import { afterEach, beforeEach, describe, expect, it } from 'bun:test';
import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { dirname, join, resolve } from 'node:path';
import { fileURLToPath } from 'node:url';
import { defineEvents, openBus } from '@celilo/event-bus';
import { parse as parseYaml } from 'yaml';
import { closeDb, getDb } from '../db/client';
import { ModuleManifestSchema } from '../manifest/schema';
import { ModuleSubscriptionSchema } from '../manifest/schema';
import type { ModuleManifest } from '../manifest/schema';
import { setupTestDatabaseAt as migrateDbFile } from '../test-utils/database';
import { resetTestDbPath } from '../test-utils/db-path';
import {
  registerModuleSubscriptions,
  resolveSubscription,
  resyncAllSubscriptions,
  unregisterModuleSubscriptions,
} from './module-subscriptions';

const baseManifest = (overrides: Partial<ModuleManifest> = {}): ModuleManifest => ({
  celilo_contract: '1.0',
  id: 'lunacycle',
  name: 'Lunacycle',
  version: '1.0.0',
  requires: { capabilities: [] },
  provides: { capabilities: [] },
  variables: { owns: [], imports: [] },
  ...overrides,
});

describe('resolveSubscription', () => {
  it('substitutes $self in pattern and ${MODULE_PATH} in handler', () => {
    const resolved = resolveSubscription(
      {
        name: 'smoke-after-deploy',
        pattern: 'deploy.completed.$self',
        handler: 'bun ${MODULE_PATH}/celilo/scripts/smoke.ts',
      },
      'lunacycle',
      '/var/lib/celilo/modules/lunacycle',
    );

    expect(resolved.name).toBe('lunacycle.smoke-after-deploy');
    expect(resolved.pattern).toBe('deploy.completed.lunacycle');
    expect(resolved.handler).toBe('bun /var/lib/celilo/modules/lunacycle/celilo/scripts/smoke.ts');
    expect(resolved.registeredBy).toBe('lunacycle');
  });

  // The other half of the invariant `checkSubscribers` classifies by (#624):
  // core rows are not scoped under their registrar id, module rows always are.
  // Stop scoping the name here and every module row reads as core-owned — the
  // stale check would go silently blind rather than noisily wrong.
  it('always scopes the name under the id it stamps in registeredBy', () => {
    const resolved = resolveSubscription({ name: 'a', pattern: 'x', handler: 'echo' }, 'foo', '/p');
    expect(resolved.name.startsWith(`${resolved.registeredBy}.`)).toBe(true);
  });

  it('only substitutes $self when followed by . or end-of-string', () => {
    // `$selfish` would not be a real pattern but we want to ensure the
    // substitution doesn't accidentally rewrite identifier-like names.
    const resolved = resolveSubscription(
      {
        name: 'a',
        pattern: '$self.x.$selfish',
        handler: 'echo',
      },
      'foo',
      '/p',
    );
    expect(resolved.pattern).toBe('foo.x.$selfish');
  });

  it('passes through max_attempts and timeout_ms when set', () => {
    const resolved = resolveSubscription(
      {
        name: 'a',
        pattern: 'x',
        handler: 'echo',
        max_attempts: 5,
        timeout_ms: 90000,
      },
      'foo',
      '/p',
    );
    expect(resolved.maxAttempts).toBe(5);
    expect(resolved.timeoutMs).toBe(90000);
  });

  it('synthesizes a `celilo events run-hook` handler for a hook subscription', () => {
    const resolved = resolveSubscription(
      {
        name: 'dns-register-system',
        pattern: 'system.created.*',
        hook: 'on_system_event',
        hook_inputs: { op: 'register' },
      },
      'technitium',
      '/var/lib/celilo/modules/technitium',
    );
    // The runner re-reads the manifest by (module, sub-name); the handler
    // carries exactly those two identifiers and nothing module-path-specific.
    expect(resolved.handler).toBe('celilo events run-hook technitium dns-register-system');
    expect(resolved.name).toBe('technitium.dns-register-system');
    expect(resolved.pattern).toBe('system.created.*');
  });
});

