// ENQUEUEING PUBLISH STAGES (release 18): one stage = one job on the editor's
// `publish` queue, run as a child process.
//
//   enqueueStage(paths, req)        one stage — refused (info) when the same
//                                   stage is already queued or running
//   enqueuePublishRun(paths, plan)  a whole plan at once (Publish now, the
//                                   site page's Build & deploy)
//
// Each job is `runManagedCommand` over `stageCommand(paths, req)` (S1's
// stageRun.ts: `tsx bin/archilyzer.ts stage <kind> <target> …`, cwd common/),
// so Cancel kills the child — and the child's tree (next build, wrangler,
// docker). The queue is ONE FIFO (`publish`, concurrency 1): stages run one at
// a time, beside the publish lock the child takes.
//
// ORDER IS ON DISK, NOT IN MEMORY: a run's builds carry `indexAfter` and its
// deploys `builtAfter` (the plan sets them), and a child whose precondition is
// not met when it starts exits 3 — the job ends `failed` with the sentence in
// its log. That is intended: nothing here chains one job to another. One
// window it leaves: a click's run (foreground) that finds the lane's index
// update already QUEUED (background — behind a job on another queue's slot,
// say) reuses it as `existing`, and its own build, foreground, can then start
// first and end "waiting for the index update this run started". The lane
// yields while any publish stage is queued, so the window is narrow; Retry
// re-enqueues the build.
//
// The spec IS the request: `{kind: "publish-<stage>", slug: target, params:
// {runId, …req}}` (`parseJobSpec` requires `slug`; there is no `channelSlug`),
// which is what Retry (editor jobReplayRegistry.ts) re-enqueues from.

import { getRegistry, type JobRecord } from "../jobs/registry";
import type { JobSpec } from "../jobs/jobSpec";
import { runManagedCommand, type StreamActionResult } from "../jobs/streamCommand";
import { getPaths, type Paths } from "../lib/paths";
import type { PlanSkip, PublishPlan } from "./publishPlan";
import { stageCommand } from "./stageRun";
import {
  STAGES,
  STAGE_FLAGS,
  deployKindOf,
  isStageKind,
  parseStageArgs,
  type StageRequest,
} from "./stages";
import { newStampId } from "./stamps";

export const PUBLISH_QUEUE = "publish";

/** A fresh run id (one Publish now, one lane pass): `<prefix>-<time>-<rand>`. */
export function newPublishRunId(prefix = "run", now = Date.now()): string {
  return `${prefix}-${newStampId(now)}`;
}

/**
 * What makes two stage requests the same stage: kind, target and — for a
 * deploy — its destination (production, local, or which preview). A
 * production deploy queued behind a preview of the same site is another stage.
 */
export function stageIdentity(r: Pick<StageRequest, "kind" | "target" | "preview" | "to">): string {
  const where = r.kind.startsWith("deploy") ? `|${deployKindOf(r)}|${r.preview ?? ""}` : "";
  return `${r.kind}|${r.target}${where}`;
}

/** The request a stage job's spec carries, or null when it is not one. */
export function stageRequestFromSpec(spec: JobSpec | null | undefined): StageRequest | null {
  if (!spec || !spec.kind.startsWith("publish-")) return null;
  const kind = spec.kind.slice("publish-".length);
  if (!isStageKind(kind)) return null;
  const p = spec.params ?? {};
  if (p.kind !== undefined && p.kind !== kind) return null;
  // Through the stage row's own parser: one definition of a valid request.
  const flags: Record<string, string | boolean | undefined> = {};
  for (const [flag, type] of Object.entries(STAGE_FLAGS)) {
    const key = flag.replace(/-([a-z])/g, (_, c: string) => c.toUpperCase());
    const v = p[key];
    if (v === undefined || v === null) continue;
    if (type === "boolean") {
      if (v === true) flags[flag] = true;
    } else if (typeof v === "string" || typeof v === "number") {
      flags[flag] = String(v);
    }
  }
  flags["run-id"] = typeof p.runId === "string" ? p.runId : undefined;
  const parsed = parseStageArgs([kind, spec.slug], flags);
  return "error" in parsed ? null : parsed;
}

