// Workflows in Spec Library + Activation Lens v0 — workflow scanner.
//
// SpecLibraryService knows how to classify rig + agent YAML files. This
// scanner adds workflow_specs as a third library kind by reading
// directly from the workflow_specs SQLite cache (PL-004 Phase D) +
// re-parsing each cached row to extract the per-step routing for the
// review payload's topology graph.
//
// The cache is the single source of truth for which workflow_specs the
// daemon has seen — both built-in starters (seeded at startup by
// loadStarterWorkflowSpecs) and
// operator-authored specs at workspace paths that the workspace-
// surface reconciliation contract has read through. Reading from the
// cache (rather than re-walking directories) means: no new env config,
// no duplicate parse logic, and the scanner stays consistent with
// `rig workflow specs` (same rows, same source-of-truth).
//
// Built-in detection: a row's `source_path` is "built in" iff it falls
// under the daemon's workflowBuiltinSpecsDir (the same `path.sep`
// boundary check used by /api/workflow/specs).

import * as path from "node:path";
import { existsSync, readdirSync, readFileSync, statSync } from "node:fs";
import { createHash } from "node:crypto";
import type Database from "better-sqlite3";
import type { WorkflowSpec, WorkflowExitKind, WorkflowAgentHarness, WorkflowGateSpec } from "./workflow-types.js";
import { RETAINED_STATUS, type WorkflowSpecCache } from "./workflow-spec-cache.js";
import type { EventBus } from "./event-bus.js";

export interface SpecLibraryWorkflowEntry {
  id: string;
  kind: "workflow";
  name: string;
  version: string;
  /** "builtin" when source_path is under workflowBuiltinSpecsDir; else "user_file". */
  sourceType: "builtin" | "user_file";
  sourcePath: string;
  relativePath: string;
  updatedAt: string;
  summary?: string;
  /** True iff sourcePath is under the daemon's bundled built-in dir. */
  isBuiltIn: boolean;
  /** Cheap counts surfaced in the row summary line. */
  rolesCount: number;
  stepsCount: number;
  terminalTurnRule: string;
  targetRig: string | null;
  /**
   * Slice 11 (workflow-spec-folder-discovery) — diagnostic state.
   * "valid" rows have parsed payload and are operable.
   * "error" rows came from a malformed YAML in the workflows folder
   * scan; Library UI renders them with error styling and surfaces
   * the errorMessage so the operator can fix the file in place.
   */
  status: "valid" | "error";
  /** Populated only when status === "error" — concrete parse/validate diagnostic. */
  errorMessage: string | null;
}

export interface SpecLibraryWorkflowReview {
  kind: "workflow";
  name: string;
  version: string;
  purpose: string | null;
  targetRig: string | null;
  terminalTurnRule: string;
  rolesCount: number;
  stepsCount: number;
  isBuiltIn: boolean;
  sourcePath: string;
  cachedAt: string;
  /** Topology graph projection: nodes from `roles`, edges derived from
   *  each step's next_hop.suggested_roles → next-step ids. Same shape
   *  Slice Story View v1 uses for its Topology tab — consumers reuse
   *  the same UI primitives for rendering.
   *
   *  OPR.0.4.6.WF4 Q1 (arch-ruled): the projection ADDS branch edges
   *  from `next_hop.on` (routingType 'branch' + the triggering exit),
   *  corrects the false-terminal defect (a step with on-targets is not
   *  terminal), and projects optional harness/host/gate node fields —
   *  all byte-identity-by-omission: a suggested-roles-only spec with no
   *  on/harness/host/gate projects EXACTLY as before (no 'branch' edge,
   *  no branchOn key, no optional node key). */
  topology: {
    nodes: Array<{
      stepId: string;
      role: string;
      objective: string | null;
      preferredTarget: string | null;
      isEntry: boolean;
      isTerminal: boolean;
      /** OPR.0.4.6.WF4 Q1-P1 — OMITTED when the step declares none. */
      harness?: WorkflowAgentHarness;
      host?: string;
      gate?: WorkflowGateSpec;
    }>;
    edges: Array<{
      fromStepId: string;
      toStepId: string;
      /** OPR.0.4.6.WF4 Q1-P2 — closed-vocabulary extension of 'direct'. */
      routingType: "direct" | "branch";
      /** The triggering exit; present on 'branch' edges ONLY. */
      branchOn?: WorkflowExitKind;
    }>;
  };
  /** Per-step list rendered below the graph. */
  steps: Array<{
    stepId: string;
    role: string;
    objective: string | null;
    allowedExits: string[];
    /** Resolved destinations from next_hop.suggested_roles → step ids. */
    allowedNextSteps: Array<{ stepId: string; role: string }>;
  }>;
}

