COMERCIAL-GTI/tests/jobs-manager.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { EventEmitter } from 'node:events';
import { PassThrough } from 'node:stream';
import { createJobManager, recoverOrphans } from '../src/jobs/manager.mjs';
import { openDb } from '../src/db.mjs';

test('close cancela jobs, aguarda exit e mata ao exceder prazo', async () => {
  const children = [];
  const spawn = () => {
    const child = new EventEmitter();
    child.stdout = new PassThrough();
    child.stderr = new PassThrough();
    child.stdin = { write: (value) => { child.cancel = value; } };
    child.kill = () => { child.killed = true; };
    children.push(child);
    return child;
  };
  const jobs = createJobManager({ spawn, killGraceMs: 5 });
  jobs.startJob({ type: 'teste', script: 'fake.mjs' });
  await Promise.race([
    jobs.close(),
    new Promise((resolve, reject) => setTimeout(() => reject(new Error('close nao respeitou prazo')), 50))
  ]);
  assert.equal(children[0].cancel, 'cancel\n');
  assert.equal(children[0].killed, true);
  assert.equal(jobs.activeJobs().length, 0);
});

test('recoverOrphans recupera heartbeat stale mesmo com PID reutilizado e preserva lease recente', () => {
  const db = openDb(':memory:');
  db.prepare("INSERT INTO capture_runs (params_json, status, pid, heartbeat_at) VALUES ('{}', 'running', ?, '2000-01-01 00:00:00')").run(process.pid);
  db.prepare("INSERT INTO extractions (uf, status, pid, heartbeat_at) VALUES ('BA', 'pending', 999, datetime('now'))").run();
  db.prepare("INSERT INTO exports (tipo, status, pid, heartbeat_at) VALUES ('xlsx', 'running', 999, '2000-01-01 00:00:00')").run();

  recoverOrphans(db, () => {}, { processAlive: (pid) => pid === process.pid });

  assert.equal(db.prepare('SELECT status FROM capture_runs').get().status, 'error');
  assert.equal(db.prepare('SELECT status FROM extractions').get().status, 'pending');
  const recovered = db.prepare('SELECT status, error, finished_at FROM exports').get();
  assert.equal(recovered.status, 'error');
  assert.match(recovered.error, /interrompido/);
  assert.ok(recovered.finished_at);
  db.close();
});

test('manager aguarda close para publicar done e converte close nao zero apos done em erro', async () => {
  const children = [];
  const spawn = () => {
    const child = new EventEmitter();
    child.stdout = new PassThrough();
    child.stderr = new PassThrough();
    child.stdin = { write() {} };
    child.kill = () => {};
    children.push(child);
    return child;
  };
  const jobs = createJobManager({ spawn });
  const firstId = jobs.startJob({ type: 'teste', script: 'fake' });
  const firstEvents = [];
  jobs.subscribe(firstId, (event) => firstEvents.push(event));
  children[0].stdout.write('{"t":"done","summary":{"total":1}}\n');
  await new Promise((resolve) => setImmediate(resolve));
  assert.equal(jobs.getJob(firstId).status, 'running');
  assert.equal(firstEvents.some((event) => event.t === 'done'), false);
  children[0].emit('exit', 0, null);
  assert.equal(jobs.getJob(firstId).status, 'running');
  children[0].emit('close', 0, null);
  assert.equal(jobs.getJob(firstId).status, 'done');
  assert.equal(firstEvents.at(-1).t, 'done');

  const secondId = jobs.startJob({ type: 'teste', script: 'fake' });
  const secondEvents = [];
  jobs.subscribe(secondId, (event) => secondEvents.push(event));
  children[1].stdout.write('{"t":"done","summary":{}}\n');
  await new Promise((resolve) => setImmediate(resolve));
  children[1].emit('exit', 1, null);
  children[1].emit('close', 1, null);
  assert.equal(jobs.getJob(secondId).status, 'error');
  assert.equal(secondEvents.at(-1).t, 'error');
});

test('exit nao zero corrige estado done persistido prematuramente pelo filho', async () => {
  const db = openDb(':memory:');
  const runId = Number(db.prepare("INSERT INTO capture_runs (params_json, status) VALUES ('{}', 'running')").run().lastInsertRowid);
  const child = new EventEmitter();
  child.stdout = new PassThrough();
  child.stderr = new PassThrough();
  child.stdin = { write() {} };
  child.kill = () => {};
  const jobs = createJobManager({ spawn: () => child, db });

  jobs.startJob({ type: 'captura', script: 'fake', meta: { runId } });
  db.prepare("UPDATE capture_runs SET status = 'done' WHERE id = ?").run(runId);
  child.stdout.write('{"t":"done","summary":{}}\n');
  await new Promise((resolve) => setImmediate(resolve));
  child.emit('exit', 1, null);
  child.emit('close', 1, null);

  assert.equal(db.prepare('SELECT status FROM capture_runs WHERE id = ?').get(runId).status, 'error');
  db.close();
});

