"""Admin domain - data access layer.""" from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from typing import Iterable, Optional from uuid import UUID, uuid4 from sqlalchemy import or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from src.domain.admin.models import Admin, PlanPriceVersion, SubscriptionPlan from src.domain.billing.normalization import is_payable_plan from src.domain.workspaces.models import Workspace from src.domain.billing.models import Subscription _PAYING_SUBSCRIPTION_STATUSES = frozenset({"active", "past_due"}) _SUBSCRIBED_PLAN_STATUSES = frozenset({"active", "trialing", "past_due"}) @dataclass(slots=True) class AdminOwnedWorkspaceBillingSummary: is_paying_customer: bool = False subscribed_plans: list[str] = field(default_factory=list) def summarize_owned_workspace_billing( rows: Iterable[tuple[UUID, UUID, Optional[str], Optional[str], Optional[str], Optional[datetime]]], ) -> dict[UUID, AdminOwnedWorkspaceBillingSummary]: summaries: dict[UUID, AdminOwnedWorkspaceBillingSummary] = {} seen_workspace_ids: set[UUID] = set() for owner_id, workspace_id, selected_plan_slug, subscription_plan_slug, subscription_status, _created_at in rows: summary = summaries.setdefault(owner_id, AdminOwnedWorkspaceBillingSummary()) if workspace_id in seen_workspace_ids: continue plan_slug = (subscription_plan_slug or selected_plan_slug or "").strip().lower() or None status = (subscription_status or "").strip().lower() if not plan_slug or not is_payable_plan(plan_slug): continue if status not in _SUBSCRIBED_PLAN_STATUSES: continue seen_workspace_ids.add(workspace_id) if plan_slug not in summary.subscribed_plans: summary.subscribed_plans.append(plan_slug) if status in _PAYING_SUBSCRIPTION_STATUSES: summary.is_paying_customer = True return summaries class AdminRepository: """Platform admin membership - check if user is admin.""" def __init__(self, db: AsyncSession): self._db = db async def is_admin(self, user_id: UUID) -> bool: result = await self._db.execute(select(Admin).where(Admin.user_id == user_id)) return result.scalar_one_or_none() is not None async def filter_platform_admin_ids(self, user_ids: list[UUID]) -> set[UUID]: if not user_ids: return set() result = await self._db.execute(select(Admin.user_id).where(Admin.user_id.in_(user_ids))) return {row[0] for row in result.all()} async def ensure_platform_admin( self, user_id: UUID, *, created_by: Optional[UUID] = None, ) -> tuple[Admin, bool]: """Insert into ``admins`` if missing. Returns (row, created).""" result = await self._db.execute(select(Admin).where(Admin.user_id == user_id)) existing = result.scalar_one_or_none() if existing: return existing, False row = Admin(id=uuid4(), user_id=user_id, created_by=created_by) self._db.add(row) await self._db.flush() await self._db.refresh(row) return row, True async def list_owned_workspace_billing_summaries( self, user_ids: list[UUID], ) -> dict[UUID, AdminOwnedWorkspaceBillingSummary]: if not user_ids: return {} result = await self._db.execute( select( Workspace.owner_id, Workspace.id, Workspace.selected_plan_slug, Subscription.plan_slug, Subscription.status, Subscription.created_at, ) .select_from(Workspace) .outerjoin(Subscription, Subscription.workspace_id == Workspace.id) .where(Workspace.owner_id.in_(user_ids)) .where(Workspace.deleted_at.is_(None)) .order_by(Workspace.id.asc(), Subscription.created_at.desc()) ) rows = [ ( owner_id, workspace_id, selected_plan_slug, subscription_plan_slug, subscription_status, created_at, ) for owner_id, workspace_id, selected_plan_slug, subscription_plan_slug, subscription_status, created_at in result.all() if owner_id is not None ] return summarize_owned_workspace_billing(rows) class SubscriptionPlanRepository: def __init__(self, db: AsyncSession): self._db = db async def get_by_id(self, plan_id: UUID) -> Optional[SubscriptionPlan]: result = await self._db.execute( select(SubscriptionPlan).where(SubscriptionPlan.id == plan_id) ) return result.scalar_one_or_none() async def get_by_slug(self, slug: str) -> Optional[SubscriptionPlan]: result = await self._db.execute( select(SubscriptionPlan).where(SubscriptionPlan.slug == slug) ) return result.scalar_one_or_none() async def get_by_stripe_price_id(self, stripe_price_id: str) -> Optional[SubscriptionPlan]: result = await self._db.execute( select(SubscriptionPlan).where( or_( SubscriptionPlan.stripe_monthly_price_id == stripe_price_id, SubscriptionPlan.stripe_yearly_price_id == stripe_price_id, SubscriptionPlan.metered_stripe_price_id == stripe_price_id, ) ) ) return result.scalar_one_or_none() async def list_all(self, active_only: bool = False) -> list[SubscriptionPlan]: q = select(SubscriptionPlan).order_by(SubscriptionPlan.sort_order) if active_only: q = q.where(SubscriptionPlan.is_active.is_(True)) result = await self._db.execute(q) return list(result.scalars().all()) async def create(self, **kwargs) -> SubscriptionPlan: plan = SubscriptionPlan(**kwargs) self._db.add(plan) await self._db.flush() await self._db.refresh(plan) return plan async def update(self, plan_id: UUID, **kwargs) -> Optional[SubscriptionPlan]: plan = await self.get_by_id(plan_id) if not plan: return None for k, v in kwargs.items(): if v is not None and hasattr(plan, k): setattr(plan, k, v) await self._db.flush() await self._db.refresh(plan) return plan class PlanPriceVersionRepository: """Append-only price version history. Plans reference current version.""" def __init__(self, db: AsyncSession): self._db = db async def create( self, subscription_plan_id: UUID, interval: str, stripe_price_id: str, price_minor: int, currency: str = "usd", stripe_product_id: Optional[str] = None, is_current: bool = True, ) -> PlanPriceVersion: version = PlanPriceVersion( subscription_plan_id=subscription_plan_id, interval=interval, stripe_price_id=stripe_price_id, price_minor=price_minor, currency=currency, stripe_product_id=stripe_product_id, is_current=is_current, ) self._db.add(version) await self._db.flush() await self._db.refresh(version) return version async def set_current_exclusive(self, subscription_plan_id: UUID, interval: str, new_version_id: UUID) -> None: """Set new version as current; clear is_current on others for this plan+interval.""" await self._db.execute( update(PlanPriceVersion) .where( PlanPriceVersion.subscription_plan_id == subscription_plan_id, PlanPriceVersion.interval == interval, PlanPriceVersion.id != new_version_id, ) .values(is_current=False) ) await self._db.flush() async def get_by_stripe_price_id(self, stripe_price_id: str) -> Optional[PlanPriceVersion]: result = await self._db.execute( select(PlanPriceVersion).where(PlanPriceVersion.stripe_price_id == stripe_price_id) ) return result.scalar_one_or_none()