//! /subscribe websocket server — legacy-compat live tail
//!
//! each subscriber runs a pull loop on its own thread, reading the shared
//! hot tail at its own cursor (no per-client queues). filters apply at read
//! time; options_update swaps them atomically. contract and quirks per
//! docs/upstream-subscribe-protocol.md.
//!
//! v1 and v2 share the pull fanout while retaining their distinct wire,
//! cursor-floor, and skipped-event contracts.

const std = @import("std");
const websocket = @import("websocket");
const zat = @import("zat");
const archive_mod = @import("../storage/archive.zig");
const cold = @import("../storage/cold.zig");
const disk_space = @import("../storage/disk_space.zig");
const endpoints = @import("endpoints.zig");
const filter_mod = @import("filter.zig");
const homepage = @import("homepage.zig");
const http_time = @import("http_time.zig");
const metrics = @import("../runtime/metrics.zig");
const process_metrics = @import("../runtime/process_metrics.zig");
const repo_action_limiter = @import("repo_action_limiter.zig");
const repo_export = @import("repo_export.zig");
const repo_store = @import("../ingest/backfill/repo_store.zig");
const phase_mod = @import("../ingest/backfill/phase.zig");
const status_page = @import("status_page.zig");
const api_keys = @import("api_keys.zig");
const tail_mod = @import("tail.zig");
const wire = @import("wire.zig");
const xrpcapi = @import("xrpcapi.zig");
const zstd = @import("../storage/zstd.zig");

const Io = std.Io;
const Allocator = std.mem.Allocator;
const log = std.log.scoped(.stream);

/// 8 MB, matching the Io.Threaded stacks (docs/lessons-from-zlay.md #1)
const subscriber_stack_size: usize = 8 * 1024 * 1024;
const ping_stack_size: usize = subscriber_stack_size;
const slow_lag_threshold: u64 = 100_000;
pub const default_slow_window_ns: u64 = 60 * std.time.ns_per_s;
pub const default_slow_min_rate: f64 = 5;
pub const default_subscribe_read_batch: usize = 1024;
const ping_interval: Io.Clock.Duration = .{ .raw = Io.Duration.fromSeconds(30), .clock = .awake };
const frame_write_timeout_ms: u32 = 5_000;

const SlowDetector = struct {
    window_ns: u64 = default_slow_window_ns,
    min_rate: f64 = default_slow_min_rate,
    streak_start_ns: i96 = 0,
    streak_start_pos: u64 = 0,
    in_streak: bool = false,

    fn observe(self: *SlowDetector, now_ns: i96, pos: u64, lag: u64) bool {
        if (lag <= slow_lag_threshold) {
            self.in_streak = false;
            return false;
        }
        if (!self.in_streak) {
            self.in_streak = true;
            self.streak_start_ns = now_ns;
            self.streak_start_pos = pos;
            return false;
        }
        const elapsed_ns = now_ns - self.streak_start_ns;
        if (elapsed_ns < self.window_ns) return false;
        const progress = pos -| self.streak_start_pos;
        const elapsed_seconds = @as(f64, @floatFromInt(elapsed_ns)) / std.time.ns_per_s;
        const rate = @as(f64, @floatFromInt(progress)) / elapsed_seconds;
        if (rate >= self.min_rate) {
            self.streak_start_ns = now_ns;
            self.streak_start_pos = pos;
            return false;
        }
        return true;
    }
};

const CursorPlan = struct {
    /// null means live: anchor at the published tip when delivery starts,
    /// after requireHello if requested. Replay plans are fixed pre-upgrade.
    start_idx: ?u64 = null,
    /// First raw network sequence to scan. This advances even when filtering,
    /// v1 skip rules, or message caps suppress every frame in a batch.
    start_seq: ?u64 = null,
    mode: metrics.CursorMode = .live,
    /// v2 timestamp cursor clamped to the retention floor: the subscriber is
    /// owed one #info OutdatedCursor frame before events (lexicon contract)
    outdated_info: bool = false,
};

pub const Hub = struct {
    allocator: Allocator,
    io: Io,
    tail: *tail_mod.Tail,
    stats: *metrics.Stats,
    archive: ?*archive_mod.Archive = null,
    cold_reader: ?*cold.ColdReader = null,
    xrpc: ?*xrpcapi.Api = null,
    repo_status: ?*repo_store.Store = null,
    data_dir: []const u8 = ".",
    /// Identity directory used by the public account diagnostics collector.
    /// Production always configures it, matching upstream's resolver-first
    /// handle semantics; tests may disable resolution explicitly.
    plc_url: []const u8 = "https://plc.directory",
    status_identity_resolution: bool = true,
    repo_action_rate_limits: bool = true,
    repo_action_limiter: repo_action_limiter.Limiter = .{},
    active: std.atomic.Value(u32) = .init(0),
    /// admission caps, 0 = unbounded. a recorded divergence: upstream keeps
    /// connection admission at its CDN/proxy layer (docs/README.md §7); we
    /// have no CDN, and only the process knows a cursor is about to cost
    /// archive reads (docs/semantic-parity.md).
    max_subscribers: u32 = 0,
    max_cold_readers: u32 = 0,
    cold_active: std.atomic.Value(u32) = .init(0),
    bootstrap_enabled: bool = false,
    compaction_enabled: bool = false,
    retry_enabled: bool = false,
    /// Upstream cursor replay window. Zero makes both endpoints pure-live and
    /// classifies an otherwise-valid cursor as disabled.
    cursor_lookback_ns: u64 = 36 * std.time.ns_per_hour,
    subscribe_slow_window_ns: u64 = default_slow_window_ns,
    subscribe_slow_min_rate: f64 = default_slow_min_rate,
    subscribe_read_batch: usize = default_subscribe_read_batch,
    /// serving gate: false until the bootstrap lifecycle reaches
    /// steady_state — subscribe + xrpc return 503 (upstream contract);
    /// /healthz and /metrics stay reachable
    serving: std.atomic.Value(bool) = .init(true),
    /// Set before listener cancellation begins. An upgrade that was already
    /// accepted but has not reached afterInit observes this gate and sends the
    /// same 1001 shutdown close instead of escaping the drain snapshot.
    draining: std.atomic.Value(bool) = .init(false),
    /// True only after the public listener and the optional debug listener
    /// have both been bound and their accept loops started. This is the
    /// upstream /readyz contract; it deliberately says nothing about ingest
    /// health or bootstrap steady state.
    listeners_ready: std.atomic.Value(bool) = .init(false),
    /// False while main is still opening storage (active-segment recovery,
    /// archive manifest, rocksdb) behind an already-accepting listener.
    /// Upstream starts its server first and 503-gates data paths until ready
    /// (jetstreamd runtime.go: lifecycle.IsSteadyState on subscribe, the
    /// xrpcapi Ready hook on the archive surface); this is that gate. The
    /// release store in main publishes the hub's store pointers, so any path
    /// that observes true may dereference them. Defaults true so test hubs
    /// that publish pointers before serving are unaffected.
    storage_ready: std.atomic.Value(bool) = .init(true),
    /// per-key archive metering ring, rendered into /metrics when set
    key_ring: ?*api_keys.KeyRing = null,

    /// Bounded cold-replay concurrency. fetchAdd-then-undo keeps the pair
    /// wait-free; transient overshoot is bounded by the number of racing
    /// subscriber threads and never admits past the cap.
    fn tryAcquireColdSlot(self: *Hub) bool {
        if (self.max_cold_readers == 0) return true;
        if (self.cold_active.fetchAdd(1, .acquire) >= self.max_cold_readers) {
            _ = self.cold_active.fetchSub(1, .release);
            return false;
        }
        return true;
    }

    fn releaseColdSlot(self: *Hub) void {
        if (self.max_cold_readers == 0) return;
        _ = self.cold_active.fetchSub(1, .release);
    }

    pub fn deinit(self: *Hub) void {
        self.repo_action_limiter.deinit(self.allocator);
    }
};

pub const ListenerRole = enum {
    /// Historical Stream mode: protocol and operator routes share one port.
    legacy,
    /// Upstream public listener: protocol routes only.
    public,
    /// Upstream debug listener: health, readiness, and metrics only.
    debug,
};

pub const ListenerContext = struct {
    hub: *Hub,
    role: ListenerRole,
};

fn contextHub(context: anytype) *Hub {
    const T = @TypeOf(context);
    if (T == *Hub) return context;
    if (T == *ListenerContext) return context.hub;
    @compileError("Stream websocket listener context must be *Hub or *ListenerContext");
}

fn contextRole(context: anytype) ListenerRole {
    return if (@TypeOf(context) == *Hub) .legacy else context.role;
}

