278 lines
10 KiB
Python
278 lines
10 KiB
Python
"""
|
|
models/platform.py — Platform-level models (superadmin scope).
|
|
Tables: system_users, plans, tenants, tenant_billing_history,
|
|
tenant_setting_overrides, audit_log, jwt_blocklist
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from app.extensions import db
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SystemUser(db.Model):
|
|
__tablename__ = "system_users"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
email = db.Column(db.String(255), unique=True, nullable=False, index=True)
|
|
password_hash = db.Column(db.String(255), nullable=False)
|
|
name = db.Column(db.String(100), nullable=False)
|
|
role = db.Column(db.String(50), nullable=False, default="superadmin")
|
|
is_active = db.Column(db.Boolean, nullable=False, default=True)
|
|
failed_login_attempts = db.Column(db.Integer, nullable=False, default=0)
|
|
locked_until = db.Column(db.DateTime, nullable=True)
|
|
password_reset_token = db.Column(db.String(255), nullable=True)
|
|
password_reset_expires_at = db.Column(db.DateTime, nullable=True)
|
|
last_login_at = db.Column(db.DateTime, nullable=True)
|
|
created_at = db.Column(db.DateTime, nullable=False,
|
|
default=lambda: datetime.now(timezone.utc))
|
|
|
|
# Flask-Login interface
|
|
@property
|
|
def is_authenticated(self):
|
|
return True
|
|
|
|
@property
|
|
def is_anonymous(self):
|
|
return False
|
|
|
|
def get_id(self):
|
|
return f"system:{self.id}"
|
|
|
|
def is_locked(self):
|
|
if self.locked_until is None:
|
|
return False
|
|
return datetime.now(timezone.utc) < self.locked_until.replace(tzinfo=timezone.utc)
|
|
|
|
def record_login(self):
|
|
from datetime import datetime as _dt, timezone as _tz
|
|
self.failed_login_attempts = 0
|
|
self.locked_until = None
|
|
self.last_login_at = _dt.now(_tz.utc)
|
|
|
|
def record_failed_login(self, max_attempts: int, lockout_minutes: int):
|
|
from datetime import datetime as _dt, timezone as _tz, timedelta as _td
|
|
self.failed_login_attempts = (self.failed_login_attempts or 0) + 1
|
|
if self.failed_login_attempts >= max_attempts:
|
|
self.locked_until = _dt.now(_tz.utc) + _td(minutes=lockout_minutes)
|
|
|
|
def __repr__(self):
|
|
return f"<SystemUser {self.email}>"
|
|
|
|
|
|
class Plan(db.Model):
|
|
__tablename__ = "plans"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(50), unique=True, nullable=False)
|
|
price_monthly = db.Column(db.Numeric(8, 2), nullable=False)
|
|
max_staff = db.Column(db.Integer, nullable=True) # None = unlimited
|
|
max_locations = db.Column(db.Integer, nullable=True) # None = unlimited
|
|
features_json = db.Column(db.JSON, nullable=False, default=dict)
|
|
is_active = db.Column(db.Boolean, nullable=False, default=True)
|
|
|
|
tenants = db.relationship("Tenant", back_populates="plan", lazy="dynamic")
|
|
|
|
def has_feature(self, flag: str) -> bool:
|
|
return bool(self.features_json.get(flag, False))
|
|
|
|
def __repr__(self):
|
|
return f"<Plan {self.name}>"
|
|
|
|
|
|
class Tenant(db.Model):
|
|
__tablename__ = "tenants"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
slug = db.Column(db.String(80), unique=True, nullable=False, index=True)
|
|
name = db.Column(db.String(150), nullable=False)
|
|
owner_email = db.Column(db.String(255), nullable=False)
|
|
plan_id = db.Column(db.Integer, db.ForeignKey("plans.id"), nullable=False)
|
|
status = db.Column(db.String(20), nullable=False, default="trial")
|
|
# status: 'active' | 'trial' | 'suspended' | 'cancelled'
|
|
trial_ends_at = db.Column(db.DateTime, nullable=True)
|
|
subscription_expires_at = db.Column(db.DateTime, nullable=True)
|
|
is_demo = db.Column(db.Boolean, nullable=False, default=False)
|
|
created_at = db.Column(db.DateTime, nullable=False,
|
|
default=lambda: datetime.now(timezone.utc))
|
|
updated_at = db.Column(db.DateTime, nullable=False,
|
|
default=lambda: datetime.now(timezone.utc),
|
|
onupdate=lambda: datetime.now(timezone.utc))
|
|
|
|
plan = db.relationship("Plan", back_populates="tenants")
|
|
billing_history = db.relationship(
|
|
"TenantBillingHistory", back_populates="tenant", lazy="dynamic"
|
|
)
|
|
setting_overrides = db.relationship(
|
|
"TenantSettingOverride", back_populates="tenant", lazy="dynamic"
|
|
)
|
|
|
|
def is_suspended(self):
|
|
return self.status in ("suspended", "cancelled")
|
|
|
|
def is_active_status(self):
|
|
return self.status in ("active", "trial")
|
|
|
|
def get_setting(self, key: str, default=None):
|
|
"""
|
|
Resolve a tenant setting, honouring superadmin overrides.
|
|
Override takes precedence if lifted_at is NULL.
|
|
Falls back to tenant_settings, then the supplied default.
|
|
"""
|
|
override = (
|
|
TenantSettingOverride.query
|
|
.filter_by(tenant_id=self.id, setting_key=key)
|
|
.filter(TenantSettingOverride.lifted_at.is_(None))
|
|
.first()
|
|
)
|
|
if override:
|
|
return override.setting_value
|
|
|
|
from app.models.salon import TenantSetting
|
|
setting = TenantSetting.query.filter_by(
|
|
tenant_id=self.id, setting_key=key
|
|
).first()
|
|
return setting.setting_value if setting else default
|
|
|
|
def has_feature(self, flag: str) -> bool:
|
|
"""Check plan feature flag, allowing active overrides to force-enable/disable."""
|
|
override = (
|
|
TenantSettingOverride.query
|
|
.filter_by(tenant_id=self.id, setting_key=f"feature_{flag}")
|
|
.filter(TenantSettingOverride.lifted_at.is_(None))
|
|
.first()
|
|
)
|
|
if override:
|
|
return override.setting_value.lower() in ("true", "1", "yes")
|
|
return self.plan.has_feature(flag) if self.plan else False
|
|
|
|
def __repr__(self):
|
|
return f"<Tenant {self.slug}>"
|
|
|
|
|
|
class TenantBillingHistory(db.Model):
|
|
__tablename__ = "tenant_billing_history"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
tenant_id = db.Column(db.Integer, db.ForeignKey("tenants.id"),
|
|
nullable=False, index=True)
|
|
amount = db.Column(db.Numeric(10, 2), nullable=False)
|
|
description = db.Column(db.String(255), nullable=False)
|
|
paid_at = db.Column(db.DateTime, nullable=True)
|
|
invoice_ref = db.Column(db.String(100), nullable=True)
|
|
recorded_by = db.Column(db.Integer, db.ForeignKey("system_users.id"),
|
|
nullable=True)
|
|
created_at = db.Column(db.DateTime, nullable=False,
|
|
default=lambda: __import__('datetime').datetime.now(
|
|
__import__('datetime').timezone.utc))
|
|
|
|
tenant = db.relationship("Tenant", back_populates="billing_history")
|
|
recorder = db.relationship("SystemUser")
|
|
|
|
def __repr__(self):
|
|
return f"<BillingHistory tenant={self.tenant_id} amount={self.amount}>"
|
|
|
|
|
|
class TenantSettingOverride(db.Model):
|
|
__tablename__ = "tenant_setting_overrides"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
tenant_id = db.Column(db.Integer, db.ForeignKey("tenants.id"),
|
|
nullable=False, index=True)
|
|
setting_key = db.Column(db.String(100), nullable=False)
|
|
setting_value = db.Column(db.Text, nullable=True)
|
|
overridden_by = db.Column(db.Integer, db.ForeignKey("system_users.id"),
|
|
nullable=False)
|
|
overridden_at = db.Column(db.DateTime, nullable=False,
|
|
default=lambda: datetime.now(timezone.utc))
|
|
lifted_at = db.Column(db.DateTime, nullable=True)
|
|
note = db.Column(db.Text, nullable=True)
|
|
|
|
tenant = db.relationship("Tenant", back_populates="setting_overrides")
|
|
admin = db.relationship("SystemUser")
|
|
|
|
@property
|
|
def is_active(self):
|
|
return self.lifted_at is None
|
|
|
|
def __repr__(self):
|
|
return f"<SettingOverride tenant={self.tenant_id} key={self.setting_key}>"
|
|
|
|
|
|
class AuditLog(db.Model):
|
|
"""
|
|
Immutable, append-only audit log.
|
|
Never issue UPDATE or DELETE against this table from application code.
|
|
Retention: records older than 365 days purged monthly via APScheduler.
|
|
"""
|
|
__tablename__ = "audit_log"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
actor_id = db.Column(db.Integer, nullable=False)
|
|
actor_type = db.Column(db.String(50), nullable=False)
|
|
# actor_type: 'system_user' | 'tenant_user'
|
|
action = db.Column(db.String(100), nullable=False)
|
|
target_type = db.Column(db.String(100), nullable=True)
|
|
target_id = db.Column(db.Integer, nullable=True)
|
|
before_json = db.Column(db.JSON, nullable=True)
|
|
after_json = db.Column(db.JSON, nullable=True)
|
|
ip_address = db.Column(db.String(45), nullable=True)
|
|
created_at = db.Column(db.DateTime, nullable=False,
|
|
default=lambda: datetime.now(timezone.utc),
|
|
index=True)
|
|
|
|
@classmethod
|
|
def log(cls, actor_id, actor_type, action,
|
|
target_type=None, target_id=None,
|
|
before=None, after=None, ip_address=None):
|
|
"""
|
|
Helper to append an audit entry and flush to DB.
|
|
Usage:
|
|
AuditLog.log(
|
|
actor_id=current_user.id,
|
|
actor_type='system_user',
|
|
action='tenant.suspend',
|
|
target_type='tenant',
|
|
target_id=tenant.id,
|
|
before={'status': 'active'},
|
|
after={'status': 'suspended'},
|
|
ip_address=request.remote_addr,
|
|
)
|
|
"""
|
|
entry = cls(
|
|
actor_id=actor_id,
|
|
actor_type=actor_type,
|
|
action=action,
|
|
target_type=target_type,
|
|
target_id=target_id,
|
|
before_json=before,
|
|
after_json=after,
|
|
ip_address=ip_address,
|
|
)
|
|
db.session.add(entry)
|
|
logger.info(
|
|
"AUDIT | actor=%s(%s) action=%s target=%s/%s",
|
|
actor_type, actor_id, action, target_type, target_id,
|
|
)
|
|
return entry
|
|
|
|
def __repr__(self):
|
|
return f"<AuditLog {self.action} by {self.actor_type}:{self.actor_id}>"
|
|
|
|
|
|
class JWTBlocklist(db.Model):
|
|
"""
|
|
Stores revoked JWT refresh token JTIs.
|
|
Checked on every token refresh request.
|
|
"""
|
|
__tablename__ = "jwt_blocklist"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
jti = db.Column(db.String(36), nullable=False, unique=True, index=True)
|
|
token_type = db.Column(db.String(20), nullable=False, default="refresh")
|
|
revoked_at = db.Column(db.DateTime, nullable=False,
|
|
default=lambda: datetime.now(timezone.utc))
|
|
|
|
def __repr__(self):
|
|
return f"<JWTBlocklist jti={self.jti}>" |