import { Mutex } from '@livekit/mutex';
import {
  type AddTrackRequest,
  ClientConfigSetting,
  ClientConfiguration,
  type ConnectionQualityUpdate,
  DataChannelInfo,
  DataChannelReceiveState,
  DataPacket,
  DataPacket_Kind,
  DataTrackSubscriberHandles,
  DisconnectReason,
  EncryptedPacket,
  EncryptedPacketPayload,
  Encryption_Type,
  type JoinResponse,
  type LeaveRequest,
  LeaveRequest_Action,
  MediaSectionsRequirement,
  ParticipantInfo,
  ConnectionQuality as ProtoConnectionQuality,
  PublishDataTrackResponse,
  ReconnectReason,
  type ReconnectResponse,
  type RegionSettings,
  RequestResponse,
  Room as RoomModel,
  RoomMovedResponse,
  RpcAck,
  ServerInfo,
  SessionDescription,
  SignalTarget,
  SpeakerInfo,
  type StreamStateUpdate,
  SubscribedQualityUpdate,
  type SubscriptionPermissionUpdate,
  type SubscriptionResponse,
  SyncState,
  TrackInfo,
  type TrackPublishedResponse,
  TrackUnpublishedResponse,
  Transcription,
  UnpublishDataTrackResponse,
  UpdateSubscription,
  type UserPacket,
} from '@livekit/protocol';
import { EventEmitter } from 'events';
import type { MediaAttributes } from 'sdp-transform';
import type TypedEventEmitter from 'typed-emitter';
import type { SignalOptions } from '../api/SignalClient';
import {
  SignalClient,
  SignalConnectionState,
  toProtoSessionDescription,
} from '../api/SignalClient';
import type { BaseE2EEManager } from '../e2ee/E2eeManager';
import { asEncryptablePacket, isInsertableStreamSupported } from '../e2ee/utils';
import {
  hasFrameMetadataPublishOptions,
  isFrameMetadataSupported,
  shouldUseFrameMetadataScriptTransform,
} from '../frameMetadata/utils';
import log, { LoggerNames, getLogger } from '../logger';
import type { InternalRoomOptions } from '../options';
import type { NonSharedUint8Array } from '../type-polyfills/non-shared-typed-arrays';
import TypedPromise from '../utils/TypedPromise';
import { TTLMap } from '../utils/ttlmap';
import PCTransport, { PCEvents } from './PCTransport';
import { PCTransportManager, PCTransportState } from './PCTransportManager';
import type { ReconnectContext, ReconnectPolicy } from './ReconnectPolicy';
import { DataChannelManager } from './data-channel/DataChannelManager';
import type { FlowControlledDataChannel } from './data-channel/FlowControlledDataChannel';
import type { LossyDataChannel } from './data-channel/LossyDataChannel';
import type { ReliableDataChannel } from './data-channel/ReliableDataChannel';
import { DataChannelKind } from './data-channel/types';
import { DataTrackInfo } from './data-track/types';
import { roomConnectOptionDefaults } from './defaults';
import {
  ConnectionError,
  ConnectionErrorReason,
  NegotiationError,
  PublishDataError,
  SignalReconnectError,
  TrackInvalidError,
  UnexpectedConnectionState,
} from './errors';
import { EngineEvent } from './events';
import CriticalTimers from './timers';
import type LocalTrack from './track/LocalTrack';
import type LocalTrackPublication from './track/LocalTrackPublication';
import LocalVideoTrack from './track/LocalVideoTrack';
import type { SimulcastTrackInfo } from './track/LocalVideoTrack';
import type RemoteTrackPublication from './track/RemoteTrackPublication';
import type { Track } from './track/Track';
import type { TrackPublishOptions, VideoCodec } from './track/options';
import { getTrackPublicationInfo } from './track/utils';
import type { LoggerOptions } from './types';
import {
  isPublisherOfferWithJoinSupported,
  isReactNative,
  isVideoCodec,
  isVideoTrack,
  isWeb,
  negotiateDependencyDescriptor,
  sleep,
  supportsAddTrack,
  supportsTransceiver,
  toHttpUrl,
} from './utils';

const minReconnectWait = 2 * 1000;
const leaveReconnect = 'leave-reconnect';

/**
 * How long local connection quality must stay `LOST` while connected and publishing before we
 * force a full reconnect — `LOST` is the server's verdict that it isn't receiving our media.
 */
const connectionQualityLostTimeout = 10 * 1000;
const reliabeReceiveStateTTL = 30_000;

const initialMediaSectionsAudio = 3;
const initialMediaSectionsVideo = 3;

enum PCState {
  New,
  Connected,
  Disconnected,
  Reconnecting,
  Closed,
}

export { DataChannelKind };

// Default data-channel max message size (bytes), used when the remote SDP
// answer does not advertise an `a=max-message-size` attribute (RFC 8841).
// `0` means "no limit".
const DEFAULT_MAX_MESSAGE_SIZE = 64_000;

/** @internal */
export default class RTCEngine extends (EventEmitter as new () => TypedEventEmitter<EngineEventCallbacks>) {
  client: SignalClient;

  rtcConfig: RTCConfiguration = {};

  peerConnectionTimeout: number = roomConnectOptionDefaults.peerConnectionTimeout;

  fullReconnectOnNext: boolean = false;

  pcManager?: PCTransportManager;

  /**
   * @internal
   */
  latestJoinResponse?: JoinResponse;

  /**
   * @internal
   */
  latestRemoteOfferId: number = 0;

  /** @internal */
  e2eeManager: BaseE2EEManager | undefined;

  get isClosed() {
    return this._isClosed;
  }

  get isNewlyCreated() {
    return this._isNewlyCreated;
  }

  get pendingReconnect() {
    return !!this.reconnectTimeout;
  }

  /**
   * Owns the data channels: the three flow-controlled publisher wrappers (engine-lifetime; the
   * RTCDataChannel handles underneath are attached/detached as peer connections come and go, with
   * waiter invalidation built into the turnover) plus the subscriber-side receive handles.
   */
  private dataChannels: DataChannelManager;

  private get reliableChannel(): ReliableDataChannel {
    return this.dataChannels.reliable;
  }

  private get lossyChannel(): LossyDataChannel {
    return this.dataChannels.lossy;
  }

  private get dataTrackChannel(): LossyDataChannel {
    return this.dataChannels.dataTrack;
  }

  private subscriberPrimary: boolean = false;

  private pcState: PCState = PCState.New;

  private _isClosed: boolean = true;

  private _isNewlyCreated: boolean = true;

  private pendingTrackResolvers: {
    [key: string]: { resolve: (info: TrackInfo) => void; reject: () => void };
  } = {};

  // keep join info around for reconnect, this could be a region url
  private url?: string;

  private token?: string;

  private signalOpts?: SignalOptions;

  private reconnectAttempts: number = 0;

  private reconnectStart: number = 0;

  private clientConfiguration?: ClientConfiguration;

  private attemptingReconnect: boolean = false;

  private reconnectPolicy: ReconnectPolicy;

  private reconnectTimeout?: ReturnType<typeof setTimeout>;

  private participantSid?: string;

  /** keeps track of how often an initial join connection has been tried */
  private joinAttempts: number = 0;

  /** specifies how often an initial join connection is allowed to retry */
  private maxJoinAttempts: number = 1;

  private closingLock: Mutex;

  private dataProcessLock: Mutex;

  private shouldFailNext: boolean = false;

  private shouldFailOnV1Path: boolean = false;

  private regionStrategy?: RegionStrategy;

  private log = log;

  private loggerOptions: LoggerOptions;

  private publisherConnectionPromise: Promise<void> | undefined;

  private reliableReceivedState: TTLMap<string, number> = new TTLMap(reliabeReceiveStateTTL);

  private midToTrackId: { [key: string]: string } = {};

  /** used to indicate whether the browser is currently waiting to reconnect */
  private isWaitingForNetworkReconnect: boolean = false;

  /** set while the local participant's connection quality is `LOST`; forces a full reconnect on timeout */
  private lostQualityTimeout?: ReturnType<typeof setTimeout>;

  /** timestamp (ms) the primary transport entered `CONNECTING`, used to bound how long we tolerate it */
  private transportConnectingSince?: number;

