//! Sends the user's changes to Gmail as they make them, on the event
//! loop. A thread has one request out at most, so its changes land in
//! the order they were made; what changes meanwhile waits, and goes as
//! one request when the first comes back. A change that waits is undone
//! by its opposite, such as an undo, so neither goes; once it has gone,
//! the opposite follows it.

const std = @import("std");
const siv = @import("siv");
const Gmail = @import("Gmail.zig");
const Http = @import("Http.zig");
const Mailbox = @import("Mailbox.zig");
const ascii = std.ascii;
const mem = std.mem;
const testing = std.testing;
const Allocator = std.mem.Allocator;
const ArenaAllocator = std.heap.ArenaAllocator;
const ArrayList = std.ArrayList;
const Io = std.Io;
const log = std.log.scoped(.sync);

const Sync = @This();

/// Why a change has not reached Gmail.
pub const Trouble = enum {
    /// The access token expired. Changes wait until `proceed`.
    unauthorized,
    /// Gmail refused it or could not be reached. It is dropped, and the
    /// log says why.
    failed,
};

/// Hears trouble, on the app's thread.
pub const Report = *const fn (context: *anyopaque, trouble: Trouble) void;

gpa: Allocator,
/// The account; its authorization changes when the token is refreshed.
gmail: Gmail,
context: *anyopaque,
report: Report,
/// Gmail's ids of the threads and labels changed so far. A change names
/// them by place here, which outlasts the mailbox it came from.
names: std.array_hash_map.String(void),
names_arena: ArenaAllocator,
/// Changes not sent yet, by thread.
pending: std.array_hash_map.Auto(Name, Delta),
/// The request out for each thread that has one.
out: std.array_hash_map.Auto(Name, *Out),
/// Set until there is a token, and from a 401 until a new one: nothing
/// is sent meanwhile.
paused: bool,

/// A thread's or a label's place in `names`.
const Name = u32;

/// What a thread needs changed in Gmail, from where the request out
/// for it leaves it. The mailbox tells only real changes, so a change
/// to the opposite of one here takes it back.
const Delta = struct {
    /// Labels to put on (true) or take off (false).
    labels: std.array_hash_map.Auto(Name, bool),
    unread: ?bool,
    trash: ?Gmail.Bin,

    const empty: Delta = .{ .labels = .empty, .unread = null, .trash = null };

    fn deinit(d: *Delta, gpa: Allocator) void {
        d.labels.deinit(gpa);
        d.* = undefined;
    }

    fn label(d: *Delta, gpa: Allocator, l: Name, on: bool) Allocator.Error!void {
        if (d.labels.get(l)) |was| if (was != on) {
            _ = d.labels.swapRemove(l);
            return;
        };
        try d.labels.put(gpa, l, on);
    }

    /// `newer`'s changes after these.
    fn add(d: *Delta, gpa: Allocator, newer: *const Delta) Allocator.Error!void {
        var it = newer.labels.iterator();
        while (it.next()) |e| try d.label(gpa, e.key_ptr.*, e.value_ptr.*);
        if (newer.unread) |u| d.unread = then(bool, d.unread, u);
        if (newer.trash) |t| d.trash = then(Gmail.Bin, d.trash, t);
    }

    /// `next` after `was`: its opposite takes it back.
    fn then(comptime T: type, was: ?T, next: T) ?T {
        return if (was != null and was.? != next) null else next;
    }
};

/// A request out, and the changes it carries.
const Out = struct {
    sync: *Sync,
    thread: Name,
    delta: Delta,
};

/// Starts paused, as there is no token yet: changes wait for `proceed`.
pub fn init(gpa: Allocator, gmail: Gmail, context: *anyopaque, report: Report) Sync {
    return .{
        .gpa = gpa,
        .gmail = gmail,
        .context = context,
        .report = report,
        .names = .empty,
        .names_arena = .init(gpa),
        .pending = .empty,
        .out = .empty,
        .paused = true,
    };
}

/// Drops what is unsent. Deinit the Http first, so no reply comes after.
pub fn deinit(self: *Sync) void {
    for (self.pending.values()) |*d| d.deinit(self.gpa);
    self.pending.deinit(self.gpa);
    for (self.out.values()) |out| {
        out.delta.deinit(self.gpa);
        self.gpa.destroy(out);
    }
    self.out.deinit(self.gpa);
    self.names.deinit(self.gpa);
    self.names_arena.deinit();
    self.* = undefined;
}

/// What to set as a mailbox's listener.
pub fn listener(self: *Sync) Mailbox.Listener {
    return .{ .context = self, .changed = changed };
}

