continuum-backend/src/dispatch/index.ts

import { and, asc, eq, inArray, isNotNull, isNull, lt, sql } from "drizzle-orm";
import { db } from "../db";
import { printers, printJobs } from "../db/schema";
import { getDownloadUrl, headObject } from "../s3";
import { isGatewayOnline, sendPrinterCommand } from "../routes/ws/edge";
import { jobStatusFromGatewayEvent, nextJobStatus } from "./job_status";
import { rememberStart, takeStart } from "./late_failure";
import { dispatchEnabled, dispatchSwitchWarning } from "./enabled";

/**
 * The job scheduler: picks up `queued` jobs and puts them on idle printers.
 *
 * Until now `POST /jobs` ended at a `queued` row and an upload URL, and
 * nothing ever moved it. This is the join - deliberately the simplest thing
 * that is correct, because every interesting failure here is a physical one.
 *
 * ## Why a timer rather than an event
 *
 * A job becomes dispatchable for two unrelated reasons: someone queued it, or
 * a printer became free. The second has no event the backend owns - it
 * arrives as a telemetry snapshot from a gateway, five seconds after the fact
 * at best. A timer catches both without needing a nudge from either, and a
 * scheduler that only ran on job creation would leave a queue stuck forever
 * behind one busy printer.
 *
 * `POST /jobs` nudges it anyway, so a queued job on an idle farm starts in
 * well under a second rather than waiting out the tick.
 *
 * ## What it will not do
 *
 * It sends to a printer whose *last reported* status is `idle`, which is a
 * claim about several seconds ago. Two things follow. A printer someone
 * started by hand at the machine can still be sent a job, and the gateway's
 * own guard is what refuses it - `start_uploaded_file` checks the printer is
 * in a terminal state first. And a job is only ever dispatched once, because
 * `claimNextJob` moves the row out of `queued` in the same statement that
 * selects it; two ticks racing cannot both win.
 */

/** How often to look for dispatchable work. */
const TICK_MS = 5_000;

/**
 * How long the gateway gets to download the file from R2 and transfer it to
 * the printer before the presigned URL stops working.
 *
 * Generous on purpose. The FTPS transfer to the printer alone was measured at
 * 0.2 MiB/s on a real P1P, so a 62 MB project - which is the size of the
 * largest file sitting on that printer's card - is over five minutes of
 * upload before a byte of printing happens. A URL that expires mid-transfer
 * would fail the job for a reason that looks like a storage outage.
 */
const DOWNLOAD_URL_TTL_S = 1800;

/**
 * Printer states that may be sent work.
 *
 * `idle` only. `offline` is the trap worth naming: a printer the gateway has
 * not heard from keeps whatever status it last had, so `offline` can mean
 * "powered off mid-print", and a printer in `error` may be sitting in a
 * paused state with filament loaded. Neither is a machine to put a job on.
 */
const DISPATCHABLE_PRINTER_STATUS = ["idle"] as const;

/**
 * How long a job may sit in `uploading` before it is given up on.
 *
 * `uploading` means the gateway has accepted the start and is downloading the
 * file and transferring it to the printer. That reply is deliberately an
 * "accepted, working on it" - the gateway answers immediately because
 * awaiting the transfer inline would stop its telemetry for minutes, and
 * telemetry is what every job's state is derived from.
 *
 * The consequence is that a failure *after* the ack has no `CommandResult` to
 * carry it, so a job whose download 403s on an expired URL, or whose archive
 * the printer refuses, would otherwise sit in `uploading` forever claiming to
 * be nearly printing.
 *
 * Twenty minutes: the transfer alone is over five for the largest project
 * seen (62 MB at the measured 0.2 MiB/s), and the printer then spends minutes
 * in bed levelling before it reports `RUNNING` - a P1P sat at layer 0 for
 * several while `percent` climbed to 14. Anything tighter would reap jobs
 * that are working perfectly.
 */
