flo-bit.dev
[READ-ONLY] Mirror of https://github.com/flo-bit/skywatched-backend. backend/jetstream consumer for skywatched.app
skywatched.app
13 kB
472 lines
1import { existsSync } from 'node:fs';
2import { join } from 'node:path';
3import { Database, SQLQueryBindings } from 'bun:sqlite';
4import { getFormattedDetails } from './tmdb';
5import { getAllRated, getProfile } from './atp';
6
7const litefsDir = process.env.NODE_ENV === 'production' ? '/var/lib/litefs' : './litefs';
8const litefsPath = join(litefsDir, 'db.sqlite');
9
10if (!existsSync(litefsDir)) {
11 console.error('Unable to reach LiteFS directory at', litefsDir);
12 process.exit(1);
13}
14
15const db = new Database(litefsPath);
16
17type MainRecord = {
18 uri: string;
19 cid: string;
20
21 author: {
22 did: string;
23 handle: string;
24 displayName?: string;
25 avatar?: string;
26 };
27
28 indexedAt: string;
29 createdAt: string;
30 updatedAt: string;
31
32 record: {
33 $type: string;
34 item: {
35 ref: string;
36 value: string;
37 };
38 note?: {
39 value: string;
40 createdAt: string;
41 updatedAt: string;
42 };
43 rating?: {
44 value: number;
45 createdAt: string;
46 };
47 metadata?: {
48 title: string;
49 poster_path: string;
50 backdrop_path: string;
51 tagline: string;
52 overview: string;
53 genres: string[];
54 release_date?: string;
55 };
56 crosspost?: {
57 uri: string;
58 likes?: number;
59 reposts?: number;
60 replies?: number;
61 };
62
63 likes?: number;
64 }
65};
66
67type Record = {
68 uri: string;
69 cid: string;
70
71 author_did: string;
72 author_handle: string;
73 author_displayName: string;
74 author_avatar: string;
75
76 indexedAt: string;
77 createdAt: string;
78 updatedAt: string;
79
80 record_type: string;
81 record_item_ref: string;
82 record_item_value: string;
83
84 record_note_value?: string;
85 record_note_createdAt?: string;
86 record_note_updatedAt?: string;
87
88 record_rating_value?: number;
89 record_rating_createdAt?: string;
90
91 record_metadata_title?: string;
92 record_metadata_poster_path?: string;
93 record_metadata_backdrop_path?: string;
94 record_metadata_tagline?: string;
95 record_metadata_overview?: string;
96 record_metadata_genres?: string;
97 record_metadata_release_date?: string;
98
99 record_crosspost_uri?: string;
100 record_crosspost_likes?: number;
101 record_crosspost_reposts?: number;
102 record_crosspost_replies?: number;
103
104 record_likes?: number;
105};
106
107
108function createTables() {
109 db.run(`
110 CREATE TABLE IF NOT EXISTS records (
111 uri TEXT PRIMARY KEY,
112 cid TEXT NOT NULL,
113
114 author_did TEXT NOT NULL,
115 author_handle TEXT NOT NULL,
116 author_displayName TEXT,
117 author_avatar TEXT,
118
119 indexedAt TEXT NOT NULL,
120 createdAt TEXT NOT NULL,
121 updatedAt TEXT NOT NULL,
122
123 record_type TEXT NOT NULL,
124 record_item_ref TEXT NOT NULL,
125 record_item_value TEXT NOT NULL,
126
127 record_note_value TEXT,
128 record_note_createdAt TEXT,
129 record_note_updatedAt TEXT,
130
131 record_rating_value INTEGER,
132 record_rating_createdAt TEXT,
133
134 record_metadata_title TEXT,
135 record_metadata_poster_path TEXT,
136 record_metadata_backdrop_path TEXT,
137 record_metadata_tagline TEXT,
138 record_metadata_overview TEXT,
139 record_metadata_genres TEXT,
140 record_metadata_release_date TEXT,
141
142 record_crosspost_uri TEXT,
143 record_crosspost_likes INTEGER,
144 record_crosspost_reposts INTEGER,
145 record_crosspost_replies INTEGER,
146
147 record_likes INTEGER
148 );
149
150 CREATE INDEX IF NOT EXISTS idx_uri ON records (uri);
151 CREATE INDEX IF NOT EXISTS idx_author_did ON records (author_did);
152 CREATE INDEX IF NOT EXISTS idx_indexedAt ON records (indexedAt);
153 CREATE INDEX IF NOT EXISTS idx_createdAt ON records (createdAt);
154 CREATE INDEX IF NOT EXISTS idx_updatedAt ON records (updatedAt);
155 CREATE INDEX IF NOT EXISTS idx_item_ref_value ON records (record_item_ref, record_item_value);
156 CREATE INDEX IF NOT EXISTS idx_crosspost_uri ON records (record_crosspost_uri);
157 CREATE INDEX IF NOT EXISTS idx_likes ON records (record_likes);
158 `);
159
160 db.run(`
161 CREATE TABLE IF NOT EXISTS latest_indexedAt (
162 indexedAt TEXT PRIMARY KEY
163 );
164
165 CREATE INDEX IF NOT EXISTS idx_indexedAt ON latest_indexedAt (indexedAt);
166 `);
167
168 db.run(`
169 CREATE TABLE IF NOT EXISTS likes (
170 uri TEXT PRIMARY KEY,
171 author_did TEXT NOT NULL,
172 subject_cid TEXT NOT NULL,
173 subject_uri TEXT NOT NULL,
174 createdAt TEXT NOT NULL
175 );
176
177 CREATE INDEX IF NOT EXISTS idx_uri ON likes (uri);
178 CREATE INDEX IF NOT EXISTS idx_author_did ON likes (author_did);
179 CREATE INDEX IF NOT EXISTS idx_subject_uri ON likes (subject_uri);
180 CREATE INDEX IF NOT EXISTS idx_createdAt ON likes (createdAt);
181 `);
182}
183
184// Function to get the latest timestamp (either creation or update)
185export function getLatestTimestamp(): string | null {
186 const row = db
187 .query("SELECT MAX(indexedAt) AS latest FROM latest_indexedAt;")
188 .get() as { latest: string } | undefined;
189 return row?.latest ?? null;
190}
191
192export function setLatestTimestamp(indexedAt: string): void {
193 db.query("DELETE FROM latest_indexedAt").run();
194 db.query(
195 "INSERT OR REPLACE INTO latest_indexedAt (indexedAt) VALUES (?)"
196 ).run(indexedAt);
197}
198
199export function getRecord(uri: string): MainRecord | null {
200 const row = db.query("SELECT * FROM records WHERE uri = ?").get(uri) as
201 | Record
202 | undefined;
203 return row ? transformRecord(row) : null;
204}
205
206export function getAuthorDids(): string[] {
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);
211}
212
213export function createRecord(
214 record: MainRecord,
215 override: boolean = false
216): void {
217 const sql = `
218 INSERT ${override ? "OR REPLACE" : ""} INTO records (
219 uri, cid,
220
221 author_did, author_handle, author_displayName, author_avatar,
222
223 indexedAt, createdAt, updatedAt,
224
225 record_type,
226
227 record_item_ref, record_item_value,
228
229 record_note_value, record_note_createdAt, record_note_updatedAt,
230
231 record_rating_value, record_rating_createdAt,
232
233 record_metadata_title, record_metadata_poster_path, record_metadata_backdrop_path, record_metadata_tagline, record_metadata_overview, record_metadata_genres, record_metadata_release_date,
234
235 record_crosspost_uri, record_crosspost_likes, record_crosspost_reposts, record_crosspost_replies
236 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
237 `;
238 const params = [
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,
256
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,
264
265 record.record.crosspost?.uri ?? null,
266 record.record.crosspost?.likes ?? null,
267 record.record.crosspost?.reposts ?? null,
268 record.record.crosspost?.replies ?? null,
269 ];
270 db.query(sql).run(...(params as SQLQueryBindings[]));
271}
272
273export function transformRecord(record: Record): MainRecord {
274 return {
275 uri: record.uri,
276 cid: record.cid,
277 indexedAt: record.indexedAt,
278 createdAt: record.createdAt,
279 updatedAt: record.updatedAt,
280 author: {
281 did: record.author_did,
282 handle: record.author_handle,
283 displayName: record.author_displayName,
284 avatar: record.author_avatar,
285 },
286 record: {
287 $type: record.record_type,
288 item: {
289 ref: record.record_item_ref,
290 value: record.record_item_value,
291 },
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,
306 metadata: {
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 ?? "",
314 },
315 likes: record.record_likes ?? 0,
316 },
317 };
318}
319
320// Function to get the most recent created records with pagination
321export function getMostRecentRecords(
322 limit: number = 100,
323 cursor: string | null = null
324): MainRecord[] {
325 let sql = "SELECT * FROM records WHERE 1=1";
326 const params: (string | number)[] = [];
327
328 if (cursor) {
329 sql += " AND createdAt < ?";
330 params.push(cursor);
331 }
332
333 sql += " ORDER BY createdAt DESC LIMIT ?";
334 params.push(limit);
335
336 const rows = db.query(sql).all(...params) as Record[];
337 return rows.map((row) => transformRecord(row));
338}
339
340// // Function to update a record
341// export function updateRecord(uri: string, updates: Partial<MainRecord>): void {
342// const fields = Object.keys(updates).map(key => `${key} = ?`).join(', ');
343// const values = Object.values(updates);
344// const sql = `UPDATE records SET ${fields} WHERE uri = ?`;
345// db.query(sql).run(...values as SQLQueryBindings[], uri);
346// }
347
348// Function to get the most recent records for a specific user
349export 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 = ?";
355 const params: (string | number)[] = [user_did];
356
357 if (cursor) {
358 sql += " AND createdAt < ?";
359 params.push(cursor);
360 }
361
362 sql += " ORDER BY createdAt DESC LIMIT ?";
363 params.push(limit);
364
365 const rows = db.query(sql).all(...params) as Record[];
366 return rows.map((row) => transformRecord(row));
367}
368
369// Function to get the most recent records for a specific item ref and value with pagination
370export 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 = ?";
378 const params: (string | number)[] = [item_ref, item_value];
379
380 if (cursor) {
381 sql += " AND createdAt < ?";
382 params.push(cursor);
383 }
384
385 sql += " ORDER BY createdAt DESC LIMIT ?";
386 params.push(limit);
387
388 const rows = db.query(sql).all(...params) as Record[];
389 return rows.map((row) => transformRecord(row));
390}
391
392// Function to delete a record
393// export function deleteRecord(uri: string): void {
394// const sql = 'DELETE FROM records WHERE uri = ?';
395// db.query(sql).run(uri);
396// }
397
398export function deleteAllByUser(user_did: string): void {
399 const sql = "DELETE FROM records WHERE author_did = ?";
400 db.query(sql).run(user_did);
401}
402
403export function deleteRecord(uri: string): void {
404 const sql = "DELETE FROM records WHERE uri = ?";
405 db.query(sql).run(uri);
406}
407
408export function deleteAllRecords(): void {
409 const sql = "DELETE FROM records";
410 db.query(sql).run();
411}
412
413export function recreateTables() {
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");
417
418 // create the tables again
419 createTables();
420}
421
422export async function backfillUserIfNecessary(did: string): Promise<void> {
423 // check if at least one record exists for this user
424 const records = getRecentRecordsByUser(did, 1);
425 if (records.length !== 0) return;
426
427 const items = await getAllRated({ did });
428 const profile = await getProfile({ did });
429
430 for (const item of items) {
431 if (item.value.item.ref !== "tmdb:s" && item.value.item.ref !== "tmdb:m") {
432 continue;
433 }
434
435 const metadata = await getFormattedDetails(
436 item.value.item.value,
437 item.value.item.ref
438 );
439
440 createRecord({
441 uri: item.uri,
442 cid: item.cid,
443 author: {
444 did: did,
445 handle: profile.handle,
446 displayName: profile.displayName,
447 avatar: profile.avatar,
448 },
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(),
461 record: {
462 $type: item.value.$type,
463 item: item.value.item,
464 rating: item.value.rating,
465 note: item.value.note,
466 metadata,
467 },
468 });
469 }
470}
471
472export { db, MainRecord };