Merge pull request #44 from Tanzania-AI-Community/fredy/development/onboarding-flow

Fredy/development/onboarding flow

Author: jurmy24

.env.template

@@ -18,6 +18,11 @@ WHATSAPP_VERIFY_TOKEN=
 # 60 day access token to use META's API for the Twiga bot
 WHATSAPP_API_TOKEN=
 
+# Whatsapp number for testing purposes
+RECIPIENT_WAID=
+
+DAILY_MESSAGE_LIMIT=100
+
 """OpenAI API credentials and other environment variables"""
 
 # OpenAI

.gitignore

@@ -173,4 +173,6 @@ cython_debug/
 .DS_Store
 
 # Ignore vector store
-twiga_vector_store/
+twiga_vector_store/
+private.pem
+public.pem

app/config.py

@@ -21,9 +21,17 @@ class Settings(BaseSettings):
     meta_api_version: str
     meta_app_id: str
     meta_app_secret: SecretStr
+    # WhatsApp settings
     whatsapp_cloud_number_id: str
     whatsapp_verify_token: SecretStr
     whatsapp_api_token: SecretStr
+    whatsapp_business_public_key: SecretStr
+    whatsapp_business_private_key: SecretStr
+    whatsapp_business_private_key_password: SecretStr
+    # Flows settings
+    personal_and_school_info_flow_id: str
+    subject_class_info_flow_id: str
+    flow_token_encryption_key: SecretStr
     # Rate limit settings
     daily_message_limit: int
     # Database settings

app/database/db.py

@@ -1,6 +1,7 @@
-from sqlmodel import SQLModel, Field, select
+from sqlmodel import SQLModel, Field, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 import logging
+from datetime import datetime
 
 from app.database.models import *
 from app.database.engine import db_engine
@@ -31,6 +32,11 @@ class UserQueryError(UserDatabaseError):
     """Raised when user query fails"""
 
     pass
+  
+class UserUpdateError(UserDatabaseError):
+    """Raised when user update fails"""
+
+    pass
 
 
 async def get_or_create_user(wa_id: str, name: Optional[str] = None) -> User:
@@ -99,7 +105,76 @@ async def get_user_by_waid(wa_id: str) -> Optional[User]:
             logger.error(f"Failed to query user {wa_id}: {str(e)}")
             raise UserQueryError(f"Failed to query user: {str(e)}")
 
+# TODO: rename this function
+async def update_user_by_waid(user: User) -> User:
+    """
+    Update any information about an existing user and return the updated user.
+    """
+    if user is None:
+        logger.error("Cannot update user: user object is None")
+        raise UserUpdateError("Cannot update user: user object is None")
+
+    # Convert the User object to a dictionary
+    user_data = user.__dict__.copy()
+
+    logger.debug(f"Updating user {user_data}")
+
+    # Remove the _sa_instance_state attribute
+    user_data.pop("_sa_instance_state", None)
+
+    # Extract the wa_id
+    wa_id = user_data.pop("wa_id")
+
+    # Remove the id attribute
+    user_data.pop("wa_id", None)
+    user_data.pop("id", None)
+
+    # Handle the birthday field if necessary
+    if "birthday" in user_data and isinstance(user_data["birthday"], str):
+        user_data["birthday"] = datetime.strptime(
+            user_data["birthday"], "%Y-%m-%d"
+        ).date()
+
+    async with AsyncSession(db_engine) as session:
+        try:
+            statement = update(User).where(User.wa_id == wa_id).values(**user_data)
+            await session.execute(statement)
+            await session.commit()
+
+            # Fetch the updated user
+            result = await session.execute(select(User).filter_by(wa_id=wa_id))
+            updated_user = result.scalar_one_or_none()
+
+            logger.info(f"Updated user {wa_id} with {user_data}")
+            return updated_user
+        except Exception as e:
+            await session.rollback()
+            logger.error(f"Failed to update user {wa_id}: {str(e)}")
+            raise UserUpdateError(f"Failed to update user: {str(e)}")
+ 
+
+# TODO: This should be replaced with get_user_by_waid or the get_or_create_user function
+async def get_user_data(wa_id: str) -> dict:
+    """
+    Retrieve user data based on wa_id.
+    """
+    async with AsyncSession(db_engine) as session:
+        try:
+            statement = select(User).where(User.wa_id == wa_id)
+            result = await session.execute(statement)
+            user = result.scalar_one_or_none()
+            if user:
+                user_data = user.model_dump()
+                logger.info(f"Retrieved user data for {wa_id}: {user_data}")
+                return user_data
+            else:
+                logger.warning(f"No user found with wa_id {wa_id}")
+                return None
+        except Exception as e:
+            logger.error(f"Failed to query user {wa_id}: {str(e)}")
+            raise UserQueryError(f"Failed to query user: {str(e)}")
 
