"""Centralized accessors for singleton objects stored on ``app.state``.

**Getters** (used by routers): raise 503 when a required dependency is
missing, except ``get_store`` which returns ``None``.

Initialization is handled directly in ``app.py`` via :class:`AsyncExitStack`.
"""

from __future__ import annotations

from collections.abc import AsyncGenerator, Callable
from contextlib import AsyncExitStack, asynccontextmanager
from typing import TYPE_CHECKING, TypeVar, cast

from fastapi import FastAPI, HTTPException, Request
from langgraph.types import Checkpointer

from deerflow.config.app_config import AppConfig
from deerflow.persistence.feedback import FeedbackRepository
from deerflow.runtime import RunContext, RunManager, StreamBridge
from deerflow.runtime.events.store.base import RunEventStore
from deerflow.runtime.runs.store.base import RunStore

if TYPE_CHECKING:
    from app.gateway.auth.local_provider import LocalAuthProvider
    from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
    from deerflow.persistence.agents import AgentStore
    from deerflow.persistence.notifications import NotificationStore
    from deerflow.persistence.scheduled_tasks import ScheduledTaskStore
    from deerflow.persistence.skills import SkillStore
    from deerflow.persistence.tags import TagStore
    from deerflow.persistence.thread_meta.base import ThreadMetaStore


T = TypeVar("T")


def get_config(request: Request) -> AppConfig:
    """Return the app-scoped ``AppConfig`` stored on ``app.state``."""
    config = getattr(request.app.state, "config", None)
    if config is None:
        raise HTTPException(status_code=503, detail="Configuration not available")
    return config


@asynccontextmanager
async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
    """Bootstrap and tear down all LangGraph runtime singletons.

    Usage in ``app.py``::

        async with langgraph_runtime(app):
            yield
    """
    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
    from deerflow.runtime import make_store, make_stream_bridge
    from deerflow.runtime.checkpointer.async_provider import make_checkpointer
    from deerflow.runtime.events.store import make_run_event_store

    async with AsyncExitStack() as stack:
        config = getattr(app.state, "config", None)
        if config is None:
            raise RuntimeError("langgraph_runtime() requires app.state.config to be initialized")

        app.state.stream_bridge = await stack.enter_async_context(make_stream_bridge(config))

        # Initialize persistence engine BEFORE checkpointer so that
        # auto-create-database logic runs first (postgres backend).
        await init_engine_from_config(config.database)

        app.state.checkpointer = await stack.enter_async_context(make_checkpointer(config))
        app.state.store = await stack.enter_async_context(make_store(config))

        # Initialize repositories — one get_session_factory() call for all.
        sf = get_session_factory()
        if sf is not None:
            from deerflow.persistence.feedback import FeedbackRepository
            from deerflow.persistence.run import RunRepository

            app.state.run_store = RunRepository(sf)
            app.state.feedback_repo = FeedbackRepository(sf)
        else:
            from deerflow.runtime.runs.store.memory import MemoryRunStore

            app.state.run_store = MemoryRunStore()
            app.state.feedback_repo = None

        from deerflow.persistence.thread_meta import make_thread_store

        app.state.thread_store = make_thread_store(sf, app.state.store)

        from deerflow.persistence.scheduled_tasks import make_scheduled_task_store

        app.state.scheduled_task_store = make_scheduled_task_store(sf)

        from deerflow.persistence.llm_metrics import make_llm_metrics_store

        app.state.llm_metrics_store = make_llm_metrics_store(sf)

        from deerflow.persistence.tool_metrics import make_tool_metrics_store

        app.state.tool_metrics_store = make_tool_metrics_store(sf)

        from deerflow.persistence.agents import make_agent_store

        app.state.agent_store = make_agent_store(sf)

        from deerflow.persistence.skills import make_skill_store

        app.state.skill_store = make_skill_store(sf)

        from deerflow.persistence.notifications import (
            make_notification_store,
            set_notification_store,
        )

        app.state.notification_store = make_notification_store(sf)
        # Make the sender helper (harness layer) able to write without going
        # through FastAPI request context — used by cascade.py.
        set_notification_store(app.state.notification_store)
        from deerflow.persistence.recommended_questions import make_recommended_question_store

        app.state.recommended_question_store = make_recommended_question_store(sf)

        from deerflow.persistence.tags import make_tag_store

        app.state.tag_store = make_tag_store(sf)

        from deerflow.persistence.thread_shares import make_thread_share_store

        app.state.thread_share_store = make_thread_share_store(sf)

        from deerflow.persistence.ai_writing_sessions import (
            make_ai_writing_session_store,
            register_default_store as _register_ai_writing_default_store,
        )

        app.state.ai_writing_session_store = make_ai_writing_session_store(sf)
        # 让 LangGraph Server 路径下的 AI 写作节点能跨 app/harness 边界拿到 repo。
        # 老 /start /stream /resume 路由仍走 request.app.state，互不影响。
        _register_ai_writing_default_store(app.state.ai_writing_session_store)

        from deerflow.persistence.article_types import make_article_type_store

        app.state.article_type_store = make_article_type_store(sf)

        from deerflow.persistence.memory_config import make_user_memory_config_store

        app.state.memory_config_store = make_user_memory_config_store(sf)

        from deerflow.persistence.user_prompts import make_user_prompt_store

        app.state.user_prompt_store = make_user_prompt_store(sf)

        from deerflow.persistence.roundtable_drafts import make_roundtable_draft_store

        app.state.roundtable_draft_store = make_roundtable_draft_store(sf)

        # Run event store (has its own factory with config-driven backend selection)
        run_events_config = getattr(config, "run_events", None)
        app.state.run_event_store = make_run_event_store(run_events_config)

        # RunManager with store backing for persistence
        app.state.run_manager = RunManager(store=app.state.run_store)

        try:
            yield
        finally:
            from deerflow.persistence.notifications import set_notification_store

            set_notification_store(None)
            await close_engine()


