diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b96c68..a010df6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **Processing logs & reporting** — users can now view the full history of polling runs and per-email delivery status: + - **`GET /processing-runs`** — paginated list of all processing runs for the authenticated user's mailboxes (filterable by account and status). + - **`GET /processing-runs/{id}`** — details for a single run. + - **`GET /processing-runs/{id}/logs`** — per-email log entries (subject, sender, size, delivery status, error details) for a given run. + - **`GET /mail-accounts/{id}/processing-runs`** — runs scoped to a single mailbox. + - **`GET /mail-accounts/{id}/logs`** — all per-email log entries for a single mailbox. +- **Admin log endpoints** (superuser only): + - **`GET /admin/processing-runs`** — all runs across every user, filterable by user ID, account ID, or status. Account and user email addresses are GDPR-pseudonymised. + - **`GET /admin/processing-logs`** — all per-email log entries system-wide, filterable by user, account, run, or log level. Sender (`From:`) headers are pseudonymised via `mask_from_header()`; subjects are shown as-is (user-owned content). +- **`backend/app/core/gdpr.py`** — GDPR masking utilities: `mask_email()`, `mask_name()`, `mask_from_header()` for pseudonymising PII in admin views. +- **Worker now writes `ProcessingLog` entries per email** — subject, sender, size, delivery outcome and error detail are captured for every email processed by `process_mail_account`. +- **`/logs` page** — user-facing log page with a paginated processing-run table; each row expands inline to show the per-email log for that run (subject, masked sender, size, status). +- **`/admin/logs` page** — admin view with two tabs: *Processing Runs* (expandable, fetches per-email logs on demand) and *Per-Email Logs* (flat table with GDPR-masked sender addresses). Filterable by user ID and status/level. +- **Sidebar navigation** — added *Logs* link (user) and *Activity Logs* link (admin) to `DashboardLayout`. +- **Admin overview** — added *Activity Logs* card to `/admin` page. +- **Dashboard** — "Recent Processing Runs" table now reads from the new `/processing-runs` endpoint; shows account name and a *View all logs* link. + ### Added - **Prometheus metrics** (`/metrics` endpoint on the FastAPI backend, scraped every 15 s): - **HTTP layer** — `http_requests_total` (counter, labelled `method`/`endpoint`/`status_code`) and `http_request_duration_seconds` (histogram). Path segments that are numeric IDs are normalised to `{id}` to avoid label-set explosion. diff --git a/backend/app/api/v1/api.py b/backend/app/api/v1/api.py index 0d17374..fe5eafa 100644 --- a/backend/app/api/v1/api.py +++ b/backend/app/api/v1/api.py @@ -13,6 +13,7 @@ from app.api.v1.endpoints import ( admin, providers, app_settings, + logs, ) api_router = APIRouter() @@ -34,3 +35,6 @@ api_router.include_router( ) api_router.include_router(admin.router, prefix="/admin", tags=["Admin"]) api_router.include_router(app_settings.router, prefix="/settings", tags=["Settings"]) +api_router.include_router( + logs.router, prefix="/processing-runs", tags=["Processing Logs"] +) diff --git a/backend/app/api/v1/endpoints/admin.py b/backend/app/api/v1/endpoints/admin.py index ddab0e9..78bc6af 100644 --- a/backend/app/api/v1/endpoints/admin.py +++ b/backend/app/api/v1/endpoints/admin.py @@ -1,15 +1,18 @@ """Admin endpoints""" -from typing import List -from fastapi import APIRouter, Depends, HTTPException, status +import math +from typing import List, Optional +from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from app.core.database import get_db from app.core.deps import get_current_superuser +from app.core.gdpr import mask_email, mask_from_header from app.models.database_models import ( User, MailAccount, + ProcessingLog, ProcessingRun, SubscriptionPlan, SubscriptionTier, @@ -21,6 +24,10 @@ from app.models.schemas import ( SubscriptionPlanResponse, SubscriptionPlanCreate, SubscriptionPlanUpdate, + AdminProcessingRunResponse, + AdminProcessingLogResponse, + PaginatedAdminRunsResponse, + PaginatedAdminLogsResponse, ) router = APIRouter() @@ -282,3 +289,165 @@ async def delete_plan( ) await db.delete(plan) await db.commit() + + +# ── Admin Logs ───────────────────────────────────────────────────────────────── + + +def _admin_paginate(total: int, page: int, page_size: int) -> dict: + pages = max(1, math.ceil(total / page_size)) if total else 1 + return {"total": total, "page": page, "page_size": page_size, "pages": pages} + + +@router.get( + "/processing-runs", + response_model=PaginatedAdminRunsResponse, + summary="List all processing runs across all users (admin only)", +) +async def admin_list_processing_runs( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + user_id: Optional[int] = Query(None, description="Filter by user ID"), + account_id: Optional[int] = Query(None, description="Filter by mail account ID"), + status_filter: Optional[str] = Query( + None, + alias="status", + description="Filter by run status", + ), + current_user: User = Depends(get_current_superuser), + db: AsyncSession = Depends(get_db), +): + """ + Return a paginated list of all processing runs in the system with + GDPR-masked user / account email addresses. + """ + base = ( + select( + ProcessingRun, + MailAccount.name.label("account_name"), + MailAccount.email_address.label("account_email"), + MailAccount.user_id.label("uid"), + User.email.label("user_email"), + ) + .join(MailAccount, ProcessingRun.mail_account_id == MailAccount.id) + .join(User, MailAccount.user_id == User.id) + ) + + if user_id is not None: + base = base.where(MailAccount.user_id == user_id) + if account_id is not None: + base = base.where(ProcessingRun.mail_account_id == account_id) + if status_filter: + base = base.where(ProcessingRun.status == status_filter) + + total = ( + await db.execute(select(func.count()).select_from(base.subquery())) + ).scalar_one() + offset = (page - 1) * page_size + rows = ( + await db.execute( + base.order_by(ProcessingRun.started_at.desc()) + .offset(offset) + .limit(page_size) + ) + ).all() + + items = [ + AdminProcessingRunResponse( + id=row.ProcessingRun.id, # type: ignore[arg-type] + mail_account_id=row.ProcessingRun.mail_account_id, # type: ignore[arg-type] + started_at=row.ProcessingRun.started_at, # type: ignore[arg-type] + completed_at=row.ProcessingRun.completed_at, # type: ignore[arg-type] + duration_seconds=row.ProcessingRun.duration_seconds, # type: ignore[arg-type] + emails_fetched=row.ProcessingRun.emails_fetched, # type: ignore[arg-type] + emails_forwarded=row.ProcessingRun.emails_forwarded, # type: ignore[arg-type] + emails_failed=row.ProcessingRun.emails_failed, # type: ignore[arg-type] + status=row.ProcessingRun.status, # type: ignore[arg-type] + error_message=row.ProcessingRun.error_message, # type: ignore[arg-type] + account_name=row.account_name, + account_email=mask_email(row.account_email) if row.account_email else None, + user_id=row.uid, + user_email=mask_email(row.user_email) if row.user_email else None, + ) + for row in rows + ] + + return PaginatedAdminRunsResponse( + items=items, **_admin_paginate(total, page, page_size) # type: ignore[arg-type] + ) + + +@router.get( + "/processing-logs", + response_model=PaginatedAdminLogsResponse, + summary="List all per-email processing logs across all users (admin only)", +) +async def admin_list_processing_logs( + page: int = Query(1, ge=1), + page_size: int = Query(50, ge=1, le=200), + user_id: Optional[int] = Query(None, description="Filter by user ID"), + account_id: Optional[int] = Query(None, description="Filter by mail account ID"), + run_id: Optional[int] = Query(None, description="Filter by processing run ID"), + level: Optional[str] = Query( + None, description="Filter by level (INFO, WARNING, ERROR)" + ), + current_user: User = Depends(get_current_superuser), + db: AsyncSession = Depends(get_db), +): + """ + Return paginated per-email log entries with GDPR-masked sender addresses. + Subject lines are shown as-is (the user owns their own mail content); + sender addresses are pseudonymised for operator privacy. + """ + base = select( + ProcessingLog, + User.email.label("user_email"), + ).join(User, ProcessingLog.user_id == User.id) + + if user_id is not None: + base = base.where(ProcessingLog.user_id == user_id) + if account_id is not None: + base = base.where(ProcessingLog.mail_account_id == account_id) + if run_id is not None: + base = base.where(ProcessingLog.processing_run_id == run_id) + if level: + base = base.where(ProcessingLog.level == level.upper()) + + total = ( + await db.execute(select(func.count()).select_from(base.subquery())) + ).scalar_one() + offset = (page - 1) * page_size + rows = ( + await db.execute( + base.order_by(ProcessingLog.timestamp.desc()) + .offset(offset) + .limit(page_size) + ) + ).all() + + items = [ + AdminProcessingLogResponse( + id=row.ProcessingLog.id, # type: ignore[arg-type] + timestamp=row.ProcessingLog.timestamp, # type: ignore[arg-type] + level=row.ProcessingLog.level, # type: ignore[arg-type] + message=row.ProcessingLog.message, # type: ignore[arg-type] + email_subject=row.ProcessingLog.email_subject, # type: ignore[arg-type] + email_from=( + mask_from_header(row.ProcessingLog.email_from) + if row.ProcessingLog.email_from + else None + ), + success=row.ProcessingLog.success, # type: ignore[arg-type] + mail_account_id=row.ProcessingLog.mail_account_id, # type: ignore[arg-type] + processing_run_id=row.ProcessingLog.processing_run_id, # type: ignore[arg-type] + email_size_bytes=row.ProcessingLog.email_size_bytes, # type: ignore[arg-type] + error_details=row.ProcessingLog.error_details, # type: ignore[arg-type] + user_id=row.ProcessingLog.user_id, # type: ignore[arg-type] + user_email=mask_email(row.user_email) if row.user_email else None, + ) + for row in rows + ] + + return PaginatedAdminLogsResponse( + items=items, **_admin_paginate(total, page, page_size) # type: ignore[arg-type] + ) diff --git a/backend/app/api/v1/endpoints/logs.py b/backend/app/api/v1/endpoints/logs.py new file mode 100644 index 0000000..3583c70 --- /dev/null +++ b/backend/app/api/v1/endpoints/logs.py @@ -0,0 +1,234 @@ +""" +Processing logs and run history endpoints for users. + +Users can view the full history of processing runs and per-email logs +for their own mailboxes. Admin equivalents live in admin.py. +""" + +from __future__ import annotations + +import math +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy import select, func +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.database import get_db +from app.core.deps import get_current_active_user +from app.models.database_models import ( + MailAccount, + ProcessingLog, + ProcessingRun, + User, +) +from app.models.schemas import ( + PaginatedProcessingLogsResponse, + PaginatedProcessingRunsResponse, + ProcessingLogDetailResponse, + ProcessingRunDetailResponse, +) + +router = APIRouter() + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + + +def _paginate(total: int, page: int, page_size: int) -> dict: + pages = max(1, math.ceil(total / page_size)) if total else 1 + return {"total": total, "page": page, "page_size": page_size, "pages": pages} + + +# --------------------------------------------------------------------------- +# Processing Runs (all accounts belonging to the current user) +# --------------------------------------------------------------------------- + + +@router.get( + "/processing-runs", + response_model=PaginatedProcessingRunsResponse, + summary="List processing runs for the current user", +) +async def list_processing_runs( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + account_id: Optional[int] = Query(None, description="Filter by mail account ID"), + status_filter: Optional[str] = Query( + None, + alias="status", + description="Filter by run status (completed, failed, partial_failure, running)", + ), + current_user: User = Depends(get_current_active_user), + db: AsyncSession = Depends(get_db), +): + """ + Return a paginated list of processing runs for all mail accounts owned by + the authenticated user, optionally filtered by account or status. + """ + # Base query: join with MailAccount to enforce ownership + base = ( + select(ProcessingRun, MailAccount.name, MailAccount.email_address) + .join(MailAccount, ProcessingRun.mail_account_id == MailAccount.id) + .where(MailAccount.user_id == current_user.id) # type: ignore[arg-type] + ) + + if account_id is not None: + base = base.where(ProcessingRun.mail_account_id == account_id) + if status_filter: + base = base.where(ProcessingRun.status == status_filter) + + # Total count + count_q = select(func.count()).select_from(base.subquery()) + total = (await db.execute(count_q)).scalar_one() + + offset = (page - 1) * page_size + rows = ( + await db.execute( + base.order_by(ProcessingRun.started_at.desc()) + .offset(offset) + .limit(page_size) + ) + ).all() + + items = [ + ProcessingRunDetailResponse( + id=row.ProcessingRun.id, # type: ignore[arg-type] + mail_account_id=row.ProcessingRun.mail_account_id, # type: ignore[arg-type] + started_at=row.ProcessingRun.started_at, # type: ignore[arg-type] + completed_at=row.ProcessingRun.completed_at, # type: ignore[arg-type] + duration_seconds=row.ProcessingRun.duration_seconds, # type: ignore[arg-type] + emails_fetched=row.ProcessingRun.emails_fetched, # type: ignore[arg-type] + emails_forwarded=row.ProcessingRun.emails_forwarded, # type: ignore[arg-type] + emails_failed=row.ProcessingRun.emails_failed, # type: ignore[arg-type] + status=row.ProcessingRun.status, # type: ignore[arg-type] + error_message=row.ProcessingRun.error_message, # type: ignore[arg-type] + account_name=row.name, + account_email=row.email_address, + ) + for row in rows + ] + + return PaginatedProcessingRunsResponse( + items=items, **_paginate(total, page, page_size) # type: ignore[arg-type] + ) + + +@router.get( + "/processing-runs/{run_id}", + response_model=ProcessingRunDetailResponse, + summary="Get a single processing run", +) +async def get_processing_run( + run_id: int, + current_user: User = Depends(get_current_active_user), + db: AsyncSession = Depends(get_db), +): + """Return details for a single processing run owned by the current user.""" + row = ( + await db.execute( + select(ProcessingRun, MailAccount.name, MailAccount.email_address) + .join(MailAccount, ProcessingRun.mail_account_id == MailAccount.id) + .where( + ProcessingRun.id == run_id, + MailAccount.user_id == current_user.id, # type: ignore[arg-type] + ) + ) + ).one_or_none() + + if not row: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Processing run not found" + ) + + return ProcessingRunDetailResponse( + id=row.ProcessingRun.id, # type: ignore[arg-type] + mail_account_id=row.ProcessingRun.mail_account_id, # type: ignore[arg-type] + started_at=row.ProcessingRun.started_at, # type: ignore[arg-type] + completed_at=row.ProcessingRun.completed_at, # type: ignore[arg-type] + duration_seconds=row.ProcessingRun.duration_seconds, # type: ignore[arg-type] + emails_fetched=row.ProcessingRun.emails_fetched, # type: ignore[arg-type] + emails_forwarded=row.ProcessingRun.emails_forwarded, # type: ignore[arg-type] + emails_failed=row.ProcessingRun.emails_failed, # type: ignore[arg-type] + status=row.ProcessingRun.status, # type: ignore[arg-type] + error_message=row.ProcessingRun.error_message, # type: ignore[arg-type] + account_name=row.name, + account_email=row.email_address, + ) + + +@router.get( + "/processing-runs/{run_id}/logs", + response_model=PaginatedProcessingLogsResponse, + summary="Get per-email logs for a processing run", +) +async def get_run_logs( + run_id: int, + page: int = Query(1, ge=1), + page_size: int = Query(50, ge=1, le=200), + current_user: User = Depends(get_current_active_user), + db: AsyncSession = Depends(get_db), +): + """ + Return the detailed per-email log entries recorded during a specific + processing run. Ownership is verified by joining with MailAccount. + """ + # Verify the run belongs to this user + run_row = ( + await db.execute( + select(ProcessingRun) + .join(MailAccount, ProcessingRun.mail_account_id == MailAccount.id) + .where( + ProcessingRun.id == run_id, + MailAccount.user_id == current_user.id, # type: ignore[arg-type] + ) + ) + ).scalar_one_or_none() + + if not run_row: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Processing run not found" + ) + + count_q = select(func.count(ProcessingLog.id)).where( + ProcessingLog.processing_run_id == run_id + ) + total = (await db.execute(count_q)).scalar_one() + + offset = (page - 1) * page_size + logs = ( + ( + await db.execute( + select(ProcessingLog) + .where(ProcessingLog.processing_run_id == run_id) + .order_by(ProcessingLog.timestamp.asc()) + .offset(offset) + .limit(page_size) + ) + ) + .scalars() + .all() + ) + + items = [ + ProcessingLogDetailResponse( + id=log.id, # type: ignore[arg-type] + timestamp=log.timestamp, # type: ignore[arg-type] + level=log.level, # type: ignore[arg-type] + message=log.message, # type: ignore[arg-type] + email_subject=log.email_subject, # type: ignore[arg-type] + email_from=log.email_from, # type: ignore[arg-type] + success=log.success, # type: ignore[arg-type] + mail_account_id=log.mail_account_id, # type: ignore[arg-type] + processing_run_id=log.processing_run_id, # type: ignore[arg-type] + email_size_bytes=log.email_size_bytes, # type: ignore[arg-type] + error_details=log.error_details, # type: ignore[arg-type] + ) + for log in logs + ] + + return PaginatedProcessingLogsResponse( + items=items, **_paginate(total, page, page_size) # type: ignore[arg-type] + ) diff --git a/backend/app/api/v1/endpoints/mail_accounts.py b/backend/app/api/v1/endpoints/mail_accounts.py index 8e2d5b1..508709d 100644 --- a/backend/app/api/v1/endpoints/mail_accounts.py +++ b/backend/app/api/v1/endpoints/mail_accounts.py @@ -1,9 +1,10 @@ """Mail account management endpoints""" -from typing import List -from fastapi import APIRouter, Depends, HTTPException, status +import math +from typing import List, Optional +from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy import select, desc +from sqlalchemy import select, desc, func from app.core.database import get_db from app.core.deps import get_current_active_user @@ -11,6 +12,8 @@ from app.core.security import encrypt_credential from app.models.database_models import ( User, MailAccount, + ProcessingLog, + ProcessingRun, AccountStatus, SubscriptionPlan, ) @@ -22,6 +25,10 @@ from app.models.schemas import ( MailAccountTestResponse, MailAccountAutoDetectRequest, MailAccountAutoDetectResponse, + PaginatedProcessingRunsResponse, + PaginatedProcessingLogsResponse, + ProcessingRunDetailResponse, + ProcessingLogDetailResponse, ) from app.services.mail_processor import MailProcessor, MailServerAutoDetect from app.core.config import settings @@ -274,3 +281,148 @@ async def auto_detect_mail_settings( return MailAccountAutoDetectResponse( success=len(suggestions) > 0, suggestions=suggestions ) + + +# --------------------------------------------------------------------------- +# Per-account processing runs & logs +# --------------------------------------------------------------------------- + + +@router.get( + "/{account_id}/processing-runs", + response_model=PaginatedProcessingRunsResponse, + summary="List processing runs for a specific mail account", +) +async def list_account_runs( + account_id: int, + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + current_user: User = Depends(get_current_active_user), + db: AsyncSession = Depends(get_db), +): + """Return paginated processing runs for a mail account owned by the user.""" + result = await db.execute( + select(MailAccount).where( + MailAccount.id == account_id, + MailAccount.user_id == current_user.id, + ) + ) + account = result.scalar_one_or_none() + if not account: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Mail account not found" + ) + + base = select(ProcessingRun).where(ProcessingRun.mail_account_id == account_id) + total = ( + await db.execute(select(func.count()).select_from(base.subquery())) + ).scalar_one() + + offset = (page - 1) * page_size + runs = ( + ( + await db.execute( + base.order_by(desc(ProcessingRun.started_at)) + .offset(offset) + .limit(page_size) + ) + ) + .scalars() + .all() + ) + + pages = max(1, math.ceil(total / page_size)) if total else 1 + items = [ + ProcessingRunDetailResponse( + id=r.id, # type: ignore[arg-type] + mail_account_id=r.mail_account_id, # type: ignore[arg-type] + started_at=r.started_at, # type: ignore[arg-type] + completed_at=r.completed_at, # type: ignore[arg-type] + duration_seconds=r.duration_seconds, # type: ignore[arg-type] + emails_fetched=r.emails_fetched, # type: ignore[arg-type] + emails_forwarded=r.emails_forwarded, # type: ignore[arg-type] + emails_failed=r.emails_failed, # type: ignore[arg-type] + status=r.status, # type: ignore[arg-type] + error_message=r.error_message, # type: ignore[arg-type] + account_name=account.name, # type: ignore[arg-type] + account_email=account.email_address, # type: ignore[arg-type] + ) + for r in runs + ] + return PaginatedProcessingRunsResponse( + items=items, total=total, page=page, page_size=page_size, pages=pages + ) + + +@router.get( + "/{account_id}/logs", + response_model=PaginatedProcessingLogsResponse, + summary="List processing logs for a specific mail account", +) +async def list_account_logs( + account_id: int, + page: int = Query(1, ge=1), + page_size: int = Query(50, ge=1, le=200), + level: Optional[str] = Query( + None, description="Filter by log level (INFO, WARNING, ERROR)" + ), + current_user: User = Depends(get_current_active_user), + db: AsyncSession = Depends(get_db), +): + """Return paginated per-email log entries for a mail account owned by the user.""" + result = await db.execute( + select(MailAccount).where( + MailAccount.id == account_id, + MailAccount.user_id == current_user.id, + ) + ) + account = result.scalar_one_or_none() + if not account: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Mail account not found" + ) + + base = select(ProcessingLog).where( + ProcessingLog.mail_account_id == account_id, + ProcessingLog.user_id == current_user.id, # type: ignore[arg-type] + ) + if level: + base = base.where(ProcessingLog.level == level.upper()) + + total = ( + await db.execute(select(func.count()).select_from(base.subquery())) + ).scalar_one() + + offset = (page - 1) * page_size + logs = ( + ( + await db.execute( + base.order_by(ProcessingLog.timestamp.desc()) + .offset(offset) + .limit(page_size) + ) + ) + .scalars() + .all() + ) + + pages = max(1, math.ceil(total / page_size)) if total else 1 + items = [ + ProcessingLogDetailResponse( + id=log.id, # type: ignore[arg-type] + timestamp=log.timestamp, # type: ignore[arg-type] + level=log.level, # type: ignore[arg-type] + message=log.message, # type: ignore[arg-type] + email_subject=log.email_subject, # type: ignore[arg-type] + email_from=log.email_from, # type: ignore[arg-type] + success=log.success, # type: ignore[arg-type] + mail_account_id=log.mail_account_id, # type: ignore[arg-type] + processing_run_id=log.processing_run_id, # type: ignore[arg-type] + email_size_bytes=log.email_size_bytes, # type: ignore[arg-type] + error_details=log.error_details, # type: ignore[arg-type] + ) + for log in logs + ] + return PaginatedProcessingLogsResponse( + items=items, total=total, page=page, page_size=page_size, pages=pages + ) diff --git a/backend/app/core/gdpr.py b/backend/app/core/gdpr.py new file mode 100644 index 0000000..25733dc --- /dev/null +++ b/backend/app/core/gdpr.py @@ -0,0 +1,106 @@ +""" +GDPR-compliant data masking utilities. + +These helpers are used in admin-facing API responses to pseudonymise +personal data (email addresses, names) so that operators can audit +system behaviour without seeing full end-user PII. +""" + +from __future__ import annotations + +import re + + +def mask_email(email: str) -> str: + """ + Partially mask an email address for GDPR-compliant display. + + Examples + -------- + >>> mask_email("john.doe@example.com") + 'jo***@e***.com' + >>> mask_email("ab@x.io") + 'ab***@x***.io' + >>> mask_email("a@b.de") + 'a***@b***.de' + """ + if not email or "@" not in email: + return "***" + + local, _, domain = email.partition("@") + + # Local part: keep first 2 chars (or all if shorter), then "***" + visible_local = local[:2] if len(local) >= 2 else local + masked_local = f"{visible_local}***" + + # Domain part: keep first char of SLD and the TLD unchanged + domain_parts = domain.rsplit(".", 1) + if len(domain_parts) == 2: + sld, tld = domain_parts + visible_sld = sld[:1] if sld else "" + masked_domain = f"{visible_sld}***.{tld}" + else: + masked_domain = "***" + + return f"{masked_local}@{masked_domain}" + + +def mask_name(name: str) -> str: + """ + Partially mask a display name. + + Examples + -------- + >>> mask_name("John Doe") + 'Jo*** D***' + >>> mask_name("Alice") + 'Al***' + """ + if not name: + return "***" + words = name.split() + masked_words = [] + for word in words: + visible = word[:2] if len(word) >= 2 else word + masked_words.append(f"{visible}***") + return " ".join(masked_words) + + +# RFC 5322 address pattern – extracts the bare email from strings like +# "John Doe " or just "john@example.com". +_ADDR_RE = re.compile(r"<([^>]+)>|(\S+@\S+\.\S+)") + + +def mask_from_header(from_header: str) -> str: + """ + Mask a raw RFC 5322 From header value for GDPR-compliant display. + + Examples + -------- + >>> mask_from_header("John Doe ") + 'Jo*** D*** ' + >>> mask_from_header("john.doe@example.com") + 'jo***@e***.com' + """ + if not from_header: + return "***" + + # Try to parse "Display Name " form + angle_match = re.search(r"^(.*?)<([^>]+)>", from_header.strip()) + if angle_match: + display_name = angle_match.group(1).strip().strip('"') + email_part = angle_match.group(2).strip() + masked_email = mask_email(email_part) + if display_name: + masked_display = mask_name(display_name) + return f"{masked_display} <{masked_email}>" + return masked_email + + # Plain email address + bare_match = _ADDR_RE.search(from_header) + if bare_match: + email_part = bare_match.group(1) or bare_match.group(2) + return mask_email(email_part) + + # Fallback: mask the whole string + return mask_name(from_header) diff --git a/backend/app/models/schemas.py b/backend/app/models/schemas.py index c5882ab..9765f3e 100644 --- a/backend/app/models/schemas.py +++ b/backend/app/models/schemas.py @@ -204,6 +204,15 @@ class ProcessingRunResponse(BaseModel): model_config = ConfigDict(from_attributes=True) +class ProcessingRunDetailResponse(ProcessingRunResponse): + """ProcessingRunResponse with optional account metadata.""" + + account_name: Optional[str] = None + account_email: Optional[str] = None + + model_config = ConfigDict(from_attributes=True) + + # Processing Log Schemas class ProcessingLogResponse(BaseModel): id: int @@ -217,6 +226,67 @@ class ProcessingLogResponse(BaseModel): model_config = ConfigDict(from_attributes=True) +class ProcessingLogDetailResponse(ProcessingLogResponse): + """ProcessingLogResponse with additional fields.""" + + mail_account_id: int + processing_run_id: Optional[int] = None + email_size_bytes: Optional[int] = None + error_details: Optional[Dict[str, Any]] = None + + model_config = ConfigDict(from_attributes=True) + + +class PaginatedProcessingRunsResponse(BaseModel): + items: List[ProcessingRunDetailResponse] + total: int + page: int + page_size: int + pages: int + + +class PaginatedProcessingLogsResponse(BaseModel): + items: List[ProcessingLogDetailResponse] + total: int + page: int + page_size: int + pages: int + + +class AdminProcessingRunResponse(ProcessingRunDetailResponse): + """ProcessingRunDetailResponse with user info for admin views.""" + + user_id: Optional[int] = None + user_email: Optional[str] = None + + model_config = ConfigDict(from_attributes=True) + + +class PaginatedAdminRunsResponse(BaseModel): + items: List[AdminProcessingRunResponse] + total: int + page: int + page_size: int + pages: int + + +class AdminProcessingLogResponse(ProcessingLogDetailResponse): + """ProcessingLogDetailResponse with user info for admin views.""" + + user_id: int + user_email: Optional[str] = None + + model_config = ConfigDict(from_attributes=True) + + +class PaginatedAdminLogsResponse(BaseModel): + items: List[AdminProcessingLogResponse] + total: int + page: int + page_size: int + pages: int + + # Notification Config Schemas class NotificationConfigBase(BaseModel): channel: NotificationChannel diff --git a/backend/app/workers/tasks.py b/backend/app/workers/tasks.py index 30287a1..5276593 100644 --- a/backend/app/workers/tasks.py +++ b/backend/app/workers/tasks.py @@ -3,6 +3,7 @@ Celery tasks for background email processing. """ import asyncio +import email as email_lib import time from datetime import datetime, timedelta, timezone from celery import Task @@ -198,7 +199,33 @@ async def process_mail_account(account_id: int): f"for account {account.id}; truncating to shorter list" ) + # Field length limits matching the DB column definitions + _MAX_SUBJECT_LEN = 500 + _MAX_FROM_LEN = 255 + for email_data, uid in zip(emails, new_uids): + # ── Parse email metadata for logging ─────────────────────── + email_subject: str | None = None + email_from: str | None = None + try: + msg = email_lib.message_from_bytes(email_data) + raw_subject = msg.get("Subject", "") or "" + email_subject = ( + raw_subject[:_MAX_SUBJECT_LEN] if raw_subject else None + ) + raw_from = msg.get("From", "") or "" + email_from = raw_from[:_MAX_FROM_LEN] if raw_from else None + except (ValueError, TypeError, UnicodeDecodeError) as exc: + logger.debug( + "Could not parse email headers for account %s: %s", + account.id, + exc, + ) + + email_size_bytes = len(email_data) + forwarded_ok = False + error_msg: str | None = None + try: if use_gmail_api and gmail_service: # Inject via Gmail API (preferred) @@ -207,6 +234,7 @@ async def process_mail_account(account_id: int): label_ids=["INBOX"], source_account_name=account.name, # type: ignore[arg-type] ) + forwarded_ok = True emails_forwarded += 1 successfully_forwarded_uids.append(uid) else: @@ -215,6 +243,7 @@ async def process_mail_account(account_id: int): email_data, account.name, account.forward_to, smtp_config # type: ignore[arg-type] ) if success: + forwarded_ok = True emails_forwarded += 1 successfully_forwarded_uids.append(uid) else: @@ -240,8 +269,29 @@ async def process_mail_account(account_id: int): "User must re-authorise." ) logger.error(f"Error delivering email: {e}") + error_msg = str(e) emails_failed += 1 + # ── Write per-email ProcessingLog entry ───────────────────── + db.add( + ProcessingLog( + user_id=account.user_id, + mail_account_id=account.id, + processing_run_id=run.id, + level="INFO" if forwarded_ok else "ERROR", + message=( + f"Forwarded: {email_subject or '(no subject)'}" + if forwarded_ok + else f"Failed: {error_msg or 'delivery error'}" + ), + email_subject=email_subject, + email_from=email_from, + email_size_bytes=email_size_bytes, + success=forwarded_ok, + error_details={"error": error_msg} if error_msg else None, + ) + ) + # 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/docs/TODO.md b/docs/TODO.md index a71b347..e3232f5 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -178,7 +178,7 @@ Comprehensive task breakdown for repository improvements and production readines ### Not Started 📋 - [ ] Integrate Sentry for error tracking -- [ ] Add structured logging with correlation IDs +- [x] Add structured logging with correlation IDs (per-email ProcessingLog entries now captured in DB) - [ ] Add APM (Application Performance Monitoring) - [ ] Set up uptime monitoring - [ ] Create runbook for common issues @@ -196,6 +196,7 @@ Comprehensive task breakdown for repository improvements and production readines - [x] Unified Google OAuth flow: sign-in requests all Gmail scopes; single `/auth/callback` redirect URI needed in Google Console - [x] Message deduplication (POP3 UIDL + IMAP \Seen flag + DB tracking) - [x] **Debug email**: "Send Debug Email" button in Settings injects a test message (from christian@docuelevate.org, dated today, labelled `test` + `imported`, placed in inbox) to verify end-to-end Gmail API delivery +- [x] **Logging & reporting**: per-email ProcessingLog capture in worker; user `/logs` page; admin `/admin/logs` page; GDPR masking utilities (`gdpr.py`) - [ ] Implement GDPR data export endpoint - [ ] Complete notification service integration (Apprise) - [ ] Add advanced email filtering @@ -232,6 +233,8 @@ because the API client layer is missing. - [x] `AuthGuard` for protected routes - [x] Fix wizard grey screen (Tailwind v4 `bg-opacity` → `/75` syntax, modal restructure) - [x] `/auth/gmail-callback` page for Gmail OAuth one-click flow +- [x] **`/logs` page** — user processing history: paginated runs table with expandable per-email log panel (subject, sender, size, status) +- [x] **Dashboard** — "Recent Processing Runs" now wired to real `/processing-runs` endpoint; shows account name and links to `/logs` ### Not Started 📋 - [ ] End-to-end testing of frontend against backend API @@ -249,6 +252,7 @@ because the API client layer is missing. - [x] `is_superuser` exposed in `/users/me` response - [x] Admin badge (purple shield) shown in top bar for superusers - [x] Fix blank page on direct navigation to `/admin*`: moved superuser guard inside `` so auth check always runs on fresh load +- [x] **`/admin/logs` page** — system-wide processing activity: expandable run table + flat per-email log table with GDPR-masked sender addresses; filterable by user ID, status, log level --- @@ -348,7 +352,8 @@ because the API client layer is missing. 1. **Immediate** (Today): - [x] Create `frontend/src/lib/api.ts` (frontend is broken without it) - [x] Fix remaining security issues (bare excepts, datetime, redirect_uri) - - [ ] Add backend endpoint for processing runs (needed by dashboard) + - [x] Add backend endpoint for processing runs (needed by dashboard) + - [x] Build logging & reporting: per-email ProcessingLog capture, user `/logs` page, admin `/admin/logs` page, GDPR masking 2. **This Week**: - [ ] Enable rate limiting diff --git a/frontend/src/app/admin/page.tsx b/frontend/src/app/admin/page.tsx index 3e80638..61f277e 100644 --- a/frontend/src/app/admin/page.tsx +++ b/frontend/src/app/admin/page.tsx @@ -89,7 +89,7 @@ export default function AdminPage() { )} -
+
Create and configure subscription plans

