/**
 * KHAELOR
 * File: src/session/bus.ts
 * Description: Typed in-process event bus (write-ahead durable publish) and the ~16 ms delta coalescer.
 *
 * Author: Simon-Pierre Boucher
 * Contact: contact@spboucher.ai
 */

import { KhaelorError, ulid } from "../shared/index.js";
import type { Unsubscribe } from "../shared/index.js";
import {
  EVENT_SCHEMA_VERSION,
  isDurableEventType,
  isEphemeralEventType,
} from "./events.js";
import type {
  DurableEvent,
  DurableEventInput,
  EphemeralEvent,
  EphemeralEventInput,
  KhaelorEvent,
} from "./events.js";

// ───────────────────────────── event bus ─────────────────────────────

/**
 * Write-ahead sink for durable events. The SessionLog implements this:
 * it assigns the envelope (v/id/sessionId/seq/ts) and enqueues the
 * serialized append BEFORE the bus fans the event out (EVENT_MODEL.md §5.2).
 */
export interface DurableAppender {
  append(input: DurableEventInput): DurableEvent;
}

/** Bus API (EVENT_MODEL.md §8). */
export interface EventBus {
  /** Append (write-ahead) then publish. Returns the full envelope. */
  publishDurable(input: DurableEventInput): DurableEvent;
  /** Fire-and-forget to live subscribers. Never persisted. */
  publishEphemeral(input: EphemeralEventInput): void;
  /** Typed subscription — handler parameter narrows by `type`. */
  on<T extends KhaelorEvent["type"]>(
    type: T,
    handler: (e: Extract<KhaelorEvent, { type: T }>) => void,
  ): Unsubscribe;
  /** Wildcard (projections, logging, coalescer). */
  onAny(handler: (e: KhaelorEvent) => void): Unsubscribe;
}

export interface SessionEventBusOptions {
  sessionId: string;
  /** Write-ahead durable sink. When absent (tests, ephemeral-only usage) the bus self-assigns envelopes. */
  appender?: DurableAppender;
  /** Handler exceptions are caught and reported here — never rethrown, never block other subscribers. */
  onHandlerError?: (error: unknown, event: KhaelorEvent) => void;
}

type AnyHandler = (e: KhaelorEvent) => void;

/**
 * Synchronous, in-process, per-session event bus. Fan-out is synchronous and
 * in `seq` order; handlers must be fast and non-throwing (EVENT_MODEL.md §8).
 */
export class SessionEventBus implements EventBus {
  private readonly sessionId: string;
  private readonly appender: DurableAppender | undefined;
  private readonly onHandlerError: ((error: unknown, event: KhaelorEvent) => void) | undefined;
  private readonly byType = new Map<string, Set<AnyHandler>>();
  private readonly anyHandlers = new Set<AnyHandler>();
  /** Fallback seq counter used only when no appender is attached. */
  private selfSeq = 0;

  constructor(options: SessionEventBusOptions) {
    this.sessionId = options.sessionId;
    this.appender = options.appender;
    this.onHandlerError = options.onHandlerError;
  }

  publishDurable(input: DurableEventInput): DurableEvent {
    if (!isDurableEventType(input.type)) {
      throw new KhaelorError("invalid-event", `publishDurable: not a durable event type: ${String(input.type)}`);
    }
    const event: DurableEvent = this.appender
      ? this.appender.append(input)
      : ({
          v: EVENT_SCHEMA_VERSION,
          id: ulid(),
          sessionId: this.sessionId,
          seq: ++this.selfSeq,
          ts: Date.now(),
          type: input.type,
          payload: input.payload,
        } as DurableEvent);
    this.dispatch(event);
    return event;
  }

  publishEphemeral(input: EphemeralEventInput): void {
    if (!isEphemeralEventType(input.type)) {
      throw new KhaelorError("invalid-event", `publishEphemeral: not an ephemeral event type: ${String(input.type)}`);
    }
    const event: EphemeralEvent = {
      id: ulid(),
      sessionId: this.sessionId,
      ts: Date.now(),
      type: input.type,
      payload: input.payload,
    } as EphemeralEvent;
    this.dispatch(event);
  }

