chad/server/plugins/socket/webrtc/Gateway.ts

import type { types } from 'mediasoup'
import type { ChadSocket, ChadSocketServer } from '../types.ts'
import type { Channel } from './Channel.ts'
import type { ChannelManager } from './ChannelManager.ts'
import type { Client } from './Client.ts'
import { consola } from 'consola'

export class WebRtcGateway {
  readonly #io: ChadSocketServer
  readonly #socket: ChadSocket
  readonly #client: Client
  readonly #channels: ChannelManager

  constructor(
    io: ChadSocketServer,
    socket: ChadSocket,
    client: Client,
    channels: ChannelManager,
  ) {
    this.#io = io
    this.#socket = socket
    this.#client = client
    this.#channels = channels

    this.register()
  }

  register(): void {
    this.#client.on('signal:new-consumer', async (data, onAcked) => {
      await this.#socket.emitWithAck('new-consumer', data)
      await onAcked()
    })
    this.#client.on('consumer:closed', consumerId => this.#socket.emit('consumer-closed', { consumerId }))
    this.#client.on('consumer:paused', consumerId => this.#socket.emit('consumer-paused', { consumerId }))
    this.#client.on('consumer:resumed', consumerId => this.#socket.emit('consumer-resumed', { consumerId }))
    this.#client.on('transport:closed', () => this.#socket.disconnect())
    this.#client.on('updated', () => this.#io.emit('client-updated', this.#client.serialize()))

    this.#socket.on('join-channel', this.#onJoinChannel.bind(this))
    this.#socket.on('create-transport', this.#onCreateTransport.bind(this))
    this.#socket.on('connect-transport', this.#onConnectTransport.bind(this))
    this.#socket.on('produce', this.#onProduce.bind(this))
    this.#socket.on('close-producer', this.#onCloseProducer.bind(this))
    this.#socket.on('pause-producer', this.#onPauseProducer.bind(this))
    this.#socket.on('resume-producer', this.#onResumeProducer.bind(this))
    this.#socket.on('pause-consumer', this.#onPauseConsumer.bind(this))
    this.#socket.on('resume-consumer', this.#onResumeConsumer.bind(this))
    this.#socket.on('update-client', this.#onUpdateClient.bind(this))
    this.#socket.on('disconnect', this.#onDisconnect.bind(this))
  }

  async #onJoinChannel({ channelId }: { channelId: string }): Promise<void> {
    if (this.#client.channelId === channelId)
      return

    const newChannel = this.#channels.get(channelId)

    if (!newChannel) {
      consola.error('[Gateway]', `Channel not found: ${channelId}`)
      return
    }

    const oldChannel = this.#channels.get(this.#client.channelId)

    if (oldChannel)
      this.#leaveChannel(oldChannel)

    this.#client.clearConsumers()

    this.#socket.join(newChannel.id)
    newChannel.addClient(this.#client)
    await newChannel.wireClient(this.#client)

    this.#io.emit('client-switched-channel', this.#client.serialize())
  }

  async #onCreateTransport(
    { producing, consuming }: { producing: boolean, consuming: boolean },
    cb: (result: any) => void,
  ): Promise<void> {
    try {
      const transportData = await this.#client.createTransport({ producing, consuming })
      cb(transportData)

      if (consuming) {
        const channel = this.#channels.get(this.#client.channelId)

        if (channel)
          await channel.wireClient(this.#client)
      }
    }
    catch (error) {
      if (error instanceof Error) {
        consola.error('[Gateway]', '[createTransport]', error.message)
        cb({ error: error.message })
      }
    }
  }

  async #onConnectTransport(
    { transportId, dtlsParameters }: { transportId: string, dtlsParameters: types.DtlsParameters },
    cb: (result: any) => void,
  ): Promise<void> {
    try {
      await this.#client.connectTransport(transportId, dtlsParameters)
      cb({ ok: true })
    }
    catch (error) {
      if (error instanceof Error) {
        consola.error('[Gateway]', '[connectTransport]', error.message)
        cb({ error: error.message })
      }
    }
  }

  async #onProduce(
    { transportId, kind, rtpParameters, appData }: {
      transportId: string
      kind: types.MediaKind
      rtpParameters: types.RtpParameters
      appData: { source: string }
    },
    cb: (result: any) => void,
  ): Promise<void> {
    try {
      const producer = await this.#client.produce(transportId, kind, rtpParameters, appData)
      cb({ id: producer.id })

      const channel = this.#channels.get(this.#client.channelId)
      if (channel) {
        for (const peer of channel.clients) {
          if (peer.socketId !== this.#client.socketId)
            await peer.createConsumerFor(producer, this.#client.socketId)
        }

        if (kind === 'audio')
          await channel.addAudioProducer(producer)
      }
    }
    catch (error) {
      if (error instanceof Error) {
        consola.error('[Gateway]', '[produce]', error.message)
        cb({ error: error.message })
      }
    }
  }

  #onCloseProducer(
    { producerId }: { producerId: string },
    cb: (result: any) => void,
  ): void {
    try {
      this.#client.closeProducer(producerId)
      cb({ ok: true })
    }
    catch (error) {
      if (error instanceof Error) {
        consola.error('[Gateway]', '[closeProducer]', error.message)
        cb({ error: error.message })
      }
    }
  }

  async #onPauseProducer(
    { producerId }: { producerId: string },
    cb: (result: any) => void,
  ): Promise<void> {
    try {
      await this.#client.pauseProducer(producerId)
      cb({ ok: true })
    }
    catch (error) {
      if (error instanceof Error) {
        consola.error('[Gateway]', '[pauseProducer]', error.message)
        cb({ error: error.message })
      }
    }
  }

  async #onResumeProducer(
    { producerId }: { producerId: string },
    cb: (result: any) => void,
  ): Promise<void> {
    try {
      await this.#client.resumeProducer(producerId)
      cb({ ok: true })
    }
    catch (error) {
      if (error instanceof Error) {
        consola.error('[Gateway]', '[resumeProducer]', error.message)
        cb({ error: error.message })
      }
    }
  }

  async #onPauseConsumer(
    { consumerId }: { consumerId: string },
    cb: (result: any) => void,
  ): Promise<void> {
    try {
      await this.#client.pauseConsumer(consumerId)
      cb({ ok: true })
    }
    catch (error) {
      if (error instanceof Error) {
        consola.error('[Gateway]', '[pauseConsumer]', error.message)
        cb({ error: error.message })
      }
    }
  }

  async #onResumeConsumer(
    { consumerId }: { consumerId: string },
    cb: (result: any) => void,
  ): Promise<void> {
    try {
      await this.#client.resumeConsumer(consumerId)
      cb({ ok: true })
    }
    catch (error) {
      if (error instanceof Error) {
        consola.error('[Gateway]', '[resumeConsumer]', error.message)
        cb({ error: error.message })
      }
    }
  }

  #onUpdateClient(
    patch: { inputMuted?: boolean, outputMuted?: boolean },
    cb: (result: any) => void,
  ): void {
    this.#client.update(patch)
    cb(this.#client.serialize())
  }

  #leaveChannel(channel: Channel): void {
    channel.unwireClient(this.#client)
    channel.kickClient(this.#client)
    this.#socket.leave(channel.id)
  }

  #onDisconnect(): void {
    consola.info('[Gateway]', 'Client disconnected:', this.#client.socketId)

    this.#socket.broadcast.emit('client-disconnected', this.#client.socketId)

    const channel = this.#channels.get(this.#client.channelId)

    if (channel)
      this.#leaveChannel(channel)

    this.#client.close()
  }
}