feat(entitlements): feature catalog, resolver, subscriptions, /me/entitlements

One catalog declares each feature's rollout flag and per-plan defaults, and
two evaluators read it: the flag service for deployment rollout and the new
resolver for what a principal holds. Plan state is a status machine on the
subscriptions document, versioned per owner in Redis and invalidated inside
every repository write. Replaces users.plan, the TIER rollout and CurrentUser.tier.

Author: Zingzy

app.py

@@ -30,6 +30,7 @@
 from infrastructure.oauth_clients import OAUTH_STATE_TTL_SECONDS, init_oauth
 from infrastructure.queue_redis import connect_queue_redis
 from infrastructure.templates import configure_template_globals, templates
+from middleware.entitlements import EntitlementsVersionMiddleware
 from middleware.error_handler import register_error_handlers
 from middleware.logging import RequestLoggingMiddleware
 from middleware.openapi import (
@@ -314,6 +315,9 @@ async def docs(request: Request):
     #    writes view_rate_limit into shared scope state during endpoint
     #    execution, before any response starts flowing outward
     app.add_middleware(RateLimitHeadersMiddleware)
+    # 8. Entitlement version header on authenticated responses; reads the
+    #    version the Entitled dependency stored, or one cache lookup.
+    app.add_middleware(EntitlementsVersionMiddleware)
 
     # ── Error handlers + rate limiter ────────────────────────────────────
     app.state.limiter = limiter

config.py

@@ -664,6 +664,29 @@ class SchedulerSettings(BaseSettings):
     lease_seconds: int = Field(default=600, ge=30)
 
 
+class BillingSettings(BaseSettings):
+    """Billing provider and the display prices the plans endpoint shows.
+
+    ``billing_provider="none"`` is the self-host default: no billing, and the
+    resolver hands every account the ``selfhost`` plan. Prices are display
+    values only; the provider owns what is charged.
+    """
+
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+    billing_provider: Literal["none", "paddle"] = "none"
+    pro_monthly_usd: int = Field(default=15, ge=0)
+    pro_year_usd: int = Field(default=144, ge=0)
+    founding_monthly_usd: int = Field(default=9, ge=0)
+    founding_year_usd: int = Field(default=90, ge=0)
+    founding_seats: int = Field(default=100, ge=0)
+    founding_window_days: int = Field(default=90, ge=0)
+
+    @property
+    def selfhost(self) -> bool:
+        return self.billing_provider == "none"
+
+
 class AppSettings(BaseSettings):
     model_config = SettingsConfigDict(env_file=".env", extra="ignore")
 
@@ -854,6 +877,7 @@ def _password_max_length_sane(cls, v: int) -> int:
     scheduler: SchedulerSettings | None = None
     llm: LlmSettings | None = None
     posthog_erasure: PostHogErasureSettings | None = None
+    billing: BillingSettings | None = None
 
     @model_validator(mode="after")
     def _populate_sub_configs_and_secret(self) -> AppSettings:
@@ -903,6 +927,8 @@ def _populate_sub_configs_and_secret(self) -> AppSettings:
             self.scheduler = SchedulerSettings()
         if self.posthog_erasure is None:
             self.posthog_erasure = PostHogErasureSettings()
+        if self.billing is None:
+            self.billing = BillingSettings()
         if self.webhooks.enabled and not self.secret_key:
             # Signing secrets are encrypted with a key derived from
             # SECRET_KEY; an empty master would mean a predictable key.

dependencies/__init__.py

@@ -35,6 +35,12 @@
     require_session_or_scopes,
     require_verified_email,
 )
