Metamigo -> Bridge

apps/bridge-worker/tasks/import-label-studio.ts

@@ -0,0 +1,208 @@
+/* eslint-disable camelcase */
+import { convert } from "html-to-text";
+import fetch from "node-fetch";
+import { URLSearchParams } from "url";
+import { withDb, AppDatabase } from "../db";
+import { loadConfig } from "@digiresilience/bridge-config";
+import { tagMap } from "../lib/tag-map";
+
+type FormattedZammadTicket = {
+  data: Record<string, unknown>;
+  predictions: Record<string, unknown>[];
+};
+
+const getZammadTickets = async (
+  page: number,
+  minUpdatedTimestamp: Date,
+): Promise<[boolean, FormattedZammadTicket[]]> => {
+  const {
+    leafcutter: { zammadApiUrl, zammadApiKey, contributorName, contributorId },
+  } = await loadConfig();
+  const headers = { Authorization: `Token ${zammadApiKey}` };
+  let shouldContinue = false;
+  const docs = [];
+  const ticketsQuery = new URLSearchParams({
+    expand: "true",
+    sort_by: "updated_at",
+    order_by: "asc",
+    query: "state.name: closed",
+    per_page: "25",
+    page: `${page}`,
+  });
+  const rawTickets = await fetch(
+    `${zammadApiUrl}/tickets/search?${ticketsQuery}`,
+    { headers },
+  );
+  const tickets: any = await rawTickets.json();
+  console.log({ tickets });
+  if (!tickets || tickets.length === 0) {
+    return [shouldContinue, docs];
+  }
+
+  for await (const ticket of tickets) {
+    const { id: source_id, created_at, updated_at, close_at } = ticket;
+    const source_created_at = new Date(created_at);
+    const source_updated_at = new Date(updated_at);
+    const source_closed_at = new Date(close_at);
+    shouldContinue = true;
+
+    if (source_closed_at <= minUpdatedTimestamp) {
+      console.log(`Skipping ticket`, {
+        source_id,
+        source_updated_at,
+        source_closed_at,
+        minUpdatedTimestamp,
+      });
+      continue;
+    }
+
+    console.log(`Processing ticket`, {
+      source_id,
+      source_updated_at,
+      source_closed_at,
+      minUpdatedTimestamp,
+    });
+
+    const rawArticles = await fetch(
+      `${zammadApiUrl}/ticket_articles/by_ticket/${source_id}`,
+      { headers },
+    );
+    const articles: any = await rawArticles.json();
+    let articleText = "";
+
+    for (const article of articles) {
+      const { content_type: contentType, body } = article;
+
+      if (contentType === "text/html") {
+        const cleanArticleText = convert(body);
+        articleText += cleanArticleText + "\n\n";
+      } else {
+        articleText += body + "\n\n";
+      }
+    }
+
+    const tagsQuery = new URLSearchParams({
+      object: "Ticket",
+      o_id: source_id,
+    });
+
+    const rawTags = await fetch(`${zammadApiUrl}/tags?${tagsQuery}`, {
+      headers,
+    });
+    const { tags }: any = await rawTags.json();
+    const transformedTags = [];
+    for (const tag of tags) {
+      const outputs = tagMap[tag];
+      if (outputs) {
+        transformedTags.push(...outputs);
+      }
+    }
+
+    const doc: FormattedZammadTicket = {
+      data: {
+        ticket: articleText,
+        contributor_id: contributorId,
+        source_id,
+        source_closed_at,
+        source_created_at,
+        source_updated_at,
+      },
+      predictions: [],
+    };
+
+    const result = transformedTags.map((tag) => {
+      return {
+        type: "choices",
+        value: {
+          choices: [tag.value],
+        },
+        to_name: "ticket",
+        from_name: tag.field,
+      };
+    });
+
+    if (result.length > 0) {
+      doc.predictions.push({
+        model_version: `${contributorName}TranslatorV1`,
+        result,
+      });
+    }
+
+    docs.push(doc);
+  }
+
+  return [shouldContinue, docs];
+};
+
+const fetchFromZammad = async (
+  minUpdatedTimestamp: Date,
+): Promise<FormattedZammadTicket[]> => {
+  const pages = [...Array.from({ length: 10000 }).keys()];
+  const allTickets: FormattedZammadTicket[] = [];
+
+  for await (const page of pages) {
+    const [shouldContinue, tickets] = await getZammadTickets(
+      page + 1,
+      minUpdatedTimestamp,
+    );
+
+    if (!shouldContinue) {
+      break;
+    }
+
+    if (tickets.length > 0) {
+      allTickets.push(...tickets);
+    }
+  }
+
+  return allTickets;
+};
+
+const sendToLabelStudio = async (tickets: FormattedZammadTicket[]) => {
+  const {
+    leafcutter: { labelStudioApiUrl, labelStudioApiKey },
+  } = await loadConfig();
+
+  const headers = {
+    Authorization: `Token ${labelStudioApiKey}`,
+    "Content-Type": "application/json",
+    Accept: "application/json",
+  };
+
+  for await (const ticket of tickets) {
+    const res = await fetch(`${labelStudioApiUrl}/projects/1/import`, {
+      method: "POST",
+      headers,
+      body: JSON.stringify([ticket]),
+    });
+    const importResult = await res.json();
+
+    console.log(JSON.stringify(importResult, undefined, 2));
+  }
+};
+
+const importLabelStudioTask = async (): Promise<void> => {
+  withDb(async (db: AppDatabase) => {
+    const {
+      leafcutter: { contributorName },
+    } = await loadConfig();
+    const settingName = `${contributorName}ImportLabelStudioTask`;
+    const res: any = await db.settings.findByName(settingName);
+    const startTimestamp = res?.value?.minUpdatedTimestamp
+      ? new Date(res.value.minUpdatedTimestamp as string)
+      : new Date("2023-03-01");
+    const tickets = await fetchFromZammad(startTimestamp);
+
+    if (tickets.length > 0) {
+      await sendToLabelStudio(tickets);
+      const lastTicket = tickets.pop();
+      const newLastTimestamp = lastTicket.data.source_closed_at;
+      console.log({ newLastTimestamp });
+      await db.settings.upsert(settingName, {
+        minUpdatedTimestamp: newLastTimestamp,
+      });
+    }
+  });
+};
+
+export default importLabelStudioTask;

