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;
    
    /**
     * Global transaction initiator, using @GlobalTransactional annotation
     */
    @Override
    @GlobalTransactional(name = "create-order", rollbackFor = Exception.class)
    public void createOrder(Order order) {
        // 1. Create order
        orderMapper.insert(order);
        
        // 2. Deduct stock (remote call)
        storageService.deduct(order.getProductId(), order.getCount());
        
        // 3. Deduct account balance (remote call)
        accountService.debit(order.getUserId(), order.getMoney());
        
        // 4. Update order status
        order.setStatus(1);
        orderMapper.update(order);
    }
}

// Stock service
@Service
public class StorageServiceImpl implements StorageService {
    
    @Autowired
    private StorageMapper storageMapper;
    
    @Override
    public void deduct(Long productId, Integer count) {
        // Execute business logic directly, no extra transaction annotation needed
        storageMapper.deduct(productId, count);
    }
}

TC (Transaction Coordinator) - Transaction coordinator
├── Global transaction management
├── Branch transaction management
└── Global lock management

TM (Transaction Manager) - Transaction manager
├── Global transaction start
├── Global transaction commit
└── Global transaction rollback

RM (Resource Manager) - Resource manager
├── Branch transaction registration
├── Branch transaction reporting
└── Branch transaction commit/rollback

// DataSourceProxy proxy data source
public class DataSourceProxy extends AbstractDataSourceProxy {
    
    @Override
    public ConnectionProxy getConnection() throws SQLException {
        Connection targetConnection = targetDataSource.getConnection();
        return new ConnectionProxy(this, targetConnection);
    }
}

// ConnectionProxy core logic
public class ConnectionProxy extends AbstractConnectionProxy {
    
    @Override
    public void commit() throws SQLException {
        try {
            // Register branch transaction
            register();
        } catch (TransactionException e) {
            // Identify and handle exceptions
            recognizeLockKeyConflictException(e, context.buildLockKeys());
        }
        
        try {
            // Execute local transaction commit
            targetConnection.commit();
        } catch (Throwable ex) {
            // Report transaction execution failure
            report(false);
            throw ex;
        }
        
        // Report transaction execution success
        report(true);
    }
}

// ExecuteTemplate execution template
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. Before image: query data before modification
        TableRecords beforeImage = buildBeforeImage(statementProxy, sqlRecognizer);
        
        // 2. Execute business SQL
        T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
        
        // 3. After image: query data after modification
        TableRecords afterImage = buildAfterImage(statementProxy, sqlRecognizer);
        
        // 4. Construct undo_log
        prepareUndoLog(beforeImage, afterImage);
        
        return result;
    }
}

// Before image and after image data structure
public class TableRecords {
    private TableMeta tableMeta;
    private List<Row> rows;
    
    // Row data
    public static class Row {
        private List<Field> fields;
        
        public static class Field {
            private String name;        // Field name
            private int keyType;        // Primary key type
            private Object value;       // Field value
        }
    }
}

// Undo Log structure
public class BranchUndoLog {
    private String xid;                    // Global transaction ID
    private long branchId;                 // Branch transaction ID
    private List<SQLUndoLog> sqlUndoLogs;  // SQL rollback logs
    
    public static class SQLUndoLog {
        private String sqlType;            // INSERT/UPDATE/DELETE
        private String tableName;          // Table name
        private TableRecords beforeImage;  // Before image
        private TableRecords afterImage;   // After image
    }
}

public class AsyncWorker implements ResourceManagerInbound {
    
    /**
     * Asynchronously commit branch transaction
     */
    public BranchStatus branchCommit(String xid, long branchId, 
                                     String resourceId) {
        // In AT mode, phase one has committed, phase two only needs to delete undo_log
        return asyncCommit(xid, branchId, resourceId);
    }
    
    private BranchStatus asyncCommit(String xid, long branchId, 
                                     String resourceId) {
        // Add to async deletion queue
        addToCommitQueue(xid, branchId, resourceId);
        return BranchStatus.PhaseTwo_Committed;
    }
    
    // Asynchronously delete 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 {
    
