eks_relay/tests/integration/pipeline.test.ts

import { after, before, beforeEach, test } from 'node:test';
import assert from 'node:assert/strict';
import { prepareTestDatabase } from '../helpers/testServer.js';
import { applyTestConfig } from '../helpers/testConfig.js';
import type { SupportMessageEvent } from '../../src/domain/messageNormalizer.js';
import type { ChatwootClient } from '../../src/clients/chatwootClient.js';
import type { FlowiseClient, FlowiseResult } from '../../src/clients/flowiseClient.js';

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

const { processMessageEvent, buildFlowisePayload } = await import(
  '../../src/domain/conversationPipeline.js'
);
const { db, disconnectDb } = await import('../../src/store/db.js');
const { claimMessage } = await import('../../src/store/idempotencyStore.js');

/** Records every Chatwoot side effect instead of performing it. */
class FakeChatwoot {
  sentMessages: { conversationId: number; content: string }[] = [];
  labels: { conversationId: number; label: string }[] = [];
  attributes: { conversationId: number; attrs: Record<string, unknown> }[] = [];
  unassigned: number[] = [];
  statusUpdates: { conversationId: number; status: string }[] = [];
  conversation: Record<string, unknown> = { id: 1311, labels: [], custom_attributes: {} };

  async getConversation(): Promise<Record<string, unknown>> {
    return this.conversation;
  }
  async addLabel(conversationId: number, label: string): Promise<void> {
    this.labels.push({ conversationId, label });
  }
  async setCustomAttributes(conversationId: number, attrs: Record<string, unknown>): Promise<void> {
    this.attributes.push({ conversationId, attrs });
    const existing = (this.conversation.custom_attributes ?? {}) as Record<string, unknown>;
    this.conversation.custom_attributes = { ...existing, ...attrs };
  }
  async unassignConversation(conversationId: number): Promise<boolean> {
    this.unassigned.push(conversationId);
    return true;
  }
  async setConversationStatus(conversationId: number, status: 'open' | 'resolved' | 'pending'): Promise<boolean> {
    this.statusUpdates.push({ conversationId, status });
    return true;
  }
  async sendOutgoingMessage(conversationId: number, content: string): Promise<unknown> {
    this.sentMessages.push({ conversationId, content });
    return {};
  }
}

class FakeFlowise {
  calls: unknown[] = [];
  constructor(private readonly result: FlowiseResult) {}
  async predict(payload: unknown): Promise<FlowiseResult> {
    this.calls.push(payload);
    return this.result;
  }
}

function evt(overrides: Partial<SupportMessageEvent> = {}): SupportMessageEvent {
  return {
    source: 'chatwoot',
    channel: 'email',
    conversationId: 1311,
    messageId: `m-${Math.random().toString(36).slice(2)}`,
    inboxId: 1,
    content: 'Czy macie gaz R134a?',
    subject: 'Pytanie',
    senderEmail: 'klient@example.com',
    senderName: 'Klient Testowy',
    senderId: '332',
    labels: [],
    customAttributes: {},
    additionalAttributes: {},
    attachmentCount: 0,
    needsConversationFetch: false,
    ...overrides,
  };
}

const deps = (chatwoot: FakeChatwoot, flowise: FakeFlowise) => ({
  chatwoot: chatwoot as unknown as ChatwootClient,
  flowise: flowise as unknown as FlowiseClient,
});

before(() => applyTestConfig({ DATABASE_URL: dbUrl }));
beforeEach(async () => {
  await db().ticket.deleteMany({});
  await db().auditEvent.deleteMany({});
  await db().processedMessage.deleteMany({});
});
after(async () => {
  await disconnectDb();
});

