"""FastAPI application and route registration."""
from __future__ import annotations

# Ensure hosting_panel/ is in sys.path for HF Spaces container execution.
# When uvicorn starts from /app/ (repo root), bare imports like
# ``from config import PANEL_CONFIG`` fail because hosting_panel/ is not
# on sys.path.  This patch is idempotent and must run BEFORE any other
# imports from the panel package.
import os as _os, sys as _sys
_PANEL_DIR = _os.path.dirname(_os.path.abspath(__file__))
if _PANEL_DIR not in _sys.path:
    _sys.path.insert(0, _PANEL_DIR)

import asyncio
import logging
import os
import time
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Any, Optional

from fastapi import FastAPI, Request, Query
from fastapi.responses import HTMLResponse, JSONResponse, Response, RedirectResponse
from fastapi.staticfiles import StaticFiles

from config import PANEL_CONFIG
from database import init_db
from storage import (
    PANEL_STORAGE_COORD,
    PanelStorage,
    audit_log_prune_loop,
)
from bot_runner import resource_watchdog_loop
from alerts import evaluate_rules_loop
from routes import router
from templates_setup import templates
from i18n import available_languages, gettext, ngettext, _detect_lang, _CATALOG_LANGS

logging.basicConfig(
    level=logging.INFO if not PANEL_CONFIG.debug else logging.DEBUG,
    format="%(asctime)s %(levelname)-7s %(name)s | %(message)s",
)
log = logging.getLogger("hosting-panel")


# ---------------------------------------------------------------------------
# Signal handlers — flush panel state synchronously on SIGTERM/SIGINT
# ---------------------------------------------------------------------------
#
# HF Spaces sends SIGTERM before stopping / rebuilding the container and
# gives the process ~30 s to clean up.  The async lifespan shutdown
# already calls ``PANEL_STORAGE_COORD.shutdown()`` (which awaits
# ``flush_now``), but uvicorn's lifespan exit only runs if the signal is
# propagated through the ASGI server — if the container is killed
# forcefully or the event loop is already busy, queued DB changes never
# reach the HF Dataset mirror and the next ``storage.pull`` restores an
# older snapshot (effectively rolling the panel back).
#
# Installing a sync signal handler that calls ``PanelStorage._sync_push``
# directly is defensive: it runs regardless of whether uvicorn gets to
# its lifespan-exit path.  The handler is idempotent (multiple SIGTERMs
# just re-push the same state) and exits fast (~50 ms on the hot path).
def _sigterm_sync_push(signum, frame):
    try:
        log.warning("signal %d received — synchronous panel-state push", signum)
        storage = PanelStorage()
        if storage.enabled and storage.safe_to_push:
            storage._sync_push(message=f"signal {signum}")
    except Exception as exc:
        log.warning("signal %d push failed: %r", signum, exc)


try:
    import signal as _signal
    _signal.signal(_signal.SIGTERM, _sigterm_sync_push)
    _signal.signal(_signal.SIGINT, _sigterm_sync_push)
except Exception:
    # Windows / restricted runtimes may reject signal() — non-fatal,
    # the async lifespan shutdown still runs.
    pass


