// 幂等键构建示例
public class IdempotentKey {
    private final String bizType;      // 业务类型：MEETING_BOOK
    private final String bizId;        // 业务主键：会议单 ID（预创建阶段为 UUID）
    private final String operatorId;   // 操作人 ID
    private final int    actionCode;   // 动作码：1=创建 2=取消 3=变更
    private final long   timestamp;    // 请求时间窗口（分钟级），防止跨周期冲突
}

graph TD
    A[客户端请求] --> B[网关层：X-Request-Id 去重]
    B --> C[信令接入层：业务幂等键校验]
    C --> D[领域服务层：状态机前置条件校验]
    D --> E[持久化层：唯一索引约束兜底]

# Spring Cloud Gateway 过滤器配置
idempotent:
  enabled: true
  headerName: X-Request-Id
  ttlSeconds: 300
  redisTemplate: stringRedisTemplate

@Aspect
@Component
@RequiredArgsConstructor
public class IdempotentAspect {
    private final StringRedisTemplate redis;
    private final IdempotentKeyResolver resolver;

    @Around("@annotation(idempotent)")
    public Object handle(ProceedingJoinPoint pjp, Idempotent idempotent) throws Throwable {
        IdempotentKey key = resolver.resolve(pjp.getArgs(), idempotent);
        String redisKey = key.toRedisKey();
        
        // SET NX EX 原子操作
        Boolean acquired = redis.opsForValue()
            .setIfAbsent(redisKey, "1", Duration.ofMinutes(idempotent.ttlMinutes()));
        
        if (!acquired) {
            throw new IdempotentRejectedException("重复请求", key.getBizId());
        }
        try {
            return pjp.proceed();
        } finally {
            // 业务成功才删除，失败保留键供排查
            if (isSuccessResult()) redis.delete(redisKey);
        }
    }
}

@Entity
@Table(uniqueConstraints = @UniqueConstraint(columnNames = {"meeting_id", "version"}))
public class MeetingAggregate {
    @Version
    private Long version;  // 乐观锁版本号
    
    private MeetingStatus status; // DRAFT -> CONFIRMED -> CANCELLED
    
    public void confirm(ConfirmCommand cmd) {
        // 状态机前置条件：仅 DRAFT 可确认
        if (this.status != MeetingStatus.DRAFT) {
            throw new IllegalStateException("会议状态非草稿，不可确认");
        }
        this.status = MeetingStatus.CONFIRMED;
        this.version = cmd.getExpectedVersion(); // 乐观锁校验
    }
}

public enum SagaState {
    STARTED,           // 事务发起
    RESOURCE_LOCKED,   // 会议室/终端锁定成功
    INVITATION_SENT,   // 邀请信令投递成功
    CONFIRMED,         // 全员确认/超时自动确认
    COMPENSATING,      // 补偿进行中
    COMPENSATED,       // 补偿完成
    FAILED             // 补偿失败，需人工介入
}

@Entity
@Table(name = "saga_meeting_book")
public class MeetingBookSaga {
    @Id String sagaId;           // 业务主键即 Saga ID
    @Enumerated(EnumType.STRING) SagaState state;
    String payloadJson;          // 完整上下文，供补偿反序列化
    int retryCount;              // 重试次数
    LocalDateTime nextRetryTime; // 下次重试时间
    String failedStep;           // 失败步骤标识
}

@Component
@RequiredArgsConstructor
public class SagaOrchestrator {
    private final MeetingBookSagaRepository sagaRepo;
    private final ResourceService resourceService;
    private final MeetingRepository meetingRepo;
    private final SignalingGateway signalingGateway;
    private final MeterRegistry meter;

    @Transactional
    public void execute(MeetingBookSaga saga) {
        try {
            switch (saga.getState()) {
                case STARTED -> stepLockResource(saga);
                case RESOURCE_LOCKED -> stepPersistMeeting(saga);
                case INVITATION_SENT -> stepWaitConfirm(saga); // 实际由异步事件驱动
                default -> throw new IllegalStateException("未知状态: " + saga.getState());
            }
            sagaRepo.save(saga);
        } catch (Exception e) {
            handleFailure(saga, e);
        }
    }

