"""HTTP router for MAF roundtable Conversation API at ``/api/maf-roundtable/conversations``.

Phase 1 (R1) scope:
- POST   /conversations          — create blank user conversation
- GET    /conversations          — list current user's history
- GET    /conversations/{id}     — public conversation DTO
- GET    /conversations/{id}/messages — paginated message history
- POST   /conversations/{id}/messages — atomic send + run start

Phase 2 (R2) adds:
- GET    /conversations/{id}/stream — SSE real-time event stream

User isolation: owner-only reads/writes. Admin diagnostic access is read-only.
Internal fields (owner_user_id, internal_session_id, active_run_id, thread ids)
are never returned.
"""

from __future__ import annotations

import asyncio
import json
import logging
from datetime import UTC, datetime
from typing import Any

from fastapi import APIRouter, HTTPException, Query, Request, status
from fastapi.responses import StreamingResponse

from app.maf_roundtable.conversation_event_hub import ConversationEventHub
from app.maf_roundtable.conversation_schemas import (
    ConversationListResponse,
    ConversationResponse,
    CreateConversationRequest,
    MessageListResponse,
    SendMessageRequest,
    SendMessageResponse,
)
from app.maf_roundtable.conversation_service import (
    ConversationConflictError,
    ConversationDisabledError,
    ConversationNoSessionConfigError,
    ConversationNotFoundError,
    ConversationServiceError,
    MafConversationService,
)

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/maf-roundtable", tags=["maf-roundtable-conversations"])

_SSE_HEADERS = {
    "Cache-Control": "no-cache",
    "Connection": "keep-alive",
    "X-Accel-Buffering": "no",
}


def _get_conversation_service(request: Request) -> MafConversationService:
    svc = getattr(request.app.state, "maf_conversation_service", None)
    if svc is None:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="MAF conversation service not available")
    return svc


def _get_event_hub(request: Request) -> ConversationEventHub | None:
    return getattr(request.app.state, "maf_conversation_event_hub", None)


def _user_ctx(request: Request) -> tuple[str, bool]:
    """Match the main MAF router's authenticated-user convention.

    ``AuthMiddleware`` stamps the resolved account on ``request.state.user``;
    it does not create ``user_id`` / ``is_admin`` attributes.  Reading those
    nonexistent attributes collapses every authenticated caller into the
    fallback account and breaks owner isolation.
    """
    user = getattr(request.state, "user", None)
    if user is None:
        return "default", False
    return str(user.id), getattr(user, "system_role", None) == "admin"


# ── REST endpoints (Phase 1) ──────────────────────────────────────────────


def _check_stream_ready(request: Request) -> None:
    stream_ready = getattr(request.app.state, "maf_conversation_stream_ready", True)
    if not stream_ready:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=getattr(request.app.state, "maf_conversation_stream_status", "stream_not_ready"),
        )


@router.post("/conversations", response_model=ConversationResponse, status_code=status.HTTP_201_CREATED)
async def create_conversation(body: CreateConversationRequest, request: Request):
    _check_stream_ready(request)
    svc = _get_conversation_service(request)
    user_id, _ = _user_ctx(request)
    try:
        return await svc.create_conversation(
            user_id=user_id,
            title=body.title.strip(),
            task_context=body.task_context,
        )
    except ConversationDisabledError:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="MAF roundtable feature is disabled",
        )


@router.get("/conversations", response_model=ConversationListResponse)
async def list_conversations(
    request: Request,
    limit: int = Query(default=50, ge=1, le=200),
    offset: int = Query(default=0, ge=0),
):
    svc = _get_conversation_service(request)
    user_id, _ = _user_ctx(request)
    conversations, total = await svc.list_conversations(user_id=user_id, limit=limit, offset=offset)
    return {"conversations": conversations, "total": total}


@router.get("/conversations/{conversation_id}", response_model=ConversationResponse)
async def get_conversation(conversation_id: str, request: Request):
    svc = _get_conversation_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        return await svc.get_conversation(conversation_id=conversation_id, user_id=user_id, is_admin=is_admin)
    except ConversationNotFoundError:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")


