nxdns/src/storage/repositories/events_repo.zig

//! `operational_events` in `config.db` — the SQL behind `storage/events.zig`.
//!
//! One row is one *episode*: a failure of one subject that opened at
//! `first_seen`, was last seen at `last_seen`, has happened `occurrences` times,
//! and is either still going (`resolved_at IS NULL`) or over. A subject that
//! fails, recovers and fails again gets a second row, never a reopened first
//! one, so the history reads as a sequence of episodes rather than one row whose
//! meaning changes.
//!
//! The partial unique index on `(code, subject_key) WHERE resolved_at IS NULL`
//! is what enforces that: at most one active row per subject, and any number of
//! resolved ones. Every function here is written to work with it rather than
//! around it.
//!
//! Only `events.Store` calls this in production, and it holds its mutex across
//! every call — nothing here takes a lock or assumes one. Strings are borrowed
//! for the duration of the call: `Stmt.bindText` binds with `SQLITE_TRANSIENT`,
//! so SQLite copies before the function returns.
//!
//! Nothing here retries and nothing here logs. The store owns what a failed
//! write means.

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

const db = @import("../db.zig");

/// The stored widths. `events.Store` re-exports them as its own caps, so a
/// value that reaches a column is always a value the column can hold, and the
/// mirror's fixed row buffers below are always big enough for a stored row.
pub const code_capacity = 32;
pub const key_capacity = 256;

/// Ruling 11's page cap, shared with `/api/queries` so a client cannot ask
/// either endpoint for a whole table.
pub const max_limit: u32 = 1000;

pub const severity_warning = "warning";
pub const severity_error = "error";

// ---------------------------------------------------------------------------
// writes
// ---------------------------------------------------------------------------

const insert_sql =
    \\INSERT INTO operational_events
    \\  (code, subject_key, subject_label, severity,
    \\   first_seen, last_seen, occurrences, resolved_at, detail)
    \\VALUES (?1, ?2, ?3, ?4, ?5, ?5, 1, ?6, ?7)
;

/// Opens an episode and returns its row id. `resolved_at` is NULL, so the
/// partial unique index refuses a second active row for the same subject —
/// which is why the store probes before it inserts whenever it cannot see the
/// whole active set.
pub fn insertActive(
    database: *db.Db,
    now_s: i64,
    code: []const u8,
    subject_key: []const u8,
    subject_label: []const u8,
    severity: []const u8,
    detail: []const u8,
) db.Error!i64 {
    try insertRow(database, now_s, code, subject_key, subject_label, severity, detail, null);
    return database.lastInsertRowid();
}

/// A one-shot event: something that happened once and is already over
/// (`query_log.recreated`). It never becomes an active row, so it never
/// competes for the partial unique index and never needs resolving.
pub fn insertResolved(
    database: *db.Db,
    now_s: i64,
    code: []const u8,
    subject_key: []const u8,
    subject_label: []const u8,
    severity: []const u8,
    detail: []const u8,
) db.Error!void {
    return insertRow(database, now_s, code, subject_key, subject_label, severity, detail, now_s);
}

fn insertRow(
    database: *db.Db,
    now_s: i64,
    code: []const u8,
    subject_key: []const u8,
    subject_label: []const u8,
    severity: []const u8,
    detail: []const u8,
    resolved_at: ?i64,
) db.Error!void {
    var stmt = try database.prepare(insert_sql);
    defer stmt.deinit();
    try stmt.bindText(1, code);
    try stmt.bindText(2, subject_key);
    try stmt.bindText(3, subject_label);
    try stmt.bindText(4, severity);
    try stmt.bindInt(5, now_s);
    if (resolved_at) |at| try stmt.bindInt(6, at) else try stmt.bindNull(6);
    try stmt.bindText(7, detail);
    return stmt.exec();
}

/// The severity `CASE` is the "raise, never lower" rule: a warning episode that
/// takes an error becomes an error episode, and an error episode that takes a
/// warning stays an error. Written in SQL rather than read-modify-write so the
/// rule holds in one statement.
const touch_sql =
    \\UPDATE operational_events
    \\   SET last_seen = ?2,
    \\       occurrences = occurrences + 1,
    \\       detail = ?3,
    \\       severity = CASE WHEN ?4 = 'error' THEN 'error' ELSE severity END
    \\ WHERE id = ?1 AND resolved_at IS NULL
;

/// Another failure of an episode already open. False means the row was not
/// there or was already resolved — the store's mirror was stale.
pub fn touchActive(
    database: *db.Db,
    id: i64,
    now_s: i64,
    severity: []const u8,
    detail: []const u8,
) db.Error!bool {
    var stmt = try database.prepare(touch_sql);
    defer stmt.deinit();
    try stmt.bindInt(1, id);
    try stmt.bindInt(2, now_s);
    try stmt.bindText(3, detail);
    try stmt.bindText(4, severity);
    try stmt.exec();
    return database.changes() != 0;
}

const resolve_by_id_sql =
    "UPDATE operational_events SET resolved_at = ?2 WHERE id = ?1 AND resolved_at IS NULL";

/// Closes one episode by row id. False means it was not open, which the store
/// treats as a stale mirror entry rather than an error.
pub fn resolveActive(database: *db.Db, id: i64, now_s: i64) db.Error!bool {
    var stmt = try database.prepare(resolve_by_id_sql);
    defer stmt.deinit();
    try stmt.bindInt(1, id);
    try stmt.bindInt(2, now_s);
    try stmt.exec();
    return database.changes() != 0;
}

const resolve_by_key_sql =
    \\UPDATE operational_events SET resolved_at = ?3
    \\ WHERE code = ?1 AND subject_key = ?2 AND resolved_at IS NULL
;

