use crate::{
    domain::*,
    error::{Error, Result, ensure},
    models::ProviderObjectRow,
    providers::Publication,
    store::{Store, WorkspaceRecord},
};
use chrono::{DateTime, Timelike, Utc};
use reqwest::{Client, Method, StatusCode};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use uuid::Uuid;

#[derive(Clone)]
pub struct Jsm {
    client: Client,
    base: String,
    email: String,
    token: String,
    team: String,
    users: BTreeMap<Uuid, String>,
    directory_base: Option<String>,
}

impl Jsm {
    pub fn from_environment(record: &WorkspaceRecord) -> Result<Self> {
        let Integration::Jsm {
            cloud_id,
            team_id,
            users,
            ..
        } = &record.document.integration
        else {
            return Err(Error::Validation(
                crate::i18n::text("err_jsm_is_not_configured_4e6594").into(),
            ));
        };
        let prefix = format!(
            "MEGURI_JSM_{}",
            record.id.simple().to_string().to_uppercase()
        );
        let email = std::env::var(format!("{prefix}_EMAIL")).map_err(|_| {
            Error::Provider(crate::i18n::message(
                "err_missing_prefix_email_deployment_secret_6e5753",
                &[("prefix", (prefix).to_string())],
            ))
        })?;
        let token = std::env::var(format!("{prefix}_TOKEN")).map_err(|_| {
            Error::Provider(crate::i18n::message(
                "err_missing_prefix_token_deployment_secret_1f0352",
                &[("prefix", (prefix).to_string())],
            ))
        })?;
        Ok(Self::new(
            format!("https://api.atlassian.com/jsm/ops/api/{cloud_id}/v1"),
            email,
            token,
            team_id.clone(),
            users.clone(),
        )?
        .with_directory_base(format!(
            "https://api.atlassian.com/ex/jira/{cloud_id}/rest/api/3"
        )))
    }
    pub fn new(
        base: String,
        email: String,
        token: String,
        team: String,
        users: BTreeMap<Uuid, String>,
    ) -> Result<Self> {
        let client = Client::builder()
            .timeout(std::time::Duration::from_secs(20))
            .redirect(reqwest::redirect::Policy::none())
            .build()
            .map_err(|_| {
                Error::Provider(
                    crate::i18n::text("err_could_not_initialize_jsm_client_09bad0").into(),
                )
            })?;
        Ok(Self {
            client,
            base,
            email,
            token,
            team,
            users,
            directory_base: None,
        })
    }

    /// Explicit injection for the fake HTTP server; production always uses Atlassian's host.
    pub fn with_directory_base(mut self, base: String) -> Self {
        self.directory_base = Some(base);
        self
    }

    async fn resolve_email(&self, email: &str) -> Result<String> {
        let base = self.directory_base.as_ref().ok_or_else(|| {
            Error::Provider(
                crate::i18n::text("err_configure_the_atlassian_directory_or_provide_acc_266b32")
                    .into(),
            )
        })?;
        let mut matches = BTreeSet::new();
        for offset in (0..1000).step_by(100) {
            let response = self
                .client
                .get(format!("{base}/user/search"))
                .basic_auth(&self.email, Some(&self.token))
                .query(&[
                    ("query", email.to_owned()),
                    ("startAt", offset.to_string()),
                    ("maxResults", "100".into()),
                ])
                .send()
                .await
                .map_err(|_| {
                    Error::Provider(
                        crate::i18n::text("err_atlassian_user_lookup_could_not_connect_6f312b")
                            .into(),
                    )
                })?;
            if !response.status().is_success() {
                return Err(Error::Provider(crate::i18n::message(
                    "err_atlassian_user_lookup_returned_http_p0_grant_bro_a0f887",
                    &[("p0", (response.status().as_u16()).to_string())],
                )));
            }
            let users: Vec<Value> = response.json().await.map_err(|_| {
                Error::Provider(
                    crate::i18n::text("err_invalid_atlassian_directory_response_1c1889").into(),
                )
            })?;
            for user in &users {
                if user["active"].as_bool() == Some(true)
                    && user["emailAddress"]
                        .as_str()
                        .is_some_and(|e| e.trim().eq_ignore_ascii_case(email))
                    && let Some(id) = user["accountId"].as_str().filter(|id| !id.is_empty())
                {
                    matches.insert(id.to_owned());
                }
            }
            if users.len() < 100 {
                break;
            }
            if offset == 900 {
                return Err(Error::Provider(
                    crate::i18n::text(
                        "err_atlassian_search_limit_reached_provide_an_explic_d1fe8a",
                    )
                    .into(),
                ));
            }
        }
        if matches.len() != 1 {
            return Err(Error::Provider(crate::i18n::message(
                "err_could_not_find_one_active_atlassian_account_with_39e9a9",
                &[("email", (email).to_string())],
            )));
        }
        Ok(matches.into_iter().next().unwrap())
    }