  on<T extends KhaelorEvent["type"]>(
    type: T,
    handler: (e: Extract<KhaelorEvent, { type: T }>) => void,
  ): Unsubscribe {
    let set = this.byType.get(type);
    if (!set) {
      set = new Set();
      this.byType.set(type, set);
    }
    const anyHandler = handler as AnyHandler;
    set.add(anyHandler);
    return () => {
      set.delete(anyHandler);
    };
  }

  onAny(handler: (e: KhaelorEvent) => void): Unsubscribe {
    this.anyHandlers.add(handler);
    return () => {
      this.anyHandlers.delete(handler);
    };
  }

  /** Number of live subscriptions (leak checks in tests). */
  subscriberCount(): number {
    let n = this.anyHandlers.size;
    for (const set of this.byType.values()) n += set.size;
    return n;
  }

  private dispatch(event: KhaelorEvent): void {
    const typed = this.byType.get(event.type);
    // Snapshot so handlers may unsubscribe during delivery.
    const handlers: AnyHandler[] = [];
    if (typed) handlers.push(...typed);
    handlers.push(...this.anyHandlers);
    for (const handler of handlers) {
      try {
        handler(event);
      } catch (error) {
        this.onHandlerError?.(error, event);
      }
    }
  }
}

// ───────────────────────────── coalescer ─────────────────────────────

/** Key for streaming text/thinking blocks: `${requestId}#${blockIndex}`. */
export type BlockKey = string;

export function blockKey(requestId: string, blockIndex: number): BlockKey {
  return `${requestId}#${blockIndex}`;
}

/** One render frame worth of coalesced deltas + durable events (EVENT_MODEL.md §7). */
export interface CoalescedFrame {
  textAppends: Map<BlockKey, string>; // concatenated ModelTextDelta / ModelThinkingDelta
  toolInputPreviews: Map<string, string>; // latest accumulated partial JSON per toolUseId
  toolOutputAppends: Map<string, string>; // concatenated ToolOutput per toolUseId
  processOutputAppends: Map<string, string>; // concatenated ProcessOutput per processId
  durables: DurableEvent[]; // durable events in this window, in seq order
}

export interface CoalescerOptions {
  /** Flush window in milliseconds (contract: ~16 ms). */
  windowMs?: number;
  /** Per-key per-frame cap on buffered append strings (default 16 KiB of UTF-16 units). */
  maxAppendPerKey?: number;
}

export const COALESCER_TRUNCATION_MARKER = "…[preview truncated]";

const DEFAULT_WINDOW_MS = 16;
const DEFAULT_MAX_APPEND_PER_KEY = 16 * 1024;

/**
 * The single sanctioned buffering point between the bus and the TUI.
 * Buffers ephemeral deltas, flushes at most one frame per ~16 ms window,
 * and flushes immediately when a durable event settles a buffered key
 * (deltas delivered in the same frame, before the durable) or on Interrupted.
 */
export class Coalescer {
  private readonly windowMs: number;
  private readonly maxAppendPerKey: number;
  private readonly subscribers = new Set<(f: CoalescedFrame) => void>();
  private readonly unsubscribeBus: Unsubscribe;

  private timer: ReturnType<typeof setTimeout> | null = null;
  private textAppends = new Map<BlockKey, string>();
  private toolOutputAppends = new Map<string, string>();
  private processOutputAppends = new Map<string, string>();
  private durables: DurableEvent[] = [];
  /** toolUseIds whose input preview changed in the current window. */
  private toolInputTouched = new Set<string>();
  /** Accumulated (capped) tool input previews — persists across frames until the tool settles. */
  private readonly toolInputAccum = new Map<string, string>();
  /** Keys that overflowed the per-frame cap in the current window. */
  private readonly overflowedKeys = new Set<string>();

  constructor(bus: EventBus, options: CoalescerOptions = {}) {
    this.windowMs = options.windowMs ?? DEFAULT_WINDOW_MS;
    this.maxAppendPerKey = options.maxAppendPerKey ?? DEFAULT_MAX_APPEND_PER_KEY;
    this.unsubscribeBus = bus.onAny((event) => {
      this.ingest(event);
    });
  }

  subscribe(onFrame: (frame: CoalescedFrame) => void): Unsubscribe {
    this.subscribers.add(onFrame);
    return () => {
      this.subscribers.delete(onFrame);
    };
  }

  /** Detach from the bus and drop any buffered state without delivering it. */
  dispose(): void {
    this.unsubscribeBus();
    if (this.timer !== null) {
      clearTimeout(this.timer);
      this.timer = null;
    }
    this.resetWindow();
    this.subscribers.clear();
  }

