feat: Mission 31 — Meta Platform: Instagram DM, Messenger, and Lead Ads Full Lifecycle

- OAuth 2.0 connect flow (start + callback with JWT state, token exchange chain)
- Webhook subscription API (auto-subscribe after OAuth, manual trigger endpoint)
- Message normalization: Instagram DM + Messenger channel discrimination, attachment parsing
- Meta delivery/read status webhooks via watermark-based updates
- Lead Ads ingestion: fetch lead data from Graph API, enrich contacts
- MetaAdapter: send_media implementation + get_user_profile
- Dispatch service: Meta media dispatch (mirrors WhatsApp pattern)
- Channel-level observability metrics (instagram/messenger discrimination)
- 21 new tests (288 total), all passing + 15/15 E2E checks

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

backend/alembic/versions/k1l2m3n4o5p6_mission_31_meta_platform.py

@@ -0,0 +1,21 @@
+"""Mission 31 — Meta Platform: Instagram DM, Messenger, Lead Ads
+
+Revision ID: k1l2m3n4o5p6
+Revises: j0k1l2m3n4o5
+Create Date: 2026-03-01
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "k1l2m3n4o5p6"
+down_revision = "j0k1l2m3n4o5"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("channelidentity", sa.Column("channel", sa.String, nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("channelidentity", "channel")

backend/app/api/v1/integrations.py

@@ -185,3 +185,42 @@ async def integrations_health_check(
     )
     await db.commit()
     return wrap_data({"status": "ok", "checked_count": len(integrations), "errors_found": errors_found})
+
+@router.post("/meta/webhooks/subscribe", response_model=ResponseEnvelope[dict], dependencies=[Depends(require_module_enabled(MODULE_INTEGRATIONS_CONNECT, "write")), Depends(require_entitlement("integrations_connect"))])
+async def subscribe_meta_webhooks(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    workspace: Workspace = Depends(deps.get_active_workspace),
+    current_user: User = Depends(deps.get_current_user),
+) -> Any:
+    """Manually trigger webhook subscription for a connected Meta integration."""
+    result = await db.execute(
+        select(Integration).where(
+            Integration.workspace_id == workspace.id,
+            Integration.provider == "meta",
+            Integration.status == IntegrationStatus.CONNECTED,
+        )
+    )
+    integration = result.scalars().first()
+    if not integration or not integration.encrypted_config:
+        return wrap_error("No connected Meta integration found")
+
+    config = decrypt_data(integration.encrypted_config)
+    page_id = integration.provider_workspace_id
+    page_access_token = config.get("access_token")
+
+    from app.services.meta_service import subscribe_page_webhooks
+    success = await subscribe_page_webhooks(page_id, page_access_token)
+
+    await audit_event(
+        db, action="meta_webhook_subscribe", entity_type="integration",
+        entity_id=str(integration.id), actor_user_id=current_user.id,
+        outcome="success" if success else "failure",
+        workspace_id=workspace.id, request=request,
+        metadata={"provider": "meta", "page_id": page_id},
+    )
+    await db.commit()
+
+    if success:
+        return wrap_data({"message": "Webhook subscription successful", "page_id": page_id})
+    return wrap_error("Failed to subscribe to webhooks")

backend/app/api/v1/meta_oauth.py

@@ -0,0 +1,260 @@
+"""Meta OAuth 2.0 connect flow — start and callback endpoints."""
+import logging
+from datetime import datetime, timedelta
+from urllib.parse import urlencode
+
+import httpx
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import RedirectResponse
+from jose import jwt, JWTError
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.exc import IntegrityError
+from sqlmodel import select
+
+from app.api import deps
+from app.core.config import settings
+from app.core.db import get_db
+from app.core.modules import require_module_enabled, MODULE_INTEGRATIONS_CONNECT
+from app.core.security import encrypt_data
+from app.models.models import Integration, IntegrationStatus, User, Workspace
+from app.schemas.envelope import wrap_data, wrap_error
+from app.services.audit_service import audit_event
+from app.services.entitlements import require_entitlement
+from app.services.meta_service import subscribe_page_webhooks
+from app.services.metrics_service import metrics
+from app.services.runtime_event_service import log_event
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+GRAPH_API_VERSION = "v18.0"
+GRAPH_BASE = f"https://graph.facebook.com/{GRAPH_API_VERSION}"
+META_SCOPES = "pages_messaging,pages_manage_metadata,instagram_manage_messages,leads_retrieval,pages_read_engagement"
+
+
+@router.get(
+    "/start",
+    dependencies=[
+        Depends(require_module_enabled(MODULE_INTEGRATIONS_CONNECT, "write")),
+        Depends(require_entitlement("integrations_connect")),
+    ],
+)
+async def meta_oauth_start(
+    db: AsyncSession = Depends(get_db),
+    workspace: Workspace = Depends(deps.get_active_workspace),
+    current_user: User = Depends(deps.get_current_user),
+):
+    """Build Meta authorization URL and return it."""
+    if not settings.META_CLIENT_ID or not settings.META_CLIENT_SECRET:
+        return wrap_error("META_CLIENT_ID and META_CLIENT_SECRET must be configured")
+
+    redirect_uri = settings.META_OAUTH_REDIRECT_URI
+    if not redirect_uri:
+        base = settings.APP_BASE_URL or settings.FRONTEND_URL
+        redirect_uri = f"{base}/api/v1/integrations/meta/oauth/callback"
+
+    # Encode workspace + user into JWT state param (10-min expiry)
+    state_payload = {
+        "sub": str(current_user.id),
+        "workspace_id": str(workspace.id),
+        "exp": datetime.utcnow() + timedelta(minutes=10),
+        "purpose": "meta_oauth",
+    }
+    state_token = jwt.encode(
+        state_payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM
+    )
+
+    params = {
+        "client_id": settings.META_CLIENT_ID,
+        "redirect_uri": redirect_uri,
+        "scope": META_SCOPES,
+        "state": state_token,
+        "response_type": "code",
+    }
+    auth_url = f"https://www.facebook.com/{GRAPH_API_VERSION}/dialog/oauth?{urlencode(params)}"
+
+    metrics.increment("meta_oauth_started")
+    await log_event(
+        db, event_type="meta.oauth.started", source="meta_oauth",
+        workspace_id=workspace.id,
+    )
+    await db.commit()
+
+    return wrap_data({"auth_url": auth_url})
+
+
+@router.get("/callback")
+async def meta_oauth_callback(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+):
+    """Exchange authorization code for token, subscribe webhooks, store integration."""
+    code = request.query_params.get("code")
+    state = request.query_params.get("state")
+    error = request.query_params.get("error")
+    frontend_base = settings.FRONTEND_URL
+
+    if error or not code or not state:
+        metrics.increment("meta_oauth_failed")
+        return RedirectResponse(
+            f"{frontend_base}/settings/integrations?oauth=meta&status=error&reason={error or 'missing_params'}"
+        )
+
+    # Decode state JWT
+    try:
+        state_data = jwt.decode(
+            state, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM]
+        )
+        if state_data.get("purpose") != "meta_oauth":
+            raise JWTError("Invalid state purpose")
+        user_id = state_data["sub"]
+        workspace_id = state_data["workspace_id"]
+    except JWTError:
+        metrics.increment("meta_oauth_failed")
+        return RedirectResponse(
+            f"{frontend_base}/settings/integrations?oauth=meta&status=error&reason=invalid_state"
+        )
+
+    redirect_uri = settings.META_OAUTH_REDIRECT_URI
+    if not redirect_uri:
+        base = settings.APP_BASE_URL or settings.FRONTEND_URL
+        redirect_uri = f"{base}/api/v1/integrations/meta/oauth/callback"
+
+    try:
+        async with httpx.AsyncClient() as client:
+            # 1. Exchange code for short-lived user token
+            token_resp = await client.get(
+                f"{GRAPH_BASE}/oauth/access_token",
+                params={
+                    "client_id": settings.META_CLIENT_ID,
+                    "redirect_uri": redirect_uri,
+                    "client_secret": settings.META_CLIENT_SECRET,
+                    "code": code,
+                },
+            )
+            if token_resp.status_code != 200:
+                raise Exception(f"Token exchange failed: {token_resp.text}")
+            short_token = token_resp.json().get("access_token")
+
+            # 2. Exchange for long-lived token
+            ll_resp = await client.get(
+                f"{GRAPH_BASE}/oauth/access_token",
+                params={
+                    "grant_type": "fb_exchange_token",
+                    "client_id": settings.META_CLIENT_ID,
+                    "client_secret": settings.META_CLIENT_SECRET,
+                    "fb_exchange_token": short_token,
+                },
+            )
+            if ll_resp.status_code != 200:
+                raise Exception(f"Long-lived token exchange failed: {ll_resp.text}")
+            long_lived_user_token = ll_resp.json().get("access_token")
+
+            # 3. Fetch pages
+            pages_resp = await client.get(
+                f"{GRAPH_BASE}/me/accounts",
+                params={"access_token": long_lived_user_token},
+            )
+            if pages_resp.status_code != 200:
+                raise Exception(f"Page fetch failed: {pages_resp.text}")
+            pages = pages_resp.json().get("data", [])
+            if not pages:
+                raise Exception("No pages found for this user")
+
+            page = pages[0]
+            page_id = page["id"]
+            page_access_token = page["access_token"]
+            page_name = page.get("name", "")
+
+            # 4. Fetch Instagram Business Account (optional)
+            ig_account_id = None
+            try:
+                ig_resp = await client.get(
+                    f"{GRAPH_BASE}/{page_id}",
+                    params={
+                        "fields": "instagram_business_account",
+                        "access_token": page_access_token,
+                    },
+                )
+                if ig_resp.status_code == 200:
+                    ig_data = ig_resp.json().get("instagram_business_account", {})
+                    ig_account_id = ig_data.get("id")
+            except Exception:
+                logger.warning("Could not fetch Instagram Business Account ID")
+
+            # 5. Subscribe webhooks
+            await subscribe_page_webhooks(page_id, page_access_token)
+
+        # 6. Store Integration
+        config_data = {
+            "access_token": page_access_token,
+            "page_name": page_name,
+            "instagram_business_account_id": ig_account_id,
+        }
+        encrypted_config = encrypt_data(config_data)
+
+        from uuid import UUID
+        ws_id = UUID(workspace_id)
+
+        result = await db.execute(
+            select(Integration).where(
+                Integration.workspace_id == ws_id,
+                Integration.provider == "meta",
+            )
+        )
+        integration = result.scalars().first()
+
+        if not integration:
+            integration = Integration(workspace_id=ws_id, provider="meta")
+            db.add(integration)
+
+        integration.status = IntegrationStatus.CONNECTED
+        integration.encrypted_config = encrypted_config
+        integration.provider_workspace_id = page_id
+        integration.connected_at = datetime.utcnow()
+        integration.last_checked_at = datetime.utcnow()
+        integration.last_error = None
+
+        try:
+            await db.commit()
+        except IntegrityError:
+            await db.rollback()
+            metrics.increment("meta_oauth_failed")
+            return RedirectResponse(
+                f"{frontend_base}/settings/integrations?oauth=meta&status=error&reason=page_already_linked"
+            )
+
+        await audit_event(
+            db, action="integration_connect", entity_type="integration",
+            entity_id=str(integration.id), actor_user_id=UUID(user_id),
+            outcome="success", workspace_id=ws_id, request=request,
+            metadata={"provider": "meta", "method": "oauth", "page_name": page_name},
+        )
+        await db.commit()
+
+        metrics.increment("meta_oauth_completed")
+        await log_event(
+            db, event_type="meta.oauth.completed", source="meta_oauth",
+            workspace_id=ws_id,
+            payload={"page_id": page_id, "page_name": page_name},
+        )
+        await db.commit()
+
+        return RedirectResponse(
+            f"{frontend_base}/settings/integrations?oauth=meta&status=success"
+        )
+
+    except Exception as e:
+        logger.exception(f"Meta OAuth callback error: {e}")
+        metrics.increment("meta_oauth_failed")
+        try:
+            await log_event(
+                db, event_type="meta.oauth.failed", source="meta_oauth",
+                outcome="failure", error_message=str(e),
+            )
+            await db.commit()
+        except Exception:
+            pass
+        return RedirectResponse(
+            f"{frontend_base}/settings/integrations?oauth=meta&status=error&reason=callback_error"
+        )

backend/app/api/v1/webhooks.py

@@ -224,9 +224,12 @@ async def meta_webhook(
     except (IndexError, AttributeError):
         pass
 
-    metrics.increment("webhooks_received", labels={"provider": "meta"})
+    # Channel discrimination for metrics
+    object_type = payload.get("object", "")
+    channel = "instagram" if object_type == "instagram" else "messenger"
+    metrics.increment("webhooks_received", labels={"provider": "meta", "channel": channel})
     await log_event(db, event_type="webhook.received", source="webhook",
-                    payload={"provider": "meta", "page_id": page_id})
+                    payload={"provider": "meta", "page_id": page_id, "channel": channel})
 
     # 3. Resolve Workspace
     workspace_id = await resolve_workspace(db, "meta", page_id) if page_id else None

backend/app/core/config.py

@@ -36,6 +36,7 @@ class Settings(BaseSettings):
     META_CLIENT_ID: Optional[str] = None
     META_CLIENT_SECRET: Optional[str] = None
     META_APP_SECRET: Optional[str] = None
+    META_OAUTH_REDIRECT_URI: Optional[str] = None
     WHATSAPP_VERIFY_TOKEN: Optional[str] = None
     ZOHO_CLIENT_ID: Optional[str] = None
     ZOHO_CLIENT_SECRET: Optional[str] = None

backend/app/integrations/meta/adapter.py

@@ -64,5 +64,61 @@ class MetaAdapter(MessagingAdapter):
     async def send_media(
         self, to: str, media_type: str, media_url: str, caption: Optional[str] = None
     ) -> str:
-        """Media sending not yet implemented for Meta."""
-        raise NotImplementedError("Meta media sending not yet supported")
+        """Send a media message via Meta Graph API (Messenger/Instagram)."""
+        headers = {"Content-Type": "application/json"}
+        params = {"access_token": self.page_access_token}
+        payload = {
+            "recipient": {"id": to},
+            "message": {
+                "attachment": {
+                    "type": media_type,
+                    "payload": {"url": media_url, "is_reusable": True},
+                }
+            },
+            "messaging_type": "RESPONSE",
+        }
+
+        async with httpx.AsyncClient() as client:
+            try:
+                response = await client.post(
+                    self.base_url, headers=headers, params=params, json=payload
+                )
+                data = response.json()
+
+                if response.status_code == 200:
+                    provider_message_id = data.get("message_id")
+                    logger.info(f"Meta media sent: {provider_message_id}")
+                    return provider_message_id
+
+                error = data.get("error", {})
+                error_code = error.get("code")
+                error_message = error.get("message", "Unknown error")
+                logger.error(f"Meta API media error: {error_message} (Code: {error_code})")
+
+                if response.status_code in [400, 401, 403]:
+                    raise Exception(f"PERMANENT_FAILURE: {error_message} (Code: {error_code})")
+                if response.status_code in [429, 500, 502, 503, 504]:
+                    raise Exception(f"TRANSIENT_FAILURE: {error_message} (Code: {error_code})")
+                raise Exception(f"UNCLASSIFIED_FAILURE: {error_message} (Status: {response.status_code})")
+
+            except httpx.RequestError as e:
+                logger.error(f"Meta Network error (media): {str(e)}")
+                raise Exception(f"TRANSIENT_FAILURE: Network error: {str(e)}")
+
+    async def get_user_profile(self, user_id: str) -> dict:
+        """Fetch user name and profile pic from Meta Graph API. Non-critical."""
+        url = f"https://graph.facebook.com/v18.0/{user_id}"
+        params = {
+            "fields": "first_name,last_name,profile_pic",
+            "access_token": self.page_access_token,
+        }
+        async with httpx.AsyncClient() as client:
+            try:
+                response = await client.get(url, params=params)
+                if response.status_code == 200:
+                    return response.json()
+                logger.warning(f"Failed to fetch profile for {user_id}: {response.status_code}")
+                return {}
+            except httpx.RequestError as e:
+                logger.warning(f"Network error fetching profile: {e}")
+                return {}

backend/app/models/models.py

@@ -268,6 +268,7 @@ class ChannelIdentity(BaseIDModel, table=True):
     contact_id: UUID = Field(foreign_key="contact.id")
     provider: str # whatsapp, meta, etc
     provider_user_id: str # phone number or fb id
+    channel: Optional[str] = Field(default=None) # instagram, messenger (sub-channel of meta)
     
 class Conversation(WorkspaceScopedModel, table=True):
     __table_args__ = (

backend/app/services/dispatch_service.py

@@ -227,7 +227,12 @@ class DispatchService:
                     page_id=integration.provider_workspace_id,
                     page_access_token=config.get("access_token")
                 )
-                provider_message_id = await adapter.send_text(recipient_id, msg.content)
+                if media_url and media_type:
+                    provider_message_id = await adapter.send_media(
+                        recipient_id, media_type, media_url, caption=media_caption
+                    )
+                else:
+                    provider_message_id = await adapter.send_text(recipient_id, msg.content)
             else:
                 raise Exception(f"Unsupported platform: {msg.platform}")
 

backend/app/services/meta_service.py

@@ -0,0 +1,111 @@
+"""Meta Graph API service — centralized helpers for OAuth, webhook subscription, Lead Ads."""
+import httpx
+import logging
+
+from app.services.metrics_service import metrics
+
+logger = logging.getLogger(__name__)
+
+GRAPH_API_VERSION = "v18.0"
+GRAPH_BASE_URL = f"https://graph.facebook.com/{GRAPH_API_VERSION}"
+
+
+async def subscribe_page_webhooks(page_id: str, page_access_token: str) -> bool:
+    """Subscribe a page to webhook fields for messaging + lead ads."""
+    url = f"{GRAPH_BASE_URL}/{page_id}/subscribed_apps"
+    params = {"access_token": page_access_token}
+    data = {"subscribed_fields": "messages,messaging_postbacks,feed,leadgen"}
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.post(url, params=params, data=data)
+            if response.status_code == 200 and response.json().get("success"):
+                metrics.increment("meta_webhook_subscribed")
+                logger.info(f"Subscribed page {page_id} to webhooks")
+                return True
+            logger.error(f"Webhook subscription failed for {page_id}: {response.text}")
+            return False
+        except httpx.RequestError as e:
+            logger.error(f"Network error subscribing webhooks for {page_id}: {e}")
+            return False
+
+
+async def verify_page_subscription(page_id: str, page_access_token: str) -> dict:
+    """Check current webhook subscriptions for a page."""
+    url = f"{GRAPH_BASE_URL}/{page_id}/subscribed_apps"
+    params = {"access_token": page_access_token}
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.get(url, params=params)
+            if response.status_code == 200:
+                return response.json()
+            return {"error": response.text}
+        except httpx.RequestError as e:
+            return {"error": str(e)}
+
+
+async def fetch_lead_data(leadgen_id: str, page_access_token: str) -> dict:
+    """Fetch lead submission data from Meta Graph API."""
+    url = f"{GRAPH_BASE_URL}/{leadgen_id}"
+    params = {"access_token": page_access_token}
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.get(url, params=params)
+            if response.status_code == 200:
+                return response.json()
+            error_msg = response.text
+            logger.error(f"Failed to fetch lead {leadgen_id}: {error_msg}")
+            metrics.increment("meta_leads_failed")
+            raise Exception(f"Failed to fetch lead data: {error_msg}")
+        except httpx.RequestError as e:
+            logger.error(f"Network error fetching lead {leadgen_id}: {e}")
+            metrics.increment("meta_leads_failed")
+            raise Exception(f"Network error fetching lead: {e}")
+
+
+def parse_lead_fields(field_data: list) -> dict:
+    """Normalize Meta lead field_data array into a flat dict."""
+    result = {}
+    field_map = {
+        "email": "email",
+        "phone_number": "phone",
+        "full_name": "full_name",
+        "first_name": "first_name",
+        "last_name": "last_name",
+    }
+
+    for field in field_data:
+        name = field.get("name", "").lower()
+        values = field.get("values", [])
+        if name in field_map and values:
+            result[field_map[name]] = values[0]
+
+    # Split full_name into first/last if separate fields missing
+    if "full_name" in result and "first_name" not in result:
+        parts = result["full_name"].split(" ", 1)
+        result["first_name"] = parts[0]
+        result["last_name"] = parts[1] if len(parts) > 1 else ""
+
+    return result
+
+
+async def get_user_profile(user_id: str, page_access_token: str) -> dict:
+    """Fetch user profile from Meta Graph API. Non-critical — returns {} on failure."""
+    url = f"{GRAPH_BASE_URL}/{user_id}"
+    params = {
+        "fields": "first_name,last_name,profile_pic",
+        "access_token": page_access_token,
+    }
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.get(url, params=params)
+            if response.status_code == 200:
+                return response.json()
+            logger.warning(f"Failed to fetch profile for {user_id}: {response.status_code}")
+            return {}
+        except httpx.RequestError as e:
+            logger.warning(f"Network error fetching profile for {user_id}: {e}")
+            return {}

backend/app/workers/tasks.py

@@ -42,6 +42,11 @@ def parse_webhook_payload(provider: str, payload: Dict) -> Dict:
         "media_type": None,
         "mime_type": None,
         "caption": None,
+        "channel": None,            # instagram | messenger | lead_ads
+        "attachment_url": None,
+        "leadgen_id": None,
+        "meta_delivery_watermark": None,
+        "meta_read_watermark": None,
     }
 
     try:
@@ -77,18 +82,50 @@ def parse_webhook_payload(provider: str, payload: Dict) -> Dict:
 
         elif provider == "meta":
             entry = payload.get("entry", [])[0]
+
+            # Channel discrimination: instagram vs messenger
+            object_type = payload.get("object", "")
+            if object_type == "instagram":
+                normalized["channel"] = "instagram"
+            else:
+                normalized["channel"] = "messenger"
+
             # Messaging (Messenger/Instagram)
             if "messaging" in entry:
                 msg_event = entry["messaging"][0]
-                normalized["trigger_type"] = "MESSAGE_INBOUND"
-                normalized["provider_user_id"] = msg_event.get("sender", {}).get("id")
-                normalized["content"] = msg_event.get("message", {}).get("text")
+
+                # Delivery receipts (watermark-based)
+                if "delivery" in msg_event:
+                    normalized["trigger_type"] = "META_STATUS_UPDATE"
+                    normalized["meta_delivery_watermark"] = msg_event["delivery"].get("watermark")
+                elif "read" in msg_event:
+                    normalized["trigger_type"] = "META_STATUS_UPDATE"
+                    normalized["meta_read_watermark"] = msg_event["read"].get("watermark")
+                elif msg_event.get("message"):
+                    # Standard inbound message
+                    normalized["trigger_type"] = "MESSAGE_INBOUND"
+                    normalized["provider_user_id"] = msg_event.get("sender", {}).get("id")
+                    message = msg_event.get("message", {})
+                    normalized["content"] = message.get("text")
+
+                    # Attachment parsing
+                    attachments = message.get("attachments", [])
+                    if attachments:
+                        att = attachments[0]
+                        att_type = att.get("type")
+                        normalized["media_type"] = att_type
+                        normalized["attachment_url"] = att.get("payload", {}).get("url")
+                        if not normalized["content"]:
+                            normalized["content"] = f"[{att_type}]"
+
             # Lead Ads
             elif "changes" in entry:
                 change = entry["changes"][0]
                 if change.get("field") == "leadgen":
                     normalized["trigger_type"] = "LEAD_AD_SUBMIT"
-                    normalized["provider_user_id"] = change.get("value", {}).get("leadgen_id")
+                    normalized["leadgen_id"] = change.get("value", {}).get("leadgen_id")
+                    normalized["provider_user_id"] = normalized["leadgen_id"]
+                    normalized["channel"] = "lead_ads"
                     normalized["content"] = "Lead Ad Submission"
     except (IndexError, KeyError, AttributeError):
         pass
@@ -163,6 +200,64 @@ async def _process_status_updates(session, statuses: list, workspace_id, correla
                         related_ids={"message_id": str(msg.id)},
                         payload={"status": wa_status, "provider_message_id": provider_msg_id})
 
+
+async def _process_meta_status_updates(session, info: dict, workspace_id, correlation_id: str):
+    """Process Meta delivery/read status webhooks using watermark timestamps."""
+    from app.services.runtime_event_service import log_event
+    from sqlmodel import or_
+
+    delivery_watermark = info.get("meta_delivery_watermark")
+    read_watermark = info.get("meta_read_watermark")
+
+    if delivery_watermark:
+        watermark_ts = datetime.utcfromtimestamp(int(delivery_watermark) / 1000)
+        result = await session.execute(
+            select(Message).where(
+                Message.workspace_id == workspace_id,
+                Message.platform == "meta",
+                Message.direction == "outbound",
+                Message.delivery_status == DeliveryStatus.SENT,
+                Message.sent_at < watermark_ts,
+            )
+        )
+        for msg in result.scalars().all():
+            old_priority = _STATUS_PRIORITY.get(msg.delivery_status, 0)
+            new_priority = _STATUS_PRIORITY.get(DeliveryStatus.DELIVERED, 0)
+            if new_priority > old_priority:
+                msg.delivery_status = DeliveryStatus.DELIVERED
+                msg.delivered_at = watermark_ts
+                session.add(msg)
+
+    if read_watermark:
+        watermark_ts = datetime.utcfromtimestamp(int(read_watermark) / 1000)
+        result = await session.execute(
+            select(Message).where(
+                Message.workspace_id == workspace_id,
+                Message.platform == "meta",
+                Message.direction == "outbound",
+                or_(
+                    Message.delivery_status == DeliveryStatus.SENT,
+                    Message.delivery_status == DeliveryStatus.DELIVERED,
+                ),
+                Message.sent_at < watermark_ts,
+            )
+        )
+        for msg in result.scalars().all():
+            old_priority = _STATUS_PRIORITY.get(msg.delivery_status, 0)
+            new_priority = _STATUS_PRIORITY.get(DeliveryStatus.READ, 0)
+            if new_priority > old_priority:
+                msg.delivery_status = DeliveryStatus.READ
+                msg.read_at = watermark_ts
+                if not msg.delivered_at:
+                    msg.delivered_at = watermark_ts
+                session.add(msg)
+
+    await log_event(session, event_type="message.meta_status_updated", source="webhook",
+                    workspace_id=workspace_id, correlation_id=correlation_id,
+                    payload={"delivery_watermark": delivery_watermark,
+                             "read_watermark": read_watermark})
+
+
 @celery_app.task(name="app.workers.tasks.process_webhook_event")
 def process_webhook_event(event_id: str):
     """
