"""示例 10：常驻任务的完整启停流程 —— **一个脚本跑通，不需要第二个终端**。

前面几轮把桶归属、跨进程、多终端全搅在一起了，所以这个示例刻意只用一条主线：

    启动 → 边跑边看（进度/心跳/租约）→ 暂停 → 恢复 → 停止 → 校验

每一步都打印"你该看什么"，最后**断言**进程内的结论与存储里的状态一致——不一致会
失败退出，不会让你盯着两段输出自己猜。

运行::

    uv run python examples/10_worker_walkthrough.py

命令行版本（需要第二个终端，因为 ``run`` 是前台进程）见文件末尾的说明。
"""

from __future__ import annotations

import sys
import time
from typing import Any

from chongming_task import Controller, SqliteTaskStore, render_table


def consume(ctx: Any, source: str, batch: int = 1000) -> int:
    """常驻消费：无限循环 + 批次进度 + 收到停止请求后优雅收尾。"""
    total = 0
    while not ctx.is_stop_requested:
        ctx.checkpoint()  # ← 不调用它 = 停不掉
        time.sleep(0.05)
        total += batch
        ctx.progress(total, unit="msg", message=source)
    return total


def main() -> int:
    # 用**文件**存储：这样存储里的状态与进程内的结论是两份数据，能对得上才有意义。
    store = SqliteTaskStore(":memory:")  # 想留下文件就把 ":memory:" 换成 "tasks.db"
    controller = Controller(store=store, owner="demo-node")
    controller.register("consume", consume)

    # ---------------------------------------------------------------- 1. 启动
    print("=" * 72)
    print("1) 启动")
    print("=" * 72)
    run = controller.submit("consume", "kafka://orders", task_id="orders-worker")
    print("   controller.submit(...) 立即返回，任务在后台跑：")
    print(f"   {run.snapshot.task_id}  状态={run.status.value}")
    print("\n   注意：`submit` 之后**必须有人让进程活着**（这里的 main 就是那个人）。")

    # ------------------------------------------------- 2. 边跑边看（两个视角）
    print("\n" + "=" * 72)
    print("2) 边跑边看：进程内的句柄 vs 存储里的事实")
    print("=" * 72)
    first = _watch(store, seconds=1.0)
    second = _watch(store, seconds=1.0)
    assert second["progress"] > first["progress"], "进度没有推进——回写链路有问题"
    print(f"\n   ✅ 进度在推进：{first['progress']} → {second['progress']}")
    print('   ✅ 租约新鲜 → 有监督者在管它（进程崩溃时这里会显示"已过期"）')

    # ---------------------------------------------------------------- 3. 暂停
    print("\n" + "=" * 72)
    print('3) pause：日常"先停一下"用这个，不要用 stop')
    print("=" * 72)
    confirmed = controller.pause("orders-worker", timeout=5)
    print(f"   pause 返回 {confirmed}（True = 已确认挂起；False = 任务没到检查点，状态已回滚）")
    print(f"   状态: {store.get('orders-worker').status.value}")  # type: ignore[union-attr]
    assert confirmed is True

    frozen = _watch(store, seconds=0.6, show_progress=False)
    print(f"\n   挂起期间进度不再推进（{frozen['progress']}）—— 但注意：它**仍占着执行槽**")

    print("\n   resume 后状态: ", end="")
    controller.resume("orders-worker")
    time.sleep(0.2)
    print(store.get("orders-worker").status.value)  # type: ignore[union-attr]

    # ---------------------------------------------------------------- 4. 停止
    print("\n" + "=" * 72)
    print('4) stop：终态、不可逆。返回的是"请求已受理"，不是"已停止"')
    print("=" * 72)
    controller.stop("orders-worker", reason="发布窗口")
    outcome = controller.wait("orders-worker", timeout=15)
    print(f"   终态: {outcome.status.value}   返回值: {outcome.value}")
    print("   ↑ succeeded 而不是 stopped：因为任务体看到 stop 请求后**正常返回**了，")
    print("     已经处理好的数量被完整保留。（让它抛 TaskCancelled 才是 stopped。）")

    # ---------------------------------------------------------------- 5. 校验
    print("\n" + "=" * 72)
    print("5) 校验：进程内的结论 与 存储里的状态 必须一致")
    print("=" * 72)
    print(render_table(controller.list_tasks()))

    stored = store.get("orders-worker")
    assert stored is not None
    assert stored.status is outcome.status, (
        f"存储={stored.status.value} 与进程内={outcome.status.value} 不一致！"
    )
    assert stored.status_version == outcome.snapshot.status_version, "版本号不一致"
    assert stored.outcome == {"value": outcome.value}, "返回值没有落库"
    print(
        f"\n   ✅ 一致：status={stored.status.value} "
        f"version={stored.status_version} outcome={stored.outcome}"
    )

    print("\n   审计轨迹（终态为什么是 succeeded，看这里就清楚了）：")
    for record in store.history("orders-worker"):
        reason = f"  原因={record.reason}" if record.reason else ""
        print(
            f"     {record.version}  {record.from_status.value:>9} -> "
            f"{record.to_status.value:<9} {record.trigger}{reason}"
        )

    failures = controller.persist_failures or "无"
    print(f"\n   写穿失败（应当为空；非空说明存储落后于内存）：{failures}")
    print("   永久分叉的任务（应当为空）：", controller.desynced_tasks or "无")
    assert controller.persist_failures == (), "有写穿失败——存储不可信"
    assert controller.desynced_tasks == {}, "有任务永久分叉"

    controller.close(timeout=10, close_store=True)

    print("\n" + "=" * 72)
    print("全部通过 ✅")
    print("=" * 72)
    _cli_recipe()
    return 0


