// PB ↔ AutoManager-typed conversions. The SDK passes raw protobuf
// payloads through for the new AutoManager events; AutoManager runs
// these converters at the boundary so the rest of the codebase works
// against the cleaner typed shapes from `./types`.
//
// Converters are deliberately permissive about defaults: an empty map
// from PB becomes `undefined` to match the optional-field convention
// used in the typed model, an absent ID wrapper stays the empty string.

import { create } from "@bufbuild/protobuf";
import { timestampDate, timestampFromDate } from "@bufbuild/protobuf/wkt";
import * as ManagerPB from "@norskvideo/norsk-api/lib/manager_pb";
import * as CommonPB from "@norskvideo/norsk-api/lib/shared/common_pb";
import {
  BundleIdSchema,
  CapabilityRequirementSchema,
  CapabilitySchema,
  OptionalIntSchema,
  OptionalStringSchema,
} from "@norskvideo/norsk-api/lib/shared/common_pb";
import { fromProductTemplateRef, toProductTemplateRef } from "@norskvideo/norsk-manager-sdk";
import { JobSchema as PbJobSchema } from "@norskvideo/norsk-api/lib/manager_pb";
import {
  Bundle,
  BundleId,
  BundleJobSpec,
  BundleState,
  Capability,
  CapabilityRequirement,
  AntiAffinity,
  GpuResource,
  IntraReplicaRules,
  JobRequirements,
  LaunchSpec,
  NodeAssignedJob,
  NodeInventory,
  OnLaunch,
  StartMode,
  OnRuntimeFailure,
  ResiliencePolicy,
} from "./types";

// ----- enums -----------------------------------------------------------

export function fromPbOnRuntimeFailure(
  pb: ManagerPB.OnRuntimeFailure
): OnRuntimeFailure {
  switch (pb) {
    case ManagerPB.OnRuntimeFailure.PREFER_SPARE:
      return "preferSpare";
    case ManagerPB.OnRuntimeFailure.AVOID_SPARE:
      return "avoidSpare";
    case ManagerPB.OnRuntimeFailure.DROP:
      return "drop";
  }
}

export function fromPbPlacementRule(
  pb: ManagerPB.PlacementRule
): "off" | "soft" | "hard" {
  switch (pb) {
    case ManagerPB.PlacementRule.OFF:
      return "off";
    case ManagerPB.PlacementRule.SOFT:
      return "soft";
    case ManagerPB.PlacementRule.HARD:
      return "hard";
  }
}

// ----- capabilities ----------------------------------------------------

export function fromPbCapability(pb: CommonPB.Capability): Capability {
  return {
    name: pb.name,
    count: pb.count,
    attributes: emptyMapToUndefined(pb.attributes),
  };
}

export function fromPbCapabilityRequirement(
  pb: CommonPB.CapabilityRequirement
): CapabilityRequirement {
  return {
    name: pb.name,
    count: pb.count,
    attributeMatches: emptyMapToUndefined(pb.attributeMatches),
  };
}

// ----- GPU -------------------------------------------------------------

export function fromPbGpuResource(pb: CommonPB.GpuResource): GpuResource {
  return {
    index: pb.index,
    model: pb.model?.value,
    totalCapacity: pb.totalCapacity,
    reservedCapacity: pb.reservedCapacity,
    exclusivelyReserved: pb.exclusivelyReserved,
  };
}

// ----- requirements ----------------------------------------------------

export function fromPbJobRequirements(
  pb: ManagerPB.JobRequirements
): JobRequirements {
  return {
    requiredCapacity: pb.requiredCapacity,
    requiredCores: pb.requiredCores?.value,
    requiredGpuCapacity: pb.requiredGpuCapacity?.value,
    requiredGpuModel: pb.requiredGpuModel?.value,
    requiredCapabilities: pb.requiredCapabilities.map(fromPbCapabilityRequirement),
    requiredHostPorts: emptyArrayToUndefined(pb.requiredHostPorts),
    onRuntimeFailure: fromPbOnRuntimeFailure(pb.onRuntimeFailure),
    ...(pb.maxRestartsPerHour ? { maxRestartsPerHour: pb.maxRestartsPerHour.value } : {}),
    ...(pb.essential ? { essential: pb.essential.value } : {}),
  };
}

