//! parcel's own search: the words of each message the corpus keeps,
//! and the messages each word is in. Most are in the segment, a file
//! mapped into memory (`Segment.zig`); the messages the corpus gained
//! since it was written are indexed here in memory, numbered after its
//! messages and threads, so a search goes through both as one. A
//! search finds the
//! messages every word of which some word of theirs starts with, so
//! "fede" finds Federico, and ranks their threads: a word in the
//! subject counts most, then the sender, those it went to, its files,
//! its text, and last what it quotes; a word typed whole counts twice
//! what a word's start does. Of threads ranked alike, the newer comes
//! first.
//!
//! Labels and flags are not kept, so a search that names them (label:,
//! is:, in:anywhere), or "me", is left to the server, as is one with no
//! words.

const std = @import("std");
const Corpus = @import("Corpus.zig");
const Query = @import("Query.zig");
const Segment = @import("Segment.zig");
const ascii = std.ascii;
const assert = std.debug.assert;
const mem = std.mem;
const Allocator = std.mem.Allocator;
const ArenaAllocator = std.heap.ArenaAllocator;
const ArrayList = std.ArrayList;
const Io = std.Io;
const testing = std.testing;

const Index = @This();

/// The file's index, if any, and those of its messages gone since it
/// was written.
segment: ?Segment,
segment_gone: std.bit_set.Dynamic,
/// How many messages are gone, in the file or here.
gone: u32,
/// The messages after the segment's, by the places postings name, past
/// its messages'.
docs: ArrayList(Doc),
/// The threads the segment does not have, by their ids, which docs name
/// by place, past its threads'.
threads: std.array_hash_map.String(void),
/// Each word, lowercased, and the messages it is in, in order.
words: std.array_hash_map.String(ArrayList(Posting)),
/// Holds the words and the threads' ids.
strings: ArenaAllocator,

/// A message: its thread, by place, when it came, and whether it has
/// since gone from the account, which search then leaves out.
const Doc = struct { thread: u32, date: i64, gone: bool };

/// Where a message has a word.
pub const Fields = packed struct(u8) {
    subject: bool = false,
    from: bool = false,
    to: bool = false,
    cc: bool = false,
    bcc: bool = false,
    attachment: bool = false,
    text: bool = false,
    quoted: bool = false,

    const all: Fields = @bitCast(@as(u8, 0xff));
};

/// A message a word is in, and where.
pub const Posting = packed struct(u32) { fields: Fields, doc: u24 };

/// The longest word kept: a longer run of letters is no word anyone
/// types, such as base64.
const max_word = 32;

/// The most words a search matches; past them, words are ignored.
const max_words = 32;

pub const empty: Index = .{
    .segment = null,
    .segment_gone = .{},
    .gone = 0,
    .docs = .empty,
    .threads = .empty,
    .words = .empty,
    .strings = undefined,
};

pub fn init(gpa: Allocator) Index {
    var index: Index = .empty;
    index.strings = .init(gpa);
    return index;
}

pub fn deinit(self: *Index, gpa: Allocator) void {
    for (self.words.values()) |*p| p.deinit(gpa);
    self.words.deinit(gpa);
    self.docs.deinit(gpa);
    self.threads.deinit(gpa);
    self.strings.deinit();
    if (self.segment) |*seg| seg.close();
    self.segment_gone.deinit(gpa);
    self.* = undefined;
}

/// Takes `seg` as what this index holds before what it is given; it
/// holds nothing yet.
pub fn attach(self: *Index, gpa: Allocator, seg: Segment) !void {
    assert(self.segment == null);
    assert(self.docs.items.len == 0);
    self.segment_gone = try .initEmpty(gpa, seg.docs.len);
    self.gone = 0;
    for (seg.docs) |d| self.gone += @intFromBool(d.gone != 0);
    self.segment = seg;
}

/// The messages and threads the segment holds, which come first.
fn base(self: *const Index) struct { docs: u32, threads: u32 } {
    const seg = self.segment orelse return .{ .docs = 0, .threads = 0 };
    return .{ .docs = @intCast(seg.docs.len), .threads = @intCast(seg.threads.len) };
}

/// The messages not gone.
pub fn count(self: *const Index) u32 {
    return self.base().docs + @as(u32, @intCast(self.docs.items.len)) - self.gone;
}

/// The messages indexed in memory rather than in the segment.
pub fn countInMemory(self: *const Index) u32 {
    return @intCast(self.docs.items.len);
}

