depo

defmodule Depo.DB do
  @moduledoc """
  A DB is a GenServer that manages an open database connection
  and a cache of prepared SQL statements. 
  """
  use GenServer

  defp init_with_db(db) do
    {:ok, %{
      # An open connection to the database.
      db: db,
      # An atom-keyed cache of named prepared SQL statements.
      cache: %{}
    }}
  end

  def terminate(_reason, %{db: db}) do
    :ok = :esqlite3.close(db)
  end

  def format_status(_, [_, state]) do
    %{cache: Map.keys(state.cache)}
  end

  # Get a stream of row tuples from the compiled statement.
  defp rows_stream(stmt) do
    Stream.unfold(stmt, fn stmt ->
      case :esqlite3.step(stmt) do
        {:row, row} -> {row, stmt}
        _ -> nil
      end
    end)
  end

  # Convert a tuple of row values and a list of keys to a map.
  defp row_to_map(row, keys) do
    columns = 0..(tuple_size(keys)-1)
    Enum.reduce(columns, %{}, fn(i, m) ->
      Map.put(m, elem(keys, i), elem(row, i))
    end)  
  end

  # Get a stream of results from a compiled SQL statement. 
  defp results_stream({:values, stmt}) do
    rows_stream(stmt)
  end
  defp results_stream({:single, stmt}) do
    # Get a stream of a single value from the compiled statement.
    Stream.unfold(stmt, fn
      nil -> nil
      stmt ->
        {:row, row} = :esqlite3.step(stmt)
        {elem(row, 0), nil}
    end)
  end
  defp results_stream(stmt) do
    keys = :esqlite3.column_names(stmt)
    # Convert the row stream into a map stream and return it.
    rows_stream(stmt)
    |> Stream.map(&(row_to_map(&1, keys)))
  end
  defp results_stream(stmt, values) when is_list(values) do
    :ok = :esqlite3.bind(stmt, values)
    results_stream(stmt)
  end
  defp results_stream(stmt, value) do
    results_stream(stmt, [value])
  end

  # Get a list of results from a compiled statement.
  defp get_results({:single, stmt}) do
    [single] = results_stream({:single, stmt})
    |> Enum.into([])
    single
  end
  defp get_results(stmt) do
    stmt
    |> results_stream()
    |> Enum.into([])
    |> Enum.reverse()
  end
  defp get_results({:single, stmt}, values) when is_list(values) do
    :ok = :esqlite3.bind(stmt, values)
    get_results({:single, stmt})
  end
  defp get_results({:single, stmt}, value) do
    get_results({:single, stmt}, [value])
  end
  defp get_results(stmt, values) when is_list(values) do
    :ok = :esqlite3.bind(stmt, values)
    get_results(stmt)
  end
  defp get_results(stmt, value) do
    get_results(stmt, [value])
  end

  # Stream the results from an isolated process.
  defp spawn_streamer(pid, stmt) do
    spawn_link(fn ->
      results_stream(stmt)
      |> Enum.each(fn result ->
        send(pid, {self(), result})
      end)
    end)
  end
  defp spawn_streamer(pid, stmt, values) when is_list(values) do
    :ok = :esqlite3.bind(stmt, values)
    spawn_streamer(pid, stmt)
  end
  defp spawn_streamer(pid, stmt, value) do
    spawn_streamer(pid, stmt, [value])
  end

  def init(:memory) do
    {:ok, db} = :esqlite3.open(':memory:')
    init_with_db(db)
  end
  def init(create: path) when is_binary(path) do
    if File.exists?(path) do
      {:stop, :file_already_exists}
    else
      {:ok, db} = path
      |> to_charlist() 
      |> :esqlite3.open()
      init_with_db(db)
    end
  end
  def init(path) when is_binary(path) do
    if File.exists?(path) do
      {:ok, db} = path 
      |> to_charlist() 
      |> :esqlite3.open()
      init_with_db(db)
    else
      {:stop, :no_such_file}
    end
  end

  def handle_cast({:write, sql}, state)
    when is_binary(sql) do
    :esqlite3.exec(to_charlist(sql), state.db)
    {:noreply, state}
  end

  def handle_cast({:write, cmd}, state)
    when is_atom(cmd) do
    if Map.has_key?(state.cache, cmd) do
      :"$done" = :esqlite3.step(state.cache[cmd])
    end
    {:noreply, state}
  end

  def handle_cast({:write, sql, values}, state)
    when is_binary(sql) do
    {:ok, stmt} = :esqlite3.prepare(to_charlist(sql), state.db)
    Stream.run(results_stream(stmt, values))
    {:noreply, state}
  end

  def handle_cast({:write, cmd, values}, state)
    when is_atom(cmd) do
    if Map.has_key?(state.cache, cmd) do
      stmt = state.cache[cmd]
      Stream.run(results_stream(stmt, values))
    end
    {:noreply, state}
  end

  def handle_cast({:teach, cmds}, state) when is_map(cmds) do
    # Prepare the statements and add them to the cache.
    cache = Enum.reduce(cmds, state.cache, fn {name, cmd}, cache -> 
      true = is_atom(name)
      Map.put(cache, name, prepare_cmd(state.db, cmd))
    end) |> Enum.into(%{})
    {:noreply, Map.put(state, :cache, cache)}
  end

  # Prepare (compile) a command with return options.
  defp prepare_cmd(db, {:values, sql}) do
    {:values, prepare_stmt(db, sql)}
  end
  defp prepare_cmd(db, {:single, sql}) do
    {:single, prepare_stmt(db, sql)}
  end
  defp prepare_cmd(db, sql) when is_binary(sql) do
    prepare_stmt(db, sql)
  end

  defp prepare_stmt(db, sql) do
    {:ok, stmt} = :esqlite3.prepare(to_charlist(sql), db)
    stmt
  end

  def handle_call({:read, sql}, _from, state)
    when is_binary(sql) do
    case :esqlite3.prepare(to_char_list(sql), state.db) do
      {:error, reason} ->
        {:reply, {:error, reason}, state}
      {:ok, stmt} ->  
        {:reply, get_results(stmt), state}
    end
  end

  def handle_call({:read, cmd}, _from, state)
    when is_atom(cmd) do
    if Map.has_key?(state.cache, cmd) do
      stmt = state.cache[cmd]
      {:reply, get_results(stmt), state}
    else
      {:reply, {:error, :unknown_command}, state}
    end
  end

  def handle_call({:read, sql, values}, _from, state)
    when is_binary(sql) do
    case :esqlite3.prepare(to_char_list(sql), state.db) do
      {:error, reason} ->
        {:reply, {:error, reason}, state}
      {:ok, stmt} ->
        {:reply, get_results(stmt, values), state}
    end
  end

  def handle_call({:read, query, values}, _from, state)
    when is_atom(query) do
    if Map.has_key?(state.cache, query) do
      stmt = state.cache[query]
      {:reply, get_results(stmt, values), state}
    else
      {:reply, {:error, :unknown_query}, state}
    end
  end

  def handle_call({:stream, pid, query}, _from, state)
    when is_binary(query) do
    case :esqlite3.prepare(to_char_list(query), state.db) do
      {:error, _} ->
        {:noreply, state}
      {:ok, stmt} ->
        {:reply, spawn_streamer(pid, stmt), state}
    end    
  end
  def handle_call({:stream, pid, query}, _from, state)
    when is_atom(query) do
    if Map.has_key?(state.cache, query) do
      stmt = state.cache[query]
      {:reply, spawn_streamer(pid, stmt), state}
    else
      {:reply, {:error, :unknown_query}, state}
    end
  end

  def handle_call({:stream, pid, query, values}, _from, state)
    when is_binary(query) do
    case :esqlite3.prepare(to_char_list(query), state.db) do
      {:error, _} ->
        {:noreply, state}
      {:ok, stmt} ->
        {:reply, spawn_streamer(pid, stmt, values), state}
    end    
  end
  def handle_call({:stream, pid, query, values}, _from, state)
    when is_atom(query) do
    if Map.has_key?(state.cache, query) do
      stmt = state.cache[query]
      {:reply, spawn_streamer(pid, stmt, values), state}
    else
      {:reply, {:error, :unknown_query}, state}
    end
  end

end