const Subscriber = struct {
    hub: *Hub,
    conn: *websocket.Conn,
    /// guarded by opts_mutex; swapped whole by options_update
    filter: filter_mod.Filter,
    opts_mutex: Io.Mutex = .init,
    /// uncompressed-size cap per event; 0 = no cap. quirk: garbage → 0.
    max_msg_size: std.atomic.Value(u64),
    idx: u64,
    cursor_seq: u64,
    alive: std.atomic.Value(bool) = .init(true),
    /// Serializes socket writes with close signaling. A normal peer close must
    /// wait for an in-flight write and prevent the next one before closing the
    /// descriptor; otherwise Threaded treats the resulting BADF as a bug.
    write_mutex: Io.Mutex = .init,
    /// inverse of alive, in the shape tail.next wants for its stop check
    stopped: std.atomic.Value(bool) = .init(false),
    /// Consecutive cold passes that made no progress while hot rows were
    /// resident: a gap between the archive and the retained hot range, which
    /// upstream treats as a contract violation and disconnects on.
    cold_stall_passes: u8 = 0,
    /// v1 zstd scheme: per-message dictionary frames as BINARY ws messages.
    /// the cap still compares the UNCOMPRESSED length (documented upstream
    /// divergence from original v1, kept deliberately).
    compressor: ?zstd.DictCompressor = null,
    /// network.bsky.jetstream.subscribeEvents: proposal-0015 wire (lexicon
    /// message frames, #sync, strict seq cursors)
    v2: bool = false,
    thread: ?std.Thread = null,
    ping_thread: ?std.Thread = null,
    /// Neither worker may touch the connection or trigger Handler.close until
    /// both join handles are installed and the subscriber is published.
    start_gate: Io.Event = .unset,
    ping_stop: Io.Event = .unset,
    slow: SlowDetector = .{},

    /// the pull loop. cleanup happens in Handler.close, which joins this
    /// thread — the subscriber never frees itself, so the handler can always
    /// safely signal it.
    fn run(self: *Subscriber) void {
        self.start_gate.waitUncancelable(self.hub.io);
        if (!self.alive.load(.acquire)) return;
        defer self.workerDone();
        while (self.alive.load(.acquire)) {
            // Upstream Tail.ReadFrom: the hot/cold boundary is decided per
            // read from the cursor alone — no sticky replay mode, so there is
            // no handoff to miss (the 2026-08-08 seam wedge). cursor_seq == 0
            // means seq-unknown (live attach or a time-indexed start): follow
            // self.idx hot until the first delivered entry teaches us a seq.
            if (self.cursor_seq != 0) {
                const committed = if (self.hub.archive) |archive|
                    archive.committed_seq.load(.acquire)
                else
                    0;
                if (self.hub.tail.indexForSeq(self.cursor_seq)) |idx| {
                    self.idx = idx; // resident (or at tip, which blocks): hot
                    self.cold_stall_passes = 0;
                } else if (self.cursor_seq > committed) {
                    // nothing archived at/past the cursor: wait at the tip
                    // for the next append (upstream guards the cold delegate
                    // with cursor < nextSeq and blocks otherwise — cold is
                    // never probed to discover exhaustion)
                    self.cold_stall_passes = 0;
                    self.idx = self.hub.tail.publishedTip();
                } else switch (self.coldPass()) {
                    .progressed => {
                        self.cold_stall_passes = 0;
                        continue;
                    },
                    .empty => {
                        // committed says rows exist at/past the cursor yet the
                        // cold snapshot served none: a rotation seam. Upstream
                        // retries once, then disconnects rather than spinning;
                        // the client's reconnect resolves a fresh position.
                        self.cold_stall_passes +|= 1;
                        if (self.cold_stall_passes >= 2) {
                            log.err("subscriber cursor {d} unserveable: cold exhausted below committed {d} (rotation seam invariant violated)", .{ self.cursor_seq, committed });
                            _ = self.hub.stats.subscribe_cold_stalls_total.fetchAdd(1, .monotonic);
                            return self.noteExit(.cold_stall);
                        }
                        continue;
                    },
                    .failed => return self.noteExit(.cold_failed),
                    .contended => {
                        // cold-reader slots full: brief pause and retry —
                        // deliberate backpressure, not a stall or an error
                        self.cold_stall_passes = 0;
                        self.hub.io.sleep(Io.Duration.fromMilliseconds(20), .awake) catch
                            return self.noteExit(.cold_failed);
                        continue;
                    },
                }
            }

            var arena = std.heap.ArenaAllocator.init(self.hub.allocator);
            defer arena.deinit();
            const result = self.hub.tail.readBatch(
                arena.allocator(),
                &self.idx,
                &self.stopped,
                self.v2,
                self.hub.subscribe_read_batch,
            ) catch |err| {
                log.warn("subscriber read batch failed: {s}", .{@errorName(err)});
                return self.noteExit(.read_batch_error);
            };
            switch (result) {
                .closed => {
                    // readBatch reports .closed for both "the tail is gone"
                    // (shutdown) and "this subscriber was stopped" (its
                    // handler saw the socket close). Different stories.
                    return self.noteExit(if (self.stopped.load(.acquire)) .client_closed else .tail_closed);
                },
                .fell_behind => {
                    // evicted from under us: the next pass re-decides from
                    // cursor_seq (seq-unknown subscribers clamp to the floor,
                    // v1's documented below-floor semantics)
                    if (self.cursor_seq == 0) self.idx = self.hub.tail.gauges().base;
                    continue;
                },
                .batch => |batch| {
                    _ = self.hub.stats.subscribe_hot_reads_total.fetchAdd(1, .monotonic);
                    for (batch.entries) |entry| {
                        if (entry.seq != 0) self.cursor_seq = entry.seq +| 1;
                        if (!self.wantsPred(entry.kind, entry.did, entry.collection, entry.skip_v1)) continue;
                        const cap = self.max_msg_size.load(.monotonic);
                        if (cap != 0 and entry.json.len > cap) {
                            _ = self.hub.stats.subscribe_events_oversize_total.fetchAdd(1, .monotonic);
                            continue;
                        }
                        const wrote = self.write(entry.json) catch |err| {
                            if (err == error.Closed) return self.noteExit(.write_closed);
                            log.warn("subscriber write failed: {s}", .{@errorName(err)});
                            return self.noteExit(.write_error);
                        };
                        if (wrote == .encode_error)
                            _ = self.hub.stats.subscribe_encode_errors_total.fetchAdd(1, .monotonic);
                    }
                },
            }
            if (self.tooSlow()) return self.noteExit(.too_slow);
        }
    }

    fn noteExit(self: *Subscriber, reason: metrics.SubscriberExit) void {
        _ = self.hub.stats.subscribe_worker_exits.getPtr(reason).fetchAdd(1, .monotonic);
    }

    fn pingLoop(self: *Subscriber) void {
        self.start_gate.waitUncancelable(self.hub.io);
        if (!self.alive.load(.acquire)) return;
        while (self.alive.load(.acquire)) {
            self.ping_stop.waitTimeout(self.hub.io, .{ .duration = ping_interval }) catch |err| switch (err) {
                error.Timeout => {},
                error.Canceled => return,
            };
            if (!self.alive.load(.acquire)) return;
            self.write_mutex.lockUncancelable(self.hub.io);
            if (!self.alive.load(.acquire)) {
                self.write_mutex.unlock(self.hub.io);
                return;
            }
            var payload: [0]u8 = .{};
            self.conn.writePing(&payload) catch {
                self.noteExit(.ping_write_error);
                const interrupt = self.alive.swap(false, .acq_rel);
                self.stopped.store(true, .release);
                self.write_mutex.unlock(self.hub.io);
                self.ping_stop.set(self.hub.io);
                self.hub.tail.wake();
                if (interrupt) self.conn.interruptRead();
                return;
            };
            self.write_mutex.unlock(self.hub.io);
        }
    }

    /// A subscriber-worker failure must end the connection, not leave a
    /// ping-only zombie. Half-shutdown wakes the server reader; that reader
    /// remains the sole owner of final descriptor cleanup.
    fn workerDone(self: *Subscriber) void {
        self.write_mutex.lockUncancelable(self.hub.io);
        const interrupt = self.alive.swap(false, .acq_rel);
        self.stopped.store(true, .release);
        self.write_mutex.unlock(self.hub.io);
        self.ping_stop.set(self.hub.io);
        self.hub.tail.wake();
        if (interrupt) self.conn.interruptRead();
    }

    fn write(self: *Subscriber, json: []const u8) !cold.WriteResult {
        self.write_mutex.lockUncancelable(self.hub.io);
        defer self.write_mutex.unlock(self.hub.io);
        if (!self.alive.load(.acquire)) return error.Closed;
        const scheme = self.compression();
        if (self.compressor) |*comp| {
            const frame = comp.compress(self.hub.allocator, json) catch {
                return .encode_error;
            };
            defer self.hub.allocator.free(frame);
            try self.conn.writeBin(frame);
            _ = self.hub.stats.subscribe_events_sent.getPtr(scheme).fetchAdd(1, .monotonic);
            _ = self.hub.stats.subscribe_bytes_sent.getPtr(scheme).fetchAdd(@intCast(frame.len), .monotonic);
        } else {
            try self.conn.writeText(json);
            _ = self.hub.stats.subscribe_events_sent.getPtr(scheme).fetchAdd(1, .monotonic);
            _ = self.hub.stats.subscribe_bytes_sent.getPtr(scheme).fetchAdd(@intCast(json.len), .monotonic);
        }
        _ = self.hub.stats.subscribe_bytes_encoded.getPtr(scheme).fetchAdd(@intCast(json.len), .monotonic);
        return .sent;
    }

    fn compression(self: *const Subscriber) metrics.Compression {
        if (self.compressor != null) return .zstd;
        return if (self.conn.compression != null) .deflate else .none;
    }

    const ColdPass = enum { progressed, empty, failed, contended };

    /// One stateless cold read from cursor_seq (upstream Tail.cold). No stop
    /// floor: overlapping the hot range is harmless because the cursor is
    /// monotonic and the caller's next pass re-decides hot-vs-cold anyway.
    fn coldPass(self: *Subscriber) ColdPass {
        const hub = self.hub;
        const archive = hub.archive orelse return .failed;
        const reader = hub.cold_reader orelse return .failed;
        const from_seq = self.cursor_seq;
        if (!hub.tryAcquireColdSlot()) {
            _ = hub.stats.subscribe_cold_contended_total.fetchAdd(1, .monotonic);
            return .contended;
        }
        defer hub.releaseColdSlot();
        _ = hub.stats.subscribe_cold_reads_total.fetchAdd(1, .monotonic);
        const batch = reader.readSeqBatch(
            archive,
            self,
            wantsPred,
            self,
            coldWriteCached,
            coldObserve,
            self.v2,
            self.compressor != null,
            from_seq,
            0,
            hub.subscribe_read_batch,
        ) catch |err| {
            log.warn("seq cold replay failed: {s}", .{@errorName(err)});
            return .failed;
        };
        log.debug("seq cold batch scanned {d} entries and served {d} frames", .{ batch.scanned, batch.frames });
        if (batch.next_seq <= from_seq) return .empty;
        self.cursor_seq = batch.next_seq;
        if (self.tooSlow()) return .failed;
        return .progressed;
    }

    fn tooSlow(self: *Subscriber) bool {
        // seq-unknown (idx-anchored) subscribers have no measurable lag; a
        // 0 cursor would read as astronomically behind and drop idle live
        // connections after one quiet window.
        if (self.cursor_seq == 0) return false;
        var tip = self.hub.tail.publishedNextSeq();
        if (self.hub.archive) |archive| tip = @max(tip, archive.committed_seq.load(.acquire) +| 1);
        const lag = tip -| self.cursor_seq;
        if (!self.slow.observe(Io.Timestamp.now(self.hub.io, .real).toNanoseconds(), self.cursor_seq, lag)) return false;
        _ = self.hub.stats.subscribe_adversarial_drops_total.fetchAdd(1, .monotonic);
        if (self.v2) {
            // proposal-0015: an error frame is terminal — best-effort send,
            // then the caller disconnects (upstream handler.go sendError)
            var err_buf: [256]u8 = undefined;
            var w: std.Io.Writer = .fixed(&err_buf);
            var msg_buf: [160]u8 = undefined;
            const msg = std.fmt.bufPrint(&msg_buf, "reading below the floor rate {d} events behind the tip; reconnect with cursor={d}", .{ lag, self.cursor_seq }) catch "reading below the floor rate";
            wire.encodeErrorV2(&w, "ConsumerTooSlow", msg) catch {};
            if (w.buffered().len > 0) _ = self.write(w.buffered()) catch {};
        }
        return true;
    }

    fn coldWriteCached(self: *Subscriber, frame: cold.CachedFrame) anyerror!cold.WriteResult {
        if (!self.alive.load(.acquire)) return error.Closed;
        const cap = self.max_msg_size.load(.monotonic);
        if (cap != 0 and frame.body.len > cap) {
            _ = self.hub.stats.subscribe_events_oversize_total.fetchAdd(1, .monotonic);
            return .oversize;
        }
        if (!frame.compressed) return self.write(frame.body);

        self.write_mutex.lockUncancelable(self.hub.io);
        defer self.write_mutex.unlock(self.hub.io);
        if (!self.alive.load(.acquire)) return error.Closed;
        try self.conn.writeBin(frame.payload);
        _ = self.hub.stats.subscribe_events_sent.getPtr(.zstd).fetchAdd(1, .monotonic);
        _ = self.hub.stats.subscribe_bytes_sent.getPtr(.zstd).fetchAdd(@intCast(frame.payload.len), .monotonic);
        _ = self.hub.stats.subscribe_bytes_encoded.getPtr(.zstd).fetchAdd(@intCast(frame.body.len), .monotonic);
        return .sent;
    }

    fn coldObserve(self: *Subscriber, _: u64, result: cold.ReplayResult) anyerror!void {
        switch (result) {
            .skipped_sync => _ = self.hub.stats.subscribe_events_skipped_sync_total.fetchAdd(1, .monotonic),
            .skipped_resync => _ = self.hub.stats.subscribe_events_skipped_resync_total.fetchAdd(1, .monotonic),
            .encode_error => _ = self.hub.stats.subscribe_encode_errors_total.fetchAdd(1, .monotonic),
            .sent, .filtered, .oversize => {},
        }
    }

    fn wantsPred(self: *Subscriber, kind: wire.Kind, did: []const u8, collection: []const u8, skip: tail_mod.SkipV1) bool {
        switch (skip) {
            .sync => {
                _ = self.hub.stats.subscribe_events_skipped_sync_total.fetchAdd(1, .monotonic);
                return false;
            },
            .resync_replacement => {
                _ = self.hub.stats.subscribe_events_skipped_resync_total.fetchAdd(1, .monotonic);
                return false;
            },
            .none => {},
        }
        self.opts_mutex.lockUncancelable(self.hub.io);
        defer self.opts_mutex.unlock(self.hub.io);
        const wanted = self.filter.wants(kind, did, collection);
        if (!wanted) _ = self.hub.stats.subscribe_events_filtered_total.fetchAdd(1, .monotonic);
        return wanted;
    }

    /// atomically replace the whole filter (options_update contract)
    fn updateOptions(self: *Subscriber, new_filter: filter_mod.Filter, max_msg_size: u64) void {
        self.opts_mutex.lockUncancelable(self.hub.io);
        var old = self.filter;
        self.filter = new_filter;
        self.opts_mutex.unlock(self.hub.io);
        old.deinit();
        self.max_msg_size.store(max_msg_size, .monotonic);
    }
};

const HttpObservation = struct {
    hub: *Hub,
    started: Io.Timestamp,
    handler: ?metrics.HttpHandler,
    method: metrics.HttpMethod,
    get_block: bool = false,
    code: u16 = 500,
    served_bytes: u64 = 0,

    fn init(hub: *Hub, path: []const u8, method: []const u8) HttpObservation {
        return .{
            .hub = hub,
            .started = Io.Timestamp.now(hub.io, .awake),
            .handler = metrics.httpHandler(path, method),
            .method = metrics.httpMethod(method),
        };
    }

    fn finish(self: *HttpObservation) void {
        const elapsed = self.started.durationTo(Io.Timestamp.now(self.hub.io, .awake)).toMicroseconds();
        const duration_us: u64 = @intCast(@max(0, elapsed));
        if (self.handler) |handler|
            self.hub.stats.observeHttp(handler, self.method, self.code, duration_us);
        if (self.get_block)
            self.hub.stats.observeGetBlock(self.code, self.served_bytes, duration_us);
    }
};

pub const subscribe_events_path = "/xrpc/network.bsky.jetstream.subscribeEvents";
/// the one subprotocol the v2 endpoint speaks (proposal 0015); also the
/// lexicon-declared default, so negotiation is a pure header echo
pub const v2_subprotocol = "xrpc.v1.json";

