validation.go

package mysql

import (
	"errors"
	"fmt"
	"log/slog"
	"strconv"
	"strings"

	"github.com/go-mysql-org/go-mysql/client"
	"github.com/go-mysql-org/go-mysql/mysql"
	"go.temporal.io/sdk/log"
)

const (
	MySQLMinVersionForBinlogRowMetadata            = "8.0.1"
	MariaDBMinVersionForBinlogRowMetadata          = "10.5.0"
	MySQLMinVersionForBinlogTransactionCompression = "8.0.20"
	MySQLMinVersionForInvisibleColumns             = "8.0.23"
	MariaDBMinVersionForInvisibleColumns           = "10.3.0"
	MySQLMinVersionForGeneratedInvisiblePrimaryKey = "8.0.30"
)

func CompareServerVersion(conn *client.Conn, version string) (int, error) {
	rr, err := conn.Execute("SELECT version()")
	if err != nil {
		return 0, fmt.Errorf("failed to get server version: %w", err)
	}

	serverVersion, err := rr.GetString(0, 0)
	if err != nil {
		return 0, fmt.Errorf("failed to get server version: %w", err)
	}

	cmp, err := mysql.CompareServerVersions(serverVersion, version)
	if err != nil {
		return 0, fmt.Errorf("failed to compare server version: %w", err)
	}
	return cmp, nil
}

// only for MySQL 5.7 and below
func CheckMySQL5BinlogSettings(conn *client.Conn, logger log.Logger) error {
	if cmp, err := CompareServerVersion(conn, "5.5.0"); err != nil {
		return fmt.Errorf("failed to get server version: %w", err)
	} else if cmp < 0 {
		// TODO: revisit this when we have more data
		logger.Warn("cannot validate mysql prior to 5.5.0, bypass")
		return nil
	}

	// MySQL 5.6.6 introduced GTIDs so they may work, no need to enforce filepos for now

	query := "SELECT @@binlog_format"
	checkBinlogRowImage := false

	if cmp, err := CompareServerVersion(conn, "5.6.2"); err != nil {
		return fmt.Errorf("failed to get server version: %w", err)
	} else if cmp >= 0 {
		query += ", @@binlog_row_image"
		checkBinlogRowImage = true
	}

	// binlog_expire_logs_seconds was introduced in 8.0 https://dev.mysql.com/worklog/task/?id=10924
	// since expire_logs_days has day granularity, all settings of it work for us so not checking
	rs, err := conn.Execute(query)
	if err != nil {
		return fmt.Errorf("failed to retrieve settings <5.7: %w", err)
	}
	if len(rs.Values) == 0 {
		return fmt.Errorf("no value returned for settings <5.7")
	}
	row := rs.Values[0]

	binlogFormat := string(row[0].AsString())
	if binlogFormat != "ROW" {
		return fmt.Errorf("binlog_format must be set to 'ROW', currently " + binlogFormat)
	}
	if checkBinlogRowImage {
		binlogRowImage := string(row[1].AsString())
		if binlogRowImage != "FULL" {
			return fmt.Errorf("binlog_row_image must be set to 'FULL', currently " + binlogRowImage)
		}
	}

	return nil
}

func CheckMySQL8BinlogSettings(conn *client.Conn, logger log.Logger) error {
	query := "SELECT @@binlog_expire_logs_seconds, @@binlog_format, @@binlog_row_image, @@binlog_row_metadata"
	// check if binlog_row_value_options is available
	checkRowValueOptions := false
	if cmp, err := CompareServerVersion(conn, "8.0.3"); err != nil {
		return fmt.Errorf("failed to get server version: %w", err)
	} else if cmp >= 0 {
		checkRowValueOptions = true
		query += ", @@binlog_row_value_options"
	}

	rs, err := conn.Execute(query)
	if err != nil {
		return fmt.Errorf("failed to retrieve settings: %w", err)
	}
	if len(rs.Values) == 0 {
		return fmt.Errorf("no value returned for settings")
	}
	row := rs.Values[0]

	binlogExpireLogsSeconds := row[0].AsUint64()
	if binlogExpireLogsSeconds < 86400 && binlogExpireLogsSeconds != 0 {
		return fmt.Errorf(
			"binlog_expire_logs_seconds must be set to at least 86400 (24 hours), currently " +
				strconv.FormatUint(binlogExpireLogsSeconds, 10))
	}
	binlogFormat := string(row[1].AsString())
	if binlogFormat != "ROW" {
		return fmt.Errorf("binlog_format must be set to 'ROW', currently " + binlogFormat)
	}
	binlogRowImage := string(row[2].AsString())
	if binlogRowImage != "FULL" {
		return fmt.Errorf("binlog_row_image must be set to 'FULL', currently " + binlogRowImage)
	}
	binlogRowMetadata := string(row[3].AsString())
	if binlogRowMetadata != "FULL" {
		return fmt.Errorf("binlog_row_metadata must be set to 'FULL', currently " + binlogRowMetadata)
	}
	if checkRowValueOptions {
		binlogRowValueOptions := string(row[4].AsString())
		if binlogRowValueOptions != "" {
			return fmt.Errorf("binlog_row_value_options must be disabled, currently " + binlogRowValueOptions)
		}
	}

	return nil
}

