Event replay poc

extra/lib/plausible/session/computed_salts.ex

@@ -0,0 +1,66 @@
+defmodule Plausible.Session.ComputedSalts do
+  @moduledoc """
+  Cache for computed salts used for replayed events.
+
+  It's not cleaned, however the range of possible values is limited
+  to the number of replayed sessions over the lifetime of the node
+  and sessions IDs can't be provided externally by users.
+
+  Despite that, the cache is still purged every couple hours.
+  """
+
+  @purge_interval :timer.seconds(7200)
+
+  use GenServer
+
+  alias Plug.Crypto.KeyGenerator
+
+  @spec start_link(Keyword.t()) :: GenServer.on_start()
+  def start_link(opts) do
+    GenServer.start_link(__MODULE__, opts, name: opts[:name] || __MODULE__)
+  end
+
+  @impl true
+  def init(opts) do
+    name = opts[:name] || __MODULE__
+
+    ^name =
+      :ets.new(name, [
+        :named_table,
+        :set,
+        :public,
+        {:read_concurrency, true}
+      ])
+
+    schedule_purge()
+
+    {:ok, name}
+  end
+
+  @spec fetch(module() | atom(), pos_integer()) :: %{previous: nil, current: binary()}
+  def fetch(name \\ __MODULE__, replay_session_id) do
+    computed_salt =
+      secret_key_base()
+      |> KeyGenerator.generate(:binary.encode_unsigned(replay_session_id), cache: name)
+      |> binary_part(0, 16)
+
+    %{previous: nil, current: computed_salt}
+  end
+
+  @impl true
+  def handle_info(:purge, name) do
+    :ets.delete_all_objects(name)
+    schedule_purge()
+
+    {:noreply, name}
+  end
+
+  defp schedule_purge() do
+    Process.send_after(self(), :purge, @purge_interval)
+  end
+
+  defp secret_key_base() do
+    Application.get_env(:plausible, PlausibleWeb.Endpoint)
+    |> Keyword.fetch!(:secret_key_base)
+  end
+end

lib/plausible/application.ex

@@ -1,3 +1,3 @@
 defmodule Plausible.Application do
   @moduledoc false
 
@@ -194,6 +194,9 @@
         end,
         Plausible.Ingestion.Counters,
         Plausible.Session.Salts,
+        on_ee do
+          Plausible.Session.ComputedSalts
+        end,
         Supervisor.child_spec(Plausible.Event.WriteBuffer, id: Plausible.Event.WriteBuffer),
         Supervisor.child_spec(Plausible.Session.WriteBuffer, id: Plausible.Session.WriteBuffer),
         ReferrerBlocklist,

lib/plausible/clickhouse_event_v2.ex

@@ -3,6 +3,7 @@ defmodule Plausible.ClickhouseEventV2 do
   Event schema for when NumericIDs migration is complete
   """
   use Ecto.Schema
+  use Plausible
   import Ecto.Changeset
 
   @primary_key false
@@ -48,6 +49,11 @@ defmodule Plausible.ClickhouseEventV2 do
 
     field :acquisition_channel, Ch, type: "LowCardinality(String)", writable: :never
 
+    on_ee do
+      # Field used during event replay
+      field :replay_session_id, Ch, type: "UInt64"
+    end
+
     # Virtual field used during event processing
     field :interactive?, :boolean, default: true, virtual: true, writable: :never
   end
@@ -72,7 +78,7 @@ defmodule Plausible.ClickhouseEventV2 do
         :revenue_reporting_amount,
         :revenue_reporting_currency,
         :interactive?
-      ]
+      ] ++ on_ee(do: [:replay_session_id], else: [])
     )
     |> validate_required([:name, :site_id, :hostname, :pathname, :user_id, :timestamp])
   end

lib/plausible/clickhouse_session_v2.ex

@@ -3,6 +3,7 @@ defmodule Plausible.ClickhouseSessionV2 do
   Session schema for when NumericIDs migration is complete
   """
   use Ecto.Schema
