lingniu-vehicle-ingest/go/vehicle-gateway/internal/gateway/tcp_server_test.go

package gateway

import (
	"context"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"os"
	"strings"
	"syscall"
	"testing"
	"time"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/authentication"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/gb32960"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/jt808"
)

func TestEnrichConnectionPlatformPromotesSourceMetadata(t *testing.T) {
	env := envelope.FrameEnvelope{
		Protocol:    envelope.ProtocolGB32960,
		MessageID:   "0x02",
		ParseStatus: envelope.ParseOK,
		Fields:      map[string]any{},
		Parsed:      map[string]any{},
	}
	state := &connectionState{platformName: "Hyundai"}

	enrichConnectionPlatform(&env, state)

	if env.PlatformName != "Hyundai" || env.SourceCode != "Hyundai" || env.SourceKind != "PLATFORM" {
		t.Fatalf("source metadata = code:%q platform:%q kind:%q", env.SourceCode, env.PlatformName, env.SourceKind)
	}
	if got := fmt.Sprint(env.Fields["platform_account"]); got != "Hyundai" {
		t.Fatalf("platform_account = %q", got)
	}
	if got := fmt.Sprint(env.Parsed["platform_name"]); got != "Hyundai" {
		t.Fatalf("parsed platform_name = %q", got)
	}
}

func TestTCPServerPublishesGoodFrameToRawAndFieldsByDefault(t *testing.T) {
	frame, err := hex.DecodeString("7E020000320133077954250001000000000048000301D2C4C707376139000A00E6004F26063016235701040001900C2504000000000202000030011F31010F867E")
	if err != nil {
		t.Fatal(err)
	}
	sink := &recordingSink{}
	server := newTestServer(t, TCPProtocol{
		Protocol: envelope.ProtocolJT808,
		Addr:     ":0",
		Extract:  jt808.ExtractFrames,
		Parse:    jt808.ParseFrame,
	}, sink)
	server.resolver = vinResolver{vin: "LNBVIN00000000001"}

	client, done := runPipe(t, server)
	if _, err := client.Write(frame); err != nil {
		t.Fatalf("client.Write() error = %v", err)
	}
	_ = client.Close()
	<-done

	if len(sink.raw) != 1 || len(sink.unified) != 0 {
		t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified))
	}
	if len(sink.fields) != 1 {
		t.Fatalf("fields=%d, want 1", len(sink.fields))
	}
	if sink.raw[0].Phone != "13307795425" {
		t.Fatalf("phone = %q", sink.raw[0].Phone)
	}
	if sink.raw[0].EventKind != envelope.EventKindRaw {
		t.Fatalf("raw event kind = %q, want %q", sink.raw[0].EventKind, envelope.EventKindRaw)
	}
	if len(sink.raw[0].Fields) != 0 {
		t.Fatalf("canonical raw must not carry bare standardized fields: %#v", sink.raw[0].Fields)
	}
	if got := sink.raw[0].ParsedFields["jt808.location.total_mileage_km"]; got == nil {
		t.Fatalf("raw parsed fields missing total mileage: %#v", sink.raw[0].ParsedFields)
	}
	if got, want := sink.fields[0].Fields["jt808.location.total_mileage_km"], sink.raw[0].ParsedFields["jt808.location.total_mileage_km"]; got != want {
		t.Fatalf("fields event should reuse raw parsed field, got %#v want %#v", got, want)
	}
	if sink.fields[0].EventKind != envelope.EventKindFields {
		t.Fatalf("fields event kind = %q, want %q", sink.fields[0].EventKind, envelope.EventKindFields)
	}
	if got, want := sink.fields[0].SourceEventID, sink.raw[0].StableEventID(); got != want {
		t.Fatalf("fields source event id = %#v, want %s", got, want)
	}
	if sink.fields[0].FieldMapping == "" {
		t.Fatal("fields event should expose field mapping version")
	}
	if len(sink.fields[0].Parsed) != 0 || len(sink.fields[0].ParsedFields) != 0 {
		t.Fatalf("fields envelope should not duplicate parsed payload: parsed=%#v parsed_fields=%#v", sink.fields[0].Parsed, sink.fields[0].ParsedFields)
	}
}

