package realtime

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

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

type SnapshotExecer interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
}

type PlateResolver interface {
	PlateByVIN(context.Context, string) (string, error)
}

type CachedPlateResolver struct {
	delegate   PlateResolver
	ttl        time.Duration
	maxEntries int
	now        func() time.Time
	observe    func(PlateCacheStats)
	mu         sync.Mutex
	entries    map[string]cachedPlateEntry
	evictions  int
}

type PlateCacheStats struct {
	Entries    int
	MaxEntries int
	Evictions  int
}

type cachedPlateEntry struct {
	plate     string
	notFound  bool
	expiresAt time.Time
}

func NewCachedPlateResolver(delegate PlateResolver, ttl time.Duration) *CachedPlateResolver {
	return NewCachedPlateResolverWithMaxEntries(delegate, ttl, 200000)
}

func NewCachedPlateResolverWithMaxEntries(delegate PlateResolver, ttl time.Duration, maxEntries int) *CachedPlateResolver {
	if delegate == nil {
		panic("cached plate resolver delegate must not be nil")
	}
	if ttl <= 0 {
		ttl = 10 * time.Minute
	}
	if maxEntries < 0 {
		maxEntries = 0
	}
	return &CachedPlateResolver{
		delegate:   delegate,
		ttl:        ttl,
		maxEntries: maxEntries,
		now:        time.Now,
		entries:    map[string]cachedPlateEntry{},
	}
}

func (r *CachedPlateResolver) SetStatsObserver(observe func(PlateCacheStats)) {
	r.mu.Lock()
	r.observe = observe
	r.mu.Unlock()
}

func (r *CachedPlateResolver) CacheStats() PlateCacheStats {
	r.mu.Lock()
	defer r.mu.Unlock()
	return PlateCacheStats{Entries: len(r.entries), MaxEntries: r.maxEntries, Evictions: r.evictions}
}

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.enforceLimitLocked(now)
	stats := PlateCacheStats{Entries: len(r.entries), MaxEntries: r.maxEntries, Evictions: r.evictions}
	observe := r.observe
	r.mu.Unlock()
	if observe != nil {
		observe(stats)
	}
	if err != nil {
		return "", err
	}
	return strings.TrimSpace(plate), nil
}

func (r *CachedPlateResolver) enforceLimitLocked(now time.Time) {
	if r.maxEntries <= 0 || len(r.entries) <= r.maxEntries {
		return
	}
	for vin, entry := range r.entries {
		if now.After(entry.expiresAt) || now.Equal(entry.expiresAt) {
			delete(r.entries, vin)
			r.evictions++
		}
	}
	for len(r.entries) > r.maxEntries {
		victim := ""
		var victimExpiresAt time.Time
		for vin, entry := range r.entries {
			if victim == "" || entry.expiresAt.Before(victimExpiresAt) {
				victim = vin
				victimExpiresAt = entry.expiresAt
			}
		}
		if victim == "" {
			return
		}
		delete(r.entries, victim)
		r.evictions++
	}
}

type SnapshotWriter struct {
	exec          SnapshotExecer
	plateResolver PlateResolver
}

func NewSnapshotWriter(exec SnapshotExecer) *SnapshotWriter {
	return NewSnapshotWriterWithPlateResolver(exec, nil)
}

func NewSnapshotWriterWithPlateResolver(exec SnapshotExecer, plateResolver PlateResolver) *SnapshotWriter {
	if exec == nil {
		panic("snapshot execer must not be nil")
	}
	return &SnapshotWriter{exec: exec, plateResolver: plateResolver}
}

func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error {
	if _, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL); err != nil {
		return err
	}
	for _, statement := range realtimeSnapshotCompatibilitySQL {
		if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) {
			return err
		}
	}
	if _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL); err != nil {
		return err
	}
	for _, statement := range realtimeLocationCompatibilitySQL {
		if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) {
			return err
		}
	}
	for _, statement := range cleanupInvalidRealtimeTotalMileageSQL {
		if _, err := w.exec.ExecContext(ctx, statement); err != nil {
			return err
		}
	}
	if _, err := w.exec.ExecContext(ctx, backfillRealtimeAccessProjectionSQL); err != nil {
		return err
	}
	return nil
}

