[WIP] first push

Author: martinyonatann

.gitignore

@@ -0,0 +1,25 @@
+# If you prefer the allow list template instead of the deny list, see community template:
+# https://github.com/github/gitignore/blob/main/community/Golang/Go.AllowList.gitignore
+#
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/
+
+# Go workspace file
+go.work
+go.work.sum
+
+# env file
+.env

cmd/mysql-schema-diff/apply_cmd.go

@@ -0,0 +1,172 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/DoWithLogic/mysql-schema-diff/pkg/diff"
+	_ "github.com/DoWithLogic/mysql-schema-diff/pkg/diff"
+	"github.com/DoWithLogic/mysql-schema-diff/pkg/log"
+	"github.com/go-sql-driver/mysql"
+	"github.com/spf13/cobra"
+)
+
+func buildApplyCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "apply",
+		Short: "Migrate your database to the match the inputted schema (apply the schema to the database)",
+	}
+
+	connFlags := createConnFlags(cmd)
+	planFlags := createPlanFlags(cmd)
+	allowedHazardsTypesStrs := cmd.Flags().StringSlice("allow-hazards", nil,
+		"Specify the hazards that are allowed. Order does not matter, and duplicates are ignored. If the"+
+			" migration plan contains unwanted hazards (hazards not in this list), then the migration will fail to run"+
+			" (example: --allowed-hazards DELETES_DATA,INDEX_BUILD)")
+	skipConfirmPrompt := cmd.Flags().Bool("skip-confirm-prompt", false, "Skips prompt asking for user to confirm before applying")
+	cmd.RunE = func(cmd *cobra.Command, args []string) error {
+		logger := log.SimpleLogger()
+
+		connConfig, err := parseConnConfig(*connFlags, logger)
+		if err != nil {
+			return err
+		}
+
+		planConfig, err := parsePlanConfig(*planFlags)
+		if err != nil {
+			return err
+		}
+
+		cmd.SilenceUsage = true
+
+		plan, err := generatePlan(context.Background(), logger, connConfig, planConfig)
+		if err != nil {
+			return err
+		} else if len(plan.Statements) == 0 {
+			fmt.Println("Schema matches expected. No plan generated")
+			return nil
+		}
+
+		fmt.Println(header("Review plan"))
+		fmt.Print(planToPrettyS(plan), "\n\n")
+
+		if err := failIfHazardsNotAllowed(plan, *allowedHazardsTypesStrs); err != nil {
+			return err
+		}
+
+		if !*skipConfirmPrompt {
+			if err := mustContinuePrompt(
+				fmt.Sprintf(
+					"Apply migration with the following hazards: %s?",
+					strings.Join(*allowedHazardsTypesStrs, ", "),
+				),
+			); err != nil {
+				return err
+			}
+		}
+
+		if err := runPlan(context.Background(), connConfig, plan); err != nil {
+			return err
+		}
+		fmt.Println("Schema applied successfully")
+		return nil
+	}
+
+	return cmd
+}
+
+func failIfHazardsNotAllowed(plan diff.Plan, allowedHazardsTypesStrs []string) error {
+	isAllowedByHazardType := make(map[diff.MigrationHazardType]bool)
+	for _, val := range allowedHazardsTypesStrs {
+		isAllowedByHazardType[strings.ToUpper(val)] = true
+	}
+	var disallowedHazardMsgs []string
+	for i, stmt := range plan.Statements {
+		var disallowedTypes []diff.MigrationHazardType
+		for _, hzd := range stmt.Hazards {
+			if !isAllowedByHazardType[hzd.Type] {
+				disallowedTypes = append(disallowedTypes, hzd.Type)
+			}
+		}
+		if len(disallowedTypes) > 0 {
+			disallowedHazardMsgs = append(disallowedHazardMsgs,
+				fmt.Sprintf("- Statement %d: %s", getDisplayableStmtIdx(i), strings.Join(disallowedTypes, ", ")),
+			)
+		}
+
+	}
+	if len(disallowedHazardMsgs) > 0 {
+		return errors.New(fmt.Sprintf(
+			"Prohited hazards found\n"+
+				"These hazards must be allowed via the allow-hazards flag, e.g., --allow-hazards %s\n"+
+				"Prohibited hazards in the following statements:\n%s",
+			strings.Join(getHazardTypes(plan), ","),
+			strings.Join(disallowedHazardMsgs, "\n"),
+		))
+	}
+	return nil
+}
+
+func runPlan(ctx context.Context, connConfig *mysql.Config, plan diff.Plan) error {
+	connPool, err := newMySQL(connConfig.FormatDSN())
+	if err != nil {
+		return err
+	}
+	defer connPool.Close()
+
+	conn, err := connPool.Conn(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	// Due to the way *sql.Db works, when a statement_timeout is set for the session, it will NOT reset
+	// by default when it's returned to the pool.
+	//
+	// We can't set the timeout at the TRANSACTION-level (for each transaction) because `ADD INDEX CONCURRENTLY`
+	// must be executed within its own transaction block. Postgres will error if you try to set a TRANSACTION-level
+	// timeout for it. SESSION-level statement_timeouts are respected by `ADD INDEX CONCURRENTLY`
+	for i, stmt := range plan.Statements {
+		fmt.Println(header(fmt.Sprintf("Executing statement %d", getDisplayableStmtIdx(i))))
+		fmt.Printf("%s\n\n", statementToPrettyS(stmt))
+		start := time.Now()
+		if _, err := conn.ExecContext(ctx, fmt.Sprintf("SET SESSION statement_timeout = %d", stmt.Timeout.Milliseconds())); err != nil {
+			return fmt.Errorf("setting statement timeout: %w", err)
+		}
+		if _, err := conn.ExecContext(ctx, fmt.Sprintf("SET SESSION lock_timeout = %d", stmt.Timeout.Milliseconds())); err != nil {
+			return fmt.Errorf("setting lock timeout: %w", err)
+		}
+		if _, err := conn.ExecContext(ctx, stmt.ToSQL()); err != nil {
+			return fmt.Errorf("executing migration statement. the database maybe be in a dirty state: %s: %w", stmt, err)
+		}
+		fmt.Printf("Finished executing statement. Duration: %s\n", time.Since(start))
+	}
+	fmt.Println(header("Complete"))
+
+	return nil
+}
+
+func getHazardTypes(plan diff.Plan) []diff.MigrationHazardType {
+	seenHazardTypes := make(map[diff.MigrationHazardType]bool)
+	var hazardTypes []diff.MigrationHazardType
+	for _, stmt := range plan.Statements {
+		for _, hazard := range stmt.Hazards {
+			if !seenHazardTypes[hazard.Type] {
+				seenHazardTypes[hazard.Type] = true
+				hazardTypes = append(hazardTypes, hazard.Type)
+			}
+		}
+	}
+	sort.Slice(hazardTypes, func(i, j int) bool {
+		return hazardTypes[i] < hazardTypes[j]
+	})
+	return hazardTypes
+}
+
+func getDisplayableStmtIdx(i int) int {
+	return i + 1
+}

cmd/mysql-schema-diff/cli.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"fmt"
+	"math"
+	"strings"
+
+	"github.com/manifoldco/promptui"
+)
+
+func header(header string) string {
+	const headerTargetWidth = 80
+
+	if len(header) > headerTargetWidth {
+		return header
+	}
+
+	if len(header) > 0 {
+		header = fmt.Sprintf(" %s ", header)
+	}
+	hashTagsOnSide := int(math.Ceil(float64(headerTargetWidth-len(header)) / 2))
+
+	rightHashTags := strings.Repeat("#", hashTagsOnSide)
+	leftHashTags := rightHashTags
+	if headerTargetWidth-len(header)-2*hashTagsOnSide > 0 {
+		leftHashTags += "#"
+	}
+	return fmt.Sprintf("%s%s%s", leftHashTags, header, rightHashTags)
+}
+
+// MustContinuePrompt prompts the user if they want to continue, and returns an error otherwise.
+// promptui requires the ContinueLabel to be one line
+func mustContinuePrompt(continueLabel string) error {
+	if len(continueLabel) == 0 {
+		continueLabel = "Continue?"
+	}
+	if _, result, err := (&promptui.Select{
+		Label: continueLabel,
+		Items: []string{"No", "Yes"},
+	}).Run(); err != nil {
+		return err
+	} else if result == "No" {
+		return fmt.Errorf("user aborted")
+	}
+	return nil
+}

