go-migrate/cmd/go_migrate/main.go

package main

import (
	"context"
	"sync"
	"time"

	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper"
	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/etl/extractors"
	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/etl/loaders"
	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/etl/table_analyzers"
	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/etl/transformers"
	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/models"
	log "github.com/sirupsen/logrus"
	"golang.org/x/sync/errgroup"
)

func main() {
	configureLog()

	migrationConfig, err := config.ReadMigrationConfig()
	if err != nil {
		log.Fatalf("error leyendo configuracion: %v", err)
	}

	log.Debugf("Config: %+v", migrationConfig)

	startTime := time.Now()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	log.Info("=== Starting migration ===")

	var wgConnect errgroup.Group
	var sourceDb, targetDb dbwrapper.DbWrapper

	wgConnect.Go(func() error {
		var err error
		sourceDb, err = connectWithTimeout(ctx, migrationConfig.SourceDbType, config.App.SourceDbUrl, 20*time.Second)
		if err != nil {
			return err
		}

		return nil
	})

	wgConnect.Go(func() error {
		var err error
		targetDb, err = connectWithTimeout(ctx, migrationConfig.TargetDbType, config.App.TargetDbUrl, 20*time.Second)
		if err != nil {
			return err
		}

		return nil
	})

	if err := wgConnect.Wait(); err != nil {
		log.Error("Connection error: ", err)
		return
	}
	defer sourceDb.Close()
	defer targetDb.Close()

	results := processMigrationJobs(ctx, sourceDb, targetDb, migrationConfig.Jobs, migrationConfig.MaxParallelWorkers)

	log.Info("=== RESUMEN DE MIGRACIÓN ===")
	var totalProcessed, totalErrors int64

	for _, res := range results {
		status := "OK"
		if res.Error != nil {
			status = "FAILED"
			log.Infof("[%s] Status: %s | Read: %d | Loaded: %d | Errors: %d | Time: %v | Error: %v", res.JobName, status, res.RowsRead, res.RowsLoaded, res.RowsFailed, res.Duration, res.Error)
		} else {
			log.Infof("[%s] Status: %s | Read: %d | Loaded: %d | Errors: %d | Time: %v", res.JobName, status, res.RowsRead, res.RowsLoaded, res.RowsFailed, res.Duration)
		}

		totalProcessed += res.RowsLoaded
		if res.Error != nil {
			totalErrors++
		}
	}

	log.Infof("Migración terminada. Tablas: %d, Errores: %d, Filas totales: %d", len(results), totalErrors, totalProcessed)

	totalDuration := time.Since(startTime)
	log.Infof("=== Migration completed successfully! ===")
	log.Infof("Total migration time: %v", totalDuration)
}

func processMigrationJobs(
	ctx context.Context,
	sourceDb dbwrapper.DbWrapper,
	targetDb dbwrapper.DbWrapper,
	jobs []config.Job,
	maxParallelWorkers int,
) []models.JobResult {
	if len(jobs) == 0 {
		log.Info("No migration jobs configured")
		return []models.JobResult{}
	}

	if maxParallelWorkers <= 0 {
		maxParallelWorkers = 1
	}

	if maxParallelWorkers > len(jobs) {
		maxParallelWorkers = len(jobs)
	}

	log.Infof("Starting migration with %d parallel worker(s)", maxParallelWorkers)

	chJobResults := make(chan models.JobResult, len(jobs))
	chJobs := make(chan config.Job, len(jobs))
	var wgJobs sync.WaitGroup

	sourceTableAnalyzer := table_analyzers.NewMssqlTableAnalyzer(sourceDb)
	targetTableAnalyzer := table_analyzers.NewPostgresTableAnalyzer(targetDb)
	extractor := extractors.NewMssqlExtractor(sourceDb)
	transformer := transformers.NewMssqlTransformer()
	loader := loaders.NewGenericLoader(targetDb)

	for i := range maxParallelWorkers {
		wgJobs.Go(func() {
			for job := range chJobs {
				log.Infof("[worker %d] >>> Processing job: %s.%s <<<", i, job.SourceTable.Schema, job.SourceTable.Table)
				res := processMigrationJob(
					ctx,
					targetDb,
					sourceTableAnalyzer,
					targetTableAnalyzer,
					extractor,
					transformer,
					loader,
					job,
				)

				chJobResults <- res
			}
		})
	}

	for _, job := range jobs {
		chJobs <- job
	}
	close(chJobs)

	go func() {
		wgJobs.Wait()
		close(chJobResults)
	}()

	var finalResults []models.JobResult
	for res := range chJobResults {
		finalResults = append(finalResults, res)
	}

	return finalResults
}

func connectWithTimeout(ctx context.Context, dbType string, dbUrl string, timeout time.Duration) (dbwrapper.DbWrapper, error) {
	localCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	sourceDb, err := dbwrapper.New(dbType)
	if err != nil {
		return nil, err
	}

	if err = sourceDb.Connect(localCtx, dbUrl); err != nil {
		return nil, err
	}

	return sourceDb, nil
}