import express from "express";
import { createHash, randomUUID, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";
import { Pool } from "pg";
import { validateDataProductPublish, validateIntakeBatch } from "@nodedc/external-provider-contract/data-plane";
import { readConfig } from "./config.mjs";
import {
  loadDataProductDefinition,
  assertPublishMatchesDefinition,
  persistDataProductDefinition,
  persistDataProductPublish,
  pruneBatchReceipts,
  pruneDataProductHistory,
  prunePatchOutbox,
  readDataProductHistory,
  readDataProductSnapshot,
  readPatchEvents,
} from "./data-product-delivery.mjs";
import { normalizeDataProductDefinition, safeDataProductDefinition } from "./data-product-policy.mjs";
import { reconcileDataProductDefinitions } from "./definitions.mjs";
import { assertBatchTimeBounds, rawRetentionExpiry } from "./intake-policy.mjs";
import {
  createManagedProvisionerRequestVerifier,
  ManagedProvisionerReplayCache,
} from "./managed-provisioner-auth.mjs";
import {
  assertReaderProduct,
  consumerReaderBindingRequestHash,
  createReaderToken,
  hashReaderToken,
  normalizeManagedConsumerReaderBindingRequest,
  normalizeManagedConsumerReaderPlanRequest,
  normalizeManagedReaderBindingRequest,
  normalizeReaderBindingRequest,
  readerBindingRequestHash,
  safeManagedConsumerReaderBinding,
  safeReaderBinding,
} from "./reader-binding.mjs";
import { resolveManagedReaderScope, resolveManagedReaderSourceConnection } from "./reader-source-scope.mjs";
import { migrate } from "./schema.mjs";
import {
  createWriterToken,
  canMigrateManagedWriterBindingToDurable,
  hashWriterToken,
  materializeDataProductPublish,
  materializeWriterBoundBatch,
  normalizeManagedWriterBindingRequest,
  normalizeWriterBindingRequest,
  safeWriterBinding,
  writerBindingRequestHash,
} from "./writer-binding.mjs";

const config = readConfig();
const managedProvisionerReplayCache = config.managedProvisionerApiEnabled
  ? new ManagedProvisionerReplayCache({ maxEntries: config.managedProvisionerReplayCacheMaxEntries })
  : null;
const verifyManagedProvisionerRequest = config.managedProvisionerApiEnabled
  ? createManagedProvisionerRequestVerifier({
    publicKey: config.managedProvisionerPublicKey,
    serviceId: config.managedProvisionerServiceId,
    keyId: config.managedProvisionerKeyId,
    audience: config.managedProvisionerAudience,
    maxSkewMs: config.managedProvisionerMaxSkewMs,
    replayCache: managedProvisionerReplayCache,
  })
  : null;
const foundryProvisionerReplayCache = config.foundryProvisionerApiEnabled
  ? new ManagedProvisionerReplayCache({ maxEntries: config.foundryProvisionerReplayCacheMaxEntries })
  : null;
const verifyFoundryProvisionerRequest = config.foundryProvisionerApiEnabled
  ? createManagedProvisionerRequestVerifier({
    publicKey: config.foundryProvisionerPublicKey,
    serviceId: config.foundryProvisionerServiceId,
    keyId: config.foundryProvisionerKeyId,
    audience: config.foundryProvisionerAudience,
    maxSkewMs: config.foundryProvisionerMaxSkewMs,
    replayCache: foundryProvisionerReplayCache,
  })
  : null;
const pool = new Pool({ connectionString: config.databaseUrl, max: config.databasePoolSize });
const app = express();
const httpServer = createServer(app);
let retentionSweepTimer = null;
let lastRetentionSweepAt = null;
const activeReaderStreams = new Map();
const activeStreamResponses = new Set();
let shuttingDown = false;
let shutdownPromise = null;

app.disable("x-powered-by");
app.use((req, _res, next) => {
  req.rawBody = Buffer.alloc(0);
  req.rawBodyCaptured = false;
  next();
});
app.use(express.json({
  limit: config.maxBatchBytes,
  inflate: false,
  verify(req, _res, buffer) {
    req.rawBody = Buffer.from(buffer);
    req.rawBodyCaptured = true;
  },
}));

app.get("/healthz", asyncRoute(async (_req, res) => {
  await pool.query("select 1");
  res.json({
    ok: true,
    service: "nodedc-external-data-plane",
    database: "ready",
    internalApiConfigured: Boolean(config.internalAccessToken),
    providerLogic: "absent",
    commandTransport: "absent",
    writerBindings: "supported",
    managedWriterBindings: "digest+idempotent-generation+explicit-revoke",
    readerBindings: "supported",
    managedReaderBindings: "digest+idempotent-generation+explicit-revoke",
    readerSourceScope: "writer-resolved+fail-closed-ambiguity",
    dataProductDelivery: "snapshot+history+durable-patch",
    legacyIntake: config.legacyIntakeEnabled ? "migration-only" : "disabled",
    legacyCapabilityProvisioning: config.provisionerApiEnabled ? "enabled" : "disabled",
    managedWriterBindingProvisioning: config.managedProvisionerApiEnabled ? "enabled" : "disabled",
    managedWriterBindingLifetime: config.managedProvisionerApiEnabled ? "explicit-revoke" : "disabled",
    managedReaderBindingProvisioning: config.managedProvisionerApiEnabled ? "enabled" : "disabled",
    managedReaderBindingLifetime: config.managedProvisionerApiEnabled ? "explicit-revoke" : "disabled",
    foundryReaderBindingProvisioning: config.foundryProvisionerApiEnabled ? "digest+server-resolved-source" : "disabled",
    foundryReaderBindingLifetime: config.foundryProvisionerApiEnabled ? "explicit-revoke" : "disabled",
    rawRetentionSweep: {
      mode: "server-scheduled",
      lastSweepAt: lastRetentionSweepAt,
    },
  });
}));

app.put("/internal/data-plane/v1/data-products/:dataProductId", requireProvisionerApi, asyncRoute(async (req, res) => {
  const definition = normalizeDataProductDefinition({ ...req.body, id: req.params.dataProductId });
  const saved = await persistDataProductDefinition(pool, definition);
  res.json({ ok: true, dataProduct: safeDataProductDefinition(saved) });
}));

app.post("/internal/data-plane/v1/data-products/:dataProductId/publish", requireWriterBinding, asyncRoute(async (req, res) => {
  if (hasScopeHeaders(req)) throw httpError(400, "data_product_publish_scope_headers_forbidden");
  const validation = validateDataProductPublish(req.body, {
    maxFacts: config.maxFactsPerPublish,
    maxAttributesBytes: config.maxAttributesBytesPerFact,
  });
  if (!validation.ok) throw httpError(422, validation.errors[0] || "invalid_data_product_publish");
  const dataProductId = requireIdentifier(req.params.dataProductId, "data_product_id_invalid");
  const definition = await loadDataProductDefinition(pool, dataProductId);
  if (!definition) throw httpError(404, "data_product_not_found");
  const batch = materializeDataProductPublish(req.body, req.writerBinding, definition, dataProductId);
  const canonicalValidation = validateIntakeBatch(batch);
  if (!canonicalValidation.ok) throw httpError(422, "materialized_publish_invalid");
  assertBatchTimeBounds(batch, { maxFutureSkewSeconds: config.maxFutureSkewSeconds });
  const receipt = await persistDataProductPublish(pool, batch, definition, {
    maxPatchOperations: config.maxPatchOperations,
    maxPatchBytes: config.maxPatchBytes,
  });
  res.status(receipt.idempotent ? 200 : 201).json({ ok: true, ...receipt });
}));

app.get("/internal/data-plane/v1/writer/data-products", requireWriterBinding, asyncRoute(async (req, res) => {
  const dataProducts = await listGrantedProducts(req.writerBinding);
  res.json({ ok: true, dataProducts });
}));

app.get("/internal/data-plane/v1/reader/data-products", requireReaderBinding, asyncRoute(async (req, res) => {
  const dataProducts = await listGrantedProducts(req.readerBinding);
  res.json({ ok: true, dataProducts });
}));

app.post("/internal/data-plane/v1/intake", requireLegacyIntake, requireInternalApi, asyncRoute(async (req, res) => {
  const validation = validateIntakeBatch(req.body);
  if (!validation.ok) throw httpError(422, "invalid_intake_batch");
  assertBatchTimeBounds(req.body, { maxFutureSkewSeconds: config.maxFutureSkewSeconds });
  await assertLegacyBatchProduct(req.body);
  const scope = requireScope(req);
  if (scope.tenantId !== req.body.source.tenantId || scope.connectionId !== req.body.source.connectionId) {
    throw httpError(403, "scope_mismatch");
  }

  const result = await persistBatch(req.body);
  res.status(result.idempotent ? 200 : 201).json({ ok: true, ...result });
}));

app.post("/internal/data-plane/v1/intake/writer-bound", requireLegacyIntake, requireWriterBinding, asyncRoute(async (req, res) => {
  const batch = materializeWriterBoundBatch(req.body, req.writerBinding, {
    hasScopeHeaders: hasScopeHeaders(req),
  });
  const validation = validateIntakeBatch(batch);
  if (!validation.ok) throw httpError(422, "invalid_intake_batch");
  assertBatchTimeBounds(batch, { maxFutureSkewSeconds: config.maxFutureSkewSeconds });
  await assertLegacyBatchProduct(batch);

  const result = await persistBatch(batch);
  res.status(result.idempotent ? 200 : 201).json({ ok: true, ...result });
}));

app.put("/internal/data-plane/v1/writer-bindings/by-key/:bindingKey", requireManagedProvisionerApi, asyncRoute(async (req, res) => {
  const bindingKey = requireIdentifier(req.params.bindingKey, "managed_writer_binding_key_invalid");
  const policy = normalizeManagedWriterBindingRequest(req.body);
  await assertRegisteredProductIds(policy.allowedDataProductIds);
  const requestHash = writerBindingRequestHash(policy);
  const client = await pool.connect();
  let response;
  try {
    await client.query("begin");
    await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
    const existing = await client.query(
      `select id, token_hash as "capabilityDigest", binding_key as "bindingKey", request_hash as "requestHash", generation,
         tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
         allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
         active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
       from external_data_plane_writer_bindings
       where binding_key = $1 and generation = $2
       for update`,
      [bindingKey, policy.generation],
    );
    if (existing.rowCount) {
      const binding = existing.rows[0];
      if (binding.requestHash !== requestHash) {
        const legacyCanMigrate = canMigrateManagedWriterBindingToDurable(binding, policy);
        if (!legacyCanMigrate) throw httpError(409, "managed_writer_binding_request_conflict");
        const migrated = await client.query(
          `update external_data_plane_writer_bindings
           set request_hash = $3, expires_at = null, rotated_at = now()
           where binding_key = $1 and generation = $2 and token_hash = $4
             and active = true and expires_at > now()
           returning id, binding_key as "bindingKey", generation,
             tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
             allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
             active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
          [bindingKey, policy.generation, requestHash, policy.capabilityDigest],
        );
        if (!migrated.rowCount) throw httpError(409, "managed_writer_binding_generation_inactive");
        response = { status: 200, idempotent: false, migrated: true, binding: migrated.rows[0] };
      } else {
        if (binding.active !== true || binding.expiresAt !== null) {
          throw httpError(409, "managed_writer_binding_generation_inactive");
        }
        response = { status: 200, idempotent: true, migrated: false, binding };
      }
    } else {
      const inserted = await client.query(
        `insert into external_data_plane_writer_bindings (
          id, token_hash, binding_key, request_hash, generation,
          tenant_id, connection_id, provider_id, allowed_data_product_ids, expires_at
        ) values ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10)
        returning id, binding_key as "bindingKey", generation,
          tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
          allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
          active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
        [
          randomUUID(),
          policy.capabilityDigest,
          bindingKey,
          requestHash,
          policy.generation,
          policy.tenantId,
          policy.connectionId,
          policy.providerId,
          JSON.stringify(policy.allowedDataProductIds),
          policy.expiresAt,
        ],
      );
      response = { status: 201, idempotent: false, binding: inserted.rows[0] };
    }
    await client.query("commit");
  } catch (error) {
    await client.query("rollback");
    if (error?.code === "23505") {
      throw httpError(409, "managed_writer_binding_capability_digest_conflict");
    }
    throw error;
  } finally {
    client.release();
  }
  res.set("Cache-Control", "no-store, max-age=0");
  res.status(response.status).json({
    ok: true,
    idempotent: response.idempotent,
    migrated: response.migrated === true,
    writerBinding: safeWriterBinding(response.binding),
  });
}));

app.post("/internal/data-plane/v1/writer-bindings/by-key/:bindingKey/generations/:generation/revoke", requireManagedProvisionerApi, asyncRoute(async (req, res) => {
  const bindingKey = requireIdentifier(req.params.bindingKey, "managed_writer_binding_key_invalid");
  const generation = requirePositiveInteger(req.params.generation, "managed_writer_binding_generation_invalid");
  const client = await pool.connect();
  let binding;
  let idempotent;
  try {
    await client.query("begin");
    await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
    const existing = await client.query(
      `select id, binding_key as "bindingKey", generation,
         tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
         allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
         active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
       from external_data_plane_writer_bindings
       where binding_key = $1 and generation = $2
       for update`,
      [bindingKey, generation],
    );
    if (!existing.rowCount) throw httpError(404, "managed_writer_binding_not_found");
    if (existing.rows[0].active === true) {
      const revoked = await client.query(
        `update external_data_plane_writer_bindings
         set active = false, revoked_at = now()
         where binding_key = $1 and generation = $2 and active = true
         returning id, binding_key as "bindingKey", generation,
           tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
           allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
           active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
        [bindingKey, generation],
      );
      binding = revoked.rows[0];
      idempotent = false;
    } else {
      binding = existing.rows[0];
      idempotent = true;
    }
    await client.query("commit");
  } catch (error) {
    await client.query("rollback");
    throw error;
  } finally {
    client.release();
  }
  res.set("Cache-Control", "no-store, max-age=0");
  res.json({ ok: true, idempotent, writerBinding: safeWriterBinding(binding) });
}));

app.post("/internal/data-plane/v1/writer-bindings", requireProvisionerApi, asyncRoute(async (req, res) => {
  const policy = normalizeWriterBindingRequest(req.body, {
    maxTtlDays: config.writerBindingMaxTtlDays,
  });
  await assertRegisteredProductIds(policy.allowedDataProductIds);
  const token = createWriterToken();
  const bindingId = randomUUID();
  const result = await pool.query(
    `insert into external_data_plane_writer_bindings (
      id, token_hash, tenant_id, connection_id, provider_id,
      allowed_data_product_ids, expires_at
    ) values ($1, $2, $3, $4, $5, $6::jsonb, $7)
    returning id, tenant_id as "tenantId", connection_id as "connectionId",
      provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
      expires_at as "expiresAt", active, created_at as "createdAt",
      rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
    [
      bindingId,
      hashWriterToken(token),
      policy.tenantId,
      policy.connectionId,
      policy.providerId,
      JSON.stringify(policy.allowedDataProductIds),
      policy.expiresAt,
    ],
  );
  // The token is intentionally returned exactly once to a trusted provisioner.
  // It must be placed directly into an opaque Engine credential reference and
  // must never be logged, stored in L2 graph data or shown to a consumer.
  sendOneTimeCapability(res, 201, { ok: true, writerBinding: safeWriterBinding(result.rows[0]), token });
}));

app.post("/internal/data-plane/v1/writer-bindings/:bindingId/rotate", requireProvisionerApi, asyncRoute(async (req, res) => {
  const bindingId = requireUuid(req.params.bindingId, "writer_binding_id_invalid");
  const token = createWriterToken();
  const result = await pool.query(
    `update external_data_plane_writer_bindings
     set token_hash = $2, rotated_at = now()
     where id = $1 and binding_key is null and active = true and expires_at > now()
     returning id, tenant_id as "tenantId", connection_id as "connectionId",
       provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
       expires_at as "expiresAt", active, created_at as "createdAt",
       rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
    [bindingId, hashWriterToken(token)],
  );
  if (!result.rowCount) throw httpError(404, "writer_binding_not_found_or_inactive");
  sendOneTimeCapability(res, 200, { ok: true, writerBinding: safeWriterBinding(result.rows[0]), token });
}));

app.post("/internal/data-plane/v1/writer-bindings/:bindingId/revoke", requireProvisionerApi, asyncRoute(async (req, res) => {
  const bindingId = requireUuid(req.params.bindingId, "writer_binding_id_invalid");
  const result = await pool.query(
    `update external_data_plane_writer_bindings
     set active = false, revoked_at = now()
     where id = $1 and binding_key is null and active = true
     returning id, tenant_id as "tenantId", connection_id as "connectionId",
       provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
       expires_at as "expiresAt", active, created_at as "createdAt",
       rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
    [bindingId],
  );
  if (!result.rowCount) throw httpError(404, "writer_binding_not_found_or_inactive");
  res.json({ ok: true, writerBinding: safeWriterBinding(result.rows[0]) });
}));

app.post("/internal/data-plane/v1/consumer-reader-bindings/plan", requireFoundryProvisionerApi, asyncRoute(async (req, res) => {
  const policy = normalizeManagedConsumerReaderPlanRequest(req.body);
  await assertRegisteredProductIds(policy.allowedDataProductIds);
  await resolveManagedReaderScope(pool, policy.allowedDataProductIds);
  const dataProducts = [];
  for (const dataProductId of policy.allowedDataProductIds) {
    const definition = await loadDataProductDefinition(pool, dataProductId);
    if (!definition || definition.active === false) throw httpError(404, "data_product_not_found");
    dataProducts.push(safeDataProductDefinition(definition));
  }
  res.set("Cache-Control", "no-store, max-age=0");
  res.json({ ok: true, sourceScope: "resolved-server-side", dataProducts });
}));

app.put("/internal/data-plane/v1/consumer-reader-bindings/by-key/:bindingKey", requireFoundryProvisionerApi, asyncRoute(async (req, res) => {
  const bindingKey = requireIdentifier(req.params.bindingKey, "managed_consumer_reader_binding_key_invalid");
  const policy = normalizeManagedConsumerReaderBindingRequest(req.body);
  await assertRegisteredProductIds(policy.allowedDataProductIds);
  const requestHash = consumerReaderBindingRequestHash(policy);
  const client = await pool.connect();
  let response;
  try {
    await client.query("begin");
    await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
    const existing = await client.query(
      `select id, binding_key as "bindingKey", request_hash as "requestHash", generation,
         tenant_id as "tenantId", connection_id as "connectionId",
         source_connection_id as "sourceConnectionId", provider_id as "providerId",
         allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
         active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
       from external_data_plane_reader_bindings
       where binding_key = $1 and generation = $2
       for update`,
      [bindingKey, policy.generation],
    );
    if (existing.rowCount) {
      const binding = existing.rows[0];
      if (binding.requestHash !== requestHash) throw httpError(409, "managed_consumer_reader_binding_request_conflict");
      if (binding.active !== true || binding.expiresAt !== null) {
        throw httpError(409, "managed_consumer_reader_binding_generation_inactive");
      }
      response = { status: 200, idempotent: true, binding };
    } else {
      const source = await resolveManagedReaderScope(client, policy.allowedDataProductIds);
      const inserted = await client.query(
        `insert into external_data_plane_reader_bindings (
          id, token_hash, binding_key, request_hash, generation,
          tenant_id, connection_id, source_connection_id, provider_id,
          allowed_data_product_ids, expires_at
        ) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11)
        returning id, binding_key as "bindingKey", generation,
          tenant_id as "tenantId", connection_id as "connectionId",
          source_connection_id as "sourceConnectionId", provider_id as "providerId",
          allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
          active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
        [
          randomUUID(),
          policy.capabilityDigest,
          bindingKey,
          requestHash,
          policy.generation,
          source.tenantId,
          source.connectionId,
          source.connectionId,
          source.providerId,
          JSON.stringify(policy.allowedDataProductIds),
          policy.expiresAt,
        ],
      );
      response = { status: 201, idempotent: false, binding: inserted.rows[0] };
    }
    await client.query("commit");
  } catch (error) {
    await client.query("rollback");
    if (error?.code === "23505") throw httpError(409, "managed_consumer_reader_capability_digest_conflict");
    throw error;
  } finally {
    client.release();
  }
  res.set("Cache-Control", "no-store, max-age=0");
  res.status(response.status).json({
    ok: true,
    idempotent: response.idempotent,
    readerBinding: safeManagedConsumerReaderBinding(response.binding),
  });
}));

app.post("/internal/data-plane/v1/consumer-reader-bindings/by-key/:bindingKey/generations/:generation/revoke", requireFoundryProvisionerApi, asyncRoute(async (req, res) => {
  const bindingKey = requireIdentifier(req.params.bindingKey, "managed_consumer_reader_binding_key_invalid");
  const generation = requirePositiveInteger(req.params.generation, "managed_consumer_reader_generation_invalid");
  const client = await pool.connect();
  let binding;
  let idempotent;
  try {
    await client.query("begin");
    await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
    const existing = await client.query(
      `select id, binding_key as "bindingKey", generation,
         tenant_id as "tenantId", connection_id as "connectionId",
         source_connection_id as "sourceConnectionId", provider_id as "providerId",
         allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
         active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
       from external_data_plane_reader_bindings
       where binding_key = $1 and generation = $2
       for update`,
      [bindingKey, generation],
    );
    if (!existing.rowCount) throw httpError(404, "managed_consumer_reader_binding_not_found");
    if (existing.rows[0].active === true) {
      const revoked = await client.query(
        `update external_data_plane_reader_bindings
         set active = false, revoked_at = now()
         where binding_key = $1 and generation = $2 and active = true
         returning id, binding_key as "bindingKey", generation,
           tenant_id as "tenantId", connection_id as "connectionId",
           source_connection_id as "sourceConnectionId", provider_id as "providerId",
           allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
           active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
        [bindingKey, generation],
      );
      binding = revoked.rows[0];
      idempotent = false;
    } else {
      binding = existing.rows[0];
      idempotent = true;
    }
    await client.query("commit");
  } catch (error) {
    await client.query("rollback");
    throw error;
  } finally {
    client.release();
  }
  res.set("Cache-Control", "no-store, max-age=0");
  res.json({ ok: true, idempotent, readerBinding: safeManagedConsumerReaderBinding(binding) });
}));

app.put("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey", requireManagedProvisionerApi, asyncRoute(async (req, res) => {
  const bindingKey = requireIdentifier(req.params.bindingKey, "managed_reader_binding_key_invalid");
  const policy = normalizeManagedReaderBindingRequest(req.body);
  await assertRegisteredProductIds(policy.allowedDataProductIds);
  const requestHash = readerBindingRequestHash(policy);
  const client = await pool.connect();
  let response;
  try {
    await client.query("begin");
    await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
    const existing = await client.query(
      `select id, binding_key as "bindingKey", request_hash as "requestHash", generation,
         tenant_id as "tenantId", connection_id as "connectionId",
         source_connection_id as "sourceConnectionId", provider_id as "providerId",
         allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
         active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
       from external_data_plane_reader_bindings
       where binding_key = $1 and generation = $2
       for update`,
      [bindingKey, policy.generation],
    );
    if (existing.rowCount) {
      let binding = existing.rows[0];
      if (binding.requestHash !== requestHash) {
        throw httpError(409, "managed_reader_binding_request_conflict");
      }
      if (binding.active !== true || binding.expiresAt !== null) {
        throw httpError(409, "managed_reader_binding_generation_inactive");
      }
      if (!binding.sourceConnectionId) {
        const sourceConnectionId = await resolveManagedReaderSourceConnection(client, policy);
        const resolved = await client.query(
          `update external_data_plane_reader_bindings
           set source_connection_id = $3
           where binding_key = $1 and generation = $2 and source_connection_id is null
           returning id, binding_key as "bindingKey", request_hash as "requestHash", generation,
             tenant_id as "tenantId", connection_id as "connectionId",
             source_connection_id as "sourceConnectionId", provider_id as "providerId",
             allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
             active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
          [bindingKey, policy.generation, sourceConnectionId],
        );
        binding = resolved.rows[0] || binding;
      }
      response = { status: 200, idempotent: true, binding };
    } else {
      const sourceConnectionId = await resolveManagedReaderSourceConnection(client, policy);
      const inserted = await client.query(
        `insert into external_data_plane_reader_bindings (
          id, token_hash, binding_key, request_hash, generation,
          tenant_id, connection_id, source_connection_id, provider_id,
          allowed_data_product_ids, expires_at
        ) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11)
        returning id, binding_key as "bindingKey", generation,
          tenant_id as "tenantId", connection_id as "connectionId",
          source_connection_id as "sourceConnectionId", provider_id as "providerId",
          allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
          active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
        [
          randomUUID(),
          policy.capabilityDigest,
          bindingKey,
          requestHash,
          policy.generation,
          policy.tenantId,
          policy.connectionId,
          sourceConnectionId,
          policy.providerId,
          JSON.stringify(policy.allowedDataProductIds),
          policy.expiresAt,
        ],
      );
      response = { status: 201, idempotent: false, binding: inserted.rows[0] };
    }
    await client.query("commit");
  } catch (error) {
    await client.query("rollback");
    if (error?.code === "23505") {
      throw httpError(409, "managed_reader_binding_capability_digest_conflict");
    }
    throw error;
  } finally {
    client.release();
  }
  res.set("Cache-Control", "no-store, max-age=0");
  res.status(response.status).json({
    ok: true,
    idempotent: response.idempotent,
    readerBinding: safeReaderBinding(response.binding),
  });
}));

app.post("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey/generations/:generation/revoke", requireManagedProvisionerApi, asyncRoute(async (req, res) => {
  const bindingKey = requireIdentifier(req.params.bindingKey, "managed_reader_binding_key_invalid");
  const generation = requirePositiveInteger(req.params.generation, "managed_reader_binding_generation_invalid");
  const client = await pool.connect();
  let binding;
  let idempotent;
  try {
    await client.query("begin");
    await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
    const existing = await client.query(
      `select id, binding_key as "bindingKey", generation,
         tenant_id as "tenantId", connection_id as "connectionId",
         source_connection_id as "sourceConnectionId", provider_id as "providerId",
         allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
         active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
       from external_data_plane_reader_bindings
       where binding_key = $1 and generation = $2
       for update`,
      [bindingKey, generation],
    );
    if (!existing.rowCount) throw httpError(404, "managed_reader_binding_not_found");
    if (existing.rows[0].active === true) {
      const revoked = await client.query(
        `update external_data_plane_reader_bindings
         set active = false, revoked_at = now()
         where binding_key = $1 and generation = $2 and active = true
         returning id, binding_key as "bindingKey", generation,
           tenant_id as "tenantId", connection_id as "connectionId",
           source_connection_id as "sourceConnectionId", provider_id as "providerId",
           allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
           active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
        [bindingKey, generation],
      );
      binding = revoked.rows[0];
      idempotent = false;
    } else {
      binding = existing.rows[0];
      idempotent = true;
    }
    await client.query("commit");
  } catch (error) {
    await client.query("rollback");
    throw error;
  } finally {
    client.release();
  }
  res.set("Cache-Control", "no-store, max-age=0");
  res.json({ ok: true, idempotent, readerBinding: safeReaderBinding(binding) });
}));

app.post("/internal/data-plane/v1/reader-bindings", requireProvisionerApi, asyncRoute(async (req, res) => {
  const policy = normalizeReaderBindingRequest(req.body, {
    maxTtlDays: config.writerBindingMaxTtlDays,
  });
  await assertRegisteredProductIds(policy.allowedDataProductIds);
  const token = createReaderToken();
  const bindingId = randomUUID();
  const result = await pool.query(
    `insert into external_data_plane_reader_bindings (
      id, token_hash, tenant_id, connection_id, source_connection_id, provider_id,
      allowed_data_product_ids, expires_at
    ) values ($1, $2, $3, $4, $4, $5, $6::jsonb, $7)
    returning id, tenant_id as "tenantId", connection_id as "connectionId",
      source_connection_id as "sourceConnectionId",
      provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
      expires_at as "expiresAt", active, created_at as "createdAt",
      rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
    [
      bindingId,
      hashReaderToken(token),
      policy.tenantId,
      policy.connectionId,
      policy.providerId,
      JSON.stringify(policy.allowedDataProductIds),
      policy.expiresAt,
    ],
  );
  sendOneTimeCapability(res, 201, { ok: true, readerBinding: safeReaderBinding(result.rows[0]), token });
}));

app.post("/internal/data-plane/v1/reader-bindings/:bindingId/rotate", requireProvisionerApi, asyncRoute(async (req, res) => {
  const bindingId = requireUuid(req.params.bindingId, "reader_binding_id_invalid");
  const token = createReaderToken();
  const result = await pool.query(
    `update external_data_plane_reader_bindings
     set token_hash = $2, rotated_at = now()
     where id = $1 and binding_key is null and active = true and expires_at > now()
     returning id, tenant_id as "tenantId", connection_id as "connectionId",
       source_connection_id as "sourceConnectionId",
       provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
       expires_at as "expiresAt", active, created_at as "createdAt",
       rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
    [bindingId, hashReaderToken(token)],
  );
  if (!result.rowCount) throw httpError(404, "reader_binding_not_found_or_inactive");
  sendOneTimeCapability(res, 200, { ok: true, readerBinding: safeReaderBinding(result.rows[0]), token });
}));

app.post("/internal/data-plane/v1/reader-bindings/:bindingId/revoke", requireProvisionerApi, asyncRoute(async (req, res) => {
  const bindingId = requireUuid(req.params.bindingId, "reader_binding_id_invalid");
  const result = await pool.query(
    `update external_data_plane_reader_bindings
     set active = false, revoked_at = now()
     where id = $1 and binding_key is null and active = true
     returning id, tenant_id as "tenantId", connection_id as "connectionId",
       source_connection_id as "sourceConnectionId",
       provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
       expires_at as "expiresAt", active, created_at as "createdAt",
       rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
    [bindingId],
  );
  if (!result.rowCount) throw httpError(404, "reader_binding_not_found_or_inactive");
  res.json({ ok: true, readerBinding: safeReaderBinding(result.rows[0]) });
}));

app.get("/internal/data-plane/v1/data-products/:dataProductId/snapshot", requireReaderBinding, asyncRoute(async (req, res) => {
  if (hasScopeHeaders(req)) throw httpError(400, "data_product_read_scope_headers_forbidden");
  const dataProductId = assertReaderProduct(req.readerBinding, req.params.dataProductId);
  const definition = await loadDataProductDefinition(pool, dataProductId);
  if (!definition) throw httpError(404, "data_product_not_found");
  const limit = boundedLimit(req.query.limit, 5000, 1, 5000);
  const snapshot = await readDataProductSnapshot(pool, req.readerBinding, definition, { limit });
  res.json(snapshot);
}));

app.get("/internal/data-plane/v1/data-products/:dataProductId/history", requireReaderBinding, asyncRoute(async (req, res) => {
  if (hasScopeHeaders(req)) throw httpError(400, "data_product_read_scope_headers_forbidden");
  const dataProductId = assertReaderProduct(req.readerBinding, req.params.dataProductId);
  const definition = await loadDataProductDefinition(pool, dataProductId);
  if (!definition) throw httpError(404, "data_product_not_found");
  const history = await readDataProductHistory(pool, req.readerBinding, definition, {
    from: req.query.from,
    to: req.query.to,
    resolutionMs: req.query.resolutionMs,
    sourceIds: parseSourceIds(req.query.sourceIds),
    limit: req.query.limit,
    cursor: req.query.cursor,
  });
  res.json(history);
}));

app.get("/internal/data-plane/v1/data-products/:dataProductId/stream", requireReaderBinding, asyncRoute(async (req, res) => {
  if (shuttingDown) throw httpError(503, "service_shutting_down");
  if (hasScopeHeaders(req)) throw httpError(400, "data_product_read_scope_headers_forbidden");
  const dataProductId = assertReaderProduct(req.readerBinding, req.params.dataProductId);
  const definition = await loadDataProductDefinition(pool, dataProductId);
  if (!definition) throw httpError(404, "data_product_not_found");
  if (definition.deliveryMode !== "snapshot+patch") throw httpError(409, "data_product_stream_not_supported");
  const streamCount = activeReaderStreams.get(req.readerBinding.id) || 0;
  if (streamCount >= config.maxReaderStreams) throw httpError(429, "reader_stream_limit_exceeded");
  activeReaderStreams.set(req.readerBinding.id, streamCount + 1);
  activeStreamResponses.add(res);

  let poll = null;
  let heartbeat = null;
  let polling = false;
  let heartbeatWriting = false;
  let cleanedUp = false;
  const cleanup = () => {
    if (cleanedUp) return;
    cleanedUp = true;
    if (poll) clearInterval(poll);
    if (heartbeat) clearInterval(heartbeat);
    activeStreamResponses.delete(res);
    const remaining = Math.max(0, (activeReaderStreams.get(req.readerBinding.id) || 1) - 1);
    if (remaining) activeReaderStreams.set(req.readerBinding.id, remaining);
    else activeReaderStreams.delete(req.readerBinding.id);
  };
  req.once("close", cleanup);
  res.once("close", cleanup);

  try {
    let cursor = parseCursor(req.get("last-event-id") || req.query.after || "0");
    const initialEvents = await readPatchEvents(pool, req.readerBinding, definition, cursor);
    if (res.destroyed || cleanedUp) return;
    res.status(200);
    res.set({
      "Cache-Control": "no-cache, no-transform",
      "Content-Type": "text/event-stream",
      Connection: "keep-alive",
      "X-Accel-Buffering": "no",
    });
    res.flushHeaders();
    await writeSseFrame(res, ": nodedc-data-product-stream\n\n");
    await writeSseFrame(res, `event: nodedc.data-product.ready.v1\ndata: ${JSON.stringify({
      schemaVersion: "nodedc.data-product.ready/v1",
      dataProductId,
      cursor: cursor.toString(),
      emittedAt: new Date().toISOString(),
    })}\n\n`);
    for (const event of initialEvents) {
      await writePatchEvent(res, event);
      cursor = BigInt(event.cursor);
    }

    const pump = async () => {
      if (polling || res.writableEnded || res.destroyed || shuttingDown) return;
      polling = true;
      try {
        await assertReaderStreamAccess(req.readerBinding, dataProductId);
        const events = await readPatchEvents(pool, req.readerBinding, definition, cursor);
        for (const event of events) {
          await writePatchEvent(res, event);
          cursor = BigInt(event.cursor);
        }
      } catch (error) {
        const code = safeErrorCode(error);
        if (!res.writableEnded && !res.destroyed) {
          await writeSseFrame(res, `event: error\ndata: ${JSON.stringify({ ok: false, error: code })}\n\n`).catch(() => {});
          res.end();
        }
      } finally {
        polling = false;
      }
    };
    poll = setInterval(() => { void pump(); }, config.streamPollMs);
    poll.unref();
    heartbeat = setInterval(() => {
      if (polling || heartbeatWriting || res.writableEnded || res.destroyed) return;
      heartbeatWriting = true;
      void writeSseFrame(res, `: heartbeat ${Date.now()}\n\n`)
        .catch(() => { if (!res.writableEnded) res.end(); })
        .finally(() => { heartbeatWriting = false; });
    }, config.streamHeartbeatMs);
    heartbeat.unref();
  } catch (error) {
    cleanup();
    if (res.headersSent) {
      if (!res.writableEnded) res.end();
      return;
    }
    throw error;
  }
}));

app.get("/internal/data-plane/v1/data-products/:dataProductId/current", requireLegacyIntake, requireInternalApi, asyncRoute(async (req, res) => {
  const scope = requireScope(req);
  const dataProductId = String(req.params.dataProductId || "");
  if (!isIdentifier(dataProductId)) throw httpError(400, "data_product_id_invalid");
  const limit = boundedLimit(req.query.limit, 200, 1, 1000);
  const rows = await pool.query(
    `select
       provider_id as "providerId", data_product_id as "dataProductId",
       source_id as "sourceId", semantic_type as "semanticType",
       observed_at as "observedAt", received_at as "receivedAt", attributes,
       case when geometry is null then null
            else jsonb_build_object('type', 'Point', 'coordinates', jsonb_build_array(
              ST_X(geometry::geometry), ST_Y(geometry::geometry)
            )) end as geometry
     from external_data_plane_current
     where tenant_id = $1 and connection_id = $2 and data_product_id = $3
     order by observed_at desc, source_id asc
     limit $4`,
    [scope.tenantId, scope.connectionId, dataProductId, limit],
  );
  res.json({
    ok: true,
    dataProductId,
    tenantId: scope.tenantId,
    connectionId: scope.connectionId,
    facts: rows.rows,
  });
}));

app.get("/internal/data-plane/v1/status", requireLegacyIntake, requireInternalApi, asyncRoute(async (req, res) => {
  const scope = requireScope(req);
  const result = await pool.query(
    `select count(*)::integer as "currentFactCount", max(received_at) as "lastReceivedAt"
     from external_data_plane_current where tenant_id = $1 and connection_id = $2`,
    [scope.tenantId, scope.connectionId],
  );
  res.json({ ok: true, ...scope, ...result.rows[0], providerLogic: "absent", commandTransport: "absent" });
}));

app.use((error, _req, res, _next) => {
  const status = Number(error?.status || 500);
  const publicStatus = status >= 400 && status < 600 ? status : 500;
  console.error(JSON.stringify({ event: "external_data_plane_error", error: safeErrorCode(error), status: publicStatus }));
  res.status(publicStatus).json({ ok: false, error: publicStatus >= 500 ? "internal_error" : safeErrorCode(error) });
});

await migrate(pool);
const bundledDataProducts = await reconcileDataProductDefinitions(pool);
retentionSweepTimer = setInterval(() => {
  void sweepRetention().catch((error) => {
    console.error(JSON.stringify({ event: "external_data_plane_retention_sweep_failed", error: safeErrorCode(error) }));
  });
}, config.retentionSweepMs);
retentionSweepTimer.unref();

httpServer.listen(config.port, "0.0.0.0", () => {
  console.log(`NODE.DC External Data Plane listening on http://0.0.0.0:${config.port}`);
  console.log(JSON.stringify({ event: "external_data_plane_definitions_ready", count: bundledDataProducts.length }));
  setImmediate(() => {
    void sweepRetention().catch((error) => {
      console.error(JSON.stringify({ event: "external_data_plane_retention_sweep_failed", error: safeErrorCode(error) }));
    });
  });
});

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

async function persistBatch(batch) {
  const client = await pool.connect();
  try {
    await client.query("begin");
    const existing = await client.query(
      `select id, fact_count as "factCount", inserted_fact_count as "insertedFactCount"
       from external_data_plane_batches
       where tenant_id = $1 and connection_id = $2 and provider_id = $3
         and data_product_id = $4 and idempotency_key = $5
       for update`,
      [
        batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
        batch.contract.dataProductId, batch.batch.idempotencyKey,
      ],
    );
    if (existing.rowCount) {
      await client.query("commit");
      return { batchId: existing.rows[0].id, idempotent: true, ...existing.rows[0] };
    }

    const batchId = randomUUID();
    await client.query(
      `insert into external_data_plane_batches (
        id, tenant_id, connection_id, provider_id, data_product_id, contract_version,
        ontology_revision, run_id, sequence, idempotency_key, received_at, fact_count
      ) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`,
      [
        batchId, batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
        batch.contract.dataProductId, batch.contract.version, batch.contract.ontologyRevision,
        batch.batch.runId, batch.batch.sequence, batch.batch.idempotencyKey,
        batch.batch.receivedAt, batch.facts.length,
      ],
    );

    if (batch.raw) await persistRawEnvelope(client, batchId, batch);

    const records = batch.facts.map((fact) => ({
      id: randomUUID(),
      source_id: fact.sourceId,
      semantic_type: fact.semanticType,
      observed_at: fact.observedAt,
      attributes: fact.attributes || {},
      geometry: fact.geometry || null,
      fingerprint: hash(fact),
    }));
    const inserted = await client.query(
      `insert into external_data_plane_facts (
        id, batch_id, tenant_id, connection_id, provider_id, data_product_id,
        source_id, semantic_type, observed_at, received_at, attributes, geometry, fingerprint
      )
      select item.id::uuid, $1, $2, $3, $4, $5,
             item.source_id, item.semantic_type, item.observed_at, $6,
             coalesce(item.attributes, '{}'::jsonb),
             case when item.geometry is null then null
                  else ST_SetSRID(ST_MakePoint(
                    (item.geometry->'coordinates'->>0)::double precision,
                    (item.geometry->'coordinates'->>1)::double precision
                  ), 4326)::geography end,
             item.fingerprint
      from jsonb_to_recordset($7::jsonb) as item(
        id text, source_id text, semantic_type text, observed_at timestamptz,
        attributes jsonb, geometry jsonb, fingerprint text
      )
      on conflict do nothing
      returning id`,
      [
        batchId, batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
        batch.contract.dataProductId, batch.batch.receivedAt, JSON.stringify(records),
      ],
    );
    await client.query(
      `insert into external_data_plane_current (
        tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type,
        observed_at, received_at, attributes, geometry, fingerprint, updated_at
      )
      select tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type,
             observed_at, received_at, attributes, geometry, fingerprint, now()
      from external_data_plane_facts where batch_id = $1
      on conflict (tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type)
      do update set
        observed_at = excluded.observed_at,
        received_at = excluded.received_at,
        attributes = excluded.attributes,
        geometry = excluded.geometry,
        fingerprint = excluded.fingerprint,
        updated_at = now()
      where excluded.observed_at >= external_data_plane_current.observed_at`,
      [batchId],
    );
    await client.query(
      "update external_data_plane_batches set inserted_fact_count = $2 where id = $1",
      [batchId, inserted.rowCount],
    );
    await client.query("commit");
    return { batchId, idempotent: false, factCount: batch.facts.length, insertedFactCount: inserted.rowCount };
  } catch (error) {
    await client.query("rollback");
    throw error;
  } finally {
    client.release();
  }
}

async function persistRawEnvelope(client, batchId, batch) {
  const serialized = batch.raw.payload === undefined ? null : JSON.stringify(batch.raw.payload);
  const expiresAt = rawRetentionExpiry({ rawRetentionDays: config.rawRetentionDays });
  await client.query(
    `insert into external_data_plane_raw_envelopes (
      id, batch_id, payload_hash, content_type, payload, payload_ref, payload_bytes, received_at, expires_at
    ) values ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9)`,
    [
      randomUUID(), batchId, batch.raw.hash || hash(batch.raw.payload), batch.raw.contentType, serialized, batch.raw.ref || null,
      serialized ? Buffer.byteLength(serialized) : null, batch.batch.receivedAt, expiresAt,
    ],
  );
}

function requireInternalApi(req, _res, next) {
  if (!config.internalAccessToken) return next(httpError(503, "internal_api_not_configured"));
  const value = bearerToken(req);
  if (!value || !safeEqual(value, config.internalAccessToken)) return next(httpError(401, "unauthorized"));
  return next();
}

function requireLegacyIntake(_req, _res, next) {
  if (!config.legacyIntakeEnabled) return next(httpError(410, "legacy_intake_disabled"));
  return next();
}

function requireProvisionerApi(req, _res, next) {
  if (!config.provisionerApiEnabled) return next(httpError(503, "provisioner_api_disabled"));
  return requireProvisionerCredential(req, next);
}

function requireManagedProvisionerApi(req, _res, next) {
  if (!config.managedProvisionerApiEnabled) return next(httpError(503, "managed_provisioner_api_disabled"));
  if (!verifyManagedProvisionerRequest) return next(httpError(503, "managed_provisioner_api_not_configured"));
  try {
    req.managedProvisionerIdentity = verifyManagedProvisionerRequest(req);
    return next();
  } catch (error) {
    return next(error);
  }
}

function requireFoundryProvisionerApi(req, _res, next) {
  if (!config.foundryProvisionerApiEnabled) return next(httpError(503, "foundry_provisioner_api_disabled"));
  if (!verifyFoundryProvisionerRequest) return next(httpError(503, "foundry_provisioner_api_not_configured"));
  try {
    req.managedProvisionerIdentity = verifyFoundryProvisionerRequest(req);
    return next();
  } catch (error) {
    return next(error);
  }
}

function requireProvisionerCredential(req, next) {
  if (!config.provisionerAccessToken) return next(httpError(503, "provisioner_api_not_configured"));
  const value = bearerToken(req);
  if (!value || !safeEqual(value, config.provisionerAccessToken)) return next(httpError(401, "provisioner_unauthorized"));
  return next();
}

function requireWriterBinding(req, _res, next) {
  return resolveWriterBinding(req)
    .then((binding) => {
      req.writerBinding = binding;
      next();
    })
    .catch(next);
}

function requireReaderBinding(req, _res, next) {
  return resolveReaderBinding(req)
    .then((binding) => {
      req.readerBinding = binding;
      next();
    })
    .catch(next);
}

async function resolveWriterBinding(req) {
  const token = bearerToken(req);
  if (!token) throw httpError(401, "writer_binding_unauthorized");
  const result = await pool.query(
    `select id, tenant_id as "tenantId", connection_id as "connectionId",
       provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
       expires_at as "expiresAt", active, created_at as "createdAt",
       rotated_at as "rotatedAt", revoked_at as "revokedAt"
     from external_data_plane_writer_bindings
     where token_hash = $1 and active = true
       and (expires_at is null or expires_at > now())`,
    [hashWriterToken(token)],
  );
  if (!result.rowCount) throw httpError(401, "writer_binding_unauthorized");
  return result.rows[0];
}

async function resolveReaderBinding(req) {
  const token = bearerToken(req);
  if (!token) throw httpError(401, "reader_binding_unauthorized");
  const result = await pool.query(
    `select id, binding_key as "bindingKey", token_hash as "tokenHash",
       tenant_id as "tenantId", connection_id as "connectionId",
       source_connection_id as "sourceConnectionId",
       provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
       expires_at as "expiresAt", active, created_at as "createdAt",
       rotated_at as "rotatedAt", revoked_at as "revokedAt"
     from external_data_plane_reader_bindings
     where token_hash = $1 and active = true
       and (expires_at is null or expires_at > now())`,
    [hashReaderToken(token)],
  );
  if (!result.rowCount) throw httpError(401, "reader_binding_unauthorized");
  if (result.rows[0].bindingKey && !result.rows[0].sourceConnectionId) {
    throw httpError(409, "reader_source_scope_unresolved");
  }
  return result.rows[0];
}

async function assertReaderStreamAccess(binding, dataProductId) {
  const result = await pool.query(
    `select binding.id
     from external_data_plane_reader_bindings as binding
     join external_data_plane_products as product
       on product.id = $3 and product.active = true
     where binding.id = $1 and binding.token_hash = $2
       and binding.active = true
       and (binding.expires_at is null or binding.expires_at > now())
       and binding.allowed_data_product_ids ? $3`,
    [binding.id, binding.tokenHash, dataProductId],
  );
  if (!result.rowCount) throw httpError(401, "reader_stream_access_revoked");
}

async function assertLegacyBatchProduct(batch) {
  const definition = await loadDataProductDefinition(pool, batch.contract.dataProductId);
  if (!definition) throw httpError(422, "legacy_data_product_not_registered");
  if (definition.version !== batch.contract.version || definition.ontologyRevision !== batch.contract.ontologyRevision) {
    throw httpError(422, "legacy_data_product_contract_mismatch");
  }
  assertPublishMatchesDefinition(batch, definition);
}

async function listGrantedProducts(binding) {
  const allowed = Array.isArray(binding.allowedDataProductIds) ? binding.allowedDataProductIds : [];
  if (!allowed.length) return [];
  const result = await pool.query(
    `select id, version, ontology_revision as "ontologyRevision",
       delivery_mode as "deliveryMode", semantic_types as "semanticTypes",
       fields, field_contracts as "fieldContracts", history_policy as "historyPolicy", active,
       created_at as "createdAt", updated_at as "updatedAt"
     from external_data_plane_products
     where active = true and id = any($1::text[])
     order by id asc`,
    [allowed],
  );
  return result.rows.map(safeDataProductDefinition);
}

async function assertRegisteredProductIds(dataProductIds) {
  const result = await pool.query(
    "select id from external_data_plane_products where active = true and id = any($1::text[])",
    [dataProductIds],
  );
  const registered = new Set(result.rows.map((row) => row.id));
  if (dataProductIds.some((id) => !registered.has(id))) throw httpError(422, "binding_data_product_not_registered");
}

function requireScope(req) {
  const tenantId = String(req.get("x-nodedc-tenant-id") || "");
  const connectionId = String(req.get("x-nodedc-connection-id") || "");
  if (!isIdentifier(tenantId) || !isIdentifier(connectionId)) throw httpError(400, "scope_headers_required");
  return { tenantId, connectionId };
}

function hasScopeHeaders(req) {
  return Object.hasOwn(req.headers, "x-nodedc-tenant-id") || Object.hasOwn(req.headers, "x-nodedc-connection-id");
}

function bearerToken(req) {
  return String(req.get("authorization") || "").replace(/^Bearer\s+/i, "");
}

function requireUuid(value, code) {
  const normalized = String(value || "");
  if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(normalized)) {
    throw httpError(400, code);
  }
  return normalized;
}

function requireIdentifier(value, code) {
  const normalized = String(value || "");
  if (!isIdentifier(normalized)) throw httpError(400, code);
  return normalized;
}

function requirePositiveInteger(value, code) {
  const normalized = String(value ?? "");
  if (!/^[1-9]\d*$/.test(normalized)) throw httpError(400, code);
  const number = Number(normalized);
  if (!Number.isSafeInteger(number) || number > 2_147_483_647) throw httpError(400, code);
  return number;
}

function parseCursor(value) {
  const normalized = String(value ?? "");
  if (!/^(?:0|[1-9]\d*)$/.test(normalized)) throw httpError(400, "data_product_cursor_invalid");
  const cursor = BigInt(normalized);
  if (cursor > 9_223_372_036_854_775_807n) throw httpError(400, "data_product_cursor_invalid");
  return cursor;
}

function parseSourceIds(value) {
  if (value === undefined || value === null || value === "") return [];
  const values = Array.isArray(value) ? value : String(value).split(",");
  return values.map((item) => String(item || "").trim()).filter(Boolean);
}

function writePatchEvent(res, event) {
  return writeSseFrame(res, `id: ${event.cursor}\nevent: nodedc.data-product.patch.v1\ndata: ${JSON.stringify(event)}\n\n`);
}

function writeSseFrame(res, frame) {
  if (res.writableEnded || res.destroyed) return Promise.reject(httpError(499, "reader_stream_closed"));
  if (res.write(frame)) return Promise.resolve();
  return new Promise((resolve, reject) => {
    const cleanup = () => {
      res.off("drain", onDrain);
      res.off("close", onClose);
      res.off("error", onError);
    };
    const onDrain = () => { cleanup(); resolve(); };
    const onClose = () => { cleanup(); reject(httpError(499, "reader_stream_closed")); };
    const onError = (error) => { cleanup(); reject(error); };
    res.once("drain", onDrain);
    res.once("close", onClose);
    res.once("error", onError);
  });
}

function sendOneTimeCapability(res, status, payload) {
  // Create/rotate is the only boundary where a capability exists in
  // plaintext. A trusted provisioner must consume the response in memory and
  // place it directly into its opaque destination; intermediaries must never
  // cache or persist it.
  res.set({
    "Cache-Control": "no-store, max-age=0",
    Pragma: "no-cache",
    Expires: "0",
  });
  return res.status(status).json(payload);
}

function safeEqual(left, right) {
  const leftBuffer = Buffer.from(left);
  const rightBuffer = Buffer.from(right);
  return leftBuffer.length === rightBuffer.length && timingSafeEqual(leftBuffer, rightBuffer);
}

function boundedLimit(value, fallback, min, max) {
  const candidate = Number.parseInt(String(value ?? ""), 10);
  if (!Number.isInteger(candidate)) return fallback;
  return Math.max(min, Math.min(max, candidate));
}

function hash(value) {
  return createHash("sha256").update(JSON.stringify(value)).digest("hex");
}

function isIdentifier(value) {
  return /^[a-z][a-z0-9._:-]{2,127}$/i.test(value);
}

function asyncRoute(handler) {
  return (req, res, next) => Promise.resolve(handler(req, res, next)).catch(next);
}

function httpError(status, code) {
  return Object.assign(new Error(code), { status, code });
}

function safeErrorCode(error) {
  return String(error?.code || error?.message || "internal_error").replace(/[^a-z0-9_.:-]/gi, "_").slice(0, 120);
}

async function sweepExpiredRawEnvelopes() {
  const result = await pool.query(
    `with expired as (
       select id from external_data_plane_raw_envelopes
       where expires_at < now()
       order by expires_at asc
       limit $1
     )
     delete from external_data_plane_raw_envelopes as target
     using expired where target.id = expired.id`,
    [config.retentionDeleteLimit],
  );
  return result.rowCount;
}

async function sweepRetention() {
  const rawEnvelopeCount = await sweepExpiredRawEnvelopes();
  const patchEventCount = await prunePatchOutbox(pool, {
    retentionMs: config.patchRetentionMs,
    limit: config.retentionDeleteLimit,
  });
  const historyFactCount = await pruneDataProductHistory(pool, { limit: config.retentionDeleteLimit });
  const batchReceiptCount = await pruneBatchReceipts(pool, {
    retentionMs: config.receiptRetentionMs,
    limit: config.retentionDeleteLimit,
  });
  lastRetentionSweepAt = new Date().toISOString();
  return { rawEnvelopeCount, patchEventCount, historyFactCount, batchReceiptCount };
}

async function shutdown() {
  if (shutdownPromise) return shutdownPromise;
  shutdownPromise = (async () => {
    shuttingDown = true;
    if (retentionSweepTimer) clearInterval(retentionSweepTimer);
    for (const response of activeStreamResponses) {
      if (!response.writableEnded) response.end();
    }
    const closed = new Promise((resolve) => httpServer.close(resolve));
    httpServer.closeIdleConnections?.();
    const forceClose = setTimeout(() => httpServer.closeAllConnections?.(), 250);
    await Promise.race([closed, new Promise((resolve) => setTimeout(resolve, 2_000))]);
    clearTimeout(forceClose);
    httpServer.closeAllConnections?.();
    await pool.end();
  })();
  return shutdownPromise;
}