refactor(go): make raw topics the realtime source

2026-07-03 08:25:14 +08:00
parent c2d058bf75
commit 4c34c1221b
14 changed files with 151 additions and 44 deletions
--- a/docs/architecture/iot-data-platform-principles.md
+++ b/docs/architecture/iot-data-platform-principles.md
@@ -52,8 +52,8 @@ flowchart LR
 | MySQL | `vehicle_daily_mileage` | Queryable daily mileage | vin, date, protocol, daily mileage, source mileage | temporary vehicle keys, generic metric key/value rows, 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 | `vehicle:latest:{vehicleKey}` | Latest merged realtime state | cross-protocol latest fields only | full parsed payloads and historical data |
-| Redis | `vehicle:realtime-raw:{protocol}:{vehicleKey}` | Latest full realtime protocol state | latest protocol parsed payload | historical data |
+| Redis | `vehicle:latest:{vin}` | Latest merged realtime state | cross-protocol latest fields only for VIN-bound vehicles | full parsed payloads, temporary identities, and historical data |
+| Redis | `vehicle:realtime-raw:{protocol}:{vin}` | Latest full realtime protocol state | latest protocol parsed payload for VIN-bound vehicles | historical data |

 ## Event Envelope Rules

@@ -91,7 +91,7 @@ Runtime metrics are exposed as Prometheus text from `/metrics`. They are operati
 Core counters:

 - `vehicle_gateway_frames_total`: received protocol frames by protocol and parse status.
- `vehicle_gateway_publish_total`: raw and unified publish results by protocol.
+- `vehicle_gateway_publish_total`: raw publish results by protocol. Unified publish metrics appear only when the explicit compatibility stream is enabled.
 - `vehicle_realtime_kafka_messages_total`: realtime consumer messages by topic and status.
 - `vehicle_realtime_updates_total`: Redis/MySQL realtime projector updates by topic and status.
 - `vehicle_history_writes_total`: TDengine history writes by topic and status.
--- a/docs/architecture/production-data-plane-inventory.md
+++ b/docs/architecture/production-data-plane-inventory.md
@@ -25,9 +25,9 @@ NATS 部署在 Kafka ECS 内网 `172.17.111.56:4222`。
 | 项 | 当前值 |
 | --- | --- |
 | Stream | `VEHICLE_INGEST` |
-| Subjects | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1`、`vehicle.event.go.unified.v1` |
+| Subjects | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1` |
 | Durable consumer | `vehicle-kafka-bridge` |
-| 语义 | Gateway 先写 NATS；bridge 写 Kafka 成功后才 ACK NATS |
+| 语义 | Gateway 先写 NATS raw subject；bridge 写 Kafka 成功后才 ACK NATS |

 ### Kafka

@@ -38,7 +38,7 @@ NATS 部署在 Kafka ECS 内网 `172.17.111.56:4222`。
 | `vehicle.raw.go.gb32960.v1` | 12 | NATS Kafka bridge | history、stat、realtime |
 | `vehicle.raw.go.jt808.v1` | 12 | NATS Kafka bridge | history、stat、realtime |
 | `vehicle.raw.go.yutong-mqtt.v1` | 12 | NATS Kafka bridge | history、stat、realtime |
-| `vehicle.event.go.unified.v1` | 12 | NATS Kafka bridge | 预留给统一事件消费 |
+| `vehicle.event.go.unified.v1` | 12 | 显式打开 `PUBLISH_UNIFIED_ENABLED=true` 后才写入 | 兼容旧统一事件消费；当前最小链路不依赖 |

 审计时 Kafka 上仍存在旧 Java/Xinda/telemetry topic，例如 `vehicle.raw.gb32960.v1`、`vehicle.raw.jt808.v1`、`vehicle.event.xinda.v1`、`vehicle.raw.telemetry-input.v1`。这些不属于当前 Go 最小链路，后续如果确认没有消费者依赖，应单独做 Kafka topic 下线清单，不在业务代码里继续引用。

--- a/docs/go-version-data-flow.html
+++ b/docs/go-version-data-flow.html
@@ -380,7 +380,7 @@
          </div>
          <div class="node bus">
            <strong>Subjects</strong>
