//! Sync 1.1 verification — signatures, repository-chain state, and repair
//! authentication. Live commits are checked as signed MST diffs. Full-repo
//! repair heads use the separate unbounded complete-CAR path; the asynchronous
//! repair coordinator consumes the explicit `needs_resync` outcome.
//!
//! semantics (upstream-faithful: no event is served or archived
//! unverified when the DID resolves — stream's consumers inherit its
//! diligence, and verification blocks rather than passing through):
//! - cache miss → resolve synchronously; parallelism comes from the
//!   ingest pipeline's verify workers, not from skipping work
//! - resolution failure → reject and count. Until transparent resync is
//!   implemented, availability must not turn into unverified archive data.
//! - signature mismatch → evict key, re-resolve once (rotation), retry;
//!   still bad → the event is dropped by the caller
//! - #identity events evict the cached key (rotation signal)
//!
const std = @import("std");
const zat = @import("zat");
const lru = @import("../storage/lru.zig");
const metrics = @import("../runtime/metrics.zig");
const meta_store = @import("../storage/meta_store.zig");
const repos = @import("backfill/repos.zig");

const Io = std.Io;
const Allocator = std.mem.Allocator;
const log = std.log.scoped(.stream);
const max_commit_blocks_bytes = 2_000_000;
const max_firehose_message_bytes = 2 * 1024 * 1024;

const CachedKey = struct {
    key_type: zat.multicodec.KeyType,
    raw: [33]u8,
    len: u8,
};

pub const Outcome = enum {
    valid,
    /// event rev is at or behind the durable/provisional repo head
    replay,
    /// signature failed against a freshly resolved key — drop the event
    invalid_signature,
    /// could not resolve the DID — reject; authoritative repair cannot be
    /// authenticated without the same identity material
    unverified,
    /// input-integrity failure which an authoritative fetch cannot repair:
    /// future rev, size violation, or outer/inner field mismatch
    invalid_commit,
    /// Chain state cannot be advanced from this event alone. Upstream
    /// performs authoritative getRepo here; Stream's coordinator consumes
    /// this outcome and withholds the triggering event from the archive.
    needs_resync,
};

pub const ChainUpdate = struct {
    rev: [13]u8,
    data: [128]u8 = undefined,
    data_len: u8,

    pub fn dataSlice(self: *const ChainUpdate) []const u8 {
        return self.data[0..self.data_len];
    }
};

pub const Result = struct {
    outcome: Outcome = .valid,
    /// Exact signed post-state proven by this commit. The writer stages it
    /// against the event's local archive boundary for durable checkpointing.
    chain: ?ChainUpdate = null,
    hosting: ?HostingUpdate = null,
};

const CommitFailureClass = enum { signature, invalid, resync, fatal };

/// Match Atmos's default PolicyResync boundary. Failures in decoding,
/// inversion, duplicate paths, and op-CID proof are repairable by replacing
/// local state from authoritative getRepo. Envelope/inner disagreement is a
/// producer-integrity failure and bypasses repair; allocation failure must
/// remain fatal rather than being disguised as a network repair.
fn classifyCommitFailure(err: anyerror) CommitFailureClass {
    return switch (err) {
        error.SignatureVerificationFailed => .signature,
        error.FieldMismatch => .invalid,
        error.OutOfMemory => .fatal,
        else => .resync,
    };
}

pub const HostingUpdate = struct {
    active: bool,
    seq: i64,
    status: [32]u8 = undefined,
    status_len: u8 = 0,
    time: [64]u8 = undefined,
    time_len: u8 = 0,

    pub fn statusSlice(self: *const HostingUpdate) []const u8 {
        return self.status[0..self.status_len];
    }

    pub fn timeSlice(self: *const HostingUpdate) []const u8 {
        return self.time[0..self.time_len];
    }
};

pub fn stageChain(writes: *meta_store.Batch, did: []const u8, state: ChainUpdate) !void {
    var key_buf: [512]u8 = undefined;
    const key = try chainKey(&key_buf, did);
    var value: [1 + 13 + 1 + 128]u8 = undefined;
    value[0] = 1;
    @memcpy(value[1..14], &state.rev);
    value[14] = state.data_len;
    @memcpy(value[15..][0..state.data_len], state.dataSlice());
    writes.put(key, value[0 .. 15 + state.data_len]);
}

pub fn stageHosting(writes: *meta_store.Batch, did: []const u8, state: HostingUpdate) !void {
    var key_buf: [512]u8 = undefined;
    const key = try hostingKey(&key_buf, did);
    var value: [1 + 1 + 8 + 1 + 32 + 1 + 64]u8 = undefined;
    value[0] = 1;
    value[1] = @intFromBool(state.active);
    std.mem.writeInt(i64, value[2..10], state.seq, .little);
    value[10] = state.status_len;
    @memcpy(value[11..][0..state.status_len], state.statusSlice());
    const time_at = 11 + state.status_len;
    value[time_at] = state.time_len;
    @memcpy(value[time_at + 1 ..][0..state.time_len], state.timeSlice());
    writes.put(key, value[0 .. time_at + 1 + state.time_len]);
}