pub const Handler = struct {
    hub: *Hub,
    conn: *websocket.Conn,
    subscriber: ?*Subscriber = null,
    /// options parsed from the query string, held until the subscriber
    /// starts (immediately, or on hello when requireHello)
    pending_filter: ?filter_mod.Filter,
    pending_max_msg_size: u64,
    cursor_plan: CursorPlan,
    awaiting_hello: bool,
    compress: bool,
    v2: bool,
    clean_disconnect_counted: bool = false,
    http_started: Io.Timestamp,
    http_handler: metrics.HttpHandler,
    http_observed: bool = false,

    pub fn init(handshake: *const websocket.Handshake, conn: *websocket.Conn, context: anytype) !Handler {
        const hub = contextHub(context);
        const url = handshake.url;
        const qi = std.mem.indexOfScalar(u8, url, '?');
        const query = if (qi) |i| url[i + 1 ..] else "";
        const path = if (qi) |i| url[0..i] else url;
        const http_started = Io.Timestamp.now(hub.io, .awake);
        const http_handler = metrics.httpHandler(path, "GET");
        const finishHttp = struct {
            fn call(h: *Hub, started: Io.Timestamp, handler: ?metrics.HttpHandler, code: u16) void {
                const label = handler orelse return;
                const elapsed = started.durationTo(Io.Timestamp.now(h.io, .awake)).toMicroseconds();
                h.stats.observeHttp(label, .get, code, @intCast(@max(0, elapsed)));
            }
        }.call;
        if (contextRole(context) == .debug or
            (!std.mem.eql(u8, path, "/subscribe") and !std.mem.eql(u8, path, subscribe_events_path)))
        {
            respond(conn, "404 Not Found", "text/plain", "not found\n");
            finishHttp(hub, http_started, http_handler, 404);
            return error.Close;
        }
        if (!hub.serving.load(.acquire)) {
            if (queryParam(query, "cursor")) |raw| {
                if (raw.len != 0)
                    _ = hub.stats.subscribe_cursor_requests.getPtr(.unavailable).fetchAdd(1, .monotonic);
            }
            respond(conn, "503 Service Unavailable", "text/plain", "bootstrap in progress\n");
            finishHttp(hub, http_started, http_handler, 503);
            return error.Close; // response already written; Close suppresses the 400
        }
        const v2 = std.mem.eql(u8, path, subscribe_events_path);
        if (v2) {
            // proposal-0015 negotiation degenerates to a header echo: the
            // lexicon-declared default IS xrpc.v1.json and it is the only
            // token we speak, so an offer is echoed and a missing offer
            // changes nothing (unnegotiated connections get identical
            // framing). Unrecognized tokens are simply not echoed.
            if (handshake.headers.get("sec-websocket-protocol")) |offered| {
                var toks = std.mem.splitScalar(u8, offered, ',');
                while (toks.next()) |tok| {
                    if (std.mem.eql(u8, std.mem.trim(u8, tok, " "), v2_subprotocol)) {
                        handshake.res_headers.add("Sec-WebSocket-Protocol", v2_subprotocol);
                        break;
                    }
                }
            }
            // the legacy names are rejected with the replacement named,
            // never silently ignored (upstream docs README §5.2)
            if (queryParam(query, "wantedCollections") != null) {
                respond(conn, "400 Bad Request", "application/json", "{\"error\":\"InvalidRequest\",\"message\":\"wantedCollections is the legacy v1 parameter; use collections\"}\n");
                finishHttp(hub, http_started, http_handler, 400);
                return error.Close;
            }
            if (queryParam(query, "wantedDids") != null) {
                respond(conn, "400 Bad Request", "application/json", "{\"error\":\"InvalidRequest\",\"message\":\"wantedDids is the legacy v1 parameter; use dids\"}\n");
                finishHttp(hub, http_started, http_handler, 400);
                return error.Close;
            }
        }

        var compress = false;
        const legacy_compress = if (queryParam(query, "compress")) |v| std.mem.eql(u8, v, "true") else false;
        const header_zstd = if (handshake.headers.get("socket-encoding")) |enc|
            std.mem.indexOf(u8, enc, "zstd") != null
        else
            false;
        if (v2) {
            if (legacy_compress or header_zstd) {
                respond(conn, "400 Bad Request", "text/plain", "compress=true / Socket-Encoding: zstd is the /subscribe (v1) opt-in; subscribeEvents uses zstdDictionary=<id> with the dictionary from getZstdDictionary\n");
                finishHttp(hub, http_started, http_handler, 400);
                return error.Close;
            }
            if (queryParam(query, "zstdDictionary")) |raw| if (raw.len != 0) {
                const id = std.fmt.parseInt(u32, raw, 10) catch {
                    respond(conn, "400 Bad Request", "text/plain", "zstdDictionary must be a positive integer zstd dictionary ID\n");
                    finishHttp(hub, http_started, http_handler, 400);
                    return error.Close;
                };
                if (id == 0) {
                    respond(conn, "400 Bad Request", "text/plain", "zstdDictionary must be a positive integer zstd dictionary ID\n");
                    finishHttp(hub, http_started, http_handler, 400);
                    return error.Close;
                }
                if (id != zstd.v2_dictionary_id) {
                    var buf: [224]u8 = undefined;
                    const body = std.fmt.bufPrint(&buf, "{{\"error\":\"UnknownZstdDictionary\",\"message\":\"dictionary {d} is no longer current; fetch dictionary {d} and reconnect\"}}\n", .{ id, zstd.v2_dictionary_id }) catch unreachable;
                    respond(conn, "400 Bad Request", "application/json", body);
                    finishHttp(hub, http_started, http_handler, 400);
                    return error.Close;
                }
                compress = true;
            };
        } else {
            compress = legacy_compress or header_zstd;
            if (compress) if (handshake.headers.get("sec-websocket-extensions")) |extensions| {
                if (offersPerMessageDeflate(extensions)) {
                    respond(conn, "400 Bad Request", "text/plain", "choose one compression scheme: custom zstd (compress=true / Socket-Encoding: zstd) or RFC 7692 permessage-deflate, not both\n");
                    finishHttp(hub, http_started, http_handler, 400);
                    return error.Close;
                }
            };
        }
        // Upstream only negotiates RFC 7692 on ordinary v1 subscriptions.
        // V2 and either custom-zstd mode explicitly decline the extension.
        if (v2 or compress) conn.disableCompression();

        var filter = (if (v2)
            filter_mod.Filter.parseQueryV2(hub.allocator, query)
        else
            filter_mod.Filter.parseQuery(hub.allocator, query)) catch |err| {
            log.debug("subscribe: invalid options: {s}", .{@errorName(err)});
            if (v2 and err != error.OutOfMemory) {
                const msg = switch (err) {
                    error.InvalidKind => "{\"error\":\"InvalidRequest\",\"message\":\"kinds values must be commit, identity, account, or sync\"}\n",
                    error.CollectionsWithoutCommits => "{\"error\":\"InvalidRequest\",\"message\":\"collections can never apply when kinds excludes commit\"}\n",
                    else => "{\"error\":\"InvalidRequest\",\"message\":\"invalid filter parameters\"}\n",
                };
                respond(conn, "400 Bad Request", "application/json", msg);
                finishHttp(hub, http_started, http_handler, 400);
                return error.Close;
            }
            finishHttp(hub, http_started, http_handler, if (err == error.OutOfMemory) 500 else 400);
            return error.InvalidRequest;
        };
        errdefer filter.deinit();

        // Cursor resolution includes parsing and happens before upgrade, so
        // malformed and too-old v2 cursors remain ordinary HTTP failures.
        const resolve_started = Io.Timestamp.now(hub.io, .real).toMicroseconds();
        const cursor_plan = resolveCursor(hub, v2, queryParam(query, "cursor")) catch |err| {
            const finished = Io.Timestamp.now(hub.io, .real).toMicroseconds();
            hub.stats.observeCursorResolve(@intCast(@max(finished - resolve_started, 0)));
            if (err == error.CursorResolveFailed) {
                _ = hub.stats.subscribe_cursor_requests.getPtr(.resolve_failed).fetchAdd(1, .monotonic);
                respond(conn, "503 Service Unavailable", "text/plain", "service not ready: cursor resolution failed\n");
                finishHttp(hub, http_started, http_handler, 503);
                return error.Close;
            }
            if (err == error.CursorTooOld) {
                _ = hub.stats.subscribe_cursor_requests.getPtr(.too_old).fetchAdd(1, .monotonic);
                const requested = std.fmt.parseInt(i64, queryParam(query, "cursor").?, 10) catch 0;
                const floor = if (hub.archive) |archive|
                    archive.manifest.lookbackFloor(Io.Timestamp.now(hub.io, .real).toMicroseconds(), hub.cursor_lookback_ns).seq
                else
                    0;
                var body_buf: [256]u8 = undefined;
                const body = if (v2)
                    std.fmt.bufPrint(&body_buf, "{{\"error\":\"CursorTooOld\",\"message\":\"cursor {d} is below the lookback floor {d}; re-backfill from your last seq via the archive\"}}\n", .{ requested, floor }) catch "{\"error\":\"CursorTooOld\"}\n"
                else
                    std.fmt.bufPrint(&body_buf, "subscribe: cursor too old: cursor {d} below lookback floor {d}; re-backfill from your last seq\n", .{ requested, floor }) catch "subscribe: cursor too old\n";
                respond(conn, "400 Bad Request", if (v2) "application/json" else "text/plain; charset=utf-8", body);
                finishHttp(hub, http_started, http_handler, 400);
                return error.Close;
            }
            finishHttp(hub, http_started, http_handler, 400);
            return error.InvalidRequest;
        };
        const resolve_finished = Io.Timestamp.now(hub.io, .real).toMicroseconds();
        hub.stats.observeCursorResolve(@intCast(@max(resolve_finished - resolve_started, 0)));
        _ = hub.stats.subscribe_cursor_requests.getPtr(cursor_plan.mode).fetchAdd(1, .monotonic);

        // quirk: requireHello is true iff the value is exactly "true".
        // v2 is server-push only — the hello mechanism does not exist there.
        const awaiting_hello = if (!v2) if (queryParam(query, "requireHello")) |v|
            std.mem.eql(u8, v, "true")
        else
            false else false;

        conn.writeTimeout(frame_write_timeout_ms) catch |err| {
            finishHttp(hub, http_started, http_handler, 500);
            return err;
        };

        return .{
            .hub = hub,
            .conn = conn,
            .pending_filter = filter,
            .pending_max_msg_size = parseMaxMsgSize(queryParam(query, "maxMessageSizeBytes")),
            .cursor_plan = cursor_plan,
            .awaiting_hello = awaiting_hello,
            .compress = compress,
            .v2 = v2,
            .http_started = http_started,
            .http_handler = http_handler orelse unreachable,
        };
    }

    /// runs after the handshake reply is on the wire — safe to start writing.
    /// with requireHello, delivery is deferred until the first valid
    /// options_update (no pre-hello buffering: the start index is resolved
    /// when the subscriber actually starts).
    pub fn afterInit(self: *Handler, context: anytype) !void {
        _ = context;
        if (self.hub.draining.load(.acquire)) {
            self.conn.writeClose(.{ .code = 1001, .reason = "server shutting down" }) catch {};
            self.noteCleanDisconnect();
            return error.ServerDraining;
        }
        if (self.awaiting_hello) return;
        try self.startSubscriber();
    }

    fn startSubscriber(self: *Handler) !void {
        const hub = self.hub;
        if (hub.max_subscribers > 0 and hub.active.load(.monotonic) >= hub.max_subscribers) {
            _ = hub.stats.subscribe_capacity_rejects_total.fetchAdd(1, .monotonic);
            self.conn.writeClose(.{ .code = 1013, .reason = "at capacity; retry later" }) catch {};
            self.noteCleanDisconnect();
            return error.AtCapacity;
        }
        const plan = self.cursor_plan;
        const start_idx = plan.start_idx orelse hub.tail.publishedTip();
        // cursor_seq == 0 means seq-unknown: live and time-indexed plans are
        // idx-anchored (live must NOT replay the resident suffix) and learn
        // their seq from the first delivered entry. Seq-cursor plans dispatch
        // hot-vs-cold from cursor_seq on every read-loop pass.
        const start_seq = plan.start_seq orelse 0;

        var compressor: ?zstd.DictCompressor = if (self.compress)
            if (self.v2)
                try zstd.DictCompressor.initLevel(&zstd.v2_dictionary, 1)
            else
                try zstd.DictCompressor.init(zstd.v1_dictionary)
        else
            null;
        errdefer if (compressor) |*comp| comp.deinit();

        const sub = try hub.allocator.create(Subscriber);
        sub.* = .{
            .hub = hub,
            .conn = self.conn,
            .filter = self.pending_filter.?,
            .max_msg_size = .init(self.pending_max_msg_size),
            .idx = start_idx,
            .cursor_seq = start_seq,
            .v2 = self.v2,
            .compressor = compressor,
            .slow = .{
                .window_ns = hub.subscribe_slow_window_ns,
                .min_rate = hub.subscribe_slow_min_rate,
            },
        };
        self.pending_filter = null; // ownership moved to the subscriber
        if (self.v2 and plan.outdated_info) {
            // lexicon contract: a clamped timestamp cursor owes the client
            // one seq-less #info OutdatedCursor frame before any event
            var info_buf: [256]u8 = undefined;
            var w: std.Io.Writer = .fixed(&info_buf);
            wire.encodeInfoV2(&w, "OutdatedCursor", "requested timestamp was below retention; resumed at the floor") catch {};
            if (w.buffered().len > 0) _ = sub.write(w.buffered()) catch {};
        }
        const scheme = sub.compression();
        sub.thread = std.Thread.spawn(.{ .stack_size = subscriber_stack_size }, Subscriber.run, .{sub}) catch |err| {
            sub.filter.deinit();
            hub.allocator.destroy(sub);
            return err;
        };
        sub.ping_thread = std.Thread.spawn(.{ .stack_size = ping_stack_size }, Subscriber.pingLoop, .{sub}) catch |err| {
            log.err("subscriber ping thread spawn failed: {s}", .{@errorName(err)});
            sub.alive.store(false, .release);
            sub.stopped.store(true, .release);
            sub.start_gate.set(hub.io);
            sub.thread.?.join();
            sub.filter.deinit();
            hub.allocator.destroy(sub);
            return err;
        };
        // Publish only after both join handles exist.  A replay failure may
        // interrupt the server reader immediately after this release; close()
        // can now always join both workers exactly once.
        self.subscriber = sub;
        _ = hub.active.fetchAdd(1, .monotonic);
        _ = hub.stats.subscribe_active.getPtr(scheme).fetchAdd(1, .monotonic);
        sub.start_gate.set(hub.io);
    }

    /// client → server messages: SubscriberSourcedMessage envelope.
    /// non-text frames are ignored; unknown types are ignored; malformed
    /// JSON closes 1007; a payload failing validation closes 1008. an
    /// invalid update while awaiting hello disconnects.
    pub fn clientMessage(self: *Handler, allocator: Allocator, data: []const u8, tpe: websocket.MessageTextType) !void {
        if (self.v2) {
            // server-push only: ANY client data frame closes the connection
            // (control frames are handled below this layer), matching the
            // atmos xrpcserver subscription contract
            return self.closeWith(1008, "network.bsky.jetstream.subscribeEvents is server-push only");
        }
        if (tpe != .text) return;

        const parsed = std.json.parseFromSlice(std.json.Value, allocator, data, .{}) catch {
            self.noteOptionsError(.bad_envelope_json);
            return self.closeWith(1007, "bad SubscriberSourcedMessage envelope");
        };
        defer parsed.deinit();
        const root = parsed.value;
        if (root != .object) {
            self.noteOptionsError(.bad_envelope_json);
            return self.closeWith(1007, "bad SubscriberSourcedMessage envelope");
        }

        const type_val = root.object.get("type") orelse {
            self.noteOptionsError(.bad_envelope_json);
            return self.closeWith(1007, "bad SubscriberSourcedMessage envelope");
        };
        if (type_val != .string) {
            self.noteOptionsError(.bad_envelope_json);
            return self.closeWith(1007, "bad SubscriberSourcedMessage envelope");
        }
        if (!std.mem.eql(u8, type_val.string, "options_update")) {
            // quirk: unknown types are logged and ignored, never fatal
            log.debug("ignoring client message type: {s}", .{type_val.string});
            return;
        }

        const payload = root.object.get("payload") orelse {
            self.noteOptionsError(.bad_payload_json);
            return self.closeWith(1007, "bad options_update payload");
        };
        if (payload != .object) {
            self.noteOptionsError(.bad_payload_json);
            return self.closeWith(1007, "bad options_update payload");
        }

        var new_filter = filter_mod.Filter.init(self.hub.allocator);
        var ok = true;
        var reason: []const u8 = "invalid options";
        if (payload.object.get("wantedCollections")) |v| {
            ok = addAll(&new_filter, v, filter_mod.Filter.addCollection, &reason);
        }
        if (ok) if (payload.object.get("wantedDids")) |v| {
            ok = addAll(&new_filter, v, filter_mod.Filter.addDid, &reason);
        };
        if (!ok) {
            new_filter.deinit();
            self.noteOptionsError(.invalid_options);
            return self.closeWith(1008, reason);
        }

        // maxMessageSizeBytes here is a JSON integer; negatives clamp to 0
        var max_size: u64 = 0;
        if (payload.object.get("maxMessageSizeBytes")) |v| {
            if (v == .integer and v.integer > 0) max_size = @intCast(v.integer);
        }

        if (self.subscriber) |sub| {
            sub.updateOptions(new_filter, max_size);
        } else {
            // hello: adopt the new options and start delivering
            if (self.pending_filter) |*f| f.deinit();
            self.pending_filter = new_filter;
            self.pending_max_msg_size = max_size;
            self.awaiting_hello = false;
            self.startSubscriber() catch |err| {
                log.err("subscriber start failed: {s}", .{@errorName(err)});
                return self.closeWith(1011, "internal error");
            };
        }
        _ = self.hub.stats.subscribe_options_updates_total.fetchAdd(1, .monotonic);
    }

    fn noteOptionsError(self: *Handler, reason: metrics.OptionsErrorReason) void {
        _ = self.hub.stats.subscribe_options_update_errors.getPtr(reason).fetchAdd(1, .monotonic);
    }

    /// Parser-level failures happen before clientMessage. The websocket
    /// dependency reports its actual size-limit error through this hook.
    pub fn clientError(self: *Handler, err: anyerror) void {
        if (err == error.TooLarge) self.noteOptionsError(.oversize);
    }

    fn closeWith(self: *Handler, code: u16, reason: []const u8) !void {
        self.conn.close(.{ .code = code, .reason = reason }) catch {};
        return error.Closed;
    }

    pub fn clientClose(self: *Handler, _: []const u8) !void {
        if (self.subscriber) |sub| self.signalStop(sub);
        self.noteCleanDisconnect();
        self.conn.close(.{}) catch {};
    }

    /// Called only by the websocket server's intentional shutdown path.
    pub fn serverClose(self: *Handler) void {
        if (self.subscriber) |sub| self.signalStop(sub);
        self.conn.writeClose(.{ .code = 1001, .reason = "server shutting down" }) catch {};
        self.noteCleanDisconnect();
    }

    /// always called by the server when the connection ends — the single
    /// place the subscriber thread is joined and freed.
    pub fn close(self: *Handler) void {
        if (self.subscriber) |sub| {
            self.signalStop(sub);
            if (sub.thread) |t| t.join();
            if (sub.ping_thread) |t| t.join();
            if (sub.compressor) |*comp| comp.deinit();
            sub.filter.deinit();
            _ = self.hub.active.fetchSub(1, .monotonic);
            _ = self.hub.stats.subscribe_active.getPtr(sub.compression()).fetchSub(1, .monotonic);
            self.hub.allocator.destroy(sub);
            self.subscriber = null;
        }
        if (self.pending_filter) |*f| f.deinit();
        if (!self.http_observed) {
            self.http_observed = true;
            const elapsed = self.http_started.durationTo(Io.Timestamp.now(self.hub.io, .awake)).toMicroseconds();
            self.hub.stats.observeHttp(self.http_handler, .get, 200, @intCast(@max(0, elapsed)));
        }
    }

    fn signalStop(self: *Handler, sub: *Subscriber) void {
        sub.write_mutex.lockUncancelable(self.hub.io);
        sub.alive.store(false, .release);
        sub.stopped.store(true, .release);
        sub.write_mutex.unlock(self.hub.io);
        sub.ping_stop.set(self.hub.io);
        self.hub.tail.wake();
    }

    fn noteCleanDisconnect(self: *Handler) void {
        if (self.clean_disconnect_counted) return;
        self.clean_disconnect_counted = true;
        _ = self.hub.stats.subscribe_clean_disconnects_total.fetchAdd(1, .monotonic);
    }

    /// plain-HTTP requests on the websocket port: /metrics and /healthz
    pub fn httpFallback(
        conn: *websocket.Conn,
        method: []const u8,
        url: []const u8,
        body: []const u8,
        headers: *const websocket.Handshake.KeyValue,
        context: anytype,
    ) void {
        const hub = contextHub(context);
        const role = contextRole(context);
        const path = if (std.mem.indexOfScalar(u8, url, '?')) |i| url[0..i] else url;
        const query = if (std.mem.indexOfScalar(u8, url, '?')) |i| url[i + 1 ..] else "";
        var observation = HttpObservation.init(hub, path, method);
        defer observation.finish();
        const debug_path = std.mem.eql(u8, path, "/healthz") or
            std.mem.eql(u8, path, "/readyz") or
            std.mem.eql(u8, path, "/metrics");
        if (role == .public and debug_path) {
            observation.code = 404;
            return respond(conn, "404 Not Found", "text/plain", "not found\n");
        }
        if (role == .debug) {
            if (!debug_path) {
                observation.code = 404;
                return respond(conn, "404 Not Found", "text/plain", "not found\n");
            }
            const head = std.mem.eql(u8, method, "HEAD");
            if (!std.mem.eql(u8, method, "GET") and !head) {
                observation.code = 405;
                return respond(conn, "405 Method Not Allowed", "text/plain", "method not allowed\n");
            }
            if (std.mem.eql(u8, path, "/healthz")) {
                observation.code = 200;
                return respondMaybeHead(conn, "200 OK", "text/plain", "ok\n", head);
            }
            if (std.mem.eql(u8, path, "/readyz")) {
                if (!hub.listeners_ready.load(.acquire)) {
                    observation.code = 503;
                    return respondMaybeHead(conn, "503 Service Unavailable", "text/plain", "not ready\n", head);
                }
                observation.code = 200;
                return respondMaybeHead(conn, "200 OK", "text/plain", "ok\n", head);
            }
            return respondMetrics(conn, hub, &observation, head);
        }
        // CORS preflight answers before ANY auth or method check: an OPTIONS
        // request never carries credentials, so the archive key gate must not
        // see it (the hosted edge's contract; ours 401'd preflights until a
        // browser consumer appeared the day after launch).
        if (std.mem.eql(u8, method, "OPTIONS")) {
            observation.code = 204;
            conn.writeFramed(cors_preflight) catch {};
            return;
        }
        // Startup gate: the listener accepts before storage exists. Only
        // surfaces that touch no store serve until main flips storage_ready;
        // /status answers 200 with a minimal page (uptime + upstream seq in
        // the normal field format) so probes see "starting", not an outage.
        if (!hub.storage_ready.load(.acquire)) {
            if (std.mem.eql(u8, path, "/status")) {
                const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s);
                var starting_buf: [224]u8 = undefined;
                const starting_body = std.fmt.bufPrint(
                    &starting_buf,
                    "state            starting (storage opening; data paths return 503)\nupstream seq     {d}\nuptime           {d}s\n",
                    .{ hub.stats.upstream_seq.load(.monotonic), now_s - hub.stats.start_time_s },
                ) catch "state            starting\n";
                observation.code = 200;
                return respond(conn, "200 OK", "text/plain; charset=utf-8", starting_body);
            }
            const storage_free = std.mem.eql(u8, path, "/healthz") or
                std.mem.eql(u8, path, "/readyz") or
                std.mem.eql(u8, path, "/metrics") or
                std.mem.eql(u8, path, "/") or
                std.mem.eql(u8, path, endpoints.json_path);
            if (!storage_free) {
                observation.code = 503;
                if (std.mem.startsWith(u8, path, "/xrpc/"))
                    return respond(conn, "503 Service Unavailable", "application/json", "{\"error\":\"ServiceUnavailable\",\"message\":\"starting\"}");
                return respond(conn, "503 Service Unavailable", "text/plain", "starting\n");
            }
        }
        // the homepage stays up during bootstrap too (it says so)
        const import_ops = std.mem.eql(u8, path, "/xrpc/network.bsky.jetstream.importTimestamps") or
            std.mem.eql(u8, path, "/xrpc/network.bsky.jetstream.getImportStatus");
        const dictionary_path = std.mem.eql(u8, path, "/xrpc/network.bsky.jetstream.getZstdDictionary");
        const ops_path = std.mem.eql(u8, path, "/healthz") or std.mem.eql(u8, path, "/readyz") or std.mem.eql(u8, path, "/metrics") or std.mem.eql(u8, path, "/status") or std.mem.eql(u8, path, "/") or std.mem.eql(u8, path, endpoints.json_path) or import_ops or dictionary_path;
        if (!hub.serving.load(.acquire) and !ops_path) {
            observation.code = 503;
            if (std.mem.startsWith(u8, path, "/xrpc/"))
                return respond(conn, "503 Service Unavailable", "application/json", "{\"error\":\"ServiceUnavailable\",\"message\":\"bootstrap in progress\"}");
            return respond(conn, "503 Service Unavailable", "text/plain", "bootstrap in progress\n");
        }
        if (hub.xrpc) |api| {
            var responder: XrpcResponder = .{ .conn = conn, .observation = &observation };
            observation.get_block = std.mem.eql(u8, method, "GET") and
                std.mem.eql(u8, path, "/xrpc/network.bsky.jetstream.getBlock");
            if (api.handle(
                &responder,
                method,
                path,
                query,
                body,
                headers.get("authorization"),
                .{
                    .range = headers.get("range"),
                    .if_match = headers.get("if-match"),
                    .if_unmodified_since = headers.get("if-unmodified-since"),
                    .if_none_match = headers.get("if-none-match"),
                    .if_modified_since = headers.get("if-modified-since"),
                    .if_range = headers.get("if-range"),
                },
            )) return;
        }
        const status_head = std.mem.eql(u8, method, "HEAD") and std.mem.eql(u8, path, "/status");
        if (!std.mem.eql(u8, method, "GET") and !status_head) {
            observation.code = 405;
            return respond(conn, "405 Method Not Allowed", "text/plain", "method not allowed\n");
        }
        if (std.mem.eql(u8, path, "/healthz")) {
            observation.code = 200;
            return respond(conn, "200 OK", "text/plain", "ok\n");
        }
        if (std.mem.eql(u8, path, "/readyz")) {
            if (!hub.listeners_ready.load(.acquire)) {
                observation.code = 503;
                return respond(conn, "503 Service Unavailable", "text/plain", "not ready\n");
            }
            observation.code = 200;
            return respond(conn, "200 OK", "text/plain", "ok\n");
        }
        if (std.mem.eql(u8, path, endpoints.json_path)) {
            observation.code = 200;
            return respond(conn, "200 OK", "application/json", endpoints.json);
        }
        if (std.mem.eql(u8, path, "/")) {
            observation.code = 200;
            const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s);
            const args: homepage.Args = .{
                .seq = hub.stats.upstream_seq.load(.monotonic),
                .events_total = hub.stats.events_total.load(.monotonic),
                .subscribers = hub.active.load(.monotonic),
                .uptime_s = now_s - hub.stats.start_time_s,
                .serving = hub.serving.load(.acquire),
            };
            var page_buf: [24 * 1024]u8 = undefined;
            const wants_html = if (headers.get("accept")) |a| std.mem.indexOf(u8, a, "text/html") != null else false;
            if (wants_html)
                return respond(conn, "200 OK", "text/html; charset=utf-8", homepage.renderHtml(&page_buf, args));
            return respond(conn, "200 OK", "text/plain; charset=utf-8", homepage.render(&page_buf, args));
        }
        if (std.mem.eql(u8, path, "/status")) {
            const wants_html = if (headers.get("accept")) |a| std.mem.indexOf(u8, a, "text/html") != null else false;
            const account_query = status_page.accountFromQueryAlloc(hub.allocator, query) catch {
                observation.code = 400;
                return respond(conn, "400 Bad Request", "text/plain", "invalid account query\n");
            };
            if (account_query) |raw_account| {
                defer hub.allocator.free(raw_account);
                const store = hub.repo_status orelse {
                    observation.code = 500;
                    return respond(conn, "500 Internal Server Error", "text/plain", "account diagnostics unavailable\n");
                };
                var arena = std.heap.ArenaAllocator.init(hub.allocator);
                defer arena.deinit();
                const account = status_page.collectAccount(
                    arena.allocator(),
                    hub.io,
                    store,
                    hub.plc_url,
                    raw_account,
                    hub.status_identity_resolution,
                ) catch {
                    observation.code = 500;
                    return respond(conn, "500 Internal Server Error", "text/plain", "failed to read account diagnostics\n");
                };
                const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds();
                var verification: ?repo_export.VerifyReport = null;
                var verification_error: []const u8 = "";
                if (!status_head and account.found) if (hub.archive) |archive| verify: {
                    if (hub.repo_action_rate_limits) {
                        var peer_ip_buf: [64]u8 = undefined;
                        const peer_ip = conn.peerIp(&peer_ip_buf) catch {
                            verification_error = "unable to identify source IP for repo verification";
                            break :verify;
                        };
                        const allowed = hub.repo_action_limiter.allow(hub.allocator, hub.io, peer_ip, now_us) catch {
                            verification_error = "unable to apply repo action rate limit";
                            break :verify;
                        };
                        if (!allowed) {
                            verification_error = "repo action rate limit exceeded";
                            break :verify;
                        }
                    }
                    verification = repo_export.verify(
                        arena.allocator(),
                        hub.io,
                        archive,
                        hub.data_dir,
                        hub.plc_url,
                        account.did,
                    ) catch |err| {
                        verification_error = std.fmt.allocPrint(arena.allocator(), "repoexport: verify {s}: {s}", .{ account.did, @errorName(err) }) catch "repo verification failed";
                        break :verify;
                    };
                };
                const page = status_page.renderAccountAlloc(hub.allocator, account, now_us, wants_html, verification, verification_error) catch {
                    observation.code = 500;
                    return respond(conn, "500 Internal Server Error", "text/plain", "failed to render account diagnostics\n");
                };
                defer hub.allocator.free(page);
                observation.code = 200;
                return respondStatus(conn, if (wants_html) "text/html; charset=utf-8" else "text/plain; charset=utf-8", page, status_head, now_us);
            }
            if (status_page.segmentTabFromQuery(query)) |which| {
                const archive = hub.archive orelse {
                    observation.code = 500;
                    return respond(conn, "500 Internal Server Error", "text/plain", "archive diagnostics unavailable\n");
                };
                const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds();
                const page = renderArchiveTab(hub, archive, which, wants_html) catch {
                    observation.code = 500;
                    return respond(conn, "500 Internal Server Error", "text/plain", "failed to read archive diagnostics\n");
                };
                defer hub.allocator.free(page);
                observation.code = 200;
                return respondStatus(conn, if (wants_html) "text/html; charset=utf-8" else "text/plain; charset=utf-8", page, status_head, now_us);
            }
            if (status_page.hostSortFromQuery(query)) |sort| {
                const store = hub.repo_status orelse {
                    observation.code = 500;
                    return respond(conn, "500 Internal Server Error", "text/plain", "host diagnostics unavailable\n");
                };
                const statuses = store.listHostStatuses(hub.allocator) catch {
                    observation.code = 500;
                    return respond(conn, "500 Internal Server Error", "text/plain", "failed to read host diagnostics\n");
                };
                defer store.freeHostStatuses(hub.allocator, statuses);
                const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds();
                const page = status_page.renderHostsAlloc(hub.allocator, statuses, sort, now_us, wants_html) catch {
                    observation.code = 500;
                    return respond(conn, "500 Internal Server Error", "text/plain", "failed to render host diagnostics\n");
                };
                defer hub.allocator.free(page);
                observation.code = 200;
                return respondStatus(conn, if (wants_html) "text/html; charset=utf-8" else "text/plain; charset=utf-8", page, status_head, now_us);
            }
            const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s);
            const lifecycle_facts = readLifecycleFacts(hub, now_s);
            const repo_facts = readRepoFacts(hub);
            const retention = readRetention(hub);
            var import_buf: [64]u8 = undefined;
            const import_job = readImportJob(hub, &import_buf);
            const args: homepage.StatusArgs = .{
                .upstream_seq = hub.stats.upstream_seq.load(.monotonic),
                .durable_seq = if (hub.archive) |archive| archive.committed_seq.load(.acquire) else 0,
                .process_events = hub.stats.events_total.load(.monotonic),
                .subscribers = hub.active.load(.monotonic),
                .uptime_s = now_s - hub.stats.start_time_s,
                .serving = hub.serving.load(.acquire),
                .bootstrap_enabled = hub.bootstrap_enabled,
                .compaction_enabled = hub.compaction_enabled,
                .retry_enabled = hub.retry_enabled,
                .verify_valid = hub.stats.verify_valid.load(.monotonic),
                .verify_invalid = hub.stats.verify_invalid.load(.monotonic),
                .phase = lifecycle_facts.phase,
                .phase_age_s = lifecycle_facts.age_s,
                .backfill_duration_us = lifecycle_facts.backfill_duration_us,
                .repos_total = repo_facts.total,
                .repos_complete = repo_facts.complete,
                .repos_pending = repo_facts.pending,
                .repos_failed = repo_facts.failed,
                .repos_unavailable = repo_facts.unavailable,
                .next_seq = if (hub.archive) |a| a.next_seq else 0,
                .lookback_s = hub.cursor_lookback_ns / std.time.ns_per_s,
                .oldest_retained_seq = retention.oldest_seq,
                .sealed_segments = retention.sealed_segments,
                .import_job = import_job,
            };
            var page_buf: [32 * 1024]u8 = undefined;
            observation.code = 200;
            if (wants_html)
                return respondStatus(conn, "text/html; charset=utf-8", homepage.renderStatusHtml(&page_buf, args), status_head, now_s * std.time.us_per_s);
            return respondStatus(conn, "text/plain; charset=utf-8", homepage.renderStatus(&page_buf, args), status_head, now_s * std.time.us_per_s);
        }
        if (std.mem.eql(u8, path, "/metrics")) {
            return respondMetrics(conn, hub, &observation, false);
        }
        observation.code = 404;
        respond(conn, "404 Not Found", "text/plain", "not found\n");
    }
};

