Merge pull request #105 from Climate-Vision/merge/pr-62

feat: alert delivery worker with email/webhook support (#62, rebased)

Author: obielin

.env.example

@@ -51,3 +51,16 @@ CLIMATEVISION_LLM_MODEL=claude-haiku-4-5-20251001
 # SQLite is used by default. Use an absolute path or a project-relative path.
 # For production with high traffic, switch to PostgreSQL and update db.py.
 DATABASE_URL=sqlite:///outputs/climatevision.sqlite3
+
+# ---------------------------------------------------------------------------
+# SMTP / Alert Delivery (for the alert delivery worker)
+# ---------------------------------------------------------------------------
+SMTP_HOST=smtp.example.com
+SMTP_PORT=587
+SMTP_USERNAME=your-email@example.com
+SMTP_PASSWORD=your-email-password
+SMTP_USE_TLS=true
+SMTP_FROM_EMAIL=alerts@climatevision.org
+
+ALERT_DELIVERY_MAX_RETRIES=3
+ALERT_DELIVERY_POLL_INTERVAL_SECONDS=60

src/climatevision/api/main.py

@@ -19,6 +19,8 @@
 from pathlib import Path
 from typing import Any, Optional, Literal
 
+from contextlib import asynccontextmanager
+
 from pydantic import field_validator
 
 from fastapi import FastAPI, File, Form, HTTPException, UploadFile, Header, Query, Depends, Request
@@ -40,6 +42,7 @@
     get_subscriptions_for_organization,
     create_organization_alert,
     get_alerts_for_organization,
+    get_pending_alerts,
     acknowledge_alert,
     mark_alert_delivered,
 )
@@ -48,6 +51,7 @@
 from climatevision.api.auth import require_api_key
 from climatevision.governance import explain_prediction, SHAPExplainer
 from climatevision.security.api_security import SecurityMiddleware
+from climatevision.workers.alert_delivery import AlertDeliveryWorker
 
 logger = logging.getLogger(__name__)
 
@@ -387,6 +391,18 @@ async def dispatch(self, request: Request, call_next: Any) -> Response:
 def create_app() -> FastAPI:
     init_db()
 
+    # Set up alert delivery worker
+    alert_worker = AlertDeliveryWorker()
+
+    @asynccontextmanager
+    async def lifespan(_app: FastAPI):
+        """Start background workers on app startup, stop on shutdown."""
+        await alert_worker.start()
+        logger.info("Alert delivery worker started")
+        yield
+        await alert_worker.stop()
+        logger.info("Alert delivery worker stopped")
+
     app = FastAPI(
         title="ClimateVision API",
         version="0.2.0",
@@ -402,6 +418,7 @@ def create_app() -> FastAPI:
         docs_url="/docs",
         redoc_url="/redoc",
         openapi_url="/openapi.json",
+        lifespan=lifespan,
     )
 
     app.add_middleware(AuditLogMiddleware)
@@ -1146,6 +1163,39 @@ def mark_alert_as_delivered(
             raise HTTPException(status_code=404, detail="Alert not found")
         return {"success": True, "alert_id": alert_id}
 
+    @app.get("/api/organizations/{org_id}/alerts/pending")
+    def list_pending_alerts(
+        org_id: int,
+        limit: int = Query(default=50, le=200),
+    ) -> list[AlertResponse]:
+        """List pending (undelivered) alerts for an organization.
+
+        This endpoint is designed for monitoring and dashboard use,
+        showing alerts that the delivery worker has not yet processed.
+        """
+        org = get_organization(org_id)
+        if not org:
+            raise HTTPException(status_code=404, detail="Organization not found")
+
+        alerts = get_pending_alerts(max_attempts=3, limit=limit)
+        # Filter to this specific organization's alerts
+        org_alerts = [a for a in alerts if a["organization_id"] == org_id]
+
+        return [
+            AlertResponse(
+                id=alert["id"],
+                organization_id=alert["organization_id"],
+                alert_type=alert["alert_type"],
+                severity=alert["severity"],
+                title=alert["title"],
+                message=alert["message"],
+                delivered=bool(alert["delivered"]),
+                acknowledged=bool(alert["acknowledged"]),
+                created_at=alert["created_at"],
+            )
+            for alert in org_alerts[:limit]
+        ]
+
     # ===== Static Files =====
 
     # Serve built frontend when dist exists (production mode)

src/climatevision/db.py

@@ -502,7 +502,7 @@ def acknowledge_alert(alert_id: int, acknowledged_by: Optional[str] = None) -> b
 def mark_alert_delivered(alert_id: int) -> bool:
     """Mark an alert as delivered."""
     now = _utc_now_iso()
-    
+
     with get_connection() as conn:
         cursor = conn.execute(
             """
@@ -513,3 +513,55 @@ def mark_alert_delivered(alert_id: int) -> bool:
             (now, alert_id),
         )
         return cursor.rowcount > 0
+
+
+def get_pending_alerts(
+    max_attempts: int = 3,
+    limit: int = 100,
+) -> list:
+    """Get all alerts that have not been delivered and are still within retry limit.
+
+    Returns alerts ordered by created_at ascending (oldest first).
+    """
+    import sqlite3
+    with get_connection() as conn:
+        return conn.execute(
+            """
+            SELECT * FROM organization_alerts
+            WHERE delivered = 0 AND delivery_attempts < ?
+            ORDER BY created_at ASC
+            LIMIT ?
+            """,
+            (max_attempts, limit),
+        ).fetchall()
+
+
+def increment_delivery_attempt(alert_id: int) -> bool:
+    """Increment the delivery attempt counter without marking delivered."""
+    with get_connection() as conn:
+        cursor = conn.execute(
+            """
+            UPDATE organization_alerts
+            SET delivery_attempts = delivery_attempts + 1
+            WHERE id = ? AND delivered = 0
+            """,
+            (alert_id,),
+        )
+        return cursor.rowcount > 0
+
+
+def mark_alert_failed(alert_id: int) -> bool:
+    """Mark an alert as permanently failed after exhausting retries."""
+    now = _utc_now_iso()
+    with get_connection() as conn:
+        cursor = conn.execute(
+            """
+            UPDATE organization_alerts
+            SET delivery_attempts = delivery_attempts + 1,
+                details = COALESCE(details || ' | ', '') ||
+                    'FAILED: all delivery attempts exhausted at ' || ?
+            WHERE id = ? AND delivered = 0
+            """,
+            (now, alert_id),
+        )
+        return cursor.rowcount > 0

src/climatevision/workers/__init__.py

@@ -0,0 +1,12 @@
+"""
+Background workers for ClimateVision.
+
+Workers handle async operations that should not block the API:
+- Alert delivery (email + webhook)
+- Report generation
+- Scheduled analysis runs
+"""
+
+from climatevision.workers.alert_delivery import AlertDeliveryWorker
+
+__all__ = ["AlertDeliveryWorker"]