//! firehose event → jetstream v1 wire frames
//!
//! pure conversion, no I/O. mirrors upstream's validation gate
//! (docs/upstream-harness.md): bad upstream input is dropped and counted,
//! never fatal. a #commit fans out to one frame per surviving op; #sync and
//! #info produce nothing on the v1 wire.

const std = @import("std");
const zat = @import("zat");
const wire = @import("../serve/wire.zig");

const firehose = zat.firehose;
const Allocator = std.mem.Allocator;

pub const DropReason = enum {
    invalid_rev,
    invalid_collection,
    invalid_rkey,
    field_too_long,
    missing_block,
    /// A repository may hold a record the subscriber encoders cannot render
    /// as AT Protocol JSON (a CBOR float, say). Upstream memoizes the encode
    /// error on the entry and each subscriber skips it; the row stays durable
    /// and ingest is unaffected. Resource exhaustion is NOT this: OutOfMemory
    /// stays fatal so a full disk or heap can never look like a bad record.
    unencodable_record,
};

pub const Drops = std.enums.EnumArray(DropReason, u64);

pub const Frame = struct {
    kind: wire.Kind,
    did: []const u8, // borrows from the decoded event
    collection: []const u8, // empty for identity/account
    time_us: i64,
    seq: u64, // local archive seq (0 = unarchived)
    json: []u8, // v1 wire; empty for kinds v1 never emits (#sync)
    json_v2: []u8, // v2 superset wire
};

/// convert one decoded firehose event into wire frames (both wires),
/// appended to `out`. Output and intermediate buffers live in `arena`;
/// `did`/`collection` borrow from `event`.
pub fn convert(
    arena: *std.heap.ArenaAllocator,
    event: firehose.Event,
    time_us: i64,
    first_seq: u64,
    out: *std.ArrayList(Frame),
    drops: *Drops,
) !void {
    return convertStamped(arena, event, time_us, first_seq, null, out, drops);
}

/// Convert using the exact rows already accepted and stamped by the archive.
/// This keeps immediate live delivery byte-for-byte aligned with cold replay
/// after an operator timestamp import.
pub fn convertStamped(
    arena: *std.heap.ArenaAllocator,
    event: firehose.Event,
    time_us: i64,
    first_seq: u64,
    stamped_rows: ?[]const @import("../storage/segment.zig").Event,
    out: *std.ArrayList(Frame),
    drops: *Drops,
) !void {
    const allocator = arena.allocator();
    const out_start = out.items.len;
    switch (event) {
        .commit => |c| try convertCommit(allocator, c, time_us, first_seq, stamped_rows, out_start, out, drops),
        .identity => |i| {
            const display_time = try rowTime(stamped_rows, 0, time_us);
            var buf: std.Io.Writer.Allocating = .init(allocator);
            errdefer buf.deinit();
            try wire.encodeIdentity(&buf.writer, .{
                .did = i.did,
                .time_us = display_time,
                .seq = i.seq,
                .time = i.time,
                .handle = i.handle,
            });
            var buf2: std.Io.Writer.Allocating = .init(allocator);
            errdefer buf2.deinit();
            try wire.encodeIdentityV2(&buf2.writer, .{
                .did = i.did,
                .time_us = display_time,
                .seq = i.seq,
                .time = i.time,
                .handle = i.handle,
            }, first_seq);
            try out.append(allocator, .{
                .kind = .identity,
                .did = i.did,
                .collection = "",
                .time_us = display_time,
                .seq = first_seq,
                .json = try buf.toOwnedSlice(),
                .json_v2 = try buf2.toOwnedSlice(),
            });
        },
        .account => |a| {
            const display_time = try rowTime(stamped_rows, 0, time_us);
            const status: ?[]const u8 = if (a.status) |st| @tagName(st) else null;
            var buf: std.Io.Writer.Allocating = .init(allocator);
            errdefer buf.deinit();
            try wire.encodeAccount(&buf.writer, .{
                .did = a.did,
                .time_us = display_time,
                .seq = a.seq,
                .time = a.time,
                .active = a.active,
                .status = status,
            });
            var buf2: std.Io.Writer.Allocating = .init(allocator);
            errdefer buf2.deinit();
            try wire.encodeAccountV2(&buf2.writer, .{
                .did = a.did,
                .time_us = display_time,
                .seq = a.seq,
                .time = a.time,
                .active = a.active,
                .status = status,
            }, first_seq);
            try out.append(allocator, .{
                .kind = .account,
                .did = a.did,
                .collection = "",
                .time_us = display_time,
                .seq = first_seq,
                .json = try buf.toOwnedSlice(),
                .json_v2 = try buf2.toOwnedSlice(),
            });
        },
        // #sync: v2-only (v1 wire skips it for parity)
        .sync => |sy| {
            if (zat.Tid.parse(sy.rev) == null) {
                drops.getPtr(.invalid_rev).* += 1;
                return;
            }
            const display_time = try rowTime(stamped_rows, 0, time_us);
            var buf2: std.Io.Writer.Allocating = .init(allocator);
            errdefer buf2.deinit();
            try wire.encodeSyncV2(&buf2.writer, .{
                .did = sy.did,
                .time_us = display_time,
                .upstream_seq = sy.seq,
                .rev = sy.rev,
                .time = sy.time,
                .blocks = sy.blocks,
            }, first_seq);
            try out.append(allocator, .{
                .kind = .sync,
                .did = sy.did,
                .collection = "",
                .time_us = display_time,
                .seq = first_seq,
                .json = try allocator.alloc(u8, 0),
                .json_v2 = try buf2.toOwnedSlice(),
            });
        },
        .info => {},
    }
    if (stamped_rows) |rows| if (rows.len != out.items.len - out_start) return error.StampedRowMismatch;
}

