Merge persistent custom order backend for issue #575

This commit is contained in:
sagitchu
2026-11-10 08:21:21 +08:00
9 changed files with 726 additions and 32 deletions
@@ -0,0 +1,45 @@
package handler
import (
"errors"
"go-backend/internal/http/response"
"go-backend/internal/store/repo"
"net/http"
)
func (h *Handler) userDisplayOrder(w http.ResponseWriter, r *http.Request) {
updateDisplayOrder(w, r, h.repo.UpdateUserDisplayOrder)
}
func (h *Handler) speedLimitDisplayOrder(w http.ResponseWriter, r *http.Request) {
updateDisplayOrder(w, r, h.repo.UpdateSpeedLimitDisplayOrder)
}
func updateDisplayOrder(w http.ResponseWriter, r *http.Request, save func([]int64, []int64) error) {
if r.Method != http.MethodPost {
response.WriteJSON(w, response.Err(400, "请求方法错误"))
return
}
var req struct {
IDs []int64 `json:"ids"`
ExpectedIDs []int64 `json:"expectedIds"`
}
if err := decodeJSON(r.Body, &req); err != nil {
response.WriteJSON(w, response.Err(400, "请求参数错误"))
return
}
if err := save(req.IDs, req.ExpectedIDs); err != nil {
code := 500
if errors.Is(err, repo.ErrInvalidDisplayOrder) {
code = 400
}
if errors.Is(err, repo.ErrDisplayOrderConflict) {
code = 409
}
msg := "排序保存失败,请刷新后重试"
if code != 500 {
msg = err.Error()
}
response.WriteJSON(w, response.Err(code, msg))
return
}
response.WriteJSON(w, response.OKEmpty())
}
@@ -248,6 +248,8 @@ func (h *Handler) Register(mux *http.ServeMux) {
mux.HandleFunc("/api/v1/forward/batch-resume", h.forwardBatchResume)
mux.HandleFunc("/api/v1/forward/batch-redeploy", h.forwardBatchRedeploy)
mux.HandleFunc("/api/v1/forward/batch-change-tunnel", h.forwardBatchChangeTunnel)
mux.HandleFunc("/api/v1/user/update-order", h.userDisplayOrder)
mux.HandleFunc("/api/v1/speed-limit/update-order", h.speedLimitDisplayOrder)
mux.HandleFunc("/api/v1/speed-limit/list", h.speedLimitList)
mux.HandleFunc("/api/v1/speed-limit/create", h.speedLimitCreate)
mux.HandleFunc("/api/v1/speed-limit/update", h.speedLimitUpdate)
+1 -1
View File
@@ -180,7 +180,7 @@ func requiresAdmin(path string) bool {
}
switch path {
case "/api/v1/user/create", "/api/v1/user/list", "/api/v1/user/update", "/api/v1/user/delete", "/api/v1/user/reset":
case "/api/v1/user/update-order", "/api/v1/user/create", "/api/v1/user/list", "/api/v1/user/update", "/api/v1/user/delete", "/api/v1/user/reset":
return true
case "/api/v1/config/update", "/api/v1/config/update-single":
return true
+20 -16
View File
@@ -10,6 +10,7 @@ import "database/sql"
// User maps to the "user" table. PostgreSQL treats "user" as a reserved
// word, so TableName() is required for correct quoting.
type User struct {
DisplayOrder int64 `gorm:"not null;default:0"`
ID int64 `gorm:"primaryKey;autoIncrement"`
User string `gorm:"column:user;type:varchar(100);not null"`
Pwd string `gorm:"type:varchar(100);not null"`
@@ -151,14 +152,15 @@ type NftCounterState struct {
func (NftCounterState) TableName() string { return "nft_counter_state" }
type SpeedLimit struct {
ID int64 `gorm:"primaryKey;autoIncrement"`
Name string `gorm:"type:varchar(100);not null"`
Speed int `gorm:"not null"`
TunnelID sql.NullInt64 `gorm:"column:tunnel_id"`
TunnelName sql.NullString `gorm:"column:tunnel_name;type:varchar(100)"`
CreatedTime int64 `gorm:"column:created_time;not null"`
UpdatedTime sql.NullInt64 `gorm:"column:updated_time"`
Status int `gorm:"not null"`
DisplayOrder int64 `gorm:"not null;default:0"`
ID int64 `gorm:"primaryKey;autoIncrement"`
Name string `gorm:"type:varchar(100);not null"`
Speed int `gorm:"not null"`
TunnelID sql.NullInt64 `gorm:"column:tunnel_id"`
TunnelName sql.NullString `gorm:"column:tunnel_name;type:varchar(100)"`
CreatedTime int64 `gorm:"column:created_time;not null"`
UpdatedTime sql.NullInt64 `gorm:"column:updated_time"`
Status int `gorm:"not null"`
}
func (SpeedLimit) TableName() string { return "speed_limit" }
@@ -417,6 +419,7 @@ type BackupData struct {
}
type UserBackup struct {
DisplayOrder *int64 `json:"displayOrder,omitempty"`
ID int64 `json:"id"`
User string `json:"user"`
Pwd string `json:"pwd"`
@@ -540,14 +543,15 @@ type UserTunnelBackup struct {
}
type SpeedLimitBackup struct {
ID int64 `json:"id"`
Name string `json:"name"`
Speed int64 `json:"speed"`
TunnelID *int64 `json:"tunnelId,omitempty"`
TunnelName string `json:"tunnelName,omitempty"`
CreatedTime int64 `json:"createdTime"`
UpdatedTime int64 `json:"updatedTime,omitempty"`
Status int `json:"status"`
DisplayOrder *int64 `json:"displayOrder,omitempty"`
ID int64 `json:"id"`
Name string `json:"name"`
Speed int64 `json:"speed"`
TunnelID *int64 `json:"tunnelId,omitempty"`
TunnelName string `json:"tunnelName,omitempty"`
CreatedTime int64 `json:"createdTime"`
UpdatedTime int64 `json:"updatedTime,omitempty"`
Status int `json:"status"`
}
type TunnelGroupBackup struct {
+36 -13
View File
@@ -914,7 +914,7 @@ func (r *Repository) ListUsers() ([]map[string]interface{}, error) {
return nil, errors.New("repository not initialized")
}
var users []model.User
if err := r.db.Where("role_id != ?", 0).Order("id DESC").Find(&users).Error; err != nil {
if err := r.db.Where("role_id != ?", 0).Order("display_order ASC, id DESC").Find(&users).Error; err != nil {
return nil, err
}
userIDs := make([]int64, 0, len(users))
@@ -954,7 +954,7 @@ func (r *Repository) ListSpeedLimits() ([]map[string]interface{}, error) {
return nil, errors.New("repository not initialized")
}
var limits []model.SpeedLimit
if err := r.db.Order("id DESC").Find(&limits).Error; err != nil {
if err := r.db.Order("display_order ASC, id DESC").Find(&limits).Error; err != nil {
return nil, err
}
items := make([]map[string]interface{}, 0, len(limits))
@@ -2093,7 +2093,8 @@ func (r *Repository) exportUsers() ([]model.UserBackup, error) {
out := make([]model.UserBackup, 0, len(users))
for _, u := range users {
b := model.UserBackup{
ID: u.ID, User: u.User, Pwd: u.Pwd, RoleID: u.RoleID,
DisplayOrder: &u.DisplayOrder,
ID: u.ID, User: u.User, Pwd: u.Pwd, RoleID: u.RoleID,
ExpTime: u.ExpTime, Flow: u.Flow, FlowMiB: u.FlowMiB, InFlow: u.InFlow, OutFlow: u.OutFlow,
FlowResetTime: u.FlowResetTime, Num: u.Num,
CreatedTime: u.CreatedTime, Status: u.Status,
@@ -2290,7 +2291,8 @@ func (r *Repository) exportSpeedLimits() ([]model.SpeedLimitBackup, error) {
out := make([]model.SpeedLimitBackup, 0, len(sls))
for _, sl := range sls {
b := model.SpeedLimitBackup{
ID: sl.ID, Name: sl.Name, Speed: int64(sl.Speed),
DisplayOrder: &sl.DisplayOrder,
ID: sl.ID, Name: sl.Name, Speed: int64(sl.Speed),
CreatedTime: sl.CreatedTime, Status: sl.Status,
}
if sl.UpdatedTime.Valid {
@@ -2374,6 +2376,17 @@ func (r *Repository) Import(backup *model.BackupData, types []string) (*model.Im
}
err := r.db.Transaction(func(tx *gorm.DB) error {
// Lock in a stable order before touching either membership.
if typeSet["users"] {
if err := lockDisplayOrder(tx, "user"); err != nil {
return err
}
}
if typeSet["speedLimits"] {
if err := lockDisplayOrder(tx, "speed_limit"); err != nil {
return err
}
}
now := unixMilliNow()
if typeSet["users"] && len(backup.Users) > 0 {
@@ -2478,12 +2491,17 @@ func importUsers(tx *gorm.DB, users []model.UserBackup, now int64) (int, error)
Status: status,
PasswordChangedAt: now,
}
// An omitted old-backup rank preserves existing order on conflict.
// New legacy records use rank zero and ID descending, matching upgrades.
columns := []string{"user", "pwd", "role_id", "exp_time", "flow", "flow_mib", "in_flow", "out_flow",
"flow_reset_time", "num", "updated_time", "status", "password_changed_at"}
if u.DisplayOrder != nil {
item.DisplayOrder = *u.DisplayOrder
columns = append(columns, "display_order")
}
err = tx.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns([]string{
"user", "pwd", "role_id", "exp_time", "flow", "flow_mib", "in_flow", "out_flow",
"flow_reset_time", "num", "updated_time", "status", "password_changed_at",
}),
Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns(columns),
}).Create(&item).Error
if err != nil {
return count, err
@@ -2766,11 +2784,16 @@ func importSpeedLimits(tx *gorm.DB, speedLimits []model.SpeedLimitBackup, now in
UpdatedTime: sql.NullInt64{Int64: now, Valid: true},
Status: sl.Status,
}
// An omitted old-backup rank preserves existing order on conflict.
// New legacy records use rank zero and ID descending, matching upgrades.
columns := []string{"name", "speed", "tunnel_id", "tunnel_name", "updated_time", "status"}
if sl.DisplayOrder != nil {
item.DisplayOrder = *sl.DisplayOrder
columns = append(columns, "display_order")
}
err := tx.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns([]string{
"name", "speed", "tunnel_id", "tunnel_name", "updated_time", "status",
}),
Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns(columns),
}).Create(&item).Error
if err != nil {
return count, err
@@ -0,0 +1,109 @@
package repo
import (
"errors"
"gorm.io/gorm"
"slices"
)
var ErrDisplayOrderConflict = errors.New("列表已变更,请刷新后重试")
var ErrInvalidDisplayOrder = errors.New("排序必须包含全部条目且不能重复")
// Lock the entire membership, including an empty list and concurrent INSERTs.
// PostgreSQL row locks alone cannot protect the missing-row/phantom case.
// SQLite's first write acquires its writer reservation before reading the list.
func lockDisplayOrder(tx *gorm.DB, table string) error {
if tx.Dialector.Name() == "postgres" {
return tx.Exec("LOCK TABLE \"" + table + "\" IN SHARE ROW EXCLUSIVE MODE").Error
}
return tx.Table(table).Where("1 = 0").UpdateColumn("display_order", gorm.Expr("display_order")).Error
}
func (r *Repository) UpdateUserDisplayOrder(ids, expected []int64) error {
return r.updateDisplayOrder("user", ids, expected)
}
func (r *Repository) UpdateSpeedLimitDisplayOrder(ids, expected []int64) error {
return r.updateDisplayOrder("speed_limit", ids, expected)
}
func (r *Repository) updateDisplayOrder(table string, ids, expected []int64) error {
if ids == nil || expected == nil {
return ErrInvalidDisplayOrder
}
seen := map[int64]bool{}
for _, id := range ids {
if id <= 0 || seen[id] {
return ErrInvalidDisplayOrder
}
seen[id] = true
}
seenExpected := map[int64]bool{}
for _, id := range expected {
if id <= 0 || seenExpected[id] {
return ErrInvalidDisplayOrder
}
seenExpected[id] = true
}
if len(ids) != len(expected) {
return ErrInvalidDisplayOrder
}
for id := range seen {
if !seenExpected[id] {
return ErrInvalidDisplayOrder
}
}
return r.db.Transaction(func(tx *gorm.DB) error {
if err := lockDisplayOrder(tx, table); err != nil {
return err
}
q := tx.Table(table)
if table == "user" {
var admins int64
if err := tx.Table(table).Where("role_id = ? AND id IN ?", 0, ids).Count(&admins).Error; err != nil {
return err
}
if admins != 0 {
return ErrInvalidDisplayOrder
}
q = q.Where("role_id != ?", 0)
}
var current []int64
if err := q.Order("display_order ASC, id DESC").Pluck("id", &current).Error; err != nil {
return err
}
if !slices.Equal(current, expected) {
return ErrDisplayOrderConflict
}
for i, id := range ids {
result := tx.Table(table).Where("id = ?", id).UpdateColumn("display_order", int64(i))
if result.Error != nil {
return result.Error
}
if result.RowsAffected != 1 {
return ErrDisplayOrderConflict
}
}
return nil
})
}
// New items precede all saved ranks. Zero is only the legacy migration default.
func createFirst(tx *gorm.DB, table string, item interface{}, rank *int64) error {
return tx.Transaction(func(tx *gorm.DB) error {
if err := lockDisplayOrder(tx, table); err != nil {
return err
}
q := tx.Table(table)
if table == "user" {
q = q.Where("role_id != ?", 0)
}
var first int64
if err := q.Select("COALESCE(MIN(display_order), 0)").Scan(&first).Error; err != nil {
return err
}
if first == -1<<63 {
return errors.New("display order exhausted")
}
*rank = first - 1
return tx.Create(item).Error
})
}
@@ -65,7 +65,7 @@ func (r *Repository) CreateUser(username, pwdHash string, roleID int, expTime, f
Status: status,
PasswordChangedAt: now,
}
if err := r.db.Create(&user).Error; err != nil {
if err := createFirst(r.db, "user", &user, &user.DisplayOrder); err != nil {
return 0, err
}
return user.ID, nil
@@ -940,7 +940,7 @@ func (r *Repository) CreateSpeedLimit(name string, speed int, now int64, status
UpdatedTime: sql.NullInt64{Int64: now, Valid: true},
Status: status,
}
if err := r.db.Create(&sl).Error; err != nil {
if err := createFirst(r.db, "speed_limit", &sl, &sl.DisplayOrder); err != nil {
return 0, err
}
return sl.ID, nil
@@ -0,0 +1,144 @@
package contract_test
import (
"bytes"
"encoding/json"
"fmt"
"go-backend/internal/auth"
"go-backend/internal/http/response"
"net/http"
"net/http/httptest"
"reflect"
"testing"
)
func orderRequest(t *testing.T, router http.Handler, token, resource, body string) response.R {
t.Helper()
req := httptest.NewRequest(http.MethodPost, "/api/v1/"+resource+"/update-order", bytes.NewBufferString(body))
req.Header.Set("Authorization", token)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
var out response.R
if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil {
t.Fatalf("decode: %v (%s)", err, w.Body.String())
}
return out
}
func TestDisplayOrderUserContract(t *testing.T) {
router, r := setupContractRouter(t, "order-secret")
token := mustAdminToken(t, "order-secret")
a, err := r.CreateUser("order-a", "hash", 1, 0, 10, 1, 2, 1, 3, 0)
if err != nil {
t.Fatal(err)
}
b, err := r.CreateUser("order-b", "hash", 1, 0, 20, 1, 2, 1, 3, 0)
if err != nil {
t.Fatal(err)
}
out := orderRequest(t, router, token, "user", fmt.Sprintf(`{"ids":[%d,%d],"expectedIds":[%d,%d]}`, a, b, b, a))
if out.Code != 0 {
t.Fatalf("reorder: %+v", out)
}
users, err := r.ListUsers()
if err != nil {
t.Fatal(err)
}
if got := []int64{users[0]["id"].(int64), users[1]["id"].(int64)}; !reflect.DeepEqual(got, []int64{a, b}) {
t.Fatalf("order %v", got)
}
}
func TestDisplayOrderValidationAndLifecycle(t *testing.T) {
for _, resource := range []string{"user", "speed-limit"} {
t.Run(resource, func(t *testing.T) {
router, r := setupContractRouter(t, "order-secret")
token := mustAdminToken(t, "order-secret")
create := func(name string) int64 {
t.Helper()
var id int64
var err error
if resource == "user" {
id, err = r.CreateUser(name, "hash", 1, 0, 10, 1, 2, 1, 3, 0)
} else {
id, err = r.CreateSpeedLimit(name, 100, 0, 1)
}
if err != nil {
t.Fatal(err)
}
return id
}
list := func() []int64 {
t.Helper()
out := postBatchRequest(t, router, token, "/api/v1/"+resource+"/list", "{}")
rows, ok := out.Data.([]interface{})
if !ok {
t.Fatalf("list data: %T", out.Data)
}
ids := []int64{}
for _, row := range rows {
ids = append(ids, int64(row.(map[string]interface{})["id"].(float64)))
}
return ids
}
a, b := create("a"), create("b")
body := func(ids, expected []int64) string {
v, _ := json.Marshal(map[string]interface{}{"ids": ids, "expectedIds": expected})
return string(v)
}
check := func(payload string, code int) {
t.Helper()
out := orderRequest(t, router, token, resource, payload)
if out.Code != code {
t.Fatalf("want %d got %+v", code, out)
}
}
check(body([]int64{a, b}, []int64{b, a}), 0)
for _, payload := range []string{`{}`, `{"ids":null,"expectedIds":[]}`, `{"ids":[1.1],"expectedIds":[1]}`, body([]int64{a, a}, []int64{a, b}), body([]int64{a}, []int64{a, b}), body([]int64{a, b, 99999}, []int64{a, b}), body([]int64{a, 99999}, []int64{a, b}), body([]int64{a, b}, []int64{a, a})} {
check(payload, 400)
}
if resource == "user" {
check(body([]int64{1, a, b}, []int64{a, b}), 400)
check(body([]int64{1, a, b}, []int64{1, a, b}), 400)
}
check(body([]int64{b, a}, []int64{b, a}), 409)
c := create("c")
if got := list(); !reflect.DeepEqual(got, []int64{c, a, b}) {
t.Fatalf("new item first: %v", got)
}
check(body([]int64{b, a}, []int64{a, b}), 409)
if resource == "user" {
if err := r.UpdateUserWithoutPassword(a, "edited", 10, 2, 0, 1, 1, 3, 0); err != nil {
t.Fatal(err)
}
} else {
if err := r.UpdateSpeedLimit(a, "edited", 200, 1, 0); err != nil {
t.Fatal(err)
}
}
if got := list(); !reflect.DeepEqual(got, []int64{c, a, b}) {
t.Fatalf("edit changed order: %v", got)
}
if got := orderRequest(t, router, "", resource, body([]int64{b, a, c}, []int64{c, a, b})); got.Code != 401 {
t.Fatalf("missing auth: %+v", got)
}
normal, err := auth.GenerateToken(a, "edited", 1, "order-secret")
if err != nil {
t.Fatal(err)
}
// Speed-limit fixture needs a real ordinary account for auth lookup.
if resource == "speed-limit" {
uid, err := r.CreateUser("normal", "hash", 1, 0, 10, 1, 2, 1, 3, 0)
if err != nil {
t.Fatal(err)
}
normal, err = auth.GenerateToken(uid, "normal", 1, "order-secret")
if err != nil {
t.Fatal(err)
}
}
if got := orderRequest(t, router, normal, resource, body([]int64{b, a, c}, []int64{c, a, b})); got.Code != 403 {
t.Fatalf("non-admin auth: %+v", got)
}
})
}
}
@@ -0,0 +1,367 @@
package contract_test
import (
"database/sql"
"errors"
"fmt"
"go-backend/internal/security"
"go-backend/internal/store/model"
"go-backend/internal/store/repo"
"os"
"path/filepath"
"reflect"
"strconv"
"testing"
"time"
)
func TestDisplayOrderBackupAndRestart(t *testing.T) {
path := filepath.Join(t.TempDir(), "order.db")
r, err := repo.Open(path)
if err != nil {
t.Fatal(err)
}
a, err := r.CreateSpeedLimit("a", 100, 0, 1)
if err != nil {
t.Fatal(err)
}
b, err := r.CreateSpeedLimit("b", 200, 0, 1)
if err != nil {
t.Fatal(err)
}
if err := r.UpdateSpeedLimitDisplayOrder([]int64{a, b}, []int64{b, a}); err != nil {
t.Fatal(err)
}
check := func(r *repo.Repository) {
t.Helper()
rows, err := r.ListSpeedLimits()
if err != nil {
t.Fatal(err)
}
ids := []int64{}
for _, row := range rows {
ids = append(ids, row["id"].(int64))
}
if !reflect.DeepEqual(ids, []int64{a, b}) {
t.Fatalf("order: %v", ids)
}
}
if err := r.Close(); err != nil {
t.Fatal(err)
}
r, err = repo.Open(path)
if err != nil {
t.Fatal(err)
}
defer r.Close()
check(r)
backup, err := r.ExportPartial([]string{"speedLimits"})
if err != nil {
t.Fatal(err)
}
dest, err := repo.Open(filepath.Join(t.TempDir(), "restore.db"))
if err != nil {
t.Fatal(err)
}
defer dest.Close()
if _, err := dest.Import(backup, []string{"speedLimits"}); err != nil {
t.Fatal(err)
}
check(dest)
}
// Lifecycle and migration fixtures use DDL only to reproduce a pre-upgrade
// database and inject a write failure. All observations use public interfaces.
func TestDisplayOrderLifecycleDialects(t *testing.T) {
for _, dialect := range []string{"sqlite", "postgres"} {
t.Run(dialect, func(t *testing.T) {
open := orderTestDatabase(t, dialect)
r := open()
for _, resource := range []string{"user", "speed_limit"} {
t.Run(resource, func(t *testing.T) {
create := func(name string) int64 {
t.Helper()
var id int64
var err error
if resource == "user" {
id, err = r.CreateUser(name, security.MD5("test-password"), 1, 0, 123, 1, 4, 1, 5, 0)
} else {
id, err = r.CreateSpeedLimit(name, 123, 0, 1)
}
if err != nil {
t.Fatal(err)
}
return id
}
list := func(db *repo.Repository) []int64 {
t.Helper()
var rows []map[string]interface{}
var err error
if resource == "user" {
rows, err = db.ListUsers()
} else {
rows, err = db.ListSpeedLimits()
}
if err != nil {
t.Fatal(err)
}
ids := []int64{}
for _, row := range rows {
ids = append(ids, row["id"].(int64))
}
return ids
}
save := func(db *repo.Repository, ids, expected []int64) error {
if resource == "user" {
return db.UpdateUserDisplayOrder(ids, expected)
}
return db.UpdateSpeedLimitDisplayOrder(ids, expected)
}
a, b, c := create("a"), create("b"), create("c")
// Drop the new column to reproduce a database last opened by the old release.
if err := r.DB().Exec("ALTER TABLE \"" + resource + "\" DROP COLUMN display_order").Error; err != nil {
t.Fatal(err)
}
if err := r.Close(); err != nil {
t.Fatal(err)
}
r = open()
if got := list(r); !reflect.DeepEqual(got, []int64{c, b, a}) {
t.Fatalf("legacy order %v", got)
}
if err := save(r, []int64{a, c, b}, []int64{c, b, a}); err != nil {
t.Fatal(err)
}
backup, err := r.ExportPartial([]string{"users", "speedLimits"})
if err != nil {
t.Fatal(err)
}
restore := orderTestDatabase(t, dialect)()
defer restore.Close()
types := []string{"speedLimits"}
if resource == "user" {
types = []string{"users"}
}
if _, err := restore.Import(backup, types); err != nil {
t.Fatal(err)
}
if got := list(restore); !reflect.DeepEqual(got, []int64{a, c, b}) {
t.Fatalf("new backup order: %v", got)
}
before, err := r.ExportPartial([]string{"users", "speedLimits"})
if err != nil {
t.Fatal(err)
}
if err := r.Close(); err != nil {
t.Fatal(err)
}
r = open()
second := open()
defer second.Close()
if got := list(second); !reflect.DeepEqual(got, []int64{a, c, b}) {
t.Fatalf("cross-client/restart order %v", got)
}
// The next save fails on its second row; the first row must roll back.
if dialect == "sqlite" {
if err := r.DB().Exec(fmt.Sprintf("CREATE TRIGGER reject_order BEFORE UPDATE OF display_order ON \"%s\" WHEN NEW.id = %d BEGIN SELECT RAISE(ABORT, 'injected failure'); END", resource, a)).Error; err != nil {
t.Fatal(err)
}
} else {
if err := r.DB().Exec(fmt.Sprintf("ALTER TABLE \"%s\" ADD CONSTRAINT reject_order CHECK (id != %d OR display_order = 0)", resource, a)).Error; err != nil {
t.Fatal(err)
}
}
if err := save(r, []int64{b, a, c}, []int64{a, c, b}); err == nil {
t.Fatal("expected injected failure")
}
if got := list(second); !reflect.DeepEqual(got, []int64{a, c, b}) {
t.Fatalf("partial write: %v", got)
}
if dialect == "sqlite" {
if err := r.DB().Exec("DROP TRIGGER reject_order").Error; err != nil {
t.Fatal(err)
}
} else {
if err := r.DB().Exec("ALTER TABLE \"" + resource + "\" DROP CONSTRAINT reject_order").Error; err != nil {
t.Fatal(err)
}
}
// Exactly one cross-client save against the same observed order may win.
start := make(chan struct{})
results := make(chan error, 2)
for _, db := range []*repo.Repository{r, second} {
go func(db *repo.Repository) { <-start; results <- save(db, []int64{b, a, c}, []int64{a, c, b}) }(db)
}
close(start)
e1, e2 := <-results, <-results
if !((e1 == nil && errors.Is(e2, repo.ErrDisplayOrderConflict)) || (e2 == nil && errors.Is(e1, repo.ErrDisplayOrderConflict))) {
t.Fatalf("concurrent saves: %v / %v", e1, e2)
}
// Reordering must not mutate any exported fields besides rank.
after, err := r.ExportPartial([]string{"users", "speedLimits"})
if err != nil {
t.Fatal(err)
}
strip := func(b *model.BackupData) {
b.ExportedAt = 0
for i := range b.Users {
b.Users[i].DisplayOrder = nil
}
for i := range b.SpeedLimits {
b.SpeedLimits[i].DisplayOrder = nil
}
}
strip(before)
strip(after)
if !reflect.DeepEqual(before, after) {
t.Fatal("reorder modified non-order backup fields")
}
// Old backups omit ranks and must not overwrite an existing saved order.
legacy := orderTestDatabase(t, dialect)()
defer legacy.Close()
if _, err := legacy.Import(after, types); err != nil {
t.Fatal(err)
}
if got := list(legacy); !reflect.DeepEqual(got, []int64{c, b, a}) {
t.Fatalf("new legacy records: %v", got)
}
if _, err := r.Import(after, types); err != nil {
t.Fatal(err)
}
if got := list(r); !reflect.DeepEqual(got, []int64{b, a, c}) {
t.Fatalf("old backup reset existing order: %v", got)
}
d := create("d")
if got := list(r); !reflect.DeepEqual(got, []int64{d, b, a, c}) {
t.Fatalf("create after reorder: %v", got)
}
if err := save(second, []int64{a, b, c}, []int64{b, a, c}); !errors.Is(err, repo.ErrDisplayOrderConflict) {
t.Fatalf("stale membership: %v", err)
}
})
}
r.Close()
})
}
}
func orderTestDatabase(t *testing.T, dialect string) func() *repo.Repository {
t.Helper()
if dialect == "sqlite" {
path := filepath.Join(t.TempDir(), "lifecycle.db")
return func() *repo.Repository {
r, err := repo.Open(path)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { r.Close() })
return r
}
}
dsn := os.Getenv("FLVX_POSTGRES_TEST_DSN")
if dsn == "" {
t.Skip("set FLVX_POSTGRES_TEST_DSN for isolated PostgreSQL contracts")
}
admin, err := sql.Open("pgx", dsn)
if err != nil {
t.Fatal(err)
}
schema := "contract_display_order_" + strconv.FormatInt(time.Now().UnixNano(), 36)
if _, err := admin.Exec("CREATE SCHEMA \"" + schema + "\""); err != nil {
t.Fatal(err)
}
t.Cleanup(func() { admin.Exec("DROP SCHEMA \"" + schema + "\" CASCADE"); admin.Close() })
scoped, err := withSearchPath(dsn, schema)
if err != nil {
t.Fatal(err)
}
return func() *repo.Repository {
r, err := repo.OpenPostgres(scoped)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { r.Close() })
return r
}
}
func TestDisplayOrderConcurrentCreate(t *testing.T) {
for _, dialect := range []string{"sqlite", "postgres"} {
t.Run(dialect, func(t *testing.T) {
for _, resource := range []string{"user", "speed-limit"} {
t.Run(resource, func(t *testing.T) {
open := orderTestDatabase(t, dialect)
r, other := open(), open()
create := func(db *repo.Repository, name string) (int64, error) {
if resource == "user" {
return db.CreateUser(name, "hash", 1, 0, 10, 1, 2, 1, 3, 0)
}
return db.CreateSpeedLimit(name, 100, 0, 1)
}
save := func(ids, expected []int64) error {
if resource == "user" {
return r.UpdateUserDisplayOrder(ids, expected)
}
return r.UpdateSpeedLimitDisplayOrder(ids, expected)
}
a, err := create(r, "a")
if err != nil {
t.Fatal(err)
}
b, err := create(r, "b")
if err != nil {
t.Fatal(err)
}
start := make(chan struct{})
saved := make(chan error, 1)
type creation struct {
id int64
err error
}
created := make(chan creation, 1)
go func() { <-start; saved <- save([]int64{a, b}, []int64{b, a}) }()
go func() { <-start; id, err := create(other, "c"); created <- creation{id, err} }()
close(start)
saveErr, c := <-saved, <-created
if c.err != nil {
t.Fatal(c.err)
}
if saveErr != nil && !errors.Is(saveErr, repo.ErrDisplayOrderConflict) {
t.Fatalf("concurrent save: %v", saveErr)
}
var rows []map[string]interface{}
if resource == "user" {
rows, err = r.ListUsers()
} else {
rows, err = r.ListSpeedLimits()
}
if err != nil {
t.Fatal(err)
}
got := []int64{}
for _, row := range rows {
got = append(got, row["id"].(int64))
}
want := []int64{c.id, a, b}
if saveErr != nil {
want = []int64{c.id, b, a}
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("concurrent create order %v want %v", got, want)
}
if resource == "user" {
err = r.DeleteUserCascade(c.id)
} else {
err = r.DeleteSpeedLimit(c.id)
}
if err != nil {
t.Fatal(err)
}
if err := save(got, got); !errors.Is(err, repo.ErrDisplayOrderConflict) {
t.Fatalf("deleted membership not rejected: %v", err)
}
})
}
})
}
}