Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393//! Goes down Gmail's list of the threads a query shows, a page of ids at//! a time, and reads the threads among them a mailbox wants: those it//! lacks, or has an older copy of. It keeps its place between calls, so//! each call goes on where the last stopped. The list as the user//! scrolls it (Session.zig) and the backfill (Backfill.zig) each have//! one.//!//! A page of ids costs 10 quota units however long, so pages are as long//! as Gmail allows; a thread read costs 40, so a call reads few.
const std = @import("std");const siv = @import("siv");const Gmail = @import("Gmail.zig");const Mailbox = @import("Mailbox.zig");const assert = std.debug.assert;const testing = std.testing;const Allocator = std.mem.Allocator;const ArenaAllocator = std.heap.ArenaAllocator;const ArrayList = std.ArrayList;
const Pager = @This();
gpa: Allocator,/// The account, whose authorization changes when the token is/// refreshed.gmail: *const Gmail,/// Holds `query`.query_arena: ArenaAllocator,query: Gmail.Query,/// Holds the place: `page`, `next`, and `rest`.place_arena: ArenaAllocator,/// The token of the page `rest` is from; empty for the first.page: []const u8,/// The token of the page after; empty for the first, null past the end.next: ?[]const u8,/// The threads of the page not gone through yet.rest: []const Gmail.Listed,/// The call out; one at a time.call: ?*Call,
/// What a mailbox wants read of the threads a list names.pub const Want = enum { /// Those it lacks. unknown, /// Those it lacks, and those it has an older copy of. unknown_or_changed, /// Only those it has an older copy of. changed,
fn wants(want: Want, known: *const Mailbox, listed: Gmail.Listed) bool { const id = known.findThread(listed.id) orelse return want != .changed; return want != .unknown and known.get(id).version < listed.version; }};
pub const Options = struct { /// The most threads read in one call, 40 quota units each. reads: u16, /// The most pages of ids gone down in one call, 10 units each. pages: u16, want: Want,};
/// What a call went through.pub const Page = struct { /// Gmail's ids of the threads gone through, newest first: those read, /// and those the mailbox did not want. ids: []const []const u8, threads: []const Gmail.Thread, /// Whether Gmail's list goes on past them. more: bool,};
/// Hears how a call went. What it is given lasts until it returns. A/// failed call leaves the place where it was, to try again.pub const Paged = *const fn (context: *anyopaque, result: anyerror!Page) void;
/// The most ids Gmail gives in a page.const page_size = 500;
pub fn init(gpa: Allocator, gmail: *const Gmail) Pager { return .{ .gpa = gpa, .gmail = gmail, .query_arena = .init(gpa), .query = .{ .labels = &.{}, .search = "" }, .place_arena = .init(gpa), .page = "", .next = "", .rest = &.{}, .call = null, };}
/// Drops the call out, if any; deinit the Http first, so no reply comes/// after.pub fn deinit(self: *Pager) void { if (self.call) |c| c.destroy(); self.place_arena.deinit(); self.query_arena.deinit(); self.* = undefined;}
/// Whether a call is out.pub fn busy(self: *const Pager) bool { return self.call != null;}
/// Starts down `query`'s list at the page `from` names, or the first/// when it is empty. Not while a call is out.pub fn reset(self: *Pager, query: Gmail.Query, from: []const u8) Allocator.Error!void { assert(self.call == null); _ = self.query_arena.reset(.retain_capacity); const a = self.query_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.query = .{ .labels = labels, .search = try a.dupe(u8, query.search) }; _ = self.place_arena.reset(.retain_capacity); self.page = try self.place_arena.allocator().dupe(u8, from); self.next = self.page; self.rest = &.{};}
/// Whether the list goes on past what was gone through.pub fn more(self: *const Pager) bool { return self.rest.len > 0 or self.next != null;}
/// The token of the page to go on from in another run; empty for the/// first.pub fn resumeFrom(self: *const Pager) []const u8 { return self.page;}
/// Goes on down the list, reading what `known` wants, and `done` hears/// what was gone through: before this returns only when Gmail need not/// be asked, as at the end of the list. `known` must outlast the call.pub fn goOn(self: *Pager, known: *const Mailbox, options: Options, context: *anyopaque, done: Paged) !void { assert(self.call == null); assert(options.reads > 0); const c = try self.gpa.create(Call); c.* = .{ .pager = self, .arena_state = .init(self.gpa), .known = known, .options = options, .context = context, .done = done, .page = self.page, .next = self.next, .rest = self.rest, .ids = .empty, .wanted = .empty, .pages = 0, }; errdefer c.destroy(); self.call = c; errdefer self.call = null; c.step() catch |err| switch (err) { error.NothingToAsk => c.finish(&.{}), else => return err, };}
/// A call under way, with the place it has got to, which becomes the/// pager's once it succeeds.const Call = struct { pager: *Pager, arena_state: ArenaAllocator, known: *const Mailbox, options: Options, context: *anyopaque, done: Paged, page: []const u8, next: ?[]const u8, rest: []const Gmail.Listed, ids: ArrayList([]const u8), wanted: ArrayList([]const u8), /// Pages listed so far. pages: u16,
fn destroy(c: *Call) void { const gpa = c.pager.gpa; c.arena_state.deinit(); gpa.destroy(c); }
/// Goes through the rest of the page, then asks for what is wanted, /// or for the next page when nothing was; else ends. fn step(c: *Call) !void { const arena = c.arena_state.allocator(); while (c.rest.len > 0 and c.wanted.items.len < c.options.reads) { const next = c.rest[0]; c.rest = c.rest[1..]; const id = try arena.dupe(u8, next.id); try c.ids.append(arena, id); if (c.options.want.wants(c.known, next)) try c.wanted.append(arena, id); } const gmail = c.pager.gmail; if (c.wanted.items.len > 0) return gmail.read(arena, c.wanted.items, c, onRead); const next = c.next orelse return error.NothingToAsk; if (c.rest.len > 0 or c.pages == c.options.pages) return error.NothingToAsk; c.pages += 1; try gmail.list(arena, c.pager.query, next, page_size, c, onList); }
fn onList(context: *anyopaque, result: anyerror!Gmail.List) void { const c: *Call = @ptrCast(@alignCast(context)); c.listed(result) catch |err| switch (err) { error.NothingToAsk => c.finish(&.{}), else => c.fail(err), }; }
fn listed(c: *Call, result: anyerror!Gmail.List) !void { const list = try result; c.page = c.next.?; c.next = list.next; c.rest = list.threads; try c.step(); }
fn onRead(context: *anyopaque, result: anyerror![]const Gmail.Thread) void { const c: *Call = @ptrCast(@alignCast(context)); const threads = result catch |err| return c.fail(err); c.finish(threads); }
fn fail(c: *Call, err: anyerror) void { const p = c.pager; p.call = null; defer c.destroy(); c.done(c.context, err); }
/// Makes the place the pager's, then tells what was gone through, /// which `done` may follow with another call. fn finish(c: *Call, threads: []const Gmail.Thread) void { const p = c.pager; p.call = null; defer c.destroy(); p.keepPlace(c.page, c.next, c.rest) catch |err| return c.done(c.context, err); c.done(c.context, .{ .ids = c.ids.items, .threads = threads, .more = p.more() }); }};
/// Copies the place a call got to, which its arena holds, into the/// pager's own.fn keepPlace(self: *Pager, page: []const u8, next: ?[]const u8, rest: []const Gmail.Listed) Allocator.Error!void { var fresh: ArenaAllocator = .init(self.gpa); errdefer fresh.deinit(); const a = fresh.allocator(); const kept = try a.alloc(Gmail.Listed, rest.len); for (rest, kept) |l, *out| out.* = .{ .id = try a.dupe(u8, l.id), .version = l.version }; const kept_page = try a.dupe(u8, page); const kept_next = if (next) |n| try a.dupe(u8, n) else null; self.place_arena.deinit(); self.place_arena = fresh; self.page = kept_page; self.next = kept_next; self.rest = kept;}
test "a pager reads what the mailbox lacks, a few at a time, and goes on from there" { var fake: Gmail.Fake = undefined; try fake.start(&routes); defer fake.stop(); var mail: Mailbox = try .init(testing.allocator); defer mail.deinit(); // The mailbox has b as it is, and d older than Gmail's. for ([_][]const u8{ "b", "d" }, [_]u64{ 2, 1 }) |id, version| { _ = try mail.put(.{ .id = id, .participants = "", .subject = "", .snippet = "", .date = 0, .messages = 1, .labels = &.{}, .unread = false, .notes = &.{}, }, version); } var pager: Pager = .init(testing.allocator, &fake.gmail); defer pager.deinit(); try pager.reset(.{ .labels = &.{"INBOX"}, .search = "" }, ""); const user: Options = .{ .reads = 2, .pages = 3, .want = .unknown };
// a and c: b is known, so it is gone through but not read. try expectPage(&fake, &pager, &mail, user, &.{ "a", "b", "c" }, &.{ "a", "c" }, true); // On to page 2, past the d it has: e. try expectPage(&fake, &pager, &mail, user, &.{ "d", "e" }, &.{"e"}, false); try testing.expect(!pager.more()); try testing.expectEqualStrings("p2", pager.resumeFrom());
// The backfill's way: one at a time, the changed too. Resumed at page // 2, it reads d again, as Gmail's is newer. try pager.reset(.{ .labels = &.{"INBOX"}, .search = "" }, "p2"); const backfill: Options = .{ .reads = 1, .pages = 1, .want = .unknown_or_changed }; try expectPage(&fake, &pager, &mail, backfill, &.{"d"}, &.{"d"}, true); try testing.expectEqual(5, mail.get(mail.findThread("d").?).version); // e is as Gmail has it: gone through, not read, and the list ends. try expectPage(&fake, &pager, &mail, backfill, &.{"e"}, &.{}, false);
// A failed read leaves the place as it was. try pager.reset(.{ .labels = &.{"Label_1"}, .search = "" }, ""); var heard: Heard = .{ .mail = &mail }; defer heard.deinit(); try pager.goOn(&mail, user, &heard, Heard.put); try testing.expectError(error.Unauthorized, heard.wait(&fake)); try testing.expectEqualStrings("", pager.resumeFrom()); try testing.expect(pager.more() and pager.rest.len == 0);}
fn expectPage( fake: *Gmail.Fake, pager: *Pager, mail: *Mailbox, options: Options, ids: []const []const u8, read: []const []const u8, more_after: bool,) !void { var heard: Heard = .{ .mail = mail }; defer heard.deinit(); try pager.goOn(mail, options, &heard, Heard.put); try heard.wait(fake); try testing.expectEqual(ids.len, heard.ids.items.len); for (ids, heard.ids.items) |want, got| try testing.expectEqualStrings(want, got); try testing.expectEqual(read.len, heard.read.items.len); for (read, heard.read.items) |want, got| try testing.expectEqualStrings(want, got); try testing.expectEqual(more_after, heard.more.?);}
/// What a call heard, kept past it: the ids, and the threads read, which/// go in the mailbox as the session puts them.const Heard = struct { mail: *Mailbox, arena: ArenaAllocator = .init(testing.allocator), ids: ArrayList([]const u8) = .empty, read: ArrayList([]const u8) = .empty, more: ?bool = null, failure: ?anyerror = null,
fn deinit(self: *Heard) void { self.arena.deinit(); self.* = undefined; }
fn put(context: *anyopaque, result: anyerror!Page) void { const self: *Heard = @ptrCast(@alignCast(context)); self.keep(result) catch |err| { self.failure = err; }; }
fn keep(self: *Heard, result: anyerror!Page) !void { const page = try result; const a = self.arena.allocator(); for (page.ids) |id| try self.ids.append(a, try a.dupe(u8, id)); for (page.threads) |t| try self.read.append(a, try a.dupe(u8, t.id)); try Gmail.put(self.mail, page.threads); self.more = page.more; }
fn wait(self: *Heard, fake: *Gmail.Fake) !void { var waited: u32 = 0; while (self.more == null and self.failure == null and waited < 100) : (waited += 1) { try siv.appDispatch(fake.app, 100); } if (self.failure) |err| return err; if (self.more == null) return error.Timeout; }};
const list_path = "/threads?maxResults=500&fields=threads(id,historyId),nextPageToken";
const routes = [_]Gmail.Fake.Route{ .ok(list_path ++ "&labelIds=INBOX&pageToken=p2", \\{"threads":[{"id":"d","historyId":"5"},{"id":"e","historyId":"6"}]} ), .ok(list_path ++ "&labelIds=INBOX", \\{"threads":[{"id":"a","historyId":"3"},{"id":"b","historyId":"2"},{"id":"c","historyId":"4"}], \\"nextPageToken":"p2"} ), .ok(list_path ++ "&labelIds=Label_1", "{\"threads\":[{\"id\":\"broken\",\"historyId\":\"1\"}]}"), .ok("/threads/a?", Gmail.Fake.thread("a", "INBOX", 3)), .ok("/threads/c?", Gmail.Fake.thread("c", "INBOX", 4)), .ok("/threads/d?", Gmail.Fake.thread("d", "INBOX", 5)), .ok("/threads/e?", Gmail.Fake.thread("e", "INBOX", 6)), .{ .path = "/threads/broken?", .body = "{}", .status = .unauthorized },};