// Bounded-cascade graph engine for `wmill pipeline run --to`.
//
// MIRROR of frontend/src/lib/components/assets/AssetGraph/boundedCascade.ts —
// the two have no shared import path (frontend is a separate package), so keep
// them in sync. The grammar is intentionally tiny: there is no dbt-style
// `--select` string. The user names a start (a schedule / manual root) and one
// or more end nodes; the run is the "path between" them:
//
//     descendants(start) ∩ (ancestors(ends) ∪ ends) ∪ {start}
//
// Node ids: assets `${kind}:${path}` (e.g. `datatable:main/raw`); runnables
// `script:${path}`. Operates on the asset-graph payload shape used by
// pipeline.ts.

export type BCGraph = {
  runnables: {
    path: string;
    usage_kind: "script" | "flow" | "job";
    // `// partitioned <kind>` on the script (daily/hourly/weekly/monthly/dynamic).
    // Emitted by both the deployed graph endpoint and the local builder.
    partition_kind?: string;
    // `// macros` library: non-empty on a workspace macro library (deployed
    // graph only). Definition-only — never a runnable cascade step.
    macros?: { name: string }[];
  }[];
  assets: { kind: string; path: string }[];
  edges: {
    runnable_kind: string;
    runnable_path: string;
    asset_kind: string;
    asset_path: string;
    access_type?: "r" | "w" | "rw";
  }[];
  triggers: (
    | {
        trigger_kind: "asset";
        asset_kind: string;
        asset_path: string;
        runnable_kind: string;
        runnable_path: string;
      }
    | { trigger_kind: string; runnable_kind: string; runnable_path: string }
  )[];
  // `// data_test` ordering edges (HD-1): the referenced asset's producer must
  // materialize before the tested script runs. Fed into the lineage DAG so a
  // bounded/cold cascade orders the referenced dimension first. Optional — the
  // deployed graph omits it when empty, and older local graphs never emit it.
  test_edges?: {
    producer_kind: string;
    producer_path: string;
    runnable_kind: string;
    runnable_path: string;
    asset_kind: string;
    asset_path: string;
  }[];
};

export const SCRIPT_PREFIX = "script:";
export const scriptNodeId = (path: string): string => `${SCRIPT_PREFIX}${path}`;
export const isScriptNode = (id: string): boolean => id.startsWith(SCRIPT_PREFIX);
export const scriptPathOf = (id: string): string => id.slice(SCRIPT_PREFIX.length);
const assetNodeId = (kind: string, path: string): string => `${kind}:${path}`;

// Native trigger kinds whose scripts must NOT be auto-run by the CLI cascade:
// event triggers fan out per external event, and `webhook`/`data_upload` are UI
// entrypoints that need caller-supplied input (a request body / an uploaded
// S3Object) — previewing any of them with empty args runs the wrong thing.
// The deployed graph omits `webhook`/`data_upload` rows, but the local graph
// emits them, so a `// on data_upload` script would otherwise read as a manual
// root and get auto-run without its upload argument.
const NON_AUTORUN_TRIGGER_KINDS = new Set([
  "kafka",
  "mqtt",
  "amqp",
  "nats",
  "postgres",
  "sqs",
  "gcp",
  "email",
  "webhook",
  "data_upload",
]);

/** Resolve an asset URI (`datatable://x`, `s3://b/k`, …) to its node id. */
export function assetUriToNodeId(uri: string): string | undefined {
  const m = uri.match(/^([a-z0-9_]+):\/\/(.+)$/i);
  if (!m) return undefined;
  const prefix = m[1].toLowerCase();
  const kind = prefix === "s3" ? "s3object" : prefix;
  // The suffix is kept verbatim (mirrors Rust `parse_asset_syntax`): an S3
  // path encodes the storage, with a leading `/` for the workspace default
  // (`s3:///key` → `/key`) vs `s3://secondary/key` → `secondary/key`.
  return `${kind}:${m[2]}`;
}

export type LineageDag = {
  down: Map<string, Set<string>>;
  up: Map<string, Set<string>>;
  nodes: Set<string>;
};

function addEdge(dag: LineageDag, a: string, b: string) {
  if (a === b) return;
  dag.nodes.add(a);
  dag.nodes.add(b);
  (dag.down.get(a) ?? dag.down.set(a, new Set()).get(a)!).add(b);
  (dag.up.get(b) ?? dag.up.set(b, new Set()).get(b)!).add(a);
}

