Merge NEXT: planning, generation map and portable configuration on top of the OIDC access model

NEXT predates the access work, so the identity layer follows master: access comes from
configured groups, instance administrators replace the site administration, and accounts are
created by SSO or by an administrator. Kept from NEXT: manual publication with the Review
changes panel and shared generator drafts, portable import/export (meguri/v2), the layered
generation map with binding provenance, JSM provider changes, contact/approval semantics and
the account-owned Person. Dropped: site administration, local sign-up, and the mandatory
unique email with its contact/account migrations. The manual-publication migration is
renumbered m_0010 after m_0009_access.

Accounts without an email still get a Person in every workspace they reach, a changed sign-in
email updates the account's person, and an account that loses access keeps its linked person so
schedules stay valid. Instance administrators may use Sync and shared drafts.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Aleksandr BogomiakovandClaude committed 2026-10-07 16:06:10 +01:00
commit 38105c65c2
57 files changed
+6890 -431

No files matched your search

+6
View File
@@ -185,6 +185,9 @@ pub async fn create_password_user(
}
let id = store::uuid(&row.id)?;
apply_access(db, config, id, Some(groups.clone())).await?;
if let Err(error) = crate::identity::link_contacts(db, id).await {
tracing::warn!(user = %id, %error, "could not add the new account to its workspaces");
}
db.instance_event(
actor,
"user.created",
@@ -239,6 +242,9 @@ pub async fn set_user_groups(
let before = editable_target(db, actor, target).await?;
let groups = checked_groups(config, groups)?;
apply_access(db, config, target, Some(groups.clone())).await?;
if let Err(error) = crate::identity::link_contacts(db, target).await {
tracing::warn!(user = %target, %error, "could not add the account to its workspaces");
}
db.instance_event(
Some(actor),
"user.groups_changed",
+2 -1
View File
@@ -334,7 +334,7 @@ pub async fn seed(db: &Store) -> Result<()> {
username: username.into(),
display_name: display.into(),
password: notice.password.clone(),
email: None,
email: Some(format!("{username}@example.com")),
groups: groups.iter().map(|g| (*g).to_owned()).collect(),
country: Some(country.into()),
},
@@ -470,6 +470,7 @@ async fn seed_workspace(
_ => service_tiers(monday, alice, carol),
};
let mut draft = Draft {
shared_revision: None,
revision: workspace.revision,
variables,
rules,
+2 -2
View File
@@ -200,7 +200,7 @@ pub async fn reset(db: &Database) -> Result<()> {
LOCK TABLE public.meguri_runtime_guard IN EXCLUSIVE MODE;
IF NOT EXISTS(SELECT 1 FROM public.meguri_runtime_guard WHERE id AND signature='meguri.runtime.v1' AND last_mode='demo') THEN RAISE EXCEPTION 'MEGURI_DEMO_PROTECTED'; END IF;
IF EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND backend_type='client backend' AND pid<>pg_backend_pid() AND application_name<>current_setting('application_name')) THEN RAISE EXCEPTION 'Demo database has another active client; stop the other process before restarting'; END IF;
SELECT string_agg(format('public.%I',name),',') INTO tables FROM unnest(ARRAY['meguri__workspaces','meguri__workspace_members','meguri__workspace_history','meguri__sessions','meguri__external_identities','meguri__team_members','meguri__tenant_members','meguri__team_history','meguri__audit_events','meguri__sync_jobs','meguri__provider_objects','meguri__teams','meguri__tenants','meguri__users','meguri__demo_pins','meguri__instance_audit','meguri__instance_state','meguri__oidc_flows']) name WHERE to_regclass('public.'||name) IS NOT NULL;
SELECT string_agg(format('public.%I',name),',') INTO tables FROM unnest(ARRAY['meguri__workspace_drafts','meguri__workspaces','meguri__workspace_members','meguri__workspace_history','meguri__sessions','meguri__external_identities','meguri__team_members','meguri__tenant_members','meguri__team_history','meguri__audit_events','meguri__sync_jobs','meguri__provider_objects','meguri__teams','meguri__tenants','meguri__users','meguri__demo_pins','meguri__instance_audit','meguri__instance_state','meguri__oidc_flows']) name WHERE to_regclass('public.'||name) IS NOT NULL;
IF tables IS NOT NULL THEN EXECUTE 'TRUNCATE '||tables||' RESTART IDENTITY'; END IF;
UPDATE public.meguri_runtime_guard SET started_at=clock_timestamp() WHERE id;
END $$;").await?;
@@ -248,7 +248,7 @@ requests AS (
UPDATE meguri__workspaces SET document=jsonb_set(document::jsonb,'{requests}',COALESCE((SELECT jsonb_agg(r) FROM jsonb_array_elements(document::jsonb->'requests') r WHERE (r->>'created_at')::timestamptz>$1-make_interval(secs=>$4::bigint::double precision)),'[]'::jsonb))::text,revision=revision+1
WHERE EXISTS(SELECT 1 FROM guard) AND demo_deleted_at IS NULL AND id NOT IN(SELECT id FROM expired) AND EXISTS(SELECT 1 FROM jsonb_array_elements(document::jsonb->'requests') r WHERE (r->>'created_at')::timestamptz<=$1-make_interval(secs=>$4::bigint::double precision)) RETURNING *
), history AS (INSERT INTO meguri__workspace_history(id,workspace_id,revision,document,effective_at) SELECT id||':'||revision::text,id,revision,document,$1 FROM requests),
request_jobs AS (INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,revision,'pending',0,$1 FROM requests ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision,state='pending',next_attempt=$1)
request_jobs AS (INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,revision,'draft',0,$1 FROM requests ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision)
INSERT INTO meguri__audit_events(workspace_id,action,details,created_at) SELECT id,'demo.requests_expired','{}',$1 FROM requests;
"#;
const PURGE: &str = r#"
+22 -9
View File
@@ -45,20 +45,25 @@ impl Permission {
pub struct User {
#[serde(default)]
pub language: Option<String>,
pub id: Uuid,
/// Canonical account-owned person; email is required by account persistence.
#[serde(flatten)]
pub person: Person,
pub username: String,
pub display_name: String,
pub country: String,
/// None follows the browser timezone; this never changes scheduling rules.
pub timezone: Option<String>,
#[serde(default)]
pub email: Option<String>,
#[serde(default)]
pub instance_admin: bool,
#[serde(default)]
pub has_password: bool,
}
impl std::ops::Deref for User {
type Target = Person;
fn deref(&self) -> &Person {
&self.person
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Member {
pub user: User,
@@ -191,9 +196,7 @@ impl WorkspaceDocument {
)?;
validate_name(&person.display_name)?;
ensure(
person
.user_id
.is_none_or(|id| id == person.id && members.contains(&id)),
person.user_id.is_none_or(|id| id == person.id),
crate::i18n::text("err_choose_a_user_who_has_access_to_this_workspace_41cf15"),
)?;
ensure(
@@ -450,7 +453,8 @@ impl WorkspaceDocument {
} = &self.integration
{
ensure(
Uuid::parse_str(cloud_id).is_ok() && Uuid::parse_str(team_id).is_ok(),
Uuid::parse_str(cloud_id).is_ok()
&& Uuid::parse_str(team_id.strip_prefix("og-").unwrap_or(team_id)).is_ok(),
crate::i18n::text("err_jsm_cloud_and_team_ids_must_be_uuids_f75b5f"),
)?;
ensure(
@@ -726,3 +730,12 @@ pub struct Timeline {
pub to: DateTime<Utc>,
pub assignments: Vec<Assignment>,
}
/// A compact representation of a verified calendar-day cycle, independent of providers.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RecurringRotation {
pub start: DateTime<Utc>,
pub end: DateTime<Utc>,
pub shift_days: u16,
pub participants: Vec<Uuid>,
}
+113
View File
@@ -170,6 +170,63 @@ pub fn timeline(
})
}
/// Compress only coverage identical to the base cycle. History or approved
/// replacements that change any effective span must retain their exact timeline.
pub fn recurring_rotation(
doc: &WorkspaceDocument,
schedule: &Schedule,
periods: &[Assignment],
) -> Result<Option<RecurringRotation>> {
let ScheduleKind::Rotation {
participants,
anchor,
shift_days,
} = &schedule.kind
else {
return Ok(None);
};
let (Some(first), Some(last)) = (periods.first(), periods.last()) else {
return Ok(None);
};
let mut base = doc.clone();
base.schedules = vec![schedule.clone()];
base.requests.clear();
let expected = timeline(&base, first.start, last.end)?;
// Compare coverage at every interval boundary; reason labels and redundant
// adjacent spans do not change who is on call.
if periods.windows(2).any(|p| p[0].end != p[1].start)
|| periods.iter().any(|p| p.start >= p.end || p.unavailable)
{
return Ok(None);
}
for p in &expected.assignments {
if periods
.iter()
.filter(|a| a.start < p.end && p.start < a.end)
.any(|a| a.user_id != p.user_id || a.unavailable != p.unavailable)
{
return Ok(None);
}
}
let day = schedule_duty_day(doc, schedule.id, first.start)?;
let shift = (day - *anchor)
.num_days()
.div_euclid(i64::from(*shift_days));
let start = local_time(
doc,
*anchor + Duration::days(shift * i64::from(*shift_days)),
schedule.handover,
)?;
let mut ordered = participants.clone();
ordered.rotate_left(shift.rem_euclid(participants.len() as i64) as usize);
Ok(Some(RecurringRotation {
start,
end: last.end,
shift_days: *shift_days,
participants: ordered,
}))
}
fn resolve(
doc: &WorkspaceDocument,
id: Uuid,
@@ -834,4 +891,60 @@ mod tests {
assert!(restored.effective_legs().is_empty());
assert_eq!(restored.legs.len(), 2);
}
#[test]
fn compact_cycle_preserves_phase_calendar_days_and_rejects_changed_coverage() {
let (mut doc, people, id, _) = fixture();
doc.timezone = "Europe/Bucharest".into();
doc.schedules.truncate(1);
for (start, end, expected_start, first_person) in [
(
"2026-03-10T12:00:00Z",
"2026-04-20T08:00:00Z",
"2026-03-09T09:00:00Z",
people[2],
),
(
"2026-10-20T12:00:00Z",
"2026-11-23T09:00:00Z",
"2026-10-19T08:00:00Z",
people[1],
),
] {
let from = start.parse().unwrap();
let to = end.parse().unwrap();
let periods = timeline(&doc, from, to).unwrap().assignments;
let compact = recurring_rotation(&doc, &doc.schedules[0], &periods)
.unwrap()
.unwrap();
assert_eq!(
compact.start,
expected_start.parse::<DateTime<Utc>>().unwrap()
);
assert_eq!(compact.participants[0], first_person);
let zone: chrono_tz::Tz = doc.timezone.parse().unwrap();
for p in &periods {
let days = (p.start.with_timezone(&zone).date_naive()
- compact.start.with_timezone(&zone).date_naive())
.num_days();
assert_eq!(p.user_id, compact.participants[(days / 7) as usize % 3]);
}
assert!(
periods
.windows(2)
.any(|p| (p[0].end - p[0].start).num_hours() != 168)
);
let mut changed = periods.clone();
changed[1].user_id = people
.iter()
.copied()
.find(|p| *p != changed[1].user_id)
.unwrap();
assert!(
recurring_rotation(&doc, &doc.schedules[0], &changed)
.unwrap()
.is_none()
);
assert_eq!(periods[0].schedule_id, id);
}
}
}
+71 -6
View File
@@ -84,6 +84,10 @@ pub struct Step {
#[serde(default)]
pub user: bool,
#[serde(default)]
pub team_all: bool,
#[serde(default)]
pub team_random: bool,
#[serde(default)]
pub condition: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -97,6 +101,8 @@ pub struct Condition {
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(from = "BlockWire")]
pub struct Block {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub portable_key: Option<String>,
pub id: Uuid,
pub kind: String,
pub fields: BTreeMap<String, Binding>,
@@ -107,6 +113,8 @@ pub struct Block {
}
#[derive(Deserialize)]
struct BlockWire {
#[serde(default)]
pub portable_key: Option<String>,
pub id: Uuid,
pub kind: String,
pub fields: BTreeMap<String, Binding>,
@@ -133,6 +141,7 @@ impl From<BlockWire> for Block {
);
}
Self {
portable_key: value.portable_key,
id: value.id,
kind: value.kind,
fields: value.fields,
@@ -506,6 +515,38 @@ impl Resolver<'_> {
}
Ok(node)
}
fn provenance(&self, b: &Binding, iterations: &[Iteration], depth: usize) -> Value {
if depth > 16 {
return Value::Null;
}
let mut source = b.clone();
source.path.clear();
source.prefix.clear();
source.suffix.clear();
let snapshot = |binding: &Binding| match self.eval(binding, 0) {
Ok(node) => json!({"value":node.json(),"node_id":node.id}),
Err(error) => json!({"error":error.to_string()}),
};
let mut stages = vec![json!({"path":[],"result":snapshot(&source)})];
for segment in &b.path {
source.path.push(segment.clone());
stages.push(json!({"path":source.path,"result":snapshot(&source)}));
}
let mut parents = vec![];
if matches!(b.kind.as_str(), "item" | "key")
&& let Some(iteration) = iterations.iter().find(|i| Some(i.id) == b.iteration)
{
parents.push(self.provenance(&iteration.source, iterations, depth + 1));
}
if let Some(key) = &b.key {
parents.push(self.provenance(key, iterations, depth + 1));
}
for item in &b.items {
parents.push(self.provenance(item, iterations, depth + 1));
}
json!({"binding":b,"stages":stages,"result":snapshot(b),"parents":parents})
}
fn value(&self, b: &Binding) -> Result<Value> {
Ok(self.eval(b, 0)?.json())
}
@@ -730,7 +771,14 @@ fn render(
&[("p0", (i + 1).to_string())],
))
})?;
let target = resolver.value(&step.recipient)?;
ensure(
[step.user, step.team_all, step.team_random]
.into_iter()
.filter(|v| *v)
.count()
<= 1,
crate::i18n::text("generation_recipient_conflict"),
)?;
steps.push(EscalationStep {
delay_minutes: delay as u16,
condition: if step.condition == "if-not-closed" {
@@ -738,13 +786,17 @@ fn render(
} else {
EscalationCondition::IfNotAcked
},
recipient: if step.user {
recipient: if step.team_all {
Recipient::TeamAll
} else if step.team_random {
Recipient::TeamRandom
} else if step.user {
Recipient::User {
id: recipient(&target, workspace, doc, members)?,
id: recipient(&resolver.value(&step.recipient)?, workspace, doc, members)?,
}
} else {
Recipient::Schedule {
id: id_value(target)?,
id: id_value(resolver.value(&step.recipient)?)?,
}
},
});
@@ -1014,7 +1066,20 @@ pub fn expand(
} else {
format!("{path} / {}", context.labels.join(" / "))
};
let mut row_trace = json!({"labels":context.labels,"inputs":context.items.iter().map(|(id,n)|(id.to_string(),n.json())).collect::<BTreeMap<_,_>>(),"status":"included","resources":[],"blocks":[]});
let bindings: Vec<_> = rule
.iterations
.iter()
.map(|i| &i.source)
.chain(rule.when.iter().flat_map(|p| [&p.left, &p.right]))
.chain(rule.blocks.iter().flat_map(|b| {
b.fields
.values()
.chain(b.steps.iter().flat_map(|s| [&s.delay, &s.recipient]))
.chain(b.conditions.iter().map(|c| &c.value))
}))
.map(|b| resolver.provenance(b, &rule.iterations, 0))
.collect();
let mut row_trace = json!({"bindings":bindings,"labels":context.labels,"inputs":context.items.iter().map(|(id,n)|(id.to_string(),n.json())).collect::<BTreeMap<_,_>>(),"status":"included","resources":[],"blocks":[]});
let managed_start = managed.len();
let row_issue_start = issues.len();
if let Some(predicate) = &rule.when {
@@ -1051,7 +1116,7 @@ pub fn expand(
row_trace["blocks"].as_array_mut().unwrap().push(json!({
"block":block.id,
"fields":block.fields.iter().map(|(key,b)|(key.clone(),resolved(b))).collect::<BTreeMap<_,_>>(),
"steps":block.steps.iter().map(|s|json!({"delay":resolved(&s.delay),"recipient":resolved(&s.recipient),"condition":s.condition})).collect::<Vec<_>>(),
"steps":block.steps.iter().map(|s|json!({"delay":resolved(&s.delay),"recipient":if s.team_all {json!(crate::i18n::text("everyone_on_the_team_235b55"))} else if s.team_random {json!(crate::i18n::text("a_random_team_member_8f35f3"))} else {resolved(&s.recipient)},"condition":s.condition})).collect::<Vec<_>>(),
"conditions":block.conditions.iter().map(|c|json!({"field":c.field,"operation":c.operation,"value":resolved(&c.value),"negate":c.negate})).collect::<Vec<_>>()
}));
if manual_ids.contains(&id) {
+112 -11
View File
@@ -8,12 +8,14 @@ use crate::{
store::{Store, WorkspaceRecord},
};
use chrono::{Duration, Utc};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use uuid::Uuid;
#[derive(Debug, Clone, Deserialize)]
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct Draft {
pub revision: i64,
#[serde(default)]
pub shared_revision: Option<i64>,
pub variables: Vec<Variable>,
pub rules: Vec<Rule>,
#[serde(default)]
@@ -32,6 +34,7 @@ async fn prepare(db: &Store, actor: Uuid, workspace: Uuid, draft: &Draft) -> Res
.prepare(workspace, draft.revision, actor, Permission::EditSchedules)
.await?;
let directory = db.directory(workspace).await?;
crate::store::account_people(&mut record.document, &directory)?;
let members = directory.iter().map(|u| u.id).collect();
let now = Utc::now();
let mut expanded = generation::expand(
@@ -61,14 +64,7 @@ async fn prepare(db: &Store, actor: Uuid, workspace: Uuid, draft: &Draft) -> Res
.collect();
for user in &directory {
if used.contains(&user.id) && expanded.document.person(user.id).is_none() {
expanded.document.people.push(Person {
id: user.id,
user_id: Some(user.id),
display_name: user.display_name.clone(),
email: None,
country: Some(user.country.clone()),
aliases: vec![],
});
expanded.document.people.push(user.person.clone());
}
}
if expanded.issues.is_empty() {
@@ -123,7 +119,11 @@ async fn prepare(db: &Store, actor: Uuid, workspace: Uuid, draft: &Draft) -> Res
draft.route_position
]))
.to_string();
let report = json!({"review_id":review,"changes":expanded.changes,"cancelled_requests":cancelled,"result":result,"trace":expanded.trace,"issues":issues});
let publication_changes = crate::store::publication_changes(
&expanded.document,
db.published_document(workspace).await?.as_ref(),
)?;
let report = json!({"shared_revision":draft.shared_revision,"publication_changes":publication_changes,"review_id":review,"changes":expanded.changes,"cancelled_requests":cancelled,"result":result,"trace":expanded.trace,"issues":issues});
record.document = expanded.document;
Ok(Prepared {
report,
@@ -162,3 +162,104 @@ pub async fn apply(db: &Store, actor: Uuid, workspace: Uuid, draft: Draft) -> Re
.await?;
Ok(json!({"ok":true}))
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Transfer {
pub revision: i64,
pub shared_revision: i64,
#[serde(default)]
pub text: String,
#[serde(default)]
pub review_id: Option<String>,
}
async fn transfer_base(
db: &Store,
actor: Uuid,
workspace: Uuid,
request: &Transfer,
) -> Result<(WorkspaceRecord, Draft)> {
let record = db
.prepare(
workspace,
request.revision,
actor,
Permission::EditSchedules,
)
.await?;
let shared = db.shared_draft(workspace).await?;
ensure(
shared["revision"].as_i64() == Some(request.shared_revision),
crate::i18n::text("err_shared_draft_stale"),
)?;
let mut draft: Draft = if shared["payload"].is_null() {
Draft {
revision: record.revision,
shared_revision: Some(request.shared_revision),
variables: record.document.generation.variables.clone(),
rules: record.document.generation.rules.clone(),
route_position: record.document.generation.route_position,
review_id: None,
}
} else {
serde_json::from_value(shared["payload"].clone())?
};
draft.revision = record.revision;
draft.shared_revision = Some(request.shared_revision);
Ok((record, draft))
}
pub async fn export_config(
db: &Store,
actor: Uuid,
workspace: Uuid,
request: Transfer,
) -> Result<Value> {
let (record, draft) = transfer_base(db, actor, workspace, &request).await?;
Ok(
json!({"text":crate::portable::format_config(&crate::portable::export(&record.document,&draft)?)}),
)
}
pub async fn import_config(
db: &Store,
actor: Uuid,
workspace: Uuid,
request: Transfer,
apply: bool,
) -> Result<Value> {
let (record, base) = transfer_base(db, actor, workspace, &request).await?;
let (draft, inputs) =
crate::portable::import(workspace, &record.document, &base, &request.text)?;
let prepared = prepare(db, actor, workspace, &draft).await?;
let review = generation::stable_id(&json!([
prepared.review,
request.shared_revision,
request.text
]))
.to_string();
if apply {
ensure(
prepared.issues.is_empty(),
crate::i18n::text("transfer_fix_errors"),
)?;
ensure(
request.review_id.as_deref() == Some(review.as_str()),
crate::i18n::text("err_the_preview_is_stale_check_the_updated_impact_be_85cacd"),
)?;
return db
.save_shared_draft(
workspace,
actor,
request.shared_revision,
serde_json::to_value(draft)?,
)
.await;
}
let contacts: Vec<_> = prepared
.record
.document
.people
.iter()
.filter(|p| record.document.person(p.id).is_none() && p.user_id.is_none())
.collect();
Ok(json!({"review_id":review,"inputs":inputs,"contacts":contacts,"preview":prepared.report}))
}
+48 -33
View File
@@ -72,7 +72,7 @@ WITH workspace AS (
), history AS (
INSERT INTO meguri__workspace_history(id,workspace_id,revision,document,effective_at,actor_id) SELECT id||':1',id,1,document,clock_timestamp(),NULL FROM workspace
), jobs AS (
INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,1,'pending',0,now() FROM workspace
INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,1,'draft',0,now() FROM workspace
)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at) SELECT id,NULL,'workspace.created',$4,now() FROM workspace
"#;
@@ -81,8 +81,8 @@ const ENABLE_WORKSPACE: &str = r#"
WITH enabled AS (
UPDATE meguri__workspaces SET disabled_at=NULL WHERE id=$1 AND disabled_at IS NOT NULL AND demo_deleted_at IS NULL RETURNING id,revision
), jobs AS (
INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,revision,'pending',0,now() FROM enabled
ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision,state='pending',attempts=0,next_attempt=now(),last_error=NULL
INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,revision,'draft',0,now() FROM enabled
ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision
)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at) SELECT id,NULL,'workspace.enabled','{}',now() FROM enabled
"#;
@@ -356,7 +356,11 @@ pub async fn reconcile_all(db: &Store, config: &Config) -> Result<usize> {
if row.demo_deleted_at.is_some() {
continue;
}
apply_access(db, config, store::uuid(&row.id)?, None).await?;
let id = store::uuid(&row.id)?;
apply_access(db, config, id, None).await?;
if let Err(error) = link_contacts(db, id).await {
tracing::warn!(user = %id, %error, "could not refresh the account person in its workspaces");
}
count += 1;
}
Ok(count)
@@ -735,9 +739,7 @@ pub async fn link_contacts(db: &Store, user: Uuid) -> Result<()> {
.get(&db.db)
.await?
.ok_or(Error::NotFound)?;
let Some(email) = row.email.as_deref().and_then(normalize_email) else {
return Ok(());
};
let email = row.email.as_deref().and_then(normalize_email);
for membership in query!(WorkspaceMemberRow,$user_id==id).all(&db.db).await? {
let workspace = store::uuid(&membership.workspace_id)?;
for _ in 0..3 {
@@ -745,8 +747,11 @@ pub async fn link_contacts(db: &Store, user: Uuid) -> Result<()> {
break;
};
let members = db.member_ids(workspace).await?;
let Some((document, details)) = linked_document(&record.document, &row, user, &email)
else {
let changed = match &email {
Some(email) => linked_document(&record.document, &row, user, email),
None => account_person_added(&record.document, &row, user),
};
let Some((document, details)) = changed else {
break;
};
record.document = document;
@@ -767,6 +772,28 @@ pub async fn link_contacts(db: &Store, user: Uuid) -> Result<()> {
Ok(())
}
/// Accounts without an email still own a Person in every workspace they reach, so planning
/// can use them at once; nothing else can be linked without an address.
fn account_person_added(
document: &WorkspaceDocument,
user: &UserRow,
user_id: Uuid,
) -> Option<(WorkspaceDocument, serde_json::Value)> {
if document.people.iter().any(|p| p.id == user_id) {
return None;
}
let mut linked = document.clone();
linked.people.push(Person {
id: user_id,
user_id: Some(user_id),
display_name: user.display_name.clone(),
email: None,
country: Some(user.country.clone()),
aliases: vec![],
});
Some((linked, json!({"user_id":user_id,"merged":false})))
}
/// Pure transformation for [`link_contacts`]. `None` means nothing to change.
pub fn linked_document(
document: &WorkspaceDocument,
@@ -789,13 +816,6 @@ pub fn linked_document(
if contact.user_id.is_some() {
return None;
}
if document
.people
.iter()
.any(|p| p.id == user_id && p.email.as_deref().is_some_and(|e| e != email))
{
return None;
}
let old = contact.id;
let mut working = document.clone();
working.people.retain(|p| p.id != user_id);
@@ -816,10 +836,7 @@ pub fn linked_document(
match document.people.iter().position(|p| p.id == user_id) {
Some(index) => {
let person = &document.people[index];
if person.email.is_some() && person.user_id == Some(user_id) {
return None;
}
if person.email.as_deref().is_some_and(|e| e != email) {
if person.email.as_deref() == Some(email) && person.user_id == Some(user_id) {
return None;
}
let mut linked = document.clone();
@@ -914,19 +931,17 @@ mod tests {
taken.user_id = Some(Uuid::new_v4());
doc.people = vec![taken];
assert!(linked_document(&doc, &row(), user, "dana@corp.test").is_none());
// The user's own person carries a different email: leave both alone.
doc.people = vec![
Person {
id: user,
user_id: Some(user),
display_name: "Dana".into(),
email: Some("other@corp.test".into()),
country: None,
aliases: vec![],
},
contact("dana@corp.test"),
];
assert!(linked_document(&doc, &row(), user, "dana@corp.test").is_none());
// The account owns its person: when the sign-in email changed, the person follows it.
doc.people = vec![Person {
id: user,
user_id: Some(user),
display_name: "Dana".into(),
email: Some("old@corp.test".into()),
country: None,
aliases: vec![],
}];
let (linked, _) = linked_document(&doc, &row(), user, "dana@corp.test").unwrap();
assert_eq!(linked.people[0].email.as_deref(), Some("dana@corp.test"));
}
#[test]
+2
View File
@@ -16,6 +16,8 @@ pub mod identity;
pub mod migrations;
pub mod models;
pub mod oidc;
pub mod portable;
mod portable_v2;
pub mod providers;
pub mod service;
pub mod store;
+2
View File
@@ -7,6 +7,7 @@ pub mod m_0006_demo_lifecycle;
pub mod m_0007_language;
pub mod m_0008_workspaces;
pub mod m_0009_access;
pub mod m_0010_manual_publication;
pub const MIGRATIONS: &[&cot::db::migrations::SyncDynMigration] = &[
&m_0001_initial::Migration,
&m_0002_constraints::Migration,
@@ -17,4 +18,5 @@ pub const MIGRATIONS: &[&cot::db::migrations::SyncDynMigration] = &[
&m_0007_language::Migration,
&m_0008_workspaces::Migration,
&m_0009_access::Migration,
&m_0010_manual_publication::Migration,
];
@@ -0,0 +1,22 @@
use cot::db::migrations::{MigrationContext, MigrationDependency, Operation, migration_op};
#[derive(Debug, Copy, Clone)]
pub struct Migration;
impl cot::db::migrations::Migration for Migration {
const APP_NAME: &'static str = "meguri";
const MIGRATION_NAME: &'static str = "m_0010_manual_publication";
const DEPENDENCIES: &'static [MigrationDependency] =
&[MigrationDependency::migration("meguri", "m_0009_access")];
const OPERATIONS: &'static [Operation] = &[Operation::custom(upgrade).build()];
}
#[migration_op]
async fn upgrade(context: MigrationContext<'_>) -> cot::db::Result<()> {
context.db.raw(r#"
DO $$ BEGIN
ALTER TABLE meguri__sync_jobs ADD COLUMN authorized_revision bigint, ADD COLUMN authorized_document text, ADD COLUMN published_document text;
UPDATE meguri__sync_jobs j SET published_document=h.document FROM meguri__workspace_history h WHERE h.workspace_id=j.workspace_id AND h.revision=j.published_revision;
UPDATE meguri__sync_jobs SET state=CASE WHEN published_revision=requested_revision THEN 'synced' ELSE 'draft' END,lease_token=NULL,lease_until=NULL;
CREATE TABLE meguri__workspace_drafts(workspace_id text PRIMARY KEY REFERENCES meguri__workspaces(id) ON DELETE CASCADE, revision bigint NOT NULL, payload text NOT NULL);
END $$;
"#).await?;
Ok(())
}
+12
View File
@@ -96,6 +96,9 @@ pub struct AuditRow {
#[derive(Debug, Clone)]
#[model(table_name = "sync_jobs")]
pub struct SyncJobRow {
pub authorized_revision: Option<i64>,
pub authorized_document: Option<String>,
pub published_document: Option<String>,
#[model(primary_key)]
pub workspace_id: String,
pub requested_revision: i64,
@@ -140,6 +143,15 @@ pub struct InstanceStateRow {
pub value: String,
}
#[derive(Debug, Clone)]
#[model(table_name = "workspace_drafts")]
pub struct WorkspaceDraftRow {
#[model(primary_key)]
pub workspace_id: String,
pub revision: i64,
pub payload: String,
}
#[derive(Debug, Clone)]
#[model(table_name = "instance_audit")]
pub struct InstanceAuditRow {
+738
View File
@@ -0,0 +1,738 @@
//! Portable, non-executable generation documents. Import only prepares a shared draft.
use crate::{
domain::*,
error::{Error, Result, ensure},
generation::{self, Node, Rule, Variable},
generation_service::Draft,
};
use serde_json::{Value, json};
use std::collections::{BTreeMap, BTreeSet};
use uuid::Uuid;
pub(super) fn fail(detail: impl ToString) -> Error {
Error::Validation(crate::i18n::message(
"transfer_invalid",
&[("detail", detail.to_string())],
))
}
pub(super) fn object(v: &Value, allowed: &[&str]) -> Result<()> {
let o = v.as_object().ok_or_else(|| fail(v))?;
if let Some(k) = o.keys().find(|k| !allowed.contains(&k.as_str())) {
return Err(fail(k));
}
Ok(())
}
fn text(v: &Value) -> Result<&str> {
v.as_str()
.filter(|s| !s.trim().is_empty())
.ok_or_else(|| fail(v))
}
fn array(v: &Value) -> Result<&[Value]> {
v.as_array().map(Vec::as_slice).ok_or_else(|| fail(v))
}
fn decode<T: serde::de::DeserializeOwned>(v: Value) -> Result<T> {
serde_json::from_value(v).map_err(fail)
}
fn normalized(s: &str) -> String {
s.trim().to_lowercase()
}
fn unique_names(values: impl Iterator<Item = String>) -> Result<()> {
let mut seen = BTreeSet::new();
for name in values {
ensure(
!name.trim().is_empty() && seen.insert(normalized(&name)),
crate::i18n::message("transfer_duplicate", &[("name", name)]),
)?;
}
Ok(())
}
fn email(doc: &WorkspaceDocument, value: &str) -> Result<String> {
if email_address::EmailAddress::is_valid(value) {
return Ok(value.trim().to_lowercase());
}
let found = Uuid::parse_str(value)
.ok()
.and_then(|id| doc.person(id))
.and_then(|p| p.email.as_deref());
found
.filter(|s| email_address::EmailAddress::is_valid(s))
.map(|s| s.trim().to_lowercase())
.ok_or_else(|| {
Error::Validation(crate::i18n::message(
"transfer_email_required",
&[(
"person",
doc.people
.iter()
.find(|p| p.id.to_string() == value)
.map(|p| p.display_name.clone())
.unwrap_or_else(|| value.into()),
)],
))
})
}
fn resolve_email(doc: &WorkspaceDocument, value: &str) -> Result<Value> {
ensure(
email_address::EmailAddress::is_valid(value),
crate::i18n::message("transfer_email_required", &[("person", value.into())]),
)?;
let matches: Vec<_> = doc
.people
.iter()
.filter(|p| {
p.email
.as_ref()
.is_some_and(|e| e.eq_ignore_ascii_case(value.trim()))
})
.collect();
ensure(
matches.len() <= 1,
crate::i18n::message("transfer_duplicate", &[("name", value.into())]),
)?;
Ok(matches
.first()
.map(|p| json!(p.id))
.unwrap_or_else(|| json!(value.trim().to_lowercase())))
}
fn literal(doc: &WorkspaceDocument, value: &mut Value, role: &str, export: bool) -> Result<()> {
if let Some(a) = value.as_array_mut() {
for v in a {
literal(doc, v, role, export)?;
}
return Ok(());
}
if value.is_null() || value.as_str() == Some("") {
return Ok(());
}
if role.is_empty() {
if !export && value.get("$person").is_some() {
object(value, &["$person"])?;
*value = resolve_email(doc, text(&value["$person"])?)?;
} else if export
&& value
.as_str()
.and_then(|s| Uuid::parse_str(s).ok())
.is_some_and(|id| doc.person(id).is_some())
{
*value = json!({"$person":email(doc,text(value)?)?});
} else if let Some(fields) = value.as_object_mut() {
for field in fields.values_mut() {
literal(doc, field, "", export)?;
}
}
return Ok(());
}
if role == "person" {
*value = if export {
json!(email(doc, text(value)?)?)
} else {
resolve_email(doc, text(value)?)?
};
} else if matches!(role, "schedule" | "escalation") {
let resources: Vec<_> = if role == "schedule" {
doc.schedules
.iter()
.map(|s| (s.id, s.name.as_str()))
.collect()
} else {
doc.escalations
.iter()
.map(|s| (s.id, s.name.as_str()))
.collect()
};
let input = text(value)?;
let found: Vec<_> = resources
.into_iter()
.filter(|(id, name)| {
if export {
id.to_string() == input
} else {
*name == input
}
})
.collect();
ensure(
found.len() == 1,
crate::i18n::message("transfer_reference", &[("name", input.into())]),
)?;
*value = if export {
json!(found[0].1)
} else {
json!(found[0].0)
};
}
Ok(())
}
fn compact(value: &mut Value) {
if let Some(o) = value.as_object_mut() {
o.retain(|k, v| {
k == "value"
|| (!v.is_null()
&& v.as_str() != Some("")
&& !v.as_array().is_some_and(Vec::is_empty))
});
}
}
fn field_role(field: &str) -> &str {
match field {
"participants" | "primary" | "secondary" => "person",
"fallback" => "schedule",
"escalation" => "escalation",
_ => "",
}
}
struct Links {
variables: BTreeMap<String, String>,
iterations: BTreeMap<String, String>,
outputs: BTreeMap<String, Value>,
}
fn binding(
b: &mut Value,
links: &Links,
doc: &WorkspaceDocument,
role: &str,
export: bool,
depth: usize,
) -> Result<()> {
ensure(depth <= 16, crate::i18n::text("transfer_limit"))?;
object(
b,
&[
"kind",
"value",
"variable",
"iteration",
"output",
"key",
"items",
"path",
"prefix",
"suffix",
],
)?;
if export {
let kind = b["kind"].as_str().unwrap_or_default().to_owned();
b.as_object_mut()
.unwrap()
.retain(|key, _| match key.as_str() {
"value" => kind == "literal",
"variable" => matches!(kind.as_str(), "variable" | "lookup"),
"iteration" => matches!(kind.as_str(), "item" | "key"),
"output" => kind == "output",
"items" => kind == "list",
"key" => kind == "lookup",
_ => true,
});
}
for field in ["variable", "iteration", "output"] {
if b.get(field).is_none_or(Value::is_null) {
continue;
}
let key = if field == "output" && !export {
b[field].to_string()
} else {
text(&b[field])?.into()
};
b[field] = if field == "output" {
links.outputs.get(&key).cloned()
} else {
(if field == "variable" {
&links.variables
} else {
&links.iterations
})
.get(&key)
.map(|s| json!(s))
}
.ok_or_else(|| {
Error::Validation(crate::i18n::message("transfer_reference", &[("name", key)]))
})?;
}
if b["kind"] == "literal" {
literal(doc, &mut b["value"], role, export)?;
}
if let Some(items) = b.get_mut("items") {
for item in items.as_array_mut().ok_or_else(|| fail("items"))? {
binding(item, links, doc, role, export, depth + 1)?;
}
}
if let Some(key) = b.get_mut("key").filter(|v| !v.is_null()) {
binding(key, links, doc, "", export, depth + 1)?;
}
if export {
compact(b);
}
Ok(())
}
fn rule_bindings(
rule: &mut Value,
links: &Links,
doc: &WorkspaceDocument,
export: bool,
) -> Result<()> {
if let Some(iterations) = rule.get_mut("iterations") {
for i in iterations
.as_array_mut()
.ok_or_else(|| fail("iterations"))?
{
binding(&mut i["source"], links, doc, "", export, 0)?;
}
}
if let Some(p) = rule.get_mut("when").filter(|v| !v.is_null()) {
object(p, &["left", "operation", "right"])?;
binding(&mut p["left"], links, doc, "", export, 0)?;
if export && p["operation"] == "not_empty" {
p["right"] = json!({"kind":"literal","value":null});
}
binding(&mut p["right"], links, doc, "", export, 0)?;
}
for block in rule["blocks"]
.as_array_mut()
.ok_or_else(|| fail("blocks"))?
{
object(
block,
&[
"id",
"kind",
"fields",
"steps",
"conditions",
"portable_key",
],
)?;
let allowed = match block["kind"].as_str() {
Some("rotation") => vec!["name", "participants", "anchor", "handover", "shift_days"],
Some("priority") => vec!["name", "participants", "fallback", "handover"],
Some("escalation") => vec!["name", "repeat_count", "repeat_wait"],
Some("route") => vec!["name", "escalation", "working"],
_ => return Err(fail(&block["kind"])),
};
object(&block["fields"], &allowed)?;
if block["kind"] != "escalation"
&& block
.get("steps")
.is_some_and(|v| v.as_array().is_none_or(|a| !a.is_empty()))
{
return Err(fail("steps"));
}
if block["kind"] != "route"
&& block
.get("conditions")
.is_some_and(|v| v.as_array().is_none_or(|a| !a.is_empty()))
{
return Err(fail("conditions"));
}
for (field, b) in block["fields"]
.as_object_mut()
.ok_or_else(|| fail("fields"))?
{
binding(b, links, doc, field_role(field), export, 0)?;
}
if let Some(steps) = block.get_mut("steps") {
for step in steps.as_array_mut().ok_or_else(|| fail("steps"))? {
object(
step,
&[
"delay",
"recipient",
"user",
"team_all",
"team_random",
"condition",
],
)?;
if step.get("condition").is_some_and(|v| {
!matches!(v.as_str(), Some("" | "if-not-acked" | "if-not-closed"))
}) {
return Err(fail(&step["condition"]));
}
binding(&mut step["delay"], links, doc, "", export, 0)?;
let role = if step["team_all"] == true || step["team_random"] == true {
""
} else if step["user"] == true {
"person"
} else {
"schedule"
};
if (step["team_all"] == true || step["team_random"] == true) && export {
step["recipient"] = json!({"kind":"literal","value":null});
}
binding(&mut step["recipient"], links, doc, role, export, 0)?;
}
}
if let Some(conditions) = block.get_mut("conditions") {
for c in conditions
.as_array_mut()
.ok_or_else(|| fail("conditions"))?
{
object(c, &["field", "operation", "value", "negate"])?;
binding(&mut c["value"], links, doc, "", export, 0)?;
}
}
if export {
compact(block);
}
}
Ok(())
}
pub(super) fn export_node(node: &Node, doc: &WorkspaceDocument) -> Result<Value> {
let mut v = serde_json::to_value(node)?;
if node.kind == "person" {
v["value"] = json!(email(doc, text(&node.value)?)?);
}
v["entries"] = Value::Array(
node.entries
.iter()
.map(|n| export_node(n, doc))
.collect::<Result<_>>()?,
);
compact(&mut v);
Ok(v)
}
/// IDs on nested entries are portable identity keys, never authority to access another workspace.
pub fn export_v1(doc: &WorkspaceDocument, draft: &Draft) -> Result<Value> {
unique_names(draft.variables.iter().map(|v| v.name.clone()))?;
unique_names(draft.rules.iter().map(|r| r.name.clone()))?;
let mut links = Links {
variables: draft
.variables
.iter()
.map(|v| (v.id.to_string(), v.name.clone()))
.collect(),
iterations: BTreeMap::new(),
outputs: BTreeMap::new(),
};
for r in &draft.rules {
for b in &r.blocks {
links
.outputs
.insert(b.id.to_string(), json!([r.name, b.id.to_string()]));
}
}
let mut rules = vec![];
for r in &draft.rules {
unique_names(r.iterations.iter().map(|i| i.name.clone()))?;
links.iterations = r
.iterations
.iter()
.map(|i| (i.id.to_string(), i.name.clone()))
.collect();
let mut value = serde_json::to_value(r)?;
value.as_object_mut().unwrap().remove("id");
rule_bindings(&mut value, &links, doc, true)?;
for i in value["iterations"].as_array_mut().unwrap() {
i.as_object_mut().unwrap().remove("id");
}
compact(&mut value);
rules.push(value);
}
Ok(
json!({"format":"meguri/v1","variables":draft.variables.iter().map(|v| Ok(json!({"name":v.name,"node":export_node(&v.node,doc)?}))).collect::<Result<Vec<_>>>()?,"generators":rules}),
)
}
fn identity(
scope: Uuid,
key: Option<&Value>,
fallback: &str,
existing: Option<Uuid>,
owned: &BTreeSet<Uuid>,
) -> Result<Uuid> {
if let Some(key) = key {
let key = text(key)?;
if let Ok(id) = Uuid::parse_str(key)
&& owned.contains(&id)
{
return Ok(id);
}
Ok(generation::stable_id(&json!(["portable", scope, key])))
} else {
Ok(
existing
.unwrap_or_else(|| generation::stable_id(&json!(["portable", scope, fallback]))),
)
}
}
fn node_ids(n: &Node, into: &mut BTreeSet<Uuid>) {
into.insert(n.id);
for child in &n.entries {
node_ids(child, into);
}
}
fn import_node(
v: &Value,
old: Option<&Node>,
scope: Uuid,
path: &str,
owned: &BTreeSet<Uuid>,
doc: &WorkspaceDocument,
depth: usize,
) -> Result<Node> {
ensure(depth <= 8, crate::i18n::text("transfer_limit"))?;
object(v, &["id", "kind", "key", "value", "entries"])?;
if matches!(v["kind"].as_str(), Some("list" | "dictionary"))
&& v.get("value")
.is_some_and(|v| !v.is_null() && v.as_str() != Some(""))
{
return Err(fail("entries"));
}
let id = identity(scope, v.get("id"), path, old.map(|n| n.id), owned)?;
let mut result = v.clone();
result["id"] = json!(id);
if v["kind"] == "person" {
result["value"] = resolve_email(doc, text(&v["value"])?)?;
}
if let Some(entries) = v.get("entries") {
result["entries"] = Value::Array(
array(entries)?
.iter()
.enumerate()
.map(|(index, entry)| {
let child_old = old.and_then(|n| {
if n.kind == "dictionary" {
n.entries
.iter()
.find(|c| Some(c.key.as_str()) == entry["key"].as_str())
} else {
n.entries.get(index)
}
});
import_node(
entry,
child_old,
scope,
&format!("{path}/{index}"),
owned,
doc,
depth + 1,
)
.and_then(|n| Ok(serde_json::to_value(n)?))
})
.collect::<Result<_>>()?,
);
}
decode(result)
}
/// Merge only named inputs. Omitted inputs and all manual resources are retained.
pub fn import(
workspace: Uuid,
doc: &WorkspaceDocument,
base: &Draft,
source: &str,
) -> Result<(Draft, Value)> {
ensure(
source.len() <= 1_000_000,
crate::i18n::text("transfer_limit"),
)?;
let mut input: Value = serde_json::from_str(source).map_err(fail)?;
if input["format"] == "meguri/v2" {
input = crate::portable_v2::expand(&input, doc, base)?;
}
object(&input, &["format", "variables", "generators"])?;
ensure(
input["format"] == "meguri/v1",
crate::i18n::text("transfer_version"),
)?;
let vars = array(&input["variables"])?;
let rules = array(&input["generators"])?;
ensure(
vars.len() <= 100 && rules.len() <= 50,
crate::i18n::text("transfer_limit"),
)?;
unique_names(
vars.iter()
.map(|v| v["name"].as_str().unwrap_or_default().into()),
)?;
unique_names(
rules
.iter()
.map(|v| v["name"].as_str().unwrap_or_default().into()),
)?;
unique_names(base.variables.iter().map(|v| v.name.clone()))?;
unique_names(base.rules.iter().map(|v| v.name.clone()))?;
let mut draft = base.clone();
let mut changes = vec![];
for v in vars {
object(v, &["name", "node"])?;
let name = text(&v["name"])?;
let old = base
.variables
.iter()
.find(|v| normalized(&v.name) == normalized(name));
let id = old.map(|v| v.id).unwrap_or_else(|| {
generation::stable_id(&json!([workspace, "variable", normalized(name)]))
});
let mut owned = BTreeSet::new();
if let Some(v) = old {
node_ids(&v.node, &mut owned);
}
let new = Variable {
id,
name: name.into(),
node: import_node(&v["node"], old.map(|v| &v.node), id, "root", &owned, doc, 0)?,
};
let same =
old.is_some_and(|o| serde_json::to_value(o).ok() == serde_json::to_value(&new).ok());
changes.push(json!({"kind":"variable","name":name,"action":if same {"unchanged"} else if old.is_some(){"update"}else{"create"}}));
if let Some(at) = draft.variables.iter().position(|v| v.id == id) {
draft.variables[at] = new;
} else {
draft.variables.push(new);
}
}
let mut prepared = vec![];
let mut links = Links {
variables: draft
.variables
.iter()
.map(|v| (v.name.clone(), v.id.to_string()))
.collect(),
iterations: BTreeMap::new(),
outputs: BTreeMap::new(),
};
for r in &base.rules {
for b in &r.blocks {
links
.outputs
.insert(json!([r.name, b.id.to_string()]).to_string(), json!(b.id));
}
}
for r in rules {
object(r, &["name", "enabled", "iterations", "when", "blocks"])?;
let name = text(&r["name"])?;
let old = base
.rules
.iter()
.find(|r| normalized(&r.name) == normalized(name));
let id = old.map(|r| r.id).unwrap_or_else(|| {
generation::stable_id(&json!([workspace, "rule", normalized(name)]))
});
let mut new = r.clone();
new["id"] = json!(id);
if new.get("enabled").is_none() {
new["enabled"] = json!(true);
}
if new.get("iterations").is_none() {
new["iterations"] = json!([]);
}
let owned: BTreeSet<_> = old
.into_iter()
.flat_map(|r| r.blocks.iter().map(|b| b.id))
.collect();
let blocks = new["blocks"].as_array_mut().ok_or_else(|| fail("blocks"))?;
let mut keys = BTreeSet::new();
for (index, b) in blocks.iter_mut().enumerate() {
let key = b
.get("id")
.map(text)
.transpose()?
.map(str::to_owned)
.unwrap_or_else(|| format!("block-{}", index + 1));
ensure(
keys.insert(key.clone()),
crate::i18n::message("transfer_duplicate", &[("name", key.clone())]),
)?;
let bid = identity(
id,
b.get("id"),
&key,
old.and_then(|r| r.blocks.get(index)).map(|b| b.id),
&owned,
)?;
links
.outputs
.insert(json!([name, key]).to_string(), json!(bid));
b["id"] = json!(bid);
}
prepared.push(new);
}
for mut r in prepared {
let name = text(&r["name"])?.to_owned();
let old = base
.rules
.iter()
.find(|r| normalized(&r.name) == normalized(&name));
let id: Uuid = decode(r["id"].clone())?;
let iterations = r["iterations"].as_array_mut().unwrap();
unique_names(
iterations
.iter()
.map(|i| i["name"].as_str().unwrap_or_default().into()),
)?;
links.iterations.clear();
for i in iterations {
object(i, &["name", "source"])?;
let name = text(&i["name"])?;
let iid = old
.and_then(|r| r.iterations.iter().find(|i| i.name == name))
.map(|i| i.id)
.unwrap_or_else(|| generation::stable_id(&json!([id, "iteration", name])));
links.iterations.insert(name.into(), iid.to_string());
i["id"] = json!(iid);
}
rule_bindings(&mut r, &links, doc, false)?;
let new: Rule = decode(r)?;
let same =
old.is_some_and(|o| serde_json::to_value(o).ok() == serde_json::to_value(&new).ok());
changes.push(json!({"kind":"rule","name":name,"action":if same {"unchanged"} else if old.is_some(){"update"}else{"create"}}));
if let Some(at) = draft.rules.iter().position(|r| r.id == new.id) {
draft.rules[at] = new;
} else {
draft.rules.push(new);
}
}
Ok((draft, Value::Array(changes)))
}
pub fn export(doc: &WorkspaceDocument, draft: &Draft) -> Result<Value> {
crate::portable_v2::compact(&export_v1(doc, draft)?, draft)
}
/// Keep small references and scalar collections on one line for hand editing.
pub fn format_config(value: &Value) -> String {
fn render(v: &Value, depth: usize) -> String {
let short = v.to_string();
if short.len() + depth * 2 <= 110 || (!v.is_array() && !v.is_object()) {
return short;
}
let indent = " ".repeat(depth + 1);
let end = " ".repeat(depth);
match v {
Value::Array(a) => format!(
"[\n{}\n{}]",
a.iter()
.map(|v| format!("{}{}", indent, render(v, depth + 1)))
.collect::<Vec<_>>()
.join(",\n"),
end
),
Value::Object(o) => {
let preferred: &[&str] = if o.contains_key("format") {
&["format", "variables", "generators"]
} else if o.contains_key("outputs") {
&["name", "enabled", "for", "when", "outputs"]
} else if o.contains_key("type") && o.contains_key("key") {
&["key", "type", "name"]
} else {
&[]
};
let keys: Vec<_> = preferred
.iter()
.filter_map(|k| o.get_key_value(*k).map(|(k, _)| k))
.chain(o.keys().filter(|k| !preferred.contains(&k.as_str())))
.collect();
format!(
"{{\n{}\n{}}}",
keys.iter()
.map(|k| format!("{}{}: {}", indent, json!(k), render(&o[*k], depth + 1)))
.collect::<Vec<_>>()
.join(",\n"),
end
)
}
_ => short,
}
}
render(value, 0)
}
+537
View File
@@ -0,0 +1,537 @@
//! Compact JSON syntax translates to the existing, validated portable importer.
use crate::{
domain::WorkspaceDocument,
error::{Result, ensure},
generation,
generation_service::Draft,
portable::{self, fail, object},
};
use serde_json::{Value, json};
use std::collections::{BTreeMap, BTreeSet};
fn arr(v: &Value) -> Result<&Vec<Value>> {
v.as_array().ok_or_else(|| fail(v))
}
fn txt(v: &Value) -> Result<&str> {
v.as_str().filter(|s| !s.is_empty()).ok_or_else(|| fail(v))
}
fn norm(s: &str) -> String {
s.trim().to_lowercase()
}
fn aliases(draft: &Draft) -> Result<BTreeMap<String, String>> {
let mut result = BTreeMap::new();
for r in &draft.rules {
let mut used: BTreeSet<String> = r
.blocks
.iter()
.filter_map(|b| b.portable_key.clone())
.collect();
ensure(
used.len() == r.blocks.iter().filter(|b| b.portable_key.is_some()).count()
&& used.iter().all(|k| !k.trim().is_empty() && k.len() <= 120),
crate::i18n::text("transfer_v2_binding"),
)?;
for b in &r.blocks {
let key = b.portable_key.clone().unwrap_or_else(|| {
let mut key = b.kind.clone();
let mut n = 2;
while used.contains(&key) {
key = format!("{}-{n}", b.kind);
n += 1;
}
used.insert(key.clone());
key
});
result.insert(b.id.to_string(), key);
}
}
Ok(result)
}
fn compact_node(n: &Value) -> Result<Value> {
Ok(match txt(&n["kind"])? {
"list" => Value::Array(
n["entries"]
.as_array()
.unwrap_or(&vec![])
.iter()
.map(compact_node)
.collect::<Result<_>>()?,
),
"dictionary" => {
let entries = n["entries"].as_array().cloned().unwrap_or_default();
let keys: Vec<_> = entries
.iter()
.map(|e| e["key"].as_str().unwrap_or_default())
.collect();
if keys.iter().any(|k| k.starts_with('$')) || !keys.windows(2).all(|w| w[0] < w[1]) {
json!({"$object":entries.iter().map(|e|Ok(json!([e["key"],compact_node(e)?]))).collect::<Result<Vec<_>>>()?})
} else {
let mut values = serde_json::Map::new();
for e in entries {
values.insert(txt(&e["key"])?.into(), compact_node(&e)?);
}
Value::Object(values)
}
}
"person" => n["value"].clone(),
"text"
if n["value"]
.as_str()
.is_some_and(email_address::EmailAddress::is_valid) =>
{
json!({"$text":n["value"]})
}
"text" | "number" | "boolean" => n["value"].clone(),
kind @ ("date" | "time" | "duration") => {
let mut o = serde_json::Map::new();
o.insert(format!("${kind}"), n["value"].clone());
Value::Object(o)
}
_ => return Err(fail(&n["kind"])),
})
}
fn compact_binding(b: &Value, names: &BTreeMap<String, String>) -> Result<Value> {
let kind = txt(&b["kind"])?;
let mut out = match kind {
"literal" => {
if b["value"].is_object() {
json!({"$literal":b["value"]})
} else {
b["value"].clone()
}
}
"list" => {
json!({"$list":arr(&b["items"] )?.iter().map(|v|compact_binding(v,names)).collect::<Result<Vec<_>>>()?})
}
"variable" | "item" | "key" => {
let key = if kind == "variable" {
"$var"
} else if kind == "item" {
"$item"
} else {
"$key"
};
let root = if kind == "variable" {
b["variable"].clone()
} else {
b["iteration"].clone()
};
let mut path = vec![root.clone()];
if let Some(a) = b["path"].as_array() {
path.extend(a.iter().cloned());
}
let mut o = serde_json::Map::new();
o.insert(key.into(), if path.len() == 1 { root } else { json!(path) });
Value::Object(o)
}
"output" => {
let pair = arr(&b["output"])?;
let alias = names.get(txt(&pair[1])?).ok_or_else(|| fail(&pair[1]))?;
json!({"$output":[pair[0],alias]})
}
"lookup" => json!({"$lookup":[b["variable"],compact_binding(&b["key"],names)?]}),
_ => return Err(fail(kind)),
};
let mut modifiers = serde_json::Map::new();
for k in ["prefix", "suffix", "path"] {
if k == "path" && matches!(kind, "variable" | "item" | "key") {
continue;
}
if let Some(v) = b.get(k).filter(|v| {
!v.is_null() && v.as_str() != Some("") && !v.as_array().is_some_and(Vec::is_empty)
}) {
modifiers.insert(k.into(), v.clone());
}
}
if !modifiers.is_empty() {
if kind == "literal" || !out.is_object() {
out = json!({"$literal":b["value"]});
}
out.as_object_mut().unwrap().extend(modifiers);
}
Ok(out)
}
fn expand_binding(v: &Value) -> Result<Value> {
if !v.is_object() {
return Ok(json!({"kind":"literal","value":v}));
}
let o = v.as_object().unwrap();
let tags: Vec<_> = o.keys().filter(|k| k.starts_with('$')).collect();
ensure(tags.len() == 1, crate::i18n::text("transfer_v2_binding"))?;
let tag = tags[0].as_str();
object(v, &[tag, "path", "prefix", "suffix"])?;
let mut out = match tag {
"$literal" => json!({"kind":"literal","value":v[tag]}),
"$var" | "$item" | "$key" => {
let (root, path) = if let Some(s) = v[tag].as_str() {
(json!(s), vec![])
} else {
let a = arr(&v[tag])?;
ensure(!a.is_empty(), crate::i18n::text("transfer_v2_binding"))?;
(a[0].clone(), a[1..].to_vec())
};
txt(&root)?;
if tag == "$var" {
json!({"kind":"variable","variable":root,"path":path})
} else {
json!({"kind":if tag=="$key"{"key"}else{"item"},"iteration":root,"path":path})
}
}
"$output" => {
let a = arr(&v[tag])?;
ensure(a.len() == 2, crate::i18n::text("transfer_v2_binding"))?;
txt(&a[0])?;
txt(&a[1])?;
json!({"kind":"output","output":a})
}
"$lookup" => {
let a = arr(&v[tag])?;
ensure(a.len() == 2, crate::i18n::text("transfer_v2_binding"))?;
txt(&a[0])?;
json!({"kind":"lookup","variable":a[0],"key":expand_binding(&a[1])?})
}
"$list" => {
json!({"kind":"list","items":arr(&v[tag])?.iter().map(expand_binding).collect::<Result<Vec<_>>>()?})
}
_ => return Err(fail(tag)),
};
for key in ["path", "prefix", "suffix"] {
if let Some(val) = v.get(key) {
if key == "path" && out[key].as_array().is_some_and(|a| !a.is_empty()) {
return Err(fail("path"));
}
out[key] = val.clone();
}
}
Ok(out)
}
fn default_field(k: &str, v: &Value) -> bool {
match k {
"handover" => v == "11:00" || v == "11:00:00",
"shift_days" => v == 7,
"repeat_count" | "repeat_wait" => v == 0,
"working" => v == false,
_ => false,
}
}
pub(super) fn compact(v1: &Value, draft: &Draft) -> Result<Value> {
let names = aliases(draft)?;
let mut variables = serde_json::Map::new();
for v in arr(&v1["variables"])? {
variables.insert(txt(&v["name"])?.into(), compact_node(&v["node"])?);
}
let mut generators = vec![];
for r in arr(&v1["generators"])? {
let mut rule = json!({"name":r["name"],"outputs":[]});
if r["enabled"] == false {
rule["enabled"] = json!(false);
}
if let Some(iterations) = r["iterations"].as_array().filter(|a| !a.is_empty()) {
rule["for"] = Value::Array(
iterations
.iter()
.map(|i| Ok(json!({"as":i["name"],"in":compact_binding(&i["source"],&names)?})))
.collect::<Result<_>>()?,
);
}
if let Some(p) = r.get("when").filter(|v| !v.is_null()) {
rule["when"] = json!([
compact_binding(&p["left"], &names)?,
p["operation"],
compact_binding(&p["right"], &names)?
]);
}
for b in arr(&r["blocks"])? {
let mut block = json!({"key":names.get(txt(&b["id"] )?).ok_or_else(||fail(&b["id"]))?,"type":b["kind"]});
for (field, binding) in b["fields"].as_object().ok_or_else(|| fail("fields"))? {
let value = compact_binding(binding, &names)?;
if !default_field(field, &value) {
block[field] = value;
}
}
if let Some(steps) = b["steps"].as_array().filter(|a| !a.is_empty()) {
block["steps"] = Value::Array(
steps
.iter()
.map(|s| {
let mut step = if s["team_all"] == true {
json!({"team":"all"})
} else if s["team_random"] == true {
json!({"team":"random"})
} else {
let key = if s["user"] == true {
"person"
} else {
"schedule"
};
let mut m = serde_json::Map::new();
m.insert(key.into(), compact_binding(&s["recipient"], &names)?);
Value::Object(m)
};
let after = compact_binding(&s["delay"], &names)?;
if after != 0 {
step["after"] = after;
}
if s["condition"] == "if-not-closed" {
step["if"] = json!("not-closed");
}
Ok(step)
})
.collect::<Result<_>>()?,
);
}
if let Some(conditions) = b["conditions"].as_array().filter(|a| !a.is_empty()) {
block["conditions"] = Value::Array(
conditions
.iter()
.map(|c| {
let mut a = vec![
c["field"].clone(),
c["operation"].clone(),
compact_binding(&c["value"], &names)?,
];
if c["negate"] == true {
a.push(json!("not"));
}
Ok(json!(a))
})
.collect::<Result<_>>()?,
);
}
rule["outputs"].as_array_mut().unwrap().push(block);
}
generators.push(rule);
}
Ok(json!({"format":"meguri/v2","variables":variables,"generators":generators}))
}
fn node_key(v: &Value) -> Option<String> {
if let Some(name) = v.get("name").and_then(Value::as_str) {
Some(format!("name:{name}"))
} else if !v.is_array() && !v.is_object() {
Some(format!("value:{v}"))
} else {
None
}
}
fn expand_node(v: &Value, old: Option<&Value>, path: &str, depth: usize) -> Result<Value> {
ensure(depth <= 8, crate::i18n::text("transfer_limit"))?;
let mut n = if let Some(a) = v.as_array() {
let prior = old
.and_then(|o| o["entries"].as_array())
.cloned()
.unwrap_or_default();
let values = prior.iter().map(compact_node).collect::<Result<Vec<_>>>()?;
let mut used = BTreeSet::new();
let mut matches = vec![None; a.len()];
for (i, item) in a.iter().enumerate() {
let candidates: Vec<_> = values
.iter()
.enumerate()
.filter(|(j, x)| {
!used.contains(j)
&& (*x == item
|| node_key(item).is_some_and(|key| node_key(x) == Some(key)))
})
.map(|(j, _)| j)
.collect();
if candidates.len() == 1 {
matches[i] = Some(candidates[0]);
used.insert(candidates[0]);
}
}
// Preserve a lone in-place edit, but never borrow an identity already matched elsewhere.
for (i, m) in matches.iter_mut().enumerate() {
if m.is_none() && prior.len() == a.len() && !used.contains(&i) {
*m = Some(i);
used.insert(i);
}
}
let entries = a
.iter()
.enumerate()
.map(|(i, item)| {
expand_node(
item,
matches[i].map(|j| &prior[j]),
&format!("{path}/{i}"),
depth + 1,
)
})
.collect::<Result<Vec<_>>>()?;
json!({"kind":"list","entries":entries,"value":old.map(|v|v["value"].clone()).unwrap_or(Value::Null)})
} else if let Some(o) = v.as_object() {
if let Some((tag, value)) = o.iter().find(|(k, _)| k.starts_with('$')) {
object(v, &[tag])?;
match tag.as_str() {
"$text" | "$person" | "$date" | "$time" | "$duration" => {
json!({"kind":&tag[1..],"value":value})
}
"$object" => dictionary(
arr(value)?
.iter()
.map(|pair| {
let pair = arr(pair)?;
ensure(pair.len() == 2, crate::i18n::text("transfer_v2_binding"))?;
Ok((txt(&pair[0])?.to_owned(), pair[1].clone()))
})
.collect::<Result<Vec<_>>>()?,
old,
path,
depth,
)?,
_ => return Err(fail(tag)),
}
} else {
dictionary(
o.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
old,
path,
depth,
)?
}
} else if let Some(s) = v.as_str() {
json!({"kind":if email_address::EmailAddress::is_valid(s){"person"}else{"text"},"value":v})
} else if v.is_boolean() {
json!({"kind":"boolean","value":v})
} else if v.is_number() {
json!({"kind":"number","value":v})
} else {
return Err(fail(v));
};
n["id"] = old
.map(|o| o["id"].clone())
.unwrap_or_else(|| json!(generation::stable_id(&json!(["v2-entry", path, v]))));
Ok(n)
}
fn dictionary(
entries: Vec<(String, Value)>,
old: Option<&Value>,
path: &str,
depth: usize,
) -> Result<Value> {
let mut seen = BTreeSet::new();
let mut nodes = vec![];
for (key, value) in entries {
ensure(
seen.insert(key.clone()),
crate::i18n::message("transfer_duplicate", &[("name", key.clone())]),
)?;
let previous = old
.and_then(|o| o["entries"].as_array())
.and_then(|a| a.iter().find(|n| n["key"] == key));
let mut node = expand_node(&value, previous, &format!("{path}/{key}"), depth + 1)?;
node["key"] = json!(key);
nodes.push(node);
}
Ok(
json!({"kind":"dictionary","entries":nodes,"value":old.map(|v|v["value"].clone()).unwrap_or(Value::Null)}),
)
}
pub(super) fn expand(v: &Value, doc: &WorkspaceDocument, base: &Draft) -> Result<Value> {
object(v, &["format", "variables", "generators"])?;
let old = json!({"variables":base.variables.iter().map(|v|json!({"name":v.name,"node":portable::export_node(&v.node,doc).unwrap_or_else(|_|serde_json::to_value(&v.node).expect("typed node"))})).collect::<Vec<_>>()});
let names = aliases(base)?;
let mut vars = vec![];
for (name, value) in v["variables"]
.as_object()
.ok_or_else(|| fail("variables"))?
{
let previous = old["variables"]
.as_array()
.unwrap()
.iter()
.find(|p| norm(p["name"].as_str().unwrap_or_default()) == norm(name));
vars.push(
json!({"name":name,"node":expand_node(value,previous.map(|p|&p["node"]),name,0)?}),
);
}
let mut outputs = BTreeMap::new();
for r in &base.rules {
for b in &r.blocks {
outputs.insert(
json!([norm(&r.name), names[&b.id.to_string()]]).to_string(),
json!(b.id),
);
}
}
for r in arr(&v["generators"])? {
for b in arr(&r["outputs"])? {
let key = txt(&b["key"])?;
outputs
.entry(json!([norm(txt(&r["name"])?), key]).to_string())
.or_insert_with(|| json!(key));
}
}
let mut rules = vec![];
for r in arr(&v["generators"])? {
object(r, &["name", "enabled", "for", "when", "outputs"])?;
let name = txt(&r["name"])?;
let mut rule = json!({"name":name,"enabled":r.get("enabled").cloned().unwrap_or(json!(true)),"blocks":[]});
if let Some(iterations) = r.get("for") {
rule["iterations"] = Value::Array(
arr(iterations)?
.iter()
.map(|i| {
object(i, &["as", "in"])?;
Ok(json!({"name":i["as"],"source":expand_binding(&i["in"])?}))
})
.collect::<Result<_>>()?,
);
}
if let Some(p) = r.get("when") {
let p = arr(p)?;
ensure(p.len() == 3, crate::i18n::text("transfer_v2_binding"))?;
rule["when"] = json!({"left":expand_binding(&p[0])?,"operation":p[1],"right":expand_binding(&p[2])?});
}
for b in arr(&r["outputs"])? {
let key = txt(&b["key"])?;
let mut block = json!({"id":outputs.get(&json!([norm(name),key]).to_string()).ok_or_else(||fail(key))?,"portable_key":key,"kind":b["type"],"fields":{}});
for (field, value) in b.as_object().ok_or_else(|| fail(b))? {
if !["key", "type", "steps", "conditions"].contains(&field.as_str()) {
block["fields"][field] = expand_binding(value)?;
}
}
if let Some(steps) = b.get("steps") {
block["steps"]=Value::Array(arr(steps)?.iter().map(|s|{
object(s,&["after","if","person","schedule","team"])?;
let recipients:Vec<_>=["person","schedule","team"].into_iter().filter(|k|s.get(k).is_some()).collect();ensure(recipients.len()==1,crate::i18n::text("transfer_v2_binding"))?;
let condition=match s.get("if").and_then(Value::as_str){None|Some("not-acked")=>"if-not-acked",Some("not-closed")=>"if-not-closed",_=>return Err(fail(&s["if"]))};
let recipient=if recipients[0]=="team"{ensure(s["team"]=="all" || s["team"]=="random",crate::i18n::text("transfer_v2_binding"))?;json!({"kind":"literal","value":null})}else{expand_binding(&s[recipients[0]])?};
Ok(json!({"delay":expand_binding(s.get("after").unwrap_or(&json!(0)))?,"condition":condition,"user":recipients[0]=="person","team_all":s["team"]=="all","team_random":s["team"]=="random","recipient":recipient}))
}).collect::<Result<_>>()?);
}
if let Some(conditions) = b.get("conditions") {
block["conditions"]=Value::Array(arr(conditions)?.iter().map(|c|{let a=arr(c)?;ensure(a.len()==3||(a.len()==4&&a[3]=="not"),crate::i18n::text("transfer_v2_binding"))?;Ok(json!({"field":a[0],"operation":a[1],"value":expand_binding(&a[2])?,"negate":a.len()==4}))}).collect::<Result<_>>()?);
}
rule["blocks"].as_array_mut().unwrap().push(block);
}
remap_outputs(&mut rule, &outputs)?;
rules.push(rule);
}
Ok(json!({"format":"meguri/v1","variables":vars,"generators":rules}))
}
fn remap_outputs(v: &mut Value, outputs: &BTreeMap<String, Value>) -> Result<()> {
if v["kind"] == "literal" {
return Ok(());
}
if v["kind"] == "output" {
let old = v["output"].clone();
let id = outputs
.get(&json!([norm(txt(&old[0])?), old[1]]).to_string())
.ok_or_else(|| fail(&old))?;
v["output"] = json!([old[0], id]);
}
match v {
Value::Object(o) => {
for v in o.values_mut() {
remap_outputs(v, outputs)?;
}
}
Value::Array(a) => {
for v in a {
remap_outputs(v, outputs)?;
}
}
_ => (),
};
Ok(())
}
+588 -68
View File
@@ -8,6 +8,7 @@ use crate::{
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;
@@ -168,7 +169,13 @@ impl Jsm {
let mut ids: BTreeSet<_> = plan
.schedules
.iter()
.flat_map(|s| s.periods.iter().map(|p| p.user_id))
.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
@@ -241,6 +248,7 @@ impl Jsm {
})?))
}
async fn list(&self, path: &str) -> Result<Vec<Value>> {
const PAGE_SIZE: usize = 50;
let mut all = vec![];
let mut offset = 0;
loop {
@@ -248,7 +256,7 @@ impl Jsm {
let page = self
.request(
Method::GET,
&format!("{path}{sep}offset={offset}&size=100"),
&format!("{path}{sep}offset={offset}&size={PAGE_SIZE}"),
None,
)
.await?
@@ -269,7 +277,7 @@ impl Jsm {
.pointer("/links/next")
.and_then(Value::as_str)
.is_none_or(str::is_empty)
&& count < 100
&& count < PAGE_SIZE
{
break;
}
@@ -326,8 +334,36 @@ impl Jsm {
desired: &Value,
previous: Option<&ProviderObjectRow>,
overwrite: bool,
as_of: DateTime<Utc>,
) -> Result<String> {
let name = desired["name"].as_str().unwrap_or_default();
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 {
@@ -335,7 +371,11 @@ impl Jsm {
.list(&self.collection(kind))
.await?
.into_iter()
.filter(|v| v["name"].as_str() == Some(name))
.filter(|v| {
v["name"]
.as_str()
.is_some_and(|name| name.ends_with(&marker) || name == legacy_name)
})
.collect();
ensure(
found.len() <= 1,
@@ -354,18 +394,32 @@ impl Jsm {
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 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));
if !matches_desired && !known && !overwrite {
.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())],
@@ -373,48 +427,75 @@ impl Jsm {
}
}
if kind == "schedule"
&& current.is_some()
&& 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,
)
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();
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(),
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
@@ -450,8 +531,303 @@ impl Jsm {
)));
}
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,
@@ -459,6 +835,9 @@ impl Jsm {
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?
@@ -481,8 +860,8 @@ impl Jsm {
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 (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,
@@ -492,8 +871,11 @@ impl Jsm {
&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 {
@@ -520,7 +902,7 @@ impl Jsm {
};
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 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,
@@ -530,6 +912,7 @@ impl Jsm {
&body,
previous.get(key.as_str()).copied(),
overwrite,
plan.as_of,
)
.await?;
policies.insert(e.id, remote);
@@ -551,7 +934,7 @@ impl Jsm {
} 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});
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 = [
@@ -577,6 +960,7 @@ impl Jsm {
&body,
previous.get(key.as_str()).copied(),
overwrite,
plan.as_of,
)
.await?;
}
@@ -595,19 +979,21 @@ impl Jsm {
.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(),
));
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,
@@ -617,14 +1003,84 @@ impl Jsm {
.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 owned_name(team: Uuid, id: Uuid) -> String {
format!("meguri-{}-{}", team.simple(), id.simple())
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()
@@ -646,9 +1102,45 @@ pub fn equivalent(actual: &Value, expected: &Value) -> bool {
{
return false;
}
expected
.iter()
.all(|(k, v)| equivalent(actual.get(k).unwrap_or(&Value::Null), v))
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))
@@ -670,6 +1162,34 @@ pub fn equivalent(actual: &Value, expected: &Value) -> bool {
#[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"}]}]});
+44
View File
@@ -15,11 +15,14 @@ pub struct PublishedSchedule {
pub name: String,
pub timezone: String,
pub periods: Vec<Assignment>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub recurring: Option<RecurringRotation>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Publication {
pub workspace_id: Uuid,
pub revision: i64,
pub as_of: DateTime<Utc>,
pub from: DateTime<Utc>,
pub to: DateTime<Utc>,
pub schedules: Vec<PublishedSchedule>,
@@ -149,6 +152,7 @@ pub fn compile(
id: schedule.id,
name: schedule.name.clone(),
timezone: record.document.timezone.clone(),
recurring: crate::engine::recurring_rotation(&record.document, schedule, &periods)?,
periods,
});
}
@@ -156,6 +160,7 @@ pub fn compile(
Ok(Publication {
workspace_id: record.id,
revision: record.revision,
as_of: now,
from: timeline.from,
to: timeline.to,
schedules,
@@ -230,6 +235,45 @@ pub async fn publish(
mod tests {
use super::*;
#[test]
fn jsm_team_identifier_preserves_operations_prefix_and_rejects_invalid_paths() {
let cloud = Uuid::new_v4().to_string();
let team = Uuid::new_v4().to_string();
for valid_team in [team.clone(), format!("og-{team}")] {
let mut doc = WorkspaceDocument::new("Platform".into());
doc.integration = Integration::Jsm {
cloud_id: cloud.clone(),
team_id: valid_team.clone(),
advanced_escalations: false,
users: Default::default(),
};
doc.validate(&Default::default()).unwrap();
let saved: WorkspaceDocument =
serde_json::from_value(serde_json::to_value(&doc).unwrap()).unwrap();
assert!(
matches!(saved.integration, Integration::Jsm { team_id, .. } if team_id == valid_team)
);
for invalid in [
"og-".to_owned(),
format!("og-og-{team}"),
format!("og-{team}/routing-rules"),
format!("og-{team}?x=1"),
] {
if let Integration::Jsm { team_id, .. } = &mut doc.integration {
*team_id = invalid;
}
assert!(doc.validate(&Default::default()).is_err());
}
if let Integration::Jsm {
cloud_id, team_id, ..
} = &mut doc.integration
{
*team_id = valid_team;
*cloud_id = format!("og-{cloud}");
}
assert!(doc.validate(&Default::default()).is_err());
}
}
#[test]
fn unsupported_escalation_is_not_silently_dropped() {
let mut doc = WorkspaceDocument::new("Platform".into());
doc.integration = Integration::Jsm {
+101 -67
View File
@@ -192,8 +192,15 @@ pub async fn update_document(
!query!(UserRow,$id==id).exists(&db.db).await?,
crate::i18n::text("err_select_the_workspace_user_from_the_directory_725423"),
)?;
} else if let Some(id) = person.user_id {
// A new person may only link to an account that currently has access here.
ensure(
members.contains(&id),
crate::i18n::text("err_choose_a_user_who_has_access_to_this_workspace_41cf15"),
)?;
}
}
store::account_people(&mut document, &db.directory(workspace).await?)?;
let interval = db.working_interval(workspace).await?;
document.timezone = interval.timezone.clone();
for route in &mut document.routes {
@@ -479,27 +486,21 @@ async fn prepare_absence(
!query!(UserRow,$id==key).exists(&db.db).await?,
crate::i18n::text("err_select_an_existing_workspace_user_from_the_direc_a09a03"),
)?;
} else if let Some(id) = person.user_id {
ensure(
participants.contains(&id),
crate::i18n::text("err_choose_a_user_who_has_access_to_this_workspace_41cf15"),
)?;
}
record.document.people.push(person);
}
store::account_people(&mut record.document, &db.directory(workspace).await?)?;
record.document.validate(&participants)?;
let mut plan = crate::absence::plan(&record.document, &input.draft, &participants, now)?;
if !input.force {
let members = db.member_ids(workspace).await?;
for id in plan
.groups
.iter()
.flatten()
.map(|l| l.to_user)
.collect::<BTreeSet<_>>()
{
if !members.contains(&id) {
plan.errors.push(
crate::i18n::text(
"err_a_selected_replacement_has_no_team_access_to_con_dd538a",
)
.into(),
);
for legs in &plan.groups {
if let Err(error) = validate_approval_access(&record.document, legs, &participants) {
plan.errors.push(error.to_string());
}
}
}
@@ -513,7 +514,9 @@ pub async fn preview_absence(
input: NewAbsence,
) -> Result<serde_json::Value> {
let (record, _, plan) = prepare_absence(db, actor, workspace, &input, Utc::now()).await?;
Ok(json!({"revision":record.revision,"plan":plan}))
let members = db.member_ids(workspace).await?;
let approval_groups:Vec<_>=plan.groups.iter().map(|legs|json!({"legs":legs,"required_approvers":if input.force {BTreeSet::new()} else {required_approvers(actor,legs,&members)}})).collect();
Ok(json!({"revision":record.revision,"plan":plan,"approval_groups":approval_groups}))
}
pub async fn create_absence(
@@ -533,6 +536,7 @@ pub async fn create_absence(
record.document.requests.len() + plan.groups.len() <= 5000,
crate::i18n::text("err_replacement_history_limit_reached_3b1168"),
)?;
let members = db.member_ids(workspace).await?;
record.document.absences.push(input.draft.absence.clone());
for legs in &plan.groups {
let mut request =
@@ -540,20 +544,18 @@ pub async fn create_absence(
request.id = Uuid::new_v4();
request.requested_by = actor;
request.reason = input.reason.trim().into();
request.required_approvers = legs
.iter()
.flat_map(|l| [l.from_user, l.to_user])
.filter(|id| *id != actor)
.collect();
request.status = if input.force {
request.required_approvers = required_approvers(actor, legs, &members);
let apply = input.force || request.required_approvers.is_empty();
request.status = if apply {
RequestStatus::Applied
} else {
RequestStatus::Pending
};
request.applied_at = input.force.then_some(now);
request.applied_at = apply.then_some(now);
request.forced_by = input.force.then_some(actor);
request.force_reason = input.force.then_some(input.reason.clone());
if input.force {
if apply {
crate::absence::validate_request(&record.document, &request, &members, now)?;
engine::supersede_overrides(&mut record.document, &request.legs);
}
record.document.requests.push(request);
@@ -572,11 +574,7 @@ pub async fn request_replacement(
input.reason.len() <= 1000,
crate::i18n::text("err_comment_must_be_at_most_1000_characters_e5d5ef"),
)?;
let permission = if input.force {
Permission::ForceOverrides
} else {
Permission::RequestOverrides
};
let permission = replacement_permission(actor, &input);
let mut record = db
.prepare(workspace, input.revision, actor, permission)
.await?;
@@ -588,14 +586,12 @@ pub async fn request_replacement(
record.document.requests.len() < 5000,
crate::i18n::text("err_replacement_history_limit_reached_3b1168"),
)?;
validate_request_participation(actor, &input)?;
validate_request_access(&input, &members)?;
let required_approvers = input
.legs
.iter()
.flat_map(|l| [l.from_user, l.to_user])
.filter(|id| *id != actor)
.collect();
validate_request_participation(actor, &input, &record.document)?;
if !input.force {
validate_approval_access(&record.document, &input.legs, &members)?;
}
let required_approvers = required_approvers(actor, &input.legs, &members);
let apply = input.force || required_approvers.is_empty();
let request = ReplacementRequest {
effective_legs: None,
id: Uuid::new_v4(),
@@ -606,17 +602,17 @@ pub async fn request_replacement(
legs: input.legs,
required_approvers,
approvals: BTreeSet::new(),
status: if input.force {
status: if apply {
RequestStatus::Applied
} else {
RequestStatus::Pending
},
applied_at: input.force.then_some(now),
applied_at: apply.then_some(now),
forced_by: input.force.then_some(actor),
force_reason: input.force.then_some(input.reason),
};
let details = serde_json::to_value(&request)?;
if input.force {
if apply {
engine::supersede_overrides(&mut record.document, &request.legs);
}
record.document.requests.push(request);
@@ -636,25 +632,58 @@ pub async fn request_replacement(
Ok(record)
}
fn validate_request_participation(actor: Uuid, input: &NewRequest) -> Result<()> {
fn planning_contact(doc: &WorkspaceDocument, id: Uuid) -> bool {
doc.person(id).is_some_and(|p| p.user_id.is_none())
}
fn required_approvers(
actor: Uuid,
legs: &[ReplacementLeg],
members: &BTreeSet<Uuid>,
) -> BTreeSet<Uuid> {
legs.iter()
.flat_map(|l| [l.from_user, l.to_user])
.filter(|id| *id != actor && members.contains(id))
.collect()
}
fn replacement_permission(actor: Uuid, input: &NewRequest) -> Permission {
if input.force {
Permission::ForceOverrides
} else if input
.legs
.iter()
.any(|l| l.from_user != actor && l.to_user != actor)
{
Permission::EditSchedules
} else {
Permission::RequestOverrides
}
}
fn validate_request_participation(
actor: Uuid,
input: &NewRequest,
doc: &WorkspaceDocument,
) -> Result<()> {
ensure(
input.force
|| input
.legs
.iter()
.all(|leg| leg.from_user == actor || leg.to_user == actor),
|| input.legs.iter().all(|leg| {
leg.from_user == actor
|| leg.to_user == actor
|| planning_contact(doc, leg.from_user)
|| planning_contact(doc, leg.to_user)
}),
crate::i18n::text("err_you_can_give_your_own_shift_or_offer_to_cover_a_5e6f57"),
)
}
fn validate_request_access(input: &NewRequest, members: &BTreeSet<Uuid>) -> Result<()> {
fn validate_approval_access(
doc: &WorkspaceDocument,
legs: &[ReplacementLeg],
members: &BTreeSet<Uuid>,
) -> Result<()> {
ensure(
input.force
|| input
.legs
.iter()
.all(|l| members.contains(&l.from_user) && members.contains(&l.to_user)),
crate::i18n::text("err_both_people_need_team_access_to_approve_ask_a_le_7b0fc5"),
legs.iter()
.flat_map(|l| [l.from_user, l.to_user])
.all(|id| members.contains(&id) || planning_contact(doc, id)),
crate::i18n::text("replacement_account_access"),
)
}
@@ -673,17 +702,15 @@ pub async fn preview_replacement(
workspace: Uuid,
mut input: NewRequest,
) -> Result<serde_json::Value> {
let permission = if input.force {
Permission::ForceOverrides
} else {
Permission::RequestOverrides
};
let permission = replacement_permission(actor, &input);
let record = db
.prepare(workspace, input.revision, actor, permission)
.await?;
validate_request_participation(actor, &input)?;
validate_request_participation(actor, &input, &record.document)?;
let members = db.member_ids(workspace).await?;
validate_request_access(&input, &members)?;
if !input.force {
validate_approval_access(&record.document, &input.legs, &members)?;
}
ensure(
input.reason.len() <= 1000,
crate::i18n::text("err_comment_must_be_at_most_1000_characters_e5d5ef"),
@@ -691,7 +718,14 @@ pub async fn preview_replacement(
let now = Utc::now();
clip_requested_interval(&mut input, now);
engine::validate_legs(&record.document, &input.legs, &members, now)?;
Ok(serde_json::json!({"legs": input.legs}))
let approvers = if input.force {
BTreeSet::new()
} else {
required_approvers(actor, &input.legs, &members)
};
Ok(
serde_json::json!({"legs": input.legs,"required_approvers":approvers,"applies_immediately":approvers.is_empty()}),
)
}
#[derive(Debug, Deserialize)]
@@ -958,16 +992,16 @@ WITH previous AS MATERIALIZED (SELECT * FROM meguri__users WHERE id=$1 FOR NO KE
UPDATE meguri__users SET display_name=$2,country=$3,timezone=$4,password_hash=$5,language=$6 WHERE id IN(SELECT id FROM previous) RETURNING *
), changed_workspaces AS (
UPDATE meguri__workspaces t SET revision=t.revision+1,document=jsonb_set(t.document::jsonb,'{people}',
COALESCE((SELECT jsonb_agg(CASE WHEN p->>'user_id'=$1 THEN p||jsonb_build_object('country',$3::text) ELSE p END) FROM jsonb_array_elements(COALESCE(t.document::jsonb->'people','[]'::jsonb)) p),'[]'::jsonb)
|| CASE WHEN EXISTS(SELECT 1 FROM jsonb_array_elements(COALESCE(t.document::jsonb->'people','[]'::jsonb)) p WHERE p->>'id'=$1) THEN '[]'::jsonb ELSE jsonb_build_array(jsonb_build_object('id',$1::text,'user_id',$1::text,'display_name',$2::text,'email',null,'country',$3::text)) END)::text
WHERE EXISTS(SELECT 1 FROM previous JOIN changed USING(id) WHERE previous.country<>changed.country)
COALESCE((SELECT jsonb_agg(CASE WHEN p->>'user_id'=$1 THEN p||jsonb_build_object('country',$3::text,'display_name',$2::text,'email',(SELECT email FROM changed)) ELSE p END) FROM jsonb_array_elements(COALESCE(t.document::jsonb->'people','[]'::jsonb)) p),'[]'::jsonb)
|| CASE WHEN EXISTS(SELECT 1 FROM jsonb_array_elements(COALESCE(t.document::jsonb->'people','[]'::jsonb)) p WHERE p->>'id'=$1) THEN '[]'::jsonb ELSE jsonb_build_array(jsonb_build_object('id',$1::text,'user_id',$1::text,'display_name',$2::text,'email',(SELECT email FROM changed),'country',$3::text)) END)::text
WHERE EXISTS(SELECT 1 FROM previous JOIN changed USING(id) WHERE previous.country<>changed.country OR previous.display_name<>changed.display_name)
AND (EXISTS(SELECT 1 FROM meguri__workspace_members WHERE workspace_id=t.id AND user_id=$1) OR EXISTS(SELECT 1 FROM jsonb_array_elements(COALESCE(t.document::jsonb->'people','[]'::jsonb)) p WHERE p->>'user_id'=$1)) RETURNING t.*
), history AS (
INSERT INTO meguri__workspace_history(id,workspace_id,revision,document,effective_at,actor_id) SELECT id||':'||revision::text,id,revision,document,clock_timestamp(),$1 FROM changed_workspaces RETURNING workspace_id,effective_at
), jobs AS (
INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,revision,'pending',0,now() FROM changed_workspaces
ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision,state='pending',attempts=0,next_attempt=now(),last_error=NULL
INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,revision,'draft',0,now() FROM changed_workspaces
ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision
)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at)
SELECT id,$1,'person.country_changed',jsonb_build_object('user_id',$1::text,'country',$3::text)::text,(SELECT effective_at FROM history WHERE workspace_id=changed_workspaces.id) FROM changed_workspaces
SELECT id,$1,'person.profile_changed',jsonb_build_object('user_id',$1::text,'display_name',$2::text,'country',$3::text)::text,(SELECT effective_at FROM history WHERE workspace_id=changed_workspaces.id) FROM changed_workspaces
"#;
+174 -26
View File
@@ -30,16 +30,43 @@ pub fn uuid(value: &str) -> Result<Uuid> {
pub fn user(row: UserRow) -> Result<User> {
Ok(User {
language: row.language,
id: uuid(&row.id)?,
person: Person {
id: uuid(&row.id)?,
user_id: Some(uuid(&row.id)?),
display_name: row.display_name,
email: row.email,
country: Some(row.country),
aliases: vec![],
},
username: row.username,
display_name: row.display_name,
country: row.country,
timezone: row.timezone,
email: row.email,
instance_admin: row.instance_admin,
has_password: row.password_hash.is_some(),
})
}
/// Workspace copies of account people are immutable identity snapshots.
pub fn account_people(document: &mut WorkspaceDocument, users: &[User]) -> Result<()> {
for user in users {
if let Some(person) = document.people.iter_mut().find(|p| p.id == user.id) {
crate::error::ensure(
person.user_id == Some(user.id)
&& person
.email
.as_ref()
.is_none_or(|e| Some(e) == user.email.as_ref()),
crate::i18n::text("account_person_authoritative"),
)?;
// The account owns the profile; merged contact IDs stay as aliases for history.
let aliases = std::mem::take(&mut person.aliases);
*person = user.person.clone();
person.aliases = aliases;
} else {
document.people.push(user.person.clone());
}
}
Ok(())
}
pub fn record(row: WorkspaceRow) -> Result<WorkspaceRecord> {
if row.demo_deleted_at.is_some() || row.disabled_at.is_some() {
return Err(Error::NotFound);
@@ -200,6 +227,11 @@ impl Store {
details: Value,
role_permissions: Option<Vec<Permission>>,
) -> Result<()> {
let users = self.directory(record.id).await?;
account_people(&mut record.document, &users)?;
record
.document
.validate(&users.iter().map(|u| u.id).collect())?;
let workspace = record.id.to_string();
let actor = actor.to_string();
let document = serde_json::to_string(&record.document)?;
@@ -295,7 +327,7 @@ impl Store {
pub async fn audit(&self, workspace: Uuid) -> Result<Vec<Value>> {
let mut names = crate::audit::Names::new();
for person in self.directory(workspace).await? {
names.insert(person.id.to_string(), person.display_name);
names.insert(person.id.to_string(), person.display_name.clone());
}
let current = self.workspace(workspace).await?;
crate::audit::remember(&mut names, &current.document);
@@ -342,20 +374,79 @@ impl Store {
output.sort_by_key(|e| std::cmp::Reverse(e["id"].as_i64().unwrap_or_default()));
Ok(output)
}
pub async fn published_document(&self, workspace: Uuid) -> Result<Option<WorkspaceDocument>> {
let id = workspace.to_string();
query!(SyncJobRow, $workspace_id == id)
.get(&self.db)
.await?
.and_then(|r| r.published_document)
.map(|s| serde_json::from_str(&s).map_err(Into::into))
.transpose()
}
pub async fn sync_status(&self, workspace: Uuid) -> Result<Value> {
let workspace = workspace.to_string();
let record = self.workspace(workspace).await?;
let baseline = self.published_document(workspace).await?;
let changed = serde_json::to_value(&record.document)? != serde_json::to_value(&baseline)?;
let id = workspace.to_string();
let mut status = match query!(SyncJobRow, $workspace_id == id)
.get(&self.db)
.await?
{
None => json!({"state":"draft"}),
Some(r) => {
let publication_busy = r.lease_until.is_some_and(|until| until > Utc::now());
json!({"publication_busy":publication_busy,"lease_until":r.lease_until,"requested_revision":r.requested_revision,"authorized_revision":r.authorized_revision,"published_revision":r.published_revision,"state":r.state,"attempts":r.attempts,"last_error":r.last_error,"last_success":r.last_success,"window_end":r.window_end})
}
};
status["has_changes"] = json!(changed);
status["changes"] = publication_changes(&record.document, baseline.as_ref())?;
Ok(status)
}
pub async fn shared_draft(&self, workspace: Uuid) -> Result<Value> {
let id = workspace.to_string();
Ok(
match query!(SyncJobRow,$workspace_id==workspace)
match query!(WorkspaceDraftRow, $workspace_id == id)
.get(&self.db)
.await?
{
None => json!({"state":"pending"}),
Some(r) => {
json!({"requested_revision":r.requested_revision,"published_revision":r.published_revision,"state":r.state,"attempts":r.attempts,"last_error":r.last_error,"last_success":r.last_success,"window_end":r.window_end})
Some(row) => {
json!({"revision":row.revision,"payload":serde_json::from_str::<Value>(&row.payload)?})
}
None => json!({"revision":0,"payload":null}),
},
)
}
pub async fn save_shared_draft(
&self,
workspace: Uuid,
actor: Uuid,
revision: i64,
payload: Value,
) -> Result<Value> {
self.require(workspace, actor, Permission::EditSchedules)
.await?;
// Deserialize into the bounded typed input; never store arbitrary integration secrets.
let draft: crate::generation_service::Draft = serde_json::from_value(payload)?;
self.prepare(workspace, draft.revision, actor, Permission::EditSchedules)
.await?;
crate::error::ensure(
draft.variables.len() <= 100 && draft.rules.len() <= 50,
crate::i18n::text("err_shared_draft_limit"),
)?;
let payload = serde_json::to_string(&draft)?;
let changed = self.db.raw_with(r#"
WITH allowed AS (SELECT id FROM meguri__workspaces WHERE id=$1 AND revision=$5 AND demo_deleted_at IS NULL AND disabled_at IS NULL AND (EXISTS(SELECT 1 FROM meguri__users WHERE id=$4 AND instance_admin) OR EXISTS(SELECT 1 FROM meguri__workspace_members WHERE workspace_id=$1 AND user_id=$4 AND (is_owner OR permissions::jsonb @> '["edit_schedules"]'::jsonb))))
, saved AS (INSERT INTO meguri__workspace_drafts(workspace_id,revision,payload) SELECT id,1,$3 FROM allowed WHERE $2=0 OR EXISTS(SELECT 1 FROM meguri__workspace_drafts WHERE workspace_id=$1)
ON CONFLICT(workspace_id) DO UPDATE SET revision=meguri__workspace_drafts.revision+1,payload=EXCLUDED.payload WHERE meguri__workspace_drafts.revision=$2 RETURNING workspace_id,revision)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at) SELECT workspace_id,$4,'generation.draft_saved',jsonb_build_object('draft_revision',revision)::text,now() FROM saved
"#, &[&workspace.to_string(), &revision, &payload, &actor.to_string(), &draft.revision]).await?;
if changed.rows_affected().0 == 0 {
return Err(Error::Conflict(
crate::i18n::text("err_shared_draft_stale").into(),
));
}
Ok(json!({"revision":revision+1,"payload":draft}))
}
pub async fn historical_timeline(
&self,
workspace: &WorkspaceRecord,
@@ -442,19 +533,20 @@ impl Store {
query!(ProviderObjectRow,$id==id).delete(&self.db).await?;
Ok(())
}
pub async fn retry(&self, workspace: Uuid, actor: Uuid) -> Result<()> {
self.require(workspace, actor, Permission::ManageIntegration)
pub async fn retry(&self, workspace: Uuid, actor: Uuid, revision: i64) -> Result<()> {
self.prepare(workspace, revision, actor, Permission::ManageIntegration)
.await?;
let mut record = self.workspace(workspace).await?;
self.commit(
&mut record,
actor,
Permission::ManageIntegration,
"publication.requested",
json!({}),
None,
)
.await
let result = self.db.raw_with(r#"
WITH selected AS (SELECT w.* FROM meguri__workspaces w WHERE id=$1 AND revision=$3 AND demo_deleted_at IS NULL AND disabled_at IS NULL AND (EXISTS(SELECT 1 FROM meguri__users WHERE id=$2 AND instance_admin) OR EXISTS(SELECT 1 FROM meguri__workspace_members WHERE workspace_id=$1 AND user_id=$2 AND (is_owner OR permissions::jsonb @> '["manage_integration"]'::jsonb))) FOR NO KEY UPDATE),
queued AS (UPDATE meguri__sync_jobs j SET authorized_revision=w.revision,authorized_document=w.document,state='pending',attempts=0,next_attempt=now(),last_error=NULL FROM selected w WHERE j.workspace_id=w.id AND (lease_until IS NULL OR lease_until<now()) RETURNING j.workspace_id)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at) SELECT workspace_id,$2,'publication.requested',jsonb_build_object('revision',$3::bigint)::text,now() FROM queued
"#, &[&workspace.to_string(), &actor.to_string(), &revision]).await?;
if result.rows_affected().0 == 0 {
return Err(Error::Conflict(
crate::i18n::text("err_publication_busy_or_stale").into(),
));
}
Ok(())
}
}
@@ -462,6 +554,7 @@ const COMMIT_WORKSPACE: &str = r#"
WITH updated AS (
UPDATE meguri__workspaces SET document=$3,revision=revision+1
WHERE id=$1 AND revision=$2 AND demo_deleted_at IS NULL AND disabled_at IS NULL
AND ($5 <> 'generation.applied' OR ($6::jsonb->>'shared_revision') IS NULL OR EXISTS(SELECT 1 FROM meguri__workspace_drafts WHERE workspace_id=$1 AND revision=($6::jsonb->>'shared_revision')::bigint))
AND (EXISTS(SELECT 1 FROM meguri__users WHERE id=$4 AND instance_admin)
OR EXISTS(SELECT 1 FROM meguri__workspace_members WHERE workspace_id=$1 AND user_id=$4 AND (is_owner OR permissions::jsonb @> $7::jsonb))) RETURNING *
), roles AS (
@@ -471,8 +564,8 @@ WITH updated AS (
SELECT id||':'||revision::text,id,revision,document,clock_timestamp(),$4 FROM updated RETURNING effective_at
), jobs AS (
INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt)
SELECT id,revision,'pending',0,now() FROM updated
ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision,state='pending',attempts=0,next_attempt=now(),last_error=NULL
SELECT id,revision,'draft',0,now() FROM updated
ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision
)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at)
SELECT id,$4,$5,$6,(SELECT effective_at FROM history) FROM updated
@@ -487,9 +580,64 @@ WITH updated AS (
SELECT id||':'||revision::text,id,revision,document,clock_timestamp(),$4::text FROM updated RETURNING effective_at
), jobs AS (
INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt)
SELECT id,revision,'pending',0,now() FROM updated
ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision,state='pending',attempts=0,next_attempt=now(),last_error=NULL
SELECT id,revision,'draft',0,now() FROM updated
ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision
)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at)
SELECT id,$4::text,$5,$6,(SELECT effective_at FROM history) FROM updated
"#;
/// Compare desired configuration with the last verified publication, never provider GETs.
pub fn publication_changes(
document: &WorkspaceDocument,
baseline: Option<&WorkspaceDocument>,
) -> Result<Value> {
let desired = serde_json::to_value(document)?;
let old = serde_json::to_value(baseline)?;
let mut changes = vec![];
let schedule_context_changed = [
"timezone",
"people",
"holidays",
"absences",
"requests",
"horizon_days",
]
.iter()
.any(|key| desired[key] != old[key]);
for (collection, kind) in [
("schedules", "schedule"),
("escalations", "escalation"),
("routes", "route"),
] {
let empty = vec![];
let before = old[collection].as_array().unwrap_or(&empty);
let after = desired[collection].as_array().unwrap_or(&empty);
for resource in after {
let previous = before.iter().find(|r| r["id"] == resource["id"]);
let action = match previous {
None => "create",
Some(prior)
if prior != resource
|| (kind == "schedule" && schedule_context_changed)
|| desired["integration"] != old["integration"] =>
{
"update"
}
_ => "unchanged",
};
changes.push(
json!({"id":resource["id"],"name":resource["name"],"kind":kind,"action":action}),
);
}
for resource in before
.iter()
.filter(|r| !after.iter().any(|n| n["id"] == r["id"]))
{
changes.push(
json!({"id":resource["id"],"name":resource["name"],"kind":kind,"action":"remove"}),
);
}
}
Ok(json!(changes))
}
+32 -2
View File
@@ -69,9 +69,13 @@ impl App for MeguriApp {
"/api/workspaces/{workspace}/routes/order",
"/api/workspaces/{workspace}/absence-preview",
"/api/workspaces/{workspace}/absences",
"/api/workspaces/{workspace}/generation-export",
"/api/workspaces/{workspace}/generation-import-preview",
"/api/workspaces/{workspace}/generation-import",
"/api/workspaces/{workspace}/generation-preview",
"/api/workspaces/{workspace}/generation",
"/api/workspaces/{workspace}/member-permissions",
"/api/workspaces/{workspace}/generation-draft",
"/api/workspaces/{workspace}/requests",
"/api/workspaces/{workspace}/request-preview",
"/api/workspaces/{workspace}/requests/{request}",
@@ -484,10 +488,36 @@ async fn dispatch(request: Request) -> Result<Response> {
None
};
return response(
json!({"coverage":coverage,"workspace":record,"is_owner":db.owner(workspace,actor).await.is_ok(),"instance_admin":identity.user.instance_admin,"access":access,"permissions":grants,"members":db.members(workspace).await?,"directory":db.directory(workspace).await?,"sync":db.sync_status(workspace).await?,"audit":audit}),
json!({"generation_draft":db.shared_draft(workspace).await?,"coverage":coverage,"workspace":record,"is_owner":db.owner(workspace,actor).await.is_ok(),"instance_admin":identity.user.instance_admin,"access":access,"permissions":grants,"members":db.members(workspace).await?,"directory":db.directory(workspace).await?,"sync":db.sync_status(workspace).await?,"audit":audit}),
);
}
let action = parts.get(3).copied().unwrap_or_default();
if parts.len() == 4 && method == Method::PUT && action == "generation-draft" {
return response(
db.save_shared_draft(workspace, actor, revision(&body)?, body["payload"].clone())
.await?,
);
}
if parts.len() == 4 && method == Method::POST && action == "generation-export" {
return response(
crate::generation_service::export_config(&db, actor, workspace, decode(body)?).await?,
);
}
if parts.len() == 4
&& method == Method::POST
&& matches!(action, "generation-import-preview" | "generation-import")
{
return response(
crate::generation_service::import_config(
&db,
actor,
workspace,
decode(body)?,
action == "generation-import",
)
.await?,
);
}
if parts.len() == 4 && method == Method::POST && action == "generation-preview" {
return response(
crate::generation_service::preview(&db, actor, workspace, decode(body)?).await?,
@@ -718,7 +748,7 @@ async fn dispatch(request: Request) -> Result<Response> {
)?);
}
if action == "sync" && method == Method::POST {
db.retry(workspace, actor).await?;
db.retry(workspace, actor, revision(&body)?).await?;
return response(json!({"ok":true}));
}
Err(Error::NotFound)
+20 -6
View File
@@ -23,12 +23,16 @@ pub async fn run(db: Store) {
pub async fn tick(db: &Store) -> Result<()> {
let now = Utc::now();
let mut jobs = query!(SyncJobRow,$next_attempt<=now).all(&db.db).await?;
jobs.retain(|job| job.lease_until.is_none_or(|until| until <= now));
jobs.retain(|job| {
job.authorized_revision.is_some()
&& ["pending", "publishing", "failed", "conflict"].contains(&job.state.as_str())
&& job.lease_until.is_none_or(|until| until <= now)
});
jobs.sort_by_key(|job| job.next_attempt);
for job in jobs.into_iter().take(20) {
let token = Uuid::new_v4().to_string();
// Publication is bounded to four minutes; a five-minute lease survives a lost worker.
let claimed=db.db.raw_with("UPDATE meguri__sync_jobs SET lease_token=$1,lease_until=now()+interval '5 minutes',state='publishing' WHERE workspace_id=$2 AND (lease_until IS NULL OR lease_until<now()) AND next_attempt<=now()",&[&token,&job.workspace_id]).await?;
let claimed=db.db.raw_with("UPDATE meguri__sync_jobs SET lease_token=$1,lease_until=now()+interval '5 minutes',state='publishing' WHERE workspace_id=$2 AND (lease_until IS NULL OR lease_until<now()) AND next_attempt<=now() AND authorized_revision=$3 AND state IN ('pending','publishing','failed','conflict')",&[&token,&job.workspace_id,&job.authorized_revision]).await?;
if claimed.rows_affected().0 == 0 {
continue;
}
@@ -46,7 +50,7 @@ pub async fn tick(db: &Store) -> Result<()> {
});
match result {
Ok((revision, end)) => {
db.db.raw_with("WITH published AS (UPDATE meguri__sync_jobs SET published_revision=$1,state=CASE WHEN requested_revision=$1 THEN 'synced' ELSE 'pending' END,attempts=0,last_error=NULL,last_success=now(),window_end=$2,next_attempt=CASE WHEN requested_revision=$1 THEN now()+interval '15 minutes' ELSE now() END,lease_token=NULL,lease_until=NULL WHERE workspace_id=$3 AND lease_token=$4 RETURNING *) INSERT INTO meguri__audit_events(workspace_id,action,details,created_at) SELECT workspace_id,'publication.succeeded',$5,now() FROM published",&[&revision,&end,&job.workspace_id,&token,&json!({"revision":revision,"window_end":end}).to_string()]).await?;
db.db.raw_with("WITH published AS (UPDATE meguri__sync_jobs SET published_revision=$1,published_document=authorized_document,state='synced',attempts=0,last_error=NULL,last_success=now(),window_end=$2,next_attempt=now(),lease_token=NULL,lease_until=NULL WHERE workspace_id=$3 AND lease_token=$4 RETURNING *) INSERT INTO meguri__audit_events(workspace_id,action,details,created_at) SELECT workspace_id,'publication.succeeded',$5,now() FROM published",&[&revision,&end,&job.workspace_id,&token,&json!({"revision":revision,"window_end":end}).to_string()]).await?;
}
Err(error) => {
let conflict = matches!(error, Error::Conflict(_));
@@ -57,7 +61,7 @@ pub async fn tick(db: &Store) -> Result<()> {
(5_i64 * 2_i64.pow(job.attempts.clamp(0, 9) as u32)).min(1800)
};
let retry = Utc::now() + Duration::seconds(delay);
db.db.raw_with("UPDATE meguri__sync_jobs SET state=CASE WHEN requested_revision=$6 THEN $1 ELSE 'pending' END,last_error=CASE WHEN requested_revision=$6 THEN $2 ELSE NULL END,attempts=CASE WHEN requested_revision=$6 THEN attempts+1 ELSE 0 END,next_attempt=CASE WHEN requested_revision=$6 THEN $3 ELSE now() END,lease_token=NULL,lease_until=NULL WHERE workspace_id=$4 AND lease_token=$5",&[&state,&error.to_string(),&retry,&job.workspace_id,&token,&job.requested_revision]).await?;
db.db.raw_with("UPDATE meguri__sync_jobs SET state=$1,last_error=$2,attempts=attempts+1,next_attempt=$3,lease_token=NULL,lease_until=NULL WHERE workspace_id=$4 AND lease_token=$5",&[&state,&error.to_string(),&retry,&job.workspace_id,&token]).await?;
tracing::warn!(workspace=%job.workspace_id,%error,"publication did not complete");
}
}
@@ -66,10 +70,20 @@ pub async fn tick(db: &Store) -> Result<()> {
}
pub async fn publish_one(db: &Store, workspace: Uuid) -> Result<(i64, chrono::DateTime<Utc>)> {
let record = db.workspace(workspace).await?;
// Only the explicitly authorized snapshot may leave the application.
let id = workspace.to_string();
let job = query!(SyncJobRow, $workspace_id == id)
.get(&db.db)
.await?
.ok_or(Error::NotFound)?;
let record = crate::store::WorkspaceRecord {
id: workspace,
revision: job.authorized_revision.ok_or(Error::NotFound)?,
document: serde_json::from_str(job.authorized_document.as_deref().ok_or(Error::NotFound)?)?,
};
let now = Utc::now();
let (from, to) = engine::publication_window(&record.document, now)?;
let timeline = db.historical_timeline(&record, from, to).await?;
let timeline = engine::timeline(&record.document, from, to)?;
let plan = providers::compile(&record, timeline, now)?;
providers::publish(db, &record, &plan, record.document.overwrite_drift).await?;
Ok((record.revision, to))