export interface ScanWorkflowSpecsOpts {
  db: Database.Database;
  /** Absolute path to the daemon's bundled builtin starter dir;
   *  null/undefined → no isBuiltIn detection (all rows render as user_file). */
  workflowBuiltinSpecsDir: string | null;
}

interface SpecRow {
  spec_id: string;
  name: string;
  version: string;
  purpose: string | null;
  target_rig: string | null;
  roles_json: string;
  steps_json: string;
  coordination_terminal_turn_rule: string;
  source_path: string;
  source_hash: string;
  cached_at: string;
  // Slice 11 — diagnostic columns from migration 040. May be missing
  // when the test harness applies only the 033 schema; treat absence
  // as status='valid', error_message=null to preserve back-compat.
  status?: string;
  error_message?: string | null;
}

export function scanWorkflowSpecs(opts: ScanWorkflowSpecsOpts): SpecLibraryWorkflowEntry[] {
  let rows: SpecRow[] = [];
  try {
    rows = opts.db.prepare(
      `SELECT * FROM workflow_specs ORDER BY name, version`,
    ).all() as SpecRow[];
  } catch {
    // workflow_specs table absent (test harness without the migration):
    // empty library. Same graceful degradation as Slice Story View
    // v0's slice indexer.
    return [];
  }

  const out: SpecLibraryWorkflowEntry[] = [];
  for (const row of rows) {
    // A version kept only for unfinished work after its file was deleted is not discoverable (#511).
    if (row.status === RETAINED_STATUS) continue;
    const status: "valid" | "error" = row.status === "error" ? "error" : "valid";
    const isBuiltIn = opts.workflowBuiltinSpecsDir
      ? isUnderDir(row.source_path, opts.workflowBuiltinSpecsDir)
      : false;

    // Slice 11 — diagnostic rows render without parsed payload. Use the
    // file basename as the row label (already stored in name by
    // writeDiagnostic) and zero counts; errorMessage carries the reason.
    if (status === "error") {
      out.push({
        // Diagnostic rows don't have stable name+version (version is
        // empty when YAML couldn't be parsed); the library id falls
        // back to source_path so the UI can route uniquely.
        id: `workflow:error:${row.source_path}`,
        kind: "workflow",
        name: row.name,
        version: row.version,
        sourceType: isBuiltIn ? "builtin" : "user_file",
        sourcePath: row.source_path,
        relativePath: row.source_path,
        updatedAt: row.cached_at,
        summary: row.error_message ?? undefined,
        isBuiltIn,
        rolesCount: 0,
        stepsCount: 0,
        terminalTurnRule: row.coordination_terminal_turn_rule || "hot_potato",
        targetRig: null,
        status: "error",
        errorMessage: row.error_message ?? null,
      });
      continue;
    }

    let roles: WorkflowSpec["roles"];
    let steps: WorkflowSpec["steps"];
    try {
      roles = JSON.parse(row.roles_json) as WorkflowSpec["roles"];
      steps = JSON.parse(row.steps_json) as WorkflowSpec["steps"];
    } catch {
      // Malformed JSON in cache — skip with no entry; the daemon's
      // /api/workflow/specs surface will surface the row anyway.
      continue;
    }
    out.push({
      // Stable id derived from name+version so the SpecLibrary's review
      // endpoint can resolve workflow entries the same way it resolves
      // rig/agent entries by id.
      id: workflowLibraryId(row.name, row.version),
      kind: "workflow",
      name: row.name,
      version: row.version,
      sourceType: isBuiltIn ? "builtin" : "user_file",
      sourcePath: row.source_path,
      relativePath: row.source_path,
      updatedAt: row.cached_at,
      summary: row.purpose ?? undefined,
      isBuiltIn,
      rolesCount: Object.keys(roles ?? {}).length,
      stepsCount: Array.isArray(steps) ? steps.length : 0,
      terminalTurnRule: row.coordination_terminal_turn_rule || "hot_potato",
      targetRig: row.target_rig,
      status: "valid",
      errorMessage: null,
    });
  }
  return out;
}

