//! Topology-preserving sealed-segment display timestamp patching.
//!
//! This is the Zig equivalent of upstream `segment.Patch`: rows, blocks,
//! witness/sequence envelopes, blooms, and collection indexes are invariant.
//! Only dirty blocks are re-encoded, while clean block frames and the complete
//! bloom/collection footer tail are copied byte-for-byte. A zero-mutation pass
//! returns no replacement bytes and an in-place patch uses tmp + fsync + rename
//! + parent-directory fsync.

const std = @import("std");
const crashpoint = @import("../runtime/crashpoint.zig");
const gloom = @import("gloom.zig");
const segment = @import("segment.zig");
const writer = @import("segment_writer.zig");
const segment_io = @import("segment_io.zig");
const zstd = @import("zstd.zig");

const Allocator = std.mem.Allocator;
const Io = std.Io;

pub const Result = struct {
    patched: bool = false,
    rows_mutated: u64 = 0,
    blocks_touched: u32 = 0,
    new_checksum: u64 = 0,
};

const Guard = struct {
    seq: u64,
    witnessed_at: i64,
    kind: segment.Kind,
    did_len: usize,
    did_hash: u64,
    collection_len: usize,
    collection_hash: u64,
    rkey_len: usize,
    rkey_hash: u64,
    rev_len: usize,
    rev_hash: u64,
    payload_len: usize,
    payload_hash: u64,

    fn init(ev: segment.Event) Guard {
        return .{
            .seq = ev.seq,
            .witnessed_at = ev.witnessed_at,
            .kind = ev.kind,
            .did_len = ev.did.len,
            .did_hash = hash(ev.did),
            .collection_len = ev.collection.len,
            .collection_hash = hash(ev.collection),
            .rkey_len = ev.rkey.len,
            .rkey_hash = hash(ev.rkey),
            .rev_len = ev.rev.len,
            .rev_hash = hash(ev.rev),
            .payload_len = ev.payload.len,
            .payload_hash = hash(ev.payload),
        };
    }

    fn verify(self: Guard, ev: segment.Event) !void {
        if (ev.seq != self.seq) return error.PatchChangedImmutableField;
        if (ev.witnessed_at != self.witnessed_at) return error.PatchChangedImmutableField;
        if (ev.kind != self.kind) return error.PatchChangedImmutableField;
        if (ev.did.len != self.did_len or hash(ev.did) != self.did_hash) return error.PatchChangedImmutableField;
        if (ev.collection.len != self.collection_len or hash(ev.collection) != self.collection_hash) return error.PatchChangedImmutableField;
        if (ev.rkey.len != self.rkey_len or hash(ev.rkey) != self.rkey_hash) return error.PatchChangedImmutableField;
        if (ev.rev.len != self.rev_len or hash(ev.rev) != self.rev_hash) return error.PatchChangedImmutableField;
        if (ev.payload.len != self.payload_len or hash(ev.payload) != self.payload_hash) return error.PatchChangedImmutableField;
    }
};

fn hash(bytes: []const u8) u64 {
    return std.hash.XxHash3.hash(0, bytes);
}

