//! relay broadcaster — fan-out firehose frames to downstream consumers
//!
//! features:
//!   - ref-counted shared frames (one copy per broadcast, not per consumer)
//!   - ConsumerTooSlow CBOR error frame before disconnect
//!   - ping/pong keepalive (30s idle, 5s deadline)
//!   - non-blocking broadcast via per-consumer ring buffers + write threads
//!   - health/stats HTTP endpoints via websocket handshake routing

const std = @import("std");
const Io = std.Io;
const websocket = @import("websocket");
const zat = @import("zat");
const ring_buffer = @import("util/ring_buffer.zig");
const event_log_mod = @import("event_log.zig");

const builtin = @import("builtin");
const build_options = @import("build_options");
const util = @import("util/util.zig");
const Allocator = std.mem.Allocator;
const log = std.log.scoped(.relay);

const timestamp = util.timestamp;

// --- stats ---

pub const Stats = struct {
    seq: std.atomic.Value(u64) = .{ .raw = 0 },
    relay_seq: std.atomic.Value(u64) = .{ .raw = 0 },
    consumer_count: std.atomic.Value(u64) = .{ .raw = 0 },
    frames_in: std.atomic.Value(u64) = .{ .raw = 0 },
    frames_out: std.atomic.Value(u64) = .{ .raw = 0 },
    validated: std.atomic.Value(u64) = .{ .raw = 0 },
    failed: std.atomic.Value(u64) = .{ .raw = 0 },
    // per-reason failure breakdown (subsets of `failed`)
    failed_bad_did: std.atomic.Value(u64) = .{ .raw = 0 },
    failed_bad_rev: std.atomic.Value(u64) = .{ .raw = 0 },
    failed_missing_blocks: std.atomic.Value(u64) = .{ .raw = 0 },
    failed_oversized_blocks: std.atomic.Value(u64) = .{ .raw = 0 },
    failed_bad_structure: std.atomic.Value(u64) = .{ .raw = 0 },
    failed_signature: std.atomic.Value(u64) = .{ .raw = 0 },
    failed_commit_integrity: std.atomic.Value(u64) = .{ .raw = 0 },
    // Proof failures split by cause, because they mean opposite things: a
    // sender that omitted prevData is behaving as expected for a pre-Sync-1.1
    // PDS, while a proof that fails to invert means the commit or our
    // verification is wrong. Counting both as sync_1_1 hid a zat MST bug that
    // was dropping valid commits at 5% of validated traffic.
    failed_sync_1_1: std.atomic.Value(u64) = .{ .raw = 0 },
    failed_prev_data_missing: std.atomic.Value(u64) = .{ .raw = 0 },
    failed_host_authority: std.atomic.Value(u64) = .{ .raw = 0 },
    failed_future_rev: std.atomic.Value(u64) = .{ .raw = 0 },
    skipped: std.atomic.Value(u64) = .{ .raw = 0 },
    rate_limited: std.atomic.Value(u64) = .{ .raw = 0 },
    rate_limited_sec: std.atomic.Value(u64) = .{ .raw = 0 },
    rate_limited_hour: std.atomic.Value(u64) = .{ .raw = 0 },
    rate_limited_day: std.atomic.Value(u64) = .{ .raw = 0 },
    decode_errors: std.atomic.Value(u64) = .{ .raw = 0 },
    cache_hits: std.atomic.Value(u64) = .{ .raw = 0 },
    cache_misses: std.atomic.Value(u64) = .{ .raw = 0 },
    slow_consumers: std.atomic.Value(u64) = .{ .raw = 0 },
    connected_inbound: std.atomic.Value(u64) = .{ .raw = 0 },
    cache_evictions: std.atomic.Value(u64) = .{ .raw = 0 },
    // chain continuity failures by reason (since vs stored rev, prevData vs stored data CID)
    chain_breaks_since: std.atomic.Value(u64) = .{ .raw = 0 },
    chain_breaks_prev_data: std.atomic.Value(u64) = .{ .raw = 0 },
    pool_backpressure: std.atomic.Value(u64) = .{ .raw = 0 },
    // keepalive pinger (ws_ping.zig) — pings sent, connections torn down for
    // unanswered pings, and ping writes that themselves failed
    ws_pings_sent: std.atomic.Value(u64) = .{ .raw = 0 },
    ws_ping_timeout_closes: std.atomic.Value(u64) = .{ .raw = 0 },
    ws_ping_write_failures: std.atomic.Value(u64) = .{ .raw = 0 },
    // host authority resolution metrics
    host_authority_checks: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_is_new: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_host_changed: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_time_us: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_cache_hits: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_network_checks: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_network_time_us: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_pool_wait_us: std.atomic.Value(u64) = .{ .raw = 0 },
    // per-branch reject breakdown (subsets of failed_host_authority).
    // lets us tell whether rejects come from stale cached forgeries
    // (host_mismatch), genuinely forged DIDs (resolve/no_endpoint),
    // or a mismatch between DID doc endpoint and our host table.
    host_authority_reject_parse_did: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_reject_resolve: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_reject_no_endpoint: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_reject_bad_url: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_reject_unknown_host: std.atomic.Value(u64) = .{ .raw = 0 },
    host_authority_reject_host_mismatch: std.atomic.Value(u64) = .{ .raw = 0 },
    // subscriber reconnect-cycle disconnect breakdown. only handshake-or-earlier
    // failures accumulate to failed_attempts (reset_failures fires after
    // ws_handshake succeeds, before read_loop). 15 consecutive of these →
    // host moves to 'exhausted'. read_loop disconnects do not directly
    // exhaust unless the next handshake also fails.
    subscriber_disconnect_dns_connect: std.atomic.Value(u64) = .{ .raw = 0 },
    subscriber_disconnect_tls_init: std.atomic.Value(u64) = .{ .raw = 0 },
    subscriber_disconnect_ws_handshake: std.atomic.Value(u64) = .{ .raw = 0 },
    subscriber_disconnect_read_loop: std.atomic.Value(u64) = .{ .raw = 0 },
    // frame pool memory pressure
    pool_queued_bytes: std.atomic.Value(u64) = .{ .raw = 0 },
    // persist/broadcast pipeline contention
    persist_order_spins: std.atomic.Value(u64) = .{ .raw = 0 },
    broadcast_queue_push_lock_spins: std.atomic.Value(u64) = .{ .raw = 0 },
    broadcast_queue_full: std.atomic.Value(u64) = .{ .raw = 0 },
    broadcast_queue_depth_hwm: std.atomic.Value(u32) = .{ .raw = 0 },
    broadcast_no_consumers: std.atomic.Value(u64) = .{ .raw = 0 },
    start_time: i64 = 0,
};

// --- shared frame (ref-counted) ---

pub const SharedFrame = struct {
    data: []const u8,
    ref_count: std.atomic.Value(u32),
    allocator: Allocator,

    pub fn create(allocator: Allocator, data: []const u8) !*SharedFrame {
        const frame = try allocator.create(SharedFrame);
        const duped = try allocator.dupe(u8, data);
        frame.* = .{
            .data = duped,
            .ref_count = .{ .raw = 1 },
            .allocator = allocator,
        };
        return frame;
    }

    pub fn acquire(self: *SharedFrame) void {
        _ = self.ref_count.fetchAdd(1, .monotonic);
    }

    pub fn release(self: *SharedFrame) void {
        if (self.ref_count.fetchSub(1, .acq_rel) == 1) {
            self.allocator.free(self.data);
            self.allocator.destroy(self);
        }
    }
};

// --- ConsumerTooSlow error frame ---

/// build the ConsumerTooSlow CBOR error frame using the SDK encoder.
/// header: {op: -1}, payload: {error: "ConsumerTooSlow", message: "..."}
fn buildErrorFrame(allocator: Allocator) ?[]const u8 {
    const cbor = zat.cbor;

    // header: {op: -1} — spec says no `t` field on error frames
    const header: cbor.Value = .{ .map = &.{
        .{ .key = "op", .value = .{ .negative = -1 } },
    } };

    // payload: {error: "ConsumerTooSlow", message: "consumer buffer full"}
    const payload: cbor.Value = .{ .map = &.{
        .{ .key = "error", .value = .{ .text = "ConsumerTooSlow" } },
        .{ .key = "message", .value = .{ .text = "consumer buffer full" } },
    } };

    const header_bytes = cbor.encodeAlloc(allocator, header) catch return null;
    const payload_bytes = cbor.encodeAlloc(allocator, payload) catch {
        allocator.free(header_bytes);
        return null;
    };

    var frame = allocator.alloc(u8, header_bytes.len + payload_bytes.len) catch {
        allocator.free(header_bytes);
        allocator.free(payload_bytes);
        return null;
    };
    @memcpy(frame[0..header_bytes.len], header_bytes);
    @memcpy(frame[header_bytes.len..], payload_bytes);

    allocator.free(header_bytes);
    allocator.free(payload_bytes);

    return frame;
}

/// encode an error frame into caller-provided buffer.
/// used for InvalidCursor / FutureCursor error frames.
/// header: {op: -1}, payload: {error: <name>, message: <msg>}
fn encodeErrorFrame(error_name: []const u8, message: []const u8, out: []u8) ?[]const u8 {
    const cbor = zat.cbor;
    const alloc = std.heap.c_allocator;

    const header: cbor.Value = .{ .map = &.{
        .{ .key = "op", .value = .{ .negative = -1 } },
    } };
    const payload: cbor.Value = .{ .map = &.{
        .{ .key = "error", .value = .{ .text = error_name } },
        .{ .key = "message", .value = .{ .text = message } },
    } };

    const header_bytes = cbor.encodeAlloc(alloc, header) catch return null;
    defer alloc.free(header_bytes);
    const payload_bytes = cbor.encodeAlloc(alloc, payload) catch return null;
    defer alloc.free(payload_bytes);

    const total = header_bytes.len + payload_bytes.len;
    if (total > out.len) return null;
    @memcpy(out[0..header_bytes.len], header_bytes);
    @memcpy(out[header_bytes.len..][0..payload_bytes.len], payload_bytes);

    return out[0..total];
}