-            <p><code>vehicle.raw.go.gb32960.v1</code><br><code>vehicle.raw.go.jt808.v1</code><br><code>vehicle.raw.go.yutong-mqtt.v1</code><br><code>vehicle.event.go.unified.v1</code></p>
+            <p><code>vehicle.raw.go.gb32960.v1</code><br><code>vehicle.raw.go.jt808.v1</code><br><code>vehicle.raw.go.yutong-mqtt.v1</code></p>
          </div>
          <div class="node warning">
            <strong>Async + Retry</strong>
@@ -396,7 +396,7 @@
          </div>
          <div class="node bus">
            <strong>Kafka Topics</strong>
-            <p>与 NATS subject 同名：raw 三个协议 + unified 一个实时主题。</p>
+            <p>与 NATS subject 同名：三个协议 RAW topic。<code>vehicle.event.go.unified.v1</code> 仅作为显式兼容开关保留。</p>
          </div>
          <div class="node warning">
            <strong>失败语义</strong>
@@ -416,7 +416,7 @@
          </div>
          <div class="node store">
            <strong>realtime-api 消费器</strong>
-            <p>消费 unified topic，写 Redis 在线状态、实时合并快照、各协议 realtime-raw。</p>
+            <p>消费三个 RAW topic，写 Redis 在线状态、实时合并快照、各协议 realtime-raw。</p>
          </div>
          <div class="node api">
            <strong>HTTP API</strong>
@@ -433,7 +433,7 @@
        <div class="mini-step"><b>2. 拆包</b><span>TCP 按协议提取完整帧；MQTT 直接按消息处理。</span></div>
        <div class="mini-step"><b>3. 解析</b><span>生成 parsed 全量结构化数据和 fields 核心字段。</span></div>
        <div class="mini-step"><b>4. 绑定 VIN</b><span>根据 VIN/phone/device_id/plate 生成 vehicle_key。</span></div>
-        <div class="mini-step"><b>5. 发 NATS</b><span>写 RAW subject；同时写 unified subject。</span></div>
+        <div class="mini-step"><b>5. 发 NATS</b><span>写协议 RAW subject；unified 只在兼容开关打开时写。</span></div>
        <div class="mini-step"><b>6. 桥接 Kafka</b><span>bridge 写 Kafka 成功后 ACK NATS。</span></div>
        <div class="mini-step"><b>7. 多路消费</b><span>历史、实时、统计各自消费 Kafka，互不阻塞。</span></div>
      </div>
--- a/docs/ops/go-service-observability.md
+++ b/docs/ops/go-service-observability.md
@@ -35,7 +35,7 @@ curl -fsS http://127.0.0.1:20200/readyz
 | --- | --- |
 | `vehicle_gateway_active_connections` | Current TCP connections by protocol. Labels: `protocol`. |
 | `vehicle_gateway_frames_total` | Protocol frames received and parsed by the gateway. Labels: `protocol`, `status`. |
-| `vehicle_gateway_publish_total` | Gateway publish result for raw and unified events. Labels: `protocol`, `kind`, `status`. |
+| `vehicle_gateway_publish_total` | Gateway publish result for raw events. Unified appears only when the compatibility stream is explicitly enabled. Labels: `protocol`, `kind`, `status`. |
 | `vehicle_bridge_messages_total` | NATS messages fetched by the bridge. Labels: `subject`, `status`. |
 | `vehicle_bridge_kafka_writes_total` | Bridge writes to Kafka. Labels: `topic`, `status`. |
 | `vehicle_bridge_nats_acks_total` | NATS ack results after Kafka write. Labels: `subject`, `status`. |
--- a/docs/ops/vehicle-ingest-runbook.md
+++ b/docs/ops/vehicle-ingest-runbook.md
@@ -33,8 +33,8 @@
 | --- | --- |
 | `lingniu-go-history-writer.service` | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1` |
 | `lingniu-go-stat-writer.service` | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1` |
