feat(homecore-api): serve bounded recorder history

v2/crates/homecore-api/src/app.rs

@@ -42,6 +42,11 @@ pub fn router(state: SharedState) -> Router {
         .route("/api/template", post(rest::render_template))
         .route("/api/config/core/check_config", post(rest::check_config))
         .route("/api/error_log", get(rest::error_log))
+        .route("/api/history/period", get(rest::get_history))
+        .route(
+            "/api/history/period/:start_time",
+            get(rest::get_history_period),
+        )
         .route("/api/homecore/compatibility", get(rest::compatibility))
         .route("/api/websocket", get(ws::websocket_handler))
         .layer(cors)

v2/crates/homecore-api/src/error.rs

@@ -14,6 +14,8 @@ pub enum ApiError {
     Unauthorized,
     #[error("service not registered: {domain}.{service}")]
     ServiceNotRegistered { domain: String, service: String },
+    #[error("service unavailable: {0}")]
+    Unavailable(String),
     #[error("internal error: {0}")]
     Internal(String),
 }
@@ -21,7 +23,9 @@ pub enum ApiError {
 pub type ApiResult<T> = Result<T, ApiError>;
 
 #[derive(Serialize)]
-struct ErrorPayload { message: String }
+struct ErrorPayload {
+    message: String,
+}
 
 impl IntoResponse for ApiError {
     fn into_response(self) -> Response {
@@ -30,6 +34,7 @@ impl IntoResponse for ApiError {
             Self::BadRequest(_) => (StatusCode::BAD_REQUEST, self.to_string()),
             Self::Unauthorized => (StatusCode::UNAUTHORIZED, self.to_string()),
             Self::ServiceNotRegistered { .. } => (StatusCode::BAD_REQUEST, self.to_string()),
+            Self::Unavailable(_) => (StatusCode::SERVICE_UNAVAILABLE, self.to_string()),
             Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, self.to_string()),
         };
         (status, Json(ErrorPayload { message })).into_response()

v2/crates/homecore-api/src/rest.rs

@@ -1,4 +1,4 @@
-use axum::extract::{Path, State};
+use axum::extract::{Path, Query, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::IntoResponse;
 use axum::Json;
@@ -92,6 +92,141 @@ pub struct StateView {
     pub context: ContextView,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct HistoryQuery {
+    filter_entity_id: Option<String>,
+    end_time: Option<String>,
+    #[serde(default)]
+    minimal_response: bool,
+    #[serde(default)]
+    no_attributes: bool,
+    #[serde(default)]
+    significant_changes_only: bool,
+}
+
+const MAX_HISTORY_ENTITIES: usize = 32;
+
+pub async fn get_history(
+    headers: HeaderMap,
+    State(s): State<SharedState>,
+    Query(query): Query<HistoryQuery>,
+) -> ApiResult<Json<Vec<Vec<StateView>>>> {
+    history_response(headers, s, None, query).await
+}
+
+pub async fn get_history_period(
+    headers: HeaderMap,
+    State(s): State<SharedState>,
+    Path(start_time): Path<String>,
+    Query(query): Query<HistoryQuery>,
+) -> ApiResult<Json<Vec<Vec<StateView>>>> {
+    history_response(headers, s, Some(start_time), query).await
+}
+
+async fn history_response(
+    headers: HeaderMap,
+    state: SharedState,
+    start_time: Option<String>,
+    query: HistoryQuery,
+) -> ApiResult<Json<Vec<Vec<StateView>>>> {
+    let _ = BearerAuth::from_headers(&headers, state.tokens()).await?;
+    let recorder = state
+        .recorder()
+        .ok_or_else(|| ApiError::Unavailable("recorder is disabled".into()))?;
+    let now = chrono::Utc::now();
+    let start = match start_time {
+        Some(value) => parse_history_time(&value)?,
+        None => now - chrono::Duration::days(1),
+    };
+    let end = match query.end_time.as_deref() {
+        Some(value) => parse_history_time(value)?,
+        None => now,
+    };
+    if end < start {
+        return Err(ApiError::BadRequest(
+            "end_time must not precede start_time".into(),
+        ));
+    }
+
+    let entity_ids = match query.filter_entity_id.as_deref() {
+        Some(raw) => raw
+            .split(',')
+            .map(str::trim)
+            .filter(|value| !value.is_empty())
+            .map(|value| {
+                EntityId::parse(value)
+                    .map_err(|error| ApiError::BadRequest(format!("invalid entity_id: {error}")))
+            })
+            .collect::<ApiResult<Vec<_>>>()?,
+        None => state
+            .homecore()
+            .states()
+            .all()
+            .into_iter()
+            .map(|snapshot| snapshot.entity_id.clone())
+            .collect(),
+    };
+    if entity_ids.len() > MAX_HISTORY_ENTITIES {
+        return Err(ApiError::BadRequest(format!(
+            "history queries are limited to {MAX_HISTORY_ENTITIES} entities"
+        )));
+    }
+
+    let mut result = Vec::with_capacity(entity_ids.len());
+    for entity_id in entity_ids {
+        let rows = recorder
+            .get_state_history(&entity_id, start, end)
+            .await
+            .map_err(|error| ApiError::Internal(format!("history query failed: {error}")))?;
+        let mut previous_state: Option<String> = None;
+        let states = rows
+            .into_iter()
+            .filter_map(|row| {
+                if query.significant_changes_only
+                    && previous_state.as_deref() == Some(row.state.as_str())
+                {
+                    return None;
+                }
+                previous_state = Some(row.state.clone());
+                let changed = history_timestamp(row.last_changed_ts);
+                let updated = history_timestamp(row.last_updated_ts);
+                Some(StateView {
+                    entity_id: row.entity_id.as_str().to_owned(),
+                    state: row.state,
+                    attributes: if query.no_attributes || query.minimal_response {
+                        serde_json::json!({})
+                    } else {
+                        row.attributes
+                    },
+                    last_changed: changed,
+                    last_updated: updated,
+                    context: ContextView {
+                        id: row.context_id.unwrap_or_default(),
+                        user_id: None,
+                        parent_id: None,
+                    },
+                })
+            })
+            .collect();
+        result.push(states);
+    }
+    Ok(Json(result))
+}
+
+fn parse_history_time(value: &str) -> ApiResult<chrono::DateTime<chrono::Utc>> {
+    chrono::DateTime::parse_from_rfc3339(value)
+        .map(|value| value.with_timezone(&chrono::Utc))
+        .map_err(|_| ApiError::BadRequest("history timestamps must be RFC 3339".into()))
+}
+
+fn history_timestamp(seconds: f64) -> String {
+    let whole = seconds.floor() as i64;
+    let nanos = ((seconds - seconds.floor()) * 1_000_000_000.0).round() as u32;
+    chrono::DateTime::<chrono::Utc>::from_timestamp(whole, nanos.min(999_999_999))
+        .unwrap_or(chrono::DateTime::<chrono::Utc>::UNIX_EPOCH)
+        .to_rfc3339()
+}
+
 #[derive(Serialize)]
 pub struct ContextView {
     pub id: String,
@@ -373,7 +508,7 @@ pub async fn compatibility(
             "template": "implemented",
             "check_config": "implemented",
             "error_log": "implemented",
-            "history": "requires_recorder",
+            "history": "implemented_when_recorder_enabled",
             "camera_calendar_media": "integration_dependent"
         },
         "websocket": {

v2/crates/homecore-api/src/state.rs

@@ -1,4 +1,5 @@
 use homecore::HomeCore;
+use homecore_recorder::Recorder;
 use std::sync::Arc;
 
 use crate::tokens::LongLivedTokenStore;
@@ -13,6 +14,7 @@ struct SharedStateInner {
     pub homecore_version: String,
     pub location_name: String,
     pub tokens: LongLivedTokenStore,
+    pub recorder: Option<Recorder>,
 }
 
 impl SharedState {
@@ -50,6 +52,19 @@ impl SharedState {
                 homecore_version: homecore_version.into(),
                 location_name: location_name.into(),
                 tokens,
+                recorder: None,
+            }),
+        }
+    }
+
+    pub fn with_recorder(self, recorder: Option<Recorder>) -> Self {
+        Self {
+            inner: Arc::new(SharedStateInner {
+                homecore: self.inner.homecore.clone(),
+                homecore_version: self.inner.homecore_version.clone(),
+                location_name: self.inner.location_name.clone(),
+                tokens: self.inner.tokens.clone(),
+                recorder,
             }),
         }
     }
@@ -66,4 +81,7 @@ impl SharedState {
     pub fn tokens(&self) -> &LongLivedTokenStore {
         &self.inner.tokens
     }
+    pub fn recorder(&self) -> Option<&Recorder> {
+        self.inner.recorder.as_ref()
+    }
 }