// re-export the post-write audit so consumers (graph-mcp shim)
// import from the same package as `writeNodeWithEdges`. Keeps the
// `[graph-*]` log family colocated.
export * from "./audit.js";

// re-export the conversation-provenance injector. MCP wrappers
// (contacts, memory) import it from the package root so the helper, the
// gate, and the label set live in one import surface.
export * from "./conversation-provenance.js";

/**
 * Write doctrine: a node without at least one adjacency is noise,
 * not knowledge. `writeNodeWithEdges` is the single primitive every graph
 * writer should call; it rejects zero-relationship writes, verifies every
 * relationship target resolves before the node is created, and commits the
 * node + all edges in one managed transaction. Provenance is stamped on the
 * node as flattened `createdBy*` properties (Neo4j does not persist nested
 * maps, and flat props are queryable — `MATCH (n) WHERE n.createdBySession
 * = $id RETURN n` is the forensic entry point).
 *
 * Process provenance doctrine (widened): writes
 * targeting any label in `ACTION_PROVENANCE_LABELS` (Person, UserProfile,
 * AdminUser, Organization, LocalBusiness, CloudflareTunnel,
 * CloudflareHostname) must include at least one inbound `:PRODUCED` edge
 * whose source carries one of `:Task`, `:Conversation`, or `:Message` (the
 * label set is checked against the actual `labels()` of the target node, so
 * subtype labels like `:AdminConversation`, `:UserMessage`, `:AssistantMessage`,
 * and `:AdminMessage` all qualify). This makes every durable LLM-tool-driven
 * entity creation traversable from either:
 *   - the Task that produced it (autonomous workflows: onboarding skill,
 *     cloudflare tunnel-login, etc.), or
 *   - the Conversation/Message turn that triggered it (direct admin asks —
 *     "add Anneke Roux as person" stamps a `:PRODUCED` edge from the active
 *     `:AdminConversation` so the entity is traceable to its causal turn
 *     without forcing the agent to mint a synthetic `:Task` per ask).
 * The Task-mediated path stays canonical for autonomous workflows. Direct
 * admin asks attach via the active Conversation/Message turn to avoid
 * redundant-Task bloat. Bootstrap writes (`createdBy.agent === 'system'`)
 * are exempt — installer and lazy-create paths run as system writers.
 *
 * Rejection paths (every one emits a stderr log the admin server pipes to
 * server.log, so orphan pressure is visible per-write not just in the
 * hourly [graph-health] signal):
 *   - zero relationships          → `[graph-write] reject reason=zero-relationships`
 *   - unresolved target id        → `[graph-write] reject reason=unresolved-target`
 *   - missing provenance          → `[graph-write] warn reason=missing-provenance`
 *     (Task 580 — soft warn; write proceeds. Composer-spawned admin sessions
 *     inherit a bare env that never carries SESSION_NODE_ID, so the prior
 *     hard reject was failing every direct admin write to a gated label.)
 *   - removed-feature write       → `[graph-write] reject reason=removed-feature`
 *     (`:ReviewAlert` and `:Event {actionTool:"review-digest-compose"}` are
 *     deleted features; the gate catches doctrine-compliant writers
 *     re-introducing them. The vitest tombstone grep is the primary fence
 *     for raw `session.run("MERGE …")` callers that bypass this primitive.)
 *
 * The `createdBy.agent` field is advisory, not authoritative — it is sourced
 * from the MCP server's AGENT_SLUG env var, which is set by the trusted
 * spawning code (admin server / workflow runner). A misconfigured spawner
 * will write "unknown"; that's the signal something is wrong. Do not use
 * this field for access control — it is forensic, not a security boundary.
 */

