Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447//! 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; } }};