const UPLOADING_TIMEOUT_MS = 20 * 60_000;

/**
 * Takes one queued job for one printer, atomically.
 *
 * The `where` repeats `status = 'queued'`, which is what makes this safe: an
 * `UPDATE ... WHERE status = 'queued'` that returns a row is proof this
 * caller is the one that moved it. Selecting and then updating would let two
 * ticks - or a tick and a nudge - both believe they owned the same job and
 * send two `start_print` commands for it.
 */
async function claimNextJob(farmId: string, printerId: string) {
  const next = await db.query.printJobs.findFirst({
    columns: { id: true },
    where: and(
      eq(printJobs.farmId, farmId),
      eq(printJobs.status, "queued"),
      // A job that already names a printer was aimed at one deliberately;
      // this scheduler only places the unassigned ones. Honouring a request
      // for a *specific* printer is a separate feature and wants its own
      // rule for what happens when that printer is busy.
      isNull(printJobs.printerId),
    ),
    // Oldest first. Not a priority system - just the absence of one, made
    // explicit so the order is not whatever Postgres felt like returning.
    orderBy: [asc(printJobs.createdAt)],
  });
  if (!next) return undefined;

  const [claimed] = await db
    .update(printJobs)
    .set({ printerId, status: "uploading", startedAt: new Date(), updatedAt: new Date() })
    .where(and(eq(printJobs.id, next.id), eq(printJobs.status, "queued")))
    .returning();

  return claimed;
}

/** Puts a job back for another tick, or fails it outright. */
async function releaseJob(jobId: string, outcome: "requeue" | "fail", reason: string) {
  const metadataNote = { lastDispatchError: reason, lastDispatchErrorAt: new Date().toISOString() };

  if (outcome === "requeue") {
    await db
      .update(printJobs)
      .set({
        status: "queued",
        printerId: null,
        startedAt: null,
        updatedAt: new Date(),
        metadata: metadataNote,
      })
      .where(eq(printJobs.id, jobId));
    return;
  }

  await db
    .update(printJobs)
    .set({ status: "failed", completedAt: new Date(), updatedAt: new Date(), metadata: metadataNote })
    .where(eq(printJobs.id, jobId));
}

/**
 * One pass: for every farm with a connected gateway and an idle printer, try
 * to place a job.
 *
 * Returns how many jobs it dispatched, which is what the tests assert on -
 * a scheduler that silently does nothing is the failure mode to catch.
 */
export { nextJobStatus } from "./job_status";

