replypier/backend/app/channels/meta_base.py

"""
Copyright 2024-2026 ReplyPier

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

import asyncio
import hashlib
import hmac
from datetime import datetime, timezone
from typing import ClassVar, Iterable, Optional

import httpx

from app.channels.base import ChannelAdapter, SendResult, WindowStatus
from app.core.config import settings
from app.core.security import decrypt_credentials
from app.models.channels import ChannelAccount, ChannelConversation
from app.core.logger import get_logger
from app.core.external_urls import external_base

logger = get_logger(__name__)

GRAPH_BASE = external_base("META_GRAPH_BASE", "https://graph.facebook.com")
# Instagram API with Instagram Login speaks to its own host with an Instagram
# user token; the Facebook graph rejects those. Page-based channels keep using
# GRAPH_BASE, so every helper below takes the base rather than assuming one.
GRAPH_INSTAGRAM_BASE = external_base("META_GRAPH_INSTAGRAM_BASE", "https://graph.instagram.com")
INSTAGRAM_OAUTH_BASE = "https://api.instagram.com"
REQUEST_TIMEOUT_SECONDS = 15.0
SEND_RETRIES = 2
# Graph error codes meaning the access token is unusable: 190 expired or
# invalidated token, 102 expired session, 0 WhatsApp's "Authentication Error"
GRAPH_AUTH_ERROR_CODES = frozenset({0, 102, 190})
# Meta customer-service window: free-form replies allowed for 24h after the
# customer's last inbound message (applies to WhatsApp, Messenger and Instagram).
WINDOW_HOURS = 24

# Shared keep-alive client to graph.facebook.com (process-lifetime, like the
# Telegram adapter's) — avoids a TCP+TLS handshake per outbound message.
_http_client: Optional[httpx.AsyncClient] = None


def _get_http_client() -> httpx.AsyncClient:
    global _http_client
    if _http_client is None or _http_client.is_closed:
        _http_client = httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SECONDS)
    return _http_client


def graph_url(path: str, base: str = GRAPH_BASE) -> str:
    return f"{base}/{settings.META_GRAPH_VERSION}/{path}"


def verify_meta_signature(raw_body: bytes, signature_header: str) -> bool:
    """Validate the X-Hub-Signature-256 header against our app secrets.

    The signature covers the raw request body; it must be checked before the
    body is parsed. Uses a constant-time comparison.

    Instagram Login webhooks are signed with the Instagram app secret rather
    than the Meta one, and the only thing that would say which product sent this
    is inside the body we have not authenticated yet. So this tries each secret
    we own instead of trusting unverified content to choose one; an attacker
    still has to forge one of them.
    """
    return signed_by_any(raw_body, signature_header,
                         (settings.META_APP_SECRET, settings.INSTAGRAM_APP_SECRET))


def signed_by_any(raw_body: bytes, signature_header: str, app_secrets: Iterable[Optional[str]]) -> bool:
    """Whether X-Hub-Signature-256 was made with one of these app secrets
    (constant-time; empty secrets are skipped)."""
    if not signature_header or not signature_header.startswith("sha256="):
        return False
    provided = signature_header[len("sha256="):]
    for app_secret in app_secrets:
        if not app_secret:
            continue
        expected = hmac.new(app_secret.encode(), raw_body, hashlib.sha256).hexdigest()
        if hmac.compare_digest(expected, provided):
            return True
    return False


def verify_challenge(mode: Optional[str], token: Optional[str], challenge: Optional[str]) -> Optional[str]:
    """Answer a webhook GET verification handshake.

    Returns the challenge string to echo when the subscribe token matches, else
    None (caller should 403).
    """
    if mode == "subscribe" and token and token == settings.META_WEBHOOK_VERIFY_TOKEN:
        return challenge
    return None


async def graph_post(path: str, access_token: str, payload: dict,
                     base: str = GRAPH_BASE) -> SendResult:
    """POST to the Graph API with small exponential backoff on transient errors.

    Returns a SendResult; 4xx errors are terminal (no retry), network/5xx are
    retried up to SEND_RETRIES times. `base` selects the host — Instagram Login
    accounts send via graph.instagram.com, everything else via graph.facebook.com.
    """
    url = graph_url(path, base)
    headers = {"Authorization": f"Bearer {access_token}"}
    last_error = "unknown error"
    for attempt in range(SEND_RETRIES + 1):
        try:
            response = await _get_http_client().post(url, json=payload, headers=headers)
            if response.status_code < 300:
                data = response.json()
                message_id = _extract_message_id(data)
                return SendResult(ok=True, external_message_id=message_id)
            # 4xx are our fault (bad token, closed window) — don't retry
            if response.status_code < 500:
                return SendResult(ok=False, error=_graph_error(response),
                                  auth_failed=_is_auth_error(response))
            last_error = _graph_error(response)
        except Exception as e:
            last_error = str(e)
        if attempt < SEND_RETRIES:
            await asyncio.sleep(0.5 * (2 ** attempt))
    return SendResult(ok=False, error=last_error)


async def _graph_request(method: str, path: str, access_token: Optional[str],
                         params: Optional[dict] = None,
                         json_body: Optional[dict] = None,
                         base: str = GRAPH_BASE) -> tuple[bool, dict]:
    """Issue a Graph call and return (ok, decoded-json).

    Unlike graph_post this keeps the whole response body, for calls where the
    body is the result rather than just a message id. The token goes in the
    Authorization header so it never appears in URLs or proxy logs; pass None
    for the one endpoint that authenticates with app credentials instead. These
    are management calls, not sends, so a failure surfaces to the caller
    instead of being retried.
    """
    try:
        response = await _get_http_client().request(
            method,
            graph_url(path, base),
            params=params or None,
            json=json_body,
            headers={"Authorization": f"Bearer {access_token}"} if access_token else {},
        )
        return (response.status_code < 300, response.json())
    except Exception as e:
        logger.error(f"Graph {method} {path} failed: {e}")
        return (False, {"error": {"message": str(e)}})


async def graph_get(path: str, access_token: str, params: Optional[dict] = None,
                    base: str = GRAPH_BASE) -> tuple[bool, dict]:
    """GET a Graph node (validating tokens during onboarding, listing templates)."""
    return await _graph_request("GET", path, access_token, params=params, base=base)


def graph_detail(data: dict, fallback: str) -> str:
    """Meta's own error text where it has one, so the UI can show the real
    reason (e.g. a template that cannot be deleted) rather than a generic
    failure.

    error_user_msg first: `message` is often a generic OAuth string that says
    nothing actionable — Graph will pair a `message` of "Application does not
    have permission for this action" with an error_user_msg that names the
    actual asset and rule. Only one of those is a clue.
    """
    error = data.get("error", {})
    if not isinstance(error, dict):
        return fallback
    return error.get("error_user_msg") or error.get("message") or fallback


async def graph_post_json(path: str, access_token: str, payload: dict) -> tuple[bool, dict]:
    """POST to a Graph node, keeping the response body (e.g. creating a message
    template returns its id, status and category)."""
    return await _graph_request("POST", path, access_token, json_body=payload)


async def debug_token(input_token: str) -> tuple[bool, dict]:
    """Inspect a token: what it is, who it belongs to, and whether it is ours.

    Returns Meta's `data` object, which for a Page token carries
    `type: "PAGE"`, `profile_id` (the page id), `app_id`, `is_valid` and the
    granted `scopes`.

    This authenticates with app credentials rather than the token being
    inspected, so it needs no permission on the asset. Reading the Page node
    directly (`me?fields=id,name`) would instead demand `pages_read_engagement`
    — unrelated to messaging, and absent from a token that can nonetheless send
    perfectly well.
    """
    app_token = f"{settings.META_APP_ID}|{settings.META_APP_SECRET}"
    ok, data = await _graph_request("GET", "debug_token", app_token,
                                    params={"input_token": input_token})
    if not ok:
        return False, data
    return True, data.get("data", {})


async def exchange_signup_code(code: str, redirect_uri: Optional[str] = None) -> tuple[bool, dict]:
    """Trade a login code for the customer's access token.

    The app credentials authenticate this call, so it is the one Graph request
    that carries no bearer token.

    redirect_uri must match whatever the login dialog used, or Graph rejects the
    code as a mismatch:
      - WhatsApp Embedded Signup issues its code through the JS SDK with no
        redirect_uri, so it passes None and we omit the param.
      - Facebook Login for Business runs as a first-party OAuth popup bound to
        our own callback URL, so that flow passes that exact callback URL.
    """
    params = {
        "client_id": settings.META_APP_ID,
        "client_secret": settings.META_APP_SECRET,
        "code": code,
    }
    if redirect_uri is not None:
        params["redirect_uri"] = redirect_uri
    return await _graph_request("GET", "oauth/access_token", None, params=params)


async def exchange_for_long_lived_token(short_lived_token: str) -> tuple[bool, dict]:
    """Trade a user token for a 60-day one.

    Page tokens inherit their lifetime from the user token they were read with:
    from a short-lived one they die in about an hour, from a long-lived one they
    never expire. We store the Page token and send with it for months, so this
    runs before /me/accounts. An already-long-lived token comes back unchanged,
    so it is safe to call unconditionally. App credentials authenticate it, so
    like exchange_signup_code it carries no bearer token.
    """
    return await _graph_request("GET", "oauth/access_token", None, params={
        "grant_type": "fb_exchange_token",
        "client_id": settings.META_APP_ID,
        "client_secret": settings.META_APP_SECRET,
        "fb_exchange_token": short_lived_token,
    })


def instagram_token_payload(data: dict) -> dict:
    """The token object out of an Instagram Login exchange response.

    Instagram has returned this both flat and wrapped in a single-element `data`
    list depending on the API version, and the difference is invisible until it
    silently yields no token — so accept either shape.
    """
    entries = data.get("data")
    if isinstance(entries, list) and entries and isinstance(entries[0], dict):
        return entries[0]
    return data


async def exchange_instagram_code(code: str, redirect_uri: str) -> tuple[bool, dict]:
    """Trade an Instagram Business Login code for a short-lived user token.

    Instagram Login is a separate OAuth from the Facebook graph: its own host,
    its own app credentials, form-encoded rather than query params, and it
    returns an Instagram user token plus the account's user_id — no Page and no
    Page token anywhere in the flow.
    """
    try:
        response = await _get_http_client().request(
            "POST",
            f"{INSTAGRAM_OAUTH_BASE}/oauth/access_token",
            data={
                "client_id": settings.INSTAGRAM_APP_ID,
                "client_secret": settings.INSTAGRAM_APP_SECRET,
                "grant_type": "authorization_code",
                "redirect_uri": redirect_uri,
                "code": code,
            },
        )
        return (response.status_code < 300, response.json())
    except Exception as e:
        logger.error(f"Instagram code exchange failed: {e}")
        return (False, {"error": {"message": str(e)}})


async def exchange_instagram_long_lived(short_lived_token: str) -> tuple[bool, dict]:
    """Trade a short-lived Instagram token (~1 hour) for a 60-day one.

    Without this the connection dies about an hour after onboarding, with no
    signal beyond messages quietly failing.
    """
    try:
        response = await _get_http_client().request(
            "GET",
            f"{GRAPH_INSTAGRAM_BASE}/access_token",
            params={
                "grant_type": "ig_exchange_token",
                "client_secret": settings.INSTAGRAM_APP_SECRET,
                "access_token": short_lived_token,
            },
        )
        return (response.status_code < 300, response.json())
    except Exception as e:
        logger.error(f"Instagram long-lived exchange failed: {e}")
        return (False, {"error": {"message": str(e)}})


async def subscribe_instagram_app(access_token: str) -> bool:
    """Subscribe our app to this Instagram account's message webhooks.

    Best-effort, like subscribe_app: onboarding still succeeds if it fails, and
    it can be retried by reconnecting.
    """
    ok, _ = await _graph_request("POST", "me/subscribed_apps", access_token,
                                 params={"subscribed_fields": "messages"},
                                 base=GRAPH_INSTAGRAM_BASE)
    return ok


async def register_phone_number(phone_number_id: str, access_token: str, pin: str) -> tuple[bool, dict]:
    """Enable a number for Cloud API sending.

    A number onboarded through Embedded Signup is attached to the WABA but not
    yet usable; registering it with a two-step-verification PIN is what makes it
    able to send.
    """
    return await _graph_request("POST", f"{phone_number_id}/register", access_token,
                                json_body={"messaging_product": "whatsapp", "pin": pin})


# Template fields worth reading back from Graph; components carries the body
# text and any {{n}} variables the UI has to prompt for.
TEMPLATE_FIELDS = "name,status,category,language,components"
TEMPLATE_PAGE_LIMIT = 100
# Backstop against paging forever on a malformed cursor. Well above Meta's own
# per-WABA template ceiling, so a real account never reaches it.
TEMPLATE_MAX_PAGES = 10


async def graph_list_all(path: str, access_token: str, params: dict,
                         max_pages: int = TEMPLATE_MAX_PAGES) -> tuple[bool, list | dict]:
    """Every node on a Graph edge, following its cursor.

    A node that exists but isn't listed is an invisible failure — a Page the
    customer administers but never sees offered, a template an agent cannot pick
    — so this pages rather than stopping at the first page. Returns
    (True, [node dicts]) or (False, graph error body), matching the other
    helpers here. A `limit` in params lets a short final page end the walk one
    round-trip early; without one it stops when Graph offers no `after` cursor.
    """
    items: list = []
    limit = params.get("limit")
    for _ in range(max_pages):
        ok, data = await graph_get(path, access_token, params=params)
        if not ok:
            return False, data
        page = data.get("data")
        if not isinstance(page, list):
            return False, data
        items.extend(page)

        after = (data.get("paging") or {}).get("cursors", {}).get("after")
        if not after or (limit is not None and len(page) < limit):
            return True, items
        params = {**params, "after": after}

    logger.warning(f"Stopped paging {path} at {max_pages} pages")
    return True, items


async def fetch_message_templates(waba_id: str, access_token: str) -> tuple[bool, list | dict]:
    """Every template on the WABA (see graph_list_all for the paging rationale)."""
    return await graph_list_all(
        f"{waba_id}/message_templates", access_token,
        {"fields": TEMPLATE_FIELDS, "limit": TEMPLATE_PAGE_LIMIT})


async def subscribe_app(node_id: str, access_token: str, subscribed_fields: Optional[str] = None) -> bool:
    """Subscribe our app to a Page/WABA so its messages reach our webhook.
    Best-effort: onboarding still succeeds if this fails (can be retried)."""
    payload = {}
    if subscribed_fields:
        payload["subscribed_fields"] = subscribed_fields
    result = await graph_post(f"{node_id}/subscribed_apps", access_token, payload)
    return result.ok


def _extract_message_id(data: dict) -> Optional[str]:
    # WhatsApp: {"messages":[{"id":...}]}; Messenger/IG: {"message_id":...}
    if isinstance(data.get("messages"), list) and data["messages"]:
        return str(data["messages"][0].get("id", "")) or None
    return data.get("message_id")


def graph_auth_refused(body) -> bool:
    """Whether a Graph error body says the access token itself is unusable
    (expired, revoked, or never valid), as opposed to refusing this one call."""
    error = body.get("error") if isinstance(body, dict) else None
    return isinstance(error, dict) and error.get("code") in GRAPH_AUTH_ERROR_CODES


def _is_auth_error(response: httpx.Response) -> bool:
    if response.status_code == 401:
        return True
    try:
        return graph_auth_refused(response.json())
    except Exception:
        return False


def _graph_error(response: httpx.Response) -> str:
    try:
        body = response.json()
        return str(body.get("error", {}).get("message") or body)
    except Exception:
        return f"HTTP {response.status_code}"


class MetaBaseAdapter(ChannelAdapter):
    """Shared behavior for the Meta messaging channels (WhatsApp, Messenger,
    Instagram): one webhook signature scheme, one Graph client, one 24h
    customer-service window. Subclasses implement parse_inbound and send_text."""

    # What an expired window means for this channel: WhatsApp can reopen with a
    # template, Messenger/Instagram simply cannot deliver.
    expired_status: ClassVar[WindowStatus] = WindowStatus.UNDELIVERABLE

    async def verify_webhook(self, headers: dict, raw_body: bytes, account: Optional[ChannelAccount]) -> bool:
        return verify_meta_signature(raw_body, headers.get("x-hub-signature-256", ""))

    def check_delivery_window(self, conversation: ChannelConversation) -> WindowStatus:
        last_inbound = conversation.last_inbound_at
        if last_inbound is None:
            return self.expired_status
        if last_inbound.tzinfo is None:
            last_inbound = last_inbound.replace(tzinfo=timezone.utc)
        age_hours = (datetime.now(timezone.utc) - last_inbound).total_seconds() / 3600
        return WindowStatus.OK if age_hours < WINDOW_HOURS else self.expired_status

    @staticmethod
    def access_token(account: ChannelAccount) -> str:
        """Decrypt the account's Graph access token (page or WABA token)."""
        return decrypt_credentials(account.encrypted_credentials)["access_token"]

    @staticmethod
    def _timestamp(value) -> Optional[datetime]:
        try:
            return datetime.fromtimestamp(int(value), tz=timezone.utc)
        except (TypeError, ValueError):
            return None