leocrm/app/plugins/builtins/mail/services.py

"""Service layer for the Mail plugin: encryption, IMAP sync, SMTP send, rules, vacation, PGP."""

from __future__ import annotations

import base64
import json
import logging
import mimetypes
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
import nh3
import pgpy
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from sqlalchemy import and_, func, or_, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.core.notifications import create_notification
from app.plugins.builtins.mail.models import (
    Mail,
    MailAccount,
    MailAttachment,
    MailFolder,
    MailLabel,
    MailLabelAssignment,
    MailRule,
    MailSignature,
    MailTemplate,
    VacationSentLog,
)

logger = logging.getLogger(__name__)

# ─── 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 re
            def decode_q(match):
                charset, encoding, encoded = match.group(1), match.group(2).upper(), match.group(3)
                if encoding == 'B':
                    import base64
                    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,
    }


# ─── AES-256 Encryption (Fernet) ───

MAIL_ENCRYPTION_KEY = os.environ.get("MAIL_ENCRYPTION_KEY", "leocrm-mail-encryption-key-2024")


def _derive_key(password: str, salt: bytes = b"leocrm-mail-salt") -> bytes:
    """Derive a 32-byte Fernet key from a password using PBKDF2."""
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        iterations=480000,
    )
    return base64.urlsafe_b64encode(kdf.derive(password.encode()))


_fernet = Fernet(_derive_key(MAIL_ENCRYPTION_KEY))


def encrypt_password(plaintext: str) -> str:
    """Encrypt a password using AES-256 (Fernet). Returns base64 ciphertext."""
    return _fernet.encrypt(plaintext.encode()).decode()


def decrypt_password(ciphertext: str) -> str:
    """Decrypt a password encrypted with encrypt_password."""
    return _fernet.decrypt(ciphertext.encode()).decode()


# ─── HTML Sanitization (F-MAIL: no script tags) ───


def sanitize_html(raw_html: str) -> str:
    """Sanitize HTML using nh3 — removes script tags and dangerous attributes."""
    if not raw_html:
        return ""
    return nh3.clean(
        raw_html,
        tags={
            "a",
            "b",
            "br",
            "div",
            "em",
            "h1",
            "h2",
            "h3",
            "h4",
            "h5",
            "h6",
            "hr",
            "i",
            "img",
            "li",
            "ol",
            "p",
            "span",
            "strong",
            "table",
            "tbody",
            "td",
            "th",
            "thead",
            "tr",
            "u",
            "ul",
            "blockquote",
            "code",
            "pre",
            "font",
            "center",
        },
        attributes={
            "a": {"href", "title", "target"},
            "img": {"src", "alt", "width", "height"},
            "span": {"style"},
            "div": {"style"},
            "font": {"color", "size", "face"},
            "p": {"style"},
            "td": {"style"},
            "th": {"style"},
        },
    )


# ─── 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 <used> <limit>)
            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:
        pass
    return None


# ─── Mail Account Service ───


async def create_mail_account(
    db: AsyncSession, *, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict
) -> MailAccount:
    """Create a new mail account with encrypted password."""
    account = MailAccount(
        tenant_id=tenant_id,
        user_id=user_id,
        email_address=data["email_address"],
        display_name=data.get("display_name", ""),
        imap_host=data["imap_host"],
        imap_port=data.get("imap_port", 993),
        imap_ssl=data.get("imap_ssl", True),
        smtp_host=data["smtp_host"],
        smtp_port=data.get("smtp_port", 587),
        smtp_tls=data.get("smtp_tls", True),
        username=data.get("username") or data["email_address"],
        encrypted_password=encrypt_password(data["password"]),
        is_shared=data.get("is_shared", False),
        is_active=True,
        sent_folder_imap_name=data.get("sent_folder_imap_name"),
        drafts_folder_imap_name=data.get("drafts_folder_imap_name"),
        spam_folder_imap_name=data.get("spam_folder_imap_name"),
        trash_folder_imap_name=data.get("trash_folder_imap_name"),
    )
    db.add(account)
    await db.flush()

    # Create INBOX first so subfolders can reference it as parent
    inbox_folder = MailFolder(
        tenant_id=tenant_id,
        account_id=account.id,
        name="Posteingang",
        imap_name="INBOX",
        is_standard=True,
    )
    db.add(inbox_folder)
    await db.flush()

    # Create standard subfolders under INBOX (IMAP server uses '.' delimiter)
    for fname, imap_name in [
        ("Gesendet", "INBOX.Sent"),
        ("Entwürfe", "INBOX.Drafts"),
        ("Papierkorb", "INBOX.Trash"),
        ("Spam", "INBOX.spam"),
    ]:
        folder = MailFolder(
            tenant_id=tenant_id,
            account_id=account.id,
            name=fname,
            imap_name=imap_name,
            parent_id=inbox_folder.id,
            is_standard=True,
        )
        db.add(folder)
    await db.flush()
    return account


async def update_mail_account(db: AsyncSession, account: MailAccount, data: dict) -> MailAccount:
    """Update a mail account, encrypting password if changed."""
    field_map = {
        "email": "email_address",
        "email_address": "email_address",
        "display_name": "display_name",
        "imap_host": "imap_host",
        "imap_port": "imap_port",
        "imap_ssl": "imap_ssl",
        "smtp_host": "smtp_host",
        "smtp_port": "smtp_port",
        "smtp_tls": "smtp_tls",
        "username": "username",
        "is_shared": "is_shared",
        "is_active": "is_active",
        "sent_folder_imap_name": "sent_folder_imap_name",
        "drafts_folder_imap_name": "drafts_folder_imap_name",
        "spam_folder_imap_name": "spam_folder_imap_name",
        "trash_folder_imap_name": "trash_folder_imap_name",
    }
    for api_field, model_field in field_map.items():
        if api_field in data and data[api_field] is not None:
            setattr(account, model_field, data[api_field])
    if "password" in data and data["password"] is not None:
        account.encrypted_password = encrypt_password(data["password"])
    await db.flush()
    await db.refresh(account)
    return account


