Feature/notification dedup

.env.example

@@ -7,4 +7,10 @@ LLM_PROVIDER=anthropic
 LLM_MODEL=claude-sonnet-4-20250514
 ANTHROPIC_API_KEY=sk-ant-...
 DAILY_COST_CAP_USD=10.0
-ADMIN_USER_IDS=your-admin-user-id
+ADMIN_USER_IDS=your-admin-user-id
+# --- Notification Service Settings ---
+# SMTP configuration for LPI Platform notifications
+SMTP_SERVER=smtp.gmail.com
+SMTP_PORT=587
+[email protected]
+SMTP_PASS=your-app-specific-password

README.md

@@ -417,6 +417,10 @@ cp .env.example .env
 | `SUPABASE_SERVICE_ROLE_KEY` | ✅ | Same as above; used by logging utilities |
 | `SUPABASE_JWT_SECRET` | ✅ | JWT signing secret — from `supabase status` or dashboard → Project Settings → API |
 | `GROQ_API_KEY` | ✅ | Groq API key for the LLM layer (free tier) — [console.groq.com](https://console.groq.com) |
+|`SMTP_SERVER`   | ✅ | SMTP server for notifications (e.g., smtp.gmail.com) |
+|`SMTP_PORT`   | ✅ | Port for SMTP (e.g., 587) |
+|`SMTP_USER`   | ✅ | Email address for notification dispatch |
+|`SMTP_PASS`   | ✅ | App-specific password for email account |
 | `ANTHROPIC_API_KEY` | Optional | Claude API key — alternate LLM provider |
 | `LLM_PROVIDER` | Optional | `groq` (default) or `anthropic` |
 | `LLM_MODEL` | Optional | Default: `llama-3.3-70b-versatile` (Groq) |
@@ -814,6 +818,9 @@ All migrations live in `supabase/migrations/`. Run `supabase db push` to apply t
 | `20260615000000_signals_rls_and_log_action.sql` | RLS on `activity_signals` + CHECK fix | Adil |
 | `20260621000000_create_recommendation_feedback.sql` | `recommendation_feedback` | Aryan |
 | `20260625000000_activity_signals_goal_fk.sql` | `goal_id` FK on `activity_signals` + strict goal-scoped RLS | Jaivardhan |
+| `20260627000000_create_notifications.sql` | notifications | Aditi |
+| `20260630000002_create_users.sql` | users profile table | Aditi |
+| `20260630000003_user_sync_trigger.sql` | Auth-to-Profile Sync Trigger | Aditi |
 
 ### Table overview
 
@@ -825,6 +832,8 @@ All migrations live in `supabase/migrations/`. Run `supabase db push` to apply t
 | `system_logs` | Platform-level events (`info`, `warning`, `error`) |
 | `activity_signals` | All ingested activity events from all streams |
 | `recommendation_feedback` | User accept/dismiss decisions on recommendation cards |
+| `notifications` | Audit trail of sent notifications (prevents duplicates) |
+| `users` | User authentication data + profile metadata (name, email, dob, gender, bio) |
 
 ### Useful SQL queries
 

src/lpi/config.py

@@ -19,6 +19,10 @@ class Settings(BaseSettings):
     github_client_id: str = ""
     github_client_secret: str = ""
     admin_user_ids: str = ""
+    smtp_server: str = "smtp.gmail.com"
+    smtp_port: int = 587
+    smtp_user: str = ""
+    smtp_pass: str = ""
 
     @property
     def admin_ids_list(self) -> list[str]:
@@ -33,6 +37,9 @@ def admin_ids_list(self) -> list[str]:
         "supabase_jwt_secret",
         "anthropic_api_key",
         "groq_api_key",
+        "smtp_server",
+        "smtp_user",
+        "smtp_pass",
         mode="before",
     )
     @classmethod

src/lpi/notifications.py

