This repository has no description
fix
author
damedotblog
date
(Mar 9, 2025, 6:33 PM -0400)
commit
8cefc4ea
8cefc4ea5291d9e583dd513619bcf14dfa956957
parent
b87dca9f
b87dca9f3252fcd207d8c0ef84bb2c14c3b9b67b
+671
-210
+560
-189
app/scripts/firehose-worker.js
+560
-189
app/scripts/firehose-worker.js
···
1
-
const WebSocket = require("ws");
2
-
const path = require("path");
3
-
const fs = require("fs");
4
-
require("dotenv").config({ path: path.resolve(__dirname, "../../.env") }); // Load environment variables from .env file in app root
1
+
// jetstream-consumer.js
2
+
// Script to consume Bluesky firehose via Jetstream and save records to Supabase
5
3
6
-
const JETSTREAM_URL = "wss://jetstream2.us-west.bsky.network/subscribe";
7
-
const FLUSHING_STATUS_NSID = "im.flushing.right.now";
8
-
const LOG_FILE = path.resolve(__dirname, "flushing-logs.jsonl");
4
+
import WebSocket from 'ws';
5
+
import { createClient } from '@supabase/supabase-js';
6
+
import dotenv from 'dotenv';
7
+
import fs from 'fs';
8
+
import path from 'path';
9
+
import https from 'https';
10
+
import { promisify } from 'util';
9
11
10
-
console.log("Starting firehose worker with file storage");
11
-
console.log(`Will save records to: ${LOG_FILE}`);
12
+
// Load environment variables
13
+
dotenv.config();
12
14
13
-
// Create log directory if needed
14
-
const logDir = path.dirname(LOG_FILE);
15
-
if (!fs.existsSync(logDir)) {
16
-
fs.mkdirSync(logDir, { recursive: true });
17
-
}
15
+
// Configure Supabase client
16
+
const supabaseUrl = process.env.SUPABASE_URL;
17
+
const supabaseKey = process.env.SUPABASE_KEY;
18
+
const supabase = createClient(supabaseUrl, supabaseKey);
18
19
19
-
// Function to append records to a log file
20
-
function saveRecord(record) {
21
-
return new Promise((resolve, reject) => {
22
-
const logEntry = JSON.stringify(record) + "\n";
23
-
fs.appendFile(LOG_FILE, logEntry, (err) => {
24
-
if (err) {
25
-
console.error("Error writing to log file:", err);
26
-
reject(err);
27
-
} else {
28
-
resolve({ success: true });
29
-
}
30
-
});
31
-
});
32
-
}
20
+
// Configure Jetstream connection
21
+
const JETSTREAM_URL = 'wss://jetstream2.us-east.bsky.network/subscribe';
22
+
const WANTED_COLLECTION = 'im.flushing.right.now';
23
+
const CURSOR_FILE_PATH = path.join(process.cwd(), 'cursor.txt');
33
24
34
-
// Test file writing on startup
35
-
async function setupSystem() {
36
-
try {
37
-
console.log("Testing file logging system...");
38
-
39
-
// Test if we can write to the log file
40
-
const testRecord = {
41
-
type: "startup",
42
-
timestamp: new Date().toISOString(),
43
-
message: "Firehose worker started"
44
-
};
45
-
46
-
await saveRecord(testRecord);
47
-
console.log("✅ File logging test successful");
48
-
49
-
// Create stats counter file if it doesn't exist
50
-
const statsFile = path.resolve(__dirname, "flushing-stats.json");
51
-
if (!fs.existsSync(statsFile)) {
52
-
const initialStats = {
53
-
total_records: 0,
54
-
start_time: new Date().toISOString(),
55
-
last_update: new Date().toISOString()
56
-
};
57
-
fs.writeFileSync(statsFile, JSON.stringify(initialStats, null, 2));
58
-
console.log("Created new stats file");
59
-
}
60
-
61
-
} catch (err) {
62
-
console.error("Error setting up file logging:", err);
63
-
process.exit(1);
25
+
// Read cursor from file if it exists
26
+
function loadCursor() {
27
+
try {
28
+
if (fs.existsSync(CURSOR_FILE_PATH)) {
29
+
const cursor = fs.readFileSync(CURSOR_FILE_PATH, 'utf8').trim();
30
+
console.log(`Loaded cursor: ${cursor}`);
31
+
return cursor;
64
32
}
33
+
} catch (error) {
34
+
console.error('Error loading cursor:', error);
35
+
}
36
+
return null;
65
37
}
66
38
67
-
let messageCount = 0;
68
-
let flushingFoundCount = 0;
39
+
// Save cursor to file
40
+
function saveCursor(cursor) {
41
+
try {
42
+
fs.writeFileSync(CURSOR_FILE_PATH, cursor.toString());
43
+
} catch (error) {
44
+
console.error('Error saving cursor:', error);
45
+
}
46
+
}
69
47
70
-
function connect() {
71
-
console.log("Connecting to Jetstream");
48
+
// Utility function to add response headers to avoid rate limiting
49
+
function getRequestOptions(url) {
50
+
const parsedUrl = new URL(url);
51
+
return {
52
+
hostname: parsedUrl.hostname,
53
+
path: parsedUrl.pathname + parsedUrl.search,
54
+
headers: {
55
+
'User-Agent': 'FlushingRecorder/1.0 (https://example.com/)',
56
+
'Accept': 'application/json'
57
+
},
58
+
timeout: 10000
59
+
};
60
+
}
72
61
73
-
const wsUrl = JETSTREAM_URL + "?wantedCollections=" + FLUSHING_STATUS_NSID;
74
-
const ws = new WebSocket(wsUrl);
62
+
// Resolve a DID to a handle using multiple methods
63
+
async function resolveDIDToHandle(did) {
64
+
console.log(`Attempting to resolve DID: ${did}`);
65
+
66
+
// Make sure the DID is properly formatted
67
+
if (!did || !did.startsWith('did:')) {
68
+
console.error(`Invalid DID format: ${did}`);
69
+
return null;
70
+
}
71
+
72
+
// Method 1: Try the Bluesky API (most reliable)
73
+
try {
74
+
console.log(`Trying Bluesky API method for ${did}`);
75
+
const handle = await resolveDIDWithBskyAPI(did);
76
+
if (handle) {
77
+
console.log(`Bluesky API resolved ${did} to ${handle}`);
78
+
return handle;
79
+
}
80
+
} catch (error) {
81
+
console.error(`Bluesky API method failed for ${did}:`, error);
82
+
}
83
+
84
+
// Method 2: Try the PLC directory
85
+
try {
86
+
console.log(`Trying PLC directory method for ${did}`);
87
+
const handle = await resolveDIDWithPLC(did);
88
+
if (handle) {
89
+
console.log(`PLC directory resolved ${did} to ${handle}`);
90
+
return handle;
91
+
}
92
+
} catch (error) {
93
+
console.error(`PLC directory method failed for ${did}:`, error);
94
+
}
95
+
96
+
// Method 3: Try handle resolver (unlikely to work for DIDs, but worth a try)
97
+
try {
98
+
console.log(`Trying handle resolver method for ${did}`);
99
+
const handle = await resolveDIDWithHandleResolver(did);
100
+
if (handle) {
101
+
console.log(`Handle resolver resolved ${did} to ${handle}`);
102
+
return handle;
103
+
}
104
+
} catch (error) {
105
+
console.error(`Handle resolver method failed for ${did}:`, error);
106
+
}
107
+
108
+
console.log(`All resolution methods failed for ${did}`);
109
+
return null;
110
+
}
75
111
76
-
ws.on("open", () => {
77
-
console.log("Connected to Jetstream");
78
-
});
79
-
80
-
ws.on("message", async (data) => {
81
-
messageCount++;
82
-
if (messageCount % 1000 === 0) {
83
-
console.log("Messages:", messageCount);
112
+
// Method 1: Resolve using PLC directory
113
+
async function resolveDIDWithPLC(did) {
114
+
return new Promise((resolve, reject) => {
115
+
const url = `https://plc.directory/${encodeURIComponent(did)}`;
116
+
console.log(`Making PLC directory request to: ${url}`);
117
+
118
+
const options = getRequestOptions(url);
119
+
120
+
const req = https.get(options, (res) => {
121
+
let data = '';
122
+
123
+
// Log response status
124
+
console.log(`PLC Directory response status: ${res.statusCode}`);
125
+
126
+
res.on('data', (chunk) => {
127
+
data += chunk;
128
+
});
129
+
130
+
res.on('end', () => {
131
+
try {
132
+
console.log(`PLC raw response for ${did}: ${data.substring(0, 300)}...`);
133
+
134
+
if (res.statusCode !== 200) {
135
+
console.warn(`Failed to resolve DID ${did} with PLC: HTTP ${res.statusCode}`);
136
+
resolve(null);
137
+
return;
138
+
}
139
+
140
+
// Try to parse as JSON first
141
+
try {
142
+
const didDoc = JSON.parse(data);
143
+
144
+
// Extract handle from alsoKnownAs
145
+
if (didDoc.alsoKnownAs && Array.isArray(didDoc.alsoKnownAs) && didDoc.alsoKnownAs.length > 0) {
146
+
console.log(`Found alsoKnownAs entries: ${JSON.stringify(didDoc.alsoKnownAs)}`);
147
+
148
+
// Look for value starting with "at://"
149
+
const atValue = didDoc.alsoKnownAs.find(value => value.startsWith('at://'));
150
+
if (atValue) {
151
+
const handle = atValue.replace('at://', '');
152
+
console.log(`Successfully resolved ${did} to handle: ${handle}`);
153
+
resolve(handle);
154
+
return;
155
+
} else {
156
+
console.warn(`No 'at://' prefix found in alsoKnownAs for ${did}`);
157
+
}
158
+
} else {
159
+
console.warn(`No alsoKnownAs property found in DID document for ${did}`);
160
+
}
161
+
} catch (jsonError) {
162
+
console.log(`JSON parsing failed, trying regex: ${jsonError.message}`);
163
+
}
164
+
165
+
// If JSON parsing fails or doesn't find handle, try regex as fallback
166
+
const atMatch = data.match(/at:\/\/([^"'\\s]+)/);
167
+
if (atMatch && atMatch[1]) {
168
+
const handle = atMatch[1];
169
+
console.log(`Regex extracted handle for ${did}: ${handle}`);
170
+
resolve(handle);
171
+
return;
172
+
}
173
+
174
+
resolve(null); // No handle found
175
+
} catch (error) {
176
+
console.error(`Error parsing PLC directory response for ${did}:`, error);
177
+
resolve(null);
84
178
}
179
+
});
180
+
});
181
+
182
+
req.on('error', (error) => {
183
+
console.error(`Error fetching PLC document for ${did}:`, error);
184
+
resolve(null);
185
+
});
186
+
187
+
req.on('timeout', () => {
188
+
console.error(`PLC request timeout for ${did}`);
189
+
req.destroy();
190
+
resolve(null);
191
+
});
192
+
});
193
+
}
85
194
195
+
// Method 2: Resolve using Bluesky API
196
+
async function resolveDIDWithBskyAPI(did) {
197
+
return new Promise((resolve, reject) => {
198
+
// The Bluesky API endpoint for DID-to-handle resolution
199
+
const url = `https://api.bsky.app/xrpc/com.atproto.repo.describeRepo?repo=${encodeURIComponent(did)}`;
200
+
console.log(`Making Bluesky API request to: ${url}`);
201
+
202
+
const options = getRequestOptions(url);
203
+
204
+
const req = https.get(options, (res) => {
205
+
let data = '';
206
+
207
+
// Log response status
208
+
console.log(`Bluesky API response status: ${res.statusCode}`);
209
+
210
+
res.on('data', (chunk) => {
211
+
data += chunk;
212
+
});
213
+
214
+
res.on('end', () => {
86
215
try {
87
-
const message = JSON.parse(data.toString());
88
-
89
-
if (message.kind === "commit" &&
90
-
message.commit &&
91
-
message.commit.collection === FLUSHING_STATUS_NSID) {
92
-
93
-
flushingFoundCount++;
94
-
console.log("Found flushing record:", flushingFoundCount);
95
-
console.log(JSON.stringify(message, null, 2));
96
-
97
-
const recordPath = message.commit.collection + "/" + message.commit.rkey;
98
-
const authorDid = message.did;
99
-
const cid = message.commit.cid || "cid_" + Date.now();
100
-
101
-
let recordText = "No text found";
102
-
let recordEmoji = "🚽";
103
-
let recordCreatedAt = new Date().toISOString();
104
-
let recordType = FLUSHING_STATUS_NSID;
105
-
106
-
if (message.commit.record) {
107
-
if (message.commit.record.text) {
108
-
recordText = message.commit.record.text;
109
-
}
110
-
if (message.commit.record.emoji) {
111
-
recordEmoji = message.commit.record.emoji;
112
-
}
113
-
if (message.commit.record.createdAt) {
114
-
recordCreatedAt = message.commit.record.createdAt;
115
-
}
116
-
if (message.commit.record.$type) {
117
-
recordType = message.commit.record.$type;
118
-
}
119
-
}
120
-
121
-
console.log("Author:", authorDid);
122
-
console.log("Path:", recordPath);
123
-
console.log("Text:", recordText);
124
-
console.log("Emoji:", recordEmoji);
125
-
console.log("Created at:", recordCreatedAt);
126
-
127
-
const uri = "at://" + authorDid + "/" + recordPath;
128
-
console.log("URI:", uri);
129
-
130
-
// Save to log file
131
-
try {
132
-
// Create a record with all the info we need - matching structure
133
-
const flushingRecord = {
134
-
did: authorDid,
135
-
collection: message.commit.collection,
136
-
type: recordType,
137
-
created_at: recordCreatedAt,
138
-
emoji: recordEmoji,
139
-
text: recordText,
140
-
cid: cid,
141
-
uri: uri,
142
-
handle: "unknown", // We'll add real handle resolution later
143
-
indexed_at: new Date().toISOString()
144
-
};
145
-
146
-
// Save to file
147
-
console.log("Saving record to log file...");
148
-
const result = await saveRecord(flushingRecord);
149
-
150
-
if (result.success) {
151
-
console.log("Record saved successfully!");
152
-
153
-
// Update stats counter
154
-
try {
155
-
const statsFile = path.resolve(__dirname, "flushing-stats.json");
156
-
let stats = { total_records: 0 };
157
-
158
-
if (fs.existsSync(statsFile)) {
159
-
stats = JSON.parse(fs.readFileSync(statsFile, 'utf8'));
160
-
}
161
-
162
-
stats.total_records++;
163
-
stats.last_update = new Date().toISOString();
164
-
165
-
fs.writeFileSync(statsFile, JSON.stringify(stats, null, 2));
166
-
167
-
// Only log every 10 records to reduce noise
168
-
if (flushingFoundCount % 10 === 0) {
169
-
console.log(`Total records processed: ${stats.total_records}`);
170
-
}
171
-
} catch (statsErr) {
172
-
console.error("Error updating stats:", statsErr.message);
173
-
}
174
-
} else {
175
-
console.error("Failed to save record to file");
176
-
}
177
-
178
-
// Keep track of how many records we've processed
179
-
if (flushingFoundCount % 50 === 0) {
180
-
console.log(`Processed ${flushingFoundCount} flushing records in this session`);
181
-
}
182
-
183
-
} catch (err) {
184
-
console.error("Error processing record:", err.message);
185
-
}
186
-
}
187
-
} catch (err) {
188
-
console.error("Error processing message:", err.message);
216
+
if (res.statusCode !== 200) {
217
+
console.warn(`Failed to resolve DID ${did} with Bluesky API: HTTP ${res.statusCode}`);
218
+
resolve(null);
219
+
return;
220
+
}
221
+
222
+
const repoInfo = JSON.parse(data);
223
+
224
+
if (repoInfo && repoInfo.handle) {
225
+
const handle = repoInfo.handle;
226
+
console.log(`Successfully resolved ${did} to handle: ${handle} using Bluesky API`);
227
+
resolve(handle);
228
+
return;
229
+
}
230
+
231
+
resolve(null); // No handle found
232
+
} catch (error) {
233
+
console.error(`Error parsing Bluesky API response for ${did}:`, error);
234
+
resolve(null);
189
235
}
236
+
});
190
237
});
191
-
192
-
ws.on("error", (error) => {
193
-
console.error("WebSocket error:", error.message);
238
+
239
+
req.on('error', (error) => {
240
+
console.error(`Error fetching from Bluesky API for ${did}:`, error);
241
+
resolve(null);
194
242
});
243
+
244
+
req.on('timeout', () => {
245
+
console.error(`Bluesky API request timeout for ${did}`);
246
+
req.destroy();
247
+
resolve(null);
248
+
});
249
+
});
250
+
}
195
251
196
-
ws.on("close", () => {
197
-
console.log("Connection closed, reconnecting in 5s");
198
-
setTimeout(connect, 5000);
252
+
// Method 3: Try Bluesky official handle resolver
253
+
async function resolveDIDWithHandleResolver(did) {
254
+
try {
255
+
// First check if this is already a handle format (user.bsky.social)
256
+
if (did.includes('.') && !did.startsWith('did:')) {
257
+
console.log(`Input appears to be a handle already: ${did}`);
258
+
return did;
259
+
}
260
+
261
+
return new Promise((resolve, reject) => {
262
+
const url = `https://api.bsky.app/xrpc/com.atproto.identity.resolveHandle?handle=${encodeURIComponent(did)}`;
263
+
console.log(`Making handle resolver request to: ${url}`);
264
+
265
+
const options = getRequestOptions(url);
266
+
267
+
const req = https.get(options, (res) => {
268
+
let data = '';
269
+
270
+
// Log response status
271
+
console.log(`Handle resolver response status: ${res.statusCode}`);
272
+
273
+
res.on('data', (chunk) => {
274
+
data += chunk;
275
+
});
276
+
277
+
res.on('end', () => {
278
+
try {
279
+
if (res.statusCode !== 200) {
280
+
console.warn(`Failed to resolve ${did} with handle resolver: HTTP ${res.statusCode}`);
281
+
resolve(null);
282
+
return;
283
+
}
284
+
285
+
const response = JSON.parse(data);
286
+
287
+
if (response && response.did === did) {
288
+
// This means we resolved a handle to a DID, but we want the opposite
289
+
resolve(null);
290
+
return;
291
+
}
292
+
293
+
resolve(null); // No handle found
294
+
} catch (error) {
295
+
console.error(`Error parsing handle resolver response for ${did}:`, error);
296
+
resolve(null);
297
+
}
298
+
});
299
+
});
300
+
301
+
req.on('error', (error) => {
302
+
console.error(`Error fetching from handle resolver for ${did}:`, error);
303
+
resolve(null);
304
+
});
305
+
306
+
req.on('timeout', () => {
307
+
console.error(`Handle resolver request timeout for ${did}`);
308
+
req.destroy();
309
+
resolve(null);
310
+
});
199
311
});
312
+
} catch (error) {
313
+
console.error(`Exception in handle resolver for ${did}:`, error);
314
+
return null;
315
+
}
200
316
}
201
317
202
-
// Start the worker
203
-
async function start() {
204
-
await setupSystem();
205
-
connect();
318
+
// Process Jetstream event
319
+
async function processEvent(event) {
320
+
try {
321
+
// Save the cursor for each event we process
322
+
saveCursor(event.time_us);
323
+
324
+
// Only process commit events
325
+
if (event.kind !== 'commit') {
326
+
// Don't log skipped events to reduce noise
327
+
return;
328
+
}
329
+
330
+
// Only process commits for our target collection
331
+
if (event.commit.collection !== WANTED_COLLECTION) {
332
+
// Don't log skipped collections to reduce noise
333
+
return;
334
+
}
335
+
336
+
// Now we can log since we know it's relevant
337
+
console.log(`Processing event: ${JSON.stringify(event).substring(0, 500)}...`);
338
+
339
+
// Extract record data
340
+
const { did, time_us } = event;
341
+
const { operation, collection, rkey, record, cid } = event.commit;
342
+
343
+
console.log(`Processing ${operation} operation for DID: ${did}, collection: ${collection}, rkey: ${rkey}`);
344
+
345
+
// Skip delete operations
346
+
if (operation === 'delete') {
347
+
console.log(`Skipping delete operation`);
348
+
return;
349
+
}
350
+
351
+
// Try different approaches to get a handle
352
+
353
+
// Approach 1: Check if handle is already in the record
354
+
let handle = null;
355
+
if (record && record.handle) {
356
+
console.log(`Found handle in record: ${record.handle}`);
357
+
handle = record.handle;
358
+
}
359
+
360
+
// Approach 2: Try to resolve via APIs
361
+
if (!handle) {
362
+
console.log(`Resolving handle for DID: ${did}`);
363
+
handle = await resolveDIDToHandle(did);
364
+
365
+
if (handle) {
366
+
console.log(`Successfully resolved handle: ${handle}`);
367
+
} else {
368
+
console.log(`Failed to resolve handle for DID: ${did}`);
369
+
370
+
// Check existing records in database for this DID
371
+
try {
372
+
const { data, error } = await supabase
373
+
.from('flushing_records')
374
+
.select('handle')
375
+
.eq('did', did)
376
+
.not('handle', 'is', null)
377
+
.not('handle', 'eq', 'unknown')
378
+
.order('indexed_at', { ascending: false })
379
+
.limit(1);
380
+
381
+
if (!error && data && data.length > 0 && data[0].handle) {
382
+
handle = data[0].handle;
383
+
console.log(`Found handle in database for DID ${did}: ${handle}`);
384
+
} else {
385
+
console.log(`No existing handle found in database for DID: ${did}`);
386
+
handle = 'unknown'; // Set explicitly to unknown
387
+
}
388
+
} catch (dbError) {
389
+
console.error(`Error checking database for existing handle: ${dbError.message}`);
390
+
handle = 'unknown'; // Set explicitly if DB query fails
391
+
}
392
+
}
393
+
}
394
+
395
+
// Double-check that we have a handle, default to 'unknown' if not
396
+
if (!handle) {
397
+
console.log(`No handle could be resolved for DID ${did}, using 'unknown'`);
398
+
handle = 'unknown';
399
+
}
400
+
401
+
// Prepare data for insertion - DO NOT include id field at all
402
+
const recordData = {
403
+
did,
404
+
collection,
405
+
type: record?.$type,
406
+
created_at: record?.createdAt || new Date().toISOString(),
407
+
emoji: record?.emoji,
408
+
text: record?.text,
409
+
cid,
410
+
uri: `at://${did}/${collection}/${rkey}`,
411
+
indexed_at: new Date().toISOString(),
412
+
handle: handle // This will never be null or undefined now
413
+
};
414
+
415
+
console.log(`Preparing to insert/update record with handle '${recordData.handle}'`);
416
+
417
+
// First check if the record already exists
418
+
const { data: existingData, error: checkError } = await supabase
419
+
.from('flushing_records')
420
+
.select('id, handle')
421
+
.eq('uri', recordData.uri)
422
+
.limit(1);
423
+
424
+
let result;
425
+
426
+
if (checkError) {
427
+
console.error(`Error checking if record exists: ${checkError.message}`);
428
+
return;
429
+
}
430
+
431
+
// If record exists, update it
432
+
if (existingData && existingData.length > 0) {
433
+
console.log(`Record with URI ${recordData.uri} already exists, updating`);
434
+
435
+
// If existing record has a valid handle and current handle is 'unknown', use the existing handle
436
+
if (existingData[0].handle && existingData[0].handle !== 'unknown' && recordData.handle === 'unknown') {
437
+
console.log(`Keeping existing handle '${existingData[0].handle}' instead of replacing with 'unknown'`);
438
+
recordData.handle = existingData[0].handle;
439
+
}
440
+
441
+
const { data, error } = await supabase
442
+
.from('flushing_records')
443
+
.update(recordData)
444
+
.eq('uri', recordData.uri);
445
+
446
+
result = { data, error };
447
+
}
448
+
// Otherwise insert a new record
449
+
else {
450
+
console.log(`Record with URI ${recordData.uri} doesn't exist, inserting with handle: ${recordData.handle}`);
451
+
const { data, error } = await supabase
452
+
.from('flushing_records')
453
+
.insert(recordData);
454
+
455
+
result = { data, error };
456
+
}
457
+
458
+
// Check the result of the operation
459
+
if (result.error) {
460
+
console.error(`Error saving record to Supabase: ${result.error.message}`);
461
+
console.error(`Failed record data: ${JSON.stringify(recordData)}`);
462
+
} else {
463
+
console.log(`Successfully saved record: ${recordData.uri} (handle: ${recordData.handle})`);
464
+
}
465
+
466
+
} catch (error) {
467
+
console.error(`Error processing event: ${error.message}`);
468
+
console.error(error.stack);
469
+
}
206
470
}
207
471
208
-
// Run the worker
209
-
start();
472
+
// Process 'identity' events when they come through the firehose
473
+
async function processIdentityEvent(event) {
474
+
try {
475
+
if (event.kind !== 'identity' || !event.identity) {
476
+
return;
477
+
}
478
+
479
+
const { did, handle } = event.identity;
480
+
481
+
if (did && handle) {
482
+
// Check if we have any records with this DID that have 'unknown' handles
483
+
try {
484
+
const { data, error } = await supabase
485
+
.from('flushing_records')
486
+
.select('uri')
487
+
.eq('did', did)
488
+
.eq('handle', 'unknown');
489
+
490
+
if (!error && data && data.length > 0) {
491
+
console.log(`Found ${data.length} records with DID ${did} and unknown handle. Updating to ${handle}...`);
492
+
493
+
// Update all matching records with the new handle
494
+
const { updateData, updateError } = await supabase
495
+
.from('flushing_records')
496
+
.update({ handle })
497
+
.eq('did', did)
498
+
.eq('handle', 'unknown');
499
+
500
+
if (updateError) {
501
+
console.error(`Error updating records with DID ${did}: ${updateError.message}`);
502
+
} else {
503
+
console.log(`Successfully updated handle for records with DID ${did} to ${handle}`);
504
+
}
505
+
}
506
+
} catch (dbError) {
507
+
console.error(`Error updating unknown handles: ${dbError.message}`);
508
+
}
509
+
}
510
+
} catch (error) {
511
+
console.error(`Error processing identity event: ${error.message}`);
512
+
}
513
+
}
514
+
515
+
// Connect to Jetstream and process events
516
+
function connectToJetstream() {
517
+
const cursor = loadCursor();
518
+
519
+
// Building the URL with query parameters - now include identity events!
520
+
// Including identity events will help us maintain DID-to-handle mapping
521
+
let url = `${JETSTREAM_URL}?wantedCollections=${WANTED_COLLECTION}`;
522
+
if (cursor) {
523
+
// Subtract a few seconds (in microseconds) to ensure no gaps
524
+
const rewindCursor = parseInt(cursor) - 5000000; // 5 seconds in microseconds
525
+
url += `&cursor=${rewindCursor}`;
526
+
}
527
+
528
+
console.log(`Connecting to Jetstream: ${url}`);
529
+
530
+
const ws = new WebSocket(url);
531
+
532
+
ws.on('open', () => {
533
+
console.log('Connected to Jetstream');
534
+
});
535
+
536
+
ws.on('message', async (data) => {
537
+
try {
538
+
const event = JSON.parse(data.toString());
539
+
540
+
// Process identity events to keep our DID-to-handle mapping up to date
541
+
if (event.kind === 'identity') {
542
+
await processIdentityEvent(event);
543
+
}
544
+
545
+
// Process other events normally
546
+
await processEvent(event);
547
+
} catch (error) {
548
+
console.error('Error parsing message:', error);
549
+
// Don't log message data to reduce noise
550
+
}
551
+
});
552
+
553
+
ws.on('error', (error) => {
554
+
console.error('WebSocket error:', error);
555
+
setTimeout(connectToJetstream, 5000); // Reconnect after 5 seconds
556
+
});
557
+
558
+
ws.on('close', () => {
559
+
console.log('Connection closed. Attempting to reconnect...');
560
+
setTimeout(connectToJetstream, 5000); // Reconnect after 5 seconds
561
+
});
562
+
563
+
// Heartbeat to keep the connection alive
564
+
const interval = setInterval(() => {
565
+
if (ws.readyState === WebSocket.OPEN) {
566
+
ws.ping();
567
+
} else {
568
+
clearInterval(interval);
569
+
}
570
+
}, 30000);
571
+
}
210
572
211
-
process.on("SIGINT", () => {
212
-
console.log("Shutting down");
573
+
// Start the application
574
+
function start() {
575
+
console.log('Starting Jetstream consumer...');
576
+
connectToJetstream();
577
+
578
+
// Handle process termination
579
+
process.on('SIGINT', () => {
580
+
console.log('Process terminated. Exiting...');
213
581
process.exit(0);
214
-
});
582
+
});
583
+
}
584
+
585
+
start();
+61
-19
app/src/app/api/bluesky/feed-direct/route.ts
+61
-19
app/src/app/api/bluesky/feed-direct/route.ts
···
87
87
88
88
entries = data || [];
89
89
} else {
90
-
// Main query: get the most recent entries
91
-
// Use a direct SQL query for maximum reliability
92
-
const { data, error } = await supabase.rpc('get_latest_entries', {
93
-
max_entries: MAX_ENTRIES
94
-
});
90
+
// First try: Direct raw SQL query via executeRaw (most reliable)
91
+
try {
92
+
// Use a direct SQL query to completely bypass any ORM and query builder caching
93
+
const rawQuery = `
94
+
SELECT * FROM flushing_records
95
+
ORDER BY id DESC
96
+
LIMIT ${MAX_ENTRIES}
97
+
`;
98
+
99
+
console.log('Executing direct SQL query:', rawQuery);
100
+
101
+
const { data: directData, error: directError } = await supabase.rpc(
102
+
'execute_raw_query',
103
+
{ raw_query: rawQuery }
104
+
);
105
+
106
+
if (directError) {
107
+
console.error('Raw SQL query error:', directError);
108
+
// Continue to next approach
109
+
} else if (directData && Array.isArray(directData) && directData.length > 0) {
110
+
console.log(`✅ Direct SQL query successful, found ${directData.length} entries`);
111
+
entries = directData;
112
+
113
+
// We got data, return early
114
+
return entries;
115
+
}
116
+
} catch (rawError) {
117
+
console.error('Exception executing raw SQL:', rawError);
118
+
// Continue to next approach
119
+
}
95
120
96
-
if (error) {
97
-
console.error('RPC function error:', error);
121
+
// Second try: Using the RPC function approach
122
+
try {
123
+
console.log('Trying RPC function approach');
98
124
99
-
// Fallback to regular query if RPC fails
100
-
console.log('Falling back to regular query');
101
-
const { data: fallbackData, error: fallbackError } = await supabase
102
-
.from('flushing_records')
103
-
.select('*')
104
-
.order('id', { ascending: false })
105
-
.limit(MAX_ENTRIES);
125
+
const { data, error } = await supabase.rpc('get_latest_entries', {
126
+
max_entries: MAX_ENTRIES
127
+
});
128
+
129
+
if (error) {
130
+
console.error('RPC function error:', error);
131
+
// Continue to fallback approach
132
+
} else if (data && Array.isArray(data) && data.length > 0) {
133
+
console.log(`✅ RPC function query successful, found ${data.length} entries`);
134
+
entries = data;
106
135
107
-
if (fallbackError) {
108
-
throw new Error(`Fallback query error: ${fallbackError.message}`);
136
+
// We got data, return early
137
+
return entries;
109
138
}
139
+
} catch (rpcError) {
140
+
console.error('Exception in RPC function:', rpcError);
141
+
// Continue to fallback approach
142
+
}
143
+
144
+
// Final fallback: Standard query builder approach
145
+
console.log('Falling back to standard query builder');
146
+
const { data: fallbackData, error: fallbackError } = await supabase
147
+
.from('flushing_records')
148
+
.select('*')
149
+
.order('id', { ascending: false })
150
+
.limit(MAX_ENTRIES);
110
151
111
-
entries = fallbackData || [];
112
-
} else {
113
-
entries = data || [];
152
+
if (fallbackError) {
153
+
throw new Error(`Fallback query error: ${fallbackError.message}`);
114
154
}
155
+
156
+
entries = fallbackData || [];
115
157
}
116
158
117
159
console.log(`Query returned ${entries.length} entries`);
+24
-2
app/src/app/page.tsx
+24
-2
app/src/app/page.tsx
···
176
176
setText('is ');
177
177
setSuccess('Your flushing status has been updated!');
178
178
179
+
// Create a temporary entry to display immediately
180
+
if (result && result.uri && result.cid) {
181
+
const tempEntry: FlushingEntry = {
182
+
id: `temp-${Date.now()}`, // Create a temporary ID
183
+
uri: result.uri,
184
+
cid: result.cid,
185
+
authorDid: did,
186
+
authorHandle: handle,
187
+
text: formattedText,
188
+
emoji: selectedEmoji,
189
+
createdAt: new Date().toISOString()
190
+
};
191
+
192
+
console.log('Adding temporary entry to feed:', tempEntry);
193
+
194
+
// Add the temporary entry to the top of the feed
195
+
setEntries(prevEntries => [tempEntry, ...prevEntries]);
196
+
197
+
// Also mark it as a new entry for animation
198
+
setNewEntryIds(new Set([tempEntry.id]));
199
+
}
200
+
179
201
// Close status form after successful submission
180
202
setTimeout(() => {
181
203
setStatusOpen(false);
182
204
}, 2000);
183
205
184
-
// Refresh the feed to show the new status
206
+
// Still refresh the feed after a delay to get the actual database entry
185
207
setTimeout(() => {
186
208
fetchLatestEntries(true);
187
-
}, 1000);
209
+
}, 3000);
188
210
} catch (err: any) {
189
211
console.error('Failed to update status:', err);
190
212
setStatusError(`Failed to update status: ${err.message || 'Unknown error'}`);
+26
execute_raw_query.sql
+26
execute_raw_query.sql
···
1
+
-- Create a function to execute raw SQL queries safely
2
+
-- This allows direct SQL execution for better performance and reliability
3
+
4
+
CREATE OR REPLACE FUNCTION execute_raw_query(raw_query TEXT)
5
+
RETURNS JSONB
6
+
LANGUAGE plpgsql
7
+
SECURITY DEFINER -- This runs with the privileges of the function creator
8
+
AS $$
9
+
DECLARE
10
+
result JSONB;
11
+
BEGIN
12
+
-- Only allow SELECT queries for security
13
+
IF position('SELECT' in upper(raw_query)) != 1 THEN
14
+
RAISE EXCEPTION 'Only SELECT queries are allowed';
15
+
END IF;
16
+
17
+
-- Execute the query and convert result to JSON
18
+
EXECUTE 'SELECT json_agg(t) FROM (' || raw_query || ') t' INTO result;
19
+
20
+
-- Return empty array instead of null if no results
21
+
RETURN COALESCE(result, '[]'::jsonb);
22
+
END;
23
+
$$;
24
+
25
+
-- Test the function
26
+
SELECT execute_raw_query('SELECT id, did, handle, text FROM flushing_records ORDER BY id DESC LIMIT 3');