//! jetstream wire formats — JSON event encoding
//!
//! v1: one bare JSON object per websocket text frame (legacy-compatible; the
//! envelope carries "cursor" when the local seq is known, matching atmos's
//! JetstreamEvent). v2: network.bsky.jetstream.subscribeEvents — one
//! xrpc.v1.json message frame per event (atproto proposal 0015), payload
//! shapes and key order pinned to upstream's lexgen output (alphabetical).
//! record CBOR is rendered in the atproto data model's JSON representation
//! ({"$link": ...} for CIDs, {"$bytes": ...} for byte strings); the v2 wire
//! carries no duplicate base64 CBOR (upstream #327).

const std = @import("std");
const zat = @import("zat");

const cbor = zat.cbor;
const multibase = zat.multibase;
const Allocator = std.mem.Allocator;

pub const Kind = enum { commit, identity, account, sync };

/// Write a byte slice as a JSON *string*, always.
///
/// `Stringify.write` renders a []const u8 whose contents are not valid UTF-8
/// as an ARRAY OF NUMBERS (std/json/Stringify.zig: "If the content is not
/// valid UTF-8, rendered as an array of numbers instead"). A single bad byte
/// in a did/collection/rkey therefore changes the type of a wire field and
/// breaks every consumer parsing it — observed 2026-08-09 as a subscriber
/// receiving `did` as a list after fault injection tore a row.
///
/// Go's encoding/json substitutes U+FFFD instead, so upstream jetstream can
/// never emit this shape; match that. The valid case is the fast path and
/// keeps byte-for-byte parity.
fn writeString(s: *std.json.Stringify, bytes: []const u8) !void {
    if (std.unicode.utf8ValidateSlice(bytes)) return s.write(bytes);

    try s.beginWriteRaw();
    const w = s.writer;
    try w.writeByte('"');
    var i: usize = 0;
    while (i < bytes.len) {
        const len = std.unicode.utf8ByteSequenceLength(bytes[i]) catch {
            try w.writeAll("\u{FFFD}");
            i += 1;
            continue;
        };
        if (i + len > bytes.len or !std.unicode.utf8ValidateSlice(bytes[i..][0..len])) {
            try w.writeAll("\u{FFFD}");
            i += 1;
            continue;
        }
        try std.json.Stringify.encodeJsonStringChars(bytes[i..][0..len], .{}, w);
        i += len;
    }
    try w.writeByte('"');
    s.endWriteRaw();
}

pub const CommitOp = struct {
    did: []const u8,
    time_us: i64,
    /// local archive seq; 0 = unknown (the envelope omits "cursor", matching
    /// atmos's omitempty so legacy frames never gain a misleading cursor:0)
    seq: u64 = 0,
    rev: []const u8,
    operation: []const u8, // "create" | "update" | "delete"
    collection: []const u8,
    rkey: []const u8,
    record: ?cbor.Value = null, // null for deletes
    cid: ?cbor.Cid = null, // null for deletes
};

/// encode a commit-op event as a complete v1 JSON frame into `w`.
pub fn encodeCommit(allocator: Allocator, w: *std.Io.Writer, op: CommitOp) !void {
    var s: std.json.Stringify = .{ .writer = w };
    try s.beginObject();
    try s.objectField("did");
    try writeString(&s, op.did);
    try s.objectField("time_us");
    try s.write(op.time_us);
    if (op.seq != 0) {
        try s.objectField("cursor");
        try s.write(op.seq);
    }
    try s.objectField("kind");
    try s.write("commit");
    try s.objectField("commit");
    try s.beginObject();
    try s.objectField("rev");
    try writeString(&s, op.rev);
    try s.objectField("operation");
    try s.write(op.operation);
    try s.objectField("collection");
    try writeString(&s, op.collection);
    try s.objectField("rkey");
    try writeString(&s, op.rkey);
    if (op.record) |record| {
        try s.objectField("record");
        try writeCborValue(&s, record);
    }
    if (op.cid) |cid| {
        try s.objectField("cid");
        const cid_str = try cidString(allocator, cid);
        defer allocator.free(cid_str);
        try s.write(cid_str);
    }
    try s.endObject();
    try s.endObject();
}

