vfxreview/lib/render-pipeline/machines.ts

import { Machine, Prisma } from "@prisma/client";
import { db } from "@/lib/db";
import { PipelineError } from "./errors";
import { getWorkerConfig } from "./config";

/**
 * Machine availability (§7.10):
 *   ALWAYS   — claims whenever idle (default when availability is null)
 *   SCHEDULE — claims only inside configured windows
 *   MANUAL   — never claims unless explicitly triggered
 * Overrides, in priority order: enabled=false beats everything; renderNowUntil
 * allows normal claims until it expires; urgent jobs may claim anytime on
 * machines with allowUrgentAnytime.
 */
export interface AvailabilityWindow {
  days: string[]; // ["mon", ..., "sun"]
  from: string; // "19:00"
  to: string; // "08:00" — from > to spans midnight
}

export interface MachineAvailability {
  mode?: "ALWAYS" | "SCHEDULE" | "MANUAL";
  windows?: AvailabilityWindow[];
  allowUrgentAnytime?: boolean;
}

const DAY_NAMES = ["sun", "mon", "tue", "wed", "thu", "fri", "sat"];

function parseHm(hm: string): number | null {
  const m = /^(\d{1,2}):(\d{2})$/.exec(hm);
  if (!m) return null;
  const mins = parseInt(m[1], 10) * 60 + parseInt(m[2], 10);
  return mins >= 0 && mins <= 24 * 60 ? mins : null;
}

export function isInWindow(windows: AvailabilityWindow[], now: Date): boolean {
  const day = DAY_NAMES[now.getDay()];
  const prevDay = DAY_NAMES[(now.getDay() + 6) % 7];
  const t = now.getHours() * 60 + now.getMinutes();

  for (const w of windows) {
    const from = parseHm(w.from);
    const to = parseHm(w.to);
    if (from == null || to == null || !Array.isArray(w.days)) continue;
    const days = w.days.map((d) => d.toLowerCase().slice(0, 3));
    if (from < to) {
      if (days.includes(day) && t >= from && t < to) return true;
    } else if (from > to) {
      // overnight window, e.g. 19:00 → 08:00
      if (days.includes(day) && t >= from) return true;
      if (days.includes(prevDay) && t < to) return true;
    } else {
      // from === to → full 24 h on listed days
      if (days.includes(day)) return true;
    }
  }
  return false;
}

export type ClaimGate = "normal" | "urgent-only" | "none";

export function machineCanClaim(machine: Machine, now: Date): ClaimGate {
  if (!machine.enabled) return "none";
  if (machine.renderNowUntil && machine.renderNowUntil > now) return "normal";

  const availability = (machine.availability ?? null) as MachineAvailability | null;
  const mode = availability?.mode ?? "ALWAYS";
  const urgent = availability?.allowUrgentAnytime ? ("urgent-only" as const) : ("none" as const);

  switch (mode) {
    case "ALWAYS":
      return "normal";
    case "SCHEDULE":
      return isInWindow(availability?.windows ?? [], now) ? "normal" : urgent;
    case "MANUAL":
      return urgent;
    default:
      return "normal";
  }
}

/** E6 — idempotent register/upsert on machine name. */
export async function registerMachine(body: {
  name?: string;
  hostname?: string;
  workerVersion?: string;
  aeVersion?: string;
  capabilities?: unknown;
}) {
  const name = body.name?.trim();
  if (!name) throw new PipelineError(422, "name is required");
  const hostname = body.hostname?.trim() || name;

  const machine = await db.machine.upsert({
    where: { name },
    create: {
      name,
      hostname,
      status: "ONLINE",
      lastSeenAt: new Date(),
      workerVersion: body.workerVersion ?? null,
      aeVersion: body.aeVersion ?? null,
      capabilities: (body.capabilities ?? undefined) as Prisma.InputJsonValue | undefined,
    },
    update: {
      hostname,
      status: "ONLINE",
      lastSeenAt: new Date(),
      workerVersion: body.workerVersion ?? undefined,
      aeVersion: body.aeVersion ?? undefined,
      capabilities: (body.capabilities ?? undefined) as Prisma.InputJsonValue | undefined,
    },
  });

  return {
    machine: { id: machine.id, name: machine.name, enabled: machine.enabled },
    config: await getWorkerConfig(),
  };
}

