3x-ui/internal/web/service/inbound_node_reconcile_test.go

package service

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"net/url"
	"slices"
	"sort"
	"strconv"
	"strings"
	"sync"
	"testing"

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

// fakeNodePanel serves just enough of the node API for ReconcileNode: the
// inbound list plus update/del endpoints, recording which remote ids get
// deleted.
func fakeNodePanel(t *testing.T, tagToID map[string]int) (*httptest.Server, func() []int) {
	t.Helper()
	var mu sync.Mutex
	var deleted []int
	writeOK := func(w http.ResponseWriter, obj any) {
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{"success": true, "msg": "", "obj": obj})
	}
	mux := http.NewServeMux()
	mux.HandleFunc("/panel/api/inbounds/list", func(w http.ResponseWriter, _ *http.Request) {
		type row struct {
			Id  int    `json:"id"`
			Tag string `json:"tag"`
		}
		rows := make([]row, 0, len(tagToID))
		for tag, id := range tagToID {
			rows = append(rows, row{Id: id, Tag: tag})
		}
		writeOK(w, rows)
	})
	mux.HandleFunc("/panel/api/inbounds/update/", func(w http.ResponseWriter, _ *http.Request) {
		writeOK(w, nil)
	})
	mux.HandleFunc("/panel/api/inbounds/del/", func(w http.ResponseWriter, r *http.Request) {
		id, err := strconv.Atoi(strings.TrimPrefix(r.URL.Path, "/panel/api/inbounds/del/"))
		if err != nil {
			http.Error(w, "bad id", http.StatusBadRequest)
			return
		}
		mu.Lock()
		deleted = append(deleted, id)
		mu.Unlock()
		writeOK(w, nil)
	})
	ts := httptest.NewServer(mux)
	t.Cleanup(ts.Close)
	return ts, func() []int {
		mu.Lock()
		defer mu.Unlock()
		out := append([]int(nil), deleted...)
		sort.Ints(out)
		return out
	}
}

func reconcileTestNode(t *testing.T, ts *httptest.Server, name, mode string, tags []string) *model.Node {
	t.Helper()
	u, err := url.Parse(ts.URL)
	if err != nil {
		t.Fatalf("parse test server URL: %v", err)
	}
	port, err := strconv.Atoi(u.Port())
	if err != nil {
		t.Fatalf("parse test server port: %v", err)
	}
	n := &model.Node{
		Name:                name,
		Scheme:              "http",
		Address:             u.Hostname(),
		Port:                port,
		BasePath:            "/",
		ApiToken:            "tok",
		Enable:              true,
		AllowPrivateAddress: true,
		Status:              "online",
		InboundSyncMode:     mode,
		InboundTags:         tags,
		InboundsAdoptedAt:   1,
	}
	if err := database.GetDB().Create(n).Error; err != nil {
		t.Fatalf("create node: %v", err)
	}
	return n
}

// In "selected" sync mode the panel never imports the unselected inbounds, so
// reconcile must not treat their absence from the local DB as a deletion: only
// a *selected* tag missing locally may be swept from the node.
func TestReconcileNode_SelectedModeLeavesUnselectedRemoteInbounds(t *testing.T) {
	setupConflictDB(t)

	ts, deletedIDs := fakeNodePanel(t, map[string]int{
		"keep":          1,
		"selected-gone": 2,
		"unmanaged":     3,
	})
	node := reconcileTestNode(t, ts, "sel-node", "selected", []string{"keep", "selected-gone"})
	seedInboundConflictNode(t, "keep", "", 443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id)

	svc := InboundService{}
	if err := svc.ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node); err != nil {
		t.Fatalf("ReconcileNode: %v", err)
	}

	got := deletedIDs()
	if len(got) != 1 || got[0] != 2 {
		t.Fatalf("deleted remote ids = %v, want [2] (unmanaged inbound 3 must survive)", got)
	}
}

