wacrm/src/app/api/whatsapp/webhook/route.ts
Arnas Donauskas 5d5c5f0a20
fix(broadcasts): per-recipient variables, webhook status sync, rate limit
Three correctness bugs were making the existing broadcast flow wrong,
not just incomplete.

1. Per-recipient variables were broken
   use-broadcast-sending.ts computed templateParams per contact but shipped
   only templateParams[0] to the API, so every contact in a batch received
   contact 0's personalization (name/email/etc).

   Fix: extend POST /api/whatsapp/broadcast to accept a per-recipient
   shape `{ recipients: [{ phone, params }], ... }`. Legacy
   `{ phone_numbers, template_params }` still accepted. Hook sends per-
   recipient payloads; API returns whatsapp_message_id per phone.

2. Webhook never updated broadcast_recipients
   handleStatusUpdate in webhook/route.ts only mirrored Meta status onto
   the `messages` table. broadcast_recipients had no correlation column,
   so delivered_count / read_count / replied_count never advanced for
   real broadcasts.

   Fix (migration 003):
     - broadcast_recipients.whatsapp_message_id (+ UNIQUE index) for
       Meta-message-id correlation
     - composite index (broadcast_id, status) for fast aggregate scans
     - AFTER INSERT/UPDATE/DELETE trigger that recomputes parent
       broadcasts counts via COUNT(*) FILTER — sender + webhook only
       touch the recipient row, aggregates stay consistent

   Webhook:
     - handleStatusUpdate also updates broadcast_recipients with forward-
       only status guard (read cannot regress to sent on a replay)
     - processMessage flags the replied recipient row when an inbound
       message arrives from a recent broadcast contact

3. Rate limiting did not match spec
   Batch 50 / no delay → Batch 10 / 1s between batches, matching the spec
   and Meta's per-phone-number rate ceiling.

Also: extract statusConfig from both page components into
src/lib/broadcast-status.ts (single source of truth for PR B). Add
whatsapp_message_id to BroadcastRecipient type.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-17 11:22:38 +03:00

577 lines
17 KiB
TypeScript

