shortner

cache.ex

defmodule Shortener.LinkManager.Cache do
  @moduledoc false
  use GenServer

  @table_name :my_cache
  
  def start_link(args) do
    GenServer.start_link(__MODULE__, args, name: __MODULE__)
  end

  def lookup(cache \\ __MODULE__, key) do
    try do
      GenServer.call(cache, {:lookup, key})
    catch
      :exit, {{:nodedown, _}, _} -> {:error, :node_down}
    end
  end

  def insert(cache \\ __MODULE__, key, value) do
    try do
      GenServer.call(cache, {:insert, key, value})
    catch
      :exit, {{:nodedown, _}, _} -> {:error, :node_down}
    end
  end

  def broadcast_insert(cache \\ __MODULE__, key, value) do
    GenServer.abcast(Node.list(), cache, {:insert, key, value})
  end

  def flush(cache \\ __MODULE__) do
    GenServer.call(cache, :flush)
  end

  def init(args) do
    # TODO - Replace nil with real table
    :ets.new(@table_name, [:named_table, :set, :public])
    {:ok, %{table: @table_name}}
  end

  def handle_cast({:insert, key, value}, data) do
#    IO.inspect({node(), "Cache.insert", key, value})
    :ets.insert(@table_name, {key, value})
    {:noreply, data}
  end
  
  def handle_call({:lookup, key}, _from, data) do
#    IO.inspect({node(), "Cache.lookup", key}, label: "Cache.lookup")
    value =
        case :ets.lookup(@table_name, key) do 
          [{^key, value}] -> {:ok, value}
          [] -> {:error, :not_found}
        end

    {:reply, value, data}
  end

  def handle_call({:insert, key, value}, _from, data) do
#    IO.inspect({node(), "Cache.insert", key, value})
    :ets.insert(@table_name, {key, value})
    {:reply, :ok, data}
  end

  def handle_call(:flush, _from, data) do
    :ets.delete_all_objects(data.table)
    {:reply, :ok, data}
  end
end

cluster.ex

defmodule Shortener.Cluster do
  @moduledoc """
  This module provides an interface for updating clusters as well as a
  supervision tree for starting and stopping node discovery.
  """

  alias Shortener.Storage

  alias ExHashRing.HashRing

  @ring_key {__MODULE__, :hash_ring}

  def child_spec(_args) do
    children = [
      {Cluster.Supervisor, [topology(), [name: Shortener.ClusterSupervisor]]},
    ]

    %{
      id: __MODULE__,
      type: :supervisor,
      start: {Supervisor, :start_link, [children, [strategy: :one_for_one]]}
    }
  end

  def find_node(key) do
    # TODO - Update with hash ring lookup
    ring = :persistent_term.get(@ring_key)
    HashRing.find_node(ring, key) |> IO.inspect(label: "find_node")
  end

  # Sets the canonical set of nodes into persistent storage.
  def set_canonical_nodes(nodes) do
    bin = :erlang.term_to_binary(nodes)
    :ok = Storage.set("shortener:cluster", bin)
  end

  def update_ring do
    # TODO - Fetch nodes from persistent store, update hash ring
    # put the hash ring into persistent term storage.
    {:ok, nodes} = Storage.get("shortener:cluster")
    nodes = :erlang.binary_to_term(nodes)

    ring = nodes
    |> Enum.reduce(HashRing.new, fn n, ring ->
        {:ok, new_ring} = HashRing.add_node(ring, n)
        new_ring
      end
    )
    :persistent_term.put(@ring_key, ring)
    :ok
  end

  defp topology do
    [
      shortener: [
        strategy: Cluster.Strategy.Gossip,
      ]
    ]
  end
end

link_manager.ex

defmodule Shortener.LinkManager do
  @moduledoc """
  Manages the lifecycles of links
  """

  alias Shortener.Storage
  alias Shortener.LinkManager.Cache
  alias Shortener.Cluster

  @lookup_sup __MODULE__.LookupSupervisor

  def child_spec(_args) do
    children = [
      Cache,
      # TODO - Extend this supervision tree to support remote lookups
    ]

    %{
      id: __MODULE__,
      type: :supervisor,
      start: {Supervisor, :start_link, [children, [strategy: :one_for_one]]}
    }
  end

  def create(url) do
    short_code = generate_short_code(url)
    node = Cluster.find_node(short_code)

    res = Cache.insert({Cache, node}, short_code, url)
    
    case res do
      {:error, :node_down} -> {:error, :node_down}
      :ok -> 
        Storage.set(short_code, url)
        {:ok, short_code}
    end # |> IO.inspect(label: "link_manager.create")
  end

  def lookup(short_code) do
    with {:error, :not_found} <- Cache.lookup(short_code),
      {:ok, url} <- Storage.get(short_code)
    do
      Cache.insert(short_code, url)
      {:ok, url}
    end
  end

  def remote_lookup(short_code) do
    node = Cluster.find_node(short_code) # |> IO.inspect(label: "remote_lookup node")
#    Cache.lookup({Cache, node}, short_code)

    with {:error, :not_found} <- Cache.lookup({Cache, node}, short_code),
      {:ok, url} <- Storage.get(short_code)
    do
      Cache.insert({Cache, node}, short_code, url)
      {:ok, url}
    else
      {:error, :node_down} -> Storage.get(short_code)
    end
  end

  def generate_short_code(url) do
    url
    |> hash
    |> Base.encode16(case: :lower)
    |> String.to_integer(16)
    |> pack_bitstring
    |> Base.url_encode64
    |> String.replace(~r/==\n?/, "")
  end

  defp hash(str), do: :crypto.hash(:sha256, str)

  defp pack_bitstring(int), do: << int :: big-unsigned-32 >>
end