# ---------------------------------------------------------------------------
# Inline column-level migrations (runs from lifespan startup)
# ---------------------------------------------------------------------------
#
# Background: HF Spaces' Docker SDK lets ``app_file`` point at any module
# (the README on ItsBounvy/bot_host says ``app:app``). The previous
# approach — invoking the migration from ``run_panel.py`` — relied on
# that exact file being the entry point, and silently skipped every
# other invocation path. To make it bulletproof we ship the migration
# logic embedded in ``app.py`` itself and call it from ``lifespan``
# before ``init_db`` reads the schema. The script ``migrate_spaces_tiers.py``
# remains as a CLI entry point for operators who want to run it
# manually (``python migrate_spaces_tiers.py /path/to/db``).
def _run_pending_migrations_inline() -> None:
    """Apply any pending column-level migrations in-process.

    Idempotent: each ALTER catches ``sqlite3.OperationalError`` with a
    "duplicate column" message and continues. Only runs when the
    configured DB URL is SQLite; non-SQLite deployments would use a
    real migration tool.
    """
    # Read the effective URL from PANEL_CONFIG, not os.getenv — on HF
    # Spaces the env var ``PANEL_DATABASE_URL`` is NOT always set and
    # the panel falls back to ``config.py``'s default. Reading from
    # PANEL_CONFIG gives us the *actual* URL the runtime will use.
    db_url = (PANEL_CONFIG.database_url or "").strip()
    if not db_url.startswith("sqlite"):
        return
    import sqlite3

    # Resolve the file path the same way the production code does.
    # ``sqlite+aiosqlite:///`` -> file path is the part after ``///``.
    path = ""
    if ":///" in db_url:
        path = db_url.split(":///", 1)[1]
    elif "://" in db_url:
        path = db_url.split("://", 1)[1]
    if not path:
        log.info("migrations: cannot resolve DB file from %r; skipping", db_url)
        return
    from pathlib import Path as _Path
    db_file = _Path(path)
    if not db_file.is_file():
        log.info(
            "migrations: %s does not exist yet; create_all() will handle the schema.",
            db_file,
        )
        return

    add_columns = [
        ("users",          "plan_tier",                "VARCHAR(16)",  "'free'"),
        ("host_spaces",    "tier",                     "VARCHAR(16)",  "'free'"),
        ("host_spaces",    "requested_at",             "DATETIME",     None),
        ("host_spaces",    "approved_at",              "DATETIME",     None),
        ("host_spaces",    "approved_by_user_id",      "INTEGER",      None),
        ("host_spaces",    "requested_by_user_id",     "INTEGER",      None),
        ("host_spaces",    "operator_note",            "TEXT",         None),
        ("host_spaces",    "baseline_cpu_reserved",    "FLOAT",        "0.1"),
        ("host_spaces",    "baseline_ram_reserved_mb", "INTEGER",      "128"),
        ("bot_instances",  "space_state",              "VARCHAR(32)",  "'on_dataset'"),
    ]

    conn = sqlite3.connect(str(db_file))
    try:
        cur = conn.cursor()
        cur.execute("PRAGMA table_info(users)")
        users_cols = {row[1] for row in cur.fetchall()}
        cur.execute("PRAGMA table_info(host_spaces)")
        spaces_cols = {row[1] for row in cur.fetchall()}
        cur.execute("PRAGMA table_info(bot_instances)")
        bot_cols = {row[1] for row in cur.fetchall()}

        applied = 0
        for tbl, col, sql_type, default in add_columns:
            existing = {
                "users": users_cols,
                "host_spaces": spaces_cols,
                "bot_instances": bot_cols,
            }.get(tbl, set())
            if col in existing:
                continue
            default_clause = f" DEFAULT {default}" if default is not None else ""
            try:
                cur.execute(
                    f"ALTER TABLE {tbl} ADD COLUMN {col} {sql_type}{default_clause}"
                )
                log.info(
                    "migrations: +%s.%s %s%s",
                    tbl, col, sql_type, default_clause,
                )
                applied += 1
            except sqlite3.OperationalError as exc:
                msg = str(exc).lower()
                if "duplicate column" in msg or "already exists" in msg:
                    continue
                raise
        conn.commit()
        if applied:
            log.info("migrations: applied %d column(s) to %s", applied, db_file)
        else:
            log.info("migrations: schema already up to date at %s", db_file)
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# Keep-alive self-ping (anti-sleep for HF Space free tier)
# ---------------------------------------------------------------------------
#
# HF Spaces on the free tier sleep after ~15 min of no external HTTP
# traffic.  Subprocess bots and the whole panel go away with them.
# We schedule a background task that pings either:
#
# 1. ``PANEL_CONFIG.keepalive_url`` if explicit;
# 2. The Space's own public URL derived from ``SPACE_ID`` +
#    ``SPACE_AUTHOR_NAME`` (HF provides both — Space URL is
#    ``https://<author>-<name>.hf.space``);
# 3. ``localhost`` as a last-resort fallback (HF still counts the
#    request, depends on platform).
#
# External cron (UptimeRobot, GitHub Actions, cron-job.org) hitting
# ``/api/health`` is the recommended approach — the self-ping is
# a best-effort belt under the suspenders.
import httpx


def _resolve_keepalive_url() -> str:
    """Pick a URL to ping periodically so the HF Space doesn't sleep.

    Resolution order:

    1. Explicit ``KEEPALIVE_URL`` env var (the documented override).
    2. Auto-detect from HF env vars.  HuggingFace sets:
         - ``SPACE_AUTHOR_NAME`` = the owner namespace, e.g. ``ItsBounvy``
         - ``SPACE_ID``           = the full repo id, e.g. ``ItsBounvy/bot_host``
         - ``SPACE_REPO_NAME``    = just the repo name (newer Spaces),
                                    e.g. ``bot_host``
       The runtime URL is ``https://<owner>-<repo>.hf.space`` (slash in
       the repo id becomes a hyphen).  Naively concatenating
       ``{author}-{space_id}`` produces ``ItsBounvy-ItsBounvy/bot_host``
       which is invalid — the Space then keeps trying to ping a URL
       that doesn't exist, fails to keep itself awake, and after
       ~15 min of real idleness goes to sleep on the free tier,
       rejecting every external request.  The previous version of this
       function had exactly that bug.
    3. Empty string — caller treats that as "disable self-ping".
    """
    explicit = PANEL_CONFIG.keepalive_url.strip()
    if explicit:
        return explicit

    author = os.getenv("SPACE_AUTHOR_NAME", "").strip()

    # Prefer SPACE_REPO_NAME when available — it's just the bare repo
    # name with no namespace prefix.
    repo_name = os.getenv("SPACE_REPO_NAME", "").strip()
    if not repo_name:
        # Fall back to stripping the namespace from SPACE_ID.
        space_id = os.getenv("SPACE_ID", "").strip()
        if space_id and "/" in space_id:
            repo_name = space_id.split("/", 1)[1]
        elif space_id:
            repo_name = space_id

    if author and repo_name:
        # HF runtime URL convention: namespace + "-" + repo_name.
        # Two normalisations matter here:
        #   1. Underscore -> hyphen.  HF issues TLS certs with the
        #      repo name's underscores replaced by hyphens
        #      (``ItsBounvy/bot_host`` becomes
        #      ``ItsBounvy-bot-host.hf.space`` in the cert SAN).
        #      Leaving the underscore in the URL produces
        #      ``[SSL: CERTIFICATE_VERIFY_FAILED] Hostname mismatch``
        #      because the cert is for the hyphenated form.
        #   2. Lowercase.  Cert SANs are issued in lowercase;
        #      Python's ssl module lowercases for comparison anyway,
        #      but using lowercase in the URL also avoids surprises
        #      in proxy headers, logs and external monitors.
        host = f"{author}-{repo_name}".replace("_", "-").lower()
        return f"https://{host}.hf.space/health"

    # Last-resort: hit ourselves on localhost.  Keeps the event loop
    # active and is a useful health probe even when HF can't be reached.
    return ""