/**
 * Labels that require an inbound `:PRODUCED` edge from a `:Task`,
 * `:Conversation`, or `:Message`. Bootstrap writes with
 * `createdBy.agent === 'system'` are exempt — that carve-out covers
 * PIN-setup `writeAdminUserAndPerson`, schema apply, and the lazy
 * `loadUserProfile` MERGE path that bypasses this primitive entirely
 * (raw Cypher in `platform/ui/app/lib/neo4j-store.ts`).
 *
 * The set is intentionally not exhaustive — labels added here become a
 * blocking gate for every LLM-tool write of that label. Add a label only
 * after every legitimate writer of it can either (a) carry a PRODUCED edge
 * from one of the accepted source families, or (b) declare itself as
 * `agent: 'system'`.
 */
export const ACTION_PROVENANCE_LABELS: ReadonlySet<string> = new Set([
  "Person",
  "UserProfile",
  "AdminUser",
  "Organization",
  "LocalBusiness",
  "CloudflareTunnel",
  "CloudflareHostname",
]);

/**
 * Source labels that satisfy the inbound-PRODUCED gate. The check is
 * performed against the full `labels()` array of the target node, so
 * subtype labels (`:AdminConversation`, `:UserMessage`, etc.) qualify
 * because they carry the parent `:Conversation` / `:Message` label too.
 */
export const PROVENANCE_SOURCE_LABELS: ReadonlySet<string> = new Set([
  "Task",
  "Conversation",
  "Message",
]);

function requiresActionProvenance(labels: readonly string[]): boolean {
  for (const label of labels) {
    if (ACTION_PROVENANCE_LABELS.has(label)) return true;
  }
  return false;
}

/**
 * Labels carrying a `CREATE VECTOR INDEX … ON (n.embedding)` in
 * platform/neo4j/schema.cypher. A node of one of these labels written with a
 * null embedding is invisible to the vector half of memory-search until a
 * manual reindex. This constant is the single runtime source of truth —
 * graph-write stays fs-free (it is bundled into the ESM payload and must not
 * read schema.cypher at runtime), so the memory-plugin drift test
 * (vector-indexed-labels-drift) pins it to the schema instead.
 */
export const VECTOR_INDEXED_LABELS: ReadonlySet<string> = new Set([
  "Question", "DefinedTerm", "Review", "Service", "Person", "LocalBusiness",
  "Organization", "PriceSpecification", "Task", "CreativeWork", "DigitalDocument",
  "KnowledgeDocument", "ConversationArchive", "Section", "Chunk", "Project",
  "Position", "Conversation", "Message", "Event", "Workflow", "Preference",
  "Listing", "Idea", "TimelineEvent", "Report", "FileArtifact",
  // Construction vertical (Task 652; renamed Task 1084) — the four
  // text-bearing labels: Job, the priced :QuoteLine, its :Quote, :VariationNote.
  "Job", "QuoteLine", "Quote", "VariationNote",
  // Knowledge-work vertical — the seven gap labels, each carrying a
  // CREATE VECTOR INDEX in schema.cypher so the chokepoint auto-embeds them.
  "Objective", "KeyResult", "Decision", "Risk", "Source", "Finding", "Hypothesis",
]);

/**
 * Document/archive labels whose primary content is a long body. The safety-net
 * embed prefers a bounded `summary` field over the raw `body`/`text` blob for
 * these; the dedicated archive-ingest path adds the per-:Section embeddings.
 */
export const DOCUMENT_ARCHIVE_LABELS: ReadonlySet<string> = new Set([
  "KnowledgeDocument", "DigitalDocument", "ConversationArchive",
]);

function labelsAreIndexed(labels: readonly string[]): boolean {
  for (const l of labels) if (VECTOR_INDEXED_LABELS.has(l)) return true;
  return false;
}

/**
 * Bounded props→text for the chokepoint safety-net embed. Mirrors
 * memory-write's buildTextRepresentation bound. For document/archive labels the
 * raw `body`/`text` blob is skipped (a `summary` field, if present, carries the
 * gist) so the embed text stays small.
 */