func TestTCPServerRecordsFieldsPublishedMetric(t *testing.T) {
	frame, err := hex.DecodeString("7E020000320133077954250001000000000048000301D2C4C707376139000A00E6004F26063016235701040001900C2504000000000202000030011F31010F867E")
	if err != nil {
		t.Fatal(err)
	}
	sink := &recordingSink{}
	registry := metrics.NewRegistry()
	server := newTestServer(t, TCPProtocol{
		Protocol: envelope.ProtocolJT808,
		Addr:     ":0",
		Extract:  jt808.ExtractFrames,
		Parse:    jt808.ParseFrame,
	}, sink)
	server.metrics = registry
	server.resolver = vinResolver{vin: "LNBVIN00000000001"}

	client, done := runPipe(t, server)
	if _, err := client.Write(frame); err != nil {
		t.Fatalf("client.Write() error = %v", err)
	}
	_ = client.Close()
	<-done

	text := registry.Render()
	for _, want := range []string{
		`vehicle_gateway_fields_total{protocol="JT808",status="published"} 1`,
		`vehicle_gateway_fields_count{protocol="JT808",status="published"} `,
		`vehicle_gateway_fields_count_histogram_count{protocol="JT808",status="published"} 1`,
		`vehicle_gateway_publish_total{kind="fields",protocol="JT808",status="ok"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("metrics missing %s:\n%s", want, text)
		}
	}
}

func TestTCPServerRecordsFrameMetrics(t *testing.T) {
	registry := metrics.NewRegistry()
	server, err := NewTCPServer(TCPServerConfig{
		Protocol: 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:    &recordingSink{},
		Logger:  slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
		Metrics: registry,
	})
	if err != nil {
		t.Fatalf("NewTCPServer() error = %v", err)
	}

	server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808", &connectionState{})

	text := registry.Render()
	for _, want := range []string{
		`vehicle_gateway_frames_total{protocol="JT808",status="OK"} 1`,
		`vehicle_gateway_last_frame_unix_seconds{protocol="JT808",status="OK"} `,
		`vehicle_gateway_identity_total{protocol="JT808",status="unresolved"} 1`,
		`vehicle_gateway_identity_issues_total{message_id="0x0200",protocol="JT808",reason="no_binding",status="unresolved"} 1`,
		`vehicle_gateway_identity_duration_ms{protocol="JT808",status="unresolved"}`,
		`vehicle_gateway_identity_duration_ms_histogram_bucket{le="+Inf",protocol="JT808",status="unresolved"} 1`,
		`vehicle_gateway_identity_duration_ms_histogram_count{protocol="JT808",status="unresolved"} 1`,
		`vehicle_gateway_identity_duration_ms_histogram_sum{protocol="JT808",status="unresolved"}`,
		`vehicle_gateway_publish_total{kind="raw",protocol="JT808",status="ok"} 1`,
		`vehicle_gateway_last_publish_unix_seconds{kind="raw",protocol="JT808",status="ok"} `,
		`vehicle_gateway_fields_total{protocol="JT808",status="skipped_non_realtime"} 1`,
		`vehicle_gateway_frame_duration_ms{protocol="JT808",status="OK"}`,
		`vehicle_gateway_frame_duration_ms_histogram_bucket{le="+Inf",protocol="JT808",status="OK"} 1`,
		`vehicle_gateway_frame_duration_ms_histogram_count{protocol="JT808",status="OK"} 1`,
		`vehicle_gateway_frame_duration_ms_histogram_sum{protocol="JT808",status="OK"}`,
	} {
		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 TestTCPServerRecordsNonRealtimeFieldsSkipMetric(t *testing.T) {
	registry := metrics.NewRegistry()
	sink := &recordingSink{}
	server := newTestServer(t, TCPProtocol{
		Protocol: envelope.ProtocolGB32960,
		Addr:     ":0",
		Extract:  gb32960.ExtractFrames,
		Parse:    gb32960.ParseFrame,
	}, sink)
	server.metrics = registry

	server.handleFrame(context.Background(), nil, buildGBFrame(0x07, 0xfe, "LNBSCB3D4R1234567", nil), "127.0.0.1:32960", &connectionState{})

	if len(sink.raw) != 1 {
		t.Fatalf("raw=%d, want 1", len(sink.raw))
	}
	if len(sink.fields) != 0 {
		t.Fatalf("fields=%d, want 0", len(sink.fields))
	}
	if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_fields_total{protocol="GB32960",status="skipped_non_realtime"} 1`) {
		t.Fatalf("non realtime fields skip metric missing:\n%s", text)
	}
}

