Repository: ariga/atlas
Pull request #1642 (2023-05-09 17:25:39)

sql/sqltool: wrap flyway dir with fs
### Extending FlywayDir Functionality with fs.FS

By wrapping `FlywayDir` with `fs.FS`, we can extend its functionality to include local directories and directories in memory or remote directories.

giautm: > Like the idea. However, I suggest to wrap `migrate.Dir` instead of `fs.FS`.

Because `migrate.Dir` has more method than `fs.FS` so we can't use FlywayDir with `os.DirFS` or other file-system.

Patch:
diff --git a/cmd/atlas/internal/cloudapi/client.go b/cmd/atlas/internal/cloudapi/client.go
index 2e7f26a6..e47de92c 100644
--- a/cmd/atlas/internal/cloudapi/client.go
+++ b/cmd/atlas/internal/cloudapi/client.go
@@ -91,6 +91,7 @@ type (
 		CurrentVersion string              `json:"currentVersion"`
 		Error          *string             `json:"error,omitempty"`
 		Files          []DeployedFileInput `json:"files"`
+		Log            string              `json:"log"`
 	}
 
 	// DeployedTargetInput represents the input type for a deployed target.
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index 8e24963e..170892f5 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -120,9 +120,7 @@ If run with the "--dry-run" flag, atlas will not execute any SQL.`,
 					return migrateApplyRun(cmd, args, flags, &MigrateReport{}) // nop reporter
 				}
 				set := NewMigrateReportSet()
-				defer func() {
-					err = errors.Join(err, set.Flush(cmd.Context()))
-				}()
+				defer set.Flush(cmd)
 				return cmdEnvsRun(setMigrateEnvFlags, cmd, func(env *Env) error {
 					return migrateApplyRun(cmd, args, flags, set.ReportFor(env))
 				})
@@ -310,15 +308,21 @@ func (s *MigrateReportSet) ReportFor(e *Env) *MigrateReport {
 // Flush report the migration deployment to the cloud.
 // The current implementation is simplistic and sends each
 // report separately without marking them as part of a group.
-func (s *MigrateReportSet) Flush(ctx context.Context) error {
-	var errs []error
+//
+// Note that reporting errors are logged, but not cause Atlas to fail.
+func (s *MigrateReportSet) Flush(cmd *cobra.Command) {
 	for _, r := range s.reports {
 		client := r.e.cfg.Client
-		if err := client.ReportMigration(ctx, *r.r); err != nil {
-			errs = append(errs, err)
+		if err := client.ReportMigration(cmd.Context(), *r.r); err != nil {
+			txt := fmt.Sprintf("Error: %s", strings.TrimRight(err.Error(), "\n"))
+			// Ensure errors are printed in new lines.
+			if cmd.Flags().Changed(flagFormat) {
+				txt = "\n" + txt
+			}
+			cmd.PrintErrln(txt)
 		}
 	}
-	return errors.Join(errs...)
+	return
 }
 
 // Init the report if the necessary dependencies.
@@ -341,11 +345,14 @@ func (r *MigrateReport) RecordTargetID(ctx context.Context) error {
 
 // Done closes and flushes this report.
 func (r *MigrateReport) Done(cmd *cobra.Command, flags migrateApplyFlags) error {
-	err := logApply(cmd, flags, r.log)
 	if !r.CloudEnabled() {
-		return err
+		return logApply(cmd, cmd.OutOrStdout(), flags, r.log)
 	}
-	var ver string
+	var (
+		ver  string
+		clog bytes.Buffer
+		err  = logApply(cmd, io.MultiWriter(cmd.OutOrStdout(), &clog), flags, r.log)
+	)
 	switch rev, err1 := r.rrw.CurrentRevision(cmd.Context()); {
 	case ent.IsNotFound(err1):
 	case err1 != nil:
@@ -389,6 +396,7 @@ func (r *MigrateReport) Done(cmd *cobra.Command, flags migrateApplyFlags) error
 			}
 			return files
 		}(),
+		Log: clog.String(),
 	})
 	return err
 }
@@ -398,7 +406,7 @@ func (r *MigrateReport) CloudEnabled() bool {
 	return r.env != nil && r.env.cfg != nil && r.env.cfg.Client != nil && r.env.cfg.Project != ""
 }
 
-func logApply(cmd *cobra.Command, flags migrateApplyFlags, r *cmdlog.MigrateApply) error {
+func logApply(cmd *cobra.Command, w io.Writer, flags migrateApplyFlags, r *cmdlog.MigrateApply) error {
 	var (
 		err error
 		f   = cmdlog.MigrateApplyTemplate
@@ -409,7 +417,7 @@ func logApply(cmd *cobra.Command, flags migrateApplyFlags, r *cmdlog.MigrateAppl
 			return fmt.Errorf("parse format: %w", err)
 		}
 	}
-	if err = f.Execute(cmd.OutOrStdout(), r); err != nil {
+	if err = f.Execute(w, r); err != nil {
 		return fmt.Errorf("execute log template: %w", err)
 	}
 	// In case a custom logging was configured, avoid reporting errors twice.
diff --git a/sql/sqltool/tool.go b/sql/sqltool/tool.go
index a0768751..debb9070 100644
--- a/sql/sqltool/tool.go
+++ b/sql/sqltool/tool.go
@@ -7,6 +7,7 @@ package sqltool
 import (
 	"bufio"
 	"bytes"
+	"errors"
 	"fmt"
 	"io/fs"
 	"path/filepath"
@@ -290,9 +291,9 @@ func (f *DBMateFile) Stmts() ([]string, error) {
 }
 
 type (
-	// FlywayDir wraps migrate.LocalDir and provides a migrate.Scanner implementation able to understand files
+	// FlywayDir wraps fs.FS and provides a migrate.Scanner implementation able to understand files
 	// generated by the FlywayFormatter for migration directory replaying.
-	FlywayDir struct{ *migrate.LocalDir }
+	FlywayDir struct{ fs.FS }
 	// FlywayFile wraps migrate.LocalFile with custom statements function.
 	FlywayFile struct{ *migrate.LocalFile }
 )
@@ -311,12 +312,16 @@ func NewFlywayDir(path string) (*FlywayDir, error) {
 // and filename.
 func (d *FlywayDir) Files() ([]migrate.File, error) {
 	var ff flywayFiles
-	if err := fs.WalkDir(d, "", func(path string, e fs.DirEntry, err error) error {
+	if err := fs.WalkDir(d, ".", func(path string, e fs.DirEntry, err error) error {
 		if err != nil {
 			return err
 		}
-		if path != "" && e.IsDir() {
-			h, err := hidden(filepath.Join(d.Path(), path))
+		if path != "." && e.IsDir() {
+			fullPath := path
+			if p, ok := d.FS.(interface{ Path() string }); ok {
+				fullPath = filepath.Join(p.Path(), path)
+			}
+			h, err := hidden(fullPath)
 			if err != nil {
 				return err
 			}
@@ -351,6 +356,26 @@ func (d *FlywayDir) Files() ([]migrate.File, error) {
 	return ret, nil
 }
 
+// Checksum implements Dir.Checksum. By default, it calls Files() and creates a checksum from them.
+func (d *FlywayDir) Checksum() (migrate.HashFile, error) {
+	if d, ok := d.FS.(migrate.Dir); ok {
+		return d.Checksum()
+	}
+	files, err := d.Files()
+	if err != nil {
+		return nil, err
+	}
+	return migrate.NewHashFile(files)
+}
+
+// WriteFile implements Dir.WriteFile.
+func (d *FlywayDir) WriteFile(name string, b []byte) error {
+	if d, ok := d.FS.(migrate.Dir); ok {
+		return d.WriteFile(name, b)
+	}
+	return errors.New("sql/sqltool: write not supported")
+}
+
 // Desc implements File.Desc.
 func (f FlywayFile) Desc() string {
 	return flywayDesc(f.Name())


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index d5b76941..0c548ecc 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -875,6 +875,7 @@ env {
 				Schema: "main",
 				URL:    u,
 			},
+			Log: "No migration files to execute\n",
 		}, report)
 	})
 
@@ -907,7 +908,7 @@ env {
 		}
 	})
 
-	t.Run("Error", func(t *testing.T) {
+	t.Run("PrintError", func(t *testing.T) {
 		status = http.StatusInternalServerError
 		require.NoError(t, dir.WriteFile("3.sql", []byte("create table baz (id int)")))
 		cmd := migrateCmd()
@@ -919,9 +920,23 @@ env {
 			"--url", u,
 			"--var", "cloud_url="+srv.URL,
 		)
-		require.EqualError(t, err, "unexpected status code: 500")
+		require.NoError(t, err)
 		// Reporting error should not affect the migration execution.
 		require.True(t, strings.HasSuffix(s, "  -- 1 migrations \n  -- 1 sql statements\nError: unexpected status code: 500\n"))
+
+		// Custom logging.
+		cmd = migrateCmd()
+		cmd.AddCommand(migrateApplyCmd())
+		s, err = runCmd(
+			cmd, "apply",
+			"-c", "file://"+path,
+			"--env", "local",
+			"--url", u,
+			"--var", "cloud_url="+srv.URL,
+			"--format", "{{ .Env.Driver }}",
+		)
+		require.NoError(t, err)
+		require.Equal(t, "sqlite3\nError: unexpected status code: 500\n", s)
 	})
 }
 
diff --git a/sql/sqltool/testdata/flyway/v3/V3_1__fourth_migration.sql b/sql/sqltool/testdata/flyway/v3/V3_1__fourth_migration.sql
new file mode 100644
index 00000000..bd094311
--- /dev/null
+++ b/sql/sqltool/testdata/flyway/v3/V3_1__fourth_migration.sql
@@ -0,0 +1,1 @@
+ALTER TABLE tbl_2 ADD col_2 INTEGER NOT NULL;
\ No newline at end of file
diff --git a/sql/sqltool/tool_test.go b/sql/sqltool/tool_test.go
index 83cea39b..da21c963 100644
--- a/sql/sqltool/tool_test.go
+++ b/sql/sqltool/tool_test.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io/fs"
 	"testing"
+	"testing/fstest"
 	"time"
 
 	"ariga.io/atlas/sql/migrate"
@@ -206,14 +207,15 @@ func TestScanners(t *testing.T) {
 				require.NoError(t, err)
 				return d
 			}(),
-			versions:     []string{"2", "3", ""},
-			descriptions: []string{"baseline", "third_migration", "views"},
+			versions:     []string{"2", "3", "3_1", ""},
+			descriptions: []string{"baseline", "third_migration", "fourth_migration", "views"},
 			stmts: [][]string{
 				{
 					"CREATE TABLE post\n(\n    id    int NOT NULL,\n    title text,\n    body  text,\n    created_at TIMESTAMP NOT NULL\n    PRIMARY KEY (id)\n);",
 					"INSERT INTO post (title, created_at) VALUES (\n'This is\nmy multiline\n\nvalue', NOW());",
 				},
 				{"ALTER TABLE tbl_2 ADD col_1 INTEGER NOT NULL;"},
+				{"ALTER TABLE tbl_2 ADD col_2 INTEGER NOT NULL;"},
 				{"CREATE VIEW `my_view` AS SELECT * FROM `post`;"},
 			},
 		},
@@ -320,6 +322,27 @@ func TestChecksum(t *testing.T) {
 				"V3__third_migration.sql",
 			},
 		},
+		{
+			name: "flyway non-local directory",
+			dir: func() migrate.Dir {
+				fs := fstest.MapFS{
+					"U1__initial.sql":               &fstest.MapFile{Data: []byte("U1__initial")},
+					"V1__initial.sql":               &fstest.MapFile{Data: []byte("V1__initial")},
+					"R__views.sql":                  &fstest.MapFile{Data: []byte("R__views")},
+					"B2__baseline.sql":              &fstest.MapFile{Data: []byte("B2__baseline")},
+					"V2__second_migration.sql":      &fstest.MapFile{Data: []byte("V2__second_migration")},
+					"V3__third_migration.sql":       &fstest.MapFile{Data: []byte("V3__third_migration")},
+					"v3/V3_1__fourth_migration.sql": &fstest.MapFile{Data: []byte("V3__1_fourth_migration")},
+				}
+				return &sqltool.FlywayDir{&fs}
+			}(),
+			files: []string{
+				"B2__baseline.sql",
+				"V3__third_migration.sql",
+				"v3/V3_1__fourth_migration.sql",
+				"R__views.sql",
+			},
+		},
 		{
 			name: "liquibase",
 			dir: func() migrate.Dir {