Hello Wallet

สร้าง Top-up Workflow ที่รันได้จริงตั้งแต่ต้นจนจบ พร้อมทดลองฆ่า worker กลางคันเพื่อดูว่า durable execution ทำงานยังไง

ติดตั้ง Temporal CLI#

# macOS
brew install temporal

บน Linux ใช้ brew install temporal หรือ snap install temporal ส่วน Windows ให้ดาวน์โหลด tarball จาก temporal.download แล้วเพิ่ม binary เข้า PATH

ตรวจสอบ แล้วเปิด dev server:

temporal --version
temporal server start-dev --db-filename ./temporal.db

Web UI จะอยู่ที่ http://localhost:8233 และ gRPC endpoint ที่ localhost:7233

โครงสร้าง project#

wallet/
├── go.mod
├── workflows/
│   └── topup.go          # workflow เท่านั้น — ห้ามมี I/O
├── activities/
│   └── activities.go     # activity เป็น struct method
├── worker/
│   └── main.go           # ประกอบ dependency + รัน worker
└── starter/
    └── main.go           # client code สำหรับสั่ง start

go mod init wallet
go get go.temporal.io/sdk

กำหนด type ที่ใช้ร่วมกัน#

package workflows

// TopUpRequest คือ input ของ workflow
// ทุก field ต้อง exported เพราะถูก serialize เป็น JSON
type TopUpRequest struct {
	TransactionID string // ใช้เป็น idempotency key ตลอดทั้ง flow
	WalletID      string
	BankAccountID string
	AmountSatang  int64 // เก็บเป็นหน่วยย่อยที่สุดเสมอ ห้ามใช้ float
}

type TopUpResult struct {
	TransactionID string
	BankRef       string
	NewBalance    int64
}

เขียน Workflow#

package workflows

import (
	"time"

	"go.temporal.io/sdk/temporal"
	"go.temporal.io/sdk/workflow"

	"wallet/activities"
)

func TopUpWorkflow(ctx workflow.Context, req TopUpRequest) (TopUpResult, error) {
	logger := workflow.GetLogger(ctx)
	logger.Info("top-up started", "txnID", req.TransactionID, "amount", req.AmountSatang)

	ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout: 30 * time.Second,
		RetryPolicy: &temporal.RetryPolicy{
			InitialInterval:    time.Second,
			BackoffCoefficient: 2.0,
			MaximumInterval:    time.Minute,
			// ไม่กำหนด MaximumAttempts = retry ไม่จำกัด
			// เหมาะกับกรณีธนาคารล่มยาว เราอยากให้ระบบรอจนกว่าจะกลับมา
		},
	})

	var a *activities.Activities // ใช้เป็น type reference เท่านั้น ไม่ต้อง instantiate

	// 1) ตรวจ limit และ KYC tier
	if err := workflow.ExecuteActivity(ctx, a.CheckLimit, req).Get(ctx, nil); err != nil {
		return TopUpResult{}, err
	}

	// 2) หักเงินจากบัญชีธนาคาร
	var bankRef string
	if err := workflow.ExecuteActivity(ctx, a.ChargeBank, req).Get(ctx, &bankRef); err != nil {
		return TopUpResult{}, err
	}

	// 3) เครดิตเข้า wallet ledger
	var newBalance int64
	err := workflow.ExecuteActivity(ctx, a.CreditWallet, req, bankRef).Get(ctx, &newBalance)
	if err != nil {
		// ตรงนี้อันตราย: เงินออกจากธนาคารแล้วแต่เข้า wallet ไม่ได้
		// บทที่ 6 จะสอนวิธีจัดการด้วย Saga
		return TopUpResult{}, err
	}

	// 4) แจ้งเตือน — ล้มได้ ไม่ควรทำให้ธุรกรรมล้มตาม
	notifyCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout: 10 * time.Second,
		RetryPolicy:         &temporal.RetryPolicy{MaximumAttempts: 3},
	})
	if err := workflow.ExecuteActivity(notifyCtx, a.Notify, req).Get(notifyCtx, nil); err != nil {
		logger.Warn("notification failed, continuing anyway", "error", err)
	}

	return TopUpResult{
		TransactionID: req.TransactionID,
		BankRef:       bankRef,
		NewBalance:    newBalance,
	}, nil
}

3 จุดที่ควรสังเกต:

• ใช้ workflow.GetLogger(ctx) ไม่ใช่ fmt.Println — logger ตัวนี้จะกด log ซ้ำระหว่าง replay ให้อัตโนมัติ
• var a *activities.Activities เป็นการอ้าง method โดยไม่สร้าง instance SDK ใช้แค่ชื่อ method ในการหา activity ตัวจริงที่ worker ลงทะเบียนไว้
• Notify มี ActivityOptions ของตัวเอง — งานที่ล้มได้ควรมี policy ต่างจากงานที่เกี่ยวกับเงิน

เขียน Activity#

package activities

import (
	"context"
	"database/sql"

	"go.temporal.io/sdk/activity"
	"go.temporal.io/sdk/temporal"

	"wallet/workflows"
)

// Activities ถือ dependency ทั้งหมด — inject ตอนสร้าง worker
type Activities struct {
	DB   *sql.DB
	Bank BankClient
	Push PushClient
}