pub fn wordCount(self: *const Index) u32 {
    const in_file: u32 = if (self.segment) |seg| @intCast(seg.words.len) else 0;
    return in_file + @as(u32, @intCast(self.words.count()));
}

/// How many bytes of the segment are mapped.
pub fn mappedBytes(self: *const Index) u64 {
    const seg = self.segment orelse return 0;
    return seg.map.memory.len;
}

/// Whether the message at place `doc` has gone from the account.
pub fn isGone(self: *const Index, doc: u32) bool {
    const b = self.base().docs;
    if (doc >= b) return self.docs.items[doc - b].gone;
    return self.segment.?.docs[doc].gone != 0 or self.segment_gone.isSet(doc);
}

/// When message `doc` came, in seconds since the epoch.
pub fn dateOf(self: *const Index, doc: u32) i64 {
    const b = self.base().docs;
    return if (doc < b) self.segment.?.docs[doc].date else self.docs.items[doc - b].date;
}

/// Writes what this index holds, which is in memory alone, as the
/// segment in `dir`: the corpus up to `corpus_end`, whose messages
/// `ids` names, by their places here.
pub fn save(
    self: *const Index,
    gpa: Allocator,
    io: Io,
    dir: Io.Dir,
    ids: *const std.array_hash_map.String(u32),
    corpus_end: u64,
    walked: bool,
    reached: i64,
) !void {
    assert(self.segment == null);
    var arena_state: ArenaAllocator = .init(gpa);
    defer arena_state.deinit();
    const a = arena_state.allocator();
    // The threads in order, and each one's new place by its old.
    const thread_ids = self.threads.keys();
    const order = try a.alloc(u32, thread_ids.len);
    for (order, 0..) |*o, i| o.* = @intCast(i);
    mem.sortUnstable(u32, order, thread_ids, byKey);
    const new_place = try a.alloc(u32, order.len);
    const sorted_threads = try a.alloc([]const u8, order.len);
    for (order, 0..) |old, i| {
        new_place[old] = @intCast(i);
        sorted_threads[i] = thread_ids[old];
    }
    const docs = try a.alloc(Segment.Doc, self.docs.items.len);
    for (self.docs.items, docs) |d, *out| {
        out.* = .{ .date = d.date, .thread = new_place[d.thread], .gone = @intFromBool(d.gone) };
    }
    const Message = @typeInfo(@FieldType(Segment.Parts, "messages")).pointer.child;
    const messages = try a.alloc(Message, ids.count());
    for (ids.keys(), ids.values(), messages) |id, doc, *m| m.* = .{ .id = id, .doc = doc };
    mem.sortUnstable(Message, messages, {}, struct {
        fn lessThan(_: void, x: Message, y: Message) bool {
            return mem.order(u8, x.id, y.id) == .lt;
        }
    }.lessThan);
    const word_order = try a.alloc(u32, self.words.count());
    for (word_order, 0..) |*o, i| o.* = @intCast(i);
    mem.sortUnstable(u32, word_order, self.words.keys(), byKey);
    const words = try a.alloc([]const u8, word_order.len);
    const postings = try a.alloc([]const Posting, word_order.len);
    for (word_order, words, postings) |i, *w, *p| {
        w.* = self.words.keys()[i];
        p.* = self.words.values()[i].items;
    }
    try Segment.write(io, dir, .{
        .corpus_end = corpus_end,
        .walked = walked,
        .reached = reached,
        .docs = docs,
        .threads = sorted_threads,
        .messages = messages,
        .words = words,
        .postings = postings,
    });
}

fn byKey(keys: []const []const u8, x: u32, y: u32) bool {
    return mem.order(u8, keys[x], keys[y]) == .lt;
}

/// About how many bytes the index holds in memory, for the log.
pub fn bytes(self: *const Index) u64 {
    var total: u64 = self.strings.queryCapacity() + self.docs.capacity * @sizeOf(Doc);
    total += self.segment_gone.capacity() / 8;
    total += self.words.capacity() * (@sizeOf([]const u8) + @sizeOf(ArrayList(Posting)) + 8);
    for (self.words.values()) |p| total += p.capacity * @sizeOf(Posting);
    return total;
}