fn chainKey(buf: []u8, did: []const u8) ![]const u8 {
    return std.fmt.bufPrint(buf, "verify/chain/{s}", .{did});
}

fn hostingKey(buf: []u8, did: []const u8) ![]const u8 {
    return std.fmt.bufPrint(buf, "verify/host/{s}", .{did});
}

fn decodeChain(value: []const u8) !ChainUpdate {
    if (value.len < 15 or value[0] != 1) return error.InvalidVerifierState;
    const data_len: usize = value[14];
    if (data_len > 128 or value.len != 15 + data_len) return error.InvalidVerifierState;
    var state: ChainUpdate = .{ .rev = undefined, .data_len = @intCast(data_len) };
    @memcpy(&state.rev, value[1..14]);
    @memcpy(state.data[0..data_len], value[15..]);
    return state;
}

fn decodeHosting(value: []const u8) !HostingUpdate {
    if (value.len < 12 or value[0] != 1 or value[1] > 1) return error.InvalidVerifierState;
    const status_len: usize = value[10];
    if (status_len > 32 or 11 + status_len >= value.len) return error.InvalidVerifierState;
    const time_at = 11 + status_len;
    const time_len: usize = value[time_at];
    if (time_len > 64 or value.len != time_at + 1 + time_len) return error.InvalidVerifierState;
    var state: HostingUpdate = .{
        .active = value[1] == 1,
        .seq = std.mem.readInt(i64, value[2..10], .little),
        .status_len = @intCast(status_len),
        .time_len = @intCast(time_len),
    };
    @memcpy(state.status[0..status_len], value[11..time_at]);
    @memcpy(state.time[0..time_len], value[time_at + 1 ..]);
    return state;
}

