telemt-api/internal/aggregate/fetch.go

package aggregate

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"time"

	"github.com/telemt/telemt-api/internal/config"
	"github.com/telemt/telemt-api/internal/proxy"
)

const statsUsersPath = "stats/users"

// UpstreamCallMeta describes one upstream Telemt GET outcome (before typed data).
type UpstreamCallMeta struct {
	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"`
}

type telemtEnvelope[T any] struct {
	OK       bool   `json:"ok"`
	Data     T      `json:"data"`
	Revision string `json:"revision"`
}

// FetchTelemtGET performs GET {base}{path_prefix}/{relPath} and decodes Telemt success envelope into T.
// On upstream/network errors, returns zero T and meta with OK=false; meta.Error is set.
func FetchTelemtGET[T any](ctx context.Context, client *http.Client, parsed *config.Parsed, alias string, relPath string) (out T, meta UpstreamCallMeta) {
	var zero T
	srv := parsed.ByAlias[alias]
	if srv == nil {
		return zero, UpstreamCallMeta{OK: false, Error: "unknown alias"}
	}
	u, err := url.Parse(srv.BaseURL)
	if err != nil {
		return zero, UpstreamCallMeta{OK: false, Error: err.Error()}
	}
	target := proxy.JoinPathPrefix(u, srv.PathPrefix, relPath)
	start := time.Now()
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), nil)
	if err != nil {
		return zero, UpstreamCallMeta{OK: false, LatencyMs: time.Since(start).Milliseconds(), Error: err.Error()}
	}
	if auth := parsed.AuthByAlias[alias]; auth != "" {
		req.Header.Set("Authorization", auth)
	}
	resp, err := client.Do(req)
	latency := time.Since(start).Milliseconds()
	if err != nil {
		return zero, UpstreamCallMeta{OK: false, LatencyMs: latency, Error: err.Error()}
	}
	body, readErr := io.ReadAll(io.LimitReader(resp.Body, 64<<20))
	_ = resp.Body.Close()
	if readErr != nil {
		return zero, UpstreamCallMeta{OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency, Error: readErr.Error()}
	}
	if resp.StatusCode != http.StatusOK {
		return zero, UpstreamCallMeta{
			OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency,
			Error: fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(body), 200)),
		}
	}
	var env telemtEnvelope[json.RawMessage]
	if err := json.Unmarshal(body, &env); err != nil {
		return zero, UpstreamCallMeta{OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency, Error: "invalid json: " + err.Error()}
	}
	if !env.OK {
		return zero, UpstreamCallMeta{OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency, Error: "upstream ok=false"}
	}
	var data T
	if err := json.Unmarshal(env.Data, &data); err != nil {
		return zero, UpstreamCallMeta{OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency, Error: "decode data: " + err.Error()}
	}
	return data, UpstreamCallMeta{OK: true, HTTPStatus: resp.StatusCode, LatencyMs: latency, Revision: env.Revision}
}

// FetchStatsUsers calls GET stats/users for each alias using FetchTelemtGET.
func FetchStatsUsers(ctx context.Context, client *http.Client, parsed *config.Parsed, aliases []string) []ServerFetchResult {
	out := make([]ServerFetchResult, 0, len(aliases))
	for _, alias := range aliases {
		users, meta := FetchTelemtGET[[]UserInfo](ctx, client, parsed, alias, statsUsersPath)
		fr := ServerFetchResult{
			Alias:      alias,
			OK:         meta.OK,
			HTTPStatus: meta.HTTPStatus,
			LatencyMs:  meta.LatencyMs,
			Error:      meta.Error,
			Revision:   meta.Revision,
		}
		if meta.OK {
			fr.Users = users
		}
		out = append(out, fr)
	}
	return out
}

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n] + "…"
}