Merge branch 'main' into feat/onboarding-focused-workflow

Author: Dharya4242

backend/app/agents/devrel/nodes/gather_context.py

@@ -4,18 +4,25 @@
 
 from app.agents.state import AgentState
 from app.services.auth.management import get_or_create_user_by_discord
+from app.database.supabase.services import ensure_user_exists, get_conversation_context
 
 logger = logging.getLogger(__name__)
 
 async def gather_context_node(state: AgentState) -> Dict[str, Any]:
     """Gather additional context for the user and their request"""
     logger.info(f"Gathering context for session {state.session_id}")
 
-    # TODO: Add context gathering from databases
-    # Currently, context is simple
-    # In production, query databases for user history, etc.
-
     original_message = state.context.get("original_message", "")
+    author_info = state.context.get("author", {})
+
+    # Ensure user exists in database
+    user_uuid = await ensure_user_exists(
+        user_id=state.user_id,
+        platform=state.platform,
+        username=author_info.get("username"),
+        display_name=author_info.get("display_name"),
+        avatar_url=author_info.get("avatar_url")
+    )
 
     new_message = {
         "role": "user",
@@ -47,9 +54,36 @@ async def gather_context_node(state: AgentState) -> Dict[str, Any]:
     context_data = {
         "user_profile": profile_data or {"user_id": state.user_id, "platform": state.platform},
         "conversation_context": len(state.messages) + 1,  # +1 for the new message
-        "session_info": {"session_id": state.session_id}
+        "session_info": {"session_id": state.session_id},
+        "user_uuid": user_uuid
     }
 
+    # Only retrieve from database if we don't have conversation context already
+    should_fetch_from_db = not state.conversation_summary and not state.key_topics
+
+    if user_uuid and should_fetch_from_db:
+        logger.info(f"No existing context in state, fetching from database for user {user_uuid}")
+        prev_context = await get_conversation_context(user_uuid)
+        if prev_context:
+            logger.info(f"Retrieved previous conversation context from database")
+            context_data["previous_conversation"] = prev_context
+
+            # Populate state with previous conversation summary and topics
+            return {
+                "messages": [new_message],
+                "context": {**state.context, **context_data},
+                "conversation_summary": prev_context.get("conversation_summary"),
+                "key_topics": prev_context.get("key_topics", []),
+                "current_task": "context_gathered",
+                "last_interaction_time": datetime.now()
+            }
+        else:
+            logger.info(f"No previous conversation context found in database")
+    else:
+        if not should_fetch_from_db:
+            logger.info(
+                f"Using existing context from state (conversation_summary: {bool(state.conversation_summary)}, key_topics: {len(state.key_topics)})")
+
     updated_context = {**state.context, **context_data}
 
     result: Dict[str, Any] = {

backend/app/agents/devrel/nodes/generate_response.py

@@ -1,9 +1,11 @@
 import logging
 import json
 from typing import Dict, Any
+from datetime import datetime
 from app.agents.state import AgentState
 from langchain_core.messages import HumanMessage
 from ..prompts.response_prompt import RESPONSE_PROMPT
+from app.database.supabase.services import store_interaction
 
 logger = logging.getLogger(__name__)
 
@@ -16,6 +18,9 @@ async def generate_response_node(state: AgentState, llm) -> Dict[str, Any]:
     try:
         final_response = await _create_response(state, llm)
 
+        # Store interaction to database
+        await _store_interaction_to_db(state, final_response)
+
         return {
             "final_response": final_response,
             "current_task": "response_generated"
@@ -99,3 +104,43 @@ def _get_latest_message(state: AgentState) -> str:
     if state.messages:
         return state.messages[-1].get("content", "")
     return state.context.get("original_message", "")
+
+async def _store_interaction_to_db(state: AgentState, final_response: str) -> None:
+    """Store the interaction to database"""
+    try:
+        user_uuid = state.context.get("user_uuid")
+
+        if not user_uuid:
+            logger.warning(f"No user_uuid in context, skipping interaction storage for session {state.session_id}")
+            return
+
+        # Get the latest user message content
+        latest_message = _get_latest_message(state)
+
+        # Extract classification data
+        classification = state.context.get("classification", {})
+        # TODO: intent key not present in classification schema (contains: needs_devrel, priority, reasoning)
+        # Modify prompt to include intent key
+        intent = classification.get("reasoning")  # Fallback to reasoning for intent
+
+        # Store the interaction
+        await store_interaction(
+            user_uuid=user_uuid,
+            platform=state.platform,
+            platform_specific_id=f"{state.session_id}_{datetime.now().timestamp()}",
+            channel_id=state.channel_id,
+            thread_id=state.thread_id,
+            content=latest_message,
+            interaction_type="message",
+            intent_classification=intent,
+            topics_discussed=state.key_topics if state.key_topics else None,
+            metadata={
+                "session_id": state.session_id,
+                "response": final_response[:500] if final_response else None,
+                "tools_used": state.tools_used,
+                "classification": classification
+            }
+        )
+
+    except Exception as e:
+        logger.error(f"Error storing interaction to database: {str(e)}")

backend/app/agents/devrel/nodes/summarization.py

@@ -145,18 +145,22 @@ async def store_summary_to_database(state: AgentState) -> None:
             logger.error(f"Missing required fields: user_id={state.user_id}, platform={state.platform}")
             return
 
-        platform_id = state.user_id
-        platform_column = f"{state.platform}_id"
+        user_uuid = state.context.get("user_uuid")
 
-        # Fetch the user's UUID from the 'users' table
-        user_response = await supabase.table("users").select("id").eq(platform_column, platform_id).limit(1).execute()
+        if not user_uuid:
+            platform_id = state.user_id
+            platform_column = f"{state.platform}_id"
 
-        if not user_response.data:
-            logger.error(f"User with {platform_column} '{platform_id}' not found in users table.")
-            return
+            user_response = await supabase.table("users").select("id").eq(platform_column, platform_id).limit(1).execute()
+
+            if not user_response.data:
+                logger.error(f"User with {platform_column} '{platform_id}' not found in users table.")
+                return
 
-        user_uuid = user_response.data[0]['id']
-        logger.info(f"Found user UUID: {user_uuid} for {platform_column}: {platform_id}")
+            user_uuid = user_response.data[0]['id']
+            logger.info(f"Found user UUID: {user_uuid} for {platform_column}: {platform_id}")
+        else:
+            logger.info(f"Using cached user UUID from context: {user_uuid}")
 
         # Record to insert/update
         record = {
@@ -178,4 +182,4 @@ async def store_summary_to_database(state: AgentState) -> None:
             logger.error(f"❌ Supabase upsert failed for session {state.session_id}: {response}")
 
     except Exception as e:
-        logger.error(f"Unexpected error while storing summary: {str(e)}")
+        logger.error(f"Unexpected error while storing summary: {str(e)}")

backend/app/api/router.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter
 from .v1.auth import router as auth_router
 from .v1.health import router as health_router
+from .v1.integrations import router as integrations_router
 
 api_router = APIRouter()
 
@@ -16,4 +17,10 @@
     tags=["Health"]
 )
 
+api_router.include_router(
+    integrations_router,
+    prefix="/v1/integrations",
+    tags=["Integrations"]
+)
+
 __all__ = ["api_router"]

backend/app/api/v1/integrations.py

@@ -0,0 +1,105 @@
+from fastapi import APIRouter, HTTPException, Depends, status
+from uuid import UUID
+from app.models.integration import (
+    IntegrationCreateRequest,
+    IntegrationUpdateRequest,
+    IntegrationResponse,
+    IntegrationListResponse,
+    IntegrationStatusResponse
+)
+from app.services.integration_service import integration_service, NotFoundError
+from app.core.dependencies import get_current_user
+
+router = APIRouter()
+
+
+@router.post("/", response_model=IntegrationResponse, status_code=status.HTTP_201_CREATED)
+async def create_integration(
+    request: IntegrationCreateRequest,
+    user_id: UUID = Depends(get_current_user)
+):
+    """Create a new organization integration."""
+    try:
+        return await integration_service.create_integration(user_id, request)
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) from e
+
+
+@router.get("/", response_model=IntegrationListResponse)
+async def list_integrations(user_id: UUID = Depends(get_current_user)):
+    """List all integrations for the current user."""
+    try:
+        integrations = await integration_service.get_integrations(user_id)
+        return IntegrationListResponse(integrations=integrations, total=len(integrations))
+    except Exception as e:
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) from e
+
+
+@router.get("/status/{platform}", response_model=IntegrationStatusResponse)
+async def get_integration_status(
+    platform: str,
+    user_id: UUID = Depends(get_current_user)
+):
+    """Get the status of a specific platform integration."""
+    try:
+        return await integration_service.get_integration_status(user_id, platform)
+    except Exception as e:
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) from e
+
+@router.get("/{integration_id}", response_model=IntegrationResponse)
+async def get_integration(
+    integration_id: UUID,
+    user_id: UUID = Depends(get_current_user)
+):
+    """Get a specific integration."""
+    try:
+        integration = await integration_service.get_integration(user_id, integration_id)
+
+        if not integration:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Integration not found"
+            )
+
+        return integration
+    except Exception as e:
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) from e
+
+@router.put("/{integration_id}", response_model=IntegrationResponse)
+async def update_integration(
+    integration_id: UUID,
+    request: IntegrationUpdateRequest,
+    user_id: UUID = Depends(get_current_user)
+):
+    """Update an existing integration."""
+    try:
+        return await integration_service.update_integration(user_id, integration_id, request)
+    except NotFoundError as e:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) from e
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to update integration: {str(e)}"
+        ) from e
+
+@router.delete("/{integration_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_integration(
+    integration_id: UUID,
+    user_id: UUID = Depends(get_current_user)
+):
+    """Delete an integration."""
+    try:
+        await integration_service.delete_integration(user_id, integration_id)
+    except NotFoundError as e:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) from e
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to delete integration: {str(e)}"
+        ) from e

