windmill/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts

import type {
	AssetGraphMacroEdge,
	AssetGraphResponse,
	NativeTriggerKind,
	PipelineTriggerDraft
} from './types'
import {
	mergeColumnLineage,
	parsePipelineAnnotations,
	scd2CurrentTargetPath,
	type ColumnLineage,
	type PipelineAnnotations
} from './parsePipelineAnnotations'
import {
	extractWrites,
	extractReads,
	type AssetKind,
	type AssetWithAltAccessType
} from '$lib/components/assets/lib'

/** Minimal structural shape of a pipeline draft `resolveGraph` needs. */
export type GraphDraft = {
	script: { content: string }
	outputAssets?: Array<{ kind: AssetKind; path: string }>
	/** Reads captured on pane teardown. `undefined` = not captured (legacy
	 * bundle) → fall back to the session cache; `[]` = authoritative none. */
	inputAssets?: Array<{ kind: AssetKind; path: string }>
}

export type ResolveGraphInput = {
	/** Persisted graph from the backend (`/assets/graph`). */
	base: AssetGraphResponse
	/** In-flight drafts keyed by script path. */
	drafts: Map<string, GraphDraft>
	/** Body assets inferred for the currently-open script (live keystrokes). */
	liveBodyAssets: {
		scriptPath: string | undefined
		assets: AssetWithAltAccessType[]
		/** Body-inferred column lineage (DuckDB SQL AST) for the open script. */
		columnLineage?: ColumnLineage[]
	}
	/** Pipeline annotations parsed from the currently-open buffer. */
	liveAnnotations: { scriptPath: string | undefined; annotations: PipelineAnnotations }
	/** Sticky session caches of inferred body writes/reads per script path. */
	inferredWritesByPath: Map<string, Array<{ kind: AssetKind; path: string }>>
	inferredReadsByPath: Map<string, Array<{ kind: AssetKind; path: string }>>
	/**
	 * Sticky cache of native trigger kinds declared via `// on <kind>` in
	 * each script's deployed source. Filled by the load-time prefetch
	 * sweep. Used here to emit "missing" placeholders for scripts whose
	 * annotation has no matching trigger row in `base.triggers`.
	 */
	annotatedNativeKindsByPath: Map<string, Set<NativeTriggerKind>>
	/** Undeployed triggers; each stands in for its script's missing trigger of that kind. */
	triggerDrafts?: Iterable<PipelineTriggerDraft>
}

/** Mutable bag the sequential passes accumulate the resolved graph into. */
type Accumulator = {
	runnables: AssetGraphResponse['runnables']
	assets: AssetGraphResponse['assets']
	edges: AssetGraphResponse['edges']
	extraTriggers: AssetGraphResponse['triggers']
}

/**
 * Native trigger kinds with a persisted (non-asset) trigger row pointing at
 * `path`. These bind by script path and survive content edits, so they dedup
 * against draft/live annotations to avoid showing a real source node next to a
 * red "missing" placeholder for the same kind.
 */
function persistedNativeKinds(base: AssetGraphResponse, path: string): Set<string> {
	return new Set(
		base.triggers
			.filter(
				(t) =>
					t.trigger_kind !== 'asset' && t.runnable_kind === 'script' && t.runnable_path === path
			)
			.map((t) => t.trigger_kind)
	)
}

// Read-asset kinds that auto-derive a cascade trigger edge inside a
// `// pipeline`. Mirror of the backend `is_auto_trigger_kind`
// (windmill-common assets.rs): ducklake tables and s3 objects.
//
// `dbt` is excluded there and must stay excluded here: a warehouse relation is
// usually built by the dbt project that reads it, and a dbt run does not
// dispatch, so deriving an edge from every `dbt://` read would draw cascade
// arrows that mostly never fire. An EXPLICIT `// on dbt://…` is drawn — the
// relations a native `// materialize manual dbt://…` script writes do wake
// subscribers, and the deploy is what refuses the case that cannot (a relation
// dbt alone builds; `sole_dbt_producer` in windmill-common assets.rs).
const AUTO_TRIGGER_KINDS: ReadonlySet<AssetKind> = new Set(['ducklake', 's3object'])

/** `kind:path` refs of a script's `// materialize` write target(s) (base +
 *  the scd2 `<dim>_current` companion), which the body `SELECT` doesn't express. */
function materializeWriteRefs(parsed: PipelineAnnotations): string[] {
	const m = parsed.materialize
	if (!m) return []
	const refs = [`${m.targetKind}:${m.targetPath}`]
	const current = scd2CurrentTargetPath(m)
	if (current) refs.push(`${m.targetKind}:${current}`)
	return refs
}

/**
 * Backend-mirror of `derive_pipeline_asset_trigger_refs` (windmill-common
 * assets.rs): a pipeline script's ducklake/s3 read auto-wires a cascade
 * trigger edge from the FROM clause, so `// on <asset>` is only needed for
 * edges inference can't see. Excluded: assets the script also writes (`writes`
 * covers `w`/`rw`, so an `rw` self-read can't loop-trigger), muted assets,
 * `// mute all`, and any explicit `// on` (which already emits its own edge).
 * Returns the `{kind, path}` refs to overlay as unsaved asset triggers, deduped.
 */
