/**
 * ResponderFanout
 *
 * One-to-many (fan-out) dispatcher for a WebRTC responder. A single local source
 * (e.g. one camera) is served to MANY simultaneous remote peers, each as its own
 * peer connection.
 *
 * Why this exists: peers that share one logical channel (a peripheral twin) all
 * sign their signaling with the SAME sourceTwinId (the peripheral id), so the
 * responder cannot tell them apart by twin identity. Instead, each initiator
 * mints a per-session `peerId` (see PeerConnectionManager) and stamps it on every
 * offer/answer/ice. This dispatcher owns the single transport subscription,
 * demultiplexes incoming signaling by `peerId`, and runs one responder session
 * (a MediaStreamHandler / DataChannelHandler in selfManagedSignaling mode) per
 * peer — feeding each only its own messages.
 *
 * The per-peer sessions own the media/data lifecycle (tracks, channels,
 * reconnect); this layer is purely the router.
 */

import { TwinTransport, PeerInfo, SignalingSink, PeerCapReachedError } from './types';

// Sentinel sourceTwinId for an offer delivered out-of-band (an HTTP request body)
// rather than over the socket transport. The viewer has no twin to address, so
// its answer/candidates flow back through a signalingSink instead of the wire.
const DIRECT_HTTP_SOURCE_TWIN_ID = 'direct-http';

/**
 * Minimal contract a per-peer responder session must satisfy. Both
 * MediaStreamHandler and DataChannelHandler already match it.
 */
export interface FanoutSession<H> {
  /** Establish the peer connection and resolve with the per-peer handle. */
  connect(): Promise<H>;
  /** Feed one demultiplexed signaling message (offer/answer/ice). */
  ingestSignalingMessage(message: any): Promise<void> | void;
  /** Tear down this peer's connection and resources. */
  close(): void;
}

export interface ResponderFanoutOptions<H> {
  transport: TwinTransport;
  /** The logical channel twin everyone signals on (e.g. the camera peripheral id). */
  targetTwinId: string;
  /** Signaling channel prefix to match, e.g. 'media-default' or 'dc-ping-pong'. */
  channelPrefix: string;
  /** Optional cap on concurrent peers; offers from new peers beyond it are ignored. */
  maxPeers?: number;
  /**
   * How long (ms) a spawned peer has to reach 'connected' before it is evicted.
   * The per-peer manager retries connection forever, so without this a peer that
   * offers but never completes ICE would hold its slot (and a maxPeers seat)
   * indefinitely. Default 45000 (covers the STUN-toggle retry).
   */
  connectTimeoutMs?: number;
  /**
   * Build a per-peer responder session.
   * - `replyToTwinId` is the offer's sourceTwinId — the twin this peer's
   *   answers/ice must be addressed to (the peripheral twin for peripheral
   *   channels, or the remote instance twin for instance-to-instance channels).
   * - `onClosed` MUST be wired to the session's disconnect/teardown so the
   *   dispatcher can drop the peer from its map.
   * - `signalingSink`, when present (one-shot HTTP answer path), diverts this
   *   session's outbound answer/ice to the sink instead of the transport.
   */
  createSession: (
    peerId: string,
    replyToTwinId: string,
    onClosed: () => void,
    signalingSink?: SignalingSink,
  ) => FanoutSession<H>;
  /** Called once per peer when its session connects, with the peer's identity. */
  onPeer: (handle: H, peerId: string, info: PeerInfo) => void;
  /** Called when a peer's session fails to connect or closes. */
  onPeerClosed?: (peerId: string, error?: Error) => void;
}

// Cap on stray ICE candidates buffered for a peer we haven't seen an offer from
// yet — guards against an unbounded buffer from stray/old candidates.
const MAX_PREOFFER_ICE = 32;
// Cap on the number of DISTINCT not-yet-offered peers we will buffer ICE for —
// guards against a sender emitting candidates under many random peerIds (which
// would otherwise grow preOfferBuffer without bound, one key per peerId).
const MAX_PENDING_PEERS = 64;
const DEFAULT_CONNECT_TIMEOUT_MS = 45000;

export class ResponderFanout<H> {
  private opts: ResponderFanoutOptions<H>;
  private peers: Map<string, FanoutSession<H>> = new Map();
  private preOfferBuffer: Map<string, any[]> = new Map();
  // Per-peer "must connect by" timers; cleared once a peer connects or is removed.
  private connectTimers: Map<string, ReturnType<typeof setTimeout>> = new Map();
  private messageHandler: ((message: any) => void) | null = null;
  private started = false;
  private closed = false;

