Add map import and export with durable refresh progress
This commit is contained in:
198
lib/fabric-jobs.ts
Normal file
198
lib/fabric-jobs.ts
Normal file
@@ -0,0 +1,198 @@
|
||||
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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user