NODEDC_PLATFORM/device-plane/packages/device-edge-channel-contract/src/index.mjs

export const DEVICE_EDGE_CHANNEL_SCHEMA =
  "nodedc.device-edge.channel-envelope.v1";

export const DEVICE_EDGE_CHANNEL_PATH =
  "/internal/v1/device-edge/channel";

export const DEVICE_EDGE_CHANNEL_LIMITS = Object.freeze({
  maxEnvelopeBytes: 1024 * 1024,
  maxPendingAcceptances: 128,
  keepaliveMs: 15_000,
  deadPeerMs: 45_000,
  reconnectMinimumMs: 1_000,
  reconnectMaximumMs: 30_000,
});

export const EDGE_TO_CORE_MESSAGE_KINDS = Object.freeze([
  "channel.hello",
  "channel.heartbeat",
  "tracker.session-opened",
  "tracker.session-closed",
  "discovery.observed",
  "adapter.message",
  "delivery.acknowledged",
  "command.status",
  "channel.counters",
]);

export const CORE_TO_EDGE_MESSAGE_KINDS = Object.freeze([
  "channel.accepted",
  "channel.heartbeat",
  "flow.window",
  "session.disposition",
  "event.accepted",
  "event.rejected",
]);

const OPAQUE_REF_RE = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/;
const ISO_TIMESTAMP_RE = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{3})?Z$/;
const forbiddenKeyFragments = Object.freeze([
  "password",
  "secret",
  "credential",
  "authorization",
  "privatekey",
  "bearertoken",
]);

export function normalizeChannelEnvelope(input, options = {}) {
  assertPlainObject(input, "device_edge_channel_envelope");
  const direction = normalizeDirection(options.direction);
  const maxEnvelopeBytes = normalizeLimit(
    options.maxEnvelopeBytes,
    DEVICE_EDGE_CHANNEL_LIMITS.maxEnvelopeBytes,
  );
  const allowedKeys = new Set([
    "schemaVersion",
    "edgeRegistrationId",
    "channelGeneration",
    "trackerSessionId",
    "adapterProfileRef",
    "sequence",
    "eventAt",
    "receivedAt",
    "payloadBytes",
    "messageKind",
    "correlationId",
    "payload",
  ]);
  rejectUnexpectedKeys(input, allowedKeys);
  rejectForbiddenKeys(input);

  if (input.schemaVersion !== DEVICE_EDGE_CHANNEL_SCHEMA) {
    throw new TypeError("device_edge_channel_schema_invalid");
  }
  const allowedKinds = direction === "edge-to-core"
    ? EDGE_TO_CORE_MESSAGE_KINDS
    : CORE_TO_EDGE_MESSAGE_KINDS;
  if (!allowedKinds.includes(input.messageKind)) {
    throw new TypeError("device_edge_channel_message_kind_invalid");
  }
  const sequence = Number(input.sequence);
  if (!Number.isSafeInteger(sequence) || sequence < 1) {
    throw new TypeError("device_edge_channel_sequence_invalid");
  }
  assertJsonValue(input.payload, 0);
  const payload = cloneJsonValue(input.payload);
  const payloadBytes = Buffer.byteLength(JSON.stringify(payload), "utf8");
  if (payloadBytes > maxEnvelopeBytes) {
    throw new TypeError("device_edge_channel_payload_too_large");
  }
  if (Number(input.payloadBytes) !== payloadBytes) {
    throw new TypeError("device_edge_channel_payload_length_mismatch");
  }

  const normalized = {
    schemaVersion: DEVICE_EDGE_CHANNEL_SCHEMA,
    edgeRegistrationId: normalizeRef(
      input.edgeRegistrationId,
      "edge_registration_id",
    ),
    channelGeneration: normalizeRef(
      input.channelGeneration,
      "channel_generation",
    ),
    trackerSessionId: normalizeRef(
      input.trackerSessionId,
      "tracker_session_id",
    ),
    adapterProfileRef: normalizeRef(
      input.adapterProfileRef,
      "adapter_profile_ref",
    ),
    sequence,
    eventAt: normalizeTimestamp(input.eventAt, "event_at"),
    receivedAt: normalizeTimestamp(input.receivedAt, "received_at"),
    payloadBytes,
    messageKind: input.messageKind,
    correlationId: normalizeRef(input.correlationId, "correlation_id"),
    payload,
  };
  return deepFreeze(normalized);
}

export function createChannelEnvelope(fields, options = {}) {
  assertPlainObject(fields, "device_edge_channel_fields");
  const payload = fields.payload ?? {};
  assertJsonValue(payload, 0);
  const clonedPayload = cloneJsonValue(payload);
  return normalizeChannelEnvelope({
    ...fields,
    schemaVersion: DEVICE_EDGE_CHANNEL_SCHEMA,
    payloadBytes: Buffer.byteLength(JSON.stringify(clonedPayload), "utf8"),
    payload: clonedPayload,
  }, options);
}

