//! The stable owner of the query logger (milestone 34 §S4).
//!
//! `logging.query_log_buffer_max` is the queue's capacity, and the queue is
//! allocated once at boot. Changing it means a new buffer, a new `Logger`, a
//! new writer task and a new database connection — a whole new generation. The
//! controller is what everything else holds instead of the `Logger`, so a
//! resize swaps the generation underneath `QuerySink`, `/metrics` and
//! `/api/health` without any of them learning that generations exist.
//!
//! Three invariants make that safe, and each has a mechanism:
//!
//! 1. **No producer ever enqueues into a closed queue.** A producer borrows the
//!    live generation through `acquire`/`release`; retirement waits for every
//!    borrow of the old generation to be released before it closes that queue.
//!    There is no resize drop window, and no resize-specific drop counter — if
//!    one were needed, this file would be wrong.
//! 2. **Publish cannot fail.** Everything fallible — the buffer, the `Logger`,
//!    the connection, the writer task and its prepared statements — happens in
//!    `prepare`, before the settings transaction commits. The new writer is
//!    spawned already parked on an activation gate, so `publish` is a pointer
//!    swap and one `Event.set`.
//! 3. **One task joins each writer.** `Future.await` is not thread-safe
//!    (`Io.zig:1198`), so the reaper — spawned once at construction — is the
//!    sole joiner of retired writers, and the shutdown path is the sole joiner
//!    of the live one. At most one retired generation exists at a time, which
//!    is what bounds writers, connections and buffers; a second resize while
//!    one is still draining is refused rather than queued.
//!
//! What must stay continuous across a swap lives here rather than in a
//! generation: the counters, the disk-gate episode and the drop stamp are
//! summed and merged across the outstanding generations, and a retired
//! generation's finals fold into `base` the moment its writer is joined, so no
//! reader ever sees a counter go backwards. `writer_failed` is the exception
//! and reports the LIVE writer only — a successful resize means a writer that
//! prepared cleanly is now serving, which is the truth the health endpoint
//! needs; the rows a failed retired writer lost are already in the drop count.

const std = @import("std");
const Allocator = std.mem.Allocator;

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

const scope = std.log.scoped(.query_logger);

/// Where a generation's writer connection comes from. The controller opens one
/// per generation and closes it after that generation's writer is joined: two
/// writers must never share a `db.Db`.
pub const Source = struct {
    dir: std.Io.Dir,
    path: [:0]const u8,
};

/// The cumulative totals of every generation whose writer has been joined.
/// Live generations are added to these on every read.
const Totals = struct {
    queries_dropped: u64 = 0,
    rows_written: u64 = 0,
    batches_gated: u64 = 0,
    last_drop_s: i64 = 0,

    fn fold(self: *Totals, lg: *const logger.Logger) void {
        self.queries_dropped += lg.queries_dropped.load(.monotonic);
        self.rows_written += lg.rows_written.load(.monotonic);
        self.batches_gated += lg.batches_gated.load(.monotonic);
        self.last_drop_s = @max(self.last_drop_s, lg.last_drop_s.load(.monotonic));
    }
};

/// What a facade reader sees: one coherent reading of every outstanding
/// generation, taken under the controller's mutex.
pub const Sample = struct {
    queries_dropped: u64,
    rows_written: u64,
    batches_gated: u64,
    /// Null while nothing has been dropped, exactly as `Logger.lastDropSeconds`.
    last_drop_s: ?i64,
    /// The LIVE writer's state. A retired writer's failures are in the counters.
    writer_failed: bool,
    /// The worst episode any outstanding generation is in. A retired writer
    /// still holding a batch against a full disk is a current fault whether or
    /// not the live one has noticed the disk yet.
    gate_episode: logger.GateEpisode,
};

/// One buffer, one `Logger`, one writer task and one connection, with the
/// producer refcount that decides when all four can go away.
///
/// Heap-allocated and never moved: the queue holds waiting tasks in intrusive
/// lists, and `logger` points into `owned`.
pub const Generation = struct {
    /// The producers' view. Points at `owned.state` for a generation this
    /// controller built, and at the caller's `Logger` for a borrowed one.
    logger: *logger.Logger,
    /// How many producers are inside `transformed`+`logTransformed` right now.
    /// Guarded by the controller's mutex.
    borrows: usize = 0,
    /// Set by `publish` under the controller's mutex.
    retired: bool = false,
    /// Null for a borrowed generation, which owns nothing and is never retired.
    owned: ?Owned = null,

    const Owned = struct {
        /// Pointed at by `Generation.logger`. Never moved.
        state: logger.Logger,
        queue_buf: []logger.Entry,
        database: db.Db,
        writer: std.Io.Future(std.Io.Cancelable!void),
        /// Set by the writer once its statements are prepared, or once
        /// preparing them has failed. `prepare` waits on it.
        ready: std.Io.Event = .unset,
        /// Opened by `publish`, or by `abandon` to release a writer that will
        /// never serve. The writer consumes nothing until one of the two.
        activation: std.Io.Event = .unset,
        /// What `BatchWriter.init` refused with, read after `ready`.
        prepare_error: ?db.Error = null,
        /// Set before `activation` when the candidate is thrown away.
        aborted: bool = false,
        capacity: u32,
    };

    fn deinit(self: *Generation, gpa: Allocator) void {
        if (self.owned) |*owned| {
            owned.database.close();
            gpa.free(owned.queue_buf);
        }
        gpa.destroy(self);
    }
};