fn convertCommit(
    allocator: Allocator,
    c: firehose.CommitEvent,
    time_us: i64,
    first_seq: u64,
    stamped_rows: ?[]const @import("../storage/segment.zig").Event,
    out_start: usize,
    out: *std.ArrayList(Frame),
    drops: *Drops,
) !void {
    // non-TID rev drops the whole event: rev ordering is load-bearing upstream
    // and the input can never become valid
    if (zat.Tid.parse(c.rev) == null) {
        drops.getPtr(.invalid_rev).* += 1;
        return;
    }

    for (c.ops) |op| {
        if (zat.Nsid.parse(op.collection) == null) {
            drops.getPtr(.invalid_collection).* += 1;
            continue;
        }
        if (zat.Rkey.parse(op.rkey) == null) {
            drops.getPtr(.invalid_rkey).* += 1;
            continue;
        }
        const is_delete = op.action == .delete;
        // create/update need the record block from the CAR diff; partial CARs
        // are spec-legal, so a missing block drops only this op
        if (!is_delete and (op.record == null or op.cid == null)) {
            drops.getPtr(.missing_block).* += 1;
            continue;
        }

        if (!representableFields(c.repo, op.collection, op.rkey, c.rev, 0)) {
            drops.getPtr(.field_too_long).* += 1;
            continue;
        }

        const row_index = out.items.len - out_start;
        const display_time = try rowTime(stamped_rows, row_index, time_us);
        const row_seq = if (first_seq == 0) 0 else first_seq + out.items.len;
        const wire_op: wire.CommitOp = .{
            .did = c.repo,
            .time_us = display_time,
            .seq = row_seq,
            .rev = c.rev,
            .operation = @tagName(op.action),
            .collection = op.collection,
            .rkey = op.rkey,
            .record = if (is_delete) null else op.record,
            .cid = if (is_delete) null else op.cid,
        };
        // record_cbor = the byte-exact archived payload (round-trip encode
        // is byte-identical per zat's tests)
        const record_cbor: []const u8 = if (is_delete) "" else try zat.cbor.encodeAlloc(allocator, op.record.?);
        if (!representableFields(c.repo, op.collection, op.rkey, c.rev, record_cbor.len)) {
            drops.getPtr(.field_too_long).* += 1;
            continue;
        }
        // Encode each wire independently. A record that one encoder cannot
        // render must not take down ingest: the row is already durable by the
        // time we get here, so propagating would kill the process over remote
        // bytes and do it again on every restart. An empty body is the
        // established "not on this wire" signal and subscribers skip it.
        // Each wire is memoized independently upstream (Entry.Encoded and
        // Entry.EncodedV2 hold separate errors), so a record v1 cannot render
        // must still reach v2 subscribers.
        var v1_failed = false;
        var v2_failed = false;
        var buf: std.Io.Writer.Allocating = .init(allocator);
        errdefer buf.deinit();
        wire.encodeCommit(allocator, &buf.writer, wire_op) catch |err| switch (err) {
            error.OutOfMemory => return err,
            else => v1_failed = true,
        };
        var buf2: std.Io.Writer.Allocating = .init(allocator);
        errdefer buf2.deinit();
        wire.encodeCommitV2(allocator, &buf2.writer, wire_op, row_seq) catch |err| switch (err) {
            error.OutOfMemory => return err,
            else => v2_failed = true,
        };
        if (v1_failed or v2_failed) drops.getPtr(.unencodable_record).* += 1;
        try out.append(allocator, .{
            .kind = .commit,
            .did = c.repo,
            .collection = op.collection,
            .time_us = display_time,
            .seq = row_seq,
            .json = if (v1_failed) "" else try buf.toOwnedSlice(),
            .json_v2 = if (v2_failed) "" else try buf2.toOwnedSlice(),
        });
    }
}