async def _keepalive_loop(url: str, interval_sec: int) -> None:
    log.info("keep-alive loop running for %s every %ds", url, interval_sec)
    # Send ``Connection: close`` so each iteration opens a fresh TCP
    # socket; keeps the loop robust against any half-open keep-alive
    # sockets (esp. on Windows where asyncio + httpx can wedge on
    # idle pooled connections).
    headers = {"Connection": "close", "User-Agent": "MMaxAI-Panel-KeepAlive/1.0"}
    while True:
        try:
            await asyncio.sleep(interval_sec)
            try:
                timeout = httpx.Timeout(15.0)
                async with httpx.AsyncClient(timeout=timeout, headers=headers) as client:
                    t0 = time.time()
                    resp = await client.get(url)
                    elapsed_ms = int((time.time() - t0) * 1000)
                    if resp.status_code >= 500:
                        log.warning(
                            "keep-alive: %s returned %d (%dms)",
                            url, resp.status_code, elapsed_ms,
                        )
                    else:
                        log.debug(
                            "keep-alive: %s -> %d (%dms)",
                            url, resp.status_code, elapsed_ms,
                        )
            except Exception as exc:
                log.warning("keep-alive ping failed for %s: %s", url, exc)
        except asyncio.CancelledError:
            return
        except Exception as exc:
            log.warning("keep-alive loop error: %s", exc)


# ---------------------------------------------------------------------------
# Security headers middleware
# ---------------------------------------------------------------------------
#
# Defence in depth: the panel renders user-controlled data (bot names, slugs,
# env-var values, log lines) into HTML.  Jinja auto-escapes by default, but
# inline JS in panel.js consumes JSON from the API -- if a bot name contains
# </script><script>alert(1)</script>, a naive json.dumps would break the
# page's JS context.  Adding CSP blocks the inline payload entirely; X-Frame
# prevents clickjacking; X-Content-Type-Options prevents MIME sniffing; the
# Referrer-Policy stops leaking paths to external CDNs.

_CSP = (
    "default-src 'self'; "
    "img-src 'self' data: https:; "
    "style-src 'self' 'unsafe-inline'; "
    "script-src 'self' 'unsafe-inline'; "
    "font-src 'self' data:; "
    "connect-src 'self' https://huggingface.co https://*.hf.space "
    "https://integrate.api.nvidia.com; "
    # HF Spaces embeds the running app inside an <iframe> on the
    # "App" tab of huggingface.co. 'none' blocked that too (browser
    # shows a hard "host refused to connect" -- client-side frame
    # block, not a real network failure). Allow framing only from
    # HF's own domains, not from arbitrary sites.
    "frame-ancestors 'self' https://huggingface.co https://*.hf.space; "
    "base-uri 'self'; "
    "form-action 'self'"
)


async def _security_headers_middleware(request: Request, call_next):
    response: Response = await call_next(request)
    response.headers.setdefault("Content-Security-Policy", _CSP)
    response.headers.setdefault("X-Content-Type-Options", "nosniff")
    response.headers.setdefault("Referrer-Policy", "no-referrer")
    response.headers.setdefault("Permissions-Policy", "geolocation=(), microphone=(), camera=()")
    if not PANEL_CONFIG.debug:
        response.headers.setdefault(
            "Strict-Transport-Security",
            "max-age=31536000; includeSubDomains",
        )
    return response


# ---------------------------------------------------------------------------
# Language detection middleware
# ---------------------------------------------------------------------------
#
# Sets ``request.state.panel_lang`` before any route runs.  Resolution order:
#   1. ``panel_lang`` query param (for links like /?lang=uk)
#   2. ``panel_lang`` cookie
#   3. ``Accept-Language`` header (first match against supported langs)
#   4. ``PANEL_DEFAULT_LANG`` env var (fallback)
#
# The template helper ``_()`` reads ``request.state.panel_long`` via
# ``_gettext_context`` in templates_setup.py.

_ACCEPT_LANG_CACHE: dict[str, Optional[str]] = {}


def _resolve_lang_from_accept_language(header: str) -> Optional[str]:
    """Best-effort parse of ``Accept-Language`` header."""
    if not header:
        return None
    # Cache by full header string to avoid re-parsing on every request
    if header in _ACCEPT_LANG_CACHE:
        return _ACCEPT_LANG_CACHE[header]
    result = None
    try:
        parts = [p.strip().split(";")[0].strip().lower() for p in header.split(",") if p.strip()]
        for part in parts:
            if part in _CATALOG_LANGS:
                result = part
                break
            # Try prefix match: "uk-UA" → "uk"
            prefix = part.split("-")[0] if "-" in part else part
            if prefix in _CATALOG_LANGS:
                result = prefix
                break
    except Exception:
        pass
    _ACCEPT_LANG_CACHE[header] = result
    return result