pub const InitError = Allocator.Error || std.Io.ConcurrentError;

pub const Options = struct {
    gpa: Allocator,
    /// The connection `cli.DataDir.openQuerylogDb` established, moved in. The
    /// controller closes it when the boot generation retires or the process
    /// stops.
    database: db.Db,
    /// Where later generations open their own connections. The controller is
    /// resizable only when this is set.
    source: ?Source,
    logging: model.Logging,
    monitor: ?*disk_monitor.Monitor = null,
    diagnostics: ?*events.Store = null,
};

/// What `prepare` refuses with. Every variant leaves the running logger and the
/// database untouched.
pub const PrepareError = Allocator.Error || db.Error || std.Io.ConcurrentError || error{
    /// `config/validate.zig`'s floor.
    BufferTooSmall,
    /// `config/validate.zig`'s ceiling, `logger.query_log_buffer_max`.
    BufferTooLarge,
    /// A previous resize's generation is still draining. At most one retired
    /// generation may exist, so this bounds writers, connections and buffers
    /// rather than queueing an unbounded chain of them.
    PreviousResizeDraining,
    /// The candidate writer could not prepare its statements.
    WriterStatementsFailed,
    /// A borrowed controller owns no generation to replace.
    NotResizable,
};

/// `config/validate.zig`'s own wording for a refused buffer size, so a settings
/// change and a startup rejection say the same thing about the same value.
/// Returns null for the failures that are not about the size.
pub fn sizeMessage(err: PrepareError, requested: u32, buf: []u8) ?[]const u8 {
    return switch (err) {
        error.BufferTooSmall => "must be at least 1",
        error.BufferTooLarge => std.fmt.bufPrint(
            buf,
            "must be at most {d}, got {d}",
            .{ logger.query_log_buffer_max, requested },
        ) catch buf[0..],
        else => null,
    };
}

/// A generation built, connected and parked, waiting for a commit that has not
/// happened yet. Exactly one of `publish` or `abandon` must consume it.
pub const Prepared = struct {
    generation: *Generation,
};