/** Unified upstream→downstream lineage DAG over scripts ∪ assets. */
export function buildLineageDag(g: BCGraph): LineageDag {
  const dag: LineageDag = { down: new Map(), up: new Map(), nodes: new Set() };
  for (const r of g.runnables ?? []) {
    if (r.usage_kind === "script") dag.nodes.add(scriptNodeId(r.path));
  }
  for (const a of g.assets ?? []) dag.nodes.add(assetNodeId(a.kind, a.path));
  for (const e of g.edges ?? []) {
    if (e.runnable_kind !== "script") continue;
    const aid = assetNodeId(e.asset_kind, e.asset_path);
    const access = e.access_type ?? "r";
    if (access === "w" || access === "rw") {
      addEdge(dag, scriptNodeId(e.runnable_path), aid); // producer
    } else if (access === "r") {
      addEdge(dag, aid, scriptNodeId(e.runnable_path)); // pure reader
    }
  }
  for (const t of g.triggers ?? []) {
    if (t.trigger_kind !== "asset" || t.runnable_kind !== "script") continue;
    const at = t as Extract<BCGraph["triggers"][number], { trigger_kind: "asset" }>;
    addEdge(dag, assetNodeId(at.asset_kind, at.asset_path), scriptNodeId(at.runnable_path));
  }
  // Data-test ordering edges: route through the referenced asset node so the
  // existing producer → asset write edge extends into producer → asset → testing
  // script (mirrors frontend boundedCascade.ts) — the tested script runs after
  // the referenced dimension materializes.
  for (const t of g.test_edges ?? []) {
    if (t.runnable_kind !== "script") continue;
    addEdge(dag, assetNodeId(t.asset_kind, t.asset_path), scriptNodeId(t.runnable_path));
  }
  return dag;
}

function closure(adj: Map<string, Set<string>>, start: string): Set<string> {
  const seen = new Set<string>();
  const queue = [start];
  while (queue.length > 0) {
    const cur = queue.shift()!;
    for (const n of adj.get(cur) ?? []) {
      if (seen.has(n)) continue;
      seen.add(n);
      queue.push(n);
    }
  }
  // A cycle back to `start` would have re-added it; the contract excludes
  // the node itself.
  seen.delete(start);
  return seen;
}

export const descendants = (dag: LineageDag, n: string): Set<string> => closure(dag.down, n);
export const ancestors = (dag: LineageDag, n: string): Set<string> => closure(dag.up, n);

/**
 * Nodes reachable from `starts` over the lineage DAG, treating `barriers` as cut
 * points: a barrier node is neither included NOR traversed through. So a node
 * reachable ONLY via a barrier is excluded, while one also reachable via another
 * path stays. Used for whole-pipeline runs to keep event handlers AND their
 * event-only downstream closure out (subtracting only the handler would leave a
 * consumer whose producer was skipped, which topoOrder would then run with
 * missing/stale inputs).
 */
export function reachableCutting(
  dag: LineageDag,
  starts: Iterable<string>,
  barriers: Set<string>,
): Set<string> {
  const seen = new Set<string>();
  const queue: string[] = [];
  for (const s of starts) {
    if (barriers.has(s) || seen.has(s)) continue;
    seen.add(s);
    queue.push(s);
  }
  while (queue.length > 0) {
    const n = queue.shift()!;
    for (const next of dag.down.get(n) ?? []) {
      if (barriers.has(next) || seen.has(next)) continue;
      seen.add(next);
      queue.push(next);
    }
  }
  return seen;
}

export type BoundedResult = {
  nodes: Set<string>;
  reachableEnds: string[];
  droppedEnds: string[];
};

/** Path-between node set for `start` and `ends` (inclusive). */
export function boundedSet(dag: LineageDag, start: string, ends: string[]): BoundedResult {
  const downSet = new Set(descendants(dag, start));
  downSet.add(start);
  const reachableEnds = ends.filter((e) => downSet.has(e));
  const droppedEnds = ends.filter((e) => !downSet.has(e));
  if (reachableEnds.length === 0) {
    return { nodes: new Set([start]), reachableEnds, droppedEnds };
  }
  const upClosure = new Set<string>();
  for (const e of reachableEnds) {
    upClosure.add(e);
    for (const a of ancestors(dag, e)) upClosure.add(a);
  }
  const nodes = new Set<string>();
  for (const n of downSet) if (upClosure.has(n)) nodes.add(n);
  nodes.add(start);
  return { nodes, reachableEnds, droppedEnds };
}

/** Script node ids eligible to start a bounded run. */
export function validStarts(g: BCGraph): Set<string> {
  const subscribers = new Set<string>();
  const scheduleScripts = new Set<string>();
  const nonAutorunScripts = new Set<string>();
  for (const t of g.triggers ?? []) {
    if (t.runnable_kind !== "script") continue;
    if (t.trigger_kind === "asset") subscribers.add(t.runnable_path);
    else if (t.trigger_kind === "schedule") scheduleScripts.add(t.runnable_path);
    else if (NON_AUTORUN_TRIGGER_KINDS.has(t.trigger_kind)) nonAutorunScripts.add(t.runnable_path);
  }
  const out = new Set<string>();
  for (const r of g.runnables ?? []) {
    if (r.usage_kind !== "script") continue;
    const p = r.path;
    if (scheduleScripts.has(p)) out.add(scriptNodeId(p));
    else if (!subscribers.has(p) && !nonAutorunScripts.has(p)) out.add(scriptNodeId(p));
  }
  return out;
}

