nxdns/src/web/handlers/live.zig

//! `GET /api/queries/live` — the query log as it happens (ruling 20).
//!
//! Server-sent events over chunked transfer. The response buffer is EMPTY on
//! purpose: `BodyWriter.flush` pushes only the protocol writer, never the body
//! writer's own buffer (http.zig:780), so with a zero-length buffer every
//! write lands in the chunked drain and one `flush` puts the frame on the
//! wire. `retry: 3000` goes out first so a dropped stream reconnects on the
//! browser's side without configuration.
//!
//! The subscriber owns one hub slot and drains it between waits. A ring
//! overflow means this client is too slow for the query rate; the stream ends
//! cleanly and the reconnecting client re-syncs through `/api/queries` —
//! dropping the client beats holding queries back (PLAN §11.4). The `: ping`
//! heartbeat every 15 s keeps middleboxes from reaping an idle connection.
//!
//! The route is rate-limit exempt (a long-lived stream must not drain its
//! address's token bucket) but pays the per-address SSE connection cap, which
//! binds loopback too: hub slots are a fixed resource. The address it keys on is
//! `client_addr`, so behind a trusted proxy each remote client holds its own
//! budget rather than all of them sharing the proxy's.

const std = @import("std");

const address = @import("../../platform/address.zig");
const http_util = @import("../http_util.zig");
const provenance_view = @import("../provenance_view.zig");
const server = @import("../server.zig");
const sse = @import("../sse.zig");

pub const retry_preamble = "retry: 3000\n\n";
pub const heartbeat = ": ping\n\n";

/// Ruling 20's heartbeat cadence. Awake clock: a suspended box owes no pings.
pub const heartbeat_interval: std.Io.Clock.Duration = .{
    .raw = .fromSeconds(15),
    .clock = .awake,
};

/// One event's `data:` payload: the shared full-provenance DTO, exactly. A live
/// event says everything `GET /api/queries/{id}` would say about the same query
/// except its id, which does not exist yet — the entry precedes its own insert.
pub const EventView = provenance_view.Provenance;

pub fn view(entry: *const sse.Entry) EventView {
    return provenance_view.fromEntry(entry);
}

/// One `event: query` frame. JSON never contains a raw newline, so the whole
/// payload is a single `data:` line.
pub fn writeEvent(w: *std.Io.Writer, entry: *const sse.Entry) std.Io.Writer.Error!void {
    try w.writeAll("event: query\ndata: ");
    var stringify: std.json.Stringify = .{ .writer = w };
    try stringify.write(view(entry));
    try w.writeAll("\n\n");
}

pub fn stream(
    state: *server.WebState,
    io: std.Io,
    request: *http_util.Request,
) http_util.HandlerError!void {
    const hub = state.hub orelse
        return http_util.respondError(request, .service_unavailable, "live stream unavailable");

    // The effective client, not the socket peer: behind a trusted proxy every
    // stream would otherwise share one address's budget. Acquire and release
    // read the same value, so a release can never miss the slot it took.
    const client = address.NetAddress.fromIp(request.client_addr);
    if (state.limiter) |limiter| {
        if (!limiter.tryAcquireSse(io, std.Io.Clock.awake.now(io), client))
            return http_util.respondError(request, .too_many_requests, "too many live streams from this address");
    }
    defer if (state.limiter) |limiter| limiter.releaseSse(io, client);

    const id = hub.subscribe(io) orelse
        return http_util.respondError(request, .service_unavailable, "live stream is full");
    defer hub.unsubscribe(io, id);

    var response = try request.http.respondStreaming(&.{}, .{
        .respond_options = .{
            .extra_headers = &.{
                .{ .name = "content-type", .value = "text/event-stream" },
                .{ .name = "cache-control", .value = "no-store" },
            },
        },
    });
    const w = &response.writer;
    try w.writeAll(retry_preamble);
    // The browser acts on the headers, not the first event; send them now.
    try response.flush();

    while (true) {
        while (hub.next(io, id)) |entry| try writeEvent(w, &entry);
        try response.flush();

        // Checked after the drain: entries that predate the overflow still
        // reach the client before the stream ends.
        if (hub.overflowed(io, id)) break;

        const wake = hub.wait(io, id, heartbeat_interval) catch return;
        switch (wake) {
            // Ruling 11 of milestone 16: the server is shutting down. Returning
            // without `end` leaves the response unterminated, which is what a
            // shutdown is; the client reconnects or gives up on its own.
            .closed => return,
            .timeout => {
                try w.writeAll(heartbeat);
                try response.flush();
            },
            .ready => {},
        }
    }

    try response.end();
}

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

const testing = std.testing;

test "the event payload is the detail body minus its id, name and type for name" {
    const detail_fields = @typeInfo(provenance_view.QueryDetail).@"struct".fields;
    const view_fields = @typeInfo(EventView).@"struct".fields;
    comptime {
        std.debug.assert(view_fields.len == detail_fields.len - 1);
        std.debug.assert(std.mem.eql(u8, detail_fields[0].name, "id"));
        for (detail_fields[1..], view_fields) |detail_field, view_field| {
            std.debug.assert(std.mem.eql(u8, detail_field.name, view_field.name));
            // Names alone would let a group keep its key while changing what it
            // holds, which is the drift a live viewer would see and a detail
            // page would not.
            std.debug.assert(detail_field.type == view_field.type);
        }
    }
}

test "a frame is one event line and one data line of JSON" {
    const entry: sse.Entry = .init(.{
        .timestamp = 1_700_000_000,
        .domain = "ads.example",
        .client_ip = "192.0.2.10",
        .qtype = 1,
        .qclass = 1,
        .rcode = 0,
        .blocked = true,
        .group_id = 1,
        .group_name = "default",
        .policy_action = .block,
        .policy_reason = .blocklist_domain,
        .matched = "ads.example",
        .source_id = 3,
        .source_name = "StevenBlack",
        .route_kind = .blocked,
        .response_time_us = 42,
        .cache_hit = false,
    });

    var buf: [1024]u8 = undefined;
    var writer: std.Io.Writer = .fixed(&buf);
    try writeEvent(&writer, &entry);
    const frame = writer.buffered();

    try testing.expect(std.mem.startsWith(u8, frame, "event: query\ndata: {"));
    try testing.expect(std.mem.endsWith(u8, frame, "}\n\n"));
    try testing.expectEqual(@as(usize, 3), std.mem.count(u8, frame, "\n"));
    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"time\":1700000000"));
    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"domain\":\"ads.example\""));
    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"group\":{\"id\":1,\"name\":\"default\"}"));
    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"reason\":\"blocklist_domain\""));
    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"source_name\":\"StevenBlack\""));
    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"kind\":\"blocked\""));
}

test "an unlogged field stays null and an empty string stays a string" {
    const entry: sse.Entry = .init(.{
        .timestamp = 1,
        .domain = "safe.example",
        .client_ip = "192.0.2.11",
    });

    var buf: [1024]u8 = undefined;
    var writer: std.Io.Writer = .fixed(&buf);
    try writeEvent(&writer, &entry);
    const frame = writer.buffered();

    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"qtype\":null"));
    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"duration_us\":null"));
    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"upstream\":\"\""));
    try testing.expect(std.mem.containsAtLeast(u8, frame, 1, "\"id\":null"));
}