perf(go): cache realtime plate lookups

go/vehicle-gateway/internal/realtime/snapshot_writer.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"errors"
 	"strings"
+	"sync"
 	"time"
 
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
@@ -18,6 +19,69 @@ type PlateResolver interface {
 	PlateByVIN(context.Context, string) (string, error)
 }
 
+type CachedPlateResolver struct {
+	delegate PlateResolver
+	ttl      time.Duration
+	now      func() time.Time
+	mu       sync.Mutex
+	entries  map[string]cachedPlateEntry
+}
+
+type cachedPlateEntry struct {
+	plate     string
+	notFound  bool
+	expiresAt time.Time
+}
+
+func NewCachedPlateResolver(delegate PlateResolver, ttl time.Duration) *CachedPlateResolver {
+	if delegate == nil {
+		panic("cached plate resolver delegate must not be nil")
+	}
+	if ttl <= 0 {
+		ttl = 10 * time.Minute
+	}
+	return &CachedPlateResolver{
+		delegate: delegate,
+		ttl:      ttl,
+		now:      time.Now,
+		entries:  map[string]cachedPlateEntry{},
+	}
+}
+
+func (r *CachedPlateResolver) PlateByVIN(ctx context.Context, vin string) (string, error) {
+	vin = strings.TrimSpace(vin)
+	if vin == "" {
+		return "", sql.ErrNoRows
+	}
+	now := r.now()
+	r.mu.Lock()
+	entry, ok := r.entries[vin]
+	if ok && now.Before(entry.expiresAt) {
+		r.mu.Unlock()
+		if entry.notFound {
+			return "", sql.ErrNoRows
+		}
+		return entry.plate, nil
+	}
+	r.mu.Unlock()
+
+	plate, err := r.delegate.PlateByVIN(ctx, vin)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return "", err
+	}
+	r.mu.Lock()
+	r.entries[vin] = cachedPlateEntry{
+		plate:     strings.TrimSpace(plate),
+		notFound:  errors.Is(err, sql.ErrNoRows),
+		expiresAt: now.Add(r.ttl),
+	}
+	r.mu.Unlock()
+	if err != nil {
+		return "", err
+	}
+	return strings.TrimSpace(plate), nil
+}
+
 type SnapshotWriter struct {
 	exec          SnapshotExecer
 	plateResolver PlateResolver

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

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/DATA-DOG/go-sqlmock"
 
@@ -212,6 +213,44 @@ func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) {
 	}
 }
 
+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,
+		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"}
@@ -358,9 +397,11 @@ 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
 }