function deriveAutoAssetTriggers(
	reads: Array<{ kind: AssetKind; path: string }>,
	writes: Array<{ kind: AssetKind; path: string }>,
	parsed: PipelineAnnotations
): Array<{ kind: AssetKind; path: string }> {
	// Auto-derivation is scoped to `// pipeline` scripts (backend gates on
	// `in_pipeline`); `// mute all` opts a pipeline script back out.
	if (!parsed.inPipeline || parsed.muteAll) return []
	const skip = new Set<string>([
		...writes.map((w) => `${w.kind}:${w.path}`),
		// The `// materialize` target(s) are this script's own output — the body
		// `SELECT` doesn't express the write, so an incremental model that reads
		// its own target would otherwise auto-derive a self-cascade edge (backend
		// parity: the deploy path upgrades the same read to `rw`).
		...materializeWriteRefs(parsed),
		...parsed.muteAssets.map((a) => `${a.kind}:${a.path}`),
		...parsed.triggerAssets.map((a) => `${a.kind}:${a.path}`)
	])
	const out: Array<{ kind: AssetKind; path: string }> = []
	for (const r of reads) {
		const key = `${r.kind}:${r.path}`
		if (!AUTO_TRIGGER_KINDS.has(r.kind) || skip.has(key)) continue
		skip.add(key) // dedup within reads too
		out.push({ kind: r.kind, path: r.path })
	}
	return out
}

/** Edge key matching a lineage/trigger edge's `(asset, runnable)` pair. */
function edgeKey(a: {
	asset_kind: string
	asset_path: string
	runnable_kind: string
	runnable_path: string
}): string {
	return `${a.asset_kind}:${a.asset_path}->${a.runnable_kind}:${a.runnable_path}`
}

/**
 * Keys of "muted read" edges: a ducklake/s3 asset a script reads read-*only*
 * (`access_type === 'r'`) yet has NO cascade trigger for. Auto-derivation is the
 * default, so a supported read with no trigger means the author opted it out
 * with `// mute <asset>` / `// mute all` (or it's an explicit non-triggering
 * read). The canvas badges these — the exception — instead of every derived
 * edge. Self-reads are excluded: a script that also writes the asset carries
 * `'rw'` on the deployed graph (one edge) or a separate `'w'` edge in the live
 * overlay (a `// materialize` producer reading its own target), and neither
 * should badge as muted — it's the script's own output, not a suppressed input.
 *
 * Only `// pipeline` scripts are considered: auto-derivation never applies to a
 * plain script or a flow, so a non-pipeline runnable reading a ducklake/s3
 * asset has no auto trigger to suppress and must render as ordinary lineage.
 */
export function computeMutedReadKeys(
	edges: AssetGraphResponse['edges'],
	triggers: AssetGraphResponse['triggers'],
	runnables: AssetGraphResponse['runnables']
): Set<string> {
	const pipelineScripts = new Set(
		runnables.filter((r) => r.usage_kind === 'script' && r.in_pipeline).map((r) => r.path)
	)
	const cascaded = new Set(
		triggers.filter((t) => t.trigger_kind === 'asset').map((t) => edgeKey(t))
	)
	const written = new Set(
		edges.filter((e) => e.access_type === 'w' || e.access_type === 'rw').map((e) => edgeKey(e))
	)
	const muted = new Set<string>()
	for (const e of edges) {
		if (e.access_type !== 'r' || !AUTO_TRIGGER_KINDS.has(e.asset_kind)) continue
		if (e.runnable_kind !== 'script' || !pipelineScripts.has(e.runnable_path)) continue
		const key = edgeKey(e)
		if (!cascaded.has(key) && !written.has(key)) muted.add(key)
	}
	return muted
}

/** `kind:path` keys of persisted asset (`// on <asset>`) triggers for `path`. */
function persistedAssetKeys(base: AssetGraphResponse, path: string): Set<string> {
	return new Set(
		base.triggers
			.filter(
				(t) =>
					t.trigger_kind === 'asset' && t.runnable_kind === 'script' && t.runnable_path === path
			)
			.map((t) => (t.trigger_kind === 'asset' ? `${t.asset_kind}:${t.asset_path}` : ''))
	)
}

/**
 * Emit a red "missing" placeholder for each annotated native kind that has no
 * matching attached trigger row. `opts.unsaved` marks editor-driven overlays
 * (drafts / live buffer); deployed-but-unswept scripts pass `unsaved: false`.
 */
function pushMissingNativeTriggers(
	extraTriggers: AssetGraphResponse['triggers'],
	annotatedKinds: Iterable<NativeTriggerKind>,
	attachedKinds: Set<string>,
	path: string,
	opts: { unsaved: boolean }
) {
	for (const kind of annotatedKinds) {
		if (attachedKinds.has(kind)) continue
		extraTriggers.push({
			trigger_kind: kind,
			runnable_kind: 'script',
			runnable_path: path,
			...(opts.unsaved ? { unsaved: true } : {}),
			missing: true
		})
	}
}

/**
 * Merge the persisted base graph with the draft, session-inferred and
 * open-script live overlays into one `AssetGraphResponse`.
 *
 * Precedence (lowest → highest): base < session-inferred (read/write
 * lineage) < per-draft seeded triggers/outputs < live annotations for the
 * one currently-open script (which replace that path's seeded triggers).
 * Overlay edges are tagged `unsaved` and deduped against persisted edges of
 * the matching access direction.
 *
 * Pure (no Svelte reactivity) so the precedence matrix is unit-testable in
 * isolation — extracted verbatim from the pipeline page's former
 * `graphWithDraft` `$derived`. See `resolveGraph.test.ts`.
 */
