add JetstreamConsumer service and jetstream:consume Ace command

+388

apps/web/app/services/jetstream_consumer.ts

··· 1 + import type { EngagementEventRow } from '@skystar/clickhouse' 2 + import type { ClickHouseStore } from '@skystar/clickhouse' 3 + import { parseJetstreamEvent } from '@skystar/atproto' 4 + 5 + // --------------------------------------------------------------------------- 6 + // WebSocket-like interface for dependency injection in tests 7 + // --------------------------------------------------------------------------- 8 + 9 + export interface WebSocketLike { 10 + // Use 'any' for the event shapes so that both the browser-native WebSocket 11 + // (which passes MessageEvent<any>) and test fakes (which pass plain objects) 12 + // satisfy this interface without requiring full MessageEvent compatibility. 13 + 14 + onmessage?: ((ev: any) => void) | null 15 + 16 + onclose?: ((ev: any) => void) | null 17 + 18 + onerror?: ((ev: any) => void) | null 19 + 20 + onopen?: ((ev?: any) => void) | null 21 + close(): void 22 + } 23 + 24 + // --------------------------------------------------------------------------- 25 + // Deps interface (injectable seams for testing) 26 + // --------------------------------------------------------------------------- 27 + 28 + export interface JetstreamConsumerDeps { 29 + /** 30 + * Factory that creates a WebSocket connected to the given URL. 31 + * In production: (url) => new WebSocket(url) 32 + * In tests: (url) => new FakeWebSocket() 33 + */ 34 + createWebSocket: (url: string) => WebSocketLike 35 + 36 + /** 37 + * Returns the current set of tracked DIDs from the database. 38 + * In production: reads from SQLite users table. 39 + * In tests: returns a controlled set. 40 + */ 41 + readTrackedDids: () => Promise<Set<string>> 42 + 43 + /** 44 + * Reads the last durable cursor from storage. Returns null if none. 45 + * In production: raw SQLite query. 46 + * In tests: returns a controlled value. 47 + */ 48 + readCursor: () => Promise<bigint | null> 49 + 50 + /** 51 + * Durably writes the cursor to storage. 52 + * In production: INSERT OR REPLACE into jetstream_cursor. 53 + * In tests: records the call. 54 + */ 55 + writeCursor: (cursor: bigint) => Promise<void> 56 + 57 + /** 58 + * Returns the current time. Injected so tests can control time. 59 + */ 60 + now: () => Date 61 + } 62 + 63 + // --------------------------------------------------------------------------- 64 + // JetstreamConsumer 65 + // --------------------------------------------------------------------------- 66 + 67 + /** 68 + * Consumes the Bluesky Jetstream WebSocket firehose and inserts filtered 69 + * engagement events (likes, reposts) into ClickHouse for tracked users. 70 + * 71 + * Design decisions: 72 + * - compress=false: zstd decompression requires a Bluesky-provided zstd 73 + * dictionary. For Task 9 we omit compression to keep the wire protocol 74 + * simple. We can add zstd in a follow-up if bandwidth becomes an issue. 75 + * - Like/repost DELETE events are intentionally ignored (spec §10: unlikes 76 + * and unreposts are not tracked; counts may drift up slightly over time). 77 + * - Cursor uses BigInt internally because Jetstream time_us is a 64-bit 78 + * microsecond timestamp that exceeds JavaScript's safe integer range. 79 + */ 80 + export class JetstreamConsumer { 81 + // ------------------------------------------------------------------------- 82 + // State 83 + // ------------------------------------------------------------------------- 84 + 85 + private trackedDids: Set<string> = new Set() 86 + private eventBuffer: EngagementEventRow[] = [] 87 + private shutdownRequested = false 88 + private ws: WebSocketLike | null = null 89 + private cursorRefreshTimer: ReturnType<typeof setInterval> | null = null 90 + private bufferFlushTimer: ReturnType<typeof setInterval> | null = null 91 + 92 + /** 93 + * The largest time_us (as BigInt) across all events buffered in the current 94 + * flush cycle. Updated whenever we push an event into the buffer. 95 + */ 96 + private pendingCursor: bigint | null = null 97 + 98 + /** 99 + * The last cursor value that was durably written to SQLite. 100 + * On reconnect we use THIS value (not pendingCursor) to avoid replaying 101 + * events we wrote to CH but that we may not have checkpointed yet. 102 + */ 103 + private lastFlushedCursor: bigint | null = null 104 + 105 + /** 106 + * Current reconnect backoff in milliseconds. Resets to 1000 on success. 107 + */ 108 + private reconnectBackoffMs = 1000 109 + 110 + // ------------------------------------------------------------------------- 111 + // Constructor 112 + // ------------------------------------------------------------------------- 113 + 114 + constructor( 115 + private readonly clickHouseStore: ClickHouseStore, 116 + private readonly deps: JetstreamConsumerDeps 117 + ) {} 118 + 119 + // ------------------------------------------------------------------------- 120 + // Public API 121 + // ------------------------------------------------------------------------- 122 + 123 + /** 124 + * Starts the consumer. Connects to Jetstream, sets up timers, and blocks 125 + * until `shutdown()` is called. 126 + */ 127 + async start(): Promise<void> { 128 + // 1. Load initial tracked-DID set and cursor from storage 129 + this.trackedDids = await this.deps.readTrackedDids() 130 + this.lastFlushedCursor = await this.deps.readCursor() 131 + 132 + // 2. Connect to Jetstream 133 + await this.connect() 134 + 135 + // 3. Set up 1s tracked-DID refresh 136 + this.cursorRefreshTimer = setInterval(() => { 137 + void this.refreshTrackedDids() 138 + }, 1000) 139 + 140 + // 4. Set up 500ms / 1000-row flush interval 141 + this.bufferFlushTimer = setInterval(() => { 142 + void this.flushIfNeeded() 143 + }, 500) 144 + 145 + // 5. Wait until shutdown is requested 146 + await this.waitForShutdown() 147 + } 148 + 149 + /** 150 + * Graceful shutdown: flushes the buffer, checkpoints the cursor, closes 151 + * the WebSocket, and allows `start()` to return. 152 + */ 153 + async shutdown(): Promise<void> { 154 + this.shutdownRequested = true 155 + 156 + if (this.cursorRefreshTimer !== null) { 157 + clearInterval(this.cursorRefreshTimer) 158 + this.cursorRefreshTimer = null 159 + } 160 + if (this.bufferFlushTimer !== null) { 161 + clearInterval(this.bufferFlushTimer) 162 + this.bufferFlushTimer = null 163 + } 164 + 165 + // One final flush before closing 166 + await this.flushBuffer() 167 + 168 + if (this.ws !== null) { 169 + this.ws.close() 170 + this.ws = null 171 + } 172 + } 173 + 174 + // ------------------------------------------------------------------------- 175 + // Connection management 176 + // ------------------------------------------------------------------------- 177 + 178 + private buildUrl(baseUrl: string): string { 179 + const url = new URL(baseUrl) 180 + 181 + url.searchParams.set('wantedCollections', 'app.bsky.feed.like') 182 + url.searchParams.append('wantedCollections', 'app.bsky.feed.repost') 183 + url.searchParams.append('wantedCollections', 'app.bsky.feed.post') 184 + 185 + // compress=false — see design decision comment at the top of the file 186 + url.searchParams.set('compress', 'false') 187 + 188 + if (this.lastFlushedCursor !== null) { 189 + url.searchParams.set('cursor', this.lastFlushedCursor.toString()) 190 + } 191 + 192 + return url.toString() 193 + } 194 + 195 + private async connect(): Promise<void> { 196 + // Construct URL here so reconnects always read lastFlushedCursor at the 197 + // moment of reconnect, not the moment of the last successful flush. 198 + const url = this.buildUrl( 199 + process.env.JETSTREAM_URL ?? 'wss://jetstream2.us-east.bsky.network/subscribe' 200 + ) 201 + 202 + const ws = this.deps.createWebSocket(url) 203 + this.ws = ws 204 + 205 + ws.onmessage = (ev) => { 206 + this.handleMessage(ev.data) 207 + } 208 + 209 + ws.onclose = (ev) => { 210 + if (this.shutdownRequested) return 211 + 212 + console.error( 213 + `[JetstreamConsumer] WebSocket closed (code=${ev.code}, reason=${ev.reason}). Reconnecting in ${this.reconnectBackoffMs}ms...` 214 + ) 215 + 216 + const delay = this.reconnectBackoffMs 217 + this.reconnectBackoffMs = Math.min(this.reconnectBackoffMs * 2, 30_000) 218 + 219 + setTimeout(() => { 220 + if (!this.shutdownRequested) { 221 + void this.connect() 222 + } 223 + }, delay) 224 + } 225 + 226 + ws.onerror = (err) => { 227 + console.error('[JetstreamConsumer] WebSocket error:', err) 228 + } 229 + 230 + ws.onopen = () => { 231 + // Reset backoff on successful connection 232 + this.reconnectBackoffMs = 1000 233 + } 234 + } 235 + 236 + // ------------------------------------------------------------------------- 237 + // Message handling 238 + // ------------------------------------------------------------------------- 239 + 240 + private handleMessage(data: string | Buffer | ArrayBuffer): void { 241 + let text: string 242 + if (typeof data === 'string') { 243 + text = data 244 + } else if (Buffer.isBuffer(data)) { 245 + text = data.toString('utf8') 246 + } else { 247 + text = Buffer.from(data).toString('utf8') 248 + } 249 + 250 + let rawJson: unknown 251 + try { 252 + rawJson = JSON.parse(text) 253 + } catch { 254 + // Malformed JSON — drop silently 255 + return 256 + } 257 + 258 + const event = parseJetstreamEvent(rawJson) 259 + if (event === null) return 260 + 261 + if (event.kind === 'like' || event.kind === 'repost') { 262 + if (!this.trackedDids.has(event.postAuthorDid)) return 263 + 264 + // Extract time_us from the raw message for cursor tracking. 265 + // The parsed event uses `ingestedAt` (a Date), but we need the raw 266 + // microsecond BigInt for cursor storage. 267 + let timeUs: bigint | null = null 268 + if (typeof rawJson === 'object' && rawJson !== null && 'time_us' in rawJson) { 269 + const raw = (rawJson as Record<string, unknown>)['time_us'] 270 + if (typeof raw === 'number') { 271 + timeUs = BigInt(Math.floor(raw)) 272 + } 273 + } 274 + 275 + this.eventBuffer.push({ 276 + postUri: event.postUri, 277 + postAuthorDid: event.postAuthorDid, 278 + actorDid: event.actorDid, 279 + rkey: event.rkey, 280 + kind: event.kind, 281 + eventCreatedAt: event.createdAt, 282 + }) 283 + 284 + if (timeUs !== null) { 285 + if (this.pendingCursor === null || timeUs > this.pendingCursor) { 286 + this.pendingCursor = timeUs 287 + } 288 + } 289 + 290 + // Flush immediately if the buffer hits 1000 rows 291 + if (this.eventBuffer.length >= 1000) { 292 + void this.flushBuffer() 293 + } 294 + } 295 + 296 + // kind === 'post' | 'post-delete' | 'account' | 'identity': 297 + // Dropped in Task 9. Task 10 will add handling. 298 + } 299 + 300 + // ------------------------------------------------------------------------- 301 + // Buffer flush 302 + // ------------------------------------------------------------------------- 303 + 304 + /** Called by the interval timer — only flushes if there's anything to flush. */ 305 + private async flushIfNeeded(): Promise<void> { 306 + if (this.eventBuffer.length > 0) { 307 + await this.flushBuffer() 308 + } 309 + } 310 + 311 + /** 312 + * Drains the event buffer and inserts into ClickHouse. 313 + * On failure: re-buffers the events at the front so the next flush retries. 314 + * On success: checkpoints the cursor. 315 + */ 316 + async flushBuffer(): Promise<void> { 317 + if (this.eventBuffer.length === 0) return 318 + 319 + const toFlush = this.eventBuffer.splice(0) 320 + const cursorAtFlushTime = this.pendingCursor 321 + this.pendingCursor = null 322 + 323 + try { 324 + await this.clickHouseStore.insertEngagementEvents(toFlush) 325 + } catch (err) { 326 + console.error(`[JetstreamConsumer] flush failed (${toFlush.length} events); will retry:`, err) 327 + // Re-buffer the failed batch at the front 328 + this.eventBuffer.unshift(...toFlush) 329 + // Restore pendingCursor 330 + if (cursorAtFlushTime !== null) { 331 + if (this.pendingCursor === null || cursorAtFlushTime > this.pendingCursor) { 332 + this.pendingCursor = cursorAtFlushTime 333 + } 334 + } 335 + return 336 + } 337 + 338 + // Successful flush — checkpoint the cursor 339 + if (cursorAtFlushTime !== null) { 340 + this.lastFlushedCursor = cursorAtFlushTime 341 + await this.checkpointCursor(cursorAtFlushTime) 342 + } 343 + } 344 + 345 + // ------------------------------------------------------------------------- 346 + // Cursor checkpointing 347 + // ------------------------------------------------------------------------- 348 + 349 + private async checkpointCursor(cursor: bigint): Promise<void> { 350 + try { 351 + await this.deps.writeCursor(cursor) 352 + } catch (err) { 353 + // Non-fatal: we'll try again on the next flush 354 + console.error('[JetstreamConsumer] cursor checkpoint failed:', err) 355 + } 356 + } 357 + 358 + // ------------------------------------------------------------------------- 359 + // Tracked-DID refresh 360 + // ------------------------------------------------------------------------- 361 + 362 + private async refreshTrackedDids(): Promise<void> { 363 + try { 364 + const newSet = await this.deps.readTrackedDids() 365 + this.trackedDids = newSet // atomic swap 366 + } catch (err) { 367 + // Non-fatal: keep the old set until the next refresh succeeds 368 + console.error('[JetstreamConsumer] tracked-DID refresh failed:', err) 369 + } 370 + } 371 + 372 + // ------------------------------------------------------------------------- 373 + // Shutdown coordination 374 + // ------------------------------------------------------------------------- 375 + 376 + private waitForShutdown(): Promise<void> { 377 + return new Promise((resolve) => { 378 + const check = () => { 379 + if (this.shutdownRequested) { 380 + resolve() 381 + } else { 382 + setTimeout(check, 100) 383 + } 384 + } 385 + check() 386 + }) 387 + } 388 + }

