nxdns/src/server/query_sink.zig

//! Where the query path hands off a finished query (PLAN §11.4).
//!
//! Milestone 6 gave the handler a `Logger`; milestone 8 gives it a second
//! consumer, the SSE hub. `QuerySink` is that fanout, and it exists so the
//! handler still makes one call and the privacy transforms still run exactly
//! once, before either consumer sees the entry.
//!
//! Order is load-bearing: PLAN:455 puts fanout ahead of persistence, so a live
//! stream shows a query while the row is still queued for the database.
//! `Hub.publish` copies and returns, so publishing first costs the query path
//! nothing it would not have paid anyway.

const std = @import("std");

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

pub const QuerySink = struct {
    /// The controller, not a `Logger`: `logging.query_log_buffer_max` can
    /// change while the server runs, and the generation a producer enqueues
    /// into has to be the one that is live at that moment.
    controller: *logger_controller.Controller,
    /// Null when `web.enabled` is false: nothing subscribes, so nothing needs
    /// a hub, and the DNS path pays one null check.
    hub: ?*sse.Hub,

    pub fn init(controller: *logger_controller.Controller, hub: ?*sse.Hub) QuerySink {
        return .{ .controller = controller, .hub = hub };
    }

    /// Transforms once, publishes, then enqueues. Never blocks the query path
    /// and never fails: both consumers drop rather than wait.
    ///
    /// The borrow spans both halves. A resize that lands between them would
    /// otherwise leave this entry going into a queue retirement has already
    /// closed, and that is exactly the drop window the controller exists to
    /// make impossible.
    pub fn log(self: *QuerySink, io: std.Io, entry: logger.Entry) void {
        const generation = self.controller.acquire(io);
        defer self.controller.release(io, generation);

        const transformed = generation.logger.transformed(entry);
        if (self.hub) |hub| hub.publish(io, transformed);
        generation.logger.logTransformed(io, transformed);
    }
};

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

const testing = std.testing;

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

test "the sink publishes and logs the same entry" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try testing.allocator.create(sse.Hub);
    defer testing.allocator.destroy(hub);
    hub.init();

    var queue_buf: [4]logger.Entry = undefined;
    var query_logger: logger.Logger = .init(.{}, &queue_buf);
    var owner: logger_controller.Borrowed = .{};
    var sink: QuerySink = .init(owner.over(&query_logger), hub);

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

    sink.log(io, sampleEntry(11, "example.com"));

    const streamed = hub.next(io, id).?;
    try testing.expectEqualStrings("example.com", streamed.domain());
    try testing.expectEqual(@as(i64, 11), streamed.timestamp);

    const queued = try query_logger.queue.getOne(io);
    try testing.expectEqualStrings("example.com", queued.domain());
    try testing.expectEqual(@as(i64, 11), queued.timestamp);
}

test "fanout does not depend on the entry reaching the queue" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try testing.allocator.create(sse.Hub);
    defer testing.allocator.destroy(hub);
    hub.init();

    var queue_buf: [4]logger.Entry = undefined;
    var query_logger: logger.Logger = .init(.{}, &queue_buf);
    var owner: logger_controller.Borrowed = .{};
    var sink: QuerySink = .init(owner.over(&query_logger), hub);

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

    // A closed queue drops what it is handed. The stream still carries the
    // query, which is only true because the publish happens first.
    query_logger.shutdown(io);
    sink.log(io, sampleEntry(3, "ordered.example"));

    try testing.expectEqualStrings("ordered.example", hub.next(io, id).?.domain());
    try testing.expectEqual(@as(u64, 1), query_logger.queries_dropped.load(.monotonic));
}

test "the privacy transforms run once, before both consumers" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const hub = try testing.allocator.create(sse.Hub);
    defer testing.allocator.destroy(hub);
    hub.init();

    var queue_buf: [4]logger.Entry = undefined;
    var query_logger: logger.Logger = .init(
        .{ .hide_domains = true, .hide_client_ips = true },
        &queue_buf,
    );
    var owner: logger_controller.Borrowed = .{};
    var sink: QuerySink = .init(owner.over(&query_logger), hub);

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

    sink.log(io, sampleEntry(4, "tracker.example"));

    const streamed = hub.next(io, id).?;
    try testing.expectEqualStrings(logger.hidden_marker, streamed.domain());
    try testing.expectEqualStrings(logger.hidden_marker, streamed.clientIp());

    const queued = try query_logger.queue.getOne(io);
    try testing.expectEqualStrings(logger.hidden_marker, queued.domain());
    try testing.expectEqualStrings(logger.hidden_marker, queued.clientIp());
}

test "a sink without a hub still logs" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var queue_buf: [4]logger.Entry = undefined;
    var query_logger: logger.Logger = .init(.{}, &queue_buf);
    var owner: logger_controller.Borrowed = .{};
    var sink: QuerySink = .init(owner.over(&query_logger), null);

    sink.log(io, sampleEntry(5, "nohub.example"));

    const queued = try query_logger.queue.getOne(io);
    try testing.expectEqualStrings("nohub.example", queued.domain());
    try testing.expectEqual(@as(u64, 0), query_logger.queries_dropped.load(.monotonic));
}