export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse {
	const { base } = input
	const ctx = makeContext(input)

	// Pass order mirrors the precedence comment, lowest → highest:
	const acc = seedAccumulator(input, ctx)
	seedDraftOverlays(acc, input)
	applyLiveBufferOverlay(acc, input, ctx)
	crossCheckSweptScripts(acc, input)
	overlayInferredLineage(acc, input)

	// Drop persisted ASSET triggers for drafted paths — those come from the
	// deployed `// on <asset>` annotations, which the draft's live/parsed
	// annotations now own. Native triggers (kafka/schedule/…) are kept: they
	// bind by `script_path`, which the draft shares, so the attachment is still
	// valid regardless of content edits.
	const baseTriggers = base.triggers.filter((t) => {
		if (t.trigger_kind !== 'asset') return true
		if (ctx.isDrafted(t.runnable_kind, t.runnable_path)) return false
		if (
			t.runnable_kind === 'script' &&
			t.runnable_path === ctx.openPath &&
			ctx.staleForOpen(t.asset_kind, t.asset_path)
		)
			return false
		return true
	})
	// Mirror the backend's skip-if-empty: no `macro_edges` key at all when
	// there is nothing to show (also keeps the no-macros response shape
	// byte-identical to before the feature).
	const macroEdges = resolveMacroEdges(input)
	const triggers = withTriggerDrafts(
		[...baseTriggers, ...acc.extraTriggers],
		input.triggerDrafts ?? []
	)
	return {
		...base,
		assets: dropOrphanedAssets(acc.assets, base, acc.edges, triggers),
		runnables: acc.runnables,
		edges: acc.edges,
		triggers,
		...(macroEdges.length > 0 || base.macro_edges ? { macro_edges: macroEdges } : {})
	}
}

function withTriggerDrafts(
	triggers: AssetGraphResponse['triggers'],
	triggerDrafts: Iterable<PipelineTriggerDraft>
): AssetGraphResponse['triggers'] {
	const drafts = [...triggerDrafts]
	if (drafts.length === 0) return triggers
	const drafted = new Set(drafts.map((d) => `${d.kind}:${d.config.script_path}`))
	return [
		...triggers.filter(
			(t) =>
				!(
					t.trigger_kind !== 'asset' &&
					t.missing &&
					t.runnable_kind === 'script' &&
					drafted.has(`${t.trigger_kind}:${t.runnable_path}`)
				)
		),
		...drafts.map((d) => ({
			trigger_kind: d.kind,
			path: d.config.path,
			runnable_kind: 'script' as const,
			runnable_path: d.config.script_path,
			unsaved: true,
			draft: true
		}))
	]
}

/**
 * Nodes the pipeline cannot work with, with the reason. Assets (keyed
 * `kind:path`): one nothing uses at all, and one a script subscribes to
 * (`// on <asset>`) that no runnable in the graph writes. Scripts (keyed by
 * path): the subscriber, whose subscription can never fire. A body read of an
 * external asset is fine and is not passed here: only the explicit `// on`
 * annotations, per script path, count as subscriptions.
 */
export function pipelineNodeErrors(
	graph: AssetGraphResponse,
	explicitOnByPath: Map<string, Array<{ kind: AssetKind; path: string }>>
): { assets: Map<string, string>; scripts: Map<string, string> } {
	const referenced = referencedAssetKeys(graph)
	const written = new Set(
		graph.edges
			.filter((e) => e.access_type === 'w' || e.access_type === 'rw')
			.map((e) => `${e.asset_kind}:${e.asset_path}`)
	)
	// The dbt project's own edges are hidden from the pipeline graph (see
	// `hideDbtRunnables`), so its relations are judged by their provenance: the
	// project uses every one, and materializes all but its sources.
	for (const a of graph.assets) {
		if (!a.dbt) continue
		referenced.add(`${a.kind}:${a.path}`)
		if (a.dbt.resource_type !== 'source') written.add(`${a.kind}:${a.path}`)
	}
	const errors = new Map<string, string>()
	const scriptErrors = new Map<string, string>()
	for (const a of graph.assets) {
		const key = `${a.kind}:${a.path}`
		if (!referenced.has(key)) errors.set(key, 'nothing in the pipeline uses it')
	}
	const scripts = new Set(
		graph.runnables.filter((r) => r.usage_kind === 'script').map((r) => r.path)
	)
	for (const [path, refs] of explicitOnByPath) {
		if (!scripts.has(path)) continue
		for (const ref of refs) {
			const key = `${ref.kind}:${ref.path}`
			if (written.has(key)) continue
			if (!errors.has(key)) {
				errors.set(key, `${path} runs on it, but nothing in the pipeline writes it`)
			}
			if (!scriptErrors.has(path)) {
				scriptErrors.set(path, `runs on ${ref.path}, which nothing in the pipeline writes`)
			}
		}
	}
	return { assets: errors, scripts: scriptErrors }
}

function referencedAssetKeys(
	graph: Pick<AssetGraphResponse, 'edges' | 'triggers' | 'test_edges' | 'dbt_edges'>
): Set<string> {
	const keys = new Set<string>()
	for (const e of graph.edges) keys.add(`${e.asset_kind}:${e.asset_path}`)
	for (const t of graph.triggers)
		if (t.trigger_kind === 'asset') keys.add(`${t.asset_kind}:${t.asset_path}`)
	for (const e of graph.test_edges ?? []) keys.add(`${e.asset_kind}:${e.asset_path}`)
	for (const e of graph.dbt_edges ?? []) {
		keys.add(`dbt:${e.from_asset_path}`)
		keys.add(`dbt:${e.to_asset_path}`)
	}
	return keys
}

