tracks lexicons and how many times they appeared on the jetstream
feat: push events to db if a lot of them accumulate (10k), or on exit
+40
-9
+4
-1
src/hooks.server.ts
+4
-1
src/hooks.server.ts
···
1
-
import { startTracking } from "$lib/jetstream.js";
1
+
import { startTracking, writeEvents } from "$lib/jetstream.js";
2
2
3
3
// Start tracking when the server starts
4
4
startTracking().catch(console.error);
5
+
process.on("SIGINT", writeEvents);
6
+
process.on("SIGTERM", writeEvents);
7
+
process.on("SIGQUIT", writeEvents);
+13
-6
src/lib/db.ts
+13
-6
src/lib/db.ts
···
81
81
this.insertNsidQuery.run(ALL_NSID);
82
82
}
83
83
84
-
addEvent = (nsid: string, timestamp: number, deleted: boolean) => {
85
-
const tx = this.db.transaction(() => {
84
+
writeEvents = (
85
+
events: { nsid: string; timestamp: number; deleted: boolean }[],
86
+
) => {
87
+
this.db.transaction(() => {
88
+
for (const event of events) {
89
+
this.insertEventQuery.run(event.nsid, event.timestamp, event.deleted);
90
+
}
91
+
})();
92
+
};
93
+
94
+
recordEvent = (nsid: string, timestamp: number, deleted: boolean) => {
95
+
this.db.transaction(() => {
86
96
this.insertNsidQuery.run(nsid);
87
-
this.insertEventQuery.run(nsid, timestamp, deleted);
88
97
this.updateCountQuery.run({
89
98
$nsid: nsid,
90
99
$deleted: deleted,
···
95
104
$deleted: deleted,
96
105
$timestamp: timestamp,
97
106
});
98
-
});
99
-
100
-
tx();
107
+
})();
101
108
};
102
109
103
110
getNsidCounts = (): EventRecord[] => {
+23
-2
src/lib/jetstream.ts
+23
-2
src/lib/jetstream.ts
···
2
2
import { eventTracker } from "./db.js";
3
3
4
4
let subscription: JetstreamSubscription | null = null;
5
+
let eventsToCommit: {
6
+
nsid: string;
7
+
timestamp: number;
8
+
deleted: boolean;
9
+
}[] = [];
10
+
11
+
export const writeEvents = () => {
12
+
eventTracker.writeEvents(eventsToCommit);
13
+
eventsToCommit = [];
14
+
};
5
15
6
16
export const startTracking = async () => {
7
17
subscription = new JetstreamSubscription({
8
18
url: "wss://jetstream2.us-east.bsky.network",
9
-
// Don't filter by collections - we want to track all of them
19
+
validateEvents: false, // trust the jetstream :3
10
20
});
11
21
12
22
for await (const event of subscription) {
···
16
26
17
27
const { operation, collection } = event.commit;
18
28
19
-
eventTracker.addEvent(collection, event.time_us, operation === "delete");
29
+
eventTracker.recordEvent(collection, event.time_us, operation === "delete");
30
+
eventsToCommit.push({
31
+
nsid: collection,
32
+
timestamp: event.time_us,
33
+
deleted: operation === "delete",
34
+
});
35
+
36
+
if (eventsToCommit.length > 10000) {
37
+
writeEvents();
38
+
}
20
39
}
40
+
41
+
writeEvents();
21
42
};