pub const Controller = struct {
    gpa: Allocator,
    source: ?Source,
    monitor: ?*disk_monitor.Monitor,
    diagnostics: ?*events.Store,

    /// Guards `live`, `retired`, `phase`, `stopping`, `base` and every
    /// generation's `borrows`. Held briefly and never across an await — the
    /// CertStore discipline (`cert_store.zig:199`).
    mutex: std.Io.Mutex = .init,
    /// Signalled whenever the reaper's work might have changed: a retirement
    /// was published, a borrow was released, or shutdown began.
    progress: std.Io.Condition = .init,

    live: *Generation,
    /// The one generation a resize may leave behind.
    retired: ?*Generation = null,
    phase: Phase = .idle,
    stopping: bool = false,
    stopped: bool = false,
    base: Totals = .{},

    /// Null for a borrowed controller. The sole joiner of retired writers.
    reaper: ?std.Io.Future(void) = null,

    const Phase = enum {
        idle,
        /// A retired generation is waiting for its last producer borrow.
        waiting_borrows,
        /// The reaper owns the retired generation and is joining its writer.
        draining,
    };

    /// Builds the boot generation around `opts.database` and starts its writer
    /// and the reaper.
    ///
    /// The boot writer takes the plain `Logger.runWriter` path rather than the
    /// parked one: a process that cannot prepare its statements at boot still
    /// serves DNS with `writer_failed` set, which is today's behaviour and the
    /// right one — refusing to start would cost more than the log rows do.
    /// Only a resize can afford to refuse.
    pub fn init(self: *Controller, io: std.Io, opts: Options) InitError!void {
        var database = opts.database;
        const generation = build(opts.gpa, database, opts.logging, opts.diagnostics) catch |err| {
            database.close();
            return err;
        };
        errdefer generation.deinit(opts.gpa);
        const owned = &generation.owned.?;

        self.* = .{
            .gpa = opts.gpa,
            .source = opts.source,
            .monitor = opts.monitor,
            .diagnostics = opts.diagnostics,
            .live = generation,
        };

        owned.writer = try io.concurrent(logger.Logger.runWriter, .{
            generation.logger,
            io,
            opts.gpa,
            &owned.database,
            opts.monitor,
        });
        errdefer {
            generation.logger.shutdown(io);
            owned.writer.await(io) catch {};
        }

        self.reaper = try io.concurrent(runReaper, .{ self, io });
    }

    /// Stops every generation in the order the writer deadlock demands, and
    /// joins every task this controller started.
    ///
    /// Producers must already be quiesced (`app.zig` cancels their group
    /// first): an entry enqueued after the close is a dropped entry, and a
    /// borrow still held here would stall the reaper.
    ///
    /// The order is today's safe order, once per generation. The live queue
    /// closes before its `draining` flag is set, because a writer that sees
    /// `draining` must still be able to reach the end of its queue and only a
    /// closed queue reports its end. The retired generation gets `draining`
    /// too — without it a retired writer parked on a shut disk gate waits for a
    /// recovery that is never coming, and the reaper would never join it.
    /// Idempotent: the teardown path and `deinit` both call it.
    pub fn shutdown(self: *Controller, io: std.Io) void {
        if (self.stopped) return;
        self.stopped = true;

        self.mutex.lockUncancelable(io);
        const live = self.live;
        if (live.owned == null) {
            self.mutex.unlock(io);
            return;
        }
        live.logger.shutdown(io);
        if (self.retired) |old| old.logger.draining.store(true, .release);
        self.stopping = true;
        self.progress.broadcast(io);
        self.mutex.unlock(io);

        live.owned.?.writer.await(io) catch {};
        if (self.reaper) |*reaper| {
            reaper.await(io);
            self.reaper = null;
        }
    }

    /// Frees the live generation and its connection. Stops everything first if
    /// the caller has not.
    pub fn deinit(self: *Controller, io: std.Io) void {
        self.shutdown(io);
        std.debug.assert(self.retired == null);
        if (self.live.owned != null) self.live.deinit(self.gpa);
    }

    // -----------------------------------------------------------------------
    // the producer side
    // -----------------------------------------------------------------------

    /// Pins the live generation for the length of one entry's transform and
    /// enqueue. Every `acquire` must be paired with a `release`.
    pub fn acquire(self: *Controller, io: std.Io) *Generation {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        self.live.borrows += 1;
        return self.live;
    }

    /// Releases a borrow. The last borrow of a retired generation is what lets
    /// the reaper close that queue.
    pub fn release(self: *Controller, io: std.Io, generation: *Generation) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        std.debug.assert(generation.borrows > 0);
        generation.borrows -= 1;
        if (generation.retired and generation.borrows == 0) self.progress.broadcast(io);
    }

    // -----------------------------------------------------------------------
    // the live-configuration setters (milestone 34 §S1, addressed here so they
    // survive a resize)
    // -----------------------------------------------------------------------

    /// Installs the privacy policy on every outstanding generation.
    ///
    /// Each generation keeps its own packed word and a producer still decides
    /// one entry from one load of the generation it borrowed, so a mixed policy
    /// remains impossible: the two stores below are to two different queues'
    /// producers, and no producer reads both.
    pub fn setPrivacy(self: *Controller, io: std.Io, p: logger.Privacy) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        self.live.logger.setPrivacy(p);
        if (self.retired) |old| old.logger.setPrivacy(p);
    }

    pub fn setFlushInterval(self: *Controller, io: std.Io, seconds: u16) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        self.live.logger.setFlushInterval(seconds);
        if (self.retired) |old| old.logger.setFlushInterval(seconds);
    }

    // -----------------------------------------------------------------------
    // the facade's read side
    // -----------------------------------------------------------------------

    /// Every counter and state the metrics and health surfaces read, in one
    /// coherent reading across the outstanding generations.
    pub fn sample(self: *Controller, io: std.Io) Sample {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);

        var totals = self.base;
        totals.fold(self.live.logger);
        var episode = self.live.logger.gateEpisode();
        if (self.retired) |old| {
            totals.fold(old.logger);
            episode = @enumFromInt(@max(@intFromEnum(episode), @intFromEnum(old.logger.gateEpisode())));
        }

        return .{
            .queries_dropped = totals.queries_dropped,
            .rows_written = totals.rows_written,
            .batches_gated = totals.batches_gated,
            .last_drop_s = if (totals.last_drop_s == 0) null else totals.last_drop_s,
            .writer_failed = self.live.logger.writer_failed.load(.acquire),
            .gate_episode = episode,
        };
    }

    /// Whether a previous resize's generation is still outstanding. This is
    /// what `prepare` refuses on, and what a caller polls to know when the one
    /// retirement slot is free again.
    pub fn retirementPending(self: *Controller, io: std.Io) bool {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        return self.retired != null;
    }

    /// The live queue's capacity, which is what a resize changes.
    pub fn capacity(self: *Controller, io: std.Io) u32 {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        const owned = self.live.owned orelse return @intCast(self.live.logger.queue.capacity());
        return owned.capacity;
    }

    // -----------------------------------------------------------------------
    // resize: prepare → (the caller commits) → publish, or abandon
    // -----------------------------------------------------------------------

    /// Builds a complete replacement generation and parks its writer. Fallible
    /// from end to end and side-effect free: on any failure nothing has been
    /// published, no connection is left open and the running logger has not
    /// been touched.
    ///
    /// The caller commits its database row only after this returns, and then
    /// calls exactly one of `publish` or `abandon`.
    ///
    /// `cfg` is the FINAL MERGED logging configuration, not the live one: the
    /// same write that resizes the queue may also change the privacy flags or
    /// the flush interval, and those are published against the generation this
    /// candidate is about to replace. Seeding from the live values would leave
    /// the replacement carrying the policy the write just retired.
    pub fn prepare(self: *Controller, io: std.Io, cfg: model.Logging) PrepareError!Prepared {
        const entries = cfg.query_log_buffer_max;
        if (entries < 1) return error.BufferTooSmall;
        if (entries > logger.query_log_buffer_max) return error.BufferTooLarge;

        const source = self.source orelse return error.NotResizable;

        {
            self.mutex.lockUncancelable(io);
            defer self.mutex.unlock(io);
            if (self.retired != null) return error.PreviousResizeDraining;
        }
        var database = try querylog_schema.reopen(io, source.dir, source.path);
        const generation = build(self.gpa, database, cfg, self.diagnostics) catch |err| {
            database.close();
            return err;
        };
        errdefer generation.deinit(self.gpa);
        const owned = &generation.owned.?;

        owned.writer = try io.concurrent(runParkedWriter, .{ generation, io, self.gpa, self.monitor });

        // The statements are prepared before anything is published, so a
        // failure here is a refused settings change rather than a writer that
        // dies under a queue producers are already filling.
        owned.ready.waitUncancelable(io);
        if (owned.prepare_error) |err| {
            owned.writer.await(io) catch {};
            scope.warn(
                "query logger resize: preparing the batch statements failed: {s}",
                .{@errorName(err)},
            );
            // `database` is closed by the errdefers above, which still cover
            // everything this function allocated.
            return error.WriterStatementsFailed;
        }

        return .{ .generation = generation };
    }

    /// Throws a prepared generation away: the transaction it was built for did
    /// not commit. Releases the parked writer, joins it and closes everything.
    pub fn abandon(self: *Controller, io: std.Io, prepared: Prepared) void {
        const owned = &prepared.generation.owned.?;
        owned.aborted = true;
        owned.activation.set(io);
        owned.writer.await(io) catch {};
        prepared.generation.deinit(self.gpa);
    }

    /// Infallible and I/O-free: swap the live pointer, mark the old generation
    /// retired, open the new writer's activation gate. Nothing is closed,
    /// joined or freed here — that is the reaper's work, and it happens after
    /// the last producer of the old generation releases it.
    pub fn publish(self: *Controller, io: std.Io, prepared: Prepared) void {
        const fresh = prepared.generation;

        self.mutex.lockUncancelable(io);
        const old = self.live;
        self.live = fresh;
        old.retired = true;
        self.retired = old;
        self.phase = .waiting_borrows;
        self.progress.broadcast(io);
        self.mutex.unlock(io);

        fresh.owned.?.activation.set(io);
    }

    // -----------------------------------------------------------------------
    // the reaper
    // -----------------------------------------------------------------------

    /// The sole joiner of retired writers, for the controller's whole life.
    ///
    /// `publish` never spawns: a spawn can fail, and publish may not. This task
    /// exists from construction and waits for work instead.
    fn runReaper(self: *Controller, io: std.Io) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);

        while (true) {
            if (self.phase == .waiting_borrows and self.retired.?.borrows == 0) {
                const old = self.retired.?;
                self.phase = .draining;
                self.mutex.unlock(io);
                drain(old, io);
                self.mutex.lockUncancelable(io);

                // Folded before the generation is freed, so the totals never
                // dip: a reader between the fold and the free sees the same
                // numbers either way.
                self.base.fold(old.logger);
                self.retired = null;
                self.phase = .idle;
                old.deinit(self.gpa);
                self.progress.broadcast(io);
                continue;
            }
            if (self.stopping and self.retired == null) return;
            self.progress.waitUncancelable(io, &self.mutex);
        }
    }
};

