23749185424

Committed 30 Mar 2026 02:09PM UTC coverage: 54.481% (+0.05%) from 54.432%

Build Type

github

Committed by Jim Wordelman

Commit Message

fix: handle host query parameter for unix socket connections in postgres driver

When host is specified as a query parameter (e.g., ?host=/var/run/postgresql
for unix socket connections) and the URL has no host in the authority section,
lib/pq silently connects to localhost:5432 instead. Convert to key-value
connection string so the host parameter is reliably used.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Pull Request Pull Request #1378: fix: handle host query parameter for unix socket connections

Coverage Stats

12 of 14 new or added lines in 1 file covered. (85.71%)

4389 of 8056 relevant lines covered (54.48%)

49.13 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.87

/database/postgres/postgres.go

//go:build go1.9

package postgres

import (
        "context"
        "database/sql"
        "errors"
        "fmt"
        "io"
        nurl "net/url"
        "regexp"
        "strconv"
        "strings"
        "sync/atomic"
        "time"

        "github.com/golang-migrate/migrate/v4"
        "github.com/golang-migrate/migrate/v4/database"
        "github.com/golang-migrate/migrate/v4/database/multistmt"
        "github.com/lib/pq"
)

func init() {
        db := Postgres{}
        database.Register("postgres", &db)
        database.Register("postgresql", &db)
}

var (
        multiStmtDelimiter = []byte(";")

        DefaultMigrationsTable       = "schema_migrations"
        DefaultMultiStatementMaxSize = 10 * 1 << 20 // 10 MB
)

var (
        ErrNilConfig      = fmt.Errorf("no config")
        ErrNoDatabaseName = fmt.Errorf("no database name")
        ErrNoSchema       = fmt.Errorf("no schema")
        ErrDatabaseDirty  = fmt.Errorf("database is dirty")
)

type Config struct {
        MigrationsTable       string
        MigrationsTableQuoted bool
        MultiStatementEnabled bool
        DatabaseName          string
        SchemaName            string
        migrationsSchemaName  string
        migrationsTableName   string
        StatementTimeout      time.Duration
        MultiStatementMaxSize int
}

type Postgres struct {
        // Locking and unlocking need to use the same connection
        conn     *sql.Conn
        db       *sql.DB
        isLocked atomic.Bool

        // Open and WithInstance need to guarantee that config is never nil
        config *Config
}

func WithConnection(ctx context.Context, conn *sql.Conn, config *Config) (*Postgres, error) {
        if config == nil {
                return nil, ErrNilConfig
        }

        if err := conn.PingContext(ctx); err != nil {
                return nil, err
        }

        if config.DatabaseName == "" {
                query := `SELECT CURRENT_DATABASE()`
                var databaseName string
                if err := conn.QueryRowContext(ctx, query).Scan(&databaseName); err != nil {
                        return nil, &database.Error{OrigErr: err, Query: []byte(query)}
                }

                if len(databaseName) == 0 {
                        return nil, ErrNoDatabaseName
                }

                config.DatabaseName = databaseName
        }

        if config.SchemaName == "" {
                query := `SELECT CURRENT_SCHEMA()`
                var schemaName sql.NullString
                if err := conn.QueryRowContext(ctx, query).Scan(&schemaName); err != nil {
                        return nil, &database.Error{OrigErr: err, Query: []byte(query)}
                }

                if !schemaName.Valid {
                        return nil, ErrNoSchema
                }

                config.SchemaName = schemaName.String
        }

        if len(config.MigrationsTable) == 0 {
                config.MigrationsTable = DefaultMigrationsTable
        }

        config.migrationsSchemaName = config.SchemaName
        config.migrationsTableName = config.MigrationsTable
        if config.MigrationsTableQuoted {
                re := regexp.MustCompile(`"(.*?)"`)
                result := re.FindAllStringSubmatch(config.MigrationsTable, -1)
                config.migrationsTableName = result[len(result)-1][1]
                if len(result) == 2 {
                        config.migrationsSchemaName = result[0][1]
                } else if len(result) > 2 {
                        return nil, fmt.Errorf("\"%s\" MigrationsTable contains too many dot characters", config.MigrationsTable)
                }
        }

        px := &Postgres{
                conn:   conn,
                config: config,
        }

        if err := px.ensureVersionTable(); err != nil {
                return nil, err
        }

        return px, nil
}

func WithInstance(instance *sql.DB, config *Config) (database.Driver, error) {
        ctx := context.Background()

        if err := instance.Ping(); err != nil {
                return nil, err
        }

        conn, err := instance.Conn(ctx)
        if err != nil {
                return nil, err
        }

        px, err := WithConnection(ctx, conn, config)
        if err != nil {
                return nil, err
        }
        px.db = instance
        return px, nil
}

