diff --git a/Cargo.lock b/Cargo.lock index 9eb299a9..10c7ab54 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -54,6 +54,7 @@ dependencies = [ "clap", "colored", "futures", + "reqwest", "serde", "serde_json", "serde_yaml", @@ -62,6 +63,7 @@ dependencies = [ "thiserror 1.0.69", "tokio", "tokio-test", + "toml 0.8.23", ] [[package]] @@ -1806,8 +1808,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "335ff9f135e4384c8150d6f27c6daed433577f86b4750418338c01a1a2528592" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -1817,9 +1821,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 5.3.0", "wasip2", + "wasm-bindgen", ] [[package]] @@ -2053,6 +2059,7 @@ dependencies = [ "hyper", "hyper-util", "rustls", + "rustls-native-certs", "tokio", "tokio-rustls", "tower-service", @@ -2422,6 +2429,12 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "mach2" version = "0.4.3" @@ -2547,10 +2560,10 @@ dependencies = [ "libc", "log", "openssl", - "openssl-probe", + "openssl-probe 0.1.6", "openssl-sys", "schannel", - "security-framework", + "security-framework 2.11.1", "security-framework-sys", "tempfile", ] @@ -2665,6 +2678,12 @@ version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" +[[package]] +name = "openssl-probe" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" + [[package]] name = "openssl-sys" version = "0.9.116" @@ -2813,6 +2832,61 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "quinn" +version = "0.11.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e20a958963c291dc322d98411f541009df2ced7b5a4f2bd52337638cfccf20" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror 2.0.17", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fcb935c5bec503c2f0e306bdd3e58bb9029dcb14fa8d9ac76e3a5256ac0763e" +dependencies = [ + "bytes", + "getrandom 0.3.4", + "lru-slab", + "rand 0.9.4", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror 2.0.17", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "addec6a0dcad8a8d96a771f815f0eaf55f9d1805756410b39f5fa81332574cbd" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.60.2", +] + [[package]] name = "quote" version = "1.0.41" @@ -2973,6 +3047,9 @@ dependencies = [ "native-tls", "percent-encoding", "pin-project-lite", + "quinn", + "rustls", + "rustls-native-certs", "rustls-pki-types", "serde", "serde_json", @@ -2980,6 +3057,7 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-native-tls", + "tokio-rustls", "tower", "tower-http", "tower-service", @@ -3048,6 +3126,12 @@ dependencies = [ "smallvec", ] +[[package]] +name = "rustc-hash" +version = "2.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "357703d41365b4b27c590e3ed91eabb1b663f07c4c084095e60cbed4362dff0d" + [[package]] name = "rustc_version" version = "0.2.3" @@ -3077,18 +3161,32 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4" dependencies = [ "once_cell", + "ring", "rustls-pki-types", "rustls-webpki", "subtle", "zeroize", ] +[[package]] +name = "rustls-native-certs" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" +dependencies = [ + "openssl-probe 0.2.1", + "rustls-pki-types", + "schannel", + "security-framework 3.5.1", +] + [[package]] name = "rustls-pki-types" version = "1.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94182ad936a0c91c324cd46c6511b9510ed16af436d7b5bab34beab0afd55f7a" dependencies = [ + "web-time", "zeroize", ] @@ -3164,6 +3262,19 @@ dependencies = [ "security-framework-sys", ] +[[package]] +name = "security-framework" +version = "3.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3297343eaf830f66ede390ea39da1d462b6b0c1b000f420d0a83f898bbbe6ef" +dependencies = [ + "bitflags 2.11.0", + "core-foundation 0.10.1", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + [[package]] name = "security-framework-sys" version = "2.15.0" @@ -3679,6 +3790,21 @@ dependencies = [ "serde_json", ] +[[package]] +name = "tinyvec" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa5fdc3bce6191a1dbc8c02d5c8bffcf557bafa17c124c5264a458f1b0613fa" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.48.0" @@ -4406,6 +4532,7 @@ dependencies = [ "aimdb-core", "aimdb-data-contracts", "aimdb-mqtt-connector", + "aimdb-tcp-connector", "aimdb-tokio-adapter", "tokio", "tracing", @@ -4424,6 +4551,27 @@ dependencies = [ "serde_json", ] +[[package]] +name = "weather-station" +version = "0.1.0" +dependencies = [ + "aimdb-core", + "aimdb-data-contracts", + "aimdb-mqtt-connector", + "aimdb-tokio-adapter", + "chrono", + "clap", + "reqwest", + "rumqttc", + "serde", + "serde_json", + "tokio", + "toml 0.8.23", + "tracing", + "tracing-subscriber", + "weather-mesh-common", +] + [[package]] name = "weather-station-alpha" version = "0.1.0" @@ -4487,6 +4635,7 @@ dependencies = [ "rand 0.10.1", "static_cell", "stm32-fmc 0.3.2", + "toml 0.8.23", "weather-mesh-common", ] @@ -4500,6 +4649,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "winapi" version = "0.3.9" diff --git a/Cargo.toml b/Cargo.toml index e2032720..45e56e4a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,6 +32,7 @@ members = [ "examples/remote-access-demo", "examples/weather-mesh-demo/weather-mesh-common", "examples/weather-mesh-demo/weather-hub", + "examples/weather-mesh-demo/weather-station", "examples/weather-mesh-demo/weather-station-alpha", "examples/weather-mesh-demo/weather-station-beta", "examples/weather-mesh-demo/weather-station-gamma", diff --git a/Makefile b/Makefile index 2999ff7f..a3a56b3c 100644 --- a/Makefile +++ b/Makefile @@ -178,7 +178,7 @@ test: fmt: @printf "$(GREEN)Formatting code (workspace members only)...$(NC)\n" - @for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async aimdb-bench; do \ + @for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async aimdb-bench; do \ printf "$(YELLOW) → Formatting $$pkg$(NC)\n"; \ cargo fmt -p $$pkg 2>/dev/null || true; \ done @@ -187,7 +187,7 @@ fmt: fmt-check: @printf "$(GREEN)Checking code formatting (workspace members only)...$(NC)\n" @FAILED=0; \ - for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async aimdb-bench; do \ + for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async aimdb-bench; do \ printf "$(YELLOW) → Checking $$pkg$(NC)\n"; \ if ! cargo fmt -p $$pkg -- --check 2>&1; then \ printf "$(RED)❌ Formatting check failed for $$pkg$(NC)\n"; \ @@ -398,6 +398,8 @@ examples: cargo build --package weather-station-alpha @printf "$(YELLOW) → Building weather-mesh-demo: weather-station-beta (edge, synthetic)$(NC)\n" cargo build --package weather-station-beta + @printf "$(YELLOW) → Building weather-mesh-demo: weather-station (public mesh, profile-driven)$(NC)\n" + cargo build --package weather-station @printf "$(YELLOW) → Building weather-station-gamma (embedded, embassy runtime)$(NC)\n" cargo build --package weather-station-gamma --target thumbv7em-none-eabihf @printf "$(YELLOW) → Building remote-access-demo (AimX server + client)$(NC)\n" diff --git a/README.md b/README.md index f30d8ea0..aecab713 100644 --- a/README.md +++ b/README.md @@ -239,6 +239,8 @@ See the [MCP server docs](tools/aimdb-mcp/) for Claude Desktop and other editors | **MQTT** — `aimdb-mqtt-connector` | ✅ Ready | std, no_std | | **KNX** — `aimdb-knx-connector` | ✅ Ready | std, no_std | | **WebSocket** — `aimdb-websocket-connector` | ✅ Ready | std, wasm | +| **Serial (COBS/UART)** — `aimdb-serial-connector` | ✅ Ready | std, no_std | +| **Unix domain socket** — `aimdb-uds-connector` | ✅ Ready | std | | **TCP** — `aimdb-tcp-connector` | ✅ Ready | std, no_std | | **Kafka** | 📋 Planned | std | | **Modbus** | 📋 Planned | std, no_std | diff --git a/aimdb-mqtt-connector/src/tokio_client.rs b/aimdb-mqtt-connector/src/tokio_client.rs index 923e0ce7..e4b2ca00 100644 --- a/aimdb-mqtt-connector/src/tokio_client.rs +++ b/aimdb-mqtt-connector/src/tokio_client.rs @@ -177,6 +177,12 @@ impl MqttConnectorImpl { mqtt_opts.set_credentials(username, password); } + // mqtts:// selects the TLS transport (system trust roots via + // native-tls); rumqttc otherwise speaks plain TCP regardless of port. + if connector_url.scheme == "mqtts" { + mqtt_opts.set_transport(rumqttc::Transport::Tls(rumqttc::TlsConfiguration::Native)); + } + // Wrap router early so we can count topics for capacity calculation let router_arc = Arc::new(router); let topic_count = router_arc.resource_ids().len(); @@ -364,4 +370,18 @@ mod tests { let connector = MqttConnectorImpl::build_internal("not-a-valid-url", None, router).await; assert!(connector.is_err()); } + + #[tokio::test] + async fn test_connector_mqtts_url_with_credentials() { + // mqtts:// with URL-embedded credentials must parse and build; the TLS + // handshake itself only happens once the event loop is polled. + let router = RouterBuilder::new().build(); + let connector = MqttConnectorImpl::build_internal( + "mqtts://hub-sub:secret@broker.example.com:8883", + None, + router, + ) + .await; + assert!(connector.is_ok()); + } } diff --git a/docs/design/042-public-weather-mesh-flagship.md b/docs/design/042-public-weather-mesh-flagship.md new file mode 100644 index 00000000..56284925 --- /dev/null +++ b/docs/design/042-public-weather-mesh-flagship.md @@ -0,0 +1,268 @@ +# 042 — Public weather mesh flagship + +**Status:** 📝 Proposed + +**Scope:** `examples/weather-mesh-demo` (hub slot pool, station config), `tools/aimdb-cli` (new `join` subcommand), a small closed-source provisioning service (private ops repo), EMQX Cloud broker setup, and the public landing/dashboard pages. Possible follow-up in `aimdb-mqtt-connector` (Embassy TLS, §8.1). + +**Goal:** turn the three-station weather-mesh demo into a single public, hosted mesh that anyone can join with one command — a live demonstration of typed data contracts across MCU → edge → cloud, and a source of real propose/admit traffic for dogfooding. + +--- + +## 1. Context + +`examples/weather-mesh-demo` is a working three-tier demo: typed, versioned contracts (`Temperature`, `Humidity`, `DewPoint`, `GpsLocation`, including a `TemperatureV1 → V2` migration), an Open-Meteo-fed station (`alpha`), an Embassy/STM32H563 hardware station (`gamma`), and a Tokio hub. A hosted instance already streams live to a browser dashboard, and the mesh is queryable by any LLM agent via the public MCP endpoint (`aimdb.dev/mcp`, over the `transport-tcp` client transport — [`aimdb-client/src/endpoint.rs`](../../aimdb-client/src/endpoint.rs)). + +What separates this from "a public mesh anyone can join" is membership: the hub configures exactly three stations via closed `TempKey`/`HumidityKey`/`DewPointKey` enums with compile-time `#[link_address]` topics ([`weather-mesh-common/src/lib.rs`](../../examples/weather-mesh-demo/weather-mesh-common/src/lib.rs)). Admitting a fourth station today means a new enum variant and a recompile. The demo broker config is also demo-grade: [`docker/mosquitto.conf`](../../examples/weather-mesh-demo/docker/mosquitto.conf) allows anonymous, unlimited connections. + +This spec closes both gaps and defines the join UX end to end. + +## 2. Decisions + +| # | Decision | Rationale | +|---|---|---| +| D1 | **One flagship instance**, operated by the project — not a multi-tenant, self-serve platform. | Scope control; the value is a live artifact, not a product. | +| D2 | Hub membership becomes a **bounded slot pool of N string-keyed records** (`StringKey::intern`), configured at startup. | No core changes, no enum surgery; N doubles as the abuse cap. See §4. | +| D3 | Genuinely dynamic (post-`run()`) registration is **out of scope** — deferred to the knowledge-graph layer work, which faces the same static/dynamic-plane problem. Solve it once there. | Avoid solving the hard problem twice. | +| D4 | Broker is **EMQX Cloud** (managed, EU region), replacing self-hosted Mosquitto for the public instance. | Real auth/ACLs/rate limits with zero broker ops; keeps public write ingress off the web host. See §5. | +| D5 | Admission is **self-serve, gated by GitHub identity** (OAuth device flow in the CLI), not a human queue. | Identity substitutes for approval: one active slot per GitHub account, minimum account age, global cap N, post-hoc revocation, and a kill switch to pause admissions. Joining is one command; the maintainer moderates after the fact instead of gating before. See §6. | +| D6 | Credentials are issued by a **small closed-source provisioning service** (server-side, private ops repo). Everything that runs on a contributor's machine stays **open source**. | Client-side secrets are extractable — closed-source client code is obfuscation, not security, and closed-source Rust is impractical for `no_std` targets anyway. Server-side is the only place the EMQX API key and abuse control can actually live. See §6.2. | +| D7 | The join client is a new **`aimdb join` subcommand** in the existing CLI, generic over a provisioning endpoint URL — not a mesh-specific tool, and never a direct client of the EMQX management API. | One tool joins the mesh *and* verifies data arrival over AimX; "how do new edge nodes get admitted" is a product question, not a demo hack. See §7. | +| D8 | Station GPS is **coarsened to 2 decimal places (~1 km)** before publication; precise location is never collected. | The dashboard publishes contributor home locations otherwise. See §9. | +| D9 | Retention: **30 days of live data**; slots silent for 30 days are recycled; per-station credentials are individually revocable. | Policy stated up front on the join page. Storage cost is negligible at weather cadence — these are policy choices, not cost choices. | +| D10 | Slot cap **N = 64** initially (env-configurable, raised by restart). | Large enough that "mesh full" is unlikely at launch; small enough to bound broker connections and dashboard fan-out. | + +## 3. Non-goals + +- Multi-tenant / bring-your-own-cloud hosting. +- Runtime (post-startup) record registration in `aimdb-core` (D3). +- Any closed-source code in the client path — connector, station, or CLI (D6). +- Proprietary application logic in this deployment. The private ops repo holds deployment configuration and the provisioning shim only; the hub and contracts are the OSS crates. +- Topic validation or schema negotiation at admission time — a station that publishes malformed payloads is rejected by the existing contract deserialization at the hub, visibly (§9). + +## 4. Hub: the slot pool + +The closed enums are a property of the *example*, not the engine. `aimdb-core` already provides everything needed: + +- [`StringKey::intern`](../../aimdb-core/src/record_id.rs) exists precisely for runtime-constructed keys. +- `reg.link_from(&topic)` already takes a runtime string (the current hub passes `key.link_address()` through a `String`). +- Inbound runtime topic resolution is covered by design [018](./018-M7-dynamic-mqtt-topics.md). + +The hub change is therefore a loop: + +```rust +let n: u16 = std::env::var("MESH_SLOTS") + .ok().and_then(|v| v.parse().ok()).unwrap_or(64); + +for slot in 0..n { + let key = StringKey::intern(format!("station.{slot}.temperature")); + let topic = format!("mqtt://station/{slot}/temperature"); + builder.configure::(key, |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 100 }); + reg.observe(); + reg.link_from(&topic) + .with_deserializer(|_ctx, data: &[u8]| Temperature::from_bytes(data)) + .finish(); + }); +} +// … same loop for Humidity; DewPoint derived per slot as today. +``` + +Properties: + +- **No recompile per station.** Admission assigns an unused slot; the record and subscription already exist. +- **Slot count is fixed per process start.** Raising N is a hub restart, which is acceptable for a single flagship instance (D1). +- **N is the abuse cap.** The broker ACL (§5) is the enforcement point; N bounds worst-case hub state and dashboard fan-out. +- The dashboard filters to slots with recent data, so empty slots are invisible rather than a wall of grey tiles. + +Topic scheme: `station/{slot}/{temperature|humidity|dew_point}` — flat, slot-scoped, and matching the ACL pattern below. The existing `sensors/{alpha,beta,gamma}/…` topics remain in the local docker-compose demo, which is unchanged by this spec. + +The typed-contract story is unaffected: schema types stay typed end to end; only the *key namespace* moves from a closed enum to bounded strings. The three-enum version remains in the repo as the local demo, where compile-time keys are the point. + +## 5. Broker: EMQX Cloud + +The public instance uses a managed EMQX Cloud deployment (EU region) instead of self-hosted Mosquitto/EMQX: + +- **Isolation:** the public write ingress no longer shares a host with the project website; broker abuse cannot starve the web server. +- **Auth & ACL:** one credential per station, created at admission via the EMQX Cloud API, restricted to publish on `station/{slot}/#` only. The hub uses a separate subscribe-only credential. +- **Limits:** per-client rate and connection limits configured at the broker — the only place abuse control is enforceable (D6). +- **Revocation:** deleting a station's credential evicts it without touching any other station (D9). + +The hub connects **outbound** to the broker over `mqtts://…:8883` — supported today by the Tokio client ([`tokio_client.rs`](../../aimdb-mqtt-connector/src/tokio_client.rs), `use-native-tls`). The serverless tier is TLS-only, which is the right default for internet-facing stations but creates one open problem for the MCU path — see §8.1 before committing to a tier. + +## 6. Admission and provisioning + +### 6.1 Flow + +``` +contributor CLI GitHub provisioning svc EMQX Cloud + │ aimdb join │ │ │ + │── device-code request ─────────▶│ │ │ + │◀─ user code + verify URL ───────│ │ │ + │ (user confirms code at │ │ │ + │ github.com/login/device) │ │ │ + │◀─ access token (poll) ──────────│ │ │ + │ prompts: station name, city │ │ │ + │── POST /v1/join {token, app} ──────────────────────────── ▶│ │ + │ │◀── verify identity ──────│ │ + │ │ check admission rules │── create user + ACL ▶│ + │ │ │ allocate slot │ + │◀────────────── station profile (station.toml) ──────────── │ │ +``` + +- **GitHub device flow** replaces the issue/Action/email plumbing: the CLI prints a short code, the user confirms it at `github.com/login/device`, and the CLI polls for the access token. The OAuth app's client ID is public by design — nothing secret ships in the CLI. No scopes beyond identity are requested. +- **Station metadata** (name, city-level location) is collected by interactive CLI prompts and sent in the join request — no form, no issue template. +- **Admission rules are enforced server-side**, where they cannot be bypassed: one active slot per GitHub account, minimum account age (e.g. 90 days), global cap N (D10), and an admissions kill switch. Mass abuse thus requires mass *aged* GitHub accounts, and N bounds the blast radius regardless. +- **Moderation is post-hoc**, proportionate to the exposure: payloads are schema-validated at the hub, each station can only write its own `station/{slot}/#` topics, and revoking one credential evicts one station (D9). The worst case is garbage numbers in the offender's own slot until revoked. +- The public roster of who's on the mesh is the station pages (§8) — no admission queue needed for transparency. + +### 6.2 Provisioning service + +A thin HTTPS shim, closed source, in the private ops repo. It is the *only* holder of the EMQX Cloud API key. Responsibilities: verify GitHub identities, enforce the admission rules (§6.1), allocate slots, create/delete broker credentials and ACLs, return station profiles, recycle silent slots (D9). State is one small table: `github_account ↔ slot ↔ broker credential`. + +What it is **not**: it is not in the data path (stations speak MQTT to the broker directly), and it is not a client-side component. The boundary is the same one the deployment already draws — ops/config private, contracts/connectors/examples open. + +### 6.3 Provisioning endpoint format (open) + +The request/response format is documented openly so `aimdb join` (§7) is generic, not flagship-specific. Version 1: + +``` +POST /v1/join +Content-Type: application/json + +{ + "auth": { "kind": "github", "token": "gho_…" }, + "app": { "name": "graz-balcony", "location": "Graz" } +} +``` + +`auth.kind` keeps the format deployment-agnostic: the flagship accepts `github`; a private deployment can accept `claim-token` (pre-issued one-time tokens, no public self-serve) with the same envelope. + +``` +200 OK +{ + "profile_version": 1, + "station_id": "slot-17", + "broker": { + "url": "mqtts://xxxx.eu-central-1.emqx.cloud:8883", + "username": "station-17", + "password": "…" + }, + "app": { + "name": "graz-balcony", + "lat": 47.07, + "lon": 15.44 + } +} +``` + +- `app` is an opaque, deployment-specific map: the CLI collects it via prompts (driven by the deployment's docs) and passes it through; the service validates, coarsens the location (D8), and echoes the final values back in the profile. Other AimDB deployments define their own `app` contents. +- Errors: `403` (identity rejected: account too new, admissions paused), `409` (account already holds a slot — body points at the existing profile), `503` (mesh full). Bodies carry a human-readable `message` the CLI prints as-is. +- `profile_version` is versioned and backward-compatible from day one: the CLI ships on the workspace release cadence and must not need a release in lockstep with mesh-side changes. + +## 7. `aimdb join` + +New subcommand in [`tools/aimdb-cli`](../../tools/aimdb-cli) (binary is already named `aimdb`, clap-based): + +``` +aimdb join [--token ] [--out station.toml] +``` + +- Default path: runs the GitHub device flow, prompts for the `app` fields, calls `/v1/join` (§6.3), and writes the profile to `station.toml` (or `--out`). `--token` selects the `claim-token` auth kind instead, for deployments without public self-serve. +- **Generic by construction:** the endpoint URL is an argument, the `app` map is passed through untouched, and no weather-mesh types appear in the CLI. `join` is a product capability — "how a new edge node gets admitted to a deployment" — that debuts on the flagship. +- **Never talks to the EMQX management API.** Doing so would require the admin key in a public binary — the same client-side-secret problem D6 exists to prevent, except leaking the key that can mint credentials for the whole broker. +- One new dependency (`reqwest` with rustls), gated behind a `join` cargo feature, on by default in released binaries. + +The three-command loop this enables is the demo: + +``` +$ aimdb join https://mesh.aimdb.dev + → To authorize, enter code WDJB-MJHT at https://github.com/login/device + ✓ authorized as @alexschnoerch + Station name: graz-balcony + Location (city): Graz + ✓ slot-17 "graz-balcony" — credentials written to station.toml + +$ cargo run -p weather-station -- --config station.toml & + +$ aimdb record list --url tcp://aimdb.dev:7433 + ✓ station.17.temperature 21.4 °C 2s ago +``` + +Join, run, and verify through the hub's own remote-access protocol — one tool, with the last command quietly demonstrating AimX introspection. + +## 8. Station UX + +Three tiers, ordered by commitment: + +- **Tier 0 — look (30 s):** `aimdb.dev/mesh` — live map + charts, and a copy-pasteable MCP snippet ("point your agent at `aimdb.dev/mcp` and ask which station is coldest"). Already running; landing-page work only. +- **Tier 1 — run locally (5 min, no admission):** the existing docker-compose demo, unchanged. Gives impatient visitors something immediate and shows the contracts are ordinary, readable Rust. +- **Tier 2 — join the public mesh:** §6 flow, then §7 commands. Target: **under 10 minutes from `aimdb join` to dot on the map** — with self-serve admission there is no waiting step left in the funnel. On first successful publish, the station prints the direct URL to its live chart and a ready-made MCP prompt for its own data — the shareable moment. + +**Hardware path:** same `station.toml`, consumed at firmware build time and flashed via probe-rs — see §8.2. + +**Station identity pages** (`aimdb.dev/mesh/`): live chart, uptime, last reading. Gives contributors a link to share and a reason to keep stations running — which sustains the propose/admit traffic this exists to generate. + +### 8.1 Open problem: Embassy TLS + +EMQX Cloud serverless is TLS-only, and [`embassy_client.rs`](../../aimdb-mqtt-connector/src/embassy_client.rs) has no TLS support — so the MCU station cannot currently connect to the public broker directly. Options: + +1. **Add TLS to the Embassy MQTT client** (e.g. `embedded-tls`). Real work — the TLS session itself, entropy (the STM32H5 on-chip TRNG, which `weather-station-gamma` already binds via the `rng` interrupt), and a time source for certificate validation (e.g. an SNTP task, since the board has no RTC battery) — but a valuable connector feature in its own right, and it keeps the "MCU speaks directly to a managed cloud broker" claim true. **Recommended.** +2. EMQX Cloud dedicated tier with a plain-TCP listener. Solves it with money and reopens the unencrypted-public-broker question this migration closes. +3. A local bridge/gateway in front of the MCU. Cheapest; quietly breaks the direct-to-cloud story. + +This is the one place the broker choice touches the product. Decide before committing to a broker tier; option 1 can land after launch if the hardware station joins later than the cloud-fed path. + +### 8.2 MCU join path (Embassy) + +Premise: the `aimdb` CLI runs on the **host machine used for flashing** — the same laptop that drives probe-rs, exactly `weather-station-gamma`'s workflow today. The reference board (Nucleo STM32H563ZI) uses Ethernet with DHCP, so there is **zero on-device network configuration** — no Wi-Fi credentials, nothing to type on the device. + +The flow is the cloud-station flow with the flash step swapped in for "run the binary": + +``` +$ aimdb join https://mesh.aimdb.dev --out station.toml # §6 — same self-serve flow, on the host + +$ cd examples/weather-mesh-demo/weather-station-gamma +$ MESH_CONFIG=../station.toml cargo run --release # build → probe-rs flash → defmt logs + +$ aimdb record list --url tcp://aimdb.dev:7433 # §7 — same verification, same tool +``` + +**Mechanism — build-time config embedding.** Today gamma hardcodes `MQTT_BROKER_IP`/`MQTT_BROKER_PORT` as consts with an "update this" comment. Replace that with the station profile consumed at build time: [`build.rs`](../../examples/weather-mesh-demo/weather-station-gamma/build.rs) reads the profile at the path in `MESH_CONFIG`, generates a `station_config.rs` into `OUT_DIR` (`include!`d by `main.rs`), and declares `cargo:rerun-if-env-changed=MESH_CONFIG` / `cargo:rerun-if-changed=`. With `MESH_CONFIG` unset, the generated config falls back to the local docker-compose demo defaults, so the existing tier-1 demo stays zero-setup. + +Why build-time rather than a config flash partition or serial provisioning: + +- Per-station firmware builds are the norm for this audience — the crate already pins its toolchain (`rust-toolchain.toml`) and builds locally; there is no identical-binary distribution problem to solve. +- It reuses the exact workflow embedded Rust developers already know: `cargo run` under the existing probe-rs runner *is* flash + RTT/defmt streaming. The §8 success banner (station chart URL + ready-made MCP prompt) prints over defmt, so the dot-on-the-map moment happens in the same terminal that flashed the board. +- The split build/flash workflow ([`flash.sh`](../../examples/weather-mesh-demo/weather-station-gamma/flash.sh): build in the dev container, flash from the host) keeps working unchanged — `MESH_CONFIG` matters only at build time. +- A config partition (probe-rs writing a config blob at a fixed flash address, firmware reading it at boot) is the natural v2 if prebuilt-image distribution ever becomes desirable; it isn't needed for one flagship's contributors. + +**Trust model:** the slot-scoped credential ends up inside the firmware image in `target/`, which is equivalent to `station.toml` sitting on the same disk — it is the user's own credential, scoped to their own `station/{slot}/#` topics. Built images shouldn't be shared; revocation (D9) covers mistakes. + +**Contingency:** this path reaches the public broker only once §8.1 is resolved — until then, `MESH_CONFIG`-built firmware can target the local demo broker only. + +## 9. Privacy, lifecycle, and failure UX + +- **Location:** only city-level location is collected at admission; published coordinates are rounded to 2 decimals (~1 km) (D8). A short privacy note on the join page states exactly what station metadata is published and for how long. +- **Retention:** 30 days of live data; optionally, downsampled aggregates kept indefinitely (D9). +- **Slot lifecycle:** silent for 30 days → credential revoked, slot recycled, stated up front (D9). +- **Schema errors are a feature:** a malformed payload is rejected by contract deserialization at the hub; the station's own logs say *what contract was violated and what was expected*. The differentiator, demonstrated at the exact moment a user is paying attention. +- **Failure paths:** rejected identity or full mesh → the CLI prints the service's `message` as-is (§6.3); revoked slot → the station says so instead of retrying forever; re-running `aimdb join` with an account that already holds a slot points at the existing profile rather than erroring opaquely. + +## 10. Sequencing + +1. **Hub slot pool** (§4) — OSS, `examples/weather-mesh-demo`, independent of everything else. +2. **EMQX Cloud setup** (§5) — EU region; resolve §8.1 before choosing the tier. +3. **Provisioning service** (§6.2) — private ops repo; includes the GitHub OAuth app registration, identity/admission checks, and slot-recycling job. +4. **`aimdb join`** (§7) + the open endpoint format doc (§6.3) — OSS. +5. **Landing page, station pages, privacy note** (§8, §9). +6. Launch; the hardware-station on-ramp — Embassy TLS (§8.1 option 1) plus the gamma build-time config plumbing (§8.2) — may follow after launch. The `build.rs` embedding in §8.2 is small and independent, so it can also land with step 1. + +Steps 1 and 4 are pure OSS and unblocked today. Step 3 is the only new operational component, and it is a thin shim in front of the EMQX Cloud API. + +## 11. References + +- [016 — RecordKey trait](./016-M6-record-key-trait.md) (`StringKey`, `intern`) +- [018 — Dynamic MQTT topics](./018-M7-dynamic-mqtt-topics.md) (runtime topic resolution) +- [041 — Data contracts as first-class capabilities](./041-data-contracts-integration.md) (contract tiers; `Simulatable` stays dev-only) +- [`examples/weather-mesh-demo`](../../examples/weather-mesh-demo) — the demo this spec extends +- [`tools/aimdb-cli`](../../tools/aimdb-cli), [`aimdb-client`](../../aimdb-client) — AimX client surface, `transport-tcp` diff --git a/docs/design/043-join-endpoint-v1.md b/docs/design/043-join-endpoint-v1.md new file mode 100644 index 00000000..e78a7bb4 --- /dev/null +++ b/docs/design/043-join-endpoint-v1.md @@ -0,0 +1,179 @@ +# 043 — Join endpoint format v1 + +**Status:** 📝 Proposed + +**Scope:** the open wire format of the provisioning endpoint `aimdb join` +speaks (`POST /v1/join`), and the station profile file the CLI writes. No +code in this repo implements the *server* side — the flagship's provisioning +service lives in the private ops repo ([042](./042-public-weather-mesh-flagship.md) +D6) — but the format is public so any AimDB deployment can implement it and +the CLI stays deployment-agnostic. + +**Consumers:** `aimdb join` ([`tools/aimdb-cli`](../../tools/aimdb-cli), +042 §7) is built against this document; `weather-station` +([`examples/weather-mesh-demo`](../../examples/weather-mesh-demo)) consumes +the profile file defined in §4. + +--- + +## 1. Overview + +A *provisioning endpoint* admits a new edge node to a deployment: it verifies +an identity, applies the deployment's admission rules, allocates whatever +resources a member needs (for the weather mesh flagship: a slot number and a +scoped MQTT credential), and returns a **station profile** the node runs +with. The CLI is generic over the endpoint URL: + +``` +aimdb join [--token ] [--out station.toml] +``` + +`aimdb join https://mesh.aimdb.dev` sends `POST https://mesh.aimdb.dev/v1/join`. +The path is fixed; the base URL is the deployment. TLS is required — the +request carries a bearer-equivalent token and the response carries a broker +credential. + +## 2. Request + +``` +POST /v1/join +Content-Type: application/json + +{ + "auth": { "kind": "github", "token": "gho_…" }, + "app": { "name": "graz-balcony", "location": "Graz" } +} +``` + +### 2.1 `auth` — identity envelope + +`auth.kind` keeps the format deployment-agnostic. v1 defines two kinds: + +| `kind` | `token` | Used by | +|---|---|---| +| `github` | A GitHub **user access token**, obtained by the CLI via the OAuth device flow (no scopes — public identity only). The service verifies it against the GitHub API and applies identity-based admission rules (one active slot per account, minimum account age, …). | The public flagship (self-serve). | +| `claim-token` | A pre-issued, one-time opaque token, distributed out of band by the deployment operator. | Private deployments without public self-serve. `aimdb join --token ` selects this kind. | + +Deployments accept the kinds they support and reject others with `403`. + +### 2.2 `app` — deployment-specific metadata + +`app` is an **opaque string-keyed map**: the CLI collects its fields via +interactive prompts (driven by the deployment's docs) and passes them through +untouched — no deployment-specific types exist in the CLI. The service +validates, normalizes, and echoes the final values back in the profile (§3). + +For the weather mesh flagship, v1 `app` fields are: + +| Field | Meaning | +|---|---| +| `name` | Station name — becomes the public identity page (`aimdb.dev/mesh/`). | +| `location` | City-level location. The service geocodes and **coarsens it to 2 decimal places (~1 km)** before it appears anywhere (042 D8); precise location is never collected. | + +## 3. Response + +### 3.1 Success — `200 OK` + +```json +{ + "profile_version": 1, + "station_id": "slot-17", + "broker": { + "url": "mqtts://xxxx.eu-central-1.emqx.cloud:8883", + "username": "station-17", + "password": "…" + }, + "app": { + "name": "graz-balcony", + "lat": 47.07, + "lon": 15.44 + } +} +``` + +| Field | Contract | +|---|---| +| `profile_version` | Format version of this response, `1` for this document. See §5. | +| `station_id` | The node's identity within the deployment — opaque to the CLI. The flagship uses `slot-`, where `` is the assigned slot in the hub's pool (042 §4); the station binary derives its topic/record namespace (`station//…`, `station..…`) from it. | +| `broker.url` | Connector URL of the data-plane broker, without credentials. `mqtts://` for the flagship (broker is TLS-only). | +| `broker.username`, `broker.password` | The per-station credential, scoped server-side to the station's own topics (flagship: publish-only on `station//#`). Individually revocable (042 D9). | +| `app` | The **final** deployment-specific values (post validation/coarsening) — clients must use these, not what they sent. The flagship echoes `name` and the coarsened `lat`/`lon` the station feeds to its weather source. | + +### 3.2 Errors + +Error responses are JSON with a human-readable `message` the CLI prints +**as-is** and exits non-zero — the service owns the failure UX text: + +```json +{ "message": "The mesh is full (64/64 slots) — try again later." } +``` + +| Status | Meaning | Flagship examples | +|---|---|---| +| `403` | Identity rejected. | Account too new; admissions paused (kill switch); unsupported `auth.kind`. | +| `409` | The identity already holds a membership. The `message` points at the existing profile (station name/id) rather than erroring opaquely. | "Your account already runs `graz-balcony` (slot-17). Revoke it first or reuse its profile." | +| `503` | Deployment full or temporarily not admitting. | Mesh at cap N. | + +Any other non-2xx status is unexpected; the CLI reports it with the body if +one is present. + +## 4. The station profile file (`station.toml`) + +`aimdb join` writes the §3.1 response as TOML (mode `0600` — it contains a +credential), default path `station.toml`, `--out` to override: + +```toml +profile_version = 1 +station_id = "slot-17" + +[broker] +url = "mqtts://xxxx.eu-central-1.emqx.cloud:8883" +username = "station-17" +password = "…" + +[app] +name = "graz-balcony" +lat = 47.07 +lon = 15.44 +``` + +The mapping is mechanical (JSON object → TOML table, `app` passed through); +consumers (`weather-station --config station.toml`, gamma's `MESH_CONFIG` +build embedding — 042 §8.2) read this file, never the HTTP response. + +## 5. Versioning and compatibility + +The CLI ships on the workspace release cadence and must not need a release +in lockstep with mesh-side changes (042 §6.3). Therefore: + +- **Unknown fields are ignored**, both in the response (CLI) and in the + profile (station binaries). Services may add fields without a version bump. +- `profile_version` is bumped only for **breaking** changes to the meaning + of existing fields. A client that receives a `profile_version` newer than + it knows writes the profile anyway and warns; a station binary that cannot + interpret a profile says which version it expected. +- The request envelope is versioned by the URL path (`/v1/join`). New auth + kinds may be added to v1 (a service rejects unknown kinds with `403`, which + the CLI reports verbatim). + +## 6. What this format deliberately is not + +- **Not a client of the broker's management API.** The response carries a + ready-made credential; how it was minted (EMQX Cloud API for the flagship) + is invisible to the client, so no admin key can end up in a public binary + (042 D6/D7). +- **Not a schema registry.** Payload validity is enforced by contract + deserialization at the hub (042 §9); join-time carries no topic lists or + schemas. +- **Not a session.** One request, one response, no state on the client + beyond the profile file. Re-joining with an identity that already holds a + membership is a `409`, not an idempotent re-issue — credential re-issue is + a service-side policy decision. + +## 7. References + +- [042 — Public weather mesh flagship](./042-public-weather-mesh-flagship.md) + §6 (admission flow), §7 (`aimdb join`), D5–D8 +- [Implementation plan — 042](../plan/042-public-weather-mesh-flagship.md) + WP3/WP4 +- [GitHub OAuth device flow](https://docs.github.com/en/apps/oauth-apps/building-oauth-apps/authorizing-oauth-apps#device-flow) diff --git a/docs/plan/042-public-weather-mesh-flagship.md b/docs/plan/042-public-weather-mesh-flagship.md new file mode 100644 index 00000000..9f725d6c --- /dev/null +++ b/docs/plan/042-public-weather-mesh-flagship.md @@ -0,0 +1,243 @@ +# Implementation plan — 042 Public weather mesh flagship + +**Design:** [042 — Public weather mesh flagship](../design/042-public-weather-mesh-flagship.md) + +**Status:** 🚧 OSS work packages implemented 2026-07-09 (WP1, WP3's 043 doc, WP4, WP5, WP6 — plus the mqtts:// TLS-transport fix in the Tokio MQTT client the WP1 premise relied on, a read-only AimX TCP endpoint in the hub's mesh mode so the §7 `aimdb record list` loop has a server, and CLI observability metadata so signal gauges round-trip). External WPs (WP2 broker, WP3 provisioning service, WP8 website) and WP7 (Embassy TLS) remain open. + +This plan translates the design's sequencing (§10) into concrete work packages +(WPs). Each WP is marked by where the work happens: + +- **OSS** — this repository; file-level detail below. +- **External** — the private ops repo, the EMQX Cloud console, or the + aimdb.dev website. Tracked here as checklists so the launch dependency + graph is complete, but implemented elsewhere. + +Codebase facts this plan relies on (verified against the current tree): + +- `StringKey::intern` ([`aimdb-core/src/record_id.rs`](../../aimdb-core/src/record_id.rs)) + and `reg.link_from(&topic)` with a runtime string already work — the hub + already passes `key.link_address()` through a `String`. The slot pool + needs **no `aimdb-core` changes**, as the design claims (D2). +- The Tokio MQTT client supports `mqtts://` and URL-embedded credentials + ([`tokio_client.rs`](../../aimdb-mqtt-connector/src/tokio_client.rs), + `use-native-tls`; credentials are parsed from the connector URL), so hub + and cloud stations can authenticate to EMQX Cloud today. The Embassy + client has no TLS — design §8.1, WP7. +- The CLI ([`tools/aimdb-cli`](../../tools/aimdb-cli), binary `aimdb`) is + clap-based, one file per command in `src/commands/`; it has no HTTP + client dependency yet. +- `weather-station-gamma` hardcodes `MQTT_BROKER_IP`/`MQT_BROKER_PORT` + consts in `main.rs`; its `build.rs` is currently just linker args — the + §8.2 config generation slots in there. + +--- + +## WP1 — Hub slot pool (OSS) — design §4 + +**Crate:** `examples/weather-mesh-demo/weather-hub` + +The hub gains an env-gated *mesh mode* alongside the existing three-enum +demo mode. The local docker-compose demo must remain byte-for-byte +unchanged in behaviour (design §4: "the existing `sensors/{alpha,beta,gamma}/…` +topics remain in the local docker-compose demo"). + +Changes in [`weather-hub/src/main.rs`](../../examples/weather-mesh-demo/weather-hub/src/main.rs): + +1. **Mode switch:** if `MESH_SLOTS` is set, configure the slot pool; if + unset, run the current `TempKey`/`HumidityKey` enum configuration + untouched. +2. **Slot pool loop:** for `slot in 0..n` (`n` = `MESH_SLOTS`, default 64, + D10): + - `StringKey::intern(format!("station.{slot}.temperature"))`, buffer + `SpmcRing { capacity: 100 }`, `.observe()`, and + `.link_from("mqtt://station/{slot}/temperature")` with the existing + `Temperature::from_bytes` deserializer — same for `Humidity`. + - `DewPoint` derived per slot via `transform_join` on that slot's + Temperature + Humidity, reusing the derivation already written in + [`weather-station-alpha/src/main.rs`](../../examples/weather-mesh-demo/weather-station-alpha/src/main.rs) + (the `t.celsius - (100.0 - h.percent) / 5.0` join). Deriving at the + hub keeps the dashboard supplied even if a station publishes only + temperature/humidity. + - Drop the per-value `.log()` console lines in mesh mode — 64 slots of + per-message logging is noise; `.observe()` metrics remain. +3. **Broker URL:** accept a full connector URL via `MQTT_URL` + (e.g. `mqtts://hub-sub:…@xxxx.eu-central-1.emqx.cloud:8883`), falling + back to the current `MQTT_BROKER` host-only var for the local demo. The + hub uses the subscribe-only EMQX credential (design §5). Enable the + `use-native-tls` feature of `aimdb-mqtt-connector` in + `weather-hub/Cargo.toml`. + +**Verification:** local mosquitto + `MESH_SLOTS=4 cargo run -p weather-hub`; +`mosquitto_pub -t station/2/temperature -m `; confirm +`station.2.temperature` updates via `aimdb record list` and that a malformed +payload is rejected with a contract error (design §9). Unset `MESH_SLOTS` +and confirm the docker-compose demo behaves exactly as before. + +## WP2 — EMQX Cloud setup (external) — design §5 + +Checklist, in the ops repo / EMQX console: + +- [ ] Decide the tier — **blocked on the §8.1 decision** (serverless is + TLS-only; see WP7). Per design, option 1 (Embassy TLS) is + recommended, which permits the serverless tier. +- [ ] Deployment in EU region. +- [ ] Hub credential: subscribe-only on `station/#`. +- [ ] Station ACL template: publish-only on `station/{slot}/#`. +- [ ] Per-client rate and connection limits. +- [ ] API key for the provisioning service (never leaves the service, D6). + +## WP3 — Provisioning service (external) + open endpoint doc (OSS) — design §6 + +The service itself is a thin closed-source HTTPS shim in the private ops +repo: GitHub identity verification, admission rules (one slot per account, +minimum account age, cap N, kill switch), slot allocation, EMQX credential +create/delete, 30-day silent-slot recycling, one +`github_account ↔ slot ↔ credential` table. Includes registering the GitHub +OAuth app (device flow; client ID is public). + +**OSS artifact in this WP:** the open endpoint format doc — +`docs/design/043-join-endpoint-v1.md` (next free number) — specifying +`POST /v1/join` v1 exactly as design §6.3 defines it: the +`auth.kind = github | claim-token` envelope, the opaque `app` map, the +`profile_version`-ed response, and the `403/409/503` error bodies with +human-readable `message`. This doc is the contract `aimdb join` (WP4) is +built against, so it lands **before or with** WP4. + +## WP4 — `aimdb join` (OSS) — design §7 + +**Crate:** `tools/aimdb-cli` + +- New `src/commands/join.rs`, wired as `Command::Join(JoinCommand)` in + `src/main.rs` following the existing per-command pattern: + + ``` + aimdb join [--token ] [--out station.toml] + ``` + +- Default auth path: GitHub device flow — POST to + `github.com/login/device/code` with the compiled-in public client ID, + print the user code + verification URL, poll + `github.com/login/oauth/access_token` respecting `interval`/`slow_down`. + `--token` switches the envelope to `claim-token` and skips the flow. +- Prompt on stdin for the deployment's `app` fields (v1: station name, + city) and pass them through untouched — no weather-mesh types in the CLI. +- `POST /v1/join` per the 043 doc; on `403/409/503` print the body's + `message` as-is and exit non-zero; on success write the profile as TOML + to `--out` (default `station.toml`, mode 0600) and print the assigned + slot + station name. +- Dependencies, gated behind a `join` cargo feature (default **on**): + `reqwest` (rustls, no default features) and `toml`. Never a client of + the EMQX management API (D6/D7). + +**Verification:** unit tests for profile serialization and error-body +handling against a mock server (`tools/aimdb-cli` already has +`tokio-test`/`tempfile` dev-deps); manual device-flow run against the real +GitHub endpoint with the flagship OAuth app. + +## WP5 — Generic cloud station (OSS) — design §7/§8 + +**New crate:** `examples/weather-mesh-demo/weather-station` (workspace +member), the binary in the design's demo loop: +`cargo run -p weather-station -- --config station.toml`. + +- Parses `station.toml` (`serde` + `toml`): broker URL/credentials, + `station_id` (slot), `app.name`, coarsened `app.lat`/`app.lon`. +- Configures `Temperature`/`Humidity` with `StringKey::intern` + (`station.{slot}.temperature`, …) and `.link_to("mqtt://station/{slot}/temperature")`, + plus the derived `DewPoint` — a config-driven port of alpha's setup, + reusing alpha's Open-Meteo client at the profile's lat/lon. +- First-successful-publish banner: direct URL to the station's live chart + (`aimdb.dev/mesh/`) and a ready-made MCP prompt — the shareable + moment (design §8 tier 2). +- Revoked-credential failure UX: on MQTT auth failure, say the slot was + likely revoked and point at the join page instead of retrying forever + (design §9). +- Alpha, beta, and the docker-compose demo remain untouched. + +**Verification:** hand-written `station.toml` against local mosquitto + +WP1 hub in mesh mode; then the full three-command loop of design §7 once +WP2/WP3 are live. + +## WP6 — Gamma build-time config (OSS) — design §8.2 + +**Crate:** `examples/weather-mesh-demo/weather-station-gamma` + +- [`build.rs`](../../examples/weather-mesh-demo/weather-station-gamma/build.rs): + read the profile at `$MESH_CONFIG` (same `station.toml`), generate + `station_config.rs` into `OUT_DIR` (broker host/port/credentials, slot, + name); with `MESH_CONFIG` unset, emit the current local-demo defaults + (`192.168.1.3:1883`, `sensors/gamma/…` topics) so tier 1 stays + zero-setup. Declare `cargo:rerun-if-env-changed=MESH_CONFIG` and + `cargo:rerun-if-changed=`. +- `main.rs`: replace the `MQTT_BROKER_IP`/`MQTT_BROKER_PORT` consts with + `include!(concat!(env!("OUT_DIR"), "/station_config.rs"))`; success + banner over defmt. +- Small and independent — can land with WP1. **Contingency:** until WP7, + `MESH_CONFIG`-built firmware can only target the local demo broker + (no TLS on Embassy). + +**Verification:** build with and without `MESH_CONFIG` and diff the +generated file; flash via the existing `cargo run` probe-rs runner and via +`flash.sh` (split build/flash must keep working). + +## WP7 — Embassy TLS (OSS, post-launch acceptable) — design §8.1 option 1 + +**Crate:** `aimdb-mqtt-connector` (`embassy_client.rs`) + +Real connector work, to be scoped in its own design doc before +implementation: `embedded-tls` session over the Embassy TCP socket, entropy +from the STM32H5 TRNG (already bound via the `rng` interrupt in gamma), and +an SNTP task as the time source for certificate validation (no RTC +battery). Keeps the "MCU speaks directly to a managed cloud broker" claim +true. Blocks only the *hardware* on-ramp to the public mesh; the cloud +path (WP1–WP5) launches without it. + +## WP8 — Landing, station pages, privacy note (external) — design §8/§9 + +Website repo checklist: + +- [ ] `aimdb.dev/mesh` landing: live map + charts, MCP snippet, join + instructions (tier 0/2 funnel). +- [ ] Station identity pages `aimdb.dev/mesh/`: live chart, uptime, + last reading. +- [ ] Dashboard filters to slots with recent data (empty slots invisible, + design §4). +- [ ] Privacy note: what metadata is published, 2-decimal coordinate + rounding (D8), 30-day retention and slot recycling (D9). + +--- + +## Sequencing and dependencies + +``` +WP1 hub slot pool ──────────────┐ +WP3(OSS) 043 endpoint doc ─► WP4 aimdb join ─┐ +WP5 weather-station ────────────┤ ├─► launch (cloud path) +WP6 gamma build config ─────────┘ │ +WP2 EMQX setup ◄─ tier decision (§8.1) ───────┤ +WP3(ext) provisioning service ────────────────┘ +WP7 Embassy TLS ─► hardware on-ramp (post-launch OK) +WP8 website ──────► launch +``` + +- **Unblocked today (pure OSS):** WP1, WP3's 043 doc, WP4, WP5, WP6. +- **WP2** waits only on the §8.1 tier decision (recommended: serverless + + WP7 later). +- **WP3 (service)** is the only new operational component and gates the + live end-to-end flow. +- **Launch =** WP1 + WP2 + WP3 + WP4 + WP5 + WP8. **WP6/WP7** complete the + hardware story and may follow. + +## End-to-end verification + +1. **Local drill (no cloud dependencies):** mosquitto + WP1 hub with + `MESH_SLOTS=8` + WP5 station with a hand-written `station.toml` → + `aimdb record list --url tcp://localhost:7433` shows the slot updating; + a deliberately malformed publish surfaces a contract error at the hub. +2. **Hosted smoke test:** the design §7 three-command loop against the real + stack — `aimdb join https://mesh.aimdb.dev`, run the station, verify via + `aimdb record list --url tcp://aimdb.dev:7433` — measured against the + design's target of **under 10 minutes** from join to dot on the map. +3. **Regression:** `docker compose up` in `examples/weather-mesh-demo` + behaves exactly as before (tier 1 unchanged), and the gamma local-demo + build with `MESH_CONFIG` unset is byte-identical in behaviour. diff --git a/examples/weather-mesh-demo/README.md b/examples/weather-mesh-demo/README.md index e7942031..fee771a1 100644 --- a/examples/weather-mesh-demo/README.md +++ b/examples/weather-mesh-demo/README.md @@ -27,6 +27,7 @@ A distributed weather monitoring mesh demonstrating AimDB's multi-tier architect | `weather-station-alpha` | Tokio (Linux Edge) | Real weather data (Open-Meteo API) | | `weather-station-beta` | Tokio (Linux Edge) | Synthetic sensor data | | `weather-station-gamma` | Embassy (STM32H563ZITx) | Portable/remote MCU sensor (hardware) | +| `weather-station` | Tokio (anywhere) | Public-mesh station, driven by a `station.toml` profile | | `weather-mesh-common` | no_std compatible | Shared contracts and configuration | | `mqtt` | Eclipse Mosquitto | Message bus for all nodes | @@ -110,6 +111,33 @@ cargo build --release probe-rs run --chip STM32H563ZITx target/thumbv8m.main-none-eabihf/release/weather-station-gamma ``` +## Public mesh mode (design 042) + +The same crates also run the hosted public mesh anyone can join +([design 042](../../docs/design/042-public-weather-mesh-flagship.md)). The +local demo above is unchanged; mesh mode is opt-in per environment variable: + +- **Hub** — `MESH_SLOTS=64 cargo run -p weather-hub` replaces the three-enum + configuration with a bounded pool of string-keyed slot records + (`station.{slot}.temperature/humidity/dew_point` from + `station/{slot}/…` topics; dew point derived at the hub). `MQTT_URL` takes + a full connector URL (e.g. `mqtts://hub-sub:…@…:8883` for EMQX Cloud), and + a read-only AimX endpoint is served at `AIMX_BIND` + (default `127.0.0.1:7433`) for `aimdb record list/get/watch`. +- **Station** — the three-command join loop + ([format doc 043](../../docs/design/043-join-endpoint-v1.md)): + + ```bash + aimdb join https://mesh.aimdb.dev # writes station.toml + cargo run -p weather-station -- --config station.toml + aimdb record list --connect tcp://aimdb.dev:7433 + ``` + +- **MCU station** — same profile, consumed at build time: + `MESH_CONFIG=../station.toml cargo run --release` inside + `weather-station-gamma` (plain-TCP brokers only until the Embassy MQTT + client speaks TLS — design 042 §8.1). + ## Inspecting with VS Code Copilot The weather mesh exposes an AimX API socket that Copilot can read through MCP (Model Context Protocol). Simply ask Copilot in natural language: diff --git a/examples/weather-mesh-demo/weather-hub/Cargo.toml b/examples/weather-mesh-demo/weather-hub/Cargo.toml index 73bb9c1a..ec44dc0c 100644 --- a/examples/weather-mesh-demo/weather-hub/Cargo.toml +++ b/examples/weather-mesh-demo/weather-hub/Cargo.toml @@ -29,6 +29,11 @@ aimdb-data-contracts = { path = "../../../aimdb-data-contracts", features = [ aimdb-mqtt-connector = { path = "../../../aimdb-mqtt-connector", features = [ "tokio-runtime", ] } +# AimX server for mesh mode: `aimdb record list --url tcp://…:7433` +# (design 042 §7). Unused in the local demo path. +aimdb-tcp-connector = { path = "../../../aimdb-tcp-connector", features = [ + "tokio-runtime", +] } tokio = { version = "1", features = ["full"] } tracing = "0.1" diff --git a/examples/weather-mesh-demo/weather-hub/src/main.rs b/examples/weather-mesh-demo/weather-hub/src/main.rs index 94aefd8f..38a7f77e 100644 --- a/examples/weather-mesh-demo/weather-hub/src/main.rs +++ b/examples/weather-mesh-demo/weather-hub/src/main.rs @@ -1,25 +1,33 @@ -use aimdb_core::{buffer::BufferCfg, AimDbBuilder, RecordKey}; +use aimdb_core::{buffer::BufferCfg, AimDbBuilder, RecordKey, StringKey}; use aimdb_data_contracts::{Linkable, ObservableRegistrarExt}; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; -use weather_mesh_common::{Humidity, HumidityKey, TempKey, Temperature}; +use weather_mesh_common::{DewPoint, Humidity, HumidityKey, TempKey, Temperature}; #[tokio::main] async fn main() -> aimdb_core::DbResult<()> { // Initialize logging + // `aimdb` is the runtime-context log target (ctx.log() via the adapter) — + // without it in the fallback filter, contract-violation reports from the + // mesh deserializers are invisible. The docker-compose demo always sets + // RUST_LOG explicitly, so this only affects bare `cargo run`. tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| "weather_hub=info,aimdb_core=info".into()), + .unwrap_or_else(|_| "weather_hub=info,aimdb_core=info,aimdb=info".into()), ) .init(); tracing::info!("🚀 Starting Weather Hub"); - // Configuration from environment - let mqtt_broker = std::env::var("MQTT_BROKER").unwrap_or_else(|_| "localhost".to_string()); - let mqtt_url = format!("mqtt://{}", mqtt_broker); + // Broker: MQTT_URL takes a full connector URL (e.g. the public mesh broker, + // mqtts://hub-sub:…@xxxx.eu-central-1.emqx.cloud:8883); otherwise fall back + // to the local demo's host-only MQTT_BROKER variable. + let mqtt_url = std::env::var("MQTT_URL").unwrap_or_else(|_| { + let mqtt_broker = std::env::var("MQTT_BROKER").unwrap_or_else(|_| "localhost".to_string()); + format!("mqtt://{}", mqtt_broker) + }); - tracing::info!("📡 Connecting to MQTT broker: {}", mqtt_url); + tracing::info!("📡 Connecting to MQTT broker: {}", redact_url(&mqtt_url)); let runtime = std::sync::Arc::new(TokioAdapter::new()?); @@ -27,6 +35,35 @@ async fn main() -> aimdb_core::DbResult<()> { aimdb_mqtt_connector::MqttConnector::new(&mqtt_url).with_client_id("weather-hub"), ); + // MESH_SLOTS switches the hub into public-mesh mode (design 042 §4): a + // bounded pool of string-keyed slot records instead of the closed + // three-station enums. Unset = the local docker-compose demo, unchanged. + match std::env::var("MESH_SLOTS").ok() { + Some(v) => { + let slots: u16 = v.parse().unwrap_or(64); + tracing::info!("🕸️ Mesh mode: {} slots", slots); + + // AimX introspection endpoint — the `aimdb record list --url + // tcp://…:7433` half of the demo loop (design 042 §7). Read-only + // (the default policy); loopback unless AIMX_BIND says otherwise + // (the hosted deployment fronts it as aimdb.dev:7433). + let aimx_bind = + std::env::var("AIMX_BIND").unwrap_or_else(|_| "127.0.0.1:7433".to_string()); + tracing::info!("🔌 AimX endpoint: tcp://{}", aimx_bind); + builder = builder.with_connector(aimdb_tcp_connector::TcpServer::new(&aimx_bind)); + + configure_mesh(&mut builder, slots); + } + None => configure_demo(&mut builder), + } + + tracing::info!("✅ Weather Hub configured, starting..."); + + builder.run().await +} + +/// The local demo configuration: the three closed-enum stations. +fn configure_demo(builder: &mut AimDbBuilder) { // Configure Temperature records for all nodes with tap functions for key in [TempKey::Alpha, TempKey::Beta, TempKey::Gamma] { let topic = key.link_address().unwrap().to_string(); @@ -75,8 +112,100 @@ async fn main() -> aimdb_core::DbResult<()> { .finish(); }); } +} - tracing::info!("✅ Weather Hub configured, starting..."); +/// The public-mesh slot pool (design 042 §4). +/// +/// Every slot's records exist from startup; admission just assigns an unused +/// slot number, so joining never needs a hub recompile. A malformed payload is +/// rejected by `from_bytes` (contract deserialization) — that error surfacing +/// at the hub is the admission-time schema check. +/// +/// No per-value `.log()` here: 64 slots of per-message console lines is noise. +/// `.observe()` keeps the signal gauges on `record list`/`record get`. +fn configure_mesh(builder: &mut AimDbBuilder, slots: u16) { + for slot in 0..slots { + let temp_key = StringKey::intern(format!("station.{slot}.temperature")); + let temp_topic = format!("mqtt://station/{slot}/temperature"); + builder.configure::(temp_key, |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 100 }); + reg.observe(); + // JSON codec for AimX record.get/subscribe — the dashboard and + // `aimdb record get` read the mesh through this. + reg.with_remote_access(); + let key_name = temp_key.as_str(); + reg.link_from(&temp_topic) + .with_deserializer(move |ctx, data: &[u8]| { + Temperature::from_bytes(data).map_err(|e| { + // Design 042 §9: schema errors are a feature — a + // malformed payload is visibly rejected at the hub. + ctx.log() + .error(&format!("{key_name}: rejected payload: {e}")); + e + }) + }) + .finish(); + }); - builder.run().await + let humidity_key = StringKey::intern(format!("station.{slot}.humidity")); + let humidity_topic = format!("mqtt://station/{slot}/humidity"); + builder.configure::(humidity_key, |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 100 }); + reg.observe(); + reg.with_remote_access(); + let key_name = humidity_key.as_str(); + reg.link_from(&humidity_topic) + .with_deserializer(move |ctx, data: &[u8]| { + Humidity::from_bytes(data).map_err(|e| { + ctx.log() + .error(&format!("{key_name}: rejected payload: {e}")); + e + }) + }) + .finish(); + }); + + // Derived at the hub so the dashboard stays supplied even when a + // station publishes only temperature/humidity. + let dew_point_key = StringKey::intern(format!("station.{slot}.dew_point")); + builder.configure::(dew_point_key, |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 100 }); + reg.observe(); + reg.with_remote_access(); + reg.transform_join(|b| { + b.input::(temp_key) + .input::(humidity_key) + .on_triggers(|mut rx, producer| async move { + let mut last_temp: Option = None; + let mut last_hum: Option = None; + while let Ok(trigger) = rx.recv().await { + match trigger.index() { + 0 => last_temp = trigger.as_input::().cloned(), + 1 => last_hum = trigger.as_input::().cloned(), + _ => {} + } + if let (Some(t), Some(h)) = (&last_temp, &last_hum) { + // Magnus approximation: T_dp ≈ T - (100 - RH) / 5 + let dew_point = t.celsius - (100.0 - h.percent) / 5.0; + let timestamp = t.timestamp.max(h.timestamp); + producer.produce(DewPoint { + celsius: dew_point, + timestamp, + }); + } + } + }) + }); + }); + } +} + +/// Strip URL-embedded credentials before logging. +fn redact_url(url: &str) -> String { + if let Some((scheme, rest)) = url.split_once("://") { + if let Some((_creds, host)) = rest.rsplit_once('@') { + return format!("{scheme}://…@{host}"); + } + } + url.to_string() } diff --git a/examples/weather-mesh-demo/weather-station-gamma/Cargo.toml b/examples/weather-mesh-demo/weather-station-gamma/Cargo.toml index b5341446..c73a46fd 100644 --- a/examples/weather-mesh-demo/weather-station-gamma/Cargo.toml +++ b/examples/weather-mesh-demo/weather-station-gamma/Cargo.toml @@ -94,6 +94,10 @@ embedded-alloc = { version = "0.6", features = ["llff"] } # network-stack seed uses the STM32 HAL RNG (embassy_stm32), not this crate. rand = { version = "0.10.1", default-features = false, optional = true } +[build-dependencies] +# Parses the $MESH_CONFIG station profile (station.toml, design 042 §8.2) to +# generate station_config.rs. Host-side only — nothing lands in the firmware. +toml = "0.8" [package.metadata.embassy] build = [ diff --git a/examples/weather-mesh-demo/weather-station-gamma/build.rs b/examples/weather-mesh-demo/weather-station-gamma/build.rs index 8cd32d7e..0f46d184 100644 --- a/examples/weather-mesh-demo/weather-station-gamma/build.rs +++ b/examples/weather-mesh-demo/weather-station-gamma/build.rs @@ -1,5 +1,152 @@ +//! Linker args + build-time station configuration (design 042 §8.2). +//! +//! With `MESH_CONFIG` pointing at a station profile (`station.toml`, written +//! by `aimdb join` — design 043 §4), the broker address, slot topics, and +//! station name are generated into `OUT_DIR/station_config.rs` and baked into +//! the firmware: `MESH_CONFIG=../station.toml cargo run --release` is +//! build → flash → live. With `MESH_CONFIG` unset, the generated file carries +//! the local docker-compose demo defaults, so the tier-1 demo stays +//! zero-setup. + +use std::env; +use std::fs; +use std::path::PathBuf; + fn main() { println!("cargo:rustc-link-arg-bins=--nmagic"); println!("cargo:rustc-link-arg-bins=-Tlink.x"); println!("cargo:rustc-link-arg-bins=-Tdefmt.x"); + + println!("cargo:rerun-if-env-changed=MESH_CONFIG"); + + let config = match env::var("MESH_CONFIG") { + Ok(path) => { + println!("cargo:rerun-if-changed={path}"); + mesh_station_config(&path) + } + Err(_) => local_demo_config(), + }; + + let out_dir = PathBuf::from(env::var("OUT_DIR").unwrap()); + fs::write(out_dir.join("station_config.rs"), config).unwrap(); +} + +/// Local docker-compose demo defaults (tier 1 — no profile, no mesh). +/// The topics mirror the `TempKey::Gamma`/…`::Gamma` link addresses in +/// `weather-mesh-common`. +fn local_demo_config() -> String { + station_config("192.168.1.3", 1883, None, None, "gamma", "sensors/gamma") +} + +/// Configuration from the station profile at `path` (design 043 §4). +fn mesh_station_config(path: &str) -> String { + let text = fs::read_to_string(path) + .unwrap_or_else(|e| panic!("MESH_CONFIG: cannot read '{path}': {e}")); + let profile: toml::Value = toml::from_str(&text) + .unwrap_or_else(|e| panic!("MESH_CONFIG: '{path}' is not valid TOML: {e}")); + + let version = profile + .get("profile_version") + .and_then(|v| v.as_integer()) + .unwrap_or_else(|| panic!("MESH_CONFIG: '{path}' has no profile_version")); + assert!( + version == 1, + "MESH_CONFIG: unsupported profile_version {version} (this firmware understands 1)" + ); + + let str_field = |table: &str, key: &str| -> &str { + profile + .get(table) + .and_then(|t| t.get(key)) + .and_then(|v| v.as_str()) + .unwrap_or_else(|| panic!("MESH_CONFIG: '{path}' has no [{table}] {key}")) + }; + + let broker_url = str_field("broker", "url"); + // The Embassy MQTT client has no TLS yet (design 042 §8.1) — a profile for + // the public broker (mqtts://) cannot work; only a plain-TCP broker (e.g. + // the local demo one) is reachable until that lands. + let rest = broker_url.strip_prefix("mqtt://").unwrap_or_else(|| { + panic!( + "MESH_CONFIG: broker URL '{broker_url}' is not mqtt:// — the Embassy \ + MQTT client has no TLS support yet (design 042 §8.1), so this \ + firmware can only target a plain-TCP broker" + ) + }); + let authority = rest.split('/').next().unwrap_or(rest); + let (host, port) = match authority.rsplit_once(':') { + Some((host, port)) => ( + host, + port.parse::() + .unwrap_or_else(|_| panic!("MESH_CONFIG: invalid broker port in '{broker_url}'")), + ), + None => (authority, 1883), + }; + + let station_id = profile + .get("station_id") + .and_then(|v| v.as_str()) + .unwrap_or_else(|| panic!("MESH_CONFIG: '{path}' has no station_id")); + let slot: u16 = station_id + .strip_prefix("slot-") + .and_then(|n| n.parse().ok()) + .unwrap_or_else(|| { + panic!("MESH_CONFIG: station_id '{station_id}' is not of the form slot-") + }); + + let name = str_field("app", "name"); + let credentials = Some(( + str_field("broker", "username"), + str_field("broker", "password"), + )); + + station_config( + host, + port, + credentials, + Some(slot), + name, + &format!("station/{slot}"), + ) +} + +/// Render the generated `station_config.rs`. String values go through `{:?}` +/// so arbitrary profile content stays a valid Rust literal. +fn station_config( + host: &str, + port: u16, + credentials: Option<(&str, &str)>, + slot: Option, + name: &str, + topic_prefix: &str, +) -> String { + let credential_const = |value: Option<&str>| match value { + Some(v) => format!("Some({v:?})"), + None => "None".to_string(), + }; + let username = credential_const(credentials.map(|(u, _)| u)); + let password = credential_const(credentials.map(|(_, p)| p)); + let slot = match slot { + Some(n) => format!("Some({n})"), + None => "None".to_string(), + }; + + format!( + "// Generated by build.rs — station configuration (design 042 §8.2).\n\ + // Source: $MESH_CONFIG station profile, or local demo defaults. Do not edit.\n\ + pub const MQTT_BROKER_HOST: &str = {host:?};\n\ + pub const MQTT_BROKER_PORT: u16 = {port};\n\ + /// Broker credential from the profile. Unused until the Embassy MQTT\n\ + /// client supports authentication + TLS (design 042 §8.1, WP7).\n\ + #[allow(dead_code)]\n\ + pub const MQTT_USERNAME: Option<&str> = {username};\n\ + #[allow(dead_code)]\n\ + pub const MQTT_PASSWORD: Option<&str> = {password};\n\ + /// Assigned mesh slot; `None` in the local demo.\n\ + pub const STATION_SLOT: Option = {slot};\n\ + pub const STATION_NAME: &str = {name:?};\n\ + pub const TEMPERATURE_TOPIC: &str = \"mqtt://{topic_prefix}/temperature\";\n\ + pub const HUMIDITY_TOPIC: &str = \"mqtt://{topic_prefix}/humidity\";\n\ + pub const DEW_POINT_TOPIC: &str = \"mqtt://{topic_prefix}/dew_point\";\n" + ) } diff --git a/examples/weather-mesh-demo/weather-station-gamma/src/main.rs b/examples/weather-mesh-demo/weather-station-gamma/src/main.rs index 6c0eeb4f..4af85344 100644 --- a/examples/weather-mesh-demo/weather-station-gamma/src/main.rs +++ b/examples/weather-mesh-demo/weather-station-gamma/src/main.rs @@ -13,18 +13,26 @@ //! //! ## Running //! -//! 1. Start MQTT broker on your network -//! 2. Update MQTT_BROKER_IP constant below -//! 3. Flash to target: +//! The broker address, topics, and station identity are generated at build +//! time by `build.rs` (design 042 §8.2): +//! +//! - **Local demo (default):** no setup — the generated config carries the +//! docker-compose demo defaults. Point it at your broker by editing the +//! defaults in `build.rs`, then: //! ```bash //! cd examples/weather-mesh-demo/weather-station-gamma -//! cargo build --release -//! cargo flash --release +//! cargo run --release # build → probe-rs flash → defmt logs +//! ``` +//! - **Public mesh:** feed it the profile written by `aimdb join`: +//! ```bash +//! MESH_CONFIG=../station.toml cargo run --release //! ``` +//! (Reaches the public broker only once the Embassy client speaks TLS — +//! design 042 §8.1; until then a profile must name a plain-TCP broker.) extern crate alloc; -use aimdb_core::{AimDbBuilder, RecordKey}; +use aimdb_core::AimDbBuilder; #[cfg(feature = "sim")] use aimdb_data_contracts::{RandomWalkParams, SimProfile, SimulatableRegistrarExt}; use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyBufferType, EmbassyRecordRegistrarExtCustom}; @@ -48,9 +56,9 @@ use {defmt_rtt as _, panic_probe as _}; #[global_allocator] static ALLOCATOR: embedded_alloc::LlffHeap = embedded_alloc::LlffHeap::empty(); -// MQTT broker address - update this to match your network -const MQTT_BROKER_IP: &str = "192.168.1.3"; -const MQTT_BROKER_PORT: u16 = 1883; +// Station configuration generated by build.rs: the $MESH_CONFIG station +// profile, or the local demo defaults when unset (design 042 §8.2). +include!(concat!(env!("OUT_DIR"), "/station_config.rs")); bind_interrupts!(struct Irqs { ETH => eth::InterruptHandler; @@ -248,14 +256,17 @@ async fn main(spawner: Spawner) { // Build MQTT broker URL use alloc::format; - let broker_url = format!("mqtt://{}:{}", MQTT_BROKER_IP, MQTT_BROKER_PORT); + let broker_url = format!("mqtt://{}:{}", MQTT_BROKER_HOST, MQTT_BROKER_PORT); let mut builder = AimDbBuilder::new().runtime(runtime.clone()).with_connector( MqttConnectorBuilder::new(&broker_url, stack).with_client_id("weather-station-gamma"), ); - // Configure temperature record - let temp_topic = TempKey::Gamma.link_address().unwrap(); + // Configure temperature record. The record keys stay the compile-time + // enums (they name the records in *this* db); only the MQTT topics come + // from the generated config, so a mesh profile re-points the publishes to + // its assigned slot (station//…). + let temp_topic = TEMPERATURE_TOPIC; builder.configure::(TempKey::Gamma, |reg| { reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing); @@ -286,7 +297,7 @@ async fn main(spawner: Spawner) { }); // Configure humidity record - let humidity_topic = HumidityKey::Gamma.link_address().unwrap(); + let humidity_topic = HUMIDITY_TOPIC; builder.configure::(HumidityKey::Gamma, |reg| { reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing); @@ -319,7 +330,7 @@ async fn main(spawner: Spawner) { // // Showcases the transform_join task model: the closure owns its state and // can hold borrows across .await without Box::pin or manual cloning. - let dew_point_topic = DewPointKey::Gamma.link_address().unwrap(); + let dew_point_topic = DEW_POINT_TOPIC; builder.configure::(DewPointKey::Gamma, |reg| { reg.buffer_sized::<8, 1>(EmbassyBufferType::SpmcRing) .transform_join(|b| { @@ -378,7 +389,7 @@ async fn main(spawner: Spawner) { info!(" Temperature: {}", temp_topic); info!(" Humidity: {}", humidity_topic); info!(" DewPoint: {}", dew_point_topic); - info!(" Broker: {}:{}", MQTT_BROKER_IP, MQTT_BROKER_PORT); + info!(" Broker: {}:{}", MQTT_BROKER_HOST, MQTT_BROKER_PORT); info!(""); static DB_CELL: StaticCell = StaticCell::new(); @@ -386,7 +397,24 @@ async fn main(spawner: Spawner) { let _db = DB_CELL.init(db); info!("✅ Database running"); - info!("🎯 Weather Station Gamma ready!"); + // The dot-on-the-map moment, in the same terminal that flashed the board + // (design 042 §8.2): mesh builds print the station's live chart URL and a + // ready-made MCP prompt over defmt. + match STATION_SLOT { + Some(slot) => { + info!( + "🎯 Station \"{}\" (slot {}) is publishing to the mesh!", + STATION_NAME, slot + ); + info!("🗺️ Live chart: https://aimdb.dev/mesh/{}", STATION_NAME); + info!("🤖 Point an MCP agent at https://aimdb.dev/mcp and try:"); + info!( + " \"What is the latest temperature at station {}?\"", + STATION_NAME + ); + } + None => info!("🎯 Weather Station Gamma ready!"), + } // Drive the AimDB runner and LED blink concurrently. embassy_futures::join::join(db_runner.run(), async { diff --git a/examples/weather-mesh-demo/weather-station/Cargo.toml b/examples/weather-mesh-demo/weather-station/Cargo.toml new file mode 100644 index 00000000..567fbac5 --- /dev/null +++ b/examples/weather-mesh-demo/weather-station/Cargo.toml @@ -0,0 +1,42 @@ +[package] +name = "weather-station" +version = "0.1.0" +edition = "2021" +description = "Generic cloud weather station - joins the public mesh from a station.toml profile" +license = "MIT OR Apache-2.0" +publish = false + +[[bin]] +name = "weather-station" +path = "src/main.rs" + +[dependencies] +weather-mesh-common = { path = "../weather-mesh-common", features = [ + "linkable", + "migratable", +] } +aimdb-core = { path = "../../../aimdb-core" } +aimdb-tokio-adapter = { path = "../../../aimdb-tokio-adapter" } +aimdb-data-contracts = { path = "../../../aimdb-data-contracts", features = [ + "linkable", + "migratable", +] } +aimdb-mqtt-connector = { path = "../../../aimdb-mqtt-connector", features = [ + "tokio-runtime", +] } + +tokio = { version = "1", features = ["full"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +reqwest = { version = "0.12", features = ["json"] } +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +chrono = "0.4" +clap = { version = "4", features = ["derive"] } +toml = "0.8" +# Pre-flight CONNECT probe only (revoked-credential UX, design 042 §9); the +# data path goes through aimdb-mqtt-connector. Same feature set as the +# connector's own rumqttc. +rumqttc = { version = "0.25", default-features = false, features = [ + "use-native-tls", +] } diff --git a/examples/weather-mesh-demo/weather-station/src/main.rs b/examples/weather-mesh-demo/weather-station/src/main.rs new file mode 100644 index 00000000..d2ad005a --- /dev/null +++ b/examples/weather-mesh-demo/weather-station/src/main.rs @@ -0,0 +1,488 @@ +//! # Weather Station (generic) +//! +//! The public-mesh station binary (design 042 §7/§8): configured entirely by +//! the `station.toml` profile that `aimdb join` writes — broker URL and +//! credentials, assigned slot, station name, and the (coarsened) location the +//! real weather data is fetched for. One profile, one command: +//! +//! ```bash +//! aimdb join https://mesh.aimdb.dev # writes station.toml +//! cargo run -p weather-station -- --config station.toml +//! ``` +//! +//! Same contracts, same Open-Meteo source as station alpha — only the key +//! namespace differs: this station publishes into its assigned slot +//! (`station/{slot}/…`) instead of a compile-time enum topic. + +use aimdb_core::{buffer::BufferCfg, AimDbBuilder, RecordKey, StringKey}; +use aimdb_data_contracts::Linkable; +use aimdb_mqtt_connector::MqttConnector; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +use clap::Parser; +use serde::Deserialize; +use std::sync::Arc; +use std::time::Duration; +use tracing::info; +use weather_mesh_common::{DewPoint, Humidity, Temperature}; + +/// Generic cloud weather station — runs from a `station.toml` profile +#[derive(Debug, Parser)] +#[command(name = "weather-station", version, about)] +struct Cli { + /// Path to the station profile written by `aimdb join` (design 043 §4) + #[arg(long)] + config: std::path::PathBuf, +} + +/// The station profile (design 043 §4). Unknown fields are ignored so the +/// service can extend the format without a version bump. +#[derive(Debug, Deserialize)] +struct StationProfile { + profile_version: u64, + station_id: String, + broker: BrokerProfile, + app: AppProfile, +} + +#[derive(Debug, Deserialize)] +struct BrokerProfile { + url: String, + username: String, + password: String, +} + +/// The flagship's `app` map: name + service-coarsened coordinates (042 D8). +#[derive(Debug, Deserialize)] +struct AppProfile { + name: String, + lat: f64, + lon: f64, +} + +#[tokio::main] +async fn main() { + // Display-format errors (revoked slot, bad profile, …) instead of the + // default Debug dump — these messages are the failure UX (design 042 §9). + if let Err(e) = run().await { + eprintln!("Error: {e}"); + std::process::exit(1); + } +} + +async fn run() -> Result<(), Box> { + // Initialize logging + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "weather_station=info,aimdb_core=info".into()), + ) + .init(); + + let cli = Cli::parse(); + let profile: StationProfile = toml::from_str(&std::fs::read_to_string(&cli.config)?)?; + + if profile.profile_version != 1 { + return Err(format!( + "unsupported profile_version {} (this station understands 1) — \ + re-run `aimdb join` with a current CLI", + profile.profile_version + ) + .into()); + } + + // The flagship assigns slot-scoped identities: station_id = "slot-" + // maps onto the hub's pool records station..* (design 042 §4). + let slot: u16 = profile + .station_id + .strip_prefix("slot-") + .and_then(|n| n.parse().ok()) + .ok_or_else(|| { + format!( + "station_id '{}' is not of the form slot- — \ + is this profile from a weather-mesh deployment?", + profile.station_id + ) + })?; + + info!("🚀 Starting Weather Station \"{}\"", profile.app.name); + info!( + "📡 Broker: {} (slot {})", + redact_url(&profile.broker.url), + slot + ); + info!( + "🌍 Weather location: {:.2}°N, {:.2}°E", + profile.app.lat, profile.app.lon + ); + + let client_id = format!("weather-station-{slot}"); + + // Fail fast on a dead credential instead of retrying forever: a revoked + // slot should say so at the moment the user is looking (design 042 §9). + preflight_broker_check(&profile.broker, &client_id).await?; + + let mqtt_url = url_with_credentials(&profile.broker)?; + + // Create runtime adapter + let adapter = Arc::new(TokioAdapter::new()?); + + // Build database with MQTT connector + let mut builder = AimDbBuilder::new() + .runtime(adapter) + .with_connector(MqttConnector::new(&mqtt_url).with_client_id(&client_id)); + + // Configure temperature record + let temp_key = StringKey::intern(format!("station.{slot}.temperature")); + let temp_topic = format!("mqtt://station/{slot}/temperature"); + builder.configure::(temp_key, |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) + .link_to(&temp_topic) + .with_serializer(|_ctx, t: &Temperature| { + t.to_bytes() + .map_err(|_| aimdb_core::connector::SerializeError::InvalidData) + }) + .finish(); + }); + + // Configure humidity record + let humidity_key = StringKey::intern(format!("station.{slot}.humidity")); + let humidity_topic = format!("mqtt://station/{slot}/humidity"); + builder.configure::(humidity_key, |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) + .link_to(&humidity_topic) + .with_serializer(|_ctx, h: &Humidity| { + h.to_bytes() + .map_err(|_| aimdb_core::connector::SerializeError::InvalidData) + }) + .finish(); + }); + + // Configure dew point record — derived from temperature and humidity + let dew_point_key = StringKey::intern(format!("station.{slot}.dew_point")); + let dew_point_topic = format!("mqtt://station/{slot}/dew_point"); + builder.configure::(dew_point_key, |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) + .transform_join(|b| { + b.input::(temp_key) + .input::(humidity_key) + .on_triggers(|mut rx, producer| async move { + let mut last_temp: Option = None; + let mut last_hum: Option = None; + while let Ok(trigger) = rx.recv().await { + match trigger.index() { + 0 => last_temp = trigger.as_input::().cloned(), + 1 => last_hum = trigger.as_input::().cloned(), + _ => {} + } + if let (Some(t), Some(h)) = (&last_temp, &last_hum) { + // Magnus approximation: T_dp ≈ T - (100 - RH) / 5 + let dew_point = t.celsius - (100.0 - h.percent) / 5.0; + let timestamp = t.timestamp.max(h.timestamp); + producer.produce(DewPoint { + celsius: dew_point, + timestamp, + }); + } + } + }) + }) + .link_to(&dew_point_topic) + .with_serializer(|_ctx, d: &DewPoint| { + d.to_bytes() + .map_err(|_| aimdb_core::connector::SerializeError::InvalidData) + }) + .finish(); + }); + + let (db, runner) = builder.build().await?; + tokio::spawn(runner.run()); + + info!("✅ Database initialized with 3 record types"); + + // Get producers + let temp_producer = db.producer::(temp_key.as_str())?; + let humidity_producer = db.producer::(humidity_key.as_str())?; + + // Spawn weather data producer + info!("🌤️ Starting weather data producer..."); + let station_name = profile.app.name.clone(); + let (lat, lon) = (profile.app.lat, profile.app.lon); + tokio::spawn(async move { + let client = OpenMeteoClient::new(); + let mut first_publish = true; + + loop { + match client.fetch_current(lat, lon).await { + Ok(data) => { + let now = chrono::Utc::now().timestamp_millis() as u64; + + temp_producer.produce(Temperature::new(data.temperature as f32, now)); + humidity_producer.produce(Humidity { + percent: data.humidity as f32, + timestamp: now, + }); + + tracing::info!( + "📊 Published: {:.1}°C, {:.1}%", + data.temperature, + data.humidity + ); + + if first_publish { + first_publish = false; + print_live_banner(&station_name); + } + } + Err(e) => { + tracing::warn!("Failed to fetch weather data: {}", e); + } + } + + // OpenMeteo updates every 15 minutes, poll every 5 min + tokio::time::sleep(Duration::from_secs(300)).await; + } + }); + + info!(""); + info!("🎯 Weather Station \"{}\" ready!", profile.app.name); + info!("📡 Publishing to MQTT topics:"); + info!(" - {}", temp_topic); + info!(" - {}", humidity_topic); + info!(" - {}", dew_point_topic); + info!(""); + info!("Press Ctrl+C to stop"); + + // Keep running + tokio::signal::ctrl_c().await?; + info!("🛑 Shutting down..."); + + Ok(()) +} + +/// The shareable moment (design 042 §8, tier 2): after the first successful +/// publish, hand the user their station's live chart and a ready-made MCP +/// prompt for its own data. +fn print_live_banner(name: &str) { + println!(); + println!("🗺️ Your station is live: https://aimdb.dev/mesh/{name}"); + println!("🤖 Ask any MCP agent about it — point it at https://aimdb.dev/mcp and try:"); + println!(" \"What is the latest temperature and dew point at station {name}?\""); + println!(); +} + +/// One CONNECT → CONNACK round-trip before building the database. +/// +/// The connector's event loop retries connection errors forever (by design — +/// brokers come and go). For an admission-scoped credential that is the wrong +/// UX: a revoked slot would retry silently for eternity. Probe once and turn +/// an auth rejection into a human answer (design 042 §9). +async fn preflight_broker_check( + broker: &BrokerProfile, + client_id: &str, +) -> Result<(), Box> { + let (tls, host, port) = split_broker_url(&broker.url)?; + + let mut opts = rumqttc::MqttOptions::new(format!("{client_id}-preflight"), host, port); + opts.set_keep_alive(Duration::from_secs(10)); + opts.set_credentials(&broker.username, &broker.password); + if tls { + opts.set_transport(rumqttc::Transport::Tls(rumqttc::TlsConfiguration::Native)); + } + + let (client, mut event_loop) = rumqttc::AsyncClient::new(opts, 4); + let result = tokio::time::timeout(Duration::from_secs(15), async { + loop { + match event_loop.poll().await { + Ok(rumqttc::Event::Incoming(rumqttc::Packet::ConnAck(_))) => return Ok(()), + Ok(_) => continue, + Err(e) => return Err(e), + } + } + }) + .await; + let _ = client.disconnect().await; + + match result { + Ok(Ok(())) => { + info!("✅ Broker accepted the station credential"); + Ok(()) + } + Ok(Err(rumqttc::ConnectionError::ConnectionRefused(code))) => Err(format!( + "the broker rejected this station's credential ({code:?}).\n \ + The slot was likely revoked (silent for 30 days, or by the operator).\n \ + Re-join the mesh for a fresh slot: aimdb join " + ) + .into()), + Ok(Err(e)) => Err(format!( + "cannot reach the broker at {}: {e}", + redact_url(&broker.url) + ) + .into()), + Err(_) => Err(format!( + "timed out connecting to the broker at {}", + redact_url(&broker.url) + ) + .into()), + } +} + +/// Split a `mqtt[s]://host[:port]` broker URL into (tls, host, port). +fn split_broker_url(url: &str) -> Result<(bool, String, u16), String> { + let (scheme, rest) = url + .split_once("://") + .ok_or_else(|| format!("broker URL '{url}' has no scheme"))?; + let tls = match scheme { + "mqtt" => false, + "mqtts" => true, + other => return Err(format!("unsupported broker scheme '{other}'")), + }; + let authority = rest.split('/').next().unwrap_or(rest); + let (host, port) = match authority.rsplit_once(':') { + Some((host, port)) => ( + host.to_string(), + port.parse() + .map_err(|_| format!("invalid broker port in '{url}'"))?, + ), + None => (authority.to_string(), if tls { 8883 } else { 1883 }), + }; + Ok((tls, host, port)) +} + +/// Embed the profile's credential into the connector URL +/// (`mqtts://user:pass@host:port`), the form the MQTT connector parses. +fn url_with_credentials(broker: &BrokerProfile) -> Result { + let (scheme, rest) = broker + .url + .split_once("://") + .ok_or_else(|| format!("broker URL '{}' has no scheme", broker.url))?; + Ok(format!( + "{scheme}://{}:{}@{rest}", + broker.username, broker.password + )) +} + +/// Strip URL-embedded credentials before logging. +fn redact_url(url: &str) -> String { + if let Some((scheme, rest)) = url.split_once("://") { + if let Some((_creds, host)) = rest.rsplit_once('@') { + return format!("{scheme}://…@{host}"); + } + } + url.to_string() +} + +// --- OpenMeteo API Client (same source as station alpha) --- + +struct OpenMeteoClient { + client: reqwest::Client, +} + +#[derive(Debug, Deserialize)] +struct OpenMeteoResponse { + current: CurrentWeather, +} + +#[derive(Debug, Deserialize)] +struct CurrentWeather { + temperature_2m: f64, + relative_humidity_2m: f64, +} + +struct WeatherData { + temperature: f64, + humidity: f64, +} + +impl OpenMeteoClient { + fn new() -> Self { + Self { + client: reqwest::Client::new(), + } + } + + async fn fetch_current(&self, lat: f64, lon: f64) -> Result { + let url = format!( + "https://api.open-meteo.com/v1/forecast?latitude={}&longitude={}¤t=temperature_2m,relative_humidity_2m", + lat, lon + ); + + let resp: OpenMeteoResponse = self.client.get(&url).send().await?.json().await?; + + Ok(WeatherData { + temperature: resp.current.temperature_2m, + humidity: resp.current.relative_humidity_2m, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn broker(url: &str) -> BrokerProfile { + BrokerProfile { + url: url.to_string(), + username: "station-17".to_string(), + password: "s3cret".to_string(), + } + } + + #[test] + fn profile_parses_the_043_sample() { + let profile: StationProfile = toml::from_str( + r#" + profile_version = 1 + station_id = "slot-17" + + [broker] + url = "mqtts://xxxx.eu-central-1.emqx.cloud:8883" + username = "station-17" + password = "s3cret" + + [app] + name = "graz-balcony" + lat = 47.07 + lon = 15.44 + "#, + ) + .unwrap(); + assert_eq!(profile.profile_version, 1); + assert_eq!(profile.station_id, "slot-17"); + assert_eq!(profile.app.name, "graz-balcony"); + } + + #[test] + fn broker_url_splits_with_scheme_defaults() { + assert_eq!( + split_broker_url("mqtts://broker.example.com:8883").unwrap(), + (true, "broker.example.com".to_string(), 8883) + ); + assert_eq!( + split_broker_url("mqtts://broker.example.com").unwrap(), + (true, "broker.example.com".to_string(), 8883) + ); + assert_eq!( + split_broker_url("mqtt://localhost").unwrap(), + (false, "localhost".to_string(), 1883) + ); + assert!(split_broker_url("http://x").is_err()); + assert!(split_broker_url("no-scheme").is_err()); + } + + #[test] + fn connector_url_carries_credentials() { + assert_eq!( + url_with_credentials(&broker("mqtts://broker.example.com:8883")).unwrap(), + "mqtts://station-17:s3cret@broker.example.com:8883" + ); + } + + #[test] + fn redaction_hides_credentials() { + assert_eq!( + redact_url("mqtts://station-17:s3cret@broker.example.com:8883"), + "mqtts://…@broker.example.com:8883" + ); + assert_eq!(redact_url("mqtt://localhost"), "mqtt://localhost"); + } +} diff --git a/tools/aimdb-cli/Cargo.toml b/tools/aimdb-cli/Cargo.toml index f2399ff4..1ada495c 100644 --- a/tools/aimdb-cli/Cargo.toml +++ b/tools/aimdb-cli/Cargo.toml @@ -19,9 +19,13 @@ path = "src/main.rs" aimdb-client = { version = "0.6.0", path = "../../aimdb-client" } aimdb-codegen = { version = "0.2.0", path = "../../aimdb-codegen" } -# Core dependencies - reuse protocol types from aimdb-core +# Core dependencies - reuse protocol types from aimdb-core. +# `observability` keeps the optional RecordMetadata fields (signal_stats, +# stage_profiling, buffer metrics) in the client-side struct — without it +# serde silently drops them from `record list --format json` output. aimdb-core = { version = "1.1.0", path = "../../aimdb-core", features = [ "std", + "observability", ] } serde = { version = "1", features = ["derive"] } serde_json = "1" @@ -53,9 +57,23 @@ chrono = "0.4" # Optional: YAML output serde_yaml = { version = "0.9", optional = true } +# `join` only (design 043): HTTP client for the provisioning endpoint + the +# GitHub device flow, and TOML for the station profile. rustls so released +# binaries don't depend on a system OpenSSL; native roots (system trust +# store) rather than the bundled webpki-roots, whose CDLA license is not on +# the deny.toml allowlist. +reqwest = { version = "0.12", default-features = false, features = [ + "rustls-tls-native-roots", + "json", +], optional = true } +toml = { version = "0.8", optional = true } + [features] -default = [] +default = ["join"] yaml = ["serde_yaml"] +# `aimdb join ` — admit this machine to a deployment +# (design 043). On by default in released binaries. +join = ["dep:reqwest", "dep:toml"] # Add the serial transport to the endpoint resolver, so `--connect serial://…` # works. Off by default (pulls tokio-serial → libudev on Linux). transport-serial = ["aimdb-client/transport-serial"] diff --git a/tools/aimdb-cli/src/commands/join.rs b/tools/aimdb-cli/src/commands/join.rs new file mode 100644 index 00000000..b3f6c315 --- /dev/null +++ b/tools/aimdb-cli/src/commands/join.rs @@ -0,0 +1,479 @@ +//! Join Command — admit this machine to an AimDB deployment +//! +//! Speaks the open provisioning format of design 043 (`POST /v1/join`): +//! authenticate (GitHub device flow, or a pre-issued claim token), pass the +//! deployment's `app` metadata through untouched, and write the returned +//! station profile as `station.toml`. Generic by construction — the endpoint +//! URL is an argument and no deployment-specific types appear here. + +use std::io::Write as _; +use std::path::PathBuf; +use std::time::Duration; + +use anyhow::{anyhow, Context}; +use clap::Args; +use serde::{Deserialize, Serialize}; + +use crate::error::CliResult; + +/// Public OAuth client ID of the flagship's GitHub App — public by design, +/// nothing secret ships in the CLI (design 042 §6.1). Deployments building +/// their own binaries bake their own app in at build time: +/// `AIMDB_GITHUB_CLIENT_ID=… cargo build`. +const GITHUB_CLIENT_ID: Option<&str> = option_env!("AIMDB_GITHUB_CLIENT_ID"); + +const GITHUB_DEVICE_CODE_URL: &str = "https://github.com/login/device/code"; +const GITHUB_ACCESS_TOKEN_URL: &str = "https://github.com/login/oauth/access_token"; +const GITHUB_USER_URL: &str = "https://api.github.com/user"; + +/// The station profile format this CLI understands (design 043 §3/§5). +const PROFILE_VERSION: u64 = 1; + +/// `app` fields collected via prompts (design 043 §2.2). The values are +/// passed through untouched; the service validates and echoes the final +/// values back in the profile. +const APP_PROMPTS: &[(&str, &str)] = &[("name", "Station name"), ("location", "Location (city)")]; + +/// Join an AimDB deployment via its provisioning endpoint +#[derive(Debug, Args)] +pub struct JoinCommand { + /// Base URL of the provisioning endpoint (the CLI POSTs to /v1/join) + provisioning_url: String, + + /// Pre-issued claim token — selects the `claim-token` auth kind and skips + /// the GitHub device flow (for deployments without public self-serve) + #[arg(long)] + token: Option, + + /// Where to write the station profile (contains the broker credential; + /// written with mode 0600) + #[arg(long, default_value = "station.toml")] + out: PathBuf, +} + +/// Identity envelope of the join request (design 043 §2.1). +#[derive(Debug, Serialize)] +struct AuthEnvelope { + kind: &'static str, + token: String, +} + +/// The station profile returned on success (design 043 §3.1). Unknown fields +/// are ignored on both ends so services can extend the format without a +/// version bump (043 §5). +#[derive(Debug, Serialize, Deserialize)] +struct StationProfile { + profile_version: u64, + station_id: String, + broker: BrokerProfile, + app: serde_json::Map, +} + +#[derive(Debug, Serialize, Deserialize)] +struct BrokerProfile { + url: String, + username: String, + password: String, +} + +impl JoinCommand { + pub async fn execute(self) -> CliResult<()> { + let client = reqwest::Client::builder() + .user_agent(concat!("aimdb-cli/", env!("CARGO_PKG_VERSION"))) + .timeout(Duration::from_secs(30)) + .build() + .context("failed to build HTTP client")?; + + let auth = match self.token { + Some(token) => AuthEnvelope { + kind: "claim-token", + token, + }, + None => AuthEnvelope { + kind: "github", + token: github_device_flow(&client).await?, + }, + }; + + let mut app = serde_json::Map::new(); + for (key, label) in APP_PROMPTS { + app.insert((*key).to_string(), prompt(label)?.into()); + } + + let profile = request_join(&client, &self.provisioning_url, &auth, &app).await?; + + if profile.profile_version > PROFILE_VERSION { + eprintln!( + "⚠ profile_version {} is newer than this CLI understands ({}); writing it anyway", + profile.profile_version, PROFILE_VERSION + ); + } + + write_profile(&self.out, &profile)?; + + let name = profile + .app + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("station"); + println!( + "✓ {} \"{}\" — credentials written to {}", + profile.station_id, + name, + self.out.display() + ); + Ok(()) + } +} + +/// GitHub OAuth device flow (design 042 §6.1): print a short code, the user +/// confirms it at github.com/login/device, poll for the access token. No +/// scopes are requested — public identity only. +async fn github_device_flow(client: &reqwest::Client) -> CliResult { + let client_id = GITHUB_CLIENT_ID.ok_or_else(|| { + anyhow!( + "this build has no GitHub OAuth client ID compiled in\n \ + Hint: use --token for claim-token deployments, or rebuild with \ + AIMDB_GITHUB_CLIENT_ID= for GitHub self-serve" + ) + })?; + + #[derive(Deserialize)] + struct DeviceCode { + device_code: String, + user_code: String, + verification_uri: String, + expires_in: u64, + interval: u64, + } + + let device: DeviceCode = client + .post(GITHUB_DEVICE_CODE_URL) + .header("Accept", "application/json") + .json(&serde_json::json!({ "client_id": client_id })) + .send() + .await + .context("device-code request to GitHub failed")? + .error_for_status() + .context("device-code request to GitHub failed")? + .json() + .await + .context("unexpected device-code response from GitHub")?; + + println!( + "→ To authorize, enter code {} at {}", + device.user_code, device.verification_uri + ); + + #[derive(Deserialize)] + struct TokenPoll { + access_token: Option, + error: Option, + interval: Option, + } + + let deadline = std::time::Instant::now() + Duration::from_secs(device.expires_in); + let mut interval = device.interval.max(1); + let token = loop { + tokio::time::sleep(Duration::from_secs(interval)).await; + if std::time::Instant::now() > deadline { + return Err( + anyhow!("device code expired before authorization — run join again").into(), + ); + } + + let poll: TokenPoll = client + .post(GITHUB_ACCESS_TOKEN_URL) + .header("Accept", "application/json") + .json(&serde_json::json!({ + "client_id": client_id, + "device_code": device.device_code, + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + })) + .send() + .await + .context("access-token poll to GitHub failed")? + .json() + .await + .context("unexpected access-token response from GitHub")?; + + if let Some(token) = poll.access_token { + break token; + } + match poll.error.as_deref() { + Some("authorization_pending") => {} + // GitHub asks us to back off; it may supply the new interval. + Some("slow_down") => interval = poll.interval.unwrap_or(interval + 5), + Some("expired_token") => { + return Err( + anyhow!("device code expired before authorization — run join again").into(), + ) + } + Some("access_denied") => { + return Err(anyhow!("authorization was denied on GitHub").into()) + } + other => { + return Err(anyhow!( + "GitHub device flow failed: {}", + other.unwrap_or("no error code in response") + ) + .into()) + } + } + }; + + // Cosmetic only — the service re-verifies the identity server-side. + #[derive(Deserialize)] + struct GithubUser { + login: String, + } + if let Ok(resp) = client + .get(GITHUB_USER_URL) + .header("Accept", "application/vnd.github+json") + .bearer_auth(&token) + .send() + .await + { + if let Ok(user) = resp.json::().await { + println!("✓ authorized as @{}", user.login); + } + } + + Ok(token) +} + +/// `POST /v1/join` and interpret the response per design 043. +async fn request_join( + client: &reqwest::Client, + base_url: &str, + auth: &AuthEnvelope, + app: &serde_json::Map, +) -> CliResult { + let url = format!("{}/v1/join", base_url.trim_end_matches('/')); + let response = client + .post(&url) + .json(&serde_json::json!({ "auth": auth, "app": app })) + .send() + .await + .with_context(|| format!("request to {url} failed"))?; + + let status = response.status().as_u16(); + let body = response + .text() + .await + .with_context(|| format!("failed to read response from {url}"))?; + parse_join_response(status, &body) +} + +/// Interpret a `/v1/join` response (design 043 §3). On 403/409/503 the body's +/// `message` is surfaced as-is — the service owns the failure UX text. +fn parse_join_response(status: u16, body: &str) -> CliResult { + #[derive(Deserialize)] + struct ErrorBody { + message: String, + } + + match status { + 200 => serde_json::from_str(body) + .context("malformed station profile in join response") + .map_err(Into::into), + 403 | 409 | 503 => { + let message = serde_json::from_str::(body) + .map(|e| e.message) + .unwrap_or_else(|_| format!("join rejected with status {status}")); + Err(anyhow!("{message}").into()) + } + other => Err(anyhow!( + "unexpected response from provisioning endpoint (status {other}): {}", + body.trim() + ) + .into()), + } +} + +/// Write the profile as TOML (design 043 §4), mode 0600 — it carries the +/// broker credential. +fn write_profile(path: &PathBuf, profile: &StationProfile) -> CliResult<()> { + let toml = + toml::to_string_pretty(profile).context("failed to serialize station profile as TOML")?; + + let mut options = std::fs::OpenOptions::new(); + options.write(true).create(true).truncate(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let mut file = options + .open(path) + .with_context(|| format!("failed to open {}", path.display()))?; + file.write_all(toml.as_bytes()) + .with_context(|| format!("failed to write {}", path.display()))?; + Ok(()) +} + +/// Prompt on stdin for one `app` field; re-asks until non-empty. +fn prompt(label: &str) -> CliResult { + loop { + print!("{label}: "); + std::io::stdout().flush()?; + let mut line = String::new(); + std::io::stdin().read_line(&mut line)?; + let value = line.trim(); + if !value.is_empty() { + return Ok(value.to_string()); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// The design 043 §3.1 sample response. + const PROFILE_JSON: &str = r#"{ + "profile_version": 1, + "station_id": "slot-17", + "broker": { + "url": "mqtts://xxxx.eu-central-1.emqx.cloud:8883", + "username": "station-17", + "password": "s3cret" + }, + "app": { "name": "graz-balcony", "lat": 47.07, "lon": 15.44 } + }"#; + + #[test] + fn profile_survives_json_to_toml_round_trip() { + let profile = parse_join_response(200, PROFILE_JSON).unwrap(); + let toml_str = toml::to_string_pretty(&profile).unwrap(); + + let reread: StationProfile = toml::from_str(&toml_str).unwrap(); + assert_eq!(reread.profile_version, 1); + assert_eq!(reread.station_id, "slot-17"); + assert_eq!(reread.broker.username, "station-17"); + assert_eq!(reread.broker.password, "s3cret"); + assert_eq!( + reread.app.get("name").and_then(|v| v.as_str()), + Some("graz-balcony") + ); + assert_eq!(reread.app.get("lat").and_then(|v| v.as_f64()), Some(47.07)); + } + + #[test] + fn unknown_profile_fields_are_ignored() { + // 043 §5: services may add fields without a version bump. + let body = PROFILE_JSON.replacen( + "\"profile_version\": 1,", + "\"profile_version\": 1, \"issued_at\": \"2026-07-09\",", + 1, + ); + let profile = parse_join_response(200, &body).unwrap(); + assert_eq!(profile.station_id, "slot-17"); + } + + #[test] + fn rejection_surfaces_service_message_as_is() { + for status in [403, 409, 503] { + let err = parse_join_response( + status, + r#"{ "message": "The mesh is full (64/64 slots) — try again later." }"#, + ) + .unwrap_err(); + assert_eq!( + err.to_string(), + "The mesh is full (64/64 slots) — try again later." + ); + } + } + + #[test] + fn rejection_without_json_body_still_reports_status() { + let err = parse_join_response(503, "upstream timeout").unwrap_err(); + assert!(err.to_string().contains("503"), "got: {err}"); + } + + #[test] + fn unexpected_status_is_an_error() { + let err = parse_join_response(500, "oops").unwrap_err(); + assert!(err.to_string().contains("500"), "got: {err}"); + } + + #[test] + fn malformed_success_body_is_an_error() { + let err = parse_join_response(200, r#"{ "hello": 1 }"#).unwrap_err(); + assert!(err.to_string().contains("profile"), "got: {err}"); + } + + #[test] + fn profile_written_with_owner_only_permissions() { + let profile = parse_join_response(200, PROFILE_JSON).unwrap(); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("station.toml"); + write_profile(&path, &profile).unwrap(); + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(&path).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, 0o600); + } + let reread: StationProfile = + toml::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap(); + assert_eq!(reread.station_id, "slot-17"); + } + + /// One-shot HTTP server: accepts a single connection, reads the request, + /// answers with the canned status/body, then closes. + async fn mock_join_server(status_line: &'static str, body: &'static str) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + // Drain the request headers + body far enough to respond politely. + let mut buf = [0u8; 4096]; + let _ = socket.read(&mut buf).await; + let response = format!( + "HTTP/1.1 {status_line}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + socket.write_all(response.as_bytes()).await.unwrap(); + }); + format!("http://{addr}") + } + + #[tokio::test] + async fn join_request_round_trips_against_mock_server() { + let base = mock_join_server("200 OK", PROFILE_JSON).await; + let client = reqwest::Client::new(); + let auth = AuthEnvelope { + kind: "claim-token", + token: "t".into(), + }; + let profile = request_join(&client, &base, &auth, &serde_json::Map::new()) + .await + .unwrap(); + assert_eq!(profile.station_id, "slot-17"); + } + + #[tokio::test] + async fn join_request_surfaces_mock_rejection_message() { + let base = mock_join_server( + "409 Conflict", + r#"{ "message": "Your account already runs graz-balcony (slot-17)." }"#, + ) + .await; + let client = reqwest::Client::new(); + let auth = AuthEnvelope { + kind: "claim-token", + token: "t".into(), + }; + let err = request_join(&client, &base, &auth, &serde_json::Map::new()) + .await + .unwrap_err(); + assert_eq!( + err.to_string(), + "Your account already runs graz-balcony (slot-17)." + ); + } +} diff --git a/tools/aimdb-cli/src/commands/mod.rs b/tools/aimdb-cli/src/commands/mod.rs index dfb501bb..70264ba5 100644 --- a/tools/aimdb-cli/src/commands/mod.rs +++ b/tools/aimdb-cli/src/commands/mod.rs @@ -5,6 +5,8 @@ pub mod generate; pub mod graph; pub mod instance; +#[cfg(feature = "join")] +pub mod join; pub mod record; pub mod watch; diff --git a/tools/aimdb-cli/src/main.rs b/tools/aimdb-cli/src/main.rs index 21f8d100..23240a50 100644 --- a/tools/aimdb-cli/src/main.rs +++ b/tools/aimdb-cli/src/main.rs @@ -4,6 +4,8 @@ //! AimDB instances via the AimX remote access protocol (v2 NDJSON wire). use clap::{Parser, Subcommand}; +#[cfg(feature = "join")] +use commands::join::JoinCommand; use commands::{ generate::GenerateCommand, graph::GraphCommand, instance::InstanceCommand, record::RecordCommand, watch::WatchCommand, @@ -54,6 +56,11 @@ enum Command { /// Generate architecture artefacts from state.toml #[command(name = "generate")] Generate(GenerateCommand), + + /// Join an AimDB deployment via its provisioning endpoint (design 043) + #[cfg(feature = "join")] + #[command(name = "join")] + Join(JoinCommand), } #[tokio::main] @@ -73,6 +80,8 @@ async fn main() { Command::Graph(cmd) => cmd.execute(endpoint).await, Command::Watch(cmd) => cmd.execute(endpoint).await, Command::Generate(cmd) => cmd.execute().await, + #[cfg(feature = "join")] + Command::Join(cmd) => cmd.execute().await, }; if let Err(e) = result {