/**
 * The backend only returns assets something references, so a deployed asset
 * whose every edge and trigger the draft/live overlays stripped — an output the
 * open script renamed — would linger as an edgeless node until the deploy
 * lands. Assets that were already unreferenced in `base` are left alone.
 */
function dropOrphanedAssets(
	assets: AssetGraphResponse['assets'],
	base: AssetGraphResponse,
	edges: AssetGraphResponse['edges'],
	triggers: AssetGraphResponse['triggers']
): AssetGraphResponse['assets'] {
	const before = referencedAssetKeys(base)
	const after = referencedAssetKeys({ ...base, edges, triggers })
	return assets.filter((a) => {
		const key = `${a.kind}:${a.path}`
		return !before.has(key) || after.has(key)
	})
}

/**
 * Macro-library → consumer edges: the deployed base edges, with `// use`
 * declarations of overlaid scripts (drafts + the open buffer) taking over
 * their consumer's `via_use` edges so adding/removing a `// use` line updates
 * the canvas live. Detection-based edges (macro calls in the deployed body)
 * are backend-owned and only refresh on redeploy.
 */
function resolveMacroEdges(input: ResolveGraphInput): AssetGraphMacroEdge[] {
	const { base, drafts, liveAnnotations } = input
	const libMacroNames = new Map<string, string[]>()
	for (const r of base.runnables) {
		if (r.usage_kind === 'script' && r.macros?.length) {
			libMacroNames.set(
				r.path,
				r.macros.map((m) => m.name)
			)
		}
	}
	const useByPath = new Map<string, string[]>()
	for (const [path, d] of drafts) {
		useByPath.set(path, parsePipelineAnnotations(d.script.content).useLibs)
	}
	if (liveAnnotations.scriptPath) {
		// `?? []` — callers may hand a minimal annotations object (tests, older
		// call sites) that predates the field.
		useByPath.set(liveAnnotations.scriptPath, liveAnnotations.annotations.useLibs ?? [])
	}
	const out: AssetGraphMacroEdge[] = []
	for (const e of base.macro_edges ?? []) {
		if (e.via_use && useByPath.has(e.consumer_path)) continue
		out.push({ ...e })
	}
	for (const [path, libs] of useByPath) {
		for (const lib of libs) {
			const existing = out.find((e) => e.lib_path === lib && e.consumer_path === path)
			if (existing) {
				// Upgrade the detection edge in place: `// use` pulls in the whole
				// library, so the edge covers every macro the lib defines.
				existing.via_use = true
				existing.unsaved = true
				existing.macro_names = [
					...new Set([...existing.macro_names, ...(libMacroNames.get(lib) ?? [])])
				]
			} else {
				out.push({
					lib_path: lib,
					consumer_path: path,
					macro_names: libMacroNames.get(lib) ?? [],
					via_use: true,
					unsaved: true
				})
			}
		}
	}
	return out
}

// Light regex extraction of a draft macro library's definitions for the live
// node badge. The strict grammar lives in the Rust `parse_macro_library` at
// deploy; this only needs name/params/table-ness for display (nested parens
// in a default value may truncate the shown signature, never the deploy).
const MACRO_DEF_RE =
	/create\s+(?:or\s+replace\s+)?(?:temp(?:orary)?\s+)?(?:macro|function)\s+([a-zA-Z_][a-zA-Z0-9_]*)\s*\(([^)]*)\)\s*as\s+(table\b)?/gi

export function extractDraftMacros(
	content: string
): { name: string; params: string; is_table: boolean }[] {
	const out: { name: string; params: string; is_table: boolean }[] = []
	for (const m of content.matchAll(MACRO_DEF_RE)) {
		out.push({ name: m[1].toLowerCase(), params: m[2].trim(), is_table: m[3] !== undefined })
	}
	return out
}

type ResolveContext = {
	draftedPaths: Set<string>
	isDrafted: (kind: string, p: string) => boolean
	openPath: string | undefined
	staleForOpen: (kind: AssetKind, path: string) => boolean
}

/** Derive the shared "what's drafted / open / stale" predicates once. */
function makeContext(input: ResolveGraphInput): ResolveContext {
	const { drafts, liveBodyAssets, liveAnnotations } = input

	// A drafted path's content-derived lineage is owned entirely by the draft
	// overlay (live inference for the active draft, parsed annotations /
	// snapshot for inactive ones). Drop persisted base asset edges for those
	// paths so it shows only the draft's I/O, not the union of the saved version
	// and the in-flight edits. (Base asset triggers are dropped at the return;
	// native triggers are kept — they bind by path.)
	const draftedPaths = new Set(drafts.keys())
	const isDrafted = (kind: string, p: string) => kind === 'script' && draftedPaths.has(p)

	const openPath = liveBodyAssets.scriptPath
	const openIsSavedEdit = openPath !== undefined && !draftedPaths.has(openPath)
	const liveRefKeys = new Set<string>()
	if (openIsSavedEdit) {
		if (liveAnnotations.scriptPath === openPath) {
			for (const a of liveAnnotations.annotations.triggerAssets)
				liveRefKeys.add(`${a.kind}:${a.path}`)
			// The `// materialize <asset>` target is a declared *output*, but it
			// lives in an annotation (not the SQL body), so neither triggerAssets
			// (inputs) nor the body-inferred assets cover it. Without this its
			// persisted write-edge is judged stale and dropped the moment the
			// script is selected/edited — leaving the output asset unlinked. A
			// managed scd2 producer persists a second write to the `<dim>_current`
			// companion view, so keep that too or a consumer of only the view
			// orphans while the producer is open for editing.
			const m = liveAnnotations.annotations.materialize
			if (m) {
				liveRefKeys.add(`${m.targetKind}:${m.targetPath}`)
				const currentPath = scd2CurrentTargetPath(m)
				if (currentPath) liveRefKeys.add(`${m.targetKind}:${currentPath}`)
			}
		}
		for (const a of liveBodyAssets.assets) liveRefKeys.add(`${a.kind}:${a.path}`)
	}
	const staleForOpen = (kind: AssetKind, path: string) =>
		openIsSavedEdit && !liveRefKeys.has(`${kind}:${path}`)

	return { draftedPaths, isDrafted, openPath, staleForOpen }
}