/// Return replacement sealed bytes when at least one indexed_at value changes.
/// `mutate` may change only `indexed_at`; the post-callback guard checks every
/// row after every callback, including cross-row retained-pointer mutations.
pub fn patch(
    allocator: Allocator,
    old_bytes: []const u8,
    result: *Result,
    candidate_dids: []const []const u8,
    ctx: anytype,
    comptime mutate: fn (@TypeOf(ctx), *segment.Event) void,
) !?[]u8 {
    result.* = .{};
    var sealed = try segment.Sealed.parse(allocator, old_bytes);
    defer sealed.deinit(allocator);
    const header = sealed.header;

    const block_index_len = @as(usize, header.block_count) * segment.BlockIndexEntry.encoded_size;
    if (header.did_bloom_offset != header.footer_offset + block_index_len)
        return error.NonContiguousFooter;
    if (header.did_bloom_offset > header.block_did_bloom_offset or
        header.block_did_bloom_offset > header.collection_index_offset or
        header.collection_index_offset > old_bytes.len)
        return error.BadFooter;

    if (candidate_dids.len > 0) {
        var filter = try gloom.unmarshal(allocator, old_bytes[header.did_bloom_offset..header.block_did_bloom_offset]);
        defer filter.deinit(allocator);
        var possible = false;
        for (candidate_dids) |did| {
            if (did.len > 0 and filter.contains(did)) {
                possible = true;
                break;
            }
        }
        if (!possible) return null;
    }

    var out: std.ArrayList(u8) = .empty;
    defer out.deinit(allocator);
    try out.appendSlice(allocator, old_bytes[0..segment.header_size]);

    var infos = try allocator.alloc(segment.BlockIndexEntry, sealed.block_index.len);
    defer allocator.free(infos);

    for (sealed.block_index, 0..) |old_info, block_index| {
        var block = try sealed.readBlock(allocator, block_index);
        defer block.deinit(allocator);
        if (block.events.len != old_info.event_count) return error.BlockEventCountMismatch;

        const guards = try allocator.alloc(Guard, block.events.len);
        defer allocator.free(guards);
        const old_indexed = try allocator.alloc(i64, block.events.len);
        defer allocator.free(old_indexed);
        for (block.events, guards, old_indexed) |ev, *guard, *indexed| {
            guard.* = .init(ev);
            indexed.* = ev.indexed_at;
        }
        for (block.events) |*ev| mutate(ctx, ev);

        var dirty = false;
        for (block.events, guards, old_indexed) |ev, guard, indexed| {
            try guard.verify(ev);
            if (ev.indexed_at != indexed) {
                dirty = true;
                result.rows_mutated += 1;
            }
        }

        var info = old_info;
        info.offset = out.items.len;
        if (dirty) {
            const encoded = try writer.encodeBlock(allocator, block.events);
            defer allocator.free(encoded);
            if (encoded.len != old_info.uncompressed_size) return error.PatchChangedBlockSize;
            const frame = zstd.compress(allocator, encoded) catch |err| switch (err) {
                error.OutOfMemory => return error.OutOfMemory,
                else => return error.CompressFailed,
            };
            defer allocator.free(frame);
            var prefix: [8]u8 = undefined;
            std.mem.writeInt(u64, &prefix, frame.len, .little);
            try out.appendSlice(allocator, &prefix);
            try out.appendSlice(allocator, frame);
            info.compressed_size = @intCast(frame.len);
            result.blocks_touched += 1;
        } else {
            const frame_end = old_info.offset + 8 + old_info.compressed_size;
            try out.appendSlice(allocator, old_bytes[old_info.offset..frame_end]);
        }
        infos[block_index] = info;
    }

    if (result.rows_mutated == 0) return null;

    const new_footer_offset = out.items.len;
    for (infos) |info| try appendBlockIndex(allocator, &out, info);
    // DID bloom, per-block blooms, and collection index do not describe
    // indexed_at and therefore remain byte-identical.
    try out.appendSlice(allocator, old_bytes[header.did_bloom_offset..]);

    const did_bloom_len = header.block_did_bloom_offset - header.did_bloom_offset;
    const block_bloom_len = header.collection_index_offset - header.block_did_bloom_offset;
    const h = out.items[0..segment.header_size];
    std.mem.writeInt(u64, h[4..12], 0, .little);
    std.mem.writeInt(u64, h[58..66], new_footer_offset, .little);
    std.mem.writeInt(u64, h[66..74], new_footer_offset + block_index_len, .little);
    std.mem.writeInt(u64, h[74..82], new_footer_offset + block_index_len + did_bloom_len, .little);
    std.mem.writeInt(u64, h[82..90], new_footer_offset + block_index_len + did_bloom_len + block_bloom_len, .little);
    std.mem.writeInt(u64, h[90..98], new_footer_offset, .little);
    var hasher = std.hash.XxHash3.init(0);
    hasher.update(h[12..segment.header_size]);
    hasher.update(out.items[new_footer_offset..]);
    result.new_checksum = hasher.final();
    std.mem.writeInt(u64, h[4..12], result.new_checksum, .little);
    result.patched = true;
    return try out.toOwnedSlice(allocator);
}

