feat(entitlements): lifecycle job for grace, lapse, reminders and reconcile

One scheduled tick a minute owns every transition no webhook fires: term
end to grace, grace end to lapsed, the seven-day past-due ceiling, the
30/7/1-day reminders and the grace and lapsed emails (deduped on the audit
log), and expired override pruning. Daily tasks compare live subscriptions
with the provider and list pro features a switched-off flag withholds.

Author: Zingzy

axiom/monitors/entitlements-lifecycle-stalled.json

@@ -0,0 +1,14 @@
+{
+  "name": "Entitlements lifecycle job stalled",
+  "description": "The lifecycle job ticks every minute and logs entitlements_lifecycle_tick. No tick for five minutes means subscriptions freeze in the generous direction: nobody moves to grace or lapses, reminders stop. Check the click worker (or the embedded scheduler) and the scheduled_tasks lease.",
+  "type": "Threshold",
+  "intervalMinutes": 5,
+  "rangeMinutes": 5,
+  "aplQuery": "['spoo-prod'] | where event == \"entitlements_lifecycle_tick\" | summarize count() by bin(_time, 5m)",
+  "operator": "Below",
+  "threshold": 1,
+  "alertOnNoData": true,
+  "notifierIds": [
+    "xd9Sj0KGJ0e2TYPnl2"
+  ]
+}

dependencies/wiring.py

@@ -93,6 +93,7 @@
 from services.domain_intel_service import DomainIntelService
 from services.edge_cache.og_writethrough import OgEdgeWritethrough
 from services.entitlements import EntitlementService
+from services.entitlements.lifecycle import LifecycleService, lifecycle_tasks
 from services.entitlements.over_limit import OverLimitService
 from services.events.sinks import (
     InlineDomainEventSink,
@@ -330,6 +331,18 @@ def build_entitlement_service(
     )
 
 
+def build_billing_provider(settings: AppSettings, http_client) -> BillingProvider:
+    billing = settings.billing
+    if billing.provider == "paddle":
+        return PaddleProvider(
+            http_client,
+            api_key=billing.paddle_api_key,
+            webhook_secret=billing.paddle_webhook_secret,
+            env=billing.paddle_env,
+        )
+    return NullBillingProvider()
+
+
 def build_billing_service(
     db,
     settings: AppSettings,
@@ -340,18 +353,8 @@ def build_billing_service(
     subscriptions: SubscriptionRepository,
 ) -> BillingService:
     billing = settings.billing
-    provider: BillingProvider
-    if billing.provider == "paddle":
-        provider = PaddleProvider(
-            http_client,
-            api_key=billing.paddle_api_key,
-            webhook_secret=billing.paddle_webhook_secret,
-            env=billing.paddle_env,
-        )
-    else:
-        provider = NullBillingProvider()
     return BillingService(
-        provider,
+        build_billing_provider(settings, http_client),
         entitlements,
         subscriptions,
         BillingEventRepository(db["billing_events"]),
@@ -361,6 +364,32 @@ def build_billing_service(
     )
 
 
+def build_lifecycle_service(
+    db,
+    settings: AppSettings,
+    *,
+    store,
+    entitlements: EntitlementService,
+    provider: BillingProvider,
+    mailer,
+) -> LifecycleService:
+    """``store`` is the caller's entitlement store tuple. An unconfigured
+    mailer becomes None so the job never retries a send it cannot make."""
+    _, events, subscriptions, overrides = store
+    return LifecycleService(
+        entitlements,
+        subscriptions,
+        overrides,
+        events,
+        UserRepository(db["users"]),
+        CustomDomainRepository(db["custom_domains"]),
+        FeatureFlagRepository(db["feature_flags"]),
+        mailer if settings.email.zepto_api_token else None,
+        app_url=settings.app_url,
+        provider=provider,
+    )
+
+
 def build_account_erasure_service(
     db,
     settings: AppSettings,
@@ -1178,6 +1207,16 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
                 ),
             ),
             erasure_sweep_task(account_erasure_service),
+            *lifecycle_tasks(
+                build_lifecycle_service(
+                    db,
+                    settings,
+                    store=ent_store,
+                    entitlements=app.state.entitlement_service,
+                    provider=build_billing_provider(settings, http_client),
+                    mailer=app.state.email_provider,
+                )
+            ),
         ]
     )
     app.state.task_scheduler = TaskScheduler(

infrastructure/email/zeptomail.py

@@ -141,6 +141,20 @@ async def send_welcome_email(self, email: str, user_name: str | None) -> bool:
         )
         return await self._send(email, user_name, subject, html_body, text_body)
 