/** Base graph plus the persisted edges that survive the draft/open filter. */
function seedAccumulator(input: ResolveGraphInput, ctx: ResolveContext): Accumulator {
	const { base } = input
	const edges = base.edges.filter((e) => {
		if (ctx.isDrafted(e.runnable_kind, e.runnable_path)) return false
		if (
			e.runnable_kind === 'script' &&
			e.runnable_path === ctx.openPath &&
			ctx.staleForOpen(e.asset_kind, e.asset_path)
		)
			return false
		return true
	})
	return {
		runnables: [...base.runnables],
		assets: [...base.assets],
		edges,
		extraTriggers: []
	}
}

/**
 * The write output(s) a `// materialize <asset>` annotation declares: the
 * target itself, plus — for a managed scd2 materialize — the `<dim>_current`
 * companion view (mirrors the deploy path), so a consumer of only the view
 * links back to this producer instead of orphaning.
 */
function materializeOuts(
	parsed: PipelineAnnotations
): Array<{ kind: AssetKind; path: string; derivedFrom?: string }> {
	if (!parsed.materialize) return []
	const outs: Array<{ kind: AssetKind; path: string; derivedFrom?: string }> = [
		{ kind: parsed.materialize.targetKind, path: parsed.materialize.targetPath }
	]
	const currentPath = scd2CurrentTargetPath(parsed.materialize)
	if (currentPath) {
		outs.push({
			kind: parsed.materialize.targetKind,
			path: currentPath,
			derivedFrom: parsed.materialize.targetPath
		})
	}
	return outs
}

/** Add an output asset node + unsaved write edge for `runnablePath`, deduped
 *  against the assets/edges already accumulated (base survivors included). */
function pushWriteOut(
	acc: Accumulator,
	runnablePath: string,
	out: { kind: AssetKind; path: string; derivedFrom?: string }
) {
	const existing = acc.assets.find((a) => a.kind === out.kind && a.path === out.path)
	if (!existing) {
		acc.assets.push({
			kind: out.kind,
			path: out.path,
			...(out.derivedFrom ? { derived_from: out.derivedFrom } : {})
		})
	} else if (out.derivedFrom && existing.derived_from == undefined) {
		existing.derived_from = out.derivedFrom
	}
	const hasWriteEdge = acc.edges.some(
		(e) =>
			e.runnable_kind === 'script' &&
			e.runnable_path === runnablePath &&
			e.asset_kind === out.kind &&
			e.asset_path === out.path &&
			(e.access_type === 'w' || e.access_type === 'rw')
	)
	if (hasWriteEdge) return
	acc.edges.push({
		runnable_path: runnablePath,
		runnable_kind: 'script',
		asset_kind: out.kind,
		asset_path: out.path,
		access_type: 'w',
		unsaved: true
	})
}

/** Add an input asset node + unsaved read edge for `runnablePath`, deduped
 *  against the assets/edges already accumulated. */
function pushReadIn(
	acc: Accumulator,
	runnablePath: string,
	inp: { kind: AssetKind; path: string }
) {
	if (!acc.assets.some((a) => a.kind === inp.kind && a.path === inp.path)) {
		acc.assets.push({ kind: inp.kind, path: inp.path })
	}
	const hasReadEdge = acc.edges.some(
		(e) =>
			e.runnable_kind === 'script' &&
			e.runnable_path === runnablePath &&
			e.asset_kind === inp.kind &&
			e.asset_path === inp.path &&
			(e.access_type === 'r' || e.access_type === 'rw')
	)
	if (hasReadEdge) return
	acc.edges.push({
		runnable_path: runnablePath,
		runnable_kind: 'script',
		asset_kind: inp.kind,
		asset_path: inp.path,
		access_type: 'r',
		unsaved: true
	})
}

/**
 * Every draft contributes: a runnable, output asset(s), a write edge, live
 * read lineage for the active draft, plus its seeded asset/native triggers
 * (template includes `// on schedule …` by default). Iterates the whole
 * `drafts` map so multiple concurrent drafts all render at once.
 */
