diff --git a/backend/app/services/mail_processor.py b/backend/app/services/mail_processor.py index 63e373c..d9e40b7 100644 --- a/backend/app/services/mail_processor.py +++ b/backend/app/services/mail_processor.py @@ -213,7 +213,6 @@ class MailProcessor: fetched: List[bytes] = [] fetched_uids: List[str] = [] - messages_to_delete: List[int] = [] fetched_count = 0 for msg_num, uid in uid_map.items(): @@ -233,7 +232,6 @@ class MailProcessor: email_data = b"\r\n".join(lines) fetched.append(email_data) fetched_uids.append(uid) - messages_to_delete.append(msg_num) fetched_count += 1 logger.info( f"Retrieved message {msg_num} (uid={uid}) " @@ -242,14 +240,6 @@ class MailProcessor: except Exception as e: logger.error(f"Error retrieving message {msg_num}: {e}") - # Delete messages if configured - if self.account.delete_after_forward: - for msg_id in messages_to_delete: - try: - pop_conn.dele(msg_id) - except Exception as e: - logger.error(f"Error deleting message {msg_id}: {e}") - pop_conn.quit() return fetched, fetched_uids @@ -370,19 +360,23 @@ class MailProcessor: f"{self.account.id}: {e}" ) - # Fetch each new message. RFC822 FETCH implicitly sets \Seen on - # the server so no extra STORE per message is required. - fetched_uids: List[bytes] = [] + # Fetch each new message using BODY.PEEK[] so the \Seen flag is NOT + # set implicitly. Successfully forwarded messages are marked \Seen + # (and optionally \Deleted) afterwards via post_process_imap(). + # This ensures that emails which fail to forward remain \Unseen and + # are therefore retried on the next SEARCH UNSEEN run. for uid in uids_to_fetch: uid_str = uid.decode() if isinstance(uid, bytes) else str(uid) try: - fetch_response = await imap_client.uid("fetch", uid_str, "(RFC822)") + fetch_response = await imap_client.uid( + "fetch", uid_str, "(BODY.PEEK[])" + ) # Extract raw email bytes from the FETCH response lines. # A UID FETCH response looks like: - # [b' (UID RFC822 {}', , b')'] - # Skip the header line (contains "RFC822") and grab the - # first substantive bytes value. + # [b' (UID BODY[] {}', , b')'] + # Skip the status/header line (contains "BODY[" or "RFC822") + # and grab the first substantive bytes value. # # aioimaplib stores IMAP literal data (the actual email # content) as bytearray, not bytes. We must accept both @@ -392,7 +386,7 @@ class MailProcessor: for line in fetch_response.lines: if not isinstance(line, (bytes, bytearray)): continue - if b"RFC822" in line: + if b"RFC822" in line or b"BODY[" in line: continue if line.startswith(b"*"): continue @@ -404,14 +398,13 @@ class MailProcessor: if email_data and email_data.strip(): emails.append(email_data) new_uids.append(uid_str) - fetched_uids.append(uid) else: logger.warning( f"No email data extracted for UID {uid_str} on " f"account {self.account.id} " f"(raw bytes: {len(email_data) if email_data else 0}); " "this may indicate a server-side error producing empty " - "IMAP RFC822 responses" + "IMAP responses" ) except Exception as e: @@ -420,20 +413,6 @@ class MailProcessor: f"{self.account.id}: {e}" ) - # Batch-mark fetched messages for deletion if configured (one STORE - # command covers all UIDs instead of N individual commands). - # RFC 3501 requires parentheses around flag names: +FLAGS (\Deleted). - if self.account.delete_after_forward and fetched_uids: - uid_set = b",".join(fetched_uids).decode() - try: - await imap_client.uid("store", uid_set, "+FLAGS", "(\\Deleted)") - await imap_client.expunge() - except Exception as e: - logger.warning( - f"Failed to delete messages for account " - f"{self.account.id}: {e}" - ) - except MailFetchError: raise except Exception as e: @@ -453,8 +432,144 @@ class MailProcessor: return emails, new_uids - @staticmethod - async def forward_email( + async def post_process_imap( + self, successfully_forwarded_uids: List[str] + ) -> None: + """Mark successfully forwarded IMAP messages as \\Seen and optionally delete. + + Because fetch uses ``BODY.PEEK[]`` (which does NOT set \\Seen), this + step is required so successfully processed messages stop appearing in + ``SEARCH UNSEEN``. Messages that failed to forward are intentionally + left \\Unseen so they are retried on the next processing run. + + If ``delete_after_forward`` is enabled, the messages are also flagged + \\Deleted and the mailbox is expunged. + """ + if not successfully_forwarded_uids: + return + + imap_client = None + try: + if self.account.protocol == MailProtocol.IMAP_SSL: + imap_client = aioimaplib.IMAP4_SSL( + host=self.account.host, port=self.account.port, timeout=30 + ) + else: + imap_client = aioimaplib.IMAP4( + host=self.account.host, port=self.account.port, timeout=30 + ) + + await imap_client.wait_hello_from_server() + await imap_client.login(self.account.username, self.password) + await imap_client.select("INBOX") + + uid_set = ",".join(successfully_forwarded_uids) + + # Mark as \Seen so they no longer appear in SEARCH UNSEEN. + await imap_client.uid("store", uid_set, "+FLAGS", "(\\Seen)") + + if self.account.delete_after_forward: + await imap_client.uid("store", uid_set, "+FLAGS", "(\\Deleted)") + await imap_client.expunge() + + except Exception as e: + logger.warning( + "Failed to post-process IMAP messages for account %s: %s", + self.account.id, + e, + ) + finally: + if imap_client is not None: + try: + await imap_client.logout() + except Exception: + pass + + async def post_process_pop3( + self, successfully_forwarded_uids: List[str] + ) -> None: + """Delete successfully forwarded POP3 messages from the source mailbox. + + Opens a fresh POP3 session, maps stable UIDs back to current message + numbers via UIDL, and deletes only the messages that were successfully + forwarded. Messages that failed to forward are intentionally left on + the server so they are retried on the next processing run. + + This method is a no-op when ``delete_after_forward`` is ``False``. + """ + if not successfully_forwarded_uids or not self.account.delete_after_forward: + return + + loop = asyncio.get_event_loop() + uid_set = set(successfully_forwarded_uids) + + def _delete_pop3() -> None: + if self.account.protocol == MailProtocol.POP3_SSL: + context = ssl.create_default_context() + pop_conn: poplib.POP3 = poplib.POP3_SSL( + str(self.account.host), + int(self.account.port), + context=context, + timeout=30, + ) + else: + pop_conn = poplib.POP3( + str(self.account.host), int(self.account.port), timeout=30 + ) + + pop_conn.user(str(self.account.username)) + pop_conn.pass_(self.password) + + uidl_response = pop_conn.uidl() + for entry in uidl_response[1]: + parts = entry.decode().split(" ", 1) + if len(parts) != 2: + continue + msg_num = int(parts[0]) + uid = parts[1].strip() + if uid in uid_set: + try: + pop_conn.dele(msg_num) + except Exception as e: + logger.error( + "Error deleting POP3 message %s (uid=%s) " + "for account %s: %s", + msg_num, + uid, + self.account.id, + e, + ) + + pop_conn.quit() + + try: + await loop.run_in_executor(None, _delete_pop3) + except Exception as e: + logger.warning( + "Failed to delete POP3 messages for account %s: %s", + self.account.id, + e, + ) + + async def post_process_messages( + self, successfully_forwarded_uids: List[str] + ) -> None: + """Post-process successfully forwarded messages. + + Routes to the protocol-specific post-processor: + * IMAP: marks messages \\Seen (and \\Deleted + expunges if configured). + * POP3: deletes messages from the source mailbox if configured. + + Must be called *after* the forwarding loop so that messages which + failed to forward remain untouched in the source mailbox and are + retried on the next run. + """ + if self.account.protocol in [MailProtocol.POP3, MailProtocol.POP3_SSL]: + await self.post_process_pop3(successfully_forwarded_uids) + else: + await self.post_process_imap(successfully_forwarded_uids) + + email_data: bytes, source_account_name: str, destination: str, diff --git a/backend/app/workers/tasks.py b/backend/app/workers/tasks.py index 3928cf7..65f32e2 100644 --- a/backend/app/workers/tasks.py +++ b/backend/app/workers/tasks.py @@ -313,6 +313,15 @@ async def process_mail_account(account_id: int): emails_forwarded += 1 successfully_forwarded_uids.append(uid) else: + logger.warning( + "Email delivery returned False for uid=%s " + "on account %s (subject=%r, from=%r); " + "message will be retried on next run", + uid, + account.id, + email_subject, + email_from, + ) emails_failed += 1 except Exception as e: @@ -371,6 +380,20 @@ async def process_mail_account(account_id: int): ) ) + # Post-process: mark successfully forwarded messages as \Seen (IMAP) + # and/or delete them from the source mailbox. This is done AFTER + # the forwarding loop so that any message that failed to forward is + # left untouched in the source and will be retried on the next run. + if successfully_forwarded_uids: + try: + await processor.post_process_messages(successfully_forwarded_uids) + except Exception as post_exc: + logger.warning( + "Failed to post-process messages for account %s: %s", + account.id, + post_exc, + ) + # Persist new message UIDs so they are not processed again for uid in successfully_forwarded_uids: if uid not in already_seen_uids: diff --git a/backend/tests/unit/test_mail_processor_imap.py b/backend/tests/unit/test_mail_processor_imap.py index 9c148b3..4485439 100644 --- a/backend/tests/unit/test_mail_processor_imap.py +++ b/backend/tests/unit/test_mail_processor_imap.py @@ -163,7 +163,7 @@ class TestFetchImapEmailsUidCommands: # The fetch call should use UID FETCH fetch_call = [c for c in mock_imap.uid.call_args_list if c.args[0] == "fetch"] assert len(fetch_call) == 1 - assert fetch_call[0] == call("fetch", "42", "(RFC822)") + assert fetch_call[0] == call("fetch", "42", "(BODY.PEEK[])") async def test_no_per_message_store_for_seen(self, processor, mock_imap): """RFC822 implicitly marks \\Seen; no extra STORE per message is needed.""" @@ -260,8 +260,12 @@ class TestFetchImapEmailsUidCommands: # RFC 3501 parenthesised flag syntax assert store_calls[0].args[3] == "(\\Seen)" - async def test_delete_after_forward_uses_single_batch_store(self): - """delete_after_forward=True must issue one UID STORE \\Deleted command.""" + async def test_fetch_does_not_delete_after_forward(self): + """_fetch_imap_emails must NOT issue \\Deleted STORE even when delete_after_forward=True. + + Deletion is deferred to post_process_imap() so that only successfully + forwarded messages are removed from the source mailbox. + """ from app.services.mail_processor import MailProcessor account = _make_account(delete_after_forward=True) @@ -295,17 +299,15 @@ class TestFetchImapEmailsUidCommands: assert new_uids == ["7", "8"] - store_calls = [c for c in mock_imap.uid.call_args_list if c.args[0] == "store"] - # Exactly one STORE for deletion covering both UIDs - assert len(store_calls) == 1 - uid_set_arg = store_calls[0].args[1] - parts = set(uid_set_arg.split(",")) - assert parts == {"7", "8"} - assert store_calls[0].args[2] == "+FLAGS" - # RFC 3501 parenthesised flag syntax - assert store_calls[0].args[3] == "(\\Deleted)" - # expunge must also be called - mock_imap.expunge.assert_awaited_once() + # No STORE for \\Deleted should be issued during fetch + delete_store_calls = [ + c + for c in mock_imap.uid.call_args_list + if c.args[0] == "store" and "(\\Deleted)" in c.args + ] + assert delete_store_calls == [], "fetch must not delete messages" + # expunge must NOT be called during fetch either + mock_imap.expunge.assert_not_awaited() async def test_individual_fetch_failure_does_not_abort(self, processor, mock_imap): """A single message fetch error should be logged but not stop processing.""" @@ -709,3 +711,170 @@ class TestFetchImapEdgeCases: assert len(emails) == 1 assert emails[0] == email_bytes assert new_uids == ["99"] + + +# --------------------------------------------------------------------------- +# post_process_imap tests +# --------------------------------------------------------------------------- + + +class TestPostProcessImap: + """Tests for the post_process_imap() method.""" + + def _make_processor(self, delete_after_forward=False, protocol="imap_ssl"): + from app.services.mail_processor import MailProcessor + + account = _make_account( + delete_after_forward=delete_after_forward, protocol=protocol + ) + return MailProcessor(account=account, decrypted_password="secret") + + def _mock_imap_client(self): + client = MagicMock() + client.wait_hello_from_server = AsyncMock() + client.login = AsyncMock() + client.select = AsyncMock() + client.uid = AsyncMock(return_value=_make_imap_response()) + client.expunge = AsyncMock() + client.logout = AsyncMock() + return client + + async def test_marks_seen_without_delete(self): + """post_process_imap marks UIDs \\Seen when delete_after_forward=False.""" + processor = self._make_processor(delete_after_forward=False) + mock_imap = self._mock_imap_client() + + with patch( + "app.services.mail_processor.aioimaplib.IMAP4_SSL", + return_value=mock_imap, + ): + await processor.post_process_imap(["10", "11"]) + + store_calls = [c for c in mock_imap.uid.call_args_list if c.args[0] == "store"] + assert len(store_calls) == 1 + assert set(store_calls[0].args[1].split(",")) == {"10", "11"} + assert store_calls[0].args[3] == "(\\Seen)" + mock_imap.expunge.assert_not_awaited() + + async def test_marks_seen_and_deleted_with_delete(self): + """post_process_imap marks \\Seen + \\Deleted and expunges when configured.""" + processor = self._make_processor(delete_after_forward=True) + mock_imap = self._mock_imap_client() + + with patch( + "app.services.mail_processor.aioimaplib.IMAP4_SSL", + return_value=mock_imap, + ): + await processor.post_process_imap(["5", "6"]) + + store_calls = [c for c in mock_imap.uid.call_args_list if c.args[0] == "store"] + assert len(store_calls) == 2 + flags_used = {c.args[3] for c in store_calls} + assert "(\\Seen)" in flags_used + assert "(\\Deleted)" in flags_used + mock_imap.expunge.assert_awaited_once() + + async def test_noop_on_empty_uid_list(self): + """post_process_imap does nothing when the UID list is empty.""" + processor = self._make_processor() + mock_imap = self._mock_imap_client() + + with patch( + "app.services.mail_processor.aioimaplib.IMAP4_SSL", + return_value=mock_imap, + ): + await processor.post_process_imap([]) + + mock_imap.login.assert_not_awaited() + + async def test_exception_is_swallowed_and_logged(self): + """post_process_imap catches exceptions and does not propagate them.""" + processor = self._make_processor() + mock_imap = self._mock_imap_client() + mock_imap.login = AsyncMock(side_effect=OSError("connection refused")) + + with patch( + "app.services.mail_processor.aioimaplib.IMAP4_SSL", + return_value=mock_imap, + ): + await processor.post_process_imap(["1"]) # must not raise + + mock_imap.logout.assert_awaited_once() + + async def test_uses_imap4_for_non_ssl(self): + """Plain IMAP accounts must use IMAP4, not IMAP4_SSL.""" + from app.services.mail_processor import MailProcessor + + account = _make_account(protocol="imap") + processor = MailProcessor(account=account, decrypted_password="pw") + mock_imap = MagicMock() + mock_imap.wait_hello_from_server = AsyncMock() + mock_imap.login = AsyncMock() + mock_imap.select = AsyncMock() + mock_imap.uid = AsyncMock(return_value=_make_imap_response()) + mock_imap.logout = AsyncMock() + + with patch( + "app.services.mail_processor.aioimaplib.IMAP4", + return_value=mock_imap, + ) as mock_cls, patch( + "app.services.mail_processor.aioimaplib.IMAP4_SSL" + ) as mock_ssl_cls: + await processor.post_process_imap(["1"]) + mock_cls.assert_called_once() + mock_ssl_cls.assert_not_called() + + async def test_logout_called_even_on_error(self): + """logout() is called in the finally block even when an exception occurs.""" + processor = self._make_processor() + mock_imap = self._mock_imap_client() + mock_imap.uid = AsyncMock(side_effect=Exception("store failed")) + + with patch( + "app.services.mail_processor.aioimaplib.IMAP4_SSL", + return_value=mock_imap, + ): + await processor.post_process_imap(["99"]) + + mock_imap.logout.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# post_process_messages routing test +# --------------------------------------------------------------------------- + + +class TestPostProcessMessages: + """post_process_messages() routes to the correct protocol handler.""" + + async def test_routes_to_imap_for_imap_ssl(self): + from app.services.mail_processor import MailProcessor + + account = _make_account(protocol="imap_ssl") + processor = MailProcessor(account=account, decrypted_password="pw") + with patch.object( + processor, "post_process_imap", new_callable=AsyncMock + ) as mock_imap, patch.object( + processor, "post_process_pop3", new_callable=AsyncMock + ) as mock_pop3: + await processor.post_process_messages(["1", "2"]) + mock_imap.assert_awaited_once_with(["1", "2"]) + mock_pop3.assert_not_awaited() + + async def test_routes_to_pop3_for_pop3_ssl(self): + from app.services.mail_processor import MailProcessor, MailProcessor + + account = _make_account(protocol="imap_ssl") + account.protocol = __import__( + "app.models.database_models", fromlist=["MailProtocol"] + ).MailProtocol.POP3_SSL + processor = MailProcessor(account=account, decrypted_password="pw") + with patch.object( + processor, "post_process_pop3", new_callable=AsyncMock + ) as mock_pop3, patch.object( + processor, "post_process_imap", new_callable=AsyncMock + ) as mock_imap: + await processor.post_process_messages(["a"]) + mock_pop3.assert_awaited_once_with(["a"]) + mock_imap.assert_not_awaited() + diff --git a/backend/tests/unit/test_mail_processor_pop3.py b/backend/tests/unit/test_mail_processor_pop3.py index f01ab2e..817eaf0 100644 --- a/backend/tests/unit/test_mail_processor_pop3.py +++ b/backend/tests/unit/test_mail_processor_pop3.py @@ -438,8 +438,12 @@ class TestFetchPop3Emails: assert len(uids) == 2 @patch("app.services.mail_processor.poplib") - async def test_delete_after_forward(self, mock_poplib): - """delete_after_forward=True issues dele() for fetched messages.""" + async def test_no_delete_during_fetch(self, mock_poplib): + """_fetch_pop3_emails must NOT call dele() regardless of delete_after_forward. + + Deletion is deferred to post_process_pop3() so that only successfully + forwarded messages are removed from the source mailbox. + """ mock_conn = self._make_pop3_mock(uid_entries=[(1, "a"), (2, "b")]) mock_poplib.POP3_SSL.return_value = mock_conn @@ -447,9 +451,7 @@ class TestFetchPop3Emails: proc = MailProcessor(account, "secret") await proc._fetch_pop3_emails(10, set()) - assert mock_conn.dele.call_count == 2 - mock_conn.dele.assert_any_call(1) - mock_conn.dele.assert_any_call(2) + mock_conn.dele.assert_not_called() @patch("app.services.mail_processor.poplib") async def test_no_delete_when_disabled(self, mock_poplib): @@ -483,17 +485,16 @@ class TestFetchPop3Emails: assert "b" not in uids @patch("app.services.mail_processor.poplib") - async def test_delete_error_does_not_abort(self, mock_poplib): - """A dele() error is logged but doesn't raise.""" + async def test_fetch_does_not_call_dele(self, mock_poplib): + """_fetch_pop3_emails never calls dele() — deletion is in post_process_pop3.""" mock_conn = self._make_pop3_mock(uid_entries=[(1, "a")]) - mock_conn.dele.side_effect = Exception("delete failed") mock_poplib.POP3_SSL.return_value = mock_conn account = _make_account(protocol="pop3_ssl", delete_after_forward=True) proc = MailProcessor(account, "secret") - # Should not raise emails, uids = await proc._fetch_pop3_emails(10, set()) assert len(emails) == 1 + mock_conn.dele.assert_not_called() @patch("app.services.mail_processor.poplib") async def test_connection_failure_raises_mail_fetch_error(self, mock_poplib): @@ -559,10 +560,81 @@ class TestFetchPop3Emails: # --------------------------------------------------------------------------- -# forward_email +# post_process_pop3 tests # --------------------------------------------------------------------------- +class TestPostProcessPop3: + """Unit tests for post_process_pop3().""" + + @patch("app.services.mail_processor.poplib") + async def test_deletes_only_forwarded_uids(self, mock_poplib): + """Only successfully forwarded UIDs are deleted from the source mailbox.""" + mock_conn = MagicMock() + mock_conn.uidl.return_value = (b"+OK", [b"1 uid-a", b"2 uid-b", b"3 uid-c"], 0) + mock_poplib.POP3_SSL.return_value = mock_conn + + account = _make_account(protocol="pop3_ssl", delete_after_forward=True) + proc = MailProcessor(account, "secret") + await proc.post_process_pop3(["uid-a", "uid-c"]) + + # Only messages 1 and 3 should be deleted (uid-a and uid-c) + assert mock_conn.dele.call_count == 2 + mock_conn.dele.assert_any_call(1) + mock_conn.dele.assert_any_call(3) + mock_conn.quit.assert_called_once() + + @patch("app.services.mail_processor.poplib") + async def test_noop_when_delete_after_forward_false(self, mock_poplib): + """post_process_pop3 does nothing when delete_after_forward=False.""" + mock_conn = MagicMock() + mock_poplib.POP3_SSL.return_value = mock_conn + + account = _make_account(protocol="pop3_ssl", delete_after_forward=False) + proc = MailProcessor(account, "secret") + await proc.post_process_pop3(["uid-a"]) + + mock_poplib.POP3_SSL.assert_not_called() + + @patch("app.services.mail_processor.poplib") + async def test_noop_on_empty_uid_list(self, mock_poplib): + """post_process_pop3 does nothing when the UID list is empty.""" + mock_conn = MagicMock() + mock_poplib.POP3_SSL.return_value = mock_conn + + account = _make_account(protocol="pop3_ssl", delete_after_forward=True) + proc = MailProcessor(account, "secret") + await proc.post_process_pop3([]) + + mock_poplib.POP3_SSL.assert_not_called() + + @patch("app.services.mail_processor.poplib") + async def test_dele_error_does_not_abort(self, mock_poplib): + """A dele() error is logged but post_process_pop3 does not raise.""" + mock_conn = MagicMock() + mock_conn.uidl.return_value = (b"+OK", [b"1 uid-a"], 0) + mock_conn.dele.side_effect = Exception("delete failed") + mock_poplib.POP3_SSL.return_value = mock_conn + + account = _make_account(protocol="pop3_ssl", delete_after_forward=True) + proc = MailProcessor(account, "secret") + await proc.post_process_pop3(["uid-a"]) # must not raise + + mock_conn.quit.assert_called_once() + + @patch("app.services.mail_processor.poplib") + async def test_connection_error_is_swallowed(self, mock_poplib): + """A connection failure is logged but does not propagate.""" + mock_poplib.POP3_SSL.side_effect = OSError("connection refused") + + account = _make_account(protocol="pop3_ssl", delete_after_forward=True) + proc = MailProcessor(account, "secret") + await proc.post_process_pop3(["uid-a"]) # must not raise + + + + + class TestForwardEmail: """Unit tests for forward_email().""" diff --git a/backend/tests/unit/test_tasks.py b/backend/tests/unit/test_tasks.py index 56bea67..4a2d330 100644 --- a/backend/tests/unit/test_tasks.py +++ b/backend/tests/unit/test_tasks.py @@ -589,6 +589,7 @@ class TestProcessMailAccount: session.refresh = AsyncMock() mock_processor = AsyncMock() + mock_processor.post_process_messages = AsyncMock() # Two emails: first succeeds, second fails mock_processor.fetch_emails.return_value = ( [raw_email, raw_email], @@ -612,6 +613,8 @@ class TestProcessMailAccount: await process_mail_account.run(1) assert session.commit.await_count >= 2 + # Only the successfully forwarded UID should be post-processed + mock_processor.post_process_messages.assert_awaited_once_with(["uid-1"]) @pytest.mark.asyncio async def test_gmail_credential_revocation_on_401(self):