/// Adds the words of message `r`, and returns its place, which `drop`
/// takes.
pub fn add(self: *Index, gpa: Allocator, r: Corpus.Record) !u32 {
    const b = self.base();
    if (b.docs + self.docs.items.len > std.math.maxInt(u24)) return error.TooManyMessages;
    const doc: u24 = @intCast(b.docs + self.docs.items.len);
    const in_segment: ?u32 = if (self.segment) |*seg| seg.findThread(r.thread) else null;
    const place = in_segment orelse blk: {
        const entry = try self.threads.getOrPut(gpa, r.thread);
        if (!entry.found_existing) entry.key_ptr.* = try self.strings.allocator().dupe(u8, r.thread);
        break :blk b.threads + @as(u32, @intCast(entry.index));
    };
    try self.docs.append(gpa, .{ .thread = place, .date = r.date, .gone = false });
    try self.addText(gpa, doc, .{ .subject = true }, r.subject);
    try self.addPerson(gpa, doc, .{ .from = true }, r.from);
    for (r.to) |p| try self.addPerson(gpa, doc, .{ .to = true }, p);
    for (r.cc) |p| try self.addPerson(gpa, doc, .{ .cc = true }, p);
    for (r.bcc) |p| try self.addPerson(gpa, doc, .{ .bcc = true }, p);
    for (r.attachments) |a| try self.addText(gpa, doc, .{ .attachment = true }, a.name);
    try self.addText(gpa, doc, .{ .text = true }, r.text);
    try self.addText(gpa, doc, .{ .quoted = true }, r.quoted);
    return doc;
}

/// Leaves the message at place `doc` out of searches from now on.
pub fn drop(self: *Index, doc: u32) void {
    if (self.isGone(doc)) return;
    self.gone += 1;
    const b = self.base().docs;
    if (doc >= b) self.docs.items[doc - b].gone = true else self.segment_gone.set(doc);
}

fn addPerson(self: *Index, gpa: Allocator, doc: u24, field: Fields, p: Corpus.Person) !void {
    try self.addText(gpa, doc, field, p.name);
    try self.addText(gpa, doc, field, p.address);
}

fn addText(self: *Index, gpa: Allocator, doc: u24, field: Fields, text: []const u8) !void {
    var rest = text;
    var buf: [max_word]u8 = undefined;
    while (nextWord(&rest, &buf)) |word| {
        const entry = try self.words.getOrPut(gpa, word);
        if (!entry.found_existing) {
            entry.key_ptr.* = try self.strings.allocator().dupe(u8, word);
            entry.value_ptr.* = .empty;
        }
        const list = entry.value_ptr;
        // A message's words come together, so it is last if it has this one.
        if (list.items.len > 0 and list.items[list.items.len - 1].doc == doc) {
            const last = &list.items[list.items.len - 1];
            last.fields = @bitCast(@as(u8, @bitCast(last.fields)) | @as(u8, @bitCast(field)));
        } else try list.append(gpa, .{ .fields = field, .doc = doc });
    }
}

/// The next word of `rest`, lowercased into `buf`, and `rest` past it: a
/// run of letters and digits, every byte of a UTF-8 sequence a letter.
/// Runs longer than `max_word` are skipped.
fn nextWord(rest: *[]const u8, buf: *[max_word]u8) ?[]const u8 {
    const s = rest.*;
    var i: usize = 0;
    while (i < s.len) {
        while (i < s.len and !isLetter(s[i])) i += 1;
        const start = i;
        while (i < s.len and isLetter(s[i])) i += 1;
        if (i == start or i - start > max_word) continue;
        rest.* = s[i..];
        return ascii.lowerString(buf, s[start..i]);
    }
    rest.* = s[s.len..];
    return null;
}

fn isLetter(c: u8) bool {
    return ascii.isAlphanumeric(c) or c >= 0x80;
}

/// A word a search matches, and where.
const Want = struct { word: []const u8, fields: Fields };

