//! tombstones — which rows a compaction pass may physically drop
//!
//! upstream internal/tombstone (docs/upstream-compaction-spec.md): two
//! tombstone families, both keyed by seq with STRICT inequality:
//! - record: (did, collection, rkey) -> max superseding seq, produced by
//!   KindDelete and KindUpdate rows (update = delete + replace)
//! - did: did -> seq, produced by KindSync rows (repo diverged) and
//!   KindAccount rows whose payload says active=false && status=="deleted"
//!
//! suppression drops ONLY materialization rows (create/update/create_resync)
//! with seq strictly below the tombstone seq; marker rows (delete/account/
//! identity/sync) survive forever — clients fold the stream themselves,
//! there is no read-time overlay. tombstones are never persisted separately:
//! they ARE the event rows, and snapshots are rebuilt by folding decoded
//! rows. keys are cloned on insert — decoded events alias the decompressed
//! block buffer (same hazard as upstream).

const std = @import("std");
const zat = @import("zat");
const segment = @import("segment.zig");

const Allocator = std.mem.Allocator;

pub const DidReason = enum { sync, account };
pub const DropReason = enum { record, sync, account };

pub const DidTombstone = struct {
    seq: u64,
    reason: DidReason,
};

pub const Counts = struct {
    record: u64 = 0,
    sync: u64 = 0,
    account: u64 = 0,
};