// ----- bundle ----------------------------------------------------------

export function fromPbBundle(pb: ManagerPB.Bundle): Bundle {
  return {
    bundleId: bundleIdString(pb.bundleId),
    pool: pb.pool,
    jobs: pb.jobs.map(fromPbBundleJobSpec),
    intraReplicaPlacement: pb.intraReplicaPlacement
      ? fromPbIntraReplicaRules(pb.intraReplicaPlacement)
      : undefined,
    resiliencePolicy: pb.resiliencePolicy ? fromPbResiliencePolicy(pb.resiliencePolicy) : undefined,
    state: fromPbBundleState(pb.state),
    onLaunch: fromPbOnLaunch(pb.onLaunch),
    ...(pb.jobs.some((j) => j.job?.tags?.[OVERRIDE_RESERVATION_TAG] === "true")
      ? { overrideReservation: true }
      : {}),
  };
}

export function fromPbOnLaunch(pb: ManagerPB.OnLaunch): OnLaunch {
  switch (pb) {
    case ManagerPB.OnLaunch.ALLOW_SPARE:
      return "allowSpare";
    default:
      return "avoidSpare";
  }
}

export function toPbOnLaunch(v: OnLaunch): PbOnLaunch {
  switch (v) {
    case "allowSpare":
      return PbOnLaunch.ALLOW_SPARE;
    case "avoidSpare":
      return PbOnLaunch.AVOID_SPARE;
  }
}

export function fromPbBundleState(pb: ManagerPB.BundleState): BundleState {
  switch (pb) {
    case ManagerPB.BundleState.DELETING:
      return "deleting";
    case ManagerPB.BundleState.FAILED:
      return "failed";
    default:
      return "live";
  }
}

export function toPbBundleState(s: BundleState): ManagerPB.BundleState {
  switch (s) {
    case "deleting":
      return ManagerPB.BundleState.DELETING;
    case "failed":
      return ManagerPB.BundleState.FAILED;
    case "live":
      return ManagerPB.BundleState.LIVE;
  }
}

/** The daemon's own reading (`apiToStartMode`): "manual", else auto —
 *  the empty string a pre-startMode bundle carries included. */
export function fromPbStartMode(pb: string): StartMode {
  return pb === "manual" ? "manual" : "auto";
}

export function toPbStartMode(v: StartMode): string {
  switch (v) {
    case "auto":
      return "auto";
    case "manual":
      return "manual";
  }
}

export function fromPbBundleJobSpec(
  pb: ManagerPB.BundleJobSpec
): BundleJobSpec {
  return {
    jobName: pb.jobName,
    requirements: pb.requirements
      ? fromPbJobRequirements(pb.requirements)
      : ({} as JobRequirements),
    startMode: fromPbStartMode(pb.startMode),
    coLocateWith: emptyArrayToUndefined(pb.coLocateWith),
    separateFrom: emptyArrayToUndefined(pb.separateFrom),
    ...(pb.exclusive ? { exclusive: true } : {}),
    ...(pb.singleUse ? { singleUse: true } : {}),
    ...(Object.keys(pb.backupParameterOverrides).length > 0
      ? { backupParameterOverrides: pb.backupParameterOverrides }
      : {}),
    launch: pb.job ? fromPbLaunchSpec(pb.job) : undefined,
  };
}

/**
 * Reconstruct the typed `LaunchSpec` from the wire-form `Job` carried
 * on a `BundleJobSpec`. If the carried Job has `productTemplate` set, this
 * is a product-template launch; otherwise it's the legacy services/volumes
 * path. The legacy variant passes the Job through as-is (cast to the
 * SDK's typed shape — same trick the v0 code used).
 */
