Activity Tracking in Phoenix LiveView

Most SaaS apps need some form of activity tracking — who did what and when. Whether it's for an admin audit log, a user-facing activity feed, or just debugging "who deleted that record?", having a trail of events is incredibly useful.

In this tutorial, we'll build an automatic activity tracking system that hooks directly into Ecto's Repo. The idea is simple: override Repo.insert/2, Repo.update/2, and Repo.delete/2 so that trackable schemas automatically get logged — without touching any of your existing context functions.

We'll also add a Trackable protocol so each schema can decide which actions to track and how to describe itself in the feed.

Step 1: The Migration

We need two tables: one for actors (the users performing actions) and one for events (the actions themselves). The actor table might seem unnecessary at first — why not just put user_id directly on the event? The reason is that it gives us a single place to upsert per user and makes it easy to extend later (say, with an account_id for multi-tenancy).

# priv/repo/migrations/20240229072201_add_activity_feed_tables.exs
defmodule MyApp.Repo.Migrations.AddActivityFeedTables do
  use Ecto.Migration

  def change do
    create table(:activity_actors) do
      add :user_id, references(:users, on_delete: :nilify_all)

      timestamps(type: :utc_datetime)
    end

    create unique_index(:activity_actors, :user_id)

    create table(:activity_events) do
      add :actor_id, references(:activity_actors, on_delete: :delete_all), null: false
      add :action, :string, null: false
      add :subject, :string
      add :context, :string
      add :data, :map

      timestamps(type: :utc_datetime, updated_at: false)
    end

    create index(:activity_events, :actor_id)
    create index(:activity_events, :data, using: "gin")
  end
end

A few things to note:

updated_at: false on events — events are immutable. Once recorded, they never change.
GIN index on data — we store the tracked record reference as a JSONB field, and the GIN index lets us query it efficiently (e.g. "show me all events for this specific post").
on_delete: :nilify_all on user_id — if a user is deleted, we keep the events but lose the actor link. You could also use :delete_all if you prefer to purge everything.

Run the migration:

mix ecto.migrate

Step 2: The Schemas

The ActivityActor schema is thin on purpose. One actor per user, and it links to many events:

# lib/my_app/activity_tracking/activity_actor.ex
defmodule MyApp.ActivityTracking.ActivityActor do
  use Ecto.Schema
  import Ecto.Changeset

  schema "activity_actors" do
    belongs_to :user, MyApp.Accounts.User
    has_many :events, MyApp.ActivityTracking.ActivityEvent, foreign_key: :actor_id

    timestamps(type: :utc_datetime)
  end

  def changeset(actor, attrs) do
    actor
    |> cast(attrs, [:user_id])
    |> validate_required([:user_id])
  end
end

The ActivityEvent schema holds the actual event data:

# lib/my_app/activity_tracking/activity_event.ex
defmodule MyApp.ActivityTracking.ActivityEvent do
  use Ecto.Schema
  import Ecto.Changeset

  schema "activity_events" do
    field :action, Ecto.Enum, values: [:insert, :update, :delete], default: :insert
    field :subject, :string
    field :context, :string
    field :data, :map

    belongs_to :actor, MyApp.ActivityTracking.ActivityActor

    timestamps(type: :utc_datetime, updated_at: false)
  end

  def changeset(event, attrs) do
    event
    |> cast(attrs, [:action, :subject, :context, :data])
    |> validate_required([:action, :subject])
  end
end

The fields:

action — one of :insert, :update, or :delete. Using Ecto.Enum keeps it typed.
subject — a human-readable description like "post My First Blog Post" or "user John". This is what shows up in the feed.
context — optional extra context (e.g. which page the action happened on).
data — a JSONB map. We store a "record" key here that points to the tracked struct using a normalized string like "Posts.Post:42". This lets us query all events for a specific record.

Step 3: The Trackable Protocol

Not every schema should be tracked. A Trackable protocol lets each schema opt in and decide how it's described in the activity feed:

# lib/my_app/activity_tracking/trackable.ex
defprotocol MyApp.ActivityTracking.Trackable do
  @doc "Which actions to track for this struct. Returns a list like [:insert, :update, :delete]."
  def allowed_actions(struct)

  @doc "Returns a human-readable subject string for the activity feed."
  def to_activity(struct)
end

Then implement it for any schema you want tracked:

defimpl MyApp.ActivityTracking.Trackable, for: MyApp.Blog.Post do
  def allowed_actions(_post), do: [:insert, :update, :delete]

  def to_activity(post) do
    "post #{post.title}"
  end
