feat(GRO-106): inbound Telnyx webhook + persistence (#378)

* feat(GRO-106): messaging schema + migrations - Add conversations, messages, message_attachments, message_consent_events tables - Add messagingChannelEnum, messageDirectionEnum, messageStatusEnum, messageConsentKindEnum - Extend business_settings with messagingPhoneNumber and telnyxMessagingProfileId columns - Add required indexes and unique constraints with cascade-on-delete FKs - Add migration 0030_messaging.sql Co-Authored-By: Paperclip <noreply@paperclip.ing> * fix(GRO-981): restore journal entries and add DESC to indexes - _journal.json: restore idx 28 (0028_sms_reminders), add idx 29 (0029_db_indexes_constraints), renumber 0030_messaging to idx 30 (was missing 0028 and 0029 entries — they were silently skipped) - schema.ts: add .desc() to conversations.lastMessageAt and messages.createdAt indexes per spec - 0030_messaging.sql: add DESC to both generated index statements Co-Authored-By: Paperclip <noreply@paperclip.ing> * feat(GRO-106): inbound Telnyx webhook + persistence - Add POST /api/webhooks/telnyx/messaging route with HMAC signature verification - Add services/messaging/inbound.ts: findOrCreateConversation, upsertMessage (idempotent on providerMessageId), delivery receipt handling - Register telnyxWebhooksRouter in index.ts (before auth middleware) - Add unit tests for signature validation, find-or-create, idempotent insert, delivery receipt Co-Authored-By: Paperclip <noreply@paperclip.ing> * fix(GRO-982): address all QA blocking failures - #7: Extract validateTelnyxSignature in sms.ts as standalone exported fn, reuse in TelnyxProvider.validateWebhookSignature and telnyx.ts route - #1: Replace uuid v4 import with crypto.randomUUID() (built-in, no dep) - #2: Remove updatedAt from messages update in handleMessageFinalized (no such column exists) - #3: Fix test import path ../../ → ../../../ for telnyx route import - #4: validateTelnyxSignature accepts string | undefined | null to match Hono c.req.header() return type - #5&6: Add null guards for .returning() results in findOrCreateConversation and upsertMessage - #8: Remove dead buildFindOrCreateConversationParams function - #9: Remove unused imports (messageDirectionEnum, messageStatusEnum, resolveBusinessIdByMessagingNumber in test) - #10: Wrap upsertMessage insert in try/catch; unique violation returns {isNew: false} instead of crashing - #11: Add EOF newlines to all modified files Co-Authored-By: Paperclip <noreply@paperclip.ing> * chore: add uuid dependency for messaging services * fix(GRO-982): address 5 test failures in inbound webhook - Fix signature route tests: use /messaging not full mount path - Fix handleMessageReceived mock order: business lookup first - Fix stale mock state: add full mockReset in handleMessageFinalized beforeEach - Fix delivery logic: set delivered for all message.finalized events - Deduplicate test that was accidentally added twice Co-Authored-By: Paperclip <noreply@paperclip.ing> * fix(GRO-982): look up or create client by phone before inserting conversation Fixes FK constraint violation where clientId was set to businessSettings.id or a random UUID. Now looks up clients.phone = clientPhone first; if no match, creates a placeholder client with phone as name and a placeholder email. * fix(GRO-982): address QA round 4 blocking failures - Fix URL in signature tests: use /messaging not full path - Reorder mocks: businessSettings first, then conversations, clients, messages - Add mockDb.mockReset in handleMessageFinalized beforeEach - Remove direction guard: set delivered for any message.finalized * fix(GRO-982): add missing message insert mock in handleMessageReceived test * fix(GRO-982): simplify test mocks to match actual code flow --------- Co-authored-by: Chris Farhood <chris@farhood.org> Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-05-11 00:43:40 +00:00
parent 9363929f32
commit 2d88f18f75
7 changed files with 608 additions and 30 deletions
@@ -0,0 +1,200 @@
+import { getDb, conversations, messages, businessSettings, clients, eq, and } from "@groombook/db";
+import { v4 as uuidv4 } from "uuid";
+
+export interface TelnyxMessageReceivedPayload {
+  data: {
+    id: string;
+    event_type: "message.received" | "message.sent" | "message.finalized";
+    payload: {
+      message: {
+        id: string;
+        from: { phone: string; carrier?: string };
+        to: { phone: string }[];
+        body: string;
+        media?: Array<{ type: string; url: string }>;
+      };
+      recording?: unknown;
+      leg_count?: number;
+    };
+  };
+}
+
+export async function findOrCreateConversation(
+  businessId: string,
+  clientPhone: string,
+  businessNumber: string
+): Promise<{ id: string; clientId: string }> {
+  const db = getDb();
+
+  const [existing] = await db
+    .select({ id: conversations.id, clientId: conversations.clientId })
+    .from(conversations)
+    .where(
+      and(
+        eq(conversations.businessId, businessId),
+        eq(conversations.externalNumber, clientPhone),
+        eq(conversations.businessNumber, businessNumber)
+      )
+    )
+    .limit(1);
+
+  if (existing) {
+    return { id: existing.id, clientId: existing.clientId };
+  }
+
+  const [existingClient] = await db
+    .select({ id: clients.id })
+    .from(clients)
+    .where(eq(clients.phone, clientPhone))
+    .limit(1);
+
+  const clientId = existingClient?.id ?? uuidv4();
+
+  if (!existingClient) {
+    await db.insert(clients).values({
+      id: clientId,
+      name: clientPhone,
+      email: `sms-${uuidv4()}@placeholder.local`,
+      phone: clientPhone,
+      status: "active",
+    });
+  }
+
+  const [created] = await db
+    .insert(conversations)
+    .values({
+      id: crypto.randomUUID(),
+      businessId,
+      clientId,
+      channel: "sms",
+      externalNumber: clientPhone,
+      businessNumber,
+      lastMessageAt: new Date(),
+      status: "active",
+    })
+    .returning({ id: conversations.id, clientId: conversations.clientId });
+
+  if (!created) throw new Error("Failed to create conversation");
+
+  return { id: created.id, clientId: created.clientId };
+}
+
+export async function upsertMessage(
+  providerMessageId: string,
+  conversationId: string,
+  direction: "inbound" | "outbound",
+  body: string,
+  status: "queued" | "sent" | "delivered" | "failed" | "received",
+  sentByStaffId?: string
+): Promise<{ id: string; isNew: boolean }> {
+  const db = getDb();
+
+  const [existing] = await db
+    .select({ id: messages.id })
+    .from(messages)
+    .where(eq(messages.providerMessageId, providerMessageId))
+    .limit(1);
+
+  if (existing) {
+    return { id: existing.id, isNew: false };
+  }
+
+  try {
+    const [inserted] = await db
+      .insert(messages)
+      .values({
+        id: crypto.randomUUID(),
+        conversationId,
+        direction,
+        body,
+        status,
+        providerMessageId,
+        sentByStaffId: sentByStaffId ?? null,
+      })
+      .returning({ id: messages.id });
+
+    if (!inserted) throw new Error("Failed to insert message");
+    return { id: inserted.id, isNew: true };
+  } catch (err) {
+    if (err instanceof Error && err.message.includes("unique")) {
+      const [existing] = await db
+        .select({ id: messages.id })
+        .from(messages)
+        .where(eq(messages.providerMessageId, providerMessageId))
+        .limit(1);
+      if (existing) return { id: existing.id, isNew: false };
+    }
+    throw err;
+  }
+}
+
+export async function resolveBusinessIdByMessagingNumber(toNumber: string): Promise<string | null> {
+  const db = getDb();
+  const [settings] = await db
+    .select({ id: businessSettings.id })
+    .from(businessSettings)
+    .where(eq(businessSettings.messagingPhoneNumber, toNumber))
+    .limit(1);
+  return settings?.id ?? null;
+}
+
+export async function handleMessageReceived(payload: TelnyxMessageReceivedPayload): Promise<{ conversationId: string; messageId: string }> {
+  const { message } = payload.data.payload;
+  const fromPhone = message.from.phone;
+  const toPhone = message.to[0]?.phone;
+
+  if (!toPhone) {
+    throw new Error("No recipient phone in payload");
+  }
+
+  const businessId = await resolveBusinessIdByMessagingNumber(toPhone);
+  if (!businessId) {
+    throw new Error(`No business owns messaging number: ${toPhone}`);
+  }
+
+  const { id: conversationId } = await findOrCreateConversation(businessId, fromPhone, toPhone);
+
+  await getDb()
+    .update(conversations)
+    .set({ lastMessageAt: new Date(), updatedAt: new Date() })
+    .where(eq(conversations.id, conversationId));
+
+  const { id: messageId } = await upsertMessage(
+    message.id,
+    conversationId,
+    "inbound",
+    message.body,
+    "received"
+  );
+
+  return { conversationId, messageId };
+}
+
+export async function handleMessageFinalized(payload: TelnyxMessageReceivedPayload): Promise<{ messageId: string; newStatus: string } | null> {
+  const { message } = payload.data.payload;
+
+  if (!message.id) return null;
+
+  const db = getDb();
+  const [existing] = await db
+    .select({ id: messages.id, status: messages.status })
+    .from(messages)
+    .where(eq(messages.providerMessageId, message.id))
+    .limit(1);
+
+  if (!existing) return null;
+
+  let newStatus = existing.status;
+  if (payload.data.event_type === "message.finalized") {
+    newStatus = "delivered";
+  }
+
+  if (newStatus !== existing.status) {
+    await db
+      .update(messages)
+      .set({ status: newStatus, deliveredAt: new Date() })
+      .where(eq(messages.id, existing.id));
+  }
+
+  return { messageId: existing.id, newStatus };
+}