MineGeo/lib/fabric-jobs.ts

import { randomUUID } from "node:crypto";
import type { SupabaseClient } from "@supabase/supabase-js";
import type { Database } from "../shared/database";
import { datasetSchema } from "../shared/protocol";
import { processFabricImages } from "./fabric-images";

type Job = Database["public"]["Tables"]["upload_jobs"]["Row"];
type JobStatus = Pick<Job, "state" | "lease_until">;
export function needsProcessing(job: JobStatus | null | undefined) {
  return (
    !!job &&
    (job.state === "queued" ||
      (job.state === "running" &&
        !!job.lease_until &&
        Date.parse(job.lease_until) < Date.now()))
  );
}
function checked<T>(result: {
  data: T;
  error: { message: string } | null;
}): NonNullable<T> {
  if (result.error) throw new Error(result.error.message);
  if (result.data == null) throw new Error("Processing record not found");
  return result.data!;
}

// Same queue, attempts and two-minute renewable leases as the standalone worker.
// Scope each invocation to a dataset the API has already authorized.
export async function processFabricJob(
  client: SupabaseClient<Database>,
  datasetId: string,
  deadlineAt = Date.now() + 250000,
) {
  const pending = checked(
    await client
      .from("upload_jobs")
      .select("*")
      .eq("dataset_id", datasetId)
      .maybeSingle(),
  );
  if (
    !needsProcessing(pending) ||
    pending.attempts >= 3 ||
    deadlineAt - Date.now() < 20000
  )
    return "skipped";
  const mapOnly = pending.kind === "map";
  const sourceId = mapOnly ? pending.source_dataset_id : pending.dataset_id;
  if (!sourceId) throw new Error("Missing capture for map refresh");
  const dataset = checked(
    await client
      .from("datasets")
      .select("owner_id,source")
      .eq("id", sourceId)
      .single(),
  );
  // Terrain exports still use the standalone renderer; do not consume a retry.
  const input = checked(
    await client
      .from("dataset_inputs")
      .select("manifest")
      .eq("dataset_id", sourceId)
      .single(),
  );
  const manifest = datasetSchema.parse(input.manifest);
  if (dataset.source !== "fabric" && !manifest.mediaFormat) return "skipped";
  const worker = "vercel-" + randomUUID();
  const now = Date.now();
  let claim = client
    .from("upload_jobs")
    .update({
      state: "running",
      attempts: pending.attempts + 1,
      lease_owner: worker,
      lease_until: new Date(now + 120000).toISOString(),
      progress: 0,
      error: null,
      updated_at: new Date(now).toISOString(),
    })
    .eq("id", pending.id)
    .eq("state", pending.state)
    .eq("attempts", pending.attempts);
  if (pending.state === "running")
    claim = claim.lt("lease_until", new Date(now).toISOString());
  const rows = checked(await claim.select("*"));
  if (!rows.length) return "skipped";
  const job = rows[0];
  const abort = new AbortController();
  // Leave time to release the lease before Vercel's 300-second limit.
  const deadline = setTimeout(
    () =>
      abort.abort(new Error("Image processing time limit reached; retrying")),
    Math.max(1, deadlineAt - Date.now()),
  );
  let lastProgress = 0;
  async function update(
    values: Database["public"]["Tables"]["upload_jobs"]["Update"],
  ) {
    const rows = checked(
      await client
        .from("upload_jobs")
        .update(values)
        .eq("id", job.id)
        .eq("lease_owner", worker)
        .eq("state", "running")
        .select("id"),
    );
    if (!rows.length) abort.abort(new Error("Lease lost or cancelled"));
    abort.signal.throwIfAborted();
  }
  const heartbeat = setInterval(() => {
    void update({
      lease_until: new Date(Date.now() + 120000).toISOString(),
      updated_at: new Date().toISOString(),
    }).catch((error) => abort.abort(error));
  }, 20000);
  try {
    console.log(
      `Vercel worker: processing ${job.id} (${manifest.viewpoints.length} panoramas)`,
    );
    await processFabricImages(manifest, {
      mapOnly,
      signal: abort.signal,
      async download(path) {
        const blob = checked(
          await client.storage
            .from("minegeo-raw")
            .download(`${dataset.owner_id}/${sourceId}/${path}`),
        );
        return Buffer.from(await blob.arrayBuffer());
      },
      async save(path, bytes) {
        checked(
          await client.storage
            .from("minegeo-media")
            .upload(
              mapOnly
                ? `${job.dataset_id}/maps/${job.id}.webp`
                : `${job.dataset_id}/${path}`,
              bytes,
              { contentType: "image/webp", upsert: true, cacheControl: "60" },
            ),
        );
      },
      async progress(percent) {
        if (Date.now() - lastProgress < 2000 && percent < 95) return;
        lastProgress = Date.now();
        await update({
          progress: Math.floor(percent),
          updated_at: new Date().toISOString(),
        });
      },
    });
    abort.signal.throwIfAborted();
    const result = await client.rpc("complete_job", {
      p_job: job.id,
      p_worker: worker,
      p_count: manifest.viewpoints.length,
      p_cover: manifest.viewpoints[0].id,
    });
    if (result.error) throw new Error(result.error.message);
    console.log(`Vercel worker: ${job.id} complete; private draft ready`);
    return "complete";
  } catch (error) {
    const message =
      error instanceof Error ? error.message : "Image processing failed";
    console.error(`Vercel worker: ${job.id}: ${message}`);
    const state = job.attempts >= 3 ? "failed" : "queued";
    const released = checked(
      await client
        .from("upload_jobs")
        .update({
          state,
          error: message,
          lease_owner: null,
          lease_until: null,
          progress: 0,
          updated_at: new Date().toISOString(),
        })
        .eq("id", job.id)
        .eq("lease_owner", worker)
        .eq("state", "running")
        .select("id"),
    );
    if (!mapOnly && state === "failed" && released.length) {
      const result = await client
        .from("datasets")
        .update({ state: "failed", rejection_reason: message })
        .eq("id", job.dataset_id)
        .eq("state", "processing");
      if (result.error) throw new Error(result.error.message);
    }
    return released.length ? state : "skipped";
  } finally {
    clearInterval(heartbeat);
    clearTimeout(deadline);
  }
}