package realtime

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"encoding/json"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)

func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.EnsureSchema(context.Background()); err != nil {
		t.Fatalf("EnsureSchema() error = %v", err)
	}
	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:       envelope.ProtocolGB32960,
		MessageID:      "0x02",
		VIN:            "VIN001",
		Phone:          "13307795425",
		DeviceID:       "iccid-1",
		SourceEndpoint: "1.2.3.4:32960",
		EventTimeMS:    1782918600000,
		ReceivedAtMS:   1782918601000,
		Parsed: map[string]any{
			"data_units": []any{
				map[string]any{"name": "vehicle", "type": "0x01", "value": map[string]any{"soc_percent": 90}},
				map[string]any{"name": "gd_fc_vendor_tlv", "type": "vendor", "value": map[string]any{"foo": "bar"}},
			},
		},
		ParsedFields: map[string]any{
			"gb32960.vehicle.soc_percent":  "90",
			"gb32960.gd_fc_vendor_tlv.foo": "bar",
		},
		Fields: map[string]any{envelope.FieldSOCPercent: 90},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}

	if len(exec.calls) != 19 {
		t.Fatalf("exec calls = %d, want 19", len(exec.calls))
	}
	if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot") {
		t.Fatalf("schema query = %s", exec.calls[0].query)
	}
	if !strings.Contains(exec.calls[11].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") {
		t.Fatalf("location schema query = %s", exec.calls[11].query)
	}
	for _, call := range exec.calls {
		if strings.Contains(call.query, "vehicle_realtime_kv") {
			t.Fatalf("snapshot writer should not create or write MySQL realtime kv: %s", call.query)
		}
	}
	for _, call := range []snapshotExecCall{exec.calls[0], exec.calls[11]} {
		if strings.Contains(call.query, "fields_json") {
			t.Fatalf("realtime schema should not contain fields_json: %s", call.query)
		}
		for _, column := range []string{"id BIGINT", "created_at", "message_id", "sequence_id", "source_endpoint"} {
			if strings.Contains(call.query, column) {
				t.Fatalf("realtime schema should not contain %s: %s", column, call.query)
			}
		}
		if !strings.Contains(call.query, "PRIMARY KEY (protocol, vin)") {
			t.Fatalf("realtime current-state table should key by protocol/vin: %s", call.query)
		}
	}
	for _, want := range []string{"platform_name", "peer", "parsed_json"} {
		if !strings.Contains(exec.calls[0].query, want) {
			t.Fatalf("snapshot schema should contain %s: %s", want, exec.calls[0].query)
		}
	}
	if strings.Contains(exec.calls[11].query, "idx_location") {
		t.Fatalf("realtime location table should not keep unused geo index: %s", exec.calls[11].query)
	}
	if !strings.Contains(exec.calls[12].query, "total_mileage_event_time") {
		t.Fatalf("location compatibility migration should add total mileage event time: %s", exec.calls[12].query)
	}
	for _, call := range exec.calls[13:17] {
		if !strings.Contains(call.query, "total_mileage") {
			t.Fatalf("schema bootstrap should clean invalid realtime mileage: %s", call.query)
		}
	}
	if !strings.Contains(exec.calls[17].query, "snapshot_backfill") {
		t.Fatalf("access projection baseline should be backfilled explicitly: %s", exec.calls[17].query)
	}
	upsert := exec.calls[18]
	if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") {
		t.Fatalf("upsert query = %s", upsert.query)
	}
	if !strings.Contains(upsert.query, "parsed_json") || !strings.Contains(upsert.query, "platform_name") || !strings.Contains(upsert.query, "peer") {
		t.Fatalf("snapshot upsert should write realtime context columns: %s", upsert.query)
	}
	for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} {
		if strings.Contains(upsert.query, column) {
			t.Fatalf("snapshot upsert should not write %s: %s", column, upsert.query)
		}
	}
	if got, want := upsert.args[0], "GB32960"; got != want {
		t.Fatalf("protocol arg = %#v, want %q", got, want)
	}
	if got, want := upsert.args[1], "VIN001"; got != want {
		t.Fatalf("vin arg = %#v, want %q", got, want)
	}
	if got, want := upsert.args[4], "1.2.3.4:32960"; got != want {
		t.Fatalf("peer arg = %#v, want %q", got, want)
	}
	if got := upsert.args[5]; !strings.Contains(got.(string), `"gb32960.vehicle.soc_percent":"90"`) {
		t.Fatalf("parsed json arg = %#v", got)
	}
	if len(upsert.args) != 12 {
		t.Fatalf("snapshot upsert args = %d, want 12", len(upsert.args))
	}
	for _, want := range []string{"access_first_seen_at", "access_previous_received_at", "access_latest_received_at", "access_report_interval_ms", "access_sample_count", "access_latest_event_id", "live_writer", "TIMESTAMPDIFF(MICROSECOND"} {
		if !strings.Contains(upsert.query, want) {
			t.Fatalf("access projection upsert missing %q: %s", want, upsert.query)
		}
	}
}