async def get_account_password(account: MailAccount) -> str:
    """Decrypt and return the account password (internal use only)."""
    return decrypt_password(account.encrypted_password)


def account_to_response(account: MailAccount) -> dict:
    """Convert MailAccount to response dict, NEVER including password."""
    return {
        "id": str(account.id),
        "email": account.email_address,
        "email_address": account.email_address,
        "display_name": account.display_name,
        "imap_host": account.imap_host,
        "imap_port": account.imap_port,
        "imap_ssl": account.imap_ssl,
        "smtp_host": account.smtp_host,
        "smtp_port": account.smtp_port,
        "smtp_tls": account.smtp_tls,
        "username": account.username,
        "is_shared": account.is_shared,
        "is_active": account.is_active,
        "sent_folder_imap_name": account.sent_folder_imap_name,
        "drafts_folder_imap_name": account.drafts_folder_imap_name,
        "spam_folder_imap_name": account.spam_folder_imap_name,
        "trash_folder_imap_name": account.trash_folder_imap_name,
        "created_at": account.created_at,
        "updated_at": account.updated_at,
    }


# ─── 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 — mail may have been moved to Trash by IMAP server
            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.tenant_id == tenant_id,
                                    )
                                )
                            )
                        ).scalar_one_or_none()

            if existing_mail:
                # If mail has deleted_at but is still on the IMAP server,
                # the IMAP delete failed.
                # Only restore if the mail is in the SAME folder we're syncing.
                # If the mail was moved to Trash in DB (folder_id=Trash) but is
                # still in INBOX on IMAP, DON'T restore to INBOX — keep it
                # in Trash and let the sync queue retry the IMAP MOVE.
                if existing_mail.deleted_at is not None:
                    if existing_mail.folder_id == folder.id:
                        # Mail is in the right folder on IMAP — restore
                        existing_mail.deleted_at = None
                        logger.info("imap_sync_folder: restored mail %s (still on IMAP server in correct folder)", existing_mail.id)
                    else:
                        # Mail is in a different folder in DB — IMAP move failed
                        # Don't restore, don't change folder_id — keep in Trash
                        logger.info("imap_sync_folder: skipping mail %s (deleted, in folder %s but found in %s on IMAP, move failed)", existing_mail.id, existing_mail.folder_id, folder.imap_name)
                        continue
                # 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:
                    pass

            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

            # 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

        # Update folder counts
        total = (
            await db.execute(
                select(func.count()).select_from(Mail).where(
                    and_(
                        Mail.folder_id == folder_id,
                        Mail.tenant_id == tenant_id,
                        Mail.deleted_at.is_(None),
                    )
                )
            )
        ).scalar() or 0
        unread = (
            await db.execute(
                select(func.count()).select_from(Mail).where(
                    and_(
                        Mail.folder_id == folder_id,
                        Mail.tenant_id == tenant_id,
                        Mail.deleted_at.is_(None),
                        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:
                pass


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:
            pass
        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:
            pass
        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:
            pass
        try:
            await client.logout()
        except Exception:
            pass
        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:
                pass
    except Exception:
        pass

    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:
                            pass

                    # 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.tenant_id == tenant_id,
                                    )
                                )
                            )
                        ).scalar_one_or_none()

                    if existing_mail:
                        # If mail has deleted_at but is still on the IMAP server,
                        # the IMAP delete failed.
                        # Only restore if the mail is in the SAME folder we're syncing.
                        # If the mail was moved to Trash in DB (folder_id=Trash) but is
                        # still in INBOX on IMAP, DON'T restore to INBOX — keep it
                        # in Trash and let the sync queue retry the IMAP MOVE.
                        if existing_mail.deleted_at is not None:
                            if existing_mail.folder_id == folder.id:
                                # Mail is in the right folder on IMAP — restore
                                existing_mail.deleted_at = None
                                logger.info("imap_sync_account: restored mail %s (still on IMAP server in correct folder)", existing_mail.id)
                            else:
                                # Mail is in a different folder in DB — IMAP move failed
                                # Don't restore, don't change folder_id — keep in Trash
                                logger.info("imap_sync_account: skipping mail %s (deleted, in folder %s but found in %s on IMAP, move failed)", existing_mail.id, existing_mail.folder_id, folder.imap_name)
                                continue
                        # 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()

                # 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:
                            pass
                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:
                        pass
                await db.flush()
            except Exception:
                pass

        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) ───


