diff --git a/Cargo.toml b/Cargo.toml index 7ea81ff..ba0d83d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,7 +30,7 @@ serde = { version = "1.0.229", features = ["derive"] } serde_json = { version = "1.0.154", features = ["preserve_order"] } sha1 = "0.11.0" sha2 = "0.11.1" -sqlx = { version = "0.9", features = ["runtime-tokio", "sqlite", "chrono"] } +sqlx = { version = "0.9", features = ["runtime-tokio", "sqlite", "postgres", "chrono"] } tokio = { version = "1", features = ["full"] } tokio-xmpp = { version = "6", features = ["insecure-tcp"] } tower-http = { version = "0.7", features = ["fs"] } diff --git a/README.md b/README.md index 3f0e769..2e2cb7a 100644 --- a/README.md +++ b/README.md @@ -2,8 +2,8 @@ A rewrite of the [Movim](https://movim.eu) XMPP web client backend in Rust, keeping Movim's frontend (CSS/JS/theme, in `frontend/`). The PHP daemon, widgets and -templates were replaced by a single Rust server (tokio, axum, tokio-xmpp, sqlx/SQLite, -minijinja). Scope: a Conversations-like chat client. +templates were replaced by a single Rust server (tokio, axum, tokio-xmpp, sqlx with SQLite +or PostgreSQL, minijinja). Scope: a Conversations-like chat client. ## Features 1:1 chat, group chats (MUC, bookmarks, invitations, moderation, configuration), @@ -35,8 +35,22 @@ XMPP server's archive expiry), dev only: `MOVIM_XMPP_PLAINTEXT=host:port`, `MOVIM_ALLOW_PRIVATE_URLS=1`, `MOVIM_DEBUG_XMPP=1`. +## Database +SQLite by default (`data/movim.sqlite`). For PostgreSQL, set +`MOVIM_DATABASE_URL=postgres://user:password@host/database`; the tables are +created at startup. To move an existing SQLite database, stop Movim, create an +empty PostgreSQL database and run: + + movim migrate-to-postgres postgres://user:password@host/database [--from sqlite:///path/movim.sqlite] + +Without `--from` it copies the SQLite database Movim would use (`MOVIM_DATABASE_URL` +or the one in `MOVIM_DATA_DIR`). Everything is copied in one transaction, ids +included, and the command refuses a database that already has data. The SQLite +file is left untouched. + ## Tests -`cargo test` runs the unit tests. `tests/e2e/` holds end-to-end scripts that +`cargo test` runs the unit tests (`MOVIM_TEST_POSTGRES_URL=postgres://…` also +runs them against PostgreSQL, in throwaway databases). `tests/e2e/` holds end-to-end scripts that drive the server (and the real frontend in headless Chromium) against a local Prosody; see `tests/e2e/README.md`. diff --git a/data/movim.sqlite b/data/movim.sqlite index 6f93d91..782ec7b 100644 Binary files a/data/movim.sqlite and b/data/movim.sqlite differ diff --git a/data/movim.sqlite-shm b/data/movim.sqlite-shm deleted file mode 100644 index 07da3a4..0000000 Binary files a/data/movim.sqlite-shm and /dev/null differ diff --git a/data/movim.sqlite-wal b/data/movim.sqlite-wal deleted file mode 100644 index 82936b9..0000000 Binary files a/data/movim.sqlite-wal and /dev/null differ diff --git a/frontend/widgets/Account/account.js b/frontend/widgets/Account/account.js index 7b8940e..8108453 100644 --- a/frontend/widgets/Account/account.js +++ b/frontend/widgets/Account/account.js @@ -76,10 +76,13 @@ var Account = { Object.keys(localStorage).filter(k => k.startsWith(prefix)).forEach(k => keys[k] = localStorage.getItem(k)); let deviceId = await store.getLocalRegistrationId(); + // The decrypted bodies too: our own messages are not encrypted for this device and, + // the sessions being reset on restore, the older ones cannot be decrypted again let payload = JSON.stringify({ jid: store.jid, deviceId: deviceId, - keys: keys + keys: keys, + messages: await ChatOmemoDB.getAllMessages() }); const salt = crypto.getRandomValues(new Uint8Array(16)); @@ -143,6 +146,15 @@ var Account = { Object.keys(localStorage).filter(k => k.startsWith(prefix)).forEach(k => localStorage.removeItem(k)); Object.entries(payload.keys).forEach(([k, v]) => localStorage.setItem(k, v)); + // The ratchets of the backup are behind the ones of our contacts: they cannot decrypt + // their new messages and break our next ones, new sessions are built on the next message + store.removeAllSessions(); + + // Backups made before the messages were included don't have them + if (Array.isArray(payload.messages)) { + await ChatOmemoDB.putMissingMessages(payload.messages); + } + localStorage.setObject(OMEMO_RESTORED_KEY, { replaced: replaced != payload.deviceId ? replaced ?? null : null }); // The page reload publishes the restored key (ChatOmemo.ownDevicesReceived) diff --git a/frontend/widgets/ChatOmemo/chatomemo.js b/frontend/widgets/ChatOmemo/chatomemo.js index 5624013..75e29ee 100644 --- a/frontend/widgets/ChatOmemo/chatomemo.js +++ b/frontend/widgets/ChatOmemo/chatomemo.js @@ -272,12 +272,29 @@ var ChatOmemo = { var store = new ChatOmemoStorage(); - let resolvedId = message.mine + // Our own messages are stored under the id this browser sent them with: in 1:1 + // the ones coming back from the archive (MAM) have another local id (m_…) + let resolvedId = (message.mine || message.jidfrom == USER_JID) && message.messageid ? message.messageid : message.id; + // A received message can be decrypted only once, but its local id changes when it comes + // back from the archive (MAM): its plaintext is also kept under the sender and the id they chose + let stableId = resolvedId == message.id && message.messageid + ? ['received', message.jidfrom, message.type == 'groupchat' ? message.resource : '', message.messageid].join('|') + : null; + + if (stableId) { + let stable = await ChatOmemoDB.getMessage(stableId); + if (typeof stable == 'string') return stable; + } + let maybeDecrypted = await ChatOmemoDB.getMessage(resolvedId); + if (typeof maybeDecrypted == 'string' && stableId) { + ChatOmemoDB.putMessage(stableId, maybeDecrypted); + } + if (maybeDecrypted !== undefined) { return maybeDecrypted; } @@ -369,6 +386,7 @@ var ChatOmemo = { let plaintext = MovimUtils.arrayBufferToString(decryptedBuffer); ChatOmemoDB.putMessage(resolvedId, plaintext); + if (stableId) ChatOmemoDB.putMessage(stableId, plaintext); return plaintext; }, enableContactState: function (jid, muc) { diff --git a/frontend/widgets/ChatOmemo/chatomemo_db.js b/frontend/widgets/ChatOmemo/chatomemo_db.js index cd41bbe..1745c97 100644 --- a/frontend/widgets/ChatOmemo/chatomemo_db.js +++ b/frontend/widgets/ChatOmemo/chatomemo_db.js @@ -65,6 +65,41 @@ var ChatOmemoDB = { return promise; }, + // All the decrypted bodies, for the key backup (Account.omemoBackup) + getAllMessages: async function() { + return new Promise(resolve => { + var tx = ChatOmemoDB.db.result.transaction(ChatOmemoDB.storeName, 'readonly'); + var request = tx.objectStore(ChatOmemoDB.storeName).getAll(); + request.onsuccess = function(event) { + resolve(event.target.result.filter(m => typeof m.body == 'string')); + } + request.onerror = function() { + resolve([]); + } + }); + }, + + // Restore them without overwriting what this browser already decrypted + putMissingMessages: async function(messages) { + return new Promise(resolve => { + var tx = ChatOmemoDB.db.result.transaction(ChatOmemoDB.storeName, 'readwrite'); + var os = tx.objectStore(ChatOmemoDB.storeName); + + messages.forEach(m => { + if (typeof m.id != 'string' || typeof m.body != 'string') return; + + os.get(m.id).onsuccess = function(event) { + if (!event.target.result || typeof event.target.result.body != 'string') { + os.put({"id" : m.id, "body" : m.body}); + } + } + }); + + tx.oncomplete = () => resolve(); + tx.onerror = () => resolve(); + }); + }, + getMessage: async function(id) { let promise = new Promise(resolve => { if (id in ChatOmemoDB.cachedMessages) { diff --git a/src/db.rs b/src/db.rs index ffe4a5a..7253996 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1,14 +1,91 @@ -//! SQLite persistence: users, quick-login passwords, contacts (vCards/avatars), -//! messages, files, reactions and MAM state. +//! Persistence (SQLite or PostgreSQL): users, quick-login passwords, contacts +//! (vCards/avatars), messages, files, reactions and MAM state. +//! +//! Queries are written once, in SQL both databases understand, with SQLite's `?` +//! placeholders: `Db::sql` rewrites them for PostgreSQL. `on!` compiles the code +//! running a query once per database, rows keep their native types. use anyhow::Result; use serde::Serialize; +use sqlx::postgres::{PgPool, PgPoolOptions}; use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous}; use sqlx::{AssertSqlSafe, FromRow, SqlitePool}; use std::str::FromStr; use std::time::Duration; -pub type Db = SqlitePool; +#[derive(Clone, Debug)] +pub enum Db { + Sqlite(SqlitePool), + Postgres(PgPool), +} + +/// Run `$body` with `$p` bound to the pool of the database (a trailing `;` +/// discards its value, the types of query results differ between databases) +macro_rules! on { + ($db:expr, $p:ident => $body:expr ;) => { + match $db { + Db::Sqlite($p) => { + $body; + } + Db::Postgres($p) => { + $body; + } + } + }; + ($db:expr, $p:ident => $body:expr) => { + match $db { + Db::Sqlite($p) => $body, + Db::Postgres($p) => $body, + } + }; +} + +/// SQLite's CURRENT_TIMESTAMP, dates are stored as text in this format +const POSTGRES_NOW: &str = "to_char(now() AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS')"; + +/// `?` placeholders become `$1`, `$2`… and CURRENT_TIMESTAMP keeps SQLite's format +pub fn to_postgres(sql: &str) -> String { + let sql = sql.replace("CURRENT_TIMESTAMP", POSTGRES_NOW); + let mut out = String::with_capacity(sql.len() + 16); + let (mut n, mut quoted) = (0, false); + for c in sql.chars() { + match c { + '\'' => { + quoted = !quoted; + out.push(c); + } + '?' if !quoted => { + n += 1; + out.push_str(&format!("${n}")); + } + _ => out.push(c), + } + } + out +} + +impl Db { + pub fn sql(&self, sql: &str) -> AssertSqlSafe { + match self { + Db::Sqlite(_) => AssertSqlSafe(sql.to_string()), + Db::Postgres(_) => AssertSqlSafe(to_postgres(sql)), + } + } + + #[cfg(test)] + pub async fn close(&self) { + on!(self, p => p.close().await) + } +} + +pub fn is_postgres_url(url: &str) -> bool { + url.starts_with("postgres://") || url.starts_with("postgresql://") +} + +/// A date `days` ago, in the stored format +fn days_ago_sql(days: i64) -> String { + (chrono::Utc::now() - chrono::Duration::days(days)).format("%Y-%m-%d %H:%M:%S").to_string() +} const SCHEMA: &str = r#" CREATE TABLE IF NOT EXISTS users ( @@ -164,7 +241,180 @@ CREATE TABLE IF NOT EXISTS push_subscriptions ( ); "#; +/// Same tables as `SCHEMA`: real booleans, 64 bits integers, and the binary +/// collation of SQLite on the texts (the dates are compared as texts). +/// push_subscriptions gets a `rowid` column, the implicit one of SQLite. +const POSTGRES_SCHEMA: &str = r#" +CREATE TABLE IF NOT EXISTS users ( + id TEXT COLLATE "C" PRIMARY KEY, + nickname TEXT, + language TEXT, + nightmode BOOLEAN NOT NULL DEFAULT FALSE, + chatmain BOOLEAN NOT NULL DEFAULT TRUE, + notificationchat BOOLEAN NOT NULL DEFAULT TRUE, + notificationcall BOOLEAN NOT NULL DEFAULT TRUE, + omemoenabled BOOLEAN NOT NULL DEFAULT TRUE, + accentcolor TEXT, + admin BOOLEAN NOT NULL DEFAULT FALSE, + chats TEXT NOT NULL DEFAULT '{}', + created_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS encrypted_passwords ( + id TEXT COLLATE "C" NOT NULL, + user_id TEXT COLLATE "C" NOT NULL, + data TEXT NOT NULL, + updated_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (user_id, id) +); + +CREATE TABLE IF NOT EXISTS contacts ( + id TEXT COLLATE "C" PRIMARY KEY, + fn TEXT, + name TEXT, + nickname TEXT, + email TEXT, + url TEXT, + description TEXT, + avatarhash TEXT, + avatartype TEXT, + updated_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS avatars ( + hash TEXT COLLATE "C" PRIMARY KEY, + mime TEXT NOT NULL, + data BYTEA NOT NULL +); + +CREATE TABLE IF NOT EXISTS messages ( + mid BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + user_id TEXT COLLATE "C" NOT NULL, + id TEXT COLLATE "C" NOT NULL, + messageid TEXT COLLATE "C", + originid TEXT COLLATE "C", + stanzaid TEXT COLLATE "C", + replaceid TEXT COLLATE "C", + jidto TEXT COLLATE "C" NOT NULL, + jidfrom TEXT COLLATE "C" NOT NULL, + resource TEXT COLLATE "C", + type TEXT COLLATE "C" NOT NULL DEFAULT 'chat', + subject TEXT, + thread TEXT COLLATE "C", + body TEXT, + html TEXT, + quoted BOOLEAN NOT NULL DEFAULT FALSE, + markable BOOLEAN NOT NULL DEFAULT FALSE, + seen BOOLEAN NOT NULL DEFAULT FALSE, + delivered TEXT COLLATE "C", + displayed TEXT COLLATE "C", + retracted BOOLEAN NOT NULL DEFAULT FALSE, + encrypted BOOLEAN NOT NULL DEFAULT FALSE, + omemoheader TEXT, + bundleid BIGINT, + parentmid BIGINT, + resolved BOOLEAN NOT NULL DEFAULT FALSE, + published TEXT COLLATE "C" NOT NULL, + created_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE (user_id, jidfrom, id) +); +CREATE INDEX IF NOT EXISTS messages_user_jidfrom ON messages (user_id, jidfrom); +CREATE INDEX IF NOT EXISTS messages_user_jidto ON messages (user_id, jidto); +CREATE INDEX IF NOT EXISTS messages_user_published ON messages (user_id, published); +CREATE INDEX IF NOT EXISTS messages_published ON messages (published); +CREATE INDEX IF NOT EXISTS messages_user_stanzaid ON messages (user_id, stanzaid); +CREATE INDEX IF NOT EXISTS messages_user_originid ON messages (user_id, originid); +CREATE INDEX IF NOT EXISTS messages_user_messageid ON messages (user_id, messageid); +CREATE INDEX IF NOT EXISTS messages_user_replaceid ON messages (user_id, replaceid); + +CREATE TABLE IF NOT EXISTS message_files ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + message_mid BIGINT NOT NULL, + name TEXT, + type TEXT, + size BIGINT, + url TEXT NOT NULL, + disposition TEXT, + thumbnail_url TEXT, + thumbnail_type TEXT, + thumbnail_width BIGINT, + thumbnail_height BIGINT +); +CREATE INDEX IF NOT EXISTS message_files_mid ON message_files (message_mid); + +CREATE TABLE IF NOT EXISTS reactions ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + message_mid BIGINT NOT NULL, + jidfrom TEXT COLLATE "C" NOT NULL, + emoji TEXT COLLATE "C" NOT NULL, + created_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE (message_mid, jidfrom, emoji) +); + +CREATE TABLE IF NOT EXISTS mam_earliests ( + user_id TEXT COLLATE "C" NOT NULL, + jid TEXT COLLATE "C" NOT NULL, + earliest TEXT COLLATE "C", + PRIMARY KEY (user_id, jid) +); + +CREATE TABLE IF NOT EXISTS history_clears ( + user_id TEXT COLLATE "C" NOT NULL, + jid TEXT COLLATE "C" NOT NULL, + cleared_at TEXT COLLATE "C" NOT NULL, + PRIMARY KEY (user_id, jid) +); + +CREATE TABLE IF NOT EXISTS omemo_states ( + user_id TEXT COLLATE "C" NOT NULL, + jid TEXT COLLATE "C" NOT NULL, + enabled BOOLEAN NOT NULL, + PRIMARY KEY (user_id, jid) +); + +CREATE TABLE IF NOT EXISTS open_chats ( + user_id TEXT COLLATE "C" NOT NULL, + jid TEXT COLLATE "C" NOT NULL, + updated_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (user_id, jid) +); + +CREATE TABLE IF NOT EXISTS urls ( + hash TEXT COLLATE "C" PRIMARY KEY, + url TEXT NOT NULL, + data TEXT NOT NULL, + created_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS push_subscriptions ( + rowid BIGINT GENERATED BY DEFAULT AS IDENTITY UNIQUE, + endpoint TEXT COLLATE "C" PRIMARY KEY, + user_id TEXT COLLATE "C" NOT NULL, + p256dh TEXT, + auth TEXT, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + activity_at TEXT COLLATE "C" NOT NULL DEFAULT CURRENT_TIMESTAMP +); +"#; + +/// Tables with an automatic id: its sequence must follow copied rows +pub const IDENTITY_COLUMNS: &[(&str, &str)] = + &[("messages", "mid"), ("message_files", "id"), ("reactions", "id"), ("push_subscriptions", "rowid")]; + pub async fn connect(url: &str) -> Result { + if is_postgres_url(url) { + let pool = PgPoolOptions::new().max_connections(16).connect(url).await?; + for statement in POSTGRES_SCHEMA.split(';') { + let statement = statement.trim(); + if !statement.is_empty() { + sqlx::query(AssertSqlSafe(to_postgres(statement))).execute(&pool).await?; + } + } + return Ok(Db::Postgres(pool)); + } + // WAL: readers never block the writer. synchronous=NORMAL is safe in WAL // (no corruption, a power loss can only drop the last commits) and avoids an // fsync per message. Writers queue on the busy timeout instead of failing. @@ -179,7 +429,7 @@ pub async fn connect(url: &str) -> Result { sqlx::query(statement).execute(&pool).await?; } } - Ok(pool) + Ok(Db::Sqlite(pool)) } // ---------------------------------------------------------------- Users @@ -203,21 +453,21 @@ impl User { } pub async fn user_get(db: &Db, id: &str) -> Result> { - Ok(sqlx::query_as::<_, User>( + let sql = db.sql( "SELECT id, nickname, language, nightmode, chatmain, notificationchat, notificationcall, omemoenabled, accentcolor, admin, chats FROM users WHERE id = ?", - ) - .bind(id) - .fetch_optional(db) - .await?) + ); + Ok(on!(db, p => sqlx::query_as::<_, User>(sql).bind(id).fetch_optional(p).await?)) } pub async fn user_get_or_create(db: &Db, id: &str) -> Result { - sqlx::query("INSERT OR IGNORE INTO users (id) VALUES (?)").bind(id).execute(db).await?; + let sql = db.sql("INSERT INTO users (id) VALUES (?) ON CONFLICT DO NOTHING"); + on!(db, p => sqlx::query(sql).bind(id).execute(p).await?;); Ok(user_get(db, id).await?.unwrap_or_default()) } pub async fn user_count(db: &Db) -> Result { - Ok(sqlx::query_scalar("SELECT COUNT(*) FROM users").fetch_one(db).await?) + let sql = db.sql("SELECT COUNT(*) FROM users"); + Ok(on!(db, p => sqlx::query_scalar(sql).fetch_one(p).await?)) } pub async fn user_set(db: &Db, id: &str, field: &str, value: &serde_json::Value) -> Result<()> { @@ -233,58 +483,52 @@ pub async fn user_set(db: &Db, id: &str, field: &str, value: &serde_json::Value) "chats", ]; anyhow::ensure!(ALLOWED.contains(&field), "field not allowed"); - let sql = format!("UPDATE users SET {field} = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?"); - let q = sqlx::query(AssertSqlSafe(sql)); - let q = match value { - serde_json::Value::Bool(b) => q.bind(*b), - serde_json::Value::Null => q.bind(None::), - serde_json::Value::String(s) => q.bind(s.clone()), - other => q.bind(other.to_string()), - }; - q.bind(id).execute(db).await?; + let sql = db.sql(&format!("UPDATE users SET {field} = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?")); + on!(db, p => { + let q = sqlx::query(sql); + let q = match value { + serde_json::Value::Bool(b) => q.bind(*b), + serde_json::Value::Null => q.bind(None::), + serde_json::Value::String(s) => q.bind(s.clone()), + other => q.bind(other.to_string()), + }; + q.bind(id).execute(p).await?; + }); Ok(()) } // ---------------------------------------------------------------- Quick login pub async fn encrypted_password_set(db: &Db, user_id: &str, id: &str, data: &str) -> Result<()> { - sqlx::query("INSERT OR REPLACE INTO encrypted_passwords (id, user_id, data, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP)") - .bind(id) - .bind(user_id) - .bind(data) - .execute(db) - .await?; + let sql = db.sql( + "INSERT INTO encrypted_passwords (id, user_id, data, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP) + ON CONFLICT(user_id, id) DO UPDATE SET data = excluded.data, updated_at = excluded.updated_at", + ); + on!(db, p => sqlx::query(sql).bind(id).bind(user_id).bind(data).execute(p).await?;); Ok(()) } pub async fn encrypted_password_get(db: &Db, user_id: &str, id: &str) -> Result> { - let data = sqlx::query_scalar("SELECT data FROM encrypted_passwords WHERE user_id = ? AND id = ?") - .bind(user_id) - .bind(id) - .fetch_optional(db) - .await?; + let sql = db.sql("SELECT data FROM encrypted_passwords WHERE user_id = ? AND id = ?"); + let data: Option = on!(db, p => sqlx::query_scalar(sql).bind(user_id).bind(id).fetch_optional(p).await?); if data.is_some() { - sqlx::query("UPDATE encrypted_passwords SET updated_at = CURRENT_TIMESTAMP WHERE user_id = ? AND id = ?") - .bind(user_id) - .bind(id) - .execute(db) - .await?; + let sql = db.sql("UPDATE encrypted_passwords SET updated_at = CURRENT_TIMESTAMP WHERE user_id = ? AND id = ?"); + on!(db, p => sqlx::query(sql).bind(user_id).bind(id).execute(p).await?;); } Ok(data) } pub async fn encrypted_passwords_clear(db: &Db, user_id: &str) -> Result<()> { - sqlx::query("DELETE FROM encrypted_passwords WHERE user_id = ?").bind(user_id).execute(db).await?; + let sql = db.sql("DELETE FROM encrypted_passwords WHERE user_id = ?"); + on!(db, p => sqlx::query(sql).bind(user_id).execute(p).await?;); Ok(()) } pub async fn cleanup(db: &Db) -> Result<()> { - sqlx::query("DELETE FROM encrypted_passwords WHERE updated_at < datetime('now', '-7 days')") - .execute(db) - .await?; - sqlx::query("DELETE FROM push_subscriptions WHERE activity_at < datetime('now', '-30 days')") - .execute(db) - .await?; + let sql = db.sql("DELETE FROM encrypted_passwords WHERE updated_at < ?"); + on!(db, p => sqlx::query(sql).bind(days_ago_sql(7)).execute(p).await?;); + let sql = db.sql("DELETE FROM push_subscriptions WHERE activity_at < ?"); + on!(db, p => sqlx::query(sql).bind(days_ago_sql(30)).execute(p).await?;); Ok(()) } @@ -324,10 +568,8 @@ impl Contact { } pub async fn contact_get(db: &Db, id: &str) -> Result> { - Ok(sqlx::query_as::<_, Contact>("SELECT id, fn, name, nickname, email, url, description, avatarhash, avatartype FROM contacts WHERE id = ?") - .bind(id) - .fetch_optional(db) - .await?) + let sql = db.sql("SELECT id, fn, name, nickname, email, url, description, avatarhash, avatartype FROM contacts WHERE id = ?"); + Ok(on!(db, p => sqlx::query_as::<_, Contact>(sql).bind(id).fetch_optional(p).await?)) } pub async fn contact_get_or_new(db: &Db, id: &str) -> Contact { @@ -339,81 +581,79 @@ pub async fn contacts_get(db: &Db, ids: &[String]) -> Result> { return Ok(vec![]); } let placeholders = vec!["?"; ids.len()].join(","); - let sql = format!("SELECT id, fn, name, nickname, email, url, description, avatarhash, avatartype FROM contacts WHERE id IN ({placeholders})"); - let mut q = sqlx::query_as::<_, Contact>(AssertSqlSafe(sql)); - for id in ids { - q = q.bind(id); - } - Ok(q.fetch_all(db).await?) + let sql = db.sql(&format!( + "SELECT id, fn, name, nickname, email, url, description, avatarhash, avatartype FROM contacts WHERE id IN ({placeholders})" + )); + Ok(on!(db, p => { + let mut q = sqlx::query_as::<_, Contact>(sql); + for id in ids { + q = q.bind(id); + } + q.fetch_all(p).await? + })) } pub async fn contact_save(db: &Db, c: &Contact) -> Result<()> { - sqlx::query( + let sql = db.sql( "INSERT INTO contacts (id, fn, name, nickname, email, url, description, avatarhash, avatartype, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(id) DO UPDATE SET fn = excluded.fn, name = excluded.name, nickname = excluded.nickname, email = excluded.email, url = excluded.url, description = excluded.description, avatarhash = excluded.avatarhash, avatartype = excluded.avatartype, updated_at = CURRENT_TIMESTAMP", - ) - .bind(&c.id) - .bind(&c.fn_) - .bind(&c.name) - .bind(&c.nickname) - .bind(&c.email) - .bind(&c.url) - .bind(&c.description) - .bind(&c.avatarhash) - .bind(&c.avatartype) - .execute(db) - .await?; + ); + on!(db, p => { + sqlx::query(sql) + .bind(&c.id) + .bind(&c.fn_) + .bind(&c.name) + .bind(&c.nickname) + .bind(&c.email) + .bind(&c.url) + .bind(&c.description) + .bind(&c.avatarhash) + .bind(&c.avatartype) + .execute(p) + .await?; + }); Ok(()) } pub async fn contact_set_avatar(db: &Db, id: &str, hash: Option<&str>, kind: &str) -> Result<()> { - sqlx::query( + let sql = db.sql( "INSERT INTO contacts (id, avatarhash, avatartype) VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET avatarhash = excluded.avatarhash, avatartype = excluded.avatartype, updated_at = CURRENT_TIMESTAMP", - ) - .bind(id) - .bind(hash) - .bind(kind) - .execute(db) - .await?; + ); + on!(db, p => sqlx::query(sql).bind(id).bind(hash).bind(kind).execute(p).await?;); Ok(()) } pub async fn contact_set_nickname(db: &Db, id: &str, nickname: Option<&str>) -> Result<()> { - sqlx::query( + let sql = db.sql( "INSERT INTO contacts (id, nickname) VALUES (?, ?) ON CONFLICT(id) DO UPDATE SET nickname = excluded.nickname, updated_at = CURRENT_TIMESTAMP", - ) - .bind(id) - .bind(nickname) - .execute(db) - .await?; + ); + on!(db, p => sqlx::query(sql).bind(id).bind(nickname).execute(p).await?;); Ok(()) } pub async fn avatar_exists(db: &Db, hash: &str) -> Result { - let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM avatars WHERE hash = ?").bind(hash).fetch_one(db).await?; + let sql = db.sql("SELECT COUNT(*) FROM avatars WHERE hash = ?"); + let n: i64 = on!(db, p => sqlx::query_scalar(sql).bind(hash).fetch_one(p).await?); Ok(n > 0) } pub async fn avatar_save(db: &Db, hash: &str, mime: &str, data: &[u8]) -> Result<()> { - sqlx::query("INSERT OR REPLACE INTO avatars (hash, mime, data) VALUES (?, ?, ?)") - .bind(hash) - .bind(mime) - .bind(data) - .execute(db) - .await?; + let sql = db.sql( + "INSERT INTO avatars (hash, mime, data) VALUES (?, ?, ?) + ON CONFLICT(hash) DO UPDATE SET mime = excluded.mime, data = excluded.data", + ); + on!(db, p => sqlx::query(sql).bind(hash).bind(mime).bind(data).execute(p).await?;); Ok(()) } pub async fn avatar_get(db: &Db, hash: &str) -> Result)>> { - Ok(sqlx::query_as::<_, (String, Vec)>("SELECT mime, data FROM avatars WHERE hash = ?") - .bind(hash) - .fetch_optional(db) - .await?) + let sql = db.sql("SELECT mime, data FROM avatars WHERE hash = ?"); + Ok(on!(db, p => sqlx::query_as::<_, (String, Vec)>(sql).bind(hash).fetch_optional(p).await?)) } // ---------------------------------------------------------------- Messages @@ -520,141 +760,134 @@ pub struct Reaction { /// Find the stored version of an incoming message, see `Message::findByStanza` pub async fn message_find_by_stanzaid(db: &Db, user: &str, jidfrom: &str, stanzaid: &str) -> Result> { - Ok(sqlx::query_as::<_, Message>(AssertSqlSafe(format!( + let sql = db.sql(&format!( "SELECT {MESSAGE_COLUMNS} FROM messages WHERE user_id = ? AND jidfrom = ? AND stanzaid = ? LIMIT 1" - ))) - .bind(user) - .bind(jidfrom) - .bind(stanzaid) - .fetch_optional(db) - .await?) + )); + Ok(on!(db, p => sqlx::query_as::<_, Message>(sql).bind(user).bind(jidfrom).bind(stanzaid).fetch_optional(p).await?)) } pub async fn message_find_by_originid(db: &Db, user: &str, jidfrom: &str, originid: &str) -> Result> { - Ok(sqlx::query_as::<_, Message>(AssertSqlSafe(format!( + let sql = db.sql(&format!( "SELECT {MESSAGE_COLUMNS} FROM messages WHERE user_id = ? AND jidfrom = ? AND (originid = ? OR id = ?) LIMIT 1" - ))) - .bind(user) - .bind(jidfrom) - .bind(originid) - .bind(originid) - .fetch_optional(db) - .await?) + )); + Ok(on!(db, p => { + sqlx::query_as::<_, Message>(sql).bind(user).bind(jidfrom).bind(originid).bind(originid).fetch_optional(p).await? + })) } pub async fn message_by_mid(db: &Db, user: &str, mid: i64) -> Result> { - Ok(sqlx::query_as::<_, Message>(AssertSqlSafe(format!("SELECT {MESSAGE_COLUMNS} FROM messages WHERE user_id = ? AND mid = ?"))) - .bind(user) - .bind(mid) - .fetch_optional(db) - .await?) + let sql = db.sql(&format!("SELECT {MESSAGE_COLUMNS} FROM messages WHERE user_id = ? AND mid = ?")); + Ok(on!(db, p => sqlx::query_as::<_, Message>(sql).bind(user).bind(mid).fetch_optional(p).await?)) } pub async fn message_by_id(db: &Db, user: &str, id: &str) -> Result> { - Ok(sqlx::query_as::<_, Message>(AssertSqlSafe(format!("SELECT {MESSAGE_COLUMNS} FROM messages WHERE user_id = ? AND id = ? LIMIT 1"))) - .bind(user) - .bind(id) - .fetch_optional(db) - .await?) + let sql = db.sql(&format!("SELECT {MESSAGE_COLUMNS} FROM messages WHERE user_id = ? AND id = ? LIMIT 1")); + Ok(on!(db, p => sqlx::query_as::<_, Message>(sql).bind(user).bind(id).fetch_optional(p).await?)) } /// Resolve a message referenced by another one (reply, reaction, retraction, receipt) /// from the conversation `jid`, by its message id, origin id or stanza id. pub async fn message_resolve_reference(db: &Db, user: &str, jid: &str, reference: &str) -> Result> { - Ok(sqlx::query_as::<_, Message>(AssertSqlSafe(format!( + let sql = db.sql(&format!( "SELECT {MESSAGE_COLUMNS} FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?) AND (messageid = ? OR originid = ? OR stanzaid = ? OR id = ?) ORDER BY mid DESC LIMIT 1" - ))) - .bind(user) - .bind(jid) - .bind(jid) - .bind(reference) - .bind(reference) - .bind(reference) - .bind(reference) - .fetch_optional(db) - .await?) + )); + Ok(on!(db, p => { + sqlx::query_as::<_, Message>(sql) + .bind(user) + .bind(jid) + .bind(jid) + .bind(reference) + .bind(reference) + .bind(reference) + .bind(reference) + .fetch_optional(p) + .await? + })) } /// Insert or update a message, returns its mid pub async fn message_save(db: &Db, m: &Message) -> Result { if m.mid > 0 { - sqlx::query( + let sql = db.sql( "UPDATE messages SET id = ?, messageid = ?, originid = ?, stanzaid = ?, replaceid = ?, jidto = ?, jidfrom = ?, resource = ?, type = ?, subject = ?, thread = ?, body = ?, html = ?, quoted = ?, markable = ?, seen = ?, delivered = ?, displayed = ?, retracted = ?, encrypted = ?, omemoheader = ?, bundleid = ?, parentmid = ?, resolved = ?, published = ?, updated_at = CURRENT_TIMESTAMP WHERE mid = ?", - ) - .bind(&m.id) - .bind(&m.messageid) - .bind(&m.originid) - .bind(&m.stanzaid) - .bind(&m.replaceid) - .bind(&m.jidto) - .bind(&m.jidfrom) - .bind(&m.resource) - .bind(&m.type_) - .bind(&m.subject) - .bind(&m.thread) - .bind(&m.body) - .bind(&m.html) - .bind(m.quoted) - .bind(m.markable) - .bind(m.seen) - .bind(&m.delivered) - .bind(&m.displayed) - .bind(m.retracted) - .bind(m.encrypted) - .bind(&m.omemoheader) - .bind(m.bundleid) - .bind(m.parentmid) - .bind(m.resolved) - .bind(&m.published) - .bind(m.mid) - .execute(db) - .await?; + ); + on!(db, p => { + sqlx::query(sql) + .bind(&m.id) + .bind(&m.messageid) + .bind(&m.originid) + .bind(&m.stanzaid) + .bind(&m.replaceid) + .bind(&m.jidto) + .bind(&m.jidfrom) + .bind(&m.resource) + .bind(&m.type_) + .bind(&m.subject) + .bind(&m.thread) + .bind(&m.body) + .bind(&m.html) + .bind(m.quoted) + .bind(m.markable) + .bind(m.seen) + .bind(&m.delivered) + .bind(&m.displayed) + .bind(m.retracted) + .bind(m.encrypted) + .bind(&m.omemoheader) + .bind(m.bundleid) + .bind(m.parentmid) + .bind(m.resolved) + .bind(&m.published) + .bind(m.mid) + .execute(p) + .await?; + }); return Ok(m.mid); } - let res = sqlx::query( + let sql = db.sql( "INSERT INTO messages (user_id, id, messageid, originid, stanzaid, replaceid, jidto, jidfrom, resource, type, subject, thread, body, html, quoted, markable, seen, delivered, displayed, retracted, encrypted, omemoheader, bundleid, parentmid, resolved, published) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(user_id, jidfrom, id) DO UPDATE SET body = excluded.body, stanzaid = COALESCE(excluded.stanzaid, stanzaid), - seen = MAX(seen, excluded.seen), updated_at = CURRENT_TIMESTAMP + ON CONFLICT(user_id, jidfrom, id) DO UPDATE SET body = excluded.body, stanzaid = COALESCE(excluded.stanzaid, messages.stanzaid), + seen = (messages.seen OR excluded.seen), updated_at = CURRENT_TIMESTAMP RETURNING mid", - ) - .bind(&m.user_id) - .bind(&m.id) - .bind(&m.messageid) - .bind(&m.originid) - .bind(&m.stanzaid) - .bind(&m.replaceid) - .bind(&m.jidto) - .bind(&m.jidfrom) - .bind(&m.resource) - .bind(&m.type_) - .bind(&m.subject) - .bind(&m.thread) - .bind(&m.body) - .bind(&m.html) - .bind(m.quoted) - .bind(m.markable) - .bind(m.seen) - .bind(&m.delivered) - .bind(&m.displayed) - .bind(m.retracted) - .bind(m.encrypted) - .bind(&m.omemoheader) - .bind(m.bundleid) - .bind(m.parentmid) - .bind(m.resolved) - .bind(&m.published) - .fetch_one(db) - .await?; - - use sqlx::Row; - Ok(res.get::(0)) + ); + Ok(on!(db, p => { + sqlx::query_scalar(sql) + .bind(&m.user_id) + .bind(&m.id) + .bind(&m.messageid) + .bind(&m.originid) + .bind(&m.stanzaid) + .bind(&m.replaceid) + .bind(&m.jidto) + .bind(&m.jidfrom) + .bind(&m.resource) + .bind(&m.type_) + .bind(&m.subject) + .bind(&m.thread) + .bind(&m.body) + .bind(&m.html) + .bind(m.quoted) + .bind(m.markable) + .bind(m.seen) + .bind(&m.delivered) + .bind(&m.displayed) + .bind(m.retracted) + .bind(m.encrypted) + .bind(&m.omemoheader) + .bind(m.bundleid) + .bind(m.parentmid) + .bind(m.resolved) + .bind(&m.published) + .fetch_one(p) + .await? + })) } /// Messages of a conversation, newest first @@ -684,118 +917,121 @@ pub async fn messages_for( sql.push_str(if after.is_some() { " ORDER BY published ASC, mid ASC" } else { " ORDER BY published DESC, mid DESC" }); sql.push_str(" LIMIT ?"); - let mut q = sqlx::query_as::<_, Message>(AssertSqlSafe(sql)).bind(user).bind(jid).bind(jid); - if let Some(b) = before { - q = q.bind(b); - } - if let Some(a) = after { - q = q.bind(a); - } - Ok(q.bind(limit).fetch_all(db).await?) + let sql = db.sql(&sql); + Ok(on!(db, p => { + let mut q = sqlx::query_as::<_, Message>(sql).bind(user).bind(jid).bind(jid); + if let Some(b) = before { + q = q.bind(b); + } + if let Some(a) = after { + q = q.bind(a); + } + q.bind(limit).fetch_all(p).await? + })) } pub async fn messages_count(db: &Db, user: &str, jid: &str) -> Result { - Ok(sqlx::query_scalar("SELECT COUNT(*) FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?)") - .bind(user) - .bind(jid) - .bind(jid) - .fetch_one(db) - .await?) + let sql = db.sql("SELECT COUNT(*) FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?)"); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).bind(jid).bind(jid).fetch_one(p).await?)) } /// Clear the local history of a conversation. The archive is marked as fully /// known and the clear date is kept, so the cleared messages are not fetched /// again from MAM or MUC history (`archive_key`: jid, or `muc:` + room). pub async fn messages_clear(db: &Db, user: &str, jid: &str, archive_key: &str) -> Result<()> { - let mut tx = db.begin().await?; - for sql in [ - "DELETE FROM reactions WHERE message_mid IN (SELECT mid FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?))", - "DELETE FROM message_files WHERE message_mid IN (SELECT mid FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?))", - "DELETE FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?)", - ] { - sqlx::query(sql).bind(user).bind(jid).bind(jid).execute(&mut *tx).await?; - } - sqlx::query("INSERT OR REPLACE INTO mam_earliests (user_id, jid, earliest) VALUES (?, ?, NULL)") + on!(db, p => { + let mut tx = p.begin().await?; + for sql in [ + "DELETE FROM reactions WHERE message_mid IN (SELECT mid FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?))", + "DELETE FROM message_files WHERE message_mid IN (SELECT mid FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?))", + "DELETE FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?)", + ] { + sqlx::query(db.sql(sql)).bind(user).bind(jid).bind(jid).execute(&mut *tx).await?; + } + sqlx::query(db.sql( + "INSERT INTO mam_earliests (user_id, jid, earliest) VALUES (?, ?, NULL) + ON CONFLICT(user_id, jid) DO UPDATE SET earliest = excluded.earliest", + )) .bind(user) .bind(archive_key) .execute(&mut *tx) .await?; - sqlx::query("INSERT OR REPLACE INTO history_clears (user_id, jid, cleared_at) VALUES (?, ?, ?)") + sqlx::query(db.sql( + "INSERT INTO history_clears (user_id, jid, cleared_at) VALUES (?, ?, ?) + ON CONFLICT(user_id, jid) DO UPDATE SET cleared_at = excluded.cleared_at", + )) .bind(user) .bind(archive_key) .bind(crate::utils::now_sql()) .execute(&mut *tx) .await?; - tx.commit().await?; + tx.commit().await?; + }); Ok(()) } /// Undo `messages_clear`: the cleared messages can be fetched again from MAM pub async fn history_restore(db: &Db, user: &str, archive_key: &str) -> Result<()> { - let mut tx = db.begin().await?; - for sql in [ - "DELETE FROM history_clears WHERE user_id = ? AND jid = ?", - "DELETE FROM mam_earliests WHERE user_id = ? AND jid = ?", - ] { - sqlx::query(sql).bind(user).bind(archive_key).execute(&mut *tx).await?; - } - tx.commit().await?; + on!(db, p => { + let mut tx = p.begin().await?; + for sql in [ + "DELETE FROM history_clears WHERE user_id = ? AND jid = ?", + "DELETE FROM mam_earliests WHERE user_id = ? AND jid = ?", + ] { + sqlx::query(db.sql(sql)).bind(user).bind(archive_key).execute(&mut *tx).await?; + } + tx.commit().await?; + }); Ok(()) } /// Delete the messages published more than `days` ago, with their reactions and /// files. Returns the number of deleted messages. pub async fn messages_expire(db: &Db, days: u32) -> Result { - let cutoff = (chrono::Utc::now() - chrono::Duration::days(days.into())).format("%Y-%m-%d %H:%M:%S").to_string(); - let mut tx = db.begin().await?; - for sql in [ - "DELETE FROM reactions WHERE message_mid IN (SELECT mid FROM messages WHERE published < ?)", - "DELETE FROM message_files WHERE message_mid IN (SELECT mid FROM messages WHERE published < ?)", - ] { - sqlx::query(sql).bind(&cutoff).execute(&mut *tx).await?; - } - let deleted = sqlx::query("DELETE FROM messages WHERE published < ?").bind(&cutoff).execute(&mut *tx).await?.rows_affected(); - tx.commit().await?; - Ok(deleted) + let cutoff = days_ago_sql(days.into()); + Ok(on!(db, p => { + let mut tx = p.begin().await?; + for sql in [ + "DELETE FROM reactions WHERE message_mid IN (SELECT mid FROM messages WHERE published < ?)", + "DELETE FROM message_files WHERE message_mid IN (SELECT mid FROM messages WHERE published < ?)", + ] { + sqlx::query(db.sql(sql)).bind(&cutoff).execute(&mut *tx).await?; + } + let deleted = sqlx::query(db.sql("DELETE FROM messages WHERE published < ?")) + .bind(&cutoff) + .execute(&mut *tx) + .await? + .rows_affected(); + tx.commit().await?; + deleted + })) } /// When the history of a conversation was last cleared (`archive_key` as above) pub async fn history_cleared_at(db: &Db, user: &str, archive_key: &str) -> Result> { - Ok(sqlx::query_scalar("SELECT cleared_at FROM history_clears WHERE user_id = ? AND jid = ?") - .bind(user) - .bind(archive_key) - .fetch_optional(db) - .await?) + let sql = db.sql("SELECT cleared_at FROM history_clears WHERE user_id = ? AND jid = ?"); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).bind(archive_key).fetch_optional(p).await?)) } /// Latest clear of a 1:1 conversation pub async fn latest_chat_clear(db: &Db, user: &str) -> Result> { - Ok(sqlx::query_scalar("SELECT MAX(cleared_at) FROM history_clears WHERE user_id = ? AND jid NOT LIKE 'muc:%'") - .bind(user) - .fetch_one(db) - .await?) + let sql = db.sql("SELECT MAX(cleared_at) FROM history_clears WHERE user_id = ? AND jid NOT LIKE 'muc:%'"); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).fetch_one(p).await?)) } pub async fn unreads_count(db: &Db, user: &str, jid: &str, muc: bool) -> Result { let types = if muc { MESSAGE_TYPE_MUC } else { MESSAGE_TYPE }; let types_sql = types.iter().map(|t| format!("'{t}'")).collect::>().join(","); - Ok(sqlx::query_scalar(AssertSqlSafe(format!( - "SELECT COUNT(*) FROM messages WHERE user_id = ? AND jidfrom = ? AND seen = 0 AND type IN ({types_sql})" - ))) - .bind(user) - .bind(jid) - .fetch_one(db) - .await?) + let sql = db.sql(&format!( + "SELECT COUNT(*) FROM messages WHERE user_id = ? AND jidfrom = ? AND seen = FALSE AND type IN ({types_sql})" + )); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).bind(jid).fetch_one(p).await?)) } /// Number of conversations (1:1 and rooms in `rooms`) with unread messages pub async fn unreads_total(db: &Db, user: &str, rooms: &[String]) -> Result { - let rows: Vec<(String, String)> = sqlx::query_as( - "SELECT DISTINCT jidfrom, type FROM messages WHERE user_id = ? AND seen = 0 AND jidfrom != user_id", - ) - .bind(user) - .fetch_all(db) - .await?; + let sql = db.sql("SELECT DISTINCT jidfrom, type FROM messages WHERE user_id = ? AND seen = FALSE AND jidfrom != user_id"); + let rows: Vec<(String, String)> = on!(db, p => sqlx::query_as(sql).bind(user).fetch_all(p).await?); Ok(rows .iter() .filter(|(jid, t)| if t == "groupchat" { rooms.contains(jid) } else { MESSAGE_TYPE.contains(&t.as_str()) }) @@ -803,35 +1039,28 @@ pub async fn unreads_total(db: &Db, user: &str, rooms: &[String]) -> Result } pub async fn messages_mark_seen(db: &Db, user: &str, jid: &str) -> Result { - Ok(sqlx::query("UPDATE messages SET seen = 1 WHERE user_id = ? AND jidfrom = ? AND seen = 0") - .bind(user) - .bind(jid) - .execute(db) - .await? - .rows_affected()) + let sql = db.sql("UPDATE messages SET seen = TRUE WHERE user_id = ? AND jidfrom = ? AND seen = FALSE"); + Ok(on!(db, p => sqlx::query(sql).bind(user).bind(jid).execute(p).await?.rows_affected())) } pub async fn message_set_field(db: &Db, mid: i64, field: &str, value: Option<&str>) -> Result<()> { anyhow::ensure!(matches!(field, "delivered" | "displayed" | "body"), "field not allowed"); - sqlx::query(AssertSqlSafe(format!("UPDATE messages SET {field} = ?, updated_at = CURRENT_TIMESTAMP WHERE mid = ?"))) - .bind(value) - .bind(mid) - .execute(db) - .await?; + let sql = db.sql(&format!("UPDATE messages SET {field} = ?, updated_at = CURRENT_TIMESTAMP WHERE mid = ?")); + on!(db, p => sqlx::query(sql).bind(value).bind(mid).execute(p).await?;); Ok(()) } pub async fn message_retract(db: &Db, mid: i64) -> Result<()> { - sqlx::query("UPDATE messages SET retracted = 1, body = NULL, html = NULL, updated_at = CURRENT_TIMESTAMP WHERE mid = ?") - .bind(mid) - .execute(db) - .await?; - sqlx::query("DELETE FROM message_files WHERE message_mid = ?").bind(mid).execute(db).await?; + let sql = db.sql("UPDATE messages SET retracted = TRUE, body = NULL, html = NULL, updated_at = CURRENT_TIMESTAMP WHERE mid = ?"); + on!(db, p => sqlx::query(sql).bind(mid).execute(p).await?;); + let sql = db.sql("DELETE FROM message_files WHERE message_mid = ?"); + on!(db, p => sqlx::query(sql).bind(mid).execute(p).await?;); Ok(()) } pub async fn message_set_resolved(db: &Db, mid: i64) -> Result<()> { - sqlx::query("UPDATE messages SET resolved = 1 WHERE mid = ?").bind(mid).execute(db).await?; + let sql = db.sql("UPDATE messages SET resolved = TRUE WHERE mid = ?"); + on!(db, p => sqlx::query(sql).bind(mid).execute(p).await?;); Ok(()) } @@ -842,64 +1071,69 @@ pub async fn last_message(db: &Db, user: &str, jid: &str, muc: bool) -> Result Result> { - Ok(sqlx::query_scalar( + let sql = db.sql( "SELECT CASE WHEN jidfrom = user_id THEN jidto ELSE jidfrom END AS jid FROM messages WHERE user_id = ? AND type = 'chat' GROUP BY jid ORDER BY MAX(published) DESC LIMIT ?", - ) - .bind(user) - .bind(limit) - .fetch_all(db) - .await?) + ); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).bind(limit).fetch_all(p).await?)) } // Resolved URLs (link previews) pub async fn url_get(db: &Db, hash: &str) -> Result> { - Ok(sqlx::query_scalar("SELECT data FROM urls WHERE hash = ?").bind(hash).fetch_optional(db).await?) + let sql = db.sql("SELECT data FROM urls WHERE hash = ?"); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(hash).fetch_optional(p).await?)) } pub async fn url_save(db: &Db, hash: &str, url: &str, data: &str) -> Result<()> { - sqlx::query("INSERT OR REPLACE INTO urls (hash, url, data) VALUES (?, ?, ?)").bind(hash).bind(url).bind(data).execute(db).await?; + let sql = db.sql( + "INSERT INTO urls (hash, url, data) VALUES (?, ?, ?) + ON CONFLICT(hash) DO UPDATE SET url = excluded.url, data = excluded.data, created_at = excluded.created_at", + ); + on!(db, p => sqlx::query(sql).bind(hash).bind(url).bind(data).execute(p).await?;); Ok(()) } /// Forget the previews older than a month, the pages may have changed pub async fn urls_cleanup(db: &Db) -> Result<()> { - sqlx::query("DELETE FROM urls WHERE created_at < datetime('now', '-30 days')").execute(db).await?; + let sql = db.sql("DELETE FROM urls WHERE created_at < ?"); + on!(db, p => sqlx::query(sql).bind(days_ago_sql(30)).execute(p).await?;); Ok(()) } // Files pub async fn message_file_save(db: &Db, f: &MessageFile) -> Result<()> { - sqlx::query("DELETE FROM message_files WHERE message_mid = ?").bind(f.message_mid).execute(db).await?; - sqlx::query( + let sql = db.sql("DELETE FROM message_files WHERE message_mid = ?"); + on!(db, p => sqlx::query(sql).bind(f.message_mid).execute(p).await?;); + let sql = db.sql( "INSERT INTO message_files (message_mid, name, type, size, url, disposition, thumbnail_url, thumbnail_type, thumbnail_width, thumbnail_height) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - ) - .bind(f.message_mid) - .bind(&f.name) - .bind(&f.type_) - .bind(f.size) - .bind(&f.url) - .bind(&f.disposition) - .bind(&f.thumbnail_url) - .bind(&f.thumbnail_type) - .bind(f.thumbnail_width) - .bind(f.thumbnail_height) - .execute(db) - .await?; + ); + on!(db, p => { + sqlx::query(sql) + .bind(f.message_mid) + .bind(&f.name) + .bind(&f.type_) + .bind(f.size) + .bind(&f.url) + .bind(&f.disposition) + .bind(&f.thumbnail_url) + .bind(&f.thumbnail_type) + .bind(f.thumbnail_width) + .bind(f.thumbnail_height) + .execute(p) + .await?; + }); Ok(()) } pub async fn message_file(db: &Db, mid: i64) -> Result> { - Ok(sqlx::query_as::<_, MessageFile>( + let sql = db.sql( "SELECT id, message_mid, name, type, size, url, disposition, thumbnail_url, thumbnail_type, thumbnail_width, thumbnail_height FROM message_files WHERE message_mid = ? LIMIT 1", - ) - .bind(mid) - .fetch_optional(db) - .await?) + ); + Ok(on!(db, p => sqlx::query_as::<_, MessageFile>(sql).bind(mid).fetch_optional(p).await?)) } // Reactions @@ -909,117 +1143,101 @@ pub async fn reactions_for(db: &Db, mids: &[i64]) -> Result> { return Ok(vec![]); } let placeholders = vec!["?"; mids.len()].join(","); - let sql = format!( + let sql = db.sql(&format!( "SELECT message_mid, jidfrom, emoji, created_at FROM reactions WHERE message_mid IN ({placeholders}) ORDER BY created_at, id" - ); - let mut q = sqlx::query_as::<_, Reaction>(AssertSqlSafe(sql)); - for mid in mids { - q = q.bind(mid); - } - Ok(q.fetch_all(db).await?) + )); + Ok(on!(db, p => { + let mut q = sqlx::query_as::<_, Reaction>(sql); + for mid in mids { + q = q.bind(mid); + } + q.fetch_all(p).await? + })) } pub async fn reactions_replace(db: &Db, mid: i64, jidfrom: &str, emojis: &[String]) -> Result<()> { - let mut tx = db.begin().await?; - sqlx::query("DELETE FROM reactions WHERE message_mid = ? AND jidfrom = ?").bind(mid).bind(jidfrom).execute(&mut *tx).await?; - for emoji in emojis { - sqlx::query("INSERT OR IGNORE INTO reactions (message_mid, jidfrom, emoji) VALUES (?, ?, ?)") + on!(db, p => { + let mut tx = p.begin().await?; + sqlx::query(db.sql("DELETE FROM reactions WHERE message_mid = ? AND jidfrom = ?")) .bind(mid) .bind(jidfrom) - .bind(emoji) .execute(&mut *tx) .await?; - } - tx.commit().await?; + for emoji in emojis { + sqlx::query(db.sql("INSERT INTO reactions (message_mid, jidfrom, emoji) VALUES (?, ?, ?) ON CONFLICT DO NOTHING")) + .bind(mid) + .bind(jidfrom) + .bind(emoji) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + }); Ok(()) } // MAM pub async fn mam_earliest_exists(db: &Db, user: &str, jid: &str) -> Result { - let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM mam_earliests WHERE user_id = ? AND jid = ?") - .bind(user) - .bind(jid) - .fetch_one(db) - .await?; + let sql = db.sql("SELECT COUNT(*) FROM mam_earliests WHERE user_id = ? AND jid = ?"); + let n: i64 = on!(db, p => sqlx::query_scalar(sql).bind(user).bind(jid).fetch_one(p).await?); Ok(n > 0) } pub async fn mam_earliest_set(db: &Db, user: &str, jid: &str, earliest: Option<&str>) -> Result<()> { - sqlx::query("INSERT OR REPLACE INTO mam_earliests (user_id, jid, earliest) VALUES (?, ?, ?)") - .bind(user) - .bind(jid) - .bind(earliest) - .execute(db) - .await?; + let sql = db.sql( + "INSERT INTO mam_earliests (user_id, jid, earliest) VALUES (?, ?, ?) + ON CONFLICT(user_id, jid) DO UPDATE SET earliest = excluded.earliest", + ); + on!(db, p => sqlx::query(sql).bind(user).bind(jid).bind(earliest).execute(p).await?;); Ok(()) } pub async fn earliest_message_date(db: &Db, user: &str, jid: &str) -> Result> { - Ok(sqlx::query_scalar("SELECT MIN(published) FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?)") - .bind(user) - .bind(jid) - .bind(jid) - .fetch_one(db) - .await?) + let sql = db.sql("SELECT MIN(published) FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?)"); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).bind(jid).bind(jid).fetch_one(p).await?)) } pub async fn latest_message_date(db: &Db, user: &str) -> Result> { - Ok(sqlx::query_scalar("SELECT MAX(published) FROM messages WHERE user_id = ? AND type = 'chat'") - .bind(user) - .fetch_one(db) - .await?) + let sql = db.sql("SELECT MAX(published) FROM messages WHERE user_id = ? AND type = 'chat'"); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).fetch_one(p).await?)) } pub async fn latest_muc_message_date(db: &Db, user: &str, room: &str) -> Result> { - Ok(sqlx::query_scalar("SELECT MAX(published) FROM messages WHERE user_id = ? AND jidfrom = ? AND type = 'groupchat'") - .bind(user) - .bind(room) - .fetch_one(db) - .await?) + let sql = db.sql("SELECT MAX(published) FROM messages WHERE user_id = ? AND jidfrom = ? AND type = 'groupchat'"); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).bind(room).fetch_one(p).await?)) } // OMEMO per-conversation state pub async fn omemo_state(db: &Db, user: &str, jid: &str) -> Result> { - Ok(sqlx::query_scalar("SELECT enabled FROM omemo_states WHERE user_id = ? AND jid = ?") - .bind(user) - .bind(jid) - .fetch_optional(db) - .await?) + let sql = db.sql("SELECT enabled FROM omemo_states WHERE user_id = ? AND jid = ?"); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).bind(jid).fetch_optional(p).await?)) } pub async fn omemo_state_set(db: &Db, user: &str, jid: &str, enabled: bool) -> Result<()> { - sqlx::query("INSERT OR REPLACE INTO omemo_states (user_id, jid, enabled) VALUES (?, ?, ?)") - .bind(user) - .bind(jid) - .bind(enabled) - .execute(db) - .await?; + let sql = db.sql( + "INSERT INTO omemo_states (user_id, jid, enabled) VALUES (?, ?, ?) + ON CONFLICT(user_id, jid) DO UPDATE SET enabled = excluded.enabled", + ); + on!(db, p => sqlx::query(sql).bind(user).bind(jid).bind(enabled).execute(p).await?;); Ok(()) } // Push subscriptions pub async fn push_subscription_save(db: &Db, user: &str, endpoint: &str, p256dh: &str, auth: &str) -> Result<()> { - sqlx::query( + let sql = db.sql( "INSERT INTO push_subscriptions (endpoint, user_id, p256dh, auth, activity_at) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(endpoint) DO UPDATE SET user_id = excluded.user_id, p256dh = excluded.p256dh, auth = excluded.auth, activity_at = CURRENT_TIMESTAMP", - ) - .bind(endpoint) - .bind(user) - .bind(p256dh) - .bind(auth) - .execute(db) - .await?; + ); + on!(db, p => sqlx::query(sql).bind(endpoint).bind(user).bind(p256dh).bind(auth).execute(p).await?;); Ok(()) } pub async fn push_subscription_touch(db: &Db, endpoint: &str) -> Result<()> { - sqlx::query("UPDATE push_subscriptions SET activity_at = CURRENT_TIMESTAMP WHERE endpoint = ?") - .bind(endpoint) - .execute(db) - .await?; + let sql = db.sql("UPDATE push_subscriptions SET activity_at = CURRENT_TIMESTAMP WHERE endpoint = ?"); + on!(db, p => sqlx::query(sql).bind(endpoint).execute(p).await?;); Ok(()) } @@ -1027,51 +1245,42 @@ pub async fn push_subscription_touch(db: &Db, endpoint: &str) -> Result<()> { pub async fn open_chat_touch(db: &Db, user: &str, jid: &str, at: Option<&str>) -> Result<()> { let at = at.map(String::from).unwrap_or_else(crate::utils::now_sql_micro); - sqlx::query( + let sql = db.sql( "INSERT INTO open_chats (user_id, jid, updated_at) VALUES (?, ?, ?) - ON CONFLICT(user_id, jid) DO UPDATE SET updated_at = MAX(updated_at, excluded.updated_at)", - ) - .bind(user) - .bind(jid) - .bind(at) - .execute(db) - .await?; + ON CONFLICT(user_id, jid) DO UPDATE SET updated_at = CASE WHEN excluded.updated_at > open_chats.updated_at + THEN excluded.updated_at ELSE open_chats.updated_at END", + ); + on!(db, p => sqlx::query(sql).bind(user).bind(jid).bind(&at).execute(p).await?;); Ok(()) } pub async fn open_chat_exists(db: &Db, user: &str, jid: &str) -> Result { - let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM open_chats WHERE user_id = ? AND jid = ?") - .bind(user) - .bind(jid) - .fetch_one(db) - .await?; + let sql = db.sql("SELECT COUNT(*) FROM open_chats WHERE user_id = ? AND jid = ?"); + let n: i64 = on!(db, p => sqlx::query_scalar(sql).bind(user).bind(jid).fetch_one(p).await?); Ok(n > 0) } pub async fn open_chat_close(db: &Db, user: &str, jid: &str) -> Result<()> { - sqlx::query("DELETE FROM open_chats WHERE user_id = ? AND jid = ?").bind(user).bind(jid).execute(db).await?; + let sql = db.sql("DELETE FROM open_chats WHERE user_id = ? AND jid = ?"); + on!(db, p => sqlx::query(sql).bind(user).bind(jid).execute(p).await?;); Ok(()) } /// Open chats, oldest first. Conversations with unread messages are opened automatically. pub async fn open_chats(db: &Db, user: &str) -> Result> { let types_sql = MESSAGE_TYPE.iter().map(|t| format!("'{t}'")).collect::>().join(","); - let unread: Vec<(String, String)> = sqlx::query_as(AssertSqlSafe(format!( - "SELECT jidfrom, MAX(published) FROM messages WHERE user_id = ? AND seen = 0 AND jidfrom != user_id + let sql = db.sql(&format!( + "SELECT jidfrom, MAX(published) FROM messages WHERE user_id = ? AND seen = FALSE AND jidfrom != user_id AND type IN ({types_sql}) GROUP BY jidfrom" - ))) - .bind(user) - .fetch_all(db) - .await?; + )); + let unread: Vec<(String, String)> = on!(db, p => sqlx::query_as(sql).bind(user).fetch_all(p).await?); for (jid, published) in unread { if !open_chat_exists(db, user, &jid).await? { open_chat_touch(db, user, &jid, Some(&published)).await?; } } - Ok(sqlx::query_scalar("SELECT jid FROM open_chats WHERE user_id = ? ORDER BY updated_at ASC") - .bind(user) - .fetch_all(db) - .await?) + let sql = db.sql("SELECT jid FROM open_chats WHERE user_id = ? ORDER BY updated_at ASC"); + Ok(on!(db, p => sqlx::query_scalar(sql).bind(user).fetch_all(p).await?)) } /// Plain text search in a discussion @@ -1095,15 +1304,11 @@ pub async fn messages_search( let types_sql = conversation_types(muc); let words: Vec = crate::utils::search_normalize(keywords).split_whitespace().take(8).map(String::from).collect(); - let rows: Vec<(i64, String)> = sqlx::query_as(AssertSqlSafe(format!( + let sql = db.sql(&format!( "SELECT mid, body FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?) AND type IN ({types_sql}) - AND encrypted = 0 AND retracted = 0 AND body IS NOT NULL AND {NOT_REPLACED} ORDER BY published DESC" - ))) - .bind(user) - .bind(jid) - .bind(jid) - .fetch_all(db) - .await?; + AND encrypted = FALSE AND retracted = FALSE AND body IS NOT NULL AND {NOT_REPLACED} ORDER BY published DESC" + )); + let rows: Vec<(i64, String)> = on!(db, p => sqlx::query_as(sql).bind(user).bind(jid).bind(jid).fetch_all(p).await?); let mut mids: Vec = rows .into_iter() .filter(|(_, body)| { @@ -1119,17 +1324,17 @@ pub async fn messages_search( } let placeholders = vec!["?"; mids.len()].join(","); - let mut q = sqlx::query_as::<_, Message>(AssertSqlSafe(format!( + let sql = db.sql(&format!( "SELECT {MESSAGE_COLUMNS} FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?) AND type IN ({types_sql}) - AND retracted = 0 AND mid IN ({placeholders}) ORDER BY published DESC LIMIT ?" - ))) - .bind(user) - .bind(jid) - .bind(jid); - for mid in &mids { - q = q.bind(mid); - } - Ok(q.bind(limit).fetch_all(db).await?) + AND retracted = FALSE AND mid IN ({placeholders}) ORDER BY published DESC LIMIT ?" + )); + Ok(on!(db, p => { + let mut q = sqlx::query_as::<_, Message>(sql).bind(user).bind(jid).bind(jid); + for mid in &mids { + q = q.bind(mid); + } + q.bind(limit).fetch_all(p).await? + })) } /// Skips the messages that were corrected by their author (XEP-0308): only the last version is searched @@ -1137,21 +1342,17 @@ pub async fn messages_search( const NOT_REPLACED: &str = "NOT EXISTS (SELECT 1 FROM messages r WHERE r.user_id = messages.user_id AND r.replaceid IS NOT NULL AND (r.replaceid IN (messages.id, messages.messageid, messages.originid) OR r.replaceid = messages.replaceid AND r.mid > messages.mid) - AND r.jidfrom = messages.jidfrom AND (r.type = 'chat' OR r.resource IS messages.resource))"; + AND r.jidfrom = messages.jidfrom AND (r.type = 'chat' OR r.resource IS NOT DISTINCT FROM messages.resource))"; /// OMEMO messages of a conversation, newest first: (mid, id, messageid, published). /// The browser looks their decrypted bodies up with id or messageid. pub async fn encrypted_index(db: &Db, user: &str, jid: &str, muc: bool) -> Result, String)>> { let types_sql = conversation_types(muc); - Ok(sqlx::query_as(AssertSqlSafe(format!( + let sql = db.sql(&format!( "SELECT mid, id, messageid, published FROM messages WHERE user_id = ? AND (jidfrom = ? OR jidto = ?) - AND type IN ({types_sql}) AND encrypted = 1 AND retracted = 0 AND {NOT_REPLACED} ORDER BY published DESC LIMIT 5000" - ))) - .bind(user) - .bind(jid) - .bind(jid) - .fetch_all(db) - .await?) + AND type IN ({types_sql}) AND encrypted = TRUE AND retracted = FALSE AND {NOT_REPLACED} ORDER BY published DESC LIMIT 5000" + )); + Ok(on!(db, p => sqlx::query_as(sql).bind(user).bind(jid).bind(jid).fetch_all(p).await?)) } #[derive(Debug, Clone, FromRow, Serialize)] @@ -1169,51 +1370,47 @@ pub struct ConversationFile { /// Files shared in a conversation (clear text ones), newest first pub async fn conversation_files(db: &Db, user: &str, jid: &str, muc: bool, limit: i64) -> Result> { let types_sql = conversation_types(muc); - Ok(sqlx::query_as(AssertSqlSafe(format!( + let sql = db.sql(&format!( "SELECT m.mid, f.name, f.type, f.size, f.url, m.published FROM message_files f JOIN messages m ON m.mid = f.message_mid - WHERE m.user_id = ? AND (m.jidfrom = ? OR m.jidto = ?) AND m.type IN ({types_sql}) AND m.retracted = 0 + WHERE m.user_id = ? AND (m.jidfrom = ? OR m.jidto = ?) AND m.type IN ({types_sql}) AND m.retracted = FALSE ORDER BY m.published DESC LIMIT ?" - ))) - .bind(user) - .bind(jid) - .bind(jid) - .bind(limit) - .fetch_all(db) - .await?) + )); + Ok(on!(db, p => sqlx::query_as(sql).bind(user).bind(jid).bind(jid).bind(limit).fetch_all(p).await?)) } /// Forget everything stored for an account pub async fn user_delete_data(db: &Db, user: &str) -> Result<()> { - let mut tx = db.begin().await?; - for sql in [ - "DELETE FROM reactions WHERE message_mid IN (SELECT mid FROM messages WHERE user_id = ?)", - "DELETE FROM message_files WHERE message_mid IN (SELECT mid FROM messages WHERE user_id = ?)", - "DELETE FROM messages WHERE user_id = ?", - "DELETE FROM open_chats WHERE user_id = ?", - "DELETE FROM mam_earliests WHERE user_id = ?", - "DELETE FROM history_clears WHERE user_id = ?", - "DELETE FROM omemo_states WHERE user_id = ?", - "DELETE FROM encrypted_passwords WHERE user_id = ?", - "DELETE FROM push_subscriptions WHERE user_id = ?", - "DELETE FROM users WHERE id = ?", - ] { - sqlx::query(sql).bind(user).execute(&mut *tx).await?; - } - tx.commit().await?; + on!(db, p => { + let mut tx = p.begin().await?; + for sql in [ + "DELETE FROM reactions WHERE message_mid IN (SELECT mid FROM messages WHERE user_id = ?)", + "DELETE FROM message_files WHERE message_mid IN (SELECT mid FROM messages WHERE user_id = ?)", + "DELETE FROM messages WHERE user_id = ?", + "DELETE FROM open_chats WHERE user_id = ?", + "DELETE FROM mam_earliests WHERE user_id = ?", + "DELETE FROM history_clears WHERE user_id = ?", + "DELETE FROM omemo_states WHERE user_id = ?", + "DELETE FROM encrypted_passwords WHERE user_id = ?", + "DELETE FROM push_subscriptions WHERE user_id = ?", + "DELETE FROM users WHERE id = ?", + ] { + sqlx::query(db.sql(sql)).bind(user).execute(&mut *tx).await?; + } + tx.commit().await?; + }); Ok(()) } pub async fn push_subscriptions(db: &Db, user: &str) -> Result> { - Ok(sqlx::query_as( - "SELECT endpoint, COALESCE(p256dh, ''), COALESCE(auth, '') FROM push_subscriptions WHERE user_id = ? AND enabled = 1", - ) - .bind(user) - .fetch_all(db) - .await?) + let sql = db.sql( + "SELECT endpoint, COALESCE(p256dh, ''), COALESCE(auth, '') FROM push_subscriptions WHERE user_id = ? AND enabled = TRUE", + ); + Ok(on!(db, p => sqlx::query_as(sql).bind(user).fetch_all(p).await?)) } pub async fn push_subscription_delete(db: &Db, endpoint: &str) -> Result<()> { - sqlx::query("DELETE FROM push_subscriptions WHERE endpoint = ?").bind(endpoint).execute(db).await?; + let sql = db.sql("DELETE FROM push_subscriptions WHERE endpoint = ?"); + on!(db, p => sqlx::query(sql).bind(endpoint).execute(p).await?;); Ok(()) } @@ -1226,37 +1423,79 @@ pub struct PushSubscription { } pub async fn push_subscriptions_list(db: &Db, user: &str) -> Result> { - Ok(sqlx::query_as::<_, PushSubscription>( + let sql = db.sql( "SELECT rowid AS id, endpoint, enabled, activity_at FROM push_subscriptions WHERE user_id = ? ORDER BY activity_at DESC", - ) - .bind(user) - .fetch_all(db) - .await?) + ); + Ok(on!(db, p => sqlx::query_as::<_, PushSubscription>(sql).bind(user).fetch_all(p).await?)) } pub async fn push_subscription_toggle(db: &Db, user: &str, id: i64, enabled: bool) -> Result<()> { - sqlx::query("UPDATE push_subscriptions SET enabled = ? WHERE rowid = ? AND user_id = ?") - .bind(enabled) - .bind(id) - .bind(user) - .execute(db) - .await?; + let sql = db.sql("UPDATE push_subscriptions SET enabled = ? WHERE rowid = ? AND user_id = ?"); + on!(db, p => sqlx::query(sql).bind(enabled).bind(id).bind(user).execute(p).await?;); Ok(()) } #[cfg(test)] -mod tests { +pub mod tests { use super::*; fn days_ago(days: i64) -> String { (chrono::Utc::now() - chrono::Duration::days(days)).format("%Y-%m-%d %H:%M:%S").to_string() } + /// A SQLite file: every pool connection to sqlite::memory: gets its own database + pub async fn sqlite_test_db(name: &str) -> (Db, std::path::PathBuf) { + let path = std::env::temp_dir().join(format!("movim-test-{name}-{}.sqlite", std::process::id())); + remove_sqlite(&path); + (connect(&format!("sqlite://{}?mode=rwc", path.display())).await.unwrap(), path) + } + + pub fn remove_sqlite(path: &std::path::Path) { + for ext in ["", "-wal", "-shm"] { + let _ = std::fs::remove_file(format!("{}{ext}", path.display())); + } + } + + /// A new database on the server of MOVIM_TEST_POSTGRES_URL + /// (postgres://user:password@host/db), None when it is not set + pub async fn postgres_test_url(name: &str) -> Option { + let Ok(admin) = std::env::var("MOVIM_TEST_POSTGRES_URL") else { + eprintln!("MOVIM_TEST_POSTGRES_URL is not set, skipping the PostgreSQL test {name}"); + return None; + }; + let database = format!("movim_test_{name}_{}", std::process::id()); + let pool = PgPoolOptions::new().max_connections(1).connect(&admin).await.unwrap(); + sqlx::query(AssertSqlSafe(format!("DROP DATABASE IF EXISTS {database} WITH (FORCE)"))).execute(&pool).await.unwrap(); + sqlx::query(AssertSqlSafe(format!("CREATE DATABASE {database}"))).execute(&pool).await.unwrap(); + pool.close().await; + let base = admin.split('?').next().unwrap(); + Some(format!("{}/{database}", &base[..base.rfind('/').unwrap()])) + } + #[tokio::test] - async fn clear_history_is_remembered() { - // A file: every pool connection to sqlite::memory: gets its own database - let path = std::env::temp_dir().join(format!("movim-test-{}.sqlite", std::process::id())); - let db = connect(&format!("sqlite://{}?mode=rwc", path.display())).await.unwrap(); + async fn clear_history_is_remembered_sqlite() { + let (db, path) = sqlite_test_db("clear").await; + clear_history_is_remembered(&db).await; + db.close().await; + remove_sqlite(&path); + } + + #[tokio::test] + async fn clear_history_is_remembered_postgres() { + let Some(url) = postgres_test_url("clear").await else { return }; + let db = connect(&url).await.unwrap(); + clear_history_is_remembered(&db).await; + db.close().await; + } + + #[test] + fn postgres_placeholders() { + assert_eq!(to_postgres("SELECT a FROM t WHERE b = ? AND c LIKE 'x?%' AND d IN (?, ?)"), "SELECT a FROM t WHERE b = $1 AND c LIKE 'x?%' AND d IN ($2, $3)"); + assert!(to_postgres("UPDATE t SET u = CURRENT_TIMESTAMP").contains("to_char(now() AT TIME ZONE 'UTC'")); + } + + async fn clear_history_is_remembered(db: &Db) { + let db = db.clone(); let (me, bob) = ("me@example.org", "bob@example.org"); let m = Message { user_id: me.into(), @@ -1336,9 +1575,39 @@ mod tests { assert_eq!(messages_count(&db, me, bob).await.unwrap(), before - 1); assert!(reactions_for(&db, &[old_mid]).await.unwrap().is_empty()); - db.close().await; - for ext in ["", "-wal", "-shm"] { - let _ = std::fs::remove_file(format!("{}{ext}", path.display())); - } + // Upserts: a message seen once stays seen, open chats keep their latest date + let unseen = Message { id: "u1".into(), published: crate::utils::now_sql(), ..m.clone() }; + let unseen_mid = message_save(&db, &Message { seen: true, ..unseen.clone() }).await.unwrap(); + assert_eq!(message_save(&db, &unseen).await.unwrap(), unseen_mid); + assert!(message_by_mid(&db, me, unseen_mid).await.unwrap().unwrap().seen); + open_chat_touch(&db, me, bob, Some("2026-02-01 10:00:00")).await.unwrap(); + open_chat_touch(&db, me, bob, Some("2026-01-01 10:00:00")).await.unwrap(); + open_chat_touch(&db, me, "carol@example.org", Some("2026-01-15 10:00:00")).await.unwrap(); + // eve is opened too: her message is unread + assert_eq!(open_chats(&db, me).await.unwrap(), ["carol@example.org", bob, "eve@example.org"]); + + // Users, booleans and push subscriptions + let user = user_get_or_create(&db, me).await.unwrap(); + assert!(user.chatmain && !user.nightmode); + user_set(&db, me, "nightmode", &serde_json::json!(true)).await.unwrap(); + assert!(user_get(&db, me).await.unwrap().unwrap().nightmode); + omemo_state_set(&db, me, bob, false).await.unwrap(); + omemo_state_set(&db, me, bob, true).await.unwrap(); + assert_eq!(omemo_state(&db, me, bob).await.unwrap(), Some(true)); + push_subscription_save(&db, me, "https://push.example/1", "k", "a").await.unwrap(); + let subs = push_subscriptions_list(&db, me).await.unwrap(); + push_subscription_toggle(&db, me, subs[0].id, false).await.unwrap(); + assert!(push_subscriptions(&db, me).await.unwrap().is_empty()); + avatar_save(&db, "h", "image/png", &[0, 159, 255]).await.unwrap(); + avatar_save(&db, "h", "image/png", &[1, 2]).await.unwrap(); + assert_eq!(avatar_get(&db, "h").await.unwrap(), Some(("image/png".into(), vec![1, 2]))); + assert_eq!(top_contacts(&db, me, 5).await.unwrap(), [bob, "eve@example.org"]); + assert!(unreads_count(&db, me, bob, false).await.unwrap() > 0); + messages_mark_seen(&db, me, bob).await.unwrap(); + assert_eq!(unreads_count(&db, me, bob, false).await.unwrap(), 0); + + user_delete_data(&db, me).await.unwrap(); + assert_eq!(messages_count(&db, me, bob).await.unwrap(), 0); + assert!(user_get(&db, me).await.unwrap().is_none()); } } diff --git a/src/main.rs b/src/main.rs index 31802a1..d91c566 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,6 +9,7 @@ mod handler; mod i18n; mod jingle; mod mam; +mod migrate_postgres; mod page; mod pep; mod proto; @@ -38,6 +39,13 @@ async fn main() -> anyhow::Result<()> { let _ = tokio_xmpp::rustls::crypto::aws_lc_rs::default_provider().install_default(); let cfg = config::Config::from_env(); + + // movim migrate-to-postgres [--from ] + let args: Vec = std::env::args().skip(1).collect(); + if args.first().map(String::as_str) == Some("migrate-to-postgres") { + return migrate_to_postgres(&cfg, &args[1..]).await; + } + std::fs::create_dir_all(&cfg.data_dir)?; i18n::init(&cfg.frontend); @@ -102,3 +110,32 @@ async fn main() -> anyhow::Result<()> { axum::serve(listener, router).await?; Ok(()) } + +/// Copy the SQLite database (MOVIM_DATABASE_URL, the data dir one by default, or +/// `--from`) to an empty PostgreSQL database. Movim must be stopped meanwhile. +async fn migrate_to_postgres(cfg: &config::Config, args: &[String]) -> anyhow::Result<()> { + let mut to = None; + let mut from = None; + let mut args = args.iter(); + while let Some(arg) = args.next() { + match arg.as_str() { + "--from" => from = args.next().cloned(), + _ if to.is_none() => to = Some(arg.clone()), + _ => anyhow::bail!("unexpected argument {arg}"), + } + } + let usage = "usage: movim migrate-to-postgres postgres://user:password@host/database [--from sqlite:///path/movim.sqlite]"; + let to = to.ok_or_else(|| anyhow::anyhow!(usage))?; + let from = from.unwrap_or_else(|| cfg.database_url.clone()); + anyhow::ensure!(!db::is_postgres_url(&from), "{usage}\nMOVIM_DATABASE_URL is already PostgreSQL, give the SQLite one with --from"); + if let Some(path) = from.strip_prefix("sqlite://").map(|p| p.split('?').next().unwrap_or(p)) { + anyhow::ensure!(std::path::Path::new(path).exists(), "{path} does not exist"); + } + + tracing::info!("copying {from} to PostgreSQL"); + for (table, rows) in migrate_postgres::sqlite_to_postgres(&from, &to).await? { + println!("{table:>20}: {rows} rows"); + } + println!("Done. Start Movim with MOVIM_DATABASE_URL set to the PostgreSQL URL"); + Ok(()) +} diff --git a/src/migrate_postgres.rs b/src/migrate_postgres.rs new file mode 100644 index 0000000..365ec2e --- /dev/null +++ b/src/migrate_postgres.rs @@ -0,0 +1,327 @@ +//! Copy a SQLite database to PostgreSQL: `movim migrate-to-postgres`. +//! +//! The target must be empty (its tables are created if needed). Everything is +//! copied in one transaction, the ids are kept (replies, reactions and files +//! point to the messages by mid) and the sequences continue after them. + +use anyhow::{Context, Result, bail}; +use sqlx::sqlite::SqliteRow; +use sqlx::{AssertSqlSafe, PgPool, Postgres, QueryBuilder, Row, SqlitePool}; + +use crate::db::{self, Db}; + +#[derive(Clone, Copy)] +enum Kind { + Text, + Int, + Bool, + Blob, +} + +use Kind::*; + +/// The columns of `db::SCHEMA`. push_subscriptions copies the implicit SQLite rowid. +const TABLES: &[(&str, &[(&str, Kind)])] = &[ + ( + "users", + &[ + ("id", Text), + ("nickname", Text), + ("language", Text), + ("nightmode", Bool), + ("chatmain", Bool), + ("notificationchat", Bool), + ("notificationcall", Bool), + ("omemoenabled", Bool), + ("accentcolor", Text), + ("admin", Bool), + ("chats", Text), + ("created_at", Text), + ("updated_at", Text), + ], + ), + ("encrypted_passwords", &[("id", Text), ("user_id", Text), ("data", Text), ("updated_at", Text)]), + ( + "contacts", + &[ + ("id", Text), + ("fn", Text), + ("name", Text), + ("nickname", Text), + ("email", Text), + ("url", Text), + ("description", Text), + ("avatarhash", Text), + ("avatartype", Text), + ("updated_at", Text), + ], + ), + ("avatars", &[("hash", Text), ("mime", Text), ("data", Blob)]), + ( + "messages", + &[ + ("mid", Int), + ("user_id", Text), + ("id", Text), + ("messageid", Text), + ("originid", Text), + ("stanzaid", Text), + ("replaceid", Text), + ("jidto", Text), + ("jidfrom", Text), + ("resource", Text), + ("type", Text), + ("subject", Text), + ("thread", Text), + ("body", Text), + ("html", Text), + ("quoted", Bool), + ("markable", Bool), + ("seen", Bool), + ("delivered", Text), + ("displayed", Text), + ("retracted", Bool), + ("encrypted", Bool), + ("omemoheader", Text), + ("bundleid", Int), + ("parentmid", Int), + ("resolved", Bool), + ("published", Text), + ("created_at", Text), + ("updated_at", Text), + ], + ), + ( + "message_files", + &[ + ("id", Int), + ("message_mid", Int), + ("name", Text), + ("type", Text), + ("size", Int), + ("url", Text), + ("disposition", Text), + ("thumbnail_url", Text), + ("thumbnail_type", Text), + ("thumbnail_width", Int), + ("thumbnail_height", Int), + ], + ), + ("reactions", &[("id", Int), ("message_mid", Int), ("jidfrom", Text), ("emoji", Text), ("created_at", Text)]), + ("mam_earliests", &[("user_id", Text), ("jid", Text), ("earliest", Text)]), + ("history_clears", &[("user_id", Text), ("jid", Text), ("cleared_at", Text)]), + ("omemo_states", &[("user_id", Text), ("jid", Text), ("enabled", Bool)]), + ("open_chats", &[("user_id", Text), ("jid", Text), ("updated_at", Text)]), + ("urls", &[("hash", Text), ("url", Text), ("data", Text), ("created_at", Text)]), + ( + "push_subscriptions", + &[ + ("rowid", Int), + ("endpoint", Text), + ("user_id", Text), + ("p256dh", Text), + ("auth", Text), + ("enabled", Bool), + ("activity_at", Text), + ], + ), +]; + +/// Rows read from SQLite and inserted in PostgreSQL at once (65535 parameters at most) +const BATCH: i64 = 1000; + +/// Copy all the data of `from` (SQLite) into `to` (PostgreSQL), returns the +/// number of rows copied per table +pub async fn sqlite_to_postgres(from: &str, to: &str) -> Result> { + anyhow::ensure!(!db::is_postgres_url(from), "the source must be a SQLite database"); + anyhow::ensure!(db::is_postgres_url(to), "the target must be a postgres:// URL"); + + // Both are opened with `db::connect`: the SQLite tables get the columns + // added since they were created, the PostgreSQL ones are created + let Db::Sqlite(src) = db::connect(from).await.context("opening the SQLite database")? else { unreachable!() }; + let Db::Postgres(dst) = db::connect(to).await.context("opening the PostgreSQL database")? else { unreachable!() }; + + for (table, _) in TABLES { + let n: i64 = sqlx::query_scalar(AssertSqlSafe(format!("SELECT COUNT(*) FROM {table}"))).fetch_one(&dst).await?; + if n > 0 { + bail!("the PostgreSQL table {table} is not empty ({n} rows), migrate into an empty database"); + } + } + + let report = copy_all(&src, &dst).await; + src.close().await; + dst.close().await; + report +} + +async fn copy_all(src: &SqlitePool, dst: &PgPool) -> Result> { + let mut tx = dst.begin().await?; + let mut report = vec![]; + + for (table, columns) in TABLES { + let copied = copy_table(src, &mut tx, table, columns).await.with_context(|| format!("copying {table}"))?; + let expected: i64 = sqlx::query_scalar(AssertSqlSafe(format!("SELECT COUNT(*) FROM {table}"))).fetch_one(src).await?; + anyhow::ensure!(copied == expected as u64, "{table}: {copied} rows copied out of {expected}"); + report.push((*table, copied)); + } + + for (table, column) in db::IDENTITY_COLUMNS { + sqlx::query(AssertSqlSafe(format!( + "SELECT setval(pg_get_serial_sequence('{table}', '{column}'), COALESCE(MAX({column}), 0) + 1, false) FROM {table}" + ))) + .execute(&mut *tx) + .await?; + } + + tx.commit().await?; + Ok(report) +} + +async fn copy_table( + src: &SqlitePool, + tx: &mut sqlx::Transaction<'_, Postgres>, + table: &str, + columns: &[(&str, Kind)], +) -> Result { + // SQLite does not enforce the column types: cast them to what PostgreSQL expects + let select = columns + .iter() + .map(|(name, kind)| match kind { + Text => format!("CAST(\"{name}\" AS TEXT)"), + Int | Bool => format!("CAST(\"{name}\" AS INTEGER)"), + Blob => format!("\"{name}\""), + }) + .collect::>() + .join(", "); + let names = columns.iter().map(|(name, _)| format!("\"{name}\"")).collect::>().join(", "); + + let mut copied = 0; + let mut last_rowid = i64::MIN; + loop { + let rows: Vec = sqlx::query(AssertSqlSafe(format!( + "SELECT rowid, {select} FROM {table} WHERE rowid > ? ORDER BY rowid LIMIT ?" + ))) + .bind(last_rowid) + .bind(BATCH) + .fetch_all(src) + .await?; + let Some(last) = rows.last() else { break }; + last_rowid = last.try_get(0)?; + + let mut insert = QueryBuilder::::new(format!("INSERT INTO {table} ({names}) ")); + let mut error = None; + insert.push_values(&rows, |mut values, row| { + for (i, (name, kind)) in columns.iter().enumerate() { + let i = i + 1; + let pushed = match kind { + Text => row.try_get::, _>(i).map(|v| { + values.push_bind(v); + }), + Int => row.try_get::, _>(i).map(|v| { + values.push_bind(v); + }), + Bool => row.try_get::, _>(i).map(|v| { + values.push_bind(v.map(|v| v != 0)); + }), + Blob => row.try_get::>, _>(i).map(|v| { + values.push_bind(v); + }), + }; + if let Err(e) = pushed { + error.get_or_insert(anyhow::anyhow!("column {name}: {e}")); + values.push_bind(None::); + } + } + }); + if let Some(e) = error { + return Err(e); + } + copied += insert.build().execute(&mut **tx).await?.rows_affected(); + } + Ok(copied) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::tests::{postgres_test_url, remove_sqlite, sqlite_test_db}; + use crate::db::{Message, MessageFile}; + + #[tokio::test] + async fn copies_everything() { + let Some(to) = postgres_test_url("migrate").await else { return }; + let (src, path) = sqlite_test_db("migrate").await; + let (me, bob) = ("me@example.org", "bob@example.org"); + + db::user_get_or_create(&src, me).await.unwrap(); + db::user_set(&src, me, "nightmode", &serde_json::json!(true)).await.unwrap(); + db::encrypted_password_set(&src, me, "device", "secret").await.unwrap(); + db::contact_save(&src, &db::Contact { fn_: Some("Bob".into()), ..db::Contact::new(bob) }).await.unwrap(); + db::avatar_save(&src, "hash", "image/png", &[0, 1, 254, 255]).await.unwrap(); + let m = Message { + user_id: me.into(), + id: "m1".into(), + jidfrom: bob.into(), + jidto: me.into(), + type_: "chat".into(), + body: Some("xin chào 👋".into()), + published: "2026-01-01 10:00:00".into(), + ..Default::default() + }; + // A gap in the mids: the copy keeps them + let first = db::message_save(&src, &m).await.unwrap(); + let gone = db::message_save(&src, &Message { id: "m2".into(), ..m.clone() }).await.unwrap(); + let reply = Message { id: "m3".into(), parentmid: Some(first), encrypted: true, omemoheader: Some("{}".into()), ..m.clone() }; + let reply = db::message_save(&src, &reply).await.unwrap(); + let Db::Sqlite(pool) = &src else { unreachable!() }; + sqlx::query("DELETE FROM messages WHERE mid = ?").bind(gone).execute(pool).await.unwrap(); + db::message_file_save(&src, &MessageFile { message_mid: reply, url: "https://f".into(), size: Some(42), ..Default::default() }) + .await + .unwrap(); + db::reactions_replace(&src, first, me, &["👍".into()]).await.unwrap(); + db::mam_earliest_set(&src, me, bob, None).await.unwrap(); + db::omemo_state_set(&src, me, bob, true).await.unwrap(); + db::open_chat_touch(&src, me, bob, None).await.unwrap(); + db::url_save(&src, "u", "https://example.org", "{}").await.unwrap(); + db::push_subscription_save(&src, me, "https://push.example/1", "k", "a").await.unwrap(); + db::push_subscription_save(&src, me, "https://push.example/2", "k", "a").await.unwrap(); + let subs = db::push_subscriptions_list(&src, me).await.unwrap(); + db::push_subscription_toggle(&src, me, subs[0].id, false).await.unwrap(); + let from = format!("sqlite://{}", path.display()); + src.close().await; + + let report = sqlite_to_postgres(&from, &to).await.unwrap(); + let count = |t: &str| report.iter().find(|(name, _)| *name == t).unwrap().1; + assert_eq!((count("messages"), count("reactions"), count("push_subscriptions")), (2, 1, 2)); + + let dst = db::connect(&to).await.unwrap(); + assert!(db::user_get(&dst, me).await.unwrap().unwrap().nightmode); + assert_eq!(db::encrypted_password_get(&dst, me, "device").await.unwrap().as_deref(), Some("secret")); + assert_eq!(db::contact_get(&dst, bob).await.unwrap().unwrap().truename(), "Bob"); + assert_eq!(db::avatar_get(&dst, "hash").await.unwrap().unwrap().1, vec![0, 1, 254, 255]); + let copied = db::message_by_mid(&dst, me, reply).await.unwrap().unwrap(); + assert_eq!((copied.parentmid, copied.encrypted, copied.body.as_deref()), (Some(first), true, Some("xin chào 👋"))); + assert_eq!(db::message_file(&dst, reply).await.unwrap().unwrap().size, Some(42)); + assert_eq!(db::reactions_for(&dst, &[first]).await.unwrap().len(), 1); + assert!(db::mam_earliest_exists(&dst, me, bob).await.unwrap()); + assert_eq!(db::omemo_state(&dst, me, bob).await.unwrap(), Some(true)); + assert_eq!(db::open_chats(&dst, me).await.unwrap(), vec![bob.to_string()]); + assert_eq!(db::url_get(&dst, "u").await.unwrap().as_deref(), Some("{}")); + let copied_subs = db::push_subscriptions_list(&dst, me).await.unwrap(); + assert_eq!(copied_subs.iter().map(|s| (s.id, s.enabled)).collect::>(), subs.iter().map(|s| (s.id, s.id != subs[0].id)).collect::>()); + assert_eq!(db::push_subscriptions(&dst, me).await.unwrap().len(), 1); + + // The sequences continue after the copied ids + let next = db::message_save(&dst, &Message { id: "m4".into(), ..m.clone() }).await.unwrap(); + assert!(next > reply); + db::push_subscription_save(&dst, me, "https://push.example/3", "k", "a").await.unwrap(); + db::reactions_replace(&dst, next, me, &["❤️".into()]).await.unwrap(); + + // A second migration does not duplicate the data + let again = sqlite_to_postgres(&from, &to).await.unwrap_err().to_string(); + assert!(again.contains("not empty"), "{again}"); + + dst.close().await; + remove_sqlite(&path); + } +} diff --git a/src/widgets/chats.rs b/src/widgets/chats.rs index f0ced9a..aa6ca8b 100644 --- a/src/widgets/chats.rs +++ b/src/widgets/chats.rs @@ -23,6 +23,15 @@ fn item_id(jid: &str) -> String { format!("#{}", utils::cleanup_id(&format!("{}_chat_item", utils::slugify(jid)))) } +/// Put the item of a discussion on top of the list. The html is prepared before +/// touching the list: with an await between the remove and the prepend, two +/// concurrent calls (a message during a reload) left two items for one contact. +async fn put_on_top(ctx: &Ctx, jid: &str) { + let html = prepare_chat(ctx, jid).await; + rpc!(ctx, "MovimTpl.remove", item_id(jid)); + rpc!(ctx, "MovimTpl.prepend", "#chats", html); +} + pub async fn call(ctx: &Ctx, func: &str, p: &P<'_>) -> bool { match func { "ajaxHttpGet" => { @@ -138,7 +147,7 @@ pub async fn chat_open(ctx: &Ctx, jid: &str) { let me = ctx.jid(); if !db::open_chat_exists(ctx.db(), &me, jid).await.unwrap_or(false) { let _ = db::open_chat_touch(ctx.db(), &me, jid, None).await; - rpc!(ctx, "MovimTpl.prepend", "#chats", prepare_chat(ctx, jid).await); + put_on_top(ctx, jid).await; } else { rpc!(ctx, "MovimTpl.replace", item_id(jid), prepare_chat(ctx, jid).await); } @@ -170,8 +179,7 @@ pub async fn open(ctx: &Ctx, jid: &str, and_show: bool, history: bool) { }); } let _ = db::open_chat_touch(ctx.db(), &me, jid, None).await; - rpc!(ctx, "MovimTpl.remove", item_id(jid)); - rpc!(ctx, "MovimTpl.prepend", "#chats", prepare_chat(ctx, jid).await); + put_on_top(ctx, jid).await; rpc!(ctx, "Chats.refresh"); if notif::is_current(ctx, jid) || and_show { rpc!(ctx, "Chats.setActive", jid); @@ -209,8 +217,7 @@ pub async fn on_message(ctx: &Ctx, m: &db::Message, origin: &Origin) { return; } let _ = db::open_chat_touch(ctx.db(), &me, &jid, Some(&m.published)).await; - rpc!(ctx, "MovimTpl.remove", item_id(&jid)); - rpc!(ctx, "MovimTpl.prepend", "#chats", prepare_chat(ctx, &jid).await); + put_on_top(ctx, &jid).await; rpc!(ctx, "Chats.refresh"); if notif::is_current(ctx, &jid) { rpc!(ctx, "Chats.setActive", jid.clone()); diff --git a/tests/e2e/__pycache__/browser_edit.cpython-313.pyc b/tests/e2e/__pycache__/browser_edit.cpython-313.pyc new file mode 100644 index 0000000..a9aae45 Binary files /dev/null and b/tests/e2e/__pycache__/browser_edit.cpython-313.pyc differ diff --git a/tests/e2e/browser_edit.py b/tests/e2e/browser_edit.py index e80e321..df54c82 100644 --- a/tests/e2e/browser_edit.py +++ b/tests/e2e/browser_edit.py @@ -87,4 +87,5 @@ async def main(): print(e) await browser.close() -asyncio.run(main()) +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/e2e/browser_omemo_history.py b/tests/e2e/browser_omemo_history.py new file mode 100644 index 0000000..e54462f --- /dev/null +++ b/tests/e2e/browser_omemo_history.py @@ -0,0 +1,68 @@ +"""Own OMEMO messages coming back from the archive (MAM) are shown decrypted, like when edited: +their local id is m_…, the browser stored the plaintext under the id it sent them with.""" +import asyncio +import os +import sqlite3 +import subprocess +from playwright.async_api import async_playwright +from browser import login, BASE, OUT +from browser_edit import send, edit_last, bubbles, open_chat + +D = os.path.dirname(os.path.abspath(__file__)) + + +async def main(): + async with async_playwright() as p: + browser = await p.chromium.launch() + actx, alice, aerr = await login(browser, "alice@localhost", "alicepass", "historyalice") + await alice.wait_for_timeout(4000) + await open_chat(alice, "bob@localhost") + await alice.evaluate("ChatOmemo.enableContactState('bob@localhost', false)") + await alice.wait_for_timeout(5000) + await send(alice, "Lịch sử: bản gốc") + await edit_last(alice, "Lịch sử: đã sửa") + await send(alice, "Lịch sử: tin thứ hai") + + bctx, bob, berr = await login(browser, "bob@localhost", "bobpass", "historybob") + await bob.wait_for_timeout(4000) + await open_chat(bob, "alice@localhost") + await bob.evaluate("ChatOmemo.enableContactState('alice@localhost', false)") + await bob.wait_for_timeout(5000) + await send(bob, "Lịch sử: trả lời của bob") + await alice.wait_for_timeout(3000) + print("before, alice:", await bubbles(alice, 3)) + await bctx.close() + + # The server forgets the discussion, the history comes back from MAM + subprocess.run(["kill", open(f"{D}/server.pid").read().strip()]) + await alice.wait_for_timeout(1000) + db = sqlite3.connect(f"{D}/data/movim.sqlite") + for table in ("messages", "mam_earliests"): + db.execute(f"DELETE FROM {table} WHERE user_id = 'alice@localhost'") + db.commit() + db.close() + subprocess.run([f"{D}/run.sh"]) + + # Same browser context: the decrypted messages of IndexedDB are kept + page = await actx.new_page() + await page.goto(BASE + "/login") + await page.wait_for_timeout(3000) + if "/login" in page.url: + await page.wait_for_selector("#login_widget input[type=submit]:not([disabled])", timeout=15000) + await page.fill("input#username", "alice@localhost") + await page.fill("input#password", "alicepass") + await page.click("#login_widget input[type=submit]") + await page.wait_for_url("**/chat**", timeout=30000) + await page.wait_for_timeout(4000) + await open_chat(page, "bob@localhost") + await page.wait_for_timeout(4000) + print("ids:", await page.evaluate( + "[...document.querySelectorAll('#chat_widget .bubble > div.message')].slice(-3).map(m => m.id)")) + print("from history, alice:", await bubbles(page, 3)) + await page.screenshot(path=f"{OUT}/history_alice.png") + + for e in aerr: + print(e) + await browser.close() + +asyncio.run(main()) diff --git a/tests/e2e/browser_omemo_restore.py b/tests/e2e/browser_omemo_restore.py new file mode 100644 index 0000000..8baa726 --- /dev/null +++ b/tests/e2e/browser_omemo_restore.py @@ -0,0 +1,58 @@ +"""OMEMO key backup restored in another browser: the decrypted messages come with it, +including our own (edited) ones that are never encrypted for the sending device.""" +import asyncio +from playwright.async_api import async_playwright +from browser import login, BASE, OUT +from browser_edit import send, edit_last, bubbles, open_chat + + +async def main(): + async with async_playwright() as p: + browser = await p.chromium.launch() + actx, alice, aerr = await login(browser, "alice@localhost", "alicepass", "restorealice") + bctx, bob, berr = await login(browser, "bob@localhost", "bobpass", "restorebob") + await alice.wait_for_timeout(4000) + await open_chat(alice, "bob@localhost") + await open_chat(bob, "alice@localhost") + + await alice.evaluate("ChatOmemo.enableContactState('bob@localhost', false)") + await alice.wait_for_timeout(5000) + await send(alice, "Khôi phục: bản gốc") + await edit_last(alice, "Khôi phục: đã sửa") + await send(bob, "Khôi phục: trả lời của bob") + print("before, alice:", await bubbles(alice, 2)) + + # Backup + await alice.goto(BASE + "/configuration") + await alice.wait_for_timeout(3000) + await alice.evaluate("Account_ajaxOmemoBackup()") + await alice.wait_for_selector("form[name=omemo_backup] input[name=password]", timeout=10000) + await alice.fill("form[name=omemo_backup] input[name=password]", "secret") + await alice.fill("form[name=omemo_backup] input[name=password_confirmation]", "secret") + async with alice.expect_download() as dl: + await alice.evaluate("Account.omemoBackup()") + path = f"{OUT}/restore_backup.json" + await (await dl.value).save_as(path) + + # Restore in a fresh browser + ctx2, alice2, a2err = await login(browser, "alice@localhost", "alicepass", "restorealice2") + await alice2.wait_for_timeout(4000) + await alice2.goto(BASE + "/configuration") + await alice2.wait_for_timeout(3000) + await alice2.evaluate("Account_ajaxOmemoRestore()") + await alice2.wait_for_selector("form[name=omemo_restore] input[name=file]", timeout=10000) + await alice2.set_input_files("form[name=omemo_restore] input[name=file]", path) + await alice2.fill("form[name=omemo_restore] input[name=password]", "secret") + async with alice2.expect_navigation(): + await alice2.evaluate("Account.omemoRestore()") + await alice2.wait_for_timeout(4000) + + await open_chat(alice2, "bob@localhost") + print("restored, alice:", await bubbles(alice2, 2)) + await alice2.screenshot(path=f"{OUT}/restore_alice2.png") + + for e in aerr + berr + a2err: + print(e) + await browser.close() + +asyncio.run(main())