feat: add time range validation and historical data querying

- Introduced CalculateStepForTimeRange function for real-time updates without maxDataPoints.
- Added QueryDataPointsWithStep function to fetch full aggregated series for fixed steps.
- Implemented ValidateTimeRange and QueryHistoricalDataPoints for time range validation and querying.
- Enhanced websocket handling with metrics for initial data latency, live fanout, and event counts.
- Updated websocket tests to cover new subscription and data handling logic.

Co-authored-by: Copilot <copilot@github.com>

Author: byPixelTV

data/query.go

@@ -114,6 +114,35 @@ func CalculateOptimalStep(start string, maxDataPoints int) (string, error) {
 	return CalculateOptimalStepWithMin(start, maxDataPoints, 10)
 }
 
+// CalculateStepForTimeRange calculates the aggregation step for a time range without needing maxDataPoints
+// Used for quick step determination in real-time updates
+func CalculateStepForTimeRange(timeRange string) (string, error) {
+	rangeInMinutes, err := timeToMinutes(timeRange)
+	if err != nil {
+		return "", err
+	}
+	rangeInMinutes = math.Abs(rangeInMinutes)
+
+	// Use sensible defaults for different ranges
+	var stepMinutes float64
+	switch {
+	case rangeInMinutes <= 60: // <= 1 hour
+		stepMinutes = 10.0 / 60.0 // 10 seconds
+	case rangeInMinutes <= 360: // <= 6 hours
+		stepMinutes = 1.0 // 1 minute
+	case rangeInMinutes <= 1440: // <= 1 day
+		stepMinutes = 4.0 // 4 minutes
+	case rangeInMinutes <= 7200: // <= 5 days
+		stepMinutes = 20.0 // 20 minutes
+	case rangeInMinutes <= 10080: // <= 7 days
+		stepMinutes = 30.0 // 30 minutes
+	default:
+		stepMinutes = 60.0 // 1 hour
+	}
+
+	return minutesToTime(stepMinutes), nil
+}
+
 // CalculateOptimalStepWithMin calculates the optimal step for a given time range
 // Uses adaptive strategy to ensure meaningful data points even with sparse data
 func CalculateOptimalStepWithMin(start string, maxDataPoints, minDataPoints int) (string, error) {

data/servers.go

@@ -127,6 +127,67 @@ func QueryDataPoints(ip string, duration string) ([]ServerDataPoint, string, err
 	return points, step, nil
 }
 
+// QueryDataPointsWithStep returns the full aggregated series for a fixed step without additional downsampling.
+// This is used for websocket initial_data payloads where the frontend should receive the exact series
+// that matches the chosen chart range.
+func QueryDataPointsWithStep(ip string, duration string, step string) ([]ServerDataPoint, string, error) {
+	queryApi := database.InfluxClient.QueryAPI(os.Getenv("INFLUXDB_ORG"))
+
+	query, err := BuildInfluxQuery(duration, step, ip)
+	if err != nil {
+		return nil, step, fmt.Errorf("failed to build query: %w", err)
+	}
+
+	result, err := queryApi.Query(context.Background(), query)
+	if err != nil {
+		return nil, step, fmt.Errorf("query execution failed: %w", err)
+	}
+	defer func() { _ = result.Close() }()
+
+	var dataPoints []ServerDataPoint
+	for result.Next() {
+		record := result.Record()
+		if record == nil {
+			continue
+		}
+
+		playerCount, ok := recordValueToInt(record.Value())
+		if !ok {
+			continue
+		}
+
+		point := ServerDataPoint{
+			Timestamp:   record.Time().Unix(),
+			PlayerCount: playerCount,
+			Ip:          record.ValueByKey("ip").(string),
+			Name:        record.ValueByKey("name").(string),
+		}
+
+		if ip == "" || point.Ip == ip {
+			dataPoints = append(dataPoints, point)
+		}
+	}
+
+	if result.Err() != nil {
+		return nil, step, fmt.Errorf("result error: %w", result.Err())
+	}
+
+	sort.Slice(dataPoints, func(i, j int) bool {
+		if dataPoints[i].Timestamp != dataPoints[j].Timestamp {
+			return dataPoints[i].Timestamp < dataPoints[j].Timestamp
+		}
+		if dataPoints[i].PlayerCount != dataPoints[j].PlayerCount {
+			return dataPoints[i].PlayerCount < dataPoints[j].PlayerCount
+		}
+		if dataPoints[i].Ip != dataPoints[j].Ip {
+			return dataPoints[i].Ip < dataPoints[j].Ip
+		}
+		return dataPoints[i].Name < dataPoints[j].Name
+	})
+
+	return dataPoints, step, nil
+}
+
 func durationLongerThanADay(duration string) bool {
 	rangeInMinutes, err := timeToMinutes(duration)
 	if err != nil {

data/time_range.go

@@ -0,0 +1,100 @@
+package data
+
+import (
+	"MineTracker/database"
+	"context"
+	"fmt"
+	"os"
+	"sort"
+	"strings"
+)
+
+var allowedTimeRanges = map[string]string{
+	"1h":  "10s",
+	"6h":  "1m",
+	"12h": "2m",
+	"24h": "4m",
+	"7d":  "30m",
+	"30d": "2h",
+	"1y":  "1d",
+}
+
+// ValidateTimeRange returns the canonical time range and the exact aggregation step.
+func ValidateTimeRange(timeRange string) (string, string, error) {
+	normalized := strings.TrimSpace(timeRange)
+	step, ok := allowedTimeRanges[normalized]
+	if !ok {
+		return "", "", fmt.Errorf("unsupported time range: %s", timeRange)
+	}
+	return normalized, step, nil
+}
+
+// QueryHistoricalDataPoints queries the historical series for a validated time range.
+// The returned step is the exact aggregation step used in the query.
+func QueryHistoricalDataPoints(ip string, timeRange string) ([]ServerDataPoint, string, error) {
+	canonicalRange, step, err := ValidateTimeRange(timeRange)
+	if err != nil {
+		return nil, "", err
+	}
+
+	query, _, resolvedStep, err := BuildInfluxQueryFromParams(QueryParams{
+		Start:         "-" + canonicalRange,
+		Step:          step,
+		ServerFilter:  ip,
+		MaxDataPoints: 5000,
+		MinDataPoints: 1,
+		UseAdaptive:   false,
+	})
+	if err != nil {
+		return nil, "", err
+	}
+
+	queryApi := database.InfluxClient.QueryAPI(os.Getenv("INFLUXDB_ORG"))
+	result, err := queryApi.Query(context.Background(), query)
+	if err != nil {
+		return nil, "", fmt.Errorf("query execution failed: %w", err)
+	}
+	defer func() { _ = result.Close() }()
+
+	points := make([]ServerDataPoint, 0, 512)
+	for result.Next() {
+		record := result.Record()
+		if record == nil {
+			continue
+		}
+
+		playerCount, ok := recordValueToInt(record.Value())
+		if !ok {
+			continue
+		}
+
+		point := ServerDataPoint{
+			Timestamp:   record.Time().Unix(),
+			PlayerCount: playerCount,
+			Ip:          record.ValueByKey("ip").(string),
+			Name:        record.ValueByKey("name").(string),
+		}
+		if ip == "" || point.Ip == ip {
+			points = append(points, point)
+		}
+	}
+
+	if result.Err() != nil {
+		return nil, "", fmt.Errorf("result error: %w", result.Err())
+	}
+
+	sort.Slice(points, func(i, j int) bool {
+		if points[i].Timestamp != points[j].Timestamp {
+			return points[i].Timestamp < points[j].Timestamp
+		}
+		if points[i].PlayerCount != points[j].PlayerCount {
+			return points[i].PlayerCount < points[j].PlayerCount
+		}
+		if points[i].Ip != points[j].Ip {
+			return points[i].Ip < points[j].Ip
+		}
+		return points[i].Name < points[j].Name
+	})
+
+	return points, resolvedStep, nil
+}

data/time_range_test.go

@@ -0,0 +1,34 @@
+package data
+
+import "testing"
+
+func TestValidateTimeRange(t *testing.T) {
+	tests := []struct {
+		name      string
+		timeRange string
+		wantStep  string
+		wantErr   bool
+	}{
+		{name: "valid 1h", timeRange: "1h", wantStep: "10s"},
+		{name: "valid 7d", timeRange: "7d", wantStep: "30m"},
+		{name: "invalid", timeRange: "168h", wantErr: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, step, err := ValidateTimeRange(tt.timeRange)
+			if tt.wantErr {
+				if err == nil {
+					t.Fatal("expected error")
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if step != tt.wantStep {
+				t.Fatalf("step = %q, want %q", step, tt.wantStep)
+			}
+		})
+	}
+}

main.go

@@ -8,6 +8,7 @@ import (
 	"MineTracker/util"
 	"MineTracker/websocket"
 	"context"
+	"expvar"
 	"net/http"
 	_ "net/http/pprof"
 	"os"
@@ -87,6 +88,7 @@ func main() {
 		routes.RegisterGetBulkDatedDataRoute(r)
 		routes.RegisterGetServers(r)
 		routes.RegisterGetVersionRoute(r)
+		r.GET("/debug/vars", gin.WrapH(expvar.Handler()))
 
 		r.GET("/ws", func(c *gin.Context) {
 			websocket.HandleWebSocket(c.Writer, c.Request)

task/pingJob.go

@@ -456,15 +456,14 @@ func (j *PingJob) pingServer(server data.PingableServer, pinger serverPinger) {
 
 	pc := resp.PlayerCount
 
-	websocket.GlobalHub.SendToServer(server.IP, map[string]interface{}{
-		"type": "data_point_rt",
-		"data": data.ServerDataPoint{
-			Timestamp:   time.Now().Unix(),
-			PlayerCount: pc,
-			Ip:          server.IP,
-			Name:        server.Name,
-		},
-	})
+	dataPoint := data.ServerDataPoint{
+		Timestamp:   time.Now().Unix(),
+		PlayerCount: pc,
+		Ip:          server.IP,
+		Name:        server.Name,
+	}
+
+	websocket.GlobalHub.SendRawDataPoint(server.IP, dataPoint)
 
 	serverCacheMu.RLock()
 	existing, found := serverCacheMap[server.IP]

websocket/metrics.go

@@ -0,0 +1,43 @@
+package websocket
+
+import (
+	"expvar"
+	"time"
+)
+
+var (
+	metricInitialDataLatencyMs = expvar.NewInt("ws_initial_data_latency_ms")
+	metricLiveFanoutMs         = expvar.NewInt("ws_live_fanout_ms")
+	metricSubscriptionCount    = expvar.NewInt("ws_subscription_count")
+	metricDroppedEvents        = expvar.NewInt("ws_dropped_events_total")
+	metricOutOfOrderEvents     = expvar.NewInt("ws_out_of_order_events_total")
+	metricEventsSentTotal      = expvar.NewMap("ws_events_sent_total")
+)
+
+func recordInitialDataLatency(duration time.Duration) {
+	metricInitialDataLatencyMs.Set(duration.Milliseconds())
+}
+
+func recordLiveFanout(duration time.Duration) {
+	metricLiveFanoutMs.Set(duration.Milliseconds())
+}
+
+func recordSubscriptionAdded() {
+	metricSubscriptionCount.Add(1)
+}
+
+func recordSubscriptionRemoved() {
+	metricSubscriptionCount.Add(-1)
+}
+
+func recordDroppedEvent() {
+	metricDroppedEvents.Add(1)
+}
+
+func recordOutOfOrderEvent() {
+	metricOutOfOrderEvents.Add(1)
+}
+
+func recordEventSent(eventType string) {
+	metricEventsSentTotal.Add(eventType, 1)
+}