func TestTCPServerSkipsIdentityForGB32960PlatformControlFrame(t *testing.T) {
	registry := metrics.NewRegistry()
	sink := &recordingSink{}
	resolver := &countingResolver{vin: "LNBVIN00000000001"}
	server := newTestServer(t, TCPProtocol{
		Protocol: envelope.ProtocolGB32960,
		Addr:     ":0",
		Extract:  gb32960.ExtractFrames,
		Parse:    gb32960.ParseFrame,
	}, sink)
	server.metrics = registry
	server.resolver = resolver

	server.handleFrame(context.Background(), nil, buildGBFrame(0x05, 0xfe, "PLATFORMLOGIN0001", nil), "127.0.0.1:32960", &connectionState{})

	if resolver.calls != 0 {
		t.Fatalf("resolver calls = %d, want 0 for platform control frame", resolver.calls)
	}
	if len(sink.raw) != 1 {
		t.Fatalf("raw=%d, want 1", len(sink.raw))
	}
	text := registry.Render()
	if strings.Contains(text, "vehicle_gateway_identity_total") {
		t.Fatalf("identity metric should not be recorded for platform control frame:\n%s", text)
	}
	if !strings.Contains(text, `vehicle_gateway_identity_skips_total{protocol="GB32960",reason="non_vehicle_frame"} 1`) {
		t.Fatalf("identity skip metric missing:\n%s", text)
	}
	if !strings.Contains(text, `vehicle_gateway_fields_total{protocol="GB32960",status="skipped_non_realtime"} 1`) {
		t.Fatalf("non realtime fields skip metric missing:\n%s", text)
	}
}

func TestTCPServerAuthenticatesAndRedactsGB32960PlatformLogin(t *testing.T) {
	registry := metrics.NewRegistry()
	sink := &recordingSink{}
	server := newTestServer(t, TCPProtocol{
		Protocol:     envelope.ProtocolGB32960,
		Addr:         ":0",
		Extract:      gb32960.ExtractFrames,
		Parse:        gb32960.ParseFrame,
		Authenticate: authentication.NewGB32960PlatformAuthenticator(authentication.ModeObserve, map[string][]string{"platform-a": {"secret-a"}}),
	}, sink)
	server.metrics = registry
	body := []byte{0x1a, 0x07, 0x0d, 0x14, 0x00, 0x00, 0x00, 0x01}
	body = append(body, fixedASCIIBytes("platform-a", 12)...)
	body = append(body, fixedASCIIBytes("secret-a", 20)...)
	body = append(body, 0x01)

	server.handleFrame(context.Background(), nil, buildGBFrame(0x05, 0xfe, "12345678901234567", body), "127.0.0.1:32960", &connectionState{})

	if len(sink.raw) != 1 {
		t.Fatalf("raw=%d, want 1", len(sink.raw))
	}
	raw := sink.raw[0]
	if raw.AuthenticationMode != "observe" || raw.AuthenticationStatus != authentication.StatusAccepted || raw.AuthenticationEnforced {
		t.Fatalf("authentication metadata = mode:%q status:%q enforced:%v", raw.AuthenticationMode, raw.AuthenticationStatus, raw.AuthenticationEnforced)
	}
	if _, exists := raw.ParsedFields["gb32960.platform_login.password"]; exists {
		t.Fatalf("plaintext password leaked into parsed fields: %#v", raw.ParsedFields)
	}
	if got := raw.ParsedFields["gb32960.platform_login.password_present"]; got != "true" {
		t.Fatalf("password presence marker = %#v", got)
	}
	if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_authentication_total{mode="observe",protocol="GB32960",source="configured",status="accepted"} 1`) {
		t.Fatalf("authentication metric missing:\n%s", text)
	}
}

func TestTCPServerResolvesIdentityForGB32960RealtimeFrame(t *testing.T) {
	registry := metrics.NewRegistry()
	sink := &recordingSink{}
	resolver := &countingResolver{vin: "LNBVIN00000000001"}
	server := newTestServer(t, TCPProtocol{
		Protocol: envelope.ProtocolGB32960,
		Addr:     ":0",
		Extract:  gb32960.ExtractFrames,
		Parse:    gb32960.ParseFrame,
	}, sink)
	server.metrics = registry
	server.resolver = resolver

	server.handleFrame(context.Background(), nil, buildGBFrame(0x02, 0xfe, "LNBSCB3D4R1234567", nil), "127.0.0.1:32960", &connectionState{})

	if resolver.calls != 1 {
		t.Fatalf("resolver calls = %d, want 1 for realtime frame", resolver.calls)
	}
	if len(sink.raw) != 1 {
		t.Fatalf("raw=%d, want 1", len(sink.raw))
	}
	if sink.raw[0].VIN != "LNBVIN00000000001" {
		t.Fatalf("raw vin = %q, want resolved vin", sink.raw[0].VIN)
	}
	if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_identity_total{protocol="GB32960",status="resolved"} 1`) {
		t.Fatalf("identity resolved metric missing:\n%s", text)
	}
}

