/**
 * apps/worker — minimal HTTP health server (no NestJS), port 4003.
 *
 * Contract (TASK-0009, docs/development/development-guide.md port table):
 *   GET /health          -> 200 { status:"ok", app, version, uptimeSeconds, timestamp }
 *   GET /health/ready    -> 200 { status:"ready", checks:{redis,queue} }
 *                           | 503 { status:"not_ready", checks:{...} } naming the failing dep
 *   GET /health/search-index:lag  -> 200 SearchIndexLagResult (TASK-0100)
 *                           | 503 { available:false, ... } when the probe fails
 *   GET /metrics                  -> 200 text/plain Prometheus exposition
 *                                    (TASK-0135; 204 when OTel is disabled
 *                                    or the render fails — the scrape target
 *                                    exists either way)
 *
 * Unauthenticated by design (documented exception to the Phase 1 global guard);
 * it exposes no data beyond app name / version / dependency state / metrics.
 *
 * The worker is otherwise headless (BullMQ + stream consumers, no business HTTP
 * surface). This is the ONLY port the worker listens on.
 */
import { createServer, type Server } from "node:http";

export const APP_NAME = "factory-worker";
export const APP_VERSION = "0.1.0";

export interface WorkerHealthDeps {
  /** Redis PING — resolves true when the shared Redis is reachable. */
  pingRedis: () => Promise<boolean>;
  /** BullMQ queue probe — true when the queue backend is reachable. */
  checkQueue?: () => Promise<boolean>;
  /**
   * Search-index consumer lag probe (TASK-0100 req 1: the lag suite asserts
   * the worker consumes a 1k-event burst without backlog growth > 30 s via
   * the consumer-lag metric `search-index:lag`).
   *
   * Resolves a snapshot of the `worker-search-index` consumer group on
   * `factory:events`: `pending` = entries delivered but not yet ACKed
   * (in-flight or awaiting redelivery); `lagEntries` = entries in the stream
   * not yet delivered to THIS group (group last-delivered id → stream tail).
   * Both are "the backlog" in different senses; the E2E lag suite reads
   * `pending + lagEntries` as the composite backlog number.
   *
   * Optional: when absent (unit tests / older wiring) the endpoint reports
   * `available: false` and never fails readiness.
   */
  checkSearchIndexLag?: () => Promise<{ pending: number; lagEntries: number }>;
  /**
   * TASK-0135 — render the OTel Prometheus exposition (the worker's
   * queue_depth / queue_job_duration_seconds series + target_info). Returns
   * "" when OTel is disabled; the route answers 204 in that case so the
   * Prometheus scrape target is always present on the documented port.
   */
  metricsBody?: () => Promise<string>;
}

export interface WorkerHealthResult {
  status: "ready" | "not_ready";
  checks: {
    redis: boolean;
    queue: boolean;
  };
}

export interface SearchIndexLagResult {
  metric: "search-index:lag";
  stream: "factory:events";
  group: "worker-search-index";
  /** False when no probe is wired (or the Redis query failed). */
  available: boolean;
  /** Delivered to the group but not ACKed (in-flight / redelivery). */
  pending: number;
  /** Stream entries not yet delivered to the group (tail − last-delivered). */
  lagEntries: number;
  /** pending + lagEntries — the number the lag E2E asserts (the backlog). */
  backlog: number;
  timestamp: string;
}

/**
 * Build the health server. `readyFn` is called per readiness request (1 s
 * timeout enforced inside) so a hung dependency answers < 2 s. Liveness is
 * always 200 — it never probes dependencies.
 */
export function createHealthServer(deps: WorkerHealthDeps): Server {
  const startedAt = Date.now();

  const timeout = (ms: number) =>
    new Promise<never>((_, reject) =>
      setTimeout(() => reject(new Error("health check timeout")), ms).unref()
    );

  const ready = async (): Promise<WorkerHealthResult> => {
    const [redis, queue] = await Promise.all([
      withTimeout(deps.pingRedis(), timeout(1000)).catch(() => false),
      deps.checkQueue ? withTimeout(deps.checkQueue(), timeout(1000)).catch(() => false) : true
    ]);
    return {
      status: redis && queue ? "ready" : "not_ready",
      checks: { redis, queue }
    };
  };

  /**
   * `GET /health/search-index:lag` (TASK-0100 req 1, Backend Changes) — the
   * consumer-lag metric the lag E2E polls during a 1k-event burst.
   *
   * The probe is OBSERVABILITY, not readiness: it never gates
   * /health/ready. When the probe is not wired (deps.checkSearchIndexLag
   * absent) or Redis fails, the response is `available: false` with zeroed
   * counts — callers (the E2E) must treat that as "cannot measure" and skip
   * the assertion rather than fail the run.
   */
  const lag = async (): Promise<SearchIndexLagResult> => {
    const base: SearchIndexLagResult = {
      metric: "search-index:lag",
      stream: "factory:events",
      group: "worker-search-index",
      available: false,
      pending: 0,
      lagEntries: 0,
      backlog: 0,
      timestamp: new Date().toISOString()
    };
    if (!deps.checkSearchIndexLag) return base;
    try {
      const { pending, lagEntries } = await withTimeout(deps.checkSearchIndexLag(), timeout(1500));
      return { ...base, available: true, pending, lagEntries, backlog: pending + lagEntries };
    } catch {
      return base; // probe failure → "cannot measure", never 5xx (liveness stays 200)
    }
  };

  return createServer(async (req, res) => {
    const url = (req.url ?? "").split("?")[0];
    if (req.method !== "GET") {
      res.writeHead(405, { "content-type": "application/json" });
      res.end(JSON.stringify({ status: "method_not_allowed" }));
      return;
    }
    // Liveness at /api/v1/health (TASK-0009 contract); /health kept as an
    // alias for the documented manual curl.
    if (url === "/api/v1/health" || url === "/api/v1/health/" || url === "/health" || url === "/health/") {
      res.writeHead(200, { "content-type": "application/json" });
      res.end(
        JSON.stringify({
          status: "ok",
          app: APP_NAME,
          version: APP_VERSION,
          uptimeSeconds: Math.round((Date.now() - startedAt) / 1000),
          timestamp: new Date().toISOString()
        })
      );
      return;
    }
    if (url === "/health/ready") {
      const result = await ready();
      res.writeHead(result.status === "ready" ? 200 : 503, {
        "content-type": "application/json"
      });
      res.end(JSON.stringify(result));
      return;
    }
    if (url === "/health/search-index:lag") {
      res.writeHead(200, { "content-type": "application/json" });
      res.end(JSON.stringify(await lag()));
      return;
    }
    if (url === "/metrics") {
      const body = deps.metricsBody ? await deps.metricsBody().catch(() => "") : "";
      if (!body) {
        res.writeHead(204);
        res.end();
        return;
      }
      res.writeHead(200, { "content-type": "text/plain; version=0.0.4; charset=utf-8" });
      res.end(body);
      return;
    }
    res.writeHead(404, { "content-type": "application/json" });
    res.end(JSON.stringify({ status: "not_found" }));
  });
}

async function withTimeout<T>(p: Promise<T>, t: Promise<never>): Promise<T> {
  return Promise.race([p, t]);
}