Merge pull request #2 from getkalido/feature/run_one_migration_in_a_b…

…atch

Refactored the migrate method to support a one migration per batch mode

migrations.go

@@ -57,26 +57,52 @@ func Register(name string, up, down func(*pg.Tx) error) {
 	}
 }
 
-func Run(db *pg.DB, cmd, name, template string) error {
+/*
+Run Runs the specified command with the options they require
+Note:
+	init - no options
+	migrate - one option
+		- "" for all migrations in a single batch
+		- "one-by-one" for one migration in a batch mode
+	rollback - no options
+	create - two options
+		- name - name of the migration (must be first)
+		- template - string that contains the go code to use as a template. see migrationTemplate
+*/
+func Run(db *pg.DB, cmd string, options ...string) error {
 	switch cmd {
 	case "init":
 		return initialise(db)
+
 	case "migrate":
-		return migrate(db)
+		extra := ""
+		if len(options) > 0 {
+			extra = options[0]
+		}
+		return migrate(db, extra == "one-by-one")
+
 	case "rollback":
 		return rollback(db)
+
 	case "create":
+		name := ""
+		template := ""
+		if len(options) > 0 {
+			name = options[0]
+		}
+		if len(options) > 1 {
+			template = options[1]
+		}
 		if len(name) == 0 {
-			return errors.New("Please enter migration description")
+			return errors.New("Please enter migration name")
 		}
 
 		name = strings.Replace(name, " ", "_", -1)
 
 		return create(name, template)
-	default:
-		return errors.Errorf("unsupported command: %q", cmd)
-
 	}
+
+	return errors.Errorf("unsupported command: %q", cmd)
 }
 
 func initialise(db *pg.DB) error {
@@ -127,68 +153,135 @@ func initialise(db *pg.DB) error {
 	})
 }
 
-func migrate(db *pg.DB) error {
-	return db.RunInTransaction(func(tx *pg.Tx) (err error) {
-
-		err = lockTable(tx)
-
-		if err != nil {
-			return
-		}
+func getMigrationsToRun(tx *pg.Tx) ([]string, error) {
+	var migrations []string
 
-		var migrations []string
+	migrations, err := getCompletedMigrations(tx)
+	if err != nil {
+		return nil, err
+	}
 
-		migrations, err = getCompletedMigrations(tx)
+	missingMigrations := difference(migrations, migrationNames)
+	if len(missingMigrations) > 0 {
+		return nil, errors.Errorf("Migrations table corrupt: %+v", missingMigrations)
+	}
 
-		if err != nil {
-			return
-		}
+	migrationsToRun := difference(migrationNames, migrations)
 
-		missingMigrations := difference(migrations, migrationNames)
+	if len(migrationsToRun) > 0 {
+		sort.Strings(migrationsToRun)
+	}
 
-		if len(missingMigrations) > 0 {
-			return errors.Errorf("Migrations table corrupt: %+v", missingMigrations)
-		}
+	return migrationsToRun, nil
+}
+func migrate(db *pg.DB, oneByOne bool) error {
+	if oneByOne {
+		return migrateOneByOne(db)
+	}
+	return migrateOneBatch(db)
+}
 
-		migrationsToRun := difference(migrationNames, migrations)
+func migrateOneByOne(db *pg.DB) error {
 
-		if len(migrationsToRun) > 0 {
-			var batch int
-			batch, err = getBatchNumber(tx)
+	var migrationsToRun []string
 
+	err := db.RunInTransaction(
+		func(tx *pg.Tx) (err error) {
+			err = lockTable(tx)
 			if err != nil {
 				return
 			}
 
-			batch++
+			migrationsToRun, err = getMigrationsToRun(tx)
+			return
+		})
 
-			sort.Slice(migrationsToRun, func(i, j int) bool {
-				switch strings.Compare(migrationsToRun[i], migrationsToRun[j]) {
-				case -1:
-					return true
-				case 1:
-					return false
-				}
-				return true
-			})
+	if err != nil {
+		return err
+	}
 
-			fmt.Printf("Batch %d run: %d migrations\n", batch, len(migrationsToRun))
+	if len(migrationsToRun) == 0 {
+		return nil
+	}
 
-			for _, migration := range migrationsToRun {
-				err = allMigrations[migration].Up(tx)
+	for _, migration := range migrationsToRun {
+		err := db.RunInTransaction(
+			func(tx *pg.Tx) (err error) {
+				err = lockTable(tx)
+				if err != nil {
+					return
+				}
 
+				var batch int
+				batch, err = getBatchNumber(tx)
 				if err != nil {
-					err = errors.Wrapf(err, "%s failed to migrate", migration)
 					return
 				}
 
-				err = insertCompletedMigration(tx, migration, batch)
+				batch++
+
+				fmt.Printf("Batch %d run: 1 migration - %s\n", batch, migration)
 
+				err = allMigrations[migration].Up(tx)
 				if err != nil {
+					err = errors.Wrapf(err, "%s failed to migrate", migration)
 					return
 				}
+
+				err = insertCompletedMigration(tx, migration, batch)
+				return
+			})
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func migrateOneBatch(db *pg.DB) error {
+	return db.RunInTransaction(func(tx *pg.Tx) (err error) {
+
+		err = lockTable(tx)
+		if err != nil {
+			return
+		}
+
+		var migrationsToRun []string
+		migrationsToRun, err = getMigrationsToRun(tx)
+		if err != nil {
+			return
+		}
+
+		if len(migrationsToRun) == 0 {
+			return
+		}
+
+		var batch int
+		batch, err = getBatchNumber(tx)
+		if err != nil {
+			return
+		}
+
+		batch++
+
+		fmt.Printf("Batch %d run: %d migrations\n", batch, len(migrationsToRun))
+
+		for _, migration := range migrationsToRun {
+			err = allMigrations[migration].Up(tx)
+
+			if err != nil {
+				err = errors.Wrapf(err, "%s failed to migrate", migration)
+				return
+			}
+
+			err = insertCompletedMigration(tx, migration, batch)
+
+			if err != nil {
+				return
 			}
 		}
+
 		return
 	})
 }