feat(go): add platform principles and health probes

docs/architecture/iot-data-platform-principles.md

@@ -0,0 +1,81 @@
+# Vehicle IoT Data Platform Principles
+
+## Goal
+
+Build a production vehicle data platform that can receive GB32960, JT808, and Yutong MQTT data reliably, keep raw evidence traceable, and expose only simple business tables for realtime, history, identity, and metrics.
+
+## First Principles
+
+1. Raw data is the source of truth.
+2. Kafka is the durable replay log.
+3. Redis is realtime cache, not permanent storage.
+4. MySQL stores identity, light realtime business snapshots, and low-cardinality metrics.
+5. TDengine stores time-series raw and location history.
+6. Protocol-specific fields stay in raw parsed JSON unless they become a stable query or metric requirement.
+7. Upper business tables must not duplicate full parsed payloads.
+
+## Data Flow
+
+```mermaid
+flowchart LR
+    GB["GB32960 TCP"] --> GW["Go Gateway"]
+    JT["JT808 TCP"] --> GW
+    YM["Yutong MQTT"] --> GW
+
+    GW --> NATS["NATS JetStream ingress"]
+    NATS --> BR["NATS Kafka Bridge"]
+    BR --> KRAW["Kafka raw topics"]
+    GW -. fallback .-> KRAW
+
+    KRAW --> HW["history-writer"]
+    KRAW --> SW["stat-writer"]
+    KRAW --> RA["realtime-api projector"]
+
+    HW --> TDRAW["TDengine raw_frames"]
+    HW --> TDLOC["TDengine vehicle_locations"]
+    SW --> MYMET["MySQL vehicle_daily_metric"]
+    RA --> REDIS["Redis realtime-raw"]
+    RA --> MYSNAP["MySQL vehicle_realtime_snapshot"]
+    RA --> MYLOC["MySQL vehicle_realtime_location"]
+```
+
+## Minimal Storage Contract
+
+| Store | Table or key | Purpose | Keep | Avoid |
+| --- | --- | --- | --- | --- |
+| TDengine | `raw_frames` | Replay and audit evidence | raw hex/text, full parsed JSON, parse status, protocol tags | business-only duplicated columns |
+| TDengine | `vehicle_locations` | High-volume location history | time, vin, protocol, longitude, latitude, speed, direction, mileage | full parsed JSON |
+| MySQL | `vehicle_realtime_snapshot` | Latest per-protocol vehicle heartbeat | protocol, vin, plate, event time, received time | phone, device, message sequence, parsed JSON |
+| MySQL | `vehicle_realtime_location` | Latest business location cache | vin, plate, location, speed, mileage, SOC, event time | full raw payload and protocol internals |
+| MySQL | `vehicle_daily_metric` | Queryable daily metrics | vin or vehicle key, date, metric key, value, method, source mileage | per-frame raw details |
+| MySQL | `vehicle_identity_binding` | Manual identity mapping | vin, plate, phone, device id | registration history |
+| MySQL | `jt808_registration` | JT808 registration and auth trace | phone, device id, plate, auth code, vin match state, first/latest seen | GB32960 or MQTT records |
+| Redis | `realtime-raw:{protocol}:{vin}` | Latest full realtime protocol state | merged latest protocol payload | historical data |
+
+## Event Envelope Rules
+
+Every received frame should become one `FrameEnvelope`.
+
+- `protocol`, `message_id`, `event_time_ms`, `received_at_ms`, and `parse_status` are mandatory.
+- `vin` is preferred as the vehicle identity.
+- JT808 can use `phone` as a temporary key before VIN binding is resolved.
+- `raw_hex` or `raw_text` must be retained for replay and parser correction.
+- `parsed` stores protocol-specific fields.
+- `fields` stores only stable cross-protocol core fields: speed, total mileage, longitude, latitude, SOC.
+
+## Optimization Order
+
+1. Stabilize ingress: connection lifecycle, protocol parser correctness, bounded backpressure, durable publish.
+2. Stabilize event log: NATS to Kafka bridge, topic names, partition keys, retry and replay.
+3. Simplify storage: keep only the minimal tables above, remove duplicated payloads from business tables.
+4. Stabilize realtime: Redis full latest raw state, MySQL lightweight snapshot and location.
+5. Stabilize history: raw and location query pagination, clear time zone behavior.
+6. Stabilize metrics: idempotent daily metrics from mileage differences and raw replay.
+7. Add operations: health, readiness, metrics, lag, connection counts, deploy notes.
+
+## Current Next Steps
+
+1. Add health and readiness endpoints to each long-running Go service.
+2. Add runtime metrics for protocol frame counts, parse failures, Kafka/NATS publish failures, and writer lag.
+3. Review TDengine tables and remove unused tables only after export or explicit confirmation.
+4. Keep new business tables narrow by default. Add a column only when a query or index proves it is needed.

