import { createHash, randomUUID } from "node:crypto";
import { mkdir, readFile, readdir, rename, writeFile } from "node:fs/promises";
import { join } from "node:path";

const STATE_SCHEMA_VERSION = "nodedc.module-foundry.data-product-consumer/v1";
const PLAN_SCHEMA_VERSION = "nodedc.module-foundry.data-product-consumer-plan/v1";
const ACCEPTANCE_SCHEMA_VERSION = "nodedc.module-foundry.data-product-consumer-acceptance/v1";
const PRESENTATION_PATCH_SCHEMA_VERSION = "nodedc.foundry.presentation-patch/v1";
const identifier = /^[A-Za-z0-9._:-]{1,160}$/;
const cursorPattern = /^(?:0|[1-9]\d*)$/;

function consumerError(code, statusCode = 400) {
  return Object.assign(new Error(code), { statusCode });
}

function targetIdentity(target) {
  const applicationId = String(target?.application?.id || target?.applicationId || "");
  const pageId = String(target?.page?.id || target?.pageId || "");
  const binding = target?.binding || {};
  const bindingId = String(binding.id || target?.bindingId || "");
  if (!/^[0-9a-f-]{36}$/i.test(applicationId) || !/^[a-z0-9-]+$/i.test(pageId) || !identifier.test(bindingId)) {
    throw consumerError("data_product_consumer_target_invalid");
  }
  return { applicationId, pageId, bindingId };
}

function targetKey(target) {
  const identity = targetIdentity(target);
  return `${identity.applicationId}/${identity.pageId}/${identity.bindingId}`;
}

function stateFileName(target) {
  return `${createHash("sha256").update(targetKey(target), "utf8").digest("hex")}.json`;
}

function factKey(fact) {
  return `${fact.semanticType}\u0000${fact.sourceId}`;
}

function safeErrorCode(error) {
  const value = String(error?.message || "data_product_consumer_failed");
  return /^[a-z][a-z0-9_]{2,120}$/.test(value) ? value : "data_product_consumer_failed";
}

function safeStatusFromFact(fact, policy, nowMs) {
  const statusContract = policy.statusContract || null;
  if (statusContract) {
    const sourceStatus = fact?.attributes?.[statusContract.attribute];
    if (typeof sourceStatus !== "string" || !statusContract.allowedValues.includes(sourceStatus)) {
      throw consumerError("data_product_consumer_fact_status_invalid", 502);
    }
    if (policy.freshness === "none" || policy.terminalStatuses.includes(sourceStatus)) {
      return sourceStatus;
    }
    const observedAt = Date.parse(String(fact?.observedAt || ""));
    if (Number.isFinite(observedAt) && nowMs - observedAt > policy.staleAfterMs) return "stale";
    return sourceStatus;
  }
  const sourceStatus = [fact?.attributes?.operational_status, fact?.attributes?.status]
    .find((value) => typeof value === "string" && value.trim());
  const normalized = String(sourceStatus || "active").trim().toLowerCase().replace(/[^a-z0-9_-]+/g, "-").slice(0, 64) || "active";
  if (policy.terminalStatuses.includes(normalized)) return normalized;
  if (policy.freshness === "none") return normalized;
  const observedAt = Date.parse(String(fact?.observedAt || ""));
  if (Number.isFinite(observedAt) && nowMs - observedAt > policy.staleAfterMs) return "stale";
  return normalized;
}

function decorateFact(subject) {
  return { ...subject.fact, presentationStatus: subject.status };
}

function safeTarget(target) {
  const identity = targetIdentity(target);
  const binding = target.binding || {};
  return {
    ...identity,
    dataProductId: String(binding.dataProductId || ""),
    slotId: String(binding.slotId || ""),
    delivery: "snapshot+patch",
    semanticTypes: [...(binding.semanticTypes || [])],
    fieldProjection: [...(binding.fieldProjection || [])],
    dataClass: binding.dataClass === "restricted" ? "restricted" : "operational",
  };
}

function planHash(value) {
  return `fcp1_${createHash("sha256").update(JSON.stringify(value), "utf8").digest("base64url")}`;
}

function emptyMetrics() {
  return {
    snapshotCommits: 0,
    patchCommits: 0,
    patchOperations: 0,
    replayedPatches: 0,
    snapshotRebaseRemovals: 0,
    canonicalRemovals: 0,
    staleTransitions: 0,
    reconnects: 0,
  };
}

function safeStateSummary(record) {
  const state = record.state;
  const subjects = Object.values(state.subjects || {});
  return {
    schemaVersion: STATE_SCHEMA_VERSION,
    consumerId: state.consumerId,
    target: state.target,
    enabled: state.enabled === true,
    runtimeState: state.runtimeState,
    product: state.product,
    policy: state.policy,
    cursor: state.cursor,
    snapshotGeneration: state.snapshotGeneration,
    subjectCount: subjects.length,
    subjects: subjects.slice(0, 20).map((subject) => ({
      subjectId: subject.fact.sourceId,
      semanticType: subject.fact.semanticType,
      observedAt: subject.fact.observedAt,
      status: subject.status,
    })),
    viewerCount: record.listeners.size,
    upstreamStreamCount: record.stream ? 1 : 0,
    metrics: state.metrics,
    timestamps: state.timestamps,
    lastError: state.lastError || null,
    readerGrant: "target-scoped-server-only",
    readerGrantGeneration: Number.isSafeInteger(state.readerGrantGeneration) ? state.readerGrantGeneration : 1,
  };
}

function activeReaderGrantGeneration(state) {
  const generation = Number(state?.readerGrantGeneration ?? 1);
  if (!Number.isSafeInteger(generation) || generation < 1 || generation > 2_147_483_647) {
    throw consumerError("data_product_reader_grant_generation_invalid", 500);
  }
  return generation;
}

