harness_zbt/backend/app/infrastructure/redis/keys.py

class RedisKeyBuilder:
    def __init__(self, prefix: str) -> None:
        if not prefix.endswith(":"):
            raise ValueError("Redis Key 前缀必须以冒号结尾")
        self._prefix = prefix

    def auth_otp(self, mobile_hash: str) -> str:
        return self._key("auth", "otp", mobile_hash)

    def auth_session(self, session_id: str) -> str:
        return self._key("auth", "session", session_id)

    def auth_refresh(self, jti_hash: str) -> str:
        return self._key("auth", "refresh", jti_hash)

    def login_rate(self, subject_hash: str) -> str:
        return self._key("rate", "login", subject_hash)

    def agent_rate(self, subject_id: str) -> str:
        return self._key("rate", "agent", subject_id)

    def agent_thread_hot(self, thread_id: str) -> str:
        return self._key("agent", "thread", thread_id, "hot")

    def agent_run_events(self, run_id: str) -> str:
        return self._key("agent", "run", run_id, "events")

    def agent_run_cancel(self, run_id: str) -> str:
        return self._key("agent", "run", run_id, "cancel")

    def payment_lock(self, payment_id: str) -> str:
        return self._key("lock", "payment", payment_id)

    def issue_lock(self, order_id: str) -> str:
        return self._key("lock", "issue", order_id)

    def product_cache(self, scope_hash: str) -> str:
        return self._key("cache", "product", "available", scope_hash)

    def metrics_cache(self, metric_hash: str) -> str:
        return self._key("cache", "metrics", metric_hash)

    def worker_wakeup(self, queue: str) -> str:
        return self._key("worker", "wakeup", queue)

    def _key(self, *segments: str) -> str:
        for segment in segments:
            if not segment or ":" in segment:
                raise ValueError("Redis Key 片段不能为空或包含冒号")
        return self._prefix + ":".join(segments)