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 sse = @import("../web/sse.zig");

pub const QuerySink = struct {
    logger: *logger.Logger,
    /// 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(query_logger: *logger.Logger, hub: ?*sse.Hub) QuerySink {
        return .{ .logger = query_logger, .hub = hub };
    }

    /// Transforms once, publishes, then enqueues. Never blocks the query path
    /// and never fails: both consumers drop rather than wait.
    pub fn log(self: *QuerySink, io: std.Io, entry: logger.Entry) void {
        const transformed = self.logger.transformed(entry);
        if (self.hub) |hub| hub.publish(io, transformed);
        self.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 sink: QuerySink = .init(&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 sink: QuerySink = .init(&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 sink: QuerySink = .init(&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 sink: QuerySink = .init(&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));
}