package service import ( "context" "encoding/json" "net/http" "net/http/httptest" "net/url" "reflect" "strconv" "testing" "time" "github.com/mhsanaei/3x-ui/v3/internal/database" "github.com/mhsanaei/3x-ui/v3/internal/database/model" "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" "github.com/mhsanaei/3x-ui/v3/internal/xray" "gorm.io/gorm" ) // commitTickBetweenReadAndWrite parks the serial writer, lets op read the // inbound and queue its transaction, then commits tick ahead of that transaction. func commitTickBetweenReadAndWrite(t *testing.T, tick func(tx *gorm.DB) error, op func()) { t.Helper() resetTrafficWriterForTest(t) StartTrafficWriter() parked := make(chan struct{}) release := make(chan struct{}) tickErr := make(chan error, 1) go func() { tickErr <- submitTrafficWrite(func() error { close(parked) <-release return database.GetDB().Transaction(tick) }) }() <-parked opDone := make(chan struct{}) go func() { defer close(opDone) op() }() waitTrafficWriterQueued(t) close(release) if err := <-tickErr; err != nil { t.Fatalf("tick: %v", err) } <-opDone } // seedRenewableNeighbour builds an inbound holding a healthy client X and a // quota-disabled client Y whose auto-renew is due, as the traffic job sees them. func seedRenewableNeighbour(t *testing.T, port int, nodeID *int) *model.Inbound { t.Helper() past := time.Now().Add(-time.Hour).UnixMilli() future := time.Now().Add(30 * 24 * time.Hour).UnixMilli() clients := []model.Client{ {Email: "x@stale", ID: "aaaaaaaa-0000-0000-0000-00000000000a", SubID: "sub-x", Enable: true, ExpiryTime: future}, {Email: "y@stale", ID: "aaaaaaaa-0000-0000-0000-00000000000b", SubID: "sub-y", Enable: false, Reset: 30, ExpiryTime: past, TotalGB: 1000}, } ib := &model.Inbound{ Tag: "stale-" + strconv.Itoa(port), Enable: true, Port: port, Protocol: model.VLESS, Settings: clientsSettings(t, clients), NodeID: nodeID, } if err := database.GetDB().Create(ib).Error; err != nil { t.Fatalf("create inbound: %v", err) } if err := (&ClientService{}).SyncInbound(nil, ib.Id, clients); err != nil { t.Fatalf("SyncInbound: %v", err) } rows := []xray.ClientTraffic{ {InboundId: ib.Id, Email: "x@stale", Enable: true, ExpiryTime: future}, {InboundId: ib.Id, Email: "y@stale", Enable: false, Up: 600, Down: 400, Total: 1000, Reset: 30, ExpiryTime: past}, } if err := database.GetDB().Create(&rows).Error; err != nil { t.Fatalf("seed client_traffics: %v", err) } return ib } func autoRenewTick(tx *gorm.DB) error { _, _, err := (&InboundService{}).autoRenewClients(tx, newTrafficMutationBatch()) return err } // renewYTick writes the renewal autoRenewClients would commit for y@stale; it // skips clients hosted only on a node, so the node case applies it directly. func renewYTick(inboundId int) func(tx *gorm.DB) error { return func(tx *gorm.DB) error { var ib model.Inbound if err := tx.First(&ib, inboundId).Error; err != nil { return err } var settings map[string]any if err := json.Unmarshal([]byte(ib.Settings), &settings); err != nil { return err } for _, c := range settings["clients"].([]any) { if m := c.(map[string]any); m["email"] == "y@stale" { m["enable"] = true m["expiryTime"] = time.Now().Add(30 * 24 * time.Hour).UnixMilli() } } b, err := json.MarshalIndent(settings, "", " ") if err != nil { return err } return tx.Model(&model.Inbound{}).Where("id = ?", inboundId).Update("settings", string(b)).Error } } func settingsClient(t *testing.T, inboundId int, email string) (model.Client, bool) { t.Helper() ib, err := (&InboundService{}).GetInbound(inboundId) if err != nil { t.Fatalf("GetInbound: %v", err) } clients, err := (&InboundService{}).GetClients(ib) if err != nil { t.Fatalf("GetClients: %v", err) } for _, c := range clients { if c.Email == email { return c, true } } return model.Client{}, false } func requireNeighbourRenewed(t *testing.T, inboundId int) model.Client { t.Helper() y, ok := settingsClient(t, inboundId, "y@stale") if !ok { t.Fatal("neighbour y@stale missing from settings") } if now := time.Now().UnixMilli(); !y.Enable || y.ExpiryTime <= now { t.Fatalf("renewed neighbour rolled back in settings: enable=%v expiryTime=%d (now %d)", y.Enable, y.ExpiryTime, now) } return y } type staleClientOp struct { name string // advancesNodeFingerprint: on a node inbound the op pushes per client and // then advances the reconcile-skip fingerprint. advancesNodeFingerprint bool run func(t *testing.T, ib *model.Inbound) error } var staleClientOps = []staleClientOp{ {"edit", true, func(t *testing.T, ib *model.Inbound) error { rec := lookupClientRecord(t, "x@stale") edited := rec.ToClient() edited.Comment = "edited" _, err := (&ClientService{}).UpdateInboundClient(&InboundService{}, &model.Inbound{ Id: ib.Id, Settings: clientsSettings(t, []model.Client{*edited}), }, "x@stale") return err }}, {"add", true, func(t *testing.T, ib *model.Inbound) error { _, err := (&ClientService{}).AddInboundClient(&InboundService{}, &model.Inbound{ Id: ib.Id, Settings: clientsSettings(t, []model.Client{{Email: "z@stale", ID: "aaaaaaaa-0000-0000-0000-00000000000c", Enable: true}}), }) return err }}, {"delete", true, func(t *testing.T, ib *model.Inbound) error { _, err := (&ClientService{}).DelInboundClientByEmail(&InboundService{}, ib.Id, "x@stale", false, true) return err }}, {"bulk detach", true, func(t *testing.T, ib *model.Inbound) error { _, _, err := (&ClientService{}).BulkDetach(&InboundService{}, []string{"x@stale"}, []int{ib.Id}) return err }}, {"bulk adjust", false, func(t *testing.T, ib *model.Inbound) error { _, _, err := (&ClientService{}).BulkAdjust(&InboundService{}, []string{"x@stale"}, 1, 0, "", nil, "") return err }}, {"bulk delete", false, func(t *testing.T, ib *model.Inbound) error { _, _, err := (&ClientService{}).BulkDelete(&InboundService{}, []string{"x@stale"}, false) return err }}, {"bulk set enable", true, func(t *testing.T, ib *model.Inbound) error { _, _, err := (&ClientService{}).BulkSetEnable(&InboundService{}, []string{"x@stale"}, false) return err }}, } // Each client op reads the inbound before queueing its write; a renewal the // traffic writer commits in between must not be reverted to enable=false. func TestClientOpsKeepNeighbourRenewedMidOp(t *testing.T) { for i, op := range staleClientOps { t.Run(op.name, func(t *testing.T) { setupBulkDB(t) ib := seedRenewableNeighbour(t, 23101+i, nil) commitTickBetweenReadAndWrite(t, autoRenewTick, func() { if err := op.run(t, ib); err != nil { t.Errorf("%s: %v", op.name, err) } }) requireNeighbourRenewed(t, ib.Id) }) } } // An op on the renewed client itself keeps the fields it did not change. func TestBulkAdjustOnRenewedClientKeepsRenewal(t *testing.T) { setupBulkDB(t) ib := seedRenewableNeighbour(t, 23120, nil) commitTickBetweenReadAndWrite(t, autoRenewTick, func() { if _, _, err := (&ClientService{}).BulkAdjust(&InboundService{}, []string{"y@stale"}, 0, 500, "", nil, ""); err != nil { t.Errorf("BulkAdjust: %v", err) } }) if y := requireNeighbourRenewed(t, ib.Id); y.TotalGB != 1500 { t.Fatalf("y@stale totalGB = %d, want 1500 (the adjust itself was lost)", y.TotalGB) } } // The node got only the per-client push, so the skip fingerprint must not claim // it also holds the renewal the traffic writer committed mid-op. func TestNodeClientOpsMidRenewalStillReconcileRenewal(t *testing.T) { for i, op := range staleClientOps { if !op.advancesNodeFingerprint { continue } t.Run(op.name, func(t *testing.T) { setupBulkDB(t) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{"success":true}`)) })) t.Cleanup(srv.Close) u, err := url.Parse(srv.URL) if err != nil { t.Fatalf("parse url: %v", err) } port, _ := strconv.Atoi(u.Port()) node := &model.Node{ Name: "stale-node", Scheme: "http", Address: u.Hostname(), Port: port, BasePath: "/", ApiToken: "tok", Enable: true, Status: "online", AllowPrivateAddress: true, } if err := database.GetDB().Create(node).Error; err != nil { t.Fatalf("create node: %v", err) } remote := runtime.NewRemote(node, nil) useTestRuntimeManager(t).SetRuntimeOverride(node.Id, remote) ib := seedRenewableNeighbour(t, 23131+i, &node.Id) remote.AdoptInboundAlias(ib, runtime.RemoteInboundOption{Id: 7, Tag: ib.Tag}) commitTickBetweenReadAndWrite(t, renewYTick(ib.Id), func() { if err := op.run(t, ib); err != nil { t.Errorf("%s: %v", op.name, err) } }) requireNeighbourRenewed(t, ib.Id) saved, err := (&InboundService{}).GetInbound(ib.Id) if err != nil { t.Fatalf("GetInbound: %v", err) } pushed, err := remote.ReconcileInbound(context.Background(), saved, true) if err != nil { t.Fatalf("ReconcileInbound: %v", err) } if !pushed { t.Fatal("reconcile skipped the inbound: the node never receives y@stale's renewal") } }) } } func TestRebaseClientSettings(t *testing.T) { const base = `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}` cases := []struct { name, ours, current, want string }{ { name: "untouched client takes the committed version", ours: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":2},{"email":"b","enable":false,"expiryTime":1}]}`, current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":true,"expiryTime":9}]}`, want: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":2},{"email":"b","enable":true,"expiryTime":9}]}`, }, { name: "edited client keeps committed changes to fields the op left alone", ours: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1,"comment":"x"}]}`, current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":true,"expiryTime":9}]}`, want: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":true,"expiryTime":9,"comment":"x"}]}`, }, { name: "client the op removed stays removed", ours: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1}]}`, current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":true,"expiryTime":9}]}`, want: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1}]}`, }, { name: "client committed after the read is kept", ours: `{"decryption":"none","clients":[{"email":"a","enable":false,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`, current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1},{"email":"c"}]}`, want: `{"decryption":"none","clients":[{"email":"a","enable":false,"totalGB":1},{"email":"b","enable":false,"expiryTime":1},{"email":"c"}]}`, }, { name: "untouched client removed after the read stays removed", ours: `{"decryption":"none","clients":[{"email":"a","enable":false,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`, current: `{"decryption":"none","clients":[{"email":"a","enable":true,"totalGB":1}]}`, want: `{"decryption":"none","clients":[{"email":"a","enable":false,"totalGB":1}]}`, }, { name: "top-level key follows whichever side changed it", ours: `{"decryption":"none","testseed":[1],"clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`, current: `{"decryption":"mlkem","clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`, want: `{"decryption":"mlkem","testseed":[1],"clients":[{"email":"a","enable":true,"totalGB":1},{"email":"b","enable":false,"expiryTime":1}]}`, }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { got, err := rebaseClientSettings(base, tc.ours, tc.current) if err != nil { t.Fatalf("rebaseClientSettings: %v", err) } var gotV, wantV any if err := json.Unmarshal([]byte(got), &gotV); err != nil { t.Fatalf("unmarshal got: %v", err) } if err := json.Unmarshal([]byte(tc.want), &wantV); err != nil { t.Fatalf("unmarshal want: %v", err) } if !reflect.DeepEqual(gotV, wantV) { t.Fatalf("rebase = %s\nwant %s", got, tc.want) } }) } }