Phase 3: usage metering

usage_records — one metered observation per (deployment, resource,
period, granularity). resource_kind is the BILLING kind (droplet_hours,
spaces_gb_month, ...) so it joins straight to plan_items. sub_attribution
jsonb column present from day one for forward-compat (sub-billing) but
never written in phase 3.

metering_config — per-billing-kind granularity (daily default; flip to
hourly when fine-grained billing is wanted) + retention_days.

ArcadiaCloud.Metering:
- meter_day/1 — walks every deployment-attributed resource, maps its
  inventory kind to a billing kind, records that day's usage. Idempotent
  via the unique (deployment, resource, period, granularity) index.
- usage_for_period/3 — SUM(qty) GROUP BY billing kind over a date range,
  returns the %{kind => qty} map the quote engine takes as :usage.

Uniform accrual model so everything downstream is just SUM(qty):
- droplet      -> droplet_hours, 24/day when active (0 when off)
- spaces_bucket / snapshot / droplet_backup -> *_gb_month, size/days so
  the month sums to GB-months
- dns_zone     -> dns_zones, 1/days so the month sums to the zone count

MeteringWorker — Oban cron 01:10 UTC daily, meters the previous
complete day.

Smoke verified: a pilot deployment with 2 droplets + 2 DNS zones + 3
snapshots attributed; 10 days metered -> 70 usage records; aggregation
gave droplet_hours 480 (2x24x10), dns_zones 0.65 (2x10/31),
snapshot_gb_month 15.48; fed into the quote engine against Studio — all
within allowance so $50 base, no overage (correct for a light partial
month).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

lib/arcadia_cloud/metering.ex

