Server: InfluxDB 2 storage, live telemetry endpoint, ABRP forwarder

- MongoDB replaced by InfluxDB 2 (one bucket, installationId tag).
  Consumption is stored per chunk at its recordedAt and summed on read,
  so a chunk the car re-sends from its cache overwrites itself instead
  of being counted twice ($inc did count it twice). Lifetime is a
  running total derived on read.
- POST /v1/telemetry takes batches of samples in ABRP's tlm fields; every
  field is bounded, unknown fields are dropped, unread signals stay absent.
- After a successful write the newest fresh sample is forwarded to ABRP
  with ABRP_API_KEY / ABRP_USER_TOKEN from the server environment; rate
  limited, never queued. The car no longer needs ABRP credentials.
- Self-hosting: standalone output, Dockerfile (tests run in the build,
  runs as UID 10001), .env.example and README for the new setup.
- Dev shell gains Node 22 and InfluxDB 2.

Verified locally against InfluxDB 2.7.12: all endpoints, retry
idempotency, and the dashboard pages. Unit tests: 17 pass.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

web/src/app/v1/charging-sessions/[sessionId]/route.ts

@@ -1,6 +1,6 @@
 import { NextResponse } from "next/server";
 import { handle, parseJson, requireIngest } from "@/lib/api";
-import { chargingCollection, ensureIndexes } from "@/lib/mongo";
+import { writeChargingSession } from "@/lib/store";
 import { validateChargingSession } from "@/lib/validate";
 
 export const runtime = "nodejs";