backend/app/core/dependencies.py

@@ -1,12 +1,69 @@
-from fastapi import Request
+from fastapi import Header, HTTPException, status, Request
+from uuid import UUID
+from app.database.supabase.client import get_supabase_client
+import logging
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from main import DevRAIApplication
 
-async def get_app_instance(request: Request) -> "DevRAIApplication":
+logger = logging.getLogger(__name__)
+
+
+def get_app_instance(request: Request) -> "DevRAIApplication":
+    """Get the application instance from FastAPI app state."""
+    return request.app.state.app_instance
+
+
+async def get_current_user(authorization: str = Header(None)) -> UUID:
     """
-    Dependency to get the application instance from FastAPI's state.
-    This avoids circular imports by using dependency injection.
+    Get the current authenticated user from the Supabase JWT token.
+
+    Args:
+        authorization: The Authorization header containing the Bearer token
+
+    Returns:
+        UUID: The user's ID
+
+    Raises:
+        HTTPException: If authentication fails
     """
-    return request.app.state.app_instance
+    if not authorization:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing authorization header",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    if not authorization.startswith("Bearer "):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid authorization header format. Expected 'Bearer <token>'",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    token = authorization.replace("Bearer ", "")
+
+    try:
+        supabase = get_supabase_client()
+        # Verify the token and get user
+        user_response = supabase.auth.get_user(token)
+
+        if not user_response or not user_response.user:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Invalid or expired token",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+
+        return UUID(user_response.user.id)
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception("Authentication error")
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Authentication failed",
+            headers={"WWW-Authenticate": "Bearer"},
+        ) from e