end

defimpl MyApp.ActivityTracking.Trackable, for: MyApp.Accounts.User do
  def allowed_actions(_user), do: [:update]

  def to_activity(user) do
    "user #{user.email}"
  end
end

This is the nice part — each schema controls its own tracking behaviour. Maybe you only want to track inserts for one schema but all three actions for another. The protocol makes that clean.

Step 4: The ActivityTracking Context

This is the core module. It handles tracking events, managing the current actor via the process dictionary, and a helper for building record keys:

# lib/my_app/activity_tracking/activity_tracking.ex
defmodule MyApp.ActivityTracking do
  import Ecto.Query, warn: false

  alias MyApp.Repo
  alias MyApp.ActivityTracking.{ActivityActor, ActivityEvent}

  @actor_key {__MODULE__, :actor_id}

  # --- Actor management via process dictionary ---

  def put_actor(%{id: user_id}), do: put_actor(user_id)

  def put_actor(user_id) do
    Process.put(@actor_key, user_id)
  end

  def get_actor do
    Process.get(@actor_key)
  end

  # --- Tracking ---

  def track(user_id, attrs) do
    with {:ok, actor} <- upsert_actor(user_id) do
      %ActivityEvent{}
      |> ActivityEvent.changeset(attrs)
      |> Ecto.Changeset.put_assoc(:actor, actor)
      |> Repo.insert()
    end
  end

  def upsert_actor(user_id) do
    %ActivityActor{user_id: user_id}
    |> Repo.insert(
      conflict_target: [:user_id],
      on_conflict: {:replace_all_except, [:id, :inserted_at]},
      returning: true
    )
  end

  def tracked?(struct) do
    ActivityEvent
    |> where_record(struct)
    |> Repo.exists?()
  end

  def tracked_by?(struct, user_or_id) do
    ActivityEvent
    |> where_record(struct)
    |> where_users(user_or_id)
    |> Repo.exists?()
  end

  # --- Helpers ---

  def record_key(%{__struct__: module, id: id}) do
    module
    |> Module.split()
    |> tl()
    |> Enum.join(".")
    |> Kernel.<>(":#{id}")
  end

  defp where_record(query, struct) do
    key = record_key(struct)
    from q in query, where: fragment("? ->> ? = ?", q.data, "record", ^key)
  end

  defp where_users(query, user_or_users) do
    user_ids =
      user_or_users
      |> List.wrap()
      |> Enum.map(fn
        %{id: id} -> id
        id -> id
      end)

    from q in query,
      join: a in assoc(q, :actor),
      where: a.user_id in ^user_ids
  end
end

A few things I want to highlight:

The process dictionary is used to store the current actor (user) for the duration of a request. This is what makes the automatic Repo tracking work — when Repo.insert fires, it can look up who's performing the action without you having to pass the user around everywhere. We'll set this in a LiveView hook in a moment.

upsert_actor/1 uses Postgres ON CONFLICT to either create or fetch the actor row. This means the first time a user triggers any tracked action, their actor is created. After that, it's just a lookup.

record_key/1 (renamed from the original struct_to_normalized_string) converts a struct to a string like "Blog.Post:42". It strips the top-level app module and joins the rest. This is how we link events back to specific records.

Step 5: The Repo Override

This is where the magic happens. We override Repo.insert/2, Repo.update/2, and Repo.delete/2 so that trackable structs are automatically logged:

