CRM: Trial prospects worker

config/runtime.exs

@@ -846,7 +846,9 @@ cloud_cron = [
   # First sunday of the month, 4:00 UTC
   {"0 4 1-7 * SUN", Plausible.Workers.ClickhouseCleanSites},
   # Daily at 4:00 UTC
-  {"0 4 * * *", Plausible.Workers.SetLegacyTimeOnPageCutoff}
+  {"0 4 * * *", Plausible.Workers.SetLegacyTimeOnPageCutoff},
+  # Daily at 2:00 UTC
+  {"0 2 * * *", Plausible.Workers.ScoreTrialProspects}
 ]
 
 crontab = if(is_selfhost, do: base_cron, else: base_cron ++ cloud_cron)
@@ -876,7 +878,8 @@ cloud_queues = [
   lock_sites: 1,
   legacy_time_on_page_cutoff: 1,
   purge_cdn_cache: 1,
-  sso_domain_ownership_verification: 32
+  sso_domain_ownership_verification: 32,
+  score_trial_prospects: 1
 ]
 
 queues = if(is_selfhost, do: base_queues, else: base_queues ++ cloud_queues)

extra/lib/plausible/customer_support/trial_prospect.ex

@@ -0,0 +1,54 @@
+defmodule Plausible.CustomerSupport.TrialProspect do
+  @moduledoc """
+  Cached revenue-potential scoring for a team on (or just off) a trial.
+  Rows are (re)computed daily by `Plausible.Workers.ScoreTrialProspects` and
+  read by the customer support UI.
+  """
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  @type t() :: %__MODULE__{}
+
+  @fields [
+    :team_id,
+    :estimated_monthly,
+    :observed_days,
+    :first_data_day,
+    :kind,
+    :forced_by,
+    :pageview_limit,
+    :over_top_tier,
+    :estimated_mrr,
+    :computed_at
+  ]
+
+  schema "trial_prospects" do
+    belongs_to :team, Plausible.Teams.Team
+
+    field :estimated_monthly, :integer
+    field :observed_days, :integer
+    field :first_data_day, :date
+    field :kind, Ecto.Enum, values: [:starter, :growth, :business]
+    field :forced_by, {:array, :string}, default: []
+    field :pageview_limit, :integer
+    field :over_top_tier, :boolean, default: false
+    field :estimated_mrr, :integer
+    field :computed_at, :utc_datetime
+
+    timestamps()
+  end
+
+  def changeset(prospect, attrs) do
+    prospect
+    |> cast(attrs, @fields)
+    |> validate_required([
+      :team_id,
+      :estimated_monthly,
+      :observed_days,
+      :first_data_day,
+      :kind,
+      :computed_at
+    ])
+    |> unique_constraint(:team_id)
+  end
+end

extra/lib/plausible/customer_support/trial_prospects.ex

