feat(go): add realtime pipeline observability

go/vehicle-gateway/internal/realtime/http.go

@@ -3,7 +3,9 @@ package realtime
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/redis/go-redis/v9"
@@ -15,6 +17,14 @@ type Handler struct {
 	repository *Repository
 }
 
+type OnlineIndexHandler struct {
+	repository *Repository
+}
+
+type PipelineDebugHandler struct {
+	repository *Repository
+}
+
 func NewHandler(repository *Repository) *Handler {
 	if repository == nil {
 		panic("realtime repository must not be nil")
@@ -22,6 +32,20 @@ func NewHandler(repository *Repository) *Handler {
 	return &Handler{repository: repository}
 }
 
+func NewOnlineIndexHandler(repository *Repository) *OnlineIndexHandler {
+	if repository == nil {
+		panic("online index repository must not be nil")
+	}
+	return &OnlineIndexHandler{repository: repository}
+}
+
+func NewPipelineDebugHandler(repository *Repository) *PipelineDebugHandler {
+	if repository == nil {
+		panic("pipeline debug repository must not be nil")
+	}
+	return &PipelineDebugHandler{repository: repository}
+}
+
 func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodGet {
 		writeError(w, http.StatusMethodNotAllowed, "method not allowed")
@@ -52,6 +76,76 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (h *OnlineIndexHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		writeError(w, http.StatusMethodNotAllowed, "method not allowed")
+		return
+	}
+	if strings.Trim(r.URL.Path, "/") != "api/realtime/online" {
+		writeError(w, http.StatusNotFound, "route not found")
+		return
+	}
+	query, err := parseOnlineListQuery(r)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+	items, total, err := h.repository.ListOnline(r.Context(), query)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]any{
+		"items":  items,
+		"total":  total,
+		"limit":  query.Limit,
+		"offset": query.Offset,
+	})
+}
+
+func (h *PipelineDebugHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		writeError(w, http.StatusMethodNotAllowed, "method not allowed")
+		return
+	}
+	if strings.Trim(r.URL.Path, "/") != "api/debug/pipeline" {
+		writeError(w, http.StatusNotFound, "route not found")
+		return
+	}
+	summary, err := h.repository.PipelineSummary(r.Context())
+	writeResult(w, summary, err)
+}
+
+func parseOnlineListQuery(r *http.Request) (OnlineListQuery, error) {
+	values := r.URL.Query()
+	limit, err := parseBoundedInt(values.Get("limit"), 100, 1, 1000, "limit")
+	if err != nil {
+		return OnlineListQuery{}, err
+	}
+	offset, err := parseBoundedInt(values.Get("offset"), 0, 0, 1_000_000, "offset")
+	if err != nil {
+		return OnlineListQuery{}, err
+	}
+	return normalizeOnlineListQuery(OnlineListQuery{
+		Protocol: envelope.Protocol(values.Get("protocol")),
+		Limit:    limit,
+		Offset:   offset,
+	}), nil
+}
+
+func parseBoundedInt(raw string, fallback int, min int, max int, name string) (int, error) {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return fallback, nil
+	}
+	value, err := strconv.Atoi(raw)
+	if err != nil || value < min || value > max {
+		return 0, fmt.Errorf("%s must be between %d and %d", name, min, max)
+	}
+	return value, nil
+}
+
 func writeResult(w http.ResponseWriter, value any, err error) {
 	if err != nil {
 		if errors.Is(err, redis.Nil) {

go/vehicle-gateway/internal/realtime/model.go

@@ -37,6 +37,24 @@ type OnlineStatus struct {
 	TTLSeconds     int64               `json:"ttl_seconds"`
 }
 
+type OnlineListQuery struct {
+	Protocol envelope.Protocol `json:"protocol,omitempty"`
+	Limit    int               `json:"limit"`
+	Offset   int               `json:"offset"`
+}
+
+type ProtocolPipelineSummary struct {
+	Protocol     envelope.Protocol `json:"protocol"`
+	IndexedCount int64             `json:"indexed_count"`
+	OnlineCount  int64             `json:"online_count"`
+	LatestSeenMS int64             `json:"latest_seen_ms"`
+}
+
+type PipelineSummary struct {
+	Protocols []ProtocolPipelineSummary `json:"protocols"`
+	UpdatedMS int64                     `json:"updated_ms"`
+}
+
 type Config struct {
 	OnlineTTL time.Duration
 }

go/vehicle-gateway/internal/realtime/repository.go

@@ -195,6 +195,141 @@ func (r *Repository) IsOnline(ctx context.Context, vehicleKey string) (OnlineSta
 	return best, nil
 }
 
+func (r *Repository) ListOnline(ctx context.Context, query OnlineListQuery) ([]OnlineStatus, int64, error) {
+	query = normalizeOnlineListQuery(query)
+	total, err := r.onlineTotal(ctx, query.Protocol)
+	if err != nil {
+		return nil, 0, err
+	}
+	items := make([]OnlineStatus, 0, query.Limit)
+	start := int64(query.Offset)
+	if query.Protocol != "" {
+		start = 0
+	}
+	skipped := 0
+	batchSize := int64(query.Limit * 5)
+	if batchSize < 100 {
+		batchSize = 100
+	}
+	for len(items) < query.Limit {
+		stop := start + batchSize - 1
+		members, err := r.client.ZRevRangeWithScores(ctx, "vehicle:last_seen", start, stop).Result()
+		if err != nil {
+			return nil, 0, err
+		}
+		if len(members) == 0 {
+			break
+		}
+		for _, member := range members {
+			protocol, vin, ok := splitOnlineMember(fmt.Sprint(member.Member))
+			if !ok {
+				continue
+			}
+			if query.Protocol != "" && protocol != query.Protocol {
+				continue
+			}
+			if query.Protocol != "" && skipped < query.Offset {
+				skipped++
+				continue
+			}
+			status, err := r.onlineStatusFor(ctx, protocol, vin, int64(member.Score))
+			if err != nil {
+				return nil, 0, err
+			}
+			items = append(items, status)
+			if len(items) >= query.Limit {
+				break
+			}
+		}
+		if int64(len(members)) < batchSize {
+			break
+		}
+		start += batchSize
+	}
+	return items, total, nil
+}
+
+func (r *Repository) PipelineSummary(ctx context.Context) (PipelineSummary, error) {
+	out := PipelineSummary{
+		Protocols: make([]ProtocolPipelineSummary, 0, len(knownRealtimeProtocols())),
+		UpdatedMS: time.Now().UnixMilli(),
+	}
+	for _, protocol := range knownRealtimeProtocols() {
+		indexed, err := r.client.SCard(ctx, realtimeIndexKey(protocol)).Result()
+		if err != nil {
+			return PipelineSummary{}, err
+		}
+		onlineCount, latestSeen, err := r.protocolOnlineStats(ctx, protocol)
+		if err != nil {
+			return PipelineSummary{}, err
+		}
+		out.Protocols = append(out.Protocols, ProtocolPipelineSummary{
+			Protocol:     protocol,
+			IndexedCount: indexed,
+			OnlineCount:  onlineCount,
+			LatestSeenMS: latestSeen,
+		})
+	}
+	return out, nil
+}
+
+func (r *Repository) onlineTotal(ctx context.Context, protocol envelope.Protocol) (int64, error) {
+	if protocol != "" {
+		return r.client.SCard(ctx, realtimeIndexKey(protocol)).Result()
+	}
+	return r.client.ZCard(ctx, "vehicle:last_seen").Result()
+}
+
+func (r *Repository) protocolOnlineStats(ctx context.Context, protocol envelope.Protocol) (int64, int64, error) {
+	members, err := r.client.SMembers(ctx, realtimeIndexKey(protocol)).Result()
+	if err != nil {
+		return 0, 0, err
+	}
+	var onlineCount int64
+	var latestSeen int64
+	for _, vin := range members {
+		status, err := r.onlineStatusFor(ctx, protocol, vin, 0)
+		if err != nil {
+			return 0, 0, err
+		}
+		if status.Online {
+			onlineCount++
+		}
+		if status.LastSeenMS > latestSeen {
+			latestSeen = status.LastSeenMS
+		}
+	}
+	return onlineCount, latestSeen, nil
+}
+
+func (r *Repository) onlineStatusFor(ctx context.Context, protocol envelope.Protocol, vin string, scoreMS int64) (OnlineStatus, error) {
+	key := onlineKey(protocol, vin)
+	stateKey := onlineStateKey(protocol, vin)
+	state, err := r.client.HGetAll(ctx, stateKey).Result()
+	if err != nil {
+		return OnlineStatus{}, err
+	}
+	ttl := r.client.TTL(ctx, key).Val()
+	status := OnlineStatus{
+		VehicleKey:     firstNonEmptyString(state["vehicle_key"], vin),
+		VIN:            firstNonEmptyString(state["vin"], vin),
+		Protocol:       protocol,
+		Online:         ttl > 0,
+		LastSeenMS:     firstPositiveInt64(parseInt64(state["last_seen_ms"]), scoreMS),
+		OfflineAfterMS: parseInt64(state["offline_after_ms"]),
+		SourceEndpoint: state["source_endpoint"],
+		Protocols:      []envelope.Protocol{protocol},
+		TTLSeconds:     int64(ttl.Seconds()),
+	}
+	if status.TTLSeconds < 0 {
+		status.TTLSeconds = 0
+	}
+	if status.OfflineAfterMS <= 0 && status.LastSeenMS > 0 {
+		status.OfflineAfterMS = status.LastSeenMS + r.cfg.ttl().Milliseconds()
+	}
+	return status, nil
+}
+
 func (r *Repository) getSnapshot(ctx context.Context, key string) (Snapshot, error) {
 	var snapshot Snapshot
 	payload, err := r.client.Get(ctx, key).Bytes()
@@ -741,3 +876,48 @@ func knownRealtimeProtocols() []envelope.Protocol {
 func protocolsKey(vehicleKey string) string {
 	return "vehicle:protocols:" + strings.TrimSpace(vehicleKey)
 }
+
+func normalizeOnlineListQuery(query OnlineListQuery) OnlineListQuery {
+	query.Protocol = envelope.Protocol(strings.ToUpper(strings.TrimSpace(string(query.Protocol))))
+	if query.Limit <= 0 {
+		query.Limit = 100
+	}
+	if query.Limit > 1000 {
+		query.Limit = 1000
+	}
+	if query.Offset < 0 {
+		query.Offset = 0
+	}
+	return query
+}
+
+func splitOnlineMember(member string) (envelope.Protocol, string, bool) {
+	protocolText, vin, ok := strings.Cut(strings.TrimSpace(member), ":")
+	if !ok || strings.TrimSpace(protocolText) == "" || strings.TrimSpace(vin) == "" {
+		return "", "", false
+	}
+	return envelope.Protocol(protocolText), strings.TrimSpace(vin), true
+}
+
+func parseInt64(value string) int64 {
+	parsed, _ := strconv.ParseInt(strings.TrimSpace(value), 10, 64)
+	return parsed
+}
+
+func firstPositiveInt64(values ...int64) int64 {
+	for _, value := range values {
+		if value > 0 {
+			return value
+		}
+	}
+	return 0
+}
+
+func firstNonEmptyString(values ...string) string {
+	for _, value := range values {
+		if strings.TrimSpace(value) != "" {
+			return strings.TrimSpace(value)
+		}
+	}
+	return ""
+}

go/vehicle-gateway/internal/realtime/repository_test.go

@@ -430,6 +430,125 @@ func TestRepositoryFastUpdateOnlyWritesPermanentKVAndMinuteOnline(t *testing.T)
 	}
 }
 
+func TestRepositoryListsOnlineStatusesAndPipelineSummary(t *testing.T) {
+	repo, closeFn := newTestRepository(t)
+	defer closeFn()
+	ctx := context.Background()
+
+	for _, env := range []envelope.FrameEnvelope{
+		{
+			Protocol:     envelope.ProtocolGB32960,
+			VIN:          "VIN001",
+			EventTimeMS:  1000,
+			ReceivedAtMS: 1100,
+			Parsed: map[string]any{
+				"data_units": []any{map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 88.0}}},
+			},
+		},
+		{
+			Protocol:     envelope.ProtocolJT808,
+			VIN:          "VIN002",
+			EventTimeMS:  2000,
+			ReceivedAtMS: 2200,
+			MessageID:    "0x0200",
+			Parsed:       map[string]any{"location": map[string]any{"longitude": 121.1, "latitude": 30.2}},
+			Fields:       map[string]any{envelope.FieldLongitude: 121.1, envelope.FieldLatitude: 30.2},
+		},
+	} {
+		if err := repo.FastUpdate(ctx, env); err != nil {
+			t.Fatalf("FastUpdate() error = %v", err)
+		}
+	}
+
+	items, total, err := repo.ListOnline(ctx, OnlineListQuery{Protocol: envelope.ProtocolJT808, Limit: 10})
+	if err != nil {
+		t.Fatalf("ListOnline() error = %v", err)
+	}
+	if total != 1 || len(items) != 1 || items[0].VIN != "VIN002" || !items[0].Online {
+		t.Fatalf("jt808 online list total=%d items=%#v", total, items)
+	}
+
+	summary, err := repo.PipelineSummary(ctx)
+	if err != nil {
+		t.Fatalf("PipelineSummary() error = %v", err)
+	}
+	got := map[envelope.Protocol]ProtocolPipelineSummary{}
+	for _, item := range summary.Protocols {
+		got[item.Protocol] = item
+	}
+	if got[envelope.ProtocolGB32960].IndexedCount != 1 || got[envelope.ProtocolGB32960].OnlineCount != 1 || got[envelope.ProtocolGB32960].LatestSeenMS != 1100 {
+		t.Fatalf("gb32960 summary = %#v", got[envelope.ProtocolGB32960])
+	}
+	if got[envelope.ProtocolJT808].IndexedCount != 1 || got[envelope.ProtocolJT808].OnlineCount != 1 || got[envelope.ProtocolJT808].LatestSeenMS != 2200 {
+		t.Fatalf("jt808 summary = %#v", got[envelope.ProtocolJT808])
+	}
+}
+
+func TestOnlineIndexHandlerReturnsPagedOnlineStatuses(t *testing.T) {
+	repo, closeFn := newTestRepository(t)
+	defer closeFn()
+	ctx := context.Background()
+
+	if err := repo.FastUpdate(ctx, envelope.FrameEnvelope{
+		Protocol:     envelope.ProtocolGB32960,
+		VIN:          "VIN001",
+		EventTimeMS:  1000,
+		ReceivedAtMS: 1100,
+		Parsed: map[string]any{
+			"data_units": []any{map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 88.0}}},
+		},
+	}); err != nil {
+		t.Fatalf("FastUpdate() error = %v", err)
+	}
+
+	handler := NewOnlineIndexHandler(repo)
+	request := httptest.NewRequest(http.MethodGet, "/api/realtime/online?protocol=gb32960&limit=5", nil)
+	response := httptest.NewRecorder()
+	handler.ServeHTTP(response, request)
+
+	if response.Code != http.StatusOK {
+		t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
+	}
+	body := response.Body.String()
+	for _, want := range []string{`"protocol":"GB32960"`, `"vin":"VIN001"`, `"online":true`, `"total":1`} {
+		if !strings.Contains(body, want) {
+			t.Fatalf("response missing %s: %s", want, body)
+		}
+	}
+}
+
+func TestPipelineDebugHandlerReturnsProtocolSummary(t *testing.T) {
+	repo, closeFn := newTestRepository(t)
+	defer closeFn()
+	ctx := context.Background()
+
+	if err := repo.FastUpdate(ctx, envelope.FrameEnvelope{
+		Protocol:     envelope.ProtocolYutongMQTT,
+		VIN:          "VIN003",
+		EventTimeMS:  3000,
+		ReceivedAtMS: 3300,
+		Parsed:       map[string]any{"data": map[string]any{"TOTAL_MILEAGE": 1}},
+		Fields:       map[string]any{envelope.FieldTotalMileageKM: 1},
+	}); err != nil {
+		t.Fatalf("FastUpdate() error = %v", err)
+	}
+
+	handler := NewPipelineDebugHandler(repo)
+	request := httptest.NewRequest(http.MethodGet, "/api/debug/pipeline", nil)
+	response := httptest.NewRecorder()
+	handler.ServeHTTP(response, request)
+
+	if response.Code != http.StatusOK {
+		t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
+	}
+	body := response.Body.String()
+	for _, want := range []string{`"protocol":"YUTONG_MQTT"`, `"indexed_count":1`, `"online_count":1`, `"latest_seen_ms":3300`} {
+		if !strings.Contains(body, want) {
+			t.Fatalf("response missing %s: %s", want, body)
+		}
+	}
+}
+
 func TestRepositoryReplacesJT808LocationWhenUpdatingRealtimeRaw(t *testing.T) {
 	repo, closeFn := newTestRepository(t)
 	defer closeFn()