Add aggregate configuration support and update documentation

- Introduced AggregateConfig to manage aggregation settings in the gateway configuration.
- Added validation for reserved alias 'agg' and included tests for aggregate alias handling.
- Updated config.example.yaml to demonstrate aggregate configuration options.
- Enhanced README.md to include information about the new aggregation endpoint and its usage.
- Modified gateway.go to integrate the new aggregate handler for processing aggregation requests.

README.md

@@ -1,6 +1,6 @@
 # telemt-api
 
-HTTP‑шлюз на Go для [Telemt Control API](docs/API.md): один порт, **белый список IP (CIDR)**, маршруты вида `/api/{alias}/…` → `{base_url}/v1/…`, метрики Prometheus на `/metrics`.
+HTTP‑шлюз на Go для [Telemt Control API](docs/API.md): один порт, **белый список IP (CIDR)**, маршруты вида `/api/{alias}/…` → `{base_url}/v1/…`, агрегация нескольких инстансов — [`/api/agg/…`](docs/AGGREGATE.md), метрики Prometheus на `/metrics`.
 
 ## Быстрый старт (Linux)
 
@@ -55,6 +55,7 @@ docker compose logs -f gateway
 |----------|------------|
 | **[docs/GATEWAY_RUN.md](docs/GATEWAY_RUN.md)** | Полная инструкция: конфиг, pull/registry, Docker CLI, Compose, CI/CD, неполадки |
 | **[docs/API.md](docs/API.md)** | Контракт Telemt Control API (`/v1/…`) |
+| **[docs/AGGREGATE.md](docs/AGGREGATE.md)** | Агрегирующие эндпоинты шлюза (`/api/agg/…`) |
 
 ## Сборка и тесты без Docker
 

config.example.yaml

@@ -23,6 +23,12 @@ whitelist_cidrs:
 #   - "10.0.0.0/8"
 trusted_proxies: []
 
+# Опционально: по умолчанию /api/agg/* опрашивает все servers; можно ограничить список:
+# aggregate:
+#   include_aliases:
+#     - gt1
+#     - gt2
+
 servers:
   - alias: main_srv
     base_url: http://127.0.0.1:9091

docs/AGGREGATE.md

@@ -0,0 +1,52 @@
+# Агрегирующие эндпоинты шлюза (`/api/agg/`)
+
+Шлюз **telemt-api** опрашивает несколько upstream [Telemt Control API](API.md) (`GET /v1/stats/users` на каждом сервере из конфигурации) и отдаёт сводные JSON-ответы в формате `{"ok": true, "data": ...}`.
+
+Доступ к **одному** инстансу по-прежнему через прокси: `GET /api/{alias}/…` (например `/api/gt1/v1/stats/users`).
+
+## Маршруты
+
+| Метод | Путь | Описание |
+| --- | --- | --- |
+| GET | `/api/agg/summary` | Сводка по флоту, список опросов upstream, топ пользователей по суммарному `total_octets`. |
+| GET | `/api/agg/traffic` | Трафик по каждому пользователю в разрезе серверов (алиасов). |
+| GET | `/api/agg/unique-ips` | Уникальные IP по пользователю: на каких серверах IP есть в active/recent списках снимка. |
+| GET | `/api/agg/users` | Объединённый список пользователей с `by_server` и суммарным `total_octets`. |
+
+Все методы — **GET**; действует тот же whitelist, что и для остального API шлюза.
+
+## Query-параметры
+
+| Параметр | Где | Значение |
+| --- | --- | --- |
+| `aliases` | все | Список алиасов через запятую (например `gt1,gt2`). Если не задан — см. `aggregate.include_aliases` в YAML или все серверы из `servers`. |
+| `top_n` | `summary` | Размер топа пользователей (по умолчанию `10`, максимум `1000`). |
+| `include_links` | `users` | `true` — добавить сгенерированные `tg://proxy` ссылки (берётся первая успешная запись по пользователю). |
+| `min_total_octets` | `users` | Отфильтровать пользователей с суммарным трафиком ниже порога. |
+
+## Конфигурация (опционально)
+
+```yaml
+aggregate:
+  include_aliases:
+    - gt1
+    - gt2
+```
+
+Если блок отсутствует или `include_aliases` пуст, по умолчанию участвуют **все** записи `servers`.
+
+Имя алиаса **`agg`** в `servers` запрещено (зарезервировано под префикс `/api/agg/`).
+
+## Ограничения
+
+- **Один и тот же `username` на разных серверах** может соответствовать разным учётным записям; агрегатор сопоставляет строки по имени — учитывайте при интерпретации сумм.
+- У Telemt в `UserInfo` **нет** поля «IP последний раз подключался к серверу X». В `unique-ips` поле `primary_server` заполняется **только** если ровно один сервер видит IP в `active_unique_ips_list` на момент запроса; иначе `primary_server` отсутствует или несколько серверов в списках — это снимок, не история.
+
+## Примеры
+
+```bash
+curl -sS "http://127.0.0.1:8080/api/agg/summary"
+curl -sS "http://127.0.0.1:8080/api/agg/traffic?aliases=gt1,gt2"
+curl -sS "http://127.0.0.1:8080/api/agg/unique-ips"
+curl -sS "http://127.0.0.1:8080/api/agg/users?include_links=false&min_total_octets=1000000"
+```