/** Tags starting with this prefix are SDK-managed (well-known reserved
 *  names) — `__stopDateTime` carries the auto-stop instant for example.
 *  Callers can't set them via `LaunchSpec.tags` (silently stripped on
 *  outbound) and they're surfaced as typed fields rather than raw tags
 *  on inbound. */
export const RESERVED_TAG_PREFIX = "__";
/** Well-known reserved tag carrying the operator's submitted auto-stop
 *  instant. The daemon has no first-class concept of stop time, so we
 *  ride along on its tag-preservation behaviour. */
export const STOP_DATE_TIME_TAG = "__stopDateTime";
/** Phase G — well-known reserved tag carrying the bundle's
 *  `overrideReservation` flag. Bundle-level in the typed API, but Manager
 *  has no bundle field for it, so it rides on each product-template job's
 *  tags exactly as `stopDateTime` does. Read back off whichever job
 *  carries it. */
export const OVERRIDE_RESERVATION_TAG = "__overrideReservation";

function buildJobTags(
  stopDateTime: Date | undefined,
  userTags: Record<string, string> | undefined,
): Record<string, string> {
  const out: Record<string, string> = {};
  if (userTags) {
    for (const [k, v] of Object.entries(userTags)) {
      if (k.startsWith(RESERVED_TAG_PREFIX)) continue; // silently strip
      out[k] = v;
    }
  }
  if (stopDateTime) out[STOP_DATE_TIME_TAG] = stopDateTime.toISOString();
  return out;
}

function splitJobTags(tags: Record<string, string>): {
  stopDateTime?: Date;
  userTags?: Record<string, string>;
} {
  const userTags: Record<string, string> = {};
  let stopDateTime: Date | undefined;
  for (const [k, v] of Object.entries(tags)) {
    if (k === STOP_DATE_TIME_TAG) {
      const d = new Date(v);
      if (!Number.isNaN(d.getTime())) stopDateTime = d;
      continue;
    }
    if (k.startsWith(RESERVED_TAG_PREFIX)) continue; // unknown reserved — hide
    userTags[k] = v;
  }
  return {
    ...(stopDateTime ? { stopDateTime } : {}),
    ...(Object.keys(userTags).length > 0 ? { userTags } : {}),
  };
}

export function fromPbLaunchSpec(pbJob: ManagerPB.Job): LaunchSpec {
  if (pbJob.productTemplate) {
    const { stopDateTime, userTags } = splitJobTags(pbJob.tags ?? {});
    return {
      kind: "productTemplate",
      ref: fromProductTemplateRef(pbJob.productTemplate),
      parameters:
        Object.keys(pbJob.productTemplateParameters).length > 0
          ? pbJob.productTemplateParameters
          : undefined,
      // Preserve the operator's submitted schedule so consumers can
      // surface it even before a JobPending event has been streamed
      // (e.g. bundles in mgr's bundleMap whose materialised jobs
      // haven't arrived yet — calendar view depends on this).
      ...(pbJob.startDateTime ? { startDateTime: timestampDate(pbJob.startDateTime) } : {}),
      ...(stopDateTime ? { stopDateTime } : {}),
      ...(userTags ? { tags: userTags } : {}),
      ...(pbJob.launchConfigJson?.value !== undefined
        ? { launchConfigJson: pbJob.launchConfigJson.value }
        : {}),
      ...(pbJob.templateLaunchDefaultsJson?.value !== undefined
        ? { templateLaunchDefaultsJson: pbJob.templateLaunchDefaultsJson.value }
        : {}),
      ...(pbJob.extraSecurityGroups.length > 0
        ? { extraSecurityGroups: pbJob.extraSecurityGroups }
        : {}),
    };
  }
  return { kind: "legacy", job: pbJob as never };
}

export function fromPbIntraReplicaRules(
  pb: ManagerPB.IntraReplicaRules
): IntraReplicaRules {
  return {
    sameNodeDefault: pb.sameNodeDefault,
    sameAz: fromPbPlacementRule(pb.sameAz),
  };
}