describe('ModuleSubscriptionSchema handler/hook validation', () => {
  const base = { name: 'a', pattern: 'x' };

  it('accepts a handler-only subscription', () => {
    expect(ModuleSubscriptionSchema.safeParse({ ...base, handler: 'echo' }).success).toBe(true);
  });

  it('accepts a hook-only subscription, with or without hook_inputs', () => {
    expect(ModuleSubscriptionSchema.safeParse({ ...base, hook: 'on_system_event' }).success).toBe(
      true,
    );
    expect(
      ModuleSubscriptionSchema.safeParse({
        ...base,
        hook: 'on_system_event',
        hook_inputs: { op: 'register' },
      }).success,
    ).toBe(true);
  });

  it('rejects declaring both handler and hook', () => {
    expect(
      ModuleSubscriptionSchema.safeParse({ ...base, handler: 'echo', hook: 'on_system_event' })
        .success,
    ).toBe(false);
  });

  it('rejects declaring neither handler nor hook', () => {
    expect(ModuleSubscriptionSchema.safeParse(base).success).toBe(false);
  });

  it('rejects hook_inputs without a hook', () => {
    expect(
      ModuleSubscriptionSchema.safeParse({ ...base, handler: 'echo', hook_inputs: { op: 'x' } })
        .success,
    ).toBe(false);
  });
});