function validateConsumerContract(value) {
  if (value === undefined) return null;
  const keys = value && typeof value === "object" && !Array.isArray(value)
    ? Object.keys(value).sort()
    : [];
  if (JSON.stringify(keys) !== JSON.stringify([
    "deliveryMode",
    "fieldProjection",
    "fieldTypes",
    "geometryTypes",
    "ontologyRevision",
    "patchMode",
    "semanticTypes",
    "snapshotMode",
    "subjectIdentity",
  ])) throw consumerError("data_product_consumer_policy_invalid", 500);
  const semanticTypes = Array.isArray(value.semanticTypes) ? value.semanticTypes : [];
  const fieldProjection = Array.isArray(value.fieldProjection) ? value.fieldProjection : [];
  const geometryTypes = Array.isArray(value.geometryTypes) ? value.geometryTypes : [];
  const supportedGeometryTypes = new Set(["Point", "LineString", "Polygon", "MultiPolygon"]);
  const fieldTypes = value.fieldTypes && typeof value.fieldTypes === "object" && !Array.isArray(value.fieldTypes)
    ? value.fieldTypes
    : null;
  if (
    !identifier.test(String(value.ontologyRevision || ""))
    || value.deliveryMode !== "snapshot+patch"
    || semanticTypes.length === 0
    || semanticTypes.length > 8
    || new Set(semanticTypes).size !== semanticTypes.length
    || semanticTypes.some((item) => !identifier.test(item))
    || fieldProjection.length === 0
    || fieldProjection.length > 32
    || new Set(fieldProjection).size !== fieldProjection.length
    || fieldProjection.some((item) => !/^[A-Za-z][A-Za-z0-9_.-]{0,127}$/.test(item))
    || !fieldTypes
    || JSON.stringify(Object.keys(fieldTypes)) !== JSON.stringify(fieldProjection)
    || Object.values(fieldTypes).some((type) => !["string", "number", "boolean", "string_array"].includes(type))
    || geometryTypes.length > supportedGeometryTypes.size
    || new Set(geometryTypes).size !== geometryTypes.length
    || geometryTypes.some((type) => !supportedGeometryTypes.has(type))
    || value.subjectIdentity !== "semantic-type+source-id"
    || value.snapshotMode !== "atomic-replace"
    || value.patchMode !== "atomic"
  ) throw consumerError("data_product_consumer_policy_invalid", 500);
  return {
    ontologyRevision: value.ontologyRevision,
    deliveryMode: "snapshot+patch",
    semanticTypes: [...semanticTypes],
    fieldProjection: [...fieldProjection],
    fieldTypes: { ...fieldTypes },
    geometryTypes: [...geometryTypes],
    subjectIdentity: "semantic-type+source-id",
    snapshotMode: "atomic-replace",
    patchMode: "atomic",
  };
}

function assertConsumerContract(contract, product, binding) {
  if (!contract) return;
  const productFieldTypes = Object.fromEntries(contract.fieldProjection.map((field) => [
    field,
    product?.fieldContracts?.[field]?.type,
  ]));
  const productHasGeometry = Array.isArray(product.fields) && product.fields.includes("geometry");
  const contractHasGeometry = contract.geometryTypes.length > 0;
  if (
    product.ontologyRevision !== contract.ontologyRevision
    || product.deliveryMode !== contract.deliveryMode
    || JSON.stringify(product.semanticTypes) !== JSON.stringify(contract.semanticTypes)
    || JSON.stringify(binding.semanticTypes) !== JSON.stringify(contract.semanticTypes)
    || JSON.stringify(binding.fieldProjection) !== JSON.stringify(contract.fieldProjection)
    || !Array.isArray(product.fields)
    || contract.fieldProjection.some((field) => !product.fields.includes(field))
    || JSON.stringify(productFieldTypes) !== JSON.stringify(contract.fieldTypes)
    || productHasGeometry !== contractHasGeometry
    || (contractHasGeometry && product?.fieldContracts?.geometry?.type !== "geometry")
  ) throw consumerError("data_product_consumer_contract_mismatch", 409);
}

function assertEnvelopeProduct(envelope, product, errorCode) {
  if (envelope?.dataProduct?.id !== product.id || envelope?.dataProduct?.version !== product.version) {
    throw consumerError(errorCode, 502);
  }
}

function assertFactContract(fact, policy) {
  const contract = policy.consumerContract;
  if (!contract) return;
  if (!contract.semanticTypes.includes(fact.semanticType)) {
    throw consumerError("data_product_consumer_fact_contract_invalid", 502);
  }
  const attributes = fact?.attributes && typeof fact.attributes === "object" && !Array.isArray(fact.attributes)
    ? fact.attributes
    : {};
  for (const [field, type] of Object.entries(contract.fieldTypes)) {
    if (attributes[field] === undefined) continue;
    const value = attributes[field];
    const valid = type === "string_array"
      ? Array.isArray(value) && value.every((item) => typeof item === "string")
      : typeof value === type && (type !== "number" || Number.isFinite(value));
    if (!valid) throw consumerError("data_product_consumer_fact_contract_invalid", 502);
  }
  const hasGeometry = fact.geometry !== undefined && fact.geometry !== null;
  if (contract.geometryTypes.length === 0) {
    if (hasGeometry) throw consumerError("data_product_consumer_fact_contract_invalid", 502);
    return;
  }
  if (!hasGeometry || !contract.geometryTypes.includes(fact.geometry.type)) {
    throw consumerError("data_product_consumer_fact_contract_invalid", 502);
  }
}

