/**
 * Soft-delete primitive for the knowledge graph.
 *
 * MCP-tool callers must NEVER `DETACH DELETE` user-domain nodes. Use
 * `trashNode` to mark a node `:Trashed`; `restoreNode` removes the label;
 * `emptyTrash` hard-deletes nodes whose `trashedAt < now - graceDays`. The
 * 2026-04-20 incident wiped 19 nodes via a single autonomous `DETACH DELETE`
 * — Neo4j Community has no PITR, so properties were unrecoverable. This
 * primitive contains the blast radius for every caller.
 *
 * Unique-constraint handling: when the trashed node's labels carry single-
 * key UNIQUE constraints (e.g. `LocalBusiness.accountId`), the live values
 * are snapshotted into `_trashedKeys` (JSON) and nulled on the node, so
 * MERGE against the same key won't collide. `restoreNode` writes them back
 * and fails loudly when an active node already occupies the slot.
 *
 * Label-aware cascade: when `trashNode` trashes a `:Conversation`, it also
 * trashes every `(m:Message)-[:PART_OF]->(c)` Message in the same managed
 * transaction. This makes `MATCH (m:Trashed)` a correct audit primitive —
 * trashing a Conversation without cascading to its Messages silently
 * under-reports the blast radius. `restoreNode` reverses both sides.
 */

import type { Session } from "neo4j-driver";

/**
 * Single-key UNIQUE properties per label, mirroring `platform/neo4j/schema.cypher`.
 *
 * Composite UNIQUEs (UserProfile, Email composite, AccessGrant) are
 * represented by a single nullable component — Neo4j enforces a composite
 * constraint only when every component is non-null, so nulling one frees it.
 */
const UNIQUE_KEYS_BY_LABEL: Record<string, string[]> = {
  Person: ["email", "telephone"],
  Service: ["serviceId"],
  LocalBusiness: ["accountId"],
  Task: ["taskId"],
  Event: ["eventId"],
  // :KnowledgeDocument is keyed on `attachmentId` (document writer path —
  // file/URL/obsidian/contract attachment + legacy Gmail-thread shape under
  // `source:'email'`). Task 397 split the conversation transcripts off into
  // `:ConversationArchive`, which carries `conversationIdentity`.
  KnowledgeDocument: ["attachmentId"],
  ConversationArchive: ["conversationIdentity"],
  DigitalDocument: ["attachmentId"],
  Conversation: ["sessionId"],
  Message: ["messageId"],
  Workflow: ["workflowId"],
  WorkflowStep: ["stepId"],
  WorkflowRun: ["runId"],
  Preference: ["preferenceId"],
  Email: ["emailId", "messageId"],
  AdminUser: ["userId"],
  ToolCall: ["callId"],
  // Composite component nulls — frees the composite constraint:
  AccessGrant: ["contactValue"],     // composite (contactValue, agentSlug, accountId)
  UserProfile: ["userId"],           // composite (accountId, userId)
};

const TRASH_PROP_NAMES = [
  "trashedAt",
  "trashedBy",
  "trashReason",
  "_trashedKeys",
] as const;

export interface TrashParams {
  session: Session;
  accountId: string;
  elementId: string;
  /** Provenance marker — appears verbatim in `trashedBy` and `[trash:marked] by=`. */
  by: string;
  reason?: string;
}

export interface TrashResult {
  trashed: boolean;
  alreadyTrashed: boolean;
  nodeId: string;
  /** Labels excluding `:Trashed`. */
  labels: string[];
  trashedAt: string;
  originalKeys: Record<string, unknown>;
}