# ---------------------------------------------------------------------------
# Getters – called by routers per-request
# ---------------------------------------------------------------------------


def _require(attr: str, label: str) -> Callable[[Request], T]:
    """Create a FastAPI dependency that returns ``app.state.<attr>`` or 503."""

    def dep(request: Request) -> T:
        val = getattr(request.app.state, attr, None)
        if val is None:
            raise HTTPException(status_code=503, detail=f"{label} not available")
        return cast(T, val)

    dep.__name__ = dep.__qualname__ = f"get_{attr}"
    return dep


get_stream_bridge: Callable[[Request], StreamBridge] = _require("stream_bridge", "Stream bridge")
get_run_manager: Callable[[Request], RunManager] = _require("run_manager", "Run manager")
get_checkpointer: Callable[[Request], Checkpointer] = _require("checkpointer", "Checkpointer")
get_run_event_store: Callable[[Request], RunEventStore] = _require("run_event_store", "Run event store")
get_feedback_repo: Callable[[Request], FeedbackRepository] = _require("feedback_repo", "Feedback")
get_run_store: Callable[[Request], RunStore] = _require("run_store", "Run store")


def get_store(request: Request):
    """Return the global store (may be ``None`` if not configured)."""
    return getattr(request.app.state, "store", None)


def get_thread_store(request: Request) -> ThreadMetaStore:
    """Return the thread metadata store (SQL or memory-backed)."""
    val = getattr(request.app.state, "thread_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Thread metadata store not available")
    return val


def get_scheduled_task_store(request: Request) -> ScheduledTaskStore:
    """Return the scheduled-task store."""
    val = getattr(request.app.state, "scheduled_task_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Scheduled task store not available")
    return val


def get_llm_metrics_store(request: Request):
    """Return the LLM metrics store, or None when not configured."""
    return getattr(request.app.state, "llm_metrics_store", None)


def get_tool_metrics_store(request: Request):
    """Return the tool-call metrics store, or None when not configured."""
    return getattr(request.app.state, "tool_metrics_store", None)


def get_agent_store(request: Request) -> AgentStore:
    """Return the custom-agent store."""
    val = getattr(request.app.state, "agent_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Agent store not available")
    return val


def get_skill_store(request: Request) -> SkillStore:
    """Return the custom-skill store."""
    val = getattr(request.app.state, "skill_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Skill store not available")
    return val


def get_notification_store(request: Request) -> NotificationStore:
    """Return the in-app notification store."""
    val = getattr(request.app.state, "notification_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Notification store not available")
    return val


def get_tag_store(request: Request) -> TagStore:
    """Return the tag store."""
    val = getattr(request.app.state, "tag_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Tag store not available")
    return val


def get_thread_share_store(request: Request):
    """Return the conversation share-link store."""
    val = getattr(request.app.state, "thread_share_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Thread share store not available")
    return val