/// Whether nothing is out and nothing waits that could go: paused, the
/// changes wait for a token that may not come.
pub fn idle(self: *const Sync) bool {
    return self.out.count() == 0 and (self.paused or self.pending.count() == 0);
}

/// Sends what waited for a new token.
pub fn proceed(self: *Sync) void {
    self.paused = false;
    var waiting: ArrayList(Name) = .empty;
    defer waiting.deinit(self.gpa);
    waiting.appendSlice(self.gpa, self.pending.keys()) catch |err| return self.fail("resuming", err);
    for (waiting.items) |thread| {
        if (!self.out.contains(thread)) self.send(thread) catch |err| self.fail("sending", err);
    }
}

fn changed(context: *anyopaque, mail: *const Mailbox, change: Mailbox.Change) void {
    const self: *Sync = @ptrCast(@alignCast(context));
    self.record(mail, change) catch |err| self.fail("keeping a change", err);
}

fn record(self: *Sync, mail: *const Mailbox, change: Mailbox.Change) !void {
    const id = switch (change) {
        .label => |c| c.thread,
        .unread => |c| c.thread,
        .trash => |t| t,
        .untrash => |c| c.thread,
    };
    const thread = try self.name(mail.get(id).id);
    const label: Name = try self.name(mail.label(if (change == .label) change.label.label else .inbox).id);
    const entry = try self.pending.getOrPut(self.gpa, thread);
    if (!entry.found_existing) entry.value_ptr.* = .empty;
    const d = entry.value_ptr;
    switch (change) {
        .label => |c| try d.label(self.gpa, label, c.on),
        .unread => |c| d.unread = Delta.then(bool, d.unread, c.unread),
        .trash => d.trash = Delta.then(Gmail.Bin, d.trash, .trash),
        .untrash => |c| {
            // A trash that has not gone took nothing out of the Inbox.
            const went = d.trash != .trash;
            d.trash = Delta.then(Gmail.Bin, d.trash, .untrash);
            if (c.inbox and went) try d.label(self.gpa, label, true);
        },
    }
    if (!self.paused and !self.out.contains(thread)) try self.send(thread);
}

fn name(self: *Sync, id: []const u8) !Name {
    const entry = try self.names.getOrPut(self.gpa, id);
    if (!entry.found_existing) {
        entry.key_ptr.* = self.names_arena.allocator().dupe(u8, id) catch |err| {
            self.names.swapRemoveAt(entry.index);
            return err;
        };
    }
    return @intCast(entry.index);
}

/// Sends a thread's pending changes as one request.
fn send(self: *Sync, thread: Name) !void {
    const kv = self.pending.fetchSwapRemove(thread) orelse return;
    var delta = kv.value;
    errdefer delta.deinit(self.gpa);
    // Each change taken back by another.
    if (delta.labels.count() == 0 and delta.unread == null and delta.trash == null) return delta.deinit(self.gpa);
    // Trash or untrash goes alone; the rest follows it.
    if (delta.trash != null and (delta.labels.count() > 0 or delta.unread != null)) {
        try self.pending.put(self.gpa, thread, .{ .labels = delta.labels, .unread = delta.unread, .trash = null });
        delta = .{ .labels = .empty, .unread = null, .trash = delta.trash };
    }
    const out = try self.gpa.create(Out);
    errdefer self.gpa.destroy(out);
    out.* = .{ .sync = self, .thread = thread, .delta = delta };
    try self.out.put(self.gpa, thread, out);
    errdefer _ = self.out.swapRemove(thread);

    var arena_state: ArenaAllocator = .init(self.gpa);
    defer arena_state.deinit();
    const arena = arena_state.allocator();
    const id = self.names.keys()[thread];
    if (delta.trash) |way| return self.gmail.trash(arena, id, way, out, onSent);
    var add: ArrayList([]const u8) = .empty;
    var remove: ArrayList([]const u8) = .empty;
    var it = delta.labels.iterator();
    while (it.next()) |e| try (if (e.value_ptr.*) &add else &remove).append(arena, self.names.keys()[e.key_ptr.*]);
    if (delta.unread) |u| try (if (u) &add else &remove).append(arena, "UNREAD");
    try self.gmail.modify(arena, id, add.items, remove.items, out, onSent);
}

fn onSent(context: *anyopaque, response: Http.Error!Http.Response) void {
    const out: *Out = @ptrCast(@alignCast(context));
    const self = out.sync;
    const thread = out.thread;
    var delta = out.delta;
    _ = self.out.swapRemove(thread);
    self.gpa.destroy(out);
    self.sent(thread, &delta, response) catch |err| self.fail("after sending", err);
}