export function getWorkflowReview(opts: ScanWorkflowSpecsOpts & { name: string; version: string }): SpecLibraryWorkflowReview | null {
  let row: SpecRow | undefined;
  try {
    row = opts.db.prepare(
      `SELECT * FROM workflow_specs WHERE name = ? AND version = ?`,
    ).get(opts.name, opts.version) as SpecRow | undefined;
  } catch {
    return null;
  }
  if (!row || row.status === RETAINED_STATUS) return null;

  let roles: WorkflowSpec["roles"];
  let steps: WorkflowSpec["steps"];
  try {
    roles = JSON.parse(row.roles_json) as WorkflowSpec["roles"];
    steps = JSON.parse(row.steps_json) as WorkflowSpec["steps"];
  } catch {
    return null;
  }

  const isBuiltIn = opts.workflowBuiltinSpecsDir
    ? isUnderDir(row.source_path, opts.workflowBuiltinSpecsDir)
    : false;

  // Project the topology — same shape as Slice Story View v1 uses.
  const stepByRole = new Map<string, typeof steps[0]>();
  for (const step of steps ?? []) {
    if (!stepByRole.has(step.actor_role)) stepByRole.set(step.actor_role, step);
  }
  const entryRole = (row as unknown as { entry_role?: string }).entry_role
    ?? steps?.[0]?.actor_role;
  const entryStepId = entryRole ? stepByRole.get(entryRole)?.id : undefined;

  const stepIds = new Set((steps ?? []).map((s) => s.id));
  const topologyNodes = (steps ?? []).map((step) => {
    const roleSpec = (roles as Record<string, { preferred_targets?: string[] }>)[step.actor_role] ?? {};
    // OPR.0.4.6.WF4 Q1-P3: a step with next_hop.on targets is NOT terminal
    // (the false-terminal defect for branch-only steps like build/verify).
    const hasOnTargets = Object.keys(step.next_hop?.on ?? {}).length > 0;
    return {
      stepId: step.id,
      role: step.actor_role,
      objective: step.objective ?? null,
      preferredTarget: roleSpec.preferred_targets?.[0] ?? null,
      isEntry: step.id === entryStepId,
      isTerminal: !(step.next_hop?.suggested_roles?.length) && !hasOnTargets,
      // OPR.0.4.6.WF4 Q1-P1: optional node fields, OMITTED when absent
      // (byte-identity-by-omission — a step with none adds no key).
      ...(step.harness ? { harness: step.harness } : {}),
      ...(step.host ? { host: step.host } : {}),
      ...(step.gate ? { gate: step.gate } : {}),
    };
  });

  const topologyEdges: SpecLibraryWorkflowReview["topology"]["edges"] = [];
  for (const step of steps ?? []) {
    for (const role of step.next_hop?.suggested_roles ?? []) {
      const target = stepByRole.get(role);
      if (!target) continue;
      topologyEdges.push({ fromStepId: step.id, toStepId: target.id, routingType: "direct" });
    }
    // OPR.0.4.6.WF4 Q1-P2: branch edges from next_hop.on (exit → successor
    // STEP ID). Absent `on` pushes nothing ⇒ suggested-roles-only specs
    // project a byte-identical edge array. The exit label rides `branchOn`.
    for (const [exit, targetStepId] of Object.entries(step.next_hop?.on ?? {})) {
      if (!targetStepId || !stepIds.has(targetStepId)) continue;
      topologyEdges.push({
        fromStepId: step.id,
        toStepId: targetStepId,
        routingType: "branch",
        branchOn: exit as WorkflowExitKind,
      });
    }
  }

  const stepDetails: SpecLibraryWorkflowReview["steps"] = (steps ?? []).map((step) => ({
    stepId: step.id,
    role: step.actor_role,
    objective: step.objective ?? null,
    allowedExits: [...(step.allowed_exits ?? [])],
    allowedNextSteps: (step.next_hop?.suggested_roles ?? [])
      .map((role) => {
        const target = stepByRole.get(role);
        return target ? { stepId: target.id, role } : null;
      })
      .filter((x): x is { stepId: string; role: string } => x !== null),
  }));

  return {
    kind: "workflow",
    name: row.name,
    version: row.version,
    purpose: row.purpose,
    targetRig: row.target_rig,
    terminalTurnRule: row.coordination_terminal_turn_rule || "hot_potato",
    rolesCount: Object.keys(roles ?? {}).length,
    stepsCount: (steps ?? []).length,
    isBuiltIn,
    sourcePath: row.source_path,
    cachedAt: row.cached_at,
    topology: { nodes: topologyNodes, edges: topologyEdges },
    steps: stepDetails,
  };
}