    private void stepLockResource(MeetingBookSaga saga) {
        LockResult result = resourceService.lock(saga.getRoomId(), saga.getTerminalIds());
        saga.setLockToken(result.getToken());
        saga.setState(SagaState.RESOURCE_LOCKED);
        meter.counter("saga_step_success", "step", "lock").increment();
    }

    private void handleFailure(MeetingBookSaga saga, Exception e) {
        saga.setState(SagaState.COMPENSATING);
        saga.setFailedStep(saga.getState().name());
        saga.incrementRetry();
        saga.setNextRetryTime(calcBackoff(saga.getRetryCount()));
        sagaRepo.save(saga);
        meter.counter("saga_step_failure", "step", saga.getState().name()).increment();
        // 异步触发补偿任务
        compensationExecutor.submit(() -> compensate(saga));
    }
}

@Component
@RequiredArgsConstructor
public class SagaCompensationExecutor {
    private final MeetingBookSagaRepository sagaRepo;
    private final ResourceService resourceService;
    private final MeetingRepository meetingRepo;
    private final SignalingGateway signalingGateway;

    @Transactional
    public void compensate(MeetingBookSaga saga) {
        try {
            // 逆序补偿，幂等执行
            if (saga.getState().ordinal() >= SagaState.INVITATION_SENT.ordinal()) {
                signalingGateway.broadcastCancel(saga.getMeetingId()); // 幂等
            }
            if (saga.getState().ordinal() >= SagaState.RESOURCE_LOCKED.ordinal()) {
                resourceService.unlock(saga.getLockToken()); // 幂等
            }
            if (saga.getState().ordinal() >= SagaState.RESOURCE_LOCKED.ordinal()) {
                meetingRepo.deleteById(saga.getMeetingId());
            }
            saga.setState(SagaState.COMPENSATED);
        } catch (Exception e) {
            if (saga.getRetryCount() >= MAX_RETRY) {
                saga.setState(SagaState.FAILED);
                alertService.sendAlert("Saga补偿彻底失败", saga);
            } else {
                saga.setNextRetryTime(calcBackoff(saga.getRetryCount()));
            }
        } finally {
            sagaRepo.save(saga);
        }
    }

    private LocalDateTime calcBackoff(int retryCount) {
        // 指数退避 + 抖动：2^N * 1min + random(0-30s)
        return LocalDateTime.now().plusMinutes((long) Math.pow(2, retryCount))
                            .plusSeconds(RandomUtils.nextInt(0, 30));
    }
}

@Scheduled(cron = "0 */5 * * * ?") // 每 5 分钟扫描
public void scanStuckSagas() {
    List<MeetingBookSaga> stuck = sagaRepo.findStuckSagas(
        SagaState.COMPENSATING, 
        Duration.ofMinutes(10), // 超过 10 分钟未推进
        PageRequest.of(0, 100)
    );
    stuck.forEach(compensationExecutor::compensate);
}

graph LR
    A[请求入口] --> B{本地 Caffeine 缓存<br/>命中?}
    B -- 是 --> C[直接返回 409 Conflict]
    B -- 否 --> D[Redis SET NX EX<br/>Lua 原子脚本]
    D -- 成功 --> E[写入本地缓存<br/>TTL=Redis TTL * 0.8]
    D -- 失败 --> C
    E --> F[异步写入 MySQL<br/>幂等流水表]

@Configuration
public class LocalIdempotentCacheConfig {
    @Bean
    public Cache<String, IdempotentResult> idempotentLocalCache() {
        return Caffeine.newBuilder()
            .maximumSize(100_000)                    // 约占 200MB 堆内存
            .expireAfterWrite(Duration.ofMinutes(4)) // 短于 Redis TTL(5min)，防脏读
            .recordStats()
            .build();
    }
}