  constructor(private options: InternalRoomOptions) {
    super();
    this.log = getLogger(options.loggerName ?? LoggerNames.Engine, () => this.logContext);
    this.loggerOptions = {
      loggerName: options.loggerName,
      loggerContextCb: () => this.logContext,
    };
    this.client = new SignalClient(undefined, this.loggerOptions);
    this.client.signalLatency = this.options.expSignalLatency;
    this.reconnectPolicy = this.options.reconnectPolicy;
    this.closingLock = new Mutex();
    this.dataProcessLock = new Mutex();
    this.dataChannels = new DataChannelManager({
      isEngineClosed: () => this.isClosed,
      isReconnecting: () => this.attemptingReconnect,
      onDataMessage: (message) => this.handleDataMessage(message),
      onDataTrackMessage: (message) => this.handleDataTrackMessage(message),
      onDataError: (event) => this.handleDataError(event),
      onChannelClose: (kind) => this.handleDataChannelClose(kind)(),
      onBufferStatusChanged: (kind, isLow) =>
        this.emit(EngineEvent.DCBufferStatusChanged, isLow, kind),
    });

    this.client.onParticipantUpdate = (updates) =>
      this.emit(EngineEvent.ParticipantUpdate, updates);
    this.client.onConnectionQuality = (update) => {
      this.handleLocalConnectionQuality(update);
      this.emit(EngineEvent.ConnectionQualityUpdate, update);
    };
    this.client.onRoomUpdate = (update) => this.emit(EngineEvent.RoomUpdate, update);
    this.client.onSubscriptionError = (resp) => this.emit(EngineEvent.SubscriptionError, resp);
    this.client.onSubscriptionPermissionUpdate = (update) =>
      this.emit(EngineEvent.SubscriptionPermissionUpdate, update);
    this.client.onSpeakersChanged = (update) => this.emit(EngineEvent.SpeakersChanged, update);
    this.client.onStreamStateUpdate = (update) => this.emit(EngineEvent.StreamStateChanged, update);
    this.client.onRequestResponse = (response) =>
      this.emit(EngineEvent.SignalRequestResponse, response);
    this.client.onParticipantUpdate = (updates) =>
      this.emit(EngineEvent.ParticipantUpdate, updates);
    this.client.onJoined = (joinResponse) => this.emit(EngineEvent.Joined, joinResponse);
  }

  /** @internal */
  get logContext() {
    return {
      room: this.latestJoinResponse?.room?.name,
      roomID: this.latestJoinResponse?.room?.sid,
      participant: this.latestJoinResponse?.participant?.identity,
      participantID: this.participantSid,
    };
  }

  async join(
    url: string,
    token: string,
    opts: SignalOptions,
    abortSignal?: AbortSignal,
    /** setting this to true results in dual peer connection mode being used */
    useV0Path: boolean = false,
  ): Promise<{ joinResponse: JoinResponse; serverInfo: Partial<ServerInfo> }> {
    this._isNewlyCreated = false;
    this.url = url;
    this.token = token;
    this.signalOpts = opts;
    this.maxJoinAttempts = opts.maxRetries;
    try {
      this.joinAttempts += 1;

      this.setupSignalClientCallbacks();
      // Whether the initial publisher offer is bundled with the join request. Computed once and
      // reused after the join below. Only the (non-Firefox) offer-with-join path does this.
      const sendOfferWithJoin = !useV0Path && isPublisherOfferWithJoinSupported();

      let offerProto: SessionDescription | undefined;
      if (sendOfferWithJoin) {
        if (!this.pcManager) {
          // Firefox is excluded from offer-with-join (see isPublisherOfferWithJoinSupported):
          // customers reported ICE connectivity problems for FF on this path (#1919) that we were
          // never able to reproduce, so out of caution FF stays on the deferred path below. The
          // exact cause is unknown — note that ICE gathering does not actually start here, since
          // createInitialOffer() defers setLocalDescription (via pendingInitialOffer) until the
          // answer is applied, after updateConfiguration() has set the server's TURN servers.
          await this.configure();
          this.applyInitialPublisherLayout();
        }
        const offer = await this.pcManager?.publisher.createInitialOffer();
        if (offer) {
          offerProto = toProtoSessionDescription(offer.offer, offer.offerId);
        }
      }

      if (abortSignal?.aborted) {
        throw ConnectionError.cancelled('Connection aborted');
      }

      if (!useV0Path && this.shouldFailOnV1Path) {
        this.shouldFailOnV1Path = false;
        throw ConnectionError.serviceNotFound('Simulated v1 path failure', 'v0-rtc');
      }
      const joinResponse = await this.client.join(
        url,
        token,
        opts,
        abortSignal,
        useV0Path,
        offerProto,
      );
      this._isClosed = false;
      this.latestJoinResponse = joinResponse;
      this.participantSid = joinResponse.participant?.sid;

      this.subscriberPrimary = joinResponse.subscriberPrimary;
      if (sendOfferWithJoin) {
        this.pcManager?.updateConfiguration(this.makeRTCConfiguration(joinResponse));
      } else {
        if (!this.pcManager) {
          // Deferred path (Firefox, and V0): configure with the join response so the PC picks up
          // the server's ICE servers and topology, then negotiate separately rather than bundling
          // the offer with the join.
          await this.configure(joinResponse, !useV0Path);
          if (!useV0Path) {
            // The V1 first offer must carry the media layout so Firefox binds receive decoders for
            // subscribed tracks — without it, subscribed audio/video arrive as RTP but
            // never decode. V0 (legacy dual-PC) keeps its original lazy behavior.
            this.applyInitialPublisherLayout();
          }
        }
        // create offer
        if (!this.subscriberPrimary || joinResponse.fastPublish) {
          this.negotiate().catch((err) => {
            this.log.error(err);
          });
        }
      }

      this.registerOnLineListener();
      this.clientConfiguration = joinResponse.clientConfiguration;
      this.emit(EngineEvent.SignalConnected, joinResponse);

      let serverInfo: Partial<ServerInfo> | undefined = joinResponse.serverInfo;
      if (!serverInfo) {
        serverInfo = { version: joinResponse.serverVersion, region: joinResponse.serverRegion };
      }
      this.log.info(
        `connected to Livekit Server ${Object.entries(serverInfo)
          .map(([key, value]) => `${key}: ${value}`)
          .join(', ')}`,
      );

      return { joinResponse, serverInfo };
    } catch (e) {
      if (e instanceof ConnectionError) {
        if (e.reason === ConnectionErrorReason.ServerUnreachable) {
          this.log.warn(
            `Couldn't connect to server, attempt ${this.joinAttempts} of ${this.maxJoinAttempts}`,
          );
          if (this.joinAttempts < this.maxJoinAttempts) {
            return this.join(url, token, opts, abortSignal, useV0Path);
          }
        } else if (e.reason === ConnectionErrorReason.ServiceNotFound) {
          this.log.warn(`Initial connection failed: ${e.message} – Retrying`);
          if (this.pcManager) {
            this.pcManager.onStateChange = undefined;
            await this.cleanupPeerConnections();
          }
          return this.join(url, token, opts, abortSignal, true);
        }
      }
      throw e;
    }
  }

  async close() {
    const unlock = await this.closingLock.lock();
    if (this.isClosed) {
      unlock();
      return;
    }
    try {
      this._isClosed = true;
      this.joinAttempts = 0;
      this.emit(EngineEvent.Closing);
      this.removeAllListeners();
      this.deregisterOnLineListener();
      this.clearPendingReconnect();
      this.clearLostQualityTimeout();
      this.cleanupLossyDataStats();
      await this.cleanupPeerConnections();
      await this.cleanupClient();
    } finally {
      unlock();
    }
  }

  async cleanupPeerConnections() {
    this.dataChannels.teardown();

    await this.pcManager?.close();
    this.pcManager = undefined;
    // the connecting timestamp belongs to the transports we just tore down
    this.transportConnectingSince = undefined;

    this.reliableReceivedState.clear();
  }

  cleanupLossyDataStats() {
    this.lossyChannel.stopThresholdTuning();
  }

  async cleanupClient() {
    await this.client.close();
    this.client.resetCallbacks();
    // Any in-flight addTrack requests are orphaned by the signal reconnect — the new session
    // won't deliver `trackPublishedResponse` for them, so reject the pending resolvers and
    // clear the map. Otherwise a subsequent `addTrack` call with the same client id (e.g. a
    // publish retry after a `NegotiationError`) throws `TrackInvalidError`.
    for (const cid of Object.keys(this.pendingTrackResolvers)) {
      this.pendingTrackResolvers[cid].reject();
    }
    this.pendingTrackResolvers = {};
  }

  addTrack(req: AddTrackRequest): Promise<TrackInfo> {
    if (this.pendingTrackResolvers[req.cid]) {
      throw new TrackInvalidError('a track with the same ID has already been published');
    }
    return new Promise<TrackInfo>((resolve, reject) => {
      const publicationTimeout = CriticalTimers.setTimeout(() => {
        delete this.pendingTrackResolvers[req.cid];
        reject(
          ConnectionError.timeout('publication of local track timed out, no response from server'),
        );
      }, 10_000);
      this.pendingTrackResolvers[req.cid] = {
        resolve: (info: TrackInfo) => {
          CriticalTimers.clearTimeout(publicationTimeout);
          resolve(info);
        },
        reject: () => {
          CriticalTimers.clearTimeout(publicationTimeout);
          reject(new Error('Cancelled publication by calling unpublish'));
        },
      };
      this.client.sendAddTrack(req);
    });
  }

  /**
   * Removes sender from PeerConnection, returning true if it was removed successfully
   * and a negotiation is necessary
   * @param sender
   * @returns
   */
  removeTrack(sender: RTCRtpSender): boolean {
    if (sender.track && this.pendingTrackResolvers[sender.track.id]) {
      const { reject } = this.pendingTrackResolvers[sender.track.id];
      if (reject) {
        reject();
      }
      delete this.pendingTrackResolvers[sender.track.id];
    }
    try {
      this.pcManager!.removeTrack(sender);
      return true;
    } catch (e: unknown) {
      this.log.warn('failed to remove track', { error: e });
    }
    return false;
  }

