79 lines
3.1 KiB
TypeScript
79 lines
3.1 KiB
TypeScript
import { parentPort } from "node:worker_threads";
|
|
import { join } from "node:path";
|
|
|
|
import { WorkerAiCallGate } from "./ai-call-gate.js";
|
|
import { readConfiguredLocalDataRoot } from "./runtime-config.js";
|
|
import { RetentionCleanup } from "./retention-cleanup.js";
|
|
import { ProjectPurgeCleanup } from "./project-purge-cleanup.js";
|
|
import { StructuredJsonlLogger } from "./structured-log.js";
|
|
import { WorkerStorageStatus } from "./storage-status.js";
|
|
import { attachWorkerSupervisorControl, initializeWorkerCredentialClient, receiveWorkerCredentials } from "./supervisor-channel.js";
|
|
|
|
export function handleWorkerProbe(message: unknown) {
|
|
return message === "ping" ? "pong" : null;
|
|
}
|
|
|
|
const workerPort = parentPort;
|
|
|
|
if (workerPort) {
|
|
workerPort.on("message", (message: unknown) => {
|
|
workerPort.postMessage(handleWorkerProbe(message));
|
|
});
|
|
}
|
|
|
|
if (!workerPort && process.argv.includes("--dada-credential-stdin")) {
|
|
initializeWorkerCredentialClient(await receiveWorkerCredentials());
|
|
const controlPipeIndex = process.argv.indexOf("--dada-control-pipe");
|
|
const controlPipe = process.argv[controlPipeIndex + 1];
|
|
if (controlPipeIndex < 0 || !controlPipe) throw new Error("Supervisor control pipe name is required.");
|
|
const keepAlive = setInterval(() => undefined, 30_000);
|
|
let storage: WorkerStorageStatus | undefined;
|
|
let retention: RetentionCleanup | undefined;
|
|
let projectCleanup: ProjectPurgeCleanup | undefined;
|
|
let retentionTimer: ReturnType<typeof setInterval> | undefined;
|
|
const control = attachWorkerSupervisorControl(controlPipe, () => {
|
|
clearInterval(keepAlive);
|
|
if (retentionTimer) clearInterval(retentionTimer);
|
|
retention?.close();
|
|
projectCleanup?.close();
|
|
storage?.close();
|
|
});
|
|
let storageStatus: "active" | "unavailable" = "active";
|
|
try {
|
|
const dataRoot = readConfiguredLocalDataRoot();
|
|
const databasePath = join(dataRoot, "db", "dada.sqlite3");
|
|
storage = new WorkerStorageStatus(databasePath);
|
|
retention = new RetentionCleanup({ databasePath });
|
|
projectCleanup = new ProjectPurgeCleanup({ dataRoot, databasePath });
|
|
const runRetentionCleanup = () => {
|
|
try {
|
|
retention?.purgeExpired();
|
|
projectCleanup?.run();
|
|
} catch {
|
|
storageStatus = "unavailable";
|
|
storage?.markLogUnavailable();
|
|
control.reportStatus("storage_unavailable");
|
|
}
|
|
};
|
|
runRetentionCleanup();
|
|
retentionTimer = setInterval(runRetentionCleanup, 24 * 60 * 60 * 1_000);
|
|
const logger = new StructuredJsonlLogger({
|
|
component: "worker",
|
|
directory: join(dataRoot, "logs", "worker"),
|
|
onWriteFailure: () => {
|
|
storageStatus = "unavailable";
|
|
try {
|
|
storage?.markLogUnavailable();
|
|
} finally {
|
|
control.reportStatus("storage_unavailable");
|
|
}
|
|
},
|
|
});
|
|
logger.write({ error_category: "none", status_category: "ready" });
|
|
new WorkerAiCallGate({ getStorageStatus: () => storageStatus === "unavailable" ? storageStatus : (storage?.getStatus() ?? "unavailable"), logger });
|
|
} catch {
|
|
storageStatus = "unavailable";
|
|
control.reportStatus("storage_unavailable");
|
|
}
|
|
}
|