function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) {
	const { base, drafts, liveBodyAssets, inferredReadsByPath, inferredWritesByPath } = input
	const { runnables, assets, extraTriggers } = acc

	for (const [path, d] of drafts) {
		// The open draft's annotations come from the live buffer, not the draft
		// snapshot — the snapshot only syncs on pane teardown, so parsing it here
		// would pin annotation-derived outputs (`// materialize` target, badges)
		// to their pre-edit values until the user clicks away.
		const parsed =
			path === input.liveAnnotations.scriptPath
				? input.liveAnnotations.annotations
				: parsePipelineAnnotations(d.script.content)
		// For the open script, fold in the WASM-inferred column lineage (DuckDB
		// SQL AST) under the same annotation-wins precedence the backend applies
		// on deploy, so the live preview matches what deploys. Only the open
		// script carries live inference (`liveBodyAssets`); other drafts stay
		// annotation-only until they deploy (the backend infers then).
		const inferredCL =
			path === liveBodyAssets.scriptPath ? (liveBodyAssets.columnLineage ?? []) : []
		const mergedCL = mergeColumnLineage(inferredCL, parsed.columnLineage)
		// The `// materialize` target this draft's column lineage describes, so
		// the column graph anchors to it rather than guessing a write-edge.
		const materializeTarget = parsed.materialize
			? { kind: parsed.materialize.targetKind, path: parsed.materialize.targetPath }
			: undefined
		// A draft can coexist with a base entry — during save the refetch
		// lands before drafts cleanup, and a user re-editing a deployed
		// script also produces both. In that case we mutate the existing
		// base runnable to carry `unsaved: true` instead of pushing a
		// duplicate (which would crash svelte-flow's keyed each), so the
		// canvas + trigger-node labels reflect that there's pending body
		// editing for this path.
		// `// macros` library draft: extract the definitions for the live node
		// badge (regex-light; the strict parse happens at deploy).
		const draftMacros = parsed.macros ? extractDraftMacros(d.script.content) : []
		const baseIdx = runnables.findIndex((r) => r.usage_kind === 'script' && r.path === path)
		if (baseIdx === -1) {
			runnables.push({
				path,
				usage_kind: 'script',
				in_pipeline: true,
				partition_kind: parsed.partition?.kind,
				freshness: parsed.freshness?.duration,
				tag: parsed.tag,
				retry: parsed.retry,
				data_tests: parsed.dataTests.length > 0 ? parsed.dataTests : undefined,
				column_lineage: mergedCL.length > 0 ? mergedCL : undefined,
				materialize_target: materializeTarget,
				macros: draftMacros.length > 0 ? draftMacros : undefined,
				unsaved: true
			})
		} else {
			// Refresh annotation-derived badges from the live parse too, so
			// adding/removing `// data_test` / `// column` lines on an
			// already-deployed script updates the badge immediately (not only
			// after redeploy/refetch).
			runnables[baseIdx] = {
				...runnables[baseIdx],
				data_tests: parsed.dataTests.length > 0 ? parsed.dataTests : undefined,
				column_lineage: mergedCL.length > 0 ? mergedCL : undefined,
				materialize_target: materializeTarget,
				macros: draftMacros.length > 0 ? draftMacros : undefined,
				unsaved: true
			}
		}
		// Output asset(s): two-tier resolution.
		//   1. Active draft (the body the user is editing right now):
		//      live body inference is authoritative — renaming a
		//      CREATE TABLE target or writeS3File path retires the
		//      old output node and surfaces the new one as the user
		//      types.
		//   2. Inactive draft: its captured `outputAssets` (inferred at
		//      creation/last edit, or the seeded output for a fresh draft
		//      whose body doesn't yet write anything inferable).
		const liveForThisDraft = liveBodyAssets.scriptPath === path
		const writeOuts: Array<{ kind: AssetKind; path: string; derivedFrom?: string }> = []
		if (liveForThisDraft) {
			writeOuts.push(...extractWrites(liveBodyAssets.assets))
		} else if (d.outputAssets) {
			writeOuts.push(...d.outputAssets)
		}
		// `// materialize <asset>` declares a write output via annotation, not
		// the SQL body, so the body-inference tiers above miss it (pushWriteOut
		// dedups against existing assets/edges).
		writeOuts.push(...materializeOuts(parsed))
		for (const out of writeOuts) {
			pushWriteOut(acc, path, out)
		}
		// Read lineage: live inference for the open draft, the captured
		// `inputAssets` snapshot for inactive ones (with the session cache as a
		// legacy fallback). Without the inactive tier, merely selecting another
		// node would drop this draft's input edges from the canvas.
		const draftReads = liveForThisDraft
			? extractReads(liveBodyAssets.assets)
			: (d.inputAssets ?? inferredReadsByPath.get(path) ?? [])
		for (const inp of draftReads) {
			pushReadIn(acc, path, inp)
		}
		// Seed trigger edges from the draft's template so the graph stays
		// stable when the user clicks off this draft. Live annotations
		// (below) take over for the currently-open draft so keystroke
		// edits still update in real time.
		for (const a of parsed.triggerAssets) {
			extraTriggers.push({
				trigger_kind: 'asset',
				asset_kind: a.kind,
				asset_path: a.path,
				runnable_kind: 'script',
				runnable_path: path,
				unsaved: true
			})
			// Also synthesize the asset node so the trigger edge has a
			// target even if the upstream asset isn't in base (e.g. the
			// producer script is in another folder we haven't fetched
			// or also a draft).
			const hasTriggerAsset = assets.some((x) => x.kind === a.kind && x.path === a.path)
			if (!hasTriggerAsset) assets.push({ kind: a.kind, path: a.path })
		}
		// Auto-derived cascade edges (backend parity): a ducklake/s3 read wires
		// the edge from the body alone. Reads reuse the tier above; writes come
		// from the active draft's live inference, else the captured snapshot /
		// session cache. The open buffer's derived edges are re-computed
		// authoritatively in applyLiveBufferOverlay (which strips this path's
		// seeded triggers first), same as the explicit `// on` triggers above.
		const draftWrites = liveForThisDraft
			? extractWrites(liveBodyAssets.assets)
			: (d.outputAssets ?? inferredWritesByPath.get(path) ?? [])
		for (const a of deriveAutoAssetTriggers(draftReads, draftWrites, parsed)) {
			extraTriggers.push({
				trigger_kind: 'asset',
				asset_kind: a.kind,
				asset_path: a.path,
				runnable_kind: 'script',
				runnable_path: path,
				unsaved: true
			})
			if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) {
				assets.push({ kind: a.kind, path: a.path })
			}
		}
		// Native trigger annotations on a draft are "missing" until a
		// matching trigger row exists. A brand-new draft never has one (the
		// script isn't deployed yet), but a draft promoted from unsaved
		// edits to a *deployed* script keeps its persisted trigger rows —
		// they bind by path and are not dropped above. Dedup against those
		// so the canvas doesn't show the real schedule node next to a red
		// "missing" placeholder for the same kind.
		pushMissingNativeTriggers(
			extraTriggers,
			parsed.nativeTriggers.map((n) => n.kind),
			persistedNativeKinds(base, path),
			path,
			{ unsaved: true }
		)
	}
}