/// a fold of tombstone rows over some seq window. also the live in-memory
/// set (fed per append in steady state) — same structure, the live set is
/// used only for the cap trigger, never for drop decisions.
pub const Snapshot = struct {
    allocator: Allocator,
    /// both maps own their individually allocated keys so eviction releases
    /// real resident memory rather than retaining an arena high-water mark.
    records: std.StringHashMapUnmanaged(u64) = .empty,
    dids: std.StringHashMapUnmanaged(DidTombstone) = .empty,
    /// running total matching approxBytes' historical formula, maintained at
    /// insert/evict so approxBytes is O(1). The live set publishes it per
    /// archive append (Compactor.onAppend); the previous full-map walk cost
    /// ~80% of the writer core at 119k entries (perf, 2026-08-06).
    approx_bytes: usize = 0,

    pub fn init(allocator: Allocator) Snapshot {
        return .{ .allocator = allocator };
    }

    pub fn deinit(self: *Snapshot) void {
        var records = self.records.keyIterator();
        while (records.next()) |key| self.allocator.free(key.*);
        var dids = self.dids.keyIterator();
        while (dids.next()) |key| self.allocator.free(key.*);
        self.records.deinit(self.allocator);
        self.dids.deinit(self.allocator);
    }

    pub fn len(self: *const Snapshot) usize {
        return self.records.count() + self.dids.count();
    }

    /// observe one event row; per-key max seq wins. non-marker kinds are
    /// ignored, so callers can feed every row unconditionally.
    pub fn observe(self: *Snapshot, ev: segment.Event) !void {
        switch (ev.kind) {
            .delete, .update => {
                var key_buf: [1024]u8 = undefined;
                const key = recordKey(&key_buf, ev.did, ev.collection, ev.rkey) orelse return;
                const gop = try self.records.getOrPut(self.allocator, key);
                if (!gop.found_existing) {
                    const owned = self.allocator.dupe(u8, key) catch |err| {
                        _ = self.records.remove(key);
                        return err;
                    };
                    gop.key_ptr.* = owned;
                    gop.value_ptr.* = ev.seq;
                    self.approx_bytes += owned.len - 2 + 64;
                } else if (ev.seq > gop.value_ptr.*) {
                    gop.value_ptr.* = ev.seq;
                }
            },
            .sync => try self.observeDid(ev.did, ev.seq, .sync),
            .account => {
                if (try accountIsDelete(self.allocator, ev.payload))
                    try self.observeDid(ev.did, ev.seq, .account);
            },
            .create, .create_resync, .identity => {},
        }
    }

    fn observeDid(self: *Snapshot, did: []const u8, seq: u64, reason: DidReason) !void {
        const gop = try self.dids.getOrPut(self.allocator, did);
        if (!gop.found_existing) {
            const owned = self.allocator.dupe(u8, did) catch |err| {
                _ = self.dids.remove(did);
                return err;
            };
            gop.key_ptr.* = owned;
            gop.value_ptr.* = .{ .seq = seq, .reason = reason };
            self.approx_bytes += owned.len + 64;
        } else if (seq > gop.value_ptr.seq) {
            gop.value_ptr.* = .{ .seq = seq, .reason = reason };
        }
    }

    /// true iff a compaction pass may physically drop this row: a
    /// materialization row strictly below a covering tombstone. marker rows
    /// are never dropped.
    pub fn shouldDrop(self: *const Snapshot, ev: segment.Event) bool {
        return self.dropReason(ev) != null;
    }

    /// Upstream attributes each physically removed row to the covering
    /// tombstone. DID-wide tombstones win over record tombstones.
    pub fn dropReason(self: *const Snapshot, ev: segment.Event) ?DropReason {
        switch (ev.kind) {
            .create, .update, .create_resync => {},
            else => return null,
        }
        if (self.dids.get(ev.did)) |t| {
            if (t.seq > ev.seq) return switch (t.reason) {
                .sync => .sync,
                .account => .account,
            };
        }
        var key_buf: [1024]u8 = undefined;
        const key = recordKey(&key_buf, ev.did, ev.collection, ev.rkey) orelse return null;
        if (self.records.get(key)) |t_seq| {
            if (t_seq > ev.seq) return .record;
        }
        return null;
    }

    /// Exact counterpart of upstream Snapshot's metric accounting. Counts
    /// are derived from the final map so a later account tombstone replacing
    /// a sync tombstone is attributed to account, not its insertion history.
    pub fn countsByReason(self: *const Snapshot) Counts {
        var result: Counts = .{ .record = self.records.count() };
        var it = self.dids.iterator();
        while (it.next()) |entry| switch (entry.value_ptr.reason) {
            .sync => result.sync += 1,
            .account => result.account += 1,
        };
        return result;
    }

    /// Key bytes plus upstream's conservative 64-byte map-entry overhead.
    pub fn approxBytes(self: *const Snapshot) usize {
        return self.approx_bytes;
    }

    /// Distinct DIDs represented by this snapshot. Returns null once the
    /// upstream narrowing cutoff is exceeded, signaling a full scan.
    pub fn candidateDids(self: *const Snapshot, allocator: Allocator, max: usize) !?[][]const u8 {
        var set: std.StringHashMapUnmanaged(void) = .empty;
        defer set.deinit(allocator);
        var records = self.records.iterator();
        while (records.next()) |entry| {
            const nul = std.mem.indexOfScalar(u8, entry.key_ptr.*, 0).?;
            try set.put(allocator, entry.key_ptr.*[0..nul], {});
            if (set.count() > max) return null;
        }
        var dids = self.dids.iterator();
        while (dids.next()) |entry| {
            try set.put(allocator, entry.key_ptr.*, {});
            if (set.count() > max) return null;
        }
        const result = try allocator.alloc([]const u8, set.count());
        var index: usize = 0;
        var it = set.keyIterator();
        while (it.next()) |did| : (index += 1) result[index] = did.*;
        return result;
    }

    /// drop entries with seq <= watermark (they cover only rows already
    /// physically compacted). live-set maintenance; snapshots are throwaway.
    pub fn evict(self: *Snapshot, watermark: u64) void {
        var rit = self.records.iterator();
        while (rit.next()) |e| {
            if (e.value_ptr.* <= watermark) {
                const key = e.key_ptr.*;
                self.records.removeByPtr(e.key_ptr);
                self.approx_bytes -= key.len - 2 + 64;
                self.allocator.free(key);
            }
        }
        var dit = self.dids.iterator();
        while (dit.next()) |e| {
            if (e.value_ptr.seq <= watermark) {
                const key = e.key_ptr.*;
                self.dids.removeByPtr(e.key_ptr);
                self.approx_bytes -= key.len + 64;
                self.allocator.free(key);
            }
        }
    }
};

