"""
Signal Worker - Redis Streams Consumer
=======================================
Phase 6.1: Event Bus Implementation
Phase 15.2: Redis Trace Context Propagation (2026-03-26)

功能:
- XREADGROUP 消費 stream:awoooi_signals
- Signal → Incident 聚合邏輯 (Phase 6.3 實作)
- 失敗重試 + ACK 機制
- Graceful Shutdown
- **Phase 15.2**: Trace Context 還原 (零斷鏈觀測)

Redis Streams 概念:
- Stream: stream:awoooi_signals (訊息佇列)
- Consumer Group: awoooi_workers (消費者群組)
- Consumer: worker_{hostname} (單一消費者)

Trace Context 傳遞 (Phase 15.2):
- Producer (webhooks.py) 寫入 _trace_id, _span_id 到 Redis
- Consumer (此檔案) 還原 Context，建立子 Span
- 實現 API → Redis → Worker 完整追蹤鏈

統帥鐵律:
- 使用 XREADGROUP 確保訊息只被處理一次
- 處理完成後必須 XACK
- 失敗訊息進入 Pending List，需定期清理
"""

import asyncio
import socket
from typing import Any

import structlog

from src.core.redis_client import get_redis, get_worker_redis
from src.core.telemetry import restore_trace_context
from src.services.incident_engine import get_incident_engine

logger = structlog.get_logger(__name__)


# =============================================================================
# Constants
# =============================================================================

# 2026-03-27 ogt: 統一 Stream Key 格式 (P0 修復)
STREAM_KEY = "awoooi:signals"
CONSUMER_GROUP = "awoooi_workers"
CONSUMER_NAME = f"worker_{socket.gethostname()}"

# 每次讀取的訊息數量
BATCH_SIZE = 10
# 讀取超時 (毫秒) - 0 表示阻塞等待
BLOCK_MS = 5000
# 失敗重試上限
MAX_RETRIES = 3

# =============================================================================
# ADR-038/039: 安全網配置 (Wave 1)
# =============================================================================
# XCLAIM: 閒置訊息回收閾值（毫秒）
PENDING_IDLE_MS = 60_000  # 1 分鐘無 ACK 則可被其他 Worker 回收
# Active Sweeper: 掃描間隔（秒）
SWEEP_INTERVAL_S = 30
# Graceful Shutdown: 最大等待時間（秒）
GRACEFUL_SHUTDOWN_TIMEOUT_S = 75  # K8s terminationGracePeriodSeconds: 90


# =============================================================================
# Signal Worker
# =============================================================================