apps/bridge-worker/tasks/import-leafcutter.ts

@@ -0,0 +1,174 @@
+/* eslint-disable camelcase */
+import fetch from "node-fetch";
+import { URLSearchParams } from "url";
+import { withDb, AppDatabase } from "../db";
+import { loadConfig } from "@digiresilience/bridge-config";
+
+type LabelStudioTicket = {
+  id: string;
+  is_labeled: boolean;
+  annotations: Record<string, unknown>[];
+  data: Record<string, unknown>;
+  updated_at: string;
+};
+
+type LeafcutterTicket = {
+  id: string;
+  incident: string[];
+  technology: string[];
+  targeted_group: string[];
+  country: string[];
+  region: string[];
+  continent: string[];
+  date: Date;
+  origin: string;
+  origin_id: string;
+  source_created_at: string;
+  source_updated_at: string;
+};
+
+const getLabelStudioTickets = async (
+  page: number,
+): Promise<LabelStudioTicket[]> => {
+  const {
+    leafcutter: { labelStudioApiUrl, labelStudioApiKey },
+  } = await loadConfig();
+
+  const headers = {
+    Authorization: `Token ${labelStudioApiKey}`,
+    Accept: "application/json",
+  };
+  const ticketsQuery = new URLSearchParams({
+    page_size: "50",
+    page: `${page}`,
+  });
+  console.log({ url: `${labelStudioApiUrl}/projects/1/tasks?${ticketsQuery}` });
+  const res = await fetch(
+    `${labelStudioApiUrl}/projects/1/tasks?${ticketsQuery}`,
+    { headers },
+  );
+  console.log({ res });
+  const tasksResult: any = await res.json();
+  console.log({ tasksResult });
+
+  return tasksResult;
+};
+
+const fetchFromLabelStudio = async (
+  minUpdatedTimestamp: Date,
+): Promise<LabelStudioTicket[]> => {
+  const pages = [...Array.from({ length: 10000 }).keys()];
+  const allDocs: LabelStudioTicket[] = [];
+
+  for await (const page of pages) {
+    const docs = await getLabelStudioTickets(page + 1);
+    console.log({ page, docs });
+
+    if (docs && docs.length > 0) {
+      for (const doc of docs) {
+        const updatedAt = new Date(doc.updated_at);
+        console.log({ updatedAt, minUpdatedTimestamp });
+        if (updatedAt > minUpdatedTimestamp) {
+          console.log(`Adding doc`, { doc });
+          allDocs.push(doc);
+        }
+      }
+    } else {
+      break;
+    }
+  }
+
+  console.log({ allDocs });
+  return allDocs;
+};
+
+const sendToLeafcutter = async (tickets: LabelStudioTicket[]) => {
+  const {
+    leafcutter: {
+      contributorId,
+      opensearchApiUrl,
+      opensearchUsername,
+      opensearchPassword,
+    },
+  } = await loadConfig();
+
+  console.log({ tickets });
+  const filteredTickets = tickets.filter((ticket) => ticket.is_labeled);
+  console.log({ filteredTickets });
+  const finalTickets: LeafcutterTicket[] = filteredTickets.map((ticket) => {
+    const {
+      id,
+      annotations,
+      data: { source_id, source_created_at, source_updated_at },
+    } = ticket;
+
+    const getTags = (tags: Record<string, any>[], name: string) =>
+      tags
+        .filter((tag) => tag.from_name === name)
+        .map((tag) => tag.value.choices)
+        .flat();
+
+    const allTags = annotations.map(({ result }) => result).flat();
+    const incident = getTags(allTags, "incidentType tag");
+    const technology = getTags(allTags, "platform tag");
+    const country = getTags(allTags, "country tag");
+    const targetedGroup = getTags(allTags, "targetedGroup tag");
+
+    return {
+      id,
+      incident,
+      technology,
+      targeted_group: targetedGroup,
+      country,
+      region: [],
+      continent: [],
+      date: new Date(source_created_at as string),
+      origin: contributorId,
+      origin_id: source_id as string,
+      source_created_at: source_created_at as string,
+      source_updated_at: source_updated_at as string,
+    };
+  });
+
+  console.log("Sending to Leafcutter");
+  console.log({ finalTickets });
+
+  const result = await fetch(opensearchApiUrl, {
+    method: "POST",
+    headers: {
+      "Content-Type": "application/json",
+      Authorization: `Basic ${Buffer.from(`${opensearchUsername}:${opensearchPassword}`).toString("base64")}`,
+    },
+    body: JSON.stringify({ tickets: finalTickets }),
+  });
+  console.log({ result });
+};
+
+const importLeafcutterTask = async (): Promise<void> => {
+  withDb(async (db: AppDatabase) => {
+    const {
+      leafcutter: { contributorName },
+    } = await loadConfig();
+    const settingName = `${contributorName}ImportLeafcutterTask`;
+    const res: any = await db.settings.findByName(settingName);
+    const startTimestamp = res?.value?.minUpdatedTimestamp
+      ? new Date(res.value.minUpdatedTimestamp as string)
+      : new Date("2023-03-01");
+    const newLastTimestamp = new Date();
+    console.log({
+      contributorName,
+      settingName,
+      res,
+      startTimestamp,
+      newLastTimestamp,
+    });
+    const tickets = await fetchFromLabelStudio(startTimestamp);
+    console.log({ tickets });
+    await sendToLeafcutter(tickets);
+    await db.settings.upsert(settingName, {
+      minUpdatedTimestamp: newLastTimestamp,
+    });
+  });
+};
+
+export default importLeafcutterTask;

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