+110

apps/web/commands/jetstream_consume.ts

··· 1 + import { BaseCommand } from '@adonisjs/core/ace' 2 + import db from '@adonisjs/lucid/services/db' 3 + import { ClickHouseStore } from '@skystar/clickhouse' 4 + import { JetstreamConsumer } from '#services/jetstream_consumer' 5 + 6 + /** 7 + * Ace command: node ace jetstream:consume 8 + * 9 + * Connects to the Bluesky Jetstream WebSocket firehose and streams 10 + * engagement events (likes, reposts) for tracked users into ClickHouse. 11 + * 12 + * Designed to run as a long-lived daemon process. Uses `staysAlive: true` 13 + * so the Adonis framework does not exit when `run()` returns — the command 14 + * runs indefinitely until SIGTERM/SIGINT. 15 + * 16 + * Cursor persistence: 17 + * On startup, reads the last checkpointed Jetstream cursor from SQLite and 18 + * uses it as the `cursor=` query parameter so we replay any events missed 19 + * during downtime. After each successful ClickHouse flush, the cursor is 20 + * written back to SQLite via raw query (bypassing Lucid ORM since BigInt 21 + * values exceed JS safe integer range and Lucid doesn't coerce BigInt 22 + * columns natively). 23 + * 24 + * Tracked-DID filter: 25 + * Every 1 second the worker refreshes its in-memory Set<did> of tracked 26 + * users from SQLite. Events for users not in this set are discarded. 27 + * 28 + * Compression: 29 + * compress=false — see JetstreamConsumer for rationale. 30 + */ 31 + export default class JetstreamConsume extends BaseCommand { 32 + static commandName = 'jetstream:consume' 33 + static description = 'Consume the Bluesky Jetstream firehose into ClickHouse' 34 + 35 + /** 36 + * staysAlive: true — tells Adonis NOT to exit when run() returns. 37 + * The process remains alive until shutdown() is called (on SIGTERM/SIGINT), 38 + * which causes start() to resolve and run() to return normally. 39 + * See: https://docs.adonisjs.com/guides/ace/creating-commands#stay-alive-mode 40 + */ 41 + static options = { startApp: true, staysAlive: true } 42 + 43 + async run() { 44 + const clickHouseStore = await this.app.container.make(ClickHouseStore) 45 + 46 + const consumer = new JetstreamConsumer(clickHouseStore, { 47 + // Production WebSocket factory — uses Node 24's built-in global WebSocket 48 + createWebSocket: (url: string) => { 49 + return new WebSocket(url) 50 + }, 51 + 52 + // Read tracked DIDs from SQLite users table 53 + readTrackedDids: async () => { 54 + const rows = await db.from('users').select('did') 55 + return new Set<string>(rows.map((r: { did: string }) => r.did)) 56 + }, 57 + 58 + // Read the persisted Jetstream cursor from SQLite via raw query. 59 + // We use a raw query (not the Lucid model) because the cursor_us column 60 + // is a 64-bit integer and Lucid returns JS numbers that lose precision 61 + // above 2^53. The raw @adonisjs/lucid db.rawQuery returns the value as 62 + // a JS number from better-sqlite3; we convert via BigInt to preserve it. 63 + readCursor: async () => { 64 + const result = await db.rawQuery( 65 + 'SELECT cursor_us FROM jetstream_cursor WHERE id = 1 LIMIT 1' 66 + ) 67 + const rows = result as unknown as Array<{ cursor_us: number | bigint }> 68 + if (!rows || rows.length === 0) return null 69 + const raw = rows[0].cursor_us 70 + return typeof raw === 'bigint' ? raw : BigInt(Math.floor(Number(raw))) 71 + }, 72 + 73 + // Write the cursor to SQLite via raw query (INSERT OR REPLACE singleton row). 74 + writeCursor: async (cursor: bigint) => { 75 + const nowMs = Date.now() 76 + await db.rawQuery( 77 + 'INSERT OR REPLACE INTO jetstream_cursor (id, cursor_us, updated_at) VALUES (1, ?, ?)', 78 + [cursor.toString(), nowMs] 79 + ) 80 + }, 81 + 82 + now: () => new Date(), 83 + }) 84 + 85 + // Graceful shutdown on SIGTERM (Docker stop, systemd stop, k8s eviction) 86 + const handleSignal = () => { 87 + this.logger.info('Shutdown signal received — flushing and closing...') 88 + consumer 89 + .shutdown() 90 + .then(() => { 91 + this.logger.info('JetstreamConsumer shut down cleanly') 92 + process.exit(0) 93 + }) 94 + .catch((err: unknown) => { 95 + const message = err instanceof Error ? err.message : String(err) 96 + this.logger.error(`Error during shutdown: ${message}`) 97 + process.exit(1) 98 + }) 99 + } 100 + 101 + process.on('SIGTERM', handleSignal) 102 + process.on('SIGINT', handleSignal) 103 + 104 + this.logger.info( 105 + `Starting JetstreamConsumer (JETSTREAM_URL=${process.env.JETSTREAM_URL ?? 'wss://jetstream2.us-east.bsky.network/subscribe'})` 106 + ) 107 + 108 + await consumer.start() 109 + } 110 + }

+518

apps/web/tests/unit/services/jetstream_consumer.spec.ts

··· 1 + /** 2 + * Unit tests for JetstreamConsumer. 3 + * 4 + * All dependencies (WebSocket, SQLite reads/writes, ClickHouseStore) are 5 + * injected as fakes so these tests run without any external services. 6 + */ 7 + import { test } from '@japa/runner' 8 + import type { ClickHouseStore } from '@skystar/clickhouse' 9 + import type { EngagementEventRow } from '@skystar/clickhouse' 10 + import { 11 + JetstreamConsumer, 12 + type WebSocketLike, 13 + type JetstreamConsumerDeps, 14 + } from '#services/jetstream_consumer' 15 + 16 + // --------------------------------------------------------------------------- 17 + // FakeWebSocket — controllable WebSocket stub 18 + // --------------------------------------------------------------------------- 19 + 20 + class FakeWebSocket implements WebSocketLike { 21 + onmessage?: ((ev: any) => void) | null 22 + 23 + onclose?: ((ev: any) => void) | null 24 + 25 + onerror?: ((ev: any) => void) | null 26 + onopen?: (() => void) | null 27 + 28 + closeCallCount = 0 29 + constructedUrl = '' 30 + 31 + close() { 32 + this.closeCallCount++ 33 + } 34 + 35 + /** Emit a parsed event object as a JSON message. */ 36 + emit(event: unknown) { 37 + this.onmessage?.({ data: JSON.stringify(event) }) 38 + } 39 + 40 + /** Emit a close event (simulates unexpected disconnect). */ 41 + emitClose(code = 1006) { 42 + this.onclose?.({ code, reason: 'test close' }) 43 + } 44 + 45 + /** Trigger the open handler to reset reconnect backoff. */ 46 + emitOpen() { 47 + this.onopen?.() 48 + } 49 + } 50 + 51 + // --------------------------------------------------------------------------- 52 + // FakeClickHouseStore 53 + // --------------------------------------------------------------------------- 54 + 55 + interface FakeStoreConfig { 56 + shouldThrow?: boolean 57 + } 58 + 59 + function makeFakeStore(config: FakeStoreConfig = {}): ClickHouseStore & { 60 + insertCalls: EngagementEventRow[][] 61 + } { 62 + const insertCalls: EngagementEventRow[][] = [] 63 + 64 + return { 65 + insertCalls, 66 + async insertEngagementEvents(events: EngagementEventRow[]) { 67 + if (config.shouldThrow) throw new Error('ClickHouse unavailable') 68 + insertCalls.push([...events]) 69 + }, 70 + } as unknown as ClickHouseStore & { insertCalls: EngagementEventRow[][] } 71 + } 72 + 73 + // --------------------------------------------------------------------------- 74 + // Test fixture event factories 75 + // --------------------------------------------------------------------------- 76 + 77 + const TRACKED_AUTHOR_DID = 'did:plc:trackedauthor123' 78 + const UNTRACKED_AUTHOR_DID = 'did:plc:untrackedauthor456' 79 + const ACTOR_DID = 'did:plc:actor789' 80 + 81 + function makeLikeEvent(postAuthorDid: string, timeUs = 1725911162329308, rkey = 'likerkey1') { 82 + return { 83 + did: ACTOR_DID, 84 + time_us: timeUs, 85 + kind: 'commit', 86 + commit: { 87 + rev: 'rev1', 88 + operation: 'create', 89 + collection: 'app.bsky.feed.like', 90 + rkey, 91 + record: { 92 + $type: 'app.bsky.feed.like', 93 + createdAt: '2024-09-09T19:46:02.102Z', 94 + subject: { 95 + cid: 'bafycid1', 96 + uri: `at://${postAuthorDid}/app.bsky.feed.post/postrkeyabc`, 97 + }, 98 + }, 99 + cid: 'bafylikecid', 100 + }, 101 + } 102 + } 103 + 104 + function makeRepostEvent(postAuthorDid: string, timeUs = 1725911162329308, rkey = 'repostrkey1') { 105 + return { 106 + did: ACTOR_DID, 107 + time_us: timeUs, 108 + kind: 'commit', 109 + commit: { 110 + rev: 'rev2', 111 + operation: 'create', 112 + collection: 'app.bsky.feed.repost', 113 + rkey, 114 + record: { 115 + $type: 'app.bsky.feed.repost', 116 + createdAt: '2024-09-09T19:47:00.000Z', 117 + subject: { 118 + cid: 'bafycid2', 119 + uri: `at://${postAuthorDid}/app.bsky.feed.post/postrkeydef`, 120 + }, 121 + }, 122 + cid: 'bafyrepostcid', 123 + }, 124 + } 125 + } 126 + 127 + function makeLikeDeleteEvent() { 128 + return { 129 + did: ACTOR_DID, 130 + time_us: 1725911162329308, 131 + kind: 'commit', 132 + commit: { 133 + rev: 'rev3', 134 + operation: 'delete', 135 + collection: 'app.bsky.feed.like', 136 + rkey: 'likedeletedkey', 137 + }, 138 + } 139 + } 140 + 141 + function makePostEvent(authorDid: string) { 142 + return { 143 + did: authorDid, 144 + time_us: 1725911162329308, 145 + kind: 'commit', 146 + commit: { 147 + rev: 'rev4', 148 + operation: 'create', 149 + collection: 'app.bsky.feed.post', 150 + rkey: 'postrkeyxyz', 151 + record: { 152 + $type: 'app.bsky.feed.post', 153 + text: 'Hello Bluesky!', 154 + createdAt: '2024-09-09T19:48:00.000Z', 155 + }, 156 + cid: 'bafypostcid', 157 + }, 158 + } 159 + } 160 + 161 + // --------------------------------------------------------------------------- 162 + // Consumer factory 163 + // --------------------------------------------------------------------------- 164 + 165 + interface ConsumerSetup { 166 + fakeWs: FakeWebSocket 167 + store: ClickHouseStore & { insertCalls: EngagementEventRow[][] } 168 + consumer: JetstreamConsumer 169 + writeCursorCalls: bigint[] 170 + lastConnectedUrl: { value: string } 171 + } 172 + 173 + function makeConsumer( 174 + options: { 175 + trackedDids?: Set<string> 176 + initialCursor?: bigint | null 177 + storeShouldThrow?: boolean 178 + } = {} 179 + ): ConsumerSetup { 180 + const { 181 + trackedDids = new Set([TRACKED_AUTHOR_DID]), 182 + initialCursor = null, 183 + storeShouldThrow = false, 184 + } = options 185 + 186 + const fakeWs = new FakeWebSocket() 187 + const store = makeFakeStore({ shouldThrow: storeShouldThrow }) 188 + const writeCursorCalls: bigint[] = [] 189 + const lastConnectedUrl: { value: string } = { value: '' } 190 + 191 + let currentTrackedDids = trackedDids 192 + 193 + const deps: JetstreamConsumerDeps = { 194 + createWebSocket(url: string) { 195 + lastConnectedUrl.value = url 196 + fakeWs.constructedUrl = url 197 + return fakeWs 198 + }, 199 + readTrackedDids: async () => currentTrackedDids, 200 + readCursor: async () => initialCursor, 201 + writeCursor: async (cursor: bigint) => { 202 + writeCursorCalls.push(cursor) 203 + }, 204 + now: () => new Date('2024-09-09T19:46:00.000Z'), 205 + } 206 + 207 + const consumer = new JetstreamConsumer(store, deps) 208 + 209 + // Expose the ability to update tracked dids for test 10 210 + ;(consumer as any).__setTrackedDids = (newSet: Set<string>) => { 211 + currentTrackedDids = newSet 212 + } 213 + 214 + return { fakeWs, store, consumer, writeCursorCalls, lastConnectedUrl } 215 + } 216 + 217 + // --------------------------------------------------------------------------- 218 + // Helper: start consumer without blocking (it blocks until shutdown) 219 + // --------------------------------------------------------------------------- 220 + 221 + async function startConsumerInBackground(consumer: JetstreamConsumer): Promise<void> { 222 + // We don't await start() — it loops forever until shutdown() 223 + // Just kick it off and let the event loop run 224 + void consumer.start() 225 + // Give the event loop a tick to let connect() run synchronously 226 + await new Promise((r) => setTimeout(r, 0)) 227 + } 228 + 229 + // --------------------------------------------------------------------------- 230 + // Tests 231 + // --------------------------------------------------------------------------- 232 + 233 + test.group('JetstreamConsumer — like event for tracked user', () => { 234 + test('like event for tracked user is buffered and flushed', async ({ assert }) => { 235 + const { fakeWs, store, consumer } = makeConsumer() 236 + 237 + await startConsumerInBackground(consumer) 238 + 239 + fakeWs.emit(makeLikeEvent(TRACKED_AUTHOR_DID)) 240 + 241 + await consumer.flushBuffer() 242 + 243 + assert.equal(store.insertCalls.length, 1) 244 + assert.equal(store.insertCalls[0].length, 1) 245 + assert.equal(store.insertCalls[0][0].postAuthorDid, TRACKED_AUTHOR_DID) 246 + assert.equal(store.insertCalls[0][0].actorDid, ACTOR_DID) 247 + assert.equal(store.insertCalls[0][0].kind, 'like') 248 + 249 + await consumer.shutdown() 250 + }) 251 + }) 252 + 253 + test.group('JetstreamConsumer — like event for untracked user', () => { 254 + test('like event for untracked user is dropped', async ({ assert }) => { 255 + const { fakeWs, store, consumer } = makeConsumer() 256 + 257 + await startConsumerInBackground(consumer) 258 + 259 + fakeWs.emit(makeLikeEvent(UNTRACKED_AUTHOR_DID)) 260 + 261 + await consumer.flushBuffer() 262 + 263 + assert.equal(store.insertCalls.length, 0) 264 + 265 + await consumer.shutdown() 266 + }) 267 + }) 268 + 269 + test.group('JetstreamConsumer — repost event for tracked user', () => { 270 + test('repost event for tracked user is buffered and flushed', async ({ assert }) => { 271 + const { fakeWs, store, consumer } = makeConsumer() 272 + 273 + await startConsumerInBackground(consumer) 274 + 275 + fakeWs.emit(makeRepostEvent(TRACKED_AUTHOR_DID)) 276 + 277 + await consumer.flushBuffer() 278 + 279 + assert.equal(store.insertCalls.length, 1) 280 + assert.equal(store.insertCalls[0][0].kind, 'repost') 281 + assert.equal(store.insertCalls[0][0].postAuthorDid, TRACKED_AUTHOR_DID) 282 + 283 + await consumer.shutdown() 284 + }) 285 + }) 286 + 287 + test.group('JetstreamConsumer — like delete events are dropped', () => { 288 + test('like delete event is dropped (spec §10: unlikes not tracked)', async ({ assert }) => { 289 + const { fakeWs, store, consumer } = makeConsumer() 290 + 291 + await startConsumerInBackground(consumer) 292 + 293 + fakeWs.emit(makeLikeDeleteEvent()) 294 + 295 + await consumer.flushBuffer() 296 + 297 + assert.equal(store.insertCalls.length, 0) 298 + 299 + await consumer.shutdown() 300 + }) 301 + }) 302 + 303 + test.group('JetstreamConsumer — post events do not crash', () => { 304 + test('post create event for tracked author does not crash and inserts nothing (Task 9)', async ({ 305 + assert, 306 + }) => { 307 + const { fakeWs, store, consumer } = makeConsumer() 308 + 309 + await startConsumerInBackground(consumer) 310 + 311 + // Should not throw 312 + fakeWs.emit(makePostEvent(TRACKED_AUTHOR_DID)) 313 + 314 + await consumer.flushBuffer() 315 + 316 + // Task 9: no insert for post events 317 + assert.equal(store.insertCalls.length, 0) 318 + 319 + await consumer.shutdown() 320 + }) 321 + }) 322 + 323 + test.group('JetstreamConsumer — buffer flushes after 1000 events', () => { 324 + test('buffer auto-flushes at 1000 events', async ({ assert }) => { 325 + const { fakeWs, store, consumer } = makeConsumer() 326 + 327 + await startConsumerInBackground(consumer) 328 + 329 + // Emit 1001 like events; the 1000th should trigger an automatic flush 330 + for (let i = 0; i < 1001; i++) { 331 + fakeWs.emit(makeLikeEvent(TRACKED_AUTHOR_DID, 1725911162329308 + i, `rkey${i}`)) 332 + } 333 + 334 + // Wait for the async flush triggered at 1000 to complete 335 + await new Promise((r) => setTimeout(r, 10)) 336 + 337 + // At least one flush should have happened (the 1000-row auto-flush) 338 + assert.isAbove(store.insertCalls.length, 0) 339 + 340 + // The first flush should have exactly 1000 rows 341 + assert.equal(store.insertCalls[0].length, 1000) 342 + 343 + await consumer.shutdown() 344 + }) 345 + }) 346 + 347 + test.group('JetstreamConsumer — buffer flushes after 500ms interval', () => { 348 + test('5 events are flushed by the 500ms interval timer', async ({ assert }) => { 349 + const { fakeWs, store, consumer } = makeConsumer() 350 + 351 + await startConsumerInBackground(consumer) 352 + 353 + for (let i = 0; i < 5; i++) { 354 + fakeWs.emit(makeLikeEvent(TRACKED_AUTHOR_DID, 1725911162329308 + i, `rkey${i}`)) 355 + } 356 + 357 + // Wait longer than the 500ms flush interval 358 + await new Promise((r) => setTimeout(r, 600)) 359 + 360 + assert.isAbove(store.insertCalls.length, 0) 361 + const totalFlushed = store.insertCalls.reduce((acc, batch) => acc + batch.length, 0) 362 + assert.equal(totalFlushed, 5) 363 + 364 + await consumer.shutdown() 365 + }) 366 + }) 367 + 368 + test.group('JetstreamConsumer — cursor checkpointed after flush', () => { 369 + test('writeCursor is called with the event time_us after a successful flush', async ({ 370 + assert, 371 + }) => { 372 + const { fakeWs, consumer, writeCursorCalls } = makeConsumer() 373 + 374 + await startConsumerInBackground(consumer) 375 + 376 + const timeUs = 1725911162329308 377 + fakeWs.emit(makeLikeEvent(TRACKED_AUTHOR_DID, timeUs)) 378 + 379 + await consumer.flushBuffer() 380 + 381 + assert.equal(writeCursorCalls.length, 1) 382 + assert.equal(writeCursorCalls[0], BigInt(timeUs)) 383 + 384 + await consumer.shutdown() 385 + }) 386 + }) 387 + 388 + test.group('JetstreamConsumer — failed insert retains events in buffer', () => { 389 + test('events remain in buffer after a failed flush', async ({ assert }) => { 390 + const { fakeWs, store, consumer } = makeConsumer({ storeShouldThrow: true }) 391 + 392 + await startConsumerInBackground(consumer) 393 + 394 + fakeWs.emit(makeLikeEvent(TRACKED_AUTHOR_DID)) 395 + 396 + // flushBuffer throws internally, but must not throw to caller 397 + await consumer.flushBuffer() 398 + 399 + // The event should still be in the buffer (accessible via another flush attempt) 400 + // We can verify by making the store stop throwing and flushing again 401 + ;(store as any).insertEngagementEvents = async (events: EngagementEventRow[]) => { 402 + store.insertCalls.push([...events]) 403 + } 404 + 405 + await consumer.flushBuffer() 406 + 407 + assert.equal(store.insertCalls.length, 1) 408 + assert.equal(store.insertCalls[0].length, 1) 409 + 410 + await consumer.shutdown() 411 + }) 412 + }) 413 + 414 + test.group('JetstreamConsumer — tracked-DID refresh', () => { 415 + test('refreshTrackedDids updates the in-memory set so new users are tracked', async ({ 416 + assert, 417 + }) => { 418 + const aliceDid = 'did:plc:alice' 419 + const bobDid = 'did:plc:bob' 420 + 421 + const { fakeWs, store, consumer } = makeConsumer({ 422 + trackedDids: new Set([aliceDid]), 423 + }) 424 + 425 + await startConsumerInBackground(consumer) 426 + 427 + // Bob's like is dropped initially 428 + fakeWs.emit(makeLikeEvent(bobDid, 1725911162329308, 'rkey-bob-1')) 429 + await consumer.flushBuffer() 430 + assert.equal(store.insertCalls.length, 0, 'bob should be dropped before refresh') 431 + 432 + // Update the DID source to return both alice + bob 433 + ;(consumer as any).__setTrackedDids(new Set([aliceDid, bobDid])) 434 + 435 + // Trigger a refresh (call the private method via the 1s interval mechanism) 436 + await (consumer as any).refreshTrackedDids() 437 + 438 + // Bob's like should now be accepted 439 + fakeWs.emit(makeLikeEvent(bobDid, 1725911162329309, 'rkey-bob-2')) 440 + await consumer.flushBuffer() 441 + 442 + assert.equal(store.insertCalls.length, 1) 443 + assert.equal(store.insertCalls[0][0].postAuthorDid, bobDid) 444 + 445 + await consumer.shutdown() 446 + }) 447 + }) 448 + 449 + test.group('JetstreamConsumer — shutdown flushes remaining events', () => { 450 + test('shutdown flushes all buffered events before closing', async ({ assert }) => { 451 + const { fakeWs, store, consumer } = makeConsumer() 452 + 453 + await startConsumerInBackground(consumer) 454 + 455 + for (let i = 0; i < 5; i++) { 456 + fakeWs.emit(makeLikeEvent(TRACKED_AUTHOR_DID, 1725911162329308 + i, `rkey${i}`)) 457 + } 458 + 459 + await consumer.shutdown() 460 + 461 + const totalFlushed = store.insertCalls.reduce((acc, batch) => acc + batch.length, 0) 462 + assert.equal(totalFlushed, 5) 463 + }) 464 + }) 465 + 466 + test.group('JetstreamConsumer — shutdown writes final cursor checkpoint', () => { 467 + test('shutdown writes the cursor for all flushed events', async ({ assert }) => { 468 + const { fakeWs, consumer, writeCursorCalls } = makeConsumer() 469 + 470 + await startConsumerInBackground(consumer) 471 + 472 + const timeUs = 1725911162340000 473 + for (let i = 0; i < 3; i++) { 474 + fakeWs.emit(makeLikeEvent(TRACKED_AUTHOR_DID, timeUs + i, `rkey${i}`)) 475 + } 476 + 477 + await consumer.shutdown() 478 + 479 + // The final cursor should be the largest time_us 480 + assert.isAbove(writeCursorCalls.length, 0) 481 + const lastWritten = writeCursorCalls[writeCursorCalls.length - 1] 482 + assert.equal(lastWritten, BigInt(timeUs + 2)) 483 + }) 484 + }) 485 + 486 + test.group('JetstreamConsumer — reconnect uses lastFlushedCursor', () => { 487 + test('reconnect URL contains the last durable cursor, not the in-memory pending one', async ({ 488 + assert, 489 + }) => { 490 + const durableCursor = 999_000_000n 491 + const { fakeWs, consumer, lastConnectedUrl } = makeConsumer({ 492 + initialCursor: durableCursor, 493 + }) 494 + 495 + await startConsumerInBackground(consumer) 496 + 497 + // Verify initial connection URL contains the durable cursor 498 + assert.include(lastConnectedUrl.value, `cursor=${durableCursor.toString()}`) 499 + 500 + // Emit an event with a new time_us (in-memory cursor advances, but not yet flushed) 501 + fakeWs.emit(makeLikeEvent(TRACKED_AUTHOR_DID, 1725911162329308, 'rkey1')) 502 + 503 + // Simulate unexpected WebSocket close BEFORE flushing 504 + // (so lastFlushedCursor is still durableCursor) 505 + const urlBeforeReconnect = lastConnectedUrl.value 506 + fakeWs.emitClose(1006) 507 + 508 + // Give the reconnect setTimeout(0 delay would be 1s, but we can check the 509 + // URL was set from the initial connect) — in this test we just verify that 510 + // the initial URL contained the durable cursor value, and that if we called 511 + // buildUrl again it would still use lastFlushedCursor (durableCursor). 512 + // The actual reconnect with backoff uses setTimeout(1000) so we can't easily 513 + // test it without fake timers. Instead, verify the initial URL. 514 + assert.include(urlBeforeReconnect, `cursor=${durableCursor.toString()}`) 515 + 516 + await consumer.shutdown() 517 + }) 518 + })

Configure Feed

Configure Feed