carbonite

# SPDX-License-Identifier: Apache-2.0

defmodule Carbonite do
  @readme Path.join([__DIR__, "../README.md"])
  @external_resource @readme

  @moduledoc @readme
             |> File.read!()
             |> String.split("<!-- MDOC -->")
             |> Enum.drop(1)
             |> Enum.take_every(2)
             |> Enum.join("\n")

  @moduledoc since: "0.1.0"

  import Ecto.Query
  alias Carbonite.{Outbox, Prefix, Query, Schema, Transaction, Trigger}
  require Prefix
  require Schema

  @type prefix :: binary()
  @type repo :: Ecto.Repo.t()

  @type prefix_option :: {:carbonite_prefix, prefix()}

  @doc "Returns the default audit trail prefix."
  @doc since: "0.1.0"
  @spec default_prefix() :: prefix()
  def default_prefix, do: Prefix.default_prefix()

  @doc """
  Inserts a `t:Carbonite.Transaction.t/0` into the database.

  Make sure to run this within a transaction.

  ## Parameters

  * `repo` - the Ecto repository
  * `params` - map of params for the `Carbonite.Transaction` (e.g., `:meta`)
  * `opts` - optional keyword list

  ## Options

  * `carbonite_prefix` - defines the audit trail's schema, defaults to `"carbonite_default"`

  ## Multiple inserts in the same transaction

  Normally, you should have exactly one `insert_transaction/3` call per database transaction. In
  practise, there are two scenarios in this function may be called multiple times:

  1. If an operation A, which calls `insert_transaction/3`, sometimes is nested within an outer
     operation B, which also calls `insert_transaction/3`.
  2. In tests using Ecto's SQL sandbox, subsequent calls to transactional operations (even to the
     same operation twice) are wrapped inside the overarching test transaction, and hence also
     effectively call `insert_transaction/3` within the same transaction.

  While the first scenario can be resolved using appropriate control flow (e.g. by conditionally
  disabling the inner `insert_transaction/3` call), the second scenario is quite common and often
  unavoidable.

  Therefore, `insert_transaction/3` **ignores** subsequent calls within the same database
  transaction (equivalent to `ON CONFLICT DO NOTHING`), **discarding metadata** passed to all
  calls but the first.
  """
  @doc since: "0.4.0"
  @spec insert_transaction(repo()) :: {:ok, Transaction.t()} | {:error, Ecto.Changeset.t()}
  @spec insert_transaction(repo(), params :: map()) ::
          {:ok, Transaction.t()} | {:error, Ecto.Changeset.t()}
  @spec insert_transaction(repo(), params :: map(), [prefix_option()]) ::
          {:ok, Transaction.t()} | {:error, Ecto.Changeset.t()}
  def insert_transaction(repo, params \\ %{}, opts \\ []) do
    carbonite_prefix = Keyword.get(opts, :carbonite_prefix, default_prefix())

    # NOTE: ON CONFLICT DO NOTHING does not combine with RETURNING, so we're forcing an UPDATE.

    params
    |> Transaction.changeset()
    |> repo.insert(
      prefix: carbonite_prefix,
      on_conflict: {:replace, [:xact_id]},
      conflict_target: [:id],
      returning: true
    )
  end

  @doc """
  Deletes the current `t:Carbonite.Transaction.t/0` if no changes have been recorded.

  This is sometimes useful to avoid "orphaned" transactions (without change records)
  for operations that usually modify any tracked tables but in rare cases do nothing.

  As the INSERT and DELETE on the `transactions` table has a performance cost, it is
  usually preferable to skip the transaction entirely, if possible.

  ## Example

      MyApp.Repo.transaction(fn ->
        Carbonite.insert_transaction(MyApp.Repo)
        do_something_that_may_or_may_not_cause_a_change_to_be_recorded()
        Carbonite.delete_transaction_if_empty(MyApp.Repo)
      end)

  ## Parameters

  * `repo` - the Ecto repository
  * `opts` - optional keyword list

  ## Options

  * `carbonite_prefix` - defines the audit trail's schema, defaults to `"carbonite_default"`
  """
  @doc since: "0.16.0"
  @spec delete_transaction_if_empty(repo()) :: {:ok, non_neg_integer()}
  @spec delete_transaction_if_empty(repo(), [prefix_option()]) :: {:ok, non_neg_integer()}
  def delete_transaction_if_empty(repo, opts \\ []) do
    {rows_deleted, _} =
      opts
      |> Query.current_transaction()
      |> Query.without_changes()
      |> repo.delete_all()

    {:ok, rows_deleted}
  end

  @doc """
  Fetches all changes of the current transaction from the database.

  Make sure to run this within a transaction.

  ## Parameters

  * `repo` - the Ecto repository
  * `opts` - optional keyword list

  ## Options

  * `carbonite_prefix` - defines the audit trail's schema, defaults to `"carbonite_default"`
  """
  @doc since: "0.5.0"
  @spec fetch_changes(repo()) :: {:ok, [Carbonite.Change.t()]}
  @spec fetch_changes(repo(), [prefix_option()]) :: {:ok, [Carbonite.Change.t()]}
  def fetch_changes(repo, opts \\ []) do
    %Carbonite.Transaction{changes: changes} =
      [preload: true]
      |> Keyword.merge(opts)
      |> Query.current_transaction()
      |> repo.one()

    {:ok, changes}
  end

  @doc """
  Sets the current transaction to "override mode" for all tables in the audit log.

  ## Parameters

  * `repo` - the Ecto repository
  * `opts` - optional keyword list

  ## Options

  * `to` - allows to specify the target mode
  * `carbonite_prefix` - defines the audit trail's schema, defaults to `"carbonite_default"`
  """
  @doc since: "0.4.0"
  @spec override_mode(repo()) :: :ok
  @spec override_mode(repo(), [{:to, Trigger.mode()} | prefix_option()]) :: :ok
  def override_mode(repo, opts \\ []) do
    carbonite_prefix = Keyword.get(opts, :carbonite_prefix, default_prefix())
    mode = Keyword.get(opts, :to, :override)

    repo.query!("SET LOCAL #{carbonite_prefix}.override_mode = '#{mode}';")

    :ok
  end

  @type process_option ::
          Carbonite.Query.outbox_queue_option()
          | {:filter, (Ecto.Query.t() -> Ecto.Query.t()) | Ecto.Query.dynamic_expr()}
          | {:chunk, pos_integer()}

  @type process_func_option :: {:memo, Outbox.memo()} | {:discard_last, boolean()}

  @typedoc """
  This type defines the callback function signature for `Carbonite.process/3`.

  The processor function receives the current chunk of transactions and the memo of the last
  function application, and must return one of

  * `:cont` - continue processing
  * `:halt` - stop processing after this chunk
  * `{:cont | :halt, opts}` - cont/halt and set some options

  After the process function invocation the Outbox is updated with new attributes.

  ## Options

  Returned options can be:

  * `memo` - memo to store on Outbox, defaults to previous memo
  * `last_transaction_id` - last transaction id to remember as processed, defaults to previous
                            `last_transaction_id` on `:halt`, defaults to last id in current
                            chunk when `:cont` is returned
  """
  @type process_func ::
          ([Transaction.t()], Outbox.memo() ->
             :cont | :halt | {:cont | :halt, [process_func_option()]})

  @doc """
  Processes an outbox queue.

  This function sends chunks of persisted transactions to a user-supplied processing function. It
  looks up the current "reading position" from a given `Carbonite.Outbox` and yields transactions
  matching the given filter criteria (`min_age`, etc.) until either the input source is exhausted
  or a processing function application returns `:halt`.

  Returns either `{:ok, outbox}` or `{:halt, outbox}` depending on whether processing was halted
  explicitly or due the exhausted input source.

  See `Carbonite.Query.outbox_queue/2` for query options.

  ## Examples

      Carbonite.process(MyApp.Repo, "rabbit_holes", fn [transaction], _memo ->
        # The transaction has its changes preloaded.
        transaction
        |> MyApp.Foo.serialize()
        |> MyApp.Foo.send_to_external_database()

        :cont
      end)

  ### Memo passing

  The `memo` is useful to carry data between each processor application. Let's say you wanted to
  generate a hashsum chain on your processed data:

      Carbonite.process(MyApp.Repo, "rabbit_holes", fn [transaction], %{"checksum" => checksum} ->
        {payload, checksum} = MyApp.Foo.serialize_and_hash(transaction, checksum)

        MyApp.Foo.send_to_external_database(payload)

        {:cont, memo: %{"checksum" => checksum}}
      end)

  ### Chunking / Limiting

  The examples above received a single-element list as their first parameter: This is because the
  transactions are actually processed in "chunks" and the default chunk size is 1. If you would
  like to process more transactions in one chunk, set the `chunk` option:

      Carbonite.process(MyApp.Repo, "rabbit_holes", [chunk: 50], fn transactions, _memo ->
        for transaction <- transactions do
          transaction
          |> MyApp.Foo.serialize()
          |> MyApp.Foo.send_to_external_database()
        end

        :cont
      end)

  The query that is executed to fetch the data from the database is controlled with the `limit`
  option and is independent of the chunk size.

  ### Error handling

  In case you run into an error midway into processing a batch, you may choose to halt processing
  while remembering about the last processed transaction. This is equivalent to raising an
  exception from the processing function.

      Carbonite.process(MyApp.Repo, "rabbit_holes", fn [transaction], _memo ->
        case send_to_external_database(transaction) do
          :ok ->
            :cont

          {:error, _term} ->
            :halt
        end
      end

  You can, however, if you know the first half of a batch has been processed, still update the
  `memo` and `last_transaction_id`.

      Carbonite.process(MyApp.Repo, "rabbit_holes", fn transactions, _memo ->
        case process_transactions(transactions) do
          {:error, last_successful_transaction} ->
            {:halt, last_transaction_id: last_successful_transaction.id}

          :ok ->
            :cont
        end
      end

  ## Parameters

  * `repo` - the Ecto repository
  * `outbox_name` - name of the outbox to process
  * `opts` - optional keyword list
  * `process_func` - see `t:process_func/0` for details

  ## Options

  * `min_age` - the minimum age of a record, defaults to 300 seconds (set nil to disable)
  * `limit` - limits the query in size, defaults to 100 (set nil to disable)
  * `filter` - Ecto `dynamic/2` or function for refining the batch query, defaults to nil
  * `chunk` - defines the size of the chunk passed to the process function, defaults to 1
  * `carbonite_prefix` - defines the audit trail's schema, defaults to `"carbonite_default"`
  """
  @doc since: "0.4.0"
  @spec process(repo(), Outbox.name(), process_func()) :: {:ok | :halt, Outbox.t()}
  @spec process(repo(), Outbox.name(), [process_option()], process_func()) ::
          {:ok | :halt, Outbox.t()}
  def process(repo, outbox_name, opts \\ [], process_func) do
    outbox = load_outbox(repo, outbox_name, opts)

    functions = %{
      query: query_func(repo, opts),
      process: process_func(process_func),
      update: update_func(repo)
    }

    process_outbox(outbox, functions)
  end

  defp load_outbox(repo, outbox_name, opts) do
    opts = Keyword.take(opts, [:carbonite_prefix])

    outbox_name
    |> Carbonite.Query.outbox(opts)
    |> repo.one!()
  end

  defp query_func(repo, opts) do
    filter = Keyword.get(opts, :filter, dynamic(true))
    chunk = Keyword.get(opts, :chunk, 1)

    fn outbox ->
      outbox
      |> Carbonite.Query.outbox_queue(opts)
      |> apply_filter(filter)
      |> repo.all()
      |> Enum.chunk_every(chunk)
    end
  end

  defp apply_filter(query, filter) when is_function(filter), do: filter.(query)
  defp apply_filter(query, %Ecto.Query.DynamicExpr{} = filter), do: where(query, ^filter)

  defp process_func(process_func) do
    fn chunk, outbox ->
      case process_func.(chunk, outbox.memo) do
        cont_or_halt when is_atom(cont_or_halt) -> {cont_or_halt, []}
        {cont_or_halt, opts} -> {cont_or_halt, opts}
      end
    end
  end

  defp update_func(repo) do
    fn outbox, attrs ->
      outbox
      |> Outbox.changeset(attrs)
      |> repo.update!()
    end
  end

  defp process_outbox(outbox, functions) do
    case functions.query.(outbox) do
      [] -> {:ok, outbox}
      chunks -> process_chunks(chunks, outbox, functions)
    end
  end

  defp process_chunks([], outbox, functions) do
    process_outbox(outbox, functions)
  end

  defp process_chunks([chunk | rest], outbox, functions) do
    case process_chunk(chunk, outbox, functions) do
      {:cont, outbox} -> process_chunks(rest, outbox, functions)
      halt -> halt
    end
  end

  defp process_chunk(chunk, outbox, functions) do
    {cont_or_halt, result_opts} = functions.process.(chunk, outbox)

    defaults =
      if cont_or_halt == :cont do
        %{last_transaction_id: List.last(chunk).id}
      else
        %{}
      end

    results =
      result_opts
      |> Keyword.take([:memo, :last_transaction_id])
      |> Map.new()

    outbox = functions.update.(outbox, Map.merge(defaults, results))

    {cont_or_halt, outbox}
  end

  @type purge_option :: Carbonite.Query.outbox_done_option()

  @doc """
  Deletes transactions that have been fully processed.

  See `Carbonite.Query.outbox_done/1` for query options.

  Returns the number of deleted transactions.

  ## Parameters

  * `repo` - the Ecto repository
  * `opts` - optional keyword list

  ## Options

  * `min_age` - the minimum age of a record, defaults to 300 seconds (set nil to disable)
  * `carbonite_prefix` - defines the audit trail's schema, defaults to `"carbonite_default"`
  """
  @doc since: "0.4.0"
  @spec purge(repo()) :: {:ok, non_neg_integer()}
  @spec purge(repo(), [purge_option()]) :: {:ok, non_neg_integer()}
  def purge(repo, opts \\ []) do
    {deleted, nil} =
      opts
      |> Carbonite.Query.outbox_done()
      |> repo.delete_all()

    {:ok, deleted}
  end
end