3x-ui-agent/internal/xray/api.go

// Package xray provides integration with the Xray proxy core.
// It includes API client functionality, configuration management, traffic monitoring,
// and process control for Xray instances.
package xray

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"net/netip"
	"regexp"
	"strings"
	"time"

	"github.com/mhsanaei/3x-ui/v3/internal/logger"
	"github.com/mhsanaei/3x-ui/v3/internal/util/common"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/encoding"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protowire"
)

// This build no longer links xray-core. What remains of the core API client:
//
//   - StatsService (QueryStats / GetUsersStats) is implemented at the
//     protobuf wire level with protowire: those two RPCs carry the traffic
//     accounting and online-user data, and their messages are small and
//     stable. The agent's traffic reporting and the local traffic job keep
//     working exactly as before.
//   - The mutation RPCs (HandlerService add/remove/alter, RoutingService
//     rule/balancer operations) cannot be reimplemented without the core's
//     typed config builders — AddInbound hands the core a fully built
//     protobuf handler config, not JSON. They return
//     ErrHotApplyUnavailable instead. Every caller already treats an API
//     failure as "fall back": hot-apply degrades to a full core restart,
//     the balancer view reports not-running, the node bridge probe gives
//     up, and the IP-limit temp-disconnect logs and moves on. On a
//     control-plane master (the lite build's target) the local core never
//     runs, so these paths were unreachable there to begin with.

// ErrHotApplyUnavailable is returned by the core mutation RPCs that this
// build cannot perform without linking xray-core. Callers fall back to
// restarting the core process, which applies the same configuration.
var ErrHotApplyUnavailable = errors.New("xray core hot-apply API unavailable in this build; the change is applied by restarting the core")

// Compiled once at package load: GetTraffic runs on every traffic-stats tick,
// so recompiling these per call is wasted work.
var (
	trafficRegex       = regexp.MustCompile(`(inbound|outbound)>>>([^>]+)>>>traffic>>>(downlink|uplink)`)
	clientTrafficRegex = regexp.MustCompile(`user>>>([^>]+)>>>traffic>>>(downlink|uplink)`)
)

// XrayAPI is a client for the Xray core's gRPC API. Only the stats service
// is functional in this build; see the package comment.
type XrayAPI struct {
	grpcClient      *grpc.ClientConn
	isConnected     bool
	StatsLastValues map[string]int64
}

// Init connects to the Xray API server.
func (x *XrayAPI) Init(apiPort int) error {
	if apiPort <= 0 || apiPort > math.MaxUint16 {
		return fmt.Errorf("invalid Xray API port: %d", apiPort)
	}

	addr := fmt.Sprintf("127.0.0.1:%d", apiPort)
	conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		return fmt.Errorf("failed to connect to Xray API: %w", err)
	}

	x.grpcClient = conn
	x.isConnected = true
	if x.StatsLastValues == nil {
		x.StatsLastValues = make(map[string]int64)
	}

	return nil
}

// Close closes the gRPC connection and resets the XrayAPI client state.
func (x *XrayAPI) Close() {
	if x.grpcClient != nil {
		x.grpcClient.Close()
	}
	x.isConnected = false
}

// rawBytes is a pre-encoded protobuf message passed through the gRPC stack
// untouched by rawCodec.
type rawBytes []byte

// rawCodec is a gRPC codec that forwards rawBytes as-is. Its name keeps the
// standard "proto" content subtype the core's server expects.
type rawCodec struct{}

func (rawCodec) Marshal(v any) ([]byte, error) {
	b, ok := v.(rawBytes)
	if !ok {
		return nil, fmt.Errorf("xray api: raw codec cannot marshal %T", v)
	}
	return b, nil
}

func (rawCodec) Unmarshal(data []byte, v any) error {
	p, ok := v.(*rawBytes)
	if !ok {
		return fmt.Errorf("xray api: raw codec cannot unmarshal into %T", v)
	}
	*p = append((*p)[:0], data...)
	return nil
}

func (rawCodec) Name() string { return "proto" }

var _ encoding.Codec = rawCodec{}

const (
	statsServiceQueryStats    = "/xray.app.stats.command.StatsService/QueryStats"
	statsServiceGetUsersStats = "/xray.app.stats.command.StatsService/GetUsersStats"
)