async def send_mail_via_smtp(
    db: AsyncSession,
    *,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    account: MailAccount,
    to_addrs: list[str],
    cc_addrs: list[str] = None,
    bcc_addrs: list[str] = None,
    subject: str = "",
    body_html: str = "",
    body_text: str = "",
    in_reply_to: str | None = None,
    references_header: str | None = None,
    signature: MailSignature | None = None,
    attachment_paths: list[dict] | None = None,
) -> dict:
    """Send an email via SMTP using aiosmtplib.

    Args:
        attachment_paths: list of dicts with keys 'path', 'filename', 'mime_type'
                          pointing to files on disk to attach.
    """
    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:
        body_html = body_html + f"<br><br>{signature.body_html}"
        if body_text:
            body_text = body_text + "\n\n-- \n" + _strip_html(signature.body_html)

    # Build email message
    msg = EmailMessage()
    msg["From"] = formataddr((account.display_name or "", account.email_address))
    msg["To"] = ", ".join(to_addrs)
    if cc_addrs:
        msg["Cc"] = ", ".join(cc_addrs)
    msg["Subject"] = subject
    msg["Date"] = formatdate(localtime=True)
    msg_id = make_msgid(
        domain=account.email_address.split("@")[-1] if "@" in account.email_address else "localhost"
    )
    msg["Message-ID"] = msg_id
    if in_reply_to:
        msg["In-Reply-To"] = in_reply_to
    if references_header:
        msg["References"] = references_header

    if body_html:
        msg.set_content(body_text or _strip_html(body_html), subtype="plain")
        msg.add_alternative(body_html, subtype="html")
    else:
        msg.set_content(body_text, subtype="plain")

    # Add attachments to the message
    for att_info in attachment_paths:
        file_path = att_info.get("path", "")
        filename = att_info.get("filename", os.path.basename(file_path))
        mime_type = att_info.get("mime_type", "application/octet-stream")
        if not file_path or not os.path.exists(file_path):
            continue
        with open(file_path, "rb") as f:  # noqa: ASYNC230
            content = f.read()
        # Determine maintype/subtype from mime_type
        if "/" in mime_type:
            maintype, subtype = mime_type.split("/", 1)
        else:
            maintype, subtype = "application", "octet-stream"
        msg.add_attachment(
            content,
            maintype=maintype,
            subtype=subtype,
            filename=filename,
        )

    # Send via SMTP
    password = await get_account_password(account)
    try:
        smtp = aiosmtplib.SMTP(
            hostname=account.smtp_host,
            port=account.smtp_port,
            use_tls=account.smtp_tls,
        )
        await smtp.connect()
        await smtp.login(account.username, password)
        recipients = to_addrs + cc_addrs + bcc_addrs
        await smtp.send_message(msg, recipients=recipients)
        await smtp.quit()

        # Store sent mail in Sent folder — use configured mapping if set,
        # otherwise flexible lookup to handle different IMAP naming conventions
        if account.sent_folder_imap_name:
            sent_folder = (
                await db.execute(
                    select(MailFolder).where(
                        and_(
                            MailFolder.account_id == account.id,
                            MailFolder.imap_name == account.sent_folder_imap_name,
                        )
                    )
                )
            ).scalar_one_or_none()
        else:
            sent_folder = (
                await db.execute(
                    select(MailFolder).where(
                        and_(
                            MailFolder.account_id == account.id,
                            MailFolder.imap_name.in_(
                                ["Sent", "INBOX.Sent", "Sent Items", "Sent Mail"]
                            ),
                        )
                    )
                )
            ).scalar_one_or_none()

            if not sent_folder:
                sent_folder = (
                    await db.execute(
                        select(MailFolder).where(
                            and_(
                                MailFolder.account_id == account.id,
                                MailFolder.is_standard == True,
                                MailFolder.imap_name.ilike("%sent%"),
                            )
                        )
                    )
                ).scalar_one_or_none()

        if sent_folder:
            thread_id = _compute_thread_id(msg_id, references_header or "", in_reply_to)
            sent_mail = Mail(
                tenant_id=tenant_id,
                account_id=account.id,
                folder_id=sent_folder.id,
                message_id=msg_id,
                thread_id=thread_id,
                in_reply_to=in_reply_to,
                references_header=references_header,
                subject=subject,
                from_address=account.email_address,
                to_addresses=", ".join(to_addrs),
                cc_addresses=", ".join(cc_addrs),
                bcc_addresses=", ".join(bcc_addrs),
                body_text=body_text or _strip_html(body_html),
                body_html=body_html,
                body_html_sanitized=sanitize_html(body_html),
                is_seen=True,
                is_answered=bool(in_reply_to),
                sent_at=datetime.now(UTC),
            )
            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(
                db, tenant_id, user_id,
                "mail_sent",
                "E-Mail gesendet",
                subject,
            )
            await db.flush()
        except Exception:
            pass

        return {"status": "sent", "message_id": msg_id}
    except Exception as e:
        # ── Notification: send error ──
        try:
            await create_notification(
                db, tenant_id, user_id,
                "mail_send_error",
                "E-Mail konnte nicht gesendet werden",
                str(e),
            )
            await db.flush()
        except Exception:
            pass
        return {"status": "error", "error": str(e)}


async def reply_to_mail(
    db: AsyncSession,
    *,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    original_mail: Mail,
    account: MailAccount,
    body_html: str,
    body_text: str = "",
    reply_to_all: bool = False,
    signature: MailSignature | None = None,
) -> dict:
    """Reply to a mail, setting In-Reply-To and References headers (F-MAIL-02)."""
    to_addrs = [original_mail.from_address]
    if reply_to_all and original_mail.cc_addresses:
        to_addrs.extend([a.strip() for a in original_mail.cc_addresses.split(",") if a.strip()])

    refs = original_mail.references_header or ""
    new_refs = f"{refs} {original_mail.message_id}".strip()

    result = await send_mail_via_smtp(
        db,
        tenant_id=tenant_id,
        user_id=user_id,
        account=account,
        to_addrs=to_addrs,
        subject=f"Re: {original_mail.subject}".replace("Re: Re: ", "Re: "),
        body_html=body_html,
        body_text=body_text,
        in_reply_to=original_mail.message_id,
        references_header=new_refs,
        signature=signature,
    )

    # Mark original as answered
    original_mail.is_answered = True
    await db.flush()
    return result


async def forward_mail(
    db: AsyncSession,
    *,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    original_mail: Mail,
    account: MailAccount,
    to_addrs: list[str],
    cc_addrs: list[str] = None,
    body_html: str = "",
    body_text: str = "",
    signature: MailSignature | None = None,
) -> dict:
    """Forward a mail with original as forwarded content (F-MAIL-02)."""
    fwd_subject = f"Fwd: {original_mail.subject}".replace("Fwd: Fwd: ", "Fwd: ")
    fwd_body = (
        f"<br><br>----- Original Message -----<br>"
        f"From: {original_mail.from_address}<br>"
        f"Subject: {original_mail.subject}<br><br>"
        f"{original_mail.body_html or original_mail.body_text}"
    )
    full_html = body_html + fwd_body
    full_text = (body_text or _strip_html(body_html)) + "\n\n----- Original Message -----\n"

    result = await send_mail_via_smtp(
        db,
        tenant_id=tenant_id,
        user_id=user_id,
        account=account,
        to_addrs=to_addrs,
        cc_addrs=cc_addrs or [],
        subject=fwd_subject,
        body_html=full_html,
        body_text=full_text,
        signature=signature,
    )

    original_mail.is_forwarded = True
    await db.flush()
    return result


# ─── Template Service (F-MAIL-06) ───