pub const Verifier = struct {
    allocator: Allocator,
    io: Io,
    /// thread-safe LRU (internal mutex); resolution I/O runs on the
    /// CALLER's resolver — each verify worker owns one, so misses resolve
    /// in parallel (duplicate concurrent resolves of one DID are rare and
    /// harmless). LRU eviction replaces the old clear-when-full reset,
    /// which would have caused a resolve stampede at real-network scale.
    cache: lru.LruCache(CachedKey),
    stats: ?*metrics.Stats = null,
    meta: *meta_store.Store,
    chain_mu: Io.Mutex = .init,
    /// Provisional verified heads. Durability is staged by the ordered
    /// writer only after corresponding archive rows are assigned.
    chains: std.StringHashMapUnmanaged(ChainUpdate) = .empty,
    hosting: std.StringHashMapUnmanaged(HostingUpdate) = .empty,

    /// plc_url: base URL of the PLC directory. for the simulator this is its
    /// http address (it serves GET /did:...); production is plc.directory.
    plc_url: []const u8,

    pub fn init(allocator: Allocator, io: Io, plc_url: []const u8, meta: *meta_store.Store) Verifier {
        return .{
            .allocator = allocator,
            .io = io,
            .plc_url = plc_url,
            .meta = meta,
            .cache = lru.LruCache(CachedKey).init(allocator, 250_000, io),
        };
    }

    /// one per verify worker; owns a keep-alive connection
    pub fn makeResolver(self: *Verifier) zat.DidResolver {
        var r = zat.DidResolver.init(self.io, self.allocator);
        r.plc_url = self.plc_url;
        return r;
    }

    pub fn deinit(self: *Verifier) void {
        self.cache.deinit();
        var it = self.chains.iterator();
        while (it.next()) |entry| self.allocator.free(entry.key_ptr.*);
        self.chains.deinit(self.allocator);
        var hosting_it = self.hosting.iterator();
        while (hosting_it.next()) |entry| self.allocator.free(entry.key_ptr.*);
        self.hosting.deinit(self.allocator);
    }

    /// verify the commit signature in `blocks` (the event's CAR diff)
    /// against `did`'s signing key.
    pub fn verifyCommit(self: *Verifier, resolver: *zat.DidResolver, commit: zat.firehose.CommitEvent) !Result {
        // Upstream bypasses resync policy for these input-integrity gates.
        const rev = zat.Tid.parse(commit.rev) orelse return self.rejectInvalid();
        const now_us = Io.Timestamp.now(self.io, .real).toMicroseconds();
        if (rev.timestamp() > @as(u64, @intCast(@max(now_us, 0))) + 5 * 60 * std.time.us_per_s)
            return self.rejectInvalid();
        if (commit.blocks.len > max_commit_blocks_bytes or commit.ops.len > 200)
            return self.rejectInvalid();

        // key lookup/resolution under the lock; signature crypto outside it
        // so verify workers actually run in parallel
        const prior = try self.chainFor(commit.repo);
        var legacy = false;
        if (prior) |state| {
            if (std.mem.order(u8, commit.rev, &state.rev) != .gt)
                return .{ .outcome = .replay };
            if (commit.prev_data) |claimed| {
                if (!std.mem.eql(u8, claimed.raw, state.dataSlice()))
                    return self.needsResync();
            } else if (isLegacyCommit(commit)) {
                // Atmos defaults to LegacyAccept: Sync-1.0-shaped commits
                // cannot prove continuity, but still pass signature and
                // post-state op-CID verification and advance the chain.
                legacy = true;
            } else return self.needsResync();
        }

        const key = self.lockedGetKey(resolver, commit.repo, false) orelse {
            self.count("verify_unverified");
            return .{ .outcome = .unverified };
        };
        if (checkCommit(self.allocator, commit, key, if (legacy) null else prior)) |chain| {
            self.count("verify_valid");
            try self.rememberChain(commit.repo, chain);
            return .{ .chain = chain };
        } else |err| switch (classifyCommitFailure(err)) {
            .signature => {},
            .invalid => return self.rejectInvalid(),
            .resync => return self.needsResync(),
            .fatal => return err,
        }
        // key may have rotated: re-resolve once and retry
        const fresh = self.lockedGetKey(resolver, commit.repo, true) orelse {
            self.count("verify_unverified");
            return .{ .outcome = .unverified };
        };
        if (checkCommit(self.allocator, commit, fresh, if (legacy) null else prior)) |chain| {
            self.count("verify_valid");
            try self.rememberChain(commit.repo, chain);
            return .{ .chain = chain };
        } else |err| switch (classifyCommitFailure(err)) {
            .signature => {},
            .invalid => return self.rejectInvalid(),
            .resync => return self.needsResync(),
            .fatal => return err,
        }
        self.count("verify_invalid");
        return .{ .outcome = .invalid_signature };
    }

    pub fn verifySync(self: *Verifier, resolver: *zat.DidResolver, sync_event: zat.firehose.SyncEvent) !Result {
        // Atmos deliberately leaves an unparseable #sync rev to the ingest
        // gate. A divergent sync must first repair chain state from getRepo;
        // only then does the gate reject the entire original directive.
        if (zat.Tid.parse(sync_event.rev)) |rev| {
            const now_us = Io.Timestamp.now(self.io, .real).toMicroseconds();
            if (rev.timestamp() > @as(u64, @intCast(@max(now_us, 0))) + 5 * 60 * std.time.us_per_s)
                return self.rejectInvalid();
        }

        const prior = try self.chainFor(sync_event.did) orelse return .{ .outcome = .needs_resync };
        if (std.mem.order(u8, sync_event.rev, &prior.rev) != .gt)
            return .{ .outcome = .replay };
        if (sync_event.blocks.len == 0) return .{ .outcome = .needs_resync };

        var arena = std.heap.ArenaAllocator.init(self.allocator);
        defer arena.deinit();
        const loaded = zat.loadCommitFromCAR(arena.allocator(), sync_event.blocks) catch
            return .{ .outcome = .needs_resync };
        if (!std.mem.eql(u8, loaded.commit.data_cid, prior.dataSlice()))
            return .{ .outcome = .needs_resync };
        // Atmos checks envelope/inner identity only on its no-op fast path.
        // A divergent data root is repaired authoritatively first; the
        // downstream ingest gate then owns malformed envelope fields.
        if (!std.mem.eql(u8, loaded.commit.did, sync_event.did) or
            !std.mem.eql(u8, loaded.commit.rev, sync_event.rev))
            return self.rejectInvalid();

        const key = self.lockedGetKey(resolver, sync_event.did, false) orelse
            return .{ .outcome = .unverified };
        if (!checkSyncSignature(arena.allocator(), sync_event, key)) {
            const fresh = self.lockedGetKey(resolver, sync_event.did, true) orelse
                return .{ .outcome = .unverified };
            if (!checkSyncSignature(arena.allocator(), sync_event, fresh)) {
                self.count("verify_invalid");
                return .{ .outcome = .invalid_signature };
            }
        }

        var update = prior;
        @memcpy(&update.rev, sync_event.rev);
        try self.rememberChain(sync_event.did, update);
        self.count("verify_valid");
        return .{ .chain = update };
    }

    /// Verify the authoritative full-repo commit returned by getRepo. This is
    /// deliberately separate from bootstrap: ordinary network backfill
    /// mirrors upstream and does not seed verifier state, while a transparent
    /// repair MUST authenticate the fetched head before it can replace the
    /// chain. The returned update is provisional only; the repair coordinator
    /// applies it under the per-DID FSM and the archive writer promotes it at
    /// the matching durable row boundary.
    pub fn verifyFetchedRepo(
        self: *Verifier,
        resolver: *zat.DidResolver,
        did: []const u8,
        prepared: *const repos.PreparedRepo,
    ) !Result {
        const key = self.lockedGetKey(resolver, did, false) orelse {
            self.count("verify_unverified");
            return .{ .outcome = .unverified };
        };
        const update = checkFetchedRepo(did, prepared, key) catch |err| switch (err) {
            error.SignatureVerificationFailed => retry: {
                const fresh = self.lockedGetKey(resolver, did, true) orelse {
                    self.count("verify_unverified");
                    return .{ .outcome = .unverified };
                };
                break :retry checkFetchedRepo(did, prepared, fresh) catch |fresh_err| {
                    if (fresh_err == error.SignatureVerificationFailed) {
                        self.count("verify_invalid");
                        return .{ .outcome = .invalid_signature };
                    }
                    return self.rejectInvalid();
                };
            },
            else => return self.rejectInvalid(),
        };
        self.count("verify_valid");
        return .{ .chain = update };
    }

    pub fn onAccount(self: *Verifier, account: zat.firehose.AccountEvent) !Result {
        if (zat.Did.parse(account.did) == null) return self.rejectInvalid();
        if (try self.hostingFor(account.did)) |prior| {
            if (account.seq <= prior.seq) return .{ .outcome = .replay };
        }
        const status = if (account.status) |value| @tagName(value) else "";
        if (status.len > 32 or account.time.len > 64) return self.rejectInvalid();
        var update: HostingUpdate = .{
            .active = account.active,
            .seq = account.seq,
            .status_len = @intCast(status.len),
            .time_len = @intCast(account.time.len),
        };
        @memcpy(update.status[0..status.len], status);
        @memcpy(update.time[0..account.time.len], account.time);
        try self.rememberHosting(account.did, update);
        return .{ .hosting = update };
    }

    fn checkSyncSignature(allocator: Allocator, sync_event: zat.firehose.SyncEvent, key: CachedKey) bool {
        const public_key = zat.multicodec.PublicKey{
            .key_type = key.key_type,
            .raw = key.raw[0..key.len],
        };
        const result = zat.verifyCommitCar(allocator, sync_event.blocks, public_key, .{
            .verify_mst = false,
            .expected_did = sync_event.did,
            .max_car_size = max_firehose_message_bytes,
        }) catch return false;
        return std.mem.eql(u8, result.commit_rev, sync_event.rev);
    }

    fn rejectInvalid(self: *Verifier) Result {
        self.count("verify_invalid");
        return .{ .outcome = .invalid_commit };
    }

    fn needsResync(self: *Verifier) Result {
        // This is a verification failure, even though the coordinator can
        // repair it asynchronously. Keep the failure visible to operators.
        self.count("verify_invalid");
        return .{ .outcome = .needs_resync };
    }

    fn lockedGetKey(self: *Verifier, resolver: *zat.DidResolver, did: []const u8, force_resolve: bool) ?CachedKey {
        if (!force_resolve) {
            if (self.cache.get(did)) |k| {
                self.count("verify_cache_hits");
                return k;
            }
            self.count("verify_cache_misses");
        } else {
            self.evict(did);
        }
        return self.resolveAndCache(resolver, did);
    }

    /// resolve on the caller's own resolver, outside the cache lock
    fn resolveAndCache(self: *Verifier, resolver: *zat.DidResolver, did: []const u8) ?CachedKey {
        // did:web fetches attacker-named hosts, but zat's resolver routes
        // those through network_safety (private/loopback rejection, DoH
        // preflight, no redirects, checked-connection dial) — lesson #10
        // is handled at the SDK layer
        const doc_or_null: ?zat.DidDocument = doc: {
            const parsed = zat.Did.parse(did) orelse break :doc null;
            break :doc resolver.resolve(parsed) catch |err| {
                log.debug("DID resolve failed for {s}: {s}", .{ did, @errorName(err) });
                break :doc null;
            };
        };
        var doc = doc_or_null orelse return null;
        defer doc.deinit();

        const vm = doc.signingKey() orelse return null;
        const key_bytes = zat.multibase.decode(self.allocator, vm.public_key_multibase) catch return null;
        defer self.allocator.free(key_bytes);
        const public_key = zat.multicodec.parsePublicKey(key_bytes) catch return null;
        if (public_key.raw.len > 33) return null;

        var cached = CachedKey{
            .key_type = public_key.key_type,
            .raw = undefined,
            .len = @intCast(public_key.raw.len),
        };
        @memcpy(cached.raw[0..public_key.raw.len], public_key.raw);

        self.cache.put(did, cached) catch {};
        return cached;
    }

    fn count(self: *Verifier, comptime field: []const u8) void {
        if (self.stats) |st| _ = @field(st, field).fetchAdd(1, .monotonic);
    }

    /// rotation signal: #identity events invalidate the cached key
    pub fn evict(self: *Verifier, did: []const u8) void {
        _ = self.cache.remove(did);
    }

    fn checkCommit(allocator: Allocator, commit: zat.firehose.CommitEvent, key: CachedKey, prior: ?ChainUpdate) !ChainUpdate {
        const public_key = zat.multicodec.PublicKey{
            .key_type = key.key_type,
            .raw = key.raw[0..key.len],
        };
        var arena = std.heap.ArenaAllocator.init(allocator);
        defer arena.deinit();
        const ops = try commit.toMstOperations(arena.allocator());
        const result = zat.verifyCommitDiff(
            arena.allocator(),
            commit.blocks,
            ops,
            if (prior) |*state| state.dataSlice() else null,
            public_key,
            .{
                .expected_did = null,
                .max_car_size = max_commit_blocks_bytes,
            },
        ) catch |err| switch (err) {
            // Match upstream's default LenientInversion carve-out narrowly:
            // the envelope already proved prevData == our stored head, and
            // inversion completed but produced a different root. Signature,
            // outer/inner fields, and post-state op CIDs remain mandatory.
            error.PrevDataMismatch => if (prior != null)
                try zat.verifyCommitDiff(
                    arena.allocator(),
                    commit.blocks,
                    ops,
                    prior.?.dataSlice(),
                    public_key,
                    .{
                        .expected_did = null,
                        .skip_inversion = true,
                        .max_car_size = max_commit_blocks_bytes,
                    },
                )
            else
                return err,
            else => return err,
        };
        // The signed inner commit and relay envelope must identify the same
        // revision and CAR root. A valid signature over relabeled outer
        // fields is still an invalid firehose event.
        if (!std.mem.eql(u8, result.commit_did, commit.repo) or
            !std.mem.eql(u8, result.commit_rev, commit.rev)) return error.FieldMismatch;
        const outer_cid = commit.commit orelse return error.FieldMismatch;
        if (!std.mem.eql(u8, result.commit_cid, outer_cid.raw)) return error.FieldMismatch;
        try checkOpCids(arena.allocator(), commit, result.data_cid);
        if (result.data_cid.len > 128) return error.FieldMismatch;
        var update: ChainUpdate = .{
            .rev = undefined,
            .data_len = @intCast(result.data_cid.len),
        };
        @memcpy(&update.rev, result.commit_rev);
        @memcpy(update.data[0..result.data_cid.len], result.data_cid);
        return update;
    }

    fn checkFetchedRepo(
        did: []const u8,
        prepared: *const repos.PreparedRepo,
        key: CachedKey,
    ) !ChainUpdate {
        const public_key = zat.multicodec.PublicKey{
            .key_type = key.key_type,
            .raw = key.raw[0..key.len],
        };
        // prepareRepo already indexed the CAR and proved MST/record
        // completeness. Authenticate those retained commit bytes directly;
        // reparsing and re-walking multi-gigabyte repositories here would be
        // both semantically redundant and operationally unacceptable.
        try prepared.authenticate(did, public_key);
        if (prepared.rev.len != 13 or zat.Tid.parse(prepared.rev) == null) return error.InvalidCommit;
        if (prepared.data_cid.len > 128) return error.InvalidCommit;
        var update: ChainUpdate = .{
            .rev = undefined,
            .data_len = @intCast(prepared.data_cid.len),
        };
        @memcpy(&update.rev, prepared.rev);
        @memcpy(update.data[0..prepared.data_cid.len], prepared.data_cid);
        return update;
    }

    fn checkOpCids(allocator: Allocator, commit: zat.firehose.CommitEvent, data_cid: []const u8) !void {
        if (commit.ops.len == 0) return;
        const loaded = try zat.loadCommitFromCAR(allocator, commit.blocks);
        var tree = zat.mst.Mst.loadFromBlocks(allocator, loaded.repo_car, data_cid) catch
            return error.OpCidMismatch;
        for (commit.ops) |op| {
            const path = if (op.path.len > 0)
                op.path
            else
                try std.fmt.allocPrint(allocator, "{s}/{s}", .{ op.collection, op.rkey });
            const found = tree.getLazy(path) catch return error.OpCidMismatch;
            switch (op.action) {
                .create, .update => {
                    const claimed = op.cid orelse return error.OpCidMismatch;
                    const actual = found orelse return error.OpCidMismatch;
                    if (!std.mem.eql(u8, claimed.raw, actual.raw)) return error.OpCidMismatch;
                },
                .delete => if (found != null) return error.OpCidMismatch,
            }
        }
    }

    pub fn chainFor(self: *Verifier, did: []const u8) !?ChainUpdate {
        self.chain_mu.lockUncancelable(self.io);
        if (self.chains.get(did)) |state| {
            self.chain_mu.unlock(self.io);
            return state;
        }
        self.chain_mu.unlock(self.io);

        var key_buf: [512]u8 = undefined;
        const key = try chainKey(&key_buf, did);
        const encoded = try self.meta.getAlloc(self.allocator, key) orelse return null;
        defer self.allocator.free(encoded);
        const state = try decodeChain(encoded);
        try self.rememberChain(did, state);
        return state;
    }

    pub fn rememberChain(self: *Verifier, did: []const u8, state: ChainUpdate) !void {
        self.chain_mu.lockUncancelable(self.io);
        defer self.chain_mu.unlock(self.io);
        if (self.chains.getPtr(did)) |existing| {
            existing.* = state;
            return;
        }
        const owned = try self.allocator.dupe(u8, did);
        errdefer self.allocator.free(owned);
        try self.chains.put(self.allocator, owned, state);
    }

    pub fn hostingFor(self: *Verifier, did: []const u8) !?HostingUpdate {
        self.chain_mu.lockUncancelable(self.io);
        if (self.hosting.get(did)) |state| {
            self.chain_mu.unlock(self.io);
            return state;
        }
        self.chain_mu.unlock(self.io);
        var key_buf: [512]u8 = undefined;
        const key = try hostingKey(&key_buf, did);
        const encoded = try self.meta.getAlloc(self.allocator, key) orelse return null;
        defer self.allocator.free(encoded);
        const state = try decodeHosting(encoded);
        try self.rememberHosting(did, state);
        return state;
    }

    fn rememberHosting(self: *Verifier, did: []const u8, state: HostingUpdate) !void {
        self.chain_mu.lockUncancelable(self.io);
        defer self.chain_mu.unlock(self.io);
        if (self.hosting.getPtr(did)) |existing| {
            existing.* = state;
            return;
        }
        const owned = try self.allocator.dupe(u8, did);
        errdefer self.allocator.free(owned);
        try self.hosting.put(self.allocator, owned, state);
    }
};