/// One heap-stable generation with its writer future still unset. Takes
/// ownership of `database`.
fn build(
    gpa: Allocator,
    database: db.Db,
    cfg: model.Logging,
    diagnostics: ?*events.Store,
) Allocator.Error!*Generation {
    const generation = try gpa.create(Generation);
    errdefer gpa.destroy(generation);

    const queue_buf = try gpa.alloc(logger.Entry, cfg.query_log_buffer_max);
    errdefer gpa.free(queue_buf);

    generation.* = .{
        .logger = undefined,
        .owned = .{
            .state = .init(cfg, queue_buf),
            .queue_buf = queue_buf,
            .database = database,
            .writer = undefined,
            .capacity = cfg.query_log_buffer_max,
        },
    };
    const owned = &generation.owned.?;
    generation.logger = &owned.state;
    owned.state.diagnostics = diagnostics;
    return generation;
}

/// Closes a retired generation's queue, waits for its writer to empty it, and
/// closes its connection. Runs with the controller's mutex released: it awaits.
fn drain(generation: *Generation, io: std.Io) void {
    const owned = &generation.owned.?;
    // `retire`, not `shutdown`: with `draining` set a gate-held writer would
    // take the `GatedAtShutdown` path and DROP the batch it is holding
    // (`logger.zig`). A replaced generation is not a stopping process — the
    // disk can still recover, and those rows are still going to be written.
    generation.logger.retire(io);
    owned.writer.await(io) catch {};
}

