// 示例：业务幂等键生成策略
String idempotentKey = String.format("meeting:book:%s:%s:%s", 
    tenantId,           // 租户隔离
    organizerUserId,    // 发起人
    clientRequestId     // 客户端生成的 UUID (必传)
);

// 伪代码：Gateway Filter / Interceptor
public boolean tryAcquireIdempotentLock(String key, long ttlSeconds) {
    // Lua 脚本保证原子性：不存在则设置并返回 1，存在则返回 0
    String script = "if redis.call('setnx', KEYS[1], ARGV[1]) == 1 then redis.call('expire', KEYS[1], ARGV[2]) return 1 else return 0 end";
    return redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), 
        Collections.singletonList(key), traceId, ttlSeconds) == 1L;
}

-- 核心表结构设计
CREATE TABLE `meeting_info` (
  `id` BIGINT NOT NULL AUTO_INCREMENT,
  `tenant_id` BIGINT NOT NULL,
  `organizer_id` BIGINT NOT NULL,
  `client_request_id` VARCHAR(64) NOT NULL COMMENT '客户端幂等键',
  `meeting_status` TINYINT NOT NULL DEFAULT 0 COMMENT '0:草稿 1:已预约 2:进行中 3:已结束 4:已取消',
  `version` INT NOT NULL DEFAULT 0 COMMENT '乐观锁版本号',
  PRIMARY KEY (`id`),
  -- 核心幂等唯一索引
  UNIQUE KEY `uk_tenant_organizer_client` (`tenant_id`, `organizer_id`, `client_request_id`)
) ENGINE=InnoDB;

// 状态流转合法性校验
public enum MeetingStatus {
    DRAFT(0), BOOKED(1), IN_PROGRESS(2), ENDED(3), CANCELLED(4);

    private static final Map<Integer, Set<Integer>> TRANSITIONS = Map.of(
        DRAFT.code, Set.of(BOOKED.code, CANCELLED.code),
        BOOKED.code, Set.of(IN_PROGRESS.code, CANCELLED.code),
        IN_PROGRESS.code, Set.of(ENDED.code, CANCELLED.code)
    );

    public boolean canTransitionTo(int targetCode) {
        return TRANSITIONS.getOrDefault(this.code, Collections.emptySet()).contains(targetCode);
    }
}

// Service 层应用
@Transactional
public Meeting bookMeeting(BookCmd cmd) {
    Meeting meeting = meetingRepo.findByIdempotentKey(cmd.getKey());
    if (meeting != null) {
        // 幂等返回：仅允许从 DRAFT 推进到 BOOKED
        if (meeting.getStatus() == MeetingStatus.DRAFT && cmd.getAction() == Action.BOOK) {
            return transitionToBooked(meeting);
        }
        // 非法重复/状态冲突，抛出业务异常或直接返回现状
        return meeting; 
    }
    // 首次创建：插入 DRAFT 状态，再推进
    return createAndBook(cmd);
}

@Entity
@Table(name = "saga_transaction_log")
public class SagaTransactionLog {
    @Id String sagaId;           // 业务主键，如 meetingId
    String currentStep;          // 当前执行步骤
    String status;               // RUNNING, COMPLETED, COMPENSATING, COMPENSATED, FAILED
    @Lob String payload;         // 完整上下文 JSON (参会人、会议室ID、TraceID等)
    @Lob String compensationPayload; // 补偿所需上下文
    LocalDateTime createTime;
    LocalDateTime updateTime;
    int retryCount;              // 补偿重试次数
}

@Component
public class MeetingBookingSagaOrchestrator {

    private static final List<SagaStep> STEPS = List.of(
        new SagaStep("CREATE_MEETING", meetingService::createDraft, meetingService::cancelMeeting),
        new SagaStep("LOCK_RESOURCE", resourceService::lockRoom, resourceService::releaseRoom),
        new SagaStep("SEND_NOTIFICATION", notificationService::sendInvite, notificationService::sendCancel),
        new SagaStep("PUSH_SIGNAL", signalingService::pushCreate, signalingService::pushCancel)
    );

    @Transactional
    public void execute(BookingContext ctx) {
        SagaTransactionLog log = new SagaTransactionLog();
        log.setSagaId(ctx.getMeetingId());
        log.setStatus("RUNNING");
        logRepo.save(log);

        for (int i = 0; i < STEPS.size(); i++) {
            SagaStep step = STEPS.get(i);
            try {
                log.setCurrentStep(step.name);
                step.execute(ctx); // 执行正向操作
                logRepo.save(log); // 持久化进度
            } catch (Exception e) {
                // 触发补偿：逆序执行已成功步骤的补偿动作
                triggerCompensation(log, i - 1, ctx);
                throw new SagaExecutionException("Step " + step.name + " failed", e);
            }
        }
        log.setStatus("COMPLETED");
        logRepo.save(log);
    }

