zat.dev
atproto utils for zig
zat.dev
atproto
sdk
zig
fix: align firehose event types with sync spec
- add missing CommitEvent fields: since, commit CID, blobs
- add cid to RepoOp for downstream CID verification
- make rev, time required on CommitEvent (spec: required)
- make time required on IdentityEvent and AccountEvent (spec: required)
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
zzstoatzz.io
+91
-51
src/internal/firehose.zig
+91
-51
src/internal/firehose.zig
···
49
49
pub const CommitEvent = struct {
50
50
seq: i64,
51
51
repo: []const u8, // DID
52
-
rev: ?[]const u8 = null,
53
-
time: ?[]const u8 = null,
52
+
rev: []const u8, // TID — revision of the commit
53
+
time: []const u8, // datetime — when event was received
54
+
since: ?[]const u8 = null, // TID — rev of preceding commit (null = full repo export)
55
+
commit: ?cbor.Cid = null, // CID of the commit object
54
56
ops: []const RepoOp,
57
+
blobs: []const cbor.Cid = &.{}, // new blobs referenced by records in this commit
55
58
too_big: bool = false,
56
59
};
57
60
···
59
62
action: CommitAction,
60
63
collection: []const u8,
61
64
rkey: []const u8,
65
+
cid: ?cbor.Cid = null, // CID of the record (null for deletes)
62
66
record: ?cbor.Value = null, // decoded DAG-CBOR record from CAR block
63
67
};
64
68
65
69
pub const IdentityEvent = struct {
66
70
seq: i64,
67
71
did: []const u8,
68
-
time: ?[]const u8 = null,
72
+
time: []const u8, // datetime — when event was received
69
73
handle: ?[]const u8 = null,
70
74
};
71
75
72
76
pub const AccountEvent = struct {
73
77
seq: i64,
74
78
did: []const u8,
75
-
time: ?[]const u8 = null,
79
+
time: []const u8, // datetime — when event was received
76
80
active: bool = true,
77
81
status: ?AccountStatus = null,
78
82
};
···
137
141
fn decodeCommit(allocator: Allocator, payload: cbor.Value) DecodeError!Event {
138
142
const seq_val = payload.getInt("seq") orelse return error.MissingField;
139
143
const repo = payload.getString("repo") orelse return error.MissingField;
144
+
const rev = payload.getString("rev") orelse return error.MissingField;
145
+
const time = payload.getString("time") orelse return error.MissingField;
146
+
147
+
// parse commit CID
148
+
var commit_cid: ?cbor.Cid = null;
149
+
if (payload.get("commit")) |commit_val| {
150
+
switch (commit_val) {
151
+
.cid => |c| commit_cid = c,
152
+
else => {},
153
+
}
154
+
}
155
+
156
+
// parse blobs array (array of CID links)
157
+
var blobs: std.ArrayList(cbor.Cid) = .{};
158
+
if (payload.getArray("blobs")) |blob_values| {
159
+
for (blob_values) |blob_val| {
160
+
switch (blob_val) {
161
+
.cid => |c| try blobs.append(allocator, c),
162
+
else => {},
163
+
}
164
+
}
165
+
}
140
166
141
167
// parse CAR blocks
142
168
const blocks_bytes = payload.getBytes("blocks");
···
160
186
const collection = path[0..slash];
161
187
const rkey = path[slash + 1 ..];
162
188
163
-
// look up record from CAR blocks via CID
189
+
// extract CID from op and look up record from CAR blocks
190
+
var op_cid: ?cbor.Cid = null;
164
191
var record: ?cbor.Value = null;
165
-
if (parsed_car) |c| {
166
-
if (op_val.get("cid")) |cid_val| {
167
-
switch (cid_val) {
168
-
.cid => |cid| {
192
+
if (op_val.get("cid")) |cid_val| {
193
+
switch (cid_val) {
194
+
.cid => |cid| {
195
+
op_cid = cid;
196
+
if (parsed_car) |c| {
169
197
if (car.findBlock(c, cid.raw)) |block_data| {
170
198
record = cbor.decodeAll(allocator, block_data) catch null;
171
199
}
172
-
},
173
-
else => {},
174
-
}
200
+
}
201
+
},
202
+
else => {},
175
203
}
176
204
}
177
205
···
179
207
.action = action,
180
208
.collection = collection,
181
209
.rkey = rkey,
210
+
.cid = op_cid,
182
211
.record = record,
183
212
});
184
213
}
···
187
216
return .{ .commit = .{
188
217
.seq = seq_val,
189
218
.repo = repo,
190
-
.rev = payload.getString("rev"),
191
-
.time = payload.getString("time"),
219
+
.rev = rev,
220
+
.time = time,
221
+
.since = payload.getString("since"),
222
+
.commit = commit_cid,
192
223
.ops = try ops.toOwnedSlice(allocator),
224
+
.blobs = try blobs.toOwnedSlice(allocator),
193
225
.too_big = payload.getBool("tooBig") orelse false,
194
226
} };
195
227
}
···
198
230
return .{ .identity = .{
199
231
.seq = payload.getInt("seq") orelse return error.MissingField,
200
232
.did = payload.getString("did") orelse return error.MissingField,
201
-
.time = payload.getString("time"),
233
+
.time = payload.getString("time") orelse return error.MissingField,
202
234
.handle = payload.getString("handle"),
203
235
} };
204
236
}
···
208
240
return .{ .account = .{
209
241
.seq = payload.getInt("seq") orelse return error.MissingField,
210
242
.did = payload.getString("did") orelse return error.MissingField,
211
-
.time = payload.getString("time"),
243
+
.time = payload.getString("time") orelse return error.MissingField,
212
244
.active = payload.getBool("active") orelse true,
213
245
.status = if (status_str) |s| AccountStatus.parse(s) else null,
214
246
} };
···
293
325
.blocks = car_blocks.items,
294
326
};
295
327
const blocks_bytes = try car.writeAlloc(allocator, car_data);
328
+
329
+
// build blobs array
330
+
var blob_values: std.ArrayList(cbor.Value) = .{};
331
+
defer blob_values.deinit(allocator);
332
+
for (commit.blobs) |blob| {
333
+
try blob_values.append(allocator, .{ .cid = blob });
334
+
}
296
335
297
336
// build payload entries
298
337
var entries: std.ArrayList(cbor.Value.MapEntry) = .{};
299
338
defer entries.deinit(allocator);
300
339
301
-
if (blocks_bytes.len > 0) {
302
-
try entries.append(allocator, .{ .key = "blocks", .value = .{ .bytes = blocks_bytes } });
340
+
try entries.append(allocator, .{ .key = "blocks", .value = .{ .bytes = blocks_bytes } });
341
+
if (commit.commit) |c| {
342
+
try entries.append(allocator, .{ .key = "commit", .value = .{ .cid = c } });
303
343
}
344
+
try entries.append(allocator, .{ .key = "blobs", .value = .{ .array = blob_values.items } });
304
345
try entries.append(allocator, .{ .key = "ops", .value = .{ .array = op_values.items } });
305
346
try entries.append(allocator, .{ .key = "repo", .value = .{ .text = commit.repo } });
306
-
if (commit.rev) |rev| {
307
-
try entries.append(allocator, .{ .key = "rev", .value = .{ .text = rev } });
308
-
}
347
+
try entries.append(allocator, .{ .key = "rev", .value = .{ .text = commit.rev } });
309
348
try entries.append(allocator, .{ .key = "seq", .value = .{ .unsigned = @intCast(commit.seq) } });
310
-
if (commit.time) |t| {
311
-
try entries.append(allocator, .{ .key = "time", .value = .{ .text = t } });
349
+
if (commit.since) |s| {
350
+
try entries.append(allocator, .{ .key = "since", .value = .{ .text = s } });
312
351
}
352
+
try entries.append(allocator, .{ .key = "time", .value = .{ .text = commit.time } });
313
353
if (commit.too_big) {
314
354
try entries.append(allocator, .{ .key = "tooBig", .value = .{ .boolean = true } });
315
355
}
···
326
366
try entries.append(allocator, .{ .key = "handle", .value = .{ .text = h } });
327
367
}
328
368
try entries.append(allocator, .{ .key = "seq", .value = .{ .unsigned = @intCast(identity.seq) } });
329
-
if (identity.time) |t| {
330
-
try entries.append(allocator, .{ .key = "time", .value = .{ .text = t } });
331
-
}
369
+
try entries.append(allocator, .{ .key = "time", .value = .{ .text = identity.time } });
332
370
333
371
try cbor.encode(allocator, writer, .{ .map = entries.items });
334
372
}
···
345
383
if (account.status) |s| {
346
384
try entries.append(allocator, .{ .key = "status", .value = .{ .text = @tagName(s) } });
347
385
}
348
-
if (account.time) |t| {
349
-
try entries.append(allocator, .{ .key = "time", .value = .{ .text = t } });
350
-
}
386
+
try entries.append(allocator, .{ .key = "time", .value = .{ .text = account.time } });
351
387
352
388
try cbor.encode(allocator, writer, .{ .map = entries.items });
353
389
}
···
502
538
defer arena.deinit();
503
539
const alloc = arena.allocator();
504
540
505
-
// header: {op: 1, t: "#identity"}
506
-
const header_bytes = [_]u8{
507
-
0xa2, // map(2)
508
-
0x62, 'o', 'p', 0x01, // "op": 1
509
-
0x61, 't', 0x69, '#', 'i', 'd', 'e', 'n', 't', 'i', 't', 'y', // "t": "#identity"
510
-
};
511
-
// payload: {seq: 42, did: "did:plc:test"}
512
-
const payload_bytes = [_]u8{
513
-
0xa2, // map(2)
514
-
0x63, 's', 'e', 'q', 0x18, 42, // "seq": 42
515
-
0x63, 'd', 'i', 'd', 0x6c, 'd', 'i', 'd', ':', 'p', 'l', 'c', ':', 't', 'e', 's', 't', // "did": "did:plc:test"
516
-
};
541
+
// build frame via encoder for cleaner test
542
+
const original = Event{ .identity = .{
543
+
.seq = 42,
544
+
.did = "did:plc:test",
545
+
.time = "2024-01-15T10:30:00Z",
546
+
} };
547
+
const frame = try encodeFrame(alloc, original);
517
548
518
-
var frame: [header_bytes.len + payload_bytes.len]u8 = undefined;
519
-
@memcpy(frame[0..header_bytes.len], &header_bytes);
520
-
@memcpy(frame[header_bytes.len..], &payload_bytes);
521
-
522
-
const event = try decodeFrame(alloc, &frame);
549
+
const event = try decodeFrame(alloc, frame);
523
550
const identity = event.identity;
524
551
try std.testing.expectEqual(@as(i64, 42), identity.seq);
525
552
try std.testing.expectEqualStrings("did:plc:test", identity.did);
553
+
try std.testing.expectEqualStrings("2024-01-15T10:30:00Z", identity.time);
526
554
}
527
555
528
556
test "Event.seq works" {
···
532
560
const identity_event = Event{ .identity = .{
533
561
.seq = 42,
534
562
.did = "did:plc:test",
563
+
.time = "2024-01-15T10:30:00Z",
535
564
} };
536
565
try std.testing.expectEqual(@as(i64, 42), identity_event.seq().?);
537
566
}
···
563
592
const original = Event{ .identity = .{
564
593
.seq = 42,
565
594
.did = "did:plc:test123",
566
-
.handle = "alice.bsky.social",
567
595
.time = "2024-01-15T10:30:00Z",
596
+
.handle = "alice.bsky.social",
568
597
} };
569
598
570
599
const frame = try encodeFrame(alloc, original);
···
573
602
const id = decoded.identity;
574
603
try std.testing.expectEqual(@as(i64, 42), id.seq);
575
604
try std.testing.expectEqualStrings("did:plc:test123", id.did);
605
+
try std.testing.expectEqualStrings("2024-01-15T10:30:00Z", id.time);
576
606
try std.testing.expectEqualStrings("alice.bsky.social", id.handle.?);
577
-
try std.testing.expectEqualStrings("2024-01-15T10:30:00Z", id.time.?);
578
607
}
579
608
580
609
test "encode → decode account frame" {
···
585
614
const original = Event{ .account = .{
586
615
.seq = 100,
587
616
.did = "did:plc:suspended",
617
+
.time = "2024-01-15T10:30:00Z",
588
618
.active = false,
589
619
.status = .suspended,
590
-
.time = "2024-01-15T10:30:00Z",
591
620
} };
592
621
593
622
const frame = try encodeFrame(alloc, original);
···
596
625
const acct = decoded.account;
597
626
try std.testing.expectEqual(@as(i64, 100), acct.seq);
598
627
try std.testing.expectEqualStrings("did:plc:suspended", acct.did);
628
+
try std.testing.expectEqualStrings("2024-01-15T10:30:00Z", acct.time);
599
629
try std.testing.expectEqual(false, acct.active);
600
630
try std.testing.expectEqual(AccountStatus.suspended, acct.status.?);
601
631
}
···
613
643
const original = Event{ .commit = .{
614
644
.seq = 999,
615
645
.repo = "did:plc:poster",
616
-
.rev = "abc123",
646
+
.rev = "3k2abc000000",
617
647
.time = "2024-01-15T10:30:00Z",
648
+
.since = "3k2abd000000",
618
649
.ops = &.{.{
619
650
.action = .create,
620
651
.collection = "app.bsky.feed.post",
···
629
660
const commit = decoded.commit;
630
661
try std.testing.expectEqual(@as(i64, 999), commit.seq);
631
662
try std.testing.expectEqualStrings("did:plc:poster", commit.repo);
632
-
try std.testing.expectEqualStrings("abc123", commit.rev.?);
663
+
try std.testing.expectEqualStrings("3k2abc000000", commit.rev);
664
+
try std.testing.expectEqualStrings("2024-01-15T10:30:00Z", commit.time);
665
+
try std.testing.expectEqualStrings("3k2abd000000", commit.since.?);
666
+
try std.testing.expectEqual(@as(usize, 0), commit.blobs.len);
633
667
try std.testing.expectEqual(@as(usize, 1), commit.ops.len);
634
668
635
669
const op = commit.ops[0];
636
670
try std.testing.expectEqual(CommitAction.create, op.action);
637
671
try std.testing.expectEqualStrings("app.bsky.feed.post", op.collection);
638
672
try std.testing.expectEqualStrings("3k2abc", op.rkey);
673
+
try std.testing.expect(op.cid != null);
639
674
640
675
// record should be decoded from the CAR blocks
641
676
const rec = op.record.?;
···
651
686
const original = Event{ .commit = .{
652
687
.seq = 500,
653
688
.repo = "did:plc:deleter",
689
+
.rev = "3k2xyz000000",
690
+
.time = "2024-01-15T10:30:00Z",
654
691
.ops = &.{.{
655
692
.action = .delete,
656
693
.collection = "app.bsky.feed.post",
···
663
700
const decoded = try decodeFrame(alloc, frame);
664
701
665
702
try std.testing.expectEqual(@as(i64, 500), decoded.commit.seq);
703
+
try std.testing.expectEqualStrings("3k2xyz000000", decoded.commit.rev);
704
+
try std.testing.expectEqualStrings("2024-01-15T10:30:00Z", decoded.commit.time);
666
705
try std.testing.expectEqual(@as(usize, 1), decoded.commit.ops.len);
667
706
try std.testing.expectEqual(CommitAction.delete, decoded.commit.ops[0].action);
707
+
try std.testing.expect(decoded.commit.ops[0].cid == null);
668
708
try std.testing.expect(decoded.commit.ops[0].record == null);
669
709
}