//! The relay's firehose event handler: buffers actor events, flushes batches to worker.
//!
//! Consumes com.atproto.sync.subscribeRepos directly rather than jetstream.
//! jetstream is a lossy projection of the relay: pds.zat.dev repos were absent
//! from a 7-day jetstream replay (0 events, 6/6 repos) while the relay was
//! rev-current with that PDS and re-emitted its commits live (verified
//! 2026-07-28). The loss is silent and partial, so it cannot be spot-checked.

const std = @import("std");
const io = std.Options.debug_io;
const mem = std.mem;
const json = std.json;
const Allocator = mem.Allocator;
const zat = @import("zat");
const HttpTransport = zat.HttpTransport;
const BloomFilter = @import("bloom.zig").BloomFilter;
const identity = @import("identity.zig");
const logfire = @import("logfire");
const fatal = @import("fatal.zig");

const log = std.log.scoped(.ingester);

fn cGetenv(name: [*:0]const u8) ?[]const u8 {
    if (std.c.getenv(name)) |p| return mem.span(p);
    return null;
}

pub fn timestamp() i64 {
    return @intCast(@divFloor(std.Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s));
}

// per-flush event cap. the flush POST is SYNCHRONOUS — it blocks the jetstream
// read loop for one worker round-trip (~1-2s), so catch-up throughput is
// MAX_BATCH / flush_time. at 300 the consumer capped at ~0.8x of peak firehose
// even on a fast host and the gap ran to ~3h (2026-06-15); 1500 lifts the
// ceiling well above realtime so the backlog drains. the worker accepts
// 10k/POST as one Turso transaction; catch-up batches are mostly bare-DID
// INSERT OR IGNORE (light locks). PROPER fix is decoupling the read loop from
// the flusher so POSTs don't stall reads — this is the stopgap.
// 1500 was tuned for jetstream's server-FILTERED stream. The relay's firehose
// delivers ~4x the actor events, and each POST now fans into several sequential
// Turso transactions worker-side (250 statements each, to stay inside Turso's
// 5s transaction window). At 1500 the whole request outran Cloudflare's origin
// timeout, 524'd, and the ingester retried the same oversized batch every 5s —
// a self-sustaining wedge that kept Turso's single writer busy even after the
// job that triggered it was killed (2026-07-28: writes down ~25min, cursor
// frozen, watchdog restarts that could not clear it).
//
// 400 keeps one POST comfortably inside the timeout. Throughput is fine because
// flushes are far more frequent than the old 5s cadence.
pub const MAX_BATCH: usize = 400;
const MAX_BACKLOG = MAX_BATCH * 10;
const BLOOM_BITS: usize = 10_000_000; // ~1.2MB fixed
const BLOOM_HASHES: usize = 7;

// liveness watchdog. a wedged process looks healthy from the outside while
// ingesting nothing — process-wide TLS rot + a black-holed websocket silently
// dropped 13 days of firehose (2026-05-29 → 06-12). exiting instead lets fly
// restart us; resume-from-cursor makes that lossless within jetstream's
// ~25h replay window.
const WATCHDOG_STALL_SECS: i64 = 10 * 60;
const WATCHDOG_MAX_CONSECUTIVE_FAILURES: u32 = 60; // ≥5min at the 5s retry cadence
const HEARTBEAT_INTERVAL_SECS: i64 = 60;

pub var source_event_at: std.atomic.Value(i64) = .init(0);
pub var queued_ingest: std.atomic.Value(usize) = .init(0);
pub var queued_deletes: std.atomic.Value(usize) = .init(0);
pub var backpressure_count: std.atomic.Value(u64) = .init(0);
pub var backpressure_active: std.atomic.Value(bool) = .init(false);

pub var last_event_at: std.atomic.Value(i64) = .init(0);
pub var last_flush_ok_at: std.atomic.Value(i64) = .init(0);

pub fn watchdogLoop() void {
    const boot = timestamp();
    last_event_at.store(boot, .monotonic);
    last_flush_ok_at.store(boot, .monotonic);
    while (true) {
        io.sleep(std.Io.Duration.fromSeconds(60), .awake) catch {};
        const now = timestamp();
        const event_age = now - last_event_at.load(.monotonic);
        const flush_age = now - last_flush_ok_at.load(.monotonic);
        if (event_age >= WATCHDOG_STALL_SECS or flush_age >= WATCHDOG_STALL_SECS) {
            log.err("watchdog: stalled (last event {d}s ago, last ok flush {d}s ago), exiting for restart", .{
                event_age, flush_age,
            });
            fatal.exitForRestart();
        }
    }
}

pub const Config = struct {
    worker_url: []const u8,
    secret: []const u8,
};

pub fn getConfig() Config {
    return .{
        .worker_url = cGetenv("TYPEAHEAD_URL") orelse
            @panic("TYPEAHEAD_URL not set"),
        .secret = cGetenv("TYPEAHEAD_SECRET") orelse
            @panic("TYPEAHEAD_SECRET not set"),
    };
}

