Files
meguri/src/providers/jsm.rs
T

690 lines
26 KiB
Rust
Raw Normal View History

2026-09-24 14:35:05 +03:00
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 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))
.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>> {
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=100"),
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 < 100
{
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,
) -> Result<String> {
let name = desired["name"].as_str().unwrap_or_default();
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() == Some(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()?;
if previous.is_some() && current.is_none() && !overwrite {
return Err(Error::Conflict(crate::i18n::message(
"err_jsm_kind_was_deleted_externally_publication_paus_88543d",
&[("kind", (kind).to_string())],
)));
}
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));
if !matches_desired && !known && !overwrite {
return Err(Error::Conflict(crate::i18n::message(
"err_jsm_kind_changed_outside_meguri_publication_paus_ac8220",
&[("kind", (kind).to_string())],
)));
}
}
if kind == "schedule"
&& let Some(id) = &existing_id
{
let overrides = self.list(&format!("/schedules/{id}/overrides")).await?;
if !overrides.is_empty() && !overwrite {
return Err(Error::Conflict(
crate::i18n::text(
"err_jsm_schedule_contains_external_overrides_publica_ff6a5d",
)
.into(),
));
}
for item in overrides {
let alias = item["alias"].as_str().ok_or_else(|| {
Error::Provider(crate::i18n::text("err_override_has_no_alias_0bab5d").into())
})?;
ensure(
safe_id(alias),
crate::i18n::text("err_unexpected_jsm_override_identifier_896f12"),
)?;
self.request(
Method::DELETE,
&format!("/schedules/{id}/overrides/{alias}"),
None,
)
.await?;
}
}
let id = if matches_desired {
existing_id.unwrap()
} else if current.is_some() {
let id = existing_id.unwrap();
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?;
Ok(id)
}
pub async fn publish(
&self,
db: &Store,
record: &WorkspaceRecord,
plan: &Publication,
overwrite: bool,
) -> Result<()> {
// 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=s.periods.iter().map(|p|Ok(json!({"name":format!("shift-{}",p.start.timestamp()),"startDate":p.start,"endDate":p.end,"type":"daily","length":1,"participants":[{"type":"user","id":self.mapped(p.user_id)?}]}))).collect::<Result<Vec<_>>>()?;
let body = json!({"name":owned_name(record.id,s.id),"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,
)
.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),"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,
)
.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),"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,
)
.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"
&& !overwrite
&& !self
.list(&format!("/schedules/{}/overrides", old.remote_id))
.await?
.is_empty()
{
return Err(Error::Conflict(
crate::i18n::text(
"err_an_obsolete_jsm_schedule_has_external_overrides_c6de63",
)
.into(),
));
}
self.request(
Method::DELETE,
&format!("{}/{}", self.collection(kind), old.remote_id),
None,
)
.await?;
}
db.forget(record.id, &old.object_key).await?;
}
}
Ok(())
}
}
fn owned_name(team: Uuid, id: Uuid) -> String {
format!("meguri-{}-{}", team.simple(), id.simple())
}
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;
}
expected
.iter()
.all(|(k, v)| 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 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));
}
}