/// Closes one episode by subject. This is the slow path the store takes only
/// when its mirror cannot answer — the mirror is full and some active row is
/// outside it.
pub fn resolveActiveByKey(
    database: *db.Db,
    now_s: i64,
    code: []const u8,
    subject_key: []const u8,
) db.Error!bool {
    var stmt = try database.prepare(resolve_by_key_sql);
    defer stmt.deinit();
    try stmt.bindText(1, code);
    try stmt.bindText(2, subject_key);
    try stmt.bindInt(3, now_s);
    try stmt.exec();
    return database.changes() != 0;
}

const select_active_id_sql =
    "SELECT id FROM operational_events WHERE code = ?1 AND subject_key = ?2 AND resolved_at IS NULL";

/// The row id of the open episode for a subject, or null when there is none.
///
/// The store's overflow probe: when active rows exist outside its mirror it
/// cannot know whether a subject already has one open, and inserting blind
/// would collide with the partial unique index. It never runs while the mirror
/// covers every active row.
pub fn selectActiveId(database: *db.Db, code: []const u8, subject_key: []const u8) db.Error!?i64 {
    var stmt = try database.prepare(select_active_id_sql);
    defer stmt.deinit();
    try stmt.bindText(1, code);
    try stmt.bindText(2, subject_key);
    if (!try stmt.step()) return null;
    return stmt.columnInt(0);
}

/// The value phase one stamps on the rows it claims, so phase two can find
/// exactly those and no others.
///
/// A timestamp cannot serve: `resolveExcept` may run in the same second as an
/// ordinary `resolve`, and a mark of `now_s` would make phase two revive the
/// episode that `resolve` had just closed. It must also satisfy the table's
/// `resolved_at >= first_seen` check, which rules out a negative sentinel, and
/// it never reaches a reader — the three statements are one transaction, and a
/// rollback takes the mark with it.
const sweep_mark: i64 = std.math.maxInt(i64);

const mark_all_of_code_sql =
    "UPDATE operational_events SET resolved_at = ?2 WHERE code = ?1 AND resolved_at IS NULL";

/// Un-marks the row phase one claimed for `subject_key`, leaving the episode
/// open. The partial unique index allows only one active row per
/// `(code, subject_key)`, so there is at most one to find; `ORDER BY id DESC
/// LIMIT 1` says so to SQLite rather than trusting the plan.
const unmark_kept_sql =
    \\UPDATE operational_events SET resolved_at = NULL
    \\ WHERE id = (SELECT id FROM operational_events
    \\              WHERE code = ?1 AND subject_key = ?2 AND resolved_at = ?3
    \\              ORDER BY id DESC LIMIT 1)
;

/// Turns what is still marked — the episodes no kept key claimed — into a real
/// resolve at the caller's instant.
const resolve_marked_sql =
    "UPDATE operational_events SET resolved_at = ?2 WHERE code = ?1 AND resolved_at = ?3";

const count_active_sql = "SELECT count(*) FROM operational_events WHERE resolved_at IS NULL";

/// Resolves every active episode of `code` whose subject is not in `kept`, and
/// returns how many active rows the **whole table** has left.
///
/// Mark-then-unmark rather than a `NOT IN` list, so the statement text is fixed
/// no matter how many keys arrive and each key is bound on its own. One
/// transaction, so no reader ever sees a marked row, and only the rows this
/// call marked can be touched by the two statements after it: an episode some
/// other write resolved a moment earlier is not among them, whatever second it
/// happened in.
///
/// The count comes from inside that transaction and is returned only after the
/// commit succeeds: the store sets its overflow count from it, and a number
/// from a transaction that then rolled back would be a lie it kept.
pub fn resolveExcept(
    database: *db.Db,
    now_s: i64,
    code: []const u8,
    kept: []const []const u8,
) db.Error!i64 {
    var tx = try db.Tx.begin(database);
    errdefer tx.rollback();

    {
        var stmt = try database.prepare(mark_all_of_code_sql);
        defer stmt.deinit();
        try stmt.bindText(1, code);
        try stmt.bindInt(2, sweep_mark);
        try stmt.exec();
    }

    if (kept.len != 0) {
        var stmt = try database.prepare(unmark_kept_sql);
        defer stmt.deinit();
        for (kept) |key| {
            try stmt.reset();
            try stmt.bindText(1, code);
            try stmt.bindText(2, key);
            try stmt.bindInt(3, sweep_mark);
            try stmt.exec();
        }
    }

    {
        var stmt = try database.prepare(resolve_marked_sql);
        defer stmt.deinit();
        try stmt.bindText(1, code);
        try stmt.bindInt(2, now_s);
        try stmt.bindInt(3, sweep_mark);
        try stmt.exec();
    }

    const total_active = try database.queryInt(count_active_sql);
    try tx.commit();
    return total_active;
}

/// What `purgeResolved` did. `active` and `absent` are different answers to the
/// operator — one says "not yet", the other "already gone" — so the delete alone
/// cannot report the outcome and the row is probed when it deletes nothing.
pub const PurgeOutcome = enum { deleted, active, absent };

const purge_one_sql =
    "DELETE FROM operational_events WHERE id = ?1 AND resolved_at IS NOT NULL";

const row_exists_sql = "SELECT 1 FROM operational_events WHERE id = ?1";

/// The operator's manual delete of one resolved event.
///
/// The `resolved_at IS NOT NULL` clause is the whole guarantee: an open episode
/// is the current state of the box, and deleting it would make the store's
/// mirror point at a row that no longer exists. A delete that removed nothing is
/// probed, in the same transaction, so `active` and `absent` cannot be confused
/// by a row that resolved between the two statements.
pub fn purgeResolved(database: *db.Db, id: i64) db.Error!PurgeOutcome {
    var tx = try db.Tx.begin(database);
    errdefer tx.rollback();

    {
        var stmt = try database.prepare(purge_one_sql);
        defer stmt.deinit();
        try stmt.bindInt(1, id);
        try stmt.exec();
    }

    const outcome: PurgeOutcome = if (database.changes() != 0) .deleted else blk: {
        var stmt = try database.prepare(row_exists_sql);
        defer stmt.deinit();
        try stmt.bindInt(1, id);
        break :blk if (try stmt.step()) .active else .absent;
    };

    try tx.commit();
    return outcome;
}

