feat(admin): authorize batch purchase starts
This commit is contained in:
QiuSW
@@ -0,0 +1,128 @@
|
||||
package tasks
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"math"
|
||||
"math/big"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
_ "time/tzdata"
|
||||
)
|
||||
|
||||
const maxStartItems = 100
|
||||
|
||||
var (
|
||||
ErrStartConflict = errors.New("purchase start conflicts with current task state")
|
||||
ErrInvalidStart = errors.New("invalid purchase start request")
|
||||
)
|
||||
|
||||
type StartPolicy struct {
|
||||
AuthorizationTTL time.Duration
|
||||
MaxQuantity int
|
||||
MaxTotalPrice string
|
||||
}
|
||||
type StartItem struct {
|
||||
TaskID string `json:"task_id"`
|
||||
ExpectedTaskVersion int `json:"expected_task_version"`
|
||||
}
|
||||
type StartCommand struct {
|
||||
StartKey string `json:"start_key"`
|
||||
Tasks []StartItem `json:"tasks"`
|
||||
}
|
||||
type AuthorizedTask struct {
|
||||
TaskID string `json:"task_id"`
|
||||
TaskVersion int `json:"task_version"`
|
||||
AuthorizationID string `json:"authorization_id"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
type StartResult struct {
|
||||
StartKey string `json:"start_key"`
|
||||
AuthorizedCount int `json:"authorized_count"`
|
||||
Tasks []AuthorizedTask `json:"tasks"`
|
||||
PaymentAutomated bool `json:"payment_automated"`
|
||||
}
|
||||
type TaskFilter struct{ Keyword, Status, CreatedFrom, CreatedTo string }
|
||||
type TaskRow struct {
|
||||
ID, Title, GoodsID, SKUColor, SKUSize, MaxTotalPrice, Status string
|
||||
Quantity, Version int
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
func normalizeCents(value string) (string, *big.Int, bool) {
|
||||
if value == "" || strings.TrimSpace(value) != value {
|
||||
return "", nil, false
|
||||
}
|
||||
parts := strings.Split(value, ".")
|
||||
if len(parts) != 2 || len(parts[0]) == 0 || len(parts[1]) != 2 || (len(parts[0]) > 1 && parts[0][0] == '0') {
|
||||
return "", nil, false
|
||||
}
|
||||
for _, part := range parts {
|
||||
for _, ch := range part {
|
||||
if ch < '0' || ch > '9' {
|
||||
return "", nil, false
|
||||
}
|
||||
}
|
||||
}
|
||||
cents := new(big.Int)
|
||||
if _, ok := cents.SetString(parts[0]+parts[1], 10); !ok || cents.Sign() <= 0 {
|
||||
return "", nil, false
|
||||
}
|
||||
return value, cents, true
|
||||
}
|
||||
|
||||
func startItems(command StartCommand) ([]StartItem, error) {
|
||||
if !validUUID(command.StartKey) || len(command.Tasks) == 0 || len(command.Tasks) > maxStartItems {
|
||||
return nil, ErrInvalidStart
|
||||
}
|
||||
items := append([]StartItem(nil), command.Tasks...)
|
||||
sort.Slice(items, func(i, j int) bool { return items[i].TaskID < items[j].TaskID })
|
||||
for i, item := range items {
|
||||
if !validUUID(item.TaskID) || item.ExpectedTaskVersion <= 0 || item.ExpectedTaskVersion == math.MaxInt || (i > 0 && item.TaskID == items[i-1].TaskID) {
|
||||
return nil, ErrInvalidStart
|
||||
}
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
func validTaskStatus(value string) bool {
|
||||
if value == "" {
|
||||
return true
|
||||
}
|
||||
for _, status := range []string{"DRAFT", "PENDING", "CLAIMED", "ORDERING", "NEEDS_MANUAL", "WAITING_PAYMENT", "RECONCILIATION_REQUIRED", "SUCCEEDED", "FAILED", "CANCELED"} {
|
||||
if value == status {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func ShanghaiRange(from, to string) (time.Time, time.Time, error) {
|
||||
if from == "" && to == "" {
|
||||
return time.Time{}, time.Time{}, nil
|
||||
}
|
||||
location, err := time.LoadLocation("Asia/Shanghai")
|
||||
if err != nil {
|
||||
return time.Time{}, time.Time{}, err
|
||||
}
|
||||
parse := func(value string) (time.Time, error) { return time.ParseInLocation("2006-01-02", value, location) }
|
||||
var start, end time.Time
|
||||
if from != "" {
|
||||
start, err = parse(from)
|
||||
if err != nil {
|
||||
return time.Time{}, time.Time{}, ErrInvalidStart
|
||||
}
|
||||
start = start.UTC()
|
||||
}
|
||||
if to != "" {
|
||||
end, err = parse(to)
|
||||
if err != nil {
|
||||
return time.Time{}, time.Time{}, ErrInvalidStart
|
||||
}
|
||||
end = end.AddDate(0, 0, 1).UTC()
|
||||
}
|
||||
if !start.IsZero() && !end.IsZero() && !start.Before(end) {
|
||||
return time.Time{}, time.Time{}, ErrInvalidStart
|
||||
}
|
||||
return start, end, nil
|
||||
}
|
||||
@@ -0,0 +1,424 @@
|
||||
package tasks
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"reflect"
|
||||
"sort"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"cmbuyer/admin/internal/migrations"
|
||||
)
|
||||
|
||||
var fixedStartTime = time.Date(2026, 8, 4, 9, 2, 3, 456000000, time.FixedZone("UTC+8", 8*60*60))
|
||||
|
||||
func TestStartPurchasesPersistsCompleteSnapshotsForOneAndHundredTasks(t *testing.T) {
|
||||
for _, count := range []int{1, 100} {
|
||||
t.Run(fmt.Sprintf("%d tasks", count), func(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store := configuredStartStore(t, database)
|
||||
store.now = func() time.Time { return fixedStartTime }
|
||||
items := make([]StartItem, 0, count)
|
||||
wantDrafts := make(map[string]Draft, count)
|
||||
for index := 1; index <= count; index++ {
|
||||
id := startTestUUID(index)
|
||||
draft := Draft{
|
||||
ID: id,
|
||||
Title: fmt.Sprintf("task-%03d", index),
|
||||
GoodsID: fmt.Sprintf("937122%06d", index),
|
||||
SKUColor: fmt.Sprintf("color-%03d", index),
|
||||
SKUSize: fmt.Sprintf("size-%03d", index),
|
||||
Quantity: index%10 + 1,
|
||||
MaxTotalPrice: fmt.Sprintf("%d.%02d", index+10, index%100),
|
||||
}
|
||||
if _, err := store.CreateDraft(context.Background(), draft); err != nil {
|
||||
t.Fatalf("create draft %d: %v", index, err)
|
||||
}
|
||||
items = append(items, StartItem{TaskID: id, ExpectedTaskVersion: 1})
|
||||
wantDrafts[id] = draft
|
||||
}
|
||||
sort.Slice(items, func(i, j int) bool { return items[i].TaskID > items[j].TaskID })
|
||||
command := StartCommand{StartKey: startTestUUID(1001 + count), Tasks: items}
|
||||
|
||||
result, err := store.StartPurchases(context.Background(), command, "authenticated-admin")
|
||||
if err != nil {
|
||||
t.Fatalf("StartPurchases: %v", err)
|
||||
}
|
||||
if result.StartKey != command.StartKey || result.AuthorizedCount != count || result.PaymentAutomated || len(result.Tasks) != count {
|
||||
t.Fatalf("result = %#v", result)
|
||||
}
|
||||
wantCreated := fixedStartTime.UTC()
|
||||
wantExpires := wantCreated.Add(15 * time.Minute)
|
||||
seenAuthorizationIDs := map[string]bool{}
|
||||
for index, authorized := range result.Tasks {
|
||||
if index > 0 && result.Tasks[index-1].TaskID >= authorized.TaskID {
|
||||
t.Fatalf("result is not in canonical task order: %#v", result.Tasks)
|
||||
}
|
||||
if authorized.TaskVersion != 2 || !authorized.ExpiresAt.Equal(wantExpires) || !validUUID(authorized.AuthorizationID) || seenAuthorizationIDs[authorized.AuthorizationID] {
|
||||
t.Fatalf("authorized task = %#v", authorized)
|
||||
}
|
||||
seenAuthorizationIDs[authorized.AuthorizationID] = true
|
||||
want := wantDrafts[authorized.TaskID]
|
||||
var taskStatus, taskUpdated, authTaskID, authStartKey, goodsID, color, size, priceCap, authStatus, createdBy, createdAt, expiresAt string
|
||||
var taskVersion, authTaskVersion, quantity int
|
||||
err := database.QueryRow(`
|
||||
SELECT t.status,t.version,t.updated_at,
|
||||
a.task_id,a.task_version,a.start_key,a.goods_id,a.sku_color,a.sku_size,a.quantity,a.total_price_cap,a.status,a.created_by,a.created_at,a.expires_at
|
||||
FROM tasks t JOIN order_authorizations a ON a.task_id=t.id WHERE a.id=?`, authorized.AuthorizationID).
|
||||
Scan(&taskStatus, &taskVersion, &taskUpdated, &authTaskID, &authTaskVersion, &authStartKey, &goodsID, &color, &size, &quantity, &priceCap, &authStatus, &createdBy, &createdAt, &expiresAt)
|
||||
if err != nil {
|
||||
t.Fatalf("read authorization snapshot: %v", err)
|
||||
}
|
||||
if taskStatus != "PENDING" || taskVersion != 2 || taskUpdated != wantCreated.Format(time.RFC3339Nano) ||
|
||||
authTaskID != want.ID || authTaskVersion != 2 || authStartKey != command.StartKey ||
|
||||
goodsID != want.GoodsID || color != want.SKUColor || size != want.SKUSize || quantity != want.Quantity || priceCap != want.MaxTotalPrice ||
|
||||
authStatus != "ACTIVE" || createdBy != "authenticated-admin" || createdAt != wantCreated.Format(time.RFC3339Nano) || expiresAt != wantExpires.Format(time.RFC3339Nano) {
|
||||
t.Fatalf("stored task/authorization mismatch for %s", want.ID)
|
||||
}
|
||||
}
|
||||
var distinctCreated, distinctExpires int
|
||||
if err := database.QueryRow(`SELECT COUNT(DISTINCT created_at), COUNT(DISTINCT expires_at) FROM order_authorizations WHERE start_key=?`, command.StartKey).Scan(&distinctCreated, &distinctExpires); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if distinctCreated != 1 || distinctExpires != 1 {
|
||||
t.Fatalf("batch timestamps are not shared: created=%d expires=%d", distinctCreated, distinctExpires)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartPurchasesRejectsInvalidCommandsAndPolicyWithoutWrites(t *testing.T) {
|
||||
validItem := StartItem{TaskID: startTestUUID(1), ExpectedTaskVersion: 1}
|
||||
hundredOne := make([]StartItem, 101)
|
||||
for index := range hundredOne {
|
||||
hundredOne[index] = StartItem{TaskID: startTestUUID(index + 1), ExpectedTaskVersion: 1}
|
||||
}
|
||||
for name, command := range map[string]StartCommand{
|
||||
"invalid start key": {StartKey: "not-a-uuid", Tasks: []StartItem{validItem}},
|
||||
"empty tasks": {StartKey: startTestUUID(1001)},
|
||||
"over batch limit": {StartKey: startTestUUID(1001), Tasks: hundredOne},
|
||||
"invalid task id": {StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: "1", ExpectedTaskVersion: 1}}},
|
||||
"duplicate task": {StartKey: startTestUUID(1001), Tasks: []StartItem{validItem, validItem}},
|
||||
"zero version": {StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: validItem.TaskID}}},
|
||||
"overflow version": {StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: validItem.TaskID, ExpectedTaskVersion: math.MaxInt}}},
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store := configuredStartStore(t, database)
|
||||
_, err := store.StartPurchases(context.Background(), command, "admin")
|
||||
if !errors.Is(err, ErrInvalidStart) {
|
||||
t.Fatalf("error = %v, want ErrInvalidStart", err)
|
||||
}
|
||||
assertAuthorizationCount(t, database, 0)
|
||||
})
|
||||
}
|
||||
|
||||
for name, mutate := range map[string]func(*SQLiteStore){
|
||||
"zero ttl": func(store *SQLiteStore) { store.policy.AuthorizationTTL = 0 },
|
||||
"zero quantity": func(store *SQLiteStore) { store.policy.MaxQuantity = 0 },
|
||||
"bad max price": func(store *SQLiteStore) { store.policy.MaxTotalPrice = "999" },
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store := configuredStartStore(t, database)
|
||||
createStartDraft(t, store, validItem.TaskID)
|
||||
mutate(store)
|
||||
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{validItem}}, "admin")
|
||||
if !errors.Is(err, ErrInvalidStart) {
|
||||
t.Fatalf("error = %v, want ErrInvalidStart", err)
|
||||
}
|
||||
assertDraftUnchanged(t, database, validItem.TaskID)
|
||||
assertAuthorizationCount(t, database, 0)
|
||||
})
|
||||
}
|
||||
|
||||
database := migratedDatabase(t)
|
||||
store := configuredStartStore(t, database)
|
||||
createStartDraft(t, store, validItem.TaskID)
|
||||
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{validItem}}, "")
|
||||
if !errors.Is(err, ErrInvalidStart) {
|
||||
t.Fatalf("empty created_by error = %v", err)
|
||||
}
|
||||
assertDraftUnchanged(t, database, validItem.TaskID)
|
||||
}
|
||||
|
||||
func TestStartPurchasesRejectsEveryTaskConflictWithoutAuthorization(t *testing.T) {
|
||||
for name, mutate := range map[string]func(*testing.T, *SQLiteStore, string, *StartItem){
|
||||
"missing": func(_ *testing.T, _ *SQLiteStore, _ string, item *StartItem) {
|
||||
item.TaskID = startTestUUID(99)
|
||||
},
|
||||
"not draft": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
|
||||
execTestSQL(t, store.database, `UPDATE tasks SET status='PENDING' WHERE id=?`, id)
|
||||
},
|
||||
"version mismatch": func(_ *testing.T, _ *SQLiteStore, _ string, item *StartItem) {
|
||||
item.ExpectedTaskVersion = 2
|
||||
},
|
||||
"empty goods id": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
|
||||
execTestSQL(t, store.database, `UPDATE tasks SET goods_id='' WHERE id=?`, id)
|
||||
},
|
||||
"nondigit goods id": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
|
||||
execTestSQL(t, store.database, `UPDATE tasks SET goods_id='937x' WHERE id=?`, id)
|
||||
},
|
||||
"empty color": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
|
||||
execTestSQL(t, store.database, `UPDATE tasks SET sku_color='' WHERE id=?`, id)
|
||||
},
|
||||
"empty size": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
|
||||
execTestSQL(t, store.database, `UPDATE tasks SET sku_size='' WHERE id=?`, id)
|
||||
},
|
||||
"quantity over policy": func(_ *testing.T, store *SQLiteStore, _ string, _ *StartItem) {
|
||||
store.policy.MaxQuantity = 1
|
||||
},
|
||||
"noncanonical price one decimal": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
|
||||
execTestSQL(t, store.database, `UPDATE tasks SET max_total_price='12.8' WHERE id=?`, id)
|
||||
},
|
||||
"noncanonical leading zero": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
|
||||
execTestSQL(t, store.database, `UPDATE tasks SET max_total_price='012.80' WHERE id=?`, id)
|
||||
},
|
||||
"price over policy": func(_ *testing.T, store *SQLiteStore, _ string, _ *StartItem) {
|
||||
store.policy.MaxTotalPrice = "12.79"
|
||||
},
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store := configuredStartStore(t, database)
|
||||
id := startTestUUID(1)
|
||||
createStartDraft(t, store, id)
|
||||
item := StartItem{TaskID: id, ExpectedTaskVersion: 1}
|
||||
mutate(t, store, id, &item)
|
||||
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{item}}, "admin")
|
||||
if !errors.Is(err, ErrStartConflict) {
|
||||
t.Fatalf("error = %v, want ErrStartConflict", err)
|
||||
}
|
||||
assertAuthorizationCount(t, database, 0)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartPurchasesRollsBackWholeBatchForLateConflictAndSQLFailure(t *testing.T) {
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
breakBatch func(*testing.T, *SQLiteStore, string)
|
||||
}{
|
||||
{name: "late validation conflict", breakBatch: func(t *testing.T, store *SQLiteStore, secondID string) {
|
||||
execTestSQL(t, store.database, `UPDATE tasks SET sku_size='' WHERE id=?`, secondID)
|
||||
}},
|
||||
{name: "late SQL failure", breakBatch: func(t *testing.T, store *SQLiteStore, secondID string) {
|
||||
statement := fmt.Sprintf(`CREATE TRIGGER reject_second_authorization BEFORE INSERT ON order_authorizations WHEN NEW.task_id='%s' BEGIN SELECT RAISE(ABORT, 'test failure'); END`, secondID)
|
||||
execTestSQL(t, store.database, statement)
|
||||
}},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store := configuredStartStore(t, database)
|
||||
firstID, secondID := startTestUUID(1), startTestUUID(2)
|
||||
createStartDraft(t, store, firstID)
|
||||
createStartDraft(t, store, secondID)
|
||||
test.breakBatch(t, store, secondID)
|
||||
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: firstID, ExpectedTaskVersion: 1}, {TaskID: secondID, ExpectedTaskVersion: 1}}}, "admin")
|
||||
if err == nil {
|
||||
t.Fatal("StartPurchases unexpectedly succeeded")
|
||||
}
|
||||
assertDraftUnchanged(t, database, firstID)
|
||||
var secondStatus string
|
||||
var secondVersion int
|
||||
if err := database.QueryRow(`SELECT status,version FROM tasks WHERE id=?`, secondID).Scan(&secondStatus, &secondVersion); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if secondStatus != "DRAFT" || secondVersion != 1 {
|
||||
t.Fatalf("second task = %s/v%d, want DRAFT/v1", secondStatus, secondVersion)
|
||||
}
|
||||
assertAuthorizationCount(t, database, 0)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartPurchasesReplayIsStableAndRejectsDifferentOrIncompleteSets(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store := configuredStartStore(t, database)
|
||||
firstID, secondID, thirdID := startTestUUID(1), startTestUUID(2), startTestUUID(3)
|
||||
for _, id := range []string{firstID, secondID, thirdID} {
|
||||
createStartDraft(t, store, id)
|
||||
}
|
||||
command := StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: secondID, ExpectedTaskVersion: 1}, {TaskID: firstID, ExpectedTaskVersion: 1}}}
|
||||
first, err := store.StartPurchases(context.Background(), command, "admin")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
command.Tasks[0], command.Tasks[1] = command.Tasks[1], command.Tasks[0]
|
||||
replay, err := store.StartPurchases(context.Background(), command, "admin")
|
||||
if err != nil || !reflect.DeepEqual(replay, first) {
|
||||
t.Fatalf("replay = (%#v, %v), want %#v", replay, err, first)
|
||||
}
|
||||
assertAuthorizationCount(t, database, 2)
|
||||
|
||||
conflicting := []StartCommand{
|
||||
{StartKey: command.StartKey, Tasks: command.Tasks[:1]},
|
||||
{StartKey: command.StartKey, Tasks: []StartItem{{TaskID: firstID, ExpectedTaskVersion: 2}, {TaskID: secondID, ExpectedTaskVersion: 1}}},
|
||||
{StartKey: command.StartKey, Tasks: []StartItem{{TaskID: firstID, ExpectedTaskVersion: 1}, {TaskID: secondID, ExpectedTaskVersion: 1}, {TaskID: thirdID, ExpectedTaskVersion: 1}}},
|
||||
}
|
||||
for _, changed := range conflicting {
|
||||
if _, err := store.StartPurchases(context.Background(), changed, "admin"); !errors.Is(err, ErrStartConflict) {
|
||||
t.Fatalf("different payload error = %v", err)
|
||||
}
|
||||
}
|
||||
assertAuthorizationCount(t, database, 2)
|
||||
assertDraftUnchanged(t, database, thirdID)
|
||||
|
||||
execTestSQL(t, database, `DELETE FROM order_authorizations WHERE task_id=?`, secondID)
|
||||
if _, err := store.StartPurchases(context.Background(), command, "admin"); !errors.Is(err, ErrStartConflict) {
|
||||
t.Fatalf("incomplete replay error = %v", err)
|
||||
}
|
||||
assertAuthorizationCount(t, database, 1)
|
||||
}
|
||||
|
||||
func TestStartPurchasesConcurrentReplayAndVersionRace(t *testing.T) {
|
||||
t.Run("same key replays one stable result", func(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store := configuredStartStore(t, database)
|
||||
id := startTestUUID(1)
|
||||
createStartDraft(t, store, id)
|
||||
command := StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: id, ExpectedTaskVersion: 1}}}
|
||||
const callers = 16
|
||||
start := make(chan struct{})
|
||||
results := make(chan StartResult, callers)
|
||||
errorsChannel := make(chan error, callers)
|
||||
var group sync.WaitGroup
|
||||
for range callers {
|
||||
group.Add(1)
|
||||
go func() {
|
||||
defer group.Done()
|
||||
<-start
|
||||
result, err := store.StartPurchases(context.Background(), command, "admin")
|
||||
if err != nil {
|
||||
errorsChannel <- err
|
||||
return
|
||||
}
|
||||
results <- result
|
||||
}()
|
||||
}
|
||||
close(start)
|
||||
group.Wait()
|
||||
close(results)
|
||||
close(errorsChannel)
|
||||
for err := range errorsChannel {
|
||||
t.Fatalf("concurrent replay: %v", err)
|
||||
}
|
||||
var want StartResult
|
||||
for result := range results {
|
||||
if want.StartKey == "" {
|
||||
want = result
|
||||
} else if !reflect.DeepEqual(result, want) {
|
||||
t.Fatalf("unstable replay: %#v != %#v", result, want)
|
||||
}
|
||||
}
|
||||
assertAuthorizationCount(t, database, 1)
|
||||
var version int
|
||||
if err := database.QueryRow(`SELECT version FROM tasks WHERE id=?`, id).Scan(&version); err != nil || version != 2 {
|
||||
t.Fatalf("task version = %d, err=%v", version, err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("different keys race one expected version", func(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store := configuredStartStore(t, database)
|
||||
id := startTestUUID(1)
|
||||
createStartDraft(t, store, id)
|
||||
start := make(chan struct{})
|
||||
errorsChannel := make(chan error, 2)
|
||||
var group sync.WaitGroup
|
||||
for _, key := range []string{startTestUUID(1001), startTestUUID(1002)} {
|
||||
group.Add(1)
|
||||
go func(startKey string) {
|
||||
defer group.Done()
|
||||
<-start
|
||||
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startKey, Tasks: []StartItem{{TaskID: id, ExpectedTaskVersion: 1}}}, "admin")
|
||||
errorsChannel <- err
|
||||
}(key)
|
||||
}
|
||||
close(start)
|
||||
group.Wait()
|
||||
close(errorsChannel)
|
||||
successes, conflicts := 0, 0
|
||||
for err := range errorsChannel {
|
||||
switch {
|
||||
case err == nil:
|
||||
successes++
|
||||
case errors.Is(err, ErrStartConflict):
|
||||
conflicts++
|
||||
default:
|
||||
t.Fatalf("unexpected race error: %v", err)
|
||||
}
|
||||
}
|
||||
if successes != 1 || conflicts != 1 {
|
||||
t.Fatalf("success/conflict = %d/%d, want 1/1", successes, conflicts)
|
||||
}
|
||||
assertAuthorizationCount(t, database, 1)
|
||||
})
|
||||
}
|
||||
|
||||
func TestSQLiteStoreRejectsV1SchemaAtStartup(t *testing.T) {
|
||||
database := openDatabase(t)
|
||||
if err := migrations.Run(context.Background(), database, migrationDirectory(t), "up-by-one"); err != nil {
|
||||
t.Fatalf("migrate to v1: %v", err)
|
||||
}
|
||||
if _, err := NewSQLiteStore(database); err == nil {
|
||||
t.Fatal("NewSQLiteStore accepted the v1 two-pass schema")
|
||||
}
|
||||
}
|
||||
|
||||
func configuredStartStore(t *testing.T, database *sql.DB) *SQLiteStore {
|
||||
t.Helper()
|
||||
store, err := NewSQLiteStore(database)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSQLiteStore: %v", err)
|
||||
}
|
||||
store.SetStartPolicy(StartPolicy{AuthorizationTTL: 15 * time.Minute, MaxQuantity: 10, MaxTotalPrice: "999.99"})
|
||||
return store
|
||||
}
|
||||
|
||||
func createStartDraft(t *testing.T, store *SQLiteStore, id string) {
|
||||
t.Helper()
|
||||
draft := Draft{ID: id, Title: "test", GoodsID: "937122477375", SKUColor: "黑色", SKUSize: "M", Quantity: 2, MaxTotalPrice: "12.80"}
|
||||
if _, err := store.CreateDraft(context.Background(), draft); err != nil {
|
||||
t.Fatalf("CreateDraft: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func startTestUUID(number int) string {
|
||||
return fmt.Sprintf("%08x-1234-4abc-a123-%012x", number, number)
|
||||
}
|
||||
|
||||
func assertAuthorizationCount(t *testing.T, database *sql.DB, want int) {
|
||||
t.Helper()
|
||||
var got int
|
||||
if err := database.QueryRow(`SELECT COUNT(*) FROM order_authorizations`).Scan(&got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got != want {
|
||||
t.Fatalf("authorization count = %d, want %d", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func assertDraftUnchanged(t *testing.T, database *sql.DB, id string) {
|
||||
t.Helper()
|
||||
var status string
|
||||
var version int
|
||||
if err := database.QueryRow(`SELECT status,version FROM tasks WHERE id=?`, id).Scan(&status, &version); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if status != "DRAFT" || version != 1 {
|
||||
t.Fatalf("task %s = %s/v%d, want DRAFT/v1", id, status, version)
|
||||
}
|
||||
}
|
||||
|
||||
func execTestSQL(t *testing.T, database *sql.DB, statement string, arguments ...any) {
|
||||
t.Helper()
|
||||
if _, err := database.Exec(statement, arguments...); err != nil {
|
||||
t.Fatalf("execute test SQL: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,229 @@
|
||||
package tasks
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"cmbuyer/admin/internal/domain"
|
||||
)
|
||||
|
||||
// ErrInvalidFilter 表示任务筛选值无效，路由应按字段重新渲染而不是泄露内部错误。
|
||||
var ErrInvalidFilter = errors.New("invalid task filter")
|
||||
|
||||
// SetStartPolicy is called during startup; policy is explicit because authorization limits must not be implicit defaults.
|
||||
func (store *SQLiteStore) SetStartPolicy(policy StartPolicy) { store.policy = policy }
|
||||
|
||||
func (store *SQLiteStore) ListTasks(ctx context.Context, filter TaskFilter) ([]TaskRow, error) {
|
||||
if !ValidateTaskFilter(filter).Valid() {
|
||||
return nil, ErrInvalidFilter
|
||||
}
|
||||
from, to, err := ShanghaiRange(filter.CreatedFrom, filter.CreatedTo)
|
||||
if err != nil {
|
||||
return nil, ErrInvalidFilter
|
||||
}
|
||||
clauses, args := []string{"1=1"}, []any{}
|
||||
if filter.Status != "" {
|
||||
clauses = append(clauses, "status = ?")
|
||||
args = append(args, filter.Status)
|
||||
}
|
||||
if filter.Keyword != "" {
|
||||
escaped := strings.NewReplacer("\\", "\\\\", "%", "\\%", "_", "\\_").Replace(filter.Keyword)
|
||||
clauses = append(clauses, "(title LIKE ? ESCAPE '\\' OR goods_id LIKE ? ESCAPE '\\')")
|
||||
args = append(args, "%"+escaped+"%", "%"+escaped+"%")
|
||||
}
|
||||
if !from.IsZero() {
|
||||
clauses = append(clauses, "julianday(created_at) >= julianday(?)")
|
||||
args = append(args, from.Format(time.RFC3339Nano))
|
||||
}
|
||||
if !to.IsZero() {
|
||||
clauses = append(clauses, "julianday(created_at) < julianday(?)")
|
||||
args = append(args, to.Format(time.RFC3339Nano))
|
||||
}
|
||||
rows, err := store.database.QueryContext(ctx, "SELECT id,title,goods_id,sku_color,sku_size,quantity,max_total_price,status,version,created_at FROM tasks WHERE "+strings.Join(clauses, " AND ")+" ORDER BY julianday(created_at) DESC,rowid DESC", args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
result := []TaskRow{}
|
||||
for rows.Next() {
|
||||
var item TaskRow
|
||||
var created string
|
||||
if err := rows.Scan(&item.ID, &item.Title, &item.GoodsID, &item.SKUColor, &item.SKUSize, &item.Quantity, &item.MaxTotalPrice, &item.Status, &item.Version, &created); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
item.CreatedAt, err = time.Parse(time.RFC3339Nano, created)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, rows.Err()
|
||||
}
|
||||
|
||||
// ValidateTaskFilter 返回可关联到字段的错误，使服务端页面拒绝篡改参数时仍能保留输入值。
|
||||
func ValidateTaskFilter(filter TaskFilter) Errors {
|
||||
validation := Errors{}
|
||||
if !validTaskStatus(filter.Status) {
|
||||
validation["status"] = "请选择有效的任务状态。"
|
||||
}
|
||||
location, err := time.LoadLocation("Asia/Shanghai")
|
||||
if err != nil {
|
||||
validation["created_from"] = "日期筛选暂不可用，请稍后重试。"
|
||||
validation["created_to"] = "日期筛选暂不可用，请稍后重试。"
|
||||
return validation
|
||||
}
|
||||
parseDate := func(field, value string) (time.Time, bool) {
|
||||
if value == "" {
|
||||
return time.Time{}, true
|
||||
}
|
||||
parsed, parseErr := time.ParseInLocation("2006-01-02", value, location)
|
||||
if parseErr != nil {
|
||||
validation[field] = "请输入有效日期。"
|
||||
return time.Time{}, false
|
||||
}
|
||||
return parsed, true
|
||||
}
|
||||
from, fromOK := parseDate("created_from", filter.CreatedFrom)
|
||||
to, toOK := parseDate("created_to", filter.CreatedTo)
|
||||
if fromOK && toOK && !from.IsZero() && !to.IsZero() && from.After(to) {
|
||||
validation["created_to"] = "结束日期不能早于开始日期。"
|
||||
}
|
||||
return validation
|
||||
}
|
||||
|
||||
func (store *SQLiteStore) StartPurchases(ctx context.Context, command StartCommand, createdBy string) (StartResult, error) {
|
||||
items, err := startItems(command)
|
||||
if err != nil || createdBy == "" {
|
||||
return StartResult{}, ErrInvalidStart
|
||||
}
|
||||
if store.policy.AuthorizationTTL <= 0 || store.policy.MaxQuantity <= 0 {
|
||||
return StartResult{}, ErrInvalidStart
|
||||
}
|
||||
_, ceiling, ok := normalizeCents(store.policy.MaxTotalPrice)
|
||||
if !ok {
|
||||
return StartResult{}, ErrInvalidStart
|
||||
}
|
||||
writeCtx, cancel := context.WithTimeout(ctx, sqliteWriteTimeout)
|
||||
defer cancel()
|
||||
select {
|
||||
case store.writeGate <- struct{}{}:
|
||||
defer func() { <-store.writeGate }()
|
||||
case <-writeCtx.Done():
|
||||
return StartResult{}, writeCtx.Err()
|
||||
}
|
||||
tx, err := store.database.BeginTx(writeCtx, nil)
|
||||
if err != nil {
|
||||
return StartResult{}, err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
// Replay precedes any DRAFT check. One service process serializes this check with creation; SQLite uniqueness remains the cross-transaction backstop.
|
||||
result, found, err := replayStart(writeCtx, tx, command.StartKey, items)
|
||||
if err != nil {
|
||||
return StartResult{}, err
|
||||
}
|
||||
if found {
|
||||
if err := tx.Commit(); err != nil {
|
||||
return StartResult{}, err
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
now := store.now().UTC()
|
||||
expires := now.Add(store.policy.AuthorizationTTL)
|
||||
result = StartResult{StartKey: command.StartKey, AuthorizedCount: len(items), Tasks: make([]AuthorizedTask, 0, len(items)), PaymentAutomated: false}
|
||||
for _, item := range items {
|
||||
var title, goods, color, size, price, status string
|
||||
var quantity, version int
|
||||
if err := tx.QueryRowContext(writeCtx, "SELECT title,goods_id,sku_color,sku_size,quantity,max_total_price,status,version FROM tasks WHERE id=?", item.TaskID).Scan(&title, &goods, &color, &size, &quantity, &price, &status, &version); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return StartResult{}, ErrStartConflict
|
||||
}
|
||||
return StartResult{}, err
|
||||
}
|
||||
if status != "DRAFT" || version != item.ExpectedTaskVersion || !goodsIDValid(goods) || color == "" || size == "" || quantity < 1 || quantity > store.policy.MaxQuantity {
|
||||
return StartResult{}, ErrStartConflict
|
||||
}
|
||||
canonical, cents, ok := normalizeCents(price)
|
||||
if !ok || canonical != price || cents.Cmp(ceiling) > 0 {
|
||||
return StartResult{}, ErrStartConflict
|
||||
}
|
||||
if _, err := domain.TransitionTask(domain.TaskStatusDraft, domain.TaskStatusPending); err != nil {
|
||||
return StartResult{}, err
|
||||
}
|
||||
id, err := NewCreateKey()
|
||||
if err != nil {
|
||||
return StartResult{}, err
|
||||
}
|
||||
next := version + 1
|
||||
if _, err = tx.ExecContext(writeCtx, "INSERT INTO order_authorizations (id,task_id,task_version,start_key,goods_id,sku_color,sku_size,quantity,total_price_cap,status,created_by,created_at,expires_at) VALUES (?,?,?,?,?,?,?,?,?,'ACTIVE',?,?,?)", id, item.TaskID, next, command.StartKey, goods, color, size, quantity, price, createdBy, now.Format(time.RFC3339Nano), expires.Format(time.RFC3339Nano)); err != nil {
|
||||
return StartResult{}, err
|
||||
}
|
||||
updated, err := tx.ExecContext(writeCtx, "UPDATE tasks SET status='PENDING',version=version+1,updated_at=? WHERE id=? AND status='DRAFT' AND version=?", now.Format(time.RFC3339Nano), item.TaskID, version)
|
||||
if err != nil {
|
||||
return StartResult{}, err
|
||||
}
|
||||
affected, err := updated.RowsAffected()
|
||||
if err != nil {
|
||||
return StartResult{}, err
|
||||
}
|
||||
if affected != 1 {
|
||||
return StartResult{}, ErrStartConflict
|
||||
}
|
||||
result.Tasks = append(result.Tasks, AuthorizedTask{TaskID: item.TaskID, TaskVersion: next, AuthorizationID: id, ExpiresAt: expires})
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return StartResult{}, err
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func goodsIDValid(value string) bool {
|
||||
if value == "" {
|
||||
return false
|
||||
}
|
||||
for _, ch := range value {
|
||||
if ch < '0' || ch > '9' {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func replayStart(ctx context.Context, tx *sql.Tx, startKey string, items []StartItem) (StartResult, bool, error) {
|
||||
rows, err := tx.QueryContext(ctx, "SELECT id,task_id,task_version,expires_at FROM order_authorizations WHERE start_key=? ORDER BY task_id", startKey)
|
||||
if err != nil {
|
||||
return StartResult{}, false, err
|
||||
}
|
||||
defer rows.Close()
|
||||
result := StartResult{StartKey: startKey, PaymentAutomated: false}
|
||||
for rows.Next() {
|
||||
var item AuthorizedTask
|
||||
var expires string
|
||||
if err := rows.Scan(&item.AuthorizationID, &item.TaskID, &item.TaskVersion, &expires); err != nil {
|
||||
return StartResult{}, false, err
|
||||
}
|
||||
item.ExpiresAt, err = time.Parse(time.RFC3339Nano, expires)
|
||||
if err != nil {
|
||||
return StartResult{}, false, err
|
||||
}
|
||||
result.Tasks = append(result.Tasks, item)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return StartResult{}, false, err
|
||||
}
|
||||
if len(result.Tasks) == 0 {
|
||||
return StartResult{}, false, nil
|
||||
}
|
||||
if len(result.Tasks) != len(items) {
|
||||
return StartResult{}, false, ErrStartConflict
|
||||
}
|
||||
for i := range items {
|
||||
if result.Tasks[i].TaskID != items[i].TaskID || result.Tasks[i].TaskVersion-1 != items[i].ExpectedTaskVersion {
|
||||
return StartResult{}, false, ErrStartConflict
|
||||
}
|
||||
}
|
||||
result.AuthorizedCount = len(result.Tasks)
|
||||
return result, true, nil
|
||||
}
|
||||
@@ -0,0 +1,195 @@
|
||||
package tasks
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestListTasksTreatsLikeMetacharactersLiterally(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store, err := NewSQLiteStore(database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
created := "2026-08-04T01:00:00Z"
|
||||
insertTaskRow(t, database, "percent", "100%纯棉", "100", "DRAFT", created)
|
||||
insertTaskRow(t, database, "underscore", "尺码_A", "101", "DRAFT", created)
|
||||
insertTaskRow(t, database, "backslash", `路径\名称`, "102", "DRAFT", created)
|
||||
insertTaskRow(t, database, "plain", "普通商品", "103", "DRAFT", created)
|
||||
|
||||
for _, test := range []struct {
|
||||
keyword string
|
||||
wantID string
|
||||
}{
|
||||
{keyword: "%", wantID: "percent"},
|
||||
{keyword: "_", wantID: "underscore"},
|
||||
{keyword: `\`, wantID: "backslash"},
|
||||
} {
|
||||
t.Run(test.wantID, func(t *testing.T) {
|
||||
rows, err := store.ListTasks(context.Background(), TaskFilter{Keyword: test.keyword})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 1 || rows[0].ID != test.wantID {
|
||||
t.Fatalf("keyword %q rows = %#v, want only %q", test.keyword, rows, test.wantID)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestListTasksSupportsEveryStatusAndEmptyMeansAll(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store, err := NewSQLiteStore(database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
statuses := []string{"DRAFT", "PENDING", "CLAIMED", "ORDERING", "NEEDS_MANUAL", "WAITING_PAYMENT", "RECONCILIATION_REQUIRED", "SUCCEEDED", "FAILED", "CANCELED"}
|
||||
for index, status := range statuses {
|
||||
insertTaskRow(t, database, status, status, "200", status, time.Date(2026, 8, 4, 1, 0, index, 0, time.UTC).Format(time.RFC3339Nano))
|
||||
}
|
||||
|
||||
all, err := store.ListTasks(context.Background(), TaskFilter{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(all) != len(statuses) {
|
||||
t.Fatalf("all-status rows = %d, want %d", len(all), len(statuses))
|
||||
}
|
||||
for _, status := range statuses {
|
||||
rows, err := store.ListTasks(context.Background(), TaskFilter{Status: status})
|
||||
if err != nil {
|
||||
t.Fatalf("status %s: %v", status, err)
|
||||
}
|
||||
if len(rows) != 1 || rows[0].Status != status {
|
||||
t.Fatalf("status %s rows = %#v", status, rows)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestListTasksUsesShanghaiHalfOpenDateRange(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store, err := NewSQLiteStore(database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
insertTaskRow(t, database, "before", "before", "300", "DRAFT", "2026-08-03T15:59:59Z")
|
||||
insertTaskRow(t, database, "at-start", "at-start", "301", "DRAFT", "2026-08-03T16:00:00Z")
|
||||
insertTaskRow(t, database, "before-end", "before-end", "302", "DRAFT", "2026-08-04T15:59:59Z")
|
||||
insertTaskRow(t, database, "at-end", "at-end", "303", "DRAFT", "2026-08-04T16:00:00Z")
|
||||
|
||||
rows, err := store.ListTasks(context.Background(), TaskFilter{CreatedFrom: "2026-08-04", CreatedTo: "2026-08-04"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 2 || rows[0].ID != "before-end" || rows[1].ID != "at-start" {
|
||||
t.Fatalf("Shanghai day rows = %#v, want [before-end at-start]", rows)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListTasksBreaksEqualTimestampsByDescendingRowID(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store, err := NewSQLiteStore(database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
created := "2026-08-04T01:02:03Z"
|
||||
insertTaskRow(t, database, "first", "first", "400", "DRAFT", created)
|
||||
insertTaskRow(t, database, "second", "second", "401", "DRAFT", created)
|
||||
|
||||
rows, err := store.ListTasks(context.Background(), TaskFilter{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 2 || rows[0].ID != "second" || rows[1].ID != "first" {
|
||||
t.Fatalf("equal-time rows = %#v, want descending rowid", rows)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListTasksRejectsInvalidStatusAndDates(t *testing.T) {
|
||||
store, err := NewSQLiteStore(migratedDatabase(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for name, filter := range map[string]TaskFilter{
|
||||
"status": {Status: "UNKNOWN"},
|
||||
"from date": {CreatedFrom: "2026-02-30"},
|
||||
"to date": {CreatedTo: "04/08/2026"},
|
||||
"reverse range": {CreatedFrom: "2026-08-05", CreatedTo: "2026-08-04"},
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
rows, err := store.ListTasks(context.Background(), filter)
|
||||
if !errors.Is(err, ErrInvalidFilter) || rows != nil {
|
||||
t.Fatalf("ListTasks(%#v) = (%#v, %v), want ErrInvalidFilter", filter, rows, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartPurchasesIsAtomicAndReplaysSameSet(t *testing.T) {
|
||||
database := migratedDatabase(t)
|
||||
store, err := NewSQLiteStore(database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store.SetStartPolicy(StartPolicy{AuthorizationTTL: time.Hour, MaxQuantity: 10, MaxTotalPrice: "999.99"})
|
||||
store.now = func() time.Time { return time.Date(2026, 8, 4, 1, 2, 3, 0, time.UTC) }
|
||||
for _, draft := range []Draft{testDraft(testKey, "one"), testDraft("b3c9f507-7473-4fa6-8d71-8786c34c6301", "two")} {
|
||||
if _, err := store.CreateDraft(context.Background(), draft); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
command := StartCommand{StartKey: "c3c9f507-7473-4fa6-8d71-8786c34c6301", Tasks: []StartItem{{TaskID: "b3c9f507-7473-4fa6-8d71-8786c34c6301", ExpectedTaskVersion: 1}, {TaskID: testKey, ExpectedTaskVersion: 1}}}
|
||||
first, err := store.StartPurchases(context.Background(), command, "admin")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if first.AuthorizedCount != 2 || first.PaymentAutomated {
|
||||
t.Fatalf("start result=%#v", first)
|
||||
}
|
||||
command.Tasks[0], command.Tasks[1] = command.Tasks[1], command.Tasks[0]
|
||||
replay, err := store.StartPurchases(context.Background(), command, "admin")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if replay.Tasks[0].AuthorizationID != first.Tasks[0].AuthorizationID || replay.Tasks[1].AuthorizationID != first.Tasks[1].AuthorizationID {
|
||||
t.Fatalf("replay=%#v first=%#v", replay, first)
|
||||
}
|
||||
var pending, auths int
|
||||
if err := database.QueryRow(`SELECT COUNT(*) FROM tasks WHERE status='PENDING' AND version=2`).Scan(&pending); err != nil || pending != 2 {
|
||||
t.Fatalf("pending=%d err=%v", pending, err)
|
||||
}
|
||||
if err := database.QueryRow(`SELECT COUNT(*) FROM order_authorizations WHERE status='ACTIVE' AND created_by='admin'`).Scan(&auths); err != nil || auths != 2 {
|
||||
t.Fatalf("auths=%d err=%v", auths, err)
|
||||
}
|
||||
_, err = store.StartPurchases(context.Background(), StartCommand{StartKey: command.StartKey, Tasks: command.Tasks[:1]}, "admin")
|
||||
if !errors.Is(err, ErrStartConflict) {
|
||||
t.Fatalf("subset err=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestShanghaiRangeAndMoneyAreFailClosed(t *testing.T) {
|
||||
start, end, err := ShanghaiRange("2026-08-04", "2026-08-04")
|
||||
if err != nil || start.Format(time.RFC3339) != "2026-08-03T16:00:00Z" || end.Format(time.RFC3339) != "2026-08-04T16:00:00Z" {
|
||||
t.Fatalf("range=(%s,%s,%v)", start, end, err)
|
||||
}
|
||||
for _, value := range []string{"0.01", "12.80", "999999999999999999999999.99"} {
|
||||
if _, _, ok := normalizeCents(value); !ok {
|
||||
t.Fatalf("money %q rejected", value)
|
||||
}
|
||||
}
|
||||
for _, value := range []string{"1", "01.20", "0.00", "1.234", "1.", " 1.00", "1e2"} {
|
||||
if _, _, ok := normalizeCents(value); ok {
|
||||
t.Fatalf("money %q accepted", value)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func insertTaskRow(t *testing.T, database *sql.DB, id, title, goodsID, status, createdAt string) {
|
||||
t.Helper()
|
||||
if _, err := database.Exec(`INSERT INTO tasks (id, source, title, goods_id, sku_color, sku_size, quantity, max_total_price, status, version, created_at, updated_at) VALUES (?, 'MANUAL', ?, ?, '黑色', 'M', 2, '12.80', ?, 1, ?, ?)`, id, title, goodsID, status, createdAt, createdAt); err != nil {
|
||||
t.Fatalf("insert task %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
@@ -13,21 +13,27 @@ const sqliteWriteTimeout = 2 * time.Second
|
||||
type Store interface {
|
||||
CreateDraft(context.Context, Draft) (Draft, error)
|
||||
ListDrafts(context.Context) ([]Draft, error)
|
||||
ListTasks(context.Context, TaskFilter) ([]TaskRow, error)
|
||||
StartPurchases(context.Context, StartCommand, string) (StartResult, error)
|
||||
}
|
||||
type SQLiteStore struct {
|
||||
database *sql.DB
|
||||
now func() time.Time
|
||||
createGate chan struct{}
|
||||
database *sql.DB
|
||||
now func() time.Time
|
||||
writeGate chan struct{}
|
||||
policy StartPolicy
|
||||
}
|
||||
|
||||
func NewSQLiteStore(database *sql.DB) (*SQLiteStore, error) {
|
||||
if database == nil {
|
||||
return nil, errors.New("database is required")
|
||||
}
|
||||
if _, err := database.Exec("SELECT 1 FROM tasks LIMIT 1"); err != nil {
|
||||
if _, err := database.Exec("SELECT task_version, start_key, total_price_cap FROM order_authorizations LIMIT 1"); err != nil {
|
||||
return nil, fmt.Errorf("tasks migration is not available: %w", err)
|
||||
}
|
||||
return &SQLiteStore{database: database, now: time.Now, createGate: make(chan struct{}, 1)}, nil
|
||||
if _, err := database.Exec("SELECT 1 FROM purchase_attempts LIMIT 1"); err != nil {
|
||||
return nil, fmt.Errorf("single-pass migration is not available: %w", err)
|
||||
}
|
||||
return &SQLiteStore{database: database, now: time.Now, writeGate: make(chan struct{}, 1)}, nil
|
||||
}
|
||||
|
||||
func (store *SQLiteStore) CreateDraft(ctx context.Context, draft Draft) (Draft, error) {
|
||||
@@ -36,8 +42,8 @@ func (store *SQLiteStore) CreateDraft(ctx context.Context, draft Draft) (Draft,
|
||||
// SQLite permits one writer at a time. Serializing this store's short create
|
||||
// transaction prevents concurrent retries of one create key from surfacing as busy.
|
||||
select {
|
||||
case store.createGate <- struct{}{}:
|
||||
defer func() { <-store.createGate }()
|
||||
case store.writeGate <- struct{}{}:
|
||||
defer func() { <-store.writeGate }()
|
||||
case <-writeContext.Done():
|
||||
return Draft{}, writeContext.Err()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user