ligbox-ops-platform/projects/wizard/backend/app/services/mail_bundle.py

"""
Spec 035/043 — Provisionamento mail bundle pós-activação Desk/FOSS.

Garante domínio Carbonio, conta gerente (admin@domínio), quotas e metadata no registry.
Deploy: /opt/ligbox-wizard/backend/app/services/mail_bundle.py
"""

from __future__ import annotations

import json
import secrets
import string
from datetime import datetime, timezone
from typing import Any

from app.services import activity_log, carbonio, domain_registry
from app.services.carbonio import CarbonioError

_BUNDLE_VAULT = __import__("pathlib").Path("/var/lib/ligbox-wizard/mail_bundles")


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _generate_password(length: int = 16) -> str:
    alphabet = string.ascii_letters + string.digits + "!@#$%&*"
    while True:
        pwd = "".join(secrets.choice(alphabet) for _ in range(length))
        if any(c.islower() for c in pwd) and any(c.isupper() for c in pwd) and any(c.isdigit() for c in pwd):
            return pwd


def _bundle_path(domain: str) -> __import__("pathlib").Path:
    safe = domain.lower().strip().replace("/", "_")
    return _BUNDLE_VAULT / f"{safe}.json"


def load_bundle(domain: str) -> dict[str, Any] | None:
    p = _bundle_path(domain)
    if not p.is_file():
        return None
    try:
        return json.loads(p.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return None


def save_bundle(domain: str, data: dict[str, Any]) -> None:
    _BUNDLE_VAULT.mkdir(parents=True, exist_ok=True)
    _bundle_path(domain).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def provision_mail_bundle(
    *,
    domain: str,
    admin_email: str,
    admin_name: str | None = None,
    seats: int = 25,
    mail_gb_per_seat: int = 30,
    files_gb_per_seat: int = 200,
    foss_order_id: str | int | None = None,
    admin_password: str | None = None,
) -> dict[str, Any]:
    """
    Idempotente — reutiliza domínio/conta existentes; actualiza quotas e metadata.
    """
    domain = domain.lower().strip().rstrip(".")
    email = admin_email.lower().strip()
    if not email.endswith(f"@{domain}"):
        raise ValueError(f"admin_email deve ser @{domain}")

    existing = load_bundle(domain)
    if existing and existing.get("provisioned"):
        return {
            **existing,
            "already_provisioned": True,
            "mail_host": existing.get("mail_host") or f"mail.{domain}",
        }

    quota_mb = max(1, int(mail_gb_per_seat)) * 1024
    password = admin_password or _generate_password()
    display = admin_name or email.split("@")[0]

    activity_log.info(f"mail-bundle: domínio {domain} seats={seats}", source="internal")
    try:
        if not carbonio.domain_exists(domain):
            try:
                carbonio.create_domain(domain)
            except CarbonioError as exc:
                if "DOMAIN_EXISTS" not in str(exc):
                    raise
                carbonio.set_domain_public_hostname(domain)
        else:
            carbonio.set_domain_public_hostname(domain)

        _msg, reused_account = carbonio.ensure_onboarding_account(
            email, password, display_name=display
        )
        if hasattr(carbonio, "set_mail_quota"):
            carbonio.set_mail_quota(email, quota_mb)
    except CarbonioError as exc:
        activity_log.error(f"mail-bundle Carbonio: {exc}", source="internal")
        raise

    mail_host = f"mail.{domain}"
    bundle = {
        "domain": domain,
        "admin_email": email,
        "admin_name": display,
        "seats": int(seats),
        "mail_gb_per_seat": int(mail_gb_per_seat),
        "files_gb_per_seat": int(files_gb_per_seat),
        "foss_order_id": foss_order_id,
        "mail_host": mail_host,
        "webmail_url": f"https://{mail_host}/",
        "files_url": f"https://files.{domain}/",
        "provisioned": True,
        "provisioned_at": _now(),
        "account_reused": reused_account,
    }
    save_bundle(domain, bundle)

    rec = domain_registry.get_domain_record(domain) or {}
    rec.update(
        {
            "domain": domain,
            "mail_bundle": {
                "seats": bundle["seats"],
                "mail_gb_per_seat": bundle["mail_gb_per_seat"],
                "files_gb_per_seat": bundle["files_gb_per_seat"],
                "foss_order_id": foss_order_id,
                "provisioned_at": bundle["provisioned_at"],
            },
            "manager_email": email,
            "manager_name": display,
        }
    )
    domain_registry.save_domain_record(domain, rec)

    activity_log.ok(f"mail-bundle OK {domain} → {mail_host}", source="internal")
    return {
        **bundle,
        "already_provisioned": False,
        "password_generated": not bool(admin_password),
    }