/**
 * apps/worker — main entry (headless: no business HTTP surface).
 *
 * The worker listens on exactly ONE port: the HTTP health server
 * (TASK-0009 item 4, default 4003) — /health + /health/ready. No other
 * routes are exposed.
 *
 * Startup order:
 *   1. config (zod, dev-safe defaults) + logger
 *   2. Redis (shared connection: BullMQ + streams + heartbeat)
 *   3. Prisma connect (surfaces bad DATABASE_URL immediately)
 *   4. OpenSearch indices (idempotent create)
 *   5. seven BullMQ Workers (concurrency = WORKER_CONCURRENCY)
 *   6. four StreamConsumers on factory:events (one consumer group per queue:
 *      worker-search-index / worker-webhooks / worker-automations /
 *      worker-email — the TASK-0094 notification→email decision)
 *   7. repeatable maintenance jobs + schedule-automation reconciliation
 *      (re-run every 5 min so DB edits to schedules take effect)
 *   8. heartbeat: SET worker:heartbeat:<WORKER_ID> EX 90 every 30s (deep
 *      liveness for the api) + HTTP health server on HEALTH_PORT (4003)
 *
 * Graceful shutdown (SIGTERM/SIGINT):
 *   close workers (finish active jobs, 30s cap) → stop consumers →
 *   disconnect prisma + redis → exit 0.
 */
import { setTimeout as delay } from "node:timers/promises";
import { Queue, Worker } from "bullmq";
import { initOtel, shutdownOtel, m, renderPrometheus } from "@factory/observability";
import { loadEnv } from "./config.js";
import { getLogger } from "./logger.js";
import { getRedis, closeRedis, createRedis } from "./lib/redis.js";
import { createHealthServer } from "./health/server.js";
import { connectPrisma, closePrisma } from "./lib/prisma.js";
import { S3Client } from "./lib/s3.js";
import { SearchClient } from "@factory/search";
import { QUEUES, enqueue } from "./queues.js";
import { createFilesWorker } from "./processors/files/processor.js";
import { assertScannerConfigured } from "./processors/files/steps/scan.js";
import { cleanupTempFiles } from "./processors/files/temp.js";
import {
  createSearchIndexWorker,
  createSearchIndexConsumer,
  ensureContentIndex,
  recordSearchDlq
} from "./processors/search-index/index.js";
import {
  createWebhooksWorker,
  createWebhooksConsumer,
  handleWebhookFinalFailure
} from "./processors/webhooks.processor.js";
import {
  createAutomationsWorker,
  createAutomationsConsumer
} from "./processors/automations.processor.js";
import { reconcileScheduleAutomations } from "./processors/automations/schedule.js";
import { runWatchdog } from "./processors/automations/watchdog.js";
import { createEmailWorker } from "./processors/email.processor.js";
import { createNotificationEmailConsumer } from "./processors/email-decision.processor.js";
import { createAiWorker } from "./processors/ai.processor.js";
import {
  createMaintenanceWorker,
  MAINTENANCE_REPEATABLES
} from "./processors/maintenance.processor.js";

const log = getLogger("main");

const SHUTDOWN_CAP_MS = 30_000;
const HEARTBEAT_INTERVAL_MS = 30_000;
const HEARTBEAT_TTL_SEC = 90;

