//! The `std.log` sink: runtime level filter, stderr or rotating-file output,
//! and upstream-error deduplication (PLAN §11.6).
//!
//! ## The sink lock
//!
//! `std.Options.logFn` (std.zig:132) receives no `std.Io`, so the sink cannot
//! guard itself with `std.Io.Mutex`: both `lock` and `unlock` (Io.zig:1602,
//! Io.zig:1640) take an `Io` parameter, and before `install` the sink holds
//! none. `std.Thread.Mutex` is not an option either — 0.16.0 has no
//! `lib/std/Thread/` directory and `lib/std/Thread.zig` declares no `Mutex`.
//!
//! The sink therefore takes the lock `std.log.defaultLog` itself takes
//! (log.zig:96-109): `std.debug.lockStderr` / `std.debug.unlockStderr`. Those
//! need no `Io` because they read `std.Options.debug_io` (debug.zig:283), they
//! are documented as recursive (debug.zig:263-270), and `Io/Threaded.zig`
//! implements that recursion per OS thread (Threaded.zig:13787-13796). nxdns
//! runs on the `std.Io.Threaded` instance the stdlib start code constructs
//! (start.zig:724, handed to `main` via `std.process.Init`), so one task is
//! one thread and the recursion holds. Taking it across the file writes as well keeps the
//! file path and the stderr fallback path from interleaving with each other,
//! with `std.Progress`, or with a panic dump.
//!
//! Every `state` field below is read and written only while that lock is held.
//!
//! Nothing in this file may call `std.log`: this file is what `std.log` calls,
//! and a log line emitted from the sink would recurse.

const std = @import("std");
const builtin = @import("builtin");
const model = @import("../config/model.zig");

/// Two upstream failures with the same key inside this window produce one line.
pub const dedup_window_ns: i96 = 60 * std.time.ns_per_s;
pub const dedup_slots = 64;
pub const max_message_key_bytes = 96;
pub const max_scope_name_bytes = 32;
pub const max_dedup_key_bytes = max_scope_name_bytes + 1 + max_message_key_bytes;

/// A formatted message longer than this is truncated rather than dropped: a
/// log line must never be a hard failure.
pub const max_message_bytes = 2048;

/// `writeEscaped` expands one byte to at most four (`\xNN`).
pub const max_escaped_message_bytes = max_message_bytes * 4;

/// The widest possible file record header and terminator:
/// `-9223372036854775808` + ` ` + `warning` + `(` + a bounded scope name +
/// `)` + `: ` + `\n`. A scope name is bounded by `boundedScopeName`, so a
/// record can never outgrow a buffer of this size plus the escaped message.
pub const max_line_header_bytes = 20 + 1 + 7 + 1 + max_scope_name_bytes + 1 + 2 + 1;

/// Room for the longest configured path plus the `".255"` rotation suffix.
pub const max_rotated_path_bytes = std.Io.Dir.max_path_bytes + 4;

pub const Stats = struct {
    lines_written: u64 = 0,
    lines_deduped: u64 = 0,
    /// Messages that did not fit `max_message_bytes` and carry the
    /// `truncation_marker` in place of their last bytes. Counted where the
    /// truncation happens, so a truncated message the dedup window then
    /// swallows counts here and under `lines_deduped`.
    lines_truncated: u64 = 0,
    rotations: u64 = 0,
    sink_errors: u64 = 0,
};

/// What a truncated message ends in, the same three bytes `safe_url.zig` uses
/// so both truncations read alike to an operator.
pub const truncation_marker = "...";

// ---------------------------------------------------------------------------
// Level
// ---------------------------------------------------------------------------

pub fn toStdLevel(level: model.LogLevel) std.log.Level {
    return switch (level) {
        .err => .err,
        .warn => .warn,
        .info => .info,
        .debug => .debug,
    };
}

/// `std.log.Level` orders the most severe tag first, so a message passes when
/// its ordinal is at or below the configured threshold's.
pub fn enabled(message_level: std.log.Level, threshold: std.log.Level) bool {
    return @intFromEnum(message_level) <= @intFromEnum(threshold);
}

/// `std.log.Level.asText` takes its level `comptime`; the sink filters at
/// runtime and needs the same strings.
fn levelText(level: std.log.Level) []const u8 {
    return switch (level) {
        .err => "error",
        .warn => "warning",
        .info => "info",
        .debug => "debug",
    };
}

/// A scope name reaches the record header, which shares one fixed buffer with
/// the message. Nothing bounds the name of a scope outside the deduplicated
/// four, so the sink bounds it here and both output paths use the result: one
/// event reads the same on stderr as in the file.
pub fn boundedScopeName(scope_name: []const u8) []const u8 {
    return scope_name[0..@min(scope_name.len, max_scope_name_bytes)];
}

// ---------------------------------------------------------------------------
// Deduplication
// ---------------------------------------------------------------------------

/// PLAN §11.6 rate-limits exactly the scopes whose failures repeat once per
/// query; every other scope logs unconditionally. `.tls_server` joins them
/// (milestone-16 ruling 16): its warnings are peer-driven, so an unhappy client
/// could otherwise evict genuine warnings from the rotating log.
pub fn isDedupScope(comptime scope: @EnumLiteral()) bool {
    return scope == .doh_client or scope == .dot_client or
        scope == .pool or scope == .forward_client or scope == .tls_server;
}

pub fn buildKey(
    buf: *[max_dedup_key_bytes]u8,
    scope_name: []const u8,
    message: []const u8,
) []const u8 {
    const scope_len = @min(scope_name.len, max_scope_name_bytes);
    @memcpy(buf[0..scope_len], scope_name[0..scope_len]);
    buf[scope_len] = 0;
    const message_len = @min(message.len, max_message_key_bytes);
    @memcpy(buf[scope_len + 1 ..][0..message_len], message[0..message_len]);
    return buf[0 .. scope_len + 1 + message_len];
}

/// Fixed-size and allocation-free: the sink runs inside `logFn`, which has no
/// allocator and must not fail.
pub const DedupTable = struct {
    entries: [dedup_slots]Entry = @splat(.{}),

    const Entry = struct {
        key: [max_dedup_key_bytes]u8 = undefined,
        key_len: u8 = 0,
        stamp_ns: i96 = 0,
        occupied: bool = false,
    };

    /// True means emit the line. A key seen less than `dedup_window_ns` ago is
    /// dropped; a key that is new to a full table replaces the entry with the
    /// oldest stamp. `now` is an `.awake` timestamp.
    pub fn admit(self: *DedupTable, now: std.Io.Timestamp, key: []const u8) bool {
        std.debug.assert(key.len <= max_dedup_key_bytes);
        var free_slot: ?usize = null;
        var oldest: ?usize = null;
        for (&self.entries, 0..) |*entry, i| {
            if (!entry.occupied) {
                if (free_slot == null) free_slot = i;
                continue;
            }
            if (std.mem.eql(u8, entry.key[0..entry.key_len], key)) {
                if (now.nanoseconds - entry.stamp_ns < dedup_window_ns) return false;
                entry.stamp_ns = now.nanoseconds;
                return true;
            }
            if (oldest == null or entry.stamp_ns < self.entries[oldest.?].stamp_ns) oldest = i;
        }
        const slot = free_slot orelse oldest.?;
        const entry = &self.entries[slot];
        @memcpy(entry.key[0..key.len], key);
        entry.key_len = @intCast(key.len);
        entry.stamp_ns = now.nanoseconds;
        entry.occupied = true;
        return true;
    }
};

// ---------------------------------------------------------------------------
// Rotation naming
// ---------------------------------------------------------------------------

/// `nxdns.log` plus generation `n` is `nxdns.log.n`.
pub fn rotatedName(buf: []u8, path: []const u8, n: u8) error{NameTooLong}![]const u8 {
    var w: std.Io.Writer = .fixed(buf);
    w.print("{s}.{d}", .{ path, n }) catch return error.NameTooLong;
    return w.buffered();
}

// ---------------------------------------------------------------------------
// Sink state
// ---------------------------------------------------------------------------

