Add service usage recording (#1270)

.tool-versions

@@ -1,2 +1,2 @@
-erlang 24.3.4.13
+erlang 24.3.4.14
 elixir 1.12.3

apps/core/lib/core/schema/cluster.ex

@@ -12,13 +12,14 @@ defmodule Core.Schema.Cluster do
   @expiry 1
 
   schema "clusters" do
-    field :provider,    Provider
-    field :name,        :string
-    field :console_url, :string
-    field :source,      Source
-    field :git_url,     :string
-    field :domain,      :string
-    field :pinged_at,   :utc_datetime_usec
+    field :provider,      Provider
+    field :name,          :string
+    field :console_url,   :string
+    field :source,        Source
+    field :git_url,       :string
+    field :domain,        :string
+    field :pinged_at,     :utc_datetime_usec
+    field :service_count, :integer
 
     belongs_to :owner,   User
     belongs_to :account, Account
@@ -80,7 +81,7 @@ defmodule Core.Schema.Cluster do
     from(c in query, order_by: ^order)
   end
 
-  @valid ~w(owner_id account_id provider name domain console_url source git_url pinged_at)a
+  @valid ~w(owner_id account_id provider name domain console_url source git_url pinged_at service_count)a
 
   def changeset(model, attrs \\ %{}) do
     model

apps/core/lib/core/schema/cluster_usage_history.ex

@@ -3,8 +3,9 @@ defmodule Core.Schema.ClusterUsageHistory do
   alias Core.Schema.{Account, Cluster}
 
   schema "cluster_usage_history" do
-    field :cpu,    :integer
-    field :memory, :integer
+    field :cpu,      :integer
+    field :memory,   :integer
+    field :services, :integer
 
     belongs_to :cluster, Cluster
     belongs_to :account, Account
@@ -28,7 +29,7 @@ defmodule Core.Schema.ClusterUsageHistory do
     from(u in query, order_by: ^order)
   end
 
-  @valid ~w(cpu memory cluster_id account_id)a
+  @valid ~w(cpu memory cluster_id services account_id)a
 
   def changeset(model, attrs \\ %{}) do
     model

apps/core/lib/core/schema/platform_plan.ex

@@ -26,13 +26,14 @@ defmodule Core.Schema.PlatformPlan do
   end
 
   schema "platform_plans" do
-    field :name,        :string
-    field :visible,     :boolean, default: true
-    field :trial,       :boolean, default: false
-    field :enterprise,  :boolean
-    field :cost,        :integer
-    field :period,      Period
-    field :external_id, :string
+    field :name,         :string
+    field :visible,      :boolean, default: true
+    field :trial,        :boolean, default: false
+    field :enterprise,   :boolean
+    field :cost,         :integer
+    field :period,       Period
+    field :external_id,  :string
+    field :service_plan, :string
 
     embeds_one :features, Features, on_replace: :update do
       boolean_fields [:vpn, :user_management, :audit, :multi_cluster, :database_management, :cd]
@@ -60,7 +61,7 @@ defmodule Core.Schema.PlatformPlan do
 
   def features(), do: __MODULE__.Features.__schema__(:fields) -- [:id]
 
-  @valid ~w(name visible cost period external_id trial)a
+  @valid ~w(name visible cost period external_id trial service_plan)a
 
   def changeset(schema, attrs \\ %{}) do
     schema

apps/core/lib/core/schema/platform_subscription.ex

@@ -26,8 +26,9 @@ defmodule Core.Schema.PlatformSubscription do
   end
 
   schema "platform_subscriptions" do
-    field :status, Status
-    field :external_id, :string
+    field :status,       Status
+    field :external_id,  :string
+    field :metered_id,   :string
 
     embeds_many :line_items, LineItem, on_replace: :delete
     belongs_to  :account,    Account
@@ -70,7 +71,7 @@ defmodule Core.Schema.PlatformSubscription do
     |> unique_constraint(:account_id)
   end
 
-  @stripe_valid ~w(external_id status)a
+  @stripe_valid ~w(external_id metered_id status)a
 
   def stripe_changeset(schema, attrs \\ %{}) do
     schema

apps/core/lib/core/services/clusters.ex

@@ -76,9 +76,20 @@ defmodule Core.Services.Clusters do
   """
   @spec save_usage(map, Cluster.t) :: {:ok, ClusterUsageHistory.t} | error
   def save_usage(attrs, %Cluster{id: id, account_id: account_id}) do
-    %ClusterUsageHistory{cluster_id: id, account_id: account_id}
-    |> ClusterUsageHistory.changeset(attrs)
-    |> Core.Repo.insert()
+    start_transaction()
+    |> add_operation(:usage, fn _ ->
+      %ClusterUsageHistory{cluster_id: id, account_id: account_id}
+      |> ClusterUsageHistory.changeset(attrs)
+      |> Core.Repo.insert()
+    end)
+    |> add_operation(:cluster, fn
+      %{usage: %ClusterUsageHistory{services: svcs}} when is_integer(svcs) and svcs > 0 ->
+        get_cluster!(id)
+        |> Cluster.changeset(%{service_count: svcs})
+        |> Core.Repo.update()
+      _ -> {:ok, nil}
+    end)
+    |> execute(extract: :usage)
   end
 
   @doc """

apps/core/lib/core/services/payments.ex

@@ -541,7 +541,7 @@ defmodule Core.Services.Payments do
       |> execute()
     end)
     |> add_operation(:finalized, fn %{db: %{line_items: items} = db, stripe: %{base: %{id: id}} = stripe} ->
-      rest = Map.delete(stripe, :base)
+      rest = Map.drop(stripe, [:base])
 
       items = Enum.map(items, fn %{dimension: dimension} = item ->
         Piazza.Ecto.Schema.mapify(item)
@@ -762,7 +762,7 @@ defmodule Core.Services.Payments do
     |> add_operation(:stripe, fn %{account: %Account{billing_customer_id: cust_id}, db: db} ->
       Stripe.Subscription.create(%{
         customer: cust_id,
-        items: sub_line_items(plan, db),
+        items: sub_line_items(plan, db) ++ metered_plan(plan),
         payment_behavior: "default_incomplete",
         off_session: true
       })
@@ -772,9 +772,13 @@ defmodule Core.Services.Payments do
         stripe: %{id: sub_id, items: %{data: items}},
         db: subscription
       } ->
+      metered = Enum.find(items, & &1.plan.id == plan.service_plan)
+      items = Enum.filter(items, & &1.plan.id != plan.service_plan)
+
       subscription
       |> PlatformSubscription.stripe_changeset(%{
         external_id: sub_id,
+        metered_id: metered && metered.id,
         line_items: rebuild_line_items(subscription, plan, items)
       })
       |> Core.Repo.update()
@@ -785,6 +789,9 @@ defmodule Core.Services.Payments do
   def create_platform_subscription(attrs, plan_id, %User{} = user),
     do: create_platform_subscription(attrs, get_platform_plan!(plan_id), user)
 
+  defp metered_plan(%PlatformPlan{service_plan: p}) when is_binary(p), do: [%{plan: p}]
+  defp metered_plan(_), do: []
+
 
   defp provision_customer(%Account{billing_customer_id: nil} = account, args) do
     account = Core.Repo.preload(account, [:root_user])
@@ -1105,17 +1112,26 @@ defmodule Core.Services.Payments do
     end)
     |> add_operation(:stripe, fn %{db: updated} ->
       Stripe.Subscription.update(sub.external_id, %{
-        items: old_items(sub) ++ sub_line_items(updated.plan, updated)
+        items: (
+          old_items(sub) ++
+          (if sub.metered_id, do: [%{id: sub.metered_id, deleted: true}], else: []) ++
+          sub_line_items(updated.plan, updated) ++
+          metered_plan(plan)
+        )
       })
     end)
     |> add_operation(:finalized, fn
       %{
         stripe: %{id: sub_id, items: %{data: items}},
         db: subscription
       } ->
+      metered = Enum.find(items, & &1.plan.id == plan.service_plan)
+      items = Enum.filter(items, & &1.plan.id != plan.service_plan)
+
       subscription
       |> PlatformSubscription.stripe_changeset(%{
         external_id: sub_id,
+        metered_id: metered && metered.id,
         line_items: rebuild_line_items(subscription, subscription.plan, items)
       })
       |> Core.Repo.update()

apps/core/priv/repo/migrations/20231107002129_service_usage.exs

@@ -0,0 +1,13 @@
+defmodule Core.Repo.Migrations.ServiceUsage do
+  use Ecto.Migration
+
+  def change do
+    alter table(:clusters) do
+      add :service_count, :integer, default: 0
+    end
+
+    alter table(:cluster_usage_history) do
+      add :services, :integer, default: 0
+    end
+  end
+end

apps/core/priv/repo/migrations/20231107013705_add_service_product.exs

@@ -0,0 +1,13 @@
+defmodule Core.Repo.Migrations.AddServiceProduct do
+  use Ecto.Migration
+
+  def change do
+    alter table(:platform_plans) do
+      add :service_plan, :string
+    end
+
+    alter table(:platform_subscriptions) do
+      add :metered_id, :string
+    end
+  end
+end

apps/core/test/services/payments_test.exs

@@ -243,6 +243,7 @@ defmodule Core.Services.PaymentsTest do
       user = insert(:user, roles: %{admin: true}, account: account)
       plan = insert(:platform_plan,
         external_id: "plan_id",
+        service_plan: "service_plan",
         line_items: [
           %{name: "user", dimension: :user, external_id: "id_user", period: :monthly},
           %{name: "cluster", dimension: :cluster, external_id: "id_cluster", period: :monthly},
@@ -251,12 +252,13 @@ defmodule Core.Services.PaymentsTest do
 
       expect(Stripe.Subscription, :create, fn %{
         customer: "cus_id",
-        items: [%{plan: "id_user", quantity: 2}, %{plan: "id_cluster", quantity: 0}]
+        items: [%{plan: "id_user", quantity: 2}, %{plan: "id_cluster", quantity: 0}, %{plan: "service_plan"}]
       } ->
         {:ok, %{
           id: "sub_id",
           items: %{
             data: [
+              %{id: "metered_id", plan: %{id: "service_plan"}},
               %{id: "user_id", plan: %{id: "id_user"}},
               %{id: "cluster_id", plan: %{id: "id_cluster"}}
             ]
@@ -267,6 +269,7 @@ defmodule Core.Services.PaymentsTest do
       {:ok, subscription} = Payments.create_platform_subscription(%{}, plan, user)
 
       assert subscription.plan_id == plan.id
+      assert subscription.metered_id == "metered_id"
       assert subscription.account_id == user.account_id
       assert subscription.external_id == "sub_id"
 
@@ -647,11 +650,13 @@ defmodule Core.Services.PaymentsTest do
         "sub_id",
         %{items: [
           %{id: "si_1", deleted: true},
-          %{plan: "id_user", quantity: 1}
+          %{plan: "id_user", quantity: 1},
+          %{plan: "service_plan"}
         ]} -> {:ok, %{
           id: "sub_id",
           items: %{
             data: [
+              %{id: "metered_id", plan: %{id: "service_plan"}},
               %{id: "user_id", plan: %{id: "id_user"}}
             ]
           }
@@ -661,6 +666,7 @@ defmodule Core.Services.PaymentsTest do
       user = insert(:user, roles: %{admin: true})
       plan = insert(:platform_plan,
         external_id: "pl_id2",
+        service_plan: "service_plan",
         line_items: [
           %{name: "user", dimension: :user, external_id: "id_user", period: :monthly},
         ]
@@ -685,6 +691,8 @@ defmodule Core.Services.PaymentsTest do
       {:ok, updated} = Payments.update_platform_plan(plan, user)
 
       assert updated.external_id == "sub_id"
+      assert updated.metered_id == "metered_id"
+
       [%{dimension: :user} = user] = updated.line_items
 
       assert user.id

apps/graphql/lib/graphql/schema/cluster.ex

@@ -26,6 +26,7 @@ defmodule GraphQl.Schema.Cluster do
     field :console_url,  :string, description: "The URL of the console running on the cluster."
     field :domain,       :string, description: "The domain name used for applications deployed on the cluster."
     field :pinged_at,    :datetime, description: "The last time the cluster was pinged."
+    field :service_count, :integer, description: "the services deployed from this cluster"
     field :upgrade_info, list_of(:upgrade_info), description: "pending upgrades for each installed app", resolve: fn
       cluster, _, _ -> Cluster.upgrade_info(cluster)
     end
@@ -76,8 +77,9 @@ defmodule GraphQl.Schema.Cluster do
 
   @desc "A record of the utilization in a given cluster"
   object :cluster_usage_history do
-    field :cpu,    :integer
-    field :memory, :integer
+    field :cpu,      :integer
+    field :memory,   :integer
+    field :services, :integer
 
     field :cluster, :cluster, resolve: dataloader(Cluster)
     field :account, :account, resolve: dataloader(Account)

apps/rtc/test/rtc_web/channels/upgrade_channel_test.exs

@@ -52,12 +52,15 @@ defmodule RtcWeb.UpgradeChannelTest do
       {:ok, socket} = mk_socket(user)
       {:ok, _, socket} = subscribe_and_join(socket, "queues:#{q.id}", %{})
 
-      ref = push(socket, "usage", %{"cpu" => 1000, "memory" => 10000})
+      ref = push(socket, "usage", %{"cpu" => 1000, "memory" => 10000, "services" => 5})
       assert_reply ref, :ok, _
 
       [hist] = Core.Schema.ClusterUsageHistory.for_cluster(cluster.id) |> Core.Repo.all()
       assert hist.cpu == 1000
       assert hist.memory == 10000
+      assert hist.services == 5
+
+      assert refetch(cluster).service_count == 5
     end
   end
 end

schema/schema.graphql

@@ -1580,6 +1580,9 @@ type Cluster {
   "The last time the cluster was pinged."
   pingedAt: DateTime
 
+  "the services deployed from this cluster"
+  serviceCount: Int
+
   "pending upgrades for each installed app"
   upgradeInfo: [UpgradeInfo]
 
@@ -2173,6 +2176,7 @@ type Stack {
 type ClusterUsageHistory {
   cpu: Int
   memory: Int
+  services: Int
   cluster: Cluster
   account: Account
   insertedAt: DateTime

www/src/generated/graphql.ts

@@ -433,6 +433,8 @@ export type Cluster = {
   provider: Provider;
   /** The upgrade queue for applications running on the cluster. */
   queue?: Maybe<UpgradeQueue>;
+  /** the services deployed from this cluster */
+  serviceCount?: Maybe<Scalars['Int']['output']>;
   /** The source of the cluster. */
   source?: Maybe<Source>;
   /** whether all installations in the cluster have been synced */
@@ -514,6 +516,7 @@ export type ClusterUsageHistory = {
   cpu?: Maybe<Scalars['Int']['output']>;
   insertedAt?: Maybe<Scalars['DateTime']['output']>;
   memory?: Maybe<Scalars['Int']['output']>;
+  services?: Maybe<Scalars['Int']['output']>;
   updatedAt?: Maybe<Scalars['DateTime']['output']>;
 };