feat: add per-source mail import trigger

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

@@ -202,6 +202,88 @@ def _import_to_response(row: MailSourceImport) -> MailSourceImportResponse:
     )
 
 
+def _fetch_response(source: MailSource, results: Dict[str, Any]) -> Dict[str, Any]:
+    """Build the common response payload for a manual source fetch."""
+    return {
+        "source_id": source.id,
+        "name": source.name,
+        "success": bool(results.get("success", False)),
+        "processed": int(results.get("processed", 0)),
+        "reports_found": int(results.get("reports_found", 0)),
+        "duplicate_reports": int(results.get("duplicate_reports", 0)),
+        "new_domains": [str(d) for d in results.get("new_domains", [])],
+        "error_count": len(results.get("errors", [])),
+        "timestamp": datetime.now().isoformat(),
+    }
+
+
+def _fetch_gmail_source(source: MailSource, db: Session) -> Dict[str, Any]:
+    """Run one Gmail API import and persist source/import metadata."""
+    if not source.gmail_access_token:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Gmail account not yet authorised. Complete OAuth2 flow first.",
+        )
+
+    already = GmailClient.load_ingested_ids(source.gmail_ingested_ids)
+    client = GmailClient(
+        client_id=source.gmail_client_id or "",
+        client_secret=source.gmail_client_secret or "",
+        access_token=source.gmail_access_token,
+        refresh_token=source.gmail_refresh_token or "",
+        already_ingested_ids=already,
+        db=db,
+    )
+
+    started_at = datetime.utcnow()
+    results = client.fetch_reports()
+
+    if results.get("new_ingested_ids"):
+        all_ids = list(dict.fromkeys(already + results["new_ingested_ids"]))
+        source.gmail_ingested_ids = GmailClient.dump_ingested_ids(all_ids)
+
+    refreshed = client.get_refreshed_tokens()
+    if refreshed:
+        source.gmail_access_token = refreshed["access_token"]
+        if "refresh_token" in refreshed:
+            source.gmail_refresh_token = refreshed["refresh_token"]
+
+    source.last_checked = datetime.utcnow()
+    record_import_attempt(db, source, results, started_at=started_at, trigger="manual")
+    db.commit()
+    return results
+
+
+def _fetch_imap_source(source: MailSource, db: Session, days: int) -> Dict[str, Any]:
+    """Run one IMAP import and persist source/import metadata."""
+    client = IMAPClient(
+        server=source.server,
+        port=source.port or 993,
+        username=source.username,
+        password=source.password,
+        delete_emails=False,
+        db=db,
+    )
+    started_at = datetime.utcnow()
+    results = client.fetch_reports(days=days)
+    source.last_checked = datetime.utcnow()
+    record_import_attempt(db, source, results, started_at=started_at, trigger="manual")
+    db.commit()
+    return results
+
+
+def _fetch_source(source: MailSource, db: Session, days: int) -> Dict[str, Any]:
+    """Dispatch a manual fetch for one configured mail source."""
+    if source.method == "GMAIL_API":
+        return _fetch_gmail_source(source, db)
+    if source.method == "IMAP":
+        return _fetch_imap_source(source, db, days)
+    raise HTTPException(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        detail=f"Manual fetch is not available for method '{source.method}'.",
+    )
+
+
 # ---------------------------------------------------------------------------
 # Routes
 # ---------------------------------------------------------------------------