const State = struct {
    installed: bool = false,
    io: std.Io = undefined,
    threshold: std.log.Level = .info,
    output: model.LogOutput = .stderr,
    path_buf: [std.Io.Dir.max_path_bytes]u8 = undefined,
    path_len: usize = 0,
    max_files: u8 = 0,
    max_bytes: u64 = 0,
    file: ?std.Io.File = null,
    file_pos: u64 = 0,
    /// A rotation step failed. The live file stays closed until a later line
    /// completes the rotation: reopening it would append past `max_bytes`.
    rotate_pending: bool = false,
    dedup: DedupTable = .{},
    stats: Stats = .{},

    fn path(self: *const State) []const u8 {
        return self.path_buf[0..self.path_len];
    }
};

var state: State = .{};

/// Called once from `main` after the config parse. Never called by tests: a
/// sink installed under the test runner would swallow the harness's own logs.
pub fn install(io: std.Io, cfg: model.Logging) void {
    installWithMaxBytes(io, cfg, model.maxLogBytes(cfg));
}

/// Test-only entry point. `install`'s smallest possible `max_size_mb` is 1 MiB,
/// which would make the integration rotation case write a megabyte per
/// generation; this overrides only that threshold.
pub fn installForTest(io: std.Io, cfg: model.Logging, max_bytes_override: u64) void {
    installWithMaxBytes(io, cfg, max_bytes_override);
}

fn installWithMaxBytes(io: std.Io, cfg: model.Logging, max_bytes: u64) void {
    var stderr_buf: [64]u8 = undefined;
    _ = std.debug.lockStderr(&stderr_buf);
    defer std.debug.unlockStderr();

    closeFileLocked();
    state.io = io;
    state.threshold = toStdLevel(cfg.level);
    state.max_files = cfg.max_files;
    state.max_bytes = max_bytes;
    state.dedup = .{};
    state.rotate_pending = false;
    state.installed = true;

    if (cfg.file_path.len > state.path_buf.len) {
        // A path that cannot be stored whole would name a different file.
        state.path_len = 0;
        state.output = .stderr;
        state.stats.sink_errors += 1;
        return;
    }
    state.path_len = cfg.file_path.len;
    @memcpy(state.path_buf[0..state.path_len], cfg.file_path);
    state.output = cfg.output;
    if (state.output == .file) openFileLocked();
}

// ---------------------------------------------------------------------------
// Hot apply (milestone-34 S3.5)
// ---------------------------------------------------------------------------

/// Which of the three disjoint shapes a `logging` apply takes. The case is
/// decided from the FINAL MERGED config against the live sink, and it depends
/// only on `output` and `file_path` — fields nothing but an apply writes.
/// Rotation and write-failure recovery move `file`, `file_pos` and
/// `rotate_pending`, never these two, so a case decided in one lock hold is
/// still the right case in the next.
pub const ApplyCase = enum {
    /// The merged config wants a file, and it is not the file that is open:
    /// the path differs, or output is switching TO file.
    target_changed,
    /// Output is switching away from file. Nothing to open.
    target_removed,
    /// Everything else — output stays stderr/syslog, or output stays file on
    /// the SAME path. Only config fields move; the handle and its position and
    /// rotation state stay with the rotation machinery that owns them.
    target_unchanged,
};

/// The complete new target state for a `target_changed` apply. The handle
/// couples to both other fields: inheriting the old `file_pos` would write
/// past the new file's end, and inheriting a pending rotation would rotate the
/// new target on its first line.
pub const PreparedSink = struct {
    file: std.Io.File,
    file_pos: u64,
    rotate_pending: bool = false,
};

pub const PrepareError = error{
    /// `file_path` does not fit in the sink's path buffer, so the sink could
    /// not name the file it was told to write.
    PathTooLong,
    /// The new target could not be opened, created, or measured.
    TargetUnopenable,
};

/// A validated `logging` apply, owning everything publish needs. Publish takes
/// no borrow from the request arena, so this outlives the request that built
/// it.
pub const PreparedApply = struct {
    case: ApplyCase,
    threshold: std.log.Level,
    output: model.LogOutput,
    path_buf: [std.Io.Dir.max_path_bytes]u8,
    path_len: usize,
    max_files: u8,
    max_bytes: u64,
    sink: ?PreparedSink,

    pub fn path(self: *const PreparedApply) []const u8 {
        return self.path_buf[0..self.path_len];
    }
};

/// Prepare: everything fallible happens here, and nothing is published. A
/// `target_changed` apply opens the NEW file and MEASURES it — the open is the
/// step that can fail, and doing it here closes the close-then-reopen window
/// `installWithMaxBytes` has, where a bad new path leaves no sink at all.
pub fn prepareApply(io: std.Io, cfg: model.Logging) PrepareError!PreparedApply {
    return prepareApplyWithMaxBytes(io, cfg, model.maxLogBytes(cfg));
}

/// Test-only entry point, mirroring `installForTest`.
pub fn prepareApplyForTest(io: std.Io, cfg: model.Logging, max_bytes: u64) PrepareError!PreparedApply {
    return prepareApplyWithMaxBytes(io, cfg, max_bytes);
}

fn prepareApplyWithMaxBytes(io: std.Io, cfg: model.Logging, max_bytes: u64) PrepareError!PreparedApply {
    if (cfg.file_path.len > std.Io.Dir.max_path_bytes) return error.PathTooLong;

    var prepared: PreparedApply = .{
        .case = undefined,
        .threshold = toStdLevel(cfg.level),
        .output = cfg.output,
        .path_buf = undefined,
        .path_len = cfg.file_path.len,
        .max_files = cfg.max_files,
        .max_bytes = max_bytes,
        .sink = null,
    };
    @memcpy(prepared.path_buf[0..prepared.path_len], cfg.file_path);
    prepared.case = classifyApply(cfg);

    if (prepared.case == .target_changed) {
        prepared.sink = try openTarget(io, prepared.path());
    }
    return prepared;
}

fn classifyApply(cfg: model.Logging) ApplyCase {
    var stderr_buf: [64]u8 = undefined;
    _ = std.debug.lockStderr(&stderr_buf);
    defer std.debug.unlockStderr();

    const currently_file = state.output == .file;
    if (cfg.output != .file) return if (currently_file) .target_removed else .target_unchanged;
    if (!currently_file) return .target_changed;
    return if (std.mem.eql(u8, state.path(), cfg.file_path)) .target_unchanged else .target_changed;
}

/// Opens `p` and measures it, without touching the live sink.
fn openTarget(io: std.Io, p: []const u8) PrepareError!PreparedSink {
    if (p.len == 0) return error.TargetUnopenable;

    const prev = io.swapCancelProtection(.blocked);
    defer _ = io.swapCancelProtection(prev);

    const dir: std.Io.Dir = .cwd();
    const file = dir.openFile(io, p, .{ .mode = .write_only }) catch |open_err| switch (open_err) {
        error.FileNotFound => dir.createFile(io, p, .{ .truncate = false }) catch
            return error.TargetUnopenable,
        else => return error.TargetUnopenable,
    };
    errdefer file.close(io);

    // A pre-existing nonempty target is appended to, so the new position is
    // its measured length rather than zero.
    const length = file.length(io) catch return error.TargetUnopenable;
    return .{ .file = file, .file_pos = length, .rotate_pending = false };
}

/// Publish: infallible and I/O-free, one hold of the sink lock. Returns the
/// DETACHED old handle, which `retireApply` closes — closing a file is retire
/// work, and doing it here would put a syscall inside the publish.
pub fn publishApply(prepared: PreparedApply) ?std.Io.File {
    var stderr_buf: [64]u8 = undefined;
    _ = std.debug.lockStderr(&stderr_buf);
    defer std.debug.unlockStderr();

    state.threshold = prepared.threshold;
    state.output = prepared.output;
    state.path_len = prepared.path_len;
    @memcpy(state.path_buf[0..state.path_len], prepared.path());
    state.max_files = prepared.max_files;
    state.max_bytes = prepared.max_bytes;

    switch (prepared.case) {
        // The handle and its position and rotation state are not this apply's
        // to move; a broken handle is repaired by the existing per-write
        // recovery, not by a config change.
        .target_unchanged => return null,
        .target_changed => {
            const detached = state.file;
            const sink = prepared.sink.?;
            state.file = sink.file;
            state.file_pos = sink.file_pos;
            state.rotate_pending = sink.rotate_pending;
            return detached;
        },
        .target_removed => {
            const detached = state.file;
            state.file = null;
            state.file_pos = 0;
            state.rotate_pending = false;
            return detached;
        },
    }
}

