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 :

    - :peer_db
      
      List of
        { id, pid | nil, ip, port }

    - :shard_db
    
      List of
        { id, manifest, pid | nil }

    - :shard_state

      List of
        { id, state }

    - :shard_procs

      List of
        { {id, path}, pid }

    - :shard_peer_db

      Mult-list of
        { shard_id, peer_id }


    And an internal table :

    - :outbox
      
      Multi-list of
        { peer_id, message, time_inserted }

  """

  use GenServer

  require Logger

  @peer_db [Application.get_env(:shard, :data_path), "peer_db"] |> Path.join |> String.to_atom
  @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
  @shard_peer_db [Application.get_env(:shard, :data_path), "shard_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(@peer_db, [type: :set])
    for [{id, _pid, ip, port}] <- :dets.match @peer_db, :"$1" do
      :dets.insert @peer_db, {id, nil, ip, port}
      # connect blindly to everyone
      add_peer(ip, port)
    end

    :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(@shard_peer_db, [type: :bag])

    :ets.new(:shard_procs, [:set, :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_id, shards}, state) do
    for shard_id <- shards do
      case :dets.lookup(@shard_db, shard_id) do
        [{ ^shard_id, _, pid }] ->
          :dets.insert(@shard_peer_db, {shard_id, peer_id})
          GenServer.cast(pid, {:interested, peer_id})
        [] -> 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


  # ================
  # PUBLIC INTERFACE
  # ================
  

  @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"""
  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"""
  Dispatch incoming message to correct shard process
  """
  def dispatch(peer_id, {shard_id, path, msg}) do
    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

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

  def dispatch(peer_id, {:not_interested, shard}) do
    GenServer.cast(__MODULE__, {:not_interested, peer_id, shard})
  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"""
  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"""
  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 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

  @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
end