feat: add provider-backed microservices
This commit is contained in:
Codex
@@ -32,9 +32,45 @@ interface RuntimeConfig {
|
||||
providerToken: string;
|
||||
heartbeatTimeoutMs: number;
|
||||
taskPendingTimeoutMs: number;
|
||||
microservices: MicroserviceConfig[];
|
||||
logFile: string;
|
||||
}
|
||||
|
||||
interface MicroserviceConfig {
|
||||
id: string;
|
||||
name: string;
|
||||
providerId: string;
|
||||
description: string;
|
||||
repository: {
|
||||
url: string;
|
||||
commitId: string;
|
||||
dockerfile: string;
|
||||
composeFile: string;
|
||||
composeService: string;
|
||||
containerName: string;
|
||||
};
|
||||
backend: {
|
||||
nodeBaseUrl: string;
|
||||
nodeBindHost: string;
|
||||
nodePort: number;
|
||||
proxyMode: string;
|
||||
frontendOnly: boolean;
|
||||
public: boolean;
|
||||
allowedPathPrefixes: string[];
|
||||
healthPath: string;
|
||||
timeoutMs: number;
|
||||
};
|
||||
development: {
|
||||
providerId: string;
|
||||
sshPassthrough: boolean;
|
||||
worktreePath: string;
|
||||
};
|
||||
frontend: {
|
||||
route: string;
|
||||
integrated: boolean;
|
||||
};
|
||||
}
|
||||
|
||||
interface WsData {
|
||||
providerId?: string;
|
||||
channel?: "provider" | "ssh";
|
||||
@@ -86,6 +122,88 @@ function readOptionalNumberEnv(name: string, fallback: number): number {
|
||||
return parsed;
|
||||
}
|
||||
|
||||
function asRecord(value: unknown, name: string): Record<string, unknown> {
|
||||
if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error(`${name} must be an object`);
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
|
||||
function stringFromRecord(value: Record<string, unknown>, key: string, path: string): string {
|
||||
const field = value[key];
|
||||
if (typeof field !== "string" || field.length === 0) throw new Error(`${path}.${key} must be a non-empty string`);
|
||||
return field;
|
||||
}
|
||||
|
||||
function numberFromRecord(value: Record<string, unknown>, key: string, path: string): number {
|
||||
const field = value[key];
|
||||
if (typeof field !== "number" || !Number.isFinite(field) || field <= 0) throw new Error(`${path}.${key} must be a positive number`);
|
||||
return field;
|
||||
}
|
||||
|
||||
function booleanFromRecord(value: Record<string, unknown>, key: string, path: string): boolean {
|
||||
const field = value[key];
|
||||
if (typeof field !== "boolean") throw new Error(`${path}.${key} must be a boolean`);
|
||||
return field;
|
||||
}
|
||||
|
||||
function stringArrayFromRecord(value: Record<string, unknown>, key: string, path: string): string[] {
|
||||
const field = value[key];
|
||||
if (!Array.isArray(field) || field.some((item) => typeof item !== "string" || item.length === 0)) {
|
||||
throw new Error(`${path}.${key} must be an array of non-empty strings`);
|
||||
}
|
||||
return field as string[];
|
||||
}
|
||||
|
||||
function parseMicroserviceConfig(value: unknown, index: number): MicroserviceConfig {
|
||||
const path = `MICROSERVICES_JSON[${index}]`;
|
||||
const item = asRecord(value, path);
|
||||
const repository = asRecord(item.repository, `${path}.repository`);
|
||||
const backend = asRecord(item.backend, `${path}.backend`);
|
||||
const development = asRecord(item.development, `${path}.development`);
|
||||
const frontend = asRecord(item.frontend, `${path}.frontend`);
|
||||
return {
|
||||
id: stringFromRecord(item, "id", path),
|
||||
name: stringFromRecord(item, "name", path),
|
||||
providerId: stringFromRecord(item, "providerId", path),
|
||||
description: stringFromRecord(item, "description", path),
|
||||
repository: {
|
||||
url: stringFromRecord(repository, "url", `${path}.repository`),
|
||||
commitId: stringFromRecord(repository, "commitId", `${path}.repository`),
|
||||
dockerfile: stringFromRecord(repository, "dockerfile", `${path}.repository`),
|
||||
composeFile: stringFromRecord(repository, "composeFile", `${path}.repository`),
|
||||
composeService: stringFromRecord(repository, "composeService", `${path}.repository`),
|
||||
containerName: stringFromRecord(repository, "containerName", `${path}.repository`),
|
||||
},
|
||||
backend: {
|
||||
nodeBaseUrl: stringFromRecord(backend, "nodeBaseUrl", `${path}.backend`),
|
||||
nodeBindHost: stringFromRecord(backend, "nodeBindHost", `${path}.backend`),
|
||||
nodePort: numberFromRecord(backend, "nodePort", `${path}.backend`),
|
||||
proxyMode: stringFromRecord(backend, "proxyMode", `${path}.backend`),
|
||||
frontendOnly: booleanFromRecord(backend, "frontendOnly", `${path}.backend`),
|
||||
public: booleanFromRecord(backend, "public", `${path}.backend`),
|
||||
allowedPathPrefixes: stringArrayFromRecord(backend, "allowedPathPrefixes", `${path}.backend`),
|
||||
healthPath: stringFromRecord(backend, "healthPath", `${path}.backend`),
|
||||
timeoutMs: numberFromRecord(backend, "timeoutMs", `${path}.backend`),
|
||||
},
|
||||
development: {
|
||||
providerId: stringFromRecord(development, "providerId", `${path}.development`),
|
||||
sshPassthrough: booleanFromRecord(development, "sshPassthrough", `${path}.development`),
|
||||
worktreePath: stringFromRecord(development, "worktreePath", `${path}.development`),
|
||||
},
|
||||
frontend: {
|
||||
route: stringFromRecord(frontend, "route", `${path}.frontend`),
|
||||
integrated: booleanFromRecord(frontend, "integrated", `${path}.frontend`),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function readMicroservicesEnv(): MicroserviceConfig[] {
|
||||
const raw = process.env.MICROSERVICES_JSON;
|
||||
if (raw === undefined || raw.length === 0) return [];
|
||||
const parsed = JSON.parse(raw) as unknown;
|
||||
if (!Array.isArray(parsed)) throw new Error("MICROSERVICES_JSON must be an array");
|
||||
return parsed.map(parseMicroserviceConfig);
|
||||
}
|
||||
|
||||
function readConfig(): RuntimeConfig {
|
||||
return {
|
||||
port: readNumberEnv("PORT"),
|
||||
@@ -94,6 +212,7 @@ function readConfig(): RuntimeConfig {
|
||||
providerToken: requiredEnv("PROVIDER_TOKEN"),
|
||||
heartbeatTimeoutMs: readNumberEnv("HEARTBEAT_TIMEOUT_MS"),
|
||||
taskPendingTimeoutMs: readOptionalNumberEnv("TASK_PENDING_TIMEOUT_MS", 10 * 60 * 1000),
|
||||
microservices: readMicroservicesEnv(),
|
||||
logFile: requiredEnv("LOG_FILE"),
|
||||
};
|
||||
}
|
||||
@@ -735,20 +854,54 @@ async function getOverview(): Promise<JsonValue> {
|
||||
};
|
||||
}
|
||||
|
||||
async function dispatchTask(req: Request): Promise<Response> {
|
||||
const body = (await req.json()) as { providerId?: unknown; command?: unknown; payload?: unknown };
|
||||
const providerId = typeof body.providerId === "string" ? body.providerId : "";
|
||||
const command = isProviderDispatchCommand(body.command) ? body.command : null;
|
||||
const payload = typeof body.payload === "object" && body.payload !== null ? (body.payload as Record<string, JsonValue>) : {};
|
||||
if (!providerId) {
|
||||
return jsonResponse({ ok: false, error: "providerId is required" }, 400);
|
||||
}
|
||||
if (command === null) {
|
||||
return jsonResponse({ ok: false, error: "command must be one of docker.ps, provider.upgrade, host.ssh, echo" }, 400);
|
||||
}
|
||||
if (command === "host.ssh" && !(await providerSupports(providerId, "host.ssh"))) {
|
||||
return jsonResponse({ ok: false, error: `provider does not declare host.ssh capability: ${providerId}` }, 409);
|
||||
}
|
||||
function microserviceById(id: string): MicroserviceConfig | null {
|
||||
return config.microservices.find((service) => service.id === id) ?? null;
|
||||
}
|
||||
|
||||
function dockerStatusRecord(status: JsonValue | null): Record<string, unknown> {
|
||||
return typeof status === "object" && status !== null && !Array.isArray(status) ? status as Record<string, unknown> : {};
|
||||
}
|
||||
|
||||
function findContainer(status: JsonValue | null, containerName: string): JsonValue | null {
|
||||
const containers = dockerStatusRecord(status).containers;
|
||||
if (!Array.isArray(containers)) return null;
|
||||
const container = containers.find((item) => {
|
||||
if (typeof item !== "object" || item === null || Array.isArray(item)) return false;
|
||||
return (item as Record<string, unknown>).name === containerName;
|
||||
});
|
||||
return container === undefined ? null : container as JsonValue;
|
||||
}
|
||||
|
||||
async function getMicroservices(): Promise<JsonValue[]> {
|
||||
const nodes = await getNodes();
|
||||
const dockerStatuses = await getNodeDockerStatuses();
|
||||
return config.microservices.map((service) => {
|
||||
const node = nodes.find((item) => item.providerId === service.providerId) ?? null;
|
||||
const docker = dockerStatuses.find((item) => item.providerId === service.providerId) ?? null;
|
||||
const container = findContainer(docker?.dockerStatus ?? null, service.repository.containerName);
|
||||
return {
|
||||
...service,
|
||||
runtime: {
|
||||
providerStatus: node?.status ?? "missing",
|
||||
providerName: node?.name ?? service.providerId,
|
||||
providerLastHeartbeat: node?.lastHeartbeat ?? null,
|
||||
container,
|
||||
backendPortMapping: {
|
||||
providerId: service.providerId,
|
||||
node: `${service.backend.nodeBindHost}:${service.backend.nodePort}`,
|
||||
mainServerAccess: "frontend-only backend proxy",
|
||||
public: service.backend.public,
|
||||
},
|
||||
},
|
||||
} satisfies JsonValue;
|
||||
});
|
||||
}
|
||||
|
||||
async function createAndSendTask(
|
||||
providerId: string,
|
||||
command: CoreDispatchMessage["command"],
|
||||
payload: Record<string, JsonValue>,
|
||||
): Promise<{ taskId: string; providerOnline: boolean }> {
|
||||
const taskId = `task_${Date.now()}_${Math.random().toString(16).slice(2)}`;
|
||||
await sql`
|
||||
INSERT INTO unidesk_tasks (id, provider_id, command, status, payload, result)
|
||||
@@ -757,7 +910,7 @@ async function dispatchTask(req: Request): Promise<Response> {
|
||||
const socket = activeProviders.get(providerId);
|
||||
if (!socket) {
|
||||
await recordEvent("task_queued_provider_offline", providerId, { taskId, providerId, command });
|
||||
return jsonResponse({ ok: true, taskId, status: "queued", providerOnline: false });
|
||||
return { taskId, providerOnline: false };
|
||||
}
|
||||
const dispatch: CoreDispatchMessage = { type: "dispatch", taskId, command, payload };
|
||||
socket.send(JSON.stringify(dispatch));
|
||||
@@ -767,7 +920,127 @@ async function dispatchTask(req: Request): Promise<Response> {
|
||||
WHERE id = ${taskId} AND status = 'queued'
|
||||
`;
|
||||
await recordEvent("task_dispatched", providerId, { taskId, providerId, command });
|
||||
return jsonResponse({ ok: true, taskId, status: "dispatched", providerOnline: true });
|
||||
return { taskId, providerOnline: true };
|
||||
}
|
||||
|
||||
async function rawTask(taskId: string): Promise<{ id: string; provider_id: string; command: string; status: string; payload: JsonValue; result: JsonValue | null; updated_at: Date | string } | null> {
|
||||
const rows = await sql<Array<{ id: string; provider_id: string; command: string; status: string; payload: JsonValue; result: JsonValue | null; updated_at: Date | string }>>`
|
||||
SELECT id, provider_id, command, status, payload, result, updated_at
|
||||
FROM unidesk_tasks
|
||||
WHERE id = ${taskId}
|
||||
LIMIT 1
|
||||
`;
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
async function waitForTaskTerminal(taskId: string, timeoutMs: number): Promise<ReturnType<typeof rawTask> extends Promise<infer T> ? T : never> {
|
||||
const started = Date.now();
|
||||
let latest = await rawTask(taskId);
|
||||
while (Date.now() - started < timeoutMs) {
|
||||
latest = await rawTask(taskId);
|
||||
if (latest !== null && (latest.status === "succeeded" || latest.status === "failed")) return latest;
|
||||
await Bun.sleep(200);
|
||||
}
|
||||
return latest;
|
||||
}
|
||||
|
||||
function isMicroservicePathAllowed(service: MicroserviceConfig, path: string): boolean {
|
||||
return service.backend.allowedPathPrefixes.some((prefix) => path === prefix || path.startsWith(prefix));
|
||||
}
|
||||
|
||||
function readMicroserviceArrayLimits(url: URL): { query: string; jsonArrayLimits: Record<string, JsonValue> } {
|
||||
const params = new URLSearchParams(url.searchParams);
|
||||
const jsonArrayLimits: Record<string, JsonValue> = {};
|
||||
const rawLimits = params.getAll("__unideskArrayLimit");
|
||||
params.delete("__unideskArrayLimit");
|
||||
for (const raw of rawLimits) {
|
||||
for (const entry of raw.split(",")) {
|
||||
const [path = "", limitText = ""] = entry.split(":");
|
||||
if (!/^[A-Za-z0-9_.-]+$/.test(path)) continue;
|
||||
const limit = Number(limitText);
|
||||
if (Number.isInteger(limit) && limit > 0 && limit <= 500) jsonArrayLimits[path] = limit;
|
||||
}
|
||||
}
|
||||
const search = params.toString();
|
||||
return { query: search.length > 0 ? `?${search}` : "", jsonArrayLimits };
|
||||
}
|
||||
|
||||
function responseFromMicroserviceResult(task: Awaited<ReturnType<typeof rawTask>>): Response {
|
||||
if (task === null) return jsonResponse({ ok: false, error: "microservice proxy task missing" }, 502);
|
||||
if (task.status !== "succeeded") return jsonResponse({ ok: false, error: "microservice proxy task failed", task }, 502);
|
||||
const result = dockerStatusRecord(task.result);
|
||||
const status = Number(result.status);
|
||||
const contentType = typeof result.contentType === "string" ? result.contentType : "application/json; charset=utf-8";
|
||||
const bodyText = typeof result.bodyText === "string" ? result.bodyText : "";
|
||||
if (!Number.isInteger(status) || status < 100 || status > 599) {
|
||||
return jsonResponse({ ok: false, error: "microservice proxy returned invalid upstream status", task }, 502);
|
||||
}
|
||||
return new Response(bodyText, {
|
||||
status,
|
||||
headers: { "content-type": contentType },
|
||||
});
|
||||
}
|
||||
|
||||
async function microserviceRoute(req: Request, url: URL): Promise<Response> {
|
||||
const rest = url.pathname.slice("/api/microservices/".length);
|
||||
const slashIndex = rest.indexOf("/");
|
||||
const serviceId = decodeURIComponent(slashIndex === -1 ? rest : rest.slice(0, slashIndex));
|
||||
const suffix = slashIndex === -1 ? "" : rest.slice(slashIndex + 1);
|
||||
const service = microserviceById(serviceId);
|
||||
if (service === null) return jsonResponse({ ok: false, error: `microservice not found: ${serviceId}` }, 404);
|
||||
if (suffix === "" || suffix === "status") return jsonResponse({ ok: true, microservice: (await getMicroservices()).find((item) => recordValue(item, "id") === serviceId) ?? service });
|
||||
if (req.method !== "GET" && req.method !== "HEAD") return jsonResponse({ ok: false, error: "microservice frontend proxy only supports GET/HEAD" }, 405);
|
||||
|
||||
const proxyPrefix = "proxy";
|
||||
const targetPath = suffix === "health"
|
||||
? service.backend.healthPath
|
||||
: suffix === proxyPrefix
|
||||
? "/"
|
||||
: suffix.startsWith(`${proxyPrefix}/`)
|
||||
? `/${suffix.slice(proxyPrefix.length + 1)}`
|
||||
: "";
|
||||
if (targetPath.length === 0) return jsonResponse({ ok: false, error: "microservice route must be /status, /health, or /proxy/<path>" }, 404);
|
||||
if (!isMicroservicePathAllowed(service, targetPath)) {
|
||||
return jsonResponse({ ok: false, error: "microservice path is not allowed", serviceId, targetPath }, 403);
|
||||
}
|
||||
if (!(await providerSupports(service.providerId, "microservice.http"))) {
|
||||
return jsonResponse({ ok: false, error: `provider does not declare microservice.http capability: ${service.providerId}` }, 409);
|
||||
}
|
||||
const proxyOptions = readMicroserviceArrayLimits(url);
|
||||
const { taskId, providerOnline } = await createAndSendTask(service.providerId, "microservice.http", {
|
||||
source: "microservice-frontend-proxy",
|
||||
serviceId: service.id,
|
||||
method: req.method,
|
||||
targetBaseUrl: service.backend.nodeBaseUrl,
|
||||
path: targetPath,
|
||||
query: proxyOptions.query,
|
||||
jsonArrayLimits: proxyOptions.jsonArrayLimits,
|
||||
timeoutMs: service.backend.timeoutMs,
|
||||
});
|
||||
if (!providerOnline) return jsonResponse({ ok: false, error: `provider is offline: ${service.providerId}`, taskId }, 503);
|
||||
const task = await waitForTaskTerminal(taskId, service.backend.timeoutMs + 3000);
|
||||
return responseFromMicroserviceResult(task);
|
||||
}
|
||||
|
||||
async function dispatchTask(req: Request): Promise<Response> {
|
||||
const body = (await req.json()) as { providerId?: unknown; command?: unknown; payload?: unknown };
|
||||
const providerId = typeof body.providerId === "string" ? body.providerId : "";
|
||||
const command = isProviderDispatchCommand(body.command) ? body.command : null;
|
||||
const payload = typeof body.payload === "object" && body.payload !== null ? (body.payload as Record<string, JsonValue>) : {};
|
||||
if (!providerId) {
|
||||
return jsonResponse({ ok: false, error: "providerId is required" }, 400);
|
||||
}
|
||||
if (command === null) {
|
||||
return jsonResponse({ ok: false, error: "command must be one of docker.ps, provider.upgrade, host.ssh, microservice.http, echo" }, 400);
|
||||
}
|
||||
if (command === "host.ssh" && !(await providerSupports(providerId, "host.ssh"))) {
|
||||
return jsonResponse({ ok: false, error: `provider does not declare host.ssh capability: ${providerId}` }, 409);
|
||||
}
|
||||
if (command === "microservice.http" && !(await providerSupports(providerId, "microservice.http"))) {
|
||||
return jsonResponse({ ok: false, error: `provider does not declare microservice.http capability: ${providerId}` }, 409);
|
||||
}
|
||||
const { taskId, providerOnline } = await createAndSendTask(providerId, command, payload);
|
||||
return jsonResponse({ ok: true, taskId, status: providerOnline ? "dispatched" : "queued", providerOnline });
|
||||
}
|
||||
|
||||
function numberFromUnknown(value: unknown, fallback: number, min: number, max: number): number {
|
||||
@@ -889,6 +1162,8 @@ async function route(req: Request, server: Server<WsData>): Promise<Response | u
|
||||
if (url.pathname === "/api/nodes/docker-status") return jsonResponse({ ok: true, dockerStatuses: await getNodeDockerStatuses() });
|
||||
if (url.pathname === "/api/events") return jsonResponse({ ok: true, events: await getEvents(readLimit(url, 100)) });
|
||||
if (url.pathname === "/api/tasks") return jsonResponse({ ok: true, tasks: await getTasks(readLimit(url, 100), url.searchParams.get("status") ?? "all") });
|
||||
if (url.pathname === "/api/microservices") return jsonResponse({ ok: true, microservices: await getMicroservices() });
|
||||
if (url.pathname.startsWith("/api/microservices/")) return microserviceRoute(req, url);
|
||||
if (url.pathname === "/api/dispatch" && req.method === "POST") return dispatchTask(req);
|
||||
if (url.pathname === "/logs") return jsonResponse({ ok: true, logs: recentLogs.slice(-readLimit(url, 100)) });
|
||||
if (url.pathname === "/favicon.ico") return textResponse("", 204);
|
||||
|
||||
Reference in New Issue
Block a user