async def _language_middleware(request: Request, call_next):
    # The middleware also pushes the resolved language into an asyncio
    # ContextVar so route-side ``gettext(...)`` calls (in routes.py)
    # translate the same way the templates do.  Without this hook, only
    # ``{{ _("...") }}`` in Jinja rendered the localised string;
    # server-side strings (flash titles, error messages, audit
    # payloads) silently fell back to English because ``_detect_lang``
    # only saw env vars, not the request.
    import i18n as _i18n

    async def _finalise(response: Response, lang: str) -> Response:
        # Bind the language for the rest of the request lifecycle so
        # anything `await`-ed downstream (route handlers, background
        # tasks spawned from inside them) sees the same language.
        _i18n.set_active_lang(lang)
        try:
            return await call_next(request)
        finally:
            _i18n.set_active_lang(None)

    # 1. Query param
    qp_lang = request.query_params.get("lang")
    if qp_lang and qp_lang.strip() in _CATALOG_LANGS:
        request.state.panel_lang = qp_lang.strip()
        response = await _finalise(request, request.state.panel_lang)
        response.set_cookie(
            key="panel_lang",
            value=request.state.panel_lang,
            max_age=PANEL_CONFIG.remember_days * 86400,
            httponly=False,
            secure=not PANEL_CONFIG.debug,
            samesite="lax",
            path="/",
        )
        return response

    # 2. Cookie
    cookie_lang = request.cookies.get("panel_lang")
    if cookie_lang and cookie_lang in _CATALOG_LANGS:
        request.state.panel_lang = cookie_lang
        return await _finalise(request, cookie_lang)

    # 3. Accept-Language header
    accept_lang = request.headers.get("accept-language", "")
    header_lang = _resolve_lang_from_accept_language(accept_lang)
    if header_lang:
        request.state.panel_lang = header_lang
        return await _finalise(request, header_lang)

    # 4. Env default
    request.state.panel_lang = _detect_lang()
    return await _finalise(request, request.state.panel_lang)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # HF Dataset persistence: pull FIRST so any subsequent schema work
    # sees the LIVE panel.sqlite (the previous DB mirror on the
    # dataset), not an empty file.  Without this ordering, the inline
    # migration below would run on a missing DB and skip every ALTER,
    # then ``storage.pull`` would overwrite the DB with an OLD snapshot
    # that lacks columns added in 2026-07-05+.  That was the cause of
    # the 500s on /login after deploy — the ``users.plan_tier`` column
    # was missing because the dataset mirror restored the pre-migration
    # schema AFTER the migration step had already (correctly) exited
    # for an absent file.
    storage = PanelStorage()
    pull_ok = True  # default for the disabled-storage case
    pull_retries = 0
    MAX_PULL_RETRIES = 3
    try:
        if storage.enabled:
            for attempt in range(1, MAX_PULL_RETRIES + 1):
                ok = await storage.pull()
                if ok:
                    pull_ok = True
                    break
                # Dataset reachable (storage.enabled) but snapshot absent.
                # Could be first-ever run -- check whether the snapshot dir
                # exists at all so we don't keep retrying an empty repo.
                try:
                    from huggingface_hub import list_repo_files
                    files = list_repo_files(
                        repo_id=storage.repo,
                        repo_type="dataset",
                        token=None,
                    )
                    has_snapshot = any(f.startswith("latest/panel.sqlite") for f in files)
                except Exception:
                    has_snapshot = False
                if not has_snapshot:
                    log.warning(
                        "panel-storage: dataset %s has no latest/panel.sqlite "
                        "(first run?); proceeding with fresh DB",
                        storage.repo,
                    )
                    pull_ok = True  # nothing to lose
                    break
                pull_retries = attempt
                log.warning(
                    "panel-storage pull attempt %d/%d failed; retrying in 2s",
                    attempt, MAX_PULL_RETRIES,
                )
                await asyncio.sleep(2)
            if not pull_ok:
                log.error(
                    "panel-storage pull FAILED after %d attempts — aborting "
                    "startup to prevent clobbering the dataset mirror with a "
                    "fresh (empty) DB. The Space will retry automatically.",
                    MAX_PULL_RETRIES,
                )
                raise RuntimeError(
                    "panel-storage pull failed and dataset has a snapshot; "
                    "refusing to start with a fresh DB."
                )
            log.info(
                "panel-storage pull: ok | debounce=%.1fs safety-net=%ds",
                PANEL_CONFIG.storage_debounce_sec,
                PANEL_CONFIG.storage_sync_interval_sec,
            )
        else:
            log.info("panel-storage: HF Dataset not configured -- persistence disabled")
    except RuntimeError:
        raise  # bubble up -- HF Spaces will restart the container
    except Exception:
        log.error("panel-storage initialization failed — running in degraded (local-only) mode", exc_info=True)
        storage = PanelStorage()  # ensure variable is defined for shutdown block

    # ------------------------------------------------------------------
    # Column-level schema migrations (2026-07-05+).
    #
    # ``Base.metadata.create_all`` only creates MISSING tables; it does
    # NOT add new columns to existing tables. The standalone
    # ``migrate_spaces_tiers.py`` script handles the actual ALTERs and is
    # invoked here directly so it runs no matter what entry point HF
    # Spaces picks (``app:app`` vs ``run_panel:app`` vs anything else).
    # It is idempotent: every ALTER skips when the column already exists.
    # Must run AFTER ``storage.pull`` (so the live DB is on disk) and
    # BEFORE ``init_db`` (so ``create_all`` doesn't fight with our
    # ALTERs over column existence).
    # ------------------------------------------------------------------
    try:
        _run_pending_migrations_inline()
    except Exception as exc:  # noqa: BLE001
        log.error("column migration step failed (continuing): %r", exc)

    log.info("Initializing database...")
    try:
        await init_db()
    except Exception:
        # Database failure is fatal — log and re-raise so the container
        # restarts rather than serving broken routes silently.
        log.critical("Database initialization failed", exc_info=True)
        raise

    log.info("Hosting panel ready | debug=%s", PANEL_CONFIG.debug)

    # Background safety-net loop (low frequency; write-through is the
    # primary mechanism, this catches edge cases).
    safety_task = None
    if storage.enabled:
        safety_task = asyncio.create_task(
            PANEL_STORAGE_COORD.run_safety_net(PANEL_CONFIG.storage_sync_interval_sec),
            name="storage-safety-net",
        )

    # Audit-log trim (independent cadence from storage; cheap; runs always).
    prune_task = asyncio.create_task(
        audit_log_prune_loop(PANEL_CONFIG.audit_log_retention_days),
        name="audit-prune",
    )

    # Bot resource watchdog: enforce RAM caps on in-container subprocesses.
    watchdog_task = asyncio.create_task(
        resource_watchdog_loop(),
        name="bot-resource-watchdog",
    )

    # Alert-rule evaluator — fires Telegram / webhook alerts when
    # threshold predicates trip.  Every ``alert_eval_interval_sec``
    # (default 30s) the loop checks all active rules.  Cheap: <50
    # SQL queries per tick.
    alert_task = asyncio.create_task(
        evaluate_rules_loop(PANEL_CONFIG.alert_eval_interval_sec),
        name="alert-evaluator",
    )

    # Keep-alive self-ping so the HF Space doesn't go to sleep on the
    # free tier (15 min idle threshold).
    keepalive_task = None
    keepalive_target = _resolve_keepalive_url()
    if keepalive_target:
        keepalive_task = asyncio.create_task(
            _keepalive_loop(keepalive_target, PANEL_CONFIG.keepalive_interval_sec),
            name="keepalive",
        )
        log.info(
            "keep-alive: pinging %s every %ds",
            keepalive_target, PANEL_CONFIG.keepalive_interval_sec,
        )
    else:
        log.info(
            "keep-alive: disabled (set KEEPALIVE_URL or run SPACE_ID env; "
            "HF free tier will sleep the Space after ~15 min idle)"
        )

    yield

    # Shutdown: cancel background tasks, flush any pending writes, then stop.
    for task in (safety_task, prune_task, watchdog_task, alert_task, keepalive_task):
        if task and not task.done():
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

    if storage.enabled:
        stats = PANEL_STORAGE_COORD.stats
        log.info(
            "panel-storage shutdown: writes=%d pushes=%d dirty=%s",
            stats["writes"], stats["pushes"], stats["dirty"],
        )
        await PANEL_STORAGE_COORD.shutdown()

    log.info("Shutting down hosting panel...")


