nxdns/src/storage/logger.zig

//! Async query logger (PLAN §11.4). The query path hands an `Entry` to `log`
//! and never touches the database: one writer task owns the `db.Db` handle, and
//! everything between the two is an `std.Io.Queue`.
//!
//! `Io.Queue` copies elements as raw bytes (`Io.zig:2189`), so an `Entry` owns
//! every byte it carries — a slice into the caller's packet buffer would dangle
//! the moment the query finishes. That is the whole reason this file has fixed
//! buffers instead of slices.
//!
//! The privacy transforms of §11.4 run in `transformed`, before the entry is
//! enqueued, so nothing downstream — the database or the event stream — can
//! observe a value the operator asked to hide. `log` is the two halves in
//! order; `QuerySink` calls them separately so both of its consumers see the
//! one transformed entry.
//!
//! Log rows are expendable. A full queue drops the oldest unflushed entry, a
//! failed batch is dropped whole, and a disk that crossed the critical
//! threshold holds batches back indefinitely. Each of the three has a counter.
//! A writer that cannot prepare its statements closes the queue and marks
//! `writer_failed`, so the loss is visible rather than silent.

const std = @import("std");

const db = @import("db.zig");
const disk_monitor = @import("disk_monitor.zig");
const events = @import("events.zig");
const model = @import("../config/model.zig");
const queries_repo = @import("repositories/queries_repo.zig");

/// Named `scope` rather than `log`: `Logger.log` is the enqueue entry point,
/// and the two names collide inside the struct.
const scope = std.log.scoped(.query_logger);

/// Flush tuning is comptime: §12.1 defines no configuration keys for it and a
/// household deployment has no reason to tune it.
pub const flush_batch = 100;
pub const flush_interval_ms = 100;

/// What `hide_domains` and `hide_client_ips` store instead of the real value.
pub const hidden_marker = "hidden";

/// How long a batch waits before it re-reads the disk monitor.
pub const gate_retry_s = 1;

/// The widths of the query log's text columns, and the single source of them:
/// every producer that formats into one of these fields sizes its own buffer
/// from the constant here, so nothing can format wider than the row stores.
pub const max_domain_len = 253;
/// RFC 5952 text of any IPv6 address, zone identifier included.
pub const max_client_len = 45;
pub const max_reason_len = 32;
pub const max_upstream_len = 64;

/// One row on its way to `query_log`, carrying its own bytes.
pub const Entry = struct {
    timestamp: i64,
    domain_buf: [max_domain_len]u8,
    domain_len: u8,
    client_buf: [max_client_len]u8,
    client_len: u8,
    qtype: ?u16,
    blocked: bool,
    reason_buf: [max_reason_len]u8,
    reason_len: u8,
    response_time_us: ?i64,
    cache_hit: ?bool,
    upstream_buf: [max_upstream_len]u8,
    upstream_len: u8,

    /// The borrowed shape of an entry. `init` copies out of it, so a caller can
    /// build one from slices that die with the query.
    pub const Fields = struct {
        timestamp: i64,
        domain: []const u8,
        client_ip: []const u8,
        qtype: ?u16 = null,
        blocked: bool = false,
        /// Empty means "no reason", which reaches the database as NULL.
        block_reason: []const u8 = "",
        response_time_us: ?i64 = null,
        cache_hit: ?bool = null,
        /// Empty means "no upstream", which reaches the database as NULL.
        upstream: []const u8 = "",
    };

    /// Copies each string in, truncated to what its buffer holds. A name longer
    /// than 253 bytes is not a valid domain name, so truncation here means the
    /// caller skipped the parser, not that a real name was lost.
    pub fn init(f: Fields) Entry {
        var entry: Entry = .{
            .timestamp = f.timestamp,
            .domain_buf = undefined,
            .domain_len = 0,
            .client_buf = undefined,
            .client_len = 0,
            .qtype = f.qtype,
            .blocked = f.blocked,
            .reason_buf = undefined,
            .reason_len = 0,
            .response_time_us = f.response_time_us,
            .cache_hit = f.cache_hit,
            .upstream_buf = undefined,
            .upstream_len = 0,
        };
        entry.setDomain(f.domain);
        entry.setClientIp(f.client_ip);
        entry.reason_len = copyInto(&entry.reason_buf, f.block_reason);
        entry.upstream_len = copyInto(&entry.upstream_buf, f.upstream);
        return entry;
    }

    pub fn setDomain(self: *Entry, value: []const u8) void {
        self.domain_len = copyInto(&self.domain_buf, value);
    }

    pub fn setClientIp(self: *Entry, value: []const u8) void {
        self.client_len = copyInto(&self.client_buf, value);
    }

    pub fn domain(self: *const Entry) []const u8 {
        return self.domain_buf[0..self.domain_len];
    }

    pub fn clientIp(self: *const Entry) []const u8 {
        return self.client_buf[0..self.client_len];
    }

    pub fn blockReason(self: *const Entry) []const u8 {
        return self.reason_buf[0..self.reason_len];
    }

    pub fn upstream(self: *const Entry) []const u8 {
        return self.upstream_buf[0..self.upstream_len];
    }
};