/** E7 — heartbeat; also the server→worker command channel (cancel signals). */
export async function recordHeartbeat(
  machineId: string,
  body: { cpuPercent?: number; memPercent?: number; diskFreeGb?: number; currentJobId?: string | null }
) {
  const machine = await db.machine.findUnique({ where: { id: machineId } });
  if (!machine) throw new PipelineError(404, "Machine not found");

  await db.$transaction([
    db.machine.update({
      where: { id: machineId },
      data: { lastSeenAt: new Date(), status: machine.enabled ? "ONLINE" : "DISABLED" },
    }),
    db.workerHeartbeat.create({
      data: {
        machineId,
        cpuPercent: body.cpuPercent ?? null,
        memPercent: body.memPercent ?? null,
        diskFreeGb: body.diskFreeGb ?? null,
        currentJobId: body.currentJobId ?? null,
      },
    }),
  ]);

  // Cancelled-but-unacknowledged jobs held by this machine → CANCEL_JOB commands
  const cancelled = await db.renderJob.findMany({
    where: { machineId, status: "CANCELLED", finishedAt: null },
    select: { id: true },
  });
  return {
    ok: true,
    commands: cancelled.map((j) => ({ type: "CANCEL_JOB" as const, jobId: j.id })),
  };
}

/** Machine list for the monitoring page, with latest heartbeat + current job. */
export async function listMachines() {
  const heartbeatSeconds = 30;
  const machines = await db.machine.findMany({
    orderBy: { name: "asc" },
    include: {
      heartbeats: { orderBy: { createdAt: "desc" }, take: 1 },
      renderJobs: {
        where: { status: { in: ["CLAIMED", "RUNNING"] } },
        take: 1,
        include: { export: { select: { id: true, versionString: true, shot: { select: { shotCode: true } } } } },
      },
    },
  });
  const now = Date.now();
  return machines.map((m) => ({
    id: m.id,
    name: m.name,
    hostname: m.hostname,
    enabled: m.enabled,
    // ONLINE = heartbeat within 3× the heartbeat interval (§5.6)
    status: !m.enabled
      ? "DISABLED"
      : m.lastSeenAt && now - m.lastSeenAt.getTime() < heartbeatSeconds * 3 * 1000
        ? "ONLINE"
        : "OFFLINE",
    lastSeenAt: m.lastSeenAt,
    workerVersion: m.workerVersion,
    aeVersion: m.aeVersion,
    capabilities: m.capabilities,
    availability: m.availability,
    renderNowUntil: m.renderNowUntil,
    latestHeartbeat: m.heartbeats[0]
      ? {
          createdAt: m.heartbeats[0].createdAt,
          cpuPercent: m.heartbeats[0].cpuPercent,
          memPercent: m.heartbeats[0].memPercent,
          diskFreeGb: m.heartbeats[0].diskFreeGb,
        }
      : null,
    currentJob: m.renderJobs[0]
      ? {
          id: m.renderJobs[0].id,
          exportId: m.renderJobs[0].exportId,
          shotCode: m.renderJobs[0].export?.shot.shotCode ?? null,
          versionString: m.renderJobs[0].export?.versionString ?? null,
          progress: m.renderJobs[0].progress,
          etaSeconds: m.renderJobs[0].etaSeconds,
        }
      : null,
  }));
}

export async function updateMachine(
  machineId: string,
  patch: { enabled?: boolean; availability?: MachineAvailability | null; renderNowHours?: number | null }
) {
  const machine = await db.machine.findUnique({ where: { id: machineId } });
  if (!machine) throw new PipelineError(404, "Machine not found");

  const data: Prisma.MachineUpdateInput = {};
  if (patch.enabled !== undefined) {
    data.enabled = patch.enabled;
    data.status = patch.enabled ? machine.status === "DISABLED" ? "OFFLINE" : machine.status : "DISABLED";
  }
  if (patch.availability !== undefined) {
    data.availability = patch.availability === null ? Prisma.DbNull : (patch.availability as unknown as Prisma.InputJsonValue);
  }
  if (patch.renderNowHours !== undefined) {
    data.renderNowUntil =
      patch.renderNowHours === null || patch.renderNowHours <= 0
        ? null
        : new Date(Date.now() + patch.renderNowHours * 3600 * 1000);
  }
  const updated = await db.machine.update({ where: { id: machineId }, data });
  return {
    machine: {
      id: updated.id,
      name: updated.name,
      enabled: updated.enabled,
      availability: updated.availability,
      renderNowUntil: updated.renderNowUntil,
    },
  };
}