diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index 5dd7ae9..7cba21a 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -1350,6 +1350,37 @@ impl DatabaseEngine { }) } + /// Return the subset of `ids` already present in the `messages` table, as a single + /// `SELECT id FROM messages WHERE id IN (...)` query. + /// + /// Lets a caller classify a whole fetched batch as new-vs-seen with one round trip instead + /// of one existence lookup per id. Passing an empty slice short-circuits to an empty set + /// without querying at all, since `IN ()` is not valid SQL. + pub async fn existing_message_ids( + &self, + ids: &[String], + ) -> Result, DatabaseError> { + if ids.is_empty() { + return Ok(std::collections::HashSet::new()); + } + + let mut builder: sqlx::QueryBuilder<'_, sqlx::Sqlite> = + sqlx::QueryBuilder::new("SELECT id FROM messages WHERE id IN ("); + let mut separated = builder.separated(", "); + for id in ids { + separated.push_bind(id); + } + builder.push(")"); + + let rows: Vec<(String,)> = builder + .build_query_as() + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)?; + + Ok(rows.into_iter().map(|(id,)| id).collect()) + } + /// Save a [`nuncio_core::model::CalendarEvent`] to SQLite (INSERT OR REPLACE). /// /// Unlike [`Self::save_email`], calendar event summary/location are never encrypted at @@ -3730,6 +3761,42 @@ mod tests { } } + #[tokio::test] + async fn existing_message_ids_returns_only_the_ids_already_persisted() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + engine + .save_email(&export_test_email("msg-1", "acct-a", "inbox")) + .await + .unwrap(); + engine + .save_email(&export_test_email("msg-2", "acct-a", "inbox")) + .await + .unwrap(); + + let found = engine + .existing_message_ids(&[ + "msg-1".to_string(), + "msg-2".to_string(), + "msg-3-never-saved".to_string(), + ]) + .await + .unwrap(); + + assert_eq!(found.len(), 2); + assert!(found.contains("msg-1")); + assert!(found.contains("msg-2")); + assert!(!found.contains("msg-3-never-saved")); + } + + #[tokio::test] + async fn existing_message_ids_with_empty_input_returns_empty_set_without_querying() { + let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); + + let found = engine.existing_message_ids(&[]).await.unwrap(); + + assert!(found.is_empty()); + } + #[tokio::test] async fn list_messages_for_export_with_no_filters_returns_every_message() { let (engine, _dir) = DatabaseEngine::connect_ephemeral().await.unwrap(); diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index bbab07d..98dc78a 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -195,39 +195,26 @@ pub async fn apply_filter_actions( applied } -/// Reports whether `message_id` is not yet present in the store. -/// -/// Sync is not yet incremental (a re-sync re-fetches every message the -/// backend still reports, not just genuinely new ones), so this is the guard -/// that keeps filter actions firing exactly once per message: it must be -/// checked BEFORE `save_email` persists (or re-persists) the message, since -/// `save_email` is `INSERT OR REPLACE` and would otherwise erase the -/// distinction between "arriving for the first time" and "seen again". -/// -/// A lookup failure that is not "not found" (a genuine DB error) is logged -/// and treated as "not new", the conservative choice: firing filters again on -/// a message we can't positively identify as new risks duplicate remote -/// mutations, which is worse than occasionally missing a fire on a message -/// that was in fact new. -async fn is_new_message(db: &nuncio_store::db::DatabaseEngine, message_id: &str) -> bool { - match db.get_message(message_id).await { - Ok(_) => false, - Err(e) if e.is_not_found() => true, - Err(e) => { - tracing::warn!( - "could not determine whether message '{message_id}' is new before sync \ - (treating as not-new to avoid duplicate filter firing): {e}" - ); - false - } - } -} - /// Fetch every folder and every message in every folder from `backend`, /// persisting each message via [`DatabaseEngine::save_email`] and, for /// messages genuinely new to the store, evaluating it against -/// `filter_engine` via [`apply_filter_actions`] (see [`is_new_message`] for -/// why this must be gated rather than run on every persisted message). +/// `filter_engine` via [`apply_filter_actions`]. +/// +/// Sync is not yet incremental (a re-sync re-fetches every message the +/// backend still reports, not just genuinely new ones), so new-vs-seen must +/// be determined BEFORE `save_email` persists (or re-persists) anything -- +/// `save_email` is `INSERT OR REPLACE` and would otherwise erase the +/// distinction between "arriving for the first time" and "seen again". Doing +/// that with one existence lookup per message (`DatabaseEngine::get_message`) +/// is an N+1 query pattern over a fetched chunk; instead, this classifies an +/// entire folder's chunk with a single batched +/// `DatabaseEngine::existing_message_ids` lookup up front, then folds in a +/// per-batch "already seen this pass" set while iterating so a chunk that +/// itself contains a repeated id (e.g. a backend surfacing the same message +/// twice in one fetch) still fires at most once -- exactly as if each +/// message's existence had been checked one at a time immediately before its +/// own `save_email`. +/// /// Returns the total number of messages processed (a message id "processed" /// more than once, e.g. by a backend that returns the same id from multiple /// folders, is counted once per occurrence -- `save_email` itself is @@ -260,8 +247,19 @@ async fn fetch_and_persist( .sync_messages(&folder.id, last_state.as_deref()) .await?; let fetched = emails.len(); + + // One round trip classifies the whole chunk instead of one lookup per + // message. A lookup failure that is not "not found" would be a + // genuine DB error; `existing_message_ids` surfaces those as `Err` + // rather than silently treating the batch as "not new", so callers + // never persist and skip-filter on an unverified assumption. + let chunk_ids: Vec = emails.iter().map(|e| e.id.clone()).collect(); + let already_present = db.existing_message_ids(&chunk_ids).await?; + let mut seen_this_pass = std::collections::HashSet::new(); + for email in emails { - let is_new = is_new_message(db, &email.id).await; + let is_new = + !already_present.contains(&email.id) && seen_this_pass.insert(email.id.clone()); db.save_email(&email).await?; if is_new { apply_filter_actions(db, filter_engine, &email).await; @@ -1304,6 +1302,81 @@ mod tests { ); } + #[tokio::test] + async fn sync_with_backend_fires_filters_only_for_new_messages_in_a_mixed_batch() { + // A single fetched chunk containing a mix of an already-seen message + // (persisted by a prior sync) and genuinely new ones must fire filter + // actions ONLY for the new ones, via the batched + // `existing_message_ids` classification -- not a per-message lookup. + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let event_bus = EventBus::new(); + + let mark_read_rule = nuncio_filter::NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse rule"); + let filter_engine = FilterEngine::new(vec![mark_read_rule]).expect("compile rule"); + + // Pre-seed one message directly, simulating it having landed in a + // prior sync pass. It must NOT re-fire even though this pass's + // backend still reports it (sync is not yet incremental). + db.save_email(&mock_email( + "m-already-seen", + "inbox", + "Urgent: known issue", + )) + .await + .expect("pre-seed existing message"); + + let mock = MockMailBackend::new(); + mock.add_folder(Folder { + id: "inbox".to_string(), + name: "Inbox".to_string(), + total_messages: 3, + unread_messages: 3, + }); + // Mixed chunk: one already-present id, two genuinely new ids. + mock.add_message(mock_email("m-already-seen", "inbox", "Urgent: known issue")); + mock.add_message(mock_email("m-new-1", "inbox", "Urgent: brand new")); + mock.add_message(mock_email("m-new-2", "inbox", "Urgent: also new")); + + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + .await + .expect("sync succeeds"); + assert_eq!(synced, 3, "every fetched message is still persisted"); + + let logs = db + .list_filter_execution_logs(10) + .await + .expect("list execution logs"); + assert_eq!( + logs.len(), + 2, + "filters fire only for the two genuinely new messages in the mixed batch" + ); + let fired_ids: std::collections::HashSet = + logs.iter().map(|l| l.message_id.clone()).collect(); + assert!(fired_ids.contains("m-new-1")); + assert!(fired_ids.contains("m-new-2")); + assert!(!fired_ids.contains("m-already-seen")); + + // The pre-seeded message must still be persisted (INSERT OR REPLACE + // semantics are unaffected by the batched new/seen classification) + // but its read flag must be untouched by this pass's MARK READ rule. + let already_seen = db + .get_message("m-already-seen") + .await + .expect("still persisted"); + assert!( + !already_seen.read, + "a message that already existed before this sync must not re-fire MARK READ" + ); + } + #[tokio::test] async fn sync_with_backend_persists_and_threads_the_folder_checkpoint_across_syncs() { // The regression this story exists to kill: the returned checkpoint