Migrations functionality

gbus/abstractions.go

@@ -205,6 +205,7 @@ type TxProvider interface {
 	New() (*sql.Tx, error)
 	Dispose()
 	Ping(timeoutInSeconds time.Duration) bool
+	GetDb() *sql.DB
 }
 
 //TxOutbox abstracts the transactional outgoing channel type

gbus/builder/builder.go

@@ -80,6 +80,7 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 				panic(err)
 			}
 			gb.TxProvider = mysqltx
+			mysql.EnsureSchema(mysqltx.GetDb(), gb.SvcName)
 			sagaStore = mysql.NewSagaStore(gb.SvcName, mysqltx)
 			if builder.purgeOnStartup {
 				err := sagaStore.Purge()

gbus/tx/mysql/migrations.go

@@ -0,0 +1,74 @@
+package mysql
+
+import (
+	"database/sql"
+
+	"fmt"
+	"github.com/lopezator/migrator"
+	"github.com/wework/grabbit/gbus/tx"
+)
+
+func SagaStoreTableMigration(svcName string) *migrator.Migration {
+	tblName := tx.GetSagatableName(svcName)
+
+	createTableQuery := `CREATE TABLE IF NOT EXISTS ` + tblName + ` (
+		rec_id INT PRIMARY KEY AUTO_INCREMENT,
+		saga_id VARCHAR(255) UNIQUE NOT NULL,
+		saga_type VARCHAR(255)  NOT NULL,
+		saga_data LONGBLOB NOT NULL,
+		version integer NOT NULL DEFAULT 0,
+		last_update timestamp  DEFAULT NOW(),
+		INDEX ` + tblName + `_sagatype_idx (saga_type))`
+
+	return &migrator.Migration{
+		Name: "create saga store table",
+		Func: func(tx *sql.Tx) error {
+			if _, err := tx.Exec(createTableQuery); err != nil {
+				return err
+			}
+			return nil
+		},
+	}
+}
+
+func OutboxMigrations(svcName string) *migrator.Migration {
+
+	query := `CREATE TABLE IF NOT EXISTS ` + getOutboxName(svcName) + ` (
+	rec_id int NOT NULL AUTO_INCREMENT,
+	message_id varchar(50) NOT NULL UNIQUE,
+	message_type varchar(50) NOT NULL,
+	exchange	varchar(50) NOT NULL,
+	routing_key	varchar(50) NOT NULL,
+	publishing	longblob NOT NULL,
+	status	int(11) NOT NULL,
+	relay_id varchar(50)  NULL,
+	delivery_tag	bigint(20) NOT NULL,
+	delivery_attempts int NOT NULL DEFAULT 0,
+	insert_date	timestamp DEFAULT CURRENT_TIMESTAMP,
+	PRIMARY KEY(rec_id),
+	INDEX status_delivery (rec_id, status, delivery_attempts))`
+
+	return &migrator.Migration{
+		Name: "create outbox table",
+		Func: func(tx *sql.Tx) error {
+			if _, err := tx.Exec(query); err != nil {
+				return err
+			}
+			return nil
+		},
+	}
+}
+
+func EnsureSchema(db *sql.DB, svcName string) {
+
+	migrationsTable := fmt.Sprintf("grabbitMigrations_%s", svcName)
+
+	migrate := migrator.NewNamed(migrationsTable,
+		OutboxMigrations(svcName),
+		SagaStoreTableMigration(svcName),
+	)
+	err := migrate.Migrate(db)
+	if err != nil {
+		panic(err)
+	}
+}

gbus/tx/mysql/sagastore.go

@@ -1,9 +1,6 @@
 package mysql
 
 import (
-	"database/sql"
-	"fmt"
-
 	log "github.com/sirupsen/logrus"
 	"github.com/wework/grabbit/gbus"
 	"github.com/wework/grabbit/gbus/saga"
@@ -19,81 +16,6 @@ func (store *SagaStore) log() *log.Entry {
 	return log.WithField("saga_store", "mysql")
 }
 
-func (store *SagaStore) ensureSchema() {
-	store.log().Info("ensuring saga schema exists")
-	if tablesExists := store.sagaTablesExist(); !tablesExists {
-
-		store.log().Info("could not find saga schema, attempting to creat schema")
-
-		store.createSagaTables()
-	}
-}
-
-func (store *SagaStore) sagaTablesExist() bool {
-
-	tblName := store.GetSagatableName()
-	tx := store.NewTx()
-	defer func() {
-		err := tx.Commit()
-		if err != nil {
-			store.log().WithError(err).Error("could not commit sagaTablesExist")
-		}
-	}()
-
-	selectSQL := `SELECT 1 FROM ` + tblName + ` LIMIT 1;`
-
-	store.log().Info(selectSQL)
-
-	row := tx.QueryRow(selectSQL)
-	var exists int
-	err := row.Scan(&exists)
-	if err != nil && err != sql.ErrNoRows {
-		return false
-	}
-
-	return true
-}
-
-func (store *SagaStore) createSagaTables() {
-	tblName := store.GetSagatableName()
-	tx := store.NewTx()
-
-	createTable := `CREATE TABLE ` + tblName + ` (
-		rec_id INT PRIMARY KEY AUTO_INCREMENT,
-		saga_id VARCHAR(255) UNIQUE NOT NULL,
-		saga_type VARCHAR(255)  NOT NULL,
-		saga_data LONGBLOB NOT NULL,
-		version integer NOT NULL DEFAULT 0,
-		last_update timestamp  DEFAULT NOW()
-		)`
-	createSagaTypeIndex := `CREATE INDEX ` + tblName + `_sagatype_idx ON ` + tblName + ` (saga_type)`
-
-	sqls := [2]string{
-		createTable,
-		createSagaTypeIndex}
-
-	for i, sql := range sqls {
-		_, error := tx.Exec(sql)
-
-		store.log().WithFields(log.Fields{"step": i, "sql": sql}).Info("creating saga tables")
-
-		if error != nil {
-			txErr := fmt.Errorf("failed to create saga tables.\n%v", error)
-			if rollbackErr := tx.Rollback(); rollbackErr != nil {
-				log.WithError(rollbackErr).Error("could not rollback transaction")
-			}
-			panic(txErr)
-		}
-	}
-
-	error := tx.Commit()
-	if error != nil {
-		txErr := fmt.Errorf("failed to create saga tables.\n%v", error)
-		panic(txErr)
-	}
-
-}
-
 //NewSagaStore creates a bew SagaStore
 func NewSagaStore(svcName string, txProvider gbus.TxProvider) saga.Store {
 
@@ -103,7 +25,6 @@ func NewSagaStore(svcName string, txProvider gbus.TxProvider) saga.Store {
 		ParamsMarkers: getParamsMarker()}
 	store := &SagaStore{
 		base}
-	store.ensureSchema()
 	return store
 }
 

gbus/tx/mysql/txoutbox.go

@@ -5,15 +5,14 @@ import (
 	"database/sql"
 	"encoding/gob"
 	"fmt"
-	"strconv"
-	"strings"
-	"sync"
-	"time"
-
 	"github.com/rs/xid"
 	log "github.com/sirupsen/logrus"
 	"github.com/streadway/amqp"
 	"github.com/wework/grabbit/gbus"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
 )
 
 var (
@@ -55,13 +54,6 @@ func (outbox *TxOutbox) Start(amqpOut *gbus.AMQPOutbox) error {
 	if e != nil {
 		panic(fmt.Sprintf("passed in transaction provider failed with the following error\n%s", e))
 	}
-	if ensureErr := outbox.ensureSchema(tx, outbox.svcName); ensureErr != nil {
-		err := tx.Rollback()
-		if err != nil {
-			outbox.log().WithError(err).Error("could not rollback the transaction for creation of schemas")
-		}
-		return ensureErr
-	}
 	if outbox.purgeOnStartup {
 		if purgeErr := outbox.purge(tx); purgeErr != nil {
 			outbox.log().WithError(purgeErr).Error("failed to purge transactional outbox")
@@ -321,75 +313,6 @@ func (outbox *TxOutbox) sendMessages(recordSelector func(tx *sql.Tx) (*sql.Rows,
 	return nil
 }
 
-func (outbox *TxOutbox) ensureSchema(tx *sql.Tx, svcName string) error {
-
-	schemaExists := outbox.outBoxTablesExists(tx, svcName)
-
-	if schemaExists {
-		/*
-			The following  performs an alter schema to accommodate for breaking change introduced in commit 6a9f5df
-			so that earlier consumers of grabbit will not break once the upgrade to the 1.0.0 release.
-			Once a proper DB migration stratagy will be in place and implemented (post 1.0.0) the following code
-			will be deleted.
-		*/
-
-		return outbox.migrate0_9To1_0(tx, svcName)
-	}
-
-	createTablesSQL := `CREATE TABLE IF NOT EXISTS ` + getOutboxName(svcName) + ` (
-	rec_id int NOT NULL AUTO_INCREMENT,
-	message_id varchar(50) NOT NULL UNIQUE,
-	message_type varchar(50) NOT NULL,
-	exchange	varchar(50) NOT NULL,
-	routing_key	varchar(50) NOT NULL,
-	publishing	longblob NOT NULL,
-	status	int(11) NOT NULL,
-	relay_id varchar(50)  NULL,
-	delivery_tag	bigint(20) NOT NULL,
-	delivery_attempts int NOT NULL DEFAULT 0,
-	insert_date	timestamp DEFAULT CURRENT_TIMESTAMP,
-	PRIMARY KEY(rec_id),
-	INDEX status_delivery (rec_id, status, delivery_attempts))`
-
-	_, createErr := tx.Exec(createTablesSQL)
-
-	return createErr
-
-}
-
-func (outbox *TxOutbox) outBoxTablesExists(tx *sql.Tx, svcName string) bool {
-
-	tblName := getOutboxName(svcName)
-
-	selectSQL := `SELECT 1 FROM ` + tblName + ` LIMIT 1;`
-
-	outbox.log().Info(selectSQL)
-
-	row := tx.QueryRow(selectSQL)
-	var exists int
-	err := row.Scan(&exists)
-	if err != nil && err != sql.ErrNoRows {
-		return false
-	}
-
-	return true
-}
-
-func (outbox *TxOutbox) migrate0_9To1_0(tx *sql.Tx, svcName string) error {
-	tblName := getOutboxName(svcName)
-	alter := `ALTER TABLE ` + tblName + ` CHANGE COLUMN delivery_attemtps delivery_attempts int NOT NULL DEFAULT 0;`
-	_, execErr := tx.Exec(alter)
-	if execErr != nil {
-		outbox.log().WithField("sql_err", execErr).Info("migration:renaming column")
-	}
-	addIndex := `ALTER TABLE ` + tblName + ` ADD INDEX status_delivery (rec_id, status, delivery_attempts);`
-	_, indexErr := tx.Exec(addIndex)
-	if indexErr != nil {
-		outbox.log().WithField("sql_err", execErr).Info("migration:adding index column")
-	}
-	return nil
-}
-
 func getOutboxName(svcName string) string {
 
 	return strings.ToLower("grabbit_" + sanitizeTableName(svcName) + "_outbox")

gbus/tx/provider.go

@@ -38,3 +38,8 @@ func (provider *Provider) Ping(timeoutInSeconds time.Duration) bool {
 
 	return hasPing
 }
+
+func (provider *Provider) GetDb() *sql.DB {
+	return provider.Database
+
+}

gbus/tx/sagastore.go

@@ -62,7 +62,7 @@ func (store *SagaStore) scanInstances(rows *sql.Rows) ([]*saga.Instance, error)
 //GetSagasByType implements interface method store.GetSagasByType
 func (store *SagaStore) GetSagasByType(tx *sql.Tx, sagaType reflect.Type) (instances []*saga.Instance, err error) {
 
-	tblName := store.GetSagatableName()
+	tblName := GetSagatableName(store.SvcName)
 	selectSQL := "SELECT saga_id, saga_type, saga_data, version FROM " + tblName + " WHERE saga_type=" + store.ParamsMarkers[0]
 
 	rows, err := tx.Query(selectSQL, sagaType.String())
@@ -86,7 +86,7 @@ func (store *SagaStore) GetSagasByType(tx *sql.Tx, sagaType reflect.Type) (insta
 
 //UpdateSaga implements interface method store.UpdateSaga
 func (store *SagaStore) UpdateSaga(tx *sql.Tx, instance *saga.Instance) (err error) {
-	tblName := store.GetSagatableName()
+	tblName := GetSagatableName(store.SvcName)
 	currentVersion := instance.ConcurrencyCtrl
 	nextVersion := instance.ConcurrencyCtrl + 1
 	instance.ConcurrencyCtrl = nextVersion
@@ -115,7 +115,7 @@ func (store *SagaStore) RegisterSagaType(saga gbus.Saga) {
 
 //DeleteSaga implements interface method store.DeleteSaga
 func (store *SagaStore) DeleteSaga(tx *sql.Tx, instance *saga.Instance) error {
-	tblName := store.GetSagatableName()
+	tblName := GetSagatableName(store.SvcName)
 	deleteSQL := `DELETE FROM ` + tblName + ` WHERE saga_id= ?`
 	result, err := tx.Exec(deleteSQL, instance.ID)
 	if err != nil {
@@ -132,7 +132,7 @@ func (store *SagaStore) DeleteSaga(tx *sql.Tx, instance *saga.Instance) error {
 //GetSagaByID implements interface method store.GetSagaByID
 func (store *SagaStore) GetSagaByID(tx *sql.Tx, sagaID string) (*saga.Instance, error) {
 
-	tblName := store.GetSagatableName()
+	tblName := GetSagatableName(store.SvcName)
 	selectSQL := `SELECT saga_id, saga_type, saga_data, version FROM ` + tblName + ` WHERE saga_id=` + store.ParamsMarkers[0] + ``
 
 	rows, err := tx.Query(selectSQL, sagaID)
@@ -144,7 +144,7 @@ func (store *SagaStore) GetSagaByID(tx *sql.Tx, sagaID string) (*saga.Instance,
 	}()
 	if err != nil {
 		store.log().WithError(err).
-			WithFields(log.Fields{"saga_id": sagaID, "table_name": store.GetSagatableName()}).
+			WithFields(log.Fields{"saga_id": sagaID, "table_name": GetSagatableName(store.SvcName)}).
 			Error("Failed to fetch saga")
 
 		return nil, err
@@ -162,7 +162,7 @@ func (store *SagaStore) GetSagaByID(tx *sql.Tx, sagaID string) (*saga.Instance,
 //SaveNewSaga implements interface method store.SaveNewSaga
 func (store *SagaStore) SaveNewSaga(tx *sql.Tx, sagaType reflect.Type, newInstance *saga.Instance) (err error) {
 	store.RegisterSagaType(newInstance.UnderlyingInstance)
-	tblName := store.GetSagatableName()
+	tblName := GetSagatableName(store.SvcName)
 	insertSQL := `INSERT INTO ` + tblName + ` (saga_id, saga_type, saga_data, version) VALUES (?, ?, ?, ?)`
 
 	var buf []byte
@@ -181,8 +181,8 @@ func (store *SagaStore) SaveNewSaga(tx *sql.Tx, sagaType reflect.Type, newInstan
 //Purge cleans up the saga store, to be used in tests and in extreme situations in production
 func (store *SagaStore) Purge() error {
 	tx := store.NewTx()
-	store.log().WithField("saga_table", store.GetSagatableName()).Info("Purging saga table")
-	deleteSQL := fmt.Sprintf("DELETE FROM %s", store.GetSagatableName())
+	store.log().WithField("saga_table", GetSagatableName(store.SvcName)).Info("Purging saga table")
+	deleteSQL := fmt.Sprintf("DELETE FROM %s", GetSagatableName(store.SvcName))
 	results, err := tx.Exec(deleteSQL)
 	if err != nil {
 		store.log().WithError(err).Error("failed to purge saga table")
@@ -220,10 +220,10 @@ func (store *SagaStore) NewTx() *sql.Tx {
 }
 
 //GetSagatableName returns the table name in which to store the Sagas
-func (store *SagaStore) GetSagatableName() string {
+func GetSagatableName(svcName string) string {
 
 	var re = regexp.MustCompile(`-|;|\\|`)
-	sanitized := re.ReplaceAllString(store.SvcName, "")
+	sanitized := re.ReplaceAllString(svcName, "")
 
 	return strings.ToLower("grabbit_" + sanitized + "_sagas")
 }