"""Mandanten-Verwaltung: Firmen anlegen/auflisten inkl. Kennzahlen. Genutzt von: - routers/reseller.py – Reseller verwaltet seine EIGENEN Firmen - routers/tenants.py – SUPER_ADMIN verwaltet ALLE Firmen + Reseller RLS sorgt für die Sichtbarkeitsgrenze: ein Reseller (app.reseller_id) sieht via companies-/users-Policy automatisch nur seine eigenen Firmen. Das Anlegen einer Firma bootstrappt unter kurzzeitigem Bypass (privilegierte, auditierte Operation) und stellt danach den RLS-Kontext des Actors wieder her. """ import re import secrets from datetime import datetime, timedelta, timezone from uuid import UUID from fastapi import HTTPException from sqlalchemy import func, select, text from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import generate_invite_token, hash_password from app.models.audit_log import AuditLog from app.models.company import Company from app.models.user import User, UserRole from app.schemas.reseller import TenantCreate, TenantOut def _slugify(name: str) -> str: return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")[:80] class TenantService: async def _unique_slug(self, name: str, db: AsyncSession) -> str: base = _slugify(name) or "firma" slug, counter = base, 1 while await db.scalar(select(Company).where(Company.slug == slug)): slug = f"{base}-{counter}" counter += 1 return slug async def _unique_email(self, local_part: str, domain: str, db: AsyncSession) -> str: """Erzeugt eine eindeutige, syntaktisch gültige interne Login-Kennung, falls keine echte E-Mail angegeben wurde (z.B. 'max.mustermann@alpha-gmbh.local').""" local = re.sub(r"[^a-z0-9.]+", ".", local_part.lower()).strip(".") or "admin" dom = re.sub(r"[^a-z0-9.-]+", "-", domain.lower()).strip("-.") or "tenant" email, counter = f"{local}@{dom}.local", 1 while await db.scalar(select(User).where(User.email == email)): email = f"{local}{counter}@{dom}.local" counter += 1 return email async def create_tenant( self, data: TenantCreate, actor: User, db: AsyncSession, reseller_id: UUID | None, ) -> tuple[Company, User, str | None]: """Legt Firma + Erst-Admin an. Mit admin_email → Einladung per Mail (Passwort wird beim Accept gesetzt). Ohne admin_email → interne Login-Kennung + Temp-Passwort (einmalig zurück- gegeben als 3. Tupel-Element), Account sofort aktiv, kein Mailversand. """ # E-Mail global eindeutig (User.email ist unique) if data.admin_email and await db.scalar(select(User).where(User.email == data.admin_email)): raise HTTPException(status_code=400, detail="E-Mail ist bereits vergeben.") slug = await self._unique_slug(data.name, db) # Bootstrap unter Bypass – neue, leere Firma; danach Kontext wiederherstellen. await db.execute(text("SET LOCAL app.bypass_rls = 'on'")) initial_password: str | None = None try: company = Company( name=data.name, slug=slug, plan=data.plan or "trial", country=data.country or "DE", reseller_id=reseller_id, is_active=True, ) db.add(company) await db.flush() if data.admin_email: raw_token, token_hash = generate_invite_token() admin = User( company_id=company.id, email=data.admin_email, first_name=data.admin_first_name, last_name=data.admin_last_name, role=UserRole.COMPANY_ADMIN, password_hash=hash_password(raw_token), # temporär, beim Accept überschrieben invite_token_hash=token_hash, invite_expires=datetime.now(timezone.utc) + timedelta(days=7), is_active=False, ) else: # Kein Mailversand: interne Kennung + sofort nutzbares Temp-Passwort login = await self._unique_email( f"{data.admin_first_name}.{data.admin_last_name}", slug, db ) initial_password = secrets.token_urlsafe(9) admin = User( company_id=company.id, email=login, first_name=data.admin_first_name, last_name=data.admin_last_name, role=UserRole.COMPANY_ADMIN, password_hash=hash_password(initial_password), is_active=True, ) db.add(admin) await db.flush() from app.services.absence_service import absence_service await absence_service.create_defaults_for_company(company.id, db) db.add(AuditLog( company_id=company.id, user_id=actor.id, action="tenant_created", entity_type="company", entity_id=company.id, new_value={ "name": company.name, "reseller_id": str(reseller_id) if reseller_id else None, "admin_email": admin.email, "invited": bool(data.admin_email), }, )) await db.flush() if data.admin_email: from app.services.email_service import email_service await email_service.send_invite(admin, actor, raw_token, db) finally: # RLS-Kontext des Actors wiederherstellen if actor.role == UserRole.RESELLER: await db.execute(text("SET LOCAL app.bypass_rls = 'off'")) # SUPER_ADMIN: Bypass bleibt aktiv (Normalzustand) return company, admin, initial_password async def list_tenants(self, db: AsyncSession) -> list[TenantOut]: """Alle für den aktuellen RLS-Kontext sichtbaren Firmen + Kennzahlen.""" companies = (await db.scalars(select(Company).order_by(Company.name))).all() if not companies: return [] ids = [c.id for c in companies] # Nutzeranzahl pro Firma counts: dict[UUID, tuple[int, int]] = {} rows = await db.execute( select( User.company_id, func.count(User.id), func.count(User.id).filter(User.is_active.is_(True)), ).where(User.company_id.in_(ids)).group_by(User.company_id) ) for cid, total, active in rows: counts[cid] = (total, active) # Erst-Admin-E-Mail pro Firma admin_email: dict[UUID, str] = {} arows = await db.execute( select(User.company_id, User.email) .where(User.company_id.in_(ids), User.role == UserRole.COMPANY_ADMIN) .order_by(User.company_id, User.created_at) ) for cid, email in arows: admin_email.setdefault(cid, email) # Reseller-Namen reseller_ids = [c.reseller_id for c in companies if c.reseller_id] reseller_name: dict[UUID, str] = {} if reseller_ids: rrows = await db.execute( select(User.id, User.first_name, User.last_name).where(User.id.in_(reseller_ids)) ) for rid, fn, ln in rrows: reseller_name[rid] = f"{fn} {ln}".strip() out: list[TenantOut] = [] for c in companies: total, active = counts.get(c.id, (0, 0)) out.append(TenantOut( id=c.id, name=c.name, slug=c.slug, plan=c.plan, is_active=c.is_active, country=c.country, reseller_id=c.reseller_id, reseller_name=reseller_name.get(c.reseller_id) if c.reseller_id else None, user_count=total, active_user_count=active, admin_email=admin_email.get(c.id), )) return out tenant_service = TenantService()