fix: print notices from migration statements (#3091)

* fix: print notices from migration statements

* chore: update unit tests

internal/db/diff/diff_test.go

@@ -283,7 +283,8 @@ func TestDiffDatabase(t *testing.T) {
 		// Check error
 		assert.Empty(t, diff)
 		assert.ErrorContains(t, err, `ERROR: schema "public" already exists (SQLSTATE 42P06)
-At statement 0: create schema public`)
+At statement 0:
+create schema public`)
 		assert.Empty(t, apitest.ListUnmatchedRequests())
 	})
 

internal/db/push/push_test.go

@@ -105,7 +105,7 @@ func TestMigrationPush(t *testing.T) {
 		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: "+migration.INSERT_MIGRATION_VERSION)
+		assert.ErrorContains(t, err, "At statement 0:\n"+migration.INSERT_MIGRATION_VERSION)
 	})
 }
 

pkg/migration/file.go

@@ -1,13 +1,15 @@
 package migration
 
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
 	"io"
 	"io/fs"
 	"path/filepath"
 	"regexp"
+	"strings"
 
 	"github.com/go-errors/errors"
 	"github.com/jackc/pgconn"
@@ -86,11 +88,32 @@ func (m *MigrationFile) ExecBatch(ctx context.Context, conn *pgx.Conn) error {
 		if i < len(m.Statements) {
 			stat = m.Statements[i]
 		}
-		return errors.Errorf("%w\nAt statement %d: %s", err, i, stat)
+		var pgErr *pgconn.PgError
+		if errors.As(err, &pgErr) {
+			stat = markError(stat, int(pgErr.Position))
+		}
+		return errors.Errorf("%w\nAt statement %d:\n%s", err, i, stat)
 	}
 	return nil
 }
 
+func markError(stat string, pos int) string {
+	lines := strings.Split(stat, "\n")
+	for j, r := range lines {
+		if c := len(r); pos > c {
+			pos -= c + 1
+			continue
+		}
+		// Show a caret below the error position
+		if pos > 0 {
+			caret := append(bytes.Repeat([]byte{' '}, pos-1), '^')
+			lines = append(lines[:j+1], string(caret))
+		}
+		break
+	}
+	return strings.Join(lines, "\n")
+}
+
 func (m *MigrationFile) insertVersionSQL(conn *pgx.Conn, batch *pgconn.Batch) error {
 	value := pgtype.TextArray{}
 	if err := value.Set(m.Statements); err != nil {

pkg/migration/file_test.go

@@ -75,6 +75,6 @@ func TestMigrationFile(t *testing.T) {
 		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")
+		assert.ErrorContains(t, err, "At statement 0:\ncreate schema public")
 	})
 }

pkg/migration/history.go

@@ -10,7 +10,7 @@ import (
 )
 
 const (
-	SET_LOCK_TIMEOUT         = "SET LOCAL lock_timeout = '4s'"
+	SET_LOCK_TIMEOUT         = "SET lock_timeout = '4s'"
 	CREATE_VERSION_SCHEMA    = "CREATE SCHEMA IF NOT EXISTS supabase_migrations"
 	CREATE_VERSION_TABLE     = "CREATE TABLE IF NOT EXISTS supabase_migrations.schema_migrations (version text NOT NULL PRIMARY KEY)"
 	ADD_STATEMENTS_COLUMN    = "ALTER TABLE supabase_migrations.schema_migrations ADD COLUMN IF NOT EXISTS statements text[]"

pkg/pgxv5/connect.go

@@ -2,8 +2,11 @@ package pgxv5
 
 import (
 	"context"
+	"fmt"
+	"os"
 
 	"github.com/go-errors/errors"
+	"github.com/jackc/pgconn"
 	"github.com/jackc/pgx/v4"
 )
 
@@ -14,6 +17,9 @@ func Connect(ctx context.Context, connString string, options ...func(*pgx.ConnCo
 	if err != nil {
 		return nil, errors.Errorf("failed to parse connection string: %w", err)
 	}
+	config.OnNotice = func(pc *pgconn.PgConn, n *pgconn.Notice) {
+		fmt.Fprintf(os.Stderr, "%s (%s): %s\n", n.Severity, n.Code, n.Message)
+	}
 	// Apply config overrides
 	for _, op := range options {
 		op(config)