CREATE TABLE `undo_log` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(100) NOT NULL,
  `context` varchar(128) NOT NULL,
  `rollback_info` longblob NOT NULL,
  `log_status` int(11) NOT NULL,
  `log_created` datetime NOT NULL,
  `log_modified` datetime NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

@Service
public class OrderServiceImpl implements OrderService {
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private AccountService accountService;
    
    @Autowired
    private StorageService storageService;
    
    /**
     * 全局事务发起者，使用 @GlobalTransactional 注解
     */
    @Override
    @GlobalTransactional(name = "create-order", rollbackFor = Exception.class)
    public void createOrder(Order order) {
        // 1. 创建订单
        orderMapper.insert(order);
        
        // 2. 扣减库存（远程调用）
        storageService.deduct(order.getProductId(), order.getCount());
        
        // 3. 扣减账户余额（远程调用）
        accountService.debit(order.getUserId(), order.getMoney());
        
        // 4. 更新订单状态
        order.setStatus(1);
        orderMapper.update(order);
    }
}

// 库存服务
@Service
public class StorageServiceImpl implements StorageService {
    
    @Autowired
    private StorageMapper storageMapper;
    
    @Override
    public void deduct(Long productId, Integer count) {
        // 直接执行业务逻辑，无需额外事务注解
        storageMapper.deduct(productId, count);
    }
}

// DataSourceProxy 代理数据源
public class DataSourceProxy extends AbstractDataSourceProxy {
    
    @Override
    public ConnectionProxy getConnection() throws SQLException {
        Connection targetConnection = targetDataSource.getConnection();
        return new ConnectionProxy(this, targetConnection);
    }
}

// ConnectionProxy 核心逻辑
public class ConnectionProxy extends AbstractConnectionProxy {
    
    @Override
    public void commit() throws SQLException {
        try {
            // 注册分支事务
            register();
        } catch (TransactionException e) {
            // 识别并处理异常
            recognizeLockKeyConflictException(e, context.buildLockKeys());
        }
        
        try {
            // 执行本地事务提交
            targetConnection.commit();
        } catch (Throwable ex) {
            // 上报事务执行失败
            report(false);
            throw ex;
        }
        
        // 上报事务执行成功
        report(true);
    }
}

// ExecuteTemplate 执行模板
public class ExecuteTemplate {
    
    public static <T, S extends Statement> T execute(
            SQLRecognizer sqlRecognizer,
            StatementProxy<S> statementProxy,
            StatementCallback<T, S> statementCallback,
            Object... args) throws SQLException {
        
        // 1. 前置镜像：查询修改前的数据
        TableRecords beforeImage = buildBeforeImage(statementProxy, sqlRecognizer);
        
        // 2. 执行业务 SQL
        T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
        
        // 3. 后置镜像：查询修改后的数据
        TableRecords afterImage = buildAfterImage(statementProxy, sqlRecognizer);
        
        // 4. 构造 undo_log
        prepareUndoLog(beforeImage, afterImage);
        
        return result;
    }
}

// 前置镜像和后置镜像数据结构
public class TableRecords {
    private TableMeta tableMeta;
    private List<Row> rows;
    
    // 行数据
    public static class Row {
        private List<Field> fields;
        
        public static class Field {
            private String name;        // 字段名
            private int keyType;        // 主键类型
            private Object value;       // 字段值
        }
    }
}

// Undo Log 结构
public class BranchUndoLog {
    private String xid;                    // 全局事务ID
    private long branchId;                 // 分支事务ID
    private List<SQLUndoLog> sqlUndoLogs;  // SQL回滚日志
    
    public static class SQLUndoLog {
        private String sqlType;            // INSERT/UPDATE/DELETE
        private String tableName;          // 表名
        private TableRecords beforeImage;  // 前置镜像
        private TableRecords afterImage;   // 后置镜像
    }
}

public class AsyncWorker implements ResourceManagerInbound {
    
    /**
     * 异步提交分支事务
     */
    public BranchStatus branchCommit(String xid, long branchId, 
                                     String resourceId) {
        // AT 模式下，一阶段已经提交，二阶段只需删除 undo_log
        return asyncCommit(xid, branchId, resourceId);
    }
    
    private BranchStatus asyncCommit(String xid, long branchId, 
                                     String resourceId) {
        // 加入异步删除队列
        addToCommitQueue(xid, branchId, resourceId);
        return BranchStatus.PhaseTwo_Committed;
    }
    
    // 异步删除 undo_log
    private void deleteUndoLog(String xid, long branchId) {
        String sql = "DELETE FROM undo_log WHERE xid = ? AND branch_id = ?";
        executeUpdate(sql, xid, branchId);
    }
}

public class UndoLogManager {
    
