multitenantsaasplatformwithexternalintegrations-7f7k45/mock_services/communication_service.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
from datetime import datetime
import uuid
from enum import Enum

app = FastAPI(title="Mock Communication Service", version="1.0.0")

class MessageType(str, Enum):
    EMAIL = "email"
    SMS = "sms"
    PUSH_NOTIFICATION = "push_notification"

class MessageStatus(str, Enum):
    PENDING = "pending"
    SENT = "sent"
    DELIVERED = "delivered"
    BOUNCED = "bounced"
    FAILED = "failed"

class EmailMessage(BaseModel):
    id: Optional[str] = None
    organization_id: int
    to: str
    from_email: str
    subject: str
    body: str
    status: MessageStatus = MessageStatus.PENDING
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

class SMSMessage(BaseModel):
    id: Optional[str] = None
    organization_id: int
    to: str
    message: str
    status: MessageStatus = MessageStatus.PENDING
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

class PushNotification(BaseModel):
    id: Optional[str] = None
    organization_id: int
    user_id: str
    title: str
    body: str
    status: MessageStatus = MessageStatus.PENDING
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

# Mock data storage
messages_db: Dict[str, Dict[str, Any]] = {}
organization_communications: Dict[int, List[str]] = {}

@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "communication", "timestamp": datetime.utcnow()}

@app.post("/emails", response_model=EmailMessage)
async def send_email(email: EmailMessage):
    message_id = str(uuid.uuid4())
    email.id = message_id
    email.created_at = datetime.utcnow()
    
    # Simulate email sending
    import random
    delivery_rate = 0.92  # 92% delivery rate
    
    if random.random() < delivery_rate:
        email.status = MessageStatus.DELIVERED
        event_type = "email.delivered"
    else:
        email.status = MessageStatus.BOUNCED
        event_type = "email.bounced"
    
    email.updated_at = datetime.utcnow()
    messages_db[message_id] = {**email.dict(), "type": MessageType.EMAIL}
    
    # Add to organization communications
    if email.organization_id not in organization_communications:
        organization_communications[email.organization_id] = []
    organization_communications[email.organization_id].append(message_id)
    
    # Send webhook (simulated)
    await send_webhook(event_type, email.dict())
    
    return email

@app.post("/sms", response_model=SMSMessage)
async def send_sms(sms: SMSMessage):
    message_id = str(uuid.uuid4())
    sms.id = message_id
    sms.created_at = datetime.utcnow()
    
    # Simulate SMS sending
    import random
    delivery_rate = 0.95  # 95% delivery rate
    
    if random.random() < delivery_rate:
        sms.status = MessageStatus.DELIVERED
        event_type = "sms.delivered"
    else:
        sms.status = MessageStatus.FAILED
        event_type = "sms.failed"
    
    sms.updated_at = datetime.utcnow()
    messages_db[message_id] = {**sms.dict(), "type": MessageType.SMS}
    
    # Add to organization communications
    if sms.organization_id not in organization_communications:
        organization_communications[sms.organization_id] = []
    organization_communications[sms.organization_id].append(message_id)
    
    # Send webhook (simulated)
    await send_webhook(event_type, sms.dict())
    
    return sms

@app.post("/notifications", response_model=PushNotification)
async def send_notification(notification: PushNotification):
    message_id = str(uuid.uuid4())
    notification.id = message_id
    notification.created_at = datetime.utcnow()
    
    # Simulate push notification sending
    import random
    delivery_rate = 0.88  # 88% delivery rate
    
    if random.random() < delivery_rate:
        notification.status = MessageStatus.DELIVERED
        event_type = "notification.delivered"
    else:
        notification.status = MessageStatus.FAILED
        event_type = "notification.failed"
    
    notification.updated_at = datetime.utcnow()
    messages_db[message_id] = {**notification.dict(), "type": MessageType.PUSH_NOTIFICATION}
    
    # Add to organization communications
    if notification.organization_id not in organization_communications:
        organization_communications[notification.organization_id] = []
    organization_communications[notification.organization_id].append(message_id)
    
    # Send webhook (simulated)
    await send_webhook(event_type, notification.dict())
    
    return notification

@app.get("/messages/{message_id}/status")
async def get_delivery_status(message_id: str):
    if message_id not in messages_db:
        raise HTTPException(status_code=404, detail="Message not found")
    
    message = messages_db[message_id]
    return {
        "message_id": message_id,
        "status": message["status"],
        "type": message["type"],
        "updated_at": message["updated_at"]
    }

@app.get("/organizations/{organization_id}/communications")
async def get_communication_history(organization_id: int):
    if organization_id not in organization_communications:
        return {"messages": [], "total_messages": 0}
    
    org_messages = []
    for message_id in organization_communications[organization_id]:
        if message_id in messages_db:
            org_messages.append(messages_db[message_id])
    
    # Group by type
    emails = [m for m in org_messages if m.get("type") == MessageType.EMAIL]
    sms_messages = [m for m in org_messages if m.get("type") == MessageType.SMS]
    notifications = [m for m in org_messages if m.get("type") == MessageType.PUSH_NOTIFICATION]
    
    return {
        "emails": emails,
        "sms_messages": sms_messages,
        "notifications": notifications,
        "total_messages": len(org_messages),
        "summary": {
            "emails_sent": len(emails),
            "sms_sent": len(sms_messages),
            "notifications_sent": len(notifications)
        }
    }

async def send_webhook(event_type: str, data: Dict[str, Any]):
    """Simulate sending webhook to main service"""
    webhook_data = {
        "event_id": str(uuid.uuid4()),
        "event_type": event_type,
        "timestamp": datetime.utcnow().isoformat(),
        "data": data
    }
    
    # In a real implementation, this would send HTTP request to main service
    print(f"Webhook sent: {event_type} - {webhook_data['event_id']}")
    
    return webhook_data

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8003)