Revenue tracking: Ingestion and breakdown queries (#2957)

* Add revenue fields to ClickHouse events

This commit adds 4 fields to the ClickHouse events_v2 table:

* `revenue_source_amount` and `revenue_source_currency` store revenue in
  the original currency sent during ingestion

* `revenue_reporting_amount` and `revenue_reporting_currency` store
  revenue in a common currency to perform calculations, and this
  currency is defined by the user when setting up the goal

The type of amount fields is `Nullable(Decimal64(3))`. That covers all
fiat currencies and allows us to store huge amounts. Even though
ClickHouse does not suggest using `Nullable`, this is a good use case,
because otherwise additional work would have to be done to
differentiate missing values from real zeroes.

I ran a benchmark with the data pattern we expect in production, where
we have more missing values than real decimals. I created 100 million
records where 90% of decimals are missing. The difference between the
tables in storage is just 0.4Mb.

* Add revenue parameter to Events API

This commit adds support for sending revenue data in ingestion using the
`revenue` parameter - aliased to `$`.

* Add revenue parameter to mix send_pageview

* Add average and total revenue to breakdown queries

config/runtime.exs

@@ -335,7 +335,8 @@ config :plausible, Plausible.ImportDeletionRepo,
   pool_size: 1
 
 config :ex_money,
-  open_exchange_rates_app_id: get_var_from_path_or_env(config_dir, "OPEN_EXCHANGE_RATES_APP_ID")
+  open_exchange_rates_app_id: get_var_from_path_or_env(config_dir, "OPEN_EXCHANGE_RATES_APP_ID"),
+  retrieve_every: :timer.hours(24)
 
 case mailer_adapter do
   "Bamboo.PostmarkAdapter" ->

lib/mix/tasks/send_pageview.ex

@@ -25,7 +25,9 @@ defmodule Mix.Tasks.SendPageview do
     referrer: :string,
     host: :string,
     event: :string,
-    props: :string
+    props: :string,
+    revenue_currency: :string,
+    revenue_amount: :string
   ]
 
   def run(opts) do
@@ -85,12 +87,21 @@ defmodule Mix.Tasks.SendPageview do
     event = Keyword.get(opts, :event, @default_event)
     props = Keyword.get(opts, :props, @default_props)
 
+    revenue =
+      if Keyword.get(opts, :revenue_currency) do
+        %{
+          currency: Keyword.get(opts, :revenue_currency),
+          amount: Keyword.get(opts, :revenue_amount)
+        }
+      end
+
     %{
       name: event,
       url: "http://#{domain}#{page}",
       domain: domain,
       referrer: referrer,
-      props: props
+      props: props,
+      revenue: revenue
     }
   end
 

lib/plausible/clickhouse_event_v2.ex

@@ -36,6 +36,12 @@ defmodule Plausible.ClickhouseEventV2 do
 
     field :"meta.key", {:array, :string}
     field :"meta.value", {:array, :string}
+
+    field :revenue_source_amount, Ch, type: "Nullable(Decimal64(3))"
+    field :revenue_source_currency, Ch, type: "FixedString(3)"
+    field :revenue_reporting_amount, Ch, type: "Nullable(Decimal64(3))"
+    field :revenue_reporting_currency, Ch, type: "FixedString(3)"
+
     field :transferred_from, :string
   end
 
@@ -67,7 +73,11 @@ defmodule Plausible.ClickhouseEventV2 do
         :city_geoname_id,
         :screen_size,
         :"meta.key",
-        :"meta.value"
+        :"meta.value",
+        :revenue_source_amount,
+        :revenue_source_currency,
+        :revenue_reporting_amount,
+        :revenue_reporting_currency
       ],
       empty_values: [nil, ""]
     )

lib/plausible/ingestion/event.ex

@@ -98,6 +98,7 @@ defmodule Plausible.Ingestion.Event do
       &put_utm_tags/1,
       &put_geolocation/1,
       &put_props/1,
+      &put_revenue/1,
       &put_salts/1,
       &put_user_id/1,
       &validate_clickhouse_event/1,
@@ -206,6 +207,38 @@ defmodule Plausible.Ingestion.Event do
 
   defp put_props(%__MODULE__{} = event), do: event
 
