defmodule EctoTrailer do @moduledoc """ EctoTrailer allows to store changeset changes into a separate `audit_log` table. ## Usage 1. Add `ecto_trailer` to your list of dependencies in `mix.exs`: def deps do [{:ecto_trailer, "~> 0.1.0"}] end 2. Add a migration that creates `audit_log` table to `priv/repo/migrations` folder: defmodule EctoTrailer.TestRepo.Migrations.CreateAuditLogTable do @moduledoc false use Ecto.Migration def change do create table(:audit_log, primary_key: false) do add :id, :uuid, primary_key: true add :actor_id, :string, null: false add :resource, :string, null: false add :resource_id, :string, null: false add :changeset, :map, null: false timestamps([type: :utc_datetime, updated_at: false]) end end end 4. Use `EctoTrailer` in your repo: defmodule MyApp.Repo do use Ecto.Repo, otp_app: :my_app use EctoTrailer end 5. Use logging functions instead of defaults. See `EctoTrailer` module docs. """ alias Ecto.Changeset alias EctoTrailer.Changelog alias Ecto.Multi require Logger @type action_type :: :insert | :update | :upsert | :delete # Cache frequently accessed config to avoid repeated lookups @redacted_fields_config Application.compile_env(:ecto_trailer, :redacted_fields, nil) @changelog_fields [:actor_id, :resource, :resource_id, :changeset, :change_type] @not_loaded_pattern "Ecto.Association.NotLoaded" defmacro __using__(_) do quote do @type action_type :: :insert | :update | :upsert | :delete @doc """ Store changes in a `change_log` table. """ @spec log( struct_or_changeset :: Ecto.Schema.t() | Ecto.Changeset.t(), changes :: map(), actor_id :: String.T, action_type :: action_type() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def log(struct_or_changeset, changes, actor_id, action_type), do: EctoTrailer.log(__MODULE__, struct_or_changeset, changes, actor_id, action_type) @doc """ Store bulk changes in a `change_log` table. """ @spec log_bulk( structs :: [Ecto.Schema.t()], changes :: [map()], actor_id :: String.t(), action_type :: action_type() ) :: :ok def log_bulk(structs, changes, actor_id, action_type), do: EctoTrailer.log_bulk(__MODULE__, structs, changes, actor_id, action_type) @doc """ Call `c:Ecto.Repo.insert/2` operation and store changes in a `change_log` table. Insert arguments, return and options same as `c:Ecto.Repo.insert/2` has. """ @spec insert_and_log( struct_or_changeset :: Ecto.Schema.t() | Ecto.Changeset.t(), actor_id :: String.T, opts :: Keyword.t() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def insert_and_log(struct_or_changeset, actor_id, opts \\ []), do: EctoTrailer.insert_and_log(__MODULE__, struct_or_changeset, actor_id, opts) @doc """ Call `c:Ecto.Repo.update/2` operation and store changes in a `change_log` table. Insert arguments, return and options same as `c:Ecto.Repo.update/2` has. """ @spec update_and_log( changeset :: Ecto.Changeset.t(), actor_id :: String.T, opts :: Keyword.t() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def update_and_log(changeset, actor_id, opts \\ []), do: EctoTrailer.update_and_log(__MODULE__, changeset, actor_id, opts) @doc """ Call `c:Ecto.Repo.upsert/2` operation and store changes in a `change_log` table. Insert arguments, return and options same as `c:Ecto.Repo.upsert/2` has. """ @spec upsert_and_log( struct_or_changeset :: Ecto.Schema.t() | Ecto.Changeset.t(), actor_id :: String.T, opts :: Keyword.t() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def upsert_and_log(struct_or_changeset, actor_id, opts \\ []), do: EctoTrailer.upsert_and_log(__MODULE__, struct_or_changeset, actor_id, opts) @doc """ Call `c:Ecto.Repo.delete/2` operation and store deleted objext in a `change_log` table. """ @spec delete_and_log( struct_or_changeset :: Ecto.Schema.t() | Ecto.Changeset.t(), actor_id :: String.T, opts :: Keyword.t() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def delete_and_log(struct_or_changeset, actor_id, opts \\ []), do: EctoTrailer.delete_and_log(__MODULE__, struct_or_changeset, actor_id, opts) end end @doc """ Store changes in a `change_log` table. """ @spec log( repo :: Ecto.Repo.t(), struct_or_changeset :: Ecto.Schema.t() | Ecto.Changeset.t(), changes :: map(), actor_id :: String.T, action_type :: action_type() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def log(repo, struct_or_changeset, changes, actor_id, action_type) do Multi.new() |> Multi.run(:operation, fn _, _ -> {:ok, struct_or_changeset} end) |> run_logging_transaction_alone(repo, struct_or_changeset, changes, actor_id, action_type) end @doc """ Store bulk changes in a `change_log` table. """ @spec log_bulk( repo :: Ecto.Repo.t(), structs :: [Ecto.Schema.t()], changes :: [map()], actor_id :: String.t(), action_type :: action_type() ) :: :ok def log_bulk(repo, structs, changes, actor_id, action_type) do actor_id_str = to_actor_id_string(actor_id) Enum.zip(structs, changes) |> Enum.each(fn {struct, change} -> Multi.new() |> Multi.run(:operation, fn _, _ -> {:ok, struct} end) |> run_logging_transaction_alone(repo, struct, change, actor_id_str, action_type) end) end @doc """ Call `c:Ecto.Repo.insert/2` operation and store changes in a `change_log` table. Insert arguments, return and options same as `c:Ecto.Repo.insert/2` has. """ @spec insert_and_log( repo :: Ecto.Repo.t(), struct_or_changeset :: Ecto.Schema.t() | Ecto.Changeset.t(), actor_id :: String.T, opts :: Keyword.t() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def insert_and_log(repo, struct_or_changeset, actor_id, opts \\ []) do Multi.new() |> Multi.insert(:operation, struct_or_changeset, opts) |> run_logging_transaction(repo, struct_or_changeset, actor_id, :insert) end @doc """ Call `c:Ecto.Repo.update/2` operation and store changes in a `change_log` table. Insert arguments, return and options same as `c:Ecto.Repo.update/2` has. """ @spec update_and_log( repo :: Ecto.Repo.t(), changeset :: Ecto.Changeset.t(), actor_id :: String.T, opts :: Keyword.t() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def update_and_log(repo, changeset, actor_id, opts \\ []) do Multi.new() |> Multi.update(:operation, changeset, opts) |> run_logging_transaction(repo, changeset, actor_id, :update) end @doc """ Call `c:Ecto.Repo.upsert/2` operation and store changes in a `change_log` table. Insert arguments, return and options same as `c:Ecto.Repo.upsert/2` has. """ @spec upsert_and_log( repo :: Ecto.Repo.t(), struct_or_changeset :: Ecto.Schema.t() | Ecto.Changeset.t(), actor_id :: String.T, opts :: Keyword.t() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def upsert_and_log(repo, struct_or_changeset, actor_id, opts \\ []) do Multi.new() |> Multi.insert_or_update(:operation, struct_or_changeset, opts) |> run_logging_transaction(repo, struct_or_changeset, actor_id, :upsert) end @doc """ Call `c:Ecto.Repo.delete/2` operation and store deleted objext in a `change_log` table. """ @spec delete_and_log( repo :: Ecto.Repo.t(), struct_or_changeset :: Ecto.Schema.t() | Ecto.Changeset.t(), actor_id :: String.T, opts :: Keyword.t() ) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()} def delete_and_log(repo, struct_or_changeset, actor_id, opts \\ []) do Multi.new() |> Multi.delete(:operation, struct_or_changeset, opts) |> run_logging_transaction(repo, struct_or_changeset, actor_id, :delete) end defp run_logging_transaction(multi, repo, struct_or_changeset, actor_id, operation_type) do multi |> Multi.run(:changelog, &log_changes(&1, &2, struct_or_changeset, actor_id, operation_type)) |> repo.transaction() |> build_result() end defp run_logging_transaction_alone(multi, repo, struct, changes, actor_id, operation_type) do multi |> Multi.run( :changelog, &log_changes_alone(&1, &2, struct, changes, actor_id, operation_type) ) |> repo.transaction() |> build_result() end defp build_result({:ok, %{operation: operation}}), do: {:ok, operation} defp build_result({:error, :operation, reason, _changes_so_far}), do: {:error, reason} defp log_changes_alone( repo, %{operation: operation} = _multi_acc, _struct_or_changeset, changes, actor_id, operation_type ) do resource = operation.__struct__.__schema__(:source) actor_id_str = to_actor_id_string(actor_id) resource_id_str = to_string(operation.id) %{ actor_id: actor_id_str, resource: resource, resource_id: resource_id_str, changeset: changes, change_type: operation_type } |> changelog_changeset() |> repo.insert() |> case do {:ok, changelog} -> {:ok, changelog} {:error, reason} -> Logger.error( "Failed to store changes in audit log: #{inspect(operation)} " <> "by actor #{inspect(actor_id)}. Reason: #{inspect(reason)}" ) {:ok, reason} end end defp log_changes(repo, %{operation: operation} = _multi_acc, struct_or_changeset, actor_id, operation_type) do associations = operation.__struct__.__schema__(:associations) resource = operation.__struct__.__schema__(:source) embeds = operation.__struct__.__schema__(:embeds) struct_or_changeset = prepare_struct_or_changeset(struct_or_changeset, operation_type) changes = struct_or_changeset |> get_changes() |> get_embed_changes(embeds) |> get_assoc_changes(associations) |> redact_custom_fields() |> validate_changes(struct_or_changeset, operation_type) actor_id_str = to_actor_id_string(actor_id) resource_id_str = to_string(operation.id) %{ actor_id: actor_id_str, resource: resource, resource_id: resource_id_str, changeset: changes, change_type: operation_type } |> changelog_changeset() |> repo.insert() |> case do {:ok, changelog} -> {:ok, changelog} {:error, reason} -> Logger.error( "Failed to store changes in audit log: #{inspect(struct_or_changeset)} " <> "by actor #{inspect(actor_id)}. Reason: #{inspect(reason)}" ) {:ok, reason} end end defp prepare_struct_or_changeset(%Changeset{data: data} = _changeset, :delete), do: data defp prepare_struct_or_changeset(struct_or_changeset, _), do: struct_or_changeset defp to_actor_id_string(actor_id) when is_binary(actor_id), do: actor_id defp to_actor_id_string(actor_id), do: to_string(actor_id) defp validate_changes(_changes, schema, :delete) do # Special case for delete operations {_, return} = schema |> Map.from_struct() |> Map.pop(:__meta__) remove_empty_associations(return) end defp validate_changes(changes, _schema, _operation_type), do: changes # defp redact_custom_fields(changeset) when is_nil(@redacted_fields_config), do: changeset defp redact_custom_fields(changeset), do: redact_fields(changeset, @redacted_fields_config) defp redact_fields(changeset, redacted_fields) do Enum.reduce(redacted_fields, changeset, fn field, acc -> # Handle both string and atom keys string_field = to_string(field) # Only redact if the field exists in the changeset if Map.has_key?(acc, field) or Map.has_key?(acc, string_field) do acc |> Map.put(field, "[REDACTED]") |> Map.put(string_field, "[REDACTED]") else acc end end) end defp remove_empty_associations(struct) do struct |> Enum.map(fn {key, %{__struct__: _} = value} -> if not_loaded?(value), do: {key, nil}, else: {key, value} entry -> entry end) |> Map.new() end defp not_loaded?(value) do value |> Kernel.inspect() |> String.contains?(@not_loaded_pattern) end # Pattern matching for empty changeset defp get_changes(%Changeset{changes: changes}) when changes == %{}, do: %{} defp get_changes(%Changeset{changes: changes}), do: map_custom_ecto_types(changes) # Handle struct case defp get_changes(%{__struct__: _} = changes) do changes |> Map.from_struct() |> Map.drop([:__meta__]) |> map_custom_ecto_types() end # Handle regular map case defp get_changes(changes) when is_map(changes) do changes |> map_custom_ecto_types() end # Handle list case defp get_changes(changes) when is_list(changes) do Enum.map(changes, &get_changes/1) end # Handle other values (string, etc.) defp get_changes(value) do if not_loaded?(value), do: nil, else: value end defp get_embed_changes(changeset, []), do: changeset defp get_embed_changes(changeset, embeds) do Enum.reduce(embeds, changeset, fn embed, acc -> case Map.get(acc, embed) do nil -> acc embed_changes -> Map.put(acc, embed, get_changes(embed_changes)) end end) end defp get_assoc_changes(changeset, []), do: changeset defp get_assoc_changes(changeset, associations) do Enum.reduce(associations, changeset, fn assoc, acc -> case Map.get(acc, assoc) do nil -> acc assoc_changes when is_struct(assoc_changes) -> if not_loaded?(assoc_changes) do Map.put(acc, assoc, nil) else Map.put(acc, assoc, get_changes(assoc_changes)) end assoc_changes -> Map.put(acc, assoc, get_changes(assoc_changes)) end end) end defp map_custom_ecto_types(changes) do Map.new(changes, &map_custom_ecto_type/1) end defp map_custom_ecto_type({_field, %Changeset{}} = input), do: input defp map_custom_ecto_type({field, %{__struct__: _} = value}), do: {field, inspect(value)} defp map_custom_ecto_type({field, value}) when is_map(value) and is_map_key(value, :__struct__), do: {field, inspect(value)} defp map_custom_ecto_type({field, value}) when is_map(value), do: {field, value} defp map_custom_ecto_type(value), do: value defp changelog_changeset(attrs) do Changeset.cast(%Changelog{}, attrs, @changelog_fields) end end