func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	writer := NewSnapshotWriter(db)

	mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN platform_name").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN peer").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN parsed_json").
		WillReturnResult(sqlmock.NewResult(0, 0))
	for _, column := range []string{"access_first_seen_at", "access_previous_received_at", "access_latest_received_at", "access_report_interval_ms", "access_sample_count", "access_latest_event_id", "access_first_seen_source"} {
		mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN " + column).
			WillReturnResult(sqlmock.NewResult(0, 0))
	}
	mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("ALTER TABLE vehicle_realtime_location ADD COLUMN total_mileage_event_time").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("UPDATE vehicle_realtime_location").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("UPDATE vehicle_realtime_snapshot").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("UPDATE vehicle_realtime_snapshot").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("UPDATE vehicle_realtime_snapshot").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("UPDATE vehicle_realtime_snapshot").
		WillReturnResult(sqlmock.NewResult(0, 0))

	if err := writer.EnsureSchema(context.Background()); err != nil {
		t.Fatalf("EnsureSchema() error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unexpected compatibility migration query: %v", err)
	}
}

func TestRealtimeAccessProjectionSQLProtectsDuplicateAndOutOfOrderReceipts(t *testing.T) {
	accessStart := strings.Index(upsertRealtimeSnapshotSQL, "access_previous_received_at =")
	if accessStart < 0 {
		t.Fatal("access projection assignments are missing")
	}
	accessSQL := upsertRealtimeSnapshotSQL[accessStart:]
	for _, want := range []string{
		"VALUES(access_latest_received_at) > vehicle_realtime_snapshot.access_latest_received_at",
		"VALUES(access_latest_event_id) <> vehicle_realtime_snapshot.access_latest_event_id",
		"TIMESTAMPDIFF(MICROSECOND",
		"access_sample_count + 1",
	} {
		if !strings.Contains(accessSQL, want) {
			t.Fatalf("access projection SQL missing %q:\n%s", want, accessSQL)
		}
	}
	if strings.Contains(accessSQL, "VALUES(event_time)") {
		t.Fatalf("access receipt projection must not be gated by device event time:\n%s", accessSQL)
	}
	ordered := []string{"access_previous_received_at =", "access_report_interval_ms =", "access_sample_count =", "access_latest_received_at =", "access_latest_event_id ="}
	previous := -1
	for _, assignment := range ordered {
		position := strings.Index(accessSQL, assignment)
		if position <= previous {
			t.Fatalf("access assignment %q must preserve old latest receipt before advancing it: %s", assignment, accessSQL)
		}
		previous = position
	}
	for _, want := range []string{"COALESCE(access_first_seen_at, received_at, updated_at)", "snapshot_backfill", "access_sample_count = 0"} {
		if !strings.Contains(backfillRealtimeAccessProjectionSQL, want) {
			t.Fatalf("access baseline backfill missing %q: %s", want, backfillRealtimeAccessProjectionSQL)
		}
	}
}

