"""需要预先运行的 Temporal Server；不启动服务、不调用模型，支付为本地模拟。"""
import argparse
import asyncio
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from temporalio import activity
from temporalio.client import Client
from temporalio.exceptions import ApplicationError
from temporalio.worker import Worker
from runtime_store import RuntimeStore
from temporal_workflow import RefundWorkflow


def make_activity(directory: Path):
    @activity.defn(name="apply_refund")
    def apply_refund(task_id: str, decision: dict) -> str:
        # Activity 可重试。task ID、decision ID 和支付 operation ID 都保持不变。
        store = RuntimeStore(directory)
        try:
            store.submit(task_id)
            store.decide(task_id, decision["approved"], decision["decision_id"])
            return store.advance(task_id)["status"] if decision["approved"] else "rejected"
        except (ValueError, KeyError) as error:
            raise ApplicationError(str(error), type="InvalidRequest", non_retryable=True) from error
    return apply_refund


async def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=["worker", "start", "approve", "reject", "status"])
    parser.add_argument("task_id", nargs="?", default="T-temporal")
    parser.add_argument("--address", default="localhost:7233")
    parser.add_argument("--data-dir", type=Path, default=Path(".temporal-refund-demo"))
    args = parser.parse_args()
    client = await Client.connect(args.address)
    if args.command == "worker":
        # 同步 Activity 使用独立线程池，避免阻塞异步 Worker 调度循环。
        # 本例数据库在本机，多副本部署必须替换为共享业务服务。
        with ThreadPoolExecutor(max_workers=4) as pool:
            async with Worker(client, task_queue="refund-demo", workflows=[RefundWorkflow],
                              activities=[make_activity(args.data_dir)], activity_executor=pool):
                await asyncio.Event().wait()
        return
    if args.command == "start":
        await client.start_workflow(RefundWorkflow.run, args.task_id,
                                    id=args.task_id, task_queue="refund-demo")
        print("任务已提交；在另一个终端发送审批。重复 start 不表示安全创建新业务申请。")
    else:
        handle = client.get_workflow_handle(args.task_id)
        if args.command in {"approve", "reject"}:
            await handle.signal(RefundWorkflow.decide,
                                {"approved": args.command == "approve", "decision_id": "decision-1"})
            print("审批信号已发送，业务完成情况需要继续查询。")
        else:
            print(await handle.query(RefundWorkflow.current_status))


if __name__ == "__main__":
    asyncio.run(main())