/// The threads, by their ids, best first, of the messages that match
/// `q`; null if `q` asks what the index cannot tell. They point into
/// the index, and `arena` holds the slice.
pub fn search(self: *const Index, arena: Allocator, q: *const Query) !?[]const []const u8 {
    var wants: [max_words]Want = undefined;
    var n: u8 = 0;
    var after: i64 = std.math.minInt(i64);
    var before: i64 = std.math.maxInt(i64);
    const excluded = try arena.alloc(bool, self.allDocs());
    @memset(excluded, false);
    for (q.terms[0..q.len]) |*t| {
        const value, const fields: Fields = switch (t.what) {
            .text => |s| .{ s, .all },
            .from => |s| .{ s, .{ .from = true } },
            .to => |s| .{ s, .{ .to = true } },
            .cc => |s| .{ s, .{ .cc = true } },
            .subject => |s| .{ s, .{ .subject = true } },
            .before => |time| {
                if (t.negated) after = @max(after, time) else before = @min(before, time);
                continue;
            },
            .after => |time| {
                if (t.negated) before = @min(before, time) else after = @max(after, time);
                continue;
            },
            .label, .unread, .starred, .anywhere => return null,
        };
        if (fields != Fields.all and ascii.eqlIgnoreCase(value, "me")) return null;
        var term_wants: [max_words]Want = undefined;
        const got = try wordsOf(arena, value, fields, &term_wants);
        if (t.negated) {
            try self.exclude(arena, got, excluded);
            continue;
        }
        for (got) |w| {
            if (n == max_words) break;
            wants[n] = w;
            n += 1;
        }
    }
    if (n == 0) return null;
    const matched = try arena.alloc(u8, self.allDocs());
    const scores = try arena.alloc(u32, self.allDocs());
    @memset(matched, 0);
    @memset(scores, 0);
    for (wants[0..n], 0..) |w, i| self.match(w, @intCast(i), matched, scores);
    const threads = try self.rank(arena, n, matched, scores, excluded, after, before);
    return threads;
}

/// The words of a term's `value`, each to be found in `fields`, copied
/// into `arena`.
fn wordsOf(arena: Allocator, value: []const u8, fields: Fields, out: *[max_words]Want) ![]const Want {
    var rest = value;
    var buf: [max_word]u8 = undefined;
    var n: usize = 0;
    while (nextWord(&rest, &buf)) |word| {
        if (n == out.len) break;
        out[n] = .{ .word = try arena.dupe(u8, word), .fields = fields };
        n += 1;
    }
    return out[0..n];
}

/// Marks the messages that have every one of `wants`.
fn exclude(self: *const Index, arena: Allocator, wants: []const Want, excluded: []bool) !void {
    if (wants.len == 0) return;
    const matched = try arena.alloc(u8, self.allDocs());
    const scores = try arena.alloc(u32, self.allDocs());
    @memset(matched, 0);
    for (wants, 0..) |w, i| self.match(w, @intCast(i), matched, scores);
    for (matched, excluded) |m, *e| {
        if (m == wants.len) e.* = true;
    }
}

/// Counts `w`, the `i`th word, for the messages that had every word
/// before it: the word itself first, then the words it starts.
fn match(self: *const Index, w: Want, i: u8, matched: []u8, scores: []u32) void {
    if (self.segment) |*seg| {
        const range = seg.wordsStarting(w.word);
        for (range.start..range.end) |at| {
            const whole = seg.words[at].word.len == w.word.len;
            matchPostings(seg.postingsOf(@intCast(at)), w.fields, i, if (whole) 2 else 1, matched, scores);
        }
    }
    if (self.words.get(w.word)) |postings| matchPostings(postings.items, w.fields, i, 2, matched, scores);
    for (self.words.keys(), self.words.values()) |word, postings| {
        if (word.len <= w.word.len or !mem.startsWith(u8, word, w.word)) continue;
        matchPostings(postings.items, w.fields, i, 1, matched, scores);
    }
}

fn matchPostings(postings: []const Posting, fields: Fields, i: u8, times: u32, matched: []u8, scores: []u32) void {
    const want: u8 = @bitCast(fields);
    for (postings) |p| {
        const where: u8 = @as(u8, @bitCast(p.fields)) & want;
        if (where == 0 or matched[p.doc] != i) continue;
        matched[p.doc] = i + 1;
        scores[p.doc] += weight(@bitCast(where)) * times;
    }
}

/// What the best of the fields a word is in counts for.
fn weight(f: Fields) u32 {
    if (f.subject) return 8;
    if (f.from) return 6;
    if (f.to or f.cc or f.attachment) return 3;
    if (f.bcc or f.text) return 2;
    return 1;
}

const Hit = struct { thread: u32, score: u32, date: i64 };