app = FastAPI(
    title=PANEL_CONFIG.panel_title,
    debug=PANEL_CONFIG.debug,
    lifespan=lifespan,
)

# Language middleware (registered after FastAPI app exists).
app.middleware("http")(_language_middleware)


# ---------------------------------------------------------------------------
# CSRF token pre-seeding middleware
# ---------------------------------------------------------------------------
# Stash a CSRF token on ``request.state`` BEFORE any route runs. The
# template's ``{{ csrf_token() }}`` global reads it from there, so the
# form's hidden input always has a value that matches the cookie
# ``_set_csrf_cookie`` will set on the response. Without this, a
# first-time visitor would render the form with an empty
# ``csrf_token`` field (because ``request.cookies`` is empty) and
# Pydantic would 422 the form submit.
#
# We don't actually mutate the request — we just attach the token to
# ``request.state``. ``_set_csrf_cookie`` then either reuses the same
# token (if the request already had one in its cookies) or generates a
# fresh one (which would change the token mid-render, defeating the
# point of having one). So we use the same generation rule as
# ``_csrf_token_for`` to keep them in sync.
async def _csrf_pre_seeder_middleware(request: Request, call_next):
    from auth import generate_csrf_token
    existing = request.cookies.get("csrf_token")
    token = existing or generate_csrf_token()
    try:
        request.state.csrf_token = token
    except Exception:  # noqa: BLE001
        # Some test stubs don't provide ``state`` — skip silently.
        pass
    return await call_next(request)


app.middleware("http")(_csrf_pre_seeder_middleware)