  updateMuteStatus(trackSid: string, muted: boolean) {
    this.client.sendMuteTrack(trackSid, muted);
  }

  get dataSubscriberReadyState(): string | undefined {
    return this.dataChannelForKind(DataChannelKind.RELIABLE, true)?.readyState;
  }

  async getConnectedServerAddress(): Promise<string | undefined> {
    return this.pcManager?.getConnectedAddress();
  }

  /* @internal */
  setRegionStrategy(strategy: RegionStrategy | undefined) {
    this.regionStrategy = strategy;
  }

  private async configure(joinResponse?: JoinResponse, useSinglePeerConnection?: boolean) {
    // already configured
    if (this.pcManager && this.pcManager.currentState !== PCTransportState.NEW) {
      return;
    }

    if (!joinResponse) {
      const rtcConfig = this.makeRTCConfiguration();
      this.pcManager = new PCTransportManager('publisher-only', this.loggerOptions, rtcConfig);
    } else {
      this.participantSid = joinResponse.participant?.sid;
      const rtcConfig = this.makeRTCConfiguration(joinResponse);
      this.pcManager = new PCTransportManager(
        useSinglePeerConnection
          ? 'publisher-only'
          : joinResponse.subscriberPrimary
            ? 'subscriber-primary'
            : 'publisher-primary',
        this.loggerOptions,
        rtcConfig,
      );
    }

    this.emit(EngineEvent.TransportsCreated, this.pcManager.publisher, this.pcManager.subscriber);

    this.pcManager.onIceCandidate = (candidate, target) => {
      this.client.sendIceCandidate(candidate, target);
    };

    this.pcManager.onPublisherOffer = (offer, offerId) => {
      this.client.sendOffer(offer, offerId);
    };

    this.pcManager.onDataChannel = this.handleDataChannel;
    this.pcManager.onStateChange = async (connectionState, publisherState, subscriberState) => {
      this.log.debug(`primary PC state changed ${connectionState}`);

      // Record when the primary transport actually entered CONNECTING so
      // verifyTransport() can bound how long we tolerate it. Deriving it from the
      // real transition (this handler only fires on state changes) rather than from
      // observation time keeps it from going stale across peer-connection rebuilds.
      if (connectionState === PCTransportState.CONNECTING) {
        this.transportConnectingSince = Date.now();
      } else {
        this.transportConnectingSince = undefined;
      }

      if (['closed', 'disconnected', 'failed'].includes(publisherState)) {
        // reset publisher connection promise
        this.publisherConnectionPromise = undefined;
      }
      if (connectionState === PCTransportState.CONNECTED) {
        const shouldEmit = this.pcState === PCState.New;
        this.pcState = PCState.Connected;
        if (shouldEmit) {
          this.emit(EngineEvent.Connected, this.latestJoinResponse!);
        }
      } else if (connectionState === PCTransportState.FAILED) {
        // on Safari, PeerConnection will switch to 'disconnected' during renegotiation
        if (this.pcState === PCState.Connected || this.pcState === PCState.Reconnecting) {
          this.pcState = PCState.Disconnected;

          this.handleDisconnect(
            'peerconnection failed',
            subscriberState === 'failed'
              ? ReconnectReason.RR_SUBSCRIBER_FAILED
              : ReconnectReason.RR_PUBLISHER_FAILED,
          );
        }
      }

      // detect cases where both signal client and peer connection are severed and assume that user has lost network connection
      const isSignalSevered =
        this.client.isDisconnected ||
        this.client.currentState === SignalConnectionState.RECONNECTING;
      const isPCSevered = [
        PCTransportState.FAILED,
        PCTransportState.CLOSING,
        PCTransportState.CLOSED,
      ].includes(connectionState);
      if (isSignalSevered && isPCSevered && !this._isClosed) {
        this.emit(EngineEvent.Offline);
      }
    };
    this.pcManager.onTrack = (ev: RTCTrackEvent) => {
      // this fires after the underlying transceiver is stopped and potentially
      // peer connection closed, so do not bubble up if there are no streams
      if (ev.streams.length === 0) return;
      this.emit(EngineEvent.MediaTrackAdded, ev.track, ev.streams[0], ev.receiver);
    };
  }

  private setupSignalClientCallbacks() {
    // configure signaling client
    this.client.onAnswer = async (sd, offerId, midToTrackId) => {
      if (!this.pcManager) {
        return;
      }
      this.log.debug('received server answer', {
        RTCSdpType: sd.type,
        sdp: sd.sdp,
        midToTrackId,
      });
      this.midToTrackId = midToTrackId;
      await this.pcManager.setPublisherAnswer(sd, offerId);
    };

    // add candidate on trickle
    this.client.onTrickle = (candidate, target) => {
      if (!this.pcManager) {
        return;
      }
      this.log.debug('got ICE candidate from peer', { candidate, target });
      this.pcManager.addIceCandidate(candidate, target);
    };

    // when server creates an offer for the client
    this.client.onOffer = async (sd, offerId, midToTrackId) => {
      this.latestRemoteOfferId = offerId;
      if (!this.pcManager) {
        return;
      }
      this.midToTrackId = midToTrackId;
      const answer = await this.pcManager.createSubscriberAnswerFromOffer(sd, offerId);
      if (answer) {
        this.client.sendAnswer(answer, offerId);
      }
    };

    this.client.onLocalTrackPublished = (res: TrackPublishedResponse) => {
      this.log.debug('received trackPublishedResponse', {
        cid: res.cid,
        track: res.track?.sid,
      });
      if (!this.pendingTrackResolvers[res.cid]) {
        this.log.error(`missing track resolver for ${res.cid}`, { cid: res.cid });
        return;
      }
      const { resolve } = this.pendingTrackResolvers[res.cid];
      delete this.pendingTrackResolvers[res.cid];
      resolve(res.track!);
    };

    this.client.onLocalTrackUnpublished = (response: TrackUnpublishedResponse) => {
      this.emit(EngineEvent.LocalTrackUnpublished, response);
    };

    this.client.onLocalTrackSubscribed = (trackSid: string) => {
      this.emit(EngineEvent.LocalTrackSubscribed, trackSid);
    };

    this.client.onTokenRefresh = (token: string) => {
      this.token = token;
      this.emit(EngineEvent.TokenRefreshed, token);
    };

    this.client.onRemoteMuteChanged = (trackSid: string, muted: boolean) => {
      this.emit(EngineEvent.RemoteMute, trackSid, muted);
    };

    this.client.onSubscribedQualityUpdate = (update: SubscribedQualityUpdate) => {
      this.emit(EngineEvent.SubscribedQualityUpdate, update);
    };

    this.client.onRoomMoved = (res: RoomMovedResponse) => {
      this.participantSid = res.participant?.sid;
      if (this.latestJoinResponse) {
        this.latestJoinResponse.room = res.room;
      }
      this.emit(EngineEvent.RoomMoved, res);
    };

    this.client.onMediaSectionsRequirement = (requirement: MediaSectionsRequirement) => {
      this.addMediaSections(requirement.numAudios, requirement.numVideos);
      this.negotiate();
    };

    this.client.onPublishDataTrackResponse = (event: PublishDataTrackResponse) => {
      this.emit(EngineEvent.PublishDataTrackResponse, event);
    };

    this.client.onUnPublishDataTrackResponse = (event: UnpublishDataTrackResponse) => {
      this.emit(EngineEvent.UnPublishDataTrackResponse, event);
    };

    this.client.onDataTrackSubscriberHandles = (event: DataTrackSubscriberHandles) => {
      this.emit(EngineEvent.DataTrackSubscriberHandles, event);
    };

    this.client.onClose = () => {
      this.handleDisconnect('signal', ReconnectReason.RR_SIGNAL_DISCONNECTED);
    };

    this.client.onLeave = (leave: LeaveRequest) => {
      this.log.info(`client leave request received (action=${leave?.action})`, {
        reason: leave?.reason,
      });
      if (leave.regions) {
        this.log.debug('updating regions');
        this.emit(EngineEvent.ServerRegionsReported, leave.regions);
      }
      switch (leave.action) {
        case LeaveRequest_Action.DISCONNECT:
          this.emit(EngineEvent.Disconnected, leave?.reason);
          this.close();
          break;
        case LeaveRequest_Action.RECONNECT:
          this.fullReconnectOnNext = true;
          // reconnect immediately instead of waiting for next attempt
          this.handleDisconnect(leaveReconnect);
          break;
        case LeaveRequest_Action.RESUME:
          // reconnect immediately instead of waiting for next attempt
          this.handleDisconnect(leaveReconnect);
        default:
          break;
      }
    };
  }