fn sent(self: *Sync, thread: Name, delta: *Delta, response: Http.Error!Http.Response) !void {
    var arena_state: ArenaAllocator = .init(self.gpa);
    defer arena_state.deinit();
    const arena = arena_state.allocator();
    const what = try arena.print("POST /threads/{s}", .{self.names.keys()[thread]});
    if (Gmail.check(what, arena, response)) |_| {
        delta.deinit(self.gpa);
    } else |err| switch (err) {
        error.Unauthorized => {
            try self.requeue(thread, delta);
            if (!self.paused) {
                self.paused = true;
                self.report(self.context, .unauthorized);
            }
            return;
        },
        else => {
            delta.deinit(self.gpa);
            self.report(self.context, .failed);
        },
    }
    if (!self.paused) try self.send(thread);
}

/// Puts a refused request's changes back in line, under what changed
/// since it went.
fn requeue(self: *Sync, thread: Name, delta: *Delta) !void {
    errdefer delta.deinit(self.gpa);
    const entry = try self.pending.getOrPut(self.gpa, thread);
    if (entry.found_existing) {
        const newer = entry.value_ptr;
        try delta.add(self.gpa, newer);
        newer.deinit(self.gpa);
    }
    entry.value_ptr.* = delta.*;
}

fn fail(self: *Sync, what: []const u8, err: anyerror) void {
    log.err("{s}: {t}", .{ what, err });
    self.report(self.context, .failed);
}

test "changes reach Gmail in order, one request per thread at a time, wait out a 401, and are undone" {
    var address: Io.net.IpAddress = .{ .ip4 = .loopback(0) };
    var server = try address.listen(testing.io, .{});
    defer server.deinit(testing.io);
    var fake: Fake = .{};
    defer fake.deinit();
    var serving = try testing.io.concurrent(Fake.serve, .{ &fake, &server, testing.io });
    defer serving.cancel(testing.io);

    const app = try siv.appCreate(.{
        .backend = .headless,
        .flags = .{ .no_system_fonts = true, .no_system_icons = true, .reduced_motion = true },
    });
    defer siv.appDestroy(app);
    const http = try Http.init(testing.allocator, app);
    defer http.deinit();
    var base_buf: [64]u8 = undefined;
    const gmail: Gmail = .{
        .http = http,
        .base = try mem.print(&base_buf, "http://127.0.0.1:{d}", .{server.socket.address.getPort()}),
        .authorization = "authorization: Bearer t",
    };
    var troubles: std.EnumArray(Trouble, u32) = .initFill(0);
    var sync: Sync = .init(testing.allocator, gmail, &troubles, struct {
        fn report(context: *anyopaque, trouble: Trouble) void {
            const t: *std.EnumArray(Trouble, u32) = @ptrCast(@alignCast(context));
            t.getPtr(trouble).* += 1;
        }
    }.report);
    defer sync.deinit();
    sync.proceed();

    var mail: Mailbox = try .init(testing.allocator);
    defer mail.deinit();
    const black: Mailbox.Rgb = .{ .r = 0, .g = 0, .b = 0 };
    const work = try mail.addLabel(.{ .id = "Label_1", .name = "Work", .kind = .user, .color = black });
    var ids: [6]Mailbox.Id = undefined;
    for ([_][]const u8{ "t1", "t2", "t3", "t4", "t5", "t6" }, &ids) |id, *out| out.* = try mail.addThread(.{
        .id = id,
        .participants = "Ana",
        .subject = "Hi",
        .snippet = "",
        .date = 1,
        .messages = 1,
        .labels = &.{.inbox},
        .unread = false,
        .notes = &.{},
    });
    mail.listener = sync.listener();

    // The first goes at once and is refused; the rest of t1's changes
    // wait behind it, Work's two toggles ending off.
    mail.archive(ids[0]);
    mail.toggleLabel(ids[0], work);
    mail.toggleLabel(ids[0], work);
    mail.toggleLabel(ids[0], work);
    mail.setUnread(ids[0], true);
    mail.trash(ids[1]);
    mail.archive(ids[1]); // trashed, so not in the inbox: nothing to send

    // Undone before it goes, a change goes not at all; after, its
    // opposite follows it. Untrash puts the Inbox back by name.
    var inbox = try mail.get(ids[3]).labels.clone(testing.allocator);
    defer inbox.deinit(testing.allocator);
    mail.toggleLabel(ids[2], .starred);
    mail.archive(ids[2]);
    mail.setLabel(ids[2], .inbox, true);
    mail.trash(ids[3]);
    mail.untrash(ids[3], &inbox);
    mail.toggleLabel(ids[4], .starred);
    mail.trash(ids[4]);
    mail.untrash(ids[4], &inbox);
    mail.archive(ids[5]);
    mail.setLabel(ids[5], .inbox, true);
    var waited: u32 = 0;
    while (!sync.paused and waited < 100) : (waited += 1) try siv.appDispatch(app, 100);
    try testing.expectEqual(1, troubles.get(.unauthorized));
    sync.proceed();
    waited = 0;
    while (!sync.idle() and waited < 100) : (waited += 1) try siv.appDispatch(app, 100);
    try testing.expect(sync.idle());
    try testing.expectEqual(0, troubles.get(.failed));

    try fake.mutex.lock(testing.io);
    defer fake.mutex.unlock(testing.io);
    const archived = "{\"addLabelIds\":[],\"removeLabelIds\":[\"INBOX\"]}";
    const inboxed = "{\"addLabelIds\":[\"INBOX\"],\"removeLabelIds\":[]}";
    const starred = "{\"addLabelIds\":[\"STARRED\"],\"removeLabelIds\":[]}";
    const want = [_]struct { []const u8, []const []const u8 }{
        .{ "t1", &.{
            "/threads/t1/modify " ++ archived,
            "/threads/t1/modify {\"addLabelIds\":[\"Label_1\",\"UNREAD\"],\"removeLabelIds\":[\"INBOX\"]}",
        } },
        .{ "t2", &.{"/threads/t2/trash "} },
        .{ "t3", &.{"/threads/t3/modify " ++ starred} },
        .{ "t4", &.{ "/threads/t4/trash ", "/threads/t4/untrash ", "/threads/t4/modify " ++ inboxed } },
        .{ "t5", &.{"/threads/t5/modify " ++ starred} },
        .{ "t6", &.{ "/threads/t6/modify " ++ archived, "/threads/t6/modify " ++ inboxed } },
    };
    for (want) |w| {
        var seen: ArrayList([]const u8) = .empty;
        defer seen.deinit(testing.allocator);
        var path_buf: [8]u8 = undefined;
        const path = try mem.print(&path_buf, "/{s}/", .{w[0]});
        for (fake.seen.items) |line| if (mem.find(u8, line, path) != null) try seen.append(testing.allocator, line);
        try testing.expectEqual(w[1].len, seen.items.len);
        for (w[1], seen.items) |expected, line| try testing.expectEqualStrings(expected, line);
    }
}