export function encodeChannelEnvelope(envelope, options = {}) {
  const normalized = normalizeChannelEnvelope(envelope, options);
  const encoded = Buffer.from(`${JSON.stringify(normalized)}\n`, "utf8");
  const maxEnvelopeBytes = normalizeLimit(
    options.maxEnvelopeBytes,
    DEVICE_EDGE_CHANNEL_LIMITS.maxEnvelopeBytes,
  );
  if (encoded.length > maxEnvelopeBytes) {
    throw new TypeError("device_edge_channel_envelope_too_large");
  }
  return encoded;
}

export function createChannelEnvelopeDecoder(options = {}) {
  const direction = normalizeDirection(options.direction);
  const maxEnvelopeBytes = normalizeLimit(
    options.maxEnvelopeBytes,
    DEVICE_EDGE_CHANNEL_LIMITS.maxEnvelopeBytes,
  );
  let buffered = Buffer.alloc(0);

  return Object.freeze({
    push(chunk) {
      if (!Buffer.isBuffer(chunk) && !(chunk instanceof Uint8Array)) {
        throw new TypeError("device_edge_channel_chunk_invalid");
      }
      let incoming = Buffer.from(chunk);
      const envelopes = [];
      while (incoming.length > 0) {
        const newlineIndex = incoming.indexOf(0x0a);
        if (newlineIndex < 0) {
          if (buffered.length + incoming.length > maxEnvelopeBytes) {
            throw new TypeError("device_edge_channel_envelope_too_large");
          }
          buffered = buffered.length === 0
            ? incoming
            : Buffer.concat([buffered, incoming]);
          break;
        }
        if (buffered.length + newlineIndex === 0) {
          throw new TypeError("device_edge_channel_envelope_empty");
        }
        if (buffered.length + newlineIndex + 1 > maxEnvelopeBytes) {
          throw new TypeError("device_edge_channel_envelope_too_large");
        }
        const segment = incoming.subarray(0, newlineIndex);
        const line = buffered.length === 0
          ? segment
          : Buffer.concat([buffered, segment]);
        buffered = Buffer.alloc(0);
        incoming = incoming.subarray(newlineIndex + 1);
        let parsed;
        try {
          parsed = JSON.parse(line.toString("utf8"));
        } catch {
          throw new TypeError("device_edge_channel_envelope_json_invalid");
        }
        envelopes.push(normalizeChannelEnvelope(parsed, {
          direction,
          maxEnvelopeBytes,
        }));
      }
      return envelopes;
    },
    finish() {
      if (buffered.length !== 0) {
        throw new TypeError("device_edge_channel_envelope_truncated");
      }
    },
    bufferedBytes() {
      return buffered.length;
    },
  });
}

export function normalizeCertificateFingerprint(value) {
  const compact = String(value || "").replaceAll(":", "").toUpperCase();
  if (!/^[A-F0-9]{64}$/.test(compact)) {
    throw new TypeError("device_edge_channel_certificate_fingerprint_invalid");
  }
  return compact.match(/.{2}/g).join(":");
}

export function normalizeCertificateIdentities(value) {
  if (!Array.isArray(value) || value.length < 1 || value.length > 2) {
    throw new TypeError("device_edge_channel_certificate_identities_invalid");
  }
  const generations = new Set();
  const fingerprints = new Set();
  let activeCount = 0;
  const identities = value.map((identity) => {
    assertPlainObject(identity, "device_edge_channel_certificate_identity");
    rejectUnexpectedKeys(
      identity,
      new Set(["generationRef", "fingerprint", "status"]),
    );
    if (!["active", "staged"].includes(identity.status)) {
      throw new TypeError("device_edge_channel_certificate_identity_status_invalid");
    }
    if (identity.status === "active") activeCount += 1;
    const generationRef = normalizeRef(
      identity.generationRef,
      "certificate_generation_ref",
    );
    const fingerprint = normalizeCertificateFingerprint(identity.fingerprint);
    if (generations.has(generationRef) || fingerprints.has(fingerprint)) {
      throw new TypeError("device_edge_channel_certificate_identity_duplicate");
    }
    generations.add(generationRef);
    fingerprints.add(fingerprint);
    return Object.freeze({
      generationRef,
      fingerprint,
      status: identity.status,
    });
  });
  if (activeCount !== 1) {
    throw new TypeError("device_edge_channel_active_certificate_identity_invalid");
  }
  return Object.freeze(identities);
}