+  use Plausible
 
   defmodule BoolUInt8 do
     @moduledoc """
@@ -75,6 +76,11 @@ defmodule Plausible.ClickhouseSessionV2 do
     field :transferred_from, :string
 
     field :acquisition_channel, Ch, type: "LowCardinality(String)", writable: :never
+
+    on_ee do
+      # Field used during event replay
+      field :replay_session_id, Ch, type: "UInt64"
+    end
   end
 
   def random_uint64() do

lib/plausible/ingestion/event.ex

@@ -276,7 +276,7 @@ defmodule Plausible.Ingestion.Event do
   end
 
   defp put_basic_info(%__MODULE__{} = event, _context) do
-    update_event_attrs(event, %{
+    attrs = %{
       domain: event.domain,
       site_id: event.site.id,
       timestamp: event.request.timestamp,
@@ -286,7 +286,13 @@ defmodule Plausible.Ingestion.Event do
       scroll_depth: event.request.scroll_depth,
       engagement_time: event.request.engagement_time,
       interactive?: event.request.interactive?
-    })
+    }
+
+    on_ee do
+      attrs = Map.put(attrs, :replay_session_id, event.request.replay_session_id)
+    end
+
+    update_event_attrs(event, attrs)
   end
 
   defp put_source_info(%__MODULE__{} = event, _context) do
@@ -380,8 +386,18 @@ defmodule Plausible.Ingestion.Event do
     end
   end
 
-  defp put_salts(%__MODULE__{} = event, _context) do
-    %{event | salts: Plausible.Session.Salts.fetch()}
+  on_ee do
+    defp put_salts(%__MODULE__{} = event, _context) do
+      if replay_session_id = event.clickhouse_event_attrs.replay_session_id do
+        %{event | salts: Plausible.Session.ComputedSalts.fetch(replay_session_id)}
+      else
+        %{event | salts: Plausible.Session.Salts.fetch()}
+      end
+    end
+  else
+    defp put_salts(%__MODULE__{} = event, _context) do
+      %{event | salts: Plausible.Session.Salts.fetch()}
+    end
   end
 
   defp put_user_id(%__MODULE__{} = event, _context) do

lib/plausible/ingestion/request.ex

@@ -59,6 +59,9 @@ defmodule Plausible.Ingestion.Request do
 
     on_ee do
       field :revenue_source, :map
+
+      # field for replayed events
+      field :replay_session_id, :integer
     end
 
     field :query_params, :map
@@ -91,6 +94,7 @@ defmodule Plausible.Ingestion.Request do
           |> put_uri(request_body)
           |> put_hostname()
           |> put_user_agent(conn)
+          |> put_replay_data(conn)
           |> put_request_params(request_body)
           |> put_referrer(request_body)
           |> put_pathname()
@@ -131,6 +135,53 @@ defmodule Plausible.Ingestion.Request do
     defp put_revenue_source(changeset, _request_body), do: changeset
   end
 
+  on_ee do
+    @replay_session_id_header "x-replay-session-id"
+    @replay_time_header "x-replay-time"
+
+    defp put_replay_data(changeset, conn) do
+      now = NaiveDateTime.utc_now(:second)
+
+      replay_session_id =
+        conn
+        |> Plug.Conn.get_req_header(@replay_session_id_header)
+        |> List.first()
+        |> to_integer()
+
+      if replay_session_id do
+        time =
+          conn
+          |> Plug.Conn.get_req_header(@replay_time_header)
+          |> List.first()
+          |> NaiveDateTime.from_iso8601!()
+
+        if NaiveDateTime.compare(time, now) in [:lt, :eq] do
+          changeset
+          |> Changeset.put_change(:replay_session_id, replay_session_id)
+          |> Changeset.put_change(:timestamp, time)
+        else
+          changeset
+        end
+      else
+        changeset
+      end
+    end
+
+    defp to_integer(s) when is_binary(s) do
+      case Integer.parse(s) do
+        {n, ""} when n > 0 ->
+          n
+
+        _ ->
+          nil
+      end
+    end
+
+    defp to_integer(_), do: nil
+  else
+    defp put_replay_data(changeset, _conn), do: changeset
+  end
+
   defp put_remote_ip(changeset, conn) do
     Changeset.put_change(changeset, :remote_ip, PlausibleWeb.RemoteIP.get(conn))
   end

lib/plausible/session/cache_store.ex

@@ -3,6 +3,8 @@ defmodule Plausible.Session.CacheStore do
   Session management on the basis of incoming events.
   """
 
