telemt-api/internal/aggregate/merge.go

package aggregate

import (
	"sort"
	"strings"
	"time"
)

// BuildTraffic builds traffic matrix from successful fetches only.
func BuildTraffic(results []ServerFetchResult) []TrafficRow {
	perUser := map[string]map[string]TrafficServerStats{}

	for _, fr := range results {
		if !fr.OK {
			continue
		}
		for _, u := range fr.Users {
			if perUser[u.Username] == nil {
				perUser[u.Username] = map[string]TrafficServerStats{}
			}
			perUser[u.Username][fr.Alias] = TrafficServerStats{
				TotalMegabytes:     octetsToMegabytes(u.TotalOctets),
				CurrentConnections: u.CurrentConnections,
				Revision:           fr.Revision,
			}
		}
	}

	names := make([]string, 0, len(perUser))
	for n := range perUser {
		names = append(names, n)
	}
	sort.Strings(names)

	rows := make([]TrafficRow, 0, len(names))
	for _, name := range names {
		rows = append(rows, TrafficRow{
			Username: name,
			Servers:  perUser[name],
		})
	}
	return rows
}

// BuildUniqueIPs builds IP × server visibility per user.
func BuildUniqueIPs(results []ServerFetchResult) []UniqueIPsRow {
	type ipKey struct {
		user string
		ip   string
	}
	active := map[ipKey]map[string]struct{}{}
	recent := map[ipKey]map[string]struct{}{}

	for _, fr := range results {
		if !fr.OK {
			continue
		}
		for _, u := range fr.Users {
			for _, ip := range u.ActiveUniqueIPsList {
				ip = strings.TrimSpace(ip)
				if ip == "" {
					continue
				}
				k := ipKey{user: u.Username, ip: ip}
				if active[k] == nil {
					active[k] = map[string]struct{}{}
				}
				active[k][fr.Alias] = struct{}{}
			}
			for _, ip := range u.RecentUniqueIPsList {
				ip = strings.TrimSpace(ip)
				if ip == "" {
					continue
				}
				k := ipKey{user: u.Username, ip: ip}
				if recent[k] == nil {
					recent[k] = map[string]struct{}{}
				}
				recent[k][fr.Alias] = struct{}{}
			}
		}
	}

	users := map[string][]ipKey{}
	for k := range active {
		users[k.user] = append(users[k.user], k)
	}
	for k := range recent {
		found := false
		for _, x := range users[k.user] {
			if x == k {
				found = true
				break
			}
		}
		if !found {
			users[k.user] = append(users[k.user], k)
		}
	}

	userNames := make([]string, 0, len(users))
	for u := range users {
		userNames = append(userNames, u)
	}
	sort.Strings(userNames)

	out := make([]UniqueIPsRow, 0, len(userNames))
	for _, uname := range userNames {
		keys := users[uname]
		sort.Slice(keys, func(i, j int) bool { return keys[i].ip < keys[j].ip })
		ips := make([]IPAssignments, 0, len(keys))
		for _, k := range keys {
			a := sortedKeys(active[k])
			r := sortedKeys(recent[k])
			var primary *string
			if len(a) == 1 {
				s := a[0]
				primary = &s
			}
			ips = append(ips, IPAssignments{
				IP:              k.ip,
				ActiveOnServers: a,
				RecentOnServers: r,
				PrimaryServer:   primary,
			})
		}
		out = append(out, UniqueIPsRow{Username: uname, IPs: ips})
	}
	return out
}

func sortedKeys(m map[string]struct{}) []string {
	if len(m) == 0 {
		return nil
	}
	s := make([]string, 0, len(m))
	for k := range m {
		s = append(s, k)
	}
	sort.Strings(s)
	return s
}

type userOnServer struct {
	alias string
	u     UserInfo
}

// BuildUsers merged rows with totals and optional links from first successful server per user.
func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets uint64) []UsersRow {
	type acc struct {
		byServer map[string]TrafficServerStats
		links    *UserLinks
		total    uint64
		act      uint64
		rec      uint64
	}
	m := map[string]*acc{}
	perUserServers := map[string][]userOnServer{}

	for _, fr := range results {
		if !fr.OK {
			continue
		}
		for _, u := range fr.Users {
			a := m[u.Username]
			if a == nil {
				a = &acc{byServer: map[string]TrafficServerStats{}}
				m[u.Username] = a
			}
			a.byServer[fr.Alias] = TrafficServerStats{
				TotalMegabytes:     octetsToMegabytes(u.TotalOctets),
				CurrentConnections: u.CurrentConnections,
				Revision:           fr.Revision,
			}
			a.total += u.TotalOctets
			if u.ActiveUniqueIPs > a.act {
				a.act = u.ActiveUniqueIPs
			}
			if u.RecentUniqueIPs > a.rec {
				a.rec = u.RecentUniqueIPs
			}
			if includeLinks && u.Links != nil && a.links == nil {
				a.links = u.Links
			}
			perUserServers[u.Username] = append(perUserServers[u.Username], userOnServer{alias: fr.Alias, u: u})
		}
	}

	names := make([]string, 0, len(m))
	for n := range m {
		if m[n].total >= minTotalOctets {
			names = append(names, n)
		}
	}
	sort.Strings(names)

	rows := make([]UsersRow, 0, len(names))
	for _, name := range names {
		a := m[name]
		sort.Slice(perUserServers[name], func(i, j int) bool {
			return perUserServers[name][i].alias < perUserServers[name][j].alias
		})
		ad, mt, ex, dq, mu := mergeUserLimitFields(perUserServers[name])
		rows = append(rows, UsersRow{
			Username:          name,
			TotalMegabytes:    octetsToMegabytes(a.total),
			ByServer:          a.byServer,
			Links:             a.links,
			ActiveUniqueIPs:   a.act,
			RecentUniqueIPs:   a.rec,
			UserAdTag:         ad,
			MaxTCPConns:       mt,
			ExpirationRFC3339: ex,
			DataQuotaBytes:    dq,
			MaxUniqueIPs:      mu,
		})
	}
	return rows
}

