/**
 * Transactional creation of a Wertstrom from a confirmed QAF import preview
 * (QVS-P2, KAR-971, architecture.md §2/§4).
 *
 * Re-derives nodes/connections from the DB (qaf-source.ts + mapper.ts)
 * rather than trusting a client-supplied node array wholesale — the only
 * inputs trusted from the client are the previewToken (staleness check),
 * title/description, and which rowIndexes to deselect. This keeps E4
 * (project_id forced from the QAF source) and the "0 rows = not found" IDOR
 * discipline (architecture.md §9) intact all the way to the write.
 *
 * The actual atomic write (value_stream_maps + value_stream_imports in one
 * transaction) happens in the create_value_stream_from_qaf SECURITY DEFINER
 * RPC (supabase-migration-value-stream-imports.sql) — this module's job is
 * preparing exactly the payload that RPC needs and translating its
 * result/errors back into a typed outcome.
 */

import type { SupabaseClient } from '@supabase/supabase-js'
import type { VsmConnection, VsmNode } from '@/lib/vsm-types'
import { MAPPING_VERSION } from './mapper'
import { loadQafSourceRows, mapAndFingerprint } from './qaf-source'
import type { MissingFieldReason, QvsWarning, QvsWarningCode, QvsWarningSeverity } from './types'

export interface QvsPreviewConfirmation {
  qafFileId: string
  /** From the QvsImportPreview the client is confirming (preview.ts). */
  previewToken: string
  title: string
  description?: string | null
  /**
   * rowIndex values the user deselected in the preview. References
   * `VsmNode.qafSource.rowIndex`, NOT `node.id` — node ids are freshly
   * randomized every call, rowIndex is the stable identifier across a
   * preview→confirm round trip. P2 shipped the mechanism; P3 (PR #337) now
   * supplies the actual caller — `components/wertstrom/qvs-import-preview-
   * dialog.tsx`'s `handleCreate` builds this via `buildConfirmationPayload`
   * (qvs-preview-logic.ts) and passes it through
   * `createValueStreamFromQafAction`. The path stays unreached in any
   * current deployment only because `qafValueStream` is still `false` in
   * all 3 profiles (requireFlagEnabled in app/wertstrom/qaf-actions.ts),
   * not because there is no dialog anymore.
   */
  deselectedRowIndexes?: readonly number[]
}

export interface QvsCreationResult {
  valueStreamId: string
  importId: string
  /** true when this call hit an existing import for the same previewToken —
   * nothing new was written, the existing pair was returned as-is
   * (idempotent re-submit; see the RPC's ON CONFLICT handling). */
  idempotentHit: boolean
}

export type QvsCreationError =
  | { code: 'not_found' }
  | { code: 'title_required' }
  /** The qaf_file's step set changed between preview and confirm (replaced /
   * reparsed) — the caller must re-preview rather than confirm stale data. */
  | { code: 'stale_preview' }
  /** Every step was deselected — creating a 0-node "Wertstrom" from a QAF
   * import is a degenerate result, not a real import (E6 spirit: an import
   * record must reflect something that was actually imported). */
  | { code: 'no_steps_selected' }
  /** RPC rejected `p_imported_step_count`/`p_excluded_step_count` as null or
   * negative (migration's `invalid_step_count` guard). Not reachable via
   * this module's own app-sanctioned path — creation.ts always derives both
   * counts from real arrays (never negative, never null) — only via a
   * direct `.rpc()` call with tampered parameters. A permanent validation
   * failure, not a transient DB hiccup a retry would fix. */
  | { code: 'invalid_step_count' }
  /** RPC rejected a caller-supplied `p_import_id` that already exists
   * (migration's `import_id_conflict` guard). Not reachable via this
   * module's own app-sanctioned path — creation.ts always generates a fresh
   * `crypto.randomUUID()` — only via a direct `.rpc()` call with a
   * colliding id. A permanent conflict, not a transient DB hiccup a retry
   * would fix. */
  | { code: 'import_id_conflict' }
  | { code: 'db_error'; message: string }

export type QvsCreationOutcome = { ok: true; result: QvsCreationResult } | { ok: false; error: QvsCreationError }

