Fix session duration for out-of-order events (#593)

lib/plausible/session/store.ex

@@ -1,7 +1,6 @@
 defmodule Plausible.Session.Store do
   use GenServer
   use Plausible.Repo
-  alias Plausible.Session.WriteBuffer
   import Ecto.Query, only: [from: 2]
   require Logger
 
@@ -11,7 +10,8 @@ defmodule Plausible.Session.Store do
     GenServer.start_link(__MODULE__, opts, name: __MODULE__)
   end
 
-  def init(_opts) do
+  def init(opts) do
+    buffer = Keyword.get(opts, :buffer, Plausible.Session.WriteBuffer)
     timer = Process.send_after(self(), :garbage_collect, @garbage_collect_interval_milliseconds)
 
     latest_sessions =
@@ -36,14 +36,18 @@ defmodule Plausible.Session.Store do
         _e -> %{}
       end
 
-    {:ok, %{timer: timer, sessions: sessions}}
+    {:ok, %{timer: timer, sessions: sessions, buffer: buffer}}
   end
 
-  def on_event(event, prev_user_id) do
-    GenServer.call(__MODULE__, {:on_event, event, prev_user_id})
+  def on_event(event, prev_user_id, pid \\ __MODULE__) do
+    GenServer.call(pid, {:on_event, event, prev_user_id})
   end
 
-  def handle_call({:on_event, event, prev_user_id}, _from, %{sessions: sessions} = state) do
+  def handle_call(
+        {:on_event, event, prev_user_id},
+        _from,
+        %{sessions: sessions, buffer: buffer} = state
+      ) do
     found_session = sessions[event.user_id] || (prev_user_id && sessions[prev_user_id])
     active = is_active?(found_session, event)
 
@@ -51,17 +55,17 @@ defmodule Plausible.Session.Store do
       cond do
         found_session && active ->
           new_session = update_session(found_session, event)
-          WriteBuffer.insert([%{new_session | sign: 1}, %{found_session | sign: -1}])
+          buffer.insert([%{new_session | sign: 1}, %{found_session | sign: -1}])
           Map.put(sessions, event.user_id, new_session)
 
         found_session && !active ->
           new_session = new_session_from_event(event)
-          WriteBuffer.insert([new_session])
+          buffer.insert([new_session])
           Map.put(sessions, event.user_id, new_session)
 
         true ->
           new_session = new_session_from_event(event)
-          WriteBuffer.insert([new_session])
+          buffer.insert([new_session])
           Map.put(sessions, event.user_id, new_session)
       end
 
@@ -80,7 +84,7 @@ defmodule Plausible.Session.Store do
         timestamp: event.timestamp,
         exit_page: event.pathname,
         is_bounce: false,
-        duration: Timex.diff(event.timestamp, session.start, :second),
+        duration: Timex.diff(event.timestamp, session.start, :second) |> abs,
         pageviews:
           if(event.name == "pageview", do: session.pageviews + 1, else: session.pageviews),
         events: session.events + 1

test/plausible/session/store_test.exs

@@ -0,0 +1,99 @@
+defmodule Plausible.Session.StoreTest do
+  use Plausible.DataCase
+  import Double
+  alias Plausible.Session.{Store, WriteBuffer}
+
+  setup do
+    buffer =
+      WriteBuffer
+      |> stub(:insert, fn _sessions -> nil end)
+
+    {:ok, store} = GenServer.start_link(Store, buffer: buffer)
+    [store: store, buffer: buffer]
+  end
+
+  test "creates a session from an event", %{store: store} do
+    event =
+      build(:event,
+        name: "pageview",
+        referrer: "ref",
+        referrer_source: "refsource",
+        utm_medium: "medium",
+        utm_source: "source",
+        utm_campaign: "campaign",
+        browser: "browser",
+        browser_version: "55",
+        country_code: "EE",
+        screen_size: "Desktop",
+        operating_system: "Mac",
+        operating_system_version: "11"
+      )
+
+    Store.on_event(event, nil, store)
+
+    assert_receive({WriteBuffer, :insert, [sessions]})
+    assert [session] = sessions
+    assert session.hostname == event.hostname
+    assert session.domain == event.domain
+    assert session.user_id == event.user_id
+    assert session.entry_page == event.pathname
+    assert session.exit_page == event.pathname
+    assert session.is_bounce == true
+    assert session.duration == 0
+    assert session.pageviews == 1
+    assert session.events == 1
+    assert session.referrer == event.referrer
+    assert session.referrer_source == event.referrer_source
+    assert session.utm_medium == event.utm_medium
+    assert session.utm_source == event.utm_source
+    assert session.utm_campaign == event.utm_campaign
+    assert session.country_code == event.country_code
+    assert session.screen_size == event.screen_size
+    assert session.operating_system == event.operating_system
+    assert session.operating_system_version == event.operating_system_version
+    assert session.browser == event.browser
+    assert session.browser_version == event.browser_version
+    assert session.timestamp == event.timestamp
+    assert session.start === event.timestamp
+  end
+
+  test "updates a session", %{store: store} do
+    timestamp = Timex.now()
+    event1 = build(:event, name: "pageview", timestamp: timestamp |> Timex.shift(seconds: -10))
+
+    event2 =
+      build(:event,
+        domain: event1.domain,
+        user_id: event1.user_id,
+        name: "pageview",
+        timestamp: timestamp
+      )
+
+    Store.on_event(event1, nil, store)
+    Store.on_event(event2, nil, store)
+    assert_receive({WriteBuffer, :insert, [[session, _negative_record]]})
+    assert session.is_bounce == false
+    assert session.duration == 10
+    assert session.pageviews == 2
+    assert session.events == 2
+  end
+
+  test "calculates duration correctly for out-of-order events", %{store: store} do
+    timestamp = Timex.now()
+    event1 = build(:event, name: "pageview", timestamp: timestamp |> Timex.shift(seconds: 10))
+
+    event2 =
+      build(:event,
+        domain: event1.domain,
+        user_id: event1.user_id,
+        name: "pageview",
+        timestamp: timestamp
+      )
+
+    Store.on_event(event1, nil, store)
+    Store.on_event(event2, nil, store)
+
+    assert_receive({WriteBuffer, :insert, [[session, _negative_record]]})
+    assert session.duration == 10
+  end
+end