fn respondMetrics(conn: *websocket.Conn, hub: *Hub, observation: *HttpObservation, head: bool) void {
    observation.code = 200;
    const g = hub.tail.gauges();
    const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s);
    var buf: [256 * 1024]u8 = undefined;
    const out = metrics.format(&buf, hub.stats, hub.active.load(.monotonic), .{
        .entries = g.entries,
        .bytes = g.bytes,
        .pinned_bytes = g.pinned_bytes,
        .pinned_overrun_bytes = g.pinned_overrun_bytes,
        .base = g.base,
        .tip = g.tip,
    }, now_s, process_metrics.collect(hub.io), if (hub.archive) |archive| disk_space.freeBytes(archive.dir) else null, readDurableRepos(hub));
    var key_buf: [16 * 1024]u8 = undefined;
    var key_writer: Io.Writer = .fixed(&key_buf);
    // sealed-archive totals are public by design (the site's ON DISK tile
    // reads them here; listSegments carries per-segment sizes but is
    // key-gated)
    if (hub.archive) |archive| {
        const totals = archive.manifest.sealedTotals();
        key_writer.print("# TYPE stream_archive_sealed_bytes gauge\nstream_archive_sealed_bytes {d}\n# TYPE stream_archive_sealed_segments gauge\nstream_archive_sealed_segments {d}\n", .{ totals.bytes, totals.segments }) catch {};
    }
    if (hub.key_ring) |ring| ring.renderMetrics(hub.io, &key_writer);
    respondPartsMaybeHead(conn, "200 OK", "text/plain; version=0.0.4", out, key_writer.buffered(), head);
}