@@ -0,0 +1,83 @@
+import logging
+import os
+import smtplib
+from collections import defaultdict
+from email.message import EmailMessage
+
+from dotenv import load_dotenv
+
+from lpi.store import _get_client, get_user_email
+
+load_dotenv()
+logger = logging.getLogger(__name__)
+
+_NOTIF_TEMPLATES = {
+    "PushEvent": ("New GitHub Push 📤", "Activity detected in {repo}.\n\n💡 Insight: {explanation}"),
+    "pr_merged":        ("PR Merged 🎉", "You merged PR #{pr_number}: '{title}' in {repo}.\n\n💡 Insight: {explanation}"),
+    "commit_pushed":    ("New Commits 📦", "A new push was detected in {repo}.\n\n💡 Insight: {explanation}"),
+    "phase_advanced":   ("SMILE Phase Advanced ✨", "Goal '{title}' moved to {phase}."),
+    "inactivity_alert": ("Inactivity Detected ⚠️", "No activity detected in {repo} for {days} days."),
+    "pr_opened": ("PR Opened 🚀", "{actor} opened PR #{pr_number}: '{title}' in {repo}.\n\n💡 Insight: {explanation}")
+}
+
+def _dispatch_email(target_email: str, title: str, body: str) -> None:
+    SMTP_SERVER = os.getenv("SMTP_SERVER", "smtp.gmail.com")
+    SMTP_PORT = int(os.getenv("SMTP_PORT", 587))
+    SMTP_USER = os.getenv("SMTP_USER")
+    SMTP_PASS = os.getenv("SMTP_PASS")
+
+    if not SMTP_USER or not SMTP_PASS:
+        logger.warning("Email blocked: SMTP credentials missing.")
+        return
+
+    msg = EmailMessage()
+    msg.set_content(body)
+    msg["Subject"] = title
+    msg["From"] = SMTP_USER
+    msg["To"] = target_email
+
+    try:
+        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
+            server.starttls()
+            server.login(SMTP_USER, SMTP_PASS)
+            server.send_message(msg)
+        logger.info(f"📧 SUCCESS: Email sent to {target_email}")
+    except Exception as e:
+        logger.exception(f"SMTP ERROR for {target_email}: {e}")
+
+def create_notification_if_new(user_id: str, signal_id: str, event_type: str, payload: dict) -> bool:
+    template = _NOTIF_TEMPLATES.get(event_type)
+    if not template: 
+        return False
+
+    title_tmpl, body_tmpl = template
+    safe_payload = defaultdict(lambda: "[N/A]", payload)
+
+    # Fallback explanation
+    if "explanation" not in safe_payload:
+        safe_payload["explanation"] = "Every contribution helps move your project forward."
+
+    try:
+        body = body_tmpl.format_map(safe_payload)
+    except Exception:
+        body = body_tmpl 
+
+    try:
+        client = _get_client()
+        client.table("notifications").insert({
+            "user_id":  user_id,
+            "signal_id": signal_id,
+            "type":     event_type,
+            "title":    title_tmpl,
+            "body":     body,
+        }).execute()
+    except Exception as e:
+        if "unique" in str(e).lower(): 
+            return False
+        logger.exception(f"Failed to record notification: {signal_id}")
+        return False
+
+    target_email = get_user_email(user_id)
+    if target_email:
+        _dispatch_email(target_email, title_tmpl, body)
+    return True

src/lpi/routers/me.py

@@ -1,11 +1,49 @@
-from fastapi import APIRouter, Depends
+from datetime import date
 
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel
+
+from lpi import store
 from lpi.middleware.auth import UserContext, get_current_user_context
 
 router = APIRouter()
 
+# 1. Define the expected payload from the frontend
+class ProfileUpdate(BaseModel):
+    name: str | None = None
+    gender: str | None = None
+    dob: date | None = None
+    bio: str | None = None
 
+# 2. Existing GET route
 @router.get("/", response_model=UserContext)
 def get_me(user_context: UserContext = Depends(get_current_user_context)) -> UserContext:
     """Return the authenticated user's context (including admin status)."""
     return user_context