go/vehicle-gateway/cmd/gateway/main.go

@@ -16,6 +16,7 @@ import (
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/gateway"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/gb32960"
@@ -39,6 +40,7 @@ func main() {
 		os.Exit(1)
 	}
 	defer closeResolver()
+	health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-gateway", nil))
 
 	protocols := []gateway.TCPProtocol{
 		{

go/vehicle-gateway/cmd/history-writer/main.go

@@ -14,6 +14,7 @@ import (
 	_ "github.com/taosdata/driver-go/v3/taosWS"
 
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/history"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
 )
@@ -34,6 +35,9 @@ func main() {
 		logger.Error("tdengine ping failed", "error", err)
 		os.Exit(1)
 	}
+	health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-history-writer", []health.Check{
+		{Name: "tdengine", Check: db.PingContext},
+	}))
 
 	writer := history.NewWriter(db)
 	if cfg.EnsureSchema {

go/vehicle-gateway/cmd/nats-kafka-bridge/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/segmentio/kafka-go"
 
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
 )
 
@@ -32,6 +33,14 @@ func main() {
 		os.Exit(1)
 	}
 	defer conn.Close()
+	health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "nats-kafka-bridge", []health.Check{
+		{Name: "nats", Check: func(context.Context) error {
+			if conn.Status() != nats.CONNECTED {
+				return fmt.Errorf("nats status is %s", conn.Status().String())
+			}
+			return nil
+		}},
+	}))
 	js, err := conn.JetStream()
 	if err != nil {
 		logger.Error("nats jetstream init failed", "error", err)

go/vehicle-gateway/cmd/realtime-api/main.go

@@ -18,6 +18,7 @@ import (
 	_ "github.com/taosdata/driver-go/v3/taosWS"
 
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/history"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime"
@@ -46,6 +47,9 @@ func main() {
 	})
 
 	mux := http.NewServeMux()
+	healthChecks := []health.Check{
+		{Name: "redis", Check: func(ctx context.Context) error { return client.Ping(ctx).Err() }},
+	}
 	mux.Handle("/openapi.json", realtime.NewOpenAPIHandler())
 	mux.Handle("/swagger-ui/", realtime.NewSwaggerUIHandler())
 	mux.Handle("/api/realtime/vehicles/", realtime.NewHandler(repository))
@@ -62,6 +66,7 @@ func main() {
 			logger.Error("stats mysql ping failed", "error", err)
 			os.Exit(1)
 		}
+		healthChecks = append(healthChecks, health.Check{Name: "mysql", Check: db.PingContext})
 		closeStats = func() { _ = db.Close() }
 		bindingTable := env("VEHICLE_IDENTITY_TABLE", "vehicle_identity_binding")
 		snapshotWriter := realtime.NewSnapshotWriterWithPlateResolver(db, realtime.NewBindingPlateResolver(db, bindingTable))
@@ -108,6 +113,7 @@ func main() {
 			logger.Error("history tdengine ping failed", "driver", driver, "error", err)
 			os.Exit(1)
 		}
+		healthChecks = append(healthChecks, health.Check{Name: "tdengine", Check: db.PingContext})
 		closeHistory = func() { _ = db.Close() }
 		database := env("TDENGINE_DATABASE", history.DefaultDatabase)
 		mux.Handle("/api/history/raw-frames", history.NewRawFrameHandler(history.NewRawFrameRepository(db, database)))
@@ -126,6 +132,9 @@ func main() {
 		logger.Warn("TDENGINE_DSN is empty; history query api disabled")
 	}
 	defer closeHistory()
+	healthHandler := health.NewHandler("vehicle-realtime-api", healthChecks)
+	mux.Handle("/healthz", healthHandler)
+	mux.Handle("/readyz", healthHandler)
 
 	server := &http.Server{
 		Addr:              env("HTTP_ADDR", ":20210"),

go/vehicle-gateway/cmd/stat-writer/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/segmentio/kafka-go"
 
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats"
 )
@@ -34,6 +35,9 @@ func main() {
 		logger.Error("mysql ping failed", "error", err)
 		os.Exit(1)
 	}
+	health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-stat-writer", []health.Check{
+		{Name: "mysql", Check: db.PingContext},
+	}))
 
 	writer := stats.NewWriter(db, cfg.Location)
 	if cfg.EnsureSchema {

go/vehicle-gateway/internal/health/health.go

@@ -0,0 +1,132 @@
+package health
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strings"
+	"time"
+)
+
+type Check struct {
+	Name  string
+	Check func(context.Context) error
+}
+
+type Handler struct {
+	service string
+	checks  []Check
+}
+
+type Logger interface {
+	Info(string, ...any)
+	Error(string, ...any)
+}
+
+func NewHandler(service string, checks []Check) *Handler {
+	service = strings.TrimSpace(service)
+	if service == "" {
+		service = "vehicle-service"
+	}
+	return &Handler{service: service, checks: checks}
+}
+
+func NewMux(service string, checks []Check) *http.ServeMux {
+	mux := http.NewServeMux()
+	handler := NewHandler(service, checks)
+	mux.Handle("/healthz", handler)
+	mux.Handle("/readyz", handler)
+	return mux
+}
+
+func NewServer(addr string, service string, checks []Check) *http.Server {
+	addr = strings.TrimSpace(addr)
+	if addr == "" {
+		return nil
+	}
+	return &http.Server{
+		Addr:              addr,
+		Handler:           NewMux(service, checks),
+		ReadHeaderTimeout: 5 * time.Second,
+	}
+}
+
+func Start(ctx context.Context, logger Logger, server *http.Server) {
+	if server == nil {
+		return
+	}
+	go func() {
+		<-ctx.Done()
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = server.Shutdown(shutdownCtx)
+	}()
+	go func() {
+		if logger != nil {
+			logger.Info("health server started", "addr", server.Addr)
+		}
+		if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) && logger != nil {
+			logger.Error("health server failed", "error", err)
+		}
+	}()
+}
+
+func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		writeHealth(w, http.StatusMethodNotAllowed, map[string]any{"error": "method not allowed"})
+		return
+	}
+	path := strings.Trim(r.URL.Path, "/")
+	if path != "healthz" && path != "readyz" {
+		writeHealth(w, http.StatusNotFound, map[string]any{"error": "route not found"})
+		return
+	}
+	if path == "healthz" {
+		writeHealth(w, http.StatusOK, map[string]any{
+			"service": h.service,
+			"status":  "ok",
+			"time":    time.Now().UTC().Format(time.RFC3339),
+		})
+		return
+	}
+	h.writeReadiness(w, r)
+}
+
+func (h *Handler) writeReadiness(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
+	defer cancel()
+
+	status := "ok"
+	code := http.StatusOK
+	checks := make(map[string]any, len(h.checks))
+	for _, check := range h.checks {
+		name := strings.TrimSpace(check.Name)
+		if name == "" {
+			continue
+		}
+		if check.Check == nil {
+			checks[name] = map[string]any{"status": "ok"}
+			continue
+		}
+		if err := check.Check(ctx); err != nil {
+			status = "degraded"
+			code = http.StatusServiceUnavailable
+			checks[name] = map[string]any{"status": "error", "error": err.Error()}
+			continue
+		}
+		checks[name] = map[string]any{"status": "ok"}
+	}
+	writeHealth(w, code, map[string]any{
+		"service": h.service,
+		"status":  status,
+		"time":    time.Now().UTC().Format(time.RFC3339),
+		"checks":  checks,
+	})
+}
+
+func writeHealth(w http.ResponseWriter, status int, payload map[string]any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(payload)
+}