const purge_all_resolved_sql = "DELETE FROM operational_events WHERE resolved_at IS NOT NULL";

/// Drops every resolved row and returns how many went. Active episodes are
/// never candidates, so the store's mirror describes the same set of rows before
/// and after.
pub fn purgeAllResolved(database: *db.Db) db.Error!i64 {
    var stmt = try database.prepare(purge_all_resolved_sql);
    defer stmt.deinit();
    try stmt.exec();
    return database.changes();
}

const prune_expired_sql =
    "DELETE FROM operational_events WHERE resolved_at IS NOT NULL AND resolved_at < ?1";

/// The cap keeps the newest resolved rows by the same order the API pages them
/// in, so the row that survives the cap is the row a reader would have seen
/// first. Active rows are never candidates: an episode still going is the state
/// of the box, not history.
const prune_over_cap_sql =
    \\DELETE FROM operational_events
    \\ WHERE resolved_at IS NOT NULL AND id NOT IN (
    \\       SELECT id FROM operational_events WHERE resolved_at IS NOT NULL
    \\        ORDER BY resolved_at DESC, id DESC LIMIT ?1)
;

/// Drops resolved rows older than `cutoff_s`, then resolved rows beyond
/// `max_rows`, and returns how many went. One transaction: two passes that
/// half-applied would leave the cap enforced against a set the retention pass
/// had already changed.
pub fn pruneResolved(database: *db.Db, cutoff_s: i64, max_rows: i64) db.Error!i64 {
    var tx = try db.Tx.begin(database);
    errdefer tx.rollback();

    var deleted: i64 = 0;
    {
        var stmt = try database.prepare(prune_expired_sql);
        defer stmt.deinit();
        try stmt.bindInt(1, cutoff_s);
        try stmt.exec();
        deleted += database.changes();
    }
    {
        var stmt = try database.prepare(prune_over_cap_sql);
        defer stmt.deinit();
        try stmt.bindInt(1, max_rows);
        try stmt.exec();
        deleted += database.changes();
    }

    try tx.commit();
    return deleted;
}

// ---------------------------------------------------------------------------
// the store's mirror
// ---------------------------------------------------------------------------

/// One active row, as the store's mirror needs it. By value with fixed buffers
/// rather than by slice: the store loads these into a stack array and the
/// statement's own storage dies at the next `step`.
pub const ActiveRow = struct {
    id: i64,
    code_buf: [code_capacity]u8,
    code_len: u8,
    key_buf: [key_capacity]u8,
    key_len: u16,

    pub fn code(self: *const ActiveRow) []const u8 {
        return self.code_buf[0..self.code_len];
    }

    pub fn subjectKey(self: *const ActiveRow) []const u8 {
        return self.key_buf[0..self.key_len];
    }
};

/// The length bounds are not defensive noise. A row wider than the mirror's
/// buffers cannot be mirrored, and truncating it would give the store a *wrong*
/// identity — a later resolve would match the truncation and close the wrong
/// episode, or miss and silently no-op. Excluding it here leaves it counted as
/// untracked by `countActive`, which is the honest outcome.
const load_active_sql =
    \\SELECT id, code, subject_key FROM operational_events
    \\ WHERE resolved_at IS NULL AND id > ?1
    \\   AND length(code) <= ?3 AND length(subject_key) <= ?4
    \\ ORDER BY id LIMIT ?2
;

/// Fills `out` with active rows whose id is above `after_id`, ascending, and
/// returns how many were written. The store pages through with the last id it
/// received, so nothing here has to hold a cursor open across a call.
pub fn loadActive(database: *db.Db, out: []ActiveRow, after_id: i64) db.Error!usize {
    if (out.len == 0) return 0;

    var stmt = try database.prepare(load_active_sql);
    defer stmt.deinit();
    try stmt.bindInt(1, after_id);
    try stmt.bindInt(2, @intCast(out.len));
    try stmt.bindInt(3, code_capacity);
    try stmt.bindInt(4, key_capacity);

    var written: usize = 0;
    while (written < out.len and try stmt.step()) {
        const code = stmt.columnText(1);
        const key = stmt.columnText(2);
        // The `length()` bounds above are on characters and these are on bytes;
        // a multi-byte code or key could pass the first and not the second.
        if (code.len > code_capacity or key.len > key_capacity) continue;

        const row = &out[written];
        row.* = .{
            .id = stmt.columnInt(0),
            .code_buf = @splat(0),
            .code_len = @intCast(code.len),
            .key_buf = @splat(0),
            .key_len = @intCast(key.len),
        };
        @memcpy(row.code_buf[0..code.len], code);
        @memcpy(row.key_buf[0..key.len], key);
        written += 1;
    }
    return written;
}

/// Every active row, mirrored or not. The store's overflow count is this minus
/// what it managed to mirror.
pub fn countActive(database: *db.Db) db.Error!i64 {
    return database.queryInt(count_active_sql);
}

/// What `/api/health` and `/metrics` report, and the `active` block of the
/// diagnostics page.
pub const Counts = struct {
    warnings: u32 = 0,
    errors: u32 = 0,
};

const active_counts_sql =
    \\SELECT coalesce(sum(severity = 'warning'), 0), coalesce(sum(severity = 'error'), 0)
    \\  FROM operational_events WHERE resolved_at IS NULL
;

pub fn activeCounts(database: *db.Db) db.Error!Counts {
    var stmt = try database.prepare(active_counts_sql);
    defer stmt.deinit();
    // A bare aggregate always produces exactly one row; no row means the
    // statement is not the one this function prepared.
    if (!try stmt.step()) return error.Misuse;
    return .{
        .warnings = try countOf(stmt.columnInt(0)),
        .errors = try countOf(stmt.columnInt(1)),
    };
}

/// `sum` over a boolean cannot go negative; a negative value means the row came
/// from something other than this schema.
fn countOf(value: i64) db.Error!u32 {
    if (value < 0) return error.Mismatch;
    return std.math.cast(u32, value) orelse error.Mismatch;
}

