//! Live query fanout for `GET /api/queries/live` (PLAN §11.4:455).
//!
//! The DNS query path publishes through `QuerySink`, which calls `publish`
//! before it hands the same entry to the logger: the event stream must never
//! wait on a database. `publish` therefore copies and returns — it allocates
//! nothing, touches no I/O, and holds one mutex across a scan of 32 slots.
//!
//! A subscriber that cannot keep up loses its stream rather than the queries:
//! a full ring sets `overflowed`, the subscriber task sees the flag and ends
//! the response, and the browser's `EventSource` reconnects on its own.
//!
//! `logger.Entry` carries its own bytes, so a ring slot is a plain copy with
//! nothing borrowed from the query that produced it.

const std = @import("std");

const logger = @import("../storage/logger.zig");

pub const Entry = logger.Entry;

/// Concurrent live streams. The per-IP cap (`web.sse_max_connections_per_ip`)
/// keeps one client from taking all of them; `subscribe` returning null is the
/// backstop and answers 503.
pub const max_subscribers = 32;

/// Entries one subscriber may fall behind by. At household query rates this is
/// several seconds of slack on a stalled TCP connection.
///
/// The ring is embedded in the slot, so this multiplies `@sizeOf(logger.Entry)`
/// — about 1.8 KiB once milestone 28 widened it for provenance. One subscriber
/// therefore costs roughly 115 KiB of ring and the whole hub roughly 3.6 MiB,
/// allocated once for the life of the process. That is the reason to keep both
/// this and `max_subscribers` small: they are paid whether or not anyone is
/// watching. `logger.query_log_buffer_max` bounds the other consumer of the
/// same width, the writer queue.
pub const ring_capacity = 64;

pub const SubscriberId = enum(u8) { _ };

/// What `wait` returns: an entry (or the overflow flag) is ready, the caller's
/// timeout passed and it owes the client a heartbeat, or the hub is closing and
/// the subscriber must end its response now.
pub const Wake = enum { ready, timeout, closed };