/// encode an #info message frame into caller-provided buffer.
/// used for OutdatedCursor. this is a message frame (op: 1), not an error frame.
/// header: {op: 1, t: "#info"}, payload: {name: <name>, message: <msg>}
fn encodeInfoMessage(name: []const u8, message: []const u8, out: []u8) ?[]const u8 {
    const cbor = zat.cbor;
    const alloc = std.heap.c_allocator;

    const header: cbor.Value = .{ .map = &.{
        .{ .key = "op", .value = .{ .unsigned = 1 } },
        .{ .key = "t", .value = .{ .text = "#info" } },
    } };
    const payload: cbor.Value = .{ .map = &.{
        .{ .key = "name", .value = .{ .text = name } },
        .{ .key = "message", .value = .{ .text = message } },
    } };

    const header_bytes = cbor.encodeAlloc(alloc, header) catch return null;
    defer alloc.free(header_bytes);
    const payload_bytes = cbor.encodeAlloc(alloc, payload) catch return null;
    defer alloc.free(payload_bytes);

    const total = header_bytes.len + payload_bytes.len;
    if (total > out.len) return null;
    @memcpy(out[0..header_bytes.len], header_bytes);
    @memcpy(out[header_bytes.len..][0..payload_bytes.len], payload_bytes);

    return out[0..total];
}

// --- frame re-sequencing ---

/// resequence a firehose frame: replace the payload's "seq" with relay_seq.
/// header bytes are preserved as-is, only the payload is re-encoded.
/// returns null on decode failure; caller owns returned memory via allocator.
pub fn resequenceFrame(allocator: Allocator, data: []const u8, relay_seq: u64) ?[]u8 {
    const cbor = zat.cbor;

    // split frame: [header CBOR] [payload CBOR]
    const header_result = cbor.decode(allocator, data) catch return null;
    const header_end = header_result.consumed;
    const payload_data = data[header_end..];

    // check header type — commit frames need tooBig normalization
    const is_commit = blk: {
        const header_val = switch (header_result.value) {
            .map => |entries| entries,
            else => break :blk false,
        };
        for (header_val) |entry| {
            if (std.mem.eql(u8, entry.key, "t")) {
                const t = switch (entry.value) {
                    .text => |s| s,
                    else => break :blk false,
                };
                break :blk std.mem.eql(u8, t, "#commit");
            }
        }
        break :blk false;
    };

    // decode payload map
    const payload = cbor.decodeAll(allocator, payload_data) catch return null;
    const old_entries = switch (payload) {
        .map => |entries| entries,
        else => return null,
    };

    // rebuild map entries with seq replaced
    var new_entries: std.ArrayListUnmanaged(cbor.Value.MapEntry) = .empty;
    var found_seq = false;
    var found_too_big = false;
    for (old_entries) |entry| {
        if (std.mem.eql(u8, entry.key, "seq")) {
            new_entries.append(allocator, .{ .key = "seq", .value = .{ .unsigned = relay_seq } }) catch return null;
            found_seq = true;
        } else {
            if (std.mem.eql(u8, entry.key, "tooBig")) found_too_big = true;
            new_entries.append(allocator, entry) catch return null;
        }
    }
    if (!found_seq) {
        new_entries.append(allocator, .{ .key = "seq", .value = .{ .unsigned = relay_seq } }) catch return null;
    }
    // lexicon requires tooBig on #commit — some PDSes omit it, hydrant et al. reject the frame
    if (is_commit and !found_too_big) {
        new_entries.append(allocator, .{ .key = "tooBig", .value = .{ .boolean = false } }) catch return null;
    }

    const new_payload: cbor.Value = .{ .map = new_entries.items };
    const new_payload_bytes = cbor.encodeAlloc(allocator, new_payload) catch return null;

    // combine: original header bytes + resequenced payload
    var result = allocator.alloc(u8, header_end + new_payload_bytes.len) catch return null;
    @memcpy(result[0..header_end], data[0..header_end]);
    @memcpy(result[header_end..], new_payload_bytes);

    return result;
}

// --- broadcast queue (worker → fiber handoff) ---

/// item produced by frame workers, consumed by the broadcaster fiber.
/// data is heap-allocated by the worker; the broadcaster fiber frees it after broadcast.
pub const BroadcastItem = struct {
    seq: u64,
    data: []const u8,
};

/// MPSC ring buffer: multiple frame worker threads push, one broadcaster fiber pops.
/// uses an atomic spinlock for push (no Io.Mutex — works from any execution context).
/// pop is single-consumer (broadcaster fiber only), no locking needed.
pub const BroadcastQueue = struct {
    const CAPACITY = 8192;

    items: [CAPACITY]BroadcastItem = undefined,
    head: std.atomic.Value(u32) = .{ .raw = 0 },
    tail: std.atomic.Value(u32) = .{ .raw = 0 },
    push_lock: std.atomic.Value(u32) = .{ .raw = 0 },
    allocator: Allocator,

    pub fn init(allocator: Allocator) BroadcastQueue {
        return .{ .allocator = allocator };
    }

    /// approximate queue depth (may be slightly stale — for metrics only)
    pub fn depth(self: *const BroadcastQueue) u32 {
        const tail = self.tail.load(.monotonic);
        const head = self.head.load(.monotonic);
        return if (tail >= head) tail - head else CAPACITY - head + tail;
    }

    /// push an item (called by worker threads). spins until space is available.
    /// matches Indigo semantics: every persisted event reaches live broadcast.
    /// the broadcaster fiber drains at memory speed (no I/O), so spin is brief.
    pub fn push(self: *BroadcastQueue, seq: u64, data: []const u8, stats: *Stats) void {
        var full_spins: u32 = 0;
        while (true) {
            // acquire spinlock
            var lock_spins: u32 = 0;
            while (self.push_lock.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
                lock_spins += 1;
                std.atomic.spinLoopHint();
            }
            if (lock_spins > 0) {
                _ = stats.broadcast_queue_push_lock_spins.fetchAdd(lock_spins, .monotonic);
            }

            const tail = self.tail.load(.monotonic);
            const next_tail = (tail + 1) % CAPACITY;
            if (next_tail == self.head.load(.acquire)) {
                // full — release lock, yield, retry
                self.push_lock.store(0, .release);
                // saturate: this is a diagnostic counter, and the queue staying
                // full is exactly when it must not take the process down. a
                // stalled drain spun this past 2^32 and panicked the worker
                // under ReleaseSafe, turning "consumers are slow" into a crash.
                full_spins +|= 1;
                std.atomic.spinLoopHint();
                continue;
            }

            self.items[tail] = .{ .seq = seq, .data = data };
            self.tail.store(next_tail, .release);
            self.push_lock.store(0, .release);

            if (full_spins > 0) {
                _ = stats.broadcast_queue_full.fetchAdd(full_spins, .monotonic);
            }
            // update high-water mark (racy but directionally correct)
            const d = self.depth();
            const hwm = stats.broadcast_queue_depth_hwm.load(.monotonic);
            if (d > hwm) {
                _ = stats.broadcast_queue_depth_hwm.cmpxchgWeak(hwm, d, .monotonic, .monotonic);
            }
            return;
        }
    }

    /// peek the seq at the head without removing it (broadcaster fiber only).
    /// used to gate emission on the durable watermark. null if empty.
    pub fn peekSeq(self: *BroadcastQueue) ?u64 {
        const head = self.head.load(.monotonic);
        if (head == self.tail.load(.acquire)) return null; // empty
        return self.items[head].seq;
    }

    /// pop an item (called by broadcaster fiber only). returns null if empty.
    pub fn pop(self: *BroadcastQueue) ?BroadcastItem {
        const head = self.head.load(.monotonic);
        if (head == self.tail.load(.acquire)) return null; // empty

        const item = self.items[head];
        self.head.store((head + 1) % CAPACITY, .release);
        return item;
    }
};

// --- consumer ---

const ping_interval_ns: u64 = 30 * std.time.ns_per_s;
const ping_timeout_ns: u64 = 5 * std.time.ns_per_s;

pub const Consumer = struct {
    const BUFFER_CAP = 65536;

    conn: *websocket.Conn,
    allocator: Allocator,
    buf: [BUFFER_CAP]*SharedFrame = undefined,
    write_pos: usize = 0,
    read_pos: usize = 0,
    buf_len: usize = 0,
    alive: std.atomic.Value(bool) = .{ .raw = true },
    mutex: Io.Mutex = Io.Mutex.init,
    cond: Io.Condition = Io.Condition.init,
    future: ?Io.Future(void) = null,
    last_send_time: i128 = 0,
    io: Io,

    /// push a shared frame to this consumer's send buffer.
    /// acquires a reference. returns false if full (consumer too slow) or
    /// dead — a disconnected consumer must not keep accumulating frames
    /// (replay for a flapping consumer compounded into an OOM otherwise).
    pub fn enqueue(self: *Consumer, frame: *SharedFrame) bool {
        self.mutex.lockUncancelable(self.io);
        defer self.mutex.unlock(self.io);

        if (!self.alive.load(.acquire)) return false;
        if (self.buf_len == BUFFER_CAP) return false;

        frame.acquire();
        self.buf[self.write_pos] = frame;
        self.write_pos = (self.write_pos + 1) % BUFFER_CAP;
        self.buf_len += 1;
        self.cond.signal(self.io);
        return true;
    }

    /// push raw bytes (for replay, error frames). allocates a new SharedFrame.
    pub fn enqueueRaw(self: *Consumer, data: []const u8) bool {
        const frame = SharedFrame.create(self.allocator, data) catch return false;
        // enqueue will acquire, so we start with refcount 1 (for our ownership)
        // then release our ownership after enqueue
        if (self.enqueue(frame)) {
            frame.release(); // drop our ref, consumer write loop holds one
            return true;
        }
        frame.release();
        return false;
    }

    fn dequeue(self: *Consumer) ?*SharedFrame {
        // caller must hold mutex
        if (self.buf_len == 0) return null;
        const frame = self.buf[self.read_pos];
        self.read_pos = (self.read_pos + 1) % BUFFER_CAP;
        self.buf_len -= 1;
        return frame;
    }

    fn writeLoop(self: *Consumer) void {
        self.last_send_time = Io.Timestamp.now(self.io, .real).nanoseconds;

        while (self.alive.load(.acquire)) {
            var frame: ?*SharedFrame = null;
            {
                self.mutex.lockUncancelable(self.io);
                defer self.mutex.unlock(self.io);
                if (self.buf_len == 0) {
                    // no data: poll briefly instead of blocking on cond
                    // (Io.Condition has no timedWait, so we poll to allow periodic ping checks)
                    self.mutex.unlock(self.io);
                    self.io.sleep(Io.Duration.fromMilliseconds(100), .awake) catch {};
                    self.mutex.lockUncancelable(self.io);
                }
                frame = self.dequeue();
            }
            if (frame) |f| {
                defer f.release();
                self.conn.writeBin(f.data) catch {
                    self.alive.store(false, .release);
                    break;
                };
                self.last_send_time = Io.Timestamp.now(self.io, .real).nanoseconds;
            } else {
                // no data — check if we should send a ping
                self.maybePing();
            }
        }
        // drain remaining buffered frames
        while (true) {
            self.mutex.lockUncancelable(self.io);
            const frame = self.dequeue();
            self.mutex.unlock(self.io);
            if (frame) |f| {
                f.release();
            } else break;
        }
        // close the connection from the writeLoop thread to unblock readLoop.
        // this MUST happen here, not from the broadcast thread — closing from
        // broadcast races with readLoop's setsockopt/read calls on the socket,
        // causing EBADF → unreachable panic under ReleaseSafe.
        self.conn.close(.{}) catch {};
    }

    fn maybePing(self: *Consumer) void {
        const now = Io.Timestamp.now(self.io, .real).nanoseconds;
        const elapsed: u64 = @intCast(@max(0, now - self.last_send_time));
        if (elapsed >= ping_interval_ns) {
            // send ping
            self.conn.writePing(&.{}) catch {
                self.alive.store(false, .release);
                return;
            };
            self.last_send_time = now;
        }
    }

    pub fn shutdown(self: *Consumer) void {
        self.alive.store(false, .release);
        self.cond.signal(self.io);
        if (self.future) |*f| f.cancel(self.io);
        self.future = null;
    }
};

