//! Client auto-materialisation (PLAN §7.2): every address that asks a question
//! ends up as a row in `clients`, so the operator can name it and assign it a
//! group without typing an address by hand.
//!
//! A DNS query must never wait on a database write, so `track` only records the
//! address in a fixed-capacity table in memory. A background loop drains that
//! table every `flush_interval_s` and writes one row per distinct client, and
//! prunes the rows of devices that went quiet on every
//! `prune_every_passes`-th pass.
//!
//! The table is bounded at `max_pending`. A full table drops the address and
//! counts it under `dropped_full`: a burst of spoofed source addresses must not
//! be able to grow this allocation, and a dropped address costs nothing, since
//! the next query from that client tracks it again.
//!
//! A materialised row does not reach the matcher until the next manager reload.
//! Nothing depends on it: `groupForClient` already falls back to the prefix
//! rules and then to the default group, so the row exists for the operator's
//! benefit, not for resolution.

const std = @import("std");

const address = @import("../platform/address.zig");
const clients_repo = @import("../storage/repositories/clients_repo.zig");
const db = @import("../storage/db.zig");
const disk_monitor = @import("../storage/disk_monitor.zig");

const log = std.log.scoped(.clients);

/// RFC 5952 text of any IPv6 address. `format` never writes more than this, so
/// the formatting in `flushOnce` cannot fail.
const max_ip_text = 45;

/// One address waiting for its row, with the wall-clock second of its most
/// recent query.
const Pending = struct {
    addr: address.NetAddress,
    last_seen: i64,
};