@@ -278,6 +360,36 @@ async def list_mail_source_imports(
     return [_import_to_response(row) for row in rows]
 
 
+@router.post("/{source_id}/fetch", response_model=Dict[str, Any])
+async def fetch_mail_source(
+    source_id: int,
+    days: int = 7,
+    db: Session = Depends(get_db),
+    _auth: dict = Depends(require_admin_auth),
+) -> Dict[str, Any]:
+    """Manually fetch DMARC reports for one configured mail source."""
+    if days < 1 or days > 365:
+        raise HTTPException(status_code=400, detail="Days parameter must be between 1 and 365")
+
+    source = _get_source_or_404(source_id, db)
+    results = _fetch_source(source, db, days)
+    logger.info(
+        "Manual fetch for source id=%d: processed=%d reports_found=%d duplicates=%d",
+        int(source_id),
+        int(results.get("processed", 0)),
+        int(results.get("reports_found", 0)),
+        int(results.get("duplicate_reports", 0)),
+    )
+    for err in results.get("errors", []):
+        logger.warning(
+            "Manual fetch warning for source id=%d: %s",
+            int(source_id),
+            _sanitize_for_log(err),
+        )
+
+    return _fetch_response(source, results)
+
+
 @router.put("/{source_id}", response_model=MailSourceResponse)
 async def update_mail_source(
     source_id: int,

backend/app/templates/mail_sources.html

@@ -105,6 +105,15 @@
                                                     <polyline points="22 4 12 14.01 9 11.01"></polyline>
                                                 </svg>
                                             </button>
+                                            <button class="btn btn-ghost btn-xs" x-on:click="fetchSource(source)" title="Run import now"
+                                                :disabled="Boolean(fetching[source.id])">
+                                                <svg xmlns="http://www.w3.org/2000/svg" width="14" height="14" viewBox="0 0 24 24"
+                                                    fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round"
+                                                    stroke-linejoin="round">
+                                                    <path d="M21 12a9 9 0 1 1-9-9c2.52 0 4.8 1.04 6.43 2.72"></path>
+                                                    <path d="M21 3v6h-6"></path>
+                                                </svg>
+                                            </button>
                                             <button class="btn btn-ghost btn-xs" x-on:click="loadImportHistory(source)" title="Import history">
                                                 <svg xmlns="http://www.w3.org/2000/svg" width="14" height="14" viewBox="0 0 24 24"
                                                     fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round"
@@ -471,6 +480,7 @@ function mailSourcesApp() {
         isTesting: false,
         isSaving: false,
         testing: {},
+        fetching: {},
         historySource: null,
         importHistory: [],
         historyLoading: false,
@@ -509,6 +519,33 @@ function mailSourcesApp() {
             }
         },
 
+        async fetchSource(source) {
+            this.fetching[source.id] = true;
+            this.feedback = { message: '', type: '' };
+            try {
+                const resp = await fetch(`/api/v1/mail-sources/${source.id}/fetch`, {
+                    method: 'POST',
+                });
+                const result = await resp.json();
+                if (!resp.ok) {
+                    throw new Error(result.detail || 'Import failed');
+                }
+                this.feedback = {
+                    message: `Import finished for ${source.name}: ${result.reports_found} report${result.reports_found === 1 ? '' : 's'}, ${result.duplicate_reports || 0} duplicate${result.duplicate_reports === 1 ? '' : 's'}.`,
+                    type: result.success ? 'success' : 'error',
+                };
+                await this.loadSources();
+                if (this.historySource && this.historySource.id === source.id) {
+                    const updated = this.sources.find(s => s.id === source.id) || source;
+                    await this.loadImportHistory(updated);
+                }
+            } catch (e) {
+                this.feedback = { message: `Import error: ${e.message}`, type: 'error' };
+            } finally {
+                this.fetching[source.id] = false;
+            }
+        },
+
         async loadImportHistory(source) {
             this.historySource = source;
             this.importHistory = [];

backend/app/tests/test_mail_sources.py

@@ -282,6 +282,31 @@ class TestMailSourcesAPIAuthed:
         assert data[0]["new_domains"] == ["example.com"]
         assert data[0]["errors"] == ["sanitized error"]
 
+    def test_list_import_history_handles_malformed_json(
+        self, authed_client: TestClient, db_session: Session
+    ):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources", json={"name": "Bad History JSON", "method": "IMAP"}
+        )
+        source_id = create_resp.json()["id"]
+
+        db_session.add(
+            MailSourceImport(
+                mail_source_id=source_id,
+                trigger="manual",
+                status="warning",
+                new_domains="not-json",
+                errors='{"not": "a list"}',
+            )
+        )
+        db_session.commit()
+
+        resp = authed_client.get(f"/api/v1/mail-sources/{source_id}/imports")
+
+        assert resp.status_code == 200
+        assert resp.json()[0]["new_domains"] == []
+        assert resp.json()[0]["errors"] == []
+
     def test_list_import_history_unknown_source_returns_404(self, authed_client: TestClient):
         resp = authed_client.get("/api/v1/mail-sources/99999/imports")
         assert resp.status_code == 404
@@ -716,6 +741,178 @@ class TestGmailAPIMailSource:
         assert data["new_domains"] == ["example.com"]
 
 
+class TestManualSourceFetchEndpoint:
+    """Tests for POST /api/v1/mail-sources/{source_id}/fetch."""
+
+    def test_fetch_imap_source(self, authed_client: TestClient, caplog):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources",
+            json={
+                "name": "Fetch IMAP",
+                "method": "IMAP",
+                "server": "imap.example.com",
+                "username": "user@example.com",
+                "password": "secret",
+            },
+        )
+        source_id = create_resp.json()["id"]
+        mock_imap = MagicMock()
+        mock_imap.fetch_reports.return_value = {
+            "success": True,
+            "processed": 5,
+            "reports_found": 3,
+            "duplicate_reports": 1,
+            "new_domains": ["example.com"],
+            "errors": ["bad attachment\nwith newline"],
+        }
+
+        with (
+            caplog.at_level("WARNING"),
+            patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_imap),
+        ):
+            resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch?days=30")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is True
+        assert data["processed"] == 5
+        assert data["reports_found"] == 3
+        assert data["duplicate_reports"] == 1
+        assert data["error_count"] == 1
+        assert "bad attachment with newline" in caplog.text
+        mock_imap.fetch_reports.assert_called_once_with(days=30)
+
+    def test_fetch_gmail_source(self, authed_client: TestClient, db_session: Session):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources",
+            json={
+                "name": "Fetch Gmail Generic",
+                "method": "GMAIL_API",
+                "gmail_client_id": "cid",
+                "gmail_client_secret": "csec",
+            },
+        )
+        source_id = create_resp.json()["id"]
+        source = db_session.get(MailSource, source_id)
+        source.gmail_access_token = "tok"
+        source.gmail_refresh_token = "refresh"
+        db_session.commit()
+
+        mock_gmail = MagicMock()
+        mock_gmail.fetch_reports.return_value = {
+            "success": True,
+            "processed": 2,
+            "reports_found": 1,
+            "duplicate_reports": 0,
+            "new_domains": [],
+            "errors": [],
+            "new_ingested_ids": ["id1"],
+        }
+        mock_gmail.get_refreshed_tokens.return_value = None
+
+        with (
+            patch("app.api.api_v1.endpoints.mail_sources.GmailClient", return_value=mock_gmail),
+            patch(
+                "app.api.api_v1.endpoints.mail_sources.GmailClient.load_ingested_ids",
+                return_value=[],
+            ),
+            patch(
+                "app.api.api_v1.endpoints.mail_sources.GmailClient.dump_ingested_ids",
+                return_value='["id1"]',
+            ),
+        ):
+            resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is True
+        assert data["processed"] == 2
+        assert data["reports_found"] == 1
+        assert data["source_id"] == source_id
+
+    def test_fetch_gmail_source_rejects_missing_token(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources",
+            json={"name": "Fetch Gmail No Token", "method": "GMAIL_API"},
+        )
+        source_id = create_resp.json()["id"]
+
+        resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch")
+
+        assert resp.status_code == 400
+        assert "OAuth2" in resp.json()["detail"]
+
+    def test_fetch_gmail_source_persists_refreshed_tokens(
+        self, authed_client: TestClient, db_session: Session
+    ):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources",
+            json={
+                "name": "Fetch Gmail Refresh",
+                "method": "GMAIL_API",
+                "gmail_client_id": "cid",
+                "gmail_client_secret": "csec",
+            },
+        )
+        source_id = create_resp.json()["id"]
+        source = db_session.get(MailSource, source_id)
+        source.gmail_access_token = "old-access"
+        source.gmail_refresh_token = "old-refresh"
+        db_session.commit()
+
+        mock_gmail = MagicMock()
+        mock_gmail.fetch_reports.return_value = {
+            "success": True,
+            "processed": 1,
+            "reports_found": 1,
+            "duplicate_reports": 0,
+            "new_domains": [],
+            "errors": [],
+            "new_ingested_ids": [],
+        }
+        mock_gmail.get_refreshed_tokens.return_value = {
+            "access_token": "new-access",
+            "refresh_token": "new-refresh",
+        }
+
+        with (
+            patch("app.api.api_v1.endpoints.mail_sources.GmailClient", return_value=mock_gmail),
+            patch(
+                "app.api.api_v1.endpoints.mail_sources.GmailClient.load_ingested_ids",
+                return_value=[],
+            ),
+        ):
+            resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch")
+
+        db_session.refresh(source)
+        assert resp.status_code == 200
+        assert source.gmail_access_token == "new-access"
+        assert source.gmail_refresh_token == "new-refresh"
+
+    def test_fetch_rejects_invalid_days(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources",
+            json={"name": "Bad Days", "method": "IMAP"},
+        )
+        source_id = create_resp.json()["id"]
+
+        resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch?days=0")
+
+        assert resp.status_code == 400
+
+    def test_fetch_rejects_unsupported_source_method(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources",
+            json={"name": "Fetch POP3", "method": "POP3"},
+        )
+        source_id = create_resp.json()["id"]
+
+        resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch")
+
+        assert resp.status_code == 400
+        assert "not available" in resp.json()["detail"]
+
+
 # ---------------------------------------------------------------------------
 # GmailClient unit tests
 # ---------------------------------------------------------------------------

