agentic.email/backend/app/services/org_members.py

"""Organization member provisioning for company admins."""

from __future__ import annotations

import secrets
from datetime import date
from pwdlib import PasswordHash
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import Account
from app.schemas import OrgMemberCreate, OrgMemberRead
from app.services.domain_provisioning import account_is_domain_owner

password_hasher = PasswordHash.recommended()
PLACEHOLDER_BIRTHDATE = date(1990, 1, 1)


async def can_manage_org_members(
    session: AsyncSession,
    account: Account,
) -> bool:
    if account.account_type != "company" or not account.company_domain:
        return False
    if account.member_status == "suspended":
        return False
    if account.org_role == "admin":
        return True
    return await account_is_domain_owner(
        session,
        account.id,
        account.company_domain,
    )


async def member_read(
    session: AsyncSession,
    member: Account,
) -> OrgMemberRead:
    is_owner = await account_is_domain_owner(
        session,
        member.id,
        member.company_domain,
    )
    role = "owner" if is_owner else member.org_role
    return OrgMemberRead(
        id=member.id,
        email=member.email,
        first_name=member.first_name,
        last_name=member.last_name,
        org_role=role,
        member_status=member.member_status,
        is_domain_owner=is_owner,
        groups=[],
        created_at=member.created_at,
    )


async def list_org_members(
    session: AsyncSession,
    company_domain: str,
) -> tuple[list[OrgMemberRead], dict[str, int]]:
    result = await session.scalars(
        select(Account)
        .where(
            Account.account_type == "company",
            Account.company_domain == company_domain,
        )
        .order_by(Account.created_at.asc())
    )
    members = list(result.all())
    items = [await member_read(session, member) for member in members]
    counts = {
        "all": len(items),
        "active": sum(1 for item in items if item.member_status == "active"),
        "invited": sum(1 for item in items if item.member_status == "invited"),
        "suspended": sum(1 for item in items if item.member_status == "suspended"),
    }
    return items, counts


def _placeholder_phone() -> str:
    digits = "".join(str(secrets.randbelow(10)) for _ in range(10))
    return f"+1{digits}"


def _email_local_part(email: str) -> str:
    return email.split("@", 1)[0].lower()


async def create_org_member(
    session: AsyncSession,
    *,
    manager: Account,
    payload: OrgMemberCreate,
) -> OrgMemberRead:
    company_domain = manager.company_domain
    if company_domain is None:
        raise ValueError("Company domain is required.")

    email = str(payload.email).lower()
    if not email.endswith(f"@{company_domain}"):
        raise ValueError(f"Email must use your company domain @{company_domain}.")

    if payload.send_invitation:
        raw_password = secrets.token_urlsafe(24)
        member_status = "invited"
    else:
        if not payload.password:
            raise ValueError("Password is required when invitation email is disabled.")
        raw_password = payload.password
        member_status = "active"

    existing = await session.scalar(
        select(Account).where(
            func.lower(Account.email) == email,
        )
    )
    if existing:
        raise ValueError("That email address is already registered.")

    phone_number = _placeholder_phone()
    while await session.scalar(
        select(Account).where(Account.phone_number == phone_number)
    ):
        phone_number = _placeholder_phone()

    member = Account(
        account_type="company",
        email=email,
        first_name=payload.first_name.strip(),
        last_name=payload.last_name.strip(),
        date_of_birth=PLACEHOLDER_BIRTHDATE,
        phone_number=phone_number,
        company_domain=company_domain,
        org_role=payload.org_role,
        member_status=member_status,
        created_by_id=manager.id,
        password_hash=password_hasher.hash(raw_password),
    )
    session.add(member)
    await session.flush()
    return await member_read(session, member)


async def update_org_member_role(
    session: AsyncSession,
    *,
    manager: Account,
    member_id,
    org_role: str,
) -> OrgMemberRead:
    company_domain = manager.company_domain
    if company_domain is None:
        raise ValueError("Company domain is required.")

    if not await account_is_domain_owner(
        session,
        manager.id,
        company_domain,
    ):
        raise ValueError("Only the domain owner can assign roles.")

    member = await session.scalar(
        select(Account).where(
            Account.id == member_id,
            Account.company_domain == company_domain,
        )
    )
    if member is None:
        raise ValueError("User not found in your organization.")

    if await account_is_domain_owner(session, member.id, company_domain):
        raise ValueError("The domain owner role cannot be changed.")

    if member.id == manager.id:
        raise ValueError("You cannot change your own role.")

    member.org_role = org_role
    await session.flush()
    return await member_read(session, member)