export async function trashNode(params: TrashParams): Promise<TrashResult> {
  const { session, accountId, elementId, by, reason } = params;

  const lookup = await session.run(
    `MATCH (n) WHERE elementId(n) = $eid AND n.accountId = $accountId
     RETURN labels(n) AS labels, properties(n) AS props`,
    { eid: elementId, accountId },
  );
  if (lookup.records.length === 0) {
    throw new Error(
      `trashNode: node not found (elementId=${elementId} accountId=${accountId.slice(0, 8)}…)`,
    );
  }

  const allLabels = lookup.records[0].get("labels") as string[];
  const props = lookup.records[0].get("props") as Record<string, unknown>;
  const baseLabels = allLabels.filter((l) => l !== "Trashed");

  if (allLabels.includes("Trashed")) {
    return {
      trashed: false,
      alreadyTrashed: true,
      nodeId: elementId,
      labels: baseLabels,
      trashedAt: String(props.trashedAt ?? ""),
      originalKeys: {},
    };
  }

  const uniqueKeys = new Set<string>();
  for (const label of baseLabels) {
    for (const key of UNIQUE_KEYS_BY_LABEL[label] ?? []) uniqueKeys.add(key);
  }

  const originalKeys: Record<string, unknown> = {};
  for (const k of uniqueKeys) {
    if (props[k] !== undefined && props[k] !== null) originalKeys[k] = props[k];
  }

  const setNullClauses = Object.keys(originalKeys)
    .map((k) => `n.\`${k}\` = null`)
    .join(", ");
  const setNullSuffix = setNullClauses ? `, ${setNullClauses}` : "";

  const trashedAt = new Date().toISOString();
  const isConversation = baseLabels.includes("Conversation");
  // Message unique-key list computed once — matches UNIQUE_KEYS_BY_LABEL["Message"].
  const messageUniqueKeys = UNIQUE_KEYS_BY_LABEL["Message"] ?? [];

  // Root trash + Conversation→Message cascade in one managed transaction so a
  // DB error mid-sweep can't leave a Conversation :Trashed with some Messages
  // un-trashed (the 2026-04-22 audit correctness gap). For non-Conversation
  // labels, the cascade branch is a no-op — same one-write shape as before.
  let cascadedMessageCount = 0;
  await session.executeWrite(async (tx) => {
    await tx.run(
      `MATCH (n) WHERE elementId(n) = $eid
       SET n:Trashed,
           n.trashedAt = datetime($trashedAt),
           n.trashedBy = $by,
           n.trashReason = $reason,
           n._trashedKeys = $trashedKeysJson${setNullSuffix}`,
      {
        eid: elementId,
        trashedAt,
        by,
        reason: reason ?? null,
        trashedKeysJson: JSON.stringify(originalKeys),
      },
    );

    if (isConversation) {
      // Two-step cascade: collect Message elementIds and their unique-key
      // values first, then apply :Trashed + snapshot + null in one SET per
      // message using parameter-passed snapshot JSON. This keeps JSON
      // construction in the Node process (where JSON.stringify is trivially
      // correct) rather than building escaped JSON inside Cypher.
      const collectKeys = messageUniqueKeys.length > 0
        ? messageUniqueKeys.map((k) => `\`${k}\`: m.\`${k}\``).join(", ")
        : "";
      const collectMsgProps = collectKeys ? `, {${collectKeys}}` : ", {}";

      const collected = await tx.run(
        `MATCH (c) WHERE elementId(c) = $eid
         MATCH (m:Message)-[:PART_OF]->(c)
         WHERE NOT m:Trashed
         RETURN elementId(m) AS meid${collectMsgProps} AS keys`,
        { eid: elementId },
      );

      for (const rec of collected.records) {
        const meid = rec.get("meid") as string;
        const keys = rec.get("keys") as Record<string, unknown>;
        const liveKeys: Record<string, unknown> = {};
        for (const k of messageUniqueKeys) {
          if (keys[k] !== undefined && keys[k] !== null) liveKeys[k] = keys[k];
        }
        const msgSetNulls = Object.keys(liveKeys).length > 0
          ? ", " + Object.keys(liveKeys).map((k) => `m.\`${k}\` = null`).join(", ")
          : "";
        await tx.run(
          `MATCH (m) WHERE elementId(m) = $meid
           SET m:Trashed,
               m.trashedAt = datetime($trashedAt),
               m.trashedBy = $by,
               m.trashReason = $reason,
               m._trashedKeys = $trashedKeysJson${msgSetNulls}`,
          {
            meid,
            trashedAt,
            by: `${by}:cascade-from-conversation`,
            reason: reason ?? `cascade from Conversation ${elementId}`,
            trashedKeysJson: JSON.stringify(liveKeys),
          },
        );
      }
      cascadedMessageCount = collected.records.length;
    }
  });

  process.stderr.write(
    `[trash:marked] accountId=${accountId} elementId=${elementId} labels=${baseLabels.join(",")} by=${by} reason=${reason ?? "null"}\n`,
  );
  if (isConversation) {
    process.stderr.write(
      `[trash:cascaded] accountId=${accountId} conversationElementId=${elementId} messageCount=${cascadedMessageCount} by=${by}\n`,
    );
  }

  return {
    trashed: true,
    alreadyTrashed: false,
    nodeId: elementId,
    labels: baseLabels,
    trashedAt,
    originalKeys,
  };
}