@@ -0,0 +1,57 @@
+import Wreck from "@hapi/wreck";
+import * as R from "remeda";
+import { withDb, AppDatabase } from "../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/signald-message.ts

@@ -0,0 +1,90 @@
+/* eslint-disable camelcase */
+import logger from "../logger";
+import { IncomingMessagev1 } from "@digiresilience/node-signald/build/main/generated";
+import { withDb, AppDatabase } from "../db";
+import workerUtils from "../utils";
+
+interface WebhookPayload {
+  to: string;
+  from: string;
+  message_id: string;
+  sent_at: string;
+  message: string;
+  attachment: string | null;
+  filename: string | null;
+  mime_type: string | null;
+}
+
+interface SignaldMessageTaskOptions {
+  message: IncomingMessagev1;
+  botId: string;
+  botPhoneNumber: string;
+  attachment: string;
+  filename: string;
+  mimetype: string;
+}
+
+const formatPayload = (opts: SignaldMessageTaskOptions): WebhookPayload => {
+  const { botId, botPhoneNumber, message, attachment, filename, mimetype } = opts;
+  const { source, timestamp, data_message: dataMessage } = message;
+
+  const { number }: any = source;
+
+  const { body, attachments }: any = dataMessage;
+
+  return {
+    to: botPhoneNumber,
+    from: number,
+    message_id: `${botId}-${timestamp}`,
+    sent_at: `${timestamp}`,
+    message: body,
+    attachment,
+    filename,
+    mime_type: mimetype,
+  };
+};
+
+const notifyWebhooks = async (
+  db: AppDatabase,
+  messageInfo: SignaldMessageTaskOptions
+) => {
+  const {
+    botId,
+    message: { timestamp },
+  } = messageInfo;
+  const webhooks = await db.webhooks.findAllByBackendId("signal", botId);
+  if (webhooks && webhooks.length === 0) {
+    logger.debug({ botId }, "no webhooks registered for signal bot");
+    return;
+  }
+
+  webhooks.forEach(({ id }) => {
+    const payload = formatPayload(messageInfo);
+    logger.debug(
+      { payload },
+      "formatted signal bot payload for notify-webhook"
+    );
+    workerUtils.addJob(
+      "notify-webhook",
+      {
+        payload,
+        webhookId: id,
+      },
+      {
+        // this de-deduplicates the job
+        jobKey: `webhook-${id}-message-${botId}-${timestamp}`,
+      }
+    );
+  });
+};
+
+const signaldMessageTask = async (
+  options: SignaldMessageTaskOptions
+): Promise<void> => {
+  console.log(options);
+  withDb(async (db: AppDatabase) => {
+    await notifyWebhooks(db, options);
+  });
+};
+
+export default signaldMessageTask;