/// A resize generation's writer: prepare, report, park, then serve.
fn runParkedWriter(
    generation: *Generation,
    io: std.Io,
    gpa: Allocator,
    monitor: ?*disk_monitor.Monitor,
) std.Io.Cancelable!void {
    const owned = &generation.owned.?;

    var writer = queries_repo.BatchWriter.init(gpa, &owned.database) catch |err| {
        owned.prepare_error = err;
        owned.ready.set(io);
        return;
    };
    defer writer.deinit();
    owned.ready.set(io);

    // Nothing is consumed until `publish` or `abandon` decides. The queue this
    // writer will serve has no producers yet either — it is not live.
    owned.activation.waitUncancelable(io);
    if (owned.aborted) return;

    return generation.logger.runPrepared(io, &writer, monitor);
}

/// A controller over a `Logger` the caller already owns: no connection, no
/// writer, no reaper and no resize. Owns nothing, so there is nothing to tear
/// down; both parts are inline so a caller keeps them on its stack beside the
/// sink — `upstream/owner.zig`'s `Borrowed`.
pub const Borrowed = struct {
    generation: Generation = undefined,
    controller: Controller = undefined,

    pub fn over(self: *Borrowed, lg: *logger.Logger) *Controller {
        self.generation = .{ .logger = lg };
        self.controller = .{
            .gpa = undefined,
            .source = null,
            .monitor = null,
            .diagnostics = null,
            .live = &self.generation,
        };
        return &self.controller;
    }
};

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

const testing = std.testing;

/// `std.testing.tmpDir` creates its directory against `std.testing.io`, so
/// every call into the code under test uses the same `Io` instance — an
/// `Io.Threaded`, which is what makes `io.concurrent` available to the writers.
const test_io = testing.io;

/// Where `std.testing.tmpDir` puts its directories.
const tmp_prefix = ".zig-cache/tmp/";
const sub_path_len = @typeInfo(@FieldType(testing.TmpDir, "sub_path")).array.len;

/// A real `querylog.db` in a temporary directory with a controller over it.
///
/// Must not be moved once `start` has run: the reaper holds `&self.controller`,
/// and the queue keeps waiting tasks in intrusive lists.
const Harness = struct {
    tmp: testing.TmpDir = undefined,
    path_buf: [tmp_prefix.len + sub_path_len + "/querylog.db".len + 1]u8 = undefined,
    path: [:0]const u8 = undefined,
    monitor: ?*disk_monitor.Monitor = null,
    controller: Controller = undefined,
    cfg: model.Logging = .{},

    fn start(self: *Harness, cfg: model.Logging) !void {
        self.cfg = cfg;
        self.tmp = testing.tmpDir(.{});
        errdefer self.tmp.cleanup();

        var root_buf: [tmp_prefix.len + sub_path_len]u8 = undefined;
        @memcpy(root_buf[0..tmp_prefix.len], tmp_prefix);
        @memcpy(root_buf[tmp_prefix.len..], &self.tmp.sub_path);
        self.path = try std.fmt.bufPrintZ(&self.path_buf, "{s}/querylog.db", .{&root_buf});

        var opened = try querylog_schema.open(test_io, std.Io.Dir.cwd(), self.path);
        errdefer opened.database.close();

        try self.controller.init(test_io, .{
            .gpa = testing.allocator,
            .database = opened.database,
            .source = .{ .dir = std.Io.Dir.cwd(), .path = self.path },
            .logging = cfg,
            .monitor = self.monitor,
        });
    }

    fn stop(self: *Harness) void {
        self.controller.deinit(test_io);
        self.tmp.cleanup();
    }

    /// A resize of the started configuration, which is what a settings write
    /// hands `prepare`: the whole merged section with one field moved.
    fn prepareResize(self: *Harness, entries: u32) PrepareError!Prepared {
        var cfg = self.cfg;
        cfg.query_log_buffer_max = entries;
        return self.controller.prepare(test_io, cfg);
    }

    /// The rows on disk, read through a connection of its own after the
    /// controller has closed every one of its own.
    fn storedRows(self: *Harness) !i64 {
        var database = try querylog_schema.reopen(test_io, std.Io.Dir.cwd(), self.path);
        defer database.close();
        return queries_repo.countRows(&database);
    }

    /// A connection for a test that needs to change the schema under a running
    /// controller.
    fn sideConnection(self: *Harness) !db.Db {
        return querylog_schema.reopen(test_io, std.Io.Dir.cwd(), self.path);
    }
};

