wip

Author: tmus

Migrator.go

@@ -16,15 +16,27 @@ func NewMigrator(driver Driver) (*Migrator, error) {
 	return m, nil
 }
 
-// TODO replace dir with opts, which includes direction and fresh commands, maybe more.
-func (m *Migrator) Run(dir string, migrations ...Migration) error {
+func (m *Migrator) Run(args []string, migrations ...Migration) error {
+	opts := gatherOptions(args)
+
 	if err := m.driver.Init(); err != nil {
 		return fmt.Errorf("cannot initialise migration driver: %w", err)
 	}
 
-	if err := m.driver.Run(migrations); err != nil {
+	if err := m.driver.Run(opts, migrations); err != nil {
 		return fmt.Errorf("unable to run migrations: %w", err)
 	}
 
 	return nil
 }
+
+func gatherOptions(args []string) Options {
+	dir := Unknown
+	if len(args) > 0 {
+		dir = directionFromString(args[1])
+	}
+
+	return Options{
+		direction: dir,
+	}
+}

driver.go

@@ -5,7 +5,7 @@ type Driver interface {
 	Init() error
 	// Run takes a slice of exodus.Migration and runs it against the
 	// underlying data store defined in the driver.
-	Run(payload []Migration) error
+	Run(options Options, payload []Migration) error
 	// Close closes the underlying database instance.
 	Close() error
 }

driver/mysql/const.go

@@ -1,6 +1,7 @@
 package mysql
 
 const (
+	dropTableSchema   = `DROP TABLE %s;`
 	renameTableSchema = `ALTER TABLE %s RENAME TO %s;`
 	createTableSchema = `CREATE TABLE %s (
 	%s

driver/mysql/mysql.go

@@ -35,7 +35,6 @@ func NewDriver(datasource string) (*mysqlDriver, error) {
 }
 
 func (d mysqlDriver) Init() error {
-	d.fresh()
 	// Before running migrations, make sure that the migrations table exists on
 	// the underlying database. This table is used to track which migrations
 	// have already been ran. If it doesn't exist, then create it.
@@ -52,7 +51,18 @@ func (d mysqlDriver) Close() error {
 	return d.db.Close()
 }
 
-func (d mysqlDriver) Run(migrations []exodus.Migration) error {
+func (d mysqlDriver) Run(opts exodus.Options, migrations []exodus.Migration) error {
+	switch opts.Direction() {
+	case exodus.Up:
+		return d.runUp(opts, migrations)
+	case exodus.Down:
+		return d.runDown(opts, migrations)
+	default:
+		return errors.New("not running migrations, direction not specified")
+	}
+}
+
+func (d mysqlDriver) runUp(opts exodus.Options, migrations []exodus.Migration) error {
 	// First, retrieve the list of migrations that have previously been ran. These
 	// migrations are then used to determine which of the incoming migrations
 	// should be ran against the database.
@@ -78,6 +88,29 @@ func (d mysqlDriver) Run(migrations []exodus.Migration) error {
 	return nil
 }
 
+func (d mysqlDriver) runDown(opts exodus.Options, migrations []exodus.Migration) error {
+	// First, retrieve the list of migrations that have previously been ran. These
+	// will be used to retrieve the Down method of each migration to run.
+	previous, err := d.getLastBatchRan()
+	if err != nil {
+		return fmt.Errorf("unable to process migrations: %w", err)
+	}
+	migrations = filterRanMigrations(migrations, previous)
+
+	if len(migrations) == 0 {
+		d.log.Info().Msg("Nothing to rollback.")
+		return nil
+	}
+
+	for _, migration := range migrations {
+		if err := d.processDown(migration); err != nil {
+			return fmt.Errorf("unable to execute SQL: %w", err)
+		}
+	}
+
+	return nil
+}
+
 func (d mysqlDriver) getRan() ([]string, error) {
 	rows, err := d.db.Query("SELECT migration FROM migrations")
 	if err != nil {
@@ -99,6 +132,27 @@ func (d mysqlDriver) getRan() ([]string, error) {
 	return ran, nil
 }
 
+func (d mysqlDriver) getLastBatchRan() ([]string, error) {
+	rows, err := d.db.Query("SELECT migration FROM migrations WHERE batch = (SELECT MAX(batch) FROM migrations)")
+	if err != nil {
+		return []string{}, fmt.Errorf("unable to get previous migrations from database: %w", err)
+	}
+
+	var ran []string
+	for rows.Next() {
+		var migration string
+		if err := rows.Scan(&migration); err != nil {
+			return []string{}, fmt.Errorf("unable to get previous migrations from database: %w", err)
+		}
+		ran = append(ran, migration)
+	}
+	if err := rows.Err(); err != nil {
+		return []string{}, fmt.Errorf("unable to get previous migrations from database: %w", err)
+	}
+
+	return ran, nil
+}
+
 func (d mysqlDriver) process(migration exodus.Migration, batch int) error {
 	builder := &exodus.MigrationPayload{}
 	migration.Up(builder)
@@ -125,6 +179,36 @@ func (d mysqlDriver) process(migration exodus.Migration, batch int) error {
 	return nil
 }
 
+func (d mysqlDriver) processDown(migration exodus.Migration) error {
+	builder := &exodus.MigrationPayload{}
+	migration.Down(builder)
+	start := time.Now()
+	d.log.Info().Msgf("Rolling back: %s", reflect.TypeOf(migration).String())
+	for _, p := range builder.Operations() {
+		switch p.Operation() {
+		case exodus.CREATE_TABLE:
+			if err := d.createTable(p); err != nil {
+				return err
+			}
+		case exodus.RENAME_TABLE:
+			if err := d.renameTable(p); err != nil {
+				return err
+			}
+		case exodus.DROP_TABLE:
+			if err := d.dropTable(p); err != nil {
+				return err
+			}
+		default:
+			return errors.New("operation not supported")
+		}
+	}
+
+	d.removeMigrationLog(migration)
+
+	d.log.Info().Msgf("Rolled back: %s in %v", reflect.TypeOf(migration).String(), time.Since(start))
+	return nil
+}
+
 func (d mysqlDriver) logMigration(migration exodus.Migration, batch int) error {
 	stmt, err := d.db.Prepare("INSERT INTO migrations (migration, batch) VALUES ( ?, ? )")
 	if err != nil {
@@ -139,6 +223,20 @@ func (d mysqlDriver) logMigration(migration exodus.Migration, batch int) error {
 	return nil
 }
 
+func (d mysqlDriver) removeMigrationLog(migration exodus.Migration) error {
+	stmt, err := d.db.Prepare("DELETE FROM migrations WHERE migration = ?")
+	if err != nil {
+		log.Fatalln("Cannot create `migrations` remove statement. ")
+	}
+	defer stmt.Close()
+
+	if _, err = stmt.Exec(reflect.TypeOf(migration).String()); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // nextBatchNumber retreives the highest batch number from the
 // migrations table and increments it by one.
 func (d mysqlDriver) nextBatchNumber() int {
@@ -154,6 +252,17 @@ func (d mysqlDriver) lastBatchNumber() int {
 	return num
 }
 
+func (d mysqlDriver) dropTable(payload *exodus.MigrationOperation) error {
+	table := payload.Table()
+	sql := fmt.Sprintf(dropTableSchema, table)
+
+	if _, err := d.db.Exec(sql); err != nil {
+		return fmt.Errorf("unable to drop table `%s`: %w", table, err)
+	}
+
+	return nil
+}
+
 func (d mysqlDriver) renameTable(payload *exodus.MigrationOperation) error {
 	from := payload.Table()
 	to := payload.Payload().(string)
@@ -254,6 +363,18 @@ func filterPendingMigrations(migrations []exodus.Migration, existing []string) [
 	return response
 }
 
+func filterRanMigrations(migrations []exodus.Migration, existing []string) []exodus.Migration {
+	var response []exodus.Migration
+
+	for _, migration := range migrations {
+		if exists(migration, existing) {
+			response = append(response, migration)
+		}
+	}
+
+	return response
+}
+
 func exists(migration exodus.Migration, existing []string) bool {
 	for _, ex := range existing {
 		if reflect.TypeOf(migration).String() == ex {

options.go

@@ -0,0 +1,29 @@
+package exodus
+
+type direction int
+
+const (
+	Unknown direction = iota
+	Up
+	Down
+)
+
+func directionFromString(dir string) direction {
+	if dir == "up" {
+		return Up
+	}
+
+	if dir == "down" {
+		return Down
+	}
+
+	return Unknown
+}
+
+type Options struct {
+	direction direction
+}
+
+func (o Options) Direction() direction {
+	return o.direction
+}

usage/migrations/20220115115948_create_users_table.go

@@ -20,5 +20,5 @@ func (migration20220115115948create_users_table) Up(schema *exodus.MigrationPayl
 }
 
 func (migration20220115115948create_users_table) Down(schema *exodus.MigrationPayload) {
-
+	schema.Drop("example")
 }

usage/migrations/20220120041331_create_posts_table.go

@@ -14,4 +14,6 @@ func (migration20220120041331create_posts_table) Up(schema *exodus.MigrationPayl
 	})
 }
 
-func (migration20220120041331create_posts_table) Down(schema *exodus.MigrationPayload) {}
+func (migration20220120041331create_posts_table) Down(schema *exodus.MigrationPayload) {
+	schema.Drop("posts")
+}