import { Effect } from "effect";
import type {
  DeviceId,
  EngineId,
  HostProfile,
  InstanceRecord,
  InstanceState,
  LaunchFailure,
  EngineRuntimeKind,
  ServingOptions,
} from "./contracts";
import { fetchLocal } from "../../http/local-fetch";
import { applyDevices } from "./engines/devices";
import { engineSpec, planLaunch, supportsRuntime } from "./engines/registry";
import { toEvent } from "./failures";
import type { Launcher } from "./launchers/launcher";
import type { InstanceStore } from "./instances/store";

/**
 * The only mutator in the compute layer. Everything it knows lives in the instance
 * records; there is no coordinator state to go stale, and cancellation is one flag per
 * name rather than an abort controller + an intent serial + a boolean racing each other.
 */

const STOP_GRACE_MS = 20_000;

/** Operators tune cold-start budgets per box (large MoE + AOT compile can exceed any
 *  default); the legacy env override keeps working. */
const readyDeadlineOverrideMs = (): number | null => {
  const raw = process.env["LOCAL_STUDIO_READY_TIMEOUT_MS"];
  const parsed = raw ? Number(raw) : NaN;
  return Number.isFinite(parsed) && parsed > 0 ? parsed : null;
};
const HEALTH_PROBE_TIMEOUT_MS = 3_000;

export interface ComputeDeps {
  readonly store: InstanceStore;
  readonly launcherFor: (runtime: EngineRuntimeKind) => Launcher;
  readonly host: () => Effect.Effect<HostProfile>;
  readonly freeDevices: () => Effect.Effect<readonly DeviceId[]>;
  readonly onEvent: (name: string, stage: string, message: string) => Effect.Effect<void>;
}

export interface ComputeLaunchInput {
  readonly name: string;
  readonly engine: EngineId;
  readonly recipeId: string;
  readonly runtime: EngineRuntimeKind;
  readonly deviceCount: number;
  /** Pin the launch to these devices (recipe GPU selectors); default = any free. */
  readonly devices?: readonly DeviceId[];
  /** Serve on exactly this port (legacy inference_port); default = engine base scan. */
  readonly portOverride?: number;
  /** Verbatim launch argv (recipe custom launch command); replaces the engine plan. */
  readonly commandOverride?: readonly string[];
  readonly modelPath: string;
  readonly servedModelName: string;
  readonly options: ServingOptions;
  readonly extraArgs: readonly string[];
  readonly env: Readonly<Record<string, string>>;
  readonly dockerImage: string | null;
  readonly binary: string | null;
}

export interface InstanceView {
  readonly record: InstanceRecord;
  readonly state: InstanceState;
}

export interface ComputeService {
  readonly launch: (input: ComputeLaunchInput) => Effect.Effect<InstanceRecord, LaunchFailure>;
  readonly stop: (name: string) => Effect.Effect<boolean>;
  readonly cancel: (name: string) => Effect.Effect<boolean>;
  readonly stateOf: (record: InstanceRecord) => Effect.Effect<InstanceState>;
  readonly instances: () => Effect.Effect<readonly InstanceView[]>;
  /** One supervisor pass: drop records whose handle is gone. The only reaper. */
  readonly superviseOnce: () => Effect.Effect<number>;
}

