feat(device-plane): supervise pinned edge channels in core

device-plane/services/device-control-core/src/app.mjs

@@ -2,16 +2,11 @@ import { createHash, timingSafeEqual } from "node:crypto";
 import { createServer } from "node:http";
 
 import {
-  assertSafeProjection,
   hashRestrictedIdentifier,
   maskRestrictedIdentifier,
-  normalizeAdapterAcceptance,
-  normalizeAdapterMessage,
-  normalizeDiscoverySignal,
   normalizeRestrictedIdentifier,
-  toSafeDiscoveryView,
-  toSafeAdapterMessageView,
 } from "../../../packages/device-protocol-contract/src/index.mjs";
+import { createDeviceGatewayIngest } from "./gateway-ingest.mjs";
 import {
   normalizeManagementActor,
 } from "./project-management.mjs";
@@ -59,6 +54,8 @@ export function createControlCoreApp({
   discoveryIngestEnabled = false,
   managementApiEnabled = false,
   managementToken = "",
+  gatewayIngest = null,
+  edgeChannelStatusProvider = null,
 } = {}) {
   if (!repository || typeof repository.health !== "function") {
     throw new TypeError("device_repository_required");
@@ -88,6 +85,24 @@ export function createControlCoreApp({
       throw new TypeError("device_identifier_pepper_invalid");
     }
   }
+  const ingest = discoveryIngestEnabled
+    ? gatewayIngest ?? createDeviceGatewayIngest({ repository, identifierPepper })
+    : gatewayIngest;
+  if (
+    ingest
+    && (
+      typeof ingest.observeDiscovery !== "function"
+      || typeof ingest.acceptMessage !== "function"
+    )
+  ) {
+    throw new TypeError("device_gateway_ingest_invalid");
+  }
+  if (
+    edgeChannelStatusProvider != null
+    && typeof edgeChannelStatusProvider !== "function"
+  ) {
+    throw new TypeError("device_edge_channel_status_provider_invalid");
+  }
 
   const server = createServer(async (request, response) => {
     response.setHeader("Content-Type", "application/json; charset=utf-8");
@@ -108,6 +123,9 @@ export function createControlCoreApp({
           database,
           discoveryIngest: discoveryIngestEnabled ? "enabled" : "disabled",
           managementApi: managementApiEnabled ? "enabled" : "disabled",
+          edgeChannels: edgeChannelStatusProvider
+            ? edgeChannelStatusProvider()
+            : { enabled: false, configured: 0, accepted: 0, degraded: 0 },
           commandTransport: "disabled",
         });
       }
@@ -236,22 +254,11 @@ export function createControlCoreApp({
         }
 
         const input = await readJsonBody(request, 32 * 1024);
-        const signal = normalizeDiscoverySignal(input);
-        const identifierDigest = hashRestrictedIdentifier(
-          signal.identifier,
-          identifierPepper,
-        );
-        const safeView = assertSafeProjection(toSafeDiscoveryView(signal));
-        const discovery = await repository.upsertQuarantineDiscovery({
-          identifierDigest,
-          safeView,
-          sessionRef: signal.sessionRef,
-          routeRef: signal.routeRef ?? null,
-        });
+        const discovery = await ingest.observeDiscovery(input);
         return writeJson(response, discovery.created ? 201 : 200, {
           ok: true,
           created: discovery.created,
-          discovery: assertSafeProjection(discovery.value),
+          discovery: discovery.value,
         });
       }
 
@@ -273,31 +280,7 @@ export function createControlCoreApp({
         }
 
         const input = await readJsonBody(request, 1024 * 1024);
-        const message = normalizeAdapterMessage(input);
-        const identifierDigest = hashRestrictedIdentifier(
-          message.identifier,
-          identifierPepper,
-        );
-        const safeView = assertSafeProjection(toSafeAdapterMessageView(message));
-        const requestDigest = gatewayMessageRequestDigest({
-          edgeRef: safeView.edgeRef,
-          adapterRef: safeView.adapterRef,
-          protocolProfileRef: safeView.protocolProfileRef,
-          protocol: safeView.protocol,
-          routeRef: safeView.routeRef ?? null,
-          idempotencyKey: safeView.idempotencyKey,
-          identifierKind: safeView.identifier.kind,
-          identifierDigest,
-          payloadSchemaRef: safeView.payloadSchemaRef,
-          payload: safeView.payload,
-        });
-        const acceptance = normalizeAdapterAcceptance(
-          await repository.acceptAdapterMessage({
-            identifierDigest,
-            requestDigest,
-            safeView,
-          }),
-        );
+        const acceptance = await ingest.acceptMessage(input);
         return writeJson(response, acceptance.replayed ? 200 : 201, {
           ok: true,
           acceptance,
@@ -421,12 +404,6 @@ function managementRequestDigest(value) {
     .digest("hex")}`;
 }
 
-function gatewayMessageRequestDigest(value) {
-  return `sha256:${createHash("sha256")
-    .update(JSON.stringify(value), "utf8")
-    .digest("hex")}`;
-}
-
 function matchesBearer(header, expected) {
   if (typeof header !== "string" || !header.startsWith("Bearer ")) return false;
   const actual = Buffer.from(header.slice("Bearer ".length), "utf8");