// 切面增强版
@Around("@annotation(idempotent)")
public Object handleWithLocalCache(ProceedingJoinPoint pjp, Idempotent idempotent) {
    String key = resolver.resolveKey(pjp.getArgs(), idempotent);
    
    // 1. 本地极速判重（纳秒级）
    IdempotentResult cached = localCache.getIfPresent(key);
    if (cached != null) {
        metrics.increment("idempotent.local.hit");
        throw new IdempotentRejectedException(cached.getBizId());
    }

    // 2. Redis 分布式判重（毫秒级）
    boolean acquired = redisScript.execute(KEYS, key, ttlSeconds);
    if (!acquired) {
        // 回种本地缓存，防止穿透
        localCache.put(key, IdempotentResult.rejected(bizId));
        metrics.increment("idempotent.redis.conflict");
        throw new IdempotentRejectedException(bizId);
    }

    // 3. 执行业务
    try {
        Object result = pjp.proceed();
        // 成功才回种“成功”标记，失败不缓存允许重试
        localCache.put(key, IdempotentResult.success(bizId));
        asyncPersistIdempotentLog(key, bizId, "SUCCESS");
        return result;
    } catch (Exception e) {
        // 业务失败：删除 Redis Key 释放锁，本地不缓存失败态
        redis.del(key);
        throw e;
    }
}

-- 按 biz_type + 日期分表，单表 < 5000 万行
CREATE TABLE idempotent_log_202412 (
    id BIGINT AUTO_INCREMENT,
    biz_type VARCHAR(32) NOT NULL,
    biz_id VARCHAR(64) NOT NULL,
    request_id VARCHAR(64) NOT NULL, -- 网关 X-Request-Id
    idempotent_key VARCHAR(128) NOT NULL,
    status TINYINT NOT NULL, -- 1=SUCCESS 2=CONFLICT 3=FAILED
    request_body JSON,       -- 脱敏后的请求体，供审计
    response_code VARCHAR(16),
    created_at DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3),
    PRIMARY KEY (id, created_at),
    UNIQUE KEY uk_idem_key (idempotent_key), -- 全局唯一约束兜底
    KEY idx_biz_id (biz_id)
) ENGINE=InnoDB PARTITION BY RANGE (TO_DAYS(created_at)) (
    PARTITION p20241201 VALUES LESS THAN (TO_DAYS('2024-12-02')),
    ...
);

graph TB
    subgraph Region A
        GW_A[网关] --> IS_A[信令服务]
    end
    subgraph Region B
        GW_B[网关] --> IS_B[信令服务]
    end
    IS_A --> GKS[(全局幂等服务<br/>基于 OceanBase/Spanner)]
    IS_B --> GKS
    GKS -->|强一致写入| OB[(OceanBase<br/>Primary Zone 就近读写)]

// 统一状态存储接口，屏蔽底层多活数据库差异
public interface SagaStateRepository {
    // 乐观锁 CAS 更新状态
    boolean compareAndSetState(String sagaId, SagaState expected, SagaState next, String payload);
    
    // 查询当前状态用于幂等补偿
    SagaSnapshot getSnapshot(String sagaId);
}

// OceanBase 实现：利用 ROW_VERSION 隐藏列实现无锁 CAS
@Repository
public class OceanBaseSagaRepo implements SagaStateRepository {
    @Override
    public boolean compareAndSetState(String sagaId, SagaState expected, SagaState next, String payload) {
        int updated = jdbcTemplate.update(
            "UPDATE saga_meeting_book SET state=?, payload=?, version=version+1, gmt_modified=NOW(3) " +
            "WHERE saga_id=? AND state=? AND version=?",
            next.name(), payload, sagaId, expected.name(), currentVersion
        );
        return updated == 1;
    }
}

-- 对比两 Region 会议单最终状态一致性
SELECT a.meeting_id, a.status AS status_A, b.status AS status_B
FROM region_a.meeting_main a
JOIN region_b.meeting_main b ON a.meeting_id = b.meeting_id
WHERE a.status != b.status 
  AND a.gmt_modified > DATE_SUB(NOW(), INTERVAL 1 DAY);

@Aspect
@Component
@RequiredArgsConstructor
public class AuditLogAspect {
    private final AuditLogProducer auditProducer;
    private final DataMaskingService masking;

    @AfterReturning(pointcut = "@annotation(audit)", returning = "result")
    public void logSuccess(JoinPoint jp, AuditLog audit, Object result) {
        AuditEvent event = buildEvent(jp, audit, result, "SUCCESS", null);
        auditProducer.sendAsync(event); // 异步非阻塞
    }

    @AfterThrowing(pointcut = "@annotation(audit)", throwing = "ex")
    public void logFailure(JoinPoint jp, AuditLog audit, Exception ex) {
        AuditEvent event = buildEvent(jp, audit, null, "FAILED", ex.getMessage());
        auditProducer.sendAsync(event);
    }

