add websocket functionality

src/main.ts

@@ -3,33 +3,35 @@ import {
   OllamaResponse,
   NewStatusBody,
   Notification,
+  WSEvent,
 } from "../types.js";
 import striptags from "striptags";
 import { PrismaClient } from "../generated/prisma/client.js";
+import { createWebsocket } from "./websocket.js";
 
 const prisma = new PrismaClient();
 
-const getNotifications = async () => {
-  try {
-    const request = await fetch(
-      `${process.env.PLEROMA_INSTANCE_URL}/api/v1/notifications?types[]=mention`,
-      {
-        method: "GET",
-        headers: {
-          Authorization: `Bearer ${process.env.INSTANCE_BEARER_TOKEN}`,
-        },
-      }
-    );
+// const getNotifications = async () => {
+//   try {
+//     const request = await fetch(
+//       `${process.env.PLEROMA_INSTANCE_URL}/api/v1/notifications?types[]=mention`,
+//       {
+//         method: "GET",
+//         headers: {
+//           Authorization: `Bearer ${process.env.INSTANCE_BEARER_TOKEN}`,
+//         },
+//       }
+//     );
 
-    const notifications: Notification[] = await request.json();
+//     const notifications: Notification[] = await request.json();
 
-    return notifications;
-  } catch (error: any) {
-    throw new Error(error.message);
-  }
-};
+//     return notifications;
+//   } catch (error: any) {
+//     throw new Error(error.message);
+//   }
+// };
 
-const notifications = await getNotifications();
+// const notifications = await getNotifications();
 
 const storeUserData = async (notification: Notification): Promise<void> => {
   try {
@@ -171,13 +173,35 @@ const postReplyToStatus = async (
   }
 };
 
-if (notifications) {
-  await Promise.all(
-    notifications.map(async (notification) => {
-      const ollamaResponse = await generateOllamaRequest(notification);
-      if (ollamaResponse) {
-        postReplyToStatus(notification, ollamaResponse);
-      }
-    })
+const ws = createWebsocket();
+
+ws.on("upgrade", () => {
+  console.log(
+    `Websocket connection to ${process.env.PLEROMA_INSTANCE_DOMAIN} successful.`
   );
-}
+});
+
+ws.on("message", async (data) => {
+  const message: WSEvent = JSON.parse(data.toString("utf-8"));
+  if (message.event !== "notification") {
+    // only watch for notification events
+    return;
+  }
+  console.log("Websocket message received.");
+  const payload = JSON.parse(message.payload) as Notification;
+  const ollamaResponse = await generateOllamaRequest(payload);
+  if (ollamaResponse) {
+    await postReplyToStatus(payload, ollamaResponse);
+  }
+});
+
+// if (notifications) {
+//   await Promise.all(
+//     notifications.map(async (notification) => {
+//       const ollamaResponse = await generateOllamaRequest(notification);
+//       if (ollamaResponse) {
+//         postReplyToStatus(notification, ollamaResponse);
+//       }
+//     })
+//   );
+// }

src/websocket.ts

@@ -0,0 +1,22 @@
+import { WebSocket } from "ws";
+
+const scheme = process.env.PLEROMA_INSTANCE_URL?.startsWith("https")
+  ? "wss"
+  : "ws"; // this is so nigger rigged
+const host = process.env.PLEROMA_INSTANCE_DOMAIN;
+
+export const createWebsocket = (): WebSocket => {
+  try {
+    const ws = new WebSocket( // only connects to Soapbox frontends right now, but could pretty easily connect to Pleroma frontends with some tweaking
+      `${scheme}://${host}/api/v1/streaming?stream=user`,
+      [process.env.SOAPBOX_WS_PROTOCOL as string],
+      {
+        followRedirects: true,
+      }
+    );
+
+    return ws;
+  } catch (error: any) {
+    throw new Error(error);
+  }
+};