mtproto

defmodule MTProto do
  use Connection

  require Logger

  alias MTProto.{Auth, Crypto, Packet}

  def start_link(notifier_pid) do
    Connection.start_link(__MODULE__, [notifier: notifier_pid])
  end

  def notifier_process(client, notifier_pid) do
    Connection.call(client, {:notifier_process, notifier_pid})
  end

  def authorize(client) do
    Connection.call(client, :authorize)
  end

  def authorize(client, auth_key, auth_key_hash, server_salt) do
    Connection.call(client, {:authorize, auth_key, auth_key_hash, server_salt})
  end

  def send_request(client, request) do
    Connection.call(client, {:send, request})
  end

  def dump_state(client) do
    Connection.call(client, :dump_state)
  end

  # ...

  defmodule State do
    @moduledoc """
    :notifier - pid of the process where MTProto client sends notifications;
    :socket - TCP socket;
    :packet_buffer - gluing incomplete TCP packets into fully one;
    :session_id - MTProto session ID, generated for every client instance (process);

    :auth_state authentication state, can be one of these values:
      * :connected - when client only connected to the telegram server;
      * :req_pq - first request to the server in authorization flow,
        requests P and Q numbers;
      * :req_dh_params - second request to the server, for getting DH data;
      * :dh_gen - last request to the server, to check new_nonce_hashN;
      * :encrypted - authorized state of the client, when it has auth_key,
        auth_key_hash and server_salt.

    :auth_params - for storing authentication params between steps, used AuthParams struct;
    :auth_key - authentication key, see MTProto docs for details;
    :auth_key_hash - authentication key hash;
    :server_salt - telegram server salt;
    :msg_seqno - used for storing message sequence number, for using in MTProto packets;
    :msg_ids - list of message IDs, using for server acks;
    :dc_options - list of telegram datacenters;
    :dc - current datacenter ID.
    """
    defstruct [:notifier, :socket,
               :packet_buffer,
               :session_id,
               :auth_state, :auth_params,
               :auth_key, :auth_key_hash, :server_salt,
               :msg_seqno, :msg_ids,
               :dc_options, :dc]
  end

  defmodule AuthParams do
    defstruct [:nonce, :server_nonce, :new_nonce, :nonce_hash1]
  end

  def init(opts) do
    {:connect, :init,
      %State{notifier: opts[:notifier], packet_buffer: <<>>,
             msg_seqno: 0, msg_ids: []}}
  end

  def connect(_, %{socket: nil,} = state) do
    {host, port} = choose_server(state)
    Logger.debug("connect to #{inspect host}, #{inspect port}")

    case :gen_tcp.connect(host, port, [:binary, active: false], 5000) do
      {:ok, socket} ->
        Logger.debug("connected")
        send(self, :after_connect)
        send(state.notifier, :connected)
        session_id = Crypto.make_session_id()
        {:ok, %{state|socket: socket, session_id: session_id, auth_state: :connected}}
      {:error, _} ->
        {:backoff, 1000, state}
    end
  end

  def disconnect(info, %{socket: socket} = state) do
    Logger.debug("disconnected #{inspect info}")
    :ok = :gen_tcp.close(socket)
    :error_logger.format("Connection error: ~p~n", [info])
    {:connect, :reconnect, %{state | socket: nil}}
  end

  def handle_call(:dump_state, _, state) do
    {:reply, state, state}
  end
  def handle_call(:authorize, _, state) do
    # init MTProto authorization
    state = Auth.init(self(), state)

    {:reply, :ok, state}
  end
  def handle_call({:authorize, auth_key, auth_key_hash, server_salt}, _, state) do
    # set auth credentials
    state =
      %{state|auth_state: :encrypted, auth_key: auth_key,
              auth_key_hash: auth_key_hash, server_salt: server_salt}

    # notify client about authentication
    send(self, :authorized)

    # set auth credentials
    {:reply, :ok, state}
  end
  def handle_call({:notifier_process, pid}, _, state) do
    {:reply, :ok, %{state|notifier: pid}}
  end
  def handle_call({:send, request}, _, %{socket: socket} = state) do
    case send_rpc_request(socket, request, state) do
      {:ok, state} -> {:reply, :ok, state}
      {:error, reason, state} -> {:disconnect, {:error, reason}, state}
    end
  end
  def handle_call(:close, from, s) do
    {:disconnect, {:close, from}, s}
  end
  def handle_call(_, _, state) do
    {:noreply, state}
  end

  def handle_info(:after_connect, %{socket: socket} = state) do
    # first packet
    :ok = :gen_tcp.send(socket, <<0xef>>)

    # allow to receive first packet
    next_packet(socket)

    Logger.debug("connection ok!")

    {:noreply, state}
  end
  def handle_info({:reconnect, reason}, state) do
    Logger.debug("reconnect #{inspect reason}")
    {:disconnect, {:reconnect, reason}, state}
  end
  def handle_info(:authorized, %{socket: socket} = state) do
    # notify about authorization result
    send(state.notifier,
      {:authorized, state.auth_key, state.auth_key_hash, state.server_salt})

    # init connection
    case send_rpc_request(socket, init_connection_request(), state) do
      {:ok, state} -> {:noreply, state}
      {:error, reason, state} -> {:disconnect, {:error, reason}, state}
    end
  end
  def handle_info({:send_bare, packet}, %{socket: socket} = state) do
    case :gen_tcp.send(socket, packet) do
      :ok ->
        {:noreply, state}
      {:error, reason} ->
        {:disconnect, {:error, reason}, state}
    end
  end
  def handle_info({:tcp, socket, packet}, state) do
    case Packet.decode(<<state.packet_buffer :: binary, packet :: binary>>) do
      {:more, _need_bytes, incomplete_packet} ->
        next_packet(socket)
        {:noreply, %{state|packet_buffer: incomplete_packet}}
      {:ok, packet, rest} ->
        state = %{state|packet_buffer: rest}

        # IO.puts " |-| packet:  #{inspect packet, limit: 50_000}"
        decode_result = Packet.decode_packet(packet, state)

        # IO.puts " |-| decoded: #{inspect decode_result}"
        case decode_result do
          {:error, reason} ->
            {:stop, {:error, reason}, state}
          {:ok, decoded_packet} ->
            case state.auth_state do
              :encrypted ->
                state = handle_packet(state, decoded_packet)
                next_packet(socket)
                {:noreply, state}
              _ ->
                handle_auth(self(), state, decoded_packet)
            end
        end
    end
  end
  def handle_info({:tcp_closed, socket}, state) do
    {:disconnect, :tcp_closed, state}
  end
  def handle_info(_request, state) do
    {:noreply, state}
  end

  defp handle_packet(state, packet) do
    IO.puts "\n\n ---- handle_packet #{inspect packet, limit: 100_000}\n"
    case packet do
      %TL.MTProto.Msg.Container{messages: messages} ->
        # calls `handle_packet#(state, %TL.MTProto.Message{...})` on each message
        Enum.reduce(messages, state, fn(message, state) ->
          handle_packet(state, message)
        end)
      %TL.MTProto.Message{msg_id: msg_id, body: body} ->
        handle_packet(state, body)
      %TL.MTProto.Msgs.Ack{msg_ids: msg_ids} ->
        %{state|msg_ids: state.msg_ids -- msg_ids}
      %TL.MTProto.New.Session.Created{server_salt: server_salt} ->
        %{state|server_salt: <<server_salt :: little-size(64)>>}
      %TL.MTProto.Rpc.Error{error_code: code, error_message: message} ->
        send(state.notifier, {:error, code, message})
        # TODO do we need to reconnect when server responds with error?
        # send(self, {:reconnect, :change_dc})
        state
      %TL.MTProto.Rpc.Result{req_msg_id: msg_id, result: result} ->
        state = handle_packet(state, result)
        # FIXME do we need to remove msg_id from current state in this place?
        %{state|msg_ids: state.msg_ids -- [msg_id]}
      %TL.MTProto.Gzip.Packed{packed_data: packed_data} ->
        {:ok, data} = TL.Serializer.decode(:zlib.gunzip(packed_data))
        handle_packet(state, data)
      # stores this_dc and dc list, changes when server fails
      # or returns Rpc.Error, or accidentally disconnected
      %TL.Config{dc_options: dc_options, this_dc: dc} ->
        Logger.debug("set dc options, current: #{inspect dc}, list: #{inspect dc_options}")
        %{state|dc_options: dc_options, dc: dc}
      result ->
        # IO.puts " --- handle_packet result: #{inspect result}"
        send(state.notifier, {:result, result})
        state
    end
  end

  defp handle_auth(client, state, packet) do
    case Auth.handle(client, state, packet) do
      {:ok, new_state} ->
        next_packet(state.socket)
        {:noreply, new_state}
      {:error, reason, new_state} ->
        {:stop, {:error, reason}, new_state}
    end
  end

  defp init_connection_request do
    %TL.InvokeWithLayer{
      layer: 57,
      query: %TL.InitConnection{
        api_id: config(:api_id),
        device_model: config(:device_model, "elixir"),
        system_version: config(:system_version, "0.1.0"),
        app_version: config(:app_version, "0.0.1"),
        lang_code: config(:lang_code, "en"),
        query: %TL.Help.GetConfig{}}}
  end

  def send_rpc_request(socket, request, state) do
    {packet, state} = Packet.encode(request, state)

    case :gen_tcp.send(socket, packet) do
      :ok ->
        {:ok, state}
      {:error, reason} ->
        {:error, reason, state}
    end
  end

  defp next_packet(socket) do
    :ok = :inet.setopts(socket, active: :once)
  end

  defp choose_server(%{dc: dc, dc_options: dc_options} = state) when length(dc_options) > 0 do
    acceptable_dcs = Enum.filter(dc_options, fn(dc_opt) ->
        dc_opt.ipv6 == false and dc_opt.id != dc
      end)

    new_dc = Enum.random(acceptable_dcs)

    {String.to_charlist(new_dc.ip_address), new_dc.port}
  end
  defp choose_server(_state) do
    {telegram_host(), telegram_port()}
  end

  defp telegram_host do
    config(:host)
  end

  defp telegram_port do
    config(:port)
  end

  defp config(key, default \\ nil) do
    Application.get_env(:mtproto, key, default)
  end
end