describe('register / unregister roundtrip', () => {
  let dir: string;
  let dbPath: string;

  beforeEach(() => {
    dir = mkdtempSync(join(tmpdir(), 'modsubs-test-'));
    dbPath = join(dir, 'events.db');
    process.env.EVENT_BUS_DB = dbPath;
  });
  afterEach(() => {
    delete process.env.EVENT_BUS_DB;
    try {
      rmSync(dir, { recursive: true, force: true });
    } catch {
      /* ignore */
    }
  });

  it('manifest with no subscriptions is a no-op', () => {
    const result = registerModuleSubscriptions(baseManifest(), '/p');
    expect(result.registered).toBe(0);
  });

  it('a module with an on_backup hook arms the scheduled backup sweep', () => {
    // The sweep is a system-level subscriber, not a module one, so it does not
    // count toward `registered`. It appears the moment the fleet has something
    // to back up — including on `module update`, which is how a manifest that
    // newly declares a cadence reaches an already-deployed fleet.
    const result = registerModuleSubscriptions(
      baseManifest({ hooks: { on_backup: { script: 'backup.ts' } } }),
      '/p',
    );
    expect(result.registered).toBe(0);

    const bus = openBus({ dbPath, events: defineEvents({}) });
    try {
      const row = bus.db
        .query<{ name: string; pattern: string; handler: string }, []>(
          "SELECT name, pattern, handler FROM subscribers WHERE name = 'celilo-backup-sweep'",
        )
        .get();
      expect(row).toEqual({
        name: 'celilo-backup-sweep',
        pattern: 'timer.tick.1h',
        handler: 'celilo backup sweep',
      });
    } finally {
      bus.close();
    }
  });

  // Any module can hold the operation lock, so registering ANY module — even
  // one with no subscriptions and no backup hook — is enough to arm the sweep
  // that reclaims abandoned rows (#581).
  it('arms the abandoned-operations sweep for any module', () => {
    registerModuleSubscriptions(baseManifest({}), '/p');

    const bus = openBus({ dbPath, events: defineEvents({}) });
    try {
      const row = bus.db
        .query<{ pattern: string; handler: string }, []>(
          "SELECT pattern, handler FROM subscribers WHERE name = 'celilo-operations-sweep'",
        )
        .get();
      expect(row).toEqual({
        pattern: 'timer.tick.1h',
        handler: 'celilo module operations clear',
      });
    } finally {
      bus.close();
    }
  });

  it('registers each subscription as a row, names scoped to module id', () => {
    const result = registerModuleSubscriptions(
      baseManifest({
        subscriptions: [
          {
            name: 'smoke',
            pattern: 'deploy.completed.$self',
            handler: 'bun ${MODULE_PATH}/x.ts',
          },
          {
            name: 'on-cert-rotated',
            pattern: 'cert.rotated',
            handler: 'echo',
          },
        ],
      }),
      '/var/lib/celilo/modules/lunacycle',
    );
    expect(result.registered).toBe(2);

    const bus = openBus({ dbPath, events: defineEvents({}) });
    try {
      const rows = bus.db
        .query<{ name: string; pattern: string; handler: string }, []>(
          "SELECT name, pattern, handler FROM subscribers WHERE name LIKE '%.%' ORDER BY name",
        )
        .all();
      expect(rows).toEqual([
        {
          name: 'lunacycle.on-cert-rotated',
          pattern: 'cert.rotated',
          handler: 'echo',
        },
        {
          name: 'lunacycle.smoke',
          pattern: 'deploy.completed.lunacycle',
          handler: 'bun /var/lib/celilo/modules/lunacycle/x.ts',
        },
      ]);
    } finally {
      bus.close();
    }
  });

  it('re-registering the same manifest is idempotent (updates rows in place)', () => {
    const m = baseManifest({
      subscriptions: [{ name: 's', pattern: 'a', handler: 'echo first' }],
    });
    registerModuleSubscriptions(m, '/p');
    // Edit the in-memory manifest and re-register; same name, new handler.
    if (!m.subscriptions) throw new Error('subscriptions missing');
    m.subscriptions[0].handler = 'echo second';
    registerModuleSubscriptions(m, '/p');

    const bus = openBus({ dbPath, events: defineEvents({}) });
    try {
      const rows = bus.db
        .query<{ count: number }, []>(
          "SELECT COUNT(*) AS count FROM subscribers WHERE name LIKE '%.%'",
        )
        .get();
      expect(rows?.count).toBe(1);
      const row = bus.db
        .query<{ handler: string }, []>("SELECT handler FROM subscribers WHERE name LIKE '%.%'")
        .get();
      expect(row?.handler).toBe('echo second');
    } finally {
      bus.close();
    }
  });

  it('unregister removes all subscriptions for the module, scoped by name prefix', () => {
    registerModuleSubscriptions(
      baseManifest({
        id: 'lunacycle',
        subscriptions: [
          { name: 'a', pattern: 'x.$self', handler: 'echo' },
          { name: 'b', pattern: 'y.$self', handler: 'echo' },
        ],
      }),
      '/p',
    );
    // Another module's subs should NOT be touched.
    registerModuleSubscriptions(
      baseManifest({
        id: 'authentik',
        subscriptions: [{ name: 'a', pattern: 'z.$self', handler: 'echo' }],
      }),
      '/p',
    );

    const result = unregisterModuleSubscriptions('lunacycle');
    expect(result.unregistered).toBe(2);

    const bus = openBus({ dbPath, events: defineEvents({}) });
    try {
      const rows = bus.db
        .query<{ name: string }, []>(
          "SELECT name FROM subscribers WHERE name LIKE '%.%' ORDER BY name",
        )
        .all();
      expect(rows).toEqual([{ name: 'authentik.a' }]);
    } finally {
      bus.close();
    }
  });
});