/// (did, collection, rkey) -> "did\x00collection\x00rkey". components never
/// contain NUL (did/nsid/rkey syntax); null on overflow (caller drops the
/// observation — such a row also fails segment validation upstream of here).
fn recordKey(buf: []u8, did: []const u8, collection: []const u8, rkey: []const u8) ?[]const u8 {
    return std.fmt.bufPrint(buf, "{s}\x00{s}\x00{s}", .{ did, collection, rkey }) catch null;
}

/// does an #account payload mark the account deleted? upstream decodes the
/// envelope per fold; matches active == false && status == "deleted".
pub fn accountIsDelete(allocator: Allocator, payload: []const u8) !bool {
    var arena = std.heap.ArenaAllocator.init(allocator);
    defer arena.deinit();
    const env = try zat.cbor.decodeAll(arena.allocator(), payload);
    const active = env.getBool("active") orelse return false;
    if (active) return false;
    const status = env.getString("status") orelse return false;
    return std.mem.eql(u8, status, "deleted");
}

// === tests ===

const testing = std.testing;

fn row(kind: segment.Kind, seq: u64, did: []const u8, coll: []const u8, rkey: []const u8, payload: []const u8) segment.Event {
    return .{
        .seq = seq,
        .witnessed_at = 1,
        .indexed_at = 0,
        .kind = kind,
        .did = did,
        .collection = coll,
        .rkey = rkey,
        .rev = "3l3qo2vutsw2b",
        .payload = payload,
    };
}

// {"active": false, "status": "deleted"} / {"active": true} as dag-cbor
const account_deleted = "\xa2\x66\x61\x63\x74\x69\x76\x65\xf4\x66\x73\x74\x61\x74\x75\x73\x67\x64\x65\x6c\x65\x74\x65\x64";
const account_active = "\xa1\x66\x61\x63\x74\x69\x76\x65\xf5";

test "record tombstones: strict inequality, max-seq wins, markers survive" {
    var snap = Snapshot.init(testing.allocator);
    defer snap.deinit();

    try snap.observe(row(.delete, 10, "did:plc:a", "app.bsky.feed.post", "k1", ""));
    try snap.observe(row(.update, 7, "did:plc:a", "app.bsky.feed.post", "k1", "x")); // lower, ignored
    try testing.expectEqual(@as(usize, 1), snap.len());

    // strictly below 10 → drop; at 10 (the marker's own seq) or above → keep
    try testing.expect(snap.shouldDrop(row(.create, 9, "did:plc:a", "app.bsky.feed.post", "k1", "x")));
    try testing.expect(snap.shouldDrop(row(.update, 9, "did:plc:a", "app.bsky.feed.post", "k1", "x")));
    try testing.expect(snap.shouldDrop(row(.create_resync, 9, "did:plc:a", "app.bsky.feed.post", "k1", "x")));
    try testing.expect(!snap.shouldDrop(row(.create, 10, "did:plc:a", "app.bsky.feed.post", "k1", "x")));
    try testing.expect(!snap.shouldDrop(row(.create, 11, "did:plc:a", "app.bsky.feed.post", "k1", "x")));
    // different rkey untouched; marker kinds never drop
    try testing.expect(!snap.shouldDrop(row(.create, 9, "did:plc:a", "app.bsky.feed.post", "k2", "x")));
    try testing.expect(!snap.shouldDrop(row(.delete, 9, "did:plc:a", "app.bsky.feed.post", "k1", "")));
    try testing.expect(!snap.shouldDrop(row(.identity, 9, "did:plc:a", "", "", "x")));
}

