`apps/workers/src/index.ts`

59 lines
import {
  work,
  stopQueue,
  installGracefulShutdown,
  startQueueHealthCanary,
  computeQueueHealth,
  getQueueHealthState,
} from "@homesonspec/shared";
import { meridianHomesAdapter } from "@homesonspec/collector-meridian-homes";
import { runPipeline } from "./pipeline";
import { startHealthServer, closeHealthServer } from "./health-server";

/**
 * Queue-backed worker daemon (pg-boss). Stages stay pure functions; this
 * process merely subscribes to job names and invokes them. Scheduling of
 * refresh jobs per source-registry intervals lands with live sources.
 */
const ADAPTERS = { [meridianHomesAdapter.key]: meridianHomesAdapter };
const JOB_NAME = "pipeline.run";
const HEALTH_PORT = Number(process.env.WORKER_HEALTH_PORT ?? 9976);

async function main() {
  await work<{ adapterKey: string }>(JOB_NAME, async ({ adapterKey }) => {
    const adapter = ADAPTERS[adapterKey];
    if (!adapter) throw new Error(`unknown adapter ${adapterKey}`);
    const summary = await runPipeline(adapter);
    console.log(`[pipeline.run] ${adapterKey}:`, JSON.stringify(summary));
  });
  console.log("workers: subscribed to pipeline.run");

  // Liveness probe: distinguishes a healthy-idle worker from a wedged-but-online
  // one (pg-boss no longer crashes on a DB error, so pm2 would otherwise report
  // "online" through an outage). 503 when the queue has errored recently without
  // a canary recovery — a pm2/monitoring health-check can act on it. Bind failure
  // (e.g. EADDRINUSE on the shared host) is non-fatal — see startHealthServer.
  const healthServer = startHealthServer({
    port: HEALTH_PORT,
    health: () => computeQueueHealth(getQueueHealthState(), Date.now()),
  });

  const canary = startQueueHealthCanary({ queueName: JOB_NAME });

  // pm2 reload/restart -> SIGINT/SIGTERM: stop the canary + health server and
  // drain the queue (see stopQueue) instead of dropping in-flight leases.
  installGracefulShutdown({
    stop: async () => {
      clearInterval(canary);
      closeHealthServer(healthServer);
      await stopQueue();
    },
  });
  console.log("workers: graceful shutdown wired (SIGINT/SIGTERM)");
}

main().catch((error) => {
  console.error(error);
  process.exit(1);
});