"""SSE transport for observable Builder chat turns."""

from __future__ import annotations

import json
import logging
from collections.abc import Callable, Iterator
from queue import Queue
from threading import Thread
from typing import Any

from fastapi.responses import StreamingResponse

from .fsm import FSMError
from .models import ActionResponse


BuilderOperation = Callable[[], ActionResponse]
PHASE_LABELS = {
    "suggesting_next": "正在判断下一步建议",
    "structuring": "正在整理信息",
    "checking_gaps": "正在检查可补充的信息",
    "rewriting": "正在生成候选改写",
    "saving": "正在写入简历",
}


def stream_builder_message(operation: BuilderOperation) -> StreamingResponse:
    events: Queue[tuple[str, dict[str, Any]] | None] = Queue()

    def emit(event: str, data: dict[str, Any] | None = None) -> None:
        events.put((event, data or {}))

    def worker() -> None:
        try:
            result = operation()
            _emit_statuses(emit, tuple(result.builder_stream_phases))
            for chunk in _chunks(str((result.turn.content if result.turn else "") or "")):
                emit("delta", {"text": chunk})
            emit("complete", result.model_dump(mode="json"))
        except FSMError as exc:
            emit("error", {"code": exc.code, "message": exc.message, "status_code": exc.status_code})
        except Exception as exc:  # pragma: no cover - defensive transport boundary
            logging.getLogger(__name__).exception("builder SSE operation failed")
            emit("error", {"code": "builder_stream_failed", "message": "Resume assistant stream failed. Please retry.", "status_code": 502, "reason_code": type(exc).__name__})
        finally:
            events.put(None)

    def generate() -> Iterator[str]:
        thread = Thread(target=worker, name="resume-builder-sse", daemon=True)
        thread.start()
        while True:
            item = events.get()
            if item is None:
                break
            event, data = item
            yield _frame(event, data)

    return StreamingResponse(generate(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


def _emit_statuses(emit: Callable[[str, dict[str, Any]], None], phases: tuple[str, ...]) -> None:
    for phase in phases:
        emit("status", {"phase": phase, "label": PHASE_LABELS[phase]})


def _chunks(text: str, size: int = 24) -> Iterator[str]:
    if not text:
        return
    for index in range(0, len(text), size):
        yield text[index : index + size]


def _frame(event: str, data: dict[str, Any]) -> str:
    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False, separators=(',', ':'))}\n\n"