defmodule SendFromFile.PeerHandler do
  require Logger

  import Bitwise

  alias ExWebRTC.{
    ICECandidate,
    MediaStreamTrack,
    PeerConnection,
    RTPCodecParameters,
    SessionDescription
  }

  alias ExWebRTC.Media.{IVF, Ogg}
  alias ExWebRTC.RTP.Payloader

  @behaviour WebSock

  @video_file "./video.ivf"
  @audio_file "./audio.ogg"

  @max_rtp_timestamp (1 <<< 32) - 1
  @max_rtp_seq_no (1 <<< 16) - 1

  @ice_servers [
    %{urls: "stun:stun.l.google.com:19302"}
  ]

  @video_codecs [
    %RTPCodecParameters{
      payload_type: 96,
      mime_type: "video/VP8",
      clock_rate: 90_000
    }
  ]

  @audio_codecs [
    %RTPCodecParameters{
      payload_type: 111,
      mime_type: "audio/opus",
      clock_rate: 48_000,
      channels: 2
    }
  ]

  @impl true
  def init(_) do
    {:ok, pc} =
      PeerConnection.start_link(
        ice_servers: @ice_servers,
        video_codecs: @video_codecs,
        audio_codecs: @audio_codecs
      )

    stream_id = MediaStreamTrack.generate_stream_id()
    video_track = MediaStreamTrack.new(:video, [stream_id])
    audio_track = MediaStreamTrack.new(:audio, [stream_id])

    {:ok, _sender} = PeerConnection.add_track(pc, video_track)
    {:ok, _sender} = PeerConnection.add_track(pc, audio_track)

    {:ok, _header, video_reader} = IVF.Reader.open(@video_file)
    {:ok, video_payloader} = @video_codecs |> hd() |> Payloader.new(max_payload_size: 800)

    {:ok, audio_reader} = Ogg.Reader.open(@audio_file)
    {:ok, audio_payloader} = @audio_codecs |> hd() |> Payloader.new()

    state = %{
      peer_connection: pc,
      video_track_id: video_track.id,
      audio_track_id: audio_track.id,
      video_reader: video_reader,
      video_payloader: video_payloader,
      audio_reader: audio_reader,
      audio_payloader: audio_payloader,
      next_video_timestamp: Enum.random(0..@max_rtp_timestamp),
      next_audio_timestamp: Enum.random(0..@max_rtp_timestamp),
      next_video_sequence_number: Enum.random(0..@max_rtp_seq_no),
      next_audio_sequence_number: Enum.random(0..@max_rtp_seq_no)
    }

    {:ok, offer} = PeerConnection.create_offer(pc)
    :ok = PeerConnection.set_local_description(pc, offer)

    offer_json = SessionDescription.to_json(offer)

    msg =
      %{"type" => "offer", "data" => offer_json}
      |> Jason.encode!()

    Logger.info("Sent SDP offer: #{inspect(offer_json)}")

    {:push, {:text, msg}, state}
  end

  @impl true
  def handle_in({msg, [opcode: :text]}, state) do
    msg
    |> Jason.decode!()
    |> handle_ws_msg(state)

    {:ok, state}
  end

  @impl true
  def handle_info({:ex_webrtc, _from, msg}, state) do
    handle_webrtc_msg(msg, state)
  end

  @impl true
  def handle_info(:send_video, state) do
    # 30 =~ 1000 millisecond / 30 FPS
    Process.send_after(self(), :send_video, 30)

    case IVF.Reader.next_frame(state.video_reader) do
      {:ok, frame} ->
        {rtp_packets, payloader} = Payloader.payload(state.video_payloader, frame.data)

        # 3_000 = 90_000 (VP8 clock rate) / 30 FPS
        next_sequence_number =
          Enum.reduce(rtp_packets, state.next_video_sequence_number, fn packet, sequence_number ->
            packet = %{
              packet
              | timestamp: state.next_video_timestamp,
                sequence_number: sequence_number
            }

            PeerConnection.send_rtp(state.peer_connection, state.video_track_id, packet)

            sequence_number + 1 &&& @max_rtp_seq_no
          end)

        next_timestamp = state.next_video_timestamp + 3_000 &&& @max_rtp_timestamp

        state = %{
          state
          | video_payloader: payloader,
            next_video_timestamp: next_timestamp,
            next_video_sequence_number: next_sequence_number
        }

        {:ok, state}

      :eof ->
        Logger.info("Video file finished. Looping...")
        {:ok, _header, reader} = IVF.Reader.open(@video_file)
        {:ok, %{state | video_reader: reader}}

      {:error, reason} ->
        Logger.error("Error when reading IVF, reason: #{inspect(reason)}")
        {:ok, state}
    end
  end

  @impl true
  def handle_info(:send_audio, state) do
    case Ogg.Reader.next_packet(state.audio_reader) do
      {:ok, {packet, duration}, reader} ->
        # in real-life scenario, you will need to conpensate for `Process.send_after/3` error
        # and time spent on reading and parsing the file
        Process.send_after(self(), :send_audio, duration)

        {[rtp_packet], payloader} = Payloader.payload(state.audio_payloader, packet)

        rtp_packet = %{
          rtp_packet
          | timestamp: state.next_audio_timestamp,
            sequence_number: state.next_audio_sequence_number
        }

        PeerConnection.send_rtp(state.peer_connection, state.audio_track_id, rtp_packet)

        # Ogg.Reader.next_packet/1 returns duration in ms
        # we have to convert it to RTP timestamp difference
        timestamp_delta = trunc(duration * 48_000 / 1000)
        next_timestamp = state.next_audio_timestamp + timestamp_delta &&& @max_rtp_timestamp
        next_sequence_number = state.next_audio_sequence_number + 1 &&& @max_rtp_seq_no

        state = %{
          state
          | audio_reader: reader,
            audio_payloader: payloader,
            next_audio_timestamp: next_timestamp,
            next_audio_sequence_number: next_sequence_number
        }

        {:ok, state}

      :eof ->
        send(self(), :send_audio)
        Logger.info("Audio file finished. Looping...")
        {:ok, reader} = Ogg.Reader.open(@audio_file)
        {:ok, %{state | audio_reader: reader}}

      {:error, reason} ->
        Logger.error("Error when reading Ogg, reason: #{inspect(reason)}")
        {:ok, state}
    end
  end

  @impl true
  def handle_info({:EXIT, pc, reason}, %{peer_connection: pc} = state) do
    # Bandit traps exits under the hood so our PeerConnection.start_link
    # won't automatically bring this process down.
    Logger.info("Peer connection process exited, reason: #{inspect(reason)}")
    {:stop, {:shutdown, :pc_closed}, state}
  end

  @impl true
  def terminate(reason, _state) do
    Logger.info("WebSocket connection was terminated, reason: #{inspect(reason)}")
  end

  defp handle_ws_msg(%{"type" => "answer", "data" => data}, state) do
    Logger.info("Received SDP answer: #{inspect(data)}")

    answer = SessionDescription.from_json(data)
    :ok = PeerConnection.set_remote_description(state.peer_connection, answer)
  end

  defp handle_ws_msg(%{"type" => "ice", "data" => data}, state) do
    Logger.info("Received ICE candidate: #{inspect(data)}")

    candidate = ICECandidate.from_json(data)
    :ok = PeerConnection.add_ice_candidate(state.peer_connection, candidate)
  end

  defp handle_webrtc_msg({:ice_candidate, candidate}, state) do
    candidate_json = ICECandidate.to_json(candidate)

    msg =
      %{"type" => "ice", "data" => candidate_json}
      |> Jason.encode!()

    Logger.info("Sent ICE candidate: #{inspect(candidate_json)}")

    {:push, {:text, msg}, state}
  end

  defp handle_webrtc_msg({:connection_state_change, :connected}, state) do
    Logger.info("Connection established, starting to send media")

    send(self(), :send_video)
    send(self(), :send_audio)

    {:ok, state}
  end

  defp handle_webrtc_msg({:connection_state_change, conn_state}, state) do
    Logger.info("Connection state changed: #{conn_state}")

    if conn_state == :failed do
      {:stop, {:shutdown, :pc_failed}, state}
    else
      {:ok, state}
    end
  end

  defp handle_webrtc_msg(_msg, state), do: {:ok, state}
end
