eks_relay/tests/integration/webhook.test.ts

import { after, before, test } from 'node:test';
import assert from 'node:assert/strict';
import { prepareTestDatabase, startServer, type RunningServer } from '../helpers/testServer.js';
import { applyTestConfig } from '../helpers/testConfig.js';
import {
  incomingEmailMessage,
  noLabelsPayload,
  outgoingMessage,
  privateNote,
  statusUpdate,
} from '../fixtures/chatwoot.js';

const dbUrl = prepareTestDatabase('test-webhook');

// Imported after the DATABASE_URL is in place.
const { createApp } = await import('../../src/http/app.js');
const { db, disconnectDb } = await import('../../src/store/db.js');

let running: RunningServer;

before(async () => {
  applyTestConfig({ DATABASE_URL: dbUrl });
  running = await startServer(createApp());
});

after(async () => {
  await running.close();
  await disconnectDb();
});

async function post(path: string, body: unknown, headers: Record<string, string> = {}) {
  const res = await fetch(`${running.baseUrl}${path}`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', ...headers },
    body: JSON.stringify(body),
  });
  return { status: res.status, body: (await res.json()) as Record<string, unknown> };
}

test('GET /health returns 200 without touching any dependency', async () => {
  const res = await fetch(`${running.baseUrl}/health`);
  assert.equal(res.status, 200);
  const body = (await res.json()) as Record<string, unknown>;
  assert.equal(body.ok, true);
  assert.equal(body.service, 'eks-relay');
});

test('an incoming message is accepted with 202 and recorded', async () => {
  const res = await post('/webhooks/chatwoot', incomingEmailMessage);
  assert.equal(res.status, 202);
  assert.equal(res.body.accepted, true);
  assert.equal(res.body.conversationId, 1311);
  assert.ok(res.body.jobId);

  const row = await db().processedMessage.findUnique({
    where: { source_messageId: { source: 'chatwoot', messageId: '90210' } },
  });
  assert.ok(row, 'ProcessedMessage row should exist');
  assert.equal(row.conversationId, 1311);
  assert.equal(row.status, 'queued');

  const jobs = await db().job.findMany({ where: { type: 'chatwoot_message' } });
  assert.equal(jobs.length, 1);
});

test('the same message id a second time is a no-op duplicate', async () => {
  const res = await post('/webhooks/chatwoot', incomingEmailMessage);
  assert.equal(res.status, 200);
  assert.equal(res.body.duplicate, true);

  const jobs = await db().job.findMany({ where: { type: 'chatwoot_message' } });
  assert.equal(jobs.length, 1, 'a duplicate must not create a second job');

  const rows = await db().processedMessage.findMany({ where: { messageId: '90210' } });
  assert.equal(rows.length, 1);
});

test('concurrent deliveries of the same message create exactly one job', async () => {
  const payload = { ...incomingEmailMessage, id: 99001 };
  const results = await Promise.all([
    post('/webhooks/chatwoot', payload),
    post('/webhooks/chatwoot', payload),
    post('/webhooks/chatwoot', payload),
  ]);
  const accepted = results.filter((r) => r.status === 202);
  assert.equal(accepted.length, 1, 'exactly one delivery may be accepted');

  const jobs = await db().job.findMany({});
  const forMessage = jobs.filter((j) => j.payloadJson.includes('"messageId":"99001"'));
  assert.equal(forMessage.length, 1);
});

test('the webhook stores an audit event', async () => {
  const events = await db().auditEvent.findMany({ where: { messageId: '90210' } });
  assert.ok(events.length >= 1);
  assert.equal(events[0]?.eventType, 'webhook_accepted');
});

test('outgoing messages, private notes and status events are skipped with 200', async () => {
  for (const payload of [outgoingMessage, privateNote, statusUpdate]) {
    const res = await post('/webhooks/chatwoot', payload);
    assert.equal(res.status, 200);
    assert.equal(res.body.skipped, true);
  }
  const skippedRows = await db().processedMessage.findMany({
    where: { messageId: { in: ['90211', '90212', '90213'] } },
  });
  assert.equal(skippedRows.length, 0, 'skipped events must not be recorded as processed');
});

test('a payload without labels is still accepted and flagged for a fetch', async () => {
  const res = await post('/webhooks/chatwoot', noLabelsPayload);
  assert.equal(res.status, 202);

  const job = await db().job.findFirst({
    where: { payloadJson: { contains: '"messageId":"90217"' } },
  });
  assert.ok(job);
  const payload = JSON.parse(job.payloadJson) as { needsConversationFetch: boolean };
  assert.equal(payload.needsConversationFetch, true);
});

test('garbage bodies are answered 200 without creating work', async () => {
  const before = await db().job.count();
  const res = await post('/webhooks/chatwoot', { totally: 'unrelated' });
  assert.equal(res.status, 200);
  assert.equal(res.body.skipped, true);
  assert.equal(await db().job.count(), before);
});

test('an already-ticketed conversation is settled at the webhook without queueing', async () => {
  const jobsBefore = await db().job.count();

  const res = await post('/webhooks/chatwoot', {
    ...incomingEmailMessage,
    id: 99500,
    conversation: {
      ...incomingEmailMessage.conversation,
      labels: ['ticket'],
      custom_attributes: { ticket_number: 'EKS-20260820-1311', handoff: true },
    },
  });

  assert.equal(res.status, 200);
  assert.equal(res.body.skipped, true);
  assert.equal(res.body.reason, 'ticket_mode');

  assert.equal(await db().job.count(), jobsBefore, 'no job may be queued for a ticketed conversation');

  const row = await db().processedMessage.findUnique({
    where: { source_messageId: { source: 'chatwoot', messageId: '99500' } },
  });
  assert.equal(row?.status, 'skipped');
  assert.equal(row?.reason, 'ticket_mode');

  const events = await db().auditEvent.findMany({ where: { messageId: '99500' } });
  assert.equal(events[0]?.eventType, 'skipped_ticket_mode');
  assert.match(events[0]?.metaJson ?? '', /"stage":"webhook"/);
});

test('a ticketed conversation is still deduplicated on redelivery', async () => {
  const payload = {
    ...incomingEmailMessage,
    id: 99501,
    conversation: {
      ...incomingEmailMessage.conversation,
      labels: ['ticket'],
      custom_attributes: {},
    },
  };
  const first = await post('/webhooks/chatwoot', payload);
  assert.equal(first.body.reason, 'ticket_mode');

  const second = await post('/webhooks/chatwoot', payload);
  assert.equal(second.status, 200);
  assert.equal(second.body.duplicate, true);

  const rows = await db().processedMessage.findMany({ where: { messageId: '99501' } });
  assert.equal(rows.length, 1);
});

test('a payload lacking labels is still queued, so the worker can fetch them', async () => {
  const res = await post('/webhooks/chatwoot', { ...noLabelsPayload, id: 99502 });
  assert.equal(res.status, 202, 'the ticket state is unknown here — it must not be guessed');
  assert.ok(res.body.jobId);
});

test('the queued audit event records the jobId in its own column', async () => {
  const res = await post('/webhooks/chatwoot', { ...incomingEmailMessage, id: 99503 });
  assert.equal(res.status, 202);

  const event = await db().auditEvent.findFirst({
    where: { messageId: '99503', eventType: 'webhook_accepted' },
  });
  assert.equal(event?.jobId, res.body.jobId);
});