pavel444-byte
c9fc9496eb
Some checks failed
Deploy MineGeo to Dokploy / deploy (push) Has been cancelled
447 lines
15 KiB
TypeScript
447 lines
15 KiB
TypeScript
import { createClient } from "@supabase/supabase-js";
|
||
import { existsSync } from "node:fs";
|
||
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 { processFabricImages } from "../lib/fabric-images";
|
||
import { renderSnapshot } from "../renderer/render.mjs";
|
||
import type { Database } from "../shared/database";
|
||
if (existsSync(".env.local")) process.loadEnvFile(".env.local");
|
||
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;
|
||
let activeAbort: AbortController | null = null;
|
||
function stopWorker() {
|
||
stopping = true;
|
||
activeAbort?.abort();
|
||
}
|
||
process.on("SIGINT", stopWorker);
|
||
process.on("SIGTERM", stopWorker);
|
||
function check<T>(r: {
|
||
data: T;
|
||
error: { message: string } | null;
|
||
}): NonNullable<T> {
|
||
if (r.error) throw new Error(r.error.message);
|
||
return r.data!;
|
||
}
|
||
function summarizeWorkerError(error: unknown) {
|
||
const message = (error instanceof Error ? error.message : String(error))
|
||
.replace(/\s+/g, " ")
|
||
.trim();
|
||
if (/<(?:!doctype|html|head|body)\b/i.test(message)) {
|
||
const status = message.match(/\b([45]\d{2})\b/)?.[1];
|
||
const provider = /cloudflare/i.test(message) ? "Cloudflare" : "upstream";
|
||
return `${provider} returned an HTML error${status ? ` (${status})` : ""}`;
|
||
}
|
||
return message.length > 500 ? `${message.slice(0, 497)}…` : message;
|
||
}
|
||
async function processJob(job: {
|
||
id: string;
|
||
dataset_id: string;
|
||
attempts: number;
|
||
kind?: string;
|
||
source_dataset_id?: string | null;
|
||
}) {
|
||
const mapOnly = job.kind === "map";
|
||
const sourceId = mapOnly ? job.source_dataset_id! : job.dataset_id;
|
||
const abort = new AbortController(),
|
||
directory = `.cache/jobs/${job.id}`;
|
||
activeAbort = abort;
|
||
await mkdir(directory, { recursive: true });
|
||
let lastProgressWrite = 0;
|
||
let lastLoggedAt = 0;
|
||
let lastLoggedPercent = -1;
|
||
async function progress(percent: number, message: string) {
|
||
if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
|
||
const value = Math.max(0, Math.min(99, Math.floor(percent)));
|
||
const now = Date.now();
|
||
if (now - lastLoggedAt >= 30000 || value >= lastLoggedPercent + 5) {
|
||
console.log(`Job ${job.id}: ${value}% — ${message}`);
|
||
lastLoggedAt = now;
|
||
lastLoggedPercent = value;
|
||
}
|
||
if (now - lastProgressWrite < 2000 && value !== 99) return;
|
||
lastProgressWrite = now;
|
||
const rows = check(
|
||
await client
|
||
.from("upload_jobs")
|
||
.update({
|
||
progress: value,
|
||
error: null,
|
||
updated_at: new Date().toISOString(),
|
||
})
|
||
.eq("id", job.id)
|
||
.eq("lease_owner", worker)
|
||
.eq("state", "running")
|
||
.select("id"),
|
||
);
|
||
if (!rows.length) {
|
||
abort.abort();
|
||
throw new Error("Lease lost or cancelled");
|
||
}
|
||
}
|
||
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", sourceId).single(),
|
||
);
|
||
const input = check(
|
||
await client
|
||
.from("dataset_inputs")
|
||
.select("manifest")
|
||
.eq("dataset_id", sourceId)
|
||
.single(),
|
||
);
|
||
const manifest = datasetSchema.parse(input.manifest);
|
||
const assets = manifest.assets.filter(
|
||
(a) =>
|
||
!mapOnly ||
|
||
a.path ===
|
||
(manifest.mediaFormat
|
||
? "map.webp"
|
||
: manifest.source === "plugin"
|
||
? "terrain.json.gz"
|
||
: "map.png"),
|
||
);
|
||
await progress(0, `Downloading ${assets.length} export files`);
|
||
let downloaded = 0;
|
||
for (const asset of 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}/${sourceId}/${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 progress(
|
||
(20 * ++downloaded) / assets.length,
|
||
`Downloaded ${downloaded}/${assets.length} export files`,
|
||
);
|
||
}
|
||
const output = `${directory}/media`;
|
||
await mkdir(output, { recursive: true });
|
||
if (manifest.mediaFormat) {
|
||
await processFabricImages(manifest, {
|
||
mapOnly,
|
||
signal: abort.signal,
|
||
download: (path) => readFile(`${directory}/raw/${path}`),
|
||
async save(path, bytes) {
|
||
await mkdir(
|
||
`${output}/${path}`.slice(0, `${output}/${path}`.lastIndexOf("/")),
|
||
{ recursive: true },
|
||
);
|
||
await writeFile(`${output}/${path}`, bytes);
|
||
},
|
||
progress: (percent) =>
|
||
progress(20 + percent * 0.7, "Validating imported map images"),
|
||
});
|
||
} else if (manifest.source === "plugin") {
|
||
await progress(20, "Reading the terrain snapshot");
|
||
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 (let i = 0; i < terrain.blocks.length; i++) {
|
||
const b = terrain.blocks[i];
|
||
if (
|
||
!Array.isArray(b) ||
|
||
b.length < 4 ||
|
||
!Number.isSafeInteger(b[0]) ||
|
||
!Number.isSafeInteger(b[1]) ||
|
||
!Number.isSafeInteger(b[2]) ||
|
||
b[0] < terrain.bounds.minX ||
|
||
b[0] > terrain.bounds.maxX ||
|
||
b[2] < terrain.bounds.minZ ||
|
||
b[2] > terrain.bounds.maxZ ||
|
||
b[1] < -4096 ||
|
||
b[1] > 4096 ||
|
||
typeof b[3] !== "string" ||
|
||
!/^minecraft:[a-z0-9_]+(\[[a-z0-9_=,]+\])?$/.test(b[3])
|
||
)
|
||
throw new Error("Invalid block record");
|
||
// Avoid starving lease heartbeats while validating millions of records.
|
||
if (i % 65536 === 0)
|
||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||
}
|
||
console.log(
|
||
`Job ${job.id}: ${terrain.blocks.length.toLocaleString()} blocks, ${manifest.viewpoints.length} panoramas; starting cached geometry builder.`,
|
||
);
|
||
const report = await renderSnapshot(
|
||
terrain,
|
||
mapOnly ? [] : (manifest.availableViewpoints ?? manifest.viewpoints),
|
||
output,
|
||
{
|
||
size: Number(process.env.MINEGEO_FACE_SIZE || 1024),
|
||
minecraftVersion: manifest.minecraftVersion,
|
||
signal: abort.signal,
|
||
onProgress: ({ percent, message }) => progress(percent, message),
|
||
},
|
||
);
|
||
console.log(`Rendered ${job.dataset_id}: ${JSON.stringify(report)}`);
|
||
} else {
|
||
await progress(25, "Compressing the captured map");
|
||
await sharp(await readFile(`${directory}/raw/map.png`), {
|
||
limitInputPixels: 16777216,
|
||
})
|
||
.webp({ lossless: true })
|
||
.toFile(`${output}/map.webp`);
|
||
let processedFaces = 0;
|
||
for (const point of mapOnly
|
||
? []
|
||
: (manifest.availableViewpoints ?? 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`);
|
||
await progress(
|
||
35 +
|
||
(55 * ++processedFaces) /
|
||
(manifest.viewpoints.length * faces.length),
|
||
`Compressing panorama faces ${processedFaces}/${manifest.viewpoints.length * faces.length}`,
|
||
);
|
||
}
|
||
}
|
||
}
|
||
const mapMeta = await sharp(`${output}/map.webp`).metadata();
|
||
if (!mapMeta.width) throw new Error("Missing map output");
|
||
await progress(90, "Saving processed images");
|
||
const images = (await readdir(output, { recursive: true })).filter(
|
||
(entry) => entry.endsWith(".webp"),
|
||
);
|
||
let uploaded = 0,
|
||
nextImage = 0;
|
||
let uploadError: unknown;
|
||
// Small bounded pool avoids paying Storage round-trip latency serially.
|
||
// Finish every in-flight operation before cleanup/retry can touch the files.
|
||
await Promise.all(
|
||
Array.from({ length: Math.min(4, images.length) }, async () => {
|
||
try {
|
||
while (nextImage < images.length && !uploadError) {
|
||
if (abort.signal.aborted)
|
||
throw new Error("Lease lost or cancelled");
|
||
const entry = images[nextImage++],
|
||
normalized = entry.replaceAll("\\", "/");
|
||
check(
|
||
await client.storage
|
||
.from("minegeo-media")
|
||
.upload(
|
||
mapOnly
|
||
? `${job.dataset_id}/maps/${job.id}.webp`
|
||
: `${job.dataset_id}/${normalized}`,
|
||
await readFile(`${output}/${entry}`),
|
||
{
|
||
contentType: "image/webp",
|
||
upsert: true,
|
||
// Drafts can be repaired before publication.
|
||
cacheControl: "60",
|
||
},
|
||
),
|
||
);
|
||
await progress(
|
||
90 + (9 * ++uploaded) / images.length,
|
||
`Saved ${uploaded}/${images.length} processed images`,
|
||
);
|
||
}
|
||
} catch (error) {
|
||
uploadError ??= error;
|
||
}
|
||
}),
|
||
);
|
||
if (uploadError) throw uploadError;
|
||
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,
|
||
}),
|
||
);
|
||
console.log(`Job ${job.id}: 100% — private draft ready for preview.`);
|
||
} catch (error) {
|
||
const message =
|
||
error instanceof Error ? error.message : "Processing failed";
|
||
console.error(`Job ${job.id}: ${message}`);
|
||
const state = stopping ? "queued" : job.attempts >= 3 ? "failed" : "queued";
|
||
const rows = check(
|
||
await client
|
||
.from("upload_jobs")
|
||
.update({
|
||
state,
|
||
error: stopping ? null : message,
|
||
lease_owner: null,
|
||
lease_until: null,
|
||
progress: 0,
|
||
attempts: stopping ? Math.max(0, job.attempts - 1) : job.attempts,
|
||
})
|
||
.eq("id", job.id)
|
||
.eq("lease_owner", worker)
|
||
.eq("state", "running")
|
||
.select("id"),
|
||
);
|
||
if (!mapOnly && state === "failed" && rows.length)
|
||
await client
|
||
.from("datasets")
|
||
.update({ state: "failed", rejection_reason: message })
|
||
.eq("id", job.dataset_id)
|
||
.eq("state", "processing");
|
||
} finally {
|
||
if (activeAbort === abort) activeAbort = null;
|
||
clearInterval(heartbeat);
|
||
await rm(directory, {
|
||
recursive: true,
|
||
force: true,
|
||
maxRetries: 5,
|
||
retryDelay: 300,
|
||
});
|
||
}
|
||
}
|
||
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 (retrying in 5s): ${summarizeWorkerError(error)}`,
|
||
);
|
||
await new Promise((r) => setTimeout(r, 5000));
|
||
}
|
||
}
|