rickub
nandi/gleanpublic Fork
⑂ c5a7a68
Branches
claude/modal-container-min-76527a main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/glean.git
git clone ssh://git@rickub.com/nandi/glean.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

Download ZIP

Add a Jetstream firehose consumer

Reads ATProto commits as JSON over a websocket. Jetstream is public, so
unlike the OAuth and XRPC work this is verified end to end against the live
network: 17/17.

Two capabilities are confirmed rather than assumed, because both fail
invisibly. Server-side filtering: the probe subscribes to one collection and
asserts nothing else arrives, since an unfiltered subscription is the entire
network and a consumer that lost its query parameters would still look
healthy. Cursor replay: rewinding must actually rewind rather than quietly
resume live, so the probe replays from a cursor it saw and asserts the
events overlap what it already received -- the parameter being accepted
proves nothing.

Two traps worth recording. websock's Uri overload forwards only uri.path and
drops the query string, which for Jetstream means losing every filter, so
connect uses the host/path form and assembles path-plus-query itself. And
wantedCollections must be repeated per value rather than comma-joined; a
joined value matches no collection.

The seemingly unused `import chronicles` is load-bearing: websock's logging
does not compile unless the stream symbol resolves in the importing module's
scope.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
nandithebull committed 2026-09-20T21:55:33-07:00 Browse files
c5a7a68 parent: 72af361
modified spike/nim/.gitignore +1 -0
@@ -4,3 +4,4 @@ sqlite_probe
44 dpop_probe
55 oauth_probe
66 xrpc_probe
7+jetstream_probe
@@ -4,3 +4,4 @@ sqlite_probe
4 dpop_probe4 dpop_probe
5 oauth_probe5 oauth_probe
6 xrpc_probe6 xrpc_probe
7+jetstream_probe
added spike/nim/atproto/jetstream.nim +184 -0
new file mode 100644
@@ -0,0 +1,184 @@
1+## Jetstream consumer: the ATProto firehose as JSON over a websocket.
2+##
3+## Jetstream is a public, unauthenticated feed of repository commits, so this
4+## is the one part of the port that can be verified end to end without a
5+## consented session.
6+##
7+## Filtering happens server-side via query parameters (wantedCollections,
8+## wantedDids), which matters: the unfiltered firehose is the entire network.
9+## A subscription that silently loses its filters still delivers events, so a
10+## consumer that looks healthy can be reading the wrong stream entirely --
11+## see the note on websock's Uri overload in `connect`.
12+
13+import std/[json, strutils, uri]
14+import pkg/chronicles
15+import pkg/chronos
16+import pkg/websock/websock
17+
18+type
19+ JetstreamError* = object of CatchableError
20+
21+ EventKind* = enum
22+ ekCommit = "commit"
23+ ekIdentity = "identity"
24+ ekAccount = "account"
25+ ekUnknown = "unknown"
26+
27+ CommitOperation* = enum
28+ coCreate = "create"
29+ coUpdate = "update"
30+ coDelete = "delete"
31+
32+ Commit* = object
33+ rev*: string
34+ operation*: CommitOperation
35+ collection*: string
36+ rkey*: string
37+ cid*: string
38+ record*: JsonNode ## nil for a delete
39+
40+ Event* = object
41+ did*: string
42+ timeUs*: int64 ## microseconds; also the replay cursor
43+ kind*: EventKind
44+ commit*: Commit ## meaningful only when kind is ekCommit
45+ raw*: JsonNode
46+
47+ JetstreamConfig* = object
48+ url*: string
49+ wantedCollections*: seq[string]
50+ wantedDids*: seq[string]
51+ cursor*: int64 ## 0 to start live
52+ userAgent*: string
53+
54+ EventHandler* = proc(ev: Event) {.gcsafe, raises: [].}
55+
56+const
57+ DefaultJetstream* = "wss://jetstream1.us-east.bsky.network"
58+ DefaultUserAgent* = "glean-nim/0.1"
59+
60+proc bytesToString(bs: openArray[byte]): string =
61+ ## websock hands back seq[byte]; Jetstream frames are UTF-8 JSON text.
62+ result = newString(bs.len)
63+ if bs.len > 0:
64+ copyMem(result[0].addr, bs[0].unsafeAddr, bs.len)
65+
66+proc parseEventKind(s: string): EventKind =
67+ case s
68+ of "commit": ekCommit
69+ of "identity": ekIdentity
70+ of "account": ekAccount
71+ else: ekUnknown
72+
73+proc parseOperation(s: string): CommitOperation =
74+ case s
75+ of "create": coCreate
76+ of "update": coUpdate
77+ of "delete": coDelete
78+ else: coCreate
79+
80+proc parseEvent*(raw: string): Event =
81+ ## Decode one Jetstream frame.
82+ let j =
83+ try: parseJson(raw)
84+ except CatchableError:
85+ raise newException(JetstreamError, "frame is not valid JSON")
86+
87+ result = Event(
88+ did: j{"did"}.getStr,
89+ timeUs: j{"time_us"}.getBiggestInt.int64,
90+ kind: parseEventKind(j{"kind"}.getStr),
91+ raw: j,
92+ )
93+
94+ if result.kind == ekCommit and j.hasKey("commit"):
95+ let c = j["commit"]
96+ result.commit = Commit(
97+ rev: c{"rev"}.getStr,
98+ operation: parseOperation(c{"operation"}.getStr),
99+ collection: c{"collection"}.getStr,
100+ rkey: c{"rkey"}.getStr,
101+ cid: c{"cid"}.getStr,
102+ record: c{"record"},
103+ )
104+
105+proc subscribeUrl*(cfg: JetstreamConfig): string =
106+ ## Build the /subscribe URL, including filters.
107+ var base = cfg.url.strip(chars = {'/'})
108+ if not base.endsWith("/subscribe"):
109+ base &= "/subscribe"
110+
111+ var params: seq[(string, string)]
112+ # Repeated keys, not a comma-joined list: Jetstream reads these as a
113+ # multi-valued parameter.
114+ for c in cfg.wantedCollections:
115+ params.add ("wantedCollections", c)
116+ for d in cfg.wantedDids:
117+ params.add ("wantedDids", d)
118+ if cfg.cursor > 0:
119+ params.add ("cursor", $cfg.cursor)
120+
121+ if params.len > 0: base & "?" & params.encodeQuery else: base
122+
123+proc connect*(cfg: JetstreamConfig): Future[WSSession] {.async.} =
124+ ## Open the websocket.
125+ ##
126+ ## Deliberately not websock's `Uri` overload: it forwards only `uri.path`
127+ ## and drops `uri.query`, which for Jetstream means losing every filter and
128+ ## silently subscribing to the entire network firehose instead of the
129+ ## handful of collections asked for.
130+ let full = subscribeUrl(cfg)
131+ let u = parseUri(full)
132+ if u.scheme notin ["ws", "wss"]:
133+ raise newException(JetstreamError, "not a websocket URL: " & full)
134+
135+ let
136+ secure = u.scheme == "wss"
137+ port = if u.port.len > 0: u.port else: (if secure: "443" else: "80")
138+ pathAndQuery = (if u.path.len > 0: u.path else: "/") &
139+ (if u.query.len > 0: "?" & u.query else: "")
140+
141+ await WebSocket.connect(
142+ host = u.hostname & ":" & port,
143+ path = pathAndQuery,
144+ hostName = u.hostname,
145+ secure = secure,
146+ )
147+
148+proc readEvents*(ws: WSSession, handler: EventHandler,
149+ maxEvents = 0) {.async.} =
150+ ## Pump frames until the socket closes, or until `maxEvents` have been
151+ ## handled when that is non-zero.
152+ var seen = 0
153+ while ws.readyState != ReadyState.Closed:
154+ let frame = await ws.recvMsg()
155+ if frame.len == 0:
156+ # A zero-length read means the peer closed; recvMsg does not raise.
157+ break
158+ var ev: Event
159+ try:
160+ ev = parseEvent(bytesToString(frame))
161+ except JetstreamError:
162+ # One malformed frame should not end a long-lived subscription.
163+ continue
164+ handler(ev)
165+ inc seen
166+ if maxEvents > 0 and seen >= maxEvents:
167+ break
168+
169+proc collectEvents*(cfg: JetstreamConfig, count: int,
170+ timeout = 30.seconds): Future[seq[Event]] {.async.} =
171+ ## Connect, gather `count` events, and close. For probes and tests.
172+ var got: seq[Event]
173+ let ws = await connect(cfg)
174+ defer:
175+ try: await ws.close()
176+ except CatchableError: discard
177+
178+ proc onEvent(ev: Event) {.gcsafe, raises: [].} =
179+ got.add ev
180+
181+ let pump = readEvents(ws, onEvent, maxEvents = count)
182+ if not await pump.withTimeout(timeout):
183+ await pump.cancelAndWait()
184+ got
new file mode 100644
@@ -0,0 +1,184 @@
1+## Jetstream consumer: the ATProto firehose as JSON over a websocket.
2+##
3+## Jetstream is a public, unauthenticated feed of repository commits, so this
4+## is the one part of the port that can be verified end to end without a
5+## consented session.
6+##
7+## Filtering happens server-side via query parameters (wantedCollections,
8+## wantedDids), which matters: the unfiltered firehose is the entire network.
9+## A subscription that silently loses its filters still delivers events, so a
10+## consumer that looks healthy can be reading the wrong stream entirely --
11+## see the note on websock's Uri overload in `connect`.
12+
13+import std/[json, strutils, uri]
14+import pkg/chronicles
15+import pkg/chronos
16+import pkg/websock/websock
17+
18+type
19+ JetstreamError* = object of CatchableError
20+
21+ EventKind* = enum
22+ ekCommit = "commit"
23+ ekIdentity = "identity"
24+ ekAccount = "account"
25+ ekUnknown = "unknown"
26+
27+ CommitOperation* = enum
28+ coCreate = "create"
29+ coUpdate = "update"
30+ coDelete = "delete"
31+
32+ Commit* = object
33+ rev*: string
34+ operation*: CommitOperation
35+ collection*: string
36+ rkey*: string
37+ cid*: string
38+ record*: JsonNode ## nil for a delete
39+
40+ Event* = object
41+ did*: string
42+ timeUs*: int64 ## microseconds; also the replay cursor
43+ kind*: EventKind
44+ commit*: Commit ## meaningful only when kind is ekCommit
45+ raw*: JsonNode
46+
47+ JetstreamConfig* = object
48+ url*: string
49+ wantedCollections*: seq[string]
50+ wantedDids*: seq[string]
51+ cursor*: int64 ## 0 to start live
52+ userAgent*: string
53+
54+ EventHandler* = proc(ev: Event) {.gcsafe, raises: [].}
55+
56+const
57+ DefaultJetstream* = "wss://jetstream1.us-east.bsky.network"
58+ DefaultUserAgent* = "glean-nim/0.1"
59+
60+proc bytesToString(bs: openArray[byte]): string =
61+ ## websock hands back seq[byte]; Jetstream frames are UTF-8 JSON text.
62+ result = newString(bs.len)
63+ if bs.len > 0:
64+ copyMem(result[0].addr, bs[0].unsafeAddr, bs.len)
65+
66+proc parseEventKind(s: string): EventKind =
67+ case s
68+ of "commit": ekCommit
69+ of "identity": ekIdentity
70+ of "account": ekAccount
71+ else: ekUnknown
72+
73+proc parseOperation(s: string): CommitOperation =
74+ case s
75+ of "create": coCreate
76+ of "update": coUpdate
77+ of "delete": coDelete
78+ else: coCreate
79+
80+proc parseEvent*(raw: string): Event =
81+ ## Decode one Jetstream frame.
82+ let j =
83+ try: parseJson(raw)
84+ except CatchableError:
85+ raise newException(JetstreamError, "frame is not valid JSON")
86+
87+ result = Event(
88+ did: j{"did"}.getStr,
89+ timeUs: j{"time_us"}.getBiggestInt.int64,
90+ kind: parseEventKind(j{"kind"}.getStr),
91+ raw: j,
92+ )
93+
94+ if result.kind == ekCommit and j.hasKey("commit"):
95+ let c = j["commit"]
96+ result.commit = Commit(
97+ rev: c{"rev"}.getStr,
98+ operation: parseOperation(c{"operation"}.getStr),
99+ collection: c{"collection"}.getStr,
100+ rkey: c{"rkey"}.getStr,
101+ cid: c{"cid"}.getStr,
102+ record: c{"record"},
103+ )
104+
105+proc subscribeUrl*(cfg: JetstreamConfig): string =
106+ ## Build the /subscribe URL, including filters.
107+ var base = cfg.url.strip(chars = {'/'})
108+ if not base.endsWith("/subscribe"):
109+ base &= "/subscribe"
110+
111+ var params: seq[(string, string)]
112+ # Repeated keys, not a comma-joined list: Jetstream reads these as a
113+ # multi-valued parameter.
114+ for c in cfg.wantedCollections:
115+ params.add ("wantedCollections", c)
116+ for d in cfg.wantedDids:
117+ params.add ("wantedDids", d)
118+ if cfg.cursor > 0:
119+ params.add ("cursor", $cfg.cursor)
120+
121+ if params.len > 0: base & "?" & params.encodeQuery else: base
122+
123+proc connect*(cfg: JetstreamConfig): Future[WSSession] {.async.} =
124+ ## Open the websocket.
125+ ##
126+ ## Deliberately not websock's `Uri` overload: it forwards only `uri.path`
127+ ## and drops `uri.query`, which for Jetstream means losing every filter and
128+ ## silently subscribing to the entire network firehose instead of the
129+ ## handful of collections asked for.
130+ let full = subscribeUrl(cfg)
131+ let u = parseUri(full)
132+ if u.scheme notin ["ws", "wss"]:
133+ raise newException(JetstreamError, "not a websocket URL: " & full)
134+
135+ let
136+ secure = u.scheme == "wss"
137+ port = if u.port.len > 0: u.port else: (if secure: "443" else: "80")
138+ pathAndQuery = (if u.path.len > 0: u.path else: "/") &
139+ (if u.query.len > 0: "?" & u.query else: "")
140+
141+ await WebSocket.connect(
142+ host = u.hostname & ":" & port,
143+ path = pathAndQuery,
144+ hostName = u.hostname,
145+ secure = secure,
146+ )
147+
148+proc readEvents*(ws: WSSession, handler: EventHandler,
149+ maxEvents = 0) {.async.} =
150+ ## Pump frames until the socket closes, or until `maxEvents` have been
151+ ## handled when that is non-zero.
152+ var seen = 0
153+ while ws.readyState != ReadyState.Closed:
154+ let frame = await ws.recvMsg()
155+ if frame.len == 0:
156+ # A zero-length read means the peer closed; recvMsg does not raise.
157+ break
158+ var ev: Event
159+ try:
160+ ev = parseEvent(bytesToString(frame))
161+ except JetstreamError:
162+ # One malformed frame should not end a long-lived subscription.
163+ continue
164+ handler(ev)
165+ inc seen
166+ if maxEvents > 0 and seen >= maxEvents:
167+ break
168+
169+proc collectEvents*(cfg: JetstreamConfig, count: int,
170+ timeout = 30.seconds): Future[seq[Event]] {.async.} =
171+ ## Connect, gather `count` events, and close. For probes and tests.
172+ var got: seq[Event]
173+ let ws = await connect(cfg)
174+ defer:
175+ try: await ws.close()
176+ except CatchableError: discard
177+
178+ proc onEvent(ev: Event) {.gcsafe, raises: [].} =
179+ got.add ev
180+
181+ let pump = readEvents(ws, onEvent, maxEvents = count)
182+ if not await pump.withTimeout(timeout):
183+ await pump.cancelAndWait()
184+ got
added spike/nim/jetstream_probe.nim +150 -0
new file mode 100644
@@ -0,0 +1,150 @@
1+## Reads the live ATProto firehose.
2+##
3+## nim c -r -d:ssl jetstream_probe.nim
4+##
5+## Jetstream is public, so unlike the OAuth and XRPC probes this one can check
6+## the real thing end to end. The point of interest is server-side filtering:
7+## an unfiltered subscription is the whole network, so a consumer that loses
8+## its query parameters still receives a healthy-looking torrent of events
9+## that are simply the wrong ones.
10+
11+import std/[json, sequtils, sets, strformat, strutils, tables]
12+import pkg/chronos
13+import atproto/jetstream
14+
15+var failures = 0
16+
17+proc report(name: string, ok: bool, detail = "") =
18+ if not ok: inc failures
19+ let label = if ok: "PASS" else: "FAIL"
20+ if detail.len > 0: echo &" {label} {name} -- {detail}"
21+ else: echo &" {label} {name}"
22+
23+proc main() {.async.} =
24+ echo "Jetstream probe"
25+ echo ""
26+
27+ # --- URL construction ---------------------------------------------------
28+ block:
29+ let cfg = JetstreamConfig(
30+ url: DefaultJetstream,
31+ wantedCollections: @["app.bsky.feed.post", "app.bsky.feed.like"],
32+ cursor: 123,
33+ )
34+ let u = subscribeUrl(cfg)
35+ report("adds /subscribe", "/subscribe" in u)
36+ # Repeated keys, not one comma-joined value: Jetstream reads these as a
37+ # multi-valued parameter and a joined string matches no collection.
38+ report("repeats wantedCollections per value",
39+ u.count("wantedCollections=") == 2, u.split('?')[^1])
40+ report("carries the cursor", "cursor=123" in u)
41+
42+ block:
43+ let cfg = JetstreamConfig(url: "wss://example.test/subscribe")
44+ report("does not double up /subscribe",
45+ subscribeUrl(cfg).count("/subscribe") == 1)
46+
47+ # --- parsing ------------------------------------------------------------
48+ block:
49+ let raw = """{"did":"did:plc:abc","time_us":1720000000000000,
50+ "kind":"commit","commit":{"rev":"3k","operation":"create",
51+ "collection":"app.bsky.feed.post","rkey":"xyz","cid":"bafy",
52+ "record":{"text":"hi","$type":"app.bsky.feed.post"}}}"""
53+ let ev = parseEvent(raw)
54+ report("parses a commit event", ev.kind == ekCommit and ev.did == "did:plc:abc")
55+ report("parses the operation", ev.commit.operation == coCreate)
56+ report("parses the record", ev.commit.record{"text"}.getStr == "hi")
57+ report("time_us survives as int64", ev.timeUs == 1720000000000000'i64,
58+ $ev.timeUs)
59+
60+ block:
61+ # A delete carries no record; reading one must not crash a consumer.
62+ let ev = parseEvent("""{"did":"did:plc:a","time_us":1,"kind":"commit",
63+ "commit":{"operation":"delete","collection":"c","rkey":"r"}}""")
64+ report("a delete has no record",
65+ ev.commit.operation == coDelete and
66+ (ev.commit.record == nil or ev.commit.record.kind == JNull))
67+
68+ # --- the live firehose --------------------------------------------------
69+ const wanted = "app.bsky.feed.post"
70+ var events: seq[Event]
71+ try:
72+ let cfg = JetstreamConfig(
73+ url: DefaultJetstream,
74+ wantedCollections: @[wanted],
75+ )
76+ events = await collectEvents(cfg, count = 25, timeout = 30.seconds)
77+ except CatchableError as e:
78+ report("connected to the live firehose", false, e.msg)
79+ echo ""
80+ echo "RESULT: could not reach Jetstream."
81+ quit 1
82+
83+ report("received live events", events.len > 0, &"{events.len} events")
84+ if events.len == 0:
85+ echo ""
86+ echo "RESULT: no events received."
87+ quit 1
88+
89+ report("every event has a DID", events.allIt(it.did.startsWith("did:")))
90+ report("cursors advance",
91+ events[^1].timeUs >= events[0].timeUs,
92+ &"{events[0].timeUs} -> {events[^1].timeUs}")
93+
94+ # The real check: server-side filtering actually applied. Without it this
95+ # would be a mix of likes, follows, reposts and everything else.
96+ var collections: CountTable[string]
97+ for ev in events:
98+ if ev.kind == ekCommit:
99+ collections.inc ev.commit.collection
100+ let others = toSeq(collections.keys).filterIt(it != wanted and it.len > 0)
101+ report("filter applied server-side (only " & wanted & ")",
102+ others.len == 0,
103+ if others.len == 0: $collections.len & " collection"
104+ else: "also saw " & others.join(", "))
105+
106+ let posts = events.filterIt(
107+ it.kind == ekCommit and it.commit.collection == wanted and
108+ it.commit.operation == coCreate and it.commit.record != nil)
109+ report("create events carry a typed record",
110+ posts.len == 0 or posts[0].commit.record{"$type"}.getStr == wanted,
111+ if posts.len > 0: posts[0].commit.record{"$type"}.getStr else: "none seen")
112+
113+ # --- cursor replay ------------------------------------------------------
114+ # This is how a consumer catches up after downtime, so "the cursor is
115+ # accepted" is not enough -- it has to actually rewind. Replaying from an
116+ # earlier cursor must return events at or after it, including ones already
117+ # seen live.
118+ block:
119+ let rewindTo = events[0].timeUs
120+ let seenLive = events.mapIt(it.timeUs).toHashSet
121+ try:
122+ let cfg = JetstreamConfig(
123+ url: DefaultJetstream,
124+ wantedCollections: @[wanted],
125+ cursor: rewindTo,
126+ )
127+ let replayed = await collectEvents(cfg, count = 25, timeout = 30.seconds)
128+ report("replay from a cursor returns events", replayed.len > 0,
129+ &"{replayed.len} events")
130+ if replayed.len > 0:
131+ report("replayed events start at or after the cursor",
132+ replayed[0].timeUs >= rewindTo,
133+ &"cursor {rewindTo} -> first {replayed[0].timeUs}")
134+ # Proof it rewound rather than resuming live: the replay should
135+ # re-deliver timestamps from the first batch.
136+ let overlap = replayed.countIt(it.timeUs in seenLive)
137+ report("replay re-delivers already-seen events (a real rewind)",
138+ overlap > 0, &"{overlap} of {replayed.len} overlap")
139+ except CatchableError as e:
140+ report("replay from a cursor returns events", false, e.msg)
141+
142+ echo ""
143+ if failures == 0:
144+ echo "RESULT: Jetstream works, with server-side filtering and replay confirmed."
145+ else:
146+ echo &"RESULT: {failures} check(s) failed."
147+ quit 1
148+
149+when isMainModule:
150+ waitFor main()
new file mode 100644
@@ -0,0 +1,150 @@
1+## Reads the live ATProto firehose.
2+##
3+## nim c -r -d:ssl jetstream_probe.nim
4+##
5+## Jetstream is public, so unlike the OAuth and XRPC probes this one can check
6+## the real thing end to end. The point of interest is server-side filtering:
7+## an unfiltered subscription is the whole network, so a consumer that loses
8+## its query parameters still receives a healthy-looking torrent of events
9+## that are simply the wrong ones.
10+
11+import std/[json, sequtils, sets, strformat, strutils, tables]
12+import pkg/chronos
13+import atproto/jetstream
14+
15+var failures = 0
16+
17+proc report(name: string, ok: bool, detail = "") =
18+ if not ok: inc failures
19+ let label = if ok: "PASS" else: "FAIL"
20+ if detail.len > 0: echo &" {label} {name} -- {detail}"
21+ else: echo &" {label} {name}"
22+
23+proc main() {.async.} =
24+ echo "Jetstream probe"
25+ echo ""
26+
27+ # --- URL construction ---------------------------------------------------
28+ block:
29+ let cfg = JetstreamConfig(
30+ url: DefaultJetstream,
31+ wantedCollections: @["app.bsky.feed.post", "app.bsky.feed.like"],
32+ cursor: 123,
33+ )
34+ let u = subscribeUrl(cfg)
35+ report("adds /subscribe", "/subscribe" in u)
36+ # Repeated keys, not one comma-joined value: Jetstream reads these as a
37+ # multi-valued parameter and a joined string matches no collection.
38+ report("repeats wantedCollections per value",
39+ u.count("wantedCollections=") == 2, u.split('?')[^1])
40+ report("carries the cursor", "cursor=123" in u)
41+
42+ block:
43+ let cfg = JetstreamConfig(url: "wss://example.test/subscribe")
44+ report("does not double up /subscribe",
45+ subscribeUrl(cfg).count("/subscribe") == 1)
46+
47+ # --- parsing ------------------------------------------------------------
48+ block:
49+ let raw = """{"did":"did:plc:abc","time_us":1720000000000000,
50+ "kind":"commit","commit":{"rev":"3k","operation":"create",
51+ "collection":"app.bsky.feed.post","rkey":"xyz","cid":"bafy",
52+ "record":{"text":"hi","$type":"app.bsky.feed.post"}}}"""
53+ let ev = parseEvent(raw)
54+ report("parses a commit event", ev.kind == ekCommit and ev.did == "did:plc:abc")
55+ report("parses the operation", ev.commit.operation == coCreate)
56+ report("parses the record", ev.commit.record{"text"}.getStr == "hi")
57+ report("time_us survives as int64", ev.timeUs == 1720000000000000'i64,
58+ $ev.timeUs)
59+
60+ block:
61+ # A delete carries no record; reading one must not crash a consumer.
62+ let ev = parseEvent("""{"did":"did:plc:a","time_us":1,"kind":"commit",
63+ "commit":{"operation":"delete","collection":"c","rkey":"r"}}""")
64+ report("a delete has no record",
65+ ev.commit.operation == coDelete and
66+ (ev.commit.record == nil or ev.commit.record.kind == JNull))
67+
68+ # --- the live firehose --------------------------------------------------
69+ const wanted = "app.bsky.feed.post"
70+ var events: seq[Event]
71+ try:
72+ let cfg = JetstreamConfig(
73+ url: DefaultJetstream,
74+ wantedCollections: @[wanted],
75+ )
76+ events = await collectEvents(cfg, count = 25, timeout = 30.seconds)
77+ except CatchableError as e:
78+ report("connected to the live firehose", false, e.msg)
79+ echo ""
80+ echo "RESULT: could not reach Jetstream."
81+ quit 1
82+
83+ report("received live events", events.len > 0, &"{events.len} events")
84+ if events.len == 0:
85+ echo ""
86+ echo "RESULT: no events received."
87+ quit 1
88+
89+ report("every event has a DID", events.allIt(it.did.startsWith("did:")))
90+ report("cursors advance",
91+ events[^1].timeUs >= events[0].timeUs,
92+ &"{events[0].timeUs} -> {events[^1].timeUs}")
93+
94+ # The real check: server-side filtering actually applied. Without it this
95+ # would be a mix of likes, follows, reposts and everything else.
96+ var collections: CountTable[string]
97+ for ev in events:
98+ if ev.kind == ekCommit:
99+ collections.inc ev.commit.collection
100+ let others = toSeq(collections.keys).filterIt(it != wanted and it.len > 0)
101+ report("filter applied server-side (only " & wanted & ")",
102+ others.len == 0,
103+ if others.len == 0: $collections.len & " collection"
104+ else: "also saw " & others.join(", "))
105+
106+ let posts = events.filterIt(
107+ it.kind == ekCommit and it.commit.collection == wanted and
108+ it.commit.operation == coCreate and it.commit.record != nil)
109+ report("create events carry a typed record",
110+ posts.len == 0 or posts[0].commit.record{"$type"}.getStr == wanted,
111+ if posts.len > 0: posts[0].commit.record{"$type"}.getStr else: "none seen")
112+
113+ # --- cursor replay ------------------------------------------------------
114+ # This is how a consumer catches up after downtime, so "the cursor is
115+ # accepted" is not enough -- it has to actually rewind. Replaying from an
116+ # earlier cursor must return events at or after it, including ones already
117+ # seen live.
118+ block:
119+ let rewindTo = events[0].timeUs
120+ let seenLive = events.mapIt(it.timeUs).toHashSet
121+ try:
122+ let cfg = JetstreamConfig(
123+ url: DefaultJetstream,
124+ wantedCollections: @[wanted],
125+ cursor: rewindTo,
126+ )
127+ let replayed = await collectEvents(cfg, count = 25, timeout = 30.seconds)
128+ report("replay from a cursor returns events", replayed.len > 0,
129+ &"{replayed.len} events")
130+ if replayed.len > 0:
131+ report("replayed events start at or after the cursor",
132+ replayed[0].timeUs >= rewindTo,
133+ &"cursor {rewindTo} -> first {replayed[0].timeUs}")
134+ # Proof it rewound rather than resuming live: the replay should
135+ # re-deliver timestamps from the first batch.
136+ let overlap = replayed.countIt(it.timeUs in seenLive)
137+ report("replay re-delivers already-seen events (a real rewind)",
138+ overlap > 0, &"{overlap} of {replayed.len} overlap")
139+ except CatchableError as e:
140+ report("replay from a cursor returns events", false, e.msg)
141+
142+ echo ""
143+ if failures == 0:
144+ echo "RESULT: Jetstream works, with server-side filtering and replay confirmed."
145+ else:
146+ echo &"RESULT: {failures} check(s) failed."
147+ quit 1
148+
149+when isMainModule:
150+ waitFor main()
modified spike/nim/nim.cfg +5 -0
@@ -5,3 +5,8 @@
55 # load-bearing, not an optimisation.
66 --passC:"-DSQLITE_CORE"
77 --passL:"-lsqlite3 -lm"
8+
9+# websock pulls in chronicles, which will not compile unless a logging stream
10+# is configured. This spike has no use for its output, and leaving it on makes
11+# every consumer of websock inherit the requirement.
12+--define:"chronicles_enabled=off"
@@ -5,3 +5,8 @@
5 # load-bearing, not an optimisation.5 # load-bearing, not an optimisation.
6 --passC:"-DSQLITE_CORE"6 --passC:"-DSQLITE_CORE"
7 --passL:"-lsqlite3 -lm"7 --passL:"-lsqlite3 -lm"
8+
9+# websock pulls in chronicles, which will not compile unless a logging stream
10+# is configured. This spike has no use for its output, and leaving it on makes
11+# every consumer of websock inherit the requirement.
12+--define:"chronicles_enabled=off"
modified spike/nim/readme.md +40 -2
@@ -13,6 +13,7 @@ nim c -r sqlite_probe.nim # storage
1313 nim c -r dpop_probe.nim && ./dpop_probe | (cd verify && go run .) # crypto
1414 nim c -r -d:ssl oauth_probe.nim [handle] # the flow
1515 nim c -r -d:ssl xrpc_probe.nim # signing
16+nim c -r -d:ssl jetstream_probe.nim # firehose
1617 ```
1718
1819 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.
@@ -138,6 +139,41 @@ it with a 200 while ignoring the bogus credential entirely — a passing test
138139 that tested nothing. `com.atproto.server.getSession` has to evaluate the
139140 credential.
140141
142+## 5. Jetstream — works, verified against the live firehose
143+
144+`atproto/jetstream.nim` consumes the ATProto firehose as JSON over a
145+websocket, using `websock` (already vendored via chronos). Jetstream is public
146+and unauthenticated, so unlike the OAuth and XRPC probes this one checks the
147+real thing end to end: 17/17 against the live network.
148+
149+Two capabilities that matter to Glean are confirmed rather than assumed:
150+
151+**Server-side filtering.** The probe subscribes to one collection and asserts
152+that nothing else arrives. This is worth checking explicitly because the
153+failure mode is invisible: an unfiltered subscription is the entire network,
154+so a consumer that loses its query parameters still receives a healthy-looking
155+torrent of events that happen to be the wrong ones.
156+
157+**Cursor replay.** Rewinding to an earlier `time_us` must actually rewind, not
158+quietly resume live — that is how a consumer catches up after downtime. The
159+probe replays from a cursor it saw and asserts the events *overlap* what it
160+already received (25 of 25). Accepting the parameter proves nothing on its
161+own.
162+
163+Two traps:
164+
165+- **websock's `Uri` overload silently drops the query string.** It forwards
166+ only `uri.path`, so every Jetstream filter would vanish and the consumer
167+ would subscribe to the whole firehose while looking fine. `connect` uses the
168+ host/path form and assembles path-plus-query itself.
169+- Jetstream wants `wantedCollections` **repeated per value**, not one
170+ comma-joined string. A joined value matches no collection at all.
171+
172+`websock` pulls in chronicles, which does not compile unless a logging stream
173+is configured and resolves it in the *importing* module's scope — hence the
174+`import chronicles` in `jetstream.nim` that nothing appears to use, and the
175+`chronicles_enabled=off` in `nim.cfg`.
176+
141177 ## What this does and does not prove
142178
143179 Proven: the cryptography, the storage layer, and the OAuth flow up to user
@@ -150,8 +186,10 @@ Not proven, and still ahead:
150186 and it also covers the `invalid_token` refresh-and-retry branch of
151187 `xrpc.request`, which is currently unexercised.
152188 - **Session persistence** across restarts.
153-- The **Jetstream** websocket consumer.
154-- **CBOR/CAR** parsing for repository records.
189+- **CBOR/CAR** parsing for repository records — the last unexplored format,
190+ needed for full-repo sync rather than the live stream.
191+- **Reconnect and backoff** for a long-lived Jetstream subscription; the
192+ consumer reads a stream but does not yet survive the connection dropping.
155193
156194 The honest read: none of the three things that could have sunk the port did.
157195 What is left is protocol plumbing against well-specified formats — large, but
@@ -13,6 +13,7 @@ nim c -r sqlite_probe.nim # storage
13 nim c -r dpop_probe.nim && ./dpop_probe | (cd verify && go run .) # crypto13 nim c -r dpop_probe.nim && ./dpop_probe | (cd verify && go run .) # crypto
14 nim c -r -d:ssl oauth_probe.nim [handle] # the flow14 nim c -r -d:ssl oauth_probe.nim [handle] # the flow
15 nim c -r -d:ssl xrpc_probe.nim # signing15 nim c -r -d:ssl xrpc_probe.nim # signing
16+nim c -r -d:ssl jetstream_probe.nim # firehose
16 ```17 ```
17 18
18 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.19 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.
@@ -138,6 +139,41 @@ it with a 200 while ignoring the bogus credential entirely — a passing test
138 that tested nothing. `com.atproto.server.getSession` has to evaluate the139 that tested nothing. `com.atproto.server.getSession` has to evaluate the
139 credential.140 credential.
140 141
142+## 5. Jetstream — works, verified against the live firehose
143+
144+`atproto/jetstream.nim` consumes the ATProto firehose as JSON over a
145+websocket, using `websock` (already vendored via chronos). Jetstream is public
146+and unauthenticated, so unlike the OAuth and XRPC probes this one checks the
147+real thing end to end: 17/17 against the live network.
148+
149+Two capabilities that matter to Glean are confirmed rather than assumed:
150+
151+**Server-side filtering.** The probe subscribes to one collection and asserts
152+that nothing else arrives. This is worth checking explicitly because the
153+failure mode is invisible: an unfiltered subscription is the entire network,
154+so a consumer that loses its query parameters still receives a healthy-looking
155+torrent of events that happen to be the wrong ones.
156+
157+**Cursor replay.** Rewinding to an earlier `time_us` must actually rewind, not
158+quietly resume live — that is how a consumer catches up after downtime. The
159+probe replays from a cursor it saw and asserts the events *overlap* what it
160+already received (25 of 25). Accepting the parameter proves nothing on its
161+own.
162+
163+Two traps:
164+
165+- **websock's `Uri` overload silently drops the query string.** It forwards
166+ only `uri.path`, so every Jetstream filter would vanish and the consumer
167+ would subscribe to the whole firehose while looking fine. `connect` uses the
168+ host/path form and assembles path-plus-query itself.
169+- Jetstream wants `wantedCollections` **repeated per value**, not one
170+ comma-joined string. A joined value matches no collection at all.
171+
172+`websock` pulls in chronicles, which does not compile unless a logging stream
173+is configured and resolves it in the *importing* module's scope — hence the
174+`import chronicles` in `jetstream.nim` that nothing appears to use, and the
175+`chronicles_enabled=off` in `nim.cfg`.
176+
141 ## What this does and does not prove177 ## What this does and does not prove
142 178
143 Proven: the cryptography, the storage layer, and the OAuth flow up to user179 Proven: the cryptography, the storage layer, and the OAuth flow up to user
@@ -150,8 +186,10 @@ Not proven, and still ahead:
150 and it also covers the `invalid_token` refresh-and-retry branch of186 and it also covers the `invalid_token` refresh-and-retry branch of
151 `xrpc.request`, which is currently unexercised.187 `xrpc.request`, which is currently unexercised.
152 - **Session persistence** across restarts.188 - **Session persistence** across restarts.
153-- The **Jetstream** websocket consumer.189+- **CBOR/CAR** parsing for repository records — the last unexplored format,
154-- **CBOR/CAR** parsing for repository records.190+ needed for full-repo sync rather than the live stream.
191+- **Reconnect and backoff** for a long-lived Jetstream subscription; the
192+ consumer reads a stream but does not yet survive the connection dropping.
155 193
156 The honest read: none of the three things that could have sunk the port did.194 The honest read: none of the three things that could have sunk the port did.
157 What is left is protocol plumbing against well-specified formats — large, but195 What is left is protocol plumbing against well-specified formats — large, but