feat: initialize unidesk platform

src/components/backend-core/Dockerfile

@@ -0,0 +1,7 @@
+FROM oven/bun:1-alpine
+WORKDIR /app/src/components/backend-core
+COPY src/components/backend-core/package.json ./package.json
+RUN bun install --production
+COPY src/components/shared /app/src/components/shared
+COPY src/components/backend-core/src ./src
+CMD ["bun", "run", "src/index.ts"]

src/components/backend-core/package.json

@@ -0,0 +1,12 @@
+{
+  "name": "@unidesk/backend-core",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "start": "bun run src/index.ts",
+    "check": "tsc -p tsconfig.json --noEmit"
+  },
+  "dependencies": {
+    "postgres": "latest"
+  }
+}

src/components/backend-core/src/index.ts

@@ -0,0 +1,484 @@
+import { appendFileSync, mkdirSync } from "node:fs";
+import { dirname } from "node:path";
+import type { Server, ServerWebSocket } from "bun";
+import postgres from "postgres";
+import {
+  type ApiEvent,
+  type ApiNode,
+  type ApiTask,
+  type CoreDispatchMessage,
+  type JsonValue,
+  type ProviderLabels,
+  type ProviderToCoreMessage,
+  isProviderToCoreMessage,
+} from "../../shared/src/index";
+
+interface RuntimeConfig {
+  port: number;
+  databaseUrl: string;
+  providerToken: string;
+  heartbeatTimeoutMs: number;
+  logFile: string;
+}
+
+interface WsData {
+  providerId?: string;
+}
+
+type ProviderSocket = ServerWebSocket<WsData>;
+
+type SqlClient = ReturnType<typeof postgres>;
+
+const recentLogs: unknown[] = [];
+const activeProviders = new Map<string, ProviderSocket>();
+const serviceStartedAt = new Date();
+const config = readConfig();
+const logger = createLogger("backend-core", config.logFile);
+const sql = postgres(config.databaseUrl, {
+  max: 8,
+  idle_timeout: 20,
+  connect_timeout: 10,
+});
+
+let dbReady = false;
+
+function requiredEnv(name: string): string {
+  const value = process.env[name];
+  if (value === undefined || value.length === 0) {
+    throw new Error(`Missing required environment variable: ${name}`);
+  }
+  return value;
+}
+
+function readNumberEnv(name: string): number {
+  const raw = requiredEnv(name);
+  const parsed = Number(raw);
+  if (!Number.isFinite(parsed) || parsed <= 0) {
+    throw new Error(`Environment variable ${name} must be a positive number, got ${raw}`);
+  }
+  return parsed;
+}
+
+function readConfig(): RuntimeConfig {
+  return {
+    port: readNumberEnv("PORT"),
+    databaseUrl: requiredEnv("DATABASE_URL"),
+    providerToken: requiredEnv("PROVIDER_TOKEN"),
+    heartbeatTimeoutMs: readNumberEnv("HEARTBEAT_TIMEOUT_MS"),
+    logFile: requiredEnv("LOG_FILE"),
+  };
+}
+
+function createLogger(service: string, logFile: string) {
+  mkdirSync(dirname(logFile), { recursive: true });
+  return (level: "debug" | "info" | "warn" | "error", message: string, data?: JsonValue): void => {
+    const entry = data === undefined
+      ? { ts: new Date().toISOString(), service, level, message }
+      : { ts: new Date().toISOString(), service, level, message, data };
+    recentLogs.push(entry);
+    while (recentLogs.length > 500) recentLogs.shift();
+    const line = `${JSON.stringify(entry)}\n`;
+    try {
+      appendFileSync(logFile, line, "utf8");
+    } catch (error) {
+      console.error(JSON.stringify({ ts: new Date().toISOString(), service, level: "error", message: "log_write_failed", data: String(error) }));
+    }
+    const consoleMethod = level === "error" ? console.error : level === "warn" ? console.warn : console.log;
+    consoleMethod(line.trimEnd());
+  };
+}
+
+function jsonResponse(body: unknown, status = 200): Response {
+  return new Response(JSON.stringify(body, null, 2), {
+    status,
+    headers: {
+      "content-type": "application/json; charset=utf-8",
+      "access-control-allow-origin": "*",
+      "access-control-allow-methods": "GET,POST,OPTIONS",
+      "access-control-allow-headers": "content-type,x-provider-token",
+    },
+  });
+}
+
+function textResponse(text: string, status = 200): Response {
+  return new Response(text, {
+    status,
+    headers: {
+      "content-type": "text/plain; charset=utf-8",
+      "access-control-allow-origin": "*",
+    },
+  });
+}
+
+async function initDatabase(client: SqlClient): Promise<void> {
+  logger("info", "database_init_start", { databaseUrl: redactDatabaseUrl(config.databaseUrl) });
+  await client`
+    CREATE TABLE IF NOT EXISTS unidesk_nodes (
+      provider_id TEXT PRIMARY KEY,
+      name TEXT NOT NULL,
+      labels JSONB NOT NULL DEFAULT '{}'::jsonb,
+      status TEXT NOT NULL,
+      connected_at TIMESTAMPTZ,
+      last_heartbeat TIMESTAMPTZ,
+      updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+  `;
+  await client`
+    CREATE TABLE IF NOT EXISTS unidesk_events (
+      id BIGSERIAL PRIMARY KEY,
+      type TEXT NOT NULL,
+      source TEXT NOT NULL,
+      payload JSONB NOT NULL DEFAULT '{}'::jsonb,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+  `;
+  await client`
+    CREATE TABLE IF NOT EXISTS unidesk_tasks (
+      id TEXT PRIMARY KEY,
+      provider_id TEXT NOT NULL,
+      command TEXT NOT NULL,
+      status TEXT NOT NULL,
+      payload JSONB NOT NULL DEFAULT '{}'::jsonb,
+      result JSONB,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+      updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+  `;
+  dbReady = true;
+  logger("info", "database_init_complete");
+}
+
+async function initDatabaseWithRetry(): Promise<void> {
+  const started = Date.now();
+  let attempt = 0;
+  while (!dbReady) {
+    attempt += 1;
+    try {
+      await initDatabase(sql);
+      return;
+    } catch (error) {
+      const elapsedMs = Date.now() - started;
+      logger("warn", "database_init_retry", { attempt, elapsedMs, error: errorToJson(error) });
+      if (elapsedMs > 90_000) {
+        logger("error", "database_init_failed", { attempt, elapsedMs, error: errorToJson(error) });
+        throw error;
+      }
+      await Bun.sleep(Math.min(1000 * attempt, 5000));
+    }
+  }
+}
+
+function redactDatabaseUrl(value: string): string {
+  try {
+    const url = new URL(value);
+    if (url.password) url.password = "***";
+    return url.toString();
+  } catch {
+    return "<invalid-url>";
+  }
+}
+
+function errorToJson(error: unknown): JsonValue {
+  if (error instanceof Error) {
+    return { name: error.name, message: error.message, stack: error.stack ?? null };
+  }
+  return String(error);
+}
+
+function compactJson(value: unknown, depth = 0): JsonValue {
+  if (value === null || typeof value === "number" || typeof value === "boolean") return value;
+  if (typeof value === "string") return value.length > 600 ? `${value.slice(0, 600)}...<truncated:${value.length}>` : value;
+  if (Array.isArray(value)) {
+    const items = value.slice(0, 20).map((item) => compactJson(item, depth + 1));
+    if (value.length > 20) items.push({ truncatedItems: value.length - 20 });
+    return items;
+  }
+  if (typeof value === "object" && value !== null) {
+    if (depth >= 4) return "<truncated:depth>";
+    const entries = Object.entries(value as Record<string, unknown>).slice(0, 30);
+    const result: Record<string, JsonValue> = {};
+    for (const [key, item] of entries) result[key] = compactJson(item, depth + 1);
+    const total = Object.keys(value as Record<string, unknown>).length;
+    if (total > entries.length) result.truncatedKeys = total - entries.length;
+    return result;
+  }
+  return String(value);
+}
+
+async function recordEvent(type: string, source: string, payload: JsonValue): Promise<void> {
+  logger("info", type, { source, payload });
+  if (!dbReady) return;
+  try {
+    await sql`
+      INSERT INTO unidesk_events (type, source, payload)
+      VALUES (${type}, ${source}, ${sql.json(payload)})
+    `;
+  } catch (error) {
+    logger("error", "event_insert_failed", { type, source, error: errorToJson(error) });
+  }
+}
+
+async function upsertNodeOnline(providerId: string, name: string, labels: ProviderLabels): Promise<void> {
+  await sql`
+    INSERT INTO unidesk_nodes (provider_id, name, labels, status, connected_at, last_heartbeat, updated_at)
+    VALUES (${providerId}, ${name}, ${sql.json(labels)}, 'online', now(), now(), now())
+    ON CONFLICT (provider_id) DO UPDATE SET
+      name = EXCLUDED.name,
+      labels = EXCLUDED.labels,
+      status = 'online',
+      connected_at = COALESCE(unidesk_nodes.connected_at, EXCLUDED.connected_at),
+      last_heartbeat = now(),
+      updated_at = now()
+  `;
+}
+
+async function touchHeartbeat(providerId: string, labels: ProviderLabels): Promise<void> {
+  await sql`
+    UPDATE unidesk_nodes
+    SET labels = ${sql.json(labels)}, status = 'online', last_heartbeat = now(), updated_at = now()
+    WHERE provider_id = ${providerId}
+  `;
+}
+
+async function markProviderOffline(providerId: string): Promise<void> {
+  activeProviders.delete(providerId);
+  if (!dbReady) return;
+  await sql`
+    UPDATE unidesk_nodes
+    SET status = 'offline', updated_at = now()
+    WHERE provider_id = ${providerId}
+  `;
+  await recordEvent("provider_offline", providerId, { providerId });
+}
+
+async function markStaleProvidersOffline(): Promise<void> {
+  if (!dbReady) return;
+  const timeoutMs = config.heartbeatTimeoutMs;
+  const rows = await sql<{ provider_id: string }[]>`
+    UPDATE unidesk_nodes
+    SET status = 'offline', updated_at = now()
+    WHERE status = 'online'
+      AND last_heartbeat IS NOT NULL
+      AND last_heartbeat < now() - (${timeoutMs} * interval '1 millisecond')
+    RETURNING provider_id
+  `;
+  for (const row of rows) {
+    activeProviders.delete(row.provider_id);
+    await recordEvent("provider_heartbeat_timeout", row.provider_id, { providerId: row.provider_id, timeoutMs });
+  }
+}
+
+function parseMessage(raw: string | Buffer): ProviderToCoreMessage {
+  const text = typeof raw === "string" ? raw : raw.toString("utf8");
+  const parsed = JSON.parse(text) as unknown;
+  if (!isProviderToCoreMessage(parsed)) {
+    throw new Error(`Unsupported provider message: ${text.slice(0, 200)}`);
+  }
+  return parsed;
+}
+
+async function handleProviderMessage(ws: ProviderSocket, raw: string | Buffer): Promise<void> {
+  const message = parseMessage(raw);
+  ws.data.providerId = message.providerId;
+  activeProviders.set(message.providerId, ws);
+
+  if (message.type === "register") {
+    await upsertNodeOnline(message.providerId, message.name, message.labels);
+    await recordEvent("provider_registered", message.providerId, {
+      providerId: message.providerId,
+      name: message.name,
+      labels: message.labels,
+      capabilities: message.capabilities,
+    });
+    ws.send(JSON.stringify({ type: "ack", requestId: "register", ok: true, message: "registered" }));
+    return;
+  }
+
+  if (message.type === "heartbeat") {
+    await touchHeartbeat(message.providerId, message.labels);
+    logger("debug", "provider_heartbeat", { providerId: message.providerId, labels: message.labels });
+    return;
+  }
+
+  await sql`
+    UPDATE unidesk_tasks
+    SET status = ${message.status}, result = ${sql.json(message.result ?? { message: message.message })}, updated_at = now()
+    WHERE id = ${message.taskId}
+  `;
+  await recordEvent("task_status", message.providerId, {
+    providerId: message.providerId,
+    taskId: message.taskId,
+    status: message.status,
+    message: message.message,
+    result: message.result ?? null,
+  });
+}
+
+async function getNodes(): Promise<ApiNode[]> {
+  const rows = await sql<Array<Record<string, unknown>>>`
+    SELECT provider_id, name, labels, status, connected_at, last_heartbeat
+    FROM unidesk_nodes
+    ORDER BY status DESC, provider_id ASC
+  `;
+  return rows.map((row) => ({
+    providerId: String(row.provider_id),
+    name: String(row.name),
+    status: row.status === "online" ? "online" : "offline",
+    labels: (row.labels ?? {}) as ProviderLabels,
+    connectedAt: row.connected_at instanceof Date ? row.connected_at.toISOString() : row.connected_at === null ? null : String(row.connected_at),
+    lastHeartbeat: row.last_heartbeat instanceof Date ? row.last_heartbeat.toISOString() : row.last_heartbeat === null ? null : String(row.last_heartbeat),
+  }));
+}
+
+async function getEvents(limit: number): Promise<ApiEvent[]> {
+  const rows = await sql<Array<Record<string, unknown>>>`
+    SELECT id, type, source, payload, created_at
+    FROM unidesk_events
+    ORDER BY id DESC
+    LIMIT ${limit}
+  `;
+  return rows.map((row) => ({
+    id: Number(row.id),
+    type: String(row.type),
+    source: String(row.source),
+    payload: compactJson(row.payload ?? {}),
+    createdAt: row.created_at instanceof Date ? row.created_at.toISOString() : String(row.created_at),
+  }));
+}
+
+async function getTasks(limit: number): Promise<ApiTask[]> {
+  const rows = await sql<Array<Record<string, unknown>>>`
+    SELECT id, provider_id, command, status, payload, result, created_at, updated_at
+    FROM unidesk_tasks
+    ORDER BY updated_at DESC
+    LIMIT ${limit}
+  `;
+  return rows.map((row) => ({
+    id: String(row.id),
+    providerId: String(row.provider_id),
+    command: String(row.command),
+    status: String(row.status),
+    payload: compactJson(row.payload ?? {}),
+    result: compactJson(row.result ?? null),
+    createdAt: row.created_at instanceof Date ? row.created_at.toISOString() : String(row.created_at),
+    updatedAt: row.updated_at instanceof Date ? row.updated_at.toISOString() : String(row.updated_at),
+  }));
+}
+
+async function getOverview(): Promise<JsonValue> {
+  const nodes = await getNodes();
+  const tasks = await getTasks(50);
+  const online = nodes.filter((node) => node.status === "online").length;
+  const pendingTasks = tasks.filter((task) => task.status === "queued" || task.status === "dispatched" || task.status === "running").length;
+  return {
+    service: "unidesk-core",
+    ok: true,
+    dbReady,
+    uptimeSeconds: Math.floor((Date.now() - serviceStartedAt.getTime()) / 1000),
+    nodeCount: nodes.length,
+    onlineNodeCount: online,
+    pendingTaskCount: pendingTasks,
+    activeSocketCount: activeProviders.size,
+    heartbeatTimeoutMs: config.heartbeatTimeoutMs,
+  };
+}
+
+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 = body.command === "docker.ps" || body.command === "echo" ? body.command : "echo";
+  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);
+  }
+  const taskId = `task_${Date.now()}_${Math.random().toString(16).slice(2)}`;
+  await sql`
+    INSERT INTO unidesk_tasks (id, provider_id, command, status, payload, result)
+    VALUES (${taskId}, ${providerId}, ${command}, 'queued', ${sql.json(payload)}, NULL)
+  `;
+  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 });
+  }
+  const dispatch: CoreDispatchMessage = { type: "dispatch", taskId, command, payload };
+  socket.send(JSON.stringify(dispatch));
+  await sql`UPDATE unidesk_tasks SET status = 'dispatched', updated_at = now() WHERE id = ${taskId}`;
+  await recordEvent("task_dispatched", providerId, { taskId, providerId, command });
+  return jsonResponse({ ok: true, taskId, status: "dispatched", providerOnline: true });
+}
+
+async function route(req: Request, server: Server<WsData>): Promise<Response | undefined> {
+  const url = new URL(req.url);
+  if (req.method === "OPTIONS") return jsonResponse({ ok: true });
+
+  if (url.pathname === "/ws/provider") {
+    const token = url.searchParams.get("token") ?? req.headers.get("x-provider-token");
+    if (token !== config.providerToken) {
+      await recordEvent("provider_auth_failed", "unknown", { remote: req.headers.get("x-forwarded-for") ?? null });
+      return jsonResponse({ ok: false, error: "invalid provider token" }, 401);
+    }
+    const upgraded = server.upgrade(req, { data: {} satisfies WsData });
+    return upgraded ? undefined : jsonResponse({ ok: false, error: "websocket upgrade failed" }, 400);
+  }
+
+  try {
+    if (url.pathname === "/" || url.pathname === "/health") {
+      return jsonResponse({ ok: true, service: "unidesk-core", dbReady, startedAt: serviceStartedAt.toISOString() });
+    }
+    if (!dbReady && url.pathname.startsWith("/api/")) {
+      return jsonResponse({ ok: false, error: "database not ready" }, 503);
+    }
+    if (url.pathname === "/api/overview") return jsonResponse(await getOverview());
+    if (url.pathname === "/api/nodes") return jsonResponse({ ok: true, nodes: await getNodes() });
+    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)) });
+    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);
+    return jsonResponse({ ok: false, error: "not found", path: url.pathname }, 404);
+  } catch (error) {
+    logger("error", "request_failed", { path: url.pathname, error: errorToJson(error) });
+    return jsonResponse({ ok: false, error: errorToJson(error) }, 500);
+  }
+}
+
+function readLimit(url: URL, defaultLimit: number): number {
+  const raw = url.searchParams.get("limit");
+  if (raw === null) return defaultLimit;
+  const parsed = Number(raw);
+  if (!Number.isInteger(parsed) || parsed <= 0) return defaultLimit;
+  return Math.min(parsed, 500);
+}
+
+await initDatabaseWithRetry();
+
+const server = Bun.serve<WsData>({
+  port: config.port,
+  hostname: "0.0.0.0",
+  fetch: route,
+  websocket: {
+    open(ws) {
+      logger("info", "provider_socket_open", { remoteAddress: ws.remoteAddress });
+    },
+    message(ws, raw) {
+      handleProviderMessage(ws, raw).catch((error) => {
+        logger("error", "provider_message_failed", { providerId: ws.data.providerId ?? null, error: errorToJson(error) });
+        ws.send(JSON.stringify({ type: "ack", requestId: "message", ok: false, message: String(error) }));
+      });
+    },
+    close(ws) {
+      const providerId = ws.data.providerId;
+      logger("warn", "provider_socket_close", { providerId: providerId ?? null });
+      if (providerId !== undefined) {
+        markProviderOffline(providerId).catch((error) => logger("error", "provider_offline_mark_failed", { providerId, error: errorToJson(error) }));
+      }
+    },
+  },
+});
+
+setInterval(() => {
+  markStaleProvidersOffline().catch((error) => logger("error", "heartbeat_sweep_failed", { error: errorToJson(error) }));
+}, 10_000);
+
+logger("info", "server_listening", { url: `http://0.0.0.0:${server.port}`, logFile: config.logFile });

src/components/backend-core/tsconfig.json

@@ -0,0 +1,18 @@
+{
+  "compilerOptions": {
+    "composite": true,
+    "target": "ES2022",
+    "module": "ESNext",
+    "moduleResolution": "Bundler",
+    "types": ["bun", "node"],
+    "strict": true,
+    "noImplicitReturns": true,
+    "noFallthroughCasesInSwitch": true,
+    "declaration": true,
+    "emitDeclarationOnly": true,
+    "outDir": "dist",
+    "skipLibCheck": true
+  },
+  "include": ["src/**/*.ts"],
+  "references": [{ "path": "../shared" }]
+}