// --- broadcaster ---

/// per-chunk bound for disk replay: caps the transient allocation of one
/// consumer connect at ~16 MiB regardless of how much backlog its cursor
/// spans. worst-case per-frame size (5 MiB max_message_size) is still
/// covered: max_bytes is checked before each read, so a chunk holds at most
/// max_bytes + one frame.
const replay_chunk_limits: event_log_mod.PlaybackLimits = .{
    .max_entries = 4096,
    .max_bytes = 16 * 1024 * 1024,
};

const history_capacity = 50_000;
const FrameHistory = ring_buffer.RingBuffer(history_capacity);

/// function pointer for HTTP fallback handler (set by main.zig)
pub const HttpFallbackFn = *const fn (
    *websocket.Conn,
    []const u8,
    []const u8,
    []const u8,
    *const websocket.Handshake.KeyValue,
    ?*anyopaque,
) void;

pub const Broadcaster = struct {
    allocator: Allocator,
    consumers: std.ArrayListUnmanaged(*Consumer) = .empty,
    consumers_mutex: Io.Mutex = Io.Mutex.init,
    /// ordering spinlock for persist → queue push atomicity.
    /// atomic spinlock (not Io.Mutex) so both Threaded workers and Evented admin
    /// can participate in the same ordered publication path.
    persist_order: std.atomic.Value(u32) = .{ .raw = 0 },
    /// worker → fiber handoff: frame workers push here, broadcaster fiber pops.
    broadcast_queue: BroadcastQueue,
    history: FrameHistory,
    persist: ?*event_log_mod.DiskPersist = null,
    db_queue: ?*event_log_mod.DbRequestQueue = null,
    stats: Stats = .{},
    error_frame: ?[]const u8 = null,
    http_fallback: ?HttpFallbackFn = null,
    http_fallback_ctx: ?*anyopaque = null,
    io: Io,
    shutdown: *std.atomic.Value(bool),
    producers_done: std.atomic.Value(bool) = .{ .raw = false },

    pub fn init(allocator: Allocator, io: Io, shutdown: *std.atomic.Value(bool)) Broadcaster {
        return .{
            .allocator = allocator,
            .broadcast_queue = BroadcastQueue.init(allocator),
            .history = FrameHistory.init(allocator, io),
            .stats = .{ .start_time = timestamp(io) },
            .error_frame = buildErrorFrame(allocator),
            .io = io,
            .shutdown = shutdown,
        };
    }

    pub fn deinit(self: *Broadcaster) void {
        self.consumers_mutex.lockUncancelable(self.io);
        defer self.consumers_mutex.unlock(self.io);
        for (self.consumers.items) |consumer| {
            consumer.shutdown();
            self.allocator.destroy(consumer);
        }
        self.consumers.deinit(self.allocator);
        self.history.deinit();
        if (self.error_frame) |ef| self.allocator.free(ef);
        while (self.broadcast_queue.pop()) |item| self.allocator.free(@constCast(item.data));
    }

    pub fn addConsumer(self: *Broadcaster, conn: *websocket.Conn) !*Consumer {
        const consumer = try self.allocator.create(Consumer);
        consumer.* = .{
            .conn = conn,
            .allocator = self.allocator,
            .io = self.io,
        };
        consumer.future = self.io.concurrent(Consumer.writeLoop, .{consumer}) catch {
            self.allocator.destroy(consumer);
            return error.ThreadSpawnFailed;
        };

        self.consumers_mutex.lockUncancelable(self.io);
        defer self.consumers_mutex.unlock(self.io);
        self.consumers.append(self.allocator, consumer) catch {
            consumer.shutdown();
            self.allocator.destroy(consumer);
            return error.OutOfMemory;
        };
        _ = self.stats.consumer_count.fetchAdd(1, .monotonic);
        log.info("consumer connected ({d} total)", .{self.consumers.items.len});
        return consumer;
    }

    pub fn removeConsumer(self: *Broadcaster, consumer: *Consumer) void {
        {
            self.consumers_mutex.lockUncancelable(self.io);
            defer self.consumers_mutex.unlock(self.io);
            for (self.consumers.items, 0..) |c, i| {
                if (c == consumer) {
                    _ = self.consumers.swapRemove(i);
                    // only decrement if we found it — dropSlowConsumer already
                    // decremented for slow consumers removed during broadcast()
                    _ = self.stats.consumer_count.fetchSub(1, .monotonic);
                    break;
                }
            }
        }
        consumer.shutdown();
        log.info("consumer disconnected ({d} remaining)", .{self.consumers.items.len});
        self.allocator.destroy(consumer);
    }

    /// broadcast a frame to all consumers. non-blocking — just enqueues.
    /// drops slow consumers whose buffers are full after sending ConsumerTooSlow.
    pub fn broadcast(self: *Broadcaster, seq: u64, data: []const u8) void {
        self.stats.seq.store(seq, .release);

        // add to history for cursor replay
        _ = self.history.push(seq, data);

        // fast path: skip SharedFrame allocation when nobody is listening
        if (self.stats.consumer_count.load(.acquire) == 0) {
            _ = self.stats.broadcast_no_consumers.fetchAdd(1, .monotonic);
            return;
        }

        // create one shared frame for all consumers
        const frame = SharedFrame.create(self.allocator, data) catch return;
        defer frame.release(); // release broadcaster's reference

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

        var frames_sent: u64 = 0;
        var i: usize = 0;
        while (i < self.consumers.items.len) {
            const consumer = self.consumers.items[i];
            if (!consumer.alive.load(.acquire)) {
                // remove from broadcast list but don't destroy —
                // Handler.close() → removeConsumer() owns cleanup
                _ = self.consumers.swapRemove(i);
                _ = self.stats.consumer_count.fetchSub(1, .monotonic);
                continue;
            }
            if (consumer.enqueue(frame)) {
                frames_sent += 1;
            } else {
                // buffer full — send ConsumerTooSlow error, then drop
                self.dropSlowConsumer(consumer);
                _ = self.consumers.swapRemove(i);
                _ = self.stats.consumer_count.fetchSub(1, .monotonic);
                _ = self.stats.slow_consumers.fetchAdd(1, .monotonic);
                continue;
            }
            i += 1;
        }
        _ = self.stats.frames_out.fetchAdd(frames_sent, .monotonic);
    }

    fn dropSlowConsumer(self: *Broadcaster, consumer: *Consumer) void {
        log.warn("dropping slow consumer (ConsumerTooSlow)", .{});
        // try to send error frame directly (bypass buffer since it's full)
        if (self.error_frame) |ef| {
            consumer.conn.writeBin(ef) catch {};
        }
        // signal consumer to stop — writeLoop checks alive and exits, then
        // Handler.close → removeConsumer handles shutdown + destroy.
        // do NOT close the socket here: readLoop (websocket server thread) may
        // be calling setsockopt/read on it concurrently. closing from a third
        // thread causes EBADF → unreachable panic in posix.setsockopt under
        // ReleaseSafe. let writeLoop's exit trigger the natural close sequence.
        consumer.alive.store(false, .release);
        consumer.cond.signal(consumer.io);
        // consumer is removed from list by broadcast() caller (swapRemove).
        // cleanup deferred to removeConsumer to avoid double-destroy.
    }

    /// two-phase cursor replay: disk (diskpersist) first, then in-memory ring buffer.
    /// the consumer is already in the live broadcast list, so frames arriving
    /// during replay are buffered — no gap possible.
    ///
    /// disk playback runs on pool_io (Threaded) via request/reply — playback()
    /// holds the DiskPersist mutex and reads files, all of which require Threaded Io.
    pub fn replayTo(self: *Broadcaster, consumer: *Consumer, cursor: u64) void {
        // phase 1: disk replay from diskpersist via request/reply, in bounded
        // chunks. materializing the whole backlog per connect is an
        // O(retention) allocation — a flapping consumer stacking those
        // replays OOMed the canary (2026-08-06). each chunk is capped, the
        // consumer's liveness is re-checked between chunks, and a dead or
        // slow consumer aborts the replay entirely.
        if (self.persist) |dp| {
            var since = cursor;
            var replayed: usize = 0;
            while (consumer.alive.load(.acquire)) {
                var req: event_log_mod.PlaybackRequest = .{
                    .since = since,
                    .allocator = self.allocator,
                    .limits = replay_chunk_limits,
                };
                dp.enqueuePlayback(&req);

                // poll until pool_io worker completes the request.
                // SAFETY: req is stack-local — we MUST wait for the worker to finish before
                // returning, otherwise the stack frame unwinds while the worker still holds &req.
                // if sleep fails (shutdown/io error), fall back to spin-wait. the host_ops worker
                // drains all pending playback requests before exiting, so this is bounded.
                while (!req.done.load(.acquire)) {
                    self.io.sleep(Io.Duration.fromMicroseconds(100), .awake) catch {
                        while (!req.done.load(.acquire)) std.atomic.spinLoopHint();
                        break;
                    };
                }

                defer {
                    for (req.entries.items) |e| self.allocator.free(e.data);
                    req.entries.deinit(self.allocator);
                }

                if (req.err) |err| {
                    log.warn("disk replay failed: {s}, falling back to memory", .{@errorName(err)});
                    self.replayFromMemory(consumer, since);
                    return;
                }

                if (req.entries.items.len == 0) break; // caught up

                var reseq_arena = std.heap.ArenaAllocator.init(self.allocator);
                defer reseq_arena.deinit();

                for (req.entries.items) |entry| {
                    // resequence: replace upstream seq in CBOR with relay-assigned seq
                    const frame_data = resequenceFrame(reseq_arena.allocator(), entry.data, entry.seq) orelse entry.data;
                    if (!consumer.enqueueRaw(frame_data)) {
                        log.warn("replay aborted after {d} frames (consumer dead or buffer full, cursor={d})", .{ replayed, cursor });
                        return;
                    }
                    replayed += 1;
                    since = entry.seq;
                }
            }
            if (replayed > 0) {
                log.info("replayed {d} frames from disk (cursor={d})", .{ replayed, cursor });
            }
            return;
        }

        // phase 2 fallback: in-memory ring buffer (no persist configured)
        self.replayFromMemory(consumer, cursor);
    }

    fn replayFromMemory(self: *Broadcaster, consumer: *Consumer, cursor: u64) void {
        self.history.mutex.lockUncancelable(self.io);
        defer self.history.mutex.unlock(self.io);

        var i: usize = 0;
        while (i < self.history.len) : (i += 1) {
            const idx = (self.history.read_pos + i) % history_capacity;
            const entry = self.history.entries[idx];
            if (entry.seq > cursor) {
                if (!consumer.enqueueRaw(entry.data)) break;
            }
        }
    }

    pub fn consumerCount(self: *Broadcaster) usize {
        self.consumers_mutex.lockUncancelable(self.io);
        defer self.consumers_mutex.unlock(self.io);
        return self.consumers.items.len;
    }

    /// broadcast loop — drains the broadcast queue, fans out to consumers
    /// and calls broadcast() for each item. this is the only path that touches
    /// consumers_mutex / consumer.mutex / consumer.cond.
    pub fn runBroadcastLoop(self: *Broadcaster) void {
        log.info("broadcaster fiber started", .{});
        while (true) {
            const finished = self.producers_done.load(.acquire);
            // only emit frames that are durably on disk: hold any whose seq is
            // past the persister's committed watermark until the next flush
            // advances it. without a persister (degenerate no-persist mode),
            // nothing is gated. see docs/incident-2026-05-31-seq-rewind.md.
            const committed: u64 = if (self.persist) |dp|
                dp.committed_seq.load(.acquire)
            else
                std.math.maxInt(u64);

            var drained: usize = 0;
            while (self.broadcast_queue.peekSeq()) |seq| {
                if (seq > committed) break; // not durable yet — hold
                const item = self.broadcast_queue.pop() orelse break;
                self.broadcast(item.seq, item.data);
                self.allocator.free(@constCast(item.data));
                drained += 1;
            }
            if (finished) break;
            if (drained == 0) {
                // nothing emittable — yield briefly so other fibers run
                self.io.sleep(Io.Duration.fromMilliseconds(1), .awake) catch return;
            }
        }
        while (self.broadcast_queue.pop()) |item| {
            self.allocator.free(@constCast(item.data));
        }
        log.info("broadcaster fiber stopped", .{});
    }
};

