[READ-ONLY] Mirror of https://github.com/flo-bit/skywatched-backend. backend/jetstream consumer for skywatched.app
skywatched.app
add more apis
+126
-85
+3
package.json
+3
package.json
+18
-4
src/api.ts
+18
-4
src/api.ts
···
1
1
// server.ts
2
-
import { backfillUserIfNecessary, getLatestTimestamp, getMostRecentRecords, getRecentRecordsByItemRef, getRecentRecordsByUser } from './database';
2
+
import { backfillUserIfNecessary, getAuthorDids, getLatestTimestamp, getMostRecentRecords, getRecentRecordsByItemRef, getRecentRecordsByUser, getRecord } from './database';
3
3
4
4
// Define your API routes and handlers
5
5
export const handler = async (req: Request): Promise<Response> => {
···
15
15
}
16
16
17
17
if (path === "/api/most-recent-records") {
18
-
const limit = parseInt(url.searchParams.get("limit") || "50");
18
+
const limit = parseInt(url.searchParams.get("limit") ?? "50");
19
19
const cursor = url.searchParams.get("cursor");
20
20
const records = getMostRecentRecords(limit, cursor);
21
21
return new Response(JSON.stringify(records), { status: 200, headers: { "Content-Type": "application/json" } });
···
24
24
if (path === "/api/recent-records-by-user") {
25
25
const user_did = url.searchParams.get("did");
26
26
27
-
const limit = parseInt(url.searchParams.get("limit") || "50");
27
+
const limit = parseInt(url.searchParams.get("limit") ?? "50");
28
28
const cursor = url.searchParams.get("cursor");
29
29
if (!user_did) {
30
30
return new Response(JSON.stringify({ error: "did is required" }), { status: 400, headers: { "Content-Type": "application/json" } });
···
42
42
if (path === "/api/recent-records-by-item") {
43
43
const item_ref = url.searchParams.get("ref");
44
44
const item_value = url.searchParams.get("value");
45
-
const limit = parseInt(url.searchParams.get("limit") || "50");
45
+
const limit = parseInt(url.searchParams.get("limit") ?? "50");
46
46
const cursor = url.searchParams.get("cursor");
47
47
if (!item_ref || !item_value) {
48
48
return new Response(JSON.stringify({ error: "item_ref and item_value are required" }), { status: 400, headers: { "Content-Type": "application/json" } });
49
49
}
50
50
const records = getRecentRecordsByItemRef(item_ref, item_value, limit, cursor);
51
51
return new Response(JSON.stringify(records), { status: 200, headers: { "Content-Type": "application/json" } });
52
+
}
53
+
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
+
}
62
+
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" } });
52
66
}
53
67
}
54
68
+101
-77
src/database.ts
+101
-77
src/database.ts
···
1
1
import { existsSync } from 'node:fs';
2
2
import { join } from 'node:path';
3
3
import { Database, SQLQueryBindings } from 'bun:sqlite';
4
-
import { AtpBaseClient } from '@atproto/api';
5
-
import { getDetails, getFormattedDetails } from './tmdb';
4
+
import { getFormattedDetails } from './tmdb';
6
5
import { getAllRated, getProfile } from './atp';
7
6
8
7
const litefsDir = process.env.NODE_ENV === 'production' ? '/var/lib/litefs' : './litefs';
···
56
55
};
57
56
crosspost?: {
58
57
uri: string;
58
+
likes?: number;
59
+
reposts?: number;
60
+
replies?: number;
59
61
};
60
62
}
61
63
};
···
93
95
record_metadata_release_date?: string;
94
96
95
97
record_crosspost_uri?: string;
98
+
record_crosspost_likes?: number;
99
+
record_crosspost_reposts?: number;
100
+
record_crosspost_replies?: number;
96
101
};
97
102
98
103
function createTables() {
99
-
db.run(`
100
-
CREATE TABLE IF NOT EXISTS records (
101
-
uri TEXT PRIMARY KEY,
102
-
cid TEXT NOT NULL,
104
+
db.run(`
105
+
CREATE TABLE IF NOT EXISTS records (
106
+
uri TEXT PRIMARY KEY,
107
+
cid TEXT NOT NULL,
103
108
104
-
author_did TEXT NOT NULL,
105
-
author_handle TEXT NOT NULL,
106
-
author_displayName TEXT NOT NULL,
107
-
author_avatar TEXT NOT NULL,
109
+
author_did TEXT NOT NULL,
110
+
author_handle TEXT NOT NULL,
111
+
author_displayName TEXT,
112
+
author_avatar TEXT,
108
113
109
-
indexedAt TEXT NOT NULL,
110
-
createdAt TEXT NOT NULL,
111
-
updatedAt TEXT NOT NULL,
114
+
indexedAt TEXT NOT NULL,
115
+
createdAt TEXT NOT NULL,
116
+
updatedAt TEXT NOT NULL,
112
117
113
-
record_type TEXT NOT NULL,
114
-
record_item_ref TEXT NOT NULL,
115
-
record_item_value TEXT NOT NULL,
118
+
record_type TEXT NOT NULL,
119
+
record_item_ref TEXT NOT NULL,
120
+
record_item_value TEXT NOT NULL,
116
121
117
-
record_note_value TEXT,
118
-
record_note_createdAt TEXT,
119
-
record_note_updatedAt TEXT,
122
+
record_note_value TEXT,
123
+
record_note_createdAt TEXT,
124
+
record_note_updatedAt TEXT,
120
125
121
-
record_rating_value INTEGER,
122
-
record_rating_createdAt TEXT,
126
+
record_rating_value INTEGER,
127
+
record_rating_createdAt TEXT,
123
128
124
-
record_metadata_title TEXT,
125
-
record_metadata_poster_path TEXT,
126
-
record_metadata_backdrop_path TEXT,
127
-
record_metadata_tagline TEXT,
128
-
record_metadata_overview TEXT,
129
-
record_metadata_genres TEXT,
130
-
record_metadata_release_date TEXT,
129
+
record_metadata_title TEXT,
130
+
record_metadata_poster_path TEXT,
131
+
record_metadata_backdrop_path TEXT,
132
+
record_metadata_tagline TEXT,
133
+
record_metadata_overview TEXT,
134
+
record_metadata_genres TEXT,
135
+
record_metadata_release_date TEXT,
131
136
132
-
record_crosspost_uri TEXT
133
-
);
137
+
record_crosspost_uri TEXT,
138
+
record_crosspost_likes INTEGER,
139
+
record_crosspost_reposts INTEGER,
140
+
record_crosspost_replies INTEGER
141
+
);
134
142
135
-
CREATE INDEX IF NOT EXISTS idx_author_did ON records (author_did);
136
-
CREATE INDEX IF NOT EXISTS idx_indexedAt ON records (indexedAt);
137
-
CREATE INDEX IF NOT EXISTS idx_createdAt ON records (createdAt);
138
-
CREATE INDEX IF NOT EXISTS idx_updatedAt ON records (updatedAt);
139
-
CREATE INDEX IF NOT EXISTS idx_item_ref_value ON records (record_item_ref, record_item_value);
140
-
`);
143
+
CREATE INDEX IF NOT EXISTS idx_uri ON records (uri);
144
+
CREATE INDEX IF NOT EXISTS idx_author_did ON records (author_did);
145
+
CREATE INDEX IF NOT EXISTS idx_indexedAt ON records (indexedAt);
146
+
CREATE INDEX IF NOT EXISTS idx_createdAt ON records (createdAt);
147
+
CREATE INDEX IF NOT EXISTS idx_updatedAt ON records (updatedAt);
148
+
CREATE INDEX IF NOT EXISTS idx_item_ref_value ON records (record_item_ref, record_item_value);
149
+
CREATE INDEX IF NOT EXISTS idx_crosspost_uri ON records (record_crosspost_uri);
150
+
`);
141
151
142
-
db.run(`
143
-
CREATE TABLE IF NOT EXISTS latest_indexedAt (
144
-
indexedAt TEXT PRIMARY KEY
145
-
);
152
+
db.run(`
153
+
CREATE TABLE IF NOT EXISTS latest_indexedAt (
154
+
indexedAt TEXT PRIMARY KEY
155
+
);
146
156
147
-
CREATE INDEX IF NOT EXISTS idx_indexedAt ON latest_indexedAt (indexedAt);
148
-
`);
157
+
CREATE INDEX IF NOT EXISTS idx_indexedAt ON latest_indexedAt (indexedAt);
158
+
`);
149
159
}
150
160
151
161
// Function to get the latest timestamp (either creation or update)
152
162
export function getLatestTimestamp(): string | null {
153
163
const row = db.query('SELECT MAX(indexedAt) AS latest FROM latest_indexedAt;').get() as { latest: string } | undefined;
154
-
return row?.latest || null;
164
+
return row?.latest ?? null;
155
165
}
156
166
157
167
export function setLatestTimestamp(indexedAt: string): void {
168
+
db.query('DELETE FROM latest_indexedAt').run();
158
169
db.query('INSERT OR REPLACE INTO latest_indexedAt (indexedAt) VALUES (?)').run(indexedAt);
159
170
}
160
171
172
+
export function getRecord(uri: string): MainRecord | null {
173
+
const row = db.query('SELECT * FROM records WHERE uri = ?').get(uri) as Record | undefined;
174
+
return row ? transformRecord(row) : null;
175
+
}
176
+
177
+
export function getAuthorDids(): string[] {
178
+
const rows = db.query('SELECT DISTINCT author_did FROM records').all() as { author_did: string }[];
179
+
return rows.map(row => row.author_did);
180
+
}
181
+
161
182
export function createRecord(record: MainRecord): void {
162
183
const sql = `
163
184
INSERT INTO records (
···
177
198
178
199
record_metadata_title, record_metadata_poster_path, record_metadata_backdrop_path, record_metadata_tagline, record_metadata_overview, record_metadata_genres, record_metadata_release_date,
179
200
180
-
record_crosspost_uri
181
-
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
201
+
record_crosspost_uri, record_crosspost_likes, record_crosspost_reposts, record_crosspost_replies
202
+
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
182
203
`;
183
204
const params = [
184
205
record.uri, record.cid,
···
193
214
record.record.metadata?.tagline ?? null, record.record.metadata?.overview ?? null, record.record.metadata?.genres?.join(',') ?? null,
194
215
record.record.metadata?.release_date ?? null,
195
216
196
-
record.record.crosspost?.uri ?? null
217
+
record.record.crosspost?.uri ?? null,
218
+
record.record.crosspost?.likes ?? null,
219
+
record.record.crosspost?.reposts ?? null,
220
+
record.record.crosspost?.replies ?? null
197
221
];
198
222
db.query(sql).run(...params as SQLQueryBindings[]);
199
223
}
···
319
343
320
344
export async function backfillUserIfNecessary(did: string): Promise<void> {
321
345
// check if at least one record exists for this user
322
-
const records = await getRecentRecordsByUser(did, 1);
323
-
if(records.length === 0) {
324
-
const items = await getAllRated({ did });
325
-
const profile = await getProfile({ did });
326
-
327
-
for(const item of items) {
328
-
if(item.value.item.ref !== 'tmdb:s' && item.value.item.ref !== 'tmdb:m') {
329
-
continue;
330
-
}
331
-
332
-
const metadata = await getFormattedDetails(item.value.item.value, item.value.item.ref);
346
+
const records = getRecentRecordsByUser(did, 1);
347
+
if(records.length !== 0) return;
333
348
334
-
createRecord({
335
-
uri: item.uri,
336
-
cid: item.cid,
337
-
author: {
338
-
did: did,
339
-
handle: profile.handle,
340
-
displayName: profile.displayName,
341
-
avatar: profile.avatar,
342
-
},
343
-
indexedAt: item.value.note?.createdAt ?? item.value.rating?.createdAt ?? new Date().toISOString(),
344
-
createdAt: item.value.note?.createdAt ?? item.value.rating?.createdAt ?? new Date().toISOString(),
345
-
updatedAt: item.value.note?.updatedAt ?? item.value.rating?.createdAt ?? new Date().toISOString(),
346
-
record: {
347
-
$type: item.value.$type,
348
-
item: item.value.item,
349
-
rating: item.value.rating,
350
-
note: item.value.note,
351
-
metadata,
352
-
}
353
-
});
349
+
const items = await getAllRated({ did });
350
+
const profile = await getProfile({ did });
351
+
352
+
for(const item of items) {
353
+
if(item.value.item.ref !== 'tmdb:s' && item.value.item.ref !== 'tmdb:m') {
354
+
continue;
354
355
}
356
+
357
+
const metadata = await getFormattedDetails(item.value.item.value, item.value.item.ref);
358
+
359
+
createRecord({
360
+
uri: item.uri,
361
+
cid: item.cid,
362
+
author: {
363
+
did: did,
364
+
handle: profile.handle,
365
+
displayName: profile.displayName,
366
+
avatar: profile.avatar,
367
+
},
368
+
indexedAt: item.value.note?.createdAt ?? item.value.rating?.createdAt ?? new Date().toISOString(),
369
+
createdAt: item.value.note?.createdAt ?? item.value.rating?.createdAt ?? new Date().toISOString(),
370
+
updatedAt: item.value.note?.updatedAt ?? item.value.rating?.createdAt ?? new Date().toISOString(),
371
+
record: {
372
+
$type: item.value.$type,
373
+
item: item.value.item,
374
+
rating: item.value.rating,
375
+
note: item.value.note,
376
+
metadata,
377
+
}
378
+
});
355
379
}
356
380
}
357
381
+4
-4
src/jetstream.ts
+4
-4
src/jetstream.ts
···
1
1
2
2
import WebSocket from 'ws';
3
-
import { backfillUserIfNecessary, createRecord, getLatestTimestamp, MainRecord, setLatestTimestamp } from './database';
3
+
import { backfillUserIfNecessary, createRecord, dropAllTables, getLatestTimestamp, MainRecord, setLatestTimestamp } from './database';
4
4
import { getFormattedDetails } from './tmdb';
5
5
import { getProfile } from './atp';
6
6
···
129
129
130
130
export async function startJetstream() {
131
131
const nowMS = Date.now();
132
-
const twentyFourHoursAgoMS =
133
-
nowMS - 24 * 60 * 60 * 1000;
132
+
const oneMinuteAgoMS =
133
+
nowMS - 1 * 60 * 1000;
134
134
135
135
let lastTimestamp = await getLatestTimestamp();
136
136
console.log('Last timestamp:', lastTimestamp);
137
137
138
-
const currentStartTime = lastTimestamp ? new Date(lastTimestamp) : new Date(twentyFourHoursAgoMS);
138
+
const currentStartTime = lastTimestamp ? new Date(lastTimestamp) : new Date(oneMinuteAgoMS);
139
139
140
140
console.log('Using timestamp');
141
141
printTimestamp(currentStartTime.getTime() * 1000);