pub fn representableFields(did: []const u8, collection: []const u8, rkey: []const u8, rev: []const u8, payload_len: usize) bool {
    return did.len <= std.math.maxInt(u16) and
        collection.len <= std.math.maxInt(u8) and
        rkey.len <= std.math.maxInt(u8) and
        rev.len <= std.math.maxInt(u8) and
        payload_len <= std.math.maxInt(u32);
}

fn rowTime(stamped_rows: ?[]const @import("../storage/segment.zig").Event, index: usize, fallback: i64) !i64 {
    const rows = stamped_rows orelse return fallback;
    if (index >= rows.len) return error.StampedRowMismatch;
    return rows[index].displayTime();
}

// === tests ===

const testing = std.testing;

// A repository can hold a record the wire encoders cannot render. By the time
// conversion runs the row is already durable, so propagating would kill ingest
// over remote bytes and do it again on every restart. Upstream keeps the row,
// memoizes the encode error, and lets subscribers skip it.
test "unencodable record is isolated instead of failing the live batch" {
    // A CID whose base32 form overflows cidString's fixed buffer makes
    // writeCborValue fail with WriteFailed -- reachable from remote input.
    const long_cid: zat.cbor.Cid = .{ .raw = &@as([100]u8, @splat(0x55)) };
    const bad_record: zat.cbor.Value = .{ .map = &.{
        .{ .key = "subject", .value = .{ .cid = long_cid } },
    } };
    const good_record: zat.cbor.Value = .{ .map = &.{
        .{ .key = "$type", .value = .{ .text = "app.bsky.feed.like" } },
    } };
    const cid = try zat.cbor.Cid.forDagCbor(testing.allocator, "rec");
    defer testing.allocator.free(cid.raw);

    const ops = [_]firehose.RepoOp{
        .{ .action = .create, .collection = "app.bsky.feed.like", .rkey = "3l3qo2vuowo2b", .cid = cid, .record = bad_record },
        .{ .action = .create, .collection = "app.bsky.feed.like", .rkey = "3l3qo2vuowo2c", .cid = cid, .record = good_record },
    };
    const event: firehose.Event = .{ .commit = .{
        .seq = 1,
        .repo = "did:plc:abc",
        .rev = "3l3qo2vutsw2b",
        .time = "2024-09-09T19:46:02.102Z",
        .ops = &ops,
    } };

    var arena_t = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena_t.deinit();
    var frames: std.ArrayList(Frame) = .empty;
    var drops: Drops = .initFill(0);
    // The batch must succeed, not propagate.
    try convert(&arena_t, event, 99, 10, &frames, &drops);

    try testing.expectEqual(@as(usize, 2), frames.items.len);
    try testing.expectEqual(@as(u64, 1), drops.get(.unencodable_record));
    // The unencodable row keeps its seq and is skipped on the wire.
    try testing.expectEqual(@as(usize, 0), frames.items[0].json.len);
    try testing.expectEqual(@as(usize, 0), frames.items[0].json_v2.len);
    // Its healthy sibling in the same commit is unaffected.
    try testing.expect(frames.items[1].json.len > 0);
    try testing.expect(frames.items[1].json_v2.len > 0);
    try testing.expectEqual(frames.items[0].seq + 1, frames.items[1].seq);
}

