Merge pull request #183 from christianlouis/codex/m12-m365-safe-backfills

feat: add M365 safe backfill windows

backend/app/api/api_v1/endpoints/mail_sources.py

@@ -12,7 +12,7 @@ import logging
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
-from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from pydantic import BaseModel
 from sqlalchemy.orm import Session
 
@@ -446,6 +446,7 @@ def _fetch_response(source: MailSource, results: Dict[str, Any]) -> Dict[str, An
         "error_count": len(results.get("errors", [])),
         "target_mailbox": results.get("target_mailbox"),
         "target_folder": results.get("target_folder"),
+        "search_window_days": results.get("search_window_days"),
         "timestamp": datetime.now().isoformat(),
     }
 
@@ -487,7 +488,7 @@ def _fetch_gmail_source(source: MailSource, db: Session) -> Dict[str, Any]:
     return results
 
 
-def _fetch_m365_source(source: MailSource, db: Session) -> Dict[str, Any]:
+def _fetch_m365_source(source: MailSource, db: Session, days: int = 7) -> Dict[str, Any]:
     """Run one Microsoft 365 Graph import and persist source/import metadata."""
     if not source.m365_access_token:
         raise HTTPException(
@@ -510,7 +511,7 @@ def _fetch_m365_source(source: MailSource, db: Session) -> Dict[str, Any]:
     )
 
     started_at = datetime.utcnow()
-    results = client.fetch_reports()
+    results = client.fetch_reports(days=days)
 
     if results.get("new_ingested_ids"):
         all_ids = list(dict.fromkeys(already + results["new_ingested_ids"]))
@@ -551,7 +552,7 @@ def _fetch_source(source: MailSource, db: Session, days: int) -> Dict[str, Any]:
     if source.method == "GMAIL_API":
         return _fetch_gmail_source(source, db)
     if source.method == "M365_GRAPH":
-        return _fetch_m365_source(source, db)
+        return _fetch_m365_source(source, db, days=days)
     if source.method == "IMAP":
         return _fetch_imap_source(source, db, days)
     raise HTTPException(
@@ -1113,6 +1114,7 @@ async def m365_oauth_callback_post(
 @router.post("/{source_id}/m365/fetch", response_model=Dict[str, Any])
 async def m365_fetch_reports(
     source_id: int,
+    days: int = Query(7, ge=1, le=365, title="Number of days to fetch"),
     db: Session = Depends(get_db),
     _auth: dict = Depends(require_admin_auth),
 ) -> Dict[str, Any]:
@@ -1125,7 +1127,7 @@ async def m365_fetch_reports(
             detail="This endpoint is only available for M365_GRAPH sources.",
         )
 
-    results = _fetch_m365_source(source, db)
+    results = _fetch_m365_source(source, db, days=days)
     logger.info(
         "Microsoft 365 fetch for source id=%d: processed=%d reports_found=%d",
         int(source_id),

backend/app/main.py

@@ -193,7 +193,7 @@ def _poll_single_m365_source(source: MailSource) -> None:
         )
 
         started_at = datetime.utcnow()
-        results = client.fetch_reports()
+        results = client.fetch_reports(days=7)
         if src:
             if results.get("new_ingested_ids"):
                 all_ids = list(dict.fromkeys(already + results["new_ingested_ids"]))
@@ -756,7 +756,7 @@ def _trigger_poll_gmail_source(source: MailSource, db) -> dict:
     }
 
 
-def _trigger_poll_m365_source(source: MailSource, db) -> dict:
+def _trigger_poll_m365_source(source: MailSource, db, days: int = 7) -> dict:
     """Poll a single M365_GRAPH source and return a result dict for the API response."""
     global last_check_time  # pylint: disable=global-statement
 
@@ -774,7 +774,7 @@ def _trigger_poll_m365_source(source: MailSource, db) -> dict:
         db=db,
     )
     started_at = datetime.utcnow()
-    results = graph_client.fetch_reports()
+    results = graph_client.fetch_reports(days=days)
     last_check_time = datetime.now()
 
     if results.get("new_ingested_ids"):
@@ -832,7 +832,7 @@ def _poll_source_for_trigger(source: MailSource, db, days: int = 7) -> dict:  #
                 "reason": "Microsoft 365 account not yet authorised",
             }
         try:
