Mysql 如何回滚操作失误的数据（通过 binlog 日志）

package main

import (
    "flag"
    "fmt"
    "os"
    "time"

    "github.com/go-mysql-org/go-mysql/canal"
    "github.com/go-mysql-org/go-mysql/mysql"
    "github.com/go-mysql-org/go-mysql/replication"
)

// 自定义 canal 事件处理
type MyEventHandler struct {
    canal.DummyEventHandler
    rollbackSQL []string // 存储回滚 SQL
    startTime   time.Time
    endTime     time.Time
}

// 处理行事件（delete/update/insert 都会触发）
func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
    // 过滤时间范围外的事件
    eventTime := time.Unix(int64(e.Header.Timestamp), 0)
    if eventTime.Before(h.startTime) || eventTime.After(h.endTime) {
        return nil
    }

    // 表名：e.Table.Schema 是数据库名，e.Table.Name 是表名
    tableName := fmt.Sprintf("%s.%s", e.Table.Schema, e.Table.Name)

    switch e.Action {
    case "delete":
        // delete 事件：生成 insert 回滚 SQL
        // e.Rows[0] 是删除前的行数据
        h.genInsertSQL(tableName, e.Table.Columns, e.Rows[0])
    case "update":
        // update 事件：生成 update 回滚 SQL（恢复到修改前）
        // e.Rows[0] 是修改前的行数据，e.Rows[1] 是修改后的
        h.genUpdateSQL(tableName, e.Table.Columns, e.Rows[0], e.Rows[1])
    }
    return nil
}

// 生成 insert 回滚 SQL（恢复 delete 数据）
func (h *MyEventHandler) genInsertSQL(tableName string, cols []*replication.Column, row []interface{}) {
    sql := fmt.Sprintf("INSERT INTO %s (", tableName)
    values := "VALUES ("

    for i, col := range cols {
        // 拼接字段名
        if i > 0 {
            sql += ","
            values += ","
        }
        sql += fmt.Sprintf("`%s`", col.Name)

        // 拼接字段值（处理字符串类型加引号）
        val := h.formatValue(row[i])
        values += val
    }

    sql += ") " + values + ";"
    h.rollbackSQL = append(h.rollbackSQL, sql)
}

// 生成 update 回滚 SQL（恢复 update 数据）
func (h *MyEventHandler) genUpdateSQL(tableName string, cols []*replication.Column, oldRow, newRow []interface{}) {
    sql := fmt.Sprintf("UPDATE %s SET ", tableName)
    where := " WHERE "

    hasSet := false
    hasWhere := false

    for i, col := range cols {
        oldVal := h.formatValue(oldRow[i])
        newVal := h.formatValue(newRow[i])

        // 拼接 SET 部分（用旧值覆盖新值）
        if oldVal != newVal {
            if hasSet {
                sql += ","
            }
            sql += fmt.Sprintf("`%s`=%s", col.Name, oldVal)
            hasSet = true
        }

        // 拼接 WHERE 部分（用主键或所有字段定位行）
        if col.IsPK {
            if hasWhere {
                where += " AND "
            }
            where += fmt.Sprintf("`%s`=%s", col.Name, newVal)
            hasWhere = true
        }
    }

    // 若没有主键，用所有字段定位（避免更新多条数据）
    if !hasWhere {
        for i, col := range cols {
            if hasWhere {
                where += " AND "
            }
            where += fmt.Sprintf("`%s`=%s", col.Name, newVal[i])
            hasWhere = true
        }
    }

    if hasSet && hasWhere {
        sql += where + ";"
        h.rollbackSQL = append(h.rollbackSQL, sql)
    }
}

// 格式化字段值（字符串加引号，其他类型直接输出）
func (h *MyEventHandler) formatValue(val interface{}) string {
    switch v := val.(type) {
    case string:
        return fmt.Sprintf("'%s'", v)
    case []byte:
        return fmt.Sprintf("'%s'", string(v))
    default:
        return fmt.Sprintf("%v", v)
    }
}

// 保存回滚 SQL 到文件
func (h *MyEventHandler) SaveToFile(filePath string) error {
    f, err := os.Create(filePath)
    if err != nil {
        return err
    }
    defer f.Close()

    for _, sql := range h.rollbackSQL {
        _, err := f.WriteString(sql + "\n")
        if err != nil {
            return err
        }
    }
    return nil
}

func main() {
    // 命令行参数：指定 MySQL 连接信息、时间范围、输出文件
    host := flag.String("host", "127.0.0.1", "MySQL 主机地址")
    port := flag.Int("port", 3306, "MySQL 端口")
    user := flag.String("user", "root", "MySQL 用户名")
    pass := flag.String("pass", "123456", "MySQL 密码")
    db := flag.String("db", "test", "目标数据库名")
    startTime := flag.String("start", "", "开始时间（格式：2025-12-04 10:00:00）")
    endTime := flag.String("end", "", "结束时间（格式：2025-12-04 10:30:00）")
    output := flag.String("output", "rollback.sql", "回滚 SQL 输出文件")
    flag.Parse()

    // 解析时间参数
    start, err := time.Parse("2006-01-02 15:04:05", *startTime)
    if err != nil {
        fmt.Printf("解析开始时间错误：%v\n", err)
        return
    }
    end, err := time.Parse("2006-01-02 15:04:05", *endTime)
    if err != nil {
        fmt.Printf("解析结束时间错误：%v\n", err)
        return
    }

    // 初始化 canal（binlog 解析核心）
    cfg := canal.NewDefaultConfig()
    cfg.Addr = fmt.Sprintf("%s:%d", *host, *port)
    cfg.User = *user
    cfg.Passwd = *pass
    cfg.Database = *db
    cfg.IncludeTableRegex = []string{fmt.Sprintf("%s\\..*", *db)} // 只解析目标数据库的表

    c, err := canal.NewCanal(cfg)
    if err != nil {
        fmt.Printf("初始化 canal 错误：%v\n", err)
        return
    }

    // 注册自定义事件处理器
    handler := &MyEventHandler{
        startTime: start,
        endTime:   end,
    }
    c.SetEventHandler(handler)

    // 找到当前 binlog 位置并开始解析
    pos, err := c.GetMasterPos()
    if err != nil {
        fmt.Printf("获取 binlog 位置错误：%v\n", err)
        return
    }
    fmt.Printf("开始解析 binlog：%s 位置 %d\n", pos.Name, pos.Pos)

    // 开始解析（阻塞直到解析到指定结束时间）
    err = c.RunFrom(mysql.Position{Name: pos.Name, Pos: pos.Pos})
    if err != nil {
        fmt.Printf("解析 binlog 错误：%v\n", err)
        return
    }

    // 保存回滚 SQL 到文件
    err = handler.SaveToFile(*output)
    if err != nil {
        fmt.Printf("保存回滚 SQL 错误：%v\n", err)
        return
    }

    fmt.Printf("解析完成！回滚 SQL 已保存到 %s，共 %d 条\n", *output, len(handler.rollbackSQL))
}