fn appendBlockIndex(allocator: Allocator, out: *std.ArrayList(u8), info: segment.BlockIndexEntry) !void {
    var entry: [segment.BlockIndexEntry.encoded_size]u8 = undefined;
    std.mem.writeInt(u64, entry[0..8], info.offset, .little);
    std.mem.writeInt(u32, entry[8..12], info.compressed_size, .little);
    std.mem.writeInt(u32, entry[12..16], info.uncompressed_size, .little);
    std.mem.writeInt(u32, entry[16..20], info.event_count, .little);
    std.mem.writeInt(u64, entry[20..28], info.min_seq, .little);
    std.mem.writeInt(u64, entry[28..36], info.max_seq, .little);
    std.mem.writeInt(i64, entry[36..44], info.min_witnessed_at, .little);
    std.mem.writeInt(i64, entry[44..52], info.max_witnessed_at, .little);
    try out.appendSlice(allocator, &entry);
}

pub fn patchFile(
    allocator: Allocator,
    io: Io,
    dir: Io.Dir,
    name: []const u8,
    candidate_dids: []const []const u8,
    ctx: anytype,
    comptime mutate: fn (@TypeOf(ctx), *segment.Event) void,
) !Result {
    const old_bytes = try dir.readFileAlloc(io, name, allocator, .limited(1 << 31));
    defer allocator.free(old_bytes);
    var result: Result = .{};
    const new_bytes = (try patch(allocator, old_bytes, &result, candidate_dids, ctx, mutate)) orelse return result;
    defer allocator.free(new_bytes);

    var tmp_buf: [128]u8 = undefined;
    const tmp_name = try std.fmt.bufPrint(&tmp_buf, "{s}.tmp", .{name});
    dir.deleteFile(io, tmp_name) catch |err| if (err != error.FileNotFound) return err;
    var file = try dir.createFile(io, tmp_name, .{ .truncate = true });
    var file_open = true;
    errdefer {
        if (file_open) file.close(io);
        dir.deleteFile(io, tmp_name) catch {};
    }
    try segment_io.writeStreamingAll(&file, io, new_bytes);
    crashpoint.hit(.after_segment_patch_temp_written);
    try segment_io.sync(&file, io);
    crashpoint.hit(.after_segment_patch_temp_synced);
    file.close(io);
    file_open = false;
    try segment_io.rename(dir, tmp_name, dir, name, io);
    crashpoint.hit(.after_segment_patch_renamed);
    try segment_io.syncDir(dir, io);
    crashpoint.hit(.after_segment_patch_dir_synced);
    return result;
}

// === offline production-format tests ===

const testing = std.testing;

fn fixtureEvent(seq: u64, did: []const u8, collection: []const u8) segment.Event {
    return .{
        .seq = seq,
        .witnessed_at = @intCast(1000 + seq),
        .indexed_at = 0,
        .kind = if (collection.len == 0) .identity else .create,
        .did = did,
        .collection = collection,
        .rkey = if (collection.len == 0) "" else "r1",
        .rev = if (collection.len == 0) "" else "v1",
        .payload = if (collection.len == 0) "" else "\xa1\x61\x78\x01",
    };
}

fn fixture(allocator: Allocator) ![]u8 {
    var active = try writer.ActiveWriter.init(allocator);
    defer active.deinit();
    active.max_events_per_block = 2;
    try active.append(fixtureEvent(1, "did:plc:a", "app.bsky.feed.post"));
    try active.append(fixtureEvent(2, "did:plc:b", "app.bsky.feed.like"));
    try active.append(fixtureEvent(3, "did:plc:a", ""));
    try active.append(fixtureEvent(4, "did:plc:c", "app.bsky.feed.post"));
    try active.append(fixtureEvent(5, "did:plc:b", "app.bsky.feed.post"));
    try active.seal();
    return allocator.dupe(u8, active.bytes());
}

const Stamp = struct { did: []const u8, value: i64 };
fn stamp(ctx: Stamp, ev: *segment.Event) void {
    if (std.mem.eql(u8, ev.did, ctx.did)) ev.indexed_at = ctx.value;
}