internal/aggregate/fetch.go

@@ -0,0 +1,148 @@
+package aggregate
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/telemt/telemt-api/internal/config"
+)
+
+const statsUsersPath = "stats/users"
+
+// FetchStatsUsers calls GET {base}{path_prefix}/stats/users for each alias.
+func FetchStatsUsers(ctx context.Context, client *http.Client, parsed *config.Parsed, aliases []string) []ServerFetchResult {
+	out := make([]ServerFetchResult, 0, len(aliases))
+	for _, alias := range aliases {
+		srv := parsed.ByAlias[alias]
+		if srv == nil {
+			out = append(out, ServerFetchResult{
+				Alias: alias,
+				OK:    false,
+				Error: "unknown alias",
+			})
+			continue
+		}
+		u, err := url.Parse(srv.BaseURL)
+		if err != nil {
+			out = append(out, ServerFetchResult{
+				Alias: alias,
+				OK:    false,
+				Error: err.Error(),
+			})
+			continue
+		}
+		target := joinPathPrefix(u, srv.PathPrefix, statsUsersPath)
+		start := time.Now()
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), nil)
+		if err != nil {
+			out = append(out, ServerFetchResult{
+				Alias:     alias,
+				OK:        false,
+				LatencyMs: time.Since(start).Milliseconds(),
+				Error:     err.Error(),
+			})
+			continue
+		}
+		if auth := parsed.AuthByAlias[alias]; auth != "" {
+			req.Header.Set("Authorization", auth)
+		}
+		resp, err := client.Do(req)
+		latency := time.Since(start).Milliseconds()
+		if err != nil {
+			out = append(out, ServerFetchResult{
+				Alias:     alias,
+				OK:        false,
+				LatencyMs: latency,
+				Error:     err.Error(),
+			})
+			continue
+		}
+		body, readErr := io.ReadAll(io.LimitReader(resp.Body, 64<<20))
+		_ = resp.Body.Close()
+		if readErr != nil {
+			out = append(out, ServerFetchResult{
+				Alias:      alias,
+				OK:         false,
+				HTTPStatus: resp.StatusCode,
+				LatencyMs:  latency,
+				Error:      readErr.Error(),
+			})
+			continue
+		}
+		if resp.StatusCode != http.StatusOK {
+			out = append(out, ServerFetchResult{
+				Alias:      alias,
+				OK:         false,
+				HTTPStatus: resp.StatusCode,
+				LatencyMs:  latency,
+				Error:      fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(body), 200)),
+			})
+			continue
+		}
+		var env statsUsersEnvelope
+		if err := json.Unmarshal(body, &env); err != nil {
+			out = append(out, ServerFetchResult{
+				Alias:      alias,
+				OK:         false,
+				HTTPStatus: resp.StatusCode,
+				LatencyMs:  latency,
+				Error:      "invalid json: " + err.Error(),
+			})
+			continue
+		}
+		if !env.OK {
+			out = append(out, ServerFetchResult{
+				Alias:      alias,
+				OK:         false,
+				HTTPStatus: resp.StatusCode,
+				LatencyMs:  latency,
+				Error:      "upstream ok=false",
+			})
+			continue
+		}
+		out = append(out, ServerFetchResult{
+			Alias:      alias,
+			OK:         true,
+			HTTPStatus: resp.StatusCode,
+			LatencyMs:  latency,
+			Revision:   env.Revision,
+			Users:      env.Data,
+		})
+	}
+	return out
+}
+
+func joinPathPrefix(base *url.URL, pathPrefix, rest string) *url.URL {
+	rel := strings.Trim(pathPrefix, "/")
+	if rest != "" {
+		if rel != "" {
+			rel = rel + "/" + rest
+		} else {
+			rel = rest
+		}
+	}
+	var parts []string
+	for _, seg := range strings.Split(rel, "/") {
+		if seg != "" {
+			parts = append(parts, seg)
+		}
+	}
+	if len(parts) == 0 {
+		out := *base
+		return &out
+	}
+	return base.JoinPath(parts...)
+}
+
+func truncate(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return s[:n] + "…"
+}

