nxdns/src/storage/repositories/upstream_history_repo.zig

//! `upstream_minute` and its `upstream_targets` dimension table in
//! `querylog.db` (milestone-26 rulings 2, 4, 5).
//!
//! One row per upstream per wall-clock UTC minute that had at least one
//! attempt. Rows are additive facts: a flush adds to whatever is already there,
//! so a restart inside a minute continues that minute's row rather than
//! replacing it, and nothing here can lower a stored count.
//!
//! Identity is the url, not the `config.db` upstream id: ids cannot be foreign
//! keys across database files and may be deleted or reused. Editing an
//! upstream's url deliberately starts a new history.
//!
//! Nothing here retries. The accumulator owns what a failed flush means
//! (`upstream/history.zig`).

const std = @import("std");

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

/// How far back `Retention` keeps minute rows. A fixed window, not a knob
/// (m26 anti-requirements), and deliberately wider than the widest dashboard
/// period: `stats.window` derives `since = until - width * count` with
/// `until > now`, so a 30-day window never asks for anything older than
/// `now - 30d`. The extra day is slack for a retention pass that runs late.
///
/// `logging.retention_days` does not apply here. It bounds the query log, whose
/// rows are per-query; these are per-minute aggregates whose whole purpose is
/// to outlive them.
pub const retention_window_s: i64 = 31 * 86_400;

/// One minute of one upstream's outcomes, as the accumulator hands it over.
/// Every string is borrowed for the duration of the call: `Stmt.bindText` binds
/// with `SQLITE_TRANSIENT`, so SQLite copies before `flush` returns.
pub const FlushRow = struct {
    url: []const u8,
    minute_ts: i64,
    successes: u32,
    failures: u32,
    last_failure_ts: ?i64,
    /// Empty when the minute held no failure.
    last_error: []const u8,
};

const insert_target_sql = "INSERT OR IGNORE INTO upstream_targets (url) VALUES (?1)";

const select_target_sql = "SELECT id FROM upstream_targets WHERE url = ?1";

/// Additive, and the timestamp columns are max-wins, which is what makes a
/// flush safe to repeat against a row another process already wrote.
///
/// `max()` over a NULL is NULL in SQLite, so the coalesce is what keeps an
/// existing `last_failure_ts` when the incoming row carries none. The `CASE`
/// moves `last_error` with the timestamp it belongs to: a success-only upsert
/// leaves the stored failure and its name exactly as they were.
const upsert_minute_sql =
    \\INSERT INTO upstream_minute (upstream_id, minute_ts, successes, failures, last_failure_ts, last_error)
    \\VALUES (?1, ?2, ?3, ?4, ?5, ?6)
    \\ON CONFLICT(upstream_id, minute_ts) DO UPDATE SET
    \\    successes = successes + excluded.successes,
    \\    failures = failures + excluded.failures,
    \\    last_failure_ts = coalesce(max(last_failure_ts, excluded.last_failure_ts), last_failure_ts, excluded.last_failure_ts),
    \\    last_error = CASE
    \\        WHEN excluded.last_failure_ts IS NOT NULL
    \\         AND (last_failure_ts IS NULL OR excluded.last_failure_ts >= last_failure_ts)
    \\        THEN excluded.last_error
    \\        ELSE last_error
    \\    END
;

/// One transaction for the whole batch: either every minute of the pass lands
/// or none of it does, so a failed flush leaves nothing half-written for the
/// caller's merge-back to double-count.
pub fn flush(database: *db.Db, rows: []const FlushRow) db.Error!void {
    if (rows.len == 0) return;

    var insert_target = try database.prepare(insert_target_sql);
    defer insert_target.deinit();
    var select_target = try database.prepare(select_target_sql);
    defer select_target.deinit();
    var upsert = try database.prepare(upsert_minute_sql);
    defer upsert.deinit();

    var tx = try db.Tx.begin(database);
    errdefer tx.rollback();

    for (rows) |row| {
        const upstream_id = try internTarget(&insert_target, &select_target, row.url);

        try upsert.reset();
        try upsert.bindInt(1, upstream_id);
        try upsert.bindInt(2, row.minute_ts);
        try upsert.bindInt(3, row.successes);
        try upsert.bindInt(4, row.failures);
        if (row.last_failure_ts) |at| try upsert.bindInt(5, at) else try upsert.bindNull(5);
        try upsert.bindText(6, row.last_error);
        try upsert.exec();
    }

    try tx.commit();
}