    private void triggerCompensation(SagaTransactionLog log, int failedStepIndex, BookingContext ctx) {
        log.setStatus("COMPENSATING");
        logRepo.save(log);

        for (int i = failedStepIndex; i >= 0; i--) {
            SagaStep step = STEPS.get(i);
            // 补偿动作内部必须实现幂等（如：释放资源先查状态，已释放则跳过）
            retryTemplate.execute(context -> {
                step.compensate(ctx);
                return null;
            }, recoveryCallback -> {
                // 重试耗尽：告警 + 写入死信表 + 人工介入工单
                alertService.alert("Saga补偿失败需人工介入", log.getSagaId(), step.name);
                deadLetterRepo.save(new DeadLetter(log.getSagaId(), step.name, ctx));
            });
        }
        log.setStatus("COMPENSATED");
        logRepo.save(log);
    }
}

// 优化前：直接 INSERT IGNORE，依赖 DB 锁冲突
// 优化后：应用层 CAS 预检 + Redis 乐观锁分流

@Transactional(rollbackFor = Exception.class)
public Meeting bookMeeting(BookCmd cmd) {
    // 1. 极轻量 Redis 乐观锁预检（非阻塞，失败快返回）
    String optimisticLockKey = "meeting:lock:" + cmd.getMeetingRoomId() + ":" + cmd.getTimeSlotHash();
    // Lua: 校验版本号 + 原子自增，仅允许单线程通过预检
    long version = redisTemplate.execute(CHECK_AND_INCR_SCRIPT, List.of(optimisticLockKey), cmd.getExpectedVersion());
    if (version == -1L) {
        throw new ResourceConflictException("时间槽版本已变更，请刷新重试"); // 客户端走乐观锁重试流程
    }

    // 2. 核心业务入库（此时冲突概率已大幅降低）
    Meeting meeting = meetingRepo.insertDraft(cmd);
    
    // 3. 异步刷新 Redis 版本号（最终一致性）
    redisTemplate.opsForValue().set(optimisticLockKey, meeting.getVersion());
    return meeting;
}

// 补偿阶段：构建 DAG 并行执行
public void compensateParallel(SagaTransactionLog log, BookingContext ctx) {
    // 1. 解析已成功步骤依赖图 (Step 2 依赖 Step 1, Step 3/4 依赖 Step 2)
    DagGraph<String, SagaStep> dag = buildCompensationDag(log.getCompletedSteps());
    
    // 2. 拓扑层级并行执行
    List<List<SagaStep>> levels = dag.topologicalLevels();
    for (List<SagaStep> level : levels) {
        // 同层无依赖，并行补偿
        List<CompletableFuture<Void>> futures = level.stream()
            .map(step -> CompletableFuture.runAsync(() -> safeCompensate(step, ctx), compensationExecutor))
            .collect(Collectors.toList());
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    }
}

// TypeScript SDK 伪代码
enum RequestState { IDLE, SENDING, AWAITING_ACK, COMPLETED, FAILED_NEED_RETRY }

class MeetingBookingController {
    private requestMap = new Map<string, RequestContext>(); // key: clientRequestId

    async bookMeeting(params: BookParams): Promise<MeetingResult> {
        const requestId = uuidv4(); // 客户端强制生成
        const ctx = { requestId, params, state: RequestState.SENDING, retryCount: 0 };
        this.requestMap.set(requestId, ctx);

        return this.executeWithRetry(ctx);
    }

    private async executeWithRetry(ctx: RequestContext): Promise<MeetingResult> {
        while (ctx.retryCount < MAX_RETRY) {
            try {
                ctx.state = RequestState.AWAITING_ACK;
                const resp = await this.apiClient.post('/meeting/book', {
                    ...ctx.params,
                    clientRequestId: ctx.requestId // 透传幂等键
                }, { timeout: 10000 });

                // 服务端返回幂等标识
                if (resp.headers['x-idempotent-replay'] === 'true') {
                    logger.info(`[幂等命中] RequestId: ${ctx.requestId}`);
                }
                ctx.state = RequestState.COMPLETED;
                return resp.data;
            } catch (err) {
                if (this.isRetriableError(err)) {
                    ctx.retryCount++;
                    await this.sleep(backoff(ctx.retryCount)); // 指数退避 + 抖动
                    continue;
                }
                throw err; // 非重试错误（如参数校验失败）直接抛出
            }
        }
        ctx.state = RequestState.FAILED_NEED_RETRY;
        throw new MaxRetryExceededError();
    }
}