+            
 async def get_user_message_history(
     user_id: int, limit: int = 10
 ) -> Optional[List[Message]]:
@@ -202,4 +277,4 @@ async def vector_search(query: str, n_results: int, where: dict) -> List[Chunk]:
             return result.scalars().all()
         except Exception as e:
             logger.error(f"Failed to search for knowledge: {str(e)}")
-            raise Exception(f"Failed to search for knowledge: {str(e)}")
+            raise Exception(f"Failed to search for knowledge: {str(e)}")

app/database/models.py

@@ -57,7 +57,6 @@ class GradeLevel(str, Enum):
 class OnboardingState(str, Enum):
     new = "new"
     personal_info_submitted = "personal_info_submitted"
-    class_subject_info_submitted = "class_subject_info_submitted"
     completed = "completed"
 
 
@@ -106,8 +105,10 @@ class User(SQLModel, table=True):
     role: str = Field(default=Role.teacher, max_length=20)
     class_info: Optional[dict] = Field(default=None, sa_type=JSON)
     school_name: Optional[str] = Field(default=None, max_length=100)
+    school_location: Optional[str] = Field(default=None, max_length=100)
     birthday: Optional[date] = Field(default=None, sa_type=Date)
     region: Optional[str] = Field(default=None, max_length=50)
+    location: Optional[str] = Field(default=None, max_length=100)
     last_message_at: Optional[datetime] = Field(
         sa_type=DateTime(timezone=True)
     )  # user.last_message_at = datetime.now(timezone.utc) (this is how to set it when updating later)

app/main.py

@@ -1,12 +1,14 @@
-from fastapi import FastAPI, Request, Depends
+from fastapi import FastAPI, HTTPException, Request, Depends
 from fastapi.responses import JSONResponse
 import logging
 from contextlib import asynccontextmanager
 
 
 from app.security import signature_required
+from app.security import flows_signature_required
 from app.services.whatsapp_service import whatsapp_client
 from app.services.messaging_service import handle_request
+from app.services.flow_service import flow_client
 from app.database.engine import db_engine, init_db
 
 logger = logging.getLogger(__name__)
@@ -40,3 +42,45 @@ async def webhook_get(request: Request) -> JSONResponse:
 async def webhook_post(request: Request) -> JSONResponse:
     logger.debug("webhook_post is being called")
     return await handle_request(request)
+
+
+@app.post("/flows", dependencies=[Depends(flows_signature_required)])
+async def handle_flows_webhook(request: Request) -> JSONResponse:
+    try:
+        body = await request.json()
+        logger.debug(f"Received webhook: {body}")
+        return await flow_client.handle_flow_webhook(body)
+    except Exception as e:
+        logger.error(f"Error handling webhook: {e}")
+        raise HTTPException(status_code=500, detail="Internal Server Error")
+
+
+# use this when testing flows locally, the returned token will be the flow_token
+@app.post("/encrypt_flow_token")
+async def handle_encrypt_flow_token(request: Request) -> JSONResponse:
+    try:
+        body = await request.json()
+        logger.debug(f"Received request to encrypt flow token: {body}")
+        wa_id = body.get("wa_id")
+        flow_id = body.get("flow_id")
+
+        logger.info(f"Encrypting flow token for wa_id {wa_id} and flow_id {flow_id}")
+
+        return await flow_client.encrypt_flow_token(wa_id, flow_id)
+    except Exception as e:
+        logger.error(f"Error encrypting flow token: {e}")
+        raise HTTPException(status_code=500, detail="Internal Server Error")
+
+
+# decrypt_flow_token
+@app.post("/decrypt_flow_token")
+async def handle_decrypt_flow_token(request: Request) -> JSONResponse:
+    try:
+        body = await request.json()
+        logger.debug(f"Received request to decrypt flow token: {body}")
+        encrypted_flow_token = body.get("encrypted-flow-token")
+
+        return await flow_client.decrypt_flow_token(encrypted_flow_token)
+    except Exception as e:
+        logger.error(f"Error decrypting flow token: {e}")
+        raise HTTPException(status_code=500, detail="Internal Server Error")