    async fn resolve_people(&self, record: &WorkspaceRecord, plan: &Publication) -> Result<Self> {
        let mut resolved = self.clone();
        let mut ids: BTreeSet<_> = plan
            .schedules
            .iter()
            .flat_map(|s| {
                s.periods.iter().map(|p| p.user_id).chain(
                    s.recurring
                        .iter()
                        .flat_map(|r| r.participants.iter().copied()),
                )
            })
            .collect();
        ids.extend(
            plan.escalations
                .iter()
                .flat_map(|e| e.steps.iter())
                .filter_map(|s| {
                    if let Recipient::User { id } = s.recipient {
                        Some(id)
                    } else {
                        None
                    }
                }),
        );
        for id in ids {
            if resolved.users.contains_key(&id) {
                continue;
            }
            let email = record
                .document
                .person(id)
                .and_then(|p| p.email.as_deref())
                .ok_or_else(|| {
                    Error::Provider(crate::i18n::message(
                        "err_person_id_needs_an_email_or_an_atlassian_account_360521",
                        &[("id", (id).to_string())],
                    ))
                })?;
            resolved.users.insert(id, self.resolve_email(email).await?);
        }
        Ok(resolved)
    }
    async fn request(
        &self,
        method: Method,
        path: &str,
        body: Option<&Value>,
    ) -> Result<Option<Value>> {
        let mut req = self
            .client
            .request(method.clone(), format!("{}{path}", self.base))
            .basic_auth(&self.email, Some(&self.token))
            .header("Accept", "application/json");
        if let Some(body) = body {
            req = req.json(body);
        }
        let response = req.send().await.map_err(|_| {
            Error::Provider(
                crate::i18n::text("err_jsm_request_timed_out_or_could_not_connect_5ce428").into(),
            )
        })?;
        if response.status() == StatusCode::NOT_FOUND {
            return Ok(None);
        }
        if !response.status().is_success() {
            return Err(Error::Provider(crate::i18n::message(
                "err_jsm_returned_http_p0_for_p1_check_credentials_pl_b31caf",
                &[
                    ("p0", (response.status().as_u16()).to_string()),
                    ("p1", (method).to_string()),
                ],
            )));
        }
        if response.status() == StatusCode::NO_CONTENT {
            return Ok(Some(Value::Null));
        }
        Ok(Some(response.json().await.map_err(|_| {
            Error::Provider(
                crate::i18n::text("err_jsm_returned_an_invalid_json_response_af8daf").into(),
            )
        })?))
    }
    async fn list(&self, path: &str) -> Result<Vec<Value>> {
        const PAGE_SIZE: usize = 50;
        let mut all = vec![];
        let mut offset = 0;
        loop {
            let sep = if path.contains('?') { '&' } else { '?' };
            let page = self
                .request(
                    Method::GET,
                    &format!("{path}{sep}offset={offset}&size={PAGE_SIZE}"),
                    None,
                )
                .await?
                .ok_or_else(|| {
                    Error::Provider(
                        crate::i18n::text("err_jsm_collection_was_not_found_60d46c").into(),
                    )
                })?;
            let values = page
                .get("values")
                .and_then(Value::as_array)
                .ok_or_else(|| {
                    Error::Provider(crate::i18n::text("err_jsm_collection_has_no_values").into())
                })?;
            let count = values.len();
            all.extend(values.iter().cloned());
            if page
                .pointer("/links/next")
                .and_then(Value::as_str)
                .is_none_or(str::is_empty)
                && count < PAGE_SIZE
            {
                break;
            }
            if count == 0 {
                break;
            }
            offset += count;
            ensure(
                offset <= 20000,
                crate::i18n::text("err_jsm_collection_exceeds_the_supported_pagination_a1136e"),
            )?;
        }
        Ok(all)
    }
    fn mapped(&self, id: Uuid) -> Result<String> {
        self.users.get(&id).cloned().ok_or_else(|| {
            Error::Validation(crate::i18n::message(
                "err_employee_id_has_no_jsm_account_mapping_7b6db1",
                &[("id", (id).to_string())],
            ))
        })
    }
    fn collection(&self, kind: &str) -> String {
        match kind {
            "schedule" => "/schedules".into(),
            "escalation" => format!("/teams/{}/escalations", self.team),
            _ => format!("/teams/{}/routing-rules", self.team),
        }
    }
    async fn read(&self, kind: &str, id: &str) -> Result<Option<Value>> {
        let Some(mut object) = self
            .request(
                Method::GET,
                &format!("{}/{id}", self.collection(kind)),
                None,
            )
            .await?
        else {
            return Ok(None);
        };
        if kind == "schedule" {
            object["rotations"] =
                Value::Array(self.list(&format!("/schedules/{id}/rotations")).await?);
        }
        Ok(Some(object))
    }
    #[allow(clippy::too_many_arguments)]
    async fn synchronize(
        &self,
        db: &Store,
        record: &WorkspaceRecord,
        kind: &str,
        key: &str,
        desired: &Value,
        previous: Option<&ProviderObjectRow>,
        overwrite: bool,
        as_of: DateTime<Utc>,
    ) -> Result<String> {
        let local_id = key
            .split_once(':')
            .and_then(|(_, id)| Uuid::parse_str(id).ok())
            .expect("provider object keys contain a resource UUID");
        // A deleted resource is absent desired state, not an edit to an existing
        // resource. Clear stale identity before discovery so a lost replacement
        // create response can be recovered by its stable ownership marker.
        let mut previous = previous;
        if let Some(old) = previous
            && self.read(kind, &old.remote_id).await?.is_none()
        {
            db.forget(record.id, key).await?;
            db.forget(record.id, &format!("pending:{key}")).await?;
            if kind == "schedule" {
                let prefix = format!("coverage:{local_id}:");
                for child in db
                    .objects(record.id)
                    .await?
                    .iter()
                    .filter(|r| r.object_key.starts_with(&prefix))
                {
                    db.forget(record.id, &child.object_key).await?;
                }
            }
            previous = None;
        }
        let marker = ownership_marker(record.id, local_id);
        let legacy_name = format!("meguri-{}-{}", record.id.simple(), local_id.simple());
        let existing_id = if let Some(previous) = previous {
            Some(previous.remote_id.clone())
        } else {
            let found: Vec<_> = self
                .list(&self.collection(kind))
                .await?
                .into_iter()
                .filter(|v| {
                    v["name"]
                        .as_str()
                        .is_some_and(|name| name.ends_with(&marker) || name == legacy_name)
                })
                .collect();
            ensure(
                found.len() <= 1,
                crate::i18n::text("err_multiple_jsm_resources_have_the_same_meguri_owne_889887"),
            )?;
            found
                .first()
                .and_then(|v| v["id"].as_str())
                .map(str::to_string)
        };
        let current = if let Some(id) = &existing_id {
            self.read(kind, id).await?
        } else {
            None
        };
        let expected = previous
            .map(|p| serde_json::from_str::<Value>(&p.content))
            .transpose()?;
        let pending_key = format!("pending:{key}");
        let pending =
            db.objects(record.id).await?.into_iter().find(|r| {
                r.object_key == pending_key && Some(&r.remote_id) == existing_id.as_ref()
            });
        let pending_expected = pending
            .as_ref()
            .map(|r| serde_json::from_str::<Value>(&r.content))
            .transpose()?;
        let matches_desired = current.as_ref().is_some_and(|c| equivalent(c, desired));
        if let Some(current) = &current {
            let known = expected
                .as_ref()
                .is_some_and(|last| equivalent(current, last))
                || pending_expected
                    .as_ref()
                    .is_some_and(|next| equivalent(current, next));
            let legacy_matches =
                if previous.is_none() && current["name"].as_str() == Some(&legacy_name) {
                    let mut legacy = desired.clone();
                    legacy["name"] = json!(legacy_name);
                    equivalent(current, &legacy)
                } else {
                    false
                };
            if !matches_desired && !known && !legacy_matches && !overwrite {
                return Err(Error::Conflict(crate::i18n::message(
                    "err_jsm_kind_changed_outside_meguri_publication_paus_ac8220",
                    &[("kind", (kind).to_string())],
                )));
            }
        }
        if kind == "schedule"
            && current.is_some()
            && let Some(id) = &existing_id
        {
            self.check_coverage(db, record.id, local_id, id, overwrite, as_of)
                .await?;
        }
        let id = if matches_desired {
            existing_id.unwrap()
        } else if current.is_some() {
            let id = existing_id.unwrap();
            let current = current.as_ref().unwrap();
            if kind == "schedule"
                && current["rotations"]
                    .as_array()
                    .is_some_and(|r| r.len() == 1)
                && desired["rotations"]
                    .as_array()
                    .is_some_and(|r| r.len() == 1)
            {
                let rotation_id = current["rotations"][0]["id"]
                    .as_str()
                    .filter(|v| safe_id(v))
                    .ok_or_else(|| {
                        Error::Provider(
                            crate::i18n::text("err_unexpected_jsm_resource_identifier_dedd70")
                                .into(),
                        )
                    })?;
                let mut metadata = desired.clone();
                metadata.as_object_mut().unwrap().remove("rotations");
                if !equivalent(current, &metadata) {
                    let mut intermediate = metadata.clone();
                    intermediate["rotations"] = current["rotations"].clone();
                    self.schedule_step(
                        db,
                        record.id,
                        key,
                        &id,
                        &format!("/schedules/{id}"),
                        &metadata,
                        &intermediate,
                    )
                    .await?;
                }
                if !equivalent(&current["rotations"][0], &desired["rotations"][0]) {
                    self.schedule_step(
                        db,
                        record.id,
                        key,
                        &id,
                        &format!("/schedules/{id}/rotations/{rotation_id}"),
                        &desired["rotations"][0],
                        desired,
                    )
                    .await?;
                }
            } else {
                self.request(
                    Method::PATCH,
                    &format!("{}/{id}", self.collection(kind)),
                    Some(desired),
                )
                .await?
                .ok_or_else(|| {
                    Error::Conflict(
                        crate::i18n::text("err_jsm_resource_disappeared_during_publication").into(),
                    )
                })?;
            }
            id
        } else {
            let response = self
                .request(Method::POST, &self.collection(kind), Some(desired))
                .await?
                .ok_or_else(|| {
                    Error::Provider(
                        crate::i18n::text("err_jsm_create_endpoint_was_not_found_1be86a").into(),
                    )
                })?;
            response["id"]
                .as_str()
                .ok_or_else(|| {
                    Error::Provider(
                        crate::i18n::text("err_jsm_created_resource_without_an_id").into(),
                    )
                })?
                .to_string()
        };
        ensure(
            safe_id(&id),
            crate::i18n::text("err_unexpected_jsm_resource_identifier_dedd70"),
        )?;
        let actual = self.read(kind, &id).await?.ok_or_else(|| {
            Error::Provider(
                crate::i18n::text("err_jsm_resource_is_missing_after_publication_3448ad").into(),
            )
        })?;
        if !equivalent(&actual, desired) {
            return Err(Error::Provider(crate::i18n::message(
                "err_jsm_kind_read_back_does_not_match_the_requested_63440e",
                &[("kind", (kind).to_string())],
            )));
        }
        db.checkpoint(record.id, key, &id, desired).await?;
        db.forget(record.id, &pending_key).await?;
        Ok(id)
    }
    #[allow(clippy::too_many_arguments)]
    async fn schedule_step(
        &self,
        db: &Store,
        workspace: Uuid,
        key: &str,
        id: &str,
        path: &str,
        body: &Value,
        expected: &Value,
    ) -> Result<()> {
        let pending = format!("pending:{key}");
        db.checkpoint(workspace, &pending, id, expected).await?;
        self.request(Method::PATCH, path, Some(body))
            .await?
            .ok_or_else(|| {
                Error::Conflict(
                    crate::i18n::text("err_jsm_resource_disappeared_during_publication").into(),
                )
            })?;
        let actual = self.read("schedule", id).await?.ok_or_else(|| {
            Error::Provider(
                crate::i18n::text("err_jsm_resource_is_missing_after_publication_3448ad").into(),
            )
        })?;
        ensure(
            equivalent(&actual, expected),
            crate::i18n::message(
                "err_jsm_kind_read_back_does_not_match_the_requested_63440e",
                &[("kind", "schedule".into())],
            ),
        )?;
        db.checkpoint(workspace, key, id, expected).await?;
        db.forget(workspace, &pending).await?;
        Ok(())
    }
    fn schedule_payloads(
        &self,
        schedule: &super::PublishedSchedule,
        as_of: DateTime<Utc>,
    ) -> Result<(Vec<Value>, Vec<Value>)> {
        if let Some(rotation) = &schedule.recurring {
            let zone = schedule.timezone.parse::<chrono_tz::Tz>().map_err(|_| {
                Error::Provider(crate::i18n::text("jsm_invalid_coverage_interval").into())
            })?;
            if half_hour_floor(rotation.start, zone) == rotation.start
                && half_hour_floor(rotation.end, zone) == rotation.end
            {
                let participants = rotation
                    .participants
                    .iter()
                    .map(|id| self.mapped(*id).map(|id| json!({"type":"user","id":id})))
                    .collect::<Result<Vec<_>>>()?;
                let (kind, length) = if rotation.shift_days.is_multiple_of(7) {
                    ("weekly", rotation.shift_days / 7)
                } else {
                    ("daily", rotation.shift_days)
                };
                return Ok((
                    vec![
                        json!({"name":schedule.name,"startDate":rotation.start,"endDate":rotation.end,"type":kind,"length":length,"participants":participants}),
                    ],
                    vec![],
                ));
            }
        }
        if !needs_precise_coverage(schedule)? {
            let zone = schedule.timezone.parse::<chrono_tz::Tz>().map_err(|_| {
                Error::Provider(crate::i18n::text("jsm_invalid_coverage_interval").into())
            })?;
            let rotations = schedule.periods.iter().map(|p| Ok(json!({
                "name":if schedule.periods.len() == 1 { schedule.name.clone() } else {
                    format!("{} · {}", schedule.name, p.start.with_timezone(&zone).format("%d-%m-%Y %H:%M %:z"))
                }, "startDate":p.start, "endDate":p.end,
                "type":"daily", "length":1, "participants":[{"type":"user","id":self.mapped(p.user_id)?}]
            }))).collect::<Result<Vec<_>>>()?;
            return Ok((rotations, vec![]));
        }
        let zone = schedule.timezone.parse::<chrono_tz::Tz>().map_err(|_| {
            Error::Provider(crate::i18n::text("jsm_invalid_coverage_interval").into())
        })?;
        let current = schedule
            .periods
            .iter()
            .find(|p| p.start <= as_of && as_of < p.end);
        let base_person = current.map(|p| self.mapped(p.user_id)).transpose()?;
        let first = current.unwrap_or(&schedule.periods[0]);
        let last_end = schedule.periods.last().unwrap().end;
        let start = half_hour_floor(first.start, zone);
        let floor_end = half_hour_floor(last_end, zone);
        let end = if floor_end == last_end {
            floor_end
        } else {
            floor_end + chrono::Duration::minutes(30)
        };
        let participant = base_person.as_ref().map_or_else(
            || json!({"type":"noone"}),
            |id| json!({"type":"user","id":id}),
        );
        let rotations = vec![
            json!({"name":schedule.name, "startDate":start, "endDate":end,
            "type":"daily", "length":1, "participants":[participant]}),
        ];
        let mut coverage = vec![];
        // JSM rejects new overrides starting in the past. The already effective
        // assignee is the base; overrides describe only future transitions.
        for p in schedule.periods.iter().filter(|p| p.start > as_of) {
            coverage.push(
                json!({"responder":{"type":"user","id":self.mapped(p.user_id)?},
                "startDate":milliseconds(p.start), "endDate":milliseconds(p.end)}),
            );
        }
        if base_person.is_some() {
            for pair in schedule.periods.windows(2) {
                if pair[0].end > as_of && pair[0].end < pair[1].start {
                    coverage.push(json!({"responder":{"type":"noone"},
                        "startDate":milliseconds(pair[0].end), "endDate":milliseconds(pair[1].start)}));
                }
            }
            if last_end > as_of && last_end < end {
                coverage.push(json!({"responder":{"type":"noone"},
                    "startDate":milliseconds(last_end), "endDate":end}));
            }
        }
        Ok((rotations, coverage))
    }

