"""连续 N 天无活跃账户金币 / 现金清零的进程内定时任务。

结构同 app.core.daily_exchange_worker:每 `INACTIVE_CLEAR_CHECK_INTERVAL_SEC` 醒一次,
跨进北京新的一天(0 点后)就跑一轮 inactive_clear.clear_inactive_accounts。

健壮性:
- **天然幂等**:只清有余额的账户,清完余额=0,下一轮 / 重启 / 多次唤醒都不重复清。
- **当天首跑即补**:进程起来时若当天还没跑过,立即跑一轮。
- **同机多进程互斥**:文件锁保证多 worker 只有一个实际跑。
- **开关**:settings.INACTIVE_CLEAR_ENABLED=false 时不启动。

⚠️ 这是不可逆的批量资金操作(清空用户金币 + 金币现金,邀请金除外)。活跃口径与清零
细节见 app.repositories.inactive_clear。
"""
from __future__ import annotations

import asyncio
import contextlib
import logging
import os
import time
from collections.abc import Iterator
from datetime import date
from pathlib import Path

from sqlalchemy.exc import SQLAlchemyError

from app.core import rewards
from app.core.config import settings
from app.db.session import SessionLocal
from app.repositories import inactive_clear as inactive_clear_repo

logger = logging.getLogger("shagua.inactive_clear")
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "inactive_clear.lock"


def _touch_lock() -> None:
    with contextlib.suppress(FileNotFoundError):
        os.utime(_LOCK_PATH, None)


@contextlib.contextmanager
def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
    """同机多进程保护:同一时间只允许一个清零 worker 运行。"""
    _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
    fd: int | None = None
    try:
        try:
            fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        except FileExistsError:
            try:
                age = time.time() - _LOCK_PATH.stat().st_mtime
            except FileNotFoundError:
                age = stale_after_sec + 1
            if age > stale_after_sec:
                with contextlib.suppress(FileNotFoundError):
                    _LOCK_PATH.unlink()
                try:
                    fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
                except FileExistsError:
                    fd = None

        if fd is None:
            yield False
            return

        os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
        yield True
    finally:
        if fd is not None:
            os.close(fd)
            with contextlib.suppress(FileNotFoundError):
                _LOCK_PATH.unlink()


def _clear_once() -> dict:
    with SessionLocal() as db:
        return inactive_clear_repo.clear_inactive_accounts(
            db, days=settings.INACTIVE_CLEAR_DAYS
        )


async def _run_loop() -> None:
    interval = max(60, int(settings.INACTIVE_CLEAR_CHECK_INTERVAL_SEC))
    lock_stale_after = max(interval * 3, 1800)
    with _single_instance_lock(lock_stale_after) as lock_acquired:
        if not lock_acquired:
            logger.warning("inactive-clear skipped: another worker owns lock")
            return
        await _run_locked_loop(interval)


async def _run_locked_loop(interval: int) -> None:
    logger.info(
        "inactive-clear worker started interval=%ss days=%s",
        interval,
        settings.INACTIVE_CLEAR_DAYS,
    )
    # 本进程上次跑过的北京日;None=尚未跑过本进程(启动即补当天)。
    last_run: date | None = None
    try:
        while True:
            try:
                _touch_lock()
                today = rewards.cn_today()
                if last_run != today:
                    result = await asyncio.to_thread(_clear_once)
                    last_run = today
                    logger.info("inactive-clear done date=%s result=%s", today, result)
            except SQLAlchemyError:
                logger.exception("inactive-clear db error")
            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
                logger.exception("inactive-clear unexpected error")
            await asyncio.sleep(interval)
    except asyncio.CancelledError:
        logger.info("inactive-clear worker stopped")
        raise


def start_inactive_clear_worker() -> asyncio.Task | None:
    if not settings.INACTIVE_CLEAR_ENABLED:
        logger.info("inactive-clear disabled (INACTIVE_CLEAR_ENABLED=false)")
        return None
    return asyncio.create_task(_run_loop(), name="inactive-clear")


async def stop_inactive_clear_worker(task: asyncio.Task | None) -> None:
    if task is None:
        return
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task