nxdns/src/storage/querylog_fixtures.zig

//! The shipped `querylog.db` fixtures, and the proof that every supported
//! schema version reaches the current one with the operator's rows intact.
//!
//! A file of its own, not a section of `querylog_migrations.zig`, because of the
//! link contract that split `querylog_versions.zig` out in the first place. The
//! assertions here need `repositories/queries_repo.zig`'s projection-coherence
//! oracle, and that file reaches across `src/` for the config and filter types;
//! importing it from `querylog_migrations.zig` would pull all of it into the
//! module `tools/cut.zig` builds `querylog_schema.zig` as, where those paths lie
//! outside the module root and do not compile.
//!
//! The fixtures themselves — `testdata/querylog-v<N>-{schema,data}.sql` — are
//! immutable once released. Every version in `[minimum_supported_version,
//! current_version]` has a pair: the current version's pair is the next
//! migration's starting point, and an explicit break ships the new baseline.

const std = @import("std");

const db = @import("db.zig");
const migrations = @import("querylog_migrations.zig");
const queries_repo = @import("repositories/queries_repo.zig");
const querylog_schema = @import("querylog_schema.zig");
const versions = @import("querylog_versions.zig");

const testing = std.testing;

/// A temporary directory and the `querylog.db` path inside it. Deliberately not
/// `querylog_migrations.zig`'s test harness: that one builds synthetic schemas,
/// while everything here starts from the shipped fixture files.
const Harness = struct {
    threaded: std.Io.Threaded,
    tmp: std.testing.TmpDir,
    buf: [256]u8 = undefined,

    fn init() Harness {
        return .{
            .threaded = .init(testing.allocator, .{}),
            .tmp = testing.tmpDir(.{ .iterate = true }),
        };
    }

    fn deinit(self: *Harness) void {
        self.tmp.cleanup();
        self.threaded.deinit();
    }

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

    fn path(self: *Harness) [:0]const u8 {
        return std.fmt.bufPrintZ(&self.buf, ".zig-cache/tmp/{s}/querylog.db", .{self.tmp.sub_path}) catch
            unreachable;
    }

    fn openLive(self: *Harness) !db.Db {
        var database = try db.Db.open(self.path(), .{ .mode = .read_write_existing });
        errdefer database.close();
        try db.applyPragmas(&database, .{});
        return database;
    }
};

/// One shipped schema version's frozen pair. Both halves are immutable once
/// released — the release cut byte-compares them against the previous tag — and
/// every version in `[minimum_supported_version, current_version]` must have a
/// pair, which the cut gate also enforces.
const Fixture = struct {
    version: i32,
    schema: [:0]const u8,
    data: [:0]const u8,
};

const fixtures = [_]Fixture{
    .{
        .version = 1,
        .schema = @embedFile("testdata/querylog-v1-schema.sql"),
        .data = @embedFile("testdata/querylog-v1-data.sql"),
    },
};

fn fixtureFor(version: i32) ?Fixture {
    for (fixtures) |fixture| {
        if (fixture.version == version) return fixture;
    }
    return null;
}

/// Writes a fixture pair to `path` and stamps it. `stamp` is a parameter rather
/// than `fixture.version` because the legacy-fingerprint file is the same
/// version-1 bytes under a different stamp.
fn writeFixture(path: [:0]const u8, fixture: Fixture, stamp: i32) !void {
    var database = try db.Db.open(path, .{ .mode = .read_write_create });
    defer database.close();
    try db.applyPragmas(&database, .{});
    try database.exec(fixture.schema);
    try database.exec(fixture.data);

    var buf: [64]u8 = undefined;
    const sql = std.fmt.bufPrintZ(&buf, "PRAGMA user_version = {d};", .{stamp}) catch unreachable;
    try database.exec(sql);
}

/// A second path in the harness's directory, for the reference databases the
/// assertions below compare against.
fn sidePath(h: *Harness, buf: []u8, name: []const u8) [:0]const u8 {
    return std.fmt.bufPrintZ(buf, ".zig-cache/tmp/{s}/{s}", .{ h.tmp.sub_path, name }) catch unreachable;
}

/// A database holding nothing but the current `ddl`, which is what a file
/// created by this build is.
fn openFreshCurrent(h: *Harness, buf: []u8) !db.Db {
    var database = try db.Db.open(sidePath(h, buf, "fresh.db"), .{ .mode = .read_write_create });
    errdefer database.close();
    try db.applyPragmas(&database, .{});
    try database.exec(querylog_schema.ddl);
    return database;
}

