feat: refactor data extraction and loading functions for improved context handling

cmd/go_migrate/loader.go

@@ -1,12 +1,15 @@
 package main
 
 import (
+	"context"
 	"fmt"
 
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
 	log "github.com/sirupsen/logrus"
 )
 
-func fakeLoader(job MigrationJob, columns []ColumnType, in <-chan []UnknownRowValues) {
+func fakeLoader(job MigrationJob, columns []ColumnType, in <-chan [][]any) {
 	for rows := range in {
 		log.Debugf("Chunk received, loading data into...")
 
@@ -17,3 +20,35 @@ func fakeLoader(job MigrationJob, columns []ColumnType, in <-chan []UnknownRowVa
 		}
 	}
 }
+
+func loadRowsPostgres(ctx context.Context, job MigrationJob, columns []ColumnType, db *pgxpool.Pool, in <-chan []UnknownRowValues) error {
+	for rows := range in {
+		identifier := pgx.Identifier{job.Schema, job.Table}
+		colNames := Map(columns, func(col ColumnType) string {
+			return col.name
+		})
+
+		_, err := db.CopyFrom(
+			ctx,
+			identifier,
+			colNames,
+			pgx.CopyFromRows(rows),
+		)
+
+		if err != nil {
+			return err
+		}
+	}
+
+	return 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
+}