def substitute_template_vars(template_body: str, variables: dict[str, str]) -> str:
    """Replace {{placeholder}} variables in template body."""
    result = template_body
    for key, value in variables.items():
        result = result.replace(f"{{{{{key}}}}}", value)
        result = result.replace(f"{{{{{key.lower()}}}}}", value)
        result = result.replace(f"{{{{{key.upper()}}}}}", value)
    return result


# ─── Mail Rule Engine (F-MAIL-07) ───


def matches_condition(mail: Mail, conditions: dict) -> bool:
    """Check if a mail matches all rule conditions."""
    for field, expected in conditions.items():
        if field == "from_contains":
            if expected.lower() not in mail.from_address.lower():
                return False
        elif field == "subject_contains":
            if expected.lower() not in mail.subject.lower():
                return False
        elif field == "to_contains":
            if expected.lower() not in mail.to_addresses.lower():
                return False
        elif field == "body_contains":
            body = (mail.body_text + mail.body_html).lower()
            if expected.lower() not in body:
                return False
        elif field == "has_attachments":
            if mail.has_attachments != bool(expected):
                return False
        elif field == "is_flagged":
            if mail.is_flagged != bool(expected):
                return False
    return True


async def execute_rule_actions(
    db: AsyncSession, mail: Mail, actions: dict, tenant_id: uuid.UUID
) -> dict:
    """Execute rule actions on a matching mail."""
    results = {}
    for action, value in actions.items():
        if action == "move_to_folder":
            folder_id = uuid.UUID(value) if isinstance(value, str) else value
            folder = (
                await db.execute(
                    select(MailFolder).where(
                        and_(MailFolder.id == folder_id, MailFolder.tenant_id == tenant_id)
                    )
                )
            ).scalar_one_or_none()
            if folder:
                mail.folder_id = folder.id
                results["moved"] = str(folder.id)
        elif action == "label":
            label_id = uuid.UUID(value) if isinstance(value, str) else value
            existing = (
                await db.execute(
                    select(MailLabelAssignment).where(
                        and_(
                            MailLabelAssignment.mail_id == mail.id,
                            MailLabelAssignment.label_id == label_id,
                        )
                    )
                )
            ).scalar_one_or_none()
            if not existing:
                assignment = MailLabelAssignment(
                    tenant_id=tenant_id,
                    mail_id=mail.id,
                    label_id=label_id,
                )
                db.add(assignment)
                results["labeled"] = str(label_id)
        elif action == "mark_seen":
            mail.is_seen = bool(value)
            results["seen"] = bool(value)
        elif action == "mark_flagged":
            mail.is_flagged = bool(value)
            results["flagged"] = bool(value)
        elif action == "forward_to":
            results["forward_to"] = value
    await db.flush()
    return results


async def apply_rules_to_mail(db: AsyncSession, mail: Mail, tenant_id: uuid.UUID) -> list[dict]:
    """Find and apply all matching rules to a mail, sorted by priority."""
    rules = (
        (
            await db.execute(
                select(MailRule)
                .where(
                    and_(
                        MailRule.tenant_id == tenant_id,
                        MailRule.is_active,
                        or_(
                            MailRule.account_id == mail.account_id,
                            MailRule.account_id.is_(None),
                        ),
                    )
                )
                .order_by(MailRule.priority)
            )
        )
        .scalars()
        .all()
    )

    applied = []
    for rule in rules:
        conditions = json.loads(rule.conditions) if rule.conditions else {}
        actions = json.loads(rule.actions) if rule.actions else {}
        if matches_condition(mail, conditions):
            result = await execute_rule_actions(db, mail, actions, tenant_id)
            applied.append({"rule_id": str(rule.id), "rule_name": rule.name, "actions": result})
    return applied


# ─── Vacation Auto-Reply (F-MAIL-08) ───


VACATION_DEDUP_HOURS = 24


async def should_send_vacation_reply(
    db: AsyncSession,
    account_id: uuid.UUID,
    sender_address: str,
    tenant_id: uuid.UUID,
) -> bool:
    """Check if vacation auto-reply should be sent (dedup within 24h)."""
    cutoff = datetime.now(UTC) - timedelta(hours=VACATION_DEDUP_HOURS)
    existing = (
        await db.execute(
            select(VacationSentLog).where(
                and_(
                    VacationSentLog.account_id == account_id,
                    VacationSentLog.sender_address == sender_address,
                    VacationSentLog.sent_at >= cutoff,
                    VacationSentLog.tenant_id == tenant_id,
                )
            )
        )
    ).scalar_one_or_none()
    return existing is None


async def log_vacation_sent(
    db: AsyncSession,
    account_id: uuid.UUID,
    sender_address: str,
    tenant_id: uuid.UUID,
) -> None:
    """Log that a vacation auto-reply was sent to a sender."""
    log = VacationSentLog(
        tenant_id=tenant_id,
        account_id=account_id,
        sender_address=sender_address,
        sent_at=datetime.now(UTC),
    )
    db.add(log)
    await db.flush()


# ─── PGP Service (F-MAIL-12) ───


def import_pgp_private_key(private_key_armored: str, passphrase: str = "") -> tuple[str, str]:
    """Import a PGP private key. Returns (key_id, public_key_armored)."""
    key, _ = pgpy.PGPKey.from_blob(private_key_armored)
    if key.is_protected:
        with key.unlock(passphrase):
            pub_key = key.pubkey
            key_id = str(key.fingerprint).upper()[-16:]
            return key_id, str(pub_key)
    pub_key = key.pubkey
    key_id = str(key.fingerprint).upper()[-16:]
    return key_id, str(pub_key)


def import_pgp_public_key(public_key_armored: str) -> str:
    """Import a PGP public key. Returns key_id."""
    key, _ = pgpy.PGPKey.from_blob(public_key_armored)
    return str(key.fingerprint).upper()[-16:]


def pgp_encrypt_message(plaintext: str, recipient_public_key_armored: str) -> str:
    """Encrypt a message with recipient's public PGP key."""
    pub_key, _ = pgpy.PGPKey.from_blob(recipient_public_key_armored)
    msg = pgpy.PGPMessage.new(plaintext)
    encrypted = pub_key.encrypt(msg)
    return str(encrypted)