/**
 * QVS-P4 (KAR-973, gap-analysis G5, ux-flow.md §4) — Multi-QAF variant
 * tagging for ONE `createValueStreamFromQaf` call. Optional, additive 3rd
 * parameter: omitted entirely, this function's behavior is byte-identical to
 * P2/P3 (variantTags never set on a node, `variant_selector` persisted as
 * `null`, exactly as before this PR).
 */
export interface QvsVariantTagging {
  /** Applied verbatim to every resulting node's `VsmNode.variantTags` —
   * 'shared_flow' passes ALL known variant keys (the flow is shared across
   * all of them), 'per_variant' passes exactly ONE. Never a fabricated
   * per-variant cycle time or step — only the tag differs (ux-flow.md §4:
   * "identische Schritte"). */
  variantTags: readonly string[]
  /** Persisted verbatim on `value_stream_imports.variant_selector` (schema
   * already models this shape, unused until this PR — see
   * schemas/qaf-value-stream-import.schema.json). */
  variantSelector: { strategy: 'shared_flow' | 'per_variant'; variantKeys: readonly string[] }
  /**
   * Appended to the PERSISTED `engine_context.previewToken` (the RPC's
   * idempotency dedup key, `value_stream_imports_preview_token_idx`) —
   * deliberately NOT to the staleness-check comparison (`confirmation.
   * previewToken` is still compared against the true, unmodified freshToken
   * recomputed from the source; see the staleness check below). Without
   * this, N `createValueStreamsForVariants` calls sharing the SAME source
   * preview would all persist the identical previewToken and the RPC's
   * partial-unique-index would silently collapse variants 2..N into
   * `idempotent_hit: true` pointing at variant 1's Wertstrom — a correctness
   * bug, not a real idempotent resubmit. Omit for the single shared_flow
   * call (only one call happens, no collision possible).
   *
   * KAR-974 reimport review-fix: the SAME mechanism is also needed by a
   * caller that is NOT tagging a variant at all (reimport.ts's
   * `createComparisonValueStreamFromSync` — a "separate comparison copy",
   * which must never collide with the ORIGINAL import's own persisted
   * token). Rather than force that caller to fabricate a fake
   * `QvsVariantTagging`, `createValueStreamFromQaf` also accepts a bare
   * `options.persistedTokenSuffix` (see its signature below) with no
   * variant-tagging side effects at all.
   */
  persistedTokenSuffix?: string
}

/** Defensive fallback for `value_stream_imports.parser_version` (NOT NULL) —
 * `qaf_file.parser_version` is nullable in the DB (pre-versioning-era rows),
 * though in practice every ingest path stamps it (see PARSER_VERSION usage
 * in app/qaf-differences/actions.ts). Never fabricates a real version
 * string; the fallback is legible as "not recorded", not a guessed value. */
const FALLBACK_PARSER_VERSION = 'unrecorded'

function relinkConnections(nodes: readonly VsmNode[]): VsmConnection[] {
  const connections: VsmConnection[] = []
  for (let i = 0; i < nodes.length - 1; i++) {
    connections.push({ id: crypto.randomUUID(), fromNodeId: nodes[i].id, toNodeId: nodes[i + 1].id })
  }
  return connections
}

// ── Review-Fixes 4/8/9: persistence-shape transforms ────────────────────────
// mapped.warnings / mapped.missingFieldReasons (P1, types.ts) are per-
// OCCURRENCE arrays — the right internal shape for the mapper, which has no
// notion of "the persisted audit record". schemas/qaf-value-stream-import.schema.json
// is the canonical contract for THAT record and requires a different shape
// for both fields (a `severity`-tagged, `stepRef`-named warning item; a
// Map-by-field-name for missing-field reasons). Code follows the schema —
// these functions are the one place that reshapes P1's internal types into
// the schema's persisted form, right before the RPC call. The P1 types
// themselves are NOT renamed (out of scope, would ripple into mapper.ts/
// manufacturing-capability.ts and their own tests for no benefit).

