orbit/examples/task-outbox/internal/outbox/store.go

// Package outbox 演示接入方本地业务事务、稳定提交身份及有限发送恢复。
package outbox

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	_ "embed"
	"encoding/json"
	"errors"
	"regexp"
	"strings"
	"time"
	"unicode"
	"unicode/utf8"

	orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
	"github.com/go-sql-driver/mysql"
)

//go:embed schema.sql
var schema string

var (
	ErrInvalid           = errors.New("outbox scope, operation or request is invalid")
	ErrConflict          = errors.New("outbox operation already has different content; keep the original identity")
	ErrStorage           = errors.New("outbox storage failed; reconcile the original operation")
	ErrCommitUnconfirmed = errors.New("outbox commit is unconfirmed; reconcile the original operation")
	ErrIdle              = errors.New("no due outbox record")
	ErrNotFound          = errors.New("outbox operation not found in this scope")
	ErrLeaseLost         = errors.New("outbox lease was replaced; do not overwrite its result")
	idPattern            = regexp.MustCompile(`^[a-f0-9]{32}$`)
)

const maxSends = 100
const recordColumns = `namespace_id,queue_id,operation_id,idempotency_key,request_json,request_digest,state,send_count,lease_epoch,wake_at,send_deadline_at,last_code,task_id,receipt_json,created_at,updated_at`

type Store struct {
	db               *sql.DB
	namespace, queue string
}

// Open 仅连接业务方数据库，不创建表，不输出原始DSN或数据库错误。
func Open(ctx context.Context, dsn string) (*sql.DB, error) {
	cfg, err := mysql.ParseDSN(dsn)
	if err != nil || cfg.DBName == "" {
		return nil, ErrInvalid
	}
	cfg.ParseTime = true
	cfg.Loc = time.UTC
	cfg.MultiStatements = false
	cfg.AllowAllFiles = false
	cfg.Timeout, cfg.ReadTimeout, cfg.WriteTimeout = 5*time.Second, 8*time.Second, 8*time.Second
	if cfg.Params == nil {
		cfg.Params = map[string]string{}
	}
	cfg.Params["time_zone"] = "'+00:00'"
	connector, err := mysql.NewConnector(cfg)
	if err != nil {
		return nil, ErrStorage
	}
	db := sql.OpenDB(connector)
	db.SetMaxOpenConns(8)
	db.SetMaxIdleConns(4)
	db.SetConnMaxLifetime(5 * time.Minute)
	if err = db.PingContext(ctx); err != nil {
		db.Close()
		return nil, ErrStorage
	}
	return db, nil
}

// Init 仅用于专用空库，已有表会报错；生产接入使用业务方自己的迁移。
func Init(ctx context.Context, db *sql.DB) error {
	for _, statement := range strings.Split(schema, ";") {
		if strings.TrimSpace(statement) == "" {
			continue
		}
		if _, err := db.ExecContext(ctx, statement); err != nil {
			return ErrStorage
		}
	}
	return nil
}
func New(db *sql.DB, namespace, queue string) (*Store, error) {
	if db == nil || !idPattern.MatchString(namespace) || !idPattern.MatchString(queue) {
		return nil, ErrInvalid
	}
	return &Store{db: db, namespace: namespace, queue: queue}, nil
}

// document 冻结当前v1默认值及UTC表示；Outbox按文档字节去重，不冒充平台语义摘要。
func (s *Store) document(in orbit.SubmitTaskRequest) ([]byte, error) {
	// encoding/json会把非法UTF-8替换成U+FFFD；在冻结摘要前拒绝，不能静默改变业务标识。
	for _, value := range []string{in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, in.BusinessKey, in.ConcurrencyKey} {
		if !utf8.ValidString(value) {
			return nil, ErrInvalid
		}
	}
	if in.QueueID != s.queue || in.Handler == "" || in.HandlerVersion == "" || in.PayloadVersion == "" || len(in.Payload) > 256<<10 || !utf8.Valid(in.Payload) || !json.Valid(in.Payload) || bytes.Equal(bytes.TrimSpace(in.Payload), []byte("null")) {
		return nil, ErrInvalid
	}
	if in.TaskTimeoutMS == 0 {
		in.TaskTimeoutMS = 86400000
	}
	if in.RecoverySafety == "" {
		in.RecoverySafety = orbit.RecoveryReconcileOnly
	}
	if in.Retry.MaxAttempts == 0 {
		in.Retry.MaxAttempts = 1
	}
	if in.Retry.AttemptTimeoutMS == 0 {
		in.Retry.AttemptTimeoutMS = 30000
	}
	if in.Retry.MaxAttempts > 1 {
		if in.Retry.InitialDelayMS == 0 {
			in.Retry.InitialDelayMS = 1000
		}
		if in.Retry.MaxDelayMS == 0 {
			in.Retry.MaxDelayMS = 30000
		}
	}
	if in.AvailableAt != nil {
		at := in.AvailableAt.UTC()
		in.AvailableAt = &at
	}
	if in.StartDeadlineAt != nil {
		at := in.StartDeadlineAt.UTC()
		in.StartDeadlineAt = &at
	}
	raw, err := json.Marshal(in)
	if err != nil || len(raw) > 512<<10 {
		return nil, ErrInvalid
	}
	return raw, nil
}