@@ -0,0 +1,173 @@
+defmodule ArcadiaCloud.Metering do
+  @moduledoc """
+  Usage metering. Walks deployment-attributed cloud_resources and records
+  per-day usage into usage_records. The monthly invoice rollup (and the
+  quote engine) consume `usage_for_period/3`.
+
+  Metering maps an inventory kind (droplet, spaces_bucket, ...) to a
+  BILLING kind (droplet_hours, spaces_gb_month, ...) — usage_records and
+  plan_items both speak billing kinds.
+
+  Uniform accrual model: every kind contributes a daily quantity that
+  SUMS to the period total. droplet_hours sums to ~720/mo; gb_month kinds
+  record size/days_in_month so the month sums to GB-months; count kinds
+  (dns_zones) record 1/days_in_month so the month sums to the count.
+  Everything downstream is just SUM(qty).
+  """
+
+  import Ecto.Query, warn: false
+
+  alias ArcadiaCloud.Repo
+  alias ArcadiaCloud.Cloud.CloudResource
+  alias ArcadiaCloud.Metering.{UsageRecord, MeteringConfig}
+
+  @doc """
+  Meter a single day for every billable resource. Idempotent — the
+  unique index on (deployment, resource, period_start, granularity)
+  means re-running upserts. Returns count of records written.
+  """
+  def meter_day(%Date{} = date) do
+    period_start = DateTime.new!(date, ~T[00:00:00], "Etc/UTC")
+    days_in_month = Date.days_in_month(date)
+
+    billable_resources()
+    |> Enum.reduce(0, fn resource, count ->
+      case daily_usage(resource, days_in_month) do
+        nil ->
+          count
+
+        {billing_kind, qty} ->
+          upsert_record(resource, billing_kind, period_start, qty)
+          count + 1
+      end
+    end)
+  end
+
+  @doc """
+  Total usage per billing kind for a deployment over a period. Returns a
+  map %{billing_kind => total_qty(float)} — the shape the quote engine
+  takes as `:usage`.
+  """
+  def usage_for_period(deployment_id, %Date{} = from_date, %Date{} = to_date) do
+    from_dt = DateTime.new!(from_date, ~T[00:00:00], "Etc/UTC")
+    to_dt = DateTime.new!(to_date, ~T[23:59:59], "Etc/UTC")
+
+    from(u in UsageRecord,
+      where:
+        u.deployment_id == ^deployment_id and u.period_start >= ^from_dt and
+          u.period_start <= ^to_dt,
+      group_by: u.resource_kind,
+      select: {u.resource_kind, sum(u.qty)}
+    )
+    |> Repo.all()
+    |> Map.new(fn {kind, total} -> {kind, to_float(total)} end)
+  end
+
+  def list_usage_records(deployment_id, opts \\ []) do
+    from(u in UsageRecord,
+      where: u.deployment_id == ^deployment_id,
+      order_by: [desc: u.period_start]
+    )
+    |> maybe_limit(opts[:limit])
+    |> Repo.all()
+  end
+
+  # ---- config ---------------------------------------------------------------
+
+  def get_granularity(resource_kind) do
+    case Repo.get_by(MeteringConfig, resource_kind: resource_kind) do
+      nil -> "daily"
+      config -> config.granularity
+    end
+  end
+
+  def upsert_config(resource_kind, granularity, retention_days \\ 400) do
+    attrs = %{
+      resource_kind: resource_kind,
+      granularity: granularity,
+      retention_days: retention_days
+    }
+
+    case Repo.get_by(MeteringConfig, resource_kind: resource_kind) do
+      nil -> %MeteringConfig{} |> MeteringConfig.changeset(attrs) |> Repo.insert()
+      cfg -> cfg |> MeteringConfig.changeset(attrs) |> Repo.update()
+    end
+  end
+
+  # ---- internals ------------------------------------------------------------
+
+  defp billable_resources do
+    from(r in CloudResource,
+      where: not is_nil(r.deployment_id) and is_nil(r.deleted_at)
+    )
+    |> Repo.all()
+  end
+
+  # inventory kind => {billing kind, daily qty}
+  defp daily_usage(%CloudResource{kind: "droplet", status: status}, _dim) do
+    hours = if status == "active", do: 24, else: 0
+    {"droplet_hours", hours}
+  end
+
+  defp daily_usage(%CloudResource{kind: "spaces_bucket"} = r, dim) do
+    {"spaces_gb_month", attr_gb(r) / dim}
+  end
+
+  defp daily_usage(%CloudResource{kind: "snapshot"} = r, dim) do
+    {"snapshot_gb_month", attr_gb(r) / dim}
+  end
+
+  defp daily_usage(%CloudResource{kind: "droplet_backup"} = r, dim) do
+    {"snapshot_gb_month", attr_gb(r) / dim}
+  end
+
+  defp daily_usage(%CloudResource{kind: "dns_zone"}, dim) do
+    {"dns_zones", 1.0 / dim}
+  end
+
+  defp daily_usage(_resource, _dim), do: nil
+
+  defp attr_gb(%CloudResource{attrs: attrs}) do
+    (attrs && (attrs["size_gigabytes"] || attrs["disk_gb"])) || 0
+  end
+
+  defp upsert_record(resource, billing_kind, period_start, qty) do
+    granularity = get_granularity(billing_kind)
+    qty_dec = qty |> to_decimal()
+
+    case Repo.get_by(UsageRecord,
+           deployment_id: resource.deployment_id,
+           resource_id: resource.id,
+           period_start: period_start,
+           granularity: granularity
+         ) do
+      nil ->
+        %UsageRecord{}
+        |> UsageRecord.changeset(%{
+          deployment_id: resource.deployment_id,
+          resource_id: resource.id,
+          resource_kind: billing_kind,
+          period_start: period_start,
+          granularity: granularity,
+          qty: qty_dec
+        })
+        |> Repo.insert!()
+
+      existing ->
+        existing
+        |> UsageRecord.changeset(%{qty: qty_dec, resource_kind: billing_kind})
+        |> Repo.update!()
+    end
+  end
+
+  defp to_decimal(%Decimal{} = d), do: d
+  defp to_decimal(n) when is_integer(n), do: Decimal.new(n)
+  defp to_decimal(n) when is_float(n), do: Decimal.from_float(Float.round(n, 6))
+
+  defp to_float(nil), do: 0.0
+  defp to_float(%Decimal{} = d), do: Decimal.to_float(d)
+  defp to_float(n) when is_number(n), do: n / 1
+
+  defp maybe_limit(q, nil), do: q
+  defp maybe_limit(q, n), do: from(u in q, limit: ^n)
+end