More work on WhatsApp messages

apps/webapp/app/models/workflowRun.server.ts

@@ -48,6 +48,7 @@ export async function startWorkflowRun(id: string, apiKey: string) {
 
     await resolveDisconnectionStepsInRun(workflowRun.id);
   } else {
+    console.log(`[startWorkflowRun] ${workflowRun.id} is running`);
     await prisma.workflowRun.update({
       where: { id: workflowRun.id },
       data: {

apps/webapp/app/services/events/dispatch.server.ts

@@ -22,7 +22,7 @@ export class DispatchEvent {
     });
 
     if (!event) {
-      throw new Error("Event not found");
+      throw new Error(`Event not found: ${id}`);
     }
 
     console.log(

apps/webapp/app/services/externalSources/handleExternalSource.server.ts

@@ -28,7 +28,7 @@ type TriggeredEventResponse = {
     event: string;
     timestamp?: string;
     context?: any;
-  };
+  }[];
 };
 
 export type HandledExternalEventResponse =
@@ -57,28 +57,30 @@ export class HandleExternalSource {
 
     switch (possibleEvent.status) {
       case "ok": {
-        const { id, payload, event, timestamp, context } = possibleEvent.data;
+        for (let index = 0; index < possibleEvent.data.length; index++) {
+          const { id, payload, event, timestamp, context } =
+            possibleEvent.data[index];
 
-        const ingestService = new IngestEvent();
+          const ingestService = new IngestEvent();
 
-        await ingestService.call(
-          {
-            id,
-            payload,
-            name: event,
-            type: externalSource.type,
-            service: serviceIdentifier,
-            timestamp,
-            context,
-          },
-          externalSource.organization
-        );
+          await ingestService.call(
+            {
+              id,
+              payload,
+              name: event,
+              type: externalSource.type,
+              service: serviceIdentifier,
+              timestamp,
+              context,
+            },
+            externalSource.organization
+          );
+        }
 
         return true;
       }
       case "ignored": {
         console.log(`Ignored external event: ${possibleEvent.reason}`);
-
         return true;
       }
       case "error": {
@@ -175,15 +177,17 @@ export class HandleExternalSource {
 
     return {
       status: "ok",
-      data: {
-        id: ulid(),
-        payload: request.body,
-        event: source.event,
-        context: {
-          headers: request.headers,
-          externalSourceId: externalSource.id,
+      data: [
+        {
+          id: ulid(),
+          payload: request.body,
+          event: source.event,
+          context: {
+            headers: request.headers,
+            externalSourceId: externalSource.id,
+          },
         },
-      },
+      ],
     };
   }
 }

examples/whatsapp/src/index.ts

@@ -11,7 +11,7 @@ new Trigger({
     accountId: "114848614845931",
   }),
   run: async (event, ctx) => {
-    await ctx.logger.info(`Action was ${event.action}`);
+    await ctx.logger.info(`Action was ${event}`);
 
     return {};
   },

integrations/github/src/internal/webhooks.ts

@@ -87,7 +87,7 @@ export class GitHubWebhookIntegration implements WebhookIntegration {
 
     return {
       status: "ok" as const,
-      data: { id, payload: options.request.body, event, context },
+      data: [{ id, payload: options.request.body, event, context }],
     };
   }
 

integrations/whatsapp/src/events.ts

@@ -12,13 +12,7 @@ export function messageEvent(params: {
       filter: {
         service: ["whatsapp"],
         payload: {
-          object: ["whatsapp_business_account"],
-          entry: {
-            id: [params.accountId],
-            changes: {
-              field: ["messages"],
-            },
-          },
+          type: ["message"],
         },
         event: ["messages"],
       },

integrations/whatsapp/src/internal/webhooks.ts

@@ -1,6 +1,6 @@
 import crypto from "crypto";
 import { ulid } from "ulid";
-import { getAccessToken } from "@trigger.dev/integration-sdk";
+import { getAccessToken, ReceivedWebhook } from "@trigger.dev/integration-sdk";
 import type {
   DisplayProperty,
   HandleWebhookOptions,
@@ -15,9 +15,7 @@ export class WhatsAppWebhookIntegration implements WebhookIntegration {
 
     switch (whatsAppSource.subresource) {
       case "messages":
-        return `messages.${
-          whatsAppSource.accountId
-        }.${whatsAppSource.events.join(".")}`;
+        return `messages.${whatsAppSource.accountId}.${whatsAppSource.event}`;
       default:
         throw new Error(`Unknown subresource`);
     }
@@ -75,14 +73,11 @@ export class WhatsAppWebhookIntegration implements WebhookIntegration {
       "x-forwarded-proto",
     ]);
 
+    const data = getData(options.request.body, context);
+
     return {
       status: "ok" as const,
-      data: {
-        id: ulid(),
-        payload: options.request.body,
-        event: "messages",
-        context,
-      },
+      data,
     };
   }
 
@@ -97,7 +92,45 @@ function parseWebhookSource(source: unknown) {
   return WebhookSourceSchema.parse(source);
 }
 
-function handleMessagesEvent(options: HandleWebhookOptions) {}
+function getData(
+  body: any,
+  context: Record<string, string>
+): ReceivedWebhook[] {
+  const webhooks: ReceivedWebhook[] = [];
+  for (const entry of body.entry) {
+    for (const change of entry.changes) {
+      if (change.field === "messages") {
+        const messageData = change.value;
+
+        const metadata = messageData.metadata;
+        const contacts = messageData.contacts;
+
+        for (const message of messageData.messages) {
+          const timestamp = `${message.timestamp}000`;
+          webhooks.push({
+            id: message.id as string,
+            payload: {
+              type: "message",
+              contacts,
+              metadata,
+              message: {
+                ...message,
+                timestamp,
+              },
+            },
+            timestamp,
+            event: "messages",
+            context,
+          });
+        }
+      } else {
+        console.error(`Unknown field ${change.field}`);
+      }
+    }
+  }
+
+  return webhooks;
+}
 
 function omit<T extends Record<string, unknown>, K extends keyof T>(
   obj: T,

integrations/whatsapp/src/schemas/index.ts

@@ -4,8 +4,10 @@ import * as messageEvents from "./messageEvents";
 export const WebhookSourceSchema = z.object({
   subresource: z.literal("messages"),
   accountId: z.string(),
-  scopes: z.array(z.string()),
-  events: z.array(z.string()),
+  event: z.string(),
+  verifyPayload: z.object({
+    enabled: z.boolean(),
+  }),
 });
 
 export { messageEvents };

integrations/whatsapp/src/schemas/messageEvents.ts

@@ -1,3 +1,41 @@
 import { z } from "zod";
 
-export const messageEventSchema = z.any();
+const metadataSchema = z.object({
+  display_phone_number: z.string(),
+  phone_number_id: z.string(),
+});
+
+const contactSchema = z.object({
+  profile: z.object({ name: z.string() }),
+  wa_id: z.string(),
+});
+
+const commonMessageData = z.object({
+  id: z.string(),
+  from: z.string(),
+  timestamp: z.date(),
+});
+
+const textMessageEventSchema = z.object({
+  type: z.literal("text"),
+  text: z.object({ body: z.string() }),
+});
+
+const audioMessageEventSchema = z.object({
+  type: z.literal("audio"),
+  audio: z.object({
+    id: z.string(),
+    mime_type: z.string(),
+  }),
+});
+
+const messageSchema = z
+  .discriminatedUnion("type", [textMessageEventSchema, audioMessageEventSchema])
+  .and(commonMessageData);
+
+export const messageEventSchema = z.object({
+  type: z.literal("message"),
+  contacts: z.array(contactSchema),
+  metadata: metadataSchema,
+  message: messageSchema,
+});

packages/integration-sdk/src/types.ts

@@ -85,7 +85,7 @@ export interface WebhookIntegration {
   handleWebhookRequest: (
     options: HandleWebhookOptions
   ) =>
-    | { status: "ok"; data: ReceivedWebhook }
+    | { status: "ok"; data: ReceivedWebhook[] }
     | { status: "ignored"; reason: string }
     | { status: "error"; error: string };
   verifyWebhookRequest: (