app/core/startup/seeder.py

"""core/startup/seeder.py — Seed the superadmin account on first startup."""

import structlog

logger = structlog.get_logger(__name__)


async def seed_superadmin(cfg) -> None:
    """Create the superadmin account from env vars if it doesn't exist yet.

    Idempotent: if the account already exists, this is a no-op.
    If the account exists but has the wrong role, it is promoted to superadmin.
    """
    if not cfg.SUPERADMIN_EMAIL or not cfg.SUPERADMIN_PASSWORD:
        logger.warning(
            "SUPER_ADMIN_USER / SUPER_ADMIN_PASS not set — "
            "no superadmin account seeded. Register manually via /api/auth/register."
        )
        return

    from sqlalchemy import select
    from app.models.user import User
    from app.core.security import hash_password
    from app.dependencies import get_async_session

    async with get_async_session() as db:
        existing = await db.scalar(select(User).where(User.email == cfg.SUPERADMIN_EMAIL))
        if existing:
            if existing.role != "superadmin":
                existing.role = "superadmin"
                await db.commit()
                logger.info("Promoted existing user to superadmin", email=cfg.SUPERADMIN_EMAIL)
            else:
                logger.info("Superadmin already exists", email=cfg.SUPERADMIN_EMAIL)
            return

        sa = User(
            email=cfg.SUPERADMIN_EMAIL,
            password_hash=hash_password(cfg.SUPERADMIN_PASSWORD),
            role="superadmin",
            full_name="Super Admin",
            is_active=True,
            totp_secret=None,
        )
        db.add(sa)
        await db.commit()
        logger.info(
            "Superadmin account created — TOTP enrollment required on first login",
            email=cfg.SUPERADMIN_EMAIL,
        )