Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265//! subscribe filter options//!//! v1 (/subscribe): wantedCollections / wantedDids, quirks included//! (docs/upstream-subscribe-protocol.md): prefix patterns get no NSID//! validation of the head, caps fire post-dedupe, identity/account events//! bypass wantedCollections, empty-collection commits bypass the filter.//!//! v2 (network.bsky.jetstream.subscribeEvents): kinds / dids / collections —//! three orthogonal AND-composed predicates, each match-all when omitted.//! collections constrains only commit events; kinds is validated against the//! lexicon enum; the legacy v1 names are rejected upstream of this parser.
const std = @import("std");const zat = @import("zat");const wire = @import("wire.zig");
const Allocator = std.mem.Allocator;
pub const max_wanted_collections = 100;pub const max_wanted_dids = 10_000;
pub const ParseError = error{ InvalidCollection, TooManyCollections, InvalidDid, InvalidEncoding, InvalidKind, CollectionsWithoutCommits, TooManyDids, OutOfMemory,};
pub const KindSet = std.enums.EnumSet(wire.Kind);
pub const Filter = struct { /// exact collection NSIDs; null means match-all (no collection filter) collections: ?std.StringHashMapUnmanaged(void) = null, /// prefixes ending in '.', e.g. "app.bsky." prefixes: std.ArrayList([]const u8) = .empty, /// wanted DIDs; null means match-all dids: ?std.StringHashMapUnmanaged(void) = null, /// v2 kinds predicate; full set means match-all (v1 always full) kinds: KindSet = .full, arena: std.heap.ArenaAllocator,
pub fn init(allocator: Allocator) Filter { return .{ .arena = std.heap.ArenaAllocator.init(allocator) }; }
pub fn deinit(self: *Filter) void { self.arena.deinit(); }
/// parse the query string of a /subscribe URL (the part after '?'). pub fn parseQuery(allocator: Allocator, query: []const u8) ParseError!Filter { var f = Filter.init(allocator); errdefer f.deinit(); var it = std.mem.splitScalar(u8, query, '&'); while (it.next()) |pair| { if (pair.len == 0) continue; const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue; const key = pair[0..eq]; const value = try percentDecodeAlloc(f.arena.allocator(), pair[eq + 1 ..]); if (std.mem.eql(u8, key, "wantedCollections")) { try f.addCollection(value); } else if (std.mem.eql(u8, key, "wantedDids")) { try f.addDid(value); } } return f; }
/// parse a network.bsky.jetstream.subscribeEvents query: kinds / dids / /// collections. an out-of-enum kind and a collections filter that can /// never apply (kinds present but excluding commit) are both errors — /// the caller maps them to pre-upgrade HTTP 400s. pub fn parseQueryV2(allocator: Allocator, query: []const u8) ParseError!Filter { var f = Filter.init(allocator); errdefer f.deinit(); var saw_kinds = false; var saw_collections = false; var kinds = KindSet.empty; var it = std.mem.splitScalar(u8, query, '&'); while (it.next()) |pair| { if (pair.len == 0) continue; const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue; const key = pair[0..eq]; const value = try percentDecodeAlloc(f.arena.allocator(), pair[eq + 1 ..]); if (std.mem.eql(u8, key, "collections")) { saw_collections = true; try f.addCollection(value); } else if (std.mem.eql(u8, key, "dids")) { try f.addDid(value); } else if (std.mem.eql(u8, key, "kinds")) { saw_kinds = true; const kind = std.meta.stringToEnum(wire.Kind, value) orelse return error.InvalidKind; kinds.insert(kind); } } if (saw_kinds) { if (saw_collections and !kinds.contains(.commit)) return error.CollectionsWithoutCommits; f.kinds = kinds; } return f; }
pub fn addCollection(self: *Filter, raw: []const u8) ParseError!void { const alloc = self.arena.allocator(); if (std.mem.endsWith(u8, raw, ".*")) { // prefix pattern: trim the '*', keep the trailing '.' // v1 quirk: no NSID validation of the head const prefix = raw[0 .. raw.len - 1]; for (self.prefixes.items) |p| { if (std.mem.eql(u8, p, prefix)) return; // dedupe } try self.prefixes.append(alloc, try alloc.dupe(u8, prefix)); } else { if (zat.Nsid.parse(raw) == null) return error.InvalidCollection; if (self.collections == null) self.collections = .empty; const gop = self.collections.?.getOrPut(alloc, raw) catch return error.OutOfMemory; if (!gop.found_existing) gop.key_ptr.* = alloc.dupe(u8, raw) catch return error.OutOfMemory; } // cap fires post-dedupe on the unique count const unique = (if (self.collections) |c| c.count() else 0) + self.prefixes.items.len; if (unique > max_wanted_collections) return error.TooManyCollections; }
pub fn addDid(self: *Filter, raw: []const u8) ParseError!void { const alloc = self.arena.allocator(); if (zat.Did.parse(raw) == null) return error.InvalidDid; if (self.dids == null) self.dids = .empty; const gop = self.dids.?.getOrPut(alloc, raw) catch return error.OutOfMemory; if (!gop.found_existing) gop.key_ptr.* = alloc.dupe(u8, raw) catch return error.OutOfMemory; if (self.dids.?.count() > max_wanted_dids) return error.TooManyDids; }
/// does this filter want a frame? pub fn wants(self: *const Filter, kind: wire.Kind, did: []const u8, collection: []const u8) bool { if (!self.kinds.contains(kind)) return false; if (self.dids) |dids| { if (!dids.contains(did)) return false; } // identity/account/sync events bypass wantedCollections entirely; // commits with an empty collection also bypass the filter if (kind != .commit or collection.len == 0) return true; if (self.collections == null and self.prefixes.items.len == 0) return true; if (self.collections) |c| { if (c.contains(collection)) return true; } for (self.prefixes.items) |prefix| { if (std.mem.startsWith(u8, collection, prefix)) return true; } return false; }};
// === tests ===
const testing = std.testing;
test "match-all when no params" { var f = try Filter.parseQuery(testing.allocator, ""); defer f.deinit(); try testing.expect(f.wants(.commit, "did:plc:x", "app.bsky.feed.post")); try testing.expect(f.wants(.identity, "did:plc:x", ""));}
fn percentDecodeAlloc(allocator: Allocator, input: []const u8) ParseError![]const u8 { const out = allocator.alloc(u8, input.len) catch return error.OutOfMemory; var read: usize = 0; var write: usize = 0; while (read < input.len) { switch (input[read]) { '%' => { if (read + 2 >= input.len) return error.InvalidEncoding; out[write] = std.fmt.parseInt(u8, input[read + 1 .. read + 3], 16) catch return error.InvalidEncoding; read += 3; }, '+' => { out[write] = ' '; read += 1; }, else => |c| { out[write] = c; read += 1; }, } write += 1; } return out[0..write];}
test "v2 query: kinds, dids, collections compose as AND predicates" { var f = try Filter.parseQueryV2(testing.allocator, "kinds=commit&kinds=account&collections=app.bsky.feed.post&collections=sh.tangled.*"); defer f.deinit(); try testing.expect(f.wants(.commit, "did:plc:x", "app.bsky.feed.post")); try testing.expect(f.wants(.commit, "did:plc:x", "sh.tangled.repo")); try testing.expect(!f.wants(.commit, "did:plc:x", "app.bsky.graph.follow")); // collections constrains only commits; account passes because kinds allows it try testing.expect(f.wants(.account, "did:plc:x", "")); // identity excluded by kinds even though collections would not apply try testing.expect(!f.wants(.identity, "did:plc:x", ""));
try testing.expectError(error.InvalidKind, Filter.parseQueryV2(testing.allocator, "kinds=bogus")); try testing.expectError(error.CollectionsWithoutCommits, Filter.parseQueryV2(testing.allocator, "kinds=identity&collections=app.bsky.feed.post"));
// omitted kinds = match-all, sync included (v2 emits sync) var all = try Filter.parseQueryV2(testing.allocator, ""); defer all.deinit(); try testing.expect(all.wants(.sync, "did:plc:x", ""));}
test "collection exact and prefix matching" { var f = try Filter.parseQuery(testing.allocator, "wantedCollections=app.bsky.feed.post&wantedCollections=sh.tangled.*"); defer f.deinit(); try testing.expect(f.wants(.commit, "d", "app.bsky.feed.post")); try testing.expect(!f.wants(.commit, "d", "app.bsky.feed.like")); try testing.expect(f.wants(.commit, "d", "sh.tangled.repo")); // identity/account bypass collections try testing.expect(f.wants(.identity, "d", "")); try testing.expect(f.wants(.account, "d", "")); // empty-collection commit bypasses the filter try testing.expect(f.wants(.commit, "d", ""));}
test "two-segment prefix is accepted (v1 quirk)" { var f = Filter.init(testing.allocator); defer f.deinit(); try f.addCollection("app.bsky.*"); try testing.expect(f.wants(.commit, "d", "app.bsky.feed.post"));}
test "invalid collection and did rejected" { var f = Filter.init(testing.allocator); defer f.deinit(); try testing.expectError(error.InvalidCollection, f.addCollection("not a nsid")); try testing.expectError(error.InvalidCollection, f.addCollection("app.bsky.fo*")); try testing.expectError(error.InvalidDid, f.addDid("bob"));}
test "query values are percent decoded" { var f = try Filter.parseQuery(testing.allocator, "wantedDids=did%3Aplc%3Aabc&wantedCollections=app.bsky.feed.post"); defer f.deinit(); try testing.expect(f.wants(.commit, "did:plc:abc", "app.bsky.feed.post")); try testing.expect(!f.wants(.commit, "did:plc:other", "app.bsky.feed.post"));}
test "did filtering applies to all kinds" { var f = try Filter.parseQuery(testing.allocator, "wantedDids=did:plc:aaa"); defer f.deinit(); try testing.expect(f.wants(.commit, "did:plc:aaa", "c")); try testing.expect(!f.wants(.commit, "did:plc:bbb", "c")); try testing.expect(!f.wants(.identity, "did:plc:bbb", ""));}
test "caps fire post-dedupe" { var f = Filter.init(testing.allocator); defer f.deinit(); // the same collection many times never trips the cap var i: usize = 0; while (i < 200) : (i += 1) try f.addCollection("app.bsky.feed.post"); try testing.expectEqual(@as(u32, 1), f.collections.?.count());}