func CheckMariaDBBinlogSettings(conn *client.Conn, logger log.Logger, requireRowMetadata bool) error {
	query := "SELECT @@binlog_format, @@binlog_row_image"

	checkBinlogRowMetadata := false
	cmp, err := CompareServerVersion(conn, MariaDBMinVersionForBinlogRowMetadata)
	if err != nil {
		return fmt.Errorf("failed to get server version: %w", err)
	}
	if cmp >= 0 {
		checkBinlogRowMetadata = true
		query += ", @@binlog_row_metadata"
	} else {
		if requireRowMetadata {
			return fmt.Errorf("MariaDB version too old for column exclusion support, "+
				"please disable it or upgrade to >=%s (binlog_row_metadata needed)",
				MariaDBMinVersionForBinlogRowMetadata)
		}
		logger.Warn("MariaDB version does not support binlog_row_metadata, skipping check")
		// to keep index ordering
		query += ", NULL"
	}
	checkBinlogExpiry := false
	// if binlog_row_metadata is not supported, this isn't either
	if checkBinlogRowMetadata {
		cmp, err = CompareServerVersion(conn, "10.6.1")
		if err != nil {
			return fmt.Errorf("failed to get server version: %w", err)
		}
		if cmp >= 0 {
			checkBinlogExpiry = true
			query += ", @@binlog_expire_logs_seconds"
		} else {
			logger.Warn("MariaDB version does not support binlog_expire_logs_seconds, skipping check")
		}
	}

	rs, err := conn.Execute(query)
	if err != nil {
		return fmt.Errorf("failed to retrieve settings: %w", err)
	}
	if len(rs.Values) == 0 {
		return fmt.Errorf("no value returned for settings")
	}
	row := rs.Values[0]

	binlogFormat := string(row[0].AsString())
	if binlogFormat != "ROW" {
		return fmt.Errorf("binlog_format must be set to 'ROW', currently " + binlogFormat)
	}

	binlogRowImage := string(row[1].AsString())
	if binlogRowImage != "FULL" {
		return fmt.Errorf("binlog_row_image must be set to 'FULL', currently " + binlogRowImage)
	}

	if checkBinlogRowMetadata {
		binlogRowMetadata := string(row[2].AsString())
		if binlogRowMetadata != "FULL" {
			// only strictly required for column exclusion support, but let's enforce it for consistency
			return fmt.Errorf("binlog_row_metadata must be set to 'FULL', currently " + binlogRowMetadata)
		}
	}

	if checkBinlogExpiry {
		binlogExpireLogsSeconds := row[3].AsUint64()
		if binlogExpireLogsSeconds < 86400 && binlogExpireLogsSeconds != 0 {
			return fmt.Errorf(
				"binlog_expire_logs_seconds must be set to at least 86400 (24 hours), currently " +
					strconv.FormatUint(binlogExpireLogsSeconds, 10))
		}
	}

	return nil
}

func CheckRDSBinlogSettings(conn *client.Conn, logger log.Logger) error {
	// AWS RDS/Aurora has its own binlog retention setting that we need to check, minimum 24h
	// check RDS/Aurora binlog retention setting
	if rs, err := conn.Execute("SELECT value FROM mysql.rds_configuration WHERE name='binlog retention hours'"); err != nil {
		if mErr, ok := errors.AsType[*mysql.MyError](err); ok && (mErr.Code == mysql.ER_NO_SUCH_TABLE || mErr.Code == mysql.ER_TABLEACCESS_DENIED_ERROR) {
			// Table doesn't exist, which means this is not RDS/Aurora
			logger.Warn("mysql.rds_configuration table does not exist, skipping Aurora/RDS binlog retention check",
				slog.Any("error", err))
			return nil
		}
		return fmt.Errorf("failed to check RDS/Aurora binlog retention hours: " + err.Error())
	} else if len(rs.Values) > 0 {
		binlogRetentionHoursStr := string(rs.Values[0][0].AsString())
		if binlogRetentionHoursStr == "" {
			return fmt.Errorf("RDS/Aurora setting 'binlog retention hours' should be at least 24, currently unset")
		}
		if binlogRetentionHours, err := strconv.Atoi(binlogRetentionHoursStr); err != nil {
			return fmt.Errorf("failed to parse RDS/Aurora setting 'binlog retention hours': " + err.Error())
		} else if binlogRetentionHours < 24 {
			return fmt.Errorf("RDS/Aurora setting 'binlog retention hours' should be at least 24, currently " + binlogRetentionHoursStr)
		}
	} else {
		logger.Warn("binlog retention hours returned nothing, skipping Aurora/RDS binlog retention check")
	}

	return nil
}