/// The relay's firehose has no server-side collection filter (jetstream's
/// wantedCollections has no equivalent), so the allowlist moves client-side and
/// every frame is decoded before it can be discarded.
/// Which firehose collections cause us to discover an actor.
///
/// TODO: this list is hardcoded, so onboarding an app is a code change and a
/// deploy — and until that happens its users are discoverable ONLY via
/// `#identity`. That is not a theoretical gap: birds.place users write nothing
/// but `place.birds.*`, so before this entry existed the commit path could not
/// see them at all, and an actor whose `#identity` was missed had no second
/// chance. Worth moving to config (or dropping the allowlist for a deny-list of
/// known-spam collections) rather than growing this by hand.
/// Design sketch — published lexicons as the signal: docs/collection-discovery.md
pub fn wantedCollection(collection: []const u8) bool {
    const exact = [_][]const u8{
        "app.bsky.actor.profile",
        "app.bsky.feed.post",
        "app.bsky.feed.like",
        "app.bsky.graph.follow",
        "sh.tangled.actor.profile",
    };
    for (exact) |e| if (mem.eql(u8, collection, e)) return true;

    const prefixes = [_][]const u8{
        "site.standard.",
        "place.birds.",
    };
    for (prefixes) |pre| if (mem.startsWith(u8, collection, pre)) return true;
    return false;
}

/// Render a profile record's avatar blob ref as the base32 CID string the rest
/// of the system stores. DAG-CBOR carries the ref as a real CID (tag 42), where
/// jetstream handed us a pre-rendered `ref.$link` string — so this transform is
/// new with the firehose cutover and is the one that can silently produce wrong
/// avatars rather than failing loudly.
pub fn avatarCidOf(allocator: Allocator, record: zat.cbor.Value) ?[]const u8 {
    return blobCidOf(allocator, record, "avatar");
}

/// Same rendering for any blob-valued profile field (`avatar`, `banner`).
pub fn blobCidOf(allocator: Allocator, record: zat.cbor.Value, key: []const u8) ?[]const u8 {
    const blob = record.get(key) orelse return null;
    const cid = blob.getCid("ref") orelse return null;
    return zat.multibase.encode(allocator, .base32lower, cid.raw) catch null;
}

/// an actor event to POST to the worker
pub const ActorEvent = struct {
    did: []const u8,
    handle: ?[]const u8 = null,
    display_name: ?[]const u8 = null,
    avatar_cid: ?[]const u8 = null,
    description: ?[]const u8 = null,
    banner_cid: ?[]const u8 = null,
    pds: ?[]const u8 = null,
};