// ---------------------------------------------------------------------------
// the API read layer (`GET /api/diagnostics`)
// ---------------------------------------------------------------------------

/// One event as the API serializes it. `subject_key` deliberately has no field
/// here: it is the store's identity for a subject and may hold an upstream url
/// with a token in it, so it never leaves the process. `subject` is the
/// redacted display identity.
pub const Event = struct {
    id: i64,
    code: []const u8,
    component: []const u8,
    subject: []const u8,
    severity: []const u8,
    first_seen: i64,
    last_seen: i64,
    occurrences: i64,
    resolved_at: ?i64,
    detail: []const u8,
};

pub const State = enum { active, resolved, all };

/// Every field is an independent narrowing; `null` means "do not filter on it".
pub const Filter = struct {
    state: State = .all,
    /// `"warning"` or `"error"`, validated by the handler.
    severity: ?[]const u8 = null,
    /// Matched on the part of `code` before the dot.
    component: ?[]const u8 = null,
    /// The window an episode must overlap. `since` is exclusive at the
    /// resolution end and `until` is exclusive at the start end — see
    /// `where_window`.
    since: ?i64 = null,
    until: ?i64 = null,
    limit: u32 = 100,
    /// Keyset cursor: only rows with a strictly smaller `id`. Rows come back
    /// newest-first, so this is the id of the last row of the previous page.
    before: ?i64 = null,
};

const select_head =
    \\SELECT id, code, subject_label, severity, first_seen, last_seen,
    \\       occurrences, resolved_at, detail
    \\  FROM operational_events
;

/// The escape character of `where_component`. SQLite does not give string
/// literals C escapes, so `'\'` in the SQL text is one backslash.
const like_escape = '\\';

const where_active = " resolved_at IS NULL";
const where_resolved = " resolved_at IS NOT NULL";
const where_severity = " severity = ?";
const where_component = " code LIKE ? ESCAPE '\\'";
const where_before = " id < ?";

/// An episode overlaps `[since, until)` when it started before the window ended
/// and had not already been resolved when the window began. `resolved_at >
/// since` is strict, matching `queries_repo`: an episode resolved exactly at
/// `since` is over by the time the window opens.
const where_since = " (resolved_at IS NULL OR resolved_at > ?)";
const where_until = " first_seen < ?";

const select_tail = " ORDER BY id DESC LIMIT ?";

const where_keyword = " WHERE";
const and_keyword = " AND";

/// Assembles the statement from the fixed fragments above and nothing else.
///
/// **No value ever reaches this buffer.** Every filter contributes a `?` and is
/// bound afterwards, in the order the predicates were appended: an unnumbered
/// parameter takes the next free index, so append order and bind order are the
/// same single contract.
const Sql = struct {
    /// `where_keyword` is longer than `and_keyword` and is used at most once,
    /// so counting six of it bounds every reachable combination.
    const capacity = select_head.len + 6 * where_keyword.len + select_tail.len +
        where_resolved.len + where_severity.len + where_component.len +
        where_since.len + where_until.len + where_before.len;

    buf: [capacity]u8 = undefined,
    len: usize = 0,
    has_where: bool = false,

    fn put(self: *Sql, fragment: []const u8) void {
        @memcpy(self.buf[self.len..][0..fragment.len], fragment);
        self.len += fragment.len;
    }

    fn predicate(self: *Sql, fragment: []const u8) void {
        self.put(if (self.has_where) and_keyword else where_keyword);
        self.has_where = true;
        self.put(fragment);
    }

    fn text(self: *const Sql) []const u8 {
        return self.buf[0..self.len];
    }
};

/// Rows come back newest-first (`id DESC`). Every string is allocated from
/// `arena`, including the list's own storage, so the caller frees the whole
/// result by resetting the arena — there is nothing to unwind on failure.
///
/// An empty window returns nothing rather than everything: `since >= until`
/// describes no time at all, and a filter that answers a question the client
/// did not ask is worse than an empty page.
pub fn selectEvents(database: *db.Db, arena: Allocator, filter: Filter) db.Error!std.ArrayList(Event) {
    var out: std.ArrayList(Event) = .empty;
    if (filter.since) |since| {
        if (filter.until) |until| {
            if (since >= until) return out;
        }
    }

    var sql: Sql = .{};
    sql.put(select_head);
    switch (filter.state) {
        .active => sql.predicate(where_active),
        .resolved => sql.predicate(where_resolved),
        .all => {},
    }
    if (filter.severity != null) sql.predicate(where_severity);
    if (filter.component != null) sql.predicate(where_component);
    if (filter.since != null) sql.predicate(where_since);
    if (filter.until != null) sql.predicate(where_until);
    if (filter.before != null) sql.predicate(where_before);
    sql.put(select_tail);

    var stmt = try database.prepare(sql.text());
    defer stmt.deinit();

    var idx: c_int = 0;
    if (filter.severity) |v| {
        idx += 1;
        try stmt.bindText(idx, v);
    }
    if (filter.component) |v| {
        idx += 1;
        try stmt.bindText(idx, try componentPattern(arena, v));
    }
    if (filter.since) |v| {
        idx += 1;
        try stmt.bindInt(idx, v);
    }
    if (filter.until) |v| {
        idx += 1;
        try stmt.bindInt(idx, v);
    }
    if (filter.before) |v| {
        idx += 1;
        try stmt.bindInt(idx, v);
    }
    idx += 1;
    try stmt.bindInt(idx, @min(filter.limit, max_limit));

    while (try stmt.step()) {
        try out.append(arena, try rowOf(&stmt, arena));
    }
    return out;
}

const select_one_sql = select_head ++ " WHERE id = ?1";

/// Null for an id that never existed and for one retention has since removed —
/// the API cannot tell those apart and does not pretend to.
pub fn selectOne(database: *db.Db, arena: Allocator, id: i64) db.Error!?Event {
    var stmt = try database.prepare(select_one_sql);
    defer stmt.deinit();
    try stmt.bindInt(1, id);
    if (!try stmt.step()) return null;
    return try rowOf(&stmt, arena);
}

