74 lines
2.9 KiB
TypeScript
74 lines
2.9 KiB
TypeScript
import path from "node:path";
|
|
import { jobRunsInline } from "../src/config/env";
|
|
import { pendingMigrations } from "../src/infra/db/migration-manifest";
|
|
import { createCoordinator } from "../src/infra/redis";
|
|
import { createStorage } from "../src/infra/storage";
|
|
import { createWorker } from "../src/server/spreadsheet-adapter/worker-loop";
|
|
import { bootstrapScript, describeTarget, openOperationalDb, runScript } from "./_bootstrap";
|
|
|
|
/**
|
|
* Processo do worker. Só bootstrap, sinais e cadência: a lógica do tick está em
|
|
* `src/server/spreadsheet-adapter/worker-loop.ts`, que é testável sem subir
|
|
* processo.
|
|
*
|
|
* Precisa ser executado com `node --import tsx scripts/worker.ts`, não com o
|
|
* CLI do `tsx`: o CLI cria um processo filho e o SIGTERM do contêiner pararia
|
|
* no pai, sem nunca chegar ao handler que espera o lock ser liberado.
|
|
*/
|
|
|
|
const TICK_MS = Number(process.env.ADAPTER_WORKER_TICK_MS ?? 2000);
|
|
const BATCH = Number(process.env.ADAPTER_WORKER_BATCH ?? 20);
|
|
const GRACE_MS = Number(process.env.ADAPTER_WORKER_GRACE_MS ?? 20000);
|
|
const HEARTBEAT = process.env.ADAPTER_WORKER_HEARTBEAT ?? path.join("/tmp", "adaptador-worker-heartbeat");
|
|
|
|
bootstrapScript("worker");
|
|
if (jobRunsInline()) {
|
|
console.warn("[worker] ADAPTER_JOB_MODE não é worker. Este processo só é útil com a fila ligada.");
|
|
}
|
|
|
|
runScript("worker", async () => {
|
|
const db = await openOperationalDb("worker");
|
|
|
|
// O worker não aplica migrations: isso é passo explícito e aprovado do deploy.
|
|
const pending = await pendingMigrations(db);
|
|
if (pending.length > 0) {
|
|
await db.close().catch(() => undefined);
|
|
throw new Error(`migrations pendentes em ${describeTarget()}: ${pending.join(", ")}. Rode npm run migrate antes.`);
|
|
}
|
|
|
|
const deps = { db, storage: createStorage(), coordinator: createCoordinator() };
|
|
const worker = createWorker(deps, {
|
|
batch: BATCH,
|
|
heartbeatPath: HEARTBEAT,
|
|
onError: (scope, error) => {
|
|
// Nome e escopo do erro. Nunca conteúdo de célula.
|
|
console.error("worker", scope, error instanceof Error ? error.name : "error");
|
|
},
|
|
});
|
|
|
|
const timer = setInterval(() => {
|
|
void worker.tick();
|
|
}, TICK_MS);
|
|
|
|
let stopping = false;
|
|
async function shutdown(signal: string) {
|
|
if (stopping) return;
|
|
stopping = true;
|
|
console.log(`[worker] ${signal} recebido. Encerrando.`);
|
|
clearInterval(timer);
|
|
const clean = await worker.shutdown(GRACE_MS);
|
|
if (!clean) console.warn(`[worker] tick não terminou em ${GRACE_MS}ms. Encerrando de todo modo.`);
|
|
await db.close().catch(() => undefined);
|
|
console.log("[worker] encerrado.");
|
|
process.exit(0);
|
|
}
|
|
|
|
process.on("SIGTERM", () => void shutdown("SIGTERM"));
|
|
process.on("SIGINT", () => void shutdown("SIGINT"));
|
|
|
|
console.log(
|
|
`[worker] pronto. intervalo=${TICK_MS}ms lote=${BATCH} carencia=${GRACE_MS}ms alvo=${describeTarget()} heartbeat=${HEARTBEAT}`,
|
|
);
|
|
await worker.tick();
|
|
});
|