"""Plan-based feature gating helpers (INDL-59). `PLAN_FEATURES` / `PLAN_LIMITS` (src/core/constants.py) are the single source of truth for what each subscription plan unlocks. `has_feature()` is the only sanctioned way to check a plan-gated feature anywhere in the API — every previously-duplicated inline `plan in (...)` / `plan == ...` check should call this instead. `require_feature()` (src/core/dependencies.py) wraps `has_feature()` as a FastAPI dependency for routes that need a server-side 403, mirroring the pre-existing `aiSearch` gate at src/apps/public/router.py (`POST /records/ai-search`). """ from typing import Optional, Union from src.core.constants import PLAN_FEATURES, PLAN_LIMITS, SubscriptionPlan, TRIAL_RECORD_LIMIT _ORDERED_PLANS = ( SubscriptionPlan.STARTER, SubscriptionPlan.PROFESSIONAL, SubscriptionPlan.ENTERPRISE, ) _PLAN_DISPLAY_NAMES = { SubscriptionPlan.STARTER: "Starter", SubscriptionPlan.PROFESSIONAL: "Professional", SubscriptionPlan.ENTERPRISE: "Enterprise", } # Human-readable labels for 403 messages. Keys mirror the Plan Feature Matrix # in docs/59_INDL-59_Plan_Based_Feature_Gating.md exactly. _FEATURE_DISPLAY_NAMES = { "cemeteryMap": "Cemetery map", "publicMemorialPages": "Public memorial pages", "qrCodes": "QR codes", "sales": "Sales", "scheduling": "Scheduling", "aiSearch": "AI search", "aiRecordExtraction": "AI record extraction", "aiBiographyWriter": "AI biography writer", "dashboardServiceKpis": "Dashboard service KPIs", "dashboardWeeklyServices": "Dashboard weekly services", "recordMemorialTab": "Record memorial tab", "findInfoCards": "Find page info cards", "feeSchedule": "Fee schedule", "emailTemplates": "Email templates", "crewMembers": "Crew members", "billingInvoicesTab": "Billing invoices", } def _resolve_plan(account_or_plan: Union[object, str, SubscriptionPlan, None]) -> Optional[SubscriptionPlan]: """Accept either an `Account` ORM instance or a raw plan string/enum. `Account.plan` is a plain `String` column (not a DB-level enum), so any value read from the database is always a `str` — this normalizes it to the `SubscriptionPlan` enum used as the matrix key. """ raw = getattr(account_or_plan, "plan", account_or_plan) if raw is None: return None try: return SubscriptionPlan(raw) except ValueError: return None def has_feature(account_or_plan: Union[object, str, SubscriptionPlan, None], key: str) -> bool: """Return whether `account_or_plan`'s plan unlocks feature `key`. Accepts either an `Account` model instance (its `.plan` column is read) or a raw plan string/`SubscriptionPlan` value, so callers that already have an `Account` in hand and callers that only have a plan string can both use it directly without an extra branch. Unknown plan values or unknown feature keys resolve to `False` (fail-closed) rather than raising. """ plan = _resolve_plan(account_or_plan) if plan is None: return False return bool(PLAN_FEATURES.get(plan, {}).get(key, False)) def feature_limit(account_or_plan: Union[object, str, SubscriptionPlan, None], limit_key: str) -> Optional[int]: """Return the numeric limit (e.g. `records`) for a plan. Returns `None` both for "unlimited" (Enterprise `records`) and for an unrecognized plan/limit_key — callers that need to distinguish those cases should check `has_feature`/plan validity separately. """ plan = _resolve_plan(account_or_plan) if plan is None: return None return PLAN_LIMITS.get(plan, {}).get(limit_key) def effective_records_limit( account_or_plan: Union[object, str, SubscriptionPlan, None], is_trialing: bool, ) -> Optional[int]: """Records limit that should actually be enforced/displayed right now. During the free trial (`Subscription.payment_status == "trialing"`) every plan is capped at `TRIAL_RECORD_LIMIT`, regardless of the plan's real `PLAN_LIMITS[...]["records"]` ceiling — this stops trial accounts from loading in data volume they haven't paid for yet. The moment the first charge succeeds (`payment_status` flips to `"paid"`), the caller passes `is_trialing=False` and the plan's real limit applies again. """ if is_trialing: return TRIAL_RECORD_LIMIT return feature_limit(account_or_plan, "records") def records_limit_reached_message( account_or_plan: Union[object, str, SubscriptionPlan, None], is_trialing: bool, ) -> str: """Build the 403 message when a tenant hits its effective records limit.""" limit = effective_records_limit(account_or_plan, is_trialing) if is_trialing: return ( f"You've reached the {limit}-record limit for your free trial. " "Add a payment method to unlock your plan's full record limit." ) label = plan_display_name(account_or_plan) return f"You've reached the maximum of {limit} records on the {label} plan." def plan_display_name(account_or_plan: Union[object, str, SubscriptionPlan, None]) -> str: """Human-readable plan name (e.g. "Starter") for limit-reached messages.""" plan = _resolve_plan(account_or_plan) return _PLAN_DISPLAY_NAMES.get(plan, "your") if plan is not None else "your" def staff_limit_reached_message(account_or_plan: Union[object, str, SubscriptionPlan, None]) -> str: """Build the 403 message for INDL-59's staff-seat limit (2 Starter / 10 Professional / 10 Enterprise — `PLAN_LIMITS[plan]["staff"]`). Only Starter benefits from an "upgrade to add more" nudge — Professional and Enterprise share the same 10-seat ceiling today, so suggesting an upgrade there would be misleading. """ plan = _resolve_plan(account_or_plan) limit = feature_limit(account_or_plan, "staff") label = plan_display_name(account_or_plan) base = f"You've reached the maximum of {limit} staff users on the {label} plan." if plan == SubscriptionPlan.STARTER: return f"{base} Upgrade to Professional or Enterprise to add more." return base def feature_gate_message(key: str) -> str: """Build the 403 message for a gated feature. Matches the tone of the existing ai-search gate precedent: " is available on the plan(s)." """ plans_with_feature = [ plan for plan in _ORDERED_PLANS if PLAN_FEATURES.get(plan, {}).get(key) is True ] names = [_PLAN_DISPLAY_NAMES[p] for p in plans_with_feature] label = _FEATURE_DISPLAY_NAMES.get(key, key) if not names: # Not offered on any plan today — shouldn't normally be gated, but # keep the message sane rather than raising. return f"{label} is not available on your plan." if len(names) == 1: return f"{label} is available on the {names[0]} plan." if len(names) == 2: return f"{label} is available on the {names[0]} and {names[1]} plans." return f"{label} is available on the {', '.join(names[:-1])}, and {names[-1]} plans."