  constructor(opts: ResponderFanoutOptions<H>) {
    this.opts = opts;
  }

  /** Subscribe and begin accepting peers. Idempotent. */
  async start(): Promise<void> {
    if (this.started || this.closed) return;
    this.started = true;

    await this.opts.transport.subscribe(this.opts.targetTwinId).catch((err) => {
      console.error('[ResponderFanout] Failed to subscribe to twin:', err);
    });

    this.messageHandler = (message: any) => this.onMessage(message);
    this.opts.transport.onMessage(this.opts.transport.twinId, this.messageHandler);
  }

  /** Number of currently-tracked peers. */
  getPeerCount(): number {
    return this.peers.size;
  }

  /**
   * Feed a SINGLE offer delivered out-of-band (from an HTTP request body, not the
   * socket transport) into this already-armed fan-out, and resolve with the
   * matching non-trickle answer SDP (WHEP). The offer is answered on a fresh
   * per-peer session exactly like a socket-signaled peer, except the session's
   * outbound answer/ice is diverted to a per-call signaling sink — the viewer has
   * no twin to receive signaling over. The answer is emitted only after ICE
   * gathering completes (or its cap), so it carries the publisher's relay
   * candidates embedded and the viewer can connect with no further signaling.
   *
   * Reuses the same gating and spawn path as the socket flow: an existing session
   * for `peerId` is replaced, and the maxPeers cap is enforced (as a typed
   * PeerCapReachedError here, so the caller can map it to a 503). The connect
   * timeout still arms, so a viewer that takes the answer but never connects is
   * evicted and frees its slot.
   *
   * @param peerId - the viewer's id (distinguishes it from other viewers)
   * @param offerSdp - the viewer's offer SDP
   * @returns the answer SDP (gathering-complete)
   */
  async ingestExternalOffer(peerId: string, offerSdp: string): Promise<string> {
    if (this.closed) {
      throw new Error('Cannot ingest external offer: ResponderFanout is closed');
    }
    if (!this.started) {
      await this.start();
    }

    // A re-offer/retry for a viewer already in flight: drop the stale session so
    // the new offer is answered on a fresh peer connection.
    if (this.peers.has(peerId)) {
      this.removePeer(peerId);
    }

    // Same cap the socket path enforces in onMessage, surfaced as a typed error.
    if (this.opts.maxPeers && this.peers.size >= this.opts.maxPeers) {
      throw new PeerCapReachedError(peerId, this.opts.maxPeers);
    }

    return new Promise<string>((resolve, reject) => {
      const signalingSink: SignalingSink = (type, data) => {
        // Non-trickle: only the gathering-complete answer matters. ice sends are
        // swallowed upstream (PeerConnectionManager), so none reach here.
        if (type !== 'answer') return;
        const sdp = (data as RTCSessionDescriptionInit).sdp;
        if (sdp) {
          resolve(sdp);
        } else {
          reject(new Error('Failed to answer external offer: peer produced an empty answer description'));
        }
      };

      // Synthesize the signaling envelope a per-peer session expects. The type
      // prefix must match the fan-out's channelPrefix (the session gates on it);
      // the sentinel sourceTwinId marks a viewer with no return transport.
      const offerMessage = {
        data: {
          type: `${this.opts.channelPrefix}-${this.opts.targetTwinId}:offer`,
          peerId,
          data: { type: 'offer', sdp: offerSdp },
        },
        sourceTwinId: DIRECT_HTTP_SOURCE_TWIN_ID,
      };

      this.spawnPeer(peerId, offerMessage, signalingSink);
    });
  }

  /** Drop a peer and close its session. Safe to call repeatedly. */
  removePeer(peerId: string, error?: Error): void {
    this.clearConnectTimer(peerId);
    const session = this.peers.get(peerId);
    if (!session) return;
    this.peers.delete(peerId);
    try {
      session.close();
    } catch (err) {
      console.error(`[ResponderFanout] Error closing peer ${peerId}:`, err);
    }
    this.opts.onPeerClosed?.(peerId, error);
  }

  /** Tear down the dispatcher and every peer. */
  close(): void {
    if (this.closed) return;
    this.closed = true;

    if (this.messageHandler) {
      this.opts.transport.offMessage(this.opts.transport.twinId, this.messageHandler);
      this.messageHandler = null;
    }

    this.peers.forEach((session) => {
      try {
        session.close();
      } catch (err) {
        console.error('[ResponderFanout] Error closing peer session:', err);
      }
    });
    this.peers.clear();
    this.preOfferBuffer.clear();
    this.connectTimers.forEach((timer) => clearTimeout(timer));
    this.connectTimers.clear();
  }