+
+# 3. New PATCH route for profile updates
+@router.patch("/profile", summary="Update user profile")
+def update_profile(
+    profile_data: ProfileUpdate,
+    user_context: UserContext = Depends(get_current_user_context),
+):
+    """Updates the authenticated user's profile details."""
+    # exclude_unset ensures we only update fields the user actually submitted
+    updates = profile_data.model_dump(exclude_unset=True)
+
+    if "dob" in updates and updates["dob"]:
+        updates["dob"] = updates["dob"].isoformat()
+
+    if not updates:
+        return {"status": "no changes provided"}
+
+    # Use the user_id from the verified JWT context
+    updated_user = store.update_user_profile(user_id=user_context.user_id, updates=updates)
+
+    if not updated_user:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to update profile in database."
+        )
+
+    return {"status": "success", "profile": updated_user}

src/lpi/routers/signals.py

@@ -66,10 +66,18 @@
 from lpi import store
 from lpi.middleware.auth import UserContext, get_current_user, get_current_user_context
 from lpi.models import Signal, SignalCreate
-from lpi.utils.logging import log_user_activity
+from lpi.notifications import create_notification_if_new
+from lpi.utils.logging import log_user_activity, logger
 
 router = APIRouter()
 
+def _generate_explanation(event_type: str, payload: dict) -> str:
+    """Generates a rule-based explanation for signals."""
+    if event_type == "commit_pushed":
+        return "You're actively pushing code. Keep iterating!"
+    if event_type == "pr_merged":
+        return "Merging a PR is a significant milestone that moves your goal forward toward next phase."
+    return "Your project is showing activity—every small update contributes to your long-term goals."
 
 # ── Wave 2: POST /api/v1/signals/ ────────────────────────────────────────────
 