/// An untouched load of `fixture`, to compare a migrated or opened file against
/// rather than restating the fixture's contents in the assertions.
fn openPristine(h: *Harness, buf: []u8, fixture: Fixture) !db.Db {
    const path = sidePath(h, buf, "pristine.db");
    try writeFixture(path, fixture, fixture.version);
    var database = try db.Db.open(path, .{ .mode = .read_write_existing });
    errdefer database.close();
    try db.applyPragmas(&database, .{});
    return database;
}

/// Every row of every table, as one canonical text. Exact equality is only the
/// right question for a file no migration has reshaped; a migrated file is
/// checked by the counts and the watermark instead.
fn dumpContent(gpa: std.mem.Allocator, database: *db.Db, out: *std.ArrayList(u8)) !void {
    var tables: std.ArrayList([]u8) = .empty;
    defer freeOwned(gpa, &tables);
    {
        var stmt = try database.prepare(
            \\SELECT name FROM sqlite_schema
            \\WHERE type = 'table' AND name NOT LIKE 'sqlite\_%' ESCAPE '\'
            \\ORDER BY name
        );
        defer stmt.deinit();
        // Copied out before the per-table statements step: a borrowed
        // `columnText` would not survive them.
        while (try stmt.step()) try tables.append(gpa, try stmt.columnTextAlloc(gpa, 0));
    }

    // The lines are sorted here rather than by the query: the four `bucket_*`
    // tables are WITHOUT ROWID, so `ORDER BY rowid` is not available to all of
    // them and no single column list is.
    var lines: std.ArrayList([]u8) = .empty;
    defer freeOwned(gpa, &lines);

    for (tables.items) |table| {
        var sql_buf: [256]u8 = undefined;
        const sql = std.fmt.bufPrint(&sql_buf, "SELECT * FROM \"{s}\"", .{table}) catch unreachable;
        var stmt = try database.prepare(sql);
        defer stmt.deinit();
        while (try stmt.step()) {
            var line: std.ArrayList(u8) = .empty;
            errdefer line.deinit(gpa);
            try line.print(gpa, "R|{s}", .{table});
            var col: c_int = 0;
            const columns: c_int = db.c.sqlite3_column_count(stmt.handle);
            while (col < columns) : (col += 1) {
                try line.print(gpa, "|{s}", .{stmt.columnTextOrNull(col) orelse "<null>"});
            }
            try lines.append(gpa, try line.toOwnedSlice(gpa));
        }
    }

    std.mem.sortUnstable([]u8, lines.items, {}, struct {
        fn lessThan(_: void, a: []u8, b: []u8) bool {
            return std.mem.lessThan(u8, a, b);
        }
    }.lessThan);
    for (lines.items) |line| {
        try out.appendSlice(gpa, line);
        try out.append(gpa, '\n');
    }
}

/// One value, encoded so that no two different values can produce the same
/// bytes: a type tag, the byte length, then the bytes themselves.
///
/// Nothing here is a sentinel and nothing is escaped, which is the point. A
/// serialization that wrote NULL as `<null>` cannot tell a NULL apart from the
/// six-character string of the same name, and one that separated values with
/// `|` cannot tell `a|b` in one column from `a` and `b` in two — so a migration
/// that turned a NULL `upstream` into text, or shifted a value from one column
/// into its neighbour, would compare EQUAL to the original. The length prefix
/// makes the stream uniquely decodable, so equal encodings mean equal rows.
///
/// A float travels as its bit pattern rather than as printed digits: the
/// question here is whether the value survived, not whether it rounds the same.
fn writeValue(gpa: std.mem.Allocator, stmt: *db.Stmt, col: c_int, out: *std.ArrayList(u8)) !void {
    switch (db.c.sqlite3_column_type(stmt.handle, col)) {
        db.column_type.null_value => try out.print(gpa, "n0:", .{}),
        db.column_type.integer => {
            var buf: [24]u8 = undefined;
            const text = std.fmt.bufPrint(&buf, "{d}", .{stmt.columnInt(col)}) catch unreachable;
            try out.print(gpa, "i{d}:{s}", .{ text.len, text });
        },
        db.column_type.float => {
            const bits: u64 = @bitCast(db.c.sqlite3_column_double(stmt.handle, col));
            var buf: [24]u8 = undefined;
            const text = std.fmt.bufPrint(&buf, "{d}", .{bits}) catch unreachable;
            try out.print(gpa, "f{d}:{s}", .{ text.len, text });
        },
        db.column_type.blob => {
            // `columnText` on a blob hands back the same bytes SQLite stores,
            // which is what this compares; it is not read as text.
            const bytes = stmt.columnText(col);
            try out.print(gpa, "b{d}:{s}", .{ bytes.len, bytes });
        },
        else => {
            const bytes = stmt.columnText(col);
            try out.print(gpa, "t{d}:{s}", .{ bytes.len, bytes });
        },
    }
}

