130 lines
4.7 KiB
Go
130 lines
4.7 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
|
||
}
|
||
|
||
// updateTransactionMetadata replaces the provisional transaction snapshot after every tracked account
|
||
// mutation has succeeded. Gift transactions must be inserted before ledger entries for referential and
|
||
// idempotency ordering, but balance_version is authoritative only after applyTrackedAccountDelta has
|
||
// advanced the locked account state. Persisting the final JSON in the same transaction makes a later
|
||
// command_id replay return the exact original balance and version instead of reading mutable live state.
|
||
func (r *Repository) updateTransactionMetadata(ctx context.Context, tx *sql.Tx, transactionID string, metadata any, nowMS int64) error {
|
||
metadataJSON, err := json.Marshal(metadata)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
result, err := tx.ExecContext(ctx,
|
||
`UPDATE wallet_transactions
|
||
SET metadata_json = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND transaction_id = ?`,
|
||
string(metadataJSON),
|
||
nowMS,
|
||
appcode.FromContext(ctx),
|
||
transactionID,
|
||
)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
affected, err := result.RowsAffected()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// go-sql-driver/mysql 默认返回 changed rows，不返回 matched rows。当最终 metadata
|
||
// 与插入时的预测值完全一致时，这条 UPDATE 合法命中主键但 affected=0；交易行已在
|
||
// 同一事务中由 insertTransaction 成功插入，因此只能把多行更新视为不变量破坏。
|
||
if affected > 1 {
|
||
return xerr.New(xerr.Internal, "wallet transaction metadata update lost its transaction")
|
||
}
|
||
return nil
|
||
}
|