/// Retire: closes the handle `publishApply` detached, after no writer can
/// reach it — the swap happened under the sink lock, so any writer that held
/// it has already returned.
pub fn retireApply(io: std.Io, detached: ?std.Io.File) void {
    const file = detached orelse return;
    const prev = io.swapCancelProtection(.blocked);
    defer _ = io.swapCancelProtection(prev);
    file.close(io);
}

/// Discards a prepared apply that will not be published, because its commit
/// failed or a sibling owner's prepare did.
pub fn abortApply(io: std.Io, prepared: PreparedApply) void {
    const sink = prepared.sink orelse return;
    const prev = io.swapCancelProtection(.blocked);
    defer _ = io.swapCancelProtection(prev);
    sink.file.close(io);
}

/// The directory the disk monitor should measure for `cfg`: the log file's
/// directory when output is `file`, and null otherwise — with output on
/// stderr or syslog there is no log file to run out of room for.
///
/// A path with no directory component measures the working directory, which is
/// where a bare filename lands.
pub fn logDirname(cfg: model.Logging) ?[]const u8 {
    if (cfg.output != .file) return null;
    if (cfg.file_path.len == 0) return null;
    return std.fs.path.dirname(cfg.file_path) orelse ".";
}

/// Flushes and closes the file, and restores pass-through stderr formatting.
pub fn deinstall() void {
    var stderr_buf: [64]u8 = undefined;
    _ = std.debug.lockStderr(&stderr_buf);
    defer std.debug.unlockStderr();

    closeFileLocked();
    state.installed = false;
    state.output = .stderr;
    state.path_len = 0;
    state.rotate_pending = false;
}

pub fn stats() Stats {
    var stderr_buf: [64]u8 = undefined;
    _ = std.debug.lockStderr(&stderr_buf);
    defer std.debug.unlockStderr();

    return state.stats;
}

// ---------------------------------------------------------------------------
// logFn
// ---------------------------------------------------------------------------

/// Matches the `std.Options.logFn` field type verified at std.zig:132-137.
pub fn logFn(
    comptime message_level: std.log.Level,
    comptime scope: @EnumLiteral(),
    comptime format: []const u8,
    args: anytype,
) void {
    var stderr_buf: [512]u8 = undefined;
    const locked = std.debug.lockStderr(&stderr_buf);
    defer std.debug.unlockStderr();

    if (!state.installed) {
        // The lock is recursive, so `defaultLog` taking it again is sound.
        return std.log.defaultLog(message_level, scope, format, args);
    }
    if (!enabled(message_level, state.threshold)) return;

    var message_buf: [max_message_bytes]u8 = undefined;
    var mw: std.Io.Writer = .fixed(&message_buf);
    // A message longer than the buffer is truncated rather than dropped, but a
    // silently cut line reads as a complete one: the marker says the sink cut
    // it, and the counter says how often that happens.
    const message = if (mw.print(format, args)) mw.buffered() else |_| blk: {
        state.stats.lines_truncated += 1;
        // `@min` rather than a plain subtraction: a `print` that fails on its
        // first chunk leaves fewer bytes buffered than the marker is long, and
        // the marker still has to fit.
        const kept = @min(mw.buffered().len, message_buf.len - truncation_marker.len);
        @memcpy(message_buf[kept..][0..truncation_marker.len], truncation_marker);
        break :blk message_buf[0 .. kept + truncation_marker.len];
    };

    if (comptime isDedupScope(scope) and enabled(message_level, .warn)) {
        comptime std.debug.assert(@tagName(scope).len <= max_scope_name_bytes);
        var key_buf: [max_dedup_key_bytes]u8 = undefined;
        const key = buildKey(&key_buf, @tagName(scope), message);
        if (!state.dedup.admit(std.Io.Clock.awake.now(state.io), key)) {
            state.stats.lines_deduped += 1;
            return;
        }
    }

    const scope_name = comptime boundedScopeName(@tagName(scope));
    const is_default = scope == .default;
    const emitted = switch (state.output) {
        // systemd captures stderr into the journal, which is what `syslog`
        // means for the only supported deployment.
        .stderr, .syslog => writeTerminalLocked(
            locked,
            message_level,
            scope_name,
            is_default,
            message,
        ),
        .file => emitFileLocked(locked, message_level, scope_name, is_default, message),
    };
    if (emitted) state.stats.lines_written += 1;
}

/// A line reaches an operator or it counts as a sink error; it is never both
/// discarded and silent. Each failed write attempt counts once, so a line that
/// fails on the file and again on the stderr fallback counts twice.
fn writeTerminalLocked(
    locked: std.Io.LockedStderr,
    level: std.log.Level,
    scope_name: []const u8,
    is_default: bool,
    message: []const u8,
) bool {
    writeTerminal(locked.terminal(), level, scope_name, is_default, message) catch {
        state.stats.sink_errors += 1;
        return false;
    };
    return true;
}

/// A formatted message can carry bytes chosen by someone else: a query name, an
/// upstream error string. A raw newline in one would forge a second timestamped
/// record, and other control bytes reach a terminal verbatim. `std.log`'s own
/// `defaultLog` (log.zig:132) writes the message raw; this sink escapes on both
/// paths instead, because operators and journald parse its lines.
///
/// `\n`, `\r`, `\t` and `\\` become their two-character C escapes; every other
/// byte below 0x20, and DEL, becomes `\xNN`.
fn writeEscaped(w: *std.Io.Writer, message: []const u8) std.Io.Writer.Error!void {
    const hex = "0123456789abcdef";
    var plain_start: usize = 0;
    for (message, 0..) |byte, i| {
        var hex_buf: [4]u8 = undefined;
        const escape: []const u8 = switch (byte) {
            '\\' => "\\\\",
            '\n' => "\\n",
            '\r' => "\\r",
            '\t' => "\\t",
            0x00...0x08, 0x0b, 0x0c, 0x0e...0x1f, 0x7f => blk: {
                hex_buf = .{ '\\', 'x', hex[byte >> 4], hex[byte & 0x0f] };
                break :blk &hex_buf;
            },
            else => continue,
        };
        try w.writeAll(message[plain_start..i]);
        try w.writeAll(escape);
        plain_start = i + 1;
    }
    try w.writeAll(message[plain_start..]);
}

fn writeTerminal(
    t: std.Io.Terminal,
    level: std.log.Level,
    scope_name: []const u8,
    is_default: bool,
    message: []const u8,
) std.Io.Writer.Error!void {
    t.setColor(switch (level) {
        .err => .red,
        .warn => .yellow,
        .info => .green,
        .debug => .magenta,
    }) catch {};
    t.setColor(.bold) catch {};
    try t.writer.writeAll(levelText(level));
    t.setColor(.reset) catch {};
    t.setColor(.dim) catch {};
    t.setColor(.bold) catch {};
    if (!is_default) {
        try t.writer.writeAll("(");
        try t.writer.writeAll(scope_name);
        try t.writer.writeAll(")");
    }
    try t.writer.writeAll(": ");
    t.setColor(.reset) catch {};
    try writeEscaped(t.writer, message);
    try t.writer.writeAll("\n");
}