/// One query's rows, in the order the query returns them, tagged with `label` so
/// that a difference names the relation it came from. The column count is part
/// of each row for the same reason the lengths are part of each value.
fn dumpQuery(
    gpa: std.mem.Allocator,
    database: *db.Db,
    label: []const u8,
    sql: [:0]const u8,
    out: *std.ArrayList(u8),
) !void {
    var stmt = try database.prepare(sql);
    defer stmt.deinit();
    while (try stmt.step()) {
        const columns: c_int = db.c.sqlite3_column_count(stmt.handle);
        try out.print(gpa, "{s}:{d}:", .{ label, columns });
        var col: c_int = 0;
        while (col < columns) : (col += 1) {
            try writeValue(gpa, &stmt, col, out);
        }
        try out.append(gpa, '\n');
    }
}

/// The operator's data as the application means it, in an order a migration
/// cannot permute.
///
/// `q.*` rather than a column list on purpose: a migration that adds a column
/// must show that column here, and a hand-written list would quietly stop
/// covering the table the day it grows. `d.domain` rides along so that the text
/// a row names is compared, not only the id it happens to hold.
const logical_relations = [_]struct { label: []const u8, sql: [:0]const u8 }{
    .{
        .label = "query_log",
        .sql =
        \\SELECT q.*, d.domain FROM query_log q
        \\JOIN domains d ON d.id = q.domain_id
        \\ORDER BY q.id
        ,
    },
    .{ .label = "domains", .sql = "SELECT * FROM domains ORDER BY id" },
    .{ .label = "querylog_meta", .sql = "SELECT * FROM querylog_meta ORDER BY id" },
};

fn dumpLogical(gpa: std.mem.Allocator, database: *db.Db, out: *std.ArrayList(u8)) !void {
    for (logical_relations) |relation| {
        try dumpQuery(gpa, database, relation.label, relation.sql, out);
    }
}

fn freeOwned(gpa: std.mem.Allocator, list: *std.ArrayList([]u8)) void {
    for (list.items) |item| gpa.free(item);
    list.deinit(gpa);
}

fn expectSameContent(a: *db.Db, b: *db.Db) !void {
    var text_a: std.ArrayList(u8) = .empty;
    defer text_a.deinit(testing.allocator);
    var text_b: std.ArrayList(u8) = .empty;
    defer text_b.deinit(testing.allocator);
    try dumpContent(testing.allocator, a, &text_a);
    try dumpContent(testing.allocator, b, &text_b);
    try testing.expectEqualStrings(text_a.items, text_b.items);
}

fn rowCount(database: *db.Db, table: []const u8) !i64 {
    var buf: [128]u8 = undefined;
    return database.queryInt(std.fmt.bufPrint(&buf, "SELECT count(*) FROM \"{s}\"", .{table}) catch unreachable);
}

/// What must hold after a fixture has come through `openVersioned`, whatever
/// path it took: every row the operator had is still there with the same
/// CONTENT, and the projections still agree with the raw rows.
///
/// Counting rows and checking one watermark is what this used to do, and a
/// migration that shifted a timestamp, dropped a `qtype` or crossed two rows'
/// `client_ip` values passed it. The comparison is therefore the full logical
/// content of the three operator tables, `available_since` included as one
/// column of `querylog_meta` among the rest. The counts stay because a count
/// difference is the failure worth naming plainly.
///
/// The relations are named rather than derived because they are the ones holding
/// operator data; the `bucket_*` projections are derived from them, which
/// `expectProjectionsMatchRecompute` is the right check for. A future migration
/// that renames a table amends this alongside the step that does it. So does one
/// that renumbers `id` values: this asserts they survive, which every rebuild
/// written to the A.2 rule does.
fn expectFixtureSurvived(opened: *db.Db, pristine: *db.Db) !void {
    for ([_][]const u8{ "query_log", "domains", "querylog_meta" }) |table| {
        try testing.expectEqual(try rowCount(pristine, table), try rowCount(opened, table));
    }

    var opened_text: std.ArrayList(u8) = .empty;
    defer opened_text.deinit(testing.allocator);
    var pristine_text: std.ArrayList(u8) = .empty;
    defer pristine_text.deinit(testing.allocator);
    try dumpLogical(testing.allocator, opened, &opened_text);
    try dumpLogical(testing.allocator, pristine, &pristine_text);
    try testing.expectEqualStrings(pristine_text.items, opened_text.items);

    try queries_repo.expectProjectionsMatchRecompute(opened);
}

