restore: reset past source tree deletion, apply pending work

2026-03-27 10:49:39 +00:00
parent 04214ca155
commit 3571511349
21 changed files with 2799 additions and 37 deletions
@@ -0,0 +1,312 @@
+//! WhatsApp Business API integration.
+//!
+//! Provides:
+//! - [`WhatsAppTransport`] — a [`ChatTransport`] that sends messages via the
+//!   Meta Graph API (`graph.facebook.com/v21.0/{phone_number_id}/messages`).
+//! - [`MessagingWindowTracker`] — tracks the 24-hour messaging window per user.
+//! - [`webhook_verify`] / [`webhook_receive`] — Poem handlers for the WhatsApp
+//!   webhook (GET verification handshake + POST incoming messages).
+
+pub mod commands;
+pub mod format;
+pub mod history;
+pub mod meta;
+pub mod twilio;
+
+pub use history::{load_whatsapp_history, MessagingWindowTracker, WhatsAppConversationHistory};
+pub use meta::WhatsAppTransport;
+pub use twilio::{extract_twilio_text_messages, TwilioWhatsAppTransport};
+
+use serde::Deserialize;
+use std::collections::{HashMap, HashSet};
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
+use tokio::sync::{Mutex as TokioMutex, oneshot};
+
+use crate::agents::AgentPool;
+use crate::chat::ChatTransport;
+use crate::http::context::{PermissionDecision, PermissionForward};
+use crate::slog;
+use poem::{Request, Response, handler, http::StatusCode, web::Query};
+
+// ── Webhook types (Meta → us) ───────────────────────────────────────────
+
+/// Top-level webhook payload from Meta.
+#[derive(Deserialize, Debug)]
+pub struct WebhookPayload {
+    #[serde(default)]
+    pub entry: Vec<WebhookEntry>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct WebhookEntry {
+    #[serde(default)]
+    pub changes: Vec<WebhookChange>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct WebhookChange {
+    pub value: Option<WebhookValue>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct WebhookValue {
+    #[serde(default)]
+    pub messages: Vec<WebhookMessage>,
+    #[allow(dead_code)] // Present in Meta webhook JSON, kept for deserialization
+    pub metadata: Option<WebhookMetadata>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct WebhookMetadata {
+    #[allow(dead_code)]
+    pub phone_number_id: Option<String>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct WebhookMessage {
+    pub from: Option<String>,
+    pub r#type: Option<String>,
+    pub text: Option<WebhookText>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct WebhookText {
+    pub body: Option<String>,
+}
+
+/// Extract text messages from a webhook payload.
+///
+/// Returns `(sender_phone, message_body)` pairs.
+pub fn extract_text_messages(payload: &WebhookPayload) -> Vec<(String, String)> {
+    let mut messages = Vec::new();
+    for entry in &payload.entry {
+        for change in &entry.changes {
+            if let Some(value) = &change.value {
+                for msg in &value.messages {
+                    if msg.r#type.as_deref() == Some("text")
+                        && let (Some(from), Some(text)) = (&msg.from, &msg.text)
+                        && let Some(body) = &text.body
+                    {
+                        messages.push((from.clone(), body.clone()));
+                    }
+                }
+            }
+        }
+    }
+    messages
+}
+
+/// Query parameters for the webhook verification GET request.
+#[derive(Deserialize)]
+pub struct VerifyQuery {
+    #[serde(rename = "hub.mode")]
+    pub hub_mode: Option<String>,
+    #[serde(rename = "hub.verify_token")]
+    pub hub_verify_token: Option<String>,
+    #[serde(rename = "hub.challenge")]
+    pub hub_challenge: Option<String>,
+}
+
+/// Shared context for webhook handlers, injected via Poem's `Data` extractor.
+pub struct WhatsAppWebhookContext {
+    pub verify_token: String,
+    /// Active provider: `"meta"` (Meta Graph API) or `"twilio"` (Twilio REST API).
+    pub provider: String,
+    pub transport: Arc<dyn ChatTransport>,
+    pub project_root: PathBuf,
+    pub agents: Arc<AgentPool>,
+    pub bot_name: String,
+    /// The bot's "user ID" for command dispatch (e.g. "whatsapp-bot").
+    pub bot_user_id: String,
+    pub ambient_rooms: Arc<Mutex<HashSet<String>>>,
+    /// Per-sender conversation history for LLM passthrough.
+    pub history: WhatsAppConversationHistory,
+    /// Maximum number of conversation entries to keep per sender.
+    pub history_size: usize,
+    /// Tracks the 24-hour messaging window per user phone number.
+    pub window_tracker: Arc<MessagingWindowTracker>,
+    /// Phone numbers allowed to send messages to the bot.
+    /// When empty, all numbers are allowed (backwards compatible).
+    pub allowed_phones: Vec<String>,
+    /// Permission requests from the MCP `prompt_permission` tool arrive here.
+    pub perm_rx: Arc<TokioMutex<tokio::sync::mpsc::UnboundedReceiver<PermissionForward>>>,
+    /// Pending permission replies keyed by sender phone number.
+    pub pending_perm_replies:
+        Arc<TokioMutex<HashMap<String, oneshot::Sender<PermissionDecision>>>>,
+    /// Seconds before an unanswered permission prompt is auto-denied.
+    pub permission_timeout_secs: u64,
+}
+
+/// GET /webhook/whatsapp — webhook verification.
+///
+/// For Meta: responds to the `hub.mode=subscribe` challenge handshake.
+/// For Twilio: Twilio does not send GET verification; always returns 200 OK.
+#[handler]
+pub async fn webhook_verify(
+    Query(q): Query<VerifyQuery>,
+    ctx: poem::web::Data<&Arc<WhatsAppWebhookContext>>,
+) -> Response {
+    // Twilio does not use a GET challenge; just acknowledge.
+    if ctx.provider == "twilio" {
+        return Response::builder().status(StatusCode::OK).body("ok");
+    }
+
+    // Meta verification handshake.
+    if q.hub_mode.as_deref() == Some("subscribe")
+        && q.hub_verify_token.as_deref() == Some(&ctx.verify_token)
+        && let Some(challenge) = q.hub_challenge
+    {
+        slog!("[whatsapp] Webhook verification succeeded");
+        return Response::builder().status(StatusCode::OK).body(challenge);
+    }
+    slog!("[whatsapp] Webhook verification failed");
+    Response::builder()
+        .status(StatusCode::FORBIDDEN)
+        .body("Verification failed")
+}
+
+/// POST /webhook/whatsapp — receive incoming messages.
+///
+/// Dispatches to the appropriate parser based on the configured provider:
+/// - `"meta"`: parses Meta's JSON `WebhookPayload`.
+/// - `"twilio"`: parses Twilio's `application/x-www-form-urlencoded` body.
+///
+/// Both providers expect a `200 OK` response, even on parse errors.
+#[handler]
+pub async fn webhook_receive(
+    req: &Request,
+    body: poem::Body,
+    ctx: poem::web::Data<&Arc<WhatsAppWebhookContext>>,
+) -> Response {
+    let _ = req;
+    let bytes = match body.into_bytes().await {
+        Ok(b) => b,
+        Err(e) => {
+            slog!("[whatsapp] Failed to read webhook body: {e}");
+            return Response::builder()
+                .status(StatusCode::BAD_REQUEST)
+                .body("Bad request");
+        }
+    };
+
+    let messages = if ctx.provider == "twilio" {
+        let msgs = extract_twilio_text_messages(&bytes);
+        if msgs.is_empty() {
+            slog!("[whatsapp/twilio] No text messages in webhook body; ignoring");
+        }
+        msgs
+    } else {
+        let payload: WebhookPayload = match serde_json::from_slice(&bytes) {
+            Ok(p) => p,
+            Err(e) => {
+                slog!("[whatsapp] Failed to parse webhook payload: {e}");
+                // Meta expects 200 even on parse errors to avoid retries.
+                return Response::builder().status(StatusCode::OK).body("ok");
+            }
+        };
+        let msgs = extract_text_messages(&payload);
+        if msgs.is_empty() {
+            // Status updates, read receipts, etc. — acknowledge silently.
+            return Response::builder().status(StatusCode::OK).body("ok");
+        }
+        msgs
+    };
+
+    if messages.is_empty() {
+        return Response::builder().status(StatusCode::OK).body("ok");
+    }
+
+    let ctx = Arc::clone(*ctx);
+    tokio::spawn(async move {
+        for (sender, text) in messages {
+            slog!("[whatsapp] Message from {sender}: {text}");
+            commands::handle_incoming_message(&ctx, &sender, &text).await;
+        }
+    });
+
+    Response::builder().status(StatusCode::OK).body("ok")
+}
+
+// ── Tests ───────────────────────────────────────────────────────────────
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // ── Existing webhook / transport tests ────────────────────────────
+
+    #[test]
+    fn extract_text_messages_parses_valid_payload() {
+        let json = r#"{
+            "entry": [{
+                "changes": [{
+                    "value": {
+                        "messages": [{
+                            "from": "15551234567",
+                            "type": "text",
+                            "text": { "body": "help" }
+                        }],
+                        "metadata": { "phone_number_id": "123456" }
+                    }
+                }]
+            }]
+        }"#;
+        let payload: WebhookPayload = serde_json::from_str(json).unwrap();
+        let msgs = extract_text_messages(&payload);
+        assert_eq!(msgs.len(), 1);
+        assert_eq!(msgs[0].0, "15551234567");
+        assert_eq!(msgs[0].1, "help");
+    }
+
+    #[test]
+    fn extract_text_messages_ignores_non_text() {
+        let json = r#"{
+            "entry": [{
+                "changes": [{
+                    "value": {
+                        "messages": [{
+                            "from": "15551234567",
+                            "type": "image",
+                            "image": { "id": "img123" }
+                        }],
+                        "metadata": { "phone_number_id": "123456" }
+                    }
+                }]
+            }]
+        }"#;
+        let payload: WebhookPayload = serde_json::from_str(json).unwrap();
+        let msgs = extract_text_messages(&payload);
+        assert!(msgs.is_empty());
+    }
+
+    #[test]
+    fn extract_text_messages_handles_empty_payload() {
+        let json = r#"{ "entry": [] }"#;
+        let payload: WebhookPayload = serde_json::from_str(json).unwrap();
+        let msgs = extract_text_messages(&payload);
+        assert!(msgs.is_empty());
+    }
+
+    #[test]
+    fn extract_text_messages_handles_multiple_messages() {
+        let json = r#"{
+            "entry": [{
+                "changes": [{
+                    "value": {
+                        "messages": [
+                            { "from": "111", "type": "text", "text": { "body": "status" } },
+                            { "from": "222", "type": "text", "text": { "body": "help" } }
+                        ],
+                        "metadata": { "phone_number_id": "123456" }
+                    }
+                }]
+            }]
+        }"#;
+        let payload: WebhookPayload = serde_json::from_str(json).unwrap();
+        let msgs = extract_text_messages(&payload);
+        assert_eq!(msgs.len(), 2);
+        assert_eq!(msgs[0].1, "status");
+        assert_eq!(msgs[1].1, "help");
+    }
+}