mg4/server/store.go

package main

import (
	"context"
	"fmt"
	"strings"
	"time"

	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
	"github.com/influxdata/influxdb-client-go/v2/api"
	"github.com/influxdata/influxdb-client-go/v2/api/write"
)

// Store is what the handlers need from the database; a fake implements it in tests.
type Store interface {
	WriteTelemetry(ctx context.Context, installationID string, samples []TelemetrySample) error
	WriteConsumptionChunk(ctx context.Context, chunk ConsumptionChunk) error
	WriteChargingSession(ctx context.Context, session ChargingSession) error
	ReadConsumptionChunks(ctx context.Context, installationID string) ([]ConsumptionChunk, error)
	SaveDevice(ctx context.Context, d Device) error
	LoadDevices(ctx context.Context) ([]Device, error)
}

// InfluxStore keeps everything in one InfluxDB 2 bucket; every point carries an
// installationId tag. Grafana reads the same bucket directly.
//
//	telemetry          time = sample utc                  ABRP tlm fields (floats and booleans)
//	consumption_chunk  time = recordedAt   tags: date     km, kwh, hours, socDrop, sohPercent
//	charging_session   time = startedAt    tags: sessionId summary fields
//	charging_curve     time = point time   tags: sessionId powerKw, socPercent
//	device             time = createdAt    tags: deviceId  tokenHash, installationId, revoked
//
// Idempotency comes from InfluxDB itself: a point with the same measurement, tags and
// timestamp replaces the earlier one. The car re-sends a cached chunk with its original
// recordedAt and a session under its own id, so a retry overwrites instead of adding.
//
// Every number is written as a float. InfluxDB fixes a field's type on first write, so
// a 0 written as an integer would make every later 0.5 a rejected write.
type InfluxStore struct {
	client influxdb2.Client
	write  api.WriteAPIBlocking
	query  api.QueryAPI
	bucket string
}

const tagInstallation = "installationId"

func NewInfluxStore(url, token, org, bucket string) *InfluxStore {
	client := influxdb2.NewClientWithOptions(url, token,
		influxdb2.DefaultOptions().SetHTTPRequestTimeout(10).SetPrecision(time.Millisecond))
	return &InfluxStore{
		client: client,
		write:  client.WriteAPIBlocking(org, bucket),
		query:  client.QueryAPI(org),
		bucket: bucket,
	}
}

func (s *InfluxStore) Close() { s.client.Close() }

func (s *InfluxStore) WriteTelemetry(ctx context.Context, id string, samples []TelemetrySample) error {
	points := make([]*write.Point, 0, len(samples))
	for _, sample := range samples {
		fields := make(map[string]any, len(sample.Numbers)+len(sample.Flags))
		for k, v := range sample.Numbers {
			fields[k] = v
		}
		for k, v := range sample.Flags {
			fields[k] = v
		}
		if len(fields) == 0 {
			// A point needs at least one field; a bare timestamp says nothing anyway.
			continue
		}
		points = append(points, influxdb2.NewPoint("telemetry",
			map[string]string{tagInstallation: id}, fields, time.Unix(sample.UTC, 0)))
	}
	if len(points) == 0 {
		return nil
	}
	return s.write.WritePoint(ctx, points...)
}

func (s *InfluxStore) WriteConsumptionChunk(ctx context.Context, c ConsumptionChunk) error {
	fields := map[string]any{"km": c.Day.Km, "kwh": c.Day.Kwh, "hours": c.Day.Hours, "socDrop": c.Day.SocDrop}
	if c.SohPercent != nil {
		fields["sohPercent"] = *c.SohPercent
	}
	return s.write.WritePoint(ctx, influxdb2.NewPoint("consumption_chunk",
		map[string]string{tagInstallation: c.InstallationID, "date": c.Date}, fields, c.RecordedAt))
}