export async function dispatchTick(): Promise<number> {
  // Also covers the nudge from `POST /jobs`, which calls this directly.
  if (!dispatchEnabled()) return 0;
  await reapStalledUploads();

  let dispatched = 0;

  // Idle, claimed printers. The `gateway_printer_id` filter is not redundant
  // with the status check: an unclaimed printer has no id to send a command
  // to, and `sendPrinterCommand` would throw for every one of them.
  const candidates = await db.query.printers.findMany({
    columns: { id: true, farmId: true, name: true },
    where: and(
      inArray(printers.status, [...DISPATCHABLE_PRINTER_STATUS]),
      // Only printers a gateway has claimed. Not redundant with the status
      // check: an unclaimed printer has no gateway id to address, and
      // `sendPrinterCommand` throws for every one of them.
      isNotNull(printers.gatewayPrinterId),
    ),
  });

  for (const printer of candidates) {
    // Cheap and in-memory, so it goes before the queue query rather than
    // after: a farm whose gateway is offline has nothing dispatchable no
    // matter what is queued, and asking the database first would be a
    // round trip per printer to find that out.
    if (!isGatewayOnline(printer.farmId)) continue;

    const job = await claimNextJob(printer.farmId, printer.id);
    if (!job) continue;

    try {
      // Is the file even there? `POST /jobs` hands the client a presigned PUT
      // and returns, so the row exists before the object does. Asking the
      // bucket is the only honest answer: a client that crashed mid-upload
      // would never tell us, and dispatching anyway would have the gateway
      // download R2's 403 XML error document and hand it to a printer as a
      // project. A job whose file never arrives just stays queued.
      const object = await headObject(job.r2ObjectKey);
      if (!object) {
        await releaseJob(job.id, "requeue", "the job file is not in object storage yet");
        continue;
      }
      if (object.size === 0) {
        // Zero bytes is a finished upload of nothing, not an upload in
        // progress, and it will never become valid on its own.
        await releaseJob(job.id, "fail", "the job file is empty");
        console.warn(`[dispatch] job ${job.id} failed: uploaded file is 0 bytes`);
        continue;
      }

      // Presigned here, at dispatch, and never at job creation: a queued job
      // may wait days, and a URL minted when it was queued would be long
      // expired by the time anything sent it.
      const url = await getDownloadUrl(job.r2ObjectKey, DOWNLOAD_URL_TTL_S);

      const result = await sendPrinterCommand(printer.id, {
        action: "start_print",
        job_id: job.id,
        file_name: job.fileName,
        url,
        // Chosen by whoever queued the job; absent means "by the project's filament".
        ...(typeof job.metadata?.amsSlot === "number" ? { ams_slot: job.metadata.amsSlot } : {}),
      });

      // Accepted or not, a late result for this command can name the job.
      rememberStart(result.commandId, job.id);

      if (!result.ok) {
        // Requeued, not failed. The gateway refusing a start is usually
        // transient from the cloud's point of view - the printer was busier
        // than its last snapshot said, or the gateway dropped mid-command -
        // and a job that fails permanently because of a five-second-stale
        // status reading would be indefensible.
        await releaseJob(job.id, "requeue", result.error ?? "gateway reported the start failed");
        console.warn(`[dispatch] job ${job.id} requeued: ${result.error}`);
        continue;
      }

      // Note what this does *not* do: it does not mark the job `printing`.
      // The gateway accepting a start is three steps away from a running
      // print - confirmed the expensive way on 2026-10-15, when a real print
      // was accepted, uploaded byte-perfect, reached RUNNING and then paused
      // itself on a physical filament fault. The job stays `uploading` until
      // telemetry from the printer says otherwise. See `advanceJobsFrom`.
      dispatched += 1;
      console.log(`[dispatch] job ${job.id} sent to printer ${printer.name} (${printer.id})`);
    } catch (err) {
      // `sendPrinterCommand` throws only for things that were true before
      // anything was sent: no such printer, unclaimed, or no gateway. All
      // three are transient or fixable, so the job goes back in the queue.
      const reason = err instanceof Error ? err.message : String(err);
      await releaseJob(job.id, "requeue", reason);
      console.warn(`[dispatch] job ${job.id} requeued before sending: ${reason}`);
    }
  }

  return dispatched;
}

/**
 * A gateway that had accepted a `start_print` reports that the download or
 * transfer then failed. Fails the job with that reason - but only while it is
 * still `uploading`: one that has since reached `printing` is being driven by
 * the printer's own telemetry, and a late message must not overrule it.
 *
 * Failed rather than requeued, for the reason `reapStalledUploads` gives.
 */
export async function failJobFromGateway(commandId: string, error: string): Promise<void> {
  const jobId = takeStart(commandId);
  if (!jobId) return;

  const failed = await db
    .update(printJobs)
    .set({
      status: "failed",
      completedAt: new Date(),
      updatedAt: new Date(),
      metadata: sql`coalesce(${printJobs.metadata}, '{}'::jsonb) || ${JSON.stringify({
        lastDispatchError: `the gateway could not start it: ${error}`,
        lastDispatchErrorAt: new Date().toISOString(),
      })}::jsonb`,
    })
    .where(and(eq(printJobs.id, jobId), eq(printJobs.status, "uploading")))
    .returning({ id: printJobs.id });

  if (failed.length > 0) console.warn(`[dispatch] job ${jobId} failed: the gateway could not start it: ${error}`);
}

