"""PMFlow FastAPI 应用入口。"""

from __future__ import annotations

import asyncio
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import sqlalchemy as sa

from .api.router import api_router
from .core.config import settings
from .core.logging import configure_logging, get_logger
from .core.response import success_response
from .database import AsyncSessionLocal, engine
from .models import Base
from .services.scheduler_center import WorkloadRecalcScheduler
from .services.scheduler_center import KnowledgeUrlRefreshScheduler
from .services.scheduler_center import KnowledgeGraphIncrementalScheduler
from .services.scheduler_center import AIPMOCadenceScheduler
from .services.ai_orchestration_run_service import AIOrchestrationRunService
from .services.skill_service import init_system_skill_tags
from .services.feishu_long_connection import init_feishu_long_connection_manager
from .services.feishu_integration import dispatch_pending_feishu_replies
from .services.context_cache import register_context_cache_invalidation_hooks

configure_logging(settings.debug)
logger = get_logger(__name__)
register_context_cache_invalidation_hooks()


def _sqlite_backfill_columns(sync_conn) -> None:
    """为历史 SQLite 库补齐新增列，避免运行时 500。"""
    dialect_name = str(sync_conn.dialect.name or "").lower()
    if dialect_name != "sqlite":
        return

    def _columns(table_name: str) -> set[str]:
        try:
            rows = sync_conn.exec_driver_sql(f"PRAGMA table_info({table_name})").fetchall()
            return {str(row[1]) for row in rows}
        except Exception:
            return set()

    def _add_column(table_name: str, column_name: str, column_ddl: str) -> None:
        cols = _columns(table_name)
        if not cols or column_name in cols:
            return
        sync_conn.exec_driver_sql(
            f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_ddl}"
        )

    # truth_nodes
    _add_column("truth_nodes", "signal_id", "BIGINT")

    # signals
    _add_column("signals", "feishu_item_id", "BIGINT")
    _add_column("signals", "matched_project_id", "BIGINT")
    _add_column("signals", "title", "VARCHAR(200)")
    _add_column("signals", "signal_type", "VARCHAR(50)")
    _add_column("signals", "summary", "TEXT")
    _add_column("signals", "reasoning", "TEXT")
    _add_column("signals", "confidence", "FLOAT")
    _add_column("signals", "recommended_action", "VARCHAR(50)")
    _add_column("signals", "status", "VARCHAR(50) NOT NULL DEFAULT '待处理'")

    # qa_records
    _add_column("qa_records", "session_id", "INTEGER")
    _add_column("qa_records", "turn_index", "INTEGER NOT NULL DEFAULT 1")

    # ai_processing_logs
    _add_column("ai_processing_logs", "normalized_text", "TEXT")
    _add_column("ai_processing_logs", "route", "VARCHAR(80)")
    _add_column("ai_processing_logs", "intent", "VARCHAR(40)")
    _add_column("ai_processing_logs", "executor", "VARCHAR(40)")
    _add_column("ai_processing_logs", "operation_code", "VARCHAR(80)")
    _add_column("ai_processing_logs", "operation_result", "JSON")
    _add_column("ai_processing_logs", "reply_source", "VARCHAR(40)")


def _normalize_dialect(value: str) -> str:
    raw = str(value or "").strip().lower()
    if not raw:
        return ""
    if raw.startswith("postgres"):
        return "postgresql"
    if raw.startswith("sqlite"):
        return "sqlite"
    if raw.startswith("mysql"):
        return "mysql"
    if "+" in raw:
        raw = raw.split("+", 1)[0]
    if ":" in raw:
        raw = raw.split(":", 1)[0]
    return raw


def _mask_db_url(url: str) -> str:
    value = str(url or "").strip()
    if "@" not in value:
        return value
    prefix, suffix = value.split("@", 1)
    if "://" not in prefix:
        return f"***@{suffix}"
    scheme, creds = prefix.split("://", 1)
    user = creds.split(":", 1)[0] if ":" in creds else creds
    return f"{scheme}://{user}:***@{suffix}"


def _check_database_dialect_guard(sync_conn) -> None:
    configured = _normalize_dialect(settings.database_url)
    actual = _normalize_dialect(sync_conn.dialect.name)
    expected = _normalize_dialect(os.getenv("PMFLOW_EXPECTED_DB_DIALECT", ""))

    env_urls = [
        os.getenv("DATABASE_URL", ""),
        os.getenv("SQLALCHEMY_DATABASE_URL", ""),
        os.getenv("SYNC_DATABASE_URL", ""),
    ]
    configured_families = {_normalize_dialect(item) for item in env_urls if str(item or "").strip()}
    configured_families.discard("")

    logger.info(
        "database_dialect_guard",
        configured=configured,
        actual=actual,
        expected=expected or "unset",
        database_url=_mask_db_url(settings.database_url),
    )
    print(
        "[DB-Guard] "
        f"configured={configured}, actual={actual}, expected={expected or 'unset'}, "
        f"url={_mask_db_url(settings.database_url)}"
    )
    if configured and actual and configured != actual:
        raise RuntimeError(
            f"数据库方言不一致: configured={configured}, actual={actual}。"
            "请确保同一服务只连接单一数据库方言。"
        )
    if expected and actual and expected != actual:
        raise RuntimeError(
            f"数据库方言与 PMFLOW_EXPECTED_DB_DIALECT 不一致: expected={expected}, actual={actual}。"
        )
    if len(configured_families) > 1:
        raise RuntimeError(
            f"检测到多种数据库方言配置: {sorted(configured_families)}。"
            "禁止同一服务混用 SQLite/Postgres。"
        )