/**
 * Script node ids eligible as an EXPLICIT bounded-run start from *anywhere* in
 * the DAG (dbt's `--select model+`): every script that can run with empty args —
 * i.e. all scripts except non-autorun-trigger ones (kafka/mqtt/…/webhook/
 * data_upload, which fan out per event or need caller-supplied input). Unlike
 * `validStarts` (schedule/manual roots only), this INCLUDES mid-DAG asset
 * subscribers and pure readers, so `--from <mid-model>` runs that node plus its
 * transitive downstream WITHOUT re-running upstream. Roots stay a subset of this
 * set. A non-autorun handler still qualifies once it's `--upload`-bound (handled
 * by the caller, which unions in the bound node ids).
 */
export function validFromStarts(g: BCGraph): Set<string> {
  const nonAutorun = nonAutorunTriggerScripts(g);
  // Seed with the schedule/manual roots: `validStarts` lets a schedule identity
  // win over a secondary non-autorun trigger (a `// on schedule` + `// on
  // data_upload` script IS a scheduled root), so a root must stay `--from`-
  // eligible even though it's also in `nonAutorunTriggerScripts`.
  const out = new Set<string>(validStarts(g));
  for (const r of g.runnables ?? []) {
    if (r.usage_kind !== "script") continue;
    const id = scriptNodeId(r.path);
    if (!nonAutorun.has(id)) out.add(id);
  }
  return out;
}

/**
 * Script node ids that carry a trigger requiring caller-supplied input or
 * per-event fanout (kafka/mqtt/nats/postgres/sqs/gcp/email/webhook/data_upload).
 * These can't be run with empty args, so a whole-pipeline run must exclude them
 * even when they're a lineage descendant of a valid start (not just a root).
 */
export function nonAutorunTriggerScripts(g: BCGraph): Set<string> {
  const out = new Set<string>();
  for (const t of g.triggers ?? []) {
    if (t.runnable_kind === "script" && NON_AUTORUN_TRIGGER_KINDS.has(t.trigger_kind)) {
      out.add(scriptNodeId(t.runnable_path));
    }
  }
  return out;
}

/** Project a node-id set to the script paths it contains. */
export function scriptsOf(nodes: Iterable<string>): string[] {
  const out: string[] = [];
  for (const id of nodes) if (isScriptNode(id)) out.push(scriptPathOf(id));
  return out;
}

/**
 * Resolve a CLI `--to` / `--from` token to a node id, or undefined if it
 * matches nothing. Asset URIs (`kind://path`) resolve to the asset node; a bare
 * token matches a runnable by exact path or by short (last-segment) name.
 */
export function resolveToken(g: BCGraph, token: string): string | undefined {
  if (token.includes("://")) {
    const id = assetUriToNodeId(token);
    return id && g.assets.some((a) => `${a.kind}:${a.path}` === id) ? id : undefined;
  }
  const scripts = (g.runnables ?? []).filter((r) => r.usage_kind === "script");
  const exact = scripts.find((r) => r.path === token);
  if (exact) return scriptNodeId(exact.path);
  const byShort = scripts.filter((r) => (r.path.split("/").pop() ?? r.path) === token);
  return byShort.length === 1 ? scriptNodeId(byShort[0].path) : undefined;
}

/**
 * Topological order of `scripts` over the in-set producer→subscriber edges
 * (assets collapsed). Scripts on a cycle are returned in `cyclic` and excluded
 * from `order`. Serial-run friendly: every script comes after its in-set
 * upstreams.
 */
export function topoOrder(
  g: BCGraph,
  scripts: Set<string>,
): { order: string[]; cyclic: string[] } {
  const dag = buildLineageDag(g);
  const down = new Map<string, Set<string>>();
  const indegree = new Map<string, number>();
  for (const s of scripts) indegree.set(s, 0);
  // One-hop (through a single asset) script→script edges, restricted to the set.
  for (const s of scripts) {
    const sid = scriptNodeId(s);
    const oneHop = new Set<string>();
    for (const asset of dag.down.get(sid) ?? []) {
      for (const sub of dag.down.get(asset) ?? []) {
        if (isScriptNode(sub)) {
          const p = scriptPathOf(sub);
          if (p !== s && scripts.has(p)) oneHop.add(p);
        }
      }
    }
    if (oneHop.size > 0) {
      down.set(s, oneHop);
      for (const p of oneHop) indegree.set(p, (indegree.get(p) ?? 0) + 1);
    }
  }
  const ready = [...scripts].filter((s) => (indegree.get(s) ?? 0) === 0);
  const remaining = new Map(indegree);
  const order: string[] = [];
  while (ready.length > 0) {
    const n = ready.shift()!;
    order.push(n);
    for (const p of down.get(n) ?? []) {
      const d = (remaining.get(p) ?? 0) - 1;
      remaining.set(p, d);
      if (d === 0) ready.push(p);
    }
  }
  const orderedSet = new Set(order);
  const cyclic = [...scripts].filter((s) => !orderedSet.has(s));
  return { order, cyclic };
}
