/**
 * Build-bus hook dispatcher service.
 *
 * Stateful wrapper around `planHookDispatch` + `runUpstreamHook`:
 * loads installed-module manifests from the celilo DB, dispatches
 * matching `on_upstream_publish` hooks when a verified PublishEvent
 * arrives, and records every run durably in the `build_bus_hook_runs`
 * ledger (celilo#1304) — without which a failed run exists only as
 * daemon console output, which no surface reads. Since
 * module-orchestrator-primitives slice 7 it also performs the FRAMEWORK
 * self-update on the same events (tasks.md 7.2): celilo replacing its
 * own binaries on the management host is class H work that stopped
 * being a module hook (design D6).
 *
 * Started by the `celilo subscribers serve` daemon (in-process,
 * sharing the same lifetime); the receiver-server calls
 * `dispatcher.handleEvent(event)` on every verified webhook.
 *
 * Module loading is injectable so tests can drive the dispatcher
 * with synthetic ModuleHookContext lists. Production loads from the
 * DB + filesystem (manifest.yml under each module's source_path).
 */

import { readFileSync } from 'node:fs';
import { join } from 'node:path';
import type { PublishEvent } from '@celilo/event-bus/build-bus';
import { eq } from 'drizzle-orm';
import { parse as parseYaml } from 'yaml';
import { getDb } from '../../db/client';
import { buildBusHookRuns, modules } from '../../db/schema';
import { CONTROL_PLANE_MODULE_ID } from '../deployed-systems';
import {
  type HookDispatchPlan,
  type ModuleHookContext,
  type UpstreamHookEntry,
  type UpstreamHookResult,
  defaultUpstreamHookDeps,
  planHookDispatch,
  runUpstreamHook,
} from './hook-dispatch';
import { type SelfUpdatePlan, planSelfUpdate, runSelfUpdate } from './self-update';

export interface HookDispatcher {
  /** Run the matching hooks for a single event. Returns the per-hook results. */
  handleEvent(event: PublishEvent): Promise<UpstreamHookResult[]>;
  /** Clean up (currently a no-op; included for symmetry with the receiver). */
  stop(): Promise<void>;
}

export interface HookDispatcherOptions {
  /**
   * Database for the default executor hop. Production omits it and the
   * process database opens lazily; tests inject an isolated client.
   */
  db?: Parameters<typeof defaultUpstreamHookDeps>[0];
  /**
   * Override the module-loading step. Tests inject a fixture; in
   * production we omit and the dispatcher reads from the celilo
   * modules table.
   */
  loadModules?: () => ModuleHookContext[];
  /**
   * Override the executor hop for one matched upstream-publish entry.
   * Production omits it; `runUpstreamHook` over the process database runs.
   */
  runUpstreamHook?: (plan: HookDispatchPlan, event: PublishEvent) => Promise<UpstreamHookResult>;
  /**
   * Override the durable outcome record. The production default writes one
   * `build_bus_hook_runs` row per run — without it a failed hook run or
   * self-update exists only as daemon console output, which no surface
   * reads (celilo#1304). Tests inject a capture to stay DB-free. The plan
   * is passed because the jailed executor result does not carry the
   * script path; the ledger's `script_path` column is NOT NULL.
   */
  recordRun?: (event: PublishEvent, plan: HookDispatchPlan, result: UpstreamHookResult) => void;
  /**
   * Override "is the control-plane module installed", the gate for the
   * framework self-update. Production reads the modules table; tests
   * inject the answer.
   */
  controlPlaneInstalled?: () => boolean;
  /** Override the self-update runner. Production: `runSelfUpdate`. */
  selfUpdate?: (plan: SelfUpdatePlan) => ReturnType<typeof runSelfUpdate>;
}

/** Bound on error tails so one chatty hook cannot bloat the ledger. */
const TAIL_CHARS = 2000;

function tail(s: string): string {
  return s.length > TAIL_CHARS ? s.slice(-TAIL_CHARS) : s;
}

/**
 * The production outcome record: one `build_bus_hook_runs` row per run.
 * Throwing here would 500 nothing (the webhook already succeeded) but
 * would kill dispatch of later plans, so a record failure is logged and
 * swallowed — the console outcome lines below remain the fallback signal.
 *
 * The jailed executor path (slice 7) does not expose exit codes or
 * captured stdio, so `exit_code` is always null — the column's "the
 * script never reached an exit" semantics now hold of the path itself —
 * and the executor's error text lands in `stderr_tail`.
 */