/**
 * Live-parsed overlay for the currently-open script — takes precedence over
 * the seeded-template triggers for the same path by swapping them out. Scoped
 * to one path (only one pane is open at a time).
 */
function applyLiveBufferOverlay(acc: Accumulator, input: ResolveGraphInput, ctx: ResolveContext) {
	const { base, liveAnnotations } = input
	const { assets, extraTriggers } = acc
	const livePath = liveAnnotations.scriptPath
	if (!livePath) return

	// When the open buffer is a draft, its base asset triggers were dropped
	// above (draft-owned), so don't dedup live asset triggers against them —
	// every live `// on <asset>` is emitted as unsaved. For a non-draft open
	// script (viewing a saved one), keep deduping so live re-parse doesn't
	// double the persisted triggers.
	const assetKeys = ctx.draftedPaths.has(livePath)
		? new Set<string>()
		: persistedAssetKeys(base, livePath)
	// Strip seeded triggers we computed above for the active draft;
	// live annotations are authoritative for the open buffer.
	for (let i = extraTriggers.length - 1; i >= 0; i--) {
		if (extraTriggers[i].runnable_path === livePath) extraTriggers.splice(i, 1)
	}
	for (const a of liveAnnotations.annotations.triggerAssets) {
		const key = `${a.kind}:${a.path}`
		if (assetKeys.has(key)) continue
		extraTriggers.push({
			trigger_kind: 'asset',
			asset_kind: a.kind,
			asset_path: a.path,
			runnable_kind: 'script',
			runnable_path: livePath,
			unsaved: true
		})
		// Synthesize the asset node so the new trigger edge has a
		// target — without this, typing `// on s3:///...` adds an
		// edge to a node that doesn't exist and the canvas silently
		// drops it. Mirrors the draft branch above.
		if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) {
			assets.push({ kind: a.kind, path: a.path })
		}
	}
	// Auto-derived cascade edges (backend parity): a ducklake/s3 read wires
	// the edge from the FROM clause alone, keystroke-live. The open buffer's
	// live body inference is authoritative for its reads/writes. `// mute` /
	// `// mute all` and explicit `// on` (above) suppress a derived edge; a
	// derived edge already persisted for a non-draft open script is deduped
	// via `assetKeys`.
	if (input.liveBodyAssets.scriptPath === livePath) {
		const reads = extractReads(input.liveBodyAssets.assets)
		const writes = extractWrites(input.liveBodyAssets.assets)
		for (const a of deriveAutoAssetTriggers(reads, writes, liveAnnotations.annotations)) {
			if (assetKeys.has(`${a.kind}:${a.path}`)) continue
			extraTriggers.push({
				trigger_kind: 'asset',
				asset_kind: a.kind,
				asset_path: a.path,
				runnable_kind: 'script',
				runnable_path: livePath,
				unsaved: true
			})
			if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) {
				assets.push({ kind: a.kind, path: a.path })
			}
		}
	}
	// Lineage overlay for an open SAVED script. seedDraftOverlays owns drafts,
	// but a deployed script's unsaved edits are only promoted to a draft on
	// pane teardown — until then the buffer's lineage lives here. staleForOpen
	// already dropped the base edges the buffer no longer references; this adds
	// the ones it now does. Without it, retargeting `// materialize` (or a body
	// write/read) shows no output edge until the user clicks away.
	if (!ctx.draftedPaths.has(livePath)) {
		for (const out of materializeOuts(liveAnnotations.annotations)) {
			pushWriteOut(acc, livePath, out)
		}
		if (input.liveBodyAssets.scriptPath === livePath) {
			for (const out of extractWrites(input.liveBodyAssets.assets)) {
				pushWriteOut(acc, livePath, out)
			}
			for (const inp of extractReads(input.liveBodyAssets.assets)) {
				pushReadIn(acc, livePath, inp)
			}
		}
	}
	// Native trigger annotations: kinds for which a matching trigger
	// row was found in the backend response. If the live buffer
	// declares `// on kafka` and at least one kafka_trigger row points
	// at this script, the source node is already on the canvas — no
	// overlay needed. Otherwise emit a "missing" placeholder so the
	// user can either create the trigger row or remove the annotation.
	pushMissingNativeTriggers(
		extraTriggers,
		liveAnnotations.annotations.nativeTriggers.map((n) => n.kind),
		persistedNativeKinds(base, livePath),
		livePath,
		{ unsaved: true }
	)
}