fn isLegacyCommit(commit: zat.firehose.CommitEvent) bool {
    if (commit.prev_data != null) return false;
    for (commit.ops) |op| switch (op.action) {
        .update, .delete => if (op.prev != null) return false,
        .create => {},
    };
    return true;
}

test "durable verifier chain reloads from metadata store" {
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const path_len = try tmp.dir.realPath(threaded.io(), &path_buf);

    var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]);
    defer meta.deinit();
    var state: ChainUpdate = .{ .rev = undefined, .data_len = 4 };
    @memcpy(&state.rev, "3k2abd0000000");
    @memcpy(state.data[0..4], "data");
    var writes = meta.batch();
    stageChain(&writes, "did:plc:test", state) catch unreachable;
    var hosting: HostingUpdate = .{ .active = false, .seq = 42, .status_len = 8, .time_len = 4 };
    @memcpy(hosting.status[0..8], "deactive");
    @memcpy(hosting.time[0..4], "time");
    try stageHosting(&writes, "did:plc:test", hosting);
    try writes.commitDurable();
    writes.deinit();

    var verifier = Verifier.init(testing.allocator, threaded.io(), "http://localhost", &meta);
    defer verifier.deinit();
    const loaded = (try verifier.chainFor("did:plc:test")).?;
    try testing.expectEqualSlices(u8, &state.rev, &loaded.rev);
    try testing.expectEqualSlices(u8, state.dataSlice(), loaded.dataSlice());
    const loaded_hosting = (try verifier.hostingFor("did:plc:test")).?;
    try testing.expect(!loaded_hosting.active);
    try testing.expectEqual(@as(i64, 42), loaded_hosting.seq);
    try testing.expectEqualStrings("deactive", loaded_hosting.statusSlice());
    try testing.expectEqualStrings("time", loaded_hosting.timeSlice());
}