export interface RestoreParams {
  session: Session;
  accountId: string;
  elementId: string;
}

export interface RestoreResult {
  restored: boolean;
  nodeId: string;
  labels: string[];
  restoredKeys: Record<string, unknown>;
}

export async function restoreNode(params: RestoreParams): Promise<RestoreResult> {
  const { session, accountId, elementId } = params;

  const lookup = await session.run(
    `MATCH (n:Trashed) WHERE elementId(n) = $eid
     RETURN labels(n) AS labels, n._trashedKeys AS keysJson`,
    { eid: elementId },
  );
  if (lookup.records.length === 0) {
    throw new Error(
      `restoreNode: trashed node not found (elementId=${elementId})`,
    );
  }

  const allLabels = lookup.records[0].get("labels") as string[];
  const baseLabels = allLabels.filter((l) => l !== "Trashed");
  const keysJson = lookup.records[0].get("keysJson") as string | null;
  const originalKeys: Record<string, unknown> = keysJson ? JSON.parse(keysJson) : {};

  // Conflict check: an active node already holds the unique slot we want back?
  for (const label of baseLabels) {
    const uniqueKeys = UNIQUE_KEYS_BY_LABEL[label] ?? [];
    for (const k of uniqueKeys) {
      const v = originalKeys[k];
      if (v === undefined || v === null) continue;
      const conflict = await session.run(
        `MATCH (other:\`${label}\`)
         WHERE elementId(other) <> $eid
           AND NOT other:Trashed
           AND other.\`${k}\` = $val
         RETURN elementId(other) AS otherId LIMIT 1`,
        { eid: elementId, val: v },
      );
      if (conflict.records.length > 0) {
        const otherId = conflict.records[0].get("otherId") as string;
        throw new Error(
          `restoreNode: cannot restore ${label} elementId=${elementId} — active node elementId=${otherId} already holds ${k}=${JSON.stringify(v)}`,
        );
      }
    }
  }

  const setClauses = Object.keys(originalKeys)
    .map((k) => `n.\`${k}\` = $val_${k}`)
    .join(", ");
  const setSuffix = setClauses ? `, ${setClauses}` : "";
  const setParams: Record<string, unknown> = { eid: elementId };
  for (const [k, v] of Object.entries(originalKeys)) setParams[`val_${k}`] = v;

  const isConversation = baseLabels.includes("Conversation");
  let cascadedMessageCount = 0;

  await session.executeWrite(async (tx) => {
    await tx.run(
      `MATCH (n:Trashed) WHERE elementId(n) = $eid
       REMOVE n:Trashed, n.trashedAt, n.trashedBy, n.trashReason, n._trashedKeys
       SET n.restoredAt = datetime()${setSuffix}`,
      setParams,
    );

    if (isConversation) {
      // Reverse of the trashNode cascade: re-label and restore unique-key
      // snapshots for every `(m:Message)-[:PART_OF]->(c)` that was trashed
      // by the same cascade chain (i.e. m._trashedKeys exists + m was
      // trashed by a `*:cascade-from-conversation` provenance). We scope by
      // the second condition so a user-initiated single-Message trash that
      // happens to sit inside this Conversation isn't accidentally undone.
      const collected = await tx.run(
        `MATCH (c) WHERE elementId(c) = $eid
         MATCH (m:Trashed:Message)-[:PART_OF]->(c)
         WHERE m.trashedBy ENDS WITH ':cascade-from-conversation'
         RETURN elementId(m) AS meid, m._trashedKeys AS keysJson`,
        { eid: elementId },
      );
      for (const rec of collected.records) {
        const meid = rec.get("meid") as string;
        const keysJson = rec.get("keysJson") as string | null;
        const keys: Record<string, unknown> = keysJson ? JSON.parse(keysJson) : {};
        const setClause = Object.keys(keys).length > 0
          ? ", " + Object.keys(keys).map((k) => `m.\`${k}\` = $val_${k}`).join(", ")
          : "";
        const msgParams: Record<string, unknown> = { meid };
        for (const [k, v] of Object.entries(keys)) msgParams[`val_${k}`] = v;
        await tx.run(
          `MATCH (m) WHERE elementId(m) = $meid
           REMOVE m:Trashed, m.trashedAt, m.trashedBy, m.trashReason, m._trashedKeys
           SET m.restoredAt = datetime()${setClause}`,
          msgParams,
        );
      }
      cascadedMessageCount = collected.records.length;
    }
  });

  process.stderr.write(
    `[trash:restored] accountId=${accountId} elementId=${elementId} labels=${baseLabels.join(",")}\n`,
  );
  if (isConversation) {
    process.stderr.write(
      `[trash:cascade-restored] accountId=${accountId} conversationElementId=${elementId} messageCount=${cascadedMessageCount}\n`,
    );
  }

  return {
    restored: true,
    nodeId: elementId,
    labels: baseLabels,
    restoredKeys: originalKeys,
  };
}