/// adapter the xrpc api writes through (json / raw+etag / xrpc error)
const XrpcResponder = struct {
    conn: *websocket.Conn,
    observation: *HttpObservation,
    modified_s: ?i64 = null,
    cache_max_age_s: u64 = 0,

    /// bytes actually served on this request (0 until a 200 body went out);
    /// the api key ring charges per-key budgets from this
    pub fn servedBytes(self: *const XrpcResponder) u64 {
        return self.observation.served_bytes;
    }

    pub fn archiveModified(self: *XrpcResponder, modified_s: i64) void {
        // net/http treats the Unix epoch as an unspecified modification time.
        self.modified_s = if (modified_s == 0) null else modified_s;
    }

    pub fn archiveCacheMaxAge(self: *XrpcResponder, max_age_s: u64) void {
        self.cache_max_age_s = max_age_s;
    }

    fn cacheControl(self: *const XrpcResponder, buf: *[64]u8) []const u8 {
        if (self.cache_max_age_s == 0) return "public, no-cache";
        return std.fmt.bufPrint(buf, "public, max-age={d}", .{self.cache_max_age_s}) catch "public, no-cache";
    }

    fn lastModifiedHeader(self: *const XrpcResponder, buf: *[64]u8) []const u8 {
        const modified_s = self.modified_s orelse return "";
        var date_buf: [40]u8 = undefined;
        return std.fmt.bufPrint(buf, "Last-Modified: {s}\r\n", .{http_time.format(&date_buf, modified_s)}) catch "";
    }

    pub fn json(self: *XrpcResponder, status: u16, body: []const u8) void {
        self.observation.code = status;
        respond(self.conn, statusLine(status), "application/json", body);
    }

    pub fn raw(self: *XrpcResponder, status: u16, body: []const u8, etag_hex: []const u8) void {
        var modified_buf: [64]u8 = undefined;
        const modified = self.lastModifiedHeader(&modified_buf);
        var cache_buf: [64]u8 = undefined;
        const cache = self.cacheControl(&cache_buf);
        var buf: [640]u8 = undefined;
        const header = std.fmt.bufPrint(
            &buf,
            "HTTP/1.1 {s}\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n",
            .{ statusLine(status), body.len, etag_hex, cache, modified },
        ) catch return;
        self.observation.code = status;
        self.conn.writeFramed(header) catch return;
        if (body.len > 0) self.conn.writeFramed(body) catch return;
        if (status == 200) self.observation.served_bytes = body.len;
    }

    pub fn dictionary(self: *XrpcResponder, body: []const u8, id: u32, etag: []const u8, not_modified: bool) void {
        var buf: [896]u8 = undefined;
        const header = if (not_modified)
            std.fmt.bufPrint(
                &buf,
                "HTTP/1.1 304 Not Modified\r\n" ++ cors_headers ++ "ETag: {s}\r\nCache-Control: public, max-age=31536000, immutable\r\nX-Zstd-Dictionary-Id: {d}\r\nContent-Length: 0\r\nConnection: close\r\nServer: stream\r\n\r\n",
                .{ etag, id },
            ) catch return
        else
            std.fmt.bufPrint(
                &buf,
                "HTTP/1.1 200 OK\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nETag: {s}\r\nCache-Control: public, max-age=31536000, immutable\r\nX-Zstd-Dictionary-Id: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n",
                .{ body.len, etag, id },
            ) catch return;
        self.observation.code = if (not_modified) 304 else 200;
        self.conn.writeFramed(header) catch return;
        if (!not_modified) self.conn.writeFramed(body) catch return;
    }

    pub fn partial(self: *XrpcResponder, body: []const u8, etag_hex: []const u8, start: u64, end: u64, total: u64) void {
        var modified_buf: [64]u8 = undefined;
        const modified = self.lastModifiedHeader(&modified_buf);
        var cache_buf: [64]u8 = undefined;
        const cache = self.cacheControl(&cache_buf);
        var buf: [720]u8 = undefined;
        const header = std.fmt.bufPrint(
            &buf,
            "HTTP/1.1 206 Partial Content\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n",
            .{ body.len, start, end, total, etag_hex, cache, modified },
        ) catch return;
        self.observation.code = 206;
        self.conn.writeFramed(header) catch return;
        if (body.len > 0) self.conn.writeFramed(body) catch return;
    }

    pub fn multipartBytes(self: *XrpcResponder, body: []const u8, etag_hex: []const u8, boundary: *const [60]u8, ranges: anytype) void {
        const encoded_len = multipartEncodedLength(boundary, ranges, body.len) orelse
            return self.err(500, "InternalError", "multipart response too large");
        self.multipartHeader(etag_hex, boundary, encoded_len);
        if (self.observation.code != 206) return;
        for (ranges, 0..) |range, i| {
            var part_buf: [320]u8 = undefined;
            const part = multipartPartHeader(&part_buf, boundary, range.start, range.length, body.len, i == 0) orelse return;
            self.conn.writeFramed(part) catch return;
            const start: usize = @intCast(range.start);
            const end: usize = @intCast(range.start + range.length);
            if (end > start) self.conn.writeFramed(body[start..end]) catch return;
        }
        self.multipartClose(boundary);
    }

    pub fn multipartFile(self: *XrpcResponder, io: Io, source: Io.File, etag_hex: []const u8, boundary: *const [60]u8, ranges: anytype, total: u64) void {
        const encoded_len = multipartEncodedLength(boundary, ranges, total) orelse
            return self.err(500, "InternalError", "multipart response too large");
        self.multipartHeader(etag_hex, boundary, encoded_len);
        if (self.observation.code != 206) return;
        var copy_buf: [64 * 1024]u8 = undefined;
        for (ranges, 0..) |range, i| {
            var part_buf: [320]u8 = undefined;
            const part = multipartPartHeader(&part_buf, boundary, range.start, range.length, total, i == 0) orelse return;
            self.conn.writeFramed(part) catch return;
            var offset = range.start;
            var remaining = range.length;
            while (remaining > 0) {
                const want: usize = @intCast(@min(remaining, copy_buf.len));
                const n = source.readPositional(io, &.{copy_buf[0..want]}, offset) catch return;
                if (n == 0) return;
                self.conn.writeFramed(copy_buf[0..n]) catch return;
                offset += n;
                remaining -= n;
            }
        }
        self.multipartClose(boundary);
    }

    fn multipartHeader(self: *XrpcResponder, etag_hex: []const u8, boundary: *const [60]u8, body_len: u64) void {
        var modified_buf: [64]u8 = undefined;
        const modified = self.lastModifiedHeader(&modified_buf);
        var cache_buf: [64]u8 = undefined;
        const cache = self.cacheControl(&cache_buf);
        var buf: [832]u8 = undefined;
        const header = std.fmt.bufPrint(
            &buf,
            "HTTP/1.1 206 Partial Content\r\n" ++ cors_headers ++ "Content-Type: multipart/byteranges; boundary={s}\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n",
            .{ boundary, body_len, etag_hex, cache, modified },
        ) catch return;
        self.observation.code = 206;
        self.conn.writeFramed(header) catch {
            self.observation.code = 500;
            return;
        };
    }

    fn multipartClose(self: *XrpcResponder, boundary: *const [60]u8) void {
        var buf: [72]u8 = undefined;
        const close = std.fmt.bufPrint(&buf, "\r\n--{s}--\r\n", .{boundary}) catch return;
        self.conn.writeFramed(close) catch return;
    }

    /// Stream an immutable sealed file without retaining it per request. The
    /// upstream Go server reaches the same property through http.ServeContent
    /// and sendfile; chunking keeps Zig's memory bounded for slow clients.
    pub fn file(self: *XrpcResponder, io: Io, source: Io.File, etag_hex: []const u8, start: u64, end: u64, total: u64, range_requested: bool) void {
        // A satisfiable Range request is 206 even when it happens to span the
        // whole representation (e.g. bytes=0- or an oversized suffix).
        const partial_response = range_requested;
        const len = end - start + 1;
        var modified_buf: [64]u8 = undefined;
        const modified = self.lastModifiedHeader(&modified_buf);
        var cache_buf: [64]u8 = undefined;
        const cache = self.cacheControl(&cache_buf);
        var header_buf: [832]u8 = undefined;
        const header = if (partial_response)
            std.fmt.bufPrint(
                &header_buf,
                "HTTP/1.1 206 Partial Content\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n",
                .{ len, start, end, total, etag_hex, cache, modified },
            ) catch return
        else
            std.fmt.bufPrint(
                &header_buf,
                "HTTP/1.1 200 OK\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n",
                .{ len, etag_hex, cache, modified },
            ) catch return;
        self.observation.code = if (partial_response) 206 else 200;
        self.conn.writeFramed(header) catch return;

        var offset = start;
        var remaining = len;
        var buf: [64 * 1024]u8 = undefined;
        while (remaining > 0) {
            const want: usize = @intCast(@min(remaining, buf.len));
            const n = source.readPositional(io, &.{buf[0..want]}, offset) catch return;
            if (n == 0) return;
            self.conn.writeFramed(buf[0..n]) catch return;
            offset += n;
            remaining -= n;
        }
        if (!partial_response) self.observation.served_bytes = len;
    }

    pub fn rangeNotSatisfiable(self: *XrpcResponder, total: u64, no_overlap: bool) void {
        const body = if (no_overlap) "invalid range: failed to overlap\n" else "invalid range\n";
        var buf: [896]u8 = undefined;
        const header = if (no_overlap)
            std.fmt.bufPrint(
                &buf,
                "HTTP/1.1 416 Range Not Satisfiable\r\n" ++ cors_headers ++ "Content-Range: bytes */{d}\r\nContent-Type: text/plain; charset=utf-8\r\nX-Content-Type-Options: nosniff\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n",
                .{ total, body.len },
            ) catch return
        else
            std.fmt.bufPrint(
                &buf,
                "HTTP/1.1 416 Range Not Satisfiable\r\n" ++ cors_headers ++ "Content-Type: text/plain; charset=utf-8\r\nX-Content-Type-Options: nosniff\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n",
                .{body.len},
            ) catch return;
        self.observation.code = 416;
        self.conn.writeFramed(header) catch return;
        self.conn.writeFramed(body) catch return;
    }

    pub fn preconditionFailed(self: *XrpcResponder, etag_hex: []const u8) void {
        var modified_buf: [64]u8 = undefined;
        const modified = self.lastModifiedHeader(&modified_buf);
        var cache_buf: [64]u8 = undefined;
        const cache = self.cacheControl(&cache_buf);
        var buf: [640]u8 = undefined;
        const header = std.fmt.bufPrint(
            &buf,
            "HTTP/1.1 412 Precondition Failed\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Content-Length: 0\r\nConnection: close\r\nServer: stream\r\n\r\n",
            .{ etag_hex, cache, modified },
        ) catch return;
        self.observation.code = 412;
        self.conn.writeFramed(header) catch return;
    }

    pub fn notModified(self: *XrpcResponder, etag_hex: []const u8) void {
        var cache_buf: [64]u8 = undefined;
        const cache = self.cacheControl(&cache_buf);
        var buf: [512]u8 = undefined;
        const header = std.fmt.bufPrint(
            &buf,
            "HTTP/1.1 304 Not Modified\r\n" ++ cors_headers ++ "ETag: \"{s}\"\r\nCache-Control: {s}\r\nConnection: close\r\nServer: stream\r\n\r\n",
            .{ etag_hex, cache },
        ) catch return;
        self.observation.code = 304;
        self.conn.writeFramed(header) catch return;
    }

    /// every status err() can be called with must have an arm here: the
    /// else-arm rewrites unknown statuses to 400, which silently breaks
    /// retry semantics (429 fell through to 400 on the key ring's first
    /// day in production — metered clients treated the throttle as fatal)
    fn errStatusLine(status: u16) []const u8 {
        return switch (status) {
            401 => "401 Unauthorized",
            404 => "404 Not Found",
            405 => "405 Method Not Allowed",
            409 => "409 Conflict",
            429 => "429 Too Many Requests",
            500 => "500 Internal Server Error",
            501 => "501 Not Implemented",
            503 => "503 Service Unavailable",
            else => "400 Bad Request",
        };
    }

    pub fn err(self: *XrpcResponder, status: u16, name: []const u8, message: []const u8) void {
        var buf: [512]u8 = undefined;
        const wire_name = if (status == 500) "InternalServerError" else name;
        const body = std.fmt.bufPrint(&buf, "{{\"error\":\"{s}\",\"message\":\"{s}\"}}", .{ wire_name, message }) catch return;
        const status_line = errStatusLine(status);
        self.observation.code = status;
        respond(self.conn, status_line, "application/json", body);
    }

    /// The hosted Bluesky instances' archive-auth contract, byte for byte
    /// (probed 2026-08-13): 401, `WWW-Authenticate: Bearer`, and a bare
    /// {"error":"invalid bearer credential"} — NOT the XRPC error envelope.
    pub fn unauthorized(self: *XrpcResponder) void {
        self.observation.code = 401;
        const body = "{\"error\":\"invalid bearer credential\"}";
        var buf: [512]u8 = undefined;
        const header = std.fmt.bufPrint(
            &buf,
            "HTTP/1.1 401 Unauthorized\r\n" ++ cors_headers ++ "Content-Type: application/json\r\nWWW-Authenticate: Bearer\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n",
            .{body.len},
        ) catch return;
        self.conn.writeFramed(header) catch return;
        _ = self.conn.writeFramed(body) catch {};
    }
};