/** Per-code severity classification for the persisted `warnings` payload
 * (schema-required `severity` field, absent from the internal QvsWarning
 * type). A lookup table, not a blind default, so a future new QvsWarningCode
 * fails to compile here until someone deliberately assigns it a severity.
 * Both current codes are advisory, not hard failures: `sequence_ambiguous`
 * downgrades display confidence but the import still proceeds with a fully
 * usable Wertstrom; `no_eligible_rows` documents a degenerate/empty result
 * if it is ever threaded into a persisted record (today it never is — it is
 * a manufacturing-capability.ts-only warning, never part of MapperResult —
 * but QvsWarningCode already has both members, so both get a considered
 * assignment now rather than leaving one to a silent runtime fallback
 * later). Neither has its own hard stop at the RPC level (that role belongs
 * to the `no_steps_selected` / `invalid_step_count` error codes), so neither
 * is 'blocking_error'. */
const QVS_WARNING_SEVERITY: Record<QvsWarningCode, QvsWarningSeverity> = {
  sequence_ambiguous: 'warning',
  no_eligible_rows: 'warning',
}

/** P3 preview-UI lookup: the exact same severity a confirmed import would
 * persist (toPersistedWarning below), exposed read-only so the Preview
 * dialog (`QvsImportPreview.warnings`, still the plain `QvsWarning[]` shape)
 * can render a severity badge that can never drift from what `creation.ts`
 * actually writes for the same code — re-deriving a second table in the UI
 * layer would risk exactly that drift for a future new `QvsWarningCode`. */
export function severityForQvsWarningCode(code: QvsWarningCode): QvsWarningSeverity {
  return QVS_WARNING_SEVERITY[code]
}

/** QvsWarning -> the schema's persisted warning item: adds the required
 * `severity` (from the table above) and renames the positional field
 * `rowIndex` -> `stepRef` (schema name), stringified (schema types it as
 * `string | null`). `code`/`message` pass through unchanged. */
function toPersistedWarning(warning: QvsWarning): { code: string; severity: string; message: string; stepRef: string | null } {
  return {
    code: warning.code,
    severity: QVS_WARNING_SEVERITY[warning.code],
    message: warning.message,
    stepRef: warning.rowIndex !== undefined ? String(warning.rowIndex) : null,
  }
}

const NOT_NUMERIC_PREFIX = 'not_numeric['

/** The schema's `missing_field_reasons.*.reason` enum has a plain
 * `"not_numeric"` member (no interpolated text) — the mapper's
 * `not_numeric[<rawText>]` template-literal variant carries the actual
 * offending text for `rawTextSamples` instead (see below). Every other
 * MissingFieldReasonCode value is already schema-valid verbatim. */
function canonicalizeReason(reason: MissingFieldReason['reason']): string {
  return reason.startsWith(NOT_NUMERIC_PREFIX) ? 'not_numeric' : reason
}

/** Extracts the interpolated raw text from a `not_numeric[<rawText>]` reason,
 * or `null` for any other reason code. */
function extractRawText(reason: MissingFieldReason['reason']): string | null {
  if (!reason.startsWith(NOT_NUMERIC_PREFIX)) return null
  return reason.slice(NOT_NUMERIC_PREFIX.length, -1)
}

interface PersistedMissingFieldReason {
  reason: string
  count: number
  rawTextSamples?: string[]
}

/**
 * mapped.missingFieldReasons (one entry PER OCCURRENCE: `{rowIndex, nodeId,
 * field, targetField, reason}`) -> the schema's Map-by-field-name shape
 * (`{ [field]: { reason, count, rawTextSamples? } }`).
 *
 * Only occurrences whose `rowIndex` is in `includedRowIndexes` are counted —
 * same discipline as `imported_step_count`/`import_snapshot`: a reason
 * belonging to a step the user deselected has no place in the audit record
 * of what WAS actually imported (mirrors the Review-Fix 5 phantom-stepRef
 * discipline for `manual_preview_corrections`, applied here to a different
 * field). Pass `mapped.nodes`' rowIndexes (i.e. before deselection) to
 * intentionally include everything instead.
 *
 * The schema models exactly one `{reason, count}` pair per field, but real
 * QAF data can have MIXED reasons for one field across rows (e.g. some cells
 * genuinely blank, others carry an explicit "n.a." marker). When a field has
 * more than one distinct canonical reason, the most frequent one is kept as
 * the representative `reason` (ties broken by first-encountered, i.e. row
 * order — deterministic for a given input); `count` is always the TOTAL
 * occurrences for that field across all reasons, never just the
 * representative reason's share. This is a deliberate, documented
 * simplification of a many-reasons-per-field reality into the schema's
 * one-reason-per-field shape, not a fabrication: every number is real,
 * nothing is invented.
 */