export interface EmptyTrashParams {
  session: Session;
  accountId: string;
  /** Default 30. */
  graceDays?: number;
  dryRun?: boolean;
  /** Optional label whitelist — only nodes carrying any of these labels are eligible. */
  labels?: string[];
  /** Side-effect callback invoked before the `DETACH DELETE` for each
   *  emptied node, e.g. to clean disk attachments or purge sibling JSONL
   *  transcripts.
   *
   *  Contract (Task 237): if the callback **throws**, the candidate is
   *  **skipped this run** — no `DETACH DELETE` runs and no `[trash:emptied]`
   *  line emits. The thrown message lands in
   *  `[trash:empty-run] reason=onEmpty-failed` and the candidate becomes
   *  eligible again on the next sweep. Pre-Task-237 consumers
   *  (`removeAttachmentDir` for `:KnowledgeDocument`) cannot throw because
   *  they wrap `fs.rm({force:true})`, so the contract change is invisible
   *  to them. */
  onEmpty?: (node: TrashCandidate) => Promise<void>;
}

export interface TrashCandidate {
  elementId: string;
  labels: string[];
  trashedAt: string;
  trashedBy: string | null;
  /** Original unique-key snapshot — useful for KnowledgeDocument disk cleanup. */
  trashedKeys: Record<string, unknown>;
}

export interface EmptyTrashResult {
  graceDays: number;
  dryRun: boolean;
  candidates: TrashCandidate[];
  emptied: number;
}