test "patch changes only indexed_at and preserves clean frames plus footer tail" {
    const old = try fixture(testing.allocator);
    defer testing.allocator.free(old);
    var before = try segment.Sealed.parse(testing.allocator, old);
    defer before.deinit(testing.allocator);
    var result: Result = .{};
    const updated = (try patch(testing.allocator, old, &result, &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp)).?;
    defer testing.allocator.free(updated);
    try testing.expect(result.patched);
    try testing.expectEqual(@as(u64, 2), result.rows_mutated);
    try testing.expectEqual(@as(u32, 2), result.blocks_touched);

    var after = try segment.Sealed.parse(testing.allocator, updated);
    defer after.deinit(testing.allocator);
    try testing.expectEqual(before.header.block_count, after.header.block_count);
    try testing.expectEqual(before.header.event_count, after.header.event_count);
    try testing.expectEqual(before.header.min_seq, after.header.min_seq);
    try testing.expectEqual(before.header.max_seq, after.header.max_seq);
    try testing.expectEqual(before.header.min_witnessed_at, after.header.min_witnessed_at);
    try testing.expectEqual(before.header.max_witnessed_at, after.header.max_witnessed_at);
    try testing.expectEqualSlices(u8, old[before.header.did_bloom_offset..], updated[after.header.did_bloom_offset..]);

    // The final block has no did:plc:a and is copied exactly, including its
    // original zstd frame bytes.
    const old_last = before.block_index[2];
    const new_last = after.block_index[2];
    try testing.expectEqualSlices(
        u8,
        old[old_last.offset .. old_last.offset + 8 + old_last.compressed_size],
        updated[new_last.offset .. new_last.offset + 8 + new_last.compressed_size],
    );
    for (0..after.block_index.len) |bi| {
        var block = try after.readBlock(testing.allocator, bi);
        defer block.deinit(testing.allocator);
        for (block.events) |ev| {
            try testing.expectEqual(if (std.mem.eql(u8, ev.did, "did:plc:a")) @as(i64, 99) else 0, ev.indexed_at);
        }
    }
}

test "patch file exhaustively propagates segment I/O faults without silent corruption" {
    segment_io.disarm();
    defer segment_io.disarm();
    const old = try fixture(testing.allocator);
    defer testing.allocator.free(old);

    // Learn the deterministic committed output from the real implementation.
    var reference_tmp = testing.tmpDir(.{});
    defer reference_tmp.cleanup();
    {
        var file = try reference_tmp.dir.createFile(testing.io, "seg.jss", .{});
        defer file.close(testing.io);
        try file.writeStreamingAll(testing.io, old);
    }
    _ = try patchFile(testing.allocator, testing.io, reference_tmp.dir, "seg.jss", &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp);
    const committed = try reference_tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20));
    defer testing.allocator.free(committed);
    try testing.expect(!std.mem.eql(u8, old, committed));

    const cases = [_]struct { spec: []const u8, committed_after_error: bool }{
        .{ .spec = "write:1:enospc", .committed_after_error = false },
        .{ .spec = "write:1:shortwrite", .committed_after_error = false },
        .{ .spec = "sync:1:eio", .committed_after_error = false },
        .{ .spec = "rename:1:eio", .committed_after_error = false },
        .{ .spec = "sync:2:eio", .committed_after_error = true },
    };
    for (cases) |case| {
        var tmp = testing.tmpDir(.{});
        defer tmp.cleanup();
        {
            var file = try tmp.dir.createFile(testing.io, "seg.jss", .{});
            defer file.close(testing.io);
            try file.writeStreamingAll(testing.io, old);
        }
        try testing.expect(segment_io.armSpec(case.spec));
        try testing.expectError(switch (case.spec[0]) {
            'w' => if (std.mem.endsWith(u8, case.spec, "enospc")) error.NoSpaceLeft else error.ShortWrite,
            else => error.InputOutput,
        }, patchFile(testing.allocator, testing.io, tmp.dir, "seg.jss", &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp));
        segment_io.disarm();
        try testing.expectError(error.FileNotFound, tmp.dir.openFile(testing.io, "seg.jss.tmp", .{}));
        const got = try tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20));
        defer testing.allocator.free(got);
        try testing.expectEqualSlices(u8, if (case.committed_after_error) committed else old, got);
    }
}

