166 lines
11 KiB
JavaScript
166 lines
11 KiB
JavaScript
import fs from "node:fs";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import crypto from "node:crypto";
|
|
import http from "node:http";
|
|
import { fileURLToPath } from "node:url";
|
|
import { startBoundedServerJobProcess, cancelServerJobProcess } from "./server-job-process.mjs";
|
|
|
|
const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../..");
|
|
const RUNNER = path.join(ROOT, "tools/web/server-operator-job.py");
|
|
const MAGIC = Buffer.from("WBJ1");
|
|
const MAX_ENVELOPE = 513 * 1024 * 1024;
|
|
const jobs = new Map();
|
|
const operatorSchemaCache = new Map();
|
|
|
|
function fail(message, status = 400) { const error = new Error(message); error.statusCode = status; throw error; }
|
|
function digest(bytes) { return crypto.createHash("sha256").update(bytes).digest("hex"); }
|
|
function stableJSON(value) {
|
|
if (value === null || typeof value === "string" || typeof value === "boolean" || typeof value === "number") return JSON.stringify(value);
|
|
if (Array.isArray(value)) return `[${value.map(stableJSON).join(",")}]`;
|
|
return `{${Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${stableJSON(value[key])}`).join(",")}}`;
|
|
}
|
|
function validateBindings(request) {
|
|
const propertiesSha256 = digest(Buffer.from(stableJSON(request.properties ?? {})));
|
|
if (propertiesSha256 !== request.propertiesSha256) fail("SERVER_OPERATOR_HTTP_PROPERTIES_HASH_MISMATCH", 422);
|
|
const { requestSha256, ...unsigned } = request;
|
|
const canonical = stableJSON({ ...unsigned, context: unsigned.context ?? "BACKGROUND" });
|
|
if (digest(Buffer.from(canonical)) !== requestSha256) fail("SERVER_OPERATOR_HTTP_REQUEST_HASH_MISMATCH", 422);
|
|
if (/^(?:wm|window|screen|file|outliner|view3d)\./iu.test(String(request.operator ?? ""))) fail("SERVER_OPERATOR_CONTEXT_BLOCKED", 422);
|
|
}
|
|
function readEnvelope(buffer) {
|
|
if (buffer.length < 16 || !buffer.subarray(0, 4).equals(MAGIC)) fail("SERVER_OPERATOR_HTTP_INVALID_ENVELOPE");
|
|
const requestBytes = buffer.readUInt32LE(8);
|
|
const sourceBytes = buffer.readUInt32LE(12);
|
|
if (buffer.length > MAX_ENVELOPE || requestBytes < 1 || sourceBytes < 1 || 16 + requestBytes + sourceBytes !== buffer.length) fail("SERVER_OPERATOR_HTTP_INVALID_LENGTH");
|
|
const request = JSON.parse(buffer.subarray(16, 16 + requestBytes).toString("utf8"));
|
|
const source = buffer.subarray(16 + requestBytes);
|
|
if (source.length !== request.sourceBlendByteLength || digest(source) !== request.sourceBlendSha256) fail("SERVER_OPERATOR_HTTP_SOURCE_HASH_MISMATCH", 422);
|
|
validateBindings(request);
|
|
return { request, source };
|
|
}
|
|
function writeJson(response, status, value) { response.statusCode = status; response.setHeader("Content-Type", "application/json"); response.setHeader("Cache-Control", "no-store"); response.end(JSON.stringify(value)); }
|
|
function collect(request, maxBytes) {
|
|
return new Promise((resolve, reject) => { const chunks = []; let total = 0; request.on("data", (chunk) => { total += chunk.length; if (total > maxBytes) { reject(Object.assign(new Error("SERVER_OPERATOR_HTTP_BODY_TOO_LARGE"), { statusCode: 413 })); request.destroy(); return; } chunks.push(chunk); }); request.on("end", () => resolve(Buffer.concat(chunks))); request.on("error", reject); });
|
|
}
|
|
function pythonPayload(request, sourcePath, workspaceRoot, blenderBinary) { return JSON.stringify({ action: "submit", request, sourceBlendPath: sourcePath, workspaceRoot, blenderBinary }); }
|
|
|
|
async function describeServerOperator(operator, options) {
|
|
const cachePrefix = `${options.blenderBinary ?? "blender"}:${operator}`;
|
|
const cached = [...operatorSchemaCache.entries()].find(([key]) => key.startsWith(`${cachePrefix}:`));
|
|
if (cached) return cached[1];
|
|
const workspace = fs.mkdtempSync(path.join(options.workspaceRoot ?? os.tmpdir(), "server-operator-describe-"));
|
|
try {
|
|
const handle = startBoundedServerJobProcess(options.pythonBinary ?? "python3", [RUNNER], { cwd: workspace, timeoutMs: options.describeTimeoutMs ?? 120000, stdio: ["pipe", "pipe", "pipe"], maxOutputBytes: 2 * 1024 * 1024 });
|
|
const chunks = [];
|
|
handle.child.stdout.on("data", (chunk) => chunks.push(Buffer.from(chunk)));
|
|
handle.child.stdin.end(JSON.stringify({ action: "describe", operator, blenderBinary: options.blenderBinary ?? "blender" }));
|
|
await handle.completion;
|
|
const output = Buffer.concat(chunks).toString("utf8").trim();
|
|
if (!output) fail("SERVER_OPERATOR_SCHEMA_EMPTY", 502);
|
|
const result = JSON.parse(output);
|
|
if (result.status === "FAILED") fail(result.errorMessage ?? "SERVER_OPERATOR_SCHEMA_FAILED", 422);
|
|
const buildKey = result.blenderBuild?.buildSha256 ?? "unknown";
|
|
operatorSchemaCache.set(`${cachePrefix}:${buildKey}`, result);
|
|
return result;
|
|
}
|
|
finally {
|
|
fs.rmSync(workspace, { recursive: true, force: true });
|
|
}
|
|
}
|
|
|
|
async function finishJob(entry) {
|
|
if (entry.result) return entry.result;
|
|
await entry.handle.completion;
|
|
const resultPath = path.join(entry.workspace, "result.json");
|
|
const result = fs.existsSync(resultPath) ? JSON.parse(fs.readFileSync(resultPath, "utf8")) : { schemaVersion: 1, route: "SERVER_NATIVE", jobId: entry.request.jobId, status: "FAILED", requestSha256: entry.request.requestSha256, resultSha256: digest(Buffer.from("")), errorMessage: "SERVER_OPERATOR_PROCESS_FAILED" };
|
|
if (result.status === "SUCCEEDED") {
|
|
const blendPath = path.join(entry.workspace, "result.blend");
|
|
if (!fs.existsSync(blendPath)) {
|
|
result.status = "FAILED";
|
|
result.errorMessage = "SERVER_OPERATOR_RESULT_BLEND_MISSING";
|
|
}
|
|
else result.resultBlendBase64 = fs.readFileSync(blendPath).toString("base64");
|
|
}
|
|
entry.result = result;
|
|
if (entry.workspace) {
|
|
fs.rmSync(entry.workspace, { recursive: true, force: true });
|
|
entry.workspace = null;
|
|
}
|
|
return result;
|
|
}
|
|
|
|
async function handle(request, response, options) {
|
|
const pathname = (request.url ?? "/").split("?", 1)[0];
|
|
if (pathname !== (options.path ?? "/api/server-operator-job")) return false;
|
|
const headerAction = request.headers["x-server-operator-action"];
|
|
const action = headerAction ?? ((new URL(request.url, "http://vite.local")).searchParams.has("jobId") ? "read" : "submit");
|
|
try {
|
|
if (action === "describe") {
|
|
const body = JSON.parse((await collect(request, 64 * 1024)).toString("utf8") || "{}");
|
|
const operator = String(body.operator ?? "").trim();
|
|
if (!/^[A-Za-z_][A-Za-z0-9_.-]*$/.test(operator) || Buffer.byteLength(operator) > 128) fail("SERVER_OPERATOR_SCHEMA_OPERATOR_INVALID");
|
|
const result = await describeServerOperator(operator, options);
|
|
writeJson(response, 200, { status: { ready: true }, result });
|
|
return true;
|
|
}
|
|
if (action === "submit") {
|
|
const { request: jobRequest, source } = readEnvelope(await collect(request, MAX_ENVELOPE));
|
|
const workspace = fs.mkdtempSync(path.join(options.workspaceRoot ?? os.tmpdir(), "server-operator-http-"));
|
|
const sourcePath = path.join(workspace, "source.blend");
|
|
fs.writeFileSync(sourcePath, source, { mode: 0o600 });
|
|
const handle = startBoundedServerJobProcess(options.pythonBinary ?? "python3", [RUNNER], { cwd: workspace, timeoutMs: jobRequest.timeoutMs, stdio: ["pipe", "ignore", "pipe"], maxOutputBytes: 8 * 1024 * 1024 });
|
|
const entry = { request: jobRequest, workspace, handle, result: null };
|
|
jobs.set(jobRequest.jobId, entry);
|
|
handle.child.stdin.write(pythonPayload(jobRequest, sourcePath, workspace, options.blenderBinary ?? "blender"));
|
|
handle.child.stdin.end();
|
|
const result = await finishJob(entry);
|
|
writeJson(response, 200, { status: { ready: true }, result });
|
|
return true;
|
|
}
|
|
const body = request.method === "GET" ? {} : JSON.parse((await collect(request, 64 * 1024)).toString("utf8") || "{}");
|
|
const url = new URL(request.url, "http://vite.local");
|
|
const jobId = String(body.jobId ?? url.searchParams.get("jobId") ?? "");
|
|
const requestSha256 = String(body.requestSha256 ?? url.searchParams.get("requestSha256") ?? "");
|
|
const entry = jobs.get(jobId);
|
|
if (!entry || entry.request.requestSha256 !== requestSha256) fail("SERVER_OPERATOR_HTTP_JOB_NOT_FOUND", 404);
|
|
if (action === "cancel") {
|
|
if (!entry.result) await cancelServerJobProcess(entry.handle, async () => {
|
|
if (entry.workspace) {
|
|
fs.rmSync(entry.workspace, { recursive: true, force: true });
|
|
entry.workspace = null;
|
|
}
|
|
}, { graceMs: 500 });
|
|
const result = entry.result ?? { schemaVersion: 1, route: "SERVER_NATIVE", jobId, status: "CANCELLED", sourceBlendSha256: entry.request.sourceBlendSha256, sourceRevision: entry.request.sourceRevision, blenderBuild: entry.request.blenderBuild, propertiesSha256: entry.request.propertiesSha256, requestSha256, resultSha256: digest(Buffer.from("CANCELLED")) };
|
|
entry.result = { ...result, status: "CANCELLED" };
|
|
writeJson(response, 200, { status: { ready: true }, result: entry.result });
|
|
return true;
|
|
}
|
|
const result = entry.result ?? { schemaVersion: 1, route: "SERVER_NATIVE", jobId, status: "RUNNING", sourceBlendSha256: entry.request.sourceBlendSha256, sourceRevision: entry.request.sourceRevision, blenderBuild: entry.request.blenderBuild, propertiesSha256: entry.request.propertiesSha256, requestSha256, resultSha256: digest(Buffer.from("RUNNING")) };
|
|
writeJson(response, 200, { status: { ready: true }, result });
|
|
return true;
|
|
} catch (error) { writeJson(response, error.statusCode ?? 500, { status: { ready: false }, result: { status: "FAILED", errorMessage: String(error.message ?? error) } }); return true; }
|
|
}
|
|
|
|
export function serverOperatorJobProvider(options = {}) {
|
|
return { name: "server-operator-http-provider", configureServer(server) { server.middlewares.use((request, response, next) => { void handle(request, response, options).then((handled) => { if (!handled) next(); }).catch(() => next()); }); }, configurePreviewServer(server) { server.middlewares.use((request, response, next) => { void handle(request, response, options).then((handled) => { if (!handled) next(); }).catch(() => next()); }); } };
|
|
}
|
|
|
|
export { handle as handleServerOperatorJobRequest };
|
|
|
|
if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
|
|
const port = Number.parseInt(process.env.SERVER_OPERATOR_PORT ?? "8787", 10);
|
|
const host = process.env.SERVER_OPERATOR_HOST ?? "127.0.0.1";
|
|
if (!Number.isSafeInteger(port) || port < 1 || port > 65535) throw new Error("SERVER_OPERATOR_PORT_INVALID");
|
|
const options = {
|
|
blenderBinary: process.env.BLENDER_BIN ?? "blender",
|
|
pythonBinary: process.env.PYTHON_BIN ?? "python3",
|
|
workspaceRoot: process.env.SERVER_OPERATOR_WORKSPACE_ROOT,
|
|
};
|
|
http.createServer((request, response) => {
|
|
void handle(request, response, options).then((handled) => {
|
|
if (!handled) writeJson(response, 404, { status: { ready: false }, result: { status: "FAILED", errorMessage: "NOT_FOUND" } });
|
|
}).catch((error) => writeJson(response, 500, { status: { ready: false }, result: { status: "FAILED", errorMessage: String(error?.message ?? error) } }));
|
|
}).listen(port, host, () => process.stdout.write(`server-operator-provider-ready http://${host}:${port}/api/server-operator-job\n`));
|
|
}
|