export function fromPbResiliencePolicy(pb: ManagerPB.ResiliencePolicy): ResiliencePolicy {
  return {
    ...(pb.pool.length > 0 ? { pool: pb.pool } : {}),
    sameNode: fromPbAntiAffinity(pb.sameNode),
    sameAz: fromPbAntiAffinity(pb.sameAz),
    sameCloud: fromPbAntiAffinity(pb.sameCloud),
  };
}

function fromPbAntiAffinity(pb: ManagerPB.AntiAffinity): AntiAffinity {
  return pb === ManagerPB.AntiAffinity.FORBID ? "forbid" : "allow";
}

function toPbAntiAffinity(a: AntiAffinity): ManagerPB.AntiAffinity {
  return a === "forbid" ? ManagerPB.AntiAffinity.FORBID : ManagerPB.AntiAffinity.ALLOW;
}

// ----- inventory -------------------------------------------------------

/**
 * A wire inventory field whose value lay outside the contract and was
 * replaced at the boundary. `advertised` is what the producer sent, `used`
 * what the typed inventory carries.
 *
 * @public
 */
export interface InventoryContractViolation {
  field: "utilizationCap";
  advertised: number;
  used: number;
}

/**
 * `utilization_cap` is a FRACTION (0-1) of `total_capacity`: placement
 * computes `total * cap - reserved`, the UI draws the cap line when it is
 * below 1, and the design default is 0.8. A producer that sends anything
 * else — a pre-fix norsk-ctl worker sent its vCPU count, so a 16-vCPU node
 * advertised 256 units free and every production packed onto it — is
 * read as 1 (no soft cap): the node's declared capacity, never more. Zero,
 * negative and NaN are not caps either and read the same way. The
 * violation is handed to `onViolation` so the caller can say so once.
 */
function contractUtilizationCap(
  advertised: number,
  onViolation: ((v: InventoryContractViolation) => void) | undefined
): number {
  if (Number.isFinite(advertised) && advertised > 0 && advertised <= 1) return advertised;
  onViolation?.({ field: "utilizationCap", advertised, used: 1 });
  return 1;
}

export function fromPbNodeInventory(
  pb: CommonPB.NodeInventory,
  onViolation?: (v: InventoryContractViolation) => void
): NodeInventory {
  return {
    totalCapacity: pb.totalCapacity,
    utilizationCap: contractUtilizationCap(pb.utilizationCap, onViolation),
    totalCores: pb.totalCores,
    gpus: pb.gpus.map(fromPbGpuResource),
    capabilities: pb.capabilities.map(fromPbCapability),
    reservedCapacity: pb.reservedCapacity,
    reservedCores: pb.reservedCores,
    reservedCapabilities: pb.reservedCapabilities.map(fromPbCapability),
    reachable: pb.reachable,
    lastSeen: pb.lastSeen
      ? new Date(Number(pb.lastSeen.seconds) * 1000 + (pb.lastSeen.nanos ?? 0) / 1_000_000)
      : new Date(0),
    cordoned: pb.cordoned,
    assignedJobs: pb.assignedJobs.map(fromPbNodeAssignedJob),
    reservedHostPorts: emptyArrayToUndefined(pb.reservedHostPorts),
    // Phase G — what the node advertises holding, i.e. its warmth.
    heldImages: emptyArrayToUndefined(
      pb.heldImages.map((h) => ({ imageRef: h.imageRef, tags: [...h.tags] }))
    ),
  };
}

export function fromPbNodeAssignedJob(
  pb: CommonPB.NodeAssignedJob
): NodeAssignedJob {
  return {
    jobId: jobIdString(pb.jobId),
    coreSet: pb.coreSet.length > 0 ? [...pb.coreSet] : undefined,
    gpuIndex: pb.gpuIndex?.value,
  };
}

// ----- helpers ---------------------------------------------------------