function validatePolicy(policy, product) {
  if (!policy || policy.dataProductId !== product.id || policy.productVersion !== product.version) {
    throw consumerError("data_product_consumer_policy_not_found", 409);
  }
  const statusContract = policy.statusContract === undefined
    ? null
    : validateStatusContract(policy.statusContract);
  const freshness = policy.freshness ?? statusContract?.freshness ?? "observed-at";
  if (!["none", "observed-at"].includes(freshness)) {
    throw consumerError("data_product_consumer_policy_invalid", 500);
  }
  if (statusContract && statusContract.freshness !== freshness) {
    throw consumerError("data_product_consumer_policy_invalid", 500);
  }
  const staleDisabled = freshness === "none";
  if (
    (staleDisabled && policy.staleAfterMs !== null)
    || (!staleDisabled && (!Number.isInteger(policy.staleAfterMs) || policy.staleAfterMs < 1_000 || policy.staleAfterMs > 7 * 24 * 60 * 60 * 1000))
  ) {
    throw consumerError("data_product_consumer_policy_invalid", 500);
  }
  const terminalStatuses = Array.isArray(policy.terminalStatuses) ? policy.terminalStatuses : [];
  if (terminalStatuses.some((value) => typeof value !== "string" || !/^[a-z0-9_-]{1,64}$/.test(value))) {
    throw consumerError("data_product_consumer_policy_invalid", 500);
  }
  if (statusContract && terminalStatuses.some((value) => !statusContract.allowedValues.includes(value))) {
    throw consumerError("data_product_consumer_policy_invalid", 500);
  }
  if (policy.removeMode !== "canonical-tombstone-or-snapshot-rebase") {
    throw consumerError("data_product_consumer_policy_invalid", 500);
  }
  const consumerContract = validateConsumerContract(policy.consumerContract);
  const dataClass = policy.dataClass ?? "operational";
  if (!["operational", "restricted"].includes(dataClass)) {
    throw consumerError("data_product_consumer_policy_invalid", 500);
  }
  return {
    id: String(policy.id || ""),
    version: String(policy.version || ""),
    dataProductId: product.id,
    productVersion: product.version,
    dataClass,
    freshness,
    staleAfterMs: policy.staleAfterMs,
    terminalStatuses: [...terminalStatuses],
    ...(statusContract ? { statusContract } : {}),
    ...(consumerContract ? { consumerContract } : {}),
    removeMode: policy.removeMode,
  };
}

function validateStatusContract(value) {
  const keys = value && typeof value === "object" && !Array.isArray(value)
    ? Object.keys(value).sort()
    : [];
  if (JSON.stringify(keys) !== JSON.stringify(["allowedValues", "attribute", "freshness", "missing"])) {
    throw consumerError("data_product_consumer_policy_invalid", 500);
  }
  const attribute = String(value.attribute || "");
  const allowedValues = Array.isArray(value.allowedValues) ? value.allowedValues : [];
  if (
    !/^[A-Za-z][A-Za-z0-9_.-]{0,127}$/.test(attribute)
    || allowedValues.length === 0
    || allowedValues.length > 32
    || new Set(allowedValues).size !== allowedValues.length
    || allowedValues.some((item) => typeof item !== "string" || !/^[a-z0-9_-]{1,64}$/.test(item))
    || value.missing !== "reject"
    || !["none", "observed-at"].includes(value.freshness)
  ) {
    throw consumerError("data_product_consumer_policy_invalid", 500);
  }
  return {
    attribute,
    allowedValues: [...allowedValues],
    missing: "reject",
    freshness: value.freshness,
  };
}

function parseSseBlock(block) {
  const fields = { event: "message", id: "", data: [] };
  for (const line of block.split("\n")) {
    if (!line || line.startsWith(":")) continue;
    const separator = line.indexOf(":");
    const name = separator === -1 ? line : line.slice(0, separator);
    const value = separator === -1 ? "" : line.slice(separator + 1).replace(/^ /, "");
    if (name === "event") fields.event = value;
    if (name === "id") fields.id = value;
    if (name === "data") fields.data.push(value);
  }
  return { event: fields.event, id: fields.id, data: fields.data.join("\n") };
}

function waitForAbortableDelay(ms, signal) {
  if (signal.aborted) return Promise.resolve();
  return new Promise((resolve) => {
    const timer = setTimeout(done, ms);
    function done() {
      clearTimeout(timer);
      signal.removeEventListener("abort", done);
      resolve();
    }
    signal.addEventListener("abort", done, { once: true });
  });
}