/** The spec a stage job records (its replay descriptor). */
export function stageSpec(req: StageRequest): JobSpec {
  const params: Record<string, unknown> = {};
  for (const [k, v] of Object.entries(req)) if (v !== undefined) params[k] = v;
  return { kind: STAGES[req.kind].jobKind, slug: req.target, params };
}

/** A stage job already queued or running for the same stage, if any. */
export function findQueuedStage(
  records: readonly Pick<JobRecord, "id" | "kind" | "queueKey" | "status" | "spec">[],
  req: StageRequest,
): Pick<JobRecord, "id" | "status"> | undefined {
  const want = stageIdentity(req);
  return records.find((r) => {
    if (r.queueKey !== PUBLISH_QUEUE) return false;
    if (r.status !== "queued" && r.status !== "running") return false;
    if (r.kind !== STAGES[req.kind].jobKind) return false;
    const other = stageRequestFromSpec(r.spec);
    return other !== null && stageIdentity(other) === want;
  });
}

export type EnqueueStageResult =
  | Extract<StreamActionResult, { ok: true }>
  | { ok: false; error: string; info?: boolean; jobId?: string };

/**
 * Enqueue ONE stage on the `publish` queue. A duplicate (the same stage
 * already queued or running) is refused with `info: true` and the existing
 * job's id — nothing is enqueued twice. `background` puts it behind any
 * foreground stage (the lane's are background; a click is foreground).
 */
export async function enqueueStage(
  paths: Paths,
  req: StageRequest,
  opts: { background?: boolean; env?: NodeJS.ProcessEnv } = {},
): Promise<EnqueueStageResult> {
  const dup = findQueuedStage(getRegistry().list(), req);
  if (dup) {
    return {
      ok: false,
      info: true,
      jobId: dup.id,
      error: `${STAGES[req.kind].label} ${req.target} is already ${dup.status} (job ${dup.id})`,
    };
  }
  const cmd = stageCommand(paths, req, opts.env);
  return runManagedCommand({
    kind: STAGES[req.kind].jobKind,
    queueKey: PUBLISH_QUEUE,
    paths,
    spec: stageSpec(req),
    ...(opts.background ? { background: true } : {}),
    command: cmd.command,
    args: cmd.args,
    cwd: cmd.cwd,
    env: cmd.env,
  });
}

export type PublishRunJob = {
  kind: StageRequest["kind"];
  target: string;
  jobId: string;
  previewUrl?: string;
  // True when the same stage was already queued (its id is the existing job).
  existing?: boolean;
};

export type PublishRunResult = {
  runId: string;
  jobs: PublishRunJob[];
  skipped: PlanSkip[];
  refused: { kind: StageRequest["kind"]; target: string; error: string }[];
};

/**
 * Enqueue a whole plan at once, in its order, under one run id. Nobody reads
 * the jobs' streams here: each is released (its log is on disk), and the
 * caller follows the ids (`/api/jobs/<id>/log`, `--wait`).
 */
export async function enqueuePublishRun(
  paths: Paths = getPaths(),
  plan: PublishPlan,
  opts: {
    runId?: string;
    background?: boolean;
    env?: NodeJS.ProcessEnv;
    // The one-stage enqueue (tests inject it).
    enqueue?: typeof enqueueStage;
  } = {},
): Promise<PublishRunResult> {
  const enqueue = opts.enqueue ?? enqueueStage;
  const runId = opts.runId ?? newPublishRunId();
  const result: PublishRunResult = { runId, jobs: [], skipped: [...plan.skipped], refused: [] };
  for (const step of plan.steps) {
    const { reason: _reason, previewUrl, ...rest } = step;
    const req: StageRequest = { ...rest, runId };
    const res = await enqueue(paths, req, { background: opts.background, env: opts.env });
    if (res.ok) {
      void res.stream.cancel();
      result.jobs.push({ kind: req.kind, target: req.target, jobId: res.jobId, ...(previewUrl ? { previewUrl } : {}) });
    } else if (res.info && res.jobId) {
      result.jobs.push({
        kind: req.kind,
        target: req.target,
        jobId: res.jobId,
        existing: true,
        ...(previewUrl ? { previewUrl } : {}),
      });
    } else {
      result.refused.push({ kind: req.kind, target: req.target, error: res.error });
    }
  }
  return result;
}