// BuildSingleUser returns one merged user row or nil if the user is absent on all successful upstreams.
func BuildSingleUser(results []ServerFetchResult, username string, includeLinks bool) *UsersRow {
	rows := BuildUsers(results, includeLinks, 0)
	for i := range rows {
		if rows[i].Username == username {
			return &rows[i]
		}
	}
	return nil
}

func mergeUserLimitFields(rows []userOnServer) (userAdTag *string, maxTCP *uint64, exp *string, quota *uint64, maxUip *uint64) {
	for _, r := range rows {
		if r.u.UserAdTag != nil && userAdTag == nil {
			v := *r.u.UserAdTag
			userAdTag = &v
		}
	}
	var earliest time.Time
	var earliestStr string
	haveEarliest := false
	for _, r := range rows {
		if r.u.ExpirationRFC3339 == nil {
			continue
		}
		s := *r.u.ExpirationRFC3339
		t, err := time.Parse(time.RFC3339Nano, s)
		if err != nil {
			t, err = time.Parse(time.RFC3339, s)
		}
		if err != nil {
			continue
		}
		if !haveEarliest || t.Before(earliest) {
			earliest = t
			earliestStr = s
			haveEarliest = true
		}
	}
	if haveEarliest {
		exp = &earliestStr
	}
	for _, r := range rows {
		if r.u.MaxTCPConns != nil {
			v := *r.u.MaxTCPConns
			if maxTCP == nil || v < *maxTCP {
				maxTCP = new(uint64)
				*maxTCP = v
			}
		}
	}
	for _, r := range rows {
		if r.u.DataQuotaBytes != nil {
			v := *r.u.DataQuotaBytes
			if quota == nil || v < *quota {
				quota = new(uint64)
				*quota = v
			}
		}
	}
	for _, r := range rows {
		if r.u.MaxUniqueIPs != nil {
			v := *r.u.MaxUniqueIPs
			if maxUip == nil || v < *maxUip {
				maxUip = new(uint64)
				*maxUip = v
			}
		}
	}
	return userAdTag, maxTCP, exp, quota, maxUip
}

// BuildSummary computes fleet totals and top users.
func BuildSummary(results []ServerFetchResult, topN int) SummaryData {
	if topN <= 0 {
		topN = 10
	}
	if topN > 1000 {
		topN = 1000
	}

	sumByUser := map[string]uint64{}
	maxUniqueIPByUser := map[string]uint64{}
	var fleetOctets, fleetConn uint64
	ok, fail := 0, 0
	serverRows := make([]ServerFetchResult, 0, len(results))

	for _, fr := range results {
		sr := ServerFetchResult{
			Alias:      fr.Alias,
			OK:         fr.OK,
			HTTPStatus: fr.HTTPStatus,
			LatencyMs:  fr.LatencyMs,
			Error:      fr.Error,
			Revision:   fr.Revision,
		}
		serverRows = append(serverRows, sr)
		if !fr.OK {
			fail++
			continue
		}
		ok++
		for _, u := range fr.Users {
			fleetOctets += u.TotalOctets
			fleetConn += u.CurrentConnections
			sumByUser[u.Username] += u.TotalOctets
			if u.ActiveUniqueIPs > maxUniqueIPByUser[u.Username] {
				maxUniqueIPByUser[u.Username] = u.ActiveUniqueIPs
			}
		}
	}

	type pair struct {
		name string
		n    uint64
	}
	pairs := make([]pair, 0, len(sumByUser))
	for n, v := range sumByUser {
		pairs = append(pairs, pair{name: n, n: v})
	}
	sort.Slice(pairs, func(i, j int) bool {
		if pairs[i].n != pairs[j].n {
			return pairs[i].n > pairs[j].n
		}
		return pairs[i].name < pairs[j].name
	})
	if len(pairs) > topN {
		pairs = pairs[:topN]
	}
	top := make([]TopUser, 0, len(pairs))
	for _, p := range pairs {
		top = append(top, TopUser{Username: p.name, TotalMegabytes: octetsToMegabytes(p.n)})
	}

	ipPairs := make([]pair, 0, len(maxUniqueIPByUser))
	for n, v := range maxUniqueIPByUser {
		ipPairs = append(ipPairs, pair{name: n, n: v})
	}
	sort.Slice(ipPairs, func(i, j int) bool {
		if ipPairs[i].n != ipPairs[j].n {
			return ipPairs[i].n > ipPairs[j].n
		}
		return ipPairs[i].name < ipPairs[j].name
	})
	if len(ipPairs) > topN {
		ipPairs = ipPairs[:topN]
	}
	topIP := make([]TopUserByUniqueIPs, 0, len(ipPairs))
	for _, p := range ipPairs {
		topIP = append(topIP, TopUserByUniqueIPs{Username: p.name, UniqueIPs: p.n})
	}

	return SummaryData{
		Servers:               serverRows,
		ServersTotal:          len(results),
		ServersOK:             ok,
		ServersFailed:         fail,
		FleetTotalMegabytes:   octetsToMegabytes(fleetOctets),
		FleetTotalConnections: fleetConn,
		TopUsers:              top,
		TopUsersByUniqueIPs:   topIP,
	}
}