// --- websocket handler ---

pub const Handler = struct {
    consumer: ?*Consumer,
    broadcaster: *Broadcaster,
    conn: *websocket.Conn,
    cursor: ?u64,
    invalid_cursor: bool = false,

    /// called during handshake — validate path, parse cursor.
    /// do NOT start the write thread here (handshake reply hasn't been sent yet).
    pub fn init(handshake: *const websocket.Handshake, conn: *websocket.Conn, ctx: *Broadcaster) !Handler {
        const url = handshake.url;
        const path = if (std.mem.indexOfScalar(u8, url, '?')) |qi| url[0..qi] else url;

        if (!std.mem.eql(u8, path, "/xrpc/com.atproto.sync.subscribeRepos")) {
            return error.NotFound;
        }

        // parse cursor (deferred until afterInit when connection is fully upgraded)
        var cursor: ?u64 = null;
        var invalid_cursor = false;
        if (std.mem.indexOf(u8, url, "cursor=")) |cursor_start| {
            const value_start = cursor_start + "cursor=".len;
            const value_end = std.mem.indexOfScalarPos(u8, url, value_start, '&') orelse url.len;
            const cursor_str = url[value_start..value_end];
            cursor = std.fmt.parseInt(u64, cursor_str, 10) catch blk: {
                invalid_cursor = true;
                break :blk null;
            };
        }

        return .{ .consumer = null, .broadcaster = ctx, .conn = conn, .cursor = cursor, .invalid_cursor = invalid_cursor };
    }

    /// called AFTER the websocket handshake reply has been sent to the client.
    /// safe to start the consumer write thread and begin sending frames.
    pub fn afterInit(self: *Handler, ctx: *Broadcaster) !void {
        const consumer = ctx.addConsumer(self.conn) catch return error.Unexpected;
        self.consumer = consumer;

        if (self.invalid_cursor) {
            // spec: send error frame with InvalidCursor, then close
            var buf: [256]u8 = undefined;
            if (encodeErrorFrame("InvalidCursor", "cursor parameter is not a valid integer", &buf)) |frame| {
                consumer.conn.writeBin(frame) catch {};
            }
            consumer.conn.close(.{}) catch {};
            return;
        }

        if (self.cursor) |cursor| {
            const latest = ctx.stats.relay_seq.load(.acquire);

            // FutureCursor: cursor ahead of current seq — error, close
            if (latest > 0 and cursor > latest) {
                var buf: [256]u8 = undefined;
                if (encodeErrorFrame("FutureCursor", "Cursor in the future.", &buf)) |frame| {
                    consumer.conn.writeBin(frame) catch {};
                }
                consumer.conn.close(.{}) catch {};
                return;
            }

            // OutdatedCursor: cursor older than oldest available — info, continue
            const oldest = blk: {
                if (ctx.persist != null and ctx.db_queue != null) {
                    const FirstSeqReq = struct {
                        base: event_log_mod.DbRequest = .{ .callback = &execute },
                        result: ?u64 = null,

                        fn execute(b: *event_log_mod.DbRequest, dp: *event_log_mod.DiskPersist) void {
                            const s: *@This() = @fieldParentPtr("base", b);
                            s.result = dp.firstSeq();
                        }
                    };
                    var first_req: FirstSeqReq = .{};
                    ctx.db_queue.?.push(&first_req.base);
                    first_req.base.wait(ctx.io, ctx.shutdown);
                    if (first_req.result) |s| break :blk s;
                }
                break :blk ctx.history.oldestSeq() orelse 0;
            };
            if (oldest > 0 and cursor < oldest) {
                var buf: [256]u8 = undefined;
                if (encodeInfoMessage("OutdatedCursor", "Requested cursor exceeded limit. Possibly missing events", &buf)) |frame| {
                    consumer.conn.writeBin(frame) catch {};
                }
                // don't close — fall through to replayTo which replays from oldest available
            }

            ctx.replayTo(consumer, cursor);
        }
    }

    pub fn clientMessage(_: *Handler, _: []const u8) !void {
        // relay consumers are read-only
    }

    /// handle client-initiated close frame. without this, the websocket
    /// library tries to write a close reply which can panic if the socket
    /// is already dead (unreachable in std/net.zig drain).
    pub fn clientClose(self: *Handler, _: []const u8) !void {
        if (self.consumer) |c| {
            c.alive.store(false, .release);
            c.cond.signal(c.io);
        }
        self.conn.close(.{}) catch {};
    }

    pub fn close(self: *Handler) void {
        if (self.consumer) |c| self.broadcaster.removeConsumer(c);
    }

    /// called by websocket server for non-WebSocket HTTP requests
    pub fn httpFallback(
        conn: *websocket.Conn,
        method: []const u8,
        url: []const u8,
        body: []const u8,
        headers: *const websocket.Handshake.KeyValue,
        ctx: *Broadcaster,
    ) void {
        if (ctx.http_fallback) |f| {
            f(conn, method, url, body, headers, ctx.http_fallback_ctx);
        }
    }
};

pub const AttributionMetrics = struct {
    history_entries: usize = 0,
    evtbuf_entries: usize = 0,
    did_cache_entries: usize = 0,
    host_authority_cache_entries: usize = 0,
    resolve_queue_len: usize = 0,
    resolve_queued_set_count: usize = 0,
    // memory attribution — internal capacities of key data structures
    validator_cache_map_cap: usize = 0,
    did_cache_map_cap: usize = 0,
    queued_set_map_cap: usize = 0,
    evtbuf_cap: usize = 0,
    outbuf_cap: usize = 0,
    workers_count: usize = 0,
    // db_queue liveness — handled stops advancing while depth > 0 = workers wedged
    db_queue_depth: u32 = 0,
    db_queue_handled_total: u64 = 0,
};

pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, attribution: AttributionMetrics, data_dir: []const u8, buf: []u8, io: Io) []const u8 {
    const uptime: i64 = timestamp(io) - stats.start_time;
    var w: Io.Writer = .fixed(buf);

    // canary metric — proves what binary is running
    w.print(
        \\# TYPE relay_build_info gauge
        \\# HELP relay_build_info build provenance
        \\relay_build_info{{git_sha="{s}",optimize="{s}"}} 1
        \\
    , .{ build_options.git_sha, build_options.optimize }) catch {};

    w.print(
        \\# TYPE relay_frames_received_total counter
        \\relay_frames_received_total {d}
        \\
        \\# TYPE relay_frames_broadcast_total counter
        \\relay_frames_broadcast_total {d}
        \\
        \\# TYPE relay_consumers_active gauge
        \\relay_consumers_active {d}
        \\
        \\# TYPE relay_validation_total counter
        \\relay_validation_total{{result="validated"}} {d}
        \\relay_validation_total{{result="failed"}} {d}
        \\relay_validation_total{{result="skipped"}} {d}
        \\
        \\# TYPE relay_rate_limited_total counter
        \\relay_rate_limited_total {d}
        \\
        \\# TYPE relay_rate_limited_by_tier counter
        \\relay_rate_limited_by_tier{{tier="sec"}} {d}
        \\relay_rate_limited_by_tier{{tier="hour"}} {d}
        \\relay_rate_limited_by_tier{{tier="day"}} {d}
        \\
        \\# TYPE relay_pool_backpressure_total counter
        \\# HELP relay_pool_backpressure_total times pool submit blocked on full queue
        \\relay_pool_backpressure_total {d}
        \\
        \\# TYPE relay_decode_errors_total counter
        \\relay_decode_errors_total {d}
        \\
        \\# TYPE relay_cache_total counter
        \\relay_cache_total{{result="hit"}} {d}
        \\relay_cache_total{{result="miss"}} {d}
        \\
        \\# TYPE relay_slow_consumers_total counter
        \\relay_slow_consumers_total {d}
        \\
        \\# TYPE relay_connected_inbound gauge
        \\relay_connected_inbound {d}
        \\
        \\# TYPE relay_upstream_seq gauge
        \\relay_upstream_seq {d}
        \\
        \\# TYPE relay_seq gauge
        \\relay_seq {d}
        \\
        \\# TYPE relay_uptime_seconds gauge
        \\relay_uptime_seconds {d}
        \\
        \\# TYPE relay_validator_cache_entries gauge
        \\relay_validator_cache_entries {d}
        \\
        \\# TYPE relay_validator_cache_evictions_total counter
        \\relay_validator_cache_evictions_total {d}
        \\
        \\# TYPE relay_history_entries gauge
        \\# HELP relay_history_entries in-memory frame history ring buffer entries
        \\relay_history_entries {d}
        \\
        \\# TYPE relay_evtbuf_entries gauge
        \\# HELP relay_evtbuf_entries pending flush buffer entries
        \\relay_evtbuf_entries {d}
        \\
        \\# TYPE relay_did_cache_entries gauge
        \\# HELP relay_did_cache_entries DID-to-UID cache entries
        \\relay_did_cache_entries {d}
        \\
        \\# TYPE relay_resolve_queue_len gauge
        \\# HELP relay_resolve_queue_len DID resolve queue length
        \\relay_resolve_queue_len {d}
        \\
        \\# TYPE relay_resolve_queued_set_count gauge
        \\# HELP relay_resolve_queued_set_count DID resolve dedup set size
        \\relay_resolve_queued_set_count {d}
        \\
        \\# TYPE relay_host_authority_checks_total counter
        \\# HELP relay_host_authority_checks_total resolveHostAuthority calls on frame workers
        \\relay_host_authority_checks_total {d}
        \\
        \\# TYPE relay_host_authority_trigger counter
        \\# HELP relay_host_authority_trigger host authority check triggers by reason
        \\relay_host_authority_trigger{{reason="is_new"}} {d}
        \\relay_host_authority_trigger{{reason="host_changed"}} {d}
        \\
        \\# TYPE relay_host_authority_time_us_total counter
        \\# HELP relay_host_authority_time_us_total cumulative microseconds in resolveHostAuthority
        \\relay_host_authority_time_us_total {d}
        \\
        \\# TYPE relay_pool_queued_bytes gauge
        \\# HELP relay_pool_queued_bytes estimated bytes of frame data queued in worker pool
        \\relay_pool_queued_bytes {d}
        \\
    , .{
        stats.frames_in.load(.acquire),
        stats.frames_out.load(.acquire),
        stats.consumer_count.load(.acquire),
        stats.validated.load(.acquire),
        stats.failed.load(.acquire),
        stats.skipped.load(.acquire),
        stats.rate_limited.load(.acquire),
        stats.rate_limited_sec.load(.acquire),
        stats.rate_limited_hour.load(.acquire),
        stats.rate_limited_day.load(.acquire),
        stats.pool_backpressure.load(.acquire),
        stats.decode_errors.load(.acquire),
        stats.cache_hits.load(.acquire),
        stats.cache_misses.load(.acquire),
        stats.slow_consumers.load(.acquire),
        stats.connected_inbound.load(.acquire),
        stats.seq.load(.acquire),
        stats.relay_seq.load(.acquire),
        uptime,
        cache_entries,
        stats.cache_evictions.load(.acquire),
        attribution.history_entries,
        attribution.evtbuf_entries,
        attribution.did_cache_entries,
        attribution.resolve_queue_len,
        attribution.resolve_queued_set_count,
        stats.host_authority_checks.load(.acquire),
        stats.host_authority_is_new.load(.acquire),
        stats.host_authority_host_changed.load(.acquire),
        stats.host_authority_time_us.load(.acquire),
        stats.pool_queued_bytes.load(.acquire),
    }) catch return w.buffered();

    // keepalive pinger metrics (separate print to stay under 32-arg limit)
    w.print(
        \\# TYPE relay_ws_pings_sent_total counter
        \\# HELP relay_ws_pings_sent_total keepalive pings sent to silent upstream connections
        \\relay_ws_pings_sent_total {d}
        \\
        \\# TYPE relay_ws_ping_timeout_closes_total counter
        \\# HELP relay_ws_ping_timeout_closes_total connections closed as half-open after max unanswered keepalive pings
        \\relay_ws_ping_timeout_closes_total {d}
        \\
        \\# TYPE relay_ws_ping_write_failures_total counter
        \\# HELP relay_ws_ping_write_failures_total keepalive ping writes that failed (connection torn down)
        \\relay_ws_ping_write_failures_total {d}
        \\
    , .{
        stats.ws_pings_sent.load(.acquire),
        stats.ws_ping_timeout_closes.load(.acquire),
        stats.ws_ping_write_failures.load(.acquire),
    }) catch return w.buffered();

    w.print(
        \\# TYPE relay_host_authority_cache_hits_total counter
        \\# HELP relay_host_authority_cache_hits_total confirmed host mismatches rejected without network resolution
        \\relay_host_authority_cache_hits_total {d}
        \\
        \\# TYPE relay_host_authority_network_checks_total counter
        \\# HELP relay_host_authority_network_checks_total host authority decisions that performed network resolution
        \\relay_host_authority_network_checks_total {d}
        \\
        \\# TYPE relay_host_authority_network_time_us_total counter
        \\# HELP relay_host_authority_network_time_us_total worker microseconds spent waiting on host authority network decisions
        \\relay_host_authority_network_time_us_total {d}
        \\
        \\# TYPE relay_host_authority_pool_wait_us_total counter
        \\# HELP relay_host_authority_pool_wait_us_total worker microseconds spent waiting for a host resolver pool slot
        \\relay_host_authority_pool_wait_us_total {d}
        \\
        \\# TYPE relay_host_authority_cache_entries gauge
        \\# HELP relay_host_authority_cache_entries confirmed DID and incoming host mismatches cached
        \\relay_host_authority_cache_entries {d}
        \\
    , .{
        stats.host_authority_cache_hits.load(.acquire),
        stats.host_authority_network_checks.load(.acquire),
        stats.host_authority_network_time_us.load(.acquire),
        stats.host_authority_pool_wait_us.load(.acquire),
        attribution.host_authority_cache_entries,
    }) catch return w.buffered();

    // pipeline contention metrics (separate print to stay under 32-arg limit)
    w.print(
        \\# TYPE relay_chain_breaks_total counter
        \\# HELP relay_chain_breaks_total chain continuity failures by reason
        \\relay_chain_breaks_total{{reason="since"}} {d}
        \\relay_chain_breaks_total{{reason="prev_data"}} {d}
        \\
        \\# TYPE relay_persist_order_spins_total counter
        \\# HELP relay_persist_order_spins_total spin iterations waiting for persist ordering lock
        \\relay_persist_order_spins_total {d}
        \\
        \\# TYPE relay_broadcast_queue_push_lock_spins_total counter
        \\# HELP relay_broadcast_queue_push_lock_spins_total spin iterations waiting for broadcast queue push lock
        \\relay_broadcast_queue_push_lock_spins_total {d}
        \\
        \\# TYPE relay_broadcast_queue_full_total counter
        \\# HELP relay_broadcast_queue_full_total spin iterations on full broadcast queue (ring capacity)
        \\relay_broadcast_queue_full_total {d}
        \\
        \\# TYPE relay_broadcast_queue_depth_hwm gauge
        \\# HELP relay_broadcast_queue_depth_hwm high-water mark of broadcast queue depth
        \\relay_broadcast_queue_depth_hwm {d}
        \\
        \\# TYPE relay_broadcast_no_consumers_total counter
        \\# HELP relay_broadcast_no_consumers_total frames skipped broadcast (no consumers)
        \\relay_broadcast_no_consumers_total {d}
        \\
    , .{
        stats.chain_breaks_since.load(.acquire),
        stats.chain_breaks_prev_data.load(.acquire),
        stats.persist_order_spins.load(.acquire),
        stats.broadcast_queue_push_lock_spins.load(.acquire),
        stats.broadcast_queue_full.load(.acquire),
        stats.broadcast_queue_depth_hwm.load(.acquire),
        stats.broadcast_no_consumers.load(.acquire),
    }) catch {};

    // validation failure breakdown by reason
    w.print(
        \\# TYPE relay_validation_failed counter
        \\# HELP relay_validation_failed validation failures by reason
        \\relay_validation_failed{{reason="bad_did"}} {d}
        \\relay_validation_failed{{reason="bad_rev"}} {d}
        \\relay_validation_failed{{reason="missing_blocks"}} {d}
        \\relay_validation_failed{{reason="oversized_blocks"}} {d}
        \\relay_validation_failed{{reason="bad_structure"}} {d}
        \\relay_validation_failed{{reason="signature"}} {d}
        \\relay_validation_failed{{reason="commit_integrity"}} {d}
        \\relay_validation_failed{{reason="sync_1_1"}} {d}
        \\relay_validation_failed{{reason="prev_data_missing"}} {d}
        \\relay_validation_failed{{reason="host_authority"}} {d}
        \\relay_validation_failed{{reason="future_rev"}} {d}
        \\
        \\# TYPE relay_host_authority_reject counter
        \\# HELP relay_host_authority_reject host authority reject breakdown by branch
        \\relay_host_authority_reject{{branch="parse_did"}} {d}
        \\relay_host_authority_reject{{branch="resolve"}} {d}
        \\relay_host_authority_reject{{branch="no_endpoint"}} {d}
        \\relay_host_authority_reject{{branch="bad_url"}} {d}
        \\relay_host_authority_reject{{branch="unknown_host"}} {d}
        \\relay_host_authority_reject{{branch="host_mismatch"}} {d}
        \\
        \\# TYPE relay_subscriber_disconnect_total counter
        \\# HELP relay_subscriber_disconnect_total subscriber reconnect-cycle disconnects by phase
        \\relay_subscriber_disconnect_total{{phase="dns_connect"}} {d}
        \\relay_subscriber_disconnect_total{{phase="tls_init"}} {d}
        \\relay_subscriber_disconnect_total{{phase="ws_handshake"}} {d}
        \\relay_subscriber_disconnect_total{{phase="read_loop"}} {d}
        \\
    , .{
        stats.failed_bad_did.load(.acquire),
        stats.failed_bad_rev.load(.acquire),
        stats.failed_missing_blocks.load(.acquire),
        stats.failed_oversized_blocks.load(.acquire),
        stats.failed_bad_structure.load(.acquire),
        stats.failed_signature.load(.acquire),
        stats.failed_commit_integrity.load(.acquire),
        stats.failed_sync_1_1.load(.acquire),
        stats.failed_prev_data_missing.load(.acquire),
        stats.failed_host_authority.load(.acquire),
        stats.failed_future_rev.load(.acquire),
        stats.host_authority_reject_parse_did.load(.acquire),
        stats.host_authority_reject_resolve.load(.acquire),
        stats.host_authority_reject_no_endpoint.load(.acquire),
        stats.host_authority_reject_bad_url.load(.acquire),
        stats.host_authority_reject_unknown_host.load(.acquire),
        stats.host_authority_reject_host_mismatch.load(.acquire),
        stats.subscriber_disconnect_dns_connect.load(.acquire),
        stats.subscriber_disconnect_tls_init.load(.acquire),
        stats.subscriber_disconnect_ws_handshake.load(.acquire),
        stats.subscriber_disconnect_read_loop.load(.acquire),
    }) catch return w.buffered();

    // memory attribution — internal capacities help identify what's consuming RSS
    w.print(
        \\# TYPE relay_validator_cache_map_cap gauge
        \\# HELP relay_validator_cache_map_cap hashmap backing capacity of signing key LRU
        \\relay_validator_cache_map_cap {d}
        \\
        \\# TYPE relay_did_cache_map_cap gauge
        \\# HELP relay_did_cache_map_cap hashmap backing capacity of DID-to-UID LRU
        \\relay_did_cache_map_cap {d}
        \\
        \\# TYPE relay_queued_set_map_cap gauge
        \\# HELP relay_queued_set_map_cap hashmap backing capacity of resolver dedup set
        \\relay_queued_set_map_cap {d}
        \\
        \\# TYPE relay_evtbuf_cap gauge
        \\# HELP relay_evtbuf_cap allocated capacity of event flush buffer (jobs)
        \\relay_evtbuf_cap {d}
        \\
        \\# TYPE relay_outbuf_cap gauge
        \\# HELP relay_outbuf_cap allocated capacity of output byte buffer
        \\relay_outbuf_cap {d}
        \\
        \\# TYPE relay_workers_count gauge
        \\# HELP relay_workers_count active subscriber worker threads
        \\relay_workers_count {d}
        \\
        \\# TYPE relay_db_queue_depth gauge
        \\# HELP relay_db_queue_depth current depth of the DbRequestQueue (tail - head). non-zero with stagnant relay_db_queue_handled_total = workers wedged
        \\relay_db_queue_depth {d}
        \\
        \\# TYPE relay_db_queue_handled_total counter
        \\# HELP relay_db_queue_handled_total total DbRequest callbacks executed across worker threads
        \\relay_db_queue_handled_total {d}
        \\
    , .{
        attribution.validator_cache_map_cap,
        attribution.did_cache_map_cap,
        attribution.queued_set_map_cap,
        attribution.evtbuf_cap,
        attribution.outbuf_cap,
        attribution.workers_count,
        attribution.db_queue_depth,
        attribution.db_queue_handled_total,
    }) catch {};

    // linux-only process metrics from /proc
    if (comptime builtin.os.tag == .linux) {
        appendProcMetrics(&w, io);
        appendDiskMetrics(&w, data_dir);
    }

    return w.buffered();
}