    // Empty remote IDs are durable create intents, never verified publications.
    async fn check_coverage(
        &self,
        db: &Store,
        workspace: Uuid,
        schedule: Uuid,
        remote: &str,
        overwrite: bool,
        _as_of: DateTime<Utc>,
    ) -> Result<()> {
        let prefix = format!("coverage:{schedule}:");
        let rows = db.objects(workspace).await?;
        let mut owned = vec![];
        for row in rows.iter().filter(|r| r.object_key.starts_with(&prefix)) {
            let content: Value = serde_json::from_str(&row.content)?;
            if content["scheduleId"].as_str() == Some(remote) {
                owned.push((row, content));
            }
        }
        let items = self.list(&format!("/schedules/{remote}/overrides")).await?;
        for item in &items {
            let alias = coverage_alias(item)?;
            if let Some((_, content)) = owned.iter().find(|(row, _)| row.remote_id == alias) {
                if !overwrite && !equivalent(item, &content["payload"]) {
                    return Err(Error::Conflict(
                        crate::i18n::text("jsm_coverage_conflict").into(),
                    ));
                }
                continue;
            }
            let pending: Vec<_> = owned
                .iter()
                .filter(|(row, content)| {
                    row.remote_id.is_empty() && equivalent(item, &content["payload"])
                })
                .collect();
            ensure(
                pending.len() <= 1,
                crate::i18n::text("jsm_coverage_conflict"),
            )?;
            if let Some((row, content)) = pending.first() {
                ensure(
                    items
                        .iter()
                        .filter(|candidate| equivalent(candidate, &content["payload"]))
                        .count()
                        == 1,
                    crate::i18n::text("jsm_coverage_conflict"),
                )?;
                db.checkpoint(workspace, &row.object_key, alias, content)
                    .await?;
            } else if overwrite {
                self.request(
                    Method::DELETE,
                    &format!("/schedules/{remote}/overrides/{alias}"),
                    None,
                )
                .await?;
            } else {
                return Err(Error::Conflict(
                    crate::i18n::text("jsm_coverage_conflict").into(),
                ));
            }
        }
        Ok(())
    }