/**
 * Applies an outcome a gateway kept for us. Throws only when the database
 * could not be reached - the one case where the gateway must NOT be told it
 * was handled, so that it sends it again. A job that is unknown, belongs to
 * another farm, or is no longer in flight is not an error: there is nothing
 * to change, and the gateway should stop sending it.
 */
export async function applyGatewayJobEvent(
  farmId: string,
  event: { job_id: string; event: "started" | "finished" | "failed"; detail?: string | null; at_ms: number },
): Promise<"applied" | "ignored"> {
  // The farm in the `where` is the authorisation: a gateway can only move
  // jobs of the farm it is connected for.
  const job = await db.query.printJobs.findFirst({
    columns: { id: true, status: true },
    where: and(eq(printJobs.id, event.job_id), eq(printJobs.farmId, farmId)),
  });
  if (!job) return "ignored";

  const next = jobStatusFromGatewayEvent(job.status, event.event);
  if (!next) return "ignored";

  const at = new Date(event.at_ms);
  const terminal = next === "completed" || next === "failed";
  const updated = await db
    .update(printJobs)
    .set({
      status: next,
      updatedAt: new Date(),
      ...(next === "completed" ? { progressPct: 100 } : {}),
      ...(terminal ? { completedAt: at } : {}),
      ...(next === "failed"
        ? {
            metadata: sql`coalesce(${printJobs.metadata}, '{}'::jsonb) || ${JSON.stringify({
              lastDispatchError: event.detail ?? "the gateway reported the job failed",
              lastDispatchErrorAt: at.toISOString(),
            })}::jsonb`,
          }
        : {}),
    })
    // Re-checked in the write: the status may have moved since the read.
    .where(and(eq(printJobs.id, job.id), eq(printJobs.status, job.status)))
    .returning({ id: printJobs.id });

  if (updated.length === 0) return "ignored";
  console.log(`[dispatch] job ${job.id} ${job.status} -> ${next} (gateway reported ${event.event})`);
  return "applied";
}

/**
 * Fails jobs that have been `uploading` too long.
 *
 * Marked `failed` rather than requeued, deliberately. A requeue would send
 * the same file to the same printer and, if the cause was the file rather
 * than a hiccup, do it again every twenty minutes forever. More importantly
 * the printer may by now be *printing* it - the gateway could have succeeded
 * and only its telemetry be late - and requeueing would then dispatch a
 * second job onto a working machine. Failing is the recoverable mistake
 * here; a human can requeue, and the printer's own telemetry will still
 * report what it is really doing.
 */
async function reapStalledUploads(): Promise<void> {
  const cutoff = new Date(Date.now() - UPLOADING_TIMEOUT_MS);

  const stalled = await db
    .update(printJobs)
    .set({
      status: "failed",
      completedAt: new Date(),
      updatedAt: new Date(),
      metadata: {
        lastDispatchError:
          `the gateway accepted the start but the printer never reported printing within ` +
          `${Math.round(UPLOADING_TIMEOUT_MS / 60_000)} minutes - check the gateway's log for ` +
          `the download or transfer error (a failure it reports promptly fails the job sooner)`,
        lastDispatchErrorAt: new Date().toISOString(),
      },
    })
    .where(and(eq(printJobs.status, "uploading"), lt(printJobs.startedAt, cutoff)))
    .returning({ id: printJobs.id });

  for (const job of stalled) {
    console.warn(`[dispatch] job ${job.id} failed: stalled in uploading`);
  }
}

let timer: ReturnType<typeof setInterval> | undefined;