function boundedEmbedText(
  labels: readonly string[],
  props: Record<string, unknown>,
): string {
  const PER_VALUE = 2000;
  const TOTAL = 8000;
  const docPreferred = labels.some((l) => DOCUMENT_ARCHIVE_LABELS.has(l));
  const parts: string[] = [`[${labels.join(", ")}]`];
  for (const [k, v] of Object.entries(props)) {
    if (v === null || v === undefined || k === "embedding") continue;
    if (docPreferred && (k === "body" || k === "text")) continue;
    const s = String(v);
    parts.push(`${k}: ${s.length > PER_VALUE ? s.slice(0, PER_VALUE) : s}`);
  }
  const joined = parts.join(" | ");
  return joined.length > TOTAL ? joined.slice(0, TOTAL) : joined;
}

function findProvenanceCandidates(
  relationships: readonly GraphRelationship[],
): GraphRelationship[] {
  return relationships.filter(
    (r) => r.type === "PRODUCED" && r.direction === "incoming",
  );
}

import type { Session } from "neo4j-driver";
export interface GraphRelationship {
  type: string;
  direction: "outgoing" | "incoming";
  targetNodeId: string;
}

export interface CreatedBy {
  /** Agent slug (e.g. "maxy-admin", "whatsapp-public") for LLM-tool writes. */
  agent?: string;
  /** Session correlation id — same string across every node written in a conversation turn. */
  session?: string;
  /** MCP tool name ("memory-write", "contact-create", ...) for LLM-tool writes. */
  tool?: string;
  /** Subsystem name ("workflow-execute", "persist-tool-call", ...) for system writes. */
  source?: string;
}

export interface WriteNodeWithEdgesParams {
  session: Session;
  labels: string[];
  props: Record<string, unknown>;
  /** At least one relationship is required — zero-rel writes are rejected. */
  relationships: GraphRelationship[];
  createdBy: CreatedBy;
  /**
   * Override for the accountId gate. Production callers leave
   * this unset — the gate compares `props.accountId` against
   * `process.env.ACCOUNT_ID`, which the MCP server / UI server validates at
   * boot time. Tests pass an explicit value to avoid env coupling.
   * Reading from env var (instead of an on-disk dir enumeration) keeps
   * graph-write out of the `node:fs` import chain — the lib is bundled
   * transitively into the ESM payload via cloudflare-task-tracker.ts, and
   * any CJS-side fs require shims to a runtime-fatal `__require("fs")`
   * after esbuild's CJS-to-ESM conversion strips the `node:` prefix.
   */
  expectedAccountId?: string;
  /**
   * Optional embedder. When supplied and the node carries a label in
   * VECTOR_INDEXED_LABELS and `props.embedding` is absent/empty, the chokepoint
   * computes a bounded props embedding before CREATE. Injected (never imported)
   * so graph-write stays out of the node:fs / Ollama import chain. A failed
   * embed logs `[graph-write] warn reason=embed-failed` and the write proceeds —
   * a failed embed must never abort an otherwise-valid write.
   */
  embed?: (text: string) => Promise<number[]>;
}

export interface WriteNodeResult {
  nodeId: string;
  labels: string[];
  edgesCreated: number;
}

/** Stamp flattened provenance into node properties. */
export function stampCreatedBy(
  props: Record<string, unknown>,
  createdBy: CreatedBy
): Record<string, unknown> {
  return {
    ...props,
    createdByAgent: createdBy.agent ?? "unknown",
    createdBySession: createdBy.session ?? "unknown",
    createdByTool: createdBy.tool ?? null,
    createdBySource: createdBy.source ?? null,
  };
}

/**
 * Enforce the write doctrine: node + ≥1 edge, transactional, provenance stamped.
 * The label and relationship type strings are interpolated into Cypher — the
 * caller must constrain them via Zod/schema before calling. Backticks are
 * stripped to defeat the only Cypher escape char.
 */
