"""无障碍保护存活监控后台任务。

周期扫描「曾经保护过、当前 alive、心跳超时」的设备 = App 被彻底杀掉/无障碍已停(心跳断了),
**命中即在服务器终端打印告警**(本期先不接推送,工程量大,用终端打印代替真实通知);并把状态机
推进到 notified 防每轮重复打印(心跳恢复时由 repositories.device.touch_heartbeat 重置回 alive)。
结构仿 withdraw_reconcile_worker(单实例锁 + asyncio 轮询 + 优雅退出)。

见 spec: spec/accessibility-liveness-push.md。
"""
from __future__ import annotations

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

from sqlalchemy.exc import SQLAlchemyError

from app.core.config import settings
from app.db.session import SessionLocal
from app.repositories import device as device_repo

logger = logging.getLogger("shagua.heartbeat_monitor")
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "heartbeat_monitor.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 _silent_seconds(last: datetime | None) -> int | None:
    """距上次心跳的秒数(兼容 sqlite 取回的 naive datetime)。"""
    if last is None:
        return None
    ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow()
    return int((ref - last).total_seconds())


def _scan_once(timeout_minutes: int) -> dict:
    """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备,在**服务器终端打印**告警代替真实推送。

    本期不接推送(极光/厂商通道工程量大),只做服务端掉线检测:命中即 logger.warning 打印到终端,
    并把状态机推进到 notified 防每轮重复打印(心跳恢复时 touch_heartbeat 会重置回 alive)。
    """
    notified = 0
    with SessionLocal() as db:
        overdue = device_repo.list_overdue(db, timeout_minutes=timeout_minutes)
        for device in overdue:
            silent = _silent_seconds(device.last_heartbeat_at)
            logger.warning(
                "[掉线检测] user_id=%s device_id=%s 已 %s 秒无心跳(阈值 %d 分钟)"
                " → 判定 App 已被杀/无障碍已停。【已置 kill_alert_pending: 用户下次进 App 将弹「开启自启动」引导(后置检测);推送本期未接】",
                device.user_id,
                device.device_id,
                silent if silent is not None else "?",
                timeout_minutes,
            )
            device_repo.mark_notified(db, device_id_pk=device.id)
            notified += 1
        return {"checked": len(overdue), "notified": notified}


async def _run_loop() -> None:
    interval = max(10, int(settings.HEARTBEAT_SCAN_INTERVAL_SEC))
    timeout_minutes = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES))
    lock_stale_after = max(interval * 3, 600)
    with _single_instance_lock(lock_stale_after) as lock_acquired:
        if not lock_acquired:
            logger.warning("heartbeat monitor skipped: another worker owns lock")
            return
        await _run_locked_loop(interval, timeout_minutes)


async def _run_locked_loop(interval: int, timeout_minutes: int) -> None:
    logger.info(
        "heartbeat monitor started interval=%ss timeout=%sm",
        interval,
        timeout_minutes,
    )
    try:
        while True:
            try:
                _touch_lock()
                result = await asyncio.to_thread(_scan_once, timeout_minutes)
                if result["notified"]:
                    logger.info("heartbeat monitor result=%s", result)
            except SQLAlchemyError:
                logger.exception("heartbeat monitor db error")
            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
                logger.exception("heartbeat monitor unexpected error")
            await asyncio.sleep(interval)
    except asyncio.CancelledError:
        logger.info("heartbeat monitor stopped")
        raise


def start_heartbeat_monitor() -> asyncio.Task | None:
    if not settings.HEARTBEAT_MONITOR_ENABLED:
        logger.info("heartbeat monitor disabled")
        return None
    return asyncio.create_task(_run_loop(), name="heartbeat-monitor")


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