path: root/lib/polyjuice/client/sync.ex

# Copyright 2019-2020 Hubert Chathi <hubert@uhoreg.ca>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

defmodule Polyjuice.Client.Sync do
  # Matrix sync worker.  Calls /sync and sends messages to the listener.
  @moduledoc false
  use Task, restart: :permanent
  require Logger

  @doc """
  Start a sync task.
  """
  @spec start_link([...]) :: {:ok, pid}
  def start_link([client, listener | opts])
      when (is_pid(listener) or is_function(listener)) and is_list(opts) do
    Task.start_link(__MODULE__, :sync, [client, listener, opts])
  end

  @enforce_keys [:send, :access_token, :homeserver_url, :uri, :user_id, :storage]
  defstruct [
    :send,
    :conn_ref,
    :access_token,
    :homeserver_url,
    :uri,
    :user_id,
    :storage,
    :since,
    query_params: "",
    backoff: nil,
    set_filter: nil,
    initial_done: false
  ]

  @sync_path "_matrix/client/r0/sync"
  @sync_timeout 30000
  @buffer_timeout 10000

  @doc false
  def sync(
        %Polyjuice.Client{
          access_token: access_token,
          base_url: homeserver_url,
          user_id: user_id,
          storage: storage
        },
        listener,
        opts
      ) do
    # Make sure the URL ends with a slash, so that URI.merge doesn't clobber
    # the last path component.  (URI.merge is smart enough to drop the double
    # "/" if it already ends with a slash.)
    homeserver_url = homeserver_url <> "/"

    # Figure out how to handle the filter (if any): can we pass it in straight
    # to the query, or do we need to get its ID.  And if we get its ID, do we
    # already have it, or do we need to send it to the server?
    {filter, set_filter} =
      case Keyword.get(opts, :filter) do
        nil ->
          {nil, nil}

        f when is_binary(f) ->
          {f, nil}

        f when is_map(f) ->
          case Polyjuice.Client.Storage.get_filter_id(storage, f) do
            nil -> {nil, f}
            id -> {id, nil}
          end
      end

    query_params =
      Enum.reduce(
        opts,
        if filter do
          [{"timeout", @sync_timeout}, {"filter", filter}]
        else
          [{"timeout", @sync_timeout}]
        end,
        fn
          {:full_state, full_state}, acc -> [{"full_state", full_state} | acc]
          {:set_presence, set_presence}, acc -> [{"set_presence", set_presence} | acc]
          _, acc -> acc
        end
      )
      |> URI.encode_query()

    uri = URI.merge(homeserver_url, @sync_path)

    connect(%__MODULE__{
      send: if(is_function(listener), do: listener, else: &send(listener, &1)),
      access_token: access_token,
      homeserver_url: homeserver_url,
      user_id: user_id,
      uri: uri,
      query_params: query_params,
      storage: storage,
      since: Polyjuice.Client.Storage.get_sync_token(storage),
      set_filter: set_filter
    })
  end

  defp calc_backoff(backoff), do: if(backoff, do: min(backoff * 2, 30), else: 1)

  defp connect(state) do
    if state.backoff, do: :timer.sleep(state.backoff * 1000)

    uri = state.uri
    options = [recv_timeout: @sync_timeout + @buffer_timeout]

    case %URI{scheme: uri.scheme, host: uri.host, port: uri.port}
         |> URI.to_string()
         |> :hackney.connect(options) do
      {:ok, conn_ref} ->
        Logger.info("Connected to sync")
        state.send.({:connected})

        if state.set_filter do
          set_filter(%{state | conn_ref: conn_ref, backoff: nil})
        else
          do_sync(%{state | conn_ref: conn_ref, backoff: nil})
        end

      # FIXME: what errors do we need to handle differently?
      {:error, err} ->
        backoff = calc_backoff(state.backoff)
        Logger.error("Sync error: #{err}; retrying in #{backoff} seconds.")
        connect(%{state | backoff: backoff})
    end
  end

  defp set_filter(state) do
    if state.backoff, do: :timer.sleep(state.backoff * 1000)

    Logger.debug("Setting filter")

    e = &URI.encode_www_form/1

    path =
      URI.merge(
        state.homeserver_url,
        "#{Polyjuice.Client.prefix_r0()}/user/#{e.(state.user_id)}/filter"
      ).path

    headers = [
      {"Accept", "application/json"},
      {"Content-Type", "application/json"},
      {"Authorization", "Bearer #{state.access_token}"}
    ]

    case :hackney.send_request(
           state.conn_ref,
           {:post, path, headers, Jason.encode!(state.set_filter)}
         ) do
      {:ok, status_code, _resp_headers, client_ref} ->
        case status_code do
          200 ->
            {:ok, body} = :hackney.body(client_ref)

            with {:ok, %{} = json_body} <- Jason.decode(body),
                 filter_id = Map.get(json_body, "filter_id") do
              Logger.debug("got filter id #{filter_id}")

              Polyjuice.Client.Storage.set_filter_id(
                state.storage,
                state.set_filter,
                filter_id
              )

              do_sync(%{
                state
                | query_params: "#{state.query_params}&filter=#{e.(filter_id)}",
                  set_filter: nil
              })
            else
              _ ->
                backoff = calc_backoff(state.backoff)
                Logger.error("Server sent us garbage; retrying in #{backoff} seconds")
                set_filter(%{state | backoff: backoff})
            end

          _ ->
            {:ok, body} = :hackney.body(client_ref)
            Logger.warn("Unable to set filter for sync.  Ignoring.  Got message: #{body}")
            do_sync(%{state | set_filter: nil})
        end

      # if the request timed out, try again
      {:error, :timeout} ->
        Logger.info("set filter timed out")
        set_filter(%{state | backoff: nil})

      {:error, :closed} ->
        backoff = calc_backoff(state.backoff)
        Logger.error("Set filter error: closed; retrying in #{backoff} seconds.")
        connect(%{state | backoff: backoff, conn_ref: nil})
        state.send.({:disconnected})

      # FIXME: what other error codes do we need to handle?
      {:error, err} ->
        # for other errors, we retry with exponential backoff
        backoff = calc_backoff(state.backoff)
        Logger.error("Set filter error: #{err}; retrying in #{backoff} seconds.")
        set_filter(%{state | backoff: backoff})
    end
  end

  defp do_sync(state) do
    if state.backoff, do: :timer.sleep(state.backoff * 1000)

    headers = [
      {"Accept", "application/json"},
      {"Authorization", "Bearer #{state.access_token}"}
    ]

    path =
      state.uri.path <>
        "?" <>
        state.query_params <>
        if state.since, do: "&since=" <> URI.encode_www_form(state.since), else: ""

    case :hackney.send_request(state.conn_ref, {:get, path, headers, ""}) do
      {:ok, status_code, _resp_headers, client_ref} ->
        case status_code do
          200 ->
            {:ok, body} = :hackney.body(client_ref)

            with {:ok, json_body} <- Jason.decode(body),
                 %{"next_batch" => next_batch} <- json_body do
              if state.backoff, do: Logger.info("Sync resumed")
              process_body(json_body, state)
              Polyjuice.Client.Storage.set_sync_token(state.storage, next_batch)

              if not state.initial_done do
                state.send.({:initial_sync_completed})
              end

              do_sync(%{state | since: next_batch, backoff: nil, initial_done: true})
            else
              _ ->
                backoff = calc_backoff(state.backoff)
                Logger.error("Server sent us garbage; retrying in #{backoff} seconds")
                do_sync(%{state | backoff: backoff})
            end

          # FIXME: other status codes/error messages
          _ ->
            backoff = calc_backoff(state.backoff)
            Logger.error("Unexpected status code #{status_code}; retrying in #{backoff} seconds")
            do_sync(%{state | backoff: backoff})
        end

      # if the request timed out, try again
      {:error, :timeout} ->
        Logger.info("sync timed out")
        do_sync(%{state | backoff: nil})

      {:error, :closed} ->
        backoff = calc_backoff(state.backoff)
        Logger.error("Sync error: closed; retrying in #{backoff} seconds.")
        connect(%{state | backoff: backoff, conn_ref: nil})
        state.send.({:disconnected})

      # FIXME: what other error codes do we need to handle?
      {:error, err} ->
        # for other errors, we retry with exponential backoff
        backoff = calc_backoff(state.backoff)
        Logger.error("Sync error: #{err}; retrying in #{backoff} seconds.")
        do_sync(%{state | backoff: backoff})
    end
  end

  defp process_body(body, state) do
    rooms = Map.get(body, "rooms", %{})

    rooms
    |> Map.get("join", [])
    |> Enum.each(fn {k, v} -> process_room(k, v, state) end)

    rooms
    |> Map.get("invite", [])
    |> Enum.each(fn {k, v} -> process_invite(k, v, state) end)

    rooms
    |> Map.get("leave", [])
    |> Enum.each(fn {k, v} ->
      process_room(k, v, state)
      state.send.({:left, k})
    end)
  end

  defp process_room(roomname, room, state) do
    timeline = Map.get(room, "timeline", %{})

    if Map.get(timeline, "limited", false) do
      with {:ok, prev_batch} <- Map.fetch(timeline, "prev_batch") do
        state.send.({:limited, roomname, prev_batch})
      end
    end

    room
    |> Map.get("state", %{})
    |> Map.get("events", [])
    |> Enum.each(&process_event(&1, roomname, state))

    timeline
    |> Map.get("events", [])
    |> Enum.each(&process_event(&1, roomname, state))
  end

  defp process_event(
         %{
           "state_key" => _state_key
         } = event,
         roomname,
         state
       ) do
    state.send.({:state, roomname, event})

    state
  end

  defp process_event(
         %{} = event,
         roomname,
         state
       ) do
    state.send.({:message, roomname, event})

    state
  end

  defp process_event(_, _, state) do
    state
  end

  defp process_invite(roomname, room, state) do
    # The invite state is a map from state type to state key to event.
    invite_state =
      Map.get(room, "invite_state", %{})
      |> Map.get("events", [])
      |> Enum.reduce(
        %{},
        fn
          %{
            "type" => type,
            "state_key" => state_key
          } = val,
          acc ->
            Map.get(acc, type, %{})
            |> Map.put(state_key, val)
            |> (&Map.put(acc, type, &1)).()

          _, acc ->
            acc
        end
      )

    inviter =
      invite_state
      |> Map.get("m.room.member", %{})
      |> Map.get(state.user_id, %{})
      |> Map.get("sender")

    if inviter do
      state.send.({:invite, roomname, inviter, invite_state})
    end

    state
  end
end