Repository: supabase/cli
Pull request #2504 (2024-07-09 12:36:45)

feat: expose migration tool as public package
## What kind of change does this PR introduce?

feature

## What is the new behavior?

Allows 3rd party to run schema migrations the same way as supabase cli.
- Remote migrations are tracked in `supabase_migrations.schema_migrations` table
- Local migrations are tracked by a directory of `<timestampe>_name.sql` files
- Both remote and local sources can be read into a `MigrationFile` object

Exposes the following methods for remote migrations
- `CreateMigrationTable`
- `ReadMigrationTable`
- `ListRemoteMigrations`

For local migrations
- `ListLocalMigrations`
- `FindPendingMigrations`
- `ApplyMigrations`

And seeding
- `SeedData`
- `SeedGlobals`

## Additional context

Also exposes `pgtest` library for mocking postgres tcp connection.


coveralls: ## Pull Request Test Coverage Report for [Build 9857171917](https://coveralls.io/builds/68538293)


### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **5** unchanged lines in **1** file lost coverage.
* Overall coverage decreased (**-0.03%**) to **61.058%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [internal/gen/keys/keys.go](https://coveralls.io/builds/68538293/source?filename=internal%2Fgen%2Fkeys%2Fkeys.go#L101) | 5 | 11.76% |
<!-- | **Total:** | **5** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/68538293/badge)](https://coveralls.io/builds/68538293) |
| :-- | --: |
| Change from base [Build 9857104926](https://coveralls.io/builds/68538185): |  -0.03% |
| Covered Lines: | 6731 |
| Relevant Lines: | 11024 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/.golangci.yml b/.golangci.yml
index e6999619..9d518983 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -16,8 +16,8 @@ linters:
     - staticcheck
     - ineffassign
     - unused
-issues:
-  exclude-rules:
-    - linters:
-        - stylecheck
-      text: "ST1003:"
+linters-settings:
+  stylecheck:
+    checks: ["all", "-ST1003"]
+  dupl:
+    threshold: 200
diff --git a/internal/db/diff/diff.go b/internal/db/diff/diff.go
index 74b69080..a36180f9 100644
--- a/internal/db/diff/diff.go
+++ b/internal/db/diff/diff.go
@@ -25,9 +25,8 @@ import (
 	"github.com/supabase/cli/internal/db/start"
 	"github.com/supabase/cli/internal/gen/keys"
 	"github.com/supabase/cli/internal/migration/apply"
-	"github.com/supabase/cli/internal/migration/list"
-	"github.com/supabase/cli/internal/migration/repair"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
 	"github.com/supabase/cli/pkg/parser"
 )
 
@@ -158,7 +157,7 @@ func ConnectShadowDatabase(ctx context.Context, timeout time.Duration, options .
 }
 
 func MigrateShadowDatabase(ctx context.Context, container string, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {
-	migrations, err := list.LoadLocalMigrations(fsys)
+	migrations, err := migration.ListLocalMigrations(utils.MigrationsDir, afero.NewIOFS(fsys))
 	if err != nil {
 		return err
 	}
@@ -186,17 +185,7 @@ func migrateBaseDatabase(ctx context.Context, container string, fsys afero.Fs, o
 	if err := start.SetupDatabase(ctx, conn, container[:12], os.Stderr, fsys); err != nil {
 		return err
 	}
-	for _, path := range migrations {
-		fmt.Fprintln(os.Stderr, "Applying schema "+utils.Bold(path)+"...")
-		migration, err := repair.NewMigrationFromFile(path, fsys)
-		if err != nil {
-			return err
-		}
-		if err := migration.ExecBatch(ctx, conn); err != nil {
-			return err
-		}
-	}
-	return nil
+	return migration.SeedGlobals(ctx, migrations, conn, afero.NewIOFS(fsys))
 }
 
 func DiffDatabase(ctx context.Context, schema []string, config pgconn.Config, w io.Writer, fsys afero.Fs, differ func(context.Context, string, string, []string) (string, error), options ...func(*pgx.ConnConfig)) (string, error) {
diff --git a/internal/db/pull/pull.go b/internal/db/pull/pull.go
index e0e88656..c82761c5 100644
--- a/internal/db/pull/pull.go
+++ b/internal/db/pull/pull.go
@@ -19,6 +19,7 @@ import (
 	"github.com/supabase/cli/internal/migration/new"
 	"github.com/supabase/cli/internal/migration/repair"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
 )
 
 var (
@@ -121,7 +122,7 @@ func diffRemoteSchema(p utils.Program, ctx context.Context, schema []string, pat
 }
 
 func assertRemoteInSync(ctx context.Context, conn *pgx.Conn, fsys afero.Fs) error {
-	remoteMigrations, err := list.LoadRemoteMigrations(ctx, conn)
+	remoteMigrations, err := migration.ListRemoteMigrations(ctx, conn)
 	if err != nil {
 		return err
 	}
diff --git a/internal/db/push/push.go b/internal/db/push/push.go
index 8cdee030..db70a989 100644
--- a/internal/db/push/push.go
+++ b/internal/db/push/push.go
@@ -3,7 +3,6 @@ package push
 import (
 	"context"
 	"fmt"
-	"io"
 	"os"
 	"strings"
 
@@ -52,7 +51,7 @@ func Run(ctx context.Context, dryRun, ignoreVersionMismatch bool, includeRoles,
 			return errors.New(context.Canceled)
 		}
 		if includeRoles {
-			if err := CreateCustomRoles(ctx, conn, os.Stderr, fsys); err != nil {
+			if err := apply.CreateCustomRoles(ctx, conn, fsys); err != nil {
 				return err
 			}
 		}
@@ -68,14 +67,3 @@ func Run(ctx context.Context, dryRun, ignoreVersionMismatch bool, includeRoles,
 	fmt.Println("Finished " + utils.Aqua("supabase db push") + ".")
 	return nil
 }
-
-func CreateCustomRoles(ctx context.Context, conn *pgx.Conn, w io.Writer, fsys afero.Fs) error {
-	roles, err := fsys.Open(utils.CustomRolesPath)
-	if errors.Is(err, os.ErrNotExist) {
-		return nil
-	} else if err != nil {
-		return errors.Errorf("failed to load custom roles: %w", err)
-	}
-	fmt.Fprintln(w, "Creating custom roles "+utils.Bold(utils.CustomRolesPath)+"...")
-	return apply.BatchExecDDL(ctx, conn, roles)
-}
diff --git a/internal/db/remote/commit/commit.go b/internal/db/remote/commit/commit.go
index 718b2306..1f793f4e 100644
--- a/internal/db/remote/commit/commit.go
+++ b/internal/db/remote/commit/commit.go
@@ -13,9 +13,9 @@ import (
 	"github.com/supabase/cli/internal/db/diff"
 	"github.com/supabase/cli/internal/db/dump"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/migration/list"
 	"github.com/supabase/cli/internal/migration/repair"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
 )
 
 func Run(ctx context.Context, schema []string, config pgconn.Config, fsys afero.Fs) error {
@@ -65,7 +65,7 @@ func run(p utils.Program, ctx context.Context, schema []string, config pgconn.Co
 func fetchRemote(p utils.Program, ctx context.Context, schema []string, timestamp string, config pgconn.Config, fsys afero.Fs) error {
 	path := filepath.Join(utils.MigrationsDir, timestamp+"_remote_commit.sql")
 	// Special case if this is the first migration
-	if migrations, err := list.LoadLocalMigrations(fsys); err != nil {
+	if migrations, err := migration.ListLocalMigrations(utils.MigrationsDir, afero.NewIOFS(fsys)); err != nil {
 		return err
 	} else if len(migrations) == 0 {
 		p.Send(utils.StatusMsg("Committing initial migration on remote database..."))
@@ -85,11 +85,11 @@ func fetchRemote(p utils.Program, ctx context.Context, schema []string, timestam
 }
 
 func assertRemoteInSync(ctx context.Context, conn *pgx.Conn, fsys afero.Fs) error {
-	remoteMigrations, err := list.LoadRemoteMigrations(ctx, conn)
+	remoteMigrations, err := migration.ListRemoteMigrations(ctx, conn)
 	if err != nil {
 		return err
 	}
-	localMigrations, err := list.LoadLocalMigrations(fsys)
+	localMigrations, err := migration.ListLocalMigrations(utils.MigrationsDir, afero.NewIOFS(fsys))
 	if err != nil {
 		return err
 	}
@@ -104,7 +104,7 @@ func assertRemoteInSync(ctx context.Context, conn *pgx.Conn, fsys afero.Fs) erro
 
 	for i, remoteTimestamp := range remoteMigrations {
 		// LoadLocalMigrations guarantees we always have a match
-		localTimestamp := utils.MigrateFilePattern.FindStringSubmatch(localMigrations[i])[1]
+		localTimestamp := migration.MigrateFilePattern.FindStringSubmatch(localMigrations[i])[1]
 		if localTimestamp != remoteTimestamp {
 			return conflictErr
 		}
diff --git a/internal/db/reset/reset.go b/internal/db/reset/reset.go
index 7ea4c4e7..08fe208f 100644
--- a/internal/db/reset/reset.go
+++ b/internal/db/reset/reset.go
@@ -22,6 +22,7 @@ import (
 	"github.com/supabase/cli/internal/migration/apply"
 	"github.com/supabase/cli/internal/migration/repair"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
 	"github.com/supabase/cli/pkg/pgxv5"
 )
 
@@ -152,11 +153,7 @@ func initDatabase(ctx context.Context, options ...func(*pgx.ConnConfig)) error {
 		return err
 	}
 	defer conn.Close(context.Background())
-	sql := utils.InitialSchemaPg14Sql
-	if utils.Config.Db.MajorVersion == 13 {
-		sql = utils.InitialSchemaPg13Sql
-	}
-	return apply.BatchExecDDL(ctx, conn, strings.NewReader(sql))
+	return start.InitSchema14(ctx, conn)
 }
 
 // Recreate postgres database by connecting to template1
@@ -170,8 +167,8 @@ func recreateDatabase(ctx context.Context, options ...func(*pgx.ConnConfig)) err
 		return err
 	}
 	// We are not dropping roles here because they are cluster level entities. Use stop && start instead.
-	sql := repair.MigrationFile{
-		Lines: []string{
+	sql := migration.MigrationFile{
+		Statements: []string{
 			"DROP DATABASE IF EXISTS postgres WITH (FORCE)",
 			"CREATE DATABASE postgres WITH OWNER postgres",
 		},
@@ -238,15 +235,15 @@ func resetRemote(ctx context.Context, version string, config pgconn.Config, fsys
 		return err
 	}
 	// Drop all user defined schemas
-	migration := repair.MigrationFile{}
+	migration := migration.MigrationFile{}
 	for _, schema := range userSchemas {
 		sql := fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE", schema)
-		migration.Lines = append(migration.Lines, sql)
+		migration.Statements = append(migration.Statements, sql)
 	}
 	// If an extension uses a schema it doesn't create, dropping the schema will cascade to also
 	// drop the extension. But if an extension creates its own schema, dropping the schema will
 	// throw an error. Hence, we drop the extension instead so it cascades to its own schema.
-	migration.Lines = append(migration.Lines, dropObjects)
+	migration.Statements = append(migration.Statements, dropObjects)
 	if err := migration.ExecBatch(ctx, conn); err != nil {
 		return err
 	}
diff --git a/internal/db/start/start.go b/internal/db/start/start.go
index 46ea5a9d..14c2e9a1 100644
--- a/internal/db/start/start.go
+++ b/internal/db/start/start.go
@@ -19,10 +19,10 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/afero"
-	"github.com/supabase/cli/internal/db/push"
 	"github.com/supabase/cli/internal/migration/apply"
 	"github.com/supabase/cli/internal/status"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
 )
 
 var (
@@ -197,20 +197,26 @@ func initCurrentBranch(fsys afero.Fs) error {
 func initSchema(ctx context.Context, conn *pgx.Conn, host string, w io.Writer) error {
 	fmt.Fprintln(w, "Setting up initial schema...")
 	if utils.Config.Db.MajorVersion <= 14 {
-		return initSchema14(ctx, conn)
+		if file, err := migration.NewMigrationFromReader(strings.NewReader(utils.GlobalsSql)); err != nil {
+			return err
+		} else if err := file.ExecBatch(ctx, conn); err != nil {
+			return err
+		}
+		return InitSchema14(ctx, conn)
 	}
 	return initSchema15(ctx, host)
 }
 
-func initSchema14(ctx context.Context, conn *pgx.Conn) error {
-	if err := apply.BatchExecDDL(ctx, conn, strings.NewReader(utils.GlobalsSql)); err != nil {
-		return err
-	}
+func InitSchema14(ctx context.Context, conn *pgx.Conn) error {
 	sql := utils.InitialSchemaPg14Sql
 	if utils.Config.Db.MajorVersion == 13 {
 		sql = utils.InitialSchemaPg13Sql
 	}
-	return apply.BatchExecDDL(ctx, conn, strings.NewReader(sql))
+	file, err := migration.NewMigrationFromReader(strings.NewReader(sql))
+	if err != nil {
+		return err
+	}
+	return file.ExecBatch(ctx, conn)
 }
 
 func initRealtimeJob(host string) utils.DockerJob {
@@ -312,5 +318,5 @@ func SetupDatabase(ctx context.Context, conn *pgx.Conn, host string, w io.Writer
 	if err := initSchema(ctx, conn, host, w); err != nil {
 		return err
 	}
-	return push.CreateCustomRoles(ctx, conn, os.Stderr, fsys)
+	return apply.CreateCustomRoles(ctx, conn, fsys)
 }
diff --git a/internal/gen/keys/keys.go b/internal/gen/keys/keys.go
index 3a4cc284..81ed5298 100644
--- a/internal/gen/keys/keys.go
+++ b/internal/gen/keys/keys.go
@@ -7,33 +7,14 @@ import (
 	"fmt"
 	"os"
 	"strings"
-	"time"
 
 	"github.com/go-errors/errors"
 	"github.com/go-git/go-git/v5"
-	"github.com/golang-jwt/jwt/v5"
 	"github.com/spf13/afero"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/config"
 )
 
-type CustomClaims struct {
-	Ref  string `json:"ref"`
-	Role string `json:"role"`
-	jwt.RegisteredClaims
-}
-
-func NewJWTToken(ref, role string, expiry time.Time) *jwt.Token {
-	claims := CustomClaims{
-		ref,
-		role,
-		jwt.RegisteredClaims{
-			ExpiresAt: jwt.NewNumericDate(expiry),
-			Issuer:    "supabase",
-		},
-	}
-	return jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
-}
-
 type CustomName struct {
 	DbHost         string `env:"db.host,default=NEXT_PUBLIC_SUPABASE_URL"`
 	DbPassword     string `env:"db.password,default=SUPABASE_DB_PASSWORD"`
@@ -75,15 +56,20 @@ func GenerateSecrets(ctx context.Context, projectRef, branch string, fsys afero.
 	hash := sha256.Sum256([]byte(key))
 	utils.Config.Db.Password = hex.EncodeToString(hash[:])
 	// Generate JWT tokens
-	expiry := time.Now().AddDate(10, 0, 0)
-	anonToken := NewJWTToken(projectRef, "anon", expiry)
-	utils.Config.Auth.AnonKey, err = anonToken.SignedString([]byte(utils.Config.Auth.JwtSecret))
-	if err != nil {
+	anonToken := config.CustomClaims{
+		Issuer: "supabase",
+		Ref:    projectRef,
+		Role:   "anon",
+	}.NewToken()
+	if utils.Config.Auth.AnonKey, err = anonToken.SignedString([]byte(utils.Config.Auth.JwtSecret)); err != nil {
 		return errors.Errorf("failed to sign anon key: %w", err)
 	}
-	serviceToken := NewJWTToken(projectRef, "service_role", expiry)
-	utils.Config.Auth.ServiceRoleKey, err = serviceToken.SignedString([]byte(utils.Config.Auth.JwtSecret))
-	if err != nil {
+	serviceToken := config.CustomClaims{
+		Issuer: "supabase",
+		Ref:    projectRef,
+		Role:   "service_role",
+	}.NewToken()
+	if utils.Config.Auth.ServiceRoleKey, err = serviceToken.SignedString([]byte(utils.Config.Auth.JwtSecret)); err != nil {
 		return errors.Errorf("failed to sign service_role key: %w", err)
 	}
 	return nil
diff --git a/internal/link/link.go b/internal/link/link.go
index 2b6e0ba4..9d033d47 100644
--- a/internal/link/link.go
+++ b/internal/link/link.go
@@ -15,13 +15,13 @@ import (
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/afero"
 	"github.com/spf13/viper"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/utils"
 	"github.com/supabase/cli/internal/utils/credentials"
 	"github.com/supabase/cli/internal/utils/flags"
 	"github.com/supabase/cli/internal/utils/tenant"
 	"github.com/supabase/cli/pkg/api"
 	cliConfig "github.com/supabase/cli/pkg/config"
+	"github.com/supabase/cli/pkg/migration"
 )
 
 var updatedConfig ConfigCopy
@@ -187,7 +187,7 @@ func linkDatabase(ctx context.Context, config pgconn.Config, options ...func(*pg
 	defer conn.Close(context.Background())
 	updatePostgresConfig(conn)
 	// If `schema_migrations` doesn't exist on the remote database, create it.
-	return history.CreateMigrationTable(ctx, conn)
+	return migration.CreateMigrationTable(ctx, conn)
 }
 
 func linkDatabaseVersion(ctx context.Context, projectRef string, fsys afero.Fs) error {
diff --git a/internal/migration/apply/apply.go b/internal/migration/apply/apply.go
index c9723ee1..9528578f 100644
--- a/internal/migration/apply/apply.go
+++ b/internal/migration/apply/apply.go
@@ -2,18 +2,15 @@ package apply
 
 import (
 	"context"
-	"fmt"
-	"io"
 	"os"
 	"path/filepath"
 
 	"github.com/go-errors/errors"
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/afero"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/migration/list"
-	"github.com/supabase/cli/internal/migration/repair"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
 )
 
 func MigrateAndSeed(ctx context.Context, version string, conn *pgx.Conn, fsys afero.Fs) error {
@@ -28,45 +25,25 @@ func MigrateAndSeed(ctx context.Context, version string, conn *pgx.Conn, fsys af
 }
 
 func SeedDatabase(ctx context.Context, conn *pgx.Conn, fsys afero.Fs) error {
-	seed, err := repair.NewMigrationFromFile(utils.SeedDataPath, fsys)
+	err := migration.SeedData(ctx, []string{utils.SeedDataPath}, conn, afero.NewIOFS(fsys))
 	if errors.Is(err, os.ErrNotExist) {
 		return nil
-	} else if err != nil {
-		return err
 	}
-	fmt.Fprintln(os.Stderr, "Seeding data "+utils.Bold(utils.SeedDataPath)+"...")
-	// Batch seed commands, safe to use statement cache
-	return seed.ExecBatchWithCache(ctx, conn)
+	return err
 }
 
 func MigrateUp(ctx context.Context, conn *pgx.Conn, pending []string, fsys afero.Fs) error {
-	if len(pending) > 0 {
-		if err := history.CreateMigrationTable(ctx, conn); err != nil {
-			return err
-		}
-	}
-	for _, filename := range pending {
-		if err := applyMigration(ctx, conn, filename, fsys); err != nil {
-			return err
-		}
+	var paths []string
+	for _, name := range pending {
+		paths = append(paths, filepath.Join(utils.MigrationsDir, name))
 	}
-	return nil
+	return migration.ApplyMigrations(ctx, paths, conn, afero.NewIOFS(fsys))
 }
 
-func applyMigration(ctx context.Context, conn *pgx.Conn, filename string, fsys afero.Fs) error {
-	fmt.Fprintln(os.Stderr, "Applying migration "+utils.Bold(filename)+"...")
-	path := filepath.Join(utils.MigrationsDir, filename)
-	migration, err := repair.NewMigrationFromFile(path, fsys)
-	if err != nil {
-		return err
-	}
-	return migration.ExecBatch(ctx, conn)
-}
-
-func BatchExecDDL(ctx context.Context, conn *pgx.Conn, sql io.Reader) error {
-	migration, err := repair.NewMigrationFromReader(sql)
-	if err != nil {
-		return err
+func CreateCustomRoles(ctx context.Context, conn *pgx.Conn, fsys afero.Fs) error {
+	err := migration.SeedGlobals(ctx, []string{utils.CustomRolesPath}, conn, afero.NewIOFS(fsys))
+	if errors.Is(err, os.ErrNotExist) {
+		return nil
 	}
-	return migration.ExecBatch(ctx, conn)
+	return err
 }
diff --git a/internal/migration/fetch/fetch.go b/internal/migration/fetch/fetch.go
index 18df66d7..cb78c70d 100644
--- a/internal/migration/fetch/fetch.go
+++ b/internal/migration/fetch/fetch.go
@@ -10,8 +10,8 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/afero"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
 )
 
 func Run(ctx context.Context, config pgconn.Config, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {
@@ -43,11 +43,11 @@ func Run(ctx context.Context, config pgconn.Config, fsys afero.Fs, options ...fu
 	return nil
 }
 
-func fetchMigrationHistory(ctx context.Context, config pgconn.Config, options ...func(*pgx.ConnConfig)) ([]history.SchemaMigration, error) {
+func fetchMigrationHistory(ctx context.Context, config pgconn.Config, options ...func(*pgx.ConnConfig)) ([]migration.MigrationFile, error) {
 	conn, err := utils.ConnectByConfig(ctx, config, options...)
 	if err != nil {
 		return nil, err
 	}
 	defer conn.Close(context.Background())
-	return history.ReadMigrationTable(ctx, conn)
+	return migration.ReadMigrationTable(ctx, conn)
 }
diff --git a/internal/migration/list/list.go b/internal/migration/list/list.go
index 08c0c65b..df25f223 100644
--- a/internal/migration/list/list.go
+++ b/internal/migration/list/list.go
@@ -4,24 +4,17 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"os"
-	"regexp"
 	"strconv"
 
 	"github.com/charmbracelet/glamour"
 	"github.com/go-errors/errors"
 	"github.com/jackc/pgconn"
-	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/afero"
 	"github.com/supabase/cli/internal/utils"
-	"github.com/supabase/cli/pkg/pgxv5"
+	"github.com/supabase/cli/pkg/migration"
 )
 
-const LIST_MIGRATION_VERSION = "SELECT version FROM supabase_migrations.schema_migrations ORDER BY version"
-
-var initSchemaPattern = regexp.MustCompile(`([0-9]{14})_init\.sql`)
-
 func Run(ctx context.Context, config pgconn.Config, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {
 	remoteVersions, err := loadRemoteVersions(ctx, config, options...)
 	if err != nil {
@@ -41,27 +34,7 @@ func loadRemoteVersions(ctx context.Context, config pgconn.Config, options ...fu
 		return nil, err
 	}
 	defer conn.Close(context.Background())
-	return LoadRemoteMigrations(ctx, conn)
-}
-
-func LoadRemoteMigrations(ctx context.Context, conn *pgx.Conn) ([]string, error) {
-	versions, err := listMigrationVersions(ctx, conn)
-	if err != nil {
-		var pgErr *pgconn.PgError
-		if errors.As(err, &pgErr) && pgErr.Code == pgerrcode.UndefinedTable {
-			// If migration history table is undefined, the remote project has no migrations
-			return nil, nil
-		}
-	}
-	return versions, err
-}
-
-func listMigrationVersions(ctx context.Context, conn *pgx.Conn) ([]string, error) {
-	rows, err := conn.Query(ctx, LIST_MIGRATION_VERSION)
-	if err != nil {
-		return nil, errors.Errorf("failed to query rows: %w", err)
-	}
-	return pgxv5.CollectStrings(rows)
+	return migration.ListRemoteMigrations(ctx, conn)
 }
 
 func makeTable(remoteMigrations, localMigrations []string) string {
@@ -115,56 +88,25 @@ func RenderTable(markdown string) error {
 }
 
 func LoadLocalVersions(fsys afero.Fs) ([]string, error) {
-	names, err := LoadLocalMigrations(fsys)
+	names, err := migration.ListLocalMigrations(utils.MigrationsDir, afero.NewIOFS(fsys))
 	if err != nil {
 		return nil, err
 	}
 	var versions []string
 	for _, filename := range names {
 		// LoadLocalMigrations guarantees we always have a match
-		version := utils.MigrateFilePattern.FindStringSubmatch(filename)[1]
+		version := migration.MigrateFilePattern.FindStringSubmatch(filename)[1]
 		versions = append(versions, version)
 	}
 	return versions, nil
 }
 
-func LoadLocalMigrations(fsys afero.Fs) ([]string, error) {
-	return LoadPartialMigrations("", fsys)
-}
-
 func LoadPartialMigrations(version string, fsys afero.Fs) ([]string, error) {
-	localMigrations, err := afero.ReadDir(fsys, utils.MigrationsDir)
-	if err != nil && !errors.Is(err, os.ErrNotExist) {
-		return nil, errors.Errorf("failed to read directory: %w", err)
-	}
-	var names []string
-	for i, migration := range localMigrations {
-		filename := migration.Name()
-		if i == 0 && shouldSkip(filename) {
-			fmt.Fprintln(os.Stderr, "Skipping migration "+utils.Bold(filename)+`... (replace "init" with a different file name to apply this migration)`)
-			continue
-		}
-		matches := utils.MigrateFilePattern.FindStringSubmatch(filename)
-		if len(matches) == 0 {
-			fmt.Fprintln(os.Stderr, "Skipping migration "+utils.Bold(filename)+`... (file name must match pattern "<timestamp>_name.sql")`)
-			continue
-		}
-		names = append(names, filename)
-		if matches[1] == version {
-			break
-		}
-	}
-	return names, nil
-}
-
-func shouldSkip(name string) bool {
-	// NOTE: To handle backward-compatibility. `<timestamp>_init.sql` as
-	// the first migration (prev versions of the CLI) is deprecated.
-	matches := initSchemaPattern.FindStringSubmatch(name)
-	if len(matches) == 2 {
-		if timestamp, err := strconv.ParseUint(matches[1], 10, 64); err == nil && timestamp < 20211209000000 {
-			return true
-		}
+	var filter []func(string) bool
+	if len(version) > 0 {
+		filter = append(filter, func(v string) bool {
+			return v <= version
+		})
 	}
-	return false
+	return migration.ListLocalMigrations(utils.MigrationsDir, afero.NewIOFS(fsys), filter...)
 }
diff --git a/internal/migration/repair/repair.go b/internal/migration/repair/repair.go
index 5b953064..c0670000 100644
--- a/internal/migration/repair/repair.go
+++ b/internal/migration/repair/repair.go
@@ -3,21 +3,17 @@ package repair
 import (
 	"context"
 	"fmt"
-	"io"
 	"os"
 	"path/filepath"
 	"strconv"
 
 	"github.com/go-errors/errors"
 	"github.com/jackc/pgconn"
-	"github.com/jackc/pgtype"
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/afero"
-	"github.com/spf13/viper"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/migration/list"
 	"github.com/supabase/cli/internal/utils"
-	"github.com/supabase/cli/pkg/parser"
+	"github.com/supabase/cli/pkg/migration"
 )
 
 const (
@@ -60,13 +56,13 @@ func Run(ctx context.Context, config pgconn.Config, version []string, status str
 }
 
 func UpdateMigrationTable(ctx context.Context, conn *pgx.Conn, version []string, status string, repairAll bool, fsys afero.Fs) error {
-	if err := history.CreateMigrationTable(ctx, conn); err != nil {
+	if err := migration.CreateMigrationTable(ctx, conn); err != nil {
 		return err
 	}
 	// Data statements don't mutate schemas, safe to use statement cache
 	batch := &pgx.Batch{}
 	if repairAll {
-		batch.Queue(history.TRUNCATE_VERSION_TABLE)
+		batch.Queue(migration.TRUNCATE_VERSION_TABLE)
 	}
 	switch status {
 	case Applied:
@@ -75,11 +71,11 @@ func UpdateMigrationTable(ctx context.Context, conn *pgx.Conn, version []string,
 			if err != nil {
 				return err
 			}
-			batch.Queue(history.INSERT_MIGRATION_VERSION, f.Version, f.Name, f.Lines)
+			batch.Queue(migration.INSERT_MIGRATION_VERSION, f.Version, f.Name, f.Statements)
 		}
 	case Reverted:
 		if !repairAll {
-			batch.Queue(history.DELETE_MIGRATION_VERSION, version)
+			batch.Queue(migration.DELETE_MIGRATION_VERSION, version)
 		}
 	}
 	if err := conn.SendBatch(ctx, batch).Close(); err != nil {
@@ -103,118 +99,10 @@ func GetMigrationFile(version string, fsys afero.Fs) (string, error) {
 	return matches[0], nil
 }
 
-type MigrationFile struct {
-	Lines   []string
-	Version string
-	Name    string
-}
-
-func NewMigrationFromVersion(version string, fsys afero.Fs) (*MigrationFile, error) {
+func NewMigrationFromVersion(version string, fsys afero.Fs) (*migration.MigrationFile, error) {
 	name, err := GetMigrationFile(version, fsys)
 	if err != nil {
 		return nil, err
 	}
-	return NewMigrationFromFile(name, fsys)
-}
-
-func NewMigrationFromFile(path string, fsys afero.Fs) (*MigrationFile, error) {
-	sql, err := fsys.Open(path)
-	if err != nil {
-		return nil, errors.Errorf("failed to open migration file: %w", err)
-	}
-	defer sql.Close()
-	// Unless explicitly specified, Use file length as max buffer size
-	if !viper.IsSet("SCANNER_BUFFER_SIZE") {
-		if fi, err := sql.Stat(); err == nil {
-			if size := int(fi.Size()); size > parser.MaxScannerCapacity {
-				parser.MaxScannerCapacity = size
-			}
-		}
-	}
-	file, err := NewMigrationFromReader(sql)
-	if err == nil {
-		// Parse version from file name
-		filename := filepath.Base(path)
-		matches := utils.MigrateFilePattern.FindStringSubmatch(filename)
-		if len(matches) > 2 {
-			file.Version = matches[1]
-			file.Name = matches[2]
-		}
-	}
-	return file, err
-}
-
-func NewMigrationFromReader(sql io.Reader) (*MigrationFile, error) {
-	lines, err := parser.SplitAndTrim(sql)
-	if err != nil {
-		return nil, err
-	}
-	return &MigrationFile{Lines: lines}, nil
-}
-
-func (m *MigrationFile) ExecBatch(ctx context.Context, conn *pgx.Conn) error {
-	// Batch migration commands, without using statement cache
-	batch := &pgconn.Batch{}
-	for _, line := range m.Lines {
-		batch.ExecParams(line, nil, nil, nil, nil)
-	}
-	// Insert into migration history
-	if len(m.Version) > 0 {
-		if err := m.insertVersionSQL(conn, batch); err != nil {
-			return err
-		}
-	}
-	// ExecBatch is implicitly transactional
-	if result, err := conn.PgConn().ExecBatch(ctx, batch).ReadAll(); err != nil {
-		// Defaults to printing the last statement on error
-		stat := history.INSERT_MIGRATION_VERSION
-		i := len(result)
-		if i < len(m.Lines) {
-			stat = m.Lines[i]
-		}
-		return errors.Errorf("%w\nAt statement %d: %s", err, i, stat)
-	}
-	return nil
-}
-
-func (m *MigrationFile) insertVersionSQL(conn *pgx.Conn, batch *pgconn.Batch) error {
-	value := pgtype.TextArray{}
-	if err := value.Set(m.Lines); err != nil {
-		return errors.Errorf("failed to set text array: %w", err)
-	}
-	ci := conn.ConnInfo()
-	var err error
-	var encoded []byte
-	var valueFormat int16
-	if conn.Config().PreferSimpleProtocol {
-		encoded, err = value.EncodeText(ci, encoded)
-		valueFormat = pgtype.TextFormatCode
-	} else {
-		encoded, err = value.EncodeBinary(ci, encoded)
-		valueFormat = pgtype.BinaryFormatCode
-	}
-	if err != nil {
-		return errors.Errorf("failed to encode binary: %w", err)
-	}
-	batch.ExecParams(
-		history.INSERT_MIGRATION_VERSION,
-		[][]byte{[]byte(m.Version), []byte(m.Name), encoded},
-		[]uint32{pgtype.TextOID, pgtype.TextOID, pgtype.TextArrayOID},
-		[]int16{pgtype.TextFormatCode, pgtype.TextFormatCode, valueFormat},
-		nil,
-	)
-	return nil
-}
-
-func (m *MigrationFile) ExecBatchWithCache(ctx context.Context, conn *pgx.Conn) error {
-	// Data statements don't mutate schemas, safe to use statement cache
-	batch := pgx.Batch{}
-	for _, line := range m.Lines {
-		batch.Queue(line)
-	}
-	// No need to track version here because there are no schema changes
-	if err := conn.SendBatch(ctx, &batch).Close(); err != nil {
-		return errors.Errorf("failed to send batch: %w", err)
-	}
-	return nil
+	return migration.NewMigrationFromFile(name, afero.NewIOFS(fsys))
 }
diff --git a/internal/migration/squash/squash.go b/internal/migration/squash/squash.go
index 8a955ed6..e7325f49 100644
--- a/internal/migration/squash/squash.go
+++ b/internal/migration/squash/squash.go
@@ -19,10 +19,10 @@ import (
 	"github.com/supabase/cli/internal/db/dump"
 	"github.com/supabase/cli/internal/db/start"
 	"github.com/supabase/cli/internal/migration/apply"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/migration/list"
 	"github.com/supabase/cli/internal/migration/repair"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
 )
 
 var ErrMissingVersion = errors.New("version not found")
@@ -179,7 +179,7 @@ func baselineMigrations(ctx context.Context, config pgconn.Config, version strin
 		return err
 	}
 	defer conn.Close(context.Background())
-	if err := history.CreateMigrationTable(ctx, conn); err != nil {
+	if err := migration.CreateMigrationTable(ctx, conn); err != nil {
 		return err
 	}
 	m, err := repair.NewMigrationFromVersion(version, fsys)
@@ -188,8 +188,8 @@ func baselineMigrations(ctx context.Context, config pgconn.Config, version strin
 	}
 	// Data statements don't mutate schemas, safe to use statement cache
 	batch := pgx.Batch{}
-	batch.Queue(history.DELETE_MIGRATION_BEFORE, m.Version)
-	batch.Queue(history.INSERT_MIGRATION_VERSION, m.Version, m.Name, m.Lines)
+	batch.Queue(migration.DELETE_MIGRATION_BEFORE, m.Version)
+	batch.Queue(migration.INSERT_MIGRATION_VERSION, m.Version, m.Name, m.Statements)
 	if err := conn.SendBatch(ctx, &batch).Close(); err != nil {
 		return errors.Errorf("failed to update migration history: %w", err)
 	}
diff --git a/internal/migration/up/up.go b/internal/migration/up/up.go
index f7a54a43..b6894f03 100644
--- a/internal/migration/up/up.go
+++ b/internal/migration/up/up.go
@@ -11,13 +11,8 @@ import (
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/afero"
 	"github.com/supabase/cli/internal/migration/apply"
-	"github.com/supabase/cli/internal/migration/list"
 	"github.com/supabase/cli/internal/utils"
-)
-
-var (
-	errMissingRemote = errors.New("Found local migration files to be inserted before the last migration on remote database.")
-	errMissingLocal  = errors.New("Remote migration versions not found in " + utils.MigrationsDir + " directory.")
+	"github.com/supabase/cli/pkg/migration"
 )
 
 func Run(ctx context.Context, includeAll bool, config pgconn.Config, fsys afero.Fs, options ...func(*pgx.ConnConfig)) error {
@@ -34,50 +29,25 @@ func Run(ctx context.Context, includeAll bool, config pgconn.Config, fsys afero.
 }
 
 func GetPendingMigrations(ctx context.Context, includeAll bool, conn *pgx.Conn, fsys afero.Fs) ([]string, error) {
-	remoteMigrations, err := list.LoadRemoteMigrations(ctx, conn)
+	remoteMigrations, err := migration.ListRemoteMigrations(ctx, conn)
 	if err != nil {
 		return nil, err
 	}
-	localMigrations, err := list.LoadLocalMigrations(fsys)
+	localMigrations, err := migration.ListLocalMigrations(utils.MigrationsDir, afero.NewIOFS(fsys))
 	if err != nil {
 		return nil, err
 	}
-	// Find unapplied local migrations older than the latest migration on
-	// remote, and remote migrations that are missing from local.
-	var unapplied, missing []string
-	i, j := 0, 0
-	for i < len(remoteMigrations) && j < len(localMigrations) {
-		remote := remoteMigrations[i]
-		filename := localMigrations[j]
-		// Check if migration has been applied before, LoadLocalMigrations guarantees a match
-		local := utils.MigrateFilePattern.FindStringSubmatch(filename)[1]
-		if remote == local {
-			j++
-			i++
-		} else if remote < local {
-			missing = append(missing, remote)
-			i++
-		} else {
-			// Include out-of-order local migrations
-			unapplied = append(unapplied, filename)
-			j++
+	diff, err := migration.FindPendingMigrations(localMigrations, remoteMigrations)
+	if errors.Is(err, migration.ErrMissingLocal) {
+		utils.CmdSuggestion = suggestRevertHistory(diff)
+	} else if errors.Is(err, migration.ErrMissingRemote) {
+		if includeAll {
+			pending := localMigrations[len(remoteMigrations)+len(diff):]
+			return append(diff, pending...), nil
 		}
+		utils.CmdSuggestion = suggestIgnoreFlag(diff)
 	}
-	// Ensure all remote versions exist on local
-	if j == len(localMigrations) {
-		missing = append(missing, remoteMigrations[i:]...)
-	}
-	if len(missing) > 0 {
-		utils.CmdSuggestion = suggestRevertHistory(missing)
-		return nil, errMissingLocal
-	}
-	// Enforce migrations are applied in chronological order by default
-	if !includeAll && len(unapplied) > 0 {
-		utils.CmdSuggestion = suggestIgnoreFlag(unapplied)
-		return nil, errMissingRemote
-	}
-	pending := localMigrations[len(remoteMigrations)+len(unapplied):]
-	return append(unapplied, pending...), nil
+	return diff, err
 }
 
 func suggestRevertHistory(versions []string) string {
diff --git a/internal/start/start.go b/internal/start/start.go
index 166d5e3b..c3e5c092 100644
--- a/internal/start/start.go
+++ b/internal/start/start.go
@@ -282,7 +282,8 @@ EOF
 		// Special case for GitLab pipeline
 		host := utils.Docker.DaemonHost()
 		if parsed, err := client.ParseHostURL(host); err == nil && parsed.Scheme == "tcp" {
-			env = append(env, "DOCKER_HOST="+host)
+			parsed.Host = "host.docker.internal"
+			env = append(env, "DOCKER_HOST="+parsed.String())
 		} else if parsed, err := client.ParseHostURL(client.DefaultDockerHost); err == nil {
 			if host != client.DefaultDockerHost {
 				fmt.Fprintln(os.Stderr, utils.Yellow("WARNING:"), "analytics requires mounting default docker socket:", parsed.Host)
diff --git a/internal/utils/connect.go b/internal/utils/connect.go
index b3c44ad4..9479d4e4 100644
--- a/internal/utils/connect.go
+++ b/internal/utils/connect.go
@@ -10,11 +10,11 @@ import (
 	"strings"
 	"time"
 
-	"github.com/go-errors/errors"
 	"github.com/jackc/pgconn"
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/viper"
 	"github.com/supabase/cli/internal/debug"
+	"github.com/supabase/cli/pkg/pgxv5"
 )
 
 func ToPostgresURL(config pgconn.Config) string {
@@ -116,24 +116,10 @@ func ConnectLocalPostgres(ctx context.Context, config pgconn.Config, options ...
 }
 
 func ConnectByUrl(ctx context.Context, url string, options ...func(*pgx.ConnConfig)) (*pgx.Conn, error) {
-	// Parse connection url
-	config, err := pgx.ParseConfig(url)
-	if err != nil {
-		return nil, errors.Errorf("failed to parse postgres url: %w", err)
-	}
-	// Apply config overrides
-	for _, op := range options {
-		op(config)
-	}
 	if viper.GetBool("DEBUG") {
-		debug.SetupPGX(config)
-	}
-	// Connect to database
-	conn, err := pgx.ConnectConfig(ctx, config)
-	if err != nil {
-		return nil, errors.Errorf("failed to connect to postgres: %w", err)
+		options = append(options, debug.SetupPGX)
 	}
-	return conn, nil
+	return pgxv5.Connect(ctx, url, options...)
 }
 
 func ConnectByConfigStream(ctx context.Context, config pgconn.Config, w io.Writer, options ...func(*pgx.ConnConfig)) (*pgx.Conn, error) {
diff --git a/pkg/migration/apply.go b/pkg/migration/apply.go
new file mode 100644
index 00000000..0311b43f
--- /dev/null
+++ b/pkg/migration/apply.go
@@ -0,0 +1,74 @@
+package migration
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+
+	"github.com/go-errors/errors"
+	"github.com/jackc/pgx/v4"
+)
+
+var (
+	ErrMissingRemote = errors.New("Found local migration files to be inserted before the last migration on remote database.")
+	ErrMissingLocal  = errors.New("Remote migration versions not found in local migrations directory.")
+)
+
+// Find unapplied local migrations older than the latest migration on
+// remote, and remote migrations that are missing from local.
+func FindPendingMigrations(localMigrations, remoteMigrations []string) ([]string, error) {
+	var unapplied, missing []string
+	i, j := 0, 0
+	for i < len(remoteMigrations) && j < len(localMigrations) {
+		remote := remoteMigrations[i]
+		filename := localMigrations[j]
+		// Check if migration has been applied before, LoadLocalMigrations guarantees a match
+		local := MigrateFilePattern.FindStringSubmatch(filename)[1]
+		if remote == local {
+			j++
+			i++
+		} else if remote < local {
+			missing = append(missing, remote)
+			i++
+		} else {
+			// Include out-of-order local migrations
+			unapplied = append(unapplied, filename)
+			j++
+		}
+	}
+	// Ensure all remote versions exist on local
+	if j == len(localMigrations) {
+		missing = append(missing, remoteMigrations[i:]...)
+	}
+	if len(missing) > 0 {
+		// utils.CmdSuggestion = suggestRevertHistory(missing)
+		return missing, errors.New(ErrMissingLocal)
+	}
+	// Enforce migrations are applied in chronological order by default
+	if len(unapplied) > 0 {
+		// utils.CmdSuggestion = suggestIgnoreFlag(unapplied)
+		return unapplied, errors.New(ErrMissingRemote)
+	}
+	pending := localMigrations[len(remoteMigrations):]
+	return pending, nil
+}
+
+func ApplyMigrations(ctx context.Context, pending []string, conn *pgx.Conn, fsys fs.FS) error {
+	if len(pending) > 0 {
+		if err := CreateMigrationTable(ctx, conn); err != nil {
+			return err
+		}
+	}
+	for _, path := range pending {
+		filename := filepath.Base(path)
+		fmt.Fprintf(os.Stderr, "Applying migration %s...\n", filename)
+		if migration, err := NewMigrationFromFile(path, fsys); err != nil {
+			return err
+		} else if err := migration.ExecBatch(ctx, conn); err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/pkg/migration/file.go b/pkg/migration/file.go
new file mode 100644
index 00000000..1f70ddba
--- /dev/null
+++ b/pkg/migration/file.go
@@ -0,0 +1,126 @@
+package migration
+
+import (
+	"context"
+	"io"
+	"io/fs"
+	"path/filepath"
+	"regexp"
+
+	"github.com/go-errors/errors"
+	"github.com/jackc/pgconn"
+	"github.com/jackc/pgtype"
+	"github.com/jackc/pgx/v4"
+	"github.com/spf13/viper"
+	"github.com/supabase/cli/pkg/parser"
+)
+
+var MigrateFilePattern = regexp.MustCompile(`^([0-9]+)_(.*)\.sql$`)
+
+type MigrationFile struct {
+	Version    string
+	Name       string
+	Statements []string
+}
+
+func NewMigrationFromFile(path string, fsys fs.FS) (*MigrationFile, error) {
+	sql, err := fsys.Open(path)
+	if err != nil {
+		return nil, errors.Errorf("failed to open migration file: %w", err)
+	}
+	defer sql.Close()
+	// Unless explicitly specified, Use file length as max buffer size
+	if !viper.IsSet("SCANNER_BUFFER_SIZE") {
+		if fi, err := sql.Stat(); err == nil {
+			if size := int(fi.Size()); size > parser.MaxScannerCapacity {
+				parser.MaxScannerCapacity = size
+			}
+		}
+	}
+	file, err := NewMigrationFromReader(sql)
+	if err == nil {
+		// Parse version from file name
+		filename := filepath.Base(path)
+		matches := MigrateFilePattern.FindStringSubmatch(filename)
+		if len(matches) > 2 {
+			file.Version = matches[1]
+			file.Name = matches[2]
+		}
+	}
+	return file, err
+}
+
+func NewMigrationFromReader(sql io.Reader) (*MigrationFile, error) {
+	lines, err := parser.SplitAndTrim(sql)
+	if err != nil {
+		return nil, err
+	}
+	return &MigrationFile{Statements: lines}, nil
+}
+
+func (m *MigrationFile) ExecBatch(ctx context.Context, conn *pgx.Conn) error {
+	// Batch migration commands, without using statement cache
+	batch := &pgconn.Batch{}
+	for _, line := range m.Statements {
+		batch.ExecParams(line, nil, nil, nil, nil)
+	}
+	// Insert into migration history
+	if len(m.Version) > 0 {
+		if err := m.insertVersionSQL(conn, batch); err != nil {
+			return err
+		}
+	}
+	// ExecBatch is implicitly transactional
+	if result, err := conn.PgConn().ExecBatch(ctx, batch).ReadAll(); err != nil {
+		// Defaults to printing the last statement on error
+		stat := INSERT_MIGRATION_VERSION
+		i := len(result)
+		if i < len(m.Statements) {
+			stat = m.Statements[i]
+		}
+		return errors.Errorf("%w\nAt statement %d: %s", err, i, stat)
+	}
+	return nil
+}
+
+func (m *MigrationFile) insertVersionSQL(conn *pgx.Conn, batch *pgconn.Batch) error {
+	value := pgtype.TextArray{}
+	if err := value.Set(m.Statements); err != nil {
+		return errors.Errorf("failed to set text array: %w", err)
+	}
+	ci := conn.ConnInfo()
+	var err error
+	var encoded []byte
+	var valueFormat int16
+	if conn.Config().PreferSimpleProtocol {
+		encoded, err = value.EncodeText(ci, encoded)
+		valueFormat = pgtype.TextFormatCode
+	} else {
+		encoded, err = value.EncodeBinary(ci, encoded)
+		valueFormat = pgtype.BinaryFormatCode
+	}
+	if err != nil {
+		return errors.Errorf("failed to encode binary: %w", err)
+	}
+	batch.ExecParams(
+		INSERT_MIGRATION_VERSION,
+		[][]byte{[]byte(m.Version), []byte(m.Name), encoded},
+		[]uint32{pgtype.TextOID, pgtype.TextOID, pgtype.TextArrayOID},
+		[]int16{pgtype.TextFormatCode, pgtype.TextFormatCode, valueFormat},
+		nil,
+	)
+	return nil
+}
+
+func (m *MigrationFile) ExecBatchWithCache(ctx context.Context, conn *pgx.Conn) error {
+	// Data statements don't mutate schemas, safe to use statement cache
+	batch := pgx.Batch{}
+	for _, line := range m.Statements {
+		batch.Queue(line)
+	}
+	// No need to track version here because there are no schema changes
+	if err := conn.SendBatch(ctx, &batch).Close(); err != nil {
+		return errors.Errorf("failed to send batch: %w", err)
+	}
+	return nil
+}
diff --git a/internal/migration/history/history.go b/pkg/migration/history.go
similarity index 87%
rename from internal/migration/history/history.go
rename to pkg/migration/history.go
index f625dd38..6dc8b5fc 100644
--- a/internal/migration/history/history.go
+++ b/pkg/migration/history.go
@@ -1,4 +1,4 @@
-package history
+package migration
 
 import (
 	"context"
@@ -20,14 +20,10 @@ const (
 	DELETE_MIGRATION_BEFORE  = "DELETE FROM supabase_migrations.schema_migrations WHERE version <= $1"
 	TRUNCATE_VERSION_TABLE   = "TRUNCATE supabase_migrations.schema_migrations"
 	SELECT_VERSION_TABLE     = "SELECT * FROM supabase_migrations.schema_migrations"
+	LIST_MIGRATION_VERSION   = "SELECT version FROM supabase_migrations.schema_migrations ORDER BY version"
 )
 
-type SchemaMigration struct {
-	Version    string
-	Name       string
-	Statements []string
-}
-
+// TODO: support overriding `supabase_migrations.schema_migrations` with user defined <schema>.<table>
 func CreateMigrationTable(ctx context.Context, conn *pgx.Conn) error {
 	// This must be run without prepared statements because each statement in the batch depends on
 	// the previous schema change. The lock timeout will be reset when implicit transaction ends.
@@ -43,10 +39,10 @@ func CreateMigrationTable(ctx context.Context, conn *pgx.Conn) error {
 	return nil
 }
 
-func ReadMigrationTable(ctx context.Context, conn *pgx.Conn) ([]SchemaMigration, error) {
+func ReadMigrationTable(ctx context.Context, conn *pgx.Conn) ([]MigrationFile, error) {
 	rows, err := conn.Query(ctx, SELECT_VERSION_TABLE)
 	if err != nil {
 		return nil, errors.Errorf("failed to read migration table: %w", err)
 	}
-	return pgxv5.CollectRows[SchemaMigration](rows)
+	return pgxv5.CollectRows[MigrationFile](rows)
 }
diff --git a/pkg/migration/list.go b/pkg/migration/list.go
new file mode 100644
index 00000000..d4e1984a
--- /dev/null
+++ b/pkg/migration/list.go
@@ -0,0 +1,77 @@
+package migration
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"os"
+	"regexp"
+	"strconv"
+
+	"github.com/go-errors/errors"
+	"github.com/jackc/pgconn"
+	"github.com/jackc/pgerrcode"
+	"github.com/jackc/pgx/v4"
+	"github.com/supabase/cli/pkg/pgxv5"
+)
+
+func ListRemoteMigrations(ctx context.Context, conn *pgx.Conn) ([]string, error) {
+	// We query the version string only for backwards compatibility
+	rows, err := conn.Query(ctx, LIST_MIGRATION_VERSION)
+	if err != nil {
+		return nil, errors.Errorf("failed to query rows: %w", err)
+	}
+	versions, err := pgxv5.CollectStrings(rows)
+	if err != nil {
+		var pgErr *pgconn.PgError
+		if errors.As(err, &pgErr) && pgErr.Code == pgerrcode.UndefinedTable {
+			// If migration history table is undefined, the remote project has no migrations
+			return nil, nil
+		}
+	}
+	return versions, err
+}
+
+func ListLocalMigrations(path string, fsys fs.FS, filter ...func(string) bool) ([]string, error) {
+	localMigrations, err := fs.ReadDir(fsys, path)
+	if err != nil && !errors.Is(err, os.ErrNotExist) {
+		return nil, errors.Errorf("failed to read directory: %w", err)
+	}
+	var names []string
+	for i, migration := range localMigrations {
+		filename := migration.Name()
+		if i == 0 && shouldSkip(filename) {
+			fmt.Fprintf(os.Stderr, "Skipping migration %s... (replace \"init\" with a different file name to apply this migration)\n", filename)
+			continue
+		}
+		matches := MigrateFilePattern.FindStringSubmatch(filename)
+		if len(matches) == 0 {
+			fmt.Fprintf(os.Stderr, "Skipping migration %s... (file name must match pattern \"<timestamp>_name.sql\")\n", filename)
+			continue
+		}
+		if len(filter) == 0 {
+			names = append(names, filename)
+			continue
+		}
+		for _, keep := range filter {
+			if version := matches[1]; keep(version) {
+				names = append(names, filename)
+			}
+		}
+	}
+	return names, nil
+}
+
+var initSchemaPattern = regexp.MustCompile(`([0-9]{14})_init\.sql`)
+
+func shouldSkip(name string) bool {
+	// NOTE: To handle backward-compatibility. `<timestamp>_init.sql` as
+	// the first migration (prev versions of the CLI) is deprecated.
+	matches := initSchemaPattern.FindStringSubmatch(name)
+	if len(matches) == 2 {
+		if timestamp, err := strconv.ParseUint(matches[1], 10, 64); err == nil && timestamp < 20211209000000 {
+			return true
+		}
+	}
+	return false
+}
diff --git a/pkg/migration/seed.go b/pkg/migration/seed.go
new file mode 100644
index 00000000..108ecb3f
--- /dev/null
+++ b/pkg/migration/seed.go
@@ -0,0 +1,38 @@
+package migration
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+
+	"github.com/jackc/pgx/v4"
+)
+
+func SeedData(ctx context.Context, pending []string, conn *pgx.Conn, fsys fs.FS) error {
+	for _, path := range pending {
+		filename := filepath.Base(path)
+		fmt.Fprintf(os.Stderr, "Seeding data %s...", filename)
+		// Batch seed commands, safe to use statement cache
+		if seed, err := NewMigrationFromFile(path, fsys); err != nil {
+			return err
+		} else if err := seed.ExecBatchWithCache(ctx, conn); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func SeedGlobals(ctx context.Context, pending []string, conn *pgx.Conn, fsys fs.FS) error {
+	for _, path := range pending {
+		filename := filepath.Base(path)
+		fmt.Fprintf(os.Stderr, "Seeding globals %s...", filename)
+		if globals, err := NewMigrationFromFile(path, fsys); err != nil {
+			return err
+		} else if err := globals.ExecBatch(ctx, conn); err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/pkg/pgxv5/connect.go b/pkg/pgxv5/connect.go
new file mode 100644
index 00000000..f82a31d1
--- /dev/null
+++ b/pkg/pgxv5/connect.go
@@ -0,0 +1,27 @@
+package pgxv5
+
+import (
+	"context"
+
+	"github.com/go-errors/errors"
+	"github.com/jackc/pgx/v4"
+)
+
+// Extends pgx.Connect with support for programmatically overriding parsed config
+func Connect(ctx context.Context, connString string, options ...func(*pgx.ConnConfig)) (*pgx.Conn, error) {
+	// Parse connection url
+	config, err := pgx.ParseConfig(connString)
+	if err != nil {
+		return nil, errors.Errorf("failed to parse connection string: %w", err)
+	}
+	// Apply config overrides
+	for _, op := range options {
+		op(config)
+	}
+	// Connect to database
+	conn, err := pgx.ConnectConfig(ctx, config)
+	if err != nil {
+		return nil, errors.Errorf("failed to connect to postgres: %w", err)
+	}
+	return conn, nil
+}
diff --git a/pkg/storage/objects.go b/pkg/storage/objects.go
index 7f417178..8e840fdf 100644
--- a/pkg/storage/objects.go
+++ b/pkg/storage/objects.go
@@ -77,6 +77,7 @@ func ParseFileOptions(f afero.File, opts ...func(*FileOptions)) (*FileOptions, e
 			return nil, errors.Errorf("failed to read file: %w", err)
 		}
 		fo.ContentType = http.DetectContentType(buf)
+		// TODO: handle cases where file is not seekable
 		_, err = f.Seek(0, io.SeekStart)
 		if err != nil {
 			return nil, errors.Errorf("failed to seek file: %w", err)


Test patch:
diff --git a/internal/db/branch/switch_/switch__test.go b/internal/db/branch/switch_/switch__test.go
index 31b4aaa7..42960662 100644
--- a/internal/db/branch/switch_/switch__test.go
+++ b/internal/db/branch/switch_/switch__test.go
@@ -15,8 +15,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/testing/apitest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 func TestSwitchCommand(t *testing.T) {
diff --git a/internal/db/diff/diff_test.go b/internal/db/diff/diff_test.go
index 4d36a8b1..cadbd54d 100644
--- a/internal/db/diff/diff_test.go
+++ b/internal/db/diff/diff_test.go
@@ -19,12 +19,13 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/db/reset"
 	"github.com/supabase/cli/internal/db/start"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/testing/apitest"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/internal/testing/helper"
 	"github.com/supabase/cli/internal/utils"
 	"github.com/supabase/cli/pkg/config"
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
@@ -163,10 +164,10 @@ func TestMigrateShadow(t *testing.T) {
 			Reply("CREATE SCHEMA").
 			Query(utils.InitialSchemaPg14Sql).
 			Reply("CREATE SCHEMA")
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(sql).
+		helper.MockMigrationHistory(conn).
+			Query(sql).
 			Reply("CREATE SCHEMA").
-			Query(history.INSERT_MIGRATION_VERSION, "0", "test", []string{sql}).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "test", []string{sql}).
 			Reply("INSERT 0 1")
 		// Run test
 		err := MigrateShadowDatabase(context.Background(), "test-shadow-db", fsys, conn.Intercept)
@@ -336,10 +337,10 @@ At statement 0: create schema public`)
 			Reply("CREATE SCHEMA").
 			Query(utils.InitialSchemaPg14Sql).
 			Reply("CREATE SCHEMA")
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(sql).
+		helper.MockMigrationHistory(conn).
+			Query(sql).
 			Reply("CREATE SCHEMA").
-			Query(history.INSERT_MIGRATION_VERSION, "0", "test", []string{sql}).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "test", []string{sql}).
 			Reply("INSERT 0 1")
 		// Run test
 		diff, err := DiffDatabase(context.Background(), []string{"public"}, dbConfig, io.Discard, fsys, DiffSchemaMigra, conn.Intercept)
diff --git a/internal/db/lint/lint_test.go b/internal/db/lint/lint_test.go
index 144c229c..40ce0503 100644
--- a/internal/db/lint/lint_test.go
+++ b/internal/db/lint/lint_test.go
@@ -15,8 +15,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/testing/apitest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
@@ -121,13 +121,8 @@ func TestLintDatabase(t *testing.T) {
 				[]interface{}{"f2", string(r2)},
 			).
 			Query("rollback").Reply("ROLLBACK")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		result, err := LintDatabase(ctx, mock, []string{"public"})
+		result, err := LintDatabase(context.Background(), conn.MockClient(t), []string{"public"})
 		assert.NoError(t, err)
 		// Validate result
 		assert.ElementsMatch(t, expected, result)
@@ -167,15 +162,10 @@ func TestLintDatabase(t *testing.T) {
 			Query(checkSchemaScript, "private").
 			Reply("SELECT 1", []interface{}{"f2", string(r2)}).
 			Query("rollback").Reply("ROLLBACK")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		result, err := LintDatabase(ctx, mock, []string{"public", "private"})
+		result, err := LintDatabase(context.Background(), conn.MockClient(t), []string{"public", "private"})
+		// Check error
 		assert.NoError(t, err)
-		// Validate result
 		assert.ElementsMatch(t, expected, result)
 	})
 
@@ -187,13 +177,9 @@ func TestLintDatabase(t *testing.T) {
 			Query(ENABLE_PGSQL_CHECK).
 			ReplyError(pgerrcode.UndefinedFile, `could not open extension control file "/usr/share/postgresql/14/extension/plpgsql_check.control": No such file or directory"`).
 			Query("rollback").Reply("ROLLBACK")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		_, err = LintDatabase(ctx, mock, []string{"public"})
+		_, err := LintDatabase(context.Background(), conn.MockClient(t), []string{"public"})
+		// Check error
 		assert.Error(t, err)
 	})
 
@@ -207,13 +193,9 @@ func TestLintDatabase(t *testing.T) {
 			Query(checkSchemaScript, "public").
 			Reply("SELECT 1", []interface{}{"f1", "malformed"}).
 			Query("rollback").Reply("ROLLBACK")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		_, err = LintDatabase(ctx, mock, []string{"public"})
+		_, err := LintDatabase(context.Background(), conn.MockClient(t), []string{"public"})
+		// Check error
 		assert.Error(t, err)
 	})
 }
diff --git a/internal/db/pull/pull_test.go b/internal/db/pull/pull_test.go
index 76d61607..5c50e8f9 100644
--- a/internal/db/pull/pull_test.go
+++ b/internal/db/pull/pull_test.go
@@ -14,11 +14,11 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/migration/list"
 	"github.com/supabase/cli/internal/testing/apitest"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
@@ -71,7 +71,7 @@ func TestPullCommand(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			ReplyError(pgerrcode.InvalidCatalogName, `database "postgres" does not exist`)
 		// Run test
 		err := Run(context.Background(), nil, dbConfig, "", fsys, conn.Intercept)
@@ -93,16 +93,11 @@ func TestPullSchema(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		err = utils.RunProgram(context.Background(), func(p utils.Program, ctx context.Context) error {
-			return run(p, ctx, nil, "0_test.sql", mock, fsys)
+		err := utils.RunProgram(context.Background(), func(p utils.Program, ctx context.Context) error {
+			return run(p, ctx, nil, "0_test.sql", conn.MockClient(t), fsys)
 		})
 		// Check error
 		assert.NoError(t, err)
@@ -120,18 +115,13 @@ func TestPullSchema(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 1", []interface{}{"0"}).
 			Query(reset.ListSchemas, escapedSchemas).
 			ReplyError(pgerrcode.DuplicateTable, `relation "test" already exists`)
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		err = utils.RunProgram(context.Background(), func(p utils.Program, ctx context.Context) error {
-			return run(p, ctx, nil, "", mock, fsys)
+		err := utils.RunProgram(context.Background(), func(p utils.Program, ctx context.Context) error {
+			return run(p, ctx, nil, "", conn.MockClient(t), fsys)
 		})
 		// Check error
 		assert.ErrorContains(t, err, `ERROR: relation "test" already exists (SQLSTATE 42P07)`)
@@ -151,16 +141,11 @@ func TestPullSchema(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 1", []interface{}{"0"})
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		err = utils.RunProgram(context.Background(), func(p utils.Program, ctx context.Context) error {
-			return run(p, ctx, []string{"public"}, "", mock, fsys)
+		err := utils.RunProgram(context.Background(), func(p utils.Program, ctx context.Context) error {
+			return run(p, ctx, []string{"public"}, "", conn.MockClient(t), fsys)
 		})
 		// Check error
 		assert.ErrorContains(t, err, "network error")
@@ -175,15 +160,10 @@ func TestSyncRemote(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		err = assertRemoteInSync(ctx, mock, fsys)
+		err := assertRemoteInSync(context.Background(), conn.MockClient(t), fsys)
 		// Check error
 		assert.ErrorIs(t, err, os.ErrPermission)
 		assert.Empty(t, apitest.ListUnmatchedRequests())
@@ -197,15 +177,10 @@ func TestSyncRemote(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		err = assertRemoteInSync(ctx, mock, fsys)
+		err := assertRemoteInSync(context.Background(), conn.MockClient(t), fsys)
 		// Check error
 		assert.ErrorIs(t, err, errConflict)
 		assert.Empty(t, apitest.ListUnmatchedRequests())
@@ -219,15 +194,10 @@ func TestSyncRemote(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 1", []interface{}{"20220727064247"})
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		err = assertRemoteInSync(ctx, mock, fsys)
+		err := assertRemoteInSync(context.Background(), conn.MockClient(t), fsys)
 		// Check error
 		assert.ErrorIs(t, err, errConflict)
 		assert.Empty(t, apitest.ListUnmatchedRequests())
@@ -239,15 +209,10 @@ func TestSyncRemote(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		err = assertRemoteInSync(ctx, mock, fsys)
+		err := assertRemoteInSync(context.Background(), conn.MockClient(t), fsys)
 		// Check error
 		assert.ErrorIs(t, err, errMissing)
 		assert.Empty(t, apitest.ListUnmatchedRequests())
diff --git a/internal/db/push/push_test.go b/internal/db/push/push_test.go
index e2c17dad..44d9f024 100644
--- a/internal/db/push/push_test.go
+++ b/internal/db/push/push_test.go
@@ -11,11 +11,11 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/supabase/cli/internal/migration/history"
-	"github.com/supabase/cli/internal/migration/list"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/internal/testing/helper"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
@@ -35,7 +35,7 @@ func TestMigrationPush(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
 		// Run test
 		err := Run(context.Background(), true, false, true, true, dbConfig, fsys, conn.Intercept)
@@ -49,7 +49,7 @@ func TestMigrationPush(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
 		// Run test
 		err := Run(context.Background(), false, false, false, false, dbConfig, fsys, conn.Intercept)
@@ -72,7 +72,7 @@ func TestMigrationPush(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			ReplyError(pgerrcode.InvalidCatalogName, `database "target" does not exist`)
 		// Run test
 		err := Run(context.Background(), false, false, false, false, pgconn.Config{
@@ -94,16 +94,16 @@ func TestMigrationPush(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(history.INSERT_MIGRATION_VERSION, "0", "test", nil).
+		helper.MockMigrationHistory(conn).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "test", nil).
 			ReplyError(pgerrcode.NotNullViolation, `null value in column "version" of relation "schema_migrations"`)
 		// Run test
 		err := Run(context.Background(), false, false, false, false, dbConfig, fsys, conn.Intercept)
 		// Check error
 		assert.ErrorContains(t, err, `ERROR: null value in column "version" of relation "schema_migrations" (SQLSTATE 23502)`)
-		assert.ErrorContains(t, err, "At statement 0: "+history.INSERT_MIGRATION_VERSION)
+		assert.ErrorContains(t, err, "At statement 0: "+migration.INSERT_MIGRATION_VERSION)
 	})
 }
 
@@ -116,10 +116,10 @@ func TestPushAll(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(history.INSERT_MIGRATION_VERSION, "0", "test", nil).
+		helper.MockMigrationHistory(conn).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "test", nil).
 			Reply("INSERT 0 1")
 		// Run test
 		err := Run(context.Background(), false, false, true, true, dbConfig, fsys, conn.Intercept)
@@ -136,7 +136,7 @@ func TestPushAll(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
 		// Run test
 		err := Run(context.Background(), false, false, true, true, dbConfig, fsys, conn.Intercept)
@@ -152,7 +152,7 @@ func TestPushAll(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
 		// Run test
 		err := Run(context.Background(), false, false, true, false, dbConfig, fsys, conn.Intercept)
@@ -168,10 +168,10 @@ func TestPushAll(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(history.INSERT_MIGRATION_VERSION, "0", "test", nil).
+		helper.MockMigrationHistory(conn).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "test", nil).
 			Reply("INSERT 0 1")
 		// Run test
 		err := Run(context.Background(), false, false, false, true, dbConfig, fsys, conn.Intercept)
diff --git a/internal/db/reset/reset_test.go b/internal/db/reset/reset_test.go
index 6715feea..01d0e435 100644
--- a/internal/db/reset/reset_test.go
+++ b/internal/db/reset/reset_test.go
@@ -19,8 +19,8 @@ import (
 	"github.com/supabase/cli/internal/db/start"
 	"github.com/supabase/cli/internal/testing/apitest"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 func TestResetCommand(t *testing.T) {
diff --git a/internal/db/start/start_test.go b/internal/db/start/start_test.go
index d3972332..dcc93b4c 100644
--- a/internal/db/start/start_test.go
+++ b/internal/db/start/start_test.go
@@ -16,8 +16,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/testing/apitest"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 func TestInitBranch(t *testing.T) {
diff --git a/internal/db/test/test_test.go b/internal/db/test/test_test.go
index 16d00dbe..7c931099 100644
--- a/internal/db/test/test_test.go
+++ b/internal/db/test/test_test.go
@@ -12,9 +12,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/testing/apitest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
 	"github.com/supabase/cli/pkg/config"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/debug/postgres_test.go b/internal/debug/postgres_test.go
index 027d8ee9..9f46fd6b 100644
--- a/internal/debug/postgres_test.go
+++ b/internal/debug/postgres_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/jackc/pgx/v4"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 func TestPostgresProxy(t *testing.T) {
diff --git a/internal/gen/types/types_test.go b/internal/gen/types/types_test.go
index 125bc569..3c05c15b 100644
--- a/internal/gen/types/types_test.go
+++ b/internal/gen/types/types_test.go
@@ -13,9 +13,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/testing/apitest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
 	"github.com/supabase/cli/pkg/api"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 func TestGenLocalCommand(t *testing.T) {
diff --git a/internal/inspect/bloat/bloat_test.go b/internal/inspect/bloat/bloat_test.go
index 5cbbfcbf..8646565c 100644
--- a/internal/inspect/bloat/bloat_test.go
+++ b/internal/inspect/bloat/bloat_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/blocking/blocking_test.go b/internal/inspect/blocking/blocking_test.go
index d41f2406..46b63441 100644
--- a/internal/inspect/blocking/blocking_test.go
+++ b/internal/inspect/blocking/blocking_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/cache/cache_test.go b/internal/inspect/cache/cache_test.go
index 36b50818..28fa1cb7 100644
--- a/internal/inspect/cache/cache_test.go
+++ b/internal/inspect/cache/cache_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/calls/calls_test.go b/internal/inspect/calls/calls_test.go
index 5620ec6a..ded27100 100644
--- a/internal/inspect/calls/calls_test.go
+++ b/internal/inspect/calls/calls_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/index_sizes/index_sizes_test.go b/internal/inspect/index_sizes/index_sizes_test.go
index 72742e74..9071c571 100644
--- a/internal/inspect/index_sizes/index_sizes_test.go
+++ b/internal/inspect/index_sizes/index_sizes_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/index_usage/index_usage_test.go b/internal/inspect/index_usage/index_usage_test.go
index 24d61ebf..5b735bb6 100644
--- a/internal/inspect/index_usage/index_usage_test.go
+++ b/internal/inspect/index_usage/index_usage_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/locks/locks_test.go b/internal/inspect/locks/locks_test.go
index 7c342b2d..e4c55c6b 100644
--- a/internal/inspect/locks/locks_test.go
+++ b/internal/inspect/locks/locks_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/long_running_queries/long_running_queries_test.go b/internal/inspect/long_running_queries/long_running_queries_test.go
index edb97d6d..d936c61a 100644
--- a/internal/inspect/long_running_queries/long_running_queries_test.go
+++ b/internal/inspect/long_running_queries/long_running_queries_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/outliers/outliers_test.go b/internal/inspect/outliers/outliers_test.go
index b34a87cf..e5d46432 100644
--- a/internal/inspect/outliers/outliers_test.go
+++ b/internal/inspect/outliers/outliers_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/replication_slots/replication_slots_test.go b/internal/inspect/replication_slots/replication_slots_test.go
index 8ed02f13..d9e9b322 100644
--- a/internal/inspect/replication_slots/replication_slots_test.go
+++ b/internal/inspect/replication_slots/replication_slots_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/report_test.go b/internal/inspect/report_test.go
index d019976b..4de1ab60 100644
--- a/internal/inspect/report_test.go
+++ b/internal/inspect/report_test.go
@@ -27,7 +27,7 @@ import (
 	"github.com/supabase/cli/internal/inspect/total_table_sizes"
 	"github.com/supabase/cli/internal/inspect/unused_indexes"
 	"github.com/supabase/cli/internal/inspect/vacuum_stats"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/role_connections/role_connections_test.go b/internal/inspect/role_connections/role_connections_test.go
index 79c15355..32ecab76 100644
--- a/internal/inspect/role_connections/role_connections_test.go
+++ b/internal/inspect/role_connections/role_connections_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/seq_scans/seq_scans_test.go b/internal/inspect/seq_scans/seq_scans_test.go
index 3bfb2ba6..3db6caee 100644
--- a/internal/inspect/seq_scans/seq_scans_test.go
+++ b/internal/inspect/seq_scans/seq_scans_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/table_index_sizes/table_index_sizes_test.go b/internal/inspect/table_index_sizes/table_index_sizes_test.go
index 0a58a710..20ad80fc 100644
--- a/internal/inspect/table_index_sizes/table_index_sizes_test.go
+++ b/internal/inspect/table_index_sizes/table_index_sizes_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/table_record_counts/table_record_counts_test.go b/internal/inspect/table_record_counts/table_record_counts_test.go
index c5f97683..a03714d9 100644
--- a/internal/inspect/table_record_counts/table_record_counts_test.go
+++ b/internal/inspect/table_record_counts/table_record_counts_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/table_sizes/table_sizes_test.go b/internal/inspect/table_sizes/table_sizes_test.go
index f96e17e1..5cc6426a 100644
--- a/internal/inspect/table_sizes/table_sizes_test.go
+++ b/internal/inspect/table_sizes/table_sizes_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/total_index_size/total_index_size_test.go b/internal/inspect/total_index_size/total_index_size_test.go
index 6a6591f2..8eb0e0aa 100644
--- a/internal/inspect/total_index_size/total_index_size_test.go
+++ b/internal/inspect/total_index_size/total_index_size_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/total_table_sizes/total_table_sizes_test.go b/internal/inspect/total_table_sizes/total_table_sizes_test.go
index a207456c..bc548af6 100644
--- a/internal/inspect/total_table_sizes/total_table_sizes_test.go
+++ b/internal/inspect/total_table_sizes/total_table_sizes_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/unused_indexes/unused_indexes_test.go b/internal/inspect/unused_indexes/unused_indexes_test.go
index d295c164..ee418209 100644
--- a/internal/inspect/unused_indexes/unused_indexes_test.go
+++ b/internal/inspect/unused_indexes/unused_indexes_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/inspect/vacuum_stats/vacuum_stats_test.go b/internal/inspect/vacuum_stats/vacuum_stats_test.go
index 3e1b93f3..0d3cbec1 100644
--- a/internal/inspect/vacuum_stats/vacuum_stats_test.go
+++ b/internal/inspect/vacuum_stats/vacuum_stats_test.go
@@ -8,8 +8,8 @@ import (
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/supabase/cli/internal/db/reset"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/internal/link/link_test.go b/internal/link/link_test.go
index 0a4a071f..a2f63a82 100644
--- a/internal/link/link_test.go
+++ b/internal/link/link_test.go
@@ -12,13 +12,14 @@ import (
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/testing/apitest"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/internal/testing/helper"
 	"github.com/supabase/cli/internal/utils"
 	"github.com/supabase/cli/internal/utils/tenant"
 	"github.com/supabase/cli/pkg/api"
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
 	"github.com/zalando/go-keyring"
 )
 
@@ -82,7 +83,7 @@ func TestLinkCommand(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
+		helper.MockMigrationHistory(conn)
 		// Flush pending mocks after test execution
 		defer gock.OffAll()
 		gock.New(utils.DefaultApiHost).
@@ -327,7 +328,7 @@ func TestLinkDatabase(t *testing.T) {
 			"standard_conforming_strings": "on",
 		})
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
+		helper.MockMigrationHistory(conn)
 		// Run test
 		err := linkDatabase(context.Background(), dbConfig, conn.Intercept)
 		// Check error
@@ -344,7 +345,7 @@ func TestLinkDatabase(t *testing.T) {
 			"server_version":              "15.0",
 		})
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
+		helper.MockMigrationHistory(conn)
 		// Run test
 		err := linkDatabase(context.Background(), dbConfig, conn.Intercept)
 		// Check error
@@ -361,13 +362,13 @@ func TestLinkDatabase(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(history.SET_LOCK_TIMEOUT).
-			Query(history.CREATE_VERSION_SCHEMA).
+		conn.Query(migration.SET_LOCK_TIMEOUT).
+			Query(migration.CREATE_VERSION_SCHEMA).
 			Reply("CREATE SCHEMA").
-			Query(history.CREATE_VERSION_TABLE).
+			Query(migration.CREATE_VERSION_TABLE).
 			ReplyError(pgerrcode.InsufficientPrivilege, "permission denied for relation supabase_migrations").
-			Query(history.ADD_STATEMENTS_COLUMN).
-			Query(history.ADD_NAME_COLUMN)
+			Query(migration.ADD_STATEMENTS_COLUMN).
+			Query(migration.ADD_NAME_COLUMN)
 		// Run test
 		err := linkDatabase(context.Background(), dbConfig, conn.Intercept)
 		// Check error
diff --git a/internal/migration/apply/apply_test.go b/internal/migration/apply/apply_test.go
index af124c2c..9c5ec34d 100644
--- a/internal/migration/apply/apply_test.go
+++ b/internal/migration/apply/apply_test.go
@@ -6,15 +6,15 @@ import (
 	"path/filepath"
 	"testing"
 
-	"github.com/jackc/pgconn"
 	"github.com/jackc/pgerrcode"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/internal/testing/helper"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 func TestMigrateDatabase(t *testing.T) {
@@ -27,18 +27,13 @@ func TestMigrateDatabase(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(sql).
+		helper.MockMigrationHistory(conn).
+			Query(sql).
 			Reply("CREATE SCHEMA").
-			Query(history.INSERT_MIGRATION_VERSION, "0", "test", []string{sql}).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "test", []string{sql}).
 			Reply("INSERT 0 1")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		err = MigrateAndSeed(ctx, "", mock, fsys)
+		err := MigrateAndSeed(context.Background(), "", conn.MockClient(t), fsys)
 		// Check error
 		assert.NoError(t, err)
 	})
@@ -69,13 +64,10 @@ func TestSeedDatabase(t *testing.T) {
 		defer conn.Close(t)
 		conn.Query(sql).
 			Reply("INSERT 0 1")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		assert.NoError(t, SeedDatabase(ctx, mock, fsys))
+		err := SeedDatabase(context.Background(), conn.MockClient(t), fsys)
+		// Check error
+		assert.NoError(t, err)
 	})
 
 	t.Run("ignores missing seed", func(t *testing.T) {
@@ -102,13 +94,8 @@ func TestSeedDatabase(t *testing.T) {
 		defer conn.Close(t)
 		conn.Query(sql).
 			ReplyError(pgerrcode.NotNullViolation, `null value in column "age" of relation "employees"`)
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		err = SeedDatabase(ctx, mock, fsys)
+		err := SeedDatabase(context.Background(), conn.MockClient(t), fsys)
 		// Check error
 		assert.ErrorContains(t, err, `ERROR: null value in column "age" of relation "employees" (SQLSTATE 23502)`)
 	})
@@ -121,14 +108,9 @@ func TestMigrateUp(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
+		helper.MockMigrationHistory(conn)
 		// Run test
-		err = MigrateUp(context.Background(), mock, []string{"20220727064247_missing.sql"}, fsys)
+		err := MigrateUp(context.Background(), conn.MockClient(t), []string{"20220727064247_missing.sql"}, fsys)
 		// Check error
 		assert.ErrorIs(t, err, os.ErrNotExist)
 	})
diff --git a/internal/migration/list/list_test.go b/internal/migration/list/list_test.go
index cb3234a2..b23fa4a2 100644
--- a/internal/migration/list/list_test.go
+++ b/internal/migration/list/list_test.go
@@ -13,8 +13,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
@@ -32,7 +33,7 @@ func TestMigrationList(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
 		// Run test
 		err := Run(context.Background(), dbConfig, fsys, conn.Intercept)
@@ -55,7 +56,7 @@ func TestMigrationList(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
 		// Run test
 		err := Run(context.Background(), dbConfig, fsys, conn.Intercept)
@@ -69,7 +70,7 @@ func TestRemoteMigrations(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 1", []interface{}{"20220727064247"})
 		// Run test
 		versions, err := loadRemoteVersions(context.Background(), dbConfig, conn.Intercept)
@@ -89,7 +90,7 @@ func TestRemoteMigrations(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			ReplyError(pgerrcode.UndefinedTable, "relation \"supabase_migrations.schema_migrations\" does not exist")
 		// Run test
 		versions, err := loadRemoteVersions(context.Background(), dbConfig, conn.Intercept)
@@ -102,7 +103,7 @@ func TestRemoteMigrations(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 1", []interface{}{})
 		// Run test
 		_, err := loadRemoteVersions(context.Background(), dbConfig, conn.Intercept)
diff --git a/internal/migration/repair/repair_test.go b/internal/migration/repair/repair_test.go
index d5a18fc0..a291e5ea 100644
--- a/internal/migration/repair/repair_test.go
+++ b/internal/migration/repair/repair_test.go
@@ -1,26 +1,22 @@
 package repair
 
 import (
-	"bufio"
 	"context"
-	"fmt"
 	"os"
 	"path/filepath"
-	"strings"
 	"testing"
 
 	"github.com/jackc/pgconn"
 	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v4"
 	"github.com/spf13/afero"
-	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/internal/testing/helper"
 	"github.com/supabase/cli/internal/utils"
-	"github.com/supabase/cli/pkg/parser"
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
@@ -40,8 +36,8 @@ func TestRepairCommand(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(history.INSERT_MIGRATION_VERSION, "0", "test", []string{"select 1"}).
+		helper.MockMigrationHistory(conn).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "test", []string{"select 1"}).
 			Reply("INSERT 0 1")
 		// Run test
 		err := Run(context.Background(), dbConfig, []string{"0"}, Applied, fsys, conn.Intercept)
@@ -55,8 +51,8 @@ func TestRepairCommand(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(history.DELETE_MIGRATION_VERSION, []string{"0"}).
+		helper.MockMigrationHistory(conn).
+			Query(migration.DELETE_MIGRATION_VERSION, []string{"0"}).
 			Reply("DELETE 1")
 		// Run test
 		err := Run(context.Background(), dbConfig, []string{"0"}, Reverted, fsys, conn.Intercept)
@@ -90,8 +86,8 @@ func TestRepairCommand(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(history.INSERT_MIGRATION_VERSION, "0", "test", nil).
+		helper.MockMigrationHistory(conn).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "test", nil).
 			ReplyError(pgerrcode.DuplicateObject, `relation "supabase_migrations.schema_migrations" does not exist`)
 		// Run test
 		err := Run(context.Background(), dbConfig, []string{"0"}, Applied, fsys, conn.Intercept)
@@ -110,8 +106,8 @@ func TestRepairAll(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(history.TRUNCATE_VERSION_TABLE + `;INSERT INTO supabase_migrations.schema_migrations(version, name, statements) VALUES( '0' ,  'test' ,  '{select 1}' )`).
+		helper.MockMigrationHistory(conn).
+			Query(migration.TRUNCATE_VERSION_TABLE + `;INSERT INTO supabase_migrations.schema_migrations(version, name, statements) VALUES( '0' ,  'test' ,  '{select 1}' )`).
 			Reply("TRUNCATE TABLE").
 			Reply("INSERT 0 1")
 		// Run test
@@ -129,8 +125,8 @@ func TestRepairAll(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(history.TRUNCATE_VERSION_TABLE).
+		helper.MockMigrationHistory(conn).
+			Query(migration.TRUNCATE_VERSION_TABLE).
 			Reply("TRUNCATE TABLE")
 		// Run test
 		err := Run(context.Background(), dbConfig, nil, Reverted, fsys, conn.Intercept)
@@ -157,82 +153,3 @@ func TestRepairAll(t *testing.T) {
 		assert.ErrorIs(t, err, os.ErrPermission)
 	})
 }
-
-func TestMigrationFile(t *testing.T) {
-	t.Run("new from file sets max token", func(t *testing.T) {
-		viper.Reset()
-		// Setup in-memory fs
-		fsys := afero.NewMemMapFs()
-		// Setup initial migration
-		name := "20220727064247_create_table.sql"
-		path := filepath.Join(utils.MigrationsDir, name)
-		query := "BEGIN; " + strings.Repeat("a", parser.MaxScannerCapacity)
-		require.NoError(t, afero.WriteFile(fsys, path, []byte(query), 0644))
-		// Run test
-		migration, err := NewMigrationFromFile(path, fsys)
-		// Check error
-		assert.NoError(t, err)
-		assert.Len(t, migration.Lines, 2)
-		assert.Equal(t, "20220727064247", migration.Version)
-	})
-
-	t.Run("new from reader errors on max token", func(t *testing.T) {
-		viper.Reset()
-		sql := "\tBEGIN; " + strings.Repeat("a", parser.MaxScannerCapacity)
-		// Run test
-		migration, err := NewMigrationFromReader(strings.NewReader(sql))
-		// Check error
-		assert.ErrorIs(t, err, bufio.ErrTooLong)
-		assert.ErrorContains(t, err, "After statement 1: \tBEGIN;")
-		assert.Nil(t, migration)
-	})
-
-	t.Run("encodes statements in binary format", func(t *testing.T) {
-		migration := MigrationFile{
-			Lines:   []string{"create schema public"},
-			Version: "0",
-		}
-		// Setup mock postgres
-		conn := pgtest.NewConn()
-		defer conn.Close(t)
-		conn.Query(migration.Lines[0]).
-			Reply("CREATE SCHEMA").
-			Query(history.INSERT_MIGRATION_VERSION, "0", "", migration.Lines).
-			Reply("INSERT 0 1")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
-		// Run test
-		err = migration.ExecBatch(context.Background(), mock)
-		// Check error
-		assert.NoError(t, err)
-	})
-
-	t.Run("throws error on insert failure", func(t *testing.T) {
-		migration := MigrationFile{
-			Lines:   []string{"create schema public"},
-			Version: "0",
-		}
-		// Setup mock postgres
-		conn := pgtest.NewConn()
-		defer conn.Close(t)
-		conn.Query(migration.Lines[0]).
-			ReplyError(pgerrcode.DuplicateSchema, `schema "public" already exists`).
-			Query(history.INSERT_MIGRATION_VERSION, "0", "", fmt.Sprintf("{%s}", migration.Lines[0])).
-			Reply("INSERT 0 1")
-		// Connect to mock via text protocol
-		ctx := context.Background()
-		mock, err := utils.ConnectByConfig(ctx, dbConfig, conn.Intercept, func(cc *pgx.ConnConfig) {
-			cc.PreferSimpleProtocol = true
-		})
-		require.NoError(t, err)
-		defer mock.Close(ctx)
-		// Run test
-		err = migration.ExecBatch(context.Background(), mock)
-		// Check error
-		assert.ErrorContains(t, err, "ERROR: schema \"public\" already exists (SQLSTATE 42P06)")
-		assert.ErrorContains(t, err, "At statement 0: create schema public")
-	})
-}
diff --git a/internal/migration/squash/squash_test.go b/internal/migration/squash/squash_test.go
index fd810683..6cdc1712 100644
--- a/internal/migration/squash/squash_test.go
+++ b/internal/migration/squash/squash_test.go
@@ -22,12 +22,13 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/db/start"
-	"github.com/supabase/cli/internal/migration/history"
 	"github.com/supabase/cli/internal/migration/repair"
 	"github.com/supabase/cli/internal/testing/apitest"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
+	"github.com/supabase/cli/internal/testing/helper"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
@@ -81,12 +82,12 @@ func TestSquashCommand(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(sql).
+		helper.MockMigrationHistory(conn).
+			Query(sql).
 			Reply("CREATE SCHEMA").
-			Query(history.INSERT_MIGRATION_VERSION, "0", "init", []string{sql}).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "init", []string{sql}).
 			Reply("INSERT 0 1").
-			Query(history.INSERT_MIGRATION_VERSION, "1", "target", nil).
+			Query(migration.INSERT_MIGRATION_VERSION, "1", "target", nil).
 			Reply("INSERT 0 1")
 		// Run test
 		err := Run(context.Background(), "", pgconn.Config{
@@ -114,8 +115,8 @@ func TestSquashCommand(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(fmt.Sprintf("DELETE FROM supabase_migrations.schema_migrations WHERE version <=  '0' ;INSERT INTO supabase_migrations.schema_migrations(version, name, statements) VALUES( '0' ,  'init' ,  '{%s}' )", sql)).
+		helper.MockMigrationHistory(conn).
+			Query(fmt.Sprintf("DELETE FROM supabase_migrations.schema_migrations WHERE version <=  '0' ;INSERT INTO supabase_migrations.schema_migrations(version, name, statements) VALUES( '0' ,  'init' ,  '{%s}' )", sql)).
 			Reply("INSERT 0 1")
 		// Run test
 		err := Run(context.Background(), "0", dbConfig, fsys, conn.Intercept, func(cc *pgx.ConnConfig) {
@@ -304,10 +305,10 @@ func TestSquashMigrations(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(sql).
+		helper.MockMigrationHistory(conn).
+			Query(sql).
 			Reply("CREATE SCHEMA").
-			Query(history.INSERT_MIGRATION_VERSION, "0", "init", []string{sql}).
+			Query(migration.INSERT_MIGRATION_VERSION, "0", "init", []string{sql}).
 			Reply("INSERT 0 1")
 		// Run test
 		err := squashMigrations(context.Background(), []string{filepath.Base(path)}, afero.NewReadOnlyFs(fsys), conn.Intercept)
@@ -331,8 +332,8 @@ func TestBaselineMigration(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(fmt.Sprintf("DELETE FROM supabase_migrations.schema_migrations WHERE version <=  '0' ;INSERT INTO supabase_migrations.schema_migrations(version, name, statements) VALUES( '0' ,  'init' ,  '{%s}' )", sql)).
+		helper.MockMigrationHistory(conn).
+			Query(fmt.Sprintf("DELETE FROM supabase_migrations.schema_migrations WHERE version <=  '0' ;INSERT INTO supabase_migrations.schema_migrations(version, name, statements) VALUES( '0' ,  'init' ,  '{%s}' )", sql)).
 			Reply("INSERT 0 1")
 		// Run test
 		err := baselineMigrations(context.Background(), dbConfig, "", fsys, conn.Intercept, func(cc *pgx.ConnConfig) {
@@ -359,8 +360,8 @@ func TestBaselineMigration(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
-		conn.Query(fmt.Sprintf("DELETE FROM supabase_migrations.schema_migrations WHERE version <=  '%[1]s' ;INSERT INTO supabase_migrations.schema_migrations(version, name, statements) VALUES( '%[1]s' ,  'init' ,  null )", "0")).
+		helper.MockMigrationHistory(conn).
+			Query(fmt.Sprintf("DELETE FROM supabase_migrations.schema_migrations WHERE version <=  '%[1]s' ;INSERT INTO supabase_migrations.schema_migrations(version, name, statements) VALUES( '%[1]s' ,  'init' ,  null )", "0")).
 			ReplyError(pgerrcode.InsufficientPrivilege, "permission denied for relation supabase_migrations")
 		// Run test
 		err := baselineMigrations(context.Background(), dbConfig, "0", fsys, conn.Intercept, func(cc *pgx.ConnConfig) {
@@ -376,7 +377,7 @@ func TestBaselineMigration(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		pgtest.MockMigrationHistory(conn)
+		helper.MockMigrationHistory(conn)
 		// Run test
 		err := baselineMigrations(context.Background(), dbConfig, "0", fsys, conn.Intercept)
 		// Check error
diff --git a/internal/migration/up/up_test.go b/internal/migration/up/up_test.go
index c82da0b6..9fbdf328 100644
--- a/internal/migration/up/up_test.go
+++ b/internal/migration/up/up_test.go
@@ -6,14 +6,13 @@ import (
 	"path/filepath"
 	"testing"
 
-	"github.com/jackc/pgconn"
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/supabase/cli/internal/migration/list"
 	"github.com/supabase/cli/internal/testing/fstest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 func TestPendingMigrations(t *testing.T) {
@@ -33,15 +32,10 @@ func TestPendingMigrations(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 2", []interface{}{"20221201000000"}, []interface{}{"20221201000001"})
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		pending, err := GetPendingMigrations(ctx, false, mock, fsys)
+		pending, err := GetPendingMigrations(context.Background(), false, conn.MockClient(t), fsys)
 		// Check error
 		assert.NoError(t, err)
 		assert.ElementsMatch(t, files[2:], pending)
@@ -53,15 +47,10 @@ func TestPendingMigrations(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 0")
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		_, err = GetPendingMigrations(ctx, false, mock, fsys)
+		_, err := GetPendingMigrations(context.Background(), false, conn.MockClient(t), fsys)
 		// Check error
 		assert.ErrorIs(t, err, os.ErrPermission)
 	})
@@ -72,17 +61,12 @@ func TestPendingMigrations(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 1", []interface{}{"0"})
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		_, err = GetPendingMigrations(ctx, false, mock, fsys)
+		_, err := GetPendingMigrations(context.Background(), false, conn.MockClient(t), fsys)
 		// Check error
-		assert.ErrorIs(t, err, errMissingLocal)
+		assert.ErrorIs(t, err, migration.ErrMissingLocal)
 		assert.Contains(t, utils.CmdSuggestion, "supabase migration repair --status reverted 0")
 	})
 
@@ -97,17 +81,12 @@ func TestPendingMigrations(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 1", []interface{}{"1"})
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		_, err = GetPendingMigrations(ctx, false, mock, fsys)
+		_, err := GetPendingMigrations(context.Background(), false, conn.MockClient(t), fsys)
 		// Check error
-		assert.ErrorIs(t, err, errMissingRemote)
+		assert.ErrorIs(t, err, migration.ErrMissingRemote)
 	})
 }
 
@@ -128,15 +107,10 @@ func TestIgnoreVersionMismatch(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 2", []interface{}{"20221201000000"}, []interface{}{"20221201000002"})
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		pending, err := GetPendingMigrations(ctx, true, mock, fsys)
+		pending, err := GetPendingMigrations(context.Background(), true, conn.MockClient(t), fsys)
 		// Check error
 		assert.NoError(t, err)
 		assert.ElementsMatch(t, []string{files[1], files[3]}, pending)
@@ -158,17 +132,12 @@ func TestIgnoreVersionMismatch(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 2", []interface{}{"20221201000002"}, []interface{}{"20221201000004"})
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		_, err = GetPendingMigrations(ctx, true, mock, fsys)
+		_, err := GetPendingMigrations(context.Background(), true, conn.MockClient(t), fsys)
 		// Check error
-		assert.ErrorIs(t, err, errMissingLocal)
+		assert.ErrorIs(t, err, migration.ErrMissingLocal)
 		assert.Contains(t, utils.CmdSuggestion, "supabase migration repair --status reverted 20221201000004")
 	})
 
@@ -186,7 +155,7 @@ func TestIgnoreVersionMismatch(t *testing.T) {
 		// Setup mock postgres
 		conn := pgtest.NewConn()
 		defer conn.Close(t)
-		conn.Query(list.LIST_MIGRATION_VERSION).
+		conn.Query(migration.LIST_MIGRATION_VERSION).
 			Reply("SELECT 5",
 				[]interface{}{"20221201000000"},
 				[]interface{}{"20221201000001"},
@@ -194,15 +163,10 @@ func TestIgnoreVersionMismatch(t *testing.T) {
 				[]interface{}{"20221201000003"},
 				[]interface{}{"20221201000004"},
 			)
-		// Connect to mock
-		ctx := context.Background()
-		mock, err := utils.ConnectLocalPostgres(ctx, pgconn.Config{Port: 5432}, conn.Intercept)
-		require.NoError(t, err)
-		defer mock.Close(ctx)
 		// Run test
-		_, err = GetPendingMigrations(ctx, true, mock, fsys)
+		_, err := GetPendingMigrations(context.Background(), true, conn.MockClient(t), fsys)
 		// Check error
-		assert.ErrorIs(t, err, errMissingLocal)
+		assert.ErrorIs(t, err, migration.ErrMissingLocal)
 		assert.Contains(t, utils.CmdSuggestion, "supabase migration repair --status reverted 20221201000001 20221201000003 20221201000004")
 	})
 }
diff --git a/internal/start/start_test.go b/internal/start/start_test.go
index a7404091..4b4b0eaa 100644
--- a/internal/start/start_test.go
+++ b/internal/start/start_test.go
@@ -17,9 +17,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/testing/apitest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils"
 	"github.com/supabase/cli/pkg/config"
+	"github.com/supabase/cli/pkg/pgtest"
 	"github.com/supabase/cli/pkg/storage"
 )
 
diff --git a/internal/testing/apitest/helper.go b/internal/testing/apitest/docker.go
similarity index 100%
rename from internal/testing/apitest/helper.go
rename to internal/testing/apitest/docker.go
diff --git a/internal/testing/helper/history.go b/internal/testing/helper/history.go
new file mode 100644
index 00000000..b73d267e
--- /dev/null
+++ b/internal/testing/helper/history.go
@@ -0,0 +1,19 @@
+package helper
+
+import (
+	"github.com/supabase/cli/pkg/migration"
+	"github.com/supabase/cli/pkg/pgtest"
+)
+
+func MockMigrationHistory(conn *pgtest.MockConn) *pgtest.MockConn {
+	conn.Query(migration.SET_LOCK_TIMEOUT).
+		Query(migration.CREATE_VERSION_SCHEMA).
+		Reply("CREATE SCHEMA").
+		Query(migration.CREATE_VERSION_TABLE).
+		Reply("CREATE TABLE").
+		Query(migration.ADD_STATEMENTS_COLUMN).
+		Reply("ALTER TABLE").
+		Query(migration.ADD_NAME_COLUMN).
+		Reply("ALTER TABLE")
+	return conn
+}
diff --git a/internal/testing/pgtest/helper.go b/internal/testing/pgtest/helper.go
deleted file mode 100644
index c8734636..00000000
--- a/internal/testing/pgtest/helper.go
+++ /dev/null
@@ -1,17 +0,0 @@
-package pgtest
-
-import (
-	"github.com/supabase/cli/internal/migration/history"
-)
-
-func MockMigrationHistory(conn *MockConn) {
-	conn.Query(history.SET_LOCK_TIMEOUT).
-		Query(history.CREATE_VERSION_SCHEMA).
-		Reply("CREATE SCHEMA").
-		Query(history.CREATE_VERSION_TABLE).
-		Reply("CREATE TABLE").
-		Query(history.ADD_STATEMENTS_COLUMN).
-		Reply("ALTER TABLE").
-		Query(history.ADD_NAME_COLUMN).
-		Reply("ALTER TABLE")
-}
diff --git a/internal/utils/connect_test.go b/internal/utils/connect_test.go
index 585d44fe..4ea76c12 100644
--- a/internal/utils/connect_test.go
+++ b/internal/utils/connect_test.go
@@ -13,8 +13,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/supabase/cli/internal/testing/apitest"
-	"github.com/supabase/cli/internal/testing/pgtest"
 	"github.com/supabase/cli/internal/utils/cloudflare"
+	"github.com/supabase/cli/pkg/pgtest"
 )
 
 var dbConfig = pgconn.Config{
diff --git a/pkg/migration/apply_test.go b/pkg/migration/apply_test.go
new file mode 100644
index 00000000..65a75028
--- /dev/null
+++ b/pkg/migration/apply_test.go
@@ -0,0 +1,180 @@
+package migration
+
+import (
+	"context"
+	"embed"
+	"os"
+	"testing"
+	fs "testing/fstest"
+
+	"github.com/jackc/pgerrcode"
+	"github.com/stretchr/testify/assert"
+	"github.com/supabase/cli/pkg/pgtest"
+)
+
+func TestPendingMigrations(t *testing.T) {
+	t.Run("finds pending migrations", func(t *testing.T) {
+		local := []string{
+			"20221201000000_test.sql",
+			"20221201000001_test.sql",
+			"20221201000002_test.sql",
+			"20221201000003_test.sql",
+		}
+		remote := []string{
+			"20221201000000",
+			"20221201000001",
+		}
+		// Run test
+		pending, err := FindPendingMigrations(local, remote)
+		// Check error
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, local[2:], pending)
+	})
+
+	t.Run("throws error on missing local migration", func(t *testing.T) {
+		local := []string{}
+		remote := []string{"0"}
+		// Run test
+		pending, err := FindPendingMigrations(local, remote)
+		// Check error
+		assert.ErrorIs(t, err, ErrMissingLocal)
+		assert.ElementsMatch(t, remote, pending)
+	})
+
+	t.Run("throws error on missing remote version", func(t *testing.T) {
+		local := []string{
+			"0_test.sql",
+			"1_test.sql",
+		}
+		remote := []string{"1"}
+		// Run test
+		pending, err := FindPendingMigrations(local, remote)
+		// Check error
+		assert.ErrorIs(t, err, ErrMissingRemote)
+		assert.ElementsMatch(t, local[:1], pending)
+	})
+
+	t.Run("throws error on out-of-order remote migrations", func(t *testing.T) {
+		local := []string{
+			"20221201000000_test.sql",
+			"20221201000001_test.sql",
+			"20221201000002_test.sql",
+			"20221201000003_test.sql",
+			"20221201000004_test.sql",
+		}
+		remote := []string{
+			"20221201000002",
+			"20221201000004",
+		}
+		// Run test
+		missing, err := FindPendingMigrations(local, remote)
+		// Check error
+		assert.ErrorIs(t, err, ErrMissingRemote)
+		assert.ElementsMatch(t, []string{local[0], local[1], local[3]}, missing)
+	})
+
+	t.Run("throws error on out-of-order local migrations", func(t *testing.T) {
+		local := []string{
+			"20221201000000_test.sql",
+			"20221201000002_test.sql",
+		}
+		remote := []string{
+			"20221201000000",
+			"20221201000001",
+			"20221201000002",
+			"20221201000003",
+			"20221201000004",
+		}
+		// Run test
+		missing, err := FindPendingMigrations(local, remote)
+		// Check error
+		assert.ErrorIs(t, err, ErrMissingLocal)
+		assert.ElementsMatch(t, []string{remote[1], remote[3], remote[4]}, missing)
+	})
+}
+
+var (
+	//go:embed testdata
+	testMigrations embed.FS
+	//go:embed testdata/0_schema.sql
+	testSchema string
+)
+
+func TestApplyMigrations(t *testing.T) {
+	pending := []string{"testdata/0_schema.sql"}
+
+	t.Run("applies migrations and appends history", func(t *testing.T) {
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		mockMigrationHistory(conn).
+			Query(testSchema).
+			Reply("CREATE SCHEMA").
+			Query(INSERT_MIGRATION_VERSION, "0", "schema", []string{testSchema}).
+			Reply("INSERT 0 1")
+		// Run test
+		err := ApplyMigrations(context.Background(), pending, conn.MockClient(t), testMigrations)
+		// Check error
+		assert.NoError(t, err)
+	})
+
+	t.Run("throws error on history failure", func(t *testing.T) {
+		// Setup in-memory fs
+		fsys := fs.MapFS{}
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(SET_LOCK_TIMEOUT).
+			Query(CREATE_VERSION_SCHEMA).
+			Reply("CREATE SCHEMA").
+			Query(CREATE_VERSION_TABLE).
+			ReplyError(pgerrcode.InsufficientPrivilege, "permission denied for relation supabase_migrations").
+			Query(ADD_STATEMENTS_COLUMN).
+			Query(ADD_NAME_COLUMN)
+		// Run test
+		err := ApplyMigrations(context.Background(), pending, conn.MockClient(t), fsys)
+		// Check error
+		assert.ErrorContains(t, err, "ERROR: permission denied for relation supabase_migrations (SQLSTATE 42501)")
+	})
+
+	t.Run("throws error on missing file", func(t *testing.T) {
+		// Setup in-memory fs
+		fsys := fs.MapFS{}
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		mockMigrationHistory(conn)
+		// Run test
+		err := ApplyMigrations(context.Background(), pending, conn.MockClient(t), fsys)
+		// Check error
+		assert.ErrorIs(t, err, os.ErrNotExist)
+	})
+
+	t.Run("throws error on missing table", func(t *testing.T) {
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		mockMigrationHistory(conn).
+			Query(testSchema).
+			ReplyError(pgerrcode.UndefinedTable, `relation "supabase_migrations.schema_migrations" does not exist`).
+			Query(INSERT_MIGRATION_VERSION, "0", "schema", []string{testSchema}).
+			Reply("INSERT 0 1")
+		// Run test
+		err := ApplyMigrations(context.Background(), pending, conn.MockClient(t), testMigrations)
+		// Check error
+		assert.ErrorContains(t, err, `ERROR: relation "supabase_migrations.schema_migrations" does not exist (SQLSTATE 42P01)`)
+	})
+}
+
+func mockMigrationHistory(conn *pgtest.MockConn) *pgtest.MockConn {
+	conn.Query(SET_LOCK_TIMEOUT).
+		Query(CREATE_VERSION_SCHEMA).
+		Reply("CREATE SCHEMA").
+		Query(CREATE_VERSION_TABLE).
+		Reply("CREATE TABLE").
+		Query(ADD_STATEMENTS_COLUMN).
+		Reply("ALTER TABLE").
+		Query(ADD_NAME_COLUMN).
+		Reply("ALTER TABLE")
+	return conn
+}
diff --git a/pkg/migration/file_test.go b/pkg/migration/file_test.go
new file mode 100644
index 00000000..a38238f8
--- /dev/null
+++ b/pkg/migration/file_test.go
@@ -0,0 +1,80 @@
+package migration
+
+import (
+	"bufio"
+	"context"
+	"strings"
+	"testing"
+	fs "testing/fstest"
+
+	"github.com/jackc/pgerrcode"
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/assert"
+	"github.com/supabase/cli/pkg/parser"
+	"github.com/supabase/cli/pkg/pgtest"
+)
+
+func TestMigrationFile(t *testing.T) {
+	t.Run("new from file sets max token", func(t *testing.T) {
+		// Setup in-memory fs
+		path := "20220727064247_create_table.sql"
+		query := "BEGIN; " + strings.Repeat("a", parser.MaxScannerCapacity)
+		fsys := fs.MapFS{
+			path: &fs.MapFile{Data: []byte(query)},
+		}
+		// Run test
+		migration, err := NewMigrationFromFile(path, fsys)
+		// Check error
+		assert.NoError(t, err)
+		assert.Len(t, migration.Statements, 2)
+		assert.Equal(t, "20220727064247", migration.Version)
+	})
+
+	t.Run("new from reader errors on max token", func(t *testing.T) {
+		viper.Reset()
+		sql := "\tBEGIN; " + strings.Repeat("a", parser.MaxScannerCapacity)
+		// Run test
+		migration, err := NewMigrationFromReader(strings.NewReader(sql))
+		// Check error
+		assert.ErrorIs(t, err, bufio.ErrTooLong)
+		assert.ErrorContains(t, err, "After statement 1: \tBEGIN;")
+		assert.Nil(t, migration)
+	})
+
+	t.Run("encodes statements in binary format", func(t *testing.T) {
+		migration := MigrationFile{
+			Statements: []string{"create schema public"},
+			Version:    "0",
+		}
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(migration.Statements[0]).
+			Reply("CREATE SCHEMA").
+			Query(INSERT_MIGRATION_VERSION, "0", "", migration.Statements).
+			Reply("INSERT 0 1")
+		// Run test
+		err := migration.ExecBatch(context.Background(), conn.MockClient(t))
+		// Check error
+		assert.NoError(t, err)
+	})
+
+	t.Run("throws error on insert failure", func(t *testing.T) {
+		migration := MigrationFile{
+			Statements: []string{"create schema public"},
+			Version:    "0",
+		}
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(migration.Statements[0]).
+			ReplyError(pgerrcode.DuplicateSchema, `schema "public" already exists`).
+			Query(INSERT_MIGRATION_VERSION, "0", "", migration.Statements).
+			Reply("INSERT 0 1")
+		// Run test
+		err := migration.ExecBatch(context.Background(), conn.MockClient(t))
+		// Check error
+		assert.ErrorContains(t, err, "ERROR: schema \"public\" already exists (SQLSTATE 42P06)")
+		assert.ErrorContains(t, err, "At statement 0: create schema public")
+	})
+}
diff --git a/pkg/migration/list_test.go b/pkg/migration/list_test.go
new file mode 100644
index 00000000..80d09fd4
--- /dev/null
+++ b/pkg/migration/list_test.go
@@ -0,0 +1,92 @@
+package migration
+
+import (
+	"context"
+	"testing"
+	fs "testing/fstest"
+
+	"github.com/jackc/pgerrcode"
+	"github.com/stretchr/testify/assert"
+	"github.com/supabase/cli/pkg/pgtest"
+)
+
+func TestRemoteMigrations(t *testing.T) {
+	t.Run("loads migration versions", func(t *testing.T) {
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(LIST_MIGRATION_VERSION).
+			Reply("SELECT 1", []interface{}{"20220727064247"})
+		// Run test
+		versions, err := ListRemoteMigrations(context.Background(), conn.MockClient(t))
+		// Check error
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []string{"20220727064247"}, versions)
+	})
+
+	t.Run("loads empty migrations on missing table", func(t *testing.T) {
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(LIST_MIGRATION_VERSION).
+			ReplyError(pgerrcode.UndefinedTable, "relation \"supabase_migrations.schema_migrations\" does not exist")
+		// Run test
+		versions, err := ListRemoteMigrations(context.Background(), conn.MockClient(t))
+		// Check error
+		assert.NoError(t, err)
+		assert.Empty(t, versions)
+	})
+
+	t.Run("throws error on invalid row", func(t *testing.T) {
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(LIST_MIGRATION_VERSION).
+			Reply("SELECT 1", []interface{}{})
+		// Run test
+		_, err := ListRemoteMigrations(context.Background(), conn.MockClient(t))
+		// Check error
+		assert.ErrorContains(t, err, "number of field descriptions must equal number of destinations, got 0 and 1")
+	})
+}
+
+func TestLocalMigrations(t *testing.T) {
+	t.Run("loads migration versions", func(t *testing.T) {
+		// Setup in-memory fs
+		files := []string{
+			"20220727064246_test.sql",
+			"20220727064248_test.sql",
+		}
+		fsys := fs.MapFS{}
+		for _, name := range files {
+			fsys[name] = &fs.MapFile{}
+		}
+		// Run test
+		versions, err := ListLocalMigrations(".", fsys)
+		// Check error
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, files, versions)
+	})
+
+	t.Run("ignores outdated and invalid files", func(t *testing.T) {
+		// Setup in-memory fs
+		fsys := fs.MapFS{
+			"20211208000000_init.sql":   &fs.MapFile{},
+			"20211208000001_invalid.ts": &fs.MapFile{},
+		}
+		// Run test
+		versions, err := ListLocalMigrations(".", fsys)
+		// Check error
+		assert.NoError(t, err)
+		assert.Empty(t, versions)
+	})
+
+	t.Run("throws error on open failure", func(t *testing.T) {
+		// Setup in-memory fs
+		fsys := fs.MapFS{"migrations": &fs.MapFile{}}
+		// Run test
+		_, err := ListLocalMigrations("migrations", fsys)
+		// Check error
+		assert.ErrorContains(t, err, "failed to read directory:")
+	})
+}
diff --git a/pkg/migration/seed_test.go b/pkg/migration/seed_test.go
new file mode 100644
index 00000000..cf88a818
--- /dev/null
+++ b/pkg/migration/seed_test.go
@@ -0,0 +1,93 @@
+package migration
+
+import (
+	"context"
+	_ "embed"
+	"os"
+	"testing"
+	fs "testing/fstest"
+
+	"github.com/jackc/pgerrcode"
+	"github.com/stretchr/testify/assert"
+	"github.com/supabase/cli/pkg/pgtest"
+)
+
+//go:embed testdata/seed.sql
+var testSeed string
+
+func TestSeedData(t *testing.T) {
+	pending := []string{"testdata/seed.sql"}
+
+	t.Run("seeds from file", func(t *testing.T) {
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(testSeed).
+			Reply("INSERT 0 1")
+		// Run test
+		err := SeedData(context.Background(), pending, conn.MockClient(t), testMigrations)
+		// Check error
+		assert.NoError(t, err)
+	})
+
+	t.Run("throws error on missing file", func(t *testing.T) {
+		// Setup in-memory fs
+		fsys := fs.MapFS{}
+		// Run test
+		err := SeedData(context.Background(), pending, nil, fsys)
+		// Check error
+		assert.ErrorIs(t, err, os.ErrNotExist)
+	})
+
+	t.Run("throws error on insert failure", func(t *testing.T) {
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(testSeed).
+			ReplyError(pgerrcode.NotNullViolation, `null value in column "age" of relation "employees"`)
+		// Run test
+		err := SeedData(context.Background(), pending, conn.MockClient(t), testMigrations)
+		// Check error
+		assert.ErrorContains(t, err, `ERROR: null value in column "age" of relation "employees" (SQLSTATE 23502)`)
+	})
+}
+
+//go:embed testdata/globals.sql
+var testGlobals string
+
+func TestSeedGlobals(t *testing.T) {
+	pending := []string{"testdata/globals.sql"}
+
+	t.Run("seeds from file", func(t *testing.T) {
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(testGlobals).
+			Reply("CREATE ROLE")
+		// Run test
+		err := SeedGlobals(context.Background(), pending, conn.MockClient(t), testMigrations)
+		// Check error
+		assert.NoError(t, err)
+	})
+
+	t.Run("throws error on missing file", func(t *testing.T) {
+		// Setup in-memory fs
+		fsys := fs.MapFS{}
+		// Run test
+		err := SeedGlobals(context.Background(), pending, nil, fsys)
+		// Check error
+		assert.ErrorIs(t, err, os.ErrNotExist)
+	})
+
+	t.Run("throws error on create failure", func(t *testing.T) {
+		// Setup mock postgres
+		conn := pgtest.NewConn()
+		defer conn.Close(t)
+		conn.Query(testGlobals).
+			ReplyError(pgerrcode.InvalidCatalogName, `database "postgres" does not exist`)
+		// Run test
+		err := SeedGlobals(context.Background(), pending, conn.MockClient(t), testMigrations)
+		// Check error
+		assert.ErrorContains(t, err, `ERROR: database "postgres" does not exist (SQLSTATE 3D000)`)
+	})
+}
diff --git a/pkg/migration/testdata/0_schema.sql b/pkg/migration/testdata/0_schema.sql
new file mode 100644
index 00000000..ba3138ab
--- /dev/null
+++ b/pkg/migration/testdata/0_schema.sql
@@ -0,0 +1,1 @@
+create schema private
\ No newline at end of file
diff --git a/pkg/migration/testdata/globals.sql b/pkg/migration/testdata/globals.sql
new file mode 100644
index 00000000..0e17ccd4
--- /dev/null
+++ b/pkg/migration/testdata/globals.sql
@@ -0,0 +1,1 @@
+create role developer_role with login password 'developer'
\ No newline at end of file
diff --git a/pkg/migration/testdata/seed.sql b/pkg/migration/testdata/seed.sql
new file mode 100644
index 00000000..e48c726c
--- /dev/null
+++ b/pkg/migration/testdata/seed.sql
@@ -0,0 +1,1 @@
+INSERT INTO employees(name) VALUES ('Alice')
\ No newline at end of file
diff --git a/internal/testing/pgtest/mock.go b/pkg/pgtest/mock.go
similarity index 91%
rename from internal/testing/pgtest/mock.go
rename to pkg/pgtest/mock.go
index 0cfc7ad4..b73ad255 100644
--- a/internal/testing/pgtest/mock.go
+++ b/pkg/pgtest/mock.go
@@ -12,10 +12,13 @@ import (
 	"github.com/jackc/pgproto3/v2"
 	"github.com/jackc/pgtype"
 	"github.com/jackc/pgx/v4"
+	"github.com/supabase/cli/pkg/pgxv5"
 	"google.golang.org/grpc/test/bufconn"
 )
 
 type MockConn struct {
+	client *pgx.Conn
+
 	// Duplex server listener backed by in-memory buffer
 	server *bufconn.Listener
 
@@ -234,14 +237,33 @@ func (r *MockConn) ReplyError(code, message string) *MockConn {
 }
 
 func (r *MockConn) Close(t *testing.T) {
+	if r.client != nil {
+		if err := r.client.Close(context.Background()); err != nil {
+			t.Errorf("failed to close client: %v", err)
+		}
+	}
 	for err := range r.errChan {
-		t.Errorf("pgmock error: %v", err)
+		t.Errorf("pgmock error:\n%v", err)
 	}
 	if err := r.server.Close(); err != nil {
-		t.Fatalf("failed to close: %v", err)
+		t.Fatalf("failed to close server: %v", err)
 	}
 }
 
+func (r *MockConn) MockClient(t *testing.T, opts ...func(*pgx.ConnConfig)) *pgx.Conn {
+	if r.client != nil {
+		return r.client
+	}
+	opts = append(opts, r.Intercept, func(cc *pgx.ConnConfig) {
+		cc.ConnectTimeout = time.Second * 2
+	})
+	var err error
+	if r.client, err = pgxv5.Connect(context.Background(), "", opts...); err != nil {
+		t.Errorf("failed to connect: %v", err)
+	}
+	return r.client
+}
+
 func NewWithStatus(status map[string]string) *MockConn {
 	const bufSize = 1024 * 1024
 	mock := MockConn{
@@ -279,7 +301,7 @@ func NewWithStatus(status map[string]string) *MockConn {
 }
 
 func NewConn() *MockConn {
-	return NewWithStatus(map[string]string{
+	status := map[string]string{
 		"application_name":              "",
 		"client_encoding":               "UTF8",
 		"DateStyle":                     "ISO, MDY",
@@ -292,5 +314,6 @@ func NewConn() *MockConn {
 		"server_version":                "14.3 (Debian 14.3-1.pgdg110+1)",
 		"standard_conforming_strings":   "on",
 		"TimeZone":                      "UTC",
-	})
+	}
+	return NewWithStatus(status)
 }
diff --git a/internal/testing/pgtest/step.go b/pkg/pgtest/step.go
similarity index 92%
rename from internal/testing/pgtest/step.go
rename to pkg/pgtest/step.go
index 86e136ee..907b6182 100644
--- a/internal/testing/pgtest/step.go
+++ b/pkg/pgtest/step.go
@@ -28,7 +28,7 @@ func (e *extendedQueryStep) Step(backend *pgproto3.Backend) error {
 	if m, ok := msg.(*pgproto3.Parse); ok {
 		want := &pgproto3.Parse{Name: m.Name, Query: e.sql, ParameterOIDs: m.ParameterOIDs}
 		if !reflect.DeepEqual(m, want) {
-			return errors.Errorf("msg => %#v, e.want => %#v", m, want)
+			return errors.Errorf("expected => %#v\nactual => %#v", want, m)
 		}
 		// Anonymous ps falls through
 		if m.Name != "" {
@@ -64,7 +64,7 @@ func (e *extendedQueryStep) Step(backend *pgproto3.Backend) error {
 			PreparedStatement:    m.PreparedStatement,
 		}
 		if !reflect.DeepEqual(m, want) {
-			return errors.Errorf("msg => %#v, e.want => %#v", msg, want)
+			return errors.Errorf("expected => %#v\nactual => %#v", want, msg)
 		}
 		e.reply.Steps = append([]pgmock.Step{
 			pgmock.ExpectMessage(&pgproto3.Describe{ObjectType: 'P'}),
@@ -82,7 +82,7 @@ func (e *extendedQueryStep) Step(backend *pgproto3.Backend) error {
 		return e.reply.Run(backend)
 	}
 
-	return errors.Errorf("msg => %#v, e.want => %#v", msg, want)
+	return errors.Errorf("expected => %#v\nactual => %#v", want, msg)
 }
 
 // Expects a SQL query in any form: simple, prepared, or anonymous.
@@ -103,7 +103,7 @@ func (e *terminateStep) Step(backend *pgproto3.Backend) error {
 		return nil
 	}
 
-	return errors.Errorf("msg => %#v, e.want => %#v", msg, &pgproto3.Terminate{})
+	return errors.Errorf("expected => %#v\nactual => %#v", &pgproto3.Terminate{}, msg)
 }
 
 func ExpectTerminate() pgmock.Step {