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