/// The threads of the messages that matched all `n` words, were not
/// left out, and came after `after` and before `before`, best first.
fn rank(
    self: *const Index,
    arena: Allocator,
    n: u8,
    matched: []const u8,
    scores: []const u32,
    excluded: []const bool,
    after: i64,
    before: i64,
) ![]const []const u8 {
    // Each thread's best, by its place; a thread not found has no date.
    const b = self.base();
    const best = try arena.alloc(Hit, b.threads + self.threads.count());
    for (best, 0..) |*h, i| h.* = .{ .thread = @intCast(i), .score = 0, .date = std.math.minInt(i64) };
    var found: u32 = 0;
    for (0..self.allDocs()) |at| {
        const doc: u32 = @intCast(at);
        if (matched[doc] != n or excluded[doc] or self.isGone(doc)) continue;
        const thread, const date = if (doc < b.docs) d: {
            const d = self.segment.?.docs[doc];
            break :d .{ d.thread, d.date };
        } else d: {
            const d = self.docs.items[doc - b.docs];
            break :d .{ d.thread, d.date };
        };
        if (date < after or date >= before) continue;
        const h = &best[thread];
        if (h.date == std.math.minInt(i64)) found += 1;
        h.score = @max(h.score, scores[doc]);
        h.date = @max(h.date, date);
    }
    const hits = try arena.alloc(Hit, found);
    var i: u32 = 0;
    for (best) |h| if (h.date != std.math.minInt(i64)) {
        hits[i] = h;
        i += 1;
    };
    mem.sortUnstable(Hit, hits, {}, better);
    const out = try arena.alloc([]const u8, hits.len);
    for (hits, out) |h, *id| id.* = self.threadId(h.thread);
    return out;
}

fn threadId(self: *const Index, thread: u32) []const u8 {
    const b = self.base().threads;
    return if (thread < b) self.segment.?.str(self.segment.?.threads[thread]) else self.threads.keys()[thread - b];
}

fn allDocs(self: *const Index) u32 {
    return self.base().docs + @as(u32, @intCast(self.docs.items.len));
}

fn better(_: void, a: Hit, b: Hit) bool {
    if (a.score != b.score) return a.score > b.score;
    return if (a.date != b.date) a.date > b.date else a.thread < b.thread;
}

fn testRecord(thread: []const u8, date: i64, subject: []const u8, from: []const u8, text: []const u8) Corpus.Record {
    return .{
        .id = thread,
        .thread = thread,
        .message_id = "",
        .subject = subject,
        .list_id = "",
        .delivered_to = "",
        .date = date,
        .size = 0,
        .from = .{ .name = from, .address = "" },
        .to = &.{.{ .name = "", .address = "tim@example.com" }},
        .cc = &.{},
        .bcc = &.{},
        .attachments = &.{},
        .text = text,
        .quoted = "",
    };
}

fn testSearch(arena: Allocator, index: *const Index, text: []const u8) !?[]const []const u8 {
    var q: Query = .parse(text, 100 * std.time.s_per_day);
    return index.search(arena, &q);
}

test "a word's start finds it; a whole word, the subject, and the sender rank first, then the newer" {
    var index: Index = .init(testing.allocator);
    defer index.deinit(testing.allocator);
    _ = try index.add(testing.allocator, testRecord("body-old", 10, "Lunch", "Ana", "Ask Federico about it"));
    _ = try index.add(testing.allocator, testRecord("body-new", 20, "Dinner", "Bo", "federal holiday, then"));
    _ = try index.add(testing.allocator, testRecord("sender", 5, "Hello", "Federico Ruiz", "nothing here"));
    _ = try index.add(testing.allocator, testRecord("subject", 1, "Re: Fede's plan", "Cy", "ok"));
    _ = try index.add(testing.allocator, testRecord("none", 30, "Fed up", "Di", "deferred"));
    const gone = try index.add(testing.allocator, testRecord("gone", 40, "Fede", "Fede", "fede"));
    index.drop(gone);
    var arena: ArenaAllocator = .init(testing.allocator);
    defer arena.deinit();
    const a = arena.allocator();
    // "fede" typed whole in the subject outranks the sender's name it
    // starts, which outranks the text, newer first; a message gone from
    // the account is no match, however well it would rank.
    const got = (try testSearch(a, &index, "FEDE")).?;
    try testing.expectEqual(4, got.len);
    for ([_][]const u8{ "subject", "sender", "body-new", "body-old" }, got) |want, thread| {
        try testing.expectEqualStrings(want, thread);
    }
    // Every word must match, a phrase's too; from: looks at the sender
    // alone; a - leaves out what it matches.
    try testing.expectEqual(1, (try testSearch(a, &index, "fede ask")).?.len);
    try testing.expectEqual(1, (try testSearch(a, &index, "\"ask fede\"")).?.len);
    try testing.expectEqualStrings("sender", (try testSearch(a, &index, "from:fed")).?[0]);
    try testing.expectEqual(1, (try testSearch(a, &index, "from:fed")).?.len);
    const left = (try testSearch(a, &index, "fede -federico")).?;
    try testing.expectEqual(2, left.len);
    try testing.expectEqualStrings("subject", left[0]);
    // A date bounds it; a word inside another is no match.
    try testing.expectEqualStrings("body-new", (try testSearch(a, &index, "fede after:15")).?[0]);
    try testing.expectEqual(0, (try testSearch(a, &index, "erico")).?.len);
    // The server answers what the index cannot.
    try testing.expectEqual(null, try testSearch(a, &index, "fede is:unread"));
    try testing.expectEqual(null, try testSearch(a, &index, "from:me"));
    try testing.expectEqual(null, try testSearch(a, &index, "before:2026/11/01"));
}

