//! disk persistence matching indigo's diskpersist format
//!
//! append-only log files with relay-assigned sequence numbers.
//! Postgres metadata index for fast cursor→file lookup.
//!
//! on-disk entry format (28-byte LE header + CBOR payload):
//!   [4B flags LE] [4B kind LE] [4B payload_len LE] [8B uid LE] [8B seq LE] [payload]
//!
//! file naming: evts-{startSeq} (rotated every events_per_file events)
//!
//! see: indigo cmd/relay/stream/persist/diskpersist/diskpersist.go

const std = @import("std");
const Io = std.Io;
const pg = @import("pg");
const lru = @import("util/lru.zig");
const util = @import("util/util.zig");

const getenv = util.getenv;
const Allocator = std.mem.Allocator;
const log = std.log.scoped(.relay);

// --- constants matching indigo ---

pub const header_size: usize = 28; // 4 + 4 + 4 + 8 + 8

pub const EvtKind = enum(u32) {
    commit = 1,
    handle = 2, // deprecated
    tombstone = 3, // deprecated
    identity = 4,
    account = 5,
    sync = 6,
};

pub const EvtFlags = struct {
    pub const takedown: u32 = 1;
    pub const rebased: u32 = 2;
};

const default_events_per_file: u32 = 10_000;
const default_flush_interval_ms: u64 = 100;
const default_flush_threshold: usize = 400;
/// upper bound on a single record's payload, used to reject a corrupt length
/// during recovery. firehose frames are well under this; a larger value means a
/// torn/garbage header, not a real record.
const max_record_len: u32 = 16 * 1024 * 1024;

// --- header ---

pub const EvtHeader = struct {
    flags: u32,
    kind: u32,
    len: u32, // payload length (not including header)
    uid: u64,
    seq: u64,

    pub fn encode(self: EvtHeader, buf: *[header_size]u8) void {
        std.mem.writeInt(u32, buf[0..4], self.flags, .little);
        std.mem.writeInt(u32, buf[4..8], self.kind, .little);
        std.mem.writeInt(u32, buf[8..12], self.len, .little);
        std.mem.writeInt(u64, buf[12..20], self.uid, .little);
        std.mem.writeInt(u64, buf[20..28], self.seq, .little);
    }

    pub fn decode(buf: *const [header_size]u8) EvtHeader {
        return .{
            .flags = std.mem.readInt(u32, buf[0..4], .little),
            .kind = std.mem.readInt(u32, buf[4..8], .little),
            .len = std.mem.readInt(u32, buf[8..12], .little),
            .uid = std.mem.readInt(u64, buf[12..20], .little),
            .seq = std.mem.readInt(u64, buf[20..28], .little),
        };
    }
};

// --- persist job (write buffer entry) ---

const PersistJob = struct {
    data: []u8, // header + payload, owned
    seq: u64, // assigned seq (for broadcast ordering)
};

// --- disk persistence ---

/// bounds on a single playback call. a consumer connect with an old cursor
/// must never materialize the whole retained backlog in one shot — that is
/// an O(retention) allocation per connect, and a flapping consumer turns it
/// into an OOM (see docs/handoffs/HANDOFF-2026-08-06-consumer-flap-oom.md).
/// callers replay in bounded chunks instead, advancing `since` between calls.
pub const PlaybackLimits = struct {
    max_entries: usize = std.math.maxInt(usize),
    max_bytes: usize = std.math.maxInt(usize),
};

/// playback request — caller posts, pool_io worker executes
pub const PlaybackRequest = struct {
    since: u64,
    allocator: Allocator,
    limits: PlaybackLimits = .{},
    entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty,
    err: ?anyerror = null,
    done: std.atomic.Value(bool) = .{ .raw = false },
    next: std.atomic.Value(?*PlaybackRequest) = .{ .raw = null },
};

/// DB request — caller posts, pool_io worker executes.
/// callers define typed structs embedding DbRequest + @fieldParentPtr.
pub const DbRequest = struct {
    callback: *const fn (*DbRequest, *DiskPersist) void,
    done: std.atomic.Value(bool) = .{ .raw = false },
    err: ?anyerror = null,

    /// spin-wait for completion, yielding via io.sleep when available.
    /// never returns until `done` is true — the request struct lives on the
    /// caller's stack, so we must outlive the worker's callback execution.
    /// the queue's shutdown drain sets done+err on unprocessed requests,
    /// and workers set done after the callback returns, so this always terminates.
    pub fn wait(self: *DbRequest, io: Io, shutdown: *std.atomic.Value(bool)) void {
        _ = shutdown;
        while (!self.done.load(.acquire)) {
            io.sleep(Io.Duration.fromMicroseconds(100), .awake) catch {
                // sleep failed (Io shutting down) — busy-wait for worker to finish
                while (!self.done.load(.acquire)) {
                    std.atomic.spinLoopHint();
                }
                break;
            };
        }
    }
};

/// MPSC FIFO ring buffer for general DB traffic.
/// multiple producers push via spinlock,
/// multiple consumers (pool_io worker threads) pop via CAS.
pub const DbRequestQueue = struct {
    const CAPACITY = 4096;

    items: [CAPACITY]*DbRequest = undefined,
    head: std.atomic.Value(u32) = .{ .raw = 0 },
    tail: std.atomic.Value(u32) = .{ .raw = 0 },
    push_lock: std.atomic.Value(u32) = .{ .raw = 0 }, // producer spinlock
    /// total requests handled across all worker threads. exposed as
    /// relay_db_queue_handled_total — if this stops advancing while
    /// the queue has depth, the workers are dead or stuck.
    handled: std.atomic.Value(u64) = .{ .raw = 0 },
    shutdown: *std.atomic.Value(bool),
    persist: *DiskPersist,

    /// current depth = tail - head. exposed as relay_db_queue_depth
    /// so operators can correlate handler hangs with queue backup.
    pub fn depth(self: *const DbRequestQueue) u32 {
        const t = self.tail.load(.acquire);
        const h = self.head.load(.acquire);
        return t -% h;
    }

    pub fn push(self: *DbRequestQueue, req: *DbRequest) void {
        // acquire producer spinlock
        while (self.push_lock.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
            std.atomic.spinLoopHint();
        }
        defer self.push_lock.store(0, .release);

        const tail = self.tail.load(.monotonic);
        const head = self.head.load(.acquire);
        // if full, spin until space opens (workers are fast, this shouldn't happen)
        if (tail -% head >= CAPACITY) {
            self.push_lock.store(0, .release);
            while (self.tail.load(.monotonic) -% self.head.load(.acquire) >= CAPACITY) {
                std.atomic.spinLoopHint();
            }
            while (self.push_lock.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
                std.atomic.spinLoopHint();
            }
        }

        self.items[self.tail.load(.monotonic) % CAPACITY] = req;
        self.tail.store(self.tail.load(.monotonic) +% 1, .release);
    }

    pub fn pop(self: *DbRequestQueue) ?*DbRequest {
        while (true) {
            const head = self.head.load(.acquire);
            if (head == self.tail.load(.acquire)) return null;
            const item = self.items[head % CAPACITY];
            if (self.head.cmpxchgWeak(head, head +% 1, .acq_rel, .acquire) == null)
                return item;
        }
    }

    pub fn run(self: *DbRequestQueue, pool_io: Io) void {
        log.info("db_queue worker: starting", .{});
        var consecutive_sleep_errors: u32 = 0;
        while (!self.shutdown.load(.acquire)) {
            if (self.pop()) |req| {
                req.callback(req, self.persist);
                req.done.store(true, .release);
                _ = self.handled.fetchAdd(1, .monotonic);
                consecutive_sleep_errors = 0;
            } else {
                pool_io.sleep(Io.Duration.fromMilliseconds(5), .awake) catch |e| {
                    // previous behavior: silently `return`, killing the worker
                    // forever. one of these errors during a transient io hiccup
                    // would wedge requestCrawl + every other DB-bound handler.
                    // log + continue. spin-yield to avoid a tight error loop.
                    consecutive_sleep_errors += 1;
                    if (consecutive_sleep_errors == 1 or consecutive_sleep_errors % 1000 == 0) {
                        log.warn("db_queue worker: pool_io.sleep failed ({s}), continuing (consecutive={d})", .{ @errorName(e), consecutive_sleep_errors });
                    }
                    std.atomic.spinLoopHint();
                };
            }
        }
        // shutdown drain — signal error on unprocessed requests
        while (self.pop()) |req| {
            req.err = error.ShuttingDown;
            req.done.store(true, .release);
        }
        log.info("db_queue worker: exiting (shutdown=true, handled_total={d})", .{self.handled.load(.acquire)});
    }
};