// "all" mode keeps the original anti-entropy contract: every remote inbound
// missing from the local DB is deleted on the node.
func TestReconcileNode_AllModeDeletesUndesiredRemoteInbounds(t *testing.T) {
	setupConflictDB(t)

	ts, deletedIDs := fakeNodePanel(t, map[string]int{
		"keep":   1,
		"gone-a": 2,
		"gone-b": 3,
	})
	node := reconcileTestNode(t, ts, "all-node", "all", nil)
	seedInboundConflictNode(t, "keep", "", 443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id)

	svc := InboundService{}
	if err := svc.ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node); err != nil {
		t.Fatalf("ReconcileNode: %v", err)
	}

	got := deletedIDs()
	if len(got) != 2 || got[0] != 2 || got[1] != 3 {
		t.Fatalf("deleted remote ids = %v, want [2 3]", got)
	}
}

// A node whose pre-existing inbounds were never adopted into the central DB
// has zero local rows for legitimate reasons: reconcile before that first
// adoption must not sweep — it would delete every real inbound on the node
// right after onboarding (add node, save it again, watch it get wiped).
func TestReconcileNode_SkipsSweepBeforeFirstAdoption(t *testing.T) {
	setupConflictDB(t)

	ts, deletedIDs := fakeNodePanel(t, map[string]int{
		"real-a": 1,
		"real-b": 2,
		"real-c": 3,
	})
	node := reconcileTestNode(t, ts, "fresh-node", "all", nil)
	node.InboundsAdoptedAt = 0

	svc := InboundService{}
	if err := svc.ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node); err != nil {
		t.Fatalf("ReconcileNode: %v", err)
	}

	if got := deletedIDs(); len(got) != 0 {
		t.Fatalf("deleted remote ids = %v, want none before first adoption", got)
	}
}

// One inbound the node rejects (e.g. a legacy protocol failing the node's
// request validation, #5685) must not abort the reconcile: the healthy inbound
// is still pushed, the delete sweep still runs, and the returned error names
// the failed tag so the caller keeps the dirty flag set for retry.
func TestReconcileNode_ContinuesPastFailedInbound(t *testing.T) {
	setupConflictDB(t)

	var mu sync.Mutex
	updated := map[int]int{}
	var deleted []int
	tagToID := map[string]int{"legacy": 1, "healthy": 2, "gone": 3}
	writeOK := func(w http.ResponseWriter, obj any) {
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{"success": true, "msg": "", "obj": obj})
	}
	mux := http.NewServeMux()
	mux.HandleFunc("/panel/api/inbounds/list", func(w http.ResponseWriter, _ *http.Request) {
		type row struct {
			Id  int    `json:"id"`
			Tag string `json:"tag"`
		}
		rows := make([]row, 0, len(tagToID))
		for tag, id := range tagToID {
			rows = append(rows, row{Id: id, Tag: tag})
		}
		writeOK(w, rows)
	})
	mux.HandleFunc("/panel/api/inbounds/update/", func(w http.ResponseWriter, r *http.Request) {
		id, err := strconv.Atoi(strings.TrimPrefix(r.URL.Path, "/panel/api/inbounds/update/"))
		if err != nil {
			http.Error(w, "bad id", http.StatusBadRequest)
			return
		}
		if id == tagToID["legacy"] {
			http.Error(w, "request body failed validation", http.StatusBadRequest)
			return
		}
		mu.Lock()
		updated[id]++
		mu.Unlock()
		writeOK(w, nil)
	})
	mux.HandleFunc("/panel/api/inbounds/del/", func(w http.ResponseWriter, r *http.Request) {
		id, err := strconv.Atoi(strings.TrimPrefix(r.URL.Path, "/panel/api/inbounds/del/"))
		if err != nil {
			http.Error(w, "bad id", http.StatusBadRequest)
			return
		}
		mu.Lock()
		deleted = append(deleted, id)
		mu.Unlock()
		writeOK(w, nil)
	})
	ts := httptest.NewServer(mux)
	t.Cleanup(ts.Close)

	node := reconcileTestNode(t, ts, "half-broken-node", "all", nil)
	seedInboundConflictNode(t, "legacy", "", 1080, model.Protocol("socks"), ``, `{"auth":"noauth"}`, &node.Id)
	seedInboundConflictNode(t, "healthy", "", 443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id)

	svc := InboundService{}
	err := svc.ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node)
	if err == nil {
		t.Fatal("ReconcileNode: want an error naming the rejected inbound, got nil")
	}
	if !strings.Contains(err.Error(), `reconcile inbound "legacy"`) {
		t.Fatalf("ReconcileNode error = %q, want it to name inbound \"legacy\"", err)
	}

	mu.Lock()
	healthyPushes := updated[tagToID["healthy"]]
	gotDeleted := append([]int(nil), deleted...)
	mu.Unlock()
	if healthyPushes != 1 {
		t.Fatalf("healthy inbound pushed %d times, want 1", healthyPushes)
	}
	sort.Ints(gotDeleted)
	if len(gotDeleted) != 1 || gotDeleted[0] != tagToID["gone"] {
		t.Fatalf("deleted remote ids = %v, want [%d] (sweep must still run past the failure)", gotDeleted, tagToID["gone"])
	}
}