/// encode an identity event. `seq` and `time` come from the upstream frame.
pub const IdentityArgs = struct {
    did: []const u8,
    time_us: i64,
    local_seq: u64 = 0,
    seq: i64,
    time: []const u8,
    handle: ?[]const u8,
};

pub fn encodeIdentity(w: *std.Io.Writer, args: IdentityArgs) !void {
    var s: std.json.Stringify = .{ .writer = w };
    try s.beginObject();
    try s.objectField("did");
    try writeString(&s, args.did);
    try s.objectField("time_us");
    try s.write(args.time_us);
    if (args.local_seq != 0) {
        try s.objectField("cursor");
        try s.write(args.local_seq);
    }
    try s.objectField("kind");
    try s.write("identity");
    try s.objectField("identity");
    try s.beginObject();
    try s.objectField("did");
    try writeString(&s, args.did);
    if (args.handle) |h| {
        try s.objectField("handle");
        try writeString(&s, h);
    }
    try s.objectField("seq");
    try s.write(args.seq);
    try s.objectField("time");
    try s.write(args.time);
    try s.endObject();
    try s.endObject();
}

/// encode an account event.
pub const AccountArgs = struct {
    did: []const u8,
    time_us: i64,
    local_seq: u64 = 0,
    seq: i64,
    time: []const u8,
    active: bool,
    status: ?[]const u8,
};

pub fn encodeAccount(w: *std.Io.Writer, args: AccountArgs) !void {
    var s: std.json.Stringify = .{ .writer = w };
    try s.beginObject();
    try s.objectField("did");
    try writeString(&s, args.did);
    try s.objectField("time_us");
    try s.write(args.time_us);
    if (args.local_seq != 0) {
        try s.objectField("cursor");
        try s.write(args.local_seq);
    }
    try s.objectField("kind");
    try s.write("account");
    try s.objectField("account");
    try s.beginObject();
    try s.objectField("active");
    try s.write(args.active);
    try s.objectField("did");
    try writeString(&s, args.did);
    try s.objectField("seq");
    try s.write(args.seq);
    if (args.status) |st| {
        try s.objectField("status");
        try s.write(st);
    }
    try s.objectField("time");
    try s.write(args.time);
    try s.endObject();
    try s.endObject();
}

/// CIDv1 string form: multibase base32lower with 'b' prefix.
pub fn cidString(allocator: Allocator, cid: cbor.Cid) ![]u8 {
    return multibase.encode(allocator, .base32lower, cid.raw);
}

/// render a decoded DAG-CBOR value in the atproto JSON representation.
fn writeCborValue(s: *std.json.Stringify, value: cbor.Value) !void {
    switch (value) {
        .unsigned => |v| try s.write(v),
        .negative => |v| try s.write(v),
        // upstream's cbor.ToJSON rejects floats as outside the atproto data
        // model (atmos cbor/json.go), so the record is unencodable, not emitted
        .float => return error.WriteFailed,
        // record text can be any bytes a PDS chose to write; the same
        // array-instead-of-string hazard applies to it as to the envelope
        .text => |v| try writeString(s, v),
        .boolean => |v| try s.write(v),
        .null => try s.write(null),
        .bytes => |v| {
            try s.beginObject();
            try s.objectField("$bytes");
            try s.beginWriteRaw();
            try s.writer.writeByte('"');
            try writeBase64NoPad(s.writer, v);
            try s.writer.writeByte('"');
            s.endWriteRaw();
            try s.endObject();
        },
        .cid => |cid| {
            try s.beginObject();
            try s.objectField("$link");
            var buf: [128]u8 = undefined;
            var fba = std.heap.FixedBufferAllocator.init(&buf);
            const str = cidString(fba.allocator(), cid) catch return error.WriteFailed;
            try s.write(str);
            try s.endObject();
        },
        .array => |items| {
            try s.beginArray();
            for (items) |item| try writeCborValue(s, item);
            try s.endArray();
        },
        .map => |entries| {
            try s.beginObject();
            for (entries) |entry| {
                try s.objectField(entry.key);
                try writeCborValue(s, entry.value);
            }
            try s.endObject();
        },
    }
}