    /**
     * Rollback branch transaction
     */
    public void undo(DataSourceProxy dataSourceProxy, String xid, 
                     long branchId) throws SQLException {
        
        Connection conn = dataSourceProxy.getConnection();
        
        try {
            // 1. Query 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;  // Already rolled back or committed
            }
            
            // 2. Data validation
            if (!dataValidation(conn, branchUndoLog)) {
                throw new SQLException("Data validation failed");
            }
            
            // 3. Generate reverse SQL and execute
            for (SQLUndoLog sqlUndoLog : branchUndoLog.getSqlUndoLogs()) {
                AbstractUndoExecutor undoExecutor = 
                    UndoExecutorFactory.getUndoExecutor(
                        dataSourceProxy.getDbType(), sqlUndoLog);
                undoExecutor.executeOn(conn);
            }
            
            // 4. Delete 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;
        }
    }
    
    /**
     * Data validation: compare current data with after image
     */
    private boolean dataValidation(Connection conn, 
                                   BranchUndoLog branchUndoLog) {
        for (SQLUndoLog sqlUndoLog : branchUndoLog.getSqlUndoLogs()) {
            TableRecords afterImage = sqlUndoLog.getAfterImage();
            TableRecords currentRecords = queryCurrentRecords(conn, afterImage);
            
            // Compare data consistency
            if (!afterImage.equals(currentRecords)) {
                return false;  // Data was dirty-written
            }
        }
        return true;
    }
}

// UPDATE reverse SQL
public class MySQLUndoUpdateExecutor extends AbstractUndoExecutor {
    
    @Override
    protected String buildUndoSQL() {
        TableRecords beforeImage = sqlUndoLog.getBeforeImage();
        
        // Generate UPDATE statement based on before image
        StringBuilder sql = new StringBuilder("UPDATE ");
        sql.append(sqlUndoLog.getTableName()).append(" SET ");
        
        // Set field values to before image values
        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 condition (primary key)
        sql.append(" WHERE ");
        appendWhereCondition(sql, beforeImage);
        
        return sql.toString();
    }
}

// DELETE reverse SQL (generate INSERT)
public class MySQLUndoDeleteExecutor extends AbstractUndoExecutor {
    
    @Override
    protected String buildUndoSQL() {
        TableRecords beforeImage = sqlUndoLog.getBeforeImage();
        
        // Generate INSERT statement based on before image
        StringBuilder sql = new StringBuilder("INSERT INTO ");
        sql.append(sqlUndoLog.getTableName()).append(" (");
        
        // Field list
        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 reverse SQL (generate DELETE)
public class MySQLUndoInsertExecutor extends AbstractUndoExecutor {
    
    @Override
    protected String buildUndoSQL() {
        TableRecords afterImage = sqlUndoLog.getAfterImage();
        
        // Generate DELETE statement based on after image
        StringBuilder sql = new StringBuilder("DELETE FROM ");
        sql.append(sqlUndoLog.getTableName());
        sql.append(" WHERE ");
        
        appendWhereCondition(sql, afterImage);
        
        return sql.toString();
    }
}

public class LockManagerImpl implements LockManager {
    
    /**
     * Acquire global lock
     */
    public boolean acquireLock(List<RowLock> rowLocks) {
        // Lock granularity: table name + primary key value
        // Example: order_table:1,2,3
        String lockKey = buildLockKey(rowLocks);
        
        // Apply for global lock from TC
        boolean result = transactionCoordinator.acquireLock(
            xid, branchId, resourceId, lockKey);
        
        if (!result) {
            // Lock acquisition failed, enter retry logic
            return retryAcquireLock(rowLocks);
        }
        
        return true;
    }
    
    private String buildLockKey(List<RowLock> rowLocks) {
        // Format: 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 optimization
      async-commit-buffer-limit: 10000
      
    undo:
      # Only record updated columns
      only-care-update-columns: true
      
      # Compress undo_log
      compress:
        enable: true
        type: zip
        threshold: 64k

-- Clean undo_log older than 7 days
DELETE FROM undo_log 
WHERE log_created < DATE_SUB(NOW(), INTERVAL 7 DAY);