+  defp put_revenue(%__MODULE__{request: %{revenue_source: %Money{} = revenue_source}} = event) do
+    revenue_goals = Plausible.Site.Cache.get(event.domain).revenue_goals || []
+
+    matching_goal =
+      Enum.find(revenue_goals, &(&1.event_name == event.clickhouse_event_attrs.name))
+
+    cond do
+      is_nil(matching_goal) ->
+        event
+
+      matching_goal.currency == revenue_source.currency ->
+        update_attrs(event, %{
+          revenue_source_amount: Money.to_decimal(revenue_source),
+          revenue_source_currency: to_string(revenue_source.currency),
+          revenue_reporting_amount: Money.to_decimal(revenue_source),
+          revenue_reporting_currency: to_string(revenue_source.currency)
+        })
+
+      matching_goal.currency != revenue_source.currency ->
+        converted = Money.to_currency!(revenue_source, matching_goal.currency)
+
+        update_attrs(event, %{
+          revenue_source_amount: Money.to_decimal(revenue_source),
+          revenue_source_currency: to_string(revenue_source.currency),
+          revenue_reporting_amount: Money.to_decimal(converted),
+          revenue_reporting_currency: to_string(converted.currency)
+        })
+    end
+  end
+
+  defp put_revenue(event), do: event
+
   defp put_salts(%__MODULE__{} = event) do
     %{event | salts: Plausible.Session.Salts.fetch()}
   end

lib/plausible/ingestion/request.ex

@@ -38,6 +38,7 @@ defmodule Plausible.Ingestion.Request do
     field :hash_mode, :integer
     field :pathname, :string
     field :props, :map
+    field :revenue_source, :map
     field :query_params, :map
 
     field :timestamp, :naive_datetime
@@ -70,6 +71,7 @@ defmodule Plausible.Ingestion.Request do
         |> put_props(request_body)
         |> put_pathname()
         |> put_query_params()