class SignalWorker:
    """
    Redis Streams 訊號消費者

    職責:
    1. 從 stream:awoooi_signals 讀取訊號
    2. 將訊號聚合為 Incident (Phase 6.3)
    3. 更新 Working Memory (Redis)
    4. 觸發決策引擎 (Phase 6.4)

    使用方式:
        worker = SignalWorker()
        await worker.start()  # 啟動消費循環
        await worker.stop()   # 優雅關閉
    """

    def __init__(self) -> None:
        self._running = False
        self._task: asyncio.Task | None = None
        self._sweeper_task: asyncio.Task | None = None  # ADR-038: Active Sweeper

    async def _ensure_consumer_group(self) -> None:
        """
        確保 Consumer Group 存在

        XGROUP CREATE 如果 Group 已存在會報錯，
        因此使用 MKSTREAM 選項並忽略 BUSYGROUP 錯誤。
        """
        redis_client = get_redis()
        try:
            # MKSTREAM: 如果 Stream 不存在則建立
            await redis_client.xgroup_create(
                STREAM_KEY,
                CONSUMER_GROUP,
                id="0",  # 從頭開始消費
                mkstream=True,
            )
            logger.info(
                "consumer_group_created",
                stream=STREAM_KEY,
                group=CONSUMER_GROUP,
            )
        except Exception as e:
            # BUSYGROUP: Group 已存在，忽略
            if "BUSYGROUP" in str(e):
                logger.debug("consumer_group_exists", group=CONSUMER_GROUP)
            else:
                raise

    async def start(self) -> None:
        """
        啟動消費循環

        在背景執行，不阻塞主執行緒。
        """
        if self._running:
            logger.warning("signal_worker_already_running")
            return

        await self._ensure_consumer_group()

        self._running = True
        self._task = asyncio.create_task(self._consume_loop())
        self._sweeper_task = asyncio.create_task(self._sweep_loop())  # ADR-038
        logger.info(
            "signal_worker_started",
            stream=STREAM_KEY,
            group=CONSUMER_GROUP,
            consumer=CONSUMER_NAME,
            sweep_interval_s=SWEEP_INTERVAL_S,
        )

    async def stop(self) -> None:
        """
        優雅關閉 (ADR-038 Wave 1 強化)

        等待當前處理完成後停止。
        超時時間: 75 秒（搭配 K8s terminationGracePeriodSeconds: 90）
        """
        if not self._running:
            return

        logger.info(
            "signal_worker_stopping",
            timeout_s=GRACEFUL_SHUTDOWN_TIMEOUT_S,
        )
        self._running = False

        # 停止 Sweeper
        if self._sweeper_task:
            self._sweeper_task.cancel()
            try:
                await self._sweeper_task
            except asyncio.CancelledError:
                pass

        # 等待主消費循環完成
        if self._task:
            try:
                await asyncio.wait_for(
                    self._task,
                    timeout=GRACEFUL_SHUTDOWN_TIMEOUT_S,
                )
            except TimeoutError:
                logger.warning(
                    "signal_worker_stop_timeout",
                    timeout_s=GRACEFUL_SHUTDOWN_TIMEOUT_S,
                )
                self._task.cancel()
            except asyncio.CancelledError:
                pass

        logger.info("signal_worker_stopped")

    async def _consume_loop(self) -> None:
        """
        主消費循環

        XREADGROUP 阻塞等待新訊息，處理後 XACK。

        統帥鐵律 2026-03-23:
        - 使用 Worker 專屬 Redis 連線 (無超時限制)
        - 絕對禁止使用 API 共用的短超時連線
        """
        redis_client = get_worker_redis()  # Worker 專屬長連線

        while self._running:
            try:
                # XREADGROUP: 從 Consumer Group 讀取訊息
                # >: 只讀取新訊息 (不包含 Pending List)
                messages = await redis_client.xreadgroup(
                    groupname=CONSUMER_GROUP,
                    consumername=CONSUMER_NAME,
                    streams={STREAM_KEY: ">"},
                    count=BATCH_SIZE,
                    block=BLOCK_MS,
                )

                if not messages:
                    # 超時，沒有新訊息
                    continue

                # messages 格式: [[stream_name, [(id, data), ...]]]
                for _stream_name, entries in messages:
                    for message_id, data in entries:
                        await self._process_signal(message_id, data)

            except asyncio.CancelledError:
                logger.info("signal_worker_cancelled")
                break
            except Exception as e:
                logger.exception("signal_worker_error", error=str(e))
                # 避免無限快速重試
                await asyncio.sleep(1.0)

    async def _sweep_loop(self) -> None:
        """
        Active Sweeper: 定期掃描並回收閒置的 Pending 訊息

        ADR-038 Wave 1: 使用 XCLAIM 回收其他 Worker 死亡或卡住的訊息

        流程:
        1. XPENDING 取得 Pending List 摘要
        2. XPENDING 取得具體閒置訊息 (idle > PENDING_IDLE_MS)
        3. XCLAIM 回收訊息到本 Consumer
        4. 重新處理或強制 ACK (超過 MAX_RETRIES)
        """
        redis_client = get_worker_redis()

        while self._running:
            try:
                # 等待下次掃描
                await asyncio.sleep(SWEEP_INTERVAL_S)

                if not self._running:
                    break

                # 1. 取得 Pending 摘要
                pending_info = await redis_client.xpending(
                    STREAM_KEY,
                    CONSUMER_GROUP,
                )

                if not pending_info or pending_info.get("pending", 0) == 0:
                    continue  # 沒有 Pending 訊息

                pending_count = pending_info.get("pending", 0)
                logger.debug(
                    "sweep_pending_check",
                    pending_count=pending_count,
                )

                # 2. 取得具體的 Pending 訊息（最多 10 條）
                pending_messages = await redis_client.xpending_range(
                    STREAM_KEY,
                    CONSUMER_GROUP,
                    min="-",
                    max="+",
                    count=10,
                )

                for msg in pending_messages:
                    message_id = msg.get("message_id")
                    idle_ms = msg.get("time_since_delivered", 0)
                    delivery_count = msg.get("times_delivered", 0)

                    # 只處理閒置超過閾值的訊息
                    if idle_ms < PENDING_IDLE_MS:
                        continue

                    logger.info(
                        "sweep_claiming_message",
                        message_id=message_id,
                        idle_ms=idle_ms,
                        delivery_count=delivery_count,
                    )

                    # 3. XCLAIM 回收訊息
                    claimed = await redis_client.xclaim(
                        STREAM_KEY,
                        CONSUMER_GROUP,
                        CONSUMER_NAME,
                        min_idle_time=PENDING_IDLE_MS,
                        message_ids=[message_id],
                    )

                    if not claimed:
                        continue

                    # 4. 處理回收的訊息
                    for claimed_id, claimed_data in claimed:
                        if delivery_count >= MAX_RETRIES:
                            # 超過重試上限，強制 ACK 並記錄
                            logger.warning(
                                "sweep_force_ack_max_retries",
                                message_id=claimed_id,
                                delivery_count=delivery_count,
                            )
                            await redis_client.xack(
                                STREAM_KEY,
                                CONSUMER_GROUP,
                                claimed_id,
                            )
                        else:
                            # 重新處理
                            await self._process_signal(claimed_id, claimed_data)

            except asyncio.CancelledError:
                logger.info("sweep_loop_cancelled")
                break
            except Exception as e:
                logger.exception("sweep_loop_error", error=str(e))
                await asyncio.sleep(5.0)  # 錯誤後等待

    async def _process_signal(self, message_id: str, data: dict[str, Any]) -> None:
        """
        處理單一訊號

        Phase 6.3 核心邏輯:
        1. 訊號去重 (fingerprint)
        2. 訊號聚合 (30分鐘時間窗口 + 服務關聯)
        3. Incident 建立/更新 (聚合到同一 Incident)
        4. GraphRAG 爆炸半徑分析
        5. 雙層持久化 (Redis + PostgreSQL)

        Phase 15.2: Trace Context 還原
        - 從 Redis 訊息提取 _trace_id, _span_id
        - 建立子 Span 繼承原始 Trace，實現零斷鏈觀測
        """
        redis_client = get_redis()

        # =================================================================
        # Phase 15.2: 提取 Trace Context (從 Producer 注入的欄位)
        # =================================================================
        trace_context = None
        trace_id = data.pop("_trace_id", None)  # pop 避免污染 signal data
        span_id = data.pop("_span_id", None)

        if trace_id:
            trace_context = {
                "trace_id": trace_id,
                "span_id": span_id or "",
            }

        # =================================================================
        # 在還原的 Trace Context 中處理訊號
        # =================================================================
        with restore_trace_context(trace_context) as span:
            try:
                # 設置 Span 屬性 (用於 SignOz 搜尋)
                span.set_attribute("messaging.system", "redis_streams")
                span.set_attribute("messaging.destination", STREAM_KEY)
                span.set_attribute("messaging.message_id", message_id)
                span.set_attribute("signal.source", data.get("source", "unknown"))
                span.set_attribute("signal.alert_name", data.get("alert_name", "unknown"))
                span.set_attribute("signal.severity", data.get("severity", "unknown"))

                logger.info(
                    "signal_received",
                    message_id=message_id,
                    source=data.get("source", "unknown"),
                    alert_name=data.get("alert_name", "unknown"),
                    severity=data.get("severity", "unknown"),
                    namespace=data.get("namespace", "default"),
                    target=data.get("target", "unknown"),
                    trace_restored=trace_context is not None,
                )

                # Phase 6.3: 使用 IncidentEngine 處理訊號
                # - 自動聚合相關告警到同一 Incident
                # - GraphRAG 分析爆炸半徑
                # - 雙層持久化
                engine = get_incident_engine()
                incident = await engine.process_signal(data)

                if incident:
                    # 記錄 Incident 到 Span
                    span.set_attribute("incident.id", incident.incident_id)
                    span.set_attribute("incident.severity", incident.severity.value)
                    span.set_attribute("incident.signal_count", len(incident.signals))

                    logger.info(
                        "signal_processed_by_engine",
                        message_id=message_id,
                        incident_id=incident.incident_id,
                        severity=incident.severity.value,
                        signal_count=len(incident.signals),
                        affected_services=incident.affected_services,
                        persisted_to_pg=getattr(incident, "persisted_to_pg", False),  # 2026-04-01 ogt: BrainIncident 無此欄位 (ADR-046 P2-01)
                    )
                else:
                    span.set_attribute("signal.processing_failed", True)
                    logger.warning(
                        "signal_processing_failed",
                        message_id=message_id,
                        signal_data=data,
                    )

                # ACK: 確認訊息已處理
                await redis_client.xack(STREAM_KEY, CONSUMER_GROUP, message_id)
                span.set_attribute("messaging.acked", True)

                logger.debug("signal_acked", message_id=message_id)

            except Exception as e:
                span.set_attribute("error", True)
                span.set_attribute("error.message", str(e))
                logger.exception(
                    "signal_process_error",
                    message_id=message_id,
                    error=str(e),
                )
                # 不 ACK，訊息會留在 Pending List
                # Phase 6.3 將實作 Pending List 清理機制


