diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 793a759..7aa194c 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -439,6 +439,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi nodeRow.Id = nodeID unmanagedTag := unmanagedTagPredicate(&nodeRow) selfKey := effectiveNodeKey(&model.Node{Id: nodeID, Guid: nodeRow.Guid}) + guidShared := nodeRow.Guid != "" && selfKey != nodeRow.Guid originGuidFor := func(snapIb *model.Inbound) string { if snapIb.OriginNodeGuid != "" && snapIb.OriginNodeGuid != nodeRow.Guid { return snapIb.OriginNodeGuid @@ -1267,6 +1268,32 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } } + tree := snap.OnlineTree + switch { + case len(tree) == 0 && len(snap.OnlineEmails) > 0: + // Old-build node (no GUID tree): key its flat online list under its + // own effective identity so attribution still works for that branch. + tree = map[string][]string{selfKey: snap.OnlineEmails} + case guidShared && len(tree) > 0: + // Newer cloned node: its own clients arrive keyed under the shared + // panelGuid. Remap just that entry to the node-unique key so the + // clones don't merge; descendant subtrees keep their distinct GUIDs. + if _, ok := tree[nodeRow.Guid]; ok { + tree = remapGuidTreeKey(tree, nodeRow.Guid, selfKey) + } + } + s.SetNodeOnlineTree(nodeID, tree) + + activeTree := normalizeActiveInboundTreeTags(snap.ActiveInboundTree, tagToCentral) + if guidShared && len(activeTree) > 0 { + if _, ok := activeTree[nodeRow.Guid]; ok { + activeTree = remapGuidTreeKey(activeTree, nodeRow.Guid, selfKey) + } + } + if len(activeTree) > 0 { + activeTree = filterGuidTreeKeys(activeTree, activeInboundGuidKeys(snap.Inbounds, tagToCentral, originGuidFor)) + } + s.SetNodeActiveInboundTree(nodeID, activeTree) return structuralChange, nil } @@ -1294,8 +1321,11 @@ func (s *InboundService) restartRemoteNodesOnDisable(nodeIDs []int) { } func (s *InboundService) GetOnlineClients() []string { - // Lite master: no local core aggregates online clients; node onlines - // arrive through agent reports and node polling instead. + // Lite master: no local core aggregates online clients; the online + // store (see online_store.go) is fed by agent reports and node polling. + if out := sharedOnlineStore.GetOnlineClients(); out != nil { + return out + } return []string{} } @@ -1305,7 +1335,13 @@ func (s *InboundService) GetOnlineClients() []string { // node-id keying so a client three hops down is attributed to its real node, // not the intermediate one it was synced through. func (s *InboundService) GetOnlineClientsByGuid() map[string][]string { - return map[string][]string{} + out := sharedOnlineStore.GetMergedNodeTrees() + if local := sharedOnlineStore.GetLocalOnlineClients(); len(local) > 0 { + if guid := s.panelGuid(); guid != "" { + out[guid] = mergeEmails(out[guid], local) + } + } + return out } // GetActiveInboundsByGuid returns the inbound tags that carried traffic within @@ -1313,12 +1349,30 @@ func (s *InboundService) GetOnlineClientsByGuid() map[string][]string { // each inbound. A GUID missing from the map means "don't gate" for that node's // inbounds (old-build node or no active-inbound signal). func (s *InboundService) GetActiveInboundsByGuid() map[string][]string { - return map[string][]string{} + out := sharedOnlineStore.GetMergedActiveInboundTrees() + active := sharedOnlineStore.GetLocalActiveInbounds() + if len(active) == 0 { + return out + } + guid := s.panelGuid() + if guid == "" { + return out + } + out[guid] = mergeEmails(out[guid], active) + return out } -func (s *InboundService) SetNodeOnlineTree(nodeID int, tree map[string][]string) {} +func (s *InboundService) SetNodeOnlineTree(nodeID int, tree map[string][]string) { + sharedOnlineStore.SetNodeOnlineTree(nodeID, tree) +} -func (s *InboundService) ClearNodeOnlineClients(nodeID int) {} +func (s *InboundService) SetNodeActiveInboundTree(nodeID int, tree map[string][]string) { + sharedOnlineStore.SetNodeActiveInboundTree(nodeID, tree) +} + +func (s *InboundService) ClearNodeOnlineClients(nodeID int) { + sharedOnlineStore.ClearNodeOnlineClients(nodeID) +} // panelGuid returns this panel's stable self-identifier, used to key the local // panel's own clients in the per-node online maps (#4983). @@ -1451,11 +1505,13 @@ func (s *InboundService) GetClientsLastOnline() (map[string]int64, error) { } // RefreshLocalOnlineClients folds the emails and inbound tags active on this -// panel's own xray this poll into the local online/active sets, applying the -// grace window and pruning stale entries. Pass nil to only prune. See -// xray.Process for why the local sets are kept separate from the shared +// panel's own inbounds this poll into the local online/active sets, applying +// the grace window and pruning stale entries. Pass nil to only prune. See +// onlineStore for why the local sets are kept separate from the shared // last_online column. -func (s *InboundService) RefreshLocalOnlineClients(activeEmails, activeInboundTags []string) {} +func (s *InboundService) RefreshLocalOnlineClients(activeEmails, activeInboundTags []string) { + sharedOnlineStore.RefreshLocalOnline(activeEmails, activeInboundTags, time.Now().UnixMilli(), onlineGracePeriodMs) +} func (s *InboundService) FilterAndSortClientEmails(emails []string) ([]string, []string, error) { db := database.GetDB() diff --git a/internal/web/service/inbound_node_online_integration_test.go b/internal/web/service/inbound_node_online_integration_test.go new file mode 100644 index 0000000..cf0cf8c --- /dev/null +++ b/internal/web/service/inbound_node_online_integration_test.go @@ -0,0 +1,122 @@ +package service + +import ( + "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" +) + +// The traditional-node path: NodeTrafficSyncJob fetches a TrafficSnapshot +// from each node panel (runtime.Remote.FetchTrafficSnapshot) and hands it +// to InboundService.SetRemoteTraffic. Before the online-store restoration, +// the snapshot's OnlineTree / ActiveInboundTree were dropped on the floor +// inside that call (the store setters were no-op stubs), so polled nodes +// never showed anyone online. This drives the exact public entry the job +// calls and asserts the snapshot's online data lands in the store. +func TestSetRemoteTraffic_FeedsOnlineStore(t *testing.T) { + setupConflictDB(t) + db := database.GetDB() + svc := InboundService{} + + const nodeID = 4242 + const nodeGuid = "trad-node-guid-4242" + svc.ClearNodeOnlineClients(nodeID) + t.Cleanup(func() { svc.ClearNodeOnlineClients(nodeID) }) + + if err := db.Create(&model.Node{ + Id: nodeID, + Name: "trad-node", + Address: "10.0.4.2", + Port: 2053, + ApiToken: "t", + Guid: nodeGuid, + }).Error; err != nil { + t.Fatalf("create node: %v", err) + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{ + { + Tag: "in-443-tcp", + Enable: true, + Port: 443, + Protocol: model.VLESS, + Settings: `{"clients":[{"email":"trad@example.com"}]}`, + }, + }, + OnlineTree: map[string][]string{nodeGuid: {"trad@example.com"}}, + ActiveInboundTree: map[string][]string{nodeGuid: {"in-443-tcp"}}, + } + + if _, err := svc.SetRemoteTraffic(nodeID, snap, false, false); err != nil { + t.Fatalf("SetRemoteTraffic: %v", err) + } + + assertStringSet(t, "online union", svc.GetOnlineClients(), []string{"trad@example.com"}) + byGuid := svc.GetOnlineClientsByGuid() + assertStringSet(t, "online byGuid", byGuid[nodeGuid], []string{"trad@example.com"}) + + // The active tree is normalized from node-side tags to central tags; + // whatever tag the adopted central inbound carries is what must show. + var central model.Inbound + if err := db.Where("node_id = ?", nodeID).First(¢ral).Error; err != nil { + t.Fatalf("load central inbound: %v", err) + } + active := svc.GetActiveInboundsByGuid() + assertStringSet(t, "active byGuid", active[nodeGuid], []string{central.Tag}) + + // A failed probe makes the job clear the node; its clients must stop + // showing as online immediately. + svc.ClearNodeOnlineClients(nodeID) + if got := svc.GetOnlineClients(); len(got) != 0 { + t.Fatalf("online union after clear = %v, want empty", got) + } +} + +// Old-build nodes have no GUID tree endpoint: the job passes their flat +// OnlineEmails in the snapshot and SetRemoteTraffic must key them under +// the node's own effective identity. +func TestSetRemoteTraffic_FlatOnlineEmailsFallback(t *testing.T) { + setupConflictDB(t) + db := database.GetDB() + svc := InboundService{} + + const nodeID = 4243 + const nodeGuid = "flat-node-guid-4243" + svc.ClearNodeOnlineClients(nodeID) + t.Cleanup(func() { svc.ClearNodeOnlineClients(nodeID) }) + + if err := db.Create(&model.Node{ + Id: nodeID, + Name: "flat-node", + Address: "10.0.4.3", + Port: 2053, + ApiToken: "t", + Guid: nodeGuid, + }).Error; err != nil { + t.Fatalf("create node: %v", err) + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{ + { + Tag: "in-8443-tcp", + Enable: true, + Port: 8443, + Protocol: model.VLESS, + Settings: `{"clients":[{"email":"flat@example.com"}]}`, + }, + }, + OnlineEmails: []string{"flat@example.com"}, + } + + if _, err := svc.SetRemoteTraffic(nodeID, snap, false, false); err != nil { + t.Fatalf("SetRemoteTraffic: %v", err) + } + + assertStringSet(t, "online union (flat)", svc.GetOnlineClients(), []string{"flat@example.com"}) + byGuid := svc.GetOnlineClientsByGuid() + assertStringSet(t, "online byGuid (flat)", byGuid[nodeGuid], []string{"flat@example.com"}) +} diff --git a/internal/web/service/online_store.go b/internal/web/service/online_store.go new file mode 100644 index 0000000..c52b0e5 --- /dev/null +++ b/internal/web/service/online_store.go @@ -0,0 +1,302 @@ +package service + +import "sync" + +// onlineStore is the panel's in-memory online-clients store: which client +// emails are currently online, and on which inbound tags, keyed by the +// panelGuid of the node that physically hosts them (#4983). +// +// This is the restored pre-P9 implementation. The store originally lived on +// xray.Process (the local core handle) and the service wrappers reached it +// through currentXrayProcess(); P9 deleted the local core — and with it the +// only Process instance — but kept every data-arrival path (node polling, +// agent reports) calling the service setters, which were left as no-op +// stubs, so online state silently stayed empty. The lite master has no +// local core to hang the store on, so it now lives here as a package-level +// singleton shared by every InboundService value (services are constructed +// as bare values in many places, so an instance field would fragment the +// state between writers — agent controller, sync job — and readers — +// client controller, paging service). The algorithms, locking and +// grace-window semantics are ported verbatim from xray.Process. +// +// The local sets (onlineClients / localActiveInbounds) track THIS panel's +// own inbounds only. On the lite master no local core feeds them — only +// RefreshLocalOnlineClients fold-ins (e.g. the mtproto job's local actives) +// — while node sets arrive per node through the Set*Tree setters. The local +// sets are kept independent of the shared client_traffics.last_online +// column: that column is bumped by remote-node syncs too and would +// otherwise leak remote-only clients into the local set. +type onlineStore struct { + // onlineClients is the set of emails active on THIS panel's own + // inbounds within the online grace window. It is derived only from + // local fold-ins (see RefreshLocalOnline) — never from remote-node + // snapshots — so a client connected solely to a remote node is not + // reported online on local inbounds. + onlineClients []string + // localActiveInbounds is the set of THIS panel's inbound tags that + // carried traffic within the same grace window. A user>>>email stat + // aggregates across every inbound a client is attached to, so an + // online email alone can't say which inbound it actually used. Pairing + // it with the inbound>>>tag stat lets the per-inbound view drop a + // multi-inbound client from inbounds that saw no traffic this window. + localActiveInbounds []string + // localLastOnline records, per email, the last time a local fold-in + // reported traffic for it. RefreshLocalOnline rebuilds onlineClients + // from this map each tick. + localLastOnline map[string]int64 + // localInboundLastActive mirrors localLastOnline for inbound tags: the + // last tick traffic was reported through each tag. Rebuilt into + // localActiveInbounds under the same grace window so the two signals + // stay aligned — an email within grace always has the inbound it used + // within grace too. + localInboundLastActive map[string]int64 + // nodeOnlineTrees holds, per direct remote node (keyed by that node's + // panel-local id), the GUID-keyed online-emails subtree that node + // reported — its own clients under its panelGuid plus every descendant + // under theirs. Keying the stored value by GUID (not node id) lets the + // master attribute a deeply nested client to the node that physically + // hosts it across a chain (#4983); the outer node-id key is only so a + // failed probe can drop that whole branch's contribution. + // NodeTrafficSyncJob populates entries per cron tick and clears them + // when a probe fails. The mutex guards this map, onlineClients, and + // localLastOnline above so the online getters never see a torn read. + nodeOnlineTrees map[int]map[string][]string + // nodeActiveInboundTrees mirrors nodeOnlineTrees for active inbound + // tags: each direct node reports a GUID-keyed subtree of inbound tags + // that carried traffic within its own grace window. The inbounds page + // combines this with nodeOnlineTrees so a multi-inbound client is not + // shown on an idle inbound. + nodeActiveInboundTrees map[int]map[string][]string + mu sync.RWMutex +} + +// sharedOnlineStore is the one online store for the whole panel process — +// the counterpart of the single xray.Process the pre-P9 code reached via +// currentXrayProcess(). +var sharedOnlineStore = &onlineStore{} + +// GetOnlineClients returns the union of locally-online clients and +// node-online clients from every registered remote panel. Dedupes by +// email so a client connected to both a local and a node-managed inbound +// surfaces once. Cheap allocation — typical online sets are small and +// the union is recomputed on demand. +func (p *onlineStore) GetOnlineClients() []string { + p.mu.RLock() + defer p.mu.RUnlock() + + if len(p.nodeOnlineTrees) == 0 { + // Hot path for single-panel deployments: avoid the map+dedupe + // work entirely and return the local slice as-is. + return p.onlineClients + } + + seen := make(map[string]struct{}, len(p.onlineClients)) + out := make([]string, 0, len(p.onlineClients)) + add := func(emails []string) { + for _, email := range emails { + if _, dup := seen[email]; dup { + continue + } + seen[email] = struct{}{} + out = append(out, email) + } + } + add(p.onlineClients) + for _, tree := range p.nodeOnlineTrees { + for _, emails := range tree { + add(emails) + } + } + return out +} + +// GetLocalOnlineClients returns a copy of the emails online on THIS panel's +// own inbounds within the grace window. The service layer keys these under +// the panel's own GUID when assembling the per-node online view. +func (p *onlineStore) GetLocalOnlineClients() []string { + p.mu.RLock() + defer p.mu.RUnlock() + if len(p.onlineClients) == 0 { + return nil + } + out := make([]string, len(p.onlineClients)) + copy(out, p.onlineClients) + return out +} + +// GetMergedNodeTrees returns the union of every direct node's reported +// subtree, keyed by the panelGuid of the node that physically hosts each +// client set. Because each child already reports its descendants under +// their own GUIDs, merging the direct children yields the whole tree at +// any depth (#4983), so a client three hops down is attributed to its real +// node, not the intermediate one. GUIDs are globally unique, but a set +// reported under the same GUID by more than one path is deduped per key; +// empty sets are omitted. +func (p *onlineStore) GetMergedNodeTrees() map[string][]string { + p.mu.RLock() + defer p.mu.RUnlock() + if len(p.nodeOnlineTrees) == 0 { + return map[string][]string{} + } + out := make(map[string][]string) + seen := make(map[string]map[string]struct{}) + for _, tree := range p.nodeOnlineTrees { + for guid, emails := range tree { + if guid == "" || len(emails) == 0 { + continue + } + dedup := seen[guid] + if dedup == nil { + dedup = make(map[string]struct{}, len(emails)) + seen[guid] = dedup + } + for _, email := range emails { + if _, ok := dedup[email]; ok { + continue + } + dedup[email] = struct{}{} + out[guid] = append(out[guid], email) + } + } + } + return out +} + +// GetLocalActiveInbounds returns a copy of THIS panel's inbound tags that +// carried traffic within the grace window. The service layer keys these +// under the panel's own GUID before merging them with remote-node +// active-inbound subtrees. +func (p *onlineStore) GetLocalActiveInbounds() []string { + p.mu.RLock() + defer p.mu.RUnlock() + if len(p.localActiveInbounds) == 0 { + return nil + } + out := make([]string, len(p.localActiveInbounds)) + copy(out, p.localActiveInbounds) + return out +} + +// GetMergedActiveInboundTrees returns the union of every direct node's +// reported active-inbound subtree, keyed by the panelGuid of the node that +// physically hosts each inbound. Duplicate tags reported through multiple +// paths are deduped per GUID. +func (p *onlineStore) GetMergedActiveInboundTrees() map[string][]string { + p.mu.RLock() + defer p.mu.RUnlock() + if len(p.nodeActiveInboundTrees) == 0 { + return map[string][]string{} + } + out := make(map[string][]string) + seen := make(map[string]map[string]struct{}) + for _, tree := range p.nodeActiveInboundTrees { + for guid, tags := range tree { + if guid == "" || len(tags) == 0 { + continue + } + dedup := seen[guid] + if dedup == nil { + dedup = make(map[string]struct{}, len(tags)) + seen[guid] = dedup + } + for _, tag := range tags { + if tag == "" { + continue + } + if _, ok := dedup[tag]; ok { + continue + } + dedup[tag] = struct{}{} + out[guid] = append(out[guid], tag) + } + } + } + return out +} + +// RefreshLocalOnline records that each email in activeEmails and each tag +// in activeInboundTags had local traffic at now, then rebuilds +// onlineClients and localActiveInbounds from every entry seen within +// graceMs, pruning older ones. Pass nil/empty slices to only prune — +// NodeTrafficSyncJob does this so stale local entries still age out +// between fold-ins. +func (p *onlineStore) RefreshLocalOnline(activeEmails, activeInboundTags []string, now, graceMs int64) { + p.mu.Lock() + defer p.mu.Unlock() + if p.localLastOnline == nil { + p.localLastOnline = make(map[string]int64, len(activeEmails)) + } + for _, email := range activeEmails { + p.localLastOnline[email] = now + } + online := make([]string, 0, len(p.localLastOnline)) + for email, ts := range p.localLastOnline { + if now-ts < graceMs { + online = append(online, email) + } else { + delete(p.localLastOnline, email) + } + } + p.onlineClients = online + + if p.localInboundLastActive == nil { + p.localInboundLastActive = make(map[string]int64, len(activeInboundTags)) + } + for _, tag := range activeInboundTags { + p.localInboundLastActive[tag] = now + } + activeInbounds := make([]string, 0, len(p.localInboundLastActive)) + for tag, ts := range p.localInboundLastActive { + if now-ts < graceMs { + activeInbounds = append(activeInbounds, tag) + } else { + delete(p.localInboundLastActive, tag) + } + } + p.localActiveInbounds = activeInbounds +} + +// SetNodeOnlineTree records the GUID-keyed online subtree one direct remote +// node reported (its own clients under its panelGuid plus every descendant +// under theirs). Replaces any previous entry for that node — +// NodeTrafficSyncJob always sends the full subtree per tick, and the agent +// controller sends its online list per traffic report. +func (p *onlineStore) SetNodeOnlineTree(nodeID int, tree map[string][]string) { + p.mu.Lock() + defer p.mu.Unlock() + if len(tree) == 0 { + delete(p.nodeOnlineTrees, nodeID) + return + } + if p.nodeOnlineTrees == nil { + p.nodeOnlineTrees = map[int]map[string][]string{} + } + p.nodeOnlineTrees[nodeID] = tree +} + +// SetNodeActiveInboundTree records the GUID-keyed active-inbound subtree +// one direct remote node reported. Replaces any previous entry for that +// node. +func (p *onlineStore) SetNodeActiveInboundTree(nodeID int, tree map[string][]string) { + p.mu.Lock() + defer p.mu.Unlock() + if len(tree) == 0 { + delete(p.nodeActiveInboundTrees, nodeID) + return + } + if p.nodeActiveInboundTrees == nil { + p.nodeActiveInboundTrees = map[int]map[string][]string{} + } + p.nodeActiveInboundTrees[nodeID] = tree +} + +// ClearNodeOnlineClients drops a direct node's whole subtree contribution +// (online emails and active inbounds). Called when a probe fails or an +// agent disconnects, so a downed node — and everything behind it — doesn't +// keep its clients listed as "online" until the next successful probe. +func (p *onlineStore) ClearNodeOnlineClients(nodeID int) { + p.mu.Lock() + defer p.mu.Unlock() + delete(p.nodeOnlineTrees, nodeID) + delete(p.nodeActiveInboundTrees, nodeID) +} diff --git a/internal/web/service/online_store_test.go b/internal/web/service/online_store_test.go new file mode 100644 index 0000000..1256784 --- /dev/null +++ b/internal/web/service/online_store_test.go @@ -0,0 +1,193 @@ +package service + +import ( + "fmt" + "sync" + "testing" + "time" +) + +func sortedSet(in []string) map[string]bool { + out := make(map[string]bool, len(in)) + for _, s := range in { + out[s] = true + } + return out +} + +func requireSet(t *testing.T, what string, got []string, want ...string) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("%s: got %v, want set %v", what, got, want) + } + set := sortedSet(got) + for _, w := range want { + if !set[w] { + t.Fatalf("%s: got %v, missing %q (want %v)", what, got, w, want) + } + } +} + +// A node's reported tree must surface both in the flat union and in the +// GUID-keyed view, deduped against the local set and other nodes. +func TestOnlineStoreSetTreeUnionAndByGuid(t *testing.T) { + st := &onlineStore{} + st.SetNodeOnlineTree(1, map[string][]string{ + "guid-a": {"alice@example.com", "bob@example.com"}, + "guid-b": {"carol@example.com"}, + }) + st.SetNodeOnlineTree(2, map[string][]string{ + "guid-c": {"bob@example.com", "dave@example.com"}, + }) + // Local fold-in overlapping one node email (dedupe check). + st.RefreshLocalOnline([]string{"alice@example.com", "local@example.com"}, nil, 1000, 20000) + + requireSet(t, "union", st.GetOnlineClients(), + "alice@example.com", "bob@example.com", "carol@example.com", "dave@example.com", "local@example.com") + + merged := st.GetMergedNodeTrees() + requireSet(t, "byGuid guid-a", merged["guid-a"], "alice@example.com", "bob@example.com") + requireSet(t, "byGuid guid-b", merged["guid-b"], "carol@example.com") + requireSet(t, "byGuid guid-c", merged["guid-c"], "bob@example.com", "dave@example.com") + if _, ok := merged["guid-absent"]; ok { + t.Fatalf("byGuid: unexpected key for absent guid: %v", merged) + } + requireSet(t, "local", st.GetLocalOnlineClients(), "alice@example.com", "local@example.com") +} + +// The agent controller reports onlines under the literal "agent" GUID key; +// they must reach the union untouched. +func TestOnlineStoreAgentKeyTree(t *testing.T) { + st := &onlineStore{} + st.SetNodeOnlineTree(7, map[string][]string{"agent": {"agentuser@example.com"}}) + requireSet(t, "union", st.GetOnlineClients(), "agentuser@example.com") + requireSet(t, "byGuid agent", st.GetMergedNodeTrees()["agent"], "agentuser@example.com") +} + +// Setting a tree replaces the node's previous contribution; an empty tree +// deletes it outright. +func TestOnlineStoreSetTreeReplacesAndEmptyDeletes(t *testing.T) { + st := &onlineStore{} + st.SetNodeOnlineTree(3, map[string][]string{"g": {"one@example.com"}}) + st.SetNodeOnlineTree(3, map[string][]string{"g": {"two@example.com"}}) + requireSet(t, "union after replace", st.GetOnlineClients(), "two@example.com") + + st.SetNodeOnlineTree(3, map[string][]string{}) + if got := st.GetOnlineClients(); len(got) != 0 { + t.Fatalf("union after empty set: got %v, want empty", got) + } + if got := st.GetMergedNodeTrees(); len(got) != 0 { + t.Fatalf("byGuid after empty set: got %v, want empty", got) + } +} + +// Active-inbound trees behave like online trees, with per-GUID dedupe. +func TestOnlineStoreActiveInboundTrees(t *testing.T) { + st := &onlineStore{} + st.SetNodeActiveInboundTree(1, map[string][]string{"g1": {"in-a", "in-b"}}) + st.SetNodeActiveInboundTree(2, map[string][]string{"g1": {"in-b", "in-c"}}) + merged := st.GetMergedActiveInboundTrees() + requireSet(t, "active g1", merged["g1"], "in-a", "in-b", "in-c") + + st.RefreshLocalOnline(nil, []string{"local-in"}, 5000, 20000) + requireSet(t, "local active", st.GetLocalActiveInbounds(), "local-in") +} + +// Clear drops BOTH of the node's trees and leaves other nodes untouched. +func TestOnlineStoreClear(t *testing.T) { + st := &onlineStore{} + st.SetNodeOnlineTree(1, map[string][]string{"g1": {"a@example.com"}}) + st.SetNodeActiveInboundTree(1, map[string][]string{"g1": {"in-a"}}) + st.SetNodeOnlineTree(2, map[string][]string{"g2": {"b@example.com"}}) + + st.ClearNodeOnlineClients(1) + requireSet(t, "union after clear", st.GetOnlineClients(), "b@example.com") + if got := st.GetMergedActiveInboundTrees(); len(got) != 0 { + t.Fatalf("active after clear: got %v, want empty", got) + } +} + +// Local entries live for the grace window, then a prune-only refresh +// (nil, nil — what NodeTrafficSyncJob calls every tick) ages them out. +func TestOnlineStoreGraceWindowPrune(t *testing.T) { + st := &onlineStore{} + const grace = int64(20000) + st.RefreshLocalOnline([]string{"stay@example.com"}, []string{"tag-stay"}, 10000, grace) + st.RefreshLocalOnline([]string{"fresh@example.com"}, nil, 25000, grace) + + // At t=25000 the first entry is 15s old: still inside the window. + requireSet(t, "local within grace", st.GetLocalOnlineClients(), "stay@example.com", "fresh@example.com") + + // Prune-only at t=45001: stay@10000 is 35s old (gone), fresh@25000 is + // 20001ms old (also past the strict < grace comparison), tag pruned. + st.RefreshLocalOnline(nil, nil, 45001, grace) + if got := st.GetLocalOnlineClients(); len(got) != 0 { + t.Fatalf("local after prune: got %v, want empty", got) + } + if got := st.GetLocalActiveInbounds(); len(got) != 0 { + t.Fatalf("local active after prune: got %v, want empty", got) + } +} + +// Boundary: an entry exactly graceMs old is expired (original uses <). +func TestOnlineStoreGraceBoundary(t *testing.T) { + st := &onlineStore{} + st.RefreshLocalOnline([]string{"edge@example.com"}, nil, 0, 20000) + st.RefreshLocalOnline(nil, nil, 19999, 20000) + requireSet(t, "just inside", st.GetLocalOnlineClients(), "edge@example.com") + st.RefreshLocalOnline(nil, nil, 20000, 20000) + if got := st.GetLocalOnlineClients(); len(got) != 0 { + t.Fatalf("at boundary: got %v, want empty", got) + } +} + +// The store is shared across InboundService values: a write through one +// instance must be visible through another (services are bare values; an +// instance field would fragment state between writers and readers). +func TestOnlineStoreSharedAcrossServiceInstances(t *testing.T) { + writer := &InboundService{} + reader := &InboundService{} + const nodeID = 987654 + t.Cleanup(func() { writer.ClearNodeOnlineClients(nodeID) }) + + writer.SetNodeOnlineTree(nodeID, map[string][]string{"guid-shared": {"shared@example.com"}}) + requireSet(t, "union via other instance", reader.GetOnlineClients(), "shared@example.com") + byGuid := reader.GetOnlineClientsByGuid() + requireSet(t, "byGuid via other instance", byGuid["guid-shared"], "shared@example.com") +} + +// Concurrent writers/readers must not race or corrupt the union. Run with +// -race for the full guarantee; even without it, the final state is exact. +func TestOnlineStoreConcurrentAccess(t *testing.T) { + st := &onlineStore{} + const nodes = 8 + var wg sync.WaitGroup + for n := 0; n < nodes; n++ { + n := n + wg.Add(1) + go func() { + defer wg.Done() + guid := fmt.Sprintf("guid-%d", n) + for i := 0; i < 50; i++ { + st.SetNodeOnlineTree(n, map[string][]string{guid: {fmt.Sprintf("u%d-%d@example.com", n, i)}}) + _ = st.GetOnlineClients() + _ = st.GetMergedNodeTrees() + st.RefreshLocalOnline([]string{fmt.Sprintf("local-%d@example.com", n)}, nil, time.Now().UnixMilli(), 60000) + } + }() + } + wg.Wait() + // Every node's last write holds exactly one email; locals may or may + // not have been pruned by a same-ms refresh, so assert on node trees. + merged := st.GetMergedNodeTrees() + for n := 0; n < nodes; n++ { + requireSet(t, fmt.Sprintf("tree %d", n), merged[fmt.Sprintf("guid-%d", n)], + fmt.Sprintf("u%d-49@example.com", n)) + } + for n := 0; n < nodes; n++ { + st.ClearNodeOnlineClients(n) + } + if got := st.GetMergedNodeTrees(); len(got) != 0 { + t.Fatalf("trees after mass clear: got %v, want empty", got) + } +}