activity_pub_client

defmodule ActivityPubClient.SyncServer do
  @moduledoc """
    GenServer used to keep a WebSocket connection to an ActivityPub instance.

    Reference: https://docs.joinmastodon.org/methods/timelines/streaming/#websocket-a-idwebsocketa

    With inspiration from:
    https://github.com/GamgeeNL/websocket-client
  """
  use GenServer
  require Logger

  defstruct client: nil,
            type: nil,
            # These will be set after gun connects
            stream_ref: nil,
            gun_pid: nil,
            # Optionally set this if you want to control this client from elsewhere
            # Use :set_from
            from: nil

  @doc """
  Starts a `SyncServer` process.

  Expects:
  - an `%OAuth2.Client{}` struct as returned by `Client`,
  - the type of streaming, can be one of:
    - user
    - public
    - public:local
    - hashtag
    - hashtag:local
    - list
    - direct
  - an optional callback, which will be called when the WebSocket receives data.
    In the absence of a callback, the will send a message to the process that
    calls `open_ws/1`.
  """
  def start(client, type, callback \\ nil) do
    opts = [
      client: client,
      type: type,
      callback: callback,
      name: {:via, Registry, {ActivityPubClient.SyncRegistry, client.token.access_token}}
    ]

    DynamicSupervisor.start_child(
      ActivityPubClient.SyncSupervisor,
      {__MODULE__, opts}
    )
  end

  @doc """
  Upgrades the connection to WebSocket.
  """
  def open_ws(client) do
    {:ok, pid} = lookup(client)
    send_sync(pid, {:open_ws})
  end

  defp lookup(client) do
    case Registry.lookup(ActivityPubClient.SyncRegistry, client.token.access_token) do
      [{pid, _}] -> {:ok, pid}
      [] -> {:error, :not_found}
    end
  end

  # This supports sending any of the handle_call functions which are accessed from WebsocketClient.Application
  defp send_sync(pid, args_tuple) do
    GenServer.call(pid, args_tuple)
  end

  # This supports sending any of the handle_cast functions which are accessed from WebsocketClient.Application
  defp send_async(pid, args_tuple) do
    GenServer.cast(pid, args_tuple)
  end

  @impl true
  def init(opts) do
    Logger.debug("Websocket client started...")

    # Set initial state as default values and return from `init`
    {:ok, %__MODULE__{client: Keyword.fetch!(opts, :client), type: Keyword.fetch!(opts, :type)}}
  end

  # This sends back the current state of this server
  @impl true
  def handle_call({:get_state}, _from, state) do
    {:reply, state, state}
  end

  # This opens the websocket which includes
  # get > upgrade > success
  # This sets the relevant values in state
  # This is blocking so the client can't send until the websocket is open
  #
  # This also sets from so you can send the results back to the calling process
  @impl true
  def handle_call({:open_ws}, {from, _call}, state) do
    new_state = ws_upgrade(state)
    {:reply, :ok, %__MODULE__{new_state | from: from}}
  end

  # If you want to set the calling process of this server you can do it here
  @impl true
  def handle_cast({:set_from, from}, state) do
    {:noreply, %__MODULE__{state | from: from}}
  end

  # This sends a message after the websocket connection has been opened
  @impl true
  def handle_cast({:send_message, message}, %{gun_pid: gun_pid} = state) do
    case :gun.ws_send(gun_pid, {:text, message}) do
      :ok ->
        {:noreply, state}

      reply ->
        Logger.info(
          "Error on sending message from gun. Maybe the websocket is not open... #{inspect(reply)}"
        )

        {:noreply, state}
    end
  end

  ####
  # gun upgrade functions
  ####

  # This is the one it hits if upgrade is successful and from is nil (not set)
  @impl true
  def handle_info(
        {:gun_upgrade, gun_pid, stream_ref, ["websocket"], headers},
        %{stream_ref: stream_ref, gun_pid: gun_pid, from: nil, callback: nil} = state
      ) do
    Logger.info("Upgraded #{inspect(gun_pid)}. Success!\nHeaders:\n#{inspect(headers)}")
    {:noreply, state}
  end

  # This is the one it hits if upgrade is successful and, from is nil (not set)
  # but a callback is set
  @impl true
  def handle_info(
        {:gun_upgrade, gun_pid, stream_ref, ["websocket"], headers},
        %{stream_ref: stream_ref, gun_pid: gun_pid, from: nil, callback: callback} = state
      ) do
    Logger.debug("Upgraded #{inspect(gun_pid)}. Success!\nHeaders:\n#{inspect(headers)}")
    # Give from process a message that the websocket is now open.
    callback.({:websocket_open})
    {:noreply, state}
  end

  # This is the one it hits if upgrade is successful and from is set
  # This sends a message back to the calling function that the websocket is open and can send and receive messages
  @impl true
  def handle_info(
        {:gun_upgrade, gun_pid, stream_ref, ["websocket"], headers},
        %{stream_ref: stream_ref, gun_pid: gun_pid, from: from} = state
      ) do
    Logger.debug("Upgraded #{inspect(gun_pid)}. Success!\nHeaders:\n#{inspect(headers)}")
    # Give from process a message that the websocket is now open.
    send(from, {:websocket_open})
    {:noreply, state}
  end

  # Upgrade not successful
  @impl true
  def handle_info({:gun_response, _gun_pid, _, _, status, headers}, _) do
    Logger.info("Websocket upgrade failed.")
    exit({:ws_upgrade_failed, status, headers})
  end

  # Error on upgrade
  @impl true
  def handle_info({:gun_error, _gun_pid, _stream_ref, reason}, _) do
    exit({:ws_upgrade_failed, reason})
  end

  ####
  # End of upgrade functions
  ####

  ####
  # gun messages
  ####

  # Receiving message with no 'from' process
  @impl true
  def handle_info(
        {:gun_ws, gun_pid, stream_ref, {:text, message}},
        %{stream_ref: stream_ref, gun_pid: gun_pid, from: nil, callback: nil} = state
      ) do
    Logger.info("Message received #{inspect(decode_message(message))}")
    {:noreply, state}
  end

  # Receiving message with no 'from' process
  # but a callback is set
  @impl true
  def handle_info(
        {:gun_ws, gun_pid, stream_ref, {:text, message}},
        %{stream_ref: stream_ref, gun_pid: gun_pid, from: nil, callback: callback} = state
      ) do
    callback.(%{activitypub_message: decode_message(message)})
    {:noreply, state}
  end

  # Gun receives a message from the backend and we return it to the calling process.
  @impl true
  def handle_info(
        {:gun_ws, gun_pid, stream_ref, {:text, message}},
        %{stream_ref: stream_ref, gun_pid: gun_pid, from: from} = state
      ) do
    # Give it back to the calling process
    # If you're using another GenServer you should probably use GenServer.reply/2
    send(from, %{activitypub_message: decode_message(message)})
    {:noreply, state}
  end

  # If the gun_pid matches, restart? - currently restarting on below call
  @impl true
  def handle_info({:gun_down, gun_pid, _http_ws, :closed, [], []}, %{gun_pid: gun_pid} = state) do
    # Toggle this to restart here
    # {:noreply, ws_upgrade(state) }
    {:noreply, state}
  end

  # If the gun_pid doesn't match, don't restart, it's probably obsolete.
  @impl true
  def handle_info(
        {:gun_down, _alt_gun_pid, _http_ws, :closed, [], []},
        %{gun_pid: _gun_pid} = state
      ) do
    {:noreply, state}
  end

  # This is instructing gun to close, so we restart
  # This could be more granular by matching on the code which is currently ignored.
  @impl true
  def handle_info(
        {:gun_ws, gun_pid, stream_ref, {:close, _code, ""}},
        %{stream_ref: stream_ref, gun_pid: gun_pid} = state
      ) do
    # Re-open the websocket connection if it closes here.
    {:noreply, ws_upgrade(state)}
  end

  # Handle gun_up - this can be monitored if desired.
  # http_ws is :http or :ws
  @impl true
  def handle_info({:gun_up, gun_pid, _http_ws}, %{gun_pid: gun_pid} = state) do
    {:noreply, state}
  end

  # Handle gun_up - this can be monitored if desired.
  # http_ws is :http or :ws
  @impl true
  def handle_info({:gun_up, _alt_gun_pid, _http_ws}, %{gun_pid: _gun_pid} = state) do
    {:noreply, state}
  end

  # If this hits, something unexpected happened. Perhaps an error.
  # This is meant to be a catch all for all other messages.
  @impl true
  def handle_info(message, state) do
    Logger.error(
      "Unexpected message: #{inspect(message, pretty: true)} with state: #{inspect(state, pretty: true)}"
    )

    {:noreply, state}
  end

  # This gets the websocket to the state of upgrade where the upgrade message
  # needs to be received in one of the handle_info functions
  defp ws_upgrade(%{client: client, type: type} = state) do
    host = URI.parse(client.site).host |> String.to_charlist()
    port = URI.parse(client.site).port
    path = '/api/v1/streaming?stream=' ++ String.to_charlist(type)

    {:ok, _} = :application.ensure_all_started(:gun)

    connect_opts = %{
      connect_timeout: :timer.minutes(1),
      retry: 10,
      retry_timeout: 300
    }

    headers = [
      {'authorization',
       String.to_charlist(client.token.token_type) ++
         ' ' ++ String.to_charlist(client.token.access_token)}
    ]

    {:ok, gun_pid} = :gun.open(host, port, connect_opts)
    {:ok, _protocol} = :gun.await_up(gun_pid)
    # Set custom header with cookie for device id
    stream_ref = :gun.ws_upgrade(gun_pid, path, headers)
    # Return updated state
    %__MODULE__{state | stream_ref: stream_ref, gun_pid: gun_pid}
  end

  defp decode_message(message) do
    message = Jason.decode!(message)
    Map.update!(message, "payload", &Jason.decode!/1)
  end
end