95 lines
3.2 KiB
Go
95 lines
3.2 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
)
|
||
|
||
type transactionRow struct {
|
||
TransactionID string
|
||
MetadataJSON string
|
||
}
|
||
|
||
func (r *Repository) lookupTransaction(ctx context.Context, tx *sql.Tx, commandID string, requestHash string, bizType string) (transactionRow, bool, error) {
|
||
return r.lookupTransactionWithConflictCode(ctx, tx, commandID, requestHash, bizType, xerr.LedgerConflict)
|
||
}
|
||
|
||
func (r *Repository) lookupTransactionWithConflictCode(ctx context.Context, tx *sql.Tx, commandID string, requestHash string, bizType string, conflictCode xerr.Code) (transactionRow, bool, error) {
|
||
row := tx.QueryRowContext(ctx,
|
||
`SELECT transaction_id, request_hash, biz_type, COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wallet_transactions
|
||
WHERE app_code = ? AND command_id = ?
|
||
FOR UPDATE`,
|
||
appcode.FromContext(ctx),
|
||
commandID,
|
||
)
|
||
|
||
var txRow transactionRow
|
||
var storedHash string
|
||
var storedBizType string
|
||
if err := row.Scan(&txRow.TransactionID, &storedHash, &storedBizType, &txRow.MetadataJSON); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return transactionRow{}, false, nil
|
||
}
|
||
return transactionRow{}, false, err
|
||
}
|
||
if storedHash != requestHash || storedBizType != bizType {
|
||
|
||
return transactionRow{}, true, xerr.New(conflictCode, "wallet command idempotency conflict")
|
||
}
|
||
return txRow, true, nil
|
||
}
|
||
|
||
// lookupTransactionConsistent 不对不存在的唯一键做 gap lock，供能在 INSERT duplicate 后安全回读的命令使用。
|
||
// 这避免两个同 command 首发都持 gap lock、再以相反顺序等待业务行锁和唯一键插入造成死锁。
|
||
func (r *Repository) lookupTransactionConsistent(ctx context.Context, tx *sql.Tx, commandID string, requestHash string, bizType string, conflictCode xerr.Code) (transactionRow, bool, error) {
|
||
row := tx.QueryRowContext(ctx,
|
||
`SELECT transaction_id, request_hash, biz_type, COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wallet_transactions
|
||
WHERE app_code = ? AND command_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
commandID,
|
||
)
|
||
var txRow transactionRow
|
||
var storedHash string
|
||
var storedBizType string
|
||
if err := row.Scan(&txRow.TransactionID, &storedHash, &storedBizType, &txRow.MetadataJSON); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return transactionRow{}, false, nil
|
||
}
|
||
return transactionRow{}, false, err
|
||
}
|
||
if storedHash != requestHash || storedBizType != bizType {
|
||
return transactionRow{}, true, xerr.New(conflictCode, "wallet command idempotency conflict")
|
||
}
|
||
return txRow, true, nil
|
||
}
|
||
|
||
func (r *Repository) insertTransaction(ctx context.Context, tx *sql.Tx, transactionID string, commandID string, bizType string, requestHash string, externalRef string, metadata any, nowMs int64) error {
|
||
metadataJSON, err := json.Marshal(metadata)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx,
|
||
`INSERT INTO wallet_transactions (
|
||
app_code, transaction_id, command_id, biz_type, status, request_hash, external_ref, metadata_json, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
appcode.FromContext(ctx),
|
||
transactionID,
|
||
commandID,
|
||
bizType,
|
||
transactionStatusSucceeded,
|
||
requestHash,
|
||
externalRef,
|
||
string(metadataJSON),
|
||
nowMs,
|
||
nowMs,
|
||
)
|
||
return err
|
||
}
|