  private makeRTCConfiguration(
    serverResponse?: JoinResponse | ReconnectResponse,
  ): RTCConfiguration {
    const rtcConfig = { ...this.rtcConfig };

    // E2EE and packet trailer extraction both rely on encoded frame transforms.
    // Only opt into the createEncodedStreams flavor when that path will be
    // used; RTCRtpScriptTransform does not need the PeerConnection flag.
    const needsInsertableStreams =
      this.signalOpts?.e2eeEnabled ||
      (this.frameMetadataWorker && !shouldUseFrameMetadataScriptTransform());

    if (needsInsertableStreams && isInsertableStreamSupported()) {
      this.log.debug('E2EE - setting up transports with insertable streams');
      //  this makes sure that no data is sent before the transforms are ready
      // @ts-ignore
      rtcConfig.encodedInsertableStreams = true;
    }

    // @ts-ignore
    rtcConfig.sdpSemantics = 'unified-plan';
    // @ts-ignore
    rtcConfig.continualGatheringPolicy = 'gather_continually';

    if (!serverResponse) {
      return rtcConfig;
    }

    // update ICE servers before creating PeerConnection
    if (serverResponse.iceServers && !rtcConfig.iceServers) {
      const rtcIceServers: RTCIceServer[] = [];
      serverResponse.iceServers.forEach((iceServer) => {
        const rtcIceServer: RTCIceServer = {
          urls: iceServer.urls,
        };
        if (iceServer.username) rtcIceServer.username = iceServer.username;
        if (iceServer.credential) {
          rtcIceServer.credential = iceServer.credential;
        }
        rtcIceServers.push(rtcIceServer);
      });
      rtcConfig.iceServers = rtcIceServers;
    }

    if (
      serverResponse.clientConfiguration &&
      serverResponse.clientConfiguration.forceRelay === ClientConfigSetting.ENABLED
    ) {
      rtcConfig.iceTransportPolicy = 'relay';
    }

    return rtcConfig;
  }

  /**
   * Populate the publisher PC so its first offer carries the data channels + recvonly media
   * sections. Required for every V1 connection: Firefox only binds receive decoders for media
   * present in that first offer, and the offer-with-join path needs the sections to
   * build a meaningful initial offer. Must be called on a configured pcManager.
   */
  private applyInitialPublisherLayout() {
    this.createDataChannels();
    /**
     * Native libwebrtc does not support pre-populating the media sections,
     * so we skip it for React Native.
     *
     * Related: https://github.com/livekit/rust-sdks/pull/1151
     */
    if (!isReactNative()) {
      this.addMediaSections(initialMediaSectionsAudio, initialMediaSectionsVideo);
    }
  }

  private addMediaSections(numAudios: number, numVideos: number) {
    const transceiverInit: RTCRtpTransceiverInit = { direction: 'recvonly' };
    for (let i: number = 0; i < numAudios; i++) {
      this.pcManager?.addPublisherTransceiverOfKind('audio', transceiverInit);
    }
    // media only arrives on these sections when there is no subscriber connection to arrive on
    const receivesMedia = this.pcManager?.mode === 'publisher-only';
    for (let i: number = 0; i < numVideos; i++) {
      const transceiver = this.pcManager?.addPublisherTransceiverOfKind('video', transceiverInit);
      if (receivesMedia && transceiver) {
        const negotiated = negotiateDependencyDescriptor(transceiver);

        this.log.debug('dependency descriptor negotiated for received video', { negotiated });
      }
    }
  }

  private createDataChannels() {
    if (!this.pcManager) {
      return;
    }

    this.dataChannels.createPublisherChannels(this.pcManager);
  }

  private handleDataChannel = async ({ channel }: RTCDataChannelEvent) => {
    if (!channel) {
      return;
    }
    if (this.dataChannels.adoptSubscriberChannel(channel)) {
      this.log.debug(`on data channel ${channel.id}, ${channel.label}`);
    }
  };

  /** Normalizes an incoming data-channel message into bytes, or logs and returns undefined. */
  private async decodeDataMessage(message: MessageEvent): Promise<Uint8Array | undefined> {
    if (message.data instanceof ArrayBuffer) {
      return new Uint8Array(message.data);
    }
    if (message.data instanceof Blob) {
      return new Uint8Array(await message.data.arrayBuffer());
    }
    this.log.error('unsupported data type', { data: message.data });
    return undefined;
  }

  private handleDataMessage = async (message: MessageEvent) => {
    // make sure to respect incoming data message order by processing message events one after the other
    const unlock = await this.dataProcessLock.lock();
    try {
      const bytes = await this.decodeDataMessage(message);
      if (!bytes) {
        return;
      }
      const dp = DataPacket.fromBinary(bytes);

      if (dp.sequence > 0 && dp.participantSid !== '') {
        const lastSeq = this.reliableReceivedState.get(dp.participantSid);
        if (lastSeq && dp.sequence <= lastSeq) {
          // ignore duplicate or out-of-order packets in reliable channel
          return;
        }
        this.reliableReceivedState.set(dp.participantSid, dp.sequence);
      }

      if (dp.value?.case === 'speaker') {
        // dispatch speaker updates
        this.emit(EngineEvent.ActiveSpeakersUpdate, dp.value.value.speakers);
      } else if (dp.value?.case === 'encryptedPacket') {
        if (!this.e2eeManager) {
          this.log.error('Received encrypted packet but E2EE not set up');
          return;
        }
        const decryptedData = await this.e2eeManager?.handleEncryptedData(
          dp.value.value.encryptedValue as NonSharedUint8Array,
          dp.value.value.iv as NonSharedUint8Array,
          dp.participantIdentity,
          dp.value.value.keyIndex,
        );
        const decryptedPacket = EncryptedPacketPayload.fromBinary(decryptedData.payload);
        const newDp = new DataPacket({
          value: decryptedPacket.value,
          participantIdentity: dp.participantIdentity,
          participantSid: dp.participantSid,
        });
        if (newDp.value?.case === 'user') {
          // compatibility
          applyUserDataCompat(newDp, newDp.value.value);
        }
        this.emit(EngineEvent.DataPacketReceived, newDp, dp.value.value.encryptionType);
      } else {
        if (dp.value?.case === 'user') {
          // compatibility
          applyUserDataCompat(dp, dp.value.value);
        }
        this.emit(EngineEvent.DataPacketReceived, dp, Encryption_Type.NONE);
      }
    } finally {
      unlock();
    }
  };

  private handleDataTrackMessage = async (message: MessageEvent) => {
    const bytes = await this.decodeDataMessage(message);
    if (!bytes) {
      return;
    }
    this.emit('dataTrackPacketReceived', bytes);
  };

  private handleDataError = (event: Event) => {
    // Errors fired while we're tearing the connection down (e.g. the SCTP transport aborting as
    // the peer connection closes) carry no actionable information — the channel is going away
    // regardless. Suppress them so a graceful disconnect doesn't surface spurious errors.
    // See livekit/client-sdk-js#1953.
    if (this._isClosed) {
      return;
    }

    const channel = event.currentTarget as RTCDataChannel;
    const channelKind = channel.maxRetransmits === 0 ? 'lossy' : 'reliable';

    if (typeof RTCErrorEvent !== 'undefined' && event instanceof RTCErrorEvent && event.error) {
      const { error } = event;
      this.log.error(`DataChannel error on ${channelKind}: ${error.message}`, {
        error,
        errorDetail: error.errorDetail,
        sctpCauseCode: error.sctpCauseCode,
      });
    } else {
      this.log.error(`Unknown DataChannel error on ${channelKind}`, { event });
    }
  };

  private handleDataChannelClose = (kind: DataChannelKind) => () => {
    // A publisher DC closing while the session is up and the publisher PC is still
    // connected is the signature of an oversized message having aborted the channel
    // (see livekit/rust-sdks#1137). Surface it; do not attempt renegotiation.
    if (!this._isClosed && this.pcManager?.publisher.getConnectionState() === 'connected') {
      this.log.error(
        `publisher data channel '${DataChannelKind[kind]}' closed unexpectedly`,
        this.logContext,
      );
    }
  };

  async createSender(
    track: LocalTrack,
    opts: TrackPublishOptions,
    encodings?: RTCRtpEncodingParameters[],
  ) {
    let sender: RTCRtpSender;
    if (supportsTransceiver()) {
      sender = await this.createTransceiverRTCRtpSender(track, opts, encodings);
    } else if (supportsAddTrack()) {
      this.log.warn('using add-track fallback');
      sender = await this.createRTCRtpSender(track.mediaStreamTrack);
    } else {
      throw new UnexpectedConnectionState('Required webRTC APIs not supported on this device');
    }
    this.setupFrameMetadataSender(sender, opts);
    return sender;
  }

  async createSimulcastSender(
    track: LocalVideoTrack,
    simulcastTrack: SimulcastTrackInfo,
    opts: TrackPublishOptions,
    encodings?: RTCRtpEncodingParameters[],
  ) {
    let sender: RTCRtpSender | undefined;
    if (supportsTransceiver()) {
      sender = await this.createSimulcastTransceiverSender(track, simulcastTrack, opts, encodings);
    } else if (supportsAddTrack()) {
      this.log.debug('using add-track fallback');
      sender = await this.createRTCRtpSender(track.mediaStreamTrack);
    } else {
      throw new UnexpectedConnectionState('Cannot stream on this device');
    }
    if (sender) {
      this.setupFrameMetadataSender(sender, opts);
    }
    return sender;
  }