pub const DiskPersist = struct {
    allocator: Allocator,
    dir_path: []const u8,
    dir: Io.Dir,
    db: *pg.Pool,
    current_file: ?Io.File = null,
    current_file_path: ?[]const u8 = null,
    current_file_pos: u64 = 0,

    // sequence state
    cur_seq: u64 = 1,
    event_counter: u64 = 0,

    // config
    events_per_file: u32 = default_events_per_file,
    retention_hours: u64 = 72, // 3 days
    max_dir_bytes: u64 = 100 * 1024 * 1024 * 1024, // 100 GB

    // DID → UID cache (matches indigo's bidirectional ARC cache)
    did_cache: lru.LruCache(u64),

    // UID → host_id cache. host_id mutates ~5 orders of magnitude slower than
    // it's read (every event hits getAccountHostId) and setAccountHostId is its
    // sole writer, so a read-through cache eliminates a per-event point read.
    // keyed by the uid's raw bytes; only nonzero host_ids are cached (0 means
    // "not yet assigned", a transient state that resolves on the next commit).
    host_id_cache: lru.LruCache(u64),

    // write buffer (flushed periodically or when threshold hit)
    outbuf: std.ArrayListUnmanaged(u8) = .empty,
    evtbuf: std.ArrayListUnmanaged(PersistJob) = .empty,
    mutex: Io.Mutex = Io.Mutex.init,

    // flush task
    flush_future: ?Io.Future(void) = null,
    alive: std.atomic.Value(bool) = .{ .raw = true },

    // highest seq that is durably on disk. advanced only by flushLocked, after the
    // write. the broadcaster fiber refuses to emit a queued frame until its seq is
    // <= committed_seq, so a frame never reaches consumers before it is durable —
    // which means the seq recovered on restart can't trail an already-broadcast
    // seq (the post-crash rewind). see docs/incident-2026-05-31-seq-rewind.md.
    committed_seq: std.atomic.Value(u64) = .{ .raw = 0 },

    io: Io,

    /// last successful DB interaction (epoch seconds, set by Threaded workers).
    /// read by metrics server to report health without direct pg.Pool access.
    last_db_success: std.atomic.Value(i64) = .{ .raw = 0 },

    /// MPSC queue for playback requests (caller → pool_io worker)
    playback_head: std.atomic.Value(?*PlaybackRequest) = .{ .raw = null },

    /// current evtbuf entry count (for metrics — non-blocking, returns 0 if lock is contended)
    pub fn evtbufLen(self: *DiskPersist) usize {
        if (!self.mutex.tryLock()) return 0;
        defer self.mutex.unlock(self.io);
        return self.evtbuf.items.len;
    }

    /// current DID cache entry count (for metrics)
    pub fn didCacheLen(self: *DiskPersist) u32 {
        return self.did_cache.count();
    }

    /// DID cache hashmap backing capacity (for memory attribution)
    pub fn didCacheMapCap(self: *DiskPersist) u32 {
        return self.did_cache.mapCapacity();
    }

    /// evtbuf allocated capacity in jobs (for memory attribution — non-blocking)
    pub fn evtbufCap(self: *DiskPersist) usize {
        if (!self.mutex.tryLock()) return 0;
        defer self.mutex.unlock(self.io);
        return self.evtbuf.capacity;
    }

    /// outbuf allocated capacity in bytes (for memory attribution — non-blocking)
    pub fn outbufCap(self: *DiskPersist) usize {
        if (!self.mutex.tryLock()) return 0;
        defer self.mutex.unlock(self.io);
        return self.outbuf.capacity;
    }

    pub fn init(allocator: Allocator, dir_path: []const u8, database_url: []const u8, db_pool_size: u16, io: Io) !DiskPersist {
        // ensure directory exists
        try Io.Dir.cwd().createDirPath(io, dir_path);

        var dir = try Io.Dir.cwd().openDir(io, dir_path, .{ .iterate = true });
        errdefer dir.close(io);

        // connect to Postgres
        const uri = std.Uri.parse(database_url) catch return error.InvalidDatabaseUrl;
        const pool = try pg.Pool.initUri(io, allocator, uri, .{ .size = db_pool_size });
        errdefer pool.deinit();

        // create tables (matching indigo's Go relay schema)
        _ = try pool.exec(
            \\CREATE TABLE IF NOT EXISTS log_file_refs (
            \\  id BIGSERIAL PRIMARY KEY,
            \\  path TEXT NOT NULL,
            \\  archived BOOLEAN NOT NULL DEFAULT false,
            \\  seq_start BIGINT NOT NULL,
            \\  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
            \\)
        , .{});

        _ = try pool.exec(
            \\CREATE TABLE IF NOT EXISTS account (
            \\  uid BIGSERIAL PRIMARY KEY,
            \\  did TEXT NOT NULL UNIQUE,
            \\  host_id BIGINT NOT NULL DEFAULT 0,
            \\  status TEXT NOT NULL DEFAULT 'active',
            \\  upstream_status TEXT NOT NULL DEFAULT 'active',
            \\  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
            \\)
        , .{});

        // migration: add columns if they don't exist (for existing deployments)
        _ = pool.exec("ALTER TABLE account ADD COLUMN IF NOT EXISTS host_id BIGINT NOT NULL DEFAULT 0", .{}) catch {};
        _ = pool.exec("ALTER TABLE account ADD COLUMN IF NOT EXISTS upstream_status TEXT NOT NULL DEFAULT 'active'", .{}) catch {};

        // index host_id so the per-host account COUNT(*) (new-account admission)
        // is an index scan instead of an 8.7M-row seq scan. CONCURRENTLY avoids
        // locking the live account table during the build; IF NOT EXISTS makes
        // re-runs (and resuming a failed build) safe.
        _ = pool.exec("CREATE INDEX CONCURRENTLY IF NOT EXISTS account_host_id_idx ON account (host_id)", .{}) catch {};

        _ = try pool.exec(
            \\CREATE TABLE IF NOT EXISTS account_repo (
            \\  uid BIGINT PRIMARY KEY REFERENCES account(uid),
            \\  rev TEXT NOT NULL,
            \\  commit_data_cid TEXT NOT NULL
            \\)
        , .{});

        _ = try pool.exec(
            \\CREATE TABLE IF NOT EXISTS domain_ban (
            \\  id BIGSERIAL PRIMARY KEY,
            \\  domain TEXT NOT NULL UNIQUE,
            \\  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
            \\)
        , .{});

        _ = try pool.exec(
            \\CREATE TABLE IF NOT EXISTS host (
            \\  id BIGSERIAL PRIMARY KEY,
            \\  hostname TEXT NOT NULL UNIQUE,
            \\  status TEXT NOT NULL DEFAULT 'active',
            \\  last_seq BIGINT NOT NULL DEFAULT 0,
            \\  failed_attempts INTEGER NOT NULL DEFAULT 0,
            \\  account_limit BIGINT,
            \\  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
            \\  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
            \\)
        , .{});

        // migration: add account_limit column to existing host tables
        _ = try pool.exec("ALTER TABLE host ADD COLUMN IF NOT EXISTS account_limit BIGINT", .{});

        _ = try pool.exec(
            \\CREATE TABLE IF NOT EXISTS backfill_progress (
            \\  collection TEXT NOT NULL,
            \\  source TEXT NOT NULL,
            \\  cursor TEXT NOT NULL DEFAULT '',
            \\  imported_count BIGINT NOT NULL DEFAULT 0,
            \\  completed_at TIMESTAMPTZ,
            \\  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
            \\  PRIMARY KEY (collection, source)
            \\)
        , .{});

        // migrate: old schema had collection as sole PK — add source to composite PK
        _ = pool.exec(
            \\DO $$ BEGIN
            \\  IF EXISTS (
            \\    SELECT 1 FROM pg_constraint
            \\    WHERE conname = 'backfill_progress_pkey'
            \\    AND conrelid = 'backfill_progress'::regclass
            \\    AND array_length(conkey, 1) = 1
            \\  ) THEN
            \\    ALTER TABLE backfill_progress DROP CONSTRAINT backfill_progress_pkey;
            \\    ALTER TABLE backfill_progress ADD PRIMARY KEY (collection, source);
            \\  END IF;
            \\END $$
        , .{}) catch {};

        var self = DiskPersist{
            .allocator = allocator,
            .dir_path = try allocator.dupe(u8, dir_path),
            .dir = dir,
            .db = pool,
            .did_cache = lru.LruCache(u64).init(allocator, 500_000, io),
            .host_id_cache = lru.LruCache(u64).init(allocator, 500_000, io),
            .io = io,
        };

        // recover from existing log files
        try self.resumeLog();

        // everything already on disk is committed; cur_seq points one past it.
        self.committed_seq.store(self.cur_seq -| 1, .monotonic);

        return self;
    }

    pub fn deinit(self: *DiskPersist) void {
        // stop flush task (loop checks alive after each sleep interval)
        self.alive.store(false, .release);
        if (self.flush_future) |*f| f.cancel(self.io);

        // flush remaining
        self.mutex.lockUncancelable(self.io);
        self.flushLocked() catch |err| log.err("final event flush failed: {s}", .{@errorName(err)});
        self.mutex.unlock(self.io);

        // free write buffer
        for (self.evtbuf.items) |job| self.allocator.free(job.data);
        self.evtbuf.deinit(self.allocator);
        self.outbuf.deinit(self.allocator);

        self.did_cache.deinit();
        self.host_id_cache.deinit();

        if (self.current_file) |f| f.close(self.io);
        if (self.current_file_path) |p| self.allocator.free(p);
        self.dir.close(self.io);
        self.db.deinit();
        self.allocator.free(self.dir_path);
    }

    /// start the background flush thread
    pub fn start(self: *DiskPersist) !void {
        self.flush_future = try self.io.concurrent(flushLoop, .{self});
    }

    pub const UidResult = struct {
        uid: u64,
        host_changed: bool = false,
        is_new: bool = false,
    };

    /// resolve a DID to a numeric UID, checking host association.
    /// does NOT set host_id — caller must verify authority first via DID doc
    /// resolution, then call setAccountHostId on success.
    /// returns is_new=true when no host_id is set yet (first encounter),
    /// host_changed=true when host_id differs from stored value.
    pub fn uidForDidFromHost(self: *DiskPersist, did: []const u8, host_id: u64) !UidResult {
        const uid = try self.uidForDid(did);
        if (host_id > 0) {
            const current_host = self.getAccountHostId(uid) catch 0;
            if (current_host == 0) {
                return .{ .uid = uid, .is_new = true };
            } else if (current_host != host_id) {
                log.info("account {s} (uid={d}) host mismatch: current={d} new={d}", .{ did, uid, current_host, host_id });
                return .{ .uid = uid, .host_changed = true };
            }
        }
        return .{ .uid = uid };
    }

    /// check DB health without touching pg.Pool — safe from any Io context.
    /// returns true if a Threaded worker successfully queried the DB within the last 30s.
    pub fn isDbHealthy(self: *DiskPersist) bool {
        const last = self.last_db_success.load(.acquire);
        if (last == 0) return false;
        var ts: std.c.timespec = undefined;
        _ = std.c.clock_gettime(.REALTIME, &ts);
        return (@as(i64, ts.sec) - last) < 30;
    }

    fn markDbSuccess(self: *DiskPersist) void {
        var ts: std.c.timespec = undefined;
        _ = std.c.clock_gettime(.REALTIME, &ts);
        self.last_db_success.store(@as(i64, ts.sec), .release);
    }

    /// resolve a DID to a numeric UID. creates a new account row on first encounter.
    /// matches indigo's Relay.DidToUid → Account.UID mapping.
    pub fn uidForDid(self: *DiskPersist, did: []const u8) !u64 {
        // fast path: check in-memory cache
        if (self.did_cache.get(did)) |uid| return uid;

        // check database
        if (try self.db.rowUnsafe(
            "SELECT uid FROM account WHERE did = $1",
            .{did},
        )) |row| {
            var r = row;
            defer r.deinit() catch {};
            const uid: u64 = @intCast(r.get(i64, 0));
            self.didCachePut(did, uid);
            self.markDbSuccess();
            return uid;
        }

        // create new account row (ignore if already exists from concurrent insert)
        _ = self.db.exec(
            "INSERT INTO account (did) VALUES ($1) ON CONFLICT (did) DO NOTHING",
            .{did},
        ) catch |err| {
            log.warn("failed to create account for {s}: {s}", .{ did, @errorName(err) });
            return err;
        };

        // read back the UID (whether we just created it or it already existed)
        var row = try self.db.rowUnsafe(
            "SELECT uid FROM account WHERE did = $1",
            .{did},
        ) orelse return error.AccountCreationFailed;
        defer row.deinit() catch {};
        const uid: u64 = @intCast(row.get(i64, 0));

        self.didCachePut(did, uid);
        return uid;
    }

    /// insert into did_cache (LRU handles eviction at capacity).
    fn didCachePut(self: *DiskPersist, did: []const u8, uid: u64) void {
        self.did_cache.put(did, uid) catch {};
    }

    /// per-DID sync state for chain tracking
    pub const AccountState = struct {
        rev: []const u8,
        data_cid: []const u8,
    };

    /// get stored sync state for a user (from account_repo table)
    pub fn getAccountState(self: *DiskPersist, uid: u64, allocator: Allocator) !?AccountState {
        var row = (try self.db.rowUnsafe(
            "SELECT rev, commit_data_cid FROM account_repo WHERE uid = $1",
            .{@as(i64, @intCast(uid))},
        )) orelse return null;
        defer row.deinit() catch {};
        const rev = row.get([]const u8, 0);
        const data_cid = row.get([]const u8, 1);
        // A rev-only row is still prior repo state. Older zlay versions and
        // signature-only #sync handling stored an empty data CID; discarding
        // those rows disabled stale-rev and missing-prevData checks entirely.
        if (rev.len == 0) return null;
        return .{
            .rev = try allocator.dupe(u8, rev),
            .data_cid = try allocator.dupe(u8, data_cid),
        };
    }

    /// update stored sync state after a verified commit (conditional upsert into account_repo).
    /// uses WHERE clause to ensure rev only moves forward, preventing race conditions
    /// when concurrent workers process commits for the same DID.
    /// returns true if the state was updated, false if a newer rev was already stored.
    pub fn updateAccountState(self: *DiskPersist, uid: u64, rev: []const u8, data_cid: []const u8) !bool {
        const rows_affected = try self.db.exec(
            "INSERT INTO account_repo (uid, rev, commit_data_cid) VALUES ($1, $2, $3) ON CONFLICT (uid) DO UPDATE SET rev = EXCLUDED.rev, commit_data_cid = EXCLUDED.commit_data_cid WHERE account_repo.rev < EXCLUDED.rev",
            .{ @as(i64, @intCast(uid)), rev, data_cid },
        );
        return (rows_affected orelse 0) > 0;
    }

    // --- account status ---

    /// raw little-endian bytes of a uid, for use as an LRU key.
    fn uidKey(uid: u64) [8]u8 {
        return @bitCast(std.mem.nativeToLittle(u64, uid));
    }

    /// get the host_id for an account. returns 0 if not set.
    /// read-through cache: only nonzero host_ids are cached (0 is the transient
    /// pre-assignment state, which resolves via setAccountHostId on next commit).
    pub fn getAccountHostId(self: *DiskPersist, uid: u64) !u64 {
        const key = uidKey(uid);
        if (self.host_id_cache.get(&key)) |hid| return hid;

        var row = (try self.db.rowUnsafe(
            "SELECT host_id FROM account WHERE uid = $1",
            .{@as(i64, @intCast(uid))},
        )) orelse return 0;
        defer row.deinit() catch {};
        const hid = row.get(i64, 0);
        if (hid > 0) {
            const host_id: u64 = @intCast(hid);
            self.host_id_cache.put(&key, host_id) catch {};
            return host_id;
        }
        return 0;
    }

    /// count accounts on a host (Threaded pool)
    pub fn getHostAccountCount(self: *DiskPersist, host_id: u64) u64 {
        return getHostAccountCountImpl(host_id, self.db);
    }

    fn getHostAccountCountImpl(host_id: u64, db: *pg.Pool) u64 {
        var row = (db.rowUnsafe(
            "SELECT COUNT(*) FROM account WHERE host_id = $1",
            .{@as(i64, @intCast(host_id))},
        ) catch return 0) orelse return 0;
        defer row.deinit() catch {};
        const count = row.get(i64, 0);
        return if (count > 0) @intCast(count) else 0;
    }

    /// effective account count (Threaded pool)
    pub fn getEffectiveAccountCount(self: *DiskPersist, host_id: u64) u64 {
        return getEffectiveAccountCountImpl(host_id, self.db);
    }

    /// uses admin-configured limit if set, otherwise actual COUNT(*).
    /// CASE (not COALESCE) so the COUNT subquery is only evaluated when no
    /// account_limit is set — SQL doesn't short-circuit a COALESCE's aggregate.
    fn getEffectiveAccountCountImpl(host_id: u64, db: *pg.Pool) u64 {
        var row = (db.rowUnsafe(
            "SELECT CASE WHEN h.account_limit IS NOT NULL THEN h.account_limit ELSE (SELECT COUNT(*) FROM account a WHERE a.host_id = h.id) END FROM host h WHERE h.id = $1",
            .{@as(i64, @intCast(host_id))},
        ) catch return 0) orelse return 0;
        defer row.deinit() catch {};
        const count = row.get(i64, 0);
        return if (count > 0) @intCast(count) else 0;
    }

    /// set host account limit (Threaded pool)
    pub fn setHostAccountLimit(self: *DiskPersist, host_id: u64, limit: ?u64) !void {
        return setHostAccountLimitImpl(host_id, limit, self.db);
    }

    /// pass null to clear the override and revert to actual COUNT(*).
    fn setHostAccountLimitImpl(host_id: u64, limit: ?u64, db: *pg.Pool) !void {
        if (limit) |l| {
            const clamped: i64 = if (l > @as(u64, @intCast(std.math.maxInt(i64)))) std.math.maxInt(i64) else @intCast(l);
            _ = try db.exec(
                "UPDATE host SET account_limit = $2, updated_at = now() WHERE id = $1",
                .{ @as(i64, @intCast(host_id)), clamped },
            );
        } else {
            _ = try db.exec(
                "UPDATE host SET account_limit = NULL, updated_at = now() WHERE id = $1",
                .{@as(i64, @intCast(host_id))},
            );
        }
    }

    /// set the host_id for an account (first encounter or migration)
    pub fn setAccountHostId(self: *DiskPersist, uid: u64, host_id: u64) !void {
        _ = try self.db.exec(
            "UPDATE account SET host_id = $2 WHERE uid = $1",
            .{ @as(i64, @intCast(uid)), @as(i64, @intCast(host_id)) },
        );
        // write-through: sole writer, so the cache stays coherent with the DB.
        const key = uidKey(uid);
        if (host_id > 0) {
            self.host_id_cache.put(&key, host_id) catch {};
        } else {
            _ = self.host_id_cache.remove(&key);
        }
    }

    /// update the upstream status for an account (from #account events).
    /// Go relay: account.go UpdateAccountUpstreamStatus
    pub fn updateAccountUpstreamStatus(self: *DiskPersist, uid: u64, upstream_status: []const u8) !void {
        _ = try self.db.exec(
            "UPDATE account SET upstream_status = $2 WHERE uid = $1",
            .{ @as(i64, @intCast(uid)), upstream_status },
        );
    }

    /// check if an account is active (both local status and upstream status).
    /// Go relay: models.Account.IsActive()
    pub const AccountActivity = struct {
        local_ok: bool,
        upstream_ok: bool,

        pub fn active(self: AccountActivity) bool {
            return self.local_ok and self.upstream_ok;
        }
    };

    /// account activity split by source: local (relay takedown) vs upstream
    /// (PDS-reported via #account events). callers that drop events for
    /// upstream-inactive accounts use the distinction to trigger a
    /// StatusChecker re-check — local takedowns must never be re-checked away.
    pub fn accountActivity(self: *DiskPersist, uid: u64) !AccountActivity {
        var row = (try self.db.rowUnsafe(
            "SELECT status, upstream_status FROM account WHERE uid = $1",
            .{@as(i64, @intCast(uid))},
        )) orelse return .{ .local_ok = false, .upstream_ok = false };
        defer row.deinit() catch {};
        const status = row.get([]const u8, 0);
        const upstream = row.get([]const u8, 1);
        return .{
            .local_ok = std.mem.eql(u8, status, "active"),
            .upstream_ok = std.mem.eql(u8, upstream, "active"),
        };
    }

    pub fn isAccountActive(self: *DiskPersist, uid: u64) !bool {
        return (try self.accountActivity(uid)).active();
    }

    // --- host management ---

    pub const Host = struct {
        id: u64,
        hostname: []const u8,
        status: []const u8,
        last_seq: u64,
        failed_attempts: u32,
        account_limit: ?u64 = null,
    };

    const HostResult = struct { id: u64, last_seq: u64 };

    /// get or create a host row (Threaded pool)
    pub fn getOrCreateHost(self: *DiskPersist, hostname: []const u8) !HostResult {
        return getOrCreateHostImpl(hostname, self.db);
    }

    fn getOrCreateHostImpl(hostname: []const u8, db: *pg.Pool) !HostResult {
        _ = db.exec(
            "INSERT INTO host (hostname) VALUES ($1) ON CONFLICT (hostname) DO NOTHING",
            .{hostname},
        ) catch |err| {
            log.warn("failed to create host {s}: {s}", .{ hostname, @errorName(err) });
            return err;
        };

        var row = try db.rowUnsafe(
            "SELECT id, last_seq FROM host WHERE hostname = $1",
            .{hostname},
        ) orelse return error.HostCreationFailed;
        defer row.deinit() catch {};
        return .{
            .id = @intCast(row.get(i64, 0)),
            .last_seq = @intCast(row.get(i64, 1)),
        };
    }

    /// check if a host is banned or blocked by status (Threaded pool)
    pub fn isHostBanned(self: *DiskPersist, hostname: []const u8) bool {
        return isHostBannedImpl(hostname, self.db);
    }

    fn isHostBannedImpl(hostname: []const u8, db: *pg.Pool) bool {
        var row = db.rowUnsafe(
            "SELECT status FROM host WHERE hostname = $1",
            .{hostname},
        ) catch return false;
        if (row) |*r| {
            defer r.deinit() catch {};
            const status = r.get([]const u8, 0);
            return std.mem.eql(u8, status, "banned") or std.mem.eql(u8, status, "blocked");
        }
        return false;
    }

    /// update cursor position for a host
    pub fn updateHostSeq(self: *DiskPersist, host_id: u64, seq: u64) !void {
        _ = try self.db.exec(
            "UPDATE host SET last_seq = $2, updated_at = now() WHERE id = $1",
            .{ @as(i64, @intCast(host_id)), @as(i64, @intCast(seq)) },
        );
    }

    /// look up host ID by hostname (Threaded pool)
    pub fn getHostIdForHostname(self: *DiskPersist, hostname: []const u8) !?u64 {
        return getHostIdForHostnameImpl(hostname, self.db);
    }

    fn getHostIdForHostnameImpl(hostname: []const u8, db: *pg.Pool) !?u64 {
        var row = (try db.rowUnsafe(
            "SELECT id FROM host WHERE hostname = $1",
            .{hostname},
        )) orelse return null;
        defer row.deinit() catch {};
        return @intCast(row.get(i64, 0));
    }

    /// update host status (Threaded pool)
    pub fn updateHostStatus(self: *DiskPersist, host_id: u64, status: []const u8) !void {
        return updateHostStatusImpl(host_id, status, self.db);
    }

    fn updateHostStatusImpl(host_id: u64, status: []const u8, db: *pg.Pool) !void {
        _ = try db.exec(
            "UPDATE host SET status = $2, updated_at = now() WHERE id = $1",
            .{ @as(i64, @intCast(host_id)), status },
        );
    }

    /// list all active hosts (Threaded pool)
    pub fn listActiveHosts(self: *DiskPersist, allocator: Allocator) ![]Host {
        return listActiveHostsImpl(allocator, self.db);
    }

    fn listActiveHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host {
        var hosts: std.ArrayListUnmanaged(Host) = .empty;
        errdefer {
            for (hosts.items) |host| {
                allocator.free(host.hostname);
                allocator.free(host.status);
            }
            hosts.deinit(allocator);
        }

        var result = try db.query(
            "SELECT id, hostname, status, last_seq, failed_attempts, account_limit FROM host WHERE status = 'active' ORDER BY id ASC",
            .{},
        );
        defer result.deinit();

        while (result.nextUnsafe() catch null) |row| {
            try hosts.append(allocator, .{
                .id = @intCast(row.get(i64, 0)),
                .hostname = try allocator.dupe(u8, row.get([]const u8, 1)),
                .status = try allocator.dupe(u8, row.get([]const u8, 2)),
                .last_seq = @intCast(row.get(i64, 3)),
                .failed_attempts = @intCast(row.get(i32, 4)),
                .account_limit = if (row.get(?i64, 5)) |v| @as(?u64, @intCast(v)) else null,
            });
        }

        return try hosts.toOwnedSlice(allocator);
    }

    /// list all hosts (Threaded pool)
    pub fn listAllHosts(self: *DiskPersist, allocator: Allocator) ![]Host {
        return listAllHostsImpl(allocator, self.db);
    }

    fn listAllHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host {
        var hosts: std.ArrayListUnmanaged(Host) = .empty;
        errdefer {
            for (hosts.items) |host| {
                allocator.free(host.hostname);
                allocator.free(host.status);
            }
            hosts.deinit(allocator);
        }

        var result = try db.query(
            "SELECT id, hostname, status, last_seq, failed_attempts, account_limit FROM host ORDER BY id ASC",
            .{},
        );
        defer result.deinit();

        while (result.nextUnsafe() catch null) |row| {
            try hosts.append(allocator, .{
                .id = @intCast(row.get(i64, 0)),
                .hostname = try allocator.dupe(u8, row.get([]const u8, 1)),
                .status = try allocator.dupe(u8, row.get([]const u8, 2)),
                .last_seq = @intCast(row.get(i64, 3)),
                .failed_attempts = @intCast(row.get(i32, 4)),
                .account_limit = if (row.get(?i64, 5)) |v| @as(?u64, @intCast(v)) else null,
            });
        }

        return try hosts.toOwnedSlice(allocator);
    }

    /// increment failure count and return new value
    pub fn incrementHostFailures(self: *DiskPersist, host_id: u64) !u32 {
        _ = try self.db.exec(
            "UPDATE host SET failed_attempts = failed_attempts + 1, updated_at = now() WHERE id = $1",
            .{@as(i64, @intCast(host_id))},
        );
        var row = try self.db.rowUnsafe(
            "SELECT failed_attempts FROM host WHERE id = $1",
            .{@as(i64, @intCast(host_id))},
        ) orelse return 0;
        defer row.deinit() catch {};
        return @intCast(row.get(i32, 0));
    }

    /// check if a hostname (or any parent domain) is banned (Threaded pool).
    pub fn isDomainBanned(self: *DiskPersist, hostname: []const u8) bool {
        return self.isDomainBannedImpl(hostname, self.db);
    }

    /// Go relay: domain_ban.go DomainIsBanned — suffix-based check.
    fn isDomainBannedImpl(_: *DiskPersist, hostname: []const u8, db: *pg.Pool) bool {
        // check each suffix: "pds.host.example.com", "host.example.com", "example.com"
        var offset: usize = 0;
        while (offset < hostname.len) {
            const suffix = hostname[offset..];
            var row = db.rowUnsafe(
                "SELECT 1 FROM domain_ban WHERE domain = $1",
                .{suffix},
            ) catch return false;
            if (row) |*r| {
                r.deinit() catch {};
                return true;
            }
            // advance past next dot
            if (std.mem.indexOfScalarPos(u8, hostname, offset, '.')) |dot| {
                offset = dot + 1;
            } else break;
        }
        return false;
    }

    /// reset failure count (Threaded pool)
    pub fn resetHostFailures(self: *DiskPersist, host_id: u64) !void {
        return resetHostFailuresImpl(host_id, self.db);
    }

    fn resetHostFailuresImpl(host_id: u64, db: *pg.Pool) !void {
        _ = try db.exec(
            "UPDATE host SET failed_attempts = 0, updated_at = now() WHERE id = $1",
            .{@as(i64, @intCast(host_id))},
        );
    }

    /// enqueue a playback request for the pool_io worker to execute.
    /// uses MPSC push (lock-free Treiber stack).
    pub fn enqueuePlayback(self: *DiskPersist, req: *PlaybackRequest) void {
        while (true) {
            const head = self.playback_head.load(.acquire);
            req.next.store(head, .monotonic);
            if (self.playback_head.cmpxchgWeak(@as(?*PlaybackRequest, head), req, .release, .monotonic) == null) {
                return;
            }
        }
    }

    /// pop all pending playback requests (single-consumer — pool_io worker only).
    /// returns a singly-linked list via .next pointers, or null if empty.
    pub fn popPlaybackBatch(self: *DiskPersist) ?*PlaybackRequest {
        return self.playback_head.swap(null, .acq_rel);
    }

    /// persist an event. assigns a sequence number. returns the assigned seq.
    /// the event is buffered and will be flushed to disk asynchronously.
    pub fn persist(self: *DiskPersist, kind: EvtKind, uid: u64, payload: []const u8) !u64 {
        // build the on-disk record: header + payload
        const total_len = header_size + payload.len;
        const data = try self.allocator.alloc(u8, total_len);
        errdefer self.allocator.free(data);

        // write header (seq filled in later under lock)
        const header = EvtHeader{
            .flags = 0,
            .kind = @intFromEnum(kind),
            .len = @intCast(payload.len),
            .uid = uid,
            .seq = 0, // placeholder
        };
        header.encode(data[0..header_size]);
        @memcpy(data[header_size..], payload);

        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);

        if (self.evtbuf.items.len >= default_flush_threshold) try self.flushLocked();
        try self.evtbuf.ensureUnusedCapacity(self.allocator, 1);
        try self.outbuf.ensureUnusedCapacity(self.allocator, data.len);

        // assign seq
        const seq = self.cur_seq;
        self.cur_seq += 1;
        std.mem.writeInt(u64, data[20..28], seq, .little);

        self.evtbuf.appendAssumeCapacity(.{ .data = data, .seq = seq });
        self.outbuf.appendSliceAssumeCapacity(data);
        self.event_counter += 1;

        // flush if threshold hit
        if (self.evtbuf.items.len >= default_flush_threshold) {
            self.flushLocked() catch |err| log.warn("event flush deferred: {s}", .{@errorName(err)});
        }

        return seq;
    }

    /// playback events with seq > since, bounded by `limits`. a truncated
    /// result is detected by the caller via entries hitting the limits;
    /// resume by calling again with since = last returned seq.
    pub fn playback(self: *DiskPersist, since: u64, allocator: Allocator, entries: *std.ArrayListUnmanaged(PlaybackEntry), limits: PlaybackLimits) !void {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);

        const since_i: i64 = @intCast(since);

        // find the log file containing `since`
        var start_files: std.ArrayListUnmanaged(LogFileRef) = .empty;
        defer start_files.deinit(allocator);

        if (since > 0) {
            // find file whose seq_start is just before `since`
            if (try self.db.rowUnsafe(
                "SELECT id, path, seq_start FROM log_file_refs WHERE seq_start <= $1 ORDER BY seq_start DESC LIMIT 1",
                .{since_i},
            )) |row| {
                var r = row;
                defer r.deinit() catch {};
                try start_files.append(allocator, .{
                    .path = try allocator.dupe(u8, r.get([]const u8, 1)),
                    .seq_start = @intCast(r.get(i64, 2)),
                });
            }
        }

        // find all subsequent files
        {
            var result = try self.db.query(
                "SELECT id, path, seq_start FROM log_file_refs WHERE seq_start > $1 ORDER BY seq_start ASC",
                .{since_i},
            );
            defer result.deinit();
            while (result.nextUnsafe() catch null) |r| {
                try start_files.append(allocator, .{
                    .path = try allocator.dupe(u8, r.get([]const u8, 1)),
                    .seq_start = @intCast(r.get(i64, 2)),
                });
            }
        }

        defer for (start_files.items) |f| allocator.free(f.path);

        // read events from each file until the limits are hit
        var bytes_used: usize = 0;
        for (start_files.items) |ref| {
            if (entries.items.len >= limits.max_entries or bytes_used >= limits.max_bytes) break;
            var file = self.dir.openFile(self.io, ref.path, .{}) catch continue;
            defer file.close(self.io);
            try readEventsFrom(allocator, file, self.io, since, entries, limits, &bytes_used);
        }
    }

    /// last assigned sequence number, or null if empty
    pub fn lastSeq(self: *DiskPersist) ?u64 {
        if (self.cur_seq <= 1) return null;
        return self.cur_seq - 1;
    }

    /// oldest available sequence number (Threaded pool)
    pub fn firstSeq(self: *DiskPersist) ?u64 {
        return firstSeqImpl(self.db);
    }

    fn firstSeqImpl(db: *pg.Pool) ?u64 {
        if (db.rowUnsafe(
            "SELECT seq_start FROM log_file_refs ORDER BY seq_start ASC LIMIT 1",
            .{},
        ) catch null) |row| {
            var r = row;
            defer r.deinit() catch {};
            const v = r.get(i64, 0);
            return if (v > 0) @intCast(v) else null;
        }
        return null;
    }

    /// garbage-collect log files older than the retention period
    pub fn gc(self: *DiskPersist) !void {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);

        const cutoff_interval = try std.fmt.allocPrint(self.allocator, "{d} hours", .{self.retention_hours});
        defer self.allocator.free(cutoff_interval);

        // find expired refs
        var expired: std.ArrayListUnmanaged(GcRef) = .empty;
        defer {
            for (expired.items) |e| self.allocator.free(e.path);
            expired.deinit(self.allocator);
        }

        {
            var result = try self.db.query(
                "SELECT id, path FROM log_file_refs WHERE created_at < now() - $1::interval",
                .{cutoff_interval},
            );
            defer result.deinit();
            while (result.nextUnsafe() catch null) |r| {
                try expired.append(self.allocator, .{
                    .id = r.get(i64, 0),
                    .path = try self.allocator.dupe(u8, r.get([]const u8, 1)),
                });
            }
        }

        for (expired.items) |ref| {
            // skip current file
            if (self.current_file_path) |cur| {
                if (std.mem.eql(u8, ref.path, cur)) continue;
            }

            // delete db record first (prevents playback from finding it)
            _ = self.db.exec("DELETE FROM log_file_refs WHERE id = $1", .{ref.id}) catch |err| {
                log.warn("gc: failed to delete db record {d}: {s}", .{ ref.id, @errorName(err) });
                continue;
            };

            // delete file
            self.dir.deleteFile(self.io, ref.path) catch |err| {
                log.warn("gc: failed to delete {s}: {s}", .{ ref.path, @errorName(err) });
            };
        }

        if (expired.items.len > 0) {
            log.info("gc: removed {d} expired log files", .{expired.items.len});
        }

        // size-based GC: delete oldest files until total size is under max_dir_bytes
        if (self.max_dir_bytes > 0) {
            try self.gcBySize();
        }

        self.gcOrphans() catch |err| {
            log.warn("gc: orphan sweep failed: {s}", .{@errorName(err)});
        };

        self.markDbSuccess();
    }

    /// delete segment files that no log_file_refs row references and that are
    /// strictly older (by start seq, from the evts-{seq} name) than everything
    /// still registered. gc deletes the row before the file — deliberately, so
    /// playback can never resolve a missing file — which means a failed file
    /// delete leaves an orphan no later time-based pass can see: the row is
    /// gone, and the file leaks forever. that is how production accumulated
    /// ~100G / 10 weeks of unreferenced segments (found 2026-08-27). playback
    /// resolves every read through log_file_refs, so an unreferenced file
    /// below the retained window is unreachable and safe to remove while live.
    fn gcOrphans(self: *DiskPersist) !void {
        var min_registered: ?u64 = null;
        {
            var result = try self.db.query("SELECT min(seq_start) FROM log_file_refs", .{});
            defer result.deinit();
            while (result.nextUnsafe() catch null) |r| {
                if (r.get(?i64, 0)) |v| min_registered = @intCast(v);
            }
        }
        // no registered segments at all: nothing is provably expired — a
        // fresh DB over an old data dir must not eat the recovery source
        const floor = min_registered orelse return;

        var doomed: std.ArrayListUnmanaged([]u8) = .empty;
        defer {
            for (doomed.items) |p| self.allocator.free(p);
            doomed.deinit(self.allocator);
        }
        {
            var it = self.dir.iterate();
            while (it.next(self.io) catch null) |entry| {
                if (entry.kind != .file) continue;
                if (!std.mem.startsWith(u8, entry.name, "evts-")) continue;
                const seq = std.fmt.parseInt(u64, entry.name["evts-".len..], 10) catch continue;
                if (seq >= floor) continue;
                if (self.current_file_path) |cur| {
                    if (std.mem.eql(u8, entry.name, cur)) continue;
                }
                try doomed.append(self.allocator, try self.allocator.dupe(u8, entry.name));
            }
        }

        var removed: usize = 0;
        for (doomed.items) |path| {
            self.dir.deleteFile(self.io, path) catch |err| {
                log.warn("gc: failed to delete orphan {s}: {s}", .{ path, @errorName(err) });
                continue;
            };
            removed += 1;
        }
        if (removed > 0) {
            log.info("gc: removed {d} orphaned log files (unreferenced, below seq {d})", .{ removed, floor });
        }
    }

    /// delete oldest event files until total directory size is under max_dir_bytes
    fn gcBySize(self: *DiskPersist) !void {
        // query all refs ordered by creation time (oldest first)
        var refs: std.ArrayListUnmanaged(GcSizeRef) = .empty;
        defer {
            for (refs.items) |e| self.allocator.free(e.path);
            refs.deinit(self.allocator);
        }

        var total_bytes: u64 = 0;
        {
            var result = try self.db.query(
                "SELECT id, path FROM log_file_refs ORDER BY id ASC",
                .{},
            );
            defer result.deinit();
            while (result.nextUnsafe() catch null) |r| {
                const path = r.get([]const u8, 1);
                const stat = self.dir.statFile(self.io, path, .{}) catch continue;
                total_bytes += stat.size;
                try refs.append(self.allocator, .{
                    .id = r.get(i64, 0),
                    .path = try self.allocator.dupe(u8, path),
                    .size = stat.size,
                });
            }
        }

        if (total_bytes <= self.max_dir_bytes) return;

        log.info("gc: event dir size {d} MB exceeds limit {d} MB, pruning oldest files", .{
            total_bytes / (1024 * 1024),
            self.max_dir_bytes / (1024 * 1024),
        });

        var removed: u64 = 0;
        for (refs.items) |ref| {
            if (total_bytes <= self.max_dir_bytes) break;

            // never delete the current file
            if (self.current_file_path) |cur| {
                if (std.mem.eql(u8, ref.path, cur)) continue;
            }

            _ = self.db.exec("DELETE FROM log_file_refs WHERE id = $1", .{ref.id}) catch |err| {
                log.warn("gc: failed to delete db record {d}: {s}", .{ ref.id, @errorName(err) });
                continue;
            };

            self.dir.deleteFile(self.io, ref.path) catch |err| {
                log.warn("gc: failed to delete {s}: {s}", .{ ref.path, @errorName(err) });
                continue; // don't subtract size if file delete failed
            };

            total_bytes -= @min(total_bytes, ref.size);
            removed += 1;
        }

        if (removed > 0) {
            log.info("gc: size-based pruning removed {d} files, dir now {d} MB", .{
                removed,
                total_bytes / (1024 * 1024),
            });
        }
    }

    /// take down all events for a user (set flag + zero payload)
    pub fn takeDownUser(self: *DiskPersist, uid: u64) !void {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);

        // iterate all log files
        var refs: std.ArrayListUnmanaged([]const u8) = .empty;
        defer {
            for (refs.items) |p| self.allocator.free(p);
            refs.deinit(self.allocator);
        }

        {
            var result = try self.db.query("SELECT path FROM log_file_refs ORDER BY seq_start DESC", .{});
            defer result.deinit();
            while (result.nextUnsafe() catch null) |r| {
                try refs.append(self.allocator, try self.allocator.dupe(u8, r.get([]const u8, 0)));
            }
        }

        for (refs.items) |path| {
            var file = self.dir.openFile(self.io, path, .{ .mode = .read_write }) catch continue;
            defer file.close(self.io);
            mutateUserEventsInFile(file, self.io, uid) catch |err| {
                log.warn("takedown: failed to process {s}: {s}", .{ path, @errorName(err) });
            };
        }
    }

    // --- internals ---

    fn resumeLog(self: *DiskPersist) !void {
        // find most recent log file
        if (try self.db.rowUnsafe(
            "SELECT id, path, seq_start FROM log_file_refs ORDER BY seq_start DESC LIMIT 1",
            .{},
        )) |row| {
            var r = row;
            defer r.deinit() catch {};
            const path = r.get([]const u8, 1);
            const seq_start: u64 = @intCast(r.get(i64, 2));

            var file = self.dir.openFile(self.io, path, .{ .mode = .read_write }) catch {
                // file missing, start fresh
                try self.initLogFile();
                return;
            };

            // scan for last seq (validates records, reports the last good offset)
            const scan = scanForLastSeq(file, self.io, seq_start) catch {
                file.close(self.io);
                try self.initLogFile();
                return;
            };

            // heal a torn tail (partial write from an unclean kill) so subsequent
            // appends stay record-aligned and future scans don't desync.
            const full_size = (try file.stat(self.io)).size;
            if (scan.valid_end < full_size) {
                log.warn("recovered log {s} has a torn tail: truncating {d} -> {d} bytes", .{ path, full_size, scan.valid_end });
                file.setLength(self.io, scan.valid_end) catch |err| {
                    log.warn("failed to truncate torn log tail: {s}", .{@errorName(err)});
                };
            }

            if (scan.last_seq) |ls| {
                self.cur_seq = ls + 1;
                log.info("recovered seq from disk: last_seq={d}", .{ls});
            } else {
                self.cur_seq = if (seq_start > 0) seq_start else 1;
            }

            // append at the end of the last valid record (post-truncation)
            self.current_file_pos = scan.valid_end;

            self.current_file = file;
            self.current_file_path = try self.allocator.dupe(u8, path);
        } else {
            try self.initLogFile();
        }
    }

    fn initLogFile(self: *DiskPersist) !void {
        try self.createLogFile(self.cur_seq);
    }

    fn createLogFile(self: *DiskPersist, start_seq: u64) !void {
        var name_buf: [64]u8 = undefined;
        const name = std.fmt.bufPrint(&name_buf, "evts-{d}", .{start_seq}) catch unreachable;

        const path = try self.allocator.dupe(u8, name);
        errdefer self.allocator.free(path);
        const file = try self.dir.createFile(self.io, name, .{ .exclusive = true, .read = true });
        errdefer {
            file.close(self.io);
            self.dir.deleteFile(self.io, name) catch {};
        }

        // register in Postgres
        _ = try self.db.exec(
            "INSERT INTO log_file_refs (path, seq_start) VALUES ($1, $2)",
            .{ name, @as(i64, @intCast(start_seq)) },
        );

        if (self.current_file) |f| f.close(self.io);
        if (self.current_file_path) |p| self.allocator.free(p);
        self.current_file = file;
        self.current_file_pos = 0;
        self.current_file_path = path;
        self.event_counter = 0;
    }

    pub fn flush(self: *DiskPersist) !void {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);
        try self.flushLocked();
    }

    fn flushLocked(self: *DiskPersist) !void {
        if (self.evtbuf.items.len == 0) return;

        // write buffered bytes to current file
        const file = self.current_file orelse return;
        try file.writePositionalAll(self.io, self.outbuf.items, self.current_file_pos);
        self.current_file_pos += self.outbuf.items.len;

        // clear write buffer (bytes are now durable)
        self.outbuf.clearRetainingCapacity();

        // these events are now durable — publish the watermark so the broadcaster
        // fiber may emit them. evtbuf is append-ordered, so the last entry holds
        // the highest seq just written. this is the only place committed_seq moves.
        if (self.evtbuf.items.len > 0) {
            self.committed_seq.store(self.evtbuf.items[self.evtbuf.items.len - 1].seq, .release);
        }

        // free job data
        for (self.evtbuf.items) |job| {
            self.allocator.free(job.data);
        }
        self.evtbuf.clearRetainingCapacity();

        // check if we need to rotate
        if (self.event_counter >= self.events_per_file) {
            self.createLogFile(self.cur_seq) catch |err| {
                log.err("flush: log rotation failed: {s}", .{@errorName(err)});
            };
        }
    }

    fn flushLoop(self: *DiskPersist) void {
        while (self.alive.load(.acquire)) {
            // sleep for flush interval (replaces timedWait — Io.Condition has no timed variant)
            self.io.sleep(Io.Duration.fromMilliseconds(@intCast(default_flush_interval_ms)), .awake) catch {};
            self.mutex.lockUncancelable(self.io);
            self.flushLocked() catch |err| log.err("event flush failed: {s}", .{@errorName(err)});
            self.mutex.unlock(self.io);
        }
    }
};