func TestReconcileNode_AdoptsCompatibleOriginInboundWithoutRemoteMutation(t *testing.T) {
	setupConflictDB(t)

	var mu sync.Mutex
	mutations := 0
	writeOK := func(w http.ResponseWriter, obj any) {
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{"success": true, "msg": "", "obj": obj})
	}
	mux := http.NewServeMux()
	mux.HandleFunc("/panel/api/inbounds/list", func(w http.ResponseWriter, _ *http.Request) {
		writeOK(w, []map[string]any{{"id": 41, "tag": "already-deployed", "listen": "", "port": 8443, "protocol": "vless"}})
	})
	mux.HandleFunc("/panel/api/inbounds/", func(w http.ResponseWriter, _ *http.Request) {
		mu.Lock()
		mutations++
		mu.Unlock()
		writeOK(w, nil)
	})
	ts := httptest.NewServer(mux)
	t.Cleanup(ts.Close)

	node := reconcileTestNode(t, ts, "adopt-node", "all", nil)
	node.Guid = "origin-guid"
	if err := database.GetDB().Model(node).Update("guid", node.Guid).Error; err != nil {
		t.Fatalf("update node guid: %v", err)
	}
	seedInboundConflictNode(t, "desired-name", "", 8443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id)
	if err := database.GetDB().Model(&model.Inbound{}).Where("tag = ?", "desired-name").Update("origin_node_guid", node.Guid).Error; err != nil {
		t.Fatalf("set origin guid: %v", err)
	}

	svc := InboundService{}
	rt := runtime.NewRemote(node, nil)
	if err := svc.ReconcileNode(context.Background(), rt, node); err != nil {
		t.Fatalf("first ReconcileNode: %v", err)
	}
	if err := svc.ReconcileNode(context.Background(), rt, node); err != nil {
		t.Fatalf("second ReconcileNode: %v", err)
	}
	mu.Lock()
	got := mutations
	mu.Unlock()
	if got != 0 {
		t.Fatalf("remote mutations = %d, want 0 while adopting compatible deployed inbound", got)
	}
}