test "op CID proof rejects an envelope that disagrees with post-state MST" {
    const testing = std.testing;
    var arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    const record_cid = try zat.cbor.Cid.forDagCbor(allocator, "record");
    const wrong_cid = try zat.cbor.Cid.forDagCbor(allocator, "wrong");
    var tree = zat.mst.Mst.init(allocator);
    try tree.put("app.bsky.feed.post/3k2abcdefghij", record_cid);
    const data_cid = try tree.rootCid();
    const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(7));
    const did = try keypair.did(allocator);
    const signed = try zat.signCommit(allocator, .{
        .did = did,
        .rev = "3k2abcdefghij",
        .data = data_cid,
    }, &keypair);
    var blocks: std.ArrayList(zat.car.Block) = .empty;
    try blocks.append(allocator, .{ .cid_raw = signed.cid.raw, .data = signed.bytes });
    try tree.collectBlocks(&blocks);
    const car_bytes = try zat.car.writeAlloc(allocator, .{
        .roots = &.{signed.cid},
        .blocks = blocks.items,
    });
    const base: zat.firehose.CommitEvent = .{
        .seq = 1,
        .repo = did,
        .rev = "3k2abcdefghij",
        .time = "2026-07-15T00:00:00Z",
        .commit = signed.cid,
        .blocks = car_bytes,
        .ops = &.{.{
            .action = .create,
            .path = "app.bsky.feed.post/3k2abcdefghij",
            .collection = "app.bsky.feed.post",
            .rkey = "3k2abcdefghij",
            .cid = record_cid,
        }},
    };
    try Verifier.checkOpCids(allocator, base, data_cid.raw);
    var tampered = base;
    tampered.ops = &.{.{
        .action = .create,
        .path = "app.bsky.feed.post/3k2abcdefghij",
        .collection = "app.bsky.feed.post",
        .rkey = "3k2abcdefghij",
        .cid = wrong_cid,
    }};
    try testing.expectError(error.OpCidMismatch, Verifier.checkOpCids(allocator, tampered, data_cid.raw));
    try testing.expectEqual(CommitFailureClass.resync, classifyCommitFailure(error.OpCidMismatch));
}