+        |> put_revenue_source(request_body)
         |> map_domains(request_body)
         |> Changeset.validate_required([
           :event_name,
@@ -187,7 +189,7 @@ defmodule Plausible.Ingestion.Request do
   defp put_props(changeset, %{} = request_body) do
     props =
       (request_body["m"] || request_body["meta"] || request_body["p"] || request_body["props"])
-      |> decode_props_or_fallback()
+      |> decode_json_or_fallback()
       |> Enum.reject(fn {_k, v} -> is_nil(v) || is_list(v) || is_map(v) || v == "" end)
       |> Enum.take(@max_props)
       |> Map.new()
@@ -197,18 +199,6 @@ defmodule Plausible.Ingestion.Request do
     |> validate_props()
   end
 
-  defp decode_props_or_fallback(raw) do
-    with raw when is_binary(raw) <- raw,
-         {:ok, %{} = decoded} <- Jason.decode(raw) do
-      decoded
-    else
-      already_a_map when is_map(already_a_map) -> already_a_map
-      {:ok, _list_or_other} -> %{}
-      {:error, _decode_error} -> %{}
-      _any -> %{}
-    end
-  end
-
   @max_prop_key_length 300
   @max_prop_value_length 2000
   defp validate_props(changeset) do
@@ -231,6 +221,46 @@ defmodule Plausible.Ingestion.Request do
     end
   end
 
+  defp put_revenue_source(%Ecto.Changeset{} = changeset, %{} = request_body) do
+    with revenue_source <- request_body["revenue"] || request_body["$"],
+         %{"amount" => _, "currency" => _} = revenue_source <-
+           decode_json_or_fallback(revenue_source) do
+      parse_revenue_source(changeset, revenue_source)
+    else
+      _any -> changeset
+    end
+  end
+
+  @valid_currencies Plausible.Goal.valid_currencies()
+  defp parse_revenue_source(changeset, %{"amount" => amount, "currency" => currency}) do
+    with true <- currency in @valid_currencies,
+         {%Decimal{} = amount, _rest} <- parse_decimal(amount),
+         %Money{} = amount <- Money.new(currency, amount) do
+      Changeset.put_change(changeset, :revenue_source, amount)
+    else
+      _any -> changeset
+    end
+  end
+
+  defp decode_json_or_fallback(raw) do
+    with raw when is_binary(raw) <- raw,
+         {:ok, %{} = decoded} <- Jason.decode(raw) do
+      decoded
+    else
+      already_a_map when is_map(already_a_map) -> already_a_map
+      _any -> %{}
+    end
+  end
+
+  defp parse_decimal(value) do
+    case value do
+      value when is_binary(value) -> Decimal.parse(value)
+      value when is_float(value) -> {Decimal.from_float(value), nil}
+      value when is_integer(value) -> {Decimal.new(value), nil}
+      _any -> :error
+    end
+  end
+
   defp put_query_params(changeset) do
     case Changeset.get_field(changeset, :uri) do
       %{query: query} when is_binary(query) ->

lib/plausible/stats/base.ex

@@ -277,6 +277,26 @@ defmodule Plausible.Stats.Base do
     |> select_event_metrics(rest)
   end
 
+  def select_event_metrics(q, [:total_revenue | rest]) do
+    from(e in q,
+      select_merge: %{
+        total_revenue:
+          fragment("toDecimal64(sum(?) * any(_sample_factor), 3)", e.revenue_reporting_amount)
+      }
+    )
+    |> select_event_metrics(rest)
+  end
+
+  def select_event_metrics(q, [:average_revenue | rest]) do
+    from(e in q,
+      select_merge: %{
+        average_revenue:
+          fragment("toDecimal64(avg(?) * any(_sample_factor), 3)", e.revenue_reporting_amount)
+      }
+    )
+    |> select_event_metrics(rest)
+  end
+
   def select_event_metrics(q, [:sample_percent | rest]) do
     from(e in q,
       select_merge: %{

lib/plausible/stats/breakdown.ex

@@ -17,15 +17,25 @@ defmodule Plausible.Stats.Breakdown do
       |> Goals.for_site()
       |> Enum.split_with(fn goal -> goal.event_name end)
 
+    revenue_goals = Enum.filter(event_goals, &Plausible.Goal.revenue?/1)
+
     events = Enum.map(event_goals, & &1.event_name)
     event_query = %Query{query | filters: Map.put(query.filters, "event:name", {:member, events})}
 
     trace(query, property, metrics)
 
+    metrics =
+      if Enum.empty?(revenue_goals) do
+        metrics
+      else
+        metrics ++ [:average_revenue, :total_revenue]
+      end
+
     event_results =
       if Enum.any?(event_goals) do
         breakdown(site, event_query, "event:name", metrics, pagination)
         |> transform_keys(%{name: :goal})
+        |> cast_revenue_metrics_to_money(revenue_goals)
       else
         []
       end
@@ -156,6 +166,20 @@ defmodule Plausible.Stats.Breakdown do
     breakdown_sessions(site, query, property, metrics, pagination)
   end
 
+  defp cast_revenue_metrics_to_money(event_results, revenue_goals) do
+    for result <- event_results do
+      matching_goal = Enum.find(revenue_goals, &(&1.event_name == result.goal))
+
+      if matching_goal && result.total_revenue && result.average_revenue do
+        result
+        |> Map.put(:total_revenue, Money.new!(matching_goal.currency, result.total_revenue))
+        |> Map.put(:average_revenue, Money.new!(matching_goal.currency, result.average_revenue))
+      else
+        result
+      end
+    end
+  end
+
   defp zip_results(event_result, session_result, property, metrics) do
     null_row = Enum.map(metrics, fn metric -> {metric, nil} end) |> Enum.into(%{})
 

lib/plausible_web/controllers/api/stats_controller.ex

@@ -1058,6 +1058,7 @@ defmodule PlausibleWeb.Api.StatsController do
         goal
         |> Map.put(:prop_names, CustomProps.props_for_goal(site, query))
         |> Map.put(:conversion_rate, calculate_cr(total_visitors, goal[:unique_conversions]))
+        |> format_revenue_metrics()
       end)
 
     if params["csv"] do
@@ -1067,6 +1068,22 @@ defmodule PlausibleWeb.Api.StatsController do
     end
   end
 
+  defp format_revenue_metrics(%{average_revenue: %Money{}, total_revenue: %Money{}} = results) do
+    %{
+      results
+      | average_revenue: %{
+          short: Money.to_string!(results.average_revenue, format: :short, fractional_digits: 1),
+          long: Money.to_string!(results.average_revenue)
+        },
+        total_revenue: %{
+          short: Money.to_string!(results.total_revenue, format: :short, fractional_digits: 1),
+          long: Money.to_string!(results.total_revenue)
+        }
+    }
+  end
+
+  defp format_revenue_metrics(results), do: results
+
   def prop_breakdown(conn, params) do
     site = conn.assigns[:site]
     query = Query.from(site, params) |> Filters.add_prefix()

mix.exs

@@ -61,7 +61,7 @@ defmodule Plausible.MixProject do
       {:bcrypt_elixir, "~> 3.0"},
       {:bypass, "~> 2.1", only: [:dev, :test]},
       {:cachex, "~> 3.4"},
-      {:ecto_ch, "~> 0.1.0"},
+      {:ecto_ch, "~> 0.1.10"},
       {:combination, "~> 0.0.3"},
       {:connection, "~> 1.1", override: true},
       {:cors_plug, "~> 3.0"},

mix.lock

@@ -31,7 +31,7 @@
   "double": {:hex, :double, "0.8.2", "8e1cfcccdaef76c18846bc08e555555a2a699b806fa207b6468572a60513cc6a", [:mix], [], "hexpm", "90287642b2ec86125e0457aaba2ab0e80f7d7050cc80a0cef733e59bd70aa67c"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.31", "a93921cdc6b9b869f519213d5bc79d9e218ba768d7270d46fdcf1c01bacff9e2", [:mix], [], "hexpm", "317d367ee0335ef037a87e46c91a2269fef6306413f731e8ec11fc45a7efd059"},
   "ecto": {:hex, :ecto, "3.9.5", "9f0aa7ae44a1577b651c98791c6988cd1b69b21bc724e3fd67090b97f7604263", [:mix], [{:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "d4f3115d8cbacdc0bfa4b742865459fb1371d0715515842a1fb17fe31920b74c"},
-  "ecto_ch": {:hex, :ecto_ch, "0.1.9", "cb8e7bbe926c73d4160f4a1d9153a7bca5b757678b648ff2d8baf7c0dad11200", [:mix], [{:ch, "~> 0.1.14", [hex: :ch, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.9", [hex: :ecto_sql, repo: "hexpm", optional: false]}], "hexpm", "94c125d335581208c64b76a37d98961393f81131f8ef2a57ea4c4e66bbe42753"},
+  "ecto_ch": {:hex, :ecto_ch, "0.1.10", "72d21b2395cde46a242abf0d16163289bcd5a99d0bcbcb371025a99cad049d99", [:mix], [{:ch, "~> 0.1.14", [hex: :ch, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.9", [hex: :ecto_sql, repo: "hexpm", optional: false]}], "hexpm", "a365f65856d59eb1f5c04f19d27bf9e6855c1ce8e4dc57ed09bb186bda491f81"},
   "ecto_sql": {:hex, :ecto_sql, "3.9.2", "34227501abe92dba10d9c3495ab6770e75e79b836d114c41108a4bf2ce200ad5", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.9.2", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.6.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.16.0 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "1eb5eeb4358fdbcd42eac11c1fbd87e3affd7904e639d77903c1358b2abd3f70"},
   "elixir_make": {:hex, :elixir_make, "0.7.6", "67716309dc5d43e16b5abbd00c01b8df6a0c2ab54a8f595468035a50189f9169", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}], "hexpm", "5a0569756b0f7873a77687800c164cca6dfc03a09418e6fcf853d78991f49940"},
   "envy": {:hex, :envy, "1.1.1", "0bc9bd654dec24fcdf203f7c5aa1b8f30620f12cfb28c589d5e9c38fe1b07475", [:mix], [], "hexpm", "7061eb1a47415fd757145d8dec10dc0b1e48344960265cb108f194c4252c3a89"},

test/plausible/ingestion/event_test.exs

@@ -99,4 +99,37 @@ defmodule Plausible.Ingestion.EventTest do
     assert {:ok, %{buffered: [], dropped: [dropped]}} = Event.build_and_buffer(request)
     assert dropped.drop_reason == :throttle
   end
+
+  test "saves revenue amount" do
+    site = insert(:site)
+    _goal = insert(:goal, event_name: "checkout", currency: "USD", site: site)
+
+    payload = %{
+      name: "checkout",
+      url: "http://#{site.domain}",
+      revenue: %{amount: 10.2, currency: "USD"}
+    }
+
+    conn = build_conn(:post, "/api/events", payload)
+    assert {:ok, request} = Request.build(conn)
+
+    assert {:ok, %{buffered: [event], dropped: []}} = Event.build_and_buffer(request)
+    assert Decimal.eq?(event.clickhouse_event.revenue_source_amount, Decimal.new("10.2"))
+  end
+
+  test "does not save revenue amount when there is no revenue goal" do
+    site = insert(:site)
+
+    payload = %{
+      name: "checkout",
+      url: "http://#{site.domain}",
+      revenue: %{amount: 10.2, currency: "USD"}
+    }
+
+    conn = build_conn(:post, "/api/events", payload)
+    assert {:ok, request} = Request.build(conn)
+
+    assert {:ok, %{buffered: [event], dropped: []}} = Event.build_and_buffer(request)
+    assert event.clickhouse_event.revenue_source_amount == nil
+  end
 end