feat(l1): seat_enforcement service for engineer + L1 seat limits

Shared helper used by invite, accept-invite, and role-change endpoints
(integrated in T8). Counts active users by role against role-specific
seat limit on subscription (engineer → seat_limit, l1_tech → l1_seat_limit).
None limit = unlimited.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-28 12:40:48 -04:00
parent 874dee7263
commit 02fc47c832
3 changed files with 276 additions and 0 deletions

View File

@@ -0,0 +1,18 @@
from typing import Literal, Optional
from pydantic import BaseModel
Role = Literal['engineer', 'l1_tech']
class SeatCheckResult(BaseModel):
available: bool
current: int
limit: Optional[int] # None = unlimited
role: Role
class SeatUsage(BaseModel):
engineer: SeatCheckResult
l1_tech: SeatCheckResult

View File

@@ -0,0 +1,63 @@
from typing import Literal
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.account import Account
from app.models.subscription import Subscription
from app.models.user import User
from app.schemas.seat_enforcement import SeatCheckResult
Role = Literal['engineer', 'l1_tech']
def _limit_for_role(subscription: Subscription, role: Role) -> int | None:
if role == 'engineer':
return subscription.seat_limit
if role == 'l1_tech':
return subscription.l1_seat_limit
raise ValueError(f"Unknown role: {role}")
async def check_seat_available(
account: Account,
subscription: Subscription,
role: Role,
db: AsyncSession,
) -> SeatCheckResult:
"""
Count active users with the given role in the account, compare against
the role-specific seat limit on the subscription. Returns availability.
None limit = unlimited (returns available=True).
"""
limit = _limit_for_role(subscription, role)
stmt = (
select(func.count(User.id))
.where(User.account_id == account.id)
.where(User.account_role == role)
.where(User.is_active.is_(True))
)
current = (await db.execute(stmt)).scalar_one()
if limit is None:
return SeatCheckResult(available=True, current=current, limit=None, role=role)
return SeatCheckResult(
available=current < limit,
current=current,
limit=limit,
role=role,
)
async def get_seat_usage(
account: Account,
subscription: Subscription,
db: AsyncSession,
) -> tuple[SeatCheckResult, SeatCheckResult]:
"""Return (engineer, l1_tech) seat-usage tuple for the seat-counter widget."""
eng = await check_seat_available(account, subscription, 'engineer', db)
l1 = await check_seat_available(account, subscription, 'l1_tech', db)
return eng, l1