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