test "commit block limit is upstream decimal two million bytes" {
    const testing = std.testing;
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const path_len = try tmp.dir.realPath(threaded.io(), &path_buf);
    var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]);
    defer meta.deinit();
    var verifier = Verifier.init(testing.allocator, threaded.io(), "http://offline.invalid", &meta);
    defer verifier.deinit();
    const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(29));
    const public_key = try keypair.publicKey();
    var cached: CachedKey = .{ .key_type = .p256, .raw = undefined, .len = @intCast(public_key.len) };
    @memcpy(cached.raw[0..public_key.len], &public_key);
    try verifier.cache.put("did:plc:size-boundary", cached);
    const blocks = try testing.allocator.alloc(u8, max_commit_blocks_bytes + 1);
    defer testing.allocator.free(blocks);
    @memset(blocks, 0);
    const base: zat.firehose.CommitEvent = .{
        .seq = 1,
        .repo = "did:plc:size-boundary",
        .rev = "3k2abcdefghij",
        .time = "2026-07-16T00:00:00Z",
        .blocks = blocks[0..max_commit_blocks_bytes],
        .ops = &.{},
    };
    const at_limit = try verifier.verifyCommit(undefined, base);
    try testing.expectEqual(Outcome.needs_resync, at_limit.outcome);
    var over = base;
    over.blocks = blocks;
    const rejected = try verifier.verifyCommit(undefined, over);
    try testing.expectEqual(Outcome.invalid_commit, rejected.outcome);
}

