eks_relay/tests/integration/jobQueue.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';

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

const { enqueue, claimNextJob, completeJob, failJob, queueStats } = await import(
  '../../src/queue/jobQueue.js'
);
const { db, disconnectDb } = await import('../../src/store/db.js');

before(() => applyTestConfig({ DATABASE_URL: dbUrl, WORKER_MAX_ATTEMPTS: '2' }));
beforeEach(async () => {
  await db().job.deleteMany({});
});
after(async () => {
  await disconnectDb();
});

test('an enqueued job can be claimed exactly once', async () => {
  await enqueue({ type: 'chatwoot_message', payload: { conversationId: 1 } });

  const first = await claimNextJob();
  assert.ok(first);
  assert.equal(first.status, 'processing');
  assert.equal(first.attempts, 1);

  const second = await claimNextJob();
  assert.equal(second, null, 'a claimed job must not be handed out again');
});

test('completing a job clears the error and marks it done', async () => {
  const id = await enqueue({ type: 'chatwoot_message', payload: {} });
  await claimNextJob();
  await completeJob(id);

  const job = await db().job.findUnique({ where: { id } });
  assert.equal(job?.status, 'done');
  assert.equal(job?.lastError, null);
  assert.ok(job?.finishedAt);
});

test('a failure below the attempt limit is re-queued with a future runAfter', async () => {
  const id = await enqueue({ type: 'chatwoot_message', payload: {} });
  await claimNextJob();

  const disposition = await failJob(id, 'Flowise returned HTTP 502');
  assert.equal(disposition, 'retry');

  const job = await db().job.findUnique({ where: { id } });
  assert.equal(job?.status, 'queued');
  assert.ok(job && job.runAfter.getTime() > Date.now(), 'backoff must delay the retry');
  assert.match(job?.lastError ?? '', /502/);

  const claimed = await claimNextJob();
  assert.equal(claimed, null, 'a backed-off job is not runnable yet');
});

test('exhausting the attempt limit parks the job as dead', async () => {
  const id = await enqueue({ type: 'chatwoot_message', payload: {} });

  await claimNextJob();
  assert.equal(await failJob(id, 'boom 1'), 'retry');

  await db().job.update({ where: { id }, data: { runAfter: new Date(0) } });
  await claimNextJob();
  assert.equal(await failJob(id, 'boom 2'), 'dead');

  const job = await db().job.findUnique({ where: { id } });
  assert.equal(job?.status, 'dead');
  assert.equal(job?.attempts, 2);
});

test('secrets never reach the stored payload or the stored error', async () => {
  const id = await enqueue({
    type: 'chatwoot_message',
    payload: { api_token: 'super-secret-token', conversationId: 5 },
  });
  await claimNextJob();
  await failJob(id, 'failed calling https://shop.test/x?consumer_key=ck_leakedvalue');

  const job = await db().job.findUnique({ where: { id } });
  assert.ok(!job?.payloadJson.includes('super-secret-token'));
  assert.ok(!job?.lastError?.includes('ck_leakedvalue'));
});

test('queueStats counts jobs per status', async () => {
  await enqueue({ type: 'chatwoot_message', payload: {} });
  await enqueue({ type: 'chatwoot_message', payload: {} });
  const id = await claimNextJob();
  assert.ok(id);
  await completeJob(id.id);

  const stats = await queueStats();
  assert.equal(stats.queued, 1);
  assert.equal(stats.done, 1);
});