@@ -0,0 +1,131 @@
+defmodule Plausible.CustomerSupport.TrialProspects do
+  @moduledoc """
+  Pure scoring logic for ranking trial teams by revenue potential. Turns a
+  team's partial traffic sample and premium-feature usage into an estimated MRR.
+  """
+
+  # Static reference pricing (EUR/mo) + pageview ladder. Hand-maintained; keep in sync when public pricing changes.
+  @pricing_path Application.app_dir(:plausible, ["priv", "trial_prospect_pricing.json"])
+  @external_resource @pricing_path
+  @tiers @pricing_path
+         |> File.read!()
+         |> Jason.decode!()
+         |> Map.fetch!("tiers")
+         |> Enum.map(fn tier ->
+           %{
+             limit: tier["monthly_pageview_limit"],
+             starter: tier["starter"],
+             growth: tier["growth"],
+             business: tier["business"]
+           }
+         end)
+         |> Enum.sort_by(& &1.limit)
+  @ladder Enum.map(@tiers, & &1.limit)
+
+  @business_features [
+    :props,
+    :funnels,
+    :revenue_goals,
+    :stats_api
+  ]
+  @growth_features [:shared_links, :site_segments]
+
+  @starter_site_limit 1
+  @growth_site_limit 3
+  @starter_member_limit 0
+  @growth_member_limit 3
+
+  @kind_rank %{starter: 0, growth: 1, business: 2}
+
+  @doc """
+  Combines feature usage, site/member counts + a monthly estimate into the
+  persisted scoring fields: `kind`, `forced_by`, `pageview_limit`,
+  `over_top_tier`, `estimated_mrr`.
+  """
+  @spec score(non_neg_integer(), [module()], non_neg_integer(), non_neg_integer()) :: %{
+          kind: :starter | :growth | :business,
+          forced_by: [String.t()],
+          pageview_limit: pos_integer() | nil,
+          over_top_tier: boolean(),
+          estimated_mrr: non_neg_integer() | nil
+        }
+  def score(estimated_monthly, feature_modules, site_count, member_count) do
+    {kind, forced_by} = plan_kind(feature_modules, site_count, member_count)
+    {limit, over_top_tier} = pageview_rung(estimated_monthly)
+
+    %{
+      kind: kind,
+      forced_by: forced_by,
+      pageview_limit: limit,
+      over_top_tier: over_top_tier,
+      estimated_mrr: estimated_mrr(kind, limit, over_top_tier)
+    }
+  end
+
+  @spec plan_kind([module()], non_neg_integer(), non_neg_integer()) ::
+          {:starter | :growth | :business, [String.t()]}
+  defp plan_kind(feature_modules, site_count, member_count) do
+    used = Enum.map(feature_modules, & &1.name())
+
+    feature_k = feature_kind(used)
+    site_k = site_count_kind(site_count)
+    member_k = member_count_kind(member_count)
+
+    kind = feature_k |> higher_kind(site_k) |> higher_kind(member_k)
+
+    {kind, forced_by(used, kind, site_k, member_k)}
+  end
+
+  @spec pageview_rung(non_neg_integer()) :: {pos_integer() | nil, boolean()}
+  defp pageview_rung(estimated_monthly) do
+    case Enum.find(@ladder, &(&1 >= estimated_monthly)) do
+      nil -> {nil, true}
+      limit -> {limit, false}
+    end
+  end
+
+  @spec estimated_mrr(:starter | :growth | :business, pos_integer() | nil, boolean()) ::
+          non_neg_integer() | nil
+  defp estimated_mrr(_kind, _limit, true), do: nil
+
+  defp estimated_mrr(kind, limit, false) do
+    tier = Enum.find(@tiers, &(&1.limit == limit))
+    tier && Map.fetch!(tier, kind)
+  end
+
+  defp feature_kind(used) do
+    cond do
+      forcing(used, @business_features) != [] -> :business
+      forcing(used, @growth_features) != [] -> :growth
+      true -> :starter
+    end
+  end
+
+  defp site_count_kind(n) when n <= @starter_site_limit, do: :starter
+  defp site_count_kind(n) when n <= @growth_site_limit, do: :growth
+  defp site_count_kind(_), do: :business
+
+  defp member_count_kind(n) when n <= @starter_member_limit, do: :starter
+  defp member_count_kind(n) when n <= @growth_member_limit, do: :growth
+  defp member_count_kind(_), do: :business
+
+  defp higher_kind(a, b), do: if(@kind_rank[a] >= @kind_rank[b], do: a, else: b)
+
+  defp forced_by(_used, :starter, _site_k, _member_k), do: []
+
+  defp forced_by(used, kind, site_k, member_k) do
+    forcing_for(used, kind) ++
+      List.wrap(if(site_k == kind, do: "site_limit")) ++
+      List.wrap(if(member_k == kind, do: "team_member_limit"))
+  end
+
+  defp forcing_for(used, :business), do: forcing(used, @business_features)
+  defp forcing_for(used, :growth), do: forcing(used, @growth_features)
+
+  defp forcing(used, tier_features) do
+    used
+    |> Enum.filter(&(&1 in tier_features))
+    |> Enum.map(&Atom.to_string/1)
+    |> Enum.sort()
+  end
+end

extra/lib/plausible/workers/score_trial_prospects.ex

@@ -0,0 +1,91 @@
+defmodule Plausible.Workers.ScoreTrialProspects do
+  @moduledoc """
+  Daily worker that scores the revenue potential of every team on (or just off)
+  a trial and caches the result in the `trial_prospects` table.
+  """
+  use Plausible.Repo
+
+  use Oban.Worker,
+    queue: :score_trial_prospects,
+    max_attempts: 1
+
+  alias Plausible.Teams
+  alias Plausible.CustomerSupport.{TrialProspect, TrialProspects}
+
+  @max_expired_days 30
+
+  @impl Oban.Worker
+  def perform(_job) do
+    rows =
+      Date.utc_today()
+      |> trial_population()
+      |> Enum.flat_map(&score_team/1)
+
+    persist(rows)
+    :ok
+  end
+
+  defp trial_population(today) do
+    cutoff = Date.add(today, -@max_expired_days)
+
+    Repo.all(
+      from t in Teams.Team,
+        left_join: s in assoc(t, :subscription),
+        where: not is_nil(t.trial_expiry_date),
+        where: is_nil(s.id),
+        where: t.trial_expiry_date >= ^cutoff
+    )
+  end
+
+  defp score_team(team) do
+    site_ids = Teams.owned_sites_ids(team)
+
+    traffic = Plausible.Stats.Clickhouse.trial_traffic(site_ids)
+
+    case traffic do
+      %{events_in_window: 0} -> []
+      traffic -> [build_row(team, traffic, site_ids)]
+    end
+  end
+
+  defp build_row(team, traffic, site_ids) do
+    score =
+      TrialProspects.score(
+        traffic.estimated_monthly,
+        Teams.Billing.features_usage(team, site_ids),
+        length(site_ids),
+        Teams.Billing.team_member_usage(team)
+      )
+
+    now = DateTime.utc_now() |> DateTime.truncate(:second)
+    naive_now = DateTime.to_naive(now)
+
+    %{
+      team_id: team.id,
+      estimated_monthly: traffic.estimated_monthly,
+      observed_days: traffic.observed_days,
+      first_data_day: traffic.first_data_day,
+      kind: score.kind,
+      forced_by: score.forced_by,
+      pageview_limit: score.pageview_limit,
+      over_top_tier: score.over_top_tier,
+      estimated_mrr: score.estimated_mrr,
+      computed_at: now,
+      inserted_at: naive_now,
+      updated_at: naive_now
+    }
+  end
+
+  defp persist(rows) do
+    scored_team_ids = Enum.map(rows, & &1.team_id)
+
+    Repo.transaction(fn ->
+      Repo.insert_all(TrialProspect, rows,
+        on_conflict: {:replace_all_except, [:id, :team_id, :inserted_at]},
+        conflict_target: :team_id
+      )
+
+      Repo.delete_all(from p in TrialProspect, where: p.team_id not in ^scored_team_ids)
+    end)
+  end
+end