function bundleIdString(pb?: { bundleId: string }): BundleId {
  return pb?.bundleId ?? "";
}

function jobIdString(pb?: { jobId: string }): string {
  return pb?.jobId ?? "";
}

function emptyMapToUndefined<V>(
  m: Record<string, V>
): Record<string, V> | undefined {
  return Object.keys(m).length > 0 ? { ...m } : undefined;
}

function emptyArrayToUndefined<V>(arr: V[]): V[] | undefined {
  return arr.length > 0 ? [...arr] : undefined;
}

// =============================================================================
// AutoManager-typed → PB conversions. Used when AutoManager submits a Bundle
// (or future request types) over the wire.
// =============================================================================

import {
  BundleSchema,
  BundleJobSpecSchema,
  IntraReplicaRulesSchema,
  ResiliencePolicySchema,
  JobRequirementsSchema,
  PlacementRule,
  OnLaunch as PbOnLaunch,
  OnRuntimeFailure as PbOnRuntimeFailure,
} from "@norskvideo/norsk-api/lib/manager_pb";
import { OptionalFloatSchema } from "@norskvideo/norsk-api/lib/shared/common_pb";

export function toPbOnRuntimeFailure(b: OnRuntimeFailure): PbOnRuntimeFailure {
  switch (b) {
    case "preferSpare":
      return PbOnRuntimeFailure.PREFER_SPARE;
    case "avoidSpare":
      return PbOnRuntimeFailure.AVOID_SPARE;
    case "drop":
      return PbOnRuntimeFailure.DROP;
  }
}

export function toPbPlacementRule(r: "off" | "soft" | "hard"): PlacementRule {
  switch (r) {
    case "off":
      return PlacementRule.OFF;
    case "soft":
      return PlacementRule.SOFT;
    case "hard":
      return PlacementRule.HARD;
  }
}

export function toPbCapability(c: Capability): CommonPB.Capability {
  return create(CapabilitySchema, {
    name: c.name,
    count: c.count,
    attributes: c.attributes ?? {},
  });
}

export function toPbCapabilityRequirement(
  r: CapabilityRequirement
): CommonPB.CapabilityRequirement {
  return create(CapabilityRequirementSchema, {
    name: r.name,
    count: r.count,
    attributeMatches: r.attributeMatches ?? {},
  });
}

export function toPbJobRequirements(
  r: JobRequirements
): ManagerPB.JobRequirements {
  return create(JobRequirementsSchema, {
    requiredCapacity: r.requiredCapacity,
    requiredCores:
      r.requiredCores !== undefined
        ? create(OptionalIntSchema, { value: r.requiredCores })
        : undefined,
    requiredGpuCapacity:
      r.requiredGpuCapacity !== undefined
        ? create(OptionalFloatSchema, { value: r.requiredGpuCapacity })
        : undefined,
    requiredGpuModel:
      r.requiredGpuModel !== undefined
        ? create(OptionalStringSchema, { value: r.requiredGpuModel })
        : undefined,
    requiredCapabilities: r.requiredCapabilities.map(toPbCapabilityRequirement),
    requiredHostPorts: r.requiredHostPorts ?? [],
    onRuntimeFailure: toPbOnRuntimeFailure(r.onRuntimeFailure),
    ...(r.maxRestartsPerHour !== undefined ? { maxRestartsPerHour: { value: r.maxRestartsPerHour } } : {}),
    ...(r.essential !== undefined ? { essential: { value: r.essential } } : {}),
  });
}

export function toPbBundleJobSpec(j: BundleJobSpec): ManagerPB.BundleJobSpec {
  return create(BundleJobSpecSchema, {
    jobName: j.jobName,
    requirements: toPbJobRequirements(j.requirements),
    startMode: toPbStartMode(j.startMode ?? "auto"),
    coLocateWith: j.coLocateWith ?? [],
    separateFrom: j.separateFrom ?? [],
    exclusive: j.exclusive ?? false,
    singleUse: j.singleUse ?? false,
    backupParameterOverrides: j.backupParameterOverrides ?? {},
    job: j.launch ? toPbLaunchSpec(j.launch) : undefined,
  });
}

