"""Tier-based feature gating.
Maps subscription tiers to feature sets and provides helpers/decorators to
enforce access on API routes.
Tiers (cumulative):
starter (free) → dashboard, forecasting
launch ($49/mo) → + cascading_goals
growth ($99/mo) → + strategic_intelligence, scale_optimization, multi_market
command ($199/mo)→ everything + enterprise
"""
from functools import wraps
from flask import jsonify
from flask_login import current_user
# Ordered tiers, lowest → highest
TIER_ORDER = ['starter', 'launch', 'growth', 'command', 'enterprise']
# Features unlocked AT each tier (cumulative via tier ordering)
TIER_FEATURES = {
'starter': {'dashboard', 'forecasting'},
'launch': {'dashboard', 'forecasting', 'cascading_goals'},
'growth': {
'dashboard', 'forecasting', 'cascading_goals',
'strategic_intelligence', 'scale_optimization', 'multi_market',
},
'command': {
'dashboard', 'forecasting', 'cascading_goals',
'strategic_intelligence', 'scale_optimization', 'multi_market',
'enterprise',
},
'enterprise': {
'dashboard', 'forecasting', 'cascading_goals',
'strategic_intelligence', 'scale_optimization', 'multi_market',
'enterprise',
},
}
# Per-tier resource limits
TIER_LIMITS = {
'starter': {'markets': 1, 'seats': 2, 'connectors': 1},
'launch': {'markets': 1, 'seats': 5, 'connectors': 3},
'growth': {'markets': 5, 'seats': 25, 'connectors': 5},
'command': {'markets': 25, 'seats': 100, 'connectors': 99},
'enterprise': {'markets': 999, 'seats': 999, 'connectors': 99},
}
# 'enterprise' plan slug (from Stripe price map) maps to the command tier
_PLAN_TO_TIER = {
'launch': 'launch',
'growth': 'growth',
'command': 'command',
'enterprise': 'enterprise',
}
def get_company_tier(company) -> str:
"""Return the effective tier slug for a company.
Prefers the tier recorded in settings_json (set by the Stripe webhook
or admin override), then falls back to the subscription's plan_id,
then falls back to 'starter'.
"""
if company is None:
return 'starter'
settings = company.settings_json or {}
# Explicit tier in settings_json takes highest priority
# (allows admin override, testing, manual assignment)
settings_tier = settings.get('tier')
if settings_tier and settings_tier in TIER_FEATURES:
return settings_tier
sub = getattr(company, 'subscription', None)
# backref may be a list depending on relationship config
if isinstance(sub, (list, tuple)):
sub = sub[0] if sub else None
if sub is not None and sub.is_active:
tier = sub.plan_id
tier = _PLAN_TO_TIER.get(tier, tier)
return tier if tier in TIER_FEATURES else 'starter'
return 'starter'
def get_user_tier(user=None) -> str:
"""Return the effective tier for a user (via their primary company).
Super admins are treated as the top tier so internal tooling always works.
"""
from app.models import UserCompany
user = user or current_user
if not getattr(user, 'is_authenticated', False):
return 'starter'
if getattr(user, 'role', '') == 'super_admin':
return 'command'
uc = UserCompany.query.filter_by(user_id=user.id).first()
return get_company_tier(uc.company if uc else None)
def tier_has_feature(tier: str, feature: str) -> bool:
return feature in TIER_FEATURES.get(tier, TIER_FEATURES['starter'])
def get_tier_limits(tier: str) -> dict:
return TIER_LIMITS.get(tier, TIER_LIMITS['starter'])
def require_feature(feature: str):
"""Route decorator: 403 unless the current user's tier includes `feature`."""
def decorator(f):
@wraps(f)
def wrapped(*args, **kwargs):
tier = get_user_tier()
if not tier_has_feature(tier, feature):
return jsonify({
'error': 'upgrade_required',
'message': f"Your current plan ('{tier}') does not include this feature.",
'feature': feature,
'tier': tier,
}), 403
return f(*args, **kwargs)
return wrapped
return decorator