  private ingest(event: KhaelorEvent): void {
    switch (event.type) {
      case "model.text-delta":
      case "model.thinking-delta": {
        const key = blockKey(event.payload.requestId, event.payload.blockIndex);
        this.appendCapped(this.textAppends, key, event.payload.text);
        this.armTimer();
        return;
      }
      case "model.tool-input-delta": {
        const id = event.payload.toolUseId;
        const current = this.toolInputAccum.get(id) ?? "";
        this.toolInputAccum.set(
          id,
          this.capAppend(current, event.payload.partialJson, `input:${id}`),
        );
        this.toolInputTouched.add(id);
        this.armTimer();
        return;
      }
      case "tool.output": {
        this.appendCapped(this.toolOutputAppends, event.payload.toolUseId, event.payload.chunk);
        this.armTimer();
        return;
      }
      case "process.output": {
        this.appendCapped(this.processOutputAppends, event.payload.processId, event.payload.chunk);
        this.armTimer();
        return;
      }
      case "model.tool-call-started": {
        // UI hint only; nothing to buffer, but it starts a window so the
        // preview state (if any follows) flushes on cadence.
        this.armTimer();
        return;
      }
      default: {
        // Durable event.
        const durable = event as DurableEvent;
        this.durables.push(durable);
        if (this.settlesBufferedKey(durable) || durable.type === "user.interrupted") {
          this.flushNow();
        } else {
          this.armTimer();
        }
        return;
      }
    }
  }

  private settlesBufferedKey(event: DurableEvent): boolean {
    switch (event.type) {
      case "model.text-block-completed":
      case "model.thinking-block-completed":
        return this.textAppends.has(blockKey(event.payload.requestId, event.payload.blockIndex));
      case "tool.requested": {
        const id = event.payload.toolUseId;
        const buffered = this.toolInputTouched.has(id);
        // Settlement replaces: accumulated preview is dropped permanently.
        this.toolInputAccum.delete(id);
        return buffered;
      }
      case "tool.completed":
      case "tool.failed":
      case "tool.cancelled":
        return this.toolOutputAppends.has(event.payload.toolUseId);
      case "process.exited":
        return this.processOutputAppends.has(event.payload.processId);
      default:
        return false;
    }
  }

  private appendCapped(map: Map<string, string>, key: string, chunk: string): void {
    map.set(key, this.capAppend(map.get(key) ?? "", chunk, key));
  }

  private capAppend(current: string, chunk: string, capKey: string): string {
    if (this.overflowedKeys.has(capKey)) return current;
    const combined = current + chunk;
    if (combined.length <= this.maxAppendPerKey) return combined;
    this.overflowedKeys.add(capKey);
    return combined.slice(0, this.maxAppendPerKey) + COALESCER_TRUNCATION_MARKER;
  }

  private armTimer(): void {
    if (this.timer !== null) return;
    this.timer = setTimeout(() => {
      this.timer = null;
      this.flushNow();
    }, this.windowMs);
  }

  private flushNow(): void {
    if (this.timer !== null) {
      clearTimeout(this.timer);
      this.timer = null;
    }
    if (
      this.textAppends.size === 0 &&
      this.toolInputTouched.size === 0 &&
      this.toolOutputAppends.size === 0 &&
      this.processOutputAppends.size === 0 &&
      this.durables.length === 0
    ) {
      return;
    }
    const toolInputPreviews = new Map<string, string>();
    for (const id of this.toolInputTouched) {
      const preview = this.toolInputAccum.get(id);
      if (preview !== undefined) toolInputPreviews.set(id, preview);
    }
    const frame: CoalescedFrame = {
      textAppends: this.textAppends,
      toolInputPreviews,
      toolOutputAppends: this.toolOutputAppends,
      processOutputAppends: this.processOutputAppends,
      durables: this.durables,
    };
    this.resetWindow();
    for (const subscriber of [...this.subscribers]) {
      try {
        subscriber(frame);
      } catch {
        // A slow or broken renderer must never break the engine.
      }
    }
  }

  private resetWindow(): void {
    this.textAppends = new Map();
    this.toolOutputAppends = new Map();
    this.processOutputAppends = new Map();
    this.durables = [];
    this.toolInputTouched.clear();
    this.overflowedKeys.clear();
  }
}