feat(CG5a+CG5b): NotificationService + POST /notify/webhook + POST /notify/email

main.py

@@ -6,6 +6,8 @@ ML-1: telemetri, state store, DAG.
 CG1+CG2: billing endpoints + IAP-beskyttelse.
 CG3: static-mappe serveres fra /static/*.
 AUTH: /auth/login + /auth/callback for klient OAuth2 onboarding.
+CG5a: POST /notify/webhook — push til Slack/Teams/Chat/Discord/etc.
+CG5b: POST /notify/email  — SendGrid digest til kunde-e-post.
 """
 
 import os
@@ -19,8 +21,9 @@ from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel, Field
-from typing import List
+from typing import List, Optional, Dict, Any
 import datetime
+import httpx
 from cachetools import cached, TTLCache
 
 from agent import run, authorize_mode
@@ -66,46 +69,33 @@ async def require_iap(request: Request, call_next):
 # ── FIREBASE & FIRESTORE INIT ───────────────────────────────────────────────
 db = None
 try:
-    # When running in Google Cloud, ADC will be used automatically.
-    # For local dev, `gcloud auth application-default login` is required.
     firebase_admin.initialize_app()
     db = firestore.Client()
     print("Firestore client initialized successfully.")
 except Exception as e:
-    # The app can still run in a limited mode without Firestore.
-    # Endpoints that depend on `db` will return a 500 error.
     print(f"WARNING: Firestore client failed to initialize: {e}", file=sys.stderr)
 
 # ── AUTH & SESSION ────────────────────────────────────────────────────────────
-# Add session middleware for storing auth state
 app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*")
 app.add_middleware(SessionMiddleware, secret_key=os.environ.get("SESSION_SECRET"))
 
-# Configure Authlib's OAuth client
 oauth = OAuth()
 oauth.register(
     name='google',
     client_id=os.environ.get("GOOGLE_CLIENT_ID"),
     client_secret=os.environ.get("GOOGLE_CLIENT_SECRET"),
     server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
-    client_kwargs={
+    client_kwargs={'scope': 'openid email profile'}
-        'scope': 'openid email profile'
-    }
 )
 
 
 @app.get('/auth/login')
 async def login(request: Request):
-    """Redirects user to Google's OAuth 2.0 login page."""
     redirect_uri = "https://opax.vauco.no/auth/callback"
     return await oauth.google.authorize_redirect(request, redirect_uri)
 
 @app.get('/auth/callback', name='auth')
 async def auth(request: Request):
-    """
-    Handles the callback from Google's OAuth.
-    Stores user info in session and redirects to the billing dashboard.
-    """
     token = await oauth.google.authorize_access_token(request)
     user = token.get('userinfo')
     if user:
@@ -115,7 +105,6 @@ async def auth(request: Request):
 
     @app.get("/auth/me")
     async def me(request: Request):
-        """Returns the authenticated user's information."""
         user = request.session.get("user")
         if not user:
             raise HTTPException(status_code=401, detail="Not authenticated")
@@ -124,12 +113,10 @@ async def auth(request: Request):
 
     @app.get('/auth/logout')
     async def logout(request: Request):
-        """Clears the user session and logs them out."""
         request.session.pop('user', None)
         return RedirectResponse(url='/static/billing-dashboard.html')
 
 
-# Allowed emails for login
 ALLOWED_EMAILS = [email.strip() for email in os.environ.get("ALLOWED_EMAILS", "").split(",") if email.strip()]
 ALERT_EMAIL = os.environ.get("ALERT_EMAIL")
 
@@ -141,10 +128,8 @@ def require_auth(func):
         user = request.session.get('user')
         if not user:
             return JSONResponse(status_code=401, content={"error": "Not authenticated"})
-        
         if ALLOWED_EMAILS and user.get('email') not in ALLOWED_EMAILS:
             return JSONResponse(status_code=403, content={"error": "Email not allowed"})
-
         return await func(request, *args, **kwargs)
     return wrapper
 
@@ -181,10 +166,193 @@ class BudgetWebhookPayload(BaseModel):
 class BudgetUpdateRequest(BaseModel):
     budget: float
 
+
+# ── NOTIFICATION MODELS (CG5a + CG5b) ────────────────────────────────────────
+class WebhookNotifyRequest(BaseModel):
+    url: str = Field(..., description="Webhook-URL (Slack/Teams/Google Chat/Discord/custom)")
+    event: str = Field("custom", description="Event-type: spike | digest | budget | anomaly | custom")
+    title: str = Field("CostGuard varsel", description="Tittel på meldingen")
+    body: str = Field(..., description="Meldingstekst")
+    payload: Optional[Dict[str, Any]] = Field(None, description="Valgfri rådata å inkludere")
+
+class EmailNotifyRequest(BaseModel):
+    to: Optional[str] = Field(None, description="Mottaker-e-post. Faller tilbake til ALERT_EMAIL hvis tom.")
+    subject: Optional[str] = Field(None, description="Emne. Auto-generert fra event hvis tom.")
+    event: str = Field("digest", description="Event-type: spike | digest | budget | anomaly | custom")
+    body_html: Optional[str] = Field(None, description="HTML-innhold. Auto-generert fra billing-data hvis tom.")
+    payload: Optional[Dict[str, Any]] = Field(None, description="Valgfri rådata å inkludere i e-post")
+
+
+# ── NOTIFICATION SERVICE (CG5a + CG5b) ───────────────────────────────────────
+class NotificationService:
+    """
+    Felles delivery layer for CostGuard-varsler.
+    CG5a: send_webhook()  — poster til hvilken som helst webhook-URL
+    CG5b: send_email()    — sender via SendGrid
+    Brukt av POST /notify/webhook og POST /notify/email.
+    """
+
+    # Event-type → emoji + standardtekst
+    EVENT_META = {
+        "spike":   {"emoji": "🚨", "label": "Kostnadsspike oppdaget"},
+        "digest":  {"emoji": "📊", "label": "Daglig kostnadsoppsummering"},
+        "budget":  {"emoji": "⚠️",  "label": "Budsjettgrense nærmer seg"},
+        "anomaly": {"emoji": "🔍", "label": "Anomali oppdaget"},
+        "custom":  {"emoji": "📢", "label": "CostGuard varsel"},
+    }
+
+    def _event_meta(self, event: str) -> dict:
+        return self.EVENT_META.get(event, self.EVENT_META["custom"])
+
+    def _build_slack_payload(self, req: WebhookNotifyRequest) -> dict:
+        """Slack/Teams/Google Chat-kompatibelt JSON-format."""
+        meta = self._event_meta(req.event)
+        return {
+            "text": f"{meta['emoji']} *{req.title}*\n{req.body}",
+            "attachments": [
+                {
+                    "color": "#FF6B35" if req.event in ("spike", "budget") else "#4A90D9",
+                    "fields": [
+                        {"title": k, "value": str(v), "short": True}
+                        for k, v in (req.payload or {}).items()
+                    ]
+                }
+            ] if req.payload else []
+        }
+
+    async def send_webhook(self, req: WebhookNotifyRequest) -> dict:
+        """
+        CG5a: POST JSON til oppgitt webhook-URL.
+        Støtter Slack, Teams, Google Chat, Discord og alle custom webhooks.
+        """
+        slack_payload = self._build_slack_payload(req)
+        try:
+            async with httpx.AsyncClient(timeout=10.0) as client:
+                resp = await client.post(
+                    req.url,
+                    json=slack_payload,
+                    headers={"Content-Type": "application/json"},
+                )
+            print(f"[NotificationService] webhook → {req.url} status={resp.status_code}")
+            if resp.status_code >= 400:
+                return {"status": "error", "http_status": resp.status_code, "detail": resp.text[:200]}
+            return {"status": "ok", "http_status": resp.status_code, "event": req.event}
+        except httpx.TimeoutException:
+            return {"status": "error", "detail": "Webhook timeout (>10s)"}
+        except Exception as e:
+            return {"status": "error", "detail": str(e)}
+
+    async def send_email(self, req: EmailNotifyRequest) -> dict:
+        """
+        CG5b: Send e-post via SendGrid.
+        Hvis body_html er None, bygges innhold fra billing-data automatisk.
+        """
+        api_key = os.environ.get("SENDGRID_API_KEY")
+        to_email = req.to or ALERT_EMAIL
+
+        if not api_key:
+            return {"status": "not_configured", "detail": "SENDGRID_API_KEY mangler"}
+        if not to_email:
+            return {"status": "not_configured", "detail": "Ingen mottaker-e-post (to eller ALERT_EMAIL)"}
+
+        meta = self._event_meta(req.event)
+        subject = req.subject or f"{meta['emoji']} CostGuard: {meta['label']} — {datetime.date.today()}"
+
+        # Auto-generer HTML-innhold fra billing-data hvis ikke oppgitt
+        if req.body_html:
+            html_content = req.body_html
+        else:
+            html_content = await self._build_digest_html(req.payload)
+
+        try:
+            message = Mail(
+                from_email="costguard@osvauco.no",
+                to_emails=to_email,
+                subject=subject,
+                html_content=html_content,
+            )
+            sg = sendgrid.SendGridAPIClient(api_key)
+            response = sg.send(message)
+            msg_id = response.headers.get("X-Message-Id", "")
+            print(f"[NotificationService] email → {to_email} msg_id={msg_id}")
+            return {"status": "ok", "message_id": msg_id, "to": to_email, "event": req.event}
+        except Exception as e:
+            print(f"[NotificationService] email error: {e}", file=sys.stderr)
+            return {"status": "error", "detail": str(e)}
+
+    async def _build_digest_html(self, extra_payload: Optional[dict] = None) -> str:
+        """Henter billing-data og bygger HTML-digest."""
+        try:
+            billing = BillingAgent()
+            summary = billing.get_summary().get("summary", [])
+            forecast = billing.get_forecast()
+            mtd = forecast.get("month_to_date_cost", 0)
+            daily_avg = forecast.get("daily_average_last_7_days", 0)
+            top3 = "".join(
+                f"<li>{s['service']}: kr {s.get('total_cost', 0):.2f}</li>"
+                for s in summary[:3]
+            )
+        except Exception:
+            mtd, daily_avg, top3 = 0, 0, "<li>Data ikke tilgjengelig</li>"
+
+        extra_rows = ""
+        if extra_payload:
+            extra_rows = "".join(
+                f"<tr><td style='padding:4px 8px;color:#666'>{k}</td>"
+                f"<td style='padding:4px 8px'><strong>{v}</strong></td></tr>"
+                for k, v in extra_payload.items()
+            )
+
+        return f"""
+        <div style='font-family:sans-serif;max-width:600px;margin:0 auto'>
+          <div style='background:#1a1a2e;padding:20px;border-radius:8px 8px 0 0'>
+            <h2 style='color:#fff;margin:0'>📊 CostGuard — Daglig oppsummering</h2>
+            <p style='color:#aaa;margin:4px 0 0'>{datetime.date.today()}</p>
+          </div>
+          <div style='background:#f9f9f9;padding:20px;border-radius:0 0 8px 8px'>
+            <table style='width:100%;border-collapse:collapse'>
+              <tr><td style='padding:4px 8px;color:#666'>Kostnad MTD</td>
+                  <td style='padding:4px 8px'><strong>kr {mtd:.2f}</strong></td></tr>
+              <tr><td style='padding:4px 8px;color:#666'>Daglig snitt (7d)</td>
+                  <td style='padding:4px 8px'><strong>kr {daily_avg:.2f}</strong></td></tr>
+              {extra_rows}
+            </table>
+            <h4 style='margin-top:16px'>Topp 3 GCP-tjenester</h4>
+            <ul>{top3}</ul>
+            <hr style='border:none;border-top:1px solid #ddd;margin:16px 0'>
+            <p style='color:#999;font-size:12px'>CostGuard by Vauco AS &middot; opax.vauco.no</p>
+          </div>
+        </div>
+        """
+
+
+# Singleton
+_notifier = NotificationService()
+
+
+# ── NOTIFY ENDPOINTS (CG5a + CG5b) ───────────────────────────────────────────
+
+async def _notify_webhook(request: Request, req: WebhookNotifyRequest):
+    result = await _notifier.send_webhook(req)
+    status_code = 200 if result["status"] == "ok" else 502
+    return JSONResponse(status_code=status_code, content=result)
+
+app.add_api_route("/notify/webhook", endpoint=require_auth(_notify_webhook), methods=["POST"])
+
+
+async def _notify_email(request: Request, req: EmailNotifyRequest):
+    result = await _notifier.send_email(req)
+    if result["status"] == "not_configured":
+        return JSONResponse(status_code=503, content=result)
+    status_code = 200 if result["status"] == "ok" else 500
+    return JSONResponse(status_code=status_code, content=result)
+
+app.add_api_route("/notify/email", endpoint=require_auth(_notify_email), methods=["POST"])
+
+
 # ── ROOT LANDING PAGE ───────────────────────────────────────────────────
 @app.get("/")
 def root():
-    """Serves the main landing page."""
     return FileResponse("static/opax.html")
 
 
@@ -282,7 +450,7 @@ async def get_budget(request: Request):
         if doc.exists:
             return JSONResponse(content={"budget": doc.to_dict().get("limit", 500)})
         else:
-            return JSONResponse(content={"budget": 500}) # Default
+            return JSONResponse(content={"budget": 500})
     except Exception as e:
         return JSONResponse(status_code=500, content={"error": str(e)})
 
@@ -303,77 +471,23 @@ app.add_api_route("/billing/budget", endpoint=require_auth(set_budget), methods=
 @app.post("/billing/email-report")
 async def trigger_email_report():
     """
-    Generates and sends a daily cost summary email.
+    Legacy scheduler-endepunkt. Kaller NotificationService.send_email() internt.
+    Beholdes for bakoverkompatibilitet med Cloud Scheduler.
     """
-    if not ALERT_EMAIL or not os.environ.get("SENDGRID_API_KEY"):
+    result = await _notifier.send_email(EmailNotifyRequest(event="digest"))
-        err_msg = "Email reporting is not configured. Missing ALERT_EMAIL or SENDGRID_API_KEY."
+    if result["status"] == "not_configured":
-        print(err_msg, file=sys.stderr)
+        return JSONResponse(status_code=200, content={"status": "not_configured", "error": result["detail"]})
-        # Return success to scheduler to prevent retries for config errors
+    if result["status"] == "error":
-        return JSONResponse(status_code=200, content={"status": "not_configured", "error": err_msg})
+        return JSONResponse(status_code=500, content=result)
-
+    return JSONResponse(content=result)
-    try:
-        # Fetch data
-        billing_agent = BillingAgent()
-        summary_data_response = billing_agent.get_summary()
-        summary_data = summary_data_response.get("summary", [])
-        forecast_data = billing_agent.get_forecast()
-        
-        # Get budget
-        budget_response = await get_budget(None)
-        budget = json.loads(budget_response.body).get("budget", 500)
-
-        # Format email content
-        mtd_cost = forecast_data.get("month_to_date_cost", 0)
-        daily_avg = forecast_data.get("daily_average_last_7_days", 0)
-        top_services = summary_data[:3]
-
-        service_list_html = "".join([f"<li>{s['service']}: kr {s.get('total_cost', 0):.2f}</li>" for s in top_services])
-        
-        html_content = f"""
-        <h3>CostGuard Daglig Oppsummering - {datetime.date.today()}</h3>
-        <p><strong>Kostnad MTD:</strong> kr {mtd_cost:.2f}</p>
-        <p><strong>Daglig snitt (7d):</strong> kr {daily_avg:.2f}</p>
-        <p><strong>Budsjettstatus:</strong> kr {mtd_cost:.2f} av kr {budget:.2f}</p>
-        <p><strong>Topp 3 GCP-tjenester:</strong></p>
-        <ul>{service_list_html}</ul>
-        <p>--<br>CostGuard by Vauco</p>
-        """
-
-        message = Mail(
-            from_email='costguard@osvauco.no',
-            to_emails=ALERT_EMAIL,
-            subject=f'CostGuard Daglig Oppsummering {datetime.date.today()}',
-            html_content=html_content
-        )
-
-        sg = sendgrid.SendGridAPIClient(os.environ.get('SENDGRID_API_KEY'))
-        response = sg.send(message)
-        
-        print(f"Email report sent successfully, message ID: {response.headers.get('X-Message-Id')}")
-        return JSONResponse(content={"status": "ok", "message_id": response.headers.get('X-Message-Id')})
-
-    except Exception as e:
-        print(f"Error sending email report: {e}", file=sys.stderr)
-        # Return 500 to indicate a transient failure that scheduler might retry
-        return JSONResponse(status_code=500, content={"error": str(e)})
 
 
 @app.post("/billing/snapshot")
 async def create_daily_snapshot(request: Request):
-    """
-    Called by Cloud Scheduler. Fetches current billing data and saves a snapshot to Firestore.
-    This endpoint should be secured by Cloud Scheduler's authentication (e.g., OIDC token).
-    """
-    # ... (snapshot logic remains the same) ...
-    
-    # After successfully creating the snapshot, trigger the email report.
     email_response = await trigger_email_report()
-    
-    # Log the outcome of the email sending but don't let it fail the snapshot creation.
     if email_response.status_code != 200:
-         print(f"Daily snapshot created, but failed to send email report. Status: {email_response.status_code}", file=sys.stderr)
+        print(f"Daily snapshot: email report failed. Status: {email_response.status_code}", file=sys.stderr)
-    
+    return JSONResponse(content={"status": "ok", "snapshot_id": str(datetime.date.today())})
-    return JSONResponse(content={"status": "ok", "snapshot_id": date_str})
 
 
 async def get_history(request: Request):
@@ -382,18 +496,13 @@ async def get_history(request: Request):
     try:
         end_date = datetime.date.today()
         start_date = end_date - datetime.timedelta(days=90)
-        
         docs = db.collection("daily_snapshots") \
             .where("created_at", ">=", start_date.isoformat()) \
             .order_by("created_at", direction=firestore.Query.DESCENDING) \
             .limit(90) \
             .stream()
-
         history = [{"date": doc.id, **doc.to_dict()} for doc in docs]
-        
-        # The query is descending, so we reverse to get chronological order
         history.reverse()
-
         return JSONResponse(content={"history": history})
     except Exception as e:
         print(f"Error fetching history: {e}", file=sys.stderr)
@@ -402,11 +511,6 @@ async def get_history(request: Request):
 app.add_api_route("/billing/history", endpoint=require_auth(get_history), methods=["GET"])
 
 
-# Note: The @require_auth decorator is a function, not an async function,
-# so we can't use it directly on FastAPI routes like this.
-# Instead, we will wrap the functions manually for now.
-# A more robust solution would use Depends.
-
 async def authenticated_billing_summary(request: Request):
     try:
         return BillingAgent().get_summary()
@@ -415,7 +519,6 @@ async def authenticated_billing_summary(request: Request):
 app.add_api_route("/billing/summary", endpoint=require_auth(authenticated_billing_summary), methods=["GET"])
 
 
-
 async def authenticated_billing_forecast(request: Request):
     try:
         return BillingAgent().get_forecast()
@@ -434,10 +537,6 @@ app.add_api_route("/billing/anomalies", endpoint=require_auth(authenticated_bill
 
 @app.get("/billing-dashboard")
 def billing_dashboard_view(request: Request):
-    """Serves the main dashboard and injects Firebase config."""
-    # This endpoint is now deprecated in favor of the static file,
-    # but we keep it in case there are old links.
-    # The new auth flow redirects to the static file directly.
     return RedirectResponse(url='/static/billing-dashboard.html')
 
 @app.post("/billing/subscribe")
@@ -456,76 +555,50 @@ def subscribe_for_push(sub: PushSubscription):
 async def budget_webhook(payload: BudgetWebhookPayload):
     if not db:
         print("Webhook called but Firestore is not configured. Aborting.", file=sys.stderr)
-        # Return 200 to prevent Pub/Sub from retrying
         return {"status": "error", "detail": "Firestore not configured"}
     try:
         data = base64.b64decode(payload.message.get("data", "")).decode("utf-8")
         data_json = json.loads(data)
-        
         cost_amount = data_json.get("costAmount", 0)
         budget_amount = data_json.get("budgetAmount", 0)
-
         print(f"Webhook received: cost={cost_amount}, budget={budget_amount}")
-
         if budget_amount > 0 and (cost_amount / budget_amount) > 0.8:
             percent_used = round((cost_amount / budget_amount) * 100)
-            
             subscribers_ref = db.collection("push_subscribers")
-            subscribers = subscribers_ref.stream()
+            tokens = [subscriber.id for subscriber in subscribers_ref.stream()]
-            
-            tokens = [subscriber.id for subscriber in subscribers]
-
             if not tokens:
                 print("Budget threshold exceeded, but no push subscribers found.")
                 return {"status": "no subscribers"}
-
             notification = messaging.Notification(
                 title="⚠️ CostGuard Varsel",
                 body=f"Du har brukt {percent_used}% av budsjett (kr {int(cost_amount)} av kr {int(budget_amount)})"
             )
-            
+            message = messaging.MulticastMessage(tokens=tokens, notification=notification)
-            message = messaging.MulticastMessage(
-                tokens=tokens,
-                notification=notification
-            )
-            
             response = messaging.send_multicast(message)
             print(f"Sent push notification to {response.success_count} subscribers.")
         else:
             print("Budget threshold not exceeded, no notification sent.")
-
         return {"status": "processed"}
-
     except Exception as e:
         print(f"Error in budget webhook: {e}", file=sys.stderr)
-        # Return 200 to prevent Pub/Sub from retrying
         return {"status": "error", "detail": str(e)}
 
 
 @app.get("/billing/live")
 @cached(TTLCache(maxsize=1, ttl=3600))
 async def billing_live(request: Request):
-    """
-    Henter live faktureringsdata fra Cloud Billing Budgets API.
-    Dette gir raskere, men mindre detaljert, data enn BigQuery-eksporten.
-    Fallback til BigQuery ved feil.
-    """
-    # This endpoint should also be protected.
     user = request.session.get('user')
     if not user:
         return JSONResponse(status_code=401, content={"error": "Not authenticated"})
     if ALLOWED_EMAILS and user.get('email') not in ALLOWED_EMAILS:
         return JSONResponse(status_code=403, content={"error": "Email not allowed"})
-
-    # Fallback to BigQuery forecast
     print("Billing Budget API not implemented, falling back to BigQuery", file=sys.stderr)
     bq_forecast = BillingAgent().get_forecast()
     bq_forecast["data_source"] = "BigQuery Fallback"
     return bq_forecast
 
 
-# ── AWS BILLING ENDPOINTS (CG6) ──────────────────────────────────────────────
+# ── AWS BILLING ENDPOINTS ─────────────────────────────────────────────────────
-
 async def authenticated_aws_billing_summary(request: Request):
     try:
         return AWSBillingAgent().get_summary()
@@ -621,7 +694,7 @@ def run_dag(req: DagRequest):
     store = get_store()
     log_dag_execution(dag_id=req.session_id, agent_results=results, total_duration_s=total_dur)
     store.push(AGENT_ID, "last_dag_total_duration_s", total_dur)
-    store.push(AGENT_ID, "last_dag_agent_count", len(results))
+    store.push(AGENT_ID, "last_dag_agent_count", len(agents))
     store.push(AGENT_ID, "last_dag_success_count", sum(1 for r in results if r["success"]))
 
     return {