""" Incident-Approval Synchronization Service ========================================== ADR-027: Incident-Approval 同步架構 確保 Incident 和 Approval 的原子性同步： 1. 建立時原子同步 (UnitOfWork + Saga) 2. 狀態變更時雙向傳播 設計原則: - PostgreSQL 事務 (UnitOfWork) 確保資料庫原子性 - Saga Pattern 處理 Redis 寫入失敗 - 狀態同步 Hook 確保一致性 版本: v1.0 建立: 2026-03-26 (台北時區) """ from datetime import UTC, datetime from typing import TYPE_CHECKING from uuid import uuid4 import structlog from redis.exceptions import RedisError from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.core.constants import ( APPROVAL_TO_INCIDENT_STATUS, APPROVAL_TTL_SECONDS, INCIDENT_TTL_SECONDS, REDIS_KEY_APPROVAL, REDIS_KEY_INCIDENT, ) from src.core.unit_of_work import UnitOfWork from src.db.models import ApprovalRecord, IncidentRecord from src.models.approval import ApprovalRequestCreate, ApprovalStatus from src.models.incident import IncidentStatus from src.services.approval_db import add_approval_created_timeline_event if TYPE_CHECKING: from redis.asyncio import Redis logger = structlog.get_logger(__name__) class IncidentApprovalSyncError(Exception): """Incident-Approval 同步失敗""" pass class IncidentApprovalService: """ Incident-Approval 同步服務 職責: 1. 原子建立 Incident + Approval (create_with_approval) 2. 狀態同步 (on_approval_status_change) 3. TTL 統一管理 使用方式: service = IncidentApprovalService(session_factory, redis_client) # 原子建立 incident, approval = await service.create_with_approval( incident_data, approval_data ) # 狀態變更同步 await service.on_approval_status_change(approval_id, "approved") Warning: 禁止繞過此 Service 直接建立 Incident 或 Approval """ def __init__( self, session_factory: async_sessionmaker[AsyncSession], redis_client: "Redis", ): """ 初始化 IncidentApprovalService Args: session_factory: SQLAlchemy async session factory redis_client: Redis async client """ self._session_factory = session_factory self._redis = redis_client # ========================================================================= # 原子建立 (Phase 1 Core) # ========================================================================= async def create_with_approval( self, incident_data: dict, approval_data: ApprovalRequestCreate, *, link_metadata: bool = True, ) -> tuple[IncidentRecord, ApprovalRecord]: """ 原子建立 Incident + Approval 保證: 1. PostgreSQL 事務：兩者同時成功或失敗 2. Saga 補償：Redis 失敗時回滾 PostgreSQL Args: incident_data: Incident 資料 (dict for IncidentRecord) approval_data: Approval 建立請求 link_metadata: 是否在 metadata 中建立雙向連結 Returns: tuple[IncidentRecord, ApprovalRecord]: 建立的記錄 Raises: IncidentApprovalSyncError: 建立失敗 (已回滾) Warning: 不要直接使用 incident_service.create() 或 approval_db.create_approval() """ async with UnitOfWork(self._session_factory) as uow: try: # 1. 建立 Incident (PostgreSQL) incident_id = incident_data.get("incident_id") or str(uuid4()) incident_data["incident_id"] = incident_id incident_record = IncidentRecord( id=incident_id, title=incident_data.get("title", ""), description=incident_data.get("description", ""), severity=incident_data.get("severity", "P2"), status=incident_data.get("status", "investigating"), source=incident_data.get("source", "alertmanager"), fingerprint=incident_data.get("fingerprint"), extra_metadata=incident_data.get("metadata", {}), ) uow.session.add(incident_record) await uow.flush() # 取得 ID # 2. 建立 Approval (PostgreSQL) # 準備 metadata，建立雙向連結 approval_metadata = approval_data.metadata or {} if link_metadata: approval_metadata["incident_id"] = incident_id approval_record = ApprovalRecord( action=approval_data.action, description=approval_data.description, status=ApprovalStatus.PENDING, risk_level=approval_data.risk_level or "MEDIUM", required_signatures=1, current_signatures=0, signatures=[], blast_radius=approval_data.blast_radius.model_dump() if approval_data.blast_radius else {}, dry_run_checks=[ c.model_dump() for c in (approval_data.dry_run_checks or []) ], requested_by=approval_data.requested_by, expires_at=approval_data.expires_at, extra_metadata=approval_metadata, incident_id=incident_id, fingerprint=incident_data.get("fingerprint"), ) uow.session.add(approval_record) await uow.flush() # 取得 ID add_approval_created_timeline_event(uow.session, approval_record) # 更新 Incident metadata 連結 Approval if link_metadata: incident_record.extra_metadata = { **(incident_record.extra_metadata or {}), "approval_id": approval_record.id, } # 3. 寫入 Redis (事務外，需 Saga 補償) try: await self._write_to_redis(incident_record, approval_record) except RedisError as e: # Saga 補償: rollback PostgreSQL logger.error( "redis_write_failed_saga_rollback", incident_id=incident_id, approval_id=approval_record.id, error=str(e), ) await uow.rollback() raise IncidentApprovalSyncError(f"Redis write failed: {e}") from e # 4. Commit PostgreSQL await uow.commit() logger.info( "incident_approval_created", incident_id=incident_id, approval_id=approval_record.id, ) return incident_record, approval_record except IncidentApprovalSyncError: raise except Exception as e: logger.exception( "incident_approval_create_error", error=str(e), ) raise IncidentApprovalSyncError(f"Create failed: {e}") from e async def _write_to_redis( self, incident: IncidentRecord, approval: ApprovalRecord, ) -> None: """ 寫入 Redis (Working Memory) 兩個 key 必須同時成功，使用 pipeline 批量操作。 Args: incident: Incident 記錄 approval: Approval 記錄 """ incident_key = f"{REDIS_KEY_INCIDENT}{incident.id}" approval_key = f"{REDIS_KEY_APPROVAL}{approval.id}" # 序列化 incident_json = self._serialize_incident(incident) approval_json = self._serialize_approval(approval) # Pipeline 批量寫入 async with self._redis.pipeline() as pipe: pipe.set(incident_key, incident_json, ex=INCIDENT_TTL_SECONDS) pipe.set(approval_key, approval_json, ex=APPROVAL_TTL_SECONDS) await pipe.execute() logger.debug( "redis_written", incident_key=incident_key, approval_key=approval_key, ttl=INCIDENT_TTL_SECONDS, ) def _serialize_incident(self, record: IncidentRecord) -> str: """序列化 IncidentRecord 為 JSON""" import json return json.dumps({ "incident_id": record.id, "title": record.title, "description": record.description, "severity": record.severity, "status": record.status, "source": record.source, "fingerprint": record.fingerprint, "metadata": record.extra_metadata, "created_at": record.created_at.isoformat() if record.created_at else None, }) def _serialize_approval(self, record: ApprovalRecord) -> str: """序列化 ApprovalRecord 為 JSON""" import json status_val = record.status.value if hasattr(record.status, "value") else record.status risk_val = record.risk_level.value if hasattr(record.risk_level, "value") else record.risk_level return json.dumps({ "id": record.id, "action": record.action, "description": record.description, "status": status_val, "risk_level": risk_val, "required_signatures": record.required_signatures, "current_signatures": record.current_signatures, "signatures": record.signatures, "blast_radius": record.blast_radius, "requested_by": record.requested_by, "created_at": record.created_at.isoformat() if record.created_at else None, "incident_id": getattr(record, "incident_id", None), "metadata": record.extra_metadata, }) # ========================================================================= # 狀態同步 (Phase 2) # ========================================================================= async def on_approval_status_change( self, approval_id: str, new_status: ApprovalStatus | str, ) -> None: """ Approval 狀態變更時同步 Incident 行為: 1. 更新 Approval 狀態 (PostgreSQL) 2. 同步更新關聯 Incident 狀態 3. 更新 Redis TTL Args: approval_id: Approval ID new_status: 新狀態 (ApprovalStatus 或字串) Note: 呼叫時機: sign_approval() 成功後 """ if isinstance(new_status, str): new_status = ApprovalStatus(new_status) async with UnitOfWork(self._session_factory) as uow: # 1. 取得 Approval result = await uow.session.execute( select(ApprovalRecord).where(ApprovalRecord.id == approval_id) ) approval = result.scalar_one_or_none() if not approval: logger.warning("approval_not_found_for_sync", approval_id=approval_id) return # 2. 更新 Approval 狀態 approval.status = new_status if new_status in (ApprovalStatus.APPROVED, ApprovalStatus.REJECTED): approval.resolved_at = datetime.now(UTC) # 3. 取得關聯 Incident ID incident_id = approval.incident_id or (approval.extra_metadata or {}).get("incident_id") if not incident_id: logger.debug( "no_linked_incident", approval_id=approval_id, ) await uow.commit() return # 4. 更新 Incident 狀態 result = await uow.session.execute( select(IncidentRecord).where(IncidentRecord.incident_id == incident_id) ) incident = result.scalar_one_or_none() if incident: new_incident_status = IncidentStatus( APPROVAL_TO_INCIDENT_STATUS.get( new_status.value if hasattr(new_status, "value") else new_status, "investigating", ) ) incident.status = new_incident_status if new_incident_status == IncidentStatus.RESOLVED: incident.resolved_at = datetime.now(UTC) logger.info( "incident_status_synced", incident_id=incident_id, approval_id=approval_id, new_status=new_incident_status, ) # 5. 更新 Redis TTL try: await self._refresh_redis_ttl(incident_id, approval_id) except RedisError as e: # Redis 失敗不阻斷主流程 logger.warning( "redis_ttl_refresh_failed", incident_id=incident_id, error=str(e), ) await uow.commit() async def _refresh_redis_ttl( self, incident_id: str, approval_id: str, ) -> None: """ 刷新 Redis TTL 用於確保關聯的 Incident 和 Approval 同時過期。 """ incident_key = f"{REDIS_KEY_INCIDENT}{incident_id}" approval_key = f"{REDIS_KEY_APPROVAL}{approval_id}" async with self._redis.pipeline() as pipe: pipe.expire(incident_key, INCIDENT_TTL_SECONDS) pipe.expire(approval_key, APPROVAL_TTL_SECONDS) await pipe.execute() # ========================================================================= # 查詢輔助 # ========================================================================= async def get_incident_by_approval_id( self, approval_id: str, ) -> IncidentRecord | None: """ 透過 Approval ID 取得關聯的 Incident Args: approval_id: Approval ID Returns: IncidentRecord | None """ async with UnitOfWork(self._session_factory) as uow: # 取得 Approval result = await uow.session.execute( select(ApprovalRecord).where(ApprovalRecord.id == approval_id) ) approval = result.scalar_one_or_none() if not approval: return None # 取得 Incident ID incident_id = (approval.extra_metadata or {}).get("incident_id") if not incident_id: return None # 取得 Incident result = await uow.session.execute( select(IncidentRecord).where(IncidentRecord.incident_id == incident_id) ) return result.scalar_one_or_none() # ============================================================================= # Factory Function (Dependency Injection) # ============================================================================= _incident_approval_service: IncidentApprovalService | None = None def get_incident_approval_service() -> IncidentApprovalService: """ 取得 IncidentApprovalService 單例 使用方式: service = get_incident_approval_service() await service.on_approval_status_change(approval_id, "approved") Note: - 使用全域 session factory 和 redis client - 首次呼叫時初始化，後續重用 """ global _incident_approval_service if _incident_approval_service is None: from src.core.redis_client import get_redis from src.db.base import get_session_factory _incident_approval_service = IncidentApprovalService( session_factory=get_session_factory(), redis_client=get_redis(), ) return _incident_approval_service def reset_incident_approval_service() -> None: """ 重置 IncidentApprovalService 單例 (用於測試) """ global _incident_approval_service _incident_approval_service = None