Server 2.4.0, app 2.11.1: ABRP forwarding removed from the server

The car sends live data to ABRP itself, and the car and the server go
over the same mobile connection: when the car cannot reach ABRP, it
cannot reach the server either, so the server's fallback bought
nothing. The forwarder, its ABRP_* settings and abrpSentUtc are gone;
the server only stores. The app stops sending abrpSentUtc (the server
ignores unknown fields, so 2.11.0 keeps working) and its texts no longer
promise a server fallback.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QLJfZQZeAKAwSqViTmLzki

server/README.md

@@ -1,11 +1,11 @@
 # Nils' MG4 — car server
 
-A small Go service at car.k8s.stinnesbeck.com: the API the car writes to, and the ABRP
-forwarder. Data lives in InfluxDB 2; Grafana reads it from there for the dashboards
-(`../k8s/grafana/`).
+A small Go service at car.k8s.stinnesbeck.com: the API the car writes to. Data lives in
+InfluxDB 2; Grafana reads it from there for the dashboards (`../k8s/grafana/`). The car sends
+its live data to ABRP itself; the server does not talk to ABRP (removed in 2.4.0).
 
 ```
-car ──live telemetry───▶ server ──▶ ABRP  api.iternio.com/1/tlm/send
+car ──live telemetry───▶ server
     ──daily chunks─────▶   │
     ──charging sessions▶   ▼
                        InfluxDB 2 ◀── Grafana
@@ -20,14 +20,10 @@ Environment only; the server refuses to start when a required value is missing.
 | `INFLUX_URL`, `INFLUX_TOKEN`, `INFLUX_ORG` | required; the token needs read/write on the bucket |
 | `INFLUX_BUCKET` | default `mg4` |
 | `INGEST_TOKEN` | required, at least 24 characters: `openssl rand -hex 32` |
-| `ABRP_API_KEY`, `ABRP_USER_TOKEN` | optional; both needed to forward |
-| `ABRP_MIN_INTERVAL_SECONDS` | default 5 |
 | `APPROVAL_PASSWORD` | optional, at least 12 characters; switches pairing on (see below) |
 | `PUBLIC_URL` | the address a phone uses, for the pairing QR code; derived from the request if empty |
 | `LISTEN_ADDR` | default `:8080` |
 
-The ABRP credentials never reach the car.
-
 ## Endpoints
 
 The ingest endpoints accept `Authorization: Bearer <token>` with either the shared `INGEST_TOKEN`
@@ -39,10 +35,7 @@ with (403 otherwise); the shared token works for any.
   dropped, known ones bounded: a value out of range is dropped on its own, and a sample without
   a plausible time (before 2001, or more than an hour ahead) is dropped whole. The rest is stored
   and the answer is `200 {stored, droppedCount, dropped}`, even when nothing was usable, since
-  the car would only send it again. The newest sample is forwarded to ABRP if it is under
-  5 minutes old and not ahead of the server's clock. A car that sends to ABRP itself adds
-  `abrpSentUtc`, the newest sample ABRP took from it; only samples newer than that are
-  forwarded, so ABRP gets each one once.
+  the car would only send it again.
 - `PUT /v1/consumption/daily` — one delta chunk of a day's consumption.
 - `PUT /v1/charging-sessions/{sessionId}` — a finished DC session with its curve.
 - `GET /v1/consumption/history/{installationId}` — the car reads its own totals back.

server/abrp.go

@@ -1,142 +0,0 @@
-package main
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"log/slog"
-	"net/http"
-	"net/url"
-	"strings"
-	"sync"
-	"time"
-)
-
-// Forwarder sends live telemetry on to A Better Route Planner.
-//
-// The ABRP credentials exist only here, from ABRP_API_KEY and ABRP_USER_TOKEN in the
-// server's environment; the car never holds them. Forwarding is off when either is empty.
-//
-// Only the newest sample of a batch is sent, and at most one per MinInterval. A failed
-// forward is logged and dropped, never queued: ABRP plans from the car's state now, and a
-// backlog replayed later would describe a car that has moved on. The stored copy in
-// InfluxDB is unaffected either way.
-type Forwarder struct {
-	APIKey      string
-	UserToken   string
-	MinInterval time.Duration
-	Endpoint    string
-	Client      *http.Client
-
-	mu         sync.Mutex
-	lastUTC    int64
-	lastSentAt time.Time
-}
-
-const (
-	abrpEndpoint = "https://api.iternio.com/1/tlm/send"
-	// Older than this, a sample is history rather than live state, and ABRP gets nothing.
-	abrpMaxAge = 5 * time.Minute
-	// Newer than now by more than this, a sample's clock is wrong, and ABRP gets nothing.
-	abrpMaxAhead = 30 * time.Second
-)
-
-func (f *Forwarder) configured() bool {
-	return f != nil && f.APIKey != "" && f.UserToken != ""
-}
-
-// pick chooses the sample to forward, or false when there is nothing new or it is too
-// soon. It records the choice, so two batches arriving together cannot both forward.
-func (f *Forwarder) pick(samples []TelemetrySample, now time.Time) (TelemetrySample, bool) {
-	// A sample stamped ahead of now has a wrong clock. Taking it would also set lastUTC into
-	// the future, and every real sample after it would look older and never be sent.
-	latest := now.Add(abrpMaxAhead).Unix()
-	var newest TelemetrySample
-	found := false
-	for _, s := range samples {
-		if s.UTC <= latest && (!found || s.UTC > newest.UTC) {
-			newest, found = s, true
-		}
-	}
-	if !found || now.Sub(time.Unix(newest.UTC, 0)) > abrpMaxAge {
-		return TelemetrySample{}, false
-	}
-	f.mu.Lock()
-	defer f.mu.Unlock()
-	if f.lastUTC != 0 && newest.UTC <= f.lastUTC {
-		return TelemetrySample{}, false
-	}
-	if !f.lastSentAt.IsZero() && now.Sub(f.lastSentAt) < f.MinInterval {
-		return TelemetrySample{}, false
-	}
-	f.lastUTC = newest.UTC
-	f.lastSentAt = now
-	return newest, true
-}
-
-// Forward is fire-and-forget: the car's request is answered without waiting for ABRP.
-func (f *Forwarder) Forward(samples []TelemetrySample) {
-	if !f.configured() {
-		return
-	}
-	sample, ok := f.pick(samples, time.Now())
-	if !ok {
-		return
-	}
-	go func() {
-		if err := f.send(context.Background(), sample); err != nil {
-			slog.Warn("ABRP forward failed", "error", err)
-		}
-	}()
-}
-
-// tlm is the object ABRP expects: the same field names, flags as 0/1.
-func tlm(sample TelemetrySample) map[string]any {
-	out := map[string]any{"utc": sample.UTC}
-	for field, value := range sample.Numbers {
-		out[field] = value
-	}
-	for field, value := range sample.Flags {
-		if value {
-			out[field] = 1
-		} else {
-			out[field] = 0
-		}
-	}
-	return out
-}
-
-func (f *Forwarder) send(ctx context.Context, sample TelemetrySample) error {
-	payload, err := json.Marshal(tlm(sample))
-	if err != nil {
-		return err
-	}
-	// Credentials go in the form body, never the URL, which ends up in access logs.
-	form := url.Values{"api_key": {f.APIKey}, "token": {f.UserToken}, "tlm": {string(payload)}}
-	endpoint := f.Endpoint
-	if endpoint == "" {
-		endpoint = abrpEndpoint
-	}
-	ctx, cancel := context.WithTimeout(ctx, 8*time.Second)
-	defer cancel()
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, strings.NewReader(form.Encode()))
-	if err != nil {
-		return err
-	}
-	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
-	client := f.Client
-	if client == nil {
-		client = http.DefaultClient
-	}
-	resp, err := client.Do(req)
-	if err != nil {
-		// A url.Error from the client quotes the URL, which holds no credentials.
-		return err
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		// The response body can echo the request; only the status is reported.
-		return fmt.Errorf("rejected: HTTP %d", resp.StatusCode)
-	}
-	return nil
-}

server/api.go

@@ -13,7 +13,7 @@ import (
 
 // API serves the car. Every endpoint but /v1/health needs the ingest token.
 //
-//	POST /v1/telemetry                           live samples; stored, newest forwarded to ABRP
+//	POST /v1/telemetry                           live samples, stored
 //	PUT  /v1/consumption/daily                   one delta chunk of a day's consumption
 //	PUT  /v1/charging-sessions/{sessionId}       one finished DC charging session
 //	GET  /v1/consumption/history/{installationId} the car reads its own totals back
@@ -24,7 +24,6 @@ import (
 type API struct {
 	Store       Store
 	IngestToken string
-	Forwarder   *Forwarder
 	Pairing     *Pairing
 	// PublicURL is the server's address as a phone reaches it, for the pairing QR code;
 	// derived from the request when empty.
@@ -117,27 +116,11 @@ func (a *API) ingest(h handler) http.HandlerFunc {
 	}
 }
 
-// notYetAtAbrp leaves out the samples the car already sent to ABRP itself.
-func notYetAtAbrp(samples []TelemetrySample, sentUTC int64) []TelemetrySample {
-	if sentUTC == 0 {
-		return samples
-	}
-	out := make([]TelemetrySample, 0, len(samples))
-	for _, s := range samples {
-		if s.UTC > sentUTC {
-			out = append(out, s)
-		}
-	}
-	return out
-}
-
 // maxDroppedListed caps the dropped values named in a telemetry response and log line.
 const maxDroppedListed = 20
 
-// telemetry stores the batch first; only once the write succeeded is the newest sample
-// handed to ABRP, so the car's retry of a failed request cannot leave ABRP with data the
-// database never got. The answer says how many samples were stored and what was dropped;
-// a batch with nothing usable is still a 200, since sending it again would not help.
+// telemetry stores the batch. The answer says how many samples were stored and what was
+// dropped; a batch with nothing usable is still a 200, since sending it again would not help.
 func (a *API) telemetry(w http.ResponseWriter, r *http.Request, c caller, body []byte) error {
 	batch, err := validateTelemetryBatch(body, time.Now())
 	if err != nil {
@@ -150,7 +133,6 @@ func (a *API) telemetry(w http.ResponseWriter, r *http.Request, c caller, body [
 		if err := a.Store.WriteTelemetry(r.Context(), batch.InstallationID, batch.Samples); err != nil {
 			return err
 		}
-		a.Forwarder.Forward(notYetAtAbrp(batch.Samples, batch.AbrpSentUTC))
 	}
 	dropped := batch.Dropped
 	if len(dropped) > maxDroppedListed {

server/api_test.go

@@ -76,7 +76,7 @@ const (
 
 func newTestAPI() (*API, *fakeStore) {
 	store := &fakeStore{}
-	return &API{Store: store, IngestToken: testToken, Forwarder: &Forwarder{},
+	return &API{Store: store, IngestToken: testToken,
 		Pairing: NewPairing(testApprovalPassword, nil), PublicURL: "https://car.example.com"}, store
 }
 
@@ -204,23 +204,3 @@ func jsonInt(v int64) string {
 	b, _ := json.Marshal(v)
 	return string(b)
 }
-
-func TestSamplesTheCarSentToAbrpAreNotForwardedAgain(t *testing.T) {
-	samples := []TelemetrySample{{UTC: 100}, {UTC: 115}, {UTC: 130}}
-	if got := notYetAtAbrp(samples, 0); len(got) != 3 {
-		t.Fatalf("without abrpSentUtc everything is forwardable: %d", len(got))
-	}
-	if got := notYetAtAbrp(samples, 130); len(got) != 0 {
-		t.Fatalf("the car sent the newest: %d left", len(got))
-	}
-	if got := notYetAtAbrp(samples, 115); len(got) != 1 || got[0].UTC != 130 {
-		t.Fatalf("only the newer sample: %+v", got)
-	}
-	b, err := validateTelemetryBatch([]byte(`{"installationId":"`+testInstallation+`","abrpSentUtc":1790000000,"samples":[{"utc":1790000000}]}`), testNow)
-	if err != nil || b.AbrpSentUTC != 1790000000 {
-		t.Fatalf("abrpSentUtc not read: %v %d", err, b.AbrpSentUTC)
-	}
-	if _, err := validateTelemetryBatch([]byte(`{"installationId":"`+testInstallation+`","abrpSentUtc":"x","samples":[{"utc":1790000000}]}`), testNow); !isValidationError(err) {
-		t.Fatal("a malformed abrpSentUtc was accepted")
-	}
-}

server/logic_test.go

@@ -1,10 +1,6 @@
 package main
 
 import (
-	"io"
-	"net/http"
-	"net/http/httptest"
-	"net/url"
 	"testing"
 	"time"
 )
@@ -49,68 +45,6 @@ func TestAggregateDaysNewestSohWins(t *testing.T) {
 	}
 }
 
-func TestTlmSendsFlagsAsZeroOrOne(t *testing.T) {
-	got := tlm(TelemetrySample{UTC: testNowS, Numbers: map[string]float64{"soc": 50},
-		Flags: map[string]bool{"is_charging": true, "is_parked": false}})
-	if got["utc"] != int64(testNowS) || got["soc"] != 50.0 || got["is_charging"] != 1 || got["is_parked"] != 0 {
-		t.Fatalf("tlm %+v", got)
-	}
-}
-
-func TestForwarderPicksOnlyTheNewestFreshSampleOncePerInterval(t *testing.T) {
-	f := &Forwarder{MinInterval: 5 * time.Second}
-	now := time.Unix(testNowS, 0)
-	got, ok := f.pick([]TelemetrySample{{UTC: testNowS - 10}, {UTC: testNowS - 2}, {UTC: testNowS - 5}}, now)
-	if !ok || got.UTC != testNowS-2 {
-		t.Fatalf("picked %+v %v", got, ok)
-	}
-	if _, ok := f.pick([]TelemetrySample{{UTC: testNowS}}, now.Add(time.Second)); ok {
-		t.Fatal("forwarded again within the interval")
-	}
-	if _, ok := f.pick([]TelemetrySample{{UTC: testNowS - 3}}, now.Add(time.Minute)); ok {
-		t.Fatal("forwarded an older sample than the last one")
-	}
-	// A sample from ten minutes ago describes a car that has moved on.
-	fresh := &Forwarder{MinInterval: 5 * time.Second}
-	if _, ok := fresh.pick([]TelemetrySample{{UTC: testNowS - 600}}, now); ok {
-		t.Fatal("forwarded a stale sample")
-	}
-	// A sample from a clock a day ahead is skipped, and does not block the real ones after it.
-	ahead := &Forwarder{MinInterval: 5 * time.Second}
-	got, ok = ahead.pick([]TelemetrySample{{UTC: testNowS + 86400}, {UTC: testNowS - 1}}, now)
-	if !ok || got.UTC != testNowS-1 {
-		t.Fatalf("picked %+v %v", got, ok)
-	}
-	if _, ok := ahead.pick([]TelemetrySample{{UTC: testNowS + 86400}}, now.Add(time.Minute)); ok {
-		t.Fatal("forwarded a sample from the future")
-	}
-	if _, ok := ahead.pick([]TelemetrySample{{UTC: testNowS + 60}}, now.Add(time.Minute)); !ok {
-		t.Fatal("a real sample after a future one was not forwarded")
-	}
-}
-
-func TestForwarderPutsCredentialsInTheBodyNotTheURL(t *testing.T) {
-	received := make(chan *http.Request, 1)
-	var form url.Values
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		body, _ := io.ReadAll(r.Body)
-		form, _ = url.ParseQuery(string(body))
-		received <- r
-	}))
-	defer srv.Close()
-	f := &Forwarder{APIKey: "key-123", UserToken: "token-456", Endpoint: srv.URL + "/1/tlm/send"}
-	if err := f.send(t.Context(), TelemetrySample{UTC: testNowS, Numbers: map[string]float64{"soc": 80}}); err != nil {
-		t.Fatal(err)
-	}
-	r := <-received
-	if r.URL.RawQuery != "" {
-		t.Fatalf("credentials leaked into the URL: %s", r.URL.RawQuery)
-	}
-	if form.Get("api_key") != "key-123" || form.Get("token") != "token-456" || form.Get("tlm") != `{"soc":80,"utc":1790000000}` {
-		t.Fatalf("form %v", form)
-	}
-}
-
 func TestLoadConfigFailsClosed(t *testing.T) {
 	env := map[string]string{
 		"INFLUX_URL": "http://influxdb:8086", "INFLUX_TOKEN": "t", "INFLUX_ORG": "mg4",
@@ -118,7 +52,7 @@ func TestLoadConfigFailsClosed(t *testing.T) {
 	}
 	get := func(k string) string { return env[k] }
 	c, err := loadConfig(get)
-	if err != nil || c.listenAddr != ":8080" || c.influxBucket != "mg4" || c.abrpMinInterval != 5*time.Second {
+	if err != nil || c.listenAddr != ":8080" || c.influxBucket != "mg4" {
 		t.Fatalf("defaults: %+v %v", c, err)
 	}
 	env["INGEST_TOKEN"] = "short"

server/main.go

@@ -1,5 +1,6 @@
 // Command server is the Nils' MG4 car server: the API the car writes its telemetry,
-// consumption and charging sessions to, stored in InfluxDB 2 and forwarded live to ABRP.
+// consumption and charging sessions to, stored in InfluxDB 2. The car sends live data to ABRP
+// itself.
 // Dashboards are Grafana's job; Grafana reads the same bucket directly.
 package main
 
@@ -11,7 +12,6 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
-	"strconv"
 	"syscall"
 	"time"
 )
@@ -22,33 +22,28 @@ const (
 )
 
 type config struct {
-	listenAddr      string
-	influxURL       string
-	influxToken     string
-	influxOrg       string
-	influxBucket    string
-	ingestToken     string
-	abrpAPIKey      string
-	abrpUserToken   string
-	abrpMinInterval time.Duration
-	approvalPass    string
-	publicURL       string
+	listenAddr   string
+	influxURL    string
+	influxToken  string
+	influxOrg    string
+	influxBucket string
+	ingestToken  string
+	approvalPass string
+	publicURL    string
 }
 
 // loadConfig reads the environment and fails closed: a missing InfluxDB setting or a
 // short ingest token stops the server from starting rather than serving something broken.
 func loadConfig(getenv func(string) string) (config, error) {
 	c := config{
-		listenAddr:    getenv("LISTEN_ADDR"),
-		influxURL:     getenv("INFLUX_URL"),
-		influxToken:   getenv("INFLUX_TOKEN"),
-		influxOrg:     getenv("INFLUX_ORG"),
-		influxBucket:  getenv("INFLUX_BUCKET"),
-		ingestToken:   getenv("INGEST_TOKEN"),
-		abrpAPIKey:    getenv("ABRP_API_KEY"),
-		abrpUserToken: getenv("ABRP_USER_TOKEN"),
-		approvalPass:  getenv("APPROVAL_PASSWORD"),
-		publicURL:     getenv("PUBLIC_URL"),
+		listenAddr:   getenv("LISTEN_ADDR"),
+		influxURL:    getenv("INFLUX_URL"),
+		influxToken:  getenv("INFLUX_TOKEN"),
+		influxOrg:    getenv("INFLUX_ORG"),
+		influxBucket: getenv("INFLUX_BUCKET"),
+		ingestToken:  getenv("INGEST_TOKEN"),
+		approvalPass: getenv("APPROVAL_PASSWORD"),
+		publicURL:    getenv("PUBLIC_URL"),
 	}
 	if c.listenAddr == "" {
 		c.listenAddr = ":8080"
@@ -69,14 +64,6 @@ func loadConfig(getenv func(string) string) (config, error) {
 	if c.approvalPass != "" && len(c.approvalPass) < minApprovalPasswordLength {
 		return config{}, fmt.Errorf("APPROVAL_PASSWORD must be at least %d characters", minApprovalPasswordLength)
 	}
-	c.abrpMinInterval = 5 * time.Second
-	if raw := getenv("ABRP_MIN_INTERVAL_SECONDS"); raw != "" {
-		seconds, err := strconv.ParseFloat(raw, 64)
-		if err != nil || seconds < 0 {
-			return config{}, errors.New("ABRP_MIN_INTERVAL_SECONDS must be a non-negative number")
-		}
-		c.abrpMinInterval = time.Duration(seconds * float64(time.Second))
-	}
 	return c, nil
 }
 
@@ -90,12 +77,6 @@ func main() {
 
 	store := NewInfluxStore(cfg.influxURL, cfg.influxToken, cfg.influxOrg, cfg.influxBucket)
 	defer store.Close()
-	forwarder := &Forwarder{
-		APIKey:      cfg.abrpAPIKey,
-		UserToken:   cfg.abrpUserToken,
-		MinInterval: cfg.abrpMinInterval,
-		Client:      &http.Client{Timeout: 10 * time.Second},
-	}
 	// Paired cars' tokens must be known before the first request; starting without them
 	// would answer every paired car with 401, so a failed load stops the start instead.
 	loadCtx, cancelLoad := context.WithTimeout(context.Background(), 30*time.Second)
@@ -106,8 +87,7 @@ func main() {
 		os.Exit(1)
 	}
 	pairing := NewPairing(cfg.approvalPass, devices)
-	api := &API{Store: store, IngestToken: cfg.ingestToken, Forwarder: forwarder,
-		Pairing: pairing, PublicURL: cfg.publicURL}
+	api := &API{Store: store, IngestToken: cfg.ingestToken, Pairing: pairing, PublicURL: cfg.publicURL}
 
 	srv := &http.Server{
 		Addr:              cfg.listenAddr,
@@ -122,7 +102,7 @@ func main() {
 	defer stop()
 	go func() {
 		slog.Info("listening", "addr", cfg.listenAddr, "bucket", cfg.influxBucket,
-			"abrp", forwarder.configured(), "pairing", pairing.Enabled(), "pairedDevices", len(devices))
+			"pairing", pairing.Enabled(), "pairedDevices", len(devices))
 		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
 			slog.Error("server failed", "error", err)
 			os.Exit(1)

server/model.go

@@ -107,7 +107,4 @@ type TelemetryBatch struct {
 	Samples        []TelemetrySample
 	// Dropped names the values and samples validation left out, e.g. "samples[3].speed".
 	Dropped []string
-	// AbrpSentUTC is the newest sample the car delivered to ABRP itself (0: none). The
-	// server forwards only samples newer than that, so ABRP gets each one once.
-	AbrpSentUTC int64
 }

server/validate.go

@@ -273,11 +273,6 @@ func validateTelemetryBatch(body []byte, now time.Time) (TelemetryBatch, error)
 		return TelemetryBatch{}, invalid("samples is invalid")
 	}
 	batch := TelemetryBatch{InstallationID: id, Samples: make([]TelemetrySample, 0, len(raw))}
-	if o.has("abrpSentUtc") {
-		if batch.AbrpSentUTC, err = integer(o["abrpSentUtc"], "abrpSentUtc"); err != nil {
-			return TelemetryBatch{}, err
-		}
-	}
 	latest := now.Add(maxClockAhead).Unix()
 	for i, entry := range raw {
 		name := fmt.Sprintf("samples[%d]", i)