exkad

defmodule Exkad.Node do
  use GenServer
  use Bitwise
  import Supervisor.Spec

  defmodule Descriptor do
    defstruct id: :empty, loc: :empty
  end

  alias Descriptor, as: ND

  @alpha 3
  @k 10
  @request_timeout 5000

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts)
  end

  def init(opts) do

    state = opts
      |> init_desc
      |> init_tx
      |> init_kbuckets
      |> add_seed
      |> populate_kbuckets
      |> start_store
    {:ok, state}
  end


  defp init_tx(%{tx: _} = state), do: state

  defp init_tx(_) do
    raise ArgumentError, message: "You need to provide a tx module"
  end

  defp bind_to(:self), do: self
  defp bind_to(address), do: address


  defp init_desc(state) do
    id = state[:id]
    bind = bind_to(state[:bind])
    state = Dict.drop(state, [:id, :bind])

    Enum.into(state, %{desc: %ND{id: id, loc: bind}})
  end

  defp init_kbuckets(state = %{desc: %ND{id: id}}) do
    buckets = id
      |> id_to_bin
      |> Enum.map(fn _ -> [] end)
    Enum.into(state, %{buckets: buckets})
  end

  defp start_store(%{desc: %ND{id: id}} = state) do

    children = [
      worker(Exkad.Store, [[]]),
    ]

    {:ok, sup} = Supervisor.start_link(children, strategy: :one_for_one)
    
    {_, store, _, _} = Supervisor.which_children(sup)  
      |> Enum.find(fn({module, _, _, _}) -> module == Exkad.Store end)

    Enum.into(%{sup: sup, store: store}, state)
  end




  defp add_seed(state = %{seed: nil}), do: state

  defp add_seed(state = %{seed: seed}) do
    add_to_bucket(state, seed)
  end
  ##
  # There's no seed, so we're the first in the network
  defp add_seed(state), do: state


  defp populate_kbuckets(state = %{desc: %ND{id: this_id}}) do
    # Ask the seed for our id, which will give back a bunch of stuff to populate 
    {_, state} = find_k(this_id, state)
    state
  end

  ##
  # There's no seed, first in the network..
  defp populate_kbuckets(state), do: state



  defp unpack(<<a::1, b::1, c::1, d::1, e::1, f::1, g::1, h::1>>) do
    [a, b, c, d, e, f, g, h]
  end


  def id_to_bin(loc) do
    loc
      |> :erlang.term_to_binary
      |> :erlang.binary_to_list
      |> Enum.map(fn num -> 
        unpack(<<num :: size(8)>>)
      end)
      |> List.flatten
  end



  def prefix_length([a | this_rest], [a | other_rest], acc) do
    prefix_length(this_rest, other_rest, acc + 1)
  end

  def prefix_length(_, _, acc), do: acc
  
  def prefix_length(this, other) do
    prefix_length(id_to_bin(this), id_to_bin(other), 0)
  end

  ##
  # If you try to add self to bucket, it's a noop
  def add_to_bucket(state = %{desc: %ND{id: id}}, %ND{id: id}) do
    state
  end

  #
  # Add a single descriptor to the local buckets
  #
  def add_to_bucket(state = %{desc: desc, buckets: buckets}, add_desc) do
    %ND{id: id} = desc
    %ND{id: add_id} = add_desc

    index = prefix_length(id, add_id)
    bucket = [add_desc | Enum.at(buckets, index)] |> Enum.uniq |> Enum.take(@k)
    buckets = List.replace_at(buckets, index, bucket)
    Enum.into(%{buckets: buckets}, state)
  end

  #
  # Add all the nodes in `nodes` to local bucket
  #
  defp add_all_to_buckets(nodes, state) do
    Enum.reduce(nodes, state, fn(desc, acc) -> 
      add_to_bucket(acc, desc) 
    end)
  end

  #
  # xor distance between from_id and to_id
  # 
  def distance_from(from_id, to_id) do
    Enum.zip(id_to_bin(from_id), id_to_bin(to_id)) 
      |> Enum.map(fn {f, t} -> f ^^^ t end)
  end

  #
  # Get the k_closest_nodes to `to_id` in local buckets
  #
  def k_closest_nodes_to(nodes, to_id) do
    nodes
      |> Enum.uniq
      |> Enum.map(fn %ND{id: id, loc: loc} = nd -> {distance_from(id, to_id), nd} end)
      |> Enum.sort(fn {a, _}, {b, _} -> a <= b end)
      |> Enum.take(@k)
      |> Enum.map(fn {_distance, desc} -> desc end)
  end

  def k_closest_nodes(to_id, %{buckets: buckets}) do
    List.flatten(buckets) |> k_closest_nodes_to(to_id)
  end



  # return true if `closer_nodes` contains any nodes
  # that are closer to `to_find_id` than `nodes`
  defp are_closer?(closer_nodes, nodes, to_find_id) do
    Enum.any?(closer_nodes, fn %ND{id: cnode} -> 
      closer_dist = distance_from(cnode, to_find_id)
      Enum.any?(nodes, fn %ND{id: node} -> 
        closer_dist < distance_from(node, to_find_id)
      end)
    end)
  end


  #
  # Generic fan out forward to nodes
  # runs the to_do/1 function on each node
  # and waits for all nodes to return a result, 
  # giving back a list of results
  defp forward(nodes, to_do) do
    nodes
      |> Enum.map(fn(node) ->
        {node, Task.async(fn -> to_do.(node) end)}
      end)
      |> Enum.map(fn {_, task} ->
        Task.await(task, @request_timeout)
      end)
  end

  # Helper for find_k/2
  defp find_k(to_find_id, %{desc: desc, tx: tx} = state, nodes, seen) do
    closer = nodes
      |> forward(fn node -> 
        {:ok, node} = tx.find(node, to_find_id, desc) 
        node
      end)
      |> List.flatten
      |> Enum.uniq
      |> Enum.reject(fn node -> node in seen or node == desc end)
      |> k_closest_nodes_to(to_find_id)

    state = add_all_to_buckets(closer, state)

    if are_closer?(closer, nodes, to_find_id) do
        find_k(to_find_id, state, closer, seen ++ nodes)
    else
      nodes = k_closest_nodes_to(seen ++ nodes ++ closer, to_find_id)
      {nodes, state}
    end
  end

  #
  # Recursively find the k nodes that are closest to `to_find_id`
  #
  defp find_k(to_find_id, state) do
    find_k(to_find_id, state, k_closest_nodes(to_find_id, state), [])
  end


  defp extract_value([]), do: {:error, :not_found}
  defp extract_value([{:value, thing} | _]), do: {:value, thing}
  defp extract_value([first | rest]), do: extract_value(rest)



  defp fan_out(nodes, fwd_fun, key, %{tx: tx}) do
    forward(nodes, fn node -> fwd_fun.(tx, node, key) end)
      |> List.flatten
      |> Enum.uniq
  end



  #
  # Forward a find request to `nodes` for key
  #
  defp get_from_many(nodes, key, state) do
    fan_out(nodes, fn(tx, node, key) -> tx.get(node, key) end, key, state)
  end

  defp search_from_many(nodes, key, state) do
    fan_out(nodes, fn(tx, node, key) -> tx.search(node, key) end, key, state)
  end

  defp is_search_result?({:ok, results}) when is_list(results), do: True
  defp is_search_result?(_), do: False

  defp flatten_search(results) do
    Enum.map(results, fn {:ok, res} -> res end) |> List.flatten
  end

  defp rsearch(key, %{desc: desc} = state, nodes, seen, results) do
    {search_results, closer} = search_from_many(nodes, key, state) 
      |> Enum.partition(fn item -> is_search_result?(item) end)

    closer = closer
      |> Enum.reject(fn node -> node in seen or node == desc end)
      |> k_closest_nodes_to(key)

    results = results ++ flatten_search(search_results)

    if are_closer?(closer, nodes, key) do
      rsearch(key, state, closer, seen ++ nodes, results)
    else
      closer_results = search_from_many(closer, key, state) 
        |> Enum.filter(fn res -> is_search_result?(res) end)
        |> flatten_search

      closer_results ++ results
    end

  end


  #
  # Recursive find of a key
  #
  defp rfind_val(key, %{desc: desc} = state, nodes, seen) do

    closer = get_from_many(nodes, key, state)
    case extract_value(closer) do
      {:error, :not_found} -> 
        state = add_all_to_buckets(closer, state)
        closer = closer
          |> Enum.reject(fn node -> node in seen or node == desc end)
          |> k_closest_nodes_to(key)

        if are_closer?(closer, nodes, key) do
          rfind_val(key, state, closer, seen ++ nodes)
        else
          val = get_from_many(closer, key, state) |> extract_value
          {state, val}
        end
      value -> {state, value}
    end
  end



  defp rfind_val(key, state) do
    case value_or_closest(key, state) do
      {:value, value} -> {state, {:value, value}}
      k_closest -> rfind_val(key, state, k_closest, [])
    end
  end 

  defp rsearch(key, state) do
    nodes = k_closest_nodes(key, state)
    rsearch(key, state, nodes, [], [])
  end

  
  defp value_or_closest(key, %{store: store} = state) do
    case Exkad.Store.get(store, key) do
      {:error, :not_found} -> k_closest_nodes(key, state)
      value -> {:value, value}
    end
  end

  #
  # Replicate a keyval pair to k nodes
  #
  defp put_to_many(key, value, %{tx: tx} = state) do
    {nodes, _state} = find_k(key, state)
    forward(nodes, fn(node) -> tx.put(node, value) end)
  end

  #
  # Replicate metadata to k nodes
  #
  defp put_meta_to_many(meta_key, meta_term, ptr, %{tx: tx} = state) do
    {nodes, _state} = find_k(meta_key, state)
    forward(nodes, fn(node) -> tx.put_meta(node, meta_term, ptr) end)
  end



  def handle_call({:ping, from_desc}, _from, state) do
    state = add_to_bucket(state, from_desc)
    {:reply, :ok, state}
  end


  def handle_call({:make_ping, to_desc}, _from, %{tx: tx, desc: from} = state) do
    res = tx.ping(to_desc, from)
    # state = add_to_bucket(state, from_desc)
    {:reply, res, state}
  end


  def handle_call(:dump, _from, state) do
    {:reply, state, state}
  end

  ##
  # This call is cheating, it will not be implemented over http, just erlang messaging
  # to make tests easier
  def handle_call(:describe, _from, %{desc: desc} = state) do
    {:reply, {:ok, desc}, state}
  end

  def handle_call({:find, to_find_id, from_desc}, _from, state) do
    closest = k_closest_nodes(to_find_id, state)
    state = add_to_bucket(state, from_desc)
    {:reply, {:ok, closest}, state}
  end

  def handle_call({:put, value}, _from, %{store: store} = state) do
    result = Exkad.Store.put(store, value)    
    {:reply, result, state}
  end


  def handle_call({:put_meta, term, ptr}, _from, %{store: store} = state) do
    result = Exkad.Store.put_meta(store, term, ptr)    
    {:reply, result, state}
  end

  def handle_call({:search, key}, _from, %{store: store} = state) do
    result = Exkad.Store.search(store, key)
    {:reply, result, state}
  end

  def handle_call({:rsearch, term}, _from, state) do
    key = Exkad.Store.hash(term)
    result = rsearch(key, state)
    {:reply, result, state}
  end

  def handle_call({:rput, value, meta}, _from, %{store: store} = state) do
    result = Exkad.Store.put(store, value)

    {:ok, key} = result
    put_to_many(key, value, state)

    Enum.map(meta, fn {meta_name, meta_term} -> 

      {:ok, meta_key} = Exkad.Store.put_meta(store, meta_term, key)
      IO.puts "Put meta #{meta_name} : #{meta_term} | #{meta_key} --> #{key}"
      put_meta_to_many(meta_key, meta_term, key, state)
    end)
    
    {:reply, result, state}
  end

  def handle_call({:get, key}, _from, state) do
    result = value_or_closest(key, state)
    {:reply, result, state}
  end

  def handle_call({:rget, key}, _from, state) do
    {state, value} = rfind_val(key, state)
    {:reply, value, state}
  end


end