CognxSafeTrack
feat: implement whatsapp templates management with security hardening, audit logs, and crm integration
0f2f80a | import { FastifyInstance } from 'fastify'; | |
| import { whatsappService } from '../services/whatsapp'; | |
| import { z } from 'zod'; | |
| import { getOrganizationByPhoneNumberId } from '../services/organization'; | |
| import { logger } from '../logger'; | |
| import { createBullBoard } from '@bull-board/api'; | |
| import { BullMQAdapter } from '@bull-board/api/bullMQAdapter'; | |
| import { FastifyAdapter } from '@bull-board/fastify'; | |
| import { whatsappQueue, notificationQueue } from '../services/queue'; | |
| const WebhookPayloadSchema = z.object({ | |
| object: z.literal('whatsapp_business_account'), | |
| entry: z.array(z.object({ | |
| id: z.string(), | |
| changes: z.array(z.object({ | |
| value: z.object({ | |
| metadata: z.object({ phone_number_id: z.string() }).optional(), | |
| messages: z.array(z.any()).optional(), | |
| }), | |
| field: z.string(), | |
| })), | |
| })), | |
| }); | |
| export async function internalRoutes(fastify: FastifyInstance) { | |
| // ── BullBoard Dashboard ────────────────────────────────────────────────── | |
| const serverAdapter = new FastifyAdapter(); | |
| createBullBoard({ | |
| queues: [ | |
| new BullMQAdapter(whatsappQueue), | |
| new BullMQAdapter(notificationQueue) | |
| ], | |
| serverAdapter, | |
| }); | |
| // Re-enabled BullBoard for production monitoring | |
| serverAdapter.setBasePath('/v1/internal/queues'); | |
| fastify.register(async (instance) => { | |
| instance.register(serverAdapter.registerPlugin(), { | |
| prefix: '/', | |
| }); | |
| }, { | |
| prefix: '/v1/internal/queues', | |
| config: { requireAuth: true } | |
| }); | |
| // ── Handle Webhook Forwarding from Gateway (HF -> Railway) ─────────────── | |
| fastify.post('/v1/internal/whatsapp/inbound', { | |
| config: { requireAuth: true } | |
| }, async (request, reply) => { | |
| try { | |
| const parsed = WebhookPayloadSchema.safeParse(request.body); | |
| if (!parsed.success) { | |
| return reply.code(400).send({ error: 'Invalid Payload', details: parsed.error.flatten() }); | |
| } | |
| for (const entry of parsed.data.entry) { | |
| for (const change of entry.changes) { | |
| const phoneNumberId = change.value.metadata?.phone_number_id || 'unknown'; | |
| const organizationId = await getOrganizationByPhoneNumberId(phoneNumberId); | |
| if (!organizationId) { | |
| logger.warn(`[INTERNAL-WEBHOOK] Unknown phone_number_id ${phoneNumberId} — skipping`); | |
| continue; | |
| } | |
| for (const message of change.value.messages || []) { | |
| const phone = message.from; | |
| let text = ''; | |
| let audioUrl: string | undefined; | |
| let imageUrl: string | undefined; | |
| if (message.type === 'text') text = message.text?.body; | |
| else if (message.type === 'interactive') { | |
| text = message.interactive?.button_reply?.id || message.interactive?.list_reply?.id; | |
| } else if (message.type === 'audio') { | |
| audioUrl = message.audio?.id; | |
| text = ''; // Trigger transcription | |
| } else if (message.type === 'image') { | |
| imageUrl = message.image?.id; | |
| text = ''; // Trigger analysis | |
| } | |
| if (phone) { | |
| await whatsappQueue.add('handle-inbound', { | |
| phone, | |
| text, | |
| audioUrl, | |
| imageUrl, | |
| organizationId | |
| }, { | |
| attempts: 3, | |
| backoff: { type: 'exponential', delay: 1000 }, | |
| removeOnComplete: true | |
| }); | |
| } | |
| } | |
| } | |
| } | |
| return reply.code(200).send({ ok: true, status: 'queued' }); | |
| } catch (error) { | |
| logger.error(`[INTERNAL-WEBHOOK] Enqueue error: ${error}`); | |
| return reply.code(500).send({ error: 'Failed to enqueue messages' }); | |
| } | |
| }); | |
| // ── Handle standard transcribed messages from worker (Railway) ─────────── | |
| fastify.post<{ | |
| Body: { phone: string; text: string; audioUrl?: string; imageUrl?: string; organizationId?: string } | |
| }>('/v1/internal/handle-message', { | |
| config: { requireAuth: true } | |
| }, async (request, reply) => { | |
| const { phone, text, audioUrl, imageUrl, organizationId } = request.body; | |
| try { | |
| await whatsappService.handleIncomingMessage(phone, text, audioUrl, imageUrl, undefined, organizationId); | |
| return reply.send({ ok: true }); | |
| } catch (err: unknown) { | |
| return reply.code(500).send({ error: err instanceof Error ? err.message : String(err) }); | |
| } | |
| }); | |
| fastify.get('/v1/internal/ping', { config: { requireAuth: true } }, async () => ({ ok: true })); | |
| } | |