test('a normal question is answered and marked replied', async () => {
  const chatwoot = new FakeChatwoot();
  const flowise = new FakeFlowise({ type: 'reply', text: 'Tak, mamy.', actions: null });
  const event = evt();
  await claimMessage(event.source, event.messageId, event.conversationId);

  const outcome = await processMessageEvent(event, deps(chatwoot, flowise));

  assert.equal(outcome.action, 'reply');
  assert.equal(chatwoot.sentMessages.length, 1);
  assert.equal(chatwoot.sentMessages[0]?.content, 'Tak, mamy.');

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

test('a conversation already in ticket mode never reaches Flowise', async () => {
  const chatwoot = new FakeChatwoot();
  const flowise = new FakeFlowise({ type: 'reply', text: 'nie powinno wyjść', actions: null });
  const event = evt({ labels: ['ticket'] });
  await claimMessage(event.source, event.messageId, event.conversationId);

  const outcome = await processMessageEvent(event, deps(chatwoot, flowise));

  assert.deepEqual(outcome, { action: 'skipped', reason: 'ticket_mode' });
  assert.equal(flowise.calls.length, 0);
  assert.equal(chatwoot.sentMessages.length, 0);
});

test('a ticket_number custom attribute also stops the AI', async () => {
  const chatwoot = new FakeChatwoot();
  const flowise = new FakeFlowise({ type: 'reply', text: 'x', actions: null });
  const event = evt({ customAttributes: { ticket_number: 'EKS-20260820-1311' } });
  await claimMessage(event.source, event.messageId, event.conversationId);

  const outcome = await processMessageEvent(event, deps(chatwoot, flowise));
  assert.equal(outcome.action, 'skipped');
  assert.equal(flowise.calls.length, 0);
});

test('spam is blocked before Flowise and recorded', async () => {
  const chatwoot = new FakeChatwoot();
  const flowise = new FakeFlowise({ type: 'reply', text: 'x', actions: null });
  const event = evt({ senderEmail: 'mailer-daemon@example.com' });
  await claimMessage(event.source, event.messageId, event.conversationId);

  const outcome = await processMessageEvent(event, deps(chatwoot, flowise));

  assert.equal(outcome.action, 'spam');
  assert.equal(flowise.calls.length, 0, 'the spam gate must sit in front of the LLM');
  const row = await db().processedMessage.findUnique({
    where: { source_messageId: { source: 'chatwoot', messageId: event.messageId } },
  });
  assert.equal(row?.status, 'spam');
});

test('configured AgentBot guard skips conversations assigned to the legacy user account', async () => {
  applyTestConfig({
    DATABASE_URL: dbUrl,
    CHATWOOT_REQUIRE_AGENT_BOT_ASSIGNMENT: 'true',
    CHATWOOT_REQUIRED_AGENT_BOT_ID: '1',
  });
  const chatwoot = new FakeChatwoot();
  chatwoot.conversation = {
    id: 1311,
    labels: [],
    custom_attributes: {},
    assignee_agent_bot_id: null,
    meta: { assignee_type: 'User', assignee: { id: 2, name: 'KlimBot' } },
  };
  const flowise = new FakeFlowise({ type: 'reply', text: 'nie powinno wyjść', actions: null });
  const event = evt();
  await claimMessage(event.source, event.messageId, event.conversationId);

  const outcome = await processMessageEvent(event, deps(chatwoot, flowise));

  assert.deepEqual(outcome, { action: 'skipped', reason: 'not_assigned_to_bot' });
  assert.equal(flowise.calls.length, 0);
  assert.equal(chatwoot.sentMessages.length, 0);
  applyTestConfig({ DATABASE_URL: dbUrl });
});

test('configured AgentBot guard allows conversations assigned to the Chatwoot AgentBot', async () => {
  applyTestConfig({
    DATABASE_URL: dbUrl,
    CHATWOOT_REQUIRE_AGENT_BOT_ASSIGNMENT: 'true',
    CHATWOOT_REQUIRED_AGENT_BOT_ID: '1',
  });
  const chatwoot = new FakeChatwoot();
  chatwoot.conversation = {
    id: 1311,
    labels: [],
    custom_attributes: {},
    assignee_agent_bot_id: 1,
    meta: { assignee_type: 'AgentBot', assignee: { id: 1, name: 'Agent' } },
  };
  const flowise = new FakeFlowise({ type: 'reply', text: 'OK od bota', actions: null });
  const event = evt();
  await claimMessage(event.source, event.messageId, event.conversationId);

  const outcome = await processMessageEvent(event, deps(chatwoot, flowise));

  assert.equal(outcome.action, 'reply');
  assert.equal(flowise.calls.length, 1);
  assert.equal(chatwoot.sentMessages[0]?.content, 'OK od bota');
  applyTestConfig({ DATABASE_URL: dbUrl });
});

test('a handoff action creates a ticket, labels it and replies with the number', async () => {
  const chatwoot = new FakeChatwoot();
  const flowise = new FakeFlowise({
    type: 'handoff',
    text: 'Przekazuję sprawę do supportu.',
    actions: [{ type: 'handoff' }],
  });
  const event = evt();
  await claimMessage(event.source, event.messageId, event.conversationId);

  const outcome = await processMessageEvent(event, deps(chatwoot, flowise));

  assert.equal(outcome.action, 'handoff');
  assert.match((outcome as { ticketNumber: string }).ticketNumber, /^EKS-\d{8}-1311$/);
  assert.ok(chatwoot.labels.some((l) => l.label === 'ticket'));
  assert.ok(chatwoot.attributes.some((a) => 'ticket_number' in a.attrs));
  assert.deepEqual(chatwoot.unassigned, [1311]);
  assert.deepEqual(chatwoot.statusUpdates, [{ conversationId: 1311, status: 'open' }]);
  assert.equal(chatwoot.sentMessages.length, 1);
  assert.match(chatwoot.sentMessages[0]?.content ?? '', /EKS-\d{8}-1311/);
});

test('two handoffs for one conversation reuse the same ticket number', async () => {
  const chatwoot = new FakeChatwoot();
  const first = evt({ messageId: 'h1' });
  const second = evt({ messageId: 'h2' });
  await claimMessage(first.source, first.messageId, first.conversationId);
  await claimMessage(second.source, second.messageId, second.conversationId);

  const flowise = new FakeFlowise({ type: 'handoff', text: 'ok', actions: [{ type: 'handoff' }] });
  const a = await processMessageEvent(first, deps(chatwoot, flowise));

  // The second message would normally be stopped by the ticket-mode check; call
  // createTicket directly to prove the idempotency of the ticket itself.
  const { createTicket } = await import('../../src/domain/ticketService.js');
  const b = await createTicket(1311, 'retry', chatwoot as unknown as ChatwootClient);

  assert.equal((a as { ticketNumber: string }).ticketNumber, b.ticketNumber);
  assert.equal(b.status, 'existing');
  assert.equal(await db().ticket.count({ where: { conversationId: 1311 } }), 1);
});

test('an empty Flowise answer is recorded as failed and nothing is sent', async () => {
  const chatwoot = new FakeChatwoot();
  const flowise = new FakeFlowise({ type: 'unknown', text: null, actions: null });
  const event = evt();
  await claimMessage(event.source, event.messageId, event.conversationId);

  const outcome = await processMessageEvent(event, deps(chatwoot, flowise));

  assert.equal(outcome.action, 'no_reply');
  assert.equal(chatwoot.sentMessages.length, 0);
  const row = await db().processedMessage.findUnique({
    where: { source_messageId: { source: 'chatwoot', messageId: event.messageId } },
  });
  assert.equal(row?.status, 'failed');
});

test('the conversation is fetched when the webhook omitted labels', async () => {
  const chatwoot = new FakeChatwoot();
  chatwoot.conversation = { id: 1311, labels: ['ticket'], custom_attributes: {} };
  const flowise = new FakeFlowise({ type: 'reply', text: 'x', actions: null });
  const event = evt({ needsConversationFetch: true, labels: [] });
  await claimMessage(event.source, event.messageId, event.conversationId);

  const outcome = await processMessageEvent(event, deps(chatwoot, flowise));

  assert.equal(outcome.action, 'skipped');
  assert.equal(flowise.calls.length, 0);
});

test('the Flowise payload keeps the contract the custom tools depend on', () => {
  const payload = buildFlowisePayload(evt());

  assert.match(payload.question, /\[CONTACT_INFO\]/);
  assert.match(payload.question, /contact_email: klient@example\.com/);
  assert.match(payload.question, /Czy macie gaz R134a\?/);
  assert.equal(payload.overrideConfig.sessionId, 'chatwoot:1311');
  assert.equal(payload.overrideConfig.conversationId, 1311);
  assert.equal(payload.overrideConfig.inboxId, 1);
  assert.equal(payload.overrideConfig.tenantId, 'easyklima');
  assert.equal(payload.overrideConfig.channel, 'email');
  assert.equal(payload.metadata.source, 'chatwoot');
  assert.equal(payload.metadata.messageType, 'incoming');
});