pub const Hub = struct {
    /// Guards every field of every slot. `publish` runs on the DNS hot path,
    /// so the critical section is copies and flag writes only.
    mutex: std.Io.Mutex,
    /// Milestone-16 ruling 11. Set once, never cleared: a hub that is closing
    /// belongs to a server that is going away.
    closing: bool,
    slots: [max_subscribers]Slot,

    const Slot = struct {
        active: bool,
        /// Set by `publish` when the ring is full. Never cleared while the
        /// subscriber lives: the stream it belongs to is over.
        overflowed: bool,
        head: u32,
        len: u32,
        event: std.Io.Event,
        ring: [ring_capacity]Entry,
    };

    /// Initializes in place. The rings are close to a megabyte, which a
    /// by-value `init` would copy through the caller's frame.
    ///
    /// The ring storage stays undefined: `len` says which slots hold entries.
    pub fn init(self: *Hub) void {
        self.mutex = .init;
        self.closing = false;
        for (&self.slots) |*slot| {
            slot.active = false;
            slot.overflowed = false;
            slot.head = 0;
            slot.len = 0;
            slot.event = .unset;
        }
    }

    /// Claims a slot, or null when all 32 are taken.
    ///
    /// `lockUncancelable` throughout this file: `publish`'s caller is
    /// `Handler.handle`, which has no error union to carry `error.Canceled`
    /// out of (the same reasoning as `clients.Tracker.track`), and the rest of
    /// the surface shares the mutex with it.
    pub fn subscribe(self: *Hub, io: std.Io) ?SubscriberId {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);

        for (&self.slots, 0..) |*slot, index| {
            if (slot.active) continue;
            slot.active = true;
            slot.overflowed = false;
            slot.head = 0;
            slot.len = 0;
            slot.event = .unset;
            return @enumFromInt(index);
        }
        return null;
    }

    /// Releases the slot. The caller must not be waiting on it.
    pub fn unsubscribe(self: *Hub, io: std.Io, id: SubscriberId) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);

        const slot = self.slotOf(id);
        slot.active = false;
        slot.overflowed = false;
        slot.len = 0;
        slot.head = 0;
    }

    /// Milestone-16 ruling 11. Ends every live stream.
    ///
    /// Without this, a graceful drain waits out one heartbeat interval per idle
    /// subscriber: shutting the sockets down does not reach a task parked inside
    /// `wait`, which is waiting on an event, not on the peer. Called before the
    /// web listener begins its shutdown.
    pub fn close(self: *Hub, io: std.Io) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);

        self.closing = true;
        for (&self.slots) |*slot| {
            if (slot.active) slot.event.set(io);
        }
    }

    /// Copies `entry` into every live ring and wakes its subscriber. Called
    /// once per logged query.
    pub fn publish(self: *Hub, io: std.Io, entry: Entry) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);

        for (&self.slots) |*slot| {
            if (!slot.active or slot.overflowed) continue;
            if (slot.len == ring_capacity) {
                slot.overflowed = true;
            } else {
                slot.ring[(slot.head + slot.len) % ring_capacity] = entry;
                slot.len += 1;
            }
            slot.event.set(io);
        }
    }

    /// The oldest entry this subscriber has not seen, or null when its ring is
    /// empty. Check `overflowed` first: entries that predate the overflow are
    /// still readable, but the stream must end once they run out.
    pub fn next(self: *Hub, io: std.Io, id: SubscriberId) ?Entry {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);

        const slot = self.slotOf(id);
        if (slot.len == 0) return null;
        const entry = slot.ring[slot.head];
        slot.head = (slot.head + 1) % ring_capacity;
        slot.len -= 1;
        return entry;
    }

    /// True once this subscriber missed an entry. The subscriber task ends the
    /// response when it sees this.
    pub fn overflowed(self: *Hub, io: std.Io, id: SubscriberId) bool {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        return self.slotOf(id).overflowed;
    }

    /// Blocks until something is ready for this subscriber, `timeout` passes,
    /// or the hub closes; `.timeout` is the heartbeat's cue and `.closed` ends
    /// the stream.
    ///
    /// The event is reset under the mutex and only while the ring is empty, so
    /// a `publish` that lands between the check and the wait sets the event
    /// again and the wait returns at once. Only the owning subscriber task
    /// calls this, which is what `Event.reset` requires (`Io.zig:1866`).
    ///
    /// A spurious futex wakeup reports `.timeout` (`Io.zig:1824`): the caller
    /// sends one heartbeat it did not strictly owe.
    pub fn wait(
        self: *Hub,
        io: std.Io,
        id: SubscriberId,
        timeout: std.Io.Clock.Duration,
    ) std.Io.Cancelable!Wake {
        self.mutex.lockUncancelable(io);
        if (self.closing) {
            self.mutex.unlock(io);
            return .closed;
        }
        const slot = self.slotOf(id);
        if (slot.len > 0 or slot.overflowed) {
            self.mutex.unlock(io);
            return .ready;
        }
        slot.event.reset();
        self.mutex.unlock(io);

        slot.event.waitTimeout(io, .{ .duration = timeout }) catch |err| switch (err) {
            error.Timeout => return .timeout,
            error.Canceled => |e| return e,
        };

        // `close` wakes the same event a publish does, so the flag is what tells
        // the two apart. Reading it here rather than on the next call through
        // keeps the drain from writing one more frame into a dying connection.
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        return if (self.closing) .closed else .ready;
    }

    fn slotOf(self: *Hub, id: SubscriberId) *Slot {
        const slot = &self.slots[@intFromEnum(id)];
        std.debug.assert(slot.active);
        return slot;
    }
};

// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------

const testing = std.testing;

fn sampleEntry(timestamp: i64, domain: []const u8) Entry {
    return .init(.{
        .timestamp = timestamp,
        .domain = domain,
        .client_ip = "192.0.2.10",
        .qtype = 1,
    });
}

fn newHub(gpa: std.mem.Allocator) !*Hub {
    const hub = try gpa.create(Hub);
    hub.init();
    return hub;
}

test "a subscriber reads what was published, oldest first" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);

    hub.publish(io, sampleEntry(1, "first.example"));
    hub.publish(io, sampleEntry(2, "second.example"));

    try testing.expectEqualStrings("first.example", hub.next(io, id).?.domain());
    try testing.expectEqualStrings("second.example", hub.next(io, id).?.domain());
    try testing.expect(hub.next(io, id) == null);
    try testing.expect(!hub.overflowed(io, id));
}

test "an entry published before a subscription is not delivered" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    hub.publish(io, sampleEntry(1, "early.example"));

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);
    try testing.expect(hub.next(io, id) == null);
}

test "every live subscriber receives its own copy" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const first = hub.subscribe(io).?;
    const second = hub.subscribe(io).?;
    defer hub.unsubscribe(io, first);
    defer hub.unsubscribe(io, second);

    hub.publish(io, sampleEntry(7, "shared.example"));

    try testing.expectEqualStrings("shared.example", hub.next(io, first).?.domain());
    try testing.expectEqualStrings("shared.example", hub.next(io, second).?.domain());
}

test "the hub hands out every slot and then refuses" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    var ids: [max_subscribers]SubscriberId = undefined;
    for (&ids) |*id| id.* = hub.subscribe(io).?;
    try testing.expect(hub.subscribe(io) == null);

    hub.unsubscribe(io, ids[3]);
    const reused = hub.subscribe(io).?;
    try testing.expectEqual(ids[3], reused);
}

