// Phase F — NodeService reconciliation, split out of automanager.ts. A
// node service is a long-running job placed on EVERY eligible node of its
// pools (the only lifecycle, `eager`); this reconciler keeps the placed set
// equal to the eligible set. It reads the fleet through the narrow view
// below and writes only its own placements plus the job binding of each
// runtime it places — the AutoManager owns the timers that call it.
//
// Fenced behind `AutoSettings.experimental.nodeServices` (settings
// validation): what exists is a working F1 and nothing more — see
// norsk-mgr docs/internal/design/node-services.md.

import * as uuid from "uuid";
import type { JobId, NodeId, NorskManager } from "@norskvideo/norsk-manager-sdk";
import type * as ManagerSdk from "@norskvideo/norsk-manager-sdk";
import type { NodeSummary } from "./automanager";
import { debuglog } from "./debuglog";
import type { JobBindings } from "./jobBinding";
import type { CapabilityRequirement, NodeInventory as TypedNodeInventory, NodeService, ServiceId } from "./types";

/**
 * Fixed role assigned to runtime jobs that exist to back a
 * NodeService. Worker reads `WorkerJob.role === NODE_SERVICE_ROLE`
 * to recognise the instance as a service rather than a regular
 * workload, and adds the service's `provides[]` capabilities to its
 * inventory once the job reaches `ready`.
 *
 * @public
 */
export const NODE_SERVICE_ROLE = "nodeService";

/** What the reconciler needs from the AutoManager, handed in explicitly. */
export interface NodeServiceDeps {
  /** The configured services (`settings.nodeServices`), read per reconcile. */
  services: () => NodeService[];
  /** False before hydration and after close: nothing is placed then. */
  active: () => boolean;
  nodes: () => Iterable<NodeSummary>;
  inventoryOf: (nodeId: NodeId) => TypedNodeInventory | undefined;
  poolNameForNode: (nodeId: NodeId) => string;
  /** Cordoned by the operator (`cordonNode`), as opposed to by the worker. */
  isCordoned: (nodeId: NodeId) => boolean;
  norsk: Pick<NorskManager, "createJob" | "startJob" | "stopJob" | "deleteJob">;
  bindings: JobBindings;
  onError: (err: unknown) => void;
}

export class NodeServiceReconciler {
  // Per service, the set of (nodeId → runtime jobId) placed on it. Kept in
  // step with the configured eligible set by `reconcile`.
  private readonly placements: Map<ServiceId, Map<NodeId, JobId>> = new Map();

  constructor(private readonly deps: NodeServiceDeps) {}

  services(): NodeService[] {
    return this.deps.services();
  }

  /** Nodes on which the named service is placed, keyed by runtime jobId. */
  placementsFor(serviceId: ServiceId): Map<NodeId, JobId> {
    return new Map(this.placements.get(serviceId) ?? []);
  }

  /**
   * Drive every configured service toward its lifecycle target: place on
   * every eligible node that does not host it, tear down every placement
   * whose node is no longer eligible. Idempotent; safe on every event.
   */
  reconcile(): void {
    if (!this.deps.active()) return;
    const services = this.deps.services();
    if (services.length === 0) return;
    // Only `eager` reaches here: settings validation refuses any other
    // lifecycle at start-up.
    for (const svc of services) this.reconcileEager(svc);
  }

  /** The node is gone (stopping / stopped): its placements with it. */
  dropNode(nodeId: NodeId): void {
    for (const placements of this.placements.values()) placements.delete(nodeId);
  }

  /** Hydration: the fleet view is being rebuilt from the initial state. */
  clear(): void {
    this.placements.clear();
  }

  private reconcileEager(svc: NodeService): void {
    const eligible = this.eligibleNodesFor(svc);
    const placed = this.placements.get(svc.serviceId) ?? new Map<NodeId, JobId>();

    // Place on every eligible node that doesn't already host the service.
    for (const nodeId of eligible) {
      if (placed.has(nodeId)) continue;
      void this.place(svc, nodeId).catch((err) => this.deps.onError(err));
    }

    // Tear down any placement that's no longer eligible.
    for (const [nodeId, jobId] of placed) {
      if (eligible.has(nodeId)) continue;
      void this.teardown(svc.serviceId, nodeId, jobId).catch((err) => this.deps.onError(err));
    }
  }

