diff --git a/app/plugins/builtins/mail/migrations/0007_sync_queue.sql b/app/plugins/builtins/mail/migrations/0007_sync_queue.sql new file mode 100644 index 0000000..1b6e21f --- /dev/null +++ b/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); diff --git a/app/plugins/builtins/mail/models.py b/app/plugins/builtins/mail/models.py index 29458f0..125c842 100644 --- a/app/plugins/builtins/mail/models.py +++ b/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) + ) diff --git a/app/plugins/builtins/mail/plugin.py b/app/plugins/builtins/mail/plugin.py index 5a80af2..3caa934 100644 --- a/app/plugins/builtins/mail/plugin.py +++ b/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.""" - from app.plugins.builtins.mail.services import auto_sync_all_accounts + """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, 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"], ) diff --git a/app/plugins/builtins/mail/routes.py b/app/plugins/builtins/mail/routes.py index c4446cd..bdc1aa7 100644 --- a/app/plugins/builtins/mail/routes.py +++ b/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) diff --git a/app/plugins/builtins/mail/services.py b/app/plugins/builtins/mail/services.py index b4a6b5a..003f90d 100644 --- a/app/plugins/builtins/mail/services.py +++ b/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, - marks it as \\Deleted and expunges. - Non-critical: errors are logged, DB soft-delete still succeeds. + Uses stored imap_uid directly; falls back to Message-ID search if missing. + Raises exceptions on failure so caller can queue for retry. """ 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: - logger.warning("imap_delete_mail: mail %s has no message_id, cannot sync", mail_id) + if not mail.imap_uid and not mail.message_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,21 +1837,21 @@ 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) - return + raise RuntimeError(f"cannot select folder {folder.imap_name}") - 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() + # Use stored imap_uid directly; fall back to Message-ID search + if mail.imap_uid: + uid_str = mail.imap_uid else: - uids = [] - - if not uids: - logger.warning("imap_delete_mail: no UID found for Message-ID %s", mail.message_id) - return - - uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0]) + 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: + raise RuntimeError(f"no UID found for Message-ID {mail.message_id}") + uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0]) await client.uid('store', uid_str, r'+FLAGS (\\Deleted)') await client.expunge() @@ -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: - logger.warning("imap_move_mail: mail %s has no message_id, cannot sync", mail_id) + if not mail.imap_uid and not mail.message_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,21 +1946,21 @@ 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) - return + raise RuntimeError(f"cannot select folder {source_folder.imap_name}") - 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() + # Use stored imap_uid directly; fall back to Message-ID search + if mail.imap_uid: + uid_str = mail.imap_uid else: - uids = [] - - if not uids: - logger.warning("imap_move_mail: no UID found for Message-ID %s", mail.message_id) - return - - uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0]) + 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: + raise RuntimeError(f"no UID found for Message-ID {mail.message_id}") + 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 try: @@ -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) - return + raise RuntimeError(f"COPY failed for mail {mail_id}") 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()