    async fn publish_coverage(
        &self,
        db: &Store,
        workspace: Uuid,
        schedule: Uuid,
        remote: &str,
        desired: &[Value],
    ) -> Result<()> {
        let prefix = format!("coverage:{schedule}:");
        let rows = db.objects(workspace).await?;
        let items = self.list(&format!("/schedules/{remote}/overrides")).await?;
        let mut used = BTreeSet::new();
        for payload in desired {
            let digest = Sha256::digest(serde_json::to_vec(payload)?);
            let key = format!("{prefix}{digest:x}");
            used.insert(key.clone());
            let content = json!({"scheduleId":remote,"payload":payload});
            let previous = rows.iter().find(|r| r.object_key == key);
            let alias = previous
                .filter(|r| !r.remote_id.is_empty())
                .filter(|r| {
                    serde_json::from_str::<Value>(&r.content)
                        .ok()
                        .is_some_and(|v| v["scheduleId"].as_str() == Some(remote))
                })
                .map(|r| r.remote_id.as_str());
            let existing = alias.and_then(|alias| {
                items
                    .iter()
                    .find(|item| item["alias"].as_str() == Some(alias))
            });
            let alias = if let (Some(alias), Some(existing)) = (alias, existing) {
                if !equivalent(existing, payload) {
                    self.request(
                        Method::PUT,
                        &format!("/schedules/{remote}/overrides/{alias}"),
                        Some(payload),
                    )
                    .await?;
                }
                alias.to_owned()
            } else {
                db.checkpoint(workspace, &key, "", &content).await?;
                let created = self
                    .request(
                        Method::POST,
                        &format!("/schedules/{remote}/overrides"),
                        Some(payload),
                    )
                    .await?
                    .ok_or_else(|| {
                        Error::Provider(crate::i18n::text("jsm_coverage_response_invalid").into())
                    })?;
                coverage_alias(&created)?.to_owned()
            };
            let actual = self
                .request(
                    Method::GET,
                    &format!("/schedules/{remote}/overrides/{alias}"),
                    None,
                )
                .await?
                .ok_or_else(|| {
                    Error::Provider(crate::i18n::text("jsm_coverage_response_invalid").into())
                })?;
            ensure(
                equivalent(&actual, payload),
                crate::i18n::text("jsm_coverage_response_invalid"),
            )?;
            db.checkpoint(workspace, &key, &alias, &content).await?;
        }
        // Keep previous effective spans until all replacement writes are verified.
        for row in rows
            .iter()
            .filter(|r| r.object_key.starts_with(&prefix) && !used.contains(&r.object_key))
        {
            let content: Value = serde_json::from_str(&row.content)?;
            if !row.remote_id.is_empty() {
                let parent = content["scheduleId"]
                    .as_str()
                    .filter(|v| safe_id(v))
                    .ok_or_else(|| {
                        Error::Provider(crate::i18n::text("jsm_coverage_response_invalid").into())
                    })?;
                ensure(
                    safe_id(&row.remote_id),
                    crate::i18n::text("jsm_coverage_response_invalid"),
                )?;
                self.request(
                    Method::DELETE,
                    &format!("/schedules/{parent}/overrides/{}", row.remote_id),
                    None,
                )
                .await?;
            }
            db.forget(workspace, &row.object_key).await?;
        }
        Ok(())
    }