pub const IngestHandler = struct {
    allocator: Allocator,
    config: Config,
    transport: *HttpTransport,
    /// nullable so unit tests / minimal modes can run without it.
    /// in production main wires this to a pool with N concurrent slingshot workers.
    identity_pool: ?*identity.Pool,
    buffer: std.ArrayList(ActorEvent),
    /// seq of the firehose event each buffered entry came from, 1:1 with
    /// `buffer`. Without it the flush can only advance the durable cursor when
    /// the buffer fully drains — which never happens at relay volume, so the
    /// cursor stayed 0 and every restart lost its place.
    cursors: std.ArrayList(i64),
    delete_buffer: std.ArrayList([]const u8),
    /// arena owns all string data in buffer/delete_buffer
    arena: std.heap.ArenaAllocator,
    /// bloom filter: fixed-size dedup for non-profile DIDs (~1.2MB)
    bloom: BloomFilter,
    pending_cursor: i64 = 0,
    flushed_cursor: i64 = 0,
    total_ingested: u64 = 0,
    total_deleted: u64 = 0,
    last_flush: i64 = 0,
    retry_after: i64 = 0, // timestamp before which we skip flush attempts
    consecutive_failures: u32 = 0,
    last_heartbeat: i64 = 0,

    pub fn init(
        allocator: Allocator,
        config: Config,
        transport: *HttpTransport,
        identity_pool: ?*identity.Pool,
    ) !IngestHandler {
        return .{
            .allocator = allocator,
            .config = config,
            .transport = transport,
            .identity_pool = identity_pool,
            .buffer = .empty,
            .cursors = .empty,
            .delete_buffer = .empty,
            .arena = std.heap.ArenaAllocator.init(allocator),
            .bloom = try BloomFilter.init(allocator, BLOOM_BITS, BLOOM_HASHES),
            .last_flush = timestamp(),
        };
    }

    pub fn deinit(self: *IngestHandler) void {
        self.bloom.deinit(self.allocator);
        self.arena.deinit();
        self.buffer.deinit(self.allocator);
        self.cursors.deinit(self.allocator);
        self.delete_buffer.deinit(self.allocator);
    }

    /// The ONLY way to buffer an event. Keeps `cursors` 1:1 with `buffer`; a
    /// bare `buffer.append` desynchronizes them and the flush then advances the
    /// durable cursor past events it never sent.
    fn enqueue(self: *IngestHandler, event: ActorEvent) void {
        self.buffer.append(self.allocator, event) catch return;
        self.cursors.append(self.allocator, self.pending_cursor) catch {
            _ = self.buffer.pop();
        };
    }

    fn dupe(self: *IngestHandler, s: []const u8) ?[]const u8 {
        return self.arena.allocator().dupe(u8, s) catch null;
    }

    fn resetBloom(self: *IngestHandler) void {
        log.info("resetting bloom filter ({d} insertions)", .{self.bloom.count});
        self.bloom.reset();
    }

    pub fn onEvent(self: *IngestHandler, event: zat.FirehoseEvent) void {
        // BEFORE dispatch: the handlers append to `buffer`, and each append
        // records `pending_cursor` as that entry's seq. Setting it afterwards
        // would tag every entry with the PREVIOUS event's seq.
        //
        // relay `seq`, NOT jetstream's µs timestamp — different keyspace, which
        // is why the cutover writes a fresh cursor key (see fetchCursor).
        if (event.seq()) |s| self.pending_cursor = s;

        const source_time: ?[]const u8 = switch (event) {
            .commit => |c| c.time,
            .identity => |id| id.time,
            .account => |acct| acct.time,
            .sync => |sync_event| sync_event.time,
            .info => null,
        };
        if (source_time) |t| {
            if (zat.Datetime.parse(t)) |parsed| source_event_at.store(@divFloor(parsed.micros, 1_000_000), .monotonic);
        }

        switch (event) {
            .commit => |c| self.handleCommit(c),
            .identity => |id| self.handleIdentity(id),
            .account => |acct| self.handleAccount(acct),
            .sync, .info => {},
        }

        const now = timestamp();
        last_event_at.store(now, .monotonic);
        self.publishQueueDepth();
        // Block this callback before reading more frames. Keep queued writes and
        // their cursor evidence until they succeed; the watchdog can restart
        // from the persisted cursor if the upstream stays unavailable.
        if (self.buffer.items.len >= MAX_BACKLOG or self.delete_buffer.items.len >= MAX_BACKLOG) {
            self.drainBacklog(WorkerWriter{ .transport = self.transport, .config = self.config });
            return;
        }

        // respect backoff cooldown after failures
        if (now < self.retry_after) return;

        if (self.shouldFlush(now)) {
            self.flush();
        }
    }

    pub fn onError(_: *IngestHandler, err: anyerror) void {
        log.err("firehose: {s}", .{@errorName(err)});
    }

    pub fn onConnect(_: *IngestHandler, host: []const u8) void {
        log.info("connected to {s}", .{host});
    }

    /// one firehose commit carries many ops, unlike jetstream's one-op events.
    fn handleCommit(self: *IngestHandler, c: zat.firehose.CommitEvent) void {
        for (c.ops) |op| {
            if (!wantedCollection(op.collection)) continue;
            self.handleOp(c.repo, op);
        }
    }

    fn handleOp(self: *IngestHandler, repo: []const u8, op: zat.firehose.RepoOp) void {
        if (mem.eql(u8, op.collection, "app.bsky.actor.profile") or
            mem.eql(u8, op.collection, "sh.tangled.actor.profile"))
        {
            // rich extraction from profile records (bsky and tangled)
            if (op.action != .create and op.action != .update) return;

            const record = op.record orelse return;

            const did = self.dupe(repo) orelse return;
            var event = ActorEvent{ .did = did };

            // bsky profiles have displayName; tangled profiles don't
            if (record.getString("displayName")) |name| {
                event.display_name = self.dupe(name);
            }

            event.avatar_cid = avatarCidOf(self.arena.allocator(), record);
            if (record.getString("description")) |d| {
                event.description = self.dupe(d);
            }
            event.banner_cid = blobCidOf(self.arena.allocator(), record, "banner");

            // ingester only sees self-labels from profile records — it can never
            // correctly determine hidden. let refreshModeration cron handle it.

            self.enqueue(event);
        } else {
            // non-profile collections: just discover the DID
            if (op.action == .delete) return;

            // dedup: skip if bloom filter says we've seen this DID recently
            if (self.bloom.contains(repo)) return;
            self.bloom.insert(repo);

            const did = self.dupe(repo) orelse return;
            self.enqueue(.{ .did = did });
        }
    }

    fn handleIdentity(self: *IngestHandler, id: zat.firehose.IdentityEvent) void {
        // Keep discovery on the cursor-protected queue even when the async
        // resolver is full or its HTTP write fails. A bare row lets the durable
        // handle-enrichment queue find the account later.
        const did = self.dupe(id.did) orelse return;
        const handle = if (id.handle) |h|
            (if (!mem.eql(u8, h, "handle.invalid")) self.dupe(h) else null)
        else
            null;
        self.enqueue(.{ .did = did, .handle = handle });

        if (self.identity_pool) |pool| {
            if (identity.Did.fromSlice(id.did)) |resolver_did| {
                _ = pool.offer(io, .{ .did = resolver_did }); // discovery is queued above; drops show on /health
            }
        }
    }

    fn handleAccount(self: *IngestHandler, acct: zat.firehose.AccountEvent) void {
        if (!acct.active) {
            const did = self.dupe(acct.did) orelse return;
            self.delete_buffer.append(self.allocator, did) catch return;
        }
    }

    fn shouldFlush(self: *IngestHandler, now: i64) bool {
        return self.buffer.items.len >= MAX_BATCH or
            self.delete_buffer.items.len >= MAX_BATCH or
            (now - self.last_flush >= 5 and (self.buffer.items.len > 0 or self.delete_buffer.items.len > 0));
    }

    fn publishQueueDepth(self: *IngestHandler) void {
        queued_ingest.store(self.buffer.items.len, .monotonic);
        queued_deletes.store(self.delete_buffer.items.len, .monotonic);
    }

    fn drainBacklog(self: *IngestHandler, writer: anytype) void {
        _ = backpressure_count.fetchAdd(1, .monotonic);
        backpressure_active.store(true, .monotonic);
        defer backpressure_active.store(false, .monotonic);
        log.warn("backpressure: pausing relay consumption ({d} ingest, {d} delete)", .{
            self.buffer.items.len, self.delete_buffer.items.len,
        });
        while (self.buffer.items.len > 0 or self.delete_buffer.items.len > 0) {
            writer.waitUntil(self.retry_after);
            self.flushWith(writer);
        }
        log.info("backpressure: queue drained, resuming relay consumption", .{});
    }

    fn flush(self: *IngestHandler) void {
        self.flushWith(WorkerWriter{ .transport = self.transport, .config = self.config });
    }

    fn flushWith(self: *IngestHandler, writer: anytype) void {
        defer self.publishQueueDepth();
        // A slow HTTP request must not make the next single event immediately
        // due. Full batches still flush immediately during catch-up.
        defer self.last_flush = writer.now();
        var any_failure = false;
        var posted_through: ?i64 = null;

        if (self.buffer.items.len > 0) {
            const batch_end = @min(self.buffer.items.len, MAX_BATCH);
            const batch = self.buffer.items[0..batch_end];
            const ok = writer.post(batch, self.flushed_cursor);
            if (ok) {
                self.total_ingested += batch_end;
                const rss = getRssKB();
                log.info("+{d} actors (total: {d}, pending: {d}) cursor={d} rss={d}KB", .{
                    batch_end,                         self.total_ingested,
                    self.buffer.items.len - batch_end, self.pending_cursor,
                    rss,
                });
                if (batch_end < self.buffer.items.len) {
                    std.mem.copyForwards(
                        ActorEvent,
                        self.buffer.items[0 .. self.buffer.items.len - batch_end],
                        self.buffer.items[batch_end..],
                    );
                }
                self.buffer.shrinkRetainingCapacity(self.buffer.items.len - batch_end);

                // The durable cursor advances to the last seq we ACTUALLY sent,
                // not to pending_cursor — pending_cursor covers events still in
                // the buffer, so publishing it would skip them on restart.
                if (batch_end <= self.cursors.items.len) {
                    posted_through = completedBatchCursor(self.cursors.items, batch_end);
                    if (batch_end < self.cursors.items.len) {
                        std.mem.copyForwards(
                            i64,
                            self.cursors.items[0 .. self.cursors.items.len - batch_end],
                            self.cursors.items[batch_end..],
                        );
                    }
                    self.cursors.shrinkRetainingCapacity(self.cursors.items.len - batch_end);
                }
            } else {
                log.warn("ingest batch failed ({d} events, {d} pending), will retry in 5s", .{
                    batch_end, self.buffer.items.len,
                });
                any_failure = true;
            }
        }

        if (self.delete_buffer.items.len > 0) {
            const batch_end = @min(self.delete_buffer.items.len, MAX_BATCH);
            const batch = self.delete_buffer.items[0..batch_end];
            const ok = writer.delete(batch);
            if (ok) {
                self.total_deleted += batch_end;
                log.info("-{d} moderated (total: {d})", .{
                    batch_end, self.total_deleted,
                });
                if (batch_end < self.delete_buffer.items.len) {
                    const remaining = self.delete_buffer.items.len - batch_end;
                    var i: usize = 0;
                    while (i < remaining) : (i += 1) {
                        self.delete_buffer.items[i] = self.delete_buffer.items[batch_end + i];
                    }
                    self.delete_buffer.shrinkRetainingCapacity(remaining);
                } else {
                    self.delete_buffer.clearRetainingCapacity();
                }
            } else {
                log.warn("delete batch failed ({d} dids), will retry in 5s", .{batch_end});
                any_failure = true;
            }
        }

        if (any_failure) {
            self.consecutive_failures += 1;
            if (self.consecutive_failures >= WATCHDOG_MAX_CONSECUTIVE_FAILURES) {
                log.err("watchdog: {d} consecutive flush failures, exiting for restart", .{
                    self.consecutive_failures,
                });
                fatal.exitForRestart();
            }
            self.retry_after = timestamp() + 5;
        } else {
            self.consecutive_failures = 0;
            const now = timestamp();
            last_flush_ok_at.store(now, .monotonic);
            if (now - self.last_heartbeat >= HEARTBEAT_INTERVAL_SECS) {
                self.last_heartbeat = now;
                const span = logfire.span("ingest.heartbeat", .{});
                defer span.end();
                span.setAttribute("total_ingested", self.total_ingested);
                span.setAttribute("total_deleted", self.total_deleted);
                span.setAttribute("queued_ingest", self.buffer.items.len);
                span.setAttribute("queued_deletes", self.delete_buffer.items.len);
                span.setAttribute("backpressure_count", backpressure_count.load(.monotonic));
            }
            // Advance only past what was sent, and only while no deletes are
            // queued — a pending delete carries a lower seq, and skipping it
            // would drop a moderation event on restart.
            if (posted_through) |c| {
                if (self.delete_buffer.items.len == 0 and c > self.flushed_cursor) {
                    self.flushed_cursor = c;
                }
            }
            if (self.buffer.items.len == 0 and self.delete_buffer.items.len == 0) {
                if (self.pending_cursor > self.flushed_cursor) {
                    self.flushed_cursor = self.pending_cursor;
                }
                _ = self.arena.reset(.retain_capacity);
            }
        }
    }
};

