diff --git a/app/plugins/builtins/mail/attachments.py b/app/plugins/builtins/mail/attachments.py new file mode 100644 index 0000000..2d52106 --- /dev/null +++ b/app/plugins/builtins/mail/attachments.py @@ -0,0 +1,117 @@ +"""Attachment storage helpers for the Mail plugin. + +Extracted from services.py as part of the God-object split (BUG-018 pilot). +Re-exported by ``app.plugins.builtins.mail.services``. +""" + +from __future__ import annotations + +import os +import re +import uuid + +import aiofiles + +from app.config import settings +from app.plugins.builtins.mail.models import MailAttachment + +# ─── Attachment Storage Helpers ─── + +MAX_ATTACHMENT_SIZE = 25 * 1024 * 1024 # 25 MB + + +def _decode_mime_filename(filename: str) -> str: + """Decode MIME-encoded filename, handling =?charset?Q?...?= and =?charset?B?...?= patterns.""" + if not filename: + return "attachment" + # If no MIME encoding pattern, return as-is + if "=?" not in filename: + return filename + try: + from email.header import decode_header, make_header + + return str(make_header(decode_header(filename))) + except Exception: + # Fallback: manually decode Q-encoding if decode_header fails + # This handles cases where the email parser partially processed the filename + try: + import base64 + + def decode_q(match): + charset, encoding, encoded = ( + match.group(1), + match.group(2).upper(), + match.group(3), + ) + if encoding == "B": + decoded = base64.b64decode(encoded).decode( + charset or "utf-8", errors="replace" + ) + else: # Q encoding + decoded = encoded.replace("_", " ") + decoded = re.sub( + r"=([0-9A-Fa-f]{2})", + lambda m: chr(int(m.group(1), 16)), + decoded, + ) + decoded = decoded.encode("latin-1").decode( + charset or "utf-8", errors="replace" + ) + return decoded + + return re.sub(r"=\?([^?]+)\?([BbQq])\?([^?]*)\?=", decode_q, filename) + except Exception: + return filename + + +def _sanitize_filename(filename: str) -> str: + """Sanitize a filename to prevent path traversal attacks.""" + # Remove any path components — keep only the basename + filename = os.path.basename(filename or "attachment") + # Replace potentially dangerous characters + filename = re.sub(r"[^a-zA-Z0-9._-]", "_", filename) + # Ensure non-empty + if not filename: + filename = "attachment" + # Limit length + if len(filename) > 200: + name, ext = os.path.splitext(filename) + filename = name[:200 - len(ext)] + ext + return filename + + +def _attachment_storage_path(mail_id: uuid.UUID, filename: str) -> str: + """Build the on-disk storage path for a mail attachment.""" + safe_name = _sanitize_filename(filename) + return os.path.join( + settings.storage_path, + "mail_attachments", + str(mail_id), + safe_name, + ) + + +async def _save_attachment_to_storage( + mail_id: uuid.UUID, filename: str, content: bytes +) -> str: + """Save attachment content to disk and return the storage path.""" + storage_path = _attachment_storage_path(mail_id, filename) + os.makedirs(os.path.dirname(storage_path), exist_ok=True) + async with aiofiles.open(storage_path, "wb") as f: + await f.write(content) + return storage_path + + +def attachment_to_response(att: MailAttachment) -> dict: + """Convert a MailAttachment ORM object to a response dict.""" + return { + "id": str(att.id), + "mail_id": str(att.mail_id), + "filename": att.filename, + "mime_type": att.mime_type, + "size_bytes": att.size_bytes, + "size": att.size_bytes, # alias for frontend compatibility + "content_id": att.content_id, + "is_inline": bool(att.content_id), + "dms_file_id": str(att.dms_file_id) if att.dms_file_id else None, + } diff --git a/app/plugins/builtins/mail/imap_sync.py b/app/plugins/builtins/mail/imap_sync.py new file mode 100644 index 0000000..6e5c714 --- /dev/null +++ b/app/plugins/builtins/mail/imap_sync.py @@ -0,0 +1,1148 @@ +"""IMAP sync service for the Mail plugin (F-MAIL-01). + +Extracted from services.py as part of the God-object split (BUG-018 pilot). +Re-exported by ``app.plugins.builtins.mail.services``. + +Contains folder sync, account sync, thread-id computation and the German +folder name mapping for standard IMAP folders. +""" + +from __future__ import annotations + +import logging +import re +import uuid +from datetime import UTC, datetime +from email import message_from_bytes + +import aioimaplib +from sqlalchemy import and_, func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.notifications import create_notification +from app.plugins.builtins.mail.attachments import ( + _decode_mime_filename, + _sanitize_filename, + _save_attachment_to_storage, +) +from app.plugins.builtins.mail.models import ( + Mail, + MailAccount, + MailAttachment, + MailFolder, +) +from app.plugins.builtins.mail.sanitize import sanitize_html + +logger = logging.getLogger(__name__) + + +# German display names for standard IMAP folders. +# Keys are full IMAP paths (with dot delimiter) and also bare leaf names. +IMAP_FOLDER_NAME_MAP = { + "INBOX": "Posteingang", + "INBOX.Sent": "Gesendet", + "INBOX.Drafts": "Entwürfe", + "INBOX.Trash": "Papierkorb", + "INBOX.Archive": "Archiv", + "INBOX.spam": "Spam", + "INBOX.Spam": "Spam", + # Bare names (fallback for servers that don't nest under INBOX) + "Sent": "Gesendet", + "Sent Items": "Gesendet", + "Sent Mail": "Gesendet", + "Drafts": "Entwürfe", + "Draft": "Entwürfe", + "Spam": "Spam", + "Junk": "Spam", + "Junk Email": "Spam", + "Junk E-mail": "Spam", + "Trash": "Papierkorb", + "Deleted": "Papierkorb", + "Deleted Items": "Papierkorb", + "Archive": "Archiv", +} + +# Standard IMAP folders that are always considered "standard" +STANDARD_IMAP_FOLDERS = { + "INBOX", + "INBOX.Sent", "INBOX.Drafts", "INBOX.Trash", "INBOX.Archive", + "INBOX.spam", "INBOX.Spam", + "Sent", "Sent Items", "Sent Mail", + "Drafts", "Draft", + "Spam", "Junk", "Junk Email", "Junk E-mail", + "Trash", "Deleted", "Deleted Items", + "Archive", +} + +MAX_EMAILS_PER_FOLDER = 2000 + + +def _get_german_folder_name(imap_name: str) -> str: + """Return the German display name for a standard IMAP folder, or the original name.""" + if imap_name in IMAP_FOLDER_NAME_MAP: + return IMAP_FOLDER_NAME_MAP[imap_name] + # Try the leaf component (after last dot) for unknown nested folders + leaf = imap_name.rsplit(".", 1)[-1] if "." in imap_name else imap_name + if leaf in IMAP_FOLDER_NAME_MAP: + return IMAP_FOLDER_NAME_MAP[leaf] + return imap_name + + +def _parse_imap_list_response(response) -> tuple[list[tuple[str, str]], str]: + """Parse IMAP LIST response into list of (flags, folder_name) tuples. + + Handles both "/" and "." delimiters. The IMAP LIST response format is: + * LIST (\\HasChildren) "." "INBOX" + * LIST (\\HasNoChildren) "." "INBOX.Sent" + + Returns (folders, delimiter) where delimiter is the hierarchy separator + detected from the LIST response (defaults to '.' if not found). + """ + folders: list[tuple[str, str]] = [] + delimiter = '.' + lines = response.lines if hasattr(response, 'lines') else response + for line in lines: + if isinstance(line, (bytes, bytearray)): + text = line.decode('utf-8', errors='replace') + elif isinstance(line, str): + text = line + else: + continue + if 'LIST' not in text: + continue + # Extract quoted segments — the delimiter is the first quoted string, + # the folder name is the second. + parts = text.split('"') + if len(parts) >= 4: + delimiter = parts[1] + folder_name = parts[3] + flags = parts[0] if parts[0] else '' + folders.append((flags, folder_name)) + elif len(parts) >= 2: + folder_name = parts[-2] if len(parts) >= 2 else '' + if folder_name: + folders.append(('', folder_name)) + return folders, delimiter + + +def _build_folder_hierarchy( + imap_folders: list[tuple[str, str]], + account_id: uuid.UUID, + tenant_id: uuid.UUID, + existing_folders: dict[str, MailFolder], + delimiter: str = '.', + folder_mapping: dict[str, str | None] | None = None, +) -> list[MailFolder]: + """Create or update MailFolder records from IMAP LIST response. + + Handles delimiter-separated hierarchies (e.g. INBOX.Sent → parent=INBOX). + Updates existing folders in-place (name, is_standard, parent_id) so + that stale DB records with wrong imap_name values get corrected. + """ + result: list[MailFolder] = [] + + # Build reverse mapping: imap_name -> standard type (sent/drafts/spam/trash) + mapping_by_imap: dict[str, str] = {} + if folder_mapping: + for std_type, imap_name_val in folder_mapping.items(): + if imap_name_val: + mapping_by_imap[imap_name_val] = std_type + + # First pass: create or update folder records + for _flags, imap_name in imap_folders: + if not imap_name: + continue + + display_name = _get_german_folder_name(imap_name) + is_standard = imap_name in STANDARD_IMAP_FOLDERS + + # If account has explicit folder mapping, mark mapped folders as standard + if imap_name in mapping_by_imap: + is_standard = True + std_type = mapping_by_imap[imap_name] + if std_type == "sent": + display_name = "Gesendet" + elif std_type == "drafts": + display_name = "Entwürfe" + elif std_type == "spam": + display_name = "Spam" + elif std_type == "trash": + display_name = "Papierkorb" + + if imap_name in existing_folders: + folder = existing_folders[imap_name] + folder.name = display_name + folder.is_standard = is_standard + result.append(folder) + else: + folder = MailFolder( + tenant_id=tenant_id, + account_id=account_id, + name=display_name, + imap_name=imap_name, + is_standard=is_standard, + ) + result.append(folder) + + # Second pass: set parent_id based on IMAP hierarchy + folder_by_imap_name = {f.imap_name: f for f in result} + for folder in result: + if delimiter in folder.imap_name: + parts = folder.imap_name.split(delimiter) + parent_imap = delimiter.join(parts[:-1]) + if parent_imap in folder_by_imap_name: + parent = folder_by_imap_name[parent_imap] + folder.parent_id = parent.id # may be None for new folders; fixed after flush + else: + folder.parent_id = None + else: + folder.parent_id = None + + return result + + +def _parse_imap_quota_response(response) -> int | None: + """Parse an IMAP GETQUOTAROOT response and return usage percentage. + + Looks for lines like: + * QUOTA "INBOX" (STORAGE 12345 67890) + where 12345 is used and 67890 is limit. + Returns the usage percentage as an int, or None if parsing fails. + """ + try: + lines = response.lines if hasattr(response, "lines") else response + for line in lines: + if isinstance(line, (bytes, bytearray)): + line = line.decode("utf-8", errors="replace") + if not isinstance(line, str): + continue + if "QUOTA" not in line.upper(): + continue + # Extract the parenthesized storage values + # Pattern: (STORAGE ) + match = re.search(r"\(STORAGE\s+(\d+)\s+(\d+)\)", line, re.IGNORECASE) + if match: + used = int(match.group(1)) + limit = int(match.group(2)) + if limit > 0: + return int((used / limit) * 100) + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + return None + + +async def get_account_password(account: MailAccount) -> str: + """Decrypt and return the account password (internal use only). + + Uses per-account salt if available, falls back to legacy salt for old accounts. + """ + from app.plugins.builtins.mail.crypto import decrypt_password + + return decrypt_password(account.encrypted_password, account.password_salt or None) + + +async def imap_sync_folder( + db: AsyncSession, + folder_id: uuid.UUID, + tenant_id: uuid.UUID, +) -> dict: + """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__) + + folder = ( + await db.execute( + select(MailFolder).where( + and_(MailFolder.id == folder_id, MailFolder.tenant_id == tenant_id) + ) + ) + ).scalar_one_or_none() + if not folder: + return {"synced": 0, "error": "Folder not found"} + + account = ( + await db.execute( + select(MailAccount).where( + and_(MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id) + ) + ) + ).scalar_one_or_none() + if not account: + return {"synced": 0, "error": "Account not found"} + + 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: + client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) + await client.wait_hello_from_server() + await client.login(account.username, password) + + select_resp = await client.select(folder.imap_name) + if select_resp.result != 'OK': + return {"synced": 0, "error": f"Cannot select folder {folder.imap_name}"} + + # 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 = [] + + # 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) + + # Skip if we already have this UID (can happen with UID * search) + existing_mail = ( + await db.execute( + select(Mail).where( + and_( + Mail.account_id == account.id, + Mail.folder_id == folder.id, + Mail.imap_uid == uid_str, + Mail.tenant_id == tenant_id, + ) + ) + ) + ).scalar_one_or_none() + + # Also check by message_id — but ONLY within the same folder + # to prevent cross-folder moves. A mail in Trash and INBOX with + # the same message_id are separate copies (like real mail clients). + if not existing_mail: + # Fetch headers first to get Message-ID + fetch_hdr = await client.uid('fetch', uid_str, '(BODY.PEEK[HEADER.FIELDS (MESSAGE-ID)])') + hdr_raw = None + for line in (fetch_hdr.lines if hasattr(fetch_hdr, 'lines') else fetch_hdr): + if isinstance(line, bytearray): + hdr_raw = bytes(line) + break + if hdr_raw: + hdr_msg = message_from_bytes(hdr_raw) + peek_msg_id = hdr_msg.get("Message-ID", "") + if peek_msg_id: + existing_mail = ( + await db.execute( + select(Mail).where( + and_( + Mail.account_id == account.id, + Mail.message_id == peek_msg_id, + Mail.folder_id == folder.id, + Mail.tenant_id == tenant_id, + ) + ) + ) + ).scalar_one_or_none() + + if existing_mail: + # Update folder_id and imap_uid if mail moved to a different folder + if existing_mail.folder_id != folder.id: + existing_mail.folder_id = folder.id + existing_mail.imap_uid = uid_str + elif not existing_mail.imap_uid: + existing_mail.imap_uid = uid_str + continue + + # Fetch and parse the email + fetch_resp = await client.uid('fetch', uid_str, '(RFC822)') + raw_email = None + for line in (fetch_resp.lines if hasattr(fetch_resp, 'lines') else fetch_resp): + if isinstance(line, bytearray): + raw_email = bytes(line) + break + if raw_email is None: + continue + if isinstance(raw_email, str): + raw_email = raw_email.encode() + + msg = message_from_bytes(raw_email) + body_text = "" + body_html = "" + attachments = [] + + if msg.is_multipart(): + for part in msg.walk(): + ct = part.get_content_type() + if ct == "text/plain": + payload = part.get_payload(decode=True) + if payload: + body_text = payload.decode("utf-8", errors="replace") + elif ct == "text/html": + payload = part.get_payload(decode=True) + if payload: + body_html = payload.decode("utf-8", errors="replace") + elif part.get_filename(): + payload_bytes = part.get_payload(decode=True) or b"" + attachments.append({ + "filename": part.get_filename(), + "mime_type": ct, + "size": len(payload_bytes), + "content": payload_bytes, + }) + else: + ct = msg.get_content_type() + payload = msg.get_payload(decode=True) + if payload: + decoded = payload.decode("utf-8", errors="replace") + if ct == "text/html": + body_html = decoded + else: + body_text = decoded + + from email.header import decode_header, make_header + + def _decode_mime_header(value: str) -> str: + if not value: + return "" + try: + return str(make_header(decode_header(value))) + except Exception: + return value + + raw_msg_id = msg.get("Message-ID") + if raw_msg_id: + message_id = raw_msg_id + else: + message_id = f"generated-{account.id}-{folder.id}-{uid_str}" + subject = _decode_mime_header(msg.get("Subject", "")) + from_addr = _decode_mime_header(msg.get("From", "")) + to_addrs = _decode_mime_header(msg.get("To", "")) + cc_addrs = _decode_mime_header(msg.get("Cc", "")) + refs = msg.get("References", "") + in_reply_to = msg.get("In-Reply-To") + date_str = msg.get("Date", "") + + received_at = datetime.now(UTC) + if date_str: + try: + from email.utils import parsedate_to_datetime + parsed = parsedate_to_datetime(date_str) + if parsed: + received_at = parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC) + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + + thread_id = _compute_thread_id(message_id, refs, in_reply_to) + + mail = Mail( + tenant_id=tenant_id, + account_id=account.id, + folder_id=folder.id, + imap_uid=uid_str, + message_id=message_id, + thread_id=thread_id, + in_reply_to=in_reply_to, + references_header=refs, + subject=subject, + from_address=from_addr, + to_addresses=to_addrs, + cc_addresses=cc_addrs, + body_text=body_text, + body_html=body_html, + body_html_sanitized=sanitize_html(body_html), + has_attachments=len(attachments) > 0, + size_bytes=len(raw_email), + received_at=received_at, + ) + db.add(mail) + await db.flush() + synced_count += 1 + + # Publish mail.received event + from app.core.event_bus import get_event_bus + event_bus = get_event_bus() + await event_bus.publish('mail.received', { + 'mail_id': str(mail.id), + 'tenant_id': str(tenant_id), + 'account_id': str(account.id), + 'folder_id': str(folder.id), + 'subject': subject, + 'from_address': from_addr, + }) + + # Lifecycle hook: mail.after_receive + from app.core.hooks import do_action + await do_action("mail.after_receive", {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'account_id': str(account.id), 'folder_id': str(folder.id), 'subject': subject, 'from_address': from_addr}, db=db, tenant_id=tenant_id) + await do_action("mail.after_create", {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'subject': subject, 'from_address': from_addr}, db=db, tenant_id=tenant_id) + + # Outbox event: mail.received + from app.core.outbox import enqueue_outbox_event + await enqueue_outbox_event(db, tenant_id, 'mail.received', {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'account_id': str(account.id), 'folder_id': str(folder.id), 'subject': subject, 'from_address': from_addr}, aggregate_type='mail', aggregate_id=mail.id) + + # Save attachments + for att_data in attachments: + try: + raw_filename = att_data["filename"] or "attachment" + decoded_filename = _decode_mime_filename(raw_filename) + storage_path = await _save_attachment_to_storage( + mail.id, decoded_filename, att_data["content"] + ) + attachment = MailAttachment( + tenant_id=tenant_id, + mail_id=mail.id, + filename=_sanitize_filename(decoded_filename), + mime_type=att_data["mime_type"], + size_bytes=att_data["size"], + storage_path=storage_path, + ) + db.add(attachment) + except Exception as att_err: + logger.warning(f"Failed to save attachment for mail {mail.id}: {att_err}") + continue + + # Vanished-UID check: delete DB mails whose UID no longer exists on IMAP server + try: + vanished_search = await client.uid_search('ALL') + vanished_raw = vanished_search[1][0] if vanished_search[1] and vanished_search[1][0] else b'' + if isinstance(vanished_raw, (bytes, bytearray)): + imap_uids_set = {u.decode() if isinstance(u, bytes) else str(u) for u in vanished_raw.split()} + else: + imap_uids_set = set() + db_mails = ( + await db.execute( + select(Mail).where( + and_( + Mail.folder_id == folder_id, + Mail.tenant_id == tenant_id, + ) + ) + ) + ).scalars().all() + for db_mail in db_mails: + if db_mail.imap_uid: + if db_mail.imap_uid not in imap_uids_set: + logger.info("imap_sync_folder: deleting vanished mail %s (UID %s no longer on server)", db_mail.id, db_mail.imap_uid) + await db.delete(db_mail) + elif db_mail.message_id: + # Mail has no UID — try to find it on IMAP by Message-ID + try: + mid_search = await client.uid_search(f'HEADER Message-ID "{db_mail.message_id}"') + mid_raw = mid_search[1][0] if mid_search[1] and mid_search[1][0] else b'' + if isinstance(mid_raw, (bytes, bytearray)) and mid_raw: + found_uids = mid_raw.decode().split() + if found_uids: + db_mail.imap_uid = found_uids[0] + logger.info("imap_sync_folder: found UID %s for mail %s via Message-ID", found_uids[0], db_mail.id) + else: + logger.info("imap_sync_folder: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id) + await db.delete(db_mail) + else: + logger.info("imap_sync_folder: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id) + await db.delete(db_mail) + except Exception as mid_exc: + logger.warning("imap_sync_folder: Message-ID search failed for mail %s: %s", db_mail.id, mid_exc) + await db.flush() + except Exception as vanished_exc: + logger.warning("imap_sync_folder: vanished-UID check failed for folder %s: %s", folder_id, vanished_exc) + + # 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, + ) + ) + ) + ).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.is_seen.is_(False), + ) + ) + ) + ).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} + + except Exception as exc: + logger.warning("imap_sync_folder: failed for folder %s: %s", folder_id, exc) + return {"synced": 0, "error": str(exc)} + finally: + if client is not None: + try: + await client.logout() + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + + +async def imap_sync_account( + db: AsyncSession, + account_id: uuid.UUID, + tenant_id: uuid.UUID, +) -> dict: + """Sync mail folders and messages from IMAP server. + + Syncs ALL folders from the IMAP server (not just INBOX). + For each folder, fetches the last MAX_EMAILS_PER_FOLDER emails + (sorted by date descending) to avoid timeouts on large mailboxes. + Creates/updates mail_folders records with German display names. + """ + account = ( + await db.execute( + select(MailAccount).where( + and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id) + ) + ) + ).scalar_one_or_none() + if not account: + return {"synced": 0, "error": "Account not found"} + + if not account.is_active: + try: + await create_notification( + db, account.tenant_id, account.user_id, + "mail_account", + "Mail-Account deaktiviert", + f"Account {account.email_address} ist deaktiviert und wird nicht synchronisiert.", + ) + await db.flush() + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + return {"synced": 0, "error": "Account is not active"} + + password = await get_account_password(account) + + # ── IMAP connection ── + try: + client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) + await client.wait_hello_from_server() + except Exception as e: + try: + await create_notification( + db, account.tenant_id, account.user_id, + "mail_error", + "IMAP-Verbindung fehlgeschlagen", + f"Account {account.email_address}: {e}", + ) + await db.flush() + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + return {"synced": 0, "error": f"IMAP connection failed: {e}"} + + # ── IMAP login ── + try: + await client.login(account.username, password) + except Exception as e: + try: + await create_notification( + db, account.tenant_id, account.user_id, + "mail_auth", + "IMAP-Login fehlgeschlagen", + f"Account {account.email_address}: Passwort oder Anmeldedaten prüfen", + ) + await db.flush() + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + try: + await client.logout() + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + return {"synced": 0, "error": f"IMAP login failed: {e}"} + + # ── Quota check (non-critical, not all servers support QUOTA) ── + try: + quota_resp = await client.getquotaroot('INBOX') + usage_pct = _parse_imap_quota_response(quota_resp) + if usage_pct is not None and usage_pct > 80: + if usage_pct > 95: + quota_title = "Postfach voll – keine neuen Mails empfangbar" + else: + quota_title = "Postfach fast voll" + try: + await create_notification( + db, account.tenant_id, account.user_id, + "mail_quota", + quota_title, + f"Account {account.email_address}: {usage_pct}% belegt", + ) + await db.flush() + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + + try: + # 1) LIST all folders from IMAP server + list_response = await client.list('""', '"*"') + imap_folders, imap_delimiter = _parse_imap_list_response(list_response) + + # If LIST returned nothing, fall back to standard folders (dot-delimited) + if not imap_folders: + imap_folders = [ + ('', 'INBOX'), + ('', 'INBOX.Sent'), + ('', 'INBOX.Drafts'), + ('', 'INBOX.spam'), + ('', 'INBOX.Trash'), + ] + + # 2) Load existing folders from DB for this account + existing_db_folders = ( + await db.execute( + select(MailFolder).where( + and_(MailFolder.account_id == account.id, MailFolder.tenant_id == tenant_id) + ) + ) + ).scalars().all() + existing_by_imap: dict[str, MailFolder] = { + f.imap_name: f for f in existing_db_folders + } + + # 2a) Migrate stale folder names: if DB has 'Sent' but IMAP returns + # 'INBOX.Sent', update the DB record's imap_name so it matches. + imap_names_from_server = {name for _, name in imap_folders if name} + for db_folder in existing_db_folders: + if db_folder.imap_name not in imap_names_from_server: + # Try matching by leaf component using detected delimiter + leaf = db_folder.imap_name.rsplit(imap_delimiter, 1)[-1] + for srv_name in imap_names_from_server: + srv_leaf = srv_name.rsplit(imap_delimiter, 1)[-1] + if srv_leaf.lower() == leaf.lower(): + db_folder.imap_name = srv_name + existing_by_imap[srv_name] = db_folder + break + + # 3) Create/update folders in DB + folder_mapping = { + "sent": account.sent_folder_imap_name, + "drafts": account.drafts_folder_imap_name, + "spam": account.spam_folder_imap_name, + "trash": account.trash_folder_imap_name, + } + db_folders = _build_folder_hierarchy( + imap_folders, account.id, tenant_id, existing_by_imap, imap_delimiter, + folder_mapping=folder_mapping, + ) + for folder in db_folders: + if folder.id is None: + db.add(folder) + await db.flush() + + # 3a) Re-set parent_id now that new folders have IDs after flush + folder_by_imap_post_flush = {f.imap_name: f for f in db_folders} + for folder in db_folders: + if imap_delimiter in folder.imap_name: + parts = folder.imap_name.split(imap_delimiter) + parent_imap = imap_delimiter.join(parts[:-1]) + if parent_imap in folder_by_imap_post_flush: + parent = folder_by_imap_post_flush[parent_imap] + if parent.id: + folder.parent_id = parent.id + + await db.flush() + + # Build a map of imap_name -> folder_id for email sync + folder_by_imap = {f.imap_name: f for f in db_folders} + + synced_count = 0 + new_mails: list[dict] = [] + + # 4) Sync emails for each folder (limit to last 50 per folder) + for imap_name, folder in folder_by_imap.items(): + try: + # Select the folder on the IMAP server — use the raw IMAP + # name without extra quoting (aioimaplib handles it) + select_resp = await client.select(imap_name) + if select_resp.result != 'OK': + continue + + # Fetch UIDs and sort by date descending — 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 = [] + + # Limit to last MAX_EMAILS_PER_FOLDER UIDs + # UIDs are monotonically increasing, so the last ones are the newest + if len(uids) > MAX_EMAILS_PER_FOLDER: + uids = uids[-MAX_EMAILS_PER_FOLDER:] + + for uid in uids: + uid_str = uid.decode() if isinstance(uid, bytes) else str(uid) + fetch_resp = await client.uid('fetch', uid_str, '(RFC822)') + raw_email = None + for line in (fetch_resp.lines if hasattr(fetch_resp, 'lines') else fetch_resp): + if isinstance(line, bytearray): + raw_email = bytes(line) + break + if raw_email is None: + continue + if isinstance(raw_email, str): + raw_email = raw_email.encode() + + msg = message_from_bytes(raw_email) + body_text = "" + body_html = "" + attachments = [] + + if msg.is_multipart(): + for part in msg.walk(): + ct = part.get_content_type() + if ct == "text/plain": + payload = part.get_payload(decode=True) + if payload: + body_text = payload.decode("utf-8", errors="replace") + elif ct == "text/html": + payload = part.get_payload(decode=True) + if payload: + body_html = payload.decode("utf-8", errors="replace") + elif part.get_filename(): + payload_bytes = part.get_payload(decode=True) or b"" + content_id = part.get("Content-ID", None) + if content_id: + content_id = content_id.strip("<>") + attachments.append( + { + "filename": part.get_filename(), + "mime_type": ct, + "size": len(payload_bytes), + "content": payload_bytes, + "content_id": content_id, + } + ) + else: + ct = msg.get_content_type() + payload = msg.get_payload(decode=True) + if payload: + decoded = payload.decode("utf-8", errors="replace") + if ct == "text/html": + body_html = decoded + else: + body_text = decoded + + from email.header import decode_header, make_header + + def _decode_mime_header(value: str) -> str: + """Decode MIME encoded-words (=?UTF-8?Q?...?=) to readable text.""" + if not value: + return "" + try: + return str(make_header(decode_header(value))) + except Exception: + return value + + raw_msg_id = msg.get("Message-ID") + if raw_msg_id: + message_id = raw_msg_id + else: + # Deterministic ID so re-syncs don't create duplicates + message_id = f"generated-{account.id}-{folder.id}-{uid_str}" + subject = _decode_mime_header(msg.get("Subject", "")) + from_addr = _decode_mime_header(msg.get("From", "")) + to_addrs = _decode_mime_header(msg.get("To", "")) + cc_addrs = _decode_mime_header(msg.get("Cc", "")) + refs = msg.get("References", "") + in_reply_to = msg.get("In-Reply-To") + date_str = msg.get("Date", "") + + # Parse date for received_at + received_at = datetime.now(UTC) + if date_str: + try: + from email.utils import parsedate_to_datetime + parsed = parsedate_to_datetime(date_str) + if parsed: + received_at = parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC) + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + + # Compute thread_id from References/In-Reply-To + thread_id = _compute_thread_id(message_id, refs, in_reply_to) + + # Dedup: first check by (account_id, folder_id, imap_uid), + # then fall back to (account_id, message_id) for mails synced + # before the imap_uid column existed. + existing_mail = ( + await db.execute( + select(Mail).where( + and_( + Mail.account_id == account.id, + Mail.folder_id == folder.id, + Mail.imap_uid == uid_str, + Mail.tenant_id == tenant_id, + ) + ) + ) + ).scalar_one_or_none() + + if not existing_mail: + existing_mail = ( + await db.execute( + select(Mail).where( + and_( + Mail.account_id == account.id, + Mail.message_id == message_id, + Mail.folder_id == folder.id, + Mail.tenant_id == tenant_id, + ) + ) + ) + ).scalar_one_or_none() + + if existing_mail: + # 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: + existing_mail.folder_id = folder.id + existing_mail.imap_uid = uid_str + elif not existing_mail.imap_uid: + existing_mail.imap_uid = uid_str + + + # Save attachments for existing emails that have has_attachments but no records + if attachments and existing_mail.has_attachments: + existing_att_count = ( + await db.execute( + select(func.count()).select_from(MailAttachment).where( + and_( + MailAttachment.mail_id == existing_mail.id, + MailAttachment.tenant_id == tenant_id, + ) + ) + ) + ).scalar() or 0 + if existing_att_count == 0: + for att_data in attachments: + try: + raw_filename = att_data["filename"] or "attachment" + decoded_filename = _decode_mime_filename(raw_filename) + storage_path = await _save_attachment_to_storage( + existing_mail.id, decoded_filename, att_data["content"] + ) + attachment = MailAttachment( + tenant_id=tenant_id, + mail_id=existing_mail.id, + filename=_sanitize_filename(decoded_filename), + mime_type=att_data["mime_type"], + size_bytes=att_data["size"], + storage_path=storage_path, + content_id=att_data.get("content_id"), + ) + db.add(attachment) + except Exception as att_err: + logger.warning(f"Failed to save attachment for existing mail {existing_mail.id}: {att_err}") + continue + await db.flush() + continue + + mail = Mail( + tenant_id=tenant_id, + account_id=account.id, + folder_id=folder.id, + imap_uid=uid_str, + message_id=message_id, + thread_id=thread_id, + in_reply_to=in_reply_to, + references_header=refs, + subject=subject, + from_address=from_addr, + to_addresses=to_addrs, + cc_addresses=cc_addrs, + body_text=body_text, + body_html=body_html, + body_html_sanitized=sanitize_html(body_html), + has_attachments=len(attachments) > 0, + size_bytes=len(raw_email), + received_at=received_at, + ) + db.add(mail) + await db.flush() + synced_count += 1 + + # Collect for new-mail notifications + new_mails.append({ + "from": from_addr, + "subject": subject, + }) + + # Save attachments to storage and DB + for att_data in attachments: + try: + # Decode MIME-encoded filenames (e.g. =?utf-8?q?...?=) + raw_filename = att_data["filename"] or "attachment" + decoded_filename = _decode_mime_filename(raw_filename) + storage_path = await _save_attachment_to_storage( + mail.id, decoded_filename, att_data["content"] + ) + attachment = MailAttachment( + tenant_id=tenant_id, + mail_id=mail.id, + filename=_sanitize_filename(decoded_filename), + mime_type=att_data["mime_type"], + size_bytes=att_data["size"], + storage_path=storage_path, + content_id=att_data.get("content_id"), + ) + db.add(attachment) + except Exception as att_err: + logger.warning(f"Failed to save attachment for new mail {mail.id}: {att_err}") + continue + await db.flush() + + # Vanished-UID check: delete DB mails whose UID no longer exists on IMAP server + try: + vanished_search = await client.uid_search('ALL') + vanished_raw = vanished_search[1][0] if vanished_search[1] and vanished_search[1][0] else b'' + if isinstance(vanished_raw, (bytes, bytearray)): + imap_uids_set = {u.decode() if isinstance(u, bytes) else str(u) for u in vanished_raw.split()} + else: + imap_uids_set = set() + db_mails = ( + await db.execute( + select(Mail).where( + and_( + Mail.folder_id == folder.id, + Mail.tenant_id == tenant_id, + ) + ) + ) + ).scalars().all() + for db_mail in db_mails: + if db_mail.imap_uid: + if db_mail.imap_uid not in imap_uids_set: + logger.info("imap_sync_account: deleting vanished mail %s (UID %s no longer on server)", db_mail.id, db_mail.imap_uid) + await db.delete(db_mail) + elif db_mail.message_id: + # Mail has no UID — try to find it on IMAP by Message-ID + try: + mid_search = await client.uid_search(f'HEADER Message-ID "{db_mail.message_id}"') + mid_raw = mid_search[1][0] if mid_search[1] and mid_search[1][0] else b'' + if isinstance(mid_raw, (bytes, bytearray)) and mid_raw: + found_uids = mid_raw.decode().split() + if found_uids: + db_mail.imap_uid = found_uids[0] + logger.info("imap_sync_account: found UID %s for mail %s via Message-ID", found_uids[0], db_mail.id) + else: + logger.info("imap_sync_account: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id) + await db.delete(db_mail) + else: + logger.info("imap_sync_account: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id) + await db.delete(db_mail) + except Exception as mid_exc: + logger.warning("imap_sync_account: Message-ID search failed for mail %s: %s", db_mail.id, mid_exc) + await db.flush() + except Exception as vanished_exc: + logger.warning("imap_sync_account: vanished-UID check failed for folder %s: %s", folder.imap_name, vanished_exc) + + # 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) + ) + ) + ).scalar() + unread = ( + await db.execute( + select(func.count()).select_from(Mail).where( + and_( + Mail.folder_id == folder.id, + Mail.tenant_id == tenant_id, + Mail.is_seen.is_(False), + ) + ) + ) + ).scalar() + folder.total_count = total or 0 + folder.unread_count = unread or 0 + + await db.flush() + + except Exception: + # Skip folders that can't be selected (e.g. no select permission) + continue + + # ── New-mail notifications (max 10, then summary) ── + if new_mails: + try: + if len(new_mails) <= 10: + for nm in new_mails: + try: + await create_notification( + db, account.tenant_id, account.user_id, + "mail_new", + f"Neue E-Mail von {nm['from']}", + nm["subject"], + ) + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + else: + try: + await create_notification( + db, account.tenant_id, account.user_id, + "mail_new", + f"Neue E-Mails: {len(new_mails)} neue Nachrichten", + f"Account {account.email_address} hat {len(new_mails)} neue E-Mails empfangen.", + ) + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + await db.flush() + except Exception: + logger.debug("Ignored exception in mail service", exc_info=True) + + await db.flush() + await client.logout() + return {"synced": synced_count} + except Exception as e: + return {"synced": 0, "error": str(e)} + + +def _compute_thread_id(message_id: str, references: str, in_reply_to: str | None) -> str: + """Compute thread ID from References/In-Reply-To headers (F-MAIL-05).""" + ref_parts: list[str] = [] + if references: + ref_parts = [r.strip() for r in references.split() if r.strip()] + if in_reply_to and in_reply_to.strip() not in ref_parts: + ref_parts.append(in_reply_to.strip()) + if ref_parts: + return ref_parts[0] + return message_id or str(uuid.uuid4()) + + +# ─── SMTP Send Service (F-MAIL-02) ─── diff --git a/app/plugins/builtins/mail/services.py b/app/plugins/builtins/mail/services.py index 5acdf5b..24b348f 100644 --- a/app/plugins/builtins/mail/services.py +++ b/app/plugins/builtins/mail/services.py @@ -8,14 +8,13 @@ import os import re import uuid from datetime import UTC, datetime, timedelta -from email import message_from_bytes from email.message import EmailMessage from email.utils import formataddr, formatdate, make_msgid import aiofiles import aioimaplib import aiosmtplib -from sqlalchemy import and_, func, or_, select +from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings @@ -24,15 +23,23 @@ from app.core.notifications import create_notification # ── Re-exports from extracted sub-modules (BUG-018 pilot split) ── # Backwards compatibility: all consumers still import from # ``app.plugins.builtins.mail.services`` unchanged. -from app.plugins.builtins.mail.crypto import ( # noqa: E402 +from app.plugins.builtins.mail.crypto import ( # noqa: E402,F401 decrypt_password, encrypt_password, generate_salt, ) +from app.plugins.builtins.mail.imap_sync import ( # noqa: E402,F401 + _compute_thread_id, + _get_german_folder_name, + _parse_imap_list_response, + _parse_imap_quota_response, + get_account_password, + imap_sync_account, + imap_sync_folder, +) from app.plugins.builtins.mail.models import ( Mail, MailAccount, - MailAttachment, MailFolder, MailLabelAssignment, MailRule, @@ -139,36 +146,6 @@ async def _save_attachment_to_storage( # ─── IMAP Quota Parser ─── -def _parse_imap_quota_response(response) -> int | None: - """Parse an IMAP GETQUOTAROOT response and return usage percentage. - - Looks for lines like: - * QUOTA "INBOX" (STORAGE 12345 67890) - where 12345 is used and 67890 is limit. - Returns the usage percentage as an int, or None if parsing fails. - """ - try: - lines = response.lines if hasattr(response, "lines") else response - for line in lines: - if isinstance(line, (bytes, bytearray)): - line = line.decode("utf-8", errors="replace") - if not isinstance(line, str): - continue - if "QUOTA" not in line.upper(): - continue - # Extract the parenthesized storage values - # Pattern: (STORAGE ) - match = re.search(r"\(STORAGE\s+(\d+)\s+(\d+)\)", line, re.IGNORECASE) - if match: - used = int(match.group(1)) - limit = int(match.group(2)) - if limit > 0: - return int((used / limit) * 100) - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - return None - - # ─── Mail Account Service ─── @@ -265,1086 +242,6 @@ async def update_mail_account(db: AsyncSession, account: MailAccount, data: dict return account -async def get_account_password(account: MailAccount) -> str: - """Decrypt and return the account password (internal use only). - - Uses per-account salt if available, falls back to legacy salt for old accounts. - """ - return decrypt_password(account.encrypted_password, account.password_salt or None) - - -# ─── IMAP Sync Service (F-MAIL-01) ─── - - -# German display names for standard IMAP folders. -# Keys are full IMAP paths (with dot delimiter) and also bare leaf names. -IMAP_FOLDER_NAME_MAP = { - "INBOX": "Posteingang", - "INBOX.Sent": "Gesendet", - "INBOX.Drafts": "Entwürfe", - "INBOX.Trash": "Papierkorb", - "INBOX.Archive": "Archiv", - "INBOX.spam": "Spam", - "INBOX.Spam": "Spam", - # Bare names (fallback for servers that don't nest under INBOX) - "Sent": "Gesendet", - "Sent Items": "Gesendet", - "Sent Mail": "Gesendet", - "Drafts": "Entwürfe", - "Draft": "Entwürfe", - "Spam": "Spam", - "Junk": "Spam", - "Junk Email": "Spam", - "Junk E-mail": "Spam", - "Trash": "Papierkorb", - "Deleted": "Papierkorb", - "Deleted Items": "Papierkorb", - "Archive": "Archiv", -} - -# Standard IMAP folders that are always considered "standard" -STANDARD_IMAP_FOLDERS = { - "INBOX", - "INBOX.Sent", "INBOX.Drafts", "INBOX.Trash", "INBOX.Archive", - "INBOX.spam", "INBOX.Spam", - "Sent", "Sent Items", "Sent Mail", - "Drafts", "Draft", - "Spam", "Junk", "Junk Email", "Junk E-mail", - "Trash", "Deleted", "Deleted Items", - "Archive", -} - -MAX_EMAILS_PER_FOLDER = 2000 - - -def _get_german_folder_name(imap_name: str) -> str: - """Return the German display name for a standard IMAP folder, or the original name.""" - if imap_name in IMAP_FOLDER_NAME_MAP: - return IMAP_FOLDER_NAME_MAP[imap_name] - # Try the leaf component (after last dot) for unknown nested folders - leaf = imap_name.rsplit(".", 1)[-1] if "." in imap_name else imap_name - if leaf in IMAP_FOLDER_NAME_MAP: - return IMAP_FOLDER_NAME_MAP[leaf] - return imap_name - - -def _parse_imap_list_response(response) -> tuple[list[tuple[str, str]], str]: - """Parse IMAP LIST response into list of (flags, folder_name) tuples. - - Handles both "/" and "." delimiters. The IMAP LIST response format is: - * LIST (\\HasChildren) "." "INBOX" - * LIST (\\HasNoChildren) "." "INBOX.Sent" - - Returns (folders, delimiter) where delimiter is the hierarchy separator - detected from the LIST response (defaults to '.' if not found). - """ - folders: list[tuple[str, str]] = [] - delimiter = '.' - lines = response.lines if hasattr(response, 'lines') else response - for line in lines: - if isinstance(line, (bytes, bytearray)): - text = line.decode('utf-8', errors='replace') - elif isinstance(line, str): - text = line - else: - continue - if 'LIST' not in text: - continue - # Extract quoted segments — the delimiter is the first quoted string, - # the folder name is the second. - parts = text.split('"') - if len(parts) >= 4: - delimiter = parts[1] - folder_name = parts[3] - flags = parts[0] if parts[0] else '' - folders.append((flags, folder_name)) - elif len(parts) >= 2: - folder_name = parts[-2] if len(parts) >= 2 else '' - if folder_name: - folders.append(('', folder_name)) - return folders, delimiter - - -def _build_folder_hierarchy( - imap_folders: list[tuple[str, str]], - account_id: uuid.UUID, - tenant_id: uuid.UUID, - existing_folders: dict[str, MailFolder], - delimiter: str = '.', - folder_mapping: dict[str, str | None] | None = None, -) -> list[MailFolder]: - """Create or update MailFolder records from IMAP LIST response. - - Handles delimiter-separated hierarchies (e.g. INBOX.Sent → parent=INBOX). - Updates existing folders in-place (name, is_standard, parent_id) so - that stale DB records with wrong imap_name values get corrected. - """ - result: list[MailFolder] = [] - - # Build reverse mapping: imap_name -> standard type (sent/drafts/spam/trash) - mapping_by_imap: dict[str, str] = {} - if folder_mapping: - for std_type, imap_name_val in folder_mapping.items(): - if imap_name_val: - mapping_by_imap[imap_name_val] = std_type - - # First pass: create or update folder records - for _flags, imap_name in imap_folders: - if not imap_name: - continue - - display_name = _get_german_folder_name(imap_name) - is_standard = imap_name in STANDARD_IMAP_FOLDERS - - # If account has explicit folder mapping, mark mapped folders as standard - if imap_name in mapping_by_imap: - is_standard = True - std_type = mapping_by_imap[imap_name] - if std_type == "sent": - display_name = "Gesendet" - elif std_type == "drafts": - display_name = "Entwürfe" - elif std_type == "spam": - display_name = "Spam" - elif std_type == "trash": - display_name = "Papierkorb" - - if imap_name in existing_folders: - folder = existing_folders[imap_name] - folder.name = display_name - folder.is_standard = is_standard - result.append(folder) - else: - folder = MailFolder( - tenant_id=tenant_id, - account_id=account_id, - name=display_name, - imap_name=imap_name, - is_standard=is_standard, - ) - result.append(folder) - - # Second pass: set parent_id based on IMAP hierarchy - folder_by_imap_name = {f.imap_name: f for f in result} - for folder in result: - if delimiter in folder.imap_name: - parts = folder.imap_name.split(delimiter) - parent_imap = delimiter.join(parts[:-1]) - if parent_imap in folder_by_imap_name: - parent = folder_by_imap_name[parent_imap] - folder.parent_id = parent.id # may be None for new folders; fixed after flush - else: - folder.parent_id = None - else: - folder.parent_id = None - - return result - - -async def imap_sync_folder( - db: AsyncSession, - folder_id: uuid.UUID, - tenant_id: uuid.UUID, -) -> dict: - """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__) - - folder = ( - await db.execute( - select(MailFolder).where( - and_(MailFolder.id == folder_id, MailFolder.tenant_id == tenant_id) - ) - ) - ).scalar_one_or_none() - if not folder: - return {"synced": 0, "error": "Folder not found"} - - account = ( - await db.execute( - select(MailAccount).where( - and_(MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id) - ) - ) - ).scalar_one_or_none() - if not account: - return {"synced": 0, "error": "Account not found"} - - 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: - client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) - await client.wait_hello_from_server() - await client.login(account.username, password) - - select_resp = await client.select(folder.imap_name) - if select_resp.result != 'OK': - return {"synced": 0, "error": f"Cannot select folder {folder.imap_name}"} - - # 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 = [] - - # 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) - - # Skip if we already have this UID (can happen with UID * search) - existing_mail = ( - await db.execute( - select(Mail).where( - and_( - Mail.account_id == account.id, - Mail.folder_id == folder.id, - Mail.imap_uid == uid_str, - Mail.tenant_id == tenant_id, - ) - ) - ) - ).scalar_one_or_none() - - # Also check by message_id — but ONLY within the same folder - # to prevent cross-folder moves. A mail in Trash and INBOX with - # the same message_id are separate copies (like real mail clients). - if not existing_mail: - # Fetch headers first to get Message-ID - fetch_hdr = await client.uid('fetch', uid_str, '(BODY.PEEK[HEADER.FIELDS (MESSAGE-ID)])') - hdr_raw = None - for line in (fetch_hdr.lines if hasattr(fetch_hdr, 'lines') else fetch_hdr): - if isinstance(line, bytearray): - hdr_raw = bytes(line) - break - if hdr_raw: - hdr_msg = message_from_bytes(hdr_raw) - peek_msg_id = hdr_msg.get("Message-ID", "") - if peek_msg_id: - existing_mail = ( - await db.execute( - select(Mail).where( - and_( - Mail.account_id == account.id, - Mail.message_id == peek_msg_id, - Mail.folder_id == folder.id, - Mail.tenant_id == tenant_id, - ) - ) - ) - ).scalar_one_or_none() - - if existing_mail: - # Update folder_id and imap_uid if mail moved to a different folder - if existing_mail.folder_id != folder.id: - existing_mail.folder_id = folder.id - existing_mail.imap_uid = uid_str - elif not existing_mail.imap_uid: - existing_mail.imap_uid = uid_str - continue - - # Fetch and parse the email - fetch_resp = await client.uid('fetch', uid_str, '(RFC822)') - raw_email = None - for line in (fetch_resp.lines if hasattr(fetch_resp, 'lines') else fetch_resp): - if isinstance(line, bytearray): - raw_email = bytes(line) - break - if raw_email is None: - continue - if isinstance(raw_email, str): - raw_email = raw_email.encode() - - msg = message_from_bytes(raw_email) - body_text = "" - body_html = "" - attachments = [] - - if msg.is_multipart(): - for part in msg.walk(): - ct = part.get_content_type() - if ct == "text/plain": - payload = part.get_payload(decode=True) - if payload: - body_text = payload.decode("utf-8", errors="replace") - elif ct == "text/html": - payload = part.get_payload(decode=True) - if payload: - body_html = payload.decode("utf-8", errors="replace") - elif part.get_filename(): - payload_bytes = part.get_payload(decode=True) or b"" - attachments.append({ - "filename": part.get_filename(), - "mime_type": ct, - "size": len(payload_bytes), - "content": payload_bytes, - }) - else: - ct = msg.get_content_type() - payload = msg.get_payload(decode=True) - if payload: - decoded = payload.decode("utf-8", errors="replace") - if ct == "text/html": - body_html = decoded - else: - body_text = decoded - - from email.header import decode_header, make_header - - def _decode_mime_header(value: str) -> str: - if not value: - return "" - try: - return str(make_header(decode_header(value))) - except Exception: - return value - - raw_msg_id = msg.get("Message-ID") - if raw_msg_id: - message_id = raw_msg_id - else: - message_id = f"generated-{account.id}-{folder.id}-{uid_str}" - subject = _decode_mime_header(msg.get("Subject", "")) - from_addr = _decode_mime_header(msg.get("From", "")) - to_addrs = _decode_mime_header(msg.get("To", "")) - cc_addrs = _decode_mime_header(msg.get("Cc", "")) - refs = msg.get("References", "") - in_reply_to = msg.get("In-Reply-To") - date_str = msg.get("Date", "") - - received_at = datetime.now(UTC) - if date_str: - try: - from email.utils import parsedate_to_datetime - parsed = parsedate_to_datetime(date_str) - if parsed: - received_at = parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC) - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - - thread_id = _compute_thread_id(message_id, refs, in_reply_to) - - mail = Mail( - tenant_id=tenant_id, - account_id=account.id, - folder_id=folder.id, - imap_uid=uid_str, - message_id=message_id, - thread_id=thread_id, - in_reply_to=in_reply_to, - references_header=refs, - subject=subject, - from_address=from_addr, - to_addresses=to_addrs, - cc_addresses=cc_addrs, - body_text=body_text, - body_html=body_html, - body_html_sanitized=sanitize_html(body_html), - has_attachments=len(attachments) > 0, - size_bytes=len(raw_email), - received_at=received_at, - ) - db.add(mail) - await db.flush() - synced_count += 1 - - # Publish mail.received event - from app.core.event_bus import get_event_bus - event_bus = get_event_bus() - await event_bus.publish('mail.received', { - 'mail_id': str(mail.id), - 'tenant_id': str(tenant_id), - 'account_id': str(account.id), - 'folder_id': str(folder.id), - 'subject': subject, - 'from_address': from_addr, - }) - - # Lifecycle hook: mail.after_receive - from app.core.hooks import do_action - await do_action("mail.after_receive", {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'account_id': str(account.id), 'folder_id': str(folder.id), 'subject': subject, 'from_address': from_addr}, db=db, tenant_id=tenant_id) - await do_action("mail.after_create", {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'subject': subject, 'from_address': from_addr}, db=db, tenant_id=tenant_id) - - # Outbox event: mail.received - from app.core.outbox import enqueue_outbox_event - await enqueue_outbox_event(db, tenant_id, 'mail.received', {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'account_id': str(account.id), 'folder_id': str(folder.id), 'subject': subject, 'from_address': from_addr}, aggregate_type='mail', aggregate_id=mail.id) - - # Save attachments - for att_data in attachments: - try: - raw_filename = att_data["filename"] or "attachment" - decoded_filename = _decode_mime_filename(raw_filename) - storage_path = await _save_attachment_to_storage( - mail.id, decoded_filename, att_data["content"] - ) - attachment = MailAttachment( - tenant_id=tenant_id, - mail_id=mail.id, - filename=_sanitize_filename(decoded_filename), - mime_type=att_data["mime_type"], - size_bytes=att_data["size"], - storage_path=storage_path, - ) - db.add(attachment) - except Exception as att_err: - logger.warning(f"Failed to save attachment for mail {mail.id}: {att_err}") - continue - - # Vanished-UID check: delete DB mails whose UID no longer exists on IMAP server - try: - vanished_search = await client.uid_search('ALL') - vanished_raw = vanished_search[1][0] if vanished_search[1] and vanished_search[1][0] else b'' - if isinstance(vanished_raw, (bytes, bytearray)): - imap_uids_set = {u.decode() if isinstance(u, bytes) else str(u) for u in vanished_raw.split()} - else: - imap_uids_set = set() - db_mails = ( - await db.execute( - select(Mail).where( - and_( - Mail.folder_id == folder_id, - Mail.tenant_id == tenant_id, - ) - ) - ) - ).scalars().all() - for db_mail in db_mails: - if db_mail.imap_uid: - if db_mail.imap_uid not in imap_uids_set: - logger.info("imap_sync_folder: deleting vanished mail %s (UID %s no longer on server)", db_mail.id, db_mail.imap_uid) - await db.delete(db_mail) - elif db_mail.message_id: - # Mail has no UID — try to find it on IMAP by Message-ID - try: - mid_search = await client.uid_search(f'HEADER Message-ID "{db_mail.message_id}"') - mid_raw = mid_search[1][0] if mid_search[1] and mid_search[1][0] else b'' - if isinstance(mid_raw, (bytes, bytearray)) and mid_raw: - found_uids = mid_raw.decode().split() - if found_uids: - db_mail.imap_uid = found_uids[0] - logger.info("imap_sync_folder: found UID %s for mail %s via Message-ID", found_uids[0], db_mail.id) - else: - logger.info("imap_sync_folder: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id) - await db.delete(db_mail) - else: - logger.info("imap_sync_folder: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id) - await db.delete(db_mail) - except Exception as mid_exc: - logger.warning("imap_sync_folder: Message-ID search failed for mail %s: %s", db_mail.id, mid_exc) - await db.flush() - except Exception as vanished_exc: - logger.warning("imap_sync_folder: vanished-UID check failed for folder %s: %s", folder_id, vanished_exc) - - # 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, - ) - ) - ) - ).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.is_seen.is_(False), - ) - ) - ) - ).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} - - except Exception as exc: - logger.warning("imap_sync_folder: failed for folder %s: %s", folder_id, exc) - return {"synced": 0, "error": str(exc)} - finally: - if client is not None: - try: - await client.logout() - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - - -async def imap_sync_account( - db: AsyncSession, - account_id: uuid.UUID, - tenant_id: uuid.UUID, -) -> dict: - """Sync mail folders and messages from IMAP server. - - Syncs ALL folders from the IMAP server (not just INBOX). - For each folder, fetches the last MAX_EMAILS_PER_FOLDER emails - (sorted by date descending) to avoid timeouts on large mailboxes. - Creates/updates mail_folders records with German display names. - """ - account = ( - await db.execute( - select(MailAccount).where( - and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id) - ) - ) - ).scalar_one_or_none() - if not account: - return {"synced": 0, "error": "Account not found"} - - if not account.is_active: - try: - await create_notification( - db, account.tenant_id, account.user_id, - "mail_account", - "Mail-Account deaktiviert", - f"Account {account.email_address} ist deaktiviert und wird nicht synchronisiert.", - ) - await db.flush() - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - return {"synced": 0, "error": "Account is not active"} - - password = await get_account_password(account) - - # ── IMAP connection ── - try: - client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) - await client.wait_hello_from_server() - except Exception as e: - try: - await create_notification( - db, account.tenant_id, account.user_id, - "mail_error", - "IMAP-Verbindung fehlgeschlagen", - f"Account {account.email_address}: {e}", - ) - await db.flush() - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - return {"synced": 0, "error": f"IMAP connection failed: {e}"} - - # ── IMAP login ── - try: - await client.login(account.username, password) - except Exception as e: - try: - await create_notification( - db, account.tenant_id, account.user_id, - "mail_auth", - "IMAP-Login fehlgeschlagen", - f"Account {account.email_address}: Passwort oder Anmeldedaten prüfen", - ) - await db.flush() - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - try: - await client.logout() - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - return {"synced": 0, "error": f"IMAP login failed: {e}"} - - # ── Quota check (non-critical, not all servers support QUOTA) ── - try: - quota_resp = await client.getquotaroot('INBOX') - usage_pct = _parse_imap_quota_response(quota_resp) - if usage_pct is not None and usage_pct > 80: - if usage_pct > 95: - quota_title = "Postfach voll – keine neuen Mails empfangbar" - else: - quota_title = "Postfach fast voll" - try: - await create_notification( - db, account.tenant_id, account.user_id, - "mail_quota", - quota_title, - f"Account {account.email_address}: {usage_pct}% belegt", - ) - await db.flush() - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - - try: - # 1) LIST all folders from IMAP server - list_response = await client.list('""', '"*"') - imap_folders, imap_delimiter = _parse_imap_list_response(list_response) - - # If LIST returned nothing, fall back to standard folders (dot-delimited) - if not imap_folders: - imap_folders = [ - ('', 'INBOX'), - ('', 'INBOX.Sent'), - ('', 'INBOX.Drafts'), - ('', 'INBOX.spam'), - ('', 'INBOX.Trash'), - ] - - # 2) Load existing folders from DB for this account - existing_db_folders = ( - await db.execute( - select(MailFolder).where( - and_(MailFolder.account_id == account.id, MailFolder.tenant_id == tenant_id) - ) - ) - ).scalars().all() - existing_by_imap: dict[str, MailFolder] = { - f.imap_name: f for f in existing_db_folders - } - - # 2a) Migrate stale folder names: if DB has 'Sent' but IMAP returns - # 'INBOX.Sent', update the DB record's imap_name so it matches. - imap_names_from_server = {name for _, name in imap_folders if name} - for db_folder in existing_db_folders: - if db_folder.imap_name not in imap_names_from_server: - # Try matching by leaf component using detected delimiter - leaf = db_folder.imap_name.rsplit(imap_delimiter, 1)[-1] - for srv_name in imap_names_from_server: - srv_leaf = srv_name.rsplit(imap_delimiter, 1)[-1] - if srv_leaf.lower() == leaf.lower(): - db_folder.imap_name = srv_name - existing_by_imap[srv_name] = db_folder - break - - # 3) Create/update folders in DB - folder_mapping = { - "sent": account.sent_folder_imap_name, - "drafts": account.drafts_folder_imap_name, - "spam": account.spam_folder_imap_name, - "trash": account.trash_folder_imap_name, - } - db_folders = _build_folder_hierarchy( - imap_folders, account.id, tenant_id, existing_by_imap, imap_delimiter, - folder_mapping=folder_mapping, - ) - for folder in db_folders: - if folder.id is None: - db.add(folder) - await db.flush() - - # 3a) Re-set parent_id now that new folders have IDs after flush - folder_by_imap_post_flush = {f.imap_name: f for f in db_folders} - for folder in db_folders: - if imap_delimiter in folder.imap_name: - parts = folder.imap_name.split(imap_delimiter) - parent_imap = imap_delimiter.join(parts[:-1]) - if parent_imap in folder_by_imap_post_flush: - parent = folder_by_imap_post_flush[parent_imap] - if parent.id: - folder.parent_id = parent.id - - await db.flush() - - # Build a map of imap_name -> folder_id for email sync - folder_by_imap = {f.imap_name: f for f in db_folders} - - synced_count = 0 - new_mails: list[dict] = [] - - # 4) Sync emails for each folder (limit to last 50 per folder) - for imap_name, folder in folder_by_imap.items(): - try: - # Select the folder on the IMAP server — use the raw IMAP - # name without extra quoting (aioimaplib handles it) - select_resp = await client.select(imap_name) - if select_resp.result != 'OK': - continue - - # Fetch UIDs and sort by date descending — 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 = [] - - # Limit to last MAX_EMAILS_PER_FOLDER UIDs - # UIDs are monotonically increasing, so the last ones are the newest - if len(uids) > MAX_EMAILS_PER_FOLDER: - uids = uids[-MAX_EMAILS_PER_FOLDER:] - - for uid in uids: - uid_str = uid.decode() if isinstance(uid, bytes) else str(uid) - fetch_resp = await client.uid('fetch', uid_str, '(RFC822)') - raw_email = None - for line in (fetch_resp.lines if hasattr(fetch_resp, 'lines') else fetch_resp): - if isinstance(line, bytearray): - raw_email = bytes(line) - break - if raw_email is None: - continue - if isinstance(raw_email, str): - raw_email = raw_email.encode() - - msg = message_from_bytes(raw_email) - body_text = "" - body_html = "" - attachments = [] - - if msg.is_multipart(): - for part in msg.walk(): - ct = part.get_content_type() - if ct == "text/plain": - payload = part.get_payload(decode=True) - if payload: - body_text = payload.decode("utf-8", errors="replace") - elif ct == "text/html": - payload = part.get_payload(decode=True) - if payload: - body_html = payload.decode("utf-8", errors="replace") - elif part.get_filename(): - payload_bytes = part.get_payload(decode=True) or b"" - content_id = part.get("Content-ID", None) - if content_id: - content_id = content_id.strip("<>") - attachments.append( - { - "filename": part.get_filename(), - "mime_type": ct, - "size": len(payload_bytes), - "content": payload_bytes, - "content_id": content_id, - } - ) - else: - ct = msg.get_content_type() - payload = msg.get_payload(decode=True) - if payload: - decoded = payload.decode("utf-8", errors="replace") - if ct == "text/html": - body_html = decoded - else: - body_text = decoded - - from email.header import decode_header, make_header - - def _decode_mime_header(value: str) -> str: - """Decode MIME encoded-words (=?UTF-8?Q?...?=) to readable text.""" - if not value: - return "" - try: - return str(make_header(decode_header(value))) - except Exception: - return value - - raw_msg_id = msg.get("Message-ID") - if raw_msg_id: - message_id = raw_msg_id - else: - # Deterministic ID so re-syncs don't create duplicates - message_id = f"generated-{account.id}-{folder.id}-{uid_str}" - subject = _decode_mime_header(msg.get("Subject", "")) - from_addr = _decode_mime_header(msg.get("From", "")) - to_addrs = _decode_mime_header(msg.get("To", "")) - cc_addrs = _decode_mime_header(msg.get("Cc", "")) - refs = msg.get("References", "") - in_reply_to = msg.get("In-Reply-To") - date_str = msg.get("Date", "") - - # Parse date for received_at - received_at = datetime.now(UTC) - if date_str: - try: - from email.utils import parsedate_to_datetime - parsed = parsedate_to_datetime(date_str) - if parsed: - received_at = parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC) - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - - # Compute thread_id from References/In-Reply-To - thread_id = _compute_thread_id(message_id, refs, in_reply_to) - - # Dedup: first check by (account_id, folder_id, imap_uid), - # then fall back to (account_id, message_id) for mails synced - # before the imap_uid column existed. - existing_mail = ( - await db.execute( - select(Mail).where( - and_( - Mail.account_id == account.id, - Mail.folder_id == folder.id, - Mail.imap_uid == uid_str, - Mail.tenant_id == tenant_id, - ) - ) - ) - ).scalar_one_or_none() - - if not existing_mail: - existing_mail = ( - await db.execute( - select(Mail).where( - and_( - Mail.account_id == account.id, - Mail.message_id == message_id, - Mail.folder_id == folder.id, - Mail.tenant_id == tenant_id, - ) - ) - ) - ).scalar_one_or_none() - - if existing_mail: - # 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: - existing_mail.folder_id = folder.id - existing_mail.imap_uid = uid_str - elif not existing_mail.imap_uid: - existing_mail.imap_uid = uid_str - - - # Save attachments for existing emails that have has_attachments but no records - if attachments and existing_mail.has_attachments: - existing_att_count = ( - await db.execute( - select(func.count()).select_from(MailAttachment).where( - and_( - MailAttachment.mail_id == existing_mail.id, - MailAttachment.tenant_id == tenant_id, - ) - ) - ) - ).scalar() or 0 - if existing_att_count == 0: - for att_data in attachments: - try: - raw_filename = att_data["filename"] or "attachment" - decoded_filename = _decode_mime_filename(raw_filename) - storage_path = await _save_attachment_to_storage( - existing_mail.id, decoded_filename, att_data["content"] - ) - attachment = MailAttachment( - tenant_id=tenant_id, - mail_id=existing_mail.id, - filename=_sanitize_filename(decoded_filename), - mime_type=att_data["mime_type"], - size_bytes=att_data["size"], - storage_path=storage_path, - content_id=att_data.get("content_id"), - ) - db.add(attachment) - except Exception as att_err: - logger.warning(f"Failed to save attachment for existing mail {existing_mail.id}: {att_err}") - continue - await db.flush() - continue - - mail = Mail( - tenant_id=tenant_id, - account_id=account.id, - folder_id=folder.id, - imap_uid=uid_str, - message_id=message_id, - thread_id=thread_id, - in_reply_to=in_reply_to, - references_header=refs, - subject=subject, - from_address=from_addr, - to_addresses=to_addrs, - cc_addresses=cc_addrs, - body_text=body_text, - body_html=body_html, - body_html_sanitized=sanitize_html(body_html), - has_attachments=len(attachments) > 0, - size_bytes=len(raw_email), - received_at=received_at, - ) - db.add(mail) - await db.flush() - synced_count += 1 - - # Collect for new-mail notifications - new_mails.append({ - "from": from_addr, - "subject": subject, - }) - - # Save attachments to storage and DB - for att_data in attachments: - try: - # Decode MIME-encoded filenames (e.g. =?utf-8?q?...?=) - raw_filename = att_data["filename"] or "attachment" - decoded_filename = _decode_mime_filename(raw_filename) - storage_path = await _save_attachment_to_storage( - mail.id, decoded_filename, att_data["content"] - ) - attachment = MailAttachment( - tenant_id=tenant_id, - mail_id=mail.id, - filename=_sanitize_filename(decoded_filename), - mime_type=att_data["mime_type"], - size_bytes=att_data["size"], - storage_path=storage_path, - content_id=att_data.get("content_id"), - ) - db.add(attachment) - except Exception as att_err: - logger.warning(f"Failed to save attachment for new mail {mail.id}: {att_err}") - continue - await db.flush() - - # Vanished-UID check: delete DB mails whose UID no longer exists on IMAP server - try: - vanished_search = await client.uid_search('ALL') - vanished_raw = vanished_search[1][0] if vanished_search[1] and vanished_search[1][0] else b'' - if isinstance(vanished_raw, (bytes, bytearray)): - imap_uids_set = {u.decode() if isinstance(u, bytes) else str(u) for u in vanished_raw.split()} - else: - imap_uids_set = set() - db_mails = ( - await db.execute( - select(Mail).where( - and_( - Mail.folder_id == folder.id, - Mail.tenant_id == tenant_id, - ) - ) - ) - ).scalars().all() - for db_mail in db_mails: - if db_mail.imap_uid: - if db_mail.imap_uid not in imap_uids_set: - logger.info("imap_sync_account: deleting vanished mail %s (UID %s no longer on server)", db_mail.id, db_mail.imap_uid) - await db.delete(db_mail) - elif db_mail.message_id: - # Mail has no UID — try to find it on IMAP by Message-ID - try: - mid_search = await client.uid_search(f'HEADER Message-ID "{db_mail.message_id}"') - mid_raw = mid_search[1][0] if mid_search[1] and mid_search[1][0] else b'' - if isinstance(mid_raw, (bytes, bytearray)) and mid_raw: - found_uids = mid_raw.decode().split() - if found_uids: - db_mail.imap_uid = found_uids[0] - logger.info("imap_sync_account: found UID %s for mail %s via Message-ID", found_uids[0], db_mail.id) - else: - logger.info("imap_sync_account: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id) - await db.delete(db_mail) - else: - logger.info("imap_sync_account: deleting mail %s (no UID, not found on server by Message-ID)", db_mail.id) - await db.delete(db_mail) - except Exception as mid_exc: - logger.warning("imap_sync_account: Message-ID search failed for mail %s: %s", db_mail.id, mid_exc) - await db.flush() - except Exception as vanished_exc: - logger.warning("imap_sync_account: vanished-UID check failed for folder %s: %s", folder.imap_name, vanished_exc) - - # 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) - ) - ) - ).scalar() - unread = ( - await db.execute( - select(func.count()).select_from(Mail).where( - and_( - Mail.folder_id == folder.id, - Mail.tenant_id == tenant_id, - Mail.is_seen.is_(False), - ) - ) - ) - ).scalar() - folder.total_count = total or 0 - folder.unread_count = unread or 0 - - await db.flush() - - except Exception: - # Skip folders that can't be selected (e.g. no select permission) - continue - - # ── New-mail notifications (max 10, then summary) ── - if new_mails: - try: - if len(new_mails) <= 10: - for nm in new_mails: - try: - await create_notification( - db, account.tenant_id, account.user_id, - "mail_new", - f"Neue E-Mail von {nm['from']}", - nm["subject"], - ) - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - else: - try: - await create_notification( - db, account.tenant_id, account.user_id, - "mail_new", - f"Neue E-Mails: {len(new_mails)} neue Nachrichten", - f"Account {account.email_address} hat {len(new_mails)} neue E-Mails empfangen.", - ) - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - await db.flush() - except Exception: - logger.debug("Ignored exception in mail service", exc_info=True) - - await db.flush() - await client.logout() - return {"synced": synced_count} - except Exception as e: - return {"synced": 0, "error": str(e)} - - -def _compute_thread_id(message_id: str, references: str, in_reply_to: str | None) -> str: - """Compute thread ID from References/In-Reply-To headers (F-MAIL-05).""" - ref_parts: list[str] = [] - if references: - ref_parts = [r.strip() for r in references.split() if r.strip()] - if in_reply_to and in_reply_to.strip() not in ref_parts: - ref_parts.append(in_reply_to.strip()) - if ref_parts: - return ref_parts[0] - return message_id or str(uuid.uuid4()) - - # ─── SMTP Send Service (F-MAIL-02) ───