func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:       envelope.ProtocolJT808,
		MessageID:      "0x0200",
		Sequence:       9,
		VIN:            "VIN001",
		Phone:          "13307795425",
		DeviceID:       "device-1",
		Plate:          "沪A12345",
		SourceEndpoint: "1.2.3.4:808",
		EventTimeMS:    1782918600000,
		ReceivedAtMS:   1782918601000,
		EventID:        "event-1",
		Parsed:         map[string]any{"location": map[string]any{"latitude": 30.590151, "longitude": 121.069881}},
		ParsedFields: map[string]any{
			"jt808.location.latitude":         30.590151,
			"jt808.location.longitude":        121.069881,
			"jt808.location.speed_kmh":        23.0,
			"jt808.location.total_mileage_km": "10241.2",
			"jt808.location.altitude_m":       uint16(5),
			"jt808.location.direction_deg":    uint16(79),
			"jt808.location.alarm_flag":       uint32(0),
			"jt808.location.status_flag":      uint32(786435),
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}

	if len(exec.calls) != 2 {
		t.Fatalf("exec calls = %d, want 2", len(exec.calls))
	}
	locationUpsert := exec.calls[1]
	if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location") {
		t.Fatalf("location upsert query = %s", locationUpsert.query)
	}
	if strings.Contains(locationUpsert.query, "fields_json") {
		t.Fatalf("location upsert should not write fields_json: %s", locationUpsert.query)
	}
	for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} {
		if strings.Contains(locationUpsert.query, column) {
			t.Fatalf("location upsert should not write %s: %s", column, locationUpsert.query)
		}
	}
	if got, want := locationUpsert.args[0], "JT808"; got != want {
		t.Fatalf("protocol arg = %#v, want %q", got, want)
	}
	if got, want := locationUpsert.args[1], "VIN001"; got != want {
		t.Fatalf("vin arg = %#v, want %q", got, want)
	}
	if got, want := locationUpsert.args[4], 30.590151; got != want {
		t.Fatalf("latitude arg = %#v, want %v", got, want)
	}
	if got, want := locationUpsert.args[5], 121.069881; got != want {
		t.Fatalf("longitude arg = %#v, want %v", got, want)
	}
	if got, want := locationUpsert.args[6], 23.0; got != want {
		t.Fatalf("speed arg = %#v, want %v", got, want)
	}
	if got, want := locationUpsert.args[7], 10241.2; got != want {
		t.Fatalf("mileage arg = %#v, want %v", got, want)
	}
	if _, ok := locationUpsert.args[8].(time.Time); !ok {
		t.Fatalf("total mileage event time arg = %#v, want time.Time", locationUpsert.args[8])
	}
	if got := locationUpsert.args[9]; got != nil {
		t.Fatalf("JT808 location should not invent SOC, got %#v", got)
	}
	if len(locationUpsert.args) != 16 {
		t.Fatalf("location upsert args = %d, want 16", len(locationUpsert.args))
	}
}

func TestSnapshotWriterDropsNonPositiveTotalMileageFromRealtimeStores(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		MessageID:    "0x0200",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		ParsedFields: map[string]any{
			"jt808.location.latitude":         "30.590151",
			"jt808.location.longitude":        "121.069881",
			"jt808.location.speed_kmh":        "23",
			"jt808.location.total_mileage_km": "0",
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}

	if len(exec.calls) != 2 {
		t.Fatalf("exec calls = %d, want snapshot + location", len(exec.calls))
	}
	parsedJSON, ok := exec.calls[0].args[5].(string)
	if !ok {
		t.Fatalf("snapshot parsed_json arg = %#v", exec.calls[0].args[5])
	}
	if strings.Contains(parsedJSON, "total_mileage_km") {
		t.Fatalf("snapshot parsed_json should drop non-positive mileage: %s", parsedJSON)
	}
	if !strings.Contains(parsedJSON, "jt808.location.speed_kmh") {
		t.Fatalf("snapshot parsed_json should keep valid fields: %s", parsedJSON)
	}
	if exec.calls[1].args[7] != nil {
		t.Fatalf("location total mileage arg = %#v, want nil", exec.calls[1].args[7])
	}
	if exec.calls[1].args[8] != nil {
		t.Fatalf("location total mileage time arg = %#v, want nil", exec.calls[1].args[8])
	}
}

func TestSnapshotWriterNormalizesFarFutureEventTime(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)
	received := time.Date(2026, 7, 12, 9, 30, 0, 0, time.UTC)
	futureEvent := time.Date(2026, 7, 13, 9, 30, 0, 0, time.UTC)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		MessageID:    "0x0200",
		VIN:          "VIN001",
		EventTimeMS:  futureEvent.UnixMilli(),
		ReceivedAtMS: received.UnixMilli(),
		Parsed:       map[string]any{"location": map[string]any{"latitude": 30.590151, "longitude": 121.069881}},
		ParsedFields: map[string]any{
			"jt808.location.latitude":         30.590151,
			"jt808.location.longitude":        121.069881,
			"jt808.location.total_mileage_km": "10241.2",
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}

	if len(exec.calls) != 2 {
		t.Fatalf("exec calls = %d, want snapshot + location", len(exec.calls))
	}
	if got, ok := exec.calls[0].args[6].(time.Time); !ok || !got.Equal(received) {
		t.Fatalf("snapshot event_time arg = %#v, want received %s", exec.calls[0].args[6], received)
	}
	if got, ok := exec.calls[1].args[3].(time.Time); !ok || !got.Equal(received) {
		t.Fatalf("location event_time arg = %#v, want received %s", exec.calls[1].args[3], received)
	}
	if got, ok := exec.calls[1].args[8].(time.Time); !ok || !got.Equal(received) {
		t.Fatalf("total mileage event time arg = %#v, want received %s", exec.calls[1].args[8], received)
	}
}