test "a full ring marks the subscriber overflowed and stops copying" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);

    for (0..ring_capacity) |i| hub.publish(io, sampleEntry(@intCast(i), "fill.example"));
    try testing.expect(!hub.overflowed(io, id));

    hub.publish(io, sampleEntry(999, "lost.example"));
    try testing.expect(hub.overflowed(io, id));

    // What the ring already held is still readable; the entry that overflowed
    // it is not, and the flag stays set.
    var drained: usize = 0;
    while (hub.next(io, id)) |entry| : (drained += 1) {
        try testing.expectEqualStrings("fill.example", entry.domain());
    }
    try testing.expectEqual(@as(usize, ring_capacity), drained);
    try testing.expect(hub.overflowed(io, id));
}

test "the ring wraps around its head" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);

    // Two and a half laps, consuming as we go: the head passes the end of the
    // storage twice and no entry is lost.
    for (0..ring_capacity * 2 + ring_capacity / 2) |i| {
        var buf: [32]u8 = undefined;
        const domain = try std.fmt.bufPrint(&buf, "d{d}.example", .{i});
        hub.publish(io, sampleEntry(@intCast(i), domain));

        const got = hub.next(io, id).?;
        try testing.expectEqualStrings(domain, got.domain());
        try testing.expectEqual(@as(i64, @intCast(i)), got.timestamp);
    }
    try testing.expect(!hub.overflowed(io, id));
}

test "wait returns as soon as an entry is waiting" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);

    const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
    hub.publish(io, sampleEntry(1, "ready.example"));
    try testing.expectEqual(Wake.ready, try hub.wait(io, id, long));
}

test "wait times out on an idle subscriber so the heartbeat can go out" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);

    const brief: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(20), .clock = .awake };
    try testing.expectEqual(Wake.timeout, try hub.wait(io, id, brief));
    try testing.expect(hub.next(io, id) == null);
}

test "a publish wakes a waiting subscriber" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);

    const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
    var future = try io.concurrent(Hub.wait, .{ hub, io, id, long });

    hub.publish(io, sampleEntry(5, "late.example"));

    try testing.expectEqual(Wake.ready, try future.await(io));
    try testing.expectEqualStrings("late.example", hub.next(io, id).?.domain());
}

test "an overflow wakes a waiting subscriber" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);

    for (0..ring_capacity) |i| hub.publish(io, sampleEntry(@intCast(i), "fill.example"));
    while (hub.next(io, id)) |_| {}

    // The ring is empty again but its head sits mid-storage; refill it and
    // overflow, so the wake comes from the flag rather than from an entry.
    for (0..ring_capacity) |i| hub.publish(io, sampleEntry(@intCast(i), "fill.example"));

    const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
    var future = try io.concurrent(Hub.wait, .{ hub, io, id, long });
    hub.publish(io, sampleEntry(999, "lost.example"));

    try testing.expectEqual(Wake.ready, try future.await(io));
    try testing.expect(hub.overflowed(io, id));
}

test "close wakes a parked subscriber instead of leaving it on the heartbeat" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);

    // Far longer than the 15 s heartbeat the handler passes, so a pass here
    // cannot come from the timeout arm.
    const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(600), .clock = .awake };
    var future = try io.concurrent(Hub.wait, .{ hub, io, id, long });

    hub.close(io);

    try testing.expectEqual(Wake.closed, try future.await(io));
}

test "wait on a closed hub returns at once, with no entry pending" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const id = hub.subscribe(io).?;
    defer hub.unsubscribe(io, id);

    hub.close(io);

    const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(600), .clock = .awake };
    try testing.expectEqual(Wake.closed, try hub.wait(io, id, long));
}

test "publishing while subscribers come and go reaches only the live ones" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try newHub(testing.allocator);
    defer testing.allocator.destroy(hub);

    const steady = hub.subscribe(io).?;
    defer hub.unsubscribe(io, steady);

    var churner = try io.concurrent(churn, .{ hub, io });

    var published: usize = 0;
    while (published < 500) : (published += 1) {
        hub.publish(io, sampleEntry(@intCast(published), "churn.example"));
        // Keep the steady subscriber under its ring cap: this test is about
        // the churn, not about overflow.
        while (hub.next(io, steady)) |_| {}
    }
    churner.await(io);
    try testing.expect(!hub.overflowed(io, steady));

    // Every slot the churner used is free again.
    var ids: [max_subscribers - 1]SubscriberId = undefined;
    for (&ids) |*id| id.* = hub.subscribe(io).?;
    for (ids) |id| hub.unsubscribe(io, id);
}

fn churn(hub: *Hub, io: std.Io) void {
    for (0..200) |i| {
        const id = hub.subscribe(io) orelse continue;
        if (i % 3 == 0) _ = hub.next(io, id);
        hub.unsubscribe(io, id);
    }
}