go/vehicle-gateway/internal/health/health_test.go

@@ -0,0 +1,98 @@
+package health
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestHandlerReturnsHealthyWhenAllChecksPass(t *testing.T) {
+	handler := NewHandler("vehicle-realtime-api", []Check{
+		{Name: "redis", Check: func(context.Context) error { return nil }},
+		{Name: "mysql", Check: func(context.Context) error { return nil }},
+	})
+	request := httptest.NewRequest(http.MethodGet, "/readyz", 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{`"service":"vehicle-realtime-api"`, `"status":"ok"`, `"redis":{"status":"ok"`, `"mysql":{"status":"ok"`} {
+		if !strings.Contains(body, want) {
+			t.Fatalf("response missing %s: %s", want, body)
+		}
+	}
+}
+
+func TestHandlerReturnsServiceUnavailableWhenCheckFails(t *testing.T) {
+	handler := NewHandler("vehicle-history-writer", []Check{
+		{Name: "tdengine", Check: func(context.Context) error { return errors.New("ping failed") }},
+	})
+	request := httptest.NewRequest(http.MethodGet, "/readyz", nil)
+	response := httptest.NewRecorder()
+
+	handler.ServeHTTP(response, request)
+
+	if response.Code != http.StatusServiceUnavailable {
+		t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
+	}
+	body := response.Body.String()
+	for _, want := range []string{`"service":"vehicle-history-writer"`, `"status":"degraded"`, `"tdengine":`, `"status":"error"`, `"error":"ping failed"`} {
+		if !strings.Contains(body, want) {
+			t.Fatalf("response missing %s: %s", want, body)
+		}
+	}
+}
+
+func TestHandlerRejectsUnknownPath(t *testing.T) {
+	handler := NewHandler("vehicle-gateway", nil)
+	request := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+	response := httptest.NewRecorder()
+
+	handler.ServeHTTP(response, request)
+
+	if response.Code != http.StatusNotFound {
+		t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
+	}
+}
+
+func TestNewMuxRegistersHealthAndReadinessRoutes(t *testing.T) {
+	mux := NewMux("vehicle-stat-writer", []Check{
+		{Name: "mysql", Check: func(context.Context) error { return nil }},
+	})
+	for _, path := range []string{"/healthz", "/readyz"} {
+		request := httptest.NewRequest(http.MethodGet, path, nil)
+		response := httptest.NewRecorder()
+
+		mux.ServeHTTP(response, request)
+
+		if response.Code != http.StatusOK {
+			t.Fatalf("%s status = %d body=%s", path, response.Code, response.Body.String())
+		}
+	}
+}
+
+func TestNewServerReturnsNilWhenAddressIsEmpty(t *testing.T) {
+	if server := NewServer("", "vehicle-gateway", nil); server != nil {
+		t.Fatalf("server = %#v, want nil", server)
+	}
+}
+
+func TestNewServerBuildsConfiguredHTTPServer(t *testing.T) {
+	server := NewServer(":20290", "vehicle-gateway", nil)
+	if server == nil {
+		t.Fatal("server is nil")
+	}
+	if server.Addr != ":20290" {
+		t.Fatalf("addr = %q", server.Addr)
+	}
+	if server.ReadHeaderTimeout == 0 {
+		t.Fatal("ReadHeaderTimeout should be configured")
+	}
+}