Samir Boulahtit
408019dbb3
Implement the foundation for multi-platform support allowing independent business offerings (OMS, Loyalty, etc.) with their own CMS pages. Database Models: - Add Platform model for business offerings (domain, branding, config) - Add VendorPlatform junction table for many-to-many relationship - Update SubscriptionTier with platform_id and CMS limits - Update ContentPage with platform_id, is_platform_page for three-tier hierarchy - Add CMS feature codes (cms_basic, cms_custom_pages, cms_templates, etc.) Three-Tier Content Resolution: 1. Vendor override (platform_id + vendor_id + slug) 2. Vendor default (platform_id + vendor_id=NULL + is_platform_page=False) 3. Platform marketing pages (is_platform_page=True) New Components: - PlatformContextMiddleware for detecting platform from domain/path - ContentPageService updated with full three-tier resolution - Platform folder structure (app/platforms/oms/, app/platforms/loyalty/) - Alembic migration with backfill for existing data Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
757 lines
26 KiB
Python
757 lines
26 KiB
Python
# models/database/subscription.py
|
|
"""
|
|
Subscription database models for tier-based access control.
|
|
|
|
Provides models for:
|
|
- SubscriptionTier: Database-driven tier definitions with Stripe integration
|
|
- VendorSubscription: Per-vendor subscription tracking
|
|
- AddOnProduct: Purchasable add-ons (domains, SSL, email packages)
|
|
- VendorAddOn: Add-ons purchased by each vendor
|
|
- StripeWebhookEvent: Idempotency tracking for webhook processing
|
|
- BillingHistory: Invoice and payment history
|
|
|
|
Tier Structure:
|
|
- Essential (€49/mo): 100 orders/mo, 200 products, 1 user, LU invoicing
|
|
- Professional (€99/mo): 500 orders/mo, unlimited products, 3 users, EU VAT
|
|
- Business (€199/mo): 2000 orders/mo, unlimited products, 10 users, analytics, API
|
|
- Enterprise (€399+/mo): Unlimited, white-label, custom integrations
|
|
"""
|
|
|
|
import enum
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import (
|
|
Boolean,
|
|
Column,
|
|
DateTime,
|
|
ForeignKey,
|
|
Index,
|
|
Integer,
|
|
Numeric,
|
|
String,
|
|
Text,
|
|
)
|
|
from sqlalchemy.dialects.sqlite import JSON
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from app.core.database import Base
|
|
from models.database.base import TimestampMixin
|
|
|
|
|
|
class TierCode(str, enum.Enum):
|
|
"""Subscription tier codes."""
|
|
|
|
ESSENTIAL = "essential"
|
|
PROFESSIONAL = "professional"
|
|
BUSINESS = "business"
|
|
ENTERPRISE = "enterprise"
|
|
|
|
|
|
class SubscriptionStatus(str, enum.Enum):
|
|
"""Subscription status."""
|
|
|
|
TRIAL = "trial" # Free trial period
|
|
ACTIVE = "active" # Paid and active
|
|
PAST_DUE = "past_due" # Payment failed, grace period
|
|
CANCELLED = "cancelled" # Cancelled, access until period end
|
|
EXPIRED = "expired" # No longer active
|
|
|
|
|
|
class AddOnCategory(str, enum.Enum):
|
|
"""Add-on product categories."""
|
|
|
|
DOMAIN = "domain"
|
|
SSL = "ssl"
|
|
EMAIL = "email"
|
|
STORAGE = "storage"
|
|
|
|
|
|
class BillingPeriod(str, enum.Enum):
|
|
"""Billing period for add-ons."""
|
|
|
|
MONTHLY = "monthly"
|
|
ANNUAL = "annual"
|
|
ONE_TIME = "one_time"
|
|
|
|
|
|
# ============================================================================
|
|
# SubscriptionTier - Database-driven tier definitions
|
|
# ============================================================================
|
|
|
|
|
|
class SubscriptionTier(Base, TimestampMixin):
|
|
"""
|
|
Database-driven tier definitions with Stripe integration.
|
|
|
|
Replaces the hardcoded TIER_LIMITS dict for dynamic tier management.
|
|
|
|
Can be:
|
|
- Global tier (platform_id=NULL): Available to all platforms
|
|
- Platform-specific tier (platform_id set): Only for that platform
|
|
"""
|
|
|
|
__tablename__ = "subscription_tiers"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
|
|
# Platform association (NULL = global tier available to all platforms)
|
|
platform_id = Column(
|
|
Integer,
|
|
ForeignKey("platforms.id", ondelete="CASCADE"),
|
|
nullable=True,
|
|
index=True,
|
|
comment="Platform this tier belongs to (NULL = global tier)",
|
|
)
|
|
|
|
code = Column(String(30), nullable=False, index=True)
|
|
name = Column(String(100), nullable=False)
|
|
description = Column(Text, nullable=True)
|
|
|
|
# Pricing (in cents for precision)
|
|
price_monthly_cents = Column(Integer, nullable=False)
|
|
price_annual_cents = Column(Integer, nullable=True) # Null for enterprise/custom
|
|
|
|
# Limits (null = unlimited)
|
|
orders_per_month = Column(Integer, nullable=True)
|
|
products_limit = Column(Integer, nullable=True)
|
|
team_members = Column(Integer, nullable=True)
|
|
order_history_months = Column(Integer, nullable=True)
|
|
|
|
# CMS Limits (null = unlimited)
|
|
cms_pages_limit = Column(
|
|
Integer,
|
|
nullable=True,
|
|
comment="Total CMS pages limit (NULL = unlimited)",
|
|
)
|
|
cms_custom_pages_limit = Column(
|
|
Integer,
|
|
nullable=True,
|
|
comment="Custom pages limit, excluding overrides (NULL = unlimited)",
|
|
)
|
|
|
|
# Features (JSON array of feature codes)
|
|
features = Column(JSON, default=list)
|
|
|
|
# Stripe Product/Price IDs
|
|
stripe_product_id = Column(String(100), nullable=True)
|
|
stripe_price_monthly_id = Column(String(100), nullable=True)
|
|
stripe_price_annual_id = Column(String(100), nullable=True)
|
|
|
|
# Display and visibility
|
|
display_order = Column(Integer, default=0)
|
|
is_active = Column(Boolean, default=True, nullable=False)
|
|
is_public = Column(Boolean, default=True, nullable=False) # False for enterprise
|
|
|
|
# Relationship to Platform
|
|
platform = relationship(
|
|
"Platform",
|
|
back_populates="subscription_tiers",
|
|
foreign_keys=[platform_id],
|
|
)
|
|
|
|
# Unique constraint: tier code must be unique per platform (or globally if NULL)
|
|
__table_args__ = (
|
|
Index("idx_tier_platform_active", "platform_id", "is_active"),
|
|
)
|
|
|
|
def __repr__(self):
|
|
platform_info = f", platform_id={self.platform_id}" if self.platform_id else ""
|
|
return f"<SubscriptionTier(code='{self.code}', name='{self.name}'{platform_info})>"
|
|
|
|
def to_dict(self) -> dict:
|
|
"""Convert tier to dictionary (compatible with TIER_LIMITS format)."""
|
|
return {
|
|
"name": self.name,
|
|
"price_monthly_cents": self.price_monthly_cents,
|
|
"price_annual_cents": self.price_annual_cents,
|
|
"orders_per_month": self.orders_per_month,
|
|
"products_limit": self.products_limit,
|
|
"team_members": self.team_members,
|
|
"order_history_months": self.order_history_months,
|
|
"cms_pages_limit": self.cms_pages_limit,
|
|
"cms_custom_pages_limit": self.cms_custom_pages_limit,
|
|
"features": self.features or [],
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# AddOnProduct - Purchasable add-ons
|
|
# ============================================================================
|
|
|
|
|
|
class AddOnProduct(Base, TimestampMixin):
|
|
"""
|
|
Purchasable add-on products (domains, SSL, email packages).
|
|
|
|
These are separate from subscription tiers and can be added to any tier.
|
|
"""
|
|
|
|
__tablename__ = "addon_products"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
code = Column(String(50), unique=True, nullable=False, index=True)
|
|
name = Column(String(100), nullable=False)
|
|
description = Column(Text, nullable=True)
|
|
category = Column(String(50), nullable=False, index=True)
|
|
|
|
# Pricing
|
|
price_cents = Column(Integer, nullable=False)
|
|
billing_period = Column(
|
|
String(20), default=BillingPeriod.MONTHLY.value, nullable=False
|
|
)
|
|
|
|
# For tiered add-ons (e.g., email_5, email_10)
|
|
quantity_unit = Column(String(50), nullable=True) # emails, GB, etc.
|
|
quantity_value = Column(Integer, nullable=True) # 5, 10, 50, etc.
|
|
|
|
# Stripe
|
|
stripe_product_id = Column(String(100), nullable=True)
|
|
stripe_price_id = Column(String(100), nullable=True)
|
|
|
|
# Display
|
|
display_order = Column(Integer, default=0)
|
|
is_active = Column(Boolean, default=True, nullable=False)
|
|
|
|
def __repr__(self):
|
|
return f"<AddOnProduct(code='{self.code}', name='{self.name}')>"
|
|
|
|
|
|
# ============================================================================
|
|
# VendorAddOn - Add-ons purchased by vendor
|
|
# ============================================================================
|
|
|
|
|
|
class VendorAddOn(Base, TimestampMixin):
|
|
"""
|
|
Add-ons purchased by a vendor.
|
|
|
|
Tracks active add-on subscriptions and their billing status.
|
|
"""
|
|
|
|
__tablename__ = "vendor_addons"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=False, index=True)
|
|
addon_product_id = Column(
|
|
Integer, ForeignKey("addon_products.id"), nullable=False, index=True
|
|
)
|
|
|
|
# Status
|
|
status = Column(String(20), default="active", nullable=False, index=True)
|
|
|
|
# For domains: store the actual domain name
|
|
domain_name = Column(String(255), nullable=True, index=True)
|
|
|
|
# Quantity (for tiered add-ons like email packages)
|
|
quantity = Column(Integer, default=1, nullable=False)
|
|
|
|
# Stripe billing
|
|
stripe_subscription_item_id = Column(String(100), nullable=True)
|
|
|
|
# Period tracking
|
|
period_start = Column(DateTime(timezone=True), nullable=True)
|
|
period_end = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Cancellation
|
|
cancelled_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Relationships
|
|
vendor = relationship("Vendor", back_populates="addons")
|
|
addon_product = relationship("AddOnProduct")
|
|
|
|
__table_args__ = (
|
|
Index("idx_vendor_addon_status", "vendor_id", "status"),
|
|
Index("idx_vendor_addon_product", "vendor_id", "addon_product_id"),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<VendorAddOn(vendor_id={self.vendor_id}, addon={self.addon_product_id}, status='{self.status}')>"
|
|
|
|
|
|
# ============================================================================
|
|
# StripeWebhookEvent - Webhook idempotency tracking
|
|
# ============================================================================
|
|
|
|
|
|
class StripeWebhookEvent(Base, TimestampMixin):
|
|
"""
|
|
Log of processed Stripe webhook events for idempotency.
|
|
|
|
Prevents duplicate processing of the same event.
|
|
"""
|
|
|
|
__tablename__ = "stripe_webhook_events"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
event_id = Column(String(100), unique=True, nullable=False, index=True)
|
|
event_type = Column(String(100), nullable=False, index=True)
|
|
|
|
# Processing status
|
|
status = Column(String(20), default="pending", nullable=False, index=True)
|
|
processed_at = Column(DateTime(timezone=True), nullable=True)
|
|
error_message = Column(Text, nullable=True)
|
|
|
|
# Raw event data (encrypted for security)
|
|
payload_encrypted = Column(Text, nullable=True)
|
|
|
|
# Related entities (for quick lookup)
|
|
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=True, index=True)
|
|
subscription_id = Column(
|
|
Integer, ForeignKey("vendor_subscriptions.id"), nullable=True, index=True
|
|
)
|
|
|
|
__table_args__ = (Index("idx_webhook_event_type_status", "event_type", "status"),)
|
|
|
|
def __repr__(self):
|
|
return f"<StripeWebhookEvent(event_id='{self.event_id}', type='{self.event_type}', status='{self.status}')>"
|
|
|
|
|
|
# ============================================================================
|
|
# BillingHistory - Invoice and payment history
|
|
# ============================================================================
|
|
|
|
|
|
class BillingHistory(Base, TimestampMixin):
|
|
"""
|
|
Invoice and payment history for vendors.
|
|
|
|
Stores Stripe invoice data for display and reporting.
|
|
"""
|
|
|
|
__tablename__ = "billing_history"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=False, index=True)
|
|
|
|
# Stripe references
|
|
stripe_invoice_id = Column(String(100), unique=True, nullable=True, index=True)
|
|
stripe_payment_intent_id = Column(String(100), nullable=True)
|
|
|
|
# Invoice details
|
|
invoice_number = Column(String(50), nullable=True)
|
|
invoice_date = Column(DateTime(timezone=True), nullable=False)
|
|
due_date = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Amounts (in cents for precision)
|
|
subtotal_cents = Column(Integer, nullable=False)
|
|
tax_cents = Column(Integer, default=0, nullable=False)
|
|
total_cents = Column(Integer, nullable=False)
|
|
amount_paid_cents = Column(Integer, default=0, nullable=False)
|
|
currency = Column(String(3), default="EUR", nullable=False)
|
|
|
|
# Status
|
|
status = Column(String(20), nullable=False, index=True)
|
|
|
|
# PDF URLs
|
|
invoice_pdf_url = Column(String(500), nullable=True)
|
|
hosted_invoice_url = Column(String(500), nullable=True)
|
|
|
|
# Description and line items
|
|
description = Column(Text, nullable=True)
|
|
line_items = Column(JSON, nullable=True)
|
|
|
|
# Relationships
|
|
vendor = relationship("Vendor", back_populates="billing_history")
|
|
|
|
__table_args__ = (
|
|
Index("idx_billing_vendor_date", "vendor_id", "invoice_date"),
|
|
Index("idx_billing_status", "vendor_id", "status"),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<BillingHistory(vendor_id={self.vendor_id}, invoice='{self.invoice_number}', status='{self.status}')>"
|
|
|
|
|
|
# ============================================================================
|
|
# Legacy TIER_LIMITS (kept for backward compatibility during migration)
|
|
# ============================================================================
|
|
|
|
# Tier limit definitions (hardcoded for now, could be moved to DB)
|
|
TIER_LIMITS = {
|
|
TierCode.ESSENTIAL: {
|
|
"name": "Essential",
|
|
"price_monthly_cents": 4900, # €49
|
|
"price_annual_cents": 49000, # €490 (2 months free)
|
|
"orders_per_month": 100,
|
|
"products_limit": 200,
|
|
"team_members": 1,
|
|
"order_history_months": 6,
|
|
"features": [
|
|
"letzshop_sync",
|
|
"inventory_basic",
|
|
"invoice_lu",
|
|
"customer_view",
|
|
],
|
|
},
|
|
TierCode.PROFESSIONAL: {
|
|
"name": "Professional",
|
|
"price_monthly_cents": 9900, # €99
|
|
"price_annual_cents": 99000, # €990
|
|
"orders_per_month": 500,
|
|
"products_limit": None, # Unlimited
|
|
"team_members": 3,
|
|
"order_history_months": 24,
|
|
"features": [
|
|
"letzshop_sync",
|
|
"inventory_locations",
|
|
"inventory_purchase_orders",
|
|
"invoice_lu",
|
|
"invoice_eu_vat",
|
|
"customer_view",
|
|
"customer_export",
|
|
],
|
|
},
|
|
TierCode.BUSINESS: {
|
|
"name": "Business",
|
|
"price_monthly_cents": 19900, # €199
|
|
"price_annual_cents": 199000, # €1990
|
|
"orders_per_month": 2000,
|
|
"products_limit": None, # Unlimited
|
|
"team_members": 10,
|
|
"order_history_months": None, # Unlimited
|
|
"features": [
|
|
"letzshop_sync",
|
|
"inventory_locations",
|
|
"inventory_purchase_orders",
|
|
"invoice_lu",
|
|
"invoice_eu_vat",
|
|
"invoice_bulk",
|
|
"customer_view",
|
|
"customer_export",
|
|
"analytics_dashboard",
|
|
"accounting_export",
|
|
"api_access",
|
|
"automation_rules",
|
|
"team_roles",
|
|
],
|
|
},
|
|
TierCode.ENTERPRISE: {
|
|
"name": "Enterprise",
|
|
"price_monthly_cents": 39900, # €399 starting
|
|
"price_annual_cents": None, # Custom
|
|
"orders_per_month": None, # Unlimited
|
|
"products_limit": None, # Unlimited
|
|
"team_members": None, # Unlimited
|
|
"order_history_months": None, # Unlimited
|
|
"features": [
|
|
"letzshop_sync",
|
|
"inventory_locations",
|
|
"inventory_purchase_orders",
|
|
"invoice_lu",
|
|
"invoice_eu_vat",
|
|
"invoice_bulk",
|
|
"customer_view",
|
|
"customer_export",
|
|
"analytics_dashboard",
|
|
"accounting_export",
|
|
"api_access",
|
|
"automation_rules",
|
|
"team_roles",
|
|
"white_label",
|
|
"multi_vendor",
|
|
"custom_integrations",
|
|
"sla_guarantee",
|
|
"dedicated_support",
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
class VendorSubscription(Base, TimestampMixin):
|
|
"""
|
|
Per-vendor subscription tracking.
|
|
|
|
Tracks the vendor's subscription tier, billing period,
|
|
and usage counters for limit enforcement.
|
|
"""
|
|
|
|
__tablename__ = "vendor_subscriptions"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
vendor_id = Column(
|
|
Integer, ForeignKey("vendors.id"), unique=True, nullable=False, index=True
|
|
)
|
|
|
|
# Tier - tier_id is the FK, tier (code) kept for backwards compatibility
|
|
tier_id = Column(
|
|
Integer, ForeignKey("subscription_tiers.id"), nullable=True, index=True
|
|
)
|
|
tier = Column(
|
|
String(20), default=TierCode.ESSENTIAL.value, nullable=False, index=True
|
|
)
|
|
|
|
# Status
|
|
status = Column(
|
|
String(20), default=SubscriptionStatus.TRIAL.value, nullable=False, index=True
|
|
)
|
|
|
|
# Billing period
|
|
period_start = Column(DateTime(timezone=True), nullable=False)
|
|
period_end = Column(DateTime(timezone=True), nullable=False)
|
|
is_annual = Column(Boolean, default=False, nullable=False)
|
|
|
|
# Trial info
|
|
trial_ends_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Card collection tracking (for trials that require card upfront)
|
|
card_collected_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Usage counters (reset each billing period)
|
|
orders_this_period = Column(Integer, default=0, nullable=False)
|
|
orders_limit_reached_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Overrides (for custom enterprise deals)
|
|
custom_orders_limit = Column(Integer, nullable=True) # Override tier limit
|
|
custom_products_limit = Column(Integer, nullable=True)
|
|
custom_team_limit = Column(Integer, nullable=True)
|
|
|
|
# Payment info (Stripe integration)
|
|
stripe_customer_id = Column(String(100), nullable=True, index=True)
|
|
stripe_subscription_id = Column(String(100), nullable=True, index=True)
|
|
stripe_price_id = Column(String(100), nullable=True) # Current price being billed
|
|
stripe_payment_method_id = Column(String(100), nullable=True) # Default payment method
|
|
|
|
# Proration and upgrade/downgrade tracking
|
|
proration_behavior = Column(String(50), default="create_prorations")
|
|
scheduled_tier_change = Column(String(30), nullable=True) # Pending tier change
|
|
scheduled_change_at = Column(DateTime(timezone=True), nullable=True)
|
|
|
|
# Payment failure tracking
|
|
payment_retry_count = Column(Integer, default=0, nullable=False)
|
|
last_payment_error = Column(Text, nullable=True)
|
|
|
|
# Cancellation
|
|
cancelled_at = Column(DateTime(timezone=True), nullable=True)
|
|
cancellation_reason = Column(Text, nullable=True)
|
|
|
|
# Relationships
|
|
vendor = relationship("Vendor", back_populates="subscription")
|
|
tier_obj = relationship("SubscriptionTier", backref="subscriptions")
|
|
|
|
__table_args__ = (
|
|
Index("idx_subscription_vendor_status", "vendor_id", "status"),
|
|
Index("idx_subscription_period", "period_start", "period_end"),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<VendorSubscription(vendor_id={self.vendor_id}, tier='{self.tier}', status='{self.status}')>"
|
|
|
|
# =========================================================================
|
|
# Tier Limit Properties
|
|
# =========================================================================
|
|
|
|
@property
|
|
def tier_limits(self) -> dict:
|
|
"""Get the limit definitions for current tier.
|
|
|
|
Uses database tier (tier_obj) if available, otherwise falls back
|
|
to hardcoded TIER_LIMITS for backwards compatibility.
|
|
"""
|
|
# Use database tier if relationship is loaded
|
|
if self.tier_obj is not None:
|
|
return {
|
|
"orders_per_month": self.tier_obj.orders_per_month,
|
|
"products_limit": self.tier_obj.products_limit,
|
|
"team_members": self.tier_obj.team_members,
|
|
"features": self.tier_obj.features or [],
|
|
}
|
|
# Fall back to hardcoded limits
|
|
return TIER_LIMITS.get(TierCode(self.tier), TIER_LIMITS[TierCode.ESSENTIAL])
|
|
|
|
@property
|
|
def orders_limit(self) -> int | None:
|
|
"""Get effective orders limit (custom or tier default)."""
|
|
if self.custom_orders_limit is not None:
|
|
return self.custom_orders_limit
|
|
return self.tier_limits.get("orders_per_month")
|
|
|
|
@property
|
|
def products_limit(self) -> int | None:
|
|
"""Get effective products limit (custom or tier default)."""
|
|
if self.custom_products_limit is not None:
|
|
return self.custom_products_limit
|
|
return self.tier_limits.get("products_limit")
|
|
|
|
@property
|
|
def team_members_limit(self) -> int | None:
|
|
"""Get effective team members limit (custom or tier default)."""
|
|
if self.custom_team_limit is not None:
|
|
return self.custom_team_limit
|
|
return self.tier_limits.get("team_members")
|
|
|
|
@property
|
|
def features(self) -> list[str]:
|
|
"""Get list of enabled features for current tier."""
|
|
return self.tier_limits.get("features", [])
|
|
|
|
# =========================================================================
|
|
# Status Checks
|
|
# =========================================================================
|
|
|
|
@property
|
|
def is_active(self) -> bool:
|
|
"""Check if subscription allows access."""
|
|
return self.status in [
|
|
SubscriptionStatus.TRIAL.value,
|
|
SubscriptionStatus.ACTIVE.value,
|
|
SubscriptionStatus.PAST_DUE.value, # Grace period
|
|
SubscriptionStatus.CANCELLED.value, # Until period end
|
|
]
|
|
|
|
@property
|
|
def is_trial(self) -> bool:
|
|
"""Check if currently in trial."""
|
|
return self.status == SubscriptionStatus.TRIAL.value
|
|
|
|
@property
|
|
def trial_days_remaining(self) -> int | None:
|
|
"""Get remaining trial days."""
|
|
if not self.is_trial or not self.trial_ends_at:
|
|
return None
|
|
remaining = (self.trial_ends_at - datetime.now(UTC)).days
|
|
return max(0, remaining)
|
|
|
|
# =========================================================================
|
|
# Limit Checks
|
|
# =========================================================================
|
|
|
|
def can_create_order(self) -> tuple[bool, str | None]:
|
|
"""
|
|
Check if vendor can create/import another order.
|
|
|
|
Returns: (can_create, error_message)
|
|
"""
|
|
if not self.is_active:
|
|
return False, "Subscription is not active"
|
|
|
|
limit = self.orders_limit
|
|
if limit is None: # Unlimited
|
|
return True, None
|
|
|
|
if self.orders_this_period >= limit:
|
|
return False, f"Monthly order limit reached ({limit} orders). Upgrade to continue."
|
|
|
|
return True, None
|
|
|
|
def can_add_product(self, current_count: int) -> tuple[bool, str | None]:
|
|
"""
|
|
Check if vendor can add another product.
|
|
|
|
Args:
|
|
current_count: Current number of products
|
|
|
|
Returns: (can_add, error_message)
|
|
"""
|
|
if not self.is_active:
|
|
return False, "Subscription is not active"
|
|
|
|
limit = self.products_limit
|
|
if limit is None: # Unlimited
|
|
return True, None
|
|
|
|
if current_count >= limit:
|
|
return False, f"Product limit reached ({limit} products). Upgrade to add more."
|
|
|
|
return True, None
|
|
|
|
def can_add_team_member(self, current_count: int) -> tuple[bool, str | None]:
|
|
"""
|
|
Check if vendor can add another team member.
|
|
|
|
Args:
|
|
current_count: Current number of team members
|
|
|
|
Returns: (can_add, error_message)
|
|
"""
|
|
if not self.is_active:
|
|
return False, "Subscription is not active"
|
|
|
|
limit = self.team_members_limit
|
|
if limit is None: # Unlimited
|
|
return True, None
|
|
|
|
if current_count >= limit:
|
|
return False, f"Team member limit reached ({limit} members). Upgrade to add more."
|
|
|
|
return True, None
|
|
|
|
def has_feature(self, feature: str) -> bool:
|
|
"""Check if a feature is enabled for current tier."""
|
|
return feature in self.features
|
|
|
|
# =========================================================================
|
|
# Usage Tracking
|
|
# =========================================================================
|
|
|
|
def increment_order_count(self) -> None:
|
|
"""Increment the order counter for this period."""
|
|
self.orders_this_period += 1
|
|
|
|
# Track when limit was first reached
|
|
limit = self.orders_limit
|
|
if limit and self.orders_this_period >= limit and not self.orders_limit_reached_at:
|
|
self.orders_limit_reached_at = datetime.now(UTC)
|
|
|
|
def reset_period_counters(self) -> None:
|
|
"""Reset counters for new billing period."""
|
|
self.orders_this_period = 0
|
|
self.orders_limit_reached_at = None
|
|
|
|
|
|
# ============================================================================
|
|
# Capacity Planning
|
|
# ============================================================================
|
|
|
|
|
|
class CapacitySnapshot(Base, TimestampMixin):
|
|
"""
|
|
Daily snapshot of platform capacity metrics.
|
|
|
|
Used for growth trending and capacity forecasting.
|
|
Captured daily by background job.
|
|
"""
|
|
|
|
__tablename__ = "capacity_snapshots"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
snapshot_date = Column(DateTime(timezone=True), nullable=False, unique=True, index=True)
|
|
|
|
# Vendor metrics
|
|
total_vendors = Column(Integer, default=0, nullable=False)
|
|
active_vendors = Column(Integer, default=0, nullable=False)
|
|
trial_vendors = Column(Integer, default=0, nullable=False)
|
|
|
|
# Subscription metrics
|
|
total_subscriptions = Column(Integer, default=0, nullable=False)
|
|
active_subscriptions = Column(Integer, default=0, nullable=False)
|
|
|
|
# Resource metrics
|
|
total_products = Column(Integer, default=0, nullable=False)
|
|
total_orders_month = Column(Integer, default=0, nullable=False)
|
|
total_team_members = Column(Integer, default=0, nullable=False)
|
|
|
|
# Storage metrics
|
|
storage_used_gb = Column(Numeric(10, 2), default=0, nullable=False)
|
|
db_size_mb = Column(Numeric(10, 2), default=0, nullable=False)
|
|
|
|
# Capacity metrics (theoretical limits from subscriptions)
|
|
theoretical_products_limit = Column(Integer, nullable=True)
|
|
theoretical_orders_limit = Column(Integer, nullable=True)
|
|
theoretical_team_limit = Column(Integer, nullable=True)
|
|
|
|
# Tier distribution (JSON: {"essential": 10, "professional": 5, ...})
|
|
tier_distribution = Column(JSON, nullable=True)
|
|
|
|
# Performance metrics
|
|
avg_response_ms = Column(Integer, nullable=True)
|
|
peak_cpu_percent = Column(Numeric(5, 2), nullable=True)
|
|
peak_memory_percent = Column(Numeric(5, 2), nullable=True)
|
|
|
|
# Indexes
|
|
__table_args__ = (
|
|
Index("ix_capacity_snapshots_date", "snapshot_date"),
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<CapacitySnapshot(date={self.snapshot_date}, vendors={self.total_vendors})>"
|