@router.get("/conversations/{conversation_id}/messages", response_model=MessageListResponse)
async def list_messages(
    conversation_id: str,
    request: Request,
    before_seq: int | None = Query(default=None, ge=1),
    limit: int = Query(default=50, ge=1, le=200),
):
    svc = _get_conversation_service(request)
    user_id, is_admin = _user_ctx(request)
    try:
        messages, last_stream_seq = await svc.list_messages(
            conversation_id=conversation_id, user_id=user_id, is_admin=is_admin,
            before_seq=before_seq, limit=limit,
        )
        return {"messages": messages, "last_stream_seq": last_stream_seq}
    except ConversationNotFoundError:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")


@router.post("/conversations/{conversation_id}/messages", response_model=SendMessageResponse, status_code=status.HTTP_201_CREATED)
async def send_message(conversation_id: str, body: SendMessageRequest, request: Request):
    _check_stream_ready(request)
    svc = _get_conversation_service(request)
    user_id, _ = _user_ctx(request)
    try:
        return await svc.send_message(conversation_id=conversation_id, user_id=user_id, content=body.content)
    except ConversationNotFoundError:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")
    except ConversationConflictError:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="A run is already active")
    except ConversationNoSessionConfigError:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="No trusted session configuration")
    except ConversationDisabledError:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="MAF roundtable feature is disabled")
    except ConversationServiceError as exc:
        raise HTTPException(status_code=exc.status_code, detail=exc.message)


# ── SSE endpoint (Phase 2 / R2) ───────────────────────────────────────────


# Event types that mark the end of a run's lifecycle. When the SSE stream
# encounters any of these in order, it sends the event and then closes the
# connection so the client doesn't sit idle receiving heartbeats.
_TERMINAL_EVENT_TYPES = frozenset({
    "run.completed",
    "run.error",
    "run.cancelled",
})


def _format_sse_event(event: dict[str, Any]) -> str:
    """Format one conversation event as an SSE frame per §4.3 contract.

    ``event:`` field carries the event type; ``data:`` carries the full
    envelope; ``id:`` carries the event_id for Last-Event-ID reconnection.
    """
    event_type = event.get("type", "message")
    event_id = event.get("event_id")
    payload = json.dumps(event, default=str, ensure_ascii=False)
    parts = [f"event: {event_type}", f"data: {payload}"]
    if event_id:
        parts.append(f"id: {event_id}")
    parts.append("")
    parts.append("")
    return "\n".join(parts)


def _format_heartbeat() -> str:
    """SSE comment frame — keeps the connection alive without producing a
    client-visible event."""
    return ": heartbeat\n\n"


def _is_terminal(event: dict[str, Any]) -> bool:
    return event.get("type") in _TERMINAL_EVENT_TYPES


