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"}},
			},
		},
		Fields: map[string]any{envelope.FieldSOCPercent: 90},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}

	if len(exec.calls) != 6 {
		t.Fatalf("exec calls = %d, want 6", 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[4].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") {
		t.Fatalf("location schema query = %s", exec.calls[4].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[4]} {
		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[4].query, "idx_location") {
		t.Fatalf("realtime location table should not keep unused geo index: %s", exec.calls[4].query)
	}
	upsert := exec.calls[5]
	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) != 9 {
		t.Fatalf("snapshot upsert args = %d, want 9", len(upsert.args))
	}
}

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))
	mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location").
		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 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}},
		Fields: map[string]any{
			envelope.FieldLatitude:       30.590151,
			envelope.FieldLongitude:      121.069881,
			envelope.FieldSpeedKMH:       23.0,
			envelope.FieldTotalMileageKM: 10241.2,
			envelope.FieldSOCPercent:     88,
			"altitude_m":                 uint16(5),
			"direction_deg":              uint16(79),
			"alarm_flag":                 uint32(0),
			"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 got, want := locationUpsert.args[8], 88.0; got != want {
		t.Fatalf("soc arg = %#v, want %v", got, want)
	}
	if len(locationUpsert.args) != 15 {
		t.Fatalf("location upsert args = %d, want 15", len(locationUpsert.args))
	}
}

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

	existing := `{"data_units":[{"type":"0x01","name":"vehicle","value":{"soc_percent":88}}]}`
	mock.ExpectQuery("SELECT parsed_json FROM vehicle_realtime_snapshot WHERE protocol = \\? AND vin = \\?").
		WithArgs("GB32960", "VIN001").
		WillReturnRows(sqlmock.NewRows([]string{"parsed_json"}).AddRow(existing))
	mock.ExpectExec("INSERT INTO vehicle_realtime_snapshot").
		WithArgs(
			"GB32960",
			"VIN001",
			"",
			"",
			"",
			jsonFlatFieldsArg{fields: map[string]string{
				"gb32960.vehicle.soc_percent":     "88",
				"gb32960.vehicle.type":            "0x01",
				"gb32960.gd_fc_stack.stack_count": "1",
				"gb32960.gd_fc_stack.type":        "0x30",
			}},
			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}},
			},
		},
	})
	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.ExpectQuery("SELECT parsed_json FROM vehicle_realtime_snapshot WHERE protocol = \\? AND vin = \\?").
		WithArgs("GB32960", "VIN001").
		WillReturnRows(sqlmock.NewRows([]string{"parsed_json"}))
	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(),
		).
		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)

	existing := `{"data_units":[{"type":"0x30","name":"gd_fc_stack","value":{"stack_count":1,"summaries":[{"cell_count":432,"stack_water_outlet_temp_c":65,"frame_cell_start":201,"frame_cell_count":200}]}}]}`
	mock.ExpectQuery("SELECT parsed_json FROM vehicle_realtime_snapshot WHERE protocol = \\? AND vin = \\?").
		WithArgs("GB32960", "VIN001").
		WillReturnRows(sqlmock.NewRows([]string{"parsed_json"}).AddRow(existing))
	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(),
		).
		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},
						},
					},
				},
			},
		},
	})
	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 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)
		}
	}
}

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(),
		Fields: map[string]any{
			envelope.FieldLatitude:  30.590151,
			envelope.FieldLongitude: 121.069881,
		},
	}); 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(),
		Fields: map[string]any{
			envelope.FieldLatitude:  30.590151,
			envelope.FieldLongitude: 121.069881,
		},
	}

	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 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,
		Fields: map[string]any{
			envelope.FieldLatitude:  30.590151,
			envelope.FieldLongitude: 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(),
		Fields:       map[string]any{envelope.FieldSOCPercent: 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(),
		Fields:       map[string]any{envelope.FieldSOCPercent: 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{},
		},
	}); err != nil {
		t.Fatalf("Update() error = %v", err)
	}
	if len(exec.calls) != 0 {
		t.Fatalf("exec calls = %d, want 0", len(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}},
		},
	}
}

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 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
}