fn writeBase64NoPad(w: *std.Io.Writer, data: []const u8) !void {
    const encoder = std.base64.standard_no_pad.Encoder;
    var chunk: [4]u8 = undefined;
    var i: usize = 0;
    while (i < data.len) : (i += 3) {
        const n = @min(3, data.len - i);
        const out = encoder.encode(&chunk, data[i..][0..n]);
        try w.writeAll(out);
    }
}

// === tests ===

const testing = std.testing;

fn expectEncode(expected: []const u8, actual: []const u8) !void {
    try testing.expectEqualStrings(expected, actual);
}

test "encodeCommit create with record" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();

    const record: cbor.Value = .{ .map = &.{
        .{ .key = "$type", .value = .{ .text = "app.bsky.feed.like" } },
        .{ .key = "count", .value = .{ .unsigned = 3 } },
    } };
    // CIDv1 dag-cbor sha2-256 over "hi" (contents don't matter for encoding)
    const cid = try cbor.Cid.forDagCbor(testing.allocator, "hi");
    defer testing.allocator.free(cid.raw);

    try encodeCommit(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 1725911162329308,
        .rev = "3l3qo2vutsw2b",
        .operation = "create",
        .collection = "app.bsky.feed.like",
        .rkey = "3l3qo2vuowo2b",
        .record = record,
        .cid = cid,
    });

    const json = out.written();
    // envelope fields present and ordered
    try testing.expect(std.mem.startsWith(u8, json, "{\"did\":\"did:plc:abc\",\"time_us\":1725911162329308,\"kind\":\"commit\","));
    try testing.expect(std.mem.indexOf(u8, json, "\"operation\":\"create\"") != null);
    try testing.expect(std.mem.indexOf(u8, json, "\"record\":{\"$type\":\"app.bsky.feed.like\",\"count\":3}") != null);
    try testing.expect(std.mem.indexOf(u8, json, "\"cid\":\"b") != null);
}

test "encodeCommit delete omits record and cid" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    try encodeCommit(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 1,
        .rev = "3l3qo2vutsw2b",
        .operation = "delete",
        .collection = "app.bsky.feed.like",
        .rkey = "xyz",
    });
    const json = out.written();
    try testing.expect(std.mem.indexOf(u8, json, "record") == null);
    try testing.expect(std.mem.indexOf(u8, json, "cid") == null);
    try testing.expect(std.mem.indexOf(u8, json, "\"rkey\":\"xyz\"}}") != null);
}

test "encodeIdentity and encodeAccount" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    try encodeIdentity(&out.writer, .{
        .did = "did:plc:abc",
        .time_us = 42,
        .seq = 7,
        .time = "2024-09-09T19:46:02.102Z",
        .handle = "alice.test",
    });
    try expectEncode(
        "{\"did\":\"did:plc:abc\",\"time_us\":42,\"kind\":\"identity\",\"identity\":{\"did\":\"did:plc:abc\",\"handle\":\"alice.test\",\"seq\":7,\"time\":\"2024-09-09T19:46:02.102Z\"}}",
        out.written(),
    );

    out.clearRetainingCapacity();
    try encodeAccount(&out.writer, .{
        .did = "did:plc:abc",
        .time_us = 42,
        .seq = 7,
        .time = "2024-09-09T19:46:02.102Z",
        .active = false,
        .status = "takendown",
    });
    try expectEncode(
        "{\"did\":\"did:plc:abc\",\"time_us\":42,\"kind\":\"account\",\"account\":{\"active\":false,\"did\":\"did:plc:abc\",\"seq\":7,\"status\":\"takendown\",\"time\":\"2024-09-09T19:46:02.102Z\"}}",
        out.written(),
    );
}

