/**
 * Wire a module's manifest `subscriptions:` block into the SQLite event
 * bus. Called from `module/import.ts` on install and from
 * `cli/commands/module-remove.ts` on remove.
 *
 * Substitutions performed at subscribe time:
 *   - `$self` in `pattern`    → the module's id
 *   - `${MODULE_PATH}` in `handler` → the module's installed targetPath
 *   - a `hook:` subscription   → a synthesized `celilo events run-hook
 *     <module> <sub-name>` handler (openspec/specs/event-driven-hook-subscriptions/spec.md)
 *
 * The bus subscriber's name is namespaced as `<module-id>.<sub-name>`
 * so two modules can declare a subscription named `smoke` without
 * colliding.
 */

import { join } from 'node:path';
import { defineEvents, openBus } from '@celilo/event-bus';
import { inArray } from 'drizzle-orm';
import { getEventBusPath, getModuleStoragePath } from '../config/paths';
import { getDb } from '../db/client';
import { modules } from '../db/schema';
import type { ModuleManifest, ModuleSubscription } from '../manifest/schema';
import { ensureBackupSweepSubscriber } from './backup-sweep';
import { ensureOperationsSweepSubscriber } from './module-operations';

/**
 * The bus is opened by the celilo CLI without an event registry — the
 * CLI doesn't know the schemas of every module's events. The bus's
 * empty-registry path skips payload validation, leaving that to the
 * linked handlers (which open the bus *with* their own registry).
 */
const NO_SCHEMAS = defineEvents({});

/**
 * Resolve the per-module substitutions on a single subscription. Pure
 * function: takes a parsed manifest entry, returns the bus-shaped
 * subscribe options.
 */
export function resolveSubscription(
  sub: ModuleSubscription,
  moduleId: string,
  modulePath: string,
): {
  name: string;
  pattern: string;
  handler: string;
  maxAttempts?: number;
  timeoutMs?: number;
  registeredBy: string;
} {
  return {
    name: scopedName(moduleId, sub.name),
    pattern: substituteSelf(sub.pattern, moduleId),
    handler: resolveHandler(sub, moduleId, modulePath),
    maxAttempts: sub.max_attempts,
    timeoutMs: sub.timeout_ms,
    registeredBy: moduleId,
  };
}

/**
 * The bus handler string for a subscription. A `hook:` subscription becomes a
 * synthesized `celilo events run-hook <module> <sub-name>` invocation — the
 * generic runner re-reads this module's manifest by (module, sub-name) to find
 * the hook + its `hook_inputs`, then runs it in a fault-isolated subprocess
 * with backend access. A `handler:` subscription is the literal command with
 * `${MODULE_PATH}` resolved. The schema guarantees exactly one is set; the
 * final throw is defense-in-depth (no surprises).
 */
function resolveHandler(sub: ModuleSubscription, moduleId: string, modulePath: string): string {
  if (sub.hook) {
    return `celilo events run-hook ${moduleId} ${sub.name}`;
  }
  if (sub.handler) {
    return substituteModulePath(sub.handler, modulePath);
  }
  throw new Error(
    `subscription '${sub.name}' on module '${moduleId}' declares neither 'handler' nor 'hook'`,
  );
}

/**
 * Register all of a module's subscriptions on the bus. Idempotent —
 * re-running with the same manifest updates existing rows in place.
 */
