feat: NIP-77 negentropy for MLS group sync + parallel boot

Replace O(n) per-group relay fetches with NIP-77 set reconciliation
for MLS groups. Exchanges fingerprints with relays to identify only
missing events, then fetches the delta — near-instant when up to date.
Boot sync runs DM and MLS negentropy concurrently via tokio::join!,
reducing total sync from ~70s to ~1.2s.

- Add sync_mls_groups_quick() with negentropy relay racing + background
  straggler gap-filling (matches DM pattern)
- Add load_mls_negentropy_items() with SQL-level since filtering
- Modify sync_group_since_cursor to accept prefetched events (skip
  relay fetch for batched multi-group sync)
- Remove mls_processed_events cleanup (needed for negentropy state)
- Widen DM quick phase to 7 days to match MLS window
- Use negentropy for single-relay reconnection sync

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: JSKitty

src-tauri/src/commands/mls.rs

@@ -1141,7 +1141,7 @@ pub async fn sync_mls_groups_now(
 
             if let Some(id) = group_id {
                 // Sync specific group since last cursor
-                mls.sync_group_since_cursor(&id)
+                mls.sync_group_since_cursor(&id, None)
                     .await
                     .map_err(|e| e.to_string())
             } else {
@@ -1163,7 +1163,7 @@ pub async fn sync_mls_groups_now(
                 let mut total_new: u32 = 0;
 
                 for gid in group_ids {
-                    match mls.sync_group_since_cursor(&gid).await {
+                    match mls.sync_group_since_cursor(&gid, None).await {
                         Ok((processed, new_msgs)) => {
                             total_processed = total_processed.saturating_add(processed);
                             total_new = total_new.saturating_add(new_msgs);
@@ -1187,6 +1187,279 @@ pub async fn sync_mls_groups_now(
     .map_err(|e| format!("Task join error: {}", e))?
 }
 
+/// Quick MLS group sync using NIP-77 negentropy set reconciliation.
+/// Exchanges fingerprints with relays to identify only missing events, then fetches
+/// the delta — near-instant when already up to date. Only syncs groups active within 7 days.
+pub async fn sync_mls_groups_quick() -> Result<(u32, u32), String> {
+    use futures_util::StreamExt;
+
+    tokio::task::spawn_blocking(move || {
+        let rt = tokio::runtime::Handle::current();
+        rt.block_on(async move {
+            let mls = MlsService::new_persistent_static().map_err(|e| e.to_string())?;
+
+            // Load all non-evicted groups
+            let groups = db::load_mls_groups().await.unwrap_or_default();
+            let active_groups: Vec<_> = groups.into_iter()
+                .filter(|g| !g.evicted)
+                .collect();
+
+            if active_groups.is_empty() {
+                println!("[MLS] Quick sync: no groups to sync");
+                return Ok((0, 0));
+            }
+
+            // Load cursors to determine which groups are recently active
+            let cursors = mls.read_event_cursors().await.unwrap_or_default();
+            let now_secs = Timestamp::now().as_secs();
+            let seven_days_ago = now_secs.saturating_sub(7 * 24 * 3600);
+
+            // Filter to recently-active groups:
+            // - Groups with a cursor last_seen_at within 7 days (had recent messages)
+            // - Groups created within 7 days (newly joined, need initial sync)
+            let recent_groups: Vec<_> = active_groups.into_iter()
+                .filter(|g| {
+                    let cursor_recent = cursors.get(&g.group_id)
+                        .map(|c| c.last_seen_at >= seven_days_ago)
+                        .unwrap_or(false);
+                    let created_recent = g.created_at >= seven_days_ago;
+                    cursor_recent || created_recent
+                })
+                .collect();
+
+            if recent_groups.is_empty() {
+                println!("[MLS] Quick sync: no recently-active groups (within 7d)");
+                return Ok((0, 0));
+            }
+
+            // Compute earliest cursor across all recent groups for the `since` filter
+            let min_since = recent_groups.iter()
+                .map(|g| {
+                    cursors.get(&g.group_id)
+                        .map(|c| c.last_seen_at)
+                        .unwrap_or_else(|| {
+                            if g.created_at > 0 { g.created_at } else { seven_days_ago }
+                        })
+                })
+                .min()
+                .unwrap_or(seven_days_ago);
+
+            let group_ids: Vec<String> = recent_groups.iter()
+                .map(|g| g.group_id.clone())
+                .collect();
+
+            // Load known MLS event IDs for negentropy fingerprinting (SQL-filtered)
+            let neg_items = db::load_mls_negentropy_items(Some(min_since)).unwrap_or_default();
+
+            println!("[MLS] Quick sync (negentropy): {} groups, {} known items, since={}",
+                recent_groups.len(), neg_items.len(), min_since);
+
+            // Build filter for negentropy reconciliation
+            let filter = Filter::new()
+                .kind(Kind::MlsGroupMessage)
+                .since(Timestamp::from_secs(min_since))
+                .custom_tags(
+                    SingleLetterTag::lowercase(Alphabet::H),
+                    group_ids.iter().map(|s| s.as_str()),
+                );
+
+            // Negentropy dry-run: exchange fingerprints to find missing events
+            let sync_opts = nostr_sdk::SyncOptions::new()
+                .direction(nostr_sdk::SyncDirection::Down)
+                .initial_timeout(std::time::Duration::from_secs(10))
+                .dry_run();
+
+            let client = NOSTR_CLIENT.get().ok_or("Nostr client not initialized")?;
+
+            // Get Relay objects for trusted relays
+            let relay_map = client.relays().await;
+            let trusted_urls = active_trusted_relays().await;
+            let trusted_relays: Vec<(String, nostr_sdk::Relay)> = trusted_urls.iter()
+                .filter_map(|url| {
+                    let normalized = url.trim_end_matches('/');
+                    relay_map.iter()
+                        .find(|(u, _)| u.as_str().trim_end_matches('/') == normalized)
+                        .map(|(_, r)| (url.to_string(), r.clone()))
+                })
+                .collect();
+            drop(relay_map);
+
+            if trusted_relays.is_empty() {
+                println!("[MLS] Quick sync: no trusted relays available");
+                return Ok((0, 0));
+            }
+
+            // Race all trusted relays — first to reconcile drives sync
+            let mut relay_futs = futures_util::stream::FuturesUnordered::new();
+            for (url, relay) in &trusted_relays {
+                let url = url.clone();
+                let relay = relay.clone();
+                let f = filter.clone();
+                let items = neg_items.clone();
+                let opts = sync_opts.clone();
+                relay_futs.push(async move {
+                    let result = tokio::time::timeout(
+                        std::time::Duration::from_secs(10),
+                        relay.sync_with_items(f, items, &opts),
+                    ).await;
+                    (url, result)
+                });
+            }
+
+            // Drain until first successful reconciliation
+            let mut missing_ids: Vec<EventId> = Vec::new();
+            let mut primary_succeeded = false;
+            while let Some((url, result)) = relay_futs.next().await {
+                match result {
+                    Ok(Ok(recon)) => {
+                        missing_ids = recon.remote.into_iter().collect();
+                        println!("[MLS] Quick sync: {} reconciled, {} missing events",
+                            url, missing_ids.len());
+                        primary_succeeded = true;
+                        break;
+                    }
+                    Ok(Err(e)) => eprintln!("[MLS] Quick sync: {} negentropy failed: {}", url, e),
+                    Err(_) => eprintln!("[MLS] Quick sync: {} negentropy timed out (10s)", url),
+                }
+            }
+
+            // Spawn background task for remaining relays — they fill gaps silently
+            if primary_succeeded && !relay_futs.is_empty() {
+                let primary_set: std::collections::HashSet<EventId> = missing_ids.iter().copied().collect();
+                let bg_client = client.clone();
+                let bg_group_ids: Vec<String> = recent_groups.iter().map(|g| g.group_id.clone()).collect();
+                tokio::spawn(async move {
+                    let mut extra_ids: Vec<EventId> = Vec::new();
+                    while let Some((url, result)) = relay_futs.next().await {
+                        match result {
+                            Ok(Ok(recon)) => {
+                                let new: Vec<EventId> = recon.remote.into_iter()
+                                    .filter(|id| !primary_set.contains(id))
+                                    .collect();
+                                if !new.is_empty() {
+                                    println!("[MLS][BG] {} reconciled: {} additional missing events", url, new.len());
+                                    extra_ids.extend(new);
+                                } else {
+                                    println!("[MLS][BG] {} reconciled: 0 additional", url);
+                                }
+                            }
+                            Ok(Err(e)) => eprintln!("[MLS][BG] {} negentropy failed: {}", url, e),
+                            Err(_) => eprintln!("[MLS][BG] {} timed out (10s)", url),
+                        }
+                    }
+
+                    // Fetch + process extra events found by background relays
+                    if !extra_ids.is_empty() {
+                        println!("[MLS][BG] Fetching {} additional events from background relays", extra_ids.len());
+                        match bg_client.fetch_events_from(
+                            active_trusted_relays().await,
+                            Filter::new().ids(extra_ids).kind(Kind::MlsGroupMessage),
+                            std::time::Duration::from_secs(15),
+                        ).await {
+                            Ok(events) => {
+                                // Group by h-tag and process per group
+                                let mut by_group: std::collections::HashMap<String, Vec<nostr_sdk::Event>> =
+                                    std::collections::HashMap::new();
+                                for event in events {
+                                    if let Some(h_tag) = event.tags.find(TagKind::SingleLetter(
+                                        SingleLetterTag::lowercase(Alphabet::H),
+                                    )) {
+                                        if let Some(gid) = h_tag.content() {
+                                            by_group.entry(gid.to_string()).or_default().push(event);
+                                        }
+                                    }
+                                }
+                                if let Ok(mls) = MlsService::new_persistent_static() {
+                                    for gid in &bg_group_ids {
+                                        if let Some(group_events) = by_group.remove(gid) {
+                                            match mls.sync_group_since_cursor(gid, Some(group_events)).await {
+                                                Ok((_, new)) if new > 0 => {
+                                                    println!("[MLS][BG] {} new messages for group {}", new, &gid[..8.min(gid.len())]);
+                                                }
+                                                Err(e) => eprintln!("[MLS][BG] sync failed for {}: {}", &gid[..8.min(gid.len())], e),
+                                                _ => {}
+                                            }
+                                        }
+                                    }
+                                }
+                                println!("[MLS][BG] Background relay sync complete");
+                            }
+                            Err(e) => eprintln!("[MLS][BG] Fetch error: {}", e),
+                        }
+                    }
+                });
+            }
+
+            if missing_ids.is_empty() {
+                println!("[MLS] Quick sync: no missing events (already up to date)");
+                return Ok((0, 0));
+            }
+
+            // Fetch only the missing events
+            let events = client
+                .fetch_events_from(
+                    active_trusted_relays().await,
+                    Filter::new()
+                        .ids(missing_ids)
+                        .kind(Kind::MlsGroupMessage),
+                    std::time::Duration::from_secs(15),
+                )
+                .await
+                .map_err(|e| format!("MLS negentropy fetch failed: {}", e))?;
+
+            println!("[MLS] Quick sync: fetched {} missing events for {} groups",
+                events.len(), recent_groups.len());
+
+            // Group events by h-tag value
+            let mut events_by_group: std::collections::HashMap<String, Vec<nostr_sdk::Event>> =
+                std::collections::HashMap::new();
+            for event in events {
+                if let Some(h_tag) = event.tags.find(TagKind::SingleLetter(
+                    SingleLetterTag::lowercase(Alphabet::H),
+                )) {
+                    if let Some(gid) = h_tag.content() {
+                        events_by_group
+                            .entry(gid.to_string())
+                            .or_default()
+                            .push(event);
+                    }
+                }
+            }
+
+            // Process each group's events through the engine (with pre-fetched events)
+            let mut total_processed: u32 = 0;
+            let mut total_new: u32 = 0;
+
+            for group in &recent_groups {
+                let group_events = events_by_group.remove(&group.group_id).unwrap_or_default();
+                if group_events.is_empty() {
+                    continue;
+                }
+
+                match mls.sync_group_since_cursor(&group.group_id, Some(group_events)).await {
+                    Ok((processed, new_msgs)) => {
+                        total_processed = total_processed.saturating_add(processed);
+                        total_new = total_new.saturating_add(new_msgs);
+                    }
+                    Err(e) => {
+                        eprintln!("[MLS] Quick sync failed for group {}: {}", group.group_id, e);
+                    }
+                }
+
+                if let Err(e) = sync_mls_group_participants(group.group_id.clone()).await {
+                    eprintln!("[MLS] Failed to sync participants for group {}: {}", group.group_id, e);
+                }
+            }
+
+            println!("[MLS] Quick sync complete (negentropy): {} processed, {} new messages",
+                total_processed, total_new);
+            Ok((total_processed, total_new))
+        })
+    })
+    .await
+    .map_err(|e| format!("Task join error: {}", e))?
+}
+
 /// Sync the participants array for an MLS group chat with the actual members from the engine
 /// This ensures chat.participants is always up-to-date
 /// (Internal helper - not a Tauri command)
@@ -1556,7 +1829,7 @@ pub async fn accept_mls_welcome(welcome_event_id_hex: String) -> Result<bool, St
             // Immediately prefetch recent MLS messages for this group so the chat list shows previews
             // and ordering without requiring the user to open the chat. This loads a recent slice
             // (48h window by default in sync_group_since_cursor) rather than full history.
-            match mls.sync_group_since_cursor(&nostr_group_id).await {
+            match mls.sync_group_since_cursor(&nostr_group_id, None).await {
                 Ok((processed, new_msgs)) => {
                     println!("[MLS] Post-accept initial sync (epoch={}): processed={}, new={}", welcome_epoch, processed, new_msgs);
                     // Optional: let UI know initial sync finished for this group

src-tauri/src/commands/sync.rs

@@ -172,8 +172,8 @@ pub async fn fetch_messages<R: Runtime>(
             }
         }
 
-        // Also sync MLS group messages after single-relay reconnection
-        if let Err(e) = crate::commands::mls::sync_mls_groups_now(None).await {
+        // Also sync MLS group messages after single-relay reconnection (negentropy)
+        if let Err(e) = crate::commands::mls::sync_mls_groups_quick().await {
             eprintln!("[Single-Relay Sync] Failed to sync MLS groups: {}", e);
         }
 
@@ -414,10 +414,15 @@ pub async fn fetch_messages<R: Runtime>(
     } // STATE lock released — no lock held during network operations
 
     // ========================================================================
-    // Negentropy (NIP-77) set reconciliation — single-pass sync
+    // Negentropy (NIP-77) + MLS group sync — run concurrently
     // ========================================================================
 
     let sync_start = std::time::Instant::now();
+
+    // Run DM quick phase and MLS group sync concurrently
+    let (_dm_new_messages, _mls_result) = tokio::join!(
+        // Task A: DM Quick Phase (negentropy reconciliation)
+        async {
     let mut new_messages_count: u32 = 0;
 
     // Load our known wrapper IDs + timestamps for reconciliation fingerprinting
@@ -426,9 +431,9 @@ pub async fn fetch_messages<R: Runtime>(
     println!("[Sync] Loaded {} negentropy items ({} with valid timestamps)",
         negentropy_items.len(), valid_ts_count);
 
-    // Quick phase: last 2 days — tiny item set for near-instant reconciliation.
+    // Quick phase: last 7 days — small item set for near-instant reconciliation.
     // Shows recent offline messages within ~1s. Full archive sync runs in background after.
-    let quick_since = Timestamp::now().as_secs().saturating_sub(2 * 24 * 3600);
+    let quick_since = Timestamp::now().as_secs().saturating_sub(7 * 24 * 3600);
     let quick_items: Vec<(EventId, Timestamp)> = negentropy_items.iter()
         .filter(|(_, ts)| ts.as_secs() >= quick_since)
         .cloned()
@@ -437,7 +442,7 @@ pub async fn fetch_messages<R: Runtime>(
         .pubkey(my_public_key)
         .kind(Kind::GiftWrap)
         .since(Timestamp::from_secs(quick_since));
-    println!("[Sync] Quick phase: {} items (last 2d), full: {}", quick_items.len(), negentropy_items.len());
+    println!("[Sync] Quick phase: {} items (last 7d), full: {}", quick_items.len(), negentropy_items.len());
 
     // Dry-run negentropy reconciliation — exchange fingerprints only
     // This identifies which events the relay has that we don't, without transferring data.
@@ -698,6 +703,17 @@ pub async fn fetch_messages<R: Runtime>(
     // Quick phase done — recent messages visible to user
     println!("[Sync] Quick phase: {:.2?}, {} new messages", sync_start.elapsed(), new_messages_count);
 
+    new_messages_count
+        },
+        // Task B: MLS Quick Sync (batched single-request fetch for recently-active groups)
+        async {
+            let mls_start = std::time::Instant::now();
+            if let Err(e) = crate::commands::mls::sync_mls_groups_quick().await {
+                eprintln!("[Sync] Parallel MLS group sync failed: {}", e);
+            }
+            println!("[Sync] MLS group sync: {:.2?}", mls_start.elapsed());
+        }
+    );
     // ========================================================================
     // Archive sync — full negentropy reconciliation (drives sync UI)
     // ========================================================================