func TestTCPServerRecordsIdentityTimeoutMetrics(t *testing.T) {
	registry := metrics.NewRegistry()
	server, err := NewTCPServer(TCPServerConfig{
		Protocol: 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:     &recordingSink{},
		Resolver: errorResolver{err: context.DeadlineExceeded},
		Logger:   slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
		Metrics:  registry,
	})
	if err != nil {
		t.Fatalf("NewTCPServer() error = %v", err)
	}

	server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808", &connectionState{})

	text := registry.Render()
	for _, want := range []string{
		`vehicle_gateway_identity_total{protocol="JT808",status="timeout"} 1`,
		`vehicle_gateway_identity_issues_total{message_id="0x0200",protocol="JT808",reason="timeout",status="timeout"} 1`,
		`vehicle_gateway_identity_duration_ms{protocol="JT808",status="timeout"}`,
		`vehicle_gateway_identity_duration_ms_histogram_bucket{le="+Inf",protocol="JT808",status="timeout"} 1`,
		`vehicle_gateway_identity_duration_ms_histogram_count{protocol="JT808",status="timeout"} 1`,
		`vehicle_gateway_frames_total{protocol="JT808",status="PARTIAL"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("metrics missing %s:\n%s", want, text)
		}
	}
}

func TestTCPServerPreservesResolvedEnvelopeWhenIdentitySideEffectFails(t *testing.T) {
	registry := metrics.NewRegistry()
	sink := &recordingSink{}
	wantErr := errors.New("registration upsert failed")
	server, err := NewTCPServer(TCPServerConfig{
		Protocol: 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,
					ParsedFields: map[string]any{
						"jt808.location.total_mileage_km": "10241.2",
					},
					ParseStatus: envelope.ParseOK,
				}, nil
			},
		},
		Sink:     sink,
		Resolver: resolvedErrorResolver{vin: "LNBVIN00000000001", err: wantErr},
		Logger:   slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
		Metrics:  registry,
	})
	if err != nil {
		t.Fatalf("NewTCPServer() error = %v", err)
	}

	server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808", &connectionState{})

	if len(sink.raw) != 1 {
		t.Fatalf("raw count = %d, want 1", len(sink.raw))
	}
	if sink.raw[0].VIN != "LNBVIN00000000001" {
		t.Fatalf("raw vin = %q, want resolved vin", sink.raw[0].VIN)
	}
	if sink.raw[0].ParseStatus != envelope.ParsePartial {
		t.Fatalf("parse status = %q, want PARTIAL", sink.raw[0].ParseStatus)
	}
	if len(sink.raw[0].Parsed) != 0 {
		t.Fatalf("canonical raw should not duplicate parsed tree: %#v", sink.raw[0].Parsed)
	}
	if got := sink.raw[0].ParsedFields["jt808.location.total_mileage_km"]; got != "10241.2" {
		t.Fatalf("protocol field lost after identity side-effect failure: %#v", got)
	}
	for field := range sink.raw[0].ParsedFields {
		if strings.HasPrefix(field, "jt808.identity.") {
			t.Fatalf("derived identity annotation leaked into protocol fields: %s", field)
		}
	}
	if len(sink.fields) != 1 || sink.fields[0].VIN != "LNBVIN00000000001" {
		t.Fatalf("fields should preserve resolved vin, fields=%#v", sink.fields)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_gateway_identity_total{protocol="JT808",status="error"} 1`,
		`vehicle_gateway_identity_issues_total{message_id="0x0200",protocol="JT808",reason="resolver_error",status="error"} 1`,
		`vehicle_gateway_frames_total{protocol="JT808",status="PARTIAL"} 1`,
		`vehicle_gateway_fields_total{protocol="JT808",status="published"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("metric missing %s:\n%s", want, text)
		}
	}
}

func TestTCPServerRecordsActiveConnectionGauge(t *testing.T) {
	registry := metrics.NewRegistry()
	server, err := NewTCPServer(TCPServerConfig{
		Protocol: TCPProtocol{
			Protocol: envelope.ProtocolJT808,
			Addr:     ":0",
			Extract:  jt808.ExtractFrames,
			Parse:    jt808.ParseFrame,
		},
		Sink:        &recordingSink{},
		Logger:      slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
		Metrics:     registry,
		IdleTimeout: time.Minute,
	})
	if err != nil {
		t.Fatalf("NewTCPServer() error = %v", err)
	}
	client, srv := net.Pipe()
	opened := make(chan struct{})
	done := make(chan struct{})
	go func() {
		close(opened)
		server.handleConnection(context.Background(), srv)
		close(done)
	}()
	<-opened
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if strings.Contains(registry.Render(), `vehicle_gateway_active_connections{protocol="JT808"} 1`) {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_active_connections{protocol="JT808"} 1`) {
		t.Fatalf("active connection gauge missing while open:\n%s", text)
	}

	_ = client.Close()
	<-done

	if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_active_connections{protocol="JT808"} 0`) {
		t.Fatalf("active connection gauge should return to zero:\n%s", text)
	}
}

func TestTCPServerClosesActiveConnectionWhenContextCancelled(t *testing.T) {
	registry := metrics.NewRegistry()
	server, err := NewTCPServer(TCPServerConfig{
		Protocol: TCPProtocol{
			Protocol: envelope.ProtocolJT808,
			Addr:     ":0",
			Extract:  jt808.ExtractFrames,
			Parse:    jt808.ParseFrame,
		},
		Sink:        &recordingSink{},
		Logger:      slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
		Metrics:     registry,
		IdleTimeout: time.Minute,
	})
	if err != nil {
		t.Fatalf("NewTCPServer() error = %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	client, srv := net.Pipe()
	done := make(chan struct{})
	go func() {
		server.handleConnection(ctx, srv)
		close(done)
	}()
	waitForMetric(t, registry, `vehicle_gateway_active_connections{protocol="JT808"} 1`)

	cancel()
	server.closeActiveConnections()
	defer client.Close()
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("connection handler did not exit after context cancellation")
	}
	if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_connection_closes_total{protocol="JT808",reason="context_cancelled"} 1`) {
		t.Fatalf("context cancellation close metric missing:\n%s", text)
	}
}