test "a thread ranks by its best message, and a run too long for a word is none" {
    var index: Index = .init(testing.allocator);
    defer index.deinit(testing.allocator);
    _ = try index.add(testing.allocator, testRecord("t1", 1, "x", "a", "plan"));
    _ = try index.add(testing.allocator, testRecord("t2", 2, "x", "b", "planning"));
    _ = try index.add(testing.allocator, testRecord("t1", 3, "the plan", "c", ""));
    _ = try index.add(testing.allocator, testRecord("t3", 4, "", "d", "Zm9vYmFyYmF6cXV4cXV1eHF1dXhxdXV4cXV1eA rest"));
    var arena: ArenaAllocator = .init(testing.allocator);
    defer arena.deinit();
    const got = (try testSearch(arena.allocator(), &index, "plan")).?;
    try testing.expectEqual(2, got.len);
    try testing.expectEqualStrings("t1", got[0]);
    try testing.expectEqual(0, (try testSearch(arena.allocator(), &index, "zm9v")).?.len);
    try testing.expectEqual(1, (try testSearch(arena.allocator(), &index, "rest")).?.len);
    // x, a, tim, example, com, plan, b, planning, the, c, d, rest: not the run.
    try testing.expectEqual(12, index.wordCount());
}

test "a search goes through the segment and what came since as one" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var ids: std.array_hash_map.String(u32) = .empty;
    defer ids.deinit(testing.allocator);
    {
        var before: Index = .init(testing.allocator);
        defer before.deinit(testing.allocator);
        const records = [_]Corpus.Record{
            testRecord("t1", 1, "Plan", "a", ""),
            testRecord("t2", 2, "x", "b", "planning"),
            testRecord("t0", 3, "x", "c", "plan"),
        };
        for ([_][]const u8{ "m1", "m2", "m3" }, records) |id, r| {
            try ids.put(testing.allocator, id, try before.add(testing.allocator, r));
        }
        try before.save(testing.allocator, testing.io, tmp.dir, &ids, 100, true, 1_700_000_000);
    }
    var index: Index = .init(testing.allocator);
    defer index.deinit(testing.allocator);
    const seg = (try Segment.open(testing.io, tmp.dir, 100)).?;
    try testing.expect(seg.walked);
    try testing.expectEqual(1_700_000_000, seg.reached);
    try testing.expectEqual(1, seg.findMessage("m2").?);
    try testing.expectEqual(null, seg.findMessage("m4"));
    // Older than the corpus it says it indexes is no index of it.
    try testing.expectEqual(null, try Segment.open(testing.io, tmp.dir, 99));
    try index.attach(testing.allocator, seg);
    // A message in a thread the segment holds ranks that thread; one in
    // a new thread is numbered after the segment's; one gone is no match.
    _ = try index.add(testing.allocator, testRecord("t2", 9, "the plan", "d", ""));
    _ = try index.add(testing.allocator, testRecord("t3", 4, "x", "e", "planet"));
    index.drop(0);
    try testing.expectEqual(4, index.count());
    var arena: ArenaAllocator = .init(testing.allocator);
    defer arena.deinit();
    const got = (try testSearch(arena.allocator(), &index, "plan")).?;
    try testing.expectEqual(3, got.len);
    for ([_][]const u8{ "t2", "t0", "t3" }, got) |want, thread| try testing.expectEqualStrings(want, thread);
    try testing.expectEqual(1, (try testSearch(arena.allocator(), &index, "planning")).?.len);
}