|
|
@@ -19,11 +19,15 @@
|
|
|
//! [`WaEvent::Error`].
|
|
|
|
|
|
use std::collections::HashMap;
|
|
|
-use std::path::Path;
|
|
|
+use std::io::ErrorKind;
|
|
|
+use std::path::{Path, PathBuf};
|
|
|
use std::sync::{Arc, Mutex, MutexGuard};
|
|
|
+use std::time::Duration;
|
|
|
|
|
|
use anyhow::{Context, Result, anyhow, bail};
|
|
|
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
|
|
|
+use tokio::sync::watch;
|
|
|
+use tokio::task::AbortHandle;
|
|
|
use whatsapp_rust::prelude::*;
|
|
|
use whatsapp_rust::types::events::{
|
|
|
ConnectFailure, ContactUpdate, LazyHistorySync, MessageBatch, PushNameUpdate,
|
|
|
@@ -78,6 +82,25 @@ const CLIENT_STOPPED_REASON: &str = "client stopped";
|
|
|
/// Reason reported when another session replaced this one.
|
|
|
const STREAM_REPLACED_REASON: &str = "stream replaced by another session";
|
|
|
|
|
|
+/// Error returned by client operations while no session is running (during a relink).
|
|
|
+const NO_SESSION_MESSAGE: &str = "WhatsApp session not running";
|
|
|
+
|
|
|
+/// Seconds a relink waits for the old client run loop to stop after the logout.
|
|
|
+const STOP_TIMEOUT_SECONDS: u64 = 10;
|
|
|
+
|
|
|
+/// Suffixes of the SQLite side files of the session database.
|
|
|
+const SESSION_SIDE_FILE_SUFFIXES: [&str; 2] = ["-wal", "-shm"];
|
|
|
+
|
|
|
+/// Seconds the session reset waits for SQLite locks held by the stopped client.
|
|
|
+const SESSION_BUSY_TIMEOUT_SECONDS: u64 = 5;
|
|
|
+
|
|
|
+/// Table where the storage library records its applied migrations (kept by the session reset).
|
|
|
+const MIGRATIONS_TABLE: &str = "__diesel_schema_migrations";
|
|
|
+
|
|
|
+/// Query listing the session tables to empty, except SQLite internals and the migrations table.
|
|
|
+const SESSION_TABLES_QUERY: &str =
|
|
|
+ "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND name <> ?1";
|
|
|
+
|
|
|
/// Event emitted by the WhatsApp client wrapper.
|
|
|
#[derive(Debug, Clone)]
|
|
|
pub enum WaEvent {
|
|
|
@@ -134,14 +157,85 @@ pub struct DownloadedImage {
|
|
|
pub mime: String,
|
|
|
}
|
|
|
|
|
|
-/// Handle used to send messages through the running client.
|
|
|
+/// Handle to the running WhatsApp session. Clones share the same session, which
|
|
|
+/// [`WaHandle::relink`] can replace with a new one.
|
|
|
#[derive(Clone)]
|
|
|
pub struct WaHandle {
|
|
|
+ /// Running session; `None` while a relink replaces it (or after a failed relink).
|
|
|
+ session: Arc<Mutex<Option<Session>>>,
|
|
|
+ /// SQLite session file, deleted and recreated by a relink.
|
|
|
+ session_db: PathBuf,
|
|
|
+ /// Destination of the events of every session.
|
|
|
+ events: UnboundedSender<WaEvent>,
|
|
|
+}
|
|
|
+
|
|
|
+/// One running client and the tasks serving it.
|
|
|
+struct Session {
|
|
|
/// Shared client driving the connection.
|
|
|
client: Arc<Client>,
|
|
|
+ /// Translator task; it holds the client, so it is aborted once the session stops.
|
|
|
+ translator: AbortHandle,
|
|
|
+ /// Becomes `true` once the client run loop has stopped.
|
|
|
+ stopped: watch::Receiver<bool>,
|
|
|
+}
|
|
|
+
|
|
|
+impl Session {
|
|
|
+ /// Logs out and releases the client, so its session database can be deleted.
|
|
|
+ ///
|
|
|
+ /// Inputs: the session (consumed). Output: none.
|
|
|
+ /// Side effects: deregisters this device on the phone when connected (best effort),
|
|
|
+ /// disconnects, waits up to `STOP_TIMEOUT_SECONDS` for the run loop to stop and aborts the
|
|
|
+ /// translator task. After a timeout the database may stay open for a while: the deletion
|
|
|
+ /// retries and reports it.
|
|
|
+ async fn stop(mut self) {
|
|
|
+ self.client.logout().await;
|
|
|
+ let stopped = self.stopped.wait_for(|stopped| *stopped);
|
|
|
+
|
|
|
+ // Elapsed or closed both mean there is nothing more to wait for.
|
|
|
+ let _ = tokio::time::timeout(Duration::from_secs(STOP_TIMEOUT_SECONDS), stopped).await;
|
|
|
+ self.translator.abort();
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
impl WaHandle {
|
|
|
+ /// Returns the client of the running session.
|
|
|
+ ///
|
|
|
+ /// Inputs: none. Output: the client, or an error while no session is running.
|
|
|
+ /// Side effects: none.
|
|
|
+ fn client(&self) -> Result<Arc<Client>> {
|
|
|
+ self.lock_session()
|
|
|
+ .as_ref()
|
|
|
+ .map(|session| Arc::clone(&session.client))
|
|
|
+ .ok_or_else(|| anyhow!(NO_SESSION_MESSAGE))
|
|
|
+ }
|
|
|
+
|
|
|
+ /// Locks the session slot, recovering from a poisoned lock.
|
|
|
+ ///
|
|
|
+ /// Inputs: none. Output: the guard. Side effects: none.
|
|
|
+ fn lock_session(&self) -> MutexGuard<'_, Option<Session>> {
|
|
|
+ self.session
|
|
|
+ .lock()
|
|
|
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
|
|
|
+ }
|
|
|
+
|
|
|
+ /// Replaces the session with a new, unpaired one, so a new QR code is offered.
|
|
|
+ ///
|
|
|
+ /// Inputs: none. Output: `Ok(())` once the new client is started, or the deletion or start
|
|
|
+ /// error (the handle then has no session until the next successful relink).
|
|
|
+ /// Side effects: logs out and stops the current client (see [`Session::stop`]), empties the
|
|
|
+ /// session database (see [`reset_session_files`]) and starts a new client on it, whose
|
|
|
+ /// pairing QR codes arrive as [`WaEvent::Qr`].
|
|
|
+ pub async fn relink(&self) -> Result<()> {
|
|
|
+ let current = self.lock_session().take();
|
|
|
+ if let Some(current) = current {
|
|
|
+ current.stop().await;
|
|
|
+ }
|
|
|
+ reset_session_files(&self.session_db)?;
|
|
|
+ let session = open_session(&self.session_db, self.events.clone()).await?;
|
|
|
+ *self.lock_session() = Some(session);
|
|
|
+ Ok(())
|
|
|
+ }
|
|
|
+
|
|
|
/// Sends a plain text message.
|
|
|
///
|
|
|
/// Inputs: `chat_jid` is a JID in `user@server` form, `text` the body (non-empty).
|
|
|
@@ -155,7 +249,7 @@ impl WaHandle {
|
|
|
.parse()
|
|
|
.map_err(|e| anyhow!("invalid JID '{chat_jid}': {e}"))?;
|
|
|
let sent = self
|
|
|
- .client
|
|
|
+ .client()?
|
|
|
.send_message(jid, wa::Message::text(text))
|
|
|
.await
|
|
|
.with_context(|| format!("failed to send message to {chat_jid}"))?;
|
|
|
@@ -169,7 +263,7 @@ impl WaHandle {
|
|
|
/// Side effects: HTTPS request to the WhatsApp CDN; the bytes stay in memory.
|
|
|
pub async fn download_image(&self, image: &ImageRef) -> Result<DownloadedImage> {
|
|
|
let bytes = self
|
|
|
- .client
|
|
|
+ .client()?
|
|
|
.download(image.message.as_ref())
|
|
|
.await
|
|
|
.context("failed to download the image from WhatsApp")?;
|
|
|
@@ -186,10 +280,14 @@ impl WaHandle {
|
|
|
///
|
|
|
/// Inputs: none (reads the client's device identity).
|
|
|
/// Output: the phone-number JID and the LID JID, whichever are known, in
|
|
|
- /// `user@server` form without device suffix; empty before pairing completes.
|
|
|
+ /// `user@server` form without device suffix; empty before pairing completes and while no
|
|
|
+ /// session is running.
|
|
|
/// Side effects: none.
|
|
|
pub fn own_jids(&self) -> Vec<String> {
|
|
|
- [self.client.pn(), self.client.lid()]
|
|
|
+ let Ok(client) = self.client() else {
|
|
|
+ return Vec::new();
|
|
|
+ };
|
|
|
+ [client.pn(), client.lid()]
|
|
|
.into_iter()
|
|
|
.flatten()
|
|
|
.map(|jid| jid_to_string(&jid))
|
|
|
@@ -200,12 +298,83 @@ impl WaHandle {
|
|
|
/// Starts the WhatsApp client in the background.
|
|
|
///
|
|
|
/// Inputs: `session_db` is the SQLite session file (created when missing),
|
|
|
-/// `events` receives every [`WaEvent`].
|
|
|
+/// `events` receives every [`WaEvent`], of this session and of the ones started by
|
|
|
+/// [`WaHandle::relink`].
|
|
|
/// Output: a [`WaHandle`] for sending messages.
|
|
|
+/// Side effects: see [`open_session`].
|
|
|
+pub async fn start(session_db: &Path, events: UnboundedSender<WaEvent>) -> Result<WaHandle> {
|
|
|
+ let session = open_session(session_db, events.clone()).await?;
|
|
|
+ Ok(WaHandle {
|
|
|
+ session: Arc::new(Mutex::new(Some(session))),
|
|
|
+ session_db: session_db.to_path_buf(),
|
|
|
+ events,
|
|
|
+ })
|
|
|
+}
|
|
|
+
|
|
|
+/// Leaves an empty session database, so the next client starts unpaired.
|
|
|
+///
|
|
|
+/// Inputs: `session_db`, the database path. Output: `Ok(())`, or the error naming the path.
|
|
|
+/// Side effects: deletes the database and its SQLite side files. When the database is still
|
|
|
+/// open (Windows refuses to delete it: the library never releases a stopped client before the
|
|
|
+/// process exits) its tables are emptied in place instead; the stopped client writes nothing more,
|
|
|
+/// because its saver flushes once on shutdown and then exits.
|
|
|
+fn reset_session_files(session_db: &Path) -> Result<()> {
|
|
|
+ match std::fs::remove_file(session_db) {
|
|
|
+ Ok(()) => remove_side_files(session_db),
|
|
|
+ Err(e) if e.kind() == ErrorKind::NotFound => remove_side_files(session_db),
|
|
|
+ Err(e) => clear_session_tables(session_db).with_context(|| {
|
|
|
+ format!("cannot delete ({e}) nor clear session database {}", session_db.display())
|
|
|
+ }),
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/// Deletes the SQLite side files (`-wal`, `-shm`) of a deleted session database.
|
|
|
+///
|
|
|
+/// Inputs: `session_db`, the database path. Output: `Ok(())`, or the error naming the file.
|
|
|
+/// Side effects: removes files; missing ones are fine.
|
|
|
+fn remove_side_files(session_db: &Path) -> Result<()> {
|
|
|
+ for suffix in SESSION_SIDE_FILE_SUFFIXES {
|
|
|
+ let mut file = session_db.as_os_str().to_owned();
|
|
|
+ file.push(suffix);
|
|
|
+ match std::fs::remove_file(&file) {
|
|
|
+ Err(e) if e.kind() != ErrorKind::NotFound => {
|
|
|
+ return Err(e).with_context(|| format!("cannot delete session file {}", Path::new(&file).display()));
|
|
|
+ }
|
|
|
+ _ => {}
|
|
|
+ }
|
|
|
+ }
|
|
|
+ Ok(())
|
|
|
+}
|
|
|
+
|
|
|
+/// Empties every table of the session database, keeping the schema and its migration record.
|
|
|
+///
|
|
|
+/// Inputs: `session_db`, the database path. Output: `Ok(())`, or the SQLite error.
|
|
|
+/// Side effects: deletes all rows in one transaction; waits up to `SESSION_BUSY_TIMEOUT_SECONDS`
|
|
|
+/// for locks held by other connections.
|
|
|
+fn clear_session_tables(session_db: &Path) -> Result<()> {
|
|
|
+ let mut connection = rusqlite::Connection::open(session_db)?;
|
|
|
+ connection.busy_timeout(Duration::from_secs(SESSION_BUSY_TIMEOUT_SECONDS))?;
|
|
|
+ let transaction = connection.transaction()?;
|
|
|
+ let tables = transaction
|
|
|
+ .prepare(SESSION_TABLES_QUERY)?
|
|
|
+ .query_map([MIGRATIONS_TABLE], |row| row.get::<_, String>(0))?
|
|
|
+ .collect::<rusqlite::Result<Vec<_>>>()?;
|
|
|
+ for table in tables {
|
|
|
+ transaction.execute(&format!("DELETE FROM \"{table}\""), [])?;
|
|
|
+ }
|
|
|
+ transaction.commit()?;
|
|
|
+ Ok(())
|
|
|
+}
|
|
|
+
|
|
|
+/// Opens the session database and starts a client on it.
|
|
|
+///
|
|
|
+/// Inputs: `session_db` is the SQLite session file (created when missing),
|
|
|
+/// `events` receives every [`WaEvent`] of this session.
|
|
|
+/// Output: the running [`Session`].
|
|
|
/// Side effects: creates the parent directory and the session database,
|
|
|
/// spawns tokio tasks (client run loop and event translator) and opens the
|
|
|
/// network connection. Pairing and connection results arrive as events.
|
|
|
-pub async fn start(session_db: &Path, events: UnboundedSender<WaEvent>) -> Result<WaHandle> {
|
|
|
+async fn open_session(session_db: &Path, events: UnboundedSender<WaEvent>) -> Result<Session> {
|
|
|
let db_url = session_db
|
|
|
.to_str()
|
|
|
.ok_or_else(|| anyhow!("session database path is not valid UTF-8"))?;
|
|
|
@@ -228,16 +397,22 @@ pub async fn start(session_db: &Path, events: UnboundedSender<WaEvent>) -> Resul
|
|
|
|
|
|
let client = bot.client();
|
|
|
let translator = Arc::new(Translator::new(client.clone(), events));
|
|
|
- tokio::spawn(Arc::clone(&translator).run(raw_rx));
|
|
|
+ let translator_task = tokio::spawn(Arc::clone(&translator).run(raw_rx)).abort_handle();
|
|
|
|
|
|
+ let (stopped_tx, stopped) = watch::channel(false);
|
|
|
let bot_handle = bot.spawn();
|
|
|
tokio::spawn(async move {
|
|
|
// Keeping the handle alive keeps the bot running; resolves when it stops.
|
|
|
bot_handle.await;
|
|
|
translator.emit(WaEvent::Disconnected(CLIENT_STOPPED_REASON.to_owned()));
|
|
|
+ stopped_tx.send_replace(true);
|
|
|
});
|
|
|
|
|
|
- Ok(WaHandle { client })
|
|
|
+ Ok(Session {
|
|
|
+ client,
|
|
|
+ translator: translator_task,
|
|
|
+ stopped,
|
|
|
+ })
|
|
|
}
|
|
|
|
|
|
/// Event-bus handler that forwards raw library events, in order, to the translator task.
|
|
|
@@ -414,7 +589,7 @@ impl Translator {
|
|
|
match event {
|
|
|
Event::PairingQrCode(qr) => self.emit(WaEvent::Qr(qr.code.clone())),
|
|
|
Event::PairingQrCodesExhausted(_) => {
|
|
|
- self.report("pairing QR codes exhausted; restart to pair again".to_owned());
|
|
|
+ self.report("pairing QR codes exhausted; request a new QR to pair again".to_owned());
|
|
|
}
|
|
|
Event::PairError(e) => self.report(format!("pairing failed: {}", e.error)),
|
|
|
Event::Connected(_) => self.emit(WaEvent::Connected),
|
|
|
@@ -966,6 +1141,47 @@ fn image_caption_part(trimmed: &str) -> Option<&str> {
|
|
|
mod tests {
|
|
|
use super::*;
|
|
|
|
|
|
+ /// Seconds a live test waits for a pairing QR code.
|
|
|
+ const LIVE_QR_TIMEOUT_SECONDS: u64 = 60;
|
|
|
+
|
|
|
+ /// Waits for the next pairing QR code, skipping every other event.
|
|
|
+ async fn next_qr(events: &mut UnboundedReceiver<WaEvent>) -> String {
|
|
|
+ let wait = async {
|
|
|
+ loop {
|
|
|
+ match events.recv().await {
|
|
|
+ Some(WaEvent::Qr(code)) => return code,
|
|
|
+ Some(_) => {}
|
|
|
+ None => panic!("event channel closed before a QR code"),
|
|
|
+ }
|
|
|
+ }
|
|
|
+ };
|
|
|
+ tokio::time::timeout(Duration::from_secs(LIVE_QR_TIMEOUT_SECONDS), wait)
|
|
|
+ .await
|
|
|
+ .expect("no QR code in time")
|
|
|
+ }
|
|
|
+
|
|
|
+ /// Live check against the WhatsApp servers (needs network): run with `--ignored`.
|
|
|
+ #[tokio::test(flavor = "multi_thread")]
|
|
|
+ #[ignore = "connects to the WhatsApp servers"]
|
|
|
+ async fn relink_deletes_the_session_and_offers_a_new_qr() {
|
|
|
+ let session_db = PathBuf::from("target").join(format!("wh-relink-test-{}.db", std::process::id()));
|
|
|
+ let (events_tx, mut events) = unbounded_channel();
|
|
|
+ let handle = start(&session_db, events_tx).await.expect("client must start");
|
|
|
+ let first = next_qr(&mut events).await;
|
|
|
+
|
|
|
+ handle.relink().await.expect("relink must succeed");
|
|
|
+ let second = next_qr(&mut events).await;
|
|
|
+ let session = handle.lock_session().take().expect("relink must leave a session");
|
|
|
+ session.stop().await;
|
|
|
+
|
|
|
+ // The file stays open until the process exits on Windows: only empty it.
|
|
|
+ reset_session_files(&session_db).expect("test session must be resettable");
|
|
|
+
|
|
|
+ // QR payload: `ref,noise key,identity key,adv secret,client type`; a new device has new keys.
|
|
|
+ let device_keys = |qr: &str| qr.split(',').skip(1).take(3).collect::<Vec<_>>().join(",");
|
|
|
+ assert_ne!(device_keys(&first), device_keys(&second));
|
|
|
+ }
|
|
|
+
|
|
|
/// Builds a message with the given plain text.
|
|
|
fn text_message(text: &str) -> wa::Message {
|
|
|
wa::Message {
|
