diff --git a/Cargo.lock b/Cargo.lock index e6ce9bd..0f3ff06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -52,16 +52,60 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "arrayvec" +version = "0.7.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" + +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + +[[package]] +name = "blake3" +version = "1.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d9e454fc11f76977dc803893aff6304ed33d6a26efae8696573bea74baa27ae" +dependencies = [ + "arrayvec", + "cc", + "cfg-if", + "constant_time_eq", + "cpufeatures", +] + [[package]] name = "calcalist" version = "0.1.0" dependencies = [ + "blake3", "clap", "serde", + "serde_json", + "tempfile", "thiserror", "toml", ] +[[package]] +name = "cc" +version = "1.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "005ec2760ca554fae18df7a11195552ec576cd665632a881bc011d5bb2fd4d80" +dependencies = [ + "find-msvc-tools", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + [[package]] name = "clap" version = "4.6.6" @@ -108,12 +152,60 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "constant_time_eq" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" + +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + [[package]] name = "equivalent" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "fastrand" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" + +[[package]] +name = "find-msvc-tools" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e0f1c7c3a72c66fd80abe965175f7523475c0489a87d3ff9d6e8c87d87a9d2d" + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", +] + [[package]] name = "hashbrown" version = "0.17.1" @@ -142,6 +234,36 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + [[package]] name = "once_cell_polyfill" version = "1.70.2" @@ -166,6 +288,25 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rustix" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys", +] + [[package]] name = "serde" version = "1.0.229" @@ -196,6 +337,19 @@ dependencies = [ "syn", ] +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + [[package]] name = "serde_spanned" version = "1.1.1" @@ -205,6 +359,12 @@ dependencies = [ "serde_core", ] +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + [[package]] name = "strsim" version = "0.11.1" @@ -222,6 +382,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom", + "once_cell", + "rustix", + "windows-sys", +] + [[package]] name = "thiserror" version = "2.0.20" @@ -313,3 +486,9 @@ name = "winnow" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml index aff47f4..5303d43 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,10 +7,15 @@ description = "Aggregate and sync events between CalDAV, Google Calendar and iCa publish = false [dependencies] +blake3 = "1.8" clap = { version = "4.6", features = ["derive"] } serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" thiserror = "2.0" toml = "1.1" +[dev-dependencies] +tempfile = "3" + [lints.rust] unsafe_code = "forbid" diff --git a/TODO.md b/TODO.md index 0b782f2..d4755ae 100644 --- a/TODO.md +++ b/TODO.md @@ -16,17 +16,20 @@ sync semantics these items implement. Core modules: -- [ ] `state.rs` — JSON sidecar, atomic temp + fsync + rename; records each aggregate's +- [x] `state.rs` — JSON sidecar, atomic temp + fsync + rename; records each aggregate's resolved target endpoint id **and** backend type -- [ ] `vdir.rs` — read and write vdir directories -- [ ] `ical.rs` — surgical line-level `.ics` editing (UID rewrite, property injection), +- [x] `vdir.rs` — read and write vdir directories +- [x] `ical.rs` — surgical line-level `.ics` editing (UID rewrite, property injection), respecting RFC 5545 folding; no parse-and-reserialize -- [ ] `provenance.rs` — deterministic `blake3(aggregate_id, source_id, source_uid)` UIDs -- [ ] `reconcile.rs` — the aggregation engine; pure, no I/O +- [x] `provenance.rs` — deterministic `blake3(aggregate_id, source_id, source_uid)` UIDs +- [x] `mirror.rs` — the to-aggregate and to-source transforms (added; not in the + original plan, which folded these into `reconcile`) +- [x] `reconcile.rs` — the aggregation engine; pure, no I/O +- [x] `sync.rs` — one cycle over the local vdirs, applying what `reconcile` decides - [ ] `pimsync.rs` — generate `pimsync.conf` (with `on_empty skip` and `on_delete skip`), drive one-shot `pimsync sync` - [ ] `google/auth.rs`, `google/api.rs`, `google/convert.rs` -- [ ] Reintroduce `SchedulingSuppression` in `config.rs` (removed in M0 as dead code) +- [x] Reintroduce `SchedulingSuppression` in `config.rs` (removed in M0 as dead code) Safety-critical behaviour: @@ -34,21 +37,25 @@ Safety-critical behaviour: guests are on a mail sink we control and confirm no mail is emitted; repeat for update and delete under `sendUpdates=none`. The Google attendee path depends on it. Fallback if it fails: the same demotion transform used for CalDAV. -- [ ] `sync` refuses to run on aggregate target drift, before reconciliation +- [x] `sync` refuses to run on aggregate target drift, before reconciliation - [ ] `aggregate retarget` — flush unrouted creations against the old target, then re-materialise; keep old orphans by default -- [ ] Mass-deletion guard (`max_delete_fraction`), overridable with `--force` -- [ ] Echo suppression: derived UIDs are never re-ingested as source events +- [x] Mass-deletion guard (`max_delete_fraction`), overridable with `--force`, with an + absolute floor so deleting a couple of events is never refused +- [x] Echo suppression: derived UIDs are never re-ingested as source events Tests: -- [ ] `reconcile` table-driven cases: create/update/delete each direction, both-sides-changed, +- [x] `reconcile` table-driven cases: create/update/delete each direction, both-sides-changed, routing, echo suppression, mass-delete abort -- [ ] `ical` round-trip fixtures: recurring with overrides, all-day, TZID, unknown `X-` props +- [x] `ical` round-trip fixtures: recurring with overrides, all-day, TZID, unknown `X-` props - [ ] Integration against Radicale plus a `file://` WebCal fixture; assert idempotence -- [ ] Safety: no live `ATTENDEE`/`ORGANIZER` on a CalDAV-targeted mirror, `VALARM` intact, - `PARTSTAT: DECLINED` maps to `TRANSP: TRANSPARENT`, emptying a source aborts -- [ ] Retarget: drift makes `sync` exit non-zero having written nothing and losing no source event +- [x] Safety (unit level): no live `ATTENDEE`/`ORGANIZER` on a CalDAV-targeted mirror, + `VALARM` intact, `PARTSTAT: DECLINED` maps to `TRANSP: TRANSPARENT`, bulk deletion aborts +- [ ] Safety (integration): the same against a real Radicale instance with an SMTP sink, + proving no mail is emitted +- [ ] Retarget: drift makes `sync` exit non-zero having written nothing and losing no source + event (verified by hand end to end; still needs an automated test) ## M2 — interface and packaging diff --git a/src/config.rs b/src/config.rs index c69d90f..c66b0be 100644 --- a/src/config.rs +++ b/src/config.rs @@ -54,6 +54,47 @@ impl EndpointKind { pub fn is_writable(&self) -> bool { !matches!(self, Self::Webcal { .. }) } + + pub fn kind_name(&self) -> &'static str { + match self { + Self::Caldav { .. } => "caldav", + Self::Google { .. } => "google", + Self::Webcal { .. } => "webcal", + } + } + + pub fn scheduling_suppression(&self) -> SchedulingSuppression { + match self { + Self::Google { .. } => SchedulingSuppression::Native, + Self::Caldav { .. } | Self::Webcal { .. } => SchedulingSuppression::None, + } + } + + /// The calendar owner's address, used to tell their own attendance apart + /// from the other guests'. + pub fn owner(&self) -> Option<&str> { + match self { + Self::Caldav { username, .. } => Some(username), + Self::Google { calendar_id, .. } => Some(calendar_id), + Self::Webcal { .. } => None, + } + } +} + +/// How scheduling messages — invitations, cancellations, replies — can be kept +/// from being sent when writing to a backend. +/// +/// The governing rule is that writes to an aggregate must never emit scheduling +/// mail, while writes to a source schedule normally. Google can be told not to +/// notify, so attendees survive a mirror intact. CalDAV offers no portable way to +/// suppress RFC 6638 scheduling, so inertness has to be structural instead: the +/// live properties are removed and the guest list carried as inert data. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SchedulingSuppression { + /// The backend can suppress notification, so attendees may be kept verbatim. + Native, + /// No suppression available; attendees must be demoted. + None, } #[derive(Debug, Clone, Deserialize, Serialize)] diff --git a/src/ical.rs b/src/ical.rs new file mode 100644 index 0000000..78c188b --- /dev/null +++ b/src/ical.rs @@ -0,0 +1,681 @@ +//! Surgical editing of iCalendar text. +//! +//! Events are mirrored between backends that each model calendars slightly +//! differently, and calcalist deliberately models only a fraction of RFC 5545. +//! Parsing into a typed structure and re-serialising would therefore drop every +//! property we do not know about — vendor extensions especially. Instead each +//! logical line keeps a byte range into the original text: untouched lines are +//! emitted exactly as they arrived, and only lines we actually change are rebuilt. +//! +//! Line terminators are normalised to whichever style dominates the input, since +//! RFC 5545 mandates CRLF and mixed endings are a defect rather than content. + +use std::ops::Range; + +use thiserror::Error; + +/// RFC 5545 section 3.1: lines SHOULD NOT be longer than 75 octets, excluding +/// the line break. +const FOLD_LIMIT: usize = 75; + +/// Properties servers rewrite on every store, which therefore say nothing about +/// whether the event's content actually changed. Excluded from the content hash +/// so a round trip through a server does not look like a user edit. +const VOLATILE_PROPERTIES: &[&str] = &["DTSTAMP", "LAST-MODIFIED"]; + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum IcalError { + #[error("the calendar is empty")] + Empty, + #[error("line {line}: continuation without a preceding line")] + DanglingContinuation { line: usize }, + #[error("line {line}: END:{found} closes {expected}")] + MismatchedEnd { + line: usize, + found: String, + expected: String, + }, + #[error("line {line}: END:{found} with no open component")] + UnopenedEnd { line: usize, found: String }, + #[error("component {0} is never closed")] + UnclosedComponent(String), +} + +/// Where a line's text comes from: untouched input, or something we built. +#[derive(Debug, Clone)] +enum Body { + Original(Range), + Generated(String), +} + +#[derive(Debug, Clone)] +struct Line { + body: Body, + /// The logical (unfolded) content of the line. + unfolded: String, + /// Component stack the line sits in, innermost last. `BEGIN:VEVENT` and its + /// matching `END` are both considered inside `VEVENT`. + path: Vec, +} + +impl Line { + fn generated(unfolded: String, path: Vec) -> Self { + Line { + body: Body::Generated(unfolded.clone()), + unfolded, + path, + } + } + + /// True when the line sits directly inside `component`, not in a nested one. + fn is_directly_in(&self, component: &str) -> bool { + self.path + .last() + .is_some_and(|name| name.eq_ignore_ascii_case(component)) + } + + fn name(&self) -> &str { + split_property(&self.unfolded).0 + } + + fn has_name(&self, name: &str) -> bool { + self.name().eq_ignore_ascii_case(name) + } +} + +/// One property occurrence, borrowed from the calendar. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Property<'a> { + pub name: &'a str, + /// Everything between the name and the value, without the leading `;`. + pub params: &'a str, + pub value: &'a str, +} + +impl<'a> Property<'a> { + /// Looks up a parameter by name, case-insensitively, unquoting the value. + pub fn param(&self, wanted: &str) -> Option<&'a str> { + split_params(self.params) + .find_map(|(name, value)| name.eq_ignore_ascii_case(wanted).then_some(unquote(value))) + } +} + +#[derive(Debug, Clone)] +pub struct Calendar { + source: String, + lines: Vec, + newline: String, +} + +/// Two calendars are equal when they render identically. Comparing the rendered +/// form rather than the internal line spans means an edited calendar and a freshly +/// parsed one with the same content compare equal, which is what callers mean. +impl PartialEq for Calendar { + fn eq(&self, other: &Self) -> bool { + self.to_ics() == other.to_ics() + } +} + +impl Calendar { + pub fn parse(text: &str) -> Result { + if text.trim().is_empty() { + return Err(IcalError::Empty); + } + let newline = if text.contains("\r\n") { "\r\n" } else { "\n" }; + let lines = build_lines(text)?; + Ok(Calendar { + source: text.to_string(), + lines, + newline: newline.to_string(), + }) + } + + /// Renders back to iCalendar text. + pub fn to_ics(&self) -> String { + let mut out = String::with_capacity(self.source.len() + 64); + for line in &self.lines { + match &line.body { + Body::Original(span) => out.push_str(&self.source[span.clone()]), + Body::Generated(text) => out.push_str(&fold(text, &self.newline)), + } + out.push_str(&self.newline); + } + out + } + + /// The UID shared by the event and any recurrence overrides. + pub fn uid(&self) -> Option<&str> { + self.properties("VEVENT", "UID").next().map(|p| p.value) + } + + /// Every occurrence of `name` directly inside `component`. + pub fn properties<'a>( + &'a self, + component: &'a str, + name: &'a str, + ) -> impl Iterator> { + self.lines + .iter() + .filter(move |line| line.is_directly_in(component) && line.has_name(name)) + .map(|line| { + let (name, params, value) = split_property(&line.unfolded); + Property { + name, + params, + value, + } + }) + } + + /// Rewrites the UID of every `VEVENT`, preserving any parameters. + /// Recurrence overrides share the UID, so all of them move together. + pub fn set_uid(&mut self, uid: &str) { + for line in &mut self.lines { + if line.is_directly_in("VEVENT") && line.has_name("UID") { + let (name, params, _) = split_property(&line.unfolded); + let separator = if params.is_empty() { "" } else { ";" }; + let rebuilt = format!("{name}{separator}{params}:{uid}"); + line.body = Body::Generated(rebuilt.clone()); + line.unfolded = rebuilt; + } + } + } + + /// Removes every occurrence of `names` sitting directly inside `component`, + /// returning how many lines went. Nested components are untouched, so an + /// `ATTENDEE` inside a `VALARM` — an alarm recipient, not a guest — survives. + pub fn remove_properties(&mut self, component: &str, names: &[&str]) -> usize { + let before = self.lines.len(); + self.lines.retain(|line| { + !(line.is_directly_in(component) && names.iter().any(|name| line.has_name(name))) + }); + before - self.lines.len() + } + + /// The unfolded text of every `name` line directly inside `component`. + /// Used when a property must be carried across verbatim rather than reformatted. + pub fn property_lines(&self, component: &str, name: &str) -> Vec<&str> { + self.lines + .iter() + .filter(|line| line.is_directly_in(component) && line.has_name(name)) + .map(|line| line.unfolded.as_str()) + .collect() + } + + /// Appends a property to every instance of `component`, just before its `END`. + pub fn add_property(&mut self, component: &str, name: &str, value: &str) { + self.insert_before_end(component, format!("{name}:{value}")); + } + + /// Appends an already-formatted `NAME;PARAMS:VALUE` line, preserving its + /// parameters exactly as they were on the donor event. + pub fn add_raw_property(&mut self, component: &str, unfolded: &str) { + self.insert_before_end(component, unfolded.to_string()); + } + + fn insert_before_end(&mut self, component: &str, unfolded: String) { + let ends: Vec = self + .lines + .iter() + .enumerate() + .filter(|(_, line)| line.is_directly_in(component) && line.unfolded.starts_with("END:")) + .map(|(index, _)| index) + .collect(); + // Insert from the back so earlier indices stay valid. + for index in ends.into_iter().rev() { + let path = self.lines[index].path.clone(); + let line = Line::generated(unfolded.clone(), path); + self.lines.insert(index, line); + } + } + + /// A digest of the event's meaningful content. + /// + /// Lines are keyed by the component they sit in and sorted, so that property + /// reordering — which carries no meaning in iCalendar — does not register as + /// a change, while a property moving between components does. + pub fn content_hash(&self) -> String { + let mut entries: Vec = self + .lines + .iter() + .filter(|line| { + !VOLATILE_PROPERTIES + .iter() + .any(|volatile| line.has_name(volatile)) + }) + .map(|line| format!("{}\t{}", line.path.join("/"), line.unfolded)) + .collect(); + entries.sort(); + let mut hasher = blake3::Hasher::new(); + for entry in &entries { + hasher.update(&(entry.len() as u64).to_le_bytes()); + hasher.update(entry.as_bytes()); + } + hasher + .finalize() + .as_bytes() + .iter() + .take(16) + .map(|byte| format!("{byte:02x}")) + .collect() + } + + /// Replaces `name` inside `component` if present, otherwise adds it. + pub fn set_property(&mut self, component: &str, name: &str, value: &str) { + let existing: Vec = self + .lines + .iter() + .enumerate() + .filter(|(_, line)| line.is_directly_in(component) && line.has_name(name)) + .map(|(index, _)| index) + .collect(); + if existing.is_empty() { + self.add_property(component, name, value); + return; + } + for index in existing { + let path = self.lines[index].path.clone(); + self.lines[index] = Line::generated(format!("{name}:{value}"), path); + } + } +} + +/// Groups physical lines into logical ones and records the component stack. +fn build_lines(text: &str) -> Result, IcalError> { + let mut lines: Vec = Vec::new(); + let mut path: Vec = Vec::new(); + let mut pending: Option<(Range, String)> = None; + + for (number, span) in physical_lines(text).into_iter().enumerate() { + let raw = &text[span.clone()]; + if let Some(rest) = continuation(raw) { + let Some((open_span, unfolded)) = pending.as_mut() else { + return Err(IcalError::DanglingContinuation { line: number + 1 }); + }; + open_span.end = span.end; + unfolded.push_str(rest); + continue; + } + if let Some((span, unfolded)) = pending.take() { + lines.push(close_line(span, unfolded, &mut path, number)?); + } + pending = Some((span, raw.to_string())); + } + if let Some((span, unfolded)) = pending.take() { + let number = lines.len(); + lines.push(close_line(span, unfolded, &mut path, number)?); + } + if let Some(open) = path.pop() { + return Err(IcalError::UnclosedComponent(open)); + } + Ok(lines) +} + +/// Finalises one logical line, maintaining the component stack around it. +fn close_line( + span: Range, + unfolded: String, + path: &mut Vec, + number: usize, +) -> Result { + let (name, _, value) = split_property(&unfolded); + if name.eq_ignore_ascii_case("BEGIN") { + path.push(value.trim().to_ascii_uppercase()); + } else if name.eq_ignore_ascii_case("END") { + let found = value.trim().to_ascii_uppercase(); + match path.last() { + Some(open) if *open == found => {} + Some(open) => { + return Err(IcalError::MismatchedEnd { + line: number + 1, + found, + expected: open.clone(), + }); + } + None => { + return Err(IcalError::UnopenedEnd { + line: number + 1, + found, + }); + } + } + } + let line_path = path.clone(); + if name.eq_ignore_ascii_case("END") { + path.pop(); + } + Ok(Line { + body: Body::Original(span), + unfolded, + path: line_path, + }) +} + +/// Byte ranges of each physical line, excluding its terminator. +fn physical_lines(text: &str) -> Vec> { + let bytes = text.as_bytes(); + let mut spans = Vec::new(); + let mut start = 0; + for (index, byte) in bytes.iter().enumerate() { + if *byte != b'\n' { + continue; + } + let mut end = index; + if end > start && bytes[end - 1] == b'\r' { + end -= 1; + } + spans.push(start..end); + start = index + 1; + } + if start < text.len() { + spans.push(start..text.len()); + } + spans +} + +/// A folded continuation begins with a single space or tab, which is not content. +fn continuation(raw: &str) -> Option<&str> { + raw.strip_prefix(' ').or_else(|| raw.strip_prefix('\t')) +} + +/// Splits `NAME;PARAM=x:value` at the first colon outside a quoted string. +fn split_property(line: &str) -> (&str, &str, &str) { + let mut quoted = false; + let mut colon = None; + let mut semicolon = None; + for (index, ch) in line.char_indices() { + match ch { + '"' => quoted = !quoted, + ';' if !quoted && semicolon.is_none() => semicolon = Some(index), + ':' if !quoted => { + colon = Some(index); + break; + } + _ => {} + } + } + let Some(colon) = colon else { + return (line, "", ""); + }; + match semicolon.filter(|index| *index < colon) { + Some(semicolon) => ( + &line[..semicolon], + &line[semicolon + 1..colon], + &line[colon + 1..], + ), + None => (&line[..colon], "", &line[colon + 1..]), + } +} + +/// Splits a parameter list on unquoted semicolons into `(name, value)` pairs. +fn split_params(params: &str) -> impl Iterator { + let mut parts = Vec::new(); + let mut quoted = false; + let mut start = 0; + for (index, ch) in params.char_indices() { + match ch { + '"' => quoted = !quoted, + ';' if !quoted => { + parts.push(¶ms[start..index]); + start = index + 1; + } + _ => {} + } + } + if start < params.len() { + parts.push(¶ms[start..]); + } + parts.into_iter().filter_map(|part| part.split_once('=')) +} + +fn unquote(value: &str) -> &str { + value + .strip_prefix('"') + .and_then(|rest| rest.strip_suffix('"')) + .unwrap_or(value) +} + +/// Folds a logical line to the 75-octet limit, never splitting a character. +fn fold(line: &str, newline: &str) -> String { + let mut out = String::with_capacity(line.len() + line.len() / FOLD_LIMIT + 1); + let mut used = 0; + for ch in line.chars() { + let width = ch.len_utf8(); + if used + width > FOLD_LIMIT { + out.push_str(newline); + out.push(' '); + used = 1; + } + out.push(ch); + used += width; + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A deliberately awkward event: a timezone, a recurrence override sharing the + /// UID, a folded description, a vendor property, an organiser, a declined + /// attendee, and an alarm that has an ATTENDEE of its own. + const FIXTURE: &str = concat!( + "BEGIN:VCALENDAR\r\n", + "VERSION:2.0\r\n", + "PRODID:-//Example//EN\r\n", + "BEGIN:VTIMEZONE\r\n", + "TZID:Europe/Amsterdam\r\n", + "END:VTIMEZONE\r\n", + "BEGIN:VEVENT\r\n", + "UID:event-1@example.com\r\n", + "DTSTART;TZID=Europe/Amsterdam:20260910T090000\r\n", + "RRULE:FREQ=WEEKLY;BYDAY=TH\r\n", + "EXDATE;TZID=Europe/Amsterdam:20260917T090000\r\n", + "SUMMARY:Weekly sync\r\n", + "DESCRIPTION:A description long enough that it arrives folded across more\r\n", + " than one physical line in the original file.\r\n", + "ORGANIZER;CN=\"Boss; Big\":mailto:boss@example.com\r\n", + "ATTENDEE;CN=Me;PARTSTAT=DECLINED:mailto:me@example.com\r\n", + "ATTENDEE;CN=Them;PARTSTAT=ACCEPTED:mailto:them@example.com\r\n", + "X-VENDOR-THING:preserve me\r\n", + "BEGIN:VALARM\r\n", + "ACTION:EMAIL\r\n", + "TRIGGER:-PT15M\r\n", + "ATTENDEE:mailto:me@example.com\r\n", + "END:VALARM\r\n", + "END:VEVENT\r\n", + "BEGIN:VEVENT\r\n", + "UID:event-1@example.com\r\n", + "RECURRENCE-ID;TZID=Europe/Amsterdam:20260924T090000\r\n", + "SUMMARY:Weekly sync (moved)\r\n", + "END:VEVENT\r\n", + "END:VCALENDAR\r\n", + ); + + fn parsed() -> Calendar { + Calendar::parse(FIXTURE).expect("fixture should parse") + } + + #[test] + fn round_trips_byte_for_byte() { + assert_eq!(parsed().to_ics(), FIXTURE); + } + + #[test] + fn unfolds_continuation_lines() { + let calendar = parsed(); + let description = calendar + .properties("VEVENT", "DESCRIPTION") + .next() + .expect("description present"); + assert_eq!( + description.value, + "A description long enough that it arrives folded across more than one physical line in the original file." + ); + } + + #[test] + fn reads_the_shared_uid() { + assert_eq!(parsed().uid(), Some("event-1@example.com")); + } + + #[test] + fn rewriting_the_uid_moves_every_recurrence_instance() { + let mut calendar = parsed(); + calendar.set_uid("deadbeef@calcalist"); + let uids: Vec<&str> = calendar + .properties("VEVENT", "UID") + .map(|property| property.value) + .collect(); + assert_eq!(uids, vec!["deadbeef@calcalist", "deadbeef@calcalist"]); + // Nothing else moved. + assert!(calendar.to_ics().contains("X-VENDOR-THING:preserve me")); + assert!(calendar.to_ics().contains("RRULE:FREQ=WEEKLY;BYDAY=TH")); + } + + #[test] + fn stripping_guests_spares_the_alarm_recipient() { + let mut calendar = parsed(); + let removed = calendar.remove_properties("VEVENT", &["ATTENDEE", "ORGANIZER"]); + assert_eq!(removed, 3); + + let output = calendar.to_ics(); + assert!(!output.contains("ORGANIZER")); + assert!(!output.contains("them@example.com")); + // The VALARM's own ATTENDEE is who the alarm notifies, not a guest. + assert!(output.contains("BEGIN:VALARM")); + assert_eq!(calendar.properties("VALARM", "ATTENDEE").count(), 1); + } + + #[test] + fn alarms_survive_untouched() { + let mut calendar = parsed(); + calendar.set_uid("deadbeef@calcalist"); + calendar.remove_properties("VEVENT", &["ATTENDEE", "ORGANIZER"]); + let output = calendar.to_ics(); + assert!(output.contains("BEGIN:VALARM\r\nACTION:EMAIL\r\nTRIGGER:-PT15M\r\n")); + } + + #[test] + fn reads_parameters_including_quoted_ones() { + let calendar = parsed(); + let organizer = calendar + .properties("VEVENT", "ORGANIZER") + .next() + .expect("organizer present"); + // The CN contains a semicolon and is quoted; it must not split the params. + assert_eq!(organizer.param("CN"), Some("Boss; Big")); + assert_eq!(organizer.value, "mailto:boss@example.com"); + + let declined = calendar + .properties("VEVENT", "ATTENDEE") + .find(|attendee| attendee.param("PARTSTAT") == Some("DECLINED")) + .expect("a declined attendee"); + assert_eq!(declined.value, "mailto:me@example.com"); + } + + #[test] + fn added_properties_land_inside_every_event() { + let mut calendar = parsed(); + calendar.add_property("VEVENT", "X-CALCALIST-SOURCE", "work"); + assert_eq!( + calendar.properties("VEVENT", "X-CALCALIST-SOURCE").count(), + 2 + ); + // Placed before the END, so it stays within the component. + assert!( + calendar + .to_ics() + .contains("X-CALCALIST-SOURCE:work\r\nEND:VEVENT") + ); + } + + #[test] + fn setting_a_property_replaces_rather_than_duplicates() { + let mut calendar = parsed(); + calendar.set_property("VEVENT", "SUMMARY", "Replaced"); + let summaries: Vec<&str> = calendar + .properties("VEVENT", "SUMMARY") + .map(|property| property.value) + .collect(); + assert_eq!(summaries, vec!["Replaced", "Replaced"]); + } + + #[test] + fn setting_an_absent_property_adds_it() { + let mut calendar = parsed(); + calendar.set_property("VEVENT", "TRANSP", "TRANSPARENT"); + assert_eq!(calendar.properties("VEVENT", "TRANSP").count(), 2); + } + + #[test] + fn generated_lines_are_folded_within_the_octet_limit() { + let mut calendar = parsed(); + let long = "x".repeat(400); + calendar.set_property("VEVENT", "DESCRIPTION", &long); + let output = calendar.to_ics(); + for line in output.split("\r\n") { + assert!( + line.len() <= FOLD_LIMIT, + "line exceeds the fold limit: {} octets", + line.len() + ); + } + // And it survives a re-parse unchanged. + let reparsed = Calendar::parse(&output).expect("folded output should parse"); + assert_eq!( + reparsed + .properties("VEVENT", "DESCRIPTION") + .next() + .map(|property| property.value), + Some(long.as_str()) + ); + } + + #[test] + fn folding_never_splits_a_multibyte_character() { + let mut calendar = parsed(); + // Four-byte characters, so a naive 75-byte cut would land mid-character. + let long = "🎉".repeat(80); + calendar.set_property("VEVENT", "SUMMARY", &long); + let output = calendar.to_ics(); + let reparsed = Calendar::parse(&output).expect("output should still be valid UTF-8 iCal"); + assert_eq!( + reparsed + .properties("VEVENT", "SUMMARY") + .next() + .map(|property| property.value), + Some(long.as_str()) + ); + } + + #[test] + fn accepts_lf_only_input_and_keeps_that_style() { + let lf = FIXTURE.replace("\r\n", "\n"); + let calendar = Calendar::parse(&lf).expect("LF input should parse"); + assert_eq!(calendar.to_ics(), lf); + } + + #[test] + fn rejects_structural_damage() { + assert_eq!(Calendar::parse(" ").unwrap_err(), IcalError::Empty); + + let mismatched = "BEGIN:VCALENDAR\r\nBEGIN:VEVENT\r\nEND:VCALENDAR\r\n"; + assert!(matches!( + Calendar::parse(mismatched), + Err(IcalError::MismatchedEnd { .. }) + )); + + let unclosed = "BEGIN:VCALENDAR\r\nVERSION:2.0\r\n"; + assert_eq!( + Calendar::parse(unclosed).unwrap_err(), + IcalError::UnclosedComponent("VCALENDAR".into()) + ); + + let dangling = " continuation first\r\n"; + assert_eq!( + Calendar::parse(dangling).unwrap_err(), + IcalError::DanglingContinuation { line: 1 } + ); + } +} diff --git a/src/main.rs b/src/main.rs index 2b053b3..0a9e159 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,20 +3,35 @@ mod cli; mod config; mod doctor; +mod ical; +mod mirror; mod paths; mod pimsync; +mod provenance; +mod reconcile; +mod state; +mod sync; +mod vdir; use std::process::ExitCode; use clap::Parser; use crate::cli::{Cli, Command}; +use crate::config::Config; +use crate::sync::Report; fn main() -> ExitCode { - let cli = Cli::parse(); - match &cli.command { - Command::Doctor => run_doctor(&cli), - other => unimplemented(other), + match &Cli::parse() { + cli @ Cli { + command: Command::Doctor, + .. + } => run_doctor(cli), + cli @ Cli { + command: Command::Sync { dry_run, force }, + .. + } => run_sync(cli, *dry_run, *force), + Cli { command, .. } => unimplemented(command), } } @@ -32,7 +47,79 @@ fn run_doctor(cli: &Cli) -> ExitCode { } } -/// M0 ships the skeleton: the command surface is defined, but only `doctor` acts. +fn run_sync(cli: &Cli, dry_run: bool, force: bool) -> ExitCode { + let (config, _) = match Config::load(cli.config.as_deref()) { + Ok(loaded) => loaded, + Err(error) => return fail(&error), + }; + let state_dir = match paths::state_dir() { + Ok(dir) => dir, + Err(error) => return fail(&error), + }; + match sync::run(&config, &state_dir, dry_run, force) { + Ok(report) => { + print_report(&report); + ExitCode::SUCCESS + } + Err(error) => fail(&error), + } +} + +fn print_report(report: &Report) { + if report.aggregates.is_empty() { + println!("no aggregates configured"); + return; + } + for aggregate in &report.aggregates { + println!( + "{}: {} mirrored, {} written back, {} removed from the aggregate, {} removed from sources", + aggregate.id, + aggregate.mirrored, + aggregate.written_back, + aggregate.deleted_from_aggregate, + aggregate.deleted_from_sources, + ); + for conflict in &aggregate.conflicts { + println!( + " conflict: {} changed in both places; kept the version from `{}`", + conflict.source_uid, conflict.source_id + ); + } + for skipped in &aggregate.skipped { + println!(" skipped: {}", describe(skipped)); + } + } + if report.dry_run { + println!("(dry run — nothing was written)"); + } +} + +fn describe(skipped: &reconcile::Skipped) -> String { + match skipped { + reconcile::Skipped::NoSink { aggregate_uid } => format!( + "`{aggregate_uid}` was created in the aggregate, but no default_sink is set to route it to" + ), + reconcile::Skipped::ReadOnlySource { + aggregate_uid, + source_id, + } => format!( + "`{aggregate_uid}` was edited, but its source `{source_id}` is a read-only feed" + ), + } +} + +/// Prints an error with its full cause chain, which is where the useful detail is. +fn fail(error: &dyn std::error::Error) -> ExitCode { + eprintln!("calcalist: {error}"); + let mut source = error.source(); + while let Some(cause) = source { + eprintln!(" caused by: {cause}"); + source = cause.source(); + } + ExitCode::FAILURE +} + +/// M1 is still landing: `serve`, `status`, `google` and `aggregate` come later. fn unimplemented(command: &Command) -> ExitCode { let name = match command { Command::Sync { .. } => "sync", @@ -42,6 +129,6 @@ fn unimplemented(command: &Command) -> ExitCode { Command::Aggregate { .. } => "aggregate", Command::Doctor => "doctor", }; - eprintln!("calcalist: `{name}` is not implemented yet (M0 provides `doctor` only)"); + eprintln!("calcalist: `{name}` is not implemented yet"); ExitCode::from(2) } diff --git a/src/mirror.rs b/src/mirror.rs new file mode 100644 index 0000000..a7e2e00 --- /dev/null +++ b/src/mirror.rs @@ -0,0 +1,396 @@ +//! Transforming events between a source calendar and its aggregate copy. +//! +//! Mirroring is not a byte copy. The aggregate copy must be *scheduling inert* — +//! writing it must never cause a server to mail invitations or cancellations for +//! a meeting that was already invited from its source. How that is achieved +//! depends on the target backend, which is what [`SchedulingSuppression`] selects. +//! +//! Writing an edit back to the source is the inverse, with one asymmetry: the +//! demoted form has thrown away the live guest list, so the source event itself +//! is used as the donor to put it back. Otherwise editing a meeting's time in the +//! aggregate would silently drop its guests. + +use crate::config::SchedulingSuppression; +use crate::ical::Calendar; + +/// Records which source an aggregate copy came from. +pub const SOURCE_PROPERTY: &str = "X-CALCALIST-SOURCE"; +/// Records the UID the event has in its source calendar. +pub const ORIGIN_UID_PROPERTY: &str = "X-CALCALIST-ORIGIN-UID"; +/// Carries the guest list inertly when it cannot be kept as real attendees. +pub const ATTENDEES_PROPERTY: &str = "X-CALCALIST-ATTENDEES"; + +/// Separates the original description from the appended guest list, so the +/// addition can be found and removed again when writing an edit back. +pub const GUEST_MARKER: &str = "-- guests (calcalist) --"; + +/// Properties that only ever exist on an aggregate copy. +const OWN_PROPERTIES: &[&str] = &[SOURCE_PROPERTY, ORIGIN_UID_PROPERTY, ATTENDEES_PROPERTY]; + +/// Live scheduling properties, whose presence is what makes a server send mail. +const SCHEDULING_PROPERTIES: &[&str] = &["ATTENDEE", "ORGANIZER"]; + +/// Builds the aggregate copy of a source event. +pub fn to_aggregate( + source: &Calendar, + aggregate_uid: &str, + source_id: &str, + source_uid: &str, + owner: Option<&str>, + suppression: SchedulingSuppression, +) -> Calendar { + let mut mirrored = source.clone(); + mirrored.set_uid(aggregate_uid); + mirrored.remove_properties("VEVENT", OWN_PROPERTIES); + mirrored.add_property("VEVENT", SOURCE_PROPERTY, source_id); + mirrored.add_property("VEVENT", ORIGIN_UID_PROPERTY, source_uid); + + if suppression == SchedulingSuppression::Native { + return mirrored; + } + demote_attendees(&mut mirrored, owner); + mirrored +} + +/// Removes the live guest list, keeping its information in inert form. +fn demote_attendees(calendar: &mut Calendar, owner: Option<&str>) { + let guests: Vec = calendar + .properties("VEVENT", "ATTENDEE") + .map(|attendee| { + let name = attendee + .param("CN") + .unwrap_or_else(|| address(attendee.value)); + match attendee.param("PARTSTAT") { + Some(status) => format!("{name} <{}> ({status})", address(attendee.value)), + None => format!("{name} <{}>", address(attendee.value)), + } + }) + .collect(); + + // A meeting the owner declined must not read as busy in the aggregate. + if owner.is_some_and(|owner| declined(calendar, owner)) { + calendar.set_property("VEVENT", "TRANSP", "TRANSPARENT"); + } + + if !guests.is_empty() { + calendar.add_property( + "VEVENT", + ATTENDEES_PROPERTY, + &escape_text(&guests.join(", ")), + ); + append_guests_to_description(calendar, &guests); + } + calendar.remove_properties("VEVENT", SCHEDULING_PROPERTIES); +} + +/// Whether `owner` is an attendee who has declined. +fn declined(calendar: &Calendar, owner: &str) -> bool { + calendar.properties("VEVENT", "ATTENDEE").any(|attendee| { + address(attendee.value).eq_ignore_ascii_case(owner) + && attendee.param("PARTSTAT") == Some("DECLINED") + }) +} + +fn append_guests_to_description(calendar: &mut Calendar, guests: &[String]) { + let existing = calendar + .properties("VEVENT", "DESCRIPTION") + .next() + .map(|property| property.value.to_string()) + .unwrap_or_default(); + let block = format!("{GUEST_MARKER}\\n{}", escape_text(&guests.join("\n"))); + let combined = if existing.is_empty() { + block + } else { + format!("{existing}\\n\\n{block}") + }; + calendar.set_property("VEVENT", "DESCRIPTION", &combined); +} + +/// Rebuilds a source-side event from an edited aggregate copy. +/// +/// `donor` is the event as it currently stands in the source calendar, when +/// there is one. It supplies the properties the mirror transform removed, so an +/// edit made in the aggregate cannot silently strip a meeting's guests. Routing +/// a newly created event has no donor: nothing was stripped from it. +pub fn to_source(edited: &Calendar, donor: Option<&Calendar>, source_uid: &str) -> Calendar { + let mut restored = edited.clone(); + restored.set_uid(source_uid); + restored.remove_properties("VEVENT", OWN_PROPERTIES); + strip_guest_block(&mut restored); + + // If the mirror kept live attendees, the edit owns them. If it demoted them, + // they are missing here and must come back from the source. + if let Some(donor) = donor + && restored.properties("VEVENT", "ATTENDEE").next().is_none() + { + for line in donor.property_lines("VEVENT", "ORGANIZER") { + restored.add_raw_property("VEVENT", line); + } + for line in donor.property_lines("VEVENT", "ATTENDEE") { + restored.add_raw_property("VEVENT", line); + } + } + restored +} + +/// Removes the appended guest block from DESCRIPTION, leaving any real text. +fn strip_guest_block(calendar: &mut Calendar) { + let Some(description) = calendar + .properties("VEVENT", "DESCRIPTION") + .next() + .map(|property| property.value.to_string()) + else { + return; + }; + let Some(index) = description.find(GUEST_MARKER) else { + return; + }; + // Strip exactly the separator this transform inserted. Trimming a character + // set here would eat a trailing "n" from real text ("Plan" -> "Pla"). + let before = &description[..index]; + let kept = before.strip_suffix("\\n\\n").unwrap_or(before).to_string(); + if kept.is_empty() { + calendar.remove_properties("VEVENT", &["DESCRIPTION"]); + } else { + calendar.set_property("VEVENT", "DESCRIPTION", &kept); + } +} + +/// Strips a `mailto:` (or other) scheme from a calendar user address. +fn address(value: &str) -> &str { + value.split_once(':').map_or(value, |(_, rest)| rest) +} + +/// Escapes a value for an iCalendar TEXT property, per RFC 5545 section 3.3.11. +fn escape_text(value: &str) -> String { + let mut escaped = String::with_capacity(value.len()); + for ch in value.chars() { + match ch { + '\\' => escaped.push_str(r"\\"), + ';' => escaped.push_str(r"\;"), + ',' => escaped.push_str(r"\,"), + '\n' => escaped.push_str(r"\n"), + _ => escaped.push(ch), + } + } + escaped +} + +#[cfg(test)] +mod tests { + use super::*; + + const OWNER: &str = "me@example.com"; + + fn source_event(extra: &str) -> Calendar { + let text = format!( + "BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:event-1@example.com\r\nDTSTART:20260910T090000Z\r\nSUMMARY:Weekly sync\r\n{extra}BEGIN:VALARM\r\nACTION:DISPLAY\r\nTRIGGER:-PT15M\r\nEND:VALARM\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n" + ); + Calendar::parse(&text).expect("fixture should parse") + } + + const GUESTS: &str = concat!( + "ORGANIZER;CN=Boss:mailto:boss@example.com\r\n", + "ATTENDEE;CN=Me;PARTSTAT=ACCEPTED:mailto:me@example.com\r\n", + "ATTENDEE;CN=Them;PARTSTAT=NEEDS-ACTION:mailto:them@example.com\r\n", + ); + + fn mirror(source: &Calendar, suppression: SchedulingSuppression) -> Calendar { + to_aggregate( + source, + "deadbeef@calcalist", + "work", + "event-1@example.com", + Some(OWNER), + suppression, + ) + } + + #[test] + fn a_mirror_records_where_it_came_from() { + let mirrored = mirror(&source_event(""), SchedulingSuppression::Native); + assert_eq!(mirrored.uid(), Some("deadbeef@calcalist")); + assert_eq!( + mirrored + .properties("VEVENT", SOURCE_PROPERTY) + .next() + .map(|p| p.value), + Some("work") + ); + assert_eq!( + mirrored + .properties("VEVENT", ORIGIN_UID_PROPERTY) + .next() + .map(|p| p.value), + Some("event-1@example.com") + ); + } + + /// Google can suppress notification, so the guest list survives intact. + #[test] + fn a_suppressible_backend_keeps_real_attendees() { + let mirrored = mirror(&source_event(GUESTS), SchedulingSuppression::Native); + assert_eq!(mirrored.properties("VEVENT", "ATTENDEE").count(), 2); + assert_eq!(mirrored.properties("VEVENT", "ORGANIZER").count(), 1); + assert_eq!(mirrored.properties("VEVENT", ATTENDEES_PROPERTY).count(), 0); + } + + /// CalDAV cannot, so the live properties must go. + #[test] + fn an_unsuppressible_backend_emits_no_live_scheduling_properties() { + let mirrored = mirror(&source_event(GUESTS), SchedulingSuppression::None); + assert_eq!(mirrored.properties("VEVENT", "ATTENDEE").count(), 0); + assert_eq!(mirrored.properties("VEVENT", "ORGANIZER").count(), 0); + let output = mirrored.to_ics(); + assert!(!output.contains("\r\nATTENDEE")); + assert!(!output.contains("\r\nORGANIZER")); + } + + #[test] + fn the_demoted_guest_list_is_still_readable() { + let mirrored = mirror(&source_event(GUESTS), SchedulingSuppression::None); + let inert = mirrored + .properties("VEVENT", ATTENDEES_PROPERTY) + .next() + .expect("guest list carried") + .value; + assert!(inert.contains("Me (ACCEPTED)")); + assert!(inert.contains("Them (NEEDS-ACTION)")); + + let description = mirrored + .properties("VEVENT", "DESCRIPTION") + .next() + .expect("description added") + .value; + assert!(description.contains(GUEST_MARKER)); + } + + /// A meeting the owner declined should read as free, not busy. + #[test] + fn declining_marks_the_mirror_transparent() { + let declined_by_owner = concat!( + "ATTENDEE;CN=Me;PARTSTAT=DECLINED:mailto:me@example.com\r\n", + "ATTENDEE;CN=Them;PARTSTAT=ACCEPTED:mailto:them@example.com\r\n", + ); + let mirrored = mirror( + &source_event(declined_by_owner), + SchedulingSuppression::None, + ); + assert_eq!( + mirrored + .properties("VEVENT", "TRANSP") + .next() + .map(|p| p.value), + Some("TRANSPARENT") + ); + } + + /// Someone else declining says nothing about the owner's availability. + #[test] + fn another_guest_declining_does_not_free_the_owner() { + let other_declined = concat!( + "ATTENDEE;CN=Me;PARTSTAT=ACCEPTED:mailto:me@example.com\r\n", + "ATTENDEE;CN=Them;PARTSTAT=DECLINED:mailto:them@example.com\r\n", + ); + let mirrored = mirror(&source_event(other_declined), SchedulingSuppression::None); + assert_eq!(mirrored.properties("VEVENT", "TRANSP").count(), 0); + } + + #[test] + fn alarms_are_never_touched() { + for suppression in [SchedulingSuppression::Native, SchedulingSuppression::None] { + let mirrored = mirror(&source_event(GUESTS), suppression); + assert!(mirrored.to_ics().contains("BEGIN:VALARM")); + assert!(mirrored.to_ics().contains("TRIGGER:-PT15M")); + } + } + + #[test] + fn mirroring_twice_is_stable() { + let source = source_event(GUESTS); + let once = mirror(&source, SchedulingSuppression::None); + let twice = mirror(&source, SchedulingSuppression::None); + assert_eq!(once.to_ics(), twice.to_ics()); + } + + #[test] + fn writing_back_restores_the_source_uid_and_drops_our_properties() { + let source = source_event(GUESTS); + let mirrored = mirror(&source, SchedulingSuppression::None); + let back = to_source(&mirrored, Some(&source), "event-1@example.com"); + + assert_eq!(back.uid(), Some("event-1@example.com")); + for property in OWN_PROPERTIES { + assert_eq!(back.properties("VEVENT", property).count(), 0, "{property}"); + } + } + + /// The demoted mirror has no guests to give back, so editing it must not + /// silently strip the meeting's attendees. + #[test] + fn writing_back_a_demoted_mirror_recovers_the_guest_list() { + let source = source_event(GUESTS); + let mut mirrored = mirror(&source, SchedulingSuppression::None); + mirrored.set_property("VEVENT", "SUMMARY", "Weekly sync (moved)"); + + let back = to_source(&mirrored, Some(&source), "event-1@example.com"); + assert_eq!(back.properties("VEVENT", "ATTENDEE").count(), 2); + assert_eq!(back.properties("VEVENT", "ORGANIZER").count(), 1); + // The edit itself survives. + assert_eq!( + back.properties("VEVENT", "SUMMARY").next().map(|p| p.value), + Some("Weekly sync (moved)") + ); + // And the parameters came back with them. + let attendee = back + .properties("VEVENT", "ATTENDEE") + .find(|a| a.value.contains("them@")) + .expect("guest restored"); + assert_eq!(attendee.param("PARTSTAT"), Some("NEEDS-ACTION")); + } + + #[test] + fn writing_back_removes_the_appended_guest_block() { + let source = source_event(GUESTS); + let mirrored = mirror(&source, SchedulingSuppression::None); + let back = to_source(&mirrored, Some(&source), "event-1@example.com"); + assert_eq!(back.properties("VEVENT", "DESCRIPTION").count(), 0); + } + + #[test] + fn a_real_description_survives_the_round_trip() { + let with_description = format!("DESCRIPTION:Bring the plan\r\n{GUESTS}"); + let source = source_event(&with_description); + let mirrored = mirror(&source, SchedulingSuppression::None); + let back = to_source(&mirrored, Some(&source), "event-1@example.com"); + assert_eq!( + back.properties("VEVENT", "DESCRIPTION") + .next() + .map(|p| p.value), + Some("Bring the plan") + ); + } + + /// Regression: trimming a character set would truncate text ending in "n". + #[test] + fn a_description_ending_in_n_is_not_truncated() { + let with_description = format!("DESCRIPTION:Bring the plan\r\n{GUESTS}"); + let source = source_event(&with_description); + let mirrored = mirror(&source, SchedulingSuppression::None); + let back = to_source(&mirrored, Some(&source), "event-1@example.com"); + let description = back + .properties("VEVENT", "DESCRIPTION") + .next() + .expect("description kept") + .value; + assert!( + description.ends_with("plan"), + "truncated to {description:?}" + ); + } + + #[test] + fn text_escaping_follows_the_specification() { + assert_eq!(escape_text("a,b;c\\d\ne"), r"a\,b\;c\\d\ne"); + } +} diff --git a/src/provenance.rs b/src/provenance.rs new file mode 100644 index 0000000..8217672 --- /dev/null +++ b/src/provenance.rs @@ -0,0 +1,102 @@ +//! Deterministic identity for mirrored events. +//! +//! An aggregate copy's UID is derived from the aggregate, the source it came from +//! and the source's own UID. Two properties follow, and the design leans on both: +//! the state file is a cache that can be rebuilt rather than a single point of +//! failure, and a UID that derives is recognisably ours, so mirrored events are +//! never re-ingested as if a user had written them. +//! +//! The derivation deliberately excludes the aggregate's target, so moving an +//! aggregate to a different endpoint leaves every UID unchanged. + +use std::fmt; + +/// Suffix marking a UID as produced by calcalist. +pub const UID_SUFFIX: &str = "@calcalist"; + +/// Hex characters taken from the hash: 16 bytes, which is ample for identity. +const HEX_LEN: usize = 32; + +/// A UID for an event mirrored into an aggregate. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct DerivedUid(String); + +impl fmt::Display for DerivedUid { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +/// Derives the aggregate-side UID for one source event. +pub fn derive_uid(aggregate_id: &str, source_id: &str, source_uid: &str) -> DerivedUid { + let mut hasher = blake3::Hasher::new(); + for field in [aggregate_id, source_id, source_uid] { + absorb(&mut hasher, field); + } + let hash = hasher.finalize(); + let hex: String = hash + .as_bytes() + .iter() + .take(HEX_LEN / 2) + .map(|byte| format!("{byte:02x}")) + .collect(); + DerivedUid(format!("{hex}{UID_SUFFIX}")) +} + +/// Whether a UID looks like one we generated, and so belongs to a mirrored copy +/// rather than to an event a user created in the aggregate. +pub fn is_derived(uid: &str) -> bool { + let Some(hex) = uid.strip_suffix(UID_SUFFIX) else { + return false; + }; + hex.len() == HEX_LEN && hex.bytes().all(|byte| byte.is_ascii_hexdigit()) +} + +/// Length-prefixes each field so that differing splits cannot collide: +/// without this, `("ab", "c")` and `("a", "bc")` would hash identically. +fn absorb(hasher: &mut blake3::Hasher, field: &str) { + hasher.update(&(field.len() as u64).to_le_bytes()); + hasher.update(field.as_bytes()); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn derivation_is_stable() { + let first = derive_uid("unified", "work", "event-1@example.com"); + let second = derive_uid("unified", "work", "event-1@example.com"); + assert_eq!(first, second); + } + + #[test] + fn each_field_changes_the_result() { + let base = derive_uid("unified", "work", "event-1"); + assert_ne!(base, derive_uid("other", "work", "event-1")); + assert_ne!(base, derive_uid("unified", "home", "event-1")); + assert_ne!(base, derive_uid("unified", "work", "event-2")); + } + + /// The field boundaries must matter, or two different events could collide. + #[test] + fn field_boundaries_are_unambiguous() { + assert_ne!(derive_uid("a", "bc", "d"), derive_uid("ab", "c", "d")); + assert_ne!(derive_uid("a", "b", "cd"), derive_uid("a", "bc", "d")); + } + + #[test] + fn derived_uids_are_recognised() { + let uid = derive_uid("unified", "work", "event-1"); + assert!(is_derived(&uid.to_string())); + } + + #[test] + fn foreign_uids_are_not_recognised() { + assert!(!is_derived("event-1@example.com")); + assert!(!is_derived("@calcalist")); + assert!(!is_derived("nothex000000000000000000000000zz@calcalist")); + // Right shape, wrong length. + assert!(!is_derived("abc123@calcalist")); + } +} diff --git a/src/reconcile.rs b/src/reconcile.rs new file mode 100644 index 0000000..53e0201 --- /dev/null +++ b/src/reconcile.rs @@ -0,0 +1,872 @@ +//! The aggregation engine: deciding what must change, without changing anything. +//! +//! This module is pure. It takes a snapshot of every source vdir, the aggregate +//! vdir and the previous sync's state, and returns the actions that would bring +//! them into agreement. Nothing here touches the filesystem or the network, which +//! is what makes the interesting cases — a conflicting edit, a deleted mirror, a +//! source that vanished — cheap to test exhaustively. +//! +//! Change is detected by comparing content hashes against the ones recorded at +//! the end of the last cycle, which is what distinguishes "the source changed" +//! from "the aggregate changed" and so from a genuine conflict. + +use std::collections::{BTreeMap, HashSet}; +use std::path::PathBuf; + +use thiserror::Error; + +use crate::config::SchedulingSuppression; +use crate::ical::Calendar; +use crate::mirror; +use crate::provenance::{self, derive_uid}; +use crate::state::{AggregateState, Link}; +use crate::vdir::Item; + +/// One source calendar as it currently stands on disk. +#[derive(Debug, Clone, Copy)] +pub struct SourceView<'a> { + pub id: &'a str, + /// iCal feeds cannot be written to, so edits and deletions cannot go back. + pub writable: bool, + /// The calendar owner's address, for telling their attendance from others'. + pub owner: Option<&'a str>, + pub items: &'a BTreeMap, +} + +#[derive(Debug, Clone, Copy)] +pub struct Policy<'a> { + pub aggregate_id: &'a str, + pub suppression: SchedulingSuppression, + pub default_sink: Option<&'a str>, + pub propagate_deletes: bool, + pub max_delete_fraction: f64, + /// Set by `--force`, to proceed past the mass-deletion guard. + pub force: bool, +} + +/// A change to make on disk. Applying these is somebody else's job. +#[derive(Debug, Clone, PartialEq)] +pub enum Action { + WriteAggregate { + uid: String, + calendar: Calendar, + /// Path the item currently occupies, so it is replaced rather than duplicated. + replaces: Option, + }, + DeleteAggregate { + uid: String, + path: PathBuf, + }, + WriteSource { + source_id: String, + uid: String, + calendar: Calendar, + /// Path the item currently occupies, so it is replaced rather than duplicated. + replaces: Option, + }, + DeleteSource { + source_id: String, + uid: String, + path: PathBuf, + }, +} + +/// Both sides changed since the last sync. The source wins; this records that it +/// happened so the user can be told. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Conflict { + pub aggregate_uid: String, + pub source_id: String, + pub source_uid: String, +} + +/// Something that could not be done, and why. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Skipped { + /// A user created an event in the aggregate, but no sink is configured to + /// route it to, and guessing one would put it in the wrong calendar. + NoSink { aggregate_uid: String }, + /// An edit was made to a mirror of a read-only feed, which cannot accept it. + ReadOnlySource { + aggregate_uid: String, + source_id: String, + }, +} + +#[derive(Debug)] +pub struct Outcome { + pub actions: Vec, + /// The links to record once the actions have been applied. + pub links: BTreeMap, + pub conflicts: Vec, + pub skipped: Vec, +} + +#[derive(Debug, Error, PartialEq)] +pub enum ReconcileError { + #[error( + "refusing to delete {requested} of {tracked} tracked event(s) ({percent:.0}%, limit {limit:.0}%); \ + re-run with --force if this is intended" + )] + MassDeletion { + requested: usize, + tracked: usize, + percent: f64, + limit: f64, + }, +} + +/// Works out what must change to bring the aggregate and its sources into agreement. +pub fn reconcile( + policy: Policy<'_>, + sources: &[SourceView<'_>], + aggregate: &BTreeMap, + prior: Option<&AggregateState>, +) -> Result { + let empty = BTreeMap::new(); + let prior_links = prior.map_or(&empty, |state| &state.links); + + let mut outcome = Outcome { + actions: Vec::new(), + links: BTreeMap::new(), + conflicts: Vec::new(), + skipped: Vec::new(), + }; + let mut live: HashSet = HashSet::new(); + let mut deletions = 0usize; + + for source in sources { + mirror_source( + policy, + source, + aggregate, + prior_links, + &mut outcome, + &mut live, + &mut deletions, + ); + } + retire_vanished_sources(aggregate, prior_links, &live, &mut outcome, &mut deletions); + route_new_events(policy, sources, aggregate, &mut outcome); + + guard_against_mass_deletion(policy, prior_links.len(), deletions)?; + Ok(outcome) +} + +/// Brings each event of one source into agreement with its aggregate copy. +fn mirror_source( + policy: Policy<'_>, + source: &SourceView<'_>, + aggregate: &BTreeMap, + prior_links: &BTreeMap, + outcome: &mut Outcome, + live: &mut HashSet, + deletions: &mut usize, +) { + for (source_uid, item) in source.items { + // A mirror that has found its way into a source is not a source event. + // Skipping it is what stops our own writes echoing back around. + if provenance::is_derived(source_uid) { + continue; + } + let aggregate_uid = derive_uid(policy.aggregate_id, source.id, source_uid).to_string(); + let desired = mirror::to_aggregate( + &item.calendar, + &aggregate_uid, + source.id, + source_uid, + source.owner, + policy.suppression, + ); + + let Some(link) = prior_links.get(&aggregate_uid) else { + // Not seen before: a new source event needs a mirror. + record( + outcome, + &aggregate_uid, + link_to(source, source_uid, item, &desired), + Mirror { + calendar: &desired, + replaces: aggregate.get(&aggregate_uid).map(|held| held.path.clone()), + }, + ); + live.insert(aggregate_uid); + continue; + }; + live.insert(aggregate_uid.clone()); + + let source_changed = link.source_hash != item.hash; + let Some(current) = aggregate.get(&aggregate_uid) else { + // The mirror is gone: somebody deleted it in the aggregate. + if policy.propagate_deletes && source.writable { + outcome.actions.push(Action::DeleteSource { + source_id: source.id.to_string(), + uid: source_uid.clone(), + path: item.path.clone(), + }); + *deletions += 1; + live.remove(&aggregate_uid); + } else { + record( + outcome, + &aggregate_uid, + link_to(source, source_uid, item, &desired), + Mirror { + calendar: &desired, + replaces: aggregate.get(&aggregate_uid).map(|held| held.path.clone()), + }, + ); + } + continue; + }; + + let aggregate_changed = current.hash != link.aggregate_hash; + match (source_changed, aggregate_changed) { + (false, false) => { + outcome.links.insert(aggregate_uid, link.clone()); + } + (true, true) => { + // Both moved. The origin calendar is authoritative. + outcome.conflicts.push(Conflict { + aggregate_uid: aggregate_uid.clone(), + source_id: source.id.to_string(), + source_uid: source_uid.clone(), + }); + record( + outcome, + &aggregate_uid, + link_to(source, source_uid, item, &desired), + Mirror { + calendar: &desired, + replaces: aggregate.get(&aggregate_uid).map(|held| held.path.clone()), + }, + ); + } + (true, false) => { + record( + outcome, + &aggregate_uid, + link_to(source, source_uid, item, &desired), + Mirror { + calendar: &desired, + replaces: aggregate.get(&aggregate_uid).map(|held| held.path.clone()), + }, + ); + } + (false, true) => { + write_back( + policy, + source, + &aggregate_uid, + source_uid, + item, + current, + outcome, + ); + } + } + } +} + +/// Sends an aggregate-side edit back to the calendar the event came from. +fn write_back( + _policy: Policy<'_>, + source: &SourceView<'_>, + aggregate_uid: &str, + source_uid: &str, + item: &Item, + current: &Item, + outcome: &mut Outcome, +) { + if !source.writable { + // A feed cannot take the edit, and silently reverting would lose it + // without explanation, so say so instead. + outcome.skipped.push(Skipped::ReadOnlySource { + aggregate_uid: aggregate_uid.to_string(), + source_id: source.id.to_string(), + }); + outcome.links.insert( + aggregate_uid.to_string(), + Link { + source_id: source.id.to_string(), + source_uid: source_uid.to_string(), + source_hash: item.hash.clone(), + aggregate_hash: current.hash.clone(), + }, + ); + return; + } + let restored = mirror::to_source(¤t.calendar, Some(&item.calendar), source_uid); + let source_hash = restored.content_hash(); + outcome.actions.push(Action::WriteSource { + source_id: source.id.to_string(), + uid: source_uid.to_string(), + calendar: restored, + replaces: Some(item.path.clone()), + }); + outcome.links.insert( + aggregate_uid.to_string(), + Link { + source_id: source.id.to_string(), + source_uid: source_uid.to_string(), + source_hash, + aggregate_hash: current.hash.clone(), + }, + ); +} + +/// Removes mirrors whose source event no longer exists. +fn retire_vanished_sources( + aggregate: &BTreeMap, + prior_links: &BTreeMap, + live: &HashSet, + outcome: &mut Outcome, + deletions: &mut usize, +) { + for (aggregate_uid, _) in prior_links.iter().filter(|(uid, _)| !live.contains(*uid)) { + let Some(item) = aggregate.get(aggregate_uid) else { + continue; + }; + outcome.actions.push(Action::DeleteAggregate { + uid: aggregate_uid.clone(), + path: item.path.clone(), + }); + *deletions += 1; + } +} + +/// Sends events created directly in the aggregate to the configured sink, and +/// replaces them with the canonical mirror so no duplicate is left behind. +fn route_new_events( + policy: Policy<'_>, + sources: &[SourceView<'_>], + aggregate: &BTreeMap, + outcome: &mut Outcome, +) { + let user_created = aggregate + .iter() + .filter(|(uid, _)| !provenance::is_derived(uid)); + + for (uid, item) in user_created { + let sink = policy + .default_sink + .and_then(|sink| sources.iter().find(|source| source.id == sink)) + .filter(|sink| sink.writable); + let Some(sink) = sink else { + outcome.skipped.push(Skipped::NoSink { + aggregate_uid: uid.clone(), + }); + continue; + }; + + // Attendees are kept here deliberately: the user is organising a meeting, + // and the sink server inviting the guests is the intended behaviour. + let routed = mirror::to_source(&item.calendar, None, uid); + let source_hash = routed.content_hash(); + let aggregate_uid = derive_uid(policy.aggregate_id, sink.id, uid).to_string(); + let canonical = mirror::to_aggregate( + &routed, + &aggregate_uid, + sink.id, + uid, + sink.owner, + policy.suppression, + ); + + outcome.actions.push(Action::WriteSource { + source_id: sink.id.to_string(), + uid: uid.clone(), + calendar: routed, + replaces: None, + }); + outcome.links.insert( + aggregate_uid.clone(), + Link { + source_id: sink.id.to_string(), + source_uid: uid.clone(), + source_hash, + aggregate_hash: canonical.content_hash(), + }, + ); + outcome.actions.push(Action::WriteAggregate { + uid: aggregate_uid, + calendar: canonical, + replaces: None, + }); + // The user's own copy is replaced by the canonical mirror, so that the + // event does not appear twice. + outcome.actions.push(Action::DeleteAggregate { + uid: uid.clone(), + path: item.path.clone(), + }); + } +} + +/// Deletions always permitted regardless of the fraction. +/// +/// A share on its own is meaningless when the counts are small: deleting the only +/// tracked event is 100% of them, and refusing that would make the guard fire on +/// the most ordinary action there is. The guard is meant to catch a source vdir +/// that failed to populate, which shows up as a *bulk* disappearance. +const ALWAYS_ALLOWED_DELETIONS: usize = 3; + +/// Refuses a cycle that would delete an implausible share of the aggregate. +/// +/// A source vdir that failed to populate looks exactly like one whose events were +/// all deleted, and propagating that would destroy the originals. +fn guard_against_mass_deletion( + policy: Policy<'_>, + tracked: usize, + deletions: usize, +) -> Result<(), ReconcileError> { + if policy.force || tracked == 0 || deletions <= ALWAYS_ALLOWED_DELETIONS { + return Ok(()); + } + let fraction = deletions as f64 / tracked as f64; + if fraction <= policy.max_delete_fraction { + return Ok(()); + } + Err(ReconcileError::MassDeletion { + requested: deletions, + tracked, + percent: fraction * 100.0, + limit: policy.max_delete_fraction * 100.0, + }) +} + +/// The link describing a freshly mirrored source event. +fn link_to(source: &SourceView<'_>, source_uid: &str, item: &Item, desired: &Calendar) -> Link { + Link { + source_id: source.id.to_string(), + source_uid: source_uid.to_string(), + source_hash: item.hash.clone(), + aggregate_hash: desired.content_hash(), + } +} + +/// Queues a mirror write and the link that will describe it once applied. +fn record(outcome: &mut Outcome, aggregate_uid: &str, link: Link, mirror: Mirror<'_>) { + outcome.links.insert(aggregate_uid.to_string(), link); + outcome.actions.push(Action::WriteAggregate { + uid: aggregate_uid.to_string(), + calendar: mirror.calendar.clone(), + replaces: mirror.replaces, + }); +} + +/// The aggregate copy to write, and the file it supersedes if one exists. +struct Mirror<'a> { + calendar: &'a Calendar, + replaces: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::Path; + + const AGG: &str = "unified"; + const SRC: &str = "work"; + + fn calendar(uid: &str, summary: &str) -> Calendar { + Calendar::parse(&format!( + "BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:{uid}\r\nDTSTART:20260910T090000Z\r\nSUMMARY:{summary}\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n" + )) + .expect("fixture should parse") + } + + fn item(uid: &str, summary: &str) -> Item { + let calendar = calendar(uid, summary); + Item { + uid: uid.to_string(), + path: Path::new("/tmp/calcalist-test").join(format!("{uid}.ics")), + hash: calendar.content_hash(), + calendar, + } + } + + fn map(items: Vec) -> BTreeMap { + items + .into_iter() + .map(|item| (item.uid.clone(), item)) + .collect() + } + + fn derived_uid(source_uid: &str) -> String { + derive_uid(AGG, SRC, source_uid).to_string() + } + + /// The aggregate copy that mirroring `source_uid` would produce. + fn mirrored(source_uid: &str, summary: &str) -> Item { + let aggregate_uid = derived_uid(source_uid); + let calendar = mirror::to_aggregate( + &calendar(source_uid, summary), + &aggregate_uid, + SRC, + source_uid, + None, + SchedulingSuppression::None, + ); + Item { + uid: aggregate_uid, + path: Path::new("/tmp/calcalist-test/agg").join(format!("{source_uid}.ics")), + hash: calendar.content_hash(), + calendar, + } + } + + fn prior_with(source_uid: &str, summary: &str) -> AggregateState { + let source = item(source_uid, summary); + let aggregate = mirrored(source_uid, summary); + let mut links = BTreeMap::new(); + links.insert( + aggregate.uid.clone(), + Link { + source_id: SRC.into(), + source_uid: source_uid.into(), + source_hash: source.hash, + aggregate_hash: aggregate.hash, + }, + ); + AggregateState { + target: crate::state::Target { + endpoint: "gcal".into(), + kind: "caldav".into(), + }, + links, + } + } + + fn policy() -> Policy<'static> { + Policy { + aggregate_id: AGG, + suppression: SchedulingSuppression::None, + default_sink: None, + propagate_deletes: true, + max_delete_fraction: 0.5, + force: false, + } + } + + fn view<'a>(items: &'a BTreeMap, writable: bool) -> SourceView<'a> { + SourceView { + id: SRC, + writable, + owner: None, + items, + } + } + + #[test] + fn a_new_source_event_gains_a_mirror() { + let sources = map(vec![item("event-1", "Standup")]); + let aggregate = BTreeMap::new(); + let outcome = reconcile(policy(), &[view(&sources, true)], &aggregate, None).expect("ok"); + + assert_eq!(outcome.actions.len(), 1); + let uid = derived_uid("event-1"); + assert!(matches!( + &outcome.actions[0], + Action::WriteAggregate { uid: written, .. } if *written == uid + )); + assert_eq!(outcome.links.len(), 1); + assert_eq!(outcome.links[&uid].source_uid, "event-1"); + } + + #[test] + fn an_unchanged_event_produces_no_work() { + let sources = map(vec![item("event-1", "Standup")]); + let aggregate = map(vec![mirrored("event-1", "Standup")]); + let prior = prior_with("event-1", "Standup"); + + let outcome = + reconcile(policy(), &[view(&sources, true)], &aggregate, Some(&prior)).expect("ok"); + assert!(outcome.actions.is_empty(), "{:?}", outcome.actions); + assert_eq!(outcome.links.len(), 1); + } + + #[test] + fn a_changed_source_updates_the_mirror() { + let sources = map(vec![item("event-1", "Standup moved")]); + let aggregate = map(vec![mirrored("event-1", "Standup")]); + let prior = prior_with("event-1", "Standup"); + + let outcome = + reconcile(policy(), &[view(&sources, true)], &aggregate, Some(&prior)).expect("ok"); + assert!(matches!( + outcome.actions.as_slice(), + [Action::WriteAggregate { .. }] + )); + assert!(outcome.conflicts.is_empty()); + } + + /// Editing a mirrored event in the aggregate must reach the origin calendar, + /// or the aggregate would be read-only in practice. + #[test] + fn an_edit_in_the_aggregate_is_written_back_to_the_source() { + let sources = map(vec![item("event-1", "Standup")]); + let aggregate = map(vec![mirrored("event-1", "Standup rescheduled")]); + let prior = prior_with("event-1", "Standup"); + + let outcome = + reconcile(policy(), &[view(&sources, true)], &aggregate, Some(&prior)).expect("ok"); + match outcome.actions.as_slice() { + [ + Action::WriteSource { + source_id, + uid, + calendar, + .. + }, + ] => { + assert_eq!(source_id, SRC); + assert_eq!(uid, "event-1"); + assert_eq!(calendar.uid(), Some("event-1")); + assert!(calendar.to_ics().contains("SUMMARY:Standup rescheduled")); + } + other => panic!("expected a write back to the source, got {other:?}"), + } + } + + #[test] + fn when_both_sides_changed_the_source_wins_and_the_clash_is_recorded() { + let sources = map(vec![item("event-1", "Source version")]); + let aggregate = map(vec![mirrored("event-1", "Aggregate version")]); + let prior = prior_with("event-1", "Standup"); + + let outcome = + reconcile(policy(), &[view(&sources, true)], &aggregate, Some(&prior)).expect("ok"); + match outcome.actions.as_slice() { + [Action::WriteAggregate { calendar, .. }] => { + assert!(calendar.to_ics().contains("SUMMARY:Source version")); + } + other => panic!("expected the source to overwrite the aggregate, got {other:?}"), + } + assert_eq!( + outcome.conflicts, + vec![Conflict { + aggregate_uid: derived_uid("event-1"), + source_id: SRC.into(), + source_uid: "event-1".into(), + }] + ); + } + + #[test] + fn deleting_a_mirror_deletes_the_source_event() { + let sources = map(vec![item("event-1", "Standup")]); + let aggregate = BTreeMap::new(); + let prior = prior_with("event-1", "Standup"); + + let outcome = + reconcile(policy(), &[view(&sources, true)], &aggregate, Some(&prior)).expect("ok"); + assert!(matches!( + outcome.actions.as_slice(), + [Action::DeleteSource { uid, .. }] if uid == "event-1" + )); + assert!(outcome.links.is_empty()); + } + + #[test] + fn without_delete_propagation_the_mirror_comes_back_instead() { + let sources = map(vec![item("event-1", "Standup")]); + let aggregate = BTreeMap::new(); + let prior = prior_with("event-1", "Standup"); + let policy = Policy { + propagate_deletes: false, + ..policy() + }; + + let outcome = + reconcile(policy, &[view(&sources, true)], &aggregate, Some(&prior)).expect("ok"); + assert!(matches!( + outcome.actions.as_slice(), + [Action::WriteAggregate { .. }] + )); + } + + #[test] + fn an_event_deleted_at_the_source_loses_its_mirror() { + let sources = BTreeMap::new(); + let aggregate = map(vec![mirrored("event-1", "Standup")]); + let prior = prior_with("event-1", "Standup"); + + let outcome = + reconcile(policy(), &[view(&sources, true)], &aggregate, Some(&prior)).expect("ok"); + assert!(matches!( + outcome.actions.as_slice(), + [Action::DeleteAggregate { uid, .. }] if *uid == derived_uid("event-1") + )); + } + + /// A user event in the aggregate goes to the sink and is replaced by the + /// canonical mirror, so it does not end up listed twice. + #[test] + fn a_user_created_event_is_routed_to_the_sink() { + let sources = map(vec![]); + let aggregate = map(vec![item("hand-written@phone", "Dentist")]); + let policy = Policy { + default_sink: Some(SRC), + ..policy() + }; + + let outcome = reconcile(policy, &[view(&sources, true)], &aggregate, None).expect("ok"); + let written_to_source = outcome + .actions + .iter() + .any(|action| matches!(action, Action::WriteSource { uid, .. } if uid == "hand-written@phone")); + let canonical_written = outcome + .actions + .iter() + .any(|action| matches!(action, Action::WriteAggregate { .. })); + let original_removed = outcome + .actions + .iter() + .any(|action| matches!(action, Action::DeleteAggregate { uid, .. } if uid == "hand-written@phone")); + + assert!(written_to_source, "{:?}", outcome.actions); + assert!(canonical_written, "{:?}", outcome.actions); + assert!(original_removed, "{:?}", outcome.actions); + assert!(outcome.skipped.is_empty()); + } + + /// Guessing a sink would put the event in the wrong calendar, so refuse. + #[test] + fn without_a_sink_a_user_created_event_is_left_alone() { + let sources = map(vec![]); + let aggregate = map(vec![item("hand-written@phone", "Dentist")]); + + let outcome = reconcile(policy(), &[view(&sources, true)], &aggregate, None).expect("ok"); + assert!(outcome.actions.is_empty()); + assert_eq!( + outcome.skipped, + vec![Skipped::NoSink { + aggregate_uid: "hand-written@phone".into() + }] + ); + } + + /// Our own mirror appearing inside a source must not be mirrored again. + #[test] + fn mirrors_found_in_a_source_are_not_re_ingested() { + let echo = mirrored("event-1", "Standup"); + let sources = map(vec![Item { + uid: echo.uid.clone(), + path: echo.path.clone(), + hash: echo.hash.clone(), + calendar: echo.calendar.clone(), + }]); + let aggregate = BTreeMap::new(); + + let outcome = reconcile(policy(), &[view(&sources, true)], &aggregate, None).expect("ok"); + assert!(outcome.actions.is_empty(), "{:?}", outcome.actions); + assert!(outcome.links.is_empty()); + } + + /// A source vdir that failed to populate looks identical to one that was + /// emptied on purpose, so refuse rather than destroy the originals. + #[test] + fn an_implausible_number_of_deletions_is_refused() { + let names = [ + ("event-1", "One"), + ("event-2", "Two"), + ("event-3", "Three"), + ("event-4", "Four"), + ("event-5", "Five"), + ]; + let sources = BTreeMap::new(); + let aggregate = map(names.iter().map(|(uid, s)| mirrored(uid, s)).collect()); + let mut prior = prior_with("event-1", "One"); + for (uid, summary) in &names[1..] { + prior.links.extend(prior_with(uid, summary).links); + } + + let error = reconcile(policy(), &[view(&sources, true)], &aggregate, Some(&prior)) + .expect_err("should refuse"); + assert!(matches!( + error, + ReconcileError::MassDeletion { + requested: 5, + tracked: 5, + .. + } + )); + } + + /// Deleting a handful of events is ordinary use, whatever share of a small + /// calendar it represents. + #[test] + fn a_small_number_of_deletions_is_never_refused() { + let names = [("event-1", "One"), ("event-2", "Two"), ("event-3", "Three")]; + let sources = BTreeMap::new(); + let aggregate = map(names.iter().map(|(uid, s)| mirrored(uid, s)).collect()); + let mut prior = prior_with("event-1", "One"); + for (uid, summary) in &names[1..] { + prior.links.extend(prior_with(uid, summary).links); + } + + // Every tracked event goes, but three is below the floor. + let outcome = reconcile(policy(), &[view(&sources, true)], &aggregate, Some(&prior)) + .expect("a small deletion must not be refused"); + assert_eq!(outcome.actions.len(), 3); + } + + #[test] + fn force_overrides_the_deletion_guard() { + let names = [ + ("event-1", "One"), + ("event-2", "Two"), + ("event-3", "Three"), + ("event-4", "Four"), + ("event-5", "Five"), + ]; + let sources = BTreeMap::new(); + let aggregate = map(names.iter().map(|(uid, s)| mirrored(uid, s)).collect()); + let mut prior = prior_with("event-1", "One"); + for (uid, summary) in &names[1..] { + prior.links.extend(prior_with(uid, summary).links); + } + + let policy = Policy { + force: true, + ..policy() + }; + let outcome = + reconcile(policy, &[view(&sources, true)], &aggregate, Some(&prior)).expect("forced"); + assert_eq!(outcome.actions.len(), 5); + } + + /// An iCal feed cannot accept an edit; losing it silently would be worse + /// than saying so. + #[test] + fn an_edit_to_a_feed_mirror_is_reported_rather_than_lost() { + let sources = map(vec![item("event-1", "Standup")]); + let aggregate = map(vec![mirrored("event-1", "Edited locally")]); + let prior = prior_with("event-1", "Standup"); + + let outcome = + reconcile(policy(), &[view(&sources, false)], &aggregate, Some(&prior)).expect("ok"); + assert!(outcome.actions.is_empty(), "{:?}", outcome.actions); + assert_eq!( + outcome.skipped, + vec![Skipped::ReadOnlySource { + aggregate_uid: derived_uid("event-1"), + source_id: SRC.into(), + }] + ); + } + + #[test] + fn a_feed_event_is_never_deleted_by_removing_its_mirror() { + let sources = map(vec![item("event-1", "Standup")]); + let aggregate = BTreeMap::new(); + let prior = prior_with("event-1", "Standup"); + + let outcome = + reconcile(policy(), &[view(&sources, false)], &aggregate, Some(&prior)).expect("ok"); + assert!(matches!( + outcome.actions.as_slice(), + [Action::WriteAggregate { .. }] + )); + } +} diff --git a/src/state.rs b/src/state.rs new file mode 100644 index 0000000..daa9334 --- /dev/null +++ b/src/state.rs @@ -0,0 +1,267 @@ +//! The sync state sidecar. +//! +//! This is machine-local: mappings between source events and their aggregate +//! copies, the content hashes that drive change detection, and the target each +//! aggregate was last published to. It deliberately lives apart from the portable +//! config, which must be copyable between machines without carrying stale hashes. +//! +//! The file is a cache, not a system of record. Aggregate UIDs are derived, so a +//! lost state file costs a full re-materialisation rather than data. + +use std::collections::BTreeMap; +use std::fs; +use std::io::{self, Write}; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +pub const STATE_VERSION: u32 = 1; +pub const FILE_NAME: &str = "state.json"; + +#[derive(Debug, Error)] +pub enum StateError { + #[error("could not read {path}: {source}")] + Read { + path: PathBuf, + #[source] + source: io::Error, + }, + #[error("could not parse {path}: {source}")] + Parse { + path: PathBuf, + #[source] + source: serde_json::Error, + }, + #[error("could not write {path}: {source}")] + Write { + path: PathBuf, + #[source] + source: io::Error, + }, + #[error("state file {path} has version {found}, but this build writes {STATE_VERSION}")] + UnsupportedVersion { path: PathBuf, found: u32 }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct State { + pub version: u32, + #[serde(default)] + pub aggregates: BTreeMap, +} + +impl Default for State { + fn default() -> Self { + State { + version: STATE_VERSION, + aggregates: BTreeMap::new(), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AggregateState { + /// Where this aggregate was last published. Compared against the config on + /// every sync: a mismatch means the target was changed, which must never be + /// reconciled as though every event had been deleted. + pub target: Target, + /// Mirrored events, keyed by their derived aggregate UID. + #[serde(default)] + pub links: BTreeMap, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Target { + pub endpoint: String, + /// Backend kind, recorded because it decides the mirror transform: switching + /// between backends changes every event's rendered content. + pub kind: String, +} + +/// One source event and the aggregate copy mirroring it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Link { + pub source_id: String, + pub source_uid: String, + /// Content hash of the source event as of the last successful sync. + pub source_hash: String, + /// Content hash of the aggregate copy as of the last successful sync. + pub aggregate_hash: String, +} + +impl State { + /// Reads the state file. A missing file yields empty state: a first run is + /// not an error, and must not read as "everything was deleted". + pub fn load(path: &Path) -> Result { + let text = match fs::read_to_string(path) { + Ok(text) => text, + Err(error) if error.kind() == io::ErrorKind::NotFound => { + return Ok(State::default()); + } + Err(source) => { + return Err(StateError::Read { + path: path.to_path_buf(), + source, + }); + } + }; + let state: State = serde_json::from_str(&text).map_err(|source| StateError::Parse { + path: path.to_path_buf(), + source, + })?; + if state.version != STATE_VERSION { + return Err(StateError::UnsupportedVersion { + path: path.to_path_buf(), + found: state.version, + }); + } + Ok(state) + } + + /// Writes the state file atomically. + /// + /// A sync that crashes part way must leave the previous state intact, so the + /// next run repeats the cycle rather than reconciling against a half-written + /// mapping. Hence write to a temporary file, flush it to disk, then rename. + pub fn save(&self, path: &Path) -> Result<(), StateError> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).map_err(|source| StateError::Write { + path: parent.to_path_buf(), + source, + })?; + } + let text = serde_json::to_string_pretty(self).map_err(|error| StateError::Write { + path: path.to_path_buf(), + source: io::Error::other(error), + })?; + + let temporary = path.with_extension("json.tmp"); + let mut file = fs::File::create(&temporary).map_err(|source| StateError::Write { + path: temporary.clone(), + source, + })?; + file.write_all(text.as_bytes()) + .and_then(|()| file.sync_all()) + .map_err(|source| StateError::Write { + path: temporary.clone(), + source, + })?; + drop(file); + + fs::rename(&temporary, path).map_err(|source| StateError::Write { + path: path.to_path_buf(), + source, + }) + } + + pub fn aggregate(&self, id: &str) -> Option<&AggregateState> { + self.aggregates.get(id) + } + + /// The aggregate's entry, created against `target` if it has none yet. + pub fn aggregate_mut(&mut self, id: &str, target: Target) -> &mut AggregateState { + self.aggregates + .entry(id.to_string()) + .or_insert_with(|| AggregateState { + target, + links: BTreeMap::new(), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn target() -> Target { + Target { + endpoint: "gcal".into(), + kind: "google".into(), + } + } + + fn link() -> Link { + Link { + source_id: "work".into(), + source_uid: "event-1@example.com".into(), + source_hash: "aaaa".into(), + aggregate_hash: "bbbb".into(), + } + } + + #[test] + fn a_missing_file_is_empty_state_not_an_error() { + let dir = tempfile::tempdir().expect("temp dir"); + let state = State::load(&dir.path().join("state.json")).expect("first run"); + assert!(state.aggregates.is_empty()); + } + + #[test] + fn round_trips_through_the_file() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("state.json"); + + let mut state = State::default(); + state + .aggregate_mut("unified", target()) + .links + .insert("deadbeef@calcalist".into(), link()); + state.save(&path).expect("save"); + + let loaded = State::load(&path).expect("load"); + let aggregate = loaded.aggregate("unified").expect("aggregate present"); + assert_eq!(aggregate.target, target()); + assert_eq!(aggregate.links["deadbeef@calcalist"], link()); + } + + #[test] + fn saving_creates_missing_parent_directories() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("nested/deeper/state.json"); + State::default().save(&path).expect("save"); + assert!(path.exists()); + } + + #[test] + fn saving_leaves_no_temporary_file_behind() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("state.json"); + State::default().save(&path).expect("save"); + + let leftovers: Vec<_> = fs::read_dir(dir.path()) + .expect("read dir") + .filter_map(Result::ok) + .map(|entry| entry.file_name().to_string_lossy().into_owned()) + .filter(|name| name.ends_with(".tmp")) + .collect(); + assert!(leftovers.is_empty(), "left behind: {leftovers:?}"); + } + + #[test] + fn rejects_a_state_file_from_a_future_version() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("state.json"); + fs::write(&path, r#"{"version": 99, "aggregates": {}}"#).expect("write"); + + assert!(matches!( + State::load(&path), + Err(StateError::UnsupportedVersion { found: 99, .. }) + )); + } + + /// The recorded target is what makes drift detectable before reconciliation. + #[test] + fn the_recorded_target_survives_a_round_trip() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("state.json"); + let mut state = State::default(); + state.aggregate_mut("unified", target()); + state.save(&path).expect("save"); + + let loaded = State::load(&path).expect("load"); + assert_eq!( + loaded.aggregate("unified").expect("present").target, + target() + ); + } +} diff --git a/src/sync.rs b/src/sync.rs new file mode 100644 index 0000000..7bbca9f --- /dev/null +++ b/src/sync.rs @@ -0,0 +1,250 @@ +//! One synchronisation cycle. +//! +//! Every endpoint is represented by a local vdir, so this drives the reconciler +//! over local files and applies whatever it decides. Getting the events into and +//! out of those vdirs is the job of pimsync (CalDAV, WebCal) and the Google +//! module, which bracket this step. + +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; + +use thiserror::Error; + +use crate::config::{Aggregate, Config, Endpoint}; +use crate::reconcile::{self, Action, Conflict, Policy, ReconcileError, Skipped, SourceView}; +use crate::state::{State, Target}; +use crate::vdir::{self, VdirError}; + +#[derive(Debug, Error)] +pub enum SyncError { + #[error(transparent)] + Vdir(#[from] VdirError), + #[error(transparent)] + State(#[from] crate::state::StateError), + #[error("aggregate `{aggregate}`: {source}")] + Reconcile { + aggregate: String, + #[source] + source: ReconcileError, + }, + #[error( + "aggregate `{aggregate}` was last published to `{recorded}` but is now configured for \ + `{configured}`; run `calcalist aggregate retarget {aggregate} --to {configured}` rather \ + than syncing, which would read the new target as an empty calendar" + )] + TargetDrift { + aggregate: String, + recorded: String, + configured: String, + }, + #[error("aggregate `{aggregate}` refers to unknown endpoint `{endpoint}`")] + UnknownEndpoint { aggregate: String, endpoint: String }, +} + +/// What one cycle did, for reporting to the user. +#[derive(Debug, Default)] +pub struct Report { + pub aggregates: Vec, + pub dry_run: bool, +} + +#[derive(Debug)] +pub struct AggregateReport { + pub id: String, + pub mirrored: usize, + pub written_back: usize, + pub deleted_from_aggregate: usize, + pub deleted_from_sources: usize, + pub conflicts: Vec, + pub skipped: Vec, +} + +/// Runs the local half of a cycle: reconcile every aggregate and apply the result. +pub fn run( + config: &Config, + state_dir: &Path, + dry_run: bool, + force: bool, +) -> Result { + let state_path = state_dir.join(crate::state::FILE_NAME); + let mut state = State::load(&state_path)?; + let mut report = Report { + dry_run, + ..Report::default() + }; + + for aggregate in &config.aggregates { + let target = resolve(config, aggregate, &aggregate.target)?; + check_target_drift(&state, aggregate, target)?; + report.aggregates.push(sync_aggregate( + config, aggregate, target, state_dir, &mut state, dry_run, force, + )?); + } + + if !dry_run { + state.save(&state_path)?; + } + Ok(report) +} + +fn resolve<'a>( + config: &'a Config, + aggregate: &Aggregate, + endpoint: &str, +) -> Result<&'a Endpoint, SyncError> { + config + .endpoint(endpoint) + .ok_or_else(|| SyncError::UnknownEndpoint { + aggregate: aggregate.id.clone(), + endpoint: endpoint.to_string(), + }) +} + +/// Refuses to reconcile when the configured target is not the one the recorded +/// state describes. +/// +/// Without this the new, empty target would be read as an aggregate whose every +/// event had been deleted — and with delete propagation on, that would remove +/// them from every source calendar. +fn check_target_drift( + state: &State, + aggregate: &Aggregate, + target: &Endpoint, +) -> Result<(), SyncError> { + let Some(prior) = state.aggregate(&aggregate.id) else { + return Ok(()); + }; + let same = + prior.target.endpoint == aggregate.target && prior.target.kind == target.kind.kind_name(); + if same { + return Ok(()); + } + Err(SyncError::TargetDrift { + aggregate: aggregate.id.clone(), + recorded: prior.target.endpoint.clone(), + configured: aggregate.target.clone(), + }) +} + +fn sync_aggregate( + config: &Config, + aggregate: &Aggregate, + target: &Endpoint, + state_dir: &Path, + state: &mut State, + dry_run: bool, + force: bool, +) -> Result { + let target_dir = vdir_path(state_dir, &target.id); + let target_items = vdir::read(&target_dir)?; + + // Read every source up front: the reconciler works on a single consistent + // snapshot rather than re-reading as it goes. + let mut source_dirs: BTreeMap = BTreeMap::new(); + let mut source_items: BTreeMap = BTreeMap::new(); + for id in &aggregate.sources { + let endpoint = resolve(config, aggregate, id)?; + let dir = vdir_path(state_dir, &endpoint.id); + source_items.insert(id.clone(), vdir::read(&dir)?); + source_dirs.insert(id.clone(), dir); + } + + let views: Vec> = aggregate + .sources + .iter() + .filter_map(|id| { + let endpoint = config.endpoint(id)?; + Some(SourceView { + id: &endpoint.id, + writable: endpoint.kind.is_writable(), + owner: endpoint.kind.owner(), + items: source_items.get(id)?, + }) + }) + .collect(); + + let policy = Policy { + aggregate_id: &aggregate.id, + suppression: target.kind.scheduling_suppression(), + default_sink: aggregate.default_sink.as_deref(), + propagate_deletes: aggregate.propagate_deletes, + max_delete_fraction: aggregate.max_delete_fraction, + force, + }; + + let outcome = reconcile::reconcile( + policy, + &views, + &target_items, + state.aggregate(&aggregate.id), + ) + .map_err(|source| SyncError::Reconcile { + aggregate: aggregate.id.clone(), + source, + })?; + + let mut report = AggregateReport { + id: aggregate.id.clone(), + mirrored: 0, + written_back: 0, + deleted_from_aggregate: 0, + deleted_from_sources: 0, + conflicts: outcome.conflicts, + skipped: outcome.skipped, + }; + + for action in &outcome.actions { + match action { + Action::WriteAggregate { + uid, + calendar, + replaces, + } => { + report.mirrored += 1; + if !dry_run { + vdir::write(&target_dir, uid, calendar, replaces.as_deref())?; + } + } + Action::DeleteAggregate { path, .. } => { + report.deleted_from_aggregate += 1; + if !dry_run { + vdir::remove(path)?; + } + } + Action::WriteSource { + source_id, + uid, + calendar, + replaces, + } => { + report.written_back += 1; + if !dry_run && let Some(dir) = source_dirs.get(source_id) { + vdir::write(dir, uid, calendar, replaces.as_deref())?; + } + } + Action::DeleteSource { path, .. } => { + report.deleted_from_sources += 1; + if !dry_run { + vdir::remove(path)?; + } + } + } + } + + if !dry_run { + let entry = state.aggregate_mut( + &aggregate.id, + Target { + endpoint: target.id.clone(), + kind: target.kind.kind_name().to_string(), + }, + ); + entry.links = outcome.links; + } + Ok(report) +} + +/// Where an endpoint's local mirror lives. +pub fn vdir_path(state_dir: &Path, endpoint_id: &str) -> PathBuf { + state_dir.join("vdir").join(endpoint_id) +} diff --git a/src/vdir.rs b/src/vdir.rs new file mode 100644 index 0000000..b20618a --- /dev/null +++ b/src/vdir.rs @@ -0,0 +1,346 @@ +//! Reading and writing vdir directories. +//! +//! A vdir is a flat directory of `.ics` files, one per item. It is the seam +//! between calcalist and pimsync: pimsync mirrors each remote into one of these, +//! and the reconciler then works purely on local files. +//! +//! Filenames are ours to choose, so items are always keyed by the UID inside the +//! file rather than by its name — pimsync and other tools name files differently +//! and we must read whatever they wrote. + +use std::collections::BTreeMap; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; + +use thiserror::Error; + +use crate::ical::{Calendar, IcalError}; + +const EXTENSION: &str = "ics"; + +#[derive(Debug, Error)] +pub enum VdirError { + #[error("could not read directory {path}: {source}")] + ReadDir { + path: PathBuf, + #[source] + source: io::Error, + }, + #[error("could not read {path}: {source}")] + ReadFile { + path: PathBuf, + #[source] + source: io::Error, + }, + #[error("could not write {path}: {source}")] + Write { + path: PathBuf, + #[source] + source: io::Error, + }, + #[error("could not remove {path}: {source}")] + Remove { + path: PathBuf, + #[source] + source: io::Error, + }, + #[error("{path} is not valid iCalendar: {source}")] + Malformed { + path: PathBuf, + #[source] + source: IcalError, + }, + #[error("{path} contains no VEVENT UID")] + NoUid { path: PathBuf }, +} + +/// One item in a vdir, keyed by the UID found inside it. +#[derive(Debug, Clone)] +pub struct Item { + pub uid: String, + pub path: PathBuf, + pub calendar: Calendar, + /// Digest of the meaningful content, for change detection. + pub hash: String, +} + +/// Every item in `dir`, keyed by UID. +/// +/// A missing directory reads as empty: an endpoint that has never synced is not +/// an error, and must not be mistaken for one whose events were all deleted. +pub fn read(dir: &Path) -> Result, VdirError> { + let entries = match fs::read_dir(dir) { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(BTreeMap::new()), + Err(source) => { + return Err(VdirError::ReadDir { + path: dir.to_path_buf(), + source, + }); + } + }; + + let mut items = BTreeMap::new(); + for entry in entries { + let path = entry + .map_err(|source| VdirError::ReadDir { + path: dir.to_path_buf(), + source, + })? + .path(); + if !is_calendar_file(&path) { + continue; + } + let item = read_item(&path)?; + items.insert(item.uid.clone(), item); + } + Ok(items) +} + +fn read_item(path: &Path) -> Result { + let text = fs::read_to_string(path).map_err(|source| VdirError::ReadFile { + path: path.to_path_buf(), + source, + })?; + let calendar = Calendar::parse(&text).map_err(|source| VdirError::Malformed { + path: path.to_path_buf(), + source, + })?; + let uid = calendar + .uid() + .ok_or_else(|| VdirError::NoUid { + path: path.to_path_buf(), + })? + .to_string(); + let hash = calendar.content_hash(); + Ok(Item { + uid, + path: path.to_path_buf(), + calendar, + hash, + }) +} + +/// Writes `calendar` into `dir` under a name derived from its UID. +/// +/// `replaces` is the path the item currently occupies, when it has one. Filenames +/// in a vdir are arbitrary — pimsync chooses its own — so an item already present +/// under a different name must be removed, or the calendar would end up holding +/// the same event twice. +/// +/// The write is atomic, so a crash cannot leave pimsync a half-written event to push. +pub fn write( + dir: &Path, + uid: &str, + calendar: &Calendar, + replaces: Option<&Path>, +) -> Result { + fs::create_dir_all(dir).map_err(|source| VdirError::Write { + path: dir.to_path_buf(), + source, + })?; + let path = dir.join(file_name(uid)); + let temporary = path.with_extension("ics.tmp"); + fs::write(&temporary, calendar.to_ics()).map_err(|source| VdirError::Write { + path: temporary.clone(), + source, + })?; + fs::rename(&temporary, &path).map_err(|source| VdirError::Write { + path: path.clone(), + source, + })?; + if let Some(previous) = replaces + && previous != path + { + remove(previous)?; + } + Ok(path) +} + +pub fn remove(path: &Path) -> Result<(), VdirError> { + match fs::remove_file(path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(source) => Err(VdirError::Remove { + path: path.to_path_buf(), + source, + }), + } +} + +fn is_calendar_file(path: &Path) -> bool { + path.is_file() + && path + .extension() + .is_some_and(|extension| extension.eq_ignore_ascii_case(EXTENSION)) +} + +/// A filesystem-safe, stable, collision-free name for a UID. +/// +/// UIDs may contain characters no filesystem accepts, and sanitising alone would +/// let two different UIDs collide, so a digest of the original is appended. +fn file_name(uid: &str) -> String { + let sanitised: String = uid + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || matches!(ch, '.' | '-' | '_') { + ch + } else { + '_' + } + }) + .take(100) + .collect(); + let digest: String = blake3::hash(uid.as_bytes()) + .as_bytes() + .iter() + .take(4) + .map(|byte| format!("{byte:02x}")) + .collect(); + format!("{sanitised}-{digest}.{EXTENSION}") +} + +#[cfg(test)] +mod tests { + use super::*; + + fn event(uid: &str, summary: &str) -> Calendar { + let text = format!( + "BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:{uid}\r\nDTSTAMP:20260910T090000Z\r\nSUMMARY:{summary}\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n" + ); + Calendar::parse(&text).expect("fixture should parse") + } + + #[test] + fn a_missing_directory_reads_as_empty_not_as_an_error() { + let dir = tempfile::tempdir().expect("temp dir"); + let missing = dir.path().join("never-synced"); + assert!( + read(&missing) + .expect("missing dir is not an error") + .is_empty() + ); + } + + #[test] + fn writes_and_reads_back_by_uid() { + let dir = tempfile::tempdir().expect("temp dir"); + write( + dir.path(), + "event-1@example.com", + &event("event-1@example.com", "Standup"), + None, + ) + .expect("write"); + + let items = read(dir.path()).expect("read"); + assert_eq!(items.len(), 1); + let item = items.get("event-1@example.com").expect("keyed by uid"); + assert_eq!( + item.calendar.to_ics(), + event("event-1@example.com", "Standup").to_ics() + ); + } + + /// Filenames are ours, but the key must come from the file's contents, since + /// pimsync names files its own way. + #[test] + fn items_are_keyed_by_content_not_filename() { + let dir = tempfile::tempdir().expect("temp dir"); + let text = event("event-1@example.com", "Standup").to_ics(); + fs::write(dir.path().join("some-name-pimsync-chose.ics"), text).expect("write"); + + let items = read(dir.path()).expect("read"); + assert!(items.contains_key("event-1@example.com")); + } + + #[test] + fn rewriting_the_same_uid_replaces_rather_than_accumulates() { + let dir = tempfile::tempdir().expect("temp dir"); + let uid = "event-1@example.com"; + write(dir.path(), uid, &event(uid, "First"), None).expect("write"); + write(dir.path(), uid, &event(uid, "Second"), None).expect("rewrite"); + + let items = read(dir.path()).expect("read"); + assert_eq!(items.len(), 1); + assert!(items[uid].calendar.to_ics().contains("SUMMARY:Second")); + } + + /// Regression: filenames in a vdir are arbitrary and pimsync picks its own, + /// so writing an item already present under a different name must replace it + /// rather than leave the calendar holding the same event twice. + #[test] + fn writing_replaces_an_item_stored_under_a_foreign_filename() { + let dir = tempfile::tempdir().expect("temp dir"); + let uid = "event-1@example.com"; + let foreign = dir.path().join("pimsync-chose-this.ics"); + fs::write(&foreign, event(uid, "Original").to_ics()).expect("write"); + + write(dir.path(), uid, &event(uid, "Edited"), Some(&foreign)).expect("write"); + + let items = read(dir.path()).expect("read"); + assert_eq!(items.len(), 1, "the event must not appear twice"); + assert!(items[uid].calendar.to_ics().contains("SUMMARY:Edited")); + assert!(!foreign.exists(), "the superseded file should be gone"); + } + + /// Two UIDs that sanitise identically must not overwrite each other. + #[test] + fn unsafe_uids_get_distinct_filenames() { + let first = file_name("a/b:c"); + let second = file_name("a_b_c"); + assert_ne!(first, second); + assert!(!first.contains('/')); + assert!(first.ends_with(".ics")); + } + + #[test] + fn ignores_files_that_are_not_calendars() { + let dir = tempfile::tempdir().expect("temp dir"); + fs::write(dir.path().join("README.md"), "not a calendar").expect("write"); + write(dir.path(), "event-1", &event("event-1", "Standup"), None).expect("write"); + + assert_eq!(read(dir.path()).expect("read").len(), 1); + } + + #[test] + fn removing_an_absent_file_is_not_an_error() { + let dir = tempfile::tempdir().expect("temp dir"); + remove(&dir.path().join("gone.ics")).expect("removing nothing should succeed"); + } + + /// A server rewriting DTSTAMP must not look like a user edit. + #[test] + fn the_hash_ignores_volatile_properties() { + let original = event("event-1", "Standup"); + let restamped = Calendar::parse( + &original + .to_ics() + .replace("DTSTAMP:20260910T090000Z", "DTSTAMP:20261225T235959Z"), + ) + .expect("parse"); + assert_eq!(original.content_hash(), restamped.content_hash()); + } + + #[test] + fn the_hash_notices_real_edits() { + assert_ne!( + event("event-1", "Standup").content_hash(), + event("event-1", "Retrospective").content_hash() + ); + } + + /// Property order carries no meaning in iCalendar, so it must not churn the hash. + #[test] + fn the_hash_ignores_property_order() { + let reordered = Calendar::parse( + "BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nSUMMARY:Standup\r\nDTSTAMP:20260910T090000Z\r\nUID:event-1\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n", + ) + .expect("parse"); + assert_eq!( + event("event-1", "Standup").content_hash(), + reordered.content_hash() + ); + } +}