def _watch(store: SqliteTaskStore, *, seconds: float, show_progress: bool = True) -> dict[str, Any]:
    """从**存储**采一次样（模拟另一个终端跑 `status`），返回关键字段。"""
    deadline = time.monotonic() + seconds
    snapshot = None
    while time.monotonic() < deadline:
        snapshot = store.get("orders-worker")
        time.sleep(0.1)
    assert snapshot is not None
    lease = "无"
    if snapshot.lease_until is not None:
        from chongming_task import utcnow

        remaining = (snapshot.lease_until - utcnow()).total_seconds()
        lease = f"剩余 {remaining:.0f}s" if remaining > 0 else f"已过期 {-remaining:.0f}s"
    progress = snapshot.progress.done if snapshot.progress is not None else 0
    if show_progress:
        print(
            f"   存储: status={snapshot.status.value:<9} v={snapshot.status_version} "
            f"progress={progress:<7} 心跳={_hhmmss(snapshot)} 租约={lease}"
        )
    return {"progress": progress, "status": snapshot.status.value}


def _hhmmss(snapshot: Any) -> str:
    return "-" if snapshot.last_heartbeat_at is None else str(snapshot.last_heartbeat_at)[11:19]


def _cli_recipe() -> None:
    print(
        """
命令行版本（`run` 是前台进程，所以要**两个终端**）
────────────────────────────────────────────────────────
# 终端 1：启动（前台阻塞；生产上交给 systemd / nohup）
set PYTHONPATH=examples
uv run chongming-task --db ./tasks.db run --tasks worker_tasks consume ^
    --arg source=kafka://orders --task-id orders-worker

# 终端 2：看状态 → 停止 → 再确认
uv run chongming-task --db ./tasks.db status orders-worker
uv run chongming-task --db ./tasks.db stop orders-worker --reason 发布窗口
uv run chongming-task --db ./tasks.db status orders-worker     <- 这一步看到 succeeded 才算停干净
""".rstrip()
    )
    print("  提示：Ctrl-C 也能停（CLI 会走 finally 里的 close()），但 status 确认仍然要做。\n")


if __name__ == "__main__":
    sys.exit(main())