lib/plausible/stats/clickhouse.ex

@@ -1,10 +1,14 @@
 defmodule Plausible.Stats.Clickhouse do
+  @moduledoc """
+    Clickhouse utility functions
+  """
+
   use Plausible
   use Plausible.Repo
   use Plausible.ClickhouseRepo
   use Plausible.Stats.SQL.Fragments
 
-  import Ecto.Query, only: [from: 2]
+  import Ecto.Query, only: [from: 2, subquery: 1]
 
   alias Plausible.Timezones
   alias Plausible.Stats
@@ -76,6 +80,76 @@
 
   def usage_breakdown([], _date_range), do: {0, 0}
 
+  @doc """
+  Estimates a team's monthly traffic from a partial sample, computed entirely in
+  ClickHouse with one light query scoped to the team's own sites. Used by
+  `Plausible.Workers.ScoreTrialProspects`.
+
+  Billable events = pageviews + custom events (everything except engagement).
+  The window is the trailing `@window_days` complete days (up to yesterday).
+  Within it, `min(toDate(timestamp))` is the team's `first_data_day`, so the
+  elapsed complete days since then give `observed_days` (clamped to
+  `[1, @window_days]`), and the monthly figure is
+  `round(events_in_window / observed_days * @days_in_month)` — all in the DB.
+
+  Returns `%{first_data_day, events_in_window, observed_days, estimated_monthly}`.
+  `events_in_window == 0` means the team had no complete day of traffic.
+  """
+  # trailing complete days sampled to estimate the monthly run rate
+  @window_days 30
+  @days_in_month 30
+  @spec trial_traffic([pos_integer()]) :: %{
+          first_data_day: Date.t() | nil,
+          events_in_window: non_neg_integer(),
+          observed_days: non_neg_integer(),
+          estimated_monthly: non_neg_integer()
+        }
+  def trial_traffic([]) do
+    %{first_data_day: nil, events_in_window: 0, observed_days: 0, estimated_monthly: 0}
+  end
+
+  def trial_traffic(site_ids) do
+    last_complete_day = Date.add(Date.utc_today(), -1)
+    first_day = Date.add(last_complete_day, -(@window_days - 1))
+
+    aggregates =
+      from(e in "events_v2",
+        where: e.site_id in ^site_ids,
+        where: e.name != "engagement",
+        where: fragment("toDate(?)", e.timestamp) >= ^first_day,
+        where: fragment("toDate(?)", e.timestamp) <= ^last_complete_day,
+        select: %{
+          first_data_day: fragment("min(toDate(?))", e.timestamp),
+          events_in_window: fragment("count(*)"),
+          # elapsed complete days since first traffic, clamped to [1, @window_days]
+          observed_days:
+            fragment(
+              "least(greatest(dateDiff('day', min(toDate(?)), ?) + 1, 1), ?)",
+              e.timestamp,
+              ^last_complete_day,
+              ^@window_days
+            )
+        }
+      )
+
+    ClickhouseRepo.one(
+      from(t in subquery(aggregates),
+        select: %{
+          first_data_day: t.first_data_day,
+          events_in_window: t.events_in_window,
+          observed_days: t.observed_days,
+          estimated_monthly:
+            fragment(
+              "toUInt64(round(? / ? * ?))",
+              t.events_in_window,
+              t.observed_days,
+              ^@days_in_month
+            )
+        }
+      )
+    )
+  end
+
   def per_site_usage_breakdown(
         site_ids,
         date_range,