157 lines
3.7 KiB
Go
157 lines
3.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
|
|
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
log "github.com/sirupsen/logrus"
|
|
)
|
|
|
|
func main() {
|
|
log.SetFormatter(&log.TextFormatter{
|
|
FullTimestamp: true,
|
|
TimestampFormat: time.StampMilli,
|
|
})
|
|
log.SetLevel(log.DebugLevel)
|
|
|
|
log.Info("Starting migration...")
|
|
|
|
ctxSource, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
|
|
sourcePool, err := db.Connect(ctxSource, config.App.FromDb.Url)
|
|
defer db.Close(sourcePool)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
log.Info("Successfully connected to from_db")
|
|
|
|
ctxTarget, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
|
|
targetPool, err := db.Connect(ctxTarget, config.App.ToDb.Url)
|
|
defer db.Close(targetPool)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
log.Info("Successfully connected to to_db")
|
|
|
|
schema := "test"
|
|
table := "migration_test"
|
|
colNames := []string{"id", "nombre_producto", "descripcion", "stock", "precio", "es_activo", "fecha_creacion", "ultima_actualizacion", "configuracion_json", "etiquetas", "binario_test", "ip_servidor", "rango_prueba"}
|
|
|
|
rowValues, err := extractData(ctxSource, sourcePool, schema, table, colNames, 2)
|
|
|
|
if err != nil {
|
|
log.Fatal("Unexpected error when extracting data", err)
|
|
}
|
|
|
|
for index, row := range rowValues {
|
|
log.Debugf("Values for row %d", index+1)
|
|
for i, v := range row {
|
|
log.Debugf("%s: %v", colNames[i], v)
|
|
}
|
|
}
|
|
|
|
insertedRows, err := insertData(ctxTarget, targetPool, schema, table, colNames, rowValues)
|
|
if err != nil {
|
|
log.Fatal("Unexpected error when inserting rows: ", err)
|
|
}
|
|
|
|
log.Infof("Inserted rows: %d", insertedRows)
|
|
|
|
log.Info("Migration completed successfully!")
|
|
}
|
|
|
|
func buildExtractSqlSentence(schema, table string, colNames []string) string {
|
|
var sbColumns strings.Builder
|
|
|
|
for i, col := range colNames {
|
|
sbColumns.WriteString(`"`)
|
|
sbColumns.WriteString(col)
|
|
sbColumns.WriteString(`"`)
|
|
if i < len(colNames)-1 {
|
|
sbColumns.WriteString(", ")
|
|
}
|
|
}
|
|
|
|
return fmt.Sprintf(`SELECT %s FROM "%s"."%s" LIMIT $1`, sbColumns.String(), schema, table)
|
|
}
|
|
|
|
func extractData(ctx context.Context, sourcePool *pgxpool.Pool, schema string, table string, colNames []string, limit int) ([][]any, error) {
|
|
if len(colNames) == 0 {
|
|
return nil, errors.New("Empty column names received")
|
|
}
|
|
|
|
sql := buildExtractSqlSentence(schema, table, colNames)
|
|
|
|
log.Debug("Executing query: ", sql)
|
|
|
|
rows, err := sourcePool.Query(ctx, sql, limit)
|
|
if err != nil {
|
|
if !errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, fmt.Errorf("Unexpected error: %w", err)
|
|
}
|
|
|
|
log.Warn("Unexpected error", err)
|
|
return [][]any{}, nil
|
|
}
|
|
defer rows.Close()
|
|
|
|
cols := rows.FieldDescriptions()
|
|
for _, c := range cols {
|
|
log.Debugf("Column: %+v", c)
|
|
}
|
|
|
|
rowValues := make([][]any, 0, limit)
|
|
|
|
for rows.Next() {
|
|
values := make([]any, len(cols))
|
|
valuePtrs := make([]any, len(cols))
|
|
|
|
for i := range values {
|
|
valuePtrs[i] = &values[i]
|
|
}
|
|
|
|
rows.Scan(valuePtrs...)
|
|
|
|
rowValues = append(rowValues, values)
|
|
}
|
|
|
|
return rowValues, nil
|
|
}
|
|
|
|
func insertData(ctx context.Context, targetPool *pgxpool.Pool, schema string, table string, colNames []string, rowValues [][]any) (int64, error) {
|
|
identifier := pgx.Identifier{schema, table}
|
|
|
|
count, err := targetPool.CopyFrom(
|
|
ctx,
|
|
identifier,
|
|
colNames,
|
|
pgx.CopyFromRows(rowValues),
|
|
)
|
|
|
|
if err != nil {
|
|
return 0, fmt.Errorf("error en CopyFrom: %w", err)
|
|
}
|
|
|
|
return count, nil
|
|
}
|
|
|
|
func Map[T any, V any](input []T, mapper func(T) V) []V {
|
|
result := make([]V, len(input))
|
|
|
|
for i, v := range input {
|
|
result[i] = mapper(v)
|
|
}
|
|
|
|
return result
|
|
}
|