export function workflowLibraryId(name: string, version: string): string {
  // Legacy IDs round-trip every name, but only versions without a colon.
  // An opaque tuple keeps colon-bearing versions distinct without changing
  // any already-unambiguous ID. Legacy valid IDs always contain another colon.
  if (version.includes(":")) return `workflow:@${Buffer.from(JSON.stringify([name, version])).toString("base64url")}`;
  return `workflow:${name}:${version}`;
}

export function parseWorkflowLibraryId(id: string): { name: string; version: string } | null {
  if (!id.startsWith("workflow:")) return null;
  const rest = id.slice("workflow:".length);
  if (rest.startsWith("@") && !rest.includes(":")) {
    try {
      const encoded = rest.slice(1);
      const bytes = Buffer.from(encoded, "base64url");
      if (!encoded || bytes.toString("base64url") !== encoded) return null;
      const pair: unknown = JSON.parse(bytes.toString("utf8"));
      if (!Array.isArray(pair) || pair.length !== 2 || pair.some(value => typeof value !== "string")) return null;
      return { name: pair[0], version: pair[1] };
    } catch { return null; }
  }
  // Preserve legacy names (including literal percent sequences and colons).
  const lastColon = rest.lastIndexOf(":");
  if (lastColon === -1) return null;
  return { name: rest.slice(0, lastColon), version: rest.slice(lastColon + 1) };
}

function isUnderDir(childPath: string, parentDir: string): boolean {
  const child = path.resolve(childPath);
  const parent = path.resolve(parentDir);
  if (child === parent) return false;
  const parentWithSep = parent.endsWith(path.sep) ? parent : `${parent}${path.sep}`;
  return child.startsWith(parentWithSep);
}

// =================================================================
// Slice 11 (release-0.3.1 workflow-spec-folder-discovery)
// =================================================================
//
// scanWorkflowSpecFolder — filesystem walk that turns workspace.specs_root/
// workflows/ into an installable user primitive. The Library route calls
// this opportunistically on each list request (OQ-3 decision); valid YAML
// gets cached via WorkflowSpecCache.readThrough (existing path), invalid
// YAML gets cached via writeDiagnostic (slice 11 path), and files that
// disappear since the last scan get removed via removeBySourcePath
// (OQ-4 decision; deletion + audit log).
//
// OQ-3 timestamp check: a file is skipped when both its mtime/ctime are <=
// the cache row's cached_at. A file can be restored or copied with an older
// mtime, so ctime keeps the stat-only fast path honest. Files in the same
// timestamp second still use the content hash, and other newer files are
// re-parsed via the cache (which itself hashes the content and returns the
// prior row when the hash matches).

