webmail-client/backend/app/accounts/service.py

"""Account lifecycle: create (encrypt + enqueue sync), delete (cascade + blobs) (SPEC §6.2/§9.1)."""

from __future__ import annotations

import uuid

from sqlalchemy import delete, func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.accounts.schemas import CreateAccountRequest
from app.accounts.types import ServerConfig
from app.indexer import blobstore
from app.models import Account, Attachment, ComposeAttachment, Message, Thread
from app.security.crypto import encrypt_secret


class AccountExistsError(Exception):
    """The user already has an account for this email."""


def _server_dict(cfg: ServerConfig) -> dict[str, object]:
    return {"host": cfg.host, "port": cfg.port, "tls": cfg.tls, "auth": cfg.auth}


async def create_account(
    session: AsyncSession, *, user_id: uuid.UUID, req: CreateAccountRequest
) -> Account:
    """Create an account row with encrypted credentials (state=backfilling)."""
    imap = ServerConfig(req.imap.host, req.imap.port, req.imap.tls, req.imap.auth)
    smtp = ServerConfig(req.smtp.host, req.smtp.port, req.smtp.tls, req.smtp.auth)
    account = Account(
        user_id=user_id,
        email=str(req.email),
        display_name=req.display_name,
        color=req.color,
        imap_config=_server_dict(imap),
        smtp_config=_server_dict(smtp),
        secret_ct=encrypt_secret(req.password.encode("utf-8")),
        discovery_src="manual",
        state="backfilling",
        sync_window=req.sync_window,
        sync_spam=req.sync_spam,
    )
    session.add(account)
    try:
        await session.flush()
    except IntegrityError as exc:
        await session.rollback()
        raise AccountExistsError(str(req.email)) from exc
    return account


async def delete_account(
    session: AsyncSession, *, user_id: uuid.UUID, account_id: uuid.UUID
) -> bool:
    """Delete an account and its data. Removes blobs no longer referenced (dedup-aware).

    Returns False if the account does not belong to the user. DB rows (mailboxes,
    messages, body_parts, attachments) and the Postgres search vector cascade via
    FK ``ON DELETE CASCADE``.
    """
    account = await session.scalar(
        select(Account).where(Account.id == account_id, Account.user_id == user_id)
    )
    if account is None:
        return False

    # Gather blob digests this account references, before the cascade delete.
    msg_shas = set(
        await session.scalars(select(Message.blob_sha256).where(Message.account_id == account_id))
    )
    att_shas = set(
        await session.scalars(
            select(Attachment.blob_sha256)
            .join(Message, Message.id == Attachment.message_id)
            .where(Message.account_id == account_id)
        )
    )
    candidate_shas = msg_shas | att_shas

    await session.execute(delete(Account).where(Account.id == account_id))
    await session.flush()

    # Threads are user-scoped and can span accounts (Thread.account_ids), so the
    # account cascade only removes its messages — not the threads. Clean up: drop
    # threads left with no messages (else they linger as empty rows the unified list
    # and label views still show but can't open), and de-list the account from
    # threads that survive on another account's messages.
    await session.execute(
        delete(Thread).where(
            Thread.user_id == user_id,
            ~select(Message.id).where(Message.thread_id == Thread.id).exists(),
        )
    )
    await session.execute(
        update(Thread)
        .where(Thread.user_id == user_id, Thread.account_ids.contains([account_id]))
        .values(account_ids=func.array_remove(Thread.account_ids, account_id))
    )

    # Deregister from sync:live and drop any lease so the owning sync worker
    # stops the account's runner on its next renewal (refactor §2 fencing).
    from app.events import get_event_bus
    from app.redis_client import get_redis
    from app.sync import leases

    await leases.mark_not_live(get_redis(), account_id)
    # Sync v2 §3: signal the owning worker to tear the runner down now (≤30s),
    # rather than waiting for the lease to lapse.
    await get_event_bus().publish(
        "account.state_changed",
        {"user_id": str(user_id), "account_id": str(account_id), "state": "deleted"},
    )

    # Content-addressed blobs are shared across the user's accounts (dedup): only
    # remove a blob when no remaining message/attachment references it.
    for sha in candidate_shas:
        still_used = await session.scalar(
            select(Message.id)
            .where(Message.user_id == user_id, Message.blob_sha256 == sha)
            .limit(1)
        )
        if still_used is None:
            still_used = await session.scalar(
                select(Attachment.id)
                .where(Attachment.user_id == user_id, Attachment.blob_sha256 == sha)
                .limit(1)
            )
        if still_used is None:
            # A forwarded file waiting in a draft shares the hash: keep it for that.
            still_used = await session.scalar(
                select(ComposeAttachment.id)
                .where(ComposeAttachment.user_id == user_id, ComposeAttachment.blob_sha256 == sha)
                .limit(1)
            )
        if still_used is None:
            blobstore.remove(user_id, sha)
    return True