fn testEntry(timestamp: i64, domain: []const u8) logger.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",
    });
}

/// One producer, following the borrow protocol `QuerySink` follows.
fn produce(controller: *Controller, count: usize, base: i64) void {
    for (0..count) |i| {
        const generation = controller.acquire(test_io);
        defer controller.release(test_io, generation);
        generation.logger.log(test_io, testEntry(base + @as(i64, @intCast(i)), "producer.example"));
    }
}

const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };

fn awaitRows(controller: *Controller, target: u64, limit: usize) !void {
    var polls: usize = 0;
    while (controller.sample(test_io).rows_written < target) : (polls += 1) {
        try testing.expect(polls < limit);
        try poll.sleep(test_io);
    }
}

fn awaitGated(controller: *Controller, limit: usize) !void {
    var polls: usize = 0;
    while (controller.sample(test_io).batches_gated == 0) : (polls += 1) {
        try testing.expect(polls < limit);
        try poll.sleep(test_io);
    }
}

fn awaitRetired(controller: *Controller, limit: usize) !void {
    var polls: usize = 0;
    while (controller.retirementPending(test_io)) : (polls += 1) {
        try testing.expect(polls < limit);
        try poll.sleep(test_io);
    }
}

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

test "a resize under concurrent producers loses nothing and accounts for every entry" {
    var h: Harness = .{};
    // Room for every entry all four producers can write: an overflow drop here
    // would be ordinary backpressure, and this case is about the swap.
    try h.start(.{ .query_log_buffer_max = 4096, .query_log_flush_interval_s = 0 });
    defer h.stop();

    const producers = 4;
    const each = 200;
    var futures: [producers]std.Io.Future(void) = undefined;
    var started: usize = 0;
    // Declared before the spawn loop so a partial spawn still joins what ran.
    defer for (futures[0..started]) |*future| future.await(test_io);
    while (started < producers) : (started += 1) {
        futures[started] = try test_io.concurrent(produce, .{
            &h.controller,
            each,
            @as(i64, @intCast(started)) * each,
        });
    }

    // The swap happens with all four producers mid-flight. Every entry either
    // reaches the old queue before its close or the new one after the publish;
    // no third outcome exists, which is what "no resize drop window" means.
    const prepared = try h.prepareResize(8192);
    h.controller.publish(test_io, prepared);
    try testing.expectEqual(@as(u32, 8192), h.controller.capacity(test_io));

    for (futures[0..started]) |*future| future.await(test_io);
    started = 0;

    try awaitRetired(&h.controller, 400);
    h.controller.shutdown(test_io);

    const total = producers * each;
    const reading = h.controller.sample(test_io);
    // Every entry produced is accounted for exactly once, and none of them was
    // lost to the swap.
    try testing.expectEqual(@as(u64, total), reading.rows_written + reading.queries_dropped);
    try testing.expectEqual(@as(u64, 0), reading.queries_dropped);
    try testing.expectEqual(@as(i64, total), try h.storedRows());
}

test "a resize with the disk gate closed publishes at once and retires when the gate reopens" {
    var monitor = critical();
    var h: Harness = .{ .monitor = &monitor };
    try h.start(.{ .query_log_buffer_max = 64, .query_log_flush_interval_s = 0 });
    defer h.stop();

    produce(&h.controller, 8, 1);
    try awaitGated(&h.controller, 400);

    // The old writer is holding a batch against a shut gate. Publish must not
    // wait for it: nothing here is allowed to block on the disk.
    const prepared = try h.prepareResize(128);
    h.controller.publish(test_io, prepared);
    try testing.expect(h.controller.retirementPending(test_io));

    produce(&h.controller, 8, 100);

    monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
    try awaitRetired(&h.controller, 1200);
    h.controller.shutdown(test_io);

    // The retirement mode does not set `draining`, so the held batch waited for
    // the disk instead of being counted as lost.
    const reading = h.controller.sample(test_io);
    try testing.expectEqual(@as(u64, 0), reading.queries_dropped);
    try testing.expectEqual(@as(u64, 16), reading.rows_written);
    try testing.expectEqual(@as(i64, 16), try h.storedRows());
}