test "did tombstones: sync and account-delete cover the whole did" {
    var snap = Snapshot.init(testing.allocator);
    defer snap.deinit();

    try snap.observe(row(.sync, 20, "did:plc:b", "", "", ""));
    try snap.observe(row(.account, 30, "did:plc:c", "", "", account_deleted));
    try snap.observe(row(.account, 40, "did:plc:d", "", "", account_active)); // not a delete
    const counts = snap.countsByReason();
    try testing.expectEqual(@as(u64, 1), counts.sync);
    try testing.expectEqual(@as(u64, 1), counts.account);

    try testing.expect(snap.shouldDrop(row(.create, 19, "did:plc:b", "app.bsky.feed.post", "k", "x")));
    try testing.expect(!snap.shouldDrop(row(.create, 20, "did:plc:b", "app.bsky.feed.post", "k", "x")));
    try testing.expect(snap.shouldDrop(row(.create, 29, "did:plc:c", "app.bsky.feed.post", "k", "x")));
    try testing.expect(!snap.shouldDrop(row(.create, 39, "did:plc:d", "app.bsky.feed.post", "k", "x")));
}

test "evict removes entries at or below the watermark" {
    var snap = Snapshot.init(testing.allocator);
    defer snap.deinit();
    try snap.observe(row(.delete, 5, "did:plc:a", "app.bsky.feed.post", "k1", ""));
    try snap.observe(row(.delete, 15, "did:plc:a", "app.bsky.feed.post", "k2", ""));
    try snap.observe(row(.sync, 10, "did:plc:b", "", "", ""));
    snap.evict(10);
    try testing.expectEqual(@as(usize, 1), snap.len());
    try testing.expect(snap.shouldDrop(row(.create, 14, "did:plc:a", "app.bsky.feed.post", "k2", "x")));
    try testing.expect(!snap.shouldDrop(row(.create, 4, "did:plc:a", "app.bsky.feed.post", "k1", "x")));
}

test "account tombstones reject malformed CBOR and metrics reflect final reason" {
    var snap = Snapshot.init(testing.allocator);
    defer snap.deinit();
    try testing.expectError(error.UnexpectedEof, snap.observe(row(.account, 1, "did:plc:a", "", "", "\xa1")));
    try snap.observe(row(.sync, 2, "did:plc:a", "", "", ""));
    try snap.observe(row(.account, 3, "did:plc:a", "", "", account_deleted));
    const counts = snap.countsByReason();
    try testing.expectEqual(@as(u64, 0), counts.sync);
    try testing.expectEqual(@as(u64, 1), counts.account);
    try testing.expectEqual(@as(usize, "did:plc:a".len + 64), snap.approxBytes());
}

test "approxBytes counter matches a full recomputation across observe and evict" {
    var snap = Snapshot.init(testing.allocator);
    defer snap.deinit();

    const recompute = struct {
        fn walk(s: *const Snapshot) usize {
            var result: usize = 0;
            var rit = s.records.iterator();
            while (rit.next()) |entry| result += entry.key_ptr.len - 2 + 64;
            var dit = s.dids.iterator();
            while (dit.next()) |entry| result += entry.key_ptr.len + 64;
            return result;
        }
    }.walk;

    try snap.observe(row(.delete, 10, "did:plc:aaa", "app.bsky.feed.post", "rkey1", ""));
    try snap.observe(row(.update, 12, "did:plc:aaa", "app.bsky.feed.post", "rkey2", ""));
    // duplicate key, higher seq: value updates, bytes must not change
    try snap.observe(row(.delete, 15, "did:plc:aaa", "app.bsky.feed.post", "rkey1", ""));
    try snap.observe(row(.sync, 20, "did:plc:bbb", "", "", ""));
    try snap.observe(row(.account, 25, "did:plc:ccc", "", "", account_deleted));
    // non-marker kinds are ignored
    try snap.observe(row(.create, 30, "did:plc:ddd", "app.bsky.feed.like", "rkey3", ""));
    try testing.expectEqual(recompute(&snap), snap.approxBytes());

    snap.evict(14); // removes rkey2 (12), keeps rkey1 (15), keeps dids (20, 25)
    try testing.expectEqual(recompute(&snap), snap.approxBytes());

    snap.evict(std.math.maxInt(u64)); // removes everything
    try testing.expectEqual(@as(usize, 0), snap.approxBytes());
    try testing.expectEqual(recompute(&snap), snap.approxBytes());
}