"""Recharge reconciliation scheduler.

Every minute, scan pending recharge orders:
- skip very new orders to give provider callbacks time to arrive;
- query the payment provider while the local pending window is still open;
- expire old pending orders using the configured payment TTL.

Late provider callbacks are still handled by recharge_service.process_paid().
"""
from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
from sqlalchemy import select

from app.database import async_session
from app.models.recharge_order import RechargeOrder, RechargeStatus
from app.redis_client import k, redis_client
from app.services import payment_service, system_setting_service
from app.services.recharge_service import mark_expired, process_paid

log = logging.getLogger("qycw.recharge.scheduler")

POLL_INTERVAL_SECONDS = 60
MIN_AGE_SECONDS = 120
DEFAULT_PENDING_PAYMENT_TTL_MINUTES = 120


def _lock_key(out_trade_no: str) -> str:
    return k("recharge", "lock", out_trade_no)


async def _get_pending_payment_ttl_minutes() -> int:
    try:
        async with async_session() as db:
            async with db.begin():
                business_settings = await system_setting_service.get_business_settings(db)
                return max(1, int(business_settings.pending_payment_ttl_minutes))
    except Exception:
        log.exception(
            "recharge scheduler: failed to load payment ttl, fallback=%s",
            DEFAULT_PENDING_PAYMENT_TTL_MINUTES,
        )
        return DEFAULT_PENDING_PAYMENT_TTL_MINUTES


async def _job_poll_pending() -> None:
    log.debug("recharge scheduler: tick")
    now = datetime.now(timezone.utc)
    min_age_cutoff = now - timedelta(seconds=MIN_AGE_SECONDS)
    pending_payment_ttl_minutes = await _get_pending_payment_ttl_minutes()
    expire_cutoff = now - timedelta(minutes=pending_payment_ttl_minutes)

    async with async_session() as db:
        async with db.begin():
            res = await db.execute(
                select(RechargeOrder)
                .where(
                    RechargeOrder.status == RechargeStatus.PENDING,
                    RechargeOrder.created_at <= min_age_cutoff,
                )
                .order_by(RechargeOrder.created_at.asc())
                .limit(100)
            )
            orders = res.scalars().all()

    if not orders:
        return

    log.debug("recharge scheduler: %d pending orders to check", len(orders))

    for order in orders:
        out_trade_no = order.out_trade_no
        created_at = order.created_at
        if created_at.tzinfo is None:
            created_at = created_at.replace(tzinfo=timezone.utc)

        if created_at <= expire_cutoff:
            lock_key = _lock_key(out_trade_no)
            acquired = await redis_client.set(lock_key, "1", nx=True, ex=30)
            if not acquired:
                continue
            try:
                async with async_session() as db:
                    async with db.begin():
                        await mark_expired(out_trade_no, db)
            except Exception:
                log.exception("recharge scheduler: mark_expired failed for %s", out_trade_no)
            finally:
                await redis_client.delete(lock_key)
            continue

        try:
            async with async_session() as config_db:
                provider, _plugin, _config = await payment_service.get_payment_provider(
                    config_db,
                    order.payment_plugin or "epay_alipay",
                )
            query_result = await provider.query_order(out_trade_no)
        except Exception as e:
            log.warning("recharge scheduler: query_order failed for %s: %s", out_trade_no, e)
            continue

        if not query_result.paid:
            log.debug(
                "recharge scheduler: %s not paid yet (status=%s)",
                out_trade_no,
                query_result.status,
            )
            continue

        lock_key = _lock_key(out_trade_no)
        acquired = await redis_client.set(lock_key, "1", nx=True, ex=30)
        if not acquired:
            log.debug("recharge scheduler: %s lock held by another process", out_trade_no)
            continue

        try:
            async with async_session() as db:
                ok = await process_paid(
                    out_trade_no=out_trade_no,
                    trade_no=query_result.trade_no,
                    money=query_result.amount_yuan,
                    db=db,
                    source="scheduler",
                    raw_params=query_result.raw,
                    expected_payment_plugin=order.payment_plugin or "epay_alipay",
                    expected_payment_method=order.payment_method or str(getattr(provider, "method", "") or ""),
                )
            if ok:
                log.debug("recharge scheduler: %s paid and credited", out_trade_no)
        except Exception:
            log.exception("recharge scheduler: process_paid failed for %s", out_trade_no)
        finally:
            await redis_client.delete(lock_key)


_scheduler: AsyncIOScheduler | None = None


def start_recharge_scheduler() -> None:
    global _scheduler
    if _scheduler is not None:
        return
    sched = AsyncIOScheduler(timezone="UTC")
    sched.add_job(
        _job_poll_pending,
        IntervalTrigger(seconds=POLL_INTERVAL_SECONDS),
        id="recharge_poll_pending",
        max_instances=1,
        coalesce=True,
        misfire_grace_time=30,
    )
    sched.start()
    _scheduler = sched
    log.debug("recharge scheduler started: interval=%ds", POLL_INTERVAL_SECONDS)


def shutdown_recharge_scheduler() -> None:
    global _scheduler
    if _scheduler is None:
        return
    _scheduler.shutdown(wait=False)
    _scheduler = None