docs/development/roadmap.md

@@ -27,6 +27,7 @@ Recently improved:
 - Parsed upload, Gmail, and IMAP reports are now persisted to `dmarc_reports` and `report_records`.
 - Report/domain API reads can hydrate the dashboard projection from persisted data after restart.
 - Mail source import history is visible in the Mail Sources UI.
+- Individual mail sources can be manually imported from the Mail Sources UI.
 - The current Alpine-based UI is allowed by CSP and renders dynamic tables in real browsers.
 
 Implementation note:

docs/milestones.md

@@ -62,6 +62,7 @@ Recently delivered:
 - Mail source imports now persist sanitized import-history records for manual and scheduled polls.
 - Uploaded, Gmail-imported, and IMAP-imported reports are now persisted to report/record tables and can be reloaded into report/domain views.
 - Mail source import history is now visible from the Mail Sources UI.
+- A single mail source can be manually imported from the Mail Sources UI, with the result recorded in import history.
 - The current Alpine-based UI can run under the configured CSP, so dynamic tables render in real browsers.
 
 Next tasks:

docs/todo.md

@@ -91,6 +91,7 @@ This file tracks the specific implementation tasks for each milestone of the DMA
 - [x] Persist sanitized import errors for API/UI review
 - [x] Count duplicate skips separately from parse failures
 - [x] Show recent import history in the Mail Sources UI
+- [x] Add manual import trigger per mail source
 - [ ] Add retry/backfill controls per mail source
 
 ## Milestone 3: Database Integration