    /**
     * 回滚分支事务
     */
    public void undo(DataSourceProxy dataSourceProxy, String xid, 
                     long branchId) throws SQLException {
        
        Connection conn = dataSourceProxy.getConnection();
        
        try {
            // 1. 查询 undo_log
            String selectSQL = "SELECT * FROM undo_log WHERE xid = ? " +
                             "AND branch_id = ? FOR UPDATE";
            BranchUndoLog branchUndoLog = selectUndoLog(conn, xid, branchId);
            
            if (branchUndoLog == null) {
                return;  // 已经回滚或提交
            }
            
            // 2. 数据校验
            if (!dataValidation(conn, branchUndoLog)) {
                throw new SQLException("Data validation failed");
            }
            
            // 3. 生成反向 SQL 并执行
            for (SQLUndoLog sqlUndoLog : branchUndoLog.getSqlUndoLogs()) {
                AbstractUndoExecutor undoExecutor = 
                    UndoExecutorFactory.getUndoExecutor(
                        dataSourceProxy.getDbType(), sqlUndoLog);
                undoExecutor.executeOn(conn);
            }
            
            // 4. 删除 undo_log
            String deleteSQL = "DELETE FROM undo_log WHERE xid = ? " +
                             "AND branch_id = ?";
            executeUpdate(conn, deleteSQL, xid, branchId);
            
            conn.commit();
            
        } catch (Exception e) {
            conn.rollback();
            throw e;
        }
    }
    
    /**
     * 数据校验：比对当前数据和后置镜像
     */
    private boolean dataValidation(Connection conn, 
                                   BranchUndoLog branchUndoLog) {
        for (SQLUndoLog sqlUndoLog : branchUndoLog.getSqlUndoLogs()) {
            TableRecords afterImage = sqlUndoLog.getAfterImage();
            TableRecords currentRecords = queryCurrentRecords(conn, afterImage);
            
            // 比对数据是否一致
            if (!afterImage.equals(currentRecords)) {
                return false;  // 数据被脏写
            }
        }
        return true;
    }
}

// UPDATE 反向 SQL
public class MySQLUndoUpdateExecutor extends AbstractUndoExecutor {
    
    @Override
    protected String buildUndoSQL() {
        TableRecords beforeImage = sqlUndoLog.getBeforeImage();
        
        // 根据前置镜像生成 UPDATE 语句
        StringBuilder sql = new StringBuilder("UPDATE ");
        sql.append(sqlUndoLog.getTableName()).append(" SET ");
        
        // 设置字段值为前置镜像的值
        List<Field> fields = beforeImage.getRows().get(0).getFields();
        for (int i = 0; i < fields.size(); i++) {
            if (i > 0) sql.append(", ");
            sql.append(fields.get(i).getName()).append(" = ?");
        }
        
        // WHERE 条件（主键）
        sql.append(" WHERE ");
        appendWhereCondition(sql, beforeImage);
        
        return sql.toString();
    }
}

// DELETE 反向 SQL（生成 INSERT）
public class MySQLUndoDeleteExecutor extends AbstractUndoExecutor {
    
    @Override
    protected String buildUndoSQL() {
        TableRecords beforeImage = sqlUndoLog.getBeforeImage();
        
        // 根据前置镜像生成 INSERT 语句
        StringBuilder sql = new StringBuilder("INSERT INTO ");
        sql.append(sqlUndoLog.getTableName()).append(" (");
        
        // 字段列表
        List<Field> fields = beforeImage.getRows().get(0).getFields();
        for (int i = 0; i < fields.size(); i++) {
            if (i > 0) sql.append(", ");
            sql.append(fields.get(i).getName());
        }
        
        sql.append(") VALUES (");
        for (int i = 0; i < fields.size(); i++) {
            if (i > 0) sql.append(", ");
            sql.append("?");
        }
        sql.append(")");
        
        return sql.toString();
    }
}

// INSERT 反向 SQL（生成 DELETE）
public class MySQLUndoInsertExecutor extends AbstractUndoExecutor {
    
    @Override
    protected String buildUndoSQL() {
        TableRecords afterImage = sqlUndoLog.getAfterImage();
        
        // 根据后置镜像生成 DELETE 语句
        StringBuilder sql = new StringBuilder("DELETE FROM ");
        sql.append(sqlUndoLog.getTableName());
        sql.append(" WHERE ");
        
        appendWhereCondition(sql, afterImage);
        
        return sql.toString();
    }
}

public class LockManagerImpl implements LockManager {
    
    /**
     * 获取全局锁
     */
    public boolean acquireLock(List<RowLock> rowLocks) {
        // 锁的粒度：表名 + 主键值
        // 例如：order_table:1,2,3
        String lockKey = buildLockKey(rowLocks);
        
        // 向 TC 申请全局锁
        boolean result = transactionCoordinator.acquireLock(
            xid, branchId, resourceId, lockKey);
        
        if (!result) {
            // 获取锁失败，进入重试逻辑
            return retryAcquireLock(rowLocks);
        }
        
        return true;
    }
    
    private String buildLockKey(List<RowLock> rowLocks) {
        // 格式：table1:pk1,pk2;table2:pk3,pk4
        Map<String, Set<String>> lockMap = new HashMap<>();
        
        for (RowLock rowLock : rowLocks) {
            lockMap.computeIfAbsent(
                rowLock.getTableName(), 
                k -> new HashSet<>()
            ).add(rowLock.getPk());
        }
        
        return lockMap.entrySet().stream()
            .map(e -> e.getKey() + ":" + String.join(",", e.getValue()))
            .collect(Collectors.joining(";"));
    }
}

seata:
  client:
    rm:
      # 异步提交优化
      async-commit-buffer-limit: 10000
      
    undo:
      # 只记录更新字段
      only-care-update-columns: true
      
      # 压缩 undo_log
      compress:
        enable: true
        type: zip
        threshold: 64k

-- 清理 7 天前的 undo_log
DELETE FROM undo_log 
WHERE log_created < DATE_SUB(NOW(), INTERVAL 7 DAY);