From 66a2da966b7428d5c45672a300aeccb75f55db93 Mon Sep 17 00:00:00 2001 From: Maxime Gaudin Date: Thu, 10 Sep 2026 09:19:40 +0200 Subject: [PATCH 1/2] fix(archive): dismiss full context group/thread by message id Inbox dedup shows one row per context_id. Slack 1-hour channel groups and real threads share that id, so archiving only the visible message let the next sibling reappear. Archive all siblings with the same context_id (issue #67). Co-authored-by: Cursor --- crates/void-cli/src/service/writes.rs | 42 ++++++++++--- crates/void-core/src/db/database_access.rs | 5 ++ crates/void-core/src/db/messages/archive.rs | 64 +++++++++++++++++++ crates/void-core/src/db/messages/mod.rs | 2 +- crates/void-core/src/db/tests/archive.rs | 68 +++++++++++++++++++++ 5 files changed, 172 insertions(+), 9 deletions(-) diff --git a/crates/void-cli/src/service/writes.rs b/crates/void-cli/src/service/writes.rs index d7380a09..bf6ee9a9 100644 --- a/crates/void-cli/src/service/writes.rs +++ b/crates/void-cli/src/service/writes.rs @@ -471,21 +471,47 @@ async fn archive_by_ids( } } - let remote_synced = if let Some(conn) = connectors.get(&connector_key) { - conn.archive(&msg.external_id, &conv.external_id) - .await - .is_ok() + let archived = db.mark_message_archived_with_context(&msg.id)?; + + let mut remote_synced = connectors.contains_key(&connector_key); + if let Some(conn) = connectors.get(&connector_key) { + for archived_msg in &archived { + // Prefer each sibling's conversation external id when available. + let peer_conv_ext = db + .get_conversation(&archived_msg.conversation_id)? + .map(|c| c.external_id) + .unwrap_or_else(|| conv.external_id.clone()); + if conn + .archive(&archived_msg.external_id, &peer_conv_ext) + .await + .is_err() + { + remote_synced = false; + } + } + // If nothing was newly archived (already archived), still try the + // resolved message so remote state stays consistent. + if archived.is_empty() + && conn + .archive(&msg.external_id, &conv.external_id) + .await + .is_err() + { + remote_synced = false; + } } else { - false - }; + remote_synced = false; + } - db.mark_message_archived(message_id)?; - cleanup_cached_files(&msg); + for archived_msg in &archived { + cleanup_cached_files(archived_msg); + } results.push(json!({ "message_id": message_id, "is_archived": true, "remote_synced": remote_synced, + "archived_count": archived.len(), })); } diff --git a/crates/void-core/src/db/database_access.rs b/crates/void-core/src/db/database_access.rs index f4929228..76f10a5a 100644 --- a/crates/void-core/src/db/database_access.rs +++ b/crates/void-core/src/db/database_access.rs @@ -224,6 +224,11 @@ impl Database { messages::mark_archived(&*self.conn()?, id) } + /// Archive a message and all siblings sharing its `context_id` (inbox thread/group). + pub fn mark_message_archived_with_context(&self, id: &str) -> Result, DbError> { + messages::mark_archived_with_context(&*self.conn()?, id) + } + pub fn update_message_metadata( &self, id: &str, diff --git a/crates/void-core/src/db/messages/archive.rs b/crates/void-core/src/db/messages/archive.rs index 0f3a679c..761b3f4e 100644 --- a/crates/void-core/src/db/messages/archive.rs +++ b/crates/void-core/src/db/messages/archive.rs @@ -61,6 +61,70 @@ pub fn mark_archived(conn: &Connection, id: &str) -> Result { Ok(updated > 0) } +/// Archive a message and every sibling sharing its `context_id` (Slack thread / +/// time-window group, Gmail thread, …). Inbox dedup shows one row per context, so +/// archiving only the visible id leaves older siblings to reappear as the next +/// representative (GitHub issue #67). +/// +/// Messages with no `context_id` are archived alone. Returns the rows that were +/// newly marked archived (for cache cleanup). +pub fn mark_archived_with_context(conn: &Connection, id: &str) -> Result, DbError> { + let Some(msg) = super::read::get(conn, id)? else { + return Ok(Vec::new()); + }; + + let (sql, params_owned): (String, Vec>) = + if let Some(ref ctx) = msg.context_id { + ( + "SELECT id, conversation_id, connection_id, connector, external_id, sender, sender_name, sender_avatar_url, body, timestamp, synced_at, is_archived, reply_to_id, media_type, metadata, context_id, is_saved + FROM messages WHERE context_id = ?1 AND is_archived = 0" + .into(), + vec![Box::new(ctx.clone())], + ) + } else if msg.is_archived { + return Ok(Vec::new()); + } else { + ( + "SELECT id, conversation_id, connection_id, connector, external_id, sender, sender_name, sender_avatar_url, body, timestamp, synced_at, is_archived, reply_to_id, media_type, metadata, context_id, is_saved + FROM messages WHERE id = ?1 AND is_archived = 0" + .into(), + vec![Box::new(id.to_string())], + ) + }; + + let mut stmt = conn.prepare(&sql)?; + let params_ref: Vec<&dyn rusqlite::types::ToSql> = + params_owned.iter().map(|p| p.as_ref()).collect(); + let to_archive: Vec = stmt + .query_map(params_ref.as_slice(), row::row_to_message)? + .collect::>()?; + + if to_archive.is_empty() { + return Ok(Vec::new()); + } + + if let Some(ref ctx) = msg.context_id { + conn.execute( + "UPDATE messages SET is_archived = 1 WHERE context_id = ?1 AND is_archived = 0", + params![ctx], + )?; + debug!( + message_id = %id, + context_id = %ctx, + count = to_archive.len(), + "archived message context group" + ); + } else { + conn.execute( + "UPDATE messages SET is_archived = 1 WHERE id = ?1", + params![id], + )?; + debug!(message_id = %id, "archived single message (no context)"); + } + + Ok(to_archive) +} + pub fn update_metadata( conn: &Connection, id: &str, diff --git a/crates/void-core/src/db/messages/mod.rs b/crates/void-core/src/db/messages/mod.rs index b1a65af8..007bd62a 100644 --- a/crates/void-core/src/db/messages/mod.rs +++ b/crates/void-core/src/db/messages/mod.rs @@ -23,7 +23,7 @@ const DEDUP_CONTEXT_CLAUSE_UNARCHIVED: &str = pub(super) const DEDUP_CONTEXT_CLAUSE_ALIASED: &str = " AND (m.context_id IS NULL OR m.id = (SELECT m2.id FROM messages m2 WHERE m2.context_id = m.context_id ORDER BY m2.timestamp DESC, m2.id DESC LIMIT 1))"; -pub use archive::{bulk_archive_before, mark_archived, update_metadata}; +pub use archive::{bulk_archive_before, mark_archived, mark_archived_with_context, update_metadata}; pub use inbox::{ backfill_avatar_urls, enrich_with_context, messages_pending_file_download, reconcile_inbox, senders_missing_avatar, diff --git a/crates/void-core/src/db/tests/archive.rs b/crates/void-core/src/db/tests/archive.rs index 64f9d14d..2420ab28 100644 --- a/crates/void-core/src/db/tests/archive.rs +++ b/crates/void-core/src/db/tests/archive.rs @@ -172,3 +172,71 @@ fn upsert_preserves_user_archived_flag() { ); assert_eq!(loaded.body.as_deref(), Some("hello edited")); } + +#[test] +fn mark_archived_with_context_archives_all_siblings() { + let db = test_db(); + let conv = make_conversation("c1", "test-slack", "C123"); + db.upsert_conversation(&conv).unwrap(); + + let ctx = "slack-group-C123-1000"; + db.upsert_message(&make_message_with_context( + "m1", "c1", "test-slack", "old", 1_000, Some(ctx), + )) + .unwrap(); + db.upsert_message(&make_message_with_context( + "m2", "c1", "test-slack", "mid", 2_000, Some(ctx), + )) + .unwrap(); + db.upsert_message(&make_message_with_context( + "m3", "c1", "test-slack", "new", 3_000, Some(ctx), + )) + .unwrap(); + // Different context must stay unarchived. + db.upsert_message(&make_message_with_context( + "m4", + "c1", + "test-slack", + "other", + 4_000, + Some("slack-group-other"), + )) + .unwrap(); + + let archived = db.mark_message_archived_with_context("m3").unwrap(); + let mut ids: Vec<_> = archived.iter().map(|m| m.id.as_str()).collect(); + ids.sort(); + assert_eq!(ids, ["m1", "m2", "m3"]); + + assert!(db.get_message("m1").unwrap().unwrap().is_archived); + assert!(db.get_message("m2").unwrap().unwrap().is_archived); + assert!(db.get_message("m3").unwrap().unwrap().is_archived); + assert!( + !db.get_message("m4").unwrap().unwrap().is_archived, + "other context untouched" + ); + + // Inbox must not promote a sibling from the archived group. + let (rows, _) = db + .recent_messages_paginated(None, Some("slack"), 50, 0, false, true, true) + .unwrap(); + assert!( + rows.iter().all(|m| m.id != "m1" && m.id != "m2" && m.id != "m3"), + "archived context group must leave inbox" + ); +} + +#[test] +fn mark_archived_with_context_single_message_without_context() { + let db = test_db(); + let conv = make_conversation("c1", "test-slack", "C123"); + db.upsert_conversation(&conv).unwrap(); + + db.upsert_message(&make_message("solo", "c1", "test-slack", "hi", 1_000)) + .unwrap(); + + let archived = db.mark_message_archived_with_context("solo").unwrap(); + assert_eq!(archived.len(), 1); + assert_eq!(archived[0].id, "solo"); + assert!(db.get_message("solo").unwrap().unwrap().is_archived); +} From 6f13eca47ee49ec5f982fd125527299a2f1f08e0 Mon Sep 17 00:00:00 2001 From: Maxime Gaudin Date: Thu, 10 Sep 2026 10:46:44 +0200 Subject: [PATCH 2/2] fix(archive): batch remote archive and harden context group update Apply review follow-ups on context-group archiving: - Run the sibling SELECT and UPDATE in one transaction so a concurrent insert into the same context cannot be archived without being reported to the caller (which drives cache cleanup). - Add `Connector::archive_batch` (default: one call per message, errors aggregated) and override it for Gmail with `batchModify`, chunked at 1000 ids. Archiving a thread is now one request instead of N. - Group the remote push by conversation and cache conversation lookups instead of querying once per sibling. - Restore cache cleanup on the already-archived path, which stopped running once cleanup was driven by the newly-archived rows. - Document `archived_count` (rows newly archived, 0 when already archived), update docs/commands.md and the changelog. - Cover the no-op, unknown-id and cross-conversation cases, plus the remote batching, with unit tests. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 4 + Cargo.lock | 1 + crates/void-cli/Cargo.toml | 1 + crates/void-cli/src/service/writes.rs | 252 +++++++++++++++--- crates/void-core/src/connector.rs | 30 +++ crates/void-core/src/db/messages/archive.rs | 63 ++--- crates/void-core/src/db/messages/mod.rs | 4 +- crates/void-core/src/db/tests/archive.rs | 131 ++++++++- .../src/connector/connector_trait.rs | 19 ++ docs/commands.md | 2 +- 10 files changed, 434 insertions(+), 73 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e760ed3..019b2236 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **Archive** — `void archive ` now dismisses the whole context group behind the item (Slack thread, Slack 1-hour channel group, Gmail thread) instead of a single row. The inbox shows one row per context, so archiving only the visible id let an older sibling resurface as the next representative. The response gains `archived_count` (rows newly archived by the call, `0` when it was already archived), and Gmail pushes the group in one `batchModify` request. + ### Added - **Remote** — `void remote status` reports `local_version` and `remote_version` so version skew between the client and the server binary is visible at a glance. diff --git a/Cargo.lock b/Cargo.lock index c5f4d919..d7727988 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4729,6 +4729,7 @@ version = "0.11.1" dependencies = [ "anyhow", "assert_cmd", + "async-trait", "chrono", "clap", "croner", diff --git a/crates/void-cli/Cargo.toml b/crates/void-cli/Cargo.toml index b5924428..cb746e31 100644 --- a/crates/void-cli/Cargo.toml +++ b/crates/void-cli/Cargo.toml @@ -43,6 +43,7 @@ rmcp = { workspace = true } schemars = { workspace = true } [dev-dependencies] +async-trait = { workspace = true } assert_cmd = { workspace = true } predicates = { workspace = true } tempfile = { workspace = true } diff --git a/crates/void-cli/src/service/writes.rs b/crates/void-cli/src/service/writes.rs index bf6ee9a9..3a9fd345 100644 --- a/crates/void-cli/src/service/writes.rs +++ b/crates/void-cli/src/service/writes.rs @@ -473,51 +473,86 @@ async fn archive_by_ids( let archived = db.mark_message_archived_with_context(&msg.id)?; - let mut remote_synced = connectors.contains_key(&connector_key); - if let Some(conn) = connectors.get(&connector_key) { - for archived_msg in &archived { - // Prefer each sibling's conversation external id when available. - let peer_conv_ext = db - .get_conversation(&archived_msg.conversation_id)? - .map(|c| c.external_id) - .unwrap_or_else(|| conv.external_id.clone()); - if conn - .archive(&archived_msg.external_id, &peer_conv_ext) - .await - .is_err() - { - remote_synced = false; - } + let remote_synced = match connectors.get(&connector_key) { + Some(conn) => { + sync_remote_archive(db, conn.as_ref(), &msg, &conv.external_id, &archived).await? } - // If nothing was newly archived (already archived), still try the - // resolved message so remote state stays consistent. - if archived.is_empty() - && conn - .archive(&msg.external_id, &conv.external_id) - .await - .is_err() - { - remote_synced = false; - } - } else { - remote_synced = false; - } + None => false, + }; - for archived_msg in &archived { - cleanup_cached_files(archived_msg); + if archived.is_empty() { + // Already archived: no row to clean up, but the cached files of the + // resolved message may still be around from an earlier run. + cleanup_cached_files(&msg); + } else { + for archived_msg in &archived { + cleanup_cached_files(archived_msg); + } } results.push(json!({ "message_id": message_id, "is_archived": true, - "remote_synced": remote_synced, + // Rows newly archived by this call: the message plus its context + // siblings. Zero means it was already archived. "archived_count": archived.len(), + "remote_synced": remote_synced, })); } Ok(json!({ "data": results, "error": null })) } +/// Push an archive to the remote service for every row that was just archived, +/// grouped by conversation so connectors with a bulk endpoint issue one call per +/// conversation instead of one per message. Returns whether every call succeeded. +async fn sync_remote_archive( + db: &Database, + conn: &dyn Connector, + msg: &void_core::models::Message, + conv_external_id: &str, + archived: &[void_core::models::Message], +) -> anyhow::Result { + if archived.is_empty() { + // Nothing newly archived: still push the resolved message so remote + // state converges even when the local row was already archived. + return Ok(conn + .archive(&msg.external_id, conv_external_id) + .await + .is_ok()); + } + + let mut conv_ext_cache: HashMap = HashMap::new(); + let mut by_conversation: HashMap> = HashMap::new(); + for archived_msg in archived { + // Prefer each sibling's own conversation external id when available. + let peer_conv_ext = match conv_ext_cache.get(&archived_msg.conversation_id) { + Some(ext) => ext.clone(), + None => { + let ext = db + .get_conversation(&archived_msg.conversation_id)? + .map(|c| c.external_id) + .unwrap_or_else(|| conv_external_id.to_string()); + conv_ext_cache.insert(archived_msg.conversation_id.clone(), ext.clone()); + ext + } + }; + by_conversation + .entry(peer_conv_ext) + .or_default() + .push(archived_msg.external_id.as_str()); + } + + let mut synced = true; + for (peer_conv_ext, external_ids) in &by_conversation { + if let Err(e) = conn.archive_batch(external_ids, peer_conv_ext).await { + warn!(conversation = %peer_conv_ext, error = %e, "remote archive failed"); + synced = false; + } + } + Ok(synced) +} + async fn run_slack_scheduled_send( connection: &void_core::config::ConnectionConfig, channel: &str, @@ -616,6 +651,161 @@ mod tests { use void_core::config::VoidConfig; use void_core::models::{Conversation, ConversationKind}; + use std::sync::Mutex; + use void_core::db::Database as CoreDatabase; + use void_core::models::{ConnectorType, HealthStatus, Message}; + use void_core::test_fixtures::make_message; + + /// Records the archive calls it receives so grouping can be asserted. + struct RecordingConnector { + batches: Mutex)>>, + singles: Mutex>, + } + + impl RecordingConnector { + fn new() -> Self { + Self { + batches: Mutex::new(Vec::new()), + singles: Mutex::new(Vec::new()), + } + } + } + + #[async_trait::async_trait] + impl Connector for RecordingConnector { + fn connector_type(&self) -> ConnectorType { + ConnectorType::from_static("slack") + } + fn connection_id(&self) -> &str { + "test-slack" + } + async fn authenticate(&mut self) -> anyhow::Result<()> { + Ok(()) + } + async fn start_sync( + &self, + _db: std::sync::Arc, + _cancel: tokio_util::sync::CancellationToken, + ) -> anyhow::Result<()> { + Ok(()) + } + async fn health_check(&self) -> anyhow::Result { + anyhow::bail!("not used") + } + async fn send_message( + &self, + _to: &str, + _content: MessageContent, + ) -> anyhow::Result { + anyhow::bail!("not used") + } + async fn reply( + &self, + _message_id: &str, + _content: MessageContent, + _in_thread: bool, + ) -> anyhow::Result { + anyhow::bail!("not used") + } + async fn archive( + &self, + external_id: &str, + conversation_external_id: &str, + ) -> anyhow::Result<()> { + self.singles.lock().unwrap().push(( + conversation_external_id.to_string(), + external_id.to_string(), + )); + Ok(()) + } + async fn archive_batch( + &self, + external_ids: &[&str], + conversation_external_id: &str, + ) -> anyhow::Result<()> { + self.batches.lock().unwrap().push(( + conversation_external_id.to_string(), + external_ids.iter().map(|s| s.to_string()).collect(), + )); + Ok(()) + } + } + + fn seed_conversation(db: &Database, id: &str, external_id: &str) { + db.upsert_conversation(&Conversation { + id: id.into(), + connection_id: "test-slack".into(), + connector: "slack".into(), + external_id: external_id.into(), + name: None, + kind: ConversationKind::Channel, + last_message_at: None, + unread_count: 0, + is_muted: false, + metadata: None, + }) + .expect("seed conversation"); + } + + fn stub_message(id: &str, conversation_id: &str, external_id: &str) -> Message { + let mut msg = make_message(id, conversation_id, "test-slack", "body", 0); + msg.external_id = external_id.into(); + msg + } + + #[tokio::test] + async fn sync_remote_archive_groups_siblings_by_conversation() { + let db = test_db(); + seed_conversation(&db, "c1", "C111"); + seed_conversation(&db, "c2", "C222"); + let conn = RecordingConnector::new(); + + let archived = vec![ + stub_message("m1", "c1", "ts-1"), + stub_message("m2", "c1", "ts-2"), + stub_message("m3", "c2", "ts-3"), + ]; + + let synced = sync_remote_archive(&db, &conn, &archived[0], "C111", &archived) + .await + .expect("sync"); + + assert!(synced); + assert!(conn.singles.lock().unwrap().is_empty()); + let mut batches = conn.batches.lock().unwrap().clone(); + batches.sort(); + assert_eq!( + batches, + vec![ + ( + "C111".to_string(), + vec!["ts-1".to_string(), "ts-2".to_string()] + ), + ("C222".to_string(), vec!["ts-3".to_string()]), + ], + "one batch call per conversation" + ); + } + + #[tokio::test] + async fn sync_remote_archive_falls_back_to_single_when_nothing_new() { + let db = test_db(); + seed_conversation(&db, "c1", "C111"); + let conn = RecordingConnector::new(); + let msg = stub_message("m1", "c1", "ts-1"); + + let synced = sync_remote_archive(&db, &conn, &msg, "C111", &[]) + .await + .expect("sync"); + + assert!(synced); + assert!(conn.batches.lock().unwrap().is_empty()); + assert_eq!( + *conn.singles.lock().unwrap(), + vec![("C111".to_string(), "ts-1".to_string())] + ); + } + fn test_db() -> Database { Database::open(std::path::Path::new(":memory:")).expect("in-memory db") } diff --git a/crates/void-core/src/connector.rs b/crates/void-core/src/connector.rs index 0db1ee69..048527ed 100644 --- a/crates/void-core/src/connector.rs +++ b/crates/void-core/src/connector.rs @@ -78,6 +78,36 @@ pub trait Connector: Send + Sync { Ok(()) } + /// Archive several messages of the same conversation. + /// + /// Default implementation archives them one by one and reports an error if + /// any call failed, without aborting the rest. Connectors with a bulk + /// endpoint should override this to issue a single request. + async fn archive_batch( + &self, + external_ids: &[&str], + conversation_external_id: &str, + ) -> anyhow::Result<()> { + let mut failed = 0usize; + for external_id in external_ids { + if self + .archive(external_id, conversation_external_id) + .await + .is_err() + { + failed += 1; + } + } + if failed > 0 { + anyhow::bail!( + "{failed}/{} remote archive calls failed for {}", + external_ids.len(), + self.connector_type() + ); + } + Ok(()) + } + /// Forward a message to another recipient. /// `external_id` is the platform-specific message identifier. /// `conversation_external_id` is the platform-specific conversation/channel ID. diff --git a/crates/void-core/src/db/messages/archive.rs b/crates/void-core/src/db/messages/archive.rs index 761b3f4e..fc6666f3 100644 --- a/crates/void-core/src/db/messages/archive.rs +++ b/crates/void-core/src/db/messages/archive.rs @@ -67,59 +67,50 @@ pub fn mark_archived(conn: &Connection, id: &str) -> Result { /// representative (GitHub issue #67). /// /// Messages with no `context_id` are archived alone. Returns the rows that were -/// newly marked archived (for cache cleanup). +/// newly marked archived (for cache cleanup); an already-archived message yields +/// an empty vector. Select and update run in one transaction so a concurrent +/// insert into the same context cannot be archived without being reported. pub fn mark_archived_with_context(conn: &Connection, id: &str) -> Result, DbError> { let Some(msg) = super::read::get(conn, id)? else { return Ok(Vec::new()); }; - let (sql, params_owned): (String, Vec>) = - if let Some(ref ctx) = msg.context_id { - ( - "SELECT id, conversation_id, connection_id, connector, external_id, sender, sender_name, sender_avatar_url, body, timestamp, synced_at, is_archived, reply_to_id, media_type, metadata, context_id, is_saved - FROM messages WHERE context_id = ?1 AND is_archived = 0" - .into(), - vec![Box::new(ctx.clone())], - ) - } else if msg.is_archived { - return Ok(Vec::new()); - } else { - ( - "SELECT id, conversation_id, connection_id, connector, external_id, sender, sender_name, sender_avatar_url, body, timestamp, synced_at, is_archived, reply_to_id, media_type, metadata, context_id, is_saved - FROM messages WHERE id = ?1 AND is_archived = 0" - .into(), - vec![Box::new(id.to_string())], - ) - }; + // Static clauses; the value always travels as a bound parameter. + let (where_clause, param): (&str, &str) = match msg.context_id.as_deref() { + Some(ctx) => ("context_id = ?1", ctx), + None if msg.is_archived => return Ok(Vec::new()), + None => ("id = ?1", id), + }; - let mut stmt = conn.prepare(&sql)?; - let params_ref: Vec<&dyn rusqlite::types::ToSql> = - params_owned.iter().map(|p| p.as_ref()).collect(); + let tx = conn.unchecked_transaction()?; + + let mut stmt = tx.prepare(&format!( + "SELECT id, conversation_id, connection_id, connector, external_id, sender, sender_name, sender_avatar_url, body, timestamp, synced_at, is_archived, reply_to_id, media_type, metadata, context_id, is_saved + FROM messages WHERE {where_clause} AND is_archived = 0" + ))?; let to_archive: Vec = stmt - .query_map(params_ref.as_slice(), row::row_to_message)? + .query_map(params![param], row::row_to_message)? .collect::>()?; + drop(stmt); if to_archive.is_empty() { return Ok(Vec::new()); } - if let Some(ref ctx) = msg.context_id { - conn.execute( - "UPDATE messages SET is_archived = 1 WHERE context_id = ?1 AND is_archived = 0", - params![ctx], - )?; - debug!( + tx.execute( + &format!("UPDATE messages SET is_archived = 1 WHERE {where_clause} AND is_archived = 0"), + params![param], + )?; + tx.commit()?; + + match msg.context_id.as_deref() { + Some(ctx) => debug!( message_id = %id, context_id = %ctx, count = to_archive.len(), "archived message context group" - ); - } else { - conn.execute( - "UPDATE messages SET is_archived = 1 WHERE id = ?1", - params![id], - )?; - debug!(message_id = %id, "archived single message (no context)"); + ), + None => debug!(message_id = %id, "archived single message (no context)"), } Ok(to_archive) diff --git a/crates/void-core/src/db/messages/mod.rs b/crates/void-core/src/db/messages/mod.rs index 007bd62a..6c23894b 100644 --- a/crates/void-core/src/db/messages/mod.rs +++ b/crates/void-core/src/db/messages/mod.rs @@ -23,7 +23,9 @@ const DEDUP_CONTEXT_CLAUSE_UNARCHIVED: &str = pub(super) const DEDUP_CONTEXT_CLAUSE_ALIASED: &str = " AND (m.context_id IS NULL OR m.id = (SELECT m2.id FROM messages m2 WHERE m2.context_id = m.context_id ORDER BY m2.timestamp DESC, m2.id DESC LIMIT 1))"; -pub use archive::{bulk_archive_before, mark_archived, mark_archived_with_context, update_metadata}; +pub use archive::{ + bulk_archive_before, mark_archived, mark_archived_with_context, update_metadata, +}; pub use inbox::{ backfill_avatar_urls, enrich_with_context, messages_pending_file_download, reconcile_inbox, senders_missing_avatar, diff --git a/crates/void-core/src/db/tests/archive.rs b/crates/void-core/src/db/tests/archive.rs index 2420ab28..f225e549 100644 --- a/crates/void-core/src/db/tests/archive.rs +++ b/crates/void-core/src/db/tests/archive.rs @@ -181,15 +181,30 @@ fn mark_archived_with_context_archives_all_siblings() { let ctx = "slack-group-C123-1000"; db.upsert_message(&make_message_with_context( - "m1", "c1", "test-slack", "old", 1_000, Some(ctx), + "m1", + "c1", + "test-slack", + "old", + 1_000, + Some(ctx), )) .unwrap(); db.upsert_message(&make_message_with_context( - "m2", "c1", "test-slack", "mid", 2_000, Some(ctx), + "m2", + "c1", + "test-slack", + "mid", + 2_000, + Some(ctx), )) .unwrap(); db.upsert_message(&make_message_with_context( - "m3", "c1", "test-slack", "new", 3_000, Some(ctx), + "m3", + "c1", + "test-slack", + "new", + 3_000, + Some(ctx), )) .unwrap(); // Different context must stay unarchived. @@ -221,7 +236,8 @@ fn mark_archived_with_context_archives_all_siblings() { .recent_messages_paginated(None, Some("slack"), 50, 0, false, true, true) .unwrap(); assert!( - rows.iter().all(|m| m.id != "m1" && m.id != "m2" && m.id != "m3"), + rows.iter() + .all(|m| m.id != "m1" && m.id != "m2" && m.id != "m3"), "archived context group must leave inbox" ); } @@ -240,3 +256,110 @@ fn mark_archived_with_context_single_message_without_context() { assert_eq!(archived[0].id, "solo"); assert!(db.get_message("solo").unwrap().unwrap().is_archived); } + +#[test] +fn mark_archived_with_context_is_noop_when_group_already_archived() { + let db = test_db(); + db.upsert_conversation(&make_conversation("c1", "test-slack", "C123")) + .unwrap(); + + let ctx = "slack-group-C123-1000"; + db.upsert_message(&make_message_with_context( + "m1", + "c1", + "test-slack", + "old", + 1_000, + Some(ctx), + )) + .unwrap(); + db.upsert_message(&make_message_with_context( + "m2", + "c1", + "test-slack", + "new", + 2_000, + Some(ctx), + )) + .unwrap(); + + assert_eq!( + db.mark_message_archived_with_context("m2").unwrap().len(), + 2 + ); + // Second call has nothing left to archive. + assert!(db + .mark_message_archived_with_context("m2") + .unwrap() + .is_empty()); + assert!(db.get_message("m1").unwrap().unwrap().is_archived); + assert!(db.get_message("m2").unwrap().unwrap().is_archived); +} + +#[test] +fn mark_archived_with_context_is_noop_when_single_message_already_archived() { + let db = test_db(); + db.upsert_conversation(&make_conversation("c1", "test-slack", "C123")) + .unwrap(); + db.upsert_message(&make_message("solo", "c1", "test-slack", "hi", 1_000)) + .unwrap(); + + assert_eq!( + db.mark_message_archived_with_context("solo").unwrap().len(), + 1 + ); + assert!(db + .mark_message_archived_with_context("solo") + .unwrap() + .is_empty()); +} + +#[test] +fn mark_archived_with_context_spans_conversations() { + let db = test_db(); + db.upsert_conversation(&make_conversation("c1", "test-slack", "C123")) + .unwrap(); + db.upsert_conversation(&make_conversation("c2", "test-slack", "C456")) + .unwrap(); + + let ctx = "slack-thread-1000"; + db.upsert_message(&make_message_with_context( + "m1", + "c1", + "test-slack", + "here", + 1_000, + Some(ctx), + )) + .unwrap(); + db.upsert_message(&make_message_with_context( + "m2", + "c2", + "test-slack", + "there", + 2_000, + Some(ctx), + )) + .unwrap(); + + let archived = db.mark_message_archived_with_context("m1").unwrap(); + let mut convs: Vec<_> = archived + .iter() + .map(|m| m.conversation_id.as_str()) + .collect(); + convs.sort(); + assert_eq!( + convs, + ["c1", "c2"], + "siblings keep their own conversation id" + ); +} + +#[test] +fn mark_archived_with_context_unknown_id_is_noop() { + let db = test_db(); + assert!(db + .mark_message_archived_with_context("nope") + .unwrap() + .is_empty()); +} diff --git a/crates/void-gmail/src/connector/connector_trait.rs b/crates/void-gmail/src/connector/connector_trait.rs index 6a5e2e5c..1126cd2c 100644 --- a/crates/void-gmail/src/connector/connector_trait.rs +++ b/crates/void-gmail/src/connector/connector_trait.rs @@ -217,6 +217,25 @@ impl Connector for GmailConnector { Ok(()) } + async fn archive_batch( + &self, + external_ids: &[&str], + _conversation_external_id: &str, + ) -> anyhow::Result<()> { + if external_ids.is_empty() { + return Ok(()); + } + info!( + count = external_ids.len(), + "archiving Gmail messages (batch)" + ); + // batchModify caps each request at 1000 ids. + for chunk in external_ids.chunks(1000) { + self.batch_modify(chunk, &[], &["INBOX"]).await?; + } + Ok(()) + } + async fn reply( &self, message_id: &str, diff --git a/docs/commands.md b/docs/commands.md index 3c8a072f..302cc9a5 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -59,7 +59,7 @@ Most read commands accept: | `void send --via --to --message ` | Send a new message. Use `--conversation ` instead of `--to` to target an existing void conversation (e.g. WhatsApp notes-to-self / "Message yourself"). `--connection ` to pick an account, `--subject` (email), `--cc` / `--bcc` (Gmail only, comma-separated), `--file ` to attach, `--at