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