func TestReconcileNode_AmbiguousCompatibleInboundsAreNotSwept(t *testing.T) {
	setupConflictDB(t)

	ts, deletedIDs := fakeNodePanel(t, map[string]int{"alias-a": 51, "alias-b": 52})
	node := reconcileTestNode(t, ts, "ambiguous-node", "all", nil)
	node.Guid = "origin-guid"
	if err := database.GetDB().Model(node).Update("guid", node.Guid).Error; err != nil {
		t.Fatalf("update node guid: %v", err)
	}
	seedInboundConflictNode(t, "desired-name", "", 0, model.Protocol(""), `{}`, `{"clients":[]}`, &node.Id)
	if err := database.GetDB().Model(&model.Inbound{}).Where("tag = ?", "desired-name").Update("origin_node_guid", node.Guid).Error; err != nil {
		t.Fatalf("set origin guid: %v", err)
	}

	err := (&InboundService{}).ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node)
	if err == nil || !strings.Contains(err.Error(), "ambiguous compatible remote inbounds") {
		t.Fatalf("ReconcileNode error = %v, want ambiguity error", err)
	}
	if got := deletedIDs(); len(got) != 0 {
		t.Fatalf("deleted ambiguous candidates = %v, want none", got)
	}
}

func TestReconcileNode_IncompatiblePortOccupantRemainsLoud(t *testing.T) {
	setupConflictDB(t)

	writeOK := func(w http.ResponseWriter, obj any) {
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{"success": true, "msg": "", "obj": obj})
	}
	mux := http.NewServeMux()
	mux.HandleFunc("/panel/api/inbounds/list", func(w http.ResponseWriter, _ *http.Request) {
		writeOK(w, []map[string]any{{"id": 42, "tag": "port-owner", "listen": "", "port": 9443, "protocol": "trojan"}})
	})
	mux.HandleFunc("/panel/api/inbounds/add", func(w http.ResponseWriter, _ *http.Request) {
		_ = json.NewEncoder(w).Encode(map[string]any{"success": false, "msg": "port already occupied", "obj": nil})
	})
	ts := httptest.NewServer(mux)
	t.Cleanup(ts.Close)

	node := reconcileTestNode(t, ts, "drift-node", "all", nil)
	node.Guid = "origin-guid"
	seedInboundConflictNode(t, "desired-name", "", 9443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id)
	if err := database.GetDB().Model(&model.Inbound{}).Where("tag = ?", "desired-name").Update("origin_node_guid", node.Guid).Error; err != nil {
		t.Fatalf("set origin guid: %v", err)
	}

	err := (&InboundService{}).ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node)
	if err == nil || !strings.Contains(err.Error(), "port already occupied") {
		t.Fatalf("ReconcileNode error = %v, want loud incompatible-port error", err)
	}
}

func TestEnsureInboundTagAllowed(t *testing.T) {
	setupConflictDB(t)
	db := database.GetDB()
	svc := NodeService{}

	selected := &model.Node{
		Name: "ensure-sel", Address: "127.0.0.1", Port: 2096, ApiToken: "tok",
		InboundSyncMode: "selected", InboundTags: []string{"a"},
	}
	if err := db.Create(selected).Error; err != nil {
		t.Fatalf("create node: %v", err)
	}

	if err := svc.EnsureInboundTagAllowed(selected.Id, "b"); err != nil {
		t.Fatalf("EnsureInboundTagAllowed add: %v", err)
	}
	var got model.Node
	if err := db.First(&got, selected.Id).Error; err != nil {
		t.Fatalf("reload node: %v", err)
	}
	if len(got.InboundTags) != 2 || got.InboundTags[0] != "a" || got.InboundTags[1] != "b" {
		t.Fatalf("InboundTags = %#v, want [a b]", got.InboundTags)
	}

	if err := svc.EnsureInboundTagAllowed(selected.Id, "a"); err != nil {
		t.Fatalf("EnsureInboundTagAllowed existing: %v", err)
	}
	if err := db.First(&got, selected.Id).Error; err != nil {
		t.Fatalf("reload node: %v", err)
	}
	if len(got.InboundTags) != 2 {
		t.Fatalf("existing tag must not duplicate, got %#v", got.InboundTags)
	}

	all := &model.Node{
		Name: "ensure-all", Address: "127.0.0.1", Port: 2097, ApiToken: "tok",
		InboundSyncMode: "all",
	}
	if err := db.Create(all).Error; err != nil {
		t.Fatalf("create node: %v", err)
	}
	if err := svc.EnsureInboundTagAllowed(all.Id, "x"); err != nil {
		t.Fatalf("EnsureInboundTagAllowed all-mode: %v", err)
	}
	var gotAll model.Node
	if err := db.First(&gotAll, all.Id).Error; err != nil {
		t.Fatalf("reload node: %v", err)
	}
	if len(gotAll.InboundTags) != 0 {
		t.Fatalf("all-mode node must stay without tags, got %#v", gotAll.InboundTags)
	}
}

