Files
cmsp/internal/store/upload_operation.go
T

82 lines
3.0 KiB
Go

package store
import (
"database/sql"
"errors"
"fmt"
)
var ErrUploadOperationPending = errors.New("视频上传操作尚未终结")
// UploadOperation 记录本次远端写操作的恢复身份,绝不包含 API Key 或响应原文。
type UploadOperation struct {
ProductID string
ShopeeID string
IdempotencyKey string
FilePath string
FileSHA256 string
FileSize int64
Status string
OperationID string
VideoURL string
ErrorCode string
UpdatedAt string
}
func (s *Store) GetUploadOperation(productID string) (UploadOperation, bool, error) {
var op UploadOperation
err := s.db.QueryRow(`SELECT product_id, shopee_id, idempotency_key, file_path, file_sha256,
file_size, status, operation_id, video_url, error_code, updated_at
FROM video_upload_operations WHERE product_id = ?`, productID).Scan(
&op.ProductID, &op.ShopeeID, &op.IdempotencyKey, &op.FilePath, &op.FileSHA256,
&op.FileSize, &op.Status, &op.OperationID, &op.VideoURL, &op.ErrorCode, &op.UpdatedAt)
if errors.Is(err, sql.ErrNoRows) {
return UploadOperation{}, false, nil
}
if err != nil {
return UploadOperation{}, false, fmt.Errorf("读取视频上传操作失败:%w", err)
}
return op, true, nil
}
// BeginUploadOperation 只允许覆盖明确终结的记录。并发/重启后未决操作不能换键再写。
func (s *Store) BeginUploadOperation(op UploadOperation) error {
result, err := s.db.Exec(`INSERT INTO video_upload_operations (
product_id, shopee_id, idempotency_key, file_path, file_sha256, file_size, status, updated_at
) VALUES (?, ?, ?, ?, ?, ?, 'created', ?)
ON CONFLICT(product_id) DO UPDATE SET
shopee_id=excluded.shopee_id, idempotency_key=excluded.idempotency_key,
file_path=excluded.file_path, file_sha256=excluded.file_sha256,
file_size=excluded.file_size, status='created', operation_id='', video_url='', error_code='',
updated_at=excluded.updated_at
WHERE video_upload_operations.status IN ('succeeded', 'failed')`,
op.ProductID, op.ShopeeID, op.IdempotencyKey, op.FilePath, op.FileSHA256, op.FileSize, op.UpdatedAt)
if err != nil {
return fmt.Errorf("保存视频上传操作失败:%w", err)
}
changed, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("确认视频上传操作失败:%w", err)
}
if changed != 1 {
return ErrUploadOperationPending
}
return nil
}
// UpdateUploadOperation 只更新与当前幂等键匹配的操作,避免旧响应覆盖新操作。
func (s *Store) UpdateUploadOperation(op UploadOperation) error {
result, err := s.db.Exec(`UPDATE video_upload_operations SET status=?, operation_id=?,
video_url=?, error_code=?, updated_at=? WHERE product_id=? AND idempotency_key=?`,
op.Status, op.OperationID, op.VideoURL, op.ErrorCode, op.UpdatedAt,
op.ProductID, op.IdempotencyKey)
if err != nil {
return fmt.Errorf("更新视频上传操作失败:%w", err)
}
changed, err := result.RowsAffected()
if err != nil || changed != 1 {
return fmt.Errorf("视频上传操作身份已变化")
}
return nil
}