fn emitFileLocked(
    locked: std.Io.LockedStderr,
    level: std.log.Level,
    scope_name: []const u8,
    is_default: bool,
    message: []const u8,
) bool {
    // Sized for the longest possible record: an all-`\xNN` message plus the
    // widest header.
    var line_buf: [max_escaped_message_bytes + max_line_header_bytes]u8 = undefined;
    const line = buildLine(
        &line_buf,
        std.Io.Clock.real.now(state.io).toSeconds(),
        level,
        scope_name,
        is_default,
        message,
    ) catch {
        state.stats.sink_errors += 1;
        return writeTerminalLocked(locked, level, scope_name, is_default, message);
    };

    if (!prepareFileLocked(line.len)) {
        return writeTerminalLocked(locked, level, scope_name, is_default, message);
    }
    const file = state.file.?;
    writeLineLocked(file, line) catch {
        // Closing makes the next line reopen: a sink that gave up on the file
        // after one failure would silently stop logging.
        state.stats.sink_errors += 1;
        closeFileLocked();
        return writeTerminalLocked(locked, level, scope_name, is_default, message);
    };
    return true;
}

/// A record is `<unix_seconds> <level>(<scope>): <escaped message>\n`.
///
/// Returns an error rather than a partial record: a line that lost its message
/// or its terminating newline would split or forge an event exactly as an
/// unescaped newline would. With a bounded scope name and a buffer of
/// `max_escaped_message_bytes + max_line_header_bytes` this cannot fail, and
/// the caller treats a failure as a sink error rather than writing the
/// fragment.
fn buildLine(
    buf: []u8,
    unix_seconds: i64,
    level: std.log.Level,
    scope_name: []const u8,
    is_default: bool,
    message: []const u8,
) std.Io.Writer.Error![]const u8 {
    var w: std.Io.Writer = .fixed(buf);
    try w.print("{d} {s}", .{ unix_seconds, levelText(level) });
    if (!is_default) try w.print("({s})", .{scope_name});
    try w.writeAll(": ");
    try writeEscaped(&w, message);
    try w.writeAll("\n");
    return w.buffered();
}

/// True leaves `state.file` open with room for `line_len` more bytes under
/// `state.max_bytes`. False leaves it closed, and the caller falls back to
/// stderr for that line.
///
/// Every path that returns false has already counted exactly one `sink_error`,
/// at the step that failed: the caller must not count it again.
fn prepareFileLocked(line_len: usize) bool {
    // Two passes at most: the second one exists for a file that is already at
    // the bound when it is opened, which rotates before its first line.
    for (0..2) |_| {
        if (state.rotate_pending) {
            if (!rotateLocked()) return false;
            state.rotate_pending = false;
            state.stats.rotations += 1;
        }
        if (state.file == null) openFileLocked();
        if (state.file == null) return false;
        if (!overLimitLocked(line_len)) return true;
        state.rotate_pending = true;
        closeFileLocked();
    }
    // A rotation succeeded and the fresh file is still over the bound: another
    // writer owns the path. No step failed, so nothing has counted yet.
    state.stats.sink_errors += 1;
    return false;
}

/// A line that alone exceeds `max_bytes` is written to an empty file rather
/// than rotated forever: the bound governs the file, and a line is never lost.
fn overLimitLocked(line_len: usize) bool {
    if (state.max_bytes == 0) return false;
    if (state.file_pos == 0) return false;
    return state.file_pos + line_len > state.max_bytes;
}

fn writeLineLocked(file: std.Io.File, line: []const u8) !void {
    const prev = state.io.swapCancelProtection(.blocked);
    defer _ = state.io.swapCancelProtection(prev);

    var write_buf: [512]u8 = undefined;
    var fw = file.writer(state.io, &write_buf);
    try fw.seekTo(state.file_pos);
    try fw.interface.writeAll(line);
    try fw.interface.flush();
    state.file_pos += line.len;
}

/// 0.16.0 has no append mode: the writer seeks to the current end instead.
fn openFileLocked() void {
    // Checked before the cancel-protection swap: an unusable path needs no io.
    if (state.path_len == 0) {
        state.stats.sink_errors += 1;
        return;
    }

    const prev = state.io.swapCancelProtection(.blocked);
    defer _ = state.io.swapCancelProtection(prev);

    const dir: std.Io.Dir = .cwd();
    const p = state.path();
    const file = dir.openFile(state.io, p, .{ .mode = .write_only }) catch |open_err| switch (open_err) {
        error.FileNotFound => dir.createFile(state.io, p, .{ .truncate = false }) catch {
            state.stats.sink_errors += 1;
            return;
        },
        else => {
            state.stats.sink_errors += 1;
            return;
        },
    };
    state.file = file;
    state.file_pos = file.length(state.io) catch {
        state.stats.sink_errors += 1;
        file.close(state.io);
        state.file = null;
        return;
    };
}

fn closeFileLocked() void {
    const file = state.file orelse return;
    file.close(state.io);
    state.file = null;
    state.file_pos = 0;
}

/// `max_files` counts the live file, so the highest kept generation is
/// `max_files - 1`.
///
/// True means the live path is free for a fresh file. False means a step
/// failed and the live path may still hold the oversized file, so the caller
/// must leave it closed: reopening it for append would defeat the disk bound
/// exactly when the filesystem is the thing that failed.
fn rotateLocked() bool {
    const prev = state.io.swapCancelProtection(.blocked);
    defer _ = state.io.swapCancelProtection(prev);

    closeFileLocked();
    rotateStepsLocked() catch {
        // One failed rotation attempt is one sink error, counted here so the
        // caller never counts the same failure a second time.
        state.stats.sink_errors += 1;
        return false;
    };
    return true;
}

const RotateError = error{RotateFailed};

fn rotateStepsLocked() RotateError!void {
    const dir: std.Io.Dir = .cwd();
    const p = state.path();

    if (state.max_files < 2) return deleteLocked(dir, p);

    var from_buf: [max_rotated_path_bytes]u8 = undefined;
    var to_buf: [max_rotated_path_bytes]u8 = undefined;
    const highest = state.max_files - 1;

    const oldest = rotatedName(&to_buf, p, highest) catch return error.RotateFailed;
    try deleteLocked(dir, oldest);

    var n: u8 = highest;
    while (n > 1) : (n -= 1) {
        const from = rotatedName(&from_buf, p, n - 1) catch return error.RotateFailed;
        const to = rotatedName(&to_buf, p, n) catch return error.RotateFailed;
        try renameLocked(dir, from, to);
    }

    const first = rotatedName(&to_buf, p, 1) catch return error.RotateFailed;
    try renameLocked(dir, p, first);
}

const RotateFault = enum { none, fail_delete, fail_rename };

/// The two rotation steps fail only when the filesystem does, which no unit
/// test can arrange on demand, so the failure paths are driven through this
/// seam instead. The storage exists in a test build only, and
/// `rotateFaultTripped` reduces to `false` everywhere else.
const rotate_fault_seam = if (builtin.is_test) struct {
    var fault: RotateFault = .none;
} else struct {};

fn rotateFaultTripped(comptime which: RotateFault) bool {
    if (!builtin.is_test) return false;
    return rotate_fault_seam.fault == which;
}

/// A generation that does not exist yet is not a failure: the first rotations
/// of a fresh log directory find nothing to delete.
fn deleteLocked(dir: std.Io.Dir, p: []const u8) RotateError!void {
    if (rotateFaultTripped(.fail_delete)) return error.RotateFailed;
    dir.deleteFile(state.io, p) catch |err| switch (err) {
        error.FileNotFound => {},
        else => return error.RotateFailed,
    };
}

fn renameLocked(dir: std.Io.Dir, from: []const u8, to: []const u8) RotateError!void {
    if (rotateFaultTripped(.fail_rename)) return error.RotateFailed;
    dir.rename(from, dir, to, state.io) catch |err| switch (err) {
        error.FileNotFound => {},
        else => return error.RotateFailed,
    };
}

// ---------------------------------------------------------------------------
// Tests
//
// The pure pieces only. `logFn` is never exercised: installing a sink under the
// test runner would eat the harness's own output. File behaviour is S8's. The
// rotation failure paths need a filesystem that fails a delete or a rename on
// demand, which `rotate_fault_seam` supplies.
// ---------------------------------------------------------------------------