fn appendProcMetrics(w: *Io.Writer, io: Io) void {
    // RSS from /proc/self/statm (field[1] * page_size)
    if (Io.Dir.openFileAbsolute(io, "/proc/self/statm", .{})) |f| {
        defer f.close(io);
        var statm_buf: [256]u8 = undefined;
        const n = f.readPositionalAll(io, &statm_buf, 0) catch 0;
        if (n > 0) {
            const line = statm_buf[0..n];
            var iter = std.mem.splitScalar(u8, line, ' ');
            _ = iter.next(); // skip total pages
            if (iter.next()) |rss_pages| {
                if (std.fmt.parseInt(u64, rss_pages, 10)) |pages| {
                    w.print(
                        \\# TYPE relay_process_rss_bytes gauge
                        \\relay_process_rss_bytes {d}
                        \\
                    , .{pages * 4096}) catch {};
                } else |_| {}
            }
        }
    } else |_| {}

    // thread count, VmHWM, RssAnon from /proc/self/status
    if (Io.Dir.openFileAbsolute(io, "/proc/self/status", .{})) |f| {
        defer f.close(io);
        var status_buf: [4096]u8 = undefined;
        const n = f.readPositionalAll(io, &status_buf, 0) catch 0;
        if (n > 0) {
            const content = status_buf[0..n];
            const fields = .{
                .{ "Threads:", "relay_threads_total" },
                .{ "VmHWM:", "relay_vm_hwm_kb" },
                .{ "RssAnon:", "relay_rss_anon_kb" },
            };
            inline for (fields) |entry| {
                if (std.mem.indexOf(u8, content, entry[0])) |pos| {
                    const rest = content[pos + entry[0].len ..];
                    const trimmed = std.mem.trimStart(u8, rest, " \t");
                    const end = std.mem.indexOfScalar(u8, trimmed, ' ') orelse
                        (std.mem.indexOfScalar(u8, trimmed, '\n') orelse trimmed.len);
                    if (std.fmt.parseInt(u64, trimmed[0..end], 10)) |val| {
                        w.print(
                            \\# TYPE {s} gauge
                            \\{s} {d}
                            \\
                        , .{ entry[1], entry[1], val }) catch {};
                    } else |_| {}
                }
            }
        }
    } else |_| {}

    // glibc malloc stats — mallinfo fields are c_int, bitcast to u32 to extend range to 4 GiB.
    // note: mallinfo only reports the main arena, not per-thread arenas.
    // for accurate total heap picture, rely on RssAnon from /proc/self/status above.
    const mi = mallinfo();
    const arena: u64 = @as(u32, @bitCast(mi.arena));
    const in_use: u64 = @as(u32, @bitCast(mi.uordblks));
    const free_bytes: u64 = @as(u32, @bitCast(mi.fordblks));
    const mmap_bytes: u64 = @as(u32, @bitCast(mi.hblkhd));
    w.print(
        \\# TYPE relay_malloc_arena_bytes gauge
        \\relay_malloc_arena_bytes {d}
        \\
        \\# TYPE relay_malloc_in_use_bytes gauge
        \\relay_malloc_in_use_bytes {d}
        \\
        \\# TYPE relay_malloc_free_bytes gauge
        \\relay_malloc_free_bytes {d}
        \\
        \\# TYPE relay_malloc_mmap_bytes gauge
        \\relay_malloc_mmap_bytes {d}
        \\
    , .{ arena, in_use, free_bytes, mmap_bytes }) catch {};

    // all-arena heap picture via malloc_info() (mallinfo above is main-arena only).
    // with ~2670 worker threads glibc spreads allocations across many per-thread
    // arenas; post-burst fragmentation lives there, invisible to mallinfo. these
    // gauges sum every arena so a rising free_total / arena_count is observable.
    if (readMallocInfo()) |info| {
        w.print(
            \\# TYPE relay_malloc_system_bytes gauge
            \\# HELP relay_malloc_system_bytes total bytes obtained from the OS across all arenas
            \\relay_malloc_system_bytes {d}
            \\
            \\# TYPE relay_malloc_free_total_bytes gauge
            \\# HELP relay_malloc_free_total_bytes free bytes held in all arenas (fast+rest), not returned to OS
            \\relay_malloc_free_total_bytes {d}
            \\
            \\# TYPE relay_malloc_arena_count gauge
            \\# HELP relay_malloc_arena_count number of glibc malloc arenas (tune via MALLOC_ARENA_MAX)
            \\relay_malloc_arena_count {d}
            \\
        , .{ info.system_bytes, info.free_bytes, info.arena_count }) catch {};
    }
}

// glibc mallinfo struct — fields are c_int (32-bit), we bitcast to u32 for range.
const MallInfo = extern struct {
    arena: c_int,
    ordblks: c_int,
    smblks: c_int,
    hblks: c_int,
    hblkhd: c_int,
    usmblks: c_int,
    fsmblks: c_int,
    uordblks: c_int,
    fordblks: c_int,
    keepcost: c_int,
};

const mallinfo = @extern(*const fn () callconv(.c) MallInfo, .{ .name = "mallinfo" });

// all-arena heap stats, parsed from glibc malloc_info() XML. unlike mallinfo()
// these cover every per-thread arena, not just the main one.
const MallocInfo = struct {
    system_bytes: u64,
    free_bytes: u64,
    arena_count: u32,
};

const CFile = opaque {};
const open_memstream = @extern(*const fn (ptr: *[*c]u8, size: *usize) callconv(.c) ?*CFile, .{ .name = "open_memstream" });
const fclose = @extern(*const fn (*CFile) callconv(.c) c_int, .{ .name = "fclose" });
const c_free = @extern(*const fn (?*anyopaque) callconv(.c) void, .{ .name = "free" });
const malloc_info_fn = @extern(*const fn (options: c_int, stream: *CFile) callconv(.c) c_int, .{ .name = "malloc_info" });

fn readMallocInfo() ?MallocInfo {
    // comptime if prunes the linux-only externs from analysis on other hosts,
    // so `zig build test` on macOS doesn't try to link malloc_info/open_memstream.
    return if (comptime builtin.os.tag == .linux) readMallocInfoLinux() else null;
}

fn readMallocInfoLinux() ?MallocInfo {
    var buf_ptr: [*c]u8 = null;
    var buf_size: usize = 0;
    const stream = open_memstream(&buf_ptr, &buf_size) orelse return null;
    _ = malloc_info_fn(0, stream);
    _ = fclose(stream); // flushes and populates buf_ptr/buf_size
    const p = buf_ptr orelse return null;
    defer c_free(p);
    const xml = p[0..buf_size];
    return MallocInfo{
        // top-level totals are the LAST occurrence (each <heap> emits its own).
        .system_bytes = lastSizeAttr(xml, "system type=\"current\"") orelse 0,
        .free_bytes = (lastSizeAttr(xml, "total type=\"fast\"") orelse 0) +
            (lastSizeAttr(xml, "total type=\"rest\"") orelse 0),
        .arena_count = countOccurrences(xml, "<heap nr="),
    };
}

