Chore/fix committed metrics backend

server/polar/metrics/metrics.py

@@ -33,28 +33,26 @@ def _not_in_trial(t: ColumnElement[datetime], i: TimeInterval) -> ColumnElement[
     buckets = t.table.c
     return or_(
         buckets.trial_end.is_(None),
-        i.sql_date_trunc(cast(SQLColumnExpression[datetime], buckets.trial_end))
-        <= i.sql_date_trunc(t),
+        buckets.trial_end < t + i.sql_interval(),
     )
 
 
 def _in_trial(t: ColumnElement[datetime], i: TimeInterval) -> ColumnElement[bool]:
     buckets = t.table.c
     return and_(
         buckets.trial_end.is_not(None),
-        i.sql_date_trunc(cast(SQLColumnExpression[datetime], buckets.trial_end))
-        > i.sql_date_trunc(t),
+        buckets.trial_end >= t + i.sql_interval(),
     )
 
 
-def _not_ended_as_of(
-    t: ColumnElement[datetime], i: TimeInterval, when: datetime
+def _not_ended_as_of_bucket(
+    t: ColumnElement[datetime], i: TimeInterval
 ) -> ColumnElement[bool]:
     buckets = t.table.c
     return or_(
         buckets.ended_or_ends_at.is_(None),
         i.sql_date_trunc(cast(SQLColumnExpression[datetime], buckets.ended_or_ends_at))
-        < i.sql_date_trunc(when),
+        < i.sql_date_trunc(t),
     )
 
 
@@ -132,7 +130,9 @@ class CommittedSubscriptionsMetric(SQLMetric):
     def get_sql_expression(
         cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime
     ) -> ColumnElement[int]:
-        return func.count(t.table.c.subscription_id).filter(_not_ended_as_of(t, i, now))
+        return func.count(t.table.c.subscription_id).filter(
+            _not_ended_as_of_bucket(t, i)
+        )
 
     @classmethod
     def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float:
@@ -171,7 +171,7 @@ def get_sql_expression(
         buckets = t.table.c
         return func.coalesce(
             func.sum(buckets.monthly_amount).filter(
-                and_(_not_in_trial(t, i), _not_ended_as_of(t, i, now))
+                and_(_not_in_trial(t, i), _not_ended_as_of_bucket(t, i))
             ),
             0,
         )
@@ -212,7 +212,7 @@ def get_sql_expression(
     ) -> ColumnElement[int]:
         return func.coalesce(
             func.sum(t.table.c.monthly_amount).filter(
-                and_(_in_trial(t, i), _not_ended_as_of(t, i, now))
+                and_(_in_trial(t, i), _not_ended_as_of_bucket(t, i))
             ),
             0,
         )

server/tests/metrics/test_service.py

@@ -20,6 +20,7 @@
 from polar.integrations.tinybird.service import DATASOURCE_EVENTS, _event_to_tinybird
 from polar.kit.db.postgres import create_async_engine, create_async_sessionmaker
 from polar.kit.time_queries import TimeInterval
+from polar.metrics.schemas import MetricsResponse
 from polar.metrics.service import metrics as metrics_service
 from polar.models import (
     Customer,
@@ -3412,6 +3413,53 @@ async def test_committed_mrr(
         assert mar.monthly_recurring_revenue == 100_00
         assert mar.committed_monthly_recurring_revenue == 100_00
 
+    async def test_committed_mrr_is_stable_across_now_values(
+        self,
+        metrics_harness: MetricsHarness,
+        metrics_session: AsyncSession,
+    ) -> None:
+        case = QUERY_CASES_BY_LABEL["committed_mrr"]
+        org_ctx = metrics_harness.organizations[case.org_key]
+        auth_subject = _metrics_auth_subject(
+            metrics_harness.user,
+            metrics_harness.unauthorized_user,
+            org_ctx.organization,
+            case.auth_type,
+        )
+
+        async def get_metrics(now: datetime) -> MetricsResponse:
+            return await metrics_service.get_metrics(
+                metrics_session,
+                auth_subject,
+                start_date=case.start_date,
+                end_date=case.end_date,
+                timezone=ZoneInfo(case.timezone),
+                interval=case.interval,
+                organization_id=[org_ctx.organization.id]
+                if case.organization_id_filter or case.auth_type == "user"
+                else None,
+                product_id=[org_ctx.product_ids[k] for k in case.product_keys] or None,
+                billing_type=list(case.billing_types) or None,
+                customer_id=[org_ctx.customer_ids[k] for k in case.customer_keys]
+                or None,
+                metrics=list(case.metrics) if case.metrics is not None else None,
+                now=now,
+            )
+
+        early_metrics = await get_metrics(datetime(2024, 2, 15, tzinfo=UTC))
+        later_metrics = await get_metrics(datetime(2024, 4, 15, tzinfo=UTC))
+
+        assert len(early_metrics.periods) == len(later_metrics.periods) == 3
+
+        for early_period, later_period in zip(
+            early_metrics.periods, later_metrics.periods, strict=True
+        ):
+            assert early_period.timestamp == later_period.timestamp
+            assert (
+                early_period.committed_monthly_recurring_revenue
+                == later_period.committed_monthly_recurring_revenue
+            )
+
     async def test_trial_excluded_from_mrr(
         self,
         metrics_harness: MetricsHarness,
@@ -3466,6 +3514,60 @@ async def test_trial_excluded_from_mrr(
         assert mar.trial_monthly_recurring_revenue == 0
         assert mar.trial_committed_monthly_recurring_revenue == 0
 
+    async def test_trial_mrr_is_stable_across_now_values(
+        self,
+        metrics_harness: MetricsHarness,
+        metrics_session: AsyncSession,
+    ) -> None:
+        case = QUERY_CASES_BY_LABEL["trial_excluded_from_mrr"]
+        org_ctx = metrics_harness.organizations[case.org_key]
+        auth_subject = _metrics_auth_subject(
+            metrics_harness.user,
+            metrics_harness.unauthorized_user,
+            org_ctx.organization,
+            case.auth_type,
+        )
+
+        async def get_metrics(now: datetime) -> MetricsResponse:
+            return await metrics_service.get_metrics(
+                metrics_session,
+                auth_subject,
+                start_date=case.start_date,
+                end_date=case.end_date,
+                timezone=ZoneInfo(case.timezone),
+                interval=case.interval,
+                organization_id=[org_ctx.organization.id]
+                if case.organization_id_filter or case.auth_type == "user"
+                else None,
+                product_id=[org_ctx.product_ids[k] for k in case.product_keys] or None,
+                billing_type=list(case.billing_types) or None,
+                customer_id=[org_ctx.customer_ids[k] for k in case.customer_keys]
+                or None,
+                metrics=list(case.metrics) if case.metrics is not None else None,
+                now=now,
+            )
+
+        early = await get_metrics(datetime(2024, 2, 1, tzinfo=UTC))
+        later = await get_metrics(datetime(2024, 6, 1, tzinfo=UTC))
+
+        assert len(early.periods) == len(later.periods) == 3
+        for early_period, later_period in zip(
+            early.periods, later.periods, strict=True
+        ):
+            assert early_period.timestamp == later_period.timestamp
+            assert (
+                early_period.monthly_recurring_revenue
+                == later_period.monthly_recurring_revenue
+            )
+            assert (
+                early_period.trial_monthly_recurring_revenue
+                == later_period.trial_monthly_recurring_revenue
+            )
+            assert (
+                early_period.trial_committed_monthly_recurring_revenue
+                == later_period.trial_committed_monthly_recurring_revenue
+            )
+
     async def test_mrr_subscription_forever_discount(
         self,
         metrics_harness: MetricsHarness,