// Enqueue 原子创建业务记录及Outbox。duplicate只表示该本地事务已经提交，不表示平台受理。
// 调用方在第一次调用前持久化operationID；COMMIT回执未知时仍用原ID、原内容再次调用。
func (s *Store) Enqueue(parent context.Context, operationID, sourceRef string, in orbit.SubmitTaskRequest) (duplicate bool, err error) {
	if !idPattern.MatchString(operationID) || !utf8.ValidString(sourceRef) || strings.TrimSpace(sourceRef) == "" || utf8.RuneCountInString(sourceRef) > 512 || strings.ContainsFunc(sourceRef, unicode.IsControl) {
		return false, ErrInvalid
	}
	raw, err := s.document(in)
	if err != nil {
		return false, err
	}
	digest := sha256.Sum256(raw)
	ctx, stop := context.WithTimeout(parent, 5*time.Second)
	defer stop()
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return false, ErrStorage
	}
	defer tx.Rollback()
	_, err = tx.ExecContext(ctx, `INSERT INTO sender_outbox(namespace_id,queue_id,operation_id,idempotency_key,request_json,request_digest,state,send_count,lease_epoch,wake_at,send_deadline_at,last_code,created_at,updated_at) VALUES(?,?,?,?,?,?,'PENDING',0,0,UTC_TIMESTAMP(6),TIMESTAMPADD(HOUR,24,UTC_TIMESTAMP(6)),'',UTC_TIMESTAMP(6),UTC_TIMESTAMP(6))`, s.namespace, s.queue, operationID, "outbox-task-"+operationID, raw, digest[:])
	if err != nil {
		var mysqlErr *mysql.MySQLError
		if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1062 {
			return false, ErrStorage
		}
		if tx.Rollback() != nil {
			return false, ErrStorage
		}
		var saved []byte
		var queue, source string
		err = s.db.QueryRowContext(ctx, `SELECT o.request_digest,o.queue_id,j.source_ref FROM sender_outbox o JOIN sender_jobs j ON j.namespace_id=o.namespace_id AND j.operation_id=o.operation_id WHERE o.namespace_id=? AND o.operation_id=?`, s.namespace, operationID).Scan(&saved, &queue, &source)
		if err != nil {
			return false, ErrStorage
		}
		if queue != s.queue || source != sourceRef || !bytes.Equal(saved, digest[:]) {
			return false, ErrConflict
		}
		return true, nil
	}
	// 锁序统一为Outbox→业务行，不能在业务行持锁时反向调用发送确认。
	if _, err = tx.ExecContext(ctx, `INSERT INTO sender_jobs(namespace_id,operation_id,source_ref,created_at,updated_at) VALUES(?,?,?,UTC_TIMESTAMP(6),UTC_TIMESTAMP(6))`, s.namespace, operationID, sourceRef); err != nil {
		return false, ErrStorage
	}
	if tx.Commit() != nil {
		return false, ErrCommitUnconfirmed
	}
	return false, nil
}

type scanner interface{ Scan(...any) error }

func scanRecord(row scanner, r *Record, extra ...any) error {
	args := []any{&r.NamespaceID, &r.QueueID, &r.OperationID, &r.IdempotencyKey, &r.RequestJSON, &r.RequestDigest, &r.State, &r.SendCount, &r.LeaseEpoch, &r.WakeAt, &r.SendDeadlineAt, &r.LastCode, &r.TaskID, &r.ReceiptJSON, &r.CreatedAt, &r.UpdatedAt}
	return row.Scan(append(args, extra...)...)
}
func (s *Store) Status(parent context.Context, id string) (Record, error) {
	if !idPattern.MatchString(id) {
		return Record{}, ErrInvalid
	}
	ctx, stop := context.WithTimeout(parent, 5*time.Second)
	defer stop()
	var r Record
	err := scanRecord(s.db.QueryRowContext(ctx, `SELECT `+recordColumns+` FROM sender_outbox WHERE namespace_id=? AND queue_id=? AND operation_id=?`, s.namespace, s.queue, id), &r)
	if errors.Is(err, sql.ErrNoRows) {
		return Record{}, ErrNotFound
	}
	if err != nil {
		return Record{}, ErrStorage
	}
	return r, nil
}