export function nextReconnectDelay(attempt, options = {}) {
  const normalizedAttempt = Number(attempt);
  if (!Number.isSafeInteger(normalizedAttempt) || normalizedAttempt < 0) {
    throw new TypeError("device_edge_channel_reconnect_attempt_invalid");
  }
  const minimumMs = normalizeReconnectDuration(
    options.minimumMs,
    DEVICE_EDGE_CHANNEL_LIMITS.reconnectMinimumMs,
  );
  const maximumMs = normalizeReconnectDuration(
    options.maximumMs,
    DEVICE_EDGE_CHANNEL_LIMITS.reconnectMaximumMs,
  );
  if (maximumMs < minimumMs) {
    throw new TypeError("device_edge_channel_reconnect_range_invalid");
  }
  const random = options.random ?? Math.random;
  if (typeof random !== "function") {
    throw new TypeError("device_edge_channel_random_invalid");
  }
  const ceiling = Math.min(maximumMs, minimumMs * (2 ** normalizedAttempt));
  const floor = Math.max(minimumMs, Math.floor(ceiling / 2));
  const sample = Number(random());
  if (!Number.isFinite(sample) || sample < 0 || sample > 1) {
    throw new TypeError("device_edge_channel_random_invalid");
  }
  return Math.floor(floor + ((ceiling - floor) * sample));
}

function normalizeReconnectDuration(value, fallback) {
  const number = value == null ? fallback : Number(value);
  if (!Number.isSafeInteger(number) || number < 10 || number > 120_000) {
    throw new TypeError("device_edge_channel_reconnect_duration_invalid");
  }
  return number;
}

function normalizeDirection(value) {
  if (!['edge-to-core', 'core-to-edge'].includes(value)) {
    throw new TypeError("device_edge_channel_direction_invalid");
  }
  return value;
}

function normalizeLimit(value, fallback) {
  const number = value == null ? fallback : Number(value);
  if (!Number.isSafeInteger(number) || number < 256 || number > 4 * 1024 * 1024) {
    throw new TypeError("device_edge_channel_limit_invalid");
  }
  return number;
}

function normalizeRef(value, field) {
  if (typeof value !== "string" || !OPAQUE_REF_RE.test(value)) {
    throw new TypeError(`device_edge_channel_${field}_invalid`);
  }
  return value;
}

function normalizeTimestamp(value, field) {
  if (typeof value !== "string" || !ISO_TIMESTAMP_RE.test(value)) {
    throw new TypeError(`device_edge_channel_${field}_invalid`);
  }
  const date = new Date(value);
  if (!Number.isFinite(date.getTime())) {
    throw new TypeError(`device_edge_channel_${field}_invalid`);
  }
  return date.toISOString();
}

function assertPlainObject(value, name) {
  if (!value || typeof value !== "object" || Array.isArray(value)) {
    throw new TypeError(`${name}_invalid`);
  }
  const prototype = Object.getPrototypeOf(value);
  if (prototype !== Object.prototype && prototype !== null) {
    throw new TypeError(`${name}_invalid`);
  }
}

function rejectUnexpectedKeys(value, allowedKeys) {
  for (const key of Object.keys(value)) {
    if (!allowedKeys.has(key)) {
      throw new TypeError(`device_edge_channel_field_unexpected:${key}`);
    }
  }
}

function rejectForbiddenKeys(value, depth = 0) {
  if (depth > 16 || value == null || typeof value !== "object") return;
  for (const [key, child] of Object.entries(value)) {
    const compact = key.toLowerCase().replaceAll(/[^a-z0-9]/g, "");
    if (forbiddenKeyFragments.some((fragment) => compact.includes(fragment))) {
      throw new TypeError(`device_edge_channel_forbidden_field:${key}`);
    }
    rejectForbiddenKeys(child, depth + 1);
  }
}

function assertJsonValue(value, depth) {
  if (depth > 16) {
    throw new TypeError("device_edge_channel_payload_depth_invalid");
  }
  if (value == null || typeof value === "string" || typeof value === "boolean") {
    return;
  }
  if (typeof value === "number" && Number.isFinite(value)) return;
  if (Array.isArray(value)) {
    for (const child of value) assertJsonValue(child, depth + 1);
    return;
  }
  assertPlainObject(value, "device_edge_channel_payload");
  for (const child of Object.values(value)) {
    assertJsonValue(child, depth + 1);
  }
}

function cloneJsonValue(value) {
  return JSON.parse(JSON.stringify(value));
}

function deepFreeze(value) {
  if (!value || typeof value !== "object" || Object.isFrozen(value)) {
    return value;
  }
  Object.freeze(value);
  for (const child of Object.values(value)) deepFreeze(child);
  return value;
}