diff --git a/build.zig b/build.zig index b64ca9f..50247c5 100644 --- a/build.zig +++ b/build.zig @@ -77,6 +77,23 @@ pub fn build(b: *std.Build) void { const smoke_step = b.step("smoke", "run jetstream smoke test"); smoke_step.dependOn(&run_smoke.step); + // firehose smoke test (CBOR + CAR + CID on live data) + const firehose_smoke = b.addExecutable(.{ + .name = "firehose-smoke", + .root_module = b.createModule(.{ + .root_source_file = b.path("scripts/firehose_smoke.zig"), + .target = target, + .optimize = optimize, + .link_libc = true, + .imports = &.{.{ .name = "zat", .module = mod }}, + }), + }); + b.installArtifact(firehose_smoke); + + const run_firehose_smoke = b.addRunArtifact(firehose_smoke); + const firehose_smoke_step = b.step("firehose-smoke", "run firehose smoke test (CBOR/CAR/CID on live data)"); + firehose_smoke_step.dependOn(&run_firehose_smoke.step); + // CBOR codec benchmarks const cbor_bench = b.addExecutable(.{ .name = "cbor-bench", diff --git a/justfile b/justfile index 15f8b12..8b4e5df 100644 --- a/justfile +++ b/justfile @@ -19,3 +19,7 @@ test: # run CBOR codec benchmarks bench: zig build bench -Doptimize=ReleaseFast + +# run firehose smoke test (CBOR/CAR/CID on live production data) +firehose-smoke: + zig build firehose-smoke -Doptimize=ReleaseFast && ./zig-out/bin/firehose-smoke diff --git a/scripts/firehose_smoke.zig b/scripts/firehose_smoke.zig new file mode 100644 index 0000000..f78bbd3 --- /dev/null +++ b/scripts/firehose_smoke.zig @@ -0,0 +1,71 @@ +//! firehose smoke test — connects to the live AT Protocol firehose, +//! decodes CBOR frames, parses CAR blocks, and verifies CIDs. +//! exercises the full CBOR → CAR → CID pipeline on production data. +//! +//! run: just firehose-smoke + +const std = @import("std"); +const zat = @import("zat"); + +pub fn main() !void { + var da: std.heap.DebugAllocator(.{}) = .init; + defer _ = da.deinit(); + const allocator = da.allocator(); + + std.debug.print("firehose smoke test starting (CBOR + CAR + CID on live data)\n", .{}); + + var handler = Handler{}; + var client = zat.FirehoseClient.init(std.Options.debug_io, allocator, .{}); + try client.subscribe(&handler); +} + +const Handler = struct { + count: u64 = 0, + commits: u64 = 0, + records: u64 = 0, + connects: u64 = 0, + errors: u64 = 0, + + pub fn onEvent(self: *Handler, event: zat.FirehoseEvent) void { + self.count += 1; + + switch (event) { + .commit => |commit| { + self.commits += 1; + for (commit.ops) |op| { + if (op.record) |record| { + // record was decoded from CAR blocks via CBOR + _ = record.getString("$type"); + self.records += 1; + } + } + }, + else => {}, + } + + if (self.count % 1000 == 0) { + std.debug.print(" [{d}] commits={d} records={d} errors={d}\n", .{ + self.count, self.commits, self.records, self.errors, + }); + } + + // stop after 10k events + if (self.count >= 10000) { + std.debug.print("\nfirehose smoke test PASSED\n", .{}); + std.debug.print(" {d} events, {d} commits, {d} records decoded, {d} errors\n", .{ + self.count, self.commits, self.records, self.errors, + }); + std.process.exit(0); + } + } + + pub fn onConnect(self: *Handler, host: []const u8) void { + self.connects += 1; + std.debug.print("CONNECT #{d} to {s}\n", .{ self.connects, host }); + } + + pub fn onError(self: *Handler, err: anyerror) void { + self.errors += 1; + std.debug.print("ERROR: {s}\n", .{@errorName(err)}); + } +};