func TestRealtimeLocationUsesProtocolMileageMapping(t *testing.T) {
	tests := []struct {
		name        string
		protocol    envelope.Protocol
		fields      map[string]any
		wantMileage float64
	}{
		{
			name:     "gb32960 kilometers",
			protocol: envelope.ProtocolGB32960,
			fields: map[string]any{
				"gb32960.position.latitude":        30.590151,
				"gb32960.position.longitude":       121.069881,
				"gb32960.vehicle.total_mileage_km": "4123.9",
			},
			wantMileage: 4123.9,
		},
		{
			name:     "jt808 kilometers",
			protocol: envelope.ProtocolJT808,
			fields: map[string]any{
				"jt808.location.latitude":         30.590151,
				"jt808.location.longitude":        121.069881,
				"jt808.location.total_mileage_km": json.Number("10241.2"),
			},
			wantMileage: 10241.2,
		},
		{
			name:     "yutong meters",
			protocol: envelope.ProtocolYutongMQTT,
			fields: map[string]any{
				"yutong_mqtt.data.latitude":      30.590151,
				"yutong_mqtt.data.longitude":     121.069881,
				"yutong_mqtt.data.total_mileage": "86737000",
			},
			wantMileage: 86737,
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			row, ok := realtimeLocationFromEnvelope(envelope.FrameEnvelope{
				Protocol:     tc.protocol,
				EventTimeMS:  1782918600000,
				ReceivedAtMS: 1782918601000,
				ParsedFields: tc.fields,
			}, "VIN001", "")
			if !ok {
				t.Fatal("realtimeLocationFromEnvelope() should produce a location")
			}
			if got, ok := row.TotalMileageKM.(float64); !ok || got != tc.wantMileage {
				t.Fatalf("total mileage = %#v, want %v", row.TotalMileageKM, tc.wantMileage)
			}
			if _, ok := row.TotalMileageAt.(time.Time); !ok {
				t.Fatalf("total mileage event time = %#v, want time.Time", row.TotalMileageAt)
			}
		})
	}
}

func TestRealtimeLocationDoesNotAdvanceMileageFromCoreFieldOnSparseFrame(t *testing.T) {
	row, ok := realtimeLocationFromEnvelope(envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolYutongMQTT,
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		ParsedFields: map[string]any{
			"yutong_mqtt.data.latitude":  "30.590151",
			"yutong_mqtt.data.longitude": "121.069881",
		},
	}, "VIN001", "")
	if !ok {
		t.Fatal("realtimeLocationFromEnvelope() should produce a location")
	}
	if row.TotalMileageKM != nil {
		t.Fatalf("sparse frame total mileage = %#v, want nil", row.TotalMileageKM)
	}
	if row.TotalMileageAt != nil {
		t.Fatalf("sparse frame total mileage event time = %#v, want nil", row.TotalMileageAt)
	}
}

func TestRealtimeLocationRejectsBareStandardizedFields(t *testing.T) {
	_, ok := realtimeLocationFromEnvelope(envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Fields: map[string]any{
			envelope.FieldLatitude:  30.590151,
			envelope.FieldLongitude: 121.069881,
		},
	}, "VIN001", "")
	if ok {
		t.Fatal("bare standardized fields must not drive the canonical location projection")
	}
}

