eks_relay/src/http/routes/webhooks.ts

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);
  }
});