@@ -11,15 +11,7 @@ export async function PUT(request: Request, context: { params: Promise<{ session
   if (denied) return denied;
   return handle(async () => {
     const { sessionId } = await context.params;
-    const record = validateChargingSession(await parseJson(request), sessionId);
-    await ensureIndexes();
-    const collection = await chargingCollection();
-    const updatedAt = new Date();
-    await collection.updateOne(
-      { installationId: record.installationId, sessionId: record.sessionId },
-      { $set: { ...record, updatedAt }, $setOnInsert: { createdAt: updatedAt } },
-      { upsert: true },
-    );
+    await writeChargingSession(validateChargingSession(await parseJson(request), sessionId));
     return new NextResponse(null, { status: 204 });
   });
 }

web/src/app/v1/consumption/daily/route.ts

@@ -1,48 +1,21 @@
 import { NextResponse } from "next/server";
 import { handle, parseJson, requireIngest } from "@/lib/api";
-import { consumptionCollection, ensureIndexes } from "@/lib/mongo";
+import { writeConsumptionChunk } from "@/lib/store";
 import { validateConsumption } from "@/lib/validate";
 
 export const runtime = "nodejs";
 export const dynamic = "force-dynamic";
 
 /**
- * Upsert on (installationId, date), incrementally. The car sends a 2-minute delta chunk
- * (not a cumulative total), so each PUT adds to the day's running totals rather than
- * overwriting them. The app does NOT send lifetime — the backend derives it as the sum of
- * all days' totals when history is read, so an APK wipe that zeroes the local counter can
- * never corrupt the cloud's lifetime.
- *
- * $setOnInsert must not touch `day` at all: $inc on `day.km` and $setOnInsert on `day`
- * conflict in MongoDB. On an insert, $inc treats a missing field as 0 and adds to it.
+ * One delta chunk (not a cumulative total) of a day's consumption. Stored as its own point
+ * at its recordedAt; reads add a day's chunks up. A chunk the car re-sends from its offline
+ * cache carries the same recordedAt and replaces itself, so a retry is never counted twice.
  */
 export async function PUT(request: Request) {
   const denied = requireIngest(request);
   if (denied) return denied;
   return handle(async () => {
-    const record = validateConsumption(await parseJson(request));
-    await ensureIndexes();
-    const collection = await consumptionCollection();
-    const updatedAt = new Date();
-    const $set: Record<string, unknown> = { updatedAt };
-    if (record.sohPercent !== undefined) $set.sohPercent = record.sohPercent;
-    await collection.updateOne(
-      { installationId: record.installationId, date: record.date },
-      {
-        $inc: {
-          "day.km": record.day.km,
-          "day.kwh": record.day.kwh,
-          "day.hours": record.day.hours,
-          "day.socDrop": record.day.socDrop,
-        },
-        $set,
-        $setOnInsert: {
-          createdAt: updatedAt,
-        },
-      },
-      { upsert: true },
-    );
+    await writeConsumptionChunk(validateConsumption(await parseJson(request)));
     return new NextResponse(null, { status: 204 });
   });
 }
-

web/src/app/v1/consumption/history/[installationId]/route.ts

@@ -1,6 +1,6 @@
 import { NextResponse } from "next/server";
 import { handle, requireIngest } from "@/lib/api";
-import { consumptionCollection } from "@/lib/mongo";
+import { readDays } from "@/lib/store";
 import { validateUuid } from "@/lib/validate";
 
 export const runtime = "nodejs";
@@ -21,23 +21,10 @@ export async function GET(request: Request, context: { params: Promise<{ install
   if (denied) return denied;
   return handle(async () => {
     const { installationId } = await context.params;
-    const collection = await consumptionCollection();
-    const records = await collection
-      .find({ installationId: validateUuid(installationId, "installationId") }, { projection: { _id: 0 } })
-      .sort({ date: -1 })
-      .limit(400)
-      .toArray();
-    // Lifetime = sum of every day's totals. Derived, never stored, so a corrupted or
-    // stale lifetime field in any single document cannot poison the total.
-    const lifetime = records.reduce(
-      (sum, r) => ({
-        km: sum.km + (r.day?.km ?? 0),
-        kwh: sum.kwh + (r.day?.kwh ?? 0),
-        hours: sum.hours + (r.day?.hours ?? 0),
-        socDrop: sum.socDrop + (r.day?.socDrop ?? 0),
-      }),
-      { km: 0, kwh: 0, hours: 0, socDrop: 0 },
-    );
+    const days = await readDays(validateUuid(installationId, "installationId"));
+    // Lifetime = sum of every day's totals, which is the newest day's running total.
+    const lifetime = days[0]?.lifetime ?? { km: 0, kwh: 0, hours: 0, socDrop: 0 };
+    const records = days.slice(0, 400).map(({ lifetime: _lifetime, ...day }) => day);
     return NextResponse.json({ records, lifetime });
   });
 }

web/src/app/v1/history/[installationId]/charging/route.ts

@@ -1,6 +1,6 @@
 import { NextResponse } from "next/server";
 import { handle, requireReader } from "@/lib/api";
-import { chargingCollection } from "@/lib/mongo";
+import { readChargingSessions } from "@/lib/store";
 import { validateUuid } from "@/lib/validate";
 
 export const runtime = "nodejs";
@@ -11,12 +11,7 @@ export async function GET(request: Request, context: { params: Promise<{ install
   if (denied) return denied;
   return handle(async () => {
     const { installationId } = await context.params;
-    const collection = await chargingCollection();
-    const records = await collection
-      .find({ installationId: validateUuid(installationId, "installationId") }, { projection: { _id: 0 } })
-      .sort({ startedAt: -1 })
-      .limit(200)
-      .toArray();
+    const records = await readChargingSessions(validateUuid(installationId, "installationId"), 200);
     return NextResponse.json({ records });
   });
 }

web/src/app/v1/history/[installationId]/consumption/route.ts

@@ -1,6 +1,6 @@
 import { NextResponse } from "next/server";
 import { handle, requireReader } from "@/lib/api";
-import { consumptionCollection } from "@/lib/mongo";
+import { readDays } from "@/lib/store";
 import { validateUuid } from "@/lib/validate";
 
 export const runtime = "nodejs";
@@ -11,12 +11,7 @@ export async function GET(request: Request, context: { params: Promise<{ install
   if (denied) return denied;
   return handle(async () => {
     const { installationId } = await context.params;
-    const collection = await consumptionCollection();
-    const records = await collection
-      .find({ installationId: validateUuid(installationId, "installationId") }, { projection: { _id: 0 } })
-      .sort({ date: -1 })
-      .limit(400)
-      .toArray();
+    const records = (await readDays(validateUuid(installationId, "installationId"))).slice(0, 400);
     return NextResponse.json({ records });
   });
 }

web/src/app/v1/telemetry/route.ts

@@ -0,0 +1,24 @@
+import { NextResponse } from "next/server";
+import { forwardToAbrp } from "@/lib/abrp";
+import { handle, parseJson, requireIngest } from "@/lib/api";
+import { writeTelemetry } from "@/lib/store";
+import { validateTelemetryBatch } from "@/lib/validate";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+/**
+ * A batch of live samples from the car. Stored 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.
+ */
+export async function POST(request: Request) {
+  const denied = requireIngest(request);
+  if (denied) return denied;
+  return handle(async () => {
+    const batch = validateTelemetryBatch(await parseJson(request));
+    await writeTelemetry(batch.installationId, batch.samples);
+    forwardToAbrp(batch.samples);
+    return new NextResponse(null, { status: 204 });
+  });
+}

web/src/lib/abrp.ts

@@ -0,0 +1,88 @@
+import { TELEMETRY_FLAGS, type TelemetrySample } from "./types.ts";
+
+/**
+ * Forwards live telemetry to A Better Route Planner.
+ *
+ * The ABRP credentials exist only here, as ABRP_API_KEY and ABRP_USER_TOKEN in the server's
+ * environment; the car never holds them. Forwarding is off when either is unset.
+ *
+ * Only the newest sample of a batch is sent, and at most one every ABRP_MIN_INTERVAL_SECONDS
+ * (default 5). 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.
+ */
+
+const API_URL = "https://api.iternio.com/1/tlm/send";
+const TIMEOUT_MS = 8_000;
+/** Older than this, a sample is history rather than live state, and ABRP gets nothing. */
+const MAX_AGE_SECONDS = 300;
+
+const state = globalThis as unknown as { abrpLastUtc?: number; abrpLastSentMs?: number };
+
+export function abrpConfigured(): boolean {
+  return Boolean(process.env.ABRP_API_KEY && process.env.ABRP_USER_TOKEN);
+}
+
+function minIntervalMs(): number {
+  const seconds = Number(process.env.ABRP_MIN_INTERVAL_SECONDS ?? "5");
+  return Number.isFinite(seconds) && seconds >= 0 ? seconds * 1000 : 5000;
+}
+
+/** The tlm object ABRP expects: the same field names, flags as 0/1. */
+export function toTlm(sample: TelemetrySample): Record<string, number> {
+  const tlm: Record<string, number> = {};
+  for (const [field, value] of Object.entries(sample)) {
+    if (typeof value === "number") tlm[field] = value;
+  }
+  for (const field of TELEMETRY_FLAGS) {
+    const value = sample[field];
+    if (value !== undefined) tlm[field] = value ? 1 : 0;
+  }
+  return tlm;
+}
+
+/** Picks the sample to forward, or null when there is nothing new or it is too soon. */
+export function pickForForward(samples: TelemetrySample[], nowMs: number): TelemetrySample | null {
+  if (samples.length === 0) return null;
+  const newest = samples.reduce((a, b) => (b.utc > a.utc ? b : a));
+  if (nowMs / 1000 - newest.utc > MAX_AGE_SECONDS) return null;
+  if (state.abrpLastUtc !== undefined && newest.utc <= state.abrpLastUtc) return null;
+  if (state.abrpLastSentMs !== undefined && nowMs - state.abrpLastSentMs < minIntervalMs()) return null;
+  return newest;
+}
+
+/** Fire-and-forget: the car's request is answered without waiting for ABRP. */
+export function forwardToAbrp(samples: TelemetrySample[]): void {
+  if (!abrpConfigured()) return;
+  const nowMs = Date.now();
+  const sample = pickForForward(samples, nowMs);
+  if (!sample) return;
+  state.abrpLastUtc = sample.utc;
+  state.abrpLastSentMs = nowMs;
+  void send(sample).catch((error) => console.warn("ABRP forward failed:", error instanceof Error ? error.message : error));
+}
+
+async function send(sample: TelemetrySample): Promise<void> {
+  // Credentials go in the form body, never the URL, which ends up in access logs.
+  const body = new URLSearchParams({
+    api_key: process.env.ABRP_API_KEY ?? "",
+    token: process.env.ABRP_USER_TOKEN ?? "",
+    tlm: JSON.stringify(toTlm(sample)),
+  });
+  const response = await fetch(API_URL, {
+    method: "POST",
+    headers: { "Content-Type": "application/x-www-form-urlencoded" },
+    body,
+    signal: AbortSignal.timeout(TIMEOUT_MS),
+  });
+  if (!response.ok) {
+    // The response body can echo the request; only the status is logged.
+    console.warn(`ABRP forward rejected: HTTP ${response.status}`);
+  }
+}
+
+/** For tests: forget what was forwarded. */
+export function resetAbrpState(): void {
+  state.abrpLastUtc = undefined;
+  state.abrpLastSentMs = undefined;
+}

web/src/lib/api.ts

@@ -2,7 +2,7 @@ import { NextResponse } from "next/server";
 import { hasIngestToken, hasSession, checkDashboardPassword } from "./auth";
 import { ValidationError } from "./validate";
 
-/** Every endpoint touches MongoDB, which rules out the edge runtime. */
+/** Every endpoint touches InfluxDB, which rules out the edge runtime. */
 export const runtime = "nodejs";
 export const dynamic = "force-dynamic";
 

web/src/lib/data.ts

@@ -1,11 +1,12 @@
-import { chargingCollection, consumptionCollection, type ChargingSession, type DailyConsumption } from "./mongo";
+import { installationIds, readChargingSessions, readDays, readLatestTelemetry } from "./store";
+import type { ChargingSession, DailyConsumption, TelemetrySample } from "./types";
 
 /**
  * Reads for the dashboard.
  *
- * Every query is scoped to one installation id. There is one car today, but the id is in
- * the schema and in the unique indexes, so scoping now costs nothing and means a second
- * head unit does not silently merge its history into the first one's charts.
+ * Every query is scoped to one installation id. There is one car today, but the id is on
+ * every point, so scoping now costs nothing and means a second head unit does not
+ * silently merge its history into the first one's charts.
  */
 
 export interface Overview {
@@ -13,36 +14,32 @@ export interface Overview {
   installationCount: number;
   days: DailyConsumption[];
   sessions: ChargingSession[];
+  latest: (TelemetrySample & { at: Date }) | null;
 }
 
 /** The installation that reported most recently — the car in use, if there are ever two. */
 export async function latestInstallationId(): Promise<{ id: string | null; count: number }> {
-  const collection = await consumptionCollection();
-  const ids = await collection.distinct("installationId");
-  if (ids.length === 0) {
-    const charging = await chargingCollection();
-    const chargingIds = await charging.distinct("installationId");
-    return { id: chargingIds[0] ?? null, count: chargingIds.length };
-  }
-  if (ids.length === 1) return { id: ids[0], count: 1 };
-  const newest = await collection.find({}, { projection: { installationId: 1 } })
-    .sort({ date: -1 }).limit(1).next();
-  return { id: newest?.installationId ?? ids[0], count: ids.length };
+  const ids = await installationIds();
+  return { id: ids[0] ?? null, count: ids.length };
 }
 
 export async function loadOverview(dayLimit = 60, sessionLimit = 60): Promise<Overview> {
   const { id, count } = await latestInstallationId();
-  if (!id) return { installationId: null, installationCount: 0, days: [], sessions: [] };
-  const [consumption, charging] = await Promise.all([consumptionCollection(), chargingCollection()]);
-  const [days, sessions] = await Promise.all([
-    consumption.find({ installationId: id }, { projection: { _id: 0 } })
-      .sort({ date: -1 }).limit(dayLimit).toArray(),
-    charging.find({ installationId: id }, { projection: { _id: 0 } })
-      .sort({ startedAt: -1 }).limit(sessionLimit).toArray(),
+  if (!id) return { installationId: null, installationCount: 0, days: [], sessions: [], latest: null };
+  const [days, sessions, latest] = await Promise.all([
+    readDays(id),
+    readChargingSessions(id, sessionLimit),
+    readLatestTelemetry(id),
   ]);
-  // Charts read left-to-right in time; the queries sort newest-first so the limit keeps
+  // Charts read left-to-right in time; the reads come newest-first so the limit keeps
   // the recent end rather than the beginning of history.
-  return { installationId: id, installationCount: count, days: days.reverse(), sessions };
+  return {
+    installationId: id,
+    installationCount: count,
+    days: days.slice(0, dayLimit).reverse(),
+    sessions,
+    latest,
+  };
 }
 
 export interface ChargeTotals {

web/src/lib/days.ts

@@ -0,0 +1,43 @@
+import type { ConsumptionChunk, DailyConsumption, Totals } from "./types.ts";
+
+/**
+ * Folds chunks into days, newest day first, each with its running lifetime total.
+ * Pure, so the arithmetic is tested without a database.
+ */
+export function aggregateDays(chunks: ConsumptionChunk[]): DailyConsumption[] {
+  const byDate = new Map<string, DailyConsumption & { sohAt: number }>();
+  for (const chunk of chunks) {
+    let day = byDate.get(chunk.date);
+    if (!day) {
+      day = {
+        installationId: chunk.installationId,
+        date: chunk.date,
+        recordedAt: chunk.recordedAt,
+        day: { km: 0, kwh: 0, hours: 0, socDrop: 0 },
+        sohAt: -Infinity,
+      };
+      byDate.set(chunk.date, day);
+    }
+    day.day.km += chunk.day.km;
+    day.day.kwh += chunk.day.kwh;
+    day.day.hours += chunk.day.hours;
+    day.day.socDrop += chunk.day.socDrop;
+    const at = chunk.recordedAt.getTime();
+    if (at > day.recordedAt.getTime()) day.recordedAt = chunk.recordedAt;
+    // The newest manual SoH of the day wins.
+    if (chunk.sohPercent !== undefined && at > day.sohAt) {
+      day.sohPercent = chunk.sohPercent;
+      day.sohAt = at;
+    }
+  }
+  const ascending = [...byDate.values()].sort((a, b) => a.date.localeCompare(b.date));
+  const running: Totals = { km: 0, kwh: 0, hours: 0, socDrop: 0 };
+  const days = ascending.map(({ sohAt: _sohAt, ...day }) => {
+    running.km += day.day.km;
+    running.kwh += day.day.kwh;
+    running.hours += day.day.hours;
+    running.socDrop += day.day.socDrop;
+    return { ...day, lifetime: { ...running } };
+  });
+  return days.reverse();
+}

web/src/lib/influx.ts

@@ -0,0 +1,40 @@
+import { InfluxDB, type QueryApi, type WriteApi } from "@influxdata/influxdb-client";
+
+/**
+ * One InfluxDB client per server process.
+ *
+ * Configured entirely from the environment so the image carries no credentials:
+ * INFLUX_URL, INFLUX_TOKEN, INFLUX_ORG and INFLUX_BUCKET (default "mg4"). The global
+ * survives module reloads in dev, where Next re-evaluates modules on every edit.
+ */
+const globalForInflux = globalThis as unknown as { influx?: InfluxDB };
+
+function env(name: string): string {
+  const value = process.env[name];
+  if (!value) throw new Error(`${name} is not configured`);
+  return value;
+}
+
+function client(): InfluxDB {
+  if (!globalForInflux.influx) {
+    globalForInflux.influx = new InfluxDB({ url: env("INFLUX_URL"), token: env("INFLUX_TOKEN"), timeout: 10_000 });
+  }
+  return globalForInflux.influx;
+}
+
+export function bucket(): string {
+  return process.env.INFLUX_BUCKET || "mg4";
+}
+
+/** A write API with millisecond precision; the caller must close() it to flush. */
+export function writeApi(): WriteApi {
+  return client().getWriteApi(env("INFLUX_ORG"), bucket(), "ms", {
+    // The caller waits for close(); a failed write must reach it rather than be retried
+    // in the background after the car has already been told it succeeded.
+    maxRetries: 0,
+  });
+}
+
+export function queryApi(): QueryApi {
+  return client().getQueryApi(env("INFLUX_ORG"));
+}

web/src/lib/mongo.ts

@@ -1,100 +0,0 @@
-import { MongoClient, type Collection, type Db } from "mongodb";
-
-/**
- * One MongoClient per warm lambda, cached on globalThis.
- *
- * A serverless function is re-entered far more often than it is cold-started, and a fresh
- * MongoClient per request opens a fresh pool per request — Atlas starts refusing
- * connections long before the traffic justifies it. The global survives module reloads in
- * dev, where Next re-evaluates modules on every edit.
- */
-const globalForMongo = globalThis as unknown as { mongoClient?: Promise<MongoClient> };
-
-export interface Totals {
-  km: number;
-  kwh: number;
-  hours: number;
-  socDrop: number;
-}
-
-export interface DailyConsumption {
-  installationId: string;
-  date: string;
-  recordedAt: Date;
-  day: Totals;
-  /** Optional — the backend derives lifetime as the sum of all days' totals. */
-  lifetime?: Totals;
-  sohPercent?: number;
-  createdAt: Date;
-  updatedAt: Date;
-}
-
-export interface CurvePoint {
-  timestamp: Date;
-  powerKw: number;
-  socPercent?: number;
-}
-
-export interface ChargingSession {
-  installationId: string;
-  sessionId: string;
-  startedAt: Date;
-  endedAt: Date;
-  durationSeconds: number;
-  energyKwh: number;
-  chargingLossPercent: number;
-  gridEnergyKwh: number;
-  pricePerKwh: number;
-  totalCost: number;
-  isDemo?: boolean;
-  startSocPercent?: number;
-  endSocPercent?: number;
-  curve: CurvePoint[];
-  createdAt: Date;
-  updatedAt: Date;
-}
-
-function clientPromise(): Promise<MongoClient> {
-  const uri = process.env.MONGODB_URI;
-  if (!uri) throw new Error("MONGODB_URI is not configured");
-  if (!globalForMongo.mongoClient) {
-    globalForMongo.mongoClient = new MongoClient(uri, {
-      serverSelectionTimeoutMS: 10_000,
-      maxPoolSize: 10,
-    }).connect();
-  }
-  return globalForMongo.mongoClient;
-}
-
-async function db(): Promise<Db> {
-  const client = await clientPromise();
-  return client.db(process.env.MONGODB_DATABASE || "fatihsmg4");
-}
-
-export async function consumptionCollection(): Promise<Collection<DailyConsumption>> {
-  return (await db()).collection<DailyConsumption>("daily_consumption");
-}
-
-export async function chargingCollection(): Promise<Collection<ChargingSession>> {
-  return (await db()).collection<ChargingSession>("charging_sessions");
-}
-
-/**
- * Creates the indexes the Express service used to create at boot.
- *
- * There is no boot here — a lambda starts on a request — so this runs on the first write
- * of each warm instance and is skipped afterwards. `createIndex` is idempotent, but it is
- * still a round trip, and the uniqueness constraints are what make the upserts safe.
- */
-let indexesEnsured = false;
-export async function ensureIndexes(): Promise<void> {
-  if (indexesEnsured) return;
-  const [consumption, charging] = await Promise.all([consumptionCollection(), chargingCollection()]);
-  await Promise.all([
-    consumption.createIndex({ installationId: 1, date: 1 }, { unique: true }),
-    consumption.createIndex({ installationId: 1, date: -1 }),
-    charging.createIndex({ installationId: 1, sessionId: 1 }, { unique: true }),
-    charging.createIndex({ installationId: 1, startedAt: -1 }),
-  ]);
-  indexesEnsured = true;
-}

web/src/lib/store.ts

@@ -0,0 +1,225 @@
+import { Point, flux } from "@influxdata/influxdb-client";
+import { aggregateDays } from "./days.ts";
+import { bucket, queryApi, writeApi } from "./influx.ts";
+import {
+  TELEMETRY_FLAGS,
+  TELEMETRY_NUMBERS,
+  type ChargingSession,
+  type ConsumptionChunk,
+  type CurvePoint,
+  type DailyConsumption,
+  type TelemetrySample,
+} from "./types.ts";
+
+/**
+ * Storage on InfluxDB 2. One bucket; every point carries an `installationId` tag.
+ *
+ *   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
+ *   telemetry          time = sample utc                            ABRP tlm fields
+ *
+ * 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 numeric field is written as a float. InfluxDB fixes a field's type on first
+ * write, so a 0 sent as an integer would make every later 0.5 a rejected write.
+ */
+
+const TAG = "installationId";
+
+async function write(points: Point[]): Promise<void> {
+  const api = writeApi();
+  api.writePoints(points);
+  await api.close();
+}
+
+export async function writeConsumptionChunk(chunk: ConsumptionChunk): Promise<void> {
+  const point = new Point("consumption_chunk")
+    .tag(TAG, chunk.installationId)
+    .tag("date", chunk.date)
+    .floatField("km", chunk.day.km)
+    .floatField("kwh", chunk.day.kwh)
+    .floatField("hours", chunk.day.hours)
+    .floatField("socDrop", chunk.day.socDrop)
+    .timestamp(chunk.recordedAt);
+  if (chunk.sohPercent !== undefined) point.floatField("sohPercent", chunk.sohPercent);
+  await write([point]);
+}
+
+export async function writeChargingSession(session: ChargingSession): Promise<void> {
+  const summary = new Point("charging_session")
+    .tag(TAG, session.installationId)
+    .tag("sessionId", session.sessionId)
+    .intField("endedAtMs", session.endedAt.getTime())
+    .floatField("durationSeconds", session.durationSeconds)
+    .floatField("energyKwh", session.energyKwh)
+    .floatField("chargingLossPercent", session.chargingLossPercent)
+    .floatField("gridEnergyKwh", session.gridEnergyKwh)
+    .floatField("pricePerKwh", session.pricePerKwh)
+    .floatField("totalCost", session.totalCost)
+    .booleanField("isDemo", session.isDemo === true)
+    .timestamp(session.startedAt);
+  if (session.startSocPercent !== undefined) summary.floatField("startSocPercent", session.startSocPercent);
+  if (session.endSocPercent !== undefined) summary.floatField("endSocPercent", session.endSocPercent);
+  const curve = session.curve.map((p) => {
+    const point = new Point("charging_curve")
+      .tag(TAG, session.installationId)
+      .tag("sessionId", session.sessionId)
+      .floatField("powerKw", p.powerKw)
+      .timestamp(p.timestamp);
+    if (p.socPercent !== undefined) point.floatField("socPercent", p.socPercent);
+    return point;
+  });
+  await write([summary, ...curve]);
+}
+
+export async function writeTelemetry(installationId: string, samples: TelemetrySample[]): Promise<void> {
+  const points = samples.map((sample) => {
+    const point = new Point("telemetry").tag(TAG, installationId).timestamp(new Date(sample.utc * 1000));
+    for (const field of Object.keys(TELEMETRY_NUMBERS) as (keyof typeof TELEMETRY_NUMBERS)[]) {
+      const value = sample[field];
+      if (value !== undefined) point.floatField(field, value);
+    }
+    for (const field of TELEMETRY_FLAGS) {
+      const value = sample[field];
+      if (value !== undefined) point.booleanField(field, value);
+    }
+    return point;
+  });
+  await write(points);
+}
+
+// ---------- Reads ----------
+
+type Row = Record<string, unknown> & { _time: string };
+
+async function rows(query: ReturnType<typeof flux>): Promise<Row[]> {
+  return queryApi().collectRows<Row>(query);
+}
+
+function num(row: Row, field: string): number | undefined {
+  const value = row[field];
+  return typeof value === "number" && Number.isFinite(value) ? value : undefined;
+}
+
+export async function readConsumptionChunks(installationId: string): Promise<ConsumptionChunk[]> {
+  const result = await rows(flux`
+    from(bucket: ${bucket()})
+      |> range(start: 0)
+      |> filter(fn: (r) => r._measurement == "consumption_chunk" and r.installationId == ${installationId})
+      |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")`);
+  return result.map((row) => {
+    const sohPercent = num(row, "sohPercent");
+    return {
+      installationId,
+      date: String(row.date),
+      recordedAt: new Date(row._time),
+      day: {
+        km: num(row, "km") ?? 0,
+        kwh: num(row, "kwh") ?? 0,
+        hours: num(row, "hours") ?? 0,
+        socDrop: num(row, "socDrop") ?? 0,
+      },
+      ...(sohPercent === undefined ? {} : { sohPercent }),
+    };
+  });
+}
+
+export async function readDays(installationId: string): Promise<DailyConsumption[]> {
+  return aggregateDays(await readConsumptionChunks(installationId));
+}
+
+export async function readChargingSessions(installationId: string, limit: number): Promise<ChargingSession[]> {
+  const summaries = await rows(flux`
+    from(bucket: ${bucket()})
+      |> range(start: 0)
+      |> filter(fn: (r) => r._measurement == "charging_session" and r.installationId == ${installationId})
+      |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
+      |> group()
+      |> sort(columns: ["_time"], desc: true)
+      |> limit(n: ${limit})`);
+  if (summaries.length === 0) return [];
+
+  const sessions: ChargingSession[] = summaries.map((row) => {
+    const startSocPercent = num(row, "startSocPercent");
+    const endSocPercent = num(row, "endSocPercent");
+    return {
+      installationId,
+      sessionId: String(row.sessionId),
+      startedAt: new Date(row._time),
+      endedAt: new Date(num(row, "endedAtMs") ?? Date.parse(row._time)),
+      durationSeconds: num(row, "durationSeconds") ?? 0,
+      energyKwh: num(row, "energyKwh") ?? 0,
+      chargingLossPercent: num(row, "chargingLossPercent") ?? 10,
+      gridEnergyKwh: num(row, "gridEnergyKwh") ?? 0,
+      pricePerKwh: num(row, "pricePerKwh") ?? 0,
+      totalCost: num(row, "totalCost") ?? 0,
+      ...(row.isDemo === true ? { isDemo: true } : {}),
+      ...(startSocPercent === undefined ? {} : { startSocPercent }),
+      ...(endSocPercent === undefined ? {} : { endSocPercent }),
+      curve: [],
+    };
+  });
+
+  // Curves for exactly these sessions: bounded by the oldest start and the newest end.
+  const start = new Date(Math.min(...sessions.map((s) => s.startedAt.getTime())));
+  const stop = new Date(Math.max(...sessions.map((s) => s.endedAt.getTime())) + 1000);
+  const curveRows = await rows(flux`
+    from(bucket: ${bucket()})
+      |> range(start: ${start}, stop: ${stop})
+      |> filter(fn: (r) => r._measurement == "charging_curve" and r.installationId == ${installationId})
+      |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")`);
+  const bySession = new Map(sessions.map((s) => [s.sessionId, s]));
+  for (const row of curveRows) {
+    const session = bySession.get(String(row.sessionId));
+    const powerKw = num(row, "powerKw");
+    if (!session || powerKw === undefined) continue;
+    const socPercent = num(row, "socPercent");
+    const point: CurvePoint = { timestamp: new Date(row._time), powerKw };
+    if (socPercent !== undefined) point.socPercent = socPercent;
+    session.curve.push(point);
+  }
+  for (const session of sessions) session.curve.sort((a, b) => a.timestamp.getTime() - b.timestamp.getTime());
+  return sessions;
+}
+
+/** Installations that reported in the last 400 days, the most recent first. */
+export async function installationIds(): Promise<string[]> {
+  const result = await rows(flux`
+    from(bucket: ${bucket()})
+      |> range(start: -400d)
+      |> filter(fn: (r) => r._measurement == "consumption_chunk" or r._measurement == "charging_session" or r._measurement == "telemetry")
+      |> keep(columns: ["_time", "installationId"])
+      |> group(columns: ["installationId"])
+      |> max(column: "_time")
+      |> group()
+      |> sort(columns: ["_time"], desc: true)`);
+  return result.map((row) => String(row.installationId));
+}
+
+/**
+ * The latest known value of every live field over the last 30 days, for the dashboard.
+ * Per field rather than per sample: the newest sample may lack a GPS fix that the one
+ * before it had. `at` is when the newest of them arrived. Null when nothing was sent.
+ */
+export async function readLatestTelemetry(installationId: string): Promise<(TelemetrySample & { at: Date }) | null> {
+  const result = await rows(flux`
+    from(bucket: ${bucket()})
+      |> range(start: -30d)
+      |> filter(fn: (r) => r._measurement == "telemetry" and r.installationId == ${installationId})
+      |> last()`);
+  if (result.length === 0) return null;
+  const newest = Math.max(...result.map((row) => Date.parse(row._time)));
+  const sample: TelemetrySample & { at: Date } = { utc: Math.floor(newest / 1000), at: new Date(newest) };
+  const numbers = TELEMETRY_NUMBERS as Record<string, unknown>;
+  const flags = TELEMETRY_FLAGS as readonly string[];
+  for (const row of result) {
+    const field = String(row._field);
+    const value = row._value;
+    if (field in numbers && typeof value === "number") (sample as Record<string, unknown>)[field] = value;
+    if (flags.includes(field) && typeof value === "boolean") (sample as Record<string, unknown>)[field] = value;
+  }
+  return sample;
+}

web/src/lib/types.ts

@@ -0,0 +1,96 @@
+/** Record shapes shared by validation, storage and the dashboard. */
+
+export interface Totals {
+  km: number;
+  kwh: number;
+  hours: number;
+  socDrop: number;
+}
+
+/** One day's consumption: the sum of every chunk the car sent for that date. */
+export interface DailyConsumption {
+  installationId: string;
+  /** The car's local date, YYYY-MM-DD. */
+  date: string;
+  /** When the newest chunk of the day was recorded. */
+  recordedAt: Date;
+  day: Totals;
+  /** Running total up to and including this day, derived when history is read. */
+  lifetime?: Totals;
+  sohPercent?: number;
+}
+
+/** One 2-minute delta chunk as the car sends it; days are built from these. */
+export interface ConsumptionChunk {
+  installationId: string;
+  date: string;
+  recordedAt: Date;
+  day: Totals;
+  sohPercent?: number;
+}
+
+export interface CurvePoint {
+  timestamp: Date;
+  powerKw: number;
+  socPercent?: number;
+}
+
+export interface ChargingSession {
+  installationId: string;
+  sessionId: string;
+  startedAt: Date;
+  endedAt: Date;
+  durationSeconds: number;
+  energyKwh: number;
+  chargingLossPercent: number;
+  gridEnergyKwh: number;
+  pricePerKwh: number;
+  totalCost: number;
+  isDemo?: boolean;
+  startSocPercent?: number;
+  endSocPercent?: number;
+  curve: CurvePoint[];
+}
+
+/** Numeric live-telemetry fields, named and unit-ed as ABRP's tlm/send expects them. */
+export const TELEMETRY_NUMBERS = {
+  soc: [0, 100], // %
+  speed: [0, 400], // km/h
+  power: [-1000, 1000], // kW, positive leaving the battery
+  est_battery_range: [0, 2000], // km
+  capacity: [0, 500], // kWh
+  soe: [0, 500], // kWh
+  soh: [0, 100], // %
+  kwh_charged: [0, 1000], // kWh, this charge session
+  odometer: [0, 10_000_000], // km
+  lat: [-90, 90],
+  lon: [-180, 180],
+  elevation: [-1000, 10_000], // m
+  heading: [0, 360], // °
+  batt_temp: [-60, 100], // °C
+  cabin_temp: [-60, 100], // °C
+  ext_temp: [-60, 70], // °C
+  hvac_setpoint: [0, 40], // °C
+  hvac_power: [0, 50], // kW
+  voltage: [0, 1500], // V
+  current: [-2000, 2000], // A
+  tire_pressure_fl: [0, 1000], // kPa
+  tire_pressure_fr: [0, 1000],
+  tire_pressure_rl: [0, 1000],
+  tire_pressure_rr: [0, 1000],
+} as const satisfies Record<string, readonly [number, number]>;
+
+/** Boolean live-telemetry fields; ABRP takes them as 0/1. */
+export const TELEMETRY_FLAGS = ["is_charging", "is_dcfc", "is_parked"] as const;
+
+export type TelemetryNumber = keyof typeof TELEMETRY_NUMBERS;
+export type TelemetryFlag = (typeof TELEMETRY_FLAGS)[number];
+
+/** One live sample. Only `utc` is always present; an unread signal is absent, never 0. */
+export type TelemetrySample = { utc: number } & Partial<Record<TelemetryNumber, number>> &
+  Partial<Record<TelemetryFlag, boolean>>;
+
+export interface TelemetryBatch {
+  installationId: string;
+  samples: TelemetrySample[];
+}

web/src/lib/validate.ts

@@ -6,35 +6,40 @@
  * read forever. Every number is bounded; an out-of-range value is rejected rather than
  * stored and quietly plotted.
  */
-import type { ChargingSession, CurvePoint, DailyConsumption, Totals } from "./mongo";
+import {
+  TELEMETRY_FLAGS,
+  TELEMETRY_NUMBERS,
+  type ChargingSession,
+  type ConsumptionChunk,
+  type CurvePoint,
+  type TelemetryBatch,
+  type TelemetrySample,
+  type Totals,
+} from "./types.ts";
 
 export class ValidationError extends Error {}
 
 const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
 const ISO_DATE = /^\d{4}-\d{2}-\d{2}$/;
 
-export type ConsumptionInput = Omit<DailyConsumption, "createdAt" | "updatedAt">;
-export type ChargingInput = Omit<ChargingSession, "createdAt" | "updatedAt">;
-
-export function validateConsumption(body: unknown): ConsumptionInput {
+export function validateConsumption(body: unknown): ConsumptionChunk {
   const record = asObject(body);
   const date = record.date;
   if (typeof date !== "string" || !ISO_DATE.test(date)) throw new ValidationError("date is invalid");
   const sohPercent = record.sohPercent;
-  const lifetime = record.lifetime;
+  // A lifetime from an older app is still validated, so a malformed one is still a 400,
+  // but it is not stored: the backend derives lifetime from the day totals.
+  if (record.lifetime !== undefined) validateTotals(record.lifetime, "lifetime");
   return {
     installationId: validateUuid(record.installationId, "installationId"),
     date,
     recordedAt: validateDate(record.recordedAt, "recordedAt"),
     day: validateTotals(record.day, "day"),
-    // Lifetime is optional — the backend derives it from day totals, and the app no longer
-    // sends it. Legacy documents may still carry one; new chunks do not.
-    ...(lifetime === undefined ? {} : { lifetime: validateTotals(lifetime, "lifetime") }),
     ...(sohPercent === undefined ? {} : { sohPercent: bounded(sohPercent, 0, 100, "sohPercent") }),
   };
 }
 
-export function validateChargingSession(body: unknown, pathSessionId: string): ChargingInput {
+export function validateChargingSession(body: unknown, pathSessionId: string): ChargingSession {
   const record = asObject(body);
   const sessionId = validateUuid(record.sessionId, "sessionId");
   if (sessionId !== pathSessionId) throw new ValidationError("sessionId does not match path");
@@ -75,6 +80,52 @@ export function validateChargingSession(body: unknown, pathSessionId: string): C
   };
 }
 
+const MAX_SAMPLES = 500;
+/** Epoch seconds: 2001 to 2286. Milliseconds would be far above the upper bound. */
+const UTC_MIN = 1_000_000_000;
+const UTC_MAX = 9_999_999_999;
+
+/**
+ * A batch of live samples in ABRP's tlm field names and units. Fields the server does not
+ * know are dropped rather than rejected, so a newer app can add one without a server
+ * release; every known field is bounded, and a flag may be sent as 0/1 or a boolean.
+ */
+export function validateTelemetryBatch(body: unknown): TelemetryBatch {
+  const record = asObject(body);
+  const installationId = validateUuid(record.installationId, "installationId");
+  const samples = record.samples;
+  if (!Array.isArray(samples) || samples.length === 0 || samples.length > MAX_SAMPLES) {
+    throw new ValidationError("samples is invalid");
+  }
+  return { installationId, samples: samples.map((entry, i) => validateSample(entry, `samples[${i}]`)) };
+}
+
+function validateSample(entry: unknown, name: string): TelemetrySample {
+  const raw = asObject(entry);
+  const utc = raw.utc;
+  if (typeof utc !== "number" || !Number.isSafeInteger(utc) || utc < UTC_MIN || utc > UTC_MAX) {
+    throw new ValidationError(`${name}.utc is invalid`);
+  }
+  const sample: TelemetrySample = { utc };
+  for (const [field, [min, max]] of Object.entries(TELEMETRY_NUMBERS)) {
+    const value = raw[field];
+    if (value === undefined || value === null) continue;
+    (sample as Record<string, unknown>)[field] = bounded(value, min, max, `${name}.${field}`);
+  }
+  for (const field of TELEMETRY_FLAGS) {
+    const value = raw[field];
+    if (value === undefined || value === null) continue;
+    if (value === true || value === 1) sample[field] = true;
+    else if (value === false || value === 0) sample[field] = false;
+    else throw new ValidationError(`${name}.${field} is invalid`);
+  }
+  // A position is only a position as a pair.
+  if ((sample.lat === undefined) !== (sample.lon === undefined)) {
+    throw new ValidationError(`${name}.lat and lon must be sent together`);
+  }
+  return sample;
+}
+
 export function validateUuid(value: unknown, name: string): string {
   if (typeof value !== "string" || !UUID.test(value)) throw new ValidationError(`${name} is invalid`);
   return value;