// CheckLogReplicaUpdates verifies that a source acting as a replica records replicated
// events in its own binary log. Without this, rows applied to the replica through its own
// upstream replication are not written to the replica's binlog, leaving gaps that surface
// as errors when PeerDB reconnects and resumes from a binlog position/GTID.
func CheckLogReplicaUpdates(conn *client.Conn) error {
	isReplica, err := ServerIsReplica(conn)
	if err != nil {
		return fmt.Errorf("failed to determine whether source is a replica: %w", err)
	}
	if !isReplica {
		return nil
	}

	rs, err := conn.Execute("SHOW VARIABLES WHERE Variable_name IN ('log_slave_updates', 'log_replica_updates')")
	if err != nil {
		return fmt.Errorf("failed to retrieve log_replica_updates setting: %w", err)
	}
	if len(rs.Values) == 0 {
		// every MySQL/MariaDB exposes one of these variables, so a replica that reports
		// neither is unexpected and we cannot confirm it records replicated events
		return fmt.Errorf("could not read log_replica_updates/log_slave_updates on replica source")
	}
	for _, row := range rs.Values {
		name := string(row[0].AsString())
		value := string(row[1].AsString())
		if !strings.EqualFold(value, "ON") && value != "1" {
			return fmt.Errorf(
				"%s must be enabled when replicating from a replica so replicated events are written to its binlog, currently %s",
				name, value)
		}
	}

	return nil
}

// ServerIsReplica reports whether the server is configured as a replica of an upstream primary.
func ServerIsReplica(conn *client.Conn) (bool, error) {
	rs, err := conn.Execute("SHOW REPLICA STATUS")
	if err != nil {
		// older MySQL (<8.0.22) and MariaDB (<10.5.1) don't recognize SHOW REPLICA STATUS
		if mErr, ok := errors.AsType[*mysql.MyError](err); ok && mErr.Code == mysql.ER_PARSE_ERROR {
			if rs, err = conn.Execute("SHOW SLAVE STATUS"); err != nil {
				return false, fmt.Errorf("failed to check replica status: %w", err)
			}
		} else {
			return false, fmt.Errorf("failed to check replica status: %w", err)
		}
	}
	return len(rs.Values) > 0, nil
}

// check if the server is a Vitess server, currently only works for initial load only
func IsVitess(conn *client.Conn) (bool, error) {
	if _, err := conn.Execute("SHOW VITESS_TABLETS"); err != nil {
		if mErr, ok := errors.AsType[*mysql.MyError](err); ok && mErr.Code == mysql.ER_PARSE_ERROR {
			return false, nil // not a Vitess server
		}
		return false, fmt.Errorf("failed to check if Vitess: %w", err)
	}
	return true, nil // is a Vitess server
}

func IsCompressedColumnType(columnType string) bool {
	return strings.Contains(strings.ToUpper(columnType), "COMPRESSED")
}

func CheckCompressedColumns(conn *client.Conn, schema string, table string) error {
	rs, err := conn.Execute(fmt.Sprintf(
		"SELECT column_name, column_type FROM information_schema.columns WHERE table_schema = '%s' AND table_name = '%s'",
		mysql.Escape(schema), mysql.Escape(table)))
	if err != nil {
		return fmt.Errorf("failed to fetch column types for %s.%s: %w", schema, table, err)
	}

	var compressed []string
	for idx := range rs.RowNumber() {
		columnType, err := rs.GetString(idx, 1)
		if err != nil {
			return err
		}
		if IsCompressedColumnType(columnType) {
			columnName, err := rs.GetString(idx, 0)
			if err != nil {
				return err
			}
			compressed = append(compressed, columnName)
		}
	}

	if len(compressed) > 0 {
		return fmt.Errorf(
			"table %s.%s has MariaDB COMPRESSED column(s) [%s], which cannot be replicated via CDC; "+
				"convert them to a non-compressed type or remove the table from the mirror",
			schema, table, strings.Join(compressed, ", "))
	}
	return nil
}

func HasReplicaWithServerId(conn *client.Conn, serverID uint32) (bool, error) {
	// This check assumes that the number of replicas is not beyond the order of magnitude of thousands.
	// This is a reasonable assumption given that services like RDS limit them to 15 and that
	// this number is constrained by concurrent connections and main server resources.
	rs, err := conn.Execute("SHOW REPLICAS")
	if err != nil {
		if rs, err = conn.Execute("SHOW SLAVE HOSTS"); err != nil {
			return false, fmt.Errorf("failed to query source for registered replicas")
		}
	}

	serverIDCol := -1
	for i, field := range rs.Fields {
		// Case insensitive match because the column name differs between MySQL and MariaDB.
		if strings.EqualFold(string(field.Name), "Server_id") {
			serverIDCol = i
			break
		}
	}
	if serverIDCol < 0 {
		return false, fmt.Errorf("no Server_id column in replica list, got fields %v", rs.Fields)
	}

	for row := range rs.RowNumber() {
		got, err := rs.GetInt(row, serverIDCol)
		if err != nil {
			return false, err
		}
		if uint32(got) == serverID {
			return true, nil
		}
	}
	return false, nil
}