-            return _trigger_poll_m365_source(source, db)
+            return _trigger_poll_m365_source(source, db, days=days)
         except Exception as e:  # pylint: disable=broad-exception-caught
             logger.error("Error polling Microsoft 365 source id=%d: %s", source.id, str(e))
             return {
@@ -881,7 +881,7 @@ def _poll_enabled_sources_for_trigger(days: int) -> list[dict]:
 @app.post("/api/v1/admin/trigger-poll")
 async def trigger_imap_poll(
     auth: dict = Depends(require_admin_auth),
-    days: int = Query(7, ge=1, le=365, title="Number of days to fetch for IMAP sources"),
+    days: int = Query(7, ge=1, le=365, title="Number of days to fetch for mail sources"),
 ):
     """
     Manually trigger IMAP polling for all enabled mail sources (admin only).

backend/app/services/microsoft_graph_client.py

@@ -3,7 +3,9 @@
 import base64
 import json
 import logging
-from typing import Any, Dict, List, Optional
+import time
+from datetime import datetime, timedelta
+from typing import Any, Callable, Dict, List, Optional
 from urllib.parse import quote, urlencode
 
 import httpx
@@ -25,6 +27,9 @@ M365_SCOPES = [
 
 _PAGE_SIZE = 100
 _MAX_FOLDER_DEPTH = 5
+_MAX_GRAPH_RETRIES = 3
+_MAX_RETRY_DELAY_SECONDS = 30
+_RETRYABLE_STATUS_CODES = {429, 503, 504}
 _DMARC_SUBJECT_TERMS = (
     "dmarc",
     "aggregate report",
@@ -65,6 +70,7 @@ class MicrosoftGraphClient:
         folder_id: Optional[str] = None,
         already_ingested_ids: Optional[List[str]] = None,
         db: Any = None,
+        sleep: Optional[Callable[[float], None]] = None,
     ):
         self.tenant_id = tenant_id or "common"
         self.client_id = client_id
@@ -77,6 +83,7 @@ class MicrosoftGraphClient:
         self.already_ingested_ids: List[str] = list(already_ingested_ids or [])
         self.report_store = ReportStore.get_instance()
         self.db = db
+        self._sleep = sleep or time.sleep
         self._refreshed_tokens: Optional[Dict[str, str]] = None
 
     def get_refreshed_tokens(self) -> Optional[Dict[str, str]]:
@@ -227,8 +234,9 @@ class MicrosoftGraphClient:
             url = data.get("@odata.nextLink")
             params = None
 
-    def fetch_reports(self) -> Dict[str, Any]:
+    def fetch_reports(self, days: int = 7) -> Dict[str, Any]:
         """Fetch and ingest DMARC report attachments from Microsoft Graph."""
+        safe_days = max(1, min(int(days or 7), 365))
         stats: Dict[str, Any] = {
             "success": True,
             "processed": 0,
@@ -242,10 +250,11 @@ class MicrosoftGraphClient:
             "details": [],
             "target_mailbox": self._target_mailbox_label(),
             "target_folder": self._target_folder_label(),
+            "search_window_days": safe_days,
         }
 
         try:
-            messages = self._list_dmarc_messages()
+            messages = self._list_dmarc_messages(days=safe_days)
         except Exception as exc:  # pylint: disable=broad-exception-caught
             logger.error("Microsoft Graph: failed to list messages: %s", exc)
             return {**stats, "success": False, "error": str(exc), "errors": [str(exc)]}
@@ -324,14 +333,42 @@ class MicrosoftGraphClient:
         params: Optional[Dict[str, Any]] = None,
     ) -> Dict[str, Any]:
         url = path_or_url if path_or_url.startswith("http") else f"{GRAPH_BASE_URL}{path_or_url}"
-        resp = httpx.request(method, url, headers=self._headers(), params=params, timeout=30)
+        resp: Optional[httpx.Response] = None
-        if resp.status_code == 401 and self.refresh_token:
+
-            self._refresh_access_token()
+        for attempt in range(_MAX_GRAPH_RETRIES + 1):
             resp = httpx.request(method, url, headers=self._headers(), params=params, timeout=30)
+            if resp.status_code == 401 and self.refresh_token:
+                self._refresh_access_token()
+                resp = httpx.request(method, url, headers=self._headers(), params=params, timeout=30)
+            if (
+                resp.status_code in _RETRYABLE_STATUS_CODES
+                and attempt < _MAX_GRAPH_RETRIES
+            ):
+                delay = self._retry_delay_seconds(resp, attempt)
+                logger.warning(
+                    "Microsoft Graph request throttled/unavailable; retrying in %.1fs",
+                    delay,
+                )
+                self._sleep(delay)
+                continue
+            break
+
+        if resp is None:
+            raise MicrosoftGraphError("Microsoft Graph request failed before receiving a response.")
         if resp.status_code < 200 or resp.status_code >= 300:
             raise MicrosoftGraphError(self._format_error(resp))
         return resp.json() if resp.content else {}
 
+    @staticmethod
+    def _retry_delay_seconds(resp: httpx.Response, attempt: int) -> float:
+        retry_after = resp.headers.get("Retry-After")
+        if retry_after:
+            try:
+                return min(float(retry_after), _MAX_RETRY_DELAY_SECONDS)
+            except ValueError:
+                pass
+        return min(float(2**attempt), _MAX_RETRY_DELAY_SECONDS)
+
     def _refresh_access_token(self) -> None:
         data = {
             "client_id": self.client_id,
@@ -392,13 +429,15 @@ class MicrosoftGraphClient:
             or lower.endswith(".gzip")
         )
 
-    def _list_dmarc_messages(self) -> List[Dict[str, Any]]:
+    def _list_dmarc_messages(self, days: int) -> List[Dict[str, Any]]:
         messages: List[Dict[str, Any]] = []
         url = self._messages_path()
+        cutoff = datetime.utcnow() - timedelta(days=days)
         params: Optional[Dict[str, Any]] = {
             "$top": _PAGE_SIZE,
             "$select": "id,subject,from,hasAttachments,receivedDateTime",
             "$orderby": "receivedDateTime desc",
+            "$filter": f"receivedDateTime ge {cutoff.strftime('%Y-%m-%dT%H:%M:%SZ')}",
         }
 
         while url:

backend/app/tests/test_mail_sources.py

@@ -1455,6 +1455,7 @@ class TestMicrosoft365GraphMailSource:
             "new_domains": ["example.com"],
             "errors": [],
             "new_ingested_ids": ["id1", "id2", "id3"],
+            "search_window_days": 30,
         }
         mock_client.get_refreshed_tokens.return_value = None
 
@@ -1472,7 +1473,7 @@ class TestMicrosoft365GraphMailSource:
                 return_value='["id1", "id2", "id3"]',
             ),
         ):
-            resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch")
+            resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch?days=30")
 
         assert resp.status_code == 200
         data = resp.json()
@@ -1483,6 +1484,8 @@ class TestMicrosoft365GraphMailSource:
         assert data["new_domains"] == ["example.com"]
         assert data["target_mailbox"] is None
         assert data["target_folder"] is None
+        assert data["search_window_days"] == 30
+        mock_client.fetch_reports.assert_called_once_with(days=30)
 
     def test_m365_specific_fetch_persists_refreshed_tokens(
         self, authed_client: TestClient, db_session: Session
@@ -2667,12 +2670,13 @@ class TestTriggerPollM365Source:
             patch("app.main.MicrosoftGraphClient.load_ingested_ids", return_value=[]),
             patch("app.main.MicrosoftGraphClient.dump_ingested_ids", return_value='["id1"]'),
         ):
-            result = _trigger_poll_m365_source(src, mock_db)
+            result = _trigger_poll_m365_source(src, mock_db, days=30)
 
         assert result["success"] is True
         assert result["source_id"] == 8
         assert result["processed"] == 2
         assert src.m365_ingested_ids == '["id1"]'
+        mock_gc.fetch_reports.assert_called_once_with(days=30)
         mock_db.commit.assert_called_once()
 
     def test_persists_refreshed_tokens(self):
@@ -2776,10 +2780,10 @@ class TestPollSourceForTrigger:
 
         expected = {"source_id": 8, "name": "M365", "success": True}
         with patch("app.main._trigger_poll_m365_source", return_value=expected) as mock_fn:
-            result = _poll_source_for_trigger(src, MagicMock())
+            result = _poll_source_for_trigger(src, MagicMock(), days=30)
 
         assert result is expected
-        mock_fn.assert_called_once()
+        assert mock_fn.call_args.kwargs["days"] == 30
 
     def test_m365_exception_returns_failure_dict(self):
         from app.main import _poll_source_for_trigger

backend/app/tests/test_main_polling.py

@@ -133,6 +133,7 @@ def test_poll_single_m365_source_persists_import_state():
     assert db_source.m365_ingested_ids == '["message-1"]'
     assert db_source.m365_access_token == "new-tok"
     assert db_source.m365_refresh_token == "new-ref"
+    mock_client.fetch_reports.assert_called_once_with(days=7)
     db.commit.assert_called_once()
     db.close.assert_called_once()
 

backend/app/tests/test_microsoft_graph_client.py

@@ -192,6 +192,8 @@ class TestMicrosoftGraphFetchReports:
     def test_fetch_reports_uses_selected_folder_id_and_records_context(self, monkeypatch):
         def fake_request(method, url, headers=None, params=None, timeout=None):
             if url.endswith("/users/shared%40example.com/mailFolders/folder-id/messages"):
+                assert "$filter" in params
+                assert params["$filter"].startswith("receivedDateTime ge ")
                 return httpx.Response(
                     200,
                     json={
@@ -226,6 +228,55 @@ class TestMicrosoftGraphFetchReports:
         assert result["processed"] == 1
         assert result["target_mailbox"] == "shared@example.com"
         assert result["target_folder"] == "DMARC Reports"
+        assert result["search_window_days"] == 7
+
+    def test_fetch_reports_applies_requested_search_window(self, monkeypatch):
+        def fake_request(method, url, headers=None, params=None, timeout=None):
+            assert url.endswith("/me/mailFolders/inbox/messages")
+            assert params["$top"] == 100
+            assert params["$select"] == "id,subject,from,hasAttachments,receivedDateTime"
+            assert params["$orderby"] == "receivedDateTime desc"
+            assert params["$filter"].startswith("receivedDateTime ge ")
+            return httpx.Response(200, json={"value": []})
+
+        monkeypatch.setattr("app.services.microsoft_graph_client.httpx.request", fake_request)
+        client = _make_client()
+
+        result = client.fetch_reports(days=30)
+
+        assert result["success"] is True
+        assert result["search_window_days"] == 30
+        assert result["processed"] == 0
+
+    def test_request_retries_graph_throttling(self, monkeypatch):
+        responses = [
+            httpx.Response(
+                429,
+                headers={"Retry-After": "0"},
+                json={"error": {"code": "TooManyRequests", "message": "slow down"}},
+            ),
+            httpx.Response(200, json={"value": []}),
+        ]
+        sleeps = []
+
+        def fake_request(method, url, headers=None, params=None, timeout=None):
+            return responses.pop(0)
+
+        monkeypatch.setattr("app.services.microsoft_graph_client.httpx.request", fake_request)
+        client = MicrosoftGraphClient(
+            tenant_id="organizations",
+            client_id="client-id",
+            client_secret="client-secret",
+            access_token="access-token",
+            refresh_token="refresh-token",
+            sleep=sleeps.append,
+        )
+
+        result = client.fetch_reports()
+
+        assert result["success"] is True
+        assert sleeps == [0.0]
+        assert responses == []
 
     def test_fetch_reports_imports_zip_attachment(self, monkeypatch, db_session):
         attachment_bytes = _zip_xml()

docs/milestones.md

@@ -206,7 +206,7 @@ Planned:
 - Microsoft 365 mail source using OAuth (Graph) with least-privilege scopes. Delivered for delegated `User.Read`, `Mail.Read`, and `offline_access` with encrypted token storage, manual import, scheduled polling, UI setup, and operator docs.
 - Shared mailbox and folder selection support for DMARC report collection. Delivered with shared mailbox targeting, Microsoft Graph folder listing, folder-id based imports, UI selection, and mailbox/folder context in import history.
 - Import-history parity with existing sources (auditable attachment outcomes, duplicates, parse failures). Delivered for Microsoft 365 imports.
-- Backfill support with safe throttling and progressive search windows.
+- Backfill support with safe throttling and progressive search windows. Delivered with days-based Graph `receivedDateTime` filters, duplicate-safe reruns, and retry/backoff for throttled or temporarily unavailable Graph requests.
 - Secret handling mirrors existing guidance (no raw secrets in logs; 1Password-friendly).
 
 Exit criteria:

docs/user_guide/microsoft365.md

@@ -39,10 +39,12 @@ DMARQ uses delegated Microsoft Graph access. That means the authorised Microsoft
 
 ## Import Behavior
 
-DMARQ reads recent messages in the configured folder, filters for messages that look like DMARC reports, downloads Graph `fileAttachment` items, and sends `.xml`, `.zip`, `.gz`, and `.gzip` attachments through the same parser and persistence path used by upload, IMAP, and Gmail imports.
+DMARQ reads messages in the configured search window, filters for messages that look like DMARC reports, downloads Graph `fileAttachment` items, and sends `.xml`, `.zip`, `.gz`, and `.gzip` attachments through the same parser and persistence path used by upload, IMAP, and Gmail imports.
 
 Imported Graph message IDs are stored on the mail source so scheduled polling does not reprocess the same message. Import history records processed messages, imported reports, duplicates, parse failures, attachment-level details, and the mailbox/folder target used for the attempt.
 
+Manual imports and backfills use the **days** value from the Mail Sources UI or trigger-poll endpoint. DMARQ passes that window to Microsoft Graph as a `receivedDateTime` filter, so large mailboxes can be searched without scanning unrelated older mail. If Graph returns a throttling or temporary service response, DMARQ retries with `Retry-After` or exponential backoff before surfacing the sanitized error in import history.
+
 ## Troubleshooting
 
 - **Not authorised**: reconnect the source from Mail Sources.