func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) error {
	if !envelope.IsRealtimeTelemetryFrame(env) {
		return nil
	}
	vin := strings.TrimSpace(env.VIN)
	if vin == "" {
		return nil
	}
	if !isRealtimeSnapshotEvent(env) {
		return nil
	}
	plate, err := w.plateForEnvelope(ctx, env)
	if err != nil {
		return err
	}
	eventTimeMS, _ := envelope.NormalizedEventTimeMS(env)
	eventTime := nullableTime(eventTimeMS)
	receivedAt := nullableTime(env.ReceivedAtMS)
	platformName := platformNameFromEnvelope(env)
	parsed := snapshotFieldsForEnvelope(env)
	if _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL,
		string(env.Protocol),
		vin,
		plate,
		platformName,
		env.SourceEndpoint,
		marshalParsedJSON(parsed),
		eventTime,
		receivedAt,
		env.StableEventID(),
		receivedAt,
		receivedAt,
		env.StableEventID(),
	); err != nil {
		return err
	}
	location, ok := realtimeLocationFromEnvelope(env, vin, plate)
	if !ok {
		return nil
	}
	_, err = w.exec.ExecContext(ctx, upsertRealtimeLocationSQL,
		location.Protocol,
		location.VIN,
		location.Plate,
		location.EventTime,
		location.Latitude,
		location.Longitude,
		location.SpeedKMH,
		location.TotalMileageKM,
		location.TotalMileageAt,
		location.SOCPercent,
		location.AltitudeM,
		location.DirectionDeg,
		location.AlarmFlag,
		location.StatusFlag,
		location.ReceivedAt,
		location.EventID,
	)
	return err
}

func snapshotFieldsForEnvelope(env envelope.FrameEnvelope) map[string]any {
	if len(env.ParsedFields) == 0 {
		return nil
	}
	return realtimeSnapshotFlatFields(env)
}

func realtimeSnapshotFlatFields(env envelope.FrameEnvelope) map[string]any {
	fields := cloneMap(env.ParsedFields)
	filterInvalidRealtimeMeasurementFields(fields)
	if len(fields) == 0 {
		return nil
	}
	return fields
}

func platformNameFromEnvelope(env envelope.FrameEnvelope) string {
	if value := strings.TrimSpace(env.PlatformName); value != "" {
		return value
	}
	if env.Parsed != nil {
		if value := strings.TrimSpace(stringValue(env.Parsed["platform_name"])); value != "" {
			return value
		}
		if login, ok := env.Parsed["platform_login"].(map[string]any); ok {
			return strings.TrimSpace(stringValue(login["username"]))
		}
	}
	return ""
}

func stringValue(value any) string {
	switch typed := value.(type) {
	case string:
		return typed
	default:
		return ""
	}
}

func marshalParsedJSON(parsed map[string]any) any {
	if len(parsed) == 0 {
		return nil
	}
	data, err := json.Marshal(parsed)
	if err != nil {
		return nil
	}
	return string(data)
}

func (w *SnapshotWriter) plateForEnvelope(ctx context.Context, env envelope.FrameEnvelope) (string, error) {
	if plate := strings.TrimSpace(env.Plate); plate != "" {
		return plate, nil
	}
	if w.plateResolver == nil {
		return "", nil
	}
	vin := strings.TrimSpace(env.VIN)
	if vin == "" {
		return "", nil
	}
	plate, err := w.plateResolver.PlateByVIN(ctx, vin)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return "", nil
		}
		return "", err
	}
	return strings.TrimSpace(plate), nil
}

type realtimeLocationRow struct {
	Protocol       string
	VIN            string
	Plate          string
	EventTime      any
	Latitude       float64
	Longitude      float64
	SpeedKMH       any
	TotalMileageKM any
	TotalMileageAt any
	SOCPercent     any
	AltitudeM      any
	DirectionDeg   any
	AlarmFlag      any
	StatusFlag     any
	ReceivedAt     any
	EventID        string
}

func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vin string, plate string) (realtimeLocationRow, bool) {
	location, ok := telemetry.LocationProjectionForProtocol(env.Protocol, env.ParsedFields)
	if !ok {
		return realtimeLocationRow{}, false
	}
	eventTimeMS, _ := envelope.NormalizedEventTimeMS(env)
	totalMileageKM, hasTotalMileage := telemetry.TotalMileageKM(env.Protocol, env.ParsedFields)
	if totalMileageKM <= 0 {
		hasTotalMileage = false
	}
	var totalMileageValue any
	var totalMileageAt any
	if hasTotalMileage {
		totalMileageValue = totalMileageKM
		totalMileageAt = nullableTime(eventTimeMS)
	}
	return realtimeLocationRow{
		Protocol:       string(env.Protocol),
		VIN:            vin,
		Plate:          plate,
		EventTime:      nullableTime(eventTimeMS),
		Latitude:       location.Latitude,
		Longitude:      location.Longitude,
		SpeedKMH:       optionalFloatValue(location.SpeedKMH),
		TotalMileageKM: totalMileageValue,
		TotalMileageAt: totalMileageAt,
		SOCPercent:     optionalFloatValue(location.SOCPercent),
		AltitudeM:      optionalFloatValue(location.AltitudeM),
		DirectionDeg:   optionalFloatValue(location.DirectionDeg),
		AlarmFlag:      optionalInt64Value(location.AlarmFlag),
		StatusFlag:     optionalInt64Value(location.StatusFlag),
		ReceivedAt:     nullableTime(env.ReceivedAtMS),
		EventID:        env.StableEventID(),
	}, true
}