// --- playback types ---

pub const PlaybackEntry = struct {
    seq: u64,
    kind: u32,
    uid: u64,
    data: []const u8, // payload only (owned by caller)
};

const LogFileRef = struct {
    path: []const u8,
    seq_start: u64,
};

const GcRef = struct {
    id: i64,
    path: []const u8,
};

const GcSizeRef = struct {
    id: i64,
    path: []const u8,
    size: u64,
};

// --- file-level operations ---

fn readEventsFrom(allocator: Allocator, file: Io.File, io: Io, since: u64, result: *std.ArrayListUnmanaged(PlaybackEntry), limits: PlaybackLimits, bytes_used: *usize) !void {
    const file_size = (try file.stat(io)).size;

    // if since > 0, scan to the right position
    var pos: u64 = 0;
    if (since > 0) {
        pos = seekToSeq(file, io, since, file_size) catch return;
    }

    // read events
    while (pos + header_size <= file_size) {
        if (result.items.len >= limits.max_entries or bytes_used.* >= limits.max_bytes) return;
        var hdr_buf: [header_size]u8 = undefined;
        const n = file.readPositionalAll(io, &hdr_buf, pos) catch break;
        if (n < header_size) break;
        pos += header_size;

        const hdr = EvtHeader.decode(&hdr_buf);

        // skip taken down / rebased events
        if (hdr.flags & (EvtFlags.takedown | EvtFlags.rebased) != 0) {
            pos += hdr.len;
            continue;
        }

        if (hdr.seq <= since) {
            pos += hdr.len;
            continue;
        }

        // read payload
        const data = allocator.alloc(u8, hdr.len) catch break;
        const read_n = file.readPositionalAll(io, data, pos) catch {
            allocator.free(data);
            break;
        };
        if (read_n < hdr.len) {
            allocator.free(data);
            break;
        }
        pos += hdr.len;

        result.append(allocator, .{
            .seq = hdr.seq,
            .kind = hdr.kind,
            .uid = hdr.uid,
            .data = data,
        }) catch {
            allocator.free(data);
            break;
        };
        bytes_used.* += data.len;
    }
}

