From 4ea66cf42407596483a3a93f175cd6be8215610e Mon Sep 17 00:00:00 2001 From: Nils Stinnesbeck Date: Sat, 17 Oct 2026 14:09:29 +0200 Subject: [PATCH] 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 --- flake.nix | 2 +- web/.dockerignore | 7 + web/.env.example | 17 +- web/Dockerfile | 27 +++ web/README.md | 84 ++++--- web/next.config.mjs | 2 + web/package-lock.json | 155 +----------- web/package.json | 4 +- .../v1/charging-sessions/[sessionId]/route.ts | 12 +- web/src/app/v1/consumption/daily/route.ts | 37 +-- .../history/[installationId]/route.ts | 23 +- .../[installationId]/charging/route.ts | 9 +- .../[installationId]/consumption/route.ts | 9 +- web/src/app/v1/telemetry/route.ts | 24 ++ web/src/lib/abrp.ts | 88 +++++++ web/src/lib/api.ts | 2 +- web/src/lib/data.ts | 45 ++-- web/src/lib/days.ts | 43 ++++ web/src/lib/influx.ts | 40 ++++ web/src/lib/mongo.ts | 100 -------- web/src/lib/store.ts | 225 ++++++++++++++++++ web/src/lib/types.ts | 96 ++++++++ web/src/lib/validate.ts | 71 +++++- web/test/days.test.ts | 37 +++ web/test/telemetry.test.ts | 56 +++++ web/tsconfig.json | 1 + 26 files changed, 821 insertions(+), 395 deletions(-) create mode 100644 web/.dockerignore create mode 100644 web/Dockerfile create mode 100644 web/src/app/v1/telemetry/route.ts create mode 100644 web/src/lib/abrp.ts create mode 100644 web/src/lib/days.ts create mode 100644 web/src/lib/influx.ts delete mode 100644 web/src/lib/mongo.ts create mode 100644 web/src/lib/store.ts create mode 100644 web/src/lib/types.ts create mode 100644 web/test/days.test.ts create mode 100644 web/test/telemetry.test.ts diff --git a/flake.nix b/flake.nix index b0b4a1f..88d38f0 100644 --- a/flake.nix +++ b/flake.nix @@ -112,7 +112,7 @@ in { default = pkgs.mkShell { - packages = [ jdk android.androidsdk buildCar ]; + packages = [ jdk android.androidsdk buildCar pkgs.nodejs_22 pkgs.influxdb2-server pkgs.influxdb2-cli ]; ANDROID_HOME = sdk; ANDROID_SDK_ROOT = sdk; diff --git a/web/.dockerignore b/web/.dockerignore new file mode 100644 index 0000000..6614541 --- /dev/null +++ b/web/.dockerignore @@ -0,0 +1,7 @@ +node_modules +.next +.env* +!.env.example +*.tsbuildinfo +Dockerfile +.dockerignore diff --git a/web/.env.example b/web/.env.example index 5cd72df..1a81b62 100644 --- a/web/.env.example +++ b/web/.env.example @@ -1,5 +1,18 @@ -MONGODB_URI=mongodb+srv://username:password@cluster.example.mongodb.net/ -MONGODB_DATABASE=fatihsmg4 +# InfluxDB 2 — a token with read/write on the bucket. +INFLUX_URL=http://influxdb:8086 +INFLUX_TOKEN=replace-with-an-influxdb-token +INFLUX_ORG=mg4 +INFLUX_BUCKET=mg4 + +# The car's write token (at least 24 characters): openssl rand -hex 32 INGEST_TOKEN=replace-with-a-long-random-write-token +# Signs dashboard sessions; rotating it signs every browser out: openssl rand -hex 32 SESSION_SECRET=replace-with-a-long-random-session-secret +# Dashboard login, at least 8 characters. DASHBOARD_PASSWORD=replace-with-a-strong-dashboard-password + +# ABRP forwarding; off when either is empty. +ABRP_API_KEY= +ABRP_USER_TOKEN= +# Minimum seconds between two samples forwarded to ABRP. +ABRP_MIN_INTERVAL_SECONDS=5 diff --git a/web/Dockerfile b/web/Dockerfile new file mode 100644 index 0000000..22663d7 --- /dev/null +++ b/web/Dockerfile @@ -0,0 +1,27 @@ +# Self-hosted image of the car server: Next.js standalone output on Node 22. +# Configuration comes entirely from the environment at runtime (see .env.example); +# nothing secret is baked in. + +FROM node:22-alpine AS deps +WORKDIR /app +COPY package.json package-lock.json ./ +RUN npm ci --no-audit --no-fund + +FROM node:22-alpine AS build +WORKDIR /app +ENV NEXT_TELEMETRY_DISABLED=1 +COPY --from=deps /app/node_modules ./node_modules +COPY . . +RUN npm test && npm run build + +FROM node:22-alpine +WORKDIR /app +ENV NODE_ENV=production \ + NEXT_TELEMETRY_DISABLED=1 \ + PORT=3000 \ + HOSTNAME=0.0.0.0 +COPY --from=build --chown=10001:10001 /app/.next/standalone ./ +COPY --from=build --chown=10001:10001 /app/.next/static ./.next/static +USER 10001:10001 +EXPOSE 3000 +CMD ["node", "server.js"] diff --git a/web/README.md b/web/README.md index 9715a88..0b2d722 100644 --- a/web/README.md +++ b/web/README.md @@ -1,68 +1,88 @@ -# FatihsMG4 web +# Nils' MG4 — car server -Next.js app on Vercel: the ingest API the car writes to, and a phone-shaped dashboard for -reading the history back. Replaces the standalone Express service that used to live in -`server/`. +Next.js app, self-hosted at car.stinnesbeck.com: the ingest API the car writes to, the ABRP +forwarder, and a phone-shaped dashboard for reading the history back. Data lives in +InfluxDB 2. Based on upstream's Vercel/MongoDB app in FatihsMG4. -**The car never connects to MongoDB and never holds a MongoDB URI.** The URI is a Vercel -environment variable and is only ever read by a server-side route. +``` +car ──live telemetry───▶ server ──▶ ABRP api.iternio.com/1/tlm/send + ──daily chunks─────▶ │ + ──charging sessions▶ ▼ + InfluxDB 2 +``` -## Two credentials, on purpose +## Credentials | | Who holds it | What it opens | |---|---|---| -| `INGEST_TOKEN` | the car, in the APK | the `PUT` endpoints — write only | -| `DASHBOARD_PASSWORD` | you, in a browser | the dashboard and the `GET` history | +| `INGEST_TOKEN` | the car, entered in the app | the write endpoints, and the car's own history read | +| `DASHBOARD_PASSWORD` | you, in a browser | the dashboard and the `GET /v1/history` reads | +| `ABRP_API_KEY`, `ABRP_USER_TOKEN` | only the server | forwarding live telemetry to ABRP | -The APK is published, so its token must be assumed extractable: anyone can `unzip` an APK -and read its strings. That is survivable here because the write token cannot read anything -back. The password is never compiled into anything. +The ABRP credentials never reach the car. A missing or empty one switches forwarding off. ## Environment variables -Set all five in Vercel → Settings → Environment Variables, for Production and Preview. +See `.env.example`. In Kubernetes they come from the Secret in `../k8s/`. | Name | Notes | |---|---| -| `MONGODB_URI` | Atlas connection string | -| `MONGODB_DATABASE` | defaults to `fatihsmg4` | -| `INGEST_TOKEN` | `openssl rand -hex 32` | +| `INFLUX_URL`, `INFLUX_TOKEN`, `INFLUX_ORG`, `INFLUX_BUCKET` | bucket defaults to `mg4`; the token needs read/write on it | +| `INGEST_TOKEN` | at least 24 characters, or every write is refused: `openssl rand -hex 32` | | `DASHBOARD_PASSWORD` | at least 8 characters, or login is refused outright | | `SESSION_SECRET` | `openssl rand -hex 32`; rotating it signs every browser out | - -Atlas network access must allow Vercel's egress, which is not a fixed address on the Hobby -plan — either allow `0.0.0.0/0` and rely on the database user's credentials, or use Atlas -private networking on a paid plan. +| `ABRP_API_KEY`, `ABRP_USER_TOKEN` | optional; both needed to forward | +| `ABRP_MIN_INTERVAL_SECONDS` | default 5 | ## Endpoints Write, `Authorization: Bearer `: -- `PUT /v1/consumption/daily` +- `POST /v1/telemetry` — `{ installationId, samples: [...] }`, up to 500 samples in ABRP's `tlm` + field names and units (`utc` in epoch seconds; flags as 0/1 or booleans). Stored, then the + newest sample is forwarded to ABRP if it is under 5 minutes old. +- `PUT /v1/consumption/daily` — one delta chunk of a day's consumption. - `PUT /v1/charging-sessions/:sessionId` -Read, session cookie or `Authorization: Bearer `: +Read: -- `GET /v1/history/:installationId/consumption` -- `GET /v1/history/:installationId/charging` +- `GET /v1/consumption/history/:installationId` — ingest token; the car reads its own totals back. +- `GET /v1/history/:installationId/consumption`, `.../charging` — session cookie or + `Authorization: Bearer `. `GET /v1/health` is unauthenticated and returns `{"status":"ok"}`. -The car builds its URLs as `CONSUMPTION_API_URL + /v1/...`, so the Android build variable -is the bare origin — `https://your-app.vercel.app`, no trailing path. +## Storage -## Indexes +One bucket, `installationId` tag on every point: -There is no boot step on Vercel, so `ensureIndexes()` runs on the first write of each warm -lambda and is skipped afterwards. The unique indexes on `(installationId, date)` and -`(installationId, sessionId)` are what make the upserts idempotent when the car retries a -record it could not confirm. +| Measurement | Timestamp | Tags | Fields | +|---|---|---|---| +| `telemetry` | sample `utc` | — | the ABRP fields that were sent | +| `consumption_chunk` | `recordedAt` | `date` | km, kwh, hours, socDrop, sohPercent | +| `charging_session` | `startedAt` | `sessionId` | endedAtMs, duration, energy, cost, start/end SOC | +| `charging_curve` | point time | `sessionId` | powerKw, socPercent | + +A point with the same measurement, tags and timestamp replaces the earlier one, so the car's +retries are idempotent: a cached chunk is re-sent with its original `recordedAt`. Days and the +lifetime total are summed from the chunks when read. All numbers are written as floats, because +InfluxDB fixes a field's type on first write. ## Local development ```sh +nix develop # from the repository root: Node 22 and InfluxDB 2 cp .env.example .env.local # Next loads .env.local automatically npm install npm run dev -npm test # pure validation and session-signing tests, no database +npm test # validation, aggregation and forwarding rules; no database ``` + +## Container image + +```sh +podman build -t git.stinnesbeck.com/nils/mg4-server: . +podman push git.stinnesbeck.com/nils/mg4-server: +``` + +The build runs the unit tests. The image runs as UID 10001 and listens on port 3000. diff --git a/web/next.config.mjs b/web/next.config.mjs index 1f0ce49..7bd72cb 100644 --- a/web/next.config.mjs +++ b/web/next.config.mjs @@ -1,5 +1,7 @@ /** @type {import('next').NextConfig} */ const nextConfig = { + // A self-contained server in .next/standalone, for the container image. + output: "standalone", // Every page reads live telemetry; a cached shell would show yesterday's drive. experimental: { serverActions: { bodySizeLimit: "1mb" } }, }; diff --git a/web/package-lock.json b/web/package-lock.json index d4aa3dd..8ad8c25 100644 --- a/web/package-lock.json +++ b/web/package-lock.json @@ -8,7 +8,7 @@ "name": "fatihsmg4-web", "version": "1.0.0", "dependencies": { - "mongodb": "^6.20.0", + "@influxdata/influxdb-client": "^1.35.0", "next": "^15.5.4", "react": "^19.1.0", "react-dom": "^19.1.0" @@ -533,14 +533,11 @@ "url": "https://opencollective.com/libvips" } }, - "node_modules/@mongodb-js/saslprep": { - "version": "1.6.0", - "resolved": "https://registry.npmjs.org/@mongodb-js/saslprep/-/saslprep-1.6.0.tgz", - "integrity": "sha512-19jGXfpctLDJUELLUZGJ5BllcLT7rHdF+gTTaN2jzqu0bGkUhpdcYtgK2Vux+g3zuBAa0TM/W666QMIjYhYujw==", - "license": "MIT", - "dependencies": { - "sparse-bitfield": "^3.0.3" - } + "node_modules/@influxdata/influxdb-client": { + "version": "1.35.0", + "resolved": "https://registry.npmjs.org/@influxdata/influxdb-client/-/influxdb-client-1.35.0.tgz", + "integrity": "sha512-woWMi8PDpPQpvTsRaUw4Ig+nOGS/CWwAwS66Fa1Vr/EkW+NEwxI8YfPBsdBMn33jK2Y86/qMiiuX/ROHIkJLTw==", + "license": "MIT" }, "node_modules/@next/env": { "version": "15.5.27", @@ -705,30 +702,6 @@ "csstype": "^3.2.2" } }, - "node_modules/@types/webidl-conversions": { - "version": "7.0.3", - "resolved": "https://registry.npmjs.org/@types/webidl-conversions/-/webidl-conversions-7.0.3.tgz", - "integrity": "sha512-CiJJvcRtIgzadHCYXw7dqEnMNRjhGZlYK05Mj9OyktqV8uVT8fD2BFOB7S1uwBE3Kj2Z+4UyPmFw/Ixgw/LAlA==", - "license": "MIT" - }, - "node_modules/@types/whatwg-url": { - "version": "11.0.5", - "resolved": "https://registry.npmjs.org/@types/whatwg-url/-/whatwg-url-11.0.5.tgz", - "integrity": "sha512-coYR071JRaHa+xoEvvYqvnIHaVqaYrLPbsufM9BF63HkwI5Lgmy2QR8Q5K/lYDYo5AK82wOvSOS0UsLTpTG7uQ==", - "license": "MIT", - "dependencies": { - "@types/webidl-conversions": "*" - } - }, - "node_modules/bson": { - "version": "6.10.4", - "resolved": "https://registry.npmjs.org/bson/-/bson-6.10.4.tgz", - "integrity": "sha512-WIsKqkSC0ABoBJuT1LEX+2HEvNmNKKgnTAyd0fL8qzK4SH2i9NXg+t08YtdZp/V9IZ33cxe3iV4yM0qg8lMQng==", - "license": "Apache-2.0", - "engines": { - "node": ">=16.20.1" - } - }, "node_modules/caniuse-lite": { "version": "1.0.30001815", "resolved": "https://registry.npmjs.org/caniuse-lite/-/caniuse-lite-1.0.30001815.tgz", @@ -772,68 +745,6 @@ "node": ">=8" } }, - "node_modules/memory-pager": { - "version": "1.5.0", - "resolved": "https://registry.npmjs.org/memory-pager/-/memory-pager-1.5.0.tgz", - "integrity": "sha512-ZS4Bp4r/Zoeq6+NLJpP+0Zzm0pR8whtGPf1XExKLJBAczGMnSi3It14OiNCStjQjM6NU1okjQGSxgEZN8eBYKg==", - "license": "MIT" - }, - "node_modules/mongodb": { - "version": "6.21.0", - "resolved": "https://registry.npmjs.org/mongodb/-/mongodb-6.21.0.tgz", - "integrity": "sha512-URyb/VXMjJ4da46OeSXg+puO39XH9DeQpWCslifrRn9JWugy0D+DvvBvkm2WxmHe61O/H19JM66p1z7RHVkZ6A==", - "license": "Apache-2.0", - "dependencies": { - "@mongodb-js/saslprep": "^1.3.0", - "bson": "^6.10.4", - "mongodb-connection-string-url": "^3.0.2" - }, - "engines": { - "node": ">=16.20.1" - }, - "peerDependencies": { - "@aws-sdk/credential-providers": "^3.188.0", - "@mongodb-js/zstd": "^1.1.0 || ^2.0.0", - "gcp-metadata": "^5.2.0", - "kerberos": "^2.0.1", - "mongodb-client-encryption": ">=6.0.0 <7", - "snappy": "^7.3.2", - "socks": "^2.7.1" - }, - "peerDependenciesMeta": { - "@aws-sdk/credential-providers": { - "optional": true - }, - "@mongodb-js/zstd": { - "optional": true - }, - "gcp-metadata": { - "optional": true - }, - "kerberos": { - "optional": true - }, - "mongodb-client-encryption": { - "optional": true - }, - "snappy": { - "optional": true - }, - "socks": { - "optional": true - } - } - }, - "node_modules/mongodb-connection-string-url": { - "version": "3.0.2", - "resolved": "https://registry.npmjs.org/mongodb-connection-string-url/-/mongodb-connection-string-url-3.0.2.tgz", - "integrity": "sha512-rMO7CGo/9BFwyZABcKAWL8UJwH/Kc2x0g72uhDWzG48URRax5TCIcJ7Rc3RZqffZzO/Gwff/jyKwCU9TN8gehA==", - "license": "Apache-2.0", - "dependencies": { - "@types/whatwg-url": "^11.0.2", - "whatwg-url": "^14.1.0 || ^13.0.0" - } - }, "node_modules/nanoid": { "version": "3.3.20", "resolved": "https://registry.npmjs.org/nanoid/-/nanoid-3.3.20.tgz", @@ -938,21 +849,11 @@ "node": "^10 || ^12 || >=14" } }, - "node_modules/punycode": { - "version": "2.3.1", - "resolved": "https://registry.npmjs.org/punycode/-/punycode-2.3.1.tgz", - "integrity": "sha512-vYt7UD1U9Wg6138shLtLOvdAu+8DsC/ilFtEVHcH+wydcSpNE20AfSOduf6MkRFahL5FY7X1oU7nKVZFtfq8Fg==", - "license": "MIT", - "engines": { - "node": ">=6" - } - }, "node_modules/react": { "version": "19.3.0", "resolved": "https://registry.npmjs.org/react/-/react-19.3.0.tgz", "integrity": "sha512-E8LUcbtBWt20bbl2YoHfx4ZDBdxVTfOKtCZn9cDSJ4l6/nuoApcpIBcj47t2wZoVX8g2ZHuMHbiShgCR1T5Sog==", "license": "MIT", - "peer": true, "engines": { "node": ">=0.10.0" } @@ -962,7 +863,6 @@ "resolved": "https://registry.npmjs.org/react-dom/-/react-dom-19.3.0.tgz", "integrity": "sha512-JDk8dgif51OjFoDE70+OT9ICyYr+69HlmihNwp1+Nsfbna3t5sIiCa9ZJktDmQ4/1b/rn26hIAR2uYXDMr5r0Q==", "license": "MIT", - "peer": true, "dependencies": { "scheduler": "^0.28.0" }, @@ -1048,15 +948,6 @@ "node": ">=0.10.0" } }, - "node_modules/sparse-bitfield": { - "version": "3.0.3", - "resolved": "https://registry.npmjs.org/sparse-bitfield/-/sparse-bitfield-3.0.3.tgz", - "integrity": "sha512-kvzhi7vqKTfkh0PZU+2D2PIllw2ymqJKujUcyPMd9Y75Nv4nPbGJZXNhxsgdQab2BmlDct1YnfQCguEvHr7VsQ==", - "license": "MIT", - "dependencies": { - "memory-pager": "^1.0.2" - } - }, "node_modules/styled-jsx": { "version": "5.1.6", "resolved": "https://registry.npmjs.org/styled-jsx/-/styled-jsx-5.1.6.tgz", @@ -1080,18 +971,6 @@ } } }, - "node_modules/tr46": { - "version": "5.1.1", - "resolved": "https://registry.npmjs.org/tr46/-/tr46-5.1.1.tgz", - "integrity": "sha512-hdF5ZgjTqgAntKkklYw0R03MG2x/bSzTtkxmIRw/sTNV8YXsCJ1tfLAX23lhxhHJlEf3CRCOCGGWw3vI3GaSPw==", - "license": "MIT", - "dependencies": { - "punycode": "^2.3.1" - }, - "engines": { - "node": ">=18" - } - }, "node_modules/tslib": { "version": "2.8.1", "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz", @@ -1118,28 +997,6 @@ "integrity": "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==", "dev": true, "license": "MIT" - }, - "node_modules/webidl-conversions": { - "version": "7.0.0", - "resolved": "https://registry.npmjs.org/webidl-conversions/-/webidl-conversions-7.0.0.tgz", - "integrity": "sha512-VwddBukDzu71offAQR975unBIGqfKZpM+8ZX6ySk8nYhVoo5CYaZyzt3YBvYtRtO+aoGlqxPg/B87NGVZ/fu6g==", - "license": "BSD-2-Clause", - "engines": { - "node": ">=12" - } - }, - "node_modules/whatwg-url": { - "version": "14.2.0", - "resolved": "https://registry.npmjs.org/whatwg-url/-/whatwg-url-14.2.0.tgz", - "integrity": "sha512-De72GdQZzNTUBBChsXueQUnPKDkg/5A5zp7pFDuQAj5UFoENpiACU0wlCvzpAGnTkj++ihpKwKyYewn/XNUbKw==", - "license": "MIT", - "dependencies": { - "tr46": "^5.1.0", - "webidl-conversions": "^7.0.0" - }, - "engines": { - "node": ">=18" - } } } } diff --git a/web/package.json b/web/package.json index 0cf8888..8854c8f 100644 --- a/web/package.json +++ b/web/package.json @@ -13,7 +13,7 @@ "node": ">=22" }, "dependencies": { - "mongodb": "^6.20.0", + "@influxdata/influxdb-client": "^1.35.0", "next": "^15.5.4", "react": "^19.1.0", "react-dom": "^19.1.0" @@ -24,4 +24,4 @@ "typescript": "^5.8.3" }, "type": "module" -} \ No newline at end of file +} diff --git a/web/src/app/v1/charging-sessions/[sessionId]/route.ts b/web/src/app/v1/charging-sessions/[sessionId]/route.ts index d36ff88..55ee73a 100644 --- a/web/src/app/v1/charging-sessions/[sessionId]/route.ts +++ b/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 }); }); } diff --git a/web/src/app/v1/consumption/daily/route.ts b/web/src/app/v1/consumption/daily/route.ts index 7397ba6..691fab5 100644 --- a/web/src/app/v1/consumption/daily/route.ts +++ b/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 = { 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 }); }); } - diff --git a/web/src/app/v1/consumption/history/[installationId]/route.ts b/web/src/app/v1/consumption/history/[installationId]/route.ts index e44811f..ffd2bc3 100644 --- a/web/src/app/v1/consumption/history/[installationId]/route.ts +++ b/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 }); }); } diff --git a/web/src/app/v1/history/[installationId]/charging/route.ts b/web/src/app/v1/history/[installationId]/charging/route.ts index 73dd795..434bf4d 100644 --- a/web/src/app/v1/history/[installationId]/charging/route.ts +++ b/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 }); }); } diff --git a/web/src/app/v1/history/[installationId]/consumption/route.ts b/web/src/app/v1/history/[installationId]/consumption/route.ts index 19d108c..158b0a8 100644 --- a/web/src/app/v1/history/[installationId]/consumption/route.ts +++ b/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 }); }); } diff --git a/web/src/app/v1/telemetry/route.ts b/web/src/app/v1/telemetry/route.ts new file mode 100644 index 0000000..ffbf8fb --- /dev/null +++ b/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 }); + }); +} diff --git a/web/src/lib/abrp.ts b/web/src/lib/abrp.ts new file mode 100644 index 0000000..87a7756 --- /dev/null +++ b/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 { + const tlm: Record = {}; + 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 { + // 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; +} diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 327e63b..a266559 100644 --- a/web/src/lib/api.ts +++ b/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"; diff --git a/web/src/lib/data.ts b/web/src/lib/data.ts index b23bf5b..50c1d07 100644 --- a/web/src/lib/data.ts +++ b/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 { 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 { diff --git a/web/src/lib/days.ts b/web/src/lib/days.ts new file mode 100644 index 0000000..e5a36fd --- /dev/null +++ b/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(); + 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(); +} diff --git a/web/src/lib/influx.ts b/web/src/lib/influx.ts new file mode 100644 index 0000000..4bb8c7d --- /dev/null +++ b/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")); +} diff --git a/web/src/lib/mongo.ts b/web/src/lib/mongo.ts deleted file mode 100644 index 508dcc6..0000000 --- a/web/src/lib/mongo.ts +++ /dev/null @@ -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 }; - -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 { - 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 { - const client = await clientPromise(); - return client.db(process.env.MONGODB_DATABASE || "fatihsmg4"); -} - -export async function consumptionCollection(): Promise> { - return (await db()).collection("daily_consumption"); -} - -export async function chargingCollection(): Promise> { - return (await db()).collection("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 { - 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; -} diff --git a/web/src/lib/store.ts b/web/src/lib/store.ts new file mode 100644 index 0000000..7836506 --- /dev/null +++ b/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 { + const api = writeApi(); + api.writePoints(points); + await api.close(); +} + +export async function writeConsumptionChunk(chunk: ConsumptionChunk): Promise { + 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 { + 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 { + 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 & { _time: string }; + +async function rows(query: ReturnType): Promise { + return queryApi().collectRows(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 { + 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 { + return aggregateDays(await readConsumptionChunks(installationId)); +} + +export async function readChargingSessions(installationId: string, limit: number): Promise { + 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 { + 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; + 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)[field] = value; + if (flags.includes(field) && typeof value === "boolean") (sample as Record)[field] = value; + } + return sample; +} diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts new file mode 100644 index 0000000..c523a46 --- /dev/null +++ b/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; + +/** 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> & + Partial>; + +export interface TelemetryBatch { + installationId: string; + samples: TelemetrySample[]; +} diff --git a/web/src/lib/validate.ts b/web/src/lib/validate.ts index 1184fc0..945657e 100644 --- a/web/src/lib/validate.ts +++ b/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; -export type ChargingInput = Omit; - -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)[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; diff --git a/web/test/days.test.ts b/web/test/days.test.ts new file mode 100644 index 0000000..6fb6486 --- /dev/null +++ b/web/test/days.test.ts @@ -0,0 +1,37 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { aggregateDays } from "../src/lib/days.ts"; + +const ID = "6f1a2b3c-4d5e-4f60-8a9b-0c1d2e3f4a5b"; +const chunk = (date: string, at: number, km: number, kwh: number, sohPercent?: number) => ({ + installationId: ID, + date, + recordedAt: new Date(at), + day: { km, kwh, hours: km / 50, socDrop: kwh / 0.6 }, + ...(sohPercent === undefined ? {} : { sohPercent }), +}); + +test("adds chunks up per day, newest day first, with a running lifetime", () => { + const days = aggregateDays([ + chunk("2026-10-16", 1_000, 10, 2), + chunk("2026-10-17", 3_000, 5, 1), + chunk("2026-10-16", 2_000, 20, 3), + ]); + assert.deepEqual(days.map((d) => d.date), ["2026-10-17", "2026-10-16"]); + assert.equal(days[1].day.km, 30); + assert.equal(days[1].day.kwh, 5); + assert.equal(days[1].recordedAt.getTime(), 2_000); + assert.equal(days[1].lifetime?.km, 30); + assert.equal(days[0].lifetime?.km, 35); + assert.equal(days[0].lifetime?.kwh, 6); +}); + +test("the newest manual SoH of a day wins, and a day without one has none", () => { + const days = aggregateDays([ + chunk("2026-10-16", 2_000, 1, 1, 96), + chunk("2026-10-16", 1_000, 1, 1, 98), + chunk("2026-10-17", 3_000, 1, 1), + ]); + assert.equal(days[1].sohPercent, 96); + assert.equal("sohPercent" in days[0], false); +}); diff --git a/web/test/telemetry.test.ts b/web/test/telemetry.test.ts new file mode 100644 index 0000000..7adea8a --- /dev/null +++ b/web/test/telemetry.test.ts @@ -0,0 +1,56 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { pickForForward, resetAbrpState, toTlm } from "../src/lib/abrp.ts"; +import { ValidationError, validateTelemetryBatch } from "../src/lib/validate.ts"; + +const INSTALLATION = "6f1a2b3c-4d5e-4f60-8a9b-0c1d2e3f4a5b"; +const NOW_S = 1_790_000_000; + +test("accepts a sample in ABRP's field names and keeps only what was sent", () => { + const batch = validateTelemetryBatch({ + installationId: INSTALLATION, + samples: [{ utc: NOW_S, soc: 81, power: -12.5, speed: 0, lat: 52.1, lon: 7.6, is_charging: 1, is_dcfc: false }], + }); + assert.deepEqual(batch.samples[0], { + utc: NOW_S, soc: 81, power: -12.5, speed: 0, lat: 52.1, lon: 7.6, is_charging: true, is_dcfc: false, + }); + // An unread signal stays absent: a missing SOC must never become 0. + assert.equal("est_battery_range" in batch.samples[0], false); +}); + +test("drops unknown fields instead of rejecting the batch", () => { + const batch = validateTelemetryBatch({ installationId: INSTALLATION, samples: [{ utc: NOW_S, future_field: 3 }] }); + assert.deepEqual(batch.samples[0], { utc: NOW_S }); +}); + +test("rejects bad timestamps, out-of-range values, bad flags and half positions", () => { + const send = (sample: object) => validateTelemetryBatch({ installationId: INSTALLATION, samples: [sample] }); + assert.throws(() => send({ utc: NOW_S * 1000 }), ValidationError); // milliseconds, not seconds + assert.throws(() => send({ utc: 1.5 }), ValidationError); + assert.throws(() => send({ utc: NOW_S, soc: 101 }), ValidationError); + assert.throws(() => send({ utc: NOW_S, speed: "fast" }), ValidationError); + assert.throws(() => send({ utc: NOW_S, is_parked: 2 }), ValidationError); + assert.throws(() => send({ utc: NOW_S, lat: 52.1 }), ValidationError); +}); + +test("rejects empty and oversized batches and bad installation ids", () => { + assert.throws(() => validateTelemetryBatch({ installationId: INSTALLATION, samples: [] }), ValidationError); + const many = Array.from({ length: 501 }, (_, i) => ({ utc: NOW_S + i })); + assert.throws(() => validateTelemetryBatch({ installationId: INSTALLATION, samples: many }), ValidationError); + assert.throws(() => validateTelemetryBatch({ installationId: "nope", samples: [{ utc: NOW_S }] }), ValidationError); +}); + +test("ABRP gets flags as 0/1", () => { + assert.deepEqual(toTlm({ utc: NOW_S, soc: 50, is_charging: true, is_parked: false }), { + utc: NOW_S, soc: 50, is_charging: 1, is_parked: 0, + }); +}); + +test("forwards only the newest fresh sample, at most once per interval", () => { + resetAbrpState(); + const nowMs = NOW_S * 1000; + const picked = pickForForward([{ utc: NOW_S - 10 }, { utc: NOW_S - 2 }, { utc: NOW_S - 5 }], nowMs); + assert.equal(picked?.utc, NOW_S - 2); + // A sample from ten minutes ago describes a car that has moved on. + assert.equal(pickForForward([{ utc: NOW_S - 600 }], nowMs), null); +}); diff --git a/web/tsconfig.json b/web/tsconfig.json index 90069c0..deb57ab 100644 --- a/web/tsconfig.json +++ b/web/tsconfig.json @@ -6,6 +6,7 @@ "skipLibCheck": true, "strict": true, "noEmit": true, + "allowImportingTsExtensions": true, "esModuleInterop": true, "module": "esnext", "moduleResolution": "bundler",