fn copyInto(buf: []u8, value: []const u8) u8 {
    const n = @min(buf.len, value.len);
    @memcpy(buf[0..n], value[0..n]);
    return @intCast(n);
}

/// The row borrows from `entry`, which must outlive the `writeBatch` call.
fn toRow(entry: *const Entry) queries_repo.Row {
    return .{
        .timestamp = entry.timestamp,
        .domain = entry.domain(),
        .client_ip = entry.clientIp(),
        .qtype = entry.qtype,
        .blocked = entry.blocked,
        .block_reason = emptyAsNull(entry.blockReason()),
        .response_time_us = entry.response_time_us,
        .cache_hit = entry.cache_hit,
        .upstream = emptyAsNull(entry.upstream()),
    };
}

fn emptyAsNull(value: []const u8) ?[]const u8 {
    return if (value.len == 0) null else value;
}

const EntryQueue = std.Io.Queue(Entry);

/// What the flush interval race can produce. `Select` demands that each field
/// type match its task's return type exactly.
const Outcome = union(enum) {
    entry: std.Io.Cancelable!?Entry,
    expiry: std.Io.Cancelable!void,
};

pub const Logger = struct {
    cfg: model.Logging,
    queue: EntryQueue,
    queries_dropped: std.atomic.Value(u64),
    rows_written: std.atomic.Value(u64),
    batches_gated: std.atomic.Value(u64),
    /// Set when `runWriter` gives up before it consumed anything. The queue is
    /// closed and every entry counts as dropped from that point, so a caller
    /// that sees this must not expect rows.
    writer_failed: std.atomic.Value(bool),
    /// Wired by the composition root after `init`, following the
    /// `gate: ?*disk_monitor.Monitor` idiom. Null in every unit test here.
    diagnostics: ?*events.Store = null,

    /// `queue_buf.len` is the backpressure cap — the composition root
    /// (`app.zig:311`) allocates `cfg.logging.query_log_buffer_max` entries,
    /// which `config/validate.zig` bounds. The queue holds waiting tasks in
    /// intrusive lists, so a `Logger` must not be moved once anything has
    /// touched it.
    pub fn init(cfg: model.Logging, queue_buf: []Entry) Logger {
        return .{
            .cfg = cfg,
            .queue = .init(queue_buf),
            .queries_dropped = .init(0),
            .rows_written = .init(0),
            .batches_gated = .init(0),
            .writer_failed = .init(false),
        };
    }

    /// Applies the privacy transforms and enqueues without ever blocking the
    /// query path. A full queue loses its oldest unflushed entry (§11.4).
    pub fn log(self: *Logger, io: std.Io, entry: Entry) void {
        self.logTransformed(io, self.transformed(entry));
    }

    /// The §11.4 privacy transforms, on their own. `QuerySink` runs them once
    /// and hands the result to every consumer, so nothing downstream — the
    /// database or the event stream — can observe a value the operator asked
    /// to hide.
    pub fn transformed(self: *const Logger, entry: Entry) Entry {
        var out = entry;
        if (self.cfg.hide_domains) out.setDomain(hidden_marker);
        if (self.cfg.hide_client_ips) out.setClientIp(hidden_marker);
        return out;
    }

    /// `log` without the transforms, for a caller that already applied them.
    pub fn logTransformed(self: *Logger, io: std.Io, entry: Entry) void {
        self.enqueue(io, entry);
    }

    /// Retries until the put succeeds, and each failed attempt drops exactly
    /// one oldest entry. A fixed attempt cap would break the policy under
    /// contention: a producer that steals the slot this call freed would make
    /// this call pay for two entries, the dropped one and its own.
    fn enqueue(self: *Logger, io: std.Io, entry: Entry) void {
        while (true) {
            // A closed queue or a canceled task means shutdown is underway;
            // both leave this entry unwritten, which is what the counter says.
            const put = self.queue.put(io, &.{entry}, 0) catch break;
            if (put == 1) return;

            // A zero-capacity queue holds nothing to drop: the put above was
            // this entry's one chance at a waiting getter.
            if (self.queue.capacity() == 0) break;

            var oldest: [1]Entry = undefined;
            const got = self.queue.get(io, &oldest, 0) catch break;
            if (got == 1) self.countDropped(1);
        }
        self.countDropped(1);
    }

    /// The writer task: owns `database` and its prepared statements for its
    /// whole life. Returns when `shutdown` closes the queue and the last batch
    /// is flushed, or when the task is canceled.
    ///
    /// `monitor` is the §11.6 gate. Null disables gating.
    pub fn runWriter(
        self: *Logger,
        io: std.Io,
        database: *db.Db,
        monitor: ?*disk_monitor.Monitor,
    ) std.Io.Cancelable!void {
        var writer = queries_repo.BatchWriter.init(database) catch |err| {
            scope.warn("query logger: preparing the batch statements failed: {s}", .{@errorName(err)});
            // Without a writer there is no consumer, so leaving the queue open
            // would silently swallow every later entry.
            self.writer_failed.store(true, .release);
            // No recovery path claims this episode: the writer is gone for the
            // life of the process, so the row stays active, which is the truth.
            self.reportWrite(io, "writer", "preparing the batch statements failed", @errorName(err), 0);
            self.queue.close(io);
            self.dropRemaining(io);
            return;
        };
        defer writer.deinit();

        var batch: [flush_batch]Entry = undefined;
        while (true) {
            // A closed queue hands over its buffered elements before it reports
            // `Closed` (`Io.zig:2118`), so this drains before it returns.
            batch[0] = self.queue.getOne(io) catch |err| switch (err) {
                error.Closed => return,
                error.Canceled => |e| return e,
            };
            const deadline: std.Io.Clock.Timestamp = .fromNow(io, .{
                .raw = .fromMilliseconds(flush_interval_ms),
                .clock = .awake,
            });
            // `n` is live across both calls: entries already taken off the
            // queue are lost if either one is canceled, so they must count.
            var n: usize = 1;
            self.fill(io, &batch, deadline, &n) catch |err| {
                self.countDropped(n);
                return err;
            };
            self.flush(io, &writer, batch[0..n], monitor) catch |err| {
                self.countDropped(n);
                return err;
            };
        }
    }

    /// Counts every entry left in a closed queue as dropped. The drain is
    /// uncancelable: a cancellation racing the writer's own failure would
    /// otherwise abandon the buffered entries without counting them.
    fn dropRemaining(self: *Logger, io: std.Io) void {
        var leftover: [flush_batch]Entry = undefined;
        while (true) {
            const n = self.queue.getUncancelable(io, &leftover, 0) catch |err| switch (err) {
                error.Closed => break,
            };
            if (n == 0) break;
            self.countDropped(n);
        }
    }

    /// Closes the queue. `log` drops from here on and `runWriter` returns once
    /// it has flushed what was left.
    ///
    /// A writer held by the disk gate keeps holding: it flushes when the disk
    /// recovers, and the `group.cancel` that follows this call in `app.zig`
    /// stops a writer that will not wait. A canceled
    /// writer counts the batch it holds under `queries_dropped`.
    pub fn shutdown(self: *Logger, io: std.Io) void {
        self.queue.close(io);
    }

    /// Fills `batch` behind the entry already in slot 0, until it is full or
    /// `deadline` passes. `n` counts the slots that hold an entry, and stays
    /// accurate on the cancellation path so the caller can count what is lost.
    fn fill(
        self: *Logger,
        io: std.Io,
        batch: *[flush_batch]Entry,
        deadline: std.Io.Clock.Timestamp,
        n: *usize,
    ) std.Io.Cancelable!void {
        n.* += self.drainAvailable(io, batch[n.*..]);

        while (n.* < batch.len) {
            const remaining = deadline.durationFromNow(io);
            if (remaining.raw.nanoseconds <= 0) break;
            const entry = try self.getWithin(io, remaining) orelse break;
            batch[n.*] = entry;
            n.* += 1;
            n.* += self.drainAvailable(io, batch[n.*..]);
        }
    }

    /// Whatever is already queued, without blocking.
    fn drainAvailable(self: *Logger, io: std.Io, room: []Entry) usize {
        if (room.len == 0) return 0;
        return self.queue.get(io, room, 0) catch 0;
    }

    /// Races one blocking `getOne` against the rest of the flush interval —
    /// `std.Io.Condition` has no timed wait, so the timer is a task.
    ///
    /// The loser is drained rather than discarded: a `getOne` that finishes
    /// just after the timer has already taken an entry off the queue, and
    /// `Select.cancelDiscard` would throw that entry away.
    fn getWithin(
        self: *Logger,
        io: std.Io,
        budget: std.Io.Clock.Duration,
    ) std.Io.Cancelable!?Entry {
        var outcomes: [2]Outcome = undefined;
        var race: std.Io.Select(Outcome) = .init(io, &outcomes);

        race.concurrent(.entry, takeOne, .{ &self.queue, io }) catch |err| switch (err) {
            // No second unit of concurrency: the caller flushes what it holds
            // rather than block past the interval.
            error.ConcurrencyUnavailable => return null,
        };
        race.concurrent(.expiry, expire, .{ io, budget }) catch |err| switch (err) {
            error.ConcurrencyUnavailable => return drainRace(&race),
        };

        const first = race.await() catch |err| {
            // Teardown: the entry the getter already took has nowhere to go.
            if (drainRace(&race)) |_| self.countDropped(1);
            return err;
        };
        const late = drainRace(&race);
        return outcomeEntry(first) orelse late;
    }

    /// One batch, one transaction. A batch is dropped whole on a database
    /// failure: these are log rows, and blocking on them would fill the queue
    /// and cost live queries instead.
    fn flush(
        self: *Logger,
        io: std.Io,
        writer: *queries_repo.BatchWriter,
        entries: []const Entry,
        monitor: ?*disk_monitor.Monitor,
    ) std.Io.Cancelable!void {
        if (entries.len == 0) return;

        if (monitor) |m| {
            const pause: std.Io.Clock.Duration = .{
                .raw = .fromSeconds(gate_retry_s),
                .clock = .awake,
            };
            while (!m.writesAllowed()) {
                _ = self.batches_gated.fetchAdd(1, .monotonic);
                try pause.sleep(io);
            }
        }

        var rows: [flush_batch]queries_repo.Row = undefined;
        for (entries, rows[0..entries.len]) |*entry, *row| row.* = toRow(entry);

        writer.writeBatch(rows[0..entries.len]) catch |err| {
            scope.warn("query log batch of {d} rows dropped: {s}", .{ entries.len, @errorName(err) });
            self.countDropped(entries.len);
            self.reportWrite(io, "batch", "a query log batch was dropped", @errorName(err), entries.len);
            return;
        };
        _ = self.rows_written.fetchAdd(entries.len, .monotonic);
        if (self.diagnostics) |store| {
            store.resolve(io, std.Io.Clock.real.now(io).toSeconds(), .query_log_write, "batch");
        }
    }

    /// An error, not a warning: dropped query rows are gone, and a writer that
    /// never started means every later row is gone too.
    fn reportWrite(
        self: *Logger,
        io: std.Io,
        operation: []const u8,
        message: []const u8,
        error_name: []const u8,
        rows: usize,
    ) void {
        const store = self.diagnostics orelse return;
        var buf: [events.Store.max_detail_len]u8 = undefined;
        const detail = std.fmt.bufPrint(&buf, "{s}: {s} ({d} rows)", .{
            message,
            error_name,
            rows,
        }) catch buf[0..];
        store.report(
            io,
            std.Io.Clock.real.now(io).toSeconds(),
            .query_log_write,
            operation,
            operation,
            .@"error",
            detail,
        );
    }

    fn countDropped(self: *Logger, n: usize) void {
        _ = self.queries_dropped.fetchAdd(n, .monotonic);
    }
};