/// scan file headers to find position of first event with seq > target
fn seekToSeq(file: Io.File, io: Io, target: u64, file_size: u64) !u64 {
    var pos: u64 = 0;
    while (pos + header_size <= file_size) {
        var hdr_buf: [header_size]u8 = undefined;
        const n = try file.readPositionalAll(io, &hdr_buf, pos);
        if (n < header_size) break;

        const hdr = EvtHeader.decode(&hdr_buf);
        if (hdr.seq > target) {
            return pos;
        }
        pos += header_size + hdr.len;
    }
    return pos;
}

const ScanResult = struct {
    /// highest seq among fully-valid records, or null if the file has none
    last_seq: ?u64,
    /// byte offset just past the last fully-valid record. anything beyond is a
    /// torn/corrupt tail (a partial final write after an unclean kill) and must
    /// be truncated before appending, or it desyncs every future scan.
    valid_end: u64,
};

/// scan a log file for the last valid sequence number. each record is validated
/// (known kind, plausible length, body fully present, AND a seq that matches the
/// dense monotonic space the writer guarantees) before being accepted; the scan
/// stops at the first record it can't trust rather than following a garbage
/// length and silently mis-recovering. this is what bounds a torn tail to the
/// torn record itself — see docs/incident-2026-05-31-seq-rewind.md.
///
/// `seq_start` is the file's declared starting seq (from its name / log_file_refs
/// row). the writer assigns seqs densely (seq = cur_seq; cur_seq += 1) and a file
/// begins at exactly seq_start, so the first record must have seq == seq_start and
/// every subsequent record seq == prev + 1. a record whose seq breaks that chain
/// is a torn/garbage tail even if its kind+len happen to look plausible — this is
/// what stops `scanForLastSeq` from accepting CBOR payload bytes as a seq and
/// relocating the whole sequencer into a junk space (the 2026-06 quintillion-seq
/// incident). pass 0 to skip the seq-chain check (genuinely unknown start).
fn scanForLastSeq(file: Io.File, io: Io, seq_start: u64) !ScanResult {
    const file_size = (try file.stat(io)).size;

    var last_seq: ?u64 = null;
    var pos: u64 = 0;
    while (pos + header_size <= file_size) {
        var hdr_buf: [header_size]u8 = undefined;
        const n = try file.readPositionalAll(io, &hdr_buf, pos);
        if (n < header_size) break;

        const hdr = EvtHeader.decode(&hdr_buf);

        // a record we can't trust — unknown kind, implausible length, or a body
        // that runs past EOF (partial final write) — marks the torn tail. stop;
        // `pos` is the end of the last good record.
        const valid_kind = hdr.kind >= @intFromEnum(EvtKind.commit) and hdr.kind <= @intFromEnum(EvtKind.sync);
        if (!valid_kind or hdr.len > max_record_len or pos + header_size + hdr.len > file_size) break;

        // seq must follow the dense monotonic chain: the first record == seq_start,
        // each next == prev + 1. a break here is a torn/garbage tail.
        if (seq_start > 0) {
            const expected = if (last_seq) |ls| ls + 1 else seq_start;
            if (hdr.seq != expected) break;
        }

        last_seq = hdr.seq;
        pos += header_size + hdr.len;
    }
    return .{ .last_seq = last_seq, .valid_end = pos };
}

