func (t *Migrator) execMigrationDir(tx pgx.Tx, dirName string) error {
	f, err := t.sqlFiles.Open(dirName)
	if err != nil {
		return fmt.Errorf("unable to get migration scripts: name %s, err %w", dirName, err)
	}
	var (
		entries []string
		stat    os.FileInfo
		file    http.File
	)
	if myToC, ok := t.toc[dirName]; ok {
		entries = make([]string, 0, len(myToC))
		for _, fileName := range myToC {
			fullName := filepath.Join(dirName, fileName)
			file, err = t.sqlFiles.Open(fullName)
			if err != nil {
				return fmt.Errorf("unable to get migration script from toc: name %s, err %w", fullName, err)
			}
			if stat, err = file.Stat(); err != nil {
				return fmt.Errorf("unable to stat migration script from toc: name %s, err %w", fullName, err)
			}
			if stat.IsDir() {
				log.Warn("msg", "Ignoring directory entry in migration ToC", "dir", dirName)
				continue
			}
			entries = append(entries, fileName)
		}
	} else {
		fileEntries, err := f.Readdir(-1)
		if err != nil {
			return fmt.Errorf("unable to read migration scripts directory: name %s, err %w", dirName, err)
		}
		entries = orderFilesNaturally(fileEntries)
	}
	for _, e := range entries {
		fileName := filepath.Join(dirName, e)
		err := t.execMigrationFile(tx, fileName)
		if err != nil {
			return err
		}
	}
	return nil
}
execMigrationDir finds all the migration files in a directory, orders them
(either by ToC or by their numerical prefix) and executes them in a transaction.
package pgmodel
import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"github.com/blang/semver/v4"
	"github.com/jackc/pgconn"
	"github.com/jackc/pgx/v4"
	"github.com/timescale/promscale/pkg/log"
	"github.com/timescale/promscale/pkg/migrations"
	"github.com/timescale/promscale/pkg/pgmodel/common/errors"
	"github.com/timescale/promscale/pkg/pgmodel/common/extension"
	"github.com/timescale/promscale/pkg/pgmodel/common/schema"
)
const (
	createMigrationsTable   = "CREATE TABLE IF NOT EXISTS prom_schema_migrations (version text not null primary key)"
	getVersion              = "SELECT version FROM prom_schema_migrations LIMIT 1"
	setVersion              = "INSERT INTO prom_schema_migrations (version) VALUES ($1)"
	truncateMigrationsTable = "TRUNCATE prom_schema_migrations"
	preinstallScripts = "preinstall"
	versionScripts    = "versions/dev"
	idempotentScripts = "idempotent"
)
var (
	tableOfContents = map[string][]string{
		"idempotent": {
			"base.sql",
			"tag-operators.sql",
			"matcher-functions.sql",
			"ha.sql",
			"metric-metadata.sql",
			"exemplar.sql",
			"tracing-private.sql",
			"tracing-public.sql",
			"tracing-public-views.sql",
			"apply_permissions.sql",  
		},
	}
	migrateMutex = &sync.Mutex{}
	migrationFileNameRegexp = regexp.MustCompile(`([[:digit:]]+)-[[:word:]]+.sql`)
)
type VersionInfo struct {
	Version    string
	CommitHash string
}
type prefixedName struct {
	prefix int
	name   string
}
type prefixedNames []prefixedName
func (p prefixedNames) Len() int {
	return len(p)
}
func (p prefixedNames) Less(i, j int) bool {
	return p[i].prefix < p[j].prefix
}
func (p prefixedNames) Swap(i, j int) {
	p[i], p[j] = p[j], p[i]
}
func (p prefixedNames) getNames() []string {
	names := make([]string, len(p))
	for i, e := range p {
		names[i] = e.name
	}
	return names
}
func Migrate(db *pgx.Conn, versionInfo VersionInfo, extOptions extension.ExtensionMigrateOptions) (err error) {
	migrateMutex.Lock()
	defer migrateMutex.Unlock()
	appVersion, err := semver.Make(versionInfo.Version)
	if err != nil {
		return errors.ErrInvalidSemverFormat
	}
	mig := NewMigrator(db, migrations.MigrationFiles, tableOfContents)
	err = mig.Migrate(appVersion)
	if err != nil {
		return fmt.Errorf("Error encountered during migration: %w", err)
	}
	return nil
}
func CheckDependencies(db *pgx.Conn, versionInfo VersionInfo, migrationFailedDueToLockError bool, extOptions extension.ExtensionMigrateOptions) (err error) {
	if err = CheckSchemaVersion(context.Background(), db, versionInfo, migrationFailedDueToLockError); err != nil {
		return err
	}
	return extension.CheckVersions(db, migrationFailedDueToLockError, extOptions)
}
func CheckSchemaVersion(ctx context.Context, conn *pgx.Conn, versionInfo VersionInfo, migrationFailedDueToLockError bool) error {
	expectedVersion := semver.MustParse(versionInfo.Version)
	dbVersion, err := getSchemaVersionOnConnection(ctx, conn)
	if err != nil {
		return fmt.Errorf("failed to check schema version: %w", err)
	}
	if versionCompare := dbVersion.Compare(expectedVersion); versionCompare != 0 {
		if versionCompare < 0 && migrationFailedDueToLockError {
			return fmt.Errorf("Failed to acquire the migration lock to upgrade the schema version and unable to run with the old version. Please ensure that no other Promscale connectors with the old schema version are running. Received schema version %v but expected %v", dbVersion, expectedVersion)
		}
		return fmt.Errorf("Error while comparing schema version: received schema version %v but expected %v", dbVersion, expectedVersion)
	}
	return nil
}
type Migrator struct {
	db       *pgx.Conn
	sqlFiles http.FileSystem
	toc      map[string][]string
}
func NewMigrator(db *pgx.Conn, sqlFiles http.FileSystem, toc map[string][]string) *Migrator {
	return &Migrator{db: db, sqlFiles: sqlFiles, toc: toc}
}
func (t *Migrator) Migrate(appVersion semver.Version) error {
	if err := ensureVersionTable(t.db); err != nil {
		return fmt.Errorf("error ensuring version table: %w", err)
	}
	dbVersion, err := getSchemaVersion(t.db)
	if err != nil {
		return fmt.Errorf("failed to get the version from database: %w", err)
	}
	if dbVersion.Compare(appVersion) == 0 {
		devRelease := false
		for _, pre := range appVersion.Pre {
			if pre.String() == "dev" {
				devRelease = true
			}
		}
		if devRelease {
			tx, err := t.db.Begin(context.Background())
			if err != nil {
				return fmt.Errorf("unable to start transaction: %w", err)
			}
			defer func() {
				_ = tx.Rollback(context.Background())
			}()
			if err = t.execMigrationDir(tx, idempotentScripts); err != nil {
				return err
			}
			if err = tx.Commit(context.Background()); err != nil {
				return fmt.Errorf("unable to commit migration transaction: %w", err)
			}
			return nil
		}
		return nil
	}
	if dbVersion.Compare(appVersion) > 0 {
		return fmt.Errorf("schema version (%v) is above the application version (%v), cannot migrate", dbVersion, appVersion)
	}
	tx, err := t.db.Begin(context.Background())
	if err != nil {
		return fmt.Errorf("unable to start transaction: %w", err)
	}
	defer func() {
		_ = tx.Rollback(context.Background())
	}()
	if dbVersion.Compare(semver.Version{}) == 0 {
		if err = t.execMigrationDir(tx, preinstallScripts); err != nil {
			return err
		}
	} else if err = t.upgradeVersion(tx, dbVersion, appVersion); err != nil {
		return err
	}
	if err = t.execMigrationDir(tx, idempotentScripts); err != nil {
		return err
	}
	if err = setDBVersion(tx, &appVersion); err != nil {
		return fmt.Errorf("error setting clean app version to DB: %w", err)
	}
	if err = tx.Commit(context.Background()); err != nil {
		return fmt.Errorf("unable to commit migration transaction: %w", err)
	}
	return nil
}
func ensureVersionTable(db *pgx.Conn) error {
	_, err := db.Exec(context.Background(), createMigrationsTable)
	if err != nil {
		return fmt.Errorf("error creating migration table: %w", err)
	}
	_, err = db.Exec(context.Background(), "GRANT SELECT ON prom_schema_migrations TO public")
	if err != nil {
		return fmt.Errorf("error creating migration table: %w", err)
	}
	return nil
}
func getSchemaVersion(db *pgx.Conn) (semver.Version, error) {
	return getSchemaVersionOnConnection(context.Background(), db)
}
func getSchemaVersionOnConnection(ctx context.Context, db *pgx.Conn) (semver.Version, error) {
	var version semver.Version
	res, err := db.Query(ctx, getVersion)
	if err != nil {
		return version, fmt.Errorf("Error getting DB version: %w", err)
	}
	defer res.Close()
	for res.Next() {
		err = res.Scan(&version)
	}
	if err != nil {
		return version, fmt.Errorf("Error getting DB version: %w", err)
	}
	err = res.Err()
	if err != nil {
		return version, fmt.Errorf("Error getting DB version: %w", err)
	}
	return version, nil
}
func (t *Migrator) execMigrationFile(tx pgx.Tx, fileName string) error {
	f, err := t.sqlFiles.Open(fileName)
	if err != nil {
		return fmt.Errorf("unable to get migration script: name %s, err %w", fileName, err)
	}
	contents, err := replaceSchemaNames(f)
	if err != nil {
		return fmt.Errorf("unable to read migration script: name %s, err %w", fileName, err)
	}
	_, err = tx.Exec(context.Background(), contents)
	if err != nil {
		pgErr, ok := err.(*pgconn.PgError)
		if ok && pgErr.Position > 0 {
			strC := contents
			code := strC[pgErr.Position-1:]
			return fmt.Errorf("error executing migration script: name %s, err %w, code at error position:\n  %s", fileName, err, code)
		}
		return fmt.Errorf("error executing migration script: name %s, err %w", fileName, err)
	}
	return nil
}