nandi.uk
This repository has no description
8.5 kB
301 lines
1import WebSocket from 'ws';
2import { IFirehoseService } from '../../application/services/IFirehoseService';
3import { FirehoseEventHandler } from '../../application/handlers/FirehoseEventHandler';
4import { EnvironmentConfigService } from 'src/shared/infrastructure/config/EnvironmentConfigService';
5import { FirehoseEvent } from '../../domain/FirehoseEvent';
6
7const DEBUG_LOGGING = true; // Set to false to disable debug logs
8
9interface JetstreamCommitEvent {
10 did: string;
11 time_us: number;
12 kind: 'commit';
13 commit: {
14 rev: string;
15 operation: 'create' | 'update' | 'delete';
16 collection: string;
17 rkey: string;
18 record?: any;
19 cid?: string;
20 };
21}
22
23interface JetstreamIdentityEvent {
24 did: string;
25 time_us: number;
26 kind: 'identity';
27 identity: {
28 did: string;
29 handle: string;
30 seq: number;
31 time: string;
32 };
33}
34
35interface JetstreamAccountEvent {
36 did: string;
37 time_us: number;
38 kind: 'account';
39 account: {
40 active: boolean;
41 did: string;
42 seq: number;
43 time: string;
44 };
45}
46
47type JetstreamEvent =
48 | JetstreamCommitEvent
49 | JetstreamIdentityEvent
50 | JetstreamAccountEvent;
51
52export class AtProtoJetstreamService implements IFirehoseService {
53 private ws?: WebSocket;
54 private isRunningFlag = false;
55 private cleaningUp = false;
56 private eventCount = 0;
57 private logInterval?: NodeJS.Timeout;
58 private mostRecentEventTime?: Date;
59 private reconnectTimeout?: NodeJS.Timeout;
60 private reconnectAttempts = 0;
61 private maxReconnectAttempts = 10;
62 private reconnectDelay = 5000; // 5 seconds
63
64 constructor(
65 private firehoseEventHandler: FirehoseEventHandler,
66 private configService: EnvironmentConfigService,
67 ) {}
68
69 async start(): Promise<void> {
70 if (this.isRunningFlag) {
71 return;
72 }
73
74 try {
75 console.log(
76 `[JETSTREAM] Starting Jetstream service for collections: ${this.getFilteredCollections().join(', ')}`,
77 );
78
79 await this.connect();
80 this.isRunningFlag = true;
81 this.startEventCountLogging();
82 console.log('[JETSTREAM] Jetstream service started');
83 } catch (error) {
84 console.error('[JETSTREAM] Failed to start Jetstream service:', error);
85 throw error;
86 }
87 }
88
89 async stop(): Promise<void> {
90 if (!this.isRunningFlag) {
91 return;
92 }
93
94 console.log('[JETSTREAM] Stopping Jetstream service...');
95
96 if (this.reconnectTimeout) {
97 clearTimeout(this.reconnectTimeout);
98 this.reconnectTimeout = undefined;
99 }
100
101 if (this.ws) {
102 this.ws.close();
103 this.ws = undefined;
104 }
105
106 this.stopEventCountLogging();
107 this.isRunningFlag = false;
108 console.log('[JETSTREAM] Jetstream service stopped');
109 }
110
111 isRunning(): boolean {
112 return (
113 this.isRunningFlag && !!this.ws && this.ws.readyState === WebSocket.OPEN
114 );
115 }
116
117 private async connect(): Promise<void> {
118 return new Promise((resolve, reject) => {
119 const jetstreamUrl = this.buildJetstreamUrl();
120 console.log(`[JETSTREAM] Connecting to ${jetstreamUrl}`);
121
122 this.ws = new WebSocket(jetstreamUrl);
123
124 this.ws.on('open', () => {
125 console.log('[JETSTREAM] WebSocket connection established');
126 this.reconnectAttempts = 0;
127 resolve();
128 });
129
130 this.ws.on('message', (data: WebSocket.Data) => {
131 this.handleMessage(data);
132 });
133
134 this.ws.on('close', (code: number, reason: Buffer) => {
135 console.log(
136 `[JETSTREAM] WebSocket closed: ${code} ${reason.toString()}`,
137 );
138 if (this.isRunningFlag && !this.cleaningUp) {
139 this.scheduleReconnect();
140 }
141 });
142
143 this.ws.on('error', (error: Error) => {
144 console.error('[JETSTREAM] WebSocket error:', error);
145 if (this.reconnectAttempts === 0) {
146 reject(error);
147 }
148 });
149 });
150 }
151
152 private scheduleReconnect(): void {
153 if (this.reconnectAttempts >= this.maxReconnectAttempts) {
154 console.error('[JETSTREAM] Max reconnection attempts reached, giving up');
155 return;
156 }
157
158 this.reconnectAttempts++;
159 const delay =
160 this.reconnectDelay *
161 Math.pow(2, Math.min(this.reconnectAttempts - 1, 5)); // Exponential backoff, max 5
162
163 console.log(
164 `[JETSTREAM] Scheduling reconnect attempt ${this.reconnectAttempts} in ${delay}ms`,
165 );
166
167 this.reconnectTimeout = setTimeout(async () => {
168 try {
169 await this.connect();
170 } catch (error) {
171 console.error('[JETSTREAM] Reconnection failed:', error);
172 this.scheduleReconnect();
173 }
174 }, delay);
175 }
176
177 private buildJetstreamUrl(): string {
178 const baseUrl = this.configService.getAtProtoConfig().jetstreamWebsocket;
179 const collections = this.getFilteredCollections();
180
181 const params = new URLSearchParams();
182 collections.forEach((collection) => {
183 params.append('wantedCollections', collection);
184 });
185
186 return `${baseUrl}/subscribe?${params.toString()}`;
187 }
188
189 private handleMessage(data: WebSocket.Data): void {
190 try {
191 const message = JSON.parse(data.toString()) as JetstreamEvent;
192 this.eventCount++;
193 this.mostRecentEventTime = new Date(message.time_us / 1000); // Convert microseconds to milliseconds
194
195 if (message.kind === 'commit') {
196 this.handleCommitEvent(message);
197 }
198 // Ignore identity and account events
199 } catch (error) {
200 console.error('[JETSTREAM] Failed to parse message:', error);
201 }
202 }
203
204 private async handleCommitEvent(event: JetstreamCommitEvent): Promise<void> {
205 try {
206 // Convert Jetstream event to our FirehoseEvent format
207 const atUri = `at://${event.did}/${event.commit.collection}/${event.commit.rkey}`;
208
209 if (DEBUG_LOGGING) {
210 console.log(
211 `[JETSTREAM] Processing commit event: ${event.commit.operation} for ${event.did}`,
212 );
213 }
214
215 // Create a synthetic Event object that matches the ATProto firehose format
216 const syntheticEvent = {
217 event: event.commit.operation,
218 did: event.did,
219 uri: atUri,
220 cid: event.commit.cid || null,
221 record: event.commit.record,
222 seq: Math.floor(event.time_us / 1000), // Use timestamp as seq
223 time: new Date(event.time_us / 1000).toISOString(),
224 rev: event.commit.rev,
225 collection: event.commit.collection,
226 };
227
228 // Create FirehoseEvent from the synthetic event
229 const firehoseEventResult = FirehoseEvent.fromEvent(
230 syntheticEvent as any,
231 );
232 if (firehoseEventResult.isErr()) {
233 if (!firehoseEventResult.error.message.includes('is not processable')) {
234 console.error(
235 '[JETSTREAM] Failed to create FirehoseEvent:',
236 firehoseEventResult.error,
237 );
238 }
239 return;
240 }
241
242 const result = await this.firehoseEventHandler.handle(
243 firehoseEventResult.value,
244 );
245
246 if (result.isErr()) {
247 console.error(
248 '[JETSTREAM] Failed to process Jetstream event:',
249 result.error,
250 );
251 } else if (DEBUG_LOGGING) {
252 console.log(`[JETSTREAM] Successfully processed event`);
253 }
254 } catch (error) {
255 console.error('[JETSTREAM] Unhandled error in handleCommitEvent:', error);
256 }
257 }
258
259 private getFilteredCollections(): string[] {
260 const collections = this.configService.getAtProtoCollections();
261
262 return [
263 collections.card,
264 collections.collection,
265 collections.collectionLink,
266 collections.marginBookmark,
267 collections.marginCollection,
268 collections.marginCollectionItem,
269 collections.collectionLinkRemoval,
270 ];
271 }
272
273 private startEventCountLogging(): void {
274 this.logInterval = setInterval(
275 () => {
276 const now = new Date();
277 let timingInfo = '';
278
279 if (this.mostRecentEventTime) {
280 const gapSeconds = Math.floor(
281 (now.getTime() - this.mostRecentEventTime.getTime()) / 1000,
282 );
283 timingInfo = ` | Most recent event: ${this.mostRecentEventTime.toISOString()} | Gap: ${gapSeconds}s`;
284 }
285
286 console.log(
287 `[JETSTREAM] Events processed in last 10 minutes: ${this.eventCount}${timingInfo}`,
288 );
289 this.eventCount = 0; // Reset counter
290 },
291 10 * 60 * 1000,
292 ); // 10 minute intervals
293 }
294
295 private stopEventCountLogging(): void {
296 if (this.logInterval) {
297 clearInterval(this.logInterval);
298 this.logInterval = undefined;
299 }
300 }
301}