会员权益时灵时不灵：权限变更的缓存穿透与即时生效实践

def get_user_perms(user_id):
    # 第一层：角色权限缓存（键=角色ID，变更少、复用高）
    role_ids = db.fetchall("SELECT role_id FROM user_role WHERE user_id=%s", (user_id,))
    perms = set()
    for rid in role_ids:
        key = f"role:perms:{rid}"
        rp = redis.smembers(key)
        if not rp:  # 缓存未命中才查库并回填
            rp = db.fetchall("SELECT perm_code FROM role_perm WHERE role_id=%s", (rid,))
            redis.sadd(key, *rp); redis.expire(key, 3600)
        perms |= set(rp)
    # 第二层：用户级单独授权（如单用户VIP、临时权限）
    user_key = f"user:perms:{user_id}"
    extra = redis.smembers(user_key)
    if not extra:
        extra = db.fetchall("SELECT perm_code FROM user_perm WHERE user_id=%s", (user_id,))
        redis.sadd(user_key, *extra); redis.expire(user_key, 600)
        extra = set(extra)
    return perms | set(extra)

# 权限变更方（后台管理服务）
def update_role_perm(role_id, new_perm_codes):
    db.execute("UPDATE role_perm SET perm_code=%s WHERE role_id=%s", ...)
    # 清缓存 + 广播变更
    redis.delete(f"role:perms:{role_id}")
    redis.publish("perm:change", json.dumps({"type": "role", "id": role_id}))

# 各实例监听变更，主动清除本地可能存在的旧权限副本
def listen_perm_change():
    pubsub = redis.pubsub()
    pubsub.subscribe("perm:change")
    for msg in pubsub.listen():
        data = json.loads(msg["data"])
        if data["type"] == "role":
            local_cache.pop(f"role:perms:{data['id']}", None)  # 本地副本一并清

# 1. 互斥重建：同角色同一时间只允许一个请求回填缓存
def rebuild_role_perm(role_id):
    if redis.set(f"rebuild:lock:{role_id}", 1, nx=True, ex=5):
        try:
            perms = db.fetchall("SELECT perm_code FROM role_perm WHERE role_id=%s", (role_id,))
            redis.delete(f"role:perms:{role_id}")
            if perms:
                redis.sadd(f"role:perms:{role_id}", *perms); redis.expire(f"role:perms:{role_id}", 3600)
            else:
                redis.set(f"role:empty:{role_id}", 1, ex=60)  # 空结果也缓存，防击穿
        finally:
            redis.delete(f"rebuild:lock:{role_id}")

# 2. 降级：缓存/Redis 全挂时，直接查库但不写缓存，保证功能可用
def get_user_perms_safe(user_id):
    try:
        return get_user_perms(user_id)
    except RedisError:
        return db.fetchall("SELECT perm_code FROM user_role ur JOIN role_perm rp ON ur.role_id=rp.role_id WHERE ur.user_id=%s", (user_id,))

CREATE TABLE perm_audit_log (
  id        BIGINT PRIMARY KEY AUTO_INCREMENT,
  operator  VARCHAR(64) NOT NULL,      -- 谁改的
  action    VARCHAR(32) NOT NULL,      -- grant/revoke/update
  target_type VARCHAR(16) NOT NULL,    -- role/user
  target_id BIGINT NOT NULL,
  perm_codes JSON NOT NULL,            -- 改了哪些权限
  source_ip VARCHAR(64),
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  KEY idx_target (target_type, target_id, created_at)
);

def audit(operator, action, target_type, target_id, perm_codes, ip):
    db.execute(
        "INSERT INTO perm_audit_log(operator, action, target_type, target_id, perm_codes, source_ip) "
        "VALUES(%s,%s,%s,%s,%s,%s)",
        (operator, action, target_type, target_id, json.dumps(perm_codes), ip),
    )