export const makeComputeService = (deps: ComputeDeps): ComputeService => {
  const cancelRequested = new Set<string>();

  const launcherOf = (record: InstanceRecord): Launcher => deps.launcherFor(record.runtime);

  const recordAlive = (record: InstanceRecord): Effect.Effect<boolean> => {
    if (record.ref === null) return Effect.succeed(false);
    // Pinned holds have no supervised process; they live until explicitly released.
    if (record.ref.kind === "pinned") return Effect.succeed(true);
    return launcherOf(record).alive(record.ref);
  };

  const healthy = (record: InstanceRecord): Effect.Effect<boolean> =>
    Effect.gen(function* () {
      const spec = engineSpec(record.engine);
      const response = yield* fetchLocal(record.port, spec.health.path, {
        timeoutMs: HEALTH_PROBE_TIMEOUT_MS,
      }).pipe(Effect.catch(() => Effect.succeed(null)));
      return response !== null && response.ok;
    });

  /** Liveness first, then health, then the deadline. Never stored anywhere. */
  const stateOf = (record: InstanceRecord): Effect.Effect<InstanceState> =>
    Effect.gen(function* () {
      if (record.ref === null) return "reserving";
      if (!(yield* recordAlive(record))) return "exited";
      if (yield* healthy(record)) return "ready";
      return Date.now() < Date.parse(record.readyDeadlineAt) ? "starting" : "unhealthy";
    });

  const stopRecord = (record: InstanceRecord): Effect.Effect<void> =>
    Effect.gen(function* () {
      if (record.ref === null) return;
      const launcher = launcherOf(record);
      // Never signal what we cannot prove is ours — a recycled pid or a hand-recreated
      // container just gets its record dropped.
      if (yield* launcher.owns(record.ref, record)) {
        yield* launcher.stop(record.ref, STOP_GRACE_MS);
      }
    });

  const failCleanup = (
    record: InstanceRecord,
    failure: LaunchFailure,
  ): Effect.Effect<never, LaunchFailure> =>
    Effect.gen(function* () {
      yield* stopRecord(record);
      deps.store.drop(record.name);
      cancelRequested.delete(record.name);
      const event = toEvent(failure);
      yield* deps.onEvent(record.name, event.stage, event.message);
      return yield* Effect.fail(failure);
    });

  const waitReady = (record: InstanceRecord): Effect.Effect<void, LaunchFailure> =>
    Effect.gen(function* () {
      const spec = engineSpec(record.engine);
      const startedAt = Date.now();
      const deadline = Date.parse(record.readyDeadlineAt);
      while (Date.now() < deadline) {
        if (cancelRequested.has(record.name)) {
          return yield* failCleanup(record, { kind: "cancelled" });
        }
        // Liveness before health: if our daemon died, a 200 on this port is someone else.
        if (!(yield* recordAlive(record))) {
          const logTail =
            record.ref === null ? "" : yield* launcherOf(record).logTail(record.ref, record);
          return yield* failCleanup(record, {
            kind: "exited-early",
            exitCode: null,
            signal: null,
            logTail,
          });
        }
        if (yield* healthy(record)) return;
        yield* Effect.sleep(spec.health.intervalMs);
      }
      const logTail =
        record.ref === null ? "" : yield* launcherOf(record).logTail(record.ref, record);
      return yield* failCleanup(record, {
        kind: "unhealthy-timeout",
        waitedMs: Date.now() - startedAt,
        logTail,
      });
    });

  const launch = (input: ComputeLaunchInput): Effect.Effect<InstanceRecord, LaunchFailure> =>
    Effect.gen(function* () {
      const host = yield* deps.host();
      const spec = engineSpec(input.engine);
      const support = spec.supports(host);
      if (!support.ok) {
        return yield* Effect.fail<LaunchFailure>({
          kind: "unsupported",
          engine: input.engine,
          reason: support.reason,
        });
      }
      if (!supportsRuntime(input.engine, host, input.runtime)) {
        return yield* Effect.fail<LaunchFailure>({
          kind: "unsupported",
          engine: input.engine,
          reason: `runtime "${input.runtime}" not available (offers: ${support.runtimes.join(", ")})`,
        });
      }

      const existing = deps.store.read(input.name);
      if (existing) {
        const state = yield* stateOf(existing);
        if (state === "ready" || state === "starting" || state === "reserving") {
          return yield* Effect.fail<LaunchFailure>({ kind: "already-running", name: input.name });
        }
        // exited / unhealthy: reclaim the name.
        yield* stopRecord(existing);
        deps.store.drop(existing.name);
      }

      cancelRequested.delete(input.name);
      const candidates = input.devices ?? (yield* deps.freeDevices());
      const record = yield* deps.store.reserve(
        {
          name: input.name,
          nodeId: host.nodeId,
          engine: input.engine,
          recipeId: input.recipeId,
          runtime: input.runtime,
          candidates,
          need: input.devices
            ? input.devices.length
            : Math.min(input.deviceCount, Math.max(candidates.length, 0)),
          shareable: host.unifiedMemory && !input.devices,
          basePort: spec.defaultPort,
          ...(input.portOverride !== undefined ? { exactPort: input.portOverride } : {}),
          readyDeadlineMs: readyDeadlineOverrideMs() ?? spec.health.readyDeadlineMs,
        },
        recordAlive,
      );

      yield* deps.onEvent(record.name, "launching", `${input.engine} on :${record.port}`);

      const plan = input.commandOverride
        ? // A custom launch command is used verbatim — the recipe author owns the argv;
          // only device selection is still folded in.
          applyDevices(
            {
              kind: input.runtime,
              argv: [...input.commandOverride],
              env: input.env,
              ports: [{ container: record.port, host: record.port }],
              mounts: [],
              devices: record.devices,
              health: spec.health,
              ...(input.dockerImage ? { image: input.dockerImage } : {}),
            },
            host.accelerator,
          )
        : planLaunch({
            engine: input.engine,
            host,
            runtime: input.runtime,
            devices: record.devices,
            port: record.port,
            modelPath: input.modelPath,
            servedModelName: input.servedModelName,
            options: input.options,
            extraArgs: input.extraArgs,
            env: input.env,
            dockerImage: input.dockerImage,
            binary: input.binary ?? spec.defaultBinary,
          });

      const reference = yield* deps
        .launcherFor(input.runtime)
        .start(plan, record)
        .pipe(Effect.catch((failure) => failCleanup(record, failure)));

      const started: InstanceRecord = { ...record, ref: reference };
      deps.store.write(started);

      yield* waitReady(started);
      cancelRequested.delete(started.name);
      yield* deps.onEvent(started.name, "ready", `healthy on :${started.port}`);
      return started;
    });

  const stop = (name: string): Effect.Effect<boolean> =>
    Effect.gen(function* () {
      const record = deps.store.read(name);
      if (!record) return false;
      yield* stopRecord(record);
      deps.store.drop(name);
      yield* deps.onEvent(name, "stopped", `freed :${record.port}`);
      return true;
    });

  const cancel = (name: string): Effect.Effect<boolean> =>
    Effect.sync(() => {
      if (deps.store.read(name) === null) return false;
      cancelRequested.add(name);
      return true;
    });

  const instances = (): Effect.Effect<readonly InstanceView[]> =>
    Effect.gen(function* () {
      const views: InstanceView[] = [];
      for (const record of deps.store.all()) {
        views.push({ record, state: yield* stateOf(record) });
      }
      return views;
    });

  const superviseOnce = (): Effect.Effect<number> =>
    Effect.gen(function* () {
      let reaped = 0;
      for (const record of deps.store.all()) {
        // Pinned holds are freed by explicit release, never by the reaper.
        if (record.ref?.kind === "pinned") continue;
        // A reservation that never got a handle is a crashed launch; give it a minute.
        if (record.ref === null) {
          const age = Date.now() - Date.parse(record.startedAt);
          if (age > 60_000) {
            deps.store.drop(record.name);
            reaped += 1;
          }
          continue;
        }
        if (!(yield* recordAlive(record))) {
          // Dropping the record frees its devices by construction — there is no release
          // call to forget and no cache to invalidate.
          deps.store.drop(record.name);
          reaped += 1;
          yield* deps.onEvent(record.name, "exited", "process gone; record reaped");
        }
      }
      return reaped;
    });

  return { launch, stop, cancel, stateOf, instances, superviseOnce };
};