919 lines
31 KiB
Go
919 lines
31 KiB
Go
package dbmigration
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"net"
|
|
"net/url"
|
|
"regexp"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.pcux.cn/xuans/Smail/config"
|
|
"git.pcux.cn/xuans/Smail/utils/credential"
|
|
mysql "github.com/go-sql-driver/mysql"
|
|
"github.com/lib/pq"
|
|
)
|
|
|
|
var (
|
|
ErrSQLCopyUnavailable = errors.New("structured database copier unavailable")
|
|
ErrSQLTargetExists = errors.New("structured database migration target already exists")
|
|
ErrSQLUnsupportedObject = errors.New("structured database contains unsupported object")
|
|
ErrSQLConnectionUnavailable = errors.New("structured database connection unavailable")
|
|
ErrSQLDataVerification = errors.New("structured database data verification failed")
|
|
ErrSQLSourceBusy = errors.New("structured database source has active connections")
|
|
ErrSwitchOutcomeUnknown = errors.New("database configuration switch outcome is unknown")
|
|
)
|
|
|
|
type AuthorizedDatabaseSwitch func(context.Context, uint64, config.DatabaseConfig, config.AuditActor) error
|
|
|
|
type StructuredDatabaseCopier interface {
|
|
Copy(context.Context, config.DatabaseConfig, config.DatabaseConfig) error
|
|
}
|
|
|
|
type StructuredDatabaseCleanup interface {
|
|
Cleanup(context.Context, config.DatabaseConfig, config.DatabaseConfig) error
|
|
Complete(config.DatabaseConfig, config.DatabaseConfig)
|
|
}
|
|
|
|
type DatabaseExecutor struct {
|
|
sqlite Runner
|
|
structured Runner
|
|
}
|
|
|
|
func NewDatabaseExecutor(sqlite, structured Runner) Runner {
|
|
return &DatabaseExecutor{sqlite: sqlite, structured: structured}
|
|
}
|
|
|
|
func (e *DatabaseExecutor) Run(ctx context.Context, plan Plan, update func(Status) error) error {
|
|
if e == nil {
|
|
return ErrExecutorUnavailable
|
|
}
|
|
if plan.sourceConfig.Type == config.DatabaseTypeSQLite {
|
|
if e.sqlite == nil {
|
|
return ErrExecutorUnavailable
|
|
}
|
|
return e.sqlite.Run(ctx, plan, update)
|
|
}
|
|
if e.structured == nil {
|
|
return ErrExecutorUnavailable
|
|
}
|
|
return e.structured.Run(ctx, plan, update)
|
|
}
|
|
|
|
type SQLExecutor struct {
|
|
copier StructuredDatabaseCopier
|
|
authorizedSwitch AuthorizedDatabaseSwitch
|
|
}
|
|
|
|
func NewSQLExecutor(copier StructuredDatabaseCopier, switchConfig AuthorizedDatabaseSwitch) *SQLExecutor {
|
|
return &SQLExecutor{copier: copier, authorizedSwitch: switchConfig}
|
|
}
|
|
|
|
func (e *SQLExecutor) Run(ctx context.Context, plan Plan, update func(Status) error) error {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
if e == nil || update == nil {
|
|
return ErrInvalidRequest
|
|
}
|
|
if err := contextError(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := validateSQLCopyRequest(plan.sourceConfig, plan.targetConfig); err != nil {
|
|
return err
|
|
}
|
|
if e.copier == nil {
|
|
return ErrSQLCopyUnavailable
|
|
}
|
|
if e.authorizedSwitch == nil {
|
|
return ErrSwitchUnavailable
|
|
}
|
|
if err := update(StatusVerifying); err != nil {
|
|
return err
|
|
}
|
|
if err := e.copier.Copy(ctx, plan.sourceConfig, plan.targetConfig); err != nil {
|
|
cleanupStructuredTarget(e.copier, plan.sourceConfig, plan.targetConfig)
|
|
return err
|
|
}
|
|
if err := contextError(ctx); err != nil {
|
|
cleanupStructuredTarget(e.copier, plan.sourceConfig, plan.targetConfig)
|
|
return err
|
|
}
|
|
if err := update(StatusSwitching); err != nil {
|
|
cleanupStructuredTarget(e.copier, plan.sourceConfig, plan.targetConfig)
|
|
return err
|
|
}
|
|
if e.authorizedSwitch == nil {
|
|
cleanupStructuredTarget(e.copier, plan.sourceConfig, plan.targetConfig)
|
|
return ErrSwitchUnavailable
|
|
}
|
|
if err := e.authorizedSwitch(ctx, plan.expectedGeneration, plan.targetConfig, plan.actor); err != nil {
|
|
if isKnownSwitchRejection(err) {
|
|
cleanupStructuredTarget(e.copier, plan.sourceConfig, plan.targetConfig)
|
|
return err
|
|
}
|
|
return errors.Join(ErrSwitchOutcomeUnknown, err)
|
|
}
|
|
if cleanup, ok := e.copier.(StructuredDatabaseCleanup); ok {
|
|
cleanup.Complete(plan.sourceConfig, plan.targetConfig)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func isKnownSwitchRejection(err error) bool {
|
|
return errors.Is(err, config.ErrAuthorizationRequired) ||
|
|
errors.Is(err, config.ErrDatabaseMigrationTarget) ||
|
|
errors.Is(err, config.ErrGenerationConflict) ||
|
|
errors.Is(err, config.ErrStepUpRequired) ||
|
|
errors.Is(err, config.ErrInvalidRuntimeConfig)
|
|
}
|
|
|
|
func cleanupStructuredTarget(copier StructuredDatabaseCopier, source, target config.DatabaseConfig) {
|
|
cleanup, ok := copier.(StructuredDatabaseCleanup)
|
|
if !ok {
|
|
return
|
|
}
|
|
cleanupContext, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
_ = cleanup.Cleanup(cleanupContext, source, target)
|
|
}
|
|
|
|
func sameStructuredConnection(source, target config.DatabaseConfig) bool {
|
|
return source.Type == target.Type &&
|
|
source.Host == target.Host &&
|
|
source.Port == target.Port &&
|
|
source.Username == target.Username &&
|
|
source.Password == target.Password &&
|
|
source.LocalRef == target.LocalRef
|
|
}
|
|
|
|
type SQLConnectionFactory func(context.Context, config.DatabaseConfig, string) (*sql.DB, error)
|
|
|
|
type SQLCopier struct {
|
|
open SQLConnectionFactory
|
|
mu sync.Mutex
|
|
created map[string]struct{}
|
|
}
|
|
|
|
func NewSQLCopier(open SQLConnectionFactory) *SQLCopier {
|
|
return &SQLCopier{open: open, created: make(map[string]struct{})}
|
|
}
|
|
|
|
func (c *SQLCopier) Copy(ctx context.Context, source, target config.DatabaseConfig) error {
|
|
if c == nil || c.open == nil {
|
|
return ErrSQLConnectionUnavailable
|
|
}
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
if err := contextError(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := validateSQLCopyRequest(source, target); err != nil {
|
|
return err
|
|
}
|
|
adminDB, err := c.open(ctx, source, "")
|
|
if err != nil || adminDB == nil {
|
|
return ErrSQLConnectionUnavailable
|
|
}
|
|
defer adminDB.Close()
|
|
if err := contextError(ctx); err != nil {
|
|
return err
|
|
}
|
|
switch source.Type {
|
|
case config.DatabaseTypeMySQL:
|
|
sourceDB, err := c.open(ctx, source, source.Name)
|
|
if err != nil || sourceDB == nil {
|
|
return ErrSQLConnectionUnavailable
|
|
}
|
|
defer sourceDB.Close()
|
|
return c.copyMySQL(ctx, sourceDB, adminDB, source, target)
|
|
case config.DatabaseTypePostgreSQL:
|
|
return c.copyPostgreSQL(ctx, adminDB, source, target)
|
|
default:
|
|
return ErrUnsupportedDatabase
|
|
}
|
|
}
|
|
|
|
func (c *SQLCopier) copyMySQL(ctx context.Context, sourceDB, adminDB *sql.DB, source, target config.DatabaseConfig) error {
|
|
sourceName := source.Name
|
|
targetName := target.Name
|
|
if err := c.validateMySQLObjects(ctx, adminDB, sourceName); err != nil {
|
|
return err
|
|
}
|
|
var exists int
|
|
if err := adminDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM information_schema.schemata WHERE schema_name = ?", targetName).Scan(&exists); err != nil {
|
|
return fmt.Errorf("check mysql migration target: %w", err)
|
|
}
|
|
if exists != 0 {
|
|
return ErrSQLTargetExists
|
|
}
|
|
if _, err := adminDB.ExecContext(ctx, "CREATE DATABASE "+quoteMySQLIdentifier(targetName)); err != nil {
|
|
if isDatabaseAlreadyExistsError(err, config.DatabaseTypeMySQL) {
|
|
return ErrSQLTargetExists
|
|
}
|
|
return fmt.Errorf("create mysql migration target: %w", err)
|
|
}
|
|
c.markCreated(source, target)
|
|
created := true
|
|
defer func() {
|
|
if created {
|
|
cleanupContext, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
if _, err := adminDB.ExecContext(cleanupContext, "DROP DATABASE "+quoteMySQLIdentifier(targetName)); err == nil {
|
|
c.forgetCreated(source, target)
|
|
}
|
|
}
|
|
}()
|
|
|
|
rows, err := sourceDB.QueryContext(ctx, "SHOW FULL TABLES FROM "+quoteMySQLIdentifier(sourceName))
|
|
if err != nil {
|
|
return fmt.Errorf("list mysql tables: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
tableNames := make([]string, 0)
|
|
sourceRowCounts := make(map[string]int64)
|
|
for rows.Next() {
|
|
if err := contextError(ctx); err != nil {
|
|
return err
|
|
}
|
|
var tableName, tableType string
|
|
if err := rows.Scan(&tableName, &tableType); err != nil {
|
|
return fmt.Errorf("read mysql table: %w", err)
|
|
}
|
|
if !strings.EqualFold(tableType, "BASE TABLE") {
|
|
return ErrSQLUnsupportedObject
|
|
}
|
|
tableNames = append(tableNames, tableName)
|
|
sourceTable := quoteMySQLIdentifier(sourceName) + "." + quoteMySQLIdentifier(tableName)
|
|
targetTable := quoteMySQLIdentifier(targetName) + "." + quoteMySQLIdentifier(tableName)
|
|
rowCount, err := countRows(ctx, sourceDB, sourceTable)
|
|
if err != nil {
|
|
return fmt.Errorf("count mysql source rows: %w", err)
|
|
}
|
|
sourceRowCounts[tableName] = rowCount
|
|
if _, err := adminDB.ExecContext(ctx, "CREATE TABLE "+targetTable+" LIKE "+sourceTable); err != nil {
|
|
return fmt.Errorf("copy mysql table structure: %w", err)
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return fmt.Errorf("read mysql tables: %w", err)
|
|
}
|
|
for _, tableName := range tableNames {
|
|
if err := contextError(ctx); err != nil {
|
|
return err
|
|
}
|
|
sourceTable := quoteMySQLIdentifier(sourceName) + "." + quoteMySQLIdentifier(tableName)
|
|
targetTable := quoteMySQLIdentifier(targetName) + "." + quoteMySQLIdentifier(tableName)
|
|
if _, err := adminDB.ExecContext(ctx, "INSERT INTO "+targetTable+" SELECT * FROM "+sourceTable); err != nil {
|
|
return fmt.Errorf("copy mysql table data: %w", err)
|
|
}
|
|
targetRowCount, err := countRows(ctx, adminDB, targetTable)
|
|
if err != nil {
|
|
return fmt.Errorf("count mysql target rows: %w", err)
|
|
}
|
|
finalSourceRowCount, err := countRows(ctx, sourceDB, sourceTable)
|
|
if err != nil {
|
|
return fmt.Errorf("verify mysql source rows: %w", err)
|
|
}
|
|
if sourceRowCounts[tableName] != targetRowCount || sourceRowCounts[tableName] != finalSourceRowCount {
|
|
return fmt.Errorf("%w: mysql table %s row count changed during migration", ErrSQLDataVerification, tableName)
|
|
}
|
|
}
|
|
if err := verifyMySQLTableStructure(ctx, adminDB, sourceName, targetName, tableNames); err != nil {
|
|
return err
|
|
}
|
|
if err := c.copyMySQLForeignKeys(ctx, adminDB, sourceName, targetName, tableNames); err != nil {
|
|
return err
|
|
}
|
|
if err := verifyMySQLForeignKeys(ctx, adminDB, sourceName, targetName); err != nil {
|
|
return err
|
|
}
|
|
created = false
|
|
return nil
|
|
}
|
|
|
|
type mysqlForeignKey struct {
|
|
tableName string
|
|
constraintName string
|
|
columns []string
|
|
referencedSchema string
|
|
referencedTable string
|
|
referencedColumns []string
|
|
updateRule string
|
|
deleteRule string
|
|
}
|
|
|
|
func (c *SQLCopier) validateMySQLObjects(ctx context.Context, database *sql.DB, sourceName string) error {
|
|
checks := []struct {
|
|
query string
|
|
args []any
|
|
}{
|
|
{
|
|
query: "SELECT COUNT(*) FROM information_schema.triggers WHERE trigger_schema = ?",
|
|
args: []any{sourceName},
|
|
},
|
|
{
|
|
query: "SELECT COUNT(*) FROM information_schema.events WHERE event_schema = ?",
|
|
args: []any{sourceName},
|
|
},
|
|
{
|
|
query: "SELECT COUNT(*) FROM information_schema.routines WHERE routine_schema = ?",
|
|
args: []any{sourceName},
|
|
},
|
|
}
|
|
for _, check := range checks {
|
|
var count int
|
|
if err := database.QueryRowContext(ctx, check.query, check.args...).Scan(&count); err != nil {
|
|
return fmt.Errorf("%w: inspect mysql database objects", ErrSQLConnectionUnavailable)
|
|
}
|
|
if count > 0 {
|
|
return ErrSQLUnsupportedObject
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *SQLCopier) copyMySQLForeignKeys(ctx context.Context, database *sql.DB, sourceName, targetName string, tableNames []string) error {
|
|
if len(tableNames) == 0 {
|
|
return nil
|
|
}
|
|
foreignKeys, err := listMySQLForeignKeys(ctx, database, sourceName)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: inspect mysql foreign keys", ErrSQLConnectionUnavailable)
|
|
}
|
|
for _, foreignKey := range foreignKeys {
|
|
if err := contextError(ctx); err != nil {
|
|
return err
|
|
}
|
|
referencedSchema := foreignKey.referencedSchema
|
|
if referencedSchema == sourceName {
|
|
referencedSchema = targetName
|
|
}
|
|
columns := quoteMySQLIdentifiers(foreignKey.columns)
|
|
referencedColumns := quoteMySQLIdentifiers(foreignKey.referencedColumns)
|
|
query := "ALTER TABLE " + quoteMySQLIdentifier(targetName) + "." + quoteMySQLIdentifier(foreignKey.tableName) + " ADD CONSTRAINT " + quoteMySQLIdentifier(foreignKey.constraintName) + " FOREIGN KEY (" + strings.Join(columns, ", ") + ") REFERENCES " + quoteMySQLIdentifier(referencedSchema) + "." + quoteMySQLIdentifier(foreignKey.referencedTable) + " (" + strings.Join(referencedColumns, ", ") + ") ON UPDATE " + foreignKey.updateRule + " ON DELETE " + foreignKey.deleteRule
|
|
if _, err := database.ExecContext(ctx, query); err != nil {
|
|
return fmt.Errorf("copy mysql foreign key: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func verifyMySQLForeignKeys(ctx context.Context, database *sql.DB, sourceName, targetName string) error {
|
|
source, err := listMySQLForeignKeys(ctx, database, sourceName)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list mysql source foreign keys for verification", ErrSQLConnectionUnavailable)
|
|
}
|
|
target, err := listMySQLForeignKeys(ctx, database, targetName)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list mysql target foreign keys for verification", ErrSQLConnectionUnavailable)
|
|
}
|
|
if !sameMySQLForeignKeys(source, target, sourceName, targetName) {
|
|
return fmt.Errorf("%w: mysql foreign key structure changed during migration", ErrSQLDataVerification)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func listMySQLForeignKeys(ctx context.Context, database *sql.DB, schemaName string) ([]mysqlForeignKey, error) {
|
|
rows, err := database.QueryContext(ctx, "SELECT kcu.TABLE_NAME, kcu.CONSTRAINT_NAME, kcu.COLUMN_NAME, kcu.REFERENCED_TABLE_SCHEMA, kcu.REFERENCED_TABLE_NAME, kcu.REFERENCED_COLUMN_NAME, rc.UPDATE_RULE, rc.DELETE_RULE FROM information_schema.KEY_COLUMN_USAGE kcu JOIN information_schema.TABLE_CONSTRAINTS tc ON tc.CONSTRAINT_SCHEMA = kcu.CONSTRAINT_SCHEMA AND tc.TABLE_NAME = kcu.TABLE_NAME AND tc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME JOIN information_schema.REFERENTIAL_CONSTRAINTS rc ON rc.CONSTRAINT_SCHEMA = kcu.CONSTRAINT_SCHEMA AND rc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME AND rc.TABLE_NAME = kcu.TABLE_NAME WHERE tc.CONSTRAINT_SCHEMA = ? AND tc.CONSTRAINT_TYPE = 'FOREIGN KEY' AND kcu.REFERENCED_TABLE_NAME IS NOT NULL ORDER BY kcu.TABLE_NAME, kcu.CONSTRAINT_NAME, kcu.ORDINAL_POSITION", schemaName)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
foreignKeys := make([]mysqlForeignKey, 0)
|
|
indexes := make(map[string]int)
|
|
for rows.Next() {
|
|
if err := contextError(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
var item mysqlForeignKey
|
|
var column, referencedColumn string
|
|
if err := rows.Scan(&item.tableName, &item.constraintName, &column, &item.referencedSchema, &item.referencedTable, &referencedColumn, &item.updateRule, &item.deleteRule); err != nil {
|
|
return nil, err
|
|
}
|
|
if !validMySQLReferenceRule(item.updateRule) || !validMySQLReferenceRule(item.deleteRule) {
|
|
return nil, ErrSQLUnsupportedObject
|
|
}
|
|
key := item.tableName + "\x00" + item.constraintName
|
|
index, ok := indexes[key]
|
|
if !ok {
|
|
indexes[key] = len(foreignKeys)
|
|
foreignKeys = append(foreignKeys, item)
|
|
index = len(foreignKeys) - 1
|
|
}
|
|
foreignKeys[index].columns = append(foreignKeys[index].columns, column)
|
|
foreignKeys[index].referencedColumns = append(foreignKeys[index].referencedColumns, referencedColumn)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return foreignKeys, nil
|
|
}
|
|
|
|
func sameMySQLForeignKeys(source, target []mysqlForeignKey, sourceName, targetName string) bool {
|
|
if len(source) != len(target) {
|
|
return false
|
|
}
|
|
sourceSignatures := make([]string, len(source))
|
|
for index, item := range source {
|
|
sourceSignatures[index] = mysqlForeignKeySignature(item, sourceName, targetName)
|
|
}
|
|
targetSignatures := make([]string, len(target))
|
|
for index, item := range target {
|
|
targetSignatures[index] = mysqlForeignKeySignature(item, sourceName, targetName)
|
|
}
|
|
sort.Strings(sourceSignatures)
|
|
sort.Strings(targetSignatures)
|
|
return sameStringSet(sourceSignatures, targetSignatures)
|
|
}
|
|
|
|
func mysqlForeignKeySignature(item mysqlForeignKey, sourceName, targetName string) string {
|
|
referencedSchema := item.referencedSchema
|
|
if referencedSchema == sourceName {
|
|
referencedSchema = targetName
|
|
}
|
|
return strings.Join([]string{
|
|
item.tableName,
|
|
item.constraintName,
|
|
strings.Join(item.columns, "\x00"),
|
|
referencedSchema,
|
|
item.referencedTable,
|
|
strings.Join(item.referencedColumns, "\x00"),
|
|
strings.ToUpper(strings.TrimSpace(item.updateRule)),
|
|
strings.ToUpper(strings.TrimSpace(item.deleteRule)),
|
|
}, "\x00")
|
|
}
|
|
|
|
func validMySQLReferenceRule(value string) bool {
|
|
switch strings.ToUpper(strings.TrimSpace(value)) {
|
|
case "CASCADE", "SET NULL", "SET DEFAULT", "RESTRICT", "NO ACTION":
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func quoteMySQLIdentifiers(values []string) []string {
|
|
quoted := make([]string, len(values))
|
|
for index, value := range values {
|
|
quoted[index] = quoteMySQLIdentifier(value)
|
|
}
|
|
return quoted
|
|
}
|
|
|
|
func (c *SQLCopier) copyPostgreSQL(ctx context.Context, adminDB *sql.DB, source, target config.DatabaseConfig) error {
|
|
sourceName := source.Name
|
|
targetName := target.Name
|
|
var exists bool
|
|
if err := adminDB.QueryRowContext(ctx, "SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname = $1)", targetName).Scan(&exists); err != nil {
|
|
return fmt.Errorf("check postgresql migration target: %w", err)
|
|
}
|
|
if exists {
|
|
return ErrSQLTargetExists
|
|
}
|
|
var activeConnections int
|
|
if err := adminDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM pg_stat_activity WHERE datname = $1 AND pid <> pg_backend_pid()", sourceName).Scan(&activeConnections); err != nil {
|
|
return fmt.Errorf("inspect postgresql source connections: %w", err)
|
|
}
|
|
if activeConnections > 0 {
|
|
return ErrSQLSourceBusy
|
|
}
|
|
if _, err := adminDB.ExecContext(ctx, "CREATE DATABASE "+quotePostgreSQLIdentifier(targetName)+" TEMPLATE "+quotePostgreSQLIdentifier(sourceName)); err != nil {
|
|
if isDatabaseAlreadyExistsError(err, config.DatabaseTypePostgreSQL) {
|
|
return ErrSQLTargetExists
|
|
}
|
|
return fmt.Errorf("copy postgresql database: %w", err)
|
|
}
|
|
c.markCreated(source, target)
|
|
created := true
|
|
defer func() {
|
|
if created {
|
|
cleanupContext, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
if err := c.Cleanup(cleanupContext, source, target); err == nil {
|
|
created = false
|
|
}
|
|
}
|
|
}()
|
|
if err := c.verifyPostgreSQLClone(ctx, source, target); err != nil {
|
|
return err
|
|
}
|
|
created = false
|
|
return nil
|
|
}
|
|
|
|
func (c *SQLCopier) verifyPostgreSQLClone(ctx context.Context, source, target config.DatabaseConfig) error {
|
|
if c == nil || c.open == nil {
|
|
return ErrSQLConnectionUnavailable
|
|
}
|
|
sourceDB, err := c.open(ctx, source, source.Name)
|
|
if err != nil {
|
|
return ErrSQLConnectionUnavailable
|
|
}
|
|
defer sourceDB.Close()
|
|
targetDB, err := c.open(ctx, target, target.Name)
|
|
if err != nil {
|
|
return ErrSQLConnectionUnavailable
|
|
}
|
|
defer targetDB.Close()
|
|
|
|
tables, err := listPostgreSQLTables(ctx, sourceDB)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list postgresql source tables", ErrSQLConnectionUnavailable)
|
|
}
|
|
targetTables, err := listPostgreSQLTables(ctx, targetDB)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list postgresql target tables", ErrSQLConnectionUnavailable)
|
|
}
|
|
if !samePostgreSQLTables(tables, targetTables) {
|
|
return fmt.Errorf("%w: postgresql table structure changed during migration", ErrSQLDataVerification)
|
|
}
|
|
for _, table := range tables {
|
|
sourceColumns, err := listPostgreSQLColumns(ctx, sourceDB, table)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list postgresql source columns", ErrSQLConnectionUnavailable)
|
|
}
|
|
targetColumns, err := listPostgreSQLColumns(ctx, targetDB, table)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list postgresql target columns", ErrSQLConnectionUnavailable)
|
|
}
|
|
if !sameStringSet(sourceColumns, targetColumns) {
|
|
return fmt.Errorf("%w: postgresql table %s columns changed during migration", ErrSQLDataVerification, table.name)
|
|
}
|
|
sourceCount, err := countRows(ctx, sourceDB, quotePostgreSQLIdentifier(table.schema)+"."+quotePostgreSQLIdentifier(table.name))
|
|
if err != nil {
|
|
return fmt.Errorf("%w: count postgresql source rows", ErrSQLConnectionUnavailable)
|
|
}
|
|
targetCount, err := countRows(ctx, targetDB, quotePostgreSQLIdentifier(table.schema)+"."+quotePostgreSQLIdentifier(table.name))
|
|
if err != nil {
|
|
return fmt.Errorf("%w: count postgresql target rows", ErrSQLConnectionUnavailable)
|
|
}
|
|
if sourceCount != targetCount {
|
|
return fmt.Errorf("%w: postgresql table %s row count changed during migration", ErrSQLDataVerification, table.name)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateSQLCopyRequest(source, target config.DatabaseConfig) error {
|
|
source = config.NormalizeDatabaseConfig(source)
|
|
target = config.NormalizeDatabaseConfig(target)
|
|
if source.Type != target.Type ||
|
|
(source.Type != config.DatabaseTypeMySQL && source.Type != config.DatabaseTypePostgreSQL) {
|
|
return ErrUnsupportedDatabase
|
|
}
|
|
if err := config.ValidateDatabaseConfig(source); err != nil {
|
|
return ErrSourceDatabase
|
|
}
|
|
if err := config.ValidateDatabaseConfig(target); err != nil {
|
|
return ErrInvalidTargetName
|
|
}
|
|
if !validStructuredDatabaseName(source.Name) {
|
|
return ErrSourceDatabase
|
|
}
|
|
if !validStructuredDatabaseName(target.Name) {
|
|
return ErrInvalidTargetName
|
|
}
|
|
if source.Name == target.Name ||
|
|
!sameStructuredConnection(source, target) {
|
|
return ErrImmutableSetting
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func verifyMySQLTableStructure(ctx context.Context, database *sql.DB, sourceName, targetName string, tableNames []string) error {
|
|
sourceTables, err := listMySQLTables(ctx, database, sourceName)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list mysql source tables for verification", ErrSQLConnectionUnavailable)
|
|
}
|
|
targetTables, err := listMySQLTables(ctx, database, targetName)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list mysql target tables for verification", ErrSQLConnectionUnavailable)
|
|
}
|
|
if !sameStringSet(sourceTables, targetTables) || !sameStringSet(sourceTables, tableNames) {
|
|
return fmt.Errorf("%w: mysql table structure changed during migration", ErrSQLDataVerification)
|
|
}
|
|
for _, tableName := range sourceTables {
|
|
sourceColumns, err := listMySQLColumns(ctx, database, sourceName, tableName)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list mysql source columns", ErrSQLConnectionUnavailable)
|
|
}
|
|
targetColumns, err := listMySQLColumns(ctx, database, targetName, tableName)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: list mysql target columns", ErrSQLConnectionUnavailable)
|
|
}
|
|
if !sameStringSet(sourceColumns, targetColumns) {
|
|
return fmt.Errorf("%w: mysql table %s columns changed during migration", ErrSQLDataVerification, tableName)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func listMySQLTables(ctx context.Context, database *sql.DB, databaseName string) ([]string, error) {
|
|
rows, err := database.QueryContext(ctx, "SELECT TABLE_NAME FROM information_schema.tables WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE' ORDER BY TABLE_NAME", databaseName)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
result := make([]string, 0)
|
|
for rows.Next() {
|
|
if err := contextError(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
var name string
|
|
if err := rows.Scan(&name); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, name)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func listMySQLColumns(ctx context.Context, database *sql.DB, databaseName, tableName string) ([]string, error) {
|
|
rows, err := database.QueryContext(ctx, "SELECT ORDINAL_POSITION, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COALESCE(COLUMN_DEFAULT, ''), EXTRA, COLUMN_KEY FROM information_schema.columns WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? ORDER BY ORDINAL_POSITION", databaseName, tableName)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
result := make([]string, 0)
|
|
for rows.Next() {
|
|
if err := contextError(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
var ordinal int
|
|
var name, columnType, nullable, defaultValue, extra, key string
|
|
if err := rows.Scan(&ordinal, &name, &columnType, &nullable, &defaultValue, &extra, &key); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, fmt.Sprintf("%d\x00%s\x00%s\x00%s\x00%s\x00%s\x00%s", ordinal, name, columnType, nullable, defaultValue, extra, key))
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func listPostgreSQLColumns(ctx context.Context, database *sql.DB, table postgreSQLTable) ([]string, error) {
|
|
rows, err := database.QueryContext(ctx, "SELECT ordinal_position, column_name, data_type, udt_name, is_nullable, COALESCE(column_default, ''), COALESCE(character_maximum_length, -1), COALESCE(numeric_precision, -1), COALESCE(numeric_scale, -1) FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2 ORDER BY ordinal_position", table.schema, table.name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
result := make([]string, 0)
|
|
for rows.Next() {
|
|
if err := contextError(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
var ordinal, characterLength, precision, scale int
|
|
var name, dataType, udtName, nullable, defaultValue string
|
|
if err := rows.Scan(&ordinal, &name, &dataType, &udtName, &nullable, &defaultValue, &characterLength, &precision, &scale); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, fmt.Sprintf("%d\x00%s\x00%s\x00%s\x00%s\x00%s\x00%d\x00%d\x00%d", ordinal, name, dataType, udtName, nullable, defaultValue, characterLength, precision, scale))
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return result, nil
|
|
}
|
|
func samePostgreSQLTables(source, target []postgreSQLTable) bool {
|
|
if len(source) != len(target) {
|
|
return false
|
|
}
|
|
seen := make(map[string]struct{}, len(source))
|
|
for _, table := range source {
|
|
seen[table.schema+"\x00"+table.name] = struct{}{}
|
|
}
|
|
for _, table := range target {
|
|
if _, ok := seen[table.schema+"\x00"+table.name]; !ok {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func sameStringSet(source, target []string) bool {
|
|
if len(source) != len(target) {
|
|
return false
|
|
}
|
|
seen := make(map[string]struct{}, len(source))
|
|
for _, value := range source {
|
|
seen[value] = struct{}{}
|
|
}
|
|
for _, value := range target {
|
|
if _, ok := seen[value]; !ok {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
type postgreSQLTable struct {
|
|
schema string
|
|
name string
|
|
}
|
|
|
|
func listPostgreSQLTables(ctx context.Context, database *sql.DB) ([]postgreSQLTable, error) {
|
|
rows, err := database.QueryContext(ctx, "SELECT table_schema, table_name FROM information_schema.tables WHERE table_type = 'BASE TABLE' AND table_schema NOT IN ('pg_catalog', 'information_schema') ORDER BY table_schema, table_name")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
tables := make([]postgreSQLTable, 0)
|
|
for rows.Next() {
|
|
var table postgreSQLTable
|
|
if err := rows.Scan(&table.schema, &table.name); err != nil {
|
|
return nil, err
|
|
}
|
|
tables = append(tables, table)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return tables, nil
|
|
}
|
|
|
|
func isDatabaseAlreadyExistsError(err error, databaseType config.DatabaseType) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
switch databaseType {
|
|
case config.DatabaseTypeMySQL:
|
|
var mysqlErr *mysql.MySQLError
|
|
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1007
|
|
case config.DatabaseTypePostgreSQL:
|
|
var postgresErr *pq.Error
|
|
return errors.As(err, &postgresErr) && string(postgresErr.Code) == "42P04"
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (c *SQLCopier) markCreated(source, target config.DatabaseConfig) {
|
|
if c == nil {
|
|
return
|
|
}
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
if c.created == nil {
|
|
c.created = make(map[string]struct{})
|
|
}
|
|
c.created[databaseTargetKey(source, target)] = struct{}{}
|
|
}
|
|
|
|
func (c *SQLCopier) hasCreated(source, target config.DatabaseConfig) bool {
|
|
if c == nil {
|
|
return false
|
|
}
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
_, ok := c.created[databaseTargetKey(source, target)]
|
|
return ok
|
|
}
|
|
|
|
func (c *SQLCopier) forgetCreated(source, target config.DatabaseConfig) {
|
|
if c == nil {
|
|
return
|
|
}
|
|
c.mu.Lock()
|
|
delete(c.created, databaseTargetKey(source, target))
|
|
c.mu.Unlock()
|
|
}
|
|
|
|
func (c *SQLCopier) Complete(source, target config.DatabaseConfig) {
|
|
c.forgetCreated(source, target)
|
|
}
|
|
|
|
func (c *SQLCopier) Cleanup(ctx context.Context, source, target config.DatabaseConfig) error {
|
|
if !c.hasCreated(source, target) {
|
|
return nil
|
|
}
|
|
if c.open == nil {
|
|
return ErrSQLConnectionUnavailable
|
|
}
|
|
adminDB, err := c.open(ctx, source, "")
|
|
if err != nil {
|
|
return ErrSQLConnectionUnavailable
|
|
}
|
|
defer adminDB.Close()
|
|
var query string
|
|
switch target.Type {
|
|
case config.DatabaseTypeMySQL:
|
|
query = "DROP DATABASE " + quoteMySQLIdentifier(target.Name)
|
|
case config.DatabaseTypePostgreSQL:
|
|
query = "DROP DATABASE " + quotePostgreSQLIdentifier(target.Name)
|
|
default:
|
|
return ErrUnsupportedDatabase
|
|
}
|
|
if _, err := adminDB.ExecContext(ctx, query); err != nil {
|
|
return fmt.Errorf("cleanup structured migration target: %w", err)
|
|
}
|
|
c.forgetCreated(source, target)
|
|
return nil
|
|
}
|
|
|
|
func countRows(ctx context.Context, database *sql.DB, table string) (int64, error) {
|
|
var count int64
|
|
if err := database.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+table).Scan(&count); err != nil {
|
|
return 0, err
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
func databaseTargetKey(source, target config.DatabaseConfig) string {
|
|
return strings.Join([]string{
|
|
string(target.Type),
|
|
target.Host,
|
|
strconv.Itoa(int(target.Port)),
|
|
target.Username,
|
|
target.Name,
|
|
source.Name,
|
|
}, "\x00")
|
|
}
|
|
|
|
func quoteMySQLIdentifier(value string) string {
|
|
return "`" + strings.ReplaceAll(value, "`", "``") + "`"
|
|
}
|
|
|
|
func quotePostgreSQLIdentifier(value string) string {
|
|
return "\"" + strings.ReplaceAll(value, "\"", "\"\"") + "\""
|
|
}
|
|
|
|
func NewRuntimeSQLConnectionFactory(root string) (SQLConnectionFactory, error) {
|
|
vault, err := credential.NewVault(root)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return func(ctx context.Context, database config.DatabaseConfig, databaseName string) (*sql.DB, error) {
|
|
password, err := config.OpenStoredSecret(vault, database.Password)
|
|
if err != nil {
|
|
return nil, ErrSQLConnectionUnavailable
|
|
}
|
|
driverName, dsn, err := runtimeSQLDSN(database, password, databaseName)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
connection, err := sql.Open(driverName, dsn)
|
|
if err != nil {
|
|
return nil, ErrSQLConnectionUnavailable
|
|
}
|
|
if err := connection.PingContext(ctx); err != nil {
|
|
_ = connection.Close()
|
|
return nil, ErrSQLConnectionUnavailable
|
|
}
|
|
return connection, nil
|
|
}, nil
|
|
}
|
|
|
|
func runtimeSQLDSN(database config.DatabaseConfig, password, databaseName string) (string, string, error) {
|
|
if database.Host == "" || database.Port == 0 || database.Username == "" || password == "" {
|
|
return "", "", ErrSQLConnectionUnavailable
|
|
}
|
|
address := net.JoinHostPort(database.Host, strconv.Itoa(int(database.Port)))
|
|
switch database.Type {
|
|
case config.DatabaseTypeMySQL:
|
|
settings := mysql.Config{
|
|
User: database.Username,
|
|
Passwd: password,
|
|
Net: "tcp",
|
|
Addr: address,
|
|
DBName: databaseName,
|
|
ParseTime: true,
|
|
Params: map[string]string{"charset": "utf8mb4"},
|
|
}
|
|
return "mysql", settings.FormatDSN(), nil
|
|
case config.DatabaseTypePostgreSQL:
|
|
maintenanceDatabase := databaseName
|
|
if maintenanceDatabase == "" {
|
|
maintenanceDatabase = "postgres"
|
|
}
|
|
databaseURL := url.URL{
|
|
Scheme: "postgres",
|
|
User: url.UserPassword(database.Username, password),
|
|
Host: address,
|
|
Path: "/" + maintenanceDatabase,
|
|
RawQuery: "sslmode=prefer",
|
|
}
|
|
return "postgres", databaseURL.String(), nil
|
|
default:
|
|
return "", "", ErrUnsupportedDatabase
|
|
}
|
|
}
|
|
|
|
var structuredDatabaseNamePattern = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_-]{0,62}$`)
|
|
var mysqlOptionIdentifierPattern = regexp.MustCompile(`^[A-Za-z0-9_]+$`)
|
|
|
|
func validStructuredDatabaseName(value string) bool {
|
|
return structuredDatabaseNamePattern.MatchString(strings.TrimSpace(value))
|
|
}
|