  private get frameMetadataWorker(): Worker | undefined {
    return (this.options.frameMetadata ?? this.options.packetTrailer)?.worker;
  }

  private setupFrameMetadataSender(sender: RTCRtpSender, opts: TrackPublishOptions = {}) {
    const worker = this.frameMetadataWorker;
    if (!worker || this.signalOpts?.e2eeEnabled) {
      return;
    }

    const frameMetadata = opts.frameMetadata ?? opts.packetTrailer;
    const hasMetadata = hasFrameMetadataPublishOptions(frameMetadata);

    if (shouldUseFrameMetadataScriptTransform()) {
      if (hasMetadata) {
        // @ts-ignore
        sender.transform = new RTCRtpScriptTransform(worker, {
          kind: 'encode',
          packetTrailer: frameMetadata,
        });
      }
      return;
    }

    if (
      !isFrameMetadataSupported(this.options.frameMetadata ?? this.options.packetTrailer) ||
      !('createEncodedStreams' in sender)
    ) {
      if (hasMetadata) {
        this.log.warn('frame metadata transform not supported; skipping write', this.logContext);
      }
      return;
    }

    // @ts-ignore
    const { readable, writable } = sender.createEncodedStreams();
    if (hasMetadata) {
      worker.postMessage(
        {
          kind: 'encode',
          data: {
            readableStream: readable,
            writableStream: writable,
            packetTrailer: frameMetadata,
          },
        },
        [readable, writable],
      );
    } else {
      readable.pipeTo(writable);
    }
  }

  private async createTransceiverRTCRtpSender(
    track: LocalTrack,
    opts: TrackPublishOptions,
    encodings?: RTCRtpEncodingParameters[],
  ) {
    if (!this.pcManager) {
      throw new UnexpectedConnectionState('publisher is closed');
    }

    const streams: MediaStream[] = [];

    if (track.mediaStream) {
      streams.push(track.mediaStream);
    }

    if (isVideoTrack(track)) {
      track.codec = opts.videoCodec;
    }

    const transceiverInit: RTCRtpTransceiverInit = { direction: 'sendonly', streams };
    if (encodings) {
      transceiverInit.sendEncodings = encodings;
    }
    // addTransceiver for react-native is async. web is synchronous, but await won't effect it.
    const transceiver = await this.pcManager.addPublisherTransceiver(
      track.mediaStreamTrack,
      transceiverInit,
    );

    return transceiver.sender;
  }

  private async createSimulcastTransceiverSender(
    track: LocalVideoTrack,
    simulcastTrack: SimulcastTrackInfo,
    opts: TrackPublishOptions,
    encodings?: RTCRtpEncodingParameters[],
  ) {
    if (!this.pcManager) {
      throw new UnexpectedConnectionState('publisher is closed');
    }
    const transceiverInit: RTCRtpTransceiverInit = { direction: 'sendonly' };
    if (encodings) {
      transceiverInit.sendEncodings = encodings;
    }
    // addTransceiver for react-native is async. web is synchronous, but await won't effect it.
    const transceiver = await this.pcManager.addPublisherTransceiver(
      simulcastTrack.mediaStreamTrack,
      transceiverInit,
    );
    if (!opts.videoCodec) {
      return;
    }
    await track.setSimulcastTrackSender(opts.videoCodec, transceiver.sender);
    return transceiver.sender;
  }

  private async createRTCRtpSender(track: MediaStreamTrack) {
    if (!this.pcManager) {
      throw new UnexpectedConnectionState('publisher is closed');
    }
    return this.pcManager.addPublisherTrack(track);
  }

  // websocket reconnect behavior. if websocket is interrupted, and the PeerConnection
  // continues to work, we can reconnect to websocket to continue the session
  // after a number of retries, we'll close and give up permanently
  private handleDisconnect = (connection: string, disconnectReason?: ReconnectReason) => {
    if (this._isClosed) {
      return;
    }

    this.log.warn(`${connection} disconnected`);
    if (this.reconnectAttempts === 0) {
      // only reset start time on the first try
      this.reconnectStart = Date.now();
    }

    const disconnect = (duration: number) => {
      this.log.warn(
        `could not recover connection after ${this.reconnectAttempts} attempts, ${duration}ms. giving up`,
      );
      this.emit(EngineEvent.Disconnected);
      this.close();
    };

    const duration = Date.now() - this.reconnectStart;
    let delay = this.getNextRetryDelay({
      elapsedMs: duration,
      retryCount: this.reconnectAttempts,
    });

    if (delay === null) {
      disconnect(duration);
      return;
    }
    if (connection === leaveReconnect) {
      delay = 0;
    }

    this.log.debug(`reconnecting in ${delay}ms`);

    this.clearReconnectTimeout();
    if (this.token) {
      // token may have been refreshed, we do not want to recreate the regionUrlProvider
      // since the current engine may have inherited a regional url
      this.emit(EngineEvent.TokenRefreshed, this.token);
    }
    this.reconnectTimeout = CriticalTimers.setTimeout(
      () =>
        this.attemptReconnect(disconnectReason).finally(() => (this.reconnectTimeout = undefined)),
      delay,
    );
  };

  /**
   * A sustained local `LOST` while connected and publishing means the server isn't receiving
   * our media, so force a full reconnect; any non-`LOST` value cancels a pending trigger.
   */
  private handleLocalConnectionQuality(update: ConnectionQualityUpdate) {
    if (!this.participantSid) {
      return;
    }
    const localUpdate = update.updates.find((u) => u.participantSid === this.participantSid);
    if (!localUpdate) {
      return;
    }
    if (localUpdate.quality === ProtoConnectionQuality.LOST) {
      this.scheduleLostQualityReconnect();
    } else {
      this.clearLostQualityTimeout();
    }
  }

  private scheduleLostQualityReconnect() {
    if (this.lostQualityTimeout) {
      // already counting down towards a reconnect
      return;
    }
    this.lostQualityTimeout = CriticalTimers.setTimeout(() => {
      this.lostQualityTimeout = undefined;
      if (this._isClosed || this.pcState !== PCState.Connected || this.attemptingReconnect) {
        return;
      }
      if (!this.hasActivePublisherSenders()) {
        return;
      }
      this.log.warn(
        'local connection quality lost while publishing, triggering full reconnect',
        this.logContext,
      );
      this.fullReconnectOnNext = true;
      this.handleDisconnect('connection quality lost', ReconnectReason.RR_PUBLISHER_FAILED);
    }, connectionQualityLostTimeout);
  }

  private clearLostQualityTimeout() {
    if (this.lostQualityTimeout) {
      CriticalTimers.clearTimeout(this.lostQualityTimeout);
      this.lostQualityTimeout = undefined;
    }
  }

  /** Whether the publisher currently has any sender with a live track. */
  private hasActivePublisherSenders(): boolean {
    return (
      this.pcManager?.publisher
        .getSenders()
        .some((sender) => !!sender.track && sender.track.readyState === 'live') ?? false
    );
  }

  /**
   * Forces a full reconnect while keeping the engine (and its saved credentials) alive. Used by
   * Room's connection-reconcile safety net when the transport silently died but we looked connected.
   * @internal
   */
  reconnect(reason: ReconnectReason = ReconnectReason.RR_UNKNOWN) {
    this.fullReconnectOnNext = true;
    this.handleDisconnect('reconcile', reason);
  }

  private async attemptReconnect(reason?: ReconnectReason) {
    if (this._isClosed) {
      return;
    }
    // guard for attempting reconnection multiple times while one attempt is still not finished
    if (this.attemptingReconnect) {
      this.log.warn('already attempting reconnect, returning early');
      return;
    }

    // A pending Lost-quality countdown belongs to the session we're now leaving; cancel it so
    // it can't fire against the reconnected session before the server has evaluated it. (A resume
    // keeps the peer connections, so cleanupPeerConnections wouldn't cover this path.)
    this.clearLostQualityTimeout();

    if (
      this.clientConfiguration?.resumeConnection === ClientConfigSetting.DISABLED ||
      // signaling state could change to closed due to hardware sleep
      // those connections cannot be resumed
      (this.pcManager?.currentState ?? PCTransportState.NEW) === PCTransportState.NEW
    ) {
      this.fullReconnectOnNext = true;
    }

    // Consume the flag up front: capture whether this attempt is a full reconnect, then reset
    // it. From here on a `true` value unambiguously represents a *new* full-reconnect request
    // that arrived while this attempt was running (e.g. a server RECONNECT leave), which the
    // finally block dispatches — for both the resume and full-reconnect paths.
    const fullReconnect = this.fullReconnectOnNext;
    this.fullReconnectOnNext = false;

    let succeeded = false;
    try {
      this.attemptingReconnect = true;
      if (fullReconnect) {
        await this.restartConnection();
      } else {
        await this.resumeConnection(reason);
      }
      this.clearPendingReconnect();
      succeeded = true;
    } catch (e) {
      this.reconnectAttempts += 1;
      let recoverable = true;
      if (e instanceof UnexpectedConnectionState) {
        this.log.debug('received unrecoverable error', { error: e });
        // unrecoverable
        recoverable = false;
      } else if (fullReconnect || !(e instanceof SignalReconnectError)) {
        // a failed full reconnect stays a full reconnect; a failed resume can only be
        // resumed again for a signal-level error, otherwise it escalates
        this.fullReconnectOnNext = true;
      }

      if (recoverable) {
        this.handleDisconnect('reconnect', ReconnectReason.RR_UNKNOWN);
      } else {
        this.log.info(
          `could not recover connection after ${this.reconnectAttempts} attempts, ${
            Date.now() - this.reconnectStart
          }ms. giving up`,
        );
        this.emit(EngineEvent.Disconnected);
        await this.close();
      }
    } finally {
      this.attemptingReconnect = false;

      // A full reconnect requested while this attempt was running (e.g. a `RECONNECT` leave
      // during a resume or a restart) that a successful attempt didn't act on; dispatch it now
      // (the failure path already retries).
      if (succeeded && this.fullReconnectOnNext && !this._isClosed) {
        this.log.debug('full reconnect requested during in-progress attempt, dispatching');
        this.handleDisconnect('reconnect');
      }
    }
  }