// A relay commit can contribute multiple entries. If a POST splits that
// commit, resuming AFTER its seq would skip its remaining entries.
fn completedBatchCursor(cursors: []const i64, count: usize) i64 {
    const last = cursors[count - 1];
    if (count < cursors.len and cursors[count] == last) return last - 1;
    return last;
}

const WorkerWriter = struct {
    transport: *HttpTransport,
    config: Config,

    fn post(self: WorkerWriter, events: []const ActorEvent, cursor: i64) bool {
        return postBatch(self.transport, self.config, events, cursor);
    }

    fn delete(self: WorkerWriter, dids: []const []const u8) bool {
        return deleteActors(self.transport, self.config, dids);
    }

    fn now(_: WorkerWriter) i64 {
        return timestamp();
    }

    fn waitUntil(_: WorkerWriter, retry_after: i64) void {
        const delay = retry_after - timestamp();
        if (delay > 0) io.sleep(std.Io.Duration.fromSeconds(delay), .awake) catch {};
    }
};

// -- HTTP client functions (worker API) --

/// Every write to the worker is a JSON POST. The transport refuses a payload
/// that names no content type (error.MissingContentType) instead of guessing
/// one, so all three writers build their request here.
pub fn workerPost(url: []const u8, payload: []const u8, auth: []const u8) HttpTransport.FetchOptions {
    return .{
        .url = url,
        .method = .POST,
        .payload = payload,
        .content_type = "application/json",
        .authorization = auth,
    };
}