import { NextResponse } from 'next/server'
import { createClient } from '@supabase/supabase-js'
import { decrypt } from '@/lib/whatsapp/encryption'
import { getMediaUrl, downloadMedia } from '@/lib/whatsapp/meta-api'
import { normalizePhone, phonesMatch } from '@/lib/whatsapp/phone-utils'
// Lazy-initialized to avoid build-time crash when env vars are missing
// eslint-disable-next-line @typescript-eslint/no-explicit-any
let _adminClient: any = null
function supabaseAdmin() {
if (!_adminClient) {
_adminClient = createClient(
process.env.NEXT_PUBLIC_SUPABASE_URL!,
process.env.SUPABASE_SERVICE_ROLE_KEY!
)
}
return _adminClient
}
interface WhatsAppMessage {
id: string
from: string
timestamp: string
type: string
text?: { body: string }
image?: { id: string; mime_type: string; caption?: string }
video?: { id: string; mime_type: string; caption?: string }
document?: { id: string; mime_type: string; filename?: string; caption?: string }
audio?: { id: string; mime_type: string }
sticker?: { id: string; mime_type: string }
location?: { latitude: number; longitude: number; name?: string; address?: string }
reaction?: { message_id: string; emoji: string }
}
interface WhatsAppWebhookEntry {
id: string
changes: Array<{
value: {
messaging_product: string
metadata: {
display_phone_number: string
phone_number_id: string
}
contacts?: Array<{
profile: { name: string }
wa_id: string
}>
messages?: WhatsAppMessage[]
statuses?: Array<{
id: string
status: string
timestamp: string
recipient_id: string
}>
}
field: string
}>
}
// GET - Webhook verification
export async function GET(request: Request) {
try {
const { searchParams } = new URL(request.url)
const mode = searchParams.get('hub.mode')
const challenge = searchParams.get('hub.challenge')
const verifyToken = searchParams.get('hub.verify_token')
if (mode !== 'subscribe' || !challenge || !verifyToken) {
return NextResponse.json(
{ error: 'Missing verification parameters' },
{ status: 400 }
)
}
// Fetch all whatsapp configs to check verify tokens
const { data: configs, error: configError } = await supabaseAdmin()
.from('whatsapp_config')
.select('id, verify_token')
if (configError || !configs) {
console.error('Error fetching configs for verification:', configError)
return NextResponse.json(
{ error: 'Verification failed' },
{ status: 403 }
)
}
// Check if any config's verify_token matches
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const matched = configs.some((config: any) => {
if (!config.verify_token) return false
try {
const decrypted = decrypt(config.verify_token)
return decrypted === verifyToken
} catch {
return false
}
})
if (matched) {
// Return challenge as plain text
return new Response(challenge, {
status: 200,
headers: { 'Content-Type': 'text/plain' },
})
}
return NextResponse.json(
{ error: 'Verification token mismatch' },
{ status: 403 }
)
} catch (error) {
console.error('Error in webhook GET verification:', error)
return NextResponse.json(
{ error: 'Internal server error' },
{ status: 500 }
)
}
}
// POST - Receive messages
export async function POST(request: Request) {
// Always return 200 immediately to acknowledge receipt
const body = await request.json()
// Process asynchronously
processWebhook(body).catch((error) => {
console.error('Error processing webhook:', error)
})
return NextResponse.json({ status: 'received' }, { status: 200 })
}
async function processWebhook(body: { entry?: WhatsAppWebhookEntry[] }) {
if (!body.entry) return
for (const entry of body.entry) {
for (const change of entry.changes) {
const value = change.value
// Handle status updates
if (value.statuses) {
for (const status of value.statuses) {
await handleStatusUpdate(status)
}
}
// Handle incoming messages
if (!value.messages || !value.contacts) continue
const phoneNumberId = value.metadata.phone_number_id
// Find user's config by phone_number_id
const { data: config, error: configError } = await supabaseAdmin()
.from('whatsapp_config')
.select('*')
.eq('phone_number_id', phoneNumberId)
.single()
if (configError || !config) {
console.error('No config found for phone_number_id:', phoneNumberId)
continue
}
const decryptedAccessToken = decrypt(config.access_token)
for (let i = 0; i < value.messages.length; i++) {
const message = value.messages[i]
const contact = value.contacts[i] || value.contacts[0]
await processMessage(
message,
contact,
config.user_id,
decryptedAccessToken
)
}
}
}
}
// Ordered forward-only: a webhook replay must never regress a recipient
// from `read` back to `sent`. Index into this array gives the "level".
const RECIPIENT_STATUS_ORDER = [
'pending',
'sent',
'delivered',
'read',
'replied',
'failed',
] as const
function recipientStatusLevel(s: string): number {
const idx = (RECIPIENT_STATUS_ORDER as readonly string[]).indexOf(s)
return idx < 0 ? 0 : idx
}
async function handleStatusUpdate(status: {
id: string
status: string
timestamp: string
recipient_id: string
}) {
// 1) Mirror onto messages (legacy behavior) — Meta's status values
// already match the CHECK constraint on messages.status.
const { error: msgErr } = await supabaseAdmin()
.from('messages')
.update({ status: status.status })
.eq('message_id', status.id)
if (msgErr) {
console.error('Error updating message status:', msgErr)
}
// 2) Mirror onto broadcast_recipients via whatsapp_message_id
// (added in migration 003). The aggregate trigger on
// broadcast_recipients re-derives the parent broadcast's
// sent/delivered/read/failed counts automatically.
const tsIso = new Date(parseInt(status.timestamp) * 1000).toISOString()
const { data: recipient, error: recFetchErr } = await supabaseAdmin()
.from('broadcast_recipients')
.select('id, status')
.eq('whatsapp_message_id', status.id)
.maybeSingle()
if (recFetchErr) {
console.error('Error fetching broadcast recipient:', recFetchErr)
return
}
if (!recipient) return // message wasn't part of a broadcast — fine
// Forward-only status guard: only advance.
const currentLevel = recipientStatusLevel(recipient.status)
const incomingLevel = recipientStatusLevel(status.status)
if (incomingLevel <= currentLevel) return
const update: Record<string, unknown> = { status: status.status }
if (status.status === 'sent' && !('sent_at' in update)) update.sent_at = tsIso
if (status.status === 'delivered') update.delivered_at = tsIso
if (status.status === 'read') update.read_at = tsIso
const { error: recUpdateErr } = await supabaseAdmin()
.from('broadcast_recipients')
.update(update)
.eq('id', recipient.id)
if (recUpdateErr) {
console.error('Error updating broadcast recipient status:', recUpdateErr)
}
}
/**
* If an inbound message's sender is on a still-unreplied
* broadcast_recipients row, flip it to `replied` so the reply count
* advances on the parent broadcast.
*
* Runs on a best-effort basis — failures here must not break the
* main inbound-message flow, so errors are swallowed with a log.
*/
async function flagBroadcastReplyIfAny(userId: string, contactId: string) {
try {
// Most recent outbound broadcast that hasn't been replied to yet.
const { data: recs, error } = await supabaseAdmin()
.from('broadcast_recipients')
.select('id, status, broadcast_id, broadcasts!inner(user_id)')
.eq('contact_id', contactId)
.eq('broadcasts.user_id', userId)
.in('status', ['sent', 'delivered', 'read'])
.order('created_at', { ascending: false })
.limit(1)
if (error || !recs || recs.length === 0) return
const row = recs[0]
const { error: updErr } = await supabaseAdmin()
.from('broadcast_recipients')
.update({ status: 'replied', replied_at: new Date().toISOString() })
.eq('id', row.id)
if (updErr) {
console.error('Error marking broadcast recipient replied:', updErr)
}
} catch (err) {
console.error('flagBroadcastReplyIfAny failed:', err)
}
}
async function processMessage(
message: WhatsAppMessage,
contact: { profile: { name: string }; wa_id: string },
userId: string,
accessToken: string
) {
const senderPhone = normalizePhone(message.from)
const contactName = contact.profile.name
// Parse message content based on type
const { contentText, mediaUrl, mediaType } = await parseMessageContent(
message,
accessToken
)
// Find or create contact
const contactRecord = await findOrCreateContact(
userId,
senderPhone,
contactName
)
if (!contactRecord) return
// Find or create conversation
const conversation = await findOrCreateConversation(
userId,
contactRecord.id
)
if (!conversation) return
// Insert message — field names MUST match the messages table schema
// (see supabase/migrations/001_initial_schema.sql):
// conversation_id, sender_type, content_type, content_text,
// media_url, template_name, message_id, status, created_at
// `mediaType` is intentionally unused — the schema has no media_type
// column; the MIME type is only used to construct the proxy URL during
// parseMessageContent. Silence the unused-var warning:
void mediaType
// The messages.content_type CHECK constraint only allows:
// text, image, document, audio, video, location, template
// Map incoming WhatsApp types that aren't in that list to the closest
// allowed value so the INSERT doesn't fail with a constraint error.
const ALLOWED_CONTENT_TYPES = new Set([
'text', 'image', 'document', 'audio', 'video', 'location', 'template',
])
const contentType = ALLOWED_CONTENT_TYPES.has(message.type)
? message.type
: message.type === 'sticker'
? 'image' // stickers are images
: 'text' // reaction, unknown → text fallback
const { error: msgError } = await supabaseAdmin().from('messages').insert({
conversation_id: conversation.id,
sender_type: 'customer',
content_type: contentType,
content_text: contentText,
media_url: mediaUrl,
message_id: message.id,
status: 'delivered',
created_at: new Date(parseInt(message.timestamp) * 1000).toISOString(),
})
if (msgError) {
console.error('Error inserting message:', msgError)
return
}
// Update conversation
const { error: convError } = await supabaseAdmin()
.from('conversations')
.update({
last_message_text: contentText || `[${message.type}]`,
last_message_at: new Date().toISOString(),
unread_count: (conversation.unread_count || 0) + 1,
updated_at: new Date().toISOString(),
})
.eq('id', conversation.id)
if (convError) {
console.error('Error updating conversation:', convError)
}
// If this contact was a recent broadcast recipient, flag the reply
// so the broadcast's `replied_count` advances (via the aggregate
// trigger installed in migration 003).
await flagBroadcastReplyIfAny(userId, contactRecord.id)
}
async function parseMessageContent(
message: WhatsAppMessage,
accessToken: string
): Promise<{
contentText: string | null
mediaUrl: string | null
mediaType: string | null
}> {
// getMediaUrl signature is (mediaId, accessToken) — earlier code had
// the args swapped, so every verification hit an invalid Meta URL and
// fell through to the catch block, leaving mediaUrl as null. That's
// why images showed up as empty bubbles in the inbox.
const verifyAndBuildUrl = async (
mediaId: string
): Promise<string | null> => {
try {
await getMediaUrl({ mediaId, accessToken })
return `/api/whatsapp/media/${mediaId}`
} catch (error) {
console.error(
`Failed to verify media ${mediaId} with Meta:`,
error instanceof Error ? error.message : error
)
return null
}
}
switch (message.type) {
case 'text':
return {
contentText: message.text?.body || null,
mediaUrl: null,
mediaType: null,
}
case 'image':
if (message.image?.id) {
return {
contentText: message.image.caption || null,
mediaUrl: await verifyAndBuildUrl(message.image.id),
mediaType: message.image.mime_type,
}
}
return { contentText: null, mediaUrl: null, mediaType: null }
case 'video':
if (message.video?.id) {
return {
contentText: message.video.caption || null,
mediaUrl: await verifyAndBuildUrl(message.video.id),
mediaType: message.video.mime_type,
}
}
return { contentText: null, mediaUrl: null, mediaType: null }
case 'document':
if (message.document?.id) {
return {
contentText:
message.document.caption || message.document.filename || null,
mediaUrl: await verifyAndBuildUrl(message.document.id),
mediaType: message.document.mime_type,
}
}
return { contentText: null, mediaUrl: null, mediaType: null }
case 'audio':
if (message.audio?.id) {
return {
contentText: null,
mediaUrl: await verifyAndBuildUrl(message.audio.id),
mediaType: message.audio.mime_type,
}
}
return { contentText: null, mediaUrl: null, mediaType: null }
case 'sticker':
// Stickers are images under the hood. Treat them as such so the
// MessageBubble renders the <img>. The caller maps the DB
// content_type to 'image' for the CHECK constraint.
if (message.sticker?.id) {
return {
contentText: null,
mediaUrl: await verifyAndBuildUrl(message.sticker.id),
mediaType: message.sticker.mime_type,
}
}
return { contentText: null, mediaUrl: null, mediaType: null }
case 'location':
if (message.location) {
const loc = message.location
const locationText = [loc.name, loc.address, `${loc.latitude},${loc.longitude}`]
.filter(Boolean)
.join(' - ')
return {
contentText: locationText,
mediaUrl: null,
mediaType: null,
}
}
return { contentText: null, mediaUrl: null, mediaType: null }
case 'reaction':
return {
contentText: message.reaction?.emoji || null,
mediaUrl: null,
mediaType: null,
}
default:
return {
contentText: `[Unsupported message type: ${message.type}]`,
mediaUrl: null,
mediaType: null,
}
}
}
async function findOrCreateContact(
userId: string,
phone: string,
name: string
) {
// Look up existing contacts for this user
const { data: contacts, error: contactsError } = await supabaseAdmin()
.from('contacts')
.select('*')
.eq('user_id', userId)
if (contactsError) {
console.error('Error fetching contacts:', contactsError)
return null
}
// Use phonesMatch for flexible matching
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const existingContact = contacts?.find((c: any) => phonesMatch(c.phone, phone))
if (existingContact) {
// Update name if it changed
if (name && name !== existingContact.name) {
await supabaseAdmin()
.from('contacts')
.update({ name, updated_at: new Date().toISOString() })
.eq('id', existingContact.id)
}
return existingContact
}
// Create new contact
const { data: newContact, error: createError } = await supabaseAdmin()
.from('contacts')
.insert({
user_id: userId,
phone,
name: name || phone,
})
.select()
.single()
if (createError) {
console.error('Error creating contact:', createError)
return null
}
return newContact
}
async function findOrCreateConversation(userId: string, contactId: string) {
// Look for existing conversation
const { data: existing, error: findError } = await supabaseAdmin()
.from('conversations')
.select('*')
.eq('user_id', userId)
.eq('contact_id', contactId)
.single()
if (!findError && existing) {
return existing
}
// Create new conversation
const { data: newConv, error: createError } = await supabaseAdmin()
.from('conversations')
.insert({
user_id: userId,
contact_id: contactId,
})
.select()
.single()
if (createError) {
console.error('Error creating conversation:', createError)
return null
}
return newConv
}