// find the last occurrence of `marker`, then parse the first `size="N"` after it.
fn lastSizeAttr(xml: []const u8, marker: []const u8) ?u64 {
    const at = std.mem.lastIndexOf(u8, xml, marker) orelse return null;
    const rest = xml[at + marker.len ..];
    const key = "size=\"";
    const kpos = std.mem.indexOf(u8, rest, key) orelse return null;
    const num = rest[kpos + key.len ..];
    const end = std.mem.indexOfScalar(u8, num, '"') orelse return null;
    return std.fmt.parseInt(u64, num[0..end], 10) catch null;
}

fn countOccurrences(haystack: []const u8, needle: []const u8) u32 {
    var n: u32 = 0;
    var i: usize = 0;
    while (std.mem.indexOfPos(u8, haystack, i, needle)) |pos| {
        n += 1;
        i = pos + needle.len;
    }
    return n;
}

test "readMallocInfo parses all-arena totals" {
    if (comptime builtin.os.tag != .linux) return error.SkipZigTest;
    // force some allocation so the arena is non-trivial
    const a = std.heap.c_allocator;
    var kept: [64][]u8 = undefined;
    for (&kept) |*slot| slot.* = try a.alloc(u8, 4096);
    defer for (kept) |s| a.free(s);

    const info = readMallocInfo() orelse return error.MallocInfoUnavailable;
    try std.testing.expect(info.system_bytes > 0);
    try std.testing.expect(info.arena_count >= 1);
}

test "lastSizeAttr / countOccurrences against sample XML" {
    const xml =
        \\<malloc version="1">
        \\<heap nr="0">
        \\<total type="fast" count="1" size="100"/>
        \\<total type="rest" count="2" size="200"/>
        \\<system type="current" size="9999"/>
        \\</heap>
        \\<heap nr="1">
        \\<total type="fast" count="3" size="111"/>
        \\<total type="rest" count="4" size="222"/>
        \\<system type="current" size="8888"/>
        \\</heap>
        \\<total type="fast" count="4" size="211"/>
        \\<total type="rest" count="6" size="422"/>
        \\<system type="current" size="18887"/>
        \\</malloc>
    ;
    try std.testing.expectEqual(@as(?u64, 18887), lastSizeAttr(xml, "system type=\"current\""));
    try std.testing.expectEqual(@as(?u64, 211), lastSizeAttr(xml, "total type=\"fast\""));
    try std.testing.expectEqual(@as(?u64, 422), lastSizeAttr(xml, "total type=\"rest\""));
    try std.testing.expectEqual(@as(u32, 2), countOccurrences(xml, "<heap nr="));
}

// statvfs — only the fields we use (f_frsize, f_blocks, f_bavail)
const StatVfs = extern struct {
    f_bsize: c_ulong,
    f_frsize: c_ulong,
    f_blocks: c_ulong,
    f_bfree: c_ulong,
    f_bavail: c_ulong,
    f_files: c_ulong,
    f_ffree: c_ulong,
    f_favail: c_ulong,
    f_fsid: c_ulong,
    f_flag: c_ulong,
    f_namemax: c_ulong,
};

const statvfs_fn = @extern(*const fn ([*:0]const u8, *StatVfs) callconv(.c) c_int, .{ .name = "statvfs" });

fn appendDiskMetrics(w: *Io.Writer, data_dir: []const u8) void {
    // statvfs needs a null-terminated path
    var path_buf: [4096]u8 = undefined;
    if (data_dir.len >= path_buf.len) return;
    @memcpy(path_buf[0..data_dir.len], data_dir);
    path_buf[data_dir.len] = 0;

    var stat: StatVfs = undefined;
    if (statvfs_fn(@ptrCast(&path_buf), &stat) != 0) return;

    const block_size: u64 = stat.f_frsize;
    const total = stat.f_blocks * block_size;
    const available = stat.f_bavail * block_size;

    w.print(
        \\# TYPE relay_disk_total_bytes gauge
        \\relay_disk_total_bytes {d}
        \\
        \\# TYPE relay_disk_available_bytes gauge
        \\relay_disk_available_bytes {d}
        \\
    , .{ total, available }) catch {};
}

pub fn formatStatsResponse(stats: *const Stats, buf: []u8, io: Io) []const u8 {
    return std.fmt.bufPrint(buf,
        \\{{"seq":{d},"relay_seq":{d},"consumers":{d},"connected_inbound":{d},"frames_in":{d},"frames_out":{d},"validated":{d},"failed":{d},"skipped":{d},"decode_errors":{d},"cache_hits":{d},"cache_misses":{d},"slow_consumers":{d},"chain_breaks":{d},"pool_backpressure":{d},"uptime_seconds":{d}}}
    , .{
        stats.seq.load(.acquire),
        stats.relay_seq.load(.acquire),
        stats.consumer_count.load(.acquire),
        stats.connected_inbound.load(.acquire),
        stats.frames_in.load(.acquire),
        stats.frames_out.load(.acquire),
        stats.validated.load(.acquire),
        stats.failed.load(.acquire),
        stats.skipped.load(.acquire),
        stats.decode_errors.load(.acquire),
        stats.cache_hits.load(.acquire),
        stats.cache_misses.load(.acquire),
        stats.slow_consumers.load(.acquire),
        stats.chain_breaks_since.load(.acquire) + stats.chain_breaks_prev_data.load(.acquire),
        stats.pool_backpressure.load(.acquire),
        timestamp(io) - stats.start_time,
    }) catch "";
}

// --- tests ---

var test_shutdown: std.atomic.Value(bool) = .{ .raw = false };

test "broadcaster add and remove consumer" {
    var b = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown);
    defer b.deinit();

    try std.testing.expectEqual(@as(u64, 0), b.stats.seq.load(.acquire));
    try std.testing.expectEqual(@as(usize, 0), b.consumerCount());
}

test "broadcast updates stats and history" {
    var b = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown);
    defer b.deinit();

    b.broadcast(1, "frame1");
    b.broadcast(2, "frame2");
    b.broadcast(3, "frame3");

    try std.testing.expectEqual(@as(u64, 3), b.stats.seq.load(.acquire));
    // frames_in is now incremented in subscriber, not broadcast
    try std.testing.expectEqual(@as(u64, 0), b.stats.frames_in.load(.acquire));
    try std.testing.expectEqual(@as(u64, 1), b.history.oldestSeq().?);
    try std.testing.expectEqual(@as(u64, 3), b.history.newestSeq().?);
}

test "frame history supports cursor replay" {
    var b = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown);
    defer b.deinit();

    for (1..6) |i| {
        var buf: [32]u8 = undefined;
        const data = std.fmt.bufPrint(&buf, "frame{d}", .{i}) catch unreachable;
        b.broadcast(@intCast(i), data);
    }

    const frames = try b.history.framesSince(std.testing.allocator, 3);
    defer {
        for (frames) |f| std.testing.allocator.free(f.data);
        std.testing.allocator.free(frames);
    }

    try std.testing.expectEqual(@as(usize, 2), frames.len);
    try std.testing.expectEqual(@as(u64, 4), frames[0].seq);
    try std.testing.expectEqual(@as(u64, 5), frames[1].seq);
}

test "enqueue refuses frames for a dead consumer" {
    // regression: replay for an abruptly-disconnected consumer kept
    // accumulating frames, compounding into the 2026-08-06 canary OOM.
    var consumer = Consumer{
        .conn = undefined, // enqueue never touches the connection
        .allocator = std.testing.allocator,
        .io = std.testing.io,
    };

    try std.testing.expect(consumer.enqueueRaw("frame1"));

    consumer.alive.store(false, .release);
    try std.testing.expect(!consumer.enqueueRaw("frame2"));

    // drain the accepted frame
    consumer.mutex.lockUncancelable(consumer.io);
    while (consumer.dequeue()) |f| f.release();
    consumer.mutex.unlock(consumer.io);
}

test "shared frame ref counting" {
    const frame = try SharedFrame.create(std.testing.allocator, "hello");
    frame.acquire(); // ref=2
    frame.acquire(); // ref=3
    frame.release(); // ref=2
    frame.release(); // ref=1
    frame.release(); // ref=0 → freed
}

test "error frame is valid CBOR" {
    const cbor = zat.cbor;
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    const frame = buildErrorFrame(alloc) orelse return error.FrameBuildFailed;

    // decode header
    const h_result = try cbor.decode(alloc, frame);
    const h = h_result.value;
    try std.testing.expectEqual(@as(i64, -1), h.getInt("op").?);
    // spec: error frames have {op: -1} only, no `t` field
    try std.testing.expect(h.getString("t") == null);

    // decode payload
    const p = try cbor.decodeAll(alloc, frame[h_result.consumed..]);
    try std.testing.expectEqualStrings("ConsumerTooSlow", p.getString("error").?);
    try std.testing.expectEqualStrings("consumer buffer full", p.getString("message").?);
}

test "formatPrometheusMetrics produces valid output" {
    var stats = Stats{ .start_time = timestamp(std.testing.io) - 60 };
    stats.seq.store(99999, .release);
    stats.relay_seq.store(12345, .release);
    stats.consumer_count.store(3, .release);
    stats.frames_in.store(10000, .release);
    stats.frames_out.store(9000, .release);
    stats.validated.store(500, .release);
    stats.failed.store(2, .release);
    stats.skipped.store(9498, .release);
    stats.cache_hits.store(400, .release);
    stats.cache_misses.store(100, .release);
    stats.host_authority_cache_hits.store(321, .release);
    stats.host_authority_network_checks.store(12, .release);
    stats.host_authority_network_time_us.store(456_000, .release);
    stats.host_authority_pool_wait_us.store(789_000, .release);

    var buf: [65536]u8 = undefined;
    const output = formatPrometheusMetrics(&stats, 42, .{
        .host_authority_cache_entries = 9,
    }, "/tmp", &buf, std.testing.io);

    try std.testing.expect(std.mem.indexOf(u8, output, "relay_frames_received_total 10000") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_frames_broadcast_total 9000") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_consumers_active 3") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_validation_total{result=\"validated\"} 500") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_validation_total{result=\"failed\"} 2") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_connected_inbound 0") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_upstream_seq 99999") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_seq 12345") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "# TYPE relay_uptime_seconds gauge") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_validator_cache_entries 42") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_validator_cache_evictions_total 0") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_cache_hits_total 321") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_network_checks_total 12") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_network_time_us_total 456000") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_pool_wait_us_total 789000") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_cache_entries 9") != null);
}