func TestSnapshotWriterMergesGB32960ParsedJSONAcrossSplitRealtimeFramesInUpsert(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	writer := NewSnapshotWriter(db)

	mock.ExpectExec("INSERT INTO vehicle_realtime_snapshot").
		WithArgs(
			"GB32960",
			"VIN001",
			"",
			"",
			"",
			jsonFlatFieldsArg{fields: map[string]string{
				"gb32960.gd_fc_stack.stack_count": "1",
				"gb32960.gd_fc_stack.type":        "0x30",
			}},
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	err = writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x02",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed: map[string]any{
			"data_units": []any{
				map[string]any{"type": "0x30", "name": "gd_fc_stack", "value": map[string]any{"stack_count": 1}},
			},
		},
		ParsedFields: map[string]any{
			"gb32960.gd_fc_stack.stack_count": "1",
			"gb32960.gd_fc_stack.type":        "0x30",
		},
	})
	if err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestSnapshotWriterUsesEnvelopeParsedFieldsWithoutReflattening(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	writer := NewSnapshotWriter(db)

	mock.ExpectExec("INSERT INTO vehicle_realtime_snapshot").
		WithArgs(
			"GB32960",
			"VIN001",
			"",
			"",
			"",
			jsonFlatFieldsArg{fields: map[string]string{
				"gb32960.vehicle.speed_kmh": "12.3",
			}},
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	err = writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x02",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed: map[string]any{
			"data_units": []any{
				map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"speed_kmh": 99}},
			},
		},
		ParsedFields: map[string]any{
			"gb32960.vehicle.speed_kmh": "12.3",
		},
		Fields: map[string]any{envelope.FieldSpeedKMH: 12.3},
	})
	if err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestSnapshotWriterCollapsesGB32960StackFragmentsWhenMergingParsedJSON(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	writer := NewSnapshotWriter(db)

	mock.ExpectExec("INSERT INTO vehicle_realtime_snapshot").
		WithArgs(
			"GB32960",
			"VIN001",
			"",
			"",
			"",
			jsonFlatFieldsArg{
				fields: map[string]string{
					"gb32960.gd_fc_stack.stack_water_outlet_temp_c": "63",
				},
				forbidden: []string{
					"gb32960.gd_fc_stack.frame_cell_start",
					"gb32960.gd_fc_stack.frame_cell_count",
					"gb32960.gd_fc_stack.frame_max_cell_voltage_v",
					"gb32960.gd_fc_stack.frame_min_cell_voltage_v",
				},
			},
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
			sqlmock.AnyArg(),
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	err = writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x02",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed: map[string]any{
			"data_units": []any{
				map[string]any{
					"type": "0x30",
					"name": "gd_fc_stack",
					"value": map[string]any{
						"stack_count": 1,
						"summaries": []any{
							map[string]any{"cell_count": 432, "stack_water_outlet_temp_c": 63, "frame_cell_start": 401, "frame_cell_count": 32},
						},
					},
				},
			},
		},
		ParsedFields: map[string]any{
			"gb32960.gd_fc_stack.stack_water_outlet_temp_c": "63",
		},
	})
	if err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestRealtimeUpsertsDoNotOverwriteWithOlderEventTime(t *testing.T) {
	for _, tc := range []struct {
		name  string
		table string
		query string
	}{
		{name: "snapshot", table: "vehicle_realtime_snapshot", query: upsertRealtimeSnapshotSQL},
		{name: "location", table: "vehicle_realtime_location", query: upsertRealtimeLocationSQL},
	} {
		t.Run(tc.name, func(t *testing.T) {
			guard := "VALUES(event_time) IS NOT NULL AND (" + tc.table + ".event_time IS NULL OR VALUES(event_time) >= " + tc.table + ".event_time)"
			if !strings.Contains(tc.query, guard) {
				t.Fatalf("upsert should guard realtime state by event_time, missing %q in:\n%s", guard, tc.query)
			}
			if strings.Contains(tc.query, "event_time = VALUES(event_time)") {
				t.Fatalf("upsert should not unconditionally replace event_time:\n%s", tc.query)
			}
			if strings.Contains(tc.query, "event_id = VALUES(event_id)") {
				t.Fatalf("upsert should not unconditionally replace event_id:\n%s", tc.query)
			}
		})
	}
}

func TestRealtimeSnapshotUpsertMergesParsedJSONInSQL(t *testing.T) {
	for _, want := range []string{
		"JSON_MERGE_PATCH",
		"COALESCE(NULLIF(vehicle_realtime_snapshot.parsed_json, ''), JSON_OBJECT())",
		"VALUES(parsed_json)",
	} {
		if !strings.Contains(upsertRealtimeSnapshotSQL, want) {
			t.Fatalf("snapshot upsert should merge parsed_json in SQL, missing %q:\n%s", want, upsertRealtimeSnapshotSQL)
		}
	}
}