// A panel-created node inbound is stored as "n<id>-tag" and pushed to the node
// with the prefix stripped, so the sweep's selected set must match both forms.
func TestReconcileNode_SelectedModeSweepsPrefixedSelectedTag(t *testing.T) {
	setupConflictDB(t)

	ts, deletedIDs := fakeNodePanel(t, map[string]int{
		"keep":          1,
		"selected-gone": 2,
		"unmanaged":     3,
	})
	node := reconcileTestNode(t, ts, "sel-prefix-node", "selected", nil)
	prefix := fmt.Sprintf("n%d-", node.Id)
	node.InboundTags = []string{prefix + "keep", prefix + "selected-gone"}
	seedInboundConflictNode(t, prefix+"keep", "", 443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id)

	svc := InboundService{}
	if err := svc.ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node); err != nil {
		t.Fatalf("ReconcileNode: %v", err)
	}

	got := deletedIDs()
	if len(got) != 1 || got[0] != 2 {
		t.Fatalf("deleted remote ids = %v, want [2] (prefixed selected tag must be swept, unmanaged 3 must survive)", got)
	}
}

// Saving the node form marks the node dirty in the same transaction that grows
// its managed set, so reconcile would sweep a tag the panel has not imported yet.
func TestReconcileNode_SaveGrowingSelectionRearmsSweepGuard(t *testing.T) {
	cases := []struct {
		name        string
		storedTags  []string
		mode        string
		tags        []string
		wantDeleted []int
	}{
		{
			name:        "newly selected tag is imported, not swept",
			storedTags:  []string{"keep"},
			mode:        "selected",
			tags:        []string{"keep", "fresh"},
			wantDeleted: nil,
		},
		{
			name:        "switch to all mode imports before sweeping",
			storedTags:  []string{"keep"},
			mode:        "all",
			wantDeleted: nil,
		},
		{
			name:        "unchanged selection still sweeps a deleted tag",
			storedTags:  []string{"keep", "gone"},
			mode:        "selected",
			tags:        []string{"keep", "gone"},
			wantDeleted: []int{3},
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			setupConflictDB(t)
			ts, deletedIDs := fakeNodePanel(t, map[string]int{"keep": 1, "fresh": 2, "gone": 3})
			node := reconcileTestNode(t, ts, "grow-node", "selected", tc.storedTags)
			seedInboundConflictNode(t, "keep", "", 443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id)

			err := (&NodeService{}).UpdateFromRequest(node.Id, &NodeMutationRequest{
				Name:                node.Name,
				Scheme:              node.Scheme,
				Address:             node.Address,
				Port:                node.Port,
				BasePath:            node.BasePath,
				Enable:              true,
				AllowPrivateAddress: true,
				InboundSyncMode:     tc.mode,
				InboundTags:         tc.tags,
			})
			if err != nil {
				t.Fatalf("UpdateFromRequest: %v", err)
			}
			saved := &model.Node{}
			if err := database.GetDB().First(saved, node.Id).Error; err != nil {
				t.Fatalf("reload node: %v", err)
			}

			svc := InboundService{}
			if err := svc.ReconcileNode(context.Background(), runtime.NewRemote(saved, nil), saved); err != nil {
				t.Fatalf("ReconcileNode: %v", err)
			}
			if got := deletedIDs(); !slices.Equal(got, tc.wantDeleted) {
				t.Fatalf("deleted remote ids = %v, want %v", got, tc.wantDeleted)
			}
		})
	}
}