NODEDC_PLATFORM/device-plane/services/device-gateway/src/server.mjs

import { readFile } from "node:fs/promises";

import {
  DEVICE_ADAPTER_CATALOG,
} from "../../../packages/device-adapter-catalog/src/index.mjs";
import { createCoreGatewayClient } from "./core-client.mjs";
import { createDeviceGatewayRuntime } from "./runtime.mjs";

const config = await readConfig();
const coreClient = config.listenEnabled
  ? createCoreGatewayClient({
      coreUrl: config.coreUrl,
      gatewayToken: config.gatewayToken,
      timeoutMs: config.coreTimeoutMs,
    })
  : undefined;
const runtime = createDeviceGatewayRuntime({
  listenEnabled: config.listenEnabled,
  publicIngressEnabled: config.publicIngressEnabled,
  coreChannelAuthenticated: false,
  adapterRegistry: DEVICE_ADAPTER_CATALOG.registry,
  protocolProfileRef: config.protocolProfileRef,
  edgeRef: config.edgeRef,
  routeRef: config.routeRef,
  healthHost: config.healthHost,
  healthPort: config.healthPort,
  tcpHost: config.tcpHost,
  tcpPort: config.tcpPort,
  maxBufferedBytes: config.maxBufferedBytes,
  maxAggregateBufferedBytes: config.maxAggregateBufferedBytes,
  maxConcurrentSessions: config.maxConcurrentSessions,
  maxSessionsPerAddress: config.maxSessionsPerAddress,
  maxConnectionsPerMinutePerAddress:
    config.maxConnectionsPerMinutePerAddress,
  maxTrackedSourceAddresses: config.maxTrackedSourceAddresses,
  sessionTimeoutMs: config.sessionTimeoutMs,
  onDiscovery: coreClient?.observeDiscovery,
  onMessage: coreClient?.acceptMessage,
});

const addresses = await runtime.start();
console.log(JSON.stringify({
  event: "device_gateway_started",
  health: addresses.healthAddress,
  tcp: addresses.tcpAddress,
  publicIngress: config.publicIngressEnabled
    ? "telemetry-ingest"
    : "disabled",
  commandTransport: "disabled",
}));

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

async function shutdown() {
  await runtime.stop();
  process.exit(0);
}

async function readConfig() {
  const listenEnabled = parseBoolean(
    process.env.DEVICE_GATEWAY_LISTEN_ENABLED,
    false,
  );
  const publicIngressEnabled = parseBoolean(
    process.env.DEVICE_GATEWAY_PUBLIC_INGRESS_ENABLED,
    false,
  );
  return {
    listenEnabled,
    publicIngressEnabled,
    protocolProfileRef: String(
      process.env.DEVICE_GATEWAY_PROTOCOL_PROFILE_REF
      || DEVICE_ADAPTER_CATALOG.defaultProfileRef,
    ),
    edgeRef: listenEnabled
      ? requiredValue(
          process.env.DEVICE_GATEWAY_EDGE_REF,
          "device_gateway_edge_ref_required",
        )
      : "edge:disabled",
    routeRef: String(process.env.DEVICE_GATEWAY_ROUTE_REF || ""),
    healthHost: String(
      process.env.DEVICE_GATEWAY_HEALTH_HOST || "127.0.0.1",
    ),
    healthPort: parsePort(process.env.DEVICE_GATEWAY_HEALTH_PORT, 18121),
    tcpHost: String(
      process.env.DEVICE_GATEWAY_TCP_HOST
      || (publicIngressEnabled ? "0.0.0.0" : "127.0.0.1"),
    ),
    tcpPort: parsePort(process.env.DEVICE_GATEWAY_TCP_PORT, 9921),
    maxBufferedBytes: parsePositiveInt(
      process.env.DEVICE_GATEWAY_MAX_BUFFERED_BYTES,
      65536,
    ),
    maxAggregateBufferedBytes: parsePositiveInt(
      process.env.DEVICE_GATEWAY_MAX_AGGREGATE_BUFFERED_BYTES,
      32 * 1024 * 1024,
    ),
    maxConcurrentSessions: parsePositiveInt(
      process.env.DEVICE_GATEWAY_MAX_SESSIONS,
      128,
    ),
    maxSessionsPerAddress: parsePositiveInt(
      process.env.DEVICE_GATEWAY_MAX_SESSIONS_PER_ADDRESS,
      16,
    ),
    maxConnectionsPerMinutePerAddress: parsePositiveInt(
      process.env.DEVICE_GATEWAY_MAX_CONNECTIONS_PER_MINUTE_PER_ADDRESS,
      60,
    ),
    maxTrackedSourceAddresses: parsePositiveInt(
      process.env.DEVICE_GATEWAY_MAX_TRACKED_SOURCE_ADDRESSES,
      2048,
    ),
    sessionTimeoutMs: parsePositiveInt(
      process.env.DEVICE_GATEWAY_SESSION_TIMEOUT_MS,
      10000,
    ),
    coreUrl: listenEnabled
      ? requiredValue(
          process.env.DEVICE_GATEWAY_CORE_URL,
          "device_gateway_core_url_required",
        )
      : "",
    gatewayToken: listenEnabled
      ? await readRequiredSecretFile(
          process.env.DEVICE_GATEWAY_CORE_TOKEN_FILE,
          "device_gateway_core_token_file_required",
        )
      : "",
    coreTimeoutMs: parsePositiveInt(
      process.env.DEVICE_GATEWAY_CORE_TIMEOUT_MS,
      5000,
    ),
  };
}

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_gateway_port_invalid");
  }
  return parsed;
}

function parsePositiveInt(value, fallback) {
  const parsed = Number(value || fallback);
  if (!Number.isSafeInteger(parsed) || parsed < 1) {
    throw new Error("device_gateway_positive_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_gateway_boolean_invalid");
}