+  use Plausible
+
   alias Plausible.Session.WriteBuffer
 
   @lock_timeout 1000
@@ -70,7 +72,18 @@ defmodule Plausible.Session.CacheStore do
   defp find_session(_domain, nil), do: nil
 
   defp find_session(event, user_id) do
-    from_cache = Plausible.Cache.Adapter.get(:sessions, {event.site_id, user_id})
+    key =
+      on_ee do
+        if event.replay_session_id do
+          {event.site_id, user_id, event.replay_session_id}
+        else
+          {event.site_id, user_id}
+        end
+      else
+        {event.site_id, user_id}
+      end
+
+    from_cache = Plausible.Cache.Adapter.get(:sessions, key)
 
     case from_cache do
       nil ->
@@ -84,7 +97,17 @@ defmodule Plausible.Session.CacheStore do
   end
 
   defp update_session_cache(session) do
-    key = {session.site_id, session.user_id}
+    key =
+      on_ee do
+        if session.replay_session_id do
+          {session.site_id, session.user_id, session.replay_session_id}
+        else
+          {session.site_id, session.user_id}
+        end
+      else
+        {session.site_id, session.user_id}
+      end
+
     Plausible.Cache.Adapter.put(:sessions, key, session, dirty?: true)
     session
   end
@@ -126,7 +149,7 @@ defmodule Plausible.Session.CacheStore do
   end
 
   defp new_session_from_event(event, session_attributes) do
-    %Plausible.ClickhouseSessionV2{
+    new_session = %Plausible.ClickhouseSessionV2{
       sign: 1,
       session_id: Plausible.ClickhouseSessionV2.random_uint64(),
       hostname: if(event.name == "pageview", do: event.hostname, else: ""),
@@ -161,5 +184,11 @@ defmodule Plausible.Session.CacheStore do
       "entry_meta.key": Map.get(event, :"meta.key"),
       "entry_meta.value": Map.get(event, :"meta.value")
     }
+
+    on_ee do
+      %{new_session | replay_session_id: event.replay_session_id}
+    else
+      new_session
+    end
   end
 end

priv/ingest_repo/migrations/20260625082529_rename_recovery_id_to_replay_session_id.exs

@@ -0,0 +1,40 @@
+defmodule Plausible.IngestRepo.Migrations.RenameRecoveryIdToReplaySessionId do
+  use Ecto.Migration
+
+  import Plausible.MigrationUtils
+
+  @on_cluster_sessions on_cluster_statement("sessions_v2")
+  @on_cluster_events on_cluster_statement("events_v2")
+
+  def up do
+    if enterprise_edition?() do
+      execute """
+      ALTER TABLE sessions_v2
+      #{@on_cluster_sessions}
+      RENAME COLUMN IF EXISTS recovery_id TO replay_session_id
+      """
+
+      execute """
+      ALTER TABLE events_v2
+      #{@on_cluster_events}
+      RENAME COLUMN IF EXISTS recovery_id TO replay_session_id
+      """
+    end
+  end
+
+  def down do
+    if enterprise_edition?() do
+      execute """
+      ALTER TABLE sessions_v2
+      #{@on_cluster_sessions}
+      RENAME COLUMN IF EXISTS replay_session_id TO recovery_id
+      """
+
+      execute """
+      ALTER TABLE events_v2
+      #{@on_cluster_events}
+      RENAME COLUMN IF EXISTS replay_session_id TO recovery_id
+      """
+    end
+  end
+end