app/models/message_models.py

@@ -1,6 +1,8 @@
-from pydantic import BaseModel, constr, Field
-from typing import List, Dict, Literal, Optional
+from pydantic import BaseModel, constr, Field, root_validator
+from typing import List, Dict, Literal, Optional, Union
 import json
+from pydantic import BaseModel, constr, Field, model_validator
+from typing import List, Dict, Literal, Optional, Union
 
 
 class TextObject(BaseModel):
@@ -68,25 +70,80 @@ class TextMessage(BaseModel):
 
 
 """
-Main model for interactive messages
+Main model for template messages
 """
 
 
-class InteractiveMessage(BaseModel):
+class TemplateMessage(BaseModel):
+    messaging_product: Literal["whatsapp"] = "whatsapp"
+    to: str
+    type: Literal["template"] = "template"
+    template: Dict[Literal["name", "language"], str]
+
+
+"""
+Models for flow interactive messages
+"""
+
+
+class FlowActionPayload(BaseModel):
+    screen: str
+    data: Dict[str, Union[str, int]]
+
+
+class FlowActionPayload(BaseModel):
+    screen: str
+    data: Dict[str, Union[str, int]]
+
+
+class FlowParameters(BaseModel):
+    flow_message_version: str
+    flow_token: str
+    flow_name: Optional[str] = None
+    flow_id: Optional[str] = None
+    flow_cta: str
+    flow_action: str
+    flow_action_payload: FlowActionPayload
+
+    @model_validator(mode="before")
+    def check_flow_name_or_id(cls, values):
+        flow_name, flow_id = values.get("flow_name"), values.get("flow_id")
+        if not flow_name and not flow_id:
+            raise ValueError("Either flow_name or flow_id must be provided")
+        if flow_name and flow_id:
+            raise ValueError("Only one of flow_name or flow_id should be provided")
+        return values
+
+
+class FlowAction(BaseModel):
+    name: Literal["flow"]
+    parameters: FlowParameters
+
+
+class FlowInteractive(BaseModel):
+    type: Literal["flow"] = "flow"
+    header: TextObject
+    body: TextObject
+    footer: TextObject
+    action: FlowAction
+
+
+class FlowInteractiveMessage(BaseModel):
     messaging_product: Literal["whatsapp"] = "whatsapp"
     recipient_type: Literal["individual"] = "individual"
     to: str
     type: Literal["interactive"] = "interactive"
-    interactive: InteractiveButton | InteractiveList
+    interactive: FlowInteractive
 
 
 """
-Main model for template messaged
+Main model for interactive messages
 """
 
 
-class TemplateMessage(BaseModel):
+class InteractiveMessage(BaseModel):
     messaging_product: Literal["whatsapp"] = "whatsapp"
+    recipient_type: Literal["individual"] = "individual"
     to: str
-    type: Literal["template"] = "template"
-    template: Dict[Literal["name", "language"], str]
+    type: Literal["interactive"] = "interactive"
+    interactive: Union[InteractiveButton, InteractiveList, FlowInteractive]

app/security.py

@@ -32,3 +32,14 @@ async def signature_required(request: Request) -> None:
     if not validate_signature(payload.decode("utf-8"), signature):
         logger.error("Signature verification failed!")
         raise HTTPException(status_code=403, detail="Invalid signature")
+
+
+# Dependency to ensure that incoming requests to our flows webhook are signed with the correct signature.
+async def flows_signature_required(request: Request) -> None:
+    signature = request.headers.get("X-Hub-Signature-256", "")[7:]  # Removing 'sha256='
+    payload = await request.body()
+
+    if not validate_signature(payload.decode("utf-8"), signature):
+        logger.error("Business signature verification failed!")
+        # NOTE : We are using a custom status code here, 432. And user will see A generic error on the client.
+        raise HTTPException(status_code=432, detail="Invalid business signature")