internal/aggregate/handlers.go

@@ -0,0 +1,189 @@
+package aggregate
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/telemt/telemt-api/internal/config"
+)
+
+const pathPrefix = "/api/agg"
+
+// Handler serves GET /api/agg/* aggregate endpoints.
+type Handler struct {
+	Parsed *config.Parsed
+	Client *http.Client
+}
+
+// NewHandler builds an aggregate handler; client must use a non-nil Transport (e.g. gateway shared transport).
+func NewHandler(p *config.Parsed, client *http.Client) *Handler {
+	return &Handler{Parsed: p, Client: client}
+}
+
+func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		w.Header().Set("Content-Type", "application/json; charset=utf-8")
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		_ = json.NewEncoder(w).Encode(errEnvelope("method_not_allowed", "only GET is allowed"))
+		return
+	}
+	sub := strings.TrimPrefix(r.URL.Path, pathPrefix)
+	sub = strings.TrimPrefix(sub, "/")
+	switch sub {
+	case "summary":
+		h.handleSummary(w, r)
+	case "traffic":
+		h.handleTraffic(w, r)
+	case "unique-ips":
+		h.handleUniqueIPs(w, r)
+	case "users":
+		h.handleUsers(w, r)
+	default:
+		w.Header().Set("Content-Type", "application/json; charset=utf-8")
+		w.WriteHeader(http.StatusNotFound)
+		_ = json.NewEncoder(w).Encode(errEnvelope("not_found", "unknown aggregate path"))
+	}
+}
+
+func errEnvelope(code, msg string) map[string]any {
+	return map[string]any{
+		"ok": false,
+		"error": map[string]string{
+			"code":    code,
+			"message": msg,
+		},
+	}
+}
+
+func (h *Handler) resolveAliases(r *http.Request) ([]string, error) {
+	q := r.URL.Query().Get("aliases")
+	if strings.TrimSpace(q) != "" {
+		var out []string
+		for _, p := range strings.Split(q, ",") {
+			a := strings.TrimSpace(p)
+			if a == "" {
+				continue
+			}
+			if h.Parsed.ByAlias[a] == nil {
+				return nil, &resolveError{msg: "unknown alias: " + a}
+			}
+			out = append(out, a)
+		}
+		if len(out) == 0 {
+			return nil, &resolveError{msg: "aliases query produced empty list"}
+		}
+		return out, nil
+	}
+	cfg := h.Parsed.Config.Aggregate
+	if cfg != nil && len(cfg.IncludeAliases) > 0 {
+		for _, a := range cfg.IncludeAliases {
+			a = strings.TrimSpace(a)
+			if a == "" {
+				continue
+			}
+			if h.Parsed.ByAlias[a] == nil {
+				return nil, &resolveError{msg: "aggregate.include_aliases: unknown alias: " + a}
+			}
+		}
+		out := make([]string, 0, len(cfg.IncludeAliases))
+		for _, a := range cfg.IncludeAliases {
+			a = strings.TrimSpace(a)
+			if a == "" {
+				continue
+			}
+			out = append(out, a)
+		}
+		return out, nil
+	}
+	out := make([]string, 0, len(h.Parsed.Config.Servers))
+	for i := range h.Parsed.Config.Servers {
+		out = append(out, h.Parsed.Config.Servers[i].Alias)
+	}
+	return out, nil
+}
+
+type resolveError struct {
+	msg string
+}
+
+func (e *resolveError) Error() string { return e.msg }
+
+func (h *Handler) handleSummary(w http.ResponseWriter, r *http.Request) {
+	aliases, err := h.resolveAliases(r)
+	if err != nil {
+		writeBadRequest(w, err)
+		return
+	}
+	topN := 10
+	if v := r.URL.Query().Get("top_n"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			topN = n
+		}
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
+	defer cancel()
+	results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
+	data := BuildSummary(results, topN)
+	writeOK(w, data)
+}
+
+func (h *Handler) handleTraffic(w http.ResponseWriter, r *http.Request) {
+	aliases, err := h.resolveAliases(r)
+	if err != nil {
+		writeBadRequest(w, err)
+		return
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
+	defer cancel()
+	results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
+	data := BuildTraffic(results)
+	writeOK(w, data)
+}
+
+func (h *Handler) handleUniqueIPs(w http.ResponseWriter, r *http.Request) {
+	aliases, err := h.resolveAliases(r)
+	if err != nil {
+		writeBadRequest(w, err)
+		return
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
+	defer cancel()
+	results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
+	data := BuildUniqueIPs(results)
+	writeOK(w, data)
+}
+
+func (h *Handler) handleUsers(w http.ResponseWriter, r *http.Request) {
+	aliases, err := h.resolveAliases(r)
+	if err != nil {
+		writeBadRequest(w, err)
+		return
+	}
+	includeLinks := strings.EqualFold(r.URL.Query().Get("include_links"), "true")
+	minOct := uint64(0)
+	if v := r.URL.Query().Get("min_total_octets"); v != "" {
+		if n, err := strconv.ParseUint(v, 10, 64); err == nil {
+			minOct = n
+		}
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
+	defer cancel()
+	results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
+	data := BuildUsers(results, includeLinks, minOct)
+	writeOK(w, data)
+}
+
+func writeBadRequest(w http.ResponseWriter, err error) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.WriteHeader(http.StatusBadRequest)
+	_ = json.NewEncoder(w).Encode(errEnvelope("bad_request", err.Error()))
+}
+
+func writeOK(w http.ResponseWriter, data any) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	_ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "data": data})
+}

