Moved morph dependency to new repo (#19618)
```release-note NONE ``` Co-authored-by: Ibrahim Serdar Acikgoz <serdaracikgoz86@gmail.com>
Этот коммит содержится в:
Agniva De Sarker
коммит произвёл
GitHub
GitHub
родитель
9534efe534
Коммит
2a59047d07
269
vendor/github.com/mattermost/morph/drivers/postgres/lock.go
сгенерированный
поставляемый
Обычный файл
269
vendor/github.com/mattermost/morph/drivers/postgres/lock.go
сгенерированный
поставляемый
Обычный файл
@@ -0,0 +1,269 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/mattermost/morph/drivers"
|
||||
)
|
||||
|
||||
// Mutex is similar to sync.Mutex, except usable by morph to lock the db.
|
||||
//
|
||||
// Pick a unique name for each mutex your plugin requires.
|
||||
//
|
||||
// A Mutex must not be copied after first use.
|
||||
type Mutex struct {
|
||||
noCopy
|
||||
key string
|
||||
|
||||
// lock guards the variables used to manage the refresh task, and is not itself related to
|
||||
// the db lock.
|
||||
lock sync.Mutex
|
||||
stopRefresh chan bool
|
||||
refreshDone chan bool
|
||||
conn *sql.Conn
|
||||
}
|
||||
|
||||
// NewMutex creates a mutex with the given key name.
|
||||
//
|
||||
// returns error if key is empty.
|
||||
func NewMutex(key string, driver drivers.Driver) (*Mutex, error) {
|
||||
key, err := drivers.MakeLockKey(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), drivers.TTL)
|
||||
defer cancel()
|
||||
|
||||
ps, ok := driver.(*postgres)
|
||||
if !ok {
|
||||
return nil, errors.New("incorrect implementation of the driver")
|
||||
}
|
||||
|
||||
conn, err := ps.db.Conn(context.Background())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
createTableIfNotExistsQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id varchar(64) PRIMARY KEY, expireat bigint);", drivers.MutexTableName)
|
||||
if _, err = conn.ExecContext(ctx, createTableIfNotExistsQuery); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &Mutex{
|
||||
key: key,
|
||||
conn: conn,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// lock makes a single attempt to lock the mutex, returning true only if successful.
|
||||
func (m *Mutex) tryLock(ctx context.Context) (bool, error) {
|
||||
now := time.Now()
|
||||
tx, err := m.conn.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
query := fmt.Sprintf("INSERT INTO %s (id, expireat) VALUES ($1, $2)", drivers.MutexTableName)
|
||||
if _, err := tx.Exec(query, m.key, now.Add(drivers.TTL).Unix()); err != nil {
|
||||
err2 := m.releaseLock(tx, now)
|
||||
if err2 == nil { // lock has been released due to expiration
|
||||
return true, nil
|
||||
}
|
||||
|
||||
return false, fmt.Errorf("failed to lock mutex: %w", err)
|
||||
}
|
||||
|
||||
err = tx.Commit()
|
||||
if err != nil {
|
||||
if txErr := tx.Rollback(); txErr != nil {
|
||||
return false, txErr
|
||||
}
|
||||
|
||||
return false, err
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (m *Mutex) releaseLock(tx *sql.Tx, t time.Time) error {
|
||||
e, err := m.getExpireAt(tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if t.Unix() < e {
|
||||
if txErr := tx.Rollback(); txErr != nil {
|
||||
return fmt.Errorf("could not rollback: %w", txErr)
|
||||
}
|
||||
|
||||
return errors.New("could not release the lock")
|
||||
}
|
||||
|
||||
query := fmt.Sprintf("UPDATE %s SET expireat = $1 WHERE id = $2", drivers.MutexTableName)
|
||||
if err = executeTx(tx, query, t.Add(drivers.TTL).Unix(), m.key); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = tx.Commit()
|
||||
if err != nil {
|
||||
if txErr := tx.Rollback(); txErr != nil {
|
||||
return fmt.Errorf("could not rollback transaction: %w", txErr)
|
||||
}
|
||||
|
||||
return fmt.Errorf("unable to set new expireat for mutex: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Mutex) getExpireAt(tx *sql.Tx) (int64, error) {
|
||||
var expireAt int64
|
||||
query := fmt.Sprintf("SELECT expireat FROM %s WHERE id = $1", drivers.MutexTableName)
|
||||
err := tx.QueryRow(query, m.key).Scan(&expireAt)
|
||||
if err != nil {
|
||||
if txErr := tx.Rollback(); txErr != nil {
|
||||
return -1, fmt.Errorf("could not rollback: %w", txErr)
|
||||
}
|
||||
|
||||
return -1, fmt.Errorf("failed to fetch mutex from db: %w", err)
|
||||
}
|
||||
|
||||
return expireAt, nil
|
||||
}
|
||||
|
||||
// refreshLock rewrites the lock key value with a new expiry, returning nil only if successful.
|
||||
func (m *Mutex) refreshLock(ctx context.Context) error {
|
||||
tx, err := m.conn.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
e, err := m.getExpireAt(tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
tmp := time.Unix(e, 0)
|
||||
query := fmt.Sprintf("UPDATE %s SET expireat = $1 WHERE id = $2", drivers.MutexTableName)
|
||||
if err = executeTx(tx, query, tmp.Add(drivers.TTL).Unix(), m.key); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = tx.Commit()
|
||||
if err != nil {
|
||||
if txErr := tx.Rollback(); txErr != nil {
|
||||
return fmt.Errorf("could not rollback: %w", txErr)
|
||||
}
|
||||
|
||||
return fmt.Errorf("unable to refresh expireat for mutex: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Lock locks m. If the mutex is already locked by any other morph instance, including the current one,
|
||||
// the calling goroutine blocks until the mutex can be locked.
|
||||
func (m *Mutex) Lock() error {
|
||||
return m.LockWithContext(context.Background())
|
||||
}
|
||||
|
||||
// LockWithContext locks m unless the context is canceled. If the mutex is already locked by any other
|
||||
// instance, including the current one, the calling goroutine blocks until the mutex can be locked,
|
||||
// or the context is canceled.
|
||||
//
|
||||
// The mutex is locked only if a nil error is returned.
|
||||
func (m *Mutex) LockWithContext(ctx context.Context) error {
|
||||
var waitInterval time.Duration
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(waitInterval):
|
||||
}
|
||||
|
||||
ok, err := m.tryLock(ctx)
|
||||
if err != nil || !ok {
|
||||
waitInterval = drivers.NextWaitInterval(waitInterval, err)
|
||||
continue
|
||||
}
|
||||
|
||||
break
|
||||
}
|
||||
|
||||
stop := make(chan bool)
|
||||
done := make(chan bool)
|
||||
go func() {
|
||||
defer close(done)
|
||||
t := time.NewTicker(drivers.RefreshInterval)
|
||||
for {
|
||||
select {
|
||||
case <-t.C:
|
||||
err := m.refreshLock(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
case <-stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
m.lock.Lock()
|
||||
m.stopRefresh = stop
|
||||
m.refreshDone = done
|
||||
m.lock.Unlock()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Unlock unlocks m. It is a run-time error if m is not locked on entry to Unlock.
|
||||
//
|
||||
// Just like sync.Mutex, a locked Lock is not associated with a particular goroutine or a process.
|
||||
func (m *Mutex) Unlock() error {
|
||||
m.lock.Lock()
|
||||
if m.stopRefresh == nil {
|
||||
m.lock.Unlock()
|
||||
panic("mutex has not been acquired")
|
||||
}
|
||||
|
||||
close(m.stopRefresh)
|
||||
m.stopRefresh = nil
|
||||
<-m.refreshDone
|
||||
m.lock.Unlock()
|
||||
|
||||
defer m.conn.Close()
|
||||
|
||||
// If an error occurs deleting, the mutex will still expire, allowing later retry.
|
||||
query := fmt.Sprintf("DELETE FROM %s WHERE id = $1", drivers.MutexTableName)
|
||||
_, err := m.conn.ExecContext(context.Background(), query, m.key)
|
||||
return err
|
||||
}
|
||||
|
||||
func executeTx(tx *sql.Tx, query string, args ...interface{}) error {
|
||||
if _, err := tx.Exec(query, args...); err != nil {
|
||||
if txErr := tx.Rollback(); txErr != nil {
|
||||
return fmt.Errorf("could not rollback transaction: %w", txErr)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// noCopy may be embedded into structs which must not be copied
|
||||
// after the first use.
|
||||
//
|
||||
// See https://golang.org/issues/8005#issuecomment-190753527
|
||||
// for details.
|
||||
type noCopy struct{}
|
||||
|
||||
// Lock is a no-op used by -copylocks checker from `go vet`.
|
||||
func (*noCopy) Lock() {}
|
||||
392
vendor/github.com/mattermost/morph/drivers/postgres/postgres.go
сгенерированный
поставляемый
Обычный файл
392
vendor/github.com/mattermost/morph/drivers/postgres/postgres.go
сгенерированный
поставляемый
Обычный файл
@@ -0,0 +1,392 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strconv"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
|
||||
_ "github.com/lib/pq"
|
||||
"github.com/mattermost/morph/drivers"
|
||||
"github.com/mattermost/morph/models"
|
||||
)
|
||||
|
||||
var (
|
||||
driverName = "postgres"
|
||||
defaultMigrationMaxSize = 10 * 1 << 20 // 10 MB
|
||||
configParams = []string{
|
||||
"x-migration-max-size",
|
||||
"x-migrations-table",
|
||||
"x-statement-timeout",
|
||||
}
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
drivers.Config
|
||||
databaseName string
|
||||
schemaName string
|
||||
closeDBonClose bool
|
||||
}
|
||||
|
||||
type postgres struct {
|
||||
conn *sql.Conn
|
||||
db *sql.DB
|
||||
config *Config
|
||||
}
|
||||
|
||||
func WithInstance(dbInstance *sql.DB, config *Config) (drivers.Driver, error) {
|
||||
driverConfig := mergeConfigs(config, getDefaultConfig())
|
||||
|
||||
conn, err := dbInstance.Conn(context.Background())
|
||||
if err != nil {
|
||||
return nil, &drivers.DatabaseError{Driver: driverName, Command: "grabbing_connection", OrigErr: err, Message: "failed to grab connection to the database"}
|
||||
}
|
||||
|
||||
if driverConfig.databaseName, err = currentDatabaseNameFromDB(conn, driverConfig); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if driverConfig.schemaName, err = currentSchema(conn, driverConfig); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &postgres{
|
||||
conn: conn,
|
||||
db: dbInstance,
|
||||
config: driverConfig,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func Open(connURL string) (drivers.Driver, error) {
|
||||
customParams, err := drivers.ExtractCustomParams(connURL, configParams)
|
||||
if err != nil {
|
||||
return nil, &drivers.AppError{Driver: driverName, OrigErr: err, Message: "failed to parse custom parameters from url"}
|
||||
}
|
||||
|
||||
sanitizedConnURL, err := drivers.RemoveParamsFromURL(connURL, configParams)
|
||||
if err != nil {
|
||||
return nil, &drivers.AppError{Driver: driverName, OrigErr: err, Message: "failed to sanitize url from custom parameters"}
|
||||
}
|
||||
|
||||
driverConfig, err := mergeConfigWithParams(customParams, getDefaultConfig())
|
||||
if err != nil {
|
||||
return nil, &drivers.AppError{Driver: driverName, OrigErr: err, Message: "failed to merge custom params to driver config"}
|
||||
}
|
||||
|
||||
db, err := sql.Open(driverName, sanitizedConnURL)
|
||||
if err != nil {
|
||||
return nil, &drivers.DatabaseError{Driver: driverName, Command: "opening_connection", OrigErr: err, Message: "failed to open connection with the database"}
|
||||
}
|
||||
|
||||
conn, err := db.Conn(context.Background())
|
||||
if err != nil {
|
||||
return nil, &drivers.DatabaseError{Driver: driverName, Command: "grabbing_connection", OrigErr: err, Message: "failed to grab connection to the database"}
|
||||
}
|
||||
|
||||
if driverConfig.databaseName, err = extractDatabaseNameFromURL(connURL); err != nil {
|
||||
return nil, &drivers.AppError{Driver: driverName, OrigErr: err, Message: "failed to extract database name from connection url"}
|
||||
}
|
||||
|
||||
if driverConfig.schemaName, err = currentSchema(conn, driverConfig); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
driverConfig.closeDBonClose = true
|
||||
|
||||
return &postgres{
|
||||
db: db,
|
||||
config: driverConfig,
|
||||
conn: conn,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func currentSchema(conn *sql.Conn, config *Config) (string, error) {
|
||||
query := "SELECT CURRENT_SCHEMA()"
|
||||
|
||||
ctx, cancel := drivers.GetContext(config.StatementTimeoutInSecs)
|
||||
defer cancel()
|
||||
|
||||
var schemaName string
|
||||
if err := conn.QueryRowContext(ctx, query).Scan(&schemaName); err != nil {
|
||||
return "", &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "failed to fetch current schema",
|
||||
Command: "current_schema",
|
||||
Query: []byte(query),
|
||||
}
|
||||
}
|
||||
return schemaName, nil
|
||||
}
|
||||
|
||||
func mergeConfigWithParams(params map[string]string, config *Config) (*Config, error) {
|
||||
var err error
|
||||
|
||||
for _, configKey := range configParams {
|
||||
if v, ok := params[configKey]; ok {
|
||||
switch configKey {
|
||||
case "x-migration-max-size":
|
||||
if config.MigrationMaxSize, err = strconv.Atoi(v); err != nil {
|
||||
return nil, errors.New(fmt.Sprintf("failed to cast config param %s of %s", configKey, v))
|
||||
}
|
||||
case "x-migrations-table":
|
||||
config.MigrationsTable = v
|
||||
case "x-statement-timeout":
|
||||
if config.StatementTimeoutInSecs, err = strconv.Atoi(v); err != nil {
|
||||
return nil, errors.New(fmt.Sprintf("failed to cast config param %s of %s", configKey, v))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return config, nil
|
||||
}
|
||||
|
||||
func mergeConfigs(config, defaultConfig *Config) *Config {
|
||||
if config.MigrationsTable == "" {
|
||||
config.MigrationsTable = defaultConfig.MigrationsTable
|
||||
}
|
||||
|
||||
if config.StatementTimeoutInSecs == 0 {
|
||||
config.StatementTimeoutInSecs = defaultConfig.StatementTimeoutInSecs
|
||||
}
|
||||
|
||||
if config.MigrationMaxSize == 0 {
|
||||
config.MigrationMaxSize = defaultConfig.MigrationMaxSize
|
||||
}
|
||||
|
||||
return config
|
||||
}
|
||||
|
||||
func (pg *postgres) Ping() error {
|
||||
ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
|
||||
defer cancel()
|
||||
|
||||
return pg.conn.PingContext(ctx)
|
||||
}
|
||||
|
||||
func (pg *postgres) createSchemaTableIfNotExists() (err error) {
|
||||
ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
|
||||
defer cancel()
|
||||
|
||||
createTableIfNotExistsQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (version bigint not null primary key, name varchar not null)", pg.config.MigrationsTable)
|
||||
if _, err = pg.conn.ExecContext(ctx, createTableIfNotExistsQuery); err != nil {
|
||||
return &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "failed while executing query",
|
||||
Command: "create_migrations_table_if_not_exists",
|
||||
Query: []byte(createTableIfNotExistsQuery),
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (postgres) DriverName() string {
|
||||
return driverName
|
||||
}
|
||||
|
||||
func (pg *postgres) Close() error {
|
||||
if pg.conn != nil {
|
||||
if err := pg.conn.Close(); err != nil {
|
||||
return &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "failed to close database connection",
|
||||
Command: "pg_conn_close",
|
||||
Query: nil,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if pg.db != nil && pg.config.closeDBonClose {
|
||||
if err := pg.db.Close(); err != nil {
|
||||
return &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "failed to close database",
|
||||
Command: "pg_db_close",
|
||||
Query: nil,
|
||||
}
|
||||
}
|
||||
pg.db = nil
|
||||
}
|
||||
|
||||
pg.conn = nil
|
||||
return nil
|
||||
}
|
||||
|
||||
func (pg *postgres) Apply(migration *models.Migration, saveVersion bool) (err error) {
|
||||
query, readErr := migration.Query()
|
||||
if readErr != nil {
|
||||
return &drivers.AppError{
|
||||
OrigErr: readErr,
|
||||
Driver: driverName,
|
||||
Message: fmt.Sprintf("failed to read migration query: %s", migration.Name),
|
||||
}
|
||||
}
|
||||
defer migration.Close()
|
||||
|
||||
ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
|
||||
defer cancel()
|
||||
|
||||
transaction, err := pg.conn.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "error while opening a transaction to the database",
|
||||
Command: "begin_transaction",
|
||||
}
|
||||
}
|
||||
|
||||
if err = executeQuery(transaction, query); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if saveVersion {
|
||||
if err = executeQuery(transaction, pg.addMigrationQuery(migration)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
err = transaction.Commit()
|
||||
if err != nil {
|
||||
return &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "error while committing a transaction to the database",
|
||||
Command: "commit_transaction",
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (pg *postgres) AppliedMigrations() (migrations []*models.Migration, err error) {
|
||||
if pg.conn == nil {
|
||||
return nil, &drivers.AppError{
|
||||
OrigErr: errors.New("driver has no connection established"),
|
||||
Message: "database connection is missing",
|
||||
Driver: driverName,
|
||||
}
|
||||
}
|
||||
|
||||
if err := pg.createSchemaTableIfNotExists(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
query := fmt.Sprintf("SELECT version, name FROM %s", pg.config.MigrationsTable)
|
||||
ctx, cancel := drivers.GetContext(pg.config.StatementTimeoutInSecs)
|
||||
defer cancel()
|
||||
var appliedMigrations []*models.Migration
|
||||
var version uint32
|
||||
var name string
|
||||
|
||||
rows, err := pg.conn.QueryContext(ctx, query)
|
||||
if err != nil {
|
||||
return nil, &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "failed to fetch applied migrations",
|
||||
Command: "select_applied_migrations",
|
||||
Query: []byte(query),
|
||||
}
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
if err := rows.Scan(&version, &name); err != nil {
|
||||
return nil, &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "failed to scan applied migration row",
|
||||
Command: "scan_applied_migrations",
|
||||
}
|
||||
}
|
||||
|
||||
appliedMigrations = append(appliedMigrations, &models.Migration{
|
||||
Name: name,
|
||||
Version: version,
|
||||
Direction: models.Up,
|
||||
})
|
||||
}
|
||||
|
||||
return appliedMigrations, nil
|
||||
}
|
||||
|
||||
func (pg *postgres) addMigrationQuery(migration *models.Migration) string {
|
||||
if migration.Direction == models.Down {
|
||||
return fmt.Sprintf("DELETE FROM %s WHERE (Version=%d AND NAME='%s')", pg.config.MigrationsTable, migration.Version, migration.Name)
|
||||
}
|
||||
return fmt.Sprintf("INSERT INTO %s (version, name) VALUES (%d, '%s')", pg.config.MigrationsTable, migration.Version, migration.Name)
|
||||
}
|
||||
|
||||
func executeQuery(transaction *sql.Tx, query string) error {
|
||||
if _, err := transaction.Exec(query); err != nil {
|
||||
if txErr := transaction.Rollback(); txErr != nil {
|
||||
err = errors.Wrap(errors.New(err.Error()+txErr.Error()), "failed to execute query in migration transaction")
|
||||
|
||||
return &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Command: "rollback_transaction",
|
||||
}
|
||||
}
|
||||
|
||||
return &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "failed to execute migration",
|
||||
Command: "executing_query",
|
||||
Query: []byte(query),
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func currentDatabaseNameFromDB(conn *sql.Conn, config *Config) (string, error) {
|
||||
query := "SELECT CURRENT_DATABASE()"
|
||||
|
||||
ctx, cancel := drivers.GetContext(config.StatementTimeoutInSecs)
|
||||
defer cancel()
|
||||
|
||||
var databaseName string
|
||||
if err := conn.QueryRowContext(ctx, query).Scan(&databaseName); err != nil {
|
||||
return "", &drivers.DatabaseError{
|
||||
OrigErr: err,
|
||||
Driver: driverName,
|
||||
Message: "failed to fetch database name",
|
||||
Command: "current_database",
|
||||
Query: []byte(query),
|
||||
}
|
||||
}
|
||||
return databaseName, nil
|
||||
}
|
||||
|
||||
func (pg *postgres) SetConfig(key string, value interface{}) error {
|
||||
if pg.config != nil {
|
||||
switch key {
|
||||
case "StatementTimeoutInSecs":
|
||||
n, ok := value.(int)
|
||||
if ok {
|
||||
pg.config.StatementTimeoutInSecs = n
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("incorrect value type for %s", key)
|
||||
case "MigrationsTable":
|
||||
n, ok := value.(string)
|
||||
if ok {
|
||||
pg.config.MigrationsTable = n
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("incorrect value type for %s", key)
|
||||
}
|
||||
}
|
||||
|
||||
return fmt.Errorf("incorrect key name %q", key)
|
||||
}
|
||||
26
vendor/github.com/mattermost/morph/drivers/postgres/utils.go
сгенерированный
поставляемый
Обычный файл
26
vendor/github.com/mattermost/morph/drivers/postgres/utils.go
сгенерированный
поставляемый
Обычный файл
@@ -0,0 +1,26 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
|
||||
"github.com/mattermost/morph/drivers"
|
||||
)
|
||||
|
||||
func extractDatabaseNameFromURL(URL string) (string, error) {
|
||||
uri, err := url.Parse(URL)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return uri.Path[1:], nil
|
||||
}
|
||||
|
||||
func getDefaultConfig() *Config {
|
||||
return &Config{
|
||||
Config: drivers.Config{
|
||||
MigrationsTable: "db_migrations",
|
||||
StatementTimeoutInSecs: 60,
|
||||
MigrationMaxSize: defaultMigrationMaxSize,
|
||||
},
|
||||
}
|
||||
}
|
||||
Ссылка в новой задаче
Block a user