fn postBatch(transport: *HttpTransport, config: Config, events: []const ActorEvent, cursor: i64) bool {
    var output: std.Io.Writer.Allocating = .init(transport.allocator);
    defer output.deinit();

    var jw: json.Stringify = .{
        .writer = &output.writer,
        .options = .{ .emit_null_optional_fields = false },
    };
    jw.write(.{ .events = events, .cursor = cursor }) catch return false;

    var url_buf: [512]u8 = undefined;
    const url = std.fmt.bufPrint(&url_buf, "{s}/admin/ingest", .{config.worker_url}) catch return false;

    var auth_buf: [256]u8 = undefined;
    const auth = std.fmt.bufPrint(&auth_buf, "Bearer {s}", .{config.secret}) catch return false;

    const result = transport.fetch(workerPost(url, output.written(), auth)) catch |err| {
        log.err("worker write transport failed: {s}", .{@errorName(err)});
        return false;
    };
    defer transport.allocator.free(result.body);

    if (result.status != .ok) {
        log.err("ingest HTTP {d}: {s}", .{ @backingInt(result.status), result.body });
    }

    return result.status == .ok;
}

fn deleteActors(transport: *HttpTransport, config: Config, dids: []const []const u8) bool {
    var output: std.Io.Writer.Allocating = .init(transport.allocator);
    defer output.deinit();

    var jw: json.Stringify = .{ .writer = &output.writer };
    jw.write(.{ .dids = dids }) catch return false;

    var url_buf: [512]u8 = undefined;
    const url = std.fmt.bufPrint(&url_buf, "{s}/admin/delete", .{config.worker_url}) catch return false;

    var auth_buf: [256]u8 = undefined;
    const auth = std.fmt.bufPrint(&auth_buf, "Bearer {s}", .{config.secret}) catch return false;

    const result = transport.fetch(workerPost(url, output.written(), auth)) catch |err| {
        log.err("worker write transport failed: {s}", .{@errorName(err)});
        return false;
    };
    defer transport.allocator.free(result.body);

    return result.status == .ok;
}

fn fetchCursorOnce(transport: *HttpTransport, config: Config) ?i64 {
    var url_buf: [512]u8 = undefined;
    const url = std.fmt.bufPrint(&url_buf, "{s}/admin/cursor", .{config.worker_url}) catch return null;

    var auth_buf: [256]u8 = undefined;
    const auth = std.fmt.bufPrint(&auth_buf, "Bearer {s}", .{config.secret}) catch return null;

    const result = transport.fetch(.{
        .url = url,
        .authorization = auth,
    }) catch return null;
    defer transport.allocator.free(result.body);

    if (result.status != .ok) return null;

    const parsed = json.parseFromSlice(json.Value, transport.allocator, result.body, .{}) catch return null;
    defer parsed.deinit();

    const cursor_val = parsed.value.object.get("cursor") orelse return null;
    return switch (cursor_val) {
        .integer => |n| n,
        .float => |f| @intFromFloat(f),
        else => null,
    };
}

/// A relay `seq` is a small monotonic counter (~8.5e8 as of 2026-07). Jetstream's
/// cursor was a µs timestamp (~1.78e15). During the jetstream→relay cutover a
/// jetstream-era ingester can still be writing µs values into the shared cursor
/// key, so a stale one must be rejected rather than replayed: subscribing at a
/// seq six orders of magnitude in the future yields an empty stream, and the
/// watchdog would read that as a dead upstream instead of a bad cursor.
/// Rejecting prevents subscription; the caller must recover a valid cursor
/// rather than start live and silently skip the unavailable replay interval.
pub const MAX_PLAUSIBLE_RELAY_SEQ: i64 = 1_000_000_000_000; // 1e12

pub fn plausibleRelayCursor(cursor: ?i64) ?i64 {
    const c = cursor orelse return null;
    if (c <= 0 or c >= MAX_PLAUSIBLE_RELAY_SEQ) return null;
    return c;
}

pub fn fetchCursor(transport: *HttpTransport, config: Config) ?i64 {
    const backoff = [_]u64{ 1, 3, 10 };
    for (backoff, 0..) |delay, attempt| {
        if (fetchCursorOnce(transport, config)) |cursor| {
            if (plausibleRelayCursor(cursor)) |ok| return ok;
            log.warn("rejecting implausible relay cursor {d} (jetstream-era µs value?)", .{cursor});
            return null;
        }
        log.warn("cursor fetch failed (attempt {d}/3), retrying in {d}s", .{ attempt + 1, delay });
        io.sleep(.{ .nanoseconds = delay * std.time.ns_per_s }, .real) catch {};
    }
    return null;
}