test "record with cid link and bytes" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    const cid = try cbor.Cid.forDagCbor(testing.allocator, "x");
    defer testing.allocator.free(cid.raw);
    const record: cbor.Value = .{ .map = &.{
        .{ .key = "ref", .value = .{ .cid = cid } },
        .{ .key = "blob", .value = .{ .bytes = "\x01\x02\x03" } },
    } };
    try encodeCommit(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 1,
        .rev = "3l3qo2vutsw2b",
        .operation = "create",
        .collection = "app.bsky.feed.post",
        .rkey = "k",
        .record = record,
        .cid = cid,
    });
    const json = out.written();
    try testing.expect(std.mem.indexOf(u8, json, "\"ref\":{\"$link\":\"b") != null);
    try testing.expect(std.mem.indexOf(u8, json, "\"blob\":{\"$bytes\":\"AQID\"}") != null);
}

// Mirrors upstream: atmos decodes a finite float64 but cbor.ToJSON rejects it
// as outside the atproto data model, so the subscribe encoder errors instead
// of emitting a JSON number.
test "record with a float is unencodable" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    const cid = try cbor.Cid.forDagCbor(testing.allocator, "x");
    defer testing.allocator.free(cid.raw);
    var arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena.deinit();
    const record = try cbor.decodeAll(arena.allocator(), "\xa1\x61x\xfb\x3f\xf0\x00\x00\x00\x00\x00\x00");
    try testing.expectError(error.WriteFailed, encodeCommit(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 1,
        .rev = "3l3qo2vutsw2b",
        .operation = "create",
        .collection = "app.bsky.feed.post",
        .rkey = "k",
        .record = record,
        .cid = cid,
    }));
}

// === network.bsky.jetstream.subscribeEvents: proposal-0015 wire ===
// One xrpc.v1.json message frame per event:
//   {"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#<kind>", ...}}
// Payload key order is pinned to upstream's lexgen output (alphabetical,
// $type first); `time` is RFC 3339 UTC with exactly six fractional digits
// (upstream encoder.go wireTimeLayout). No record_cbor (upstream #327).

const message_frame_prefix = "{\"$type\":\"message\",\"payload\":";
const nsid = "network.bsky.jetstream.subscribeEvents";

/// render a unix-microseconds timestamp as upstream's wireTimeLayout:
/// "2006-01-02T15:04:05.000000Z" — one rendering per instant, byte-stable.
pub fn writeWireTime(w: *std.Io.Writer, us: i64) !void {
    const text = zat.Datetime.fromMicros(us).toString();
    try w.writeAll(&text);
}

fn writePayloadType(s: *std.json.Stringify, fragment: []const u8) !void {
    try s.objectField("$type");
    try s.beginWriteRaw();
    try s.writer.writeByte('"');
    try s.writer.writeAll(nsid);
    try s.writer.writeByte('#');
    try s.writer.writeAll(fragment);
    try s.writer.writeByte('"');
    s.endWriteRaw();
}

fn writeEnvelopeTail(s: *std.json.Stringify, seq: u64, time_us: i64) !void {
    try s.objectField("seq");
    try s.write(seq);
    try s.objectField("time");
    try s.beginWriteRaw();
    try s.writer.writeByte('"');
    try writeWireTime(s.writer, time_us);
    try s.writer.writeByte('"');
    s.endWriteRaw();
}