pub const Tracker = struct {
    /// Distinct clients one flush interval can carry. A household LAN holds two
    /// orders of magnitude fewer; the headroom is for the spoofing case.
    pub const max_pending = 512;
    pub const flush_interval_s = 60;
    /// One day at `flush_interval_s` seconds per pass.
    pub const prune_every_passes = 1440;

    /// `tracked` counts the `track` calls that landed in the table, whether
    /// they created an entry or refreshed one, so `tracked + dropped_full` is
    /// the number of `track` calls.
    pub const Stats = struct {
        tracked: u64 = 0,
        flushed: u64 = 0,
        dropped_full: u64 = 0,
        pruned: u64 = 0,
        flush_failures: u64 = 0,
    };

    /// Guards `pending`, `count`, `passes` and `stats`. Every field below is
    /// written under it, so a reader takes it too; see `snapshotStats`.
    mutex: std.Io.Mutex,
    retention_days: u16,
    pending: [max_pending]Pending,
    count: u32,
    passes: u64,
    stats: Stats,

    /// `retention_days` is `logging.retention_days`, the same knob the query log
    /// prunes by (milestone-7 ruling 16). A client silent for that long is as
    /// uninteresting as a query that old.
    pub fn init(retention_days: u16) Tracker {
        return .{
            .mutex = .init,
            .retention_days = retention_days,
            .pending = undefined,
            .count = 0,
            .passes = 0,
            .stats = .{},
        };
    }

    /// Records `addr` as seen now. Called from the query path, so it writes no
    /// database and returns no error: a full table drops the address.
    ///
    /// `lockUncancelable` rather than `lock`: the caller is `Handler.handle`,
    /// which has no error union to carry `error.Canceled` out of. The critical
    /// section is a scan of at most `max_pending` addresses and holds no I/O.
    pub fn track(self: *Tracker, io: std.Io, addr: address.NetAddress) void {
        self.trackAt(io, addr, std.Io.Clock.real.now(io).toSeconds());
    }

    /// `track` with the timestamp supplied, so a test does not depend on the
    /// wall clock.
    pub fn trackAt(self: *Tracker, io: std.Io, addr: address.NetAddress, now_s: i64) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);

        for (self.pending[0..self.count]) |*entry| {
            if (!entry.addr.eql(addr)) continue;
            entry.last_seen = now_s;
            self.stats.tracked += 1;
            return;
        }
        if (self.count == max_pending) {
            self.stats.dropped_full += 1;
            return;
        }
        self.pending[self.count] = .{ .addr = addr, .last_seen = now_s };
        self.count += 1;
        self.stats.tracked += 1;
    }

    /// Flush loop, first flush one interval in: an empty table at startup has
    /// nothing to write.
    ///
    /// `boot` rather than `awake`, so a box that suspends still sees its day
    /// elapse and prunes on schedule.
    ///
    /// `database` must be a connection dedicated to this loop: no other task may
    /// use the same handle while it runs. `FULLMUTEX` (`db.zig:218`) serializes
    /// one SQLite call against another, but a transaction is connection state,
    /// not call state, so a flush that lands between another writer's BEGIN and
    /// COMMIT would commit or roll back with that writer's batch. Milestone-7
    /// ruling 21 gives this loop its own `config.db` connection; the tracker
    /// opens nothing itself.
    ///
    /// `monitor` gates the pass: while free space is critical the tracker writes
    /// nothing, and the addresses it would have written stay dropped.
    pub fn run(
        self: *Tracker,
        io: std.Io,
        database: *db.Db,
        monitor: ?*disk_monitor.Monitor,
    ) std.Io.Cancelable!void {
        const interval: std.Io.Clock.Duration = .{
            .raw = .fromSeconds(flush_interval_s),
            .clock = .boot,
        };
        while (true) {
            try interval.sleep(io);
            const writes_allowed = if (monitor) |m| m.writesAllowed() else true;
            self.flushOnce(io, database, writes_allowed);
        }
    }

    /// One pass: drain the table, write a row per client, and prune on every
    /// `prune_every_passes`-th pass.
    ///
    /// A gated pass does nothing at all, not even count: the work it skipped is
    /// still owed, and the pending addresses it leaves behind are re-tracked by
    /// the next query from each client.
    ///
    /// Every database failure logs one line at `warn` and counts. Nothing
    /// retries within a pass: the dropped addresses come back on their own, and
    /// a failed prune repeats the same work a day later against the same rows.
    ///
    /// Only `run` may call this concurrently with itself — the drain buffer is
    /// this call's stack, but the pass counter and the prune schedule assume a
    /// single caller.
    pub fn flushOnce(self: *Tracker, io: std.Io, database: *db.Db, writes_allowed: bool) void {
        if (!writes_allowed) return;

        var drained: [max_pending]Pending = undefined;
        const batch = self.drain(io, &drained);

        var flushed: u64 = 0;
        var failures: u64 = 0;
        for (batch) |entry| {
            var buf: [max_ip_text]u8 = undefined;
            var w: std.Io.Writer = .fixed(&buf);
            entry.addr.format(&w) catch unreachable;

            if (clients_repo.upsertSeen(database, w.buffered(), entry.last_seen)) {
                flushed += 1;
            } else |err| {
                if (failures == 0) {
                    log.warn("materialising client {s} failed: {s}", .{ w.buffered(), @errorName(err) });
                }
                failures += 1;
            }
        }

        self.mutex.lockUncancelable(io);
        self.passes += 1;
        self.stats.flushed += flushed;
        self.stats.flush_failures += failures;
        const due = self.passes % prune_every_passes == 0;
        self.mutex.unlock(io);

        if (!due) return;
        const cutoff = std.Io.Clock.real.now(io).toSeconds() - @as(i64, self.retention_days) * 86_400;
        if (clients_repo.pruneStale(database, cutoff)) |deleted| {
            self.mutex.lockUncancelable(io);
            self.stats.pruned += deleted;
            self.mutex.unlock(io);
        } else |err| {
            log.warn("pruning clients before {d} failed: {s}", .{ cutoff, @errorName(err) });
            self.mutex.lockUncancelable(io);
            self.stats.flush_failures += 1;
            self.mutex.unlock(io);
        }
    }

    pub fn snapshotStats(self: *Tracker, io: std.Io) Stats {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        return self.stats;
    }

    /// Clients waiting for their row. Reaching `max_pending` is what turns
    /// further addresses into `dropped_full`.
    pub fn pendingClients(self: *Tracker, io: std.Io) u32 {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        return self.count;
    }

    /// Empties the table into `out` and returns what it copied. The lock is
    /// released before any database call, so the query path never waits on
    /// SQLite.
    fn drain(self: *Tracker, io: std.Io, out: *[max_pending]Pending) []const Pending {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);

        const n = self.count;
        @memcpy(out[0..n], self.pending[0..n]);
        self.count = 0;
        return out[0..n];
    }
};

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

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

const testing = std.testing;

