package realtime import ( "context" "encoding/json" "errors" "sort" "strconv" "strings" "time" "github.com/redis/go-redis/v9" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" ) type Repository struct { client *redis.Client cfg Config } func NewRepository(client *redis.Client, cfg Config) *Repository { if client == nil { panic("redis client must not be nil") } return &Repository{client: client, cfg: cfg} } func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error { vin := strings.TrimSpace(env.VIN) vehicleKey := strings.TrimSpace(env.VehicleKey()) if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") { return nil } nowMS := time.Now().UnixMilli() eventMS := env.EventTimeMS if eventMS <= 0 { eventMS = env.ReceivedAtMS } protocolSnapshot := Snapshot{ VehicleKey: vehicleKey, VIN: vin, Protocol: env.Protocol, EventID: env.StableEventID(), EventTimeMS: eventMS, ReceivedAtMS: env.ReceivedAtMS, SourceEndpoint: env.SourceEndpoint, Fields: cloneFields(env.Fields), FieldTimesMS: fieldTimes(env.Fields, eventMS), UpdatedAtMS: nowMS, } if err := r.setJSON(ctx, protocolKey(vehicleKey, env.Protocol), protocolSnapshot, r.cfg.ttl()); err != nil { return err } merged, err := r.GetMerged(ctx, vehicleKey) if err != nil && !errors.Is(err, redis.Nil) { return err } if merged.VehicleKey == "" { merged = Snapshot{VehicleKey: vehicleKey, VIN: vin, Fields: map[string]any{}, FieldTimesMS: map[string]int64{}} } else if merged.VIN == "" && vin != "" { merged.VIN = vin } mergeFields(&merged, env.Fields, eventMS) if eventMS >= merged.EventTimeMS { merged.EventTimeMS = eventMS merged.EventID = env.StableEventID() merged.ReceivedAtMS = env.ReceivedAtMS merged.SourceEndpoint = env.SourceEndpoint } merged.UpdatedAtMS = nowMS if err := r.setJSON(ctx, mergedKey(vehicleKey), merged, r.cfg.ttl()); err != nil { return err } protocols, err := r.addProtocol(ctx, vehicleKey, env.Protocol) if err != nil { return err } online := OnlineStatus{ VehicleKey: vehicleKey, VIN: vin, Online: true, LastSeenMS: env.ReceivedAtMS, Protocols: protocols, TTLSeconds: int64(r.cfg.ttl().Seconds()), } if err := r.setJSON(ctx, onlineKey(vehicleKey), online, r.cfg.ttl()); err != nil { return err } return r.client.ZAdd(ctx, "vehicle:last_seen", redis.Z{Score: float64(env.ReceivedAtMS), Member: vehicleKey}).Err() } func (r *Repository) GetMerged(ctx context.Context, vehicleKey string) (Snapshot, error) { return r.getSnapshot(ctx, mergedKey(vehicleKey)) } func (r *Repository) GetProtocol(ctx context.Context, vehicleKey string, protocol envelope.Protocol) (Snapshot, error) { return r.getSnapshot(ctx, protocolKey(vehicleKey, protocol)) } func (r *Repository) IsOnline(ctx context.Context, vehicleKey string) (OnlineStatus, error) { vehicleKey = strings.TrimSpace(vehicleKey) var status OnlineStatus payload, err := r.client.Get(ctx, onlineKey(vehicleKey)).Bytes() if err != nil { if errors.Is(err, redis.Nil) { return OnlineStatus{VehicleKey: vehicleKey, VIN: vehicleKey, Online: false}, nil } return status, err } if err := json.Unmarshal(payload, &status); err != nil { return status, err } ttl := r.client.TTL(ctx, onlineKey(vehicleKey)).Val() status.Online = ttl > 0 status.TTLSeconds = int64(ttl.Seconds()) return status, nil } func (r *Repository) getSnapshot(ctx context.Context, key string) (Snapshot, error) { var snapshot Snapshot payload, err := r.client.Get(ctx, key).Bytes() if err != nil { return snapshot, err } return snapshot, json.Unmarshal(payload, &snapshot) } func (r *Repository) setJSON(ctx context.Context, key string, value any, ttl time.Duration) error { payload, err := json.Marshal(value) if err != nil { return err } return r.client.Set(ctx, key, payload, ttl).Err() } func (r *Repository) addProtocol(ctx context.Context, vehicleKey string, protocol envelope.Protocol) ([]envelope.Protocol, error) { key := protocolsKey(vehicleKey) if err := r.client.SAdd(ctx, key, string(protocol)).Err(); err != nil { return nil, err } _ = r.client.Expire(ctx, key, r.cfg.ttl()).Err() values, err := r.client.SMembers(ctx, key).Result() if err != nil { return nil, err } sort.Strings(values) out := make([]envelope.Protocol, 0, len(values)) for _, value := range values { out = append(out, envelope.Protocol(value)) } return out, nil } func mergeFields(snapshot *Snapshot, fields map[string]any, eventMS int64) { if snapshot.Fields == nil { snapshot.Fields = map[string]any{} } if snapshot.FieldTimesMS == nil { snapshot.FieldTimesMS = map[string]int64{} } for key, value := range fields { if key == envelope.FieldTotalMileageKM && !positiveNumber(value) { continue } if eventMS >= snapshot.FieldTimesMS[key] { snapshot.Fields[key] = value snapshot.FieldTimesMS[key] = eventMS } } } func positiveNumber(value any) bool { switch typed := value.(type) { case float64: return typed > 0 case float32: return typed > 0 case int: return typed > 0 case int64: return typed > 0 case uint16: return typed > 0 case uint32: return typed > 0 case string: parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64) return err == nil && parsed > 0 default: return false } } func cloneFields(fields map[string]any) map[string]any { if len(fields) == 0 { return map[string]any{} } out := make(map[string]any, len(fields)) for key, value := range fields { out[key] = value } return out } func fieldTimes(fields map[string]any, eventMS int64) map[string]int64 { out := make(map[string]int64, len(fields)) for key := range fields { out[key] = eventMS } return out } func mergedKey(vehicleKey string) string { return "vehicle:latest:" + strings.TrimSpace(vehicleKey) } func protocolKey(vehicleKey string, protocol envelope.Protocol) string { return "vehicle:latest:" + strings.TrimSpace(vehicleKey) + ":" + string(protocol) } func onlineKey(vehicleKey string) string { return "vehicle:online:" + strings.TrimSpace(vehicleKey) } func protocolsKey(vehicleKey string) string { return "vehicle:protocols:" + strings.TrimSpace(vehicleKey) }