/**
 * Cross-check for already-deployed scripts (not the open buffer): if a
 * script's persisted body declares `// on kafka` but no matching kafka_trigger
 * row points at it, surface a red placeholder. The annotated-kinds map is
 * filled by the page-level prefetch sweep (one read per script in the folder);
 * drafts and the active editor are handled by the earlier passes. Scripts not
 * yet swept contribute nothing — they surface on the next refetch.
 */
function crossCheckSweptScripts(acc: Accumulator, input: ResolveGraphInput) {
	const { base, drafts, liveAnnotations, annotatedNativeKindsByPath } = input
	const livePath = liveAnnotations.scriptPath
	for (const [scriptPath, kinds] of annotatedNativeKindsByPath) {
		if (drafts.has(scriptPath)) continue
		if (scriptPath === livePath) continue
		pushMissingNativeTriggers(
			acc.extraTriggers,
			kinds,
			persistedNativeKinds(base, scriptPath),
			scriptPath,
			{ unsaved: false }
		)
	}
}

/**
 * Live body-asset lineage for any persisted script inferred at least once this
 * session (maps filled by `handleAssetsChange` + the load prefetch). Drafts
 * are handled by `seedDraftOverlays`. For scripts whose deploy didn't persist
 * their body assets (e.g. older WASM at save time, or object-form
 * writeS3File), this keeps the lineage edge on the canvas across selection
 * changes — not just while selected.
 */
function overlayInferredLineage(acc: Accumulator, input: ResolveGraphInput) {
	const { drafts, inferredWritesByPath, inferredReadsByPath } = input
	const { assets, edges } = acc

	const overlayLineage = (
		byPath: Map<string, Array<{ kind: AssetKind; path: string }>>,
		access: 'w' | 'r'
	) => {
		for (const [scriptPath, refs] of byPath) {
			if (drafts.has(scriptPath)) continue
			for (const a of refs) {
				// Dedup against the ACCUMULATED edges, not just base: for the open
				// script these same refs may already be overlaid by
				// applyLiveBufferOverlay (which feeds the maps on the route page),
				// and a duplicate would collide on the canvas's endpoint-derived
				// edge ids.
				const hasEdge = edges.some(
					(e) =>
						e.runnable_path === scriptPath &&
						e.runnable_kind === 'script' &&
						e.asset_kind === a.kind &&
						e.asset_path === a.path &&
						(e.access_type === access || e.access_type === 'rw')
				)
				if (hasEdge) continue
				if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) {
					assets.push({ kind: a.kind, path: a.path })
				}
				edges.push({
					runnable_path: scriptPath,
					runnable_kind: 'script',
					asset_kind: a.kind,
					asset_path: a.path,
					access_type: access,
					unsaved: true
				})
			}
		}
	}
	overlayLineage(inferredWritesByPath, 'w')
	overlayLineage(inferredReadsByPath, 'r')
}

/** dbt project ↔ relation association, for a graph that does NOT draw it.
 *
 * A dbt script owns every relation of its project, so drawing one edge per
 * model buries the lineage that matters — `ref()` between models, and native
 * consumers — under a fan-out that grows with the project. The canvas suppresses
 * those edges and conveys the association through the model's badge instead:
 * `ownerByAsset` powers badge → project, `writesByOwner` powers project →
 * the transforms it materializes (its sources are inputs, not transforms).
 *
 * Only the DRAWING is dropped. The producer rows these are derived from still
 * answer "who produced this".
 */
export function dbtAssociations(
	runnables: Array<{ usage_kind: string; path: string; dbt?: unknown }>,
	edges: Array<{
		runnable_kind: string
		runnable_path: string
		asset_kind: string
		asset_path: string
		access_type?: string | null
	}>
): { ownerByAsset: Map<string, string>; writesByOwner: Map<string, Set<string>> } {
	const dbtRunnableIds = new Set(
		runnables.filter((r) => r.dbt).map((r) => `${r.usage_kind}:${r.path}`)
	)
	const ownerByAsset = new Map<string, string>()
	const writesByOwner = new Map<string, Set<string>>()
	// Candidates per relation, kept apart so the choice does not depend on the
	// order the edges arrive in. Two things decide it: a project that WRITES a
	// relation outranks one that merely declares it as a SOURCE, and among
	// several writers — which the backend permits, two selections of one project
	// materializing the same model — the smallest id wins. Anything less makes
	// the badge lead somewhere else between two loads of the same graph.
	const writersByAsset = new Map<string, string[]>()
	const readersByAsset = new Map<string, string[]>()
	for (const e of edges) {
		const runnableId = `${e.runnable_kind}:${e.runnable_path}`
		if (!dbtRunnableIds.has(runnableId)) continue
		const assetId = `asset:${e.asset_kind}:${e.asset_path}`
		const writes = e.access_type === 'w' || e.access_type === 'rw'
		const into = writes ? writersByAsset : readersByAsset
		let candidates = into.get(assetId)
		if (!candidates) into.set(assetId, (candidates = []))
		candidates.push(runnableId)
		if (writes) {
			let set = writesByOwner.get(runnableId)
			if (!set) writesByOwner.set(runnableId, (set = new Set()))
			set.add(assetId)
		}
	}
	for (const [assetId, candidates] of [...readersByAsset, ...writersByAsset]) {
		ownerByAsset.set(assetId, [...candidates].sort()[0])
	}
	return { ownerByAsset, writesByOwner }
}