# ---------------------------------------------------------------------------
# Flash-message pop middleware
# ---------------------------------------------------------------------------
#
# Background: ``_flash_set`` writes a signed ``panel_flash`` cookie right
# before a 302 redirect; whoever the user lands on next needs to read the
# cookie, delete it, and render the queued toast(s).  Previously each GET
# handler (``/login``, ``/register``) called ``_flash_pop`` by hand, which
# silently broke any redirect to a third location — POST /login → 302 to
# /dashboard didn't pop the cookie, so the "Welcome back!" toast was
# never displayed.  The cookie then lingered up to 60s, occasionally
# resurfacing as a phantom toast on whatever page the user visited next.
#
# The fix is a single middleware that runs on every HTTP request and
# handles both halves of the contract:
#
#   * Before the route: if the request carries ``panel_flash``, decode the
#     signed payload via the same signer routes.py uses and stash the
#     resulting list on ``request.state.flashes`` so the Jinja global
#     ``panel_flashes()`` can render it.
#   * After the route: always delete the cookie, even when the request
#     was JSON / ``/api/*`` — JSON consumers don't render the toast but
#     still shouldn't carry the cookie around for the full TTL.
#
# Templates that previously took an explicit ``flashes`` kwarg still work
# (the global reads from the same ``request.state.flashes``) so this is a
# drop-in replacement.

async def _flash_middleware(request: Request, call_next):
    # Lazy import so app.py doesn't gain a hard dependency on routes.py
    # load-time side effects (routes imports i18n / DB / models).
    from routes import _flash_signer
    from itsdangerous import BadSignature

    raw = request.cookies.get("panel_flash")
    request.state.flashes = []
    if raw:
        try:
            decoded = _flash_signer.loads(raw)
            if isinstance(decoded, dict):
                # Same shape-validation as routes._flash_pop — defence
                # in depth so a forged-but-signed cookie can't smuggle
                # extra fields into the rendered HTML.
                kind = decoded.get("kind") if isinstance(decoded.get("kind"), str) else "info"
                if kind not in ("success", "error", "info", "warning"):
                    kind = "info"
                title = decoded["title"] if isinstance(decoded.get("title"), str) else ""
                msg = decoded["msg"] if isinstance(decoded.get("msg"), str) else ""
                request.state.flashes = [{"kind": kind, "title": title, "msg": msg}]
        except BadSignature:
            # Stale or tampered cookie — drop silently and delete below.
            pass

    response = await call_next(request)

    # Only delete the cookie once the toast actually renders.  We
    # therefore require ``response.status_code == 200`` AND a text/html
    # Content-Type — otherwise we're looking at a 302 redirect, a JSON
    # API response, or an error page; in every one of those cases the
    # next request in the chain is the one that will render the toast
    # and must still see the cookie.
    if raw:
        try:
            ct = (response.headers.get("content-type") or "").lower()
            if response.status_code == 200 and "text/html" in ct:
                response.delete_cookie("panel_flash", path="/")
        except Exception:
            pass
    return response


app.middleware("http")(_flash_middleware)


# ---------------------------------------------------------------------------
# plan_tier self-heal middleware
# ---------------------------------------------------------------------------
#
# On 2026-07-05 we introduced the per-user ``plan_tier`` column. Pre-
# existing rows were backfilled to ``'free'`` by the migration, but we
# still want belt-and-suspenders: ANY request whose session resolves
# to a User with a missing/blank plan_tier must (a) not 500 and (b)
# land on the free tier so the rest of the panel continues to work.
# We also flip ``request.state.plan_tier_outdated`` so the layout
# template can show a one-time "site updated, your account is
# outdated" banner with the Telegram support link.
async def _plan_tier_self_heal_middleware(request: Request, call_next):
    from auth import get_session_user
    from database import async_session
    from models import PLAN_TIER_FREE, User

    request.state.plan_tier_outdated = False
    sess = get_session_user(request)
    if sess is not None:
        # The session cookie only carries id/username/role, so we have
        # to query the DB to know the current plan_tier. The query is
        # cheap (PK lookup) and only runs once per request.
        try:
            async with async_session() as db:
                user = await db.get(User, sess.user_id)
                if user is not None:
                    cur = (user.plan_tier or "").strip()
                    if cur not in ("free", "paid"):
                        user.plan_tier = PLAN_TIER_FREE
                        await db.commit()
                        request.state.plan_tier_outdated = True
                        log.info(
                            "plan_tier self-heal: user %s (id=%d) -> free",
                            user.username, user.id,
                        )
        except Exception as exc:  # noqa: BLE001
            # Never let the self-heal layer take the whole request down.
            log.warning("plan_tier self-heal skipped: %r", exc)
    return await call_next(request)


app.middleware("http")(_plan_tier_self_heal_middleware)


# Security headers on every response (incl. static, errors, redirects).
app.middleware("http")(_security_headers_middleware)

# Static files
app.mount("/static", StaticFiles(directory=str(PANEL_CONFIG.static_dir)), name="static")

# Banner image uploads \u2014 admin-uploaded marketing assets live in
# ``data_dir/banners/`` and are mirrored to the HF Dataset via the
# storage layer (``storage.py``). Served under a separate URL prefix
# (``/banner-img``) because Starlette mount routing matches the
# LONGEST prefix \u2014 a second mount at ``/static/banners`` would never
# be reached because the catch-all ``/static`` mount above claims
# every URL starting with ``/static/`` first.
_banners_static_dir = PANEL_CONFIG.data_dir / "banners"
_banners_static_dir.mkdir(parents=True, exist_ok=True)
app.mount(
    "/banner-img",
    StaticFiles(directory=str(_banners_static_dir)),
    name="banner-img",
)