// statEntry is one counter from a QueryStats response
// (xray.app.stats.command.Stat: name = 1, value = 2).
type statEntry struct {
	Name  string
	Value int64
}

// queryStats calls StatsService.QueryStats with an empty pattern and no
// reset — the same request the typed client sent. Both fields are proto3
// zero values, so the encoded request is empty.
func (x *XrayAPI) queryStats(ctx context.Context) ([]statEntry, error) {
	var resp rawBytes
	if err := x.grpcClient.Invoke(ctx, statsServiceQueryStats, rawBytes{}, &resp, grpc.ForceCodec(rawCodec{})); err != nil {
		return nil, err
	}
	return parseQueryStatsResponse(resp)
}

func parseQueryStatsResponse(b []byte) ([]statEntry, error) {
	var out []statEntry
	for len(b) > 0 {
		num, typ, n := protowire.ConsumeTag(b)
		if n < 0 {
			return nil, protowire.ParseError(n)
		}
		b = b[n:]
		if num == 1 && typ == protowire.BytesType {
			v, n := protowire.ConsumeBytes(b)
			if n < 0 {
				return nil, protowire.ParseError(n)
			}
			st, err := parseStat(v)
			if err != nil {
				return nil, err
			}
			out = append(out, st)
			b = b[n:]
			continue
		}
		n = protowire.ConsumeFieldValue(num, typ, b)
		if n < 0 {
			return nil, protowire.ParseError(n)
		}
		b = b[n:]
	}
	return out, nil
}

func parseStat(b []byte) (statEntry, error) {
	var st statEntry
	for len(b) > 0 {
		num, typ, n := protowire.ConsumeTag(b)
		if n < 0 {
			return st, protowire.ParseError(n)
		}
		b = b[n:]
		switch {
		case num == 1 && typ == protowire.BytesType:
			v, n := protowire.ConsumeBytes(b)
			if n < 0 {
				return st, protowire.ParseError(n)
			}
			st.Name = string(v)
			b = b[n:]
		case num == 2 && typ == protowire.VarintType:
			v, n := protowire.ConsumeVarint(b)
			if n < 0 {
				return st, protowire.ParseError(n)
			}
			st.Value = int64(v)
			b = b[n:]
		default:
			n := protowire.ConsumeFieldValue(num, typ, b)
			if n < 0 {
				return st, protowire.ParseError(n)
			}
			b = b[n:]
		}
	}
	return st, nil
}

// userStatEntry is one user from a GetUsersStats response
// (xray.app.stats.command.UserStat: email = 1, ips = 2;
// OnlineIPEntry: ip = 1, last_seen = 2).
type userStatEntry struct {
	Email string
	IPs   []OnlineIP
}

// getUsersStats calls StatsService.GetUsersStats without traffic and
// without reset — again all proto3 zero values, so the request is empty.
func (x *XrayAPI) getUsersStats(ctx context.Context) ([]userStatEntry, error) {
	var resp rawBytes
	if err := x.grpcClient.Invoke(ctx, statsServiceGetUsersStats, rawBytes{}, &resp, grpc.ForceCodec(rawCodec{})); err != nil {
		return nil, err
	}
	return parseGetUsersStatsResponse(resp)
}

func parseGetUsersStatsResponse(b []byte) ([]userStatEntry, error) {
	var out []userStatEntry
	for len(b) > 0 {
		num, typ, n := protowire.ConsumeTag(b)
		if n < 0 {
			return nil, protowire.ParseError(n)
		}
		b = b[n:]
		if num == 1 && typ == protowire.BytesType {
			v, n := protowire.ConsumeBytes(b)
			if n < 0 {
				return nil, protowire.ParseError(n)
			}
			u, err := parseUserStat(v)
			if err != nil {
				return nil, err
			}
			out = append(out, u)
			b = b[n:]
			continue
		}
		n = protowire.ConsumeFieldValue(num, typ, b)
		if n < 0 {
			return nil, protowire.ParseError(n)
		}
		b = b[n:]
	}
	return out, nil
}