export interface ScanWorkflowSpecFolderOpts {
  /** SQLite handle (used for direct lookups). */
  db: Database.Database;
  /** Cache handle for readThrough / writeDiagnostic / removeBySourcePath. */
  cache: WorkflowSpecCache;
  /** Absolute path to the workspace's workflows folder
   *  (typically `<workspace.specs_root>/workflows`). Missing folder
   *  → empty scan summary; not an error. */
  folder: string;
  /** Daemon's bundled builtin starter dir; used to skip removal logic
   *  for built-in rows (the scanner only owns the folder it walks). */
  builtinDir: string | null;
  /** Optional EventBus — when wired, the scanner emits a
   *  workflow_spec.removed audit event for each cache row removed because
   *  its source file disappeared (OQ-4 acceptance: deletion + audit log).
   *  Omitted in unit tests that don't care about emission. */
  eventBus?: EventBus;
  /** A user file with the same bytes and identity as an existing built-in row. */
  onBuiltinAlias?: (name: string, version: string, sourcePath: string, sourceHash: string) => void;
}

export interface ScanWorkflowSpecFolderResult {
  /** Total YAML/YML files found in folder. */
  scanned: number;
  /** Files parsed + cached successfully on this scan. */
  valid: number;
  /** Files that failed parse/validate; recorded as diagnostic rows. */
  errors: number;
  /** Cache rows removed because their source_path no longer exists. */
  removed: number;
  /** Files skipped via timestamp check (unchanged since last scan). */
  skipped: number;
}

function isWorkflowYamlFile(name: string): boolean {
  return /\.ya?ml$/i.test(name);
}