+from dependencies.entitlements import (
+    Entitled,
+    EntitlementSvc,
+    get_entitlement_service,
+    get_entitlements,
+)
 from dependencies.infra import (
     AppRegistryDep,
     GeoIP,
@@ -132,6 +138,8 @@
     "CustomDomainSvc",
     "DeviceAuthSvc",
     "DomainIntelSvc",
+    "Entitled",
+    "EntitlementSvc",
     "ExportSvc",
     "FeatureFlagSvc",
     "GeoIP",
@@ -175,6 +183,8 @@
     "get_db",
     "get_device_auth_service",
     "get_email_provider",
+    "get_entitlement_service",
+    "get_entitlements",
     "get_export_service",
     "get_feature_flag_service",
     "get_geoip_service",

dependencies/auth.py

@@ -64,10 +64,9 @@ class CurrentUser:
     # access tokens minted before the claim existed; those users match by
     # user_id only until their next token refresh.
     email: str | None = field(default=None)
-    # UserDoc.plan value (e.g. "FREE") — consumed by FeatureFlagService's
-    # TIER rollout via getattr(user, "tier"). Populated from the DB on the
-    # API-key path and from the (future) "plan" claim on the JWT path.
-    tier: str | None = field(default=None)
+    # The JWT "plan" claim: a hint for the fail mode only, never authority.
+    # None on the API-key path; the resolver reads the owner's plan by id.
+    plan_claim: str | None = field(default=None)
 
 
 async def get_current_user(
@@ -189,7 +188,6 @@ async def get_current_user(
                 # The owning UserDoc is already fetched above for
                 # email_verified — no extra DB hit to carry the email.
                 email=user.email.lower() if user and user.email else None,
-                tier=user.plan.value if user and user.plan else None,
             )
 
     # ── JWT path ──────────────────────────────────────────────────────────────
@@ -246,9 +244,7 @@ async def get_current_user(
             email_verified=email_verified,
             amr=amr,
             email=email,
-            # Not issued yet — the paid-plans launch adds the claim; TIER
-            # flag rollouts become a pure data change at that point.
-            tier=claims.get("plan"),
+            plan_claim=claims.get("plan"),
             scopes=scopes,
             app_id=claims.get("app_id"),
         )

dependencies/entitlements.py

@@ -0,0 +1,38 @@
+"""
+``Entitled``: the resolved entitlements of the request's principal.
+
+Runs the resolver once per request (cache hit in the common case) and hands
+the map to routes and services. Nothing on the request path reads
+``subscriptions`` or overrides directly.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import Depends, Request
+
+from dependencies.auth import CurrentUser, get_current_user
+from services.entitlements import EntitlementService, Resolved
+
+
+def get_entitlement_service(request: Request) -> EntitlementService:
+    return request.app.state.entitlement_service
+
+
+async def get_entitlements(
+    request: Request,
+    user: CurrentUser | None = Depends(get_current_user),
+    service: EntitlementService = Depends(get_entitlement_service),
+) -> Resolved:
+    resolved = await service.resolve_for(
+        user.user_id if user else None,
+        plan_hint=user.plan_claim if user else None,
+    )
+    if user is not None and not resolved.degraded:
+        request.state.entitlements_version = resolved.version
+    return resolved
+
+
+Entitled = Annotated[Resolved, Depends(get_entitlements)]
+EntitlementSvc = Annotated[EntitlementService, Depends(get_entitlement_service)]

dependencies/wiring.py

@@ -12,6 +12,7 @@
 from fastapi import FastAPI
 
 from config import AppSettings
+from infrastructure.cache.entitlement_cache import EntitlementCache
 from infrastructure.cache.feature_flag_cache import FeatureFlagCache
 from infrastructure.cache.meta_fetch_cache import MetaFetchCache
 from infrastructure.cache.onboarding_cache import OnboardingCache
@@ -37,6 +38,10 @@
 from repositories.blocked_url_repository import BlockedUrlRepository
 from repositories.click_repository import ClickRepository
 from repositories.custom_domain_repository import CustomDomainRepository
+from repositories.entitlement_event_repository import EntitlementEventRepository
+from repositories.entitlement_override_repository import (
+    EntitlementOverrideRepository,
+)
 from repositories.feature_flag_repository import FeatureFlagRepository
 from repositories.feed_domain_repository import FeedDomainRepository
 from repositories.legacy.emoji_url_repository import EmojiUrlRepository
@@ -47,6 +52,7 @@
     ReportSubmissionRepository,
 )
 from repositories.scheduled_task_repository import ScheduledTaskRepository
+from repositories.subscription_repository import SubscriptionRepository
 from repositories.tag_repository import TagRepository
 from repositories.token_repository import TokenRepository
 from repositories.url_repository import UrlRepository
@@ -79,6 +85,7 @@
 from services.custom_domain_service import CustomDomainService
 from services.domain_intel_service import DomainIntelService
 from services.edge_cache.og_writethrough import OgEdgeWritethrough
+from services.entitlements import EntitlementService
 from services.events.sinks import (
     InlineDomainEventSink,
     NullDomainEventSink,
@@ -258,6 +265,47 @@ def build_posthog_eraser(settings: AppSettings, http_client) -> PostHogEraser:
     )
 
 
+def build_entitlement_store(
+    db, redis_client
+) -> tuple[
+    EntitlementCache,
+    EntitlementEventRepository,
+    SubscriptionRepository,
+    EntitlementOverrideRepository,
+]:
+    """The three entitlement repositories sharing one cache, so every write
+    to subscriptions or overrides invalidates the same ``ent:{id}`` key."""
+    cache = EntitlementCache(redis_client)
+    events = EntitlementEventRepository(db["entitlement_events"])
+    subscriptions = SubscriptionRepository(db["subscriptions"], events, cache)
+    overrides = EntitlementOverrideRepository(
+        db["entitlement_overrides"], events, cache
+    )
+    return cache, events, subscriptions, overrides
+
+
+def build_entitlement_service(
+    db, settings: AppSettings, redis_client
+) -> EntitlementService:
+    cache, events, subscriptions, overrides = build_entitlement_store(db, redis_client)
+    return EntitlementService(
+        subscriptions,
+        overrides,
+        events,
+        cache,
+        selfhost=settings.billing.selfhost,
+        usage={
+            "custom_domains_max": CustomDomainRepository(
+                db["custom_domains"]
+            ).count_by_owner,
+            "webhook_endpoints_max": WebhookEndpointRepository(
+                db["webhook-endpoints"]
+            ).count_by_user,
+            "api_keys_max": ApiKeyRepository(db["api-keys"]).count_by_user,
+        },
+    )
+
+
 def build_account_erasure_service(
     db,
     settings: AppSettings,
@@ -355,6 +403,9 @@ def build_account_erasure_service(
         redis_client=redis_client,
         url_service=url_service,
     )
+    _, ent_events, subscription_repo, override_repo = build_entitlement_store(
+        db, redis_client
+    )
 
     return AccountErasureService(
         user_repo=user_repo,
@@ -372,6 +423,9 @@ def build_account_erasure_service(
         report_repo=ReportRepository(db["reports"]),
         report_submission_repo=ReportSubmissionRepository(db["report_submissions"]),
         feature_flag_repo=FeatureFlagRepository(db["feature_flags"]),
+        subscription_repo=subscription_repo,
+        override_repo=override_repo,
+        entitlement_event_repo=ent_events,
         r2_storage=r2_storage,
         posthog=build_posthog_eraser(settings, http_client),
         mailer=build_erasure_mailer(settings, http_client),
@@ -403,6 +457,12 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
     blocked_url_repo = BlockedUrlRepository(db["blocked-urls"])
     app_grant_repo = AppGrantRepository(db["app-grants"])
     feature_flag_repo = FeatureFlagRepository(db["feature_flags"])
+    _, ent_events, subscription_repo, override_repo = build_entitlement_store(
+        db, redis_client
+    )
+    app.state.entitlement_service = build_entitlement_service(
+        db, settings, redis_client
+    )
 
     # ── Infrastructure ───────────────────────────────────────────────────
     url_cache = UrlCache(redis_client, ttl_seconds=settings.redis.redis_ttl_seconds)
@@ -771,7 +831,9 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
         max_active_keys=settings.max_active_api_keys,
     )
     app.state.page_layout_service = PageLayoutService(page_layout_repo)
-    token_factory = TokenFactory(settings.jwt)
+    token_factory = TokenFactory(
+        settings.jwt, plan_of=app.state.entitlement_service.plan_hint_for
+    )
     otp_service = OtpService(token_repo)
 
     app.state.user_repo = user_repo
@@ -1010,6 +1072,9 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
         report_repo=report_repo,
         report_submission_repo=report_submission_repo,
         feature_flag_repo=feature_flag_repo,
+        subscription_repo=subscription_repo,
+        override_repo=override_repo,
+        entitlement_event_repo=ent_events,
         r2_storage=r2_storage,
         posthog=build_posthog_eraser(settings, http_client),
         mailer=erasure_mailer,

errors.py

@@ -28,6 +28,10 @@
         "details": NotRequired[Any],
         "message": NotRequired[str],
         "hint": NotRequired[str],
+        "feature": NotRequired[str],
+        "limit": NotRequired[str],
+        "max": NotRequired[int],
+        "current": NotRequired[int],
     },
 )
 
@@ -121,6 +125,47 @@ class ConflictError(AppError):
     error_code = "conflict"
 
 
+class InvalidTransitionError(ConflictError):
+    """A subscription event arrived in a status it cannot legally change."""
+
+    error_code = "invalid_transition"
+
+
+class PlanRequiredError(ForbiddenError):
+    """The account's plan does not include the feature being written."""
+
+    error_code = "plan_required"
+
+    def __init__(self, feature: str) -> None:
+        super().__init__(f"{feature} is not included in your plan")
+        self.feature = feature
+
+    def to_dict(self) -> ErrorBody:
+        d = super().to_dict()
+        d["feature"] = self.feature
+        return d
+
+
+class LimitReachedError(ForbiddenError):
+    """A per-plan count limit is full; ``max`` and ``current`` let the UI
+    render the counter."""
+
+    error_code = "limit_reached"
+
+    def __init__(self, limit: str, *, max_: int, current: int) -> None:
+        super().__init__(f"{limit} limit reached ({current}/{max_})")
+        self.limit = limit
+        self.max = max_
+        self.current = current
+
+    def to_dict(self) -> ErrorBody:
+        d = super().to_dict()
+        d["limit"] = self.limit
+        d["max"] = self.max
+        d["current"] = self.current
+        return d
+
+
 class BlockedUrlError(AppError):
     status_code = 451
     error_code = "blocked"