import { TransferTransportError } from '../errors.js';
import type {
  ChunkAck,
  ChunkSendOptions,
  IControlChannel,
  ITransferTransport,
  TransferResumeState,
} from './transport.js';
import type { StreamControlMessage } from '@shade/streams';

/**
 * In-process control channel useful for tests and same-process upload/
 * download (e.g., a CLI that uploads to its own embedded server).
 *
 * Two endpoints are paired via {@link MemoryControlChannel.linked}: a
 * `send()` on one delivers to the other's subscribers.
 */
export class MemoryControlChannel implements IControlChannel {
  private peer: MemoryControlChannel | null = null;
  private myAddress: string;
  private handlers = new Set<
    (from: string, message: StreamControlMessage) => void | Promise<void>
  >();

  private constructor(address: string) {
    this.myAddress = address;
  }

  static linked(addressA: string, addressB: string): {
    a: MemoryControlChannel;
    b: MemoryControlChannel;
  } {
    const a = new MemoryControlChannel(addressA);
    const b = new MemoryControlChannel(addressB);
    a.peer = b;
    b.peer = a;
    return { a, b };
  }

  async send(peerAddress: string, message: StreamControlMessage): Promise<void> {
    if (this.peer === null) {
      throw new TransferTransportError('MemoryControlChannel: not linked');
    }
    if (peerAddress !== this.peer.myAddress) {
      throw new TransferTransportError(
        `MemoryControlChannel: peer mismatch (expected ${this.peer.myAddress}, got ${peerAddress})`,
      );
    }
    const target = this.peer;
    const from = this.myAddress;
    // Awaiting handler completion is critical: callers (the engine) rely on
    // stream-init being processed BEFORE chunk POSTs go out. Real transports
    // (HTTP/WS) await delivery similarly.
    for (const handler of [...target.handlers]) {
      try {
        await handler(from, message);
      } catch (err) {
        console.error('[MemoryControlChannel] handler error:', err);
        throw err;
      }
    }
  }

  onMessage(
    handler: (from: string, message: StreamControlMessage) => void | Promise<void>,
  ): () => void {
    this.handlers.add(handler);
    return () => this.handlers.delete(handler);
  }
}

/**
 * In-process chunk transport. Routes 0x11 envelopes directly to a paired
 * receiver-engine via a registered hook. Used in tests and same-process
 * usage to avoid an HTTP roundtrip.
 */
export class MemoryTransferTransport implements ITransferTransport {
  private peer: MemoryTransferTransport | null = null;
  private chunkHandler:
    | ((from: string, streamId: string, laneId: number, seq: bigint, bytes: Uint8Array) => Promise<ChunkAck>)
    | null = null;
  private resumeProvider:
    | ((from: string, streamId: string) => Promise<TransferResumeState | null>)
    | null = null;
  private myAddress: string;

  private constructor(address: string) {
    this.myAddress = address;
  }

  static linked(addressA: string, addressB: string): {
    a: MemoryTransferTransport;
    b: MemoryTransferTransport;
  } {
    const a = new MemoryTransferTransport(addressA);
    const b = new MemoryTransferTransport(addressB);
    a.peer = b;
    b.peer = a;
    return { a, b };
  }

  /** Receiver-side: register the handler invoked when a chunk arrives. */
  setChunkHandler(
    handler: (
      from: string,
      streamId: string,
      laneId: number,
      seq: bigint,
      bytes: Uint8Array,
    ) => Promise<ChunkAck>,
  ): void {
    this.chunkHandler = handler;
  }

  /** Receiver-side: register the resume-state lookup. */
  setResumeProvider(
    provider: (from: string, streamId: string) => Promise<TransferResumeState | null>,
  ): void {
    this.resumeProvider = provider;
  }

  async probe(peerAddress: string): Promise<void> {
    if (this.peer === null || peerAddress !== this.peer.myAddress) {
      throw new TransferTransportError(`Memory peer ${peerAddress} not reachable`);
    }
  }

  async sendChunk(
    peerAddress: string,
    streamId: string,
    laneId: number,
    seq: number | bigint,
    bytes: Uint8Array,
    _options?: ChunkSendOptions,
  ): Promise<ChunkAck> {
    if (this.peer === null) throw new TransferTransportError('Not linked');
    if (peerAddress !== this.peer.myAddress) {
      throw new TransferTransportError(`Peer mismatch: ${peerAddress} vs ${this.peer.myAddress}`);
    }
    if (this.peer.chunkHandler === null) {
      throw new TransferTransportError('Peer has no chunk handler');
    }
    return this.peer.chunkHandler(
      this.myAddress,
      streamId,
      laneId,
      typeof seq === 'bigint' ? seq : BigInt(seq),
      bytes,
    );
  }

  async fetchResumeState(
    peerAddress: string,
    streamId: string,
  ): Promise<TransferResumeState | null> {
    if (this.peer === null) throw new TransferTransportError('Not linked');
    if (peerAddress !== this.peer.myAddress) {
      throw new TransferTransportError(`Peer mismatch: ${peerAddress} vs ${this.peer.myAddress}`);
    }
    if (this.peer.resumeProvider === null) return null;
    return this.peer.resumeProvider(this.myAddress, streamId);
  }
}
