NODEDC_DEVICE_CORE/services/device-control-core/src/server.mjs

import {
  createPrivateKey,
  createPublicKey,
  timingSafeEqual,
  X509Certificate,
} from "node:crypto";
import { lstat, readFile } from "node:fs/promises";

import { createControlCoreApp } from "./app.mjs";
import { resolveDeviceDatabaseUrl } from "./database-config.mjs";
import {
  createDeviceEdgeChannelSupervisor,
} from "./edge-channel-supervisor.mjs";
import { createDeviceGatewayIngest } from "./gateway-ingest.mjs";
import { PostgresDeviceRepository } from "./postgres-repository.mjs";
import { createTypedCommandRuntime } from "./typed-command-runtime.mjs";

const config = await readConfig();
const repository = new PostgresDeviceRepository({
  databaseUrl: config.databaseUrl,
  poolSize: config.databasePoolSize,
});

await repository.migrate();
const typedCommandRuntime = config.managementApiEnabled && config.edgeChannelEnabled
  ? createTypedCommandRuntime({ repository })
  : null;

const gatewayIngest = config.discoveryIngestEnabled || config.edgeChannelEnabled
  ? createDeviceGatewayIngest({
      repository,
      identifierPepper: config.identifierPepper,
    })
  : null;
const edgeChannelSupervisor = config.edgeChannelEnabled
  ? createDeviceEdgeChannelSupervisor({
      repository,
      gatewayIngest,
      coreIdentity: config.edgeChannelCoreIdentity,
      trustRoot: config.edgeChannelTrustRoot,
      maxEdges: config.edgeChannelMaxEdges,
      reconcileIntervalMs: config.edgeChannelReconcileIntervalMs,
      typedCommandRuntime,
    })
  : null;
await edgeChannelSupervisor?.start();

const server = createControlCoreApp({
  repository,
  gatewayToken: config.gatewayToken,
  identifierPepper: config.identifierPepper,
  discoveryIngestEnabled: config.discoveryIngestEnabled,
  managementApiEnabled: config.managementApiEnabled,
  managementToken: config.managementToken,
  gatewayIngest,
  edgeChannelStatusProvider: edgeChannelSupervisor
    ? () => edgeChannelSupervisor.status()
    : null,
  typedCommandRuntime,
});

server.listen(config.port, config.host, () => {
  console.log(JSON.stringify({
    event: "device_control_core_started",
    host: config.host,
    port: config.port,
    discoveryIngest: config.discoveryIngestEnabled,
    managementApi: config.managementApiEnabled,
    edgeChannels: config.edgeChannelEnabled,
    commandTransport: typedCommandRuntime ? "typed-service-ping-v1" : "disabled",
  }));
});

process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);

async function shutdown() {
  server.close(async () => {
    await edgeChannelSupervisor?.stop();
    await repository.close();
    process.exit(0);
  });
}

async function readConfig() {
  const discoveryIngestEnabled = parseBoolean(
    process.env.DEVICE_DISCOVERY_INGEST_ENABLED,
    false,
  );
  const managementApiEnabled = parseBoolean(
    process.env.DEVICE_MANAGEMENT_API_ENABLED,
    false,
  );
  const edgeChannelEnabled = parseBoolean(
    process.env.DEVICE_EDGE_CHANNEL_ENABLED,
    false,
  );
  const edgeChannelCoreIdentity = edgeChannelEnabled
    ? await readCoreIdentity(process.env)
    : null;
  return {
    host: String(process.env.HOST || "127.0.0.1").trim(),
    port: parsePort(process.env.PORT, 18120),
    databaseUrl: await resolveDeviceDatabaseUrl(process.env),
    databasePoolSize: parsePositiveInt(
      process.env.DEVICE_DATABASE_POOL_SIZE,
      10,
    ),
    discoveryIngestEnabled,
    managementApiEnabled,
    edgeChannelEnabled,
    gatewayToken: discoveryIngestEnabled
      ? await readRequiredSecretFile(
          process.env.DEVICE_GATEWAY_CORE_TOKEN_FILE,
          "device_gateway_core_token_file_required",
        )
      : "",
    identifierPepper:
      discoveryIngestEnabled || managementApiEnabled || edgeChannelEnabled
      ? await readRequiredSecretFile(
          process.env.DEVICE_IDENTIFIER_PEPPER_FILE,
          "device_identifier_pepper_file_required",
        )
      : "",
    managementToken: managementApiEnabled
      ? await readRequiredSecretFile(
          process.env.DEVICE_MANAGEMENT_CORE_TOKEN_FILE,
          "device_management_core_token_file_required",
        )
      : "",
    edgeChannelCoreIdentity,
    edgeChannelTrustRoot: edgeChannelEnabled
      ? await readRequiredDirectory(
          process.env.DEVICE_EDGE_CHANNEL_TRUST_ROOT,
          "device_edge_channel_trust_root_required",
        )
      : "",
    edgeChannelMaxEdges: parseBoundedInt(
      process.env.DEVICE_EDGE_CHANNEL_MAX_EDGES,
      32,
      1,
      64,
    ),
    edgeChannelReconcileIntervalMs: parseBoundedInt(
      process.env.DEVICE_EDGE_CHANNEL_RECONCILE_INTERVAL_MS,
      15_000,
      1_000,
      300_000,
    ),
  };
}