fn internTarget(insert: *db.Stmt, select: *db.Stmt, url: []const u8) db.Error!i64 {
    try insert.reset();
    try insert.bindText(1, url);
    try insert.exec();

    try select.reset();
    try select.bindText(1, url);
    // The insert above either created the row or found it already there, so a
    // miss means the table changed under this connection.
    if (!try select.step()) return error.NotFound;
    const id = select.columnInt(0);
    // A statement stopped on a row keeps its cursor open until it is reset;
    // the transaction must not carry that to the next row.
    try select.reset();
    return id;
}

/// What `GET /api/upstream/health` reports for one upstream over one window.
pub const WindowStats = struct {
    attempts: u64,
    successes: u64,
    failures: u64,
    last_failure_ts: ?i64,
    /// The error name of the row holding the newest `last_failure_ts` in the
    /// window; empty when the window holds no failure.
    ///
    /// By value rather than by slice: the caller loops over upstreams and
    /// reuses one `WindowStats`, so a borrowed slice would dangle into the
    /// storage the next iteration overwrites.
    last_failure_error_buf: [health.error_name_capacity]u8,
    last_failure_error_len: u8,

    pub fn lastFailureError(self: *const WindowStats) []const u8 {
        return self.last_failure_error_buf[0..self.last_failure_error_len];
    }
};

/// **One statement, deliberately.** Two statements would not share a SQLite
/// snapshot: the flush connection can commit between them, and the read would
/// then pair a `max(last_failure_ts)` taken from one state with an error text
/// taken from another.
///
/// The error lookup is a scalar subquery for the same reason it is not a bare
/// column: `SELECT max(last_failure_ts), last_error` lets SQLite return the
/// `last_error` of an arbitrary row of the group. `ORDER BY ... DESC, minute_ts
/// DESC` makes the choice deterministic when two minutes share a timestamp.
const window_stats_sql =
    \\SELECT coalesce(sum(m.successes), 0), coalesce(sum(m.failures), 0), max(m.last_failure_ts),
    \\       (SELECT e.last_error FROM upstream_minute e
    \\        WHERE e.upstream_id = m.upstream_id AND e.minute_ts >= ?2 AND e.minute_ts < ?3
    \\          AND e.last_failure_ts IS NOT NULL
    \\        ORDER BY e.last_failure_ts DESC, e.minute_ts DESC LIMIT 1)
    \\  FROM upstream_minute m JOIN upstream_targets t ON t.id = m.upstream_id
    \\ WHERE t.url = ?1 AND m.minute_ts >= ?2 AND m.minute_ts < ?3
;

/// Aggregates `[since, until)` by `minute_ts`. An unknown url or an empty
/// window is zeros, a null timestamp and the empty error — not an error.
pub fn windowStats(database: *db.Db, url: []const u8, since: i64, until: i64) db.Error!WindowStats {
    var stmt = try database.prepare(window_stats_sql);
    defer stmt.deinit();
    try stmt.bindText(1, url);
    try stmt.bindInt(2, since);
    try stmt.bindInt(3, until);

    // 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;

    const successes = try countOf(stmt.columnInt(0));
    const failures = try countOf(stmt.columnInt(1));

    var out: WindowStats = .{
        .attempts = successes + failures,
        .successes = successes,
        .failures = failures,
        .last_failure_ts = if (stmt.isNull(2)) null else stmt.columnInt(2),
        .last_failure_error_buf = @splat(0),
        .last_failure_error_len = 0,
    };
    const name = stmt.columnText(3);
    const copied = @min(name.len, out.last_failure_error_buf.len);
    @memcpy(out.last_failure_error_buf[0..copied], name[0..copied]);
    out.last_failure_error_len = @intCast(copied);
    return out;
}