func TestTCPServerRecordsConnectionRejectionMetric(t *testing.T) {
	registry := metrics.NewRegistry()
	server, err := NewTCPServer(TCPServerConfig{
		Protocol: TCPProtocol{
			Protocol: envelope.ProtocolJT808,
			Addr:     ":0",
			Extract:  jt808.ExtractFrames,
			Parse:    jt808.ParseFrame,
		},
		Sink:    &recordingSink{},
		Logger:  slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
		Metrics: registry,
	})
	if err != nil {
		t.Fatalf("NewTCPServer() error = %v", err)
	}

	server.recordConnectionRejection("max_connections")

	if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_connection_rejections_total{protocol="JT808",reason="max_connections"} 1`) {
		t.Fatalf("connection rejection metric missing:\n%s", text)
	}
}

func TestTCPServerRecordsConnectionCloseReasonMetric(t *testing.T) {
	registry := metrics.NewRegistry()
	server, err := NewTCPServer(TCPServerConfig{
		Protocol: TCPProtocol{
			Protocol: envelope.ProtocolJT808,
			Addr:     ":0",
			Extract:  jt808.ExtractFrames,
			Parse:    jt808.ParseFrame,
		},
		Sink:    &recordingSink{},
		Logger:  slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
		Metrics: registry,
	})
	if err != nil {
		t.Fatalf("NewTCPServer() error = %v", err)
	}

	server.recordConnectionClose("read_timeout")

	if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_connection_closes_total{protocol="JT808",reason="read_timeout"} 1`) {
		t.Fatalf("connection close metric missing:\n%s", text)
	}
}

func TestTCPServerConnectionLifecycleLogsStayDebug(t *testing.T) {
	source, err := os.ReadFile("tcp_server.go")
	if err != nil {
		t.Fatal(err)
	}
	text := string(source)
	for _, forbidden := range []string{
		`Info("tcp connection opened"`,
		`Info("tcp connection closed"`,
		`Warn("tcp connection idle timeout"`,
		`Warn("tcp connection closed by peer"`,
	} {
		if strings.Contains(text, forbidden) {
			t.Fatalf("high-volume connection lifecycle log should not be info/warn: %s", forbidden)
		}
	}
	for _, want := range []string{
		`Debug("tcp connection opened"`,
		`Debug("tcp connection closed"`,
		`Debug("tcp connection idle timeout"`,
		`Debug("tcp connection closed by peer"`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("connection lifecycle log should remain debug: %s", want)
		}
	}
}

