snapcast

defmodule Snapcast.Session do
  @moduledoc """
  One connected snapclient. Owns the TCP socket, speaks the snapcast protocol, and
  writes the audio chunks the stream hands it.

  On `Hello` it replies with `ServerSettings` and registers with `Snapcast.Server`. The
  server then assigns the active stream and sends *that stream's* `CodecHeader` exactly
  once (PCM via `set_format/3`, FLAC via the stream's `attach`) — the equivalent of
  snapserver's `send(stream->getHeader())`. A connecting client therefore receives one
  codec header for the stream it will play (or none while idle), not a default header
  followed by the real one. It answers `Time` requests for clock sync and tracks the
  client's volume/mute from `ClientInfo`.
  """

  use GenServer, restart: :temporary

  alias Snapcast.{Clock, Protocol, Server}
  require Logger

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts)

  @doc "Hand a framed message to this session to write to its socket."
  def send_frame(pid, frame), do: send(pid, {:snap_send, frame})

  @doc "Start reading from the socket (after the acceptor transfers ownership)."
  def activate(pid), do: GenServer.cast(pid, :activate)

  @doc "Set this client's playback volume (0-100) via a fresh ServerSettings."
  def set_volume(pid, volume), do: GenServer.cast(pid, {:set_volume, volume})

  @doc """
  Re-send ServerSettings (format-scaled bufferMs) for the next stream, plus — for the
  `:pcm` transport — the matching PCM CodecHeader. The `:flac` transport sends its
  CodecHeader from the stream via `send_codec_header/2` once it has been parsed.
  """
  def set_format(pid, format, codec \\ :pcm),
    do: GenServer.cast(pid, {:set_format, format, codec})

  @doc "Write a pre-built codec header payload (e.g. the FLAC stream header) to the client."
  def send_codec_header(pid, payload), do: send(pid, {:snap_codec_header, payload})

  @doc false
  def client_info(pid), do: GenServer.call(pid, :client_info)

  @impl true
  def init(opts) do
    socket = Keyword.fetch!(opts, :socket)

    {:ok,
     %{
       socket: socket,
       buffer: <<>>,
       client_id: nil,
       name: nil,
       volume: 100,
       muted: false,
       registered: false,
       format: Snapcast.format(),
       connected_at: nil
     }}
  end

  @impl true
  def handle_cast(:activate, state) do
    :inet.setopts(state.socket, active: :once)
    {:noreply, state}
  end

  def handle_cast({:set_volume, volume}, state) do
    volume = volume |> round() |> max(0) |> min(100)

    :gen_tcp.send(
      state.socket,
      Protocol.server_settings(Snapcast.buffer_ms_for(state.format), 0, volume, state.muted,
        sent: Clock.now_tv()
      )
    )

    {:noreply, %{state | volume: volume}}
  end

  # A new stream format changes the buffer depth (hi-res gets a deeper buffer), so
  # re-send ServerSettings with the format-scaled bufferMs before the CodecHeader.
  def handle_cast({:set_format, format, codec}, state) do
    fmt = Snapcast.normalize_format(format) || Snapcast.format()

    Logger.debug(
      "[snap] -> #{state.client_id} stream format=#{format_str(fmt)} codec=#{codec} " <>
        "bufferMs=#{Snapcast.buffer_ms_for(fmt)}"
    )

    write(
      state,
      Protocol.server_settings(Snapcast.buffer_ms_for(fmt), 0, state.volume, state.muted,
        sent: Clock.now_tv()
      )
    )

    # PCM advertises its WAV header here; FLAC's header is parsed from the stream and
    # sent via send_codec_header/2, so the decoder is created for the right codec.
    if codec != :flac, do: write_codec_header(state, fmt)
    {:noreply, %{state | format: fmt}}
  end

  @impl true
  def handle_call(:client_info, _from, %{registered: true} = state) do
    {:reply, {:ok, %{pid: self(), client_id: state.client_id, name: state.name}}, state}
  end

  def handle_call(:client_info, _from, state), do: {:reply, :unregistered, state}

  @impl true
  def handle_info({:snap_send, frame}, state) do
    # Time the socket write. A blocking write here also stalls this process's replies to
    # the client's Time-sync requests (same mailbox); if the client stops getting sync
    # replies it hits its ~2s timeout and reconnects. A "slow socket write" warning here
    # therefore pinpoints write-starvation as the cause of a reconnect loop.
    t0 = System.monotonic_time(:millisecond)
    result = :gen_tcp.send(state.socket, frame)
    dt = System.monotonic_time(:millisecond) - t0

    if dt >= 200 do
      {:message_queue_len, qlen} = Process.info(self(), :message_queue_len)

      Logger.warning(
        "[snap] slow socket write #{dt}ms (mailbox=#{qlen}, bytes=#{byte_size(frame)}): " <>
          "id=#{state.client_id}#{connected_for(state)}"
      )
    end

    case result do
      :ok ->
        {:noreply, state}

      {:error, reason} ->
        Logger.warning(
          "[snap] write failed: id=#{state.client_id} reason=#{inspect(reason)}#{connected_for(state)}"
        )

        {:stop, :normal, state}
    end
  end

  def handle_info({:snap_codec_header, payload}, state) do
    Logger.debug("[snap] -> #{state.client_id} flac codec header (#{byte_size(payload)} bytes)")
    write(state, Protocol.codec_header_flac(payload, sent: Clock.now_tv()))
    {:noreply, state}
  end

  def handle_info({:tcp, socket, data}, state) do
    :inet.setopts(socket, active: :once)
    state = process(%{state | buffer: state.buffer <> data})
    {:noreply, state}
  end

  def handle_info({:tcp_closed, _socket}, state) do
    Logger.info(
      "[snap] client disconnected: id=#{state.client_id} (tcp_closed)#{connected_for(state)}"
    )

    {:stop, :normal, state}
  end

  def handle_info({:tcp_error, _socket, reason}, state) do
    Logger.warning(
      "[snap] client tcp_error: id=#{state.client_id} reason=#{inspect(reason)}#{connected_for(state)}"
    )

    {:stop, :normal, state}
  end

  def handle_info(_msg, state), do: {:noreply, state}

  @impl true
  def terminate(_reason, state) do
    if state.registered, do: Server.unregister(self())
    :ok
  end

  # --- message loop ----------------------------------------------------------

  defp process(state) do
    case Protocol.decode(state.buffer) do
      {:ok, msg, rest} ->
        state = handle_message(msg, %{state | buffer: rest})
        process(state)

      :incomplete ->
        state

      {:error, _buffer} ->
        # Unparseable — drop the connection rather than spin.
        :gen_tcp.close(state.socket)
        state
    end
  end

  # Hello → ServerSettings + CodecHeader, then register for audio.
  defp handle_message(%{type: 5, id: id, body: {:hello, hello}}, state) do
    client_id = hello["ID"] || hello["MAC"] || hello["mac"] || "unknown"
    name = hello["HostName"] || hello["ClientName"] || client_id

    # Log the client's reported build — the snapclient version/OS is the quickest way to
    # spot a client that can't handle a given transport (e.g. a build without FLAC).
    Logger.info(
      "[snap] client connected: id=#{client_id} name=#{name} version=#{hello["Version"]} " <>
        "os=#{hello["OS"]} arch=#{hello["Arch"]} proto=#{hello["SnapStreamProtocolVersion"]}"
    )

    # Hello response is ServerSettings only. The CodecHeader is the assigned stream's
    # header and is sent once when the server assigns the stream (Server.register ->
    # set_format/attach), mirroring snapserver's `send(stream->getHeader())`. Sending a
    # default header here would make the client build a decoder it immediately discards.
    write(
      state,
      Protocol.server_settings(Snapcast.buffer_ms(), 0, state.volume, state.muted,
        sent: Clock.now_tv(),
        refers_to: id
      )
    )

    Server.register(self(), client_id, name)
    %{state | client_id: client_id, name: name, registered: true, connected_at: now_ms()}
  end

  # Time → reply so the client can sync its clock. latency = serverReceived -
  # clientSent; the response's base `sent` is the server time at send.
  defp handle_message(%{type: 4, id: id, body: {:time, _latency}, sent: client_sent}, state) do
    server_recv_us = Clock.now_us()
    latency = Clock.us_to_tv(server_recv_us - Clock.tv_to_us(client_sent))
    write(state, Protocol.time_response(latency, id: id, refers_to: id, sent: Clock.now_tv()))
    state
  end

  defp handle_message(%{type: 7, body: {:client_info, info}}, state) do
    %{state | volume: info["volume"] || state.volume, muted: info["muted"] || false}
  end

  defp handle_message(_msg, state), do: state

  defp write_codec_header(state, {rate, bits, channels}) do
    write(state, Protocol.codec_header_pcm(rate, bits, channels, sent: Clock.now_tv()))
  end

  defp write(state, frame), do: :gen_tcp.send(state.socket, frame)

  defp now_ms, do: System.monotonic_time(:millisecond)

  # " after Nms" once the client has registered, for connection-lifetime logging.
  defp connected_for(%{connected_at: at}) when is_integer(at), do: " after #{now_ms() - at}ms"
  defp connected_for(_state), do: ""

  defp format_str({rate, bits, channels}), do: "#{rate}/#{bits}/#{channels}"
  defp format_str(_format), do: "?"
end