/**
 * Project a typed `LaunchSpec` into the wire-form `Job` slot on
 * `BundleJobSpec`. Manager stamps this template per replica during
 * bundle expansion (setting jobId, tags, state, currentHash). For the
 * product-template variant only `productTemplate`, `productTemplateParameters`, and
 * `startDateTime` are meaningful — the rest stays at proto defaults.
 * The worker reads `productTemplate` + `productTemplateParameters` and ignores
 * services / volumes / writeFiles.
 *
 * `startDateTime` defaults to `new Date()` (immediate) when callers
 * omit it; Manager-daemon's `apiToJobTemplate` rejects unset values.
 */
export function toPbLaunchSpec(launch: LaunchSpec): ManagerPB.Job {
  if (launch.kind === "productTemplate") {
    const tags = buildJobTags(launch.stopDateTime, launch.tags);
    return create(PbJobSchema, {
      startDateTime: timestampFromDate(launch.startDateTime ?? new Date()),
      productTemplate: toProductTemplateRef(launch.ref),
      productTemplateParameters: launch.parameters ?? {},
      ...(Object.keys(tags).length > 0 ? { tags } : {}),
      ...(launch.launchConfigJson !== undefined
        ? { launchConfigJson: create(OptionalStringSchema, { value: launch.launchConfigJson }) }
        : {}),
      ...(launch.templateLaunchDefaultsJson !== undefined
        ? {
            templateLaunchDefaultsJson: create(OptionalStringSchema, {
              value: launch.templateLaunchDefaultsJson,
            }),
          }
        : {}),
      ...(launch.extraSecurityGroups?.length ? { extraSecurityGroups: launch.extraSecurityGroups } : {}),
    });
  }
  return launch.job as never;
}

export function toPbIntraReplicaRules(
  r: IntraReplicaRules
): ManagerPB.IntraReplicaRules {
  return create(IntraReplicaRulesSchema, {
    sameNodeDefault: r.sameNodeDefault,
    sameAz: toPbPlacementRule(r.sameAz),
  });
}

export function toPbResiliencePolicy(r: ResiliencePolicy): ManagerPB.ResiliencePolicy {
  return create(ResiliencePolicySchema, {
    pool: r.pool ?? "",
    sameNode: toPbAntiAffinity(r.sameNode),
    sameAz: toPbAntiAffinity(r.sameAz),
    sameCloud: toPbAntiAffinity(r.sameCloud),
  });
}

/** Phase G — carry the bundle-level `overrideReservation` on each
 *  product-template job's tags (see OVERRIDE_RESERVATION_TAG). A legacy
 *  job has no tag map of ours to write into and is left alone. */
function stampOverrideReservation(
  spec: ManagerPB.BundleJobSpec,
  overrideReservation: boolean | undefined
): ManagerPB.BundleJobSpec {
  if (overrideReservation !== true || !spec.job?.productTemplate) return spec;
  spec.job.tags = { ...(spec.job.tags ?? {}), [OVERRIDE_RESERVATION_TAG]: "true" };
  return spec;
}

export function toPbBundle(b: Bundle): ManagerPB.Bundle {
  return create(BundleSchema, {
    bundleId: create(BundleIdSchema, { bundleId: b.bundleId }),
    pool: b.pool,
    jobs: b.jobs.map((j) => stampOverrideReservation(toPbBundleJobSpec(j), b.overrideReservation)),
    intraReplicaPlacement: b.intraReplicaPlacement
      ? toPbIntraReplicaRules(b.intraReplicaPlacement)
      : undefined,
    resiliencePolicy: b.resiliencePolicy ? toPbResiliencePolicy(b.resiliencePolicy) : undefined,
    state: toPbBundleState(b.state ?? "live"),
    onLaunch: toPbOnLaunch(b.onLaunch ?? "avoidSpare"),
  });
}