feat: automate business upgrades and sync reviewed frontend
This commit is contained in:
Yvan
@@ -0,0 +1,317 @@
|
||||
package swmigration
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"math/big"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func quoteName(value string) string { return "`" + strings.ReplaceAll(value, "`", "``") + "`" }
|
||||
|
||||
// backupDatabase writes affected data to a private, durable SQL archive.
|
||||
// Large raw-data tables are not rewritten by the upgrade; their original
|
||||
// schemas are included as comments. tab_promote is included with data because
|
||||
// its primary-key width changes. Cleanup snapshots include every drop target.
|
||||
func backupDatabase(ctx context.Context, db *gorm.DB, root, identity, phase string, cleanupTargets map[string]bool, logger *slog.Logger) (path, digest string, err error) {
|
||||
if root == "" {
|
||||
return "", "", fmt.Errorf("a persistent configuration directory is required for migration backups")
|
||||
}
|
||||
if err = os.MkdirAll(root, 0700); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
info, err := os.Lstat(root)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 {
|
||||
return "", "", fmt.Errorf("migration backup location must be a private directory")
|
||||
}
|
||||
if err = os.Chmod(root, 0700); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
key := sha256.Sum256([]byte(identity))
|
||||
name := fmt.Sprintf("sw-%x-%s-%d.sql.gz", key[:8], phase, time.Now().UnixNano())
|
||||
path = filepath.Join(root, name)
|
||||
partial := path + ".partial"
|
||||
file, err := os.OpenFile(partial, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
defer func() {
|
||||
_ = file.Close()
|
||||
if err != nil {
|
||||
_ = os.Remove(partial)
|
||||
}
|
||||
}()
|
||||
hash := sha256.New()
|
||||
compressed, err := gzip.NewWriterLevel(io.MultiWriter(file, hash), gzip.BestSpeed)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
defer compressed.Close()
|
||||
writer := bufio.NewWriterSize(compressed, 128*1024)
|
||||
write := func(value string) error { _, err := writer.WriteString(value); return err }
|
||||
if err = write("-- Automatic SW upgrade backup. Restore only in a maintenance session.\n-- Contains affected/retired data; untouched raw-data schemas are comments.\nSET @sw_restore_mode=@@SESSION.sql_mode;\nSET @sw_restore_fk=@@SESSION.foreign_key_checks;\nSET @sw_restore_tz=@@SESSION.time_zone;\nSET SESSION sql_mode='NO_AUTO_VALUE_ON_ZERO';\nSET SESSION foreign_key_checks=0;\nSET SESSION time_zone='+00:00';\nSET NAMES utf8mb4;\n"); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
pool, err := db.DB()
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
conn, err := pool.Conn(ctx)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
defer conn.Close()
|
||||
var oldTimeZone, oldSQLMode, oldClientCharset, oldCollation string
|
||||
var oldResultCharset sql.NullString
|
||||
if err = conn.QueryRowContext(ctx, "SELECT @@SESSION.time_zone,@@SESSION.sql_mode,@@SESSION.character_set_client,@@SESSION.character_set_results,@@SESSION.collation_connection").Scan(&oldTimeZone, &oldSQLMode, &oldClientCharset, &oldResultCharset, &oldCollation); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if _, err = conn.ExecContext(ctx, "SET SESSION time_zone='+00:00',sql_mode='NO_AUTO_VALUE_ON_ZERO'"); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
defer func() {
|
||||
clean, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
||||
defer cancel()
|
||||
var resultCharset any
|
||||
if oldResultCharset.Valid {
|
||||
resultCharset = oldResultCharset.String
|
||||
}
|
||||
_, restoreErr := conn.ExecContext(clean, "SET SESSION time_zone=?,sql_mode=?,character_set_client=?,character_set_results=?,collation_connection=?", oldTimeZone, oldSQLMode, oldClientCharset, resultCharset, oldCollation)
|
||||
if err == nil {
|
||||
err = restoreErr
|
||||
}
|
||||
}()
|
||||
if _, err = conn.ExecContext(ctx, "SET NAMES utf8mb4"); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
tx, err := conn.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true})
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
var schema string
|
||||
if err = tx.QueryRowContext(ctx, "SELECT DATABASE()").Scan(&schema); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
rows, err := tx.QueryContext(ctx, "SELECT table_name FROM information_schema.tables WHERE table_schema=DATABASE() AND table_type='BASE TABLE' ORDER BY table_name")
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
var tables []string
|
||||
for rows.Next() {
|
||||
var table string
|
||||
if err = rows.Scan(&table); err != nil {
|
||||
rows.Close()
|
||||
return "", "", err
|
||||
}
|
||||
if table != TableName {
|
||||
tables = append(tables, table)
|
||||
}
|
||||
}
|
||||
err = rows.Err()
|
||||
rows.Close()
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
for _, table := range tables {
|
||||
if err = ctx.Err(); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if cleanupTargets != nil && !cleanupTargets[table] {
|
||||
continue
|
||||
}
|
||||
var unused, ddl string
|
||||
if err = tx.QueryRowContext(ctx, "SHOW CREATE TABLE "+quoteName(table)).Scan(&unused, &ddl); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
includeData := cleanupTargets != nil || !strings.HasPrefix(table, "tab_") || table == "tab_promote"
|
||||
if !includeData {
|
||||
if err = write("-- Retained raw table; schema reference only:\n-- " + strings.ReplaceAll(ddl, "\n", "\n-- ") + ";\n"); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err = write("DROP TABLE IF EXISTS " + quoteName(table) + ";\n" + ddl + ";\n"); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
columns, types, columnErr := backupColumns(ctx, tx, schema, table)
|
||||
if columnErr != nil {
|
||||
return "", "", columnErr
|
||||
}
|
||||
quoted := make([]string, len(columns))
|
||||
for i, col := range columns {
|
||||
quoted[i] = quoteName(col)
|
||||
}
|
||||
prefix := "INSERT INTO " + quoteName(table) + " (" + strings.Join(quoted, ",") + ") VALUES ("
|
||||
dataRows, queryErr := tx.QueryContext(ctx, "SELECT "+strings.Join(quoted, ",")+" FROM "+quoteName(table))
|
||||
if queryErr != nil {
|
||||
return "", "", queryErr
|
||||
}
|
||||
var count int64
|
||||
for dataRows.Next() {
|
||||
values := make([]any, len(columns))
|
||||
dest := make([]any, len(columns))
|
||||
for i := range values {
|
||||
dest[i] = &values[i]
|
||||
}
|
||||
if err = dataRows.Scan(dest...); err != nil {
|
||||
dataRows.Close()
|
||||
return "", "", err
|
||||
}
|
||||
literals := make([]string, len(values))
|
||||
for i, value := range values {
|
||||
literals[i], err = backupLiteral(value, types[i])
|
||||
if err != nil {
|
||||
dataRows.Close()
|
||||
return "", "", fmt.Errorf("backup %s.%s: %w", table, columns[i], err)
|
||||
}
|
||||
}
|
||||
if err = write(prefix + strings.Join(literals, ",") + ");\n"); err != nil {
|
||||
dataRows.Close()
|
||||
return "", "", err
|
||||
}
|
||||
count++
|
||||
}
|
||||
err = dataRows.Err()
|
||||
dataRows.Close()
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
logger.Info("migration backup table completed", "table", table, "rows", count, "phase", phase)
|
||||
}
|
||||
if err = tx.Commit(); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if err = write("SET SESSION foreign_key_checks=@sw_restore_fk;\nSET SESSION sql_mode=@sw_restore_mode;\nSET SESSION time_zone=@sw_restore_tz;\n-- SW BACKUP COMPLETE\n"); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if err = writer.Flush(); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if err = compressed.Close(); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if err = file.Sync(); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if err = file.Close(); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if err = os.Rename(partial, path); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
digest = hex.EncodeToString(hash.Sum(nil))
|
||||
if err = verifyBackup(path, digest); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
logger.Info("migration backup verified", "phase", phase, "path", path, "sha256", digest)
|
||||
return path, digest, nil
|
||||
}
|
||||
|
||||
func backupColumns(ctx context.Context, tx *sql.Tx, schema, table string) ([]string, []string, error) {
|
||||
rows, err := tx.QueryContext(ctx, "SELECT column_name,data_type,extra FROM information_schema.columns WHERE table_schema=? AND table_name=? ORDER BY ordinal_position", schema, table)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var names, types []string
|
||||
for rows.Next() {
|
||||
var name, kind, extra string
|
||||
if err := rows.Scan(&name, &kind, &extra); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if strings.Contains(extra, "VIRTUAL GENERATED") || strings.Contains(extra, "STORED GENERATED") {
|
||||
continue
|
||||
}
|
||||
names = append(names, name)
|
||||
types = append(types, kind)
|
||||
}
|
||||
return names, types, rows.Err()
|
||||
}
|
||||
|
||||
func backupLiteral(value any, kind string) (string, error) {
|
||||
if value == nil {
|
||||
return "NULL", nil
|
||||
}
|
||||
var raw []byte
|
||||
switch v := value.(type) {
|
||||
case []byte:
|
||||
raw = v
|
||||
case string:
|
||||
raw = []byte(v)
|
||||
case time.Time:
|
||||
if v.IsZero() {
|
||||
raw = []byte("0000-00-00 00:00:00")
|
||||
} else {
|
||||
raw = []byte(v.Format("2006-01-02 15:04:05.999999"))
|
||||
}
|
||||
case int64:
|
||||
raw = []byte(strconv.FormatInt(v, 10))
|
||||
case uint64:
|
||||
raw = []byte(strconv.FormatUint(v, 10))
|
||||
case float64:
|
||||
raw = []byte(strconv.FormatFloat(v, 'g', -1, 64))
|
||||
case bool:
|
||||
if v {
|
||||
raw = []byte("1")
|
||||
} else {
|
||||
raw = []byte("0")
|
||||
}
|
||||
default:
|
||||
return "", fmt.Errorf("unsupported SQL value %T", value)
|
||||
}
|
||||
if kind == "bit" {
|
||||
if v, ok := value.(int64); ok {
|
||||
return strconv.FormatInt(v, 10), nil
|
||||
}
|
||||
if v, ok := value.(uint64); ok {
|
||||
return strconv.FormatUint(v, 10), nil
|
||||
}
|
||||
return new(big.Int).SetBytes(raw).String(), nil
|
||||
}
|
||||
encoded := "FROM_BASE64('" + base64.StdEncoding.EncodeToString(raw) + "')"
|
||||
switch kind {
|
||||
case "binary", "varbinary", "blob", "tinyblob", "mediumblob", "longblob":
|
||||
return encoded, nil
|
||||
case "geometry", "point", "linestring", "polygon", "multipoint", "multilinestring", "multipolygon", "geometrycollection":
|
||||
return "", fmt.Errorf("unsupported geometry backup type %s", kind)
|
||||
default:
|
||||
return "CONVERT(" + encoded + " USING utf8mb4)", nil
|
||||
}
|
||||
}
|
||||
|
||||
func verifyBackup(path, expected string) error {
|
||||
if path == "" || len(expected) != 64 {
|
||||
return fmt.Errorf("verified migration backup is missing")
|
||||
}
|
||||
file, err := os.Open(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open migration backup: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
hash := sha256.New()
|
||||
if _, err = io.Copy(hash, file); err != nil {
|
||||
return err
|
||||
}
|
||||
if hex.EncodeToString(hash.Sum(nil)) != expected {
|
||||
return fmt.Errorf("migration backup checksum mismatch: %s", path)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user