fn rowOf(stmt: *db.Stmt, arena: Allocator) db.Error!Event {
    const code = try stmt.columnTextAlloc(arena, 1);
    return .{
        .id = stmt.columnInt(0),
        .code = code,
        .component = componentOf(code),
        .subject = try stmt.columnTextAlloc(arena, 2),
        .severity = try stmt.columnTextAlloc(arena, 3),
        .first_seen = stmt.columnInt(4),
        .last_seen = stmt.columnInt(5),
        .occurrences = stmt.columnInt(6),
        .resolved_at = if (stmt.isNull(7)) null else stmt.columnInt(7),
        .detail = try stmt.columnTextAlloc(arena, 8),
    };
}

/// The part of a wire code before the dot, borrowed from `code` itself. A code
/// with no dot is its own component; `events.Code` cannot produce one, and a
/// row that came from somewhere else still gets an answer rather than a crash.
pub fn componentOf(code: []const u8) []const u8 {
    const dot = std.mem.indexOfScalar(u8, code, '.') orelse return code;
    return code[0..dot];
}

/// `component.%`, with the two `LIKE` metacharacters neutralised so a filter of
/// `query_log` matches `query_log.write` and not `queryxlog.write`. The escape
/// character escapes itself.
fn componentPattern(arena: Allocator, component: []const u8) Allocator.Error![]const u8 {
    var out: std.ArrayList(u8) = try .initCapacity(arena, component.len * 2 + 2);
    for (component) |ch| {
        if (ch == '%' or ch == '_' or ch == like_escape) out.appendAssumeCapacity(like_escape);
        out.appendAssumeCapacity(ch);
    }
    out.appendAssumeCapacity('.');
    out.appendAssumeCapacity('%');
    return out.items;
}

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

const migrations = @import("../migrations.zig");
const testing = std.testing;

fn openConfig() !db.Db {
    var database = try db.Db.open(":memory:", .{ .mode = .memory });
    errdefer database.close();
    try db.applyPragmas(&database, .{});
    _ = try migrations.migrate(&database);
    return database;
}

fn open(database: *db.Db, now_s: i64, code: []const u8, key: []const u8) !i64 {
    return insertActive(database, now_s, code, key, key, severity_warning, "first");
}

test "an episode opens active and closes exactly once" {
    var database = try openConfig();
    defer database.close();

    const id = try open(&database, 100, "blocklist.refresh", "https://a.example");
    try testing.expectEqual(@as(i64, 1), try countActive(&database));

    try testing.expect(try resolveActive(&database, id, 200));
    try testing.expectEqual(@as(i64, 0), try countActive(&database));
    // Already closed: the second call is a no-op, not an error, and it must not
    // move `resolved_at` to the later instant.
    try testing.expect(!try resolveActive(&database, id, 300));
    try testing.expectEqual(
        @as(i64, 200),
        try database.queryInt("SELECT resolved_at FROM operational_events"),
    );
}

test "touchActive counts the failure, replaces the detail and raises severity without lowering it" {
    var database = try openConfig();
    defer database.close();
    const id = try open(&database, 100, "disk.space", "data");

    try testing.expect(try touchActive(&database, id, 150, severity_warning, "second"));
    try testing.expect(try touchActive(&database, id, 200, severity_error, "third"));
    // A warning after an error leaves the episode an error: the row records the
    // worst the subject reached, not the last thing it did.
    try testing.expect(try touchActive(&database, id, 250, severity_warning, "fourth"));

    var stmt = try database.prepare(
        "SELECT occurrences, last_seen, first_seen, severity, detail FROM operational_events WHERE id = ?1",
    );
    defer stmt.deinit();
    try stmt.bindInt(1, id);
    try testing.expect(try stmt.step());
    try testing.expectEqual(@as(i64, 4), stmt.columnInt(0));
    try testing.expectEqual(@as(i64, 250), stmt.columnInt(1));
    try testing.expectEqual(@as(i64, 100), stmt.columnInt(2));
    try testing.expectEqualStrings("error", stmt.columnText(3));
    try testing.expectEqualStrings("fourth", stmt.columnText(4));
}

test "touching or resolving a row that is not open reports it rather than failing" {
    var database = try openConfig();
    defer database.close();
    const id = try open(&database, 100, "disk.space", "data");
    try testing.expect(try resolveActive(&database, id, 200));

    // What a stale mirror entry looks like from here.
    try testing.expect(!try touchActive(&database, id, 300, severity_warning, "late"));
    try testing.expect(!try resolveActive(&database, 9999, 300));
    try testing.expect(!try resolveActiveByKey(&database, 300, "disk.space", "data"));
}

test "resolveActiveByKey closes the open episode and leaves the earlier ones alone" {
    var database = try openConfig();
    defer database.close();

    const first = try open(&database, 100, "upstream.exchange", "https://a.example");
    try testing.expect(try resolveActive(&database, first, 150));
    const second = try open(&database, 200, "upstream.exchange", "https://a.example");

    try testing.expect(try resolveActiveByKey(&database, 300, "upstream.exchange", "https://a.example"));
    try testing.expectEqual(@as(i64, 0), try countActive(&database));
    var stmt = try database.prepare("SELECT resolved_at FROM operational_events ORDER BY id");
    defer stmt.deinit();
    try testing.expect(try stmt.step());
    try testing.expectEqual(@as(i64, 150), stmt.columnInt(0));
    try testing.expect(try stmt.step());
    try testing.expectEqual(@as(i64, 300), stmt.columnInt(0));
    // By key, so the row it closed is the one the key had open.
    try testing.expectEqual(
        second,
        try database.queryInt("SELECT id FROM operational_events WHERE resolved_at = 300"),
    );
}

