import uuid import httpx import json import logging from datetime import datetime, timezone, timedelta from typing import Dict, Any, List, Optional from sqlalchemy.orm import Session from sqlalchemy import func import hmac import hashlib from app.models.auth.module_environment_model import ModuleEnvironment from app.models.auth.module_model import Module from app.models.auth.tenant_module_model import TenantModule from app.models.system.event_log_model import EventLog, EventStatus from app.services.auth.trust_service import TrustService from app.core.redis import sync_redis_client logger = logging.getLogger(__name__) class EventService: @staticmethod def emit_event( db: Session, event_type: str, payload: Dict[str, Any], tenant_id: Optional[uuid.UUID] = None, follow_up_event: Optional[Dict[str, Any]] = None ): """ Emits an event by writing it to the Outbox (event_logs). Scopes delivery to relevant modules based on tenant_id. """ event_id = str(uuid.uuid4()) timestamp = datetime.now(timezone.utc).isoformat() if "event_id" not in payload: payload["event_id"] = event_id final_payload = { "event_id": event_id, "event_type": event_type, "timestamp": timestamp, "data": payload } targets: List[ModuleEnvironment] = [] payload_data = payload.get("data", payload) if isinstance(payload, dict) else {} targets_list = payload.get("targets") if targets_list and isinstance(targets_list, list): target_configs = targets_list for target in target_configs: module_id = target.get("module_id") env_slug = target.get("environment_slug") if module_id and env_slug: env = db.query(ModuleEnvironment).filter( ModuleEnvironment.module_id == module_id, ModuleEnvironment.slug == env_slug ).first() if env: targets.append(env) elif tenant_id: tenant_modules = db.query(TenantModule).filter( TenantModule.tenant_id == tenant_id, TenantModule.is_active == True ).all() for tm in tenant_modules: env = db.query(ModuleEnvironment).filter( ModuleEnvironment.module_id == tm.module_id, ModuleEnvironment.slug == (tm.assigned_environment_slug or "prod") ).first() if not env: env = db.query(ModuleEnvironment).filter( ModuleEnvironment.module_id == tm.module_id, ModuleEnvironment.is_default == True ).first() if env: targets.append(env) if not targets: logger.warning(f"Event {event_type} emitted with no resolved targets. Payload scoping: {'Explicit' if 'targets' in payload.get('data', {}) else 'Implicit'}") return for env in targets: base = env.backend_base_url.rstrip('/') if event_type in {"TENANT_PROVISION_REQUESTED", "TENANT_UPDATED", "TENANT_STATUS_CHANGED", "TENANT_DEPROVISION_REQUESTED"} and env.provisioning_endpoint: endpoint = env.provisioning_endpoint.lstrip('/') logger.info(f"Trace: base='{base}', endpoint='{endpoint}'") target_url = f"{base}/{endpoint}" logger.info(f"Trace: Calculated target_url='{target_url}'") else: logger.info(f"Using default event stream for env '{env.slug}'. ProvEndpoint: '{env.provisioning_endpoint}'") target_url = f"{base}/api/internal/events" log = EventLog( event_id=uuid.UUID(event_id), event_type=event_type, payload=final_payload, target_module_id=env.module_id, target_environment_slug=env.slug, target_url=target_url, status=EventStatus.PENDING, next_retry_at=datetime.now(timezone.utc), follow_up_event=follow_up_event ) db.add(log) db.flush() try: sync_redis_client.rpush("saas:events:queue", event_id) except Exception as e: logger.error(f"Failed to push event to Redis queue: {e}") @staticmethod def process_queue_item(db: Session, event_id: str): """ Process all pending EventLogs associated with the given logical event_id. """ logs = db.query(EventLog).filter( EventLog.event_id == uuid.UUID(event_id), EventLog.status == EventStatus.PENDING ).all() if not logs: return 0 processed_count = 0 for log in logs: try: env = db.query(ModuleEnvironment).filter( ModuleEnvironment.module_id == log.target_module_id, ModuleEnvironment.slug == log.target_environment_slug ).first() if not env: log.status = EventStatus.FAILED log.error_log = "Target environment config missing" continue payload_json = json.dumps(log.payload) secret = env.trust_credentials.get("hmac_secret") if env.trust_credentials else None if secret: signature = hmac.new( secret.encode("utf-8"), payload_json.encode("utf-8"), hashlib.sha256 ).hexdigest() else: signature = "" headers = { "Content-Type": "application/json", "X-SaaS-Signature": signature, "X-SaaS-Event-Source": "saas-core" } logger.info(f"Sending event {log.event_type} to {log.target_url}") response = httpx.post(log.target_url, content=payload_json, headers=headers, timeout=5) if response.status_code in range(200, 300): log.status = EventStatus.COMPLETED log.error_log = None processed_count += 1 if log.follow_up_event: follow_up = log.follow_up_event logger.info(f"Triggering follow-up event {follow_up.get('event_type')} after {log.event_type} completed") EventService.emit_event( db, event_type=follow_up["event_type"], payload=follow_up["payload"], tenant_id=uuid.UUID(follow_up["tenant_id"]) if follow_up.get("tenant_id") else None ) else: log.retry_count += 1 backoff = min(60 * (2 ** log.retry_count), 86400) log.next_retry_at = datetime.now(timezone.utc) + timedelta(seconds=backoff) log.error_log = f"HTTP {response.status_code}: {response.text}" if log.retry_count > 10: log.status = EventStatus.FAILED except Exception as e: log.retry_count += 1 backoff = min(60 * (2 ** log.retry_count), 86400) log.next_retry_at = datetime.now(timezone.utc) + timedelta(seconds=backoff) log.error_log = str(e) db.commit() return processed_count @staticmethod def process_outbox(db: Session, batch_size: int = 50): """ Worker method to process pending events. """ now = datetime.now(timezone.utc) logs = db.query(EventLog).filter( EventLog.status == EventStatus.PENDING, EventLog.next_retry_at <= now ).limit(batch_size).all() if logs: logger.info(f"Found {len(logs)} events to process (next_retry_at <= {now})") for log in logs: try: env = db.query(ModuleEnvironment).filter( ModuleEnvironment.module_id == log.target_module_id, ModuleEnvironment.slug == log.target_environment_slug ).first() if not env: log.status = EventStatus.FAILED log.error_log = "Target environment config missing" continue payload_json = json.dumps(log.payload) secret = env.trust_credentials.get("hmac_secret") if secret: signature = hmac.new( secret.encode("utf-8"), payload_json.encode("utf-8"), hashlib.sha256 ).hexdigest() else: signature = "" headers = { "Content-Type": "application/json", "X-SaaS-Signature": signature, "X-SaaS-Event-Source": "saas-core" } logger.info(f"Sending event {log.event_type} to {log.target_url}. Payload: {payload_json}") response = httpx.post(log.target_url, content=payload_json, headers=headers, timeout=5) if response.status_code in range(200, 300): log.status = EventStatus.COMPLETED log.error_log = None if log.follow_up_event: follow_up = log.follow_up_event logger.info(f"Triggering follow-up event {follow_up.get('event_type')} after {log.event_type} completed") EventService.emit_event( db, event_type=follow_up["event_type"], payload=follow_up["payload"], tenant_id=uuid.UUID(follow_up["tenant_id"]) if follow_up.get("tenant_id") else None ) else: log.retry_count += 1 backoff = min(60 * (2 ** log.retry_count), 86400) log.next_retry_at = now + timedelta(seconds=backoff) log.error_log = f"HTTP {response.status_code}: {response.text}" if log.retry_count > 10: log.status = EventStatus.FAILED except Exception as e: log.retry_count += 1 backoff = min(60 * (2 ** log.retry_count), 86400) log.next_retry_at = now + timedelta(seconds=backoff) log.error_log = str(e) db.commit() return len(logs)