diff --git a/apps/api/src/api/v1/audit_logs.py b/apps/api/src/api/v1/audit_logs.py index a1caca917..7a2ead249 100644 --- a/apps/api/src/api/v1/audit_logs.py +++ b/apps/api/src/api/v1/audit_logs.py @@ -11,17 +11,14 @@ Endpoints: 提供 K8s 操作執行的完整審計軌跡。 """ -from datetime import timedelta from typing import Any from fastapi import APIRouter, HTTPException, Query, status from pydantic import BaseModel -from sqlalchemy import func, select from src.core.logging import get_logger -from src.db.base import get_db_context from src.db.models import AuditLog -from src.utils.timezone import now_taipei +from src.repositories.audit_log_repository import get_audit_log_repository router = APIRouter(prefix="/audit-logs", tags=["Audit Logs"]) logger = get_logger("awoooi.audit") @@ -122,46 +119,33 @@ async def list_audit_logs( Returns: AuditLogListResponse: 分頁稽核日誌 """ - async with get_db_context() as db: - # Build query - query = select(AuditLog) + # Phase 17 P0: Router 層違規修復 - 改用 Repository 層 + repo = get_audit_log_repository() - if success is not None: - query = query.where(AuditLog.success == success) - if operation_type: - query = query.where(AuditLog.operation_type == operation_type) - if namespace: - query = query.where(AuditLog.namespace == namespace) + logs, total_count = await repo.list_logs( + page=page, + page_size=page_size, + success_only=success, + namespace=namespace, + ) - # Count total - count_query = select(func.count()).select_from(query.subquery()) - total_result = await db.execute(count_query) - total_count = total_result.scalar() or 0 + # TODO: operation_type 篩選暫未實作，需要擴展 Repository + total_pages = (total_count + page_size - 1) // page_size if total_count > 0 else 1 - # Pagination - offset = (page - 1) * page_size - query = query.order_by(AuditLog.created_at.desc()) - query = query.offset(offset).limit(page_size) + logger.info( + "audit_logs_listed", + count=len(logs), + page=page, + total=total_count, + ) - result = await db.execute(query) - logs = result.scalars().all() - - total_pages = (total_count + page_size - 1) // page_size if total_count > 0 else 1 - - logger.info( - "audit_logs_listed", - count=len(logs), - page=page, - total=total_count, - ) - - return AuditLogListResponse( - count=total_count, - logs=[audit_log_to_response(log) for log in logs], - page=page, - page_size=page_size, - total_pages=total_pages, - ) + return AuditLogListResponse( + count=total_count, + logs=[audit_log_to_response(log) for log in logs], + page=page, + page_size=page_size, + total_pages=total_pages, + ) # ============================================================================= @@ -190,71 +174,26 @@ async def get_audit_stats() -> AuditStatsResponse: Returns: AuditStatsResponse: 統計資訊 """ + # Phase 17 P0: Router 層違規修復 - 改用 Repository 層 + repo = get_audit_log_repository() + stats = await repo.get_stats(since_hours=24) - async with get_db_context() as db: - # Total count - total_result = await db.execute(select(func.count(AuditLog.id))) - total_count = total_result.scalar() or 0 + logger.info( + "audit_stats_fetched", + total=stats["total"], + success_rate=stats["success_rate"], + ) - # Success/Failure count - success_result = await db.execute( - select(func.count(AuditLog.id)).where(AuditLog.success is True) - ) - success_count = success_result.scalar() or 0 - failure_count = total_count - success_count - - # Success rate - success_rate = (success_count / total_count * 100) if total_count > 0 else 0.0 - - # Average duration - avg_result = await db.execute( - select(func.avg(AuditLog.execution_duration_ms)).where( - AuditLog.execution_duration_ms.isnot(None) - ) - ) - avg_duration = avg_result.scalar() - - # By operation type - op_result = await db.execute( - select( - AuditLog.operation_type, - func.count(AuditLog.id) - ).group_by(AuditLog.operation_type) - ) - by_operation = {row[0]: row[1] for row in op_result.fetchall()} - - # By namespace - ns_result = await db.execute( - select( - AuditLog.namespace, - func.count(AuditLog.id) - ).group_by(AuditLog.namespace) - ) - by_namespace = {row[0]: row[1] for row in ns_result.fetchall()} - - # Last 24 hours - cutoff = now_taipei() - timedelta(hours=24) - last24_result = await db.execute( - select(func.count(AuditLog.id)).where(AuditLog.created_at >= cutoff) - ) - last_24h_count = last24_result.scalar() or 0 - - logger.info( - "audit_stats_fetched", - total=total_count, - success_rate=round(success_rate, 2), - ) - - return AuditStatsResponse( - total_executions=total_count, - success_count=success_count, - failure_count=failure_count, - success_rate=round(success_rate, 2), - avg_duration_ms=round(avg_duration, 2) if avg_duration else None, - by_operation_type=by_operation, - by_namespace=by_namespace, - last_24h_count=last_24h_count, - ) + return AuditStatsResponse( + total_executions=stats["total"], + success_count=stats["success_count"], + failure_count=stats["failure_count"], + success_rate=stats["success_rate"], + avg_duration_ms=stats["avg_duration_ms"], + by_operation_type=stats["by_operation_type"], + by_namespace=stats["by_namespace"], + last_24h_count=stats["last_24h_count"], + ) # ============================================================================= @@ -280,21 +219,19 @@ async def get_audit_log(log_id: str) -> AuditLogResponse: Raises: HTTPException: 404 找不到日誌 """ - async with get_db_context() as db: - result = await db.execute( - select(AuditLog).where(AuditLog.id == log_id) - ) - log = result.scalar_one_or_none() + # Phase 17 P0: Router 層違規修復 - 改用 Repository 層 + repo = get_audit_log_repository() + log = await repo.get_by_id(log_id) - if log is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Audit log not found", - ) - - logger.info( - "audit_log_fetched", - log_id=log_id, + if log is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Audit log not found", ) - return audit_log_to_response(log) + logger.info( + "audit_log_fetched", + log_id=log_id, + ) + + return audit_log_to_response(log) diff --git a/apps/api/src/api/v1/incidents.py b/apps/api/src/api/v1/incidents.py index 092c5117c..c4cb154ee 100644 --- a/apps/api/src/api/v1/incidents.py +++ b/apps/api/src/api/v1/incidents.py @@ -23,13 +23,13 @@ from fastapi import APIRouter, HTTPException, status from pydantic import BaseModel, Field from src.core.logging import get_logger -from src.core.redis_client import get_redis from src.models.approval import ApprovalRequestResponse from src.models.incident import Incident, IncidentStatus, Severity # Phase 16 R3.3b (2026-03-25 台北時區): Repository 層整合 from src.repositories.incident_repository import get_incident_repository from src.services.decision_manager import get_decision_manager +from src.services.incident_service import get_incident_service from src.services.proposal_service import get_proposal_service from src.utils.timezone import now_taipei @@ -121,40 +121,13 @@ async def list_incidents() -> IncidentListResponse: Returns: IncidentListResponse: 事件清單與計數 (含決策令牌) """ - redis_client = get_redis() + # Phase 17 P0: Router 層違規修復 - 改用 Service 層 + incident_service = get_incident_service() decision_manager = get_decision_manager() - incidents = [] try: - # 掃描所有 incident:INC-* keys - cursor = 0 - while True: - cursor, keys = await redis_client.scan( - cursor=cursor, - match="incident:INC-*", - count=100, - ) - - for key in keys: - try: - data = await redis_client.get(key) - if data: - incident = Incident.model_validate_json(data) - # 只返回活躍事件 - if incident.status in ( - IncidentStatus.INVESTIGATING, - IncidentStatus.MITIGATING, - ): - incidents.append(incident) - except Exception as e: - logger.warning( - "incident_parse_error", - key=key, - error=str(e), - ) - - if cursor == 0: - break + # 透過 Service 取得活躍事件 + incidents = await incident_service.get_active_incidents() # 按時間排序 (最新優先) incidents.sort(key=lambda i: i.created_at, reverse=True) @@ -234,19 +207,17 @@ async def get_incident(incident_id: str) -> IncidentResponse: Raises: HTTPException: 404 事件不存在 """ - redis_client = get_redis() - key = f"incident:{incident_id}" + # Phase 17 P0: Router 層違規修復 - 改用 Service 層 + incident_service = get_incident_service() try: - data = await redis_client.get(key) - if not data: + incident = await incident_service.get_from_working_memory(incident_id) + if incident is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Incident not found: {incident_id}", ) - incident = Incident.model_validate_json(data) - logger.info( "incident_fetched", incident_id=incident_id, @@ -325,13 +296,12 @@ async def generate_proposal(incident_id: str) -> ProposalGenerateResponse: risk_level=approval.risk_level.value, ) - # 取得更新後的 Incident 狀態 - redis_client = get_redis() + # Phase 17 P0: 改用 Service 層取得狀態 + incident_service = get_incident_service() incident_status = None try: - data = await redis_client.get(f"incident:{incident_id}") - if data: - incident = Incident.model_validate_json(data) + incident = await incident_service.get_from_working_memory(incident_id) + if incident: incident_status = incident.status.value except Exception: pass @@ -419,105 +389,46 @@ async def submit_feedback( Raises: HTTPException: 404 事件不存在 """ + # Phase 17 P0: Router 層違規修復 - 改用 Service 層 + incident_service = get_incident_service() - # Phase 16 R3.3b (2026-03-25): 移除直接 DB import，改用 Repository - # --- 以下為舊 import，已由 Repository 取代 --- - # from sqlalchemy import select - # from src.db.base import get_db_context - # from src.db.models import IncidentRecord - # --- 封存結束 --- - from src.models.incident import IncidentOutcome - - redis_client = get_redis() - redis_key = f"incident:{incident_id}" - - # 1. 取得現有 Incident try: - data = await redis_client.get(redis_key) - if not data: + incident = await incident_service.update_outcome( + incident_id=incident_id, + effectiveness_score=request.effectiveness_score, + human_feedback=request.human_feedback, + learning_notes=request.learning_notes, + should_remember=request.should_remember, + ) + + if incident is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Incident not found: {incident_id}", ) - incident = Incident.model_validate_json(data) - except HTTPException: - raise - except Exception as e: - logger.exception("feedback_redis_read_error", incident_id=incident_id) - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to read incident: {str(e)}", - ) from e - # 2. 更新或建立 outcome - if incident.outcome is None: - incident.outcome = IncidentOutcome() - - if request.effectiveness_score is not None: - incident.outcome.effectiveness_score = request.effectiveness_score - if request.human_feedback is not None: - incident.outcome.human_feedback = request.human_feedback - if request.learning_notes is not None: - incident.outcome.learning_notes = request.learning_notes - incident.outcome.should_remember = request.should_remember - incident.updated_at = now_taipei() - - # 3. 寫入 Redis - try: - await redis_client.set( - redis_key, - incident.model_dump_json(), - ex=604800, # 7 天 TTL - ) logger.info( - "feedback_redis_updated", + "feedback_updated", incident_id=incident_id, effectiveness_score=request.effectiveness_score, ) + + return FeedbackResponse( + success=True, + message="Feedback submitted successfully", + incident_id=incident_id, + outcome=incident.outcome.model_dump(mode="json") if incident.outcome else {}, + ) + + except HTTPException: + raise except Exception as e: - logger.exception("feedback_redis_write_error", incident_id=incident_id) + logger.exception("feedback_update_error", incident_id=incident_id) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Failed to update Redis: {str(e)}", + detail=f"Failed to update feedback: {str(e)}", ) from e - # 4. 同步到 PostgreSQL (Episodic Memory) - # Phase 16 R3.3b (2026-03-25 台北時區): 改用 Repository 層 - # 執行者: Claude Code - # 原因: 消除 Router 層直接 DB 存取，統一透過 Repository - try: - repo = get_incident_repository() - updated = await repo.update_outcome( - incident_id=incident_id, - outcome=incident.outcome.model_dump(mode="json"), - updated_at=now_taipei(), - ) - if updated: - logger.info( - "feedback_db_updated", - incident_id=incident_id, - ) - else: - logger.warning( - "feedback_db_record_not_found", - incident_id=incident_id, - message="將在下次 memory 同步時建立", - ) - except Exception as e: - logger.warning( - "feedback_db_write_error", - incident_id=incident_id, - error=str(e), - ) - # DB 寫入失敗不阻止回應 (Redis 已更新) - - return FeedbackResponse( - success=True, - message="Feedback submitted successfully", - incident_id=incident_id, - outcome=incident.outcome.model_dump(mode="json"), - ) - # ============================================================================= # DEBUG: 測試 Incident 狀態更新 @@ -533,68 +444,49 @@ async def debug_resolve_incident(incident_id: str) -> dict[str, Any]: DEBUG: 直接更新 Incident 狀態為 RESOLVED 用於測試 resolve_incident_after_approval 邏輯 - Phase 16 R3.4 (2026-03-26): 重構為使用 Repository 層 + Phase 17 P0: Router 層違規修復 - 改用 Service 層 """ + # Phase 17 P0: 改用 Service 層 + incident_service = get_incident_service() repo = get_incident_repository() - redis_client = get_redis() error_msg = None - # 1. 取得 Redis 當前狀態 + # 1. 取得更新前狀態 before_redis = None + before_db = None try: - data = await redis_client.get(f"incident:{incident_id}") - if data: - incident = Incident.model_validate_json(data) + incident = await incident_service.get_from_working_memory(incident_id) + if incident: before_redis = incident.status.value except Exception as e: error_msg = f"Redis read error: {e}" - # 2. 取得 DB 當前狀態 (透過 Repository) - before_db = None try: before_db = await repo.get_status(incident_id) except Exception as e: error_msg = f"DB read error: {e}" - # 3. 直接更新 Redis + # 2. 透過 Service 更新狀態 redis_updated = False - try: - data = await redis_client.get(f"incident:{incident_id}") - if data: - incident = Incident.model_validate_json(data) - incident.status = IncidentStatus.RESOLVED - incident.updated_at = now_taipei() - await redis_client.set( - f"incident:{incident_id}", - incident.model_dump_json(), - ex=604800, - ) - redis_updated = True - except Exception as e: - error_msg = f"Redis update error: {e}" - - # 4. 更新 DB (透過 Repository) db_updated = False try: - db_updated = await repo.update_status( - incident_id=incident_id, - status="resolved", - updated_at=now_taipei(), - ) + resolved = await incident_service.resolve_incident(incident_id) + if resolved: + redis_updated = True + db_updated = True except Exception as e: - error_msg = f"DB update error: {e}" + error_msg = f"Resolve error: {e}" - # 5. 驗證更新後狀態 + # 3. 驗證更新後狀態 after_redis = None + after_db = None try: - data = await redis_client.get(f"incident:{incident_id}") - if data: - incident = Incident.model_validate_json(data) + incident = await incident_service.get_from_working_memory(incident_id) + if incident: after_redis = incident.status.value except Exception: pass - after_db = None try: after_db = await repo.get_status(incident_id) except Exception: @@ -636,13 +528,16 @@ class SyncResult(BaseModel): async def sync_incidents_from_approvals() -> SyncResult: """ 掃描所有 pending Approvals，為缺少 Incident 的補建 + + Phase 17 P0: Router 層違規修復 - 改用 Service 層 """ from uuid import UUID from src.models.incident import Signal from src.services.approval_db import get_approval_service - redis_client = get_redis() + # Phase 17 P0: 改用 Service 層 + incident_service = get_incident_service() approval_service = get_approval_service() synced = 0 @@ -656,29 +551,9 @@ async def sync_incidents_from_approvals() -> SyncResult: for approval in pending_approvals: approval_id = str(approval.id) - # 檢查是否已有 Incident 關聯此 Approval - has_incident = False - cursor = 0 - while True: - cursor, keys = await redis_client.scan( - cursor=cursor, - match="incident:INC-*", - count=100, - ) - for key in keys: - try: - data = await redis_client.get(key) - if data: - incident = Incident.model_validate_json(data) - if UUID(approval_id) in incident.proposal_ids: - has_incident = True - break - except Exception: - pass - if has_incident or cursor == 0: - break - - if has_incident: + # Phase 17 P0: 透過 Service 查找關聯 Incident + existing = await incident_service.find_by_proposal_id(approval_id) + if existing: skipped += 1 continue @@ -716,12 +591,11 @@ async def sync_incidents_from_approvals() -> SyncResult: updated_at=now_taipei(), ) - key = f"incident:{incident.incident_id}" - await redis_client.set( - key, - incident.model_dump_json(), - ex=604800, # 7 days - ) + # Phase 17 P0: 透過 Service 儲存 + saved = await incident_service.save_to_working_memory(incident) + if not saved: + errors.append(f"Failed to save incident for approval {approval_id}") + continue synced += 1 details.append({ diff --git a/apps/api/src/api/v1/stats.py b/apps/api/src/api/v1/stats.py index ef168927a..c99c79e52 100644 --- a/apps/api/src/api/v1/stats.py +++ b/apps/api/src/api/v1/stats.py @@ -13,7 +13,6 @@ # - Redis 快取 (TTL 5 分鐘) # ============================================================================= -import json from datetime import datetime, timedelta from typing import Any @@ -23,55 +22,22 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from src.core.logging import get_logger -from src.core.redis_client import get_redis from src.db.base import get_db +from src.services.stats_service import get_stats_service from src.db.models import IncidentRecord from src.models.incident import IncidentStatus logger = get_logger(__name__) -# 快取 TTL (秒) +# Phase 17 P0: 快取 TTL 移至 StatsService +# 保留常量供參考 STATS_CACHE_TTL = 300 # 5 分鐘 -async def get_cached_or_compute( - cache_key: str, - compute_fn, - ttl: int = STATS_CACHE_TTL, -) -> dict[str, Any]: - """ - 快取包裝器: 先查 Redis，沒有則計算並快取 +# Phase 17 P0: 快取包裝器已移至 StatsService +# 使用方式: stats_service = get_stats_service() +# result = await stats_service.get_cached_or_compute(key, compute_fn) - Args: - cache_key: Redis key - compute_fn: 計算函數 (async callable) - ttl: 快取時間 (秒) - - Returns: - 快取或計算結果 - """ - redis_client = get_redis() - - # 嘗試從快取取得 - try: - cached = await redis_client.get(cache_key) - if cached: - logger.debug("stats_cache_hit", key=cache_key) - return json.loads(cached) - except Exception as e: - logger.warning("stats_cache_read_error", key=cache_key, error=str(e)) - - # 計算結果 - result = await compute_fn() - - # 寫入快取 - try: - await redis_client.set(cache_key, json.dumps(result), ex=ttl) - logger.debug("stats_cache_set", key=cache_key, ttl=ttl) - except Exception as e: - logger.warning("stats_cache_write_error", key=cache_key, error=str(e)) - - return result router = APIRouter(prefix="/stats", tags=["Statistics"]) diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index 7142dc44d..e8bb81c25 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -32,11 +32,8 @@ from pydantic import BaseModel, Field from src.core.config import settings from src.core.logging import get_logger -# Phase 6.1: Event Bus (Redis Streams) -from src.core.redis_client import get_redis - -# Phase 15.2: Trace Context 傳遞 -from src.core.telemetry import get_trace_context +# Phase 15.2: Trace Context (moved to SignalProducerService) +# get_trace_context 已移至 Service 層 from src.models.approval import ( ApprovalRequestCreate, BlastRadius, @@ -47,6 +44,10 @@ from src.models.approval import ( from src.models.incident import Incident, IncidentStatus, Severity, Signal from src.services.approval_db import get_approval_service +# Phase 17 P0: Service 層 (消除 Router 直接存取 Redis) +from src.services.incident_service import get_incident_service +from src.services.signal_producer import SignalData, get_signal_producer + # Phase 5: OpenClaw AI Engine from src.services.openclaw import get_openclaw @@ -99,7 +100,8 @@ async def create_incident_for_approval( """ from uuid import UUID - redis_client = get_redis() + # Phase 17 P0: Router 層違規修復 - 改用 Service 層 + incident_service = get_incident_service() # 映射嚴重度 severity = RISK_TO_SEVERITY.get(risk_level.lower(), Severity.P2) @@ -123,13 +125,8 @@ async def create_incident_for_approval( proposal_ids=[UUID(approval_id)], ) - # 存入 Redis (Working Memory) - key = f"incident:{incident.incident_id}" - await redis_client.set( - key, - incident.model_dump_json(), - ex=INCIDENT_TTL_SECONDS, - ) + # Phase 17 P0: 透過 Service 存入 Working Memory + await incident_service.save_to_working_memory(incident) logger.info( "incident_created_for_approval", @@ -539,54 +536,26 @@ async def produce_signal_to_stream(signal: SignalPayload) -> str: """ 將 Signal 寫入 Redis Stream - Phase 15.2: 注入 Trace Context 解決斷鏈問題 - - 使用 XADD 命令: - - MAXLEN ~10000: 限制 Stream 長度，自動裁剪舊訊息 - - *: 自動生成 Message ID + Phase 17 P0: Router 層違規修復 - 改用 Service 層 Returns: str: Redis Stream Message ID """ - redis_client = get_redis() + # Phase 17 P0: 透過 Service 寫入 Stream + producer = get_signal_producer() - # 組裝 Signal 字典 (所有值必須是字串) - signal_dict = { - "source": signal.source, - "alert_name": signal.alert_name, - "severity": signal.severity, - "namespace": signal.namespace, - "target": signal.target, - "message": signal.message, - "labels": str(signal.labels or {}), - "annotations": str(signal.annotations or {}), - "received_at": now_taipei().isoformat(), - } - - # Phase 15.2: 注入 Trace Context - trace_ctx = get_trace_context() - if trace_ctx: - signal_dict["_trace_id"] = trace_ctx.get("trace_id", "") - signal_dict["_span_id"] = trace_ctx.get("span_id", "") - - # XADD 寫入 Stream - message_id = await redis_client.xadd( - SIGNAL_STREAM_KEY, - signal_dict, - maxlen=SIGNAL_STREAM_MAXLEN, - approximate=True, # ~MAXLEN 近似裁剪，效能更好 - ) - - logger.info( - "signal_produced", - message_id=message_id, + signal_data = SignalData( source=signal.source, alert_name=signal.alert_name, severity=signal.severity, - trace_id=trace_ctx.get("trace_id") if trace_ctx else None, + namespace=signal.namespace, + target=signal.target, + message=signal.message, + labels=signal.labels, + annotations=signal.annotations, ) - return message_id + return await producer.produce(signal_data) @router.post( diff --git a/apps/api/src/repositories/audit_log_repository.py b/apps/api/src/repositories/audit_log_repository.py new file mode 100644 index 000000000..e6cef3803 --- /dev/null +++ b/apps/api/src/repositories/audit_log_repository.py @@ -0,0 +1,173 @@ +""" +Audit Log Repository - Phase 17 P0 Router 層違規修復 +==================================================== + +封裝 AuditLog 的資料庫操作，消除 Router 層直接執行 SQL。 + +符合 leWOOOgo 積木化規範: +- Router -> Service -> Repository -> DB +""" + +from datetime import timedelta +from typing import Any + +import structlog +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.db.base import get_db_context +from src.db.models import AuditLog + +logger = structlog.get_logger(__name__) + + +class AuditLogRepository: + """ + AuditLog Repository + + 封裝 AuditLog 表的 CRUD 操作 + """ + + async def list_logs( + self, + page: int = 1, + page_size: int = 20, + success_only: bool | None = None, + namespace: str | None = None, + ) -> tuple[list[AuditLog], int]: + """ + 分頁查詢 AuditLog + + Args: + page: 頁碼 + page_size: 每頁數量 + success_only: 只顯示成功/失敗 + namespace: 篩選 namespace + + Returns: + (logs, total_count) + """ + async with get_db_context() as db: + # 建立基本查詢 + query = select(AuditLog).order_by(AuditLog.created_at.desc()) + count_query = select(func.count(AuditLog.id)) + + # 篩選條件 + if success_only is not None: + query = query.where(AuditLog.success == success_only) + count_query = count_query.where(AuditLog.success == success_only) + + if namespace: + query = query.where(AuditLog.namespace == namespace) + count_query = count_query.where(AuditLog.namespace == namespace) + + # 總數 + total_result = await db.execute(count_query) + total = total_result.scalar() or 0 + + # 分頁 + offset = (page - 1) * page_size + query = query.offset(offset).limit(page_size) + + result = await db.execute(query) + logs = list(result.scalars().all()) + + return logs, total + + async def get_by_id(self, log_id: str) -> AuditLog | None: + """ + 根據 ID 取得 AuditLog + + Args: + log_id: 日誌 ID + + Returns: + AuditLog | None + """ + async with get_db_context() as db: + result = await db.execute( + select(AuditLog).where(AuditLog.id == log_id) + ) + return result.scalar_one_or_none() + + async def get_stats(self, since_hours: int = 24) -> dict[str, Any]: + """ + 取得統計資訊 + + Args: + since_hours: 統計最近幾小時 (預設 24) + + Returns: + 統計資訊字典 + """ + from src.utils.timezone import now_taipei + + async with get_db_context() as db: + # 總數 + total_result = await db.execute(select(func.count(AuditLog.id))) + total = total_result.scalar() or 0 + + # 成功數 + success_result = await db.execute( + select(func.count(AuditLog.id)).where(AuditLog.success.is_(True)) + ) + success_count = success_result.scalar() or 0 + + # 平均執行時間 + avg_result = await db.execute( + select(func.avg(AuditLog.execution_duration_ms)).where( + AuditLog.execution_duration_ms.isnot(None) + ) + ) + avg_duration = avg_result.scalar() + + # 按操作類型統計 + op_result = await db.execute( + select( + AuditLog.operation_type, + func.count(AuditLog.id), + ).group_by(AuditLog.operation_type) + ) + by_operation = {str(row[0]): row[1] for row in op_result.all()} + + # 按 namespace 統計 + ns_result = await db.execute( + select( + AuditLog.namespace, + func.count(AuditLog.id), + ).group_by(AuditLog.namespace) + ) + by_namespace = {str(row[0]): row[1] for row in ns_result.all()} + + # 最近 24 小時 + since = now_taipei() - timedelta(hours=since_hours) + last24_result = await db.execute( + select(func.count(AuditLog.id)).where(AuditLog.created_at >= since) + ) + last24_count = last24_result.scalar() or 0 + + return { + "total": total, + "success_count": success_count, + "failure_count": total - success_count, + "success_rate": round((success_count / total * 100) if total > 0 else 0, 2), + "avg_duration_ms": round(avg_duration, 2) if avg_duration else None, + "by_operation_type": by_operation, + "by_namespace": by_namespace, + "last_24h_count": last24_count, + } + + +# ============================================================================= +# Singleton +# ============================================================================= + +_audit_log_repository: AuditLogRepository | None = None + + +def get_audit_log_repository() -> AuditLogRepository: + """取得 AuditLogRepository 實例 (Singleton)""" + global _audit_log_repository + if _audit_log_repository is None: + _audit_log_repository = AuditLogRepository() + return _audit_log_repository diff --git a/apps/api/src/services/incident_service.py b/apps/api/src/services/incident_service.py index 00142228b..f7936c08a 100644 --- a/apps/api/src/services/incident_service.py +++ b/apps/api/src/services/incident_service.py @@ -520,6 +520,193 @@ class IncidentService: return {} return {} + # ========================================================================= + # Phase 17 P0: Router 層違規修復 - 新增方法 + # ========================================================================= + + async def update_outcome( + self, + incident_id: str, + effectiveness_score: int | None = None, + human_feedback: str | None = None, + learning_notes: str | None = None, + should_remember: bool = True, + ) -> Incident | None: + """ + 更新 Incident 的 outcome (人類回饋) + + Phase 17: 從 Router 層遷移至 Service 層 + + Args: + incident_id: 事件 ID + effectiveness_score: 有效性評分 (1-5) + human_feedback: 文字回饋 + learning_notes: 學習筆記 + should_remember: 是否納入長期記憶 + + Returns: + Incident | None: 更新後的事件，失敗返回 None + """ + from src.models.incident import IncidentOutcome + from src.repositories.incident_repository import get_incident_repository + from src.utils.timezone import now_taipei + + # 1. 從 Working Memory 讀取 + incident = await self.get_from_working_memory(incident_id) + if incident is None: + logger.warning("incident_not_found_for_outcome", incident_id=incident_id) + return None + + # 2. 更新 outcome + if incident.outcome is None: + incident.outcome = IncidentOutcome() + + if effectiveness_score is not None: + incident.outcome.effectiveness_score = effectiveness_score + if human_feedback is not None: + incident.outcome.human_feedback = human_feedback + if learning_notes is not None: + incident.outcome.learning_notes = learning_notes + incident.outcome.should_remember = should_remember + incident.updated_at = now_taipei() + + # 3. 寫入 Working Memory + redis_success = await self.save_to_working_memory(incident) + if not redis_success: + logger.error("outcome_redis_write_failed", incident_id=incident_id) + return None + + # 4. 同步到 Episodic Memory (PostgreSQL) + try: + repo = get_incident_repository() + await repo.update_outcome( + incident_id=incident_id, + outcome=incident.outcome.model_dump(mode="json"), + updated_at=now_taipei(), + ) + logger.info("outcome_db_updated", incident_id=incident_id) + except Exception as e: + logger.warning( + "outcome_db_update_failed", + incident_id=incident_id, + error=str(e), + ) + # DB 失敗不影響主流程 + + return incident + + async def resolve_incident(self, incident_id: str) -> Incident | None: + """ + 將 Incident 狀態更新為 RESOLVED + + Phase 17: 從 Router 層遷移至 Service 層 + + Args: + incident_id: 事件 ID + + Returns: + Incident | None: 更新後的事件，失敗返回 None + """ + from src.repositories.incident_repository import get_incident_repository + from src.utils.timezone import now_taipei + + # 1. 從 Working Memory 讀取 + incident = await self.get_from_working_memory(incident_id) + if incident is None: + logger.warning("incident_not_found_for_resolve", incident_id=incident_id) + return None + + # 2. 更新狀態 + incident.status = IncidentStatus.RESOLVED + incident.resolved_at = now_taipei() + incident.updated_at = now_taipei() + + # 3. 寫入 Working Memory + redis_success = await self.save_to_working_memory(incident) + if not redis_success: + logger.error("resolve_redis_write_failed", incident_id=incident_id) + return None + + # 4. 同步到 Episodic Memory + try: + repo = get_incident_repository() + await repo.update_status( + incident_id=incident_id, + status="resolved", + updated_at=now_taipei(), + ) + logger.info("resolve_db_updated", incident_id=incident_id) + except Exception as e: + logger.warning( + "resolve_db_update_failed", + incident_id=incident_id, + error=str(e), + ) + + return incident + + async def find_by_proposal_id(self, proposal_id: str) -> Incident | None: + """ + 根據 proposal_id 查找關聯的 Incident + + Phase 17: 從 Router 層遷移至 Service 層 + + Args: + proposal_id: 提案 ID (UUID 字串) + + Returns: + Incident | None: 找到的事件，未找到返回 None + """ + from uuid import UUID + + redis_client = get_redis() + + try: + target_uuid = UUID(proposal_id) + + async for key in redis_client.scan_iter( + match=f"{INCIDENT_KEY_PREFIX}INC-*", + count=100, + ): + data = await redis_client.get(key) + if data is None: + continue + + try: + # 方案 C: 正規化舊格式 Enum 值 + incident_dict = json.loads(data) + if "status" in incident_dict: + incident_dict["status"] = normalize_status(incident_dict["status"]) + if "severity" in incident_dict: + incident_dict["severity"] = normalize_severity(incident_dict["severity"]) + + for signal in incident_dict.get("signals", []): + if "severity" in signal: + signal["severity"] = normalize_severity(signal["severity"]) + + incident = Incident.model_validate(incident_dict) + + if target_uuid in incident.proposal_ids: + return incident + + except Exception as e: + logger.warning( + "incident_parse_error_in_find", + key=key, + error=str(e), + ) + continue + + return None + + except Exception as e: + logger.exception( + "find_by_proposal_id_error", + proposal_id=proposal_id, + error=str(e), + ) + return None + # ============================================================================= # Singleton diff --git a/apps/api/src/services/signal_producer.py b/apps/api/src/services/signal_producer.py new file mode 100644 index 000000000..fa58631bb --- /dev/null +++ b/apps/api/src/services/signal_producer.py @@ -0,0 +1,120 @@ +""" +Signal Producer Service - Phase 17 P0 Router 層違規修復 +====================================================== + +封裝 Redis Stream 的 Signal 生產邏輯，消除 Router 層直接存取 Redis。 + +功能: +- XADD 寫入 Redis Stream +- Trace Context 傳遞 + +符合 leWOOOgo 積木化規範: +- Router -> Service -> Redis +""" + +from dataclasses import dataclass +from typing import Any + +import structlog + +from src.core.redis_client import get_redis +from src.core.telemetry import get_trace_context +from src.utils.timezone import now_taipei + +logger = structlog.get_logger(__name__) + +# Redis Stream 配置 +SIGNAL_STREAM_KEY = "awoooi:signals" +SIGNAL_STREAM_MAXLEN = 10000 + + +@dataclass +class SignalData: + """Signal 資料結構""" + source: str + alert_name: str + severity: str + namespace: str + target: str + message: str + labels: dict[str, Any] | None = None + annotations: dict[str, Any] | None = None + + +class SignalProducerService: + """ + Signal 生產者服務 + + 封裝 Redis Stream 的 XADD 操作 + """ + + async def produce(self, signal: SignalData) -> str: + """ + 將 Signal 寫入 Redis Stream + + Phase 17: 從 Router 層遷移至 Service 層 + + 使用 Redis Streams (XADD) 實現非同步事件總線: + - MAXLEN ~10000: 限制 Stream 長度，自動裁剪舊訊息 + - *: 自動生成 Message ID + + Args: + signal: Signal 資料 + + Returns: + str: Redis Stream Message ID + """ + redis_client = get_redis() + + # 組裝 Signal 字典 (所有值必須是字串) + signal_dict = { + "source": signal.source, + "alert_name": signal.alert_name, + "severity": signal.severity, + "namespace": signal.namespace, + "target": signal.target, + "message": signal.message, + "labels": str(signal.labels or {}), + "annotations": str(signal.annotations or {}), + "received_at": now_taipei().isoformat(), + } + + # Phase 15.2: 注入 Trace Context + trace_ctx = get_trace_context() + if trace_ctx: + signal_dict["_trace_id"] = trace_ctx.get("trace_id", "") + signal_dict["_span_id"] = trace_ctx.get("span_id", "") + + # XADD 寫入 Stream + message_id = await redis_client.xadd( + SIGNAL_STREAM_KEY, + signal_dict, + maxlen=SIGNAL_STREAM_MAXLEN, + approximate=True, # ~MAXLEN 近似裁剪，效能更好 + ) + + logger.info( + "signal_produced", + message_id=message_id, + source=signal.source, + alert_name=signal.alert_name, + severity=signal.severity, + trace_id=trace_ctx.get("trace_id") if trace_ctx else None, + ) + + return message_id + + +# ============================================================================= +# Singleton +# ============================================================================= + +_signal_producer: SignalProducerService | None = None + + +def get_signal_producer() -> SignalProducerService: + """取得 SignalProducerService 實例 (Singleton)""" + global _signal_producer + if _signal_producer is None: + _signal_producer = SignalProducerService() + return _signal_producer diff --git a/apps/api/src/services/stats_service.py b/apps/api/src/services/stats_service.py new file mode 100644 index 000000000..bfcfaa76f --- /dev/null +++ b/apps/api/src/services/stats_service.py @@ -0,0 +1,109 @@ +""" +Stats Service - Phase 17 P0 Router 層違規修復 +============================================= + +封裝統計 API 的快取邏輯，消除 Router 層直接存取 Redis。 + +功能: +- 快取包裝器 (Redis) +- 統計計算 (透過 Repository) + +符合 leWOOOgo 積木化規範: +- Router -> Service -> Redis/Repository +""" + +import json +from typing import Any, Callable, Coroutine + +import structlog + +from src.core.redis_client import get_redis + +logger = structlog.get_logger(__name__) + +# 快取 TTL (秒) +STATS_CACHE_TTL = 300 # 5 分鐘 + + +class StatsService: + """ + 統計服務 + + 封裝統計 API 的快取邏輯 + """ + + async def get_cached_or_compute( + self, + cache_key: str, + compute_fn: Callable[[], Coroutine[Any, Any, dict[str, Any]]], + ttl: int = STATS_CACHE_TTL, + ) -> dict[str, Any]: + """ + 快取包裝器: 先查 Redis，沒有則計算並快取 + + Phase 17: 從 Router 層遷移至 Service 層 + + Args: + cache_key: Redis key + compute_fn: 計算函數 (async callable) + ttl: 快取時間 (秒) + + Returns: + 快取或計算結果 + """ + redis_client = get_redis() + + # 嘗試從快取取得 + try: + cached = await redis_client.get(cache_key) + if cached: + logger.debug("stats_cache_hit", key=cache_key) + return json.loads(cached) + except Exception as e: + logger.warning("stats_cache_read_error", key=cache_key, error=str(e)) + + # 計算結果 + result = await compute_fn() + + # 寫入快取 + try: + await redis_client.set(cache_key, json.dumps(result), ex=ttl) + logger.debug("stats_cache_set", key=cache_key, ttl=ttl) + except Exception as e: + logger.warning("stats_cache_write_error", key=cache_key, error=str(e)) + + return result + + async def invalidate_cache(self, cache_key: str) -> bool: + """ + 清除指定快取 + + Args: + cache_key: Redis key + + Returns: + 是否成功清除 + """ + redis_client = get_redis() + try: + await redis_client.delete(cache_key) + logger.info("stats_cache_invalidated", key=cache_key) + return True + except Exception as e: + logger.warning("stats_cache_invalidate_error", key=cache_key, error=str(e)) + return False + + +# ============================================================================= +# Singleton +# ============================================================================= + +_stats_service: StatsService | None = None + + +def get_stats_service() -> StatsService: + """取得 StatsService 實例 (Singleton)""" + global _stats_service + if _stats_service is None: + _stats_service = StatsService() + return _stats_service