/**
 * Signal runtime — binds declarations to a protocol driver and exposes
 * emit / subscribe / live / drain for `fx` and tests.
 */

import type {
  DeadLetter,
  LiveEvent,
  SignalBus,
  SignalDriver,
  SignalEmitOptions,
} from "../../drivers/signal-types.ts";
import type { SignalDecl } from "./declare.ts";

/** Options for {@link createSignalRuntime}. */
export interface CreateSignalRuntimeOptions {
  /** Protocol driver (`memory` · `postgres` · `redis` · `nats`). */
  readonly driver: SignalDriver;
  /** Clock. */
  readonly now?: () => number;
  /** Visibility lease for `once` claims (ms). */
  readonly leaseMs?: number;
  /** Durable path for chaos recovery. */
  readonly durablePath?: string;
  /** Injected clients forwarded to the driver. */
  readonly sql?: unknown;
  readonly redis?: import("../../drivers/signal-types.ts").SignalRedisClientLike;
  readonly nats?: import("../../drivers/signal-types.ts").SignalNatsClientLike;
  /** Forwarded to the driver. Postgres poll when `listen` is absent. */
  readonly pollMs?: number;
  /** Redis consumer-group consume for `once`. */
  readonly compete?: boolean;
  /** Redis consumer name. */
  readonly consumerId?: string;
}

/** Signal runtime. */
export interface SignalRuntime {
  /** Bound protocol driver id (`memory` · `postgres` · `redis` · `nats`). */
  readonly driverId: string;
  /** Underlying bus (after {@link SignalRuntime.start}). */
  readonly bus: SignalBus | null;
  /**
   * Register a signal declaration.
   *
   * @param decl - From {@link signal}
   */
  register(decl: SignalDecl): void;
  /** Registered declarations. */
  readonly declarations: ReadonlyMap<string, SignalDecl>;
  /**
   * Open the driver bus. Idempotent.
   */
  start(): Promise<SignalBus>;
  /**
   * Emit via the bus (auto-starts).
   *
   * @param name - Signal name
   * @param payload - Payload
   * @param options - Optional emit options (`key` for per-key once ordering)
   */
  emit(name: string, payload?: unknown, options?: SignalEmitOptions): Promise<void>;
  /**
   * Dead-lettered messages for one signal (auto-starts).
   *
   * @param name - Signal name
   */
  deadLetters(name: string): Promise<readonly DeadLetter[]>;
  /**
   * Live feed (auto-starts). Replays history then new events.
   *
   * @param name - Signal name
   * @param opts - Optional resume cursor
   */
  live(name: string, opts?: { readonly afterId?: string }): AsyncIterable<LiveEvent>;
  /**
   * Validate a live resume cursor (auto-starts). Throws OKE1210 when missing.
   *
   * @param name - Signal name
   * @param afterId - SSE cursor
   */
  checkLiveResume(name: string, afterId: string): Promise<void>;
  /** Close the bus. */
  close(): Promise<void>;
}

/**
 * Create a signal runtime bound to one protocol driver.
 *
 * @param options - Driver and bindings
 */
export function createSignalRuntime(options: CreateSignalRuntimeOptions): SignalRuntime {
  const declarations = new Map<string, SignalDecl>();
  let bus: SignalBus | null = null;

  const runtime: SignalRuntime = {
    driverId: options.driver.id,
    get bus() {
      return bus;
    },
    declarations,
    register(decl) {
      declarations.set(decl.name, decl);
    },
    async start() {
      if (bus) return bus;
      bus = await options.driver.open({
        signals: declarations,
        now: options.now,
        leaseMs: options.leaseMs,
        durablePath: options.durablePath,
        sql: options.sql,
        redis: options.redis,
        nats: options.nats,
        pollMs: options.pollMs,
        compete: options.compete,
        consumerId: options.consumerId,
      });
      return bus;
    },
    async emit(name, payload, options) {
      const b = await this.start();
      await b.emit(name, payload, options);
    },
    async deadLetters(name) {
      const b = await this.start();
      return b.deadLetters(name);
    },
    live(name, opts) {
      return {
        [Symbol.asyncIterator]() {
          let inner: AsyncIterator<LiveEvent> | undefined;
          return {
            async next() {
              if (!inner) {
                const b = await runtime.start();
                inner = b.live(name, opts)[Symbol.asyncIterator]();
              }
              return inner.next();
            },
            async return() {
              if (!inner) {
                const b = await runtime.start();
                inner = b.live(name, opts)[Symbol.asyncIterator]();
              }
              return inner.return?.() ?? { done: true, value: undefined };
            },
          };
        },
      };
    },
    async checkLiveResume(name, afterId) {
      const b = await runtime.start();
      await b.checkLiveResume(name, afterId);
    },
    async close() {
      if (bus) {
        await bus.close();
        bus = null;
      }
    },
  };

  return runtime;
}