// Mirrors upstream: the live consumer archives the raw record block without
// decoding it (internal/ingest/live/events.go), and the float is only rejected
// when the subscribe encoder renders JSON (internal/subscribe/handler.go).
test "live float record keeps its row and is skipped on the wire" {
    var arena_t = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena_t.deinit();
    const float_record = try zat.cbor.decodeAll(arena_t.allocator(), "\xa1\x61x\xfb\x3f\xf0\x00\x00\x00\x00\x00\x00");
    const cid = try zat.cbor.Cid.forDagCbor(testing.allocator, "rec");
    defer testing.allocator.free(cid.raw);

    const ops = [_]firehose.RepoOp{
        .{ .action = .create, .collection = "app.bsky.feed.like", .rkey = "3l3qo2vuowo2b", .cid = cid, .record = float_record },
    };
    const event: firehose.Event = .{ .commit = .{
        .seq = 1,
        .repo = "did:plc:abc",
        .rev = "3l3qo2vutsw2b",
        .time = "2024-09-09T19:46:02.102Z",
        .ops = &ops,
    } };

    var frames: std.ArrayList(Frame) = .empty;
    var drops: Drops = .initFill(0);
    try convert(&arena_t, event, 99, 10, &frames, &drops);

    try testing.expectEqual(@as(usize, 1), frames.items.len);
    try testing.expectEqual(@as(u64, 0), drops.get(.missing_block));
    try testing.expectEqual(@as(u64, 1), drops.get(.unencodable_record));
    try testing.expectEqual(@as(usize, 0), frames.items[0].json.len);
    try testing.expectEqual(@as(usize, 0), frames.items[0].json_v2.len);
}

test "commit fans out per op, drops invalid ops" {
    const record: zat.cbor.Value = .{ .map = &.{
        .{ .key = "$type", .value = .{ .text = "app.bsky.feed.like" } },
    } };
    const cid = try zat.cbor.Cid.forDagCbor(testing.allocator, "rec");
    defer testing.allocator.free(cid.raw);

    const ops = [_]firehose.RepoOp{
        .{ .action = .create, .collection = "app.bsky.feed.like", .rkey = "3l3qo2vuowo2b", .cid = cid, .record = record },
        .{ .action = .create, .collection = "not a nsid", .rkey = "k", .cid = cid, .record = record },
        .{ .action = .create, .collection = "app.bsky.feed.like", .rkey = "k2" }, // missing block
        .{ .action = .delete, .collection = "app.bsky.feed.like", .rkey = "k3" }, // deletes need no block
    };
    const event: firehose.Event = .{ .commit = .{
        .seq = 1,
        .repo = "did:plc:abc",
        .rev = "3l3qo2vutsw2b",
        .time = "2024-09-09T19:46:02.102Z",
        .ops = &ops,
    } };

    var arena_t = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena_t.deinit();
    var frames: std.ArrayList(Frame) = .empty;
    var drops: Drops = .initFill(0);
    try convert(&arena_t, event, 99, 10, &frames, &drops);

    try testing.expectEqual(@as(usize, 2), frames.items.len);
    try testing.expectEqual(@as(u64, 1), drops.get(.invalid_collection));
    try testing.expectEqual(@as(u64, 1), drops.get(.missing_block));
    try testing.expect(std.mem.indexOf(u8, frames.items[1].json, "\"operation\":\"delete\"") != null);
}