test "the survival comparison tells a NULL apart from text that looks like one" {
    // The mutation a count-and-watermark check misses entirely, and a
    // sentinel-string serialization misses just as completely: one column of one
    // row stops being NULL and becomes the very text the sentinel used. Every
    // count, the watermark and the projections all still agree.
    const fixture = fixtureFor(1) orelse return error.MissingFixture;

    var h: Harness = .init();
    defer h.deinit();
    try writeFixture(h.path(), fixture, fixture.version);

    var side: [256]u8 = undefined;
    var pristine = try openPristine(&h, &side, fixture);
    defer pristine.close();

    var corrupted = try h.openLive();
    defer corrupted.close();
    try testing.expectEqual(@as(i64, 0), corrupted.changes());
    try corrupted.exec(
        \\UPDATE query_log SET upstream = '<null>'
        \\WHERE id = (SELECT min(id) FROM query_log WHERE upstream IS NULL)
    );
    // The fixture has to actually carry a NULL `upstream` for this to be a test
    // of anything.
    try testing.expectEqual(@as(i64, 1), corrupted.changes());

    // Everything the old check looked at still agrees, which is why it passed.
    for ([_][]const u8{ "query_log", "domains", "querylog_meta" }) |table| {
        try testing.expectEqual(try rowCount(&pristine, table), try rowCount(&corrupted, table));
    }
    try testing.expectEqual(
        try pristine.queryInt("SELECT available_since FROM querylog_meta WHERE id = 1"),
        try corrupted.queryInt("SELECT available_since FROM querylog_meta WHERE id = 1"),
    );
    try queries_repo.expectProjectionsMatchRecompute(&corrupted);

    // The dumps are compared here rather than through `expectFixtureSurvived`
    // so that a PASSING run stays silent: `expectEqualStrings` prints the whole
    // diff before it returns its error, and this is the comparison that function
    // makes.
    var corrupted_text: std.ArrayList(u8) = .empty;
    defer corrupted_text.deinit(testing.allocator);
    var pristine_text: std.ArrayList(u8) = .empty;
    defer pristine_text.deinit(testing.allocator);
    try dumpLogical(testing.allocator, &corrupted, &corrupted_text);
    try dumpLogical(testing.allocator, &pristine, &pristine_text);
    try testing.expect(!std.mem.eql(u8, corrupted_text.items, pristine_text.items));
}

test "every supported version ships a fixture pair" {
    // The cut gate enforces this against a release; the suite enforces it
    // against a commit, so a version bump that forgot its fixtures fails here
    // long before anyone reaches for `zig build cut`.
    var version = versions.minimum_supported_version;
    while (version <= versions.current_version) : (version += 1) {
        try testing.expect(fixtureFor(version) != null);
    }
}

test "each shipped fixture pair is coherent before any migration touches it" {
    for (fixtures) |fixture| {
        var h: Harness = .init();
        defer h.deinit();
        try writeFixture(h.path(), fixture, fixture.version);

        var database = try h.openLive();
        defer database.close();
        try testing.expectEqual(
            @as(i64, fixture.version),
            try database.queryInt("PRAGMA user_version"),
        );
        // An incoherent fixture has to fail as a fixture, not later as a
        // migration that appears to have corrupted the projections.
        try queries_repo.expectProjectionsMatchRecompute(&database);
        try testing.expectEqual(@as(i64, 0), try database.queryInt("SELECT count(*) FROM pragma_foreign_key_check"));
    }
}

test "every fixture below the current version migrates to it through the shipped chain" {
    // Empty while the chain is: `minimum_supported_version == current_version`
    // today. It is written as the loop so that the day a step ships, the
    // fixture it starts from is proved through the REAL production plan with no
    // edit to this test.
    var version = versions.minimum_supported_version;
    while (version < versions.current_version) : (version += 1) {
        const fixture = fixtureFor(version) orelse return error.MissingFixture;

        var h: Harness = .init();
        defer h.deinit();
        try writeFixture(h.path(), fixture, version);

        var side: [256]u8 = undefined;
        var pristine = try openPristine(&h, &side, fixture);
        defer pristine.close();

        var result = try querylog_schema.open(h.io(), std.Io.Dir.cwd(), h.path());
        defer result.database.close();
        try testing.expectEqual(@as(?querylog_schema.RecreateReason, null), result.recreated);
        try testing.expectEqual(
            @as(i64, versions.current_version),
            try result.database.queryInt("PRAGMA user_version"),
        );

        var fresh_buf: [256]u8 = undefined;
        var fresh = try openFreshCurrent(&h, &fresh_buf);
        defer fresh.close();
        try testing.expect(try migrations.schemaEquivalent(testing.allocator, &result.database, &fresh));
        try expectFixtureSurvived(&result.database, &pristine);
    }
}