func (p *Postgres) Open(url string) (database.Driver, error) {
        purl, err := nurl.Parse(url)
        if err != nil {
                return nil, err
        }

        filteredURL := migrate.FilterCustomQuery(purl)
        connStr := filteredURL.String()

        // When host is specified as a query parameter (e.g., for unix socket
        // connections via ?host=/var/run/postgresql) and the URL has no host in
        // the authority section, convert to a key-value connection string so
        // lib/pq reliably uses the host parameter. Without this, the driver
        // may silently connect to localhost:5432 instead of the specified host.
        if purl.Host == "" && purl.Query().Get("host") != "" {
                connStr, err = pq.ParseURL(connStr)
                if err != nil {
                        return nil, err
                }
        }

        db, err := sql.Open("postgres", connStr)
        if err != nil {
                return nil, err
        }

        migrationsTable := purl.Query().Get("x-migrations-table")
        migrationsTableQuoted := false
        if s := purl.Query().Get("x-migrations-table-quoted"); len(s) > 0 {
                migrationsTableQuoted, err = strconv.ParseBool(s)
                if err != nil {
                        return nil, fmt.Errorf("unable to parse option x-migrations-table-quoted: %w", err)
                }
        }
        if (len(migrationsTable) > 0) && (migrationsTableQuoted) && ((migrationsTable[0] != '"') || (migrationsTable[len(migrationsTable)-1] != '"')) {
                return nil, fmt.Errorf("x-migrations-table must be quoted (for instance '\"migrate\".\"schema_migrations\"') when x-migrations-table-quoted is enabled, current value is: %s", migrationsTable)
        }

        statementTimeoutString := purl.Query().Get("x-statement-timeout")
        statementTimeout := 0
        if statementTimeoutString != "" {
                statementTimeout, err = strconv.Atoi(statementTimeoutString)
                if err != nil {
                        return nil, err
                }
        }

        multiStatementMaxSize := DefaultMultiStatementMaxSize
        if s := purl.Query().Get("x-multi-statement-max-size"); len(s) > 0 {
                multiStatementMaxSize, err = strconv.Atoi(s)
                if err != nil {
                        return nil, err
                }
                if multiStatementMaxSize <= 0 {
                        multiStatementMaxSize = DefaultMultiStatementMaxSize
                }
        }

        multiStatementEnabled := false
        if s := purl.Query().Get("x-multi-statement"); len(s) > 0 {
                multiStatementEnabled, err = strconv.ParseBool(s)
                if err != nil {
                        return nil, fmt.Errorf("unable to parse option x-multi-statement: %w", err)
                }
        }

        px, err := WithInstance(db, &Config{
                DatabaseName:          purl.Path,
                MigrationsTable:       migrationsTable,
                MigrationsTableQuoted: migrationsTableQuoted,
                StatementTimeout:      time.Duration(statementTimeout) * time.Millisecond,
                MultiStatementEnabled: multiStatementEnabled,
                MultiStatementMaxSize: multiStatementMaxSize,
        })

        if err != nil {
                return nil, err
        }

        return px, nil
}

func (p *Postgres) Close() error {
        connErr := p.conn.Close()
        var dbErr error
        if p.db != nil {
                dbErr = p.db.Close()
        }

        if connErr != nil || dbErr != nil {
                return fmt.Errorf("conn: %v, db: %v", connErr, dbErr)
        }
        return nil
}

// https://www.postgresql.org/docs/9.6/static/explicit-locking.html#ADVISORY-LOCKS
func (p *Postgres) Lock() error {
        return database.CasRestoreOnErr(&p.isLocked, false, true, database.ErrLocked, func() error {
                aid, err := database.GenerateAdvisoryLockId(p.config.DatabaseName, p.config.migrationsSchemaName, p.config.migrationsTableName)
                if err != nil {
                        return err
                }

                // This will wait indefinitely until the lock can be acquired.
                query := `SELECT pg_advisory_lock($1)`
                if _, err := p.conn.ExecContext(context.Background(), query, aid); err != nil {
                        return &database.Error{OrigErr: err, Err: "try lock failed", Query: []byte(query)}
                }

                return nil
        })
}

func (p *Postgres) Unlock() error {
        return database.CasRestoreOnErr(&p.isLocked, true, false, database.ErrNotLocked, func() error {
                aid, err := database.GenerateAdvisoryLockId(p.config.DatabaseName, p.config.migrationsSchemaName, p.config.migrationsTableName)
                if err != nil {
                        return err
                }

                query := `SELECT pg_advisory_unlock($1)`
                if _, err := p.conn.ExecContext(context.Background(), query, aid); err != nil {
                        return &database.Error{OrigErr: err, Query: []byte(query)}
                }
                return nil
        })
}