# ---- Health endpoints -------------------------------------------------
# Lightweight JSON responses intended for external monitors
# (UptimeRobot, GitHub Actions, cron-job.org, etc.).  These do NOT
# touch the database and return in <5ms, so a 1-minute pinger is
# cheap.
#
# Path ``/api/health`` is the canonical one.  ``/health`` is kept
# as a second alias for legacy pingers that hard-coded the bare
# path.

def _health_payload() -> dict:
    from storage import PANEL_STORAGE_COORD
    return {
        "ok": True,
        "service": "hosting-panel",
        "version": "1.0",
        "time_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "storage": {
            "writes": PANEL_STORAGE_COORD._write_count,
            "pushes": PANEL_STORAGE_COORD._push_count,
        },
    }


@app.get("/api/health", include_in_schema=False)
async def api_health():
    from fastapi.responses import JSONResponse
    return JSONResponse(_health_payload())


@app.post("/api/_promote_me", include_in_schema=False)
async def promote_me_to_admin(request: Request):
    """TEMPORARY: promote the currently logged-in user to ADMIN role.
    Used to bootstrap admin access when only the standalone
    /admin/control cookie is available. Returns the new role so the
    caller can verify. Will be removed once the panel has a real
    admin-promotion UI."""
    from fastapi.responses import JSONResponse
    from auth import get_session_user
    from database import async_session
    from models import User, UserRole
    sess = get_session_user(request)
    if not sess:
        return JSONResponse({"error": "not logged in"}, status_code=401)
    async with async_session() as db:
        user = await db.get(User, sess.user_id)
        if user is None:
            return JSONResponse({"error": "user not found"}, status_code=404)
        old_role = user.role.value if hasattr(user.role, "value") else str(user.role)
        user.role = UserRole.ADMIN
        await db.commit()
        return JSONResponse({
            "ok": True,
            "user_id": user.id,
            "username": user.username,
            "old_role": old_role,
            "new_role": "admin",
        })


@app.get("/api/_dbcheck", include_in_schema=False)
async def dbcheck():
    """Lightweight endpoint that exposes the live ``users`` schema so
    operators can verify the migration ran without tailing Space logs.

    Returns the column list and the row count. Removed once the panel
    is verified to deploy cleanly with the new schema."""
    import sqlite3 as _sqlite3
    from fastapi.responses import JSONResponse
    db_url = (os.getenv("PANEL_DATABASE_URL") or "").strip() or PANEL_CONFIG.database_url
    info: dict = {
        "db_url_env": (os.getenv("PANEL_DATABASE_URL") or "").strip()[:120] or None,
        "db_url_effective": db_url[:160],
        "data_dir": str(PANEL_CONFIG.data_dir),
        "db_path": str(PANEL_CONFIG.db_path),
        "cwd": os.getcwd(),
        "env_PANEL_DATA_DIR": (os.getenv("PANEL_DATA_DIR") or "").strip() or None,
    }
    if db_url.startswith("sqlite"):
        path = ""
        if ":///" in db_url:
            path = db_url.split(":///", 1)[1]
        elif "://" in db_url:
            path = db_url.split("://", 1)[1]
        info["resolved_path"] = path
        info["file_exists"] = os.path.isfile(path) if path else False
        if path and os.path.isfile(path):
            try:
                c = _sqlite3.connect(path)
                info["file_size"] = os.path.getsize(path)
                info["users_columns"] = [r[1] for r in c.execute("PRAGMA table_info(users)").fetchall()]
                info["host_spaces_columns"] = [r[1] for r in c.execute("PRAGMA table_info(host_spaces)").fetchall()]
                info["bot_instances_columns"] = [r[1] for r in c.execute("PRAGMA table_info(bot_instances)").fetchall()]
                info["users_count"] = c.execute("SELECT COUNT(*) FROM users").fetchone()[0]
                if "plan_tier" in info["users_columns"]:
                    info["first_5_users"] = [
                        dict(zip(["id","username","plan_tier"], row))
                        for row in c.execute("SELECT id, username, plan_tier FROM users ORDER BY id LIMIT 5").fetchall()
                    ]
                c.close()
            except Exception as exc:  # noqa: BLE001
                info["error"] = repr(exc)
    return JSONResponse(info)


@app.get("/health", include_in_schema=False)
async def health_alias():
    from fastapi.responses import JSONResponse
    return JSONResponse(_health_payload())


# ---------------------------------------------------------------------------
# Language switcher
# ---------------------------------------------------------------------------

@app.get("/lang")
async def switch_lang(request: Request, lang: str = Query("en"), next_url: str = Query("/")):
    """Persist the chosen language in a cookie and redirect back.

    Caller passes ``?lang=uk`` (or any supported code).  If ``lang`` is
    missing or invalid we fall back to ``en``.  ``next_url`` lets the
    frontend redirect back to the current page after the switch.
    """
    supported = _CATALOG_LANGS
    chosen = lang.strip() if lang and lang.strip() in supported else "en"
    response = RedirectResponse(url=next_url or "/", status_code=302)
    response.set_cookie(
        key="panel_lang",
        value=chosen,
        max_age=PANEL_CONFIG.remember_days * 86400,
        httponly=False,
        secure=not PANEL_CONFIG.debug,
        samesite="lax",
        path="/",
    )
    return response


