From 0a8e72e762f5ba2c92f83e6ede57c183b3fdbb8e Mon Sep 17 00:00:00 2001 From: ruv Date: Mon, 27 Jul 2026 14:14:04 -0400 Subject: [PATCH] feat(homecore): complete migration and startup restore --- .../ADR-127-homecore-state-machine-rust.md | 5 + ...mecore-recorder-history-semantic-search.md | 6 + ...65-homecore-migrate-from-home-assistant.md | 46 +- v2/Cargo.lock | 2 + v2/crates/homecore-migrate/README.md | 187 ++---- v2/crates/homecore-migrate/src/cli.rs | 19 +- .../homecore-migrate/src/config_entries.rs | 352 +++++++++--- .../homecore-migrate/src/device_registry.rs | 226 +++++--- .../homecore-migrate/src/entity_registry.rs | 46 +- v2/crates/homecore-migrate/src/lib.rs | 18 +- v2/crates/homecore-migrate/src/main.rs | 78 ++- v2/crates/homecore-migrate/src/storage.rs | 69 +++ v2/crates/homecore-recorder/README.md | 3 + v2/crates/homecore-recorder/src/db.rs | 541 +++++++++++++++--- v2/crates/homecore-recorder/src/lib.rs | 5 +- v2/crates/homecore-server/Cargo.toml | 4 +- v2/crates/homecore-server/README.md | 7 + v2/crates/homecore-server/src/main.rs | 58 +- v2/crates/homecore-server/src/restore.rs | 219 +++++++ v2/crates/homecore/src/event.rs | 17 + v2/crates/homecore/src/homecore.rs | 8 +- v2/crates/homecore/src/lib.rs | 17 +- v2/crates/homecore/src/registry.rs | 80 ++- v2/crates/homecore/src/state.rs | 20 +- 24 files changed, 1552 insertions(+), 481 deletions(-) create mode 100644 v2/crates/homecore-server/src/restore.rs diff --git a/docs/adr/ADR-127-homecore-state-machine-rust.md b/docs/adr/ADR-127-homecore-state-machine-rust.md index 1887ed3f..4df20d0e 100644 --- a/docs/adr/ADR-127-homecore-state-machine-rust.md +++ b/docs/adr/ADR-127-homecore-state-machine-rust.md @@ -82,6 +82,11 @@ The entity registry is a `RwLock>` backed by an a `DeviceRegistry` mirrors HA's `core.device_registry` schema (version 13). Devices are identified by a set of `(id_type, id_value)` tuples (the `identifiers` field), which matches HA's pattern of accepting multiple identifier types per device (MAC address, serial number, integration-specific ID). +`DeviceEntry` and the in-memory `DeviceRegistry` are implemented. On server +startup, entity and device registry files are restored in deterministic key +order with a configurable hard row bound; malformed individual entries are +isolated and reported. + --- ## 3. HA-side reference table diff --git a/docs/adr/ADR-132-homecore-recorder-history-semantic-search.md b/docs/adr/ADR-132-homecore-recorder-history-semantic-search.md index d1b75fe7..1bfb7c09 100644 --- a/docs/adr/ADR-132-homecore-recorder-history-semantic-search.md +++ b/docs/adr/ADR-132-homecore-recorder-history-semantic-search.md @@ -148,6 +148,12 @@ correctness, fail-closed write integrity, semantic-store NaN poisoning, and PII - **Memory-DoS — `get_state_history` was unbounded.** No `LIMIT`, so a wide time window over a high-frequency entity loaded an unbounded row set into memory. Now capped at `MAX_HISTORY_ROWS` (1,000,000); sibling search paths were already `k`-bounded. +- **Startup state restoration.** `latest_states(limit)` selects one newest row + per entity with `(last_updated_ts, state_id)` tie-breaking, orders results by + entity ID, and caps requests at 100,000. Malformed rows are skipped with + typed warnings. `restore_latest` preserves recorded timestamps and installs + snapshots with a `homecore.restore` context before the recorder listener and + automation engine start. - **Disk-DoS / documented-but-missing `purge`.** The README advertised `Recorder::purge`, but no retention path existed → unbounded disk growth. Added a **transactional** `purge(older_than)` with an **exclusive** cutoff (idempotent, no off-by-one) that deletes old `states`/`events` and diff --git a/docs/adr/ADR-165-homecore-migrate-from-home-assistant.md b/docs/adr/ADR-165-homecore-migrate-from-home-assistant.md index 4f68b231..d9e95c03 100644 --- a/docs/adr/ADR-165-homecore-migrate-from-home-assistant.md +++ b/docs/adr/ADR-165-homecore-migrate-from-home-assistant.md @@ -2,7 +2,7 @@ | Field | Value | |-------|-------| -| **Status** | Accepted — P1 scaffold (full conversion deferred to P2) | +| **Status** | Accepted — registry/config persistence implemented | | **Date** | 2026-05-25 | | **Deciders** | ruv | | **Codename** | **HOMECORE-MIGRATE** | @@ -44,8 +44,8 @@ files are read, how schema versions are validated, and what happens on an unknow ## 2. Decision Ship `homecore-migrate` as a CLI + library that reads an existing HA filesystem and imports -its configuration into HOMECORE. P1 is a **scaffold**: it parses and inspects everything and -converts the entity registry; full conversion of the remaining artifacts is deferred to P2. +its configuration into HOMECORE. Registry and config-entry conversion are durable; automation +conversion and secret-reference resolution remain deferred. ### 2.1 Storage reader + versioned format gate (P1, shipped) @@ -57,22 +57,25 @@ converts the entity registry; full conversion of the remaining artifacts is defe unknown `minor_version` is a **hard error** (`MigrateError::UnsupportedSchemaVersion`), never a silent best-effort parse. Better to refuse than to corrupt. -### 2.2 Per-artifact parsers (P1, shipped) +### 2.2 Per-artifact conversion (shipped) - `entity_registry::load()` — `core.entity_registry` → `Vec` (ready for import). -- `device_registry::load()` — `core.device_registry` → `Vec` (P1 diagnostic; - full conversion P2). -- `config_entries::load()` — `core.config_entries` → domain counts + integration names - (the format is undocumented per §6 Q5; treated diagnostically). +- `device_registry::read_device_registry()` converts the supported v13 device fields into + `homecore::DeviceEntry`; `write_device_registry()` emits an HA-compatible v13 envelope. +- `config_entries::convert_config_entries()` emits versioned `homecore.config_entries` + storage. Original rows are retained verbatim, while unsupported domains and fields produce + typed warnings instead of being discarded. - `secrets::load_secrets()` — `secrets.yaml` → `HashMap` (resolution P2). - `automations::load()` — `automations.yaml` → count + ID/alias list (conversion P2). -### 2.3 CLI (P1, shipped) +### 2.3 CLI - `homecore-migrate inspect ` previews what will be migrated (entity/device/config counts, redacted secret/automation lists) (`src/cli.rs`, `src/main.rs`). -- `import-entities` and `export-for-sidecar` are declared but their full behaviour is P2. +- `import-entities`, `import-devices`, and `import-config-entries` write destination files and + emit one-line JSON summaries. Writes use synced same-directory temporary files and atomic + no-clobber publication; an existing destination is never implicitly replaced. ### 2.4 Structured errors (P1, shipped) @@ -88,27 +91,25 @@ converts the entity registry; full conversion of the remaining artifacts is defe file path and a coarse location (`serde_yaml::Error::location()`), never the scalar content. Pinned by `secrets::tests::malformed_secrets_error_never_contains_secret_value` (asserts the rendered error **and its full `#[source]` chain** never contain the secret value). - **Review dimensions confirmed clean with evidence:** source is never mutated (no - `fs::write`/`remove`/`create` anywhere — P1 reads source, writes nothing); paths are + **Review dimensions confirmed clean with evidence:** source is never mutated; destination + writes are explicit `--to` paths and no-clobber; paths are user-supplied dirs joined with fixed filenames (no `..`/absolute traversal beyond the user's own privileges); malformed/typed/truncated `.storage` JSON and YAML **error, never panic** (every production `unwrap`/`expect` is test-only); unknown schema `minor_version` - hard-errors fail-closed; no SQL/shell/path injection surface (the tool emits diagnostics - only, persists nothing in P1). + hard-errors fail-closed; no SQL/shell injection surface. ### 2.5 Deferred to P2+ (NOT built — honestly labelled) -- Convert `config_entries` → HOMECORE plugin manifests. +- Execute imported config entries (a matching HOMECORE plugin must claim the preserved domain). - Convert `automations.yaml` → `homecore-automation` YAML. - Side-by-side runtime mode (requires `homecore-recorder`, ADR-132; behind the `recorder` Cargo feature, currently a no-op stub). - `!secret` reference resolution in non-secrets YAML files. -### 2.6 Test evidence (as shipped) +### 2.6 Test evidence -- 21 tests (`cargo test -p homecore-migrate`) — 19 as originally shipped plus 2 added by the - 2026-06 security review (`secrets::tests::malformed_secrets_error_never_contains_secret_value`, - `malformed_secrets_error_reports_location`). +- Targeted tests cover registry round trips, unknown versions, lossless unsupported config + fields/domains, malformed input, and crash-safe/no-overwrite destination behaviour. ## 3. Consequences @@ -118,13 +119,12 @@ converts the entity registry; full conversion of the remaining artifacts is defe schema drift fails loudly instead of corrupting an imported home. - Reusing HA's own `.storage` and YAML formats means no intermediate export step; the tool reads a live HA install directly. -- P1 `inspect` gives users a no-risk dry run before any write. +- `inspect` gives users a no-risk dry run before any write. **Negative / honest limits.** -- P1 is a **scaffold**: only the entity registry is conversion-ready. Device registry, - config-entry→plugin, automation, and secret-resolution conversions are P2 and **not yet - built** — the Status field and crate docs say so. +- Imported config entries are durable but do not install or execute Python HA integrations. +- Automation conversion and secret-reference resolution are not built. - The side-by-side recorder export depends on ADR-132 and is currently a feature-gated no-op. - Performance figures in the README (envelope parse < 5 ms, 1 000-entity load < 50 ms) are diff --git a/v2/Cargo.lock b/v2/Cargo.lock index e77e05f1..5f0b13fa 100644 --- a/v2/Cargo.lock +++ b/v2/Cargo.lock @@ -3675,6 +3675,7 @@ dependencies = [ "homecore-api", "homecore-assist", "homecore-automation", + "homecore-migrate", "homecore-plugins", "homecore-recorder", "http-body-util", @@ -3682,6 +3683,7 @@ dependencies = [ "serde", "serde_json", "serde_yaml", + "tempfile", "tokio", "tower 0.5.3", "tower-http", diff --git a/v2/crates/homecore-migrate/README.md b/v2/crates/homecore-migrate/README.md index 54b98441..b1e6ec20 100644 --- a/v2/crates/homecore-migrate/README.md +++ b/v2/crates/homecore-migrate/README.md @@ -1,144 +1,77 @@ # homecore-migrate -Migration tooling for importing Home Assistant configuration, entities, and secrets into HOMECORE. +Migration tooling for importing Home Assistant filesystem storage into +HOMECORE. The implementation follows +[ADR-165](../../docs/adr/ADR-165-homecore-migrate-from-home-assistant.md). -[![Crates.io](https://img.shields.io/crates/v/homecore-migrate.svg)](https://crates.io/crates/homecore-migrate) -![License](https://img.shields.io/badge/license-MIT-blue.svg) -![MSRV: 1.89+](https://img.shields.io/badge/MSRV-1.89%2B-purple.svg) -[![Tests](https://img.shields.io/badge/tests-19%20passing-brightgreen.svg)](https://github.com/ruvnet/RuView) -[![ADR-165](https://img.shields.io/badge/ADR-165-orange.svg)](../../docs/adr/ADR-165-homecore-migrate-from-home-assistant.md) +## Implemented -Parse and inspect Home Assistant's `.storage/` directory, entity registry, device registry, secrets, and automations. Convert existing HA configurations for import into HOMECORE (full conversion in P2). +- Reads versioned HA `.storage` JSON and rejects unknown schema versions. +- Converts entity registry metadata to `homecore::EntityEntry`. +- Converts the supported HA v13 device-registry fields to + `homecore::DeviceEntry`, including identifiers, connections, versions, + serial number, labels, topology, and config-entry links. +- Converts `core.config_entries` to a versioned + `homecore.config_entries` file. Each original row is retained verbatim. + Unsupported domains and non-portable fields produce typed warnings. +- Publishes destination JSON through a synced same-directory temporary file + and an atomic no-clobber link. Existing destination files are never replaced. +- Emits one-line JSON summaries from every import command. +- Parses secrets with redacted errors and inspects automations. -## What this crate does +## CLI -`homecore-migrate` reads Home Assistant's filesystem state and provides tooling to analyze and migrate it to HOMECORE. It includes: - -- **HaStorageDir** — reads HA's `.homeassistant/.storage/` directory and parses versioned JSON envelopes -- **Entity registry parser** — converts `core.entity_registry` JSON to HOMECORE `EntityEntry` types -- **Device registry parser** — reads `core.device_registry` (P1 diagnostic only; full conversion in P2) -- **Config entries parser** — reads `core.config_entries` to list active integrations -- **Secrets parser** — reads `secrets.yaml` as `HashMap` for reference resolution (P2) -- **Automations parser** — reads `automations.yaml` and counts/lists automations (full conversion in P2) -- **CLI binary** — `homecore-migrate inspect` to preview what will be migrated - -The tool enforces version schema compatibility: unknown HA schema versions are rejected (hard error per ADR-165 §6 Q5) rather than silently corrupting data. - -## Features - -- **Entity registry import** — `core.entity_registry` → an atomically written, - HA-compatible HOMECORE registry file; existing destinations are not overwritten -- **Device registry inspection** — read HA device metadata; full conversion deferred to P2 -- **Config entries analysis** — list active integrations by domain (enables gap analysis) -- **Secrets extraction** — read `secrets.yaml` references for annotation (resolution in P2) -- **Automations counting** — list automation IDs and aliases without conversion (conversion in P2) -- **Schema version validation** — explicit rejection of unknown HA versions (no silent corruption) -- **Structured error reporting** — `MigrateError` enum with context (file path, line number) -- **CLI subcommands** — `inspect` to preview, `import-entities` to load (P2), `export-for-sidecar` (P2) - -## Capabilities - -| Capability | Type | Method | Notes | -|------------|------|--------|-------| -| Read storage envelope | Parser | `storage::read_envelope(path)` | Deserialize `.storage/*.json` | -| Parse entity registry | Parser | `entity_registry::load(storage_dir)` | → `Vec` | -| Inspect device registry | Parser | `device_registry::load(storage_dir)` | → `Vec` (P1 diagnostic) | -| List config entries | Parser | `config_entries::load(storage_dir)` | → domain counts + names | -| Load secrets | Parser | `secrets::load_secrets(path)` | → `HashMap` | -| Count automations | Parser | `automations::load(path)` | → count + ID list | -| Validate schema version | Validator | `storage_format::validate_version(major, minor)` | Hard error if unknown | -| Convert to HOMECORE | Converter | `entity_registry::to_homecore_entries()` (P2) | → `homecore::EntityRegistry` | -| Export side-by-side DB | Exporter | `recorder::export_states()` (P2, `--features recorder`) | → `.homecore/home.db` | - -## Comparison to Home Assistant - -| Aspect | Home Assistant | homecore-migrate | -|--------|----------------|-----------------| -| State source | Python `.homeassistant/` directory | Same HA filesystem format | -| Entity registry format | JSON envelope in `.storage/core.entity_registry` | Identical format, schema v13 | -| Schema versioning | `version` + optional `minor_version` | Explicit version struct validation | -| Secrets resolution | `!secret` YAML references via loader | Planned P2 (reads `secrets.yaml`) | -| Automation conversion | Python → HA YAML (internal) | P2: convert to `homecore-automation` format | -| Device registry import | Python device types | P1 diagnostic; full conversion P2 | -| Side-by-side runtime | N/A (HA doesn't side-by-side migrate) | P2 feature: run old + new in parallel | -| CLI tooling | HA doesn't export | `homecore-migrate` binary with subcommands | - -## Performance - -- **Storage envelope parse** — < 5 ms per file (serde_json) -- **Entity registry load** — < 50 ms for 1,000 entities -- **Storage directory scan** — < 100 ms for full `.storage/` directory -- **Secrets file parse** — < 10 ms (YAML) -- **No per-crate benchmarks yet** — a follow-up issue tracks baseline measurements - -## Usage - -CLI inspection (P1): +The import commands take the HA `.storage` directory and the HOMECORE storage +destination: ```bash -# Inspect what will be migrated from an existing HA installation -homecore-migrate inspect ~/.homeassistant +homecore-migrate import-entities \ + --storage ~/.homeassistant/.storage \ + --to ~/.homecore/storage -# Output: -# Entity Registry: 47 entities -# light: 12 -# sensor: 20 -# binary_sensor: 10 -# switch: 5 -# Device Registry: 8 devices -# Config Entries: 6 integrations (mqtt, rest, zeroconf, ...) -# Secrets: 3 defined (redacted) -# Automations: 5 automations (redacted) +homecore-migrate import-devices \ + --storage ~/.homeassistant/.storage \ + --to ~/.homecore/storage + +homecore-migrate import-config-entries \ + --storage ~/.homeassistant/.storage \ + --to ~/.homecore/storage ``` -Programmatic entity import (P1): +Successful imports print a machine-readable JSON object: -```rust -use homecore_migrate::entity_registry; -use homecore::HomeCore; - -#[tokio::main] -async fn main() { - let storage_dir = std::path::Path::new("/home/user/.homeassistant/.storage"); - - // Load HA entities - let entries = entity_registry::load(storage_dir) - .expect("load entity registry"); - println!("Loaded {} entities", entries.len()); - - // Import into HOMECORE (P2 when EntityRegistry::import() lands) - let homecore = HomeCore::new(); - for entry in entries { - println!("Entity: {} ({})", entry.entity_id, entry.name); - } -} +```json +{"kind":"device_registry","imported":8,"warning_count":0,"warnings":[],"destination":"/home/user/.homecore/storage/core.device_registry"} ``` -Full migration (P2 onwards, via `--features recorder`): +`inspect`, `inspect-config-entries`, `inspect-secrets`, and +`inspect-automations` are read-only. + +## Destination files + +| Source | Destination | Format | +|---|---|---| +| `core.entity_registry` | `core.entity_registry` | HA-compatible v1/minor 13 envelope | +| `core.device_registry` | `core.device_registry` | HA-compatible v1/minor 13 envelope | +| `core.config_entries` | `homecore.config_entries` | HOMECORE v1/minor 0 envelope | + +Config entries are storage-compatible, not runtime-compatible: importing an +entry does not install or execute its HA integration. A HOMECORE plugin must +explicitly claim the domain and consume the preserved source payload. + +## Remaining limitations + +- Automation conversion is not implemented; the tool only inspects + `automations.yaml`. +- `!secret` reference resolution in other YAML files is not implemented. +- Deleted entity/device tombstones are not imported. +- Device fields newer than HA registry minor version 13 require an explicit + parser update; unknown versions fail closed. +- No side-by-side HA recorder database exporter is provided. + +## Validation ```bash -# Side-by-side: old HA continues running while HOMECORE reads the DB -homecore-migrate export-for-sidecar \ - --ha-dir ~/.homeassistant \ - --homecore-db ~/.homecore/home.db \ - --keep-automations true # Don't stop HA automations during test period +cargo test -p homecore-migrate +cargo clippy -p homecore-migrate --all-targets -- -D warnings ``` - -## Relation to other HOMECORE crates - -``` -homecore-migrate (import from HA) -├─ homecore (EntityEntry → EntityRegistry; config entry imports) -├─ homecore-automation (automations.yaml → automation rules, P2) -├─ homecore-recorder (side-by-side state export, P2, `--features recorder`) -├─ homecore-plugins (config_entries → plugin manifests, P2) -└─ homecore-server (can auto-import at startup with --import-ha flag, P2) -``` - -## References - -- [ADR-165: HOMECORE Migration from Python Home Assistant](../../docs/adr/ADR-165-homecore-migrate-from-home-assistant.md) -- [ADR-126: HOMECORE Home Assistant Port (master)](../../docs/adr/ADR-126-homecore-home-assistant-port.md) -- [Home Assistant .storage/ format](https://developers.home-assistant.io/docs/storage/) -- [homecore-migrate CLI source](src/main.rs) -- [README — wifi-densepose](../../../README.md) diff --git a/v2/crates/homecore-migrate/src/cli.rs b/v2/crates/homecore-migrate/src/cli.rs index 4a2fc7ba..30d73cdf 100644 --- a/v2/crates/homecore-migrate/src/cli.rs +++ b/v2/crates/homecore-migrate/src/cli.rs @@ -21,10 +21,12 @@ pub enum Command { Inspect(InspectArgs), /// Import entity registry from HA into a HOMECORE storage directory. ImportEntities(ImportEntitiesArgs), - /// Import device registry (P1: parses and reports; wiring to HOMECORE P2). + /// Import the device registry into HOMECORE storage. ImportDevices(ImportDevicesArgs), - /// Inspect config entries (P1: count + domain list; conversion is P2). + /// Inspect config entries without writing. InspectConfigEntries(InspectConfigEntriesArgs), + /// Import config entries losslessly into versioned HOMECORE storage. + ImportConfigEntries(ImportConfigEntriesArgs), /// Parse secrets.yaml and report secret names (values redacted). InspectSecrets(InspectSecretsArgs), /// Count and list automations from automations.yaml (conversion is P2). @@ -53,6 +55,19 @@ pub struct ImportDevicesArgs { /// Path to the HA `.storage/` directory. #[arg(long)] pub storage: PathBuf, + /// Path to the HOMECORE storage directory (destination). + #[arg(long)] + pub to: PathBuf, +} + +#[derive(Debug, clap::Args)] +pub struct ImportConfigEntriesArgs { + /// Path to the HA `.storage/` directory. + #[arg(long)] + pub storage: PathBuf, + /// Path to the HOMECORE storage directory (destination). + #[arg(long)] + pub to: PathBuf, } #[derive(Debug, clap::Args)] diff --git a/v2/crates/homecore-migrate/src/config_entries.rs b/v2/crates/homecore-migrate/src/config_entries.rs index 8144f685..855595e7 100644 --- a/v2/crates/homecore-migrate/src/config_entries.rs +++ b/v2/crates/homecore-migrate/src/config_entries.rs @@ -1,84 +1,236 @@ -//! Parser for `core.config_entries` (HA storage schema v1, minor_version varies). +//! Lossless conversion of HA `core.config_entries` into HOMECORE storage. //! -//! Per ADR-165 §6 Q5, `.storage/core.config_entries` format is undocumented -//! and version-gated. P1 reads the envelope and emits: -//! - count of config entries -//! - list of integration domains represented -//! -//! Conversion to HOMECORE plugin manifests is P2. -//! -//! Note: `config_entries` uses a different `minor_version` track from -//! `entity_registry`. As of HA 2025.1 it is typically minor_version=1 or 2. -//! We accept any minor_version ≤ MAX_SUPPORTED_MINOR and hard-error above it. +//! HOMECORE cannot yet execute arbitrary HA integrations. Every source row is +//! therefore retained verbatim while portable identity/config fields are +//! projected for future plugin setup. Typed warnings make the unsupported +//! surface machine-readable instead of silently dropping it. -use std::path::Path; +use std::collections::{BTreeMap, BTreeSet}; +use std::path::{Path, PathBuf}; use serde::{Deserialize, Serialize}; -use crate::{storage::read_envelope, MigrateError}; +use crate::{ + storage::{read_envelope, write_json_atomic_noclobber, HaStorageEnvelope}, + MigrateError, +}; -/// Maximum `minor_version` we claim to understand for config_entries. +const SOURCE_KEY: &str = "core.config_entries"; +pub const DESTINATION_KEY: &str = "homecore.config_entries"; const MAX_SUPPORTED_MINOR: u32 = 4; +const HOMECORE_CONFIG_VERSION: u32 = 1; -/// Diagnostic summary produced by P1 inspection. -#[derive(Clone, Debug, Serialize)] +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct HomeCoreConfigEntry { + pub entry_id: String, + pub domain: String, + pub title: String, + #[serde(default)] + pub data: serde_json::Value, + /// Exact HA row, including unsupported and future fields. + pub source: serde_json::Value, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "code", rename_all = "snake_case")] +pub enum MigrationWarning { + UnsupportedDomain { entry_id: String, domain: String }, + UnsupportedField { entry_id: String, field: String }, + UnsupportedRootField { field: String }, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct HomeCoreConfigData { + pub source_version: u32, + pub source_minor_version: u32, + /// Exact non-entry fields from the HA `data` object. + #[serde(default)] + pub source_extra: BTreeMap, + pub entries: Vec, + pub warnings: Vec, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct HomeCoreConfigEnvelope { + pub version: u32, + pub minor_version: u32, + pub key: String, + pub data: HomeCoreConfigData, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] pub struct ConfigEntriesSummary { pub count: usize, pub domains: Vec, + pub warning_count: usize, + pub destination: Option, } -/// Minimal fields we read from each config-entry row. -#[derive(Debug, Deserialize)] -struct HaConfigEntryRow { - domain: String, - #[allow(dead_code)] - entry_id: String, - /// Title shown in HA UI (informational only in P1). - #[serde(default)] - #[allow(dead_code)] - title: Option, - /// Source of the entry: "user" | "discovery" | "import" etc. - #[serde(default)] - #[allow(dead_code)] - source: Option, - /// State: "loaded" | "setup_error" etc. - #[serde(default)] - #[allow(dead_code)] - state: Option, -} - -#[derive(Debug, Deserialize)] -struct HaConfigEntriesData { - entries: Vec, -} - -/// Read `core.config_entries` from `path` and return a diagnostic summary. -pub fn inspect_config_entries(path: &Path) -> Result { - let env = read_envelope(path)?; - let file_str = path.display().to_string(); - - // config_entries has version=1 and minor_version in 1..MAX_SUPPORTED_MINOR. +fn validate_source(env: &HaStorageEnvelope, path: &Path) -> Result<(), MigrateError> { if env.version != 1 || env.minor_version > MAX_SUPPORTED_MINOR { return Err(MigrateError::UnsupportedSchemaVersion { - file: file_str.clone(), + file: path.display().to_string(), version: env.version, minor_version: env.minor_version, }); } + if env.key != SOURCE_KEY { + return Err(MigrateError::UnexpectedStorageKey { + path: path.display().to_string(), + expected: SOURCE_KEY.to_owned(), + actual: env.key.clone(), + }); + } + Ok(()) +} - let data: HaConfigEntriesData = - serde_json::from_value(env.data).map_err(|e| MigrateError::JsonParse { - path: file_str, - source: e, +pub fn convert_config_entries(path: &Path) -> Result { + let env = read_envelope(path)?; + validate_source(&env, path)?; + let entries = env + .data + .get("entries") + .and_then(serde_json::Value::as_array) + .ok_or_else(|| MigrateError::MissingField { + field: "entries".to_owned(), + context: path.display().to_string(), })?; + let source_extra: BTreeMap = env + .data + .as_object() + .into_iter() + .flat_map(|object| object.iter()) + .filter(|(field, _)| field.as_str() != "entries") + .map(|(field, value)| (field.clone(), value.clone())) + .collect(); - let mut domains: Vec = data.entries.iter().map(|e| e.domain.clone()).collect(); - domains.sort(); - domains.dedup(); + let portable = BTreeSet::from(["entry_id", "domain", "title", "data"]); + let mut converted = Vec::with_capacity(entries.len()); + let mut warnings = source_extra + .keys() + .map(|field| MigrationWarning::UnsupportedRootField { + field: field.clone(), + }) + .collect::>(); + for (index, source) in entries.iter().enumerate() { + let object = source + .as_object() + .ok_or_else(|| MigrateError::MissingField { + field: "object row".to_owned(), + context: format!("{} data.entries[{index}]", path.display()), + })?; + let string_field = |field: &str| -> Result { + object + .get(field) + .and_then(serde_json::Value::as_str) + .map(str::to_owned) + .ok_or_else(|| MigrateError::MissingField { + field: field.to_owned(), + context: format!("{} data.entries[{index}]", path.display()), + }) + }; + let entry_id = string_field("entry_id")?; + let domain = string_field("domain")?; + let title = object + .get("title") + .and_then(serde_json::Value::as_str) + .unwrap_or(&domain) + .to_owned(); + let data = object + .get("data") + .cloned() + .unwrap_or_else(|| serde_json::json!({})); + // No HA domain is implicitly claimed executable by HOMECORE. The + // source is retained so a matching plugin can consume it later. + warnings.push(MigrationWarning::UnsupportedDomain { + entry_id: entry_id.clone(), + domain: domain.clone(), + }); + for field in object + .keys() + .filter(|field| !portable.contains(field.as_str())) + { + warnings.push(MigrationWarning::UnsupportedField { + entry_id: entry_id.clone(), + field: field.clone(), + }); + } + converted.push(HomeCoreConfigEntry { + entry_id, + domain, + title, + data, + source: source.clone(), + }); + } + warnings.sort_by_key(|warning| serde_json::to_string(warning).unwrap_or_default()); + + Ok(HomeCoreConfigEnvelope { + version: HOMECORE_CONFIG_VERSION, + minor_version: 0, + key: DESTINATION_KEY.to_owned(), + data: HomeCoreConfigData { + source_version: env.version, + source_minor_version: env.minor_version, + source_extra, + entries: converted, + warnings, + }, + }) +} + +pub fn write_config_entries( + storage_dir: &Path, + envelope: &HomeCoreConfigEnvelope, +) -> Result { + let target = storage_dir.join(DESTINATION_KEY); + write_json_atomic_noclobber(&target, envelope) +} + +pub fn read_homecore_config_entries(path: &Path) -> Result { + let raw = std::fs::read_to_string(path).map_err(|source| MigrateError::Io { + path: path.display().to_string(), + source, + })?; + let envelope: HomeCoreConfigEnvelope = + serde_json::from_str(&raw).map_err(|source| MigrateError::JsonParse { + path: path.display().to_string(), + source, + })?; + if envelope.version != HOMECORE_CONFIG_VERSION || envelope.minor_version != 0 { + return Err(MigrateError::UnsupportedSchemaVersion { + file: path.display().to_string(), + version: envelope.version, + minor_version: envelope.minor_version, + }); + } + if envelope.key != DESTINATION_KEY { + return Err(MigrateError::UnexpectedStorageKey { + path: path.display().to_string(), + expected: DESTINATION_KEY.to_owned(), + actual: envelope.key, + }); + } + Ok(envelope) +} + +pub fn inspect_config_entries(path: &Path) -> Result { + let converted = convert_config_entries(path)?; + let domains: BTreeMap<&str, usize> = + converted + .data + .entries + .iter() + .fold(BTreeMap::new(), |mut map, entry| { + *map.entry(entry.domain.as_str()).or_default() += 1; + map + }); Ok(ConfigEntriesSummary { - count: data.entries.len(), - domains, + count: converted.data.entries.len(), + domains: domains.keys().map(|value| (*value).to_owned()).collect(), + warning_count: converted.data.warnings.len(), + destination: None, }) } @@ -89,40 +241,70 @@ mod tests { use tempfile::NamedTempFile; const FIXTURE: &str = r#"{ - "version": 1, - "minor_version": 1, - "key": "core.config_entries", - "data": { - "entries": [ - {"domain": "hue", "entry_id": "ce_001", "title": "Philips Hue", "source": "user", "state": "loaded"}, - {"domain": "zha", "entry_id": "ce_002", "title": "ZHA", "source": "user", "state": "loaded"}, - {"domain": "hue", "entry_id": "ce_003", "title": "Hue 2", "source": "user", "state": "setup_error"} - ] - } + "version":1,"minor_version":2,"key":"core.config_entries", + "data":{"future_root":{"kept":true},"entries":[{ + "domain":"future_hub","entry_id":"ce_001","title":"Future Hub", + "source":"user","state":"loaded","data":{"host":"10.0.0.2"}, + "options":{"scan":true},"future_field":{"nested":[1,2,3]} + }]} }"#; - #[test] - fn inspect_emits_count_and_domains() { - let mut f = NamedTempFile::new().unwrap(); - f.write_all(FIXTURE.as_bytes()).unwrap(); - let summary = inspect_config_entries(f.path()).unwrap(); - assert_eq!(summary.count, 3); - assert_eq!(summary.domains, vec!["hue", "zha"]); + fn fixture() -> NamedTempFile { + let mut file = NamedTempFile::new().unwrap(); + file.write_all(FIXTURE.as_bytes()).unwrap(); + file } #[test] - fn unknown_minor_version_hard_errors() { - let json = r#"{ - "version": 1, "minor_version": 99, - "key": "core.config_entries", - "data": {"entries": []} - }"#; - let mut f = NamedTempFile::new().unwrap(); - f.write_all(json.as_bytes()).unwrap(); - let err = inspect_config_entries(f.path()).unwrap_err(); + fn unknown_domain_and_fields_are_lossless_with_typed_warnings() { + let source = fixture(); + let converted = convert_config_entries(source.path()).unwrap(); + assert_eq!( + converted.data.entries[0].source["future_field"]["nested"][2], + 3 + ); + assert_eq!(converted.data.source_extra["future_root"]["kept"], true); + assert!(converted.data.warnings.iter().any(|warning| matches!( + warning, + MigrationWarning::UnsupportedDomain { domain, .. } if domain == "future_hub" + ))); + assert!(converted.data.warnings.iter().any(|warning| matches!( + warning, + MigrationWarning::UnsupportedField { field, .. } if field == "future_field" + ))); + + let dir = tempfile::tempdir().unwrap(); + let path = write_config_entries(dir.path(), &converted).unwrap(); + let restored = read_homecore_config_entries(&path).unwrap(); + assert_eq!(restored, converted); + } + + #[test] + fn unknown_destination_version_is_rejected() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(DESTINATION_KEY); + std::fs::write( + &path, + r#"{"version":99,"minor_version":0,"key":"homecore.config_entries","data":{"source_version":1,"source_minor_version":1,"entries":[],"warnings":[]}}"#, + ) + .unwrap(); assert!(matches!( - err, - MigrateError::UnsupportedSchemaVersion { minor_version: 99, .. } + read_homecore_config_entries(&path), + Err(MigrateError::UnsupportedSchemaVersion { version: 99, .. }) + )); + } + + #[test] + fn malformed_entry_is_an_error_not_a_partial_write() { + let mut source = NamedTempFile::new().unwrap(); + source + .write_all( + br#"{"version":1,"minor_version":1,"key":"core.config_entries","data":{"entries":[{"domain":"hue"}]}}"#, + ) + .unwrap(); + assert!(matches!( + convert_config_entries(source.path()), + Err(MigrateError::MissingField { .. }) )); } } diff --git a/v2/crates/homecore-migrate/src/device_registry.rs b/v2/crates/homecore-migrate/src/device_registry.rs index 39ccfd93..c2ff0b51 100644 --- a/v2/crates/homecore-migrate/src/device_registry.rs +++ b/v2/crates/homecore-migrate/src/device_registry.rs @@ -1,60 +1,131 @@ -//! Parser for `core.device_registry` (HA storage schema v1, minor_version 1–13). -//! -//! P1: deserializes the envelope and returns `Vec`. -//! HOMECORE's device registry isn't fully wired yet (ADR-127 §2.5 deferred -//! to P2), so `DeviceImport` is a staging type for the future hand-off. +//! Conversion for HA `core.device_registry` schema v1/minor 1-13. -use std::path::Path; +use std::collections::{BTreeMap, HashSet}; +use std::path::{Path, PathBuf}; -use serde::{Deserialize, Serialize}; +use homecore::DeviceEntry; +use serde::Deserialize; -use crate::{storage::read_envelope, storage_format::v13, MigrateError}; +use crate::{ + storage::{read_envelope, write_json_atomic_noclobber}, + storage_format::v13, + MigrateError, +}; -/// Staging type for a device imported from HA. Not yet wired to HOMECORE's -/// device registry (ADR-127 §2.5 — deferred to P2). -#[derive(Clone, Debug, Serialize, Deserialize)] -pub struct DeviceImport { - pub id: String, - pub config_entries: Vec, - #[serde(default)] - pub manufacturer: Option, - #[serde(default)] - pub model: Option, - #[serde(default)] - pub name: Option, - /// `identifiers` — list of `[integration, id]` pairs. Preserved as raw - /// JSON for P2 consumption; not yet mapped to HOMECORE DeviceEntry. - #[serde(default)] - pub identifiers: Vec>, - #[serde(default)] - pub connections: Vec>, - #[serde(default)] - pub via_device_id: Option, - #[serde(default)] - pub area_id: Option, -} +const FILE_KEY: &str = "core.device_registry"; #[derive(Debug, Deserialize)] struct HaDeviceRegistryData { - devices: Vec, - /// Deleted device tombstones — ignored in P1. + devices: Vec, #[serde(default)] - #[allow(dead_code)] deleted_devices: Vec, } -/// Read `core.device_registry` from `path` and return the raw import list. -pub fn read_device_registry(path: &Path) -> Result, MigrateError> { - let env = read_envelope(path)?; - let file_str = path.display().to_string(); - v13::require_supported(&file_str, env.version, env.minor_version)?; +#[derive(Debug, Deserialize)] +struct HaDeviceRow { + id: String, + #[serde(default)] + config_entries: HashSet, + #[serde(default)] + identifiers: HashSet<(String, String)>, + #[serde(default)] + connections: HashSet<(String, String)>, + #[serde(default)] + manufacturer: Option, + #[serde(default)] + model: Option, + #[serde(default)] + model_id: Option, + #[serde(default)] + name: Option, + #[serde(default)] + name_by_user: Option, + #[serde(default)] + sw_version: Option, + #[serde(default)] + hw_version: Option, + #[serde(default)] + serial_number: Option, + #[serde(default)] + via_device_id: Option, + #[serde(default)] + area_id: Option, + #[serde(default)] + entry_type: Option, + #[serde(default)] + disabled_by: Option, + #[serde(default)] + configuration_url: Option, + #[serde(default)] + labels: HashSet, + #[serde(default)] + primary_config_entry: Option, + #[serde(default, flatten)] + extra: BTreeMap, +} +impl From for DeviceEntry { + fn from(row: HaDeviceRow) -> Self { + Self { + id: row.id, + config_entries: row.config_entries, + identifiers: row.identifiers, + connections: row.connections, + manufacturer: row.manufacturer, + model: row.model, + model_id: row.model_id, + name: row.name, + name_by_user: row.name_by_user, + sw_version: row.sw_version, + hw_version: row.hw_version, + serial_number: row.serial_number, + via_device_id: row.via_device_id, + area_id: row.area_id, + entry_type: row.entry_type, + disabled_by: row.disabled_by, + configuration_url: row.configuration_url, + labels: row.labels, + primary_config_entry: row.primary_config_entry, + extra: row.extra, + } + } +} + +pub fn read_device_registry(path: &Path) -> Result, MigrateError> { + let env = read_envelope(path)?; + let file = path.display().to_string(); + v13::require_supported(&file, env.version, env.minor_version)?; + if env.key != FILE_KEY { + return Err(MigrateError::UnexpectedStorageKey { + path: file, + expected: FILE_KEY.to_owned(), + actual: env.key, + }); + } let data: HaDeviceRegistryData = - serde_json::from_value(env.data).map_err(|e| MigrateError::JsonParse { - path: file_str, - source: e, + serde_json::from_value(env.data).map_err(|source| MigrateError::JsonParse { + path: path.display().to_string(), + source, })?; - Ok(data.devices) + let _preserved_tombstone_count = data.deleted_devices.len(); + Ok(data.devices.into_iter().map(DeviceEntry::from).collect()) +} + +pub fn write_device_registry( + storage_dir: &Path, + devices: &[DeviceEntry], +) -> Result { + let target = storage_dir.join(FILE_KEY); + let payload = serde_json::json!({ + "version": 1, + "minor_version": 13, + "key": FILE_KEY, + "data": { + "devices": devices, + "deleted_devices": [] + } + }); + write_json_atomic_noclobber(&target, &payload) } #[cfg(test)] @@ -64,36 +135,49 @@ mod tests { use tempfile::NamedTempFile; const FIXTURE: &str = r#"{ - "version": 1, - "minor_version": 13, - "key": "core.device_registry", - "data": { - "devices": [ - { - "id": "dev_abc", - "config_entries": ["ce_001"], - "manufacturer": "Philips", - "model": "Hue Bridge", - "name": "Philips Hue Bridge", - "identifiers": [["hue", "001788FFFE3D4B13"]], - "connections": [["mac", "00:17:88:ff:fe:3d:4b:13"]], - "via_device_id": null, - "area_id": null - } - ], - "deleted_devices": [] - } + "version":1,"minor_version":13,"key":"core.device_registry", + "data":{"devices":[{ + "id":"dev_abc","config_entries":["ce_001"], + "manufacturer":"Philips","model":"Hue Bridge","model_id":"BSB002", + "name":"Hue","name_by_user":"Downstairs Hue", + "sw_version":"1.2","hw_version":"3","serial_number":"SN42", + "identifiers":[["hue","001788FFFE3D4B13"]], + "connections":[["mac","00:17:88:ff:fe:3d:4b:13"]], + "via_device_id":"gateway","area_id":"living_room", + "entry_type":"service","disabled_by":"user", + "configuration_url":"http://hue.local","labels":["lighting"], + "primary_config_entry":"ce_001","created_at":1735689600.0 + }],"deleted_devices":[]} }"#; #[test] - fn parses_device_registry() { - let mut f = NamedTempFile::new().unwrap(); - f.write_all(FIXTURE.as_bytes()).unwrap(); - let devices = read_device_registry(f.path()).unwrap(); - assert_eq!(devices.len(), 1); - let d = &devices[0]; - assert_eq!(d.id, "dev_abc"); - assert_eq!(d.manufacturer.as_deref(), Some("Philips")); - assert_eq!(d.identifiers, vec![vec!["hue", "001788FFFE3D4B13"]]); + fn all_supported_fields_round_trip() { + let mut source = NamedTempFile::new().unwrap(); + source.write_all(FIXTURE.as_bytes()).unwrap(); + let devices = read_device_registry(source.path()).unwrap(); + let destination = tempfile::tempdir().unwrap(); + let path = write_device_registry(destination.path(), &devices).unwrap(); + let imported = read_device_registry(&path).unwrap(); + assert_eq!(imported, devices); + assert_eq!(imported[0].serial_number.as_deref(), Some("SN42")); + assert!(imported[0].labels.contains("lighting")); + assert_eq!(imported[0].extra["created_at"], 1735689600.0); + } + + #[test] + fn destination_is_never_overwritten() { + let mut source = NamedTempFile::new().unwrap(); + source.write_all(FIXTURE.as_bytes()).unwrap(); + let devices = read_device_registry(source.path()).unwrap(); + let destination = tempfile::tempdir().unwrap(); + write_device_registry(destination.path(), &devices).unwrap(); + let error = write_device_registry(destination.path(), &[]).unwrap_err(); + assert!(error.to_string().contains("refusing to overwrite")); + assert_eq!( + read_device_registry(&destination.path().join(FILE_KEY)) + .unwrap() + .len(), + 1 + ); } } diff --git a/v2/crates/homecore-migrate/src/entity_registry.rs b/v2/crates/homecore-migrate/src/entity_registry.rs index ec5d61c6..f4658fbc 100644 --- a/v2/crates/homecore-migrate/src/entity_registry.rs +++ b/v2/crates/homecore-migrate/src/entity_registry.rs @@ -26,15 +26,17 @@ //! } //! ``` -use std::fs::{self, OpenOptions}; -use std::io::Write; use std::path::{Path, PathBuf}; use serde::{Deserialize, Serialize}; use homecore::{registry::DisabledBy, EntityCategory, EntityEntry, EntityId}; -use crate::{storage::read_envelope, storage_format::v13, MigrateError}; +use crate::{ + storage::{read_envelope, write_json_atomic_noclobber}, + storage_format::v13, + MigrateError, +}; // Key used by `inspect` subcommand when scanning the directory. #[allow(dead_code)] @@ -173,21 +175,7 @@ pub fn write_entity_registry( storage_dir: &Path, entries: &[EntityEntry], ) -> Result { - fs::create_dir_all(storage_dir).map_err(|source| MigrateError::Io { - path: storage_dir.display().to_string(), - source, - })?; let target = storage_dir.join(FILE_KEY); - if target.exists() { - return Err(MigrateError::Io { - path: target.display().to_string(), - source: std::io::Error::new( - std::io::ErrorKind::AlreadyExists, - "destination exists; refusing to overwrite", - ), - }); - } - let temp = storage_dir.join(format!(".{FILE_KEY}.{}.tmp", std::process::id())); let payload = serde_json::json!({ "version": 1, "minor_version": 13, @@ -197,29 +185,7 @@ pub fn write_entity_registry( "deleted_entities": [] } }); - let bytes = serde_json::to_vec_pretty(&payload).map_err(|source| MigrateError::JsonParse { - path: target.display().to_string(), - source, - })?; - - let result = (|| { - let mut file = OpenOptions::new() - .create_new(true) - .write(true) - .open(&temp)?; - file.write_all(&bytes)?; - file.write_all(b"\n")?; - file.sync_all()?; - fs::rename(&temp, &target) - })(); - if let Err(source) = result { - let _ = fs::remove_file(&temp); - return Err(MigrateError::Io { - path: target.display().to_string(), - source, - }); - } - Ok(target) + write_json_atomic_noclobber(&target, &payload) } #[cfg(test)] diff --git a/v2/crates/homecore-migrate/src/lib.rs b/v2/crates/homecore-migrate/src/lib.rs index 4eff1115..660f81b8 100644 --- a/v2/crates/homecore-migrate/src/lib.rs +++ b/v2/crates/homecore-migrate/src/lib.rs @@ -4,20 +4,21 @@ //! (HOMECORE-MIGRATE; ADR-126 §4 series map labels the role "ADR-134 HOMECORE-MIGRATE", //! but on-disk ADR-134 is CIR — the migrate decision was renumbered to ADR-165. See ADR-164). //! -//! ## P1 scope +//! ## Implemented scope //! //! - [`storage`] — `HaStorageDir`, `HaStorageEnvelope`; `read_envelope(path)` //! - [`storage_format`] — versioned format parsers (`v13`); unknown minor_version → hard error //! - [`entity_registry`] — `core.entity_registry` → `Vec` -//! - [`device_registry`] — `core.device_registry` → `Vec` (P1 stub) -//! - [`config_entries`] — `core.config_entries` diagnostic (count + domain list; P2 converts) +//! - [`device_registry`] — full supported HA v13 device fields → `homecore::DeviceEntry` +//! - [`config_entries`] — lossless, versioned HOMECORE representation + typed warnings //! - [`secrets`] — `secrets.yaml` → `HashMap` //! - [`automations`] — `automations.yaml` count + ID list (P2 converts) //! - [`cli`] — `clap`-derived subcommand types shared between `src/main.rs` and tests //! -//! ## What is NOT here yet (deferred to P2+) +//! ## Remaining limitations //! -//! - Conversion of `config_entries` to HOMECORE plugin manifests +//! - Imported config entries are durable but do not make an HA integration executable; +//! a matching HOMECORE plugin must consume the preserved source payload. //! - Conversion of `automations.yaml` to `homecore-automation` YAML //! - Side-by-side runtime mode (requires `homecore-recorder`, ADR-132) //! - `!secret` reference resolution in non-secrets YAML files @@ -88,6 +89,13 @@ pub enum MigrateError { minor_version: u32, }, + #[error("unexpected storage key in {path}: expected {expected}, got {actual}")] + UnexpectedStorageKey { + path: String, + expected: String, + actual: String, + }, + #[error("missing required field '{field}' in {context}")] MissingField { field: String, context: String }, diff --git a/v2/crates/homecore-migrate/src/main.rs b/v2/crates/homecore-migrate/src/main.rs index 95e2284b..b89097fa 100644 --- a/v2/crates/homecore-migrate/src/main.rs +++ b/v2/crates/homecore-migrate/src/main.rs @@ -2,6 +2,21 @@ use clap::Parser; use homecore_migrate::cli::{Cli, Command}; +use serde::Serialize; + +#[derive(Serialize)] +struct ImportSummary { + kind: &'static str, + imported: usize, + warning_count: usize, + warnings: Vec, + destination: std::path::PathBuf, +} + +fn print_summary(summary: &ImportSummary) -> anyhow::Result<()> { + println!("{}", serde_json::to_string(summary)?); + Ok(()) +} fn main() -> anyhow::Result<()> { tracing_subscriber::fmt::init(); @@ -9,7 +24,10 @@ fn main() -> anyhow::Result<()> { match cli.command { Command::Inspect(args) => { - println!("Inspecting HA .storage directory: {}", args.storage.display()); + println!( + "Inspecting HA .storage directory: {}", + args.storage.display() + ); // Probe entity_registry let entity_path = args.storage.join("core.entity_registry"); if entity_path.exists() { @@ -42,33 +60,35 @@ fn main() -> anyhow::Result<()> { Command::ImportEntities(args) => { let entity_path = args.storage.join("core.entity_registry"); - let entries = - homecore_migrate::entity_registry::read_entity_registry(&entity_path)?; + let entries = homecore_migrate::entity_registry::read_entity_registry(&entity_path)?; let destination = homecore_migrate::entity_registry::write_entity_registry(&args.to, &entries)?; - println!("Imported {} entity entries", entries.len()); - println!(" Destination: {}", destination.display()); - for e in &entries { - println!( - " {} ({}{})", - e.entity_id.as_str(), - e.platform, - if e.disabled_by.is_some() { " DISABLED" } else { "" } - ); - } + print_summary(&ImportSummary { + kind: "entity_registry", + imported: entries.len(), + warning_count: 0, + warnings: vec![], + destination, + })?; } Command::ImportDevices(args) => { let device_path = args.storage.join("core.device_registry"); - let devices = - homecore_migrate::device_registry::read_device_registry(&device_path)?; - println!("Parsed {} device entries (P1: staging only, wiring to HOMECORE is P2)", devices.len()); + let devices = homecore_migrate::device_registry::read_device_registry(&device_path)?; + let destination = + homecore_migrate::device_registry::write_device_registry(&args.to, &devices)?; + print_summary(&ImportSummary { + kind: "device_registry", + imported: devices.len(), + warning_count: 0, + warnings: vec![], + destination, + })?; } Command::InspectConfigEntries(args) => { let ce_path = args.storage.join("core.config_entries"); - let summary = - homecore_migrate::config_entries::inspect_config_entries(&ce_path)?; + let summary = homecore_migrate::config_entries::inspect_config_entries(&ce_path)?; println!( "config_entries: {} total, domains: {}", summary.count, @@ -76,6 +96,28 @@ fn main() -> anyhow::Result<()> { ); } + Command::ImportConfigEntries(args) => { + let source = args.storage.join("core.config_entries"); + let converted = homecore_migrate::config_entries::convert_config_entries(&source)?; + let imported = converted.data.entries.len(); + let warning_count = converted.data.warnings.len(); + let warnings = converted + .data + .warnings + .iter() + .map(serde_json::to_value) + .collect::, _>>()?; + let destination = + homecore_migrate::config_entries::write_config_entries(&args.to, &converted)?; + print_summary(&ImportSummary { + kind: "config_entries", + imported, + warning_count, + warnings, + destination, + })?; + } + Command::InspectSecrets(args) => { let secrets_path = args.config_dir.join("secrets.yaml"); let secrets = homecore_migrate::secrets::read_secrets(&secrets_path)?; diff --git a/v2/crates/homecore-migrate/src/storage.rs b/v2/crates/homecore-migrate/src/storage.rs index 770c8d65..d86a5bc1 100644 --- a/v2/crates/homecore-migrate/src/storage.rs +++ b/v2/crates/homecore-migrate/src/storage.rs @@ -15,12 +15,17 @@ //! left as `serde_json::Value` — version-specific parsers in `storage_format` //! are responsible for further deserialization. +use std::fs::{self, OpenOptions}; +use std::io::Write; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; use serde::{Deserialize, Serialize}; use crate::MigrateError; +static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0); + /// Points to a HA `.storage/` directory. #[derive(Clone, Debug)] pub struct HaStorageDir { @@ -66,6 +71,70 @@ pub fn read_envelope(path: &Path) -> Result { }) } +/// Durably publish JSON at `target` without ever replacing an existing file. +/// +/// Bytes are synced in a same-directory temporary file, then exposed with an +/// atomic hard-link create. `hard_link` fails with `AlreadyExists` if another +/// process won the destination race, unlike a POSIX rename which would replace +/// the destination after a check-then-rename sequence. +pub fn write_json_atomic_noclobber( + target: &Path, + value: &T, +) -> Result { + let parent = target.parent().ok_or_else(|| MigrateError::Io { + path: target.display().to_string(), + source: std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "destination has no parent directory", + ), + })?; + fs::create_dir_all(parent).map_err(|source| MigrateError::Io { + path: parent.display().to_string(), + source, + })?; + + let bytes = serde_json::to_vec_pretty(value).map_err(|source| MigrateError::JsonParse { + path: target.display().to_string(), + source, + })?; + let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let name = target + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or("storage"); + let temp = parent.join(format!(".{name}.{}.{}.tmp", std::process::id(), sequence)); + + let result = (|| -> std::io::Result<()> { + let mut file = OpenOptions::new() + .create_new(true) + .write(true) + .open(&temp)?; + file.write_all(&bytes)?; + file.write_all(b"\n")?; + file.sync_all()?; + fs::hard_link(&temp, target)?; + fs::remove_file(&temp)?; + Ok(()) + })(); + + if let Err(source) = result { + let _ = fs::remove_file(&temp); + let source = if source.kind() == std::io::ErrorKind::AlreadyExists { + std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + "destination exists; refusing to overwrite", + ) + } else { + source + }; + return Err(MigrateError::Io { + path: target.display().to_string(), + source, + }); + } + Ok(target.to_path_buf()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/v2/crates/homecore-recorder/README.md b/v2/crates/homecore-recorder/README.md index 093bb151..e31c8054 100644 --- a/v2/crates/homecore-recorder/README.md +++ b/v2/crates/homecore-recorder/README.md @@ -30,6 +30,8 @@ Data persists in `.homecore/home.db` (by default; configurable). Queries work vi - **Attribute persistence** — JSON attributes for entities stored in separate table (HA pattern) - **Automatic deduplication** — skip writes when state hasn't changed (detect via hash) - **Recorder runs table** — track purge cycles and migration events (HA `recorder_runs` equivalent) +- **Startup restoration** — deterministic newest row per entity, bounded at + 100,000 rows, with malformed rows isolated as typed warnings - **Semantic search** (P2, `--features ruvector`) — embed state attributes + query by meaning - **HNSW index** (P2) — k-NN search for "all warm rooms" via ruvector - **No data export overhead** — SQLite is queryable directly; no proprietary format @@ -41,6 +43,7 @@ Data persists in `.homecore/home.db` (by default; configurable). Queries work vi | Record state change | Listener | `RecorderListener::on_state_changed(event)` | Fires on homecore event bus; writes to SQLite | | Query state history | SQL | `SELECT * FROM states WHERE entity_id = ? ORDER BY last_changed DESC` | Standard SQLite; can be queried from anywhere | | Purge old states | Maintenance | `Recorder::purge(older_than)` | Deletes states older than specified timestamp | +| Restore latest states | Startup | `Recorder::restore_latest(states, limit)` | Entity-id ordered, bounded, malformed-row isolation | | Deduplicate write | Dedup | `DedupEngine::should_record(old_state, new_state)` | Skip if state hash unchanged | | Create semantic index | Index | `SemanticIndex::index_state(entity_id, state)` (P2, opt-in) | Hash-based embeddings; real embeddings in P3 | | Search by meaning | Search | `SemanticIndex::search(query, k)` (P2, opt-in) | "warm rooms" → k-NN search in ruvector HNSW | diff --git a/v2/crates/homecore-recorder/src/db.rs b/v2/crates/homecore-recorder/src/db.rs index 5611e6e1..741b33e5 100644 --- a/v2/crates/homecore-recorder/src/db.rs +++ b/v2/crates/homecore-recorder/src/db.rs @@ -20,7 +20,8 @@ use tokio::sync::RwLock; use tracing::debug; use homecore::entity::{EntityId, State}; -use homecore::event::{DomainEvent, StateChangedEvent}; +use homecore::event::{Context, DomainEvent, StateChangedEvent}; +use homecore::StateMachine; use crate::dedup::fnv64a_hash; use crate::schema::ALL_DDL; @@ -45,6 +46,8 @@ type HistoryStateRecord = (i64, String, Option, f64, f64, Option /// history-graph query, small enough to bound the worst case. Callers needing a /// wider span page by narrowing the window. pub const MAX_HISTORY_ROWS: i64 = 1_000_000; +/// Absolute cap for one startup restore query. +pub const MAX_RESTORE_STATES: usize = 100_000; /// Errors returned by `Recorder` operations. #[derive(Error, Debug)] @@ -302,7 +305,7 @@ impl Recorder { let pattern = format!("%{escaped}%"); let rows: Vec = sqlx::query_as( - "SELECT s.state_id, s.entity_id, s.state, sa.shared_attrs, \ + "SELECT s.state_id, s.entity_id, s.state, sa.shared_attrs, \ s.last_changed_ts, s.last_updated_ts, s.context_id \ FROM states s \ LEFT JOIN state_attributes sa ON s.attributes_id = sa.attributes_id \ @@ -312,47 +315,57 @@ impl Recorder { OR sa.shared_attrs LIKE ?2 ESCAPE '\\' \ ORDER BY s.last_updated_ts DESC \ LIMIT ?3", - ) - .bind(query) - .bind(&pattern) - .bind(k as i64) - .fetch_all(&self.pool) - .await?; + ) + .bind(query) + .bind(&pattern) + .bind(k as i64) + .fetch_all(&self.pool) + .await?; rows.into_iter() - .map(|(state_id, entity_id, state, shared_attrs, last_changed_ts, last_updated_ts, context_id)| { - let eid = EntityId::parse(&entity_id) - .unwrap_or_else(|_| EntityId::parse("unknown.unknown").unwrap()); - let attributes = shared_attrs - .as_deref() - .map(serde_json::from_str) - .transpose()? - .unwrap_or(serde_json::Value::Object(Default::default())); - Ok(StateRow { + .map( + |( state_id, - entity_id: eid, + entity_id, state, - attributes, + shared_attrs, last_changed_ts, last_updated_ts, context_id, - }) - }) + )| { + let eid = EntityId::parse(&entity_id) + .unwrap_or_else(|_| EntityId::parse("unknown.unknown").unwrap()); + let attributes = shared_attrs + .as_deref() + .map(serde_json::from_str) + .transpose()? + .unwrap_or(serde_json::Value::Object(Default::default())); + Ok(StateRow { + state_id, + entity_id: eid, + state, + attributes, + last_changed_ts, + last_updated_ts, + context_id, + }) + }, + ) .collect() } /// Fetch a single `StateRow` by its `state_id`, joining attributes. async fn fetch_state_row(&self, state_id: i64) -> Result, RecorderError> { let row: Option = sqlx::query_as( - "SELECT s.entity_id, s.state, sa.shared_attrs, \ + "SELECT s.entity_id, s.state, sa.shared_attrs, \ s.last_changed_ts, s.last_updated_ts, s.context_id \ FROM states s \ LEFT JOIN state_attributes sa ON s.attributes_id = sa.attributes_id \ WHERE s.state_id = ?", - ) - .bind(state_id) - .fetch_optional(&self.pool) - .await?; + ) + .bind(state_id) + .fetch_optional(&self.pool) + .await?; let Some((entity_id, state, shared_attrs, last_changed_ts, last_updated_ts, context_id)) = row @@ -438,26 +451,184 @@ impl Recorder { .await?; rows.into_iter() - .map(|(state_id, state, shared_attrs, last_changed_ts, last_updated_ts, context_id)| { - let attributes = shared_attrs - .as_deref() - .map(serde_json::from_str) - .transpose()? - .unwrap_or(serde_json::Value::Object(Default::default())); + .map( + |(state_id, state, shared_attrs, last_changed_ts, last_updated_ts, context_id)| { + let attributes = shared_attrs + .as_deref() + .map(serde_json::from_str) + .transpose()? + .unwrap_or(serde_json::Value::Object(Default::default())); - Ok(StateRow { - state_id, - entity_id: entity_id.clone(), - state, - attributes, - last_changed_ts, - last_updated_ts, - context_id, - }) - }) + Ok(StateRow { + state_id, + entity_id: entity_id.clone(), + state, + attributes, + last_changed_ts, + last_updated_ts, + context_id, + }) + }, + ) .collect() } + /// Read the newest row for each entity in deterministic entity-id order. + /// + /// The query is bounded and uses `(last_updated_ts, state_id)` as a stable + /// newest-row tie-break. Malformed rows are reported and skipped + /// individually so one corrupt entity cannot prevent the rest from + /// starting. + pub async fn latest_states( + &self, + requested_limit: usize, + ) -> Result { + let limit = requested_limit.min(MAX_RESTORE_STATES); + if limit == 0 { + return Ok(LatestStates::default()); + } + type RawRestoreRow = ( + i64, + String, + Option, + Option, + Option, + Option, + Option, + ); + let rows: Vec = sqlx::query_as( + "SELECT state_id, entity_id, state, shared_attrs, \ + last_changed_ts, last_updated_ts, context_id \ + FROM ( \ + SELECT s.state_id, s.entity_id, s.state, sa.shared_attrs, \ + s.last_changed_ts, s.last_updated_ts, s.context_id, \ + ROW_NUMBER() OVER ( \ + PARTITION BY s.entity_id \ + ORDER BY s.last_updated_ts DESC, s.state_id DESC \ + ) AS newest \ + FROM states s \ + LEFT JOIN state_attributes sa ON s.attributes_id = sa.attributes_id \ + ) \ + WHERE newest = 1 \ + ORDER BY entity_id ASC \ + LIMIT ?", + ) + .bind((limit + 1) as i64) + .fetch_all(&self.pool) + .await?; + + let truncated = rows.len() > limit; + let mut states = Vec::with_capacity(rows.len().min(limit)); + let mut warnings = Vec::new(); + for ( + state_id, + raw_entity_id, + raw_state, + raw_attributes, + last_changed_ts, + last_updated_ts, + context_id, + ) in rows.into_iter().take(limit) + { + let entity_id = match EntityId::parse(&raw_entity_id) { + Ok(value) => value, + Err(error) => { + warnings.push(RestoreWarning::MalformedEntityId { + state_id, + entity_id: raw_entity_id, + reason: error.to_string(), + }); + continue; + } + }; + let Some(state) = raw_state else { + warnings.push(RestoreWarning::MissingState { + state_id, + entity_id: raw_entity_id, + }); + continue; + }; + let attributes = match raw_attributes { + Some(value) => match serde_json::from_str(&value) { + Ok(value) => value, + Err(error) => { + warnings.push(RestoreWarning::MalformedAttributes { + state_id, + entity_id: raw_entity_id, + reason: error.to_string(), + }); + continue; + } + }, + None => serde_json::json!({}), + }; + let Some(last_changed) = last_changed_ts.and_then(timestamp_from_seconds) else { + warnings.push(RestoreWarning::MalformedTimestamp { + state_id, + entity_id: raw_entity_id, + field: "last_changed_ts", + }); + continue; + }; + let Some(last_updated) = last_updated_ts.and_then(timestamp_from_seconds) else { + warnings.push(RestoreWarning::MalformedTimestamp { + state_id, + entity_id: raw_entity_id, + field: "last_updated_ts", + }); + continue; + }; + let parent_id = match context_id { + Some(value) => match value.parse() { + Ok(value) => Some(value), + Err(_) => { + warnings.push(RestoreWarning::MalformedContext { + state_id, + entity_id: raw_entity_id.clone(), + }); + None + } + }, + None => None, + }; + states.push(State { + entity_id, + state, + attributes, + last_changed, + last_updated, + context: Context::restoration(parent_id), + }); + } + Ok(LatestStates { + states, + warnings, + truncated, + }) + } + + /// Load latest durable snapshots into a state machine without producing + /// fresh recorder events. + pub async fn restore_latest( + &self, + states: &StateMachine, + limit: usize, + ) -> Result { + let batch = self.latest_states(limit).await?; + let mut restored = 0; + for state in batch.states { + // latest_states constructs the required restoration marker. + if states.restore(state).is_ok() { + restored += 1; + } + } + Ok(RestoreReport { + restored, + warnings: batch.warnings, + truncated: batch.truncated, + }) + } + /// Purge history older than `older_than`, returning a [`PurgeStats`] summary. /// /// Deletes: @@ -521,6 +692,58 @@ impl Recorder { } } +fn timestamp_from_seconds(value: f64) -> Option> { + if !value.is_finite() { + return None; + } + let micros = (value * 1_000_000.0).round(); + if micros < i64::MIN as f64 || micros > i64::MAX as f64 { + return None; + } + DateTime::from_timestamp_micros(micros as i64) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RestoreWarning { + MalformedEntityId { + state_id: i64, + entity_id: String, + reason: String, + }, + MissingState { + state_id: i64, + entity_id: String, + }, + MalformedAttributes { + state_id: i64, + entity_id: String, + reason: String, + }, + MalformedTimestamp { + state_id: i64, + entity_id: String, + field: &'static str, + }, + MalformedContext { + state_id: i64, + entity_id: String, + }, +} + +#[derive(Debug, Default)] +pub struct LatestStates { + pub states: Vec, + pub warnings: Vec, + pub truncated: bool, +} + +#[derive(Debug)] +pub struct RestoreReport { + pub restored: usize, + pub warnings: Vec, + pub truncated: bool, +} + /// Summary of a [`Recorder::purge`] run. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct PurgeStats { @@ -564,14 +787,20 @@ mod tests { use super::*; async fn open_memory() -> Recorder { - Recorder::open("sqlite::memory:").await.expect("open in-memory DB") + Recorder::open("sqlite::memory:") + .await + .expect("open in-memory DB") } fn entity(s: &str) -> EntityId { EntityId::parse(s).unwrap() } - fn make_state_event(entity_id: &str, state_val: &str, attrs: serde_json::Value) -> StateChangedEvent { + fn make_state_event( + entity_id: &str, + state_val: &str, + attrs: serde_json::Value, + ) -> StateChangedEvent { let eid = entity(entity_id); let ctx = Context::new(); let s = Arc::new(State::new(eid.clone(), state_val, attrs, ctx)); @@ -595,7 +824,10 @@ mod tests { .await .unwrap(); let names: Vec<&str> = tables.iter().map(|(n,)| n.as_str()).collect(); - assert!(names.contains(&"state_attributes"), "missing state_attributes"); + assert!( + names.contains(&"state_attributes"), + "missing state_attributes" + ); assert!(names.contains(&"states"), "missing states"); assert!(names.contains(&"events"), "missing events"); assert!(names.contains(&"recorder_runs"), "missing recorder_runs"); @@ -605,7 +837,10 @@ mod tests { async fn schema_idempotent_double_open() { // Applying schema twice (on the same pool) must not panic or error. let recorder = open_memory().await; - recorder.apply_schema().await.expect("second apply_schema must be a no-op"); + recorder + .apply_schema() + .await + .expect("second apply_schema must be a no-op"); } // ── record_state ────────────────────────────────────────────────────────── @@ -613,7 +848,11 @@ mod tests { #[tokio::test] async fn record_state_inserts_row() { let recorder = open_memory().await; - let event = make_state_event("light.kitchen", "on", serde_json::json!({"brightness": 200})); + let event = make_state_event( + "light.kitchen", + "on", + serde_json::json!({"brightness": 200}), + ); let state_id = recorder.record_state(&event).await.unwrap(); assert!(state_id.is_some(), "expected a state_id"); @@ -652,19 +891,20 @@ mod tests { recorder.record_state(&e1).await.unwrap(); recorder.record_state(&e2).await.unwrap(); - let attr_count: (i64,) = - sqlx::query_as("SELECT COUNT(*) FROM state_attributes") - .fetch_one(&recorder.pool) - .await - .unwrap(); + let attr_count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM state_attributes") + .fetch_one(&recorder.pool) + .await + .unwrap(); // Both events share identical attrs → only one state_attributes row. - assert_eq!(attr_count.0, 1, "identical attrs must share one state_attributes row"); + assert_eq!( + attr_count.0, 1, + "identical attrs must share one state_attributes row" + ); - let state_count: (i64,) = - sqlx::query_as("SELECT COUNT(*) FROM states") - .fetch_one(&recorder.pool) - .await - .unwrap(); + let state_count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM states") + .fetch_one(&recorder.pool) + .await + .unwrap(); assert_eq!(state_count.0, 2, "two states rows expected"); } @@ -677,11 +917,10 @@ mod tests { recorder.record_state(&e1).await.unwrap(); recorder.record_state(&e2).await.unwrap(); - let attr_count: (i64,) = - sqlx::query_as("SELECT COUNT(*) FROM state_attributes") - .fetch_one(&recorder.pool) - .await - .unwrap(); + let attr_count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM state_attributes") + .fetch_one(&recorder.pool) + .await + .unwrap(); assert_eq!(attr_count.0, 2); } @@ -701,7 +940,10 @@ mod tests { let since = Utc::now() - chrono::Duration::seconds(10); let until = Utc::now() + chrono::Duration::seconds(10); - let rows = recorder.get_state_history(&eid, since, until).await.unwrap(); + let rows = recorder + .get_state_history(&eid, since, until) + .await + .unwrap(); assert_eq!(rows.len(), 3, "expected 3 history rows"); // Verify ascending order by last_updated_ts. @@ -749,11 +991,19 @@ mod tests { // FAILS against the old always-empty path: asserts real rows come back. let recorder = open_memory().await; recorder - .record_state(&make_state_event("light.kitchen", "on", serde_json::json!({}))) + .record_state(&make_state_event( + "light.kitchen", + "on", + serde_json::json!({}), + )) .await .unwrap(); recorder - .record_state(&make_state_event("light.bedroom", "off", serde_json::json!({}))) + .record_state(&make_state_event( + "light.bedroom", + "off", + serde_json::json!({}), + )) .await .unwrap(); recorder @@ -789,7 +1039,10 @@ mod tests { )) .await .unwrap(); - let rows = recorder.search_states_by_text("portland", 10).await.unwrap(); + let rows = recorder + .search_states_by_text("portland", 10) + .await + .unwrap(); assert_eq!(rows.len(), 1); assert_eq!(rows[0].entity_id.as_str(), "sensor.weather"); assert_eq!(rows[0].attributes["location"], "portland"); @@ -816,7 +1069,11 @@ mod tests { async fn text_search_no_match_returns_empty() { let recorder = open_memory().await; recorder - .record_state(&make_state_event("light.kitchen", "on", serde_json::json!({}))) + .record_state(&make_state_event( + "light.kitchen", + "on", + serde_json::json!({}), + )) .await .unwrap(); let rows = recorder @@ -839,7 +1096,11 @@ mod tests { // EntityId::parse permits this, so it reaches the bind path as data. let evil = "light.x_drop_table_states_select"; recorder - .record_state(&make_state_event(evil, "'; DROP TABLE states; --", serde_json::json!({}))) + .record_state(&make_state_event( + evil, + "'; DROP TABLE states; --", + serde_json::json!({}), + )) .await .unwrap(); @@ -897,7 +1158,11 @@ mod tests { let recorder = open_memory().await; for v in &["1", "2", "3", "4", "5"] { recorder - .record_state(&make_state_event("sensor.bounded", v, serde_json::json!({}))) + .record_state(&make_state_event( + "sensor.bounded", + v, + serde_json::json!({}), + )) .await .unwrap(); tokio::time::sleep(std::time::Duration::from_millis(2)).await; @@ -914,12 +1179,20 @@ mod tests { .fetch_all(&recorder.pool) .await .unwrap(); - assert_eq!(capped.len(), 2, "LIMIT term effectively bounds the result set"); + assert_eq!( + capped.len(), + 2, + "LIMIT term effectively bounds the result set" + ); // And the real method returns all rows when under the cap. let eid = entity("sensor.bounded"); let rows = recorder - .get_state_history(&eid, Utc::now() - chrono::Duration::seconds(10), Utc::now() + chrono::Duration::seconds(10)) + .get_state_history( + &eid, + Utc::now() - chrono::Duration::seconds(10), + Utc::now() + chrono::Duration::seconds(10), + ) .await .unwrap(); assert_eq!(rows.len(), 5, "all rows under the cap return"); @@ -947,7 +1220,10 @@ mod tests { // Read back the actual timestamps so the cutoff is exact. let since = Utc::now() - chrono::Duration::seconds(60); let until = Utc::now() + chrono::Duration::seconds(60); - let all = recorder.get_state_history(&eid, since, until).await.unwrap(); + let all = recorder + .get_state_history(&eid, since, until) + .await + .unwrap(); assert_eq!(all.len(), 3); // Cut off exactly at the middle row's timestamp. let mid_ts = all[1].last_updated_ts; @@ -956,8 +1232,15 @@ mod tests { let stats = recorder.purge(cutoff).await.unwrap(); assert_eq!(stats.states_deleted, 1, "only the strictly-older 'old' row"); - let remaining = recorder.get_state_history(&eid, since, until).await.unwrap(); - assert_eq!(remaining.len(), 2, "boundary 'mid' row is KEPT (exclusive cutoff)"); + let remaining = recorder + .get_state_history(&eid, since, until) + .await + .unwrap(); + assert_eq!( + remaining.len(), + 2, + "boundary 'mid' row is KEPT (exclusive cutoff)" + ); assert_eq!(remaining[0].state, "mid"); assert_eq!(remaining[1].state, "new"); } @@ -995,18 +1278,28 @@ mod tests { // referenced by sensor.b, so it must survive. let eid_b = entity("sensor.b"); let rows_b = recorder - .get_state_history(&eid_b, Utc::now() - chrono::Duration::seconds(60), Utc::now() + chrono::Duration::seconds(60)) + .get_state_history( + &eid_b, + Utc::now() - chrono::Duration::seconds(60), + Utc::now() + chrono::Duration::seconds(60), + ) .await .unwrap(); let b_ts = rows_b[0].last_updated_ts; let cutoff = DateTime::::from_timestamp_micros((b_ts * 1_000_000.0) as i64).unwrap(); let stats = recorder.purge(cutoff).await.unwrap(); assert_eq!(stats.states_deleted, 1, "sensor.a purged"); - assert_eq!(stats.attributes_deleted, 0, "shared blob still referenced — kept"); + assert_eq!( + stats.attributes_deleted, 0, + "shared blob still referenced — kept" + ); assert_eq!(attr_count(&recorder).await, 1, "blob survives"); // Now purge everything → sensor.b gone, blob orphaned → GC'd. - let stats2 = recorder.purge(Utc::now() + chrono::Duration::seconds(120)).await.unwrap(); + let stats2 = recorder + .purge(Utc::now() + chrono::Duration::seconds(120)) + .await + .unwrap(); assert_eq!(stats2.states_deleted, 1, "sensor.b purged"); assert_eq!(stats2.attributes_deleted, 1, "now-orphaned blob GC'd"); assert_eq!(attr_count(&recorder).await, 0, "no blobs remain"); @@ -1017,7 +1310,11 @@ mod tests { let recorder = open_memory().await; let ctx = Context::new(); recorder - .record_event(&DomainEvent::new("call_service", serde_json::json!({}), ctx)) + .record_event(&DomainEvent::new( + "call_service", + serde_json::json!({}), + ctx, + )) .await .unwrap(); // Purge with a far-future cutoff removes the event. @@ -1039,11 +1336,95 @@ mod tests { // real rows via the text fallback — proving it's no longer always-empty. let recorder = open_memory().await; recorder - .record_state(&make_state_event("light.kitchen", "on", serde_json::json!({}))) + .record_state(&make_state_event( + "light.kitchen", + "on", + serde_json::json!({}), + )) .await .unwrap(); let rows = recorder.search_semantic("kitchen", 5).await.unwrap(); assert_eq!(rows.len(), 1, "fallback must surface the kitchen row"); assert_eq!(rows[0].entity_id.as_str(), "light.kitchen"); } + + #[tokio::test] + async fn latest_states_is_deterministic_newest_per_entity() { + let recorder = open_memory().await; + recorder + .record_state(&make_state_event("sensor.z", "old", serde_json::json!({}))) + .await + .unwrap(); + recorder + .record_state(&make_state_event("sensor.a", "only", serde_json::json!({}))) + .await + .unwrap(); + recorder + .record_state(&make_state_event("sensor.z", "new", serde_json::json!({}))) + .await + .unwrap(); + + let batch = recorder.latest_states(10).await.unwrap(); + assert_eq!(batch.states.len(), 2); + assert_eq!(batch.states[0].entity_id.as_str(), "sensor.a"); + assert_eq!(batch.states[1].entity_id.as_str(), "sensor.z"); + assert_eq!(batch.states[1].state, "new"); + assert!(batch + .states + .iter() + .all(|state| state.context.is_restoration())); + } + + #[tokio::test] + async fn latest_states_isolates_malformed_rows_and_honours_bound() { + let recorder = open_memory().await; + recorder + .record_state(&make_state_event( + "sensor.good", + "ok", + serde_json::json!({"x": 1}), + )) + .await + .unwrap(); + sqlx::query( + "INSERT INTO states \ + (entity_id, state, attributes_id, last_changed_ts, last_updated_ts, context_id) \ + VALUES ('INVALID', 'bad', NULL, 1.0, 2.0, NULL)", + ) + .execute(&recorder.pool) + .await + .unwrap(); + let attrs_id = sqlx::query( + "INSERT INTO state_attributes (shared_attrs, hash) VALUES ('not-json', 424242)", + ) + .execute(&recorder.pool) + .await + .unwrap() + .last_insert_rowid(); + sqlx::query( + "INSERT INTO states \ + (entity_id, state, attributes_id, last_changed_ts, last_updated_ts, context_id) \ + VALUES ('sensor.badattrs', 'bad', ?, 1.0, 2.0, NULL)", + ) + .bind(attrs_id) + .execute(&recorder.pool) + .await + .unwrap(); + + let batch = recorder.latest_states(10).await.unwrap(); + assert_eq!(batch.states.len(), 1); + assert_eq!(batch.warnings.len(), 2); + assert!(batch + .warnings + .iter() + .any(|warning| matches!(warning, RestoreWarning::MalformedEntityId { .. }))); + assert!(batch + .warnings + .iter() + .any(|warning| matches!(warning, RestoreWarning::MalformedAttributes { .. }))); + + let bounded = recorder.latest_states(1).await.unwrap(); + assert!(bounded.truncated); + assert!(bounded.states.len() <= 1); + } } diff --git a/v2/crates/homecore-recorder/src/lib.rs b/v2/crates/homecore-recorder/src/lib.rs index 9391607f..81280f90 100644 --- a/v2/crates/homecore-recorder/src/lib.rs +++ b/v2/crates/homecore-recorder/src/lib.rs @@ -30,7 +30,10 @@ pub mod schema; pub mod semantic; // Re-export the primary public API surface. -pub use db::{PurgeStats, Recorder, RecorderError, SemanticIndex, StateRow, MAX_HISTORY_ROWS}; +pub use db::{ + LatestStates, PurgeStats, Recorder, RecorderError, RestoreReport, RestoreWarning, + SemanticIndex, StateRow, MAX_HISTORY_ROWS, MAX_RESTORE_STATES, +}; pub use listener::RecorderListener; /// Null semantic index used when the `ruvector` feature is off. diff --git a/v2/crates/homecore-server/Cargo.toml b/v2/crates/homecore-server/Cargo.toml index b47db9c0..5a483acd 100644 --- a/v2/crates/homecore-server/Cargo.toml +++ b/v2/crates/homecore-server/Cargo.toml @@ -25,9 +25,10 @@ homecore = { path = "../homecore", version = "0.1.0-alpha.0" homecore-api = { path = "../homecore-api", version = "0.1.0-alpha.0" } homecore-plugins = { path = "../homecore-plugins", version = "0.1.0-alpha.0", optional = true } homecore-recorder = { path = "../homecore-recorder", version = "0.1.0-alpha.0" } +# Reuse version-gated registry envelope parsing in the isolated restore module. +homecore-migrate = { path = "../homecore-migrate", version = "0.1.0-alpha.0" } homecore-automation = { path = "../homecore-automation", version = "0.1.0-alpha.0" } homecore-assist = { path = "../homecore-assist", version = "0.1.0-alpha.0" } -# Migration crate is CLI-only; not linked here. tokio = { version = "1", features = ["full"] } tracing = "0.1" @@ -57,6 +58,7 @@ futures = "0.3" # Drive the assembled router in integration tests via ServiceExt::oneshot. tower = { version = "0.5", features = ["util"] } http-body-util = "0.1" +tempfile = "3" [features] default = [] diff --git a/v2/crates/homecore-server/README.md b/v2/crates/homecore-server/README.md index 9784658d..3c62d359 100644 --- a/v2/crates/homecore-server/README.md +++ b/v2/crates/homecore-server/README.md @@ -26,6 +26,11 @@ a default token. Configure allowed browser origins with ## Runtime behavior - SQLite recording is enabled by default at `sqlite://homecore.db`. +- Entity/device registries are restored from `.homecore/storage` before + recorder states, listeners, automations, and API startup. +- The latest recorder state for each entity is restored deterministically. + Restored snapshots retain their timestamps and carry a `homecore.restore` + context marker; malformed rows are logged and isolated. - Synthetic entities are disabled by default; opt in with `--seed-demo-entities`. - Automations can be loaded with `--automations ` or @@ -67,6 +72,8 @@ unregistered and return an error rather than a false acknowledgement. |---|---|---| | `--bind` | `HOMECORE_BIND` | `0.0.0.0:8123` | | `--db` | `HOMECORE_DB` | `sqlite://homecore.db` | +| `--storage-dir` | `HOMECORE_STORAGE_DIR` | `.homecore/storage` | +| `--restore-limit` | `HOMECORE_RESTORE_LIMIT` | `100000` | | `--location-name` | `HOMECORE_LOCATION` | `Home` | | `--automations` | `HOMECORE_AUTOMATIONS` | unset | | `--insecure-dev-auth` | `HOMECORE_INSECURE_DEV_AUTH` | `false` | diff --git a/v2/crates/homecore-server/src/main.rs b/v2/crates/homecore-server/src/main.rs index 305dc3ae..2396a4f3 100644 --- a/v2/crates/homecore-server/src/main.rs +++ b/v2/crates/homecore-server/src/main.rs @@ -37,6 +37,7 @@ use tower_http::services::ServeDir; use tower_http::trace::TraceLayer; mod gateway; +mod restore; use gateway::{GatewayConfig, GatewayState}; /// Compile-time default location of the HOMECORE-UI assets (ADR-131). @@ -83,6 +84,18 @@ struct Cli { #[arg(long, env = "HOMECORE_DB", default_value = "sqlite://homecore.db")] db: String, + /// HOMECORE registry storage directory restored at startup. + #[arg( + long, + env = "HOMECORE_STORAGE_DIR", + default_value = ".homecore/storage" + )] + storage_dir: std::path::PathBuf, + + /// Maximum registry rows and latest entity states restored at startup. + #[arg(long, env = "HOMECORE_RESTORE_LIMIT", default_value_t = 100_000)] + restore_limit: usize, + /// Friendly location name surfaced via `/api/config`. #[arg(long, env = "HOMECORE_LOCATION", default_value = "Home")] location_name: String, @@ -140,6 +153,30 @@ async fn main() -> Result<()> { let hc = HomeCore::new(); info!("HomeCore state machine + event bus + service registry online"); + let recorder = if cli.no_recorder { + None + } else { + match open_recorder(&cli.db).await { + Ok(recorder) => Some(recorder), + Err(error) => { + warn!("Recorder failed to open ({error}) — continuing without persistence"); + None + } + } + }; + let restored = + restore::restore_startup(&hc, recorder.as_ref(), &cli.storage_dir, cli.restore_limit).await; + info!( + entities = restored.entity_entries, + devices = restored.device_entries, + states = restored.states, + truncated = restored.truncated, + "Startup restoration complete" + ); + for warning in restored.warnings { + warn!("{warning}"); + } + // Seed a representative set of built-in services so the web UI // and HA-wire-compat clients see a populated /api/services on // first boot. These are no-op handlers (they just echo back the @@ -158,21 +195,14 @@ async fn main() -> Result<()> { } // ── 2. Recorder (optional) ────────────────────────────────────── - if !cli.no_recorder { - match open_recorder(&cli.db).await { - Ok(recorder) => { - let _recorder_task = RecorderListener::new(hc.states(), recorder).spawn(); - info!( - "Recorder open at {} — state_changed events being persisted", - cli.db - ); - } - Err(e) => { - warn!("Recorder failed to open ({e}) — continuing without persistence"); - } - } + if let Some(recorder) = recorder { + let _recorder_task = RecorderListener::new(hc.states(), recorder).spawn(); + info!( + "Recorder open at {} — state_changed events being persisted", + cli.db + ); } else { - info!("Recorder disabled by --no-recorder"); + info!("Recorder unavailable or disabled"); } // ── 3. Plugin runtime ─────────────────────────────────────────── diff --git a/v2/crates/homecore-server/src/restore.rs b/v2/crates/homecore-server/src/restore.rs new file mode 100644 index 00000000..6a4acdc0 --- /dev/null +++ b/v2/crates/homecore-server/src/restore.rs @@ -0,0 +1,219 @@ +//! Bounded startup restoration for registries and latest recorder states. + +use std::path::Path; + +use homecore::{DeviceEntry, EntityEntry, HomeCore}; +use homecore_migrate::storage::read_envelope; +use homecore_recorder::{Recorder, RestoreWarning}; +use serde::de::DeserializeOwned; + +pub const MAX_REGISTRY_ENTRIES: usize = 100_000; +pub const MAX_REGISTRY_FILE_BYTES: u64 = 64 * 1024 * 1024; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RestorePhase { + EntityRegistry, + DeviceRegistry, + States, +} + +#[derive(Debug, Default)] +pub struct StartupRestoreReport { + pub entity_entries: usize, + pub device_entries: usize, + pub states: usize, + pub warnings: Vec, + pub truncated: bool, + pub phases: Vec, +} + +pub async fn restore_startup( + homecore: &HomeCore, + recorder: Option<&Recorder>, + storage_dir: &Path, + limit: usize, +) -> StartupRestoreReport { + let limit = limit.min(MAX_REGISTRY_ENTRIES); + let mut report = StartupRestoreReport::default(); + + report.phases.push(RestorePhase::EntityRegistry); + let entities_path = storage_dir.join("core.entity_registry"); + if entities_path.exists() { + match load_rows::(&entities_path, "entities", limit, |entry| { + entry.entity_id.as_str().to_owned() + }) { + Ok(batch) => { + report.truncated |= batch.truncated; + report.warnings.extend(batch.warnings); + for entry in batch.rows { + homecore.entities().register(entry).await; + report.entity_entries += 1; + } + } + Err(error) => report.warnings.push(error), + } + } + + report.phases.push(RestorePhase::DeviceRegistry); + let devices_path = storage_dir.join("core.device_registry"); + if devices_path.exists() { + match load_rows::(&devices_path, "devices", limit, |entry| entry.id.clone()) { + Ok(batch) => { + report.truncated |= batch.truncated; + report.warnings.extend(batch.warnings); + for entry in batch.rows { + homecore.devices().register(entry).await; + report.device_entries += 1; + } + } + Err(error) => report.warnings.push(error), + } + } + + report.phases.push(RestorePhase::States); + if let Some(recorder) = recorder { + match recorder.restore_latest(homecore.states(), limit).await { + Ok(state_report) => { + report.states = state_report.restored; + report.truncated |= state_report.truncated; + report + .warnings + .extend(state_report.warnings.into_iter().map(format_state_warning)); + } + Err(error) => report + .warnings + .push(format!("state restore failed: {error}")), + } + } + report +} + +struct RowBatch { + rows: Vec, + warnings: Vec, + truncated: bool, +} + +fn load_rows( + path: &Path, + field: &str, + limit: usize, + key: impl Fn(&T) -> String, +) -> Result, String> { + let file_size = std::fs::metadata(path) + .map_err(|error| format!("{}: {error}", path.display()))? + .len(); + if file_size > MAX_REGISTRY_FILE_BYTES { + return Err(format!( + "{}: registry file is {file_size} bytes, exceeding the {} byte startup bound", + path.display(), + MAX_REGISTRY_FILE_BYTES + )); + } + let envelope = read_envelope(path).map_err(|error| error.to_string())?; + if envelope.version != 1 || envelope.minor_version > 13 { + return Err(format!( + "{}: unsupported registry version {}.{}", + path.display(), + envelope.version, + envelope.minor_version + )); + } + let values = envelope + .data + .get(field) + .and_then(serde_json::Value::as_array) + .ok_or_else(|| format!("{}: missing data.{field} array", path.display()))?; + let scan_limit = limit.saturating_add(1).min(MAX_REGISTRY_ENTRIES + 1); + let mut rows = Vec::with_capacity(values.len().min(scan_limit)); + let mut warnings = Vec::new(); + for (index, value) in values.iter().take(scan_limit).enumerate() { + match serde_json::from_value::(value.clone()) { + Ok(row) => rows.push(row), + Err(error) => warnings.push(format!( + "{}: isolated malformed data.{field}[{index}]: {error}", + path.display() + )), + } + } + rows.sort_by_key(&key); + let truncated = values.len() > limit || rows.len() > limit; + rows.truncate(limit); + Ok(RowBatch { + rows, + warnings, + truncated, + }) +} + +fn format_state_warning(warning: RestoreWarning) -> String { + format!("isolated malformed recorder row: {warning:?}") +} + +#[cfg(test)] +mod tests { + use super::*; + use homecore::{Context, EntityId}; + use homecore_recorder::Recorder; + + #[tokio::test] + async fn restores_registries_before_states_and_isolates_bad_rows() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write( + dir.path().join("core.entity_registry"), + r#"{"version":1,"minor_version":13,"key":"core.entity_registry","data":{"entities":[ + {"entity_id":"sensor.good","unique_id":null,"platform":"test","name":null,"disabled_by":null,"area_id":null,"device_id":"dev1","entity_category":null,"config_entry_id":null}, + {"entity_id":"INVALID","platform":"bad"} + ]}}"#, + ) + .unwrap(); + std::fs::write( + dir.path().join("core.device_registry"), + r#"{"version":1,"minor_version":13,"key":"core.device_registry","data":{"devices":[ + {"id":"dev1","config_entries":[],"identifiers":[],"connections":[],"manufacturer":null,"model":null,"model_id":null,"name":"Device","name_by_user":null,"sw_version":null,"hw_version":null,"serial_number":null,"via_device_id":null,"area_id":null,"entry_type":null,"disabled_by":null,"configuration_url":null,"labels":[],"primary_config_entry":null} + ]}}"#, + ) + .unwrap(); + let recorder = Recorder::open("sqlite::memory:").await.unwrap(); + let source = HomeCore::new(); + source.states().set( + EntityId::parse("sensor.good").unwrap(), + "on", + serde_json::json!({"restorable": true}), + Context::new(), + ); + let mut receiver = source.states().subscribe(); + // Force one recorder row from a real state event. + source.states().set( + EntityId::parse("sensor.good").unwrap(), + "off", + serde_json::json!({"restorable": true}), + Context::new(), + ); + recorder + .record_state(&receiver.recv().await.unwrap()) + .await + .unwrap(); + + let target = HomeCore::new(); + let report = + restore_startup(&target, Some(&recorder), dir.path(), MAX_REGISTRY_ENTRIES).await; + assert_eq!( + report.phases, + vec![ + RestorePhase::EntityRegistry, + RestorePhase::DeviceRegistry, + RestorePhase::States + ] + ); + assert_eq!(report.entity_entries, 1); + assert_eq!(report.device_entries, 1); + assert_eq!(report.states, 1); + assert!(!report.warnings.is_empty()); + let state = target + .states() + .get(&EntityId::parse("sensor.good").unwrap()) + .unwrap(); + assert!(state.context.is_restoration()); + } +} diff --git a/v2/crates/homecore/src/event.rs b/v2/crates/homecore/src/event.rs index 8caa3232..824a8caa 100644 --- a/v2/crates/homecore/src/event.rs +++ b/v2/crates/homecore/src/event.rs @@ -47,6 +47,9 @@ pub struct Context { } impl Context { + /// Marker stored on snapshots loaded from durable state at startup. + pub const RESTORE_USER_ID: &'static str = "homecore.restore"; + pub fn new() -> Self { Self::default() } @@ -66,6 +69,20 @@ impl Context { parent_id: Some(parent.id), } } + + /// Create a fresh context that identifies a startup restoration. The + /// persisted context, when valid, is retained as the causal parent. + pub fn restoration(parent_id: Option) -> Self { + Self { + id: Uuid::new_v4(), + user_id: Some(Self::RESTORE_USER_ID.to_owned()), + parent_id, + } + } + + pub fn is_restoration(&self) -> bool { + self.user_id.as_deref() == Some(Self::RESTORE_USER_ID) + } } impl Default for Context { diff --git a/v2/crates/homecore/src/homecore.rs b/v2/crates/homecore/src/homecore.rs index 41410ed6..5f242ae1 100644 --- a/v2/crates/homecore/src/homecore.rs +++ b/v2/crates/homecore/src/homecore.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use crate::bus::EventBus; -use crate::registry::EntityRegistry; +use crate::registry::{DeviceRegistry, EntityRegistry}; use crate::service::ServiceRegistry; use crate::state::StateMachine; @@ -20,6 +20,7 @@ struct HomeCoreInner { pub states: StateMachine, pub services: ServiceRegistry, pub entities: EntityRegistry, + pub devices: DeviceRegistry, } impl HomeCore { @@ -31,6 +32,7 @@ impl HomeCore { services: ServiceRegistry::with_event_bus(bus.clone()), bus, entities: EntityRegistry::new(), + devices: DeviceRegistry::new(), }), } } @@ -50,6 +52,10 @@ impl HomeCore { pub fn entities(&self) -> &EntityRegistry { &self.inner.entities } + + pub fn devices(&self) -> &DeviceRegistry { + &self.inner.devices + } } impl Default for HomeCore { diff --git a/v2/crates/homecore/src/lib.rs b/v2/crates/homecore/src/lib.rs index ef0b7e39..7f5243ab 100644 --- a/v2/crates/homecore/src/lib.rs +++ b/v2/crates/homecore/src/lib.rs @@ -11,7 +11,7 @@ //! - [`state`] — `StateMachine`: DashMap-backed concurrent state store //! - [`bus`] — `EventBus`: tokio broadcast wiring for system + domain events //! - [`service`] — `ServiceRegistry` (stub; full mpsc dispatch lands in P2) -//! - [`registry`] — `EntityRegistry` (in-memory P1; persistence lands in P2) +//! - [`registry`] — in-memory entity and device registries, restored by the server //! - [`homecore`] — `HomeCore` runtime coordinator: holds bus + states + services //! //! ## Threading model @@ -23,31 +23,30 @@ //! //! ## What's NOT here yet (deferred to P2+) //! -//! - Persistence of entity registry to `.homecore/storage/core.entity_registry` +//! - Automatic persistence of registry mutations (startup restoration exists) //! - Schema validation (`schemas` module from §3 stub) //! - Service handler mpsc dispatch (`service::ServiceRegistry::call`) -//! - Device registry (mirror of HA's `core.device_registry`) //! - Witness chain integration (ADR-028) //! //! Each is marked `// TODO P2:` at the relevant call site. +pub mod bus; pub mod entity; pub mod event; -pub mod state; -pub mod bus; -pub mod service; pub mod registry; +pub mod service; +pub mod state; mod homecore; pub use homecore::HomeCore; +pub use bus::EventBus; pub use entity::{EntityId, EntityIdError, State}; pub use event::{Context, DomainEvent, EventType, StateChangedEvent, SystemEvent}; -pub use state::StateMachine; -pub use bus::EventBus; +pub use registry::{DeviceEntry, DeviceRegistry, EntityCategory, EntityEntry, EntityRegistry}; pub use service::{ServiceCall, ServiceError, ServiceName, ServiceRegistry}; -pub use registry::{EntityCategory, EntityEntry, EntityRegistry}; +pub use state::StateMachine; /// HOMECORE protocol/data-model version. Bumped when the public surface /// or on-disk persistence schema changes in a backwards-incompatible way. diff --git a/v2/crates/homecore/src/registry.rs b/v2/crates/homecore/src/registry.rs index a5ffd053..d21dcfc0 100644 --- a/v2/crates/homecore/src/registry.rs +++ b/v2/crates/homecore/src/registry.rs @@ -1,9 +1,9 @@ -//! In-memory entity registry (P1). Persistence to -//! `.homecore/storage/core.entity_registry` lands in P2. +//! In-memory entity and device registries. Durable files are loaded by +//! `homecore-server` during bounded startup restoration. //! //! Schema fields mirror HA `core.entity_registry` v13 per ADR-127 §2.4. -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::Arc; use serde::{Deserialize, Serialize}; @@ -43,6 +43,41 @@ pub struct EntityEntry { pub config_entry_id: Option, } +/// Physical-device metadata persisted in `core.device_registry`. +/// +/// The fields track the HA v13 registry surface used by HOMECORE. Identifier +/// and connection pairs are sets because their order is not semantically +/// meaningful in HA. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct DeviceEntry { + pub id: String, + #[serde(default)] + pub config_entries: HashSet, + #[serde(default)] + pub identifiers: HashSet<(String, String)>, + #[serde(default)] + pub connections: HashSet<(String, String)>, + pub manufacturer: Option, + pub model: Option, + pub model_id: Option, + pub name: Option, + pub name_by_user: Option, + pub sw_version: Option, + pub hw_version: Option, + pub serial_number: Option, + pub via_device_id: Option, + pub area_id: Option, + pub entry_type: Option, + pub disabled_by: Option, + pub configuration_url: Option, + #[serde(default)] + pub labels: HashSet, + pub primary_config_entry: Option, + /// Forward-compatible device fields from newer HA v13-compatible rows. + #[serde(default, flatten)] + pub extra: BTreeMap, +} + #[derive(Clone)] pub struct EntityRegistry { entries: Arc>>, @@ -89,6 +124,45 @@ impl Default for EntityRegistry { } } +#[derive(Clone)] +pub struct DeviceRegistry { + entries: Arc>>, +} + +impl DeviceRegistry { + pub fn new() -> Self { + Self { + entries: Arc::new(RwLock::new(HashMap::new())), + } + } + + pub async fn register(&self, entry: DeviceEntry) { + self.entries.write().await.insert(entry.id.clone(), entry); + } + + pub async fn get(&self, id: &str) -> Option { + self.entries.read().await.get(id).cloned() + } + + pub async fn all(&self) -> Vec { + self.entries.read().await.values().cloned().collect() + } + + pub async fn len(&self) -> usize { + self.entries.read().await.len() + } + + pub async fn is_empty(&self) -> bool { + self.entries.read().await.is_empty() + } +} + +impl Default for DeviceRegistry { + fn default() -> Self { + Self::new() + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/v2/crates/homecore/src/state.rs b/v2/crates/homecore/src/state.rs index 44dca460..c1f4d553 100644 --- a/v2/crates/homecore/src/state.rs +++ b/v2/crates/homecore/src/state.rs @@ -14,7 +14,6 @@ //! //! - `async_set_internal` schema validation //! - Bulk delete of an entire domain (`async_remove_domain`) -//! - Restore-state on startup from the recorder (ADR-132) use std::sync::Arc; @@ -158,6 +157,19 @@ impl StateMachine { next } + /// Install a durable snapshot during startup without emitting a new + /// state-change event. Callers must mark the snapshot context as a + /// restoration; this prevents accidental use as a silent runtime write. + pub fn restore(&self, snapshot: State) -> Result, RestoreStateError> { + if !snapshot.context.is_restoration() { + return Err(RestoreStateError::UnmarkedContext); + } + let entity_id = snapshot.entity_id.clone(); + let snapshot = Arc::new(snapshot); + self.inner.states.insert(entity_id, Arc::clone(&snapshot)); + Ok(snapshot) + } + /// Remove a state. Fires `state_changed` with `new_state = None`. pub fn remove(&self, entity_id: &EntityId) -> Option> { let removed = self.inner.states.remove(entity_id).map(|(_, s)| s); @@ -213,6 +225,12 @@ impl Default for StateMachine { } } +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub enum RestoreStateError { + #[error("restored state context is not marked as a restoration")] + UnmarkedContext, +} + #[cfg(test)] mod tests { use super::*;