internal/aggregate/handlers_test.go

@@ -0,0 +1,75 @@
+package aggregate
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/telemt/telemt-api/internal/config"
+)
+
+func TestHandlerResolveAndFetch(t *testing.T) {
+	up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/stats/users" {
+			http.NotFound(w, r)
+			return
+		}
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"ok":       true,
+			"data":     []map[string]any{{"username": "u1", "total_octets": 42, "current_connections": 0}},
+			"revision": "abc",
+		})
+	}))
+	defer up.Close()
+
+	cfg := &config.Config{
+		Servers: []config.Server{
+			{Alias: "test", BaseURL: up.URL, PathPrefix: "/v1"},
+		},
+	}
+	if err := cfg.Validate(); err != nil {
+		t.Fatal(err)
+	}
+	parsed, err := cfg.Parse()
+	if err != nil {
+		t.Fatal(err)
+	}
+	h := NewHandler(parsed, up.Client())
+	req := httptest.NewRequest(http.MethodGet, "/api/agg/summary?aliases=test", nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d body %s", rec.Code, rec.Body.String())
+	}
+	var env struct {
+		OK   bool           `json:"ok"`
+		Data SummaryData    `json:"data"`
+	}
+	if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil {
+		t.Fatal(err)
+	}
+	if !env.OK || env.Data.FleetTotalOctets != 42 {
+		t.Fatalf("data: %+v", env.Data)
+	}
+}
+
+func TestHandlerMethodNotAllowed(t *testing.T) {
+	cfg := &config.Config{
+		Servers: []config.Server{{Alias: "x", BaseURL: "http://127.0.0.1:1", PathPrefix: "/v1"}},
+	}
+	if err := cfg.Validate(); err != nil {
+		t.Fatal(err)
+	}
+	parsed, err := cfg.Parse()
+	if err != nil {
+		t.Fatal(err)
+	}
+	h := NewHandler(parsed, http.DefaultClient)
+	req := httptest.NewRequest(http.MethodPost, "/api/agg/summary", nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+	if rec.Code != http.StatusMethodNotAllowed {
+		t.Fatalf("got %d", rec.Code)
+	}
+}

internal/aggregate/merge.go

@@ -0,0 +1,272 @@
+package aggregate
+
+import (
+	"sort"
+	"strings"
+)
+
+// 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{
+				TotalOctets:        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
+}
+
+// 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{}
+
+	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{
+				TotalOctets:        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
+			}
+		}
+	}
+
+	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]
+		rows = append(rows, UsersRow{
+			Username:        name,
+			TotalOctets:     a.total,
+			ByServer:        a.byServer,
+			Links:           a.links,
+			ActiveUniqueIPs: a.act,
+			RecentUniqueIPs: a.rec,
+		})
+	}
+	return rows
+}
+
+// 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{}
+	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
+		}
+	}
+
+	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, TotalOctets: p.n})
+	}
+
+	return SummaryData{
+		Servers:               serverRows,
+		ServersTotal:          len(results),
+		ServersOK:             ok,
+		ServersFailed:         fail,
+		FleetTotalOctets:      fleetOctets,
+		FleetTotalConnections: fleetConn,
+		TopUsers:              top,
+	}
+}

