teal.fm
Your music, beautifully tracked. All yours. (coming soon)
teal.fm
teal-fm
atproto
Add new jetstream package
+21
packages/db/connect.ts
+21
packages/db/connect.ts
···
1
+
import { drizzle } from "drizzle-orm/libsql";
2
+
import * as schema from "@teal/db/schema";
3
+
import process from "node:process";
4
+
import path from "node:path";
5
+
6
+
console.log(
7
+
"Loading SQLite file at",
8
+
path.join(process.cwd(), "../../db.sqlite"),
9
+
);
10
+
11
+
export const db = drizzle({
12
+
connection:
13
+
// default is in project root / db.sqlite
14
+
process.env.DATABASE_URL ??
15
+
"file:" + path.join(process.cwd(), "../../db.sqlite"),
16
+
// doesn't seem to work?
17
+
//casing: "snake_case",
18
+
schema: schema,
19
+
});
20
+
21
+
export type Database = typeof db;
+16
packages/jetstring/package.json
+16
packages/jetstring/package.json
···
1
+
{
2
+
"name": "@teal/jetstring",
3
+
"type": "module",
4
+
"scripts": {
5
+
"dev": "tsx watch src/index.ts | pino-pretty"
6
+
},
7
+
"dependencies": {
8
+
"@skyware/jetstream": "^0.2.0",
9
+
"@teal/db": "*"
10
+
},
11
+
"devDependencies": {
12
+
"tsup": "^8.3.5",
13
+
"tsx": "^4.19.2",
14
+
"typescript": "^5.6.3"
15
+
}
16
+
}
+129
packages/jetstring/src/index.ts
+129
packages/jetstring/src/index.ts
···
1
+
import type { Database } from "@teal/db/connect";
2
+
import { db } from "@teal/db/connect";
3
+
import { status } from "@teal/db/schema";
4
+
import { CommitCreateEvent, Jetstream } from "@skyware/jetstream";
5
+
import { server } from "@teal/lexicons/generated/server/types";
6
+
import ws from "ws";
7
+
8
+
class Handler {
9
+
private static instance: Handler;
10
+
private constructor() {}
11
+
public static getInstance(): Handler {
12
+
if (!Handler.instance) {
13
+
Handler.instance = new Handler();
14
+
}
15
+
return Handler.instance;
16
+
}
17
+
18
+
handle(msg_type: string, msg: any) {
19
+
// Handle message logic here
20
+
console.log("Handling" + msg_type + "message:", msg);
21
+
if (msg_type === "xyz.statusphere.status") {
22
+
// serialize message as xyz.statusphere.status
23
+
const st = db.insert(status).values({
24
+
status: msg.status,
25
+
uri: msg.uri,
26
+
authorDid: msg.authorDid,
27
+
});
28
+
}
29
+
}
30
+
}
31
+
32
+
class Streamer {
33
+
private static instance: Streamer;
34
+
private jetstream: Jetstream;
35
+
private handler: Handler;
36
+
37
+
private wantedCollections: string[];
38
+
39
+
private constructor(wantedCollections: string[]) {
40
+
this.handler = Handler.getInstance();
41
+
console.log("Creating new jetstream with collections", wantedCollections);
42
+
this.jetstream = new Jetstream({
43
+
wantedCollections,
44
+
});
45
+
this.wantedCollections = wantedCollections;
46
+
}
47
+
48
+
public static getInstance(wantedCollections?: string[]): Streamer {
49
+
if (!Streamer.instance && wantedCollections) {
50
+
Streamer.instance = new Streamer(wantedCollections);
51
+
} else if (!Streamer.instance) {
52
+
throw Error(
53
+
"Wanted collections are required if instance does not exist!",
54
+
);
55
+
}
56
+
return Streamer.instance;
57
+
}
58
+
59
+
async setOnCreates() {
60
+
for (const collection of this.wantedCollections) {
61
+
await this.setOnCreate(collection);
62
+
}
63
+
}
64
+
65
+
async setOnCreate(collection: string) {
66
+
try {
67
+
this.jetstream.onCreate(collection, (event) => {
68
+
console.log("Received message:", event.commit.record);
69
+
this.handleCreate(collection, event);
70
+
});
71
+
} catch (error) {
72
+
console.error("Error setting onCreate:", error);
73
+
}
74
+
console.log("Started onCreate stream for", collection);
75
+
}
76
+
77
+
async handleCreate(
78
+
collection: string,
79
+
event: CommitCreateEvent<string & {}>,
80
+
) {
81
+
this.handler.handle(collection, event);
82
+
}
83
+
84
+
// Add method to start the streamer
85
+
async start() {
86
+
try {
87
+
await this.setOnCreates();
88
+
this.jetstream.start();
89
+
console.log("Streamer started successfully");
90
+
} catch (error) {
91
+
console.error("Error starting streamer:", error);
92
+
}
93
+
}
94
+
}
95
+
96
+
// Main function to run the application
97
+
async function main() {
98
+
try {
99
+
const streamer = Streamer.getInstance(["xyz.statusphere.status"]);
100
+
await streamer.start();
101
+
102
+
// Keep the process running
103
+
process.on("SIGINT", () => {
104
+
console.log("Received SIGINT. Graceful shutdown...");
105
+
process.exit(0);
106
+
});
107
+
108
+
process.on("SIGTERM", () => {
109
+
console.log("Received SIGTERM. Graceful shutdown...");
110
+
process.exit(0);
111
+
});
112
+
113
+
// Prevent the Node.js process from exiting
114
+
setInterval(() => {
115
+
// This empty interval keeps the process running
116
+
}, 1000);
117
+
118
+
console.log("Application is running. Press Ctrl+C to exit.");
119
+
} catch (error) {
120
+
console.error("Error in main:", error);
121
+
process.exit(1);
122
+
}
123
+
}
124
+
125
+
// Run the application
126
+
main().catch((error) => {
127
+
console.error("Unhandled error:", error);
128
+
process.exit(1);
129
+
});