func parseUserStat(b []byte) (userStatEntry, error) {
	var u userStatEntry
	for len(b) > 0 {
		num, typ, n := protowire.ConsumeTag(b)
		if n < 0 {
			return u, protowire.ParseError(n)
		}
		b = b[n:]
		switch {
		case num == 1 && typ == protowire.BytesType:
			v, n := protowire.ConsumeBytes(b)
			if n < 0 {
				return u, protowire.ParseError(n)
			}
			u.Email = string(v)
			b = b[n:]
		case num == 2 && typ == protowire.BytesType:
			v, n := protowire.ConsumeBytes(b)
			if n < 0 {
				return u, protowire.ParseError(n)
			}
			ip, err := parseOnlineIPEntry(v)
			if err != nil {
				return u, err
			}
			u.IPs = append(u.IPs, ip)
			b = b[n:]
		default:
			n := protowire.ConsumeFieldValue(num, typ, b)
			if n < 0 {
				return u, protowire.ParseError(n)
			}
			b = b[n:]
		}
	}
	return u, nil
}

func parseOnlineIPEntry(b []byte) (OnlineIP, error) {
	var ip OnlineIP
	for len(b) > 0 {
		num, typ, n := protowire.ConsumeTag(b)
		if n < 0 {
			return ip, protowire.ParseError(n)
		}
		b = b[n:]
		switch {
		case num == 1 && typ == protowire.BytesType:
			v, n := protowire.ConsumeBytes(b)
			if n < 0 {
				return ip, protowire.ParseError(n)
			}
			ip.IP = string(v)
			b = b[n:]
		case num == 2 && typ == protowire.VarintType:
			v, n := protowire.ConsumeVarint(b)
			if n < 0 {
				return ip, protowire.ParseError(n)
			}
			ip.LastSeen = int64(v)
			b = b[n:]
		default:
			n := protowire.ConsumeFieldValue(num, typ, b)
			if n < 0 {
				return ip, protowire.ParseError(n)
			}
			b = b[n:]
		}
	}
	return ip, nil
}

// AddInbound is unavailable in this build: it handed the core a protobuf
// handler built by xray-core's config loader. Configuration changes are
// applied by restarting the core (see ErrHotApplyUnavailable).
func (x *XrayAPI) AddInbound(inbound []byte) error {
	return ErrHotApplyUnavailable
}

// DelInbound is unavailable in this build; see AddInbound.
func (x *XrayAPI) DelInbound(tag string) error {
	return ErrHotApplyUnavailable
}

// AddOutbound is unavailable in this build; see AddInbound.
func (x *XrayAPI) AddOutbound(outbound []byte) error {
	return ErrHotApplyUnavailable
}

// DelOutbound is unavailable in this build; see AddInbound.
func (x *XrayAPI) DelOutbound(tag string) error {
	return ErrHotApplyUnavailable
}

// ApplyRoutingConfig is unavailable in this build: it handed the core a
// protobuf router config built by xray-core's config loader. Routing
// changes are applied by restarting the core.
func (x *XrayAPI) ApplyRoutingConfig(routing []byte) error {
	return ErrHotApplyUnavailable
}

// BalancerInfo is the live state of one balancer inside the running core.
type BalancerInfo struct {
	Tag string `json:"tag"`
	// Override is the outbound tag an admin forced via the API; empty when
	// the strategy is in control.
	Override string `json:"override"`
	// Selected are the outbound tags the strategy currently prefers, best
	// first (xray's "principle target" list).
	Selected []string `json:"selected"`
}

// GetBalancerInfo is unavailable in this build; callers report the
// balancer as not running, exactly as they do for a stopped core.
func (x *XrayAPI) GetBalancerInfo(tag string) (*BalancerInfo, error) {
	return nil, ErrHotApplyUnavailable
}

// SetBalancerTarget is unavailable in this build; see GetBalancerInfo.
func (x *XrayAPI) SetBalancerTarget(tag, target string) error {
	return ErrHotApplyUnavailable
}

// RouteTestRequest describes a synthetic connection to ask the running core
// which outbound its router would pick for it.
type RouteTestRequest struct {
	InboundTag string // optional: simulate arrival on this inbound
	Domain     string // target domain (sniffed/SOCKS-style destination)
	IP         string // target IP, used when Domain is empty or alongside it
	Port       int
	Network    string // "tcp" (default) or "udp"
	Protocol   string // optional sniffed protocol: http, tls, bittorrent, ...
	Email      string // optional user attribution for user-based rules
}