test "insertResolved writes a row that is over the moment it exists" {
    var database = try openConfig();
    defer database.close();
    try insertResolved(&database, 100, "query_log.recreated", "one-shot", "corrupt", severity_warning, "aside");

    try testing.expectEqual(@as(i64, 0), try countActive(&database));
    var stmt = try database.prepare("SELECT first_seen, last_seen, resolved_at, occurrences FROM operational_events");
    defer stmt.deinit();
    try testing.expect(try stmt.step());
    for (0..3) |col| try testing.expectEqual(@as(i64, 100), stmt.columnInt(@intCast(col)));
    try testing.expectEqual(@as(i64, 1), stmt.columnInt(3));

    // Nothing stops a second one: a one-shot event never competes for the
    // partial unique index, because it is never active.
    try insertResolved(&database, 200, "query_log.recreated", "one-shot", "corrupt", severity_warning, "aside");
    try testing.expectEqual(@as(i64, 2), try database.queryInt("SELECT count(*) FROM operational_events"));
}

test "selectActiveId finds the open episode of a subject and nothing else" {
    var database = try openConfig();
    defer database.close();

    const id = try open(&database, 100, "disk.probe", "statvfs");
    try testing.expectEqual(@as(?i64, id), try selectActiveId(&database, "disk.probe", "statvfs"));
    // Another code with the same subject is a different episode.
    try testing.expectEqual(@as(?i64, null), try selectActiveId(&database, "disk.space", "statvfs"));

    try testing.expect(try resolveActive(&database, id, 200));
    // A closed episode is not an open one, which is what makes the store's
    // probe answer "insert" rather than "touch".
    try testing.expectEqual(@as(?i64, null), try selectActiveId(&database, "disk.probe", "statvfs"));
}

test "resolveExcept closes what is not kept, keeps what is, and reports the whole table's active rows" {
    var database = try openConfig();
    defer database.close();

    _ = try open(&database, 100, "listener.start", "doh");
    _ = try open(&database, 100, "listener.start", "dot");
    _ = try open(&database, 100, "configuration.load", "dns.port");

    const remaining = try resolveExcept(&database, 500, "listener.start", &.{"dot"});
    // `dot` stayed open and `configuration.load` was never this code's business.
    try testing.expectEqual(@as(i64, 2), remaining);
    try testing.expectEqual(@as(i64, 2), try countActive(&database));

    var stmt = try database.prepare(
        "SELECT resolved_at FROM operational_events WHERE code = 'listener.start' AND subject_key = ?1",
    );
    defer stmt.deinit();
    try stmt.bindText(1, "doh");
    try testing.expect(try stmt.step());
    try testing.expectEqual(@as(i64, 500), stmt.columnInt(0));
}

test "a kept key revives only the row this pass resolved, never an older episode" {
    var database = try openConfig();
    defer database.close();

    // An episode of the same subject that was already over — and over at the
    // very instant this pass stamps, which is the case the `ORDER BY id DESC
    // LIMIT 1` bound exists for.
    const stale = try open(&database, 100, "listener.start", "doh");
    try testing.expect(try resolveActive(&database, stale, 500));
    const live = try open(&database, 200, "listener.start", "doh");

    try testing.expectEqual(@as(i64, 1), try resolveExcept(&database, 500, "listener.start", &.{"doh"}));
    try testing.expectEqual(@as(i64, 1), try countActive(&database));
    try testing.expectEqual(
        live,
        try database.queryInt("SELECT id FROM operational_events WHERE resolved_at IS NULL"),
    );
    // The old episode is still closed, at the timestamp it always had.
    try testing.expectEqual(
        @as(i64, 500),
        try database.queryInt("SELECT resolved_at FROM operational_events WHERE id = 1"),
    );
}

test "resolveExcept with no kept keys closes every episode of the code" {
    var database = try openConfig();
    defer database.close();
    _ = try open(&database, 100, "configuration.load", "dns.port");
    _ = try open(&database, 100, "configuration.load", "upstreams[0]");
    _ = try open(&database, 100, "disk.space", "data");

    try testing.expectEqual(@as(i64, 1), try resolveExcept(&database, 500, "configuration.load", &.{}));
    try testing.expectEqual(@as(i64, 1), try countActive(&database));
}

test "a failed resolveExcept leaves every row as it was" {
    var database = try openConfig();
    defer database.close();
    _ = try open(&database, 100, "listener.start", "doh");
    try database.exec(
        \\CREATE TRIGGER refuse_revive AFTER UPDATE ON operational_events
        \\WHEN new.resolved_at IS NULL
        \\BEGIN SELECT RAISE(ABORT, 'refused'); END;
    );

    // Phase one resolved the row and phase two could not put it back; one
    // transaction is what stops the pass from closing an episode it meant to
    // keep.
    try testing.expectError(error.Constraint, resolveExcept(&database, 500, "listener.start", &.{"doh"}));
    try testing.expectEqual(@as(i64, 1), try countActive(&database));
}

test "activeCounts splits open episodes by severity and ignores resolved ones" {
    var database = try openConfig();
    defer database.close();
    try testing.expectEqual(Counts{}, try activeCounts(&database));

    _ = try open(&database, 100, "disk.space", "data");
    _ = try insertActive(&database, 100, "listener.start", "doh", "doh", severity_error, "bind");
    const gone = try open(&database, 100, "upstream.exchange", "https://a.example");
    try testing.expect(try resolveActive(&database, gone, 200));

    try testing.expectEqual(Counts{ .warnings = 1, .errors = 1 }, try activeCounts(&database));
}

