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

  def init(:memory) do
    {:ok, db} = :esqlite3.open(':memory:')
    {:ok, %{
      # An open connection to the database.
      db: db,
      # An atom-keyed cache of named prepared SQL statements.
      cache: %{}
    }}
  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()
      {:ok, %{
        # An open connection to the database.
        db: db,
      }}
    end
  end
  def init(path) when is_binary(path) do
    if File.exists?(path) do
      {:ok, db} = path 
      |> to_charlist() 
      |> :esqlite3.open()
      {:ok, %{
        # An open connection to the database.
        db: db,
      }}
    else
      {:stop, :no_such_file}
    end
  end

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

  @doc """
  Bind the values to the statement, execute it,
  and return a stream of the row values.
  """
  def bind_stream(stmt, values) do
    :ok = :esqlite3.bind(stmt, values)
    Stream.unfold(stmt, fn stmt ->
      case :esqlite3.step(stmt) do
        {:row, row} -> {row, stmt}
        _ -> nil
      end
    end)
  end

  # Bind the values to variables, then run the query.
  defp bind_query(stmt, values) do
    keys = :esqlite3.column_names(stmt)
    # Convert the rows into maps with column names as keys.
    bind_stream(stmt, values)
    |> Stream.map(fn row ->
        Enum.reduce(0..(tuple_size(keys)-1), %{}, 
          fn(i, m) ->
            Map.put(m, elem(keys, i), elem(row, i))
          end)
      end)
    |> Enum.into([])
  end

  # Get a stream of results from a compiled SQL statement. 
  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

  # 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 list of results from a compiled statement.
  defp results_list(stmt) do
    stmt
    |> results_stream()
    |> Enum.into([])
    |> Enum.reverse()
  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(bind_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(bind_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.
    state = Map.update!(state, :cache, fn cache ->
      Enum.reduce(cmds, cache, fn {name, sql}, cache -> 
        true = is_atom(name) && is_binary(sql)
        {:ok, stmt} = :esqlite3.prepare(to_charlist(sql), state.db)
        Map.put(cache, name, stmt)
      end) |> Enum.into(%{})
    end)
    {:noreply, state}
  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, results_list(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, results_list(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, bind_query(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, bind_query(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} ->
        :ok = :esqlite3.bind(stmt, values)
        {:reply, spawn_streamer(pid, stmt), 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]
      :ok = :esqlite3.bind(stmt, values)
      {:reply, spawn_streamer(pid, stmt), state}
    else
      {:reply, {:error, :unknown_query}, state}
    end
  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
end