import * as plugins from './plugins.js';
import * as protocol from '../ts_interfaces/devtools.js';

type TConnection = Awaited<ReturnType<plugins.browserRuntime.BrowserRuntimeLease['openDevTools']>>;
type TOpenOptions = Parameters<plugins.browserRuntime.BrowserRuntimeLease['openDevTools']>[0];

/** Owns both inspector streams, including partially constructed and late-opening sessions. */
export class ControllerBrowserDevToolsTransport {
  public readonly id = plugins.crypto.randomBytes(18).toString('base64url');
  public readonly identity = Symbol('controller.browser.devtools');
  public readonly abortController = new AbortController();
  public commands?: plugins.typedrequestInterfaces.TVirtualStream<'receive'>;
  public events?: plugins.typedrequestInterfaces.TVirtualStream<'send'>;
  private opening?: Promise<TConnection>;
  private connection?: TConnection;
  private drain?: Promise<void>;
  private pending = new Set<Promise<void>>();
  private eventWrites = new Set<Promise<void>>();
  private eventAdmission: Promise<void> = Promise.resolve();
  private readonly eventBuffer = new Uint8Array(protocol.controllerDevToolsWriteBytes);
  private eventBufferBytes = 0;
  private eventFlush?: Promise<void>;
  private eventFlushImmediate?: ReturnType<typeof setImmediate>;
  private settlement?: Promise<void>;
  private timer?: ReturnType<typeof setTimeout>;

  constructor(public readonly tabId: string, private readonly parentSignal: AbortSignal) {
    parentSignal.throwIfAborted();
    parentSignal.addEventListener('abort', this.parentAborted, { once: true });
  }

  private parentAborted = (): void => { this.stop(this.parentSignal.reason); };

  public async open(open: (options: TOpenOptions) => Promise<TConnection>): Promise<TConnection> {
    this.abortController.signal.throwIfAborted();
    this.opening = Promise.resolve().then(() => {
      this.abortController.signal.throwIfAborted();
      return open({ tabId: this.tabId, onMessage: message => this.sendEvent(message),
        onClose: reason => this.stop(new Error(reason)) });
    });
    const connection = await this.opening;
    this.connection = connection;
    this.abortController.signal.throwIfAborted();
    return connection;
  }

  /** Start only after both stream descriptors exist; the RPC returns before they open. */
  public start(): void {
    if (!this.commands || !this.events || !this.connection || this.drain) throw new Error('DevTools is unavailable.');
    this.abortController.signal.throwIfAborted();
    this.timer = setTimeout(() => this.stop(new Error('DevTools stream opening timed out.')), 9000);
    this.timer.unref?.();
    this.drain = this.receiveCommands();
    void this.drain.catch(error => this.stop(error));
  }

  public watchStream(stream: plugins.typedrequestInterfaces.TVirtualStream<'send' | 'receive'>): void {
    void stream.opened.catch(error => this.stop(error));
    void stream.completion.then(() => this.stop(new Error('DevTools stream ended.')), error => this.stop(error));
  }

  private async receiveCommands(): Promise<void> {
    await Promise.all([this.commands!.opened, this.events!.opened]);
    clearTimeout(this.timer);
    const decoder = new protocol.ControllerDevToolsMessageDecoder(protocol.controllerDevToolsCommandBytes);
    while (!this.abortController.signal.aborted) {
      const chunk = await this.commands!.receive();
      this.abortController.signal.throwIfAborted();
      if (chunk === undefined) { decoder.finish(); await this.commands!.accept(); return; }
      for (const message of decoder.push(chunk)) {
        this.abortController.signal.throwIfAborted();
        if (this.pending.size >= 68) throw new Error('DevTools command capacity reached.');
        // Execution may be paused in the debugger. Resume must remain admissible.
        const command = this.connection!.send(message);
        this.pending.add(command);
        void command.then(() => this.pending.delete(command), error => {
          this.pending.delete(command); this.stop(error);
        });
      }
    }
  }

