"""cPanel Passenger WSGI entrypoint using Telegram webhooks.

Startup file: passenger_wsgi.py
Application entry point: application

Webhook mode needs no permanently open terminal. Passenger may stop an idle
worker and restart it automatically on the next Telegram POST request.
"""
from __future__ import annotations

import asyncio
import atexit
import concurrent.futures
import json
import sys
import threading
import traceback
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
if str(BASE_DIR) not in sys.path:
    sys.path.insert(0, str(BASE_DIR))

ERROR_LOG = BASE_DIR / "passenger_bot_error.log"


def _write_error() -> None:
    with ERROR_LOG.open("a", encoding="utf-8") as stream:
        stream.write("\n--- webhook processing failure ---\n")
        traceback.print_exc(file=stream)


try:
    from aiogram import Bot
    from aiogram.client.default import DefaultBotProperties
    from aiogram.enums import ParseMode
    from bot import build_dispatcher
    from config import Settings

    settings = Settings.from_env()
    dispatcher = build_dispatcher()
    startup_error: Exception | None = None
except Exception as exc:
    startup_error = exc
    _write_error()
    dispatcher = None
    settings = None

# Telegram may deliver updates for many users at the same time.  A single
# process-wide lock makes every user wait for the slowest network call and was
# the main reason the bot appeared to freeze while an admin configured images.
# Keep ordering only for updates belonging to the same user; independent users
# can then be handled concurrently.
_user_locks: dict[str, threading.Lock] = {}
_user_locks_guard = threading.Lock()

# Keep one long-lived asyncio loop and one Telegram HTTP session per Passenger
# worker.  Creating and closing a Bot/HTTP session for every webhook update is
# expensive and prevents connection reuse when many users act together.
_runtime_loop: asyncio.AbstractEventLoop | None = None
_runtime_thread: threading.Thread | None = None
_runtime_bot = None
_runtime_error: Exception | None = None
_runtime_ready = threading.Event()
_runtime_start_guard = threading.Lock()


def _update_key(payload: dict) -> str:
    """Return a stable serialization key for one Telegram conversation."""
    for update_type in (
        "message", "edited_message", "channel_post", "edited_channel_post",
        "callback_query", "inline_query", "chat_join_request",
        "my_chat_member", "chat_member", "chosen_inline_result",
    ):
        update = payload.get(update_type)
        if not isinstance(update, dict):
            continue
        user = update.get("from") or {}
        if user.get("id") is not None:
            return f"user:{user['id']}"
        chat = update.get("chat") or {}
        if chat.get("id") is not None:
            return f"chat:{chat['id']}"
    # Service updates without a user are rare.  They should not block all
    # normal users, so serialize only that individual update.
    return f"update:{payload.get('update_id', id(payload))}"


def _lock_for_update(payload: dict) -> threading.Lock:
    key = _update_key(payload)
    with _user_locks_guard:
        lock = _user_locks.get(key)
        if lock is None:
            lock = threading.Lock()
            _user_locks[key] = lock
        return lock


def _runtime_worker() -> None:
    global _runtime_loop, _runtime_bot, _runtime_error
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    _runtime_loop = loop
    try:
        assert settings is not None
        _runtime_bot = Bot(
            settings.bot_token,
            default=DefaultBotProperties(parse_mode=ParseMode.HTML),
        )
    except Exception as exc:
        _runtime_error = exc
    finally:
        _runtime_ready.set()
    if _runtime_error is None:
        loop.run_forever()


def _ensure_runtime() -> asyncio.AbstractEventLoop:
    """Start the worker lazily, after Passenger has created its worker process."""
    global _runtime_thread
    if _runtime_loop is not None and _runtime_ready.is_set():
        if _runtime_error:
            raise _runtime_error
        return _runtime_loop
    with _runtime_start_guard:
        if _runtime_thread is None:
            _runtime_thread = threading.Thread(
                target=_runtime_worker,
                name="telegram-bot-loop",
                daemon=True,
            )
            _runtime_thread.start()
        if not _runtime_ready.wait(timeout=5):
            raise RuntimeError("Telegram runtime loop did not start in time")
    if _runtime_error:
        raise _runtime_error
    assert _runtime_loop is not None
    return _runtime_loop


async def _feed_update(payload: dict) -> None:
    assert dispatcher is not None and _runtime_bot is not None
    await dispatcher.feed_raw_update(_runtime_bot, payload)


def _dispatch_update(payload: dict) -> None:
    loop = _ensure_runtime()
    future = asyncio.run_coroutine_threadsafe(_feed_update(payload), loop)
    try:
        future.result(timeout=50)
    except concurrent.futures.TimeoutError:
        future.cancel()
        raise


def _shutdown_runtime() -> None:
    loop = _runtime_loop
    bot = _runtime_bot
    if loop is None or not loop.is_running():
        return

    async def close_bot() -> None:
        if bot is not None:
            await bot.session.close()

    try:
        asyncio.run_coroutine_threadsafe(close_bot(), loop).result(timeout=3)
    except Exception:
        pass
    loop.call_soon_threadsafe(loop.stop)


atexit.register(_shutdown_runtime)


def _response(start_response, status: str, body: str):
    encoded = body.encode("utf-8")
    start_response(status, [
        ("Content-Type", "text/plain; charset=utf-8"),
        ("Content-Length", str(len(encoded))),
    ])
    return [encoded]


def application(environ, start_response):
    if startup_error is not None:
        return _response(start_response, "500 Internal Server Error", "Bot startup failed; read passenger_bot_error.log")

    method = environ.get("REQUEST_METHOD", "GET").upper()
    if method == "GET":
        return _response(start_response, "200 OK", "Telegram webhook application is running")
    if method != "POST":
        return _response(start_response, "405 Method Not Allowed", "Method not allowed")

    expected_secret = settings.webhook_secret if settings else None
    received_secret = environ.get("HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN", "")
    if expected_secret and received_secret != expected_secret:
        return _response(start_response, "403 Forbidden", "Invalid webhook secret")

    try:
        length = int(environ.get("CONTENT_LENGTH") or "0")
        if length <= 0 or length > 2_000_000:
            return _response(start_response, "400 Bad Request", "Invalid request body")
        payload = json.loads(environ["wsgi.input"].read(length).decode("utf-8"))
        # Process each update synchronously in this request so Telegram gets a
        # definitive result, but serialize only the same user's updates.  Do
        # not use one global lock here: a campaign/image request for one user
        # must not block the other 50-60 active users.
        with _lock_for_update(payload):
            _dispatch_update(payload)
    except Exception:
        _write_error()
        return _response(start_response, "500 Internal Server Error", "Update failed")

    return _response(start_response, "200 OK", "OK")