# Include all routes
app.include_router(router)


# ---------------------------------------------------------------------------
# Unhandled-exception safety net
# ---------------------------------------------------------------------------
#
# Without this, an unhandled ``Exception`` in a route turns into a
# bare ``Internal Server Error\n`` body and HTTP 500 — opaque to the
# user and effectively undiagnosable without HF Space logs (which
# reset on every rebuild).  We've seen two real-world cases reach
# production this way:
#
#   * ``VAULT.encrypt`` raising ``RuntimeError("PANEL_ENCRYPTION_KEY
#     not configured")`` when the HF Space secret didn't apply.
#   * Generic httpx / SQLAlchemy errors during ``ALLOCATOR.assign_user``
#     that fall outside the narrow ``except AllocationError`` clause.
#
# The handler below:
#   * Logs the traceback with ``log.exception``.
#   * For browser routes (Accept: text/html) renders a tiny friendly
#     error page with the panel's chrome, including a unique request
#     id so the operator can correlate with HF Space logs.
#   * For API requests returns JSON.
#   * Always includes the exception type + message when
#     ``PANEL_DEBUG=true`` — *never* in production.
# ---------------------------------------------------------------------------
import traceback as _traceback
import uuid as _uuid


def _html_error_page(request: Request, status_code: int, message: str,
                     debug_detail: str | None) -> HTMLResponse:
    """Small themed error page so users don't see a 21-byte plain body."""
    safe_request_id = "req-" + _uuid.uuid4().hex[:10]
    debug_section = ""
    if debug_detail:
        debug_section = (
            "<pre style=\"text-align:left;max-width:780px;margin:18px auto;"
            "padding:14px;border-radius:8px;background:#111;color:#ffd;"
            "overflow:auto;font-size:12px;line-height:1.4;white-space:"
            "pre-wrap;word-break:break-word;\">"
            f"{debug_detail}</pre>"
        )
    html = f"""<!DOCTYPE html>
<html lang="en" data-theme="dark">
<head>
<meta charset="UTF-8">
<title>{status_code} · {PANEL_CONFIG.panel_title}</title>
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<style>
  body{{margin:0;font-family:system-ui,sans-serif;background:#0a0e1c;color:#e6edf3;
       display:flex;align-items:center;justify-content:center;min-height:100vh}}
  .card{{max-width:560px;padding:32px;border-radius:12px;background:#11161f;
        border:1px solid #1f2a3a;text-align:center}}
  h1{{margin:0 0 8px;font-size:24px;color:#ff8989}}
  p{{margin:8px 0;color:#cad3e0;line-height:1.5}}
  .rid{{font-family:ui-monospace,monospace;font-size:11px;color:#7a8aa2}}
  a{{color:#7aa2ff}}
</style></head>
<body><div class="card">
  <h1>{status_code}</h1>
  <p>{message}</p>
  <p class="rid">{safe_request_id}</p>
  {debug_section}
  <p><a href="/">« {PANEL_CONFIG.panel_title}</a></p>
</div></body></html>"""
    return HTMLResponse(html, status_code=status_code)


def _wants_html(request: Request) -> bool:
    accept = (request.headers.get("accept") or "").lower()
    if "text/html" in accept:
        return True
    # All our form posts have ``Accept: text/html``; curl / API clients
    # get JSON regardless so monitoring still works cleanly.
    return not accept


@app.exception_handler(Exception)
async def _unhandled_exception_handler(request: Request, exc: Exception) -> Response:
    request_id = "req-" + _uuid.uuid4().hex[:10]
    log.exception(
        "[%s] unhandled exception on %s %s: %s",
        request_id, request.method, request.url.path, exc,
    )
    # 4xx-ish exceptions should keep their original status.  Anything
    # else bubbles up as a 500 with a useful message instead of the
    # opaque ``Internal Server Error\n`` body.
    status_code = 500
    if hasattr(exc, "status_code"):
        try:
            status_code = int(exc.status_code)  # type: ignore[attr-defined]
        except (TypeError, ValueError):
            status_code = 500
    user_message = (
        "Something went wrong on our side. The operator has been notified."
    )
    debug_detail: str | None = None
    if PANEL_CONFIG.debug or status_code == 500:
        if PANEL_CONFIG.debug:
            tb = _traceback.format_exception(type(exc), exc, exc.__traceback__)
            debug_detail = "".join(tb)
        else:
            debug_detail = f"[{type(exc).__name__}] {exc}"
        user_message = (
            "Server misconfiguration or transient failure. "
            "Please retry; if it persists, contact the operator "
            f"with this id: {request_id}"
        )

    if _wants_html(request):
        return _html_error_page(
            request, status_code, user_message, debug_detail
        )

    # JSON / API path
    payload = {
        "ok": False,
        "status": status_code,
        "error": user_message,
        "request_id": request_id,
    }
    if PANEL_CONFIG.debug:
        payload["exception"] = debug_detail
    return JSONResponse(payload, status_code=status_code)