func (p *Postgres) Run(migration io.Reader) error {
        if p.config.MultiStatementEnabled {
                var err error
                if e := multistmt.Parse(migration, multiStmtDelimiter, p.config.MultiStatementMaxSize, func(m []byte) bool {
                        if err = p.runStatement(m); err != nil {
                                return false
                        }
                        return true
                }); e != nil {
                        return e
                }
                return err
        }
        migr, err := io.ReadAll(migration)
        if err != nil {
                return err
        }
        return p.runStatement(migr)
}

func (p *Postgres) runStatement(statement []byte) error {
        ctx := context.Background()
        if p.config.StatementTimeout != 0 {
                var cancel context.CancelFunc
                ctx, cancel = context.WithTimeout(ctx, p.config.StatementTimeout)
                defer cancel()
        }
        query := string(statement)
        if strings.TrimSpace(query) == "" {
                return nil
        }
        if _, err := p.conn.ExecContext(ctx, query); err != nil {
                if pgErr, ok := err.(*pq.Error); ok {
                        var line uint
                        var col uint
                        var lineColOK bool
                        if pgErr.Position != "" {
                                if pos, err := strconv.ParseUint(pgErr.Position, 10, 64); err == nil {
                                        line, col, lineColOK = computeLineFromPos(query, int(pos))
                                }
                        }
                        message := fmt.Sprintf("migration failed: %s", pgErr.Message)
                        if lineColOK {
                                message = fmt.Sprintf("%s (column %d)", message, col)
                        }
                        if pgErr.Detail != "" {
                                message = fmt.Sprintf("%s, %s", message, pgErr.Detail)
                        }
                        return database.Error{OrigErr: err, Err: message, Query: statement, Line: line}
                }
                return database.Error{OrigErr: err, Err: "migration failed", Query: statement}
        }
        return nil
}

func computeLineFromPos(s string, pos int) (line uint, col uint, ok bool) {
        // replace crlf with lf
        s = strings.ReplaceAll(s, "\r\n", "\n")
        // pg docs: pos uses index 1 for the first character, and positions are measured in characters not bytes
        runes := []rune(s)
        if pos > len(runes) {
                return 0, 0, false
        }
        sel := runes[:pos]
        line = uint(runesCount(sel, newLine) + 1)
        col = uint(pos - 1 - runesLastIndex(sel, newLine))
        return line, col, true
}

const newLine = '\n'

func runesCount(input []rune, target rune) int {
        var count int
        for _, r := range input {
                if r == target {
                        count++
                }
        }
        return count
}

func runesLastIndex(input []rune, target rune) int {
        for i := len(input) - 1; i >= 0; i-- {
                if input[i] == target {
                        return i
                }
        }
        return -1
}

func (p *Postgres) SetVersion(version int, dirty bool) error {
        tx, err := p.conn.BeginTx(context.Background(), &sql.TxOptions{})
        if err != nil {
                return &database.Error{OrigErr: err, Err: "transaction start failed"}
        }

        query := `TRUNCATE ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName)
        if _, err := tx.Exec(query); err != nil {
                if errRollback := tx.Rollback(); errRollback != nil {
                        err = errors.Join(err, errRollback)
                }
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        // Also re-write the schema version for nil dirty versions to prevent
        // empty schema version for failed down migration on the first migration
        // See: https://github.com/golang-migrate/migrate/issues/330
        if version >= 0 || (version == database.NilVersion && dirty) {
                query = `INSERT INTO ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` (version, dirty) VALUES ($1, $2)`
                if _, err := tx.Exec(query, version, dirty); err != nil {
                        if errRollback := tx.Rollback(); errRollback != nil {
                                err = errors.Join(err, errRollback)
                        }
                        return &database.Error{OrigErr: err, Query: []byte(query)}
                }
        }

        if err := tx.Commit(); err != nil {
                return &database.Error{OrigErr: err, Err: "transaction commit failed"}
        }

        return nil
}