test "loadActive pages by id and skips a row too wide for the mirror" {
    var database = try openConfig();
    defer database.close();

    for (0..5) |i| {
        var key_buf: [8]u8 = undefined;
        const key = try std.fmt.bufPrint(&key_buf, "k{d}", .{i});
        _ = try open(&database, 100, "disk.probe", key);
    }
    // Wider than `key_capacity`: mirroring it under a truncated identity would
    // make a later resolve close the wrong episode.
    const wide = "w" ** (key_capacity + 1);
    _ = try open(&database, 100, "disk.probe", wide);

    var out: [2]ActiveRow = undefined;
    var seen: usize = 0;
    var after: i64 = 0;
    while (true) {
        const got = try loadActive(&database, &out, after);
        if (got == 0) break;
        for (out[0..got]) |row| {
            try testing.expect(row.subjectKey().len <= key_capacity);
            try testing.expectEqualStrings("disk.probe", row.code());
            seen += 1;
        }
        after = out[got - 1].id;
    }
    try testing.expectEqual(@as(usize, 5), seen);
    // The row that could not be mirrored is still active, so the store's
    // overflow count picks it up as the difference.
    try testing.expectEqual(@as(i64, 6), try countActive(&database));
}

test "prune drops resolved rows past the window and past the cap, never an active one" {
    var database = try openConfig();
    defer database.close();

    var live_keys: [3][8]u8 = undefined;
    for (&live_keys, 0..) |*buf, i| {
        const key = try std.fmt.bufPrint(buf, "live{d}", .{i});
        _ = try open(&database, 100, "disk.probe", key);
    }
    for (0..6) |i| {
        var key_buf: [8]u8 = undefined;
        const key = try std.fmt.bufPrint(&key_buf, "old{d}", .{i});
        const id = try open(&database, 100, "query_log.maintenance", key);
        try testing.expect(try resolveActive(&database, id, @intCast(1000 + i)));
    }

    // Nothing is old enough and nothing is over the cap.
    try testing.expectEqual(@as(i64, 0), try pruneResolved(&database, 1000, 10));
    try testing.expectEqual(@as(i64, 9), try database.queryInt("SELECT count(*) FROM operational_events"));

    // Two fall out of the retention window; the cap then takes the oldest of
    // what is left, and the three active rows are never candidates for either.
    try testing.expectEqual(@as(i64, 3), try pruneResolved(&database, 1002, 3));
    try testing.expectEqual(@as(i64, 3), try countActive(&database));
    try testing.expectEqual(
        @as(i64, 1005),
        try database.queryInt("SELECT max(resolved_at) FROM operational_events"),
    );
    try testing.expectEqual(
        @as(i64, 1003),
        try database.queryInt("SELECT min(resolved_at) FROM operational_events WHERE resolved_at IS NOT NULL"),
    );
}

test "purging one event takes the resolved row, refuses the open one and reports a missing id" {
    var database = try openConfig();
    defer database.close();

    const open_episode = try open(&database, 100, "disk.space", "data");
    const closed = try open(&database, 100, "query_log.maintenance", "vacuum");
    try testing.expect(try resolveActive(&database, closed, 200));

    // Still going: the operator is told why, and the row stays.
    try testing.expectEqual(PurgeOutcome.active, try purgeResolved(&database, open_episode));
    try testing.expectEqual(@as(i64, 2), try database.queryInt("SELECT count(*) FROM operational_events"));

    try testing.expectEqual(PurgeOutcome.deleted, try purgeResolved(&database, closed));
    try testing.expectEqual(@as(i64, 1), try database.queryInt("SELECT count(*) FROM operational_events"));
    try testing.expectEqual(@as(i64, 1), try countActive(&database));

    // The same id twice: gone is not the same answer as still open.
    try testing.expectEqual(PurgeOutcome.absent, try purgeResolved(&database, closed));
    try testing.expectEqual(PurgeOutcome.absent, try purgeResolved(&database, 9999));
}

test "purging all resolved events counts what it took and leaves every active one" {
    var database = try openConfig();
    defer database.close();

    _ = try open(&database, 100, "disk.space", "data");
    _ = try open(&database, 100, "listener.start", "doh");
    for (0..3) |i| {
        var key_buf: [8]u8 = undefined;
        const key = try std.fmt.bufPrint(&key_buf, "old{d}", .{i});
        const id = try open(&database, 100, "query_log.maintenance", key);
        try testing.expect(try resolveActive(&database, id, @intCast(200 + i)));
    }
    try insertResolved(&database, 300, "query_log.recreated", "one-shot", "corrupt", severity_warning, "aside");

    try testing.expectEqual(@as(i64, 4), try purgeAllResolved(&database));
    try testing.expectEqual(@as(i64, 2), try database.queryInt("SELECT count(*) FROM operational_events"));
    try testing.expectEqual(@as(i64, 2), try countActive(&database));

    // Nothing left to take is zero, not an error.
    try testing.expectEqual(@as(i64, 0), try purgeAllResolved(&database));
}

fn seedPage(database: *db.Db) !void {
    _ = try insertActive(database, 100, "blocklist.refresh", "https://a.example", "A", severity_warning, "d1");
    _ = try insertActive(database, 110, "blocklist.storage", "compile", "compile", severity_error, "d2");
    const closed = try insertActive(database, 120, "query_log.write", "batch", "batch", severity_error, "d3");
    try testing.expect(try resolveActive(database, closed, 400));
    _ = try insertActive(database, 130, "query_log.maintenance", "vacuum", "vacuum", severity_warning, "d4");
}

test "selectEvents returns newest first with the component derived from the code" {
    var database = try openConfig();
    defer database.close();
    try seedPage(&database);

    var arena: std.heap.ArenaAllocator = .init(testing.allocator);
    defer arena.deinit();
    const rows = try selectEvents(&database, arena.allocator(), .{});

    try testing.expectEqual(@as(usize, 4), rows.items.len);
    try testing.expectEqualStrings("query_log.maintenance", rows.items[0].code);
    try testing.expectEqualStrings("query_log", rows.items[0].component);
    try testing.expectEqualStrings("blocklist", rows.items[3].component);
    try testing.expectEqualStrings("A", rows.items[3].subject);
    try testing.expectEqual(@as(?i64, null), rows.items[0].resolved_at);
    for (rows.items[0 .. rows.items.len - 1], rows.items[1..]) |newer, older| {
        try testing.expect(newer.id > older.id);
    }
}