fn multipartPartHeader(buf: *[320]u8, boundary: *const [60]u8, start: u64, length: u64, total: u64, first: bool) ?[]const u8 {
    const end = if (length == 0) start - 1 else start + length - 1;
    return std.fmt.bufPrint(
        buf,
        "{s}--{s}\r\nContent-Range: bytes {d}-{d}/{d}\r\nContent-Type: application/octet-stream\r\n\r\n",
        .{ if (first) "" else "\r\n", boundary, start, end, total },
    ) catch null;
}

fn multipartEncodedLength(boundary: *const [60]u8, ranges: anytype, total: u64) ?u64 {
    var encoded: u64 = 0;
    for (ranges, 0..) |range, i| {
        var buf: [320]u8 = undefined;
        const part = multipartPartHeader(&buf, boundary, range.start, range.length, total, i == 0) orelse return null;
        encoded = std.math.add(u64, encoded, part.len) catch return null;
        encoded = std.math.add(u64, encoded, range.length) catch return null;
    }
    var close_buf: [72]u8 = undefined;
    const close = std.fmt.bufPrint(&close_buf, "\r\n--{s}--\r\n", .{boundary}) catch return null;
    return std.math.add(u64, encoded, close.len) catch null;
}

fn statusLine(status: u16) []const u8 {
    return switch (status) {
        200 => "200 OK",
        201 => "201 Created",
        400 => "400 Bad Request",
        401 => "401 Unauthorized",
        404 => "404 Not Found",
        405 => "405 Method Not Allowed",
        409 => "409 Conflict",
        500 => "500 Internal Server Error",
        503 => "503 Service Unavailable",
        else => "500 Internal Server Error",
    };
}

/// Render the segments or collections view from resident manifest metadata.
/// Both walk sealed segments only; neither opens a segment file, so the cost
/// is bounded by segment count rather than by archive size.
fn renderArchiveTab(
    hub: *Hub,
    archive: *archive_mod.Archive,
    which: []const u8,
    html: bool,
) ![]u8 {
    var arena = std.heap.ArenaAllocator.init(hub.allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    if (std.mem.eql(u8, which, "segments")) {
        const summaries = try archive.manifest.all(alloc);
        const rows = try alloc.alloc(status_page.SegmentRow, summaries.len);
        for (summaries, rows) |meta, *row| {
            row.* = .{
                .index = meta.idx,
                .size_bytes = meta.size,
                .event_count = meta.header.event_count,
                .unique_did_count = meta.header.unique_did_count,
                .block_count = meta.header.block_count,
                .min_seq = meta.header.min_seq,
                .max_seq = meta.header.max_seq,
                .min_witnessed_at = meta.header.min_witnessed_at,
                .max_witnessed_at = meta.header.max_witnessed_at,
            };
        }
        return status_page.renderSegmentsAlloc(hub.allocator, rows, html);
    }

    // Collections: fold each sealed segment's resident collection index into
    // one archive-wide table. `$`-prefixed entries are DID-level marker
    // sentinels, not real collections, so they stay out of the operator view.
    const resident = try archive.manifest.planningSegments(alloc);
    defer for (resident) |*item| item.deinit();
    var totals: std.StringHashMapUnmanaged(status_page.CollectionRow) = .empty;
    for (resident) |*sealed| {
        for (sealed.collection_index.entries) |entry| {
            if (entry.nsid.len == 0 or entry.nsid[0] == '$') continue;
            const gop = try totals.getOrPut(alloc, entry.nsid);
            if (gop.found_existing) {
                gop.value_ptr.event_count += entry.count;
                gop.value_ptr.segment_count += 1;
            } else {
                gop.key_ptr.* = try alloc.dupe(u8, entry.nsid);
                gop.value_ptr.* = .{
                    .nsid = gop.key_ptr.*,
                    .event_count = entry.count,
                    .segment_count = 1,
                };
            }
        }
    }
    const rows = try alloc.alloc(status_page.CollectionRow, totals.count());
    var it = totals.valueIterator();
    var at: usize = 0;
    while (it.next()) |value| : (at += 1) rows[at] = value.*;
    status_page.sortCollections(rows);
    return status_page.renderCollectionsAlloc(hub.allocator, rows, html);
}

/// Durable lifecycle facts for the status summary. A read failure is reported
/// as absence here on purpose: this is a presentation surface, and refusing to
/// render the whole page because one optional field could not be read would be
/// worse than omitting the field. Every path that acts on these values reads
/// them through the phase store, which fails closed.
const LifecycleFacts = struct {
    phase: ?[]const u8 = null,
    age_s: ?i64 = null,
    backfill_duration_us: ?i64 = null,
};

fn readLifecycleFacts(hub: *Hub, now_s: i64) LifecycleFacts {
    const store = hub.repo_status orelse return .{};
    var phase_store = phase_mod.Store.init(store.meta);
    const phase = (phase_store.read(hub.io) catch return .{}) orelse return .{};
    var facts: LifecycleFacts = .{ .phase = @tagName(phase) };
    if (phase_store.readEnteredAt(hub.io) catch null) |entered_us|
        facts.age_s = now_s - @divTrunc(entered_us, std.time.us_per_s);
    const timing = phase_store.readBackfillTiming(hub.io) catch phase_mod.BackfillTiming{};
    facts.backfill_duration_us = timing.durationUs();
    return facts;
}

/// Repository rollup for the status summary — upstream's BackfillStats. Null
/// total means no repository state exists at all, which the page must not
/// render as "zero of zero complete".
const RepoFacts = struct {
    total: ?u32 = null,
    complete: u32 = 0,
    pending: u32 = 0,
    failed: u32 = 0,
    unavailable: u32 = 0,
};

fn readRepoFacts(hub: *Hub) RepoFacts {
    const store = hub.repo_status orelse return .{};
    const total = store.count();
    if (total == 0) return .{};
    return .{
        .total = total,
        .complete = store.countByStatus(.complete),
        .pending = store.countByStatus(.pending),
        .failed = store.countByStatus(.failed),
        .unavailable = store.countByStatus(.unavailable),
    };
}

/// Retention for the status summary — upstream's CursorLookbackStats.
const Retention = struct {
    oldest_seq: ?u64 = null,
    sealed_segments: ?usize = null,
};

fn readRetention(hub: *Hub) Retention {
    const archive = hub.archive orelse return .{};
    const summaries = archive.manifest.all(hub.allocator) catch return .{};
    defer hub.allocator.free(summaries);
    if (hub.cursor_lookback_ns == 0) return .{ .sealed_segments = summaries.len };
    const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds();
    return .{
        .oldest_seq = archive.manifest.lookbackFloor(now_us, hub.cursor_lookback_ns).seq,
        .sealed_segments = summaries.len,
    };
}

/// Current or most recent timestamp-import job — upstream's ImportInfo. Null
/// when import has never run against this data dir.
fn readImportJob(hub: *Hub, buf: []u8) ?[]const u8 {
    const api = hub.xrpc orelse return null;
    const manager = api.import orelse return null;
    var parsed = (manager.current() catch return null) orelse return null;
    defer parsed.deinit();
    const record = parsed.value;
    return std.fmt.bufPrint(buf, "{s} ({s}, {s})", .{
        record.id,
        @tagName(record.state),
        @tagName(record.phase),
    }) catch null;
}

/// Durable repository counts for the progress gauges, read at scrape time.
///
/// Null only when there is no repository store at all — that is genuinely
/// unknown, and a panel should show a gap rather than invent a number. An
/// empty store is different: zero complete of zero total is an accurate
/// statement, and in a time series it reads as "nothing yet" rather than as a
/// claim. The status page hides the section instead, because the same fact
/// renders badly as prose; the distinction is medium, not truth.
fn readDurableRepos(hub: *Hub) ?metrics.DurableRepos {
    const store = hub.repo_status orelse return null;
    return .{
        .complete = store.countByStatus(.complete),
        .pending = store.countByStatus(.pending),
        .failed = store.countByStatus(.failed),
        .unavailable = store.countByStatus(.unavailable),
        .not_started = store.countByStatus(.not_started),
    };
}

// The hosted Bluesky edge's CORS contract, probed 2026-08-14: allow-origin *
// and these exposed headers on EVERY response (401s included) so browser
// consumers work — a local-first sync engine hit this the day after launch.
// Their Headwind-Quota-* names are omitted: we never send those headers.
const cors_headers = "Access-Control-Allow-Origin: *\r\nAccess-Control-Expose-Headers: ETag, Content-Range, Accept-Ranges, Retry-After, WWW-Authenticate, X-Zstd-Dictionary-Id\r\n";
const cors_preflight = "HTTP/1.1 204 No Content\r\nAccess-Control-Allow-Origin: *\r\nAccess-Control-Allow-Methods: GET, HEAD, POST, OPTIONS\r\nAccess-Control-Allow-Headers: Authorization, Content-Type, Range, If-Range\r\nAccess-Control-Max-Age: 600\r\nContent-Length: 0\r\nConnection: close\r\nServer: stream\r\n\r\n";

fn respond(conn: *websocket.Conn, status: []const u8, content_type: []const u8, resp_body: []const u8) void {
    respondMaybeHead(conn, status, content_type, resp_body, false);
}

fn respondMaybeHead(conn: *websocket.Conn, status: []const u8, content_type: []const u8, resp_body: []const u8, head: bool) void {
    var buf: [512]u8 = undefined;
    const header = std.fmt.bufPrint(
        &buf,
        "HTTP/1.1 {s}\r\nContent-Type: {s}\r\n" ++ cors_headers ++ "Content-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n",
        .{ status, content_type, resp_body.len },
    ) catch return;
    conn.writeFramed(header) catch return;
    if (!head and resp_body.len > 0) conn.writeFramed(resp_body) catch return;
}

fn respondParts(conn: *websocket.Conn, status: []const u8, content_type: []const u8, first: []const u8, second: []const u8) void {
    respondPartsMaybeHead(conn, status, content_type, first, second, false);
}

fn respondPartsMaybeHead(conn: *websocket.Conn, status: []const u8, content_type: []const u8, first: []const u8, second: []const u8, head: bool) void {
    var buf: [512]u8 = undefined;
    const header = std.fmt.bufPrint(
        &buf,
        "HTTP/1.1 {s}\r\n" ++ cors_headers ++ "Content-Type: {s}\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n",
        .{ status, content_type, first.len + second.len },
    ) catch return;
    conn.writeFramed(header) catch return;
    if (!head and first.len > 0) conn.writeFramed(first) catch return;
    if (!head and second.len > 0) conn.writeFramed(second) catch return;
}

fn respondStatus(conn: *websocket.Conn, content_type: []const u8, body: []const u8, head: bool, generated_us: i64) void {
    var time_buf: [40]u8 = undefined;
    const generated_at = formatRfc3339(&time_buf, generated_us);
    var header_buf: [768]u8 = undefined;
    const header = std.fmt.bufPrint(
        &header_buf,
        "HTTP/1.1 200 OK\r\n" ++ cors_headers ++ "Content-Type: {s}\r\nContent-Length: {d}\r\nCache-Control: no-store\r\nX-Status-Generated-At: {s}\r\nConnection: close\r\nServer: stream\r\n\r\n",
        .{ content_type, body.len, generated_at },
    ) catch return;
    conn.writeFramed(header) catch return;
    if (!head) conn.writeFramed(body) catch {};
}

fn formatRfc3339(buf: []u8, micros: i64) []const u8 {
    const text = zat.Datetime.fromMicros(micros).toString();
    @memcpy(buf[0..text.len], &text);
    return buf[0..text.len];
}

fn addAll(
    filter: *filter_mod.Filter,
    v: std.json.Value,
    comptime add: fn (*filter_mod.Filter, []const u8) filter_mod.ParseError!void,
    reason: *[]const u8,
) bool {
    if (v != .array) {
        reason.* = "invalid options";
        return false;
    }
    for (v.array.items) |item| {
        if (item != .string) {
            reason.* = "invalid options";
            return false;
        }
        add(filter, item.string) catch |err| {
            reason.* = switch (err) {
                error.InvalidCollection => "invalid options: invalid collection",
                error.InvalidDid => "invalid options: invalid DID",
                error.InvalidEncoding => "invalid options: invalid encoding",
                error.TooManyCollections => "too many wanted collections",
                error.TooManyDids => "too many wanted DIDs",
                // v2 query-only errors; unreachable through options_update
                error.InvalidKind, error.CollectionsWithoutCommits => "invalid options",
                error.OutOfMemory => "internal error",
            };
            return false;
        };
    }
    return true;
}

/// quirk: empty/malformed/negative/overflow → 0 ("no cap"), never an error
fn parseMaxMsgSize(raw: ?[]const u8) u64 {
    const s = raw orelse return 0;
    if (s.len == 0) return 0;
    const v = std.fmt.parseInt(i64, s, 10) catch return 0;
    if (v < 0) return 0;
    return @intCast(v);
}

fn offersPerMessageDeflate(extensions: []const u8) bool {
    var it = std.mem.splitScalar(u8, extensions, ',');
    while (it.next()) |extension| {
        var params = std.mem.splitScalar(u8, extension, ';');
        const name = std.mem.trim(u8, params.first(), " \t");
        if (std.ascii.eqlIgnoreCase(name, "permessage-deflate")) return true;
    }
    return false;
}

fn queryParam(query: []const u8, name: []const u8) ?[]const u8 {
    var it = std.mem.splitScalar(u8, query, '&');
    while (it.next()) |pair| {
        const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue;
        if (std.mem.eql(u8, pair[0..eq], name)) return pair[eq + 1 ..];
    }
    return null;
}

fn resolveCursor(hub: *Hub, v2: bool, raw_param: ?[]const u8) !CursorPlan {
    const raw = raw_param orelse return .{};
    if (raw.len == 0) return .{};
    if (hub.cursor_lookback_ns == 0) return .{ .mode = .disabled };
    const parsed = std.fmt.parseInt(i64, raw, 10) catch return error.InvalidCursor;
    if (parsed < 0) return error.InvalidCursor;
    const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds();

    // Both endpoints use the magnitude split. V2 differs only by rejecting
    // a seq cursor below the retained floor; v1 preserves its clamp.
    if (parsed < 1_000_000_000_000_000) {
        var seq: u64 = @intCast(parsed);
        var next_seq = hub.tail.publishedNextSeq();
        if (hub.archive) |archive| {
            const durable = archive.committed_seq.load(.acquire);
            if (durable != 0) next_seq = @max(next_seq, durable +| 1);
        }
        if (next_seq == 0 or seq >= next_seq)
            return .{ .mode = .clamped };
        var clamped = false;
        if (seq == 0) {
            seq = 1;
            clamped = true;
        }
        if (hub.archive) |archive| {
            const floor = archive.manifest.lookbackFloor(now_us, hub.cursor_lookback_ns).seq;
            if (floor != 0 and seq < floor) {
                if (v2) return error.CursorTooOld;
                seq = floor;
                clamped = true;
            }
        }
        const gauges = hub.tail.gauges();
        if (gauges.entries != 0) if (hub.tail.indexForSeq(seq)) |idx|
            return .{ .start_idx = idx, .start_seq = seq, .mode = if (clamped) .clamped else .seq };
        if (hub.archive != null)
            return .{
                .start_idx = gauges.base,
                .start_seq = seq,
                .mode = if (clamped) .clamped else .seq,
            };
        if (v2) return error.CursorTooOld;
        return .{ .start_idx = gauges.base, .mode = .clamped };
    }

    if (parsed > now_us) return .{ .mode = .clamped };
    // Hot-first: a time cursor within the retained hot window starts on the
    // hot tail directly. Resolving it through the archive instead lands at
    // the last sealed-segment boundary — up to a whole open segment behind —
    // and strands the subscriber on the cold reader, which delivers in
    // read-batch bursts at seal cadence (the "jerky resume" defect).
    if (hub.tail.oldestTime()) |oldest_hot_time| {
        if (parsed >= oldest_hot_time) return .{
            .start_idx = hub.tail.indexForCursor(parsed),
            .mode = .time_us,
        };
    }
    if (hub.archive) |archive| {
        const resolved = cold.resolveTimeToSeq(hub.allocator, hub.io, archive, parsed) catch |err| {
            log.err("timestamp cursor resolution failed for {d}: {s}", .{ parsed, @errorName(err) });
            return error.CursorResolveFailed;
        };
        var resolved_seq = resolved.seq;
        var clamped = resolved.clamped;
        const floor = archive.manifest.lookbackFloor(now_us, hub.cursor_lookback_ns).seq;
        if (floor != 0 and resolved_seq < floor) {
            resolved_seq = floor;
            clamped = true;
        }
        if (hub.tail.indexForSeq(resolved_seq)) |idx| return .{
            .start_idx = idx,
            .start_seq = resolved_seq,
            .mode = if (clamped) .clamped else .time_us,
            .outdated_info = v2 and clamped,
        };
        return .{
            .start_idx = hub.tail.gauges().base,
            .start_seq = resolved_seq,
            .mode = if (clamped) .clamped else .time_us,
            .outdated_info = v2 and clamped,
        };
    }
    var mode: metrics.CursorMode = .time_us;
    if (hub.tail.oldestTime()) |floor| {
        if (parsed < floor) mode = .clamped;
    }
    return .{ .start_idx = hub.tail.indexForCursor(parsed), .mode = mode, .outdated_info = v2 and mode == .clamped };
}

// === tests ===

test "queryParam" {
    try std.testing.expectEqualStrings("42", queryParam("a=1&cursor=42&b=2", "cursor").?);
    try std.testing.expect(queryParam("a=1", "cursor") == null);
}

test "parseMaxMsgSize quirks" {
    try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize(null));
    try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize(""));
    try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize("garbage"));
    try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize("-5"));
    try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize("99999999999999999999"));
    try std.testing.expectEqual(@as(u64, 1024), parseMaxMsgSize("1024"));
}