function recordHookRun(
  event: PublishEvent,
  plan: HookDispatchPlan,
  result: UpstreamHookResult,
): void {
  try {
    getDb()
      .insert(buildBusHookRuns)
      .values({
        eventId: event.eventId,
        packageName: event.package.name,
        packageVersion: event.package.version,
        tag: event.tag,
        moduleId: result.module,
        hookName: result.hookName,
        scriptPath: plan.hook.script,
        exitCode: null,
        timedOut: false,
        durationMs: result.durationMs,
        stdoutTail: null,
        stderrTail: result.error ? tail(result.error) : null,
      })
      .run();
  } catch (err) {
    console.warn(
      `[build-bus] could not record hook run outcome: ${err instanceof Error ? err.message : String(err)}`,
    );
  }
}

export async function startHookDispatcher(
  opts: HookDispatcherOptions = {},
): Promise<HookDispatcher> {
  const loadModules = opts.loadModules ?? loadInstalledModulesWithHooks;
  const runEntry =
    opts.runUpstreamHook ??
    ((plan, event) => runUpstreamHook(plan, event, defaultUpstreamHookDeps(opts.db)));
  const recordRun = opts.recordRun ?? recordHookRun;
  const isControlPlaneInstalled = opts.controlPlaneInstalled ?? defaultControlPlaneInstalled;
  const selfUpdateRunner = opts.selfUpdate ?? ((plan) => runSelfUpdate(plan));

  return {
    async handleEvent(event) {
      // Framework self-update first (design D6): celilo replacing celilo's
      // own binaries is not a module hook, and the dispatcher is the
      // component that already receives the publish event.
      const update = planSelfUpdate(event);
      if (update && isControlPlaneInstalled()) {
        console.log(
          `[build-bus] self-update: ${update.packageName}@${update.version} (@${event.tag})`,
        );
        const result = selfUpdateRunner(update);
        if (result.updated && result.verified) {
          console.log(
            `  ✓ ${update.binary} reports ${result.version} (was ${result.beforeVersion ?? 'not installed'})`,
          );
        } else {
          console.warn(`  ✗ self-update did not complete: ${result.error ?? 'unknown error'}`);
        }
      }

      const moduleContexts = loadModules();
      const plans = planHookDispatch(event, moduleContexts);
      if (plans.length === 0) {
        if (!update) {
          console.log(
            `[build-bus] ${event.package.name}@${event.package.version} (${event.tag}) — no module hooks match`,
          );
        }
        return [];
      }
      const results: UpstreamHookResult[] = [];
      for (const plan of plans) {
        console.log(
          `[build-bus] ${event.package.name}@${event.package.version} → ${plan.module.moduleId} (${plan.hook.name ?? plan.hook.script})`,
        );
        const result = await runEntry(plan, event);
        recordRun(event, plan, result);
        results.push(result);
        if (result.success) {
          console.log(`  ✓ success in ${result.durationMs}ms`);
        } else {
          console.warn(`  ✗ failed in ${result.durationMs}ms`);
          if (result.error) console.warn(`    ${result.error.trim().slice(0, 500)}`);
        }
      }
      return results;
    },
    async stop() {
      // No long-running connections to clean up — modules are
      // re-loaded per event so we can hot-pick up new installs.
    },
  };
}

/**
 * The framework self-update runs only where the control-plane module is
 * deployed — the same population celilo-mgmt's hook covered, since the
 * module is installed on the management host and nowhere else.
 */
function defaultControlPlaneInstalled(): boolean {
  const db = getDb();
  const row = db
    .select({ id: modules.id })
    .from(modules)
    .where(eq(modules.id, CONTROL_PLANE_MODULE_ID))
    .get();
  return row != null;
}

/**
 * Load every installed module's `on_upstream_publish` hooks from the
 * celilo DB + each module's manifest.yml. Modules without the hook
 * type are filtered out. Modules whose manifest fails to parse are
 * silently skipped — the operator sees this when they next run
 * `celilo module check`.
 */
function loadInstalledModulesWithHooks(): ModuleHookContext[] {
  const db = getDb();
  const rows = db.select({ id: modules.id, sourcePath: modules.sourcePath }).from(modules).all();
  const out: ModuleHookContext[] = [];
  for (const row of rows) {
    if (!row.sourcePath) continue;
    try {
      const parsed = parseYaml(readFileSync(join(row.sourcePath, 'manifest.yml'), 'utf-8')) as {
        hooks?: { on_upstream_publish?: UpstreamHookEntry[] };
      };
      const hooks = parsed.hooks?.on_upstream_publish ?? [];
      if (hooks.length === 0) continue;
      out.push({ moduleId: row.id, sourcePath: row.sourcePath, hooks });
    } catch {
      // Skip modules whose manifests can't be loaded — log when this
      // turns out to be a real problem rather than during routine
      // event dispatch.
    }
  }
  return out;
}