func optionalFloatValue(value *float64) any {
	if value == nil {
		return nil
	}
	return *value
}

func optionalInt64Value(value *int64) any {
	if value == nil {
		return nil
	}
	return *value
}

type BindingPlateResolver struct {
	queryer Queryer
	table   string
}

type Queryer interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

func NewBindingPlateResolver(queryer Queryer, table string) *BindingPlateResolver {
	if queryer == nil {
		panic("plate binding queryer must not be nil")
	}
	table = strings.TrimSpace(table)
	if table == "" || !safeIdentifier(table) {
		table = "vehicle_identity_binding"
	}
	return &BindingPlateResolver{queryer: queryer, table: table}
}

func (r *BindingPlateResolver) PlateByVIN(ctx context.Context, vin string) (string, error) {
	vin = strings.TrimSpace(vin)
	if vin == "" {
		return "", sql.ErrNoRows
	}
	query := "SELECT plate FROM " + r.table + " WHERE vin = ? AND plate IS NOT NULL AND plate <> ''"
	var plate string
	err := r.queryer.QueryRowContext(ctx, query, vin).Scan(&plate)
	if err != nil {
		return "", err
	}
	return strings.TrimSpace(plate), nil
}

func safeIdentifier(value string) bool {
	if value == "" {
		return false
	}
	for _, r := range value {
		if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' {
			continue
		}
		return false
	}
	return true
}

func nullableTime(ms int64) any {
	if ms <= 0 {
		return nil
	}
	return time.UnixMilli(ms)
}

func isRealtimeSnapshotEvent(env envelope.FrameEnvelope) bool {
	switch env.Protocol {
	case envelope.ProtocolGB32960:
		return (env.MessageID == "0x02" || env.MessageID == "0x03") && telemetry.HasRealtimeFields(env.Protocol, env.ParsedFields)
	case envelope.ProtocolJT808:
		return env.MessageID == "0x0200" && hasLocationFields(env)
	case envelope.ProtocolYutongMQTT:
		return telemetry.HasRealtimeFields(env.Protocol, env.ParsedFields)
	default:
		return false
	}
}

func hasLocationFields(env envelope.FrameEnvelope) bool {
	_, ok := telemetry.LocationProjectionForProtocol(env.Protocol, env.ParsedFields)
	return ok
}

const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot (
  protocol VARCHAR(32) NOT NULL,
  vin VARCHAR(32) NOT NULL DEFAULT '',
  plate VARCHAR(32) NOT NULL DEFAULT '',
  platform_name VARCHAR(64) NOT NULL DEFAULT '',
  peer VARCHAR(128) NOT NULL DEFAULT '',
  parsed_json LONGTEXT NULL,
  event_time DATETIME(3) NULL,
  received_at DATETIME(3) NULL,
  event_id VARCHAR(64) NOT NULL DEFAULT '',
  access_first_seen_at DATETIME(3) NULL,
  access_previous_received_at DATETIME(3) NULL,
  access_latest_received_at DATETIME(3) NULL,
  access_report_interval_ms BIGINT UNSIGNED NULL,
  access_sample_count BIGINT UNSIGNED NOT NULL DEFAULT 0,
  access_latest_event_id VARCHAR(64) NOT NULL DEFAULT '',
  access_first_seen_source VARCHAR(32) NOT NULL DEFAULT '',
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (protocol, vin),
  KEY idx_vin (vin),
  KEY idx_protocol_updated (protocol, updated_at),
  KEY idx_access_latest (access_latest_received_at, vin)
)`

var realtimeSnapshotCompatibilitySQL = []string{
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN platform_name VARCHAR(64) NOT NULL DEFAULT '' AFTER plate",
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN peer VARCHAR(128) NOT NULL DEFAULT '' AFTER platform_name",
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN parsed_json LONGTEXT NULL AFTER peer",
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN access_first_seen_at DATETIME(3) NULL AFTER event_id",
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN access_previous_received_at DATETIME(3) NULL AFTER access_first_seen_at",
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN access_latest_received_at DATETIME(3) NULL AFTER access_previous_received_at",
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN access_report_interval_ms BIGINT UNSIGNED NULL AFTER access_latest_received_at",
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN access_sample_count BIGINT UNSIGNED NOT NULL DEFAULT 0 AFTER access_report_interval_ms",
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN access_latest_event_id VARCHAR(64) NOT NULL DEFAULT '' AFTER access_sample_count",
	"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN access_first_seen_source VARCHAR(32) NOT NULL DEFAULT '' AFTER access_latest_event_id",
}

func isDuplicateColumnError(err error) bool {
	text := strings.ToLower(err.Error())
	return strings.Contains(text, "duplicate column") || strings.Contains(text, "1060")
}

const upsertRealtimeSnapshotSQL = `
INSERT INTO vehicle_realtime_snapshot
  (protocol, vin, plate, platform_name, peer, parsed_json, event_time, received_at, event_id,
   access_first_seen_at, access_latest_received_at, access_sample_count, access_latest_event_id, access_first_seen_source)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, 'live_writer')
