1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365//! WhatsApp Cloud API channel adapter.
//!
//! Uses the official WhatsApp Business Cloud API to send and receive messages.
//! Requires a webhook endpoint for incoming messages and the Cloud API for outgoing.
use crate::types::{ChannelAdapter, ChannelContent, ChannelMessage, ChannelType, ChannelUser};
use async_trait::async_trait;
use futures::Stream;
use std::pin::Pin;
use std::sync::Arc;
use tokio::sync::{mpsc, watch};
use tracing::{error, info};
use zeroize::Zeroizing;
const MAX_MESSAGE_LEN: usize = 4096;
/// WhatsApp Cloud API adapter.
///
/// Supports two modes:
/// - **Cloud API mode**: Uses the official WhatsApp Business Cloud API (requires Meta dev account).
/// - **Web/QR mode**: Routes outgoing messages through a local Baileys-based gateway process.
///
/// Mode is selected automatically: if `gateway_url` is set (from `WHATSAPP_WEB_GATEWAY_URL`),
/// the adapter uses Web mode. Otherwise it falls back to Cloud API mode.
pub struct WhatsAppAdapter {
/// WhatsApp Business phone number ID (Cloud API mode).
phone_number_id: String,
/// SECURITY: Access token is zeroized on drop.
access_token: Zeroizing<String>,
/// SECURITY: Verify token is zeroized on drop.
verify_token: Zeroizing<String>,
/// Port to listen for webhook callbacks (Cloud API mode).
webhook_port: u16,
/// HTTP client.
client: reqwest::Client,
/// Allowed phone numbers (empty = allow all).
allowed_users: Vec<String>,
/// Optional WhatsApp Web gateway URL for QR/Web mode (e.g. "http://127.0.0.1:3009").
gateway_url: Option<String>,
/// Shutdown signal.
shutdown_tx: Arc<watch::Sender<bool>>,
shutdown_rx: watch::Receiver<bool>,
}
impl WhatsAppAdapter {
/// Create a new WhatsApp Cloud API adapter.
pub fn new(
phone_number_id: String,
access_token: String,
verify_token: String,
webhook_port: u16,
allowed_users: Vec<String>,
) -> Self {
let (shutdown_tx, shutdown_rx) = watch::channel(false);
Self {
phone_number_id,
access_token: Zeroizing::new(access_token),
verify_token: Zeroizing::new(verify_token),
webhook_port,
client: reqwest::Client::new(),
allowed_users,
gateway_url: None,
shutdown_tx: Arc::new(shutdown_tx),
shutdown_rx,
}
}
/// Create a new WhatsApp adapter with gateway URL for Web/QR mode.
///
/// When `gateway_url` is `Some`, outgoing messages are sent via `POST {gateway_url}/message/send`
/// instead of the Cloud API. Incoming messages are handled by the gateway itself.
pub fn with_gateway(mut self, gateway_url: Option<String>) -> Self {
self.gateway_url = gateway_url.filter(|u| !u.is_empty());
self
}
/// Send a text message via the WhatsApp Cloud API.
async fn api_send_message(
&self,
to: &str,
text: &str,
) -> Result<(), Box<dyn std::error::Error>> {
let url = format!(
"https://graph.facebook.com/v21.0/{}/messages",
self.phone_number_id
);
// Split long messages
let chunks = crate::types::split_message(text, MAX_MESSAGE_LEN);
for chunk in chunks {
let body = serde_json::json!({
"messaging_product": "whatsapp",
"to": to,
"type": "text",
"text": { "body": chunk }
});
let resp = self
.client
.post(&url)
.bearer_auth(&*self.access_token)
.json(&body)
.send()
.await?;
if !resp.status().is_success() {
let status = resp.status();
let body = resp.text().await.unwrap_or_default();
error!("WhatsApp API error {status}: {body}");
return Err(format!("WhatsApp API error {status}: {body}").into());
}
}
Ok(())
}
/// Mark a message as read.
#[allow(dead_code)]
async fn api_mark_read(&self, message_id: &str) -> Result<(), Box<dyn std::error::Error>> {
let url = format!(
"https://graph.facebook.com/v21.0/{}/messages",
self.phone_number_id
);
let body = serde_json::json!({
"messaging_product": "whatsapp",
"status": "read",
"message_id": message_id
});
let _ = self
.client
.post(&url)
.bearer_auth(&*self.access_token)
.json(&body)
.send()
.await;
Ok(())
}
/// Send a text message via the WhatsApp Web gateway.
async fn gateway_send_message(
&self,
gateway_url: &str,
to: &str,
text: &str,
) -> Result<(), Box<dyn std::error::Error>> {
let url = format!("{}/message/send", gateway_url.trim_end_matches('/'));
let body = serde_json::json!({ "to": to, "text": text });
let resp = self.client.post(&url).json(&body).send().await?;
if !resp.status().is_success() {
let status = resp.status();
let body = resp.text().await.unwrap_or_default();
error!("WhatsApp gateway error {status}: {body}");
return Err(format!("WhatsApp gateway error {status}: {body}").into());
}
Ok(())
}
/// Check if a phone number is allowed.
#[allow(dead_code)]
fn is_allowed(&self, phone: &str) -> bool {
self.allowed_users.is_empty() || self.allowed_users.iter().any(|u| u == phone)
}
/// Returns true if this adapter is configured for Web/QR gateway mode.
#[allow(dead_code)]
pub fn is_gateway_mode(&self) -> bool {
self.gateway_url.is_some()
}
}
#[async_trait]
impl ChannelAdapter for WhatsAppAdapter {
fn name(&self) -> &str {
"whatsapp"
}
fn channel_type(&self) -> ChannelType {
ChannelType::WhatsApp
}
async fn start(
&self,
) -> Result<Pin<Box<dyn Stream<Item = ChannelMessage> + Send>>, Box<dyn std::error::Error>>
{
let (_tx, rx) = mpsc::channel::<ChannelMessage>(256);
let port = self.webhook_port;
let _verify_token = self.verify_token.clone();
let _allowed_users = self.allowed_users.clone();
let _access_token = self.access_token.clone();
let _phone_number_id = self.phone_number_id.clone();
let mut shutdown_rx = self.shutdown_rx.clone();
info!("Starting WhatsApp webhook listener on port {port}");
tokio::spawn(async move {
// Simple webhook polling simulation
// In production, this would be an axum HTTP server handling webhook POSTs
// For now, log that the webhook is ready
info!("WhatsApp webhook ready on port {port} (verify_token configured)");
info!("Configure your webhook URL: https://your-domain:{port}/webhook");
// Wait for shutdown
let _ = shutdown_rx.changed().await;
info!("WhatsApp adapter stopped");
});
Ok(Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx)))
}
async fn send(
&self,
user: &ChannelUser,
content: ChannelContent,
) -> Result<(), Box<dyn std::error::Error>> {
// Web/QR gateway mode: route all messages through the gateway
if let Some(ref gw) = self.gateway_url {
let text = match &content {
ChannelContent::Text(t) => t.clone(),
ChannelContent::Image { caption, .. } => {
caption
.clone()
.unwrap_or_else(|| "(Image — not supported in Web mode)".to_string())
}
ChannelContent::File { filename, .. } => {
format!("(File: {filename} — not supported in Web mode)")
}
_ => "(Unsupported content type in Web mode)".to_string(),
};
// Split long messages the same way as Cloud API mode
let chunks = crate::types::split_message(&text, MAX_MESSAGE_LEN);
for chunk in chunks {
self.gateway_send_message(gw, &user.platform_id, chunk)
.await?;
}
return Ok(());
}
// Cloud API mode (default)
match content {
ChannelContent::Text(text) => {
self.api_send_message(&user.platform_id, &text).await?;
}
ChannelContent::Image { url, caption } => {
let body = serde_json::json!({
"messaging_product": "whatsapp",
"to": user.platform_id,
"type": "image",
"image": {
"link": url,
"caption": caption.unwrap_or_default()
}
});
let api_url = format!(
"https://graph.facebook.com/v21.0/{}/messages",
self.phone_number_id
);
self.client
.post(&api_url)
.bearer_auth(&*self.access_token)
.json(&body)
.send()
.await?;
}
ChannelContent::File { url, filename } => {
let body = serde_json::json!({
"messaging_product": "whatsapp",
"to": user.platform_id,
"type": "document",
"document": {
"link": url,
"filename": filename
}
});
let api_url = format!(
"https://graph.facebook.com/v21.0/{}/messages",
self.phone_number_id
);
self.client
.post(&api_url)
.bearer_auth(&*self.access_token)
.json(&body)
.send()
.await?;
}
ChannelContent::Location { lat, lon } => {
let body = serde_json::json!({
"messaging_product": "whatsapp",
"to": user.platform_id,
"type": "location",
"location": {
"latitude": lat,
"longitude": lon
}
});
let api_url = format!(
"https://graph.facebook.com/v21.0/{}/messages",
self.phone_number_id
);
self.client
.post(&api_url)
.bearer_auth(&*self.access_token)
.json(&body)
.send()
.await?;
}
_ => {
self.api_send_message(&user.platform_id, "(Unsupported content type)")
.await?;
}
}
Ok(())
}
async fn stop(&self) -> Result<(), Box<dyn std::error::Error>> {
let _ = self.shutdown_tx.send(true);
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_whatsapp_adapter_creation() {
let adapter = WhatsAppAdapter::new(
"12345".to_string(),
"access_token".to_string(),
"verify_token".to_string(),
8443,
vec![],
);
assert_eq!(adapter.name(), "whatsapp");
assert_eq!(adapter.channel_type(), ChannelType::WhatsApp);
}
#[test]
fn test_allowed_users_check() {
let adapter = WhatsAppAdapter::new(
"12345".to_string(),
"token".to_string(),
"verify".to_string(),
8443,
vec!["+1234567890".to_string()],
);
assert!(adapter.is_allowed("+1234567890"));
assert!(!adapter.is_allowed("+9999999999"));
let open = WhatsAppAdapter::new(
"12345".to_string(),
"token".to_string(),
"verify".to_string(),
8443,
vec![],
);
assert!(open.is_allowed("+anything"));
}
}