/// {"$type":"message","payload":{"$type":"...#commit","cid"?,"collection",
///  "did","operation","record"?,"rev","rkey","seq","time"}}
pub fn encodeCommitV2(allocator: Allocator, w: *std.Io.Writer, op: CommitOp, seq: u64) !void {
    try w.writeAll(message_frame_prefix);
    var s: std.json.Stringify = .{ .writer = w };
    try s.beginObject();
    try writePayloadType(&s, "commit");
    if (op.cid) |cid| {
        try s.objectField("cid");
        const cid_str = try cidString(allocator, cid);
        defer allocator.free(cid_str);
        try s.write(cid_str);
    }
    try s.objectField("collection");
    try writeString(&s, op.collection);
    try s.objectField("did");
    try writeString(&s, op.did);
    try s.objectField("operation");
    try s.write(op.operation);
    if (op.record) |record| {
        try s.objectField("record");
        try writeCborValue(&s, record);
    }
    try s.objectField("rev");
    try writeString(&s, op.rev);
    try s.objectField("rkey");
    try writeString(&s, op.rkey);
    try writeEnvelopeTail(&s, seq, op.time_us);
    try s.endObject();
    try w.writeByte('}');
}

/// {"$type":"message","payload":{"$type":"...#identity","did","identity":{...upstream verbatim},"seq","time"}}
pub fn encodeIdentityV2(w: *std.Io.Writer, args: IdentityArgs, seq: u64) !void {
    try w.writeAll(message_frame_prefix);
    var s: std.json.Stringify = .{ .writer = w };
    try s.beginObject();
    try writePayloadType(&s, "identity");
    try s.objectField("did");
    try writeString(&s, args.did);
    try s.objectField("identity");
    try s.beginObject();
    try s.objectField("did");
    try writeString(&s, args.did);
    if (args.handle) |h| {
        try s.objectField("handle");
        try writeString(&s, h);
    }
    try s.objectField("seq");
    try s.write(args.seq);
    try s.objectField("time");
    try s.write(args.time);
    try s.endObject();
    try writeEnvelopeTail(&s, seq, args.time_us);
    try s.endObject();
    try w.writeByte('}');
}

/// {"$type":"message","payload":{"$type":"...#account","account":{...},"did","seq","time"}}
pub fn encodeAccountV2(w: *std.Io.Writer, args: AccountArgs, seq: u64) !void {
    try w.writeAll(message_frame_prefix);
    var s: std.json.Stringify = .{ .writer = w };
    try s.beginObject();
    try writePayloadType(&s, "account");
    try s.objectField("account");
    try s.beginObject();
    try s.objectField("active");
    try s.write(args.active);
    try s.objectField("did");
    try writeString(&s, args.did);
    try s.objectField("seq");
    try s.write(args.seq);
    if (args.status) |st| {
        try s.objectField("status");
        try s.write(st);
    }
    try s.objectField("time");
    try s.write(args.time);
    try s.endObject();
    try s.objectField("did");
    try writeString(&s, args.did);
    try writeEnvelopeTail(&s, seq, args.time_us);
    try s.endObject();
    try w.writeByte('}');
}

/// {"$type":"message","payload":{"$type":"...#sync","did","seq","sync":{...},"time"}}
pub fn encodeSyncV2(w: *std.Io.Writer, args: struct {
    did: []const u8,
    time_us: i64,
    upstream_seq: i64,
    rev: []const u8,
    time: []const u8,
    blocks: []const u8,
}, seq: u64) !void {
    try w.writeAll(message_frame_prefix);
    var s: std.json.Stringify = .{ .writer = w };
    try s.beginObject();
    try writePayloadType(&s, "sync");
    try s.objectField("did");
    try writeString(&s, args.did);
    try s.objectField("seq");
    try s.write(seq);
    try s.objectField("sync");
    try s.beginObject();
    // lexicon "bytes" value: the {"$bytes":"<unpadded std base64>"} sentinel
    // (atmos cbor.AppendJSONBytes) — a bare base64 string is rejected by
    // conformant decoders (caught live by the d4dd2f0 differential oracle)
    try s.objectField("blocks");
    try s.beginObject();
    try s.objectField("$bytes");
    try s.beginWriteRaw();
    try s.writer.writeByte('"');
    try writeBase64NoPad(s.writer, args.blocks);
    try s.writer.writeByte('"');
    s.endWriteRaw();
    try s.endObject();
    try s.objectField("did");
    try writeString(&s, args.did);
    try s.objectField("rev");
    try s.write(args.rev);
    try s.objectField("seq");
    try s.write(args.upstream_seq);
    try s.objectField("time");
    try s.write(args.time);
    try s.endObject();
    try s.objectField("time");
    try s.beginWriteRaw();
    try s.writer.writeByte('"');
    try writeWireTime(s.writer, args.time_us);
    try s.writer.writeByte('"');
    s.endWriteRaw();
    try s.endObject();
    try w.writeByte('}');
}