// -- utilities --

pub fn getRssKB() u64 {
    const f = std.Io.Dir.openFileAbsolute(io, "/proc/self/statm", .{}) catch return 0;
    defer f.close(io);
    var buf: [128]u8 = undefined;
    const n = f.readStreaming(io, &.{&buf}) catch return 0;
    var it = std.mem.splitScalar(u8, buf[0..n], ' ');
    _ = it.next(); // size
    const rss_pages = std.fmt.parseInt(u64, it.next() orelse return 0, 10) catch return 0;
    return rss_pages * 4;
}

const testing = std.testing;

// identity.zig is only reachable through this file, and `zig build test` roots the
// ingest suite here — without this reference its tests are silently never run.
test {
    _ = @import("identity.zig");
}

test "wantedCollection: allowlist survives the loss of jetstream's server-side filter" {
    try testing.expect(wantedCollection("app.bsky.actor.profile"));
    try testing.expect(wantedCollection("app.bsky.feed.post"));
    try testing.expect(wantedCollection("app.bsky.feed.like"));
    try testing.expect(wantedCollection("app.bsky.graph.follow"));
    try testing.expect(wantedCollection("sh.tangled.actor.profile"));
    // jetstream expressed this as the wildcard "site.standard.*"
    try testing.expect(wantedCollection("site.standard.profile"));
    try testing.expect(wantedCollection("site.standard.anything.else"));

    try testing.expect(!wantedCollection("site.standard"));
    try testing.expect(!wantedCollection("app.bsky.feed.repost"));
    try testing.expect(!wantedCollection("com.atproto.lexicon.schema"));
    try testing.expect(!wantedCollection(""));
}

test "wantedCollection: birds.place records discover their actor" {
    // Regression: a birds.place user writes only `place.birds.*`, so with this
    // missing the commit path never saw them and discovery rested entirely on a
    // single `#identity` event — which, when slingshot failed, was dropped.
    try std.testing.expect(wantedCollection("place.birds.sighting"));
    try std.testing.expect(wantedCollection("place.birds.profile"));
    try std.testing.expect(!wantedCollection("place.birdsomething.else"));
    try std.testing.expect(!wantedCollection("com.example.spam"));
}

test "avatarCidOf renders a DAG-CBOR blob ref to the stored base32 CID" {
    const allocator = testing.allocator;
    // zzstoatzz.io's real avatar CID as of 2026-07-28. cbor.Cid.raw holds the
    // CID bytes with the 0x00 identity-multibase prefix already stripped, so a
    // decode → wrap → encode round-trip must reproduce the string exactly. A
    // regression here yields a plausible-looking but wrong avatar URL.
    const want = "bafkreiegehiwbazkkpp5nucenajtdjwinxrrnz7ukawpqnudbr47f44zhq";
    const raw = try zat.multibase.decode(allocator, want);
    defer allocator.free(raw);

    const record = zat.cbor.Value{ .map = &.{
        .{ .key = "displayName", .value = .{ .text = "nate" } },
        .{ .key = "avatar", .value = .{ .map = &.{
            .{ .key = "$type", .value = .{ .text = "blob" } },
            .{ .key = "ref", .value = .{ .cid = .{ .raw = raw } } },
            .{ .key = "mimeType", .value = .{ .text = "image/jpeg" } },
        } } },
    } };

    const got = avatarCidOf(allocator, record) orelse return error.NoAvatar;
    defer allocator.free(@constCast(got));
    try testing.expectEqualStrings(want, got);

    try testing.expectEqualStrings("nate", record.getString("displayName").?);
}

test "blobCidOf reads the banner ref with the same rendering as the avatar" {
    const allocator = testing.allocator;
    const want = "bafkreiegehiwbazkkpp5nucenajtdjwinxrrnz7ukawpqnudbr47f44zhq";
    const raw = try zat.multibase.decode(allocator, want);
    defer allocator.free(raw);

    const record = zat.cbor.Value{ .map = &.{
        .{ .key = "description", .value = .{ .text = "hi" } },
        .{ .key = "banner", .value = .{ .map = &.{
            .{ .key = "$type", .value = .{ .text = "blob" } },
            .{ .key = "ref", .value = .{ .cid = .{ .raw = raw } } },
        } } },
    } };

    const got = blobCidOf(allocator, record, "banner") orelse return error.NoBanner;
    defer allocator.free(@constCast(got));
    try testing.expectEqualStrings(want, got);
    try testing.expect(avatarCidOf(allocator, record) == null);
    try testing.expectEqualStrings("hi", record.getString("description").?);
}

test "avatarCidOf returns null when the profile has no avatar" {
    const record = zat.cbor.Value{ .map = &.{
        .{ .key = "displayName", .value = .{ .text = "nate" } },
    } };
    try testing.expect(avatarCidOf(testing.allocator, record) == null);
}

test "plausibleRelayCursor rejects a jetstream-era µs cursor" {
    // the exact value /admin/cursor served mid-cutover on 2026-07-28, written by
    // a still-running jetstream ingester after the worker switched keys
    try testing.expect(plausibleRelayCursor(1785217869995374) == null);
    try testing.expect(plausibleRelayCursor(null) == null);
    try testing.expect(plausibleRelayCursor(0) == null);
    try testing.expect(plausibleRelayCursor(-1) == null);

    // real relay seqs (~8.5e8 in 2026-07) pass through untouched
    try testing.expectEqual(@as(?i64, 853_602_603), plausibleRelayCursor(853_602_603));
    try testing.expectEqual(@as(?i64, 1), plausibleRelayCursor(1));
}

