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