  /**
   * Nodes in the service's named pools whose current inventory satisfies
   * its `consumes` requirements. Used both for placement and for
   * tear-down decisions.
   */
  private eligibleNodesFor(svc: NodeService): Set<NodeId> {
    const eligible = new Set<NodeId>();
    const poolSet = new Set(svc.pools);
    for (const node of this.deps.nodes()) {
      if (node.nodeState !== "running") continue;
      if (!poolSet.has(this.deps.poolNameForNode(node.nodeId))) continue;
      const inv = this.deps.inventoryOf(node.nodeId);
      if (!inv || !inv.reachable || inv.cordoned) continue;
      if (this.deps.isCordoned(node.nodeId)) continue;
      if (nodeSatisfiesConsumes(inv, svc.consumes)) eligible.add(node.nodeId);
    }
    return eligible;
  }

  private async place(svc: NodeService, nodeId: NodeId): Promise<void> {
    const jobId = `${svc.serviceId}.${uuid.v4().slice(0, 8)}`;
    debuglog("Placing NodeService", { serviceId: svc.serviceId, nodeId, jobId });
    // Optimistically record the placement so a re-entrant reconcile
    // doesn't double-place.
    let placements = this.placements.get(svc.serviceId);
    if (!placements) {
      placements = new Map();
      this.placements.set(svc.serviceId, placements);
    }
    placements.set(nodeId, jobId);
    // Build the runtime Job: take the configured template, override
    // the jobId, and tag it as a NodeService instance.
    const job: ManagerSdk.Job = {
      ...svc.jobSpec,
      jobId,
      tags: {
        ...(svc.jobSpec.tags ?? {}),
        "auto-manager": "true",
        nodeService: svc.serviceId,
      },
    };
    let createdJob = false;
    try {
      await this.deps.norsk.createJob(job);
      createdJob = true;
      await this.deps.norsk.startJob(jobId, NODE_SERVICE_ROLE, nodeId);
      this.deps.bindings.place(jobId, nodeId);
    } catch (err) {
      // Roll back the optimistic placement on failure.
      placements.delete(nodeId);
      // R9: if createJob succeeded but startJob failed, the Manager
      // has a dangling Job row that the next reconcile would orphan
      // by allocating a fresh jobId. Best-effort delete.
      if (createdJob) {
        try {
          await this.deps.norsk.deleteJob(jobId);
        } catch (cleanupErr) {
          debuglog("placeNodeService: cleanup deleteJob failed", { jobId, err: String(cleanupErr) });
        }
      }
      this.deps.onError(err);
    }
  }

  private async teardown(serviceId: ServiceId, nodeId: NodeId, jobId: JobId): Promise<void> {
    debuglog("Tearing down NodeService instance", { serviceId, nodeId, jobId });
    this.placements.get(serviceId)?.delete(nodeId);
    try {
      await this.deps.norsk.stopJob(jobId, NODE_SERVICE_ROLE, nodeId);
    } catch (err) {
      // Job might be gone already; tolerate.
      debuglog("NodeService stopJob failed (continuing)", { serviceId, err: String(err) });
    }
  }
}

function nodeSatisfiesConsumes(inv: TypedNodeInventory, consumes: CapabilityRequirement[]): boolean {
  for (const req of consumes) {
    const cap = inv.capabilities.find((c) => c.name === req.name);
    if (!cap) return false;
    const reserved = inv.reservedCapabilities.find((c) => c.name === req.name)?.count ?? 0;
    if (cap.count - reserved < req.count) return false;
    if (req.attributeMatches) {
      for (const [k, v] of Object.entries(req.attributeMatches)) {
        if (cap.attributes?.[k] !== v) return false;
      }
    }
  }
  return true;
}