diff --git a/packages/db/connect.ts b/packages/db/connect.ts new file mode 100644 index 0000000..990b09e --- /dev/null +++ b/packages/db/connect.ts @@ -0,0 +1,21 @@ +import { drizzle } from "drizzle-orm/libsql"; +import * as schema from "@teal/db/schema"; +import process from "node:process"; +import path from "node:path"; + +console.log( + "Loading SQLite file at", + path.join(process.cwd(), "../../db.sqlite"), +); + +export const db = drizzle({ + connection: + // default is in project root / db.sqlite + process.env.DATABASE_URL ?? + "file:" + path.join(process.cwd(), "../../db.sqlite"), + // doesn't seem to work? + //casing: "snake_case", + schema: schema, +}); + +export type Database = typeof db; diff --git a/packages/jetstring/package.json b/packages/jetstring/package.json new file mode 100644 index 0000000..81e976a --- /dev/null +++ b/packages/jetstring/package.json @@ -0,0 +1,16 @@ +{ + "name": "@teal/jetstring", + "type": "module", + "scripts": { + "dev": "tsx watch src/index.ts | pino-pretty" + }, + "dependencies": { + "@skyware/jetstream": "^0.2.0", + "@teal/db": "*" + }, + "devDependencies": { + "tsup": "^8.3.5", + "tsx": "^4.19.2", + "typescript": "^5.6.3" + } +} diff --git a/packages/jetstring/src/index.ts b/packages/jetstring/src/index.ts new file mode 100644 index 0000000..abecc4d --- /dev/null +++ b/packages/jetstring/src/index.ts @@ -0,0 +1,129 @@ +import type { Database } from "@teal/db/connect"; +import { db } from "@teal/db/connect"; +import { status } from "@teal/db/schema"; +import { CommitCreateEvent, Jetstream } from "@skyware/jetstream"; +import { server } from "@teal/lexicons/generated/server/types"; +import ws from "ws"; + +class Handler { + private static instance: Handler; + private constructor() {} + public static getInstance(): Handler { + if (!Handler.instance) { + Handler.instance = new Handler(); + } + return Handler.instance; + } + + handle(msg_type: string, msg: any) { + // Handle message logic here + console.log("Handling" + msg_type + "message:", msg); + if (msg_type === "xyz.statusphere.status") { + // serialize message as xyz.statusphere.status + const st = db.insert(status).values({ + status: msg.status, + uri: msg.uri, + authorDid: msg.authorDid, + }); + } + } +} + +class Streamer { + private static instance: Streamer; + private jetstream: Jetstream; + private handler: Handler; + + private wantedCollections: string[]; + + private constructor(wantedCollections: string[]) { + this.handler = Handler.getInstance(); + console.log("Creating new jetstream with collections", wantedCollections); + this.jetstream = new Jetstream({ + wantedCollections, + }); + this.wantedCollections = wantedCollections; + } + + public static getInstance(wantedCollections?: string[]): Streamer { + if (!Streamer.instance && wantedCollections) { + Streamer.instance = new Streamer(wantedCollections); + } else if (!Streamer.instance) { + throw Error( + "Wanted collections are required if instance does not exist!", + ); + } + return Streamer.instance; + } + + async setOnCreates() { + for (const collection of this.wantedCollections) { + await this.setOnCreate(collection); + } + } + + async setOnCreate(collection: string) { + try { + this.jetstream.onCreate(collection, (event) => { + console.log("Received message:", event.commit.record); + this.handleCreate(collection, event); + }); + } catch (error) { + console.error("Error setting onCreate:", error); + } + console.log("Started onCreate stream for", collection); + } + + async handleCreate( + collection: string, + event: CommitCreateEvent, + ) { + this.handler.handle(collection, event); + } + + // Add method to start the streamer + async start() { + try { + await this.setOnCreates(); + this.jetstream.start(); + console.log("Streamer started successfully"); + } catch (error) { + console.error("Error starting streamer:", error); + } + } +} + +// Main function to run the application +async function main() { + try { + const streamer = Streamer.getInstance(["xyz.statusphere.status"]); + await streamer.start(); + + // Keep the process running + process.on("SIGINT", () => { + console.log("Received SIGINT. Graceful shutdown..."); + process.exit(0); + }); + + process.on("SIGTERM", () => { + console.log("Received SIGTERM. Graceful shutdown..."); + process.exit(0); + }); + + // Prevent the Node.js process from exiting + setInterval(() => { + // This empty interval keeps the process running + }, 1000); + + console.log("Application is running. Press Ctrl+C to exit."); + } catch (error) { + console.error("Error in main:", error); + process.exit(1); + } +} + +// Run the application +main().catch((error) => { + console.error("Unhandled error:", error); + process.exit(1); +}); diff --git a/packages/jetstring/tsconfig.json b/packages/jetstring/tsconfig.json new file mode 100644 index 0000000..fe72390 --- /dev/null +++ b/packages/jetstring/tsconfig.json @@ -0,0 +1,3 @@ +{ + "extends": "@teal/tsconfig/base.json" +} diff --git a/packages/matcher/package.json b/packages/matcher/package.json new file mode 100644 index 0000000..a688b24 --- /dev/null +++ b/packages/matcher/package.json @@ -0,0 +1,9 @@ +{ + "name": "matcher", + "version": "1.0.0", + "description": "", + "main": "index.ts", + "scripts": { + "test": "echo \"Error: no test specified\" && exit 1" + } +}