internal/aggregate/merge_test.go

@@ -0,0 +1,122 @@
+package aggregate
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestBuildTraffic(t *testing.T) {
+	results := []ServerFetchResult{
+		{
+			Alias: "a", OK: true, Revision: "r1",
+			Users: []UserInfo{
+				{Username: "u1", TotalOctets: 100, CurrentConnections: 2},
+				{Username: "u2", TotalOctets: 50, CurrentConnections: 0},
+			},
+		},
+		{
+			Alias: "b", OK: true, Revision: "r2",
+			Users: []UserInfo{
+				{Username: "u1", TotalOctets: 30, CurrentConnections: 1},
+			},
+		},
+	}
+	rows := BuildTraffic(results)
+	if len(rows) != 2 {
+		t.Fatalf("len rows: %d", len(rows))
+	}
+	var u1 *TrafficRow
+	for i := range rows {
+		if rows[i].Username == "u1" {
+			u1 = &rows[i]
+			break
+		}
+	}
+	if u1 == nil {
+		t.Fatal("missing u1")
+	}
+	if u1.Servers["a"].TotalOctets != 100 || u1.Servers["b"].TotalOctets != 30 {
+		t.Fatalf("u1 servers: %+v", u1.Servers)
+	}
+}
+
+func TestBuildUniqueIPs(t *testing.T) {
+	results := []ServerFetchResult{
+		{
+			Alias: "s1", OK: true,
+			Users: []UserInfo{
+				{
+					Username:            "alice",
+					ActiveUniqueIPsList: []string{"10.0.0.1"},
+					RecentUniqueIPsList: []string{"10.0.0.2"},
+				},
+			},
+		},
+		{
+			Alias: "s2", OK: true,
+			Users: []UserInfo{
+				{
+					Username:            "alice",
+					ActiveUniqueIPsList: []string{"10.0.0.1", "10.0.0.3"},
+				},
+			},
+		},
+	}
+	rows := BuildUniqueIPs(results)
+	if len(rows) != 1 || rows[0].Username != "alice" {
+		t.Fatalf("rows: %+v", rows)
+	}
+	var ip1 *IPAssignments
+	for i := range rows[0].IPs {
+		if rows[0].IPs[i].IP == "10.0.0.1" {
+			ip1 = &rows[0].IPs[i]
+			break
+		}
+	}
+	if ip1 == nil {
+		t.Fatal("missing 10.0.0.1")
+	}
+	if !reflect.DeepEqual(ip1.ActiveOnServers, []string{"s1", "s2"}) {
+		t.Fatalf("active: %v", ip1.ActiveOnServers)
+	}
+	if ip1.PrimaryServer != nil {
+		t.Fatalf("expected no primary, got %v", *ip1.PrimaryServer)
+	}
+}
+
+func TestBuildUniqueIPsPrimary(t *testing.T) {
+	results := []ServerFetchResult{
+		{Alias: "only", OK: true, Users: []UserInfo{
+			{Username: "bob", ActiveUniqueIPsList: []string{"192.168.1.1"}},
+		}},
+	}
+	rows := BuildUniqueIPs(results)
+	if len(rows[0].IPs) != 1 || rows[0].IPs[0].PrimaryServer == nil || *rows[0].IPs[0].PrimaryServer != "only" {
+		t.Fatalf("primary: %+v", rows[0].IPs)
+	}
+}
+
+func TestBuildSummary(t *testing.T) {
+	results := []ServerFetchResult{
+		{Alias: "a", OK: true, Users: []UserInfo{{Username: "x", TotalOctets: 100, CurrentConnections: 2}}},
+		{Alias: "b", OK: false, Error: "down"},
+	}
+	s := BuildSummary(results, 5)
+	if s.ServersOK != 1 || s.ServersFailed != 1 || s.FleetTotalOctets != 100 || s.FleetTotalConnections != 2 {
+		t.Fatalf("summary: %+v", s)
+	}
+	if len(s.TopUsers) != 1 || s.TopUsers[0].Username != "x" {
+		t.Fatalf("top: %+v", s.TopUsers)
+	}
+}
+
+func TestBuildUsersMinOctets(t *testing.T) {
+	results := []ServerFetchResult{
+		{Alias: "a", OK: true, Users: []UserInfo{{Username: "low", TotalOctets: 5}}},
+		{Alias: "b", OK: true, Users: []UserInfo{{Username: "high", TotalOctets: 100}}},
+	}
+	rows := BuildUsers(results, false, 50)
+	if len(rows) != 1 || rows[0].Username != "high" {
+		t.Fatalf("rows: %+v", rows)
+	}
+}