func TestRealtimeLocationUpsertKeepsExistingSparseFieldsWhenMQTTOnlySendsCoordinates(t *testing.T) {
	for _, column := range []string{
		"speed_kmh",
		"total_mileage_km",
		"soc_percent",
		"altitude_m",
		"direction_deg",
		"alarm_flag",
		"status_flag",
	} {
		want := column + " = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), COALESCE(VALUES(" + column + "), " + column + "), " + column + ")"
		if !strings.Contains(upsertRealtimeLocationSQL, want) {
			t.Fatalf("location upsert should keep existing %s when incoming sparse MQTT frame omits it; missing:\n%s\nin:\n%s", column, want, upsertRealtimeLocationSQL)
		}
	}
	wantMileageAt := "total_mileage_event_time = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), IF(VALUES(total_mileage_km) IS NOT NULL, COALESCE(VALUES(total_mileage_event_time), total_mileage_event_time), total_mileage_event_time), total_mileage_event_time)"
	if !strings.Contains(upsertRealtimeLocationSQL, wantMileageAt) {
		t.Fatalf("location upsert should only advance total_mileage_event_time when the incoming frame carries mileage; missing:\n%s\nin:\n%s", wantMileageAt, upsertRealtimeLocationSQL)
	}
}

func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) {
	exec := &recordingSnapshotExec{}
	resolver := &recordingPlateResolver{plate: "沪A12345"}
	writer := NewSnapshotWriterWithPlateResolver(exec, resolver)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x02",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed:       sampleGB32960RealtimeParsed(),
		ParsedFields: sampleGB32960LocationFields(),
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}

	if resolver.vin != "VIN001" {
		t.Fatalf("resolver vin = %q", resolver.vin)
	}
	if len(exec.calls) != 2 {
		t.Fatalf("exec calls = %d, want 2", len(exec.calls))
	}
	if got, want := exec.calls[0].args[2], "沪A12345"; got != want {
		t.Fatalf("snapshot plate arg = %#v, want %q", got, want)
	}
	if got, want := exec.calls[1].args[2], "沪A12345"; got != want {
		t.Fatalf("location plate arg = %#v, want %q", got, want)
	}
}

func TestBindingPlateResolverUsesPrimaryKeyLookupWithoutSort(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New() error = %v", err)
	}
	defer db.Close()
	mock.ExpectQuery("SELECT plate FROM vehicle_identity_binding WHERE vin = \\? AND plate IS NOT NULL AND plate <> ''$").
		WithArgs("VIN001").
		WillReturnRows(sqlmock.NewRows([]string{"plate"}).AddRow("沪A12345"))

	plate, err := NewBindingPlateResolver(db, "vehicle_identity_binding").PlateByVIN(context.Background(), " VIN001 ")
	if err != nil {
		t.Fatalf("PlateByVIN() error = %v", err)
	}
	if plate != "沪A12345" {
		t.Fatalf("plate = %q", plate)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestSnapshotWriterCachesBindingPlateByVIN(t *testing.T) {
	exec := &recordingSnapshotExec{}
	resolver := &recordingPlateResolver{plate: "沪A12345"}
	writer := NewSnapshotWriterWithPlateResolver(exec, NewCachedPlateResolver(resolver, time.Hour))
	event := envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x02",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed:       sampleGB32960RealtimeParsed(),
		ParsedFields: sampleGB32960LocationFields(),
	}

	if err := writer.Update(context.Background(), event); err != nil {
		t.Fatalf("first Update() error = %v", err)
	}
	event.EventTimeMS += 1000
	event.ReceivedAtMS += 1000
	if err := writer.Update(context.Background(), event); err != nil {
		t.Fatalf("second Update() error = %v", err)
	}

	if resolver.calls != 1 {
		t.Fatalf("plate resolver calls = %d, want 1", resolver.calls)
	}
	if len(exec.calls) != 4 {
		t.Fatalf("exec calls = %d, want 4", len(exec.calls))
	}
	for _, index := range []int{0, 2} {
		if got, want := exec.calls[index].args[2], "沪A12345"; got != want {
			t.Fatalf("snapshot %d plate arg = %#v, want %q", index, got, want)
		}
	}
}