+    async def send_html(
+        self,
+        to_email: str,
+        subject: str,
+        template_name: str,
+        context: dict,
+        text_body: str,
+    ) -> bool:
+        """Render ``templates/emails/<template_name>`` with ``context`` and send it."""
+        html_body = self._jinja.get_template(template_name).render(
+            app_url=self._app_url, **context
+        )
+        return await self._send(to_email, None, subject, html_body, text_body)
+
     async def send_deletion_requested(
         self, email: str, purge_after: datetime, restore_token: str | None = None
     ) -> bool:

repositories/entitlement_event_repository.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from datetime import datetime
+
 from bson import ObjectId
 
 from repositories.base import BaseRepository
@@ -21,5 +23,27 @@ async def count_for(self, user_id: ObjectId) -> int:
             {"user_id": user_id, "kind": {"$nin": list(_VERSIONLESS_KINDS)}}
         )
 
+    async def has_reminder(
+        self, user_id: ObjectId, kind: EntitlementEventKind, period: str
+    ) -> bool:
+        doc = await self._find_one_raw(
+            {"user_id": user_id, "kind": kind.value, "period": period}, {"_id": 1}
+        )
+        return doc is not None
+
+    async def users_with_override_writes_since(self, since: datetime) -> list[ObjectId]:
+        return await self._col.distinct(
+            "user_id",
+            {
+                "kind": {
+                    "$in": [
+                        EntitlementEventKind.OVERRIDE_GRANTED.value,
+                        EntitlementEventKind.OVERRIDE_REVOKED.value,
+                    ]
+                },
+                "at": {"$gte": since},
+            },
+        )
+
     async def delete_by_user(self, user_id: ObjectId) -> int:
         return await self._delete_many({"user_id": user_id})

repositories/entitlement_override_repository.py