/// #info advisory (seq-less): {"$type":"message","payload":{"$type":"...#info","message"?,"name"}}
pub fn encodeInfoV2(w: *std.Io.Writer, name: []const u8, message: ?[]const u8) !void {
    try w.writeAll(message_frame_prefix);
    var s: std.json.Stringify = .{ .writer = w };
    try s.beginObject();
    try writePayloadType(&s, "info");
    if (message) |m| {
        try s.objectField("message");
        try s.write(m);
    }
    try s.objectField("name");
    try s.write(name);
    try s.endObject();
    try w.writeByte('}');
}

/// terminal error frame: {"$type":"error","error":code,"message"?}. the
/// connection closes immediately after one is sent.
pub fn encodeErrorV2(w: *std.Io.Writer, code: []const u8, message: ?[]const u8) !void {
    var s: std.json.Stringify = .{ .writer = w };
    try s.beginObject();
    try s.objectField("$type");
    try s.write("error");
    try s.objectField("error");
    try s.write(code);
    if (message) |m| {
        try s.objectField("message");
        try s.write(m);
    }
    try s.endObject();
}

test "encodeCommitV2 emits a proposal-0015 message frame in lexgen key order" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    const record: cbor.Value = .{ .map = &.{
        .{ .key = "$type", .value = .{ .text = "app.bsky.feed.like" } },
    } };
    const payload = "\xa1\x65\x24\x74\x79\x70\x65\x72app.bsky.feed.like";
    const cid = try cbor.Cid.forDagCbor(testing.allocator, payload);
    defer testing.allocator.free(cid.raw);
    try encodeCommitV2(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 1725911162329308,
        .rev = "3l3qo2vutsw2b",
        .operation = "create",
        .collection = "app.bsky.feed.like",
        .rkey = "k",
        .record = record,
        .cid = cid,
    }, 42);
    const json = out.written();
    try testing.expect(std.mem.startsWith(u8, json, "{\"$type\":\"message\",\"payload\":{\"$type\":\"network.bsky.jetstream.subscribeEvents#commit\",\"cid\":\"b"));
    // lexgen order: cid, collection, did, operation, record, rev, rkey, seq, time
    const c = std.mem.indexOf(u8, json, "\"collection\":").?;
    const d = std.mem.indexOf(u8, json, "\"did\":").?;
    const o = std.mem.indexOf(u8, json, "\"operation\":").?;
    const r = std.mem.indexOf(u8, json, "\"record\":").?;
    try testing.expect(c < d and d < o and o < r);
    try testing.expect(std.mem.indexOf(u8, json, "\"seq\":42,\"time\":\"2024-09-09T19:46:02.329308Z\"}}") != null);
    try testing.expect(std.mem.indexOf(u8, json, "record_cbor") == null);
    try testing.expect(std.mem.indexOf(u8, json, "time_us") == null);
}

test "encodeCommitV2 delete omits cid and record" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    try encodeCommitV2(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 0,
        .rev = "r",
        .operation = "delete",
        .collection = "app.bsky.feed.like",
        .rkey = "k",
    }, 7);
    const json = out.written();
    try testing.expect(std.mem.indexOf(u8, json, "cid") == null);
    try testing.expect(std.mem.indexOf(u8, json, "record") == null);
    try testing.expect(std.mem.indexOf(u8, json, "\"time\":\"1970-01-01T00:00:00.000000Z\"") != null);
}