apps/bridge-worker/tasks/twilio-recording.ts

@@ -0,0 +1,101 @@
+import Wreck from "@hapi/wreck";
+import { withDb, AppDatabase } from "../db";
+import { twilioClientFor } from "../common";
+import { CallInstance } from "twilio/lib/rest/api/v2010/account/call";
+import workerUtils from "../utils";
+
+interface WebhookPayload {
+  startTime: string;
+  endTime: string;
+  to: string;
+  from: string;
+  duration: string;
+  callSid: string;
+  recording: string;
+  mimeType: string;
+}
+
+const getTwilioRecording = async (url: string) => {
+  try {
+    const { payload } = await Wreck.get(url);
+    return { recording: payload as Buffer };
+  } catch (error: any) {
+    console.error(error.output);
+    return { error: error.output };
+  }
+};
+
+const formatPayload = (
+  call: CallInstance,
+  recording: Buffer
+): WebhookPayload => {
+  return {
+    startTime: call.startTime.toISOString(),
+    endTime: call.endTime.toISOString(),
+    to: call.toFormatted,
+    from: call.fromFormatted,
+    duration: call.duration,
+    callSid: call.sid,
+    recording: recording.toString("base64"),
+    mimeType: "audio/mpeg",
+  };
+};
+
+const notifyWebhooks = async (
+  db: AppDatabase,
+  voiceLineId: string,
+  call: CallInstance,
+  recording: Buffer
+) => {
+  const webhooks = await db.webhooks.findAllByBackendId("voice", voiceLineId);
+  if (webhooks && webhooks.length === 0) return;
+
+  webhooks.forEach(({ id }) => {
+    const payload = formatPayload(call, recording);
+    workerUtils.addJob(
+      "notify-webhook",
+      {
+        payload,
+        webhookId: id,
+      },
+      {
+        // this de-depuplicates the job
+        jobKey: `webhook-${id}-call-${call.sid}`,
+      }
+    );
+  });
+};
+
+interface TwilioRecordingTaskOptions {
+  accountSid: string;
+  callSid: string;
+  recordingSid: string;
+  voiceLineId: string;
+}
+
+const twilioRecordingTask = async (
+  options: TwilioRecordingTaskOptions
+): Promise<void> =>
+  withDb(async (db: AppDatabase) => {
+    const { voiceLineId, accountSid, callSid, recordingSid } = options;
+
+    const voiceLine = await db.voiceLines.findById({ id: voiceLineId });
+    if (!voiceLine) return;
+
+    const provider = await db.voiceProviders.findByTwilioAccountSid(accountSid);
+    if (!provider) return;
+
+    const client = twilioClientFor(provider);
+    const meta = await client.recordings(recordingSid).fetch();
+
+    const mp3Url = "https://api.twilio.com/" + meta.uri.slice(0, -4) + "mp3";
+    const { recording, error } = await getTwilioRecording(mp3Url);
+    if (error) {
+      throw new Error(`failed to get recording for call ${callSid}`);
+    }
+
+    const call = await client.calls(callSid).fetch();
+    await notifyWebhooks(db, voiceLineId, call, recording!);
+  });
+
+export default twilioRecordingTask;