def pgp_decrypt_message(ciphertext: str, private_key_armored: str, passphrase: str = "") -> str:
    """Decrypt a PGP-encrypted message."""
    key, _ = pgpy.PGPKey.from_blob(private_key_armored)
    enc_msg = pgpy.PGPMessage.from_blob(ciphertext)
    if key.is_protected:
        with key.unlock(passphrase):
            decrypted = key.decrypt(enc_msg)
            return decrypted.message.decode("utf-8")
    decrypted = key.decrypt(enc_msg)
    return decrypted.message.decode("utf-8")


# ─── Contact Linking (F-MAIL-10) ───


def extract_email_addresses(text: str) -> list[str]:
    """Extract email addresses from a text string."""
    if not text:
        return []
    return re.findall(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", text)


# ─── Utility ───


def _strip_html(html: str) -> str:
    """Simple HTML to text conversion for plain text fallback."""
    if not html:
        return ""
    # Remove tags
    text = re.sub(r"<[^>]+>", "", html)
    # Replace HTML entities
    text = (
        text.replace("&nbsp;", " ")
        .replace("&amp;", "&")
        .replace("&lt;", "<")
        .replace("&gt;", ">")
        .replace("&quot;", '"')
    )
    return text.strip()


def mail_to_response(
    mail: Mail,
    attachments: list[MailAttachment] | None = None,
    labels: list[MailLabel] | None = None,
) -> dict:
    """Convert a Mail ORM object to a response dict."""
    resp = {
        "id": str(mail.id),
        "account_id": str(mail.account_id),
        "folder_id": str(mail.folder_id),
        "message_id": mail.message_id,
        "thread_id": mail.thread_id,
        "in_reply_to": mail.in_reply_to,
        "subject": mail.subject,
        "from_address": mail.from_address,
        "from_name": mail.from_address.split("<")[0].strip().strip('"') if "<" in mail.from_address else mail.from_address,
        "to_addresses": extract_email_addresses(mail.to_addresses) if mail.to_addresses else [],
        "cc_addresses": extract_email_addresses(mail.cc_addresses) if mail.cc_addresses else [],
        "bcc_addresses": extract_email_addresses(mail.bcc_addresses) if mail.bcc_addresses else [],
        "body_text": mail.body_text,
        "body_html": mail.body_html if mail.body_html else None,
        "body_html_sanitized": mail.body_html_sanitized,
        "sanitized_html": mail.body_html_sanitized,
        "date": mail.received_at.isoformat() if mail.received_at else (mail.sent_at.isoformat() if mail.sent_at else None),
        "is_seen": mail.is_seen,
        "is_flagged": mail.is_flagged,
        "flag_type": getattr(mail, 'flag_type', None),
        "is_draft": mail.is_draft,
        "is_answered": mail.is_answered,
        "is_forwarded": mail.is_forwarded,
        "has_attachments": mail.has_attachments,
        "size_bytes": mail.size_bytes,
        "received_at": mail.received_at,
        "sent_at": mail.sent_at,
        "contact_id": str(mail.contact_id) if mail.contact_id else None,
        "company_id": str(mail.company_id) if mail.company_id else None,
        "attachments": [],
        "labels": [],
    }
    if attachments:
        resp["attachments"] = [attachment_to_response(a) for a in attachments]
    if labels:
        resp["labels"] = [
            {"id": str(lbl.id), "name": lbl.name, "color": lbl.color} for lbl in labels
        ]
    return resp


def folder_to_response(folder: MailFolder) -> dict:
    """Convert MailFolder to response dict."""
    return {
        "id": str(folder.id),
        "account_id": str(folder.account_id),
        "name": folder.name,
        "imap_name": folder.imap_name,
        "parent_id": str(folder.parent_id) if folder.parent_id else None,
        "is_standard": folder.is_standard,
        "unread_count": folder.unread_count,
        "total_count": folder.total_count,
    }


def rule_to_response(rule: MailRule) -> dict:
    """Convert MailRule to response dict."""
    return {
        "id": str(rule.id),
        "name": rule.name,
        "account_id": str(rule.account_id) if rule.account_id else None,
        "priority": rule.priority,
        "is_active": rule.is_active,
        "conditions": json.loads(rule.conditions) if rule.conditions else {},
        "actions": json.loads(rule.actions) if rule.actions else {},
    }


def template_to_response(template: MailTemplate) -> dict:
    """Convert MailTemplate to response dict."""
    return {
        "id": str(template.id),
        "name": template.name,
        "subject": template.subject,
        "body_html": template.body_html,
    }


def signature_to_response(sig: MailSignature) -> dict:
    """Convert MailSignature to response dict."""
    return {
        "id": str(sig.id),
        "name": sig.name,
        "body_html": sig.body_html,
        "account_id": str(sig.account_id) if sig.account_id else None,
        "is_default": sig.is_default,
    }


def label_to_response(label: MailLabel) -> dict:
    """Convert MailLabel to response dict."""
    return {
        "id": str(label.id),
        "name": label.name,
        "color": label.color,
    }


# ─── IMAP Flag Sync ───


async def imap_sync_mail_flags(
    db: AsyncSession,
    mail_id: uuid.UUID,
    tenant_id: uuid.UUID,
) -> None:
    """Sync is_seen/is_flagged flags from DB to IMAP server.

    Connects to the IMAP server, selects the mail's folder,
    and uses UID STORE to set/remove \\Seen and \\Flagged flags.
    Non-critical: logs warnings on failure but does not raise.
    """
    import logging

    logger = logging.getLogger(__name__)

    # Load the mail with its folder and account
    mail = (
        await db.execute(
            select(Mail).where(
                and_(Mail.id == mail_id, Mail.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not mail:
        logger.warning("imap_sync_mail_flags: mail %s not found", mail_id)
        return

    folder = (
        await db.execute(
            select(MailFolder).where(
                and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not folder:
        logger.warning("imap_sync_mail_flags: folder %s not found for mail %s", mail.folder_id, mail_id)
        return

    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:
        logger.warning("imap_sync_mail_flags: account not found for mail %s", mail_id)
        return

    if not mail.message_id:
        logger.warning("imap_sync_mail_flags: mail %s has no message_id, cannot sync", mail_id)
        return

    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 the folder
        select_resp = await client.select(folder.imap_name)
        if select_resp.result != 'OK':
            logger.warning("imap_sync_mail_flags: cannot select folder %s", folder.imap_name)
            return

        # Find the UID by searching for the Message-ID header
        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_sync_mail_flags: 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])

        # Sync \Seen flag
        if mail.is_seen:
            await client.uid('store', uid_str, '+FLAGS (\\Seen)')
        else:
            await client.uid('store', uid_str, '-FLAGS (\\Seen)')

        # Sync \Flagged flag
        if mail.is_flagged:
            await client.uid('store', uid_str, '+FLAGS (\\Flagged)')
        else:
            await client.uid('store', uid_str, '-FLAGS (\\Flagged)')

        logger.info("imap_sync_mail_flags: synced flags for mail %s (UID %s)", mail_id, uid_str)

    except Exception as exc:
        logger.warning("imap_sync_mail_flags: failed for mail %s: %s", mail_id, exc)
    finally:
        if client is not None:
            try:
                await client.logout()
            except Exception:
                pass


# ─── IMAP Delete ───


async def _find_trash_folder_name(
    db: AsyncSession, account: MailAccount, tenant_id: uuid.UUID
) -> str | None:
    """Find the IMAP Trash folder name for an account.

    Checks account.trash_folder_imap_name, then DB folders with 'trash' in imap_name,
    then queries the IMAP server LIST for common Trash folder names.
    """
    # 1) Explicit mapping on account
    if account.trash_folder_imap_name:
        return account.trash_folder_imap_name

    # 2) DB folder with 'trash' in imap_name
    db_folders = (
        await db.execute(
            select(MailFolder).where(
                and_(MailFolder.account_id == account.id, MailFolder.tenant_id == tenant_id)
            )
        )
    ).scalars().all()
    for f in db_folders:
        if 'trash' in f.imap_name.lower():
            return f.imap_name

    # 3) Query IMAP server for common Trash folder names
    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)
        list_resp = await client.list('""', '"*"')
        imap_folders, _ = _parse_imap_list_response(list_resp)
        trash_candidates = ['Trash', 'INBOX.Trash', 'INBOX.Trash', 'Deleted', 'Deleted Items', 'Papierkorb']
        for _, name in imap_folders:
            if name in trash_candidates or 'trash' in name.lower():
                return name
    except Exception:
        pass
    finally:
        if client:
            try:
                await client.logout()
            except Exception:
                pass

    return None


async def imap_delete_mail(
    db: AsyncSession, mail_id: uuid.UUID, tenant_id: uuid.UUID, *, permanent: bool = False
) -> None:
    """Delete mail from IMAP server.

    If permanent=False (default): MOVE mail to IMAP Trash folder (like real mail clients).
    If permanent=True: STORE \\Deleted + EXPUNGE (permanent delete from server).
    Uses UID MOVE if supported, otherwise COPY + STORE \\Deleted + EXPUNGE.
    Raises exceptions on failure so caller can queue for retry.
    """
    import logging

    logger = logging.getLogger(__name__)

    mail = (
        await db.execute(
            select(Mail).where(
                and_(Mail.id == mail_id, Mail.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not mail:
        logger.warning("imap_delete_mail: mail %s not found", mail_id)
        return

    folder = (
        await db.execute(
            select(MailFolder).where(
                and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not folder:
        logger.warning("imap_delete_mail: folder %s not found for mail %s", mail.folder_id, mail_id)
        return

    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:
        logger.warning("imap_delete_mail: account not found for mail %s", mail_id)
        return

    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

    # Find Trash folder on IMAP server (only needed for non-permanent delete)
    trash_folder_name = None
    if not permanent:
        trash_folder_name = await _find_trash_folder_name(db, account, tenant_id)
        if not trash_folder_name:
            logger.warning("imap_delete_mail: no Trash folder found for account %s, will permanent delete", account.id)

    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':
            raise RuntimeError(f"cannot select folder {folder.imap_name}")

        # 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:
                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])

        if permanent or not trash_folder_name:
            # Permanent delete: STORE \Deleted + EXPUNGE
            await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
            await client.expunge()
            logger.info("imap_delete_mail: permanent deleted mail %s (UID %s)", mail_id, uid_str)
            return

        # Non-permanent: MOVE to Trash folder (like Thunderbird, Outlook, etc.)
        # Don't move if already in Trash
        if folder.imap_name.lower() == trash_folder_name.lower():
            # Already in Trash — permanent delete
            await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
            await client.expunge()
            logger.info("imap_delete_mail: permanent deleted mail %s (already in Trash, UID %s)", mail_id, uid_str)
            return

        # Try UID MOVE first
        try:
            move_resp = await client.uid('move', uid_str, trash_folder_name)
            if move_resp.result == 'OK':
                logger.info("imap_delete_mail: moved mail %s (UID %s) to Trash %s", mail_id, uid_str, trash_folder_name)
                return
        except Exception as move_exc:
            logger.info("imap_delete_mail: UID MOVE not supported, falling back: %s", move_exc)

        # Fallback: COPY to Trash + STORE \Deleted + EXPUNGE
        copy_resp = await client.uid('copy', uid_str, trash_folder_name)
        if copy_resp.result != 'OK':
            raise RuntimeError(f"COPY to Trash failed for mail {mail_id}")
        await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
        await client.expunge()
        logger.info("imap_delete_mail: moved mail %s (UID %s) to Trash via COPY+DELETE", mail_id, uid_str)

    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:
                await client.logout()
            except Exception:
                pass


# ─── IMAP Move ───


async def imap_move_mail(
    db: AsyncSession,
    mail_id: uuid.UUID,
    target_folder_id: uuid.UUID,
    tenant_id: uuid.UUID,
) -> 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.
    Raises exceptions on failure so caller can queue for retry.
    """
    import logging

    logger = logging.getLogger(__name__)

    mail = (
        await db.execute(
            select(Mail).where(
                and_(Mail.id == mail_id, Mail.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not mail:
        logger.warning("imap_move_mail: mail %s not found", mail_id)
        return

    source_folder = (
        await db.execute(
            select(MailFolder).where(
                and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not source_folder:
        logger.warning("imap_move_mail: source folder not found for mail %s", mail_id)
        return

    target_folder = (
        await db.execute(
            select(MailFolder).where(
                and_(MailFolder.id == target_folder_id, MailFolder.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not target_folder:
        logger.warning("imap_move_mail: target folder %s not found", target_folder_id)
        return

    account = (
        await db.execute(
            select(MailAccount).where(
                and_(MailAccount.id == source_folder.account_id, MailAccount.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not account:
        logger.warning("imap_move_mail: account not found for mail %s", mail_id)
        return

    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)
    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(source_folder.imap_name)
        if select_resp.result != 'OK':
            raise RuntimeError(f"cannot select folder {source_folder.imap_name}")

        # 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:
                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:
            move_resp = await client.uid('move', uid_str, target_folder.imap_name)
            if move_resp.result == 'OK':
                logger.info("imap_move_mail: moved mail %s (UID %s) via UID MOVE", mail_id, uid_str)
                return
        except Exception as move_exc:
            logger.info("imap_move_mail: UID MOVE not supported, falling back: %s", move_exc)

        # Fallback: COPY + STORE \\Deleted + EXPUNGE
        copy_resp = await client.uid('copy', uid_str, target_folder.imap_name)
        if copy_resp.result != 'OK':
            raise RuntimeError(f"COPY failed for mail {mail_id}")

        await client.uid('store', uid_str, r'+FLAGS (\\Deleted)')
        await client.expunge()

        logger.info("imap_move_mail: moved mail %s (UID %s) via COPY+DELETE", mail_id, uid_str)

    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:
                await client.logout()
            except Exception:
                pass


# ─── Draft Save / Update ───


async def save_draft(
    db: AsyncSession,
    account_id: uuid.UUID,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    data: dict,
) -> Mail:
    """Save a new draft mail to DB and IMAP Drafts folder."""
    import logging

    logger = logging.getLogger(__name__)

    # 1. Find the Drafts folder (imap_name contains 'Drafts')
    account = (
        await db.execute(
            select(MailAccount).where(
                and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not account:
        raise ValueError("Account not found")

    folders = (
        await db.execute(
            select(MailFolder).where(
                and_(MailFolder.account_id == account_id, MailFolder.tenant_id == tenant_id)
            )
        )
    ).scalars().all()

    drafts_folder = None
    for f in folders:
        if 'draft' in f.imap_name.lower():
            drafts_folder = f
            break

    if not drafts_folder:
        raise ValueError("No Drafts folder found for this account")

    # 2. Create Mail record
    to_str = ', '.join(data.get('to', []))
    cc_str = ', '.join(data.get('cc', []))
    bcc_str = ', '.join(data.get('bcc', []))
    subject = data.get('subject', '')
    body_text = data.get('body_text', '')
    body_html = data.get('body_html', '')

    msg_id = make_msgid()
    now = datetime.now(UTC)

    mail = Mail(
        tenant_id=tenant_id,
        account_id=account_id,
        folder_id=drafts_folder.id,
        message_id=msg_id,
        thread_id=msg_id,
        subject=subject,
        from_address=account.email_address,
        to_addresses=to_str,
        cc_addresses=cc_str,
        bcc_addresses=bcc_str,
        body_text=body_text,
        body_html=body_html,
        body_html_sanitized=body_html,
        is_seen=True,
        is_flagged=False,
        is_draft=True,
        is_answered=False,
        is_forwarded=False,
        has_attachments=False,
        size_bytes=len(body_text.encode('utf-8')),
        received_at=now,
        sent_at=None,
    )
    db.add(mail)
    await db.flush()

    # ── Notification: draft saved ──
    try:
        await create_notification(
            db, tenant_id, user_id,
            "mail_draft",
            "Entwurf gespeichert",
            subject or "Ohne Betreff",
        )
        await db.flush()
    except Exception:
        pass

    # 3. Build RFC822 message and APPEND to IMAP Drafts folder
    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)

        # Build RFC822 message
        email_msg = EmailMessage()
        email_msg['From'] = formataddr((account.display_name, account.email_address))
        if to_str:
            email_msg['To'] = to_str
        if cc_str:
            email_msg['Cc'] = cc_str
        email_msg['Subject'] = subject
        email_msg['Date'] = formatdate(localtime=True)
        email_msg['Message-ID'] = msg_id
        email_msg.set_content(body_text if body_text else '')
        if body_html:
            email_msg.add_alternative(body_html, subtype='html')

        rfc822_bytes = email_msg.as_bytes()

        # APPEND to Drafts folder
        append_resp = await client.append(
            drafts_folder.imap_name,
            r'(\\Draft)',
            str(int(now.timestamp())),
            rfc822_bytes,
        )
        if append_resp.result != 'OK':
            logger.warning("save_draft: IMAP APPEND failed for drafts folder %s", drafts_folder.imap_name)
        else:
            logger.info("save_draft: appended draft %s to IMAP Drafts", mail.id)

    except Exception as exc:
        logger.warning("save_draft: IMAP append failed (non-critical): %s", exc)
    finally:
        if client is not None:
            try:
                await client.logout()
            except Exception:
                pass

    return mail


async def update_draft(
    db: AsyncSession,
    mail_id: uuid.UUID,
    tenant_id: uuid.UUID,
    data: dict,
) -> Mail:
    """Update an existing draft."""
    import logging

    logger = logging.getLogger(__name__)

    mail = (
        await db.execute(
            select(Mail).where(
                and_(Mail.id == mail_id, Mail.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not mail:
        raise ValueError("Mail not found")
    if not mail.is_draft:
        raise ValueError("Mail is not a draft")

    # 2. Update fields
    to_str = ', '.join(data.get('to', []))
    cc_str = ', '.join(data.get('cc', []))
    bcc_str = ', '.join(data.get('bcc', []))
    mail.to_addresses = to_str
    mail.cc_addresses = cc_str
    mail.bcc_addresses = bcc_str
    mail.subject = data.get('subject', '')
    mail.body_text = data.get('body_text', '')
    mail.body_html = data.get('body_html', '')
    mail.body_html_sanitized = data.get('body_html', '')
    mail.size_bytes = len(mail.body_text.encode('utf-8'))
    await db.flush()

    # 3. Delete old IMAP copy and APPEND new one
    folder = (
        await db.execute(
            select(MailFolder).where(
                and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not folder:
        return mail

    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 mail

    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)

        # Delete old copy from IMAP
        select_resp = await client.select(folder.imap_name)
        if select_resp.result == 'OK' and mail.message_id:
            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 uids:
                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()

        # APPEND new copy
        email_msg = EmailMessage()
        email_msg['From'] = formataddr((account.display_name, account.email_address))
        if to_str:
            email_msg['To'] = to_str
        if cc_str:
            email_msg['Cc'] = cc_str
        email_msg['Subject'] = mail.subject
        email_msg['Date'] = formatdate(localtime=True)
        email_msg['Message-ID'] = mail.message_id
        email_msg.set_content(mail.body_text if mail.body_text else '')
        if mail.body_html:
            email_msg.add_alternative(mail.body_html, subtype='html')

        rfc822_bytes = email_msg.as_bytes()
        now = datetime.now(UTC)
        append_resp = await client.append(
            folder.imap_name,
            r'(\\Draft)',
            str(int(now.timestamp())),
            rfc822_bytes,
        )
        if append_resp.result != 'OK':
            logger.warning("update_draft: IMAP APPEND failed for drafts folder %s", folder.imap_name)
        else:
            logger.info("update_draft: appended updated draft %s to IMAP Drafts", mail.id)

    except Exception as exc:
        logger.warning("update_draft: IMAP sync failed (non-critical): %s", exc)
    finally:
        if client is not None:
            try:
                await client.logout()
            except Exception:
                pass

    return mail


# ─── IMAP Folder Create / Delete ───


async def imap_create_folder(
    db: AsyncSession, account_id: uuid.UUID, folder_name: str, tenant_id: uuid.UUID
) -> None:
    """Create folder on IMAP server.

    Connects to IMAP, creates folder with CREATE command.
    Non-critical: errors are logged, DB operation still succeeds.
    """
    import logging

    logger = logging.getLogger(__name__)

    account = (
        await db.execute(
            select(MailAccount).where(
                and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id)
            )
        )
    ).scalar_one_or_none()
    if not account:
        logger.warning("imap_create_folder: account %s not found", account_id)
        return

    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)

        resp = await client.create(folder_name)
        if resp.result != "OK":
            logger.warning(
                "imap_create_folder: CREATE failed for %s: %s", folder_name, resp
            )
        else:
            logger.info("imap_create_folder: created folder %s on IMAP", folder_name)
            try:
                await create_notification(
                    db, account.tenant_id, account.user_id,
                    "mail_folder",
                    "Ordner erstellt",
                    folder_name,
                )
                await db.flush()
            except Exception:
                pass

    except Exception as exc:
        logger.warning("imap_create_folder: failed (non-critical): %s", exc)
    finally:
        if client is not None:
            try:
                await client.logout()
            except Exception:
                pass


async def imap_delete_folder(
    db: AsyncSession, folder_id: uuid.UUID, tenant_id: uuid.UUID
) -> None:
    """Delete folder from IMAP server.

    Connects to IMAP, deletes folder with DELETE command.
    Non-critical: errors are logged, DB operation still succeeds.
    """
    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:
        logger.warning("imap_delete_folder: folder %s not found", folder_id)
        return

    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:
        logger.warning("imap_delete_folder: account not found for folder %s", folder_id)
        return

    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)

        resp = await client.delete(folder.imap_name)
        if resp.result != "OK":
            logger.warning(
                "imap_delete_folder: DELETE failed for %s: %s", folder.imap_name, resp
            )
        else:
            logger.info("imap_delete_folder: deleted folder %s on IMAP", folder.imap_name)
            try:
                await create_notification(
                    db, account.tenant_id, account.user_id,
                    "mail_folder",
                    "Ordner gelöscht",
                    folder.imap_name,
                )
                await db.flush()
            except Exception:
                pass

    except Exception as exc:
        logger.warning("imap_delete_folder: failed (non-critical): %s", exc)
    finally:
        if client is not None:
            try:
                await client.logout()
            except Exception:
                pass


# ─── Auto-Sync ───


async def auto_sync_all_accounts() -> None:
    """Auto-sync all active mail accounts.

    Called periodically by the background scheduler.
    Iterates all active mail accounts and syncs each one.
    """
    import logging

    from app.core.db import get_session_factory

    logger = logging.getLogger(__name__)

    factory = get_session_factory()
    async with factory() as db:
        accounts = (
            await db.execute(
                select(MailAccount).where(MailAccount.is_active.is_(True))
            )
        ).scalars().all()

        if not accounts:
            return

        logger.info("auto_sync_all_accounts: syncing %d active account(s)", len(accounts))

        for account in accounts:
            try:
                result = await imap_sync_account(db, account.id, account.tenant_id)
                logger.info(
                    "auto_sync_all_accounts: synced account %s (%s): %s",
                    account.id,
                    account.username,
                    result,
                )
            except Exception as exc:
                logger.warning(
                    "auto_sync_all_accounts: failed for account %s: %s",
                    account.id,
                    exc,
                )
                try:
                    await create_notification(
                        db, account.tenant_id, account.user_id,
                        "mail_sync_error",
                        "Synchronisierung fehlgeschlagen",
                        f"Account {account.email_address}: {exc}",
                    )
                except Exception:
                    pass
            # commit per-account so partial progress is saved
            try:
                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":
                is_permanent = entry.payload.get("permanent", False) if entry.payload else False
                await imap_delete_mail(db, entry.mail_id, entry.tenant_id, permanent=is_permanent)
            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()