func (s *InfluxStore) WriteChargingSession(ctx context.Context, cs ChargingSession) error {
	tags := map[string]string{tagInstallation: cs.InstallationID, "sessionId": cs.SessionID}
	summary := map[string]any{
		"endedAtMs":           cs.EndedAt.UnixMilli(),
		"durationSeconds":     cs.DurationSeconds,
		"energyKwh":           cs.EnergyKwh,
		"chargingLossPercent": cs.ChargingLossPercent,
		"gridEnergyKwh":       cs.GridEnergyKwh,
		"pricePerKwh":         cs.PricePerKwh,
		"totalCost":           cs.TotalCost,
		"isDemo":              cs.IsDemo,
	}
	if cs.StartSocPercent != nil {
		summary["startSocPercent"] = *cs.StartSocPercent
	}
	if cs.EndSocPercent != nil {
		summary["endSocPercent"] = *cs.EndSocPercent
	}
	points := []*write.Point{influxdb2.NewPoint("charging_session", tags, summary, cs.StartedAt)}
	for _, p := range cs.Curve {
		fields := map[string]any{"powerKw": p.PowerKw}
		if p.SocPercent != nil {
			fields["socPercent"] = *p.SocPercent
		}
		points = append(points, influxdb2.NewPoint("charging_curve", tags, fields, p.Timestamp))
	}
	return s.write.WritePoint(ctx, points...)
}

// fluxString quotes a value as a Flux string literal. Ids are validated UUIDs before they
// get here; the escaping is for the bucket name and as a second line of defence.
func fluxString(value string) string {
	return `"` + strings.NewReplacer(`\`, `\\`, `"`, `\"`, "$", `\$`).Replace(value) + `"`
}

func (s *InfluxStore) ReadConsumptionChunks(ctx context.Context, id string) ([]ConsumptionChunk, error) {
	query := fmt.Sprintf(`from(bucket: %s)
  |> range(start: 0)
  |> filter(fn: (r) => r._measurement == "consumption_chunk" and r.installationId == %s)
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")`,
		fluxString(s.bucket), fluxString(id))
	result, err := s.query.Query(ctx, query)
	if err != nil {
		return nil, err
	}
	defer result.Close()
	var chunks []ConsumptionChunk
	for result.Next() {
		r := result.Record()
		date, _ := r.ValueByKey("date").(string)
		chunk := ConsumptionChunk{
			InstallationID: id,
			Date:           date,
			RecordedAt:     r.Time(),
			Day: Totals{
				Km:      floatValue(r.ValueByKey("km")),
				Kwh:     floatValue(r.ValueByKey("kwh")),
				Hours:   floatValue(r.ValueByKey("hours")),
				SocDrop: floatValue(r.ValueByKey("socDrop")),
			},
		}
		if soh, ok := r.ValueByKey("sohPercent").(float64); ok {
			chunk.SohPercent = &soh
		}
		chunks = append(chunks, chunk)
	}
	return chunks, result.Err()
}

func floatValue(v any) float64 {
	f, _ := v.(float64)
	return f
}

// SaveDevice writes a paired car. Re-saving it (a revocation) overwrites the same point,
// because measurement, tag and timestamp stay the same.
func (s *InfluxStore) SaveDevice(ctx context.Context, d Device) error {
	return s.write.WritePoint(ctx, influxdb2.NewPoint("device",
		map[string]string{"deviceId": d.ID},
		map[string]any{"tokenHash": d.TokenHash, "installationId": d.InstallationID, "revoked": d.Revoked},
		d.CreatedAt))
}

func (s *InfluxStore) LoadDevices(ctx context.Context) ([]Device, error) {
	query := fmt.Sprintf(`from(bucket: %s)
  |> range(start: 0)
  |> filter(fn: (r) => r._measurement == "device")
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")`, fluxString(s.bucket))
	result, err := s.query.Query(ctx, query)
	if err != nil {
		return nil, err
	}
	defer result.Close()
	var devices []Device
	for result.Next() {
		r := result.Record()
		id, _ := r.ValueByKey("deviceId").(string)
		hash, _ := r.ValueByKey("tokenHash").(string)
		installation, _ := r.ValueByKey("installationId").(string)
		revoked, _ := r.ValueByKey("revoked").(bool)
		if id == "" || hash == "" {
			continue
		}
		devices = append(devices, Device{ID: id, InstallationID: installation, TokenHash: hash,
			CreatedAt: r.Time(), Revoked: revoked})
	}
	return devices, result.Err()
}