inbuxa/inbuxa-migrate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

MS Exchange Graph fixes

Browse Source
This commit is contained in:
Maurus Decimus
2026-08-27 19:51:18 +02:00
parent e8d98abccb
commit e00909640a
11 changed files with 1908 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cli.rs
+15
View File
@@ -1174,6 +1174,19 @@ pub struct ExchangeGraphImportArgs {
)]
exception_window_years: i32,
#[arg(
long,
help = "Skip contact photos (saves one request per contact on large address books)",
long_help = "Skip contact photos.\n Graph will not say which contacts have a photo: $expand=photo comes back empty\n even when one exists, so the only way to find out is to ask for the bytes.\n vandelay therefore spends one extra request per contact. Pass this to skip\n photos entirely on a large address book."
)]
skip_contact_photos: bool,
#[arg(
long,
help = "Skip event file attachments (they are stored as JSCalendar enclosure links)"
)]
skip_event_attachments: bool,
#[arg(
long,
value_name = "URL",
@@ -1239,6 +1252,8 @@ fn resolve_exchange_graph_import(args: ExchangeGraphImportArgs) -> Result<Action
graph_connections,
top,
exception_window_years: args.exception_window_years,
contact_photos: !args.skip_contact_photos,
event_attachments: !args.skip_event_attachments,
allow_source_change: args.allow_source_change,
},
))
src/exchange_graph/api.rs
+32
View File
@@ -153,6 +153,22 @@ impl Endpoints {
)
}
pub fn calendar_occurrences(
&self,
calendar_id: &str,
window_start: &str,
window_end: &str,
top: usize,
) -> String {
format!(
"{}/calendars/{}/calendarView?startDateTime={window_start}&endDateTime={window_end}\
&$filter=type%20eq%20%27occurrence%27&$top={top}\
&$select=id,type,seriesMasterId,start",
self.me_or_user(),
url_escape(calendar_id)
)
}
pub fn contact_folders(&self, top: usize) -> String {
format!("{}/contactFolders?$top={top}", self.me_or_user())
}
@@ -196,6 +212,22 @@ impl Endpoints {
format!("{}/contacts/{}", self.me_or_user(), url_escape(contact_id))
}
pub fn contact_photo(&self, contact_id: &str) -> String {
format!(
"{}/contacts/{}/photo/$value",
self.me_or_user(),
url_escape(contact_id)
)
}
pub fn event_attachments(&self, event_id: &str) -> String {
format!(
"{}/events/{}/attachments",
self.me_or_user(),
url_escape(event_id)
)
}
pub fn mailbox_settings_timezone(&self) -> String {
format!("{}/mailboxSettings?$select=timeZone", self.me_or_user())
}
src/exchange_graph/contact_map.rs
+44
View File
@@ -286,6 +286,38 @@ pub fn convert_contact(graph_contact: &Value) -> Result<ConvertedContact, GraphE
card.insert("links".to_owned(), Value::Object(links));
}
if let Some(cats) = graph_contact.get("categories").and_then(Value::as_array) {
let mut keywords = Map::new();
for cat in cats.iter().filter_map(Value::as_str) {
if !cat.trim().is_empty() {
keywords.insert(cat.to_ascii_lowercase(), Value::Bool(true));
}
}
if !keywords.is_empty() {
card.insert("keywords".to_owned(), Value::Object(keywords));
}
}
if let Some(ims) = graph_contact.get("imAddresses").and_then(Value::as_array) {
let mut services = Map::new();
for (i, addr) in ims
.iter()
.filter_map(Value::as_str)
.map(str::trim)
.filter(|a| !a.is_empty())
.enumerate()
{
let slot = if has_uri_scheme(addr) { "uri" } else { "user" };
services.insert(
(i + 1).to_string(),
json!({"@type": "OnlineService", slot: addr}),
);
}
if !services.is_empty() {
card.insert("onlineServices".to_owned(), Value::Object(services));
}
}
if let Some(created) = graph_contact.get("createdDateTime").and_then(Value::as_str) {
card.insert(
"created".to_owned(),
@@ -308,6 +340,18 @@ pub fn convert_contact(graph_contact: &Value) -> Result<ConvertedContact, GraphE
})
}
fn has_uri_scheme(value: &str) -> bool {
let Some((scheme, rest)) = value.split_once(':') else {
return false;
};
!rest.is_empty()
&& !scheme.is_empty()
&& scheme.starts_with(|c: char| c.is_ascii_alphabetic())
&& scheme
.chars()
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '-' | '.'))
}
pub fn synthetic_uid(graph_id: &str) -> String {
let hash = blake3::hash(graph_id.as_bytes());
format!("vandelay-graph-{}", hash.to_hex())
src/exchange_graph/expand.rs
+637
View File
@@ -0,0 +1,637 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
*
* SPDX-License-Identifier: Apache-2.0 OR MIT
*/
use chrono::{Datelike, Duration, NaiveDate, Weekday};
use serde_json::Value;
const MAX_STEPS: usize = 200_000;
pub fn expected_dates(
recurrence: &Value,
window_start: NaiveDate,
window_end: NaiveDate,
) -> Option<Vec<NaiveDate>> {
let pattern = recurrence.get("pattern")?;
let range = recurrence.get("range")?;
let start = date_of(range.get("startDate"))?;
let limit = date_of(range.get("endDate"));
let count = range
.get("numberOfOccurrences")
.and_then(Value::as_u64)
.filter(|n| *n > 0)
.map(|n| n as usize);
let interval = pattern
.get("interval")
.and_then(Value::as_u64)
.filter(|i| *i > 0)
.unwrap_or(1) as i64;
let kind = pattern.get("type").and_then(Value::as_str)?;
let mut out: Vec<NaiveDate> = Vec::new();
let mut emitted = 0usize;
let mut steps = 0usize;
let mut cursor = start;
let push = |date: NaiveDate, out: &mut Vec<NaiveDate>, emitted: &mut usize| {
if date < start {
return;
}
if let Some(end) = limit
&& date > end
{
return;
}
*emitted += 1;
if date >= window_start && date <= window_end {
out.push(date);
}
};
loop {
steps += 1;
if steps > MAX_STEPS {
break;
}
let done = match kind {
"daily" => {
push(cursor, &mut out, &mut emitted);
cursor = cursor.checked_add_signed(Duration::days(interval))?;
false
}
"weekly" => {
let days = weekdays(pattern);
if days.is_empty() {
return None;
}
let first = first_day_of_week(pattern);
let week_start = start_of_week(cursor, first);
for day in &days {
let offset = weekday_offset(first, *day);
let date = week_start + Duration::days(offset);
push(date, &mut out, &mut emitted);
}
cursor = week_start.checked_add_signed(Duration::weeks(interval))?;
false
}
"absoluteMonthly" => {
let dom = pattern.get("dayOfMonth").and_then(Value::as_u64)? as u32;
if let Some(date) = clamped_day(cursor.year(), cursor.month(), dom) {
push(date, &mut out, &mut emitted);
}
cursor = add_months(cursor.with_day(1)?, interval)?;
false
}
"relativeMonthly" => {
if let Some(date) = nth_weekday_of_month(
cursor.year(),
cursor.month(),
&weekdays(pattern),
pattern
.get("index")
.and_then(Value::as_str)
.unwrap_or("first"),
) {
push(date, &mut out, &mut emitted);
}
cursor = add_months(cursor.with_day(1)?, interval)?;
false
}
"absoluteYearly" => {
let month = pattern.get("month").and_then(Value::as_u64)? as u32;
let dom = pattern.get("dayOfMonth").and_then(Value::as_u64)? as u32;
if let Some(date) = clamped_day(cursor.year(), month, dom) {
push(date, &mut out, &mut emitted);
}
cursor = NaiveDate::from_ymd_opt(cursor.year() + interval as i32, 1, 1)?;
false
}
"relativeYearly" => {
let month = pattern.get("month").and_then(Value::as_u64)? as u32;
if let Some(date) = nth_weekday_of_month(
cursor.year(),
month,
&weekdays(pattern),
pattern
.get("index")
.and_then(Value::as_str)
.unwrap_or("first"),
) {
push(date, &mut out, &mut emitted);
}
cursor = NaiveDate::from_ymd_opt(cursor.year() + interval as i32, 1, 1)?;
false
}
_ => true,
};
if done {
return None;
}
if let Some(n) = count
&& emitted >= n
{
break;
}
if let Some(end) = limit
&& cursor > end
{
break;
}
if cursor > window_end {
break;
}
}
out.sort_unstable();
out.dedup();
Some(out)
}
fn date_of(value: Option<&Value>) -> Option<NaiveDate> {
let raw = value?.as_str()?;
NaiveDate::parse_from_str(raw.split('T').next().unwrap_or(raw), "%Y-%m-%d").ok()
}
fn weekdays(pattern: &Value) -> Vec<Weekday> {
pattern
.get("daysOfWeek")
.and_then(Value::as_array)
.map(|days| {
days.iter()
.filter_map(Value::as_str)
.filter_map(parse_weekday)
.collect()
})
.unwrap_or_default()
}
fn parse_weekday(name: &str) -> Option<Weekday> {
match name.to_ascii_lowercase().as_str() {
"monday" => Some(Weekday::Mon),
"tuesday" => Some(Weekday::Tue),
"wednesday" => Some(Weekday::Wed),
"thursday" => Some(Weekday::Thu),
"friday" => Some(Weekday::Fri),
"saturday" => Some(Weekday::Sat),
"sunday" => Some(Weekday::Sun),
_ => None,
}
}
fn first_day_of_week(pattern: &Value) -> Weekday {
pattern
.get("firstDayOfWeek")
.and_then(Value::as_str)
.and_then(parse_weekday)
.unwrap_or(Weekday::Sun)
}
fn weekday_offset(first: Weekday, day: Weekday) -> i64 {
let from = first.num_days_from_sunday() as i64;
let to = day.num_days_from_sunday() as i64;
(to - from).rem_euclid(7)
}
fn start_of_week(date: NaiveDate, first: Weekday) -> NaiveDate {
let offset = weekday_offset(first, date.weekday());
date - Duration::days(offset)
}
fn clamped_day(year: i32, month: u32, day: u32) -> Option<NaiveDate> {
if !(1..=12).contains(&month) {
return None;
}
let last = last_day_of_month(year, month)?;
NaiveDate::from_ymd_opt(year, month, day.min(last))
}
fn last_day_of_month(year: i32, month: u32) -> Option<u32> {
let first_next = if month == 12 {
NaiveDate::from_ymd_opt(year + 1, 1, 1)?
} else {
NaiveDate::from_ymd_opt(year, month + 1, 1)?
};
Some((first_next - Duration::days(1)).day())
}
fn add_months(date: NaiveDate, months: i64) -> Option<NaiveDate> {
let total = date.year() as i64 * 12 + (date.month() as i64 - 1) + months;
let year = (total.div_euclid(12)) as i32;
let month = (total.rem_euclid(12) + 1) as u32;
NaiveDate::from_ymd_opt(year, month, 1)
}
fn nth_weekday_of_month(year: i32, month: u32, days: &[Weekday], index: &str) -> Option<NaiveDate> {
if days.is_empty() || !(1..=12).contains(&month) {
return None;
}
let last = last_day_of_month(year, month)?;
let mut matches: Vec<NaiveDate> = Vec::new();
for day in 1..=last {
let date = NaiveDate::from_ymd_opt(year, month, day)?;
if days.contains(&date.weekday()) {
matches.push(date);
}
}
if matches.is_empty() {
return None;
}
match index.to_ascii_lowercase().as_str() {
"first" => matches.first().copied(),
"second" => matches.get(1).copied(),
"third" => matches.get(2).copied(),
"fourth" => matches.get(3).copied(),
"last" => matches.last().copied(),
_ => matches.first().copied(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn day(s: &str) -> NaiveDate {
NaiveDate::parse_from_str(s, "%Y-%m-%d").unwrap()
}
fn expand(recurrence: Value, from: &str, to: &str) -> Vec<String> {
expected_dates(&recurrence, day(from), day(to))
.expect("pattern is supported")
.into_iter()
.map(|d| d.to_string())
.collect()
}
#[test]
fn daily_numbered_stops_at_the_count() {
let dates = expand(
json!({
"pattern": {"type": "daily", "interval": 1},
"range": {"type": "numbered", "startDate": "2026-09-21",
"numberOfOccurrences": 6}
}),
"2026-01-01",
"2027-01-01",
);
assert_eq!(
dates,
[
"2026-09-21",
"2026-09-22",
"2026-09-23",
"2026-09-24",
"2026-09-25",
"2026-09-26"
]
);
}
#[test]
fn weekly_interval_two_keeps_all_three_weekdays() {
let dates = expand(
json!({
"pattern": {"type": "weekly", "interval": 2,
"daysOfWeek": ["monday", "wednesday", "friday"],
"firstDayOfWeek": "monday"},
"range": {"type": "endDate", "startDate": "2026-09-16",
"endDate": "2026-10-03"}
}),
"2026-01-01",
"2027-01-01",
);
assert_eq!(
dates,
[
"2026-09-16",
"2026-09-18",
"2026-09-28",
"2026-09-30",
"2026-10-02"
],
"interval 2 skips the week of the 21st entirely"
);
}
#[test]
fn relative_monthly_third_thursday() {
let dates = expand(
json!({
"pattern": {"type": "relativeMonthly", "interval": 1,
"daysOfWeek": ["thursday"], "index": "third"},
"range": {"type": "numbered", "startDate": "2026-09-17",
"numberOfOccurrences": 3}
}),
"2026-01-01",
"2027-06-01",
);
assert_eq!(dates, ["2026-09-17", "2026-10-15", "2026-11-19"]);
}
#[test]
fn relative_monthly_last_weekday() {
let dates = expand(
json!({
"pattern": {"type": "relativeMonthly", "interval": 1,
"daysOfWeek": ["friday"], "index": "last"},
"range": {"type": "numbered", "startDate": "2026-01-30",
"numberOfOccurrences": 3}
}),
"2026-01-01",
"2027-01-01",
);
assert_eq!(dates, ["2026-01-30", "2026-02-27", "2026-03-27"]);
}
#[test]
fn absolute_monthly_clamps_to_a_short_month() {
let dates = expand(
json!({
"pattern": {"type": "absoluteMonthly", "interval": 1, "dayOfMonth": 31},
"range": {"type": "numbered", "startDate": "2026-01-31",
"numberOfOccurrences": 3}
}),
"2026-01-01",
"2027-01-01",
);
assert_eq!(
dates,
["2026-01-31", "2026-02-28", "2026-03-31"],
"day 31 clamps to the last day of February"
);
}
#[test]
fn absolute_yearly_keeps_a_leap_day_series_on_the_28th_in_common_years() {
let dates = expand(
json!({
"pattern": {"type": "absoluteYearly", "interval": 1,
"month": 2, "dayOfMonth": 29},
"range": {"type": "numbered", "startDate": "2024-02-29",
"numberOfOccurrences": 3}
}),
"2020-01-01",
"2030-01-01",
);
assert_eq!(dates, ["2024-02-29", "2025-02-28", "2026-02-28"]);
}
#[test]
fn window_clips_output_but_the_count_still_tracks_the_whole_series() {
let dates = expand(
json!({
"pattern": {"type": "daily", "interval": 1},
"range": {"type": "numbered", "startDate": "2026-09-21",
"numberOfOccurrences": 6}
}),
"2026-09-24",
"2026-09-25",
);
assert_eq!(
dates,
["2026-09-24", "2026-09-25"],
"only in-window dates are returned, but occurrences 1-3 still counted \
toward the numbered range"
);
}
#[test]
fn no_end_series_stops_at_the_window() {
let dates = expand(
json!({
"pattern": {"type": "daily", "interval": 1},
"range": {"type": "noEnd", "startDate": "2026-09-21"}
}),
"2026-09-21",
"2026-09-24",
);
assert_eq!(
dates,
["2026-09-21", "2026-09-22", "2026-09-23", "2026-09-24"]
);
}
#[test]
fn unknown_pattern_is_reported_rather_than_guessed() {
assert!(
expected_dates(
&json!({
"pattern": {"type": "somethingNew", "interval": 1},
"range": {"type": "noEnd", "startDate": "2026-01-01"}
}),
day("2026-01-01"),
day("2027-01-01"),
)
.is_none()
);
}
}
pub fn graph_pattern_from_jscalendar(rule: &Value, series_start: NaiveDate) -> Option<Value> {
use serde_json::json;
let frequency = rule.get("frequency").and_then(Value::as_str)?;
let interval = rule.get("interval").and_then(Value::as_u64).unwrap_or(1);
let by_day: Vec<String> = rule
.get("byDay")
.and_then(Value::as_array)
.map(|days| {
days.iter()
.filter_map(|d| d.get("day").and_then(Value::as_str))
.filter_map(long_weekday)
.map(str::to_owned)
.collect()
})
.unwrap_or_default();
let by_month_day = rule
.get("byMonthDay")
.and_then(Value::as_array)
.and_then(|a| a.first())
.and_then(Value::as_i64);
let by_month = rule
.get("byMonth")
.and_then(Value::as_array)
.and_then(|a| a.first())
.and_then(|m| {
m.as_str()
.and_then(|s| s.parse::<i64>().ok())
.or_else(|| m.as_i64())
});
let set_position = rule
.get("bySetPosition")
.and_then(Value::as_array)
.and_then(|a| a.first())
.and_then(Value::as_i64);
let mut pattern = serde_json::Map::new();
pattern.insert("interval".to_owned(), Value::from(interval));
let kind = match frequency {
"daily" => "daily",
"weekly" => {
pattern.insert("daysOfWeek".to_owned(), json!(by_day));
"weekly"
}
"monthly" if set_position.is_some() && !by_day.is_empty() => {
pattern.insert("daysOfWeek".to_owned(), json!(by_day));
pattern.insert("index".to_owned(), Value::from(index_name(set_position?)));
"relativeMonthly"
}
"monthly" => {
pattern.insert(
"dayOfMonth".to_owned(),
Value::from(by_month_day.unwrap_or(series_start.day() as i64)),
);
"absoluteMonthly"
}
"yearly" if set_position.is_some() && !by_day.is_empty() => {
pattern.insert("daysOfWeek".to_owned(), json!(by_day));
pattern.insert("index".to_owned(), Value::from(index_name(set_position?)));
pattern.insert(
"month".to_owned(),
Value::from(by_month.unwrap_or(series_start.month() as i64)),
);
"relativeYearly"
}
"yearly" => {
pattern.insert(
"month".to_owned(),
Value::from(by_month.unwrap_or(series_start.month() as i64)),
);
pattern.insert(
"dayOfMonth".to_owned(),
Value::from(by_month_day.unwrap_or(series_start.day() as i64)),
);
"absoluteYearly"
}
_ => return None,
};
pattern.insert("type".to_owned(), Value::from(kind));
if let Some(first) = rule.get("firstDayOfWeek").and_then(Value::as_str)
&& let Some(long) = long_weekday(first)
{
pattern.insert("firstDayOfWeek".to_owned(), Value::from(long));
}
let mut range = serde_json::Map::new();
range.insert(
"startDate".to_owned(),
Value::from(series_start.format("%Y-%m-%d").to_string()),
);
if let Some(count) = rule.get("count").and_then(Value::as_u64) {
range.insert("type".to_owned(), Value::from("numbered"));
range.insert("numberOfOccurrences".to_owned(), Value::from(count));
} else if let Some(until) = rule.get("until").and_then(Value::as_str) {
range.insert("type".to_owned(), Value::from("endDate"));
range.insert(
"endDate".to_owned(),
Value::from(until.split('T').next().unwrap_or(until).to_owned()),
);
} else {
range.insert("type".to_owned(), Value::from("noEnd"));
}
Some(json!({"pattern": Value::Object(pattern), "range": Value::Object(range)}))
}
fn long_weekday(short: &str) -> Option<&'static str> {
match short.to_ascii_lowercase().as_str() {
"mo" => Some("monday"),
"tu" => Some("tuesday"),
"we" => Some("wednesday"),
"th" => Some("thursday"),
"fr" => Some("friday"),
"sa" => Some("saturday"),
"su" => Some("sunday"),
_ => None,
}
}
fn index_name(set_position: i64) -> &'static str {
match set_position {
1 => "first",
2 => "second",
3 => "third",
4 => "fourth",
-1 => "last",
_ => "first",
}
}
#[cfg(test)]
mod roundtrip_tests {
use super::*;
use crate::exchange_graph::recurrence::convert_patterned_recurrence;
use serde_json::json;
fn day(s: &str) -> NaiveDate {
NaiveDate::parse_from_str(s, "%Y-%m-%d").unwrap()
}
fn roundtrip(graph: Value, start: &str, from: &str, to: &str) {
let direct = expected_dates(&graph, day(from), day(to)).expect("graph pattern expands");
let rule = convert_patterned_recurrence(&graph).expect("converts to jscalendar");
let back = graph_pattern_from_jscalendar(&rule, day(start))
.expect("jscalendar rule maps back to a graph pattern");
let viarule = expected_dates(&back, day(from), day(to)).expect("mapped pattern expands");
assert_eq!(
direct, viarule,
"a stored rule must expand to the same dates as the Graph pattern it came from"
);
}
#[test]
fn stored_rules_expand_identically_to_their_graph_patterns() {
roundtrip(
json!({"pattern": {"type": "daily", "interval": 2},
"range": {"type": "numbered", "startDate": "2026-09-21",
"numberOfOccurrences": 5}}),
"2026-09-21",
"2026-01-01",
"2027-01-01",
);
roundtrip(
json!({"pattern": {"type": "weekly", "interval": 2,
"daysOfWeek": ["monday", "wednesday", "friday"],
"firstDayOfWeek": "monday"},
"range": {"type": "endDate", "startDate": "2026-09-16",
"endDate": "2026-12-31"}}),
"2026-09-16",
"2026-01-01",
"2027-01-01",
);
roundtrip(
json!({"pattern": {"type": "relativeMonthly", "interval": 1,
"daysOfWeek": ["thursday"], "index": "third"},
"range": {"type": "numbered", "startDate": "2026-09-17",
"numberOfOccurrences": 6}}),
"2026-09-17",
"2026-01-01",
"2028-01-01",
);
roundtrip(
json!({"pattern": {"type": "absoluteMonthly", "interval": 1, "dayOfMonth": 31},
"range": {"type": "numbered", "startDate": "2026-01-31",
"numberOfOccurrences": 4}}),
"2026-01-31",
"2026-01-01",
"2027-01-01",
);
roundtrip(
json!({"pattern": {"type": "absoluteYearly", "interval": 1,
"month": 11, "dayOfMonth": 29},
"range": {"type": "noEnd", "startDate": "2026-11-29"}}),
"2026-11-29",
"2026-01-01",
"2031-01-01",
);
roundtrip(
json!({"pattern": {"type": "relativeYearly", "interval": 1,
"daysOfWeek": ["monday"], "index": "last", "month": 5},
"range": {"type": "noEnd", "startDate": "2026-05-25"}}),
"2026-05-25",
"2026-01-01",
"2031-01-01",
);
}
}
src/exchange_graph/mod.rs
+1
View File
@@ -9,6 +9,7 @@ pub mod calendar_map;
pub mod client;
pub mod contact_map;
pub mod error;
pub mod expand;
pub mod oauth;
pub mod recurrence;
pub mod retry;
src/sync/import_exchange_graph/calendar.rs
+308 -141
View File
@@ -25,6 +25,52 @@ use crate::sync::import_jmap::pool::Pool;
use super::coordinator::{CHUNK_SIZE, GraphCoordinator};
use super::folders::CalendarFolder;
pub type EventAttachment = (String, String, Vec<u8>);
fn fetch_event_attachments(
client: &crate::exchange_graph::client::GraphClient,
endpoints: &crate::exchange_graph::api::Endpoints,
event_id: &str,
) -> Vec<EventAttachment> {
use base64::Engine;
let url = endpoints.event_attachments(event_id);
let Ok(body) = client.get_json_with_prefer(&url, &[]) else {
return Vec::new();
};
let Some(items) = body.get("value").and_then(Value::as_array) else {
return Vec::new();
};
let mut out = Vec::new();
for item in items {
if item.get("@odata.type").and_then(Value::as_str)
!= Some("#microsoft.graph.fileAttachment")
{
continue;
}
let Some(encoded) = item.get("contentBytes").and_then(Value::as_str) else {
continue;
};
let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(encoded) else {
continue;
};
if bytes.is_empty() {
continue;
}
let name = item
.get("name")
.and_then(Value::as_str)
.unwrap_or("attachment")
.to_owned();
let content_type = item
.get("contentType")
.and_then(Value::as_str)
.unwrap_or("application/octet-stream")
.to_owned();
out.push((name, content_type, bytes));
}
out
}
pub fn reconcile_all(
conn: &mut Connection,
ctx: &GraphCoordinator<'_>,
@@ -51,10 +97,16 @@ pub fn reconcile_all(
};
let mut want_ids: Vec<String> = Vec::new();
let mut occurrence_count = 0usize;
let mut series_masters: Vec<String> = Vec::new();
for stub in &stubs {
match classify_event_type(stub) {
EventType::Occurrence => occurrence_count += 1,
_ => {
if matches!(classify_event_type(stub), EventType::SeriesMaster)
&& let Some(id) = stub.get("id").and_then(Value::as_str)
{
series_masters.push(id.to_owned());
}
if let Some(id) = stub.get("id").and_then(Value::as_str) {
server_total.insert(id.to_owned());
if local.contains_key(id) {
@@ -75,29 +127,22 @@ pub fn reconcile_all(
occurrence_count
);
}
if want_ids.is_empty() {
continue;
}
let fetched = fetch_events(ctx, &want_ids);
let mut masters: Vec<(String, ConvertedEvent)> = Vec::new();
let mut exceptions: Vec<(String, ConvertedEvent)> = Vec::new();
let mut any_series = false;
for (graph_id, result) in fetched {
let mut attachments_by_id: HashMap<String, Vec<EventAttachment>> = HashMap::new();
for (graph_id, result, attachments) in fetched {
match result {
Ok(raw) => {
any_series |= is_series_master(&raw);
match convert_event(&raw, None) {
Ok(c) => match c.event_type {
EventType::Exception => exceptions.push((graph_id, c)),
_ => masters.push((graph_id, c)),
},
Err(e) => {
counts.failed += 1;
ctx.logger
.warn(&format!("graph event {graph_id} convert failed: {e}"));
}
Ok(raw) => match convert_event(&raw, None) {
Ok(c) => {
attachments_by_id.insert(graph_id.clone(), attachments);
masters.push((graph_id, c));
}
}
Err(e) => {
counts.failed += 1;
ctx.logger
.warn(&format!("graph event {graph_id} convert failed: {e}"));
}
},
Err(GraphError::Vanished) => counts.skipped += 1,
Err(e) => {
counts.failed += 1;
@@ -107,33 +152,13 @@ pub fn reconcile_all(
}
}
if any_series {
exceptions.extend(fetch_calendar_exceptions(ctx, &cal.graph_id, counts));
}
let mut master_by_graph_id: HashMap<String, ConvertedEvent> = masters.into_iter().collect();
for (ex_graph_id, ex) in exceptions {
let Some(master_graph_id) = ex.series_master_id.clone() else {
ctx.logger.warn(&format!(
"graph exception event {ex_graph_id} has no seriesMasterId; dropping"
));
counts.skipped += 1;
continue;
};
if let Some(master) = master_by_graph_id.get_mut(&master_graph_id) {
merge_exception_into(master, &ex);
} else if local.contains_key(&master_graph_id) {
merge_exception_into_existing(conn, ctx, &master_graph_id, &ex, counts);
} else {
counts.skipped += 1;
ctx.logger.warn(&format!(
"graph exception {ex_graph_id} references missing master {master_graph_id}; orphaned"
));
}
}
let master_by_graph_id: HashMap<String, ConvertedEvent> = masters.into_iter().collect();
let pairs: Vec<(String, ConvertedEvent)> = master_by_graph_id.into_iter().collect();
insert_events_chunked(conn, ctx, cal.local_id, &pairs, counts)?;
insert_events_chunked(conn, ctx, cal.local_id, &pairs, &attachments_by_id, counts)?;
if !series_masters.is_empty() {
apply_series_expansion(conn, ctx, &cal.graph_id, &series_masters, counts)?;
}
}
if any_failure {
@@ -150,8 +175,8 @@ pub fn reconcile_all(
fn fetch_events(
ctx: &GraphCoordinator<'_>,
ids: &[String],
) -> Vec<(String, Result<Value, GraphError>)> {
type R = (String, Result<Value, GraphError>);
) -> Vec<(String, Result<Value, GraphError>, Vec<EventAttachment>)> {
type R = (String, Result<Value, GraphError>, Vec<EventAttachment>);
let client = ctx.client.clone();
let endpoints: crate::exchange_graph::api::Endpoints = (*ctx.endpoints).clone();
let body_prefer = match ctx.event_body_format {
@@ -159,11 +184,18 @@ fn fetch_events(
EventBodyFormat::Html => PREFER_BODY_HTML,
};
let prefer: Vec<String> = vec![PREFER_TIMEZONE_UTC.to_owned(), body_prefer.to_owned()];
let want_attachments = ctx.event_attachments;
let pool: Pool<String, R> = Pool::new(ctx.workers, move |id: String| {
let url = endpoints.event(&id);
let prefer_refs: Vec<&str> = prefer.iter().map(String::as_str).collect();
let result = client.get_json_with_prefer(&url, &prefer_refs);
(id, result)
let attachments = match (&result, want_attachments) {
(Ok(raw), true) if raw.get("hasAttachments").and_then(Value::as_bool) == Some(true) => {
fetch_event_attachments(&client, &endpoints, &id)
}
_ => Vec::new(),
};
(id, result, attachments)
});
for id in ids {
pool.submit(id.clone());
@@ -177,10 +209,6 @@ fn fetch_events(
out
}
fn is_series_master(raw: &Value) -> bool {
raw.get("recurrence").is_some_and(|r| !r.is_null())
}
fn exception_windows(years: i32) -> Vec<(String, String)> {
let today = chrono::Utc::now().date_naive();
let span = chrono::Duration::days(
@@ -244,105 +272,206 @@ fn fetch_calendar_exceptions(
out
}
fn merge_exception_into(master: &mut ConvertedEvent, ex: &ConvertedEvent) {
let Some(raw_key) = ex.original_start.as_deref() else {
return;
};
let master_tz = master
.data
.get("timeZone")
.and_then(Value::as_str)
.map(str::to_owned);
let key = normalise_override_key(raw_key, master_tz.as_deref());
let Value::Object(map) = &mut master.data else {
return;
};
let overrides = map
.entry("recurrenceOverrides".to_owned())
.or_insert_with(|| Value::Object(serde_json::Map::new()));
let Value::Object(overrides) = overrides else {
return;
};
overrides.insert(key, override_patch_from_event(&ex.data));
}
fn merge_exception_into_existing(
fn apply_series_expansion(
conn: &mut Connection,
ctx: &GraphCoordinator<'_>,
master_graph_id: &str,
ex: &ConvertedEvent,
calendar_id: &str,
series_masters: &[String],
counts: &mut TypeCounts,
) {
let local_id = match exchange_graph_ids::local_for_graph_id(
conn,
ctx.source_id,
exchange_graph_ids::CALENDAR_EVENT,
master_graph_id,
) {
Ok(Some(id)) => id,
Ok(None) => {
counts.skipped += 1;
return;
}
Err(e) => {
counts.failed += 1;
ctx.logger.warn(&format!(
"graph exception merge: lookup of master {master_graph_id} failed: {e}"
));
return;
}
};
let Some(raw_key) = ex.original_start.as_deref() else {
counts.skipped += 1;
return;
};
if let Err(e) = merge_persisted_master(conn, local_id, raw_key, &ex.data) {
counts.failed += 1;
ctx.logger.warn(&format!(
"graph exception merge into stored master {master_graph_id} failed: {e}"
));
} else {
counts.fetched += 1;
}
}
) -> Result<(), Error> {
let exceptions = fetch_calendar_exceptions(ctx, calendar_id, counts);
let occurrences = fetch_calendar_occurrences(ctx, calendar_id, counts);
let windows = exception_windows(ctx.exception_window_years);
let ids: HashMap<String, i64> =
exchange_graph_ids::ids_of_type(conn, ctx.source_id, exchange_graph_ids::CALENDAR_EVENT)?;
fn merge_persisted_master(
conn: &Connection,
local_id: i64,
raw_key: &str,
ex_data: &Value,
) -> Result<(), String> {
let tx = conn.unchecked_transaction().map_err(|e| e.to_string())?;
let row: String = tx
.query_row(
let mut by_master: HashMap<&str, Vec<&ConvertedEvent>> = HashMap::new();
for (_, ex) in &exceptions {
if let Some(master) = ex.series_master_id.as_deref() {
by_master.entry(master).or_default().push(ex);
}
}
for master_id in series_masters {
let Some(local_id) = ids.get(master_id).copied() else {
continue;
};
let stored: String = match conn.query_row(
"SELECT data FROM calendar_events WHERE id = ?1",
params![local_id],
|row| row.get(0),
)
.map_err(|e| e.to_string())?;
let mut data: Value = serde_json::from_str(&row).map_err(|e| e.to_string())?;
let master_tz = data
.get("timeZone")
.and_then(Value::as_str)
.map(str::to_owned);
let key = normalise_override_key(raw_key, master_tz.as_deref());
if let Value::Object(map) = &mut data {
let entry = map
.entry("recurrenceOverrides".to_owned())
.or_insert_with(|| Value::Object(serde_json::Map::new()));
if let Value::Object(overrides) = entry {
overrides.insert(key, override_patch_from_event(ex_data));
) {
Ok(v) => v,
Err(_) => continue,
};
let Ok(mut card) = serde_json::from_str::<Value>(&stored) else {
continue;
};
let before = card.clone();
for ex in by_master.get(master_id.as_str()).into_iter().flatten() {
let Some(raw_key) = ex.original_start.as_deref() else {
continue;
};
let tz = card
.get("timeZone")
.and_then(Value::as_str)
.map(str::to_owned);
let key = normalise_override_key(raw_key, tz.as_deref());
if let Some(map) = card.as_object_mut() {
let entry = map
.entry("recurrenceOverrides".to_owned())
.or_insert_with(|| Value::Object(serde_json::Map::new()));
if let Some(overrides) = entry.as_object_mut() {
overrides.insert(key, override_patch_from_event(&ex.data));
}
}
}
let cancelled = cancelled_occurrence_keys(
&card,
master_id,
&occurrences,
by_master
.get(master_id.as_str())
.map(Vec::as_slice)
.unwrap_or(&[]),
&windows,
);
if !cancelled.is_empty()
&& let Some(map) = card.as_object_mut()
{
let entry = map
.entry("recurrenceOverrides".to_owned())
.or_insert_with(|| Value::Object(serde_json::Map::new()));
if let Some(overrides) = entry.as_object_mut() {
for key in cancelled {
overrides
.entry(key)
.or_insert_with(|| json!({"excluded": true}));
}
}
}
if card != before {
let tx = conn.unchecked_transaction()?;
tx.execute(
"UPDATE calendar_events SET data = ?1 WHERE id = ?2",
params![card.to_string(), local_id],
)?;
tx.commit()?;
counts.updated += 1;
}
}
tx.execute(
"UPDATE calendar_events SET data = ?1 WHERE id = ?2",
params![data.to_string(), local_id],
)
.map_err(|e| e.to_string())?;
tx.commit().map_err(|e| e.to_string())?;
Ok(())
}
fn cancelled_occurrence_keys(
card: &Value,
master_id: &str,
occurrences: &HashMap<String, Vec<String>>,
exceptions: &[&ConvertedEvent],
windows: &[(String, String)],
) -> Vec<String> {
let Some(rule) = card.get("recurrenceRule") else {
return Vec::new();
};
let Some(start) = card.get("start").and_then(Value::as_str) else {
return Vec::new();
};
let (start_date, start_time) = match start.split_once('T') {
Some((d, t)) => (d, t),
None => (start, "00:00:00"),
};
let Ok(series_start) = chrono::NaiveDate::parse_from_str(start_date, "%Y-%m-%d") else {
return Vec::new();
};
let Some(pattern) =
crate::exchange_graph::expand::graph_pattern_from_jscalendar(rule, series_start)
else {
return Vec::new();
};
let tz = card.get("timeZone").and_then(Value::as_str);
let mut actual: std::collections::HashSet<String> = std::collections::HashSet::new();
for utc in occurrences.get(master_id).into_iter().flatten() {
actual.insert(local_date_of(utc, tz));
}
for ex in exceptions {
if let Some(raw) = ex.original_start.as_deref() {
let key = normalise_override_key(raw, tz);
actual.insert(key.split('T').next().unwrap_or(&key).to_owned());
}
}
let mut out = Vec::new();
for (from, to) in windows {
let (Ok(w0), Ok(w1)) = (
chrono::NaiveDate::parse_from_str(&from[..10], "%Y-%m-%d"),
chrono::NaiveDate::parse_from_str(&to[..10], "%Y-%m-%d"),
) else {
continue;
};
let Some(expected) = crate::exchange_graph::expand::expected_dates(&pattern, w0, w1) else {
return Vec::new();
};
for date in expected {
let key = date.format("%Y-%m-%d").to_string();
if !actual.contains(&key) {
out.push(format!("{key}T{start_time}"));
}
}
}
out.sort();
out.dedup();
out
}
fn local_date_of(utc: &str, tz: Option<&str>) -> String {
let key = normalise_override_key(utc, tz);
key.split('T').next().unwrap_or(&key).to_owned()
}
fn fetch_calendar_occurrences(
ctx: &GraphCoordinator<'_>,
calendar_id: &str,
counts: &mut TypeCounts,
) -> HashMap<String, Vec<String>> {
let mut out: HashMap<String, Vec<String>> = HashMap::new();
for (from, to) in exception_windows(ctx.exception_window_years) {
let url = ctx
.endpoints
.calendar_occurrences(calendar_id, &from, &to, ctx.top);
match api::collect_all_values(ctx.client, &url, &[PREFER_TIMEZONE_UTC]) {
Ok(values) => {
for raw in values {
let Some(master) = raw.get("seriesMasterId").and_then(Value::as_str) else {
continue;
};
let Some(start) = raw
.get("start")
.and_then(|s| s.get("dateTime"))
.and_then(Value::as_str)
else {
continue;
};
out.entry(master.to_owned())
.or_default()
.push(start.to_owned());
}
}
Err(e) => {
counts.failed += 1;
ctx.logger.warn(&format!(
"graph calendar {calendar_id} occurrence sweep {from}..{to} failed: {e}; \
deleted occurrences cannot be detected for this calendar"
));
}
}
}
out
}
pub fn normalise_override_key(raw: &str, master_tz: Option<&str>) -> String {
if let Some(tz_name) = master_tz
&& let Some(local) = utc_offset_to_local(raw, tz_name)
@@ -398,12 +527,14 @@ fn insert_events_chunked(
ctx: &GraphCoordinator<'_>,
calendar_local_id: i64,
pairs: &[(String, ConvertedEvent)],
attachments: &HashMap<String, Vec<EventAttachment>>,
counts: &mut TypeCounts,
) -> Result<(), Error> {
for chunk in pairs.chunks(CHUNK_SIZE) {
let tx = conn.unchecked_transaction()?;
for (graph_id, event) in chunk {
insert_event_in_tx(&tx, ctx, calendar_local_id, graph_id, event, counts)?;
let files = attachments.get(graph_id).map(Vec::as_slice).unwrap_or(&[]);
insert_event_in_tx(&tx, ctx, calendar_local_id, graph_id, event, files, counts)?;
}
tx.commit()?;
}
@@ -416,10 +547,13 @@ fn insert_event_in_tx(
calendar_local_id: i64,
graph_id: &str,
event: &ConvertedEvent,
attachments: &[EventAttachment],
counts: &mut TypeCounts,
) -> Result<(), Error> {
let calendar_ids = json!([calendar_local_id]).to_string();
let data = event.data.to_string();
let mut card = event.data.clone();
attach_links(tx, &mut card, attachments)?;
let data = card.to_string();
tx.execute(
"INSERT INTO calendar_events (calendar_ids, is_draft, use_default_alerts, data, data_type)
VALUES (?1, ?2, ?3, ?4, 'Event')",
@@ -442,6 +576,39 @@ fn insert_event_in_tx(
Ok(())
}
fn attach_links(
tx: &Transaction<'_>,
card: &mut Value,
attachments: &[EventAttachment],
) -> Result<(), Error> {
if attachments.is_empty() {
return Ok(());
}
let mut links = serde_json::Map::new();
for (idx, (name, content_type, bytes)) in (1u32..).zip(attachments) {
let blob_id = crate::db::blobs::intern_blob(tx, bytes)?;
links.insert(
idx.to_string(),
json!({
"@type": "Link",
"@blob": blob_id,
"contentType": content_type,
"title": name,
"rel": "enclosure",
}),
);
}
if let Some(map) = card.as_object_mut() {
match map.get_mut("links").and_then(Value::as_object_mut) {
Some(existing) => existing.extend(links),
None => {
map.insert("links".to_owned(), Value::Object(links));
}
}
}
Ok(())
}
fn delete_vanished(
conn: &mut Connection,
source_id: i64,
src/sync/import_exchange_graph/contacts.rs
+93 -10
View File
@@ -21,6 +21,44 @@ use crate::sync::import_jmap::pool::Pool;
use super::coordinator::{CHUNK_SIZE, GraphCoordinator};
use super::folders::ContactFolder;
fn fetch_photo(
client: &crate::exchange_graph::client::GraphClient,
endpoints: &crate::exchange_graph::api::Endpoints,
id: &str,
) -> Option<ContactPhoto> {
let url = endpoints.contact_photo(id);
let resp = client
.get_with_prefer(&url, crate::exchange_graph::client::Accept::Binary, &[])
.ok()?;
if resp.body.is_empty() {
return None;
}
let media_type = sniff_image_type(&resp.body)
.map(str::to_owned)
.or_else(|| resp.content_type.clone())
.unwrap_or_else(|| "application/octet-stream".to_owned());
Some((resp.body, media_type))
}
fn sniff_image_type(bytes: &[u8]) -> Option<&'static str> {
if bytes.starts_with(b"\x89PNG\r\n\x1a\n") {
return Some("image/png");
}
if bytes.starts_with(&[0xFF, 0xD8, 0xFF]) {
return Some("image/jpeg");
}
if bytes.starts_with(b"GIF87a") || bytes.starts_with(b"GIF89a") {
return Some("image/gif");
}
if bytes.len() >= 12 && bytes.starts_with(b"RIFF") && &bytes[8..12] == b"WEBP" {
return Some("image/webp");
}
if bytes.starts_with(b"BM") {
return Some("image/bmp");
}
None
}
pub fn reconcile_all(
conn: &mut Connection,
ctx: &GraphCoordinator<'_>,
@@ -71,11 +109,11 @@ pub fn reconcile_all(
continue;
}
let fetched = fetch_contacts(ctx, &new_ids);
let mut converted: Vec<(String, ConvertedContact)> = Vec::new();
for (graph_id, result) in fetched {
let mut converted: Vec<(String, ConvertedContact, Option<ContactPhoto>)> = Vec::new();
for (graph_id, result, photo) in fetched {
match result {
Ok(raw) => match convert_contact(&raw) {
Ok(c) => converted.push((graph_id, c)),
Ok(c) => converted.push((graph_id, c, photo)),
Err(e) => {
counts.failed += 1;
ctx.logger
@@ -104,16 +142,25 @@ pub fn reconcile_all(
Ok(())
}
pub type ContactPhoto = (Vec<u8>, String);
fn fetch_contacts(
ctx: &GraphCoordinator<'_>,
ids: &[String],
) -> Vec<(String, Result<Value, GraphError>)> {
type R = (String, Result<Value, GraphError>);
) -> Vec<(String, Result<Value, GraphError>, Option<ContactPhoto>)> {
type R = (String, Result<Value, GraphError>, Option<ContactPhoto>);
let client = ctx.client.clone();
let endpoints: crate::exchange_graph::api::Endpoints = (*ctx.endpoints).clone();
let want_photos = ctx.contact_photos;
let pool: Pool<String, R> = Pool::new(ctx.workers, move |id: String| {
let url = endpoints.contact(&id);
(id, client.get_json_with_prefer(&url, &[]))
let contact = client.get_json_with_prefer(&url, &[]);
let photo = if want_photos && contact.is_ok() {
fetch_photo(&client, &endpoints, &id)
} else {
None
};
(id, contact, photo)
});
for id in ids {
pool.submit(id.clone());
@@ -131,13 +178,13 @@ fn insert_contacts_chunked(
conn: &mut Connection,
ctx: &GraphCoordinator<'_>,
book_local_id: i64,
pairs: &[(String, ConvertedContact)],
pairs: &[(String, ConvertedContact, Option<ContactPhoto>)],
counts: &mut TypeCounts,
) -> Result<(), Error> {
for chunk in pairs.chunks(CHUNK_SIZE) {
let tx = conn.unchecked_transaction()?;
for (graph_id, contact) in chunk {
insert_contact_in_tx(&tx, ctx, book_local_id, graph_id, contact, counts)?;
for (graph_id, contact, photo) in chunk {
insert_contact_in_tx(&tx, ctx, book_local_id, graph_id, contact, photo, counts)?;
}
tx.commit()?;
}
@@ -150,10 +197,28 @@ fn insert_contact_in_tx(
book_local_id: i64,
graph_id: &str,
converted: &ConvertedContact,
photo: &Option<ContactPhoto>,
counts: &mut TypeCounts,
) -> Result<(), Error> {
let book_ids = json!([book_local_id]).to_string();
let data = converted.data.to_string();
let mut card = converted.data.clone();
if let Some((bytes, media_type)) = photo {
let blob_id = crate::db::blobs::intern_blob(tx, bytes)?;
if let Some(map) = card.as_object_mut() {
map.insert(
"media".to_owned(),
json!({
"photo": {
"@type": "Media",
"kind": "photo",
"@blob": blob_id,
"mediaType": media_type,
}
}),
);
}
}
let data = card.to_string();
tx.execute(
"INSERT INTO contact_cards (uid, address_book_ids, data) VALUES (?1, ?2, ?3)",
params![converted.uid, book_ids, data],
@@ -204,3 +269,21 @@ fn delete_vanished(
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::sniff_image_type;
#[test]
fn photo_type_comes_from_the_bytes_not_the_header() {
let png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR";
assert_eq!(sniff_image_type(png), Some("image/png"));
assert_eq!(
sniff_image_type(&[0xFF, 0xD8, 0xFF, 0xE0]),
Some("image/jpeg")
);
assert_eq!(sniff_image_type(b"GIF89a..."), Some("image/gif"));
assert_eq!(sniff_image_type(b"RIFF____WEBPVP8 "), Some("image/webp"));
assert_eq!(sniff_image_type(b"not an image"), None);
}
}
src/sync/import_exchange_graph/coordinator.rs
+8
View File
@@ -43,6 +43,8 @@ pub struct GraphImportConfig {
pub graph_connections: usize,
pub top: usize,
pub exception_window_years: i32,
pub contact_photos: bool,
pub event_attachments: bool,
pub allow_source_change: bool,
}
@@ -57,6 +59,8 @@ pub struct GraphCoordinator<'a> {
pub logger: crate::logging::Logger,
pub event_body_format: EventBodyFormat,
pub exception_window_years: i32,
pub contact_photos: bool,
pub event_attachments: bool,
}
pub fn run(common: CommonConfig, config: GraphImportConfig) -> Result<Summary, Error> {
@@ -142,6 +146,8 @@ pub fn run(common: CommonConfig, config: GraphImportConfig) -> Result<Summary, E
logger,
event_body_format: config.event_body_format,
exception_window_years: config.exception_window_years,
contact_photos: config.contact_photos,
event_attachments: config.event_attachments,
};
let primary = !matches!(config.mailbox_kind, MailboxKind::Archive);
@@ -503,6 +509,8 @@ mod tests {
graph_connections: 4,
top: 100,
exception_window_years: 5,
contact_photos: true,
event_attachments: true,
allow_source_change: false,
};
let url = canonical_session_url(&config);