telemt-api/internal/aggregate/fetch.go

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] + "…"
}