function aggregateMissingFieldReasons(
  reasons: readonly MissingFieldReason[],
  includedRowIndexes: ReadonlySet<number>,
): Record<string, PersistedMissingFieldReason> {
  interface Bucket {
    countByReason: Map<string, number>
    rawTextSamples: string[]
  }
  const buckets = new Map<string, Bucket>()

  for (const r of reasons) {
    if (!includedRowIndexes.has(r.rowIndex)) continue
    const bucket: Bucket = buckets.get(r.field) ?? { countByReason: new Map<string, number>(), rawTextSamples: [] }
    const canonical = canonicalizeReason(r.reason)
    bucket.countByReason.set(canonical, (bucket.countByReason.get(canonical) ?? 0) + 1)
    const rawText = extractRawText(r.reason)
    // rawTextSamples: deduped, capped at the schema's `maxItems: 5` — never
    // fabricated (omitted from the output entirely when empty, see below).
    if (rawText && bucket.rawTextSamples.length < 5 && !bucket.rawTextSamples.includes(rawText)) {
      bucket.rawTextSamples.push(rawText)
    }
    buckets.set(r.field, bucket)
  }

  const out: Record<string, PersistedMissingFieldReason> = {}
  for (const [field, bucket] of buckets) {
    let bestReason = ''
    let bestCount = -1
    let total = 0
    for (const [reason, count] of bucket.countByReason) {
      total += count
      if (count > bestCount) {
        bestCount = count
        bestReason = reason
      }
    }
    out[field] = {
      reason: bestReason,
      count: total,
      ...(bucket.rawTextSamples.length > 0 ? { rawTextSamples: bucket.rawTextSamples } : {}),
    }
  }
  return out
}

