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, }, }; }