def _ensure_feishu_integrity_guards(sync_conn) -> None:
    dialect = _normalize_dialect(sync_conn.dialect.name)
    if dialect not in {"postgresql", "sqlite"}:
        return

    # Outbox 表兜底，兼容未跑 Alembic 的环境。
    outbox_meta = sa.MetaData()
    outbox_table = sa.Table(
        "feishu_reply_outbox",
        outbox_meta,
        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
        sa.Column("tenant_id", sa.BigInteger(), nullable=False, server_default=sa.text("1")),
        sa.Column("created_at", sa.TIMESTAMP(), nullable=False, server_default=sa.func.now()),
        sa.Column("updated_at", sa.TIMESTAMP(), nullable=False, server_default=sa.func.now(), onupdate=sa.func.now()),
        sa.Column("trace_id", sa.String(length=64), nullable=True),
        sa.Column("chat_id", sa.String(length=128), nullable=False),
        sa.Column("message_text", sa.Text(), nullable=False),
        sa.Column("status", sa.String(length=20), nullable=False, server_default=sa.text("'pending'")),
        sa.Column("attempts", sa.Integer(), nullable=False, server_default=sa.text("0")),
        sa.Column("next_retry_at", sa.TIMESTAMP(), nullable=True),
        sa.Column("sent_at", sa.TIMESTAMP(), nullable=True),
        sa.Column("last_error", sa.Text(), nullable=True),
    )
    outbox_table.create(sync_conn, checkfirst=True)
    sync_conn.exec_driver_sql(
        "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_tenant_id ON feishu_reply_outbox (tenant_id)"
    )
    sync_conn.exec_driver_sql(
        "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_trace_id ON feishu_reply_outbox (trace_id)"
    )
    sync_conn.exec_driver_sql(
        "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_chat_id ON feishu_reply_outbox (chat_id)"
    )
    sync_conn.exec_driver_sql(
        "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_status ON feishu_reply_outbox (status)"
    )
    sync_conn.exec_driver_sql(
        "CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_next_retry_at ON feishu_reply_outbox (next_retry_at)"
    )

    inspector = sa.inspect(sync_conn)
    if not inspector.has_table("feishu_items"):
        return

    duplicate_groups = sync_conn.execute(
        sa.text(
            """
            SELECT tenant_id, feishu_message_id, split_from_item_id, MIN(id) AS keep_id
            FROM feishu_items
            WHERE feishu_message_id IS NOT NULL
            GROUP BY tenant_id, feishu_message_id, split_from_item_id
            HAVING COUNT(*) > 1
            """
        )
    ).fetchall()

    table_exists = {
        "signals": inspector.has_table("signals"),
        "project_suggestions": inspector.has_table("project_suggestions"),
        "ai_processing_logs": inspector.has_table("ai_processing_logs"),
        "feishu_items": True,
    }

    for tenant_id, message_id, split_from_item_id, keep_id in duplicate_groups:
        if split_from_item_id is None:
            duplicate_ids = sync_conn.execute(
                sa.text(
                    """
                    SELECT id
                    FROM feishu_items
                    WHERE tenant_id = :tenant_id
                      AND feishu_message_id = :message_id
                      AND split_from_item_id IS NULL
                      AND id <> :keep_id
                    ORDER BY id ASC
                    """
                ),
                {
                    "tenant_id": int(tenant_id),
                    "message_id": str(message_id),
                    "keep_id": int(keep_id),
                },
            ).fetchall()
        else:
            duplicate_ids = sync_conn.execute(
                sa.text(
                    """
                    SELECT id
                    FROM feishu_items
                    WHERE tenant_id = :tenant_id
                      AND feishu_message_id = :message_id
                      AND split_from_item_id = :split_from_item_id
                      AND id <> :keep_id
                    ORDER BY id ASC
                    """
                ),
                {
                    "tenant_id": int(tenant_id),
                    "message_id": str(message_id),
                    "split_from_item_id": int(split_from_item_id),
                    "keep_id": int(keep_id),
                },
            ).fetchall()

        for (duplicate_id,) in duplicate_ids:
            duplicate_id = int(duplicate_id)
            canonical_id = int(keep_id)
            if table_exists["feishu_items"]:
                sync_conn.execute(
                    sa.text(
                        "UPDATE feishu_items SET split_from_item_id = :new_id "
                        "WHERE split_from_item_id = :old_id"
                    ),
                    {"new_id": canonical_id, "old_id": duplicate_id},
                )
            if table_exists["signals"]:
                sync_conn.execute(
                    sa.text(
                        "UPDATE signals SET feishu_item_id = :new_id "
                        "WHERE feishu_item_id = :old_id"
                    ),
                    {"new_id": canonical_id, "old_id": duplicate_id},
                )
            if table_exists["project_suggestions"]:
                sync_conn.execute(
                    sa.text(
                        "UPDATE project_suggestions SET feishu_item_id = :new_id "
                        "WHERE feishu_item_id = :old_id"
                    ),
                    {"new_id": canonical_id, "old_id": duplicate_id},
                )
            if table_exists["ai_processing_logs"]:
                sync_conn.execute(
                    sa.text(
                        "UPDATE ai_processing_logs SET feishu_item_id = :new_id "
                        "WHERE feishu_item_id = :old_id"
                    ),
                    {"new_id": canonical_id, "old_id": duplicate_id},
                )
            sync_conn.execute(
                sa.text("DELETE FROM feishu_items WHERE id = :id"),
                {"id": duplicate_id},
            )

    # 幂等唯一约束（含 split 维度）硬化；若存在重复历史数据将直接失败，避免静默错乱。
    sync_conn.exec_driver_sql(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS uq_feishu_items_tenant_message_root
        ON feishu_items (tenant_id, feishu_message_id)
        WHERE split_from_item_id IS NULL AND feishu_message_id IS NOT NULL
        """
    )
    sync_conn.exec_driver_sql(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS uq_feishu_items_tenant_message_split
        ON feishu_items (tenant_id, feishu_message_id, split_from_item_id)
        WHERE split_from_item_id IS NOT NULL AND feishu_message_id IS NOT NULL
        """
    )