async function main(): Promise<void> {
  const env = loadEnv();
  const started = Date.now();
  // TASK-0135: OTel FIRST (same contract as the API — the lazy metric
  // instruments bind to the SDK provider at first touch; disabled → no-op).
  // OTEL_* come straight from process.env (NOT in config.ts's zod schema —
  // keeps env.ts/config.ts untouched per the coordinator boundary).
  const otelEndpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
  await initOtel({
    serviceName: "factory-worker",
    environment: env.NODE_ENV,
    ...(process.env.OTEL_ENABLED === "true" && otelEndpoint ? { endpoint: otelEndpoint } : {})
  });
  log.info(
    {
      workerId: env.WORKER_ID,
      concurrency: env.WORKER_CONCURRENCY,
      env: env.NODE_ENV,
      smtp: env.SMTP_HOST ? "configured" : "not configured (email no-op)",
      clamav: env.CLAMD_SOCKET
        ? `unix ${env.CLAMD_SOCKET}`
        : env.CLAMAV_HOST
          ? `tcp ${env.CLAMAV_HOST}:${env.CLAMAV_PORT}`
          : "not configured (scans skipped, dev mode)",
      ai: env.AI_API_KEY ? "configured" : "not configured (ai no-op)",
      search: env.SEARCH_URL
    },
    "factory worker starting"
  );

  // --- infrastructure ------------------------------------------------------
  const redis = getRedis(env.REDIS_URL);
  const prisma = await connectPrisma();
  const s3 = new S3Client({
    endpoint: env.S3_ENDPOINT,
    region: env.S3_REGION,
    // TASK-0103 fix (mirrors apps/api/src/modules/files/s3.ts): the files
    // bucket is `S3_BUCKET_FILES` (with `S3_BUCKET` kept as a legacy alias
    // for older .env files). Previously this read the raw legacy `S3_BUCKET`,
    // so a dev .env carrying the stale pre-Phase-9 `S3_BUCKET=factory-uploads`
    // sent the pipeline's downloadTo/putObject to a bucket that no longer
    // exists (NoSuchKey 404) even though the API presigned into
    // `factory-files`. See docs/tasks/phase-09-files.md (TASK-0103 notes).
    bucket: env.S3_BUCKET_FILES || env.S3_BUCKET,
    accessKey: env.S3_ACCESS_KEY,
    secretKey: env.S3_SECRET_KEY,
    publicBase: env.S3_PUBLIC_BASE
  });
  // TASK-0102: the thumbs bucket client (same creds, `factory-thumbs`). The
  // files pipeline writes generated thumbnails here.
  const s3Thumbs = new S3Client({
    endpoint: env.S3_ENDPOINT,
    region: env.S3_REGION,
    bucket: env.S3_BUCKET_THUMBS,
    accessKey: env.S3_ACCESS_KEY,
    secretKey: env.S3_SECRET_KEY
  });
  // TASK-0102 §3: fail closed — production refuses to boot without a ClamAV
  // scanner (a silently-unscanned prod file pipeline is a security hole).
  assertScannerConfigured(env);
  // The single OpenSearch writer (TASK-0096): the @factory/search client over
  // the `factory-content` alias. Basic auth when the password is set
  // (OpenSearch security on); dev/CI run security-off.
  const search = new SearchClient({
    baseUrl: env.SEARCH_URL,
    ...(env.SEARCH_PASSWORD ? { auth: { username: env.SEARCH_USERNAME, password: env.SEARCH_PASSWORD } } : {})
  });
  const searchOps = {
    search,
    prisma,
    connection: redis,
    env,
    // TASK-0125 — incremental RAG embedding: after a page/record is
    // (re)indexed, its vectors are refreshed on the `ai` queue (the embed
    // job re-derives the chunk set from the source). Fire-and-forget by the
    // hook's own try/catch (indexDocs never fails on an enqueue error).
    enqueueEmbed: (job: { workspaceId: string; docType: "page" | "record"; refId: string }) =>
      enqueue(redis, QUEUES.ai, { type: "embed:chunks", workspaceId: job.workspaceId, docType: job.docType, refId: job.refId } as Record<string, unknown>)
  };
  await ensureContentIndex(searchOps).catch((err) => {
    log.warn(
      { err: (err as Error).message },
      "OpenSearch unavailable at startup — index will be created lazily by the first (re)index; continuing"
    );
  });

  // --- workers -------------------------------------------------------------
  const workers: Worker[] = [];
  const push = (w: Worker): void => {
    w.on("failed", (job, err) => {
      log.error(
        { queue: w.name, jobId: job?.id, err: err.message },
        "job failed (BullMQ will retry per job options)"
      );
    });
    workers.push(w);
  };

  push(
    createFilesWorker({
      connection: redis,
      prisma,
      env,
      s3,
      s3Thumbs,
      // TASK-0102 §7: the files pipeline is IO-bound (scan/render) but uses
      // disk + CPU (sharp/ffmpeg) — its own lower concurrency, not the
      // general WORKER_CONCURRENCY.
      concurrency: env.FILES_WORKER_CONCURRENCY
    })
  );
  const searchIndexWorker = createSearchIndexWorker({
    connection: redis,
    ops: searchOps,
    concurrency: env.WORKER_CONCURRENCY
  });
  searchIndexWorker.on("failed", (job, err) => {
    if (job && job.attemptsMade >= 7) {
      // 8th attempt exhausted → dead-letter record (TASK-0096 §8).
      void recordSearchDlq(redis, job, err);
    }
  });
  push(searchIndexWorker);
  const webhooksWorker = createWebhooksWorker({
    connection: redis,
    prisma,
    env,
    concurrency: env.WORKER_CONCURRENCY
  });
  webhooksWorker.on("failed", (job, err) => {
    // TASK-0130: the worker owns retry scheduling (attempts:1 on the queue —
    // every FAILED attempt re-enqueues itself as a delayed job), so a
    // BullMQ "failed" here means an UNEXPECTED job crash: dead-letter the
    // delivery row + notify admins (handleWebhookFinalFailure).
    if (job) void handleWebhookFinalFailure(prisma, redis, env)(job, err);
  });
  push(webhooksWorker);
  // TASK-0113: the pipeline is the retry manager (attempts:1 on the queue) —
  // the old BullMQ `failed` hook (attemptsMade >= 7) is gone; every failure
  // is handled inside handleRunAutomation (FAILED / RETRYING / DEAD_LETTER).
  const automationsWorker = createAutomationsWorker({
    connection: redis,
    prisma,
    env,
    concurrency: env.AUTOMATIONS_WORKER_CONCURRENCY
  });
  push(automationsWorker);
  push(
    createEmailWorker({
      connection: redis,
      prisma,
      env,
      concurrency: env.WORKER_CONCURRENCY
    })
  );
  push(
    createAiWorker({
      connection: redis,
      prisma,
      env,
      search,
      concurrency: env.WORKER_CONCURRENCY
    })
  );
  const maintenanceWorker = createMaintenanceWorker({
    connection: redis,
    prisma,
    env,
    concurrency: Math.max(1, Math.min(2, env.WORKER_CONCURRENCY))
  });
  push(maintenanceWorker);
  log.info({ count: workers.length }, "bullmq workers created");

  // TASK-0135 — queue_depth{queue} gauge: the spec's "queue-depth per queue".
  // One short-lived Queue handle per queue every 15 s (waiting + delayed =
  // the backlog the Workers/Queues dashboard shows + the >10k alert tests).
  // Fail-soft: a transient Redis blip logs at debug level, never crashes the
  // worker. m.queue.setDepth is a no-op when OTel is disabled.
  const queueDepthTick = async (): Promise<void> => {
    for (const name of Object.values(QUEUES)) {
      const q = new Queue(name, { connection: redis });
      try {
        const c = await q.getJobCounts("waiting", "delayed");
        m.queue.setDepth(name, (c.waiting ?? 0) + (c.delayed ?? 0));
      } catch (err) {
        log.debug({ queue: name, err: (err as Error).message }, "queue depth probe failed");
      } finally {
        await q.close();
      }
    }
  };
  void queueDepthTick();
  const queueDepthTimer = setInterval(() => void queueDepthTick(), 15_000);
  queueDepthTimer.unref();

  // --- stream consumers (one group per queue) -------------------------------
  // Each consumer gets its OWN ioredis connection. A StreamConsumer's read
  // loop issues XREADGROUP ... BLOCK 5000 back-to-back, and an ioredis client
  // is a single ordered-response socket: if two or more consumers share a
  // connection, their blocking reads serialize on that one socket and every
  // non-blocking command queued behind them (heartbeat SET, BullMQ lock-renew
  // PEXPIRE, boot queue ops, the search-index:lag probe) is starved. With the
  // four TASK-0094/0096/0100 consumers on the shared client, boot deadlocked
  // before the health server could start (TASK-0100 S5 diagnosis). BullMQ's
  // workers already run on dedicated named clients for the same reason.
  const searchStreamRedis = createRedis(env.REDIS_URL);
  const webhooksStreamRedis = createRedis(env.REDIS_URL);
  const automationsStreamRedis = createRedis(env.REDIS_URL);
  const emailStreamRedis = createRedis(env.REDIS_URL);
  const consumers = [
    createSearchIndexConsumer({
      redis: searchStreamRedis,
      ops: searchOps,
      consumerId: `${env.WORKER_ID}-search`
    }),
    createWebhooksConsumer({
      redis: webhooksStreamRedis,
      prisma,
      consumerId: `${env.WORKER_ID}-webhooks`
    }),
    createAutomationsConsumer({
      redis: automationsStreamRedis,
      prisma,
      consumerId: `${env.WORKER_ID}-automations`
    }),
    // TASK-0094 R6: the notification→email decision (its own group, so email
    // lag can't block the other consumers). The hourly digest is a repeatable
    // maintenance job (registered above via MAINTENANCE_REPEATABLES). `redis`
    // is the stream read connection (dedicated); `connection` stays the shared
    // client (only for occasional non-blocking enqueue).
    createNotificationEmailConsumer({
      redis: emailStreamRedis,
      connection: redis,
      prisma,
      env,
      consumerId: `${env.WORKER_ID}-email`
    })
  ];
  for (const c of consumers) await c.start();
  log.info({ count: consumers.length }, "stream consumers started");

  // --- repeatable jobs -------------------------------------------------------
  const mq = new Queue(QUEUES.maintenance, { connection: redis });
  for (const r of MAINTENANCE_REPEATABLES) {
    await mq.add(r.id, r.data as Record<string, unknown>, {
      repeat: { pattern: r.pattern },
      jobId: r.id,
      removeOnComplete: true,
      removeOnFail: false
    });
  }
  await mq.close();
  log.info({ jobs: MAINTENANCE_REPEATABLES.map((r) => r.id) }, "repeatable maintenance jobs registered");

  // TASK-0121 — nightly AI usage rollup (the ai queue owns it: it reconciles
  // the per-workspace month-to-date token totals for dashboards and sweeps
  // for workspaces that crossed their budget but have since gone quiet — the
  // API's submit-time guard only notifies on the NEXT call). Registered on a
  // SEPARATE repeatable queue instance (BullMQ scopes a repeatable's key to
  // the queue it was added on), offset from the 03:00 maintenance window.
  const aiQ = new Queue(QUEUES.ai, { connection: redis });
  await aiQ.add("ai", { type: "usage.rollup" }, {
    repeat: { pattern: "15 2 * * *" },
    jobId: "ai:usage-rollup",
    removeOnComplete: true,
    removeOnFail: false
  });
  await aiQ.close();
  log.info({ job: "ai:usage-rollup" }, "repeatable ai usage rollup registered");

  // TASK-0125 — nightly FULL re-embed (03:30 UTC, offset from the 02:15
  // usage rollup): re-embeds every ai_embeddings row, verifies count parity
  // (drift > 2% → ai.reembed_required) and is the model-change re-index rule
  // (a new embedding model simply re-embeds everything overnight). Same
  // separate-instance pattern as the usage rollup.
  const aiReembedQ = new Queue(QUEUES.ai, { connection: redis });
  await aiReembedQ.add("ai", { type: "embed:reembed" }, {
    repeat: { pattern: "30 3 * * *" },
    jobId: "ai:reembed",
    removeOnComplete: true,
    removeOnFail: false
  });
  await aiReembedQ.close();
  log.info({ job: "ai:reembed" }, "repeatable ai nightly re-embed registered");

  // Schedule automations: register now + reconcile every 5 minutes.
  const scheduleReconcile = async (): Promise<void> => {
    try {
      await reconcileScheduleAutomations(redis, prisma);
    } catch (err) {
      log.warn({ err: (err as Error).message }, "schedule reconcile failed");
    }
  };
  await scheduleReconcile();
  const scheduleTimer = setInterval(scheduleReconcile, 5 * 60_000);
  scheduleTimer.unref();

  // TASK-0113 §9: watchdog — reclaim stale runs (RUNNING > 15 min or
  // PENDING > 15 min) every 5 min. The in-process interval mirrors the
  // schedule reconcile (a BullMQ repeatable would just re-enter the same
  // pool; an interval keeps the watchdog off the queue and trivially
  // observable).
  const watchdogTick = async (): Promise<void> => {
    try {
      await runWatchdog(prisma, redis, env);
    } catch (err) {
      log.warn({ err: (err as Error).message }, "watchdog pass failed");
    }
  };
  const watchdogTimer = setInterval(watchdogTick, 5 * 60_000);
  watchdogTimer.unref();

  // TASK-0102 §3/§7: temp-file cleanup cron (every 5 min). The processor
  // deletes its own temp file in a `finally`; this reaps the crash
  // leftovers (a SIGKILL mid-pipeline) older than FILES_TEMP_MAX_AGE_MS so
  // an object "never stays on disk > scan duration + 5 min".
  const tempCleanup = async (): Promise<void> => {
    try {
      await cleanupTempFiles(env);
    } catch (err) {
      log.warn({ err: (err as Error).message }, "temp cleanup cron failed");
    }
  };
  await tempCleanup();
  const tempTimer = setInterval(tempCleanup, 5 * 60_000);
  tempTimer.unref();

  // --- heartbeat (headless liveness for api /health/deep) --------------------
  const heartbeat = async (): Promise<void> => {
    try {
      await redis.set(
        `worker:heartbeat:${env.WORKER_ID}`,
        JSON.stringify({ at: new Date().toISOString(), uptimeMs: Date.now() - started }),
        "EX",
        HEARTBEAT_TTL_SEC
      );
    } catch (err) {
      log.warn({ err: (err as Error).message }, "heartbeat write failed");
    }
  };
  await heartbeat();
  const heartbeatTimer = setInterval(heartbeat, HEARTBEAT_INTERVAL_MS);
  heartbeatTimer.unref();

  // --- HTTP health (TASK-0009 item 4): /health + /health/ready on 4003 --------
  // The worker is otherwise headless; this is its only port. BullMQ rides on the
  // same Redis backend, so the queue check is an independent Redis PING.
  //
  // TASK-0100 req 1 (Backend Changes): `search-index:lag` — the consumer-lag
  // metric the lag E2E polls while asserting the worker consumes a 1k-event
  // burst without backlog growth > 30 s.
  //   pending    = XPENDING factory:events worker-search-index (delivered,
  //                not yet ACKed — in-flight or awaiting XAUTOCLAIM redelivery)
  //   lagEntries = entries in the stream NOT yet delivered to the group:
  //                the exclusive tail after the group's last-delivered-id
  //                (XINFO GROUPS). backlog = pending + lagEntries.
  // Every Redis read fails soft → the endpoint reports available:false
  // (observed as zeroed counts) instead of 5xx; the probe never gates
  // readiness and never throws out of boot.
  // Shared probe: the /health/search-index:lag endpoint AND the 30 s OTel
  // gauge tick both read the same stream-group state (TASK-0135 added the
  // seconds dimension — the SearchIndexLagHigh alert needs an elapsed-time
  // gauge, not just the entry counts).
  const probeSearchIndexLag = async (): Promise<{
    pending: number;
    lagEntries: number;
    lagSeconds: number;
  }> => {
    const STREAM = "factory:events";
    const GROUP = "worker-search-index";
    const [pendingInfo, groups] = await Promise.all([
      (redis.xpending(STREAM, GROUP) as unknown as Promise<{ count: number }>).catch(() => ({ count: 0 })),
      (redis.xinfo("GROUPS", STREAM) as Promise<{ name: string; "last-delivered-id": string }[]>)
        .catch(() => [] as { name: string; "last-delivered-id": string }[])
    ]);
    const pending = pendingInfo?.count ?? 0;
    let lagEntries = 0;
    const groupInfo = groups.find((g) => g.name === GROUP);
    let tailId: string | null = null;
    if (groupInfo) {
      // Entries with ids strictly greater than last-delivered-id. Redis
      // exclusive id ranges (Redis ≥ 6.2; dev/CI pin redis:7): prefix the
      // exact id with "<" to exclude it.
      const lastId = groupInfo["last-delivered-id"] ?? "0-0";
      const tail = await redis
        .xrange(STREAM, `<${lastId}`, "+", "COUNT", 100_000)
        .catch((): string[][] => []);
      lagEntries = tail.length;
      // Oldest undelivered entry = first in the ascending tail (its stream
      // id embeds the creation milliseconds).
      tailId = tail[0]?.[0] ?? null;
    }
    // Lag in seconds = max(elapsed since the oldest UNDELIVERED event,
    // elapsed since the oldest PENDING (delivered, un-ACKed) event). Each
    // leg fails soft (a probe blip reports the other leg / 0, never throws).
    const nowMs = Date.now();
    let lagSeconds = 0;
    if (tailId) {
      const tailMs = Number(tailId.split("-")[0]);
      if (Number.isFinite(tailMs) && tailMs > 0) lagSeconds = Math.max(0, (nowMs - tailMs) / 1000);
    }
    try {
      const entries = await redis.xpending(STREAM, GROUP, "-", "+", 1);
      const first = (Array.isArray(entries) ? entries[0] : undefined) as
        | { "time-delivered-in-ms"?: number }
        | undefined;
      const deliveredMs = first?.["time-delivered-in-ms"];
      if (typeof deliveredMs === "number" && deliveredMs > 0) {
        lagSeconds = Math.max(lagSeconds, (nowMs - deliveredMs) / 1000);
      }
    } catch {
      // xpending with range unsupported / stream gone — the tail leg above
      // still covers undelivered lag.
    }
    return { pending, lagEntries, lagSeconds };
  };

  const searchIndexLag = async (): Promise<{ pending: number; lagEntries: number }> => {
    const r = await probeSearchIndexLag();
    return { pending: r.pending, lagEntries: r.lagEntries };
  };

  const healthServer = createHealthServer({
    pingRedis: async () => (await redis.ping()) === "PONG",
    checkQueue: async () => (await redis.ping()) === "PONG",
    checkSearchIndexLag: searchIndexLag,
    // TASK-0135: OTel Prometheus exposition on the worker's ONLY port (the
    // scrape config targets :4003/metrics — same pattern as the API).
    metricsBody: renderPrometheus
  });
  await new Promise<void>((resolve, reject) => {
    healthServer.once("error", reject);
    healthServer.listen(env.HEALTH_PORT, () => {
      healthServer.removeListener("error", reject);
      resolve();
    });
  });
  log.info({ port: env.HEALTH_PORT }, "worker health server listening");

  log.info({ ms: Date.now() - started }, "factory worker ready");

  // TASK-0135 — search_index_lag_seconds gauge: same probe as the
  // /health/search-index:lag endpoint, on its own 30 s cadence (the
  // SearchIndexLagHigh alert reads this; the 30 s window matches the
  // alert's own "for" period). No-op when OTel is disabled.
  const searchLagTick = async (): Promise<void> => {
    try {
      const { lagSeconds } = await probeSearchIndexLag();
      m.search.lag(lagSeconds);
    } catch (err) {
      log.debug({ err: (err as Error).message }, "search-index lag probe failed");
    }
  };
  void searchLagTick();
  const searchLagTimer = setInterval(() => void searchLagTick(), 30_000);
  searchLagTimer.unref();

  // --- graceful shutdown -----------------------------------------------------
  let shuttingDown = false;
  const shutdown = async (signal: string): Promise<void> => {
    if (shuttingDown) return;
    shuttingDown = true;
    log.info({ signal }, "shutting down (30s cap)");
    const cap = delay(SHUTDOWN_CAP_MS).then(() => {
      log.error("shutdown cap reached — forcing exit");
      process.exit(1);
    });
    try {
      clearInterval(heartbeatTimer);
      clearInterval(scheduleTimer);
      clearInterval(tempTimer);
      clearInterval(queueDepthTimer);
      clearInterval(searchLagTimer);
      // Health server first (no in-flight jobs; just stop accepting).
      await new Promise<void>((resolve) => {
        healthServer.close(() => resolve());
        healthServer.closeAllConnections?.();
      });
      // Workers first (finish active jobs; BullMQ closes connections after).
      await Promise.allSettled(workers.map((w) => w.close()));
      for (const c of consumers) {
        try {
          await c.stop();
        } catch (err) {
          log.warn({ err: (err as Error).message }, "consumer stop error");
        }
      }
      // Close the dedicated per-consumer stream connections (each runs an
      // XREADGROUP BLOCK read loop; the shared client is closed below).
      await Promise.allSettled(
        [searchStreamRedis, webhooksStreamRedis, automationsStreamRedis, emailStreamRedis].map(
          (r) => r.quit()
        )
      );
      // TASK-0135: flush the OTel SDK (pending spans + the final metric
      // collection) before the process dies.
      await shutdownOtel();
      await closePrisma();
      await closeRedis();
      log.info("shutdown complete");
      process.exit(0);
    } catch (err) {
      log.error({ err: (err as Error).message }, "error during shutdown");
      process.exit(1);
    }
  };
  process.on("SIGTERM", () => void shutdown("SIGTERM"));
  process.on("SIGINT", () => void shutdown("SIGINT"));
}

main().catch((err) => {
  // Boot failure: log loud + non-zero exit (orchestrator restarts).
  // eslint-disable-next-line no-console
  console.error("factory worker failed to start:", err);
  process.exit(1);
});