-| `lingniu-go-realtime-api.service` | `vehicle.event.go.unified.v1` |
-| `lingniu-go-nats-kafka-bridge.service` | NATS subjects 到同名 Kafka topic；包含三条 raw.go topic 和一条 unified topic |
+| `lingniu-go-realtime-api.service` | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1` |
+| `lingniu-go-nats-kafka-bridge.service` | NATS raw subjects 到同名 Kafka topic；`vehicle.event.go.unified.v1` 只在兼容开关打开时使用 |

 如果 stat-writer 少消费某个 raw topic，对应协议的每日里程不会进入 `vehicle_daily_mileage`。修正后可能出现短时间 Kafka lag，这是在追补历史 backlog；只要 `vehicle_stat_writes_total` 持续增长且 lag 下降，就是健康状态。

--- a/go/vehicle-gateway/cmd/gateway/main.go
+++ b/go/vehicle-gateway/cmd/gateway/main.go
@@ -44,6 +44,7 @@ func main() {
 	defer closeResolver()
 	registry := metrics.NewRegistry()
 	health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-gateway", nil, registry))
+	publishUnified := envBool("PUBLISH_UNIFIED_ENABLED", false)

 	protocols := []gateway.TCPProtocol{
 		{
@@ -74,6 +75,7 @@ func main() {
 			ReadBufferSize: envInt("TCP_READ_BUFFER_BYTES", 64*1024),
 			IdleTimeout:    time.Duration(envInt("TCP_IDLE_TIMEOUT_SECONDS", 180)) * time.Second,
 			MaxConnections: envInt("TCP_MAX_CONNECTIONS", 20_000),
+			PublishUnified: publishUnified,
 		})
 		if err != nil {
 			logger.Error("build tcp server failed", "protocol", protocol.Protocol, "error", err)
@@ -107,6 +109,7 @@ func main() {
 			Resolver:                resolver,
 			Logger:                  logger,
 			Metrics:                 registry,
+			PublishUnified:          publishUnified,
 		})
 		if err != nil {
 			logger.Error("build yutong mqtt client failed", "error", err)
@@ -183,7 +186,7 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error)
 			})
 			logger.Info("nats async publish enabled", "queue_size", queueSize, "workers", workers, "publish_timeout_ms", timeout.Milliseconds())
 		}
-		logger.Info("nats jetstream sink enabled", "url", env("NATS_URL", ""), "unified_subject", natsSinkConfigFromEnv().UnifiedSubject)
+		logger.Info("nats jetstream sink enabled", "url", env("NATS_URL", ""), "unified_subject", natsSinkConfigFromEnv().UnifiedSubject, "publish_unified_enabled", envBool("PUBLISH_UNIFIED_ENABLED", false))
 		return out, nil
 	}
 	brokers := splitCSV(os.Getenv("KAFKA_BROKERS"))
--- a/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go
+++ b/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go
@@ -105,7 +105,9 @@ func loadConfig() config {
 		env("NATS_SUBJECT_GB32960_RAW", env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960)):            env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960),
 		env("NATS_SUBJECT_JT808_RAW", env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808)):                  env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808),
 		env("NATS_SUBJECT_YUTONG_MQTT_RAW", env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT)): env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT),
-		env("NATS_SUBJECT_UNIFIED", env("KAFKA_TOPIC_UNIFIED", topics.Unified)):                       env("KAFKA_TOPIC_UNIFIED", topics.Unified),
+	}
+	if unifiedSubject, unifiedTopic, ok := unifiedRouteFromEnv(); ok {
+		route[unifiedSubject] = unifiedTopic
 	}
 	return config{
 		NATSURL:        env("NATS_URL", "nats://127.0.0.1:4222"),
@@ -124,6 +126,24 @@ func loadConfig() config {
 	}
 }

+func unifiedRouteFromEnv() (string, string, bool) {
+	subject, hasSubject := envOptional("NATS_SUBJECT_UNIFIED")
+	topic, hasTopic := envOptional("KAFKA_TOPIC_UNIFIED")
+	if !hasSubject && !hasTopic {
+		return "", "", false
+	}
+	if subject == "" {
+		subject = topic
+	}
+	if topic == "" {
+		topic = subject
+	}
+	if subject == "" || topic == "" {
+		return "", "", false
+	}
+	return subject, topic, true
+}
+
 type natsPullSubscription interface {
 	Fetch(int, ...nats.PullOpt) ([]*nats.Msg, error)
 }
@@ -290,6 +310,14 @@ func env(key string, fallback string) string {
 	return value
 }

+func envOptional(key string) (string, bool) {
+	value, ok := os.LookupEnv(key)
+	if !ok {
+		return "", false
+	}
+	return strings.TrimSpace(value), true
+}
+
 func envInt(key string, fallback int) int {
 	value := strings.TrimSpace(os.Getenv(key))
 	if value == "" {
--- a/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go
+++ b/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go
@@ -120,7 +120,6 @@ func TestLoadConfigDefaultsToGoSubjectRoutes(t *testing.T) {
 		"vehicle.raw.go.gb32960.v1":     "vehicle.raw.go.gb32960.v1",
 		"vehicle.raw.go.jt808.v1":       "vehicle.raw.go.jt808.v1",
 		"vehicle.raw.go.yutong-mqtt.v1": "vehicle.raw.go.yutong-mqtt.v1",
-		"vehicle.event.go.unified.v1":   "vehicle.event.go.unified.v1",
 	}
 	if len(cfg.Route) != len(want) {
 		t.Fatalf("route len = %d, want %d: %#v", len(cfg.Route), len(want), cfg.Route)
@@ -132,6 +131,17 @@ func TestLoadConfigDefaultsToGoSubjectRoutes(t *testing.T) {
 	}
 }

+func TestLoadConfigIncludesUnifiedOnlyWhenExplicitlyConfigured(t *testing.T) {
+	t.Setenv("NATS_SUBJECT_UNIFIED", "vehicle.event.go.unified.v1")
+	t.Setenv("KAFKA_TOPIC_UNIFIED", "vehicle.event.go.unified.v1")
+
+	cfg := loadConfig()
+
+	if got := cfg.Route["vehicle.event.go.unified.v1"]; got != "vehicle.event.go.unified.v1" {
+		t.Fatalf("unified route = %q, route=%#v", got, cfg.Route)
+	}
+}
+
 type recordingBridgeWriter struct {
 	messages []kafka.Message
 	err      error
--- a/go/vehicle-gateway/cmd/realtime-api/main.go
+++ b/go/vehicle-gateway/cmd/realtime-api/main.go
@@ -190,7 +190,8 @@ func consumeKafka(ctx context.Context, logger interface {
 }

 func kafkaTopicsFromEnv() []string {
-	return splitCSV(env("KAFKA_TOPICS", topics.Unified))
+	defaultTopics := strings.Join([]string{topics.RawGB32960, topics.RawJT808, topics.RawYutongMQTT}, ",")
+	return splitCSV(env("KAFKA_TOPICS", defaultTopics))
 }

 const kafkaMessageOperationTimeout = 30 * time.Second
--- a/go/vehicle-gateway/cmd/realtime-api/main_test.go
+++ b/go/vehicle-gateway/cmd/realtime-api/main_test.go
@@ -66,15 +66,15 @@ func TestProcessRealtimeMessageRecordsMetrics(t *testing.T) {
 		registry,
 		&contextCheckingRealtimeUpdater{},
 		&contextCheckingMessageCommitter{},
-		kafka.Message{Topic: "vehicle.event.go.unified.v1", Partition: 2, Offset: 10, HighWaterMark: 16, Value: payload},
+		kafka.Message{Topic: "vehicle.raw.go.jt808.v1", Partition: 2, Offset: 10, HighWaterMark: 16, Value: payload},
 	)

 	text := registry.Render()
 	for _, want := range []string{
-		`vehicle_realtime_kafka_messages_total{status="received",topic="vehicle.event.go.unified.v1"} 1`,
-		`vehicle_realtime_updates_total{status="ok",topic="vehicle.event.go.unified.v1"} 1`,
-		`vehicle_realtime_kafka_commits_total{status="ok",topic="vehicle.event.go.unified.v1"} 1`,
-		`vehicle_realtime_kafka_lag{partition="2",topic="vehicle.event.go.unified.v1"} 5`,
+		`vehicle_realtime_kafka_messages_total{status="received",topic="vehicle.raw.go.jt808.v1"} 1`,
+		`vehicle_realtime_updates_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`,
+		`vehicle_realtime_kafka_commits_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`,
+		`vehicle_realtime_kafka_lag{partition="2",topic="vehicle.raw.go.jt808.v1"} 5`,
 	} {
 		if !strings.Contains(text, want) {
 			t.Fatalf("metrics missing %s:\n%s", want, text)
@@ -96,9 +96,9 @@ func TestCompositeRealtimeUpdaterUpdatesBothStores(t *testing.T) {
 	}
 }

-func TestKafkaTopicsFromEnvDefaultsToGoUnifiedTopic(t *testing.T) {
+func TestKafkaTopicsFromEnvDefaultsToGoRawTopics(t *testing.T) {
 	got := strings.Join(kafkaTopicsFromEnv(), ",")
-	want := "vehicle.event.go.unified.v1"
+	want := "vehicle.raw.go.gb32960.v1,vehicle.raw.go.jt808.v1,vehicle.raw.go.yutong-mqtt.v1"
 	if got != want {
 		t.Fatalf("topics = %q, want %q", got, want)
 	}
--- a/go/vehicle-gateway/internal/gateway/mqtt_client.go
+++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go
@@ -40,6 +40,7 @@ type MQTTClientConfig struct {
 	Resolver                identity.Resolver
 	Logger                  *slog.Logger
 	Metrics                 *metrics.Registry
+	PublishUnified          bool
 }

 type MQTTClient struct {
@@ -220,12 +221,14 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []
 	if env.ParseStatus == envelope.ParseBadFrame {
 		return
 	}
-	if err := c.cfg.Sink.PublishUnified(messageCtx, env); err != nil {
-		c.recordPublishMetric("unified", "error")
-		c.cfg.Logger.Error("publish mqtt unified failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
-		return
+	if c.cfg.PublishUnified {
+		if err := c.cfg.Sink.PublishUnified(messageCtx, env); err != nil {
+			c.recordPublishMetric("unified", "error")
+			c.cfg.Logger.Error("publish mqtt unified failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
+			return
+		}
+		c.recordPublishMetric("unified", "ok")
 	}
-	c.recordPublishMetric("unified", "ok")
 }

 func (c *MQTTClient) recordFrameMetric(status envelope.ParseStatus) {
--- a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go
+++ b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go
@@ -19,7 +19,7 @@ import (
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
 )

-func TestMQTTClientHandleMessagePublishesRawAndUnified(t *testing.T) {
+func TestMQTTClientHandleMessagePublishesOnlyRawByDefault(t *testing.T) {
 	sink := &recordingSink{}
 	client, err := NewMQTTClient(MQTTClientConfig{
 		EndpointName: "endpoint-a",
@@ -39,7 +39,7 @@ func TestMQTTClientHandleMessagePublishesRawAndUnified(t *testing.T) {
 	  "data":{"METER_SPEED":52.3,"TOTAL_MILEAGE":123456.7}
 	}`))

