path: root/shard/lib/manager.ex

defprotocol Shard.Manifest do
  @moduledoc"""
  A shard manifest is a data structure that uniquely defines the identity of the shard.

  The hash of the manifest is the unique identifier of that shard on the network.
  
  The Manifest protocol is a protocol implemented by the manifest structs for the
  different shard types. It contains an operation start() that is able to launch the
  correct process for this shard and connect to other peers that use it.
  """

  @doc"""
  Start the corresponding Shard process
  """
  def start(manifest)
end

defmodule Shard.Manager do
  @moduledoc"""
    Maintains several important tables :

    - :shard_db (persistent with DETS)
    
      List of
        { id, manifest, pid | nil }

    - :shard_state (persistent with DETS)

      List of
        { id, state }

    - :peer_db (persistent with DETS)

      Mult-list of
        { shard_id, peer_info }   # TODO: add health info (last seen, ping, etc)

      peer_info := {:inet4, ip, port} | {:inet6, ip, port} | {:onion, name}

    - :shard_procs (not persistent)

      List of
        { {id, path}, pid }

  - :connections (not persistent)

      List of
        { nil | his_pk, nil | my_pk, pid, peer_info }

    And an internal table :

    - :outbox (not persistent)
      
      Multi-list of
        { dest, auth_info, message, time_inserted }

      dest := peer_info
      auth_info := nil | { his_pk, my_pk_list }

  """

  use GenServer

  require Logger

  @shard_db [Application.get_env(:shard, :data_path), "shard_db"] |> Path.join |> String.to_atom
  @shard_state [Application.get_env(:shard, :data_path), "shard_state"] |> Path.join |> String.to_atom
  @peer_db [Application.get_env(:shard, :data_path), "peer_db"] |> Path.join |> String.to_atom

  def start_link(my_port) do
    GenServer.start_link(__MODULE__, my_port, name: __MODULE__)
  end

  def init(my_port) do
    :dets.open_file(@shard_db, [type: :set])
    for [{id, manifest, _pid}] <- :dets.match @shard_db, :"$1" do
      :dets.insert @shard_db, {id, manifest, nil}
      spawn fn -> Shard.Manifest.start manifest end
    end

    :dets.open_file(@shard_state, [type: :set])
    :dets.open_file(@peer_db, [type: :bag])

    :ets.new(:shard_procs, [:set, :protected, :named_table])
    :ets.new(:connections, [:bag, :protected, :named_table])
    outbox = :ets.new(:outbox, [:bag, :private])

    {:ok, %{my_port: my_port, outbox: outbox} }
  end

  def handle_call({:register, shard_id, manifest, pid}, _from, state) do
    will_live = case :dets.lookup(@shard_db, shard_id) do
      [{ ^shard_id, _, pid }] when pid != nil -> not Process.alive?(pid)
      _ -> true
    end
    reply = if will_live do
      Process.monitor(pid)
      :dets.insert(@shard_db, {shard_id, manifest, pid})
      :ok
    else
      :redundant
    end
    {:reply, reply, state}
  end

  def handle_cast({:dispatch_to, shard_id, path, pid}, state) do
    :ets.insert(:shard_procs, { {shard_id, path}, pid })
    Process.monitor(pid)
    {:noreply, state}
  end

  def handle_cast({:interested, peer_info, shards}, state) do
    for shard_id <- shards do
      case :dets.lookup(@shard_db, shard_id) do
        [{ ^shard_id, _, pid }] ->
          :dets.insert(@peer_db, {shard_id, peer_info})
          GenServer.cast(pid, {:interested, peer_info})
        [] -> nil
      end
    end
    {:noreply, state}
  end

  def handle_cast({:not_interested, peer_id, shard_id}, state) do
    :dets.match_delete(@shard_peer_db, {shard_id, peer_id})
    {:noreply, state}
  end

  def handle_cast({:shard_peer_db_insert, shard_id, peer_id}, state) do
    :dets.insert(@shard_peer_db, {shard_id, peer_id})
    {:noreply, state}
  end

  def handle_cast({:peer_up, pk, pid, ip, port}, state) do
    for [pk2] <- :dets.match(@peer_db, {:'$1', :_, ip, port}) do
      if pk2 != pk do
        # obsolete peer information
        :dets.delete(@peer_db, pk2)
        :dets.match_delete(@shard_peer_db, {:_, pk2})
      end
    end
    :dets.insert(@peer_db, {pk, pid, ip, port})

    # Send interested message for all our shards
    id_list = (for [{id, _, _}] <- :dets.match(@shard_db, :"$1"), do: id)
    GenServer.cast(pid, {:send_msg, {:interested, id_list}})

    # Send queued messages
    for {_, msg, _} <- :ets.lookup(state.outbox, pk) do
      GenServer.cast(pid, {:send_msg, msg})
    end
    :ets.delete(state.outbox, pk)

    {:noreply, state}
  end

  def handle_cast({:peer_down, pk, ip, port}, state) do
    :dets.insert(@peer_db, {pk, nil, ip, port})
    {:noreply, state}
  end

  def handle_cast({:connect_and_send, peer_id, msg}, state) do
    case :dets.lookup(@peer_db, peer_id) do
      [{^peer_id, nil, ip, port}] ->
        add_peer(ip, port, state)
        currtime = System.os_time :second
        :ets.insert(state.outbox, {peer_id, msg, currtime})
        outbox_cleanup = [ {{:_, :_, :'$1'}, [{:<, :'$1', currtime - 60}], [true]} ]
        :ets.select_delete(state.outbox, outbox_cleanup)
      _ ->
        Logger.info "Dropping message #{inspect msg} for peer #{inspect peer_id}: peer not in database"
    end
    {:noreply, state}
  end

  def handle_cast({:try_connect, pk_list}, state) do
    for pk <- pk_list do
      case :dets.lookup(@peer_db, pk) do
        [{^pk, nil, ip, port}] ->
          add_peer(ip, port, state)
        _ -> nil
      end
    end
    {:noreply, state}
  end

  def handle_cast({:add_peer, ip, port}, state) do
    add_peer(ip, port, state)
    {:noreply, state}
  end

  def handle_info({:DOWN, _, :process, pid, _}, state) do
    :ets.match_delete(:shard_procs, {:_, pid})
    {:noreply, state}
  end

  defp add_peer(ip, port, state) do
    spawn fn ->
      case :gen_tcp.connect(ip, port, [:binary, packet: 2, active: false]) do
        {:ok, client} ->
          {:ok, pid} = DynamicSupervisor.start_child(Shard.DynamicSupervisor, {SNet.TCPConn, %{socket: client, my_port: state.my_port}})
          :ok = :gen_tcp.controlling_process(client, pid)
        _ ->
          Logger.info "Could not connect to #{inspect ip}:#{port}, some messages may be dropped"
        end
    end
  end


  # ====================
  # INTERFACE WITH PEERS
  # ====================

  def incoming(conn_pid, {:interested, shards}) do
    GenServer.cast(__MODULE__, {:interested, peer_id, shards})
  end

  def incoming(conn_pid, {:not_interested, shard}) do
    GenServer.cast(__MODULE__, {:not_interested, peer_id, shard})
  end

  @doc"""
  Dispatch incoming message to correct shard process
  """
  defp dispatch(conn_pid, {shard_id, path, msg}) do
    # TODO: auth
    case :dets.lookup(@shard_db, shard_id) do
      [] ->
        __MODULE__.send(peer_id, {:not_interested, shard_id})
      [_] ->
        case :dets.match(@shard_peer_db, {shard_id, peer_id}) do
          [] ->
            GenServer.cast(__MODULE__, {:shard_peer_db_insert, shard_id, peer_id})
          _ -> nil
        end
        case :ets.lookup(:shard_procs, {shard_id, path}) do
          [{ {^shard_id, ^path}, pid }] ->
            GenServer.cast(pid, {:msg, peer_id, shard_id, path, msg})
          [] ->
            Logger.info("Warning: dropping message for #{inspect shard_id}/#{inspect path}, no handler running.\n\t#{inspect msg}")
        end
    end
  end


  # =====================
  # INTERFACE WITH SHARDS
  # =====================

  @doc"""
  Send message to a peer specified by peer id
  """
  def send(peer_id, msg) do
    case :dets.lookup(@peer_db, peer_id) do
      [{ ^peer_id, pid, _, _}] when pid != nil->
        GenServer.cast(pid, {:send_msg, msg})
      _ -> 
        GenServer.cast(__MODULE__, {:connect_and_send, peer_id, msg})
    end
  end

  @doc"""
  Send message to a peer through an authenticated channel

  his_auth: accepted users to talk to, either single pk or list of pk

  Returns true if a corresponding channel was open and msg was sent,
  false otherwise.
  """
  def send(peer_id, my_auth, his_auth, msg) do
    # TODO
  end

  @doc"""
  Register a process as the main process for a shard.

  Returns either :ok or :redundant, in which case the process must exit.
  """
  def register(shard_id, manifest, pid) do
    GenServer.call(__MODULE__, {:register, shard_id, manifest, pid})
  end

  @doc"""
  Register a process as the handler for shard packets for a given path.
  """
  def dispatch_to(shard_id, path, pid) do
    GenServer.cast(__MODULE__, {:dispatch_to, shard_id, path, pid})
  end

  @doc"""
  Return the list of all peer IDs that are interested in a certain shard
  """
  def get_shard_peers(shard_id) do
    for [x] <- :dets.match(@shard_peer_db, {shard_id, :"$1"}), do: x
  end

  @doc"""
  Return the saved state value for a shard
  """
  def load_state(shard_id) do
    case :dets.lookup(@shard_state, shard_id) do
      [{^shard_id, state}] -> state
      _ -> nil
    end
  end

  @doc"""
  Save a state value for a shard
  """
  def save_state(shard_id, state) do
    :dets.insert(@shard_state, {shard_id, state})
  end


  # ==========================
  # INTERFACE FOR OTHER THINGS
  # ==========================

  @doc"""
  Connect to a peer specified by ip address and port
  """
  def add_peer(ip, port) do
    GenServer.cast(__MODULE__, {:add_peer, ip, port})
  end

  @doc"""
  Returns the pid for a shard if it exists
  """
  def find_proc(shard_id) do
    case :dets.lookup(@shard_db, shard_id) do
      [{^shard_id, _, pid}] -> pid
      _ -> nil
    end
  end

  @doc"""
  Return the list of all shards.
  """
  def list_shards() do
    for [x] <- :dets.match(@shard_db, :"$1"), do: x
  end

  @doc"""
  Return the list of all peers
  """
  def list_peers() do
    for [x] <- :dets.match(@peer_db, :"$1"), do: x
  end
end