# recon_task_config 内容示例 (YAML)
biz_type: "MEETING_BOOK"
schedule: "0 0/30 * * * ?" # 每半小时
source_system: "meeting_service" # 主系统
target_systems: 
  - "resource_service" # 会议室锁
  - "signaling_service" # 信令状态
compare_rules:
  - name: "会议-资源锁一致性"
    source_query: "SELECT id, room_id, status, version FROM meeting_info WHERE status IN (1,2) AND update_time > ?"
    target_query: "SELECT meeting_id, room_id, lock_status, version FROM resource_lock WHERE lock_status='LOCKED' AND update_time > ?"
    join_keys: ["id=meeting_id"]
    compare_fields: 
      - "room_id"
      - "status=lock_status" # 映射转换: 1(BOOKED) == LOCKED
    tolerance: 
      type: "TIME_WINDOW"
      window_seconds: 60 # 允许 60s 内最终一致延迟
  - name: "会议-信令状态一致性"
    source_query: "SELECT id, status FROM meeting_info WHERE status IN (1,2,3)"
    target_query: "SELECT meeting_id, signal_status FROM signaling_state WHERE signal_status IN ('CREATED','STARTED','ENDED')"
    join_keys: ["id=meeting_id"]
    compare_fields: ["status=signal_status"]
repair_strategy:
  - mismatch_type: "SOURCE_HAS_TARGET_MISSING" # 会议存在但无资源锁
    action: "TRIGGER_COMPENSATION" # 发送 MQ 触发补偿流程 (尝试补锁或取消会议)
    params: { saga_type: "REPAIR_LOCK" }
  - mismatch_type: "TARGET_HAS_SOURCE_MISSING" # 孤儿资源锁
    action: "AUTO_RELEASE" # 直接调用资源服务释放接口
    params: { force: true }

// 切面自动记录审计日志
@Aspect
@Component
@RequiredArgsConstructor
public class AuditLogAspect {
    private final AuditLogService auditLogService;

    @Around("@annotation(AuditLog)")
    public Object audit(ProceedingJoinPoint pjp) throws Throwable {
        AuditLog annotation = getAnnotation(pjp);
        String traceId = Tracer.currentSpan().traceIdString();
        String operatorId = SecurityContext.getUserId();
        
        // 执行前快照（针对修改类操作）
        Object beforeSnapshot = annotation.snapshotBefore() ? captureSnapshot(pjp) : null;
        
        long start = System.currentTimeMillis();
        Object result = pjp.proceed();
        long cost = System.currentTimeMillis() - start;
        
        // 执行后快照
        Object afterSnapshot = annotation.snapshotAfter() ? captureSnapshot(pjp) : result;
        
        // 异步落库（不阻塞主流程），写入 Kafka -> ClickHouse / ES
        auditLogService.asyncRecord(AuditLogDTO.builder()
            .traceId(traceId)
            .operatorId(operatorId)
            .action(annotation.value())
            .resourceType(annotation.resourceType())
            .resourceId(extractResourceId(pjp, result))
            .beforeState(serialize(beforeSnapshot))
            .afterState(serialize(afterSnapshot))
            .ipAddress(RequestContext.getClientIp())
            .result(ResultStatus.SUCCESS)
            .costMs(cost)
            .build());
        return result;
    }
}

# ADR-001: 会议预约分布式事务选型 Saga 而非 TCC/Seata AT

## 状态: Accepted
## 背景
会议预约涉及 4+ 下游服务，资源锁定需强一致，通知/信令允许最终一致。团队无 Seata AT 模式生产经验，TCC 开发成本高（需三接口）。

## 决策
采用 **编排式 Saga**，核心链路同步调用，非核心链路异步事务消息。
补偿动作由业务开发实现，框架仅提供调度、重试、状态持久化。

## 后果
- 正面：开发效率提升 40%，无侵入业务代码，易于理解维护。
- 负面：补偿逻辑分散在各业务 Service，需强制 Code Review 保证幂等性。
- 规避：建立《补偿接口开发规范》检查单，接入 SonarQube 规则扫描。