cmd/mysql-schema-diff/datastructs.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"fmt"
+	"sort"
+)
+
+func mustGetAndDeleteKey(m map[string]string, key string) (string, error) {
+	val, ok := m[key]
+	if !ok {
+		return "", fmt.Errorf("could not find key %q", key)
+	}
+	delete(m, key)
+
+	return val, nil
+}
+
+func keys(m map[string]string) []string {
+	var vals []string
+	for k := range m {
+		vals = append(vals, k)
+	}
+	sort.Strings(vals)
+
+	return vals
+}

cmd/mysql-schema-diff/datastructs_test.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestKeys(t *testing.T) {
+	for _, tt := range []struct {
+		name string
+		m    map[string]string
+
+		want []string
+	}{
+		{
+			name: "nil map",
+
+			want: nil,
+		},
+		{
+			name: "empty map",
+
+			want: nil,
+		},
+		{
+			name: "filled map",
+			m: map[string]string{
+				// Use an arbitrary order
+				"key2": "value2",
+				"key3": "value3",
+				"key1": "value1",
+			},
+
+			want: []string{"key1", "key2", "key3"},
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, keys(tt.m))
+		})
+	}
+}

cmd/mysql-schema-diff/flags.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/DoWithLogic/mysql-schema-diff/pkg/log"
+	"github.com/go-logfmt/logfmt"
+	"github.com/go-sql-driver/mysql"
+	"github.com/spf13/cobra"
+)
+
+type connFlags struct {
+	dsn string
+}
+
+type Config struct {
+	Host     string
+	Port     uint16
+	Database string
+	User     string
+	Password string
+}
+
+func createConnFlags(cmd *cobra.Command) *connFlags {
+	flags := new(connFlags)
+
+	// Don't mark dsn as a required flag.
+	// Allow users to user the MYSQLHOST etc environment variables like `mysql`
+	cmd.Flags().StringVar(&flags.dsn, "dsn", "", "Connection string for the database (DB password can be specified through MYSQLPASSWORD environment variable)")
+
+	return flags
+}
+
+func parseConnConfig(c connFlags, logger log.Logger) (*mysql.Config, error) {
+	if c.dsn == "" {
+		logger.Warnf("DSN flag not set. Using libpq environment variables and default values.")
+	}
+
+	cfg, err := mysql.ParseDSN(c.dsn)
+	if err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
+}
+
+func logFmtToMap(logFmt string) (map[string]string, error) {
+	logMap := make(map[string]string)
+	decoder := logfmt.NewDecoder(strings.NewReader(logFmt))
+	for decoder.ScanRecord() {
+		if _, ok := logMap[string(decoder.Key())]; ok {
+			return nil, fmt.Errorf("duplicate key %q in logfmt", string(decoder.Key()))
+		}
+		logMap[string(decoder.Key())] = string(decoder.Value())
+	}
+
+	if decoder.Err() != nil {
+		return nil, decoder.Err()
+	}
+
+	return logMap, nil
+}