/// A stand-in for Gmail that notes each POST and refuses t1's first.
const Fake = struct {
    mutex: Io.Mutex = .init,
    /// "path body" for each request, in the order they came.
    seen: ArrayList([]const u8) = .empty,
    arena: ArenaAllocator = .init(testing.allocator),

    fn deinit(fake: *Fake) void {
        fake.seen.deinit(testing.allocator);
        fake.arena.deinit();
        fake.* = undefined;
    }

    fn serve(fake: *Fake, server: *Io.net.Server, io: Io) void {
        var connections: Io.Group = .init;
        defer connections.cancel(io);
        while (true) {
            const stream = server.accept(io) catch |err| switch (err) {
                error.Canceled => return,
                else => continue,
            };
            connections.concurrent(io, connection, .{ fake, stream, io }) catch stream.close(io);
        }
    }

    fn connection(fake: *Fake, stream: Io.net.Stream, io: Io) Io.Cancelable!void {
        defer stream.close(io);
        var read_buf: [4096]u8 = undefined;
        var write_buf: [4096]u8 = undefined;
        var body_buf: [1024]u8 = undefined;
        var reader = stream.reader(io, &read_buf);
        var writer = stream.writer(io, &write_buf);
        var http: std.http.Server = .init(&reader.interface, &writer.interface);
        while (true) {
            var request = http.receiveHead() catch return;
            var auth: []const u8 = "";
            var headers = request.iterateHeaders();
            while (headers.next()) |h| {
                if (ascii.eqlIgnoreCase(h.name, "authorization")) auth = h.value;
            }
            const target = request.head.target;
            const body = request.readerExpectNone(&body_buf).allocRemaining(testing.allocator, .limited(1024)) catch
                return;
            defer testing.allocator.free(body);
            const first = blk: {
                try fake.mutex.lock(io);
                defer fake.mutex.unlock(io);
                const line = mem.concat(fake.arena.allocator(), u8, &.{ target, " ", body }) catch return;
                fake.seen.append(testing.allocator, line) catch return;
                if (mem.find(u8, target, "/t1/") == null) break :blk false;
                for (fake.seen.items[0 .. fake.seen.items.len - 1]) |before| {
                    if (mem.find(u8, before, "/t1/") != null) break :blk false;
                }
                break :blk true;
            };
            const status: std.http.Status = if (!mem.eql(u8, auth, "Bearer t") or first) .unauthorized else .ok;
            request.respond("{}", .{ .status = status }) catch return;
        }
    }
};