test "signed chain failures follow upstream repair policy and legacy default" {
    const testing = std.testing;
    var arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    var tree = zat.mst.Mst.init(allocator);
    const data_cid = try tree.rootCid();
    const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(11));
    const did = try keypair.did(allocator);
    const signed = try zat.signCommit(allocator, .{
        .did = did,
        .rev = "3k2abcdefghij",
        .data = data_cid,
    }, &keypair);
    var blocks: std.ArrayList(zat.car.Block) = .empty;
    try blocks.append(allocator, .{ .cid_raw = signed.cid.raw, .data = signed.bytes });
    try tree.collectBlocks(&blocks);
    const car_bytes = try zat.car.writeAlloc(allocator, .{
        .roots = &.{signed.cid},
        .blocks = blocks.items,
    });
    const public_key = try keypair.publicKey();
    var cached: CachedKey = .{
        .key_type = .p256,
        .raw = undefined,
        .len = @intCast(public_key.len),
    };
    @memcpy(cached.raw[0..public_key.len], &public_key);

    const event: zat.firehose.CommitEvent = .{
        .seq = 1,
        .repo = did,
        .rev = "3k2abcdefghij",
        .time = "2026-07-15T00:00:00Z",
        .commit = signed.cid,
        .blocks = car_bytes,
        .ops = &.{},
    };
    var wrong_prior: ChainUpdate = .{ .rev = undefined, .data_len = 4 };
    @memcpy(&wrong_prior.rev, "3k2abc0000000");
    @memcpy(wrong_prior.data[0..4], "nope");

    // The real signed CAR cannot invert to the deliberately wrong prior. The
    // verifier's production caller permits this exact case only after the
    // envelope's prevData has matched the stored head (Atmos's default
    // lenient-inversion carve-out), so it remains authenticated and accepted.
    const lenient = try Verifier.checkCommit(allocator, event, cached, wrong_prior);
    try testing.expectEqualSlices(u8, data_cid.raw, lenient.dataSlice());
    // Without that caller-side prevData proof, the same mismatch is the
    // recoverable failure class consumed by the repair coordinator.
    try testing.expectEqual(CommitFailureClass.resync, classifyCommitFailure(error.PrevDataMismatch));

    // With no prevData, Atmos's default LegacyAccept still authenticates the
    // exact same signed CAR and its post-state before advancing.
    const legacy = try Verifier.checkCommit(allocator, event, cached, null);
    try testing.expectEqualStrings(event.rev, &legacy.rev);
    try testing.expectEqualSlices(u8, data_cid.raw, legacy.dataSlice());
    try testing.expectEqual(CommitFailureClass.invalid, classifyCommitFailure(error.FieldMismatch));

    // Exercise the public verifier boundary too: a syntactically valid event
    // whose prevData disagrees with the stored chain is queued for repair
    // before resolution or signature work, never mislabeled permanently bad.
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const path_len = try tmp.dir.realPath(threaded.io(), &path_buf);
    var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]);
    defer meta.deinit();
    var verifier = Verifier.init(testing.allocator, threaded.io(), "http://offline.invalid", &meta);
    defer verifier.deinit();
    var current: ChainUpdate = .{ .rev = undefined, .data_len = @intCast(data_cid.raw.len) };
    @memcpy(&current.rev, "3k2abc0000000");
    @memcpy(current.data[0..data_cid.raw.len], data_cid.raw);
    try verifier.rememberChain(did, current);
    const wrong_prev = try zat.cbor.Cid.forDagCbor(allocator, "wrong prior");
    var broken = event;
    broken.prev_data = wrong_prev;
    const routed = try verifier.verifyCommit(undefined, broken);
    try testing.expectEqual(Outcome.needs_resync, routed.outcome);
}