  private getNextRetryDelay(context: ReconnectContext) {
    try {
      return this.reconnectPolicy.nextRetryDelayInMs(context);
    } catch (e) {
      this.log.warn('encountered error in reconnect policy', { error: e });
    }

    // error in user code with provided reconnect policy, stop reconnecting
    return null;
  }

  private async restartConnection(regionUrl?: string) {
    try {
      if (!this.url || !this.token) {
        // permanent failure, don't attempt reconnection
        throw new UnexpectedConnectionState('could not reconnect, url or token not saved');
      }

      this.log.info(`reconnecting, attempt: ${this.reconnectAttempts}`);
      this.emit(EngineEvent.Restarting);

      if (!this.client.isDisconnected) {
        await this.client.sendLeave();
      }
      await this.cleanupPeerConnections();
      await this.cleanupClient();

      let joinResponse: JoinResponse;
      try {
        if (!this.signalOpts) {
          this.log.warn('attempted connection restart, without signal options present');
          throw new SignalReconnectError();
        }
        // in case a regionUrl is passed, the region URL takes precedence
        joinResponse = (
          await this.join(
            regionUrl ?? this.url,
            this.token,
            this.signalOpts,
            undefined,
            !this.options.singlePeerConnection,
          )
        ).joinResponse;
      } catch (e) {
        if (e instanceof ConnectionError && e.reason === ConnectionErrorReason.NotAllowed) {
          throw new UnexpectedConnectionState('could not reconnect, token might be expired');
        }
        throw new SignalReconnectError();
      }

      if (this.shouldFailNext) {
        this.shouldFailNext = false;
        throw new Error('simulated failure');
      }

      this.client.setReconnected();
      this.emit(EngineEvent.SignalRestarted, joinResponse);

      await this.waitForPCReconnected();

      // re-check signal connection state before setting engine as resumed
      if (this.client.currentState !== SignalConnectionState.CONNECTED) {
        throw new SignalReconnectError('Signal connection got severed during reconnect');
      }

      this.regionStrategy?.resetAttempts();
      // reconnect success
      this.emit(EngineEvent.Restarted);
    } catch (error) {
      const nextRegionUrl = await this.regionStrategy?.getNextUrl();
      if (nextRegionUrl) {
        await this.restartConnection(nextRegionUrl);
        return;
      } else {
        // no more regions to try (or we're not on cloud)
        this.regionStrategy?.resetAttempts();
        throw error;
      }
    }
  }

  private async resumeConnection(reason?: ReconnectReason): Promise<void> {
    if (!this.url || !this.token) {
      // permanent failure, don't attempt reconnection
      throw new UnexpectedConnectionState('could not reconnect, url or token not saved');
    }
    // trigger publisher reconnect
    if (!this.pcManager) {
      throw new UnexpectedConnectionState('publisher and subscriber connections unset');
    }

    this.log.info(`resuming signal connection, attempt ${this.reconnectAttempts}`);
    this.emit(EngineEvent.Resuming);
    let res: ReconnectResponse | undefined;
    try {
      this.setupSignalClientCallbacks();
      res = await this.client.reconnect(this.url, this.token, this.participantSid, reason);
    } catch (error) {
      let message = '';
      if (error instanceof Error) {
        message = error.message;
        this.log.error(error.message, { error });
      }
      if (error instanceof ConnectionError && error.reason === ConnectionErrorReason.NotAllowed) {
        throw new UnexpectedConnectionState('could not reconnect, token might be expired');
      }
      if (error instanceof ConnectionError && error.reason === ConnectionErrorReason.LeaveRequest) {
        throw error;
      }
      throw new SignalReconnectError(message);
    }
    this.emit(EngineEvent.SignalResumed);

    if (res) {
      const rtcConfig = this.makeRTCConfiguration(res);
      this.pcManager.updateConfiguration(rtcConfig);
      if (this.latestJoinResponse) {
        this.latestJoinResponse.serverInfo = res.serverInfo;
      }
    } else {
      this.log.warn('Did not receive reconnect response');
    }

    if (this.shouldFailNext) {
      this.shouldFailNext = false;
      throw new Error('simulated failure');
    }

    await this.pcManager.triggerIceRestart();

    await this.waitForPCReconnected();

    // re-check signal connection state before setting engine as resumed
    if (this.client.currentState !== SignalConnectionState.CONNECTED) {
      throw new SignalReconnectError('Signal connection got severed during reconnect');
    }

    this.client.setReconnected();

    // recreate publish datachannel if it's id is null
    // (for safari https://bugs.webkit.org/show_bug.cgi?id=184688)
    const reliableDC = this.dataChannelForKind(DataChannelKind.RELIABLE);
    if (reliableDC?.readyState === 'open' && reliableDC.id === null) {
      this.createDataChannels();
    }

    if (res?.lastMessageSeq) {
      this.resendReliableMessagesForResume(res.lastMessageSeq).catch((error) => {
        this.log.warn('failed to resend reliable messages after resume', {
          ...this.logContext,
          error,
        });
      });
    }

    // resume success
    this.emit(EngineEvent.Resumed);
  }

  async waitForPCInitialConnection(timeout?: number, abortController?: AbortController) {
    if (!this.pcManager) {
      throw new UnexpectedConnectionState('PC manager is closed');
    }
    await this.pcManager.ensurePCTransportConnection(abortController, timeout);
  }

  private async waitForPCReconnected() {
    this.pcState = PCState.Reconnecting;

    this.log.debug('waiting for peer connection to reconnect');
    try {
      await sleep(minReconnectWait); // FIXME setTimeout again not ideal for a connection critical path
      if (!this.pcManager) {
        throw new UnexpectedConnectionState('PC manager is closed');
      }
      await this.pcManager.ensurePCTransportConnection(undefined, this.peerConnectionTimeout);
      this.pcState = PCState.Connected;
    } catch (e: any) {
      // TODO do we need a `failed` state here for the PC?
      this.pcState = PCState.Disconnected;
      throw ConnectionError.internal(`could not establish PC connection, ${e.message}`);
    }
  }

  waitForRestarted = () => {
    return new Promise<void>((resolve, reject) => {
      if (this.pcState === PCState.Connected) {
        resolve();
      }
      const onRestarted = () => {
        this.off(EngineEvent.Disconnected, onDisconnected);
        resolve();
      };
      const onDisconnected = () => {
        this.off(EngineEvent.Restarted, onRestarted);
        reject();
      };
      this.once(EngineEvent.Restarted, onRestarted);
      this.once(EngineEvent.Disconnected, onDisconnected);
    });
  };

  /** @internal */
  async publishRpcAck(destinationIdentity: string, requestId: string) {
    const packet = new DataPacket({
      destinationIdentities: [destinationIdentity],
      kind: DataPacket_Kind.RELIABLE,
      value: {
        case: 'rpcAck',
        value: new RpcAck({
          requestId,
        }),
      },
    });
    await this.sendDataPacket(packet, DataChannelKind.RELIABLE);
  }

