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<std::collections::HashSet<String>, 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();

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<String> = 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<String> =
+            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