func (s *Store) claim(parent context.Context) (Record, error) {
	ctx, stop := context.WithTimeout(parent, 5*time.Second)
	defer stop()
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return Record{}, ErrStorage
	}
	defer tx.Rollback()
	var r Record
	var now time.Time
	err = scanRecord(tx.QueryRowContext(ctx, `SELECT `+recordColumns+`,UTC_TIMESTAMP(6) FROM sender_outbox FORCE INDEX(ix_sender_due) WHERE namespace_id=? AND queue_id=? AND wake_at<=UTC_TIMESTAMP(6) ORDER BY wake_at,operation_id LIMIT 1 FOR UPDATE SKIP LOCKED`, s.namespace, s.queue), &r, &now)
	if errors.Is(err, sql.ErrNoRows) {
		return Record{}, ErrIdle
	}
	if err != nil {
		return Record{}, ErrStorage
	}
	if r.State != "PENDING" && r.State != "SENDING" {
		return Record{}, ErrStorage
	}
	if !now.Before(r.SendDeadlineAt) || r.SendCount >= maxSends {
		r.State = "REVIEW"
		r.WakeAt = nil
		r.LastCode = "SEND_BUDGET_EXHAUSTED"
		if !now.Before(r.SendDeadlineAt) {
			r.LastCode = "SEND_DEADLINE"
		}
	} else {
		r.State = "SENDING"
		r.SendCount++
		r.LeaseEpoch++
		wake := now.Add(30 * time.Second)
		if wake.After(r.SendDeadlineAt) {
			wake = r.SendDeadlineAt
		}
		r.WakeAt = &wake
	}
	_, err = tx.ExecContext(ctx, `UPDATE sender_outbox SET state=?,send_count=?,lease_epoch=?,wake_at=?,last_code=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, r.State, r.SendCount, r.LeaseEpoch, r.WakeAt, r.LastCode, s.namespace, r.OperationID)
	if err != nil {
		return Record{}, ErrStorage
	}
	if tx.Commit() != nil {
		return Record{}, ErrCommitUnconfirmed
	}
	return r, nil
}

// finish 只接受最近代次；过期但尚未被替代的成功回执可落库，重复HTTP只复用原键。
func (s *Store) finish(parent context.Context, claim Record, state, code string, delay time.Duration, receipt *orbit.SubmitTaskResult) error {
	ctx, stop := context.WithTimeout(parent, 5*time.Second)
	defer stop()
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return ErrStorage
	}
	defer tx.Rollback()
	var current Record
	var now time.Time
	err = scanRecord(tx.QueryRowContext(ctx, `SELECT `+recordColumns+`,UTC_TIMESTAMP(6) FROM sender_outbox WHERE namespace_id=? AND queue_id=? AND operation_id=? FOR UPDATE`, s.namespace, s.queue, claim.OperationID), &current, &now)
	if err != nil {
		return ErrStorage
	}
	if current.State != "SENDING" || current.LeaseEpoch != claim.LeaseEpoch {
		return ErrLeaseLost
	}
	var wake *time.Time
	var task *string
	var raw []byte
	if state == "ACCEPTED" {
		if receipt == nil || !validReceipt(current, *receipt) {
			return ErrInvalid
		}
		raw, err = json.Marshal(receipt)
		if err != nil || len(raw) > 1<<20 {
			return ErrInvalid
		}
		task = &receipt.Task.ID
		var existing *string
		if err = tx.QueryRowContext(ctx, `SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=? FOR UPDATE`, s.namespace, current.OperationID).Scan(&existing); err != nil {
			return ErrStorage
		}
		if existing != nil && *existing != *task {
			return ErrConflict
		}
		if _, err = tx.ExecContext(ctx, `UPDATE sender_jobs SET task_id=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, *task, s.namespace, current.OperationID); err != nil {
			return ErrStorage
		}
	} else if state == "PENDING" {
		if delay < time.Second || delay > time.Hour {
			return ErrInvalid
		}
		if !now.Before(current.SendDeadlineAt) || current.SendCount >= maxSends {
			state = "REVIEW"
			code = "SEND_BUDGET_EXHAUSTED"
		} else {
			next := now.Add(delay)
			if next.After(current.SendDeadlineAt) {
				next = current.SendDeadlineAt
			}
			wake = &next
		}
	} else if state != "REVIEW" {
		return ErrInvalid
	}
	if _, err = tx.ExecContext(ctx, `UPDATE sender_outbox SET state=?,last_code=?,wake_at=?,task_id=?,receipt_json=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, state, code, wake, task, raw, s.namespace, current.OperationID); err != nil {
		return ErrStorage
	}
	if tx.Commit() != nil {
		return ErrCommitUnconfirmed
	}
	return nil
}