@@ -147,6 +155,22 @@ def ingest_signal(
     # utils/logging.py). Wrapping it here again would be redundant and,
     # worse, gives a false impression that THIS is where a logging failure
     # gets caught — it isn't; this call simply cannot raise.
+    # MAP THE TYPE FOR THE NOTIFICATION TEMPLATE
+    mapped_type = new_signal.event_type
+    if new_signal.event_type == "PushEvent":
+        mapped_type = "commit_pushed"
+    elif new_signal.event_type == "PullRequestEvent":
+        mapped_type = "pr_merged"
+
+    create_notification_if_new(
+        user_id=user_id,
+        signal_id=new_signal.id,
+        event_type=mapped_type,  # Use the mapped semantic key
+        payload={
+            ** (new_signal.payload or {}),
+            "explanation": _generate_explanation(new_signal.event_type, new_signal.payload or {})
+        },
+    )
     log_user_activity(
         user_id=user_id,
         action="signal_ingested",
@@ -386,8 +410,13 @@ async def sync_github_events(
 
             # Build the full Signal object (mirroring the logic in ingest_signal)
             now = datetime.now(UTC)
+
+            # --- FIX 1: Deterministic UUID for Deduplication ---
+            github_event_id = str(event.get("id"))
+            consistent_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, github_event_id))
+
             new_signal = Signal(
-                id=str(uuid.uuid4()),
+                id=consistent_id,  # <- The database will now recognize duplicates!
                 user_id=user_id,
                 timestamp=now,
                 **signal_create.model_dump()
@@ -397,6 +426,36 @@ async def sync_github_events(
             store.insert_signal(new_signal)
             ingested_count += 1
 
+           # --- FIX 2: Trigger the notification service ---
+            if event_type == "PushEvent":
+                notif_type = "commit_pushed"
+                notif_payload = {
+                    "repo": repo_name,
+                    "explanation": "Your project is showing activity—every small update contributes to your long-term goals."
+                }
+            elif event_type == "PullRequestEvent":
+                notif_type = "pr_merged" 
+                gh_payload = event.get("payload", {})
+                pr_data = gh_payload.get("pull_request", {})
+
+                notif_payload = {
+                    "repo": repo_name,
+                    "pr_number": pr_data.get("number", "Unknown"),
+                    "title": pr_data.get("title", "Pull Request Updated"),
+                    "explanation": _generate_explanation("pr_merged", {})
+                }
+
+            # TRIGGER NOTIFICATION ONCE HERE
+            try:
+                create_notification_if_new(
+                    user_id=user_id,
+                    signal_id=new_signal.id,
+                    event_type=notif_type, 
+                    payload=notif_payload,
+                )
+            except Exception as e:
+                logger.error(f"Notification background task failed: {e}")
+
             # Log the activity
             log_user_activity(
                 user_id=user_id,
@@ -417,4 +476,4 @@ async def sync_github_events(
         "ingested_high_value": ingested_count,
         "repo": repo_name,
         "goal_id": goal_id
-    }
+    }

src/lpi/routers/webhooks.py

@@ -6,6 +6,7 @@
 
 from lpi import store
 from lpi.models import Signal
+from lpi.notifications import create_notification_if_new
 from lpi.routers.github_auth import repo_db
 
 router = APIRouter()
@@ -46,14 +47,19 @@ async def github_webhook_receiver(request: Request):
         }
 
     # 3. Catch Pushed Commits
-    elif event_type == "push" and payload.get("commits"):
+    elif event_type == "push":
+        # Webhook 'push' event has 'commits' and 'ref' at the top level
+        commits = payload.get("commits", [])
+        ref = payload.get("ref", "")
+
         signal_data = {
             "event_type": "commit_pushed",
             "payload": {
-                "repo": payload["repository"]["name"],
-                "branch": payload.get("ref", "").replace("refs/heads/", ""),
-                "commit_count": len(payload["commits"]),
-                "last_commit_message": payload["commits"][-1]["message"],
+                "repo": payload.get("repository", {}).get("name"),
+                "branch": ref.replace("refs/heads/", ""),
+                "commit_count": len(commits),
+                "last_commit_message": commits[-1].get("message") if commits else "New code pushed",
+                "explanation": "You're actively pushing code. Keep iterating!"
             },
         }
 
@@ -89,4 +95,11 @@ async def github_webhook_receiver(request: Request):
         store.insert_signal(signal)
         print(f"✅ AUTOMATIC DETECTION: Saved {signal_data['event_type']} for user {user_id} and goal {target_goal_id}!")
 
+        create_notification_if_new(
+            user_id=user_id,
+            signal_id=signal.id,
+            event_type=signal.event_type,
+            payload=signal.payload or {},
+        )
+
     return {"status": "success"}

src/lpi/store.py

@@ -341,6 +341,33 @@ def get_signal(signal_id: str) -> Signal | None:
         return None
     return Signal(**cast(dict, result.data[0]))
 
+def get_user_email(user_id: str) -> str | None:
+    """Fetch a user's email address for notifications."""
+    try:
+        result = _get_client().table("users").select("email").eq("id", user_id).execute()
+        # Cast result.data to a list to check length safely
+        if result.data and len(cast(list, result.data)) > 0:
+            # Cast the first row to a dict before calling .get()
+            row = cast(dict, result.data[0])
+            email = row.get("email")
+            # Ensure the return type strictly matches str | None
+            return str(email) if email else None
+        return None
+    except Exception as e:
+        print(f"Error fetching email for user {user_id}: {e}")
+        return None
+
+def update_user_profile(user_id: str, updates: dict) -> dict | None:
+    """Updates a user's profile information in the public.users table."""
+    try:
+        result = _get_client().table("users").update(updates).eq("id", user_id).execute()
+        if result.data and len(cast(list, result.data)) > 0:
+            # Explicitly return a dict to satisfy the function signature
+            return cast(dict, result.data[0])
+        return None
+    except Exception as e:
+        print(f"Error updating profile for user {user_id}: {e}")
+        return None
 
 # ── Audit log verification (new — used by tests, also useful for admin tooling) ─