def get_ai_writing_session_store(request: Request):
    """Return the AI writing session store, or None when not configured."""
    return getattr(request.app.state, "ai_writing_session_store", None)


def get_article_type_store(request: Request):
    """Return the article type store, or None when not configured."""
    return getattr(request.app.state, "article_type_store", None)


def get_memory_config_store(request: Request):
    """Return the per-user memory config store."""
    val = getattr(request.app.state, "memory_config_store", None)
    if val is None:
        raise HTTPException(status_code=503, detail="Memory config store not available")
    return val


def get_run_context(request: Request) -> RunContext:
    """Build a :class:`RunContext` from ``app.state`` singletons.

    Returns a *base* context with infrastructure dependencies.
    """
    config = get_config(request)
    return RunContext(
        checkpointer=get_checkpointer(request),
        store=get_store(request),
        event_store=get_run_event_store(request),
        run_events_config=getattr(config, "run_events", None),
        thread_store=get_thread_store(request),
        app_config=config,
    )


# ---------------------------------------------------------------------------
# Auth helpers (used by authz.py and auth middleware)
# ---------------------------------------------------------------------------

# Cached singletons to avoid repeated instantiation per request
_cached_local_provider: LocalAuthProvider | None = None
_cached_repo: SQLiteUserRepository | None = None


def get_request_access_token(request: Request) -> str | None:
    """Extract access token from cookie or Authorization header.

    Priority:
    1. ``access_token`` cookie (browser session mode)
    2. ``Authorization: Bearer <token>`` header (API client mode)
    """
    cookie_token = request.cookies.get("access_token")
    if cookie_token:
        return cookie_token

    authorization = request.headers.get("authorization", "").strip()
    if not authorization:
        return None

    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer":
        return None

    token = token.strip()
    return token or None


def get_local_provider() -> LocalAuthProvider:
    """Get or create the cached LocalAuthProvider singleton.

    Must be called after ``init_engine_from_config()`` — the shared
    session factory is required to construct the user repository.
    """
    global _cached_local_provider, _cached_repo
    if _cached_repo is None:
        from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
        from deerflow.persistence.engine import get_session_factory

        sf = get_session_factory()
        if sf is None:
            raise RuntimeError("get_local_provider() called before init_engine_from_config(); cannot access users table")
        _cached_repo = SQLiteUserRepository(sf)
    if _cached_local_provider is None:
        from app.gateway.auth.local_provider import LocalAuthProvider

        _cached_local_provider = LocalAuthProvider(repository=_cached_repo)
    return _cached_local_provider


async def get_current_user_from_request(request: Request):
    """Get the current authenticated user from request token.

    Raises HTTPException 401 if not authenticated.
    """
    from app.gateway.auth import decode_token
    from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse, TokenError, token_error_to_code

    access_token = get_request_access_token(request)
    if not access_token:
        raise HTTPException(
            status_code=401,
            detail=AuthErrorResponse(code=AuthErrorCode.NOT_AUTHENTICATED, message="Not authenticated").model_dump(),
        )

    payload = decode_token(access_token)
    if isinstance(payload, TokenError):
        raise HTTPException(
            status_code=401,
            detail=AuthErrorResponse(code=token_error_to_code(payload), message=f"Token error: {payload.value}").model_dump(),
        )

    provider = get_local_provider()
    user = await provider.get_user(payload.sub)
    if user is None:
        raise HTTPException(
            status_code=401,
            detail=AuthErrorResponse(code=AuthErrorCode.USER_NOT_FOUND, message="User not found").model_dump(),
        )

    # Token version mismatch → password was changed, token is stale
    if user.token_version != payload.ver:
        raise HTTPException(
            status_code=401,
            detail=AuthErrorResponse(code=AuthErrorCode.TOKEN_INVALID, message="Token revoked (password changed)").model_dump(),
        )

    return user


async def get_optional_user_from_request(request: Request):
    """Get optional authenticated user from request.

    Returns None if not authenticated.
    """
    try:
        return await get_current_user_from_request(request)
    except HTTPException:
        return None


async def get_current_user(request: Request) -> str | None:
    """Extract user_id from request cookie, or None if not authenticated.

    Thin adapter that returns the string id for callers that only need
    identification (e.g., ``feedback.py``). Full-user callers should use
    ``get_current_user_from_request`` or ``get_optional_user_from_request``.
    """
    user = await get_optional_user_from_request(request)
    return str(user.id) if user else None