apps/bridge-worker/tasks/voice-line-audio-update.ts

@@ -0,0 +1,48 @@
+import { createHash } from "crypto";
+import { withDb, AppDatabase } from "../db";
+import { convert } from "../lib/media-convert";
+
+interface VoiceLineAudioUpdateTaskOptions {
+  voiceLineId: string;
+}
+
+const sha1sum = (v: any) => {
+  const shasum = createHash("sha1");
+  shasum.update(v);
+  return shasum.digest("hex");
+};
+
+const voiceLineAudioUpdateTask = async (
+  payload: VoiceLineAudioUpdateTaskOptions
+): Promise<void> =>
+  withDb(async (db: AppDatabase) => {
+    const { voiceLineId } = payload;
+    const voiceLine = await db.voiceLines.findById({ id: voiceLineId });
+    if (!voiceLine) return;
+    if (!voiceLine?.promptAudio?.["audio/webm"]) return;
+
+    const webm = Buffer.from(voiceLine.promptAudio["audio/webm"], "base64");
+    const webmSha1 = sha1sum(webm);
+
+    if (
+      voiceLine.promptAudio.checksum &&
+      voiceLine.promptAudio.checksum === webmSha1
+    ) {
+      // already converted
+      return;
+    }
+
+    const mp3 = await convert(webm);
+    await db.voiceLines.updateById(
+      { id: voiceLine.id },
+      {
+        promptAudio: {
+          ...voiceLine.promptAudio,
+          "audio/mpeg": mp3.toString("base64"),
+          checksum: webmSha1,
+        },
+      }
+    );
+  });
+
+export default voiceLineAudioUpdateTask;

apps/bridge-worker/tasks/voice-line-delete.ts

@@ -0,0 +1,41 @@
+import Twilio from "twilio";
+import config from "@digiresilience/bridge-config";
+import { withDb, AppDatabase } from "../db";
+
+interface VoiceLineDeleteTaskOptions {
+  voiceLineId: string;
+  providerId: string;
+  providerLineSid: string;
+}
+
+const voiceLineDeleteTask = async (
+  payload: VoiceLineDeleteTaskOptions,
+): Promise<void> =>
+  withDb(async (db: AppDatabase) => {
+    const { voiceLineId, providerId, providerLineSid } = payload;
+    const provider = await db.voiceProviders.findById({ id: providerId });
+    if (!provider) return;
+
+    const { accountSid, apiKeySid, apiKeySecret } = provider.credentials;
+    if (!accountSid || !apiKeySid || !apiKeySecret)
+      throw new Error(
+        `twilio provider ${provider.name} does not have credentials`,
+      );
+
+    const client = Twilio(apiKeySid, apiKeySecret, {
+      accountSid,
+    });
+
+    const number = await client.incomingPhoneNumbers(providerLineSid).fetch();
+    if (
+      number &&
+      number.voiceUrl ===
+        `${config.frontend.url}/api/v1/voice/twilio/record/${voiceLineId}`
+    )
+      await client.incomingPhoneNumbers(providerLineSid).update({
+        voiceUrl: "",
+        voiceMethod: "POST",
+      });
+  });
+
+export default voiceLineDeleteTask;