test "relay seq spaces are per-relay: a cursor is only valid for its issuer" {
    // Measured 2026-07-28, first seq from each host in zat's default_hosts:
    //   bsky.network                    32_234_719_514
    //   northamerica.firehose.network    7_620_802_119
    //   europe.firehose.network          7_032_143_016
    //   asia.firehose.network            5_581_515_167
    // These are independent relays, not mirrors. The ingester failed over to
    // asia and wrote 5_581_374_644 into the cursor key while pinned-in-config
    // to bsky — resuming that against bsky would skip ~26.6 BILLION events.
    // All four are individually "plausible", which is exactly why the range
    // guard cannot catch this and why the client must pin to one host.
    const observed = [_]i64{ 32_234_719_514, 7_620_802_119, 7_032_143_016, 5_581_515_167 };
    for (observed) |s| {
        try testing.expect(plausibleRelayCursor(s) != null);
    }
    // the spread between relays dwarfs any tolerance a heuristic could apply
    try testing.expect(observed[0] - observed[3] > 26_000_000_000);
}

test "durable cursor advances past posted batches without draining the buffer" {
    // The bug: flushed_cursor only advanced when buffer AND delete_buffer were
    // both empty. Under jetstream's server-filtered stream the buffer drained
    // constantly so it worked; the relay's firehose delivers ~4x the actor events
    // and the buffer is never empty, so the cursor sat at 0 and every restart
    // silently lost its place. Verified in prod 2026-07-28: /admin/cursor
    // returned 0 while the ingester was at seq ~32,235,000,000.
    //
    // This models the invariant the fix relies on: after posting the first N of
    // M buffered entries, the durable cursor is entry N-1's seq — NOT the
    // newest seq seen (which covers entries still buffered) and NOT 0.
    var cursors: std.ArrayList(i64) = .empty;
    defer cursors.deinit(testing.allocator);
    for ([_]i64{ 100, 101, 102, 103, 104 }) |c| {
        try cursors.append(testing.allocator, c);
    }

    const batch_end: usize = 3;
    const posted_through = cursors.items[batch_end - 1];
    try testing.expectEqual(@as(i64, 102), posted_through);

    // and the remainder keeps its own seqs for the next flush
    std.mem.copyForwards(
        i64,
        cursors.items[0 .. cursors.items.len - batch_end],
        cursors.items[batch_end..],
    );
    cursors.shrinkRetainingCapacity(cursors.items.len - batch_end);
    try testing.expectEqualSlices(i64, &.{ 103, 104 }, cursors.items);

    // advancing to the newest seen seq instead would skip 103 and 104
    try testing.expect(posted_through < 104);
}

const TestWriter = struct {
    failures: usize = 0,
    delete_failures: usize = 0,
    accepted: usize = 0,
    deleted: usize = 0,
    attempts: usize = 0,
    waits: usize = 0,
    last_cursor: i64 = 0,
    clock: i64 = 0,

    fn now(self: *TestWriter) i64 {
        return self.clock;
    }

    fn post(self: *TestWriter, events: []const ActorEvent, cursor: i64) bool {
        self.attempts += 1;
        self.last_cursor = cursor;
        if (self.failures > 0) {
            self.failures -= 1;
            return false;
        }
        self.accepted += events.len;
        return true;
    }

    fn delete(self: *TestWriter, dids: []const []const u8) bool {
        if (self.delete_failures > 0) {
            self.delete_failures -= 1;
            return false;
        }
        self.deleted += dids.len;
        return true;
    }

    fn waitUntil(self: *TestWriter, _: i64) void {
        self.waits += 1;
    }
};

test "full backlog retries without losing events or advancing past failed writes" {
    var transport: HttpTransport = undefined; // fake writer owns all I/O below
    var handler = try IngestHandler.init(testing.allocator, .{ .worker_url = "", .secret = "" }, &transport, null);
    defer handler.deinit();
    for (0..MAX_BACKLOG) |i| {
        handler.pending_cursor = @intCast(i + 1);
        handler.enqueue(.{ .did = "did:plc:test" });
    }
    var writer = TestWriter{ .failures = 2 };
    handler.flushWith(&writer);
    try testing.expectEqual(MAX_BACKLOG, handler.buffer.items.len);
    try testing.expectEqual(@as(i64, 0), handler.flushed_cursor);
    try testing.expectEqual(@as(usize, 0), writer.accepted);

    handler.drainBacklog(&writer);
    try testing.expectEqual(MAX_BACKLOG, writer.accepted);
    try testing.expectEqual(@as(usize, 12), writer.attempts);
    try testing.expectEqual(@as(usize, 0), handler.buffer.items.len);
    try testing.expectEqual(@as(usize, 0), handler.cursors.items.len);
    try testing.expectEqual(@as(i64, MAX_BACKLOG), handler.flushed_cursor);
    // The sent cursor always describes the previous acknowledged batch.
    try testing.expectEqual(@as(i64, MAX_BACKLOG - MAX_BATCH), writer.last_cursor);
    try testing.expect(!backpressure_active.load(.monotonic));
}

