Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728//! A JMAP account while the app runs, such as Fastmail's: as Session is//! for Gmail. It reads the session resource, then the mail whole the//! first time, or what changed since the state the cache kept, and asks//! again as soon as the server's event stream says something changed,//! or once a minute while the stream is down, telling the desktop of new//! mail in the Inbox (Notifier.zig). The user's changes go back one thread at a time://! the thread's messages are read, then each is patched. The reader's//! threads and files, the list's pages and Enter's search, and the draft//! go through it too. The draft's id is saved, so the next run reads it//! back, with its files, for c to open. 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 Compose = @import("Compose.zig");const Contacts = @import("Contacts.zig");const Corpus = @import("Corpus.zig");const Desk = @import("Desk.zig");const Http = @import("Http.zig");const Notifier = @import("Notifier.zig");const Query = @import("Query.zig");const Jmap = @import("Jmap.zig");const Mailbox = @import("Mailbox.zig");const Reader = @import("Reader.zig");const Settings = @import("Settings.zig");const cache = @import("cache.zig");const mime = @import("mime.zig");const json = std.json;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(.jmap_session);
const JmapSession = @This();
gpa: Allocator,io: Io,app: siv.App,http: *Http,desk: *Desk,settings: ?Settings,jmap: Jmap,/// Holds what lasts as long as the session: the authorization, and/// what the session resource and the identity say.arena_state: ArenaAllocator,session_url: []const u8,/// The user's address, their From header, and the identity sending/// takes; empty until read.me: []const u8,from: []const u8,identity: []const u8,/// Whether `me` is as known as it will be: read, or not to be.me_settled: bool,/// The account's mailboxes by role, and what holds them.roles: Jmap.Roles,roles_arena: ArenaAllocator,/// Whether the session resource has been read, so requests can go.ready: bool,/// Whether this run has read which mailbox is the Inbox, the Trash, and/// the rest, which the user's changes, the draft, and a list's pages/// name; they wait until then.roles_read: bool,/// Whether the mail is being read, which ends in a poll.loading: bool,/// Whether the server said something changed while the mail was being/// read, so it is read again after.again: bool,/// Whether the last read failed, so the next failure is not told again.offline: bool,poll: ?siv.Timer,push: Push,/// Holds a read's work, emptied when it ends.load_arena: ArenaAllocator,/// The user's changes waiting to go, the first going now if `sending`.changes: ArrayList(Pending),sending: bool,/// Threads and files being read for the reader.fetches: ArrayList(*Fetch),/// What a read of changes found, until the threads it names are read.changed: ?Changed,/// The view the list last asked more of, and the page out.pager: Pager,draft: Draft,/// The draft of an earlier run, while it is read back.kept: ?*Kept,saved_at: ?Io.Timestamp,notifier: Notifier,/// What search keeps, and the backfill filling it.corpus: Corpus,walk: Walk,
/// A change waiting to go: the thread by its id, which outlives its/// place in the mailbox.const Pending = struct { thread: [64]u8, thread_len: u8, change: Mailbox.Change };
const Fetch = struct { session: *JmapSession, arena: ArenaAllocator, thread: []const u8, want: ?*Reader.Want,};
/// A view of the list: its labels' ids and its search, how far down it/// has been read, and what holds it.const View = struct { arena: ArenaAllocator, labels: []const []const u8, search: []const u8, position: u32,
fn init(gpa: Allocator) View { return .{ .arena = .init(gpa), .labels = &.{}, .search = "", .position = 0 }; }
fn set(v: *View, labels: []const []const u8, search: []const u8, position: u32) !void { _ = v.arena.reset(.retain_capacity); const a = v.arena.allocator(); const copy = try a.alloc([]const u8, labels.len); for (labels, copy) |l, *out| out.* = try a.dupe(u8, l); v.* = .{ .arena = v.arena, .labels = copy, .search = try a.dupe(u8, search), .position = position }; }
fn same(v: *const View, labels: []const []const u8, search: []const u8) bool { if (v.labels.len != labels.len or !mem.eql(u8, v.search, search)) return false; for (v.labels, labels) |a, b| if (!mem.eql(u8, a, b)) return false; return true; }};
const Pager = struct { /// What the list last asked more of. view: View, /// The view of the page out, while `busy`. out: View, busy: bool, /// Whether a page of `view` waits for the one out or for the session. wanted: bool, /// Who answers `view`'s search. reach: Desk.Reach, /// The search the index last answered, the threads it found, best /// first, since when it holds the mail, or null for all of it, and /// the page of them out, in `found_arena`. found_search: []const u8, found: []const []const u8, found_since: ?i64, found_page: []const []const u8, found_arena: ArenaAllocator,};
/// The backfill of what search keeps: down every message, newest first,/// a page of ids then the messages of it not kept, one request at a/// time. Once it has gone all the way down, a walk ends at the first/// page it has kept all of.const Walk = struct { busy: bool, /// Whether to walk again from the top once this walk ends. again: bool, /// Where the next page starts. position: u32, /// Whether the page out was a whole one, so more follow it. whole: bool, /// The messages of the page out, and those not kept yet, in `arena`. page: []const []const u8, wanted: []const []const u8, arena: ArenaAllocator, /// How many messages this walk kept. kept: u32,};
/// A walk's page of ids, and how many messages one Email/get of it reads,/// with at most `body_bytes` of each text part.const walk_page = 256;const walk_get = 50;const body_bytes = 65_536;
const DraftStep = enum { none, save, send, discard };
/// The draft kept in the Drafts mailbox: its message's id there, the/// message as last written, and what to do with it next.const Draft = struct { id: ArrayList(u8), raw: ArrayList(u8), want: DraftStep, busy: bool, /// Whether the step out sends. sending: bool,};
/// A draft kept on an earlier run, as the server gave it, and which of/// its files is read next.const Kept = struct { arena: ArenaAllocator, id: []const u8, draft: ?Jmap.Draft, next: u32,};
/// Where the id of the draft being written is saved, as Drafts saves/// Gmail's.const draft_file = "draft";
/// The server's event stream: whether it is open, and has spoken; what/// has come of it short of a whole event; the timer that opens it again,/// and how many times in a row it has ended.const Push = struct { open: bool, live: bool, buf: ArrayList(u8), timer: ?siv.Timer, ended: u8,};
/// How often to ask what has changed, without the event stream and with/// it, which says so; the poll then only stands in for an event lost.const poll_ms = 60_000;const push_poll_ms = 15 * 60_000;
/// How long to wait before opening the event stream again: longer each/// time it ends in a row, to five minutes.const relisten_ms = [_]u32{ 5_000, 15_000, 60_000, 300_000 };
/// How often the mail is saved at most while it grows.const save_every_s = 300;
/// What the list reads at a time as it is scrolled.const page_threads = 25;
/// The newest threads read the first time, at most.const first_threads = 200;
/// What an account keeps: its token, and where its session resource is/// when it is not Fastmail's.pub const token_file = "jmap-token";pub const url_file = "jmap-session";
/// Starts reading the account's mail into the desk's, and sends what the/// user changes in it from now on. `token` is copied.pub fn start( gpa: Allocator, io: Io, app: siv.App, desk: *Desk, settings: ?Settings, session_url: []const u8, token: []const u8, account: Notifier.Account,) !*JmapSession { const self = try gpa.create(JmapSession); errdefer gpa.destroy(self); const http = try Http.init(gpa, app); errdefer http.deinit(); var notifier: Notifier = try .init(gpa, app, desk, account); errdefer notifier.deinit(); self.* = .{ .gpa = gpa, .io = io, .app = app, .http = http, .desk = desk, .settings = settings, .jmap = .init(http, ""), .arena_state = .init(gpa), .session_url = "", .me = "", .from = "", .identity = "", .me_settled = false, .roles = .none, .roles_arena = .init(gpa), .ready = false, .roles_read = false, .loading = false, .again = false, .offline = false, .poll = null, .push = .{ .open = false, .live = false, .buf = .empty, .timer = null, .ended = 0 }, .load_arena = .init(gpa), .changes = .empty, .sending = false, .fetches = .empty, .changed = null, .pager = .{ .view = .init(gpa), .out = .init(gpa), .busy = false, .wanted = false, .reach = .any, .found_search = "", .found = &.{}, .found_since = null, .found_page = &.{}, .found_arena = .init(gpa), }, .draft = .{ .id = .empty, .raw = .empty, .want = .none, .busy = false, .sending = false }, .kept = null, .saved_at = null, .notifier = notifier, .corpus = undefined, .walk = .{ .busy = false, .again = false, .position = 0, .whole = false, .page = &.{}, .wanted = &.{}, .arena = .init(gpa), .kept = 0, }, }; self.corpus.open(gpa, io, app, settings, .{ .context = self, .call = corpusReady }); const a = self.arena_state.allocator(); self.jmap.authorization = try Jmap.authorizationFor(a, token); self.session_url = try a.dupe(u8, session_url); desk.mail.listener = self.listener(); desk.messages.fetcher = .{ .context = self, .fetch = fetchMessages, .attachment = fetchAttachment, .web = fetchResource, }; desk.compose.keeper = .{ .context = self, .save = saveDraft, .send = sendDraft, .discard = discardDraft }; desk.pages = .{ .context = self, .more = morePages, .pause_ms = 400, .shortest = 2 }; desk.server = if (mem.eql(u8, session_url, Jmap.fastmail)) "Fastmail" else host(self.session_url); if (desk.mail.state.len == 0) { desk.loading = true; try desk.tell("Loading mail…", .{}); } try self.jmap.askSession(self.load_arena.allocator(), self.session_url, self, onSession); self.loading = true; return self;}
/// Gives changes on their way a few seconds to land, as the app quits.pub fn finish(self: *JmapSession) void { self.stopPolling(); self.stopListening(); self.desk.compose.putAway(); var waited: u32 = 0; while (!self.idle() and waited < 50) : (waited += 1) { siv.appDispatch(self.app, 100) catch |err| { log.err("sending the last changes: {t}", .{err}); break; }; } if (!self.idle()) log.warn("quit before every change reached the server", .{}); self.save();}
/// Stops what is under way. Call before the desk goes.pub fn destroy(self: *JmapSession) void { self.stopPolling(); self.stopListening(); self.desk.mail.listener = null; self.desk.messages.fetcher = null; self.desk.compose.keeper = null; self.desk.pages = null; // It named the session's own copy of the URL. self.desk.server = "the server"; self.http.deinit(); self.push.buf.deinit(self.gpa); self.notifier.deinit(); self.corpus.deinit(); self.walk.arena.deinit(); for (self.fetches.items) |f| endFetch(f); self.fetches.deinit(self.gpa); self.changes.deinit(self.gpa); self.pager.view.arena.deinit(); self.pager.out.arena.deinit(); self.pager.found_arena.deinit(); self.draft.id.deinit(self.gpa); std.crypto.secureZero(u8, self.draft.raw.items); self.draft.raw.deinit(self.gpa); self.dropKept(); self.load_arena.deinit(); self.roles_arena.deinit(); self.arena_state.deinit(); self.gpa.destroy(self);}
fn idle(self: *const JmapSession) bool { return !self.sending and self.changes.items.len == 0 and !self.draft.busy and self.draft.want == .none;}
// Reading the mail.
fn onSession(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.loading = false; _ = self.load_arena.reset(.free_all); self.readSession(response) catch |err| return self.failed(err); self.sync() catch |err| self.failed(err);}
fn readSession(self: *JmapSession, response: Http.Error!Http.Response) !void { const r = try response; if (r.status == 401 or r.status == 403) return error.Unauthorized; if (r.status / 100 != 2) { log.err("the session resource: {d}", .{r.status}); return error.JmapFailed; } try self.jmap.readSession(self.arena_state.allocator(), r.body); self.ready = true; self.listen(); // Only to send as; a read-only token can read the mail, not send. if (!self.jmap.can_submit) { log.info("the token cannot send mail", .{}); self.me_settled = true; return self.seedContacts(); } self.askIdentity() catch |err| log.warn("asking for the identity: {t}", .{err});}
/// Reads the mail whole when there is none from before, else what/// changed since.fn sync(self: *JmapSession) !void { if (!self.ready) return; self.again = false; const states = States.of(self.desk.mail.state); if (states.emails.len == 0) return self.readWhole(); var r: Jmap.Request = undefined; try r.init(&self.jmap, self.load_arena.allocator()); try mailboxGet(&r); try r.call("Email/changes"); try r.field("sinceState", states.emails); try r.field("maxChanges", 500); try r.end("emailChanges"); try r.call("Thread/changes"); try r.field("sinceState", states.threads); try r.field("maxChanges", 500); try r.end("threadChanges"); for ([_][]const u8{ "created", "updated" }) |which| { try r.call("Email/get"); var buf: [16]u8 = undefined; try r.ref("ids", "emailChanges", "Email/changes", try mem.print(&buf, "/{s}", .{which})); try r.field("properties", .{"threadId"}); try r.end(which); } try r.send(self, onChanges); self.loading = true;}
fn mailboxGet(r: *Jmap.Request) !void { try r.call("Mailbox/get"); try r.field("ids", null); try r.field("properties", .{ "id", "name", "parentId", "role" }); try r.end("mailboxes");}
/// The newest threads, and the messages of each.fn readWhole(self: *JmapSession) !void { var r: Jmap.Request = undefined; try r.init(&self.jmap, self.load_arena.allocator()); try mailboxGet(&r); try r.call("Email/query"); try r.field("sort", .{.{ .property = "receivedAt", .isAscending = false }}); try r.field("collapseThreads", true); try r.field("limit", @min(first_threads, self.jmap.max_objects / 4)); try r.end("query"); try r.call("Email/get"); try r.ref("ids", "query", "Email/query", "/ids"); try r.field("properties", .{"threadId"}); try r.end("heads"); try threadsFrom(&r, "heads", "/list/*/threadId"); try r.send(self, onWhole); self.loading = true;}
/// A Thread/get of the ids at `path` of the reply `from`, then an/// Email/get of its messages.fn threadsFrom(r: *Jmap.Request, from: []const u8, path: []const u8) !void { try r.call("Thread/get"); try r.ref("ids", from, "Email/get", path); try r.end("threads"); try emailsOfThreads(r);}
fn emailsOfThreads(r: *Jmap.Request) !void { try r.call("Email/get"); try r.ref("ids", "threads", "Thread/get", "/list/*/emailIds"); try r.field("properties", Jmap.summary_properties); try r.end("emails");}
/// The state strings a mailbox keeps, its messages' and its threads',/// space-separated in Mailbox.state.const States = struct { emails: []const u8, threads: []const u8,
fn of(state: []const u8) States { const space = mem.findScalar(u8, state, ' ') orelse return .{ .emails = "", .threads = "" }; return .{ .emails = state[0..space], .threads = state[space + 1 ..] }; }};
fn onWhole(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.loading = false; defer _ = self.load_arena.reset(.free_all); self.readWholeReply(response) catch |err| return self.failed(err); self.loaded();}
fn readWholeReply(self: *JmapSession, response: Http.Error!Http.Response) !void { const arena = self.load_arena.allocator(); const list = try Jmap.replies("reading the mail", arena, response); const folders = try Jmap.readFolders(arena, list, "mailboxes"); const read = try Jmap.readThreads(arena, list); try self.keepRoles(folders.roles); var mail: Mailbox = try .init(self.gpa); errdefer mail.deinit(); for (folders.labels) |l| _ = try mail.addLabel(l); const lines = try Jmap.lines(arena, &mail, self.roles, read, self.me); try Jmap.put(&mail, lines, &self.desk.compose.contacts); self.desk.compose.contacts.seeded = true; try mail.setState(read.state); // What the desk had and this read did not, such as older pages. try mail.copyAbsent(self.desk.mail, null); self.show(mail);}
fn keepRoles(self: *JmapSession, roles: Jmap.Roles) !void { _ = self.roles_arena.reset(.retain_capacity); const a = self.roles_arena.allocator(); self.roles = .{ .inbox = try a.dupe(u8, roles.inbox), .sent = try a.dupe(u8, roles.sent), .drafts = try a.dupe(u8, roles.drafts), .junk = try a.dupe(u8, roles.junk), .trash = try a.dupe(u8, roles.trash), .archive = try a.dupe(u8, roles.archive), };}
/// Has the desk show `mail` in place of its own.fn show(self: *JmapSession, new_mail: Mailbox) void { var mail = new_mail; mail.listener = self.listener(); log.info("read {d} labels and {d} threads", .{ 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.state.len > 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();}
/// The changes since the state kept: which threads to read again, and/// which are gone.fn onChanges(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.loading = false; self.readChanges(response) catch |err| { _ = self.load_arena.reset(.free_all); if (err == error.CannotCalculateChanges) { log.info("reading the mail whole: the state is too old", .{}); return self.readWhole() catch |e| self.failed(e); } return self.failed(err); };}
fn readChanges(self: *JmapSession, response: Http.Error!Http.Response) !void { const arena = self.load_arena.allocator(); const list = try Jmap.replies("reading what changed", arena, response); const folders = try Jmap.readFolders(arena, list, "mailboxes"); const changes = try Jmap.readChanges(arena, list); for (changes.deleted) |id| self.corpus.remove(id) catch |err| log.err("dropping a message gone: {t}", .{err}); try self.keepRoles(folders.roles); const pending: Changed = .{ .state = changes.state, .gone = changes.gone, .labels = folders.labels, .more = changes.more, }; if (changes.threads.len == 0) return self.putChanged(pending, .empty); self.changed = pending; var r: Jmap.Request = undefined; try r.init(&self.jmap, arena); try r.call("Thread/get"); try r.field("ids", changes.threads); try r.end("threads"); try emailsOfThreads(&r); try r.send(self, onChangedThreads); self.loading = true;}
/// What a read of changes found, until the threads it names are read.const Changed = struct { state: []const u8, gone: []const []const u8, labels: []const Mailbox.LabelInfo, more: bool,};
fn onChangedThreads(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.loading = false; const arena = self.load_arena.allocator(); const changed = self.changed orelse return; self.changed = null; const read = readThreadsReply(arena, response) catch |err| { _ = self.load_arena.reset(.free_all); return self.failed(err); }; self.putChanged(changed, read) catch |err| { _ = self.load_arena.reset(.free_all); self.failed(err); };}
fn readThreadsReply(arena: Allocator, response: Http.Error!Http.Response) !Jmap.Read { return Jmap.readThreads(arena, try Jmap.replies("reading threads", arena, response));}
/// Puts the threads read again in the desk's mail, and in mail waiting/// to replace it, and takes out those gone. When the labels changed,/// which moves their places, the desk gets new mail instead.fn putChanged(self: *JmapSession, changed: Changed, r: Jmap.Read) !void { defer _ = self.load_arena.reset(.free_all); const arena = self.load_arena.allocator(); var gone: ArrayList([]const u8) = .empty; try gone.appendSlice(arena, changed.gone); try gone.appendSlice(arena, r.gone); var buf: [2]*Mailbox = undefined; const mailboxes = self.desk.mailboxes(&buf); const newest = mailboxes[mailboxes.len - 1]; if (!Jmap.sameLabels(newest, changed.labels)) { var mail: Mailbox = try .init(self.gpa); errdefer mail.deinit(); for (changed.labels) |l| _ = try mail.addLabel(l); var except: std.array_hash_map.String(void) = .empty; for (gone.items) |id| try except.put(arena, id, {}); try mail.copyAbsent(newest, &except); const lines = try Jmap.lines(arena, &mail, self.roles, r, self.me); try Jmap.put(&mail, lines, &self.desk.compose.contacts); try mail.setState(changed.state); self.show(mail); } else { var arrived: ArrayList(Mailbox.Id) = .empty; for (mailboxes) |m| { const lines = try Jmap.lines(arena, m, self.roles, r, self.me); if (m != newest) { try Jmap.put(m, lines, null); } else { // When each thread's newest message was from, to tell what is news. const then = try arena.alloc(?i64, lines.len); for (lines, then) |l, *date| date.* = if (m.findThread(l.thread.id)) |id| m.get(id).date else null; try Jmap.put(m, lines, &self.desk.compose.contacts); for (lines, then) |l, date| { const id = m.findThread(l.thread.id).?; if (m.arrived(id, date)) try arrived.append(arena, id); } } try m.setState(changed.state); } mem.sort(Mailbox.Id, arrived.items, newest, Mailbox.newerFirst); self.notifier.tell(newest, &arrived); if (r.threads.len > 0 or gone.items.len > 0) { log.info("{d} threads changed and {d} went", .{ r.threads.len, gone.items.len }); try self.desk.updated(gone.items); self.saveSometimes(); } // What it named may have been read elsewhere, as on the phone. self.notifier.withdrawRead(newest); } if (changed.more) return self.sync(); self.loaded();}
fn loaded(self: *JmapSession) void { self.offline = false; self.desk.loading = false; self.schedule(); self.walkCorpus(); // The roles and the labels they go with are in the desk's mail now. if (self.roles_read) return; self.roles_read = true; self.sendNext(); self.draftNext(); self.restoreDraft() catch |err| { log.err("reading back the draft: {t}", .{err}); self.dropKept(); }; self.askWanted() catch |err| log.err("reading more of the list: {t}", .{err}); if (self.again) self.pushed();}
// The user's changes.
fn listener(self: *JmapSession) Mailbox.Listener { return .{ .context = self, .changed = onChanged };}
fn onChanged(context: *anyopaque, mail: *const Mailbox, change: Mailbox.Change) void { const self: *JmapSession = @ptrCast(@alignCast(context)); const thread = switch (change) { .label => |c| c.thread, .unread => |c| c.thread, .trash => |t| t, .untrash => |c| c.thread, }; const id = mail.get(thread).id; var p: Pending = .{ .thread = undefined, .thread_len = @intCast(@min(id.len, 64)), .change = change }; @memcpy(p.thread[0..p.thread_len], id[0..p.thread_len]); self.notifier.withdrawRead(mail); self.changes.append(self.gpa, p) catch |err| return log.err("keeping a change: {t}", .{err}); self.sendNext();}
/// Sends the first change waiting, if none is out: its thread's/// messages are read first, to patch each.fn sendNext(self: *JmapSession) void { if (self.sending or !self.roles_read or self.changes.items.len == 0) return; self.sending = true; self.askThreadEmails() catch |err| self.sent(err);}
fn askThreadEmails(self: *JmapSession) !void { const p = &self.changes.items[0]; var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); var r: Jmap.Request = undefined; try r.init(&self.jmap, arena_state.allocator()); try r.call("Thread/get"); try r.field("ids", .{p.thread[0..p.thread_len]}); try r.end("threads"); try r.call("Email/get"); try r.ref("ids", "threads", "Thread/get", "/list/*/emailIds"); try r.field("properties", .{ "id", "mailboxIds", "keywords", "receivedAt" }); try r.end("emails"); try r.send(self, onThreadEmails);}
fn onThreadEmails(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.patch(response) catch |err| self.sent(err);}
fn patch(self: *JmapSession, response: Http.Error!Http.Response) !void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); const list = try Jmap.replies("reading a thread to change", arena, response); const p = self.changes.items[0]; const mail = self.desk.mail; // The change as the desk's mail has it now, by the thread's place. var change = p.change; const place = mail.findThread(p.thread[0..p.thread_len]) orelse return self.sent({}); switch (change) { .label => |*c| c.thread = place, .unread => |*c| c.thread = place, .trash => |*t| t.* = place, .untrash => |*c| c.thread = place, } var r: Jmap.Request = undefined; try r.init(&self.jmap, arena); try r.call("Email/set"); try r.s.objectField("update"); try Jmap.readPatches(&r.s, arena, list, mail, self.roles, change); try r.end("set"); try r.send(self, onPatched);}
fn onPatched(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); const list = Jmap.replies("changing a thread", arena, response) catch |err| return self.sent(err); const Set = struct { notUpdated: ?json.ArrayHashMap(struct { type: []const u8 = "" }) = null }; const set = Jmap.reply(Set, arena, list, "set") catch |err| return self.sent(err); if (set.notUpdated) |failed_ones| if (failed_ones.map.count() > 0) { log.err("changing a thread: {s}", .{failed_ones.map.values()[0].type}); return self.sent(error.NotUpdated); }; self.sent({});}
/// The change out has gone, or failed; the next goes.fn sent(self: *JmapSession, result: anyerror!void) void { self.sending = false; if (self.changes.items.len > 0) _ = self.changes.orderedRemove(0); result catch |err| { log.err("sending a change: {t}", .{err}); if (err == error.Unauthorized) self.refused() else self.tell("Could not save a change to the server"); }; self.sendNext();}
// The reader.
fn fetchMessages(context: *anyopaque, thread: []const u8) void { const self: *JmapSession = @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 newFetch(self: *JmapSession, thread: []const u8, want: ?*Reader.Want) !*Fetch { 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 endFetch(f); f.thread = try f.arena.allocator().dupe(u8, thread); self.fetches.appendAssumeCapacity(f); return f;}
fn startFetch(self: *JmapSession, thread: []const u8) !void { if (!self.ready) return error.NotReady; const f = try self.newFetch(thread, null); errdefer self.dropFetch(f); var r: Jmap.Request = undefined; try r.init(&self.jmap, f.arena.allocator()); try r.call("Thread/get"); try r.field("ids", .{f.thread}); try r.end("threads"); try r.call("Email/get"); try r.ref("ids", "threads", "Thread/get", "/list/*/emailIds"); try r.field("properties", Jmap.full_properties); try r.field("fetchAllBodyValues", true); try r.end("emails"); try r.send(f, onMessages);}
fn onMessages(context: *anyopaque, response: Http.Error!Http.Response) void { const f: *Fetch = @ptrCast(@alignCast(context)); const self = f.session; defer self.dropFetch(f); const arena = f.arena.allocator(); const messages: ?[]const Mailbox.Message = blk: { const list = Jmap.replies("reading a conversation", arena, response) catch break :blk null; break :blk Jmap.readMessages(arena, list, self.roles, self.me) 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, blob: []const u8, want: *Reader.Want) void { const self: *JmapSession = @ptrCast(@alignCast(context)); _ = message; self.startAttachment(blob, want) catch |err| { log.err("reading an attachment: {t}", .{err}); want.receive(null); };}
fn startAttachment(self: *JmapSession, blob: []const u8, want: *Reader.Want) !void { if (!self.ready) return error.NotReady; const f = try self.newFetch("", want); errdefer self.dropFetch(f); try self.jmap.download(f.arena.allocator(), blob, f, onBytes);}
/// 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: *JmapSession = @ptrCast(@alignCast(context)); self.startResource(url, body, want) catch |err| { log.warn("{s}: {t}", .{ url, err }); want.receive(null); };}
fn startResource(self: *JmapSession, url: []const u8, body: ?[]const u8, want: *Reader.Want) !void { const f = try self.newFetch("", want); errdefer self.dropFetch(f); const method: Http.Method = if (body == null) .get else .post; try self.http.send(.{ .method = method, .url = url, .headers = &.{}, .body = body }, f, onBytes);}
fn onBytes(context: *anyopaque, response: Http.Error!Http.Response) void { const f: *Fetch = @ptrCast(@alignCast(context)); defer f.session.dropFetch(f); const r = response catch return f.want.?.receive(null); // A 404 is no file there, as Gravatar says of an address with no // photo: the reader shows none, so it is no failure to keep. if (r.status / 100 != 2 and r.status != 404) log.info("reading a file: {d}", .{r.status}); f.want.?.receive(if (r.status / 100 == 2) r.body else null);}
fn dropFetch(self: *JmapSession, f: *Fetch) void { if (mem.findScalar(*Fetch, self.fetches.items, f)) |i| _ = self.fetches.swapRemove(i); endFetch(f);}
fn endFetch(f: *Fetch) void { const gpa = f.session.gpa; f.arena.deinit(); gpa.destroy(f);}
// The list's pages, and search.
fn morePages( context: *anyopaque, labels: []const []const u8, search: []const u8, first: bool, reach: Desk.Reach,) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.askPages(labels, 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 of `labels` and `search`, from its top if it is/// another view than the last or `first` says so. One page at a time:/// another waits for the one out.fn askPages(self: *JmapSession, labels: []const []const u8, search: []const u8, first: bool, reach: Desk.Reach) !void { const view = &self.pager.view; if (first or !view.same(labels, search)) try view.set(labels, search, 0); self.pager.reach = reach; self.pager.wanted = true; try self.askWanted();}
/// Asks the page waiting, if nothing is in the way now.fn askWanted(self: *JmapSession) !void { if (!self.pager.wanted or self.pager.busy or !self.roles_read) return; self.pager.wanted = false; const view = &self.pager.view; try self.pager.out.set(view.labels, view.search, view.position); if (view.search.len > 0 and self.pager.reach == .any and try self.searchCorpus()) return; var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); var r: Jmap.Request = undefined; try r.init(&self.jmap, arena_state.allocator()); try r.call("Email/query"); try r.s.objectField("filter"); try Jmap.filter(&r.s, self.desk.mail, self.roles, view.labels, view.search); try r.field("sort", .{.{ .property = "receivedAt", .isAscending = false }}); try r.field("collapseThreads", true); try r.field("position", view.position); try r.field("limit", page_threads); try r.end("query"); try r.call("Email/get"); try r.ref("ids", "query", "Email/query", "/ids"); try r.field("properties", .{"threadId"}); try r.end("heads"); try threadsFrom(&r, "heads", "/list/*/threadId"); try r.send(self, onPage); self.pager.busy = true;}
fn onPage(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.pager.busy = false; var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const out = &self.pager.out; self.readPage(arena_state.allocator(), response) catch |err| { log.err("reading more of the list: {t}", .{err}); if (err == error.Unauthorized) self.refused(); self.desk.paged(out.labels, out.search, null, false, null) catch |e| log.err("list: {t}", .{e}); }; self.askWanted() catch |err| { log.err("reading more of the list: {t}", .{err}); const v = self.pager.view; self.desk.paged(v.labels, v.search, null, false, null) catch |e| log.err("list: {t}", .{e}); };}
/// Puts a page's threads in the desk's mail, and tells the desk which/// they are, for the view asked.fn readPage(self: *JmapSession, arena: Allocator, response: Http.Error!Http.Response) !void { const out = &self.pager.out; const list = try Jmap.replies("reading a page", arena, response); const page = try Jmap.readPage(arena, list); try self.putThreads(arena, try Jmap.readThreads(arena, list)); const view = &self.pager.view; if (view.same(out.labels, out.search) and view.position == out.position) { view.position += @intCast(page.count); } try self.desk.paged(out.labels, out.search, page.threads, page.count == page_threads, null); if (page.threads.len > 0) self.saveSometimes();}
// What search keeps.
fn corpusReady(context: *anyopaque) void { const self: *JmapSession = @ptrCast(@alignCast(context)); if (self.roles_read) self.walkCorpus(); self.seedContacts();}
/// Gathers the people on the mail kept for search into the contacts, once/// for the account: the people of what it keeps from then on are noted as/// it keeps it. It waits for the corpus and the user's address, to know/// whom they wrote to.fn seedContacts(self: *JmapSession) void { const contacts = &self.desk.compose.contacts; if (contacts.from_corpus or !self.corpus.ready or !self.me_settled) return; const settings = self.settings orelse return; self.corpus.gatherPeople(self.app, settings, self.me, .{ .context = self, .call = gatheredPeople }) catch |err| log.warn("gathering the people on the mail: {t}", .{err});}
fn gatheredPeople(context: *anyopaque, people: *const Contacts) void { const self: *JmapSession = @ptrCast(@alignCast(context)); const contacts = &self.desk.compose.contacts; contacts.absorb(people) catch |err| return log.warn("gathering the people on the mail: {t}", .{err}); contacts.from_corpus = true; contacts.seeded = true; log.info("{d} people on the mail kept for search join the contacts", .{people.people.count()}); if (self.settings) |s| contacts.save(s);}
/// Walks down the messages from the top, once the corpus has been read,/// or again from the top once the walk out ends.fn walkCorpus(self: *JmapSession) void { if (!self.corpus.ready) return; if (self.walk.busy) { self.walk.again = true; return; } self.walk.position = 0; self.walk.kept = 0; self.askWalkIds() catch |err| self.walkFailed(err);}
fn askWalkIds(self: *JmapSession) !void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); var r: Jmap.Request = undefined; try r.init(&self.jmap, arena_state.allocator()); try r.call("Email/query"); // A draft is a new message each time it is saved, until it is sent. try r.field("filter", .{ .notKeyword = "$draft" }); try r.field("sort", .{.{ .property = "receivedAt", .isAscending = false }}); try r.field("position", self.walk.position); try r.field("limit", walk_page); try r.end("ids"); try r.send(self, onWalkIds); self.walk.busy = true;}
fn onWalkIds(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.walk.busy = false; self.readWalkIds(response) catch |err| self.walkFailed(err);}
fn readWalkIds(self: *JmapSession, response: Http.Error!Http.Response) !void { _ = self.walk.arena.reset(.retain_capacity); const a = self.walk.arena.allocator(); const list = try Jmap.replies("reading ids to keep for search", a, response); const ids = try Jmap.readIds(a, list, "ids"); var wanted: ArrayList([]const u8) = .empty; for (ids) |id| if (!self.corpus.has(id)) try wanted.append(a, id); self.walk.page = ids; self.walk.wanted = wanted.items; self.walk.whole = ids.len == walk_page; self.walk.position += @intCast(ids.len); if (wanted.items.len > 0) return self.askWalkEmails(); // Kept all of a page: what is below it was kept by an earlier walk. if (self.corpus.walked) return self.walkEnded(); try self.walkOn();}
fn askWalkEmails(self: *JmapSession) !void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const ids = self.walk.wanted[0..@min(self.walk.wanted.len, walk_get, self.jmap.max_objects)]; var r: Jmap.Request = undefined; try r.init(&self.jmap, arena_state.allocator()); try r.call("Email/get"); try r.field("ids", ids); try r.field("properties", Jmap.corpus_properties); try r.field("fetchTextBodyValues", true); try r.field("maxBodyValueBytes", body_bytes); try r.end("corpus"); try r.send(self, onWalkEmails); self.walk.busy = true;}
fn onWalkEmails(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.walk.busy = false; self.keepWalked(response) catch |err| self.walkFailed(err);}
fn keepWalked(self: *JmapSession, response: Http.Error!Http.Response) !void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const a = arena_state.allocator(); const list = try Jmap.replies("reading messages to keep for search", a, response); const records = try Jmap.readCorpus(a, list, "corpus"); for (records) |r| { try self.corpus.add(r); try Corpus.notePeople(&self.desk.compose.contacts, r, self.me); } self.walk.kept += @intCast(records.len); const asked = @min(self.walk.wanted.len, walk_get, self.jmap.max_objects); self.walk.wanted = self.walk.wanted[asked..]; if (self.walk.wanted.len > 0) return self.askWalkEmails(); try self.walkOn();}
/// Goes on to the next page, or ends the walk after the last.fn walkOn(self: *JmapSession) !void { try self.corpus.reach(self.walk.page); if (self.walk.whole) return self.askWalkIds(); try self.corpus.markWalked(); self.walkEnded();}
fn walkEnded(self: *JmapSession) void { if (self.walk.kept > 0) log.info("kept {d} more messages for search, {d} in all", .{ self.walk.kept, self.corpus.count(), }); if (!self.walk.again) return; self.walk.again = false; self.walkCorpus();}
/// Stops the walk; the next load starts another.fn walkFailed(self: *JmapSession, err: anyerror) void { log.err("keeping messages for search: {t}", .{err}); if (err == error.Unauthorized) self.refused(); self.walk.again = false;}
/// Puts threads read in the desk's mail.fn putThreads(self: *JmapSession, arena: Allocator, read: Jmap.Read) !void { var buf: [2]*Mailbox = undefined; for (self.desk.mailboxes(&buf)) |m| { const lines = try Jmap.lines(arena, m, self.roles, read, self.me); try Jmap.put(m, lines, null); }}
/// Answers the search of the page out from the index, once the backfill/// has kept a month of mail, rather than the server: a page of the/// threads it found, best first, reading those not loaded. False when the/// index cannot answer the search, which the server is asked then.fn searchCorpus(self: *JmapSession) !bool { const view = &self.pager.view; const held = self.corpus.searchable(Io.Timestamp.now(self.io, .real).toSeconds()) orelse return false; if (view.position == 0 or !mem.eql(u8, self.pager.found_search, view.search)) { if (!try self.findInCorpus(view.search)) return false; self.pager.found_since = switch (held) { .all => null, .since => |t| t, }; } const all = self.pager.found; const first = @min(view.position, all.len); const page = all[first..@min(first + page_threads, all.len)]; self.pager.found_page = page; var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const a = arena_state.allocator(); var missing: ArrayList([]const u8) = .empty; for (page) |id| if (self.desk.mail.findThread(id) == null) try missing.append(a, id); if (missing.items.len == 0) { try self.showFound(); return true; } var r: Jmap.Request = undefined; try r.init(&self.jmap, a); try r.call("Thread/get"); try r.field("ids", missing.items); try r.end("threads"); try emailsOfThreads(&r); try r.send(self, onFoundThreads); self.pager.busy = true; return true;}
/// Asks the index for `search`, keeping what it finds; false if it/// cannot tell.fn findInCorpus(self: *JmapSession, search: []const u8) !bool { const began: Io.Timestamp = .now(self.io, .awake); var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const q: Query = .parse(search, Io.Timestamp.now(self.io, .real).toSeconds()); const threads = try self.corpus.index.search(arena_state.allocator(), &q) orelse return false; _ = self.pager.found_arena.reset(.retain_capacity); const a = self.pager.found_arena.allocator(); self.pager.found_search = try a.dupe(u8, search); // The ids are the index's, which only grows. self.pager.found = try a.dupe([]const u8, threads); 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, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.pager.busy = false; self.readFound(response) catch |err| { log.err("reading what search found: {t}", .{err}); if (err == error.Unauthorized) self.refused(); const out = &self.pager.out; self.desk.paged(out.labels, out.search, null, false, null) catch |e| log.err("list: {t}", .{e}); }; self.askWanted() catch |err| { log.err("reading more of the list: {t}", .{err}); const v = self.pager.view; self.desk.paged(v.labels, v.search, null, false, null) catch |e| log.err("list: {t}", .{e}); };}
fn readFound(self: *JmapSession, response: Http.Error!Http.Response) !void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); const list = try Jmap.replies("reading what search found", arena, response); try self.putThreads(arena, try Jmap.readThreads(arena, list)); try self.showFound(); self.saveSometimes();}
/// Tells the desk the page of what the index found, for the view asked.fn showFound(self: *JmapSession) !void { const out = &self.pager.out; const view = &self.pager.view; const page = self.pager.found_page; if (view.same(out.labels, out.search) and view.position == out.position) { view.position += @intCast(page.len); } const more = out.position + page.len < self.pager.found.len; try self.desk.paged(out.labels, out.search, page, more, self.pager.found_since);}
// The draft.
fn askIdentity(self: *JmapSession) !void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); var r: Jmap.Request = undefined; try r.init(&self.jmap, arena_state.allocator()); try r.call("Identity/get"); try r.field("ids", null); try r.end("identities"); try r.send(self, onIdentity);}
fn onIdentity(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); defer { self.me_settled = true; self.seedContacts(); } var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); const list = Jmap.replies("the identity", arena, response) catch return; const found = Jmap.readIdentity(arena, list) catch return; const id = found orelse return log.warn("the account has no identity to send as", .{}); const a = self.arena_state.allocator(); self.identity = a.dupe(u8, id.id) catch return; self.me = a.dupe(u8, id.email) catch return; self.from = a.dupe(u8, id.from) catch return; self.desk.compose.setMe(self.me) catch |err| log.warn("the address: {t}", .{err}); self.desk.compose.setSignature(id.signature) catch |err| log.warn("the signature: {t}", .{err});}
fn saveDraft(context: *anyopaque, message: Compose.Outgoing) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.keepDraft(message, .save);}
fn sendDraft(context: *anyopaque, message: Compose.Outgoing) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.keepDraft(message, .send);}
fn discardDraft(context: *anyopaque) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.draft.want = .discard; self.draftNext();}
/// Keeps the message as written, to save or send once the step out ends.fn keepDraft(self: *JmapSession, message: Compose.Outgoing, want: DraftStep) void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); var random: [12]u8 = undefined; self.io.random(&random); const domain = if (mem.findScalar(u8, self.me, '@')) |at| self.me[at + 1 ..] else "parcel"; const hex = std.fmt.bytesToHex(random, .lower); const message_id = arena.print("<{s}@{s}>", .{ &hex, domain }) catch return self.draftFailed(want); const raw = mime.message(arena, message, .{ .from = self.from, .date = Io.Timestamp.now(self.io, .real).toSeconds(), .message_id = message_id, }) catch return self.draftFailed(want); std.crypto.secureZero(u8, self.draft.raw.items); self.draft.raw.clearRetainingCapacity(); self.draft.raw.appendSlice(self.gpa, raw) catch return self.draftFailed(want); // A send waiting is not undone by a save after it. if (self.draft.want != .send) self.draft.want = want; self.draftNext();}
fn draftFailed(self: *JmapSession, want: DraftStep) void { log.err("writing the draft", .{}); if (want == .send) self.desk.sent(false) catch |err| log.err("status line: {t}", .{err});}
/// Takes the next step with the draft, if none is out: the message is/// uploaded, then imported to the Drafts mailbox in place of the last/// copy, and sent from there for a send.fn draftNext(self: *JmapSession) void { if (self.draft.busy or !self.roles_read or self.draft.want == .none) return; const want = self.draft.want; self.draft.want = .none; self.draft.busy = true; self.draft.sending = want == .send; var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const result = switch (want) { .discard => self.destroyDraft(arena_state.allocator()), else => self.jmap.upload(arena_state.allocator(), self.draft.raw.items, "message/rfc822", self, onUploaded), }; result catch |err| self.draftDone(err);}
fn destroyDraft(self: *JmapSession, arena: Allocator) !void { if (self.draft.id.items.len == 0) return self.draftDone({}); var r: Jmap.Request = undefined; try r.init(&self.jmap, arena); try r.call("Email/set"); try r.field("destroy", .{self.draft.id.items}); try r.end("destroy"); self.draft.id.clearRetainingCapacity(); try r.send(self, onDraftStep);}
fn onUploaded(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.importDraft(response) catch |err| self.draftDone(err);}
fn importDraft(self: *JmapSession, response: Http.Error!Http.Response) !void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); const r0 = try response; if (r0.status == 401 or r0.status == 403) return error.Unauthorized; if (r0.status / 100 != 2) return error.UploadFailed; const blob = try Jmap.blobId(arena, r0.body); var r: Jmap.Request = undefined; try r.init(&self.jmap, arena); try r.call("Email/import"); try r.s.objectField("emails"); try r.s.beginObject(); try r.s.objectField("draft"); try r.s.beginObject(); try r.field("blobId", blob); try r.s.objectField("mailboxIds"); try r.s.beginObject(); try r.field(if (self.roles.drafts.len > 0) self.roles.drafts else self.roles.inbox, true); try r.s.endObject(); try r.field("keywords", .{ .@"$draft" = true, .@"$seen" = true }); try r.s.endObject(); try r.s.endObject(); try r.end("import"); if (self.draft.sending) { try r.call("EmailSubmission/set"); try r.s.objectField("create"); try r.s.beginObject(); try r.s.objectField("send"); try r.s.write(.{ .identityId = self.identity, .emailId = "#draft" }); try r.s.endObject(); try r.s.objectField("onSuccessUpdateEmail"); try r.s.beginObject(); try r.s.objectField("#send"); try r.s.beginObject(); var buf: [128]u8 = undefined; if (self.roles.drafts.len > 0) try r.field(try mem.print(&buf, "mailboxIds/{s}", .{self.roles.drafts}), null); if (self.roles.sent.len > 0) try r.field(try mem.print(&buf, "mailboxIds/{s}", .{self.roles.sent}), true); try r.field("keywords/$draft", null); try r.s.endObject(); try r.s.endObject(); try r.end("submit"); } if (self.draft.id.items.len > 0) { try r.call("Email/set"); try r.field("destroy", .{self.draft.id.items}); try r.end("destroy"); } try r.send(self, onDraftStep);}
fn onDraftStep(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.readDraftStep(response) catch |err| return self.draftDone(err); self.draftDone({});}
fn readDraftStep(self: *JmapSession, response: Http.Error!Http.Response) !void { var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); const list = try Jmap.replies("keeping the draft", arena, response); const Import = struct { created: ?json.ArrayHashMap(struct { id: []const u8 }) = null, notCreated: ?json.ArrayHashMap(struct { type: []const u8 = "" }) = null, }; self.draft.id.clearRetainingCapacity(); if (Jmap.reply(Import, arena, list, "import")) |got| { if (got.notCreated) |n| if (n.map.count() > 0) { log.err("keeping the draft: {s}", .{n.map.values()[0].type}); return error.NotCreated; }; if (got.created) |c| if (c.map.get("draft")) |d| try self.draft.id.appendSlice(self.gpa, d.id); } else |err| if (err != error.NoReply) return err; if (self.draft.sending) { const Submit = struct { notCreated: ?json.ArrayHashMap(struct { type: []const u8 = "" }) = null }; const got = try Jmap.reply(Submit, arena, list, "submit"); if (got.notCreated) |n| if (n.map.count() > 0) { log.err("sending: {s}", .{n.map.values()[0].type}); return error.NotSent; }; // Sent, it is no draft to replace. self.draft.id.clearRetainingCapacity(); }}
fn draftDone(self: *JmapSession, result: anyerror!void) void { self.draft.busy = false; // The next run reads it back; one sent or thrown away is no more. if (self.settings) |s| { if (self.draft.id.items.len > 0) s.save(draft_file, self.draft.id.items) else s.remove(draft_file); } const sending = self.draft.sending; self.draft.sending = false; if (result) |_| {} else |err| { log.err("the draft: {t}", .{err}); if (err == error.Unauthorized) self.refused() else if (!sending) self.tell("Could not keep the draft on the server; it stays here"); } if (sending) { const ok = if (result) |_| true else |_| false; 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) { self.stopPolling(); self.sync() catch |err| self.failed(err); } } self.draftNext();}
/// Reads back the draft the last run kept, unless compose has one,/// for c to open: the message, then its files one at a time.fn restoreDraft(self: *JmapSession) !void { const settings = self.settings orelse return; if (self.draft.id.items.len > 0 or self.draft.busy or self.draft.want != .none) return; var buf: [256]u8 = undefined; const id = settings.load(draft_file, &buf) orelse return; if (id.len == 0) return; const k = try self.gpa.create(Kept); k.* = .{ .arena = .init(self.gpa), .id = "", .draft = null, .next = 0 }; self.kept = k; const a = k.arena.allocator(); k.id = try a.dupe(u8, id); var r: Jmap.Request = undefined; try r.init(&self.jmap, a); try r.call("Email/get"); try r.field("ids", .{k.id}); try r.field("properties", Jmap.draft_properties); try r.field("fetchTextBodyValues", true); try r.end("draft"); try r.send(self, onKept);}
fn onKept(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.readKept(response) catch |err| { // It stays on the server; the next run tries again. log.err("reading back the draft: {t}", .{err}); self.dropKept(); };}
fn readKept(self: *JmapSession, response: Http.Error!Http.Response) !void { const k = self.kept.?; const a = k.arena.allocator(); const list = try Jmap.replies("reading back the draft", a, response); k.draft = try Jmap.readDraft(a, list) orelse { // Sent or thrown away elsewhere. if (self.settings) |s| s.remove(draft_file); return self.dropKept(); }; try self.gatherKept();}
/// Reads the draft's next file, or offers compose the draft once it has/// them all.fn gatherKept(self: *JmapSession) !void { const k = self.kept.?; const d = k.draft.?; while (k.next < d.blobs.len and d.blobs[k.next].len == 0) k.next += 1; if (k.next < d.blobs.len) return self.jmap.download(k.arena.allocator(), d.blobs[k.next], self, onKeptFile); defer self.dropKept(); if (!try self.desk.compose.restore(d.message)) return; // Saved again, it replaces this one. self.draft.id.clearRetainingCapacity(); try self.draft.id.appendSlice(self.gpa, k.id); self.tell("Your draft is back: c opens it");}
fn onKeptFile(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); self.readKeptFile(response) catch |err| { log.err("reading back the draft's files: {t}", .{err}); self.dropKept(); };}
fn readKeptFile(self: *JmapSession, response: Http.Error!Http.Response) !void { const k = self.kept.?; const r = try response; if (r.status / 100 != 2) { log.err("reading back the draft's files: {d}", .{r.status}); return error.DownloadFailed; } k.draft.?.attachments[k.next].data = try k.arena.allocator().dupe(u8, r.body); k.next += 1; try self.gatherKept();}
fn dropKept(self: *JmapSession) void { const k = self.kept orelse return; k.arena.deinit(); self.gpa.destroy(k); self.kept = null;}
// Keeping time.
fn schedule(self: *JmapSession) void { if (self.poll != null) return; const ms: u32 = if (self.push.live) push_poll_ms else poll_ms; self.poll = siv.timerStart(self.app, ms, onPoll, null, self) catch |err| blk: { log.err("setting the poll's timer: {t}", .{err}); break :blk null; };}
fn stopPolling(self: *JmapSession) void { if (self.poll) |t| siv.timerCancel(t); self.poll = null;}
fn onPoll(user_data: ?*anyopaque) callconv(.c) void { const self: *JmapSession = @ptrCast(@alignCast(user_data)); self.poll = null; if (self.loading) return; // Never reached, as when the network was down at start. if (!self.ready) { const arena = self.load_arena.allocator(); self.jmap.askSession(arena, self.session_url, self, onSession) catch |err| return self.failed(err); self.loading = true; return; } self.sync() catch |err| self.failed(err);}
/// Opens the server's event stream, unless it is open, waits to be/// opened again, or the server has none.fn listen(self: *JmapSession) void { if (self.push.open or self.push.timer != null or self.jmap.event_source_url.len == 0) return; var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); self.jmap.listen(arena_state.allocator(), self, onHeard, onUnheard) catch |err| { log.err("listening for changes: {t}", .{err}); return self.relisten(); }; self.push.open = true;}
fn stopListening(self: *JmapSession) void { if (self.push.timer) |t| siv.timerCancel(t); self.push.timer = null;}
fn onHeard(context: *anyopaque, bytes: []const u8) void { const self: *JmapSession = @ptrCast(@alignCast(context)); // What changed while the stream was down is read at once, unless the // first read is still to end; the poll waits longer from now. var read = false; if (!self.push.live) { self.push.live = true; self.push.ended = 0; self.stopPolling(); self.schedule(); read = self.roles_read; } const changed = self.hear(bytes) catch |err| blk: { log.warn("an event: {t}", .{err}); self.push.buf.clearRetainingCapacity(); break :blk false; }; if (read or changed) self.pushed();}
/// Reads the whole events in what has come, and whether one says the/// mail changed; the rest waits for more.fn hear(self: *JmapSession, bytes: []const u8) !bool { const buf = &self.push.buf; // No event is this long: what came is not one. if (buf.items.len + bytes.len > 1 << 20) return error.EventTooLong; try buf.appendSlice(self.gpa, bytes); var arena_state: ArenaAllocator = .init(self.gpa); defer arena_state.deinit(); const arena = arena_state.allocator(); var used: usize = 0; var changed = false; while (try Jmap.nextEvent(arena, buf.items[used..])) |next| { used += next.len; if (self.jmap.changed(arena, next.event)) changed = true; } buf.replaceRangeAssumeCapacity(0, used, &.{}); return changed;}
/// Reads what changed now, or once the read under way ends.fn pushed(self: *JmapSession) void { if (self.loading or !self.ready) { self.again = true; return; } self.stopPolling(); self.sync() catch |err| self.failed(err);}
fn onUnheard(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *JmapSession = @ptrCast(@alignCast(context)); if (response) |r| { if (r.status / 100 == 2) log.info("the event stream ended", .{}) else log.warn("the event stream: {d}", .{r.status}); } else |err| log.warn("the event stream ended: {t}", .{err}); self.push.open = false; self.push.buf.clearRetainingCapacity(); if (self.push.live) { // Back to polling every minute meanwhile. self.push.live = false; self.stopPolling(); if (!self.loading) self.schedule(); } self.relisten();}
fn relisten(self: *JmapSession) void { const ms = relisten_ms[@min(self.push.ended, relisten_ms.len - 1)]; self.push.ended +|= 1; self.push.timer = siv.timerStart(self.app, ms, onRelisten, null, self) catch |err| blk: { log.err("setting the event stream's timer: {t}", .{err}); break :blk null; };}
fn onRelisten(user_data: ?*anyopaque) callconv(.c) void { const self: *JmapSession = @ptrCast(@alignCast(user_data)); self.push.timer = null; self.listen();}
/// Keeps the desk's mail for the next run, once it has come.fn save(self: *JmapSession) void { const settings = self.settings orelse return; if (self.desk.mail.state.len == 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);}
fn saveSometimes(self: *JmapSession) void { if (self.saved_at) |at| { if (at.durationTo(.now(self.io, .awake)).toSeconds() < save_every_s) return; } self.save();}
/// Says in the status line that the mail did not load, unless it was/// said last time, and tries again later. The details are in the log.fn failed(self: *JmapSession, err: anyerror) void { log.err("loading mail: {t}", .{err}); _ = self.load_arena.reset(.free_all); self.desk.loading = false; if (err == error.Unauthorized) self.refused() else if (!self.offline) self.tell("Could not load mail from the server"); self.offline = true; self.schedule();}
fn refused(self: *JmapSession) void { self.tell("The server refused the saved token: run `parcel login` for it again");}
/// A URL's host, which the status line names a server other than/// Fastmail by.fn host(url: []const u8) []const u8 { const rest = url[if (mem.find(u8, url, "://")) |i| i + 3 else 0..]; return rest[0 .. mem.findAny(u8, rest, ":/") orelse rest.len];}
fn tell(self: *JmapSession, comptime message: []const u8) void { self.desk.tell(message, .{}) catch |err| log.err("status line: {t}", .{err});}