export async function writeNodeWithEdges(
  params: WriteNodeWithEdgesParams
): Promise<WriteNodeResult> {
  const { session, labels, props, relationships, createdBy } = params;

  const agentLabel = createdBy.agent ?? createdBy.source ?? "unknown";
  const labelCsv = labels.join(",");

  // accountId floor. Every write whose author is not the bootstrap
  // system MUST stamp `props.accountId` with the value of `ACCOUNT_ID`,
  // which the spawning process (MCP server / UI server) validated at boot
  // against the on-disk `${DATA_ROOT}/accounts/<id>/account.json` set. Read-
  // side scoping silently hides leaks, so this gate is the live floor that
  // the deletion of `pruneAlienAccounts` left to the writer surface.
  // The on-disk enumeration happens once per process at boot (in the
  // platform/lib/account-enumeration consumers — the WhatsApp helper and
  // the boot-side `[graph-health] account-enumeration` log). The gate here
  // verifies match against that pre-validated identity, keeping graph-write
  // out of the `node:fs` import chain so the ESM payload bundle does not
  // emit `__require("fs")` shims.
  // Doctrine: `.docs/neo4j.md` "Account isolation invariant".
  const isSystemBootstrap = (createdBy.agent ?? "") === "system";
  if (!isSystemBootstrap) {
    const accountId = props.accountId;
    const expectedAccountId = params.expectedAccountId ?? process.env.ACCOUNT_ID;
    if (
      typeof accountId !== "string" ||
      !expectedAccountId ||
      accountId !== expectedAccountId
    ) {
      const slice =
        typeof accountId === "string" ? accountId.slice(0, 8) : "missing";
      process.stderr.write(
        `[graph-write] reject reason=invalid-account-id accountId=${slice} writer=${agentLabel}\n`,
      );
      throw new Error(
        `Write doctrine violated: invalid-account-id (${slice}) — accountId must equal ACCOUNT_ID set by the spawning process. See .docs/neo4j.md "Account isolation invariant".`,
      );
    }
  }

  // review-detector / review-digest feature is removed entirely.
  // Reject any re-introduction via the doctrine surface. The actionTool check
  // is property-keyed (not label-keyed) so a future scheduled-Event writer
  // re-introducing the daily digest by tool name is caught here even though
  // `:Event` is otherwise a valid label. Fires before the zero-relationships
  // check because the dead-feature rejection is structurally more fundamental.
  const reviewDigestActionTool =
    typeof props.actionTool === "string" && props.actionTool === "review-digest-compose";
  if (labels.includes("ReviewAlert") || reviewDigestActionTool) {
    const actionToolField = reviewDigestActionTool
      ? "review-digest-compose"
      : "n/a";
    process.stderr.write(
      `[graph-write] reject reason=removed-feature labels=${labelCsv} actionTool=${actionToolField} agent=${agentLabel}\n`
    );
    throw new Error(
      "Write doctrine violated: review-detector feature removed — `:ReviewAlert` and `:Event {actionTool:'review-digest-compose'}` writes are not allowed."
    );
  }

  if (!relationships || relationships.length < 1) {
    process.stderr.write(
      `[graph-write] reject reason=zero-relationships labels=${labelCsv} agent=${agentLabel}\n`
    );
    throw new Error(
      "Write doctrine violated: a node must be created with at least one relationship. See .docs/neo4j.md (Write doctrine)."
    );
  }

  // Embed safety net. When the node carries a vector-indexed label, lacks an
  // embedding, and an embedder is injected, compute a bounded props embedding
  // so no indexed node enters the graph invisible to vector search. A failed
  // embed is logged and the write proceeds — never aborted (Task 636).
  const indexed = labelsAreIndexed(labels);
  let embedded =
    Array.isArray(props.embedding) && (props.embedding as unknown[]).length > 0;
  const workingProps: Record<string, unknown> = { ...props };
  if (indexed && !embedded && typeof params.embed === "function") {
    try {
      const vec = await params.embed(boundedEmbedText(labels, props));
      if (Array.isArray(vec) && vec.length > 0) {
        workingProps.embedding = vec;
        embedded = true;
      }
    } catch (err) {
      process.stderr.write(
        `[graph-write] warn reason=embed-failed labels=${labelCsv} agent=${agentLabel} detail=${err instanceof Error ? err.message : String(err)}\n`,
      );
    }
  }

  const labelStr = labels.map((l) => `\`${l.replace(/`/g, "")}\``).join(":");
  const nodeProps = stampCreatedBy(workingProps, createdBy);

  return await session.executeWrite(async (tx) => {
    const targetIds = relationships.map((r) => r.targetNodeId);
    const check = await tx.run(
      `UNWIND $ids AS id MATCH (t) WHERE elementId(t) = id RETURN elementId(t) AS id, labels(t) AS labels`,
      { ids: targetIds },
    );

    const labelsByTarget = new Map<string, string[]>();
    for (const rec of check.records) {
      labelsByTarget.set(rec.get("id") as string, rec.get("labels") as string[]);
    }

    const found = labelsByTarget.size;
    const uniqueRequested = new Set(targetIds).size;
    if (found !== uniqueRequested) {
      process.stderr.write(
        `[graph-write] reject reason=unresolved-target labels=${labelCsv} agent=${agentLabel} requested=${uniqueRequested} found=${found}\n`
      );
      throw new Error(
        `Write doctrine violated: ${uniqueRequested - found} of ${uniqueRequested} relationship target(s) did not resolve (elementId mismatch). No node created.`
      );
    }

    // Process provenance gate (widened). Labels in
    // ACTION_PROVENANCE_LABELS require an inbound :PRODUCED edge from a
    // node carrying one of PROVENANCE_SOURCE_LABELS (:Task, :Conversation,
    // :Message — subtype labels qualify via the labels() array). This
    // makes every durable LLM-tool entity write traversable to either the
    // Task that produced it (autonomous workflows) or the
    // Conversation/Message turn that triggered it (direct admin asks).
    // Bootstrap writes (createdBy.agent === 'system') are exempt —
    // installer paths run before any provenance node exists. The check
    // runs INSIDE the transaction so the labels-by-target map is
    // consistent with target resolution.
    let provenanceSourceId: string | null = null;
    let provenanceSourceLabel: string | null = null;
    if (
      requiresActionProvenance(labels) &&
      (createdBy.agent ?? "") !== "system"
    ) {
      const candidates = findProvenanceCandidates(relationships);
      const matched = candidates
        .map((r) => {
          const lbls = labelsByTarget.get(r.targetNodeId);
          if (!Array.isArray(lbls)) return null;
          const sourceLabel = lbls.find((l) => PROVENANCE_SOURCE_LABELS.has(l));
          return sourceLabel ? { rel: r, sourceLabel } : null;
        })
        .filter((m): m is { rel: GraphRelationship; sourceLabel: string } => m !== null);
      if (matched.length === 0) {
        // Task 580 relaxed this from a hard reject to a soft warn. The
        // composer-spawned admin path (PRD §6.4) inherits the per-account
        // claude-rc daemon's bare env and never receives SESSION_NODE_ID,
        // so the throw was failing every direct admin contact-create /
        // memory-write for a gated label. Operators retain visibility via
        // this single structured warn line; the audit-stage
        // missing-provenance-warning at audit.ts continues to fire
        // independently for idempotent MERGEs that didn't stamp.
        process.stderr.write(
          `[graph-write] warn reason=missing-provenance labels=${labelCsv} agent=${agentLabel}\n`
        );
      } else {
        provenanceSourceId = matched[0].rel.targetNodeId;
        provenanceSourceLabel = matched[0].sourceLabel;
      }
    }

    let nodeRes;
    try {
      nodeRes = await tx.run(
        `CREATE (n:${labelStr} $props) RETURN elementId(n) AS nodeId, labels(n) AS nodeLabels`,
        { props: nodeProps }
      );
    } catch (err) {
      // surface UserProfile uniqueness violations as a structured
      // reject line. Neo4j's ConstraintValidationFailed message names the
      // violated label + properties (e.g. "Node(N) already exists with label
      // `UserProfile` and properties accountId=… userId=…") but NOT the
      // constraint name, so we key on the helper's input `labels` parameter
      // rather than parsing the message. Other constraint violations
      // propagate without specialised logging — they are unexpected and the
      // raw error is the right signal.
      const code = (err as { code?: string } | null)?.code ?? "";
      if (code === "Neo.ClientError.Schema.ConstraintValidationFailed" && labels.includes("UserProfile")) {
        const accountIdProp = (nodeProps as Record<string, unknown>).accountId;
        const userIdProp = (nodeProps as Record<string, unknown>).userId;
        const acctSlice = typeof accountIdProp === "string" ? accountIdProp.slice(0, 8) : "unknown";
        const userSlice = typeof userIdProp === "string" ? userIdProp.slice(0, 8) : "unknown";
        process.stderr.write(
          `[graph-write] reject reason=user-profile-uniqueness-violation accountId=${acctSlice} userId=${userSlice} writer=${agentLabel}\n`
        );
      }
      throw err;
    }
    const nodeId = nodeRes.records[0].get("nodeId") as string;
    const nodeLabels = nodeRes.records[0].get("nodeLabels") as string[];

    // Neo4j Community's default isolation is read-committed (not snapshot)
    // — a target elementId that resolved during the pre-check can be
    // deleted by a concurrent transaction before the per-edge CREATE below
    // runs. If that happens, `MATCH (a), (b)` returns 0 rows and the CREATE
    // quietly produces no edge. Per-edge counter inspection catches the
    // race: any zero-result CREATE throws inside the transaction callback,
    // so `executeWrite` rolls the node back — no silent-orphan path.
    let edgesCreated = 0;
    for (const rel of relationships) {
      const type = rel.type.replace(/`/g, "");
      const q =
        rel.direction === "outgoing"
          ? `MATCH (a), (b) WHERE elementId(a) = $from AND elementId(b) = $to CREATE (a)-[:\`${type}\`]->(b)`
          : `MATCH (a), (b) WHERE elementId(a) = $from AND elementId(b) = $to CREATE (b)-[:\`${type}\`]->(a)`;
      const r = await tx.run(q, { from: nodeId, to: rel.targetNodeId });
      const created = r.summary.counters.updates().relationshipsCreated;
      if (created === 0) {
        process.stderr.write(
          `[graph-write] reject reason=unresolved-target-on-create labels=${labelCsv} agent=${agentLabel} relType=${rel.type} targetId=${rel.targetNodeId}\n`
        );
        throw new Error(
          `Write doctrine violated: relationship CREATE to target ${rel.targetNodeId} produced 0 edges (target likely deleted concurrently after pre-check). Transaction rolled back.`
        );
      }
      edgesCreated += created;
    }

    if (edgesCreated !== relationships.length) {
      // Defensive: should be unreachable given the per-edge check above.
      // The rollback throws loudly so regression shows rather than a
      // silent commit with edgesCreated < requested.
      process.stderr.write(
        `[graph-write] reject reason=edge-count-mismatch labels=${labelCsv} agent=${agentLabel} requested=${relationships.length} created=${edgesCreated}\n`
      );
      throw new Error(
        `Write doctrine violated: expected ${relationships.length} edges, created ${edgesCreated}. Transaction rolled back.`
      );
    }

    process.stderr.write(
      `[graph-write] accepted labels=${labelCsv} edges=${edgesCreated} createdByAgent=${createdBy.agent ?? "unknown"} createdByTool=${createdBy.tool ?? createdBy.source ?? "unknown"} producedBy=${provenanceSourceLabel ?? "none"}:${provenanceSourceId ?? "none"}\n`
    );

    // Per-write embed/stamp signal (Task 636). `embedded=false` on an
    // `indexed=true` line is the per-write null-embedding signature the
    // standing [embed-audit] reconciles against.
    const acctSlice =
      typeof props.accountId === "string" ? props.accountId.slice(0, 8) : "none";
    process.stderr.write(
      `[graph-write] op=node-written label=${labelCsv} accountId=${acctSlice} embedded=${embedded} indexed=${indexed}\n`
    );

    return { nodeId, labels: nodeLabels, edgesCreated };
  });
}