const testing = std.testing;

fn ts(nanoseconds: i96) std.Io.Timestamp {
    return .{ .nanoseconds = nanoseconds };
}

test "toStdLevel maps every model level" {
    try testing.expectEqual(std.log.Level.err, toStdLevel(.err));
    try testing.expectEqual(std.log.Level.warn, toStdLevel(.warn));
    try testing.expectEqual(std.log.Level.info, toStdLevel(.info));
    try testing.expectEqual(std.log.Level.debug, toStdLevel(.debug));
}

test "enabled admits at and above the threshold only" {
    try testing.expect(enabled(.err, .info));
    try testing.expect(enabled(.warn, .info));
    try testing.expect(enabled(.info, .info));
    try testing.expect(!enabled(.debug, .info));
    try testing.expect(enabled(.err, .err));
    try testing.expect(!enabled(.warn, .err));
    try testing.expect(enabled(.debug, .debug));
}

test "isDedupScope selects exactly the four upstream scopes and tls_server" {
    try testing.expect(isDedupScope(.doh_client));
    try testing.expect(isDedupScope(.dot_client));
    try testing.expect(isDedupScope(.pool));
    try testing.expect(isDedupScope(.forward_client));
    try testing.expect(isDedupScope(.tls_server));
    try testing.expect(!isDedupScope(.default));
    try testing.expect(!isDedupScope(.cache));
    try testing.expect(!isDedupScope(.dot_server));
    try testing.expect(!isDedupScope(.doh_server));
}

test "buildKey separates the scope from the message" {
    var buf: [max_dedup_key_bytes]u8 = undefined;
    const key = buildKey(&buf, "pool", "no upstream available");
    try testing.expectEqualStrings("pool\x00no upstream available", key);
}

test "buildKey truncates the message to 96 bytes" {
    var buf: [max_dedup_key_bytes]u8 = undefined;
    const long = "x" ** 200;
    const key = buildKey(&buf, "pool", long);
    try testing.expectEqual(@as(usize, 4 + 1 + max_message_key_bytes), key.len);
}

test "dedup admits a key it has not seen" {
    var table: DedupTable = .{};
    var buf: [max_dedup_key_bytes]u8 = undefined;
    const key = buildKey(&buf, "pool", "upstream timed out");
    try testing.expect(table.admit(ts(0), key));
}

test "dedup drops a repeat inside the window" {
    var table: DedupTable = .{};
    var buf: [max_dedup_key_bytes]u8 = undefined;
    const key = buildKey(&buf, "doh_client", "connect failed");
    try testing.expect(table.admit(ts(1_000), key));
    try testing.expect(!table.admit(ts(1_000), key));
    try testing.expect(!table.admit(ts(1_000 + dedup_window_ns - 1), key));
}

test "dedup admits again once the window has passed" {
    var table: DedupTable = .{};
    var buf: [max_dedup_key_bytes]u8 = undefined;
    const key = buildKey(&buf, "dot_client", "handshake failed");
    try testing.expect(table.admit(ts(0), key));
    try testing.expect(table.admit(ts(dedup_window_ns), key));
    // The admitted repeat restamps, so the window restarts from it.
    try testing.expect(!table.admit(ts(dedup_window_ns + 1), key));
}

test "dedup keys the same message under different scopes apart" {
    var table: DedupTable = .{};
    var pool_buf: [max_dedup_key_bytes]u8 = undefined;
    var doh_buf: [max_dedup_key_bytes]u8 = undefined;
    const pool_key = buildKey(&pool_buf, "pool", "connect failed");
    const doh_key = buildKey(&doh_buf, "doh_client", "connect failed");
    try testing.expect(table.admit(ts(0), pool_key));
    try testing.expect(table.admit(ts(0), doh_key));
    try testing.expect(!table.admit(ts(0), pool_key));
}

test "dedup collapses messages sharing their first 96 bytes" {
    var table: DedupTable = .{};
    var first_buf: [max_dedup_key_bytes]u8 = undefined;
    var second_buf: [max_dedup_key_bytes]u8 = undefined;
    const prefix = "y" ** max_message_key_bytes;
    const first = buildKey(&first_buf, "pool", prefix ++ " alpha");
    const second = buildKey(&second_buf, "pool", prefix ++ " beta");
    try testing.expect(table.admit(ts(0), first));
    try testing.expect(!table.admit(ts(0), second));
}

test "dedup replaces the oldest stamp when the table is full" {
    var table: DedupTable = .{};
    var buf: [max_dedup_key_bytes]u8 = undefined;
    var message_buf: [16]u8 = undefined;

    for (0..dedup_slots) |i| {
        const message = try std.fmt.bufPrint(&message_buf, "failure {d}", .{i});
        const key = buildKey(&buf, "pool", message);
        try testing.expect(table.admit(ts(@intCast(i)), key));
    }

    const newcomer = buildKey(&buf, "pool", "failure 999");
    try testing.expect(table.admit(ts(dedup_slots), newcomer));

    // Slot 0 held the oldest stamp, so its key was forgotten and is admitted
    // again inside what would otherwise still be its window. Admitting it
    // evicts the next-oldest in turn, so the survivor asserted below is the
    // newest of the original set rather than the second-oldest.
    const evicted = buildKey(&buf, "pool", "failure 0");
    try testing.expect(table.admit(ts(dedup_slots), evicted));

    const retained = buildKey(&buf, "pool", "failure 63");
    try testing.expect(!table.admit(ts(dedup_slots), retained));
}

test "dedup arithmetic holds at i96 scale" {
    var table: DedupTable = .{};
    var buf: [max_dedup_key_bytes]u8 = undefined;
    const key = buildKey(&buf, "forward_client", "zone resolver unreachable");
    const far: i96 = 1 << 80;
    try testing.expect(table.admit(ts(far), key));
    try testing.expect(!table.admit(ts(far + dedup_window_ns - 1), key));
    try testing.expect(table.admit(ts(far + dedup_window_ns), key));
}

// ---------------------------------------------------------------------------
// Escaping
// ---------------------------------------------------------------------------

fn escapeToBuf(buf: []u8, message: []const u8) ![]const u8 {
    var w: std.Io.Writer = .fixed(buf);
    try writeEscaped(&w, message);
    return w.buffered();
}

test "writeEscaped passes printable text through unchanged" {
    var buf: [64]u8 = undefined;
    try testing.expectEqualStrings(
        "upstream 10.0.0.1: timed out",
        try escapeToBuf(&buf, "upstream 10.0.0.1: timed out"),
    );
}

test "writeEscaped denies a forged record" {
    // Without the escape this query name would produce a second line that
    // parses as its own timestamped record.
    var buf: [128]u8 = undefined;
    const forged = "blocked\n1700000000 error(pool): all upstreams down";
    const escaped = try escapeToBuf(&buf, forged);
    try testing.expect(std.mem.indexOfScalar(u8, escaped, '\n') == null);
    try testing.expectEqualStrings(
        "blocked\\n1700000000 error(pool): all upstreams down",
        escaped,
    );
}

test "writeEscaped escapes the named control bytes and the backslash" {
    var buf: [64]u8 = undefined;
    try testing.expectEqualStrings(
        "a\\nb\\rc\\td\\\\e",
        try escapeToBuf(&buf, "a\nb\rc\td\\e"),
    );
}

test "writeEscaped hex-escapes every other control byte" {
    var buf: [64]u8 = undefined;
    try testing.expectEqualStrings("\\x00", try escapeToBuf(&buf, "\x00"));
    try testing.expectEqualStrings("\\x1b[31m", try escapeToBuf(&buf, "\x1b[31m"));
    try testing.expectEqualStrings("\\x7f", try escapeToBuf(&buf, "\x7f"));
    try testing.expectEqualStrings("\\x0b\\x0c", try escapeToBuf(&buf, "\x0b\x0c"));
}

