Compare commits

..

2 Commits

Author SHA1 Message Date
sagit b26e9ee9e4 fix: align card handles and show live deployment progress (#577)
* fix: keep user and limit drag handles beside names

* fix: show live node and tunnel deployment progress in list view
2026-11-10 10:55:53 +08:00
sagit 577c189737 feat: show current owner names and persist user and limit order (#576)
Resolve current forwarding-rule owner names by user ID. Add administrator-managed persistent user and speed-limit ordering, backup compatibility, and conflict recovery. Repair PostgreSQL explicit-ID import sequences transactionally.

Closes #574
Closes #575
2026-11-10 09:13:05 +08:00
32 changed files with 4044 additions and 823 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())
}
@@ -228,6 +228,7 @@ func (h *Handler) Register(mux *http.ServeMux) {
mux.HandleFunc("/api/v1/tunnel/update-order", h.tunnelUpdateOrder)
mux.HandleFunc("/api/v1/tunnel/batch-delete", h.tunnelBatchDelete)
mux.HandleFunc("/api/v1/tunnel/batch-redeploy", h.tunnelBatchRedeploy)
mux.HandleFunc("/api/v1/tunnel/batch-redeploy/stream", h.tunnelBatchRedeployStream)
mux.HandleFunc("/api/v1/tunnel/user/assign", h.userTunnelAssign)
mux.HandleFunc("/api/v1/tunnel/user/batch-assign", h.userTunnelBatchAssign)
mux.HandleFunc("/api/v1/tunnel/user/remove", h.userTunnelRemove)
@@ -248,6 +249,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)
+23 -37
View File
@@ -2046,51 +2046,37 @@ func normalizeBatchFailureReason(reason string) string {
return trimmed
}
func (h *Handler) redeployTunnelConfiguration(tunnelID int64) error {
if _, err := h.getTunnelRecord(tunnelID); err != nil {
return err
}
if nftMode, entryNodeIDs, err := h.tunnelUsesNftables(tunnelID); err != nil {
return err
} else if nftMode {
if len(entryNodeIDs) == 0 {
return errors.New("nftables 转发缺少入口节点")
}
return h.reconcileNftablesNodeByRequest(entryNodeIDs[0])
}
return h.redeployTunnelAndForwards(tunnelID)
}
func (h *Handler) tunnelBatchRedeploy(w http.ResponseWriter, r *http.Request) {
ids := idsFromBody(r, w)
if ids == nil {
return
}
success := 0
fail := 0
failures := make([]batchFailureDetail, 0)
result := batchOperationResult{}
for _, tunnelID := range ids {
tunnel, tunnelErr := h.getTunnelRecord(tunnelID)
tunnelName := ""
if tunnelErr == nil && tunnel != nil {
tunnelName, _ = h.repo.GetTunnelName(tunnelID)
name, _ := h.repo.GetTunnelName(tunnelID)
if err := h.redeployTunnelConfiguration(tunnelID); err != nil {
result.FailCount++
result.Failures = appendBatchFailure(result.Failures, tunnelID, name, err)
} else {
result.SuccessCount++
}
if tunnelErr != nil {
fail++
failures = appendBatchFailure(failures, tunnelID, tunnelName, tunnelErr)
continue
}
if nftMode, entryNodeIDs, modeErr := h.tunnelUsesNftables(tunnelID); modeErr != nil {
fail++
failures = appendBatchFailure(failures, tunnelID, tunnelName, modeErr)
continue
} else if nftMode {
if len(entryNodeIDs) == 0 {
fail++
failures = appendBatchFailureReason(failures, tunnelID, tunnelName, "nftables 转发缺少入口节点")
continue
}
if reconcileErr := h.reconcileNftablesNodeByRequest(entryNodeIDs[0]); reconcileErr != nil {
fail++
failures = appendBatchFailure(failures, tunnelID, tunnelName, reconcileErr)
continue
}
success++
continue
}
if err := h.redeployTunnelAndForwards(tunnelID); err != nil {
fail++
failures = appendBatchFailure(failures, tunnelID, tunnelName, err)
continue
}
success++
}
response.WriteJSON(w, response.OK(batchOperationResult{SuccessCount: success, FailCount: fail, Failures: failures}))
response.WriteJSON(w, response.OK(result))
}
func (h *Handler) userTunnelAssign(w http.ResponseWriter, r *http.Request) {
@@ -0,0 +1,105 @@
package handler
import (
"encoding/json"
"errors"
"net/http"
"time"
"go-backend/internal/http/response"
)
type redeployProgress struct {
Total int `json:"total"`
Completed int `json:"completed"`
Success int `json:"success"`
Failed int `json:"failed"`
}
type redeployItemResult struct {
ID int64 `json:"id"`
Name string `json:"name"`
Success bool `json:"success"`
Reason string `json:"reason,omitempty"`
}
// tunnelBatchRedeployStream reports completed tunnels, not estimated runtime.
// A disconnected caller may have an in-flight mutation whose result it did not
// receive. Finish that mutation, but never start another or automatically retry.
func (h *Handler) tunnelBatchRedeployStream(w http.ResponseWriter, r *http.Request) {
ids := idsFromBody(r, w)
if ids == nil {
return
}
// idsFromBody sorts IDs; ignore duplicate selections so each tunnel runs once.
unique := ids[:0]
for _, id := range ids {
if len(unique) == 0 || unique[len(unique)-1] != id {
unique = append(unique, id)
}
}
ids = unique
if len(ids) == 0 {
response.WriteJSON(w, response.ErrDefault("ids不能为空"))
return
}
_, err := prepareDiagnosisStreamResponse(w)
if err != nil {
response.WriteJSON(w, response.ErrDefault(err.Error()))
return
}
encoder := json.NewEncoder(w)
controller := http.NewResponseController(w)
emit := func(kind string, data interface{}) bool {
if r.Context().Err() != nil {
return false
}
// The server's WriteTimeout is an absolute request deadline. Renew only
// this stream's bounded write window as progress is emitted, so a long
// batch is not cut off despite making steady progress. Recorders and
// other non-network writers may not support deadlines.
if err := controller.SetWriteDeadline(time.Now().Add(2 * time.Minute)); err != nil && !errors.Is(err, http.ErrNotSupported) {
return false
}
event := diagnosisStreamEvent{Type: kind, Data: data, TS: time.Now().UnixMilli()}
if err := encoder.Encode(event); err != nil {
return false
}
// Buffered writes can succeed while the actual socket flush fails.
// Do not start a mutation after its processing event failed to flush.
return controller.Flush() == nil
}
progress := redeployProgress{Total: len(ids)}
result := batchOperationResult{}
if !emit("start", progress) {
return
}
for _, id := range ids {
if r.Context().Err() != nil {
return
}
name, _ := h.repo.GetTunnelName(id)
if !emit("processing", map[string]interface{}{"id": id, "name": name, "progress": progress}) {
return
}
if r.Context().Err() != nil {
return
}
item := redeployItemResult{ID: id, Name: name}
if err := h.redeployTunnelConfiguration(id); err != nil {
result.FailCount++
result.Failures = appendBatchFailure(result.Failures, id, name, err)
item.Reason = result.Failures[len(result.Failures)-1].Reason
progress.Failed++
} else {
result.SuccessCount++
item.Success = true
progress.Success++
}
progress.Completed++
if !emit("item", map[string]interface{}{"result": item, "progress": progress}) {
return
}
}
emit("done", map[string]interface{}{"progress": progress, "result": result})
}
@@ -0,0 +1,248 @@
package handler
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"go-backend/internal/http/middleware"
runtimenft "go-backend/internal/runtime/nftables"
)
type blockingRedeployManager struct {
fakeNftablesManager
entered chan struct{}
release chan struct{}
once sync.Once
}
func (m *blockingRedeployManager) Reconcile(ctx context.Context, cfg runtimenft.SSHConfig, plan runtimenft.NodePlan) (runtimenft.ApplyResult, error) {
m.once.Do(func() { close(m.entered); <-m.release })
return m.fakeNftablesManager.Reconcile(ctx, cfg, plan)
}
func TestTunnelRedeployStreamFlushesBeforeSlowMutationAndReportsMixedResults(t *testing.T) {
fixture := setupNftablesHandler(t)
h := fixture.handler
seedNftablesSSHConfig(t, h, fixture.nodeID)
id := seedTunnelForNftables(t, h, "stream-tunnel", fixture.nodeID)
manager := &blockingRedeployManager{entered: make(chan struct{}), release: make(chan struct{})}
h.nftablesManager = manager
var releaseOnce sync.Once
release := func() { releaseOnce.Do(func() { close(manager.release) }) }
defer release()
server := httptest.NewServer(http.HandlerFunc(h.tunnelBatchRedeployStream))
defer server.Close()
client := &http.Client{Timeout: 5 * time.Second}
response, err := client.Post(server.URL, "application/json", strings.NewReader(fmt.Sprintf(`{"ids":[%d,999999,%d]}`, id, id)))
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
decoder := json.NewDecoder(response.Body)
read := func(want string) map[string]interface{} {
t.Helper()
var event struct {
Type string
Data map[string]interface{}
}
if err := decoder.Decode(&event); err != nil {
t.Fatal(err)
}
if event.Type != want {
t.Fatalf("expected %s, got %s", want, event.Type)
}
return event.Data
}
if start := read("start"); start["total"] != float64(2) {
t.Fatalf("duplicate IDs not deduplicated: %+v", start)
}
processing := read("processing")
if processing["name"] != "stream-tunnel" || processing["id"] != float64(id) {
t.Fatalf("wrong current tunnel: %+v", processing)
}
select {
case <-manager.entered:
case <-time.After(2 * time.Second):
t.Fatal("did not start mutation")
}
// Both events arrived over HTTP while the actual mutation is still blocked.
release()
item := read("item")
if item["result"].(map[string]interface{})["success"] != true || item["progress"].(map[string]interface{})["completed"] != float64(1) {
t.Fatalf("wrong success result: %+v", item)
}
read("processing")
item = read("item")
if item["result"].(map[string]interface{})["success"] != false || item["result"].(map[string]interface{})["reason"] == "" {
t.Fatalf("wrong failure result: %+v", item)
}
done := read("done")
progress := done["progress"].(map[string]interface{})
result := done["result"].(map[string]interface{})
if progress["completed"] != float64(2) || progress["success"] != float64(1) || progress["failed"] != float64(1) || result["successCount"] != float64(1) || result["failCount"] != float64(1) {
t.Fatalf("incorrect final result: %+v", done)
}
if manager.reconcileCount() != 1 {
t.Fatalf("expected one execution, got %d", manager.reconcileCount())
}
if err := decoder.Decode(new(interface{})); err != io.EOF {
t.Fatalf("expected EOF after done, got %v", err)
}
}
func TestTunnelRedeployStreamStopsAfterDisconnect(t *testing.T) {
fixture := setupNftablesHandler(t)
h := fixture.handler
seedNftablesSSHConfig(t, h, fixture.nodeID)
first := seedTunnelForNftables(t, h, "first", fixture.nodeID)
second := seedTunnelForNftables(t, h, "second", fixture.nodeID)
manager := &blockingRedeployManager{entered: make(chan struct{}), release: make(chan struct{})}
h.nftablesManager = manager
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/batch-redeploy/stream", strings.NewReader(fmt.Sprintf(`{"ids":[%d,%d]}`, first, second))).WithContext(ctx)
res := httptest.NewRecorder()
finished := make(chan struct{})
go func() { defer close(finished); h.tunnelBatchRedeployStream(res, req) }()
select {
case <-manager.entered:
case <-time.After(2 * time.Second):
close(manager.release)
t.Fatal("mutation did not start")
}
cancel()
close(manager.release)
select {
case <-finished:
case <-time.After(2 * time.Second):
t.Fatal("handler did not stop")
}
if manager.reconcileCount() != 1 {
t.Fatalf("started another mutation after disconnect: %d", manager.reconcileCount())
}
if strings.Contains(res.Body.String(), `"type":"done"`) {
t.Fatal("disconnected stream claimed completion")
}
}
type failedRedeployWriter struct{ *httptest.ResponseRecorder }
func (w failedRedeployWriter) Write([]byte) (int, error) { return 0, io.ErrClosedPipe }
func TestTunnelRedeployStreamDoesNotMutateWhenInitialWriteFails(t *testing.T) {
fixture := setupNftablesHandler(t)
h := fixture.handler
seedNftablesSSHConfig(t, h, fixture.nodeID)
id := seedTunnelForNftables(t, h, "stream-tunnel", fixture.nodeID)
manager := &fakeNftablesManager{}
h.nftablesManager = manager
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(fmt.Sprintf(`{"ids":[%d]}`, id)))
h.tunnelBatchRedeployStream(failedRedeployWriter{httptest.NewRecorder()}, req)
if manager.reconcileCount() != 0 {
t.Fatal("mutation ran despite failed progress stream")
}
}
type delayedRedeployManager struct{ fakeNftablesManager }
func (m *delayedRedeployManager) Reconcile(ctx context.Context, cfg runtimenft.SSHConfig, plan runtimenft.NodePlan) (runtimenft.ApplyResult, error) {
time.Sleep(100 * time.Millisecond)
return m.fakeNftablesManager.Reconcile(ctx, cfg, plan)
}
type shortRedeployDeadlineWriter struct {
http.ResponseWriter
deadlineCalls *atomic.Int32
}
func (w shortRedeployDeadlineWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter }
func (w shortRedeployDeadlineWriter) SetWriteDeadline(deadline time.Time) error {
remaining := time.Until(deadline)
if remaining < time.Minute || remaining > 2*time.Minute {
return fmt.Errorf("expected bounded two-minute write window, got %v", remaining)
}
w.deadlineCalls.Add(1)
// Shorten the production idle window for a real, fast timeout regression.
return http.NewResponseController(w.ResponseWriter).SetWriteDeadline(time.Now().Add(200 * time.Millisecond))
}
func TestTunnelRedeployStreamRenewsDeadlineThroughRequestLog(t *testing.T) {
fixture := setupNftablesHandler(t)
h := fixture.handler
seedNftablesSSHConfig(t, h, fixture.nodeID)
ids := make([]int64, 3)
for i := range ids {
ids[i] = seedTunnelForNftables(t, h, fmt.Sprintf("slow-%d", i), fixture.nodeID)
}
manager := &delayedRedeployManager{}
h.nftablesManager = manager
var deadlineCalls atomic.Int32
logged := middleware.RequestLog(http.HandlerFunc(h.tunnelBatchRedeployStream))
server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
logged.ServeHTTP(shortRedeployDeadlineWriter{w, &deadlineCalls}, r)
}))
server.Config.WriteTimeout = 50 * time.Millisecond
server.Start()
defer server.Close()
body, _ := json.Marshal(map[string]interface{}{"ids": ids})
client := &http.Client{Timeout: 5 * time.Second}
response, err := client.Post(server.URL, "application/json", strings.NewReader(string(body)))
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
payload, err := io.ReadAll(response.Body)
if err != nil {
t.Fatalf("stream timed out across batch: %v (body %s)", err, payload)
}
if !strings.Contains(string(payload), `"type":"done"`) {
t.Fatalf("batch did not finish: %s", payload)
}
if manager.reconcileCount() != 3 {
t.Fatalf("wrong execution count: %d", manager.reconcileCount())
}
if deadlineCalls.Load() != 8 {
t.Fatalf("each of 8 events must renew its deadline, got %d", deadlineCalls.Load())
}
}
type flushFailureRedeployWriter struct {
*httptest.ResponseRecorder
flushes int
}
func (w *flushFailureRedeployWriter) FlushError() error {
w.flushes++
if w.flushes == 2 {
return io.ErrClosedPipe
}
w.ResponseRecorder.Flush()
return nil
}
func TestTunnelRedeployStreamStopsBeforeMutationOnProcessingFlushError(t *testing.T) {
fixture := setupNftablesHandler(t)
h := fixture.handler
seedNftablesSSHConfig(t, h, fixture.nodeID)
id := seedTunnelForNftables(t, h, "flush-error", fixture.nodeID)
manager := &fakeNftablesManager{}
h.nftablesManager = manager
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(fmt.Sprintf(`{"ids":[%d]}`, id)))
writer := &flushFailureRedeployWriter{ResponseRecorder: httptest.NewRecorder()}
middleware.RequestLog(http.HandlerFunc(h.tunnelBatchRedeployStream)).ServeHTTP(writer, req)
if writer.flushes != 2 {
t.Fatalf("expected start and processing flush, got %d", writer.flushes)
}
if manager.reconcileCount() != 0 {
t.Fatal("mutation ran after failed processing flush")
}
}
+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
@@ -14,6 +14,11 @@ type statusWriter struct {
status int
}
// Unwrap preserves ResponseController features such as per-response deadlines.
func (w *statusWriter) Unwrap() http.ResponseWriter {
return w.ResponseWriter
}
func (w *statusWriter) WriteHeader(code int) {
w.status = code
w.ResponseWriter.WriteHeader(code)
@@ -27,10 +32,12 @@ func (w *statusWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
return hj.Hijack()
}
func (w *statusWriter) FlushError() error {
return http.NewResponseController(w.ResponseWriter).Flush()
}
func (w *statusWriter) Flush() {
if f, ok := w.ResponseWriter.(http.Flusher); ok {
f.Flush()
}
_ = w.FlushError()
}
func (w *statusWriter) ReadFrom(r io.Reader) (int64, error) {
+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 {
+46 -16
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))
@@ -1001,7 +1001,8 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) {
var rows []fwdRow
err := r.db.Model(&model.Forward{}).
Select("forward.id, forward.user_id, forward.user_name, forward.name, forward.tunnel_id, COALESCE(tunnel.name, '') AS tunnel_name, COALESCE(tunnel.traffic_ratio, 1.0) AS traffic_ratio, forward.remote_addr, COALESCE(forward.strategy, 'fifo') AS strategy, forward.in_flow, forward.out_flow, forward.created_time, forward.status, forward.inx, forward.speed_id, forward.max_conn, forward.ip_max_conn, forward.ip_speed_id, COALESCE(ip_speed_limit.name, '') AS ip_speed_limit_name, forward.proxy_protocol, forward.proxy_protocol_receive, forward.proxy_protocol_send").
Select("forward.id, forward.user_id, COALESCE(owner.\"user\", forward.user_name) AS user_name, forward.name, forward.tunnel_id, COALESCE(tunnel.name, '') AS tunnel_name, COALESCE(tunnel.traffic_ratio, 1.0) AS traffic_ratio, forward.remote_addr, COALESCE(forward.strategy, 'fifo') AS strategy, forward.in_flow, forward.out_flow, forward.created_time, forward.status, forward.inx, forward.speed_id, forward.max_conn, forward.ip_max_conn, forward.ip_speed_id, COALESCE(ip_speed_limit.name, '') AS ip_speed_limit_name, forward.proxy_protocol, forward.proxy_protocol_receive, forward.proxy_protocol_send").
Joins(`LEFT JOIN "user" AS owner ON owner.id = forward.user_id`).
Joins("LEFT JOIN tunnel ON tunnel.id = forward.tunnel_id").
Joins("LEFT JOIN speed_limit AS ip_speed_limit ON ip_speed_limit.id = forward.ip_speed_id").
Order("forward.inx ASC, forward.id ASC").
@@ -2092,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,
@@ -2289,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 {
@@ -2372,7 +2375,24 @@ func (r *Repository) Import(backup *model.BackupData, types []string) (*model.Im
typeSet[t] = true
}
err := r.db.Transaction(func(tx *gorm.DB) error {
err := backupImportTransaction(r.db, func(tx *gorm.DB) error {
result = &model.ImportResult{}
tables, err := lockPostgresBackupTables(tx, backup, typeSet)
if err != nil {
return err
}
// SQLite still needs its writer reservation for display order.
// PostgreSQL membership locks were taken with all imported tables above.
if tx.Dialector.Name() != "postgres" && typeSet["users"] {
if err := lockDisplayOrder(tx, "user"); err != nil {
return err
}
}
if tx.Dialector.Name() != "postgres" && typeSet["speedLimits"] {
if err := lockDisplayOrder(tx, "speed_limit"); err != nil {
return err
}
}
now := unixMilliNow()
if typeSet["users"] && len(backup.Users) > 0 {
@@ -2445,7 +2465,7 @@ func (r *Repository) Import(backup *model.BackupData, types []string) (*model.Im
}
result.ConfigsImported = count
}
return nil
return advancePostgresBackupSequences(tx, tables)
})
if err != nil {
return nil, err
@@ -2477,12 +2497,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
@@ -2765,11 +2790,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,195 @@
package repo
import (
"errors"
"fmt"
"time"
"go-backend/internal/store/model"
"gorm.io/gorm"
)
var errBackupImportBusy = errors.New("database is busy; retry the backup import")
// Only retry lock acquisition failures, which occur before any backup writes.
// NOWAIT releases partial lock sets instead of deadlocking with existing
// child-before-parent deletes or forward-before-user traffic updates.
func backupImportTransaction(db *gorm.DB, importRows func(*gorm.DB) error) error {
var err error
for attempt := 0; attempt < 10; attempt++ {
err = db.Transaction(importRows)
if !errors.Is(err, errBackupImportBusy) {
return err
}
if attempt < 9 {
time.Sleep(min(10*time.Millisecond<<attempt, 100*time.Millisecond))
}
}
return err
}
// Lock every table that import can write, including generated-ID child tables.
// No writes happen until the complete set is held. Existing CRUD transactions
// can therefore finish regardless of their own parent/child mutation order.
// The returned subset contains only tables receiving explicit backup IDs.
func lockPostgresBackupTables(tx *gorm.DB, backup *model.BackupData, types map[string]bool) ([]string, error) {
if tx.Dialector.Name() != "postgres" {
return nil, nil
}
chains, grants, ports := 0, 0, 0
for _, tunnel := range backup.Tunnels {
chains += len(tunnel.ChainTunnels)
}
for _, permission := range backup.Permissions {
grants += len(permission.Grants)
}
for _, forward := range backup.Forwards {
if forward.ForwardPorts != nil {
ports++
}
}
var tables []string
for _, item := range []struct {
kind string
count int
name string
explicitIDs bool
}{
{"users", len(backup.Users), "user", true},
{"speedLimits", len(backup.SpeedLimits), "speed_limit", true},
{"nodes", len(backup.Nodes), "node", true},
{"tunnels", len(backup.Tunnels), "tunnel", true},
{"tunnels", chains, "chain_tunnel", true},
{"forwards", len(backup.Forwards), "forward", true},
{"userTunnels", len(backup.UserTunnels), "user_tunnel", true},
{"tunnelGroups", len(backup.TunnelGroups), "tunnel_group", true},
{"userGroups", len(backup.UserGroups), "user_group", true},
{"permissions", len(backup.Permissions), "group_permission", true},
{"permissions", grants, "group_permission_grant", true},
{"users", len(backup.Users), "user_quota", false},
{"forwards", ports, "forward_port", false},
{"tunnelGroups", len(backup.TunnelGroups), "tunnel_group_tunnel", false},
{"userGroups", len(backup.UserGroups), "user_group_user", false},
{"configs", len(FilterBackupConfigs(backup.Configs)), "vite_config", false},
} {
if !types[item.kind] || item.count == 0 {
continue
}
// Resolve the same relation as normal unqualified repository queries,
// even when the first search_path schema has no copy of this table.
var relation struct {
Schema string
Name string
}
if err := tx.Raw(`SELECT n.nspname AS schema, c.relname AS name
FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
WHERE c.oid = pg_catalog.to_regclass(?)`, quoteSQLIdentifier(item.name)).Scan(&relation).Error; err != nil {
return nil, err
}
if relation.Name == "" {
return nil, fmt.Errorf("missing import table %s", item.name)
}
table := quoteSQLIdentifier(relation.Schema) + "." + quoteSQLIdentifier(relation.Name)
if err := tx.Exec("LOCK TABLE " + table + " IN SHARE ROW EXCLUSIVE MODE NOWAIT").Error; err != nil {
var pgError interface{ SQLState() string }
if errors.As(err, &pgError) && pgError.SQLState() == "55P03" {
return nil, fmt.Errorf("%w: %w", errBackupImportBusy, err)
}
return nil, err
}
if item.explicitIDs {
tables = append(tables, table)
}
}
return tables, nil
}
func advancePostgresBackupSequences(tx *gorm.DB, tables []string) error {
for _, table := range tables {
if err := advancePostgresBackupSequence(tx, table); err != nil {
return fmt.Errorf("advance imported %s id sequence: %w", table, err)
}
}
return nil
}
// The caller holds the table's SHARE ROW EXCLUSIVE lock until commit. Unlike
// startup's setval repair, RESTART is transactional and invalidates other
// connections' cached values. Never return to IDs already allocated, even if
// their rows were deleted or their transactions rolled back.
func advancePostgresBackupSequence(tx *gorm.DB, table string) error {
var maxID int64
if err := tx.Raw("SELECT COALESCE(MAX(id), 0) FROM " + table).Scan(&maxID).Error; err != nil {
return err
}
if maxID <= 0 {
return nil
}
type sequenceInfo struct {
Schema string
Name string
Cache int64
Increment int64
Maximum int64
}
var sequences []sequenceInfo
// Prefer the actual default's catalog dependency: a legacy custom nextval
// default need not own its sequence, and the old owned serial may remain.
// Identity columns have no pg_attrdef, so use their owned sequence instead.
if err := tx.Raw(`WITH defaults AS (
SELECT dep.refobjid AS oid FROM pg_catalog.pg_attrdef def
JOIN pg_catalog.pg_attribute attr ON attr.attrelid = def.adrelid AND attr.attnum = def.adnum
JOIN pg_catalog.pg_depend dep ON dep.classid = 'pg_catalog.pg_attrdef'::regclass
AND dep.objid = def.oid AND dep.refclassid = 'pg_catalog.pg_class'::regclass
JOIN pg_catalog.pg_sequence seq ON seq.seqrelid = dep.refobjid
WHERE def.adrelid = ?::regclass AND attr.attname = 'id'
), candidates AS (
SELECT oid FROM defaults
UNION SELECT pg_catalog.pg_get_serial_sequence(?, 'id')::regclass WHERE NOT EXISTS (SELECT 1 FROM defaults)
)
SELECT n.nspname AS schema, c.relname AS name,
s.seqcache AS cache, s.seqincrement AS increment, s.seqmax AS maximum
FROM pg_catalog.pg_sequence s
JOIN pg_catalog.pg_class c ON c.oid = s.seqrelid
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
WHERE s.seqrelid IN (SELECT oid FROM candidates)`, table, table).Scan(&sequences).Error; err != nil {
return err
}
if len(sequences) != 1 || sequences[0].Increment <= 0 {
return fmt.Errorf("missing or ambiguous ascending ID sequence")
}
sequence := sequences[0]
name := quoteSQLIdentifier(sequence.Schema) + "." + quoteSQLIdentifier(sequence.Name)
// Reapplying CACHE without changing its setting takes a sequence DDL lock.
// Read last_value only after that lock: even direct concurrent nextval calls
// must finish first. The table locks already exclude normal INSERTs.
if err := tx.Exec(fmt.Sprintf("ALTER SEQUENCE %s CACHE %d", name, sequence.Cache)).Error; err != nil {
return err
}
var state struct {
LastValue int64
IsCalled bool
}
if err := tx.Raw("SELECT last_value, is_called FROM " + name).Scan(&state).Error; err != nil {
return err
}
if state.LastValue < 1 {
return fmt.Errorf("ID sequence must allocate positive IDs")
}
if maxID >= sequence.Maximum || (state.IsCalled && state.LastValue > sequence.Maximum-sequence.Increment) {
return fmt.Errorf("ID sequence exhausted")
}
next := state.LastValue
if state.IsCalled {
next += sequence.Increment
}
if next <= maxID {
steps := (maxID-next)/sequence.Increment + 1
if steps > (sequence.Maximum-next)/sequence.Increment {
return fmt.Errorf("ID sequence exhausted")
}
next += steps * sequence.Increment
}
return tx.Exec(fmt.Sprintf("ALTER SEQUENCE %s RESTART WITH %d", name, next)).Error
}
@@ -37,8 +37,14 @@ func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]mode
if tx == nil {
return nil, errors.New("database unavailable")
}
var forwards []model.Forward
err := tx.Where("tunnel_id = ?", tunnelID).Order("id ASC").Find(&forwards).Error
var forwards []struct {
model.Forward
CurrentUserName string
}
err := tx.Model(&model.Forward{}).
Select(`forward.*, COALESCE(owner."user", forward.user_name) AS current_user_name`).
Joins(`LEFT JOIN "user" AS owner ON owner.id = forward.user_id`).
Where("forward.tunnel_id = ?", tunnelID).Order("forward.id ASC").Find(&forwards).Error
if err != nil {
return nil, err
}
@@ -48,7 +54,7 @@ func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]mode
rows = append(rows, model.ForwardRecord{
ID: f.ID,
UserID: f.UserID,
UserName: f.UserName,
UserName: f.CurrentUserName,
Name: f.Name,
TunnelID: f.TunnelID,
RemoteAddr: f.RemoteAddr,
@@ -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)
}
})
}
})
}
}
@@ -0,0 +1,300 @@
package contract_test
import (
"bytes"
"database/sql"
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strconv"
"strings"
"testing"
"time"
"go-backend/internal/auth"
httpserver "go-backend/internal/http"
"go-backend/internal/http/handler"
"go-backend/internal/security"
"go-backend/internal/store/model"
"go-backend/internal/store/repo"
)
func TestForwardOwnerNamesFollowAccountRenames(t *testing.T) {
for _, database := range []string{"sqlite", "postgres"} {
t.Run(database, func(t *testing.T) {
for _, mode := range []string{"admin-name-only", "admin-name-and-password", "self-name-and-password"} {
t.Run(mode, func(t *testing.T) {
const secret = "issue574-isolated-test"
open := ownerNamesRepository(t, database)
r := open()
router := httpserver.NewRouter(handler.New(r, secret), secret)
for _, u := range []model.User{
{ID: 9103, User: "before-name", Pwd: security.MD5("old-password"), RoleID: 1, Status: 1, PasswordChangedAt: time.Now().Add(-10 * time.Second).UnixMilli()},
{ID: 9104, User: "other-user", Pwd: security.MD5("other-password"), RoleID: 1, Status: 1, PasswordChangedAt: time.Now().Add(-10 * time.Second).UnixMilli()},
} {
if err := r.DB().Create(&u).Error; err != nil {
t.Fatal(err)
}
}
exp := time.Now().Add(time.Hour).UnixMilli()
f := model.Forward{ID: 9301, UserID: 9103, UserName: "before-name", Name: "existing-rule", TunnelID: 9201, RemoteAddr: "127.0.0.1:8080", Status: 1, InFlow: 123, OutFlow: 456}
for _, v := range []interface{}{&model.Tunnel{ID: 9201, Name: "isolated", Type: 1, Protocol: "tls", Status: 1, Flow: 1}, &model.UserTunnel{ID: 9401, UserID: 9103, TunnelID: 9201, Status: 1, Flow: 100, Num: 10, ExpTime: exp}, &f} {
if err := r.DB().Create(v).Error; err != nil {
t.Fatal(err)
}
}
token := func(id int64, name string, role int) string {
v, e := auth.GenerateTokenAt(id, name, role, secret, time.Now().Add(time.Second))
if e != nil {
t.Fatal(e)
}
return v
}
admin := token(1, "admin_user", 0)
oldToken, e := auth.GenerateTokenAt(9103, "before-name", 1, secret, time.Now().Add(-time.Second))
if e != nil {
t.Fatal(e)
}
call := func(path, tok string, body interface{}) (int, json.RawMessage) {
b, err := json.Marshal(body)
if err != nil {
t.Fatal(err)
}
q := httptest.NewRequest(http.MethodPost, path, bytes.NewReader(b))
q.Header.Set("Authorization", tok)
q.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
router.ServeHTTP(w, q)
var out struct {
Code int `json:"code"`
Data json.RawMessage `json:"data"`
Msg string `json:"msg"`
}
if e := json.Unmarshal(w.Body.Bytes(), &out); e != nil {
t.Fatal(e)
}
return out.Code, out.Data
}
path, tok := "/api/v1/user/update", admin
body := map[string]interface{}{"id": 9103, "user": "after-name", "flow": 100, "num": 10, "expTime": exp, "flowResetTime": 1, "status": 1, "maxConn": 0}
if mode == "admin-name-and-password" {
body["pwd"] = "new-password"
}
if mode == "self-name-and-password" {
path = "/api/v1/user/updatePassword"
tok = oldToken
body = map[string]interface{}{"newUsername": "after-name", "currentPassword": "old-password", "newPassword": "new-password", "confirmPassword": "new-password"}
}
if code, _ := call(path, tok, body); code != 0 {
t.Fatalf("rename code=%d", code)
}
fresh := token(9103, "after-name", 1)
code, data := call("/api/v1/forward/list", admin, map[string]interface{}{})
if code != 0 {
t.Fatal(code)
}
var rows []map[string]interface{}
if err := json.Unmarshal(data, &rows); err != nil {
t.Fatal(err)
}
if len(rows) != 1 || rows[0]["userName"] != "after-name" || rows[0]["userId"] != float64(9103) {
t.Fatalf("expected current owner name: %s", data)
}
if rows[0]["id"] != float64(9301) || rows[0]["inFlow"] != float64(123) || rows[0]["outFlow"] != float64(456) || rows[0]["status"] != float64(1) {
t.Fatalf("rule identity/traffic/status changed: %s", data)
}
if code, data = call("/api/v1/forward/list", fresh, map[string]interface{}{}); code != 0 {
t.Fatalf("owner denied: %d", code)
}
rows = nil
if err := json.Unmarshal(data, &rows); err != nil {
t.Fatal(err)
}
if len(rows) != 1 || rows[0]["userName"] != "after-name" {
t.Fatalf("owner cannot see renamed rule: %s", data)
}
code, data = call("/api/v1/tunnel/delete-preview", admin, map[string]interface{}{"id": 9201})
if code != 0 {
t.Fatalf("preview code=%d", code)
}
var preview struct {
SampleForwards []struct {
ID int64 `json:"id"`
UserID int64 `json:"userId"`
UserName string `json:"userName"`
} `json:"sampleForwards"`
}
if err := json.Unmarshal(data, &preview); err != nil {
t.Fatal(err)
}
if len(preview.SampleForwards) != 1 || preview.SampleForwards[0].ID != 9301 || preview.SampleForwards[0].UserID != 9103 || preview.SampleForwards[0].UserName != "after-name" {
t.Fatalf("preview must show current owner: %s", data)
}
code, data = call("/api/v1/tunnel/batch-delete-preview", admin, map[string]interface{}{"ids": []int64{9201}})
if code != 0 {
t.Fatalf("batch preview code=%d", code)
}
var batch struct {
Items []struct {
SampleForwards []struct {
UserName string `json:"userName"`
} `json:"sampleForwards"`
} `json:"items"`
}
if err := json.Unmarshal(data, &batch); err != nil {
t.Fatal(err)
}
if len(batch.Items) != 1 || len(batch.Items[0].SampleForwards) != 1 || batch.Items[0].SampleForwards[0].UserName != "after-name" {
t.Fatalf("batch preview must show current owner: %s", data)
}
exported, err := r.ExportAll()
if err != nil {
t.Fatal(err)
}
if len(exported.UserTunnels) != 1 || exported.UserTunnels[0].UserID != 9103 || exported.UserTunnels[0].TunnelID != 9201 || exported.UserTunnels[0].Flow != 100 || exported.UserTunnels[0].Status != 1 {
t.Fatalf("rename changed tunnel grant: %+v", exported.UserTunnels)
}
oldCode, _ := call("/api/v1/forward/list", oldToken, map[string]interface{}{})
wantOld := 0
if mode != "admin-name-only" {
wantOld = 401
}
if oldCode != wantOld {
t.Fatalf("old token=%d want %d", oldCode, wantOld)
}
// Re-use the abandoned username for another identity; it must not acquire rules.
if code, _ := call("/api/v1/user/update", admin, map[string]interface{}{"id": 9104, "user": "before-name", "flow": 100, "num": 10, "expTime": exp, "flowResetTime": 1, "status": 1, "maxConn": 0}); code != 0 {
t.Fatalf("reuse old name: %d", code)
}
other := token(9104, "before-name", 1)
code, data = call("/api/v1/forward/list", other, map[string]interface{}{})
if code != 0 {
t.Fatal(code)
}
rows = nil
if err := json.Unmarshal(data, &rows); err != nil {
t.Fatal(err)
}
if len(rows) != 0 {
t.Fatalf("old name gained rules: %s", data)
}
if code, _ := call("/api/v1/forward/reset-flow", other, map[string]interface{}{"id": 9301}); code == 0 {
t.Fatal("other user reset original rule")
}
if code, _ := call("/api/v1/forward/reset-flow", fresh, map[string]interface{}{"id": 9301}); code != 0 {
t.Fatalf("owner reset denied: %d", code)
}
})
}
})
}
}
// Each PostgreSQL test owns only its unique schema, never the shared test database.
func ownerNamesRepository(t *testing.T, database string) func() *repo.Repository {
t.Helper()
dsn := filepath.Join(t.TempDir(), "owners.db")
if database == "postgres" {
baseDSN := strings.TrimSpace(os.Getenv("FLVX_POSTGRES_TEST_DSN"))
if baseDSN == "" {
t.Skip("set FLVX_POSTGRES_TEST_DSN to run postgres owner-name contracts")
}
schema := "contract_owner_names_" + strconv.FormatInt(time.Now().UnixNano(), 36)
db, err := sql.Open("pgx", baseDSN)
if err != nil {
t.Fatal(err)
}
if _, err = db.Exec(`CREATE SCHEMA "` + schema + `"`); err != nil {
db.Close()
t.Fatal(err)
}
t.Cleanup(func() { _, _ = db.Exec(`DROP SCHEMA IF EXISTS "` + schema + `" CASCADE`); _ = db.Close() })
dsn, err = withSearchPath(baseDSN, schema)
if err != nil {
t.Fatal(err)
}
}
return func() *repo.Repository {
t.Helper()
var r *repo.Repository
var err error
if database == "postgres" {
r, err = repo.OpenPostgres(dsn)
} else {
r, err = repo.Open(dsn)
}
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = r.Close() })
return r
}
}
func TestForwardOwnerNamesSurviveLegacyBackupAndRestart(t *testing.T) {
for _, database := range []string{"sqlite", "postgres"} {
t.Run(database, func(t *testing.T) {
open := ownerNamesRepository(t, database)
r := open()
// A legacy backup stores the rule's original name, not the user's current one.
const legacy = `{"users":[{"id":9103,"user":"current-owner","pwd":"test-only-password","roleId":1,"status":1}],
"forwards":[
{"id":9301,"userId":9103,"userName":"former-owner","name":"historical-rule","tunnelId":9201,"inFlow":123,"outFlow":456,"status":1},
{"id":9302,"userId":9199,"userName":"missing-owner","name":"orphan-rule","tunnelId":9201,"status":1},
{"id":9303,"userId":9198,"userName":"","name":"unknown-owner-rule","tunnelId":9201,"status":1}]}`
var backup model.BackupData
if err := json.Unmarshal([]byte(legacy), &backup); err != nil {
t.Fatal(err)
}
if _, err := r.Import(&backup, []string{"users", "forwards"}); err != nil {
t.Fatal(err)
}
check := func(stage string) {
t.Helper()
rows, err := r.ListForwards()
if err != nil {
t.Fatal(err)
}
if len(rows) != 3 {
t.Fatalf("%s: rows=%+v", stage, rows)
}
for i, want := range []string{"current-owner", "missing-owner", ""} {
if rows[i]["userName"] != want {
t.Fatalf("%s: row %d owner=%v want %q", stage, i, rows[i]["userName"], want)
}
}
if rows[0]["id"] != int64(9301) || rows[0]["userId"] != int64(9103) || rows[0]["inFlow"] != int64(123) || rows[0]["outFlow"] != int64(456) {
t.Fatalf("%s: rule changed: %+v", stage, rows[0])
}
previews, err := r.ListForwardsByTunnel(9201)
if err != nil {
t.Fatal(err)
}
if len(previews) != 3 || previews[0].UserName != "current-owner" || previews[1].UserName != "missing-owner" || previews[2].UserName != "" {
t.Fatalf("%s: preview owners=%+v", stage, previews)
}
exported, err := r.ExportAll()
if err != nil {
t.Fatal(err)
}
if len(exported.Forwards) != 3 || exported.Forwards[0].UserName != "former-owner" {
t.Fatalf("%s: stored legacy snapshot was rewritten: %+v", stage, exported.Forwards)
}
}
check("import")
if err := r.Close(); err != nil {
t.Fatal(err)
}
r = open()
check("restart")
// A forwards-only restore must resolve against the existing user by ID.
if _, err := r.Import(&backup, []string{"forwards"}); err != nil {
t.Fatal(err)
}
check("partial restore")
})
}
}
@@ -0,0 +1,554 @@
package contract_test
import (
"context"
"database/sql"
"fmt"
"net/http"
"net/http/httptest"
"slices"
"strconv"
"strings"
"testing"
"time"
"go-backend/internal/auth"
httpserver "go-backend/internal/http"
"go-backend/internal/http/handler"
"go-backend/internal/security"
"go-backend/internal/store/model"
"go-backend/internal/store/repo"
)
func TestPostgresBackupImportAllowsImmediateCreate(t *testing.T) {
source := ownerNamesRepository(t, "postgres")()
for _, name := range []string{"backup-one", "backup-two"} {
if _, err := source.CreateUser(name, security.MD5("test-password"), 1, 0, 10, 1, 2, 1, 3, 0); err != nil {
t.Fatal(err)
}
if _, err := source.CreateSpeedLimit(name, 100, 0, 1); err != nil {
t.Fatal(err)
}
}
backup, err := source.ExportAll()
if err != nil {
t.Fatal(err)
}
target := ownerNamesRepository(t, "postgres")()
if _, err := target.Import(backup, []string{"users", "speedLimits"}); err != nil {
t.Fatal(err)
}
if id, err := target.CreateUser("after-import", security.MD5("test-password"), 1, 0, 10, 1, 2, 1, 3, 0); err != nil || id <= 2 {
t.Errorf("create user immediately after backup import: id=%d err=%v", id, err)
}
if id, err := target.CreateSpeedLimit("after-import", 100, 0, 1); err != nil || id <= 2 {
t.Errorf("create speed limit immediately after backup import: id=%d err=%v", id, err)
}
}
func TestPostgresBackupImportPreservesAllocatedIDs(t *testing.T) {
for _, called := range []bool{false, true} {
t.Run(fmt.Sprintf("called=%t", called), func(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
// A deleted or rolled-back allocation may be higher than every row.
for _, table := range []string{"user", "speed_limit"} {
if err := r.DB().Exec(`SELECT setval(pg_get_serial_sequence(?, 'id'), 500, ?)`, `"`+table+`"`, called).Error; err != nil {
t.Fatal(err)
}
}
b := &model.BackupData{Users: []model.UserBackup{{ID: 1, User: "restored", Pwd: "$2-test-fixture", RoleID: 1}}, SpeedLimits: []model.SpeedLimitBackup{{ID: 1, Name: "restored"}}}
if _, err := r.Import(b, []string{"users", "speedLimits"}); err != nil {
t.Fatal(err)
}
want := int64(500)
if called {
want++
}
if id, err := r.CreateUser("new", "hash", 1, 0, 0, 0, 0, 1, 0, 0); err != nil || id != want {
t.Fatalf("user id=%d, want %d; err=%v", id, want, err)
}
if id, err := r.CreateSpeedLimit("new", 100, 0, 1); err != nil || id != want {
t.Fatalf("speed limit id=%d, want %d; err=%v", id, want, err)
}
})
}
}
func TestPostgresBackupImportInvalidatesCachedIDs(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
if err := r.DB().Exec(`ALTER SEQUENCE speed_limit_id_seq CACHE 20`).Error; err != nil {
t.Fatal(err)
}
pool, err := r.DB().DB()
if err != nil {
t.Fatal(err)
}
cached, err := pool.Conn(context.Background())
if err != nil {
t.Fatal(err)
}
defer cached.Close()
// Hold this backend out of the pool with unused cached IDs 2..20.
var allocated int64
if err := cached.QueryRowContext(context.Background(), `SELECT nextval('speed_limit_id_seq')`).Scan(&allocated); err != nil {
t.Fatal(err)
}
if allocated != 1 {
t.Fatal(allocated)
}
b := &model.BackupData{SpeedLimits: []model.SpeedLimitBackup{{ID: 2, Name: "restored"}}}
if _, err := r.Import(b, []string{"speedLimits"}); err != nil {
t.Fatal(err)
}
// Return only the cached backend to the pool, so the public create must use it.
pool.SetMaxIdleConns(0)
pool.SetMaxIdleConns(1)
if err := cached.Close(); err != nil {
t.Fatal(err)
}
if id, err := r.CreateSpeedLimit("new", 100, 0, 1); err != nil || id <= 20 {
t.Fatalf("cached ID reused: id=%d err=%v", id, err)
}
}
func TestPostgresBackupImportRollsBackSequenceOnCommitFailure(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
// Deferred validation fails after import has already advanced the sequence.
for _, query := range []string{
`CREATE FUNCTION reject_import_commit() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'reject imported row at commit'; END $$`,
`CREATE CONSTRAINT TRIGGER reject_import_commit AFTER INSERT ON speed_limit DEFERRABLE INITIALLY DEFERRED FOR EACH ROW WHEN (NEW.name = 'fail-at-commit') EXECUTE FUNCTION reject_import_commit()`,
} {
if err := r.DB().Exec(query).Error; err != nil {
t.Fatal(err)
}
}
b := &model.BackupData{SpeedLimits: []model.SpeedLimitBackup{{ID: 100, Name: "fail-at-commit"}}}
if _, err := r.Import(b, []string{"speedLimits"}); err == nil || !strings.Contains(err.Error(), "reject imported row at commit") {
t.Fatalf("expected commit failure, got %v", err)
}
exported, err := r.ExportAll()
if err != nil {
t.Fatal(err)
}
if len(exported.SpeedLimits) != 0 {
t.Fatal("failed import persisted rows")
}
if id, err := r.CreateSpeedLimit("after-failure", 100, 0, 1); err != nil || id != 1 {
t.Fatalf("failed import changed next allocation: id=%d err=%v", id, err)
}
}
func TestPostgresBackupImportPreservesSequenceIncrement(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
if err := r.DB().Exec(`ALTER SEQUENCE speed_limit_id_seq INCREMENT BY 10`).Error; err != nil {
t.Fatal(err)
}
b := &model.BackupData{SpeedLimits: []model.SpeedLimitBackup{{ID: 25, Name: "restored"}}}
if _, err := r.Import(b, []string{"speedLimits"}); err != nil {
t.Fatal(err)
}
if id, err := r.CreateSpeedLimit("new", 100, 0, 1); err != nil || id != 31 {
t.Fatalf("changed ID progression: id=%d, want 31; err=%v", id, err)
}
}
func TestPostgresBackupImportAdvancesEveryExplicitIDTable(t *testing.T) {
source := ownerNamesRepository(t, "postgres")()
types := []string{"users", "speedLimits", "nodes", "tunnels", "forwards", "userTunnels", "tunnelGroups", "userGroups", "permissions", "configs"}
fixture := &model.BackupData{
Users: []model.UserBackup{{ID: 1, User: "old", Pwd: "$2-test-fixture", RoleID: 1}},
SpeedLimits: []model.SpeedLimitBackup{{ID: 1, Name: "old"}},
Nodes: []model.NodeBackup{{ID: 1, Name: "old", Secret: "old-node"}},
Tunnels: []model.TunnelBackup{{ID: 1, Name: "old", ChainTunnels: []model.ChainTunnelBackup{{ID: 1, TunnelID: 1, NodeID: 1, ChainType: "1"}}}},
Forwards: []model.ForwardBackup{{ID: 1, Name: "old", UserID: 1, TunnelID: 1}},
UserTunnels: []model.UserTunnelBackup{{ID: 1, UserID: 1, TunnelID: 1}},
TunnelGroups: []model.TunnelGroupBackup{{ID: 1, Name: "old"}},
UserGroups: []model.UserGroupBackup{{ID: 1, Name: "old"}},
Permissions: []model.PermissionBackup{{ID: 1, UserGroupID: 1, TunnelGroupID: 1, Grants: []model.PermissionGrantBackup{{ID: 1, UserGroupID: 1, TunnelGroupID: 1, UserTunnelID: 1}}}},
}
if _, err := source.Import(fixture, types); err != nil {
t.Fatal(err)
}
backup, err := source.ExportAll()
if err != nil {
t.Fatal(err)
}
r := ownerNamesRepository(t, "postgres")()
if _, err := r.Import(backup, types); err != nil {
t.Fatal(err)
}
// Use the normal public creation methods, including the nested child tables.
for _, create := range []struct {
name string
run func() error
}{
{"user", func() error { _, err := r.CreateUser("new", "hash", 1, 0, 0, 0, 0, 1, 0, 0); return err }},
{"speed limit", func() error { _, err := r.CreateSpeedLimit("new", 100, 0, 1); return err }},
{"node", func() error {
return r.CreateRemoteNode("new", "new-node", "127.0.0.1", "10000-20000", 0, 1, 0, "", "", "")
}},
{"tunnel", func() error { _, err := r.CreateTunnelTx(r.DB(), "new", 1, 1, 0, 0, 1, nil, 0, "", "", 0); return err }},
{"chain tunnel", func() error { return r.CreateChainTunnelTx(r.DB(), 1, "1", 2, sql.NullInt64{}, "", 1, "tls", "") }},
{"forward", func() error {
_, err := r.CreateForwardTx(1, "old", "new", 1, "127.0.0.1:80", "", 0, 0, nil, 0, "", nil, 0, 0, nil, 0, 0, 0)
return err
}},
{"user tunnel", func() error { return r.InsertUserTunnel(2, 1, nil, 0, 0, 0, 0, 1) }},
{"tunnel group", func() error { return r.GroupCreate("tunnel_group", "new", 1, 0) }},
{"user group", func() error { return r.GroupCreate("user_group", "new", 1, 0) }},
{"permission", func() error { return r.InsertGroupPermission(2, 2, 0) }},
{"permission grant", func() error { r.InsertGroupPermissionGrant(1, 1, 2, 1, 0); return nil }},
} {
t.Run(create.name, func(t *testing.T) {
if err := create.run(); err != nil {
t.Fatal(err)
}
})
}
got, err := r.ExportAll()
if err != nil {
t.Fatal(err)
}
ids := map[string][]int64{}
for _, v := range got.Users {
ids["user"] = append(ids["user"], v.ID)
}
for _, v := range got.SpeedLimits {
ids["speed limit"] = append(ids["speed limit"], v.ID)
}
for _, v := range got.Nodes {
ids["node"] = append(ids["node"], v.ID)
}
for _, v := range got.Tunnels {
ids["tunnel"] = append(ids["tunnel"], v.ID)
for _, child := range v.ChainTunnels {
ids["chain tunnel"] = append(ids["chain tunnel"], child.ID)
}
}
for _, v := range got.Forwards {
ids["forward"] = append(ids["forward"], v.ID)
}
for _, v := range got.UserTunnels {
ids["user tunnel"] = append(ids["user tunnel"], v.ID)
}
for _, v := range got.TunnelGroups {
ids["tunnel group"] = append(ids["tunnel group"], v.ID)
}
for _, v := range got.UserGroups {
ids["user group"] = append(ids["user group"], v.ID)
}
for _, v := range got.Permissions {
ids["permission"] = append(ids["permission"], v.ID)
for _, child := range v.Grants {
ids["permission grant"] = append(ids["permission grant"], child.ID)
}
}
if len(ids) != 11 {
t.Fatalf("missing tables in export: %v", ids)
}
for name, values := range ids {
slices.Sort(values)
if !slices.Equal(values, []int64{1, 2}) {
t.Errorf("%s: expected old and new IDs [1 2], got %v", name, values)
}
}
}
func TestPostgresBackupImportKeepsExistingRowsAndEmptySelections(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
// Fixture simulates an existing explicit-ID row with an unsynchronized sequence.
if err := r.DB().Create(&model.SpeedLimit{ID: 500, Name: "existing"}).Error; err != nil {
t.Fatal(err)
}
old := &model.BackupData{SpeedLimits: []model.SpeedLimitBackup{{ID: 1, Name: "legacy"}}}
if _, err := r.Import(old, []string{"speedLimits"}); err != nil {
t.Fatal(err)
}
if id, err := r.CreateSpeedLimit("new", 100, 0, 1); err != nil || id != 501 {
t.Fatalf("existing maximum ignored: id=%d err=%v", id, err)
}
for _, b := range []*model.BackupData{{}, {Users: []model.UserBackup{{ID: 900, User: "not-selected", Pwd: "$2-test-fixture"}}}} {
if _, err := r.Import(b, []string{"speedLimits"}); err != nil {
t.Fatal(err)
}
}
if id, err := r.CreateUser("first-user", "hash", 1, 0, 0, 0, 0, 1, 0, 0); err != nil || id != 2 {
t.Fatalf("unselected sequence changed: id=%d err=%v", id, err)
}
if id, err := r.CreateSpeedLimit("after-empty", 100, 0, 1); err != nil || id != 502 {
t.Fatalf("empty import changed allocation: id=%d err=%v", id, err)
}
got, err := r.ExportAll()
if err != nil {
t.Fatal(err)
}
if len(got.Users) != 2 || len(got.SpeedLimits) != 4 || got.SpeedLimits[1].Name != "existing" {
t.Fatalf("existing or unselected data changed: %+v", got)
}
}
func TestPostgresBackupImportResolvesLegacySequenceAndSearchPath(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
pool, err := r.DB().DB()
if err != nil {
t.Fatal(err)
}
pool.SetMaxOpenConns(1)
var schema string
if err := r.DB().Raw(`SELECT current_schema()`).Scan(&schema).Error; err != nil {
t.Fatal(err)
}
prefix := "Empty-Prefix-" + strconv.FormatInt(time.Now().UnixNano(), 36)
if err := r.DB().Exec(`CREATE SCHEMA "` + prefix + `"`).Error; err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = r.DB().Exec(`DROP SCHEMA "` + prefix + `" CASCADE`).Error })
// The historical default uses a custom, unowned sequence; the original
// owned serial sequence still exists but is no longer used by INSERT.
for _, query := range []string{
`CREATE SEQUENCE "Custom Limit Sequence"`,
`ALTER TABLE speed_limit ALTER COLUMN id SET DEFAULT nextval('"Custom Limit Sequence"'::regclass)`,
`SET search_path TO "` + prefix + `", "` + schema + `"`,
} {
if err := r.DB().Exec(query).Error; err != nil {
t.Fatal(err)
}
}
b := &model.BackupData{SpeedLimits: []model.SpeedLimitBackup{{ID: 1, Name: "legacy"}}}
if _, err := r.Import(b, []string{"speedLimits"}); err != nil {
t.Fatal(err)
}
if id, err := r.CreateSpeedLimit("new", 100, 0, 1); err != nil || id != 2 {
t.Fatalf("wrong sequence or schema: id=%d err=%v", id, err)
}
}
func waitForBackupDBCondition(t *testing.T, r *repo.Repository, query string, args ...interface{}) {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
var ready bool
if err := r.DB().Raw(query, args...).Scan(&ready).Error; err != nil {
t.Fatal(err)
}
if ready {
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatal("database barrier was not reached")
}
func backupImportBarrier(t *testing.T, r *repo.Repository, table, timing, event, condition string) (*sql.Tx, int64) {
t.Helper()
pool, err := r.DB().DB()
if err != nil {
t.Fatal(err)
}
barrier, err := pool.Begin()
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = barrier.Rollback() })
key := time.Now().UnixNano() & 0x3fffffff
if _, err := barrier.Exec(`SELECT pg_advisory_xact_lock($1)`, key); err != nil {
t.Fatal(err)
}
guardEnd := ""
if condition != "" {
guardEnd = "END IF;"
}
function := fmt.Sprintf(`CREATE FUNCTION pause_backup_test() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN %s PERFORM pg_advisory_xact_lock(%d); %s RETURN NEW; END $$`, condition, key, guardEnd)
if err := r.DB().Exec(function).Error; err != nil {
t.Fatal(err)
}
trigger := fmt.Sprintf(`CREATE TRIGGER pause_backup_test %s %s ON %s FOR EACH ROW EXECUTE FUNCTION pause_backup_test()`, timing, event, table)
if err := r.DB().Exec(trigger).Error; err != nil {
t.Fatal(err)
}
return barrier, key
}
func TestPostgresBackupImportSerializesConcurrentCreatesAndImports(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
barrier, key := backupImportBarrier(t, r, "speed_limit", "BEFORE", "INSERT", "IF NEW.name = 'hold-import' THEN")
first := make(chan error, 1)
go func() {
_, err := r.Import(&model.BackupData{SpeedLimits: []model.SpeedLimitBackup{{ID: 100, Name: "hold-import"}}}, []string{"speedLimits"})
first <- err
}()
waitForBackupDBCondition(t, r, `SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype='advisory' AND objid=? AND NOT granted)`, key)
created := make(chan error, 1)
go func() {
id, err := r.CreateSpeedLimit("concurrent-create", 100, 0, 1)
if err == nil && id <= 100 {
err = fmt.Errorf("reused imported ID: %d", id)
}
created <- err
}()
waitForBackupDBCondition(t, r, `SELECT EXISTS (SELECT 1 FROM pg_locks WHERE relation='speed_limit'::regclass AND NOT granted)`)
second := make(chan error, 1)
go func() {
result, err := r.Import(&model.BackupData{SpeedLimits: []model.SpeedLimitBackup{{ID: 200, Name: "second-import"}}}, []string{"speedLimits"})
if err == nil && result.SpeedLimitsImported != 1 {
err = fmt.Errorf("wrong retry count: %+v", result)
}
second <- err
}()
if err := barrier.Commit(); err != nil {
t.Fatal(err)
}
for name, done := range map[string]<-chan error{"first import": first, "create": created, "second import": second} {
select {
case err := <-done:
if err != nil {
t.Errorf("%s: %v", name, err)
}
case <-time.After(3 * time.Second):
t.Fatalf("%s blocked", name)
}
}
if id, err := r.CreateSpeedLimit("after-both", 100, 0, 1); err != nil || id <= 200 {
t.Fatalf("second import sequence not advanced: id=%d err=%v", id, err)
}
got, err := r.ExportAll()
if err != nil {
t.Fatal(err)
}
if len(got.SpeedLimits) != 4 {
t.Fatalf("lost concurrent rows: %+v", got.SpeedLimits)
}
}
func TestPostgresBackupImportAvoidsCascadeDeleteDeadlock(t *testing.T) {
for _, mode := range []string{"retry-success", "busy-bounded"} {
t.Run(mode, func(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
user, err := r.CreateUser("delete-me", "hash", 1, 0, 0, 0, 0, 1, 0, 0)
if err != nil {
t.Fatal(err)
}
forward, err := r.CreateForwardTx(user, "delete-me", "old", 1, "127.0.0.1:80", "", 0, 0, []int64{1}, 12345, "", nil, 0, 0, nil, 0, 0, 0)
if err != nil {
t.Fatal(err)
}
// Pause a real public cascade after deleting a child row, before it
// reaches forward/user. Import must not retain partial parent locks.
barrier, key := backupImportBarrier(t, r, "forward_port", "AFTER", "DELETE", "")
deleted := make(chan error, 1)
go func() { deleted <- r.DeleteUserCascade(user) }()
waitForBackupDBCondition(t, r, `SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype='advisory' AND objid=? AND NOT granted)`, key)
emptyPorts := []model.ForwardPortBackup{}
backup := &model.BackupData{
Users: []model.UserBackup{{ID: 100, User: "restored", Pwd: "$2-test-fixture", RoleID: 1}},
Forwards: []model.ForwardBackup{{ID: forward, UserID: 100, Name: "restored", ForwardPorts: &emptyPorts}},
}
imported := make(chan error, 1)
runImport := func() {
result, err := r.Import(backup, []string{"users", "forwards"})
if err == nil && (result.UsersImported != 1 || result.ForwardsImported != 1) {
err = fmt.Errorf("wrong retry count: %+v", result)
}
imported <- err
}
go runImport()
if mode == "busy-bounded" {
select {
case err := <-imported:
if err == nil || !strings.Contains(err.Error(), "database is busy") {
t.Fatalf("expected bounded busy error, got %v", err)
}
case <-time.After(3 * time.Second):
_ = barrier.Rollback()
t.Fatal("import waited while holding partial table locks")
}
} else {
// The held child lock makes every attempt fail before writes.
// Let several bounded attempts encounter it before release.
time.Sleep(75 * time.Millisecond)
}
if err := barrier.Commit(); err != nil {
t.Fatal(err)
}
select {
case err := <-deleted:
if err != nil {
t.Fatalf("cascade: %v", err)
}
case <-time.After(3 * time.Second):
t.Fatal("cascade deadlocked")
}
if mode == "busy-bounded" {
go runImport()
}
select {
case err := <-imported:
if err != nil {
t.Fatalf("import: %v", err)
}
case <-time.After(3 * time.Second):
t.Fatal("import deadlocked")
}
if id, err := r.CreateUser("after", "hash", 1, 0, 0, 0, 0, 1, 0, 0); err != nil || id <= 100 {
t.Fatalf("create after concurrent cascade: id=%d err=%v", id, err)
}
})
}
}
func TestPostgresBackupImportRollsBackEarlierSequenceWhenLaterSequenceFails(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
if err := r.DB().Exec(`ALTER SEQUENCE speed_limit_id_seq MAXVALUE 10`).Error; err != nil {
t.Fatal(err)
}
backup := &model.BackupData{
Users: []model.UserBackup{{ID: 100, User: "rollback-user", Pwd: "$2-test-fixture", RoleID: 1}},
SpeedLimits: []model.SpeedLimitBackup{{ID: 10, Name: "exhausted"}},
}
// User sequence advances first. The exhausted limit sequence then fails.
if _, err := r.Import(backup, []string{"users", "speedLimits"}); err == nil || !strings.Contains(err.Error(), "sequence exhausted") {
t.Fatalf("expected sequence exhaustion: %v", err)
}
if id, err := r.CreateUser("after-failure", "hash", 1, 0, 0, 0, 0, 1, 0, 0); err != nil || id != 2 {
t.Fatalf("earlier sequence failed to roll back: id=%d err=%v", id, err)
}
if id, err := r.CreateSpeedLimit("after-failure", 100, 0, 1); err != nil || id != 1 {
t.Fatalf("later sequence changed: id=%d err=%v", id, err)
}
got, err := r.ExportAll()
if err != nil {
t.Fatal(err)
}
if len(got.Users) != 2 || len(got.SpeedLimits) != 1 {
t.Fatalf("partial import committed: %+v", got)
}
}
func TestPostgresBackupImportAPIAllowsImmediateCreate(t *testing.T) {
r := ownerNamesRepository(t, "postgres")()
const secret = "backup-sequence-contract"
router := httpserver.NewRouter(handler.New(r, secret), secret)
token, err := auth.GenerateToken(1, "admin_user", 0, secret)
if err != nil {
t.Fatal(err)
}
post := func(path, body string) {
t.Helper()
request := httptest.NewRequest(http.MethodPost, path, strings.NewReader(body))
request.Header.Set("Authorization", token)
request.Header.Set("Content-Type", "application/json")
response := httptest.NewRecorder()
router.ServeHTTP(response, request)
assertCode(t, response, 0)
}
post("/api/v1/backup/import", `{"version":"1.0","types":["users","speedLimits"],"users":[{"id":2,"user":"restored","pwd":"$2-test-fixture","roleId":1,"status":1}],"speedLimits":[{"id":1,"name":"restored","speed":100,"status":1}]}`)
post("/api/v1/user/create", `{"user":"after-import","pwd":"test-only-password","status":1}`)
post("/api/v1/speed-limit/create", `{"name":"after-import","speed":200}`)
got, err := r.ExportAll()
if err != nil {
t.Fatal(err)
}
if len(got.Users) != 3 || got.Users[2].ID != 3 || len(got.SpeedLimits) != 2 || got.SpeedLimits[1].ID != 2 {
t.Fatalf("API create did not retain restored records: %+v", got)
}
}
@@ -0,0 +1,98 @@
package contract_test
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"go-backend/internal/auth"
)
func TestTunnelRedeployStreamContract(t *testing.T) {
const secret = "contract-jwt-secret"
router, _ := setupContractRouter(t, secret)
token, err := auth.GenerateToken(1, "admin_user", 0, secret)
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/batch-redeploy/stream", strings.NewReader(`{"ids":[999998,999999]}`))
req.Header.Set("Authorization", token)
res := httptest.NewRecorder()
router.ServeHTTP(res, req)
if !strings.HasPrefix(res.Header().Get("Content-Type"), "application/x-ndjson") {
t.Fatalf("expected streamed progress, got %s: %s", res.Header().Get("Content-Type"), res.Body.String())
}
decoder := json.NewDecoder(res.Body)
for _, want := range []string{"start", "processing", "item", "processing", "item", "done"} {
var event struct {
Type string `json:"type"`
Data json.RawMessage `json:"data"`
}
if err := decoder.Decode(&event); err != nil {
t.Fatal(err)
}
if event.Type != want {
t.Fatalf("got %s, want %s", event.Type, want)
}
if want == "item" {
var data struct{ Result map[string]interface{} }
if err := json.Unmarshal(event.Data, &data); err != nil {
t.Fatal(err)
}
if _, ok := data.Result["name"].(string); !ok {
t.Fatalf("item name must be a string, including missing tunnels: %s", event.Data)
}
}
if want == "done" {
var data struct {
Progress struct{ Total, Completed, Success, Failed int }
Result struct {
SuccessCount, FailCount int
Failures []struct {
ID int64
Reason string
}
}
}
if err := json.Unmarshal(event.Data, &data); err != nil {
t.Fatal(err)
}
if data.Progress.Total != 2 || data.Progress.Completed != 2 || data.Progress.Failed != 2 || data.Progress.Success != 0 || data.Result.FailCount != 2 || len(data.Result.Failures) != 2 {
t.Fatalf("incorrect completion: %+v", data)
}
if data.Result.Failures[0].Reason == "" {
t.Fatal("missing failure detail")
}
}
}
if !res.Flushed {
t.Fatal("stream did not flush")
}
}
func TestTunnelRedeployStreamAuthorizationContract(t *testing.T) {
const secret = "contract-jwt-secret"
router, repository := setupContractRouter(t, secret)
seedLegacyUser(t, repository, 9111, "stream-normal-user", "password")
token, err := auth.GenerateToken(9111, "stream-normal-user", 1, secret)
if err != nil {
t.Fatal(err)
}
for _, tc := range []struct {
name, token string
code int
}{{"anonymous", "", 401}, {"non-admin", token, 403}} {
t.Run(tc.name, func(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/batch-redeploy/stream", strings.NewReader(`{"ids":[999999]}`))
req.Header.Set("Authorization", tc.token)
res := httptest.NewRecorder()
router.ServeHTTP(res, req)
assertCode(t, res, tc.code)
if strings.Contains(res.Body.String(), `"type":"start"`) {
t.Fatal("unauthorized request started stream")
}
})
}
}
+98
View File
@@ -0,0 +1,98 @@
# FLVX #574 / #575: current owner names and custom order
Status: implemented and verified locally on 2026-11-10. The user confirmed the three test boundaries before implementation and later authorized the PostgreSQL import repair discovered during validation.
Base: `3209ff8a59bf02287e6934d0b80b941006db239e` (main, verified 2026-11-10).
Sources: https://github.com/Sagit-chu/flvx/issues/574 and https://github.com/Sagit-chu/flvx/issues/575.
## Problem statement
Existing forwarding rules retain the username copied when they were created. Renaming an account leaves the rules page's owner filter and related displays showing the old name, including after restart. Users and speed limits have only ID-descending order, so an administrator cannot arrange them intentionally.
## Solution
Resolve a forwarding rule's current owner name by immutable user ID without recreating rules. Add administrator-managed global drag order for users and speed limits, persisted on the server across clients. Newly created entries appear first. Searching disables dragging so a partial result cannot overwrite the complete order.
## User stories
1. As an administrator, I see the current owner name on existing rules after renaming a user.
2. As a user changing my account name and password, my existing rules display the new name.
3. As an administrator, I see consistent names in rule groups, owner filters and tunnel deletion previews.
4. As an administrator restoring an old backup, I do not have to delete or recreate rules to correct owner names.
5. As a user, a different account reusing my former name cannot acquire my rules or permissions.
6. As an administrator, I can drag ordinary users into a global order in both card and table views.
7. As an administrator, I can drag speed limits into a global order in both card and table views.
8. As an administrator switching browsers or signing in again, I see the server's saved order.
9. As an administrator creating an item, I find it at the top; editing it keeps its position.
10. As an administrator searching a list, I see why dragging is unavailable and hidden items retain their positions.
11. As an administrator with more than ten users, I do not lose entries from the reorder set after creating a user.
12. As an administrator whose save fails or conflicts, I see the failure and recover the authoritative order rather than a false success.
13. As an operator upgrading an existing SQLite or PostgreSQL database, I retain the prior default order until I rearrange it.
14. As an operator restoring a new backup, I retain its order; older backups remain importable.
15. As a user, an administrative display-order change does not change my password, ownership, permissions, traffic totals, limits, or rule bindings.
16. As an operator restoring PostgreSQL data, I can immediately create new records without restarting, with existing sequence allocations preserved and failed imports rolled back.
## Implementation decisions
- Rule ownership remains numeric user ID. Display uses the current associated username, with the stored name (and existing unknown-owner UI) as a compatibility fallback when no associated user exists.
- Keep the legacy stored name field and backup format compatible. Display correction must work for stale data without a destructive rewrite or rule recreation.
- Cover all relevant presentation projections, not just the owner selector. Do not change data-plane service identity or authentication behavior.
- Add integer ordering to ordinary users and speed limits. Preserve old ID-descending order for tied/default ranks, set new items before the current first rank, and preserve rank on ordinary edits.
- Add administrator-only reorder endpoints for both resources. A request supplies the complete ordered ID set and its previously observed order so stale sets/orders can be rejected safely; return an actionable conflict/error rather than partial success.
- Validate input as a complete permutation of eligible current resources. Reject malformed IDs, duplicates, missing/extra IDs and administrator user IDs. All writes are transactional.
- Serialize client saves or disable dragging while a save is outstanding. On failure/conflict, reload the server's order. Cancel or ignore stale list requests when search/navigation changes.
- Reuse existing drag sensors and affordances. Support mouse, touch and keyboard. Keep delete/edit controls usable and prevent accidental drag activation.
- Search disables drag. User creation reloads the authoritative complete list, avoiding the existing ten-item truncation.
- Preserve existing API envelopes and raw JWT convention. Explicitly protect the new user reorder path in authorization middleware.
- User and speed-limit backup DTOs, export mappings and import updates carry order. Missing old-backup fields must not accidentally zero an existing custom order. New records imported without rank follow a documented deterministic rule.
- No new frontend test framework; use the repository's build/lint and isolated browser checks.
## Testing decisions (confirmed boundaries)
1. **Public HTTP API**: current-name responses, all rename entry points, ownership isolation, admin-only reorder, complete-set validation, failures/conflicts, and list/create/edit/delete behavior. This catches application behavior but not browser interactions or every database upgrade path.
2. **Repository lifecycle and backup public interfaces**: fresh/legacy databases, restart persistence, old/new backup round trips and partial import behavior for SQLite and PostgreSQL. This catches persisted compatibility but not UI behavior. Observe through public reads/exports rather than implementation-coupled assertions where possible; legacy schema fixtures may require direct fixture setup.
3. **Rendered browser UI**: table/card drag, touch/keyboard affordances, search-disabled state, recovery from failed saves, repeat actions, and navigation. This catches actual UI wiring, not production connectivity or real forwarding.
Use Matt Pocock's vertical red-to-green cycles: one failing behavior test, minimum implementation, then the next behavior. Prefer existing contract fixtures and public interfaces. Review standards and requirements independently after implementation. These boundaries were confirmed by the user before implementation tests were added.
## Work graph
- **T1 Current owner names**: implement #574 with API regression coverage and historical/backup cases. Independent.
- **T2 Persistent order backend**: implement ordering models, migrations, endpoints, permissions, transaction/conflict handling and backup compatibility. Runs in an isolated worktree; integrate after T1 and resolve any shared repository-layer edits before reporting complete.
- **T3 Drag order frontend**: implement both list views and failure/search behavior against the agreed API contract. Can proceed after T2 contract is pinned; no backend edits.
- **T4 Verify and review**: integrate all slices, full backend checks, frontend checks, isolated SQLite/PG/browser verification, independent standards/spec reviews and fixes. Depends on T1–T3. Complete.
- **T5 PostgreSQL import sequences**: user-authorized follow-on repair for the pre-existing explicit-ID import bug. Import takes the complete set of write-table locks with NOWAIT before any data writes, retries only pre-write lock contention within a bounded budget, and advances actual ID sequences transactionally without moving allocated IDs backward. Complete.
## Out of scope
Production access or changes, external comments/issues/spec publication, personal per-admin preferences, drag while searching, server pagination/cross-page drag, field-based sorting, changing actual speed-limit enforcement, agent changes, and unrelated refactors. Push/PR/merge/release were excluded from implementation and authorized separately afterward as recorded below.
## Further notes
Work is isolated on `fix/user-names-custom-order`; original checkout remains untouched. Local commits were permitted by the implementation workflow; publication required the separate authorization recorded below. Existing assessment reproductions are available in a temporary isolated checkout for reference, not as final tests. PostgreSQL testing used an isolated, disposable local PostgreSQL 16.15 container with per-test schemas. No production database or user data was used.
## Verification results
Final implementation merge: `2ad30e4b2f587b0118cb53049194c915fb32565e`.
- Full backend suite passed with PostgreSQL enabled and the cache disabled: `FLVX_POSTGRES_TEST_DSN=... go test -count=1 ./...`.
- Race-enabled owner-name, display-order and PostgreSQL backup-import contracts passed. The new PostgreSQL import contracts also passed three repeated race-enabled runs in their implementation worktree.
- `go vet ./...`, native backend build and Linux amd64/arm64 cross-builds passed using Go 1.26.10.
- Frontend `tsc --noEmit`, full `eslint .`, and production build passed using Node 24.22.1. The build still reports the existing outdated Browserslist data notice.
- SQLite and PostgreSQL contracts cover three rename paths, old-name reuse/ownership isolation, historical fallback, previews, restart and old/new backups. Order contracts cover administrator checks, complete permutations, stale order/membership, concurrent creates/reorders, new-first and edit-preserved rank.
- PostgreSQL import tests reproduce the prior 23505 failure and cover immediate public API creates without reopening the repository, all 11 explicit-ID tables, allocated/cached IDs, non-unit positive increments, unowned legacy defaults, search-path resolution, concurrent imports/creates, cascade-delete contention, bounded busy recovery, intermediate-sequence failure and commit-time rollback. Removing NOWAIT makes the contention regression fail.
- Chrome tests against synthetic localhost data passed: user and limit card keyboard reorder; user and limit table pointer reorder; 390px viewport touch reorder; refresh persistence; 13-user creation without truncation; new-first and edit-preserved order; search-disabled handles; drag cancellation; a second client's conflict followed by authoritative recovery; transport failure and successful load-error retry.
- The rendered rules page and owner filter show the current username for a historical rule whose stored snapshot still has the former name.
- Restarting the final backend binary preserved the 13-user order, four-limit order and current-name projection.
- Independent standards and requirements reviews passed after fixes. The recovered-load error is now cleared without hiding save warnings. The PostgreSQL lock-order and legacy-default findings were resolved and re-reviewed.
## Validation limits
- Linux amd64/arm64 binaries were cross-built, not run on those target machines. No live forwarding agents, production deployment, real traffic, Safari or physical mobile device were exercised. Mobile touch was tested through Chrome's emulated viewport and touch input.
- PostgreSQL identity-column fallback was reviewed but does not have a dedicated fixture. Normal repository writes and imports are covered; external SQL that reserves sequence values inside an open transaction and later acquires tables in reverse order, or concurrent manual sequence-definition changes, are outside the tested contract. Negative/non-positive ID sequences are rejected with a rolled-back import.
- Browser logs included existing React Router/dialog accessibility warnings and a password-manager-extension error. A development HMR hook-order error during the hook edit cleared on full reload; final page flows did not reproduce it.
- At completion of local verification, all changes remained on a local branch. No push, pull request, issue comment, release or production change had been performed.
## Publication stage
After reviewing the PostgreSQL identity-column testing boundary, the user separately authorized pushing the branch, creating and merging a pull request, and publishing a new tag/release on 2026-11-10. Publication must respect required checks and branch protection. The tag workflow publishes versioned/latest images and release attachments; it does not deploy production services. Production access or changes remain out of scope.
+14
View File
@@ -71,6 +71,20 @@ http {
}
# API代理到后端
location = /api/v1/tunnel/batch-redeploy/stream {
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_http_version 1.1;
proxy_buffering off;
proxy_cache off;
chunked_transfer_encoding on;
proxy_read_timeout 120s;
proxy_send_timeout 120s;
proxy_pass http://backend:6365/api/v1/tunnel/batch-redeploy/stream;
}
location = /api/v1/tunnel/diagnose/stream {
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
+10
View File
@@ -619,3 +619,13 @@ export const assignMonitorPermission = (userId: number) =>
export const removeMonitorPermission = (userId: number) =>
Network.post("/monitor/permission/remove", { userId });
// Complete ordered IDs plus the last observed server order for conflict detection.
export interface OrderUpdatePayload {
ids: number[];
expectedIds: number[];
}
export const updateUserOrder = (data: OrderUpdatePayload) =>
Network.post("/user/update-order", data);
export const updateSpeedLimitOrder = (data: OrderUpdatePayload) =>
Network.post("/speed-limit/update-order", data);
@@ -0,0 +1,287 @@
import type { BatchOperationResult } from "@/api/types";
import axios from "axios";
import { clearSession, getToken } from "@/utils/session";
const STREAM_IDLE_TIMEOUT_MS = 2 * 60 * 1000;
const MAX_EVENT_LENGTH = 1024 * 1024;
const UNCERTAIN_RESULT =
"下发进度连接中断,部分配置可能已执行,结果尚未确认。请核对节点与隧道状态后再决定是否重新下发,系统不会自动重试。";
export interface TunnelRedeployProgress {
total: number;
completed: number;
success: number;
failed: number;
}
export interface TunnelRedeployProcessing {
id: number;
name: string;
progress: TunnelRedeployProgress;
}
export interface TunnelRedeployItem {
result: {
id: number;
name: string;
success: boolean;
reason?: string;
};
progress: TunnelRedeployProgress;
}
interface TunnelRedeployCallbacks {
onStart?: (progress: TunnelRedeployProgress) => void;
onProcessing?: (payload: TunnelRedeployProcessing) => void;
onItem?: (payload: TunnelRedeployItem) => void;
}
type RawObject = Record<string, unknown>;
class RequestRejectedError extends Error {}
const object = (value: unknown): RawObject => {
if (!value || typeof value !== "object" || Array.isArray(value)) {
throw new Error(UNCERTAIN_RESULT);
}
return value as RawObject;
};
const readProgress = (value: unknown): TunnelRedeployProgress => {
const candidate = object(value);
const { total, completed, success, failed } = candidate;
if (
![total, completed, success, failed].every(
(count) =>
typeof count === "number" && Number.isSafeInteger(count) && count >= 0,
) ||
(completed as number) > (total as number) ||
(success as number) + (failed as number) !== completed
) {
throw new Error(UNCERTAIN_RESULT);
}
return { total, completed, success, failed } as TunnelRedeployProgress;
};
const expireSession = () => {
clearSession();
if (window.location.pathname !== "/") window.location.href = "/";
};
/** A mutation stream must never be retried or replayed through the legacy API. */
export const batchRedeployTunnelsStream = async (
ids: number[],
callbacks: TunnelRedeployCallbacks,
signal?: AbortSignal,
): Promise<BatchOperationResult> => {
if (signal?.aborted) throw new DOMException("Aborted", "AbortError");
if (!ids.length || ids.some((id) => !Number.isSafeInteger(id) || id <= 0)) {
throw new Error("请选择有效的隧道");
}
if (typeof fetch !== "function" || typeof TextDecoder === "undefined") {
throw new Error("当前浏览器不支持实时下发进度,请使用支持流式响应的浏览器");
}
const requestedIds = new Set(ids);
const controller = new AbortController();
const abort = () => controller.abort();
let timer: ReturnType<typeof setTimeout> | undefined;
let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
const resetTimeout = () => {
clearTimeout(timer);
timer = setTimeout(abort, STREAM_IDLE_TIMEOUT_MS);
};
signal?.addEventListener("abort", abort, { once: true });
resetTimeout();
try {
const base = axios.defaults.baseURL || "/api/v1/";
const response = await fetch(
`${base.endsWith("/") ? base : `${base}/`}tunnel/batch-redeploy/stream`,
{
method: "POST",
headers: {
"Content-Type": "application/json",
Accept: "application/x-ndjson, application/json",
Authorization: getToken() || "",
},
body: JSON.stringify({ ids: Array.from(requestedIds) }),
signal: controller.signal,
},
);
if (response.status === 401) {
expireSession();
throw new RequestRejectedError("登录已过期,请重新登录");
}
if (response.status === 404 || response.status === 405) {
throw new RequestRejectedError(
"后端尚不支持实时下发进度,请升级后端后再操作;未自动改用旧接口下发",
);
}
if (
!response.headers.get("content-type")?.includes("application/x-ndjson")
) {
let payload: RawObject | undefined;
try {
payload = object(await response.json());
} catch {
throw new Error(UNCERTAIN_RESULT);
}
if (payload.code === 401) {
expireSession();
throw new RequestRejectedError("登录已过期,请重新登录");
}
if (typeof payload.code === "number" && payload.code !== 0) {
throw new RequestRejectedError(
typeof payload.msg === "string" ? payload.msg : "下发请求失败",
);
}
throw new Error(UNCERTAIN_RESULT);
}
if (!response.ok || !response.body) throw new Error(UNCERTAIN_RESULT);
reader = response.body.getReader();
const decoder = new TextDecoder("utf-8", { fatal: true });
let buffer = "";
let progress: TunnelRedeployProgress | undefined;
let currentId: number | undefined;
const completedIds = new Set<number>();
let finalResult: BatchOperationResult | undefined;
const processLine = (line: string) => {
if (!line.trim()) return;
const event = object(JSON.parse(line));
const data = object(event.data);
const next = readProgress(event.type === "start" ? data : data.progress);
if (
next.total !== requestedIds.size ||
(progress &&
(next.success < progress.success || next.failed < progress.failed))
)
throw new Error(UNCERTAIN_RESULT);
if (event.type === "start") {
if (progress || next.completed !== 0) throw new Error(UNCERTAIN_RESULT);
progress = next;
callbacks.onStart?.(next);
return;
}
if (!progress) throw new Error(UNCERTAIN_RESULT);
if (event.type === "processing") {
if (
typeof data.id !== "number" ||
!requestedIds.has(data.id) ||
typeof data.name !== "string" ||
completedIds.has(data.id) ||
currentId !== undefined ||
next.completed !== progress.completed
)
throw new Error(UNCERTAIN_RESULT);
currentId = data.id;
callbacks.onProcessing?.({
id: data.id,
name: data.name,
progress: next,
});
} else if (event.type === "item") {
const result = object(data.result);
if (
result.id !== currentId ||
currentId === undefined ||
typeof result.name !== "string" ||
typeof result.success !== "boolean" ||
(result.reason !== undefined && typeof result.reason !== "string") ||
next.completed !== progress.completed + 1 ||
next.success !== progress.success + (result.success ? 1 : 0) ||
next.failed !== progress.failed + (result.success ? 0 : 1)
)
throw new Error(UNCERTAIN_RESULT);
completedIds.add(currentId);
currentId = undefined;
callbacks.onItem?.({
result: result as unknown as TunnelRedeployItem["result"],
progress: next,
});
} else if (event.type === "done") {
const result = object(data.result);
const failures = result.failures ?? [];
if (
currentId !== undefined ||
next.completed !== next.total ||
next.completed !== completedIds.size ||
next.completed !== progress.completed ||
result.successCount !== next.success ||
result.failCount !== next.failed ||
!Array.isArray(failures) ||
failures.length !== next.failed ||
failures.some((failure) => {
const item = object(failure);
return (
typeof item.id !== "number" ||
!completedIds.has(item.id) ||
typeof item.reason !== "string" ||
(item.name !== undefined && typeof item.name !== "string")
);
})
)
throw new Error(UNCERTAIN_RESULT);
finalResult = result as unknown as BatchOperationResult;
} else {
throw new Error(UNCERTAIN_RESULT);
}
progress = next;
};
while (!finalResult) {
const { value, done } = await reader.read();
if (done) {
buffer += decoder.decode();
if (buffer.trim()) processLine(buffer);
break;
}
resetTimeout();
buffer += decoder.decode(value, { stream: true });
let newline: number;
while ((newline = buffer.indexOf("\n")) >= 0) {
const line = buffer.slice(0, newline);
buffer = buffer.slice(newline + 1);
if (line.length > MAX_EVENT_LENGTH) throw new Error(UNCERTAIN_RESULT);
processLine(line);
if (finalResult) break;
}
if (buffer.length > MAX_EVENT_LENGTH) throw new Error(UNCERTAIN_RESULT);
}
if (!finalResult) throw new Error(UNCERTAIN_RESULT);
return finalResult;
} catch (error) {
if (signal?.aborted) throw new DOMException("Aborted", "AbortError");
if (error instanceof RequestRejectedError) throw error;
throw new Error(UNCERTAIN_RESULT);
} finally {
clearTimeout(timer);
controller.abort();
signal?.removeEventListener("abort", abort);
if (reader) {
// Cancel only the response reader, never resend an interrupted mutation.
await reader.cancel().catch(() => undefined);
reader.releaseLock();
}
}
};
@@ -0,0 +1,151 @@
import { createContext, useContext, type ReactNode } from "react";
import {
DndContext,
closestCenter,
KeyboardSensor,
MouseSensor,
TouchSensor,
useSensor,
useSensors,
type DragEndEvent,
} from "@dnd-kit/core";
import {
SortableContext,
rectSortingStrategy,
sortableKeyboardCoordinates,
useSortable,
} from "@dnd-kit/sortable";
import { CSS } from "@dnd-kit/utilities";
import { GripVertical } from "lucide-react";
import { TableRow } from "@/shadcn-bridge/heroui/table";
const HandleContext = createContext<ReactNode>(null);
export function OrderHandle() {
return <>{useContext(HandleContext)}</>;
}
export function OrderItem({
id,
label,
disabled,
row = false,
children,
}: {
id: number;
label: string;
disabled: boolean;
row?: boolean;
children: ReactNode;
}) {
const {
attributes,
listeners,
setNodeRef,
setActivatorNodeRef,
transform,
transition,
isDragging,
} = useSortable({ id, disabled });
const Component = row ? TableRow : "div";
return (
<HandleContext.Provider
value={
<button
ref={setActivatorNodeRef}
{...attributes}
{...listeners}
aria-label={`拖动排序:${label}`}
className={`inline-flex items-center justify-center shrink-0 rounded text-default-500 hover:bg-default-100 focus-visible:outline focus-visible:outline-2 focus-visible:outline-primary disabled:opacity-40 disabled:cursor-not-allowed cursor-grab active:cursor-grabbing ${row ? "p-1" : "h-8 w-8 [@media(pointer:coarse)]:h-11 [@media(pointer:coarse)]:w-11"}`}
disabled={disabled}
style={{ touchAction: "none" }}
type="button"
>
<GripVertical aria-hidden="true" className="h-4 w-4" />
</button>
}
>
<Component
ref={setNodeRef}
className={row ? undefined : "h-full min-w-0"}
style={{
transform: CSS.Transform.toString(transform),
transition,
opacity: isDragging ? 0.5 : 1,
position: "relative",
zIndex: isDragging ? 1 : undefined,
}}
>
{children}
</Component>
</HandleContext.Provider>
);
}
export function OrderList({
ids,
children,
onDragStart,
onDragEnd,
}: {
ids: number[];
children: ReactNode;
onDragStart: () => void;
onDragEnd: (event: DragEndEvent) => void;
}) {
const sensors = useSensors(
useSensor(MouseSensor, { activationConstraint: { distance: 8 } }),
useSensor(TouchSensor, {
activationConstraint: { delay: 250, tolerance: 8 },
}),
useSensor(KeyboardSensor, {
coordinateGetter: sortableKeyboardCoordinates,
}),
);
return (
<DndContext
accessibility={{
screenReaderInstructions: {
draggable: "按空格开始排序,方向键移动,再按空格保存,Escape 取消。",
},
}}
collisionDetection={closestCenter}
sensors={sensors}
onDragEnd={onDragEnd}
onDragStart={onDragStart}
>
<SortableContext items={ids} strategy={rectSortingStrategy}>
{children}
</SortableContext>
</DndContext>
);
}
export function OrderStatus({
reason,
error,
retry,
}: {
reason: string;
error: string;
retry: () => void;
}) {
return (
<div className="mb-3 text-sm text-default-500">
<p role="status">
{reason || "拖动手柄调整全局顺序;键盘可按空格、方向键、空格完成排序。"}
</p>
{error && (
<p className="text-danger mt-2" role="alert">
{error}{" "}
<button className="underline" type="button" onClick={retry}>
重新加载
</button>
</p>
)}
</div>
);
}
+122
View File
@@ -0,0 +1,122 @@
import type { DragEndEvent } from "@dnd-kit/core";
import type { OrderUpdatePayload } from "@/api";
import { useCallback, useEffect, useRef, useState } from "react";
import { arrayMove } from "@dnd-kit/sortable";
interface Result {
code: number;
msg?: string;
transportError?: boolean;
}
// Keep one complete, authoritative snapshot. Filtering is presentation-only.
export function useServerOrder<T extends { id: number }>(
fetchItems: () => Promise<T[]>,
saveOrder: (payload: OrderUpdatePayload) => Promise<Result>,
) {
const [items, setItems] = useState<T[]>([]);
const [loading, setLoading] = useState(true);
const [saving, setSaving] = useState(false);
const [loadError, setLoadError] = useState("");
const [saveWarning, setSaveWarning] = useState("");
const mounted = useRef(false);
const request = useRef(0);
const savingRef = useRef(false);
const snapshot = useRef(items);
snapshot.current = items;
const dragSnapshot = useRef<T[] | null>(null);
const reload = useCallback(async () => {
const version = ++request.current;
setLoading(true);
try {
const rows = await fetchItems();
if (!mounted.current || version !== request.current) return;
setItems(rows);
setLoadError("");
} catch {
if (!mounted.current || version !== request.current) return;
// Never expose a partial or unverified set as reorderable after failure.
setItems([]);
setLoadError("加载列表失败,请重试后再排序。");
} finally {
if (mounted.current && version === request.current) setLoading(false);
}
}, [fetchItems]);
useEffect(() => {
mounted.current = true;
void reload();
return () => {
mounted.current = false;
++request.current;
};
}, [reload]);
const onDragStart = () => {
dragSnapshot.current = snapshot.current;
};
const reorder = async ({ active, over }: DragEndEvent, disabled: boolean) => {
const before = dragSnapshot.current;
dragSnapshot.current = null;
if (
disabled ||
loading ||
savingRef.current ||
!over ||
active.id === over.id ||
before !== snapshot.current
)
return;
const from = items.findIndex((item) => item.id === active.id);
const to = items.findIndex((item) => item.id === over.id);
if (from < 0 || to < 0) return;
savingRef.current = true;
setSaving(true);
setSaveWarning("");
try {
const result = await saveOrder({
ids: arrayMove(items, from, to).map((item) => item.id),
expectedIds: items.map((item) => item.id),
});
if (!mounted.current) return;
if (result.code !== 0) {
setSaveWarning(
result.transportError
? "排序保存未确认,请检查重新加载后的服务器顺序再重试。"
: result.code === 409
? "列表已被其他操作更改,排序未保存,请按重新加载后的顺序重试。"
: `排序保存失败:${result.msg || "请重试"}。请检查重新加载后的服务器顺序。`,
);
}
} catch {
if (mounted.current)
setSaveWarning("排序保存未确认,请检查重新加载后的服务器顺序再重试。");
} finally {
if (mounted.current) {
await reload();
if (mounted.current) setSaving(false);
}
savingRef.current = false;
}
};
return {
items,
setItems,
loading,
saving,
error: [saveWarning, loadError].filter(Boolean).join(" "),
reload,
onDragStart,
reorder,
};
}
+219 -187
View File
@@ -1,12 +1,17 @@
import { useState, useEffect, useMemo } from "react";
import { useState, useMemo } from "react";
import toast from "react-hot-toast";
import { LayoutGrid, List } from "lucide-react";
import {
AnimatedPage,
StaggerList,
StaggerItem,
} from "@/components/animated-page";
OrderHandle,
OrderItem,
OrderList,
OrderStatus,
} from "@/components/server-order";
import { useServerOrder } from "@/hooks/use-server-order";
import { getOrderDisabledReason } from "@/utils/server-order";
import { isAdmin } from "@/utils/auth";
import { AnimatedPage, StaggerList } from "@/components/animated-page";
import { SearchBar } from "@/components/search-bar";
import { Card, CardBody, CardHeader } from "@/shadcn-bridge/heroui/card";
import { Button } from "@/shadcn-bridge/heroui/button";
@@ -24,11 +29,11 @@ import {
TableHeader,
TableColumn,
TableBody,
TableRow,
TableCell,
} from "@/shadcn-bridge/heroui/table";
import {
createSpeedLimit,
updateSpeedLimitOrder,
getSpeedLimitList,
updateSpeedLimit,
deleteSpeedLimit,
@@ -52,9 +57,17 @@ interface SpeedLimitForm {
status: number;
}
const fetchOrderedRules = async (): Promise<SpeedLimitRule[]> => {
const response = await getSpeedLimitList();
if (response.code !== 0) throw new Error(response.msg);
return (response.data || []) as SpeedLimitRule[];
};
export default function LimitPage() {
const [loading, setLoading] = useState(true);
const [rules, setRules] = useState<SpeedLimitRule[]>([]);
const order = useServerOrder(fetchOrderedRules, updateSpeedLimitOrder);
const { items: rules, loading, reload: loadData } = order;
const [searchKeyword, setSearchKeyword] = useLocalStorageState(
"limit-search-keyword",
"",
@@ -92,28 +105,6 @@ export default function LimitPage() {
// 表单验证错误
const [errors, setErrors] = useState<{ [key: string]: string }>({});
useEffect(() => {
loadData();
}, []);
// 加载所有数据
const loadData = async () => {
setLoading(true);
try {
const rulesRes = await getSpeedLimitList();
if (rulesRes.code === 0) {
setRules(rulesRes.data || []);
} else {
toast.error(rulesRes.msg || "获取限速规则失败");
}
} catch {
toast.error("加载数据失败");
} finally {
setLoading(false);
}
};
// 表单验证
const validateForm = (): boolean => {
const newErrors: { [key: string]: string } = {};
@@ -166,7 +157,7 @@ export default function LimitPage() {
// 确认删除规则
const confirmDelete = async () => {
if (!ruleToDelete) return;
if (!ruleToDelete || order.saving || deleteLoading) return;
setDeleteLoading(true);
try {
@@ -175,7 +166,7 @@ export default function LimitPage() {
if (res.code === 0) {
toast.success("删除成功");
setDeleteModalOpen(false);
loadData();
await loadData();
} else {
toast.error(res.msg || "删除失败");
}
@@ -188,7 +179,7 @@ export default function LimitPage() {
// 提交表单
const handleSubmit = async () => {
if (!validateForm()) return;
if (order.saving || !validateForm()) return;
setSubmitLoading(true);
try {
@@ -215,7 +206,7 @@ export default function LimitPage() {
if (res.code === 0) {
toast.success(isEdit ? "修改成功" : "创建成功");
setModalOpen(false);
loadData();
await loadData();
} else {
toast.error(res.msg || "操作失败");
}
@@ -226,9 +217,13 @@ export default function LimitPage() {
}
};
if (loading) {
return <PageLoadingState message="正在加载..." />;
}
const orderReason = getOrderDisabledReason({
isAdmin: isAdmin(),
searchKeyword,
saving: order.saving,
busy: loading || submitLoading || deleteLoading,
});
const orderDisabled = Boolean(orderReason);
return (
<AnimatedPage className="px-3 lg:px-6 py-8">
@@ -263,172 +258,207 @@ export default function LimitPage() {
</div>
</div>
{/* 统一卡片网格 */}
{filteredRules.length > 0 ? (
viewMode === "list" ? (
<Card>
<Table
aria-label="限速规则列表"
className="overflow-x-auto min-w-full"
classNames={{
wrapper:
"bg-transparent p-0 shadow-none border-none overflow-auto rounded-[24px]",
th: "bg-transparent text-default-600 font-semibold text-sm border-b border-white/20 dark:border-white/10 py-3 uppercase tracking-wider first:rounded-tl-[24px] last:rounded-tr-[24px]",
td: "py-3 border-b border-divider/50 group-data-[last=true]:border-b-0",
tr: "hover:bg-white/40 dark:hover:bg-white/10 transition-colors",
}}
>
<TableHeader>
<TableColumn>规则名称</TableColumn>
<TableColumn>速度限制</TableColumn>
<TableColumn>操作</TableColumn>
</TableHeader>
<TableBody items={filteredRules}>
{(rule) => (
<TableRow key={rule.id}>
<TableCell>
<div className="flex items-center gap-2">
<div
className={`shrink-0 w-2 h-2 rounded-full ${
rule.status === 1 ? "bg-success" : "bg-danger"
}`}
/>
<span className="font-medium text-foreground text-sm">
{rule.name}
<OrderStatus
error={order.error}
reason={orderReason}
retry={() => {
if (!order.saving) void loadData();
}}
/>
<OrderList
key={`${viewMode}:${searchKeyword}:${orderDisabled}`}
ids={filteredRules.map((rule) => rule.id)}
onDragEnd={(event) => {
void order.reorder(event, orderDisabled);
}}
onDragStart={order.onDragStart}
>
{/* 统一卡片网格 */}
{loading ? (
<PageLoadingState message="正在加载..." />
) : filteredRules.length > 0 ? (
viewMode === "list" ? (
<Card>
<Table
aria-label="限速规则列表"
className="overflow-x-auto min-w-full"
classNames={{
wrapper:
"bg-transparent p-0 shadow-none border-none overflow-auto rounded-[24px]",
th: "bg-transparent text-default-600 font-semibold text-sm border-b border-white/20 dark:border-white/10 py-3 uppercase tracking-wider first:rounded-tl-[24px] last:rounded-tr-[24px]",
td: "py-3 border-b border-divider/50 group-data-[last=true]:border-b-0",
tr: "hover:bg-white/40 dark:hover:bg-white/10 transition-colors",
}}
>
<TableHeader>
<TableColumn>规则名称</TableColumn>
<TableColumn>速度限制</TableColumn>
<TableColumn>操作</TableColumn>
</TableHeader>
<TableBody items={filteredRules}>
{(rule) => (
<OrderItem
key={rule.id}
row
disabled={orderDisabled}
id={rule.id}
label={rule.name}
>
<TableCell>
<div className="flex items-center gap-2">
<OrderHandle />
<div
className={`shrink-0 w-2 h-2 rounded-full ${
rule.status === 1 ? "bg-success" : "bg-danger"
}`}
/>
<span className="font-medium text-foreground text-sm">
{rule.name}
</span>
</div>
</TableCell>
<TableCell>
<span className="text-sm font-mono text-default-600">
{rule.speed} Mbps
</span>
</div>
</TableCell>
<TableCell>
<span className="text-sm font-mono text-default-600">
{rule.speed} Mbps
</span>
</TableCell>
<TableCell>
<div className="flex flex-wrap items-center gap-1.5 min-w-max">
<Button
className="h-6 px-2 min-w-0 text-xs bg-indigo-50 text-indigo-600 hover:bg-indigo-100 dark:bg-indigo-950/30 dark:text-indigo-400"
</TableCell>
<TableCell>
<div className="flex flex-wrap items-center gap-1.5 min-w-max">
<Button
className="h-6 px-2 min-w-0 text-xs bg-indigo-50 text-indigo-600 hover:bg-indigo-100 dark:bg-indigo-950/30 dark:text-indigo-400"
size="sm"
variant="flat"
onPress={() => handleEdit(rule)}
>
编辑
</Button>
<Button
className="h-6 px-2 min-w-0 text-xs bg-rose-50 text-rose-600 hover:bg-rose-100 dark:bg-rose-950/30 dark:text-rose-400"
size="sm"
variant="flat"
onPress={() => handleDelete(rule)}
>
删除
</Button>
</div>
</TableCell>
</OrderItem>
)}
</TableBody>
</Table>
</Card>
) : (
<StaggerList className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 xl:grid-cols-4 2xl:grid-cols-5 gap-4">
{filteredRules.map((rule) => (
<OrderItem
key={rule.id}
disabled={orderDisabled}
id={rule.id}
label={rule.name}
>
<Card className="shadow-sm overflow-hidden h-full">
<CardHeader className="pb-2 md:pb-2">
<div className="flex items-center gap-2 w-full min-w-0">
<div className="flex items-center gap-1 flex-1 min-w-0">
<OrderHandle />
<h3
className="font-semibold text-foreground truncate"
title={rule.name}
>
{rule.name}
</h3>
</div>
<Chip
className="shrink-0"
color={rule.status === 1 ? "success" : "danger"}
size="sm"
variant="flat"
>
{rule.status === 1 ? "运行" : "异常"}
</Chip>
</div>
</CardHeader>
<CardBody className="pt-0 pb-3 md:pt-0 md:pb-3">
<div className="space-y-3">
<div className="flex justify-between items-center">
<span className="text-small text-default-600">
速度限制
</span>
<Chip color="secondary" size="sm" variant="flat">
{rule.speed} Mbps
</Chip>
</div>
</div>
<div className="flex gap-2 mt-4">
<Button
className="flex-1"
color="primary"
size="sm"
startContent={
<svg
aria-hidden="true"
className="w-4 h-4"
fill="currentColor"
viewBox="0 0 20 20"
>
<path d="M13.586 3.586a2 2 0 112.828 2.828l-.793.793-2.828-2.828.793-.793zM11.379 5.793L3 14.172V17h2.828l8.38-8.379-2.83-2.828z" />
</svg>
}
variant="flat"
onPress={() => handleEdit(rule)}
>
编辑
</Button>
<Button
className="h-6 px-2 min-w-0 text-xs bg-rose-50 text-rose-600 hover:bg-rose-100 dark:bg-rose-950/30 dark:text-rose-400"
className="flex-1"
color="danger"
size="sm"
startContent={
<svg
aria-hidden="true"
className="w-4 h-4"
fill="currentColor"
viewBox="0 0 20 20"
>
<path
clipRule="evenodd"
d="M9 2a1 1 0 000 2h2a1 1 0 100-2H9z"
fillRule="evenodd"
/>
<path
clipRule="evenodd"
d="M10 18a8 8 0 100-16 8 8 0 000 16zM8 7a1 1 0 012 0v4a1 1 0 11-2 0V7zM12 7a1 1 0 012 0v4a1 1 0 11-2 0V7z"
fillRule="evenodd"
/>
</svg>
}
variant="flat"
onPress={() => handleDelete(rule)}
>
删除
</Button>
</div>
</TableCell>
</TableRow>
)}
</TableBody>
</Table>
</Card>
</CardBody>
</Card>
</OrderItem>
))}
</StaggerList>
)
) : (
<StaggerList className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 xl:grid-cols-4 2xl:grid-cols-5 gap-4">
{filteredRules.map((rule) => (
<StaggerItem key={rule.id}>
<Card className="shadow-sm overflow-hidden h-full">
<CardHeader className="pb-2 md:pb-2">
<div className="flex justify-between items-start w-full">
<div>
<h3 className="font-semibold text-foreground">
{rule.name}
</h3>
</div>
<Chip
color={rule.status === 1 ? "success" : "danger"}
size="sm"
variant="flat"
>
{rule.status === 1 ? "运行" : "异常"}
</Chip>
</div>
</CardHeader>
<CardBody className="pt-0 pb-3 md:pt-0 md:pb-3">
<div className="space-y-3">
<div className="flex justify-between items-center">
<span className="text-small text-default-600">
速度限制
</span>
<Chip color="secondary" size="sm" variant="flat">
{rule.speed} Mbps
</Chip>
</div>
</div>
<div className="flex gap-2 mt-4">
<Button
className="flex-1"
color="primary"
size="sm"
startContent={
<svg
aria-hidden="true"
className="w-4 h-4"
fill="currentColor"
viewBox="0 0 20 20"
>
<path d="M13.586 3.586a2 2 0 112.828 2.828l-.793.793-2.828-2.828.793-.793zM11.379 5.793L3 14.172V17h2.828l8.38-8.379-2.83-2.828z" />
</svg>
}
variant="flat"
onPress={() => handleEdit(rule)}
>
编辑
</Button>
<Button
className="flex-1"
color="danger"
size="sm"
startContent={
<svg
aria-hidden="true"
className="w-4 h-4"
fill="currentColor"
viewBox="0 0 20 20"
>
<path
clipRule="evenodd"
d="M9 2a1 1 0 000 2h2a1 1 0 100-2H9z"
fillRule="evenodd"
/>
<path
clipRule="evenodd"
d="M10 18a8 8 0 100-16 8 8 0 000 16zM8 7a1 1 0 012 0v4a1 1 0 11-2 0V7zM12 7a1 1 0 012 0v4a1 1 0 11-2 0V7z"
fillRule="evenodd"
/>
</svg>
}
variant="flat"
onPress={() => handleDelete(rule)}
>
删除
</Button>
</div>
</CardBody>
</Card>
</StaggerItem>
))}
</StaggerList>
)
) : (
/* 空状态 */
<Card className="shadow-sm bg-default-50/50">
<CardBody className="text-center py-20 flex flex-col items-center justify-center min-h-[240px]">
<h3 className="text-xl font-medium text-foreground tracking-tight mb-2">
暂无限速规则
</h3>
<p className="text-default-500 text-sm max-w-xs mx-auto leading-relaxed">
还没有创建任何限速规则,点击上方按钮开始创建
</p>
</CardBody>
</Card>
)}
/* 空状态 */
<Card className="shadow-sm bg-default-50/50">
<CardBody className="text-center py-20 flex flex-col items-center justify-center min-h-[240px]">
<h3 className="text-xl font-medium text-foreground tracking-tight mb-2">
暂无限速规则
</h3>
<p className="text-default-500 text-sm max-w-xs mx-auto leading-relaxed">
还没有创建任何限速规则,点击上方按钮开始创建
</p>
</CardBody>
</Card>
)}
</OrderList>
{/* 新增/编辑模态框 */}
<Modal
@@ -497,6 +527,7 @@ export default function LimitPage() {
</Button>
<Button
color="primary"
isDisabled={order.saving}
isLoading={submitLoading}
onPress={handleSubmit}
>
@@ -544,6 +575,7 @@ export default function LimitPage() {
</Button>
<Button
color="danger"
isDisabled={order.saving}
isLoading={deleteLoading}
onPress={confirmDelete}
>
+27 -15
View File
@@ -45,7 +45,6 @@ import { Chip } from "@/shadcn-bridge/heroui/chip";
import { Switch } from "@/shadcn-bridge/heroui/switch";
import { Spinner } from "@/shadcn-bridge/heroui/spinner";
import { Alert } from "@/shadcn-bridge/heroui/alert";
import { Progress } from "@/shadcn-bridge/heroui/progress";
import { Accordion, AccordionItem } from "@/shadcn-bridge/heroui/accordion";
import { Select, SelectItem } from "@/shadcn-bridge/heroui/select";
import { Checkbox } from "@/shadcn-bridge/heroui/checkbox";
@@ -70,6 +69,10 @@ import {
getConnectionStatusMeta,
getRemoteSyncErrorMessage,
} from "@/pages/node/display";
import {
NodeUpgradeProgress,
type NodeUpgradeProgressState,
} from "@/pages/node/upgrade-progress";
import { tryCopyInstallCommand } from "@/pages/node/install-command";
import {
getNodeRenewalSnapshot,
@@ -422,7 +425,7 @@ export default function NodePage() {
const [selectedVersion, setSelectedVersion] = useState("");
const [batchUpgradeLoading, setBatchUpgradeLoading] = useState(false);
const [upgradeProgress, setUpgradeProgress] = useState<
Record<number, { stage: string; percent: number; message: string }>
Record<number, NodeUpgradeProgressState>
>({});
const [infoPopoverPlacement, setInfoPopoverPlacement] = useState<
@@ -1120,6 +1123,13 @@ export default function NodePage() {
const node = nodeList.find((n) => n.id === upgradeTargetNodeId);
if (!node) return;
setUpgradeProgress((prev) => {
const next = { ...prev };
delete next[upgradeTargetNodeId];
return next;
});
setNodeList((prev) =>
prev.map((n) =>
n.id === upgradeTargetNodeId ? { ...n, upgradeLoading: true } : n,
@@ -1160,6 +1170,13 @@ export default function NodePage() {
return;
}
setUpgradeProgress((prev) => {
const next = { ...prev };
selectedLocalIds.forEach((id) => delete next[id]);
return next;
});
setBatchUpgradeLoading(true);
setUpgradeModalOpen(false);
try {
@@ -1956,6 +1973,11 @@ export default function NodePage() {
<span className="text-sm text-default-500">
{node.version || "-"}
</span>
{!isRemoteNode && (
<NodeUpgradeProgress
progress={upgradeProgress[node.id]}
/>
)}
</TableCell>
<TableCell>
<div className="flex flex-wrap items-center gap-1.5 min-w-max">
@@ -2261,19 +2283,9 @@ export default function NodePage() {
{node.version || "未知"}
</span>
</div>
{upgradeProgress[node.id] &&
upgradeProgress[node.id].percent < 100 && (
<div className="mt-1">
<Progress
showValueLabel
aria-label="升级进度"
color="warning"
label={upgradeProgress[node.id].message}
size="sm"
value={upgradeProgress[node.id].percent}
/>
</div>
)}
<NodeUpgradeProgress
progress={upgradeProgress[node.id]}
/>
<div className="flex justify-between text-sm">
<span className="text-default-600">
开机时间
@@ -0,0 +1,46 @@
import { Progress } from "@/shadcn-bridge/heroui/progress";
export interface NodeUpgradeProgressState {
stage: string;
percent: number;
message: string;
}
const STAGE_LABELS: Record<string, string> = {
downloading: "下载",
verifying: "校验",
installing: "安装 / 重启",
};
export function NodeUpgradeProgress({
progress,
}: {
progress?: NodeUpgradeProgressState;
}) {
// The final installing event means the agent is restarting, not a durable
// job state. Do not leave a stale "重启中 100%" banner after it reconnects.
if (
!progress ||
(progress.stage === "installing" && progress.percent >= 100)
) {
return null;
}
const percent = Number.isFinite(progress.percent)
? Math.min(100, Math.max(0, progress.percent))
: 0;
const stage = STAGE_LABELS[progress.stage];
return (
<div className="mt-2 min-w-40 max-w-xs break-words" role="status">
<Progress
showValueLabel
aria-label={stage ? `升级${stage}进度` : "升级进度"}
color="warning"
label={progress.message || (stage ? `正在${stage}` : "正在升级")}
size="sm"
value={percent}
/>
</div>
);
}
+60 -48
View File
@@ -64,7 +64,6 @@ import {
getNodeList,
diagnoseTunnel,
updateTunnelOrder,
batchRedeployTunnels,
previewBatchTunnelDelete,
previewTunnelDelete,
} from "@/api";
@@ -74,6 +73,11 @@ import {
getDiagnosisQualityDisplay,
type DiagnosisResult,
} from "@/pages/tunnel/diagnosis";
import { useTunnelRedeploy } from "@/pages/tunnel/use-tunnel-redeploy";
import {
TunnelRedeployStatus,
TunnelRedeploySummary,
} from "@/pages/tunnel/redeploy-progress";
import { diagnoseTunnelStream } from "@/api/diagnosis-stream";
import {
createTunnelFormDefaults,
@@ -455,6 +459,7 @@ export default function TunnelPage() {
number | null
>(null);
const [batchLoading, setBatchLoading] = useState(false);
const redeploy = useTunnelRedeploy();
const [batchProgress, setBatchProgress] = useState<BatchProgressState>({
active: false,
label: "",
@@ -1351,55 +1356,39 @@ export default function TunnelPage() {
};
const handleBatchRedeploy = async () => {
if (selectedIds.size === 0) return;
setBatchLoading(true);
setBatchProgress({
active: true,
label: `正在重新下发 ${selectedIds.size} 条隧道...`,
percent: 30,
});
if (selectedIds.size === 0 || batchLoading || redeploy.state?.running)
return;
try {
const res = await batchRedeployTunnels(Array.from(selectedIds));
const result = await redeploy.run(Array.from(selectedIds));
if (res.code === 0) {
const result = res.data;
if (result.failCount === 0) {
toast.success(`成功重新下发 ${result.successCount} 项`);
} else {
const failures = extractBatchFailures(result);
if (failures.length > 0) {
openBatchResultModal(
"批量下发结果",
`成功 ${result.successCount} 项,失败 ${result.failCount} 项`,
failures,
);
} else {
toast.error(
buildBatchFailureMessage(
result,
`成功 ${result.successCount} 项,失败 ${result.failCount} 项`,
),
);
}
}
setSelectedIds(new Set());
setSelectMode(false);
setBatchProgress({
active: true,
label: `重新下发完成:成功 ${result.successCount} 项,正在刷新列表...`,
percent: 100,
});
await refreshTunnelList(false);
if (!result) return;
if (result.failCount === 0) {
toast.success(`成功重新下发 ${result.successCount} 项`);
} else {
toast.error(res.msg || "下发失败");
const failures = extractBatchFailures(result);
if (failures.length > 0) {
openBatchResultModal(
"批量下发结果",
`成功 ${result.successCount} 项,失败 ${result.failCount} 项`,
failures,
);
} else {
toast.error(
buildBatchFailureMessage(
result,
`成功 ${result.successCount} 项,失败 ${result.failCount} 项`,
),
);
}
}
setSelectedIds(new Set());
setSelectMode(false);
await refreshTunnelList(false);
} catch (error) {
toast.error(extractApiErrorMessage(error, "下发失败"));
} finally {
setBatchProgress({ active: false, label: "", percent: 0 });
setBatchLoading(false);
toast.error(
extractApiErrorMessage(error, "下发进度中断,请检查结果后再重试"),
);
}
};
@@ -1707,7 +1696,11 @@ export default function TunnelPage() {
</Button>
<Button
color="danger"
isDisabled={selectedIds.size === 0}
isDisabled={
selectedIds.size === 0 ||
batchLoading ||
redeploy.state?.running
}
size="sm"
variant="flat"
onPress={handleOpenBatchDeleteModal}
@@ -1716,8 +1709,12 @@ export default function TunnelPage() {
</Button>
<Button
color="primary"
isDisabled={selectedIds.size === 0}
isLoading={batchLoading}
isDisabled={
selectedIds.size === 0 ||
batchLoading ||
redeploy.state?.running
}
isLoading={batchLoading || redeploy.state?.running}
size="sm"
variant="flat"
onPress={handleBatchRedeploy}
@@ -1789,6 +1786,11 @@ export default function TunnelPage() {
</div>
)}
<TunnelRedeploySummary
state={redeploy.state}
onDismiss={redeploy.clear}
/>
{/* 隧道卡片网格 */}
{tunnels.length > 0 ? (
viewMode === "list" ? (
@@ -1860,6 +1862,9 @@ export default function TunnelPage() {
{tunnel.name}
</span>
</div>
<TunnelRedeployStatus
row={redeploy.state?.rows[tunnel.id]}
/>
</TableCell>
<TableCell>
<Chip
@@ -1918,6 +1923,7 @@ export default function TunnelPage() {
<div className="flex flex-wrap items-center gap-1.5 min-w-max">
<Button
className="h-6 px-2 min-w-0 text-xs bg-indigo-50 text-indigo-600 hover:bg-indigo-100 dark:bg-indigo-950/30 dark:text-indigo-400"
isDisabled={redeploy.state?.running}
size="sm"
variant="flat"
onPress={() => handleEdit(tunnel)}
@@ -1934,6 +1940,7 @@ export default function TunnelPage() {
</Button>
<Button
className="h-6 px-2 min-w-0 text-xs bg-rose-50 text-rose-600 hover:bg-rose-100 dark:bg-rose-950/30 dark:text-rose-400"
isDisabled={redeploy.state?.running}
size="sm"
variant="flat"
onPress={() => handleDelete(tunnel)}
@@ -2015,6 +2022,9 @@ export default function TunnelPage() {
</CardHeader>
<CardBody className="pt-0 pb-3 md:pt-0 md:pb-3">
<TunnelRedeployStatus
row={redeploy.state?.rows[tunnel.id]}
/>
<div className="space-y-3">
{/* 拓扑结构 */}
<div className="pt-2 border-t border-divider">
@@ -2160,6 +2170,7 @@ export default function TunnelPage() {
<Button
className="flex-1 min-h-8"
color="primary"
isDisabled={redeploy.state?.running}
size="sm"
startContent={
<svg
@@ -2202,6 +2213,7 @@ export default function TunnelPage() {
<Button
className="flex-1 min-h-8"
color="danger"
isDisabled={redeploy.state?.running}
size="sm"
startContent={
<svg
@@ -0,0 +1,97 @@
import type {
TunnelRedeployRow,
TunnelRedeployState,
} from "./use-tunnel-redeploy";
import { Button } from "@/shadcn-bridge/heroui/button";
import { Progress } from "@/shadcn-bridge/heroui/progress";
import { Spinner } from "@/shadcn-bridge/heroui/spinner";
export function TunnelRedeployStatus({ row }: { row?: TunnelRedeployRow }) {
if (!row) return null;
const text = {
queued: "等待下发",
running: "正在下发配置…",
success: "配置下发成功",
failed: "配置下发失败",
unknown: "下发结果未确认",
}[row.status];
const color =
row.status === "failed"
? "text-danger"
: row.status === "success"
? "text-success"
: row.status === "unknown"
? "text-warning"
: "text-default-500";
return (
<div className={`mt-1 flex items-start gap-1.5 text-xs ${color}`}>
{row.status === "running" && <Spinner size="sm" />}
<span className="max-w-xs break-words" title={row.reason}>
{text}
{row.reason ? `:${row.reason}` : ""}
</span>
</div>
);
}
export function TunnelRedeploySummary({
state,
onDismiss,
}: {
state: TunnelRedeployState | null;
onDismiss: () => void;
}) {
if (!state) return null;
const { progress, running, error, currentName } = state;
const label = `${error ? "已确认" : "已完成"} ${progress.completed}/${progress.total} 条,成功 ${progress.success} 条,失败 ${progress.failed} 条`;
const percent =
progress.total > 0 ? (progress.completed / progress.total) * 100 : 0;
return (
<div
className="mb-4 space-y-2 rounded-xl border border-divider bg-content1 p-3"
role="status"
>
<div className="flex items-start justify-between gap-3">
<div className="min-w-0 space-y-1 break-words text-sm">
<p className="font-medium">
{running
? "正在下发隧道配置"
: error
? "下发进度已中断"
: "隧道配置下发结束"}
</p>
<p className="text-default-500">{label}</p>
{currentName && <p>当前:{currentName}</p>}
{running && (
<p className="text-xs text-default-500">
进度按已完成的隧道统计,当前隧道内部步骤不计百分比。请保留此页面。
</p>
)}
{error && <p className="text-warning">{error}</p>}
</div>
{!running && (
<Button
aria-label="关闭下发结果"
size="sm"
variant="light"
onPress={onDismiss}
>
关闭
</Button>
)}
</div>
<Progress
showValueLabel
aria-label="隧道配置下发进度"
color={error || progress.failed > 0 ? "warning" : "primary"}
size="sm"
value={percent}
/>
</div>
);
}
@@ -0,0 +1,118 @@
import { useEffect, useRef, useState } from "react";
import {
batchRedeployTunnelsStream,
type TunnelRedeployProgress,
} from "@/api/tunnel-redeploy-stream";
import { extractApiErrorMessage } from "@/api/error-message";
export interface TunnelRedeployRow {
status: "queued" | "running" | "success" | "failed" | "unknown";
reason?: string;
}
export interface TunnelRedeployState {
running: boolean;
progress: TunnelRedeployProgress;
rows: Record<number, TunnelRedeployRow>;
currentName?: string;
error?: string;
}
export function useTunnelRedeploy() {
const [state, setState] = useState<TunnelRedeployState | null>(null);
const abortRef = useRef<AbortController | null>(null);
useEffect(() => () => abortRef.current?.abort(), []);
const run = async (ids: number[]) => {
// Guard synchronously, before React has rendered a disabled button.
if (abortRef.current || ids.length === 0) return null;
const controller = new AbortController();
abortRef.current = controller;
setState({
running: true,
progress: { total: ids.length, completed: 0, success: 0, failed: 0 },
rows: Object.fromEntries(ids.map((id) => [id, { status: "queued" }])),
});
try {
const result = await batchRedeployTunnelsStream(
ids,
{
onStart: (progress) => {
setState((prev) => prev && { ...prev, progress });
},
onProcessing: ({ id, name, progress }) => {
setState(
(prev) =>
prev && {
...prev,
progress,
currentName: name || `隧道 #${id}`,
rows: { ...prev.rows, [id]: { status: "running" } },
},
);
},
onItem: ({ result: item, progress }) => {
setState(
(prev) =>
prev && {
...prev,
progress,
currentName: undefined,
rows: {
...prev.rows,
[item.id]: {
status: item.success ? "success" : "failed",
reason: item.reason,
},
},
},
);
},
},
controller.signal,
);
if (controller.signal.aborted) return null;
setState(
(prev) => prev && { ...prev, running: false, currentName: undefined },
);
return result;
} catch (error) {
if (controller.signal.aborted) return null;
setState(
(prev) =>
prev && {
...prev,
running: false,
currentName: undefined,
error: extractApiErrorMessage(
error,
"下发进度中断,请检查未确认的结果后再重试",
),
rows: Object.fromEntries(
Object.entries(prev.rows).map(([id, row]) => [
id,
row.status === "queued" || row.status === "running"
? { status: "unknown" }
: row,
]),
),
},
);
throw error;
} finally {
if (abortRef.current === controller) abortRef.current = null;
}
};
const clear = () => {
if (!abortRef.current) setState(null);
};
return { state, run, clear };
}
File diff suppressed because it is too large Load Diff
+20
View File
@@ -0,0 +1,20 @@
interface OrderAvailability {
isAdmin: boolean;
searchKeyword: string;
saving: boolean;
busy: boolean;
}
export function getOrderDisabledReason({
isAdmin,
searchKeyword,
saving,
busy,
}: OrderAvailability): string {
if (!isAdmin) return "仅管理员可调整全局顺序。";
if (searchKeyword.length > 0) return "搜索期间无法排序,请清空搜索后重试。";
if (saving) return "正在保存顺序,请稍候。";
if (busy) return "正在加载或更新列表,请稍候。";
return "";
}