apps/bridge-worker/tasks/voice-line-provider-update.ts

@@ -0,0 +1,38 @@
+import Twilio from "twilio";
+import config from "@digiresilience/bridge-config";
+import { withDb, AppDatabase } from "../db";
+
+interface VoiceLineUpdateTaskOptions {
+  voiceLineId: string;
+}
+
+const voiceLineUpdateTask = async (
+  payload: VoiceLineUpdateTaskOptions,
+): Promise<void> =>
+  withDb(async (db: AppDatabase) => {
+    const { voiceLineId } = payload;
+    const voiceLine = await db.voiceLines.findById({ id: voiceLineId });
+    if (!voiceLine) return;
+
+    const provider = await db.voiceProviders.findById({
+      id: voiceLine.providerId,
+    });
+    if (!provider) return;
+
+    const { accountSid, apiKeySid, apiKeySecret } = provider.credentials;
+    if (!accountSid || !apiKeySid || !apiKeySecret)
+      throw new Error(
+        `twilio provider ${provider.name} does not have credentials`,
+      );
+
+    const client = Twilio(apiKeySid, apiKeySecret, {
+      accountSid,
+    });
+
+    await client.incomingPhoneNumbers(voiceLine.providerLineSid).update({
+      voiceUrl: `${config.frontend.url}/api/v1/voice/twilio/record/${voiceLineId}`,
+      voiceMethod: "POST",
+    });
+  });
+
+export default voiceLineUpdateTask;

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

@@ -0,0 +1,94 @@
+/* eslint-disable camelcase */
+import { withDb, AppDatabase } from "../db";
+import workerUtils from "../utils";
+
+interface WebhookPayload {
+  to: string;
+  from: string;
+  message_id: string;
+  sent_at: string;
+  message: string;
+  attachment: string;
+  filename: string;
+  mime_type: string;
+}
+
+interface WhatsappMessageTaskOptions {
+  waMessageId: string;
+  waMessage: string;
+  waTimestamp: string;
+  attachment: string;
+  filename: string;
+  mimetype: string;
+  botPhoneNumber: string;
+  whatsappBotId: string;
+}
+
+const formatPayload = (
+  messageInfo: WhatsappMessageTaskOptions
+): WebhookPayload => {
+  const {
+    waMessageId,
+    waMessage,
+    waTimestamp,
+    attachment,
+    filename,
+    mimetype,
+    botPhoneNumber,
+  } = messageInfo;
+  const parsedMessage = JSON.parse(waMessage);
+  const message = parsedMessage.message?.conversation ?? 
+    parsedMessage.message?.extendedTextMessage?.text ?? 
+    parsedMessage.message?.imageMessage?.caption ??
+    parsedMessage.message?.videoMessage?.caption;
+
+  return {
+    to: botPhoneNumber,
+    from: parsedMessage.key.remoteJid,
+    message_id: waMessageId,
+    sent_at: waTimestamp,
+    message,
+    attachment,
+    filename,
+    mime_type: mimetype,
+  };
+};
+
+const notifyWebhooks = async (
+  db: AppDatabase,
+  messageInfo: WhatsappMessageTaskOptions
+) => {
+  const { waMessageId, whatsappBotId } = messageInfo;
+  const webhooks = await db.webhooks.findAllByBackendId(
+    "whatsapp",
+    whatsappBotId
+  );
+  if (webhooks && webhooks.length === 0) return;
+
+  webhooks.forEach(({ id }) => {
+    const payload = formatPayload(messageInfo);
+    console.log({ payload });
+    workerUtils.addJob(
+      "notify-webhook",
+      {
+        payload,
+        webhookId: id,
+      },
+      {
+        // this de-deduplicates the job
+        jobKey: `webhook-${id}-message-${waMessageId}`,
+      }
+    );
+  });
+};
+
+const whatsappMessageTask = async (
+  options: WhatsappMessageTaskOptions
+): Promise<void> => {
+  console.log(options);
+  withDb(async (db: AppDatabase) => {
+    await notifyWebhooks(db, options);
+  });
+};
+
+export default whatsappMessageTask;