func (p *Postgres) Version() (version int, dirty bool, err error) {
        query := `SELECT version, dirty FROM ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` LIMIT 1`
        err = p.conn.QueryRowContext(context.Background(), query).Scan(&version, &dirty)
        switch {
        case err == sql.ErrNoRows:
                return database.NilVersion, false, nil

        case err != nil:
                if e, ok := err.(*pq.Error); ok {
                        if e.Code.Name() == "undefined_table" {
                                return database.NilVersion, false, nil
                        }
                }
                return 0, false, &database.Error{OrigErr: err, Query: []byte(query)}

        default:
                return version, dirty, nil
        }
}

func (p *Postgres) Drop() (err error) {
        // select all tables in current schema
        query := `SELECT table_name FROM information_schema.tables WHERE table_schema=(SELECT current_schema()) AND table_type='BASE TABLE'`
        tables, err := p.conn.QueryContext(context.Background(), query)
        if err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }
        defer func() {
                if errClose := tables.Close(); errClose != nil {
                        err = errors.Join(err, errClose)
                }
        }()

        // delete one table after another
        tableNames := make([]string, 0)
        for tables.Next() {
                var tableName string
                if err := tables.Scan(&tableName); err != nil {
                        return err
                }
                if len(tableName) > 0 {
                        tableNames = append(tableNames, tableName)
                }
        }
        if err := tables.Err(); err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        if len(tableNames) > 0 {
                // delete one by one ...
                for _, t := range tableNames {
                        query = `DROP TABLE IF EXISTS ` + pq.QuoteIdentifier(t) + ` CASCADE`
                        if _, err := p.conn.ExecContext(context.Background(), query); err != nil {
                                return &database.Error{OrigErr: err, Query: []byte(query)}
                        }
                }
        }

        return nil
}

// ensureVersionTable checks if versions table exists and, if not, creates it.
// Note that this function locks the database, which deviates from the usual
// convention of "caller locks" in the Postgres type.
func (p *Postgres) ensureVersionTable() (err error) {
        if err = p.Lock(); err != nil {
                return err
        }

        defer func() {
                if e := p.Unlock(); e != nil {
                        err = errors.Join(err, e)
                }
        }()

        // This block checks whether the `MigrationsTable` already exists. This is useful because it allows read only postgres
        // users to also check the current version of the schema. Previously, even if `MigrationsTable` existed, the
        // `CREATE TABLE IF NOT EXISTS...` query would fail because the user does not have the CREATE permission.
        // Taken from https://github.com/mattes/migrate/blob/master/database/postgres/postgres.go#L258
        query := `SELECT COUNT(1) FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2 LIMIT 1`
        row := p.conn.QueryRowContext(context.Background(), query, p.config.migrationsSchemaName, p.config.migrationsTableName)

        var count int
        err = row.Scan(&count)
        if err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        if count == 1 {
                return nil
        }

        query = `CREATE TABLE IF NOT EXISTS ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` (version bigint not null primary key, dirty boolean not null)`
        if _, err = p.conn.ExecContext(context.Background(), query); err != nil {
                return &database.Error{OrigErr: err, Query: []byte(query)}
        }

        return nil
}