fn openMigrated() !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 lastSeen(database: *db.Db, ip: []const u8) !i64 {
    var stmt = try database.prepare("SELECT last_seen FROM clients WHERE ip = ?1");
    defer stmt.deinit();
    try stmt.bindText(1, ip);
    try testing.expect(try stmt.step());
    return stmt.columnInt(0);
}

fn parsed(text: []const u8) address.NetAddress {
    return address.NetAddress.parse(text) catch unreachable;
}

test "a client tracked twice before a flush yields one row at the later time" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openMigrated();
    defer database.close();

    var tracker: Tracker = .init(30);
    tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
    tracker.trackAt(io, parsed("192.168.1.10"), 1700000030);
    try testing.expectEqual(@as(u32, 1), tracker.pendingClients(io));

    tracker.flushOnce(io, &database, true);

    try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(i64, 1700000030), try lastSeen(&database, "192.168.1.10"));

    const stats = tracker.snapshotStats(io);
    try testing.expectEqual(@as(u64, 2), stats.tracked);
    try testing.expectEqual(@as(u64, 1), stats.flushed);
    try testing.expectEqual(@as(u64, 0), stats.flush_failures);
    try testing.expectEqual(@as(u32, 0), tracker.pendingClients(io));
}

test "distinct clients each get a row and ipv6 text is canonical" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openMigrated();
    defer database.close();

    var tracker: Tracker = .init(30);
    tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
    tracker.trackAt(io, parsed("192.168.1.11"), 1700000001);
    tracker.trackAt(io, parsed("fd00:0:0:0:0:0:0:1"), 1700000002);
    // An IPv4-mapped literal is the same client as its plain form.
    tracker.trackAt(io, address.NetAddress.fromIp(try std.Io.net.IpAddress.parse("::ffff:192.168.1.10", 53)), 1700000003);
    try testing.expectEqual(@as(u32, 3), tracker.pendingClients(io));

    tracker.flushOnce(io, &database, true);

    try testing.expectEqual(@as(i64, 3), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(i64, 1700000003), try lastSeen(&database, "192.168.1.10"));
    try testing.expectEqual(@as(i64, 1700000002), try lastSeen(&database, "fd00::1"));
    try testing.expectEqual(@as(u64, 3), tracker.snapshotStats(io).flushed);
}

test "a full table drops further clients and counts them" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openMigrated();
    defer database.close();

    var tracker: Tracker = .init(30);
    for (0..Tracker.max_pending) |i| {
        var octets: [4]u8 = undefined;
        std.mem.writeInt(u32, &octets, @intCast(i), .big);
        tracker.trackAt(io, .{ .ip4 = octets }, 1700000000);
    }
    try testing.expectEqual(@as(u32, Tracker.max_pending), tracker.pendingClients(io));
    try testing.expectEqual(@as(u64, 0), tracker.snapshotStats(io).dropped_full);

    tracker.trackAt(io, parsed("203.0.113.7"), 1700000000);
    tracker.trackAt(io, parsed("203.0.113.8"), 1700000000);
    try testing.expectEqual(@as(u32, Tracker.max_pending), tracker.pendingClients(io));

    const stats = tracker.snapshotStats(io);
    try testing.expectEqual(@as(u64, 2), stats.dropped_full);
    try testing.expectEqual(@as(u64, Tracker.max_pending), stats.tracked);

    // A tracked client still refreshes while the table is full, and the flush
    // makes room for the next newcomer.
    tracker.trackAt(io, .{ .ip4 = .{ 0, 0, 0, 0 } }, 1700000060);
    tracker.flushOnce(io, &database, true);
    try testing.expectEqual(@as(i64, Tracker.max_pending), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(i64, 1700000060), try lastSeen(&database, "0.0.0.0"));

    tracker.trackAt(io, parsed("203.0.113.7"), 1700000060);
    try testing.expectEqual(@as(u32, 1), tracker.pendingClients(io));
}

