#!/usr/bin/env python3
"""cPanel Passenger WSGI entrypoint for the SPG support bot webhook mode.

Telegram updates are first committed to SQLite before HTTP 200 is returned. A background
async worker feeds the durable inbox to python-telegram-bot. Run exactly one cPanel app
process with SQLite; for multi-process/million-user workloads, use PostgreSQL + workers on VPS.
"""
from __future__ import annotations

import asyncio
import hmac
import json
import logging
import os
import threading
import time
from datetime import datetime
from typing import Any

import config
import spg_support_bot as support
from telegram import Update

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | spg-webhook | %(message)s",
)
log = logging.getLogger("spg_webhook")
MAX_BODY_BYTES = 2 * 1024 * 1024

_loop: asyncio.AbstractEventLoop | None = None
_thread: threading.Thread | None = None
_ready = threading.Event()
_start_error: str | None = None
_ptb_app = None


def _wsgi_response(start_response, status: str, body: str, content_type: str = "text/plain; charset=utf-8"):
    data = body.encode("utf-8")
    start_response(status, [("Content-Type", content_type), ("Content-Length", str(len(data))), ("Cache-Control", "no-store")])
    return [data]


def _claim_webhook_job():
    now = time.time()
    with support.db_connect() as conn:
        conn.execute("BEGIN IMMEDIATE")
        # Recover an update interrupted by a crashed worker.
        conn.execute(
            "UPDATE webhook_inbox SET status='pending', locked_at=NULL WHERE status='processing' AND locked_at<?",
            (now - 120,),
        )
        row = conn.execute(
            "SELECT * FROM webhook_inbox WHERE status='pending' AND next_attempt_at<=? ORDER BY update_id LIMIT 1",
            (now,),
        ).fetchone()
        if row:
            conn.execute(
                "UPDATE webhook_inbox SET status='processing', locked_at=? WHERE update_id=? AND status='pending'",
                (now, row["update_id"]),
            )
        conn.commit()
    return row


async def _webhook_inbox_worker(ptb_app):
    log.info("Webhook inbox worker started")
    while True:
        job = _claim_webhook_job()
        if not job:
            await asyncio.sleep(0.2)
            continue
        try:
            payload = json.loads(job["payload"])
            update = Update.de_json(payload, ptb_app.bot)
            await ptb_app.process_update(update)
            with support.db_connect() as conn:
                # Clear raw webhook contents after processing; transcript storage has its own policy.
                conn.execute(
                    "UPDATE webhook_inbox SET status='done', processed_at=?, payload='{}', locked_at=NULL, last_error=NULL WHERE update_id=?",
                    (support.now_text(), job["update_id"]),
                )
        except Exception as exc:
            attempts = int(job["attempts"]) + 1
            delay = min(300, 2 ** min(attempts, 8))
            with support.db_connect() as conn:
                conn.execute(
                    "UPDATE webhook_inbox SET status='pending', attempts=?, next_attempt_at=?, locked_at=NULL, last_error=? WHERE update_id=?",
                    (attempts, time.time() + delay, str(exc)[:700], job["update_id"]),
                )
            log.exception("Webhook update %s processing failed; retained for retry", job["update_id"])
            await asyncio.sleep(0.1)


async def _boot_runtime():
    global _ptb_app
    support.validate_config()
    # Generated automatically and persisted privately; no fourth config value is required.
    secret = config.get_or_create_webhook_secret()
    if not secret or len(secret) < 32:
        raise RuntimeError("Could not create/read private webhook secret. Check application-root write permissions.")
    support.db_init()
    _ptb_app = support.build_application()
    await _ptb_app.initialize()
    await _ptb_app.start()
    # Application.initialize() does not execute PTB's post_init callback; start durable workers explicitly.
    await support.post_init(_ptb_app)
    # Two consumers increase throughput modestly while the transactional SQLite claim prevents
    # two workers from taking the same update. Keep a single WSGI process for SQLite deployments.
    for worker_no in range(1, 3):
        asyncio.create_task(_webhook_inbox_worker(_ptb_app), name=f"spg-webhook-inbox-{worker_no}")


def _loop_thread():
    global _loop, _start_error
    _loop = asyncio.new_event_loop()
    asyncio.set_event_loop(_loop)
    try:
        _loop.run_forever()
    finally:
        _loop.close()


def _ensure_runtime():
    global _thread, _start_error
    if _ready.is_set():
        return
    if _thread is None:
        _thread = threading.Thread(target=_loop_thread, name="spg-telegram-loop", daemon=True)
        _thread.start()
        # Wait for the event loop object to be available.
        deadline = time.time() + 10
        while _loop is None and time.time() < deadline:
            time.sleep(0.02)
        if _loop is None:
            _start_error = "event loop startup timed out"
            _ready.set()
            return
        future = asyncio.run_coroutine_threadsafe(_boot_runtime(), _loop)
        try:
            future.result(timeout=30)
        except Exception as exc:
            _start_error = str(exc)
            log.exception("SPG webhook runtime failed to initialize")
        _ready.set()


# Start once for each Passenger process. For SQLite deployment, configure a single app process.
_ensure_runtime()


def application(environ, start_response):
    """PEP-3333 WSGI callable expected by cPanel/Passenger."""
    if not _ready.is_set() or _start_error or _ptb_app is None:
        return _wsgi_response(start_response, "503 Service Unavailable", "SPG webhook service unavailable")

    method = environ.get("REQUEST_METHOD", "GET").upper()
    path = environ.get("PATH_INFO", "/")
    if method == "GET" and path in ("/", "/health"):
        return _wsgi_response(start_response, "200 OK", '{"ok":true,"mode":"webhook"}', "application/json; charset=utf-8")
    if path != "/webhook":
        return _wsgi_response(start_response, "404 Not Found", "Not found")
    if method != "POST":
        return _wsgi_response(start_response, "405 Method Not Allowed", "POST only")

    supplied = environ.get("HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN", "")
    expected = config.get_or_create_webhook_secret()
    if not supplied or not hmac.compare_digest(supplied, expected):
        return _wsgi_response(start_response, "403 Forbidden", "Forbidden")

    try:
        length = int(environ.get("CONTENT_LENGTH") or "0")
    except ValueError:
        length = 0
    if length < 1 or length > MAX_BODY_BYTES:
        return _wsgi_response(start_response, "413 Payload Too Large" if length > MAX_BODY_BYTES else "400 Bad Request", "Invalid request size")
    try:
        body = environ["wsgi.input"].read(length)
        payload = json.loads(body.decode("utf-8"))
        update_id = payload.get("update_id") if isinstance(payload, dict) else None
        if not isinstance(update_id, int):
            return _wsgi_response(start_response, "400 Bad Request", "Missing update_id")
        with support.db_connect() as conn:
            conn.execute(
                "INSERT OR IGNORE INTO webhook_inbox (update_id, payload, status, attempts, next_attempt_at, received_at) VALUES (?, ?, 'pending', 0, 0, ?)",
                (update_id, json.dumps(payload, ensure_ascii=False, separators=(",", ":")), support.now_text()),
            )
        return _wsgi_response(start_response, "200 OK", "OK")
    except Exception:
        log.exception("Webhook request could not be committed to durable inbox")
        return _wsgi_response(start_response, "500 Internal Server Error", "Retry later")