export function scanWorkflowSpecFolder(
  opts: ScanWorkflowSpecFolderOpts,
): ScanWorkflowSpecFolderResult {
  const result: ScanWorkflowSpecFolderResult = {
    scanned: 0,
    valid: 0,
    errors: 0,
    removed: 0,
    skipped: 0,
  };
  if (!existsSync(opts.folder)) return result;

  let entries: string[] = [];
  try {
    entries = readdirSync(opts.folder);
  } catch {
    return result;
  }

  const seenPaths = new Set<string>();
  for (const entry of entries) {
    if (!isWorkflowYamlFile(entry)) continue;
    const filePath = path.join(opts.folder, entry);
    let changeMs = 0;
    try {
      const stat = statSync(filePath);
      if (!stat.isFile()) continue;
      changeMs = Math.max(stat.mtimeMs, stat.ctimeMs);
    } catch {
      continue;
    }
    seenPaths.add(filePath);
    result.scanned += 1;

    // Timestamp check (OQ-3): ctime protects the stat-only skip when a
    // restored or copied file has an older mtime even though its content
    // changed. Only the same timestamp second needs a content-hash check.
    //
    // Compare at second-resolution because some filesystems (HFS+, FAT)
    // round mtime/ctime down to whole seconds while cached_at carries ms
    // precision; without this floor a freshly-written file whose change
    // time is `T - 999ms` would always look "newer" than its cached_at at
    // exactly `T` and never skip.
    // A path can hold several cached versions plus a diagnostic row (#503); the most
    // recently written row is the one that reflects the file's last scan.
    const cachedAt = opts.db
      .prepare(`SELECT cached_at, source_hash FROM workflow_specs WHERE source_path = ? AND status != '${RETAINED_STATUS}' ORDER BY cached_at DESC, rowid DESC LIMIT 1`)
      .get(filePath) as { cached_at: string; source_hash: string } | undefined;
    if (cachedAt) {
      const cachedAtMs = Date.parse(cachedAt.cached_at);
      const cachedAtSec = Math.floor(cachedAtMs / 1000);
      const changeSec = Math.floor(changeMs / 1000);
      if (Number.isFinite(cachedAtMs) && Number.isFinite(changeMs)) {
        if (cachedAtSec > changeSec) {
          result.skipped += 1;
          continue;
        }
        if (cachedAtSec === changeSec) {
          // Consult the cache's existing content hash before treating the file
          // as unchanged. Hash the UTF-8 text exactly the way the cache
          // hashes it on write (not the raw bytes): for files that are not
          // clean UTF-8 the two digests differ and every scan re-parses.
          let unchanged = false;
          try {
            unchanged = createHash("sha256").update(readFileSync(filePath, "utf-8")).digest("hex") === cachedAt.source_hash;
          } catch { /* The existing readThrough path records the diagnostic. */ }
          if (unchanged) {
            result.skipped += 1;
            continue;
          }
        }
      }
    }

    // Parse + validate via cache. If readThrough throws (parse or
    // validation error), record a diagnostic row keyed by source_path
    // so the Library UI can render the error inline.
    try {
      const cached = opts.cache.readThrough(filePath);
      if (opts.onBuiltinAlias && opts.builtinDir && cached.sourcePath !== filePath
          && isUnderDir(cached.sourcePath, opts.builtinDir)) {
        opts.onBuiltinAlias(cached.name, cached.version, filePath, cached.sourceHash);
      }
      result.valid += 1;
    } catch (err) {
      // Diagnostic: hash the raw content best-effort so we can detect
      // edits that fix the error (mtime alone is fine; this is for
      // bookkeeping). Use the empty string when content is unreadable.
      const message = err instanceof Error ? err.message : String(err);
      let sourceHash = "";
      try {
        sourceHash = createHash("sha256").update(readFileSync(filePath, "utf-8")).digest("hex");
      } catch {
        // unreadable / disappeared between stat and read; treat as
        // empty hash so the next scan re-evaluates
        sourceHash = "";
      }
      try {
        opts.cache.writeDiagnostic({
          sourcePath: filePath,
          sourceHash,
          errorMessage: message,
        });
      } catch (writeErr) {
        // One file's diagnostic must not end the scan of the rest of the folder (#511).
        console.error(`workflow spec scan: could not record the parse error for ${filePath}: ${writeErr instanceof Error ? writeErr.message : String(writeErr)}`);
      }
      result.errors += 1;
    }
  }

  // OQ-4 deletion: cache rows whose source_path lives under the scanned
  // folder AND whose file is no longer present on disk get removed. We
  // scope the removal by source_path prefix to avoid touching built-in
  // rows or rows the scanner doesn't own (e.g., other workflows folders).
  const folderPrefix = opts.folder.endsWith(path.sep) ? opts.folder : `${opts.folder}${path.sep}`;
  const cachedUnderFolder = opts.db
    .prepare(
      `SELECT source_path, spec_id, name, version FROM workflow_specs WHERE source_path LIKE ?`,
    )
    .all(`${folderPrefix}%`) as Array<{
      source_path: string;
      spec_id: string | null;
      name: string | null;
      version: string | null;
    }>;
  for (const row of cachedUnderFolder) {
    // LIKE treats _ and % in real folder names as patterns. Only this
    // literal directory owns deletion; neighboring cache rows must survive.
    if (!row.source_path.startsWith(folderPrefix)) continue;
    if (seenPaths.has(row.source_path)) continue;
    if (!row.spec_id) continue;
    // #511: a version that unfinished work is pinned to stays readable until that work ends;
    // a later scan removes it. Every other row of the vanished file goes, one event per version.
    const identity = opts.cache.storedIdentity(row.spec_id);
    if (identity && opts.cache.isPinnedByUnfinishedWork(identity.name, identity.version)) {
      opts.cache.retain(row.spec_id);
      continue;
    }
    const removed = opts.cache.removeBySpecId(row.spec_id);
    if (removed > 0) {
      result.removed += removed;
      // OQ-4 audit-log: record the disappearance so operators can trace
      // which spec was reaped when. Emission failures must not abort the
      // scan — the cache row is already gone and the next scan would be
      // a no-op for this path.
      if (opts.eventBus) {
        try {
          opts.eventBus.emit({
            type: "workflow_spec.removed",
            sourcePath: row.source_path,
            specId: row.spec_id ?? null,
            specName: row.name ?? null,
            specVersion: row.version ?? null,
            reason: "file_disappeared",
          });
        } catch { /* best-effort audit emission */ }
      }
    }
  }

  return result;
}