func TestIsRoutineTCPReadCloseClassifiesRemoteCloseNoise(t *testing.T) {
	tests := []struct {
		name string
		err  error
		want bool
	}{
		{name: "eof", err: io.EOF, want: true},
		{name: "net closed", err: net.ErrClosed, want: true},
		{name: "reset wrapped", err: fmt.Errorf("read tcp: %w", syscall.ECONNRESET), want: true},
		{name: "pipe wrapped", err: fmt.Errorf("write tcp: %w", syscall.EPIPE), want: true},
		{name: "reset text", err: errors.New("read tcp 172.17.111.55:808->117.132.196.176:22187: read: connection reset by peer"), want: true},
		{name: "unexpected", err: errors.New("checksum parser exploded"), want: false},
		{name: "nil", err: nil, want: false},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := isRoutineTCPReadClose(tt.err); got != tt.want {
				t.Fatalf("isRoutineTCPReadClose(%v) = %v, want %v", tt.err, got, tt.want)
			}
		})
	}
}

func TestTCPServerPublishesBadFrameOnlyToRaw(t *testing.T) {
	good := buildGBFrame(0x02, 0xfe, "LNBSCB3D4R1234567", nil)
	good[len(good)-1] ^= 0xff
	sink := &recordingSink{}
	registry := metrics.NewRegistry()
	server := newTestServer(t, TCPProtocol{
		Protocol: envelope.ProtocolGB32960,
		Addr:     ":0",
		Extract:  gb32960.ExtractFrames,
		Parse:    gb32960.ParseFrame,
	}, sink)
	server.metrics = registry

	client, done := runPipe(t, server)
	if _, err := client.Write(good); err != nil {
		t.Fatalf("client.Write() error = %v", err)
	}
	_ = client.Close()
	<-done

	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].ParseStatus != envelope.ParseBadFrame {
		t.Fatalf("parse status = %q", sink.raw[0].ParseStatus)
	}
	if sink.raw[0].ParseError == "" {
		t.Fatal("parse error should be recorded")
	}
	if text := registry.Render(); !strings.Contains(text, `vehicle_gateway_parse_errors_total{protocol="GB32960",reason="checksum"} 1`) {
		t.Fatalf("parse error metric missing:\n%s", text)
	}
}

