Something went wrong. Try again.
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021//! The messages kept for search: what each says, apart from what it//! quotes, and what Gmail's operators ask of it (who, when, which list,//! its files, its size). Each is a record appended to `file_name` in the//! account's directory as the backfill reads it, and the file is never//! rewritten, so a backfill of hundreds of megabytes costs only what it//! adds. Labels and flags, which change, are not kept here.//!//! The file is read by a job off the app's thread as the session starts;//! until it has been, `ready` is false and the backfill waits. A record//! cut short, as by a crash, is cut off; a file of another layout is//! started over. Only the user may read it: it holds their mail.
const std = @import("std");const siv = @import("siv");const Contacts = @import("Contacts.zig");const Index = @import("Index.zig");const Query = @import("Query.zig");const Segment = @import("Segment.zig");const Settings = @import("Settings.zig");const ascii = std.ascii;const mem = std.mem;const Allocator = std.mem.Allocator;const ArenaAllocator = std.heap.ArenaAllocator;const Io = std.Io;const Writer = Io.Writer;const log = std.log.scoped(.corpus);const testing = std.testing;
const Corpus = @This();
gpa: Allocator,io: Io,/// Null until the file is read, or with nowhere to keep it, as in tests.file: ?Io.File,/// Where the next record goes: the end of the last whole one.end: u64,/// The messages kept that the index's segment does not hold, by their/// backend's ids, which `ids_arena` holds, and their places in the index.ids: std.array_hash_map.String(u32),ids_arena: ArenaAllocator,/// The words of the messages kept, for search.index: Index,/// Whether the backfill has gone through every message once.walked: bool,/// How far down the backfill has kept every message, newest first, in/// seconds since the epoch: when the oldest of the last page it kept/// whole came. maxInt before it has kept a page.reached: i64,/// Whether the file has been read, so `has` can be trusted.ready: bool,/// The job reading the file, while it does.load: ?*Load,/// The job gathering the people on the messages, while it does.gather: ?*Gather,/// Messages gone from the account before the file was read, to drop/// once it has been.doomed: std.ArrayList([]u8),on_ready: Ready,
pub const file_name = "corpus";
/// Starts the file, and changes when its layout does.const magic = "parcel corpus\x00" ++ "\x01\x00\x00\x00";
/// How many messages past the segment's are indexed in memory as the/// corpus loads before the segment is written again with them.const rebuild_after = 5000;
/// Hears that the file has been read.pub const Ready = struct { context: *anyopaque, call: *const fn (context: *anyopaque) void,};
pub const Person = struct { name: []const u8, address: []const u8 };
pub const Attachment = struct { name: []const u8, mime: []const u8, size: u32 };
/// A message as search keeps it.pub const Record = struct { /// Its backend's ids, and its Message-ID. id: []const u8, thread: []const u8, message_id: []const u8, subject: []const u8, /// Its List-Id and Delivered-To headers, for list: and deliveredto:. list_id: []const u8, delivered_to: []const u8, /// When it arrived, in seconds since the epoch, and its size in bytes. date: i64, size: u32, from: Person, to: []const Person, cc: []const Person, bcc: []const Person, attachments: []const Attachment, /// What it says itself, and what it quotes of earlier messages. text: []const u8, quoted: []const u8,};
/// A record: a message, the backfill's end, a message gone since, or/// how far down the backfill has kept every message.const Kind = enum(u8) { message = 1, walked = 2, gone = 3, reached = 4, _ };
/// How much mail the index holds before search asks it rather than/// the server: a month back, or all of it.const enough_s = 30 * std.time.s_per_day;
/// What mail search may ask the index for: all of it, or what came/// since a time, in seconds since the epoch, as older mail is not kept/// yet.pub const Searchable = union(enum) { all, since: i64 };
/// Starts reading the corpus kept in `settings`, if any; `on_ready`/// hears once it has been, and `has` is right from then. Without/// settings, it keeps nothing and is ready at once.pub fn open(self: *Corpus, gpa: Allocator, io: Io, app: siv.App, settings: ?Settings, on_ready: Ready) void { self.* = .{ .gpa = gpa, .io = io, .file = null, .end = 0, .ids = .empty, .ids_arena = .init(gpa), .index = .init(gpa), .walked = false, .reached = std.math.maxInt(i64), .ready = false, .load = null, .gather = null, .doomed = .empty, .on_ready = on_ready, }; const s = settings orelse { self.ready = true; return; }; self.startLoad(app, s) catch |err| { log.warn("{s}: {t}; search keeps nothing this run", .{ file_name, err }); self.ready = true; };}
pub fn deinit(self: *Corpus) void { // The job's file and findings go with it. if (self.load) |l| l.corpus = null; if (self.gather) |g| g.corpus = null; if (self.file) |f| f.close(self.io); self.ids.deinit(self.gpa); self.ids_arena.deinit(); self.index.deinit(self.gpa); for (self.doomed.items) |id| self.gpa.free(id); self.doomed.deinit(self.gpa); self.* = undefined;}
/// Whether message `id` is kept.pub fn has(self: *const Corpus, id: []const u8) bool { const doc = self.place(id) orelse return false; return !self.index.isGone(doc);}
/// The place in the index of message `id`, kept or since gone.fn place(self: *const Corpus, id: []const u8) ?u32 { if (self.ids.get(id)) |doc| return doc; const seg = self.index.segment orelse return null; return seg.findMessage(id);}
pub fn count(self: *const Corpus) u32 { return self.index.count();}
/// Keeps `r`, if not already. A failure to write is returned; the/// message is not kept then.pub fn add(self: *Corpus, r: Record) !void { if (self.has(r.id)) return; var out: Writer.Allocating = .init(self.gpa); defer out.deinit(); try encode(&out.writer, r); const bytes = out.written(); mem.writeInt(u32, bytes[0..4], @intCast(bytes.len - 4), .little); try self.append(bytes); const doc = try self.index.add(self.gpa, r); try self.ids.put(self.gpa, try self.ids_arena.allocator().dupe(u8, r.id), doc);}
/// Drops message `id`, gone from the account, if it was kept: a record/// says so, and search leaves it out.pub fn remove(self: *Corpus, id: []const u8) !void { if (!self.ready) { const copy = try self.gpa.dupe(u8, id); errdefer self.gpa.free(copy); return self.doomed.append(self.gpa, copy); } const doc = self.place(id) orelse return; if (self.index.isGone(doc)) return; var out: Writer.Allocating = .init(self.gpa); defer out.deinit(); try out.writer.writeInt(u32, @intCast(1 + 4 + id.len), .little); try out.writer.writeByte(@backingInt(Kind.gone)); try string(&out.writer, id); try self.append(out.written()); _ = self.ids.swapRemove(id); self.index.drop(doc);}
/// Marks the backfill as having gone through every message.pub fn markWalked(self: *Corpus) !void { if (self.walked) return; try self.append(&.{ 1, 0, 0, 0, @backingInt(Kind.walked) }); self.walked = true;}
/// Notes that the backfill has kept every message from the newest down/// to the page `ids`, in the order the server lists them, newest first.pub fn reach(self: *Corpus, ids: []const []const u8) !void { var oldest: i64 = std.math.maxInt(i64); for (ids) |id| if (self.place(id)) |doc| { oldest = @min(oldest, self.index.dateOf(doc)); }; if (oldest >= self.reached) return; var record: [5 + 8]u8 = undefined; record[0..5].* = .{ 9, 0, 0, 0, @backingInt(Kind.reached) }; mem.writeInt(i64, record[5..], oldest, .little); try self.append(&record); self.reached = oldest;}
/// What mail search may ask the index for at `now`; null until it holds/// a month of it.pub fn searchable(self: *const Corpus, now: i64) ?Searchable { if (self.walked) return .all; if (self.reached > now - enough_s) return null; return .{ .since = self.reached };}
fn append(self: *Corpus, bytes: []const u8) !void { const f = self.file orelse return; try f.writePositionalAll(self.io, bytes, self.end); self.end += bytes.len;}
/// A message's record, as the file holds it but with its length, which/// comes first, left for the caller to fill in.fn encode(w: *Writer, r: Record) !void { try w.writeInt(u32, 0, .little); try w.writeByte(@backingInt(Kind.message)); for ([_][]const u8{ r.id, r.thread, r.message_id, r.subject, r.list_id, r.delivered_to }) |s| try string(w, s); try w.writeInt(i64, r.date, .little); try w.writeInt(u32, r.size, .little); try person(w, r.from); for ([_][]const Person{ r.to, r.cc, r.bcc }) |list| { try w.writeInt(u32, @intCast(list.len), .little); for (list) |p| try person(w, p); } try w.writeInt(u32, @intCast(r.attachments.len), .little); for (r.attachments) |a| { try string(w, a.name); try string(w, a.mime); try w.writeInt(u32, a.size, .little); } try string(w, r.text); try string(w, r.quoted);}
fn string(w: *Writer, s: []const u8) !void { try w.writeInt(u32, @intCast(s.len), .little); try w.writeAll(s);}
fn person(w: *Writer, p: Person) !void { try string(w, p.name); try string(w, p.address);}
/// The records of a file's bytes, after its magic, one at a time; a/// record cut short ends them.pub const Records = struct { bytes: []const u8, at: usize,
pub fn init(bytes: []const u8) Records { return .{ .bytes = bytes, .at = if (mem.startsWith(u8, bytes, magic)) magic.len else bytes.len }; }
/// The next record's kind and body; null past the last whole one. fn next(it: *Records) ?struct { kind: Kind, body: []const u8 } { if (it.bytes.len - it.at < 4) return null; const len = mem.readInt(u32, it.bytes[it.at..][0..4], .little); if (len == 0 or it.bytes.len - it.at - 4 < len) return null; const whole = it.bytes[it.at + 4 ..][0..len]; it.at += 4 + len; return .{ .kind = @fromBackingInt(@intCast(whole[0])), .body = whole[1..] }; }
/// The next message, its lists in `arena`; null past the last. pub fn message(it: *Records, arena: Allocator) !?Record { while (it.next()) |r| { if (r.kind != .message) continue; const m = try decode(arena, r.body); return m; } return null; }};
const Fields = struct { bytes: []const u8, at: usize = 0,
fn take(f: *Fields, n: usize) ![]const u8 { if (f.bytes.len - f.at < n) return error.Truncated; defer f.at += n; return f.bytes[f.at..][0..n]; }
fn int(f: *Fields, comptime T: type) !T { return mem.readInt(T, (try f.take(@sizeOf(T)))[0..@sizeOf(T)], .little); }
fn string(f: *Fields) ![]const u8 { return f.take(try f.int(u32)); }
fn person(f: *Fields) !Person { return .{ .name = try f.string(), .address = try f.string() }; }
fn people(f: *Fields, arena: Allocator) ![]const Person { const out = try arena.alloc(Person, try f.int(u32)); for (out) |*p| p.* = try f.person(); return out; }};
fn decode(arena: Allocator, bytes: []const u8) !Record { var f: Fields = .{ .bytes = bytes }; var r: Record = undefined; r.id = try f.string(); r.thread = try f.string(); r.message_id = try f.string(); r.subject = try f.string(); r.list_id = try f.string(); r.delivered_to = try f.string(); r.date = try f.int(i64); r.size = try f.int(u32); r.from = try f.person(); r.to = try f.people(arena); r.cc = try f.people(arena); r.bcc = try f.people(arena); const files = try arena.alloc(Attachment, try f.int(u32)); for (files) |*a| a.* = .{ .name = try f.string(), .mime = try f.string(), .size = try f.int(u32) }; r.attachments = files; r.text = try f.string(); r.quoted = try f.string(); return r;}
/// What a message's plain text says itself and what it quotes: lines/// starting with ">", and the line before them that says who wrote/// them ("On … wrote:"), are quoted, without their marks.pub fn split(arena: Allocator, text: []const u8) !struct { said: []const u8, quoted: []const u8 } { var said: std.ArrayList(u8) = .empty; var quoted: std.ArrayList(u8) = .empty; var lines = mem.splitScalar(u8, text, '\n'); // A line saying who wrote what follows, and the blank lines after it, // until a line shows whether quoted lines do follow. var held: ?[]const u8 = null; var blanks: u32 = 0; while (lines.next()) |raw| { const line = mem.trimEnd(u8, raw, "\r"); const trimmed = mem.trim(u8, line, " \t"); if (held != null and trimmed.len == 0) { blanks += 1; continue; } if (trimmed.len > 0 and trimmed[0] == '>') { if (held) |h| try addLine(arena, "ed, h); held = null; try addLine(arena, "ed, mem.trimStart(u8, trimmed, "> \t")); continue; } try unhold(arena, &said, &held, &blanks); if (mem.endsWith(u8, trimmed, "wrote:")) { held = line; continue; } try addLine(arena, &said, line); } try unhold(arena, &said, &held, &blanks); return .{ .said = mem.trim(u8, said.items, " \n"), .quoted = mem.trim(u8, quoted.items, " \n") };}
/// Says the held line, which no quote followed, and the blanks after it.fn unhold(arena: Allocator, said: *std.ArrayList(u8), held: *?[]const u8, blanks: *u32) !void { const h = held.* orelse return; try addLine(arena, said, h); for (0..blanks.*) |_| try addLine(arena, said, ""); held.* = null; blanks.* = 0;}
fn addLine(arena: Allocator, out: *std.ArrayList(u8), line: []const u8) !void { if (out.items.len > 0) try out.append(arena, '\n'); try out.appendSlice(arena, line);}
/// Notes the people on `r` in `contacts`: when it is from `me`, those it/// went to as written to, else its sender as heard from and those it went/// to as copied. `me` is never noted; empty, no one is it.pub fn notePeople(contacts: *Contacts, r: Record, me: []const u8) !void { const ms = r.date * 1000; const mine = me.len > 0 and ascii.eqlIgnoreCase(r.from.address, me); if (!mine) try contacts.noteAddress(.heard, r.from.name, r.from.address, ms); for ([_][]const Person{ r.to, r.cc }) |list| for (list) |p| { if (me.len > 0 and ascii.eqlIgnoreCase(p.address, me)) continue; try contacts.noteAddress(if (mine) .wrote else .copied, p.name, p.address, ms); };}
/// Hears the people `gatherPeople` found, to take what it wants of them.pub const People = struct { context: *anyopaque, call: *const fn (context: *anyopaque, people: *const Contacts) void,};
/// Reads every message kept for the people on it, as `notePeople` notes/// them for `me`, off the app's thread; `done` hears them. Nothing while/// it does already or without a file, as in tests.pub fn gatherPeople(self: *Corpus, app: siv.App, settings: Settings, me: []const u8, done: People) !void { if (self.file == null or self.gather != null) return; // Its own, which the job closes, as the corpus may go first. const file = try settings.dir.openFile(self.io, file_name, .{}); errdefer file.close(self.io); const g = try self.gpa.create(Gather); errdefer self.gpa.destroy(g); const own_me = try self.gpa.dupe(u8, me); errdefer self.gpa.free(own_me); g.* = .{ .corpus = self, .gpa = self.gpa, .io = self.io, .file = file, .end = self.end, .me = own_me, .people = .init(self.gpa), .done = done, .failed = null, }; try siv.jobStart(app, .{}, gatherWork, gatherDone, gatherDestroy, g); self.gather = g;}
/// The job gathering the people: the corpus it is for (null once that is/// gone), its own file to as far as the corpus had kept, and who it found.const Gather = struct { corpus: ?*Corpus, gpa: Allocator, io: Io, file: Io.File, end: u64, me: []u8, people: Contacts, done: People, failed: ?anyerror,};
fn gatherWork(user_data: ?*anyopaque) callconv(.c) void { const g: *Gather = @ptrCast(@alignCast(user_data)); gatherRead(g) catch |err| { g.failed = err; };}
fn gatherRead(g: *Gather) !void { var buf: [256 * 1024]u8 = undefined; var fr = g.file.reader(g.io, &buf); try fr.seekTo(magic.len); const r = &fr.interface; var body: std.ArrayList(u8) = .empty; defer body.deinit(g.gpa); var record_arena: ArenaAllocator = .init(g.gpa); defer record_arena.deinit(); var at: u64 = magic.len; while (g.end - at >= 4) { const len = try r.takeInt(u32, .little); if (len == 0 or g.end - at - 4 < len) break; try body.resize(g.gpa, len); try r.readSliceAll(body.items); at += 4 + len; if (body.items[0] != @backingInt(Kind.message)) continue; _ = record_arena.reset(.retain_capacity); try notePeople(&g.people, try decode(record_arena.allocator(), body.items[1..]), g.me); }}
fn gatherDone(user_data: ?*anyopaque) callconv(.c) void { const g: *Gather = @ptrCast(@alignCast(user_data)); const self = g.corpus orelse return; self.gather = null; if (g.failed) |err| return log.warn("gathering the people on the mail: {t}", .{err}); g.done.call(g.done.context, &g.people);}
fn gatherDestroy(user_data: ?*anyopaque) callconv(.c) void { const g: *Gather = @ptrCast(@alignCast(user_data)); g.file.close(g.io); g.people.deinit(); g.gpa.free(g.me); g.gpa.destroy(g);}
// Reading the file, off the app's thread.
/// The job reading the file: the corpus it is for (null once that is/// gone), the file, and what it found.const Load = struct { corpus: ?*Corpus, gpa: Allocator, io: Io, /// Null once the corpus takes it. file: ?Io.File, /// The account's directory, which holds the segment. dir: Io.Dir, ids: std.array_hash_map.String(u32), arena: ArenaAllocator, /// Whether the segment was written again. rebuilt: bool, /// The end of the last whole record, and whether the file must be /// started over: empty, or of another layout. end: u64, fresh: bool, walked: bool, reached: i64, /// The words of what it found, and how long they took to index. index: Index, index_ms: i64, /// The file's size. bytes: u64, failed: ?anyerror,};
fn startLoad(self: *Corpus, app: siv.App, settings: Settings) !void { const file = try settings.dir.createFile(self.io, file_name, .{ .read = true, .truncate = false, .permissions = @fromBackingInt(0o600), }); errdefer file.close(self.io); const l = try self.gpa.create(Load); errdefer self.gpa.destroy(l); l.* = .{ .corpus = self, .gpa = self.gpa, .io = self.io, .file = file, .dir = settings.dir, .ids = .empty, .arena = .init(self.gpa), .rebuilt = false, .end = 0, .fresh = true, .walked = false, .reached = std.math.maxInt(i64), .bytes = 0, .index = .init(self.gpa), .index_ms = 0, .failed = null, }; try siv.jobStart(app, .{}, loadWork, loadDone, loadDestroy, l); self.load = l;}
fn loadWork(user_data: ?*anyopaque) callconv(.c) void { const l: *Load = @ptrCast(@alignCast(user_data)); read(l) catch |err| { l.failed = err; };}
fn read(l: *Load) !void { const file = l.file.?; const size = try file.length(l.io); l.bytes = size; var head: [magic.len]u8 = undefined; const whole = size >= magic.len and try file.readPositionalAll(l.io, &head, 0) == magic.len; if (!whole or !mem.eql(u8, &head, magic)) { // The corpus starts over, and an index of the old one would not // be of the new one, however long it grows. l.dir.deleteFile(l.io, Segment.file_name) catch |err| switch (err) { error.FileNotFound => {}, else => return err, }; return; } l.fresh = false; const start: Io.Timestamp = .now(l.io, .awake); if (try Segment.open(l.io, l.dir, size)) |seg| { l.index.attach(l.gpa, seg) catch |err| { var s = seg; s.close(); return err; }; l.walked = seg.walked; l.reached = seg.reached; } const from = if (l.index.segment) |seg| seg.corpus_end else magic.len; const added = try readRecords(l, from, size); if (l.index.segment == null or added > rebuild_after) try rebuild(l, size); l.index_ms = start.durationTo(.now(l.io, .awake)).toMilliseconds();}
/// Indexes the records from byte `from` of the corpus to its end, `size`,/// in memory, a buffer at a time; returns how many messages it added.fn readRecords(l: *Load, from: u64, size: u64) !u32 { var buf: [256 * 1024]u8 = undefined; var fr = l.file.?.reader(l.io, &buf); try fr.seekTo(from); const r = &fr.interface; var body: std.ArrayList(u8) = .empty; defer body.deinit(l.gpa); var record_arena: ArenaAllocator = .init(l.gpa); defer record_arena.deinit(); const a = l.arena.allocator(); var at = from; var added: u32 = 0; while (size - at >= 4) { const len = try r.takeInt(u32, .little); if (len == 0 or size - at - 4 < len) break; try body.resize(l.gpa, len); try r.readSliceAll(body.items); at += 4 + len; switch (@as(Kind, @fromBackingInt(@intCast(body.items[0])))) { .message => { _ = record_arena.reset(.retain_capacity); const m = try decode(record_arena.allocator(), body.items[1..]); if (l.ids.contains(m.id)) continue; if (l.index.segment) |seg| if (seg.findMessage(m.id) != null) continue; const doc = try l.index.add(l.gpa, m); try l.ids.put(l.gpa, try a.dupe(u8, m.id), doc); added += 1; }, .walked => l.walked = true, .reached => { var f: Fields = .{ .bytes = body.items[1..] }; l.reached = @min(l.reached, try f.int(i64)); }, .gone => { var f: Fields = .{ .bytes = body.items[1..] }; const id = try f.string(); if (l.ids.fetchSwapRemove(id)) |gone| { l.index.drop(gone.value); } else if (l.index.segment) |seg| if (seg.findMessage(id)) |doc| l.index.drop(doc); }, _ => {}, } } l.end = at; return added;}
/// Writes the segment again from the whole corpus, and maps it, leaving/// nothing indexed in memory.fn rebuild(l: *Load, size: u64) !void { if (l.index.segment != null) { l.index.deinit(l.gpa); l.index = .init(l.gpa); l.ids.clearRetainingCapacity(); _ = l.arena.reset(.retain_capacity); l.walked = false; l.reached = std.math.maxInt(i64); _ = try readRecords(l, magic.len, size); } try l.index.save(l.gpa, l.io, l.dir, &l.ids, l.end, l.walked, l.reached); const seg = try Segment.open(l.io, l.dir, size) orelse return error.SegmentUnreadable; l.index.deinit(l.gpa); l.index = .init(l.gpa); l.ids.clearRetainingCapacity(); _ = l.arena.reset(.retain_capacity); l.index.attach(l.gpa, seg) catch |err| { var s = seg; s.close(); return err; }; l.rebuilt = true;}
fn loadDone(user_data: ?*anyopaque) callconv(.c) void { const l: *Load = @ptrCast(@alignCast(user_data)); const self = l.corpus orelse return; self.load = null; l.corpus = null; self.ready = true; defer self.on_ready.call(self.on_ready.context); if (l.failed) |err| { log.warn("{s}: {t}; search keeps nothing this run", .{ file_name, err }); return; } self.takeFile(l) catch |err| { log.warn("{s}: {t}; search keeps nothing this run", .{ file_name, err }); return; }; mem.swap(std.array_hash_map.String(u32), &self.ids, &l.ids); mem.swap(ArenaAllocator, &self.ids_arena, &l.arena); mem.swap(Index, &self.index, &l.index); self.walked = l.walked; self.reached = l.reached; log.info("{d} messages kept for search, {d} MB; index {s} in {d} ms: {d} words, " ++ "{d} MB mapped, {d} messages in {d} MB of memory", .{ self.count(), l.bytes >> 20, if (l.rebuilt) "written" else "read", l.index_ms, self.index.wordCount(), self.index.mappedBytes() >> 20, self.index.countInMemory(), self.index.bytes() >> 20, }); for (self.doomed.items) |id| { self.remove(id) catch |err| log.err("dropping a message gone: {t}", .{err}); self.gpa.free(id); } self.doomed.clearRetainingCapacity();}
/// Takes the job's file, started over or with what was cut short cut off.fn takeFile(self: *Corpus, l: *Load) !void { const file = l.file.?; if (l.fresh) { try file.setLength(self.io, 0); try file.writePositionalAll(self.io, magic, 0); l.end = magic.len; } else if (l.end < l.bytes) { log.warn("{s}: cutting off {d} bytes cut short", .{ file_name, l.bytes - l.end }); try file.setLength(self.io, l.end); } self.file = file; self.end = l.end; l.file = null;}
fn loadDestroy(user_data: ?*anyopaque) callconv(.c) void { const l: *Load = @ptrCast(@alignCast(user_data)); if (l.file) |f| f.close(l.io); l.ids.deinit(l.gpa); l.index.deinit(l.gpa); l.arena.deinit(); l.gpa.destroy(l);}
fn testRecord(id: []const u8, text: []const u8) Record { return .{ .id = id, .thread = "T1", .message_id = "<a@b>", .subject = "Läuft's?", .list_id = "<dev.lists.example>", .delivered_to = "me@example.com", .date = -5, .size = 70_000, .from = .{ .name = "Ana", .address = "ana@example.com" }, .to = &.{ .{ .name = "", .address = "me@example.com" }, .{ .name = "Bo", .address = "bo@x.example" } }, .cc = &.{}, .bcc = &.{.{ .name = "", .address = "hidden@x.example" }}, .attachments = &.{.{ .name = "plan.pdf", .mime = "application/pdf", .size = 1234 }}, .text = text, .quoted = "", };}
/// Reads the corpus in `settings`, as a session starts.fn testOpen(app: siv.App, settings: Settings, ready: *bool) !Corpus { ready.* = false; var c: Corpus = undefined; c.open(testing.allocator, testing.io, app, settings, .{ .context = ready, .call = struct { fn call(context: *anyopaque) void { const r: *bool = @ptrCast(@alignCast(context)); r.* = true; } }.call }); var waited: u32 = 0; while (!ready.* and waited < 50) : (waited += 1) try siv.appDispatch(app, 100); try testing.expect(c.ready); return c;}
test "messages kept are read back the next run, for the user alone, and a record cut short is cut off" { var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); const settings: Settings = .{ .io = testing.io, .dir = tmp.dir }; const app = try siv.appCreate(.{ .backend = .headless, .flags = .{ .no_system_fonts = true, .no_system_icons = true, .reduced_motion = true }, }); defer siv.appDestroy(app); var ready = false;
var c = try testOpen(app, settings, &ready); try testing.expectEqual(0, c.count()); try c.add(testRecord("M1", "first")); try c.add(testRecord("M2", "second")); try c.add(testRecord("M1", "again")); try c.markWalked(); c.deinit(); const stat = try tmp.dir.statFile(testing.io, file_name, .{}); try testing.expectEqual(0o600, @backingInt(stat.permissions) & 0o777);
// A crash partway through a record. { const f = try tmp.dir.openFile(testing.io, file_name, .{ .mode = .read_write }); defer f.close(testing.io); try f.writePositionalAll(testing.io, &.{ 200, 0, 0, 0, 1, 2, 0 }, stat.size); } c = try testOpen(app, settings, &ready); try testing.expectEqual(2, c.count()); try testing.expect(c.has("M1") and c.has("M2") and !c.has("M3")); // Read back once, as the walk wrote them, into search's index too. try testing.expectEqual(2, c.index.count()); try testing.expect(c.walked); try testing.expectEqual(stat.size, c.end); try c.add(testRecord("M3", "third")); c.deinit();
var arena: ArenaAllocator = .init(testing.allocator); defer arena.deinit(); const bytes = try tmp.dir.readFileAlloc(testing.io, file_name, testing.allocator, .unlimited); defer testing.allocator.free(bytes); var it: Records = .init(bytes); const texts = [_][]const u8{ "first", "second", "third" }; for (texts) |text| { const r = (try it.message(arena.allocator())).?; try testing.expectEqualStrings(text, r.text); try testing.expectEqualStrings("Läuft's?", r.subject); try testing.expectEqual(-5, r.date); try testing.expectEqual(70_000, r.size); try testing.expectEqualStrings("bo@x.example", r.to[1].address); try testing.expectEqual(0, r.cc.len); try testing.expectEqualStrings("hidden@x.example", r.bcc[0].address); try testing.expectEqualStrings("plan.pdf", r.attachments[0].name); try testing.expectEqual(1234, r.attachments[0].size); } try testing.expectEqual(null, try it.message(arena.allocator()));}
test "a message gone is dropped for good, even one that goes before the file is read" { var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); const settings: Settings = .{ .io = testing.io, .dir = tmp.dir }; const app = try siv.appCreate(.{ .backend = .headless, .flags = .{ .no_system_fonts = true, .no_system_icons = true, .reduced_motion = true }, }); defer siv.appDestroy(app); var ready = false; var c = try testOpen(app, settings, &ready); for ([_][]const u8{ "M1", "M2", "M3" }) |id| try c.add(testRecord(id, id)); try c.remove("M1"); try c.remove("never kept"); try testing.expect(!c.has("M1") and c.has("M2")); c.deinit();
// The next run reads M1 as gone, and drops M2 as soon as it has read // the file, though told before. ready = false; c.open(testing.allocator, testing.io, app, settings, .{ .context = &ready, .call = struct { fn call(context: *anyopaque) void { const r: *bool = @ptrCast(@alignCast(context)); r.* = true; } }.call }); defer c.deinit(); try c.remove("M2"); var waited: u32 = 0; while (!ready and waited < 50) : (waited += 1) try siv.appDispatch(app, 100); try testing.expectEqual(1, c.count()); try testing.expect(c.has("M3") and !c.has("M1") and !c.has("M2")); var arena: ArenaAllocator = .init(testing.allocator); defer arena.deinit(); for ([_][]const u8{ "m1", "m2", "m3" }, [_]usize{ 0, 0, 1 }) |word, found| { var q: Query = .parse(word, 0); try testing.expectEqual(found, (try c.index.search(arena.allocator(), &q)).?.len); }}
test "the index is written once and mapped after, what the corpus gains since kept in memory" { var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); const settings: Settings = .{ .io = testing.io, .dir = tmp.dir }; const app = try siv.appCreate(.{ .backend = .headless, .flags = .{ .no_system_fonts = true, .no_system_icons = true, .reduced_motion = true }, }); defer siv.appDestroy(app); var ready = false; var c = try testOpen(app, settings, &ready); try c.add(testRecord("M1", "m1")); try c.add(testRecord("M2", "m2")); try c.markWalked(); c.deinit();
// The next run writes the index of all of it, for the user alone. c = try testOpen(app, settings, &ready); try testing.expectEqual(0, c.index.countInMemory()); try testing.expect(c.index.mappedBytes() > 0 and c.has("M1") and c.has("M2")); const stat = try tmp.dir.statFile(testing.io, Segment.file_name, .{}); try testing.expectEqual(0o600, @backingInt(stat.permissions) & 0o777); try c.add(testRecord("M3", "m3")); try c.remove("M2"); c.deinit();
// The one after maps it as it was, and indexes what came since: a // message, and one of the index's gone; the backfill's end, which // came before the index's, holds. c = try testOpen(app, settings, &ready); defer c.deinit(); const again = try tmp.dir.statFile(testing.io, Segment.file_name, .{}); try testing.expectEqual(stat.mtime, again.mtime); try testing.expectEqual(1, c.index.countInMemory()); try testing.expect(c.walked); try testing.expectEqual(2, c.count()); try testing.expect(c.has("M1") and !c.has("M2") and c.has("M3")); var arena: ArenaAllocator = .init(testing.allocator); defer arena.deinit(); for ([_][]const u8{ "m1", "m2", "m3" }, [_]usize{ 1, 0, 1 }) |word, found| { var q: Query = .parse(word, 0); try testing.expectEqual(found, (try c.index.search(arena.allocator(), &q)).?.len); }}
test "search may ask the index once the walk has kept a month of mail, which the next runs read back" { var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); const settings: Settings = .{ .io = testing.io, .dir = tmp.dir }; const app = try siv.appCreate(.{ .backend = .headless, .flags = .{ .no_system_fonts = true, .no_system_icons = true, .reduced_motion = true }, }); defer siv.appDestroy(app); const now = 2_000_000_000; const day = std.time.s_per_day; var ready = false; var c = try testOpen(app, settings, &ready); try testing.expectEqual(null, c.searchable(now)); for ([_][]const u8{ "M1", "M2", "M3" }, [_]i64{ 10, 40, 20 }) |id, days| { var r = testRecord(id, id); r.date = now - days * day; try c.add(r); } // Twenty days down; the oldest of a page counts, and a message not // kept, deleted since it was listed, does not. try c.reach(&.{ "M1", "M3", "M9" }); try testing.expectEqual(null, c.searchable(now)); // Forty days down, and never back up. try c.reach(&.{"M2"}); try c.reach(&.{"M1"}); try testing.expectEqual(Searchable{ .since = now - 40 * day }, c.searchable(now)); c.deinit();
// Read back from the records, then from the index they were // written into. for (0..2) |_| { c = try testOpen(app, settings, &ready); defer c.deinit(); try testing.expectEqual(now - 40 * day, c.reached); } c = try testOpen(app, settings, &ready); defer c.deinit(); try c.markWalked(); try testing.expectEqual(Searchable.all, c.searchable(now + 20 * day));}
test "a file of another layout is started over" { var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); const settings: Settings = .{ .io = testing.io, .dir = tmp.dir }; try tmp.dir.writeFile(testing.io, .{ .sub_path = file_name, .data = "parcel corpus\x00\x02\x00\x00\x00junk" }); try tmp.dir.writeFile(testing.io, .{ .sub_path = Segment.file_name, .data = "an index of the old corpus" }); const app = try siv.appCreate(.{ .backend = .headless, .flags = .{ .no_system_fonts = true, .no_system_icons = true, .reduced_motion = true }, }); defer siv.appDestroy(app); var ready = false; var c = try testOpen(app, settings, &ready); defer c.deinit(); try testing.expectEqual(0, c.count()); try testing.expect(!c.walked); try testing.expectEqual(magic.len, c.end); try testing.expectEqual(magic.len, (try tmp.dir.statFile(testing.io, file_name, .{})).size); try testing.expectError(error.FileNotFound, tmp.dir.statFile(testing.io, Segment.file_name, .{}));}
test "a message's own text is told from what it quotes" { var arena: ArenaAllocator = .init(testing.allocator); defer arena.deinit(); const s = try split(arena.allocator(), \\Sounds good, see you then. Bo wrote: \\ \\that he can't come. \\ \\On Tue, Nov 10, 2026 at 9:00 AM Ana <ana@example.com> wrote: \\ \\> Lunch on Friday? \\>> Or Thursday \\ \\-- \\Tim, who wrote: nothing quoted after this ); try testing.expectEqualStrings( "Sounds good, see you then. Bo wrote:\n\nthat he can't come.\n\n\n" ++ "-- \nTim, who wrote: nothing quoted after this", s.said, ); try testing.expectEqualStrings( "On Tue, Nov 10, 2026 at 9:00 AM Ana <ana@example.com> wrote:\nLunch on Friday?\nOr Thursday", s.quoted, );}