/** Starts the scheduler. Idempotent, so a second call does not double-tick. */
export function startDispatcher() {
  if (timer) return;
  const warning = dispatchSwitchWarning();
  if (warning) console.warn(`[dispatch] ${warning}`);
  if (!dispatchEnabled()) {
    console.log("[dispatch] DISABLED (DISPATCH_ENABLED=false): queued jobs will not be sent to printers");
    return;
  }
  timer = setInterval(() => {
    dispatchTick().catch((err) => console.error("[dispatch] tick failed", err));
  }, TICK_MS);
  // Node/Bun keep the process alive for an interval; this one should not be
  // the reason a server refuses to exit.
  timer.unref?.();
  console.log(`[dispatch] scheduler started, every ${TICK_MS}ms`);
}

export function stopDispatcher() {
  if (timer) clearInterval(timer);
  timer = undefined;
}

/** Runs a tick now, without waiting for the next one. */
export function nudgeDispatcher() {
  dispatchTick().catch((err) => console.error("[dispatch] nudge failed", err));
}

/**
 * Moves job rows along from what the printers are actually reporting.
 *
 * This is the half that matters, and the reason the dispatcher above refuses
 * to mark anything `printing`. A gateway accepting `start_print` means it
 * accepted the command - no more. Confirmed the expensive way on a real P1P
 * (2026-10-15): the command was accepted, the file uploaded byte-perfect, the
 * printer reached `RUNNING`, and then it paused itself on a physical
 * broken-filament fault. A job marked `printing` on the gateway's `ok` would
 * have sat there claiming to print for the rest of the day.
 *
 * So job state is derived from the telemetry that follows, and the mapping is
 * narrow on purpose:
 *
 * - `printing` while the printer reports it is printing.
 * - `paused` likewise - including the printer pausing *itself* on a fault,
 *   which is what that P1P did and which an operator needs to see.
 * - `completed` when the printer goes idle having been printing. A Bambu
 *   reports a finished print as `FINISH`, which `mapStatus` narrows to
 *   `idle`, so "was printing, now idle" is the completion signal.
 * - `failed` when the printer reports an error. Note a Bambu reports an
 *   operator-cancelled print as `FAILED` too, so this cannot tell a
 *   cancellation from a fault; `cancelled` is set by whoever sends the stop,
 *   and is left alone here precisely so it is not overwritten by the `FAILED`
 *   that follows it.
 *
 * Progress is copied across whenever it is reported, because a job row that
 * says 0% through a two-hour print is worse than useless to an operator.
 */
export async function advanceJobsFrom(
  printerId: string,
  status: "idle" | "printing" | "paused" | "error" | undefined,
  percent: number | null | undefined,
): Promise<void> {
  if (!status) return;

  // The one job this printer could be running. `uploading` is included
  // because that is the state the dispatcher leaves a job in, and the first
  // telemetry that says `printing` is what promotes it.
  const job = await db.query.printJobs.findFirst({
    columns: { id: true, status: true },
    where: and(
      eq(printJobs.printerId, printerId),
      inArray(printJobs.status, ["uploading", "printing", "paused"]),
    ),
    orderBy: [asc(printJobs.createdAt)],
  });
  if (!job) return;

  // Narrowed explicitly rather than cast: the query's `where` already limits
  // this to the three, so a value outside them means the query and this call
  // have drifted apart - which is exactly the bug a cast would hide.
  if (job.status !== "uploading" && job.status !== "printing" && job.status !== "paused") return;

  const next = nextJobStatus(job.status, status);

  const terminal = next === "completed" || next === "failed";

  // Progress alone still deserves a write, so an unchanged status does not
  // freeze the percentage.
  if (!next && percent == null) return;

  await db
    .update(printJobs)
    .set({
      ...(next ? { status: next } : {}),
      ...(percent != null ? { progressPct: percent } : {}),
      ...(terminal ? { completedAt: new Date(), progressPct: next === "completed" ? 100 : undefined } : {}),
      updatedAt: new Date(),
    })
    .where(eq(printJobs.id, job.id));

  if (next && next !== job.status) {
    console.log(`[dispatch] job ${job.id} ${job.status} -> ${next} (printer reported ${status})`);
  }
}