    pub async fn publish(
        &self,
        db: &Store,
        record: &WorkspaceRecord,
        plan: &Publication,
        overwrite: bool,
    ) -> Result<()> {
        for schedule in &plan.schedules {
            validate_precise_intervals(schedule)?;
        }
        // Preflight every recipient before creating or modifying any external object.
        self.resolve_people(record, plan)
            .await?
            .publish_resolved(db, record, plan, overwrite)
            .await
    }

    async fn publish_resolved(
        &self,
        db: &Store,
        record: &WorkspaceRecord,
        plan: &Publication,
        overwrite: bool,
    ) -> Result<()> {
        let objects = db.objects(record.id).await?;
        let previous: BTreeMap<_, _> = objects.iter().map(|p| (p.object_key.as_str(), p)).collect();
        let mut used = BTreeSet::new();
        let mut schedules = BTreeMap::new();
        let mut policies = BTreeMap::new();
        for s in &plan.schedules {
            let key = format!("schedule:{}", s.id);
            used.insert(key.clone());
            let (rotations, coverage) = self.schedule_payloads(s, plan.as_of)?;
            let body = json!({"name":owned_name(record.id,s.id,&s.name),"description":crate::i18n::message("err_managed_by_meguri_p0_e8ef37", &[("p0", (s.name).to_string())]),"teamId":self.team,"timezone":s.timezone,"enabled":true,"rotations":rotations});
            let remote = self
                .synchronize(
                    db,
                    record,
                    "schedule",
                    &key,
                    &body,
                    previous.get(key.as_str()).copied(),
                    overwrite,
                    plan.as_of,
                )
                .await?;
            self.publish_coverage(db, record.id, s.id, &remote, &coverage)
                .await?;
            schedules.insert(s.id, remote);
        }
        for e in &plan.escalations {
            let key = format!("escalation:{}", e.id);
            used.insert(key.clone());
            let mut rules = vec![];
            for step in &e.steps {
                let (kind, id, notify) = match step.recipient {
                    Recipient::Schedule { id } => (
                        "schedule",
                        schedules.get(&id).cloned().ok_or_else(|| {
                            Error::Validation(
                                crate::i18n::text(
                                    "err_escalation_references_an_empty_schedule_2a0f3c",
                                )
                                .into(),
                            )
                        })?,
                        "default",
                    ),
                    Recipient::User { id } => ("user", self.mapped(id)?, "default"),
                    Recipient::TeamAll => ("team", self.team.clone(), "all"),
                    Recipient::TeamRandom => ("team", self.team.clone(), "random"),
                };
                rules.push(json!({"condition":step.condition,"notifyType":notify,"delay":step.delay_minutes,"recipient":{"type":kind,"id":id}}));
            }
            let body = json!({"name":owned_name(record.id,e.id,&e.name),"description":crate::i18n::message("err_managed_by_meguri_p0_e8ef37", &[("p0", (e.name).to_string())]),"enabled":true,"rules":rules,"repeat":{"count":e.repeat.count.max(1),"waitInterval":if e.repeat.count==0 {0}else{e.repeat.wait_minutes},"resetRecipientStates":e.repeat.reset_recipient_states,"closeAlertAfterAll":e.repeat.close_alert_after_all}});
            let remote = self
                .synchronize(
                    db,
                    record,
                    "escalation",
                    &key,
                    &body,
                    previous.get(key.as_str()).copied(),
                    overwrite,
                    plan.as_of,
                )
                .await?;
            policies.insert(e.id, remote);
        }
        let mut routes = plan.routes.clone();
        routes.sort_by_key(|r| r.order);
        for (order, route) in routes.iter().enumerate() {
            let key = format!("route:{}", route.id);
            used.insert(key.clone());
            let notify = match route.escalation_id {
                Some(id) => {
                    json!({"type":"escalation","id":policies.get(&id).ok_or_else(||Error::Validation(crate::i18n::text("err_missing_escalation_mapping_3fd412").into()))?})
                }
                None => json!({"type":"none"}),
            };
            let conditions:Vec<_>=route.conditions.iter().map(|c|json!({"field":c.field,"operation":c.operation,"expectedValue":c.value,"not":c.negate})).collect();
            let criteria = if conditions.is_empty() {
                json!({"type":"match-all"})
            } else {
                json!({"type":"match-all-conditions","conditions":conditions})
            };
            let mut body = json!({"name":owned_name(record.id,route.id,&route.name),"order":order,"timezone":record.document.timezone,"criteria":criteria,"notify":notify});
            if let Some(w) = &route.window {
                body["timezone"] = json!(w.timezone);
                let days = [
                    "monday",
                    "tuesday",
                    "wednesday",
                    "thursday",
                    "friday",
                    "saturday",
                    "sunday",
                ];
                let restrictions:Vec<_>=w.weekdays.iter().map(|d|json!({"startDay":days[usize::from(*d)-1],"endDay":days[usize::from(*d)-1],"startHour":w.start.hour(),"startMin":w.start.minute(),"endHour":w.end.hour(),"endMin":w.end.minute()})).collect();
                body["timeRestriction"] =
                    json!({"type":"weekday-and-time-of-day","restrictions":restrictions});
            } else {
                body["timeRestriction"] = Value::Null;
            }
            self.synchronize(
                db,
                record,
                "route",
                &key,
                &body,
                previous.get(key.as_str()).copied(),
                overwrite,
                plan.as_of,
            )
            .await?;
        }
        // Delete only checkpointed, owned resources, in reverse dependency order.
        for kind in ["route", "escalation", "schedule"] {
            for old in objects.iter().filter(|o| {
                o.object_key.starts_with(&format!("{kind}:")) && !used.contains(&o.object_key)
            }) {
                if let Some(current) = self.read(kind, &old.remote_id).await? {
                    let expected = serde_json::from_str(&old.content)?;
                    if !equivalent(&current, &expected) && !overwrite {
                        return Err(Error::Conflict(
                            crate::i18n::text(
                                "err_an_obsolete_jsm_resource_changed_externally_dele_a26bda",
                            )
                            .into(),
                        ));
                    }
                    if kind == "schedule" {
                        let local_id = old
                            .object_key
                            .split_once(':')
                            .and_then(|(_, id)| Uuid::parse_str(id).ok())
                            .unwrap();
                        self.check_coverage(
                            db,
                            record.id,
                            local_id,
                            &old.remote_id,
                            overwrite,
                            plan.as_of,
                        )
                        .await?;
                    }
                    self.request(
                        Method::DELETE,
                        &format!("{}/{}", self.collection(kind), old.remote_id),
                        None,
                    )
                    .await?;
                }
                db.forget(record.id, &old.object_key).await?;
                db.forget(record.id, &format!("pending:{}", old.object_key))
                    .await?;
                if kind == "schedule" {
                    let prefix = format!(
                        "coverage:{}:",
                        old.object_key.strip_prefix("schedule:").unwrap()
                    );
                    for child in db
                        .objects(record.id)
                        .await?
                        .iter()
                        .filter(|o| o.object_key.starts_with(&prefix))
                    {
                        db.forget(record.id, &child.object_key).await?;
                    }
                }
            }
        }
        Ok(())
    }
}