test "full-repo repair authenticates the signed complete repository head" {
    const testing = std.testing;
    var arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    const record = "signed repair record";
    const record_cid = try zat.cbor.Cid.forDagCbor(allocator, record);
    var tree = zat.mst.Mst.init(allocator);
    try tree.put("app.bsky.feed.post/3k2abcdefghij", record_cid);
    const data_cid = try tree.rootCid();
    const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(9));
    const did = try keypair.did(allocator);
    const signed = try zat.signCommit(allocator, .{
        .did = did,
        .rev = "3k2abcdefghij",
        .data = data_cid,
    }, &keypair);
    var blocks: std.ArrayList(zat.car.Block) = .empty;
    try blocks.append(allocator, .{ .cid_raw = signed.cid.raw, .data = signed.bytes });
    try tree.collectBlocks(&blocks);
    const incomplete_car = try zat.car.writeAlloc(allocator, .{
        .roots = &.{signed.cid},
        .blocks = blocks.items,
    });
    try blocks.append(allocator, .{ .cid_raw = record_cid.raw, .data = record });
    const car_bytes = try zat.car.writeAlloc(allocator, .{
        .roots = &.{signed.cid},
        .blocks = blocks.items,
    });
    const public_key = try keypair.publicKey();
    var cached: CachedKey = .{
        .key_type = .p256,
        .raw = undefined,
        .len = @intCast(public_key.len),
    };
    @memcpy(cached.raw[0..public_key.len], &public_key);

    var prepared = try repos.prepareRepo(allocator, car_bytes);
    const update = try Verifier.checkFetchedRepo(did, &prepared, cached);
    try testing.expectEqualStrings("3k2abcdefghij", &update.rev);
    try testing.expectEqualSlices(u8, data_cid.raw, update.dataSlice());

    try testing.expectError(
        error.MissingBlock,
        repos.prepareRepo(allocator, incomplete_car),
    );

    const wrong_keypair = try zat.Keypair.fromSecretKey(.p256, @splat(7));
    const wrong_public_key = try wrong_keypair.publicKey();
    var wrong: CachedKey = .{ .key_type = .p256, .raw = undefined, .len = @intCast(wrong_public_key.len) };
    @memcpy(wrong.raw[0..wrong_public_key.len], &wrong_public_key);
    try testing.expectError(error.SignatureVerificationFailed, Verifier.checkFetchedRepo(did, &prepared, wrong));
    try testing.expectError(error.InvalidCommit, Verifier.checkFetchedRepo("did:plc:wrongrepo", &prepared, cached));
}