fix: IMAP sync - use imap_uid, add sync queue with retry, restore deleted on sync, upload sent mails

- imap_delete_mail: use stored imap_uid instead of Message-ID search, raise on failure
- imap_move_mail: same imap_uid fix, raise on failure
- New MailSyncQueue model + migration 0007 for pending IMAP operations
- delete_mail route: queues failed IMAP delete for retry
- move_mail route: queues failed IMAP move for retry
- process_sync_queue: retries pending delete/move operations in auto-sync loop
- Auto-sync loop: processes sync queue before pulling new mails
- Restore soft-deleted mails if they still exist on IMAP server during sync
- Upload sent mails to IMAP Sent folder via APPEND after SMTP send
- Plugin version bumped to 1.2.0

app/plugins/builtins/mail/migrations/0007_sync_queue.sql

@@ -0,0 +1,17 @@
+-- Mail sync queue table for pending IMAP operations
+CREATE TABLE IF NOT EXISTS mail_sync_queue (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    mail_id UUID NOT NULL,
+    operation VARCHAR(50) NOT NULL,
+    payload JSONB NOT NULL DEFAULT '{}',
+    status VARCHAR(20) NOT NULL DEFAULT 'pending',
+    attempts INTEGER NOT NULL DEFAULT 0,
+    max_attempts INTEGER NOT NULL DEFAULT 3,
+    last_error TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS ix_mail_sync_queue_status ON mail_sync_queue (status);
+CREATE INDEX IF NOT EXISTS ix_mail_sync_queue_mail ON mail_sync_queue (mail_id);

app/plugins/builtins/mail/models.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import uuid
-from datetime import datetime
+from datetime import UTC, datetime
 
 from sqlalchemy import (
     Boolean,
@@ -11,6 +11,7 @@ from sqlalchemy import (
     ForeignKey,
     Index,
     Integer,
+    JSON,
     String,
     Text,
     UniqueConstraint,
@@ -415,3 +416,30 @@ class ContactPgpKey(Base, TenantMixin):
     contact_id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False)
     public_key_armored: Mapped[str] = mapped_column(Text, nullable=False)
     key_id: Mapped[str] = mapped_column(String(255), nullable=False, default="")
+
+
+class MailSyncQueue(Base, TenantMixin):
+    """Queue for pending IMAP operations that need retry."""
+
+    __tablename__ = "mail_sync_queue"
+    __table_args__ = (
+        Index("ix_mail_sync_queue_status", "status"),
+        Index("ix_mail_sync_queue_mail", "mail_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    mail_id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False)
+    operation: Mapped[str] = mapped_column(String(50), nullable=False)
+    payload: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
+    status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
+    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    max_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
+    )

app/plugins/builtins/mail/plugin.py

@@ -13,10 +13,18 @@ logger = logging.getLogger(__name__)
 
 
 async def _auto_sync_loop() -> None:
-    """Background loop: sync all active mail accounts every 5 minutes."""
+    """Background loop: process pending sync queue, then sync all active mail accounts every 5 minutes."""
-    from app.plugins.builtins.mail.services import auto_sync_all_accounts
+    from app.plugins.builtins.mail.services import auto_sync_all_accounts, process_sync_queue
+    from app.core.db import get_session_factory
 
     while True:
+        try:
+            factory = get_session_factory()
+            async with factory() as db:
+                await process_sync_queue(db)
+                await db.commit()
+        except Exception as exc:
+            logger.warning("process_sync_queue error: %s", exc)
         try:
             await auto_sync_all_accounts()
         except Exception as exc:
@@ -31,7 +39,7 @@ class MailPlugin(BasePlugin):
 
     manifest = PluginManifest(
         name="mail",
-        version="1.1.0",
+        version="1.2.0",
         display_name="Mail",
         description=(
             "Email management: IMAP sync, SMTP send, threading, "
@@ -46,7 +54,7 @@ class MailPlugin(BasePlugin):
             ),
         ],
         events=[],
-        migrations=["0001_initial.sql", "0006_flag_type.sql"],
+        migrations=["0001_initial.sql", "0006_flag_type.sql", "0007_sync_queue.sql"],
         permissions=["mail:read", "mail:send", "mail:config", "mail:share", "mail:write", "mail:delete"],
     )
 

app/plugins/builtins/mail/routes.py

@@ -31,6 +31,7 @@ from app.plugins.builtins.mail.models import (
     MailLabelAssignment,
     MailRule,
     MailSignature,
+    MailSyncQueue,
     MailTemplate,
     PgpKey,
 )
@@ -1531,12 +1532,22 @@ async def delete_mail(
     # Soft-delete: set deleted_at = now()
     mail.deleted_at = datetime.now(UTC)
     await db.flush()
-    # IMAP delete: non-critical, errors logged but DB soft-delete still succeeds
+    # IMAP delete: queue for retry if it fails
     try:
         await mail_services.imap_delete_mail(db, m_id, tenant_id)
     except Exception as exc:
         import logging
-        logging.getLogger(__name__).warning("IMAP delete failed for mail %s: %s", m_id, exc)
+        logging.getLogger(__name__).warning("IMAP delete failed for mail %s: %s, queuing for retry", m_id, exc)
+        queue_entry = MailSyncQueue(
+            tenant_id=tenant_id,
+            mail_id=m_id,
+            operation="delete",
+            payload={},
+            status="pending",
+            last_error=str(exc),
+        )
+        db.add(queue_entry)
+        await db.flush()
 
 
 @router.post("/{mail_id}/move")
@@ -1570,12 +1581,22 @@ async def move_mail(
     # Update mail.folder_id
     mail.folder_id = target_f_id
     await db.flush()
-    # IMAP move: non-critical
+    # IMAP move: queue for retry if it fails
     try:
         await mail_services.imap_move_mail(db, m_id, target_f_id, tenant_id)
     except Exception as exc:
         import logging
-        logging.getLogger(__name__).warning("IMAP move failed for mail %s: %s", m_id, exc)
+        logging.getLogger(__name__).warning("IMAP move failed for mail %s: %s, queuing for retry", m_id, exc)
+        queue_entry = MailSyncQueue(
+            tenant_id=tenant_id,
+            mail_id=m_id,
+            operation="move",
+            payload={"target_folder_id": str(target_f_id)},
+            status="pending",
+            last_error=str(exc),
+        )
+        db.add(queue_entry)
+        await db.flush()
     return mail_to_response(mail)
 
 

app/plugins/builtins/mail/services.py

@@ -852,6 +852,10 @@ async def imap_sync_account(
                         ).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: restored soft-deleted mail %s (still on IMAP)", existing_mail.id)
                         # Track folder moves: update folder_id and imap_uid if the
                         # mail now appears in a different IMAP folder.
                         if existing_mail.folder_id != folder.id:
@@ -1054,6 +1058,7 @@ async def send_mail_via_smtp(
     cc_addrs = cc_addrs or []
     bcc_addrs = bcc_addrs or []
     attachment_paths = attachment_paths or []
+    logger = logging.getLogger(__name__)
 
     # Apply signature if provided
     if signature and signature.body_html:
@@ -1184,6 +1189,27 @@ async def send_mail_via_smtp(
             db.add(sent_mail)
             await db.flush()
 
+            # Upload sent mail to IMAP Sent folder via APPEND
+            try:
+                imap_client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port)
+                await imap_client.wait_hello_from_server()
+                await imap_client.login(account.username, password)
+                # Build raw email bytes for APPEND
+                raw_email_bytes = msg.as_bytes()
+                append_resp = await imap_client.append(
+                    sent_folder.imap_name,
+                    r'(\Seen)',
+                    None,
+                    raw_email_bytes,
+                )
+                if append_resp.result == 'OK':
+                    logger.info("send_mail: uploaded sent mail to IMAP folder %s", sent_folder.imap_name)
+                else:
+                    logger.warning("send_mail: IMAP APPEND failed for sent folder %s: %s", sent_folder.imap_name, append_resp)
+                await imap_client.logout()
+            except Exception as imap_exc:
+                logger.warning("send_mail: IMAP APPEND failed (non-critical): %s", imap_exc)
+
         # ── Notification: mail sent ──
         try:
             await create_notification(
@@ -1757,9 +1783,8 @@ async def imap_delete_mail(
 ) -> None:
     """Delete mail from IMAP server.
 
-    Connects to IMAP, selects the mail's folder, finds UID by Message-ID,
+    Uses stored imap_uid directly; falls back to Message-ID search if missing.
-    marks it as \\Deleted and expunges.
+    Raises exceptions on failure so caller can queue for retry.
-    Non-critical: errors are logged, DB soft-delete still succeeds.
     """
     import logging
 
@@ -1798,8 +1823,8 @@ async def imap_delete_mail(
         logger.warning("imap_delete_mail: account not found for mail %s", mail_id)
         return
 
-    if not mail.message_id:
+    if not mail.imap_uid and not mail.message_id:
-        logger.warning("imap_delete_mail: mail %s has no message_id, cannot sync", mail_id)
+        logger.warning("imap_delete_mail: mail %s has no imap_uid or message_id, cannot sync", mail_id)
         return
 
     password = await get_account_password(account)
@@ -1812,20 +1837,20 @@ async def imap_delete_mail(
 
         select_resp = await client.select(folder.imap_name)
         if select_resp.result != 'OK':
-            logger.warning("imap_delete_mail: cannot select folder %s", folder.imap_name)
+            raise RuntimeError(f"cannot select folder {folder.imap_name}")
-            return
 
+        # Use stored imap_uid directly; fall back to Message-ID search
+        if mail.imap_uid:
+            uid_str = mail.imap_uid
+        else:
             search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"')
             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 not uids:
-            logger.warning("imap_delete_mail: no UID found for Message-ID %s", mail.message_id)
+                raise RuntimeError(f"no UID found for Message-ID {mail.message_id}")
-            return
-
             uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0])
 
         await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
@@ -1835,6 +1860,7 @@ async def imap_delete_mail(
 
     except Exception as exc:
         logger.warning("imap_delete_mail: failed for mail %s: %s", mail_id, exc)
+        raise
     finally:
         if client is not None:
             try:
@@ -1854,8 +1880,9 @@ async def imap_move_mail(
 ) -> None:
     """Move mail to another folder on IMAP server.
 
+    Uses stored imap_uid directly; falls back to Message-ID search if missing.
     Uses UID MOVE if supported, otherwise COPY + STORE \\Deleted + EXPUNGE.
-    Non-critical: errors are logged, DB update still succeeds.
+    Raises exceptions on failure so caller can queue for retry.
     """
     import logging
 
@@ -1905,8 +1932,8 @@ async def imap_move_mail(
         logger.warning("imap_move_mail: account not found for mail %s", mail_id)
         return
 
-    if not mail.message_id:
+    if not mail.imap_uid and not mail.message_id:
-        logger.warning("imap_move_mail: mail %s has no message_id, cannot sync", mail_id)
+        logger.warning("imap_move_mail: mail %s has no imap_uid or message_id, cannot sync", mail_id)
         return
 
     password = await get_account_password(account)
@@ -1919,20 +1946,20 @@ async def imap_move_mail(
 
         select_resp = await client.select(source_folder.imap_name)
         if select_resp.result != 'OK':
-            logger.warning("imap_move_mail: cannot select folder %s", source_folder.imap_name)
+            raise RuntimeError(f"cannot select folder {source_folder.imap_name}")
-            return
 
+        # Use stored imap_uid directly; fall back to Message-ID search
+        if mail.imap_uid:
+            uid_str = mail.imap_uid
+        else:
             search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"')
             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 not uids:
-            logger.warning("imap_move_mail: no UID found for Message-ID %s", mail.message_id)
+                raise RuntimeError(f"no UID found for Message-ID {mail.message_id}")
-            return
-
             uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0])
 
         # Try UID MOVE first; fall back to COPY + STORE \\Deleted + EXPUNGE
@@ -1947,8 +1974,7 @@ async def imap_move_mail(
         # Fallback: COPY + STORE \\Deleted + EXPUNGE
         copy_resp = await client.uid('copy', uid_str, target_folder.imap_name)
         if copy_resp.result != 'OK':
-            logger.warning("imap_move_mail: COPY failed for mail %s", mail_id)
+            raise RuntimeError(f"COPY failed for mail {mail_id}")
-            return
 
         await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
         await client.expunge()
@@ -1957,6 +1983,7 @@ async def imap_move_mail(
 
     except Exception as exc:
         logger.warning("imap_move_mail: failed for mail %s: %s", mail_id, exc)
+        raise
     finally:
         if client is not None:
             try:
@@ -2416,3 +2443,67 @@ async def auto_sync_all_accounts() -> None:
                 await db.commit()
             except Exception:
                 await db.rollback()
+
+
+async def process_sync_queue(db: AsyncSession) -> None:
+    """Process pending IMAP operations from the sync queue.
+
+    Called at the start of each auto-sync loop iteration.
+    Retries failed delete/move operations.
+    """
+    import logging
+
+    logger = logging.getLogger(__name__)
+
+    from app.plugins.builtins.mail.models import MailSyncQueue
+
+    pending = (
+        await db.execute(
+            select(MailSyncQueue).where(
+                and_(
+                    MailSyncQueue.status == "pending",
+                    MailSyncQueue.attempts < MailSyncQueue.max_attempts,
+                )
+            )
+        )
+    ).scalars().all()
+
+    if not pending:
+        return
+
+    logger.info("process_sync_queue: processing %d pending operation(s)", len(pending))
+
+    for entry in pending:
+        try:
+            if entry.operation == "delete":
+                await imap_delete_mail(db, entry.mail_id, entry.tenant_id)
+            elif entry.operation == "move":
+                target_folder_id = uuid.UUID(entry.payload.get("target_folder_id", ""))
+                await imap_move_mail(db, entry.mail_id, target_folder_id, entry.tenant_id)
+            else:
+                logger.warning("process_sync_queue: unknown operation %s", entry.operation)
+                entry.status = "failed"
+                entry.last_error = f"Unknown operation: {entry.operation}"
+                continue
+
+            entry.status = "completed"
+            entry.updated_at = datetime.now(UTC)
+            logger.info("process_sync_queue: completed %s for mail %s", entry.operation, entry.mail_id)
+
+        except Exception as exc:
+            entry.attempts += 1
+            entry.last_error = str(exc)
+            entry.updated_at = datetime.now(UTC)
+            if entry.attempts >= entry.max_attempts:
+                entry.status = "failed"
+                logger.warning(
+                    "process_sync_queue: giving up on %s for mail %s after %d attempts: %s",
+                    entry.operation, entry.mail_id, entry.attempts, exc,
+                )
+            else:
+                logger.info(
+                    "process_sync_queue: retry %d/%d for %s mail %s: %s",
+                    entry.attempts, entry.max_attempts, entry.operation, entry.mail_id, exc,
+                )
+
+    await db.flush()