test "a flush touches a hand-edited row without changing what the operator set" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openMigrated();
    defer database.close();
    try database.exec("INSERT INTO groups (id, name) VALUES (2, 'kids');");
    try database.exec(
        \\INSERT INTO clients (ip, name, group_id, hand_edited, first_seen, last_seen)
        \\VALUES ('192.168.1.10', 'laptop', 2, 1, 1690000000, 1690000000);
    );

    var tracker: Tracker = .init(30);
    tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
    tracker.flushOnce(io, &database, true);

    try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(i64, 1700000000), try lastSeen(&database, "192.168.1.10"));
    try testing.expectEqual(
        @as(i64, 1),
        try database.queryInt(
            \\SELECT count(*) FROM clients
            \\ WHERE ip = '192.168.1.10' AND name = 'laptop' AND group_id = 2
            \\   AND hand_edited = 1 AND first_seen = 1690000000
        ),
    );
}

test "a gated pass writes nothing and keeps the pending clients" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openMigrated();
    defer database.close();

    var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);
    try testing.expect(!monitor.writesAllowed());

    var tracker: Tracker = .init(30);
    tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
    tracker.flushOnce(io, &database, monitor.writesAllowed());

    try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(u32, 1), tracker.pendingClients(io));
    try testing.expectEqual(@as(u64, 0), tracker.snapshotStats(io).flushed);
    try testing.expectEqual(@as(u64, 0), tracker.passes);

    // Free space recovers and the same pending client lands.
    monitor.state_raw.store(@intFromEnum(disk_monitor.State.warn), .monotonic);
    tracker.flushOnce(io, &database, monitor.writesAllowed());
    try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(u64, 1), tracker.passes);
}

test "a failing upsert counts and leaves the client to be tracked again" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openMigrated();
    defer database.close();
    try database.exec(
        \\CREATE TRIGGER refuse_insert BEFORE INSERT ON clients
        \\BEGIN SELECT RAISE(ABORT, 'refused'); END;
    );

    var tracker: Tracker = .init(30);
    tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
    tracker.trackAt(io, parsed("192.168.1.11"), 1700000000);
    tracker.flushOnce(io, &database, true);

    try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
    const stats = tracker.snapshotStats(io);
    try testing.expectEqual(@as(u64, 0), stats.flushed);
    try testing.expectEqual(@as(u64, 2), stats.flush_failures);
    try testing.expectEqual(@as(u32, 0), tracker.pendingClients(io));

    try database.exec("DROP TRIGGER refuse_insert;");
    tracker.trackAt(io, parsed("192.168.1.10"), 1700000060);
    tracker.flushOnce(io, &database, true);
    try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
}

test "the pass that comes due prunes the clients that went quiet" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openMigrated();
    defer database.close();

    const now = std.Io.Clock.real.now(io).toSeconds();
    const day = 86_400;
    try clients_repo.upsertSeen(&database, "10.0.0.1", now - 40 * day);
    try clients_repo.upsertSeen(&database, "10.0.0.2", now - 29 * day);

    var tracker: Tracker = .init(30);
    // Every pass before the due one leaves both rows alone.
    for (0..Tracker.prune_every_passes - 1) |_| {
        tracker.flushOnce(io, &database, true);
    }
    try testing.expectEqual(@as(i64, 2), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(u64, 0), tracker.snapshotStats(io).pruned);

    tracker.flushOnce(io, &database, true);

    try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(i64, now - 29 * day), try lastSeen(&database, "10.0.0.2"));
    try testing.expectEqual(@as(u64, 1), tracker.snapshotStats(io).pruned);
    try testing.expectEqual(@as(u64, Tracker.prune_every_passes), tracker.passes);
}

test "a shorter retention prunes what the default keeps" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openMigrated();
    defer database.close();

    const now = std.Io.Clock.real.now(io).toSeconds();
    try clients_repo.upsertSeen(&database, "10.0.0.1", now - 3 * 86_400);

    var tracker: Tracker = .init(1);
    tracker.passes = Tracker.prune_every_passes - 1;
    tracker.flushOnce(io, &database, true);

    try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(u64, 1), tracker.snapshotStats(io).pruned);
}

test "the run loop flushes on its interval and returns on cancel" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var database = try openMigrated();
    defer database.close();

    var tracker: Tracker = .init(30);
    tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);

    var future = try io.concurrent(Tracker.run, .{
        &tracker,
        io,
        &database,
        @as(?*disk_monitor.Monitor, null),
    });
    // The first flush is one interval away, so cancelling immediately proves the
    // loop starts by sleeping rather than by writing.
    try testing.expectError(error.Canceled, future.cancel(io));
    try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
    try testing.expectEqual(@as(u32, 1), tracker.pendingClients(io));
}