export async function emptyTrash(params: EmptyTrashParams): Promise<EmptyTrashResult> {
  const t0 = Date.now();
  const { session, accountId, graceDays = 30, dryRun = false, labels: labelFilter, onEmpty } = params;

  const cutoff = new Date(Date.now() - graceDays * 24 * 60 * 60 * 1000).toISOString();

  const labelClause = labelFilter && labelFilter.length > 0
    ? `AND ANY(l IN labels(n) WHERE l IN $labelFilter)`
    : "";

  const candidatesResult = await session.run(
    `MATCH (n:Trashed)
     WHERE n.trashedAt < datetime($cutoff)
     ${labelClause}
     RETURN elementId(n) AS eid,
            labels(n) AS labels,
            toString(n.trashedAt) AS trashedAt,
            n.trashedBy AS trashedBy,
            n._trashedKeys AS keysJson
     ORDER BY n.trashedAt ASC`,
    { cutoff, ...(labelFilter ? { labelFilter } : {}) },
  );

  const candidates: TrashCandidate[] = candidatesResult.records.map((r) => {
    const keysJson = r.get("keysJson") as string | null;
    return {
      elementId: r.get("eid") as string,
      labels: (r.get("labels") as string[]).filter((l) => l !== "Trashed"),
      trashedAt: String(r.get("trashedAt")),
      trashedBy: (r.get("trashedBy") as string | null) ?? null,
      trashedKeys: keysJson ? JSON.parse(keysJson) : {},
    };
  });

  process.stderr.write(
    `[trash:empty-run] accountId=${accountId} graceDays=${graceDays} dryRun=${dryRun} candidates=${candidates.length}\n`,
  );

  if (dryRun || candidates.length === 0) {
    process.stderr.write(
      `[graph:trash:summary] accountId=${accountId} trashedNow=0 emptiedThisRun=0 graceDays=${graceDays} dryRun=${dryRun} elapsedMs=${Date.now() - t0}\n`,
    );
    return { graceDays, dryRun, candidates, emptied: 0 };
  }

  let emptied = 0;
  for (const c of candidates) {
    if (onEmpty) {
      try {
        await onEmpty(c);
      } catch (err) {
        // Task 237 — skip the DETACH DELETE so the candidate stays
        // eligible for the next sweep. Surfaces alongside the existing
        // [trash:empty-run] family so an operator's grep catches both
        // candidate listings and skip reasons in one window.
        const id8 = c.elementId.slice(0, 8);
        const message = err instanceof Error ? err.message : String(err);
        process.stderr.write(
          `[trash:empty-run] reason=onEmpty-failed elementId=${id8} labels=${c.labels.join(",")} message=${message}\n`,
        );
        continue;
      }
    }

    await session.run(
      `MATCH (n) WHERE elementId(n) = $eid DETACH DELETE n`,
      { eid: c.elementId },
    );

    const ageDays = Math.floor((Date.now() - new Date(c.trashedAt).getTime()) / (24 * 60 * 60 * 1000));
    process.stderr.write(
      `[trash:emptied] accountId=${accountId} elementId=${c.elementId} labels=${c.labels.join(",")} trashedAt=${c.trashedAt} ageDays=${ageDays}\n`,
    );
    emptied++;
  }

  process.stderr.write(
    `[graph:trash:summary] accountId=${accountId} trashedNow=0 emptiedThisRun=${emptied} graceDays=${graceDays} dryRun=${dryRun} elapsedMs=${Date.now() - t0}\n`,
  );

  return { graceDays, dryRun, candidates, emptied };
}

/**
 * Read-filter clause excluding trashed nodes for the given Cypher alias.
 *
 * Filters both the `:Trashed` label (primitive) and the `deletedAt`
 * property (KnowledgeDocument soft-delete). Belt-and-braces: a row marked
 * by either signal is excluded from reads.
 *
 *   notTrashed("node")    → "(NOT node:Trashed AND node.deletedAt IS NULL)"
 *   notTrashed("related") → "(NOT related:Trashed AND related.deletedAt IS NULL)"
 */
export function notTrashed(alias: string): string {
  return `(NOT \`${alias}\`:Trashed AND \`${alias}\`.deletedAt IS NULL)`;
}

/** Runtime accessor for the static unique-keys map (for tests + sibling tooling). */
export function uniqueKeysForLabels(labels: string[]): string[] {
  const out = new Set<string>();
  for (const l of labels) for (const k of UNIQUE_KEYS_BY_LABEL[l] ?? []) out.add(k);
  return [...out];
}

/** Property names trashNode writes — exposed so callers can re-MERGE without colliding. */
export const TRASH_METADATA_PROPS: readonly string[] = TRASH_PROP_NAMES;