adaptador-planilhas/tests/integration/worker.test.ts

import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { addSource, createJob, createModel, saveMappings, startProcess } from "@/server/spreadsheet-adapter/service";
import { downloadOutput } from "@/server/spreadsheet-adapter/process-job";
import { createWorker } from "@/server/spreadsheet-adapter/worker-loop";
import { lockKey } from "@/infra/redis/redis";
import { ctxOf, createReadyDb, depsOf, IDS } from "../helpers/world";

/**
 * Jornada assíncrona pelo loop real do worker: a web enfileira e devolve, e o
 * processamento acontece depois, no tick. É o mesmo código que o processo do
 * contêiner executa. O que estes casos não cobrem é a fronteira de processo e o
 * Redis real — isso está em `tests/services/redis.test.ts` e no CI.
 */

const previous = { mode: process.env.ADAPTER_JOB_MODE };

beforeEach(() => {
  // Modo worker: `startProcess` só enfileira e sai.
  process.env.ADAPTER_JOB_MODE = "worker";
});

afterEach(() => {
  if (previous.mode === undefined) delete process.env.ADAPTER_JOB_MODE;
  else process.env.ADAPTER_JOB_MODE = previous.mode;
});

type World = Awaited<ReturnType<typeof prepare>>;

async function prepare() {
  const db = await createReadyDb();
  const deps = depsOf(db);
  const ana = ctxOf("ana", "alpha", "owner", true);
  const model = await createModel(deps, ana, {
    name: "Clientes",
    filename: "modelo.csv",
    bytes: Buffer.from("Nome,Valor\n", "utf8"),
    requiredColumns: ["Nome", "Valor"],
  });
  const modelBytes = await deps.storage.get(`workspaces/${IDS.alpha}/models/${model.id}/original`);
  const job = await createJob(deps, ana, model.id, "job-async");
  const added = await addSource(deps, ana, job.id, {
    filename: "fonte.csv",
    bytes: Buffer.from("Nome,Valor\nAna,10\nBruno,20\n", "utf8"),
  });
  const sourceId = added.job.sources[0].id;
  await saveMappings(deps, ana, job.id, sourceId, [
    { sourceColumn: "Nome", modelColumn: "Nome" },
    { sourceColumn: "Valor", modelColumn: "Valor" },
  ]);
  return { db, deps, ana, model, modelBytes, job, sourceId };
}

async function statusOf(world: World, jobId: string): Promise<string> {
  const rows = await world.db.withService((query) =>
    query.query<{ status: string }>(`select status from public.spreadsheet_jobs where id = $1`, [jobId]),
  );
  return rows[0]?.status ?? "ausente";
}

describe("jornada assíncrona pelo worker", () => {
  it("enfileira na web, processa no tick e entrega os três formatos", async () => {
    const world = await prepare();
    const { deps, ana, job, sourceId } = world;

    const queued = await startProcess(deps, ana, job.id, sourceId, "proc-async");
    expect(queued.status).toBe("queued");
    // Nada foi processado ainda: a web só enfileirou.
    await expect(downloadOutput(deps, ana, job.id, sourceId, "csv")).rejects.toMatchObject({ code: "NOT_FOUND" });

    const worker = createWorker(deps, { heartbeatPath: null });
    const result = await worker.tick();

    expect(result.processed).toEqual([job.id]);
    expect(await statusOf(world, job.id)).toBe("completed");

    const csv = await downloadOutput(deps, ana, job.id, sourceId, "csv");
    expect(csv.body.toString("utf8")).toContain("Ana");
    const xlsx = await downloadOutput(deps, ana, job.id, sourceId, "xlsx");
    expect(xlsx.body.subarray(0, 2).toString()).toBe("PK");
    const pdf = await downloadOutput(deps, ana, job.id, sourceId, "pdf");
    expect(pdf.body.subarray(0, 4).toString()).toBe("%PDF");

    const after = await deps.storage.get(`workspaces/${IDS.alpha}/models/${world.model.id}/original`);
    expect(after.equals(world.modelBytes)).toBe(true);
  });

  it("libera o lock do usuário ao concluir, permitindo o próximo job", async () => {
    const world = await prepare();
    const { deps, ana, job, sourceId } = world;
    await startProcess(deps, ana, job.id, sourceId, "proc-lock");
    await createWorker(deps, { heartbeatPath: null }).tick();

    // Lock livre: outro processamento consegue adquirir.
    expect(await deps.coordinator.acquireLock(lockKey(ana.userId), "outro-job", 30)).toBe(true);
  });

  it("encontra o job pela varredura do banco quando a fila do Redis se perdeu", async () => {
    const world = await prepare();
    const { deps, ana, job, sourceId } = world;
    await startProcess(deps, ana, job.id, sourceId, "proc-sem-fila");

    // Simula Redis reiniciado: a fila some, o estado durável no Postgres fica.
    while (await deps.coordinator.dequeue()) {
      /* drena */
    }
    expect(await deps.coordinator.dequeue()).toBeNull();

    const result = await createWorker(deps, { heartbeatPath: null }).tick();
    expect(result.processed).toEqual([job.id]);
    expect(await statusOf(world, job.id)).toBe("completed");
  });

  it("relê o contexto do dono e não processa com o módulo desligado", async () => {
    const world = await prepare();
    const { db, deps, ana, job, sourceId } = world;
    await startProcess(deps, ana, job.id, sourceId, "proc-flag");

    // A flag cai entre o enfileiramento e o tick.
    await db.withService((query) =>
      query.query(
        `update public.workspace_module_flags set enabled = false where workspace_id = $1 and module_key = 'spreadsheet_adapter'`,
        [IDS.alpha],
      ),
    );

    const result = await createWorker(deps, { heartbeatPath: null }).tick();
    expect(result.processed).toEqual([]);
    expect(await statusOf(world, job.id)).not.toBe("completed");
    await expect(downloadOutput(deps, ana, job.id, sourceId, "csv")).rejects.toBeTruthy();
  });

  it("expira temporários no tick e preserva o modelo", async () => {
    const world = await prepare();
    const { deps, ana, job, sourceId } = world;
    await startProcess(deps, ana, job.id, sourceId, "proc-ttl");
    await createWorker(deps, { heartbeatPath: null }).tick();
    await downloadOutput(deps, ana, job.id, sourceId, "csv");

    // Avança além do TTL e roda o tick de novo.
    const expirado = { ...deps, now: () => new Date("2027-01-01T00:00:00.000Z") };
    const result = await createWorker(expirado, { heartbeatPath: null }).tick();
    expect(result.cleaned).toBeGreaterThan(0);
    await expect(downloadOutput(expirado, ana, job.id, sourceId, "csv")).rejects.toBeTruthy();

    const model = await deps.storage.get(`workspaces/${IDS.alpha}/models/${world.model.id}/original`);
    expect(model.equals(world.modelBytes)).toBe(true);
  });

  it("grava heartbeat ao fim do tick", async () => {
    const world = await prepare();
    const escritas: string[] = [];
    const worker = createWorker(world.deps, {
      heartbeatPath: "/tmp/teste-heartbeat",
      writeHeartbeat: async (target) => {
        escritas.push(target);
      },
    });
    await worker.tick();
    expect(escritas).toEqual(["/tmp/teste-heartbeat"]);
  });
});

