ab/meguri
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5260c2828baac727c82915664613e0a3e942a79e
meguri/src/providers/jsm.rs
T
Ultradesu 5260c2828b
Checks / test (push) Failing after 2m25s
Details
NEXT
2026-10-07 15:30:31 +01:00

1211 lines
47 KiB
Rust
Raw Blame History

use crate::{
domain::*,
error::{Error, Result, ensure},
models::ProviderObjectRow,
providers::Publication,
store::{Store, WorkspaceRecord},
};
use chrono::{DateTime, Timelike, Utc};
use reqwest::{Client, Method, StatusCode};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use uuid::Uuid;
#[derive(Clone)]
pub struct Jsm {
client: Client,
base: String,
email: String,
token: String,
team: String,
users: BTreeMap<Uuid, String>,
directory_base: Option<String>,
}
impl Jsm {
pub fn from_environment(record: &WorkspaceRecord) -> Result<Self> {
let Integration::Jsm {
cloud_id,
team_id,
users,
..
} = &record.document.integration
else {
return Err(Error::Validation(
crate::i18n::text("err_jsm_is_not_configured_4e6594").into(),
));
};
let prefix = format!(
"MEGURI_JSM_{}",
record.id.simple().to_string().to_uppercase()
);
let email = std::env::var(format!("{prefix}_EMAIL")).map_err(|_| {
Error::Provider(crate::i18n::message(
"err_missing_prefix_email_deployment_secret_6e5753",
&[("prefix", (prefix).to_string())],
))
})?;
let token = std::env::var(format!("{prefix}_TOKEN")).map_err(|_| {
Error::Provider(crate::i18n::message(
"err_missing_prefix_token_deployment_secret_1f0352",
&[("prefix", (prefix).to_string())],
))
})?;
Ok(Self::new(
format!("https://api.atlassian.com/jsm/ops/api/{cloud_id}/v1"),
email,
token,
team_id.clone(),
users.clone(),
)?
.with_directory_base(format!(
"https://api.atlassian.com/ex/jira/{cloud_id}/rest/api/3"
)))
}
pub fn new(
base: String,
email: String,
token: String,
team: String,
users: BTreeMap<Uuid, String>,
) -> Result<Self> {
let client = Client::builder()
.timeout(std::time::Duration::from_secs(20))
.redirect(reqwest::redirect::Policy::none())
.build()
.map_err(|_| {
Error::Provider(
crate::i18n::text("err_could_not_initialize_jsm_client_09bad0").into(),
)
})?;
Ok(Self {
client,
base,
email,
token,
team,
users,
directory_base: None,
})
}
/// Explicit injection for the fake HTTP server; production always uses Atlassian's host.
pub fn with_directory_base(mut self, base: String) -> Self {
self.directory_base = Some(base);
self
}
async fn resolve_email(&self, email: &str) -> Result<String> {
let base = self.directory_base.as_ref().ok_or_else(|| {
Error::Provider(
crate::i18n::text("err_configure_the_atlassian_directory_or_provide_acc_266b32")
.into(),
)
})?;
let mut matches = BTreeSet::new();
for offset in (0..1000).step_by(100) {
let response = self
.client
.get(format!("{base}/user/search"))
.basic_auth(&self.email, Some(&self.token))
.query(&[
("query", email.to_owned()),
("startAt", offset.to_string()),
("maxResults", "100".into()),
])
.send()
.await
.map_err(|_| {
Error::Provider(
crate::i18n::text("err_atlassian_user_lookup_could_not_connect_6f312b")
.into(),
)
})?;
if !response.status().is_success() {
return Err(Error::Provider(crate::i18n::message(
"err_atlassian_user_lookup_returned_http_p0_grant_bro_a0f887",
&[("p0", (response.status().as_u16()).to_string())],
)));
}
let users: Vec<Value> = response.json().await.map_err(|_| {
Error::Provider(
crate::i18n::text("err_invalid_atlassian_directory_response_1c1889").into(),
)
})?;
for user in &users {
if user["active"].as_bool() == Some(true)
&& user["emailAddress"]
.as_str()
.is_some_and(|e| e.trim().eq_ignore_ascii_case(email))
&& let Some(id) = user["accountId"].as_str().filter(|id| !id.is_empty())
{
matches.insert(id.to_owned());
}
}
if users.len() < 100 {
break;
}
if offset == 900 {
return Err(Error::Provider(
crate::i18n::text(
"err_atlassian_search_limit_reached_provide_an_explic_d1fe8a",
)
.into(),
));
}
}
if matches.len() != 1 {
return Err(Error::Provider(crate::i18n::message(
"err_could_not_find_one_active_atlassian_account_with_39e9a9",
&[("email", (email).to_string())],
)));
}
Ok(matches.into_iter().next().unwrap())
}
async fn resolve_people(&self, record: &WorkspaceRecord, plan: &Publication) -> Result<Self> {
let mut resolved = self.clone();
let mut ids: BTreeSet<_> = plan
.schedules
.iter()
.flat_map(|s| {
s.periods.iter().map(|p| p.user_id).chain(
s.recurring
.iter()
.flat_map(|r| r.participants.iter().copied()),
)
})
.collect();
ids.extend(
plan.escalations
.iter()
.flat_map(|e| e.steps.iter())
.filter_map(|s| {
if let Recipient::User { id } = s.recipient {
Some(id)
} else {
None
}
}),
);
for id in ids {
if resolved.users.contains_key(&id) {
continue;
}
let email = record
.document
.person(id)
.and_then(|p| p.email.as_deref())
.ok_or_else(|| {
Error::Provider(crate::i18n::message(
"err_person_id_needs_an_email_or_an_atlassian_account_360521",
&[("id", (id).to_string())],
))
})?;
resolved.users.insert(id, self.resolve_email(email).await?);
}
Ok(resolved)
}
async fn request(
&self,
method: Method,
path: &str,
body: Option<&Value>,
) -> Result<Option<Value>> {
let mut req = self
.client
.request(method.clone(), format!("{}{path}", self.base))
.basic_auth(&self.email, Some(&self.token))
.header("Accept", "application/json");
if let Some(body) = body {
req = req.json(body);
}
let response = req.send().await.map_err(|_| {
Error::Provider(
crate::i18n::text("err_jsm_request_timed_out_or_could_not_connect_5ce428").into(),
)
})?;
if response.status() == StatusCode::NOT_FOUND {
return Ok(None);
}
if !response.status().is_success() {
return Err(Error::Provider(crate::i18n::message(
"err_jsm_returned_http_p0_for_p1_check_credentials_pl_b31caf",
&[
("p0", (response.status().as_u16()).to_string()),
("p1", (method).to_string()),
],
)));
}
if response.status() == StatusCode::NO_CONTENT {
return Ok(Some(Value::Null));
}
Ok(Some(response.json().await.map_err(|_| {
Error::Provider(
crate::i18n::text("err_jsm_returned_an_invalid_json_response_af8daf").into(),
)
})?))
}
async fn list(&self, path: &str) -> Result<Vec<Value>> {
const PAGE_SIZE: usize = 50;
let mut all = vec![];
let mut offset = 0;
loop {
let sep = if path.contains('?') { '&' } else { '?' };
let page = self
.request(
Method::GET,
&format!("{path}{sep}offset={offset}&size={PAGE_SIZE}"),
None,
)
.await?
.ok_or_else(|| {
Error::Provider(
crate::i18n::text("err_jsm_collection_was_not_found_60d46c").into(),
)
})?;
let values = page
.get("values")
.and_then(Value::as_array)
.ok_or_else(|| {
Error::Provider(crate::i18n::text("err_jsm_collection_has_no_values").into())
})?;
let count = values.len();
all.extend(values.iter().cloned());
if page
.pointer("/links/next")
.and_then(Value::as_str)
.is_none_or(str::is_empty)
&& count < PAGE_SIZE
{
break;
}
if count == 0 {
break;
}
offset += count;
ensure(
offset <= 20000,
crate::i18n::text("err_jsm_collection_exceeds_the_supported_pagination_a1136e"),
)?;
}
Ok(all)
}
fn mapped(&self, id: Uuid) -> Result<String> {
self.users.get(&id).cloned().ok_or_else(|| {
Error::Validation(crate::i18n::message(
"err_employee_id_has_no_jsm_account_mapping_7b6db1",
&[("id", (id).to_string())],
))
})
}
fn collection(&self, kind: &str) -> String {
match kind {
"schedule" => "/schedules".into(),
"escalation" => format!("/teams/{}/escalations", self.team),
_ => format!("/teams/{}/routing-rules", self.team),
}
}
async fn read(&self, kind: &str, id: &str) -> Result<Option<Value>> {
let Some(mut object) = self
.request(
Method::GET,
&format!("{}/{id}", self.collection(kind)),
None,
)
.await?
else {
return Ok(None);
};
if kind == "schedule" {
object["rotations"] =
Value::Array(self.list(&format!("/schedules/{id}/rotations")).await?);
}
Ok(Some(object))
}
#[allow(clippy::too_many_arguments)]
async fn synchronize(
&self,
db: &Store,
record: &WorkspaceRecord,
kind: &str,
key: &str,
desired: &Value,
previous: Option<&ProviderObjectRow>,
overwrite: bool,
as_of: DateTime<Utc>,
) -> Result<String> {
let local_id = key
.split_once(':')
.and_then(|(_, id)| Uuid::parse_str(id).ok())
.expect("provider object keys contain a resource UUID");
// A deleted resource is absent desired state, not an edit to an existing
// resource. Clear stale identity before discovery so a lost replacement
// create response can be recovered by its stable ownership marker.
let mut previous = previous;
if let Some(old) = previous
&& self.read(kind, &old.remote_id).await?.is_none()
{
db.forget(record.id, key).await?;
db.forget(record.id, &format!("pending:{key}")).await?;
if kind == "schedule" {
let prefix = format!("coverage:{local_id}:");
for child in db
.objects(record.id)
.await?
.iter()
.filter(|r| r.object_key.starts_with(&prefix))
{
db.forget(record.id, &child.object_key).await?;
}
}
previous = None;
}
let marker = ownership_marker(record.id, local_id);
let legacy_name = format!("meguri-{}-{}", record.id.simple(), local_id.simple());
let existing_id = if let Some(previous) = previous {
Some(previous.remote_id.clone())
} else {
let found: Vec<_> = self
.list(&self.collection(kind))
.await?
.into_iter()
.filter(|v| {
v["name"]
.as_str()
.is_some_and(|name| name.ends_with(&marker) || name == legacy_name)
})
.collect();
ensure(
found.len() <= 1,
crate::i18n::text("err_multiple_jsm_resources_have_the_same_meguri_owne_889887"),
)?;
found
.first()
.and_then(|v| v["id"].as_str())
.map(str::to_string)
};
let current = if let Some(id) = &existing_id {
self.read(kind, id).await?
} else {
None
};
let expected = previous
.map(|p| serde_json::from_str::<Value>(&p.content))
.transpose()?;
let pending_key = format!("pending:{key}");
let pending =
db.objects(record.id).await?.into_iter().find(|r| {
r.object_key == pending_key && Some(&r.remote_id) == existing_id.as_ref()
});
let pending_expected = pending
.as_ref()
.map(|r| serde_json::from_str::<Value>(&r.content))
.transpose()?;
let matches_desired = current.as_ref().is_some_and(|c| equivalent(c, desired));
if let Some(current) = &current {
let known = expected
.as_ref()
.is_some_and(|last| equivalent(current, last))
|| pending_expected
.as_ref()
.is_some_and(|next| equivalent(current, next));
let legacy_matches =
if previous.is_none() && current["name"].as_str() == Some(&legacy_name) {
let mut legacy = desired.clone();
legacy["name"] = json!(legacy_name);
equivalent(current, &legacy)
} else {
false
};
if !matches_desired && !known && !legacy_matches && !overwrite {
return Err(Error::Conflict(crate::i18n::message(
"err_jsm_kind_changed_outside_meguri_publication_paus_ac8220",
&[("kind", (kind).to_string())],
)));
}
}
if kind == "schedule"
&& current.is_some()
&& let Some(id) = &existing_id
{
self.check_coverage(db, record.id, local_id, id, overwrite, as_of)
.await?;
}
let id = if matches_desired {
existing_id.unwrap()
} else if current.is_some() {
let id = existing_id.unwrap();
let current = current.as_ref().unwrap();
if kind == "schedule"
&& current["rotations"]
.as_array()
.is_some_and(|r| r.len() == 1)
&& desired["rotations"]
.as_array()
.is_some_and(|r| r.len() == 1)
{
let rotation_id = current["rotations"][0]["id"]
.as_str()
.filter(|v| safe_id(v))
.ok_or_else(|| {
Error::Provider(
crate::i18n::text("err_unexpected_jsm_resource_identifier_dedd70")
.into(),
)
})?;
let mut metadata = desired.clone();
metadata.as_object_mut().unwrap().remove("rotations");
if !equivalent(current, &metadata) {
let mut intermediate = metadata.clone();
intermediate["rotations"] = current["rotations"].clone();
self.schedule_step(
db,
record.id,
key,
&id,
&format!("/schedules/{id}"),
&metadata,
&intermediate,
)
.await?;
}
if !equivalent(&current["rotations"][0], &desired["rotations"][0]) {
self.schedule_step(
db,
record.id,
key,
&id,
&format!("/schedules/{id}/rotations/{rotation_id}"),
&desired["rotations"][0],
desired,
)
.await?;
}
} else {
self.request(
Method::PATCH,
&format!("{}/{id}", self.collection(kind)),
Some(desired),
)
.await?
.ok_or_else(|| {
Error::Conflict(
crate::i18n::text("err_jsm_resource_disappeared_during_publication").into(),
)
})?;
}
id
} else {
let response = self
.request(Method::POST, &self.collection(kind), Some(desired))
.await?
.ok_or_else(|| {
Error::Provider(
crate::i18n::text("err_jsm_create_endpoint_was_not_found_1be86a").into(),
)
})?;
response["id"]
.as_str()
.ok_or_else(|| {
Error::Provider(
crate::i18n::text("err_jsm_created_resource_without_an_id").into(),
)
})?
.to_string()
};
ensure(
safe_id(&id),
crate::i18n::text("err_unexpected_jsm_resource_identifier_dedd70"),
)?;
let actual = self.read(kind, &id).await?.ok_or_else(|| {
Error::Provider(
crate::i18n::text("err_jsm_resource_is_missing_after_publication_3448ad").into(),
)
})?;
if !equivalent(&actual, desired) {
return Err(Error::Provider(crate::i18n::message(
"err_jsm_kind_read_back_does_not_match_the_requested_63440e",
&[("kind", (kind).to_string())],
)));
}
db.checkpoint(record.id, key, &id, desired).await?;
db.forget(record.id, &pending_key).await?;
Ok(id)
}
#[allow(clippy::too_many_arguments)]
async fn schedule_step(
&self,
db: &Store,
workspace: Uuid,
key: &str,
id: &str,
path: &str,
body: &Value,
expected: &Value,
) -> Result<()> {
let pending = format!("pending:{key}");
db.checkpoint(workspace, &pending, id, expected).await?;
self.request(Method::PATCH, path, Some(body))
.await?
.ok_or_else(|| {
Error::Conflict(
crate::i18n::text("err_jsm_resource_disappeared_during_publication").into(),
)
})?;
let actual = self.read("schedule", id).await?.ok_or_else(|| {
Error::Provider(
crate::i18n::text("err_jsm_resource_is_missing_after_publication_3448ad").into(),
)
})?;
ensure(
equivalent(&actual, expected),
crate::i18n::message(
"err_jsm_kind_read_back_does_not_match_the_requested_63440e",
&[("kind", "schedule".into())],
),
)?;
db.checkpoint(workspace, key, id, expected).await?;
db.forget(workspace, &pending).await?;
Ok(())
}
fn schedule_payloads(
&self,
schedule: &super::PublishedSchedule,
as_of: DateTime<Utc>,
) -> Result<(Vec<Value>, Vec<Value>)> {
if let Some(rotation) = &schedule.recurring {
let zone = schedule.timezone.parse::<chrono_tz::Tz>().map_err(|_| {
Error::Provider(crate::i18n::text("jsm_invalid_coverage_interval").into())
})?;
if half_hour_floor(rotation.start, zone) == rotation.start
&& half_hour_floor(rotation.end, zone) == rotation.end
{
let participants = rotation
.participants
.iter()
.map(|id| self.mapped(*id).map(|id| json!({"type":"user","id":id})))
.collect::<Result<Vec<_>>>()?;
let (kind, length) = if rotation.shift_days.is_multiple_of(7) {
("weekly", rotation.shift_days / 7)
} else {
("daily", rotation.shift_days)
};
return Ok((
vec![
json!({"name":schedule.name,"startDate":rotation.start,"endDate":rotation.end,"type":kind,"length":length,"participants":participants}),
],
vec![],
));
}
}
if !needs_precise_coverage(schedule)? {
let zone = schedule.timezone.parse::<chrono_tz::Tz>().map_err(|_| {
Error::Provider(crate::i18n::text("jsm_invalid_coverage_interval").into())
})?;
let rotations = schedule.periods.iter().map(|p| Ok(json!({
"name":if schedule.periods.len() == 1 { schedule.name.clone() } else {
format!("{} · {}", schedule.name, p.start.with_timezone(&zone).format("%d-%m-%Y %H:%M %:z"))
}, "startDate":p.start, "endDate":p.end,
"type":"daily", "length":1, "participants":[{"type":"user","id":self.mapped(p.user_id)?}]
}))).collect::<Result<Vec<_>>>()?;
return Ok((rotations, vec![]));
}
let zone = schedule.timezone.parse::<chrono_tz::Tz>().map_err(|_| {
Error::Provider(crate::i18n::text("jsm_invalid_coverage_interval").into())
})?;
let current = schedule
.periods
.iter()
.find(|p| p.start <= as_of && as_of < p.end);
let base_person = current.map(|p| self.mapped(p.user_id)).transpose()?;
let first = current.unwrap_or(&schedule.periods[0]);
let last_end = schedule.periods.last().unwrap().end;
let start = half_hour_floor(first.start, zone);
let floor_end = half_hour_floor(last_end, zone);
let end = if floor_end == last_end {
floor_end
} else {
floor_end + chrono::Duration::minutes(30)
};
let participant = base_person.as_ref().map_or_else(
|| json!({"type":"noone"}),
|id| json!({"type":"user","id":id}),
);
let rotations = vec![
json!({"name":schedule.name, "startDate":start, "endDate":end,
"type":"daily", "length":1, "participants":[participant]}),
];
let mut coverage = vec![];
// JSM rejects new overrides starting in the past. The already effective
// assignee is the base; overrides describe only future transitions.
for p in schedule.periods.iter().filter(|p| p.start > as_of) {
coverage.push(
json!({"responder":{"type":"user","id":self.mapped(p.user_id)?},
"startDate":milliseconds(p.start), "endDate":milliseconds(p.end)}),
);
}
if base_person.is_some() {
for pair in schedule.periods.windows(2) {
if pair[0].end > as_of && pair[0].end < pair[1].start {
coverage.push(json!({"responder":{"type":"noone"},
"startDate":milliseconds(pair[0].end), "endDate":milliseconds(pair[1].start)}));
}
}
if last_end > as_of && last_end < end {
coverage.push(json!({"responder":{"type":"noone"},
"startDate":milliseconds(last_end), "endDate":end}));
}
}
Ok((rotations, coverage))
}
// Empty remote IDs are durable create intents, never verified publications.
async fn check_coverage(
&self,
db: &Store,
workspace: Uuid,
schedule: Uuid,
remote: &str,
overwrite: bool,
_as_of: DateTime<Utc>,
) -> Result<()> {
let prefix = format!("coverage:{schedule}:");
let rows = db.objects(workspace).await?;
let mut owned = vec![];
for row in rows.iter().filter(|r| r.object_key.starts_with(&prefix)) {
let content: Value = serde_json::from_str(&row.content)?;
if content["scheduleId"].as_str() == Some(remote) {
owned.push((row, content));
}
}
let items = self.list(&format!("/schedules/{remote}/overrides")).await?;
for item in &items {
let alias = coverage_alias(item)?;
if let Some((_, content)) = owned.iter().find(|(row, _)| row.remote_id == alias) {
if !overwrite && !equivalent(item, &content["payload"]) {
return Err(Error::Conflict(
crate::i18n::text("jsm_coverage_conflict").into(),
));
}
continue;
}
let pending: Vec<_> = owned
.iter()
.filter(|(row, content)| {
row.remote_id.is_empty() && equivalent(item, &content["payload"])
})
.collect();
ensure(
pending.len() <= 1,
crate::i18n::text("jsm_coverage_conflict"),
)?;
if let Some((row, content)) = pending.first() {
ensure(
items
.iter()
.filter(|candidate| equivalent(candidate, &content["payload"]))
.count()
== 1,
crate::i18n::text("jsm_coverage_conflict"),
)?;
db.checkpoint(workspace, &row.object_key, alias, content)
.await?;
} else if overwrite {
self.request(
Method::DELETE,
&format!("/schedules/{remote}/overrides/{alias}"),
None,
)
.await?;
} else {
return Err(Error::Conflict(
crate::i18n::text("jsm_coverage_conflict").into(),
));
}
}
Ok(())
}
async fn publish_coverage(
&self,
db: &Store,
workspace: Uuid,
schedule: Uuid,
remote: &str,
desired: &[Value],
) -> Result<()> {
let prefix = format!("coverage:{schedule}:");
let rows = db.objects(workspace).await?;
let items = self.list(&format!("/schedules/{remote}/overrides")).await?;
let mut used = BTreeSet::new();
for payload in desired {
let digest = Sha256::digest(serde_json::to_vec(payload)?);
let key = format!("{prefix}{digest:x}");
used.insert(key.clone());
let content = json!({"scheduleId":remote,"payload":payload});
let previous = rows.iter().find(|r| r.object_key == key);
let alias = previous
.filter(|r| !r.remote_id.is_empty())
.filter(|r| {
serde_json::from_str::<Value>(&r.content)
.ok()
.is_some_and(|v| v["scheduleId"].as_str() == Some(remote))
})
.map(|r| r.remote_id.as_str());
let existing = alias.and_then(|alias| {
items
.iter()
.find(|item| item["alias"].as_str() == Some(alias))
});
let alias = if let (Some(alias), Some(existing)) = (alias, existing) {
if !equivalent(existing, payload) {
self.request(
Method::PUT,
&format!("/schedules/{remote}/overrides/{alias}"),
Some(payload),
)
.await?;
}
alias.to_owned()
} else {
db.checkpoint(workspace, &key, "", &content).await?;
let created = self
.request(
Method::POST,
&format!("/schedules/{remote}/overrides"),
Some(payload),
)
.await?
.ok_or_else(|| {
Error::Provider(crate::i18n::text("jsm_coverage_response_invalid").into())
})?;
coverage_alias(&created)?.to_owned()
};
let actual = self
.request(
Method::GET,
&format!("/schedules/{remote}/overrides/{alias}"),
None,
)
.await?
.ok_or_else(|| {
Error::Provider(crate::i18n::text("jsm_coverage_response_invalid").into())
})?;
ensure(
equivalent(&actual, payload),
crate::i18n::text("jsm_coverage_response_invalid"),
)?;
db.checkpoint(workspace, &key, &alias, &content).await?;
}
// Keep previous effective spans until all replacement writes are verified.
for row in rows
.iter()
.filter(|r| r.object_key.starts_with(&prefix) && !used.contains(&r.object_key))
{
let content: Value = serde_json::from_str(&row.content)?;
if !row.remote_id.is_empty() {
let parent = content["scheduleId"]
.as_str()
.filter(|v| safe_id(v))
.ok_or_else(|| {
Error::Provider(crate::i18n::text("jsm_coverage_response_invalid").into())
})?;
ensure(
safe_id(&row.remote_id),
crate::i18n::text("jsm_coverage_response_invalid"),
)?;
self.request(
Method::DELETE,
&format!("/schedules/{parent}/overrides/{}", row.remote_id),
None,
)
.await?;
}
db.forget(workspace, &row.object_key).await?;
}
Ok(())
}
pub async fn publish(
&self,
db: &Store,
record: &WorkspaceRecord,
plan: &Publication,
overwrite: bool,
) -> Result<()> {
for schedule in &plan.schedules {
validate_precise_intervals(schedule)?;
}
// Preflight every recipient before creating or modifying any external object.
self.resolve_people(record, plan)
.await?
.publish_resolved(db, record, plan, overwrite)
.await
}
async fn publish_resolved(
&self,
db: &Store,
record: &WorkspaceRecord,
plan: &Publication,
overwrite: bool,
) -> Result<()> {
let objects = db.objects(record.id).await?;
let previous: BTreeMap<_, _> = objects.iter().map(|p| (p.object_key.as_str(), p)).collect();
let mut used = BTreeSet::new();
let mut schedules = BTreeMap::new();
let mut policies = BTreeMap::new();
for s in &plan.schedules {
let key = format!("schedule:{}", s.id);
used.insert(key.clone());
let (rotations, coverage) = self.schedule_payloads(s, plan.as_of)?;
let body = json!({"name":owned_name(record.id,s.id,&s.name),"description":crate::i18n::message("err_managed_by_meguri_p0_e8ef37", &[("p0", (s.name).to_string())]),"teamId":self.team,"timezone":s.timezone,"enabled":true,"rotations":rotations});
let remote = self
.synchronize(
db,
record,
"schedule",
&key,
&body,
previous.get(key.as_str()).copied(),
overwrite,
plan.as_of,
)
.await?;
self.publish_coverage(db, record.id, s.id, &remote, &coverage)
.await?;
schedules.insert(s.id, remote);
}
for e in &plan.escalations {
let key = format!("escalation:{}", e.id);
used.insert(key.clone());
let mut rules = vec![];
for step in &e.steps {
let (kind, id, notify) = match step.recipient {
Recipient::Schedule { id } => (
"schedule",
schedules.get(&id).cloned().ok_or_else(|| {
Error::Validation(
crate::i18n::text(
"err_escalation_references_an_empty_schedule_2a0f3c",
)
.into(),
)
})?,
"default",
),
Recipient::User { id } => ("user", self.mapped(id)?, "default"),
Recipient::TeamAll => ("team", self.team.clone(), "all"),
Recipient::TeamRandom => ("team", self.team.clone(), "random"),
};
rules.push(json!({"condition":step.condition,"notifyType":notify,"delay":step.delay_minutes,"recipient":{"type":kind,"id":id}}));
}
let body = json!({"name":owned_name(record.id,e.id,&e.name),"description":crate::i18n::message("err_managed_by_meguri_p0_e8ef37", &[("p0", (e.name).to_string())]),"enabled":true,"rules":rules,"repeat":{"count":e.repeat.count.max(1),"waitInterval":if e.repeat.count==0 {0}else{e.repeat.wait_minutes},"resetRecipientStates":e.repeat.reset_recipient_states,"closeAlertAfterAll":e.repeat.close_alert_after_all}});
let remote = self
.synchronize(
db,
record,
"escalation",
&key,
&body,
previous.get(key.as_str()).copied(),
overwrite,
plan.as_of,
)
.await?;
policies.insert(e.id, remote);
}
let mut routes = plan.routes.clone();
routes.sort_by_key(|r| r.order);
for (order, route) in routes.iter().enumerate() {
let key = format!("route:{}", route.id);
used.insert(key.clone());
let notify = match route.escalation_id {
Some(id) => {
json!({"type":"escalation","id":policies.get(&id).ok_or_else(||Error::Validation(crate::i18n::text("err_missing_escalation_mapping_3fd412").into()))?})
}
None => json!({"type":"none"}),
};
let conditions:Vec<_>=route.conditions.iter().map(|c|json!({"field":c.field,"operation":c.operation,"expectedValue":c.value,"not":c.negate})).collect();
let criteria = if conditions.is_empty() {
json!({"type":"match-all"})
} else {
json!({"type":"match-all-conditions","conditions":conditions})
};
let mut body = json!({"name":owned_name(record.id,route.id,&route.name),"order":order,"timezone":record.document.timezone,"criteria":criteria,"notify":notify});
if let Some(w) = &route.window {
body["timezone"] = json!(w.timezone);
let days = [
"monday",
"tuesday",
"wednesday",
"thursday",
"friday",
"saturday",
"sunday",
];
let restrictions:Vec<_>=w.weekdays.iter().map(|d|json!({"startDay":days[usize::from(*d)-1],"endDay":days[usize::from(*d)-1],"startHour":w.start.hour(),"startMin":w.start.minute(),"endHour":w.end.hour(),"endMin":w.end.minute()})).collect();
body["timeRestriction"] =
json!({"type":"weekday-and-time-of-day","restrictions":restrictions});
} else {
body["timeRestriction"] = Value::Null;
}
self.synchronize(
db,
record,
"route",
&key,
&body,
previous.get(key.as_str()).copied(),
overwrite,
plan.as_of,
)
.await?;
}
// Delete only checkpointed, owned resources, in reverse dependency order.
for kind in ["route", "escalation", "schedule"] {
for old in objects.iter().filter(|o| {
o.object_key.starts_with(&format!("{kind}:")) && !used.contains(&o.object_key)
}) {
if let Some(current) = self.read(kind, &old.remote_id).await? {
let expected = serde_json::from_str(&old.content)?;
if !equivalent(&current, &expected) && !overwrite {
return Err(Error::Conflict(
crate::i18n::text(
"err_an_obsolete_jsm_resource_changed_externally_dele_a26bda",
)
.into(),
));
}
if kind == "schedule" {
let local_id = old
.object_key
.split_once(':')
.and_then(|(_, id)| Uuid::parse_str(id).ok())
.unwrap();
self.check_coverage(
db,
record.id,
local_id,
&old.remote_id,
overwrite,
plan.as_of,
)
.await?;
}
self.request(
Method::DELETE,
&format!("{}/{}", self.collection(kind), old.remote_id),
None,
)
.await?;
}
db.forget(record.id, &old.object_key).await?;
db.forget(record.id, &format!("pending:{}", old.object_key))
.await?;
if kind == "schedule" {
let prefix = format!(
"coverage:{}:",
old.object_key.strip_prefix("schedule:").unwrap()
);
for child in db
.objects(record.id)
.await?
.iter()
.filter(|o| o.object_key.starts_with(&prefix))
{
db.forget(record.id, &child.object_key).await?;
}
}
}
}
Ok(())
}
}
fn milliseconds(value: DateTime<Utc>) -> DateTime<Utc> {
DateTime::from_timestamp_millis(value.timestamp_millis()).unwrap()
}
fn half_hour_floor(value: DateTime<Utc>, zone: chrono_tz::Tz) -> DateTime<Utc> {
let local = value.with_timezone(&zone);
value
- chrono::Duration::minutes(i64::from(local.minute() % 30))
- chrono::Duration::seconds(i64::from(local.second()))
- chrono::Duration::nanoseconds(i64::from(local.nanosecond()))
}
fn needs_precise_coverage(schedule: &super::PublishedSchedule) -> Result<bool> {
let zone = schedule
.timezone
.parse::<chrono_tz::Tz>()
.map_err(|_| Error::Provider(crate::i18n::text("jsm_invalid_coverage_interval").into()))?;
Ok(schedule.periods.iter().any(|p| {
[p.start, p.end].iter().any(|date| {
let local = date.with_timezone(&zone);
local.minute() % 30 != 0 || local.second() != 0 || local.nanosecond() != 0
})
}))
}
fn validate_precise_intervals(schedule: &super::PublishedSchedule) -> Result<()> {
ensure(
!schedule.periods.is_empty()
&& schedule
.periods
.iter()
.all(|p| milliseconds(p.start) < milliseconds(p.end)),
crate::i18n::text("jsm_invalid_coverage_interval"),
)
}
fn coverage_alias(value: &Value) -> Result<&str> {
value["alias"]
.as_str()
.filter(|v| safe_id(v))
.ok_or_else(|| Error::Provider(crate::i18n::text("jsm_coverage_response_invalid").into()))
}
fn ownership_marker(workspace: Uuid, id: Uuid) -> String {
let mut digest = Sha256::new();
digest.update(workspace.as_bytes());
digest.update(id.as_bytes());
let digest = digest.finalize();
format!(
"[meguri:{:032x}]",
u128::from_be_bytes(digest[..16].try_into().unwrap())
)
}
fn owned_name(workspace: Uuid, id: Uuid, name: &str) -> String {
format!("{name} {}", ownership_marker(workspace, id))
}
fn safe_id(id: &str) -> bool {
!id.is_empty()
&& id.len() <= 200
&& id
.chars()
.all(|c| c.is_ascii_alphanumeric() || "-_".contains(c))
}
/// Ignore server-generated metadata, while comparing every managed field and array entry.
pub fn equivalent(actual: &Value, expected: &Value) -> bool {
match expected {
Value::Object(expected) => {
// A restriction added to a finite rotation changes coverage, not metadata.
if expected.contains_key("participants")
&& expected.contains_key("startDate")
&& !expected.contains_key("timeRestriction")
&& actual.get("timeRestriction").is_some_and(|v| !v.is_null())
{
return false;
}
if expected.contains_key("responder")
&& expected.contains_key("startDate")
&& !expected.contains_key("rotationIds")
&& actual.get("rotationIds").is_some_and(|ids| {
!ids.is_null() && ids.as_array().is_none_or(|ids| !ids.is_empty())
})
{
return false;
}
expected.iter().all(|(k, v)| {
if k == "repeat"
&& expected.contains_key("rules")
&& actual.get(k).is_none()
&& v.get("waitInterval").and_then(Value::as_u64) == Some(0)
{
// JSM omits disabled repeats in escalation GET responses.
true
} else if k == "rotations" {
let Some(actual) = actual.get(k).and_then(Value::as_array) else {
return false;
};
let Some(expected) = v.as_array() else {
return false;
};
let mut remaining: Vec<_> = actual.iter().collect();
for expected in expected {
let Some(index) = remaining
.iter()
.position(|actual| equivalent(actual, expected))
else {
return false;
};
remaining.remove(index);
}
remaining.is_empty()
} else {
equivalent(actual.get(k).unwrap_or(&Value::Null), v)
}
})
}
Value::Array(expected) => actual.as_array().is_some_and(|a| {
a.len() == expected.len() && a.iter().zip(expected).all(|(a, e)| equivalent(a, e))
}),
Value::String(expected) => {
if let (Ok(a), Ok(b)) = (
actual.as_str().unwrap_or_default().parse::<DateTime<Utc>>(),
expected.parse::<DateTime<Utc>>(),
) {
a == b
} else {
actual.as_str() == Some(expected)
}
}
_ => actual == expected,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn omitted_escalation_repeat_only_matches_disabled_repeats() {
let mut expected = json!({"rules":[{"delay":0}],"repeat":{"count":1,"waitInterval":0,"resetRecipientStates":false,"closeAlertAfterAll":false}});
let mut remote = json!({"rules":[{"delay":0}]});
assert!(equivalent(&remote, &expected));
remote["rules"][0]["delay"] = json!(10);
assert!(!equivalent(&remote, &expected));
remote["rules"][0]["delay"] = json!(0);
expected["repeat"]["waitInterval"] = json!(5);
assert!(!equivalent(&remote, &expected));
remote["repeat"] = expected["repeat"].clone();
assert!(equivalent(&remote, &expected));
for (field, changed) in [
("count", json!(2)),
("waitInterval", json!(10)),
("resetRecipientStates", json!(true)),
("closeAlertAfterAll", json!(true)),
] {
let mut changed_remote = remote.clone();
changed_remote["repeat"][field] = changed;
assert!(!equivalent(&changed_remote, &expected));
}
expected["repeat"]["waitInterval"] = json!(0);
assert!(!equivalent(&remote, &expected));
remote["repeat"] = Value::Null;
assert!(!equivalent(&remote, &expected));
}
#[test]
fn normalization_ignores_ids_but_detects_extra_rotations_and_changed_owners() {
let expected = json!({"rotations":[{"startDate":"2026-09-23T08:00:00Z","participants":[{"id":"alice","type":"user"}]}]});
let mut remote = json!({"id":"remote","rotations":[{"id":"rotation","startDate":"2026-09-23T08:00:00.000Z","participants":[{"id":"alice","type":"user"}]}]});
assert!(equivalent(&remote, &expected));
remote["rotations"][0]["timeRestriction"] =
json!({"type":"time-of-day","restriction":{"startHour":9,"endHour":17}});
assert!(!equivalent(&remote, &expected));
remote["rotations"][0]
.as_object_mut()
.unwrap()
.remove("timeRestriction");
remote["rotations"][0]["participants"][0]["id"] = json!("bob");
assert!(!equivalent(&remote, &expected));
remote["rotations"].as_array_mut().unwrap().clear();
assert!(!equivalent(&remote, &expected));
}
}
Reference in new issue View Git Blame Copy Permalink