fn milliseconds(value: DateTime<Utc>) -> DateTime<Utc> {
    DateTime::from_timestamp_millis(value.timestamp_millis()).unwrap()
}

fn half_hour_floor(value: DateTime<Utc>, zone: chrono_tz::Tz) -> DateTime<Utc> {
    let local = value.with_timezone(&zone);
    value
        - chrono::Duration::minutes(i64::from(local.minute() % 30))
        - chrono::Duration::seconds(i64::from(local.second()))
        - chrono::Duration::nanoseconds(i64::from(local.nanosecond()))
}

fn needs_precise_coverage(schedule: &super::PublishedSchedule) -> Result<bool> {
    let zone = schedule
        .timezone
        .parse::<chrono_tz::Tz>()
        .map_err(|_| Error::Provider(crate::i18n::text("jsm_invalid_coverage_interval").into()))?;
    Ok(schedule.periods.iter().any(|p| {
        [p.start, p.end].iter().any(|date| {
            let local = date.with_timezone(&zone);
            local.minute() % 30 != 0 || local.second() != 0 || local.nanosecond() != 0
        })
    }))
}

fn validate_precise_intervals(schedule: &super::PublishedSchedule) -> Result<()> {
    ensure(
        !schedule.periods.is_empty()
            && schedule
                .periods
                .iter()
                .all(|p| milliseconds(p.start) < milliseconds(p.end)),
        crate::i18n::text("jsm_invalid_coverage_interval"),
    )
}

