//! Moving an aggregate to a different target endpoint. //! //! This is the sharpest failure mode in the design, which is why it is a command //! of its own rather than something a config edit triggers. A plain `sync` after //! the target changed would read the new, empty calendar as an aggregate whose //! every event had been deleted — and with delete propagation on, remove them //! from every source. `sync` therefore refuses on target drift and sends the user //! here. //! //! Two properties of the design make the move mechanical rather than a rebuild: //! the aggregate is a derived replica, so the sources stay authoritative //! throughout, and UID derivation excludes the target, so every derived UID is //! unchanged by the move. Only the rendered content and its location change. use std::path::Path; use thiserror::Error; use crate::config::{Config, Endpoint}; use crate::mirror; use crate::provenance; use crate::state::{State, Target}; use crate::sync::{self, SyncError, vdir_path}; use crate::vdir::{self, VdirError}; #[derive(Debug, Error)] pub enum RetargetError { #[error(transparent)] Sync(#[from] SyncError), #[error(transparent)] Vdir(#[from] VdirError), #[error(transparent)] State(#[from] crate::state::StateError), #[error("no aggregate named `{0}` in the configuration")] UnknownAggregate(String), #[error("no endpoint named `{0}` in the configuration")] UnknownEndpoint(String), #[error("endpoint `{0}` is read-only, so an aggregate cannot be published to it")] NotWritable(String), #[error( "endpoint `{endpoint}` is a source of aggregate `{aggregate}`, so targeting it would sync the aggregate into itself" )] TargetIsSource { endpoint: String, aggregate: String }, #[error( "aggregate `{0}` has never been synced, so there is nothing to move; edit the config and run `calcalist sync`" )] NeverSynced(String), } #[derive(Debug)] pub struct Outcome { pub aggregate: String, pub from: String, pub to: String, /// Events flushed to a sink before the move, which existed nowhere else. pub routed_first: usize, pub materialised: usize, pub purged: usize, } /// Moves `aggregate_id` to `new_target_id`. pub fn retarget( config: &Config, state_dir: &Path, aggregate_id: &str, new_target_id: &str, purge_old: bool, ) -> Result { let aggregate = config .aggregates .iter() .find(|candidate| candidate.id == aggregate_id) .ok_or_else(|| RetargetError::UnknownAggregate(aggregate_id.to_string()))?; let new_target = endpoint(config, new_target_id)?; if !new_target.kind.is_writable() { return Err(RetargetError::NotWritable(new_target_id.to_string())); } if aggregate .sources .iter() .any(|source| source == new_target_id) { return Err(RetargetError::TargetIsSource { endpoint: new_target_id.to_string(), aggregate: aggregate_id.to_string(), }); } let state_path = state_dir.join(crate::state::FILE_NAME); let mut state = State::load(&state_path)?; let recorded = state .aggregate(aggregate_id) .map(|entry| entry.target.clone()) .ok_or_else(|| RetargetError::NeverSynced(aggregate_id.to_string()))?; let old_target = endpoint(config, &recorded.endpoint)?.clone(); sync::prepare_vdirs(config, state_dir)?; let pimsync_config = if sync::needs_pimsync(config) { Some(crate::pimsync::write_config(config, state_dir).map_err(SyncError::from)?) } else { None }; if let Some(path) = &pimsync_config { crate::pimsync::sync(path).map_err(SyncError::from)?; } // Settle against the old target first. Events a user created in the // aggregate and that have not yet reached a sink exist only there, and would // be lost the moment we stop looking at it. let settled = sync::sync_aggregate( config, aggregate, &old_target, state_dir, &mut state, false, false, )?; let materialised = materialise(config, aggregate, new_target, state_dir, &mut state)?; let purged = if purge_old { purge(aggregate_id, &old_target, state_dir, &state)? } else { 0 }; let entry = state.aggregate_mut(aggregate_id, target_of(new_target)); entry.target = target_of(new_target); state.save(&state_path)?; if let Some(path) = &pimsync_config { crate::pimsync::sync(path).map_err(SyncError::from)?; } Ok(Outcome { aggregate: aggregate_id.to_string(), from: old_target.id.clone(), to: new_target.id.clone(), routed_first: settled.written_back, materialised, purged, }) } /// Rebuilds every mapped event on the new target, using that backend's transform. /// /// The rendered content differs between backends — a target that can suppress /// scheduling keeps real attendees, one that cannot has them demoted — so this is /// a rewrite of every event rather than a copy. fn materialise( config: &Config, aggregate: &crate::config::Aggregate, new_target: &Endpoint, state_dir: &Path, state: &mut State, ) -> Result { let new_dir = vdir_path(state_dir, &new_target.id); let existing = vdir::read(&new_dir)?; let suppression = new_target.kind.scheduling_suppression(); let links = state .aggregate(&aggregate.id) .map(|entry| entry.links.clone()) .unwrap_or_default(); let mut written = 0; let mut updated = links.clone(); for (aggregate_uid, link) in &links { let source = config.endpoint(&link.source_id); let Some(source) = source else { continue }; let source_items = vdir::read(&vdir_path(state_dir, &source.id))?; let Some(item) = source_items.get(&link.source_uid) else { // The source event is gone; the next ordinary sync will retire it. continue; }; let rebuilt = mirror::to_aggregate( &item.calendar, aggregate_uid, &source.id, &link.source_uid, source.kind.owner(), suppression, ); let replaces = existing.get(aggregate_uid).map(|held| held.path.clone()); vdir::write(&new_dir, aggregate_uid, &rebuilt, replaces.as_deref())?; if let Some(entry) = updated.get_mut(aggregate_uid) { entry.aggregate_hash = rebuilt.content_hash(); } written += 1; } state .aggregate_mut(&aggregate.id, target_of(new_target)) .links = updated; Ok(written) } /// Removes this aggregate's mirrors from the calendar it has left. /// /// Bounded by the derivation: only events whose UID this aggregate would have /// produced are touched, so anything the user keeps in that calendar survives. fn purge( aggregate_id: &str, old_target: &Endpoint, state_dir: &Path, state: &State, ) -> Result { let old_dir = vdir_path(state_dir, &old_target.id); let items = vdir::read(&old_dir)?; let links = state .aggregate(aggregate_id) .map(|entry| &entry.links) .ok_or_else(|| RetargetError::NeverSynced(aggregate_id.to_string()))?; let mut removed = 0; for (uid, item) in &items { if provenance::is_derived(uid) && links.contains_key(uid) { vdir::remove(&item.path)?; removed += 1; } } Ok(removed) } /// The state record describing where an aggregate is published. fn target_of(endpoint: &Endpoint) -> Target { Target { endpoint: endpoint.id.clone(), kind: endpoint.kind.kind_name().to_string(), } } fn endpoint<'a>(config: &'a Config, id: &str) -> Result<&'a Endpoint, RetargetError> { config .endpoint(id) .ok_or_else(|| RetargetError::UnknownEndpoint(id.to_string())) } #[cfg(test)] mod tests { use super::*; use std::fs; /// Google endpoints are used as stand-ins: they need no pimsync, so a whole /// cycle runs against local directories with nothing to reach over a network. const CONFIG: &str = r#" version = 1 [[endpoint]] id = "src" type = "google" calendar_id = "src@example.com" client_id = "x.apps.googleusercontent.com" [[endpoint]] id = "agg1" type = "google" calendar_id = "agg1@example.com" client_id = "x.apps.googleusercontent.com" [[endpoint]] id = "agg2" type = "google" calendar_id = "agg2@example.com" client_id = "x.apps.googleusercontent.com" [[aggregate]] id = "unified" target = "agg1" sources = ["src"] default_sink = "src" "#; fn config() -> Config { toml::from_str(CONFIG).expect("test config should parse") } fn event(uid: &str, summary: &str) -> String { format!( "BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:{uid}\r\nDTSTART:20260910T090000Z\r\nDTSTAMP:20260101T000000Z\r\nSUMMARY:{summary}\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n" ) } /// Reconciles one aggregate and records the result, without running a full /// cycle. A full cycle would contact Google for real, which these tests have /// no business doing — they are about what happens to local files. fn settle(config: &Config, dir: &Path) { let aggregate = &config.aggregates[0]; let target = config .endpoint(&aggregate.target) .expect("target endpoint") .clone(); let state_path = dir.join(crate::state::FILE_NAME); let mut state = State::load(&state_path).expect("state"); sync::sync_aggregate(config, aggregate, &target, dir, &mut state, false, false) .expect("reconcile"); state.save(&state_path).expect("save state"); } fn seed(config: &Config) -> tempfile::TempDir { let dir = tempfile::tempdir().expect("temp dir"); let src = vdir_path(dir.path(), "src"); fs::create_dir_all(&src).expect("create src vdir"); fs::write(src.join("a.ics"), event("a@example.com", "A")).expect("write"); fs::write(src.join("b.ics"), event("b@example.com", "B")).expect("write"); settle(config, dir.path()); dir } /// A state directory with two source events already mirrored into `agg1`. fn synced() -> tempfile::TempDir { seed(&config()) } fn count(dir: &Path, endpoint: &str) -> usize { vdir::read(&vdir_path(dir, endpoint)) .expect("read vdir") .len() } #[test] fn rebuilds_every_event_on_the_new_target() { let dir = synced(); assert_eq!(count(dir.path(), "agg1"), 2); let outcome = retarget(&config(), dir.path(), "unified", "agg2", false).expect("retarget"); assert_eq!(outcome.materialised, 2); assert_eq!(count(dir.path(), "agg2"), 2); // Kept by default: the old calendar may be wanted as a snapshot. assert_eq!(count(dir.path(), "agg1"), 2); assert_eq!(outcome.purged, 0); } #[test] fn records_the_new_target_so_sync_stops_refusing() { let dir = synced(); retarget(&config(), dir.path(), "unified", "agg2", false).expect("retarget"); let state = State::load(&dir.path().join(crate::state::FILE_NAME)).expect("state"); assert_eq!( state.aggregate("unified").expect("entry").target.endpoint, "agg2" ); } /// A purge is bounded by the derivation, so a calendar the user also keeps /// their own events in does not lose them. /// /// The aggregate has no sink here deliberately: with one configured, a /// foreign event in the aggregate is a user-created event and gets adopted /// into the sink rather than staying foreign, so it would never reach the /// purge as an outsider. #[test] fn purging_removes_only_this_aggregates_mirrors() { let sinkless: Config = toml::from_str(&CONFIG.replace(r#"default_sink = "src""#, "")) .expect("config should parse"); let dir = seed(&sinkless); let old = vdir_path(dir.path(), "agg1"); fs::write(old.join("mine.ics"), event("my-own@example.com", "Mine")).expect("write"); assert_eq!(count(dir.path(), "agg1"), 3); let outcome = retarget(&sinkless, dir.path(), "unified", "agg2", true).expect("retarget"); assert_eq!(outcome.purged, 2, "only the aggregate's own mirrors go"); let remaining = vdir::read(&old).expect("read"); assert_eq!(remaining.len(), 1); assert!( remaining.contains_key("my-own@example.com"), "the user's own event was destroyed" ); } /// The converse, and the reason the test above needs a sinkless aggregate: /// an event created in an aggregate that has a sink is adopted into it. #[test] fn a_foreign_event_is_adopted_when_a_sink_exists() { let dir = synced(); let old = vdir_path(dir.path(), "agg1"); fs::write(old.join("mine.ics"), event("my-own@example.com", "Mine")).expect("write"); retarget(&config(), dir.path(), "unified", "agg2", true).expect("retarget"); let sources = vdir::read(&vdir_path(dir.path(), "src")).expect("read"); assert!(sources.contains_key("my-own@example.com")); } /// An event created in the aggregate and not yet routed exists nowhere else, /// so the move must flush it to a sink before it stops looking at that /// calendar. #[test] fn an_unrouted_creation_reaches_the_sink_before_the_move() { let dir = synced(); let old = vdir_path(dir.path(), "agg1"); fs::write(old.join("hand.ics"), event("hand@phone", "Handwritten")).expect("write"); let outcome = retarget(&config(), dir.path(), "unified", "agg2", true).expect("retarget"); assert_eq!(outcome.routed_first, 1); let sources = vdir::read(&vdir_path(dir.path(), "src")).expect("read"); assert!( sources.contains_key("hand@phone"), "the only copy of the event was lost" ); } #[test] fn refuses_a_target_that_is_one_of_the_sources() { let dir = tempfile::tempdir().expect("temp dir"); let error = retarget(&config(), dir.path(), "unified", "src", false).expect_err("should refuse"); assert!( matches!(error, RetargetError::TargetIsSource { .. }), "{error}" ); } /// Writability is checked before any state is touched, so this needs no /// prior sync — which keeps a webcal endpoint out of the shared fixture. #[test] fn refuses_a_read_only_target() { let with_feed: Config = toml::from_str(&format!( "{CONFIG}\n[[endpoint]]\nid = \"feed\"\ntype = \"webcal\"\nurl = \"https://example.org/f.ics\"\n" )) .expect("config should parse"); let dir = tempfile::tempdir().expect("temp dir"); let error = retarget(&with_feed, dir.path(), "unified", "feed", false).expect_err("should refuse"); assert!(matches!(error, RetargetError::NotWritable(_)), "{error}"); } #[test] fn refuses_unknown_names() { let dir = tempfile::tempdir().expect("temp dir"); assert!(matches!( retarget(&config(), dir.path(), "nope", "agg2", false).expect_err("refuse"), RetargetError::UnknownAggregate(_) )); assert!(matches!( retarget(&config(), dir.path(), "unified", "nowhere", false).expect_err("refuse"), RetargetError::UnknownEndpoint(_) )); } /// Nothing has been published yet, so there is nothing to move. #[test] fn refuses_an_aggregate_that_has_never_synced() { let dir = tempfile::tempdir().expect("temp dir"); let error = retarget(&config(), dir.path(), "unified", "agg2", false).expect_err("should refuse"); assert!(matches!(error, RetargetError::NeverSynced(_)), "{error}"); } }