1	//go:build go1.9
2
3	package postgres
4
5	import (
6	"context"
7	"database/sql"
8	"errors"
9	"fmt"
10	"io"
11	nurl "net/url"
12	"regexp"
13	"strconv"
14	"strings"
15	"sync/atomic"
16	"time"
17
18	"github.com/golang-migrate/migrate/v4"
19	"github.com/golang-migrate/migrate/v4/database"
20	"github.com/golang-migrate/migrate/v4/database/multistmt"
21	"github.com/lib/pq"
22	)
23
24	func init() {	2✔
25	db := Postgres{}	2✔
26	database.Register("postgres", &db)	2✔
27	database.Register("postgresql", &db)	2✔
28	}	2✔
29
30	var (
31	multiStmtDelimiter = []byte(";")
32
33	DefaultMigrationsTable = "schema_migrations"
34	DefaultMultiStatementMaxSize = 10 * 1 << 20 // 10 MB
35	)
36
37	var (
38	ErrNilConfig = fmt.Errorf("no config")
39	ErrNoDatabaseName = fmt.Errorf("no database name")
40	ErrNoSchema = fmt.Errorf("no schema")
41	ErrDatabaseDirty = fmt.Errorf("database is dirty")
42	)
43
44	type Config struct {
45	MigrationsTable string
46	MigrationsTableQuoted bool
47	MultiStatementEnabled bool
48	DatabaseName string
49	SchemaName string
50	migrationsSchemaName string
51	migrationsTableName string
52	StatementTimeout time.Duration
53	MultiStatementMaxSize int
54	}
55
56	type Postgres struct {
57	// Locking and unlocking need to use the same connection
58	conn *sql.Conn
59	db *sql.DB
60	isLocked atomic.Bool
61
62	// Open and WithInstance need to guarantee that config is never nil
63	config *Config
64	}
65
66	func WithConnection(ctx context.Context, conn sql.Conn, config Config) (*Postgres, error) {	540✔
67	if config == nil {	540✔
68	return nil, ErrNilConfig	×
69	}	×
70
71	if err := conn.PingContext(ctx); err != nil {	540✔
72	return nil, err	×
73	}	×
74
75	if config.DatabaseName == "" {	850✔
76	query := `SELECT CURRENT_DATABASE()`	310✔
77	var databaseName string	310✔
78	if err := conn.QueryRowContext(ctx, query).Scan(&databaseName); err != nil {	310✔
79	return nil, &database.Error{OrigErr: err, Query: []byte(query)}	×
80	}	×
81
82	if len(databaseName) == 0 {	310✔
83	return nil, ErrNoDatabaseName	×
84	}	×
85
86	config.DatabaseName = databaseName	310✔
87	}
88
89	if config.SchemaName == "" {	1,080✔
90	query := `SELECT CURRENT_SCHEMA()`	540✔
91	var schemaName sql.NullString	540✔
92	if err := conn.QueryRowContext(ctx, query).Scan(&schemaName); err != nil {	540✔
93	return nil, &database.Error{OrigErr: err, Query: []byte(query)}	×
94	}	×
95
96	if !schemaName.Valid {	540✔
97	return nil, ErrNoSchema	×
98	}	×
99
100	config.SchemaName = schemaName.String	540✔
101	}
102
103	if len(config.MigrationsTable) == 0 {	1,040✔
104	config.MigrationsTable = DefaultMigrationsTable	500✔
105	}	500✔
106
107	config.migrationsSchemaName = config.SchemaName	540✔
108	config.migrationsTableName = config.MigrationsTable	540✔
109	if config.MigrationsTableQuoted {	570✔
110	re := regexp.MustCompile(`"(.*?)"`)	30✔
111	result := re.FindAllStringSubmatch(config.MigrationsTable, -1)	30✔
112	config.migrationsTableName = result[len(result)-1][1]	30✔
113	if len(result) == 2 {	50✔
114	config.migrationsSchemaName = result[0][1]	20✔
115	} else if len(result) > 2 {	40✔
116	return nil, fmt.Errorf("\"%s\" MigrationsTable contains too many dot characters", config.MigrationsTable)	10✔
117	}	10✔
118	}
119
120	px := &Postgres{	530✔
121	conn: conn,	530✔
122	config: config,	530✔
123	}	530✔
124		530✔
125	if err := px.ensureVersionTable(); err != nil {	550✔
126	return nil, err	20✔
127	}	20✔
128
129	return px, nil	510✔
130	}
131
132	func WithInstance(instance sql.DB, config Config) (database.Driver, error) {	530✔
133	ctx := context.Background()	530✔
134		530✔
135	if err := instance.Ping(); err != nil {	530✔
136	return nil, err	×
137	}	×
138
139	conn, err := instance.Conn(ctx)	530✔
140	if err != nil {	530✔
141	return nil, err	×
142	}	×
143
144	px, err := WithConnection(ctx, conn, config)	530✔
145	if err != nil {	560✔
146	return nil, err	30✔
147	}	30✔
148	px.db = instance	500✔
149	return px, nil	500✔
150	}
151
152	func (p *Postgres) Open(url string) (database.Driver, error) {	240✔
153	purl, err := nurl.Parse(url)	240✔
154	if err != nil {	240✔
155	return nil, err	×
156	}	×
157
158	filteredURL := migrate.FilterCustomQuery(purl)	240✔
159	connStr := filteredURL.String()	240✔
160		240✔
161	// When host is specified as a query parameter (e.g., for unix socket	240✔
162	// connections via ?host=/var/run/postgresql) and the URL has no host in	240✔
163	// the authority section, convert to a key-value connection string so	240✔
164	// lib/pq reliably uses the host parameter. Without this, the driver	240✔
165	// may silently connect to localhost:5432 instead of the specified host.	240✔
166	if purl.Host == "" && purl.Query().Get("host") != "" {	250✔
167	connStr, err = pq.ParseURL(connStr)	10✔
168	if err != nil {	10✔
NEW 169	return nil, err	×
NEW 170	}	×
171	}
172
173	db, err := sql.Open("postgres", connStr)	240✔
174	if err != nil {	240✔
175	return nil, err	×
176	}	×
177
178	migrationsTable := purl.Query().Get("x-migrations-table")	240✔
179	migrationsTableQuoted := false	240✔
180	if s := purl.Query().Get("x-migrations-table-quoted"); len(s) > 0 {	280✔
181	migrationsTableQuoted, err = strconv.ParseBool(s)	40✔
182	if err != nil {	40✔
183	return nil, fmt.Errorf("unable to parse option x-migrations-table-quoted: %w", err)	×
184	}	×
185	}
186	if (len(migrationsTable) > 0) && (migrationsTableQuoted) && ((migrationsTable[0] != '"') \|\| (migrationsTable[len(migrationsTable)-1] != '"')) {	250✔
187	return nil, fmt.Errorf("x-migrations-table must be quoted (for instance '\"migrate\".\"schema_migrations\"') when x-migrations-table-quoted is enabled, current value is: %s", migrationsTable)	10✔
188	}	10✔
189
190	statementTimeoutString := purl.Query().Get("x-statement-timeout")	230✔
191	statementTimeout := 0	230✔
192	if statementTimeoutString != "" {	230✔
193	statementTimeout, err = strconv.Atoi(statementTimeoutString)	×
194	if err != nil {	×
195	return nil, err	×
196	}	×
197	}
198
199	multiStatementMaxSize := DefaultMultiStatementMaxSize	230✔
200	if s := purl.Query().Get("x-multi-statement-max-size"); len(s) > 0 {	230✔
201	multiStatementMaxSize, err = strconv.Atoi(s)	×
202	if err != nil {	×
203	return nil, err	×
204	}	×
205	if multiStatementMaxSize <= 0 {	×
206	multiStatementMaxSize = DefaultMultiStatementMaxSize	×
207	}	×
208	}
209
210	multiStatementEnabled := false	230✔
211	if s := purl.Query().Get("x-multi-statement"); len(s) > 0 {	240✔
212	multiStatementEnabled, err = strconv.ParseBool(s)	10✔
213	if err != nil {	10✔
214	return nil, fmt.Errorf("unable to parse option x-multi-statement: %w", err)	×
215	}	×
216	}
217
218	px, err := WithInstance(db, &Config{	230✔
219	DatabaseName: purl.Path,	230✔
220	MigrationsTable: migrationsTable,	230✔
221	MigrationsTableQuoted: migrationsTableQuoted,	230✔
222	StatementTimeout: time.Duration(statementTimeout) * time.Millisecond,	230✔
223	MultiStatementEnabled: multiStatementEnabled,	230✔
224	MultiStatementMaxSize: multiStatementMaxSize,	230✔
225	})	230✔
226		230✔
227	if err != nil {	260✔
228	return nil, err	30✔
229	}	30✔
230
231	return px, nil	200✔
232	}
233
234	func (p *Postgres) Close() error {	180✔
235	connErr := p.conn.Close()	180✔
236	var dbErr error	180✔
237	if p.db != nil {	350✔
238	dbErr = p.db.Close()	170✔
239	}	170✔
240
241	if connErr != nil \|\| dbErr != nil {	180✔
242	return fmt.Errorf("conn: %v, db: %v", connErr, dbErr)	×
243	}	×
244	return nil	180✔
245	}
246
247	// https://www.postgresql.org/docs/9.6/static/explicit-locking.html#ADVISORY-LOCKS
248	func (p *Postgres) Lock() error {	710✔
249	return database.CasRestoreOnErr(&p.isLocked, false, true, database.ErrLocked, func() error {	1,380✔
250	aid, err := database.GenerateAdvisoryLockId(p.config.DatabaseName, p.config.migrationsSchemaName, p.config.migrationsTableName)	670✔
251	if err != nil {	670✔
252	return err	×
253	}	×
254
255	// This will wait indefinitely until the lock can be acquired.
256	query := `SELECT pg_advisory_lock($1)`	670✔
257	if _, err := p.conn.ExecContext(context.Background(), query, aid); err != nil {	670✔
258	return &database.Error{OrigErr: err, Err: "try lock failed", Query: []byte(query)}	×
259	}	×
260
261	return nil	670✔
262	})
263	}
264
265	func (p *Postgres) Unlock() error {	670✔
266	return database.CasRestoreOnErr(&p.isLocked, true, false, database.ErrNotLocked, func() error {	1,340✔
267	aid, err := database.GenerateAdvisoryLockId(p.config.DatabaseName, p.config.migrationsSchemaName, p.config.migrationsTableName)	670✔
268	if err != nil {	670✔
269	return err	×
270	}	×
271
272	query := `SELECT pg_advisory_unlock($1)`	670✔
273	if _, err := p.conn.ExecContext(context.Background(), query, aid); err != nil {	670✔
274	return &database.Error{OrigErr: err, Query: []byte(query)}	×
275	}	×
276	return nil	670✔
277	})
278	}
279
280	func (p *Postgres) Run(migration io.Reader) error {	310✔
281	if p.config.MultiStatementEnabled {	320✔
282	var err error	10✔
283	if e := multistmt.Parse(migration, multiStmtDelimiter, p.config.MultiStatementMaxSize, func(m []byte) bool {	30✔
284	if err = p.runStatement(m); err != nil {	20✔
285	return false	×
286	}	×
287	return true	20✔
288	}); e != nil {	×
289	return e	×
290	}	×
291	return err	10✔
292	}
293	migr, err := io.ReadAll(migration)	300✔
294	if err != nil {	300✔
295	return err	×
296	}	×
297	return p.runStatement(migr)	300✔
298	}
299
300	func (p *Postgres) runStatement(statement []byte) error {	320✔
301	ctx := context.Background()	320✔
302	if p.config.StatementTimeout != 0 {	320✔
303	var cancel context.CancelFunc	×
304	ctx, cancel = context.WithTimeout(ctx, p.config.StatementTimeout)	×
305	defer cancel()	×
306	}	×
307	query := string(statement)	320✔
308	if strings.TrimSpace(query) == "" {	320✔
309	return nil	×
310	}	×
311	if _, err := p.conn.ExecContext(ctx, query); err != nil {	330✔
312	if pgErr, ok := err.(*pq.Error); ok {	20✔
313	var line uint	10✔
314	var col uint	10✔
315	var lineColOK bool	10✔
316	if pgErr.Position != "" {	20✔
317	if pos, err := strconv.ParseUint(pgErr.Position, 10, 64); err == nil {	20✔
318	line, col, lineColOK = computeLineFromPos(query, int(pos))	10✔
319	}	10✔
320	}
321	message := fmt.Sprintf("migration failed: %s", pgErr.Message)	10✔
322	if lineColOK {	20✔
323	message = fmt.Sprintf("%s (column %d)", message, col)	10✔
324	}	10✔
325	if pgErr.Detail != "" {	10✔
326	message = fmt.Sprintf("%s, %s", message, pgErr.Detail)	×
327	}	×
328	return database.Error{OrigErr: err, Err: message, Query: statement, Line: line}	10✔
329	}
330	return database.Error{OrigErr: err, Err: "migration failed", Query: statement}	×
331	}
332	return nil	310✔
333	}
334
335	func computeLineFromPos(s string, pos int) (line uint, col uint, ok bool) {	74✔
336	// replace crlf with lf	74✔
337	s = strings.ReplaceAll(s, "\r\n", "\n")	74✔
338	// pg docs: pos uses index 1 for the first character, and positions are measured in characters not bytes	74✔
339	runes := []rune(s)	74✔
340	if pos > len(runes) {	82✔
341	return 0, 0, false	8✔
342	}	8✔
343	sel := runes[:pos]	66✔
344	line = uint(runesCount(sel, newLine) + 1)	66✔
345	col = uint(pos - 1 - runesLastIndex(sel, newLine))	66✔
346	return line, col, true	66✔
347	}
348
349	const newLine = '\n'
350
351	func runesCount(input []rune, target rune) int {	66✔
352	var count int	66✔
353	for _, r := range input {	1,404✔
354	if r == target {	1,442✔
355	count++	104✔
356	}	104✔
357	}
358	return count	66✔
359	}
360
361	func runesLastIndex(input []rune, target rune) int {	66✔
362	for i := len(input) - 1; i >= 0; i-- {	780✔
363	if input[i] == target {	770✔
364	return i	56✔
365	}	56✔
366	}
367	return -1	10✔
368	}
369
370	func (p *Postgres) SetVersion(version int, dirty bool) error {	440✔
371	tx, err := p.conn.BeginTx(context.Background(), &sql.TxOptions{})	440✔
372	if err != nil {	440✔
373	return &database.Error{OrigErr: err, Err: "transaction start failed"}	×
374	}	×
375
376	query := `TRUNCATE ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName)	440✔
377	if _, err := tx.Exec(query); err != nil {	440✔
378	if errRollback := tx.Rollback(); errRollback != nil {	×
379	err = errors.Join(err, errRollback)	×
380	}	×
381	return &database.Error{OrigErr: err, Query: []byte(query)}	×
382	}
383
384	// Also re-write the schema version for nil dirty versions to prevent
385	// empty schema version for failed down migration on the first migration
386	// See: https://github.com/golang-migrate/migrate/issues/330
387	if version >= 0 \|\| (version == database.NilVersion && dirty) {	840✔
388	query = `INSERT INTO ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` (version, dirty) VALUES ($1, $2)`	400✔
389	if _, err := tx.Exec(query, version, dirty); err != nil {	400✔
390	if errRollback := tx.Rollback(); errRollback != nil {	×
391	err = errors.Join(err, errRollback)	×
392	}	×
393	return &database.Error{OrigErr: err, Query: []byte(query)}	×
394	}
395	}
396
397	if err := tx.Commit(); err != nil {	440✔
398	return &database.Error{OrigErr: err, Err: "transaction commit failed"}	×
399	}	×
400
401	return nil	440✔
402	}
403
404	func (p *Postgres) Version() (version int, dirty bool, err error) {	330✔
405	query := `SELECT version, dirty FROM ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` LIMIT 1`	330✔
406	err = p.conn.QueryRowContext(context.Background(), query).Scan(&version, &dirty)	330✔
407	switch {	330✔
408	case err == sql.ErrNoRows:	110✔
409	return database.NilVersion, false, nil	110✔
410
411	case err != nil:	×
412	if e, ok := err.(*pq.Error); ok {	×
413	if e.Code.Name() == "undefined_table" {	×
414	return database.NilVersion, false, nil	×
415	}	×
416	}
417	return 0, false, &database.Error{OrigErr: err, Query: []byte(query)}	×
418
419	default:	220✔
420	return version, dirty, nil	220✔
421	}
422	}
423
424	func (p *Postgres) Drop() (err error) {	50✔
425	// select all tables in current schema	50✔
426	query := `SELECT table_name FROM information_schema.tables WHERE table_schema=(SELECT current_schema()) AND table_type='BASE TABLE'`	50✔
427	tables, err := p.conn.QueryContext(context.Background(), query)	50✔
428	if err != nil {	50✔
429	return &database.Error{OrigErr: err, Query: []byte(query)}	×
430	}	×
431	defer func() {	100✔
432	if errClose := tables.Close(); errClose != nil {	50✔
433	err = errors.Join(err, errClose)	×
434	}	×
435	}()
436
437	// delete one table after another
438	tableNames := make([]string, 0)	50✔
439	for tables.Next() {	130✔
440	var tableName string	80✔
441	if err := tables.Scan(&tableName); err != nil {	80✔
442	return err	×
443	}	×
444	if len(tableName) > 0 {	160✔
445	tableNames = append(tableNames, tableName)	80✔
446	}	80✔
447	}
448	if err := tables.Err(); err != nil {	50✔
449	return &database.Error{OrigErr: err, Query: []byte(query)}	×
450	}	×
451
452	if len(tableNames) > 0 {	100✔
453	// delete one by one ...	50✔
454	for _, t := range tableNames {	130✔
455	query = `DROP TABLE IF EXISTS ` + pq.QuoteIdentifier(t) + ` CASCADE`	80✔
456	if _, err := p.conn.ExecContext(context.Background(), query); err != nil {	80✔
457	return &database.Error{OrigErr: err, Query: []byte(query)}	×
458	}	×
459	}
460	}
461
462	return nil	50✔
463	}
464
465	// ensureVersionTable checks if versions table exists and, if not, creates it.
466	// Note that this function locks the database, which deviates from the usual
467	// convention of "caller locks" in the Postgres type.
468	func (p *Postgres) ensureVersionTable() (err error) {	530✔
469	if err = p.Lock(); err != nil {	530✔
470	return err	×
471	}	×
472
473	defer func() {	1,060✔
474	if e := p.Unlock(); e != nil {	530✔
475	err = errors.Join(err, e)	×
476	}	×
477	}()
478
479	// This block checks whether the `MigrationsTable` already exists. This is useful because it allows read only postgres
480	// users to also check the current version of the schema. Previously, even if `MigrationsTable` existed, the
481	// `CREATE TABLE IF NOT EXISTS...` query would fail because the user does not have the CREATE permission.
482	// Taken from https://github.com/mattes/migrate/blob/master/database/postgres/postgres.go#L258
483	query := `SELECT COUNT(1) FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2 LIMIT 1`	530✔
484	row := p.conn.QueryRowContext(context.Background(), query, p.config.migrationsSchemaName, p.config.migrationsTableName)	530✔
485		530✔
486	var count int	530✔
487	err = row.Scan(&count)	530✔
488	if err != nil {	530✔
489	return &database.Error{OrigErr: err, Query: []byte(query)}	×
490	}	×
491
492	if count == 1 {	830✔
493	return nil	300✔
494	}	300✔
495
496	query = `CREATE TABLE IF NOT EXISTS ` + pq.QuoteIdentifier(p.config.migrationsSchemaName) + `.` + pq.QuoteIdentifier(p.config.migrationsTableName) + ` (version bigint not null primary key, dirty boolean not null)`	230✔
497	if _, err = p.conn.ExecContext(context.Background(), query); err != nil {	250✔
498	return &database.Error{OrigErr: err, Query: []byte(query)}	20✔
499	}	20✔
500
501	return nil	210✔
502	}

golang-migrate / migrate / 23749185424

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous