This is an automated email from the ASF dual-hosted git repository.
thunguo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go-samples.git
The following commit(s) were added to refs/heads/main by this push:
new 44ff37f add saga sample (#94)
44ff37f is described below
commit 44ff37fb6fcba0b520158dc2ee5991b5507112ae
Author: lxfeng1997 <[email protected]>
AuthorDate: Tue May 12 20:05:25 2026 +0800
add saga sample (#94)
* feat: Migrating a legacy long-running process to Saga
* Fix insurance claim Saga sample runtime
* fix: use canceled spelling in saga sample
---
saga/insurance_claim/README.md | 150 ++++++++++
saga/insurance_claim/README_zh.md | 150 ++++++++++
saga/insurance_claim/config.yaml | 28 ++
saga/insurance_claim/docker-compose.yml | 39 +++
saga/insurance_claim/internal/app/config.go | 107 +++++++
saga/insurance_claim/internal/app/store.go | 310 +++++++++++++++++++++
saga/insurance_claim/internal/httpjson/httpjson.go | 67 +++++
saga/insurance_claim/legacy/main.go | 125 +++++++++
saga/insurance_claim/orchestrator/main.go | 248 +++++++++++++++++
saga/insurance_claim/seatago.yaml | 43 +++
saga/insurance_claim/services/assessment/main.go | 100 +++++++
saga/insurance_claim/services/funds/main.go | 100 +++++++
saga/insurance_claim/services/identity/main.go | 100 +++++++
saga/insurance_claim/services/surveyor/main.go | 100 +++++++
saga/insurance_claim/services/transfer/main.go | 88 ++++++
.../sql/mysql_claim_saga_schema.sql | 135 +++++++++
.../statelang/insurance_claim_saga.json | 179 ++++++++++++
17 files changed, 2069 insertions(+)
diff --git a/saga/insurance_claim/README.md b/saga/insurance_claim/README.md
new file mode 100644
index 0000000..5d2bd15
--- /dev/null
+++ b/saga/insurance_claim/README.md
@@ -0,0 +1,150 @@
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+-->
+
+# Insurance Claim Saga Sample
+
+This sample demonstrates how to migrate a legacy long-running insurance claim
process to Seata Go Saga.
+
+The workflow contains the following steps:
+
+1. Verify claimant identity
+2. Create damage assessment record
+3. Reserve payout funds
+4. Notify the assigned surveyor
+5. Execute bank transfer
+
+The legacy implementation is a sequential flow. If the first four steps
succeed but the bank transfer fails in step five, the system is left with
intermediate state such as verified identity, created assessment, reserved
funds, and sent surveyor notification, with no automatic rollback.
+
+After migrating to Saga, each step is mapped to a forward action and a
compensating action:
+
+| Forward Action | Compensating Action |
+| --- | --- |
+| VerifyIdentity | UnverifyClaim |
+| CreateDamageAssessment | DeleteDamageAssessment |
+| ReservePayoutFunds | ReleasePayoutFunds |
+| NotifyAssignedSurveyor | CancelSurveyorNotification |
+| ExecuteBankTransfer | None |
+
+When `ExecuteBankTransfer` fails, Saga compensates in reverse order:
+
+1. CancelSurveyorNotification
+2. ReleasePayoutFunds
+3. DeleteDamageAssessment
+4. UnverifyClaim
+
+## Directory Layout
+
+- `legacy/`: the legacy sequential implementation used to show the
pre-migration problem
+- `orchestrator/`: the Saga orchestrator starter
+- `services/`: five independent Go HTTP services
+- `statelang/insurance_claim_saga.json`: Saga state machine definition
+- `sql/mysql_claim_saga_schema.sql`: Saga persistence tables and business demo
tables
+- `docker-compose.yml`: MySQL and Seata Server
+
+## Start the Infrastructure
+
+```bash
+cd saga/insurance_claim
+docker-compose up -d
+```
+
+Default ports:
+
+- MySQL: `3306`
+- Seata Server: `8091`
+- identity service: `18081`
+- assessment service: `18082`
+- funds service: `18083`
+- surveyor service: `18084`
+- transfer service: `18085`
+
+## Start the Five Services
+
+Open five terminals from the repository root and run:
+
+```bash
+go run ./saga/insurance_claim/services/identity
+go run ./saga/insurance_claim/services/assessment
+go run ./saga/insurance_claim/services/funds
+go run ./saga/insurance_claim/services/surveyor
+go run ./saga/insurance_claim/services/transfer
+```
+
+You can also run each command from `saga/insurance_claim` by dropping the
`saga/insurance_claim/` prefix, for example `go run ./orchestrator`.
+
+## Run the Legacy Sequential Flow
+
+Success case:
+
+```bash
+go run ./saga/insurance_claim/legacy
+```
+
+Simulate a bank transfer failure:
+
+```bash
+go run ./saga/insurance_claim/legacy -failTransfer
+```
+
+When the flow fails, it stops immediately and prints the current business
snapshot. You will see that the intermediate state from the first four steps is
still present, which is exactly the problem with the legacy implementation.
+
+## Run the Migrated Saga Flow
+
+Success case:
+
+```bash
+go run ./saga/insurance_claim/orchestrator
+```
+
+Simulate a bank transfer failure:
+
+```bash
+go run ./saga/insurance_claim/orchestrator -failTransfer
+```
+
+The failure case prints:
+
+- `xid`
+- final Saga status
+- compensation status
+- business snapshot
+- `actionTrail`
+
+Focus on `actionTrail`. It shows:
+
+1. The first four forward actions execute
+2. The bank transfer fails
+3. The four compensating actions execute in reverse order
+
+## Implementation Notes
+
+- The five services run as independent processes and expose forward and
compensating actions over HTTP
+- The Saga orchestrator uses the built-in seata-go `http` invoker to call
those services
+- The state machine uses `CompensateState` to map each forward action to its
compensating action
+- The `failTransfer` parameter provides a stable way to reproduce the bank
transfer failure
+- `claim_step_log` records both forward and compensation execution order for
easy observation
+- MySQL can be customized with `MYSQL_HOST`, `MYSQL_PORT`, `MYSQL_USERNAME`,
`MYSQL_PASSWORD`, and `MYSQL_DB`; `MYSQL_USER` and `MYSQL_PWD` are also
accepted for compatibility
+
+## Debug with a Local seata-go Checkout
+
+If you want to debug this sample against a local `../incubator-seata-go`
checkout:
+
+```bash
+go mod edit -replace seata.apache.org/seata-go=../incubator-seata-go
+go run ./saga/insurance_claim/orchestrator -failTransfer
+go mod edit -dropreplace seata.apache.org/seata-go
+```
diff --git a/saga/insurance_claim/README_zh.md
b/saga/insurance_claim/README_zh.md
new file mode 100644
index 0000000..7d3d53e
--- /dev/null
+++ b/saga/insurance_claim/README_zh.md
@@ -0,0 +1,150 @@
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+-->
+
+# Insurance Claim Saga Sample
+
+本示例演示一个遗留的长流程理赔处理如何迁移到 Seata Go Saga。
+
+场景步骤如下：
+
+1. 校验理赔人身份
+2. 创建定损记录
+3. 预留赔付资金
+4. 通知定损员
+5. 执行银行打款
+
+遗留实现是一个串行流程。前 4 步成功后，如果第 5 步银行打款失败，系统会留下已校验身份、已创建定损单、已预留资金、已通知定损员等中间状态，无法自动回滚。
+
+Saga 迁移后，每个步骤都会拆成前向动作和补偿动作：
+
+| 前向动作 | 补偿动作 |
+| --- | --- |
+| VerifyIdentity | UnverifyClaim |
+| CreateDamageAssessment | DeleteDamageAssessment |
+| ReservePayoutFunds | ReleasePayoutFunds |
+| NotifyAssignedSurveyor | CancelSurveyorNotification |
+| ExecuteBankTransfer | 无 |
+
+当 `ExecuteBankTransfer` 失败时，Saga 会按逆序执行：
+
+1. CancelSurveyorNotification
+2. ReleasePayoutFunds
+3. DeleteDamageAssessment
+4. UnverifyClaim
+
+## 目录说明
+
+- `legacy/`：遗留串行版本，对照“迁移前”的问题
+- `orchestrator/`：Saga 编排启动器
+- `services/`：五个独立 Go HTTP 服务
+- `statelang/insurance_claim_saga.json`：Saga 状态机定义
+- `sql/mysql_claim_saga_schema.sql`：Saga 持久化表 + 业务表示例
+- `docker-compose.yml`：MySQL 与 Seata Server
+
+## 启动基础设施
+
+```bash
+cd saga/insurance_claim
+docker-compose up -d
+```
+
+默认端口：
+
+- MySQL: `3306`
+- Seata Server: `8091`
+- identity service: `18081`
+- assessment service: `18082`
+- funds service: `18083`
+- surveyor service: `18084`
+- transfer service: `18085`
+
+## 启动五个服务
+
+在仓库根目录分别打开 5 个终端：
+
+```bash
+go run ./saga/insurance_claim/services/identity
+go run ./saga/insurance_claim/services/assessment
+go run ./saga/insurance_claim/services/funds
+go run ./saga/insurance_claim/services/surveyor
+go run ./saga/insurance_claim/services/transfer
+```
+
+也可以在 `saga/insurance_claim` 目录内运行，把命令里的 `saga/insurance_claim/` 前缀去掉即可，例如 `go
run ./orchestrator`。
+
+## 运行遗留串行流程
+
+成功场景：
+
+```bash
+go run ./saga/insurance_claim/legacy
+```
+
+模拟银行打款失败：
+
+```bash
+go run ./saga/insurance_claim/legacy -failTransfer
+```
+
+失败时会直接停止，并打印当前业务快照。你会看到前 4 步留下的中间状态仍然存在，这正是遗留实现的问题。
+
+## 运行 Saga 迁移后的流程
+
+成功场景：
+
+```bash
+go run ./saga/insurance_claim/orchestrator
+```
+
+模拟银行打款失败：
+
+```bash
+go run ./saga/insurance_claim/orchestrator -failTransfer
+```
+
+失败场景下会输出：
+
+- `xid`
+- Saga 最终状态
+- compensation status
+- 业务快照
+- `actionTrail`
+
+重点看 `actionTrail`，可以看到：
+
+1. 先执行 4 个前向动作
+2. 打款失败
+3. 再按逆序执行 4 个补偿动作
+
+## 关键实现说明
+
+- 五个服务都是独立进程，使用 HTTP 暴露前向动作和补偿动作
+- Saga 编排器通过 seata-go 内置 `http` invoker 调用这些服务
+- 状态机使用 `CompensateState` 描述每个前向动作对应的补偿动作
+- `failTransfer` 参数用于稳定复现银行打款失败
+- `claim_step_log` 会记录前向和补偿执行顺序，便于观察迁移效果
+- MySQL 可通过
`MYSQL_HOST`、`MYSQL_PORT`、`MYSQL_USERNAME`、`MYSQL_PASSWORD`、`MYSQL_DB` 覆盖；同时兼容
`MYSQL_USER` 和 `MYSQL_PWD`
+
+## 使用本地 seata-go 调试
+
+如果要配合本地 `../incubator-seata-go` 调试：
+
+```bash
+go mod edit -replace seata.apache.org/seata-go=../incubator-seata-go
+go run ./saga/insurance_claim/orchestrator -failTransfer
+go mod edit -dropreplace seata.apache.org/seata-go
+```
diff --git a/saga/insurance_claim/config.yaml b/saga/insurance_claim/config.yaml
new file mode 100644
index 0000000..38d22a5
--- /dev/null
+++ b/saga/insurance_claim/config.yaml
@@ -0,0 +1,28 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+trans_operation_timeout: 60000
+service_invoke_timeout: 5000
+rm_report_success_enable: false
+saga_branch_register_enable: false
+
+state_machine_resources:
+ - statelang/*.json
+
+store_enabled: true
+store_type: mysql
+store_dsn:
"root:secret@tcp(127.0.0.1:3306)/seata_saga?parseTime=true&multiStatements=true"
+
+tc_enabled: true
diff --git a/saga/insurance_claim/docker-compose.yml
b/saga/insurance_claim/docker-compose.yml
new file mode 100644
index 0000000..f893e85
--- /dev/null
+++ b/saga/insurance_claim/docker-compose.yml
@@ -0,0 +1,39 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+services:
+ mysql:
+ image: mysql:8.0
+ container_name: insurance_claim_mysql
+ restart: unless-stopped
+ environment:
+ MYSQL_ROOT_PASSWORD: secret
+ MYSQL_DATABASE: seata_saga
+ ports:
+ - "3306:3306"
+ volumes:
+ -
./sql/mysql_claim_saga_schema.sql:/docker-entrypoint-initdb.d/1_claim_saga.sql:ro
+
+ seata-server:
+ image: seataio/seata-server:1.6.1
+ container_name: insurance_claim_seata_server
+ restart: unless-stopped
+ environment:
+ - SEATA_IP=0.0.0.0
+ - SEATA_PORT=8091
+ ports:
+ - "8091:8091"
+ depends_on:
+ - mysql
diff --git a/saga/insurance_claim/internal/app/config.go
b/saga/insurance_claim/internal/app/config.go
new file mode 100644
index 0000000..f2f7725
--- /dev/null
+++ b/saga/insurance_claim/internal/app/config.go
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package app
+
+import (
+ "fmt"
+ "os"
+)
+
+const (
+ DefaultMySQLHost = "127.0.0.1"
+ DefaultMySQLPort = "3306"
+ DefaultMySQLUser = "root"
+ DefaultMySQLPassword = "secret"
+ DefaultMySQLDB = "seata_saga"
+
+ DefaultIdentityPort = "18081"
+ DefaultAssessmentPort = "18082"
+ DefaultFundsPort = "18083"
+ DefaultSurveyorPort = "18084"
+ DefaultTransferPort = "18085"
+)
+
+type Settings struct {
+ MySQLHost string
+ MySQLPort string
+ MySQLUser string
+ MySQLPass string
+ MySQLDB string
+
+ IdentityPort string
+ AssessmentPort string
+ FundsPort string
+ SurveyorPort string
+ TransferPort string
+}
+
+func LoadSettings() Settings {
+ return Settings{
+ MySQLHost: envOrDefault("MYSQL_HOST", DefaultMySQLHost),
+ MySQLPort: envOrDefault("MYSQL_PORT", DefaultMySQLPort),
+ MySQLUser: envOrDefaultAny(DefaultMySQLUser,
"MYSQL_USERNAME", "MYSQL_USER"),
+ MySQLPass: envOrDefaultAny(DefaultMySQLPassword,
"MYSQL_PASSWORD", "MYSQL_PWD"),
+ MySQLDB: envOrDefault("MYSQL_DB", DefaultMySQLDB),
+ IdentityPort: envOrDefault("IDENTITY_SERVICE_PORT",
DefaultIdentityPort),
+ AssessmentPort: envOrDefault("ASSESSMENT_SERVICE_PORT",
DefaultAssessmentPort),
+ FundsPort: envOrDefault("FUNDS_SERVICE_PORT",
DefaultFundsPort),
+ SurveyorPort: envOrDefault("SURVEYOR_SERVICE_PORT",
DefaultSurveyorPort),
+ TransferPort: envOrDefault("TRANSFER_SERVICE_PORT",
DefaultTransferPort),
+ }
+}
+
+func (s Settings) MySQLDSN() string {
+ return
fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true&multiStatements=true",
+ s.MySQLUser, s.MySQLPass, s.MySQLHost, s.MySQLPort, s.MySQLDB)
+}
+
+func (s Settings) IdentityBaseURL() string {
+ return fmt.Sprintf("http://127.0.0.1:%s/";, s.IdentityPort)
+}
+
+func (s Settings) AssessmentBaseURL() string {
+ return fmt.Sprintf("http://127.0.0.1:%s/";, s.AssessmentPort)
+}
+
+func (s Settings) FundsBaseURL() string {
+ return fmt.Sprintf("http://127.0.0.1:%s/";, s.FundsPort)
+}
+
+func (s Settings) SurveyorBaseURL() string {
+ return fmt.Sprintf("http://127.0.0.1:%s/";, s.SurveyorPort)
+}
+
+func (s Settings) TransferBaseURL() string {
+ return fmt.Sprintf("http://127.0.0.1:%s/";, s.TransferPort)
+}
+
+func envOrDefault(key string, fallback string) string {
+ if value := os.Getenv(key); value != "" {
+ return value
+ }
+ return fallback
+}
+
+func envOrDefaultAny(fallback string, keys ...string) string {
+ for _, key := range keys {
+ if value := os.Getenv(key); value != "" {
+ return value
+ }
+ }
+ return fallback
+}
diff --git a/saga/insurance_claim/internal/app/store.go
b/saga/insurance_claim/internal/app/store.go
new file mode 100644
index 0000000..aa2ba55
--- /dev/null
+++ b/saga/insurance_claim/internal/app/store.go
@@ -0,0 +1,310 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package app
+
+import (
+ "database/sql"
+ "fmt"
+ "strings"
+
+ _ "github.com/go-sql-driver/mysql"
+)
+
+type Snapshot struct {
+ IdentityVerified bool
+ AssessmentStatus string
+ FundsStatus string
+ FundsAmount int
+ SurveyorStatus string
+ TransferStatus string
+ TransferLastError string
+ OrderedActionTrail []string
+}
+
+func OpenDB() (*sql.DB, error) {
+ settings := LoadSettings()
+ db, err := sql.Open("mysql", settings.MySQLDSN())
+ if err != nil {
+ return nil, err
+ }
+ if err := db.Ping(); err != nil {
+ _ = db.Close()
+ return nil, err
+ }
+ return db, nil
+}
+
+func EnsureBusinessSchema(db *sql.DB) error {
+ statements := []string{
+ `CREATE TABLE IF NOT EXISTS claim_identity (
+ claim_id VARCHAR(64) PRIMARY KEY,
+ business_key VARCHAR(64) NOT NULL,
+ claimant_id VARCHAR(64) NOT NULL,
+ verified TINYINT(1) NOT NULL DEFAULT 0,
+ updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
ON UPDATE CURRENT_TIMESTAMP
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`,
+ `CREATE TABLE IF NOT EXISTS claim_assessment (
+ claim_id VARCHAR(64) PRIMARY KEY,
+ business_key VARCHAR(64) NOT NULL,
+ assessment_id VARCHAR(64) NOT NULL,
+ status VARCHAR(32) NOT NULL,
+ updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
ON UPDATE CURRENT_TIMESTAMP
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`,
+ `CREATE TABLE IF NOT EXISTS claim_fund_reservation (
+ claim_id VARCHAR(64) PRIMARY KEY,
+ business_key VARCHAR(64) NOT NULL,
+ amount INT NOT NULL,
+ status VARCHAR(32) NOT NULL,
+ updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
ON UPDATE CURRENT_TIMESTAMP
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`,
+ `CREATE TABLE IF NOT EXISTS claim_surveyor_notice (
+ claim_id VARCHAR(64) PRIMARY KEY,
+ business_key VARCHAR(64) NOT NULL,
+ surveyor_id VARCHAR(64) NOT NULL,
+ status VARCHAR(32) NOT NULL,
+ updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
ON UPDATE CURRENT_TIMESTAMP
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`,
+ `CREATE TABLE IF NOT EXISTS claim_transfer (
+ claim_id VARCHAR(64) PRIMARY KEY,
+ business_key VARCHAR(64) NOT NULL,
+ bank_account VARCHAR(64) NOT NULL,
+ amount INT NOT NULL,
+ status VARCHAR(32) NOT NULL,
+ last_error VARCHAR(255) NOT NULL DEFAULT '',
+ updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
ON UPDATE CURRENT_TIMESTAMP
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`,
+ `CREATE TABLE IF NOT EXISTS claim_step_log (
+ id BIGINT PRIMARY KEY AUTO_INCREMENT,
+ business_key VARCHAR(64) NOT NULL,
+ claim_id VARCHAR(64) NOT NULL,
+ step_name VARCHAR(64) NOT NULL,
+ action_name VARCHAR(64) NOT NULL,
+ note VARCHAR(255) NOT NULL DEFAULT '',
+ created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`,
+ }
+ for _, statement := range statements {
+ if _, err := db.Exec(statement); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func EnsureSagaStoreSchema(db *sql.DB) error {
+ statements := []string{
+ `ALTER TABLE seata_state_machine_def
+ MODIFY gmt_create DATETIME(6) DEFAULT
CURRENT_TIMESTAMP(6)`,
+ `ALTER TABLE seata_state_machine_inst
+ MODIFY gmt_started DATETIME(6) DEFAULT
CURRENT_TIMESTAMP(6),
+ MODIFY gmt_end DATETIME(6) DEFAULT NULL,
+ MODIFY gmt_updated DATETIME(6) DEFAULT
CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)`,
+ `ALTER TABLE seata_state_inst
+ MODIFY gmt_started DATETIME(6) DEFAULT
CURRENT_TIMESTAMP(6),
+ MODIFY gmt_end DATETIME(6) DEFAULT NULL,
+ MODIFY gmt_updated DATETIME(6) DEFAULT
CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)`,
+ }
+ for _, statement := range statements {
+ if _, err := db.Exec(statement); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func ResetClaimData(db *sql.DB, businessKey string, claimID string) error {
+ statements := []string{
+ "DELETE FROM claim_step_log WHERE business_key = ? OR claim_id
= ?",
+ "DELETE FROM claim_transfer WHERE claim_id = ?",
+ "DELETE FROM claim_surveyor_notice WHERE claim_id = ?",
+ "DELETE FROM claim_fund_reservation WHERE claim_id = ?",
+ "DELETE FROM claim_assessment WHERE claim_id = ?",
+ "DELETE FROM claim_identity WHERE claim_id = ?",
+ }
+ for index, statement := range statements {
+ var err error
+ if index == 0 {
+ _, err = db.Exec(statement, businessKey, claimID)
+ } else {
+ _, err = db.Exec(statement, claimID)
+ }
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func RecordIdentityVerified(db *sql.DB, businessKey string, claimID string,
claimantID string) error {
+ query := `INSERT INTO claim_identity(claim_id, business_key,
claimant_id, verified)
+ VALUES(?, ?, ?, 1)
+ ON DUPLICATE KEY UPDATE business_key = VALUES(business_key),
claimant_id = VALUES(claimant_id), verified = 1`
+ if _, err := db.Exec(query, claimID, businessKey, claimantID); err !=
nil {
+ return err
+ }
+ return appendStepLog(db, businessKey, claimID, "identity", "verify",
"claimant verified")
+}
+
+func UnverifyIdentity(db *sql.DB, businessKey string, claimID string) error {
+ if _, err := db.Exec(`UPDATE claim_identity SET verified = 0 WHERE
claim_id = ?`, claimID); err != nil {
+ return err
+ }
+ return appendStepLog(db, businessKey, claimID, "identity",
"compensate", "claimant verification rolled back")
+}
+
+func CreateAssessment(db *sql.DB, businessKey string, claimID string,
assessmentID string) error {
+ query := `INSERT INTO claim_assessment(claim_id, business_key,
assessment_id, status)
+ VALUES(?, ?, ?, 'CREATED')
+ ON DUPLICATE KEY UPDATE business_key = VALUES(business_key),
assessment_id = VALUES(assessment_id), status = 'CREATED'`
+ if _, err := db.Exec(query, claimID, businessKey, assessmentID); err !=
nil {
+ return err
+ }
+ return appendStepLog(db, businessKey, claimID, "assessment", "create",
"damage assessment created")
+}
+
+func DeleteAssessment(db *sql.DB, businessKey string, claimID string) error {
+ if _, err := db.Exec(`DELETE FROM claim_assessment WHERE claim_id = ?`,
claimID); err != nil {
+ return err
+ }
+ return appendStepLog(db, businessKey, claimID, "assessment",
"compensate", "damage assessment deleted")
+}
+
+func ReserveFunds(db *sql.DB, businessKey string, claimID string, amount int)
error {
+ query := `INSERT INTO claim_fund_reservation(claim_id, business_key,
amount, status)
+ VALUES(?, ?, ?, 'RESERVED')
+ ON DUPLICATE KEY UPDATE business_key = VALUES(business_key),
amount = VALUES(amount), status = 'RESERVED'`
+ if _, err := db.Exec(query, claimID, businessKey, amount); err != nil {
+ return err
+ }
+ return appendStepLog(db, businessKey, claimID, "funds", "reserve",
fmt.Sprintf("reserved payout amount=%d", amount))
+}
+
+func ReleaseFunds(db *sql.DB, businessKey string, claimID string) error {
+ if _, err := db.Exec(`UPDATE claim_fund_reservation SET status =
'RELEASED' WHERE claim_id = ?`, claimID); err != nil {
+ return err
+ }
+ return appendStepLog(db, businessKey, claimID, "funds", "compensate",
"reserved payout released")
+}
+
+func NotifySurveyor(db *sql.DB, businessKey string, claimID string, surveyorID
string) error {
+ query := `INSERT INTO claim_surveyor_notice(claim_id, business_key,
surveyor_id, status)
+ VALUES(?, ?, ?, 'NOTIFIED')
+ ON DUPLICATE KEY UPDATE business_key = VALUES(business_key),
surveyor_id = VALUES(surveyor_id), status = 'NOTIFIED'`
+ if _, err := db.Exec(query, claimID, businessKey, surveyorID); err !=
nil {
+ return err
+ }
+ return appendStepLog(db, businessKey, claimID, "surveyor", "notify",
"assigned surveyor notified")
+}
+
+func CancelSurveyorNotification(db *sql.DB, businessKey string, claimID
string) error {
+ if _, err := db.Exec(`UPDATE claim_surveyor_notice SET status =
'CANCELED' WHERE claim_id = ?`, claimID); err != nil {
+ return err
+ }
+ return appendStepLog(db, businessKey, claimID, "surveyor",
"compensate", "surveyor notification canceled")
+}
+
+func ExecuteTransfer(db *sql.DB, businessKey string, claimID string,
bankAccount string, amount int, failTransfer bool) error {
+ status := "SUCCESS"
+ lastError := ""
+ if failTransfer {
+ status = "FAILED"
+ lastError = "BANK_TRANSFER_FAILED"
+ }
+ query := `INSERT INTO claim_transfer(claim_id, business_key,
bank_account, amount, status, last_error)
+ VALUES(?, ?, ?, ?, ?, ?)
+ ON DUPLICATE KEY UPDATE business_key = VALUES(business_key),
bank_account = VALUES(bank_account), amount = VALUES(amount), status =
VALUES(status), last_error = VALUES(last_error)`
+ if _, err := db.Exec(query, claimID, businessKey, bankAccount, amount,
status, lastError); err != nil {
+ return err
+ }
+ note := fmt.Sprintf("bank transfer amount=%d status=%s", amount, status)
+ if err := appendStepLog(db, businessKey, claimID, "transfer",
"execute", note); err != nil {
+ return err
+ }
+ if failTransfer {
+ return fmt.Errorf("BANK_TRANSFER_FAILED")
+ }
+ return nil
+}
+
+func LoadSnapshot(db *sql.DB, claimID string) (Snapshot, error) {
+ snapshot := Snapshot{
+ AssessmentStatus: "MISSING",
+ FundsStatus: "MISSING",
+ SurveyorStatus: "MISSING",
+ TransferStatus: "MISSING",
+ TransferLastError: "",
+ }
+
+ if err := db.QueryRow(`SELECT verified FROM claim_identity WHERE
claim_id = ?`, claimID).Scan(&snapshot.IdentityVerified); err != nil && err !=
sql.ErrNoRows {
+ return snapshot, err
+ }
+ if err := db.QueryRow(`SELECT status FROM claim_assessment WHERE
claim_id = ?`, claimID).Scan(&snapshot.AssessmentStatus); err != nil && err !=
sql.ErrNoRows {
+ return snapshot, err
+ }
+ if err := db.QueryRow(`SELECT status, amount FROM
claim_fund_reservation WHERE claim_id = ?`,
claimID).Scan(&snapshot.FundsStatus, &snapshot.FundsAmount); err != nil && err
!= sql.ErrNoRows {
+ return snapshot, err
+ }
+ if err := db.QueryRow(`SELECT status FROM claim_surveyor_notice WHERE
claim_id = ?`, claimID).Scan(&snapshot.SurveyorStatus); err != nil && err !=
sql.ErrNoRows {
+ return snapshot, err
+ }
+ if err := db.QueryRow(`SELECT status, last_error FROM claim_transfer
WHERE claim_id = ?`, claimID).Scan(&snapshot.TransferStatus,
&snapshot.TransferLastError); err != nil && err != sql.ErrNoRows {
+ return snapshot, err
+ }
+
+ rows, err := db.Query(`SELECT step_name, action_name, note FROM
claim_step_log WHERE claim_id = ? ORDER BY id`, claimID)
+ if err != nil {
+ return snapshot, err
+ }
+ defer rows.Close()
+
+ for rows.Next() {
+ var stepName string
+ var actionName string
+ var note string
+ if err := rows.Scan(&stepName, &actionName, ¬e); err != nil {
+ return snapshot, err
+ }
+ snapshot.OrderedActionTrail =
append(snapshot.OrderedActionTrail, fmt.Sprintf("%s:%s(%s)", stepName,
actionName, note))
+ }
+ if err := rows.Err(); err != nil {
+ return snapshot, err
+ }
+ return snapshot, nil
+}
+
+func FormatSnapshot(snapshot Snapshot) string {
+ lines := []string{
+ fmt.Sprintf("identity.verified=%t", snapshot.IdentityVerified),
+ fmt.Sprintf("assessment.status=%s", snapshot.AssessmentStatus),
+ fmt.Sprintf("funds.status=%s amount=%d", snapshot.FundsStatus,
snapshot.FundsAmount),
+ fmt.Sprintf("surveyor.status=%s", snapshot.SurveyorStatus),
+ fmt.Sprintf("transfer.status=%s lastError=%s",
snapshot.TransferStatus, snapshot.TransferLastError),
+ }
+ if len(snapshot.OrderedActionTrail) == 0 {
+ lines = append(lines, "actionTrail=<empty>")
+ } else {
+ lines = append(lines,
"actionTrail="+strings.Join(snapshot.OrderedActionTrail, " -> "))
+ }
+ return strings.Join(lines, "\n")
+}
+
+func appendStepLog(db *sql.DB, businessKey string, claimID string, stepName
string, actionName string, note string) error {
+ _, err := db.Exec(`INSERT INTO claim_step_log(business_key, claim_id,
step_name, action_name, note) VALUES(?, ?, ?, ?, ?)`,
+ businessKey, claimID, stepName, actionName, note)
+ return err
+}
diff --git a/saga/insurance_claim/internal/httpjson/httpjson.go
b/saga/insurance_claim/internal/httpjson/httpjson.go
new file mode 100644
index 0000000..01a85c7
--- /dev/null
+++ b/saga/insurance_claim/internal/httpjson/httpjson.go
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package httpjson
+
+import (
+ "encoding/json"
+ "fmt"
+ "net/http"
+)
+
+func ReadArgs(r *http.Request, expected int) ([]json.RawMessage, error) {
+ defer r.Body.Close()
+
+ var args []json.RawMessage
+ if err := json.NewDecoder(r.Body).Decode(&args); err != nil {
+ return nil, err
+ }
+ if len(args) != expected {
+ return nil, fmt.Errorf("expect %d args, got %d", expected,
len(args))
+ }
+ return args, nil
+}
+
+func StringArg(args []json.RawMessage, index int) (string, error) {
+ var value string
+ if err := json.Unmarshal(args[index], &value); err != nil {
+ return "", err
+ }
+ return value, nil
+}
+
+func IntArg(args []json.RawMessage, index int) (int, error) {
+ var value int
+ if err := json.Unmarshal(args[index], &value); err != nil {
+ return 0, err
+ }
+ return value, nil
+}
+
+func BoolArg(args []json.RawMessage, index int) (bool, error) {
+ var value bool
+ if err := json.Unmarshal(args[index], &value); err != nil {
+ return false, err
+ }
+ return value, nil
+}
+
+func WriteText(w http.ResponseWriter, statusCode int, body string) {
+ w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+ w.WriteHeader(statusCode)
+ _, _ = w.Write([]byte(body))
+}
diff --git a/saga/insurance_claim/legacy/main.go
b/saga/insurance_claim/legacy/main.go
new file mode 100644
index 0000000..9599f8f
--- /dev/null
+++ b/saga/insurance_claim/legacy/main.go
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+ "bytes"
+ "encoding/json"
+ "flag"
+ "fmt"
+ "io"
+ "net/http"
+ "os"
+
+ "seata.apache.org/seata-go-samples/saga/insurance_claim/internal/app"
+)
+
+func main() {
+ var (
+ businessKey string
+ claimID string
+ claimantID string
+ assessmentID string
+ surveyorID string
+ bankAccount string
+ payoutAmount int
+ failTransfer bool
+ )
+
+ flag.StringVar(&businessKey, "businessKey",
"insurance-claim-legacy-demo", "business key")
+ flag.StringVar(&claimID, "claimId", "claim-1001", "insurance claim ID")
+ flag.StringVar(&claimantID, "claimantId", "claimant-9001", "claimant
ID")
+ flag.StringVar(&assessmentID, "assessmentId", "assessment-7001",
"damage assessment ID")
+ flag.StringVar(&surveyorID, "surveyorId", "surveyor-3001", "surveyor
ID")
+ flag.StringVar(&bankAccount, "bankAccount", "6222020202020202", "bank
account")
+ flag.IntVar(&payoutAmount, "payoutAmount", 1500, "payout amount")
+ flag.BoolVar(&failTransfer, "failTransfer", false, "simulate a bank
transfer failure")
+ flag.Parse()
+
+ settings := app.LoadSettings()
+ db, err := app.OpenDB()
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "failed to open the database: %v\n", err)
+ os.Exit(1)
+ }
+ defer db.Close()
+
+ if err := app.EnsureBusinessSchema(db); err != nil {
+ fmt.Fprintf(os.Stderr, "failed to initialize the business
schema: %v\n", err)
+ os.Exit(1)
+ }
+ if err := app.ResetClaimData(db, businessKey, claimID); err != nil {
+ fmt.Fprintf(os.Stderr, "failed to reset the sample data: %v\n",
err)
+ os.Exit(1)
+ }
+
+ steps := []struct {
+ name string
+ url string
+ args []any
+ }{
+ {"VerifyIdentity", settings.IdentityBaseURL() +
"VerifyIdentity", []any{businessKey, claimID, claimantID}},
+ {"CreateDamageAssessment", settings.AssessmentBaseURL() +
"CreateDamageAssessment", []any{businessKey, claimID, assessmentID}},
+ {"ReservePayoutFunds", settings.FundsBaseURL() +
"ReservePayoutFunds", []any{businessKey, claimID, payoutAmount}},
+ {"NotifyAssignedSurveyor", settings.SurveyorBaseURL() +
"NotifyAssignedSurveyor", []any{businessKey, claimID, surveyorID}},
+ {"ExecuteBankTransfer", settings.TransferBaseURL() +
"ExecuteBankTransfer", []any{businessKey, claimID, bankAccount, payoutAmount,
failTransfer}},
+ }
+
+ for _, step := range steps {
+ if err := invoke(step.url, step.args); err != nil {
+ fmt.Printf("mode=legacy businessKey=%s failedStep=%s
err=%v\n", businessKey, step.name, err)
+ snapshot, snapErr := app.LoadSnapshot(db, claimID)
+ if snapErr != nil {
+ fmt.Fprintf(os.Stderr, "failed to load the
insurance claim snapshot: %v\n", snapErr)
+ os.Exit(1)
+ }
+ fmt.Println(app.FormatSnapshot(snapshot))
+ os.Exit(1)
+ }
+ }
+
+ snapshot, err := app.LoadSnapshot(db, claimID)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "failed to load the insurance claim
snapshot: %v\n", err)
+ os.Exit(1)
+ }
+
+ fmt.Printf("mode=legacy businessKey=%s status=SUCCESS\n", businessKey)
+ fmt.Println(app.FormatSnapshot(snapshot))
+}
+
+func invoke(url string, args []any) error {
+ body, err := json.Marshal(args)
+ if err != nil {
+ return err
+ }
+ resp, err := http.Post(url, "application/json", bytes.NewReader(body))
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ raw, err := io.ReadAll(resp.Body)
+ if err != nil {
+ return err
+ }
+ if resp.StatusCode >= http.StatusBadRequest {
+ return fmt.Errorf("%s", string(raw))
+ }
+ return nil
+}
diff --git a/saga/insurance_claim/orchestrator/main.go
b/saga/insurance_claim/orchestrator/main.go
new file mode 100644
index 0000000..bf436bb
--- /dev/null
+++ b/saga/insurance_claim/orchestrator/main.go
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "net/http"
+ "os"
+ "path/filepath"
+
+ "gopkg.in/yaml.v3"
+
+ "seata.apache.org/seata-go-samples/saga/insurance_claim/internal/app"
+ "seata.apache.org/seata-go/pkg/client"
+ engcfg "seata.apache.org/seata-go/pkg/saga/statemachine/engine/config"
+ "seata.apache.org/seata-go/pkg/saga/statemachine/engine/core"
+ "seata.apache.org/seata-go/pkg/saga/statemachine/engine/invoker"
+)
+
+func main() {
+ var (
+ seataConf string
+ engineConf string
+ businessKey string
+ claimID string
+ claimantID string
+ assessmentID string
+ surveyorID string
+ bankAccount string
+ payoutAmount int
+ failTransfer bool
+ )
+
+ flag.StringVar(&seataConf, "seataConf", "seatago.yaml", "path to the
seata-go client config")
+ flag.StringVar(&engineConf, "engineConf", "config.yaml", "path to the
Saga engine config")
+ flag.StringVar(&businessKey, "businessKey",
"insurance-claim-saga-demo", "business key")
+ flag.StringVar(&claimID, "claimId", "claim-1001", "insurance claim ID")
+ flag.StringVar(&claimantID, "claimantId", "claimant-9001", "claimant
ID")
+ flag.StringVar(&assessmentID, "assessmentId", "assessment-7001",
"damage assessment ID")
+ flag.StringVar(&surveyorID, "surveyorId", "surveyor-3001", "surveyor
ID")
+ flag.StringVar(&bankAccount, "bankAccount", "6222020202020202", "bank
account")
+ flag.IntVar(&payoutAmount, "payoutAmount", 1500, "payout amount")
+ flag.BoolVar(&failTransfer, "failTransfer", false, "simulate a bank
transfer failure")
+ flag.Parse()
+
+ seataConf, err := resolveSamplePath(seataConf)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "failed to locate the seata-go client
config: %v\n", err)
+ os.Exit(1)
+ }
+ engineConf, err = resolveSamplePath(engineConf)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "failed to locate the Saga engine
config: %v\n", err)
+ os.Exit(1)
+ }
+
+ client.InitPath(seataConf)
+
+ engine, err := newStateMachineEngine()
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "failed to create the state machine
engine: %v\n", err)
+ os.Exit(1)
+ }
+
+ cfgIface := engine.GetStateMachineConfig()
+ cfg, ok := cfgIface.(*engcfg.DefaultStateMachineConfig)
+ if !ok {
+ fmt.Fprintf(os.Stderr, "unexpected state machine config type:
%T\n", cfgIface)
+ os.Exit(1)
+ }
+
+ settings := app.LoadSettings()
+ runtimeEngineConf, cleanup, err :=
prepareRuntimeEngineConfig(engineConf, settings)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "failed to prepare the runtime engine
config: %v\n", err)
+ os.Exit(1)
+ }
+ defer cleanup()
+
+ if err := cfg.LoadConfig(runtimeEngineConf); err != nil {
+ fmt.Fprintf(os.Stderr, "failed to load the engine config:
%v\n", err)
+ os.Exit(1)
+ }
+ if err := cfg.Init(); err != nil {
+ fmt.Fprintf(os.Stderr, "failed to initialize the engine config:
%v\n", err)
+ os.Exit(1)
+ }
+
+ registerHTTPClients(cfgIface, settings)
+
+ db, err := app.OpenDB()
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "failed to open the database: %v\n", err)
+ os.Exit(1)
+ }
+ defer db.Close()
+
+ if err := app.EnsureBusinessSchema(db); err != nil {
+ fmt.Fprintf(os.Stderr, "failed to initialize the business
schema: %v\n", err)
+ os.Exit(1)
+ }
+ if err := app.EnsureSagaStoreSchema(db); err != nil {
+ fmt.Fprintf(os.Stderr, "failed to ensure the Saga store schema:
%v\n", err)
+ os.Exit(1)
+ }
+ if err := app.ResetClaimData(db, businessKey, claimID); err != nil {
+ fmt.Fprintf(os.Stderr, "failed to reset the sample data: %v\n",
err)
+ os.Exit(1)
+ }
+
+ params := map[string]any{
+ "businessKey": businessKey,
+ "claimId": claimID,
+ "claimantId": claimantID,
+ "assessmentId": assessmentID,
+ "surveyorId": surveyorID,
+ "bankAccount": bankAccount,
+ "payoutAmount": payoutAmount,
+ "failTransfer": failTransfer,
+ }
+
+ instance, err := engine.StartWithBusinessKey(context.Background(),
"InsuranceClaimSaga", "", businessKey, params)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "failed to start the Saga: %v\n", err)
+ os.Exit(1)
+ }
+
+ snapshot, err := app.LoadSnapshot(db, claimID)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "failed to load the insurance claim
snapshot: %v\n", err)
+ os.Exit(1)
+ }
+
+ fmt.Printf("mode=saga businessKey=%s xid=%s status=%s
compensationStatus=%s\n",
+ businessKey, instance.ID(), instance.Status(),
instance.CompensationStatus())
+ fmt.Println(app.FormatSnapshot(snapshot))
+}
+
+func registerHTTPClients(cfgIface any, settings app.Settings) {
+ cfg := cfgIface.(*engcfg.DefaultStateMachineConfig)
+ httpInvoker, ok :=
cfg.ServiceInvokerManager().ServiceInvoker("http").(*invoker.HTTPInvoker)
+ if !ok {
+ panic("http invoker is not initialized")
+ }
+
+ clientConfig := &http.Client{}
+ httpInvoker.RegisterClient("identityService",
invoker.NewHTTPClient("identityService", settings.IdentityBaseURL(),
clientConfig))
+ httpInvoker.RegisterClient("assessmentService",
invoker.NewHTTPClient("assessmentService", settings.AssessmentBaseURL(),
clientConfig))
+ httpInvoker.RegisterClient("fundsService",
invoker.NewHTTPClient("fundsService", settings.FundsBaseURL(), clientConfig))
+ httpInvoker.RegisterClient("surveyorService",
invoker.NewHTTPClient("surveyorService", settings.SurveyorBaseURL(),
clientConfig))
+ httpInvoker.RegisterClient("transferService",
invoker.NewHTTPClient("transferService", settings.TransferBaseURL(),
clientConfig))
+}
+
+func newStateMachineEngine() (*core.ProcessCtrlStateMachineEngine, error) {
+ wd, err := os.Getwd()
+ if err != nil {
+ return nil, err
+ }
+ if err := os.Chdir(os.TempDir()); err != nil {
+ return nil, err
+ }
+ defer func() {
+ _ = os.Chdir(wd)
+ }()
+ return core.NewProcessCtrlStateMachineEngine()
+}
+
+func resolveSamplePath(path string) (string, error) {
+ if filepath.IsAbs(path) {
+ return path, fileMustExist(path)
+ }
+
+ candidates := []string{
+ path,
+ filepath.Join("saga", "insurance_claim", path),
+ }
+ for _, candidate := range candidates {
+ if err := fileMustExist(candidate); err == nil {
+ return candidate, nil
+ }
+ }
+ return "", fmt.Errorf("the file does not exist: %s", path)
+}
+
+func fileMustExist(path string) error {
+ info, err := os.Stat(path)
+ if err != nil {
+ return err
+ }
+ if info.IsDir() {
+ return fmt.Errorf("the path refers to a directory: %s", path)
+ }
+ return nil
+}
+
+func prepareRuntimeEngineConfig(engineConf string, settings app.Settings)
(string, func(), error) {
+ raw, err := os.ReadFile(engineConf)
+ if err != nil {
+ return "", nil, err
+ }
+
+ var cfg map[string]any
+ if err := yaml.Unmarshal(raw, &cfg); err != nil {
+ return "", nil, err
+ }
+ cfg["store_dsn"] = settings.MySQLDSN()
+ cfg["state_machine_resources"] =
[]string{filepath.Join(filepath.Dir(engineConf), "statelang", "*.json")}
+
+ file, err := os.CreateTemp("", "insurance-claim-saga-*.yaml")
+ if err != nil {
+ return "", nil, err
+ }
+ defer file.Close()
+
+ encoder := yaml.NewEncoder(file)
+ encoder.SetIndent(2)
+ if err := encoder.Encode(cfg); err != nil {
+ _ = os.Remove(file.Name())
+ _ = encoder.Close()
+ return "", nil, err
+ }
+ if err := encoder.Close(); err != nil {
+ _ = os.Remove(file.Name())
+ return "", nil, err
+ }
+
+ cleanup := func() {
+ _ = os.Remove(file.Name())
+ }
+ return file.Name(), cleanup, nil
+}
diff --git a/saga/insurance_claim/seatago.yaml
b/saga/insurance_claim/seatago.yaml
new file mode 100644
index 0000000..5afdfe2
--- /dev/null
+++ b/saga/insurance_claim/seatago.yaml
@@ -0,0 +1,43 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+seata:
+ enabled: true
+ application-id: "insurance-claim-saga-app"
+ tx-service-group: "default_tx_group"
+
+ client:
+ rm:
+ saga-branch-register-enable: true
+ tm: {}
+
+ service:
+ vgroup-mapping:
+ default_tx_group: "default"
+ grouplist:
+ default: "127.0.0.1:8091"
+
+ transport:
+ type: TCP
+ server: NIO
+ heartbeat: true
+
+ getty:
+ reconnect-interval: 1
+ connection-num: 1
+ session:
+ compress-encoding: false
+ tcp-no-delay: true
+ keep-alive-period: 180
diff --git a/saga/insurance_claim/services/assessment/main.go
b/saga/insurance_claim/services/assessment/main.go
new file mode 100644
index 0000000..9998833
--- /dev/null
+++ b/saga/insurance_claim/services/assessment/main.go
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+ "fmt"
+ "log"
+ "net/http"
+
+ "seata.apache.org/seata-go-samples/saga/insurance_claim/internal/app"
+
"seata.apache.org/seata-go-samples/saga/insurance_claim/internal/httpjson"
+)
+
+func main() {
+ settings := app.LoadSettings()
+ db, err := app.OpenDB()
+ if err != nil {
+ log.Fatalf("failed to open the database: %v", err)
+ }
+ defer db.Close()
+
+ if err := app.EnsureBusinessSchema(db); err != nil {
+ log.Fatalf("failed to initialize the business schema: %v", err)
+ }
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
+ httpjson.WriteText(w, http.StatusOK, "OK")
+ })
+ mux.HandleFunc("/CreateDamageAssessment", func(w http.ResponseWriter, r
*http.Request) {
+ args, err := httpjson.ReadArgs(r, 3)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ businessKey, err := httpjson.StringArg(args, 0)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimID, err := httpjson.StringArg(args, 1)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ assessmentID, err := httpjson.StringArg(args, 2)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ if err := app.CreateAssessment(db, businessKey, claimID,
assessmentID); err != nil {
+ httpjson.WriteText(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
+ log.Printf("operation=CreateDamageAssessment businessKey=%s
claimId=%s assessmentId=%s status=SUCCESS", businessKey, claimID, assessmentID)
+ httpjson.WriteText(w, http.StatusOK, "ASSESSMENT_CREATED")
+ })
+ mux.HandleFunc("/DeleteDamageAssessment", func(w http.ResponseWriter, r
*http.Request) {
+ args, err := httpjson.ReadArgs(r, 2)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ businessKey, err := httpjson.StringArg(args, 0)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimID, err := httpjson.StringArg(args, 1)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ if err := app.DeleteAssessment(db, businessKey, claimID); err
!= nil {
+ httpjson.WriteText(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
+ log.Printf("operation=DeleteDamageAssessment businessKey=%s
claimId=%s status=SUCCESS", businessKey, claimID)
+ httpjson.WriteText(w, http.StatusOK, "ASSESSMENT_DELETED")
+ })
+
+ addr := fmt.Sprintf(":%s", settings.AssessmentPort)
+ log.Printf("assessment service listening on %s\n", addr)
+ log.Fatal(http.ListenAndServe(addr, mux))
+}
diff --git a/saga/insurance_claim/services/funds/main.go
b/saga/insurance_claim/services/funds/main.go
new file mode 100644
index 0000000..2ce938c
--- /dev/null
+++ b/saga/insurance_claim/services/funds/main.go
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+ "fmt"
+ "log"
+ "net/http"
+
+ "seata.apache.org/seata-go-samples/saga/insurance_claim/internal/app"
+
"seata.apache.org/seata-go-samples/saga/insurance_claim/internal/httpjson"
+)
+
+func main() {
+ settings := app.LoadSettings()
+ db, err := app.OpenDB()
+ if err != nil {
+ log.Fatalf("failed to open the database: %v", err)
+ }
+ defer db.Close()
+
+ if err := app.EnsureBusinessSchema(db); err != nil {
+ log.Fatalf("failed to initialize the business schema: %v", err)
+ }
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
+ httpjson.WriteText(w, http.StatusOK, "OK")
+ })
+ mux.HandleFunc("/ReservePayoutFunds", func(w http.ResponseWriter, r
*http.Request) {
+ args, err := httpjson.ReadArgs(r, 3)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ businessKey, err := httpjson.StringArg(args, 0)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimID, err := httpjson.StringArg(args, 1)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ amount, err := httpjson.IntArg(args, 2)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ if err := app.ReserveFunds(db, businessKey, claimID, amount);
err != nil {
+ httpjson.WriteText(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
+ log.Printf("operation=ReservePayoutFunds businessKey=%s
claimId=%s amount=%d status=SUCCESS", businessKey, claimID, amount)
+ httpjson.WriteText(w, http.StatusOK, "FUNDS_RESERVED")
+ })
+ mux.HandleFunc("/ReleasePayoutFunds", func(w http.ResponseWriter, r
*http.Request) {
+ args, err := httpjson.ReadArgs(r, 2)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ businessKey, err := httpjson.StringArg(args, 0)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimID, err := httpjson.StringArg(args, 1)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ if err := app.ReleaseFunds(db, businessKey, claimID); err !=
nil {
+ httpjson.WriteText(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
+ log.Printf("operation=ReleasePayoutFunds businessKey=%s
claimId=%s status=SUCCESS", businessKey, claimID)
+ httpjson.WriteText(w, http.StatusOK, "FUNDS_RELEASED")
+ })
+
+ addr := fmt.Sprintf(":%s", settings.FundsPort)
+ log.Printf("funds service listening on %s\n", addr)
+ log.Fatal(http.ListenAndServe(addr, mux))
+}
diff --git a/saga/insurance_claim/services/identity/main.go
b/saga/insurance_claim/services/identity/main.go
new file mode 100644
index 0000000..99afadb
--- /dev/null
+++ b/saga/insurance_claim/services/identity/main.go
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+ "fmt"
+ "log"
+ "net/http"
+
+ "seata.apache.org/seata-go-samples/saga/insurance_claim/internal/app"
+
"seata.apache.org/seata-go-samples/saga/insurance_claim/internal/httpjson"
+)
+
+func main() {
+ settings := app.LoadSettings()
+ db, err := app.OpenDB()
+ if err != nil {
+ log.Fatalf("failed to open the database: %v", err)
+ }
+ defer db.Close()
+
+ if err := app.EnsureBusinessSchema(db); err != nil {
+ log.Fatalf("failed to initialize the business schema: %v", err)
+ }
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
+ httpjson.WriteText(w, http.StatusOK, "OK")
+ })
+ mux.HandleFunc("/VerifyIdentity", func(w http.ResponseWriter, r
*http.Request) {
+ args, err := httpjson.ReadArgs(r, 3)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ businessKey, err := httpjson.StringArg(args, 0)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimID, err := httpjson.StringArg(args, 1)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimantID, err := httpjson.StringArg(args, 2)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ if err := app.RecordIdentityVerified(db, businessKey, claimID,
claimantID); err != nil {
+ httpjson.WriteText(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
+ log.Printf("operation=VerifyIdentity businessKey=%s claimId=%s
claimantId=%s status=SUCCESS", businessKey, claimID, claimantID)
+ httpjson.WriteText(w, http.StatusOK, "IDENTITY_VERIFIED")
+ })
+ mux.HandleFunc("/UnverifyClaim", func(w http.ResponseWriter, r
*http.Request) {
+ args, err := httpjson.ReadArgs(r, 2)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ businessKey, err := httpjson.StringArg(args, 0)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimID, err := httpjson.StringArg(args, 1)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ if err := app.UnverifyIdentity(db, businessKey, claimID); err
!= nil {
+ httpjson.WriteText(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
+ log.Printf("operation=UnverifyClaim businessKey=%s claimId=%s
status=SUCCESS", businessKey, claimID)
+ httpjson.WriteText(w, http.StatusOK, "IDENTITY_UNVERIFIED")
+ })
+
+ addr := fmt.Sprintf(":%s", settings.IdentityPort)
+ log.Printf("identity service listening on %s\n", addr)
+ log.Fatal(http.ListenAndServe(addr, mux))
+}
diff --git a/saga/insurance_claim/services/surveyor/main.go
b/saga/insurance_claim/services/surveyor/main.go
new file mode 100644
index 0000000..e8d2d40
--- /dev/null
+++ b/saga/insurance_claim/services/surveyor/main.go
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+ "fmt"
+ "log"
+ "net/http"
+
+ "seata.apache.org/seata-go-samples/saga/insurance_claim/internal/app"
+
"seata.apache.org/seata-go-samples/saga/insurance_claim/internal/httpjson"
+)
+
+func main() {
+ settings := app.LoadSettings()
+ db, err := app.OpenDB()
+ if err != nil {
+ log.Fatalf("failed to open the database: %v", err)
+ }
+ defer db.Close()
+
+ if err := app.EnsureBusinessSchema(db); err != nil {
+ log.Fatalf("failed to initialize the business schema: %v", err)
+ }
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
+ httpjson.WriteText(w, http.StatusOK, "OK")
+ })
+ mux.HandleFunc("/NotifyAssignedSurveyor", func(w http.ResponseWriter, r
*http.Request) {
+ args, err := httpjson.ReadArgs(r, 3)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ businessKey, err := httpjson.StringArg(args, 0)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimID, err := httpjson.StringArg(args, 1)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ surveyorID, err := httpjson.StringArg(args, 2)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ if err := app.NotifySurveyor(db, businessKey, claimID,
surveyorID); err != nil {
+ httpjson.WriteText(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
+ log.Printf("operation=NotifyAssignedSurveyor businessKey=%s
claimId=%s surveyorId=%s status=SUCCESS", businessKey, claimID, surveyorID)
+ httpjson.WriteText(w, http.StatusOK, "SURVEYOR_NOTIFIED")
+ })
+ mux.HandleFunc("/CancelSurveyorNotification", func(w
http.ResponseWriter, r *http.Request) {
+ args, err := httpjson.ReadArgs(r, 2)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ businessKey, err := httpjson.StringArg(args, 0)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimID, err := httpjson.StringArg(args, 1)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ if err := app.CancelSurveyorNotification(db, businessKey,
claimID); err != nil {
+ httpjson.WriteText(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
+ log.Printf("operation=CancelSurveyorNotification businessKey=%s
claimId=%s status=SUCCESS", businessKey, claimID)
+ httpjson.WriteText(w, http.StatusOK,
"SURVEYOR_NOTIFICATION_CANCELED")
+ })
+
+ addr := fmt.Sprintf(":%s", settings.SurveyorPort)
+ log.Printf("surveyor service listening on %s\n", addr)
+ log.Fatal(http.ListenAndServe(addr, mux))
+}
diff --git a/saga/insurance_claim/services/transfer/main.go
b/saga/insurance_claim/services/transfer/main.go
new file mode 100644
index 0000000..705ce65
--- /dev/null
+++ b/saga/insurance_claim/services/transfer/main.go
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+ "fmt"
+ "log"
+ "net/http"
+
+ "seata.apache.org/seata-go-samples/saga/insurance_claim/internal/app"
+
"seata.apache.org/seata-go-samples/saga/insurance_claim/internal/httpjson"
+)
+
+func main() {
+ settings := app.LoadSettings()
+ db, err := app.OpenDB()
+ if err != nil {
+ log.Fatalf("failed to open the database: %v", err)
+ }
+ defer db.Close()
+
+ if err := app.EnsureBusinessSchema(db); err != nil {
+ log.Fatalf("failed to initialize the business schema: %v", err)
+ }
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
+ httpjson.WriteText(w, http.StatusOK, "OK")
+ })
+ mux.HandleFunc("/ExecuteBankTransfer", func(w http.ResponseWriter, r
*http.Request) {
+ args, err := httpjson.ReadArgs(r, 5)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ businessKey, err := httpjson.StringArg(args, 0)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ claimID, err := httpjson.StringArg(args, 1)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ bankAccount, err := httpjson.StringArg(args, 2)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ amount, err := httpjson.IntArg(args, 3)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ failTransfer, err := httpjson.BoolArg(args, 4)
+ if err != nil {
+ httpjson.WriteText(w, http.StatusBadRequest,
err.Error())
+ return
+ }
+ if err := app.ExecuteTransfer(db, businessKey, claimID,
bankAccount, amount, failTransfer); err != nil {
+ log.Printf("operation=ExecuteBankTransfer
businessKey=%s claimId=%s bankAccount=%s amount=%d status=FAILED error=%s",
businessKey, claimID, bankAccount, amount, err)
+ httpjson.WriteText(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
+ log.Printf("operation=ExecuteBankTransfer businessKey=%s
claimId=%s bankAccount=%s amount=%d status=SUCCESS", businessKey, claimID,
bankAccount, amount)
+ httpjson.WriteText(w, http.StatusOK, "BANK_TRANSFER_SUCCESS")
+ })
+
+ addr := fmt.Sprintf(":%s", settings.TransferPort)
+ log.Printf("transfer service listening on %s\n", addr)
+ log.Fatal(http.ListenAndServe(addr, mux))
+}
diff --git a/saga/insurance_claim/sql/mysql_claim_saga_schema.sql
b/saga/insurance_claim/sql/mysql_claim_saga_schema.sql
new file mode 100644
index 0000000..3d9e2e2
--- /dev/null
+++ b/saga/insurance_claim/sql/mysql_claim_saga_schema.sql
@@ -0,0 +1,135 @@
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements. See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+CREATE TABLE IF NOT EXISTS `seata_state_machine_def` (
+ `id` varchar(128) NOT NULL,
+ `tenant_id` varchar(32) DEFAULT NULL,
+ `app_name` varchar(64) DEFAULT NULL,
+ `name` varchar(128) NOT NULL,
+ `status` varchar(16) DEFAULT NULL,
+ `gmt_create` datetime(6) DEFAULT CURRENT_TIMESTAMP(6),
+ `ver` varchar(16) DEFAULT NULL,
+ `type` varchar(32) DEFAULT NULL,
+ `content` mediumtext,
+ `recover_strategy` varchar(32) DEFAULT NULL,
+ `comment_` varchar(255) DEFAULT NULL,
+ PRIMARY KEY (`id`),
+ KEY `idx_smdef_name_tenant` (`name`,`tenant_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE IF NOT EXISTS `seata_state_machine_inst` (
+ `id` varchar(128) NOT NULL,
+ `machine_id` varchar(128) NOT NULL,
+ `tenant_id` varchar(32) DEFAULT NULL,
+ `parent_id` varchar(256) DEFAULT NULL,
+ `gmt_started` datetime(6) DEFAULT CURRENT_TIMESTAMP(6),
+ `gmt_end` datetime(6) DEFAULT NULL,
+ `status` varchar(16) DEFAULT NULL,
+ `compensation_status` varchar(16) DEFAULT NULL,
+ `is_running` tinyint(1) DEFAULT 0,
+ `gmt_updated` datetime(6) DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE
CURRENT_TIMESTAMP(6),
+ `business_key` varchar(128) DEFAULT NULL,
+ `start_params` mediumtext,
+ `end_params` mediumtext,
+ `excep` blob,
+ PRIMARY KEY (`id`),
+ KEY `idx_sminst_machine` (`machine_id`),
+ KEY `idx_sminst_parent` (`parent_id`),
+ KEY `idx_sminst_bizkey_tenant` (`business_key`,`tenant_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE IF NOT EXISTS `seata_state_inst` (
+ `id` varchar(128) NOT NULL,
+ `machine_inst_id` varchar(128) NOT NULL,
+ `name` varchar(128) NOT NULL,
+ `type` varchar(32) NOT NULL,
+ `gmt_started` datetime(6) DEFAULT CURRENT_TIMESTAMP(6),
+ `service_name` varchar(255) DEFAULT NULL,
+ `service_method` varchar(255) DEFAULT NULL,
+ `service_type` varchar(32) DEFAULT NULL,
+ `is_for_update` tinyint(1) DEFAULT 0,
+ `input_params` mediumtext,
+ `status` varchar(16) DEFAULT NULL,
+ `business_key` varchar(128) DEFAULT NULL,
+ `state_id_compensated_for` varchar(128) DEFAULT NULL,
+ `state_id_retried_for` varchar(128) DEFAULT NULL,
+ `output_params` mediumtext,
+ `excep` blob,
+ `gmt_end` datetime(6) DEFAULT NULL,
+ `gmt_updated` datetime(6) DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE
CURRENT_TIMESTAMP(6),
+ PRIMARY KEY (`id`),
+ KEY `idx_stinst_machine` (`machine_inst_id`),
+ KEY `idx_stinst_name` (`name`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE IF NOT EXISTS `claim_identity` (
+ `claim_id` varchar(64) NOT NULL,
+ `business_key` varchar(64) NOT NULL,
+ `claimant_id` varchar(64) NOT NULL,
+ `verified` tinyint(1) NOT NULL DEFAULT 0,
+ `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP,
+ PRIMARY KEY (`claim_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE IF NOT EXISTS `claim_assessment` (
+ `claim_id` varchar(64) NOT NULL,
+ `business_key` varchar(64) NOT NULL,
+ `assessment_id` varchar(64) NOT NULL,
+ `status` varchar(32) NOT NULL,
+ `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP,
+ PRIMARY KEY (`claim_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE IF NOT EXISTS `claim_fund_reservation` (
+ `claim_id` varchar(64) NOT NULL,
+ `business_key` varchar(64) NOT NULL,
+ `amount` int NOT NULL,
+ `status` varchar(32) NOT NULL,
+ `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP,
+ PRIMARY KEY (`claim_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE IF NOT EXISTS `claim_surveyor_notice` (
+ `claim_id` varchar(64) NOT NULL,
+ `business_key` varchar(64) NOT NULL,
+ `surveyor_id` varchar(64) NOT NULL,
+ `status` varchar(32) NOT NULL,
+ `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP,
+ PRIMARY KEY (`claim_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE IF NOT EXISTS `claim_transfer` (
+ `claim_id` varchar(64) NOT NULL,
+ `business_key` varchar(64) NOT NULL,
+ `bank_account` varchar(64) NOT NULL,
+ `amount` int NOT NULL,
+ `status` varchar(32) NOT NULL,
+ `last_error` varchar(255) NOT NULL DEFAULT '',
+ `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP,
+ PRIMARY KEY (`claim_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE IF NOT EXISTS `claim_step_log` (
+ `id` bigint NOT NULL AUTO_INCREMENT,
+ `business_key` varchar(64) NOT NULL,
+ `claim_id` varchar(64) NOT NULL,
+ `step_name` varchar(64) NOT NULL,
+ `action_name` varchar(64) NOT NULL,
+ `note` varchar(255) NOT NULL DEFAULT '',
+ `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ PRIMARY KEY (`id`),
+ KEY `idx_claim_step_log_claim` (`claim_id`),
+ KEY `idx_claim_step_log_biz` (`business_key`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git a/saga/insurance_claim/statelang/insurance_claim_saga.json
b/saga/insurance_claim/statelang/insurance_claim_saga.json
new file mode 100644
index 0000000..42981f0
--- /dev/null
+++ b/saga/insurance_claim/statelang/insurance_claim_saga.json
@@ -0,0 +1,179 @@
+{
+ "Name": "InsuranceClaimSaga",
+ "Comment": "Migrate a legacy long-running insurance claim process to Saga
and demonstrate reverse compensation after a bank transfer failure",
+ "StartState": "VerifyIdentity",
+ "Version": "1.1",
+ "Persist": true,
+ "States": {
+ "VerifyIdentity": {
+ "Type": "ServiceTask",
+ "ServiceType": "http",
+ "ServiceName": "identityService",
+ "ServiceMethod": "POST",
+ "CompensateState": "UnverifyClaim",
+ "IsPersist": true,
+ "Input": [
+ "$CEL.elContext['context']['businessKey']",
+ "$CEL.elContext['context']['claimId']",
+ "$CEL.elContext['context']['claimantId']"
+ ],
+ "Catch": [
+ {
+ "Exceptions": [
+ "HTTP error"
+ ],
+ "Next": "CompensationTrigger"
+ }
+ ],
+ "Next": "CreateDamageAssessment"
+ },
+ "CreateDamageAssessment": {
+ "Type": "ServiceTask",
+ "ServiceType": "http",
+ "ServiceName": "assessmentService",
+ "ServiceMethod": "POST",
+ "CompensateState": "DeleteDamageAssessment",
+ "IsPersist": true,
+ "Input": [
+ "$CEL.elContext['context']['businessKey']",
+ "$CEL.elContext['context']['claimId']",
+ "$CEL.elContext['context']['assessmentId']"
+ ],
+ "Catch": [
+ {
+ "Exceptions": [
+ "HTTP error"
+ ],
+ "Next": "CompensationTrigger"
+ }
+ ],
+ "Next": "ReservePayoutFunds"
+ },
+ "ReservePayoutFunds": {
+ "Type": "ServiceTask",
+ "ServiceType": "http",
+ "ServiceName": "fundsService",
+ "ServiceMethod": "POST",
+ "CompensateState": "ReleasePayoutFunds",
+ "IsPersist": true,
+ "Input": [
+ "$CEL.elContext['context']['businessKey']",
+ "$CEL.elContext['context']['claimId']",
+ "$CEL.elContext['context']['payoutAmount']"
+ ],
+ "Catch": [
+ {
+ "Exceptions": [
+ "HTTP error"
+ ],
+ "Next": "CompensationTrigger"
+ }
+ ],
+ "Next": "NotifyAssignedSurveyor"
+ },
+ "NotifyAssignedSurveyor": {
+ "Type": "ServiceTask",
+ "ServiceType": "http",
+ "ServiceName": "surveyorService",
+ "ServiceMethod": "POST",
+ "CompensateState": "CancelSurveyorNotification",
+ "IsPersist": true,
+ "Input": [
+ "$CEL.elContext['context']['businessKey']",
+ "$CEL.elContext['context']['claimId']",
+ "$CEL.elContext['context']['surveyorId']"
+ ],
+ "Catch": [
+ {
+ "Exceptions": [
+ "HTTP error"
+ ],
+ "Next": "CompensationTrigger"
+ }
+ ],
+ "Next": "ExecuteBankTransfer"
+ },
+ "ExecuteBankTransfer": {
+ "Type": "ServiceTask",
+ "ServiceType": "http",
+ "ServiceName": "transferService",
+ "ServiceMethod": "POST",
+ "IsPersist": true,
+ "Input": [
+ "$CEL.elContext['context']['businessKey']",
+ "$CEL.elContext['context']['claimId']",
+ "$CEL.elContext['context']['bankAccount']",
+ "$CEL.elContext['context']['payoutAmount']",
+ "$CEL.elContext['context']['failTransfer']"
+ ],
+ "Catch": [
+ {
+ "Exceptions": [
+ "HTTP error"
+ ],
+ "Next": "CancelSurveyorNotification"
+ }
+ ],
+ "Next": "Success"
+ },
+ "CompensationTrigger": {
+ "Type": "CompensationTrigger",
+ "Next": "Failed"
+ },
+ "UnverifyClaim": {
+ "Type": "ServiceTask",
+ "ServiceType": "http",
+ "ServiceName": "identityService",
+ "ServiceMethod": "POST",
+ "IsPersist": true,
+ "Input": [
+ "$CEL.elContext['context']['businessKey']",
+ "$CEL.elContext['context']['claimId']"
+ ],
+ "Next": "Failed"
+ },
+ "DeleteDamageAssessment": {
+ "Type": "ServiceTask",
+ "ServiceType": "http",
+ "ServiceName": "assessmentService",
+ "ServiceMethod": "POST",
+ "IsPersist": true,
+ "Input": [
+ "$CEL.elContext['context']['businessKey']",
+ "$CEL.elContext['context']['claimId']"
+ ],
+ "Next": "UnverifyClaim"
+ },
+ "ReleasePayoutFunds": {
+ "Type": "ServiceTask",
+ "ServiceType": "http",
+ "ServiceName": "fundsService",
+ "ServiceMethod": "POST",
+ "IsPersist": true,
+ "Input": [
+ "$CEL.elContext['context']['businessKey']",
+ "$CEL.elContext['context']['claimId']"
+ ],
+ "Next": "DeleteDamageAssessment"
+ },
+ "CancelSurveyorNotification": {
+ "Type": "ServiceTask",
+ "ServiceType": "http",
+ "ServiceName": "surveyorService",
+ "ServiceMethod": "POST",
+ "IsPersist": true,
+ "Input": [
+ "$CEL.elContext['context']['businessKey']",
+ "$CEL.elContext['context']['claimId']"
+ ],
+ "Next": "ReleasePayoutFunds"
+ },
+ "Success": {
+ "Type": "Succeed"
+ },
+ "Failed": {
+ "Type": "Fail",
+ "Comment": "The forward flow failed and compensation completed"
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