test "permessage-deflate offer detection" {
    try std.testing.expect(offersPerMessageDeflate("permessage-deflate"));
    try std.testing.expect(offersPerMessageDeflate("foo, permessage-deflate; client_max_window_bits"));
    try std.testing.expect(!offersPerMessageDeflate("x-webkit-deflate-frame"));
}

test "slow detector matches upstream sustained-rate semantics" {
    var stalled: SlowDetector = .{};
    var progressing: SlowDetector = .{};
    var caught_up: SlowDetector = .{};
    var stalled_pos: u64 = 0;
    var progressing_pos: u64 = 0;
    var dropped = false;
    for (1..121) |second| {
        const now_ns: i96 = @intCast(second * std.time.ns_per_s);
        stalled_pos += 1;
        progressing_pos += 100;
        if (stalled.observe(now_ns, stalled_pos, 1_000_000)) dropped = true;
        try std.testing.expect(!progressing.observe(now_ns, progressing_pos, 1_000_000));
        try std.testing.expect(!caught_up.observe(now_ns, 0, 0));
    }
    try std.testing.expect(dropped);
}

test "slow detector recovery resets its full window" {
    var detector: SlowDetector = .{};
    var pos: u64 = 0;
    for (1..31) |second| {
        pos += 1;
        try std.testing.expect(!detector.observe(@intCast(second * std.time.ns_per_s), pos, 1_000_000));
    }
    pos += 1_000_000;
    try std.testing.expect(!detector.observe(31 * std.time.ns_per_s, pos, 100));
    for (32..62) |second| {
        pos += 1;
        try std.testing.expect(!detector.observe(@intCast(second * std.time.ns_per_s), pos, 1_000_000));
    }
}

test "slow detector obeys configured fractional rate and window" {
    var below: SlowDetector = .{ .window_ns = 2 * std.time.ns_per_s, .min_rate = 2.5 };
    try std.testing.expect(!below.observe(0, 100, 1_000_000));
    try std.testing.expect(!below.observe(std.time.ns_per_s, 102, 1_000_000));
    // Four events over two seconds is 2/s: below the configured floor.
    try std.testing.expect(below.observe(2 * std.time.ns_per_s, 104, 1_000_000));

    var exact: SlowDetector = .{ .window_ns = 2 * std.time.ns_per_s, .min_rate = 2.5 };
    try std.testing.expect(!exact.observe(0, 100, 1_000_000));
    // Five events over two seconds is exactly the floor and re-anchors.
    try std.testing.expect(!exact.observe(2 * std.time.ns_per_s, 105, 1_000_000));
    try std.testing.expectEqual(@as(i96, 2 * std.time.ns_per_s), exact.streak_start_ns);
    try std.testing.expectEqual(@as(u64, 105), exact.streak_start_pos);
}

test "cursor resolution is pre-upgrade and preserves v1-v2 floor semantics" {
    var threaded: Io.Threaded = .init(std.testing.allocator, .{});
    defer threaded.deinit();
    var tail = tail_mod.Tail.init(std.testing.allocator, threaded.io(), 1 << 20);
    defer tail.deinit();
    try tail.append(.commit, "did:plc:test", "app.bsky.feed.post", 100, 100, "{}", "{}", .none);
    var stats: metrics.Stats = .{};
    var hub: Hub = .{
        .allocator = std.testing.allocator,
        .io = threaded.io(),
        .tail = &tail,
        .stats = &stats,
    };

    try std.testing.expectEqual(metrics.CursorMode.live, (try resolveCursor(&hub, false, null)).mode);
    try std.testing.expectEqual(metrics.CursorMode.clamped, (try resolveCursor(&hub, false, "50")).mode);
    try std.testing.expectError(error.CursorTooOld, resolveCursor(&hub, true, "50"));
    try std.testing.expectEqual(metrics.CursorMode.clamped, (try resolveCursor(&hub, true, "101")).mode);
    try std.testing.expectError(error.InvalidCursor, resolveCursor(&hub, true, "nope"));
    try std.testing.expectError(error.InvalidCursor, resolveCursor(&hub, true, "-1"));
}

test "near-now time cursor within hot retention starts hot, never cold" {
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20);
    defer tail.deinit();
    try tail.append(.commit, "did:plc:test", "app.bsky.feed.post", 1_750_000_000_000_000, 100, "{}", "{}", .none);
    try tail.append(.commit, "did:plc:test", "app.bsky.feed.post", 1_750_000_000_500_000, 101, "{}", "{}", .none);

    // The archive is present but knows nothing recent: resolving a hot-window
    // time cursor through it strands the subscriber on the cold reader (the
    // jerky-resume defect). Hot rows must win.
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf);
    var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir_buf[0..data_dir_len]);
    defer archive.deinit();

    var stats: metrics.Stats = .{};
    var hub: Hub = .{
        .allocator = testing.allocator,
        .io = io,
        .tail = &tail,
        .stats = &stats,
        .archive = &archive,
    };

    const plan = try resolveCursor(&hub, false, "1750000000200000");
    try testing.expectEqual(metrics.CursorMode.time_us, plan.mode);
    // hot plans are idx-anchored (seq-unknown): no start_seq, so the read
    // loop never routes them cold
    try testing.expectEqual(@as(?u64, null), plan.start_seq);
    // resumes at the first retained frame witnessed at/after the cursor
    try testing.expectEqual(tail.indexForSeq(101), plan.start_idx);

    // a cursor predating hot retention still consults the archive: start_seq
    // set, and the read loop dispatches it cold until it is resident
    const old_plan = try resolveCursor(&hub, false, "1749999999999999");
    try testing.expect(old_plan.start_seq != null);
}

test "XRPC readiness gate returns the upstream JSON service error" {
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20);
    defer tail.deinit();
    var stats: metrics.Stats = .{};
    var hub: Hub = .{
        .allocator = testing.allocator,
        .io = io,
        .tail = &tail,
        .stats = &stats,
        .serving = .init(false),
    };
    defer hub.deinit();

    var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer listener.deinit(io);
    const port = listener.socket.address.getPort();
    var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{
        .port = port,
        .address = "127.0.0.1",
        .max_conn = 4,
        .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 },
    });
    defer ws_server.deinit();
    const Fixture = struct {
        fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, fixture_hub: *Hub) !void {
            server.runIo(server_listener, fixture_hub);
        }
    };
    var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &hub });
    defer _ = future.cancel(io) catch {};

    var url_buf: [256]u8 = undefined;
    const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/xrpc/network.bsky.jetstream.listSegments", .{port});
    var transport = zat.HttpTransport.init(io, testing.allocator);
    transport.keep_alive = false;
    defer transport.deinit();
    var response = try transport.fetch(.{ .url = url, .max_response_size = 4096, .capture_headers = &.{"content-type"} });
    defer response.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.service_unavailable, response.status);
    try testing.expectEqualStrings("application/json", response.headers.get("content-type").?);
    const parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, response.body, .{});
    defer parsed.deinit();
    try testing.expectEqualStrings("ServiceUnavailable", parsed.value.object.get("error").?.string);
    try testing.expect(std.mem.indexOf(u8, parsed.value.object.get("message").?.string, "bootstrap in progress") != null);
}

test "xrpc error status lines: every emitted status has its own line" {
    // the key-ring 429 regression: unknown statuses rewrite to 400 and
    // break client retry classification
    try std.testing.expectEqualStrings("429 Too Many Requests", XrpcResponder.errStatusLine(429));
    try std.testing.expectEqualStrings("401 Unauthorized", XrpcResponder.errStatusLine(401));
    try std.testing.expectEqualStrings("503 Service Unavailable", XrpcResponder.errStatusLine(503));
    try std.testing.expectEqualStrings("400 Bad Request", XrpcResponder.errStatusLine(400));
}

test "startup gate: storage-free surfaces serve while everything else 503s as starting" {
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20);
    defer tail.deinit();
    var stats: metrics.Stats = .{};
    var hub: Hub = .{
        .allocator = testing.allocator,
        .io = io,
        .tail = &tail,
        .stats = &stats,
        .serving = .init(false),
        .storage_ready = .init(false),
    };
    defer hub.deinit();
    var context: ListenerContext = .{ .hub = &hub, .role = .public };

    var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer listener.deinit(io);
    const port = listener.socket.address.getPort();
    var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{
        .port = port,
        .address = "127.0.0.1",
        .max_conn = 4,
        .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 },
    });
    defer ws_server.deinit();
    const Fixture = struct {
        fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, listener_context: *ListenerContext) !void {
            server.runIo(server_listener, listener_context);
        }
    };
    var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &context });
    defer _ = future.cancel(io) catch {};

    var transport = zat.HttpTransport.init(io, testing.allocator);
    transport.keep_alive = false;
    defer transport.deinit();
    var url_buf: [256]u8 = undefined;

    // /status stays 200 with the probe-parseable minimal page: fleet-health
    // reads "upstream seq" and "uptime" from exactly these field names.
    var status = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/status", .{port}), .max_response_size = 1024 });
    defer status.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.ok, status.status);
    try testing.expect(std.mem.indexOf(u8, status.body, "state            starting") != null);
    try testing.expect(std.mem.indexOf(u8, status.body, "upstream seq     0") != null);
    try testing.expect(std.mem.indexOf(u8, status.body, "uptime           ") != null);

    // comptime endpoint surface needs no storage and serves during startup —
    // this is what keeps the homepage's httpInclude green through a deploy
    var eps = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}{s}", .{ port, endpoints.json_path }), .max_response_size = 16384 });
    defer eps.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.ok, eps.status);

    var home = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/", .{port}), .max_response_size = 32768 });
    defer home.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.ok, home.status);

    // storage-backed paths 503 as starting: xrpc in the JSON error shape
    var xrpc_response = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/xrpc/network.bsky.jetstream.listSegments", .{port}), .max_response_size = 1024 });
    defer xrpc_response.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.service_unavailable, xrpc_response.status);
    try testing.expect(std.mem.indexOf(u8, xrpc_response.body, "\"error\":\"ServiceUnavailable\"") != null);
    try testing.expect(std.mem.indexOf(u8, xrpc_response.body, "starting") != null);

    var tab = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/status?tab=segments", .{port}), .max_response_size = 1024 });
    defer tab.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.ok, tab.status);
    try testing.expect(std.mem.indexOf(u8, tab.body, "starting") != null);

    // once main publishes the stores, the same connection path serves fully
    hub.storage_ready.store(true, .release);
    hub.serving.store(true, .release);
    var after = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/status", .{port}), .max_response_size = 8192 });
    defer after.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.ok, after.status);
    try testing.expect(std.mem.indexOf(u8, after.body, "state            starting") == null);
}