# lib/my_app/activity_tracking/tracked_repo.ex
defmodule MyApp.ActivityTracking.TrackedRepo do
  alias MyApp.ActivityTracking
  alias MyApp.ActivityTracking.Trackable

  defmacro __using__(_opts) do
    quote do
      defoverridable(
        insert: 2,
        insert!: 2,
        update: 2,
        update!: 2,
        delete: 2,
        delete!: 2
      )

      defp trackable?(struct) do
        Trackable.impl_for(struct) != nil
      end

      defp resolve_actor(opts) do
        case Keyword.get(opts, :actor) do
          nil -> ActivityTracking.get_actor()
          %{id: user_id} -> user_id
          user_id -> user_id
        end
      end

      defp maybe_track({:ok, struct} = result, action, opts) do
        do_track(struct, action, opts)
        result
      end

      defp maybe_track(result, _action, _opts), do: result

      defp maybe_track!(struct, action, opts) do
        do_track(struct, action, opts)
        struct
      end

      defp do_track(struct, action, opts) do
        user_id = resolve_actor(opts)
        skip? = Keyword.get(opts, :track, true) == false

        if trackable?(struct) && user_id && !skip? &&
             action in Trackable.allowed_actions(struct) do
          subject =
            Keyword.get(opts, :track_subject) || Trackable.to_activity(struct)

          ActivityTracking.track(user_id, %{
            action: action,
            subject: subject,
            data: %{"record" => ActivityTracking.record_key(struct)}
          })
        end
      end

      def insert(struct, opts) do
        super(struct, opts) |> maybe_track(:insert, opts)
      end

      def update(struct, opts) do
        super(struct, opts) |> maybe_track(:update, opts)
      end

      def delete(struct, opts) do
        super(struct, opts) |> maybe_track(:delete, opts)
      end

      def insert!(struct, opts) do
        super(struct, opts) |> maybe_track!(:insert, opts)
      end

      def update!(struct, opts) do
        super(struct, opts) |> maybe_track!(:update, opts)
      end

      def delete!(struct, opts) do
        super(struct, opts) |> maybe_track!(:delete, opts)
      end
    end
  end
end

One important detail: the bang variants (insert!, update!, delete!) return the struct directly, not {:ok, struct}. So we have separate maybe_track!/3 and maybe_track/3 functions to handle both patterns.

You can also pass options to control tracking:

track: false — skip tracking for this operation: Repo.insert(changeset, track: false)
track_subject: "custom description" — override the subject text
actor: user — explicitly set the actor instead of using the process dictionary

Step 6: Wire It Into the Repo

Add a single line to your Repo module:

# lib/my_app/repo.ex
defmodule MyApp.Repo do
  use Ecto.Repo,
    otp_app: :my_app,
    adapter: Ecto.Adapters.Postgres

  use MyApp.ActivityTracking.TrackedRepo
end

That's it. Every Repo.insert, Repo.update, and Repo.delete call now checks if the struct implements Trackable and logs accordingly.

Step 7: Set the Actor in Your LiveView Hook

The last piece is telling the tracking system who is performing the action. In a LiveView app, the natural place for this is an on_mount hook:

# lib/my_app_web/live/hooks/set_actor.ex
defmodule MyAppWeb.Live.Hooks.SetActor do
  import Phoenix.LiveView

  alias MyApp.Accounts

  def on_mount(:default, _params, %{"user_token" => token} = _session, socket) do
    case Accounts.get_user_by_session_token(token) do
      %{} = user ->
        MyApp.ActivityTracking.put_actor(user)
        {:cont, assign(socket, :current_user, user)}

      nil ->
        {:halt, redirect(socket, to: "/login")}
    end
  end

  def on_mount(:default, _params, _session, socket) do
    {:cont, socket}
  end
end

If you already have a hook that loads the current user, just add the put_actor call there. No need for a separate hook.

For regular controller requests, you can do the same in a plug:

# In your auth plug, after fetching the user:
MyApp.ActivityTracking.put_actor(user)

Step 8: Manual Tracking with `with_tracking`

Sometimes you want to track something that doesn't go through Repo.insert directly — maybe a multi-step operation or something wrapped in a transaction. For that, add a with_tracking helper to the context module:

# Add to lib/my_app/activity_tracking/activity_tracking.ex
def with_tracking(user, action, subject, do: block)
    when action in [:insert, :update, :delete] do
  user_id = case user do
    %{id: id} -> id
    id -> id
  end

  case block do
    {:ok, struct} = result ->
      track(user_id, %{
        action: action,
        subject: subject,
        data: %{"record" => record_key(struct)}
      })

      result

    other ->
      other
  end
end

Use it like this:

ActivityTracking.with_tracking(current_user, :insert, "batch import") do
  Blog.create_post(attrs)
end

The event is only recorded if the block returns {:ok, struct}. If it returns an error, nothing is tracked.

Wrapping Up

With this setup, tracking is automatic for any schema that implements the Trackable protocol. You don't need to modify any of your existing context functions — just add the protocol implementation and events start flowing.

The pattern breaks down into:
Schemas — ActivityActor and ActivityEvent store the data
Protocol — Trackable lets each schema opt in and describe itself
Repo override — intercepts insert/update/delete and tracks automatically
Process dictionary — carries the current user through the request lifecycle

In the next tutorial, we'll build a LiveView to render these events as an activity feed with cursor-based pagination and real-time updates.

Tutorial