ON DUPLICATE KEY UPDATE
  plate = IF(VALUES(plate) <> '', VALUES(plate), plate),
  platform_name = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(platform_name), platform_name),
  peer = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(peer), peer),
  parsed_json = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time),
    IF(VALUES(parsed_json) IS NULL OR VALUES(parsed_json) = '', parsed_json, JSON_MERGE_PATCH(COALESCE(NULLIF(vehicle_realtime_snapshot.parsed_json, ''), JSON_OBJECT()), VALUES(parsed_json))),
    parsed_json),
  event_time = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_time), event_time),
  received_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(received_at), received_at),
  event_id = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_id), event_id),
  updated_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), CURRENT_TIMESTAMP, updated_at),
  access_previous_received_at = IF(VALUES(access_latest_received_at) IS NOT NULL AND (vehicle_realtime_snapshot.access_latest_received_at IS NULL OR VALUES(access_latest_received_at) > vehicle_realtime_snapshot.access_latest_received_at) AND VALUES(access_latest_event_id) <> vehicle_realtime_snapshot.access_latest_event_id, vehicle_realtime_snapshot.access_latest_received_at, access_previous_received_at),
  access_report_interval_ms = IF(VALUES(access_latest_received_at) IS NOT NULL AND vehicle_realtime_snapshot.access_latest_received_at IS NOT NULL AND VALUES(access_latest_received_at) > vehicle_realtime_snapshot.access_latest_received_at AND VALUES(access_latest_event_id) <> vehicle_realtime_snapshot.access_latest_event_id, TIMESTAMPDIFF(MICROSECOND, vehicle_realtime_snapshot.access_latest_received_at, VALUES(access_latest_received_at)) DIV 1000, access_report_interval_ms),
  access_sample_count = IF(VALUES(access_latest_received_at) IS NOT NULL AND (vehicle_realtime_snapshot.access_latest_received_at IS NULL OR VALUES(access_latest_received_at) > vehicle_realtime_snapshot.access_latest_received_at) AND VALUES(access_latest_event_id) <> vehicle_realtime_snapshot.access_latest_event_id, access_sample_count + 1, access_sample_count),
  access_latest_received_at = IF(VALUES(access_latest_received_at) IS NOT NULL AND (vehicle_realtime_snapshot.access_latest_received_at IS NULL OR VALUES(access_latest_received_at) > vehicle_realtime_snapshot.access_latest_received_at) AND VALUES(access_latest_event_id) <> vehicle_realtime_snapshot.access_latest_event_id, VALUES(access_latest_received_at), access_latest_received_at),
  access_latest_event_id = IF(VALUES(access_latest_received_at) IS NOT NULL AND VALUES(access_latest_received_at) = vehicle_realtime_snapshot.access_latest_received_at AND VALUES(access_latest_event_id) <> vehicle_realtime_snapshot.access_latest_event_id, VALUES(access_latest_event_id), access_latest_event_id)