test "each filter narrows the page and they compose" {
    var database = try openConfig();
    defer database.close();
    try seedPage(&database);

    var arena: std.heap.ArenaAllocator = .init(testing.allocator);
    defer arena.deinit();
    const gpa = arena.allocator();

    try testing.expectEqual(@as(usize, 3), (try selectEvents(&database, gpa, .{ .state = .active })).items.len);
    try testing.expectEqual(@as(usize, 1), (try selectEvents(&database, gpa, .{ .state = .resolved })).items.len);
    try testing.expectEqual(
        @as(usize, 2),
        (try selectEvents(&database, gpa, .{ .severity = severity_error })).items.len,
    );
    try testing.expectEqual(
        @as(usize, 2),
        (try selectEvents(&database, gpa, .{ .component = "query_log" })).items.len,
    );
    try testing.expectEqual(
        @as(usize, 1),
        (try selectEvents(&database, gpa, .{ .component = "query_log", .state = .active })).items.len,
    );
    // The component match is anchored at the dot: `query` is not `query_log`.
    try testing.expectEqual(@as(usize, 0), (try selectEvents(&database, gpa, .{ .component = "query" })).items.len);
    // And `_` is a literal, not the `LIKE` wildcard it would otherwise be.
    try testing.expectEqual(@as(usize, 0), (try selectEvents(&database, gpa, .{ .component = "query%log" })).items.len);
    try testing.expectEqual(@as(usize, 0), (try selectEvents(&database, gpa, .{ .component = "queryxlog" })).items.len);
}

test "the window selects episodes that overlap it and an empty window selects nothing" {
    var database = try openConfig();
    defer database.close();

    // Over before the window opens.
    const before = try open(&database, 100, "disk.probe", "statvfs");
    try testing.expect(try resolveActive(&database, before, 200));
    // Resolved exactly at `since`: over by the time the window begins.
    const boundary = try open(&database, 100, "disk.probe", "data_dir");
    try testing.expect(try resolveActive(&database, boundary, 300));
    // Straddles the window.
    const straddles = try open(&database, 100, "disk.probe", "log_dir");
    try testing.expect(try resolveActive(&database, straddles, 350));
    // Opens inside it and is still going.
    _ = try open(&database, 320, "disk.space", "data");
    // Opens after it closes.
    _ = try open(&database, 400, "clients.storage", "prune");

    var arena: std.heap.ArenaAllocator = .init(testing.allocator);
    defer arena.deinit();
    const gpa = arena.allocator();

    const window = try selectEvents(&database, gpa, .{ .since = 300, .until = 400 });
    try testing.expectEqual(@as(usize, 2), window.items.len);
    try testing.expectEqualStrings("disk.space", window.items[0].code);
    try testing.expectEqualStrings("log_dir", window.items[1].subject);

    // `since >= until` describes no time at all.
    try testing.expectEqual(@as(usize, 0), (try selectEvents(&database, gpa, .{ .since = 300, .until = 300 })).items.len);
    try testing.expectEqual(@as(usize, 0), (try selectEvents(&database, gpa, .{ .since = 400, .until = 300 })).items.len);
    // Each end on its own still bounds.
    try testing.expectEqual(@as(usize, 3), (try selectEvents(&database, gpa, .{ .until = 300 })).items.len);
    try testing.expectEqual(@as(usize, 3), (try selectEvents(&database, gpa, .{ .since = 300 })).items.len);
}

test "the keyset cursor walks the whole table without repeating a row" {
    var database = try openConfig();
    defer database.close();
    for (0..5) |i| {
        var key_buf: [8]u8 = undefined;
        const key = try std.fmt.bufPrint(&key_buf, "k{d}", .{i});
        _ = try open(&database, 100, "disk.probe", key);
    }

    var arena: std.heap.ArenaAllocator = .init(testing.allocator);
    defer arena.deinit();
    const gpa = arena.allocator();

    var seen: usize = 0;
    var before: ?i64 = null;
    while (true) {
        const page = try selectEvents(&database, gpa, .{ .limit = 2, .before = before });
        if (page.items.len == 0) break;
        for (page.items) |row| {
            if (before) |cursor| try testing.expect(row.id < cursor);
            seen += 1;
        }
        before = page.items[page.items.len - 1].id;
    }
    try testing.expectEqual(@as(usize, 5), seen);
}

test "the limit is capped at the repository's maximum" {
    var database = try openConfig();
    defer database.close();
    try seedPage(&database);

    var arena: std.heap.ArenaAllocator = .init(testing.allocator);
    defer arena.deinit();
    // A caller that forgets the cap cannot ask this connection for more than
    // `max_limit` rows; with four seeded it can only be observed as "all four".
    const rows = try selectEvents(&database, arena.allocator(), .{ .limit = std.math.maxInt(u32) });
    try testing.expectEqual(@as(usize, 4), rows.items.len);
}

test "selectOne answers by id and reports an unknown one as null" {
    var database = try openConfig();
    defer database.close();
    const id = try insertActive(&database, 100, "certificate.reload", "doh", "doh", severity_warning, "stat failed");

    var arena: std.heap.ArenaAllocator = .init(testing.allocator);
    defer arena.deinit();

    const row = (try selectOne(&database, arena.allocator(), id)).?;
    try testing.expectEqualStrings("certificate.reload", row.code);
    try testing.expectEqualStrings("certificate", row.component);
    try testing.expectEqualStrings("stat failed", row.detail);
    try testing.expectEqual(@as(i64, 1), row.occurrences);
    try testing.expect((try selectOne(&database, arena.allocator(), id + 1)) == null);
}

test "componentOf takes the part before the first dot" {
    try testing.expectEqualStrings("blocklist", componentOf("blocklist.refresh"));
    try testing.expectEqualStrings("query_log", componentOf("query_log.write"));
    // Not something `events.Code` can produce; it still gets an answer.
    try testing.expectEqualStrings("bare", componentOf("bare"));
    try testing.expectEqualStrings("", componentOf(""));
}