// RouteTestResult is the routing decision the core reported.
type RouteTestResult struct {
	// Matched is false when no routing rule matched — traffic would use the
	// default (first) outbound and OutboundTag is empty.
	Matched     bool   `json:"matched"`
	OutboundTag string `json:"outboundTag"`
	// GroupTags lists the balancer chain the decision went through, when any.
	GroupTags []string `json:"groupTags,omitempty"`
}

// TestRoute is unavailable in this build: it needs the RoutingService,
// whose messages embed a fully built routing context. The panel's route
// test already requires a running core and reports this error verbatim.
func (x *XrayAPI) TestRoute(req RouteTestRequest) (*RouteTestResult, error) {
	return nil, ErrHotApplyUnavailable
}

// AddUser is unavailable in this build: it handed the core a typed
// per-protocol account built by xray-core. User changes are applied by
// restarting the core (see ErrHotApplyUnavailable).
func (x *XrayAPI) AddUser(Protocol string, inboundTag string, user map[string]any) error {
	return ErrHotApplyUnavailable
}

// RemoveUser is unavailable in this build; see AddUser.
func (x *XrayAPI) RemoveUser(inboundTag, email string) error {
	return ErrHotApplyUnavailable
}

// IsMissingHandlerErr reports whether err is xray's response to removing a
// handler (inbound/outbound) that does not exist — e.g. it was already
// removed through the runtime API while the panel's config snapshot was
// stale. Safe to treat as success for removal operations.
func IsMissingHandlerErr(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "not found") ||
		strings.Contains(msg, "not enough information")
}

// IsExistingTagErr reports whether err is xray's response to adding a handler
// whose tag is already taken by a running handler.
func IsExistingTagErr(err error) bool {
	if err == nil {
		return false
	}
	return strings.Contains(strings.ToLower(err.Error()), "existing tag")
}

// IsUserExistsErr reports whether err is xray's response to adding a user whose
// email is already registered on the inbound.
func IsUserExistsErr(err error) bool {
	if err == nil {
		return false
	}
	return strings.Contains(strings.ToLower(err.Error()), "already exists")
}

// ValidateOutboundConfig checks an outbound JSON object the way the core
// would at startup for the one refusal the panel must surface early:
// since xray-core v26.7.11, an unencrypted VLESS or Trojan outbound whose
// server address is a public IP or domain fails to build. Everything else
// the core's typed loader used to check (per-protocol field shapes) is no
// longer validated in-process; the core still enforces it when it starts
// on the node. The replication below mirrors the core's check exactly,
// including that it only inspects the flat "address" settings field —
// the vnext/servers forms were never covered by it.
//
// The error texts are the core's own, verbatim: the settings layer matches
// on them for its legacy-core version gate.
func ValidateOutboundConfig(outbound []byte) error {
	var parsed struct {
		Protocol       string `json:"protocol"`
		Settings       json.RawMessage
		StreamSettings *struct {
			Security string `json:"security"`
		} `json:"streamSettings"`
	}
	if err := json.Unmarshal(outbound, &parsed); err != nil {
		return err
	}

	// A secured stream (tls/reality/…) satisfies the rule outright.
	if parsed.StreamSettings != nil && parsed.StreamSettings.Security != "" && parsed.StreamSettings.Security != "none" {
		return nil
	}

	switch strings.ToLower(parsed.Protocol) {
	case "vless":
		var settings struct {
			Encryption string          `json:"encryption"`
			Address    json.RawMessage `json:"address"`
		}
		if err := json.Unmarshal(parsed.Settings, &settings); err != nil {
			return err
		}
		if settings.Encryption != "" && settings.Encryption != "none" {
			return nil
		}
		if address := flatAddress(settings.Address); address != "" && requiresTransportSecurity(address) {
			return errors.New("vless without TLS or other encryption is prohibited unless the server address is a private IP or domain")
		}
	case "trojan":
		var settings struct {
			Address json.RawMessage `json:"address"`
		}
		if err := json.Unmarshal(parsed.Settings, &settings); err != nil {
			return err
		}
		if address := flatAddress(settings.Address); address != "" && requiresTransportSecurity(address) {
			return errors.New("trojan without TLS is prohibited unless the server address is a private IP or domain")
		}
	}
	return nil
}