  /* @internal */
  async sendDataPacket(
    packet: DataPacket,
    /**  Data-track frames don't come through here — they're sent pre-serialized via {@link sendDataTrackFrame }. */
    kind: Exclude<DataChannelKind, DataChannelKind.DATA_TRACK_LOSSY>,
  ) {
    // make sure we do have a data connection
    await this.ensurePublisherConnected(kind);

    if (this.e2eeManager && this.e2eeManager.isDataChannelEncryptionEnabled) {
      const encryptablePacket = asEncryptablePacket(packet);
      if (encryptablePacket) {
        const encryptedData = await this.e2eeManager.encryptData(
          encryptablePacket.toBinary() as NonSharedUint8Array,
        );
        packet.value = {
          case: 'encryptedPacket',
          value: new EncryptedPacket({
            encryptedValue: encryptedData.payload,
            iv: encryptedData.iv,
            keyIndex: encryptedData.keyIndex,
          }),
        };
      }
    }

    if (kind === DataChannelKind.RELIABLE) {
      packet.sequence = this.reliableChannel.nextSequence();
    }

    const msg = packet.toBinary() as Uint8Array<ArrayBuffer>;

    // Clamp to the SDK default - libwebrtc advertises larger (~256 KiB)
    // than LiveKit/pion can deliver end-to-end (~64 KiB), so we trust
    // the answer up untilthe built in ceiling.
    const maxPublisherMessageSizeBytes = Math.min(
      this.pcManager?.getMaxPublisherMessageSize() ?? DEFAULT_MAX_MESSAGE_SIZE,
      DEFAULT_MAX_MESSAGE_SIZE,
    );
    if (
      typeof maxPublisherMessageSizeBytes !== 'undefined' &&
      maxPublisherMessageSizeBytes !== 0 /* 0 means "no limit" */ &&
      msg.byteLength > maxPublisherMessageSizeBytes
    ) {
      throw new PublishDataError(
        `cannot publish data packet larger than ${maxPublisherMessageSizeBytes} bytes (got ${msg.byteLength})`,
      );
    }

    // The full-buffer policy (drop for lossy, wait/replay for reliable) is the channel's own, as
    // is emitting the buffer-status change once the send settles.
    if (kind === DataChannelKind.RELIABLE) {
      await this.reliableChannel.send(msg, packet.sequence);
    } else {
      await this.lossyChannel.send(msg);
    }
  }

  /**
   * Sends pre-serialized bytes on the data-track channel. This is the one send path that doesn't
   * go through {@link sendDataPacket} — Room's `packetAvailable` handler calls it directly with
   * bytes the data-track pipeline already serialized.
   *
   * @internal
   */
  async sendDataTrackFrame(bytes: NonSharedUint8Array) {
    // Make sure we do have a data connection: on lazily negotiated publisher connections this is
    // what kicks negotiation off, and it waits for the channel to open. Memoized, so the
    // steady-state cost is one await on an already-resolved promise.
    await this.ensurePublisherConnected(DataChannelKind.DATA_TRACK_LOSSY);
    await this.dataTrackChannel.send(bytes);
  }

  private async resendReliableMessagesForResume(lastMessageSeq: number) {
    await this.ensurePublisherConnected(DataChannelKind.RELIABLE);
    await this.reliableChannel.replay(lastMessageSeq);
  }

  /** The flow-control gate for `kind` — see {@link FlowControlledDataChannel}. */
  private flowControlFor(kind: DataChannelKind): FlowControlledDataChannel {
    return this.dataChannels.channelFor(kind);
  }

  /**
   * Resolves once the caller may send on the `kind` channel — see
   * {@link FlowControlledDataChannel.waitForHeadroomWithLock}.
   */
  async waitForBufferHeadroom(kind: DataChannelKind) {
    return this.flowControlFor(kind).waitForHeadroomWithLock();
  }

  /**
   * @internal
   */
  async ensureDataTransportConnected(
    kind: DataChannelKind,
    subscriber: boolean = this.subscriberPrimary,
  ) {
    if (!this.pcManager) {
      throw new UnexpectedConnectionState('PC manager is closed');
    }
    const transport = subscriber ? this.pcManager.subscriber : this.pcManager.publisher;
    const transportName = subscriber ? 'Subscriber' : 'Publisher';
    if (!transport) {
      throw ConnectionError.internal(`${transportName} connection not set`);
    }

    let needNegotiation = false;
    if (!subscriber && !this.dataChannelForKind(kind, subscriber)) {
      this.createDataChannels();
      needNegotiation = true;
    }

    if (
      !needNegotiation &&
      !subscriber &&
      !this.pcManager.publisher.isICEConnected &&
      this.pcManager.publisher.getICEConnectionState() !== 'checking'
    ) {
      needNegotiation = true;
    }
    if (needNegotiation) {
      // start negotiation
      this.negotiate().catch((err) => {
        this.log.error(err);
      });
    }

    const targetChannel = this.dataChannelForKind(kind, subscriber);
    if (targetChannel?.readyState === 'open') {
      return;
    }

    // wait until ICE connected
    const endTime = new Date().getTime() + this.peerConnectionTimeout;
    while (new Date().getTime() < endTime) {
      if (
        transport.isICEConnected &&
        this.dataChannelForKind(kind, subscriber)?.readyState === 'open'
      ) {
        return;
      }
      await sleep(50);
    }

    throw ConnectionError.internal(
      `could not establish ${transportName} connection, state: ${transport.getICEConnectionState()}`,
    );
  }

  private async ensurePublisherConnected(kind: DataChannelKind) {
    if (!this.publisherConnectionPromise) {
      this.publisherConnectionPromise = this.ensureDataTransportConnected(kind, false);
    }
    await this.publisherConnectionPromise;
  }

  /* @internal */
  verifyTransport(): boolean {
    if (!this.pcManager) {
      return false;
    }
    const state = this.pcManager.currentState;
    const allowedConnectionStates: PCTransportState[] = [
      PCTransportState.CONNECTING,
      PCTransportState.CONNECTED,
    ];
    if (!allowedConnectionStates.includes(state)) {
      return false;
    }

    // ensure signal is connected
    if (!this.client.ws || this.client.ws.readyState === WebSocket.CLOSED) {
      return false;
    }

    // A transport stuck in CONNECTING never reaches CONNECTED nor reports FAILED, so it would
    // otherwise look healthy forever; bound how long we tolerate it. The entry time is recorded
    // in the pcManager state-change handler (see configure()), so this is a pure read — an
    // unrecorded CONNECTING fails open rather than measuring against a stale timestamp.
    if (
      state === PCTransportState.CONNECTING &&
      this.transportConnectingSince !== undefined &&
      Date.now() - this.transportConnectingSince > this.peerConnectionTimeout
    ) {
      this.log.warn('transport stuck in connecting state', this.logContext);
      return false;
    }

    return true;
  }

  /** @internal */
  async negotiate(): Promise<void> {
    // observe signal state
    return new TypedPromise<void, NegotiationError | Error>(async (resolve, reject) => {
      if (!this.pcManager) {
        reject(new NegotiationError('PC manager is closed'));
        return;
      }

      this.pcManager.requirePublisher();
      // don't negotiate without any transceivers or data channel, it will generate sdp without ice frag then negotiate failed
      if (
        this.pcManager.publisher.getTransceivers().length == 0 &&
        !this.dataChannels.hasPublisherChannels
      ) {
        this.createDataChannels();
      }

      const abortController = new AbortController();

      const handleClosed = () => {
        abortController.abort();
        this.log.debug('engine disconnected while negotiation was ongoing');
        resolve();
        return;
      };

      if (this.isClosed) {
        reject(new NegotiationError('cannot negotiate on closed engine'));
      }
      this.on(EngineEvent.Closing, handleClosed);
      this.on(EngineEvent.Restarting, handleClosed);
      this.pcManager.publisher.off(PCEvents.RTPVideoPayloadTypes, this.onRtpMapAvailable);
      this.pcManager.publisher.once(PCEvents.RTPVideoPayloadTypes, this.onRtpMapAvailable);

      try {
        await this.pcManager.negotiate(abortController);
        resolve();
      } catch (e: unknown) {
        if (abortController.signal.aborted) {
          // negotiation was aborted due to engine close or restart, resolve
          // cleanly to avoid triggering a cascading reconnect loop
          resolve();
          return;
        }
        if (e instanceof NegotiationError) {
          this.fullReconnectOnNext = true;
        }
        this.handleDisconnect('negotiation', ReconnectReason.RR_UNKNOWN);
        if (e instanceof Error) {
          reject(e);
        } else {
          reject(new Error(String(e)));
        }
      } finally {
        this.off(EngineEvent.Closing, handleClosed);
        this.off(EngineEvent.Restarting, handleClosed);
      }
    });
  }

  dataChannelForKind(kind: DataChannelKind, sub?: boolean): RTCDataChannel | undefined {
    return this.dataChannels.getHandle(kind, sub);
  }

