Something went wrong. Try again.
GET /xrpc/tech.waow.typeahead.searchActors typeahead.waow.tech
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481//! Overlay projection (Milestone C): turn a synced actor delta into//! actor_overlay + prefix_overlay rows so index serving reflects the change//! before the next build. Called from sync.zig INSIDE the write transaction//! that upserts the actors row — so the actor write and its overlay are atomic.//!//! Only deltas newer than the snapshot watermark reach here (sync gates on it);//! older state is already baked into the base. Scores match the builder//! (scoring.scoreOf) so overlay and base entries rank on the same scale.//!//! Invariants (Codex review)://! - actor_overlay.blob = NULL is a GLOBAL DID tombstone (suppresses at//! hydrate regardless of prefix rows).//! - sync rows are source=0; operator overrides (future) are source=1 and//! survive compaction.//! - prefix diff writes added (op=1) / removed (op=0) keys, AND upserts the//! latest score for unchanged-set keys too. The original design deferred//! "score drift" to the next build, but enrichment (`quality_score`://! 0 → 850) is exactly the signal ranking needs to reflect immediately —//! waiting hours for a snapshot rebuild defeated the whole prioritized//! enrichment pipeline. compaction prunes these on the next rebuild.const std = @import("std");const zqlite = @import("zqlite");const Allocator = std.mem.Allocator;
const normalize = @import("normalize.zig");const prefixes = @import("prefixes.zig");const scoring = @import("scoring.zig");const pack = @import("pack.zig");
const log = std.log.scoped(.overlay);
/// Raw (un-normalized) actor fields from a sync delta. We normalize the/// handle/display here so prefix derivation matches the builder exactly.pub const Actor = struct { did: []const u8, handle: []const u8, display_name: []const u8, avatar_url: []const u8, labels: []const u8, associated: []const u8, pds: []const u8, created_at: []const u8, hidden: bool, quality_score: f32,};
/// Project an updated/created actor. `old_handle`/`old_display` are the/// pre-update values (the local row being replaced); pass "" for a brand-new/// actor. `updated_at` is the SOURCE timestamp (Turso updated_at), used for/// watermark-based compaction later.pub fn projectUpdate( arena: Allocator, conn: zqlite.Conn, a: Actor, old_handle: []const u8, old_display: []const u8, updated_at: i64,) !void { const nh = try normalize.handle(arena, a.handle); const nd = try normalize.display(arena, a.display_name); const new_keys = try prefixes.deriveAllUnique(arena, nh, nd);
const oh = try normalize.handle(arena, old_handle); const od = try normalize.display(arena, old_display); const old_keys = try prefixes.deriveAllUnique(arena, oh, od);
const score = scoring.scoreOf(.{ .quality_score = a.quality_score });
// Pack the RAW handle/display — `nh`/`nd` are normalized for key derivation // only. The blob is what hydrate returns to clients, so we'd otherwise // serve every enriched actor's displayName lowercased. The builder packs // raw (builder.zig); the overlay must match. const blob = try pack.packActor(arena, .{ .hidden = a.hidden, .quality_score = a.quality_score, .handle = a.handle, .display = a.display_name, .avatar = a.avatar_url, .labels = a.labels, .pds = a.pds, .created_at = a.created_at, .associated = a.associated, }); try conn.exec( "INSERT OR REPLACE INTO actor_overlay (did, blob, hidden, updated_at, source) VALUES (?, ?, ?, ?, 0)", .{ a.did, zqlite.blob(blob), @as(i64, @intFromBool(a.hidden)), updated_at }, );
try diffPrefixes(conn, a.did, old_keys, new_keys, score, updated_at);}
/// Project a deletion: write a global tombstone (NULL blob) and suppress the/// actor's known prefixes so it stops appearing as a candidate. The tombstone/// alone is sufficient for correctness (hydrate skips it); the prefix/// suppressions just keep candidate sets tidy.pub fn projectDelete( arena: Allocator, conn: zqlite.Conn, did: []const u8, old_handle: []const u8, old_display: []const u8, updated_at: i64,) !void { try conn.exec( "INSERT OR REPLACE INTO actor_overlay (did, blob, hidden, updated_at, source) VALUES (?, NULL, 0, ?, 0)", .{ did, updated_at }, );
const oh = try normalize.handle(arena, old_handle); const od = try normalize.display(arena, old_display); for (try prefixes.deriveAllUnique(arena, oh, od)) |k| { try upsertPrefix(conn, k, did, 0, 0, updated_at); }}
/// Merge-walk two sorted+deduped key lists. Writes per branch:/// - key only in `old` (rename / delete) → suppress (op=0)./// - key only in `new` (rename / new actor) → add (op=1, new score)./// - key in both (unchanged set) → upsert (op=1, new score). Carries live/// score updates (enrichment, refresh) into ranking before the next/// snapshot rebuild. INSERT OR REPLACE handles the no-change case as a/// cheap rewrite; compaction prunes on rebuild.fn diffPrefixes( conn: zqlite.Conn, did: []const u8, old_keys: []const []const u8, new_keys: []const []const u8, score: f32, updated_at: i64,) !void { var oi: usize = 0; var ni: usize = 0; while (oi < old_keys.len or ni < new_keys.len) { if (ni == new_keys.len) { try upsertPrefix(conn, old_keys[oi], did, 0, 0, updated_at); oi += 1; } else if (oi == old_keys.len) { try upsertPrefix(conn, new_keys[ni], did, score, 1, updated_at); ni += 1; } else switch (std.mem.order(u8, old_keys[oi], new_keys[ni])) { .lt => { try upsertPrefix(conn, old_keys[oi], did, 0, 0, updated_at); oi += 1; }, .gt => { try upsertPrefix(conn, new_keys[ni], did, score, 1, updated_at); ni += 1; }, .eq => { try upsertPrefix(conn, new_keys[ni], did, score, 1, updated_at); oi += 1; ni += 1; }, } }}
fn upsertPrefix(conn: zqlite.Conn, key: []const u8, did: []const u8, score: f32, op: i64, updated_at: i64) !void { try conn.exec( "INSERT OR REPLACE INTO prefix_overlay (key, did, score, op, updated_at, source) VALUES (?, ?, ?, ?, ?, 0)", .{ key, did, @as(f64, score), op, updated_at }, );}
/// Rows deleted per compactChunk call. Bounds how long the caller holds the/// write mutex: a full-overlay compaction after a long sync outage is tens of/// millions of prefix rows, and holding the mutex across all of them blocked/// the resync's next batch until the sync watchdog killed the process —/// which restarted the compaction from scratch, forever (2026-08-28).pub const COMPACT_CHUNK: u32 = 20_000;
/// One bounded step of compaction: drop up to COMPACT_CHUNK sync-derived/// overlay rows at or below the new snapshot watermark (the base now/// reflects them). Operator rows (source=1) survive — a manual hide/show/// must not vanish on promote. Returns rows deleted; 0 means done. The/// caller loops, releasing the write mutex between calls.pub fn compactChunk(conn: zqlite.Conn, new_watermark: i64) !u32 { try conn.exec( \\DELETE FROM prefix_overlay WHERE (key, did) IN ( \\ SELECT key, did FROM prefix_overlay WHERE source = 0 AND updated_at <= ? LIMIT ?) , .{ new_watermark, COMPACT_CHUNK }); const prefix_deleted: u32 = @intCast(conn.changes()); try conn.exec( \\DELETE FROM actor_overlay WHERE did IN ( \\ SELECT did FROM actor_overlay WHERE source = 0 AND updated_at <= ? LIMIT ?) , .{ new_watermark, COMPACT_CHUNK }); return prefix_deleted + @as(u32, @intCast(conn.changes()));}
/// Full compaction in one call — for callers that own the connection outright/// (tests, tooling). Production goes through LocalDb.compactOverlay so the/// write mutex is released between chunks.pub fn compact(conn: zqlite.Conn, new_watermark: i64) !void { while (try compactChunk(conn, new_watermark) > 0) {}}
// ─── tests ───────────────────────────────────────────────────────────────
const testing = std.testing;const schema = @import("schema.zig");
fn memConn() !zqlite.Conn { const conn = try zqlite.open(":memory:", zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite); try schema.initOverlay(conn); return conn;}
fn prefixOp(conn: zqlite.Conn, key: []const u8, did: []const u8) !?i64 { const row = try conn.row("SELECT op FROM prefix_overlay WHERE key = ? AND did = ?", .{ key, did }); if (row) |r| { defer r.deinit(); return r.int(0); } return null;}
test "projectUpdate writes actor blob + adds new prefixes" { var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const conn = try memConn(); defer conn.close();
try projectUpdate(arena.allocator(), conn, .{ .did = "did:plc:x", .handle = "nate.fm", .display_name = "Nate", .avatar_url = "", .labels = "[]", .associated = "{}", .pds = "", .created_at = "", .hidden = false, .quality_score = 1.0, }, "", "", 100);
// actor_overlay row exists, non-null blob, not hidden const ar = (try conn.row("SELECT hidden FROM actor_overlay WHERE did = ?", .{"did:plc:x"})).?; defer ar.deinit(); try testing.expectEqual(@as(i64, 0), ar.int(0));
// handle + exact + display prefixes added (op=1) try testing.expectEqual(@as(?i64, 1), try prefixOp(conn, "h:nate.fm", "did:plc:x")); try testing.expectEqual(@as(?i64, 1), try prefixOp(conn, "e:nate.fm", "did:plc:x")); try testing.expectEqual(@as(?i64, 1), try prefixOp(conn, "n:nate", "did:plc:x"));}
test "projectUpdate rename suppresses old keys, adds new, leaves common" { var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const conn = try memConn(); defer conn.close();
// alice.fm → alicia.fm (common prefix "h:a"/"h:al"/"h:ali" unchanged). try projectUpdate(arena.allocator(), conn, .{ .did = "did:plc:y", .handle = "alicia.fm", .display_name = "", .avatar_url = "", .labels = "[]", .associated = "{}", .pds = "", .created_at = "", .hidden = false, .quality_score = 1.0, }, "alice.fm", "", 200);
try testing.expectEqual(@as(?i64, 0), try prefixOp(conn, "e:alice.fm", "did:plc:y")); // old exact suppressed try testing.expectEqual(@as(?i64, 0), try prefixOp(conn, "h:alice", "did:plc:y")); // old handle prefix suppressed try testing.expectEqual(@as(?i64, 1), try prefixOp(conn, "e:alicia.fm", "did:plc:y")); // new exact added try testing.expectEqual(@as(?i64, 1), try prefixOp(conn, "h:alici", "did:plc:y")); // new handle prefix added try testing.expectEqual(@as(?i64, 1), try prefixOp(conn, "h:al", "did:plc:y")); // common key → upserted w/ new score (live score carry)}
test "projectUpdate packs RAW handle/display (not normalized) into the blob" { var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const conn = try memConn(); defer conn.close();
// mixed-case display name with kanji is what hydrate returns to clients; // it MUST round-trip exactly. The bug was packing normalize.display(…) // (lowercased + katakana→hiragana folded) into the blob, so every overlay // hit showed up lowercased + script-folded. try projectUpdate(arena.allocator(), conn, .{ .did = "did:plc:q", .handle = "Pfrazee.com", // raw casing preserved .display_name = "Paul Frazee 🦀", .avatar_url = "", .labels = "[]", .associated = "{}", .pds = "", .created_at = "", .hidden = false, .quality_score = 850.0, }, "", "", 100);
const row = (try conn.row( "SELECT blob FROM actor_overlay WHERE did = ?", .{"did:plc:q"}, )).?; defer row.deinit(); const blob = row.blob(0); const a = try pack.unpackActor(blob); try testing.expectEqualStrings("Pfrazee.com", a.handle); try testing.expectEqualStrings("Paul Frazee 🦀", a.display);}
test "projectUpdate score-only change carries score into overlay (enrichment bug)" { var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const conn = try memConn(); defer conn.close();
// initial project: zero-score actor (unenriched, ~99% of corpus pre-enrichment). try projectUpdate(arena.allocator(), conn, .{ .did = "did:plc:p", .handle = "pfrazee.com", .display_name = "Paul Frazee", .avatar_url = "", .labels = "[]", .associated = "{}", .pds = "", .created_at = "", .hidden = false, .quality_score = 0.0, }, "", "", 100); try testing.expectEqual(@as(?i64, 1), try prefixOp(conn, "h:pfrazee.com", "did:plc:p"));
// same handle/display, score now 850 (enrichment). ALL keys hit `.eq` — // the pre-fix updater skipped these, leaving overlay scores at 0 and // ranking unchanged. fixed updater upserts every common key with the // new score so ranking reflects the enrichment within sync's next tick. try projectUpdate(arena.allocator(), conn, .{ .did = "did:plc:p", .handle = "pfrazee.com", .display_name = "Paul Frazee", .avatar_url = "", .labels = "[]", .associated = "{}", .pds = "", .created_at = "", .hidden = false, .quality_score = 850.0, }, "pfrazee.com", "Paul Frazee", 200);
const row = (try conn.row( "SELECT op, score FROM prefix_overlay WHERE key = ? AND did = ?", .{ "h:pfrazee.com", "did:plc:p" }, )).?; defer row.deinit(); try testing.expectEqual(@as(i64, 1), row.int(0)); try testing.expect(row.float(1) > 100); // scoring.scoreOf(850) ≫ 0}
test "projectDelete writes tombstone (NULL blob) + suppresses old keys" { var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const conn = try memConn(); defer conn.close();
try projectDelete(arena.allocator(), conn, "did:plc:z", "bob.fm", "", 300);
const ar = (try conn.row("SELECT blob IS NULL FROM actor_overlay WHERE did = ?", .{"did:plc:z"})).?; defer ar.deinit(); try testing.expectEqual(@as(i64, 1), ar.int(0)); // blob IS NULL → tombstone try testing.expectEqual(@as(?i64, 0), try prefixOp(conn, "e:bob.fm", "did:plc:z")); // suppressed}
test "compact prunes sync rows ≤ watermark but keeps operator + newer rows" { var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const conn = try memConn(); defer conn.close();
// sync row @100, sync row @300, operator row @50. try conn.exec("INSERT INTO prefix_overlay (key,did,score,op,updated_at,source) VALUES ('h:a','d1',1,1,100,0)", .{}); try conn.exec("INSERT INTO prefix_overlay (key,did,score,op,updated_at,source) VALUES ('h:b','d2',1,1,300,0)", .{}); try conn.exec("INSERT INTO prefix_overlay (key,did,score,op,updated_at,source) VALUES ('h:c','d3',1,0,50,1)", .{});
try compact(conn, 200);
try testing.expectEqual(@as(?i64, null), try prefixOp(conn, "h:a", "d1")); // sync ≤200 pruned try testing.expectEqual(@as(?i64, 1), try prefixOp(conn, "h:b", "d2")); // sync >200 kept try testing.expectEqual(@as(?i64, 0), try prefixOp(conn, "h:c", "d3")); // operator kept despite ≤200}
test "compactChunk is bounded and converges across an overlay larger than one chunk" { const conn = try memConn(); defer conn.close();
const n: u32 = COMPACT_CHUNK * 2 + 7; try conn.exec("BEGIN", .{}); var i: u32 = 0; while (i < n) : (i += 1) { var key_buf: [32]u8 = undefined; const key = try std.fmt.bufPrint(&key_buf, "h:{d}", .{i}); try conn.exec("INSERT INTO prefix_overlay (key,did,score,op,updated_at,source) VALUES (?, 'd', 1, 1, 100, 0)", .{key}); } try conn.exec("INSERT INTO prefix_overlay (key,did,score,op,updated_at,source) VALUES ('keep','d', 1, 1, 100, 1)", .{}); try conn.exec("INSERT INTO actor_overlay (did,blob,hidden,updated_at,source) VALUES ('a1', x'00', 0, 100, 0)", .{}); try conn.exec("COMMIT", .{});
const first = try compactChunk(conn, 200); try testing.expectEqual(COMPACT_CHUNK + 1, first); // one chunk of prefixes + the lone actor row
var total: u32 = first; var rounds: u32 = 1; while (true) { const d = try compactChunk(conn, 200); if (d == 0) break; total += d; rounds += 1; } try testing.expectEqual(n + 1, total); try testing.expectEqual(@as(u32, 3), rounds); try testing.expectEqual(@as(?i64, 1), try prefixOp(conn, "keep", "d"));}
// Destructive only to an explicitly supplied disposable fixture. This does// not connect to Turso or start a server. Run separately on two identical// copies (one indexed) to compare actual prefix projection and commit costs.test "offline real actor fanout benchmark (disposable database only)" { const path = std.c.getenv("TYPEAHEAD_FANOUT_SCRATCH") orelse return error.SkipZigTest; const arm = std.c.getenv("TYPEAHEAD_FANOUT_ALLOW_WRITES") orelse return error.SkipZigTest; if (!std.mem.eql(u8, std.mem.span(arm), "disposable-copy")) return error.SkipZigTest; var conn = try zqlite.open(std.mem.span(path), zqlite.OpenFlags.ReadWrite); defer conn.close(); try conn.exec("PRAGMA journal_mode=WAL", .{}); try conn.exec("PRAGMA synchronous=NORMAL", .{}); try conn.exec("PRAGMA wal_autocheckpoint=1000", .{}); try conn.exec("PRAGMA cache_size=-20000", .{}); var arena_state = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena_state.deinit(); const arena = arena_state.allocator(); var actors: std.ArrayListUnmanaged(Actor) = .empty; { var rows = try conn.rows("SELECT did,blob FROM actor_overlay WHERE blob IS NOT NULL ORDER BY did LIMIT 2500", .{}); defer rows.deinit(); while (rows.next()) |row| { const blob = try arena.dupe(u8, row.blob(1)); const a = try pack.unpackActor(blob); try actors.append(arena, .{ .did = try arena.dupe(u8, row.text(0)), .handle = a.handle, .display_name = a.display, .avatar_url = a.avatar, .labels = a.labels, .associated = a.associated, .pds = a.pds, .created_at = a.created_at, .hidden = a.hidden, .quality_score = a.quality_score, }); } if (rows.err) |err| return err; } if (actors.items.len < 2500) return error.InsufficientRealActors; for ([_][]const u8{ "unchanged", "rescore", "rename", "insert", "delete" }, 0..) |scenario, si| { const start = std.Io.Timestamp.now(std.Options.debug_io, .awake).nanoseconds; try conn.exec("BEGIN IMMEDIATE", .{}); errdefer conn.exec("ROLLBACK", .{}) catch {}; for (actors.items[si * 500 .. (si + 1) * 500]) |original| { var a = original; if (si == 4) { try projectDelete(arena, conn, a.did, a.handle, a.display_name, 2000000000); continue; } if (si == 1) a.quality_score += 1; if (si == 2) { a.handle = try std.fmt.allocPrint(arena, "eval-{s}", .{a.handle}); a.display_name = try std.fmt.allocPrint(arena, "Eval {s}", .{a.display_name}); } if (si == 3) a.did = try std.fmt.allocPrint(arena, "eval:{s}", .{a.did}); try projectUpdate(arena, conn, a, if (si == 3) "" else original.handle, if (si == 3) "" else original.display_name, 2000000000); } try conn.exec("COMMIT", .{}); const end = std.Io.Timestamp.now(std.Options.debug_io, .awake).nanoseconds; std.debug.print("FANOUT scenario={s} actors=500 committed_us={d}\n", .{ scenario, @divTrunc(end - start, 1000) }); }}