vhaudiquet/rhema-claw
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(claw): the calendar table is the one truth — live, not config

Browse Source 
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.
This commit is contained in:
vhaudiquet committed 2026-10-24 01:52:09 +02:00
1 parent c6a3ed8c97
commit 818079d1f4
12 files changed
+481 -377
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/rhema-claw/src/config.rs
-360
View File
@@ -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<String>,
/// 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<CalendarWatch>,
}
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::<ClawConfig>(
"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<CalendarWatch>| 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));
crates/rhema-claw/src/daemon.rs
+8 -1
View File
@@ -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,
crates/rhema-claw/src/extension.rs
+11 -7
View File
@@ -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<CalendarWatch>,
/// 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<CalendarStore>,
}
/// 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<CalendarWatch>,
calendars: Arc<CalendarStore>,
}
impl ClawExtension for CalendarExtension {
@@ -486,9 +488,11 @@ impl ClawExtension for CalendarExtension {
}
fn sources(&self, lanes: &Lanes) -> Result<Vec<Box<dyn Source>>, 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)])
}
}
crates/rhema-claw/src/http.rs
+80 -1
View File
@@ -67,6 +67,9 @@ pub struct ClawState {
driver: Arc<Driver>,
store: Arc<SqliteStore>,
watches: Arc<ClawStore>,
/// The calendar table: `/v1/calendars` writes it, the reminders
/// source reads it fresh every pass.
calendars: Arc<rhema_calendar_integration::CalendarStore>,
notifications: Arc<NotificationStore>,
/// The scheduled-task table the `/v1/tasks` routes serve — the same
/// rows the resident scheduler fires.
@@ -117,6 +120,7 @@ impl ClawState {
driver: Arc<Driver>,
store: Arc<SqliteStore>,
watches: Arc<ClawStore>,
calendars: Arc<rhema_calendar_integration::CalendarStore>,
notifications: Arc<NotificationStore>,
tasks: Arc<TaskStore>,
bootstrap: Arc<Bootstrap>,
@@ -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<ClawState>,
Json(body): Json<rhema_calendar_integration::CalendarWatch>,
) -> 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<ClawState>,
Query(params): Query<std::collections::HashMap<String, String>>,
) -> 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<ClawState>) -> 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
crates/rhema-claw/src/interactive.rs
+6 -1
View File
@@ -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"),
),
},
))
}
crates/rhema-claw/tests/continuity.rs
-1
View File
@@ -81,7 +81,6 @@ fn config(dir: &tempfile::TempDir) -> ClawConfig {
tools_bin: None,
transcript: None,
ingest_token: None,
calendars: Vec::new(),
}
}
crates/rhema-claw/tests/e2e.rs
-1
View File
@@ -59,7 +59,6 @@ fn config(dir: &tempfile::TempDir) -> ClawConfig {
tools_bin: None,
transcript: None,
ingest_token: None,
calendars: Vec::new(),
}
}
crates/rhema-claw/tests/http.rs
+83 -1
View File
@@ -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
integrations/rhema-calendar-integration/Cargo.toml
+1
View File
@@ -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" }
integrations/rhema-calendar-integration/src/lib.rs
+2
View File
@@ -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.
integrations/rhema-calendar-integration/src/reminder.rs
+122 -4
View File
@@ -73,6 +73,7 @@ fn reminder_key_fn() -> Arc<KeyFn> {
}
/// One configured calendar, resolved to a client the source polls.
#[derive(Clone)]
pub(crate) struct Feed {
name: String,
client: Arc<dyn Calendar>,
@@ -236,7 +237,15 @@ fn nonempty(value: Option<&str>) -> Option<&str> {
pub struct CalendarSource {
dispatcher: Arc<Dispatcher>,
clock: Arc<dyn Clock>,
feeds: Vec<Feed>,
/// 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<Feed>),
Store(Arc<crate::store::CalendarStore>),
}
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<Dispatcher>,
clock: Arc<dyn Clock>,
store: Arc<crate::store::CalendarStore>,
) -> 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<Vec<Feed>, 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<Vec<Feed>, 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)]
integrations/rhema-calendar-integration/src/store.rs
+168
View File
@@ -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<Connection>,
}
impl CalendarStore {
/// Opens (creating, if needed) the calendar table inside `path`.
pub fn open(path: impl AsRef<Path>) -> Result<Self, std::io::Error> {
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<Self, std::io::Error> {
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<bool, std::io::Error> {
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<bool, std::io::Error> {
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<Vec<CalendarWatch>, 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::<Result<Vec<_>, _>>()
.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}");
}
}