  private clearConnectTimer(peerId: string): void {
    const timer = this.connectTimers.get(peerId);
    if (timer) {
      clearTimeout(timer);
      this.connectTimers.delete(peerId);
    }
  }

  // ===========================================================================
  // Private
  // ===========================================================================

  private onMessage(message: any): void {
    if (this.closed) return;

    const messageType = message?.data?.type as string | undefined;
    if (!messageType || !messageType.startsWith(`${this.opts.channelPrefix}-`)) return;

    const signalType = messageType.split(':').pop();
    // Demux strictly by the minted per-session id. A message without one cannot
    // be attributed to a peer, so ignore it.
    const peerId = message.data?.peerId as string | undefined;
    if (!peerId) {
      console.warn('[ResponderFanout] Ignoring signaling message with no peerId');
      return;
    }

    const existing = this.peers.get(peerId);
    if (existing) {
      existing.ingestSignalingMessage(message);
      return;
    }

    // Unknown peer. Only an offer starts a new session; buffer any ICE that
    // races ahead of the offer so a reordered candidate isn't dropped.
    if (signalType !== 'offer') {
      const existingBuffer = this.preOfferBuffer.get(peerId);
      if (existingBuffer) {
        if (existingBuffer.length < MAX_PREOFFER_ICE) existingBuffer.push(message);
      } else if (this.preOfferBuffer.size < MAX_PENDING_PEERS) {
        // Only start buffering for a NEW peerId while we're under the distinct-peer
        // cap; otherwise drop, so stray candidates under random ids can't grow the map.
        this.preOfferBuffer.set(peerId, [message]);
      }
      return;
    }

    if (this.opts.maxPeers && this.peers.size >= this.opts.maxPeers) {
      console.warn(`[ResponderFanout] peer cap (${this.opts.maxPeers}) reached — ignoring new peer ${peerId}`);
      return;
    }

    this.spawnPeer(peerId, message);
  }

  private spawnPeer(peerId: string, offerMessage: any, signalingSink?: SignalingSink): void {
    // Answers/ice for this peer must go back to whoever sent the offer.
    const replyToTwinId = offerMessage.sourceTwinId ?? this.opts.targetTwinId;
    // The remote's device id rides on the signaling envelope — surface it so the
    // app can tell which physical device each peer is (peerId only distinguishes
    // sessions). Undefined for peers that don't stamp one (e.g. a browser peer).
    const info: PeerInfo = { deviceId: offerMessage.sourceDeviceId };
    const session = this.opts.createSession(peerId, replyToTwinId, () => this.removePeer(peerId), signalingSink);
    this.peers.set(peerId, session);

    // Evict the peer if it never reaches 'connected' — the per-peer manager
    // retries forever, so a stuck offerer would otherwise hold its slot for good.
    const deadlineMs = this.opts.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS;
    this.connectTimers.set(
      peerId,
      setTimeout(() => {
        this.connectTimers.delete(peerId);
        if (this.peers.get(peerId) === session) {
          console.warn(`[ResponderFanout] peer ${peerId} never connected within ${deadlineMs}ms — evicting`);
          this.removePeer(peerId, new Error('connect timeout'));
        }
      }, deadlineMs),
    );

    session
      .connect()
      .then((handle) => {
        this.clearConnectTimer(peerId);
        // Bail if the dispatcher was torn down, or this peer was already removed/
        // replaced while connect() was in flight — don't hand the app a dead handle.
        if (this.closed || this.peers.get(peerId) !== session) {
          try {
            session.close();
          } catch (err) {
            console.error(`[ResponderFanout] Error closing stale peer ${peerId}:`, err);
          }
          return;
        }
        this.opts.onPeer(handle, peerId, info);
      })
      .catch((err) => {
        console.error(`[ResponderFanout] Peer ${peerId} failed to connect:`, err);
        this.removePeer(peerId, err instanceof Error ? err : new Error(String(err)));
      });

    // Replay any pre-offer ICE, then the offer. (Order is not critical: the
    // per-peer manager buffers ICE until the remote description is set.)
    const buffered = this.preOfferBuffer.get(peerId);
    if (buffered) {
      this.preOfferBuffer.delete(peerId);
      for (const bufferedMessage of buffered) session.ingestSignalingMessage(bufferedMessage);
    }
    session.ingestSignalingMessage(offerMessage);
  }
}