describe('resyncAllSubscriptions (ISS-0088)', () => {
  let dir: string;
  let busPath: string;

  beforeEach(async () => {
    closeDb();
    dir = mkdtempSync(join(tmpdir(), 'resync-test-'));
    busPath = join(dir, 'events.db');
    process.env.CELILO_DB_PATH = join(dir, 'celilo.db');
    process.env.EVENT_BUS_DB = busPath;
    // Migrate the celilo.db file getDb() will open (resyncAllSubscriptions reads it).
    const setupDb = await migrateDbFile(join(dir, 'celilo.db'));
    setupDb.$client.close();
  });
  afterEach(() => {
    closeDb();
    resetTestDbPath();
    delete process.env.EVENT_BUS_DB;
    try {
      rmSync(dir, { recursive: true, force: true });
    } catch {
      /* ignore */
    }
  });

  function seedModule(
    id: string,
    state: string,
    subs: Array<{ name: string; pattern: string; handler: string }>,
  ): void {
    const manifest = JSON.stringify({
      celilo_contract: '1.0',
      id,
      name: id,
      version: '1.0.0',
      requires: { capabilities: [] },
      provides: { capabilities: [] },
      variables: { owns: [], imports: [] },
      subscriptions: subs,
    });
    getDb().$client.run(
      'INSERT INTO modules (id, name, version, source_path, manifest_data, state) VALUES (?, ?, ?, ?, ?, ?)',
      [id, id, '1.0.0', '/p', manifest, state],
    );
  }

  function subscriberNames(): string[] {
    const bus = openBus({ dbPath: busPath, events: defineEvents({}) });
    try {
      // Module subscriptions are dot-scoped (`<module-id>.<sub-name>`); celilo's
      // own housekeeping subscribers (the backup and operations sweeps) are not,
      // and are not what these tests are about.
      return bus.db
        .query<{ name: string }, []>(
          "SELECT name FROM subscribers WHERE name LIKE '%.%' ORDER BY name",
        )
        .all()
        .map((r) => r.name);
    } finally {
      bus.close();
    }
  }

  it('registers subs for DEPLOYED modules only, skipping imported/undeployed', () => {
    seedModule('caddy', 'VERIFIED', [
      { name: 'reconcile', pattern: 'public_web.routes_changed', handler: 'echo' },
    ]);
    seedModule('technitium', 'INSTALLED', [
      { name: 'dns', pattern: 'system.created.*', handler: 'echo' },
    ]);
    seedModule('forgejo', 'IMPORTED', [{ name: 'x', pattern: 'y', handler: 'echo' }]);
    seedModule('greenwave', 'VERIFIED', []); // deployed but declares no subs

    const result = resyncAllSubscriptions();

    expect(result.modules).toBe(2); // caddy + technitium (forgejo not deployed, greenwave has none)
    expect(result.registered).toBe(2);
    expect(result.failures).toEqual([]);
    // forgejo.x absent — it was IMPORTED, not deployed.
    expect(subscriberNames()).toEqual(['caddy.reconcile', 'technitium.dns']);
  });

  it('is idempotent — a second resync does not duplicate rows', () => {
    seedModule('caddy', 'VERIFIED', [{ name: 'reconcile', pattern: 'a', handler: 'echo' }]);
    resyncAllSubscriptions();
    resyncAllSubscriptions();
    expect(subscriberNames()).toEqual(['caddy.reconcile']);
  });
});

describe('build-bus registry-poll wiring (ISS-0139)', () => {
  // The CD poll is wired as a celilo-mgmt manifest subscription so it registers
  // ONLY on the management host. Guard the exact handler/pattern against drift.
  const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '..', '..', '..', '..');

  it('celilo-mgmt subscribes the registry poll to timer.tick.15m', () => {
    const manifestPath = join(REPO_ROOT, 'modules', 'celilo-mgmt', 'manifest.yml');
    const manifest = ModuleManifestSchema.parse(parseYaml(readFileSync(manifestPath, 'utf-8')));

    const poll = manifest.subscriptions?.find((s) => s.name === 'registry-poll');
    expect(poll).toBeDefined();
    expect(poll?.pattern).toBe('timer.tick.15m');
    // A literal handler command (the CLI poll), not a hook — see manifest comment.
    // `--poll` is load-bearing: the dispatcher appends the event id to a
    // subprocess handler, and without the flag it lands in the module-name slot
    // ("Module not found: <event_id>" every tick).
    expect(poll?.handler).toBe('celilo module upgrade --poll');
    expect(poll?.hook).toBeUndefined();

    const resolved = resolveSubscription(
      // biome-ignore lint/style/noNonNullAssertion: asserted defined above
      poll!,
      'celilo-mgmt',
      '/modules/celilo-mgmt',
    );
    expect(resolved.name).toBe('celilo-mgmt.registry-poll');
    expect(resolved.handler).toBe('celilo module upgrade --poll');
  });
});