export function registerModuleSubscriptions(
  manifest: ModuleManifest,
  modulePath: string,
): { registered: number } {
  const subs = manifest.subscriptions ?? [];
  const backupSweep = Boolean(manifest.hooks?.on_backup);

  const bus = openBus({ dbPath: getEventBusPath(), events: NO_SCHEMAS });
  try {
    // Unconditional: every module can hold the operation lock (a deploy at
    // minimum), so the first module on a fleet is what arms the sweep that
    // reclaims abandoned rows. Idempotent.
    ensureOperationsSweepSubscriber(bus);

    // A module that can be backed up is also what switches the scheduled
    // backup sweep on. Registering here rather than at system init means the
    // sweep appears the moment the fleet has something to back up, and — since
    // `module update` comes through here too — a manifest that newly declares a
    // cadence arms the sweep on the same update that declares it. Idempotent.
    if (backupSweep) ensureBackupSweepSubscriber(bus);

    for (const sub of subs) {
      const resolved = resolveSubscription(sub, manifest.id, modulePath);
      bus.subscribe(resolved);
    }
    return { registered: subs.length };
  } finally {
    bus.close();
  }
}

/**
 * Tear down every bus subscription that belongs to this module. Looks
 * up rows by name prefix `<module-id>.` rather than rereading the
 * old manifest, so a stale subscription left behind by a manifest
 * change still gets cleaned up.
 *
 * Best-effort: if the bus DB doesn't exist (never opened), returns 0.
 */
export function unregisterModuleSubscriptions(moduleId: string): {
  unregistered: number;
} {
  const bus = openBus({ dbPath: getEventBusPath(), events: NO_SCHEMAS });
  try {
    const likePattern = `${moduleId}.%`;
    const rows = bus.db
      .query<{ name: string }, [string]>('SELECT name FROM subscribers WHERE name LIKE ?')
      .all(likePattern);
    for (const row of rows) {
      bus.unsubscribe(row.name);
    }
    return { unregistered: rows.length };
  } finally {
    bus.close();
  }
}

/**
 * Rebuild the event-bus `subscribers` table from every deployed module's
 * manifest `subscriptions:`. The reactive layer (who-reacts-to-what) is
 * registered at module DEPLOY time and lives in the bus (events.db), which a
 * restore/migration starts EMPTY — so after a cutover, event-driven reconciles
 * (caddy public_web, dns register, …) are dead until every module redeploys
 * (ISS-0088). This reconstructs them from durable celilo.db state instead.
 *
 * Idempotent (registerModuleSubscriptions upserts by subscriber name). Per-module
 * failures are collected, not thrown, so one bad manifest doesn't abort the rest.
 */
export function resyncAllSubscriptions(): {
  modules: number;
  registered: number;
  failures: Array<{ moduleId: string; error: string }>;
} {
  const db = getDb();
  const deployed = db
    .select()
    .from(modules)
    .where(inArray(modules.state, ['INSTALLED', 'VERIFIED']))
    .all();

  let modulesWithSubs = 0;
  let registered = 0;
  const failures: Array<{ moduleId: string; error: string }> = [];

  for (const mod of deployed) {
    const manifest = mod.manifestData as unknown as ModuleManifest;
    if (!manifest?.subscriptions?.length) continue;
    // Code always lives at ${getModuleStoragePath()}/<id> by construction
    // (import.ts) — the same path module-upgrade re-registers from (ISS-0091).
    const modulePath = join(getModuleStoragePath(), mod.id);
    try {
      const result = registerModuleSubscriptions(manifest, modulePath);
      registered += result.registered;
      modulesWithSubs += 1;
    } catch (err) {
      failures.push({ moduleId: mod.id, error: err instanceof Error ? err.message : String(err) });
    }
  }

  return { modules: modulesWithSubs, registered, failures };
}

function scopedName(moduleId: string, subName: string): string {
  return `${moduleId}.${subName}`;
}

function substituteSelf(pattern: string, moduleId: string): string {
  // `$self` matches when followed by a dot or end-of-string, so a
  // pattern like `deploy.$self.foo` substitutes correctly without
  // confusing `$selfish` if anyone wrote that. (No real reason they
  // would, but be precise.)
  return pattern.replace(/\$self(?=\.|$)/g, moduleId);
}

function substituteModulePath(handler: string, modulePath: string): string {
  return handler.replace(/\$\{MODULE_PATH\}/g, modulePath);
}