@@ -218,6 +313,18 @@ def process_webhook_event(event_id: str):
                     await session.commit()
                     return
 
+                # 1a2. Handle Meta delivery/read status updates (watermark-based)
+                if info.get("trigger_type") == "META_STATUS_UPDATE":
+                    await _process_meta_status_updates(
+                        session, info, event.workspace_id,
+                        str(event.correlation_id)
+                    )
+                    event.status = WebhookStatus.PROCESSED
+                    event.processed_at = datetime.utcnow()
+                    session.add(event)
+                    await session.commit()
+                    return
+
                 if not info["trigger_type"]:
                     logger.info(f"No actionable trigger found in payload for event {event_id}")
                     event.status = WebhookStatus.PROCESSED
@@ -240,13 +347,16 @@ def process_webhook_event(event_id: str):
 
                 if info["trigger_type"] == "MESSAGE_INBOUND":
                     msg_metadata = {}
+                    if info.get("channel"):
+                        msg_metadata["channel"] = info["channel"]
+                    if info.get("attachment_url"):
+                        msg_metadata["attachment_url"] = info["attachment_url"]
+                        msg_metadata["media_type"] = info.get("media_type")
                     if info.get("media_id"):
-                        msg_metadata = {
-                            "media_id": info["media_id"],
-                            "media_type": info["media_type"],
-                            "mime_type": info["mime_type"],
-                        }
-                        # Download and store media
+                        msg_metadata["media_id"] = info["media_id"]
+                        msg_metadata["media_type"] = info["media_type"]
+                        msg_metadata["mime_type"] = info["mime_type"]
+                        # Download and store media (WhatsApp)
                         try:
                             from app.services.media_service import download_and_store_media
                             media_path = await download_and_store_media(
@@ -269,6 +379,73 @@ def process_webhook_event(event_id: str):
                     )
                     session.add(inbound_msg)
 