fn takeOne(queue: *EntryQueue, io: std.Io) std.Io.Cancelable!?Entry {
    const entry = queue.getOne(io) catch |err| switch (err) {
        error.Closed => return null,
        error.Canceled => |e| return e,
    };
    return entry;
}

fn drainRace(race: *std.Io.Select(Outcome)) ?Entry {
    var found: ?Entry = null;
    while (race.cancel()) |outcome| {
        if (outcomeEntry(outcome)) |entry| found = entry;
    }
    return found;
}

fn expire(io: std.Io, budget: std.Io.Clock.Duration) std.Io.Cancelable!void {
    return budget.sleep(io);
}

fn outcomeEntry(outcome: Outcome) ?Entry {
    return switch (outcome) {
        .entry => |result| result catch null,
        .expiry => null,
    };
}

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

const events_fixture = @import("events_fixture.zig");
const querylog_schema = @import("querylog_schema.zig");

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,
        .blocked = false,
        .response_time_us = 900,
        .cache_hit = false,
        .upstream = "9.9.9.9",
    });
}

fn openLog() !db.Db {
    var database = try db.Db.open(":memory:", .{ .mode = .memory });
    errdefer database.close();
    try db.applyPragmas(&database, .{});
    try database.exec(querylog_schema.ddl);
    return database;
}

