feature: add saga persistence layer

go.mod

@@ -33,6 +33,7 @@ require (
 require (
 	github.com/agiledragon/gomonkey v2.0.2+incompatible
 	github.com/agiledragon/gomonkey/v2 v2.9.0
+	github.com/mattn/go-sqlite3 v1.14.19
 )
 
 require (

go.sum

@@ -515,6 +515,8 @@ github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27k
 github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
 github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
+github.com/mattn/go-sqlite3 v1.14.19 h1:fhGleo2h1p8tVChob4I9HpmVFIAkKGpiukdrgQbWfGI=
+github.com/mattn/go-sqlite3 v1.14.19/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
 github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=

pkg/saga/statemachine/constant/contant.go

@@ -13,6 +13,7 @@ const (
 	VarNameIsAsyncExecution    string = "_is_async_execution_"
 	VarNameStateInst           string = "_current_state_instance_"
 	SeqEntityStateMachineInst  string = "STATE_MACHINE_INST"
+	SeqEntityStateInst         string = "STATE_INST"
 	VarNameBusinesskey         string = "_business_key_"
 	VarNameParentId            string = "_parent_id_"
 	StateTypeServiceTask       string = "ServiceTask"

pkg/saga/statemachine/engine/process_ctrl/process_context.go

@@ -47,6 +47,12 @@ type ProcessContextImpl struct {
 	instruction Instruction
 }
 
+func NewProcessContextImpl() *ProcessContextImpl {
+	return &ProcessContextImpl{
+		mp: make(map[string]interface{}),
+	}
+}
+
 func (p *ProcessContextImpl) GetVariable(name string) interface{} {
 	p.mu.RLock()
 	defer p.mu.RUnlock()

pkg/saga/statemachine/engine/serializer/serializer.go

@@ -0,0 +1,47 @@
+package serializer
+
+import (
+	"bytes"
+	"encoding/gob"
+	"encoding/json"
+	"github.com/pkg/errors"
+)
+
+type Serializer[S any, T any] interface {
+	Serialize(object S) (T, error)
+	Deserialize(object T) (S, error)
+}
+
+type ParamsSerializer struct{}
+
+func (ParamsSerializer) Serialize(object any) (string, error) {
+	result, err := json.Marshal(object)
+	return string(result), err
+}
+
+func (ParamsSerializer) Deserialize(object string) (any, error) {
+	var result any
+	err := json.Unmarshal([]byte(object), &result)
+	return result, err
+}
+
+type ErrorSerializer struct{}
+
+func (ErrorSerializer) Serialize(object error) ([]byte, error) {
+	var buffer bytes.Buffer
+	encoder := gob.NewEncoder(&buffer)
+	if object != nil {
+		err := encoder.Encode(object.Error())
+		return buffer.Bytes(), err
+	}
+	return nil, nil
+}
+
+func (ErrorSerializer) Deserialize(object []byte) (error, error) {
+	var errorMsg string
+	buffer := bytes.NewReader(object)
+	encoder := gob.NewDecoder(buffer)
+	err := encoder.Decode(&errorMsg)
+
+	return errors.New(errorMsg), err
+}

pkg/saga/statemachine/engine/serializer/serializer_test.go

@@ -0,0 +1,17 @@
+package serializer
+
+import (
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestErrorSerializer(t *testing.T) {
+	serializer := ErrorSerializer{}
+	expected := errors.New("This is a test error")
+	serialized, err := serializer.Serialize(expected)
+	assert.Nil(t, err)
+	actual, err := serializer.Deserialize(serialized)
+	assert.Nil(t, err)
+	assert.Equal(t, expected.Error(), actual.Error())
+}

pkg/saga/statemachine/engine/store/db/db.go

@@ -0,0 +1,102 @@
+package db
+
+import (
+	"database/sql"
+	"github.com/pkg/errors"
+	"github.com/seata/seata-go/pkg/util/log"
+)
+
+const TimeLayout = "2006-01-02 15:04:05.999999999-07:00"
+
+type ExecStatement[T any] func(obj T, stmt *sql.Stmt) (int64, error)
+
+type ScanRows[T any] func(rows *sql.Rows) (T, error)
+
+type Store struct {
+	db *sql.DB
+}
+
+func SelectOne[T any](db *sql.DB, sql string, fn ScanRows[T], args ...any) (T, error) {
+	var result T
+	log.Debugf("Preparing SQL: %s", sql)
+	stmt, err := db.Prepare(sql)
+	defer stmt.Close()
+	if err != nil {
+		return result, err
+	}
+
+	log.Debugf("setting params to Stmt: %v", args)
+	rows, err := stmt.Query(args...)
+	defer rows.Close()
+	if err != nil {
+		return result, nil
+	}
+
+	if rows.Next() {
+		return fn(rows)
+	}
+	return result, errors.New("no target selected")
+}
+
+func SelectList[T any](db *sql.DB, sql string, fn ScanRows[T], args ...any) ([]T, error) {
+	result := make([]T, 0)
+
+	log.Debugf("Preparing SQL: %s", sql)
+	stmt, err := db.Prepare(sql)
+	defer stmt.Close()
+	if err != nil {
+		return result, err
+	}
+
+	log.Debugf("setting params to Stmt: %v", args)
+	rows, err := stmt.Query(args...)
+	defer rows.Close()
+	if err != nil {
+		return result, err
+	}
+
+	for rows.Next() {
+		obj, err := fn(rows)
+		if err != nil {
+			return result, err
+		}
+		result = append(result, obj)
+	}
+
+	return result, nil
+}
+
+func ExecuteUpdate[T any](db *sql.DB, sql string, fn ExecStatement[T], obj T) (int64, error) {
+	log.Debugf("Preparing SQL: %s", sql)
+	stmt, err := db.Prepare(sql)
+	defer stmt.Close()
+	if err != nil {
+		return 0, err
+	}
+
+	log.Debugf("setting params to Stmt: %v", obj)
+
+	rowsAffected, err := fn(obj, stmt)
+	if err != nil {
+		return rowsAffected, err
+	}
+
+	return rowsAffected, nil
+}
+
+func ExecuteUpdateArgs(db *sql.DB, sql string, args ...any) (int64, error) {
+	log.Debugf("Preparing SQL: %s", sql)
+	stmt, err := db.Prepare(sql)
+	defer stmt.Close()
+	if err != nil {
+		return 0, err
+	}
+
+	log.Debugf("setting params to Stmt: %v", args)
+
+	result, err := stmt.Exec(args...)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}

pkg/saga/statemachine/engine/store/db/db_test.go

@@ -0,0 +1,28 @@
+package db
+
+import (
+	"database/sql"
+	"os"
+	"sync"
+)
+
+var (
+	oncePrepareDB sync.Once
+	db            *sql.DB
+)
+
+func prepareDB() {
+	oncePrepareDB.Do(func() {
+		var err error
+		db, err = sql.Open("sqlite3", ":memory:")
+		query_, err := os.ReadFile("testdata/sql/saga/sqlite_init.sql")
+		initScript := string(query_)
+		if err != nil {
+			panic(err)
+		}
+		if _, err := db.Exec(initScript); err != nil {
+			panic(err)
+		}
+	})
+
+}

pkg/saga/statemachine/engine/store/db/statelang.go

@@ -0,0 +1,114 @@
+package db
+
+import (
+	"database/sql"
+	"github.com/pkg/errors"
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
+	"regexp"
+	"time"
+)
+
+const (
+	StateMachineFields                   = "id, tenant_id, app_name, name, status, gmt_create, ver, type, content, recover_strategy, comment_"
+	GetStateMachineByIdSql               = "SELECT " + StateMachineFields + " FROM ${TABLE_PREFIX}state_machine_def WHERE id = ?"
+	QueryStateMachinesByNameAndTenantSql = "SELECT " + StateMachineFields + " FROM ${TABLE_PREFIX}state_machine_def WHERE name = ? AND tenant_id = ? ORDER BY gmt_create DESC"
+	InsertStateMachineSql                = "INSERT INTO ${TABLE_PREFIX}state_machine_def (" + StateMachineFields + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
+	TablePrefix                          = "\\$\\{TABLE_PREFIX}"
+)
+
+type StateLangStore struct {
+	Store
+	tablePrefix                          string
+	getStateMachineByIdSql               string
+	queryStateMachinesByNameAndTenantSql string
+	insertStateMachineSql                string
+}
+
+func NewStateLangStore(db *sql.DB, tablePrefix string) *StateLangStore {
+	r := regexp.MustCompile(TablePrefix)
+
+	stateLangStore := &StateLangStore{
+		Store:                                Store{db},
+		tablePrefix:                          tablePrefix,
+		getStateMachineByIdSql:               r.ReplaceAllString(GetStateMachineByIdSql, tablePrefix),
+		queryStateMachinesByNameAndTenantSql: r.ReplaceAllString(QueryStateMachinesByNameAndTenantSql, tablePrefix),
+		insertStateMachineSql:                r.ReplaceAllString(InsertStateMachineSql, tablePrefix),
+	}
+
+	return stateLangStore
+}
+
+func (s *StateLangStore) GetStateMachineById(stateMachineId string) (statelang.StateMachine, error) {
+	return SelectOne(s.db, s.getStateMachineByIdSql, scanRowsToStateMachine, stateMachineId)
+}
+
+func (s *StateLangStore) GetLastVersionStateMachine(stateMachineName string, tenantId string) (statelang.StateMachine, error) {
+	stateMachineList, err := SelectList(s.db, s.queryStateMachinesByNameAndTenantSql, scanRowsToStateMachine, stateMachineName, tenantId)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(stateMachineList) > 0 {
+		return stateMachineList[0], nil
+	}
+	return nil, nil
+}
+
+func (s *StateLangStore) StoreStateMachine(stateMachine statelang.StateMachine) error {
+	rows, err := ExecuteUpdate(s.db, s.insertStateMachineSql, execStateMachineStatement, stateMachine)
+	if err != nil {
+		return err
+	}
+	if rows <= 0 {
+		return errors.New("affected rows is smaller than 0")
+	}
+
+	return nil
+}
+
+func scanRowsToStateMachine(rows *sql.Rows) (statelang.StateMachine, error) {
+	stateMachine := statelang.NewStateMachineImpl()
+	//var id, name, comment, version, appName, content, t, recoverStrategy, tenantId, status string
+	var id, tenantId, appName, name, status, created, version, t, content, recoverStrategy, comment string
+	//var created int64
+	err := rows.Scan(&id, &tenantId, &appName, &name, &status, &created, &version, &t, &content, &recoverStrategy, &comment)
+	if err != nil {
+		return stateMachine, err
+	}
+	stateMachine.SetID(id)
+	stateMachine.SetName(name)
+	stateMachine.SetComment(comment)
+	stateMachine.SetVersion(version)
+	stateMachine.SetAppName(appName)
+	stateMachine.SetContent(content)
+	createdTime, _ := time.Parse(TimeLayout, created)
+	stateMachine.SetCreateTime(createdTime)
+	stateMachine.SetType(t)
+	if recoverStrategy != "" {
+		stateMachine.SetRecoverStrategy(statelang.RecoverStrategy(recoverStrategy))
+	}
+	stateMachine.SetTenantId(t)
+	stateMachine.SetStatus(statelang.StateMachineStatus(status))
+	return stateMachine, nil
+}
+
+func execStateMachineStatement(obj statelang.StateMachine, stmt *sql.Stmt) (int64, error) {
+	result, err := stmt.Exec(
+		obj.ID(),
+		obj.TenantId(),
+		obj.AppName(),
+		obj.Name(),
+		obj.Status(),
+		obj.CreateTime(),
+		obj.Version(),
+		obj.Type(),
+		obj.Content(),
+		obj.RecoverStrategy(),
+		obj.Comment(),
+	)
+	if err != nil {
+		return 0, err
+	}
+	rowsAffected, err := result.RowsAffected()
+	return rowsAffected, err
+}