Server: rewrite in Go as an API-only service
Same endpoints, request formats, validation rules and InfluxDB layout as
the Next.js server, so the car needs no change: POST /v1/telemetry,
PUT /v1/consumption/daily, PUT /v1/charging-sessions/{id}, GET
/v1/consumption/history/{id}, GET /v1/health. The dashboard, login and
the /v1/history reads are gone; Grafana takes over the dashboards.
- Go 1.26 standard library (net/http routing, log/slog) plus the
official InfluxDB 2 client. Fails closed at startup on a missing
InfluxDB setting or an ingest token under 24 characters.
- ABRP forwarder ported: newest fresh sample, rate limited, never
queued, credentials only in the form body.
- 22 tests: validation, aggregation, forwarding, config, and the
handlers through a fake store. govulncheck clean (x/net 0.60.0,
toolchain go1.26.9).
- Image: static binary on distroless, about 11 MB, UID 65532, port
8080; the build runs vet and the tests.
- Dev shell: Go, gopls and govulncheck instead of Node.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QLJfZQZeAKAwSqViTmLzki
This commit is contained in:
+162
@@ -0,0 +1,162 @@
|
||||
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)
|
||||
}
|
||||
|
||||
// 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
|
||||
//
|
||||
// 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
|
||||
}
|
||||
Reference in New Issue
Block a user