test "writeEscaped leaves the printable bytes alone" {
    var buf: [256]u8 = undefined;
    // 0x20 through 0x7e: everything printable, DEL excluded.
    var message: [95]u8 = undefined;
    for (&message, 0..) |*byte, i| byte.* = @intCast(0x20 + i);
    const escaped = try escapeToBuf(&buf, &message);
    try testing.expectEqual(message.len + 1, escaped.len); // the one backslash
    try testing.expectEqual(@as(usize, 1), std.mem.count(u8, escaped, "\\\\"));
}

test "writeEscaped never exceeds four bytes per input byte" {
    var buf: [max_escaped_message_bytes]u8 = undefined;
    const worst = "\x01" ** max_message_bytes;
    const escaped = try escapeToBuf(&buf, worst);
    try testing.expectEqual(max_message_bytes * 4, escaped.len);
}

test "writeEscaped handles high bytes and an empty message" {
    var buf: [32]u8 = undefined;
    try testing.expectEqualStrings("", try escapeToBuf(&buf, ""));
    try testing.expectEqualStrings("\xc3\xa9", try escapeToBuf(&buf, "\xc3\xa9"));
}

// ---------------------------------------------------------------------------
// Record construction
// ---------------------------------------------------------------------------

test "buildLine writes one record ending in a newline" {
    var buf: [256]u8 = undefined;
    try testing.expectEqualStrings(
        "1700000000 warning(pool): no upstream available\n",
        try buildLine(&buf, 1700000000, .warn, "pool", false, "no upstream available"),
    );
}

test "buildLine omits the scope of the default scope" {
    var buf: [256]u8 = undefined;
    try testing.expectEqualStrings(
        "42 info: listening on 0.0.0.0:53\n",
        try buildLine(&buf, 42, .info, "default", true, "listening on 0.0.0.0:53"),
    );
}

test "buildLine escapes the message, so a record is one line" {
    var buf: [256]u8 = undefined;
    const line = try buildLine(&buf, 1, .err, "pool", false, "a\nb");
    try testing.expectEqualStrings("1 error(pool): a\\nb\n", line);
    try testing.expectEqual(@as(usize, 1), std.mem.count(u8, line, "\n"));
}

test "boundedScopeName truncates a scope name to the header allowance" {
    try testing.expectEqualStrings("pool", boundedScopeName("pool"));
    try testing.expectEqual(max_scope_name_bytes, boundedScopeName("s" ** 200).len);
}

test "an oversized scope name still yields a well-formed record" {
    var buf: [max_escaped_message_bytes + max_line_header_bytes]u8 = undefined;
    const line = try buildLine(
        &buf,
        1700000000,
        .warn,
        boundedScopeName("s" ** 200),
        false,
        "the message survives",
    );
    try testing.expectEqualStrings(
        "1700000000 warning(" ++ "s" ** max_scope_name_bytes ++ "): the message survives\n",
        line,
    );
}

test "the header allowance covers the widest possible header" {
    // The longest timestamp, the longest level text, a bounded scope name, and
    // the terminator, with no message at all.
    var buf: [max_line_header_bytes]u8 = undefined;
    const line = try buildLine(
        &buf,
        std.math.minInt(i64),
        .warn,
        boundedScopeName("s" ** 200),
        false,
        "",
    );
    try testing.expectEqual(max_line_header_bytes, line.len);
}

test "buildLine reports a full buffer instead of returning a partial record" {
    var buf: [16]u8 = undefined;
    try testing.expectError(
        error.WriteFailed,
        buildLine(&buf, 1700000000, .warn, "pool", false, "this does not fit"),
    );
}

test "a failed open counts exactly one sink error" {
    var stderr_buf: [64]u8 = undefined;
    _ = std.debug.lockStderr(&stderr_buf);
    const saved_stats = state.stats;
    const saved_path_len = state.path_len;
    const saved_file = state.file;
    const saved_pending = state.rotate_pending;
    defer {
        state.stats = saved_stats;
        state.path_len = saved_path_len;
        state.file = saved_file;
        state.rotate_pending = saved_pending;
        std.debug.unlockStderr();
    }

    // An empty path fails the open before it reaches `state.io`, which no test
    // ever installs. One failed attempt is one `sink_error`: the open counts
    // it and `prepareFileLocked` leaves it at that.
    state.stats = .{};
    state.path_len = 0;
    state.file = null;
    state.rotate_pending = false;

    try testing.expect(!prepareFileLocked(64));
    try testing.expectEqual(@as(u64, 1), state.stats.sink_errors);
    try testing.expectEqual(@as(u64, 0), state.stats.lines_written);
}

/// The shared body of the two rotation-failure tests: they differ only in which
/// step is made to fail and in how many `max_files` it takes to reach it.
///
/// `state.io` is normally installed by `install`, which no test calls, so a real
/// one is put in place for the duration: `rotateLocked` swaps cancel protection
/// on it before the first step runs, and the `fail_rename` case deletes the
/// oldest generation for real before it reaches the rename. That delete is why
/// the path points into a fresh tmp directory rather than at any real log.
fn expectRotationFailureCounted(fault: RotateFault, max_files: u8) !void {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();

    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buf: [128]u8 = undefined;
    const p = try std.fmt.bufPrint(&path_buf, ".zig-cache/tmp/{s}/nxdns.log", .{tmp.sub_path});

    var stderr_buf: [64]u8 = undefined;
    _ = std.debug.lockStderr(&stderr_buf);
    const saved_stats = state.stats;
    const saved_path_len = state.path_len;
    const saved_max_files = state.max_files;
    const saved_file = state.file;
    const saved_pending = state.rotate_pending;
    const saved_io = state.io;
    defer {
        rotate_fault_seam.fault = .none;
        state.stats = saved_stats;
        state.path_len = saved_path_len;
        state.max_files = saved_max_files;
        state.file = saved_file;
        state.rotate_pending = saved_pending;
        state.io = saved_io;
        std.debug.unlockStderr();
    }

    state.stats = .{};
    state.io = threaded.io();
    state.path_len = p.len;
    @memcpy(state.path_buf[0..p.len], p);
    state.max_files = max_files;
    state.file = null;
    state.rotate_pending = true;
    rotate_fault_seam.fault = fault;

    try testing.expect(!prepareFileLocked(64));
    try testing.expectEqual(@as(u64, 1), state.stats.sink_errors);
    try testing.expectEqual(@as(u64, 0), state.stats.rotations);
    // The oversized file stays closed and the rotation stays owed, so the next
    // line retries the rotation instead of appending past `max_bytes`.
    try testing.expect(state.rotate_pending);
    try testing.expectEqual(@as(?std.Io.File, null), state.file);
}

test "a failed rotation delete counts exactly one sink error" {
    // `max_files` below 2 keeps no generations, so the whole rotation is the one
    // delete of the live path.
    try expectRotationFailureCounted(.fail_delete, 1);
}

test "a failed rotation rename counts exactly one sink error" {
    // `max_files` of 2 keeps generation 1, so the steps are one delete of the
    // oldest generation followed by the rename of the live path onto it.
    try expectRotationFailureCounted(.fail_rename, 2);
}

test "rotatedName appends the generation" {
    var buf: [max_rotated_path_bytes]u8 = undefined;
    try testing.expectEqualStrings(
        "/var/log/nxdns/nxdns.log.1",
        try rotatedName(&buf, "/var/log/nxdns/nxdns.log", 1),
    );
    try testing.expectEqualStrings(
        "/var/log/nxdns/nxdns.log.4",
        try rotatedName(&buf, "/var/log/nxdns/nxdns.log", 4),
    );
    try testing.expectEqualStrings(
        "nxdns.log.255",
        try rotatedName(&buf, "nxdns.log", 255),
    );
}

test "rotatedName reports a buffer too small for the name" {
    var buf: [4]u8 = undefined;
    try testing.expectError(error.NameTooLong, rotatedName(&buf, "nxdns.log", 1));
}

