"""Author RAG Chatbot SaaS — SuperAdmin Service. Handles: client access grants/revocations, client listing, bonus token grants, subscription extensions, and audit logging. RULE: Every state-changing action MUST write to the audit log. RULE: All revocations MUST update Redis revocation blacklist immediately. """ from datetime import datetime, timedelta, timezone from typing import Literal import asyncio import structlog from redis.asyncio import Redis from sqlalchemy.ext.asyncio import AsyncSession from app.config import get_settings from app.core.access.subscription import revoke_token_in_redis from app.core.access.token_crypto import ( create_subscription_token, hash_subscription_token, ) from app.models.client_access import ClientAccess from app.models.user import User from app.repositories.access_repo import AccessRepository from app.repositories.audit_repo import AuditRepository from app.repositories.user_repo import UserRepository from app.services.token_budget import clear_budget_exhausted, clear_token_usage_counters from app.services.email_service import EmailService logger = structlog.get_logger(__name__) cfg = get_settings() PlanType = Literal["monthly", "quarterly", "semi_annual", "annual"] _PLAN_DURATION_DAYS: dict[PlanType, int] = { "monthly": 30, "quarterly": 90, "semi_annual": 180, "annual": 365, } _PLAN_TOKEN_BUDGETS: dict[PlanType, int] = { "monthly": cfg.PLAN_MONTHLY_TOKENS, "quarterly": cfg.PLAN_QUARTERLY_TOKENS, "semi_annual": cfg.PLAN_SEMI_ANNUAL_TOKENS, "annual": cfg.PLAN_ANNUAL_TOKENS, } class SuperAdminService: """Orchestrates all SuperAdmin operations with full audit logging.""" def __init__(self, db: AsyncSession, redis: Redis) -> None: """Initialize with database session and Redis connection. Args: db: SQLAlchemy async session. redis: Redis async connection. """ self._db = db self._redis = redis self._access_repo = AccessRepository(db) self._user_repo = UserRepository(db) self._audit = AuditRepository(db) self._email = EmailService() async def _notify_email_async(self, send_fn, *args, **kwargs) -> None: """Fire-and-forget email — never block API responses on SMTP.""" if not cfg.SMTP_USERNAME or not cfg.SMTP_PASSWORD: logger.debug("SMTP not configured — skipping email notification") return async def _worker() -> None: try: await asyncio.wait_for( asyncio.to_thread(send_fn, *args, **kwargs), timeout=5.0, ) except Exception as e: logger.warning("Email notification failed", error=str(e)) asyncio.create_task(_worker()) async def suspend_author( self, actor: User, author_id: str, reason: str = "" ) -> None: """Suspend an author account (set chatbot_is_active=False). Args: actor: The SuperAdmin performing the action. author_id: UUID of the author to suspend. reason: Reason for suspension. """ author = await self._user_repo.get_by_id(author_id) if not author: raise ValueError(f"Author {author_id} not found") author.chatbot_is_active = False await self._db.flush() await self._audit.log( actor_id=actor.id, actor_email=actor.email, action="suspend_author", target_type="author", target_id=author_id, details={"reason": reason}, ) logger.info("Author suspended", author_id=author_id, reason=reason) async def unsuspend_author( self, actor: User, author_id: str ) -> None: """Reinstate a suspended author account. Args: actor: The SuperAdmin performing the action. author_id: UUID of the author to reinstate. """ author = await self._user_repo.get_by_id(author_id) if not author: raise ValueError(f"Author {author_id} not found") author.chatbot_is_active = True await self._db.flush() await self._audit.log( actor_id=actor.id, actor_email=actor.email, action="unsuspend_author", target_type="author", target_id=author_id, ) logger.info("Author unsuspended", author_id=author_id) async def delete_author(self, actor: User, author_id: str) -> None: """Permanently delete an author account and all associated data. Args: actor: The SuperAdmin performing the action. author_id: UUID of the author to delete. Raises: ValueError: If author not found, is SuperAdmin, or is the actor. """ author = await self._user_repo.get_by_id(author_id) if not author: raise ValueError(f"Author {author_id} not found") if author.role == "superadmin": raise ValueError("Cannot delete SuperAdmin accounts") if author.id == actor.id: raise ValueError("Cannot delete your own account") from sqlalchemy import select result = await self._db.execute( select(ClientAccess).where(ClientAccess.author_id == author_id) ) for access in result.scalars().all(): if not access.is_revoked: await revoke_token_in_redis( redis=self._redis, token_hash=access.token_hash, expires_at=access.expires_at, ) await clear_token_usage_counters(self._redis, author_id) # R-138: Clean up ChromaDB collections to prevent orphaned vectors try: from app.core.chroma_client import get_chroma_client chroma = get_chroma_client() # SEC-4 fix: use the full 32-char hex UUID to match get_collection_name(). # The old [:12] prefix would miss collections under the updated naming contract. author_prefix = f"a{author_id.replace('-', '')}" for col in chroma.list_collections(): col_name = col.name if hasattr(col, 'name') else str(col) if col_name.startswith(author_prefix): chroma.delete_collection(col_name) logger.info("Deleted ChromaDB collection", collection=col_name) except Exception as e: logger.warning("ChromaDB cleanup failed during author delete", error=str(e)) email = author.email name = author.full_name or email await self._audit.log( actor_id=actor.id, actor_email=actor.email, action="delete_author", target_type="author", target_id=author_id, details={"email": email, "full_name": name}, ) await self._db.delete(author) await self._db.flush() logger.info("Author deleted", author_id=author_id, email=email) async def grant_access( self, actor: User, author_id: str, plan: PlanType, auto_renew: bool = False, notes: str | None = None, custom_token_budget: int | None = None, ) -> dict: """Grant subscription access to an author. Args: actor: The SuperAdmin performing the action. author_id: UUID of the author to grant access. plan: Subscription plan name. auto_renew: Whether to auto-renew before expiry. notes: Optional admin notes. custom_token_budget: Override default plan token budget. Returns: Dict with access record and generated token. Raises: ValueError: If author not found. """ author = await self._user_repo.get_by_id(author_id) if not author or author.role != "author": raise ValueError(f"Author {author_id} not found") now = datetime.now(timezone.utc) expires_at = now + timedelta(days=_PLAN_DURATION_DAYS[plan]) token_budget = custom_token_budget or _PLAN_TOKEN_BUDGETS[plan] # Generate signed subscription token # We create a temporary grant_id to embed in the token from app.models.base import generate_uuid grant_id = generate_uuid() token = create_subscription_token( author_id=author_id, grant_id=grant_id, granted_at=now, expires_at=expires_at, ) token_hash = hash_subscription_token(token) # Persist the access record access = await self._access_repo.create({ "id": grant_id, "author_id": author_id, "granted_by": actor.id, "plan": plan, "granted_at": now, "expires_at": expires_at, "token_hash": token_hash, "token_budget": token_budget, "auto_renew": auto_renew, "notes": notes, }) # Audit log await self._audit.log( actor_id=actor.id, actor_email=actor.email, action="grant_access", target_type="author", target_id=author_id, details={"plan": plan, "expires_at": expires_at.isoformat(), "token_budget": token_budget}, ) # Notify author by email (non-blocking) await self._notify_email_async( self._email.send_subscription_granted, to=author.email, name=author.full_name or author.email, plan=plan.replace("_", " ").title(), expires_at=expires_at.strftime("%B %d, %Y"), ) logger.info("Access granted", author_id=author_id, plan=plan, grant_id=grant_id) return { "token": token, "access": { "grant_id": grant_id, "plan": plan, "expires_at": expires_at, "is_revoked": False, "token_budget": token_budget, "bonus_tokens": 0, }, } async def revoke_access( self, actor: User, grant_id: str, reason: str, ) -> None: """Revoke an active subscription. Takes effect instantly via Redis. Args: actor: The SuperAdmin performing the action. grant_id: UUID of the ClientAccess record to revoke. reason: Mandatory reason for the revocation (stored in audit log). Raises: ValueError: If access record not found. """ access = await self._access_repo.get_by_grant_id(grant_id) if not access: raise ValueError(f"Access grant {grant_id} not found") now = datetime.now(timezone.utc) # Update DB record await self._access_repo.update(grant_id, access.author_id, { "is_revoked": True, "revoked_at": now, "revoked_by": actor.id, "revoke_reason": reason, }) # Instantly add to Redis revocation blacklist await revoke_token_in_redis( redis=self._redis, token_hash=access.token_hash, expires_at=access.expires_at, ) # Clear any cached budget flags await clear_budget_exhausted(self._redis, access.author_id) # Audit log await self._audit.log( actor_id=actor.id, actor_email=actor.email, action="revoke_access", target_type="subscription", target_id=grant_id, details={"reason": reason, "author_id": access.author_id}, ) # Notify author (non-blocking — revocation is already effective) author = await self._user_repo.get_by_id(access.author_id) if author: await self._notify_email_async( self._email.send_subscription_revoked, to=author.email, name=author.full_name or author.email, reason=reason, ) logger.info("Access revoked", grant_id=grant_id, reason=reason) async def add_bonus_tokens( self, actor: User, grant_id: str, bonus_tokens: int, ) -> ClientAccess: """Add bonus tokens to an existing subscription without changing expiry. Args: actor: The SuperAdmin performing the action. grant_id: UUID of the ClientAccess record. bonus_tokens: Number of additional tokens to grant. Returns: Updated ClientAccess record. Raises: ValueError: If access record not found or already revoked. """ access = await self._access_repo.get_by_grant_id(grant_id) if not access or access.is_revoked: raise ValueError("Cannot add tokens to a non-existent or revoked subscription") new_bonus = access.bonus_tokens + bonus_tokens updated = await self._access_repo.update(grant_id, access.author_id, { "bonus_tokens": new_bonus }) # Clear budget exhausted flag if it was set await clear_budget_exhausted(self._redis, access.author_id) await self._audit.log( actor_id=actor.id, actor_email=actor.email, action="add_bonus_tokens", target_type="subscription", target_id=grant_id, details={"bonus_tokens_added": bonus_tokens, "new_total_bonus": new_bonus}, ) logger.info("Bonus tokens added", grant_id=grant_id, bonus=bonus_tokens) return updated async def extend_subscription( self, actor: User, grant_id: str, extend_days: int, ) -> ClientAccess: """Extend subscription expiry without resetting token budget. Works for both active AND expired subscriptions (but not revoked). """ access = await self._access_repo.get_by_grant_id(grant_id) if not access: raise ValueError("Subscription grant not found") if access.is_revoked: raise ValueError("Cannot extend a revoked subscription — create a new grant instead") # Extend from NOW if already expired, otherwise from current expiry # R-051: Use naive UTC datetimes consistently (SQLite stores naive) now = datetime.now(timezone.utc).replace(tzinfo=None) exp = access.expires_at if exp and exp.tzinfo is not None: exp = exp.replace(tzinfo=None) # strip tz if stored as aware base = exp if (exp and exp > now) else now new_expiry = base + timedelta(days=extend_days) updated = await self._access_repo.update(grant_id, access.author_id, { "expires_at": new_expiry }) await self._audit.log( actor_id=actor.id, actor_email=actor.email, action="extend_subscription", target_type="subscription", target_id=grant_id, details={"extended_by_days": extend_days, "new_expiry": new_expiry.isoformat()}, ) logger.info("Subscription extended", grant_id=grant_id, new_expiry=new_expiry) return updated async def reset_token_usage( self, actor: User, grant_id: str, ) -> ClientAccess: """Reset tokens_used to 0 for an existing subscription. Useful when re-granting or manually topping up. """ access = await self._access_repo.get_by_grant_id(grant_id) if not access: raise ValueError("Subscription grant not found") old_used = access.tokens_used updated = await self._access_repo.update(grant_id, access.author_id, { "tokens_used": 0, "bonus_tokens": 0, }) # Clear budget exhausted flag in Redis await clear_token_usage_counters(self._redis, access.author_id) await self._audit.log( actor_id=actor.id, actor_email=actor.email, action="reset_token_usage", target_type="subscription", target_id=grant_id, details={"tokens_cleared": old_used}, ) logger.info("Token usage reset", grant_id=grant_id, tokens_cleared=old_used) return updated async def list_all_clients( self, limit: int = 100, cursor: str | None = None ) -> list[User]: """List all author accounts with their latest access status.""" return await self._user_repo.list_all_authors(limit=limit, cursor=cursor) async def get_client_detail(self, author_id: str) -> dict: """Get full detail for one author: user + active access record.""" user = await self._user_repo.get_by_id(author_id) if not user: raise ValueError(f"Author {author_id} not found") access = await self._access_repo.get_active_for_author(author_id) return {"user": user, "access": access}