fn coverage_alias(value: &Value) -> Result<&str> {
    value["alias"]
        .as_str()
        .filter(|v| safe_id(v))
        .ok_or_else(|| Error::Provider(crate::i18n::text("jsm_coverage_response_invalid").into()))
}

fn ownership_marker(workspace: Uuid, id: Uuid) -> String {
    let mut digest = Sha256::new();
    digest.update(workspace.as_bytes());
    digest.update(id.as_bytes());
    let digest = digest.finalize();
    format!(
        "[meguri:{:032x}]",
        u128::from_be_bytes(digest[..16].try_into().unwrap())
    )
}

fn owned_name(workspace: Uuid, id: Uuid, name: &str) -> String {
    format!("{name} {}", ownership_marker(workspace, id))
}
fn safe_id(id: &str) -> bool {
    !id.is_empty()
        && id.len() <= 200
        && id
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || "-_".contains(c))
}

/// Ignore server-generated metadata, while comparing every managed field and array entry.
pub fn equivalent(actual: &Value, expected: &Value) -> bool {
    match expected {
        Value::Object(expected) => {
            // A restriction added to a finite rotation changes coverage, not metadata.
            if expected.contains_key("participants")
                && expected.contains_key("startDate")
                && !expected.contains_key("timeRestriction")
                && actual.get("timeRestriction").is_some_and(|v| !v.is_null())
            {
                return false;
            }
            if expected.contains_key("responder")
                && expected.contains_key("startDate")
                && !expected.contains_key("rotationIds")
                && actual.get("rotationIds").is_some_and(|ids| {
                    !ids.is_null() && ids.as_array().is_none_or(|ids| !ids.is_empty())
                })
            {
                return false;
            }
            expected.iter().all(|(k, v)| {
                if k == "repeat"
                    && expected.contains_key("rules")
                    && actual.get(k).is_none()
                    && v.get("waitInterval").and_then(Value::as_u64) == Some(0)
                {
                    // JSM omits disabled repeats in escalation GET responses.
                    true
                } else if k == "rotations" {
                    let Some(actual) = actual.get(k).and_then(Value::as_array) else {
                        return false;
                    };
                    let Some(expected) = v.as_array() else {
                        return false;
                    };
                    let mut remaining: Vec<_> = actual.iter().collect();
                    for expected in expected {
                        let Some(index) = remaining
                            .iter()
                            .position(|actual| equivalent(actual, expected))
                        else {
                            return false;
                        };
                        remaining.remove(index);
                    }
                    remaining.is_empty()
                } else {
                    equivalent(actual.get(k).unwrap_or(&Value::Null), v)
                }
            })
        }
        Value::Array(expected) => actual.as_array().is_some_and(|a| {
            a.len() == expected.len() && a.iter().zip(expected).all(|(a, e)| equivalent(a, e))
        }),
        Value::String(expected) => {
            if let (Ok(a), Ok(b)) = (
                actual.as_str().unwrap_or_default().parse::<DateTime<Utc>>(),
                expected.parse::<DateTime<Utc>>(),
            ) {
                a == b
            } else {
                actual.as_str() == Some(expected)
            }
        }
        _ => actual == expected,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn omitted_escalation_repeat_only_matches_disabled_repeats() {
        let mut expected = json!({"rules":[{"delay":0}],"repeat":{"count":1,"waitInterval":0,"resetRecipientStates":false,"closeAlertAfterAll":false}});
        let mut remote = json!({"rules":[{"delay":0}]});
        assert!(equivalent(&remote, &expected));
        remote["rules"][0]["delay"] = json!(10);
        assert!(!equivalent(&remote, &expected));
        remote["rules"][0]["delay"] = json!(0);
        expected["repeat"]["waitInterval"] = json!(5);
        assert!(!equivalent(&remote, &expected));
        remote["repeat"] = expected["repeat"].clone();
        assert!(equivalent(&remote, &expected));
        for (field, changed) in [
            ("count", json!(2)),
            ("waitInterval", json!(10)),
            ("resetRecipientStates", json!(true)),
            ("closeAlertAfterAll", json!(true)),
        ] {
            let mut changed_remote = remote.clone();
            changed_remote["repeat"][field] = changed;
            assert!(!equivalent(&changed_remote, &expected));
        }
        expected["repeat"]["waitInterval"] = json!(0);
        assert!(!equivalent(&remote, &expected));
        remote["repeat"] = Value::Null;
        assert!(!equivalent(&remote, &expected));
    }

    #[test]
    fn normalization_ignores_ids_but_detects_extra_rotations_and_changed_owners() {
        let expected = json!({"rotations":[{"startDate":"2026-09-23T08:00:00Z","participants":[{"id":"alice","type":"user"}]}]});
        let mut remote = json!({"id":"remote","rotations":[{"id":"rotation","startDate":"2026-09-23T08:00:00.000Z","participants":[{"id":"alice","type":"user"}]}]});
        assert!(equivalent(&remote, &expected));
        remote["rotations"][0]["timeRestriction"] =
            json!({"type":"time-of-day","restriction":{"startHour":9,"endHour":17}});
        assert!(!equivalent(&remote, &expected));
        remote["rotations"][0]
            .as_object_mut()
            .unwrap()
            .remove("timeRestriction");
        remote["rotations"][0]["participants"][0]["id"] = json!("bob");
        assert!(!equivalent(&remote, &expected));
        remote["rotations"].as_array_mut().unwrap().clear();
        assert!(!equivalent(&remote, &expected));
    }
}