test "failed delete holds replay cursor until the delete succeeds" {
    var transport: HttpTransport = undefined;
    var handler = try IngestHandler.init(testing.allocator, .{ .worker_url = "", .secret = "" }, &transport, null);
    defer handler.deinit();
    handler.flushed_cursor = 10;
    handler.pending_cursor = 12;
    handler.enqueue(.{ .did = "did:plc:test" });
    try handler.delete_buffer.append(testing.allocator, "did:plc:inactive");
    var writer = TestWriter{ .delete_failures = 2 };
    handler.flushWith(&writer);
    try testing.expectEqual(@as(i64, 10), handler.flushed_cursor);
    try testing.expectEqual(@as(usize, 1), handler.delete_buffer.items.len);
    handler.drainBacklog(&writer);
    try testing.expectEqual(@as(usize, 1), writer.accepted);
    try testing.expectEqual(@as(usize, 1), writer.deleted);
    try testing.expectEqual(@as(i64, 12), handler.flushed_cursor);
}

test "a split relay commit remains replayable until all its entries succeed" {
    var transport: HttpTransport = undefined;
    var handler = try IngestHandler.init(testing.allocator, .{ .worker_url = "", .secret = "" }, &transport, null);
    defer handler.deinit();
    handler.flushed_cursor = 99;
    handler.pending_cursor = 100;
    for (0..MAX_BATCH + 1) |_| handler.enqueue(.{ .did = "did:plc:test" });
    var writer = TestWriter{};
    handler.flushWith(&writer);
    try testing.expectEqual(@as(i64, 99), handler.flushed_cursor);
    try testing.expectEqual(@as(usize, 1), handler.buffer.items.len);
    writer.failures = 1;
    handler.flushWith(&writer);
    try testing.expectEqual(@as(i64, 99), writer.last_cursor);
    try testing.expectEqual(@as(i64, 99), handler.flushed_cursor);
    handler.flushWith(&writer);
    try testing.expectEqual(@as(i64, 100), handler.flushed_cursor);
    try testing.expectEqual(@as(usize, MAX_BATCH + 1), writer.accepted);
}

test "slow flush gives the next partial batch its full collection interval" {
    var transport: HttpTransport = undefined;
    var handler = try IngestHandler.init(testing.allocator, .{ .worker_url = "", .secret = "" }, &transport, null);
    defer handler.deinit();
    handler.last_flush = 0;
    handler.enqueue(.{ .did = "did:plc:test" });
    var writer = TestWriter{ .clock = 10 }; // request finished ten seconds later
    handler.flushWith(&writer);
    handler.enqueue(.{ .did = "did:plc:next" });
    try testing.expect(!handler.shouldFlush(10));
    try testing.expect(!handler.shouldFlush(14));
    try testing.expect(handler.shouldFlush(15));
    for (1..MAX_BATCH) |_| handler.enqueue(.{ .did = "did:plc:next" });
    try testing.expect(handler.shouldFlush(10));
}

test "identity discovery without a handle survives a failed write" {
    var transport: HttpTransport = undefined;
    var handler = try IngestHandler.init(testing.allocator, .{ .worker_url = "", .secret = "" }, &transport, null);
    defer handler.deinit();
    handler.pending_cursor = 42;
    handler.handleIdentity(.{ .seq = 42, .did = "did:plc:identity", .time = "2026-09-05T00:00:00Z" });
    try testing.expectEqual(@as(usize, 1), handler.buffer.items.len);
    try testing.expectEqualStrings("did:plc:identity", handler.buffer.items[0].did);
    try testing.expect(handler.buffer.items[0].handle == null);
    var writer = TestWriter{ .failures = 1 };
    handler.flushWith(&writer);
    try testing.expectEqual(@as(usize, 1), handler.buffer.items.len);
    try testing.expectEqual(@as(i64, 0), handler.flushed_cursor);
    handler.flushWith(&writer);
    try testing.expectEqual(@as(usize, 1), writer.accepted);
    try testing.expectEqual(@as(i64, 42), handler.flushed_cursor);
}

test "identity with invalid handle still queues discovery" {
    var transport: HttpTransport = undefined;
    var handler = try IngestHandler.init(testing.allocator, .{ .worker_url = "", .secret = "" }, &transport, null);
    defer handler.deinit();
    handler.handleIdentity(.{ .seq = 1, .did = "did:plc:identity", .time = "", .handle = "handle.invalid" });
    try testing.expectEqual(@as(usize, 1), handler.buffer.items.len);
    try testing.expect(handler.buffer.items[0].handle == null);
}

test "identity with a handle queues exactly one hinted event" {
    var transport: HttpTransport = undefined;
    var handler = try IngestHandler.init(testing.allocator, .{ .worker_url = "", .secret = "" }, &transport, null);
    defer handler.deinit();
    handler.handleIdentity(.{ .seq = 1, .did = "did:plc:identity", .time = "", .handle = "example.bsky.social" });
    try testing.expectEqual(@as(usize, 1), handler.buffer.items.len);
    try testing.expectEqualStrings("example.bsky.social", handler.buffer.items[0].handle.?);
}

test {
    _ = fatal;
}

test "workerPost: the transport accepts the request it builds" {
    var transport = HttpTransport.init(std.Options.debug_io, std.testing.allocator);
    defer transport.deinit();
    // nothing listens on port 1: getting as far as the connect is the point.
    if (transport.fetch(workerPost("http://127.0.0.1:1/admin/ingest", "{}", "Bearer x"))) |result| {
        std.testing.allocator.free(result.body);
        return error.TestUnexpectedResult;
    } else |err| try std.testing.expect(err != error.MissingContentType);
}