# shared/alerts/base_service.py """ Base alert service pattern for all microservices Supports both alerts and recommendations through unified detection patterns """ import asyncio import json import random import uuid from typing import List, Dict, Any, Optional from uuid import UUID from datetime import datetime, timedelta import structlog from redis.asyncio import Redis from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from shared.messaging.rabbitmq import RabbitMQClient from shared.database.base import DatabaseManager from shared.config.rabbitmq_config import get_routing_key logger = structlog.get_logger() class BaseAlertService: """ Base class for service-specific alert and recommendation detection Implements hybrid detection patterns: scheduled jobs, event-driven, and database triggers """ def __init__(self, config): self.config = config self.db_manager = DatabaseManager(config.DATABASE_URL) self.rabbitmq_client = RabbitMQClient(config.RABBITMQ_URL, config.SERVICE_NAME) self.redis = None self.scheduler = AsyncIOScheduler() self.is_leader = False self.exchange = "alerts.exchange" # Metrics self._items_published = 0 self._checks_performed = 0 self._errors_count = 0 async def start(self): """Initialize all detection mechanisms""" try: # Connect to Redis for leader election and deduplication # Use the shared Redis URL which includes TLS configuration from redis.asyncio import from_url redis_url = self.config.REDIS_URL # Create Redis client from URL (supports TLS via rediss:// protocol) # For self-signed certificates, disable SSL verification redis_kwargs = { 'decode_responses': True, 'max_connections': 20 } # If using SSL/TLS, add SSL parameters to handle self-signed certificates if redis_url.startswith('rediss://'): redis_kwargs.update({ 'ssl_cert_reqs': None, # Disable certificate verification 'ssl_ca_certs': None, # Don't require CA certificates 'ssl_certfile': None, # Don't require client cert 'ssl_keyfile': None # Don't require client key }) self.redis = await from_url(redis_url, **redis_kwargs) logger.info("Connected to Redis", service=self.config.SERVICE_NAME, redis_url=redis_url.split("@")[-1]) # Connect to RabbitMQ await self.rabbitmq_client.connect() logger.info("Connected to RabbitMQ", service=self.config.SERVICE_NAME) # Start leader election for scheduled jobs asyncio.create_task(self.maintain_leadership()) # Setup scheduled checks (runs only on leader) self.setup_scheduled_checks() # Start database listener (runs on all instances) await self.start_database_listener() # Start event listener (runs on all instances) await self.start_event_listener() logger.info("Alert service started", service=self.config.SERVICE_NAME) except Exception as e: logger.error("Failed to start alert service", service=self.config.SERVICE_NAME, error=str(e)) raise async def stop(self): """Clean shutdown""" try: # Stop scheduler if self.scheduler.running: self.scheduler.shutdown() # Close connections if self.redis: await self.redis.aclose() # Use aclose() for modern Redis client await self.rabbitmq_client.disconnect() logger.info("Alert service stopped", service=self.config.SERVICE_NAME) except Exception as e: logger.error("Error stopping alert service", service=self.config.SERVICE_NAME, error=str(e)) # PATTERN 1: Scheduled Background Jobs def setup_scheduled_checks(self): """Configure scheduled alert checks - Override in service""" raise NotImplementedError("Subclasses must implement setup_scheduled_checks") async def maintain_leadership(self): """Leader election for scheduled jobs""" lock_key = f"scheduler_lock:{self.config.SERVICE_NAME}" lock_ttl = 60 # Generate instance_id once for the lifetime of this leadership loop # IMPORTANT: Don't regenerate on each iteration or lock extension will always fail! instance_id = getattr(self.config, 'INSTANCE_ID', str(uuid.uuid4())) logger.info("DEBUG: maintain_leadership starting", service=self.config.SERVICE_NAME, instance_id=instance_id, redis_client_type=str(type(self.redis))) while True: try: was_leader = self.is_leader # Add jitter to avoid thundering herd when multiple instances start if not was_leader: await asyncio.sleep(random.uniform(0.1, 0.5)) # Small random delay before attempting to acquire # Try to acquire new leadership if not currently leader if not self.is_leader: # Use atomic Redis operation to acquire lock result = await self.redis.set( lock_key, instance_id, ex=lock_ttl, nx=True # Only set if key doesn't exist ) acquired = result is not None self.is_leader = acquired else: # Already leader - try to extend the lock atomically # Use SET with EX and GET to atomically refresh the lock try: # SET key value EX ttl GET returns the old value (atomic check-and-set) # This is atomic and works in both standalone and cluster mode old_value = await self.redis.set( lock_key, instance_id, ex=lock_ttl, get=True # Return old value (Python redis uses 'get' param for GET option) ) # If old value matches our instance_id, we successfully extended self.is_leader = old_value == instance_id if self.is_leader: logger.debug("Lock extended successfully", service=self.config.SERVICE_NAME, instance_id=instance_id, ttl=lock_ttl) else: # Lock was taken by someone else or expired logger.info("Lost lock ownership during extension", service=self.config.SERVICE_NAME, old_owner=old_value, instance_id=instance_id) except Exception as e: # If extend fails, try to verify we still have the lock logger.warning("Failed to extend lock, verifying ownership", service=self.config.SERVICE_NAME, error=str(e)) current_check = await self.redis.get(lock_key) self.is_leader = current_check == instance_id # Handle leadership changes if self.is_leader and not was_leader: # Add a small delay to allow other instances to detect leadership change await asyncio.sleep(0.1) if self.is_leader: # Double-check we're still the leader self.scheduler.start() logger.info("Acquired scheduler leadership", service=self.config.SERVICE_NAME) elif not self.is_leader and was_leader: if self.scheduler.running: self.scheduler.shutdown() logger.info("Lost scheduler leadership", service=self.config.SERVICE_NAME) # Add jitter to reduce contention between instances await asyncio.sleep(lock_ttl // 2 + random.uniform(0, 2)) except Exception as e: import traceback logger.error("Leadership error", service=self.config.SERVICE_NAME, error=str(e), error_type=type(e).__name__, traceback=traceback.format_exc()) self.is_leader = False await asyncio.sleep(5) # PATTERN 2: Event-Driven Detection async def start_event_listener(self): """Listen for business events - Override in service""" pass # PATTERN 3: Database Triggers async def start_database_listener(self): """Listen for database notifications with connection management""" try: import asyncpg # Convert SQLAlchemy URL format to plain PostgreSQL for asyncpg database_url = self.config.DATABASE_URL if database_url.startswith('postgresql+asyncpg://'): database_url = database_url.replace('postgresql+asyncpg://', 'postgresql://') # Add connection timeout and retry logic max_retries = 3 retry_count = 0 conn = None while retry_count < max_retries and not conn: try: conn = await asyncio.wait_for( asyncpg.connect(database_url), timeout=10.0 ) break except (asyncio.TimeoutError, Exception) as e: retry_count += 1 if retry_count < max_retries: logger.warning(f"DB listener connection attempt {retry_count} failed, retrying...", service=self.config.SERVICE_NAME, error=str(e)) await asyncio.sleep(2) else: raise if conn: # Register listeners based on service await self.register_db_listeners(conn) logger.info("Database listeners registered", service=self.config.SERVICE_NAME) # Keep connection alive with periodic ping asyncio.create_task(self._maintain_db_connection(conn)) except Exception as e: logger.error("Failed to setup database listeners", service=self.config.SERVICE_NAME, error=str(e)) async def _maintain_db_connection(self, conn): """Maintain database connection for listeners""" try: while not conn.is_closed(): # Use a timeout to avoid hanging indefinitely try: await asyncio.wait_for( conn.fetchval("SELECT 1"), timeout=5.0 ) await asyncio.sleep(30) # Check every 30 seconds except asyncio.TimeoutError: logger.warning("DB ping timed out, connection may be dead", service=self.config.SERVICE_NAME) break except Exception as e: logger.error("DB listener connection lost", service=self.config.SERVICE_NAME, error=str(e)) break except Exception as e: logger.error("Error maintaining DB connection", service=self.config.SERVICE_NAME, error=str(e)) async def register_db_listeners(self, conn): """Register database listeners - Override in service""" pass # Publishing (Updated for type) async def publish_item(self, tenant_id: UUID, item: Dict[str, Any], item_type: str = 'alert'): """Publish alert or recommendation to RabbitMQ with deduplication""" try: # Generate proper deduplication key based on alert type and specific identifiers unique_id = self._generate_unique_identifier(item) item_key = f"{tenant_id}:{item_type}:{item['type']}:{unique_id}" if await self.is_duplicate_item(item_key): logger.debug("Duplicate item skipped", service=self.config.SERVICE_NAME, item_type=item_type, alert_type=item['type'], dedup_key=item_key) return False # Add metadata item['id'] = str(uuid.uuid4()) item['tenant_id'] = str(tenant_id) item['service'] = self.config.SERVICE_NAME item['timestamp'] = datetime.utcnow().isoformat() item['item_type'] = item_type # 'alert' or 'recommendation' # Determine routing key based on severity and type routing_key = get_routing_key(item_type, item['severity'], self.config.SERVICE_NAME) # Publish to RabbitMQ with timeout to prevent blocking try: success = await asyncio.wait_for( self.rabbitmq_client.publish_event( exchange_name=self.exchange, routing_key=routing_key, event_data=item ), timeout=10.0 # 10 second timeout ) except asyncio.TimeoutError: logger.error("RabbitMQ publish timed out", service=self.config.SERVICE_NAME, item_type=item_type, alert_type=item['type']) return False if success: self._items_published += 1 logger.info("Item published successfully", service=self.config.SERVICE_NAME, item_type=item_type, alert_type=item['type'], severity=item['severity'], routing_key=routing_key) else: self._errors_count += 1 logger.error("Failed to publish item", service=self.config.SERVICE_NAME, item_type=item_type, alert_type=item['type']) return success except Exception as e: self._errors_count += 1 logger.error("Error publishing item", service=self.config.SERVICE_NAME, error=str(e), item_type=item_type) return False def _generate_unique_identifier(self, item: Dict[str, Any]) -> str: """Generate unique identifier for deduplication based on alert type and content""" alert_type = item.get('type', '') metadata = item.get('metadata', {}) # Generate unique identifier based on alert type if alert_type == 'overstock_warning': return metadata.get('ingredient_id', '') elif alert_type == 'critical_stock_shortage' or alert_type == 'low_stock_warning': return metadata.get('ingredient_id', '') elif alert_type == 'expired_products': # For expired products alerts, create hash of all expired item IDs expired_items = metadata.get('expired_items', []) if expired_items: expired_ids = sorted([str(item.get('id', '')) for item in expired_items]) import hashlib return hashlib.md5(':'.join(expired_ids).encode()).hexdigest()[:16] return '' elif alert_type == 'urgent_expiry': return f"{metadata.get('ingredient_id', '')}:{metadata.get('stock_id', '')}" elif alert_type == 'temperature_breach': return f"{metadata.get('sensor_id', '')}:{metadata.get('location', '')}" elif alert_type == 'stock_depleted_by_order': return f"{metadata.get('order_id', '')}:{metadata.get('ingredient_id', '')}" elif alert_type == 'expired_batches_auto_processed': # Use processing date and total batches as identifier processing_date = metadata.get('processing_date', '')[:10] # Date only total_batches = metadata.get('total_batches_processed', 0) return f"{processing_date}:{total_batches}" elif alert_type == 'inventory_optimization': return f"opt:{metadata.get('ingredient_id', '')}:{metadata.get('recommendation_type', '')}" elif alert_type == 'waste_reduction': return f"waste:{metadata.get('ingredient_id', '')}" else: # Fallback to generic metadata.id or empty string return metadata.get('id', '') async def is_duplicate_item(self, item_key: str, window_minutes: int = 15) -> bool: """Prevent duplicate items within time window""" key = f"item_sent:{item_key}" try: result = await self.redis.set( key, "1", ex=window_minutes * 60, nx=True ) return result is None # None means duplicate except Exception as e: logger.error("Error checking duplicate", error=str(e)) return False # Allow publishing if check fails # Helper methods async def get_active_tenants(self) -> List[UUID]: """Get list of active tenant IDs""" try: from sqlalchemy import text query = text("SELECT DISTINCT tenant_id FROM tenants WHERE status = 'active'") async with self.db_manager.get_session() as session: result = await session.execute(query) return [row.tenant_id for row in result.fetchall()] except Exception as e: # If tenants table doesn't exist, skip tenant-based processing if "does not exist" in str(e): logger.debug("Tenants table not found, skipping tenant-based alert processing") return [] else: logger.error("Error fetching active tenants", error=str(e)) return [] async def get_tenant_config(self, tenant_id: UUID) -> Dict[str, Any]: """Get tenant-specific configuration""" try: from sqlalchemy import text query = text("SELECT config FROM tenants WHERE tenant_id = :tenant_id") async with self.db_manager.get_session() as session: result = await session.execute(query, {"tenant_id": tenant_id}) row = result.fetchone() return json.loads(row.config) if row and row.config else {} except Exception as e: logger.error("Error fetching tenant config", tenant_id=str(tenant_id), error=str(e)) return {} # Health and metrics def get_metrics(self) -> Dict[str, Any]: """Get service metrics""" return { "items_published": self._items_published, "checks_performed": self._checks_performed, "errors_count": self._errors_count, "is_leader": self.is_leader, "scheduler_running": self.scheduler.running, "redis_connected": self.redis and not self.redis.closed, "rabbitmq_connected": self.rabbitmq_client.connected if self.rabbitmq_client else False } async def health_check(self) -> Dict[str, Any]: """Comprehensive health check""" try: # Check Redis redis_healthy = False if self.redis and not self.redis.closed: await self.redis.ping() redis_healthy = True # Check RabbitMQ rabbitmq_healthy = self.rabbitmq_client.connected if self.rabbitmq_client else False # Check database db_healthy = False try: from sqlalchemy import text async with self.db_manager.get_session() as session: await session.execute(text("SELECT 1")) db_healthy = True except: pass status = "healthy" if all([redis_healthy, rabbitmq_healthy, db_healthy]) else "unhealthy" return { "status": status, "service": self.config.SERVICE_NAME, "components": { "redis": "healthy" if redis_healthy else "unhealthy", "rabbitmq": "healthy" if rabbitmq_healthy else "unhealthy", "database": "healthy" if db_healthy else "unhealthy", "scheduler": "running" if self.scheduler.running else "stopped" }, "metrics": self.get_metrics() } except Exception as e: return { "status": "error", "service": self.config.SERVICE_NAME, "error": str(e) } class AlertServiceMixin: """Mixin providing common alert helper methods""" def format_spanish_message(self, template_key: str, **kwargs) -> Dict[str, Any]: """Format Spanish alert message""" from shared.alerts.templates import format_item_message return format_item_message(template_key, 'es', **kwargs) def get_business_hours_severity(self, base_severity: str) -> str: """Adjust severity based on business hours""" current_hour = datetime.now().hour # Reduce non-critical severity outside business hours (7-20) if not (7 <= current_hour <= 20): if base_severity == 'medium': return 'low' elif base_severity == 'high' and current_hour < 6 or current_hour > 22: return 'medium' return base_severity def should_send_recommendation(self, tenant_id: UUID, rec_type: str) -> bool: """Check if recommendation should be sent based on tenant preferences""" # Implement tenant-specific recommendation frequency limits # This is a simplified version return True