async def _feishu_outbox_drain_loop() -> None:
    while True:
        try:
            dispatched = await dispatch_pending_feishu_replies(limit=50)
            await asyncio.sleep(0.3 if dispatched > 0 else 2.0)
        except asyncio.CancelledError:
            break
        except Exception as exc:
            logger.warning("feishu_outbox_drain_failed", error=str(exc))
            await asyncio.sleep(2.0)


@asynccontextmanager
async def lifespan(_: FastAPI):
    """应用生命周期：初始化表与系统技能标签。"""
    auto_create_tables = os.getenv("AUTO_CREATE_TABLES", "true").lower() in {
        "1",
        "true",
        "yes",
    }
    async with engine.begin() as conn:
        await conn.run_sync(_check_database_dialect_guard)
        if auto_create_tables:
            await conn.run_sync(Base.metadata.create_all)
            await conn.run_sync(_sqlite_backfill_columns)
        await conn.run_sync(_ensure_feishu_integrity_guards)

    async with AsyncSessionLocal() as db:
        try:
            created = await init_system_skill_tags(db, tenant_id=1)
            if created:
                logger.info("system_skill_tags_seeded", created=created)
        except Exception as exc:  # pragma: no cover
            logger.warning("seed_system_skills_failed", error=str(exc))
        try:
            interrupted = await AIOrchestrationRunService(db).mark_stale_runs_interrupted(older_than_minutes=10)
            if interrupted:
                await db.commit()
                logger.info("stale_ai_orchestration_runs_interrupted", count=interrupted)
        except Exception as exc:  # pragma: no cover
            logger.warning("interrupt_stale_ai_orchestration_runs_failed", error=str(exc))

    scheduler = WorkloadRecalcScheduler(AsyncSessionLocal)
    scheduler.start()
    knowledge_scheduler = KnowledgeUrlRefreshScheduler(AsyncSessionLocal)
    knowledge_scheduler.start()
    kg_scheduler = KnowledgeGraphIncrementalScheduler(AsyncSessionLocal)
    kg_scheduler.start()
    ai_pmo_cadence_scheduler = AIPMOCadenceScheduler(AsyncSessionLocal)
    ai_pmo_cadence_scheduler.start()
    feishu_lc = init_feishu_long_connection_manager(AsyncSessionLocal)
    await feishu_lc.start()
    outbox_task = asyncio.create_task(
        _feishu_outbox_drain_loop(),
        name="feishu-outbox-drain-loop",
    )

    yield
    outbox_task.cancel()
    try:
        await outbox_task
    except asyncio.CancelledError:
        pass
    await feishu_lc.stop()
    await ai_pmo_cadence_scheduler.stop()
    await kg_scheduler.stop()
    await knowledge_scheduler.stop()
    await scheduler.stop()
    await engine.dispose()


app = FastAPI(
    title=settings.app_name,
    debug=settings.debug,
    version="0.1.0",
    lifespan=lifespan,
)

origins = [item.strip() for item in settings.cors_origins.split(",") if item.strip()]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins or ["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(api_router)


@app.get("/healthz")
async def healthz() -> dict:
    return success_response({"status": "ok"})