feat: allow selecting inbounds synchronized from nodes (#5178)

* feat: select node inbounds for synchronization

Allow node owners to import either all remote inbounds or an explicit tag-based selection. Add remote inbound discovery, persistence, snapshot filtering, API documentation, tests, and localized UI labels.

* fix

* fix: scope node reconcile and orphan sweep to selected inbound tags

In 'selected' sync mode unselected inbounds never enter the panel DB, so
ReconcileNode treated them as undesired and deleted them from the node the
first time it went config-dirty. Reconcile now only sweeps remote tags that
are part of the selection; everything else on the node is unmanaged.

Panel-created or renamed inbounds on a selected-mode node also vanished:
their tag was outside the selection, so the next traffic pull filtered them
out of the snapshot and the orphan sweep silently dropped the central row.
AddInbound/UpdateInbound now allow the tag on the node before committing.

---------

Co-authored-by: Sanaei <ho3ein.sanaei@gmail.com>

internal/web/service/inbound.go

@@ -559,6 +559,14 @@ func (s *InboundService) AddInbound(inbound *model.Inbound) (*model.Inbound, boo
 		return inbound, false, err
 	}
 
+	// Before the deferred commit, so a node in "selected" sync mode cannot
+	// sweep the new central row in the gap before its tag is allowed.
+	if inbound.NodeID != nil {
+		if aErr := (&NodeService{}).EnsureInboundTagAllowed(*inbound.NodeID, inbound.Tag); aErr != nil {
+			logger.Warning("allow inbound tag on node failed:", aErr)
+		}
+	}
+
 	needRestart := false
 	if inbound.Enable {
 		rt, push, dirty, perr := s.nodePushPlan(inbound)
@@ -949,6 +957,14 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
 		}
 	}
 
+	// A rename must allow the new tag before the deferred commit, or a node in
+	// "selected" sync mode would sweep the renamed central row on the next pull.
+	if oldInbound.NodeID != nil {
+		if aErr := (&NodeService{}).EnsureInboundTagAllowed(*oldInbound.NodeID, oldInbound.Tag); aErr != nil {
+			logger.Warning("allow inbound tag on node failed:", aErr)
+		}
+	}
+
 	if err = tx.Save(oldInbound).Error; err != nil {
 		return inbound, false, err
 	}

internal/web/service/inbound_node.go

@@ -76,10 +76,11 @@ func (s *InboundService) AnyNodePending(inboundIds []int) bool {
 	return false
 }
 
-func (s *InboundService) ReconcileNode(ctx context.Context, rt *runtime.Remote, nodeID int) error {
-	if rt == nil || nodeID <= 0 {
+func (s *InboundService) ReconcileNode(ctx context.Context, rt *runtime.Remote, n *model.Node) error {
+	if rt == nil || n == nil || n.Id <= 0 {
 		return nil
 	}
+	nodeID := n.Id
 	db := database.GetDB()
 	var inbounds []*model.Inbound
 	if err := db.Model(model.Inbound{}).Where("node_id = ?", nodeID).Find(&inbounds).Error; err != nil {
@@ -104,10 +105,26 @@ func (s *InboundService) ReconcileNode(ctx context.Context, rt *runtime.Remote,
 			return fmt.Errorf("reconcile inbound %q: %w", ib.Tag, err)
 		}
 	}
+	// In "selected" sync mode the panel only manages the selected tags: the
+	// rest were never imported, so their absence from the local DB must not
+	// delete them from the node. Only a selected tag missing locally (the
+	// panel deleted it while the node was unreachable) may be swept.
+	var selected map[string]struct{}
+	if n.InboundSyncMode == "selected" {
+		selected = make(map[string]struct{}, len(n.InboundTags))
+		for _, tag := range n.InboundTags {
+			selected[tag] = struct{}{}
+		}
+	}
 	for _, tag := range remoteTags {
 		if _, want := desiredTags[tag]; want {
 			continue
 		}
+		if selected != nil {
+			if _, managed := selected[tag]; !managed {
+				continue
+			}
+		}
 		if err := rt.DelInbound(ctx, &model.Inbound{Tag: tag}); err != nil {
 			return fmt.Errorf("reconcile delete %q: %w", tag, err)
 		}

internal/web/service/inbound_node_reconcile_test.go

@@ -0,0 +1,197 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"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,
+	}
+	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), 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), 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)
+	}
+}
+
+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)
+	}
+}

internal/web/service/node.go

@@ -347,6 +347,25 @@ func (s *NodeService) normalize(n *model.Node) error {
 		n.TlsVerifyMode = "verify"
 	}
 	n.PinnedCertSha256 = strings.TrimSpace(n.PinnedCertSha256)