func TestCachedPlateResolverEvictsOldestEntryWhenLimitExceeded(t *testing.T) {
	delegate := &mapPlateResolver{plates: map[string]string{
		"VIN001": "沪A00001",
		"VIN002": "沪A00002",
		"VIN003": "沪A00003",
	}}
	resolver := NewCachedPlateResolverWithMaxEntries(delegate, time.Hour, 2)
	now := time.Date(2026, 7, 12, 10, 0, 0, 0, time.UTC)
	resolver.now = func() time.Time { return now }
	var observed PlateCacheStats
	resolver.SetStatsObserver(func(stats PlateCacheStats) {
		observed = stats
	})
	for _, vin := range []string{"VIN001", "VIN002", "VIN003"} {
		if _, err := resolver.PlateByVIN(context.Background(), vin); err != nil {
			t.Fatalf("PlateByVIN(%s) error = %v", vin, err)
		}
		now = now.Add(time.Second)
	}

	stats := resolver.CacheStats()
	if stats.Entries != 2 || stats.MaxEntries != 2 || stats.Evictions != 1 {
		t.Fatalf("cache stats = %+v, want entries=2 max=2 evictions=1", stats)
	}
	if observed != stats {
		t.Fatalf("observed stats = %+v, want %+v", observed, stats)
	}
	if _, ok := resolver.entries["VIN001"]; ok {
		t.Fatalf("oldest VIN should be evicted")
	}
	if _, ok := resolver.entries["VIN002"]; !ok {
		t.Fatalf("VIN002 should remain cached")
	}
	if _, ok := resolver.entries["VIN003"]; !ok {
		t.Fatalf("VIN003 should remain cached")
	}
}

func TestCachedPlateResolverCanDisableEntryLimit(t *testing.T) {
	delegate := &mapPlateResolver{plates: map[string]string{
		"VIN001": "沪A00001",
		"VIN002": "沪A00002",
	}}
	resolver := NewCachedPlateResolverWithMaxEntries(delegate, time.Hour, 0)
	for _, vin := range []string{"VIN001", "VIN002"} {
		if _, err := resolver.PlateByVIN(context.Background(), vin); err != nil {
			t.Fatalf("PlateByVIN(%s) error = %v", vin, err)
		}
	}
	stats := resolver.CacheStats()
	if stats.Entries != 2 || stats.MaxEntries != 0 || stats.Evictions != 0 {
		t.Fatalf("cache stats = %+v, want unlimited cache with two entries", stats)
	}
}

func TestSnapshotWriterKeepsEventPlateWhenPresent(t *testing.T) {
	exec := &recordingSnapshotExec{}
	resolver := &recordingPlateResolver{plate: "沪B99999"}
	writer := NewSnapshotWriterWithPlateResolver(exec, resolver)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		MessageID:    "0x0200",
		VIN:          "VIN001",
		Plate:        "沪A12345",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		ParsedFields: map[string]any{
			"jt808.location.latitude":  30.590151,
			"jt808.location.longitude": 121.069881,
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}

	if resolver.vin != "" {
		t.Fatalf("resolver should not be called, got vin=%q", resolver.vin)
	}
	if got, want := exec.calls[0].args[2], "沪A12345"; got != want {
		t.Fatalf("snapshot plate arg = %#v, want %q", got, want)
	}
}

func TestSnapshotWriterIgnoresMissingBindingPlate(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriterWithPlateResolver(exec, &recordingPlateResolver{err: sql.ErrNoRows})

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x02",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed:       sampleGB32960RealtimeParsed(),
		ParsedFields: map[string]any{"gb32960.vehicle.soc_percent": 90},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if got := exec.calls[0].args[2]; got != "" {
		t.Fatalf("snapshot plate arg = %#v, want empty", got)
	}
}

func TestSnapshotWriterReturnsUnexpectedPlateLookupError(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriterWithPlateResolver(exec, &recordingPlateResolver{err: errors.New("db down")})

	err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x02",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed:       sampleGB32960RealtimeParsed(),
		ParsedFields: map[string]any{"gb32960.vehicle.soc_percent": 90},
	})
	if err == nil || !strings.Contains(err.Error(), "db down") {
		t.Fatalf("Update() error = %v, want db down", err)
	}
	if len(exec.calls) != 0 {
		t.Fatalf("exec calls = %d, want 0", len(exec.calls))
	}
}

func TestSnapshotWriterSkipsUnknownVehicleKey(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol: envelope.ProtocolJT808,
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if len(exec.calls) != 0 {
		t.Fatalf("exec calls = %d, want 0", len(exec.calls))
	}
}

