Refactoring 3

apps/bridge-worker/package.json

@@ -6,16 +6,16 @@
   "license": "AGPL-3.0-or-later",
   "scripts": {
     "build": "tsc -p tsconfig.json",
-    "dev": "NODE_OPTIONS=\"--loader ts-node/esm\" graphile-worker"
+    "dev": " NODE_OPTIONS=\"--loader ts-node/esm\" graphile-worker"
   },
   "dependencies": {
     "@hapi/wreck": "^18.1.0",
+    "bridge-common": "*",
     "fluent-ffmpeg": "^2.1.2",
     "graphile-worker": "^0.16.5",
     "html-to-text": "^9.0.5",
     "jest": "^29.7.0",
     "kysely": "^0.27.3",
-    "bridge-common": "*",
     "pg": "^8.11.5",
     "remeda": "^1.61.0",
     "twilio": "^5.0.4"
@@ -25,6 +25,7 @@
     "@babel/preset-env": "7.24.4",
     "@babel/preset-typescript": "7.24.1",
     "@types/fluent-ffmpeg": "^2.1.24",
+    "dotenv": "^16.4.5",
     "eslint": "^9.0.0",
     "prettier": "^3.2.5",
     "ts-config": "*",

apps/bridge-worker/tasks/common/notify-webhook.ts

@@ -1,57 +0,0 @@
-import Wreck from "@hapi/wreck";
-import * as R from "remeda";
-import { withDb, AppDatabase } from "../../lib/db";
-// import logger from "../logger";
-
-export interface WebhookOptions {
-  webhookId: string;
-  payload: any;
-}
-
-const notifyWebhooksTask = async (options: WebhookOptions): Promise<void> =>
-  withDb(async (db: AppDatabase) => {
-    const { webhookId, payload } = options;
-
-    const webhook = await db.webhooks.findById({ id: webhookId });
-    if (!webhook) {
-      //   logger.debug(
-      //        { webhookId },
-      //        "notify-webhook: no webhook registered with id",
-      //      );
-      return;
-    }
-
-    const { endpointUrl, httpMethod, headers } = webhook;
-    const headersFormatted = R.reduce(
-      headers || [],
-      (acc: any, h: any) => {
-        acc[h.header] = h.value;
-        return acc;
-      },
-      {},
-    );
-
-    const wreck = Wreck.defaults({
-      json: true,
-      headers: headersFormatted,
-    });
-
-    // http errors will bubble up and cause the job to fail and be retried
-    try {
-      //      logger.debug(
-      //        { webhookId, endpointUrl, httpMethod },
-      //        "notify-webhook: notifying",
-      //      );
-      await (httpMethod === "post"
-        ? wreck.post(endpointUrl, { payload })
-        : wreck.put(endpointUrl, { payload }));
-    } catch (error: any) {
-      //      logger.error(
-      //        { webhookId, error: error.output },
-      //        "notify-webhook failed with this error",
-      //      );
-      throw new Error(`webhook failed webhookId=${webhookId}`);
-    }
-  });
-
-export default notifyWebhooksTask;

apps/bridge-worker/tasks/common/notify-webhooks.ts

@@ -0,0 +1,31 @@
+import { db } from "bridge-common";
+
+export interface NotifyWebhooksOptions {
+  backendId: string;
+  payload: any;
+}
+
+const notifyWebhooksTask = async (
+  options: NotifyWebhooksOptions,
+): Promise<void> => {
+  const { backendId, payload } = options;
+
+  const webhooks = await db
+    .selectFrom("Webhook")
+    .selectAll()
+    .where("backendId", "=", backendId)
+    .execute();
+
+  for (const webhook of webhooks) {
+    const { endpointUrl, httpMethod, headers } = webhook;
+    const finalHeaders = { "Content-Type": "application/json", ...headers };
+
+    await fetch(endpointUrl, {
+      method: httpMethod,
+      headers: finalHeaders,
+      body: JSON.stringify(payload),
+    });
+  }
+};
+
+export default notifyWebhooksTask;

apps/bridge-worker/tasks/facebook/receive-facebook-message.ts

@@ -17,9 +17,13 @@ const receiveFacebookMessageTask = async ({
         .selectAll()
         .where("pageId", "=", pageId)
         .executeTakeFirstOrThrow();
+      const backendId = row.id;
+      const payload = {
+        text: messaging.message.text,
+        recipient: messaging.sender.id,
+      };
 
-      console.log({ row });
-      await worker.addJob("notify_webhooks", messaging);
+      await worker.addJob("common/notify-webhooks", { backendId, payload });
     }
   }
 };

apps/bridge-worker/tasks/facebook/send-facebook-message.ts

