diff --git a/backend/app/providers/imap_smtp/parsing.py b/backend/app/providers/imap_smtp/parsing.py index f89bc4c..99a97ee 100644 --- a/backend/app/providers/imap_smtp/parsing.py +++ b/backend/app/providers/imap_smtp/parsing.py @@ -135,6 +135,41 @@ def parse_fetch_lines(lines: list[bytes | bytearray]) -> list[ParsedFetch]: return results +_VANISHED_RE = re.compile(rb"^VANISHED\s+(?:\(EARLIER\)\s+)?(\S+)", re.IGNORECASE) + + +def _expand_uid_set(token: bytes) -> list[int]: + """Expand an IMAP uid-set (``1,3:5,9``) to a flat list of ints (``*`` skipped).""" + out: list[int] = [] + for part in token.split(b","): + part = part.strip() + lo_b, sep, hi_b = part.partition(b":") + if sep: + if lo_b.isdigit() and hi_b.isdigit(): + lo, hi = int(lo_b), int(hi_b) + out.extend(range(min(lo, hi), max(lo, hi) + 1)) + elif part.isdigit(): + out.append(int(part)) + return out + + +def parse_vanished(lines: list[bytes | bytearray]) -> list[int]: + """UIDs from QRESYNC ``VANISHED [(EARLIER)] `` responses (RFC 7162). + + aioimaplib strips the leading ``* `` before buffering the line, so we match on + the bare ``VANISHED`` token. Both the SELECT-time ``(EARLIER)`` form and the + live-expunge form are accepted. + """ + out: list[int] = [] + for line in lines: + if not isinstance(line, (bytes, bytearray)): + continue + m = _VANISHED_RE.match(bytes(line).strip()) + if m: + out.extend(_expand_uid_set(m.group(1))) + return out + + def parse_search_uids(lines: list[bytes | bytearray]) -> list[int]: """Parse a UID SEARCH response's first line into a list of ints.""" if not lines: diff --git a/backend/app/providers/imap_smtp/provider.py b/backend/app/providers/imap_smtp/provider.py index 8c696ff..d3df7e3 100644 --- a/backend/app/providers/imap_smtp/provider.py +++ b/backend/app/providers/imap_smtp/provider.py @@ -35,6 +35,7 @@ from app.providers.imap_smtp.parsing import ( parse_list_line, parse_search_uids, parse_select_untagged, + parse_vanished, ) _FETCH_ITEMS = "(UID FLAGS INTERNALDATE RFC822.SIZE BODY.PEEK[])" @@ -79,6 +80,15 @@ class ImapSmtpProvider(MailProvider): self._selected: str | None = None self._uidvalidity_seen: dict[str, int] = {} self._conn_slots: list[asyncio.Semaphore] = [] + self._qresync_enabled = False + + @property + def qresync_enabled(self) -> bool: + """True once ``ENABLE QRESYNC`` succeeded on this connection (RFC 7162). + + When set, :meth:`poll` reports server-side expunges via VANISHED; otherwise + deletion detection falls back to the engine's full-UID reconcile.""" + return self._qresync_enabled # --- connection lifecycle ------------------------------------------------ @@ -138,8 +148,23 @@ class ImapSmtpProvider(MailProvider): self._release_conn_slots() # never hold a slot without an open socket raise self._imap = imap + await self._maybe_enable_qresync(imap) return imap + async def _maybe_enable_qresync(self, imap: aioimaplib.IMAP4) -> None: + """Enable QRESYNC when advertised so SELECT reports VANISHED (RFC 7162). + + Best-effort: servers without QRESYNC (e.g. Gmail, which is CONDSTORE-only) + simply keep ``_qresync_enabled`` False and use the engine's reconcile path.""" + caps = {c.upper() for c in (imap.protocol.capabilities or set())} + if "QRESYNC" not in caps: + return + try: + resp = await imap.enable("QRESYNC") + self._qresync_enabled = resp.result == "OK" + except Exception: # noqa: BLE001 - degrade gracefully to the reconcile path + self._qresync_enabled = False + async def _select(self, mailbox: str) -> dict[str, int]: imap = await self._connect() resp = await imap.select(mailbox) @@ -287,26 +312,53 @@ class ImapSmtpProvider(MailProvider): search = await imap.uid_search(f"UID {last_seen_uid + 1}:*") candidate_uids = [u for u in parse_search_uids(search.lines) if u > last_seen_uid] new_messages = await self._fetch_messages(imap, sorted(candidate_uids)) + new_uid_set = {m.uid for m in new_messages} - # Flag changes via CONDSTORE when we have a modseq baseline. flag_changes: list[FlagChange] = [] - if highestmodseq is not None: + expunged_uids: list[int] = [] + if highestmodseq is not None and self._qresync_enabled: + # QRESYNC fast resync (RFC 7162): a re-SELECT with (QRESYNC (uidvalidity + # modseq)) makes the server report, in one pass, the FLAGS changed and the + # UIDs VANISHED since ``modseq``. VANISHED is only surfaced on a *sync* + # command, so we harvest both from the SELECT response here. + qr = await self._qresync_select(mailbox, current_uidvalidity, highestmodseq) + expunged_uids = [u for u in parse_vanished(qr.lines) if u not in new_uid_set] + for pf in parse_fetch_lines(qr.lines): + if pf.uid not in new_uid_set: + flag_changes.append(FlagChange(uid=pf.uid, flags=pf.flags)) + elif highestmodseq is not None: + # CONDSTORE-only server (e.g. Gmail): flag delta via CHANGEDSINCE. There + # is no VANISHED here — the engine reconciles expunges by full-UID diff. fresp = await imap.uid("fetch", "1:*", f"(UID FLAGS) (CHANGEDSINCE {highestmodseq})") if fresp.result == "OK": - new_uid_set = {m.uid for m in new_messages} for pf in parse_fetch_lines(fresp.lines): if pf.uid not in new_uid_set: flag_changes.append(FlagChange(uid=pf.uid, flags=pf.flags)) - # TODO(spec): expunge detection needs QRESYNC/VANISHED (M2 hardening). return MailboxDelta( new=new_messages, flag_changes=flag_changes, - expunged_uids=[], + expunged_uids=expunged_uids, new_uidvalidity=None, new_highestmodseq=info.get("highestmodseq"), ) + async def _qresync_select( + self, mailbox: str, uidvalidity: int, modseq: int + ) -> aioimaplib.Response: + """Raw ``SELECT (QRESYNC (uidvalidity modseq))`` (RFC 7162). + + Issued as a low-level command because aioimaplib's high-level ``select`` + takes no parameters, and because SELECT is a *sync* command — the only kind + whose untagged responses (VANISHED, changed FETCH) aioimaplib buffers.""" + imap = await self._connect() + proto = imap.protocol + param = f"(QRESYNC ({uidvalidity} {modseq}))" + cmd = aioimaplib.Command( + "SELECT", proto.new_tag(), aioimaplib.quoted(mailbox), param, loop=proto.loop + ) + return await proto.execute(cmd) + async def idle(self, mailbox: str) -> None: imap = await self._connect() if "IDLE" not in (imap.protocol.capabilities or set()): diff --git a/backend/app/sync/engine.py b/backend/app/sync/engine.py index f1c3772..8d20d0e 100644 --- a/backend/app/sync/engine.py +++ b/backend/app/sync/engine.py @@ -323,6 +323,7 @@ async def apply_delta( results: list[PersistResult] = [] max_uid = mailbox.last_seen_uid + gone_ids: list[uuid.UUID] = [] async with session_scope() as session: for remote in delta.new: r = await persist_message( @@ -343,17 +344,49 @@ async def apply_delta( .values(flags=sorted(change.flags)) ) if delta.expunged_uids: + # Server-side expunges (QRESYNC/VANISHED, SPEC §6.3). Capture the row ids + # first so we can drop them from the search index after the delete. + gone_ids = list( + await session.scalars( + select(Message.id).where( + Message.mailbox_id == mailbox.id, Message.uid.in_(delta.expunged_uids) + ) + ) + ) await session.execute( delete(Message).where( Message.mailbox_id == mailbox.id, Message.uid.in_(delta.expunged_uids) ) ) + # A local draft whose mirrored copy vanished reverts to local-only. + await session.execute( + update(Draft) + .where(Draft.account_id == account.id, Draft.imap_uid.in_(delta.expunged_uids)) + .values(imap_uid=None) + ) updates = {"last_seen_uid": max_uid} if delta.new_highestmodseq is not None: updates["highestmodseq"] = delta.new_highestmodseq await session.execute(update(Mailbox).where(Mailbox.id == mailbox.id).values(**updates)) await _emit_batch(results, account, mailbox.path, len(results)) + + if gone_ids: + # Drop expunged messages from the search index (PG keeps its tsvector on the + # row, so this is a no-op there; Meili needs the explicit removal) and tell + # the UI to refresh, mirroring the drafts-reconcile notification. + backend = get_search_backend() + for mid in gone_ids: + await backend.delete(mid) + await bus.publish( + "message.deleted", + { + "user_id": str(account.user_id), + "account_id": str(account.id), + "mailbox": mailbox.path, + }, + ) + for change in delta.flag_changes: await bus.publish( "flags.changed", diff --git a/backend/tests/providers/test_parsing.py b/backend/tests/providers/test_parsing.py index ee79524..5d5f2ed 100644 --- a/backend/tests/providers/test_parsing.py +++ b/backend/tests/providers/test_parsing.py @@ -10,6 +10,7 @@ from app.providers.imap_smtp.parsing import ( parse_list_line, parse_search_uids, parse_select_untagged, + parse_vanished, ) @@ -57,6 +58,16 @@ def test_parse_search_uids() -> None: assert parse_search_uids([]) == [] +def test_parse_vanished() -> None: + # aioimaplib strips the leading "* " before buffering the line. + assert parse_vanished([b"VANISHED (EARLIER) 300:305,405"]) == [300, 301, 302, 303, 304, 305, 405] # noqa: E501 + # Live-expunge form (no EARLIER) and a descending range are both handled. + assert parse_vanished([b"VANISHED 41", b"VANISHED (EARLIER) 5:3"]) == [41, 3, 4, 5] + # Non-VANISHED lines are ignored; non-bytes entries are skipped. + assert parse_vanished([b"12 FETCH (UID 12 FLAGS (\\Seen))", b"OK done"]) == [] + assert parse_vanished([]) == [] + + def test_parse_select_untagged() -> None: lines = [ b"6 EXISTS", diff --git a/backend/tests/sync/test_secondary_poll.py b/backend/tests/sync/test_secondary_poll.py index 1c2b28c..4da6cee 100644 --- a/backend/tests/sync/test_secondary_poll.py +++ b/backend/tests/sync/test_secondary_poll.py @@ -167,3 +167,77 @@ async def test_drafts_poll_reconciles_expunged_rows(monkeypatch) -> None: # typ await cleanup.execute(delete(User).where(User.id == user.id)) await session.close() await engine.dispose() + + +async def test_apply_delta_expunges_vanished_uids() -> None: # type: ignore[no-untyped-def] + """A QRESYNC delta carrying ``expunged_uids`` removes those rows, drops them from + the search index (tsvector cleared), and un-links any local draft that mirrored + a vanished UID — the VANISHED path (SPEC §6.3).""" + from app.models import Message as MessageModel + + engine = create_async_engine(get_settings().database_url, poolclass=NullPool) + sm = async_sessionmaker(engine, expire_on_commit=False) + session = sm() + user = User(handle=f"van{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")) + session.add(user) + await session.flush() + account = Account( + user_id=user.id, + email="me@example.com", + imap_config={}, + smtp_config={}, + secret_ct=b"", + discovery_src="manual", + ) + session.add(account) + await session.flush() + inbox = Mailbox( + account_id=account.id, + user_id=user.id, + path="INBOX", + special_use="inbox", + last_seen_uid=10, + ) + session.add(inbox) + await session.flush() + for uid in (7, 8, 9): + session.add( + MessageModel( + user_id=user.id, + account_id=account.id, + mailbox_id=inbox.id, + uid=uid, + internal_date=datetime.now(UTC), + blob_sha256="", + ) + ) + linked = Draft(user_id=user.id, account_id=account.id, kind="new", payload={}, imap_uid=9) + session.add(linked) + await session.commit() + + class FakeProvider: + qresync_enabled = True + + async def close(self) -> None: + return None + + # uids 7 and 9 vanished on the server; 8 survives. + delta = MailboxDelta([], [], [7, 9], None, 4242) + await sync_engine.apply_delta(FakeProvider(), account, inbox, delta) # type: ignore[arg-type] + + try: + session.expunge_all() + remaining = set( + await session.scalars( + select(MessageModel.uid).where(MessageModel.mailbox_id == inbox.id) + ) + ) + assert remaining == {8} # 7 and 9 expunged + assert (await session.get(Draft, linked.id)).imap_uid is None # un-linked + refreshed = await session.get(Mailbox, inbox.id) + assert refreshed.highestmodseq == 4242 # modseq baseline advanced + finally: + async with session_scope() as cleanup: + await cleanup.execute(delete(User).where(User.id == user.id)) + await session.close() + await engine.dispose()