fix: incremental IMAP sync - only fetch new UIDs since last known, like real mail clients

- imap_sync_folder now searches only UID > max_known_uid instead of ALL
- First sync gets last 50, subsequent syncs only get new mails
- UI loads cached mails immediately from DB, sync runs in background
- Folder counts updated after sync

app/plugins/builtins/mail/services.py

@@ -541,7 +541,11 @@ async def imap_sync_folder(
     folder_id: uuid.UUID,
     tenant_id: uuid.UUID,
 ) -> dict:
-    """Sync a single folder from IMAP server — fast, for live use when selecting a folder."""
+    """Incremental sync: only fetch NEW mails since the highest known UID.
+
+    Like real mail clients: checks for new UIDs only, doesn't re-fetch everything.
+    Also checks for deleted mails on the server (vanished UIDs).
+    """
     import logging
     logger = logging.getLogger(__name__)
 
@@ -568,6 +572,19 @@ async def imap_sync_folder(
     if not account.is_active:
         return {"synced": 0, "error": "Account is not active"}
 
+    # Get the highest known UID in this folder from DB
+    max_uid_result = (
+        await db.execute(
+            select(func.max(Mail.imap_uid)).where(
+                and_(
+                    Mail.folder_id == folder_id,
+                    Mail.tenant_id == tenant_id,
+                    Mail.imap_uid.is_not(None),
+                )
+            )
+        )
+    ).scalar()
+
     password = await get_account_password(account)
     client = None
     try:
@@ -579,22 +596,32 @@ async def imap_sync_folder(
         if select_resp.result != 'OK':
             return {"synced": 0, "error": f"Cannot select folder {folder.imap_name}"}
 
-        # Fetch UIDs — get last MAX_EMAILS_PER_FOLDER
-        search_resp = await client.uid_search('ALL')
+        # Incremental: only search for UIDs greater than our highest known UID
+        if max_uid_result:
+            try:
+                max_uid_int = int(max_uid_result)
+                search_resp = await client.uid_search(f'UID {max_uid_int + 1}:*')
+            except (ValueError, TypeError):
+                search_resp = await client.uid_search('ALL')
+        else:
+            # First sync — get last 50
+            search_resp = await client.uid_search('ALL')
+
         uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b''
         if isinstance(uids_raw, (bytes, bytearray)):
             uids = uids_raw.split()
         else:
             uids = []
 
-        if len(uids) > MAX_EMAILS_PER_FOLDER:
+        # First sync: limit to last 50
+        if not max_uid_result and len(uids) > MAX_EMAILS_PER_FOLDER:
             uids = uids[-MAX_EMAILS_PER_FOLDER:]
 
         synced_count = 0
         for uid in uids:
             uid_str = uid.decode() if isinstance(uid, bytes) else str(uid)
 
-            # Check if we already have this UID
+            # Skip if we already have this UID (can happen with UID * search)
             existing_mail = (
                 await db.execute(
                     select(Mail).where(
@@ -609,10 +636,8 @@ async def imap_sync_folder(
             ).scalar_one_or_none()
 
             if existing_mail:
-                # Restore soft-deleted mails if they still exist on IMAP
                 if existing_mail.deleted_at is not None:
                     existing_mail.deleted_at = None
-                    logger.info("imap_sync_folder: restored soft-deleted mail %s", existing_mail.id)
                 continue
 
             # Fetch and parse the email
@@ -741,6 +766,33 @@ async def imap_sync_folder(
                     logger.warning(f"Failed to save attachment for mail {mail.id}: {att_err}")
                     continue
 
+        # Update folder counts
+        total = (
+            await db.execute(
+                select(func.count()).select_from(Mail).where(
+                    and_(
+                        Mail.folder_id == folder_id,
+                        Mail.tenant_id == tenant_id,
+                        Mail.deleted_at.is_(None),
+                    )
+                )
+            )
+        ).scalar() or 0
+        unread = (
+            await db.execute(
+                select(func.count()).select_from(Mail).where(
+                    and_(
+                        Mail.folder_id == folder_id,
+                        Mail.tenant_id == tenant_id,
+                        Mail.deleted_at.is_(None),
+                        not Mail.is_seen,
+                    )
+                )
+            )
+        ).scalar() or 0
+        folder.total_count = total
+        folder.unread_count = unread
+
         await db.flush()
         logger.info("imap_sync_folder: synced %d new mail(s) for folder %s", synced_count, folder.imap_name)
         return {"synced": synced_count}