From 818079d1f4545f7445832588ca034e2d0af131c1 Mon Sep 17 00:00:00 2001 From: Valentin Haudiquet Date: Sat, 24 Oct 2026 01:52:09 +0200 Subject: [PATCH] =?UTF-8?q?refactor(claw):=20the=20calendar=20table=20is?= =?UTF-8?q?=20the=20one=20truth=20=E2=80=94=20live,=20not=20config?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same move as the watches, and the config's last piece of integration state goes with it: - `rhema-calendar-integration` owns the calendar table now (`store.rs`): SQLite in the shared dispatch database, the pattern the claw's watch table established. `CalendarWatch` gains the `validate` the config's boot check used to run — per-kind URL and credential rules, plus the lead-under-horizon check — so an invalid calendar is loud at the one door every writer shares. - `CalendarSource::from_store` reads the table fresh on every detect pass (the forge source's pattern): a calendar added over the route is live next tick, a removed one stops polling, no restart in the story. The static constructor stays for tests. - `POST/DELETE/GET /v1/calendars` on the daemon: the POST body is the integration's `CalendarWatch` shape, validated at the table; DELETE takes `?name=`; the api listing names the door. - `ClawConfig.calendars` is gone. A config still carrying it is refused at parse time, like the retired `watches:` seed. The config-side validation and its tests moved with the type. The config file now carries only what boot needs: credentials, the surface, paths, the poll cadence. Everything the daemon reconfigures at runtime lives in its tables, behind its HTTP doors. --- crates/rhema-claw/src/config.rs | 360 ------------------ crates/rhema-claw/src/daemon.rs | 9 +- crates/rhema-claw/src/extension.rs | 18 +- crates/rhema-claw/src/http.rs | 81 +++- crates/rhema-claw/src/interactive.rs | 7 +- crates/rhema-claw/tests/continuity.rs | 1 - crates/rhema-claw/tests/e2e.rs | 1 - crates/rhema-claw/tests/http.rs | 84 +++- .../rhema-calendar-integration/Cargo.toml | 1 + .../rhema-calendar-integration/src/lib.rs | 2 + .../src/reminder.rs | 126 +++++- .../rhema-calendar-integration/src/store.rs | 168 ++++++++ 12 files changed, 481 insertions(+), 377 deletions(-) create mode 100644 integrations/rhema-calendar-integration/src/store.rs diff --git a/crates/rhema-claw/src/config.rs b/crates/rhema-claw/src/config.rs index 738ab61..b172ebf 100644 --- a/crates/rhema-claw/src/config.rs +++ b/crates/rhema-claw/src/config.rs @@ -14,12 +14,6 @@ use std::path::PathBuf; use serde::{Deserialize, Serialize}; -// The calendar feed types are the integration's: the reminder source -// polls them there, and the claw's config only describes them. -pub use rhema_calendar_integration::reminder::{ - CalendarKind, CalendarWatch, DEFAULT_HORIZON_HOURS, DEFAULT_LEAD_MINUTES, -}; - /// How often the resident loop runs a detect pass, when /// [`ClawConfig::poll_interval_secs`] is unset. pub const DEFAULT_POLL_INTERVAL_SECS: u64 = 60; @@ -69,17 +63,6 @@ pub enum ConfigError { "ingest_token is the same value as another credential; the ingest token must be its own secret, write-only on the event door" )] IngestTokenCollision, - /// One calendar watch is unusable; the daemon refuses to start - /// rather than poll a half-configured calendar. - #[error("calendar #{index} ({name}): {reason}")] - Calendar { - /// The calendar's position in the config file, zero-based. - index: usize, - /// The calendar's name as written. - name: String, - /// What is missing or malformed. - reason: String, - }, /// The HTTP surface would listen beyond loopback with no bearer /// token: anyone who can reach the port could run tools on this /// machine. @@ -261,10 +244,6 @@ pub struct ClawConfig { /// Unset closes the door to credentials; the human token can still /// ingest. pub ingest_token: Option, - /// The calendars to remind about. Config-file-only: a calendar is - /// read-only ambient state, not an agent-managed resource, and a - /// change takes the daemon restart it takes to edit the file. - pub calendars: Vec, } impl ClawConfig { @@ -292,7 +271,6 @@ impl Default for ClawConfig { tools_bin: None, ingest_token: None, transcript: None, - calendars: Vec::new(), } } } @@ -434,111 +412,6 @@ impl ClawConfig { return Err(ConfigError::IngestTokenCollision); } } - self.validate_calendars()?; - Ok(()) - } - - /// The calendar half of the startup checks: every calendar named, - /// complete for its kind, with leads a horizon can see. - fn validate_calendars(&self) -> Result<(), ConfigError> { - let mut seen = std::collections::HashSet::new(); - for (index, calendar) in self.calendars.iter().enumerate() { - let fail = |reason: String| ConfigError::Calendar { - index, - name: calendar.name.clone(), - reason, - }; - if calendar.name.trim().is_empty() { - return Err(fail("name must not be blank".to_owned())); - } - if !seen.insert(calendar.name.trim().to_owned()) { - return Err(fail( - "duplicate name: reminder keys would collide".to_owned(), - )); - } - match calendar.kind { - CalendarKind::Ics => { - let url = calendar - .url - .as_deref() - .map(str::trim) - .filter(|url| !url.is_empty()) - .ok_or_else(|| { - fail("ics is a feed URL: set url to the calendar's address".to_owned()) - })?; - if !(url.starts_with("https://") - || url.starts_with("http://") - || url.starts_with("webcal://")) - { - return Err(fail(format!( - "url must be https://, http://, or webcal://, got {url:?}" - ))); - } - } - CalendarKind::Caldav => { - let url = calendar - .url - .as_deref() - .map(str::trim) - .filter(|url| !url.is_empty()) - .ok_or_else(|| { - fail( - "caldav is a collection: set url to the calendar's DAV \ - address" - .to_owned(), - ) - })?; - if !(url.starts_with("https://") || url.starts_with("http://")) { - return Err(fail(format!( - "url must be https:// or http://, got {url:?}" - ))); - } - let has_user = calendar - .username - .as_deref() - .map(str::trim) - .is_some_and(|user| !user.is_empty()); - if has_user && calendar.resolved_password().is_none() { - return Err(fail( - "no password: set password in the config or CALDAV_PASSWORD".to_owned(), - )); - } - } - CalendarKind::Google => { - if calendar - .calendar_id - .as_deref() - .map(str::trim) - .filter(|id| !id.is_empty()) - .is_none() - { - return Err(fail( - "google needs a calendar: set calendar_id (\"primary\", an \ - address, a hex id)" - .to_owned(), - )); - } - if calendar.resolved_api_key().is_none() && calendar.resolved_token().is_none() - { - return Err(fail( - "no credential: set api_key (public calendars) or token in the \ - config, or CALENDAR_TOKEN" - .to_owned(), - )); - } - } - } - // The horizon is what the poll can see; a lead at or past - // it fires on an event that was never visible. - let horizon_minutes = u64::from(calendar.horizon()) * 60; - let longest = calendar.leads().into_iter().max().unwrap_or_default(); - if longest >= horizon_minutes { - return Err(fail(format!( - "lead {longest}m is not under the {horizon_minutes}m horizon: a \ - reminder cannot fire before its event is visible" - ))); - } - } Ok(()) } } @@ -687,239 +560,6 @@ ingest_token: door-key assert_eq!(config.listen_addr(), DEFAULT_LISTEN); assert_eq!(config.token, None); } - - #[test] - fn a_calendar_section_parses_field_for_field() { - let raw = "calendars: - - name: home - kind: ics - url: https://cloud.example.net/remote.php/dav/calendars/personal.ics - lead_minutes: [60, 15] - horizon_hours: 48 - - name: work - kind: google - calendar_id: primary - api_key: pub-key - - name: shared - kind: google - calendar_id: abc123@group.calendar.google.com - token: oauth-token - - name: dav - kind: caldav - url: https://cloud.example.net/remote.php/dav/calendars/me/personal/ - username: me - password: app-pass -"; - let config: ClawConfig = serde_yaml::from_str(raw).expect("parses"); - assert_eq!(config.calendars.len(), 4); - let home = &config.calendars[0]; - assert_eq!(home.name, "home"); - assert_eq!(home.kind, CalendarKind::Ics); - assert_eq!(home.leads(), vec![60, 15]); - assert_eq!(home.horizon(), 48); - assert_eq!( - config.calendars[1].resolved_api_key().as_deref(), - Some("pub-key") - ); - assert_eq!( - config.calendars[2].resolved_token().as_deref(), - Some("oauth-token") - ); - assert_eq!(config.calendars[3].kind, CalendarKind::Caldav); - assert_eq!(config.calendars[3].username.as_deref(), Some("me")); - assert_eq!( - config.calendars[3].resolved_password().as_deref(), - Some("app-pass") - ); - config.validate().expect("a complete list validates"); - } - - #[test] - fn calendar_defaults_fill_what_the_file_omits() { - let config: ClawConfig = serde_yaml::from_str( - "calendars:\n - name: home\n kind: ics\n url: https://c.example.net/a.ics\n", - ) - .expect("parses"); - assert_eq!(config.calendars[0].leads(), vec![DEFAULT_LEAD_MINUTES]); - assert_eq!(config.calendars[0].horizon(), DEFAULT_HORIZON_HOURS); - // A blank or zero lead list reads as the default, not as - // silence. - let config: ClawConfig = serde_yaml::from_str( - "calendars:\n - name: home\n kind: ics\n url: https://c.example.net/a.ics\n lead_minutes: []\n", - ) - .expect("parses"); - assert_eq!(config.calendars[0].leads(), vec![DEFAULT_LEAD_MINUTES]); - } - - #[test] - fn a_calendar_typo_is_rejected_and_named() { - let err = serde_yaml::from_str::( - "calendars:\n - name: home\n kind: ics\n uurl: https://c.example.net/a.ics\n", - ) - .expect_err("a typo'd calendar key is loud"); - assert!(err.to_string().contains("uurl"), "got: {err}"); - } - - #[test] - fn calendar_validation_fails_loudly_and_specifically() { - let config = |calendars: Vec| ClawConfig { - calendars, - ..ClawConfig::default() - }; - let ics = |name: &str, url: Option<&str>| CalendarWatch { - name: name.to_owned(), - kind: CalendarKind::Ics, - url: url.map(str::to_owned), - calendar_id: None, - token: None, - username: None, - password: None, - api_key: None, - lead_minutes: None, - horizon_hours: None, - }; - let caldav = |name: &str, url: Option<&str>| CalendarWatch { - name: name.to_owned(), - kind: CalendarKind::Caldav, - url: url.map(str::to_owned), - calendar_id: None, - token: None, - username: None, - password: None, - api_key: None, - lead_minutes: None, - horizon_hours: None, - }; - let google = |name: &str, id: Option<&str>, api_key: Option<&str>| CalendarWatch { - name: name.to_owned(), - kind: CalendarKind::Google, - url: None, - calendar_id: id.map(str::to_owned), - token: None, - username: None, - password: None, - api_key: api_key.map(str::to_owned), - lead_minutes: None, - horizon_hours: None, - }; - - let err = config(vec![ics("", Some("https://c.example.net/a.ics"))]) - .validate() - .expect_err("a blank name is loud"); - assert!(err.to_string().contains("blank"), "got: {err}"); - - let err = config(vec![ - ics("home", Some("https://c.example.net/a.ics")), - ics("home", Some("webcal://c.example.net/b.ics")), - ]) - .validate() - .expect_err("a duplicate name is loud"); - assert!(err.to_string().contains("duplicate"), "got: {err}"); - - let err = config(vec![ics("home", None)]) - .validate() - .expect_err("a url-less feed is loud"); - assert!(err.to_string().contains("feed URL"), "got: {err}"); - - let err = config(vec![ics("home", Some("ftp://c.example.net/a.ics"))]) - .validate() - .expect_err("a junk scheme is loud"); - assert!(err.to_string().contains("https://"), "got: {err}"); - - let err = config(vec![caldav("home", None)]) - .validate() - .expect_err("a url-less collection is loud"); - assert!(err.to_string().contains("DAV"), "got: {err}"); - - let err = config(vec![caldav("home", Some("webcal://c.example.net/dav/"))]) - .validate() - .expect_err("webcal is not a dav spelling"); - assert!(err.to_string().contains("https://"), "got: {err}"); - - let err = config(vec![google("work", None, Some("k"))]) - .validate() - .expect_err("an id-less google calendar is loud"); - assert!(err.to_string().contains("calendar_id"), "got: {err}"); - - let err = config(vec![google("work", Some("primary"), None)]) - .validate() - .expect_err("a credential-less google calendar is loud"); - assert!(err.to_string().contains("CALENDAR_TOKEN"), "got: {err}"); - - let mut late = ics("home", Some("https://c.example.net/a.ics")); - late.lead_minutes = Some(vec![60]); - late.horizon_hours = Some(1); - let err = config(vec![late]) - .validate() - .expect_err("a lead past the horizon is loud"); - assert!(err.to_string().contains("horizon"), "got: {err}"); - - // A username with a password validates, as does a collection - // with none of it (an open server) or a token instead. - let mut basic = caldav("home", Some("https://c.example.net/dav/home/")); - basic.username = Some("me".to_owned()); - basic.password = Some("app-pass".to_owned()); - let mut bearer = caldav("shared", Some("http://10.0.0.5/dav/shared/")); - bearer.token = Some("sekrit".to_owned()); - config(vec![ - ics("feed", Some("webcal://c.example.net/a.ics")), - basic, - bearer, - google("work", Some("primary"), Some("k")), - ]) - .validate() - .expect("a complete list validates"); - } - - #[test] - fn the_caldav_password_falls_back_to_the_env_and_blank_counts_as_unset() { - let env = |name: &str| match name { - "CALDAV_PASSWORD" => Ok("from-env".to_owned()), - _ => Err(std::env::VarError::NotPresent), - }; - assert_eq!( - CalendarWatch::resolve_password_with(env, Some(" from-config ")), - Some("from-config".to_owned()), - "the config password wins" - ); - assert_eq!( - CalendarWatch::resolve_password_with(env, None), - Some("from-env".to_owned()), - "the env password serves when no config password" - ); - assert_eq!( - CalendarWatch::resolve_password_with( - |_| Err(std::env::VarError::NotPresent), - Some(" ") - ), - None, - "a blank password counts as unset" - ); - } - - #[test] - fn the_calendar_token_falls_back_to_the_env_and_blank_counts_as_unset() { - let env = |name: &str| match name { - "CALENDAR_TOKEN" => Ok("from-env".to_owned()), - _ => Err(std::env::VarError::NotPresent), - }; - assert_eq!( - CalendarWatch::resolve_token_with(env, Some(" from-config ")), - Some("from-config".to_owned()), - "the config token wins" - ); - assert_eq!( - CalendarWatch::resolve_token_with(env, None), - Some("from-env".to_owned()), - "the env token serves when no config token" - ); - assert_eq!( - CalendarWatch::resolve_token_with(|_| Err(std::env::VarError::NotPresent), Some(" ")), - None, - "a blank token counts as unset" - ); - } - #[test] fn forge_names_parse_as_the_sources_spell_them() { assert_eq!(ForgeKind::parse("github"), Some(ForgeKind::GitHub)); diff --git a/crates/rhema-claw/src/daemon.rs b/crates/rhema-claw/src/daemon.rs index 9865724..e9dfc91 100644 --- a/crates/rhema-claw/src/daemon.rs +++ b/crates/rhema-claw/src/daemon.rs @@ -191,6 +191,10 @@ impl Daemon { // drives, jobs the jobs dispatcher answers. let jobs = Arc::new(crate::job_store::CodingJobStore::open(&db)?); + // The calendar table: one truth, read fresh by the reminders + // source every pass; `/v1/calendars` writes it. + let calendars = Arc::new(rhema_calendar_integration::CalendarStore::open(&db)?); + let forge_dispatcher = Arc::new(Dispatcher::new( dispatch, Arc::new(SystemClock), @@ -327,7 +331,9 @@ impl Daemon { identity: identity.clone(), hub: hub.clone(), workspace: interactive_workspace.clone(), - calendars: config.calendars.clone(), + // The calendar table: one truth, read fresh by the + // reminders source every pass; `/v1/calendars` writes it. + calendars: calendars.clone(), })); eprintln!("[claw] extensions: {}", extensions.names().join(", ")); // The sources register on their lanes as they build: a watch of @@ -384,6 +390,7 @@ impl Daemon { interactive_driver.clone(), store, watches, + calendars.clone(), notifications, tasks, bootstrap, diff --git a/crates/rhema-claw/src/extension.rs b/crates/rhema-claw/src/extension.rs index dc22e6f..d9f4bc0 100644 --- a/crates/rhema-claw/src/extension.rs +++ b/crates/rhema-claw/src/extension.rs @@ -36,12 +36,13 @@ use rhema_autonomous_agent::{Allowlist, Source, ToolRegistry}; use rhema_core::ToolResolver; use rhema_rsd::{Dispatcher, ResourceKey, SystemClock}; -use crate::config::{CalendarWatch, ForgeKind}; +use crate::config::ForgeKind; use crate::consents::WaitBridge; use crate::daemon::ClawError; use crate::forge::ForgeSource; use crate::hub::EventHub; use rhema_calendar_integration::CalendarSource; +use rhema_calendar_integration::CalendarStore; /// The store handles the standard set is assembled over. One struct, /// so a new extension is a field plus a constructor line, not another @@ -78,8 +79,9 @@ pub struct ExtensionDeps { /// The interactive workspace the account exec and the identity /// registry resolve against. pub workspace: PathBuf, - /// The calendar feeds the reminders source polls. - pub calendars: Vec, + /// The calendar table, shared with `/v1/calendars`: the source + /// reads it fresh every pass, so a row added at runtime is live. + pub calendars: Arc, } /// The dispatch lanes a source can feed, handed to @@ -477,7 +479,7 @@ impl ClawExtension for PhoneExtension { /// The calendar reminders: a due occurrence is an event on the resident /// lane, never a timer near the model. struct CalendarExtension { - calendars: Vec, + calendars: Arc, } impl ClawExtension for CalendarExtension { @@ -486,9 +488,11 @@ impl ClawExtension for CalendarExtension { } fn sources(&self, lanes: &Lanes) -> Result>, ClawError> { - let source = - CalendarSource::new(&lanes.interactive, Arc::new(SystemClock), &self.calendars) - .map_err(ClawError::Backend)?; + let source = CalendarSource::from_store( + &lanes.interactive, + Arc::new(SystemClock), + self.calendars.clone(), + ); Ok(vec![Box::new(source)]) } } diff --git a/crates/rhema-claw/src/http.rs b/crates/rhema-claw/src/http.rs index ff3a8f0..242ae6e 100644 --- a/crates/rhema-claw/src/http.rs +++ b/crates/rhema-claw/src/http.rs @@ -67,6 +67,9 @@ pub struct ClawState { driver: Arc, store: Arc, watches: Arc, + /// The calendar table: `/v1/calendars` writes it, the reminders + /// source reads it fresh every pass. + calendars: Arc, notifications: Arc, /// The scheduled-task table the `/v1/tasks` routes serve — the same /// rows the resident scheduler fires. @@ -117,6 +120,7 @@ impl ClawState { driver: Arc, store: Arc, watches: Arc, + calendars: Arc, notifications: Arc, tasks: Arc, bootstrap: Arc, @@ -138,6 +142,7 @@ impl ClawState { driver, store, watches, + calendars, notifications, tasks, bootstrap, @@ -232,6 +237,12 @@ pub fn router(state: ClawState) -> Router { .route("/bootstrap", get(bootstrap_status)) .route("/watches", get(watches_list).post(watches_add)) .route("/watches/{forge}/{*repo}", delete(watches_remove)) + .route( + "/calendars", + get(calendars_list) + .post(calendars_add) + .delete(calendars_remove), + ) .route("/accounts", get(accounts_list).post(accounts_add)) .route("/accounts/{id}", delete(accounts_remove)) .route("/consents", get(consents_list)) @@ -282,7 +293,7 @@ pub fn router(state: ClawState) -> Router { Json(json!({ "service": "rhema-claw", "chat": "/v1/sessions", - "api": ["/v1/sessions", "/v1/providers", "/v1/watches", + "api": ["/v1/sessions", "/v1/providers", "/v1/watches", "/v1/calendars", "/v1/accounts", "/v1/notifications", "/v1/phones", "/v1/bootstrap", "/v1/ingest"], })) @@ -618,6 +629,74 @@ async fn events( .into_response() } +/// `POST /v1/calendars`: one calendar spec, the integration's +/// `CalendarWatch` shape, validated at the table's door — the same +/// rules the config's boot check used to run. +async fn calendars_add( + State(state): State, + Json(body): Json, +) -> Response { + match state.calendars.add(&body) { + Ok(true) => ( + StatusCode::CREATED, + Json(json!({"added": body.name.trim()})), + ) + .into_response(), + Ok(false) => ( + StatusCode::OK, + Json(json!({ + "added": body.name.trim(), + "note": "a calendar with that name already exists" + })), + ) + .into_response(), + Err(err) => ( + StatusCode::BAD_REQUEST, + Json(json!({"error": err.to_string()})), + ) + .into_response(), + } +} + +/// `DELETE /v1/calendars?name=…`: forget a calendar; the reminders +/// source stops polling it on the next tick. +async fn calendars_remove( + State(state): State, + Query(params): Query>, +) -> Response { + let Some(name) = params + .get("name") + .map(|n| n.trim()) + .filter(|n| !n.is_empty()) + else { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "name is required"})), + ) + .into_response(); + }; + match state.calendars.remove(name) { + Ok(removed) => Json(json!({"removed": removed, "name": name})).into_response(), + Err(err) => ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": err.to_string()})), + ) + .into_response(), + } +} + +/// `GET /v1/calendars`: every calendar, in creation order. +async fn calendars_list(State(state): State) -> Response { + match state.calendars.list() { + Ok(list) => Json(json!({"calendars": list})).into_response(), + Err(err) => ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": err.to_string()})), + ) + .into_response(), + } +} + /// One `TurnEvent` as an SSE frame: `data: {json}`. A lagged subscriber /// gets a comment naming the drop — live delivery is best-effort; the /// record replays what completed. A serialization failure is logged with diff --git a/crates/rhema-claw/src/interactive.rs b/crates/rhema-claw/src/interactive.rs index f8296d8..8be18b3 100644 --- a/crates/rhema-claw/src/interactive.rs +++ b/crates/rhema-claw/src/interactive.rs @@ -694,7 +694,12 @@ mod tests { identity, hub, workspace: workspace.to_path_buf(), - calendars: Vec::new(), + // The test assembly"s calendar table: in-memory, and + // empty — the interactive tests do not remind. + calendars: std::sync::Arc::new( + rhema_calendar_integration::CalendarStore::in_memory() + .expect("the calendar table opens"), + ), }, )) } diff --git a/crates/rhema-claw/tests/continuity.rs b/crates/rhema-claw/tests/continuity.rs index 7cf6e46..fee3c1f 100644 --- a/crates/rhema-claw/tests/continuity.rs +++ b/crates/rhema-claw/tests/continuity.rs @@ -81,7 +81,6 @@ fn config(dir: &tempfile::TempDir) -> ClawConfig { tools_bin: None, transcript: None, ingest_token: None, - calendars: Vec::new(), } } diff --git a/crates/rhema-claw/tests/e2e.rs b/crates/rhema-claw/tests/e2e.rs index 5e983c6..93a81bd 100644 --- a/crates/rhema-claw/tests/e2e.rs +++ b/crates/rhema-claw/tests/e2e.rs @@ -59,7 +59,6 @@ fn config(dir: &tempfile::TempDir) -> ClawConfig { tools_bin: None, transcript: None, ingest_token: None, - calendars: Vec::new(), } } diff --git a/crates/rhema-claw/tests/http.rs b/crates/rhema-claw/tests/http.rs index 58d9717..8c18fac 100644 --- a/crates/rhema-claw/tests/http.rs +++ b/crates/rhema-claw/tests/http.rs @@ -65,7 +65,6 @@ fn config(dir: &tempfile::TempDir, token: Option<&str>) -> ClawConfig { tools_bin: None, ingest_token: None, transcript: None, - calendars: Vec::new(), } } @@ -893,6 +892,89 @@ async fn a_configured_token_guards_v1_and_leaves_healthz_open() { assert_eq!(whole["choices"][0]["message"]["content"], "on it"); } +/// The calendar table, over its door: add, list, refuse the invalid +/// loudly, remove — and the route shares the table the reminders +/// source reads, so no restart is in the story. +#[tokio::test] +async fn calendars_round_trip_over_their_door() { + let dir = tempfile::tempdir().expect("tempdir"); + let daemon = Daemon::build( + &config(&dir, None), + mock_registry(Arc::new(MockProvider::new(Vec::new()))), + rhema_rsd::DispatcherConfig::default(), + ) + .expect("the daemon assembles"); + let base = serve(&daemon).await; + let client = reqwest::Client::new(); + + // A valid ics lands. + let res = client + .post(format!("{base}/v1/calendars")) + .json(&json!({"name": "team", "kind": "ics", "url": "https://c.example.net/a.ics"})) + .send() + .await + .expect("the post answers"); + assert_eq!(res.status(), 201, "a new calendar is created"); + + // The invalid is loud, wherever it would enter. + let res = client + .post(format!("{base}/v1/calendars")) + .json(&json!({"name": "broken", "kind": "ics"})) + .send() + .await + .expect("the post answers"); + assert_eq!( + res.status(), + 400, + "an urlless ics is refused: {}", + res.status() + ); + let err: serde_json::Value = res.json().await.expect("the error parses"); + assert!( + err["error"].as_str().unwrap().contains("ics is a feed URL"), + "{err}" + ); + + // The list carries what landed, and nothing else. + let listed: serde_json::Value = client + .get(format!("{base}/v1/calendars")) + .send() + .await + .expect("the list answers") + .json() + .await + .expect("the list parses"); + let names: Vec<&str> = listed["calendars"] + .as_array() + .expect("an array") + .iter() + .map(|c| c["name"].as_str().expect("a name")) + .collect(); + assert_eq!(names, vec!["team"], "{listed}"); + + // Removal is a delete on the name; the next pass simply has one + // fewer feed. + let res = client + .delete(format!("{base}/v1/calendars?name=team")) + .send() + .await + .expect("the delete answers"); + assert_eq!(res.status(), 200); + let listed: serde_json::Value = client + .get(format!("{base}/v1/calendars")) + .send() + .await + .expect("the list answers") + .json() + .await + .expect("the list parses"); + assert_eq!( + listed["calendars"].as_array().map(Vec::len), + Some(0), + "{listed}" + ); +} + /// The ingest door: an external event lands on its key in the resident /// dispatcher, an identical redelivery dedups, the daemon's own source /// names cannot be re-registered, and the bearer guard covers it like diff --git a/integrations/rhema-calendar-integration/Cargo.toml b/integrations/rhema-calendar-integration/Cargo.toml index c128750..07cd92e 100644 --- a/integrations/rhema-calendar-integration/Cargo.toml +++ b/integrations/rhema-calendar-integration/Cargo.toml @@ -12,6 +12,7 @@ authors.workspace = true publish = false [dependencies] +rusqlite = { version = "0.40", features = ["bundled"] } # The reminder source: one `Source` over the feeds, detect-pass shaped # like the task scheduler's rhema-autonomous-agent = { path = "../../crates/rhema-autonomous-agent", version = "0.1.0" } diff --git a/integrations/rhema-calendar-integration/src/lib.rs b/integrations/rhema-calendar-integration/src/lib.rs index 8995bde..0102386 100644 --- a/integrations/rhema-calendar-integration/src/lib.rs +++ b/integrations/rhema-calendar-integration/src/lib.rs @@ -73,6 +73,8 @@ pub mod event; pub mod google; pub mod ics; pub mod reminder; +pub mod store; +pub use store::CalendarStore; // Re-exported so implementors of the async seam can apply #[async_trait] // without adding the crate themselves. diff --git a/integrations/rhema-calendar-integration/src/reminder.rs b/integrations/rhema-calendar-integration/src/reminder.rs index 08f5b45..39b2fb3 100644 --- a/integrations/rhema-calendar-integration/src/reminder.rs +++ b/integrations/rhema-calendar-integration/src/reminder.rs @@ -73,6 +73,7 @@ fn reminder_key_fn() -> Arc { } /// One configured calendar, resolved to a client the source polls. +#[derive(Clone)] pub(crate) struct Feed { name: String, client: Arc, @@ -236,7 +237,15 @@ fn nonempty(value: Option<&str>) -> Option<&str> { pub struct CalendarSource { dispatcher: Arc, clock: Arc, - feeds: Vec, + /// Where this pass's feeds come from: a static list (tests, and + /// callers that hold the specs) or the calendar table, read fresh + /// on every detect pass so a runtime add is live next tick. + feeds: FeedSource, +} + +enum FeedSource { + Static(Vec), + Store(Arc), } impl CalendarSource { @@ -262,10 +271,37 @@ impl CalendarSource { Ok(Self { dispatcher: dispatcher.clone(), clock, - feeds, + feeds: FeedSource::Static(feeds), }) } + /// Builds from the calendar table instead of a static list: the + /// source reads it fresh every pass, so a calendar added over the + /// HTTP route (or a chat tool) is live without a restart, and a + /// removed one stops polling on the next tick. Config validation + /// ran when the row landed; a calendar that still fails to resolve + /// here names itself and costs its tick, not the daemon. + pub fn from_store( + dispatcher: &Arc, + clock: Arc, + store: Arc, + ) -> Self { + dispatcher.register_source(CALENDARS_SOURCE, reminder_key_fn()); + Self { + dispatcher: dispatcher.clone(), + clock, + feeds: FeedSource::Store(store), + } + } + + /// The feeds this source polls, resolved for one detect pass. + fn feeds_for_pass(&self) -> Result, String> { + match &self.feeds { + FeedSource::Static(feeds) => Ok(feeds.to_vec()), + FeedSource::Store(store) => store_feeds(store), + } + } + /// A source over ready-made feeds. Tests use this; production /// builds from the config with [`CalendarSource::new`]. #[cfg(test)] @@ -278,11 +314,30 @@ impl CalendarSource { Self { dispatcher: dispatcher.clone(), clock, - feeds, + feeds: FeedSource::Static(feeds), } } } +/// Resolves the table's calendars into this pass's feeds. A feed's +/// client is rebuilt per pass — the table is small, and correctness +/// (a runtime add is live) beats a pool the row could outlive. +fn store_feeds(store: &crate::store::CalendarStore) -> Result, String> { + let calendars = store + .list() + .map_err(|err| format!("calendar table: {err}"))?; + let mut feeds = Vec::with_capacity(calendars.len()); + for calendar in &calendars { + feeds.push(Feed { + name: calendar.name.trim().to_owned(), + client: client_for(calendar)?, + leads: calendar.leads(), + horizon_hours: calendar.horizon(), + }); + } + Ok(feeds) +} + #[async_trait] impl Source for CalendarSource { fn name(&self) -> &str { @@ -298,7 +353,8 @@ impl Source for CalendarSource { let now = OffsetDateTime::from_unix_timestamp_nanos(now_ms as i128 * 1_000_000) .map_err(|err| SourceError::Backend(format!("clock: {err}")))?; let mut fed = 0; - for feed in &self.feeds { + let feeds = self.feeds_for_pass().map_err(SourceError::Backend)?; + for feed in &feeds { let window = Window::new( now, now + time::Duration::hours(i64::from(feed.horizon_hours)), @@ -608,6 +664,68 @@ impl CalendarKind { } } +impl CalendarWatch { + /// The rules a calendar must satisfy to poll — the checks the + /// config's boot validation used to run, now at the one door every + /// writer shares: the table's `add`. + pub fn validate(&self) -> Result<(), String> { + if self.name.trim().is_empty() { + return Err("name must not be blank".to_owned()); + } + match self.kind { + CalendarKind::Ics => { + let url = nonempty(self.url.as_deref()).ok_or_else(|| { + "ics is a feed URL: set url to the calendar's address".to_owned() + })?; + if !(url.starts_with("https://") + || url.starts_with("http://") + || url.starts_with("webcal://")) + { + return Err(format!( + "url must be https://, http://, or webcal://, got {url:?}" + )); + } + } + CalendarKind::Caldav => { + let url = nonempty(self.url.as_deref()).ok_or_else(|| { + "caldav is a collection: set url to the calendar's DAV address".to_owned() + })?; + if !(url.starts_with("https://") || url.starts_with("http://")) { + return Err(format!("url must be https:// or http://, got {url:?}")); + } + let has_user = nonempty(self.username.as_deref()).is_some(); + if has_user && self.resolved_password().is_none() { + return Err("no password: set password or CALDAV_PASSWORD".to_owned()); + } + } + CalendarKind::Google => { + if nonempty(self.calendar_id.as_deref()).is_none() { + return Err( + "google needs a calendar: set calendar_id (\"primary\", an address, a hex id)" + .to_owned(), + ); + } + if self.resolved_api_key().is_none() && self.resolved_token().is_none() { + return Err( + "no credential: set api_key (public calendars) or token, or CALENDAR_TOKEN" + .to_owned(), + ); + } + } + } + // The horizon is what the poll can see; a lead at or past it + // fires on an event that was never visible. + let horizon_minutes = u64::from(self.horizon()) * 60; + let longest = self.leads().into_iter().max().unwrap_or_default(); + if longest >= horizon_minutes { + return Err(format!( + "lead {longest}m is not under the {horizon_minutes}m horizon: a reminder cannot fire before its event is visible" + )); + } + Ok(()) + } +} + /// One watched calendar. Unknown keys are rejected, as on a watch. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "snake_case", deny_unknown_fields)] diff --git a/integrations/rhema-calendar-integration/src/store.rs b/integrations/rhema-calendar-integration/src/store.rs new file mode 100644 index 0000000..2a907f4 --- /dev/null +++ b/integrations/rhema-calendar-integration/src/store.rs @@ -0,0 +1,168 @@ +//! The calendar table: agent-managed calendars in SQLite, the same +//! database file the dispatcher's store uses — the pattern the claw's +//! watch table established. The config stopped carrying calendars: +//! the table is the one truth, and the source reads it fresh on every +//! detect pass, so a row added over `/v1/calendars` (or a chat tool) +//! is live without a restart. + +use crate::reminder::CalendarWatch; +use rusqlite::Connection; +use std::path::Path; +use std::sync::Mutex; + +/// An open handle on the calendar table. Its own rusqlite connection +/// to the shared database file — WAL and a busy timeout, so the +/// dispatcher's connection and this one interleave through SQLite's +/// own locking. +pub struct CalendarStore { + conn: Mutex, +} + +impl CalendarStore { + /// Opens (creating, if needed) the calendar table inside `path`. + pub fn open(path: impl AsRef) -> Result { + let path = path.as_ref(); + if let Some(dir) = path.parent() { + std::fs::create_dir_all(dir)?; + } + let conn = Connection::open(path).map_err(std::io::Error::other)?; + conn.busy_timeout(std::time::Duration::from_millis(5_000)) + .map_err(std::io::Error::other)?; + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS calendars ( + name TEXT PRIMARY KEY, + spec TEXT NOT NULL, + created_at INTEGER NOT NULL + );", + ) + .map_err(std::io::Error::other)?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + /// An in-memory table; tests use this. + pub fn in_memory() -> Result { + let conn = Connection::open_in_memory().map_err(std::io::Error::other)?; + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS calendars ( + name TEXT PRIMARY KEY, + spec TEXT NOT NULL, + created_at INTEGER NOT NULL + );", + ) + .map_err(std::io::Error::other)?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + /// Adds (or refuses, when the name is taken) a calendar. The spec + /// is validated first — the same door every writer shares, the + /// route and a future chat tool alike — so an invalid calendar is + /// loud wherever it enters. + pub fn add(&self, calendar: &CalendarWatch) -> Result { + calendar + .validate() + .map_err(std::io::Error::other)?; + let conn = self + .conn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let spec = serde_json::to_string(calendar).map_err(std::io::Error::other)?; + let inserted = conn + .execute( + "INSERT INTO calendars (name, spec, created_at) + VALUES (?1, ?2, ?3) + ON CONFLICT(name) DO NOTHING", + rusqlite::params![ + calendar.name.trim(), + spec, + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 + ], + ) + .map_err(std::io::Error::other)?; + Ok(inserted == 1) + } + + /// Removes the named calendar; `false` when there was none. + pub fn remove(&self, name: &str) -> Result { + let conn = self + .conn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let removed = conn + .execute( + "DELETE FROM calendars WHERE name = ?1", + rusqlite::params![name.trim()], + ) + .map_err(std::io::Error::other)?; + Ok(removed == 1) + } + + /// Every calendar, oldest first. The source reads this fresh on + /// every detect pass, so runtime adds take effect without a + /// restart. + pub fn list(&self) -> Result, std::io::Error> { + let conn = self + .conn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut stmt = conn + .prepare("SELECT spec FROM calendars ORDER BY created_at ASC, name ASC") + .map_err(std::io::Error::other)?; + let rows = stmt + .query_map([], |row| row.get::<_, String>(0)) + .map_err(std::io::Error::other)? + .collect::, _>>() + .map_err(std::io::Error::other)?; + rows.iter() + .map(|spec| serde_json::from_str(spec).map_err(std::io::Error::other)) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::CalendarKind; + + fn ics(name: &str) -> CalendarWatch { + serde_json::from_str(&format!( + "{{\"name\": \"{name}\", \"kind\": \"ics\", \"url\": \"https://example.net/f.ics\"}}" + )) + .expect("parses") + } + + #[test] + fn a_calendar_round_trips_through_the_table() { + let store = CalendarStore::in_memory().expect("store opens"); + assert!(store.add(&ics("team")).expect("add lands")); + assert!( + !store.add(&ics("team")).expect("add again"), + "a taken name is refused, not overwritten" + ); + let listed = store.list().expect("list reads"); + assert_eq!(listed.len(), 1); + assert_eq!(listed[0].name, "team"); + assert_eq!(listed[0].kind, CalendarKind::Ics); + assert!(store.remove("team").expect("remove lands")); + assert!(store.list().expect("list reads").is_empty()); + } + + #[test] + fn an_invalid_calendar_is_loud_wherever_it_enters() { + let store = CalendarStore::in_memory().expect("store opens"); + let blank: CalendarWatch = + serde_json::from_str("{\"name\": \" \", \"kind\": \"ics\"}").expect("parses"); + let err = store.add(&blank).expect_err("a blank name is refused"); + assert!(err.to_string().contains("name"), "got: {err}"); + let no_url: CalendarWatch = + serde_json::from_str("{\"name\": \"x\", \"kind\": \"ics\"}").expect("parses"); + let err = store.add(&no_url).expect_err("an urlless ics is refused"); + assert!(err.to_string().contains("url"), "got: {err}"); + } +}