  /** @internal */
  sendSyncState(
    remoteTracks: RemoteTrackPublication[],
    localTracks: LocalTrackPublication[],
    localDataTrackInfos: Array<DataTrackInfo>,
  ) {
    if (!this.pcManager) {
      this.log.warn('sync state cannot be sent without peer connection setup');
      return;
    }
    const previousPublisherOffer = this.pcManager.publisher.getLocalDescription();
    const previousPublisherAnswer = this.pcManager.publisher.getRemoteDescription();
    const previousSubscriberOffer = this.pcManager.subscriber?.getRemoteDescription();
    const previousSubscriberAnswer = this.pcManager.subscriber?.getLocalDescription();

    /* 1. autosubscribe on, so subscribed tracks = all tracks - unsub tracks,
          in this case, we send unsub tracks, so server add all tracks to this
          subscribe pc and unsub special tracks from it.
       2. autosubscribe off, we send subscribed tracks.
    */
    const autoSubscribe = this.signalOpts?.autoSubscribe ?? true;
    const trackSids = new Array<string>();
    const trackSidsDisabled = new Array<string>();

    remoteTracks.forEach((track) => {
      if (track.isDesired !== autoSubscribe) {
        trackSids.push(track.trackSid);
      }
      if (!track.isEnabled) {
        trackSidsDisabled.push(track.trackSid);
      }
    });

    this.client.sendSyncState(
      new SyncState({
        answer:
          this.pcManager.mode === 'publisher-only'
            ? previousPublisherAnswer
              ? toProtoSessionDescription({
                  sdp: previousPublisherAnswer.sdp,
                  type: previousPublisherAnswer.type,
                })
              : undefined
            : previousSubscriberAnswer
              ? toProtoSessionDescription({
                  sdp: previousSubscriberAnswer.sdp,
                  type: previousSubscriberAnswer.type,
                })
              : undefined,
        offer:
          this.pcManager.mode === 'publisher-only'
            ? previousPublisherOffer
              ? toProtoSessionDescription({
                  sdp: previousPublisherOffer.sdp,
                  type: previousPublisherOffer.type,
                })
              : undefined
            : previousSubscriberOffer
              ? toProtoSessionDescription({
                  sdp: previousSubscriberOffer.sdp,
                  type: previousSubscriberOffer.type,
                })
              : undefined,
        subscription: new UpdateSubscription({
          trackSids,
          subscribe: !autoSubscribe,
          participantTracks: [],
        }),
        publishTracks: getTrackPublicationInfo(localTracks),
        dataChannels: this.dataChannelsInfo(),
        trackSidsDisabled,
        datachannelReceiveStates: this.reliableReceivedState.map((seq, sid) => {
          return new DataChannelReceiveState({
            publisherSid: sid,
            lastSeq: seq,
          });
        }),
        publishDataTracks: localDataTrackInfos.map((info) => {
          return new PublishDataTrackResponse({ info: DataTrackInfo.toProtobuf(info) });
        }),
      }),
    );
  }

  /* @internal */
  failNext() {
    // debugging method to fail the next reconnect/resume attempt
    this.shouldFailNext = true;
  }

  /* @internal */
  failNextV1Path() {
    // debugging method to fail the next connection attempt for /rtc/v1 to trigger the fallback version
    this.shouldFailOnV1Path = true;
  }

  private onRtpMapAvailable = (rtpTypes: MediaAttributes['rtp']) => {
    const rtpMap = new Map<number, VideoCodec>();
    rtpTypes.forEach((rtp) => {
      const codec = rtp.codec.toLowerCase();
      if (isVideoCodec(codec)) {
        rtpMap.set(rtp.payload, codec);
      }
    });
    this.emit(EngineEvent.RTPVideoMapUpdate, rtpMap);
  };

  private dataChannelsInfo(): DataChannelInfo[] {
    const infos: DataChannelInfo[] = [];
    const getInfo = (dc: RTCDataChannel | undefined, target: SignalTarget) => {
      if (dc?.id !== undefined && dc.id !== null) {
        infos.push(
          new DataChannelInfo({
            label: dc.label,
            id: dc.id,
            target,
          }),
        );
      }
    };
    getInfo(this.dataChannelForKind(DataChannelKind.LOSSY), SignalTarget.PUBLISHER);
    getInfo(this.dataChannelForKind(DataChannelKind.RELIABLE), SignalTarget.PUBLISHER);
    getInfo(this.dataChannelForKind(DataChannelKind.LOSSY, true), SignalTarget.SUBSCRIBER);
    getInfo(this.dataChannelForKind(DataChannelKind.RELIABLE, true), SignalTarget.SUBSCRIBER);
    return infos;
  }

  private clearReconnectTimeout() {
    if (this.reconnectTimeout) {
      CriticalTimers.clearTimeout(this.reconnectTimeout);
    }
  }

  private clearPendingReconnect() {
    this.clearReconnectTimeout();
    this.reconnectAttempts = 0;
  }

  private handleBrowserOnLine = async () => {
    if (!this.url) {
      return;
    }
    const hasNetworkConnection = await fetch(toHttpUrl(this.url!), { method: 'HEAD' })
      .then((resp) => resp.ok)
      .catch(() => false);

    if (!hasNetworkConnection) {
      return;
    }
    this.log.info('detected network reconnected');

    if (
      // in case the engine is currently reconnecting, attempt a reconnect immediately after the browser state has changed to 'onLine'
      this.client.currentState === SignalConnectionState.RECONNECTING ||
      // also if the browser went offline before and the engine still thinks it's in a connected state, treat it as a network interruption that we haven't noticed yet
      (this.isWaitingForNetworkReconnect &&
        this.client.currentState === SignalConnectionState.CONNECTED)
    ) {
      this.clearReconnectTimeout();
      this.attemptReconnect(ReconnectReason.RR_SIGNAL_DISCONNECTED);
      this.isWaitingForNetworkReconnect = false;
    }
  };

  private handleBrowserOffline = async () => {
    if (!this.url) {
      return;
    }
    try {
      await Promise.race([
        fetch(toHttpUrl(this.url), { method: 'HEAD' }),
        // if there's no internet connection the fetch rejects immediately, so we only use a short timeout here
        sleep(4_000).then(() => Promise.reject()),
      ]);
    } catch (e) {
      // only set if the browser still thinks it's offline after the request failed
      if (window.navigator.onLine === false) {
        this.log.info('detected network interruption');
        this.isWaitingForNetworkReconnect = true;
      }
    }
  };

  private registerOnLineListener() {
    if (isWeb()) {
      window.addEventListener('online', this.handleBrowserOnLine);
      window.addEventListener('offline', this.handleBrowserOffline);
    }
  }

  private deregisterOnLineListener() {
    if (isWeb()) {
      window.removeEventListener('online', this.handleBrowserOnLine);
      window.removeEventListener('offline', this.handleBrowserOffline);
    }
  }

  getTrackIdForReceiver(receiver: RTCRtpReceiver): string | undefined {
    const mid = this.pcManager?.getMidForReceiver(receiver);
    if (mid) {
      const match = Object.entries(this.midToTrackId).find(([key]) => key === mid);
      if (match) {
        return match[1];
      }
    }
  }
}

export type EngineEventCallbacks = {
  connected: (joinResp: JoinResponse) => void;
  disconnected: (reason?: DisconnectReason) => void;
  resuming: () => void;
  resumed: () => void;
  restarting: () => void;
  restarted: () => void;
  signalResumed: () => void;
  signalRestarted: (joinResp: JoinResponse) => void;
  closing: () => void;
  mediaTrackAdded: (
    track: MediaStreamTrack,
    streams: MediaStream,
    receiver: RTCRtpReceiver,
  ) => void;
  activeSpeakersUpdate: (speakers: Array<SpeakerInfo>) => void;
  dataPacketReceived: (packet: DataPacket, encryptionType: Encryption_Type) => void;
  transcriptionReceived: (transcription: Transcription) => void;
  transportsCreated: (publisher: PCTransport, subscriber?: PCTransport) => void;
  /** @internal */
  trackSenderAdded: (track: Track, sender: RTCRtpSender) => void;
  rtpVideoMapUpdate: (rtpMap: Map<number, VideoCodec>) => void;
  dcBufferStatusChanged: (isLow: boolean, kind: DataChannelKind) => void;
  participantUpdate: (infos: ParticipantInfo[]) => void;
  roomUpdate: (room: RoomModel) => void;
  roomMoved: (room: RoomMovedResponse) => void;
  connectionQualityUpdate: (update: ConnectionQualityUpdate) => void;
  speakersChanged: (speakerUpdates: SpeakerInfo[]) => void;
  streamStateChanged: (update: StreamStateUpdate) => void;
  subscriptionError: (resp: SubscriptionResponse) => void;
  subscriptionPermissionUpdate: (update: SubscriptionPermissionUpdate) => void;
  subscribedQualityUpdate: (update: SubscribedQualityUpdate) => void;
  localTrackUnpublished: (unpublishedResponse: TrackUnpublishedResponse) => void;
  localTrackSubscribed: (trackSid: string) => void;
  remoteMute: (trackSid: string, muted: boolean) => void;
  offline: () => void;
  signalRequestResponse: (response: RequestResponse) => void;
  signalConnected: (joinResp: JoinResponse) => void;
  publishDataTrackResponse: (event: PublishDataTrackResponse) => void;
  unPublishDataTrackResponse: (event: UnpublishDataTrackResponse) => void;
  dataTrackSubscriberHandles: (event: DataTrackSubscriberHandles) => void;
  dataTrackPacketReceived: (packet: Uint8Array) => void;
  joined: (joinResponse: JoinResponse) => void;
  tokenRefreshed: (token: string) => void;
  serverRegionsReported: (regions: RegionSettings) => void;
};

export interface RegionStrategy {
  getNextUrl(abortSignal?: AbortSignal): Promise<string | null>;
  resetAttempts(): void;
}

function applyUserDataCompat(newObj: DataPacket, oldObj: UserPacket) {
  const participantIdentity = newObj.participantIdentity
    ? newObj.participantIdentity
    : oldObj.participantIdentity;
  newObj.participantIdentity = participantIdentity;
  oldObj.participantIdentity = participantIdentity;

  const destinationIdentities =
    newObj.destinationIdentities.length !== 0
      ? newObj.destinationIdentities
      : oldObj.destinationIdentities;
  newObj.destinationIdentities = destinationIdentities;
  oldObj.destinationIdentities = destinationIdentities;
}