test "a candidate whose writer cannot prepare its statements leaves the running logger alone" {
    var h: Harness = .{};
    try h.start(.{ .query_log_buffer_max = 16, .query_log_flush_interval_s = 0 });
    defer h.stop();

    produce(&h.controller, 4, 1);
    try awaitRows(&h.controller, 4, 400);

    // The live writer already holds its prepared statements; a new connection
    // would find no table to prepare against.
    var side = try h.sideConnection();
    try side.exec("ALTER TABLE query_log RENAME TO query_log_hidden");

    try testing.expectError(error.WriterStatementsFailed, h.prepareResize(64));

    // Untouched: same capacity, same generation, nothing retiring, and the
    // counters carry on from where they were.
    try testing.expectEqual(@as(u32, 16), h.controller.capacity(test_io));
    try testing.expect(!h.controller.retirementPending(test_io));
    try testing.expect(!h.controller.sample(test_io).writer_failed);

    try side.exec("ALTER TABLE query_log_hidden RENAME TO query_log");
    side.close();

    // And the refusal left nothing behind: the next resize works.
    const prepared = try h.prepareResize(64);
    h.controller.publish(test_io, prepared);
    try awaitRetired(&h.controller, 400);
    try testing.expectEqual(@as(u32, 64), h.controller.capacity(test_io));

    produce(&h.controller, 4, 100);
    h.controller.shutdown(test_io);
    try testing.expectEqual(@as(u64, 8), h.controller.sample(test_io).rows_written);
}

test "a prepared candidate the caller abandons closes everything it opened" {
    var h: Harness = .{};
    try h.start(.{ .query_log_buffer_max = 16, .query_log_flush_interval_s = 0 });
    defer h.stop();

    const prepared = try h.prepareResize(64);
    h.controller.abandon(test_io, prepared);

    try testing.expectEqual(@as(u32, 16), h.controller.capacity(test_io));
    try testing.expect(!h.controller.retirementPending(test_io));

    produce(&h.controller, 3, 1);
    h.controller.shutdown(test_io);
    try testing.expectEqual(@as(u64, 3), h.controller.sample(test_io).rows_written);
}

test "an invalid buffer size is refused with the message startup uses" {
    var h: Harness = .{};
    try h.start(.{ .query_log_buffer_max = 16 });
    defer h.stop();

    var buf: [64]u8 = undefined;

    try testing.expectError(error.BufferTooSmall, h.prepareResize(0));
    try testing.expectEqualStrings(
        "must be at least 1",
        sizeMessage(error.BufferTooSmall, 0, &buf).?,
    );

    const over = logger.query_log_buffer_max + 1;
    try testing.expectError(error.BufferTooLarge, h.prepareResize(over));
    var expected_buf: [64]u8 = undefined;
    // `config/validate.zig`'s wording, formatted from the same constant.
    const expected = try std.fmt.bufPrint(
        &expected_buf,
        "must be at most {d}, got {d}",
        .{ logger.query_log_buffer_max, over },
    );
    try testing.expectEqualStrings(expected, sizeMessage(error.BufferTooLarge, over, &buf).?);

    // Neither refusal touched anything.
    try testing.expectEqual(@as(u32, 16), h.controller.capacity(test_io));
}

test "a second resize is refused while the previous one is still draining" {
    var monitor = critical();
    var h: Harness = .{ .monitor = &monitor };
    try h.start(.{ .query_log_buffer_max = 32, .query_log_flush_interval_s = 0 });
    defer h.stop();

    produce(&h.controller, 4, 1);
    try awaitGated(&h.controller, 400);

    const first = try h.prepareResize(64);
    h.controller.publish(test_io, first);

    // The gate is shut, so the retired writer cannot finish. A second resize
    // would put a third buffer, connection and writer in flight.
    try testing.expectError(error.PreviousResizeDraining, h.prepareResize(128));

    monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
    try awaitRetired(&h.controller, 1200);

    // The slot is free again the moment the retirement completes.
    const second = try h.prepareResize(128);
    h.controller.publish(test_io, second);
    try testing.expectEqual(@as(u32, 128), h.controller.capacity(test_io));
    try awaitRetired(&h.controller, 400);
    h.controller.shutdown(test_io);
}