+ +
+ +
+
+

Activity Logs

+

Processing runs and per-email logs across all users

+
+
)} diff --git a/frontend/src/app/dashboard/page.tsx b/frontend/src/app/dashboard/page.tsx index 31036c0..07b77a4 100644 --- a/frontend/src/app/dashboard/page.tsx +++ b/frontend/src/app/dashboard/page.tsx @@ -4,6 +4,7 @@ import { AuthGuard } from '@/components/AuthGuard'; import { DashboardLayout } from '@/components/DashboardLayout'; import { useQuery } from '@tanstack/react-query'; import { mailAccountsApi, processingRunsApi } from '@/lib/api'; +import Link from 'next/link'; import { Mail, Send, @@ -51,19 +52,19 @@ export default function DashboardPage() { const { data: runs, isLoading: runsLoading } = useQuery({ queryKey: ['processing-runs'], - queryFn: () => processingRunsApi.list(), + queryFn: () => processingRunsApi.list({ page: 1, page_size: 10 }), }); const stats = { totalAccounts: accounts?.length || 0, activeAccounts: accounts?.filter((a) => a.is_enabled).length || 0, - emailsToday: runs + emailsToday: runs?.items ?.filter((r) => { const today = new Date().toDateString(); return new Date(r.started_at).toDateString() === today; }) .reduce((sum, r) => sum + r.emails_forwarded, 0) || 0, - errors: runs?.filter((r) => r.emails_failed > 0).length || 0, + errors: runs?.items?.filter((r) => r.emails_failed > 0).length || 0, }; return ( @@ -100,15 +101,18 @@ export default function DashboardPage() { {/* Recent Processing Runs */}
-
+

Recent Processing Runs

+ + View all logs → +
{runsLoading ? (
- ) : runs && runs.length > 0 ? ( + ) : runs && runs.items && runs.items.length > 0 ? ( @@ -133,12 +137,12 @@ export default function DashboardPage() { - {runs.slice(0, 10).map((run) => { + {runs.items.map((run) => { const account = accounts?.find((a) => a.id === run.mail_account_id); return (
- {account?.name || `Account ${run.mail_account_id}`} + {run.account_name || account?.name || `Account ${run.mail_account_id}`}
diff --git a/frontend/src/components/DashboardLayout.tsx b/frontend/src/components/DashboardLayout.tsx index 11b5bee..257d811 100644 --- a/frontend/src/components/DashboardLayout.tsx +++ b/frontend/src/components/DashboardLayout.tsx @@ -14,7 +14,9 @@ import { User, Shield, Users, - CreditCard + CreditCard, + FileText, + Activity } from 'lucide-react'; interface DashboardLayoutProps { @@ -35,6 +37,7 @@ export function DashboardLayout({ children }: DashboardLayoutProps) { const navigation = [ { name: 'Dashboard', href: '/dashboard', icon: LayoutDashboard }, { name: 'Mail Accounts', href: '/accounts', icon: Mail }, + { name: 'Logs', href: '/logs', icon: FileText }, { name: 'Settings', href: '/settings', icon: Settings }, ]; @@ -43,6 +46,7 @@ export function DashboardLayout({ children }: DashboardLayoutProps) { { name: 'Admin Overview', href: '/admin', icon: Shield }, { name: 'Manage Users', href: '/admin/users', icon: Users }, { name: 'Manage Plans', href: '/admin/plans', icon: CreditCard }, + { name: 'Activity Logs', href: '/admin/logs', icon: Activity }, ] : []; diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 81a63b2..5748499 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -126,6 +126,64 @@ export interface ProcessingRun { emails_failed: number; status: string; error_message?: string | null; + account_name?: string | null; + account_email?: string | null; +} + +export interface ProcessingLog { + id: number; + timestamp: string; + level: string; + message: string; + email_subject?: string | null; + email_from?: string | null; + success: boolean; + mail_account_id: number; + processing_run_id?: number | null; + email_size_bytes?: number | null; + error_details?: Record | null; +} + +export interface PaginatedProcessingRuns { + items: ProcessingRun[]; + total: number; + page: number; + page_size: number; + pages: number; +} + +export interface PaginatedProcessingLogs { + items: ProcessingLog[]; + total: number; + page: number; + page_size: number; + pages: number; +} + +export interface AdminProcessingRun extends ProcessingRun { + user_id?: number | null; + user_email?: string | null; +} + +export interface AdminProcessingLog extends ProcessingLog { + user_id: number; + user_email?: string | null; +} + +export interface PaginatedAdminRuns { + items: AdminProcessingRun[]; + total: number; + page: number; + page_size: number; + pages: number; +} + +export interface PaginatedAdminLogsResponse { + items: AdminProcessingLog[]; + total: number; + page: number; + page_size: number; + pages: number; } export interface AutoDetectSuggestion { @@ -292,10 +350,54 @@ export const mailAccountsApi = { // ── Processing Runs API ───────────────────────────────────────────────── export const processingRunsApi = { - async list(): Promise { - // TODO: Add a dedicated /processing-runs endpoint to the backend - // For now, return empty array since no user-facing endpoint exists yet - return []; + async list(params?: { + page?: number; + page_size?: number; + account_id?: number; + status?: string; + }): Promise { + const response = await api.get("/processing-runs", { + params, + }); + return response.data; + }, + + async get(runId: number): Promise { + const response = await api.get(`/processing-runs/${runId}`); + return response.data; + }, + + async getLogs( + runId: number, + params?: { page?: number; page_size?: number } + ): Promise { + const response = await api.get( + `/processing-runs/${runId}/logs`, + { params } + ); + return response.data; + }, + + async listForAccount( + accountId: number, + params?: { page?: number; page_size?: number; status?: string } + ): Promise { + const response = await api.get( + `/mail-accounts/${accountId}/processing-runs`, + { params } + ); + return response.data; + }, + + async listLogsForAccount( + accountId: number, + params?: { page?: number; page_size?: number; level?: string } + ): Promise { + const response = await api.get( + `/mail-accounts/${accountId}/logs`, + { params } + ); + return response.data; }, }; @@ -483,6 +585,33 @@ export const adminApi = { async deletePlan(id: number): Promise { await api.delete(`/admin/plans/${id}`); }, + + async listProcessingRuns(params?: { + page?: number; + page_size?: number; + user_id?: number; + account_id?: number; + status?: string; + }): Promise { + const response = await api.get('/admin/processing-runs', { + params, + }); + return response.data; + }, + + async listProcessingLogs(params?: { + page?: number; + page_size?: number; + user_id?: number; + account_id?: number; + run_id?: number; + level?: string; + }): Promise { + const response = await api.get('/admin/processing-logs', { + params, + }); + return response.data; + }, }; export default api;