func TestSnapshotWriterSkipsEmptyVINEvenWhenPhoneExists(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		MessageID:    "0x0200",
		Phone:        "13307795425",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Fields: map[string]any{
			envelope.FieldLatitude:  30.590151,
			envelope.FieldLongitude: 121.069881,
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if len(exec.calls) != 0 {
		t.Fatalf("exec calls = %d, want 0", len(exec.calls))
	}
}

func TestSnapshotWriterSkipsGB32960HeartbeatWithoutRealtimePayload(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:  envelope.ProtocolGB32960,
		MessageID: "0x07",
		VIN:       "ABCDE600000000009",
		Parsed: map[string]any{
			"header": map[string]any{
				"vin":                "ABCDE600000000009",
				"command":            "0x07",
				"actual_body_length": 0,
			},
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if len(exec.calls) != 0 {
		t.Fatalf("exec calls = %d, want 0", len(exec.calls))
	}
}

func TestSnapshotWriterSkipsGB32960NonRealtimeEvenWithConnectionMetadata(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x07",
		VIN:          "ABCDE600000000009",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed: map[string]any{
			"header":        map[string]any{"command": "0x07", "vin": "ABCDE600000000009"},
			"platform_name": "YueJin",
		},
		Fields: map[string]any{
			"platform_account": "YueJin",
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if len(exec.calls) != 0 {
		t.Fatalf("exec calls = %d, want 0", len(exec.calls))
	}
}

func TestSnapshotWriterSkipsJT808NonLocationEvenWithIdentityFields(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		MessageID:    "0x0100",
		VIN:          "VIN001",
		Plate:        "粤A12345",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed: map[string]any{
			"registration": map[string]any{"plate": "粤A12345"},
		},
		Fields: map[string]any{
			"plate": "粤A12345",
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if len(exec.calls) != 0 {
		t.Fatalf("exec calls = %d, want 0", len(exec.calls))
	}
}

func TestSnapshotWriterSkipsMQTTWithoutActualDataFields(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolYutongMQTT,
		MessageID:    "MQTT",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed: map[string]any{
			"topic": "/vehicle/VIN001/state",
			"data":  map[string]any{},
		},
		ParsedFields: map[string]any{
			"yutong_mqtt.metadata.topic": "/vehicle/VIN001/state",
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if len(exec.calls) != 0 {
		t.Fatalf("exec calls = %d, want 0", len(exec.calls))
	}
}

func TestSnapshotWriterAcceptsGB32960RetransmissionFields(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x03",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		ParsedFields: map[string]any{
			"gb32960.vehicle.soc_percent": 90,
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if len(exec.calls) != 1 || !strings.Contains(exec.calls[0].query, "vehicle_realtime_snapshot") {
		t.Fatalf("GB32960 retransmission should update snapshot, calls=%#v", exec.calls)
	}
}

func TestSnapshotWriterSkipsEmptyDataUnitsWithoutCoreFields(t *testing.T) {
	exec := &recordingSnapshotExec{}
	writer := NewSnapshotWriter(exec)

	if err := writer.Update(context.Background(), envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		MessageID:    "0x02",
		VIN:          "VIN001",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918601000,
		Parsed:       map[string]any{"data_units": []any{}},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if len(exec.calls) != 0 {
		t.Fatalf("exec calls = %d, want 0", len(exec.calls))
	}
}

type recordingSnapshotExec struct {
	calls []snapshotExecCall
}

func sampleGB32960RealtimeParsed() map[string]any {
	return map[string]any{
		"data_units": []any{
			map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 90}},
		},
	}
}

func sampleGB32960LocationFields() map[string]any {
	return map[string]any{
		"gb32960.position.latitude":   30.590151,
		"gb32960.position.longitude":  121.069881,
		"gb32960.vehicle.soc_percent": 90,
	}
}

type snapshotExecCall struct {
	query string
	args  []any
}

func (e *recordingSnapshotExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) {
	e.calls = append(e.calls, snapshotExecCall{query: query, args: args})
	return nil, nil
}

type recordingPlateResolver struct {
	vin   string
	plate string
	err   error
	calls int
}

func (r *recordingPlateResolver) PlateByVIN(_ context.Context, vin string) (string, error) {
	r.calls++
	r.vin = vin
	return r.plate, r.err
}

type mapPlateResolver struct {
	plates map[string]string
}

func (r *mapPlateResolver) PlateByVIN(_ context.Context, vin string) (string, error) {
	plate := strings.TrimSpace(r.plates[strings.TrimSpace(vin)])
	if plate == "" {
		return "", sql.ErrNoRows
	}
	return plate, nil
}

type jsonFlatFieldsArg struct {
	fields    map[string]string
	forbidden []string
}

func (a jsonFlatFieldsArg) Match(value driver.Value) bool {
	text, ok := value.(string)
	if !ok {
		return false
	}
	var parsed map[string]any
	if err := json.Unmarshal([]byte(text), &parsed); err != nil {
		return false
	}
	for field, want := range a.fields {
		if got, ok := parsed[field].(string); !ok || got != want {
			return false
		}
	}
	for _, field := range a.forbidden {
		if _, ok := parsed[field]; ok {
			return false
		}
	}
	return true
}