#2PC 两阶段提交

// 协调者实现（简化版）
public class Coordinator {
    private List<Participant> participants = new ArrayList<>();

    public boolean commit(String transactionId) {
        // 阶段一：发送 PREPARE
        List<ParticipantResponse> responses = new ArrayList<>();
        for (Participant participant : participants) {
            try {
                boolean ready = participant.prepare(transactionId);
                responses.add(new ParticipantResponse(participant, ready));
            } catch (Exception e) {
                responses.add(new ParticipantResponse(participant, false));
            }
        }

        // 检查所有参与者是否准备好
        boolean allReady = responses.stream()
            .allMatch(ParticipantResponse::isReady);

        if (allReady) {
            // 阶段二：发送 COMMIT
            for (Participant participant : participants) {
                try {
                    participant.commit(transactionId);
                } catch (Exception e) {
                    // COMMIT 失败，进入等待恢复状态
                    log.error("参与者提交失败，等待恢复", e);
                    waitForRecovery(participant, transactionId);
                }
            }
            return true;
        } else {
            // 任意一个失败，发送 ROLLBACK
            for (Participant participant : participants) {
                participant.rollback(transactionId);
            }
            return false;
        }
    }
}

// 参与者实现（简化版）
public class Participant {
    private String name;

    public boolean prepare(String transactionId) {
        try {
            // 1. 开启本地事务
            connection.setAutoCommit(false);

            // 2. 执行 SQL
            executeSQL();

            // 3. 预提交：锁定资源，但未真正提交
            // 此时数据库锁住，其他操作无法修改相关数据
            PreparedStatement ps = connection.prepareStatement(
                "PREPARE TO COMMIT '" + transactionId + "'"
            );
            ps.execute();

            // 4. 返回"准备好"
            return true;
        } catch (Exception e) {
            // 回滚本地事务
            connection.rollback();
            return false;
        }
    }

    public void commit(String transactionId) {
        // 正式提交
        connection.commit();
    }

    public void rollback(String transactionId) {
        // 回滚
        connection.rollback();
    }
}

// 协调者崩溃恢复
public class CoordinatorRecovery {
    // 协调者在发送 PREPARE 前，会将事务状态写入 WAL
    private WriteAheadLog wal;

    public void recover() {
        // 1. 读取 WAL 中未完成的事务
        List<TransactionRecord> inFlightTransactions = wal.getInFlightTransactions();

        for (TransactionRecord record : inFlightTransactions) {
            // 2. 查询每个参与者的状态
            TransactionState state = queryParticipantState(record.transactionId);

            if (state == TransactionState.COMMITTED) {
                // 如果参与者已提交，提交完成
                continue;
            } else if (state == TransactionState.ABORTED) {
                // 如果参与者已回滚，回滚完成
                continue;
            } else {
                // 参与者处于 READY 状态，需要重新发送决策
                // 问题：协调者不知道参与者是否真的收到了 COMMIT
                // 只能假设参与者没收到，重新发送
                resendDecision(record.transactionId);
            }
        }
    }
}

// ❌ 错误：没有处理协调者和参与者的超时
public class BadCoordinator {
    public void commit(String transactionId) {
        // 发送 PREPARE 后无限等待
        participant.prepare(transactionId);
        // 如果 participant 挂了，这里永远等待
    }
}

// ✅ 正确：处理超时
public class GoodCoordinator {
    public boolean commit(String transactionId, long timeoutMs) {
        Future<Boolean> future = executor.submit(() ->
            participant.prepare(transactionId));

        try {
            Boolean ready = future.get(timeoutMs, TimeUnit.MILLISECONDS);
            if (ready) {
                participant.commit(transactionId);
                return true;
            } else {
                participant.rollback(transactionId);
                return false;
            }
        } catch (TimeoutException e) {
            // 超时，视为失败
            participant.rollback(transactionId);
            return false;
        }
    }
}

// 参与者在 READY 状态等待协调者的决策
// 如果协调者崩溃，参与者永远等待
// 解决方案：参与者也需要有超时机制

public class Participant {
    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    public boolean prepare(String transactionId) {
        // 开启本地事务
        connection.setAutoCommit(false);
        executeSQL();

        // 注册协调者地址
        registerWithCoordinator(transactionId, coordinatorAddress);

        // 启动超时检测
        Future<?> timeoutTask = scheduler.schedule(() -> {
            // 如果超时时间内没有收到协调者指令
            // 向协调者查询事务状态
            TransactionDecision decision = queryCoordinator(transactionId);
            if (decision == null) {
                // 协调者无响应，回滚
                this.rollback(transactionId);
            }
        }, 30, TimeUnit.SECONDS);

        return true;
    }
}