export async function createValueStreamFromQaf(
  supabase: SupabaseClient,
  confirmation: QvsPreviewConfirmation,
  variantTagging?: QvsVariantTagging,
  /** Non-variant persisted-token-suffix escape hatch — see
   * `QvsVariantTagging.persistedTokenSuffix` doc. `variantTagging`'s OWN
   * suffix wins if somehow both are given (a variant call always supplies
   * its suffix via `variantTagging`, never via this parameter). */
  options?: { persistedTokenSuffix?: string },
): Promise<QvsCreationOutcome> {
  const title = confirmation.title?.trim()
  if (!title) return { ok: false, error: { code: 'title_required' } }

  const source = await loadQafSourceRows(supabase, confirmation.qafFileId)
  if (!source) return { ok: false, error: { code: 'not_found' } }

  // Real id, generated up front so it can be embedded into every node's
  // qafSource.importId (P1 contract: "FK-to-be for value_stream_imports.id")
  // AND handed to the RPC as the row it must actually insert under.
  const importId = crypto.randomUUID()
  const { mapped, previewToken: freshToken } = mapAndFingerprint(source, { importId })

  if (freshToken !== confirmation.previewToken) {
    return { ok: false, error: { code: 'stale_preview' } }
  }

  // Review-Fix 5: only rowIndexes that actually exist in the mapped nodes
  // count as "real" deselections — a client-supplied out-of-range index
  // (bug, stale UI state, or a tampered request) must not produce a phantom
  // `step_deselected` audit entry for a step that never existed. The
  // Set-based filter below was already tolerant of bogus indexes (they
  // simply match nothing, so remainingNodes was always correct); this makes
  // manual_preview_corrections equally honest by deriving it from the SAME
  // filtered list rather than the raw client input.
  const realRowIndexes = new Set(mapped.nodes.map((n) => n.qafSource!.rowIndex))
  const validDeselectedRowIndexes = (confirmation.deselectedRowIndexes ?? []).filter((rowIndex) => realRowIndexes.has(rowIndex))

  const deselected = new Set(validDeselectedRowIndexes)
  const remainingNodes = deselected.size > 0 ? mapped.nodes.filter((n) => !deselected.has(n.qafSource!.rowIndex)) : mapped.nodes

  if (remainingNodes.length === 0) {
    return { ok: false, error: { code: 'no_steps_selected' } }
  }

  // Relative source order is already preserved by the filter above — only
  // relink connections when the topology actually changed.
  const connections = deselected.size > 0 ? relinkConnections(remainingNodes) : mapped.connections

  // QVS-P4: tag every remaining node with the requested variant key(s) —
  // see QvsVariantTagging doc. No-op (nodes keep whatever variantTags the
  // mapper/editor already set, i.e. none for a fresh import) when omitted.
  if (variantTagging) {
    for (const node of remainingNodes) node.variantTags = [...variantTagging.variantTags]
  }

  const excludedStepCount = source.rows.length - remainingNodes.length
  const manualPreviewCorrections =
    validDeselectedRowIndexes.length > 0
      ? validDeselectedRowIndexes.map((rowIndex) => ({ kind: 'step_deselected', stepRef: String(rowIndex) }))
      : null

  // Review-Fix 4/8: missingFieldReasons is scoped to the steps that actually
  // ended up in remainingNodes (post-deselection) — same "describe what was
  // actually imported" discipline as imported_step_count/import_snapshot
  // above, and as validDeselectedRowIndexes just above for the opposite field.
  const includedRowIndexes = new Set(remainingNodes.map((n) => n.qafSource!.rowIndex))

  // QVS-P4/KAR-974: see QvsVariantTagging.persistedTokenSuffix doc — the
  // STALENESS check above already used the true, unmodified `freshToken`;
  // only the PERSISTED/dedup-key value differs per variant (or per
  // non-variant caller using the bare `options.persistedTokenSuffix`).
  const persistedTokenSuffix = variantTagging?.persistedTokenSuffix ?? options?.persistedTokenSuffix
  const persistedPreviewToken = persistedTokenSuffix ? `${freshToken}:${persistedTokenSuffix}` : freshToken

  const { data, error } = await supabase.rpc('create_value_stream_from_qaf', {
    p_qaf_file_id: confirmation.qafFileId,
    p_title: title,
    p_nodes: remainingNodes,
    p_connections: connections,
    p_parser_version: source.qafFile.parserVersion ?? FALLBACK_PARSER_VERSION,
    p_mapping_version: MAPPING_VERSION,
    p_imported_step_count: remainingNodes.length,
    p_excluded_step_count: excludedStepCount,
    p_import_id: importId,
    p_payload: {
      description: confirmation.description ?? null,
      engineContext: {
        previewToken: persistedPreviewToken,
        mappingVersion: MAPPING_VERSION,
        // QVS-P4: sum_planned_capacity/sum_lot_size (Datei-Ebene, always
        // present — null when the source didn't carry them, never omitted,
        // so a consumer never has to guess whether this run even looked).
        plannedCapacityPartsPerYear: source.qafFile.fileLevelContext.plannedCapacityPartsPerYear,
        lotSizeParts: source.qafFile.fileLevelContext.lotSizeParts,
      },
      warnings: mapped.warnings.map(toPersistedWarning),
      missingFieldReasons: aggregateMissingFieldReasons(mapped.missingFieldReasons, includedRowIndexes),
      importSnapshot: { nodes: remainingNodes, connections },
      manualPreviewCorrections,
      variantSelector: variantTagging?.variantSelector ?? null,
    },
  })

  if (error) {
    if (error.message?.includes('qaf_file_not_found')) return { ok: false, error: { code: 'not_found' } }
    if (error.message?.includes('title_required')) return { ok: false, error: { code: 'title_required' } }
    if (error.message?.includes('invalid_step_count')) return { ok: false, error: { code: 'invalid_step_count' } }
    if (error.message?.includes('import_id_conflict')) return { ok: false, error: { code: 'import_id_conflict' } }
    return { ok: false, error: { code: 'db_error', message: error.message } }
  }

  const row = data as { value_stream_id: string; import_id: string; idempotent_hit: boolean }
  return {
    ok: true,
    result: { valueStreamId: row.value_stream_id, importId: row.import_id, idempotentHit: row.idempotent_hit },
  }
}

