feat: add go realtime redis api

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

@@ -0,0 +1,198 @@
+package realtime
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sort"
+	"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)
+	if vin == "" {
+		return nil
+	}
+	nowMS := time.Now().UnixMilli()
+	eventMS := env.EventTimeMS
+	if eventMS <= 0 {
+		eventMS = env.ReceivedAtMS
+	}
+	protocolSnapshot := Snapshot{
+		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(vin, env.Protocol), protocolSnapshot, r.cfg.ttl()); err != nil {
+		return err
+	}
+
+	merged, err := r.GetMerged(ctx, vin)
+	if err != nil && !errors.Is(err, redis.Nil) {
+		return err
+	}
+	if merged.VIN == "" {
+		merged = Snapshot{VIN: vin, Fields: map[string]any{}, FieldTimesMS: map[string]int64{}}
+	}
+	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(vin), merged, r.cfg.ttl()); err != nil {
+		return err
+	}
+
+	protocols, err := r.addProtocol(ctx, vin, env.Protocol)
+	if err != nil {
+		return err
+	}
+	online := OnlineStatus{
+		VIN:        vin,
+		Online:     true,
+		LastSeenMS: env.ReceivedAtMS,
+		Protocols:  protocols,
+		TTLSeconds: int64(r.cfg.ttl().Seconds()),
+	}
+	if err := r.setJSON(ctx, onlineKey(vin), online, r.cfg.ttl()); err != nil {
+		return err
+	}
+	return r.client.ZAdd(ctx, "vehicle:last_seen", redis.Z{Score: float64(env.ReceivedAtMS), Member: vin}).Err()
+}
+
+func (r *Repository) GetMerged(ctx context.Context, vin string) (Snapshot, error) {
+	return r.getSnapshot(ctx, mergedKey(vin))
+}
+
+func (r *Repository) GetProtocol(ctx context.Context, vin string, protocol envelope.Protocol) (Snapshot, error) {
+	return r.getSnapshot(ctx, protocolKey(vin, protocol))
+}
+
+func (r *Repository) IsOnline(ctx context.Context, vin string) (OnlineStatus, error) {
+	var status OnlineStatus
+	payload, err := r.client.Get(ctx, onlineKey(vin)).Bytes()
+	if err != nil {
+		if errors.Is(err, redis.Nil) {
+			return OnlineStatus{VIN: vin, Online: false}, nil
+		}
+		return status, err
+	}
+	if err := json.Unmarshal(payload, &status); err != nil {
+		return status, err
+	}
+	ttl := r.client.TTL(ctx, onlineKey(vin)).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, vin string, protocol envelope.Protocol) ([]envelope.Protocol, error) {
+	key := protocolsKey(vin)
+	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 eventMS >= snapshot.FieldTimesMS[key] {
+			snapshot.Fields[key] = value
+			snapshot.FieldTimesMS[key] = eventMS
+		}
+	}
+}
+
+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(vin string) string {
+	return "vehicle:latest:" + strings.TrimSpace(vin)
+}
+
+func protocolKey(vin string, protocol envelope.Protocol) string {
+	return "vehicle:latest:" + strings.TrimSpace(vin) + ":" + string(protocol)
+}
+
+func onlineKey(vin string) string {
+	return "vehicle:online:" + strings.TrimSpace(vin)
+}
+
+func protocolsKey(vin string) string {
+	return "vehicle:protocols:" + strings.TrimSpace(vin)
+}