test "rotation covers max_files - 1 generations" {
    // The shift loop renames .{n-1} to .{n} down to .1, so `max_files` files
    // exist in total: the live file plus generations 1 through max_files - 1.
    var buf: [max_rotated_path_bytes]u8 = undefined;
    const max_files: u8 = 5;
    var names: [4][]const u8 = undefined;
    var storage: [4][max_rotated_path_bytes]u8 = undefined;
    for (1..max_files) |n| {
        const name = try rotatedName(&buf, "nxdns.log", @intCast(n));
        @memcpy(storage[n - 1][0..name.len], name);
        names[n - 1] = storage[n - 1][0..name.len];
    }
    try testing.expectEqualStrings("nxdns.log.1", names[0]);
    try testing.expectEqualStrings("nxdns.log.4", names[3]);
}

test "a message over the buffer is marked and counted" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();

    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buf: [128]u8 = undefined;
    const p = try std.fmt.bufPrint(&path_buf, ".zig-cache/tmp/{s}/nxdns.log", .{tmp.sub_path});

    var stderr_buf: [64]u8 = undefined;
    _ = std.debug.lockStderr(&stderr_buf);
    const saved = state;
    defer {
        closeFileLocked();
        state = saved;
        std.debug.unlockStderr();
    }

    state.stats = .{};
    state.io = threaded.io();
    state.threshold = .info;
    state.output = .file;
    state.path_len = p.len;
    @memcpy(state.path_buf[0..p.len], p);
    state.max_files = 0;
    state.max_bytes = 0;
    state.file = null;
    state.file_pos = 0;
    state.rotate_pending = false;
    state.installed = true;

    const oversized: [max_message_bytes + 100]u8 = @splat('a');
    logFn(.info, .default, "{s}", .{&oversized});

    try testing.expectEqual(@as(u64, 1), state.stats.lines_truncated);
    try testing.expectEqual(@as(u64, 1), state.stats.lines_written);
    try testing.expectEqual(@as(u64, 0), state.stats.sink_errors);

    closeFileLocked();
    var read_buf: [max_message_bytes * 2]u8 = undefined;
    const written = try std.Io.Dir.cwd().readFile(state.io, p, &read_buf);
    try testing.expect(std.mem.endsWith(u8, written, truncation_marker ++ "\n"));
    // The marker replaces the last bytes of the message rather than extending
    // it, so the record still fits what the buffer held.
    const colon = std.mem.indexOf(u8, written, ": ").?;
    try testing.expectEqual(max_message_bytes, written[colon + 2 ..].len - 1);
}

// ---------------------------------------------------------------------------
// hot apply (milestone-34 S3.5)
// ---------------------------------------------------------------------------

/// The sink is one process-wide `state` behind the stderr lock, so an apply
/// test has to save it, drive the apply, and put it back. Nothing here holds
/// the lock across an apply call: `classifyApply` and `publishApply` take it
/// themselves.
const ApplyFixture = struct {
    threaded: std.Io.Threaded,
    tmp: testing.TmpDir,
    saved: State,

    fn init(self: *ApplyFixture) void {
        self.threaded = .init(testing.allocator, .{});
        self.tmp = testing.tmpDir(.{});

        var stderr_buf: [64]u8 = undefined;
        _ = std.debug.lockStderr(&stderr_buf);
        defer std.debug.unlockStderr();
        self.saved = state;
        state = .{};
        state.io = self.threaded.io();
    }

    fn deinit(self: *ApplyFixture) void {
        {
            var stderr_buf: [64]u8 = undefined;
            _ = std.debug.lockStderr(&stderr_buf);
            defer std.debug.unlockStderr();
            if (state.file) |f| f.close(self.threaded.io());
            state = self.saved;
        }
        self.tmp.cleanup();
        self.threaded.deinit();
    }

    fn io(self: *ApplyFixture) std.Io {
        return self.threaded.io();
    }

    fn path(self: *ApplyFixture, buf: []u8, name: []const u8) []const u8 {
        return std.fmt.bufPrint(buf, ".zig-cache/tmp/{s}/{s}", .{ self.tmp.sub_path, name }) catch unreachable;
    }

    /// Puts the sink on `p` with a real open handle, the way a running server
    /// with `output = file` sits.
    fn openOn(self: *ApplyFixture, p: []const u8) void {
        var stderr_buf: [64]u8 = undefined;
        _ = std.debug.lockStderr(&stderr_buf);
        defer std.debug.unlockStderr();

        state.installed = true;
        state.output = .file;
        state.path_len = p.len;
        @memcpy(state.path_buf[0..p.len], p);
        state.max_bytes = 1 << 20;
        state.max_files = 3;
        openFileLocked();
        _ = self;
    }

    fn snapshot(self: *ApplyFixture) State {
        _ = self;
        var stderr_buf: [64]u8 = undefined;
        _ = std.debug.lockStderr(&stderr_buf);
        defer std.debug.unlockStderr();
        return state;
    }

    fn setHandleState(self: *ApplyFixture, file: ?std.Io.File, file_pos: u64, rotate_pending: bool) void {
        _ = self;
        var stderr_buf: [64]u8 = undefined;
        _ = std.debug.lockStderr(&stderr_buf);
        defer std.debug.unlockStderr();
        state.file = file;
        state.file_pos = file_pos;
        state.rotate_pending = rotate_pending;
    }

    /// Writes one record through the live handle, as `emitFileLocked` does.
    fn writeThroughSink(self: *ApplyFixture, text: []const u8) !void {
        _ = self;
        var stderr_buf: [64]u8 = undefined;
        _ = std.debug.lockStderr(&stderr_buf);
        defer std.debug.unlockStderr();
        try writeLineLocked(state.file.?, text);
    }

    fn read(self: *ApplyFixture, buf: []u8, name: []const u8) ![]u8 {
        return self.tmp.dir.readFileAlloc(self.io(), name, testing.allocator, .limited(buf.len)) catch |err| return err;
    }
};

fn fileCfg(p: []const u8) model.Logging {
    return .{ .output = .file, .file_path = p, .level = .info, .max_files = 3, .max_size_mb = 1 };
}

test "a bad target path is refused at prepare and the live sink is untouched" {
    var fx: ApplyFixture = undefined;
    fx.init();
    defer fx.deinit();

    var buf: [160]u8 = undefined;
    const live = fx.path(&buf, "nxdns.log");
    fx.openOn(live);
    const before = fx.snapshot();
    try testing.expect(before.file != null);

    // A directory that does not exist: the open and the create both fail.
    var bad_buf: [200]u8 = undefined;
    const bad = fx.path(&bad_buf, "no-such-dir/nxdns.log");
    try testing.expectError(
        error.TargetUnopenable,
        prepareApplyForTest(fx.io(), fileCfg(bad), 1 << 20),
    );

    const after = fx.snapshot();
    try testing.expectEqual(before.file.?.handle, after.file.?.handle);
    try testing.expectEqualStrings(live, after.path());
}

test "a target change publishes without closing, and retire closes the old handle" {
    var fx: ApplyFixture = undefined;
    fx.init();
    defer fx.deinit();

    var first_buf: [160]u8 = undefined;
    var second_buf: [160]u8 = undefined;
    const first = fx.path(&first_buf, "first.log");
    const second = fx.path(&second_buf, "second.log");
    fx.openOn(first);
    const before = fx.snapshot();

    const prepared = try prepareApplyForTest(fx.io(), fileCfg(second), 1 << 20);
    try testing.expectEqual(ApplyCase.target_changed, prepared.case);

    const detached = publishApply(prepared);
    const after = fx.snapshot();

    // Publish swapped the handle and left the old one OPEN: the old descriptor
    // still writes, which it could not if publish had closed it.
    try testing.expectEqual(before.file.?.handle, detached.?.handle);
    try testing.expect(after.file.?.handle != detached.?.handle);
    try testing.expectEqualStrings(second, after.path());
    try testing.expectEqual(@as(u64, 0), after.file_pos);
    try testing.expect(!after.rotate_pending);

    var old_writer_buf: [64]u8 = undefined;
    var ow = detached.?.writer(fx.io(), &old_writer_buf);
    try ow.interface.writeAll("still open\n");
    try ow.interface.flush();

    retireApply(fx.io(), detached);

    // The new target receives lines.
    try fx.writeThroughSink("1 info: on the new target\n");
    var read_buf: [256]u8 = undefined;
    const contents = try fx.read(&read_buf, "second.log");
    defer testing.allocator.free(contents);
    try testing.expectEqualStrings("1 info: on the new target\n", contents);
}

