"""Dedicated single-process online execution entrypoint."""

from __future__ import annotations

import asyncio
import hmac
import os
import time
from contextlib import asynccontextmanager

from starlette.responses import JSONResponse

from app.gateway.app import create_app
from app.gateway.deps import langgraph_runtime
from app.gateway.runtime_executor import configure_runtime_executor
from deerflow.config import get_app_config

from .admission import Admission
from .manager import WorkerRunManager
from .preflight import shared_storage_configured, validate_sqlite_mount
from .proxy import THREAD_PATH
from .refresh import SharedSkillRefresh


async def _measure_worker_event_loop_lag(app):
    loop = asyncio.get_running_loop()
    while True:
        if not getattr(app.state, "worker_diagnostics_enabled", False):
            app.state.worker_event_loop_lag_ms = 0.0
            await asyncio.sleep(1)
            continue
        started = loop.time()
        await asyncio.sleep(1)
        app.state.worker_event_loop_lag_ms = round(max(0.0, loop.time() - started - 1) * 1000, 2)


@asynccontextmanager
async def worker_lifespan(app):
    from .revision import code_revision
    app.state.worker_process_started = time.time()
    app.state.worker_runtime_revision = code_revision()
    config = get_app_config()
    from .runtime import verify_storage_dependencies

    verify_storage_dependencies(config)
    if not shared_storage_configured(config):
        raise RuntimeError("Managed workers require persistent PostgreSQL or SQLite chat storage")
    if "AioSandboxProvider" in str(getattr(config.sandbox, "use", "")):
        raise RuntimeError("Managed online workers cannot launch Docker sandboxes")
    validate_sqlite_mount(config)
    # Shared event records follow conversations across drained containers.
    config = config.model_copy(
        update={
            "run_events": config.run_events.model_copy(update={"backend": "db"}),
            "database": config.database.model_copy(update={"pool_size": int(os.environ.get("DEER_FLOW_WORKER_DB_POOL_SIZE", "4")), "max_overflow": 0}),
        }
    )
    app.state.config = config
    app.state.worker_skill_refresh = SharedSkillRefresh()
    app.state.worker_deferred_admission = True
    app.state.worker_admission = Admission(int(os.environ.get("DEER_FLOW_WORKER_CONCURRENCY", "4")), int(os.environ.get("DEER_FLOW_WORKER_QUEUE_LIMIT", "128")), float(os.environ.get("DEER_FLOW_WORKER_QUEUE_TIMEOUT", "60")))
    # Each execution container keeps its own bounded rolling request metrics;
    # the controller reads a compact summary with the normal status probe.
    from .telemetry import RequestRecorder

    app.state.request_recorder = RequestRecorder()
    app.state.worker_diagnostics_enabled = False
    app.state.worker_event_loop_lag_ms = 0.0
    await configure_runtime_executor(config.runtime_performance)
    async with langgraph_runtime(app, background=False):
        from langgraph.checkpoint.memory import InMemorySaver
        from langgraph.store.memory import InMemoryStore

        if isinstance(app.state.checkpointer, InMemorySaver) or isinstance(app.state.store, InMemoryStore):
            raise RuntimeError("Managed worker refuses an in-memory persistence fallback")
        close_manager = getattr(app.state.run_manager, "aclose", None)
        if close_manager is not None:
            await close_manager()
        app.state.run_manager = WorkerRunManager(store=app.state.run_store)
        retention = getattr(config, "runtime_contention", None)
        if retention is not None and hasattr(app.state.run_manager, "start_retention"):
            app.state.run_manager.start_retention(retention_seconds=retention.completed_run_retention_seconds, interval_seconds=retention.run_sweep_seconds)
        lag_task = asyncio.create_task(_measure_worker_event_loop_lag(app))
        try:
            yield
        finally:
            lag_task.cancel()
            await asyncio.gather(lag_task, return_exceptions=True)


class InternalOnly:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        expected = os.environ.get("DEER_FLOW_WORKER_TOKEN", "")
        token = dict(scope["headers"]).get(b"x-deerflow-worker-token", b"").decode()
        if not expected or not hmac.compare_digest(token, expected):
            return await JSONResponse({"detail": "Forbidden"}, 403)(scope, receive, send)
        if scope["path"] == "/internal/worker-status":
            gate = scope["app"].state.worker_admission
            recorder = getattr(scope["app"].state, "request_recorder", None)
            diagnostics_enabled = dict(scope["headers"]).get(b"x-deerflow-diagnostics-enabled") == b"true"
            scope["app"].state.worker_diagnostics_enabled = diagnostics_enabled
            if recorder is not None:
                recorder.enabled = diagnostics_enabled
            http = recorder.worker_status() if recorder is not None else None
            return await JSONResponse({"ready": True, "active": gate.active + gate.waiting, "running": gate.active, "waiting": gate.waiting, "capacity": gate.capacity,
                                       "process_started_at": scope["app"].state.worker_process_started,
                                       "runtime_revision": scope["app"].state.worker_runtime_revision,
                                       "event_loop_lag_ms": getattr(scope["app"].state, "worker_event_loop_lag_ms", None),
                                       "diagnostics_enabled": diagnostics_enabled,
                                       "http": http})(scope, receive, send)
        active_prefix = "/internal/worker-thread-active/"
        if scope["path"].startswith(active_prefix):
            from urllib.parse import unquote

            thread_id = unquote(scope["path"][len(active_prefix):])
            active = scope["app"].state.run_manager.has_active_thread(thread_id)
            return await JSONResponse({"active": active})(scope, receive, send)
        match = THREAD_PATH.fullmatch(scope["path"])
        if not match:
            return await JSONResponse({"detail": "Worker endpoint not available"}, 404)(scope, receive, send)
        scope.setdefault("state", {})["worker_node"] = os.environ.get("HOSTNAME", "worker")
        from deerflow.runtime.thread_paths import aresolve_path_user_id

        await aresolve_path_user_id(match[1])
        await self.app(scope, receive, send)


app = create_app()
app.router.lifespan_context = worker_lifespan
app.add_middleware(InternalOnly)