internal/aggregate/types.go

@@ -0,0 +1,96 @@
+package aggregate
+
+// UserInfo mirrors Telemt Control API /v1/stats/users items (subset used by aggregator).
+type UserInfo struct {
+	Username           string     `json:"username"`
+	UserAdTag          *string    `json:"user_ad_tag"`
+	MaxTCPConns        *uint64    `json:"max_tcp_conns"`
+	ExpirationRFC3339  *string    `json:"expiration_rfc3339"`
+	DataQuotaBytes     *uint64    `json:"data_quota_bytes"`
+	MaxUniqueIPs       *uint64    `json:"max_unique_ips"`
+	CurrentConnections uint64     `json:"current_connections"`
+	ActiveUniqueIPs    uint64     `json:"active_unique_ips"`
+	ActiveUniqueIPsList []string  `json:"active_unique_ips_list"`
+	RecentUniqueIPs    uint64     `json:"recent_unique_ips"`
+	RecentUniqueIPsList []string  `json:"recent_unique_ips_list"`
+	TotalOctets        uint64     `json:"total_octets"`
+	Links              *UserLinks `json:"links,omitempty"`
+}
+
+// UserLinks optional tg://proxy link groups.
+type UserLinks struct {
+	Classic []string `json:"classic,omitempty"`
+	Secure  []string `json:"secure,omitempty"`
+	TLS     []string `json:"tls,omitempty"`
+}
+
+type statsUsersEnvelope struct {
+	OK       bool       `json:"ok"`
+	Data     []UserInfo `json:"data"`
+	Revision string     `json:"revision"`
+}
+
+// ServerFetchResult is one upstream GET /v1/stats/users outcome.
+type ServerFetchResult struct {
+	Alias      string     `json:"alias"`
+	OK         bool       `json:"ok"`
+	HTTPStatus int        `json:"http_status,omitempty"`
+	LatencyMs  int64      `json:"latency_ms,omitempty"`
+	Error      string     `json:"error,omitempty"`
+	Revision   string     `json:"revision,omitempty"`
+	Users      []UserInfo `json:"-"`
+}
+
+// TrafficRow is per-user traffic broken down by server alias.
+type TrafficRow struct {
+	Username string                        `json:"username"`
+	Servers  map[string]TrafficServerStats `json:"servers"`
+}
+
+// TrafficServerStats per-server counters for one user.
+type TrafficServerStats struct {
+	TotalOctets        uint64 `json:"total_octets"`
+	CurrentConnections uint64 `json:"current_connections"`
+	Revision           string `json:"revision,omitempty"`
+}
+
+// UniqueIPsRow describes IP presence across servers for one user.
+type UniqueIPsRow struct {
+	Username string          `json:"username"`
+	IPs      []IPAssignments `json:"ips"`
+}
+
+// IPAssignments per-IP server visibility (snapshot-based).
+type IPAssignments struct {
+	IP                string   `json:"ip"`
+	ActiveOnServers   []string `json:"active_on_servers"`
+	RecentOnServers   []string `json:"recent_on_servers"`
+	PrimaryServer     *string  `json:"primary_server,omitempty"`
+}
+
+// UsersRow merged user view with optional per-server detail and links.
+type UsersRow struct {
+	Username       string                        `json:"username"`
+	TotalOctets    uint64                        `json:"total_octets"`
+	ByServer       map[string]TrafficServerStats `json:"by_server"`
+	Links          *UserLinks                    `json:"links,omitempty"`
+	ActiveUniqueIPs uint64                       `json:"active_unique_ips"`
+	RecentUniqueIPs uint64                       `json:"recent_unique_ips"`
+}
+
+// SummaryData fleet snapshot.
+type SummaryData struct {
+	Servers              []ServerFetchResult `json:"servers"`
+	ServersTotal         int                 `json:"servers_total"`
+	ServersOK            int                 `json:"servers_ok"`
+	ServersFailed        int                 `json:"servers_failed"`
+	FleetTotalOctets     uint64              `json:"fleet_total_octets"`
+	FleetTotalConnections uint64             `json:"fleet_total_connections"`
+	TopUsers             []TopUser           `json:"top_users"`
+}
+
+// TopUser by summed total_octets across servers for one username.
+type TopUser struct {
+	Username    string `json:"username"`
+	TotalOctets uint64 `json:"total_octets"`
+}