/// set takedown flag + zero payload for all events belonging to uid
fn mutateUserEventsInFile(file: Io.File, io: Io, uid: u64) !void {
    const file_size = (try file.stat(io)).size;
    var pos: u64 = 0;

    while (pos + header_size <= file_size) {
        var hdr_buf: [header_size]u8 = undefined;
        const n = try file.readPositionalAll(io, &hdr_buf, pos);
        if (n < header_size) break;

        const hdr = EvtHeader.decode(&hdr_buf);

        if (hdr.uid == uid and (hdr.flags & EvtFlags.takedown) == 0) {
            // set takedown flag
            const new_flags = hdr.flags | EvtFlags.takedown;
            var flags_buf: [4]u8 = undefined;
            std.mem.writeInt(u32, &flags_buf, new_flags, .little);
            try file.writePositionalAll(io, &flags_buf, pos);

            // zero the payload
            const payload_start = pos + header_size;
            var zeros: [4096]u8 = [_]u8{0} ** 4096;
            var remaining: u64 = hdr.len;
            var write_pos: u64 = payload_start;
            while (remaining > 0) {
                const chunk: usize = @min(remaining, zeros.len);
                try file.writePositionalAll(io, zeros[0..chunk], write_pos);
                write_pos += chunk;
                remaining -= chunk;
            }
        }

        pos += header_size + hdr.len;
    }
}