test "patch is idempotent and candidate bloom skip never calls mutate" {
    const old = try fixture(testing.allocator);
    defer testing.allocator.free(old);
    var first: Result = .{};
    const updated = (try patch(testing.allocator, old, &first, &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp)).?;
    defer testing.allocator.free(updated);
    var second: Result = .{};
    try testing.expectEqual(@as(?[]u8, null), try patch(testing.allocator, updated, &second, &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp));
    try testing.expectEqual(@as(u64, 0), second.rows_mutated);

    const Never = struct {
        fn run(_: void, _: *segment.Event) void {
            @panic("candidate bloom false-negative");
        }
    };
    var skipped: Result = .{};
    try testing.expectEqual(@as(?[]u8, null), try patch(testing.allocator, old, &skipped, &.{"did:plc:not-present"}, {}, Never.run));
}

test "patch rejects immutable and cross-row mutation" {
    const old = try fixture(testing.allocator);
    defer testing.allocator.free(old);
    const Evil = struct {
        fn run(_: void, ev: *segment.Event) void {
            ev.seq += 1;
        }
    };
    var result: Result = .{};
    try testing.expectError(error.PatchChangedImmutableField, patch(testing.allocator, old, &result, &.{}, {}, Evil.run));

    const Cross = struct {
        first: ?*segment.Event = null,
        fn run(self: *@This(), ev: *segment.Event) void {
            if (self.first) |first| {
                first.seq += 1;
            } else self.first = ev;
        }
    };
    var cross: Cross = .{};
    result = .{};
    try testing.expectError(error.PatchChangedImmutableField, patch(testing.allocator, old, &result, &.{}, &cross, Cross.run));
}

test "patchFile atomically installs a checksummed segment" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var dir = try tmp.dir.createDirPathOpen(io, "segments", .{ .open_options = .{ .iterate = true } });
    defer dir.close(io);
    const old = try fixture(testing.allocator);
    defer testing.allocator.free(old);
    var file = try dir.createFile(io, "seg_0000000000.jss", .{});
    try file.writeStreamingAll(io, old);
    file.close(io);

    const result = try patchFile(testing.allocator, io, dir, "seg_0000000000.jss", &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp);
    try testing.expect(result.patched);
    const bytes = try dir.readFileAlloc(io, "seg_0000000000.jss", testing.allocator, .limited(1 << 20));
    defer testing.allocator.free(bytes);
    var sealed = try segment.Sealed.parse(testing.allocator, bytes);
    defer sealed.deinit(testing.allocator);
    try testing.expectEqual(result.new_checksum, sealed.header.checksum);
    try testing.expectError(error.FileNotFound, dir.access(io, "seg_0000000000.jss.tmp", .{}));
}

test "patch accepts upstream-produced sealed JSS and preserves its opaque indexes" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    const old = try Io.Dir.cwd().readFileAlloc(
        io,
        "tests/fixtures/seg_0000000000.jss",
        testing.allocator,
        .limited(4 << 20),
    );
    defer testing.allocator.free(old);
    var before = try segment.Sealed.parse(testing.allocator, old);
    defer before.deinit(testing.allocator);

    const One = struct {
        seq: u64,
        value: i64,
        fn run(self: @This(), ev: *segment.Event) void {
            if (ev.seq == self.seq) ev.indexed_at = self.value;
        }
    };
    var first_block = try before.readBlock(testing.allocator, 0);
    defer first_block.deinit(testing.allocator);
    const target = One{ .seq = first_block.events[0].seq, .value = first_block.events[0].witnessed_at - 1234 };
    var result: Result = .{};
    const updated = (try patch(testing.allocator, old, &result, &.{first_block.events[0].did}, target, One.run)).?;
    defer testing.allocator.free(updated);
    try testing.expectEqual(@as(u64, 1), result.rows_mutated);
    try testing.expectEqual(@as(u32, 1), result.blocks_touched);

    var after = try segment.Sealed.parse(testing.allocator, updated);
    defer after.deinit(testing.allocator);
    try testing.expectEqualSlices(u8, old[before.header.did_bloom_offset..], updated[after.header.did_bloom_offset..]);
    var patched_block = try after.readBlock(testing.allocator, 0);
    defer patched_block.deinit(testing.allocator);
    try testing.expectEqual(target.value, patched_block.events[0].indexed_at);
    try testing.expectEqual(target.seq, patched_block.events[0].seq);
}