/** One variant's own creation outcome — see createValueStreamsForVariants doc. */
export interface QvsPerVariantOutcome {
  variantKey: string
  outcome: QvsCreationOutcome
}

/**
 * Enough of `MultiQafVariantSummary` (multi-qaf-context.ts) to build a
 * per-variant title (Review-Fix 3) — a narrow structural slice rather than
 * an import of that module's full type, so this module does not need to
 * know about the "variant context" concept at all, only "a key and a
 * label". Any `MultiQafVariantSummary` satisfies this structurally.
 */
export interface QvsVariantForCreation {
  variantKey: string
  label: string
}

/**
 * Review-Fix 3 (KAR-973 adversarial review, MAJOR — N byte-identical
 * titles): every `createValueStreamsForVariants` call used to pass the SAME
 * `confirmation` (hence the same `confirmation.title`) to all N per-variant
 * creates, so the resulting `value_stream_maps` rows were indistinguishable
 * in the /wertstrom list (the only actually-differing data, `variantTags`,
 * is never rendered anywhere). Each variant's title is now suffixed with its
 * own label (`${confirmation.title} · ${variantLabel}`) — collision-safe:
 * falls back to `variantKey` when `label` is blank (defensive; multi-qaf-
 * context.ts's own `MultiQafVariantSummary.label` doc already guarantees it
 * is never empty, but this function never trusts that from a caller it does
 * not control).
 */
function perVariantTitle(confirmation: QvsPreviewConfirmation, variant: QvsVariantForCreation): QvsPreviewConfirmation {
  const variantLabel = variant.label.trim() !== '' ? variant.label.trim() : variant.variantKey
  return { ...confirmation, title: `${confirmation.title} · ${variantLabel}` }
}

/**
 * "Je Variante ein Wertstrom" (ux-flow.md §4, gap-analysis G5) — N
 * independent `createValueStreamFromQaf` calls against the SAME confirmed
 * preview, one per Multi-QAF container variant, each producing IDENTICAL
 * nodes/connections (same source steps — "Zykluszeiten sind im QAF nicht
 * variantenspezifisch", never fabricated per-variant step data) tagged with
 * exactly ONE variant key, and (Review-Fix 3) a per-variant title suffix so
 * the N resulting Wertströme stay distinguishable in the /wertstrom list.
 *
 * Honesty about atomicity (task-mandated, not swept under the rug): each
 * individual call IS its own transaction (the RPC's existing guarantee —
 * either that one Wertstrom+import row exists afterward or neither does),
 * but this function does NOT wrap all N calls in one bigger transaction
 * (that would need an RPC/migration change; P4 ships none). A failure on
 * variant k leaves variants 1..k-1 already persisted — this function reports
 * every variant's own outcome (success or failure) so a caller can show the
 * user exactly what happened, never a misleading all-or-nothing result.
 * Sequential (not parallel) on purpose: simpler failure semantics, and this
 * is a background dialog action, not a hot path.
 */
export async function createValueStreamsForVariants(
  supabase: SupabaseClient,
  confirmation: QvsPreviewConfirmation,
  variants: readonly QvsVariantForCreation[],
): Promise<QvsPerVariantOutcome[]> {
  const results: QvsPerVariantOutcome[] = []
  for (const variant of variants) {
    const outcome = await createValueStreamFromQaf(supabase, perVariantTitle(confirmation, variant), {
      variantTags: [variant.variantKey],
      variantSelector: { strategy: 'per_variant', variantKeys: [variant.variantKey] },
      persistedTokenSuffix: `variant:${variant.variantKey}`,
    })
    results.push({ variantKey: variant.variantKey, outcome })
  }
  return results
}