// flatAddress extracts the flat outbound "address" field, which the core's
// prohibition check reads. The field is a plain string in config JSON; any
// other shape means the outbound uses the vnext/servers form the core's
// check does not inspect.
func flatAddress(raw json.RawMessage) string {
	if len(raw) == 0 {
		return ""
	}
	var address string
	if err := json.Unmarshal(raw, &address); err != nil {
		return ""
	}
	return address
}

// privateIPPrefixes is xray-core's private-IP matcher list, verbatim
// (common/geodata consts): documentation ranges, CGNAT, multicast and
// reserved space count as private for the prohibition check.
var privateIPPrefixes = []netip.Prefix{
	netip.MustParsePrefix("0.0.0.0/8"),
	netip.MustParsePrefix("10.0.0.0/8"),
	netip.MustParsePrefix("100.64.0.0/10"),
	netip.MustParsePrefix("127.0.0.0/8"),
	netip.MustParsePrefix("169.254.0.0/16"),
	netip.MustParsePrefix("172.16.0.0/12"),
	netip.MustParsePrefix("192.0.0.0/24"),
	netip.MustParsePrefix("192.0.2.0/24"),
	netip.MustParsePrefix("192.88.99.0/24"),
	netip.MustParsePrefix("192.168.0.0/16"),
	netip.MustParsePrefix("198.18.0.0/15"),
	netip.MustParsePrefix("198.51.100.0/24"),
	netip.MustParsePrefix("203.0.113.0/24"),
	netip.MustParsePrefix("224.0.0.0/3"),
	netip.MustParsePrefix("::/127"),
	netip.MustParsePrefix("fc00::/7"),
	netip.MustParsePrefix("fe80::/10"),
	netip.MustParsePrefix("ff00::/8"),
}

// privateDomainSuffixes is xray-core's private-domain matcher list,
// verbatim: each entry matches the domain itself and any subdomain.
// Dotless domains are private too (the core matches them by regexp).
var privateDomainSuffixes = []string{
	"lan",
	"localdomain",
	"example",
	"invalid",
	"localhost",
	"test",
	"local",
	"home.arpa",
	"internal",
}

var dotlessDomainRegex = regexp.MustCompile(`^[a-z]([a-z0-9-]{0,61}[a-z0-9])?$`)

// requiresTransportSecurity mirrors the core's check of the same name: an
// IP must fall outside every private range; a domain must fall outside
// every private suffix (or be dotless).
func requiresTransportSecurity(address string) bool {
	if ip, err := netip.ParseAddr(address); err == nil {
		for _, prefix := range privateIPPrefixes {
			if prefix.Contains(ip) {
				return false
			}
		}
		return true
	}
	domain := strings.TrimSuffix(strings.ToLower(address), ".")
	if dotlessDomainRegex.MatchString(domain) {
		return false
	}
	for _, suffix := range privateDomainSuffixes {
		if domain == suffix || strings.HasSuffix(domain, "."+suffix) {
			return false
		}
	}
	return true
}