async function readCoreIdentity(environment) {
  const keyPath = requiredValue(
    environment.DEVICE_EDGE_CHANNEL_CORE_KEY_FILE,
    "device_edge_channel_core_key_file_required",
  );
  const certificatePath = requiredValue(
    environment.DEVICE_EDGE_CHANNEL_CORE_CERTIFICATE_FILE,
    "device_edge_channel_core_certificate_file_required",
  );
  const [key, cert] = await Promise.all([
    readBoundedRegularFile(keyPath, 32 * 1024),
    readBoundedRegularFile(certificatePath, 32 * 1024),
  ]);
  const privatePublic = createPublicKey(createPrivateKey(key)).export({
    type: "spki",
    format: "der",
  });
  const certificatePublic = new X509Certificate(cert).publicKey.export({
    type: "spki",
    format: "der",
  });
  if (
    privatePublic.length !== certificatePublic.length
    || !timingSafeEqual(privatePublic, certificatePublic)
  ) {
    throw new Error("device_edge_channel_core_identity_mismatch");
  }
  return Object.freeze({
    identityRef: "workload:device-control-core",
    key,
    cert,
  });
}

async function readBoundedRegularFile(path, maximumBytes) {
  const state = await lstat(path);
  if (
    state.isSymbolicLink()
    || !state.isFile()
    || state.size < 1
    || state.size > maximumBytes
  ) {
    throw new Error("device_edge_channel_core_identity_file_invalid");
  }
  return readFile(path);
}

async function readRequiredDirectory(path, errorCode) {
  const normalized = requiredValue(path, errorCode);
  const state = await lstat(normalized);
  if (state.isSymbolicLink() || !state.isDirectory()) throw new Error(errorCode);
  return normalized;
}

async function readRequiredSecretFile(path, errorCode) {
  const normalized = requiredValue(path, errorCode);
  const value = (await readFile(normalized, "utf8")).trim();
  if (value.length < 32) throw new Error(errorCode);
  return value;
}

function requiredValue(value, errorCode) {
  if (typeof value !== "string" || value.trim() === "") {
    throw new Error(errorCode);
  }
  return value.trim();
}

function parsePort(value, fallback) {
  const parsed = Number(value || fallback);
  if (!Number.isSafeInteger(parsed) || parsed < 1 || parsed > 65535) {
    throw new Error("device_control_port_invalid");
  }
  return parsed;
}

function parsePositiveInt(value, fallback) {
  const parsed = Number(value || fallback);
  if (!Number.isSafeInteger(parsed) || parsed < 1) {
    throw new Error("device_positive_integer_invalid");
  }
  return parsed;
}

function parseBoundedInt(value, fallback, minimum, maximum) {
  const parsed = Number(value ?? fallback);
  if (
    !Number.isSafeInteger(parsed)
    || parsed < minimum
    || parsed > maximum
  ) {
    throw new Error("device_bounded_integer_invalid");
  }
  return parsed;
}

function parseBoolean(value, fallback) {
  if (value === undefined || value === null || value === "") return fallback;
  const normalized = String(value).trim().toLowerCase();
  if (["1", "true", "yes", "on"].includes(normalized)) return true;
  if (["0", "false", "no", "off"].includes(normalized)) return false;
  throw new Error("device_boolean_invalid");
}