+                # 2b. Lead Ads: fetch lead data and enrich contact
+                if info["trigger_type"] == "LEAD_AD_SUBMIT" and info.get("leadgen_id"):
+                    from app.services.meta_service import fetch_lead_data, parse_lead_fields
+                    from app.models.models import Integration
+                    from app.core.security import decrypt_data
+                    from app.services.metrics_service import metrics
+
+                    integration_result = await session.execute(
+                        select(Integration).where(
+                            Integration.workspace_id == event.workspace_id,
+                            Integration.provider == "meta"
+                        )
+                    )
+                    integration = integration_result.scalars().first()
+                    if integration and integration.encrypted_config:
+                        try:
+                            config = decrypt_data(integration.encrypted_config)
+                            lead_raw = await fetch_lead_data(
+                                info["leadgen_id"], config.get("access_token")
+                            )
+                            lead_fields = parse_lead_fields(
+                                lead_raw.get("field_data", [])
+                            )
+
+                            # Enrich contact metadata
+                            updated_meta = dict(contact.additional_metadata or {})
+                            if lead_fields.get("email"):
+                                updated_meta["email"] = lead_fields["email"]
+                            if lead_fields.get("phone"):
+                                updated_meta["phone"] = lead_fields["phone"]
+                            updated_meta["lead_source"] = "meta_lead_ad"
+                            updated_meta["leadgen_id"] = info["leadgen_id"]
+                            contact.additional_metadata = updated_meta
+
+                            if lead_fields.get("first_name") and not contact.first_name:
+                                contact.first_name = lead_fields["first_name"]
+                            if lead_fields.get("last_name") and not contact.last_name:
+                                contact.last_name = lead_fields["last_name"]
+                            session.add(contact)
+
+                            # Store lead as message
+                            lead_msg = Message(
+                                workspace_id=event.workspace_id,
+                                conversation_id=conversation.id,
+                                direction="inbound",
+                                content=f"Lead Ad: {lead_fields.get('full_name', lead_fields.get('first_name', 'Unknown'))}",
+                                platform="meta",
+                                delivery_status="delivered",
+                                additional_metadata={
+                                    "type": "lead_ad_submission",
+                                    "leadgen_id": info["leadgen_id"],
+                                    "lead_fields": lead_fields,
+                                    "channel": "lead_ads",
+                                },
+                            )
+                            session.add(lead_msg)
+                            metrics.increment("meta_leads_ingested")
+                            await log_event(
+                                session, event_type="meta.lead_ingested",
+                                source="webhook",
+                                workspace_id=event.workspace_id,
+                                correlation_id=str(event.correlation_id),
+                                payload={"leadgen_id": info["leadgen_id"]},
+                            )
+                        except Exception as lead_err:
+                            logger.warning(f"Lead data fetch failed for {info['leadgen_id']}: {lead_err}")
+
                 # 3. Find Matching Published Flow (prefer published_version_id pointer)
                 flow_version = None
                 flow_query = select(Flow).where(

backend/main.py

@@ -21,6 +21,7 @@ from app.api.v1.qualification import router as qualification_router
 from app.api.v1.entitlements import router as entitlements_router
 from app.api.v1.settings_profile import router as settings_profile_router
 from app.api.v1.qualification_criteria import router as qual_criteria_router
+from app.api.v1.meta_oauth import router as meta_oauth_router
 from fastapi import HTTPException
 import uuid
 import logging
@@ -167,6 +168,7 @@ app.include_router(qualification_router, prefix=f"{settings.API_V1_STR}/qualific
 app.include_router(entitlements_router, prefix=f"{settings.API_V1_STR}/entitlements", tags=["entitlements"])
 app.include_router(settings_profile_router, prefix=f"{settings.API_V1_STR}/settings", tags=["settings-profile"])
 app.include_router(qual_criteria_router, prefix=f"{settings.API_V1_STR}/prompt-studio", tags=["prompt-studio"])
+app.include_router(meta_oauth_router, prefix=f"{settings.API_V1_STR}/integrations/meta/oauth", tags=["integrations"])
 
 @app.get("/")
 async def root():

backend/tests/test_meta_platform.py

@@ -0,0 +1,644 @@
+"""
+Mission 31 — Meta Platform: Instagram DM, Messenger, and Lead Ads Tests.
+Covers: OAuth flow, webhook subscription, message parsing (Instagram/Messenger),
+delivery status watermarks, Lead Ads ingestion, MetaAdapter, and observability.
+"""
+import json
+import time
+import pytest
+import pytest_asyncio
+from datetime import datetime, timedelta
+from uuid import uuid4
+from unittest.mock import patch, AsyncMock, MagicMock
+
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.models import (
+    User, Workspace, WorkspaceMember, WorkspaceRole,
+    Message, DeliveryStatus, Integration, IntegrationStatus,
+    Contact, ChannelIdentity, Conversation,
+)
+from app.core.security import get_password_hash, create_access_token, encrypt_data
+from app.workers.tasks import parse_webhook_payload, _process_meta_status_updates
+from app.services.meta_service import parse_lead_fields, subscribe_page_webhooks
+from app.services.metrics_service import MetricsService
+
+
+# ── Helpers ──────────────────────────────────────────────────────────────
+
+
+async def _setup_workspace(db: AsyncSession, provider="meta"):
+    """Create a user + workspace + integration for testing."""
+    user = User(
+        email=f"meta_test_{uuid4().hex[:6]}@test.com",
+        hashed_password=get_password_hash("testpass123"),
+        full_name="Meta Test User",
+        is_active=True,
+        is_superuser=False,
+        email_verified_at=datetime.utcnow(),
+    )
+    db.add(user)
+    await db.flush()
+
+    ws = Workspace(name="Meta Test Workspace")
+    db.add(ws)
+    await db.flush()
+
+    mem = WorkspaceMember(user_id=user.id, workspace_id=ws.id, role=WorkspaceRole.OWNER)
+    db.add(mem)
+
+    integration = Integration(
+        workspace_id=ws.id,
+        provider=provider,
+        status=IntegrationStatus.CONNECTED,
+        provider_workspace_id="PAGE_123",
+        encrypted_config=encrypt_data({"access_token": "mock_page_token"}),
+    )
+    db.add(integration)
+    await db.flush()
+
+    return user, ws, integration
+
+
+def _messenger_text_payload(page_id="PAGE_123", sender_id="PSID_456", text="Hello from Messenger"):
+    """Build a Messenger webhook payload."""
+    return {
+        "object": "page",
+        "entry": [{
+            "id": page_id,
+            "time": int(datetime.utcnow().timestamp() * 1000),
+            "messaging": [{
+                "sender": {"id": sender_id},
+                "recipient": {"id": page_id},
+                "timestamp": int(datetime.utcnow().timestamp() * 1000),
+                "message": {
+                    "mid": f"mid.{uuid4().hex[:12]}",
+                    "text": text,
+                },
+            }],
+        }],
+    }
+
+
+def _instagram_dm_payload(ig_id="IG_PAGE_123", sender_id="IGSID_789", text="Hello from Instagram"):
+    """Build an Instagram DM webhook payload."""
+    return {
+        "object": "instagram",
+        "entry": [{
+            "id": ig_id,
+            "time": int(datetime.utcnow().timestamp() * 1000),
+            "messaging": [{
+                "sender": {"id": sender_id},
+                "recipient": {"id": ig_id},
+                "timestamp": int(datetime.utcnow().timestamp() * 1000),
+                "message": {
+                    "mid": f"mid.{uuid4().hex[:12]}",
+                    "text": text,
+                },
+            }],
+        }],
+    }
+
+
+def _instagram_attachment_payload(ig_id="IG_PAGE_123", sender_id="IGSID_789"):
+    """Build an Instagram DM payload with an image attachment."""
+    return {
+        "object": "instagram",
+        "entry": [{
+            "id": ig_id,
+            "time": int(datetime.utcnow().timestamp() * 1000),
+            "messaging": [{
+                "sender": {"id": sender_id},
+                "recipient": {"id": ig_id},
+                "timestamp": int(datetime.utcnow().timestamp() * 1000),
+                "message": {
+                    "mid": f"mid.{uuid4().hex[:12]}",
+                    "attachments": [{
+                        "type": "image",
+                        "payload": {"url": "https://example.com/photo.jpg"},
+                    }],
+                },
+            }],
+        }],
+    }
+
+
+def _messenger_echo_payload(page_id="PAGE_123"):
+    """Build a Messenger echo payload (sent by the page itself)."""
+    return {
+        "object": "page",
+        "entry": [{
+            "id": page_id,
+            "time": int(datetime.utcnow().timestamp() * 1000),
+            "messaging": [{
+                "sender": {"id": page_id},
+                "recipient": {"id": "PSID_456"},
+                "timestamp": int(datetime.utcnow().timestamp() * 1000),
+                "message": {
+                    "mid": f"mid.{uuid4().hex[:12]}",
+                    "text": "Echo message",
+                    "is_echo": True,
+                },
+            }],
+        }],
+    }
+
+
+def _delivery_receipt_payload(page_id="PAGE_123", watermark=None):
+    """Build a Meta delivery receipt payload."""
+    if watermark is None:
+        watermark = int(datetime.utcnow().timestamp() * 1000)
+    return {
+        "object": "page",
+        "entry": [{
+            "id": page_id,
+            "time": int(datetime.utcnow().timestamp() * 1000),
+            "messaging": [{
+                "sender": {"id": page_id},
+                "recipient": {"id": "PSID_456"},
+                "timestamp": int(datetime.utcnow().timestamp() * 1000),
+                "delivery": {
+                    "mids": ["mid.abc123"],
+                    "watermark": watermark,
+                },
+            }],
+        }],
+    }
+
+
+def _read_receipt_payload(page_id="PAGE_123", watermark=None):
+    """Build a Meta read receipt payload."""
+    if watermark is None:
+        watermark = int(datetime.utcnow().timestamp() * 1000)
+    return {
+        "object": "page",
+        "entry": [{
+            "id": page_id,
+            "time": int(datetime.utcnow().timestamp() * 1000),
+            "messaging": [{
+                "sender": {"id": "PSID_456"},
+                "recipient": {"id": page_id},
+                "timestamp": int(datetime.utcnow().timestamp() * 1000),
+                "read": {
+                    "watermark": watermark,
+                },
+            }],
+        }],
+    }
+
+
+def _leadgen_payload(page_id="PAGE_123", leadgen_id="LEAD_999"):
+    """Build a Lead Ads webhook payload."""
+    return {
+        "object": "page",
+        "entry": [{
+            "id": page_id,
+            "time": int(datetime.utcnow().timestamp() * 1000),
+            "changes": [{
+                "field": "leadgen",
+                "value": {
+                    "leadgen_id": leadgen_id,
+                    "page_id": page_id,
+                    "form_id": "FORM_123",
+                    "created_time": int(datetime.utcnow().timestamp()),
+                },
+            }],
+        }],
+    }
+
+
+# ── Section 1: OAuth Flow ────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_meta_oauth_start_returns_auth_url(async_client: AsyncClient, db_session: AsyncSession):
+    """GET /start returns a valid Meta authorization URL."""
+    user, ws, _ = await _setup_workspace(db_session)
+    token = create_access_token(str(user.id), workspace_id=str(ws.id))
+
+    with patch("app.api.v1.meta_oauth.settings") as mock_settings:
+        mock_settings.META_CLIENT_ID = "test_client_id"
+        mock_settings.META_CLIENT_SECRET = "test_secret"
+        mock_settings.META_OAUTH_REDIRECT_URI = "https://example.com/callback"
+        mock_settings.JWT_SECRET = "test_jwt_secret"
+        mock_settings.JWT_ALGORITHM = "HS256"
+        mock_settings.FRONTEND_URL = "http://localhost:3000"
+        mock_settings.APP_BASE_URL = None
+
+        resp = await async_client.get(
+            "/api/v1/integrations/meta/oauth/start",
+            headers={"Authorization": f"Bearer {token}", "X-Workspace-ID": str(ws.id)},
+        )
+
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["success"] is True
+    auth_url = data["data"]["auth_url"]
+    assert "facebook.com" in auth_url
+    assert "client_id=test_client_id" in auth_url
+    assert "response_type=code" in auth_url
+
+
+@pytest.mark.asyncio
+async def test_meta_oauth_start_no_credentials(async_client: AsyncClient, db_session: AsyncSession):
+    """GET /start returns error when META_CLIENT_ID is not configured."""
+    user, ws, _ = await _setup_workspace(db_session)
+    token = create_access_token(str(user.id), workspace_id=str(ws.id))
+
+    with patch("app.api.v1.meta_oauth.settings") as mock_settings:
+        mock_settings.META_CLIENT_ID = None
+        mock_settings.META_CLIENT_SECRET = None
+
+        resp = await async_client.get(
+            "/api/v1/integrations/meta/oauth/start",
+            headers={"Authorization": f"Bearer {token}", "X-Workspace-ID": str(ws.id)},
+        )
+
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["success"] is False
+    assert "configured" in data["error"].lower()
+
+
+@pytest.mark.asyncio
+async def test_meta_oauth_callback_success(async_client: AsyncClient, db_session: AsyncSession):
+    """GET /callback with valid code exchanges tokens and creates Integration."""
+    from jose import jwt as jose_jwt
+    from app.core.config import settings
+
+    user, ws, _ = await _setup_workspace(db_session, provider="whatsapp")  # no meta integration yet
+
+    state = jose_jwt.encode(
+        {"sub": str(user.id), "workspace_id": str(ws.id),
+         "exp": datetime.utcnow() + timedelta(minutes=10), "purpose": "meta_oauth"},
+        settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM,
+    )
+
+    mock_responses = [
+        # 1. Short-lived token exchange
+        MagicMock(status_code=200, json=lambda: {"access_token": "short_token"}),
+        # 2. Long-lived token exchange
+        MagicMock(status_code=200, json=lambda: {"access_token": "long_lived_token"}),
+        # 3. Fetch pages
+        MagicMock(status_code=200, json=lambda: {"data": [{"id": "PAGE_NEW", "access_token": "page_token", "name": "My Page"}]}),
+        # 4. Fetch IG account
+        MagicMock(status_code=200, json=lambda: {"instagram_business_account": {"id": "IG_BIZ_123"}}),
+        # 5. Subscribe webhooks
+        MagicMock(status_code=200, json=lambda: {"success": True}),
+    ]
+
+    call_count = {"i": 0}
+    original_get = None
+    original_post = None
+
+    async def mock_request(self, method, url, **kwargs):
+        idx = call_count["i"]
+        call_count["i"] += 1
+        if idx < len(mock_responses):
+            return mock_responses[idx]
+        return MagicMock(status_code=200, json=lambda: {})
+
+    with patch("httpx.AsyncClient.get", new=lambda self, url, **kw: mock_request(self, "GET", url, **kw)):
+        with patch("httpx.AsyncClient.post", new=lambda self, url, **kw: mock_request(self, "POST", url, **kw)):
+            resp = await async_client.get(
+                f"/api/v1/integrations/meta/oauth/callback?code=test_code&state={state}",
+            )
+
+    # Should redirect to frontend
+    # (httpx follows redirects by default in our test client)
+    assert resp.status_code == 200 or resp.status_code == 307 or resp.status_code == 302
+
+
+# ── Section 2: Webhook Subscription ──────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_subscribe_page_webhooks_success():
+    """subscribe_page_webhooks returns True on 200 + success."""
+    mock_resp = MagicMock(status_code=200, json=lambda: {"success": True})
+
+    with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_resp):
+        result = await subscribe_page_webhooks("PAGE_123", "token_abc")
+
+    assert result is True
+
+
+@pytest.mark.asyncio
+async def test_subscribe_page_webhooks_failure():
+    """subscribe_page_webhooks returns False on error."""
+    mock_resp = MagicMock(status_code=400, text="Bad Request", json=lambda: {"error": "bad"})
+
+    with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_resp):
+        result = await subscribe_page_webhooks("PAGE_123", "token_abc")
+
+    assert result is False
+
+
+# ── Section 3: Message Parsing — Instagram vs Messenger ──────────────────
+
+
+@pytest.mark.asyncio
+async def test_parse_instagram_dm_text():
+    """Instagram DM text payload → channel=instagram, correct content."""
+    payload = _instagram_dm_payload(text="Hi from IG")
+    info = parse_webhook_payload("meta", payload)
+
+    assert info["trigger_type"] == "MESSAGE_INBOUND"
+    assert info["channel"] == "instagram"
+    assert info["provider_user_id"] == "IGSID_789"
+    assert info["content"] == "Hi from IG"
+
+
+@pytest.mark.asyncio
+async def test_parse_messenger_text():
+    """Messenger text payload → channel=messenger, correct content."""
+    payload = _messenger_text_payload(text="Hi from Messenger")
+    info = parse_webhook_payload("meta", payload)
+
+    assert info["trigger_type"] == "MESSAGE_INBOUND"
+    assert info["channel"] == "messenger"
+    assert info["provider_user_id"] == "PSID_456"
+    assert info["content"] == "Hi from Messenger"
+
+
+@pytest.mark.asyncio
+async def test_parse_instagram_attachment():
+    """Instagram DM with image attachment → media_type + attachment_url extracted."""
+    payload = _instagram_attachment_payload()
+    info = parse_webhook_payload("meta", payload)
+
+    assert info["trigger_type"] == "MESSAGE_INBOUND"
+    assert info["channel"] == "instagram"
+    assert info["media_type"] == "image"
+    assert info["attachment_url"] == "https://example.com/photo.jpg"
+    assert info["content"] == "[image]"
+
+
+@pytest.mark.asyncio
+async def test_parse_messenger_echo_detected():
+    """Messenger echo payload — is_echo is not handled in parse but in process_webhook_event."""
+    payload = _messenger_echo_payload()
+    # parse_webhook_payload still returns MESSAGE_INBOUND for echoes
+    # The echo detection happens in process_webhook_event (line 192-204)
+    info = parse_webhook_payload("meta", payload)
+    assert info["trigger_type"] == "MESSAGE_INBOUND"
+    assert info["channel"] == "messenger"
+
+    # Verify the echo flag is in the raw payload for process_webhook_event to detect
+    entry = payload["entry"][0]
+    assert entry["messaging"][0]["message"]["is_echo"] is True
+
+
+# ── Section 4: Meta Delivery Status (Watermark) ─────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_meta_delivery_watermark_updates(db_session: AsyncSession):
+    """Delivery watermark updates SENT messages to DELIVERED."""
+    user, ws, integration = await _setup_workspace(db_session)
+
+    # Use time.time() for proper Unix epoch (avoids utcnow().timestamp() local TZ mismatch)
+    now_epoch = time.time()
+    sent_time = datetime.utcfromtimestamp(now_epoch - 300)  # 5 min ago
+    msg = Message(
+        workspace_id=ws.id,
+        conversation_id=uuid4(),
+        direction="outbound",
+        content="Test message",
+        platform="meta",
+        delivery_status=DeliveryStatus.SENT,
+        sent_at=sent_time,
+    )
+    db_session.add(msg)
+    await db_session.flush()
+
+    # Watermark is 1 minute ago (after sent_time)
+    watermark = int((now_epoch - 60) * 1000)
+    info = {"meta_delivery_watermark": watermark, "meta_read_watermark": None}
+
+    await _process_meta_status_updates(
+        db_session, info, ws.id, str(uuid4())
+    )
+    await db_session.flush()
+
+    await db_session.refresh(msg)
+    assert msg.delivery_status == DeliveryStatus.DELIVERED
+    assert msg.delivered_at is not None
+
+
+@pytest.mark.asyncio
+async def test_meta_read_watermark_updates(db_session: AsyncSession):
+    """Read watermark updates DELIVERED messages to READ."""
+    user, ws, integration = await _setup_workspace(db_session)
+
+    now_epoch = time.time()
+    sent_time = datetime.utcfromtimestamp(now_epoch - 300)
+    msg = Message(
+        workspace_id=ws.id,
+        conversation_id=uuid4(),
+        direction="outbound",
+        content="Test read",
+        platform="meta",
+        delivery_status=DeliveryStatus.DELIVERED,
+        sent_at=sent_time,
+        delivered_at=datetime.utcfromtimestamp(now_epoch - 240),
+    )
+    db_session.add(msg)
+    await db_session.flush()
+
+    watermark = int((now_epoch - 60) * 1000)
+    info = {"meta_delivery_watermark": None, "meta_read_watermark": watermark}
+
+    await _process_meta_status_updates(
+        db_session, info, ws.id, str(uuid4())
+    )
+    await db_session.flush()
+
+    await db_session.refresh(msg)
+    assert msg.delivery_status == DeliveryStatus.READ
+    assert msg.read_at is not None
+
+
+@pytest.mark.asyncio
+async def test_meta_status_no_regression(db_session: AsyncSession):
+    """READ message should not regress to DELIVERED."""
+    user, ws, integration = await _setup_workspace(db_session)
+
+    now_epoch = time.time()
+    sent_time = datetime.utcfromtimestamp(now_epoch - 300)
+    msg = Message(
+        workspace_id=ws.id,
+        conversation_id=uuid4(),
+        direction="outbound",
+        content="Already read",
+        platform="meta",
+        delivery_status=DeliveryStatus.READ,
+        sent_at=sent_time,
+        delivered_at=datetime.utcfromtimestamp(now_epoch - 240),
+        read_at=datetime.utcfromtimestamp(now_epoch - 180),
+    )
+    db_session.add(msg)
+    await db_session.flush()
+
+    # Delivery watermark arrives after the message was already READ
+    watermark = int(now_epoch * 1000)
+    info = {"meta_delivery_watermark": watermark, "meta_read_watermark": None}
+
+    await _process_meta_status_updates(
+        db_session, info, ws.id, str(uuid4())
+    )
+    await db_session.flush()
+
+    await db_session.refresh(msg)
+    # Should still be READ, not regressed to DELIVERED
+    assert msg.delivery_status == DeliveryStatus.READ
+
+
+# ── Section 5: Lead Ads Ingestion ────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_parse_lead_fields_standard():
+    """parse_lead_fields normalizes email + phone + full_name."""
+    field_data = [
+        {"name": "email", "values": ["john@example.com"]},
+        {"name": "phone_number", "values": ["+14155551234"]},
+        {"name": "full_name", "values": ["John Doe"]},
+    ]
+    result = parse_lead_fields(field_data)
+
+    assert result["email"] == "john@example.com"
+    assert result["phone"] == "+14155551234"
+    assert result["full_name"] == "John Doe"
+    assert result["first_name"] == "John"
+    assert result["last_name"] == "Doe"
+
+
+@pytest.mark.asyncio
+async def test_parse_lead_fields_name_split():
+    """full_name only (no separate first/last) → correctly splits."""
+    field_data = [
+        {"name": "full_name", "values": ["Jane Smith-Doe"]},
+    ]
+    result = parse_lead_fields(field_data)
+
+    assert result["first_name"] == "Jane"
+    assert result["last_name"] == "Smith-Doe"
+
+
+@pytest.mark.asyncio
+async def test_parse_leadgen_webhook():
+    """Leadgen webhook payload → trigger_type=LEAD_AD_SUBMIT, leadgen_id extracted."""
+    payload = _leadgen_payload(leadgen_id="LEAD_XYZ")
+    info = parse_webhook_payload("meta", payload)
+
+    assert info["trigger_type"] == "LEAD_AD_SUBMIT"
+    assert info["leadgen_id"] == "LEAD_XYZ"
+    assert info["channel"] == "lead_ads"
+    assert info["provider_user_id"] == "LEAD_XYZ"
+
+
+# ── Section 6: MetaAdapter ───────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_meta_send_media_success():
+    """MetaAdapter.send_media returns provider_message_id on 200."""
+    from app.integrations.meta.adapter import MetaAdapter
+
+    adapter = MetaAdapter(page_id="PAGE_123", page_access_token="token_abc")
+    mock_resp = MagicMock(
+        status_code=200,
+        json=lambda: {"message_id": "mid.media_xyz"},
+    )
+
+    with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_resp):
+        result = await adapter.send_media("PSID_456", "image", "https://example.com/img.jpg")
+
+    assert result == "mid.media_xyz"
+
+
+@pytest.mark.asyncio
+async def test_meta_send_media_permanent_failure():
+    """MetaAdapter.send_media raises PERMANENT_FAILURE on 400."""
+    from app.integrations.meta.adapter import MetaAdapter
+
+    adapter = MetaAdapter(page_id="PAGE_123", page_access_token="token_abc")
+    mock_resp = MagicMock(
+        status_code=400,
+        json=lambda: {"error": {"code": 100, "message": "Invalid attachment"}},
+    )
+
+    with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_resp):
+        with pytest.raises(Exception, match="PERMANENT_FAILURE"):
+            await adapter.send_media("PSID_456", "image", "https://example.com/bad.jpg")
+
+
+@pytest.mark.asyncio
+async def test_meta_get_user_profile():
+    """MetaAdapter.get_user_profile returns profile on success, {} on failure."""
+    from app.integrations.meta.adapter import MetaAdapter
+
+    adapter = MetaAdapter(page_id="PAGE_123", page_access_token="token_abc")
+
+    # Success case
+    success_resp = MagicMock(
+        status_code=200,
+        json=lambda: {"first_name": "John", "last_name": "Doe"},
+    )
+    with patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=success_resp):
+        profile = await adapter.get_user_profile("PSID_456")
+    assert profile["first_name"] == "John"
+
+    # Failure case
+    fail_resp = MagicMock(status_code=403, text="Forbidden")
+    with patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=fail_resp):
+        profile = await adapter.get_user_profile("PSID_456")
+    assert profile == {}
+
+
+# ── Section 7: Delivery Status Parsing ───────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_parse_delivery_receipt():
+    """Delivery receipt payload → trigger_type=META_STATUS_UPDATE, watermark extracted."""
+    watermark = int(datetime.utcnow().timestamp() * 1000)
+    payload = _delivery_receipt_payload(watermark=watermark)
+    info = parse_webhook_payload("meta", payload)
+
+    assert info["trigger_type"] == "META_STATUS_UPDATE"
+    assert info["meta_delivery_watermark"] == watermark
+    assert info["meta_read_watermark"] is None
+
+
+@pytest.mark.asyncio
+async def test_parse_read_receipt():
+    """Read receipt payload → trigger_type=META_STATUS_UPDATE, read watermark extracted."""
+    watermark = int(datetime.utcnow().timestamp() * 1000)
+    payload = _read_receipt_payload(watermark=watermark)
+    info = parse_webhook_payload("meta", payload)
+
+    assert info["trigger_type"] == "META_STATUS_UPDATE"
+    assert info["meta_read_watermark"] == watermark
+
+
+# ── Section 8: Observability ─────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_meta_metrics_tracking():
+    """Meta-specific metrics increment correctly."""
+    m = MetricsService()
+    m.increment("meta_oauth_started")
+    m.increment("meta_oauth_completed")
+    m.increment("meta_leads_ingested")
+    m.increment("meta_leads_ingested")
+    m.increment("webhooks_received", labels={"provider": "meta", "channel": "instagram"})
+
+    counts = m.get_counts()
+    assert counts["meta_oauth_started"] == 1
+    assert counts["meta_oauth_completed"] == 1
+    assert counts["meta_leads_ingested"] == 2
+    assert counts["webhooks_received{channel=instagram,provider=meta}"] == 1