/// `sum` over `CHECK (… >= 0)` columns cannot go negative; a negative value
/// means the row came from something other than this schema.
fn countOf(value: i64) db.Error!u64 {
    if (value < 0) return error.Mismatch;
    return @intCast(value);
}

/// Deletes every `upstream_minute` row strictly older than `cutoff_ts`, then
/// the targets no surviving row references, and returns how many **minute**
/// rows went.
///
/// One transaction: a target dropped without its rows, or rows dropped while
/// the target delete failed, would leave the foreign key pointing at nothing.
/// The count is minute rows only, so the metric an operator watches counts
/// aggregates rather than dimension-table housekeeping.
pub fn pruneOlderThan(database: *db.Db, cutoff_ts: i64) db.Error!i64 {
    var tx = try db.Tx.begin(database);
    errdefer tx.rollback();

    var minutes = try database.prepare("DELETE FROM upstream_minute WHERE minute_ts < ?1");
    defer minutes.deinit();
    try minutes.bindInt(1, cutoff_ts);
    try minutes.exec();
    const deleted = database.changes();

    try database.exec(
        \\DELETE FROM upstream_targets
        \\ WHERE id NOT IN (SELECT upstream_id FROM upstream_minute);
    );

    try tx.commit();
    return deleted;
}

pub fn countMinutes(database: *db.Db) db.Error!i64 {
    return database.queryInt("SELECT count(*) FROM upstream_minute");
}

pub fn countTargets(database: *db.Db) db.Error!i64 {
    return database.queryInt("SELECT count(*) FROM upstream_targets");
}

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

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

const testing = std.testing;

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

/// `columnText` is borrowed until the statement is finalized, so the stored
/// error name is copied out rather than returned as a slice.
const StoredMinute = struct {
    successes: u32,
    failures: u32,
    last_failure_ts: ?i64,
    error_buf: [health.error_name_capacity]u8,
    error_len: u8,

    fn lastError(self: *const StoredMinute) []const u8 {
        return self.error_buf[0..self.error_len];
    }
};

fn readMinute(database: *db.Db, url: []const u8, minute_ts: i64) !StoredMinute {
    var stmt = try database.prepare(
        \\SELECT m.successes, m.failures, m.last_failure_ts, m.last_error
        \\  FROM upstream_minute m JOIN upstream_targets t ON t.id = m.upstream_id
        \\ WHERE t.url = ?1 AND m.minute_ts = ?2
    );
    defer stmt.deinit();
    try stmt.bindText(1, url);
    try stmt.bindInt(2, minute_ts);
    try testing.expect(try stmt.step());
    var out: StoredMinute = .{
        .successes = @intCast(stmt.columnInt(0)),
        .failures = @intCast(stmt.columnInt(1)),
        .last_failure_ts = if (stmt.isNull(2)) null else stmt.columnInt(2),
        .error_buf = @splat(0),
        .error_len = 0,
    };
    const name = stmt.columnText(3);
    @memcpy(out.error_buf[0..name.len], name);
    out.error_len = @intCast(name.len);
    return out;
}

test "the upsert adds to the row already there rather than replacing it" {
    var database = try openLog();
    defer database.close();

    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 3, .failures = 1, .last_failure_ts = 70, .last_error = "Timeout" },
    });
    // The shape a restart inside one minute takes: a second process writes the
    // same (url, minute) and the counts continue.
    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 2, .failures = 4, .last_failure_ts = 90, .last_error = "ConnectFailed" },
    });

    const row = try readMinute(&database, "https://a.example", 60);
    try testing.expectEqual(@as(u32, 5), row.successes);
    try testing.expectEqual(@as(u32, 5), row.failures);
    try testing.expectEqual(@as(?i64, 90), row.last_failure_ts);
    try testing.expectEqualStrings("ConnectFailed", row.lastError());
    // One row and one target, not two of either.
    try testing.expectEqual(@as(i64, 1), try countMinutes(&database));
    try testing.expectEqual(@as(i64, 1), try countTargets(&database));
}

