""" Approvers 白名單管理 — ADR-093 群組遷移安全模型 chat_member 事件 → 同步 Redis Set hermes:approvers:{group_id} """ from __future__ import annotations import structlog from typing import TYPE_CHECKING if TYPE_CHECKING: pass logger = structlog.get_logger() APPROVERS_KEY_PREFIX = "hermes:approvers" def _approvers_key(group_id: str | int) -> str: return f"{APPROVERS_KEY_PREFIX}:{group_id}" async def sync_member_joined(redis, group_id: str | int, user_id: int) -> None: """成員加入群組 → 加入 Approvers Redis Set""" key = _approvers_key(group_id) await redis.sadd(key, str(user_id)) logger.info("approvers_member_joined", group_id=group_id, user_id=user_id) async def sync_member_left(redis, group_id: str | int, user_id: int) -> None: """成員離開群組 → 從 Approvers Redis Set 移除""" key = _approvers_key(group_id) await redis.srem(key, str(user_id)) logger.info("approvers_member_left", group_id=group_id, user_id=user_id) async def get_approvers(redis, group_id: str | int) -> set[int]: """取得群組 Approvers 集合(空集 = 未初始化,退回 config whitelist)""" key = _approvers_key(group_id) raw = await redis.smembers(key) return {int(uid) for uid in raw if uid} async def is_approved_member(redis, group_id: str | int, user_id: int) -> bool: """ user_id 是否在群組 Approvers Set 中。 若 Redis Set 為空(未初始化),回傳 True 讓 caller 退回 config whitelist。 """ approvers = await get_approvers(redis, group_id) if not approvers: return True # 未初始化 → 不阻擋,由 config whitelist 把關 return user_id in approvers