func TestTCPServerAnnotatesUnresolvedIdentity(t *testing.T) {
	sink := &recordingSink{}
	server := newTestServer(t, TCPProtocol{
		Protocol: envelope.ProtocolJT808,
		Addr:     ":0",
		Extract: func(raw []byte) ([][]byte, []byte, error) {
			return [][]byte{raw}, nil, nil
		},
		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.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808", &connectionState{})

	if len(sink.raw) != 1 {
		t.Fatalf("raw count = %d", len(sink.raw))
	}
	if len(sink.raw[0].Parsed) != 0 || len(sink.raw[0].ParsedFields) != 0 {
		t.Fatalf("unresolved derived metadata must not enter canonical protocol payload: parsed=%#v fields=%#v", sink.raw[0].Parsed, sink.raw[0].ParsedFields)
	}
}

func TestTCPServerWritesProtocolResponseAfterPublish(t *testing.T) {
	frame := buildGBFrame(0x07, 0xfe, "LNBSCB3D4R1234567", nil)
	registry := metrics.NewRegistry()
	sink := &recordingSink{}
	server := newTestServer(t, TCPProtocol{
		Protocol: envelope.ProtocolGB32960,
		Addr:     ":0",
		Extract:  gb32960.ExtractFrames,
		Parse:    gb32960.ParseFrame,
		Respond: func(_ []byte, env envelope.FrameEnvelope) ([]byte, bool, error) {
			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
		},
	}, sink)
	server.metrics = registry

	client, done := runPipe(t, server)
	if _, err := client.Write(frame); err != nil {
		t.Fatalf("client.Write() error = %v", err)
	}
	buf := make([]byte, 3)
	if _, err := io.ReadFull(client, buf); err != nil {
		t.Fatalf("read response error = %v", err)
	}
	if string(buf) != "ACK" {
		t.Fatalf("response = %q", string(buf))
	}
	_ = client.Close()
	<-done
	text := registry.Render()
	for _, want := range []string{
		`vehicle_gateway_response_total{message_id="0x07",protocol="GB32960",status="ok"} 1`,
		`vehicle_gateway_last_response_unix_seconds{message_id="0x07",protocol="GB32960",status="ok"} `,
		`vehicle_gateway_response_duration_ms_histogram_count{message_id="0x07",protocol="GB32960",status="ok"} 1`,
		`vehicle_gateway_response_e2e_recent_p99_ms{protocol="GB32960"} `,
		`vehicle_gateway_response_e2e_recent_samples{protocol="GB32960"} `,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("response metric missing %s:\n%s", want, text)
		}
	}
}

func TestTCPServerWritesProtocolResponseWhenFieldsPublishFails(t *testing.T) {
	registry := metrics.NewRegistry()
	sink := &recordingSink{fieldsErr: errors.New("fields queue full")}
	server := newTestServer(t, TCPProtocol{
		Protocol: envelope.ProtocolJT808,
		Addr:     ":0",
		Extract: func(raw []byte) ([][]byte, []byte, error) {
			return [][]byte{raw}, nil, nil
		},
		Parse: func(_ []byte, receivedAtMS int64, sourceEndpoint string) (envelope.FrameEnvelope, error) {
			return envelope.FrameEnvelope{
				Protocol:       envelope.ProtocolJT808,
				MessageID:      "0x0200",
				VIN:            "LNBVIN00000000001",
				Phone:          "13307795425",
				SourceEndpoint: sourceEndpoint,
				ReceivedAtMS:   receivedAtMS,
				EventTimeMS:    receivedAtMS,
				ParsedFields: map[string]any{
					"jt808.location.total_mileage_km": "10241.2",
				},
				ParseStatus: envelope.ParseOK,
			}, nil
		},
		Respond: func(_ []byte, env envelope.FrameEnvelope) ([]byte, bool, error) {
			if len(sink.raw) != 1 || sink.raw[0].EventID != env.EventID {
				t.Fatalf("response built before raw publish: raw=%d", len(sink.raw))
			}
			return []byte("ACK"), true, nil
		},
	}, sink)
	server.metrics = registry

	client, done := runPipe(t, server)
	if _, err := client.Write([]byte{0x01}); err != nil {
		t.Fatalf("client.Write() error = %v", err)
	}
	buf := make([]byte, 3)
	if _, err := io.ReadFull(client, buf); err != nil {
		t.Fatalf("read response error = %v", err)
	}
	if string(buf) != "ACK" {
		t.Fatalf("response = %q", string(buf))
	}
	_ = client.Close()
	<-done
	if len(sink.fields) != 1 {
		t.Fatalf("fields publish attempts = %d, want 1", len(sink.fields))
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_gateway_response_total{message_id="0x0200",protocol="JT808",status="ok"} 1`,
		`vehicle_gateway_fields_total{protocol="JT808",status="publish_error"} 1`,
		`vehicle_gateway_publish_total{kind="fields",protocol="JT808",status="error"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("metric missing %s:\n%s", want, text)
		}
	}
}

func TestTCPServerUsesUncancelledFrameContextForAlreadyReadFrame(t *testing.T) {
	parent, cancel := context.WithCancel(context.Background())
	cancel()

	resolver := &contextCheckingResolver{}
	sink := &contextCheckingSink{}
	server, err := NewTCPServer(TCPServerConfig{
		Protocol: 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:     sink,
		Resolver: resolver,
		Logger:   slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
	})
	if err != nil {
		t.Fatalf("NewTCPServer() error = %v", err)
	}

	server.handleFrame(parent, nil, []byte{0x01}, "127.0.0.1:808", &connectionState{})

	if resolver.ctxErr != nil {
		t.Fatalf("resolver saw cancelled context: %v", resolver.ctxErr)
	}
	if sink.rawCtxErr != nil {
		t.Fatalf("raw publish saw cancelled context: %v", sink.rawCtxErr)
	}
	if sink.unifiedCtxErr != nil {
		t.Fatalf("unified publish saw cancelled context: %v", sink.unifiedCtxErr)
	}
	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", &connectionState{})

	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
}

func (r *contextCheckingResolver) Resolve(ctx context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) {
	r.ctxErr = ctx.Err()
	return env, r.ctxErr
}

type vinResolver struct {
	vin string
}

func (r vinResolver) Resolve(_ context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) {
	env.VIN = r.vin
	return env, nil
}

type countingResolver struct {
	calls int
	vin   string
}

func (r *countingResolver) Resolve(_ context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) {
	r.calls++
	env.VIN = r.vin
	return env, nil
}

type errorResolver struct {
	err error
}

func (r errorResolver) Resolve(_ context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) {
	return env, r.err
}

type resolvedErrorResolver struct {
	vin string
	err error
}

func (r resolvedErrorResolver) Resolve(_ context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) {
	env.VIN = r.vin
	if env.Parsed == nil {
		env.Parsed = map[string]any{}
	}
	env.Parsed["identity"] = map[string]any{"resolved": true, "source": "test"}
	env.EventID = env.StableEventID()
	return env, r.err
}

type contextCheckingSink struct {
	rawCtxErr     error
	unifiedCtxErr error
	fieldsCtxErr  error
	rawCount      int
	unifiedCount  int
	fieldsCount   int
}

func (s *contextCheckingSink) PublishRaw(ctx context.Context, _ envelope.FrameEnvelope) error {
	s.rawCtxErr = ctx.Err()
	s.rawCount++
	return s.rawCtxErr
}

func (s *contextCheckingSink) PublishUnified(ctx context.Context, _ envelope.FrameEnvelope) error {
	s.unifiedCtxErr = ctx.Err()
	s.unifiedCount++
	return s.unifiedCtxErr
}

func (s *contextCheckingSink) PublishFields(ctx context.Context, _ envelope.FrameEnvelope) error {
	s.fieldsCtxErr = ctx.Err()
	s.fieldsCount++
	return s.fieldsCtxErr
}

func (s *contextCheckingSink) Close() error {
	return nil
}

func newTestServer(t *testing.T, protocol TCPProtocol, sink *recordingSink) *TCPServer {
	t.Helper()
	server, err := NewTCPServer(TCPServerConfig{
		Protocol:       protocol,
		Sink:           sink,
		Logger:         slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
		ReadBufferSize: 1024,
		IdleTimeout:    time.Second,
		MaxConnections: 1,
	})
	if err != nil {
		t.Fatalf("NewTCPServer() error = %v", err)
	}
	return server
}

func runPipe(t *testing.T, server *TCPServer) (net.Conn, <-chan struct{}) {
	t.Helper()
	client, srv := net.Pipe()
	done := make(chan struct{})
	go func() {
		defer close(done)
		server.handleConnection(context.Background(), srv)
	}()
	return client, done
}

func waitForMetric(t *testing.T, registry *metrics.Registry, want string) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if strings.Contains(registry.Render(), want) {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatalf("metric missing %s:\n%s", want, registry.Render())
}

type recordingSink struct {
	raw        []envelope.FrameEnvelope
	unified    []envelope.FrameEnvelope
	fields     []envelope.FrameEnvelope
	rawErr     error
	unifiedErr error
	fieldsErr  error
}

func (s *recordingSink) PublishRaw(_ context.Context, env envelope.FrameEnvelope) error {
	s.raw = append(s.raw, env)
	return s.rawErr
}

func (s *recordingSink) PublishUnified(_ context.Context, env envelope.FrameEnvelope) error {
	s.unified = append(s.unified, env)
	return s.unifiedErr
}

func (s *recordingSink) PublishFields(_ context.Context, env envelope.FrameEnvelope) error {
	s.fields = append(s.fields, env)
	return s.fieldsErr
}

func (s *recordingSink) Close() error {
	return nil
}

type testWriter struct {
	t *testing.T
}

func (w testWriter) Write(p []byte) (int, error) {
	w.t.Log(string(p))
	return len(p), nil
}

func buildGBFrame(command byte, response byte, vin string, body []byte) []byte {
	frame := []byte{'#', '#', command, response}
	vinBytes := []byte(vin)
	if len(vinBytes) < 17 {
		vinBytes = append(vinBytes, make([]byte, 17-len(vinBytes))...)
	}
	frame = append(frame, vinBytes[:17]...)
	frame = append(frame, 0x01, byte(len(body)>>8), byte(len(body)))
	frame = append(frame, body...)
	var bcc byte
	for _, value := range frame[2:] {
		bcc ^= value
	}
	return append(frame, bcc)
}

func fixedASCIIBytes(value string, size int) []byte {
	out := make([]byte, size)
	copy(out, []byte(value))
	return out
}