test "a success-only upsert keeps the failure timestamp and its error" {
    var database = try openLog();
    defer database.close();

    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 70, .last_error = "Timeout" },
    });
    // `max()` over a NULL is NULL in SQLite, so without the coalesce this
    // upsert would erase the timestamp it knows nothing about.
    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 5, .failures = 0, .last_failure_ts = null, .last_error = "" },
    });

    const row = try readMinute(&database, "https://a.example", 60);
    try testing.expectEqual(@as(u32, 5), row.successes);
    try testing.expectEqual(@as(u32, 1), row.failures);
    try testing.expectEqual(@as(?i64, 70), row.last_failure_ts);
    try testing.expectEqualStrings("Timeout", row.lastError());
}

test "an older failure does not overwrite the newer error already stored" {
    var database = try openLog();
    defer database.close();

    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 90, .last_error = "Timeout" },
    });
    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 70, .last_error = "ConnectFailed" },
    });

    const row = try readMinute(&database, "https://a.example", 60);
    try testing.expectEqual(@as(?i64, 90), row.last_failure_ts);
    try testing.expectEqualStrings("Timeout", row.lastError());
}

test "flush interns each url once and writes every minute of the batch" {
    var database = try openLog();
    defer database.close();

    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
        .{ .url = "https://a.example", .minute_ts = 120, .successes = 2, .failures = 0, .last_failure_ts = null, .last_error = "" },
        .{ .url = "https://b.example", .minute_ts = 60, .successes = 3, .failures = 0, .last_failure_ts = null, .last_error = "" },
    });

    try testing.expectEqual(@as(i64, 3), try countMinutes(&database));
    try testing.expectEqual(@as(i64, 2), try countTargets(&database));

    // An empty batch opens no transaction: one is already open here, so a
    // `BEGIN IMMEDIATE` would fail.
    var tx = try db.Tx.begin(&database);
    try flush(&database, &.{});
    tx.rollback();
}

test "windowStats sums only the window and pairs the newest failure with its own error" {
    var database = try openLog();
    defer database.close();

    try flush(&database, &.{
        // Before the window.
        .{ .url = "https://a.example", .minute_ts = 0, .successes = 9, .failures = 9, .last_failure_ts = 30, .last_error = "Outside" },
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 2, .failures = 1, .last_failure_ts = 100, .last_error = "ConnectFailed" },
        .{ .url = "https://a.example", .minute_ts = 120, .successes = 4, .failures = 2, .last_failure_ts = 170, .last_error = "Timeout" },
        // A later minute with no failure at all: the error must still come from
        // the minute holding the newest `last_failure_ts`, not from this one.
        .{ .url = "https://a.example", .minute_ts = 180, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
        // At the exclusive end of the window.
        .{ .url = "https://a.example", .minute_ts = 240, .successes = 7, .failures = 7, .last_failure_ts = 250, .last_error = "After" },
        // A different upstream in the same minutes.
        .{ .url = "https://b.example", .minute_ts = 120, .successes = 5, .failures = 5, .last_failure_ts = 175, .last_error = "Other" },
    });

    const stats = try windowStats(&database, "https://a.example", 60, 240);
    try testing.expectEqual(@as(u64, 7), stats.successes);
    try testing.expectEqual(@as(u64, 3), stats.failures);
    try testing.expectEqual(@as(u64, 10), stats.attempts);
    try testing.expectEqual(@as(?i64, 170), stats.last_failure_ts);
    try testing.expectEqualStrings("Timeout", stats.lastFailureError());
}

test "two minutes sharing the newest failure timestamp resolve to the later minute" {
    // The tiebreak the subquery's ORDER BY owns. `max(last_failure_ts)` alone
    // cannot choose between these two rows, so without a deterministic second
    // key the answer is whichever row SQLite happened to visit.
    var database = try openLog();
    defer database.close();
    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 100, .last_error = "Earlier" },
        .{ .url = "https://a.example", .minute_ts = 120, .successes = 0, .failures = 1, .last_failure_ts = 100, .last_error = "Later" },
    });

    const stats = try windowStats(&database, "https://a.example", 0, 1000);
    try testing.expectEqual(@as(?i64, 100), stats.last_failure_ts);
    try testing.expectEqualStrings("Later", stats.lastFailureError());
}

