The first commit, the core and all files is adding to repo
This commit is contained in:
299
worker/main.ts
Normal file
299
worker/main.ts
Normal file
@@ -0,0 +1,299 @@
|
||||
import { createClient } from "@supabase/supabase-js";
|
||||
import { randomUUID, createHash } from "node:crypto";
|
||||
import { mkdir, writeFile, readFile, readdir, rm } from "node:fs/promises";
|
||||
import { gunzipSync } from "node:zlib";
|
||||
import sharp from "sharp";
|
||||
import { datasetSchema, faces, type TerrainSnapshot } from "../shared/protocol";
|
||||
import { renderSnapshot } from "../renderer/render.mjs";
|
||||
import type { Database } from "../shared/database";
|
||||
const client = createClient<Database>(
|
||||
process.env.NEXT_PUBLIC_SUPABASE_URL!,
|
||||
process.env.SUPABASE_SECRET_KEY!,
|
||||
{ auth: { persistSession: false, autoRefreshToken: false } },
|
||||
);
|
||||
const worker = "worker-" + randomUUID();
|
||||
let stopping = false;
|
||||
process.on("SIGINT", () => (stopping = true));
|
||||
process.on("SIGTERM", () => (stopping = true));
|
||||
function check<T>(r: {
|
||||
data: T;
|
||||
error: { message: string } | null;
|
||||
}): NonNullable<T> {
|
||||
if (r.error) throw new Error(r.error.message);
|
||||
return r.data!;
|
||||
}
|
||||
async function processJob(job: {
|
||||
id: string;
|
||||
dataset_id: string;
|
||||
attempts: number;
|
||||
}) {
|
||||
const abort = new AbortController(),
|
||||
directory = `.cache/jobs/${job.id}`;
|
||||
await mkdir(directory, { recursive: true });
|
||||
const heartbeat = setInterval(async () => {
|
||||
try {
|
||||
const rows = check(
|
||||
await client
|
||||
.from("upload_jobs")
|
||||
.update({
|
||||
lease_until: new Date(Date.now() + 120000).toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
})
|
||||
.eq("id", job.id)
|
||||
.eq("lease_owner", worker)
|
||||
.eq("state", "running")
|
||||
.select("id"),
|
||||
);
|
||||
if (!rows.length) abort.abort();
|
||||
} catch {
|
||||
abort.abort();
|
||||
}
|
||||
}, 20000);
|
||||
try {
|
||||
const dataset = check(
|
||||
await client
|
||||
.from("datasets")
|
||||
.select("*")
|
||||
.eq("id", job.dataset_id)
|
||||
.single(),
|
||||
);
|
||||
const input = check(
|
||||
await client
|
||||
.from("dataset_inputs")
|
||||
.select("manifest")
|
||||
.eq("dataset_id", job.dataset_id)
|
||||
.single(),
|
||||
);
|
||||
const manifest = datasetSchema.parse(input.manifest);
|
||||
for (const asset of manifest.assets) {
|
||||
if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
|
||||
const blob = check(
|
||||
await client.storage
|
||||
.from("minegeo-raw")
|
||||
.download(`${dataset.owner_id}/${job.dataset_id}/${asset.path}`),
|
||||
);
|
||||
const bytes = Buffer.from(await blob.arrayBuffer());
|
||||
if (
|
||||
bytes.length !== asset.bytes ||
|
||||
createHash("sha256").update(bytes).digest("hex") !== asset.sha256
|
||||
)
|
||||
throw new Error(`Asset checksum/size mismatch: ${asset.path}`);
|
||||
const path = `${directory}/raw/${asset.path}`;
|
||||
await mkdir(path.slice(0, path.lastIndexOf("/")), { recursive: true });
|
||||
await writeFile(path, bytes);
|
||||
}
|
||||
await client
|
||||
.from("upload_jobs")
|
||||
.update({ progress: 20 })
|
||||
.eq("id", job.id)
|
||||
.eq("lease_owner", worker)
|
||||
.eq("state", "running");
|
||||
const output = `${directory}/media`;
|
||||
await mkdir(output, { recursive: true });
|
||||
if (manifest.source === "plugin") {
|
||||
const compressed = await readFile(`${directory}/raw/terrain.json.gz`);
|
||||
const terrain = JSON.parse(
|
||||
gunzipSync(compressed, {
|
||||
maxOutputLength: 256 * 1024 * 1024,
|
||||
}).toString(),
|
||||
) as TerrainSnapshot;
|
||||
if (
|
||||
!Array.isArray(terrain.blocks) ||
|
||||
terrain.blocks.length > 5000000 ||
|
||||
!Array.isArray(terrain.coverage)
|
||||
)
|
||||
throw new Error("Terrain snapshot exceeds supported render budget");
|
||||
if (
|
||||
["minX", "minZ", "maxX", "maxZ"].some(
|
||||
(k) =>
|
||||
terrain.bounds[k as keyof typeof terrain.bounds] !==
|
||||
manifest.bounds[k as keyof typeof manifest.bounds],
|
||||
)
|
||||
)
|
||||
throw new Error("Snapshot bounds disagree with manifest");
|
||||
for (const b of terrain.blocks) {
|
||||
if (
|
||||
!Array.isArray(b) ||
|
||||
b.length < 4 ||
|
||||
!b.slice(0, 3).every((v) => Number.isInteger(v)) ||
|
||||
typeof b[3] !== "string" ||
|
||||
!/^minecraft:[a-z0-9_]+(\[[a-z0-9_=,]+\])?$/.test(b[3])
|
||||
)
|
||||
throw new Error("Invalid block record");
|
||||
}
|
||||
const report = await renderSnapshot(
|
||||
terrain,
|
||||
manifest.viewpoints,
|
||||
output,
|
||||
{
|
||||
size: Number(process.env.MINEGEO_FACE_SIZE || 1024),
|
||||
signal: abort.signal,
|
||||
},
|
||||
);
|
||||
console.log(`Rendered ${job.dataset_id}: ${JSON.stringify(report)}`);
|
||||
} else {
|
||||
await sharp(await readFile(`${directory}/raw/map.png`), {
|
||||
limitInputPixels: 16777216,
|
||||
})
|
||||
.webp({ quality: 90 })
|
||||
.toFile(`${output}/map.webp`);
|
||||
for (const point of manifest.viewpoints) {
|
||||
await mkdir(`${output}/${point.id}`, { recursive: true });
|
||||
for (const face of faces) {
|
||||
const bytes = await readFile(
|
||||
`${directory}/raw/${point.id}/${face}.png`,
|
||||
);
|
||||
const meta = await sharp(bytes, {
|
||||
limitInputPixels: 16777216,
|
||||
}).metadata();
|
||||
if (
|
||||
!meta.width ||
|
||||
meta.width !== meta.height ||
|
||||
meta.width < 256 ||
|
||||
meta.width > 2048
|
||||
)
|
||||
throw new Error("Panorama faces must be square, 256–2048 pixels");
|
||||
await sharp(bytes, { limitInputPixels: 16777216 })
|
||||
.webp({ quality: 85 })
|
||||
.toFile(`${output}/${point.id}/${face}.webp`);
|
||||
if (face === "nz")
|
||||
await sharp(bytes)
|
||||
.resize(512, 512)
|
||||
.webp({ quality: 76 })
|
||||
.toFile(`${output}/${point.id}/preview.webp`);
|
||||
}
|
||||
}
|
||||
}
|
||||
const mapMeta = await sharp(`${output}/map.webp`).metadata();
|
||||
if (!mapMeta.width) throw new Error("Missing map output");
|
||||
for (const entry of await readdir(output, { recursive: true })) {
|
||||
if (!entry.endsWith(".webp")) continue;
|
||||
if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
|
||||
const normalized = entry.replaceAll("\\", "/");
|
||||
check(
|
||||
await client.storage
|
||||
.from("minegeo-media")
|
||||
.upload(
|
||||
`${job.dataset_id}/${normalized}`,
|
||||
await readFile(`${output}/${entry}`),
|
||||
{
|
||||
contentType: "image/webp",
|
||||
upsert: true,
|
||||
cacheControl: "31536000",
|
||||
},
|
||||
),
|
||||
);
|
||||
}
|
||||
if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
|
||||
check(
|
||||
await client.rpc("complete_job", {
|
||||
p_job: job.id,
|
||||
p_worker: worker,
|
||||
p_count: manifest.viewpoints.length,
|
||||
p_cover: manifest.viewpoints[0].id,
|
||||
}),
|
||||
);
|
||||
} catch (error) {
|
||||
const message =
|
||||
error instanceof Error ? error.message : "Processing failed";
|
||||
console.error(`Job ${job.id}: ${message}`);
|
||||
const state = job.attempts >= 3 ? "failed" : "queued";
|
||||
const rows = check(
|
||||
await client
|
||||
.from("upload_jobs")
|
||||
.update({ state, error: message, lease_until: null, progress: 0 })
|
||||
.eq("id", job.id)
|
||||
.eq("lease_owner", worker)
|
||||
.eq("state", "running")
|
||||
.select("id"),
|
||||
);
|
||||
if (state === "failed" && rows.length)
|
||||
await client
|
||||
.from("datasets")
|
||||
.update({ state: "failed", rejection_reason: message })
|
||||
.eq("id", job.dataset_id)
|
||||
.eq("state", "processing");
|
||||
} finally {
|
||||
clearInterval(heartbeat);
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
let housekeeping = 0;
|
||||
console.log(`MineGeo ${worker} ready; PostgreSQL leases through Supabase API.`);
|
||||
while (!stopping) {
|
||||
try {
|
||||
if (Date.now() - housekeeping > 300000) {
|
||||
housekeeping = Date.now();
|
||||
const dead = check(
|
||||
await client
|
||||
.from("upload_jobs")
|
||||
.update({
|
||||
state: "failed",
|
||||
error: "Worker lease expired after three attempts",
|
||||
})
|
||||
.eq("state", "running")
|
||||
.lt("lease_until", new Date().toISOString())
|
||||
.gte("attempts", 3)
|
||||
.select("dataset_id"),
|
||||
);
|
||||
for (const j of dead)
|
||||
await client
|
||||
.from("datasets")
|
||||
.update({
|
||||
state: "failed",
|
||||
rejection_reason: "Processing exhausted retry limit",
|
||||
})
|
||||
.eq("id", j.dataset_id)
|
||||
.eq("state", "processing");
|
||||
await client
|
||||
.from("pairing_codes")
|
||||
.delete()
|
||||
.lt("expires_at", new Date().toISOString());
|
||||
await client
|
||||
.from("rate_limits")
|
||||
.delete()
|
||||
.lt("expires_at", new Date().toISOString());
|
||||
await client
|
||||
.from("game_sessions")
|
||||
.delete()
|
||||
.is("owner_id", null)
|
||||
.lt("created_at", new Date(Date.now() - 30 * 86400000).toISOString());
|
||||
const abandoned = check(
|
||||
await client
|
||||
.from("datasets")
|
||||
.select("id,owner_id")
|
||||
.eq("state", "uploading")
|
||||
.lt("created_at", new Date(Date.now() - 86400000).toISOString()),
|
||||
);
|
||||
for (const d of abandoned) {
|
||||
const input = check(
|
||||
await client
|
||||
.from("dataset_inputs")
|
||||
.select("manifest")
|
||||
.eq("dataset_id", d.id)
|
||||
.single(),
|
||||
);
|
||||
await client.storage
|
||||
.from("minegeo-raw")
|
||||
.remove(
|
||||
datasetSchema
|
||||
.parse(input.manifest)
|
||||
.assets.map(
|
||||
(a: { path: string }) => `${d.owner_id}/${d.id}/${a.path}`,
|
||||
),
|
||||
);
|
||||
await client
|
||||
.from("datasets")
|
||||
.delete()
|
||||
.eq("id", d.id)
|
||||
.eq("state", "uploading");
|
||||
}
|
||||
}
|
||||
const jobs = check(await client.rpc("claim_job", { p_worker: worker }));
|
||||
if (jobs?.length) await processJob(jobs[0]);
|
||||
else await new Promise((r) => setTimeout(r, 2500));
|
||||
} catch (error) {
|
||||
console.error("Worker:", error instanceof Error ? error.message : error);
|
||||
await new Promise((r) => setTimeout(r, 5000));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user