test "encodeInfoV2 and encodeErrorV2 frames" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    try encodeInfoV2(&out.writer, "OutdatedCursor", "requested cursor was below retention; clamped");
    try testing.expect(std.mem.startsWith(u8, out.written(), "{\"$type\":\"message\",\"payload\":{\"$type\":\"network.bsky.jetstream.subscribeEvents#info\",\"message\":"));
    try testing.expect(std.mem.endsWith(u8, out.written(), "\"name\":\"OutdatedCursor\"}}"));

    out.clearRetainingCapacity();
    try encodeErrorV2(&out.writer, "ConsumerTooSlow", null);
    try testing.expectEqualStrings("{\"$type\":\"error\",\"error\":\"ConsumerTooSlow\"}", out.written());
}

test "v1 envelope carries cursor when the seq is known, never when zero" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    try encodeCommit(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 1,
        .seq = 42,
        .rev = "r",
        .operation = "delete",
        .collection = "app.bsky.feed.like",
        .rkey = "k",
    });
    try testing.expect(std.mem.startsWith(u8, out.written(), "{\"did\":\"did:plc:abc\",\"time_us\":1,\"cursor\":42,\"kind\":\"commit\","));
    out.clearRetainingCapacity();
    try encodeCommit(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 1,
        .rev = "r",
        .operation = "delete",
        .collection = "app.bsky.feed.like",
        .rkey = "k",
    });
    try testing.expect(std.mem.indexOf(u8, out.written(), "cursor") == null);
}

test "invalid utf8 in a string field stays a JSON string (never an array)" {
    // Regression: Stringify.write renders a non-UTF-8 []const u8 as an array
    // of numbers, so one torn byte turned `did` into a list on the wire and
    // broke consumers (2026-08-09, caught by the lifecycle oracle's fault
    // injection). Go's encoder substitutes U+FFFD; so do we.
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    try encodeCommit(testing.allocator, &out.writer, .{
        .did = "did:plc:a\xffb",
        .time_us = 1,
        .rev = "rev1",
        .operation = "create",
        .collection = "app.bsky.feed.post",
        .rkey = "k\xfe",
    });
    const json = out.writer.buffered();
    // the shape a consumer depends on: a quoted string, not [100,105,...]
    try testing.expect(std.mem.indexOf(u8, json, "\"did\":\"did:plc:a") != null);
    try testing.expect(std.mem.indexOf(u8, json, "\"did\":[") == null);
    try testing.expect(std.mem.indexOf(u8, json, "\"rkey\":[") == null);
    // invalid bytes became the replacement character, and it still parses
    try testing.expect(std.mem.indexOf(u8, json, "\u{FFFD}") != null);
    var parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, json, .{});
    defer parsed.deinit();
    try testing.expect(parsed.value.object.get("did").? == .string);
}

test "valid utf8 fields are untouched (byte parity with upstream)" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    try encodeCommit(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 1,
        .rev = "rev1",
        .operation = "create",
        .collection = "app.bsky.feed.post",
        .rkey = "k1",
    });
    try testing.expect(std.mem.startsWith(u8, out.writer.buffered(), "{\"did\":\"did:plc:abc\","));
}

test "record text and keys with invalid utf8 still produce parseable JSON" {
    // A PDS can put arbitrary bytes in a CBOR text value. Both the value and
    // the map key must survive as JSON strings: an array in key position is
    // not merely surprising, it is invalid JSON and breaks every parser.
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    const record: cbor.Value = .{ .map = &.{
        .{ .key = "text", .value = .{ .text = "hello \xff world" } },
    } };
    try encodeCommit(testing.allocator, &out.writer, .{
        .did = "did:plc:abc",
        .time_us = 1,
        .rev = "rev1",
        .operation = "create",
        .collection = "app.bsky.feed.post",
        .rkey = "k1",
        .record = record,
    });
    const json = out.writer.buffered();
    var parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, json, .{});
    defer parsed.deinit();
    const rec = parsed.value.object.get("commit").?.object.get("record").?;
    try testing.expect(rec.object.get("text").? == .string);
}