describe("desligamento gracioso do worker", () => {
  it("para de aceitar trabalho novo depois do shutdown", async () => {
    const world = await prepare();
    const { deps, ana, job, sourceId } = world;
    const worker = createWorker(deps, { heartbeatPath: null });

    expect(worker.isRunning()).toBe(true);
    expect(await worker.shutdown(1000)).toBe(true);
    expect(worker.isRunning()).toBe(false);

    await startProcess(deps, ana, job.id, sourceId, "proc-pos-shutdown");
    const result = await worker.tick();
    expect(result.processed).toEqual([]);
    expect(await statusOf(world, job.id)).toBe("queued");
  });

  it("espera o tick em andamento e não perde o job que não começou", async () => {
    const world = await prepare();
    const { deps, ana, job, sourceId } = world;
    await startProcess(deps, ana, job.id, sourceId, "proc-espera");

    const worker = createWorker(deps, { heartbeatPath: null });
    const inFlight = worker.tick();
    expect(worker.isBusy()).toBe(true);

    // O encerramento espera o tick corrente sair: é essa espera que dá tempo ao
    // `finally` do processamento liberar o lock de um job já iniciado.
    const clean = await worker.shutdown(10_000);
    expect(clean).toBe(true);
    expect(worker.isBusy()).toBe(false);
    await inFlight;

    // O job que ainda não havia começado continua `queued`: nada pela metade.
    expect(await statusOf(world, job.id)).toBe("queued");
    // O lock continua reservado para ESTE job, não vazou para outro dono. Em
    // modo worker ele é adquirido no enfileiramento, com TTL de 120s, e
    // `runProcess` o reaproveita no tick seguinte.
    expect(await deps.coordinator.ownsLock(lockKey(ana.userId), job.id)).toBe(true);
    expect(await deps.coordinator.acquireLock(lockKey(ana.userId), "outro-job", 5)).toBe(false);

    // O worker seguinte conclui o mesmo job, como aconteceria após um restart.
    const result = await createWorker(deps, { heartbeatPath: null }).tick();
    expect(result.processed).toEqual([job.id]);
    expect(await statusOf(world, job.id)).toBe("completed");
    // Concluído, o lock é liberado pelo `finally` do processamento.
    expect(await deps.coordinator.acquireLock(lockKey(ana.userId), "sonda", 5)).toBe(true);
  });

  it("reporta encerramento sujo quando a carência estoura", async () => {
    const world = await prepare();
    const worker = createWorker(world.deps, { heartbeatPath: null });
    const inFlight = worker.tick();
    expect(worker.isBusy()).toBe(true);
    // Carência zero: não espera nada e reporta que o tick não terminou.
    expect(await worker.shutdown(0)).toBe(false);
    await inFlight;
  });
});