export function createFoundryDataProductConsumerManager({
  stateDir,
  dataPlaneUrl,
  resolveTarget,
  readReaderToken,
  inspectReaderGrant,
  ensureReaderGrant,
  revokeReaderGrant,
  resolvePolicy,
  sanitizeSnapshot,
  sanitizePatch,
  fetchImpl = fetch,
  openStream,
  now = () => Date.now(),
  idleStopMs = 2_000,
  staleSweepMs = 5_000,
  reconnectMinMs = 500,
  reconnectMaxMs = 15_000,
}) {
  const records = new Map();
  const ready = mkdir(stateDir, { recursive: true });
  let closed = false;

  async function writeState(record) {
    await ready;
    const targetPath = join(stateDir, stateFileName(record.state.target));
    const tempPath = `${targetPath}.${randomUUID()}.tmp`;
    await writeFile(tempPath, `${JSON.stringify(record.state, null, 2)}\n`, { encoding: "utf8", mode: 0o600 });
    await rename(tempPath, targetPath);
  }

  async function readState(target) {
    await ready;
    try {
      const parsed = JSON.parse(await readFile(join(stateDir, stateFileName(target)), "utf8"));
      if (parsed?.schemaVersion !== STATE_SCHEMA_VERSION || targetKey(parsed.target) !== targetKey(target)) {
        throw consumerError("data_product_consumer_state_invalid", 500);
      }
      return parsed;
    } catch (error) {
      if (error?.code === "ENOENT") return null;
      throw error;
    }
  }

  async function recordFor(target) {
    const key = targetKey(target);
    if (records.has(key)) return records.get(key);
    const persisted = await readState(target);
    const record = {
      key,
      target,
      state: persisted,
      listeners: new Map(),
      stream: null,
      idleTimer: null,
      writing: Promise.resolve(),
    };
    records.set(key, record);
    return record;
  }

  function serialize(record, action) {
    const pending = record.writing.then(action, action);
    record.writing = pending.catch(() => undefined);
    return pending;
  }

  function emit(record, event) {
    for (const listener of record.listeners.values()) {
      try { listener(event); } catch { /* one browser cannot stop the shared consumer */ }
    }
  }

  async function catalog(target, { generation = 1, ensureGeneration = generation } = {}) {
    if (!dataPlaneUrl) throw consumerError("data_product_runtime_not_configured", 503);
    let product = null;
    let readerGrantAction = "reuse";
    let readerGrantGeneration = generation;
    if (inspectReaderGrant) {
      const inspection = await inspectReaderGrant(target, { generation });
      product = inspection?.product || null;
      readerGrantAction = inspection?.readerGrantAction === "ensure" ? "ensure" : "reuse";
      readerGrantGeneration = readerGrantAction === "ensure" ? ensureGeneration : generation;
    } else {
      const token = await readReaderToken(target.application.id, target.page.id, target.binding.id, generation);
      let response;
      try {
        response = await fetchImpl(new URL("/internal/data-plane/v1/reader/data-products", `${dataPlaneUrl}/`), {
          headers: { authorization: `Bearer ${token}`, accept: "application/json" },
          signal: AbortSignal.timeout(10_000),
        });
      } catch {
        throw consumerError("data_product_consumer_catalog_unavailable", 503);
      }
      if (response.status === 401 || response.status === 403) throw consumerError("data_product_access_denied", 403);
      if (!response.ok) throw consumerError("data_product_consumer_catalog_unavailable", 503);
      const payload = await response.json().catch(() => null);
      product = Array.isArray(payload?.dataProducts)
        ? payload.dataProducts.find((candidate) => candidate?.id === target.binding.dataProductId)
        : null;
    }
    if (!product || product.active === false) throw consumerError("data_product_not_found", 404);
    if (product.deliveryMode !== "snapshot+patch" || typeof product.version !== "string") {
      throw consumerError("data_product_consumer_delivery_not_supported", 409);
    }
    if (target.binding.semanticTypes.some((semanticType) => !product.semanticTypes?.includes(semanticType))) {
      throw consumerError("data_product_consumer_semantic_scope_mismatch", 409);
    }
    const policy = validatePolicy(resolvePolicy(product), product);
    if ((target.binding.dataClass ?? "operational") !== policy.dataClass) {
      throw consumerError("data_product_consumer_data_class_mismatch", 409);
    }
    assertConsumerContract(policy.consumerContract, product, target.binding);
    if (policy.statusContract && !target.binding.fieldProjection.includes(policy.statusContract.attribute)) {
      throw consumerError("data_product_consumer_status_field_not_projected", 409);
    }
    return {
      product,
      policy,
      readerGrantAction,
      readerGrantGeneration,
    };
  }

  async function plan(input) {
    const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
    const record = await recordFor(target);
    const currentReaderGrantGeneration = activeReaderGrantGeneration(record.state);
    const ensureReaderGrantGeneration = record.state
      ? Math.min(2_147_483_647, currentReaderGrantGeneration + 1)
      : currentReaderGrantGeneration;
    const { product, policy, readerGrantAction, readerGrantGeneration } = await catalog(target, {
      generation: currentReaderGrantGeneration,
      ensureGeneration: ensureReaderGrantGeneration,
    });
    if (readerGrantAction === "ensure" && record.state && currentReaderGrantGeneration === 2_147_483_647) {
      throw consumerError("data_product_reader_grant_generation_exhausted", 409);
    }
    const safe = safeTarget(target);
    const configuration = {
      target: safe,
      product: { id: product.id, version: product.version },
      policy,
      readerGrant: "target-scoped-server-only",
      readerGrantAction,
      readerGrantGeneration,
    };
    const existingMatches = record.state
      && JSON.stringify(record.state.target) === JSON.stringify(safe)
      && record.state.product?.id === product.id
      && record.state.product?.version === product.version
      && record.state.policy?.id === policy.id
      && record.state.policy?.version === policy.version;
    const predecessorReaderGrantGeneration = Number.isSafeInteger(record.state?.predecessorReaderGrantGeneration)
      ? record.state.predecessorReaderGrantGeneration
      : null;
    const action = !record.state
      ? "create"
      : existingMatches && predecessorReaderGrantGeneration
        ? "finalize-reader-grant-rotation"
        : existingMatches
          ? (record.state.enabled ? "refresh" : "resume")
          : "replace";
    return {
      schemaVersion: PLAN_SCHEMA_VERSION,
      planId: planHash({ configuration, action }),
      action,
      configuration,
      current: record.state ? { enabled: record.state.enabled === true, cursor: record.state.cursor, product: record.state.product } : null,
      effects: [
        ...(readerGrantAction === "ensure"
          ? [record.state ? "ensure-successor-target-scoped-reader-grant" : "ensure-target-scoped-reader-grant"]
          : []),
        "persist-consumer-state",
        ...(action === "finalize-reader-grant-rotation" ? [] : ["bootstrap-scoped-snapshot"]),
        ...(predecessorReaderGrantGeneration ? ["revoke-predecessor-reader-grant"] : []),
        "share-one-upstream-stream-per-active-binding",
      ],
      destructive: false,
    };
  }

  function initialState(target, product, policy, readerGrantGeneration) {
    const timestamp = new Date(now()).toISOString();
    return {
      schemaVersion: STATE_SCHEMA_VERSION,
      consumerId: createHash("sha256").update(targetKey(target), "utf8").digest("hex").slice(0, 32),
      target: safeTarget(target),
      enabled: true,
      runtimeState: "bootstrapping",
      product: { id: product.id, version: product.version },
      policy,
      readerGrantGeneration,
      predecessorReaderGrantGeneration: null,
      cursor: "0",
      snapshotGeneration: 0,
      subjects: {},
      metrics: emptyMetrics(),
      timestamps: { createdAt: timestamp, updatedAt: timestamp, lastSnapshotAt: null, lastPatchAt: null, lastConnectedAt: null },
      lastError: null,
    };
  }

  function upstreamUrl(record, resource, after = "") {
    const url = new URL(`/internal/data-plane/v1/data-products/${encodeURIComponent(record.state.target.dataProductId)}/${resource}`, `${dataPlaneUrl}/`);
    if (after) url.searchParams.set("after", after);
    return url;
  }

  async function fetchSnapshot(record) {
    const token = await readReaderToken(
      record.state.target.applicationId,
      record.state.target.pageId,
      record.state.target.bindingId,
      activeReaderGrantGeneration(record.state),
    );
    let response;
    try {
      response = await fetchImpl(upstreamUrl(record, "snapshot"), {
        headers: { authorization: `Bearer ${token}`, accept: "application/json" },
        signal: AbortSignal.timeout(10_000),
      });
    } catch {
      throw consumerError("data_product_runtime_unavailable", 503);
    }
    if (response.status === 401 || response.status === 403) throw consumerError("data_product_access_denied", 403);
    if (response.status === 404) throw consumerError("data_product_not_found", 404);
    if (!response.ok) throw consumerError("data_product_runtime_unavailable", 503);
    const payload = await response.json().catch(() => null);
    return sanitizeSnapshot(payload, record.target.binding);
  }

  async function bootstrap(record, { notify = true } = {}) {
    return serialize(record, async () => {
      record.state.runtimeState = "bootstrapping";
      record.state.timestamps.updatedAt = new Date(now()).toISOString();
      await writeState(record);
      try {
        const snapshot = await fetchSnapshot(record);
        assertEnvelopeProduct(snapshot, record.state.product, "data_product_snapshot_product_mismatch");
        const previousKeys = new Set(Object.keys(record.state.subjects || {}));
        const subjects = {};
        for (const fact of snapshot.facts) {
          assertFactContract(fact, record.state.policy);
          const key = factKey(fact);
          subjects[key] = {
            fact,
            status: safeStatusFromFact(fact, record.state.policy, now()),
            lastAppliedCursor: snapshot.cursor,
          };
          previousKeys.delete(key);
        }
        const timestamp = new Date(now()).toISOString();
        record.state = {
          ...record.state,
          enabled: true,
          runtimeState: record.listeners.size ? "connecting" : "idle",
          product: snapshot.dataProduct,
          cursor: snapshot.cursor,
          snapshotGeneration: record.state.snapshotGeneration + 1,
          subjects,
          metrics: {
            ...record.state.metrics,
            snapshotCommits: record.state.metrics.snapshotCommits + 1,
            snapshotRebaseRemovals: record.state.metrics.snapshotRebaseRemovals + previousKeys.size,
          },
          timestamps: { ...record.state.timestamps, updatedAt: timestamp, lastSnapshotAt: timestamp },
          lastError: null,
        };
        await writeState(record);
        if (notify) emit(record, {
          event: "nodedc.data-product.resync-required.v1",
          data: { schemaVersion: "nodedc.data-product.resync-required/v1", dataProductId: record.state.product.id },
        });
        return snapshot;
      } catch (error) {
        const timestamp = new Date(now()).toISOString();
        record.state.runtimeState = "error";
        record.state.lastError = { code: safeErrorCode(error), at: timestamp };
        record.state.timestamps.updatedAt = timestamp;
        await writeState(record);
        throw error;
      }
    });
  }

  async function apply(input) {
    const planned = await plan(input);
    if (input.planId !== planned.planId) throw consumerError("data_product_consumer_plan_mismatch", 409);
    const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
    const record = await recordFor(target);
    const plannedGeneration = planned.configuration.readerGrantGeneration;

    if (planned.action === "finalize-reader-grant-rotation") {
      const predecessorGeneration = record.state?.predecessorReaderGrantGeneration;
      if (!revokeReaderGrant || !Number.isSafeInteger(predecessorGeneration)) {
        throw consumerError("data_product_reader_grant_rotation_invalid", 500);
      }
      await revokeReaderGrant(record.target, { generation: predecessorGeneration });
      record.state.predecessorReaderGrantGeneration = null;
      record.state.timestamps.updatedAt = new Date(now()).toISOString();
      record.state.lastError = null;
      await writeState(record);
      return { plan: planned, consumer: safeStateSummary(record) };
    }

    if (planned.configuration.readerGrantAction === "ensure") {
      if (!ensureReaderGrant) throw consumerError("data_product_reader_grant_not_found", 403);
      await ensureReaderGrant(target, { generation: plannedGeneration });
    }
    const { product, policy, readerGrantAction, readerGrantGeneration } = await catalog(target, {
      generation: plannedGeneration,
      ensureGeneration: plannedGeneration,
    });
    if (readerGrantAction !== "reuse") throw consumerError("data_product_reader_grant_not_ready", 409);
    if (readerGrantGeneration !== plannedGeneration) throw consumerError("data_product_reader_grant_generation_mismatch", 409);

    const previousTarget = record.target;
    const previousState = record.state ? structuredClone(record.state) : null;
    const previousGeneration = previousState ? activeReaderGrantGeneration(previousState) : null;
    if (record.stream) await stopStreamAndWait(record);
    if (!record.state || planned.action === "replace") {
      record.target = target;
      record.state = initialState(target, product, policy, plannedGeneration);
      if (previousGeneration && previousGeneration !== plannedGeneration) {
        record.state.predecessorReaderGrantGeneration = previousGeneration;
      }
    }
    else {
      record.target = target;
      record.state.target = safeTarget(target);
      record.state.product = { id: product.id, version: product.version };
      record.state.policy = policy;
      record.state.readerGrantGeneration = plannedGeneration;
      record.state.enabled = true;
      record.state.runtimeState = "bootstrapping";
      record.state.lastError = null;
    }
    try {
      await writeState(record);
      await bootstrap(record, { notify: false });
    } catch (error) {
      if (previousState) {
        record.target = previousTarget;
        record.state = previousState;
        record.state.runtimeState = record.listeners.size ? "connecting" : "idle";
        record.state.timestamps.updatedAt = new Date(now()).toISOString();
        await writeState(record);
        if (record.listeners.size) startStream(record);
      }
      throw error;
    }

    const predecessorGeneration = record.state.predecessorReaderGrantGeneration;
    if (Number.isSafeInteger(predecessorGeneration)) {
      if (!revokeReaderGrant) throw consumerError("data_product_reader_grant_rotation_not_configured", 503);
      await revokeReaderGrant(previousTarget, { generation: predecessorGeneration });
      record.state.predecessorReaderGrantGeneration = null;
      record.state.timestamps.updatedAt = new Date(now()).toISOString();
      await writeState(record);
    }
    if (record.listeners.size) startStream(record);
    return { plan: planned, consumer: safeStateSummary(record) };
  }

  async function ensureProvisioned(target) {
    const record = await recordFor(target);
    if (record.state) {
      if (!record.state.enabled) throw consumerError("data_product_consumer_stopped", 409);
      record.target = target;
      return record;
    }
    const planned = await plan(targetIdentity(target));
    await apply({ ...targetIdentity(target), planId: planned.planId });
    return recordFor(target);
  }

  async function refreshPresentation(record) {
    if (!record.state?.enabled) return [];
    return serialize(record, async () => {
      const operations = [];
      for (const subject of Object.values(record.state.subjects || {})) {
        const status = safeStatusFromFact(subject.fact, record.state.policy, now());
        if (status === subject.status) continue;
        subject.status = status;
        operations.push({ op: "upsert", sourceId: subject.fact.sourceId, semanticType: subject.fact.semanticType, status });
      }
      if (!operations.length) return operations;
      const timestamp = new Date(now()).toISOString();
      record.state.metrics.staleTransitions += operations.filter((operation) => operation.status === "stale").length;
      record.state.timestamps.updatedAt = timestamp;
      await writeState(record);
      emit(record, {
        event: "nodedc.foundry.presentation-patch.v1",
        data: { schemaVersion: PRESENTATION_PATCH_SCHEMA_VERSION, generatedAt: timestamp, operations },
      });
      return operations;
    });
  }

  async function applyPatch(record, patch) {
    return serialize(record, async () => {
      const currentCursor = BigInt(record.state.cursor);
      const patchCursor = BigInt(patch.cursor);
      if (patchCursor <= currentCursor) {
        record.state.metrics.replayedPatches += 1;
        return { replayed: true };
      }
      if (patch.previousCursor !== record.state.cursor) throw consumerError("resync_required", 409);
      const subjects = { ...record.state.subjects };
      const emittedOperations = [];
      for (const operation of patch.operations) {
        if (operation.op === "upsert") {
          assertFactContract(operation.fact, record.state.policy);
          const key = factKey(operation.fact);
          const subject = {
            fact: operation.fact,
            status: safeStatusFromFact(operation.fact, record.state.policy, now()),
            lastAppliedCursor: patch.cursor,
          };
          subjects[key] = subject;
          emittedOperations.push({ op: "upsert", fact: decorateFact(subject) });
        } else if (operation.op === "remove") {
          const key = `${operation.semanticType}\u0000${operation.sourceId}`;
          if (subjects[key]) {
            delete subjects[key];
            emittedOperations.push(operation);
          }
        }
      }
      const timestamp = new Date(now()).toISOString();
      record.state = {
        ...record.state,
        runtimeState: "connected",
        cursor: patch.cursor,
        subjects,
        metrics: {
          ...record.state.metrics,
          patchCommits: record.state.metrics.patchCommits + 1,
          patchOperations: record.state.metrics.patchOperations + emittedOperations.length,
          canonicalRemovals: record.state.metrics.canonicalRemovals + emittedOperations.filter((operation) => operation.op === "remove").length,
        },
        timestamps: { ...record.state.timestamps, updatedAt: timestamp, lastPatchAt: timestamp },
        lastError: null,
      };
      // Persist the new cursor and semantic state before acknowledging it to
      // any browser listener. A crash can replay a patch but cannot skip one.
      await writeState(record);
      if (emittedOperations.length) emit(record, {
        event: "nodedc.data-product.patch.v1",
        id: patch.cursor,
        data: { ...patch, operations: emittedOperations },
      });
      return { replayed: false };
    });
  }

  async function consumeOnce(record, signal) {
    const token = await readReaderToken(
      record.state.target.applicationId,
      record.state.target.pageId,
      record.state.target.bindingId,
      activeReaderGrantGeneration(record.state),
    );
    const url = upstreamUrl(record, "stream", record.state.cursor);
    const response = openStream
      ? await openStream({ url, token, signal })
      : await fetchImpl(url, {
        headers: { authorization: `Bearer ${token}`, accept: "text/event-stream", connection: "close" },
        signal,
      });
    if (response.status === 409) throw consumerError("resync_required", 409);
    if (response.status === 401 || response.status === 403) throw consumerError("data_product_access_denied", 403);
    if (!response.ok || !response.body || !String(response.headers.get("content-type") || "").startsWith("text/event-stream")) {
      throw consumerError("data_product_runtime_unavailable", 503);
    }
    const timestamp = new Date(now()).toISOString();
    record.state.runtimeState = "connected";
    record.state.timestamps.lastConnectedAt = timestamp;
    record.state.timestamps.updatedAt = timestamp;
    record.state.lastError = null;
    await writeState(record);
    const reader = response.body.getReader();
    if (record.stream) record.stream.reader = reader;
    const decoder = new TextDecoder();
    let pending = "";
    let upstreamDone = false;
    try {
      while (!signal.aborted && record.listeners.size) {
        const next = await reader.read();
        if (next.done) {
          upstreamDone = true;
          break;
        }
        pending += decoder.decode(next.value, { stream: true }).replace(/\r\n/g, "\n");
        let separator;
        while ((separator = pending.indexOf("\n\n")) !== -1) {
          const event = parseSseBlock(pending.slice(0, separator));
          pending = pending.slice(separator + 2);
          if (event.event !== "nodedc.data-product.patch.v1") continue;
          let payload = null;
          try { payload = JSON.parse(event.data); } catch { payload = null; }
          const patch = sanitizePatch(payload, record.target.binding);
          if (!patch) throw consumerError("data_product_patch_contract_invalid", 502);
          assertEnvelopeProduct(patch, record.state.product, "data_product_patch_product_mismatch");
          await applyPatch(record, patch);
        }
      }
    } finally {
      if (!upstreamDone) await reader.cancel().catch(() => undefined);
      reader.releaseLock();
      if (record.stream?.reader === reader) record.stream.reader = null;
    }
  }

  function startStream(record) {
    if (closed || record.stream || !record.state?.enabled || !record.listeners.size) return;
    if (record.idleTimer) clearTimeout(record.idleTimer);
    record.idleTimer = null;
    const controller = new AbortController();
    record.stream = { controller, promise: null, reader: null };
    record.stream.promise = (async () => {
      let delay = reconnectMinMs;
      while (!closed && !controller.signal.aborted && record.listeners.size && record.state.enabled) {
        try {
          record.state.runtimeState = "connecting";
          await writeState(record);
          await consumeOnce(record, controller.signal);
          delay = reconnectMinMs;
          if (!controller.signal.aborted && record.listeners.size) throw consumerError("data_product_stream_closed", 503);
        } catch (error) {
          if (controller.signal.aborted || closed || !record.listeners.size) break;
          if (error?.message === "resync_required") {
            await bootstrap(record).catch(() => undefined);
          } else {
            const timestamp = new Date(now()).toISOString();
            record.state.runtimeState = "reconnecting";
            record.state.metrics.reconnects += 1;
            record.state.lastError = { code: safeErrorCode(error), at: timestamp };
            record.state.timestamps.updatedAt = timestamp;
            await writeState(record);
          }
          await waitForAbortableDelay(delay, controller.signal);
          delay = Math.min(reconnectMaxMs, delay * 2);
        }
      }
    })().finally(async () => {
      if (record.stream?.controller === controller) record.stream = null;
      if (record.state?.enabled && !record.listeners.size) {
        record.state.runtimeState = "idle";
        record.state.timestamps.updatedAt = new Date(now()).toISOString();
        await writeState(record).catch(() => undefined);
      }
    });
  }

  function stopStream(record) {
    if (record.idleTimer) clearTimeout(record.idleTimer);
    record.idleTimer = null;
    void record.stream?.reader?.cancel().catch(() => undefined);
    record.stream?.controller.abort();
  }

  async function stopStreamAndWait(record) {
    const pending = record.stream?.promise;
    stopStream(record);
    if (!pending) return;
    await Promise.race([
      pending.catch(() => undefined),
      new Promise((resolve) => setTimeout(resolve, 750)),
    ]);
  }

  async function snapshot(input) {
    const target = input?.binding ? input : await resolveTarget(input.applicationId, input.pageId, input.bindingId);
    const record = await ensureProvisioned(target);
    // A removed target grant revokes browser reads immediately even though the
    // last safe snapshot remains persisted for rollback/diagnostics.
    await readReaderToken(
      record.state.target.applicationId,
      record.state.target.pageId,
      record.state.target.bindingId,
      activeReaderGrantGeneration(record.state),
    );
    await refreshPresentation(record);
    return {
      schemaVersion: "nodedc.data-product.snapshot/v1",
      dataProduct: record.state.product,
      generatedAt: record.state.timestamps.lastSnapshotAt || record.state.timestamps.updatedAt,
      cursor: record.state.cursor,
      facts: Object.values(record.state.subjects).map(decorateFact),
    };
  }

  async function subscribe(input, listener) {
    const target = input?.binding ? input : await resolveTarget(input.applicationId, input.pageId, input.bindingId);
    const record = await ensureProvisioned(target);
    const after = String(input.after || "");
    if (after && !cursorPattern.test(after)) throw consumerError("stream_cursor_invalid");
    if (after && after !== record.state.cursor) return { resyncRequired: true, cursor: record.state.cursor, release() {} };
    const leaseId = randomUUID();
    record.listeners.set(leaseId, listener);
    listener({
      event: "nodedc.data-product.ready.v1",
      data: {
        schemaVersion: "nodedc.data-product.ready/v1",
        dataProductId: record.state.product.id,
        cursor: record.state.cursor,
        emittedAt: new Date(now()).toISOString(),
      },
    });
    startStream(record);
    let released = false;
    return {
      resyncRequired: false,
      cursor: record.state.cursor,
      release() {
        if (released) return;
        released = true;
        record.listeners.delete(leaseId);
        if (!record.listeners.size && record.stream && !record.idleTimer) {
          record.idleTimer = setTimeout(() => {
            record.idleTimer = null;
            if (!record.listeners.size) stopStream(record);
          }, idleStopMs);
          record.idleTimer.unref?.();
        }
      },
    };
  }

  async function status(input) {
    const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
    const record = await recordFor(target);
    if (!record.state) return { found: false, target: safeTarget(target), readerGrant: "target-scoped-server-only" };
    await refreshPresentation(record);
    return { found: true, consumer: safeStateSummary(record) };
  }

  async function accept(input) {
    const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
    const record = await ensureProvisioned(target);
    const timeoutMs = Math.min(30_000, Math.max(250, Number(input.timeoutMs ?? 5_000)));
    const minSubjectCount = Math.min(5_000, Math.max(0, Number(input.minSubjectCount ?? 1)));
    const minPatchCount = Math.min(1_000, Math.max(0, Number(input.minPatchCount ?? 0)));
    if (![timeoutMs, minSubjectCount, minPatchCount].every(Number.isInteger)) throw consumerError("data_product_consumer_acceptance_input_invalid");
    const baselinePatchCount = record.state.metrics.patchCommits;
    const lease = await subscribe(target, () => undefined);
    const deadline = now() + timeoutMs;
    try {
      while (now() < deadline) {
        const patchDelta = record.state.metrics.patchCommits - baselinePatchCount;
        if (Object.keys(record.state.subjects).length >= minSubjectCount && patchDelta >= minPatchCount) break;
        await new Promise((resolve) => setTimeout(resolve, 50));
      }
      await refreshPresentation(record);
      const subjectKeys = Object.keys(record.state.subjects);
      const duplicateCount = subjectKeys.length - new Set(subjectKeys).size;
      const patchDelta = record.state.metrics.patchCommits - baselinePatchCount;
      const upstreamStreamCount = record.stream ? 1 : 0;
      const checks = {
        bindingResolved: true,
        targetScopedReader: true,
        snapshotCommitted: record.state.metrics.snapshotCommits > 0,
        cursorPersisted: cursorPattern.test(record.state.cursor),
        minimumSubjects: subjectKeys.length >= minSubjectCount,
        minimumNewPatches: patchDelta >= minPatchCount,
        duplicateSubjects: duplicateCount === 0,
        singleUpstreamStream: upstreamStreamCount <= 1,
        secretAndEndpointExcluded: true,
      };
      return {
        schemaVersion: ACCEPTANCE_SCHEMA_VERSION,
        accepted: Object.values(checks).every(Boolean),
        checks,
        observed: { subjectCount: subjectKeys.length, newPatchCount: patchDelta, cursor: record.state.cursor, runtimeState: record.state.runtimeState },
        consumer: safeStateSummary(record),
      };
    } finally {
      lease.release();
    }
  }

  async function rollback(input) {
    const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
    const record = await recordFor(target);
    if (!record.state) throw consumerError("data_product_consumer_not_found", 404);
    stopStream(record);
    record.listeners.clear();
    record.state.enabled = false;
    record.state.runtimeState = "stopped";
    record.state.timestamps.updatedAt = new Date(now()).toISOString();
    record.state.lastError = null;
    await writeState(record);
    return { rolledBack: true, snapshotPreserved: true, consumer: safeStateSummary(record) };
  }

  async function resumePersisted() {
    await ready;
    for (const entry of await readdir(stateDir, { withFileTypes: true })) {
      if (!entry.isFile() || !/^[0-9a-f]{64}\.json$/.test(entry.name)) continue;
      try {
        const state = JSON.parse(await readFile(join(stateDir, entry.name), "utf8"));
        if (state?.schemaVersion !== STATE_SCHEMA_VERSION || !state.target) continue;
        const target = await resolveTarget(state.target.applicationId, state.target.pageId, state.target.bindingId);
        const record = await recordFor(target);
        record.target = target;
        if (record.state?.enabled) {
          record.state.runtimeState = "idle";
          record.state.timestamps.updatedAt = new Date(now()).toISOString();
          await writeState(record);
        }
      } catch {
        // One damaged or obsolete target cannot stop Foundry from serving the
        // remaining applications. Its file remains available for diagnostics.
      }
    }
  }

  const staleTimer = setInterval(() => {
    for (const record of records.values()) void refreshPresentation(record).catch(() => undefined);
  }, staleSweepMs);
  staleTimer.unref?.();

  async function shutdown() {
    if (closed) return;
    closed = true;
    clearInterval(staleTimer);
    const streams = [];
    for (const record of records.values()) {
      stopStream(record);
      record.listeners.clear();
      if (record.stream?.promise) streams.push(record.stream.promise);
    }
    await Promise.race([
      Promise.allSettled(streams),
      new Promise((resolve) => setTimeout(resolve, 750)),
    ]);
  }

  return {
    plan,
    apply,
    snapshot,
    subscribe,
    status,
    accept,
    rollback,
    resumePersisted,
    shutdown,
  };
}

export const foundryDataProductConsumerSchemas = Object.freeze({
  state: STATE_SCHEMA_VERSION,
  plan: PLAN_SCHEMA_VERSION,
  acceptance: ACCEPTANCE_SCHEMA_VERSION,
  presentationPatch: PRESENTATION_PATCH_SCHEMA_VERSION,
});