internal/config/config.go

@@ -15,11 +15,18 @@ var aliasRe = regexp.MustCompile(`^[a-z0-9][a-z0-9_-]*$`)
 
 // Config is the gateway YAML configuration.
 type Config struct {
-	Listen         string   `yaml:"listen"`
-	AllowAll       bool     `yaml:"allow_all"`
-	WhitelistCIDRs []string `yaml:"whitelist_cidrs"`
-	TrustedProxies []string `yaml:"trusted_proxies"`
-	Servers        []Server `yaml:"servers"`
+	Listen         string           `yaml:"listen"`
+	AllowAll       bool             `yaml:"allow_all"`
+	WhitelistCIDRs []string         `yaml:"whitelist_cidrs"`
+	TrustedProxies []string         `yaml:"trusted_proxies"`
+	Servers        []Server         `yaml:"servers"`
+	Aggregate      *AggregateConfig `yaml:"aggregate"`
+}
+
+// AggregateConfig controls default scope of /api/agg/* (optional).
+type AggregateConfig struct {
+	// IncludeAliases limits aggregation to these server aliases; empty means all servers.
+	IncludeAliases []string `yaml:"include_aliases"`
 }
 
 // Server maps a URL alias to an upstream base URL.
@@ -63,6 +70,9 @@ func (c *Config) Validate() error {
 		if _, ok := seen[s.Alias]; ok {
 			return fmt.Errorf("duplicate alias %q", s.Alias)
 		}
+		if s.Alias == "agg" {
+			return fmt.Errorf("servers[%d]: alias %q is reserved for /api/agg/", i, s.Alias)
+		}
 		seen[s.Alias] = struct{}{}
 		if s.BaseURL == "" {
 			return fmt.Errorf("servers[%d]: base_url is required", i)
@@ -92,6 +102,17 @@ func (c *Config) Validate() error {
 			return fmt.Errorf("trusted_proxies[%d]: %w", i, err)
 		}
 	}
+	if c.Aggregate != nil {
+		for i, a := range c.Aggregate.IncludeAliases {
+			a = strings.TrimSpace(a)
+			if a == "" {
+				return fmt.Errorf("aggregate.include_aliases[%d]: empty entry", i)
+			}
+			if !seen[a] {
+				return fmt.Errorf("aggregate.include_aliases[%d]: unknown server alias %q", i, a)
+			}
+		}
+	}
 	return nil
 }
 

