import { Router } from 'express'; import { chatwootWebhookSchema } from '../../types/chatwoot.js'; import { normalizeChatwootWebhook } from '../../domain/messageNormalizer.js'; import { isTicketMode } from '../../domain/ticketService.js'; import { claimMessage, setMessageStatus } from '../../store/idempotencyStore.js'; import { enqueue } from '../../queue/jobQueue.js'; import { audit } from '../../store/auditLog.js'; import { logger } from '../../logger.js'; import { config } from '../../config.js'; import { drainOnce } from '../../queue/worker.js'; import { evaluateRuntimeSkip } from '../../store/runtimeSettings.js'; export const webhookRouter = Router(); /** * Chatwoot `message_created` entry point. * * Contract: acknowledge fast (202) and never let a slow Flowise call cause a * Chatwoot-side timeout and retry. Anything that is not an actionable incoming * customer message is answered 200 with an explicit skip reason, so Chatwoot * does not keep retrying it. */ webhookRouter.post('/webhooks/chatwoot', async (req, res, next) => { try { const parsed = chatwootWebhookSchema.safeParse(req.body); if (!parsed.success) { logger.warn('Malformed Chatwoot webhook payload', { issues: parsed.error.issues.map((i) => i.path.join('.')), }); res.status(200).json({ ok: true, skipped: true, reason: 'invalid_payload' }); return; } const normalized = normalizeChatwootWebhook(parsed.data); if (!normalized.ok) { logger.info('Webhook ignored', { reason: normalized.reason }); res.status(200).json({ ok: true, skipped: true, reason: normalized.reason }); return; } const event = normalized.event; const runtime = await evaluateRuntimeSkip(event.messageCreatedAt ? new Date(event.messageCreatedAt) : null); if (runtime.decision.skip) { await setMessageStatus(event.source, event.messageId, 'skipped', runtime.decision.reason).catch(async () => { await claimMessage(event.source, event.messageId, event.conversationId); await setMessageStatus(event.source, event.messageId, 'skipped', runtime.decision.skip ? runtime.decision.reason : 'runtime_skip'); }); await audit({ conversationId: event.conversationId, messageId: event.messageId, eventType: `skipped_${runtime.decision.reason}`, summary: `Message not queued: ${runtime.decision.reason}`, meta: { stage: 'webhook', messageCreatedAt: event.messageCreatedAt, maxMessageAgeMinutes: runtime.settings.maxMessageAgeMinutes, ignoreMessagesBefore: runtime.settings.ignoreMessagesBefore?.toISOString() ?? null, }, }); res.status(200).json({ ok: true, skipped: true, reason: runtime.decision.reason, conversationId: event.conversationId }); return; } // Idempotency: the unique (source, messageId) insert decides the winner of // a retry race before any job is created. const claim = await claimMessage(event.source, event.messageId, event.conversationId); if (!claim.claimed) { logger.info('Duplicate message ignored', { conversationId: event.conversationId, messageId: event.messageId, previousStatus: claim.status, }); res.status(200).json({ ok: true, duplicate: true, status: claim.status, conversationId: event.conversationId, }); return; } // Fast path: when the payload already carries labels and custom attributes, // a handed-off conversation can be settled here — same data the worker // would have used, minus a queue round-trip. Conversations that need a // Chatwoot fetch are still decided in the worker. if (!event.needsConversationFetch) { const ticketed = isTicketMode({ labels: event.labels, custom_attributes: event.customAttributes, }); if (ticketed) { await setMessageStatus(event.source, event.messageId, 'skipped', 'ticket_mode'); await audit({ conversationId: event.conversationId, messageId: event.messageId, eventType: 'skipped_ticket_mode', summary: 'Conversation is in manual/ticket mode — Flowise not called (settled at webhook)', meta: { stage: 'webhook', channel: event.channel, inboxId: event.inboxId }, }); logger.info('Webhook settled without queueing: ticket mode', { conversationId: event.conversationId, messageId: event.messageId, }); res.status(200).json({ ok: true, skipped: true, reason: 'ticket_mode', conversationId: event.conversationId, }); return; } } const jobId = await enqueue({ type: 'chatwoot_message', payload: { ...event } }); await audit({ conversationId: event.conversationId, messageId: event.messageId, jobId, eventType: 'webhook_accepted', summary: `Queued job ${jobId} for conversation ${event.conversationId}`, meta: { channel: event.channel, inboxId: event.inboxId, attachments: event.attachmentCount }, }); res.status(202).json({ ok: true, accepted: true, jobId, conversationId: event.conversationId, messageId: event.messageId, }); // With the background worker disabled (one-shot runs) still make progress // once the response has been flushed. Tests drive the queue explicitly. const cfg = config(); if (!cfg.WORKER_ENABLED && cfg.NODE_ENV !== 'test') { void drainOnce().catch((err: unknown) => { logger.error('Inline drain failed', { error: err instanceof Error ? err.message : String(err), }); }); } } catch (err) { next(err); } });