test "debug readyz tracks listener readiness rather than ingest serving" {
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20);
    defer tail.deinit();
    var stats: metrics.Stats = .{};
    var hub: Hub = .{
        .allocator = testing.allocator,
        .io = io,
        .tail = &tail,
        .stats = &stats,
        .serving = .init(false),
    };
    defer hub.deinit();
    var context: ListenerContext = .{ .hub = &hub, .role = .debug };

    var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer listener.deinit(io);
    const port = listener.socket.address.getPort();
    var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{
        .port = port,
        .address = "127.0.0.1",
        .max_conn = 4,
        .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 },
    });
    defer ws_server.deinit();
    const Fixture = struct {
        fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, listener_context: *ListenerContext) !void {
            server.runIo(server_listener, listener_context);
        }
    };
    var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &context });
    defer _ = future.cancel(io) catch {};

    var url_buf: [128]u8 = undefined;
    const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/readyz", .{port});
    var transport = zat.HttpTransport.init(io, testing.allocator);
    transport.keep_alive = false;
    defer transport.deinit();
    var unavailable = try transport.fetch(.{ .url = url, .max_response_size = 256 });
    defer unavailable.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.service_unavailable, unavailable.status);
    try testing.expectEqualStrings("not ready\n", unavailable.body);

    hub.listeners_ready.store(true, .release);
    var ready = try transport.fetch(.{ .url = url, .max_response_size = 256 });
    defer ready.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.ok, ready.status);
    try testing.expectEqualStrings("ok\n", ready.body);
    try testing.expect(!hub.serving.load(.acquire));
}

test "accounts HTTP route verifies real archive state and rate limits by port-free peer IP" {
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var fixture_arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer fixture_arena.deinit();
    const alloc = fixture_arena.allocator();

    const did = "did:plc:statusverifyfixture";
    const rev = "3k2abcdefghij";
    const record = try zat.cbor.encodeAlloc(alloc, .{ .map = &.{
        .{ .key = "$type", .value = .{ .text = "app.bsky.feed.post" } },
        .{ .key = "text", .value = .{ .text = "status route verification" } },
    } });
    const record_cid = try zat.cbor.Cid.forDagCbor(alloc, record);
    var authoritative_tree = zat.mst.Mst.init(alloc);
    try authoritative_tree.put("app.bsky.feed.post/fixture", record_cid);
    const data_root = try authoritative_tree.rootCid();
    const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(43));
    const signed = try zat.signCommit(alloc, .{ .did = did, .rev = rev, .data = data_root }, &keypair);
    const commit_cid_string = try zat.multibase.base32lower.encode(alloc, signed.cid.raw);
    const get_blocks_car = try zat.car.writeAlloc(alloc, .{
        .roots = &.{},
        .blocks = &.{.{ .cid_raw = signed.cid.raw, .data = signed.bytes }},
    });

    var identity_listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer identity_listener.deinit(io);
    var identity_endpoint_buf: [96]u8 = undefined;
    const identity_endpoint = try std.fmt.bufPrint(&identity_endpoint_buf, "http://127.0.0.1:{d}", .{identity_listener.socket.address.getPort()});
    const did_doc = try std.fmt.allocPrint(alloc,
        \\{{"id":"{s}","alsoKnownAs":["at://status.example"],"verificationMethod":[],"service":[{{"id":"#atproto_pds","type":"AtprotoPersonalDataServer","serviceEndpoint":"{s}"}}]}}
    , .{ did, identity_endpoint });
    const latest_json = try std.fmt.allocPrint(alloc, "{{\"cid\":\"{s}\",\"rev\":\"{s}\"}}", .{ commit_cid_string, rev });
    var identity_calls: std.atomic.Value(u32) = .init(0);
    const IdentityFixture = struct {
        fn run(server: *Io.net.Server, task_io: Io, calls: *std.atomic.Value(u32), doc: []const u8, latest: []const u8, blocks_car: []const u8) !void {
            while (true) {
                var stream = try server.accept(task_io);
                defer stream.close(task_io);
                var read_buf: [4096]u8 = undefined;
                var write_buf: [4096]u8 = undefined;
                var reader = stream.reader(task_io, &read_buf);
                var writer = stream.writer(task_io, &write_buf);
                var http = std.http.Server.init(&reader.interface, &writer.interface);
                var request = try http.receiveHead();
                _ = calls.fetchAdd(1, .monotonic);
                const target = request.head.target;
                const response_body, const content_type = if (std.mem.indexOf(u8, target, "getLatestCommit") != null)
                    .{ latest, "application/json" }
                else if (std.mem.indexOf(u8, target, "getBlocks") != null)
                    .{ blocks_car, "application/vnd.ipld.car" }
                else
                    .{ doc, "application/json" };
                try request.respond(response_body, .{
                    .status = .ok,
                    .keep_alive = false,
                    .extra_headers = &.{.{ .name = "content-type", .value = content_type }},
                });
            }
        }
    };
    var identity_future = try io.concurrent(IdentityFixture.run, .{ &identity_listener, io, &identity_calls, did_doc, latest_json, get_blocks_car });
    defer _ = identity_future.cancel(io) catch {};

    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf);
    const data_dir = data_dir_buf[0..data_dir_len];
    var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir);
    defer archive.deinit();
    _ = try archive.append(.{
        .seq = 1,
        .witnessed_at = 1_750_000_000_000_000,
        .indexed_at = 0,
        .kind = .create,
        .did = did,
        .collection = "app.bsky.feed.post",
        .rkey = "fixture",
        .rev = rev,
        .payload = record,
    }, 1);

    const meta_store = @import("../storage/meta_store.zig");
    var meta = try meta_store.Store.open(testing.allocator, data_dir);
    defer meta.deinit();
    var repo_status = try repo_store.Store.init(testing.allocator, io, &meta);
    defer repo_status.deinit();
    try repo_status.put(did, .{
        .status = .complete,
        .active = true,
        .rev = rev,
        .latest_rev = rev,
        .handle = "status.example",
        .pds = identity_endpoint,
        .host = "127.0.0.1",
        .record_count = 1,
    });

    var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20);
    defer tail.deinit();
    var stats: metrics.Stats = .{};
    var hub: Hub = .{
        .allocator = testing.allocator,
        .io = io,
        .tail = &tail,
        .stats = &stats,
        .archive = &archive,
        .repo_status = &repo_status,
        .data_dir = data_dir,
        .plc_url = identity_endpoint,
    };
    defer hub.deinit();

    var stream_listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer stream_listener.deinit(io);
    const stream_port = stream_listener.socket.address.getPort();
    var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{
        .port = stream_port,
        .address = "127.0.0.1",
        .max_conn = 16,
        .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 },
    });
    defer ws_server.deinit();
    const StreamFixture = struct {
        fn run(server: *websocket.Server(Handler), listener: *Io.net.Server, fixture_hub: *Hub) !void {
            server.runIo(listener, fixture_hub);
        }
    };
    var stream_future = try io.concurrent(StreamFixture.run, .{ &ws_server, &stream_listener, &hub });
    defer _ = stream_future.cancel(io) catch {};

    var status_url_buf: [256]u8 = undefined;
    const status_url = try std.fmt.bufPrint(&status_url_buf, "http://127.0.0.1:{d}/status?tab=accounts&account={s}", .{ stream_port, did });
    var transport = zat.HttpTransport.init(io, testing.allocator);
    transport.keep_alive = false;
    defer transport.deinit();
    for (0..4) |_| {
        const response = try transport.fetch(.{ .url = status_url, .accept = "text/html", .max_response_size = 256 * 1024 });
        defer testing.allocator.free(response.body);
        try testing.expectEqual(std.http.Status.ok, response.status);
        try testing.expect(std.mem.indexOf(u8, response.body, "Repo verification") != null);
        try testing.expect(std.mem.indexOf(u8, response.body, "match-ok\">Match") != null);
        try testing.expect(std.mem.indexOf(u8, response.body, "repo action rate limit exceeded") == null);
    }
    try testing.expectEqual(@as(u32, 12), identity_calls.load(.monotonic));

    const limited = try transport.fetch(.{ .url = status_url, .accept = "text/html", .max_response_size = 256 * 1024 });
    defer testing.allocator.free(limited.body);
    try testing.expectEqual(std.http.Status.ok, limited.status);
    try testing.expect(std.mem.indexOf(u8, limited.body, "repo action rate limit exceeded") != null);
    try testing.expectEqual(@as(u32, 12), identity_calls.load(.monotonic));

    const head = try transport.fetch(.{ .url = status_url, .method = .HEAD, .accept = "text/html", .max_response_size = 256 * 1024 });
    defer testing.allocator.free(head.body);
    try testing.expectEqual(std.http.Status.ok, head.status);
    try testing.expectEqual(@as(usize, 0), head.body.len);
    try testing.expectEqual(@as(u32, 12), identity_calls.load(.monotonic));

    hub.repo_action_rate_limits = false;
    const trusted = try transport.fetch(.{ .url = status_url, .accept = "text/html", .max_response_size = 256 * 1024 });
    defer testing.allocator.free(trusted.body);
    try testing.expectEqual(std.http.Status.ok, trusted.status);
    try testing.expect(std.mem.indexOf(u8, trusted.body, "match-ok\">Match") != null);
    try testing.expect(std.mem.indexOf(u8, trusted.body, "repo action rate limit exceeded") == null);
    try testing.expectEqual(@as(u32, 15), identity_calls.load(.monotonic));
}

test "cold-reader slots: cap admits, saturates, releases; zero is unbounded" {
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20);
    defer tail.deinit();
    var stats: metrics.Stats = .{};
    var hub: Hub = .{
        .allocator = testing.allocator,
        .io = io,
        .tail = &tail,
        .stats = &stats,
        .max_cold_readers = 2,
    };
    try testing.expect(hub.tryAcquireColdSlot());
    try testing.expect(hub.tryAcquireColdSlot());
    try testing.expect(!hub.tryAcquireColdSlot()); // saturated
    hub.releaseColdSlot();
    try testing.expect(hub.tryAcquireColdSlot()); // freed slot readmits
    hub.releaseColdSlot();
    hub.releaseColdSlot();
    try testing.expectEqual(@as(u32, 0), hub.cold_active.load(.monotonic));

    hub.max_cold_readers = 0; // unbounded: never blocks, never counts
    try testing.expect(hub.tryAcquireColdSlot());
    hub.releaseColdSlot();
    try testing.expectEqual(@as(u32, 0), hub.cold_active.load(.monotonic));
}

test "real-HTTP archive serving does not leak per request (prod RSS ratchet regression)" {
    // 2026-08-21: prod RSS grew ~575 B per archive request (anonymous heap,
    // smaps-verified). The Api layer alone is leak-free under this same
    // allocator, so this test drives the full socket -> dispatch -> responder
    // path; testing.allocator fails the test if any request leaks.
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20);
    defer tail.deinit();
    var stats: metrics.Stats = .{};

    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buffer: [Io.Dir.max_path_bytes]u8 = undefined;
    const path_len = try tmp.dir.realPath(io, &path_buffer);
    var archive = try archive_mod.Archive.init(testing.allocator, io, path_buffer[0..path_len]);
    defer archive.deinit();
    _ = try archive.append(.{
        .seq = 1,
        .witnessed_at = 1,
        .indexed_at = 0,
        .kind = .create,
        .did = "did:plc:leakregression",
        .collection = "app.bsky.feed.post",
        .rkey = "3l3qo2vuowo2b",
        .rev = "3l3qo2vutsw2b",
        .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69",
    }, 1);
    try archive.rotate();

    var xrpc_api: xrpcapi.Api = .{ .allocator = testing.allocator, .io = io, .archive = &archive };
    var hub: Hub = .{
        .allocator = testing.allocator,
        .io = io,
        .tail = &tail,
        .stats = &stats,
        .archive = &archive,
        .xrpc = &xrpc_api,
    };
    defer hub.deinit();

    var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer listener.deinit(io);
    const port = listener.socket.address.getPort();
    var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{
        .port = port,
        .address = "127.0.0.1",
        .max_conn = 4,
        .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 },
    });
    defer ws_server.deinit();
    const Fixture = struct {
        fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, fixture_hub: *Hub) !void {
            server.runIo(server_listener, fixture_hub);
        }
    };
    var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &hub });
    defer _ = future.cancel(io) catch {};

    var url_buf: [256]u8 = undefined;
    const url = try std.fmt.bufPrint(
        &url_buf,
        "http://127.0.0.1:{d}/xrpc/network.bsky.jetstream.getBlock?segment=seg_0000000000.jss&blockIndex=0",
        .{port},
    );
    // fresh connection per request — the shape prod leaked under
    for (0..8) |_| {
        var transport = zat.HttpTransport.init(io, testing.allocator);
        transport.keep_alive = false;
        defer transport.deinit();
        var response = try transport.fetch(.{ .url = url, .max_response_size = 1 << 20 });
        defer response.deinit(testing.allocator);
        try testing.expectEqual(std.http.Status.ok, response.status);
        try testing.expect(response.body.len > 0);
    }
    // reused connection — prod leaked ~200 B/request here too
    var transport = zat.HttpTransport.init(io, testing.allocator);
    defer transport.deinit();
    for (0..8) |_| {
        var response = try transport.fetch(.{ .url = url, .max_response_size = 1 << 20 });
        defer response.deinit(testing.allocator);
        try testing.expectEqual(std.http.Status.ok, response.status);
    }
}

test "obsolete live dictionary returns the upstream JSON rejection" {
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20);
    defer tail.deinit();
    var stats: metrics.Stats = .{};
    var hub: Hub = .{
        .allocator = testing.allocator,
        .io = io,
        .tail = &tail,
        .stats = &stats,
        .serving = .init(true),
    };
    defer hub.deinit();

    var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable;
    defer listener.deinit(io);
    const port = listener.socket.address.getPort();
    var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{
        .port = port,
        .address = "127.0.0.1",
        .max_conn = 4,
        .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 },
    });
    defer ws_server.deinit();
    const Fixture = struct {
        fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, fixture_hub: *Hub) !void {
            server.runIo(server_listener, fixture_hub);
        }
    };
    var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &hub });
    defer _ = future.cancel(io) catch {};

    const address: Io.net.IpAddress = .{ .ip4 = .loopback(port) };
    var socket = try address.connect(io, .{ .mode = .stream });
    defer socket.close(io);
    var write_buf: [1024]u8 = undefined;
    var writer = socket.writer(io, &write_buf);
    try writer.interface.print("GET /xrpc/network.bsky.jetstream.subscribeEvents?zstdDictionary={d} HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Protocol: xrpc.v1.json\r\n\r\n", .{zstd.v2_dictionary_id ^ 1});
    try writer.interface.flush();
    var read_buf: [4096]u8 = undefined;
    var reader = socket.reader(io, &read_buf);
    const response = try reader.interface.allocRemaining(testing.allocator, .limited(4096));
    defer testing.allocator.free(response);
    try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 400 "));
    const split = std.mem.indexOf(u8, response, "\r\n\r\n") orelse return error.MissingHeaders;
    try testing.expect(std.mem.indexOf(u8, response[0..split], "application/json") != null);
    const parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, response[split + 4 ..], .{});
    defer parsed.deinit();
    try testing.expectEqualStrings("UnknownZstdDictionary", parsed.value.object.get("error").?.string);
    try testing.expect(std.mem.indexOf(u8, parsed.value.object.get("message").?.string, "reconnect") != null);
}