127 lines
4.4 KiB
Go
127 lines
4.4 KiB
Go
// Package outbox 演示在业务事务中使用公共 MySQL Outbox 适配包。
|
||
package jobs
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
_ "embed"
|
||
"encoding/json"
|
||
"errors"
|
||
"strings"
|
||
"unicode"
|
||
"unicode/utf8"
|
||
|
||
adapter "git.yuntuo.hk/k2safe/orbit/adapters/outbox/mysql"
|
||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||
)
|
||
|
||
//go:embed schema.sql
|
||
var schema string
|
||
var (
|
||
ErrInvalid = adapter.ErrInvalid
|
||
ErrStorage = adapter.ErrStorage
|
||
ErrConflict = adapter.ErrConflict
|
||
ErrCommitUnconfirmed = adapter.ErrCommitUnconfirmed
|
||
ErrIdle = adapter.ErrIdle
|
||
)
|
||
|
||
type Record = adapter.Record
|
||
|
||
// Store 仅补充示例业务事务；发送、查询、租约由公共包负责。
|
||
type Store struct {
|
||
*adapter.Store
|
||
db *sql.DB
|
||
namespace string
|
||
}
|
||
|
||
var Open = adapter.Open
|
||
|
||
func Init(ctx context.Context, db *sql.DB) error {
|
||
if err := adapter.Init(ctx, db); err != nil {
|
||
return err
|
||
}
|
||
if _, err := db.ExecContext(ctx, schema); err != nil {
|
||
return ErrStorage
|
||
}
|
||
return nil
|
||
}
|
||
func New(db *sql.DB, namespace, queue string) (*Store, error) {
|
||
s, err := adapter.New(db, namespace, queue, adapter.Options{OnAccepted: accepted})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &Store{s, db, namespace}, nil
|
||
}
|
||
func accepted(ctx context.Context, tx *sql.Tx, r adapter.Record, receipt orbit.SubmitTaskResult) error {
|
||
var existing *string
|
||
if err := tx.QueryRowContext(ctx, `SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=? FOR UPDATE`, r.NamespaceID, r.OperationID).Scan(&existing); err != nil {
|
||
return ErrStorage
|
||
}
|
||
if existing != nil && *existing != receipt.Task.ID {
|
||
return ErrConflict
|
||
}
|
||
_, err := tx.ExecContext(ctx, `UPDATE sender_jobs SET task_id=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, receipt.Task.ID, r.NamespaceID, r.OperationID)
|
||
if err != nil {
|
||
return ErrStorage
|
||
}
|
||
return nil
|
||
}
|
||
func (s *Store) Enqueue(ctx context.Context, id, source string, in orbit.SubmitTaskRequest) (bool, error) {
|
||
if !utf8.ValidString(source) || strings.TrimSpace(source) == "" || utf8.RuneCountInString(source) > 512 || strings.ContainsFunc(source, unicode.IsControl) {
|
||
return false, ErrInvalid
|
||
}
|
||
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
||
if err != nil {
|
||
return false, ErrStorage
|
||
}
|
||
defer tx.Rollback()
|
||
duplicate, err := s.Store.EnqueueTx(ctx, tx, id, in)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
if duplicate {
|
||
var saved string
|
||
if err = tx.QueryRowContext(ctx, `SELECT source_ref FROM sender_jobs WHERE namespace_id=? AND operation_id=? FOR UPDATE`, s.namespace, id).Scan(&saved); err != nil {
|
||
return false, ErrStorage
|
||
}
|
||
if saved != source {
|
||
return false, ErrConflict
|
||
}
|
||
} else {
|
||
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, id, source); err != nil {
|
||
return false, ErrStorage
|
||
}
|
||
}
|
||
if tx.Commit() != nil {
|
||
return false, ErrCommitUnconfirmed
|
||
}
|
||
return duplicate, nil
|
||
}
|
||
|
||
// View 是业务HTTP的读取结果；Submission只表示提交阶段，Execution来自持久化Inbox。
|
||
type View struct {
|
||
OperationID string `json:"operation_id"` // OperationID 调用方提供的稳定操作标识。
|
||
Submission string `json:"submission"` // Submission 待发/发送中/已受理/需核查，不是执行结果。
|
||
Code string `json:"code"` // Code 最近稳定的提交分类。
|
||
TaskID *string `json:"task_id,omitempty"` // TaskID 已核对的平台任务，未确认时为空。
|
||
Execution string `json:"execution,omitempty"` // Execution 业务Inbox最后收到的任务状态。
|
||
Result json.RawMessage `json:"result,omitempty"` // Result 业务Inbox中的结果，尚未到达时为空。
|
||
}
|
||
|
||
func (s *Store) View(ctx context.Context, id string) (View, error) {
|
||
r, err := s.Status(ctx, id)
|
||
if err != nil {
|
||
return View{}, err
|
||
}
|
||
v := View{OperationID: id, Submission: r.State, Code: r.LastCode, TaskID: r.TaskID}
|
||
if r.TaskID != nil {
|
||
var result []byte
|
||
err = s.db.QueryRowContext(ctx, `SELECT last_state,result_json FROM receiver_task_views WHERE consumer_name='http-service' AND queue_id=? AND task_id=?`, r.QueueID, *r.TaskID).Scan(&v.Execution, &result)
|
||
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
||
return View{}, ErrStorage
|
||
}
|
||
v.Result = result
|
||
}
|
||
return v, nil
|
||
}
|