# =============================================================================
# Singleton
# =============================================================================

_signal_worker: SignalWorker | None = None


async def init_signal_worker() -> SignalWorker:
    """
    初始化並啟動 Signal Worker

    統帥鐵律: 在 Lifespan 啟動時調用
    """
    global _signal_worker

    if _signal_worker is not None:
        return _signal_worker

    _signal_worker = SignalWorker()
    await _signal_worker.start()
    return _signal_worker


async def close_signal_worker() -> None:
    """
    關閉 Signal Worker

    統帥鐵律: 在 Lifespan 關閉時調用
    """
    global _signal_worker

    if _signal_worker is not None:
        await _signal_worker.stop()
        _signal_worker = None


def get_signal_worker() -> SignalWorker:
    """
    取得 Signal Worker 實例

    Raises:
        RuntimeError: 若 Worker 未初始化
    """
    if _signal_worker is None:
        raise RuntimeError(
            "Signal worker not initialized. Call init_signal_worker() first."
        )
    return _signal_worker


# =============================================================================
# Standalone Entry Point (for K8s Worker Deployment)
# =============================================================================

async def _write_health_files() -> None:
    """
    Write health check files for K8s probes.
    """
    from pathlib import Path

    Path("/tmp/worker-healthy").touch()
    Path("/tmp/worker-ready").touch()
    logger.info("health_files_written")


