892 lines
32 KiB
Python
892 lines
32 KiB
Python
"""
|
|
Notification Business Operations API
|
|
Handles sending, marking read, scheduling, retry, and SSE streaming
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import structlog
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Depends, HTTPException, status, Query, Path, Request, BackgroundTasks
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from typing import List, Optional, Dict, Any
|
|
from uuid import UUID
|
|
from sse_starlette.sse import EventSourceResponse
|
|
|
|
from app.schemas.notifications import (
|
|
NotificationResponse, NotificationType, NotificationStatus, NotificationPriority
|
|
)
|
|
from app.services.notification_service import EnhancedNotificationService
|
|
from app.models.notifications import NotificationType as ModelNotificationType
|
|
from app.models import AuditLog
|
|
from app.core.database import get_db
|
|
from shared.auth.decorators import get_current_user_dep, get_current_user
|
|
from shared.auth.access_control import require_user_role, admin_role_required, service_only_access
|
|
from shared.routing.route_builder import RouteBuilder
|
|
from shared.database.base import create_database_manager
|
|
from shared.monitoring.metrics import track_endpoint_metrics
|
|
from shared.security import create_audit_logger, AuditSeverity, AuditAction
|
|
|
|
logger = structlog.get_logger()
|
|
audit_logger = create_audit_logger("notification-service", AuditLog)
|
|
router = APIRouter()
|
|
route_builder = RouteBuilder('notifications')
|
|
|
|
# Dependency injection for enhanced notification service
|
|
def get_enhanced_notification_service():
|
|
database_manager = create_database_manager()
|
|
return EnhancedNotificationService(database_manager)
|
|
|
|
|
|
# ============================================================================
|
|
# BUSINESS OPERATIONS - Send, Schedule, Retry, Mark Read
|
|
# ============================================================================
|
|
|
|
@router.post(
|
|
route_builder.build_base_route("send", include_tenant_prefix=False),
|
|
response_model=NotificationResponse,
|
|
status_code=201
|
|
)
|
|
@require_user_role(["member", "admin", "owner"])
|
|
@track_endpoint_metrics("notification_send")
|
|
async def send_notification(
|
|
notification_data: Dict[str, Any],
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
notification_service: EnhancedNotificationService = Depends(get_enhanced_notification_service)
|
|
):
|
|
"""Send a single notification with enhanced validation and features"""
|
|
|
|
try:
|
|
# Check permissions for broadcast notifications (Admin+ only)
|
|
if notification_data.get("broadcast", False):
|
|
user_role = current_user.get("role", "").lower()
|
|
if user_role not in ["admin", "owner"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only admins and owners can send broadcast notifications"
|
|
)
|
|
|
|
# Log HIGH severity audit event for broadcast notifications
|
|
try:
|
|
# Note: db session would need to be passed as dependency for full audit logging
|
|
logger.info("Broadcast notification initiated",
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user["user_id"],
|
|
notification_type=notification_data.get("type"),
|
|
severity="HIGH")
|
|
except Exception as audit_error:
|
|
logger.warning("Failed to log audit event", error=str(audit_error))
|
|
|
|
# Validate required fields
|
|
if not notification_data.get("message"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Message is required"
|
|
)
|
|
|
|
if not notification_data.get("type"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Notification type is required"
|
|
)
|
|
|
|
# Convert string type to enum
|
|
try:
|
|
notification_type = ModelNotificationType(notification_data["type"])
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Invalid notification type: {notification_data['type']}"
|
|
)
|
|
|
|
# Convert priority if provided
|
|
priority = NotificationPriority.NORMAL
|
|
if "priority" in notification_data:
|
|
try:
|
|
priority = NotificationPriority(notification_data["priority"])
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Invalid priority: {notification_data['priority']}"
|
|
)
|
|
|
|
# Create notification using enhanced service
|
|
notification = await notification_service.create_notification(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
sender_id=current_user["user_id"],
|
|
notification_type=notification_type,
|
|
message=notification_data["message"],
|
|
recipient_id=notification_data.get("recipient_id"),
|
|
recipient_email=notification_data.get("recipient_email"),
|
|
recipient_phone=notification_data.get("recipient_phone"),
|
|
subject=notification_data.get("subject"),
|
|
html_content=notification_data.get("html_content"),
|
|
template_key=notification_data.get("template_key"),
|
|
template_data=notification_data.get("template_data"),
|
|
priority=priority,
|
|
scheduled_at=notification_data.get("scheduled_at"),
|
|
broadcast=notification_data.get("broadcast", False)
|
|
)
|
|
|
|
logger.info("Notification sent successfully",
|
|
notification_id=notification.id,
|
|
tenant_id=current_user.get("tenant_id"),
|
|
type=notification_type.value,
|
|
priority=priority.value)
|
|
|
|
return NotificationResponse.from_orm(notification)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to send notification",
|
|
tenant_id=current_user.get("tenant_id"),
|
|
sender_id=current_user["user_id"],
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to send notification"
|
|
)
|
|
|
|
|
|
@router.patch(
|
|
route_builder.build_operations_route("{notification_id}/read", include_tenant_prefix=False)
|
|
)
|
|
@require_user_role(["viewer", "member", "admin", "owner"])
|
|
@track_endpoint_metrics("notification_mark_read")
|
|
async def mark_notification_read(
|
|
notification_id: UUID = Path(..., description="Notification ID"),
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
notification_service: EnhancedNotificationService = Depends(get_enhanced_notification_service)
|
|
):
|
|
"""Mark a notification as read"""
|
|
|
|
try:
|
|
success = await notification_service.mark_notification_as_read(
|
|
str(notification_id),
|
|
current_user["user_id"]
|
|
)
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Notification not found or access denied"
|
|
)
|
|
|
|
return {"success": True, "message": "Notification marked as read"}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to mark notification as read",
|
|
notification_id=str(notification_id),
|
|
user_id=current_user["user_id"],
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to mark notification as read"
|
|
)
|
|
|
|
|
|
@router.patch(
|
|
route_builder.build_base_route("mark-multiple-read", include_tenant_prefix=False)
|
|
)
|
|
@require_user_role(["viewer", "member", "admin", "owner"])
|
|
@track_endpoint_metrics("notification_mark_multiple_read")
|
|
async def mark_multiple_notifications_read(
|
|
request_data: Dict[str, Any],
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
notification_service: EnhancedNotificationService = Depends(get_enhanced_notification_service)
|
|
):
|
|
"""Mark multiple notifications as read with batch processing"""
|
|
|
|
try:
|
|
notification_ids = request_data.get("notification_ids")
|
|
tenant_id = request_data.get("tenant_id")
|
|
|
|
if not notification_ids and not tenant_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Either notification_ids or tenant_id must be provided"
|
|
)
|
|
|
|
# Convert UUID strings to strings if needed
|
|
if notification_ids:
|
|
notification_ids = [str(nid) for nid in notification_ids]
|
|
|
|
marked_count = await notification_service.mark_multiple_as_read(
|
|
user_id=current_user["user_id"],
|
|
notification_ids=notification_ids,
|
|
tenant_id=tenant_id
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"marked_count": marked_count,
|
|
"message": f"Marked {marked_count} notifications as read"
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to mark multiple notifications as read",
|
|
user_id=current_user["user_id"],
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to mark notifications as read"
|
|
)
|
|
|
|
|
|
@router.patch(
|
|
route_builder.build_operations_route("{notification_id}/status", include_tenant_prefix=False)
|
|
)
|
|
@require_user_role(["admin", "owner"])
|
|
@track_endpoint_metrics("notification_update_status")
|
|
async def update_notification_status(
|
|
notification_id: UUID = Path(..., description="Notification ID"),
|
|
status_data: Dict[str, Any] = ...,
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
notification_service: EnhancedNotificationService = Depends(get_enhanced_notification_service)
|
|
):
|
|
"""Update notification status (admin/system only)"""
|
|
|
|
# Only system users or admins can update notification status
|
|
if (current_user.get("type") != "service" and
|
|
current_user.get("role") not in ["admin", "system"]):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only system services or admins can update notification status"
|
|
)
|
|
|
|
try:
|
|
new_status = status_data.get("status")
|
|
if not new_status:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Status is required"
|
|
)
|
|
|
|
# Convert string status to enum
|
|
try:
|
|
from app.models.notifications import NotificationStatus as ModelStatus
|
|
model_status = ModelStatus(new_status)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Invalid status: {new_status}"
|
|
)
|
|
|
|
updated_notification = await notification_service.update_notification_status(
|
|
notification_id=str(notification_id),
|
|
new_status=model_status,
|
|
error_message=status_data.get("error_message"),
|
|
provider_message_id=status_data.get("provider_message_id"),
|
|
metadata=status_data.get("metadata"),
|
|
response_time_ms=status_data.get("response_time_ms"),
|
|
provider=status_data.get("provider")
|
|
)
|
|
|
|
if not updated_notification:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Notification not found"
|
|
)
|
|
|
|
return NotificationResponse.from_orm(updated_notification)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to update notification status",
|
|
notification_id=str(notification_id),
|
|
status=status_data.get("status"),
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to update notification status"
|
|
)
|
|
|
|
|
|
@router.get(
|
|
route_builder.build_base_route("pending", include_tenant_prefix=False),
|
|
response_model=List[NotificationResponse]
|
|
)
|
|
@require_user_role(["admin", "owner"])
|
|
@track_endpoint_metrics("notification_get_pending")
|
|
async def get_pending_notifications(
|
|
limit: int = Query(100, ge=1, le=1000, description="Maximum number of notifications"),
|
|
notification_type: Optional[NotificationType] = Query(None, description="Filter by type"),
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
notification_service: EnhancedNotificationService = Depends(get_enhanced_notification_service)
|
|
):
|
|
"""Get pending notifications for processing (system/admin only)"""
|
|
|
|
if (current_user.get("type") != "service" and
|
|
current_user.get("role") not in ["admin", "system"]):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only system services or admins can access pending notifications"
|
|
)
|
|
|
|
try:
|
|
model_notification_type = None
|
|
if notification_type:
|
|
try:
|
|
model_notification_type = ModelNotificationType(notification_type.value)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Invalid notification type: {notification_type.value}"
|
|
)
|
|
|
|
notifications = await notification_service.get_pending_notifications(
|
|
limit=limit,
|
|
notification_type=model_notification_type
|
|
)
|
|
|
|
return [NotificationResponse.from_orm(notification) for notification in notifications]
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to get pending notifications",
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to get pending notifications"
|
|
)
|
|
|
|
|
|
@router.post(
|
|
route_builder.build_operations_route("{notification_id}/schedule", include_tenant_prefix=False)
|
|
)
|
|
@require_user_role(["member", "admin", "owner"])
|
|
@track_endpoint_metrics("notification_schedule")
|
|
async def schedule_notification(
|
|
notification_id: UUID = Path(..., description="Notification ID"),
|
|
schedule_data: Dict[str, Any] = ...,
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
notification_service: EnhancedNotificationService = Depends(get_enhanced_notification_service)
|
|
):
|
|
"""Schedule a notification for future delivery"""
|
|
|
|
try:
|
|
scheduled_at = schedule_data.get("scheduled_at")
|
|
if not scheduled_at:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="scheduled_at is required"
|
|
)
|
|
|
|
# Parse datetime if it's a string
|
|
if isinstance(scheduled_at, str):
|
|
try:
|
|
scheduled_at = datetime.fromisoformat(scheduled_at.replace('Z', '+00:00'))
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Invalid datetime format. Use ISO format."
|
|
)
|
|
|
|
# Check that the scheduled time is in the future
|
|
if scheduled_at <= datetime.utcnow():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Scheduled time must be in the future"
|
|
)
|
|
|
|
success = await notification_service.schedule_notification(
|
|
str(notification_id),
|
|
scheduled_at
|
|
)
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Notification not found or cannot be scheduled"
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Notification scheduled successfully",
|
|
"scheduled_at": scheduled_at.isoformat()
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to schedule notification",
|
|
notification_id=str(notification_id),
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to schedule notification"
|
|
)
|
|
|
|
|
|
@router.post(
|
|
route_builder.build_operations_route("{notification_id}/cancel", include_tenant_prefix=False)
|
|
)
|
|
@require_user_role(["member", "admin", "owner"])
|
|
@track_endpoint_metrics("notification_cancel")
|
|
async def cancel_notification(
|
|
notification_id: UUID = Path(..., description="Notification ID"),
|
|
cancel_data: Optional[Dict[str, Any]] = None,
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
notification_service: EnhancedNotificationService = Depends(get_enhanced_notification_service)
|
|
):
|
|
"""Cancel a pending notification"""
|
|
|
|
try:
|
|
reason = None
|
|
if cancel_data:
|
|
reason = cancel_data.get("reason", "Cancelled by user")
|
|
else:
|
|
reason = "Cancelled by user"
|
|
|
|
success = await notification_service.cancel_notification(
|
|
str(notification_id),
|
|
reason
|
|
)
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Notification not found or cannot be cancelled"
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Notification cancelled successfully",
|
|
"reason": reason
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to cancel notification",
|
|
notification_id=str(notification_id),
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to cancel notification"
|
|
)
|
|
|
|
|
|
@router.post(
|
|
route_builder.build_operations_route("{notification_id}/retry", include_tenant_prefix=False)
|
|
)
|
|
@require_user_role(["admin", "owner"])
|
|
@track_endpoint_metrics("notification_retry")
|
|
async def retry_failed_notification(
|
|
notification_id: UUID = Path(..., description="Notification ID"),
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
notification_service: EnhancedNotificationService = Depends(get_enhanced_notification_service)
|
|
):
|
|
"""Retry a failed notification (admin only)"""
|
|
|
|
# Only admins can retry notifications
|
|
if current_user.get("role") not in ["admin", "system"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only admins can retry failed notifications"
|
|
)
|
|
|
|
try:
|
|
success = await notification_service.retry_failed_notification(str(notification_id))
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Notification not found, not failed, or max retries exceeded"
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Notification queued for retry"
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Failed to retry notification",
|
|
notification_id=str(notification_id),
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to retry notification"
|
|
)
|
|
|
|
|
|
@router.get(
|
|
route_builder.build_base_route("statistics", include_tenant_prefix=False)
|
|
)
|
|
@admin_role_required
|
|
@track_endpoint_metrics("notification_get_statistics")
|
|
async def get_notification_statistics(
|
|
tenant_id: Optional[str] = Query(None, description="Filter by tenant ID"),
|
|
days_back: int = Query(30, ge=1, le=365, description="Number of days to look back"),
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
notification_service: EnhancedNotificationService = Depends(get_enhanced_notification_service)
|
|
):
|
|
"""Get comprehensive notification statistics"""
|
|
|
|
try:
|
|
stats = await notification_service.get_notification_statistics(
|
|
tenant_id=tenant_id,
|
|
days_back=days_back
|
|
)
|
|
|
|
return stats
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get notification statistics",
|
|
tenant_id=tenant_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to get notification statistics"
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# SSE STREAMING ENDPOINTS
|
|
# ============================================================================
|
|
|
|
@router.get(route_builder.build_operations_route("sse/stream/{tenant_id}", include_tenant_prefix=False))
|
|
async def stream_notifications(
|
|
request: Request,
|
|
background_tasks: BackgroundTasks,
|
|
tenant_id: str = Path(..., description="Tenant ID"),
|
|
token: Optional[str] = None
|
|
):
|
|
"""
|
|
SSE endpoint for real-time notification streaming
|
|
Supports alerts and recommendations through unified stream
|
|
"""
|
|
|
|
# Validate token and get user
|
|
current_user = None
|
|
if token:
|
|
try:
|
|
from shared.auth.jwt_handler import JWTHandler
|
|
from app.core.config import settings
|
|
|
|
jwt_handler = JWTHandler(settings.JWT_SECRET_KEY)
|
|
payload = jwt_handler.decode_access_token(token)
|
|
|
|
if not payload:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired token"
|
|
)
|
|
|
|
current_user = payload
|
|
|
|
except Exception as e:
|
|
logger.warning("Token validation failed", error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired token"
|
|
)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Authentication token required"
|
|
)
|
|
|
|
# Validate tenant access
|
|
user_tenant_id = current_user.get('tenant_id')
|
|
if user_tenant_id and str(user_tenant_id) != str(tenant_id):
|
|
logger.warning("Tenant access denied",
|
|
user_tenant_id=user_tenant_id,
|
|
requested_tenant_id=tenant_id)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied to this tenant's notifications"
|
|
)
|
|
|
|
# Get SSE service from app state
|
|
sse_service = getattr(request.app.state, 'sse_service', None)
|
|
if not sse_service:
|
|
raise HTTPException(500, "SSE service not available")
|
|
|
|
async def event_generator():
|
|
"""Generate SSE events for the client"""
|
|
client_queue = asyncio.Queue(maxsize=100) # Limit queue size
|
|
|
|
try:
|
|
# Register client
|
|
await sse_service.add_client(tenant_id, client_queue)
|
|
|
|
logger.info("SSE client connected",
|
|
tenant_id=tenant_id,
|
|
user_id=getattr(current_user, 'id', 'unknown'))
|
|
|
|
# Stream events
|
|
while True:
|
|
# Check if client disconnected
|
|
if await request.is_disconnected():
|
|
logger.info("SSE client disconnected", tenant_id=tenant_id)
|
|
break
|
|
|
|
try:
|
|
# Wait for events with timeout for keepalive
|
|
event = await asyncio.wait_for(
|
|
client_queue.get(),
|
|
timeout=30.0
|
|
)
|
|
|
|
yield event
|
|
|
|
except asyncio.TimeoutError:
|
|
# Send keepalive ping
|
|
yield {
|
|
"event": "ping",
|
|
"data": json.dumps({
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"status": "keepalive"
|
|
}),
|
|
"id": f"ping_{int(datetime.now().timestamp())}"
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Error in SSE event generator",
|
|
tenant_id=tenant_id,
|
|
error=str(e))
|
|
break
|
|
|
|
except Exception as e:
|
|
logger.error("SSE connection error",
|
|
tenant_id=tenant_id,
|
|
error=str(e))
|
|
finally:
|
|
# Clean up on disconnect
|
|
try:
|
|
await sse_service.remove_client(tenant_id, client_queue)
|
|
logger.info("SSE client cleanup completed", tenant_id=tenant_id)
|
|
except Exception as e:
|
|
logger.error("Error cleaning up SSE client",
|
|
tenant_id=tenant_id,
|
|
error=str(e))
|
|
|
|
return EventSourceResponse(
|
|
event_generator(),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no", # Disable nginx buffering
|
|
}
|
|
)
|
|
|
|
|
|
@router.post(route_builder.build_operations_route("sse/items/{item_id}/acknowledge", include_tenant_prefix=False))
|
|
async def acknowledge_item(
|
|
item_id: str,
|
|
current_user = Depends(get_current_user)
|
|
):
|
|
"""Acknowledge an alert or recommendation"""
|
|
try:
|
|
# This would update the database
|
|
# For now, just return success
|
|
|
|
logger.info("Item acknowledged",
|
|
item_id=item_id,
|
|
user_id=getattr(current_user, 'id', 'unknown'))
|
|
|
|
return {
|
|
"status": "success",
|
|
"item_id": item_id,
|
|
"acknowledged_by": getattr(current_user, 'id', 'unknown'),
|
|
"acknowledged_at": datetime.utcnow().isoformat()
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to acknowledge item", item_id=item_id, error=str(e))
|
|
raise HTTPException(500, "Failed to acknowledge item")
|
|
|
|
|
|
@router.post(route_builder.build_operations_route("sse/items/{item_id}/resolve", include_tenant_prefix=False))
|
|
async def resolve_item(
|
|
item_id: str,
|
|
current_user = Depends(get_current_user)
|
|
):
|
|
"""Resolve an alert or recommendation"""
|
|
try:
|
|
# This would update the database
|
|
# For now, just return success
|
|
|
|
logger.info("Item resolved",
|
|
item_id=item_id,
|
|
user_id=getattr(current_user, 'id', 'unknown'))
|
|
|
|
return {
|
|
"status": "success",
|
|
"item_id": item_id,
|
|
"resolved_by": getattr(current_user, 'id', 'unknown'),
|
|
"resolved_at": datetime.utcnow().isoformat()
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to resolve item", item_id=item_id, error=str(e))
|
|
raise HTTPException(500, "Failed to resolve item")
|
|
|
|
|
|
@router.get(route_builder.build_operations_route("sse/status/{tenant_id}", include_tenant_prefix=False))
|
|
async def get_sse_status(
|
|
request: Request,
|
|
tenant_id: str = Path(..., description="Tenant ID"),
|
|
current_user = Depends(get_current_user)
|
|
):
|
|
"""Get SSE connection status for a tenant"""
|
|
|
|
# Verify user has access to this tenant
|
|
if not hasattr(current_user, 'has_access_to_tenant') or not current_user.has_access_to_tenant(tenant_id):
|
|
raise HTTPException(403, "Access denied to this tenant")
|
|
|
|
try:
|
|
# Get SSE service from app state
|
|
sse_service = getattr(request.app.state, 'sse_service', None)
|
|
if not sse_service:
|
|
return {"status": "unavailable", "message": "SSE service not initialized"}
|
|
|
|
metrics = sse_service.get_metrics()
|
|
tenant_connections = len(sse_service.active_connections.get(tenant_id, set()))
|
|
|
|
return {
|
|
"status": "available",
|
|
"tenant_id": tenant_id,
|
|
"connections": tenant_connections,
|
|
"total_connections": metrics["total_connections"],
|
|
"active_tenants": metrics["active_tenants"]
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get SSE status", tenant_id=tenant_id, error=str(e))
|
|
raise HTTPException(500, "Failed to get SSE status")
|
|
|
|
|
|
# ============================================================================
|
|
# Tenant Data Deletion Operations (Internal Service Only)
|
|
# ============================================================================
|
|
|
|
@router.delete(
|
|
route_builder.build_base_route("tenant/{tenant_id}", include_tenant_prefix=False),
|
|
response_model=dict
|
|
)
|
|
@service_only_access
|
|
async def delete_tenant_data(
|
|
tenant_id: str = Path(..., description="Tenant ID to delete data for"),
|
|
current_user: dict = Depends(get_current_user_dep)
|
|
):
|
|
"""
|
|
Delete all notification data for a tenant (Internal service only)
|
|
|
|
This endpoint is called by the orchestrator during tenant deletion.
|
|
It permanently deletes all notification-related data including:
|
|
- Notifications (all types and statuses)
|
|
- Notification logs
|
|
- User notification preferences
|
|
- Tenant-specific notification templates
|
|
- Audit logs
|
|
|
|
**NOTE**: System templates (is_system=True) are preserved
|
|
|
|
**WARNING**: This operation is irreversible!
|
|
|
|
Returns:
|
|
Deletion summary with counts of deleted records
|
|
"""
|
|
from app.services.tenant_deletion_service import NotificationTenantDeletionService
|
|
from app.core.config import settings
|
|
|
|
try:
|
|
logger.info("notification.tenant_deletion.api_called", tenant_id=tenant_id)
|
|
|
|
db_manager = create_database_manager(settings.DATABASE_URL, "notification")
|
|
|
|
async with db_manager.get_session() as session:
|
|
deletion_service = NotificationTenantDeletionService(session)
|
|
result = await deletion_service.safe_delete_tenant_data(tenant_id)
|
|
|
|
if not result.success:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Tenant data deletion failed: {', '.join(result.errors)}"
|
|
)
|
|
|
|
return {
|
|
"message": "Tenant data deletion completed successfully",
|
|
"note": "System templates have been preserved",
|
|
"summary": result.to_dict()
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("notification.tenant_deletion.api_error",
|
|
tenant_id=tenant_id,
|
|
error=str(e),
|
|
exc_info=True)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to delete tenant data: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get(
|
|
route_builder.build_base_route("tenant/{tenant_id}/deletion-preview", include_tenant_prefix=False),
|
|
response_model=dict
|
|
)
|
|
@service_only_access
|
|
async def preview_tenant_data_deletion(
|
|
tenant_id: str = Path(..., description="Tenant ID to preview deletion for"),
|
|
current_user: dict = Depends(get_current_user_dep)
|
|
):
|
|
"""
|
|
Preview what data would be deleted for a tenant (dry-run)
|
|
|
|
This endpoint shows counts of all data that would be deleted
|
|
without actually deleting anything. Useful for:
|
|
- Confirming deletion scope before execution
|
|
- Auditing and compliance
|
|
- Troubleshooting
|
|
|
|
Returns:
|
|
Dictionary with entity names and their counts
|
|
"""
|
|
from app.services.tenant_deletion_service import NotificationTenantDeletionService
|
|
from app.core.config import settings
|
|
|
|
try:
|
|
logger.info("notification.tenant_deletion.preview_called", tenant_id=tenant_id)
|
|
|
|
db_manager = create_database_manager(settings.DATABASE_URL, "notification")
|
|
|
|
async with db_manager.get_session() as session:
|
|
deletion_service = NotificationTenantDeletionService(session)
|
|
preview = await deletion_service.get_tenant_data_preview(tenant_id)
|
|
|
|
total_records = sum(preview.values())
|
|
|
|
return {
|
|
"tenant_id": tenant_id,
|
|
"service": "notification",
|
|
"preview": preview,
|
|
"total_records": total_records,
|
|
"note": "System templates are not counted and will be preserved",
|
|
"warning": "These records will be permanently deleted and cannot be recovered"
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("notification.tenant_deletion.preview_error",
|
|
tenant_id=tenant_id,
|
|
error=str(e),
|
|
exc_info=True)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to preview tenant data deletion: {str(e)}"
|
|
)
|