  private sendEvent(message: string): Promise<void> {
    this.abortController.signal.throwIfAborted();
    if (!this.events) throw new Error('DevTools event stream is unavailable.');
    const admission = this.eventAdmission.then(async () => {
      const packet = protocol.encodeControllerDevToolsMessage(message, protocol.controllerDevToolsEventBytes);
      for (let offset = 0; offset < packet.length;) {
        this.abortController.signal.throwIfAborted();
        while (this.eventBufferBytes === this.eventBuffer.length) await this.flushEvents();
        const count = Math.min(packet.length - offset, this.eventBuffer.length - this.eventBufferBytes);
        this.eventBuffer.set(packet.subarray(offset, offset + count), this.eventBufferBytes);
        this.eventBufferBytes += count;
        offset += count;
        if (this.eventBufferBytes === this.eventBuffer.length) await this.flushEvents();
      }
      this.scheduleEventFlush();
    });
    this.eventAdmission = admission;
    return admission;
  }

  private scheduleEventFlush(): void {
    if (!this.eventBufferBytes || this.eventFlushImmediate || this.abortController.signal.aborted) return;
    // Batch same-turn small CDP events without adding a network round trip or
    // waiting for a large buffer to fill before delivering an interactive reply.
    this.eventFlushImmediate = setImmediate(() => {
      this.eventFlushImmediate = undefined;
      void this.flushEvents().catch(error => this.stop(error));
    });
  }

  private flushEvents(): Promise<void> {
    if (this.eventFlush) return this.eventFlush;
    const flushing = (async () => {
      while (this.eventWrites.size >= 8) await Promise.race(this.eventWrites);
      this.abortController.signal.throwIfAborted();
      if (!this.eventBufferBytes) return;
      // ACK-owned storage is immutable. Reuse only the one unsent accumulator.
      const bytes = this.eventBuffer.slice(0, this.eventBufferBytes);
      this.eventBufferBytes = 0;
      const write = this.events!.send(bytes)
        .catch(error => { this.stop(error); throw error; })
        .finally(() => { this.eventWrites.delete(write); });
      this.eventWrites.add(write);
      void write.catch(() => undefined);
    })();
    this.eventFlush = flushing.finally(() => {
      this.eventFlush = undefined;
      this.scheduleEventFlush();
    });
    return this.eventFlush;
  }

  public stop(reason: unknown): void {
    if (this.abortController.signal.aborted) return;
    this.abortController.abort(reason);
    clearTimeout(this.timer);
    clearImmediate(this.eventFlushImmediate);
    this.eventFlushImmediate = undefined;
    this.eventBufferBytes = 0;
    this.parentSignal.removeEventListener('abort', this.parentAborted);
    const streams = [this.commands, this.events].filter(stream => stream !== undefined);
    const aborted = streams.map(stream => Promise.resolve().then(() => stream.abort(reason)));
    // Keep ownership until every late open, native detach and transport resource settles.
    this.settlement = (async () => {
      const result = await Promise.allSettled([
        ...aborted, ...streams.flatMap(stream => [stream.opened, stream.completion, stream.closed]),
        this.drain,
        ...this.pending,
        this.eventAdmission, this.eventFlush, ...this.eventWrites,
        this.opening?.then(connection => connection.close(), () => undefined),
      ]);
      // Open/completion rejections are expected during abort. Native detach failures are not.
      const detached = result.at(-1)!;
      if (detached.status === 'rejected') throw detached.reason;
    })();
    void this.settlement.catch(() => undefined);
  }

  public async close(reason = new Error('DevTools closed.')): Promise<void> {
    this.stop(reason);
    let timer: ReturnType<typeof setTimeout> | undefined;
    try {
      await Promise.race([this.settlement, new Promise<never>((_, reject) => {
        timer = setTimeout(() => reject(new Error('DevTools cleanup remains pending.')), 8000);
        timer.unref?.();
      })]);
    } finally { clearTimeout(timer); }
  }
}