func (a *Activities) CheckLimit(ctx context.Context, req workflows.TopUpRequest) error {
	tier, dailyUsed, err := a.loadLimitState(ctx, req.WalletID)
	if err != nil {
		return err // error ธรรมดา = retryable
	}
	if dailyUsed+req.AmountSatang > tier.DailyLimitSatang {
		// เกิน limit คือกฎธุรกิจ ไม่ใช่ปัญหาชั่วคราว → retry ไปก็ไม่ช่วย
		return temporal.NewNonRetryableApplicationError(
			"daily top-up limit exceeded", "LimitExceeded", nil,
		)
	}
	return nil
}

func (a *Activities) ChargeBank(ctx context.Context, req workflows.TopUpRequest) (string, error) {
	logger := activity.GetLogger(ctx)
	info := activity.GetInfo(ctx)
	logger.Info("calling bank", "attempt", info.Attempt)

	// ส่ง idempotency key เดิมทุกครั้ง เพื่อให้ retry ไม่หักเงินซ้ำ
	resp, err := a.Bank.Debit(ctx, BankDebitRequest{
		IdempotencyKey: req.TransactionID,
		AccountID:      req.BankAccountID,
		AmountSatang:   req.AmountSatang,
	})
	if err != nil {
		return "", err
	}
	return resp.Reference, nil
}

func (a *Activities) CreditWallet(
	ctx context.Context, req workflows.TopUpRequest, bankRef string,
) (int64, error) {
	// เขียน ledger แบบ idempotent: UNIQUE constraint บน transaction_id
	return a.creditLedger(ctx, req.WalletID, req.AmountSatang, req.TransactionID, bankRef)
}

func (a *Activities) Notify(ctx context.Context, req workflows.TopUpRequest) error {
	return a.Push.Send(ctx, req.WalletID, "เติมเงินสำเร็จ")
}

Worker#

package main

import (
	"log"

	"go.temporal.io/sdk/client"
	"go.temporal.io/sdk/worker"

	"wallet/activities"
	"wallet/workflows"
)

const TaskQueueWalletCore = "wallet-core"

func main() {
	c, err := client.Dial(client.Options{
		HostPort:  client.DefaultHostPort, // localhost:7233
		Namespace: "default",
	})
	if err != nil {
		log.Fatalln("cannot connect to Temporal:", err)
	}
	defer c.Close()

	w := worker.New(c, TaskQueueWalletCore, worker.Options{})

	w.RegisterWorkflow(workflows.TopUpWorkflow)
	w.RegisterActivity(&activities.Activities{
		DB:   mustOpenDB(),
		Bank: newBankClient(),
		Push: newPushClient(),
	})

	// Run block จนกว่าจะได้ SIGINT/SIGTERM แล้ว shutdown แบบ graceful
	if err := w.Run(worker.InterruptCh()); err != nil {
		log.Fatalln("worker stopped:", err)
	}
}

Starter#

package main

import (
	"context"
	"fmt"
	"log"

	enumspb "go.temporal.io/api/enums/v1"
	"go.temporal.io/sdk/client"

	"wallet/workflows"
)

func main() {
	c, err := client.Dial(client.Options{})
	if err != nil {
		log.Fatalln(err)
	}
	defer c.Close()

	req := workflows.TopUpRequest{
		TransactionID: "TXN-20260807-0001",
		WalletID:      "WLT-1234",
		BankAccountID: "BNK-5678",
		AmountSatang:  100000, // 1,000.00 บาท
	}

	we, err := c.ExecuteWorkflow(context.Background(), client.StartWorkflowOptions{
		// ใช้ business ID เป็น Workflow ID — นี่คือกลไกกันซ้ำชั้นแรก
		ID:        "topup-" + req.TransactionID,
		TaskQueue: "wallet-core",
		// ถ้ามี workflow ID นี้รันอยู่แล้ว ให้ผูกกับตัวเดิม ไม่สร้างใหม่
		WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
	}, workflows.TopUpWorkflow, req)
	if err != nil {
		log.Fatalln(err)
	}

	fmt.Println("started:", we.GetID(), we.GetRunID())

	var result workflows.TopUpResult
	if err := we.Get(context.Background(), &result); err != nil {
		log.Fatalln("workflow failed:", err)
	}
	fmt.Printf("done: bankRef=%s balance=%d\n", result.BankRef, result.NewBalance)
}

รันจริง#

เปิด 3 terminal:

# 1 — dev server
temporal server start-dev --db-filename ./temporal.db

# 2 — worker
go run ./worker

# 3 — start ธุรกรรม
go run ./starter

หรือจะสั่งจาก CLI โดยไม่ต้องมี starter ก็ได้:

temporal workflow execute \
  --type TopUpWorkflow \
  --task-queue wallet-core \
  --workflow-id topup-TXN-20260807-0002 \
  --input '{"TransactionID":"TXN-20260807-0002","WalletID":"WLT-1234","BankAccountID":"BNK-5678","AmountSatang":100000}'

execute จะ block จนกว่าจะจบและคืน exit code ไม่เป็น 0 ถ้า workflow ล้ม ส่วน start จะคืนทันทีแบบ asynchronous

การทดลองที่ต้องลองด้วยตัวเอง#

คำสั่ง CLI ที่ใช้บ่อย#