defmodule Bungee do @moduledoc false defmacro __using__(module: module, type: type) do quote do use Tesla, except: [:delete] adapter(:hackney) plug(Tesla.Middleware.BaseUrl, elasticsearch_uri()) plug(Tesla.Middleware.JSON) # plug(Tesla.Middleware.Query, [refresh: Application.get_env(:bungee, :wait_for_refresh, false)]) import Bungee @spec save(%unquote(module){}) :: {:ok, :created, %unquote(module){}} | {:ok, :updated, %unquote(module){}} @spec save(%unquote(module){}) :: %unquote(module){} @spec patch(String.t(), map()) :: :ok @spec fetch(String.t()) :: {:ok, %unquote(module){}} | {:error, :not_found} @spec fetch_by(atom(), String.t()) :: {:ok, pos_integer(), list(%unquote(module){})} | {:error, :not_found} @spec filter_value(atom() | String.t()) :: String.t() @spec filter_value(any()) :: any() @spec decode(%Tesla.Env{}) :: {:ok, %unquote(module){}} | {:ok, pos_integer(), %unquote(module){}} | {:error, :not_found} @spec decode_documents(list(), list()) :: list() @spec decode_document(map()) :: %unquote(module){} @spec elasticsearch_uri() :: String.t() @spec index_write() :: String.t() @spec index_read() :: String.t() @spec module() :: String.t() @spec type() :: atom() @spec get_read_uri() :: String.t() @spec get_read_uri(String.t()) :: String.t() def save(%unquote(module){} = projection) do created_or_updated = case :erlang.function_exported(unquote(module), :key, 1) do true -> key = apply(unquote(module), :key, [projection]) persist(key, projection) false -> persist(projection) end case created_or_updated do :ok -> {:ok, :updated, projection} :created -> {:ok, :created, projection} end end def save!(%unquote(module){} = projection) do {:ok, created_or_updated, updated_projection} = save(projection) updated_projection end def save(map) when is_map(map) do projection = unquote(module) |> struct(map) save(projection) end defp persist(projection) when is_map(projection) do full_write(nil, projection) end defp persist(identifier, projection) when is_binary(identifier) and is_map(projection) do full_write(identifier, projection) end defp full_write(nil, projection) do get_write_uri() |> post(projection) |> request_ok?() end defp full_write(identifier, projection) do identifier |> get_write_uri() |> put(projection) |> request_ok?() end defp partial_write(identifier, projection) do identifier |> get_write_uri() |> Kernel.<>("/_update") |> post(%{ "doc" => projection }) |> request_ok?() end def delete(identifier) do # Elasticsearch doesn't return a 202 for DELETE request( method: :delete, url: get_write_uri(identifier), headers: [{"Content-type", "application/json"}] ) |> request_ok?() end def fetch(identifier) do identifier |> get_read_uri() |> get() |> decode() rescue error in [ArgumentError] -> require Logger Logger.error(fn -> "Failed to decode: #{inspect(error)}" end) {:error, :unexpected_error} end def fetch_by(key, value) do request( method: :get, url: get_read_uri("_search"), headers: [{"Content-type", "application/json"}], body: ~s/{ "query": { "constant_score": { "filter": { "term": { #{filter_key(key)}: #{filter_value(value)} } } } } } / ) |> decode() rescue error in [ArgumentError] -> require Logger Logger.error(fn -> "Failed to decode: #{inspect(error)}" end) {:error, :unexpected_error} end defp filter_key(key) when is_atom(key) do ~s/"#{Atom.to_string(key)}"/ end defp filter_key(key) when is_binary(key) do ~s/"#{key}"/ end defp filter_value(value) when is_integer(value) do value end defp filter_value(value) when is_binary(value) do ~s/"#{String.downcase(value)}"/ end defp elasticsearch_uri() do Application.get_env(:bungee, :elasticsearch_uri, "http://localhost:9200") end defp index_write() do Application.get_env(:bungee, :index_write, index_read()) end defp index_read() do Application.get_env(:bungee, :index, nil) end defp module() do unquote(module) end defp type() do unquote(type) end defp request_ok?(response = %Tesla.Env{status: 200}) do require Logger Logger.info(fn -> "Bungee response HTTP 200" end) Logger.debug(fn -> "\tResponse: #{inspect(response)}" end) :ok end defp request_ok?(response = %Tesla.Env{status: 201}) do require Logger Logger.info(fn -> "Bungee response HTTP 201" end) Logger.debug(fn -> "\tResponse: #{inspect(response)}" end) :created end defp request_ok?(response = %Tesla.Env{status: 202}) do require Logger Logger.info(fn -> "Bungee response HTTP 202" end) Logger.debug(fn -> "\tResponse: #{inspect(response)}" end) :deleted end defp request_ok?(response = %Tesla.Env{status: status}) do require Logger Logger.info(fn -> "Bungee response HTTP #{status}" end) Logger.debug(fn -> "\tResponse: #{inspect(response)}" end) :error end defp decode(response = %Tesla.Env{status: 404}) do require Logger Logger.info(fn -> "Bungee response HTTP 404" end) Logger.debug(fn -> "\tResponse: #{inspect(response)}" end) {:error, :not_found} end defp decode(response = %Tesla.Env{status: 200, body: %{"hits" => %{"total" => 0}}}) do require Logger Logger.info(fn -> "Bungee response HTTP 200" end) Logger.debug(fn -> "\tResponse: #{inspect(response)}" end) {:error, :not_found} end defp decode( response = %Tesla.Env{ status: 200, body: %{"hits" => %{"hits" => documents, "total" => count}} } ) do require Logger Logger.info(fn -> "Bungee response HTTP 200" end) Logger.debug(fn -> "\tResponse: #{inspect(response)}" end) {:ok, count, decode_documents(documents, [])} end defp decode(response = %Tesla.Env{status: 200, body: body}) do require Logger Logger.info(fn -> "Bungee response HTTP 200" end) Logger.debug(fn -> "\tResponse: #{inspect(response)}" end) {:ok, decode_document(body)} end defp decode(unknown) do raise ArgumentError, message: "No idea what I'm decoding (Value: #{inspect(unknown)})" end defp decode_documents([head | tail], carry) do decode_documents(tail, carry ++ [decode_document(head)]) end defp decode_documents([], carry) do carry end defp decode_document(document = %{}) do document |> Map.get("_source") |> Poison.encode!() |> Poison.decode!(as: %unquote(module){}) end defp get_read_uri() do elasticsearch_uri() <> "/" <> index_read() <> "/" <> type() end defp get_read_uri(identifier) do get_read_uri() <> "/#{identifier}" end defp get_write_uri() do elasticsearch_uri() <> "/" <> index_write() <> "/" <> type() end defp get_write_uri(identifier) do get_write_uri() <> "/#{identifier}" end end end end