// GetTraffic queries traffic statistics from the Xray core and reports what
// accrued since the previous call; the counters themselves are never reset.
// The first call of a process only records baselines, since it may be reading
// counters that already hold traffic the panel cannot attribute. After that a
// name the panel has not seen — xray creates a counter on a user's first use —
// and a counter that moved backwards because the core restarted both count
// from zero, so no client's traffic is dropped for a whole polling interval.
func (x *XrayAPI) GetTraffic() ([]*Traffic, []*ClientTraffic, error) {
	if x.grpcClient == nil {
		return nil, nil, common.NewError("xray api is not initialized")
	}

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
	defer cancel()

	stats, err := x.queryStats(ctx)
	if err != nil {
		logger.Debug("Failed to query Xray stats:", err)
		return nil, nil, err
	}

	tagTrafficMap := make(map[string]*Traffic)
	emailTrafficMap := make(map[string]*ClientTraffic)

	if x.StatsLastValues == nil {
		x.StatsLastValues = make(map[string]int64)
	}
	baselinePass := len(x.StatsLastValues) == 0

	for _, stat := range stats {
		lastValue, ok := x.StatsLastValues[stat.Name]
		x.StatsLastValues[stat.Name] = stat.Value
		if baselinePass {
			continue
		}
		if !ok || stat.Value < lastValue {
			lastValue = 0
		}
		value := stat.Value - lastValue
		if matches := trafficRegex.FindStringSubmatch(stat.Name); len(matches) == 4 {
			processTraffic(matches, value, tagTrafficMap)
		} else if matches := clientTrafficRegex.FindStringSubmatch(stat.Name); len(matches) == 3 {
			processClientTraffic(matches, value, emailTrafficMap)
		}
	}

	// Drop delta baselines for stats that no longer exist (deleted inbounds or
	// clients), which otherwise linger until the next Xray restart. Only rebuild
	// when the map has drifted past 2x the live set, so the steady-state hot path
	// stays allocation-free.
	if n := len(stats); n > 0 && len(x.StatsLastValues) > 2*n {
		pruned := make(map[string]int64, n)
		for _, stat := range stats {
			pruned[stat.Name] = x.StatsLastValues[stat.Name]
		}
		x.StatsLastValues = pruned
	}

	return mapToSlice(tagTrafficMap), mapToSlice(emailTrafficMap), nil
}

// OnlineIP is one source address of a live connection, with the unix time (seconds)
// the core last dispatched a link from it.
type OnlineIP struct {
	IP       string `json:"ip"`
	LastSeen int64  `json:"lastSeen"`
}

// OnlineUser is a client email with at least one live connection and the source
// IPs of those connections, as tracked by Xray's statsUserOnline policy.
type OnlineUser struct {
	Email string     `json:"email"`
	IPs   []OnlineIP `json:"ips"`
}

// GetOnlineUsers returns every user with at least one live connection plus their
// source IPs, via StatsService.GetUsersStats (one RPC covers all users). Requires
// statsUserOnline enabled in the policy levels; older cores return Unimplemented.
func (x *XrayAPI) GetOnlineUsers() ([]OnlineUser, error) {
	if x.grpcClient == nil {
		return nil, common.NewError("xray api is not initialized")
	}

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
	defer cancel()

	stats, err := x.getUsersStats(ctx)
	if err != nil {
		return nil, err
	}

	users := make([]OnlineUser, 0, len(stats))
	for _, u := range stats {
		if u.Email == "" {
			continue
		}
		ips := make([]OnlineIP, 0, len(u.IPs))
		for _, entry := range u.IPs {
			if entry.IP == "" {
				continue
			}
			ips = append(ips, entry)
		}
		users = append(users, OnlineUser{Email: u.Email, IPs: ips})
	}
	return users, nil
}

// IsUnimplementedErr reports whether err is the running core saying it lacks an
// RPC (an older Xray binary without the online-stats API).
func IsUnimplementedErr(err error) bool {
	return status.Code(err) == codes.Unimplemented
}

// processTraffic aggregates a traffic stat into trafficMap using regex matches and value.
func processTraffic(matches []string, value int64, trafficMap map[string]*Traffic) {
	isInbound := matches[1] == "inbound"
	tag := matches[2]
	isDown := matches[3] == "downlink"

	if tag == "api" {
		return
	}

	traffic, ok := trafficMap[tag]
	if !ok {
		traffic = &Traffic{
			IsInbound:  isInbound,
			IsOutbound: !isInbound,
			Tag:        tag,
		}
		trafficMap[tag] = traffic
	}

	if isDown {
		traffic.Down = value
	} else {
		traffic.Up = value
	}
}

// processClientTraffic updates clientTrafficMap with upload/download values for a client email.
func processClientTraffic(matches []string, value int64, clientTrafficMap map[string]*ClientTraffic) {
	email := matches[1]
	isDown := matches[2] == "downlink"

	traffic, ok := clientTrafficMap[email]
	if !ok {
		traffic = &ClientTraffic{Email: email}
		clientTrafficMap[email] = traffic
	}

	if isDown {
		traffic.Down = value
	} else {
		traffic.Up = value
	}
}

// mapToSlice converts a map of pointers to a slice of pointers.
func mapToSlice[T any](m map[string]*T) []*T {
	result := make([]*T, 0, len(m))
	for _, v := range m {
		result = append(result, v)
	}
	return result
}