cmbuyer/admin/internal/tasks/start_purchases_store_test.go

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)
	}
}