internal/config/config_test.go

@@ -62,3 +62,39 @@ func TestValidateDuplicateAlias(t *testing.T) {
 		t.Fatal("expected error")
 	}
 }
+
+func TestValidateReservedAggAlias(t *testing.T) {
+	c := &Config{
+		Servers: []Server{
+			{Alias: "agg", BaseURL: "http://x:1"},
+		},
+	}
+	if err := c.Validate(); err == nil {
+		t.Fatal("expected error for reserved alias agg")
+	}
+}
+
+func TestValidateAggregateIncludeAliases(t *testing.T) {
+	c := &Config{
+		Servers: []Server{
+			{Alias: "a", BaseURL: "http://x:1"},
+		},
+		Aggregate: &AggregateConfig{
+			IncludeAliases: []string{"a"},
+		},
+	}
+	if err := c.Validate(); err != nil {
+		t.Fatal(err)
+	}
+	c2 := &Config{
+		Servers: []Server{
+			{Alias: "a", BaseURL: "http://x:1"},
+		},
+		Aggregate: &AggregateConfig{
+			IncludeAliases: []string{"nope"},
+		},
+	}
+	if err := c2.Validate(); err == nil {
+		t.Fatal("expected error for unknown include alias")
+	}
+}

internal/server/gateway.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
+	"github.com/telemt/telemt-api/internal/aggregate"
 	"github.com/telemt/telemt-api/internal/config"
 	"github.com/telemt/telemt-api/internal/proxy"
 )
@@ -21,6 +22,7 @@ import (
 type Gateway struct {
 	parsed        *config.Parsed
 	proxies       map[string]*httputil.ReverseProxy
+	agg           *aggregate.Handler
 	log           *slog.Logger
 	transport     *http.Transport
 	promHandler   http.Handler
@@ -63,6 +65,7 @@ func NewGateway(p *config.Parsed, log *slog.Logger) (*Gateway, error) {
 		}
 		g.proxies[s.Alias] = rp
 	}
+	g.agg = aggregate.NewHandler(p, &http.Client{Transport: t})
 	return g, nil
 }
 
@@ -170,6 +173,10 @@ func (g *Gateway) serve(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	const prefix = "/api/"
+	if r.URL.Path == "/api/agg" || strings.HasPrefix(r.URL.Path, "/api/agg/") {
+		g.agg.ServeHTTP(w, r)
+		return
+	}
 	if !strings.HasPrefix(r.URL.Path, prefix) {
 		http.NotFound(w, r)
 		return