async def _heartbeat_loop(shutdown_event: asyncio.Event) -> None:
    """
    心跳循環：定期更新健康檢查文件的時間戳

    長期解決方案 (2026-03-28):
    - 每 30 秒 touch 健康文件，確保 mtime 更新
    - K8s liveness probe 可檢查 mtime 是否在 60 秒內
    - 防止 Worker 卡住但健康文件仍存在的假陽性

    @author Claude Code (首席架構師)
    @version 1.0.0
    @date 2026-03-28 (台北時間)
    """
    from pathlib import Path

    HEARTBEAT_INTERVAL = 30  # 秒

    while not shutdown_event.is_set():
        try:
            Path("/tmp/worker-healthy").touch()
            logger.debug("heartbeat_tick")
        except Exception as e:
            logger.warning("heartbeat_error", error=str(e))

        # 等待下次心跳或收到關閉信號
        try:
            await asyncio.wait_for(
                shutdown_event.wait(),
                timeout=HEARTBEAT_INTERVAL
            )
            break  # 收到關閉信號
        except TimeoutError:
            continue  # 超時，繼續下次心跳


async def _main() -> None:
    """
    Standalone worker main function.

    用於 K8s Deployment 直接執行:
    python -m src.workers.signal_worker
    """
    import signal
    import sys

    # Initialize settings first (loads env vars)
    from src.core.config import settings  # noqa: F401
    from src.core.redis_client import (
        close_redis_pool,
        close_worker_redis_pool,
        init_redis_pool,
        init_worker_redis_pool,
    )
    from src.db.base import close_db, init_db

    logger.info(
        "signal_worker_standalone_starting",
        environment=settings.ENVIRONMENT,
        redis_url=settings.REDIS_URL.split("@")[-1] if settings.REDIS_URL else "N/A",
        database_url=settings.DATABASE_URL.split("@")[-1] if settings.DATABASE_URL else "N/A",
    )

    # Initialize Redis (API pool + Worker 專屬長連線池)
    await init_redis_pool()
    await init_worker_redis_pool()  # Worker 專屬，無超時限制

    # Initialize PostgreSQL (Episodic Memory) - 確保 incidents 表存在
    try:
        await init_db()
        logger.info("postgresql_initialized", status="ok")
    except Exception as e:
        logger.error(
            "postgresql_init_failed",
            error=str(e),
            message="Episodic Memory (DB) will be unavailable - incidents won't persist",
        )

    # Write health files for K8s probes
    await _write_health_files()

    # Initialize and start worker
    worker = await init_signal_worker()

    # Setup graceful shutdown
    shutdown_event = asyncio.Event()

    def _shutdown_handler(signum: int, frame: object) -> None:
        logger.info("shutdown_signal_received", signal=signum)
        shutdown_event.set()

    signal.signal(signal.SIGTERM, _shutdown_handler)
    signal.signal(signal.SIGINT, _shutdown_handler)

    # 啟動心跳循環 (長期解決方案 - 定期更新健康文件 mtime)
    heartbeat_task = asyncio.create_task(_heartbeat_loop(shutdown_event))

    # Wait for shutdown signal
    await shutdown_event.wait()

    # 停止心跳
    heartbeat_task.cancel()
    try:
        await heartbeat_task
    except asyncio.CancelledError:
        pass

    # Graceful shutdown
    logger.info("signal_worker_shutting_down")
    await worker.stop()
    await close_worker_redis_pool()  # 關閉 Worker 專屬連線
    await close_redis_pool()
    await close_db()  # 關閉 PostgreSQL 連線池

    # Remove health files
    from pathlib import Path
    Path("/tmp/worker-healthy").unlink(missing_ok=True)
    Path("/tmp/worker-ready").unlink(missing_ok=True)

    logger.info("signal_worker_shutdown_complete")
    sys.exit(0)


if __name__ == "__main__":
    asyncio.run(_main())
