A finished signal recording goes to an outbox and is sent gzipped with
the car's token (PUT /v1/logs/{name}); the service retries every 2
minutes while the car is offline. The server keeps the newest 200 on a
new volume (LOGS_DIR); reading takes a token of its own (LOGS_TOKEN).
fetch-logs in the dev shell fetches new ones into logs/.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LhzAVaUyqcw5PkGjz4mWW6
180 lines
6.0 KiB
Go
180 lines
6.0 KiB
Go
// Command server is the Nils' MG4 car server: the API the car writes its telemetry,
|
|
// 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
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"net/http"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
minIngestTokenLength = 24
|
|
minApprovalPasswordLength = 12
|
|
)
|
|
|
|
type config struct {
|
|
listenAddr string
|
|
influxURL string
|
|
influxToken string
|
|
influxOrg string
|
|
influxBucket string
|
|
ingestToken string
|
|
approvalPass string
|
|
publicURL string
|
|
forgeURL string
|
|
forgeRepo string
|
|
forgeToken string
|
|
mapTilesURL string
|
|
mapCacheDir string
|
|
logsDir string
|
|
logsToken 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"),
|
|
approvalPass: getenv("APPROVAL_PASSWORD"),
|
|
publicURL: getenv("PUBLIC_URL"),
|
|
forgeURL: getenv("FORGE_URL"),
|
|
forgeRepo: getenv("FORGE_REPO"),
|
|
forgeToken: getenv("FORGE_TOKEN"),
|
|
mapTilesURL: getenv("MAP_TILES_URL"),
|
|
mapCacheDir: getenv("MAP_CACHE_DIR"),
|
|
logsDir: getenv("LOGS_DIR"),
|
|
logsToken: getenv("LOGS_TOKEN"),
|
|
}
|
|
if c.listenAddr == "" {
|
|
c.listenAddr = ":8080"
|
|
}
|
|
if c.influxBucket == "" {
|
|
c.influxBucket = "mg4"
|
|
}
|
|
for name, value := range map[string]string{"INFLUX_URL": c.influxURL, "INFLUX_TOKEN": c.influxToken, "INFLUX_ORG": c.influxOrg} {
|
|
if value == "" {
|
|
return config{}, fmt.Errorf("%s is not configured", name)
|
|
}
|
|
}
|
|
if len(c.ingestToken) < minIngestTokenLength {
|
|
return config{}, fmt.Errorf("INGEST_TOKEN must be at least %d characters", minIngestTokenLength)
|
|
}
|
|
// Pairing is optional; when it is on, its password guards token issuance and must not
|
|
// be guessable.
|
|
// App updates are optional; when the forge is named it must be https, so the token
|
|
// never crosses the network in the clear.
|
|
if c.forgeURL != "" && !strings.HasPrefix(c.forgeURL, "https://") {
|
|
return config{}, errors.New("FORGE_URL must be an https URL")
|
|
}
|
|
// The map's roads come from vector tiles named by a TileJSON: OpenFreeMap unless another is
|
|
// named, none with "off". Only https, like the forge.
|
|
switch c.mapTilesURL {
|
|
case "":
|
|
c.mapTilesURL = DefaultMapTilesURL
|
|
case "off":
|
|
c.mapTilesURL = ""
|
|
default:
|
|
if !strings.HasPrefix(c.mapTilesURL, "https://") {
|
|
return config{}, errors.New("MAP_TILES_URL must be an https URL or \"off\"")
|
|
}
|
|
}
|
|
// Logs from the car are optional (LOGS_DIR); reading them needs a token of their own, as
|
|
// long as the ingest token, and never the ingest token itself.
|
|
if c.logsToken != "" && c.logsDir == "" {
|
|
return config{}, errors.New("LOGS_TOKEN is set but LOGS_DIR is not")
|
|
}
|
|
if c.logsToken != "" && len(c.logsToken) < minIngestTokenLength {
|
|
return config{}, fmt.Errorf("LOGS_TOKEN must be at least %d characters", minIngestTokenLength)
|
|
}
|
|
if c.logsToken != "" && c.logsToken == c.ingestToken {
|
|
return config{}, errors.New("LOGS_TOKEN must differ from INGEST_TOKEN")
|
|
}
|
|
if c.approvalPass != "" && len(c.approvalPass) < minApprovalPasswordLength {
|
|
return config{}, fmt.Errorf("APPROVAL_PASSWORD must be at least %d characters", minApprovalPasswordLength)
|
|
}
|
|
return c, nil
|
|
}
|
|
|
|
func main() {
|
|
slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, nil)))
|
|
cfg, err := loadConfig(os.Getenv)
|
|
if err != nil {
|
|
slog.Error("invalid configuration", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
store := NewInfluxStore(cfg.influxURL, cfg.influxToken, cfg.influxOrg, cfg.influxBucket)
|
|
defer store.Close()
|
|
// 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)
|
|
devices, err := store.LoadDevices(loadCtx)
|
|
cancelLoad()
|
|
if err != nil {
|
|
slog.Error("paired devices could not be loaded", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
pairing := NewPairing(cfg.approvalPass, devices)
|
|
releases := &Releases{ForgeURL: cfg.forgeURL, Repo: cfg.forgeRepo, Token: cfg.forgeToken}
|
|
api := &API{Store: store, IngestToken: cfg.ingestToken, Pairing: pairing, Handoff: NewHandoff(),
|
|
Releases: releases, PublicURL: cfg.publicURL}
|
|
if cfg.mapTilesURL != "" {
|
|
if cfg.mapCacheDir != "" {
|
|
if err := os.MkdirAll(cfg.mapCacheDir, 0o700); err != nil {
|
|
slog.Error("map cache directory not usable", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
api.Maps = NewMaps(cfg.mapTilesURL, cfg.mapCacheDir, mapUserAgent)
|
|
}
|
|
if cfg.logsDir != "" {
|
|
if err := os.MkdirAll(cfg.logsDir, 0o700); err != nil {
|
|
slog.Error("logs directory not usable", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
api.Logs = NewLogs(cfg.logsDir, cfg.logsToken)
|
|
}
|
|
|
|
srv := &http.Server{
|
|
Addr: cfg.listenAddr,
|
|
Handler: api.Routes(),
|
|
ReadHeaderTimeout: 10 * time.Second,
|
|
ReadTimeout: 30 * time.Second,
|
|
WriteTimeout: 30 * time.Second,
|
|
IdleTimeout: 120 * time.Second,
|
|
}
|
|
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
go func() {
|
|
slog.Info("listening", "addr", cfg.listenAddr, "bucket", cfg.influxBucket,
|
|
"pairing", pairing.Enabled(), "pairedDevices", len(devices), "appUpdates", releases.enabled(),
|
|
"maps", api.Maps != nil, "logs", api.Logs != nil, "logsReadable", api.Logs != nil && cfg.logsToken != "")
|
|
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
slog.Error("server failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
}()
|
|
<-ctx.Done()
|
|
shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
if err := srv.Shutdown(shutdown); err != nil {
|
|
slog.Error("shutdown", "error", err)
|
|
}
|
|
}
|