test "switching to a pre-existing nonempty file starts at its measured length" {
    var fx: ApplyFixture = undefined;
    fx.init();
    defer fx.deinit();

    const existing = "already here\n";
    try fx.tmp.dir.writeFile(fx.io(), .{ .sub_path = "kept.log", .data = existing });

    var first_buf: [160]u8 = undefined;
    var kept_buf: [160]u8 = undefined;
    fx.openOn(fx.path(&first_buf, "first.log"));
    const kept = fx.path(&kept_buf, "kept.log");

    const prepared = try prepareApplyForTest(fx.io(), fileCfg(kept), 1 << 20);
    try testing.expectEqual(@as(u64, existing.len), prepared.sink.?.file_pos);
    retireApply(fx.io(), publishApply(prepared));

    try testing.expectEqual(@as(u64, existing.len), fx.snapshot().file_pos);

    // Inheriting the old position would have overwritten the existing bytes.
    try fx.writeThroughSink("appended\n");
    var read_buf: [256]u8 = undefined;
    const contents = try fx.read(&read_buf, "kept.log");
    defer testing.allocator.free(contents);
    try testing.expectEqualStrings(existing ++ "appended\n", contents);
}

test "a target change does not inherit a pending rotation" {
    var fx: ApplyFixture = undefined;
    fx.init();
    defer fx.deinit();

    var first_buf: [160]u8 = undefined;
    var second_buf: [160]u8 = undefined;
    fx.openOn(fx.path(&first_buf, "first.log"));
    const second = fx.path(&second_buf, "second.log");

    // A rotation the old target owed and never completed: the handle is closed
    // and the rotation is still pending.
    const stale = fx.snapshot().file.?;
    stale.close(fx.io());
    fx.setHandleState(null, 0, true);

    const prepared = try prepareApplyForTest(fx.io(), fileCfg(second), 1 << 20);
    try testing.expect(!prepared.sink.?.rotate_pending);
    retireApply(fx.io(), publishApply(prepared));

    const after = fx.snapshot();
    try testing.expect(!after.rotate_pending);
    try testing.expect(after.file != null);
    try testing.expectEqual(@as(u64, 0), after.file_pos);
}

test "a file to stderr apply detaches the handle and clears position and rotation" {
    var fx: ApplyFixture = undefined;
    fx.init();
    defer fx.deinit();

    var buf: [160]u8 = undefined;
    const live = fx.path(&buf, "nxdns.log");
    fx.openOn(live);
    fx.setHandleState(fx.snapshot().file, 4_096, true);
    const before = fx.snapshot();

    const prepared = try prepareApplyForTest(fx.io(), .{
        .output = .stderr,
        .file_path = live,
        .level = .warn,
    }, 1 << 20);
    try testing.expectEqual(ApplyCase.target_removed, prepared.case);

    const detached = publishApply(prepared);
    const after = fx.snapshot();

    try testing.expectEqual(before.file.?.handle, detached.?.handle);
    try testing.expectEqual(@as(?std.Io.File, null), after.file);
    try testing.expectEqual(@as(u64, 0), after.file_pos);
    try testing.expect(!after.rotate_pending);
    try testing.expectEqual(model.LogOutput.stderr, after.output);
    // The path still moves, so a later switch back to file opens the right one.
    try testing.expectEqualStrings(live, after.path());

    retireApply(fx.io(), detached);
}

test "a same-path apply changes only config fields, whatever the handle is doing" {
    var fx: ApplyFixture = undefined;
    fx.init();
    defer fx.deinit();

    var buf: [160]u8 = undefined;
    const live = fx.path(&buf, "nxdns.log");
    fx.openOn(live);

    // Publish takes the same lock a rotation and a write-failure closure hold,
    // so the only reachable interleavings are "before publish" and "after".
    // Both leave the handle state the apply must not touch; these are the two
    // states each of them leaves behind.
    const handle_states = [_]struct { file: bool, pos: u64, pending: bool }{
        // Mid-rotation: handle closed, rotation owed.
        .{ .file = false, .pos = 0, .pending = true },
        // Healthy and part-written.
        .{ .file = true, .pos = 8_192, .pending = false },
    };

    const open_handle = fx.snapshot().file.?;
    for (handle_states) |want| {
        fx.setHandleState(if (want.file) open_handle else null, want.pos, want.pending);

        const prepared = try prepareApplyForTest(fx.io(), .{
            .output = .file,
            .file_path = live,
            .level = .debug,
            .max_files = 9,
            .max_size_mb = 7,
        }, 4_242);
        try testing.expectEqual(ApplyCase.target_unchanged, prepared.case);

        // Nothing detached, so retire has nothing to close.
        try testing.expectEqual(@as(?std.Io.File, null), publishApply(prepared));

        const after = fx.snapshot();
        try testing.expectEqual(want.pos, after.file_pos);
        try testing.expectEqual(want.pending, after.rotate_pending);
        try testing.expectEqual(want.file, after.file != null);
        // The config fields did move.
        try testing.expectEqual(std.log.Level.debug, after.threshold);
        try testing.expectEqual(@as(u8, 9), after.max_files);
        try testing.expectEqual(@as(u64, 4_242), after.max_bytes);
    }

    fx.setHandleState(open_handle, 0, false);
}

test "a file_path change while output is stderr updates the config and touches no handle" {
    var fx: ApplyFixture = undefined;
    fx.init();
    defer fx.deinit();

    var buf: [160]u8 = undefined;
    const later = fx.path(&buf, "later.log");

    const prepared = try prepareApplyForTest(fx.io(), .{
        .output = .syslog,
        .file_path = later,
        .level = .info,
    }, 1 << 20);
    try testing.expectEqual(ApplyCase.target_unchanged, prepared.case);
    try testing.expectEqual(@as(?std.Io.File, null), publishApply(prepared));

    const after = fx.snapshot();
    try testing.expectEqual(@as(?std.Io.File, null), after.file);
    try testing.expectEqualStrings(later, after.path());

    // The later switch to file opens exactly that path.
    const to_file = try prepareApplyForTest(fx.io(), fileCfg(later), 1 << 20);
    try testing.expectEqual(ApplyCase.target_changed, to_file.case);
    retireApply(fx.io(), publishApply(to_file));
    try testing.expect(fx.snapshot().file != null);
}

test "an aborted apply closes the target it opened" {
    var fx: ApplyFixture = undefined;
    fx.init();
    defer fx.deinit();

    var buf: [160]u8 = undefined;
    const target = fx.path(&buf, "never.log");

    // The commit failed, so the prepared target must not leak its descriptor.
    const prepared = try prepareApplyForTest(fx.io(), fileCfg(target), 1 << 20);
    abortApply(fx.io(), prepared);

    const after = fx.snapshot();
    try testing.expectEqual(@as(?std.Io.File, null), after.file);
    try testing.expectEqual(model.LogOutput.stderr, after.output);
}

test "logDirname follows output and file_path in both directions" {
    try testing.expectEqualStrings("/var/log/nxdns", logDirname(.{
        .output = .file,
        .file_path = "/var/log/nxdns/nxdns.log",
    }).?);
    // A bare filename lands in the working directory.
    try testing.expectEqualStrings(".", logDirname(.{
        .output = .file,
        .file_path = "nxdns.log",
    }).?);
    // Output away from file stops the measurement whatever the path says.
    try testing.expectEqual(@as(?[]const u8, null), logDirname(.{
        .output = .stderr,
        .file_path = "/var/log/nxdns/nxdns.log",
    }));
    try testing.expectEqual(@as(?[]const u8, null), logDirname(.{
        .output = .syslog,
        .file_path = "/var/log/nxdns/nxdns.log",
    }));
}