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