    private AuditEvent buildEvent(JoinPoint jp, AuditLog audit, Object result, String status, String error) {
        MethodSignature sig = (MethodSignature) jp.getSignature();
        Object[] args = jp.getArgs();
        
        // 1. 参数脱敏
        Map<String, Object> maskedArgs = masking.maskMethodArgs(sig.getParameterNames(), args, audit.maskRules());
        
        // 2. 结果脱敏
        Object maskedResult = masking.maskResult(result, audit.resultMaskRule());
        
        return AuditEvent.builder()
            .traceId(MDC.get("traceId"))
            .spanId(MDC.get("spanId"))
            .operatorId(masking.pseudonymize((String) args[0])) // 假设第一参数为 operatorId
            .action(audit.action())
            .bizType(audit.bizType())
            .bizId(extractBizId(args, audit.bizIdExpr()))
            .requestParams(JsonUtils.toJson(maskedArgs))
            .responseResult(JsonUtils.toJson(maskedResult))
            .status(status)
            .errorMsg(error)
            .timestamp(System.currentTimeMillis())
            .build();
    }
}

@AuditLog(
    action = "MEETING_BOOK",
    bizType = "MEETING",
    bizIdExpr = "#cmd.meetingId",
    maskRules = {
        @MaskRule(field = "cmd.title", strategy = MaskStrategy.HASH_LENGTH),
        @MaskRule(field = "cmd.attendees[*].userId", strategy = MaskStrategy.PSEUDONYM),
        @MaskRule(field = "cmd.attendees[*].phone", strategy = MaskStrategy.MASK_PHONE)
    },
    resultMaskRule = @MaskRule(field = "meetingId", strategy = MaskStrategy.PSEUDONYM)
)
public MeetingBookResult book(MeetingBookCommand cmd) { ... }

// contracts/resource-service/unlock-contract.groovy
Contract.make {
    name "补偿释放会议室锁-幂等成功"
    request {
        method 'POST'
        url '/api/v1/resources/locks/release'
        headers {
            contentType(applicationJson())
            header 'X-Request-Id', 'uuid-regex'
        }
        body([
            lockToken: 'lock_abc123',
            expectedVersion: 5
        ])
    }
    response {
        status 200
        body([
            code: 'SUCCESS',
            data: true,
            traceId: 'uuid-regex'
        ])
    }
}

@SpringBootTest
@AutoConfigureStubRunner(ids = "com.company:resource-service:+:stubs:8080")
class ResourceClientContractTest {
    @Autowired ResourceClient resourceClient;

    @Test
    void unlock_idempotent_success() {
        // 契约保证：只要 Provider 通过验证，Consumer 就能正确处理幂等响应
        assertTrue(resourceClient.unlock("lock_abc123", 5));
        // 二次调用验证幂等
        assertTrue(resourceClient.unlock("lock_abc123", 5));
    }
}

# chaos/meeting-book-signaling-loss.yaml
apiVersion: chaosblade.io/v1alpha1
kind: ChaosBlade
metadata:
  name: meeting-signaling-packet-loss
spec:
  experiment:
    scope: deployment
    target: signaling-gateway
    action: loss
    mode: random
    parameters:
      percent: "30"
      interface: "eth0"
      timeout: "300" # 5分钟
  # 验证探针：自动化断言
  probes:
    - name: saga-compensated-rate
      type: prometheus
      interval: 30s
      query: |
        sum(rate(saga_state_transition_total{from="COMPENSATING",to="COMPENSATED"}[2m])) 
        / 
        sum(rate(saga_state_transition_total{from="INVITATION_SENT",to="COMPENSATING"}[2m]))
      criteria: "== 1.0" # 100% 补偿成功
      failureMessage: "补偿成功率低于 100%，存在信令丢包导致的僵尸事务"

# chaos/db-failover-saga-recovery.yaml
spec:
  experiment:
    action: pod-kill
    target: mysql-primary-pod
    parameters:
      gracePeriod: 0
  probes:
    - name: saga-stuck-detection
      type: http
      interval: 10s
      url: "http://admin-api/sagas/stuck?thresholdMinutes=2"
      criteria: "== 0" # 无卡顿事务
      failureMessage: "DB 切换后存在超过 2 分钟未推进的 Saga"