flo-bit.dev
[READ-ONLY] Mirror of https://github.com/flo-bit/skywatched-backend. backend/jetstream consumer for skywatched.app
skywatched.app
6.0 kB
226 lines
1import WebSocket from "ws";
2import {
3 backfillUserIfNecessary,
4 createRecord,
5 deleteRecord,
6 getLatestTimestamp,
7 MainRecord,
8 recreateTables,
9 setLatestTimestamp,
10} from "./database";
11import { getFormattedDetails } from "./tmdb";
12import { getProfile } from "./atp";
13import { saveLikeToDatabase } from "./db/likes";
14
15function printTimestamp(timestamp: number) {
16 const time = new Date(timestamp / 1_000).toISOString();
17 console.log("Timestamp:", time);
18}
19
20// Function to calculate microseconds
21function secondsToMicroseconds(seconds: number) {
22 return Math.floor(seconds * 1_000_000);
23}
24
25let currentTimestamp: number = 0;
26let lastPrintedTimestamp: number = 0;
27
28// Function to start the WebSocket connection
29function startWebSocket(cursor: number) {
30 const url = `wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=my.skylights.rel&wantedCollections=community.lexicon.interaction.like&cursor=${cursor}`;
31
32 const ws = new WebSocket(url);
33
34 ws.onopen = () => {
35 console.log("Connected to WebSocket");
36 };
37
38 ws.onmessage = (event) => {
39 const data = event.data as string;
40 const json = JSON.parse(data.toString());
41
42 if (
43 json.kind === "commit" &&
44 json.commit.collection === "my.skylights.rel" &&
45 json.commit.operation === "create"
46 ) {
47 saveRatingToDatabase(json);
48 }
49
50 if (json.commit?.collection === "my.skylights.rel") {
51 console.log(json);
52 }
53
54 if (
55 json.kind === "commit" &&
56 json.commit.collection === "my.skylights.rel" &&
57 json.commit.operation === "delete"
58 ) {
59 deleteRatingFromDatabase(json);
60 }
61
62 if (
63 json.kind === "commit" &&
64 json.commit.operation === "update" &&
65 json.commit.collection === "my.skylights.rel"
66 ) {
67 saveRatingToDatabase(json, true);
68 }
69
70 if (
71 json.kind === "commit" &&
72 json.commit.collection === "community.lexicon.interaction.like" &&
73 json.commit.operation === "create"
74 ) {
75 saveLike(json);
76 }
77
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 };
98}
99
100async 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 }
108
109 setLatestTimestamp(new Date(json.time_us / 1_000).toISOString());
110
111 await backfillUserIfNecessary(json.did);
112
113 let item = {
114 ref: json.commit.record.item.ref,
115 value: json.commit.record.item.value,
116 };
117
118 const details = await getFormattedDetails(item.value, item.ref);
119
120 const timestamp = new Date(json.time_us / 1_000).toISOString();
121
122 const author = await getProfile({ did: json.did });
123
124 const record: MainRecord = {
125 uri: `at://${json.did}/${json.commit.collection}/${json.commit.rkey}`,
126 cid: json.commit.cid,
127
128 author: {
129 did: json.did,
130 handle: author.handle,
131 displayName: author.displayName,
132 avatar: author.avatar,
133 },
134
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,
144
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 };
155
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 }
172}
173
174async function saveLike(json: any) {
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("/");
179
180 if (collection !== "my.skylights.rel") return;
181
182 // make sure we have the post that is being liked
183 await backfillUserIfNecessary(did);
184
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
194async function deleteRatingFromDatabase(json: any) {
195 const uri = `at://${json.did}/${json.commit.collection}/${json.commit.rkey}`;
196 console.log(uri);
197 deleteRecord(uri);
198}
199
200// Function to reconnect WebSocket with an optional delay
201function reconnectWebSocket(delay = 5000) {
202 console.log(`Reconnecting WebSocket in ${delay / 1000} seconds...`);
203 printTimestamp(currentTimestamp);
204
205 setTimeout(() => startWebSocket(currentTimestamp), delay);
206}
207
208export async function startJetstream() {
209 const nowMS = Date.now();
210 const oneMinuteAgoMS = nowMS - 1 * 60 * 1000;
211
212 // recreateTables();
213
214 let lastTimestamp = await getLatestTimestamp();
215 console.log("Last timestamp:", lastTimestamp);
216
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);
224
225 startWebSocket(currentTimestamp);
226}