16295271531

Committed 15 Jul 2025 01:55PM UTC coverage: 56.553% (+0.2%) from 56.314%

Build # 16295271531

Build Type

Pull #1294

github

Committed by

dsyers

Commit Message

lint

Pull Request Pull Request #1294: Triggers

Run Details

790 of 1325 new or added lines in 24 files covered. (59.62%)

4 existing lines in 4 files now uncovered.

5277 of 9331 relevant lines covered (56.55%)

55.43 hits per line

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

68.91

/database/postgres/postgres.go

//go:build go1.9

package postgres

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

        "go.uber.org/atomic"

        "github.com/golang-migrate/migrate/v4"
        "github.com/golang-migrate/migrate/v4/database"
        "github.com/golang-migrate/migrate/v4/database/multistmt"
        "github.com/hashicorp/go-multierror"
        "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

        Triggers map[string]func(response interface{}) error
}

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
}

type TriggerResponse struct {
        Driver  *Postgres
        Config  *Config
        Trigger string
        Detail  interface{}
}

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
        }

        db, err := sql.Open("postgres", migrate.FilterCustomQuery(purl).String())
        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
}

func (p *Postgres) AddTriggers(t map[string]func(response interface{}) error) {
        p.config.Triggers = t
}

func (p *Postgres) Trigger(name string, detail interface{}) error {
        if p.config.Triggers == nil {
                return nil
        }

        if trigger, ok := p.config.Triggers[name]; ok {
                return trigger(TriggerResponse{
                        Driver:  p,
                        Config:  p.config,
                        Trigger: name,
                        Detail:  detail,
                })
        }

        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 e := p.Trigger(database.TrigRunPre, struct {
                                Query string
                        }{Query: string(m)}); e != nil {
                                return false
                        }
                        if err = p.runStatement(m); err != nil {
                                return false
                        }
                        if e := p.Trigger(database.TrigRunPost, struct {
                                Query string
                        }{Query: string(m)}); e != nil {
                                return false
                        }
                        return true
                }); e != nil {
                        return e
                }
                return err
        }
        migr, err := io.ReadAll(migration)
        if err != nil {
                return err
        }
        if err = p.Trigger(database.TrigRunPre, struct {
                Query string
        }{Query: string(migr)}); err != nil {
                return &database.Error{OrigErr: err, Err: "failed to trigger RunPre"}
        }
        if err = p.runStatement(migr); err != nil {
                return err
        }
        if err = p.Trigger(database.TrigRunPost, struct {
                Query string
        }{Query: string(migr)}); err != nil {
                return &database.Error{OrigErr: err, Err: "failed to trigger RunPost"}
        }
        return nil
}

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.Replace(s, "\r\n", "\n", -1)
        // 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"}
        }

        if err := p.Trigger(database.TrigSetVersionPre, struct {
                Version int
                Dirty   bool
        }{Version: version, Dirty: dirty}); err != nil {
                if errRollback := tx.Rollback(); errRollback != nil {
                        err = multierror.Append(err, errRollback)
                }
                return &database.Error{OrigErr: err, Err: "failed to trigger SetVersionPre"}
        }

        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 = multierror.Append(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 = multierror.Append(err, errRollback)
                        }
                        return &database.Error{OrigErr: err, Query: []byte(query)}
                }
        }

        if err := p.Trigger(database.TrigSetVersionPost, struct {
                Version int
                Dirty   bool
        }{Version: version, Dirty: dirty}); err != nil {
                if errRollback := tx.Rollback(); errRollback != nil {
                        err = multierror.Append(err, errRollback)
                }
                return &database.Error{OrigErr: err, Err: "failed to trigger SetVersionPost"}
        }

        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 = multierror.Append(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 {
                        if err == nil {
                                err = e
                        } else {
                                err = multierror.Append(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 {
                if err := p.Trigger(database.TrigVersionTableExists, nil); err != nil {
                        return &database.Error{OrigErr: err, Err: "failed to trigger VersionTableExists"}
                }
                return nil
        }

        if err := p.Trigger(database.TrigVersionTablePre, nil); err != nil {
                return &database.Error{OrigErr: err, Err: "failed to trigger VersionTablePre"}
        }

        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)}
        }

        if err := p.Trigger(database.TrigVersionTablePost, nil); err != nil {
                return &database.Error{OrigErr: err, Err: "failed to trigger VersionTablePost"}
        }

        return nil
}

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

golang-migrate / migrate / 16295271531

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