test "an entry carries its own bytes and reads them back" {
    const entry: Entry = .init(.{
        .timestamp = 1700000000,
        .domain = "ads.example.com",
        .client_ip = "2001:db8::1",
        .qtype = 28,
        .blocked = true,
        .block_reason = "blocklist",
        .response_time_us = 42,
        .cache_hit = true,
        .upstream = "dns.example",
    });

    try testing.expectEqualStrings("ads.example.com", entry.domain());
    try testing.expectEqualStrings("2001:db8::1", entry.clientIp());
    try testing.expectEqualStrings("blocklist", entry.blockReason());
    try testing.expectEqualStrings("dns.example", entry.upstream());
    try testing.expectEqual(@as(?u16, 28), entry.qtype);
    try testing.expect(entry.blocked);
    try testing.expectEqual(@as(?i64, 42), entry.response_time_us);
    try testing.expectEqual(@as(?bool, true), entry.cache_hit);
}

test "an oversize string is truncated to what its buffer holds" {
    const long_domain = "a" ** 400;
    const entry: Entry = .init(.{
        .timestamp = 1,
        .domain = long_domain,
        .client_ip = "192.0.2.1",
        .block_reason = "r" ** 64,
        .upstream = "u" ** 128,
    });

    try testing.expectEqual(@as(usize, max_domain_len), entry.domain().len);
    try testing.expectEqual(@as(usize, max_reason_len), entry.blockReason().len);
    try testing.expectEqual(@as(usize, max_upstream_len), entry.upstream().len);
    try testing.expectEqualStrings("a" ** max_domain_len, entry.domain());
}

test "toRow maps the empty strings to null and passes the rest through" {
    const bare: Entry = .init(.{
        .timestamp = 7,
        .domain = "example.com",
        .client_ip = "192.0.2.5",
    });
    const bare_row = toRow(&bare);
    try testing.expectEqual(@as(i64, 7), bare_row.timestamp);
    try testing.expectEqualStrings("example.com", bare_row.domain);
    try testing.expectEqualStrings("192.0.2.5", bare_row.client_ip);
    try testing.expectEqual(@as(?[]const u8, null), bare_row.block_reason);
    try testing.expectEqual(@as(?[]const u8, null), bare_row.upstream);
    try testing.expectEqual(@as(?u16, null), bare_row.qtype);
    try testing.expectEqual(@as(?bool, null), bare_row.cache_hit);

    const full: Entry = .init(.{
        .timestamp = 8,
        .domain = "blocked.example",
        .client_ip = "192.0.2.6",
        .blocked = true,
        .block_reason = "blocklist",
        .upstream = "9.9.9.9",
    });
    const full_row = toRow(&full);
    try testing.expect(full_row.blocked);
    try testing.expectEqualStrings("blocklist", full_row.block_reason.?);
    try testing.expectEqualStrings("9.9.9.9", full_row.upstream.?);
}

test "log applies both privacy transforms before the entry reaches the queue" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var buf: [4]Entry = undefined;
    var logger: Logger = .init(.{ .hide_domains = true, .hide_client_ips = true }, &buf);

    logger.log(io, sampleEntry(100, "tracker.example"));

    const queued = try logger.queue.getOne(io);
    try testing.expectEqualStrings(hidden_marker, queued.domain());
    try testing.expectEqualStrings(hidden_marker, queued.clientIp());
    try testing.expectEqual(@as(i64, 100), queued.timestamp);
    try testing.expectEqual(@as(u64, 0), logger.queries_dropped.load(.monotonic));
}

test "log hides only the field its switch names" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var buf: [4]Entry = undefined;
    var domains_only: Logger = .init(.{ .hide_domains = true }, &buf);
    domains_only.log(io, sampleEntry(1, "tracker.example"));
    const hidden_domain = try domains_only.queue.getOne(io);
    try testing.expectEqualStrings(hidden_marker, hidden_domain.domain());
    try testing.expectEqualStrings("192.0.2.10", hidden_domain.clientIp());

    var clients_only: Logger = .init(.{ .hide_client_ips = true }, &buf);
    clients_only.log(io, sampleEntry(2, "tracker.example"));
    const hidden_client = try clients_only.queue.getOne(io);
    try testing.expectEqualStrings("tracker.example", hidden_client.domain());
    try testing.expectEqualStrings(hidden_marker, hidden_client.clientIp());

    var neither: Logger = .init(.{}, &buf);
    neither.log(io, sampleEntry(3, "tracker.example"));
    const untouched = try neither.queue.getOne(io);
    try testing.expectEqualStrings("tracker.example", untouched.domain());
    try testing.expectEqualStrings("192.0.2.10", untouched.clientIp());
}

test "the split halves reproduce log byte for byte" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const configs = [_]model.Logging{
        .{},
        .{ .hide_domains = true },
        .{ .hide_client_ips = true },
        .{ .hide_domains = true, .hide_client_ips = true },
    };

    for (configs) |cfg| {
        var buf: [4]Entry = undefined;
        var logger: Logger = .init(cfg, &buf);
        const source = sampleEntry(100, "tracker.example");

        logger.log(io, source);
        logger.logTransformed(io, logger.transformed(source));

        const from_log = try logger.queue.getOne(io);
        const from_halves = try logger.queue.getOne(io);
        try testing.expectEqualSlices(
            u8,
            std.mem.asBytes(&from_log),
            std.mem.asBytes(&from_halves),
        );
    }
}

test "a full queue drops the oldest entry and counts it" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var buf: [2]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    logger.log(io, sampleEntry(1, "first.example"));
    logger.log(io, sampleEntry(2, "second.example"));
    logger.log(io, sampleEntry(3, "third.example"));

    try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));

    const older = try logger.queue.getOne(io);
    const newer = try logger.queue.getOne(io);
    try testing.expectEqualStrings("second.example", older.domain());
    try testing.expectEqualStrings("third.example", newer.domain());
}

test "a zero-capacity queue drops every entry exactly once" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var buf: [0]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    for (0..5) |i| logger.log(io, sampleEntry(@intCast(i), "example.com"));
    try testing.expectEqual(@as(u64, 5), logger.queries_dropped.load(.monotonic));
}

test "log after shutdown drops instead of blocking" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var buf: [4]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);
    logger.shutdown(io);

    logger.log(io, sampleEntry(1, "example.com"));
    try testing.expectEqual(@as(u64, 1), logger.queries_dropped.load(.monotonic));
}