-	if len(sink.raw) != 1 || len(sink.unified) != 1 {
+	if len(sink.raw) != 1 || len(sink.unified) != 0 {
 		t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified))
 	}
 	if sink.raw[0].Protocol != envelope.ProtocolYutongMQTT || sink.raw[0].VIN != "LTEST000000000001" {
@@ -72,12 +72,14 @@ func TestMQTTClientRecordsMessageMetrics(t *testing.T) {
 	for _, want := range []string{
 		`vehicle_gateway_frames_total{protocol="YUTONG_MQTT",status="OK"} 1`,
 		`vehicle_gateway_publish_total{kind="raw",protocol="YUTONG_MQTT",status="ok"} 1`,
-		`vehicle_gateway_publish_total{kind="unified",protocol="YUTONG_MQTT",status="ok"} 1`,
 	} {
 		if !strings.Contains(text, want) {
 			t.Fatalf("metrics missing %s:\n%s", want, text)
 		}
 	}
+	if strings.Contains(text, `kind="unified"`) {
+		t.Fatalf("unified publish metric should not be recorded by default:\n%s", text)
+	}
 }

 func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) {
@@ -138,11 +140,37 @@ func TestMQTTClientUsesUncancelledMessageContextForReceivedMessage(t *testing.T)
 	if sink.unifiedCtxErr != nil {
 		t.Fatalf("unified publish saw cancelled context: %v", sink.unifiedCtxErr)
 	}
-	if sink.rawCount != 1 || sink.unifiedCount != 1 {
+	if sink.rawCount != 1 || sink.unifiedCount != 0 {
 		t.Fatalf("raw=%d unified=%d", sink.rawCount, sink.unifiedCount)
 	}
 }

+func TestMQTTClientPublishesUnifiedWhenExplicitlyEnabled(t *testing.T) {
+	sink := &recordingSink{}
+	client, err := NewMQTTClient(MQTTClientConfig{
+		EndpointName:   "endpoint-a",
+		Broker:         "tcp://127.0.0.1:1883",
+		ClientID:       "test-client",
+		Topics:         []string{"/ytforward/shln/+"},
+		PublishUnified: true,
+		Sink:           sink,
+		Logger:         slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
+	})
+	if err != nil {
+		t.Fatalf("NewMQTTClient() error = %v", err)
+	}
+
+	client.handleMessage(context.Background(), "/ytforward/shln/dev1", []byte(`{
+	  "device":"LTEST000000000001",
+	  "time":"20260413100000",
+	  "data":{"METER_SPEED":52.3,"TOTAL_MILEAGE":123456.7}
+	}`))
+
+	if len(sink.raw) != 1 || len(sink.unified) != 1 {
+		t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified))
+	}
+}
+
 func TestMQTTClientBuildOptionsLoadsTLSCertificates(t *testing.T) {
 	dir := t.TempDir()
 	caPath, certPath, keyPath := writeTestTLSMaterial(t, dir)
--- a/go/vehicle-gateway/internal/gateway/tcp_server.go
+++ b/go/vehicle-gateway/internal/gateway/tcp_server.go
@@ -41,6 +41,7 @@ type TCPServer struct {
 	readBufferSize int
 	idleTimeout    time.Duration
 	maxConnections int
+	publishUnified bool
 }

 type TCPServerConfig struct {
@@ -52,6 +53,7 @@ type TCPServerConfig struct {
 	ReadBufferSize int
 	IdleTimeout    time.Duration
 	MaxConnections int
+	PublishUnified bool
 }

 const frameOperationTimeout = 30 * time.Second
@@ -96,6 +98,7 @@ func NewTCPServer(cfg TCPServerConfig) (*TCPServer, error) {
 		readBufferSize: cfg.ReadBufferSize,
 		idleTimeout:    cfg.IdleTimeout,
 		maxConnections: cfg.MaxConnections,
+		publishUnified: cfg.PublishUnified,
 	}, nil
 }

@@ -226,12 +229,14 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte,
 	if env.ParseStatus == envelope.ParseBadFrame {
 		return
 	}
-	if err := s.sink.PublishUnified(frameCtx, env); err != nil {
-		s.recordPublishMetric("unified", "error")
-		s.logger.Error("publish unified failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err)
-		return
+	if s.publishUnified {
+		if err := s.sink.PublishUnified(frameCtx, env); err != nil {
+			s.recordPublishMetric("unified", "error")
+			s.logger.Error("publish unified failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err)
+			return
+		}
+		s.recordPublishMetric("unified", "ok")
 	}
-	s.recordPublishMetric("unified", "ok")
 	if s.protocol.Respond == nil {
 		return
 	}
--- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go
+++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go
@@ -16,7 +16,7 @@ import (
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/jt808"
 )

-func TestTCPServerPublishesGoodFrameToRawAndUnified(t *testing.T) {
+func TestTCPServerPublishesGoodFrameOnlyToRawByDefault(t *testing.T) {
 	frame, err := hex.DecodeString("7E020000320133077954250001000000000048000301D2C4C707376139000A00E6004F26063016235701040001900C2504000000000202000030011F31010F867E")
 	if err != nil {
 		t.Fatal(err)
@@ -36,7 +36,7 @@ func TestTCPServerPublishesGoodFrameToRawAndUnified(t *testing.T) {
 	_ = client.Close()
 	<-done

-	if len(sink.raw) != 1 || len(sink.unified) != 1 {
+	if len(sink.raw) != 1 || len(sink.unified) != 0 {
 		t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified))
 	}
 	if sink.raw[0].Phone != "13307795425" {
@@ -80,12 +80,14 @@ func TestTCPServerRecordsFrameMetrics(t *testing.T) {
 	for _, want := range []string{
 		`vehicle_gateway_frames_total{protocol="JT808",status="OK"} 1`,
 		`vehicle_gateway_publish_total{kind="raw",protocol="JT808",status="ok"} 1`,
-		`vehicle_gateway_publish_total{kind="unified",protocol="JT808",status="ok"} 1`,
 	} {
 		if !strings.Contains(text, want) {
 			t.Fatalf("metrics missing %s:\n%s", want, text)
 		}
 	}
+	if strings.Contains(text, `kind="unified"`) {
+		t.Fatalf("unified publish metric should not be recorded by default:\n%s", text)
+	}
 }

 func TestTCPServerRecordsActiveConnectionGauge(t *testing.T) {
@@ -171,8 +173,8 @@ func TestTCPServerWritesProtocolResponseAfterPublish(t *testing.T) {
 		Extract:  gb32960.ExtractFrames,
 		Parse:    gb32960.ParseFrame,
 		Respond: func(_ []byte, env envelope.FrameEnvelope) ([]byte, bool, error) {
-			if len(sink.unified) != 1 || sink.unified[0].EventID != env.EventID {
-				t.Fatalf("response built before publish: raw=%d unified=%d", len(sink.raw), len(sink.unified))
+			if len(sink.raw) != 1 || sink.raw[0].EventID != env.EventID {
+				t.Fatalf("response built before raw publish: raw=%d unified=%d", len(sink.raw), len(sink.unified))
 			}
 			return []byte("ACK"), true, nil
 		},
@@ -235,11 +237,38 @@ func TestTCPServerUsesUncancelledFrameContextForAlreadyReadFrame(t *testing.T) {
 	if sink.unifiedCtxErr != nil {
 		t.Fatalf("unified publish saw cancelled context: %v", sink.unifiedCtxErr)
 	}
-	if sink.rawCount != 1 || sink.unifiedCount != 1 {
+	if sink.rawCount != 1 || sink.unifiedCount != 0 {
 		t.Fatalf("raw=%d unified=%d", sink.rawCount, sink.unifiedCount)
 	}
 }

+func TestTCPServerPublishesUnifiedWhenExplicitlyEnabled(t *testing.T) {
+	sink := &recordingSink{}
+	server := newTestServer(t, TCPProtocol{
+		Protocol: envelope.ProtocolJT808,
+		Addr:     ":0",
+		Extract:  jt808.ExtractFrames,
+		Parse: func(_ []byte, receivedAtMS int64, sourceEndpoint string) (envelope.FrameEnvelope, error) {
+			return envelope.FrameEnvelope{
+				Protocol:       envelope.ProtocolJT808,
+				MessageID:      "0x0200",
+				Phone:          "13307795425",
+				SourceEndpoint: sourceEndpoint,
+				ReceivedAtMS:   receivedAtMS,
+				EventTimeMS:    receivedAtMS,
+				ParseStatus:    envelope.ParseOK,
+			}, nil
+		},
+	}, sink)
+	server.publishUnified = true
+
+	server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808")
+
+	if len(sink.raw) != 1 || len(sink.unified) != 1 {
+		t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified))
+	}
+}
+
 type contextCheckingResolver struct {
 	ctxErr error
 }