test "windowStats over an unknown url or an empty window is zeros and no error" {
    var database = try openLog();
    defer database.close();
    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 1, .failures = 1, .last_failure_ts = 70, .last_error = "Timeout" },
    });

    for ([_][3]i64{ .{ 0, 60, 0 }, .{ 120, 180, 0 }, .{ 60, 60, 0 } }) |window| {
        const stats = try windowStats(&database, "https://a.example", window[0], window[1]);
        try testing.expectEqual(@as(u64, 0), stats.attempts);
        try testing.expectEqual(@as(u64, 0), stats.successes);
        try testing.expectEqual(@as(u64, 0), stats.failures);
        try testing.expectEqual(@as(?i64, null), stats.last_failure_ts);
        try testing.expectEqualStrings("", stats.lastFailureError());
    }

    const unknown = try windowStats(&database, "https://never.example", 0, 1000);
    try testing.expectEqual(@as(u64, 0), unknown.attempts);
    try testing.expectEqual(@as(?i64, null), unknown.last_failure_ts);
    try testing.expectEqualStrings("", unknown.lastFailureError());
}

test "a window whose only failures are outside it reports no failure at all" {
    var database = try openLog();
    defer database.close();
    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 0, .successes = 0, .failures = 1, .last_failure_ts = 30, .last_error = "Timeout" },
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 4, .failures = 0, .last_failure_ts = null, .last_error = "" },
    });

    const stats = try windowStats(&database, "https://a.example", 60, 120);
    try testing.expectEqual(@as(u64, 4), stats.attempts);
    try testing.expectEqual(@as(?i64, null), stats.last_failure_ts);
    try testing.expectEqualStrings("", stats.lastFailureError());
}

test "pruneOlderThan counts minute rows only and drops the orphaned target" {
    var database = try openLog();
    defer database.close();

    try flush(&database, &.{
        .{ .url = "https://old.example", .minute_ts = 60, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
        .{ .url = "https://old.example", .minute_ts = 120, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
        .{ .url = "https://kept.example", .minute_ts = 120, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
        .{ .url = "https://kept.example", .minute_ts = 300, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
    });

    // Three minute rows go; the two target deletes must not join the count.
    try testing.expectEqual(@as(i64, 3), try pruneOlderThan(&database, 300));
    try testing.expectEqual(@as(i64, 1), try countMinutes(&database));
    // "old.example" has nothing left, "kept.example" still does.
    try testing.expectEqual(@as(i64, 1), try countTargets(&database));

    // The row exactly at the cutoff stays, and a second pass finds nothing.
    try testing.expectEqual(@as(i64, 0), try pruneOlderThan(&database, 300));
    try testing.expectEqual(@as(i64, 1), try countMinutes(&database));
}

test "a failed prune leaves both tables as they were" {
    var database = try openLog();
    defer database.close();
    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
    });
    try database.exec(
        \\CREATE TRIGGER refuse_target_delete BEFORE DELETE ON upstream_targets
        \\BEGIN SELECT RAISE(ABORT, 'refused'); END;
    );

    // The minute delete succeeds and the target delete does not; one
    // transaction means neither survives.
    try testing.expectError(error.Constraint, pruneOlderThan(&database, 300));
    try testing.expectEqual(@as(i64, 1), try countMinutes(&database));
    try testing.expectEqual(@as(i64, 1), try countTargets(&database));
}

test "an error name longer than the buffer is truncated, not overflowed" {
    var database = try openLog();
    defer database.close();
    const long = "A" ** 200;
    try flush(&database, &.{
        .{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 70, .last_error = long },
    });

    const stats = try windowStats(&database, "https://a.example", 60, 120);
    try testing.expectEqual(@as(usize, health.error_name_capacity), stats.lastFailureError().len);
    try testing.expectEqualStrings(long[0..health.error_name_capacity], stats.lastFailureError());
}