"""Customer WhatsApp Embedded Signup provisioning. Tokens and the platform verify token never leave this module's responses. Each WABA and phone number is stored against one Urmi user and cannot be addressed by another tenant. """ from __future__ import annotations import logging import uuid from datetime import datetime, timezone from typing import Any import httpx from fastapi import HTTPException, status from sqlalchemy import or_, select from sqlalchemy.orm import Session, selectinload from .config import settings from .models import ( UserAccount, WhatsAppBusinessAccount, WhatsAppConnectionAudit, WhatsAppMetaConnection, WhatsAppPhoneAsset, WhatsAppWebhookEvent, ) from .whatsapp_config import resolved_whatsapp_app_secret logger = logging.getLogger(__name__) RECOMMENDED_WEBHOOK_FIELDS = [ "messages", "message_template_status_update", "message_template_quality_update", "account_update", "phone_number_name_update", "phone_number_quality_update", "business_capability_update", ] def signup_public_config(db: Session) -> dict[str, Any]: app_id = settings.meta_app_id.strip() config_id = settings.meta_embedded_signup_config_id.strip() secret = resolved_whatsapp_app_secret(db) return { "app_id": app_id or None, "config_id": config_id or None, "api_version": settings.whatsapp_api_version, "ready": bool(app_id and config_id and secret), } def _audit( db: Session, user: UserAccount, event: str, detail: str, *, connection_id: Any = None, waba_id: str | None = None, ) -> None: db.add( WhatsAppConnectionAudit( user_id=user.id, connection_id=connection_id, waba_id=waba_id, event=event, detail=detail[:500], ) ) def mask_identifier(value: str | None) -> str | None: if not value: return None tail = value[-4:] return f"••••••••{tail}" def _quality_label(value: str | None) -> str | None: if not value: return None normalized = value.strip().upper() return { "GREEN": "High", "HIGH": "High", "YELLOW": "Medium", "MEDIUM": "Medium", "RED": "Low", "LOW": "Low", "UNKNOWN": "Unknown", }.get(normalized, value.title()) def _graph_error_message(data: Any, fallback: str) -> str: if isinstance(data, dict): error = data.get("error") if isinstance(error, dict) and error.get("message"): return str(error["message"])[:500] return fallback async def _graph( method: str, path: str, *, token: str | None = None, params: dict[str, Any] | None = None, json_body: dict[str, Any] | None = None, ) -> dict[str, Any]: url = f"https://graph.facebook.com/{settings.whatsapp_api_version}/{path.lstrip('/')}" headers = {"Authorization": f"Bearer {token}"} if token else {} async with httpx.AsyncClient(timeout=60.0) as client: response = await client.request(method, url, params=params, json=json_body, headers=headers) try: data = response.json() except ValueError: data = {} if response.status_code >= 400: raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=_graph_error_message(data, "Meta could not complete WhatsApp setup"), ) return data if isinstance(data, dict) else {} async def _exchange_code(db: Session, code: str) -> str: app_id = settings.meta_app_id.strip() secret = resolved_whatsapp_app_secret(db) if not app_id or not secret: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="WhatsApp Embedded Signup is not configured", ) data = await _graph( "GET", "oauth/access_token", params={"client_id": app_id, "client_secret": secret, "code": code}, ) token = str(data.get("access_token") or "") if not token: raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail="Meta did not return an authorization token", ) return token async def _debug_token(db: Session, token: str) -> dict[str, Any]: app_id = settings.meta_app_id.strip() secret = resolved_whatsapp_app_secret(db) data = await _graph( "GET", "debug_token", params={"input_token": token, "access_token": f"{app_id}|{secret}"}, ) payload = data.get("data") return payload if isinstance(payload, dict) else {} def _waba_ids_from_debug(debug: dict[str, Any]) -> list[str]: found: list[str] = [] scopes = debug.get("granular_scopes") if not isinstance(scopes, list): return found preferred = {"whatsapp_business_management", "whatsapp_business_messaging"} for item in scopes: if not isinstance(item, dict): continue if item.get("scope") not in preferred: continue for target in item.get("target_ids") or []: value = str(target) if value and value not in found: found.append(value) return found def _as_uuid(account_id: Any) -> uuid.UUID: try: return account_id if isinstance(account_id, uuid.UUID) else uuid.UUID(str(account_id)) except ValueError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="WhatsApp account not found") from exc def _owned_account(db: Session, user: UserAccount, account_id: Any) -> WhatsAppBusinessAccount: account = db.scalar( select(WhatsAppBusinessAccount) .options(selectinload(WhatsAppBusinessAccount.phones), selectinload(WhatsAppBusinessAccount.connection)) .where(WhatsAppBusinessAccount.id == _as_uuid(account_id), WhatsAppBusinessAccount.user_id == user.id) ) if account is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="WhatsApp account not found") return account def _serialize_phone(phone: WhatsAppPhoneAsset) -> dict[str, Any]: return { "id": str(phone.id), "phone_number_id": phone.phone_number_id, "display_phone": phone.display_phone, "verified_name": phone.verified_name, "quality": phone.quality, "status": phone.status, "code_verification_status": phone.code_verification_status, } def serialize_account(account: WhatsAppBusinessAccount) -> dict[str, Any]: phones = [_serialize_phone(phone) for phone in account.phones] primary = next((phone for phone in phones if phone["status"] == "connected"), phones[0] if phones else None) connection = account.connection usable = account.status == "connected" return { "id": str(account.id), "waba_id_masked": mask_identifier(account.waba_id), "name": account.name, "business_name": connection.business_name if connection else None, "business_id_masked": mask_identifier(connection.meta_business_id) if connection else None, "status": account.status, "usable": usable, "messaging_status": account.messaging_status if usable else "unavailable", "management_status": account.management_status if usable or account.status == "incomplete" else "unavailable", "templates_status": account.templates_status if account.templates_status == "active" else "unavailable", "webhooks_status": account.webhooks_status if account.webhooks_status == "active" else "unavailable", "template_count": account.template_count, "quality": primary["quality"] if primary else None, "primary_phone": primary, "phones": phones, "last_error": account.last_error, "updated_at": account.updated_at, } def list_accounts(db: Session, user: UserAccount) -> list[dict[str, Any]]: rows = db.scalars( select(WhatsAppBusinessAccount) .options(selectinload(WhatsAppBusinessAccount.phones), selectinload(WhatsAppBusinessAccount.connection)) .where(WhatsAppBusinessAccount.user_id == user.id) .order_by(WhatsAppBusinessAccount.created_at.desc()) ).all() return [serialize_account(row) for row in rows] async def _load_waba(token: str, waba_id: str) -> dict[str, Any]: return await _graph( "GET", waba_id, token=token, params={"fields": "id,name,account_review_status,owner_business_info"}, ) async def _load_phones(token: str, waba_id: str) -> list[dict[str, Any]]: data = await _graph( "GET", f"{waba_id}/phone_numbers", token=token, params={ "fields": "id,display_phone_number,verified_name,quality_rating,code_verification_status,status", "limit": "50", }, ) rows = data.get("data") return [row for row in rows if isinstance(row, dict)] if isinstance(rows, list) else [] async def _subscribe(token: str, waba_id: str) -> bool: try: await _graph("POST", f"{waba_id}/subscribed_apps", token=token) return True except HTTPException: logger.warning("WhatsApp subscribed_apps failed for waba %s", waba_id[-4:]) return False async def _template_count(token: str, waba_id: str) -> tuple[int, bool]: try: data = await _graph( "GET", f"{waba_id}/message_templates", token=token, params={"limit": "1", "summary": "total_count"}, ) except HTTPException: return 0, False summary = data.get("summary") if isinstance(summary, dict) and isinstance(summary.get("total_count"), int): return summary["total_count"], True rows = data.get("data") return (len(rows) if isinstance(rows, list) else 0), True def _claim_waba(db: Session, user: UserAccount, waba_id: str) -> WhatsAppBusinessAccount | None: existing = db.scalar(select(WhatsAppBusinessAccount).where(WhatsAppBusinessAccount.waba_id == waba_id)) if existing is not None and existing.user_id != user.id: return None return existing def _apply_phones( db: Session, user: UserAccount, account: WhatsAppBusinessAccount, phones: list[dict[str, Any]], session_phone_id: str | None, ) -> None: seen = {str(item.get("id")) for item in phones if item.get("id")} if session_phone_id and session_phone_id not in seen: phones.append({"id": session_phone_id}) current = {phone.phone_number_id: phone for phone in account.phones} for item in phones: phone_id = str(item.get("id") or "") if not phone_id: continue other = db.scalar(select(WhatsAppPhoneAsset).where(WhatsAppPhoneAsset.phone_number_id == phone_id)) if other is not None and other.user_id != user.id: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="This phone number is already connected to another Urmi workspace", ) raw_status = str(item.get("status") or "PENDING").upper() phone_status = "connected" if raw_status == "CONNECTED" else raw_status.lower() row = current.get(phone_id) or other if row is None: row = WhatsAppPhoneAsset( user_id=user.id, account_id=account.id, phone_number_id=phone_id, ) db.add(row) if row not in account.phones: account.phones.append(row) row.user_id = user.id row.account_id = account.id row.display_phone = str(item.get("display_phone_number") or row.display_phone or "") row.verified_name = item.get("verified_name") or row.verified_name row.quality = _quality_label(item.get("quality_rating")) row.status = phone_status row.code_verification_status = item.get("code_verification_status") async def _provision_waba( db: Session, user: UserAccount, connection: WhatsAppMetaConnection, token: str, waba_id: str, session_phone_id: str | None, ) -> WhatsAppBusinessAccount: claimed = _claim_waba(db, user, waba_id) if claimed is None and db.scalar(select(WhatsAppBusinessAccount.id).where(WhatsAppBusinessAccount.waba_id == waba_id)): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="This WhatsApp Business Account is already connected to another Urmi workspace", ) try: waba = await _load_waba(token, waba_id) except HTTPException as exc: account = claimed or WhatsAppBusinessAccount( user_id=user.id, connection_id=connection.id, waba_id=waba_id, name="WhatsApp Business", ) if claimed is None: db.add(account) db.flush() detail = str(exc.detail)[:500] expired = "expired" in detail.lower() or "validating access token" in detail.lower() account.status = "expired" if expired else "failed" account.last_error = detail account.messaging_status = "unavailable" account.management_status = "unavailable" connection.status = "failed" connection.last_error = account.last_error _audit(db, user, "provision_failed", account.last_error or "", connection_id=connection.id, waba_id=waba_id) return account owner = waba.get("owner_business_info") if isinstance(waba.get("owner_business_info"), dict) else {} business_id = str(owner.get("id") or connection.meta_business_id or "") business_name = str(owner.get("name") or connection.business_name or waba.get("name") or "WhatsApp Business") connection.meta_business_id = business_id or connection.meta_business_id connection.business_name = business_name connection.status = "connected" connection.last_error = None connection.access_token = token account = claimed or WhatsAppBusinessAccount( user_id=user.id, connection_id=connection.id, waba_id=waba_id, ) if claimed is None: db.add(account) db.flush() previous_connection_id = account.connection_id if claimed is not None else None account.user_id = user.id account.connection_id = connection.id if previous_connection_id and previous_connection_id != connection.id: previous = db.get(WhatsAppMetaConnection, previous_connection_id) if previous is not None and previous.user_id == user.id: previous.access_token = None account.name = str(waba.get("name") or business_name) account.last_error = None account.management_status = "active" try: phones = await _load_phones(token, waba_id) except HTTPException as exc: phones = [] account.last_error = str(exc.detail)[:500] _apply_phones(db, user, account, phones, session_phone_id) subscribed = await _subscribe(token, waba_id) template_count, templates_ok = await _template_count(token, waba_id) account.template_count = template_count account.templates_status = "active" if templates_ok else "unavailable" account.webhooks_status = "active" if subscribed else "unavailable" has_connected_phone = any(phone.status == "connected" for phone in account.phones) if has_connected_phone and account.management_status == "active": account.status = "connected" account.messaging_status = "active" _audit(db, user, "connected", "WhatsApp Business connection verified", connection_id=connection.id, waba_id=waba_id) else: account.status = "incomplete" account.messaging_status = "unavailable" account.last_error = account.last_error or "Phone number verification is not complete" _audit( db, user, "provision_incomplete", account.last_error, connection_id=connection.id, waba_id=waba_id, ) return account async def complete_signup(db: Session, user: UserAccount, payload: dict[str, Any]) -> dict[str, Any]: event = str(payload.get("event") or "").upper() error_message = str(payload.get("error_message") or "").strip()[:500] if event == "CANCEL": _audit(db, user, "signup_cancelled", "Customer closed Embedded Signup") db.commit() return {"status": "cancelled", "message": "WhatsApp setup was cancelled. Nothing was connected.", "accounts": []} if event == "ERROR" or event == "DENIED": _audit(db, user, "signup_denied", error_message or "Meta denied the requested permissions") db.commit() return { "status": "denied", "message": error_message or "Meta did not grant the permissions Urmi needs. You can try again.", "accounts": [], } if not event.startswith("FINISH"): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unrecognized Embedded Signup result") code = str(payload.get("code") or "").strip() if not code: _audit(db, user, "provision_failed", "Embedded Signup finished without an authorization code") db.commit() return { "status": "failed", "message": "Meta finished without an authorization code. Start setup again.", "accounts": [], } token = await _exchange_code(db, code) debug = await _debug_token(db, token) if debug.get("is_valid") is False: _audit(db, user, "authorization_expired", "Meta authorization was not valid") db.commit() return { "status": "expired", "message": "The Meta authorization expired before Urmi could finish setup. Start setup again.", "accounts": [], } session_waba = str(payload.get("waba_id") or "").strip() waba_ids = [session_waba] if session_waba else _waba_ids_from_debug(debug) if not waba_ids: _audit(db, user, "provision_failed", "Embedded Signup did not return a WhatsApp Business Account") db.commit() return { "status": "failed", "message": "Meta did not return a WhatsApp Business Account. Finish business and number setup in the Meta window, then try again.", "accounts": [], } scopes = debug.get("scopes") granted = ",".join(str(item) for item in scopes) if isinstance(scopes, list) else None connection = WhatsAppMetaConnection( user_id=user.id, status="pending", meta_business_id=str(payload.get("business_id") or "") or None, granted_scopes=granted, access_token=token, ) db.add(connection) db.flush() _audit(db, user, "authorization_received", "Embedded Signup returned an authorization code", connection_id=connection.id) session_phone = str(payload.get("phone_number_id") or "").strip() or None accounts: list[WhatsAppBusinessAccount] = [] for waba_id in waba_ids: accounts.append(await _provision_waba(db, user, connection, token, waba_id, session_phone)) db.commit() for account in accounts: db.refresh(account) statuses = {account.status for account in accounts} if "connected" in statuses and statuses <= {"connected"}: overall = "connected" message = "WhatsApp Business is connected." elif "connected" in statuses: overall = "connected" message = "Some WhatsApp Business accounts are connected. Others still need attention." elif "incomplete" in statuses: overall = "incomplete" message = "Meta authorization succeeded, but phone verification is not complete. Finish verification in Meta, then reconnect." else: overall = "failed" message = accounts[0].last_error or "Urmi could not finish WhatsApp setup." return {"status": overall, "message": message, "accounts": [serialize_account(account) for account in accounts]} async def refresh_account(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]: account = _owned_account(db, user, account_id) connection = account.connection token = connection.access_token if connection else None if connection is None or not token or connection.status == "disconnected": account.status = "expired" account.messaging_status = "unavailable" account.last_error = "Authorization expired. Reconnect this WhatsApp account." _audit(db, user, "authorization_expired", account.last_error, connection_id=connection.id if connection else None, waba_id=account.waba_id) db.commit() db.refresh(account) return serialize_account(account) await _provision_waba(db, user, connection, token, account.waba_id, None) db.commit() db.refresh(account) return serialize_account(account) async def disconnect_account(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]: account = _owned_account(db, user, account_id) connection = account.connection token = connection.access_token if connection else None if token: try: await _graph("DELETE", f"{account.waba_id}/subscribed_apps", token=token) except HTTPException: logger.info("Could not unsubscribe WABA %s during disconnect", account.waba_id[-4:]) account.status = "disconnected" account.messaging_status = "unavailable" account.management_status = "unavailable" account.templates_status = "unavailable" account.webhooks_status = "unavailable" account.last_error = None for phone in account.phones: phone.status = "disconnected" if connection is not None: siblings_remain = ( db.scalar( select(WhatsAppBusinessAccount.id).where( WhatsAppBusinessAccount.connection_id == connection.id, WhatsAppBusinessAccount.id != account.id, WhatsAppBusinessAccount.status != "disconnected", ) ) is not None ) if not siblings_remain: connection.status = "disconnected" connection.access_token = None connection.disconnected_at = datetime.now(timezone.utc) _audit(db, user, "disconnected", "Customer disconnected this WhatsApp Business account", connection_id=connection.id if connection else None, waba_id=account.waba_id) db.commit() db.refresh(account) return serialize_account(account) def account_detail(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]: return serialize_account(_owned_account(db, user, account_id)) async def list_templates(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]: account = _owned_account(db, user, account_id) if account.templates_status != "active" or account.connection is None or not account.connection.access_token: return {"available": False, "templates": []} data = await _graph( "GET", f"{account.waba_id}/message_templates", token=account.connection.access_token, params={"fields": "name,category,language,status,quality_score", "limit": "50"}, ) rows = data.get("data") if isinstance(data.get("data"), list) else [] templates = [] for row in rows: if not isinstance(row, dict): continue quality = row.get("quality_score") score = quality.get("score") if isinstance(quality, dict) else None templates.append( { "name": row.get("name"), "category": row.get("category"), "language": row.get("language"), "status": row.get("status"), "quality": _quality_label(str(score) if score else None), } ) account.template_count = len(templates) db.commit() return {"available": True, "templates": templates} def account_logs(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]: account = _owned_account(db, user, account_id) audits = db.scalars( select(WhatsAppConnectionAudit) .where( WhatsAppConnectionAudit.user_id == user.id, or_( WhatsAppConnectionAudit.waba_id == account.waba_id, WhatsAppConnectionAudit.connection_id == account.connection_id, ), ) .order_by(WhatsAppConnectionAudit.created_at.desc()) .limit(50) ).all() phone_ids = [phone.phone_number_id for phone in account.phones] event_filter = [WhatsAppWebhookEvent.waba_id == account.waba_id] if phone_ids: event_filter.append(WhatsAppWebhookEvent.phone_number_id.in_(phone_ids)) events = db.scalars( select(WhatsAppWebhookEvent) .where(or_(*event_filter)) .order_by(WhatsAppWebhookEvent.created_at.desc()) .limit(25) ).all() return { "audits": [ { "id": str(item.id), "event": item.event, "detail": item.detail, "created_at": item.created_at, } for item in audits ], "events": [ { "id": str(item.id), "field": item.field, "event_type": item.event_type, "from_wa_id": item.from_wa_id, "status": item.status, "created_at": item.created_at, } for item in events ], } def developer_config(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]: account = _owned_account(db, user, account_id) return { "webhook_url": settings.whatsapp_webhook_url, "verify_token_configured": bool(settings.whatsapp_verify_token), "verify_token_visible": False, "recommended_fields": RECOMMENDED_WEBHOOK_FIELDS, "webhooks_status": account.webhooks_status, "docs_url": "https://developers.facebook.com/documentation/business-messaging/whatsapp/embedded-signup", } def customer_token_for_phone(db: Session, user: UserAccount, phone_number_id: str) -> tuple[str, str]: phone = db.scalar( select(WhatsAppPhoneAsset) .options(selectinload(WhatsAppPhoneAsset.account).selectinload(WhatsAppBusinessAccount.connection)) .where(WhatsAppPhoneAsset.phone_number_id == phone_number_id, WhatsAppPhoneAsset.user_id == user.id) ) if phone is None or phone.account.status != "connected" or phone.status != "connected": raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No connected WhatsApp number for this workspace") token = phone.account.connection.access_token if phone.account.connection else None if not token: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Reconnect this WhatsApp account before sending") return token, phone.phone_number_id