test "ws ping counters render on their own lines with no cross-wiring" {
    var stats = Stats{ .start_time = timestamp(std.testing.io) };
    stats.ws_pings_sent.store(111_111, .release);
    stats.ws_ping_timeout_closes.store(222, .release);
    stats.ws_ping_write_failures.store(3, .release);

    var buf: [65536]u8 = undefined;
    const output = formatPrometheusMetrics(&stats, 0, .{
        .host_authority_cache_entries = 0,
    }, "/tmp", &buf, std.testing.io);

    try std.testing.expect(std.mem.indexOf(u8, output, "relay_ws_pings_sent_total 111111\n") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_ws_ping_timeout_closes_total 222\n") != null);
    try std.testing.expect(std.mem.indexOf(u8, output, "relay_ws_ping_write_failures_total 3\n") != null);
}

test "formatStatsResponse produces valid JSON" {
    var stats = Stats{ .start_time = timestamp(std.testing.io) - 42 };
    stats.seq.store(100, .release);
    stats.frames_in.store(200, .release);
    stats.consumer_count.store(5, .release);

    var buf: [4096]u8 = undefined;
    const response = formatStatsResponse(&stats, &buf, std.testing.io);

    try std.testing.expect(std.mem.startsWith(u8, response, "{"));
    try std.testing.expect(std.mem.indexOf(u8, response, "\"seq\":100") != null);
    try std.testing.expect(std.mem.indexOf(u8, response, "\"consumers\":5") != null);
    try std.testing.expect(std.mem.indexOf(u8, response, "\"frames_in\":200") != null);
    try std.testing.expect(std.mem.indexOf(u8, response, "\"slow_consumers\":0") != null);
}

test "resequenceFrame replaces seq in payload" {
    const cbor = zat.cbor;
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    // build a frame with upstream seq=12345
    const header: cbor.Value = .{ .map = &.{
        .{ .key = "op", .value = .{ .unsigned = 1 } },
        .{ .key = "t", .value = .{ .text = "#commit" } },
    } };
    const payload: cbor.Value = .{ .map = &.{
        .{ .key = "repo", .value = .{ .text = "did:plc:test123" } },
        .{ .key = "seq", .value = .{ .unsigned = 12345 } },
        .{ .key = "rev", .value = .{ .text = "3k2abc000000" } },
    } };

    const header_bytes = try cbor.encodeAlloc(alloc, header);
    const payload_bytes = try cbor.encodeAlloc(alloc, payload);
    const frame = try alloc.alloc(u8, header_bytes.len + payload_bytes.len);
    @memcpy(frame[0..header_bytes.len], header_bytes);
    @memcpy(frame[header_bytes.len..], payload_bytes);

    // resequence to relay seq 99999
    const result = resequenceFrame(alloc, frame, 99999) orelse return error.ResequenceFailed;

    // decode result and verify
    const h_result = try cbor.decode(alloc, result);
    const p = try cbor.decodeAll(alloc, result[h_result.consumed..]);

    // header preserved
    try std.testing.expectEqualStrings("#commit", h_result.value.getString("t").?);
    try std.testing.expectEqual(@as(i64, 1), h_result.value.getInt("op").?);
    // seq replaced with relay seq
    try std.testing.expectEqual(@as(u64, 99999), p.getUint("seq").?);
    // other fields preserved
    try std.testing.expectEqualStrings("did:plc:test123", p.getString("repo").?);
    try std.testing.expectEqualStrings("3k2abc000000", p.getString("rev").?);
    // tooBig injected for commit frames missing it
    try std.testing.expectEqual(false, p.getBool("tooBig").?);
}

test "resequenceFrame preserves identity frame fields" {
    const cbor = zat.cbor;
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    const header: cbor.Value = .{ .map = &.{
        .{ .key = "op", .value = .{ .unsigned = 1 } },
        .{ .key = "t", .value = .{ .text = "#identity" } },
    } };
    const payload: cbor.Value = .{ .map = &.{
        .{ .key = "did", .value = .{ .text = "did:plc:alice" } },
        .{ .key = "seq", .value = .{ .unsigned = 500 } },
        .{ .key = "time", .value = .{ .text = "2024-01-15T10:30:00Z" } },
    } };

    const header_bytes = try cbor.encodeAlloc(alloc, header);
    const payload_bytes = try cbor.encodeAlloc(alloc, payload);
    const frame = try alloc.alloc(u8, header_bytes.len + payload_bytes.len);
    @memcpy(frame[0..header_bytes.len], header_bytes);
    @memcpy(frame[header_bytes.len..], payload_bytes);

    const result = resequenceFrame(alloc, frame, 42) orelse return error.ResequenceFailed;

    const h_result = try cbor.decode(alloc, result);
    const p = try cbor.decodeAll(alloc, result[h_result.consumed..]);

    try std.testing.expectEqualStrings("#identity", h_result.value.getString("t").?);
    try std.testing.expectEqual(@as(u64, 42), p.getUint("seq").?);
    try std.testing.expectEqualStrings("did:plc:alice", p.getString("did").?);
    try std.testing.expectEqualStrings("2024-01-15T10:30:00Z", p.getString("time").?);
    // tooBig NOT injected for non-commit frames
    try std.testing.expectEqual(@as(?bool, null), p.getBool("tooBig"));
}

test "encodeErrorFrame produces valid error frame CBOR" {
    const cbor = zat.cbor;
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    var buf: [256]u8 = undefined;
    const frame = encodeErrorFrame("FutureCursor", "Cursor in the future.", &buf) orelse return error.EncodeFailed;

    // decode header — must be {op: -1}
    const h_result = try cbor.decode(alloc, frame);
    const h = h_result.value;
    try std.testing.expectEqual(@as(i64, -1), h.getInt("op").?);
    try std.testing.expect(h.getString("t") == null); // no t field on error frames

    // decode payload — must have error and message fields
    const p = try cbor.decodeAll(alloc, frame[h_result.consumed..]);
    try std.testing.expectEqualStrings("FutureCursor", p.getString("error").?);
    try std.testing.expectEqualStrings("Cursor in the future.", p.getString("message").?);
}

test "encodeInfoMessage produces valid #info message frame CBOR" {
    const cbor = zat.cbor;
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    var buf: [256]u8 = undefined;
    const frame = encodeInfoMessage("OutdatedCursor", "Requested cursor exceeded limit. Possibly missing events", &buf) orelse return error.EncodeFailed;

    // decode header — must be {op: 1, t: "#info"}
    const h_result = try cbor.decode(alloc, frame);
    const h = h_result.value;
    try std.testing.expectEqual(@as(i64, 1), h.getInt("op").?);
    try std.testing.expectEqualStrings("#info", h.getString("t").?);

    // decode payload — must have name and message fields (not error)
    const p = try cbor.decodeAll(alloc, frame[h_result.consumed..]);
    try std.testing.expectEqualStrings("OutdatedCursor", p.getString("name").?);
    try std.testing.expectEqualStrings("Requested cursor exceeded limit. Possibly missing events", p.getString("message").?);
    // info messages must NOT have an error field
    try std.testing.expect(p.getString("error") == null);
}

test "concurrent broadcast queue + drain produces monotonic sequences" {
    // regression test: multiple worker threads push to the broadcast queue
    // under persist_order lock, then the broadcaster fiber drains and broadcasts.
    // the ring buffer history must be strictly monotonic.

    var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown);
    defer bc.deinit();

    const num_threads = 8;
    const frames_per_thread = 500;
    const total_frames = num_threads * frames_per_thread;
    var seq_counter = std.atomic.Value(u64).init(0);

    const Worker = struct {
        fn run(bc_ptr: *Broadcaster, counter: *std.atomic.Value(u64)) void {
            for (0..frames_per_thread) |_| {
                while (bc_ptr.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
                    std.atomic.spinLoopHint();
                }
                const seq = counter.fetchAdd(1, .monotonic) + 1;
                const data = std.testing.allocator.dupe(u8, "x") catch {
                    bc_ptr.persist_order.store(0, .release);
                    continue;
                };
                bc_ptr.broadcast_queue.push(seq, data, &bc_ptr.stats);
                bc_ptr.persist_order.store(0, .release);
            }
        }
    };

    var threads: [num_threads]std.Thread = undefined;
    for (&threads) |*t| {
        t.* = std.Thread.spawn(.{}, Worker.run, .{ &bc, &seq_counter }) catch unreachable;
    }
    for (&threads) |*t| t.join();

    // drain the queue (simulating the broadcaster fiber)
    while (bc.broadcast_queue.pop()) |item| {
        bc.broadcast(item.seq, item.data);
        std.testing.allocator.free(@constCast(item.data));
    }

    // verify: ring buffer history has strictly monotonic sequences
    const frames = try bc.history.framesSince(std.testing.allocator, 0);
    defer {
        for (frames) |f| std.testing.allocator.free(f.data);
        std.testing.allocator.free(frames);
    }

    // history is capped at 50K, we pushed 4K — all should be present
    try std.testing.expectEqual(total_frames, frames.len);

    var prev_seq: u64 = 0;
    for (frames) |f| {
        if (f.seq <= prev_seq) {
            std.debug.print("monotonicity violation: seq {d} after {d}\n", .{ f.seq, prev_seq });
            return error.NonMonotonicSequence;
        }
        prev_seq = f.seq;
    }
}

test "shutdown keeps draining until the last producer finishes" {
    var shutdown = std.atomic.Value(bool){ .raw = true };
    var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &shutdown);
    defer bc.deinit();
    const drain = try std.Thread.spawn(.{}, Broadcaster.runBroadcastLoop, .{&bc});
    defer drain.join();
    defer bc.producers_done.store(true, .release);
    for (1..10_001) |seq| {
        const data = try std.testing.allocator.dupe(u8, "event");
        bc.broadcast_queue.push(seq, data, &bc.stats);
    }
    bc.producers_done.store(true, .release);
    while (bc.stats.seq.load(.acquire) != 10_000) {
        try std.testing.io.sleep(Io.Duration.fromMilliseconds(1), .awake);
    }
    try std.testing.expectEqual(@as(u32, 0), bc.broadcast_queue.depth());
}