validate.go

package connmysql

import (
	"context"
	"errors"
	"fmt"
	"log/slog"

	"github.com/go-mysql-org/go-mysql/client"
	"github.com/go-mysql-org/go-mysql/mysql"

	"github.com/PeerDB-io/peerdb/flow/generated/protos"
	"github.com/PeerDB-io/peerdb/flow/pkg/common"
	mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql"
)

func (c *MySqlConnector) CheckSourceTables(ctx context.Context, tableNames []*common.QualifiedTable) error {
	var missingTables []common.QualifiedTable
	for _, parsedTable := range tableNames {
		if _, err := c.Execute(ctx, fmt.Sprintf("SELECT * FROM %s LIMIT 0", parsedTable.MySQL())); err != nil {
			if mErr, ok := errors.AsType[*mysql.MyError](err); ok && mErr.Code == mysql.ER_NO_SUCH_TABLE {
				missingTables = append(missingTables, *parsedTable)
				continue
			}
			return fmt.Errorf("error checking table %s: %w", parsedTable.MySQL(), err)
		}
	}
	if len(missingTables) > 0 {
		return common.NewSourceTablesMissingError(missingTables)
	}
	return nil
}

func (c *MySqlConnector) CheckReplicationConnectivity(ctx context.Context) error {
	// GTID -> check GTID and error out if not enabled, check filepos as well
	// AUTO -> check GTID and fall back to filepos check
	// FILEPOS -> check filepos only
	if c.config.ReplicationMechanism != protos.MySqlReplicationMechanism_MYSQL_FILEPOS {
		if _, err := c.GetMasterGTIDSet(ctx); err != nil {
			if c.config.ReplicationMechanism == protos.MySqlReplicationMechanism_MYSQL_GTID {
				return fmt.Errorf("failed to check replication status: %w", err)
			}
			c.logger.Warn("[mysql] AUTO replication: GTID not available, will use file/position mode", slog.Any("error", err))
		}
	}
	if namePos, err := c.GetMasterPos(ctx); err != nil {
		return fmt.Errorf("failed to check replication status: %w", err)
	} else if namePos.Name == "" || namePos.Pos <= 0 {
		return fmt.Errorf("invalid replication status: missing log file or position")
	}

	return nil
}

func (c *MySqlConnector) CheckBinlogSettings(ctx context.Context, requireRowMetadata bool) error {
	conn, err := c.connect(ctx)
	if err != nil {
		return fmt.Errorf("failed to connect: %w", err)
	}

	switch c.config.Flavor {
	case protos.MySqlFlavor_MYSQL_MARIA:
		// check if binlog_row_metadata is supported is done inside this function
		return mysql_validation.CheckMariaDBBinlogSettings(conn, c.logger, requireRowMetadata)
	case protos.MySqlFlavor_MYSQL_MYSQL:
		cmp, err := c.CompareServerVersion(ctx, mysql_validation.MySQLMinVersionForBinlogRowMetadata)
		if err != nil {
			return fmt.Errorf("failed to get server version: %w", err)
		}
		if cmp < 0 {
			// as we're dispatching to a function that doesn't know about binlog_row_metadata,
			// perform the check here instead of inside
			if requireRowMetadata {
				return fmt.Errorf("MySQL version too old for column exclusion support, "+
					"please disable it or upgrade to >=%s (binlog_row_metadata needed)",
					mysql_validation.MySQLMinVersionForBinlogRowMetadata)
			}
			c.logger.Warn("Falling back to MySQL 5.7 check")
			return mysql_validation.CheckMySQL5BinlogSettings(conn, c.logger)
		} else {
			return mysql_validation.CheckMySQL8BinlogSettings(conn, c.logger)
		}
	default:
		return fmt.Errorf("unsupported MySQL flavor: %s", c.config.Flavor.String())
	}
}

func (c *MySqlConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.FlowConnectionConfigsCore) error {
	sourceTables := make([]*common.QualifiedTable, 0, len(cfg.TableMappings))
	for _, tableMapping := range cfg.TableMappings {
		parsedTable, parseErr := common.ParseTableIdentifier(tableMapping.SourceTableIdentifier)
		if parseErr != nil {
			return fmt.Errorf("invalid source table identifier: %w", parseErr)
		}
		sourceTables = append(sourceTables, parsedTable)
	}

	if err := c.CheckSourceTables(ctx, sourceTables); err != nil {
		return fmt.Errorf("provided source tables invalidated: %w", err)
	}
	// no need to check replication stuff for initial snapshot only mirrors
	if cfg.DoInitialSnapshot && cfg.InitialSnapshotOnly {
		return nil
	}

	if err := c.CheckReplicationConnectivity(ctx); err != nil {
		return fmt.Errorf("unable to establish replication connectivity: %w", err)
	}

	conn, err := c.connect(ctx)
	if err != nil {
		return fmt.Errorf("unable to connect: %w", err)
	}

	if isVitess, err := mysql_validation.IsVitess(conn); err != nil {
		return err
	} else if isVitess && !(cfg.DoInitialSnapshot && cfg.InitialSnapshotOnly) {
		return fmt.Errorf("vitess is currently not supported for MySQL mirrors in CDC")
	}
	if err := mysql_validation.CheckRDSBinlogSettings(conn, c.logger); err != nil {
		return fmt.Errorf("binlog configuration error: %w", err)
	}
	if err := mysql_validation.CheckLogReplicaUpdates(conn); err != nil {
		return fmt.Errorf("check log replica updates error: %w", err)
	}

	requireRowMetadata := false
	for _, tm := range cfg.TableMappings {
		if len(tm.Exclude) > 0 {
			requireRowMetadata = true
			break
		}
	}
	if err := c.CheckBinlogSettings(ctx, requireRowMetadata); err != nil {
		return fmt.Errorf("binlog configuration error: %w", err)
	}

	return nil
}

// HasReplicaWithServerId reports whether the source lists a replica registered with serverID.
func (c *MySqlConnector) HasReplicaWithServerId(ctx context.Context, serverID uint32) (bool, error) {
	conn, err := c.connect(ctx)
	if err != nil {
		return false, fmt.Errorf("failed to connect: %w", err)
	}
	return mysql_validation.HasReplicaWithServerId(conn, serverID)
}

func (c *MySqlConnector) ValidateCheck(ctx context.Context) error {
	if c.config.Flavor == protos.MySqlFlavor_MYSQL_UNKNOWN {
		return fmt.Errorf("flavor is set to unknown")
	}

	conn, err := c.connect(ctx)
	if err != nil {
		return fmt.Errorf("failed to connect: %w", err)
	}

	return c.validateFlavor(conn)
}

func (c *MySqlConnector) validateFlavor(conn *client.Conn) error {
	// MariaDB specific setting, introduced in MariaDB 10.0.3
	if rs, err := conn.Execute("SELECT @@gtid_strict_mode"); err != nil {
		// seems to be MySQL
		if mErr, ok := errors.AsType[*mysql.MyError](err); ok && mErr.Code == mysql.ER_UNKNOWN_SYSTEM_VARIABLE {
			if c.config.Flavor != protos.MySqlFlavor_MYSQL_MYSQL {
				return fmt.Errorf("server appears to be MySQL but MariaDB source has been selected")
			}
		} else {
			return fmt.Errorf("failed to check GTID mode: %w", err)
		}
	} else if len(rs.Values) > 0 {
		if c.config.Flavor != protos.MySqlFlavor_MYSQL_MARIA {
			return fmt.Errorf("server appears to be MariaDB but MySQL source has been selected")
		}
	}

	return nil
}