Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989//! The signed-in account while the app runs. It trades the saved refresh//! token for an access token, brings the desk's mail up to date from//! Gmail (or reads it whole, the first time), and sends the user's//! changes back, all on the event loop, so the window is up and answers//! keys from the start. An access token lasts an hour; when Gmail//! refuses one, the session gets another. Once a minute it asks Gmail//! what has changed, which costs next to nothing when nothing has, and//! tells the desktop of new mail in the Inbox while the window is not//! looked at, taking the notification back once that mail is read//! (Notifier.zig). The list reads more as it is scrolled to its end//! (Pager.zig), and older mail fills in behind it while the user does//! not need the quota (Backfill.zig), and then every message is kept for//! search (Harvest.zig), which the index answers once it has them all.//! The draft being written is kept in//! Gmail's drafts and sent from there (Drafts.zig). The mail is saved for//! the next run once it loads, every few minutes while it grows, and as//! the app quits.
const std = @import("std");const siv = @import("siv");const Backfill = @import("Backfill.zig");const Corpus = @import("Corpus.zig");const Desk = @import("Desk.zig");const Drafts = @import("Drafts.zig");const Gmail = @import("Gmail.zig");const Compose = @import("Compose.zig");const Harvest = @import("Harvest.zig");const Http = @import("Http.zig");const Notifier = @import("Notifier.zig");const Mailbox = @import("Mailbox.zig");const Pager = @import("Pager.zig");const Query = @import("Query.zig");const Reader = @import("Reader.zig");const Settings = @import("Settings.zig");const Sync = @import("Sync.zig");const cache = @import("cache.zig");const oauth = @import("oauth.zig");const mem = std.mem;const Allocator = std.mem.Allocator;const ArenaAllocator = std.heap.ArenaAllocator;const ArrayList = std.ArrayList;const Io = std.Io;const log = std.log.scoped(.session);const testing = std.testing;
const Session = @This();
gpa: Allocator,io: Io,app: siv.App,http: *Http,desk: *Desk,settings: ?Settings,sync: Sync,drafts: Drafts,/// Holds what lasts as long as the session: the client, the refresh/// token, and the authorization header.arena_state: ArenaAllocator,client: oauth.Client,refresh_token: []u8,authorization: []const u8,/// The mail being read whole, until the desk takes it.mail: ?Mailbox,/// What changed since the desk's mail, being read, until it goes in.changes: Gmail.Changes,/// Holds a load's work and the changes, and is emptied when it ends.load_arena: ArenaAllocator,/// When the access token last came, to tell a token Gmail refuses from/// one that expired.authorized_at: ?Io.Timestamp,/// Until the next look for new mail.poll: ?siv.Timer,/// Whether a token or mail is being fetched, which ends in a poll.loading: bool,/// Whether a new access token is being fetched.refreshing: bool,/// Whether to read the mail once the new token comes.reread: bool,/// Whether the last load failed, so the next failure is not told again.offline: bool,/// Threads and files being read for the reader.fetches: ArrayList(*Fetch),notifier: Notifier,/// How the mail is being read, while it is.reading: Reading,/// Goes down what the list shows as the user scrolls; its query is the/// view the desk last asked for.pager: Pager,/// A view the desk asked for while `pager` was busy, asked once it is/// not; `wanted_arena` holds it.wanted: ?struct { query: Gmail.Query, first: bool, reach: Desk.Reach },wanted_arena: ArenaAllocator,backfill: Backfill,/// Whether the backfill has started this run.filling: bool,/// When a request of the user's last ended: the backfill waits a quota/// minute after.user_at: ?Io.Timestamp,/// When the mail was last saved.saved_at: ?Io.Timestamp,/// What search keeps, and the walk filling it.corpus: Corpus,harvest: Harvest,/// What the index found for a search.found: Found,
/// A request of the reader's, and what holds its reply.const Fetch = struct { session: *Session, arena: ArenaAllocator, thread: []const u8, /// The file being read, for an attachment or a resource. `thread` /// holds the query of a search. want: ?*Reader.Want = null,};
/// The threads the index found for the search the list shows, best/// first, in `arena`, and since when it holds the mail, or null for all/// of it; how far down the list has had them; and the page of them shown/// next, whose threads not loaded `read_arena` reads.const Found = struct { query: Gmail.Query, threads: []const []const u8, since: ?i64, position: u32, page: []const []const u8, reading: bool, arena: ArenaAllocator, read_arena: ArenaAllocator,};
/// How often to ask Gmail what has changed.const poll_ms = 60_000;
const Reading = enum { changes, whole };
/// How long after the user's last request the backfill waits: Gmail's/// quota is a minute's, so the user's next minute is theirs whole.const quiet_s = 60;
/// How often the mail is saved at most while it grows; some 30 MB at/// 100,000 threads, so not each time it does.const save_every_s = 300;
/// What the list reads in one go as it is scrolled: up to 25 threads/// (1,000 quota units), going down up to ten pages of 500 ids (100) to/// find them past those loaded already.const list_options: Pager.Options = .{ .reads = 25, .pages = 10, .want = .unknown };/// A search's first page, asked at each pause in typing, reads fewer:/// the rest come as the list is scrolled.const search_options: Pager.Options = .{ .reads = 10, .pages = 10, .want = .unknown };
/// Starts bringing the desk's mail up to date, and sends what the user/// changes in it from now on. The session copies `client` and/// `refresh_token`.pub fn start( gpa: Allocator, io: Io, app: siv.App, desk: *Desk, settings: ?Settings, client: oauth.Client, refresh_token: []const u8, account: Notifier.Account,) !*Session { const self = try gpa.create(Session); errdefer gpa.destroy(self); const http = try Http.init(gpa, app); const notifier: Notifier = try .init(gpa, app, desk, account); self.* = .{ .gpa = gpa, .io = io, .app = app, .http = http, .desk = desk, .settings = settings, .sync = .init(gpa, .{ .http = http, .base = Gmail.api, .authorization = "" }, self, onTrouble), .drafts = .init(gpa, &self.sync.gmail, settings, .{ .context = self, .sent = onDraftSent, .restored = onDraftRestored, .trouble = onDraftTrouble, }), .arena_state = .init(gpa), .client = undefined, .refresh_token = &.{}, .authorization = "", .mail = null, .changes = .empty, .load_arena = .init(gpa), .authorized_at = null, .poll = null, .loading = false, .refreshing = false, .reread = false, .offline = false, .fetches = .empty, .notifier = notifier, .reading = .changes, .pager = .init(gpa, &self.sync.gmail), .wanted = null, .wanted_arena = .init(gpa), .backfill = .init(gpa, app, &self.sync.gmail, .{ .context = self, .mailboxes = backfillMailboxes, .busy = backfillBusy, .filled = onFilled, .gone = onGone, }, Backfill.default_timing), .filling = false, .user_at = null, .saved_at = null, .corpus = undefined, .harvest = .init(gpa, app, &self.sync.gmail, &self.corpus, .{ .context = self, .busy = harvestBusy, .refused = harvestRefused, }, Harvest.default_timing), .found = .{ .query = .{ .labels = &.{}, .search = "" }, .threads = &.{}, .since = null, .position = 0, .page = &.{}, .reading = false, .arena = .init(gpa), .read_arena = .init(gpa), }, }; self.corpus.open(gpa, io, app, settings, .{ .context = self, .call = corpusReady }); errdefer self.destroy(); const a = self.arena_state.allocator(); self.client = .{ .id = try a.dupe(u8, client.id), .secret = if (client.secret) |s| try a.dupe(u8, s) else null, }; self.refresh_token = try a.dupe(u8, refresh_token);
desk.mail.listener = self.listener(); desk.messages.fetcher = .{ .context = self, .fetch = fetchMessages, .attachment = fetchAttachment, .web = fetchResource, }; desk.compose.keeper = self.drafts.keeper(); try self.drafts.restore(); desk.pages = .{ .context = self, .more = morePages, .pause_ms = 1000, .shortest = 3 }; if (desk.mail.history == 0) { desk.loading = true; try desk.tell("Loading mail…", .{}); } try self.signIn(); return self;}
/// Gives changes still on their way to Gmail a few seconds to land, as/// the app quits, with the draft as it is.pub fn finish(self: *Session) void { self.stopPolling(); self.backfill.stop(); self.harvest.stop(); self.desk.compose.putAway(); var waited: u32 = 0; while (!(self.sync.idle() and self.drafts.idle()) and waited < 50) : (waited += 1) { siv.appDispatch(self.app, 100) catch |err| { log.err("sending the last changes: {t}", .{err}); break; }; } if (!self.sync.idle()) log.warn("quit before every change reached Gmail", .{}); if (!self.drafts.idle()) log.warn("quit before the draft reached Gmail", .{}); self.save();}
/// Stops what is under way. Call before the desk goes.pub fn destroy(self: *Session) void { self.stopPolling(); self.desk.messages.fetcher = null; self.desk.compose.keeper = null; self.desk.pages = null; self.http.deinit(); self.drafts.deinit(); for (self.fetches.items) |f| self.endFetch(f); self.fetches.deinit(self.gpa); self.backfill.deinit(); self.harvest.deinit(); self.corpus.deinit(); self.found.arena.deinit(); self.found.read_arena.deinit(); self.pager.deinit(); self.wanted_arena.deinit(); self.notifier.deinit(); self.sync.deinit(); if (self.mail) |*m| m.deinit(); std.crypto.secureZero(u8, self.refresh_token); self.load_arena.deinit(); self.arena_state.deinit(); self.gpa.destroy(self);}
/// Trades the refresh token for the first access token, then reads the/// mail.fn signIn(self: *Session) !void { const request = try oauth.refreshRequest(self.load_arena.allocator(), .google, self.client, self.refresh_token); try self.http.send(request, self, onToken); self.loading = true;}
fn onToken(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *Session = @ptrCast(@alignCast(context)); self.loading = false; self.loadMail(response) catch |err| self.failed(err);}
fn loadMail(self: *Session, response: Http.Error!Http.Response) !void { errdefer _ = self.load_arena.reset(.free_all); const token = oauth.readToken(self.load_arena.allocator(), try response) catch |err| { if (err == error.TokenRefused) self.tell("Google refused the saved sign-in: run `parcel login` again"); return err; }; try self.authorize(token); // The token is copied; the read below keeps its work here until it // ends. _ = self.load_arena.reset(.free_all); self.sync.proceed(); self.drafts.proceed(); self.askWanted(); // For a reply to everyone, which leaves the user out. self.sync.gmail.askAddress(self.load_arena.allocator(), self, onAddress) catch |err| log.warn("asking for the address: {t}", .{err}); self.sync.gmail.askSignature(self.load_arena.allocator(), self, onSignature) catch |err| log.warn("asking for the signature: {t}", .{err}); try self.read(if (self.desk.mail.history == 0) .whole else .changes);}
/// Reads what changed since the desk's mail into `changes`, or all the/// mail into `mail`.fn read(self: *Session, how: Reading) !void { errdefer _ = self.load_arena.reset(.free_all); const arena = self.load_arena.allocator(); const options: Gmail.LoadOptions = .{ .contacts = &self.desk.compose.contacts }; switch (how) { .changes => try self.sync.gmail.update(arena, self.desk.mail, &self.changes, options, self, onLoaded), .whole => { self.mail = try .init(self.gpa); errdefer { self.mail.?.deinit(); self.mail = null; } try self.sync.gmail.load(arena, &self.mail.?, options, self, onLoaded); }, } self.reading = how; self.loading = true;}
fn onLoaded(context: *anyopaque, result: anyerror!void) void { const self: *Session = @ptrCast(@alignCast(context)); self.loading = false; const whole = self.mail; self.mail = null; result catch |err| { if (whole) |m| { var mail = m; mail.deinit(); } _ = self.load_arena.reset(.free_all); switch (err) { error.HistoryExpired, error.TooManyChanges => { log.info("reading the mail whole: {t}", .{err}); self.read(.whole) catch |e| self.failed(e); }, error.Unauthorized => { self.reread = true; self.reauthorize() catch |e| self.failed(e); }, else => self.failed(err), } return; }; // The changes are in it. defer _ = self.load_arena.reset(.free_all); self.offline = false; self.schedule(); defer self.harvest.start(); defer self.startBackfill(); if (whole) |mail| return self.showWhole(mail); for (self.changes.deleted) |id| self.corpus.remove(id) catch |err| log.err("dropping a message gone: {t}", .{err}); // Nothing changed: the desk keeps what it shows. if (self.changes.history == self.desk.mail.history) return; self.putChanges() catch |err| log.err("showing the mail: {t}", .{err});}
/// Shows mail read whole, with the threads of the desk's it did not read/// again, as when Gmail no longer has the history since the last run:/// the backfill goes over them from the top to see what changed.fn showWhole(self: *Session, read_whole: Mailbox) void { var mail = read_whole; if (mail.history == self.desk.mail.history) return mail.deinit(); mail.copyAbsent(self.desk.mail, null) catch |err| log.err("keeping the mail read before: {t}", .{err}); self.show(mail);}
/// Puts what changed in the desk's mail in place, and in mail waiting to/// replace it: the threads read again in place of theirs, and those gone/// taken out. When the labels changed, which moves their places, the/// desk gets new mail instead.fn putChanges(self: *Session) !void { const changes = &self.changes; var buf: [2]*Mailbox = undefined; const mailboxes = self.desk.mailboxes(&buf); // What the desk will show, which may still wait for the user. const newest = mailboxes[mailboxes.len - 1]; if (!changes.sameLabels(newest)) return self.show(try changes.rebuild(self.gpa, newest)); // When each thread's newest message was from, to tell what is news. const then = try self.load_arena.allocator().alloc(?i64, changes.threads.len); for (changes.threads, then) |t, *date| { date.* = if (newest.findThread(t.id)) |id| newest.get(id).date else null; } for (mailboxes) |m| { try Gmail.put(m, changes.threads); m.history = changes.history; } log.info("{d} threads changed and {d} went", .{ changes.threads.len, changes.gone.count() }); var arrived: ArrayList(Mailbox.Id) = .empty; defer arrived.deinit(self.gpa); for (changes.threads, then) |t, date| { const id = newest.findThread(t.id).?; if (newest.arrived(id, date)) try arrived.append(self.gpa, id); } mem.sort(Mailbox.Id, arrived.items, newest, Mailbox.newerFirst); self.notifier.tell(newest, &arrived); try self.desk.updated(changes.gone.keys()); // What it named may have been read elsewhere, as on the phone. self.notifier.withdrawRead(newest); self.saveSometimes();}
/// Has the desk show `mail`, new mail in place of its own.fn show(self: *Session, new_mail: Mailbox) void { var mail = new_mail; mail.listener = self.listener(); log.info("read {d} labels and {d} threads from Gmail", .{ mail.labelCount(), mail.threads.items.len }); // Against what the desk will show, which may still wait for the user. const before: *const Mailbox = if (self.desk.pending_mail) |*p| p else self.desk.mail; // The first mail an account loads is not news. if (before.history != 0) self.notifier.announce(&mail, before); self.desk.replaceMail(mail) catch |err| log.err("showing the mail: {t}", .{err}); // What it named may have been read elsewhere, as on the phone. self.notifier.withdrawRead(if (self.desk.pending_mail) |*p| p else self.desk.mail); self.saveSometimes();}
/// Sends each change the user makes to Gmail, and takes back the new-mail/// notification once the user has read what it names.fn listener(self: *Session) Mailbox.Listener { return .{ .context = self, .changed = onChanged };}
fn onChanged(context: *anyopaque, mail: *const Mailbox, change: Mailbox.Change) void { const self: *Session = @ptrCast(@alignCast(context)); const sync = self.sync.listener(); sync.changed(sync.context, mail, change); self.notifier.withdrawRead(mail);}
/// Starts the backfill once the mail has loaded, and again from the top/// after it was read whole.fn startBackfill(self: *Session) void { if (self.filling and self.reading != .whole) return; self.filling = true; self.backfill.start(if (self.reading == .whole) .top else .where_it_was) catch |err| { log.err("starting the backfill: {t}", .{err}); };}
fn backfillMailboxes(context: *anyopaque, buf: *[2]*Mailbox) []const *Mailbox { const self: *Session = @ptrCast(@alignCast(context)); return self.desk.mailboxes(buf);}
/// Whether the backfill waits: for the session's own requests, for the/// user's, and for a quota minute after the user's last.fn backfillBusy(context: *anyopaque) bool { const self: *Session = @ptrCast(@alignCast(context)); if (self.loading or self.refreshing or self.authorized_at == null) return true; if (self.fetches.items.len > 0 or self.pager.busy() or self.found.reading) return true; if (!self.sync.idle() or !self.drafts.idle()) return true; const at = self.user_at orelse return false; return at.durationTo(.now(self.io, .awake)).toSeconds() < quiet_s;}
/// Whether the walk for search waits: for what the backfill waits for, and/// for the backfill, as the two share its part of the quota.fn harvestBusy(context: *anyopaque) bool { const self: *Session = @ptrCast(@alignCast(context)); return backfillBusy(self) or self.backfill.state != .stopped;}
/// The walk's access token has expired; it waits, busy, for the next.fn harvestRefused(context: *anyopaque) void { const self: *Session = @ptrCast(@alignCast(context)); self.reauthorize() catch |err| log.err("refreshing the token: {t}", .{err});}
/// The walk for search starts once the corpus has been read; it waits for/// the mail to load.fn corpusReady(context: *anyopaque) void { const self: *Session = @ptrCast(@alignCast(context)); self.harvest.start();}
/// Older mail shows in the list, and its people join the contacts.fn onFilled(context: *anyopaque, threads: []const Gmail.Thread) void { const self: *Session = @ptrCast(@alignCast(context)); Gmail.notePeople(&self.desk.compose.contacts, threads) catch |err| log.err("contacts: {t}", .{err}); self.desk.grew() catch |err| log.err("showing older mail: {t}", .{err}); self.saveSometimes();}
/// Threads gone from Gmail while the history had expired leave the list.fn onGone(context: *anyopaque, ids: []const []const u8) void { const self: *Session = @ptrCast(@alignCast(context)); self.desk.updated(ids) catch |err| log.err("dropping mail gone from Gmail: {t}", .{err}); self.saveSometimes();}
fn onAddress(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *Session = @ptrCast(@alignCast(context)); var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); const body = Gmail.check("GET /profile", arena, response) catch return; const address = Gmail.emailAddress(arena, body) catch |err| return log.warn("the address: {t}", .{err}); self.desk.compose.setMe(address) catch |err| log.warn("the address: {t}", .{err});}
fn onSignature(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *Session = @ptrCast(@alignCast(context)); var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); const body = Gmail.check("GET /settings/sendAs", arena, response) catch return; const text = Gmail.signature(arena, body) catch |err| return log.warn("the signature: {t}", .{err}); self.desk.compose.setSignature(text) catch |err| log.warn("the signature: {t}", .{err});}
fn onDraftSent(context: *anyopaque, ok: bool) void { const self: *Session = @ptrCast(@alignCast(context)); self.user_at = .now(self.io, .awake); self.desk.sent(ok) catch |err| log.err("status line: {t}", .{err}); // The sent message joins its thread at once rather than in a minute. if (ok and !self.loading and !self.refreshing and self.authorized_at != null) { self.stopPolling(); self.read(.changes) catch |err| self.failed(err); }}
/// The draft from the last run, which compose opens unless it has one.fn onDraftRestored(context: *anyopaque, message: Compose.Outgoing) bool { const self: *Session = @ptrCast(@alignCast(context)); const taken = self.desk.compose.restore(message) catch |err| { log.err("taking back the draft: {t}", .{err}); return false; }; if (taken) self.tell("Your draft is back: c opens it"); return taken;}
fn onDraftTrouble(context: *anyopaque, trouble: Sync.Trouble) void { const self: *Session = @ptrCast(@alignCast(context)); switch (trouble) { .failed => self.tell("Could not keep the draft in Gmail; it stays here"), .unauthorized => self.reauthorize() catch |err| { log.err("refreshing the token: {t}", .{err}); self.tell("Could not keep the draft in Gmail; it stays here"); }, }}
fn fetchMessages(context: *anyopaque, thread: []const u8) void { const self: *Session = @ptrCast(@alignCast(context)); self.startFetch(thread) catch |err| { log.err("reading a conversation: {t}", .{err}); self.desk.messages.deliver(thread, null) catch |e| log.err("reader: {t}", .{e}); };}
fn startFetch(self: *Session, thread: []const u8) !void { try self.fetches.ensureUnusedCapacity(self.gpa, 1); const f = try self.gpa.create(Fetch); f.* = .{ .session = self, .arena = .init(self.gpa), .thread = "" }; errdefer self.endFetch(f); f.thread = try f.arena.allocator().dupe(u8, thread); try self.sync.gmail.readThread(f.arena.allocator(), f.thread, f, onMessages); self.fetches.appendAssumeCapacity(f);}
fn onMessages(context: *anyopaque, response: Http.Error!Http.Response) void { const f: *Fetch = @ptrCast(@alignCast(context)); const self = f.session; if (mem.findScalar(*Fetch, self.fetches.items, f)) |i| _ = self.fetches.swapRemove(i); defer self.endFetch(f); const arena = f.arena.allocator(); const messages: ?[]const Mailbox.Message = blk: { const body = Gmail.check("GET /threads/… whole", arena, response) catch |err| { if (err == error.Unauthorized) self.reauthorize() catch |e| log.err("refreshing the token: {t}", .{e}); break :blk null; }; break :blk Gmail.messages(arena, body) catch |err| { log.err("reading a conversation: {t}", .{err}); break :blk null; }; }; self.desk.messages.deliver(f.thread, messages) catch |err| log.err("reader: {t}", .{err});}
fn fetchAttachment(context: *anyopaque, message: []const u8, attachment: []const u8, want: *Reader.Want) void { const self: *Session = @ptrCast(@alignCast(context)); self.startAttachment(message, attachment, want) catch |err| { log.err("reading an attachment: {t}", .{err}); want.receive(null); };}
fn startAttachment(self: *Session, message: []const u8, attachment: []const u8, want: *Reader.Want) !void { try self.fetches.ensureUnusedCapacity(self.gpa, 1); const f = try self.gpa.create(Fetch); f.* = .{ .session = self, .arena = .init(self.gpa), .thread = "", .want = want }; errdefer self.endFetch(f); try self.sync.gmail.readAttachment(f.arena.allocator(), message, attachment, f, onAttachment); self.fetches.appendAssumeCapacity(f);}
fn onAttachment(context: *anyopaque, response: Http.Error!Http.Response) void { const f: *Fetch = @ptrCast(@alignCast(context)); const self = f.session; if (mem.findScalar(*Fetch, self.fetches.items, f)) |i| _ = self.fetches.swapRemove(i); defer self.endFetch(f); const arena = f.arena.allocator(); const bytes: ?[]const u8 = blk: { const body = Gmail.check("GET /messages/…/attachments/…", arena, response) catch |err| { if (err == error.Unauthorized) self.reauthorize() catch |e| log.err("refreshing the token: {t}", .{e}); break :blk null; }; break :blk Gmail.attachmentData(arena, body) catch |err| { log.err("reading an attachment: {t}", .{err}); break :blk null; }; }; f.want.?.receive(bytes);}
fn morePages( context: *anyopaque, labels: []const []const u8, search: []const u8, first: bool, reach: Desk.Reach,) void { const self: *Session = @ptrCast(@alignCast(context)); self.askPages(.{ .labels = labels, .search = search }, first, reach) catch |err| { log.err("reading more of the list: {t}", .{err}); self.desk.paged(labels, search, null, false, null) catch |e| log.err("list: {t}", .{e}); };}
/// Goes on down the view `query` shows, from its top if it is another/// view than the last or `first` says so. One view at a time: another/// waits for the one out.fn askPages(self: *Session, query: Gmail.Query, first: bool, reach: Desk.Reach) !void { // Before the first token, the ask waits for it. if (self.pager.busy() or self.found.reading or self.authorized_at == null) { _ = self.wanted_arena.reset(.retain_capacity); const a = self.wanted_arena.allocator(); const labels = try a.alloc([]const u8, query.labels.len); for (query.labels, labels) |l, *out| out.* = try a.dupe(u8, l); self.wanted = .{ .query = .{ .labels = labels, .search = try a.dupe(u8, query.search) }, .first = first, .reach = reach, }; return; } if (reach == .any and try self.searchCorpus(query, first)) return; if (first or !sameQuery(self.pager.query, query)) try self.pager.reset(query, ""); const options = if (first and query.search.len > 0) search_options else list_options; try self.pager.goOn(self.desk.mail, options, self, onPaged);}
fn sameQuery(a: Gmail.Query, b: Gmail.Query) bool { if (a.labels.len != b.labels.len or !mem.eql(u8, a.search, b.search)) return false; for (a.labels, b.labels) |x, y| if (!mem.eql(u8, x, y)) return false; return true;}
/// The threads read go in the desk's mail, and so are saved with it.fn onPaged(context: *anyopaque, result: anyerror!Pager.Page) void { const self: *Session = @ptrCast(@alignCast(context)); self.user_at = .now(self.io, .awake); // The desk may ask again from `paged`, which resets the pager's query. var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const query = self.pager.query; const labels = arena_state.allocator().dupe([]const u8, query.labels) catch &.{}; const search = arena_state.allocator().dupe(u8, query.search) catch ""; const wanted = self.wanted; self.wanted = null; if (result) |page| { var buf: [2]*Mailbox = undefined; for (self.desk.mailboxes(&buf)) |m| Gmail.put(m, page.threads) catch |err| log.err("more mail: {t}", .{err}); Gmail.notePeople(&self.desk.compose.contacts, page.threads) catch |err| log.err("contacts: {t}", .{err}); self.desk.paged(labels, search, page.ids, page.more, null) catch |err| log.err("list: {t}", .{err}); if (page.threads.len > 0) self.saveSometimes(); } else |err| { log.err("reading more of the list: {t}", .{err}); if (err == error.Unauthorized) self.reauthorize() catch |e| log.err("refreshing the token: {t}", .{e}); self.desk.paged(labels, search, null, false, null) catch |e| log.err("list: {t}", .{e}); } if (self.wanted == null) self.wanted = wanted; self.askWanted();}
/// Asks what waited for the pager or the token, if they are free now.fn askWanted(self: *Session) void { const w = self.wanted orelse return; if (self.pager.busy() or self.found.reading or self.authorized_at == null) return; self.wanted = null; self.askPages(w.query, w.first, w.reach) catch |err| { log.err("reading more of the list: {t}", .{err}); self.desk.paged(w.query.labels, w.query.search, null, false, null) catch |e| log.err("list: {t}", .{e}); };}
/// Answers a search from the index once the corpus holds a month of/// mail, rather than Gmail: a page of the threads it found, best first,/// reading those not loaded. False when the index cannot answer it, as/// for one naming labels or flags, which Gmail is asked then.fn searchCorpus(self: *Session, query: Gmail.Query, first: bool) !bool { if (query.search.len == 0) return false; const held = self.corpus.searchable(Io.Timestamp.now(self.io, .real).toSeconds()) orelse return false; const found = &self.found; if (first or !sameQuery(found.query, query)) { if (!try self.findInCorpus(query)) return false; found.since = switch (held) { .all => null, .since => |t| t, }; } const start_at = @min(found.position, found.threads.len); found.page = found.threads[start_at..@min(start_at + list_options.reads, found.threads.len)]; _ = found.read_arena.reset(.retain_capacity); const a = found.read_arena.allocator(); var missing: ArrayList([]const u8) = .empty; for (found.page) |id| if (self.desk.mail.findThread(id) == null) try missing.append(a, id); if (missing.items.len == 0) { self.showFound(true); return true; } try self.sync.gmail.read(a, missing.items, self, onFoundThreads); found.reading = true; return true;}
/// Asks the index for `query`'s search, keeping what it finds; false if/// it cannot tell.fn findInCorpus(self: *Session, query: Gmail.Query) !bool { const began: Io.Timestamp = .now(self.io, .awake); var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const q: Query = .parse(query.search, Io.Timestamp.now(self.io, .real).toSeconds()); const threads = try self.corpus.index.search(arena_state.allocator(), &q) orelse return false; const found = &self.found; _ = found.arena.reset(.retain_capacity); const a = found.arena.allocator(); found.query = (try copyQuery(a, query, &.{})).query; // The ids are the index's, which only grows. found.threads = try a.dupe([]const u8, threads); found.position = 0; log.info("search: {d} threads found in the corpus in {d} ms", .{ threads.len, began.durationTo(.now(self.io, .awake)).toMilliseconds(), }); return true;}
fn onFoundThreads(context: *anyopaque, result: anyerror![]const Gmail.Thread) void { const self: *Session = @ptrCast(@alignCast(context)); self.found.reading = false; self.user_at = .now(self.io, .awake); if (result) |threads| { var buf: [2]*Mailbox = undefined; for (self.desk.mailboxes(&buf)) |m| Gmail.put(m, threads) catch |err| log.err("what search found: {t}", .{err}); Gmail.notePeople(&self.desk.compose.contacts, threads) catch |err| log.err("contacts: {t}", .{err}); self.showFound(true); if (threads.len > 0) self.saveSometimes(); } else |err| { log.err("reading what search found: {t}", .{err}); if (err == error.Unauthorized) self.reauthorize() catch |e| log.err("refreshing the token: {t}", .{e}); self.showFound(false); } self.askWanted();}
/// Tells the desk the page of what the index found, or that it could not/// be read.fn showFound(self: *Session, came: bool) void { const found = &self.found; if (came) found.position += @intCast(found.page.len); // The desk may ask again from `paged`, which may find anew. var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const a = arena_state.allocator(); const more = found.position < found.threads.len; const copy = copyQuery(a, found.query, found.page) catch |err| return log.err("list: {t}", .{err}); const ids: ?[]const []const u8 = if (came) copy.page else null; const q = copy.query; self.desk.paged(q.labels, q.search, ids, more and came, found.since) catch |err| log.err("list: {t}", .{err});}
fn copyQuery( arena: Allocator, query: Gmail.Query, page: []const []const u8,) !struct { query: Gmail.Query, page: []const []const u8 } { const labels = try arena.alloc([]const u8, query.labels.len); for (query.labels, labels) |l, *out| out.* = try arena.dupe(u8, l); const ids = try arena.alloc([]const u8, page.len); for (page, ids) |id, *out| out.* = try arena.dupe(u8, id); return .{ .query = .{ .labels = labels, .search = try arena.dupe(u8, query.search) }, .page = ids };}
/// Reads an image or a style sheet a message shows from the web, or/// posts to a list's address to leave it, with no account's/// authorization.fn fetchResource(context: *anyopaque, url: []const u8, body: ?[]const u8, want: *Reader.Want) void { const self: *Session = @ptrCast(@alignCast(context)); self.startResource(url, body, want) catch |err| { log.warn("{s}: {t}", .{ url, err }); want.receive(null); };}
fn startResource(self: *Session, url: []const u8, body: ?[]const u8, want: *Reader.Want) !void { try self.fetches.ensureUnusedCapacity(self.gpa, 1); const f = try self.gpa.create(Fetch); f.* = .{ .session = self, .arena = .init(self.gpa), .thread = "", .want = want }; errdefer self.endFetch(f); const method: Http.Method = if (body == null) .get else .post; try self.http.send(.{ .method = method, .url = url, .headers = &.{}, .body = body }, f, onResource); self.fetches.appendAssumeCapacity(f);}
fn onResource(context: *anyopaque, response: Http.Error!Http.Response) void { const f: *Fetch = @ptrCast(@alignCast(context)); const self = f.session; if (mem.findScalar(*Fetch, self.fetches.items, f)) |i| _ = self.fetches.swapRemove(i); defer self.endFetch(f); const r = response catch return f.want.?.receive(null); if (r.status / 100 != 2) log.info("GET or POST on the web: {d}", .{r.status}); f.want.?.receive(if (r.status / 100 == 2) r.body else null);}
fn endFetch(self: *Session, f: *Fetch) void { self.user_at = .now(self.io, .awake); f.arena.deinit(); self.gpa.destroy(f);}
/// Looks for new mail in a minute.fn schedule(self: *Session) void { if (self.poll != null) return; self.poll = siv.timerStart(self.app, poll_ms, onPoll, null, self) catch |err| blk: { log.err("setting the poll's timer: {t}", .{err}); break :blk null; };}
fn stopPolling(self: *Session) void { if (self.poll) |t| siv.timerCancel(t); self.poll = null;}
fn onPoll(user_data: ?*anyopaque) callconv(.c) void { const self: *Session = @ptrCast(@alignCast(user_data)); self.poll = null; if (self.loading) return; if (self.refreshing) { self.reread = true; return; } // Never signed in, as when the network was down at start. if (self.authorized_at == null) return self.signIn() catch |err| self.failed(err); self.read(if (self.desk.mail.history == 0) .whole else .changes) catch |err| self.failed(err);}
/// Keeps the desk's mail for the next run, once it has come from Gmail.fn save(self: *Session) void { const settings = self.settings orelse return; if (self.desk.mail.history == 0) return; cache.save(self.gpa, settings, self.desk.mail); if (self.desk.compose.contacts.seeded) self.desk.compose.contacts.save(settings); self.saved_at = .now(self.io, .awake);}
/// Saves, unless it was done in the last few minutes.fn saveSometimes(self: *Session) void { if (self.saved_at) |at| { if (at.durationTo(.now(self.io, .awake)).toSeconds() < save_every_s) return; } self.save();}
fn authorize(self: *Session, token: oauth.Token) !void { self.authorization = try self.arena_state.allocator().print("authorization: Bearer {s}", .{token.access}); self.sync.gmail.authorization = self.authorization; self.authorized_at = .now(self.io, .awake);}
fn onTrouble(context: *anyopaque, trouble: Sync.Trouble) void { const self: *Session = @ptrCast(@alignCast(context)); switch (trouble) { .failed => self.tell("Could not save a change to Gmail"), .unauthorized => self.reauthorize() catch |err| { log.err("refreshing the token: {t}", .{err}); self.tell("Could not save changes to Gmail"); }, }}
/// Gets a new access token, unless the one Gmail refused is new: then/// another would fare no better, and the changes wait.fn reauthorize(self: *Session) !void { if (self.refreshing) return; if (self.authorized_at) |at| { if (at.durationTo(.now(self.io, .awake)).toSeconds() < 60) return error.TokenRefused; } var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const request = try oauth.refreshRequest(arena_state.allocator(), .google, self.client, self.refresh_token); try self.http.send(request, self, onRefreshed); self.refreshing = true;}
fn onRefreshed(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *Session = @ptrCast(@alignCast(context)); self.refreshing = false; const reread = self.reread; self.reread = false; self.refreshed(response) catch |err| { self.refreshFailed(err); if (reread) self.schedule(); return; }; self.sync.proceed(); self.drafts.proceed(); if (reread) self.read(.changes) catch |err| self.failed(err);}
fn refreshed(self: *Session, response: Http.Error!Http.Response) !void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); try self.authorize(try oauth.readToken(arena_state.allocator(), try response));}
fn refreshFailed(self: *Session, err: anyerror) void { log.err("refreshing the token: {t}", .{err}); if (err == error.TokenRefused) self.tell("Google refused the saved sign-in: run `parcel login` again") else self.tell("Could not save changes to Gmail");}
/// Says in the status line that the mail did not load, unless a reason/// is there already or it was said last time, and tries again later. The/// details are in the log.fn failed(self: *Session, err: anyerror) void { log.err("loading mail: {t}", .{err}); self.desk.loading = false; if (err != error.TokenRefused and !self.offline) self.tell("Could not load mail from Gmail"); self.offline = true; self.schedule();}
fn tell(self: *Session, comptime message: []const u8) void { self.desk.tell(message, .{}) catch |err| log.err("status line: {t}", .{err});}