// === tests ===

test "header encode/decode roundtrip" {
    const hdr = EvtHeader{
        .flags = 0,
        .kind = @intFromEnum(EvtKind.commit),
        .len = 1234,
        .uid = 42,
        .seq = 99999,
    };
    var buf: [header_size]u8 = undefined;
    hdr.encode(&buf);
    const decoded = EvtHeader.decode(&buf);
    try std.testing.expectEqual(hdr.flags, decoded.flags);
    try std.testing.expectEqual(hdr.kind, decoded.kind);
    try std.testing.expectEqual(hdr.len, decoded.len);
    try std.testing.expectEqual(hdr.uid, decoded.uid);
    try std.testing.expectEqual(hdr.seq, decoded.seq);
}

test "header is little-endian" {
    const hdr = EvtHeader{ .flags = 0, .kind = 1, .len = 256, .uid = 0, .seq = 1 };
    var buf: [header_size]u8 = undefined;
    hdr.encode(&buf);
    // len=256 in LE is 0x00 0x01 0x00 0x00
    try std.testing.expectEqual(@as(u8, 0x00), buf[8]);
    try std.testing.expectEqual(@as(u8, 0x01), buf[9]);
}

test "scanForLastSeq stops at a torn tail" {
    // regression for the seq-rewind incident (docs/incident-2026-05-31-seq-rewind.md):
    // an interrupted flush leaves a partial final record. recovery must stop at the
    // last fully-written record, not trust the torn record's length and desync.
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var file = try tmp.dir.createFile(io, "log", .{ .truncate = true, .read = true });
    defer file.close(io);

    var pos: u64 = 0;
    const writeRecord = struct {
        fn go(f: Io.File, tio: Io, at: *u64, seq: u64, payload: []const u8) !void {
            var hbuf: [header_size]u8 = undefined;
            (EvtHeader{ .flags = 0, .kind = @intFromEnum(EvtKind.commit), .len = @intCast(payload.len), .uid = 0, .seq = seq }).encode(&hbuf);
            try f.writePositionalAll(tio, &hbuf, at.*);
            at.* += header_size;
            try f.writePositionalAll(tio, payload, at.*);
            at.* += payload.len;
        }
    }.go;

    // two complete records
    try writeRecord(file, io, &pos, 1, "aa");
    try writeRecord(file, io, &pos, 2, "bbbb");
    const valid_end = pos;

    // torn record: header claims a payload of 999 bytes, only 2 are present
    var hbuf: [header_size]u8 = undefined;
    (EvtHeader{ .flags = 0, .kind = @intFromEnum(EvtKind.commit), .len = 999, .uid = 0, .seq = 3 }).encode(&hbuf);
    try file.writePositionalAll(io, &hbuf, pos);
    try file.writePositionalAll(io, "xy", pos + header_size);

    const scan = try scanForLastSeq(file, io, 1);
    try std.testing.expectEqual(@as(?u64, 2), scan.last_seq); // torn record 3 rejected
    try std.testing.expectEqual(valid_end, scan.valid_end); // stops at end of record 2
}

test "scanForLastSeq rejects a record with a garbage seq" {
    // regression for the 2026-06 quintillion-seq incident: a torn tail left bytes
    // that decode to a structurally-valid header (good kind, plausible len, body
    // present) but whose seq field is actually CBOR payload (0x6363a382...). the
    // old scan trusted that seq and relocated the whole sequencer; the seq-chain
    // check must reject it so recovery resumes in the real space.
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    var file = try tmp.dir.createFile(io, "log", .{ .truncate = true, .read = true });
    defer file.close(io);

    var pos: u64 = 0;
    const writeRecord = struct {
        fn go(f: Io.File, tio: Io, at: *u64, seq: u64, payload: []const u8) !void {
            var hbuf: [header_size]u8 = undefined;
            (EvtHeader{ .flags = 0, .kind = @intFromEnum(EvtKind.commit), .len = @intCast(payload.len), .uid = 0, .seq = seq }).encode(&hbuf);
            try f.writePositionalAll(tio, &hbuf, at.*);
            at.* += header_size;
            try f.writePositionalAll(tio, payload, at.*);
            at.* += payload.len;
        }
    }.go;

    // file declares seq_start = 100; two good records continue the dense chain
    try writeRecord(file, io, &pos, 100, "aa");
    try writeRecord(file, io, &pos, 101, "bbbb");
    const valid_end = pos;

    // a record that is structurally valid (good kind, small len, body present)
    // but whose seq is the garbage CBOR value — breaks the chain (expected 102)
    try writeRecord(file, io, &pos, 0x6363a38200000000, "zz");

    const scan = try scanForLastSeq(file, io, 100);
    try std.testing.expectEqual(@as(?u64, 101), scan.last_seq); // garbage-seq record rejected
    try std.testing.expectEqual(valid_end, scan.valid_end); // stops at end of record 101

    // and a file whose very first record is already garbage recovers nothing —
    // resumeLog then falls back to seq_start rather than the junk value
    var f2 = try tmp.dir.createFile(io, "log2", .{ .truncate = true, .read = true });
    defer f2.close(io);
    var p2: u64 = 0;
    try writeRecord(f2, io, &p2, 0x6363a38200000000, "aa");
    const scan2 = try scanForLastSeq(f2, io, 100);
    try std.testing.expectEqual(@as(?u64, null), scan2.last_seq);
    try std.testing.expectEqual(@as(u64, 0), scan2.valid_end);
}

fn requireDatabaseUrl() ![]const u8 {
    return getenv("DATABASE_URL") orelse return error.SkipZigTest;
}

/// per-test database isolation. the test binary's runner executes tests
/// concurrently against one server, and the relay tables (log_file_refs /
/// account / host) are global — so sharing one database lets tests see each
/// other's rows (playback(0) returns another test's events). each test gets its
/// own freshly-created database instead, dropped on teardown. unique across
/// processes (pid) and within one (counter).
var test_db_counter: std.atomic.Value(u32) = .{ .raw = 0 };

const TestDb = struct {
    base_url: []const u8,
    name: []u8,
    url: []u8,

    fn create(base_url: []const u8) !TestDb {
        const pid: u64 = @intCast(std.c.getpid());
        const n = test_db_counter.fetchAdd(1, .monotonic);
        const name = try std.fmt.allocPrint(std.testing.allocator, "zlay_test_{d}_{d}", .{ pid, n });
        errdefer std.testing.allocator.free(name);

        {
            const uri = std.Uri.parse(base_url) catch return error.InvalidDatabaseUrl;
            const pool = try pg.Pool.initUri(std.testing.io, std.testing.allocator, uri, .{ .size = 1 });
            defer pool.deinit();
            var buf: [128]u8 = undefined;
            // name is generated from pid/counter — no injection risk
            const stmt = try std.fmt.bufPrint(&buf, "CREATE DATABASE \"{s}\"", .{name});
            _ = try pool.exec(stmt, .{});
        }

        const slash = std.mem.lastIndexOfScalar(u8, base_url, '/') orelse return error.InvalidDatabaseUrl;
        const url = try std.fmt.allocPrint(std.testing.allocator, "{s}/{s}", .{ base_url[0..slash], name });
        return .{ .base_url = base_url, .name = name, .url = url };
    }

    fn destroy(self: *TestDb) void {
        const uri = std.Uri.parse(self.base_url) catch return;
        const pool = pg.Pool.initUri(std.testing.io, std.testing.allocator, uri, .{ .size = 1 }) catch return;
        defer pool.deinit();
        var buf: [128]u8 = undefined;
        const stmt = std.fmt.bufPrint(&buf, "DROP DATABASE IF EXISTS \"{s}\" WITH (FORCE)", .{self.name}) catch return;
        _ = pool.exec(stmt, .{}) catch {};
        std.testing.allocator.free(self.name);
        std.testing.allocator.free(self.url);
    }
};

test "persist and playback" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
    defer dp.deinit();

    // persist some events (sync flush, no background thread)
    const seq1 = try dp.persist(.commit, 1, "payload-one");
    const seq2 = try dp.persist(.identity, 2, "payload-two");
    const seq3 = try dp.persist(.commit, 1, "payload-three");

    try std.testing.expectEqual(@as(u64, 1), seq1);
    try std.testing.expectEqual(@as(u64, 2), seq2);
    try std.testing.expectEqual(@as(u64, 3), seq3);

    // flush manually
    {
        dp.mutex.lockUncancelable(dp.io);
        defer dp.mutex.unlock(dp.io);
        try dp.flushLocked();
    }

    // playback from cursor=0 → all events
    var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty;
    defer {
        for (entries.items) |e| std.testing.allocator.free(e.data);
        entries.deinit(std.testing.allocator);
    }
    try dp.playback(0, std.testing.allocator, &entries, .{});

    try std.testing.expectEqual(@as(usize, 3), entries.items.len);
    try std.testing.expectEqualStrings("payload-one", entries.items[0].data);
    try std.testing.expectEqual(@as(u64, 1), entries.items[0].seq);
    try std.testing.expectEqualStrings("payload-two", entries.items[1].data);
    try std.testing.expectEqualStrings("payload-three", entries.items[2].data);
}