test "the current version's fixture pair opens on the current lane unchanged" {
    // The loop above never reaches this pair, and Gate 2 of the cut requires it
    // to exist. This is what proves it is a real, coherent file rather than one
    // shipped to satisfy a gate.
    const fixture = fixtureFor(versions.current_version) orelse return error.MissingFixture;

    var h: Harness = .init();
    defer h.deinit();
    try writeFixture(h.path(), fixture, versions.current_version);

    var side: [256]u8 = undefined;
    var pristine = try openPristine(&h, &side, fixture);
    defer pristine.close();

    var result = try querylog_schema.open(h.io(), std.Io.Dir.cwd(), h.path());
    defer result.database.close();
    try testing.expectEqual(@as(?querylog_schema.RecreateReason, null), result.recreated);
    try testing.expectEqual(
        @as(i64, versions.current_version),
        try result.database.queryInt("PRAGMA user_version"),
    );

    var fresh_buf: [256]u8 = undefined;
    var fresh = try openFreshCurrent(&h, &fresh_buf);
    defer fresh.close();
    try testing.expect(try migrations.schemaEquivalent(testing.allocator, &result.database, &fresh));
    try expectFixtureSurvived(&result.database, &pristine);
    // No migration ran, so nothing reshaped anything: byte-for-byte the rows
    // that were loaded.
    try expectSameContent(&result.database, &pristine);
}

test "a fixture carrying the 0.0.12 fingerprint restamps, and refuses once the minimum rises" {
    const fixture = fixtureFor(1) orelse return error.MissingFixture;

    // Version 1 is the legacy fingerprint's logical version, so the pair only
    // has anything to say while 1 is still supported.
    if (versions.minimum_supported_version <= 1 and versions.current_version >= 1) {
        var h: Harness = .init();
        defer h.deinit();
        try writeFixture(h.path(), fixture, versions.legacy_fingerprint);

        var side: [256]u8 = undefined;
        var pristine = try openPristine(&h, &side, fixture);
        defer pristine.close();

        var result = try querylog_schema.open(h.io(), std.Io.Dir.cwd(), h.path());
        defer result.database.close();
        try testing.expectEqual(@as(?querylog_schema.RecreateReason, null), result.recreated);
        try testing.expectEqual(
            @as(i64, versions.current_version),
            try result.database.queryInt("PRAGMA user_version"),
        );

        var fresh_buf: [256]u8 = undefined;
        var fresh = try openFreshCurrent(&h, &fresh_buf);
        defer fresh.close();
        try testing.expect(try migrations.schemaEquivalent(testing.allocator, &result.database, &fresh));
        try expectFixtureSurvived(&result.database, &pristine);
    }

    // The companion, already in the suite for the day a break raises the
    // minimum above 1: the same bytes under the same stamp are then a file this
    // build cannot reach, and it is refused without being touched.
    var h: Harness = .init();
    defer h.deinit();
    try writeFixture(h.path(), fixture, versions.legacy_fingerprint);

    const after_break: querylog_schema.Plan = .{
        .minimum = 2,
        .current = 2,
        .legacy_fingerprint = versions.legacy_fingerprint,
        .step_sql = &.{},
    };
    try testing.expect(querylog_schema.classify(versions.legacy_fingerprint, after_break).action ==
        .refuse_unsupported);
    try testing.expect(!querylog_schema.classify(versions.legacy_fingerprint, after_break).restamp);

    var handle: ?db.Db = try h.openLive();
    defer if (handle) |*open_db| open_db.close();

    migrations.expected_failures.begin();
    defer migrations.expected_failures.end();
    try testing.expectError(
        error.SchemaUnsupported,
        querylog_schema.openVersioned(h.io(), std.Io.Dir.cwd(), h.path(), &handle, after_break),
    );

    var check = try h.openLive();
    defer check.close();
    try testing.expectEqual(
        @as(i64, versions.legacy_fingerprint),
        try check.queryInt("PRAGMA user_version"),
    );
    var pristine_buf: [256]u8 = undefined;
    var pristine = try openPristine(&h, &pristine_buf, fixture);
    defer pristine.close();
    try expectSameContent(&check, &pristine);
}