`

const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location (
  protocol VARCHAR(32) NOT NULL,
  vin VARCHAR(32) NOT NULL DEFAULT '',
  plate VARCHAR(32) NOT NULL DEFAULT '',
  event_time DATETIME(3) NULL,
  latitude DECIMAL(12,6) NOT NULL,
  longitude DECIMAL(12,6) NOT NULL,
  speed_kmh DECIMAL(10,3) NULL,
  total_mileage_km DECIMAL(18,3) NULL,
  total_mileage_event_time DATETIME(3) NULL,
  soc_percent DECIMAL(6,2) NULL,
  altitude_m DECIMAL(10,3) NULL,
  direction_deg DECIMAL(10,3) NULL,
  alarm_flag BIGINT NULL,
  status_flag BIGINT NULL,
  received_at DATETIME(3) NULL,
  event_id VARCHAR(64) NOT NULL DEFAULT '',
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (protocol, vin),
  KEY idx_vin (vin),
  KEY idx_protocol_updated (protocol, updated_at)
)`

var realtimeLocationCompatibilitySQL = []string{
	"ALTER TABLE vehicle_realtime_location ADD COLUMN total_mileage_event_time DATETIME(3) NULL AFTER total_mileage_km",
}

var cleanupInvalidRealtimeTotalMileageSQL = []string{
	`UPDATE vehicle_realtime_location
SET total_mileage_km = NULL,
    total_mileage_event_time = NULL
WHERE total_mileage_km IS NOT NULL
  AND total_mileage_km <= 0`,
	`UPDATE vehicle_realtime_snapshot
SET parsed_json = JSON_REMOVE(parsed_json, '$."jt808.location.total_mileage_km"')
WHERE protocol = 'JT808'
  AND parsed_json IS NOT NULL
  AND CAST(JSON_UNQUOTE(JSON_EXTRACT(parsed_json, '$."jt808.location.total_mileage_km"')) AS DECIMAL(18,3)) <= 0`,
	`UPDATE vehicle_realtime_snapshot
SET parsed_json = JSON_REMOVE(parsed_json, '$."gb32960.vehicle.total_mileage_km"')
WHERE protocol = 'GB32960'
  AND parsed_json IS NOT NULL
  AND CAST(JSON_UNQUOTE(JSON_EXTRACT(parsed_json, '$."gb32960.vehicle.total_mileage_km"')) AS DECIMAL(18,3)) <= 0`,
	`UPDATE vehicle_realtime_snapshot
SET parsed_json = JSON_REMOVE(parsed_json, '$."yutong_mqtt.data.total_mileage_km"')
WHERE protocol = 'YUTONG_MQTT'
  AND parsed_json IS NOT NULL
	  AND CAST(JSON_UNQUOTE(JSON_EXTRACT(parsed_json, '$."yutong_mqtt.data.total_mileage_km"')) AS DECIMAL(18,3)) <= 0`,
}

const backfillRealtimeAccessProjectionSQL = `UPDATE vehicle_realtime_snapshot
SET access_first_seen_at = COALESCE(access_first_seen_at, received_at, updated_at),
    access_latest_received_at = COALESCE(access_latest_received_at, received_at, updated_at),
    access_sample_count = IF(access_sample_count = 0, 1, access_sample_count),
    access_latest_event_id = IF(access_latest_event_id = '', event_id, access_latest_event_id),
    access_first_seen_source = IF(access_first_seen_source = '', 'snapshot_backfill', access_first_seen_source)
WHERE access_first_seen_at IS NULL
   OR access_latest_received_at IS NULL
   OR access_sample_count = 0
   OR access_latest_event_id = ''
   OR access_first_seen_source = ''`

const upsertRealtimeLocationSQL = `
INSERT INTO vehicle_realtime_location
  (protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, total_mileage_event_time,
   soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
  plate = IF(VALUES(plate) <> '', VALUES(plate), plate),
  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), VALUES(event_time), event_time),
  latitude = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(latitude), latitude),
  longitude = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(longitude), longitude),
  speed_kmh = 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(speed_kmh), speed_kmh), speed_kmh),
  total_mileage_km = 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(total_mileage_km), total_mileage_km), total_mileage_km),
  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),
  soc_percent = 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(soc_percent), soc_percent), soc_percent),
  altitude_m = 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(altitude_m), altitude_m), altitude_m),
  direction_deg = 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(direction_deg), direction_deg), direction_deg),
  alarm_flag = 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(alarm_flag), alarm_flag), alarm_flag),
  status_flag = 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(status_flag), status_flag), status_flag),
  received_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(received_at), received_at),
  event_id = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), VALUES(event_id), event_id),
  updated_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_location.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location.event_time), CURRENT_TIMESTAMP, updated_at)
`