@@ -1,64 +1,34 @@
-interface SendFacebookMessageTaskOptions {}
+import { db } from "bridge-common";
+
+interface SendFacebookMessageTaskOptions {
+  token: string;
+  message: any;
+}
 
 const sendFacebookMessageTask = async (
   options: SendFacebookMessageTaskOptions,
 ): Promise<void> => {
-  console.log(options);
-
-
-  const message = await req.json();
-
-      const message = await req.json();
-    for (const entry of message.entry) {
-      for (const messaging of entry.messaging) {
-        const pageId = messaging.recipient.id;
-    const row = await db
-      .selectFrom("FacebookBot")
-      .selectAll()
-      .where("pageId", "=", pageId)
-      .executeTakeFirst();
-
-  console.log({ message });
-  const entry = message.entry[0];
-  console.log({ entry });
-  const messaging = entry?.messaging[0];
-  const pageId = messaging?.recipient?.id;
-  console.log({ pageId });
-  const row = await db
+  const { token, message } = options;
+  const { pageId, pageAccessToken } = await db
     .selectFrom("FacebookBot")
     .selectAll()
-    .where("pageId", "=", pageId)
-    .executeTakeFirst();
-
-  console.log({ row });
+    .where("token", "=", token)
+    .executeTakeFirstOrThrow();
 
   const endpoint = `https://graph.facebook.com/v19.0/${pageId}/messages`;
-  const inMessage = messaging?.message?.text;
+
   const outgoingMessage = {
-    recipient: { id: messaging?.sender?.id },
-    message: { text: `"${inMessage}", right back at you!` },
+    recipient: { id: message.recipient },
+    message: { text: `"${message.text}", right back at you!` },
     messaging_type: "RESPONSE",
-    access_token: row?.pageAccessToken,
+    access_token: pageAccessToken,
   };
-  console.log({ outgoingMessage });
-  const response = await fetch(endpoint, {
+
+  await fetch(endpoint, {
     method: "POST",
     headers: { "Content-Type": "application/json" },
     body: JSON.stringify(outgoingMessage),
   });
-
-  console.log({ response });
-
-  console.log(message);
-  const wu = await getWorkerUtils();
-
-  await wu.addJob("receive_facebook_message", message);
-
-  return NextResponse.json({ response: "ok" });
-
-  //  withDb(async (db: AppDatabase) => {
-  //    await notifyWebhooks(db, options);
-  //  });
 };
 
 export default sendFacebookMessageTask;

apps/bridge-worker/tasks/signal/receive-signal-message.ts

@@ -0,0 +1,11 @@
+import { db, getWorkerUtils } from "bridge-common";
+
+interface ReceiveSignalMessageTaskOptions {
+  message: any;
+}
+
+const receiveSignalMessageTask = async ({
+  message,
+}: ReceiveSignalMessageTaskOptions): Promise<void> => {};
+
+export default receiveSignalMessageTask;

apps/bridge-worker/tasks/signal/send-signal-message.ts

@@ -0,0 +1,11 @@
+import { db, getWorkerUtils } from "bridge-common";
+
+interface SendSignalMessageTaskOptions {
+  message: any;
+}
+
+const sendSignalMessageTask = async ({
+  message,
+}: SendSignalMessageTaskOptions): Promise<void> => {};
+
+export default sendSignalMessageTask;

apps/bridge-worker/tasks/voice/receive-voice-message.ts

@@ -0,0 +1,11 @@
+import { db, getWorkerUtils } from "bridge-common";
+
+interface ReceiveVoiceMessageTaskOptions {
+  message: any;
+}
+
+const receiveVoiceMessageTask = async ({
+  message,
+}: ReceiveVoiceMessageTaskOptions): Promise<void> => {};
+
+export default receiveVoiceMessageTask;

apps/bridge-worker/tasks/voice/send-voice-message.ts

@@ -0,0 +1,11 @@
+import { db, getWorkerUtils } from "bridge-common";
+
+interface SendVoiceMessageTaskOptions {
+  message: any;
+}
+
+const sendVoiceMessageTask = async ({
+  message,
+}: SendVoiceMessageTaskOptions): Promise<void> => {};
+
+export default sendVoiceMessageTask;

apps/bridge-worker/tasks/whatsapp/receive-whatsapp-message.ts

@@ -1,4 +1,17 @@
+import { db, getWorkerUtils } from "bridge-common";
+
+interface ReceiveWhatsappMessageTaskOptions {
+  message: any;
+}
+
+const receiveWhatsappMessageTask = async ({
+  message,
+}: ReceiveWhatsappMessageTaskOptions): Promise<void> => {};
+
+export default receiveWhatsappMessageTask;
+
 /* eslint-disable camelcase */
+/*
 import { withDb, AppDatabase } from "../../lib/db";
 import workerUtils from "../../lib/utils";
 
@@ -93,3 +106,4 @@ const whatsappMessageTask = async (
 };
 
 export default whatsappMessageTask;
+*/

apps/bridge-worker/tasks/whatsapp/send-whatsapp-message.ts

@@ -0,0 +1,11 @@
+import { db, getWorkerUtils } from "bridge-common";
+
+interface SendWhatsappMessageTaskOptions {
+  message: any;
+}
+
+const sendWhatsappMessageTask = async ({
+  message,
+}: SendWhatsappMessageTaskOptions): Promise<void> => {};
+
+export default sendWhatsappMessageTask;