test "the writer drains every entry and shutdown ends it" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();

    var buf: [512]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    var future = try io.concurrent(Logger.runWriter, .{
        &logger,
        io,
        &database,
        @as(?*disk_monitor.Monitor, null),
    });

    var names: [250][32]u8 = undefined;
    for (&names, 0..) |*name, i| {
        const written = try std.fmt.bufPrint(name, "d{d}.example", .{i % 10});
        logger.log(io, sampleEntry(@intCast(i), written));
    }
    logger.shutdown(io);
    try future.await(io);

    try testing.expectEqual(@as(u64, 0), logger.queries_dropped.load(.monotonic));
    try testing.expectEqual(@as(u64, 250), logger.rows_written.load(.monotonic));
    try testing.expectEqual(@as(i64, 250), try queries_repo.countRows(&database));
    try testing.expectEqual(@as(i64, 10), try queries_repo.countDomains(&database));
}

test "the writer flushes an entry once the interval passes" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();

    var buf: [8]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    var future = try io.concurrent(Logger.runWriter, .{
        &logger,
        io,
        &database,
        @as(?*disk_monitor.Monitor, null),
    });

    logger.log(io, sampleEntry(1, "only.example"));

    const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
    var waited: usize = 0;
    while (logger.rows_written.load(.monotonic) == 0) : (waited += 1) {
        // Ten times the interval; a flush that has not happened by then is a
        // failure, not slowness.
        try testing.expect(waited < 200);
        try poll.sleep(io);
    }

    logger.shutdown(io);
    try future.await(io);
    try testing.expectEqual(@as(i64, 1), try queries_repo.countRows(&database));
}

test "a gated flush holds the batch until the disk recovers" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();
    var writer = try queries_repo.BatchWriter.init(&database);
    defer writer.deinit();

    var buf: [4]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);
    try testing.expect(!monitor.writesAllowed());

    const entries = [_]Entry{ sampleEntry(1, "held.example"), sampleEntry(2, "held.example") };
    var future = try io.concurrent(Logger.flush, .{
        &logger,
        io,
        &writer,
        @as([]const Entry, &entries),
        @as(?*disk_monitor.Monitor, &monitor),
    });

    const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
    var waited: usize = 0;
    while (logger.batches_gated.load(.monotonic) == 0) : (waited += 1) {
        try testing.expect(waited < 200);
        try poll.sleep(io);
    }
    try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));

    monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
    try future.await(io);

    try testing.expect(logger.batches_gated.load(.monotonic) >= 1);
    try testing.expectEqual(@as(u64, 2), logger.rows_written.load(.monotonic));
    try testing.expectEqual(@as(i64, 2), try queries_repo.countRows(&database));
}

test "a failing batch is dropped whole and the writer stays usable" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();
    try database.exec(
        \\CREATE TRIGGER refuse_boom BEFORE INSERT ON query_log
        \\WHEN new.client_ip = 'boom'
        \\BEGIN SELECT RAISE(ABORT, 'refused'); END;
    );

    var writer = try queries_repo.BatchWriter.init(&database);
    defer writer.deinit();

    var buf: [4]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    var doomed = sampleEntry(10, "poison.example");
    doomed.setClientIp("boom");
    const bad = [_]Entry{ sampleEntry(9, "good.example"), doomed };
    try logger.flush(io, &writer, &bad, null);

    try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
    try testing.expectEqual(@as(u64, 0), logger.rows_written.load(.monotonic));
    try testing.expectEqual(@as(u64, 2), logger.queries_dropped.load(.monotonic));

    const good = [_]Entry{sampleEntry(11, "next.example")};
    try logger.flush(io, &writer, &good, null);
    try testing.expectEqual(@as(i64, 1), try queries_repo.countRows(&database));
    try testing.expectEqual(@as(u64, 1), logger.rows_written.load(.monotonic));
}

test "a writer that cannot prepare closes the queue and counts every entry" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // No schema: `BatchWriter.init` cannot prepare against a missing table.
    var database = try db.Db.open(":memory:", .{ .mode = .memory });
    defer database.close();

    var buf: [8]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    for (0..3) |i| logger.log(io, sampleEntry(@intCast(i), "early.example"));

    try logger.runWriter(io, &database, null);

    try testing.expect(logger.writer_failed.load(.acquire));
    try testing.expectEqual(@as(u64, 3), logger.queries_dropped.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), logger.rows_written.load(.monotonic));

    // The queue is closed, so later entries drop and count instead of piling up.
    logger.log(io, sampleEntry(99, "late.example"));
    try testing.expectEqual(@as(u64, 4), logger.queries_dropped.load(.monotonic));
}

