lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go

package realtime

import (
	"context"
	"database/sql"
	"errors"
	"strings"
	"testing"

	"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) != 3 {
		t.Fatalf("exec calls = %d, want 3", 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[1].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") {
		t.Fatalf("location schema query = %s", exec.calls[1].query)
	}
	for _, call := range exec.calls[:2] {
		if strings.Contains(call.query, "parsed_json") || strings.Contains(call.query, "fields_json") {
			t.Fatalf("realtime schema should not contain json payload columns: %s", call.query)
		}
		for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} {
			if strings.Contains(call.query, column) {
				t.Fatalf("realtime schema should not contain %s: %s", column, call.query)
			}
		}
	}
	upsert := exec.calls[2]
	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, "fields_json") {
		t.Fatalf("snapshot upsert should not write json payload 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 len(upsert.args) != 6 {
		t.Fatalf("snapshot upsert args = %d, want 6", 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("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 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:       map[string]any{"data_units": []any{}},
		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 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:       map[string]any{"data_units": []any{}},
		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:       map[string]any{"data_units": []any{}},
		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))
	}
}

type recordingSnapshotExec struct {
	calls []snapshotExecCall
}

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
}

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