func (store *SagaStore) NewTx() *sql.Tx {
	tx, err := store.Tx.New()
	if err != nil {
		e := fmt.Errorf("can't initialize sage store.\nerror:\n%s", err)
		panic(e)
	}
	return tx
}
NewTx creates a new transaction from the underlying TxProvider
package tx
import (
	"bytes"
	"database/sql"
	"encoding/gob"
	"errors"
	"fmt"
	"reflect"
	"regexp"
	"strings"
	"time"
	log "github.com/sirupsen/logrus"
	"github.com/go-sql-driver/mysql"
	"github.com/wework/grabbit/gbus"
	"github.com/wework/grabbit/gbus/saga"
)
type SagaStore struct {
	*gbus.Glogged
	Tx            gbus.TxProvider
	SvcName       string
	ParamsMarkers []string
}
func (store *SagaStore) scanInstances(rows *sql.Rows) ([]*saga.Instance, error) {
	instances := make([]*saga.Instance, 0)
	for rows.Next() {
		var sagaID, sagaType string
		var version int
		var sagaData []byte
		var startedBy sql.NullString
		var startedBySaga sql.NullString
		var startedByMsgID sql.NullString
		var startedByRPCID sql.NullString
		var createdAt mysql.NullTime
		error := rows.Scan(&sagaID, &sagaType, &sagaData, &startedBy, &startedByMsgID, &startedByRPCID, &startedBySaga, &version, &createdAt)
		if error == sql.ErrNoRows {
			return nil, error
		} else if error != nil {
			store.Log().WithError(error).Error("failed to scan saga row")
			return nil, error
		}
		reader := bytes.NewReader(sagaData)
		dec := gob.NewDecoder(reader)
		var instance saga.Instance
		instance.ConcurrencyCtrl = version
		decErr := dec.Decode(&instance)
		if startedBy.Valid {
			instance.StartedBy = startedBy.String
		}
		if startedBySaga.Valid {
			instance.StartedBySaga = startedBySaga.String
		}
		if startedByMsgID.Valid {
			instance.StartedByMessageID = startedByMsgID.String
		}
		if startedByRPCID.Valid {
			instance.StartedByRPCID = startedByRPCID.String
		}
		if createdAt.Valid {
			value, _ := createdAt.Value()
			instance.CreatedAt = value.(time.Time)
		}
		if decErr != nil {
			store.Log().WithError(decErr).Error("failed to decode saga instance")
			return nil, decErr
		}
		instances = append(instances, &instance)
	}
	return instances, nil
}
func (store *SagaStore) GetSagasByType(tx *sql.Tx, sagaType reflect.Type) (instances []*saga.Instance, err error) {
	tblName := GetSagatableName(store.SvcName)
	selectSQL := "SELECT saga_id, saga_type, saga_data, started_by_request_of_svc, started_by_msg_id, started_by_rpcid, started_by_request_of_saga, version, created_at FROM " + tblName + " WHERE saga_type=" + store.ParamsMarkers[0]
	rows, err := tx.Query(selectSQL, sagaType.String())
	defer func() {
		err := rows.Close()
		if err != nil {
			store.Log().WithError(err).Error("could not close rows")
		}
	}()
	if err != nil {
		return nil, err
	}
	if instances, err = store.scanInstances(rows); err != nil {
		store.Log().WithError(err).Error("SagaStore failed to scan saga db record")
		return nil, err
	}
	return instances, nil
}
func (store *SagaStore) UpdateSaga(tx *sql.Tx, instance *saga.Instance) (err error) {
	tblName := GetSagatableName(store.SvcName)
	currentVersion := instance.ConcurrencyCtrl
	nextVersion := instance.ConcurrencyCtrl + 1
	instance.ConcurrencyCtrl = nextVersion
	var buf []byte
	if buf, err = store.serilizeSaga(instance); err != nil {
		store.Log().WithError(err).WithField("saga_id", instance.ID).Error("SagaStore failed to encode saga")
		return err
	}
	updateSQL := `UPDATE ` + tblName + ` SET saga_data=` + store.ParamsMarkers[0] + `, version=` + store.ParamsMarkers[1] + `
WHERE saga_id=` + store.ParamsMarkers[2] + ` AND version=` + store.ParamsMarkers[3] + ``
	result, err := tx.Exec(updateSQL, buf, nextVersion, instance.ID, currentVersion)
	if err != nil {
		return err
	} else if rowsAffected, ee := result.RowsAffected(); ee != nil || rowsAffected == 0 {
		return fmt.Errorf("saga with saga_id:%v had stale data when updating. :%v", instance.ID, ee)
	}
	return nil
}
func (store *SagaStore) RegisterSagaType(saga gbus.Saga) {
	gob.Register(saga)
}
func (store *SagaStore) DeleteSaga(tx *sql.Tx, instance *saga.Instance) error {
	tblName := GetSagatableName(store.SvcName)
	deleteSQL := `DELETE FROM ` + tblName + ` WHERE saga_id= ?`
	result, err := tx.Exec(deleteSQL, instance.ID)
	if err != nil {
		return err
	}
	rowsEffected, e := result.RowsAffected()
	if rowsEffected == 0 || e != nil {
		return errors.New("couldn't delete saga, saga not found orr an error occurred")
	}
	return nil
}
func (store *SagaStore) GetSagaByID(tx *sql.Tx, sagaID string) (*saga.Instance, error) {
	tblName := GetSagatableName(store.SvcName)
	selectSQL := `SELECT saga_id, saga_type, saga_data, started_by_request_of_svc, started_by_msg_id, started_by_rpcid, started_by_request_of_saga, version, created_at FROM ` + tblName + ` WHERE saga_id=` + store.ParamsMarkers[0] + ``
	rows, err := tx.Query(selectSQL, sagaID)
	defer func() {
		err := rows.Close()
		if err != nil {
			store.Log().WithError(err).Error("could not close rows")
		}
	}()
	if err != nil && err == sql.ErrNoRows {
		return nil, saga.ErrInstanceNotFound
	}
	if err != nil {
		store.Log().WithError(err).
			WithFields(log.Fields{"saga_id": sagaID, "table_name": GetSagatableName(store.SvcName)}).
			Error("Failed to fetch saga")
		return nil, err
	}
	instances, err := store.scanInstances(rows)
	if err != nil {
		return nil, err
	}
	if len(instances) == 0 {
		return nil, fmt.Errorf("no saga found for saga with saga_id:%v", sagaID)
	}
	return instances[0], nil
}
func (store *SagaStore) SaveNewSaga(tx *sql.Tx, sagaType reflect.Type, newInstance *saga.Instance) (err error) {
	store.RegisterSagaType(newInstance.UnderlyingInstance)
	tblName := GetSagatableName(store.SvcName)
	insertSQL := `INSERT INTO ` + tblName + ` (saga_id, saga_type, saga_data, started_by_request_of_svc, started_by_msg_id, started_by_rpcid, started_by_request_of_saga, version, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`
	var buf []byte
	if buf, err = store.serilizeSaga(newInstance); err != nil {
		store.Log().WithError(err).WithField("saga_id", newInstance.ID).Error("failed to encode saga with sagaID")
		return err
	}
	_, err = tx.Exec(insertSQL, newInstance.ID, sagaType.String(), buf, newInstance.StartedBy, newInstance.StartedByMessageID, newInstance.StartedByRPCID, newInstance.StartedBySaga, newInstance.ConcurrencyCtrl, newInstance.CreatedAt)
	if err != nil {
		store.Log().WithError(err).Error("failed saving new saga")
		return err
	}
	return nil
}
func (store *SagaStore) Purge() error {
	tx := store.NewTx()
	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")
		return err
	}
	if txErr := tx.Commit(); txErr != nil {
		return txErr
	}
	rowsEffected, resultsErr := results.RowsAffected()
	if resultsErr != nil {
		store.Log().WithError(err).Warn("failed to fetch number of deleted saga records")
	} else {
		store.Log().WithField("deleted_instances", rowsEffected).Info("purged saga store")
	}
	return nil
}
func (store *SagaStore) serilizeSaga(instance *saga.Instance) ([]byte, error) {
	var buf bytes.Buffer
	enc := gob.NewEncoder(&buf)
	error := enc.Encode(instance)
	return buf.Bytes(), error
}