246 lines
6.6 KiB
Go
246 lines
6.6 KiB
Go
// Package migrate provides explicit MySQL schema migration jobs. Application
|
|
// startup never runs migrations automatically; use a separate deployment step.
|
|
package migrate
|
|
|
|
import (
|
|
"archive/zip"
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
mysqlDriver "github.com/go-sql-driver/mysql"
|
|
migration "github.com/golang-migrate/migrate/v4"
|
|
migrateMySQL "github.com/golang-migrate/migrate/v4/database/mysql"
|
|
"github.com/golang-migrate/migrate/v4/source/iofs"
|
|
"io"
|
|
"io/fs"
|
|
"path"
|
|
"regexp"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
type File struct {
|
|
Version uint64 `json:"version"`
|
|
Name string `json:"name"`
|
|
Direction string `json:"direction"`
|
|
SHA256 string `json:"sha256"`
|
|
}
|
|
|
|
var filename = regexp.MustCompile(`^([0-9]+)_([A-Za-z][A-Za-z0-9]*)\.(up|down)\.sql$`)
|
|
|
|
// Snapshot returns immutable migration bytes and a reviewable digest manifest.
|
|
// Follow lowerCamelCase names, e.g. 0001_createUsers.up.sql.
|
|
func Snapshot(source fs.FS, directory string) (fs.FS, []File, error) {
|
|
if source == nil || !fs.ValidPath(directory) {
|
|
return nil, nil, errors.New("migrate: invalid source directory")
|
|
}
|
|
entries, err := fs.ReadDir(source, directory)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
var archive bytes.Buffer
|
|
writer := zip.NewWriter(&archive)
|
|
manifest := []File{}
|
|
seen := map[string]bool{}
|
|
total := 0
|
|
for _, entry := range entries {
|
|
if entry.IsDir() {
|
|
continue
|
|
}
|
|
parts := filename.FindStringSubmatch(entry.Name())
|
|
if parts == nil {
|
|
if strings.HasSuffix(entry.Name(), ".sql") {
|
|
return nil, nil, fmt.Errorf("migrate: invalid migration filename %s", entry.Name())
|
|
}
|
|
continue
|
|
}
|
|
if len(manifest) >= 1000 {
|
|
return nil, nil, errors.New("migrate: too many files")
|
|
}
|
|
version, err := strconv.ParseUint(parts[1], 10, 32)
|
|
if err != nil || version == 0 {
|
|
return nil, nil, errors.New("migrate: version must be 1..4294967295")
|
|
}
|
|
identity := fmt.Sprint(version) + parts[3]
|
|
if seen[identity] {
|
|
return nil, nil, errors.New("migrate: duplicate migration version/direction")
|
|
}
|
|
seen[identity] = true
|
|
file, err := source.Open(path.Join(directory, entry.Name()))
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
data, err := io.ReadAll(io.LimitReader(file, (1<<20)+1))
|
|
closeErr := file.Close()
|
|
if err == nil {
|
|
err = closeErr
|
|
}
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
total += len(data)
|
|
if len(data) == 0 || len(data) > 1<<20 || total > 16<<20 {
|
|
return nil, nil, errors.New("migrate: files must be 1..1MiB and <=16MiB total")
|
|
}
|
|
sum := sha256.Sum256(data)
|
|
manifest = append(manifest, File{Version: version, Name: parts[2], Direction: parts[3], SHA256: hex.EncodeToString(sum[:])})
|
|
fileWriter, err := writer.CreateHeader(&zip.FileHeader{Name: entry.Name(), Method: zip.Store})
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if _, err = fileWriter.Write(data); err != nil {
|
|
return nil, nil, err
|
|
}
|
|
}
|
|
if len(manifest) == 0 {
|
|
return nil, nil, errors.New("migrate: no SQL migrations")
|
|
}
|
|
for _, file := range manifest {
|
|
if file.Direction == "down" && !seen[fmt.Sprint(file.Version)+"up"] {
|
|
return nil, nil, errors.New("migrate: down file lacks up migration")
|
|
}
|
|
}
|
|
sort.Slice(manifest, func(i, j int) bool {
|
|
if manifest[i].Version == manifest[j].Version {
|
|
return manifest[i].Direction < manifest[j].Direction
|
|
}
|
|
return manifest[i].Version < manifest[j].Version
|
|
})
|
|
if err = writer.Close(); err != nil {
|
|
return nil, nil, err
|
|
}
|
|
files, err := zip.NewReader(bytes.NewReader(archive.Bytes()), int64(archive.Len()))
|
|
return files, manifest, err
|
|
}
|
|
|
|
type Runner struct {
|
|
mu sync.Mutex
|
|
engine *migration.Migrate
|
|
db *sql.DB
|
|
closed bool
|
|
stopped bool
|
|
}
|
|
|
|
func NewMySQL(ctx context.Context, dsn string, source fs.FS, directory string) (runner *Runner, err error) {
|
|
if ctx == nil {
|
|
return nil, errors.New("migrate: context required")
|
|
}
|
|
snapshot, _, err := Snapshot(source, directory)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
cfg, err := mysqlDriver.ParseDSN(dsn)
|
|
if err != nil || cfg.DBName == "" {
|
|
return nil, errors.New("migrate: valid MySQL DSN with database required")
|
|
}
|
|
cfg.MultiStatements = true
|
|
if cfg.Timeout == 0 || cfg.Timeout > 5*time.Second {
|
|
cfg.Timeout = 5 * time.Second
|
|
}
|
|
if cfg.ReadTimeout == 0 || cfg.ReadTimeout > 30*time.Second {
|
|
cfg.ReadTimeout = 30 * time.Second
|
|
}
|
|
if cfg.WriteTimeout == 0 || cfg.WriteTimeout > 30*time.Second {
|
|
cfg.WriteTimeout = 30 * time.Second
|
|
}
|
|
connector, err := mysqlDriver.NewConnector(cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db := sql.OpenDB(connector)
|
|
db.SetMaxOpenConns(1)
|
|
db.SetMaxIdleConns(1)
|
|
defer func() {
|
|
if err != nil {
|
|
db.Close()
|
|
}
|
|
}()
|
|
conn, err := db.Conn(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
driver, err := migrateMySQL.WithConnection(ctx, conn, &migrateMySQL.Config{DatabaseName: cfg.DBName, StatementTimeout: 30 * time.Second})
|
|
if err != nil {
|
|
conn.Close()
|
|
return nil, err
|
|
}
|
|
src, err := iofs.New(snapshot, ".")
|
|
if err != nil {
|
|
driver.Close()
|
|
return nil, err
|
|
}
|
|
engine, err := migration.NewWithInstance("iofs", src, cfg.DBName, driver)
|
|
if err != nil {
|
|
src.Close()
|
|
driver.Close()
|
|
return nil, err
|
|
}
|
|
engine.LockTimeout = 10 * time.Second
|
|
return &Runner{engine: engine, db: db}, nil
|
|
}
|
|
|
|
// Up applies pending up scripts. MySQL DDL can partially commit; a failed
|
|
// migration remains dirty and requires operator investigation. No automatic
|
|
// down/force/dirty reset is provided. Cancellation stops after the current SQL
|
|
// operation; statements/read I/O are bounded to 30 seconds.
|
|
func (r *Runner) Up(ctx context.Context) error {
|
|
if ctx == nil {
|
|
return errors.New("migrate: context required")
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if r.closed || r.stopped {
|
|
return errors.New("migrate: closed")
|
|
}
|
|
done := make(chan error, 1)
|
|
go func() { done <- r.engine.Up() }()
|
|
var err error
|
|
select {
|
|
case err = <-done:
|
|
case <-ctx.Done():
|
|
r.stopped = true // golang-migrate retains its graceful-stop state; create a fresh runner to retry.
|
|
select {
|
|
case r.engine.GracefulStop <- true:
|
|
default:
|
|
}
|
|
err = <-done
|
|
return errors.Join(ctx.Err(), err)
|
|
}
|
|
if errors.Is(err, migration.ErrNoChange) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
func (r *Runner) Version() (uint, bool, error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if r.closed {
|
|
return 0, false, errors.New("migrate: closed")
|
|
}
|
|
version, dirty, err := r.engine.Version()
|
|
if errors.Is(err, migration.ErrNilVersion) {
|
|
return 0, false, nil
|
|
}
|
|
return version, dirty, err
|
|
}
|
|
func (r *Runner) Close() error {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if r.closed {
|
|
return nil
|
|
}
|
|
r.closed = true
|
|
a, b := r.engine.Close()
|
|
return errors.Join(a, b, r.db.Close())
|
|
}
|