test "worker failure preserves upstream cursor and successful retry reaches disk" {
    const cbor = @import("zat").cbor;
    const broadcaster = @import("broadcaster.zig");
    const validator_mod = @import("validator.zig");
    const worker = @import("frame_worker.zig");
    const ingest_ack = @import("ingest_ack.zig");
    var tdb = try TestDb.create(try requireDatabaseUrl());
    defer tdb.destroy();
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);
    var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 2, std.testing.io);
    defer dp.deinit();
    var shutdown = std.atomic.Value(bool){ .raw = false };
    var bc = broadcaster.Broadcaster.init(std.testing.allocator, std.testing.io, &shutdown);
    defer bc.deinit();
    var validator = validator_mod.Validator.init(std.testing.allocator, &bc.stats, std.testing.io);
    defer validator.deinit();
    var progress: ingest_ack.Progress = .{};
    var cursor: ?u64 = 10;
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const header = try cbor.encodeAlloc(arena.allocator(), .{ .map = &.{
        .{ .key = "op", .value = .{ .unsigned = 1 } },
        .{ .key = "t", .value = .{ .text = "#identity" } },
    } });
    const payload = try cbor.encodeAlloc(arena.allocator(), .{ .map = &.{
        .{ .key = "did", .value = .{ .text = "did:plc:cursor-test" } },
        .{ .key = "seq", .value = .{ .unsigned = 11 } },
    } });

    for (0..2) |attempt| {
        var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = 0 });
        dp.allocator = if (attempt == 0) failing.allocator() else std.testing.allocator;
        defer dp.allocator = std.testing.allocator;
        var work: worker.FrameWork = .{
            .data = try std.mem.concat(std.testing.allocator, u8, &.{ header, payload }),
            .hostname = try std.testing.allocator.dupe(u8, "example.com"),
            .host_id = 1,
            .allocator = std.testing.allocator,
            .io = std.testing.io,
            .bc = &bc,
            .validator = &validator,
            .persist = &dp,
            .collection_index = null,
            .resyncer = null,
            .receipt = try progress.reserve(std.testing.io, 11, &cursor),
        };
        bc.stats.pool_queued_bytes.store(work.data.len, .monotonic);
        worker.processFrame(&work);
        if (attempt == 0) {
            try std.testing.expectError(error.FrameProcessingFailed, progress.drain(&cursor, true));
            try std.testing.expectEqual(@as(?u64, 10), cursor);
            try std.testing.expectEqual(@as(u64, 1), dp.cur_seq);
            progress.reset();
        } else {
            try progress.drain(&cursor, true);
            try std.testing.expectEqual(@as(?u64, 11), cursor);
            try std.testing.expectEqual(@as(u64, 1), dp.committed_seq.load(.acquire));
            const scan = try scanForLastSeq(dp.current_file.?, std.testing.io, 1);
            try std.testing.expectEqual(@as(?u64, 1), scan.last_seq);
        }
    }
}

test "persist allocation failures preserve sequence and buffered records" {
    const Case = struct {
        fn run(allocator: Allocator) !void {
            var tmp = std.testing.tmpDir(.{});
            defer tmp.cleanup();
            const file = try tmp.dir.createFile(std.testing.io, "events", .{ .read = true });
            defer file.close(std.testing.io);
            var dp: DiskPersist = .{
                .allocator = allocator,
                .dir_path = "",
                .dir = tmp.dir,
                .db = undefined,
                .current_file = file,
                .did_cache = undefined,
                .host_id_cache = undefined,
                .io = std.testing.io,
            };
            defer {
                for (dp.evtbuf.items) |job| allocator.free(job.data);
                dp.evtbuf.deinit(allocator);
                dp.outbuf.deinit(allocator);
            }
            const seq = dp.persist(.commit, 1, "payload") catch |err| {
                try std.testing.expectEqual(@as(u64, 1), dp.cur_seq);
                try std.testing.expectEqual(@as(usize, 0), dp.evtbuf.items.len);
                try std.testing.expectEqual(@as(usize, 0), dp.outbuf.items.len);
                return err;
            };
            try std.testing.expectEqual(@as(u64, 1), seq);
            try dp.flushLocked();
            const scan = try scanForLastSeq(file, std.testing.io, 1);
            try std.testing.expectEqual(@as(?u64, 1), scan.last_seq);
        }
    };
    try std.testing.checkAllAllocationFailures(std.testing.allocator, Case.run, .{});
}

test "failed threshold write retains ownership and bounds pending events until retry" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const file = try tmp.dir.createFile(std.testing.io, "events", .{ .read = true });
    defer file.close(std.testing.io);
    const read_only = try tmp.dir.openFile(std.testing.io, "events", .{});
    defer read_only.close(std.testing.io);
    var dp: DiskPersist = .{
        .allocator = std.testing.allocator,
        .dir_path = "",
        .dir = tmp.dir,
        .db = undefined,
        .current_file = read_only,
        .did_cache = undefined,
        .host_id_cache = undefined,
        .io = std.testing.io,
    };
    defer {
        for (dp.evtbuf.items) |job| std.testing.allocator.free(job.data);
        dp.evtbuf.deinit(std.testing.allocator);
        dp.outbuf.deinit(std.testing.allocator);
    }
    for (0..default_flush_threshold) |_| _ = try dp.persist(.commit, 1, "payload");
    try std.testing.expectEqual(@as(u64, 0), dp.committed_seq.load(.acquire));
    try std.testing.expectError(error.NotOpenForWriting, dp.persist(.commit, 1, "next"));
    try std.testing.expectEqual(default_flush_threshold, dp.evtbuf.items.len);
    try std.testing.expectEqual(default_flush_threshold + 1, dp.cur_seq);
    dp.current_file = file;
    try dp.flush();
    const scan = try scanForLastSeq(file, std.testing.io, 1);
    try std.testing.expectEqual(@as(?u64, default_flush_threshold), scan.last_seq);
    try std.testing.expectEqual(@as(usize, 0), dp.evtbuf.items.len);
}

test "failed rotation preserves the active file and retries after database recovery" {
    var tdb = try TestDb.create(try requireDatabaseUrl());
    defer tdb.destroy();
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);
    var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 2, std.testing.io);
    defer dp.deinit();

    _ = try dp.persist(.commit, 1, "before");
    try dp.flushLocked();
    const old_pos = dp.current_file_pos;
    {
        var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = 0 });
        dp.allocator = failing.allocator();
        defer dp.allocator = std.testing.allocator;
        try std.testing.expectError(error.OutOfMemory, dp.createLogFile(2));
    }
    try std.testing.expectEqualStrings("evts-1", dp.current_file_path.?);
    try tmp.dir.createDir(std.testing.io, "evts-2", .default_dir);
    if (dp.createLogFile(2)) |_| return error.ExpectedRotationFailure else |_| {}
    try std.testing.expectEqualStrings("evts-1", dp.current_file_path.?);
    try std.testing.expectEqual(old_pos, dp.current_file_pos);
    try tmp.dir.deleteDir(std.testing.io, "evts-2");

    _ = try dp.db.exec("ALTER TABLE log_file_refs ADD CONSTRAINT reject_rotation CHECK (seq_start < 2)", .{});
    if (dp.createLogFile(2)) |_| return error.ExpectedRotationFailure else |_| {}
    try std.testing.expectEqualStrings("evts-1", dp.current_file_path.?);
    try std.testing.expectError(error.FileNotFound, tmp.dir.openFile(std.testing.io, "evts-2", .{}));
    _ = try dp.persist(.commit, 1, "after");
    try dp.flushLocked();
    const scan = try scanForLastSeq(dp.current_file.?, std.testing.io, 1);
    try std.testing.expectEqual(@as(?u64, 2), scan.last_seq);

    _ = try dp.db.exec("ALTER TABLE log_file_refs DROP CONSTRAINT reject_rotation", .{});
    try dp.createLogFile(3);
    _ = try dp.persist(.commit, 1, "rotated");
    try dp.flushLocked();
    var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty;
    defer {
        for (entries.items) |entry| std.testing.allocator.free(entry.data);
        entries.deinit(std.testing.allocator);
    }
    try dp.playback(0, std.testing.allocator, &entries, .{});
    try std.testing.expectEqual(@as(usize, 3), entries.items.len);
}

test "gc deletes expired segments and their refs, keeps the current file" {
    // written 2026-08-27, when production surfaced 10 weeks of never-pruned
    // segments (132G/150G): gc() had no test at all. this exercises the whole
    // path against real postgres — the $1::interval bind, the row decode
    // (whose `nextUnsafe() catch null` would silently turn a query error into
    // "nothing expired"), the ref-before-file delete order, and the
    // current-file skip.
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 5, std.testing.io);
    defer dp.deinit();
    dp.events_per_file = 2; // force rotation quickly
    dp.retention_hours = 1;

    // 5 events at 2/file, flushing each (rotation happens at flush end once
    // the counter passes events_per_file) → at least 3 segments registered
    for (0..5) |i| {
        var payload_buf: [16]u8 = undefined;
        const payload = std.fmt.bufPrint(&payload_buf, "payload-{d}", .{i}) catch unreachable;
        _ = try dp.persist(.commit, 1, payload);
        dp.mutex.lockUncancelable(dp.io);
        defer dp.mutex.unlock(dp.io);
        try dp.flushLocked();
    }

    var ref_count: i64 = 0;
    {
        var result = try dp.db.query("SELECT count(*) FROM log_file_refs", .{});
        defer result.deinit();
        while (result.nextUnsafe() catch null) |r| ref_count = r.get(i64, 0);
    }
    try std.testing.expect(ref_count >= 3);

    // nothing is expired yet: gc must be a no-op
    try dp.gc();
    var after_noop: i64 = 0;
    {
        var result = try dp.db.query("SELECT count(*) FROM log_file_refs", .{});
        defer result.deinit();
        while (result.nextUnsafe() catch null) |r| after_noop = r.get(i64, 0);
    }
    try std.testing.expectEqual(ref_count, after_noop);

    // age every ref past retention, then gc: everything but the current
    // file (and its ref... the ref of the skipped file survives) must go
    _ = try dp.db.exec("UPDATE log_file_refs SET created_at = now() - interval '2 hours'", .{});
    try dp.gc();

    var survivors: std.ArrayListUnmanaged([]u8) = .empty;
    defer {
        for (survivors.items) |s| std.testing.allocator.free(s);
        survivors.deinit(std.testing.allocator);
    }
    {
        var result = try dp.db.query("SELECT path FROM log_file_refs ORDER BY seq_start", .{});
        defer result.deinit();
        while (result.nextUnsafe() catch null) |r| {
            try survivors.append(std.testing.allocator, try std.testing.allocator.dupe(u8, r.get([]const u8, 0)));
        }
    }
    // only the current file's ref remains
    try std.testing.expectEqual(@as(usize, 1), survivors.items.len);
    try std.testing.expectEqualStrings(dp.current_file_path.?, survivors.items[0]);

    // expired files are gone from disk; the current file remains
    var dir = try std.Io.Dir.cwd().openDir(std.testing.io, dir_path, .{ .iterate = true });
    defer dir.close(std.testing.io);
    var files_on_disk: usize = 0;
    var it = dir.iterate();
    while (try it.next(std.testing.io)) |entry| {
        if (entry.kind != .file) continue;
        if (!std.mem.startsWith(u8, entry.name, "evts-")) continue;
        files_on_disk += 1;
        try std.testing.expectEqualStrings(dp.current_file_path.?, entry.name);
    }
    try std.testing.expectEqual(@as(usize, 1), files_on_disk);

    // orphan sweep: an unreferenced file below the retained window (the
    // production 100G leak: row deleted, file delete failed) is removed;
    // an unreferenced file at/above the retained floor is left alone
    {
        const old_orphan = try dir.createFile(std.testing.io, "evts-1", .{});
        old_orphan.close(std.testing.io);
        const new_unregistered = try dir.createFile(std.testing.io, "evts-999999", .{});
        new_unregistered.close(std.testing.io);
    }
    try dp.gc();
    try std.testing.expectError(error.FileNotFound, dir.openFile(std.testing.io, "evts-1", .{}));
    const kept = try dir.openFile(std.testing.io, "evts-999999", .{});
    kept.close(std.testing.io);
    dir.deleteFile(std.testing.io, "evts-999999") catch {};
}

test "committed_seq advances only on durable flush" {
    // regression for the seq-rewind incident (docs/incident-2026-05-31-seq-rewind.md):
    // the broadcaster fiber only emits a frame once its seq <= committed_seq, and
    // committed_seq advances ONLY here, after the durable write. so a frame can
    // never reach consumers before it is on disk, and the seq recovered on restart
    // can't trail an already-broadcast seq.
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
    defer dp.deinit();

    // fresh log: nothing durable yet
    try std.testing.expectEqual(@as(u64, 0), dp.committed_seq.load(.acquire));

    // persist below the flush threshold — assigned seqs but NOT yet committed,
    // so the broadcaster would hold all three back.
    _ = try dp.persist(.commit, 1, "a");
    _ = try dp.persist(.commit, 2, "b");
    _ = try dp.persist(.commit, 3, "c");
    try std.testing.expectEqual(@as(u64, 0), dp.committed_seq.load(.acquire));

    // flush → committed_seq jumps to the highest durable seq, releasing all three
    {
        dp.mutex.lockUncancelable(dp.io);
        defer dp.mutex.unlock(dp.io);
        try dp.flushLocked();
    }
    try std.testing.expectEqual(@as(u64, 3), dp.committed_seq.load(.acquire));
}

test "playback with cursor" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
    defer dp.deinit();

    _ = try dp.persist(.commit, 1, "a");
    _ = try dp.persist(.commit, 2, "b");
    _ = try dp.persist(.commit, 3, "c");
    {
        dp.mutex.lockUncancelable(dp.io);
        defer dp.mutex.unlock(dp.io);
        try dp.flushLocked();
    }

    // playback from cursor=2 → only seq 3
    var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty;
    defer {
        for (entries.items) |e| std.testing.allocator.free(e.data);
        entries.deinit(std.testing.allocator);
    }
    try dp.playback(2, std.testing.allocator, &entries, .{});

    try std.testing.expectEqual(@as(usize, 1), entries.items.len);
    try std.testing.expectEqual(@as(u64, 3), entries.items[0].seq);
    try std.testing.expectEqualStrings("c", entries.items[0].data);
}

test "playback respects entry and byte limits, resumable by cursor" {
    // regression for the 2026-08-06 consumer-flap OOM: an old cursor must
    // not materialize the whole backlog in one playback call.
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
    defer dp.deinit();

    for (0..10) |_| _ = try dp.persist(.commit, 1, "0123456789");
    {
        dp.mutex.lockUncancelable(dp.io);
        defer dp.mutex.unlock(dp.io);
        try dp.flushLocked();
    }

    // max_entries caps the chunk; resuming from the last seq gets the rest
    var chunk: std.ArrayListUnmanaged(PlaybackEntry) = .empty;
    defer {
        for (chunk.items) |e| std.testing.allocator.free(e.data);
        chunk.deinit(std.testing.allocator);
    }
    try dp.playback(0, std.testing.allocator, &chunk, .{ .max_entries = 4 });
    try std.testing.expectEqual(@as(usize, 4), chunk.items.len);
    try std.testing.expectEqual(@as(u64, 4), chunk.items[3].seq);

    var rest: std.ArrayListUnmanaged(PlaybackEntry) = .empty;
    defer {
        for (rest.items) |e| std.testing.allocator.free(e.data);
        rest.deinit(std.testing.allocator);
    }
    try dp.playback(chunk.items[3].seq, std.testing.allocator, &rest, .{ .max_entries = 100 });
    try std.testing.expectEqual(@as(usize, 6), rest.items.len);
    try std.testing.expectEqual(@as(u64, 5), rest.items[0].seq);
    try std.testing.expectEqual(@as(u64, 10), rest.items[5].seq);

    // max_bytes caps the chunk: 10-byte payloads, 25-byte budget → the
    // check runs before each read, so the 3rd entry crosses the budget
    // and the 4th is never read
    var byte_chunk: std.ArrayListUnmanaged(PlaybackEntry) = .empty;
    defer {
        for (byte_chunk.items) |e| std.testing.allocator.free(e.data);
        byte_chunk.deinit(std.testing.allocator);
    }
    try dp.playback(0, std.testing.allocator, &byte_chunk, .{ .max_bytes = 25 });
    try std.testing.expectEqual(@as(usize, 3), byte_chunk.items.len);
}

test "seq recovery after reinit" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    // write some events
    {
        var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
        defer dp.deinit();
        _ = try dp.persist(.commit, 1, "x");
        _ = try dp.persist(.commit, 2, "y");
        _ = try dp.persist(.account, 3, "z");
        dp.mutex.lockUncancelable(dp.io);
        defer dp.mutex.unlock(dp.io);
        try dp.flushLocked();
    }

    // reinit — should recover seq
    {
        var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
        defer dp.deinit();
        try std.testing.expectEqual(@as(u64, 3), dp.lastSeq().?);
        const seq4 = try dp.persist(.commit, 1, "w");
        try std.testing.expectEqual(@as(u64, 4), seq4);
    }
}

test "takedown zeros payload" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
    defer dp.deinit();

    _ = try dp.persist(.commit, 42, "secret-data");
    _ = try dp.persist(.commit, 99, "other-data");
    {
        dp.mutex.lockUncancelable(dp.io);
        defer dp.mutex.unlock(dp.io);
        try dp.flushLocked();
    }

    // take down user 42
    try dp.takeDownUser(42);

    // playback should skip user 42's events
    var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty;
    defer {
        for (entries.items) |e| std.testing.allocator.free(e.data);
        entries.deinit(std.testing.allocator);
    }
    try dp.playback(0, std.testing.allocator, &entries, .{});

    try std.testing.expectEqual(@as(usize, 1), entries.items.len);
    try std.testing.expectEqualStrings("other-data", entries.items[0].data);
}

test "uidForDid assigns and caches UIDs" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
    defer dp.deinit();

    // first call creates the account
    const uid1 = try dp.uidForDid("did:plc:alice");
    try std.testing.expect(uid1 > 0);

    // same DID returns same UID (from cache)
    const uid1_again = try dp.uidForDid("did:plc:alice");
    try std.testing.expectEqual(uid1, uid1_again);

    // different DID gets a different UID
    const uid2 = try dp.uidForDid("did:plc:bob");
    try std.testing.expect(uid2 > 0);
    try std.testing.expect(uid1 != uid2);
}

test "accountActivity distinguishes local takedown from stale upstream status" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 5, std.testing.io);
    defer dp.deinit();

    const uid = try dp.uidForDid("did:plc:migrant");

    // fresh account: fully active
    var activity = try dp.accountActivity(uid);
    try std.testing.expect(activity.local_ok and activity.upstream_ok);
    try std.testing.expect(activity.active());

    // the migration-stale shape: locally active, upstream deactivated.
    // this is the only combination that may trigger a StatusChecker re-check.
    try dp.updateAccountUpstreamStatus(uid, "deactivated");
    activity = try dp.accountActivity(uid);
    try std.testing.expect(activity.local_ok);
    try std.testing.expect(!activity.upstream_ok);
    try std.testing.expect(!activity.active());
    try std.testing.expect(!try dp.isAccountActive(uid));

    // recovery: the checker writes the PDS's answer back
    try dp.updateAccountUpstreamStatus(uid, "active");
    activity = try dp.accountActivity(uid);
    try std.testing.expect(activity.active());
    try std.testing.expect(try dp.isAccountActive(uid));

    // local takedown: local_ok false — must never look like the re-check shape
    _ = try dp.db.exec("UPDATE account SET status = 'takendown' WHERE uid = $1", .{@as(i64, @intCast(uid))});
    activity = try dp.accountActivity(uid);
    try std.testing.expect(!activity.local_ok);
    try std.testing.expect(activity.upstream_ok);
    try std.testing.expect(!activity.active());

    // unknown uid: inactive on both axes
    activity = try dp.accountActivity(999_999);
    try std.testing.expect(!activity.local_ok and !activity.upstream_ok);
}

test "getAccountState retains rev-only legacy state" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 5, std.testing.io);
    defer dp.deinit();

    const uid = try dp.uidForDid("did:plc:revonly");
    try std.testing.expect(try dp.updateAccountState(uid, "3k2abcdefghij", ""));

    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const state = (try dp.getAccountState(uid, arena.allocator())).?;
    try std.testing.expectEqualStrings("3k2abcdefghij", state.rev);
    try std.testing.expectEqualStrings("", state.data_cid);
}

test "host_id cache stays coherent with setAccountHostId" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
    defer dp.deinit();

    const uid = try dp.uidForDid("did:plc:hostcache");

    // unset host_id reads as 0 and is not cached (transient pre-assignment state)
    try std.testing.expectEqual(@as(u64, 0), try dp.getAccountHostId(uid));

    // write-through: set populates the cache, read returns it
    try dp.setAccountHostId(uid, 42);
    try std.testing.expectEqual(@as(u64, 42), try dp.getAccountHostId(uid));

    // a host change updates the cache in place — no stale read
    try dp.setAccountHostId(uid, 99);
    try std.testing.expectEqual(@as(u64, 99), try dp.getAccountHostId(uid));

    // clearing to 0 evicts the entry; read falls through to the DB (0)
    try dp.setAccountHostId(uid, 0);
    try std.testing.expectEqual(@as(u64, 0), try dp.getAccountHostId(uid));

    // value survives a cache miss: clear the cache, read repopulates from DB
    try dp.setAccountHostId(uid, 7);
    _ = dp.host_id_cache.remove(&DiskPersist.uidKey(uid));
    try std.testing.expectEqual(@as(u64, 7), try dp.getAccountHostId(uid));
}

test "uidForDid survives reinit" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var uid1: u64 = undefined;
    {
        var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
        defer dp.deinit();
        uid1 = try dp.uidForDid("did:plc:carol");
    }

    // reinit — UID should be the same from database
    {
        var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
        defer dp.deinit();
        const uid1_again = try dp.uidForDid("did:plc:carol");
        try std.testing.expectEqual(uid1, uid1_again);
    }
}

test "takedown with real UIDs" {
    const base_url = try requireDatabaseUrl();
    var tdb = try TestDb.create(base_url);
    defer tdb.destroy();
    const database_url = tdb.url;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
    defer std.testing.allocator.free(dir_path);

    var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io);
    defer dp.deinit();

    const alice_uid = try dp.uidForDid("did:plc:alice");
    const bob_uid = try dp.uidForDid("did:plc:bob");

    _ = try dp.persist(.commit, alice_uid, "alice-post");
    _ = try dp.persist(.commit, bob_uid, "bob-post");
    _ = try dp.persist(.commit, alice_uid, "alice-post-2");
    {
        dp.mutex.lockUncancelable(dp.io);
        defer dp.mutex.unlock(dp.io);
        try dp.flushLocked();
    }

    // take down alice
    try dp.takeDownUser(alice_uid);

    // playback should only have bob's event
    var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty;
    defer {
        for (entries.items) |e| std.testing.allocator.free(e.data);
        entries.deinit(std.testing.allocator);
    }
    try dp.playback(0, std.testing.allocator, &entries, .{});

    try std.testing.expectEqual(@as(usize, 1), entries.items.len);
    try std.testing.expectEqualStrings("bob-post", entries.items[0].data);
    try std.testing.expectEqual(bob_uid, entries.items[0].uid);
}

fn tmpDirRealPath(allocator: Allocator, tmp: std.testing.TmpDir) ![]const u8 {
    var buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const n = try tmp.dir.realPath(std.testing.io, &buf);
    return try allocator.dupe(u8, buf[0..n]);
}