test/plausible/session/cache_store_test.exs

defmodule Plausible.Session.CacheStoreTest do
  use Plausible.DataCase

  alias Plausible.Session.CacheStore

  @session_params %{
    referrer: "ref",
    referrer_source: "refsource",
    utm_medium: "medium",
    utm_source: "source",
    utm_campaign: "campaign",
    utm_content: "content",
    utm_term: "term",
    browser: "browser",
    browser_version: "55",
    country_code: "EE",
    screen_size: "Desktop",
    operating_system: "Mac",
    operating_system_version: "11"
  }

  setup do
    current_pid = self()

    buffer = fn sessions ->
      send(current_pid, {:buffer, :insert, [sessions]})
      {:ok, sessions}
    end

    slow_buffer = fn sessions ->
      Process.sleep(200)
      send(current_pid, {:slow_buffer, :insert, [sessions]})
      {:ok, sessions}
    end

    [buffer: buffer, slow_buffer: slow_buffer]
  end

  test "event processing is sequential within session", %{
    buffer: buffer,
    slow_buffer: slow_buffer,
    test: test
  } do
    telemetry_event = CacheStore.lock_telemetry_event()

    test_pid = self()

    :telemetry.attach(
      "#{test}-telemetry-handler",
      telemetry_event,
      fn ^telemetry_event, %{duration: d}, _, _ when is_integer(d) ->
        send(test_pid, {:telemetry_handled, d})
      end,
      %{}
    )

    event1 = build(:event, name: "pageview")
    event2 = build(:event, name: "pageview", user_id: event1.user_id, site_id: event1.site_id)
    event3 = build(:event, name: "pageview", user_id: event1.user_id, site_id: event1.site_id)

    CacheStore.on_event(event1, @session_params, nil, buffer_insert: buffer)

    assert_receive({:buffer, :insert, [[session1]]})
    assert_receive({:telemetry_handled, duration})
    assert is_integer(duration)

    [event2, event3]
    |> Enum.map(fn e ->
      Task.async(fn ->
        CacheStore.on_event(e, @session_params, nil, buffer_insert: slow_buffer)
      end)
    end)
    |> Task.await_many()

    assert_receive({:slow_buffer, :insert, [[removed_session11, updated_session12]]})
    assert_receive({:slow_buffer, :insert, [[removed_session12, updated_session13]]})

    # Without isolation enforced in `CacheStore.on_event/4`,
    # event2 and event3 would both get executed in parallel
    # and treat event1 as event to be updated. This would result
    # in a following set of entries in Clickhouse sessions
    # table for _the same_ session:
    #
    #             session_id | is_bounce | sign
    # (event1)      123           1         1
    # (event2)      123           1         -1
    #               123           0         1
    # (event3)      123           1         -1
    #               123           0         1
    #
    # Once collapsing merge tree table does collapsing, we'd end up with:
    #
    # session_id | is_bounce | sign
    #   123           0         1
    #   123           1         -1
    #   123           0         1
    #
    # This in turn led to sum(sign * is_bounce) < 0 which, after underflowed casting,
    # ended up with 2^32-(small n) for bounce_rate value.

    assert removed_session11 == %{session1 | sign: -1}
    assert updated_session12.sign == 1
    assert updated_session12.events == 2
    assert updated_session12.pageviews == 2
    assert removed_session12 == %{updated_session12 | sign: -1}
    assert updated_session13.sign == 1
    assert updated_session13.events == 3
    assert updated_session13.pageviews == 3
  end

  @tag :slow
  test "in case of lock kicking in, the slow event finishes processing", %{buffer: buffer} do
    current_pid = self()

    very_slow_buffer = fn sessions ->
      Process.sleep(1000)
      send(current_pid, {:very_slow_buffer, :insert, [sessions]})
      {:ok, sessions}
    end

    event1 = build(:event, name: "pageview")
    event2 = build(:event, name: "pageview", user_id: event1.user_id, site_id: event1.site_id)
    event3 = build(:event, name: "pageview", user_id: event1.user_id, site_id: event1.site_id)

    async1 =
      Task.async(fn ->
        CacheStore.on_event(event1, @session_params, nil, buffer_insert: very_slow_buffer)
      end)

    # Ensure next events are executed after processing event1 starts
    Process.sleep(100)

    async2 =
      Task.async(fn ->
        CacheStore.on_event(event2, @session_params, nil, buffer_insert: buffer)
      end)

    async3 =
      Task.async(fn ->
        CacheStore.on_event(event3, @session_params, nil, buffer_insert: buffer)
      end)

    Task.await_many([async1, async2, async3])

    assert_receive({:very_slow_buffer, :insert, [[_session]]})
    refute_receive({:buffer, :insert, [[_updated_session]]})
  end

  @tag :slow
  test "lock on slow processing of one event does not affect unrelated events", %{buffer: buffer} do
    current_pid = self()

    very_slow_buffer = fn sessions ->
      Process.sleep(1000)
      send(current_pid, {:very_slow_buffer, :insert, [sessions]})
      {:ok, sessions}
    end

    event1 = build(:event, name: "pageview")
    event2 = build(:event, name: "pageview")
    event3 = build(:event, name: "pageview", user_id: event2.user_id, site_id: event2.site_id)

    async1 =
      Task.async(fn ->
        CacheStore.on_event(event1, @session_params, nil, buffer_insert: very_slow_buffer)
      end)

    # Ensure next events are executed after processing event1 starts
    Process.sleep(100)

    async2 =
      Task.async(fn ->
        CacheStore.on_event(event2, @session_params, nil, buffer_insert: buffer)
      end)

    Process.sleep(100)

    async3 =
      Task.async(fn ->
        CacheStore.on_event(event3, @session_params, nil, buffer_insert: buffer)
      end)

    Task.await_many([async1, async2, async3])

    assert_receive({:very_slow_buffer, :insert, [[_slow_session]]})
    assert_receive({:buffer, :insert, [[new_session1]]})
    assert_receive({:buffer, :insert, [[removed_session1, updated_session1]]})
    assert new_session1.sign == 1
    assert removed_session1.session_id == new_session1.session_id
    assert removed_session1.sign == -1
    assert updated_session1.session_id == removed_session1.session_id
    assert updated_session1.sign == 1
  end

  test "exploding event processing is passed through by locking mechanism" do
    crashing_buffer = fn _sessions ->
      raise "boom"
    end

    event = build(:event, name: "pageview")

    assert_raise RuntimeError, "boom", fn ->
      CacheStore.on_event(event, @session_params, nil, buffer_insert: crashing_buffer)
    end
  end

  test "creates a session from an event", %{buffer: buffer} do
    event =
      build(:event,
        name: "pageview",
        "meta.key": ["logged_in", "darkmode"],
        "meta.value": ["true", "false"]
      )

    CacheStore.on_event(event, @session_params, nil, buffer_insert: buffer)

    assert_receive({:buffer, :insert, [sessions]})
    assert [session] = sessions
    assert session.hostname == event.hostname

    assert session.site_id == event.site_id

    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 == Map.get(@session_params, :referrer)
    assert session.referrer_source == Map.get(@session_params, :referrer_source)
    assert session.utm_medium == Map.get(@session_params, :utm_medium)
    assert session.utm_source == Map.get(@session_params, :utm_source)
    assert session.utm_campaign == Map.get(@session_params, :utm_campaign)
    assert session.utm_content == Map.get(@session_params, :utm_content)
    assert session.utm_term == Map.get(@session_params, :utm_term)
    assert session.country_code == Map.get(@session_params, :country_code)
    assert session.screen_size == Map.get(@session_params, :screen_size)
    assert session.operating_system == Map.get(@session_params, :operating_system)
    assert session.operating_system_version == Map.get(@session_params, :operating_system_version)
    assert session.browser == Map.get(@session_params, :browser)
    assert session.browser_version == Map.get(@session_params, :browser_version)
    assert session.timestamp == event.timestamp
    assert session.start === event.timestamp
    # assert Map.get(session, :"entry.meta.key") == ["logged_in", "darkmode"]
    # assert Map.get(session, :"entry.meta.value") == ["true", "false"]
  end

  test "updates session counters", %{buffer: buffer} do
    timestamp = DateTime.utc_now()

    event1 =
      build(:event, name: "pageview", timestamp: timestamp |> NaiveDateTime.shift(second: -10))

    event2 = %{
      event1
      | timestamp: timestamp
    }

    CacheStore.on_event(event1, %{}, nil, buffer_insert: buffer)
    CacheStore.on_event(event2, %{}, nil, buffer_insert: buffer)
    assert_receive({:buffer, :insert, [[_negative_record, session]]})
    assert session.is_bounce == false
    assert session.duration == 10
    assert session.pageviews == 2
    assert session.events == 2
  end

  test "does not update session counters on engagement event", %{buffer: buffer} do
    now = NaiveDateTime.utc_now(:second)
    pageview = build(:pageview, timestamp: NaiveDateTime.shift(now, second: -10))
    engagement = %{pageview | name: "engagement", timestamp: now}

    CacheStore.on_event(pageview, %{}, nil, buffer_insert: buffer)
    CacheStore.on_event(engagement, %{}, nil, buffer_insert: buffer)
    assert_receive({:buffer, :insert, [[session]]})

    assert session.is_bounce == true
    assert session.duration == 0
    assert session.pageviews == 1
    assert session.events == 1
  end

  test "keeps session in cache after engagement events", %{buffer: buffer} do
    start = NaiveDateTime.utc_now(:second) |> NaiveDateTime.shift(day: -1)

    pageview1 = build(:event, name: "pageview", timestamp: start)

    CacheStore.on_event(pageview1, %{}, nil, buffer_insert: buffer)
    assert_receive({:buffer, :insert, [[start_session]]})

    for delta <- [20, 40, 60] do
      engagement =
        Map.merge(pageview1, %{
          name: "engagement",
          timestamp: start |> NaiveDateTime.shift(minute: delta)
        })

      CacheStore.on_event(engagement, %{}, nil, buffer_insert: buffer)
    end

    pageview2 = Map.put(pageview1, :timestamp, start |> NaiveDateTime.shift(minute: 80))
    CacheStore.on_event(pageview2, %{}, nil, buffer_insert: buffer)
    assert_receive({:buffer, :insert, [[_negative_record, updated_session]]})

    assert updated_session.session_id == start_session.session_id
    assert updated_session.is_bounce == false
    assert updated_session.duration == 80 * 60
    assert updated_session.pageviews == 2
    assert updated_session.events == 2
  end

  describe "hostname-related attributes" do
    test "initial for non-pageview" do
      site_id = new_site_id()

      event =
        build(:event,
          name: "custom_event",
          site_id: site_id,
          pathname: "/path/1",
          hostname: "example.com"
        )

      flush([event])
      session = get_session(site_id)
      assert session.hostname == ""
      assert session.exit_page_hostname == ""
    end

    test "initial for pageview" do
      site_id = new_site_id()

      event =
        build(:event,
          name: "pageview",
          site_id: site_id,
          pathname: "/path/1",
          hostname: "example.com"
        )

      flush([event])
      session = get_session(site_id)
      assert session.hostname == "example.com"
      assert session.exit_page_hostname == "example.com"
    end

    test "subsequent pageview after custom_event" do
      site_id = new_site_id()

      events = [
        build(:event,
          name: "custom_event",
          site_id: site_id,
          pathname: "/path/1",
          hostname: "whatever.example.com",
          timestamp: DateTime.shift(DateTime.utc_now(), second: -5),
          user_id: 1
        ),
        build(:event,
          name: "pageview",
          site_id: site_id,
          pathname: "/path/2",
          hostname: "example.com",
          user_id: 1
        )
      ]

      flush(events)
      session = get_session(site_id)
      assert session.hostname == "example.com"
      assert session.exit_page_hostname == "example.com"
    end

    test "hostname change" do
      site_id = new_site_id()

      events = [
        build(:event,
          name: "pageview",
          site_id: site_id,
          pathname: "/landing",
          hostname: "example.com",
          timestamp: DateTime.shift(DateTime.utc_now(), second: -5),
          user_id: 1
        ),
        build(:event,
          name: "pageview",
          site_id: site_id,
          pathname: "/post/1",
          hostname: "blog.example.com",
          user_id: 1
        )
      ]

      flush(events)
      session = get_session(site_id)
      assert session.hostname == "example.com"
      assert session.exit_page_hostname == "blog.example.com"
    end

    test "hostname change with custom event in the middle" do
      site_id = new_site_id()

      events = [
        build(:event,
          name: "pageview",
          site_id: site_id,
          pathname: "/landing",
          hostname: "example.com",
          timestamp: DateTime.shift(DateTime.utc_now(), second: -5),
          user_id: 1
        ),
        build(:event,
          name: "custom_event",
          site_id: site_id,
          pathname: "/path/1",
          hostname: "analytics.example.com",
          timestamp: DateTime.shift(DateTime.utc_now(), second: -3),
          user_id: 1
        ),
        build(:event,
          name: "pageview",
          site_id: site_id,
          pathname: "/post/1",
          hostname: "blog.example.com",
          user_id: 1
        )
      ]

      flush(events)
      session = get_session(site_id)
      assert session.hostname == "example.com"
      assert session.exit_page_hostname == "blog.example.com"
    end
  end

  test "initial pageview-specific attributes" do
    site_id = new_site_id()

    event =
      build(:event,
        name: "custom_event",
        site_id: site_id,
        pathname: "/path/1",
        user_id: 1
      )

    flush([event])

    session = get_session(site_id)

    assert session.exit_page == ""
    assert session.events == 1
    assert session.pageviews == 0
  end

  test "updating pageview-specific attributes" do
    site_id = new_site_id()

    event1 =
      build(:event,
        name: "custom_event",
        site_id: site_id,
        pathname: "/path/1",
        user_id: 1
      )

    event2 =
      build(:pageview,
        pathname: "/path/2",
        site_id: site_id,
        user_id: 1
      )

    event3 =
      build(:pageview,
        pathname: "/path/3",
        site_id: site_id,
        user_id: 1
      )

    event4 =
      build(:event,
        name: "custom_event",
        site_id: site_id,
        pathname: "/path/4",
        user_id: 1
      )

    flush([event1, event2, event3, event4])

    session = get_session(site_id)

    assert session.entry_page == "/path/2"
    assert session.exit_page == "/path/3"
    assert session.events == 4
    assert session.pageviews == 2
  end

  test "calculates duration correctly for out-of-order events", %{buffer: buffer} do
    timestamp = DateTime.utc_now()

    event1 =
      build(:event, name: "pageview", timestamp: timestamp |> NaiveDateTime.shift(second: 10))

    event2 = %{event1 | timestamp: timestamp}

    CacheStore.on_event(event1, %{}, nil, buffer_insert: buffer)
    CacheStore.on_event(event2, %{}, nil, buffer_insert: buffer)

    assert_receive({:buffer, :insert, [[_negative_record, session]]})
    assert session.duration == 10
  end

  describe "collapse order" do
    test "across parts" do
      e = build(:event, name: "pageview", site_id: new_site_id())

      flush([%{e | pathname: "/"}])
      flush([%{e | pathname: "/exit"}])

      session_q = from s in Plausible.ClickhouseSessionV2, where: s.site_id == ^e.site_id
      session = Plausible.ClickhouseRepo.one!(session_q, settings: [final: true])

      refute session.is_bounce
      assert session.entry_page == "/"
      assert session.exit_page == "/exit"
    end

    test "within parts" do
      e = build(:event, name: "pageview", site_id: new_site_id())

      flush([
        %{e | pathname: "/"},
        %{e | pathname: "/exit"}
      ])

      session = get_session(e.site_id)

      refute session.is_bounce
      assert session.entry_page == "/"
      assert session.exit_page == "/exit"
    end

    test "across and within parts" do
      e = build(:event, name: "pageview", site_id: new_site_id())

      flush([
        %{e | pathname: "/"},
        %{e | pathname: "/about"}
      ])

      flush([
        %{e | pathname: "/login"},
        %{e | pathname: "/exit"}
      ])

      session = get_session(e.site_id, final: true)

      refute session.is_bounce
      assert session.entry_page == "/"
      assert session.exit_page == "/exit"
      assert session.events == 4
    end
  end

  describe "bounce detection" do
    test "single pageview is bounce" do
      session = create_session([build(:pageview)])
      assert session.is_bounce
    end

    test "two pageviews is not a bounce" do
      session = create_session([build(:pageview), build(:pageview)])
      refute session.is_bounce
    end

    test "(implicitly interactive) custom event is not a bounce" do
      session = create_session([build(:event, name: "interactive")])
      refute session.is_bounce
    end

    test "non-interactive custom event is a bounce" do
      session = create_session([build(:event, name: "non-interactive", interactive?: false)])
      assert session.is_bounce
    end

    test "pageview followed by non-interactive custom event is a bounce" do
      session =
        create_session([
          build(:pageview),
          build(:event, name: "non-interactive", interactive?: false)
        ])

      assert session.is_bounce
    end

    test "non-interactive custom event followed by pageview is a bounce" do
      session =
        create_session([
          build(:event, name: "non-interactive", interactive?: false),
          build(:pageview)
        ])

      assert session.is_bounce
    end

    test "non-interactive custom event followed by interactive custom event is not a bounce" do
      session =
        create_session([
          build(:event, name: "non-interactive", interactive?: false),
          build(:event, name: "interactive")
        ])

      refute session.is_bounce
    end
  end

  defp new_site_id() do
    [[site_id]] =
      Plausible.ClickhouseRepo.query!("select max(site_id) + rand() from sessions_v2 FINAL").rows

    site_id
  end

  defp flush(events) do
    for e <- events, do: CacheStore.on_event(e, %{}, nil)
    Plausible.Session.WriteBuffer.flush()
  end

  defp create_session(events) do
    event_params = %{site_id: new_site_id(), user_id: 1}

    events
    |> Enum.map(&Map.merge(&1, event_params))
    |> flush()

    get_session(event_params.site_id, final: true)
  end

  defp get_session(site_id, opts \\ []) do
    session_q =
      from s in Plausible.ClickhouseSessionV2,
        where: s.site_id == ^site_id,
        order_by: [desc: :events],
        limit: 1

    settings = [final: Keyword.get(opts, :final, true)]

    Plausible.ClickhouseRepo.one!(session_q, settings: settings)
  end
end