+	if n.InboundSyncMode != "selected" {
+		n.InboundSyncMode = "all"
+		n.InboundTags = nil
+	} else {
+		seen := make(map[string]struct{}, len(n.InboundTags))
+		tags := make([]string, 0, len(n.InboundTags))
+		for _, tag := range n.InboundTags {
+			tag = strings.TrimSpace(tag)
+			if tag == "" {
+				continue
+			}
+			if _, ok := seen[tag]; ok {
+				continue
+			}
+			seen[tag] = struct{}{}
+			tags = append(tags, tag)
+		}
+		n.InboundTags = tags
+	}
 	if n.TlsVerifyMode == "pin" {
 		if _, err := decodeCertPin(n.PinnedCertSha256); err != nil {
 			return common.NewError(err.Error())
@@ -368,6 +387,10 @@ func (s *NodeService) Update(id int, in *model.Node) error {
 	if err := s.normalize(in); err != nil {
 		return err
 	}
+	inboundTagsJSON, err := json.Marshal(in.InboundTags)
+	if err != nil {
+		return err
+	}
 	db := database.GetDB()
 	existing := &model.Node{}
 	if err := db.Where("id = ?", id).First(existing).Error; err != nil {
@@ -385,6 +408,8 @@ func (s *NodeService) Update(id int, in *model.Node) error {
 		"allow_private_address": in.AllowPrivateAddress,
 		"tls_verify_mode":       in.TlsVerifyMode,
 		"pinned_cert_sha256":    in.PinnedCertSha256,
+		"inbound_sync_mode":     in.InboundSyncMode,
+		"inbound_tags":          string(inboundTagsJSON),
 	}
 	if err := db.Model(model.Node{}).Where("id = ?", id).Updates(updates).Error; err != nil {
 		return err
@@ -395,6 +420,66 @@ func (s *NodeService) Update(id int, in *model.Node) error {
 	return nil
 }
 
+func (s *NodeService) GetRemoteInboundOptions(ctx context.Context, n *model.Node) ([]runtime.RemoteInboundOption, error) {
+	if err := s.normalize(n); err != nil {
+		return nil, err
+	}
+	return runtime.NewRemote(n).ListInboundOptions(ctx)
+}
+
+// EnsureInboundTagAllowed adds a panel-managed inbound's tag to the node's
+// selection when the node syncs in "selected" mode. Without it, the next
+// traffic sync would filter the tag out of the snapshot and the orphan sweep
+// would silently delete the central row the panel just created or renamed.
+// Tags are only ever added (never removed): on a rename the node may keep
+// reporting the old tag until the remote update lands, and a leftover entry
+// that matches nothing is harmless.
+func (s *NodeService) EnsureInboundTagAllowed(nodeID int, tag string) error {
+	tag = strings.TrimSpace(tag)
+	if nodeID <= 0 || tag == "" {
+		return nil
+	}
+	db := database.GetDB()
+	node := &model.Node{}
+	if err := db.Where("id = ?", nodeID).First(node).Error; err != nil {
+		return err
+	}
+	if node.InboundSyncMode != "selected" {
+		return nil
+	}
+	for _, t := range node.InboundTags {
+		if t == tag {
+			return nil
+		}
+	}
+	buf, err := json.Marshal(append(node.InboundTags, tag))
+	if err != nil {
+		return err
+	}
+	return db.Model(model.Node{}).Where("id = ?", nodeID).
+		Updates(map[string]any{"inbound_tags": string(buf)}).Error
+}
+
+func FilterNodeSnapshot(n *model.Node, snap *runtime.TrafficSnapshot) {
+	if n == nil || snap == nil || n.InboundSyncMode != "selected" {
+		return
+	}
+	allowed := make(map[string]struct{}, len(n.InboundTags))
+	for _, tag := range n.InboundTags {
+		allowed[tag] = struct{}{}
+	}
+	filtered := make([]*model.Inbound, 0, len(snap.Inbounds))
+	for _, inbound := range snap.Inbounds {
+		if inbound == nil {
+			continue
+		}
+		if _, ok := allowed[inbound.Tag]; ok {
+			filtered = append(filtered, inbound)
+		}
+	}
+	snap.Inbounds = filtered
+}
+
 func (s *NodeService) Delete(id int) error {
 	db := database.GetDB()
 	if err := db.Where("id = ?", id).Delete(model.Node{}).Error; err != nil {

internal/web/service/node_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+	"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
 )
 
 func TestNormalizeBasePath(t *testing.T) {
@@ -160,3 +161,54 @@ func TestNodeService_Normalize_OverridesUnknownScheme(t *testing.T) {
 		t.Fatalf("Scheme = %q, want https", n.Scheme)
 	}
 }
+
+func TestNodeService_NormalizeInboundSelection(t *testing.T) {
+	s := &NodeService{}
+	n := &model.Node{
+		Name:            "n",
+		Address:         "example.com",
+		Port:            443,
+		InboundSyncMode: "selected",
+		InboundTags:     []string{" alpha ", "", "beta", "alpha"},
+	}
+	if err := s.normalize(n); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if n.InboundSyncMode != "selected" {
+		t.Fatalf("InboundSyncMode = %q, want selected", n.InboundSyncMode)
+	}
+	if len(n.InboundTags) != 2 || n.InboundTags[0] != "alpha" || n.InboundTags[1] != "beta" {
+		t.Fatalf("InboundTags = %#v, want [alpha beta]", n.InboundTags)
+	}
+}
+
+func TestFilterNodeSnapshot(t *testing.T) {
+	snapshot := func() *runtime.TrafficSnapshot {
+		return &runtime.TrafficSnapshot{Inbounds: []*model.Inbound{
+			{Tag: "alpha"},
+			{Tag: "beta"},
+			{Tag: "gamma"},
+		}}
+	}
+
+	all := snapshot()
+	FilterNodeSnapshot(&model.Node{InboundSyncMode: "all"}, all)
+	if len(all.Inbounds) != 3 {
+		t.Fatalf("all mode kept %d inbounds, want 3", len(all.Inbounds))
+	}
+
+	selected := snapshot()
+	FilterNodeSnapshot(&model.Node{
+		InboundSyncMode: "selected",
+		InboundTags:     []string{"beta"},
+	}, selected)
+	if len(selected.Inbounds) != 1 || selected.Inbounds[0].Tag != "beta" {
+		t.Fatalf("selected mode produced %#v, want only beta", selected.Inbounds)
+	}
+
+	none := snapshot()
+	FilterNodeSnapshot(&model.Node{InboundSyncMode: "selected"}, none)
+	if len(none.Inbounds) != 0 {
+		t.Fatalf("empty selection kept %d inbounds, want 0", len(none.Inbounds))
+	}
+}