flo-bit.dev
[READ-ONLY] Mirror of https://github.com/flo-bit/skywatched-backend. backend/jetstream consumer for skywatched.app
skywatched.app
commit
author
Florian
date
(Jan 22, 2025, 10:17 AM +0100)
commit
2b676f69
2b676f692ec80543b4eb1943b18c7baea8fae0b1
parent
5f09e79b
5f09e79b99dc23337f63768c08022757e7e4a7af
+532
-270
+1
.prettierrc
+1
.prettierrc
···
1
+
{}
+133
-67
src/api.ts
+133
-67
src/api.ts
···
1
1
// server.ts
2
-
import { backfillUserIfNecessary, deleteAllByUser, getAuthorDids, getLatestTimestamp, getMostRecentRecords, getRecentRecordsByItemRef, getRecentRecordsByUser, getRecord } from './database';
2
+
import { processLikesApiCall } from "./api/likes";
3
+
import {
4
+
backfillUserIfNecessary,
5
+
deleteAllByUser,
6
+
getAuthorDids,
7
+
getLatestTimestamp,
8
+
getMostRecentRecords,
9
+
getRecentRecordsByItemRef,
10
+
getRecentRecordsByUser,
11
+
getRecord,
12
+
} from "./database";
3
13
4
14
// Define your API routes and handlers
5
15
export const handler = async (req: Request): Promise<Response> => {
6
-
const url = new URL(req.url);
7
-
const path = url.pathname;
8
-
const method = req.method;
16
+
const url = new URL(req.url);
17
+
const path = url.pathname;
18
+
const method = req.method;
9
19
10
-
11
-
if (method === "GET") {
12
-
if (path === "/api/latest-timestamp") {
13
-
const timestamp = getLatestTimestamp();
14
-
return new Response(JSON.stringify({ latest_timestamp: timestamp }), { status: 200, headers: { "Content-Type": "application/json" } });
15
-
}
16
-
17
-
if (path === "/api/most-recent-records") {
18
-
const limit = parseInt(url.searchParams.get("limit") ?? "50");
19
-
const cursor = url.searchParams.get("cursor");
20
-
const records = getMostRecentRecords(limit, cursor);
21
-
return new Response(JSON.stringify(records), { status: 200, headers: { "Content-Type": "application/json" } });
22
-
}
23
-
24
-
if (path === "/api/recent-records-by-user") {
25
-
const user_did = url.searchParams.get("did");
20
+
if (method === "GET") {
21
+
if (path === "/api/latest-timestamp") {
22
+
const timestamp = getLatestTimestamp();
23
+
return new Response(JSON.stringify({ latest_timestamp: timestamp }), {
24
+
status: 200,
25
+
headers: { "Content-Type": "application/json" },
26
+
});
27
+
}
26
28
27
-
const limit = parseInt(url.searchParams.get("limit") ?? "50");
28
-
const cursor = url.searchParams.get("cursor");
29
-
if (!user_did) {
30
-
return new Response(JSON.stringify({ error: "did is required" }), { status: 400, headers: { "Content-Type": "application/json" } });
31
-
}
32
-
let records = getRecentRecordsByUser(user_did, limit, cursor);
29
+
if (path === "/api/most-recent-records") {
30
+
const limit = parseInt(url.searchParams.get("limit") ?? "50");
31
+
const cursor = url.searchParams.get("cursor");
32
+
const records = getMostRecentRecords(limit, cursor);
33
+
return new Response(JSON.stringify(records), {
34
+
status: 200,
35
+
headers: { "Content-Type": "application/json" },
36
+
});
37
+
}
33
38
34
-
if(records.length === 0) {
35
-
await backfillUserIfNecessary(user_did);
36
-
records = getRecentRecordsByUser(user_did, limit, cursor);
37
-
}
39
+
if (path === "/api/recent-records-by-user") {
40
+
const user_did = url.searchParams.get("did");
38
41
39
-
return new Response(JSON.stringify(records), { status: 200, headers: { "Content-Type": "application/json" } });
40
-
}
41
-
42
-
if (path === "/api/recent-records-by-item") {
43
-
const item_ref = url.searchParams.get("ref");
44
-
const item_value = url.searchParams.get("value");
45
-
const limit = parseInt(url.searchParams.get("limit") ?? "50");
46
-
const cursor = url.searchParams.get("cursor");
47
-
if (!item_ref || !item_value) {
48
-
return new Response(JSON.stringify({ error: "item_ref and item_value are required" }), { status: 400, headers: { "Content-Type": "application/json" } });
49
-
}
50
-
const records = getRecentRecordsByItemRef(item_ref, item_value, limit, cursor);
51
-
return new Response(JSON.stringify(records), { status: 200, headers: { "Content-Type": "application/json" } });
52
-
}
42
+
const limit = parseInt(url.searchParams.get("limit") ?? "50");
43
+
const cursor = url.searchParams.get("cursor");
44
+
if (!user_did) {
45
+
return new Response(JSON.stringify({ error: "did is required" }), {
46
+
status: 400,
47
+
headers: { "Content-Type": "application/json" },
48
+
});
49
+
}
50
+
let records = getRecentRecordsByUser(user_did, limit, cursor);
53
51
54
-
if(path === "/api/record") {
55
-
const uri = url.searchParams.get("uri");
56
-
if (!uri) {
57
-
return new Response(JSON.stringify({ error: "uri is required" }), { status: 400, headers: { "Content-Type": "application/json" } });
58
-
}
59
-
const record = getRecord(uri);
60
-
return new Response(JSON.stringify(record), { status: 200, headers: { "Content-Type": "application/json" } });
61
-
}
52
+
if (records.length === 0) {
53
+
await backfillUserIfNecessary(user_did);
54
+
records = getRecentRecordsByUser(user_did, limit, cursor);
55
+
}
62
56
63
-
if(path === "/api/author-dids") {
64
-
const author_dids = getAuthorDids();
65
-
return new Response(JSON.stringify(author_dids), { status: 200, headers: { "Content-Type": "application/json" } });
66
-
}
57
+
return new Response(JSON.stringify(records), {
58
+
status: 200,
59
+
headers: { "Content-Type": "application/json" },
60
+
});
61
+
}
67
62
68
-
if(path === "/api/refresh-user") {
69
-
const did = url.searchParams.get("did");
70
-
if (!did) {
71
-
return new Response(JSON.stringify({ error: "did is required" }), { status: 400, headers: { "Content-Type": "application/json" } });
72
-
}
73
-
deleteAllByUser(did);
74
-
await backfillUserIfNecessary(did);
75
-
return new Response(JSON.stringify({ success: true }), { status: 200, headers: { "Content-Type": "application/json" } });
76
-
}
63
+
if (path === "/api/recent-records-by-item") {
64
+
const item_ref = url.searchParams.get("ref");
65
+
const item_value = url.searchParams.get("value");
66
+
const limit = parseInt(url.searchParams.get("limit") ?? "50");
67
+
const cursor = url.searchParams.get("cursor");
68
+
if (!item_ref || !item_value) {
69
+
return new Response(
70
+
JSON.stringify({ error: "item_ref and item_value are required" }),
71
+
{ status: 400, headers: { "Content-Type": "application/json" } }
72
+
);
77
73
}
74
+
const records = getRecentRecordsByItemRef(
75
+
item_ref,
76
+
item_value,
77
+
limit,
78
+
cursor
79
+
);
80
+
return new Response(JSON.stringify(records), {
81
+
status: 200,
82
+
headers: { "Content-Type": "application/json" },
83
+
});
84
+
}
78
85
79
-
// Handle 404 Not Found
80
-
return new Response(JSON.stringify({ error: "Not Found" }), { status: 404, headers: { "Content-Type": "application/json" } });
86
+
if (path === "/api/record") {
87
+
const uri = url.searchParams.get("uri");
88
+
if (!uri) {
89
+
return new Response(JSON.stringify({ error: "uri is required" }), {
90
+
status: 400,
91
+
headers: { "Content-Type": "application/json" },
92
+
});
93
+
}
94
+
const record = getRecord(uri);
95
+
return new Response(JSON.stringify(record), {
96
+
status: 200,
97
+
headers: { "Content-Type": "application/json" },
98
+
});
99
+
}
100
+
101
+
if (path === "/api/author-dids") {
102
+
const author_dids = getAuthorDids();
103
+
return new Response(JSON.stringify(author_dids), {
104
+
status: 200,
105
+
headers: { "Content-Type": "application/json" },
106
+
});
107
+
}
108
+
109
+
if (path === "/api/refresh-user") {
110
+
const did = url.searchParams.get("did");
111
+
if (!did) {
112
+
return new Response(JSON.stringify({ error: "did is required" }), {
113
+
status: 400,
114
+
headers: { "Content-Type": "application/json" },
115
+
});
116
+
}
117
+
deleteAllByUser(did);
118
+
await backfillUserIfNecessary(did);
119
+
return new Response(JSON.stringify({ success: true }), {
120
+
status: 200,
121
+
headers: { "Content-Type": "application/json" },
122
+
});
123
+
}
124
+
}
125
+
126
+
if (path.startsWith("/api/likes")) {
127
+
const response = await processLikesApiCall({
128
+
method,
129
+
path,
130
+
request: req,
131
+
url,
132
+
});
133
+
return (
134
+
response ??
135
+
new Response(JSON.stringify({ error: "Not Found" }), {
136
+
status: 404,
137
+
headers: { "Content-Type": "application/json" },
138
+
})
139
+
);
140
+
}
141
+
142
+
// Handle 404 Not Found
143
+
return new Response(JSON.stringify({ error: "Not Found" }), {
144
+
status: 404,
145
+
headers: { "Content-Type": "application/json" },
146
+
});
81
147
};
+49
src/api/likes.ts
+49
src/api/likes.ts
···
1
+
import { getAllUsersWithLikes, getLikesByUser } from "../db/likes";
2
+
import { ApiCall } from "./types";
3
+
4
+
const calls: {
5
+
[key: string]: {
6
+
method: string;
7
+
handler: (call: ApiCall) => Promise<Response>;
8
+
};
9
+
} = {
10
+
"/api/likes/user": {
11
+
method: "GET",
12
+
handler: async (call: ApiCall) => {
13
+
const user_did = call.url.searchParams.get("did");
14
+
if (!user_did) {
15
+
return new Response(JSON.stringify({ error: "Missing did" }), {
16
+
status: 400,
17
+
headers: { "Content-Type": "application/json" },
18
+
});
19
+
}
20
+
21
+
const likes = await getLikesByUser(user_did);
22
+
return new Response(JSON.stringify(likes), {
23
+
status: 200,
24
+
headers: { "Content-Type": "application/json" },
25
+
});
26
+
},
27
+
},
28
+
29
+
"/api/likes/users": {
30
+
method: "GET",
31
+
handler: async (call: ApiCall) => {
32
+
const users = await getAllUsersWithLikes();
33
+
return new Response(JSON.stringify(users), {
34
+
status: 200,
35
+
headers: { "Content-Type": "application/json" },
36
+
});
37
+
},
38
+
},
39
+
};
40
+
41
+
export async function processLikesApiCall(
42
+
call: ApiCall
43
+
): Promise<Response | undefined> {
44
+
if (calls[call.path] && calls[call.path].method === call.method) {
45
+
return calls[call.path].handler(call);
46
+
}
47
+
48
+
return undefined;
49
+
}
+6
src/api/types.ts
+6
src/api/types.ts
+124
-92
src/database.ts
+124
-92
src/database.ts
···
104
104
record_likes?: number;
105
105
};
106
106
107
-
type LikeRecord = {
108
-
uri: string;
109
-
author_did: string;
110
-
subject_cid: string;
111
-
subject_uri: string;
112
-
createdAt: string;
113
-
};
114
107
115
108
function createTables() {
116
109
db.run(`
···
190
183
191
184
// Function to get the latest timestamp (either creation or update)
192
185
export function getLatestTimestamp(): string | null {
193
-
const row = db.query('SELECT MAX(indexedAt) AS latest FROM latest_indexedAt;').get() as { latest: string } | undefined;
186
+
const row = db
187
+
.query("SELECT MAX(indexedAt) AS latest FROM latest_indexedAt;")
188
+
.get() as { latest: string } | undefined;
194
189
return row?.latest ?? null;
195
190
}
196
191
197
192
export function setLatestTimestamp(indexedAt: string): void {
198
-
db.query('DELETE FROM latest_indexedAt').run();
199
-
db.query('INSERT OR REPLACE INTO latest_indexedAt (indexedAt) VALUES (?)').run(indexedAt);
193
+
db.query("DELETE FROM latest_indexedAt").run();
194
+
db.query(
195
+
"INSERT OR REPLACE INTO latest_indexedAt (indexedAt) VALUES (?)"
196
+
).run(indexedAt);
200
197
}
201
198
202
199
export function getRecord(uri: string): MainRecord | null {
203
-
const row = db.query('SELECT * FROM records WHERE uri = ?').get(uri) as Record | undefined;
200
+
const row = db.query("SELECT * FROM records WHERE uri = ?").get(uri) as
201
+
| Record
202
+
| undefined;
204
203
return row ? transformRecord(row) : null;
205
204
}
206
205
207
206
export function getAuthorDids(): string[] {
208
-
const rows = db.query('SELECT DISTINCT author_did FROM records').all() as { author_did: string }[];
209
-
return rows.map(row => row.author_did);
207
+
const rows = db.query("SELECT DISTINCT author_did FROM records").all() as {
208
+
author_did: string;
209
+
}[];
210
+
return rows.map((row) => row.author_did);
210
211
}
211
212
212
213
export function createRecord(record: MainRecord): void {
···
232
233
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
233
234
`;
234
235
const params = [
235
-
record.uri, record.cid,
236
-
record.author.did, record.author.handle, record.author.displayName, record.author.avatar,
237
-
record.indexedAt, record.createdAt, record.updatedAt,
238
-
record.record.$type,
239
-
record.record.item.ref, record.record.item.value,
240
-
record.record.note?.value ?? null, record.record.note?.createdAt ?? null, record.record.note?.updatedAt ?? null,
241
-
record.record.rating?.value ?? null, record.record.rating?.createdAt ?? null,
236
+
record.uri,
237
+
record.cid,
238
+
record.author.did,
239
+
record.author.handle,
240
+
record.author.displayName,
241
+
record.author.avatar,
242
+
record.indexedAt,
243
+
record.createdAt,
244
+
record.updatedAt,
245
+
record.record.$type,
246
+
record.record.item.ref,
247
+
record.record.item.value,
248
+
record.record.note?.value ?? null,
249
+
record.record.note?.createdAt ?? null,
250
+
record.record.note?.updatedAt ?? null,
251
+
record.record.rating?.value ?? null,
252
+
record.record.rating?.createdAt ?? null,
242
253
243
-
record.record.metadata?.title ?? null, record.record.metadata?.poster_path ?? null, record.record.metadata?.backdrop_path ?? null,
244
-
record.record.metadata?.tagline ?? null, record.record.metadata?.overview ?? null, record.record.metadata?.genres?.join(',') ?? null,
245
-
record.record.metadata?.release_date ?? null,
254
+
record.record.metadata?.title ?? null,
255
+
record.record.metadata?.poster_path ?? null,
256
+
record.record.metadata?.backdrop_path ?? null,
257
+
record.record.metadata?.tagline ?? null,
258
+
record.record.metadata?.overview ?? null,
259
+
record.record.metadata?.genres?.join(",") ?? null,
260
+
record.record.metadata?.release_date ?? null,
246
261
247
-
record.record.crosspost?.uri ?? null,
248
-
record.record.crosspost?.likes ?? null,
249
-
record.record.crosspost?.reposts ?? null,
250
-
record.record.crosspost?.replies ?? null
262
+
record.record.crosspost?.uri ?? null,
263
+
record.record.crosspost?.likes ?? null,
264
+
record.record.crosspost?.reposts ?? null,
265
+
record.record.crosspost?.replies ?? null,
251
266
];
252
-
db.query(sql).run(...params as SQLQueryBindings[]);
267
+
db.query(sql).run(...(params as SQLQueryBindings[]));
253
268
}
254
269
255
270
export function transformRecord(record: Record): MainRecord {
···
271
286
ref: record.record_item_ref,
272
287
value: record.record_item_value,
273
288
},
274
-
note: record.record_note_value ? {
275
-
value: record.record_note_value,
276
-
createdAt: record.record_note_createdAt ?? new Date().toISOString(),
277
-
updatedAt: record.record_note_updatedAt ?? new Date().toISOString(),
278
-
} : undefined,
279
-
rating: record.record_rating_value ? {
280
-
value: record.record_rating_value,
281
-
createdAt: record.record_rating_createdAt ?? new Date().toISOString(),
282
-
} : undefined,
289
+
note: record.record_note_value
290
+
? {
291
+
value: record.record_note_value,
292
+
createdAt: record.record_note_createdAt ?? new Date().toISOString(),
293
+
updatedAt: record.record_note_updatedAt ?? new Date().toISOString(),
294
+
}
295
+
: undefined,
296
+
rating: record.record_rating_value
297
+
? {
298
+
value: record.record_rating_value,
299
+
createdAt:
300
+
record.record_rating_createdAt ?? new Date().toISOString(),
301
+
}
302
+
: undefined,
283
303
metadata: {
284
-
title: record.record_metadata_title ?? '',
285
-
poster_path: record.record_metadata_poster_path ?? '',
286
-
backdrop_path: record.record_metadata_backdrop_path ?? '',
287
-
tagline: record.record_metadata_tagline ?? '',
288
-
overview: record.record_metadata_overview ?? '',
289
-
genres: record.record_metadata_genres?.split(',') ?? [],
290
-
release_date: record.record_metadata_release_date ?? '',
304
+
title: record.record_metadata_title ?? "",
305
+
poster_path: record.record_metadata_poster_path ?? "",
306
+
backdrop_path: record.record_metadata_backdrop_path ?? "",
307
+
tagline: record.record_metadata_tagline ?? "",
308
+
overview: record.record_metadata_overview ?? "",
309
+
genres: record.record_metadata_genres?.split(",") ?? [],
310
+
release_date: record.record_metadata_release_date ?? "",
291
311
},
292
312
likes: record.record_likes ?? 0,
293
-
}
313
+
},
294
314
};
295
315
}
296
316
297
317
// Function to get the most recent created records with pagination
298
-
export function getMostRecentRecords(limit: number = 100, cursor: string | null = null): MainRecord[] {
299
-
let sql = 'SELECT * FROM records WHERE 1=1';
318
+
export function getMostRecentRecords(
319
+
limit: number = 100,
320
+
cursor: string | null = null
321
+
): MainRecord[] {
322
+
let sql = "SELECT * FROM records WHERE 1=1";
300
323
const params: (string | number)[] = [];
301
324
302
325
if (cursor) {
303
-
sql += ' AND createdAt < ?';
326
+
sql += " AND createdAt < ?";
304
327
params.push(cursor);
305
328
}
306
329
307
-
sql += ' ORDER BY createdAt DESC LIMIT ?';
330
+
sql += " ORDER BY createdAt DESC LIMIT ?";
308
331
params.push(limit);
309
332
310
333
const rows = db.query(sql).all(...params) as Record[];
311
-
return rows.map(row => transformRecord(row));
334
+
return rows.map((row) => transformRecord(row));
312
335
}
313
336
314
337
// // Function to update a record
···
320
343
// }
321
344
322
345
// Function to get the most recent records for a specific user
323
-
export function getRecentRecordsByUser(user_did: string, limit: number = 100, cursor: string | null = null): MainRecord[] {
324
-
let sql = 'SELECT * FROM records WHERE author_did = ?';
346
+
export function getRecentRecordsByUser(
347
+
user_did: string,
348
+
limit: number = 100,
349
+
cursor: string | null = null
350
+
): MainRecord[] {
351
+
let sql = "SELECT * FROM records WHERE author_did = ?";
325
352
const params: (string | number)[] = [user_did];
326
353
327
354
if (cursor) {
328
-
sql += ' AND createdAt < ?';
355
+
sql += " AND createdAt < ?";
329
356
params.push(cursor);
330
357
}
331
358
332
-
sql += ' ORDER BY createdAt DESC LIMIT ?';
359
+
sql += " ORDER BY createdAt DESC LIMIT ?";
333
360
params.push(limit);
334
361
335
362
const rows = db.query(sql).all(...params) as Record[];
336
-
return rows.map(row => transformRecord(row));
363
+
return rows.map((row) => transformRecord(row));
337
364
}
338
365
339
366
// Function to get the most recent records for a specific item ref and value with pagination
340
-
export function getRecentRecordsByItemRef(item_ref: string, item_value: string, limit: number = 100, cursor: string | null = null): MainRecord[] {
341
-
let sql = 'SELECT * FROM records WHERE record_item_ref = ? AND record_item_value = ?';
367
+
export function getRecentRecordsByItemRef(
368
+
item_ref: string,
369
+
item_value: string,
370
+
limit: number = 100,
371
+
cursor: string | null = null
372
+
): MainRecord[] {
373
+
let sql =
374
+
"SELECT * FROM records WHERE record_item_ref = ? AND record_item_value = ?";
342
375
const params: (string | number)[] = [item_ref, item_value];
343
376
344
377
if (cursor) {
345
-
sql += ' AND createdAt < ?';
378
+
sql += " AND createdAt < ?";
346
379
params.push(cursor);
347
380
}
348
381
349
-
sql += ' ORDER BY createdAt DESC LIMIT ?';
382
+
sql += " ORDER BY createdAt DESC LIMIT ?";
350
383
params.push(limit);
351
384
352
385
const rows = db.query(sql).all(...params) as Record[];
353
-
return rows.map(row => transformRecord(row));
386
+
return rows.map((row) => transformRecord(row));
354
387
}
355
388
356
389
// Function to delete a record
···
360
393
// }
361
394
362
395
export function deleteAllByUser(user_did: string): void {
363
-
const sql = 'DELETE FROM records WHERE author_did = ?';
396
+
const sql = "DELETE FROM records WHERE author_did = ?";
364
397
db.query(sql).run(user_did);
365
398
}
366
399
400
+
export function deleteRecord(uri: string): void {
401
+
const sql = "DELETE FROM records WHERE uri = ?";
402
+
db.query(sql).run(uri);
403
+
}
404
+
367
405
export function deleteAllRecords(): void {
368
-
const sql = 'DELETE FROM records';
406
+
const sql = "DELETE FROM records";
369
407
db.query(sql).run();
370
408
}
371
409
372
410
export function recreateTables() {
373
-
db.run('DROP TABLE IF EXISTS records');
374
-
db.run('DROP TABLE IF EXISTS latest_indexedAt');
375
-
db.run('DROP TABLE IF EXISTS likes');
411
+
db.run("DROP TABLE IF EXISTS records");
412
+
db.run("DROP TABLE IF EXISTS latest_indexedAt");
413
+
db.run("DROP TABLE IF EXISTS likes");
376
414
377
415
// create the tables again
378
416
createTables();
···
381
419
export async function backfillUserIfNecessary(did: string): Promise<void> {
382
420
// check if at least one record exists for this user
383
421
const records = getRecentRecordsByUser(did, 1);
384
-
if(records.length !== 0) return;
422
+
if (records.length !== 0) return;
385
423
386
424
const items = await getAllRated({ did });
387
425
const profile = await getProfile({ did });
388
-
389
-
for(const item of items) {
390
-
if(item.value.item.ref !== 'tmdb:s' && item.value.item.ref !== 'tmdb:m') {
426
+
427
+
for (const item of items) {
428
+
if (item.value.item.ref !== "tmdb:s" && item.value.item.ref !== "tmdb:m") {
391
429
continue;
392
430
}
393
431
394
-
const metadata = await getFormattedDetails(item.value.item.value, item.value.item.ref);
432
+
const metadata = await getFormattedDetails(
433
+
item.value.item.value,
434
+
item.value.item.ref
435
+
);
395
436
396
437
createRecord({
397
438
uri: item.uri,
···
402
443
displayName: profile.displayName,
403
444
avatar: profile.avatar,
404
445
},
405
-
indexedAt: item.value.note?.createdAt ?? item.value.rating?.createdAt ?? new Date().toISOString(),
406
-
createdAt: item.value.note?.createdAt ?? item.value.rating?.createdAt ?? new Date().toISOString(),
407
-
updatedAt: item.value.note?.updatedAt ?? item.value.rating?.createdAt ?? new Date().toISOString(),
446
+
indexedAt:
447
+
item.value.note?.createdAt ??
448
+
item.value.rating?.createdAt ??
449
+
new Date().toISOString(),
450
+
createdAt:
451
+
item.value.note?.createdAt ??
452
+
item.value.rating?.createdAt ??
453
+
new Date().toISOString(),
454
+
updatedAt:
455
+
item.value.note?.updatedAt ??
456
+
item.value.rating?.createdAt ??
457
+
new Date().toISOString(),
408
458
record: {
409
459
$type: item.value.$type,
410
460
item: item.value.item,
411
461
rating: item.value.rating,
412
462
note: item.value.note,
413
463
metadata,
414
-
}
464
+
},
415
465
});
416
466
}
417
-
}
418
-
419
-
export async function saveLikeToDatabase(json: LikeRecord) {
420
-
// first check if the like already exists
421
-
const existingLike = db.query('SELECT * FROM likes WHERE author_did = ? AND subject_uri = ?').get(json.author_did, json.subject_uri);
422
-
if(existingLike) return;
423
-
424
-
// check if the record exists
425
-
const record = getRecord(json.subject_uri);
426
-
if(!record) return;
427
-
428
-
// save the like to the database
429
-
const sql = 'INSERT INTO likes (uri, author_did, subject_cid, subject_uri, createdAt) VALUES (?, ?, ?, ?, ?)';
430
-
db.query(sql).run(json.uri, json.author_did, json.subject_cid, json.subject_uri, json.createdAt);
431
-
432
-
// update the record with the new like count
433
-
const newLikes = record.record.likes ? record.record.likes + 1 : 1;
434
-
db.query('UPDATE records SET record_likes = ? WHERE uri = ?').run(newLikes, json.subject_uri);
435
467
}
436
468
437
469
export { db, MainRecord };
+58
src/db/likes.ts
+58
src/db/likes.ts
···
1
+
import { db, getRecord } from "../database";
2
+
3
+
type LikeRecord = {
4
+
uri: string;
5
+
author_did: string;
6
+
subject_cid: string;
7
+
subject_uri: string;
8
+
createdAt: string;
9
+
};
10
+
11
+
export async function saveLikeToDatabase(json: LikeRecord) {
12
+
// first check if the like already exists
13
+
const existingLike = db
14
+
.query("SELECT * FROM likes WHERE author_did = ? AND subject_uri = ?")
15
+
.get(json.author_did, json.subject_uri);
16
+
if (existingLike) return;
17
+
18
+
// check if the record exists
19
+
const record = getRecord(json.subject_uri);
20
+
if (!record) return;
21
+
22
+
// save the like to the database
23
+
const sql =
24
+
"INSERT INTO likes (uri, author_did, subject_cid, subject_uri, createdAt) VALUES (?, ?, ?, ?, ?)";
25
+
db.query(sql).run(
26
+
json.uri,
27
+
json.author_did,
28
+
json.subject_cid,
29
+
json.subject_uri,
30
+
json.createdAt
31
+
);
32
+
33
+
// update the record with the new like count
34
+
const newLikes = record.record.likes ? record.record.likes + 1 : 1;
35
+
db.query("UPDATE records SET record_likes = ? WHERE uri = ?").run(
36
+
newLikes,
37
+
json.subject_uri
38
+
);
39
+
}
40
+
41
+
export async function deleteLikeFromDatabase(json: LikeRecord) {
42
+
db.query("DELETE FROM likes WHERE uri = ?").run(json.uri);
43
+
}
44
+
45
+
export async function getLikesByUser(did: string) {
46
+
return db
47
+
.query(
48
+
"SELECT * FROM likes WHERE author_did = ? ORDER BY createdAt DESC LIMIT 100"
49
+
)
50
+
.all(did);
51
+
}
52
+
53
+
export async function getAllUsersWithLikes() {
54
+
return db
55
+
.query("SELECT DISTINCT author_did FROM likes")
56
+
.all()
57
+
.map((row: any) => row.author_did);
58
+
}
+161
-111
src/jetstream.ts
+161
-111
src/jetstream.ts
···
1
1
2
2
import WebSocket from 'ws';
3
-
import { backfillUserIfNecessary, createRecord, recreateTables, getLatestTimestamp, MainRecord, saveLikeToDatabase, setLatestTimestamp } from './database';
4
-
import { getFormattedDetails } from './tmdb';
5
-
import { getProfile } from './atp';
3
+
import {
4
+
backfillUserIfNecessary,
5
+
createRecord,
6
+
deleteRecord,
7
+
getLatestTimestamp,
8
+
MainRecord,
9
+
recreateTables,
10
+
setLatestTimestamp,
11
+
} from "./database";
12
+
import { getFormattedDetails } from "./tmdb";
13
+
import { getProfile } from "./atp";
14
+
import { saveLikeToDatabase } from "./db/likes";
6
15
7
16
function printTimestamp(timestamp: number) {
8
-
const time = new Date(timestamp / 1_000).toISOString();
9
-
console.log('Timestamp:', time);
17
+
const time = new Date(timestamp / 1_000).toISOString();
18
+
console.log("Timestamp:", time);
10
19
}
11
20
12
21
// Function to calculate microseconds
13
22
function secondsToMicroseconds(seconds: number) {
14
-
return Math.floor(seconds * 1_000_000);
23
+
return Math.floor(seconds * 1_000_000);
15
24
}
16
25
17
26
let currentTimestamp: number = 0;
···
19
28
20
29
// Function to start the WebSocket connection
21
30
function startWebSocket(cursor: number) {
22
-
const url = `wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=my.skylights.rel&wantedCollections=community.lexicon.interaction.like&cursor=${cursor}`;
31
+
const url = `wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=my.skylights.rel&wantedCollections=community.lexicon.interaction.like&cursor=${cursor}`;
23
32
24
-
const ws = new WebSocket(url);
33
+
const ws = new WebSocket(url);
25
34
26
-
ws.onopen = () => {
27
-
console.log('Connected to WebSocket');
28
-
};
35
+
ws.onopen = () => {
36
+
console.log("Connected to WebSocket");
37
+
};
29
38
30
-
ws.onmessage = (event) => {
31
-
const data = event.data as string;
32
-
const json = JSON.parse(data.toString());
39
+
ws.onmessage = (event) => {
40
+
const data = event.data as string;
41
+
const json = JSON.parse(data.toString());
33
42
34
-
if(json.kind === 'commit' && json.commit.collection === 'my.skylights.rel' && json.commit.operation === 'create') {
35
-
saveRatingToDatabase(json);
36
-
}
43
+
if (
44
+
json.kind === "commit" &&
45
+
json.commit.collection === "my.skylights.rel" &&
46
+
json.commit.operation === "create"
47
+
) {
48
+
saveRatingToDatabase(json);
49
+
}
37
50
38
-
if(json.kind === 'commit' && json.commit.collection === 'community.lexicon.interaction.like' && json.commit.operation === 'create') {
39
-
saveLike(json);
40
-
}
51
+
if (json.commit?.collection === "my.skylights.rel") {
52
+
console.log(json);
53
+
}
41
54
42
-
if(!lastPrintedTimestamp || lastPrintedTimestamp < json.time_us - secondsToMicroseconds(60)) {
43
-
lastPrintedTimestamp = json.time_us;
44
-
printTimestamp(lastPrintedTimestamp);
55
+
if (
56
+
json.kind === "commit" &&
57
+
json.commit.collection === "my.skylights.rel" &&
58
+
json.commit.operation === "delete"
59
+
) {
60
+
deleteRatingFromDatabase(json);
61
+
}
45
62
46
-
setLatestTimestamp(new Date(lastPrintedTimestamp / 1_000).toISOString());
47
-
}
48
-
};
63
+
if (
64
+
json.kind === "commit" &&
65
+
json.commit.collection === "community.lexicon.interaction.like" &&
66
+
json.commit.operation === "create"
67
+
) {
68
+
saveLike(json);
69
+
}
49
70
50
-
ws.onerror = (event: any) => {
51
-
console.error('WebSocket error:', event.message);
52
-
reconnectWebSocket();
53
-
};
71
+
if (
72
+
!lastPrintedTimestamp ||
73
+
lastPrintedTimestamp < json.time_us - secondsToMicroseconds(60)
74
+
) {
75
+
lastPrintedTimestamp = json.time_us;
76
+
printTimestamp(lastPrintedTimestamp);
54
77
55
-
ws.onclose = () => {
56
-
console.log('WebSocket connection closed');
57
-
reconnectWebSocket();
58
-
};
78
+
setLatestTimestamp(new Date(lastPrintedTimestamp / 1_000).toISOString());
79
+
}
80
+
};
81
+
82
+
ws.onerror = (event: any) => {
83
+
console.error("WebSocket error:", event.message);
84
+
reconnectWebSocket();
85
+
};
86
+
87
+
ws.onclose = () => {
88
+
console.log("WebSocket connection closed");
89
+
reconnectWebSocket();
90
+
};
59
91
}
60
92
61
93
async function saveRatingToDatabase(json: any) {
62
-
if(json.commit.record.item.ref !== 'tmdb:s' && json.commit.record.item.ref !== 'tmdb:m') {
63
-
return;
64
-
}
94
+
if (
95
+
json.commit.record.item.ref !== "tmdb:s" &&
96
+
json.commit.record.item.ref !== "tmdb:m"
97
+
) {
98
+
return;
99
+
}
65
100
66
-
setLatestTimestamp(new Date(json.time_us / 1_000).toISOString());
101
+
setLatestTimestamp(new Date(json.time_us / 1_000).toISOString());
67
102
68
-
await backfillUserIfNecessary(json.did);
103
+
await backfillUserIfNecessary(json.did);
69
104
70
-
let item = {
71
-
ref: json.commit.record.item.ref,
72
-
value: json.commit.record.item.value,
73
-
}
105
+
let item = {
106
+
ref: json.commit.record.item.ref,
107
+
value: json.commit.record.item.value,
108
+
};
74
109
75
-
const details = await getFormattedDetails(item.value, item.ref);
110
+
const details = await getFormattedDetails(item.value, item.ref);
76
111
77
-
const timestamp = new Date(json.time_us / 1_000).toISOString();
112
+
const timestamp = new Date(json.time_us / 1_000).toISOString();
78
113
79
-
const author = await getProfile({ did: json.did });
114
+
const author = await getProfile({ did: json.did });
80
115
81
-
const record: MainRecord = {
82
-
uri: `at://${json.did}/${json.commit.collection}/${json.commit.rkey}`,
83
-
cid: json.commit.cid,
116
+
const record: MainRecord = {
117
+
uri: `at://${json.did}/${json.commit.collection}/${json.commit.rkey}`,
118
+
cid: json.commit.cid,
84
119
85
-
author: {
86
-
did: json.did,
87
-
handle: author.handle,
88
-
displayName: author.displayName,
89
-
avatar: author.avatar,
90
-
},
120
+
author: {
121
+
did: json.did,
122
+
handle: author.handle,
123
+
displayName: author.displayName,
124
+
avatar: author.avatar,
125
+
},
91
126
92
-
indexedAt: timestamp,
93
-
createdAt: json.commit.record.rating.createdAt ?? json.commit.record.note.createdAt ?? timestamp,
94
-
updatedAt: json.commit.record.rating.createdAt ?? json.commit.record.note.createdAt ?? timestamp,
127
+
indexedAt: timestamp,
128
+
createdAt:
129
+
json.commit.record.rating.createdAt ??
130
+
json.commit.record.note.createdAt ??
131
+
timestamp,
132
+
updatedAt:
133
+
json.commit.record.rating.createdAt ??
134
+
json.commit.record.note.createdAt ??
135
+
timestamp,
95
136
96
-
record: {
97
-
$type: json.commit.collection,
98
-
metadata: details,
99
-
item,
100
-
rating: {
101
-
value: json.commit.record.rating.value,
102
-
createdAt: json.commit.record.rating.createdAt,
103
-
},
104
-
}
105
-
}
137
+
record: {
138
+
$type: json.commit.collection,
139
+
metadata: details,
140
+
item,
141
+
rating: {
142
+
value: json.commit.record.rating.value,
143
+
createdAt: json.commit.record.rating.createdAt,
144
+
},
145
+
},
146
+
};
106
147
107
-
if(json.commit.record.note?.value) {
108
-
record.record.note = {
109
-
value: json.commit.record.note.value,
110
-
createdAt: json.commit.record.note.createdAt,
111
-
updatedAt: json.commit.record.note.createdAt,
112
-
}
113
-
}
114
-
try {
115
-
createRecord(record);
116
-
} catch(e: any) {
117
-
if(e.code === "SQLITE_CONSTRAINT_PRIMARYKEY") {
118
-
console.log("not saving record, already exists", record.uri);
119
-
} else {
120
-
console.error("FAILED TO SAVE RECORD", e.code);
121
-
}
122
-
}
148
+
if (json.commit.record.note?.value) {
149
+
record.record.note = {
150
+
value: json.commit.record.note.value,
151
+
createdAt: json.commit.record.note.createdAt,
152
+
updatedAt: json.commit.record.note.createdAt,
153
+
};
154
+
}
155
+
try {
156
+
createRecord(record);
157
+
} catch (e: any) {
158
+
if (e.code === "SQLITE_CONSTRAINT_PRIMARYKEY") {
159
+
console.log("not saving record, already exists", record.uri);
160
+
} else {
161
+
console.error("FAILED TO SAVE RECORD", e.code);
162
+
}
163
+
}
123
164
}
124
165
125
166
async function saveLike(json: any) {
126
-
// get uri
127
-
const uri = json.commit.record.subject.uri;
128
-
// split into did, collection, rkey
129
-
const [did, collection, rkey] = uri.replace('at://', '').split('/');
167
+
// get uri
168
+
const uri = json.commit.record.subject.uri;
169
+
// split into did, collection, rkey
170
+
const [did, collection] = uri.replace("at://", "").split("/");
130
171
131
-
if(collection !== 'my.skylights.rel') return;
172
+
if (collection !== "my.skylights.rel") return;
132
173
133
-
// make sure we have the post that is being liked
134
-
await backfillUserIfNecessary(did);
174
+
// make sure we have the post that is being liked
175
+
await backfillUserIfNecessary(did);
135
176
136
-
saveLikeToDatabase({
137
-
author_did: json.did,
138
-
subject_cid: json.commit.record.subject.cid,
139
-
subject_uri: json.commit.record.subject.uri,
140
-
createdAt: json.commit.record.createdAt,
141
-
});
177
+
saveLikeToDatabase({
178
+
uri: `at://${json.did}/${json.commit.collection}/${json.commit.rkey}`,
179
+
author_did: json.did,
180
+
subject_cid: json.commit.record.subject.cid,
181
+
subject_uri: json.commit.record.subject.uri,
182
+
createdAt: json.commit.record.createdAt,
183
+
});
184
+
}
185
+
186
+
async function deleteRatingFromDatabase(json: any) {
187
+
const uri = `at://${json.did}/${json.commit.collection}/${json.commit.rkey}`;
188
+
console.log(uri);
189
+
deleteRecord(uri);
142
190
}
143
191
144
192
// Function to reconnect WebSocket with an optional delay
145
193
function reconnectWebSocket(delay = 5000) {
146
-
console.log(`Reconnecting WebSocket in ${delay / 1000} seconds...`);
147
-
printTimestamp(currentTimestamp);
194
+
console.log(`Reconnecting WebSocket in ${delay / 1000} seconds...`);
195
+
printTimestamp(currentTimestamp);
148
196
149
-
setTimeout(() => startWebSocket(currentTimestamp), delay);
197
+
setTimeout(() => startWebSocket(currentTimestamp), delay);
150
198
}
151
-
152
199
153
200
export async function startJetstream() {
154
-
const nowMS = Date.now();
155
-
const oneMinuteAgoMS =
156
-
nowMS - 1 * 60 * 1000;
201
+
const nowMS = Date.now();
202
+
const oneMinuteAgoMS = nowMS - 1 * 60 * 1000;
157
203
158
-
let lastTimestamp = await getLatestTimestamp();
159
-
console.log('Last timestamp:', lastTimestamp);
204
+
// recreateTables();
205
+
206
+
let lastTimestamp = await getLatestTimestamp();
207
+
console.log("Last timestamp:", lastTimestamp);
160
208
161
-
const currentStartTime = lastTimestamp ? new Date(lastTimestamp) : new Date(oneMinuteAgoMS);
209
+
const currentStartTime = lastTimestamp
210
+
? new Date(lastTimestamp)
211
+
: new Date(oneMinuteAgoMS);
162
212
163
-
currentTimestamp = currentStartTime.getTime() * 1000;
164
-
console.log('Using timestamp', currentTimestamp);
165
-
printTimestamp(currentTimestamp);
213
+
currentTimestamp = currentStartTime.getTime() * 1000;
214
+
console.log("Using timestamp", currentTimestamp);
215
+
printTimestamp(currentTimestamp);
166
216
167
-
startWebSocket(currentTimestamp);
217
+
startWebSocket(currentTimestamp);
168
218
}