@@ -153,6 +153,45 @@ async def revoke(
         await self._cache.invalidate(user_id)
         return True
 
+    async def find_expired(
+        self, now: datetime, *, limit: int = 500
+    ) -> list[EntitlementOverrideDoc]:
+        cursor = self._col.find({"expires_at": {"$ne": None, "$lte": now}}).limit(limit)
+        docs = await cursor.to_list(length=limit)
+        return [EntitlementOverrideDoc.from_mongo(d) for d in docs]  # type: ignore[misc]
+
+    async def revoke_expired(
+        self,
+        user_id: ObjectId,
+        key: str,
+        now: datetime,
+        *,
+        actor: str,
+        reason: str,
+    ) -> bool:
+        """Revoke only while still expired: an extension granted in between wins."""
+        existing = await self._find_one({"user_id": user_id, "key": key})
+        if existing is None or existing.expires_at is None:
+            return False
+        deleted = await self._delete(
+            {"_id": existing.id, "expires_at": {"$ne": None, "$lte": now}}
+        )
+        if not deleted:
+            return False
+        await self._events.append(
+            EntitlementEventDoc(
+                user_id=user_id,
+                kind=EntitlementEventKind.OVERRIDE_REVOKED,
+                actor=actor,
+                reason=reason,
+                before=_snapshot(existing),
+                after=None,
+                at=now,
+            )
+        )
+        await self._cache.invalidate(user_id)
+        return True
+
     async def delete_by_user(self, user_id: ObjectId) -> int:
         deleted = await self._delete_many({"user_id": user_id})
         if deleted:

repositories/indexes.py

@@ -199,6 +199,8 @@ async def ensure_indexes(
     await overrides_col.create_index([("expires_at", 1)], sparse=True)
     ent_events_col = db["entitlement_events"]
     await ent_events_col.create_index([("user_id", 1), ("at", -1)])
+    # The lifecycle tick sweeps recent override writes by kind and time.
+    await ent_events_col.create_index([("kind", 1), ("at", -1)])
     # The dedupe: a webhook delivery is handled once because this insert
     # either lands or raises DuplicateKeyError.
     billing_events_col = db["billing_events"]

repositories/subscription_repository.py

@@ -145,6 +145,84 @@ async def write_transition(
         await self._cache.invalidate(user_id)
         return after
 
+    async def find_term_ended(self, now: datetime) -> list[SubscriptionDoc]:
+        """Paid terms that have run out: a scheduled cancel past its period
+        end, or a prepaid year past its end."""
+        return await self._find_many(
+            {
+                "$or": [
+                    {
+                        "status": SubscriptionStatus.CANCEL_AT_PERIOD_END.value,
+                        "current_period_end": {"$lte": now},
+                    },
+                    {
+                        "status": SubscriptionStatus.ACTIVE.value,
+                        "kind": "prepaid",
+                        "prepaid_until": {"$lte": now},
+                    },
+                ]
+            }
+        )
+
+    async def find_grace_ended(self, now: datetime) -> list[SubscriptionDoc]:
+        return await self._find_many(
+            {"status": SubscriptionStatus.GRACE.value, "grace_until": {"$lte": now}}
+        )
+
+    async def find_in_grace(self) -> list[SubscriptionDoc]:
+        return await self._find_many({"status": SubscriptionStatus.GRACE.value})
+
+    async def find_lapsed_since(
+        self, since: datetime, *, not_by: str
+    ) -> list[SubscriptionDoc]:
+        """Lapsed recently enough to still be owed the lapsed email, except
+        those lapsed by ``not_by``; a lapsed document's last write is the lapse."""
+        return await self._find_many(
+            {
+                "status": SubscriptionStatus.LAPSED.value,
+                "updated_at": {"$gte": since},
+                "lapsed_by": {"$ne": not_by},
+            }
+        )
+
+    async def find_past_due_older_than(self, cutoff: datetime) -> list[SubscriptionDoc]:
+        return await self._find_many(
+            {
+                "status": SubscriptionStatus.PAST_DUE.value,
+                "current_period_end": {"$lte": cutoff},
+            }
+        )
+
+    async def find_terms_ending_before(self, until: datetime) -> list[SubscriptionDoc]:
+        """Subscriptions whose paid term ends before ``until`` and will not
+        renew: prepaid years and scheduled cancels."""
+        return await self._find_many(
+            {
+                "$or": [
+                    {
+                        "status": SubscriptionStatus.CANCEL_AT_PERIOD_END.value,
+                        "current_period_end": {"$lte": until},
+                    },
+                    {
+                        "status": SubscriptionStatus.ACTIVE.value,
+                        "kind": "prepaid",
+                        "prepaid_until": {"$lte": until},
+                    },
+                ]
+            }
+        )
+
+    async def find_provider_live(self, provider: str) -> list[SubscriptionDoc]:
+        """Every non-lapsed recurring subscription of a provider, for reconcile."""
+        return await self._find_many(
+            {
+                "provider": provider,
+                "kind": "recurring",
+                "status": {"$ne": SubscriptionStatus.LAPSED.value},
+            },
+            limit=5000,
+        )
+
     async def find_by_provider_subscription(
         self, subscription_id: str
     ) -> SubscriptionDoc | None:

schemas/models/entitlement_event.py

@@ -32,4 +32,6 @@ class EntitlementEventDoc(MongoBaseModel):
     reason: str
     before: dict[str, Any] | None = None
     after: dict[str, Any] | None = None
+    # Dedupe key for reminders: one email per (user, kind, period).
+    period: str | None = None
     at: datetime

schemas/models/subscription.py

@@ -57,6 +57,9 @@ class SubscriptionDoc(MongoBaseModel):
     grace_until: datetime | None = None
     founding: bool = False
     founding_streak_ok: bool = False
+    # The state-machine event that lapsed it; "ended" means a refund or a
+    # manual end, not a term running out.
+    lapsed_by: str | None = None
     created_at: datetime | None = None
     updated_at: datetime | None = None