test "stamped archive rows drive immediate live display time" {
    const segment = @import("../storage/segment.zig");
    const record: zat.cbor.Value = .{ .map = &.{
        .{ .key = "$type", .value = .{ .text = "app.bsky.feed.post" } },
    } };
    const cid = try zat.cbor.Cid.forDagCbor(testing.allocator, "record");
    defer testing.allocator.free(@constCast(cid.raw));
    const ops = [_]firehose.RepoOp{.{
        .action = .create,
        .collection = "app.bsky.feed.post",
        .rkey = "3l3qo2vuowo2b",
        .cid = cid,
        .record = record,
    }};
    const event: firehose.Event = .{ .commit = .{
        .seq = 7,
        .repo = "did:plc:stamped",
        .rev = "3l3qo2vutsw2b",
        .time = "2024-09-09T19:46:02.102Z",
        .ops = &ops,
    } };
    const rows = [_]segment.Event{.{
        .seq = 42,
        .witnessed_at = 100,
        .indexed_at = 200,
        .kind = .create,
        .did = "did:plc:stamped",
        .collection = "app.bsky.feed.post",
        .rkey = "3l3qo2vuowo2b",
        .rev = "3l3qo2vutsw2b",
        .payload = "record",
    }};
    var arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena.deinit();
    var frames: std.ArrayList(Frame) = .empty;
    var drops: Drops = .initFill(0);
    try convertStamped(&arena, event, 100, 42, &rows, &frames, &drops);
    try testing.expectEqual(@as(usize, 1), frames.items.len);
    try testing.expectEqual(@as(i64, 200), frames.items[0].time_us);
    try testing.expect(std.mem.indexOf(u8, frames.items[0].json, "\"time_us\":200") != null);
    try testing.expect(std.mem.indexOf(u8, frames.items[0].json_v2, "\"time\":\"1970-01-01T00:00:00.000200Z\"") != null);
}

test "invalid rev drops whole commit" {
    const ops = [_]firehose.RepoOp{
        .{ .action = .delete, .collection = "app.bsky.feed.like", .rkey = "k" },
    };
    const event: firehose.Event = .{ .commit = .{
        .seq = 1,
        .repo = "did:plc:abc",
        .rev = "NOT_A_TID!",
        .time = "t",
        .ops = &ops,
    } };
    var arena_t = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena_t.deinit();
    var frames: std.ArrayList(Frame) = .empty;
    var drops: Drops = .initFill(0);
    try convert(&arena_t, event, 1, 0, &frames, &drops);
    try testing.expectEqual(@as(usize, 0), frames.items.len);
    try testing.expectEqual(@as(u64, 1), drops.get(.invalid_rev));
}

test "sync produces a v2-only frame; info nothing" {
    var arena_t = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena_t.deinit();
    var frames: std.ArrayList(Frame) = .empty;
    var drops: Drops = .initFill(0);
    try convert(&arena_t, .{ .info = .{} }, 1, 0, &frames, &drops);
    try convert(&arena_t, .{ .sync = .{ .seq = 1, .did = "did:plc:a", .rev = "3l3qo2vutsw2b", .time = "t", .blocks = "" } }, 1, 5, &frames, &drops);
    try testing.expectEqual(@as(usize, 1), frames.items.len);
    try testing.expectEqual(wire.Kind.sync, frames.items[0].kind);
    try testing.expectEqual(@as(usize, 0), frames.items[0].json.len);
    try testing.expect(std.mem.indexOf(u8, frames.items[0].json_v2, "subscribeEvents#sync") != null);
    try testing.expect(std.mem.indexOf(u8, frames.items[0].json_v2, "\"seq\":5") != null);
}

test "identity event encodes" {
    var arena_t = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena_t.deinit();
    var frames: std.ArrayList(Frame) = .empty;
    var drops: Drops = .initFill(0);
    try convert(&arena_t, .{ .identity = .{
        .seq = 5,
        .did = "did:plc:abc",
        .time = "2024-09-09T19:46:02.102Z",
        .handle = "alice.test",
    } }, 42, 0, &frames, &drops);
    try testing.expectEqual(@as(usize, 1), frames.items.len);
    try testing.expectEqual(wire.Kind.identity, frames.items[0].kind);
    try testing.expect(std.mem.indexOf(u8, frames.items[0].json, "\"handle\":\"alice.test\"") != null);
}