test "the counters, the drop stamp and the gate episode survive a swap" {
    var monitor = critical();
    var h: Harness = .{ .monitor = &monitor };
    // One entry of room behind a writer the gate is holding: the producers
    // below have to displace each other, so drops and a stamp exist to carry.
    try h.start(.{ .query_log_buffer_max = 1, .query_log_flush_interval_s = 0 });
    defer h.stop();

    produce(&h.controller, 1, 1);
    try awaitGated(&h.controller, 400);
    produce(&h.controller, 40, 10);

    const before = h.controller.sample(test_io);
    try testing.expect(before.queries_dropped > 0);
    try testing.expect(before.last_drop_s != null);
    try testing.expectEqual(logger.GateEpisode.losing, before.gate_episode);

    const prepared = try h.prepareResize(64);
    h.controller.publish(test_io, prepared);

    // Read across the swap: the new generation's counters start at zero, so
    // anything that reported only the live generation would go backwards here.
    const after = h.controller.sample(test_io);
    try testing.expectEqual(before.queries_dropped, after.queries_dropped);
    try testing.expectEqual(before.last_drop_s, after.last_drop_s);
    try testing.expectEqual(before.rows_written, after.rows_written);
    // The retired writer is still holding its batch against the shut gate, and
    // that is still a current fault.
    try testing.expectEqual(logger.GateEpisode.losing, after.gate_episode);

    monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
    try awaitRetired(&h.controller, 1200);

    // Folding the retired generation's finals into the base is the other half:
    // its counters must not vanish with it.
    const settled = h.controller.sample(test_io);
    try testing.expectEqual(before.queries_dropped, settled.queries_dropped);
    try testing.expectEqual(before.last_drop_s, settled.last_drop_s);
    try testing.expect(settled.rows_written >= before.rows_written);
    try testing.expectEqual(logger.GateEpisode.open, settled.gate_episode);

    h.controller.shutdown(test_io);
}

/// Flips both privacy fields together, over and over, through the controller.
fn flipPrivacy(controller: *Controller, rounds: usize) void {
    for (0..rounds) |i| {
        controller.setPrivacy(test_io, if (i % 2 == 0)
            .{}
        else
            .{ .hide_domains = true, .hide_client_ips = true });
    }
}

fn resizeRepeatedly(h: *Harness, rounds: usize) void {
    for (0..rounds) |i| {
        const prepared = h.prepareResize(if (i % 2 == 0) 48 else 64) catch continue;
        h.controller.publish(test_io, prepared);
        awaitRetired(&h.controller, 2000) catch return;
    }
}

test "no producer sees a privacy policy that redacts one field and not the other, across resizes" {
    var h: Harness = .{};
    try h.start(.{ .query_log_buffer_max = 64, .query_log_flush_interval_s = 0 });
    defer h.stop();

    const source = testEntry(1, "tracker.example");
    var flipper = try test_io.concurrent(flipPrivacy, .{ &h.controller, 20_000 });
    var resizer = try test_io.concurrent(resizeRepeatedly, .{ &h, 4 });

    // Recorded, not asserted, while the two other tasks run: an assertion that
    // returned here would leave them unjoined.
    var mixed = false;
    for (0..20_000) |_| {
        const generation = h.controller.acquire(test_io);
        const out = generation.logger.transformed(source);
        h.controller.release(test_io, generation);
        const domain_hidden = std.mem.eql(u8, out.domain(), logger.hidden_marker);
        const client_hidden = std.mem.eql(u8, out.clientIp(), logger.hidden_marker);
        if (domain_hidden != client_hidden) mixed = true;
    }
    flipper.await(test_io);
    resizer.await(test_io);
    try testing.expect(!mixed);

    try awaitRetired(&h.controller, 2000);
    h.controller.shutdown(test_io);
}

test "shutdown joins a retirement the disk gate is still blocking" {
    var monitor = critical();
    var h: Harness = .{ .monitor = &monitor };
    try h.start(.{ .query_log_buffer_max = 64, .query_log_flush_interval_s = 0 });
    defer h.stop();

    const held = 6;
    produce(&h.controller, held, 1);
    try awaitGated(&h.controller, 400);

    const prepared = try h.prepareResize(128);
    h.controller.publish(test_io, prepared);
    try testing.expect(h.controller.retirementPending(test_io));

    const after_swap = 4;
    produce(&h.controller, after_swap, 100);

    // The disk never recovers. Both writers are gate-held, one of them retired
    // and owned by the reaper — the exit still ends, each future is awaited by
    // exactly one task, and every entry is counted rather than silently lost.
    h.controller.shutdown(test_io);

    const reading = h.controller.sample(test_io);
    try testing.expectEqual(@as(u64, held + after_swap), reading.queries_dropped);
    try testing.expectEqual(@as(u64, 0), reading.rows_written);
    try testing.expectEqual(@as(i64, 0), try h.storedRows());
}

test "a borrowed controller reports its logger and refuses to resize" {
    var queue_buf: [4]logger.Entry = undefined;
    var lg: logger.Logger = .init(.{}, &queue_buf);
    lg.rows_written.store(7, .monotonic);
    lg.queries_dropped.store(2, .monotonic);

    var owner: Borrowed = .{};
    const controller = owner.over(&lg);

    const reading = controller.sample(test_io);
    try testing.expectEqual(@as(u64, 7), reading.rows_written);
    try testing.expectEqual(@as(u64, 2), reading.queries_dropped);
    try testing.expectEqual(@as(u32, 4), controller.capacity(test_io));

    try testing.expectError(error.NotResizable, controller.prepare(test_io, .{ .query_log_buffer_max = 8 }));

    // Owns nothing, so both teardown calls are no-ops over the caller's logger.
    controller.shutdown(test_io);
    controller.deinit(test_io);
}