test "a canceled writer counts the batch it was holding" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();

    var buf: [8]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);

    // Both entries are queued before the writer starts, so the batch it takes
    // into the gate holds exactly two.
    logger.log(io, sampleEntry(1, "held.example"));
    logger.log(io, sampleEntry(2, "held.example"));

    var future = try io.concurrent(Logger.runWriter, .{
        &logger,
        io,
        &database,
        @as(?*disk_monitor.Monitor, &monitor),
    });

    const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
    var waited: usize = 0;
    while (logger.batches_gated.load(.monotonic) == 0) : (waited += 1) {
        try testing.expect(waited < 400);
        try poll.sleep(io);
    }

    try testing.expectError(error.Canceled, future.cancel(io));

    try testing.expectEqual(@as(u64, 2), logger.queries_dropped.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), logger.rows_written.load(.monotonic));
    try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
}

test "an empty batch touches neither the database nor the counters" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();
    var writer = try queries_repo.BatchWriter.init(&database);
    defer writer.deinit();

    var buf: [4]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);

    var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);

    // Gated or not, an empty batch returns before it reads the monitor.
    try logger.flush(io, &writer, &.{}, &monitor);

    try testing.expectEqual(@as(u64, 0), logger.batches_gated.load(.monotonic));
    try testing.expectEqual(@as(u64, 0), logger.rows_written.load(.monotonic));
    try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(&database));
}

test "a dropped batch opens an error episode the next good batch closes" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openLog();
    defer database.close();
    try database.exec(
        \\CREATE TRIGGER refuse_boom BEFORE INSERT ON query_log
        \\WHEN new.client_ip = 'boom'
        \\BEGIN SELECT RAISE(ABORT, 'refused'); END;
    );

    var fx: events_fixture.Fixture = .{};
    try fx.init(io, 1000);
    defer fx.deinit();

    var writer = try queries_repo.BatchWriter.init(&database);
    defer writer.deinit();

    var buf: [4]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);
    logger.diagnostics = &fx.store;

    var doomed = sampleEntry(10, "poison.example");
    doomed.setClientIp("boom");
    const bad = [_]Entry{doomed};
    try logger.flush(io, &writer, &bad, null);

    try testing.expectEqualStrings("query_log.write", try fx.text("SELECT code FROM operational_events"));
    try testing.expectEqualStrings("batch", try fx.text("SELECT subject_key FROM operational_events"));
    try testing.expectEqualStrings("error", try fx.text("SELECT severity FROM operational_events"));

    const good = [_]Entry{sampleEntry(11, "next.example")};
    try logger.flush(io, &writer, &good, null);

    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT count(*) FROM operational_events"));
    try testing.expectEqual(
        @as(i64, 0),
        try fx.count("SELECT count(*) FROM operational_events WHERE resolved_at IS NULL"),
    );
}

test "a writer that cannot prepare leaves an episode no recovery path claims" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // No schema: `BatchWriter.init` cannot prepare against a missing table.
    var database = try db.Db.open(":memory:", .{ .mode = .memory });
    defer database.close();

    var fx: events_fixture.Fixture = .{};
    try fx.init(io, 1000);
    defer fx.deinit();

    var buf: [8]Entry = undefined;
    var logger: Logger = .init(.{}, &buf);
    logger.diagnostics = &fx.store;

    try logger.runWriter(io, &database, null);

    try testing.expectEqualStrings("writer", try fx.text(
        "SELECT subject_key FROM operational_events WHERE resolved_at IS NULL",
    ));
    try testing.expectEqualStrings("error", try fx.text(
        "SELECT severity FROM operational_events WHERE resolved_at IS NULL",
    ));

    // The writer returned, so nothing can ever close this. A second run finds
    // the queue closed and adds no second episode.
    try logger.runWriter(io, &database, null);
    try testing.expectEqual(
        @as(i64, 1),
        try fx.count("SELECT count(*) FROM operational_events WHERE resolved_at IS NULL"),
    );
}