def build_conversation_event_generator(
    *,
    service: MafConversationService,
    hub: ConversationEventHub,
    request: Request,
    conversation_id: str,
    after_seq: int = 0,
) -> Any:
    """Extracted async generator for SSE streaming — testable without HTTP.

    The ordering is:

    1. **Reconcile** any missed projections for the active run (P0-1 auto).
    2. **Open subscription** (register before any DB query — P0-1 fix).
    3. **Replay** from DB: ``list_events(after_seq)``.
    4. **Stabilise**: re-query while the hub generation advanced during replay.
    5. **Snapshot** marker.
    6. **Real-time**: ``wait_for_change`` → query DB → yield events.

    On terminal events the stream sends the event and closes (P0-2 fix).
    Heartbeat every ``hub.heartbeat_interval`` seconds.
    """
    store = service._store  # noqa: SLF001

    async def _gen():
        try:
            active_run_id = await _get_active_run_id(store, conversation_id)
            if active_run_id:
                try:
                    await service.reconcile_conversation_events(run_id=active_run_id)
                except Exception:  # noqa: BLE001
                    logger.warning("SSE connect reconcile failed conv=%s", conversation_id, exc_info=True)
        except Exception:  # noqa: BLE001
            logger.warning("SSE connect reconcile lookup failed conv=%s", conversation_id, exc_info=True)

        sub = hub.open_subscription(conversation_id)
        last_seq = after_seq
        try:
            try:
                replay_events = await store.conversation_events.list_events(
                    conversation_id, after_seq=last_seq,
                )
                for evt in replay_events:
                    if await request.is_disconnected():
                        return
                    yield _format_sse_event(evt)
                    last_seq = max(last_seq, int(evt.get("seq", last_seq)))
                    if _is_terminal(evt):
                        return
            except Exception:
                logger.exception("SSE replay error conv=%s", conversation_id)
                yield _format_sse_event({
                    "seq": last_seq,
                    "event_id": f"replay_error_{last_seq}",
                    "conversation_id": conversation_id,
                    "type": "run.error",
                    "occurred_at": datetime.now(UTC).isoformat(),
                    "data": {"error": "replay_failed"},
                })
                return

            while sub.has_pending:
                sub.mark_seen()
                late_events = await store.conversation_events.list_events(
                    conversation_id, after_seq=last_seq,
                )
                if not late_events:
                    break
                for evt in late_events:
                    if await request.is_disconnected():
                        return
                    yield _format_sse_event(evt)
                    last_seq = max(last_seq, int(evt.get("seq", last_seq)))
                    if _is_terminal(evt):
                        return

            snapshot = {
                "seq": last_seq,
                "event_id": f"snapshot_{conversation_id}_{last_seq}",
                "conversation_id": conversation_id,
                "type": "conversation.snapshot",
                "occurred_at": datetime.now(UTC).isoformat(),
                "data": {"last_stream_seq": last_seq},
            }
            if not await request.is_disconnected():
                yield _format_sse_event(snapshot)

            heartbeat_interval = hub.heartbeat_interval
            while True:
                if hub.is_closed:
                    return
                if await request.is_disconnected():
                    return

                changed = await sub.wait_for_change(timeout=heartbeat_interval)
                if not changed:
                    if hub.is_closed:
                        return
                    yield _format_heartbeat()
                    continue

                new_events = await store.conversation_events.list_events(
                    conversation_id, after_seq=last_seq,
                )
                for evt in new_events:
                    if await request.is_disconnected():
                        return
                    yield _format_sse_event(evt)
                    last_seq = max(last_seq, int(evt.get("seq", last_seq)))
                    if _is_terminal(evt):
                        return
        except asyncio.CancelledError:
            logger.debug("SSE stream cancelled conv=%s", conversation_id)
            raise
        except Exception:
            logger.exception("SSE stream error conv=%s", conversation_id)
            return
        finally:
            sub.close()

    return _gen()


async def _get_active_run_id(store: Any, conversation_id: str) -> str | None:
    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow
    async with store.session_factory() as db:
        row = (
            await db.execute(
                select(MafRoundtableConversationRow.active_run_id).where(
                    MafRoundtableConversationRow.id == conversation_id,
                )
            )
        ).first()
        return row[0] if row and row[0] else None


@router.get("/conversations/{conversation_id}/stream")
async def stream_conversation(
    conversation_id: str,
    request: Request,
    after_seq: int = Query(default=0, ge=0),
):
    """SSE stream for one conversation (§4.2)."""
    svc = _get_conversation_service(request)
    hub = _get_event_hub(request)
    user_id, is_admin = _user_ctx(request)

    if hub is None:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="SSE event hub not configured",
        )

    stream_ready = getattr(request.app.state, "maf_conversation_stream_ready", True)
    if not stream_ready:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=getattr(request.app.state, "maf_conversation_stream_status", "stream_not_ready"),
        )

    try:
        await svc.get_conversation(
            conversation_id=conversation_id, user_id=user_id, is_admin=is_admin,
        )
    except ConversationNotFoundError:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")

    return StreamingResponse(
        build_conversation_event_generator(
            service=svc,
            hub=hub,
            request=request,
            conversation_id=conversation_id,
            after_seq=after_seq,
        ),
        media_type="text/event-stream",
        headers=_SSE_HEADERS,
    )