test('manager persiste erro de filho inesperado e cancelamento forcado no DB', async () => {
  const db = openDb(':memory:');
  const captureId = Number(db.prepare("INSERT INTO capture_runs (params_json, status) VALUES ('{}', 'pending')").run().lastInsertRowid);
  const extractionId = Number(db.prepare("INSERT INTO extractions (uf, status) VALUES ('BA', 'pending')").run().lastInsertRowid);
  const children = [];
  const spawn = () => {
    const child = new EventEmitter();
    child.stdout = new PassThrough();
    child.stderr = new PassThrough();
    child.stdin = { write() {} };
    child.kill = () => { child.emit('exit', null, 'SIGTERM'); child.emit('close', null, 'SIGTERM'); };
    children.push(child);
    return child;
  };
  const jobs = createJobManager({ spawn, db, killGraceMs: 1 });
  jobs.startJob({ type: 'captura', script: 'fake', meta: { runId: captureId } });
  children[0].stderr.end('boom');
  children[0].emit('exit', 1, null);
  children[0].emit('close', 1, null);
  assert.equal(db.prepare('SELECT status FROM capture_runs WHERE id = ?').get(captureId).status, 'error');

  const jobId = jobs.startJob({ type: 'extracao', script: 'fake', meta: { extractionId } });
  assert.equal(jobs.cancelJob(jobId), true);
  await new Promise((resolve) => setTimeout(resolve, 5));
  assert.equal(db.prepare('SELECT status FROM extractions WHERE id = ?').get(extractionId).status, 'cancelled');
  db.close();
});

test('manager mantem cancelado quando filho relata erro depois da intencao de cancelar', async () => {
  const db = openDb(':memory:');
  const extractionId = Number(db.prepare("INSERT INTO extractions (uf, status) VALUES ('BA', 'running')").run().lastInsertRowid);
  const child = new EventEmitter();
  child.stdout = new PassThrough();
  child.stderr = new PassThrough();
  child.stdin = { write() {} };
  child.kill = () => {};
  const jobs = createJobManager({ spawn: () => child, db, killGraceMs: 100 });
  const jobId = jobs.startJob({ type: 'extracao', script: 'fake', meta: { extractionId } });
  jobs.cancelJob(jobId);
  child.stdout.write('{"t":"error","message":"stream encerrado"}\n');
  await new Promise((resolve) => setImmediate(resolve));
  assert.equal(db.prepare('SELECT status FROM extractions WHERE id = ?').get(extractionId).status, 'cancelled');
  child.emit('exit', 1, null);
  child.emit('close', 1, null);
  db.close();
});

test('linha done drenada entre exit e close resulta em done', async () => {
  const child = new EventEmitter();
  child.stdout = new PassThrough();
  child.stderr = new PassThrough();
  child.stdin = { write() {} };
  child.kill = () => {};
  const jobs = createJobManager({ spawn: () => child });
  const jobId = jobs.startJob({ type: 'teste', script: 'fake' });
  child.emit('exit', 0, null);
  child.stdout.write('{"t":"done","summary":{"late":true}}\n');
  await new Promise((resolve) => setImmediate(resolve));
  assert.equal(jobs.getJob(jobId).status, 'running');
  child.emit('close', 0, null);
  assert.equal(jobs.getJob(jobId).status, 'done');
  assert.equal(jobs.getJob(jobId).lastEvent.summary.late, true);
});

// Cenario real do capture-job.mjs: main() chama io.done() apos o trabalho
// terminar (leads ja gravados pelo proprio filho), mas o finally (fechar
// lease/db) falha depois disso — main().catch escreve uma linha "error"
// explicita e sai com codigo != 0. O done ja reportado nao pode virar erro
// so por causa da falha de limpeza que veio depois dele.
test('erro explicito apos done (falha de limpeza) nao sobrescreve o done', async () => {
  const db = openDb(':memory:');
  const runId = Number(db.prepare("INSERT INTO capture_runs (params_json, status) VALUES ('{}', 'running')").run().lastInsertRowid);
  const child = new EventEmitter();
  child.stdout = new PassThrough();
  child.stderr = new PassThrough();
  child.stdin = { write() {} };
  child.kill = () => {};
  const jobs = createJobManager({ spawn: () => child, db });

  const jobId = jobs.startJob({ type: 'captura', script: 'fake', meta: { runId } });
  const events = [];
  jobs.subscribe(jobId, (event) => events.push(event));

  child.stdout.write('{"t":"done","summary":{"new":42}}\n');
  await new Promise((resolve) => setImmediate(resolve));
  child.stdout.write('{"t":"error","message":"falha ao fechar conexao do banco"}\n');
  await new Promise((resolve) => setImmediate(resolve));
  assert.equal(jobs.getJob(jobId).status, 'running');

  child.emit('exit', 1, null);
  child.emit('close', 1, null);

  assert.equal(jobs.getJob(jobId).status, 'done');
  assert.equal(events.at(-1).t, 'done');
  assert.equal(events.at(-1).summary.new, 42);
  assert.equal(db.prepare('SELECT status FROM capture_runs WHERE id = ?').get(runId).status, 'done');
  db.close();
});

test('manager trata erro do stream stdout como falha terminal persistida', async () => {
  const db = openDb(':memory:');
  const exportId = Number(db.prepare("INSERT INTO exports (tipo, status) VALUES ('excel', 'running')").run().lastInsertRowid);
  const child = new EventEmitter();
  child.stdout = new PassThrough();
  child.stderr = new PassThrough();
  child.stdin = { write() {} };
  child.kill = () => { child.killed = true; };
  const jobs = createJobManager({ spawn: () => child, db });
  jobs.startJob({ type: 'exportacao', script: 'fake', meta: { exportId } });
  child.stdout.emit('error', new Error('pipe quebrado'));
  assert.equal(db.prepare('SELECT status FROM exports WHERE id = ?').get(exportId).status, 'error');
  assert.equal(child.killed, true);
  db.close();
});