mg4/server/abrp.go

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
)

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) {
	if len(samples) == 0 {
		return TelemetrySample{}, false
	}
	newest := samples[0]
	for _, s := range samples[1:] {
		if s.UTC > newest.UTC {
			newest = s
		}
	}
	if 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
}