//! Implicit scheduling (RFC 6638) between the principals of this server.
//!
//! `pimdav::itip` decides what a change sends to whom. This module finds the
//! recipients and their objects, and turns every message into writes that
//! commit together with the change itself. Nothing leaves the server: an
//! address outside it gets a delivery failure in its SCHEDULE-STATUS.
//!
//! Rooms and resources answer at once, from their own bookings. The outbox
//! answers free-busy requests from the recipients' calendars.

use std::collections::hash_map::Entry;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use chrono::{DateTime, Utc};
use percent_encoding::percent_decode_str;
use pimdav::calcard::icalendar::{
    ICalendar, ICalendarComponent, ICalendarComponentType, ICalendarProperty, ICalendarValue,
};
use pimdav::expand;
use pimdav::filter::TimeRange;
use pimdav::freebusy::{self, Period};
use pimdav::itip::{self, Message, Method, Role};
use pimdav::principal::UserType;
use pimdav::render;
use pimdav::xml::{CALDAV, el, hrefs, with_children};
use pimdav::zone::{self, Zone};
use sha2::{Digest, Sha256};
use tokio::sync::Mutex;
use xmltree::Element;

use super::pim::{
    INBOX, MAIL_DOMAIN, OUTBOX, collection_href, etag_of, local_part, mailto, need_privilege,
    principal_name, principal_uuid, seg,
};
use crate::api::common::blocking;
use crate::db::{PimKind, PimObject, PimOp, PimPrincipal};
use crate::error::{ApiError, AppState};

/// Held from reading a calendar object to committing the change, so that a
/// change and the writes it causes see a consistent store.
// ponytail: one lock for every object write. Per-UID locks if write
// throughput ever matters.
pub(crate) static LOCK: Mutex<()> = Mutex::const_new(());

/// The delivery status codes of RFC 6638, 3.2.9.
const DELIVERED: &str = "1.2";
/// An address in this server's domains that names no one.
const INVALID_USER: &str = "3.7";
/// An address outside this server: there is no iMIP to reach it.
const NO_ROUTE: &str = "5.2";
/// A reply the organizer's object had no room for in full.
const UNDELIVERED: &str = "5.1";
/// The recipient has no calendar for the component.
const REFUSED: &str = "5.3";
/// The recipient holds an object with this UID that is not its copy of the
/// sender's meeting (RFC 6638: no scheduling privileges).
const NO_AUTHORITY: &str = "3.8";

/// Recipients one free-busy request answers. Each costs an expansion of
/// their calendars.
const MAX_FREE_BUSY_ATTENDEES: usize = 100;

/// Who writes into a calendar, as far as scheduling cares.
pub(crate) struct Writer<'a> {
    pub owner: &'a PimPrincipal,
    /// May send messages as the owner (RFC 6638 `schedule-send`).
    pub may_schedule: bool,
    /// The writer's address when it is not the owner, for SENT-BY.
    pub sent_by: Option<String>,
    /// Stores the object without sending anything.
    pub quiet: bool,
}

impl Writer<'_> {
    /// The account `me` (principal id and name) writing into a calendar of
    /// `owner`.
    pub(crate) fn new<'a>(
        owner: &'a PimPrincipal,
        me: i64,
        name: &str,
        may_schedule: bool,
    ) -> Writer<'a> {
        Writer {
            owner,
            may_schedule,
            sent_by: (owner.id != me)
                .then(|| format!("mailto:{}", mailto(name, UserType::Individual))),
            quiet: false,
        }
    }

    /// The owner itself.
    pub(crate) fn owner(owner: &PimPrincipal) -> Writer<'_> {
        Writer {
            owner,
            may_schedule: true,
            sent_by: None,
            quiet: false,
        }
    }

    /// 403 `need-privileges` on the owner's outbox.
    fn refused(&self, privilege: &str) -> Element {
        let outbox = collection_href(&self.owner.name, PimKind::Calendar, OUTBOX, None);
        need_privilege(&outbox, CALDAV, privilege)
    }
}

/// Every principal, for mapping calendar user addresses.
#[derive(Clone)]
pub(crate) struct Directory(Vec<PimPrincipal>);

enum Recipient<'a> {
    Local(&'a PimPrincipal),
    Unknown,
    External,
}

fn found(p: Option<&PimPrincipal>) -> Recipient<'_> {
    match p {
        Some(p) => Recipient::Local(p),
        None => Recipient::Unknown,
    }
}

impl Directory {
    /// Disabled accounts included: they cannot log in, but their copies stay
    /// current.
    pub(crate) async fn load(state: &AppState) -> Result<Self, ApiError> {
        Ok(Directory(state.db.pim_principals(false).await?))
    }

    pub(crate) fn get(&self, id: i64) -> Option<&PimPrincipal> {
        self.0.iter().find(|p| p.id == id)
    }

    /// The forms `calendar-user-address-set` lists: the mailto address, the
    /// principal URL and the `urn:uuid:`. Compared without case.
    fn resolve(&self, addr: &str) -> Recipient<'_> {
        let addr = addr.trim();
        let lower = addr.to_ascii_lowercase();
        if let Some(rest) = lower.strip_prefix("mailto:") {
            let Some((local, domain)) = rest.rsplit_once('@') else {
                return Recipient::External;
            };
            let kind = match domain.strip_suffix(MAIL_DOMAIN) {
                Some("") => UserType::Individual,
                Some("rooms.") => UserType::Room,
                Some("resources.") => UserType::Resource,
                // The tombstone of a deleted principal.
                Some("deleted.") => return Recipient::Unknown,
                _ => return Recipient::External,
            };
            let name = percent_decode_str(local).decode_utf8_lossy();
            return found(
                self.0
                    .iter()
                    .find(|p| p.kind == kind && p.name.eq_ignore_ascii_case(&name)),
            );
        }
        if let Some(uuid) = lower.strip_prefix("urn:uuid:") {
            return found(self.0.iter().find(|p| principal_uuid(p.id) == uuid));
        }
        match principal_name(addr) {
            Some(name) => found(self.0.iter().find(|p| p.name.eq_ignore_ascii_case(&name))),
            None if lower.starts_with('/') || lower.starts_with("http") => Recipient::Unknown,
            None => Recipient::External,
        }
    }

    /// Whether an address names principal `id`.
    pub(crate) fn is(&self, id: i64) -> impl Fn(&str) -> bool + '_ {
        move |a: &str| matches!(self.resolve(a), Recipient::Local(p) if p.id == id)
    }
}

/// The writes that make other principals' objects forget `gone` before it
/// is deleted, after `retracted`, the writes of [`retract`]. Its addresses
/// become a tombstone in `deleted.` of the mail domain, which names no one,
/// so a later principal of the same name gets nothing meant for the old one.
/// Commit them together with the delete, holding [`LOCK`].
pub(crate) async fn forget(
    state: &AppState,
    gone: &PimPrincipal,
    mut retracted: Vec<PimOp>,
) -> Result<Vec<PimOp>, ApiError> {
    let dir = Directory(vec![gone.clone()]);
    let is_gone = dir.is(gone.id);
    let encoded = local_part(&gone.name);
    let tombstone = format!("mailto:{encoded}-{}@deleted.{MAIL_DOMAIN}", gone.id);
    let uuid = principal_uuid(gone.id);
    let needles = [gone.name.as_str(), encoded.as_str(), uuid.as_str()];
    let rewrite = |data: &[u8]| {
        itip::forget(&String::from_utf8_lossy(data), &is_gone, &tombstone).map(String::into_bytes)
    };
    let retracted_puts: HashSet<(i64, &str)> = retracted
        .iter()
        .filter_map(|op| match op {
            PimOp::Put {
                collection_id, obj, ..
            } => Some((*collection_id, obj.name.as_str())),
            _ => None,
        })
        .collect();
    let mut ops = Vec::new();
    for (collection_id, obj, data) in state.db.pim_objects_mentioning(gone.id, &needles).await? {
        if retracted_puts.contains(&(collection_id, obj.name.as_str())) {
            continue;
        }
        let Some(data) = rewrite(&data) else {
            continue;
        };
        ops.push(PimOp::Put {
            collection_id,
            obj: PimObject {
                etag: etag_of(&data),
                ..obj
            },
            data,
        });
    }
    for op in &mut retracted {
        if let PimOp::Put { obj, data, .. } | PimOp::Inbox { obj, data, .. } = op
            && let Some(new) = rewrite(data)
        {
            obj.etag = etag_of(&new);
            *data = new;
        }
    }
    // Last, because inbox writes drop the oldest messages; a rewrite after
    // them would bring a dropped one back.
    ops.extend(retracted);
    Ok(ops)
}

/// What a PUT of a calendar object stores, and what else it writes.
pub(crate) struct Stored {
    pub data: Vec<u8>,
    /// Whether `data` differs from the request body.
    pub changed: bool,
    pub schedule_tag: Option<String>,
    pub ops: Vec<PimOp>,
}

impl Stored {
    /// The write of this as `obj` into the collection, then the writes it
    /// causes. Sets the ETag and Schedule-Tag of `obj`.
    pub(crate) fn into_ops(self, collection_id: i64, obj: PimObject) -> Vec<PimOp> {
        let mut ops = vec![PimOp::Put {
            collection_id,
            obj: PimObject {
                etag: etag_of(&self.data),
                schedule_tag: self.schedule_tag,
                ..obj
            },
            data: self.data,
        }];
        ops.extend(self.ops);
        ops
    }
}

/// A PUT of `body` over `old` into collection `at.0` under the name `at.1`.
/// `Err` names a failed scheduling precondition.
pub(crate) async fn put(
    state: &AppState,
    dir: &Directory,
    w: &Writer<'_>,
    at: (i64, &str),
    old: Option<&[u8]>,
    body: &[u8],
) -> Result<Result<Stored, Element>, ApiError> {
    let parse = |b: &[u8]| render::parse(&String::from_utf8_lossy(b));
    let Some(sent) = parse(body) else {
        return Ok(Ok(unchanged(body, None)));
    };
    let owner = w.owner;
    let owns = dir.is(owner.id);
    let role = match itip::role(&sent, &owns) {
        Ok(r) => r,
        Err(refused) => return Ok(Err(refused.condition())),
    };
    if role != Role::None
        && let Some(holder) = elsewhere(state, owner, &sent, at).await?
    {
        return Ok(Err(holder));
    }
    let old = old.and_then(parse);
    let old_role = old.as_ref().and_then(|o| itip::role(o, &owns).ok());
    let now = Utc::now();
    let mut ops = Vec::new();

    let stored = match (role, old_role) {
        (Role::Organizer, _) => {
            let old = old.as_ref().filter(|_| old_role == Some(Role::Organizer));
            let (mut store, force) = itip::prepare(old, &sent, &owns);
            let mut messages = match w.quiet {
                true => Vec::new(),
                false => itip::messages(old, Some(&store), &owns, &force, now),
            };
            if !messages.is_empty() && !w.may_schedule {
                // A SEQUENCE bump alone is no invitation. The copies are not
                // reached, so the stored object keeps their SEQUENCE.
                let Some(same) = only_sequence(old, &store, &owns, &force, now) else {
                    return Ok(Err(w.refused("schedule-send-invite")));
                };
                store = same;
                messages.clear();
            }
            if !messages.is_empty() {
                itip::stamp_sender(&mut store, &owns, w.sent_by.as_deref());
                messages = itip::messages(old, Some(&store), &owns, &force, now);
            }
            // Rooms answer first, so the others' copies carry their answers.
            if answer_rooms(state, dir, owner, &mut store, &messages, &mut ops, now).await? {
                messages = itip::messages(old, Some(&store), &owns, &force, now);
            }
            let mut sent = Sent::new();
            for m in &messages {
                if let Some(status) = deliver(state, dir, owner, m, &mut ops, &mut sent).await? {
                    itip::set_attendee_status(&mut store, &m.to, status);
                }
            }
            store
        }
        (Role::Attendee, Some(Role::Attendee)) => {
            let old = old.as_ref().expect("an attendee role needs the old object");
            let (mut store, reply) = match itip::attend(old, sent.clone(), &owns, now) {
                Ok(v) => v,
                Err(refused) => return Ok(Err(refused.condition())),
            };
            if let Some(mut reply) = reply.filter(|_| !w.quiet) {
                if !w.may_schedule {
                    return Ok(Err(w.refused("schedule-send-reply")));
                }
                itip::stamp_sender(&mut store, &owns, w.sent_by.as_deref());
                itip::stamp_sender(Arc::make_mut(&mut reply.cal), &owns, w.sent_by.as_deref());
                let status = reply_to(state, dir, owner, &reply, &mut ops).await?;
                itip::set_organizer_status(&mut store, status);
            }
            store
        }
        // No longer a scheduling object, or a copy the attendee brings in
        // itself (RFC 6638, 3.2.2.2): stored as sent.
        (_, previous) => {
            if let Some(old) = old.as_ref().filter(|_| !w.quiet) {
                match removed(state, dir, w, old, previous, true).await? {
                    Ok(more) => ops.extend(more),
                    Err(refused) => return Ok(Err(refused)),
                }
            }
            let tag = (role != Role::None).then(|| etag_of(body));
            return Ok(Ok(Stored {
                ops,
                ..unchanged(body, tag)
            }));
        }
    };
    let changed = stored != sent;
    let data = match changed {
        true => render::write(&stored).into_bytes(),
        false => body.to_vec(),
    };
    Ok(Ok(Stored {
        schedule_tag: Some(etag_of(&data)),
        data,
        changed,
        ops,
    }))
}

/// `store` with the SEQUENCE values of `old`, if that leaves the attendees
/// nothing to hear.
fn only_sequence(
    old: Option<&ICalendar>,
    store: &ICalendar,
    owns: itip::Is,
    force: &[String],
    now: DateTime<Utc>,
) -> Option<ICalendar> {
    let old = old?;
    let key = |c: &ICalendarComponent| {
        c.property(&ICalendarProperty::RecurrenceId)
            .map(|e| format!("{:?}", e.values))
    };
    let sequences: HashMap<_, _> = old
        .components
        .iter()
        .filter(|c| c.has_property(&ICalendarProperty::Uid))
        .map(|c| (key(c), c.property(&ICalendarProperty::Sequence).cloned()))
        .collect();
    let mut same = store.clone();
    for c in same
        .components
        .iter_mut()
        .filter(|c| c.has_property(&ICalendarProperty::Uid))
    {
        let sequence = sequences.get(&key(c))?;
        c.entries.retain(|e| e.name != ICalendarProperty::Sequence);
        c.entries.extend(sequence.clone());
    }
    itip::messages(Some(old), Some(&same), owns, force, now)
        .is_empty()
        .then_some(same)
}

/// The resource name the server picks for an object it creates.
pub(crate) fn object_name(uid: &str, kind: PimKind) -> String {
    let hash = crate::hex(&Sha256::digest(uid));
    format!("{}.{}", &hash[..32], extension(kind))
}

/// The file extension of an object or a whole collection of `kind`.
pub(crate) fn extension(kind: PimKind) -> &'static str {
    match kind {
        PimKind::Calendar => "ics",
        PimKind::Addressbook => "vcf",
    }
}

pub(crate) fn unchanged(body: &[u8], schedule_tag: Option<String>) -> Stored {
    Stored {
        data: body.to_vec(),
        changed: false,
        schedule_tag,
        ops: Vec::new(),
    }
}

/// The writes a DELETE of `old` causes. `reply` is false for
/// `Schedule-Reply: F` (RFC 6638, 8.1). `Err` names a lacking privilege.
pub(crate) async fn delete(
    state: &AppState,
    dir: &Directory,
    w: &Writer<'_>,
    old: &[u8],
    reply: bool,
) -> Result<Result<Vec<PimOp>, Element>, ApiError> {
    let Some(old) = render::parse(&String::from_utf8_lossy(old)) else {
        return Ok(Ok(Vec::new()));
    };
    let role = itip::role(&old, &dir.is(w.owner.id)).ok();
    removed(state, dir, w, &old, role, reply).await
}

/// The writes that cancel or decline every object of the collections for
/// their attendees, as deleting each object would. Hold [`LOCK`].
pub(crate) async fn retract(
    state: &AppState,
    dir: &Directory,
    owner: &PimPrincipal,
    collection_ids: &[i64],
) -> Result<Result<Vec<PimOp>, Element>, ApiError> {
    let w = Writer::owner(owner);
    let mut ops = Vec::new();
    for &id in collection_ids {
        for (_, data) in state.db.pim_objects_with_data(id).await? {
            match delete(state, dir, &w, &data, true).await? {
                Ok(more) => ops.extend(more),
                Err(refused) => return Ok(Err(refused)),
            }
        }
    }
    Ok(Ok(ops))
}

/// An organizer object going away cancels; an attendee copy declines.
async fn removed(
    state: &AppState,
    dir: &Directory,
    w: &Writer<'_>,
    old: &ICalendar,
    role: Option<Role>,
    reply: bool,
) -> Result<Result<Vec<PimOp>, Element>, ApiError> {
    let owner = w.owner;
    let owns = dir.is(owner.id);
    let now = Utc::now();
    let mut old = old.clone();
    itip::stamp_sender(&mut old, &owns, w.sent_by.as_deref());
    let mut ops = Vec::new();
    match role {
        Some(Role::Organizer) => {
            let messages = itip::messages(Some(&old), None, &owns, &[], now);
            if !messages.is_empty() && !w.may_schedule {
                return Ok(Err(w.refused("schedule-send-invite")));
            }
            let mut sent = Sent::new();
            for m in &messages {
                deliver(state, dir, owner, m, &mut ops, &mut sent).await?;
            }
        }
        Some(Role::Attendee) if reply => {
            if let Some(m) = itip::decline(&old, &owns, now) {
                if !w.may_schedule {
                    return Ok(Err(w.refused("schedule-send-reply")));
                }
                reply_to(state, dir, owner, &m, &mut ops).await?;
            }
        }
        _ => {}
    }
    Ok(Ok(ops))
}

/// The resource of the owner that already schedules this UID elsewhere:
/// RFC 6638 allows one per UID (3.2.4.1).
async fn elsewhere(
    state: &AppState,
    owner: &PimPrincipal,
    cal: &ICalendar,
    (collection_id, name): (i64, &str),
) -> Result<Option<Element>, ApiError> {
    let Some((uid, _)) = identity(cal) else {
        return Ok(None);
    };
    let Some((holder_id, holder, _)) = state.db.pim_find_uid(owner.id, &uid).await? else {
        return Ok(None);
    };
    // A plain event with the same UID schedules nothing.
    if holder.schedule_tag.is_none() || (holder_id == collection_id && holder.name == name) {
        return Ok(None);
    }
    let slug = match state.db.pim_collection_by_id(holder_id).await? {
        Some((_, _, c)) => c.slug,
        None => return Ok(None),
    };
    let href = collection_href(&owner.name, PimKind::Calendar, &slug, None) + &seg(&holder.name);
    Ok(Some(with_children(
        el(CALDAV, "unique-scheduling-object-resource"),
        hrefs([href.as_str()]),
    )))
}

/// Rooms and resources answer their invitations at once: accepted where
/// free, declined where their bookings overlap. The answers go into the
/// organizer's `store` and inbox. Returns whether any room answered.
async fn answer_rooms(
    state: &AppState,
    dir: &Directory,
    organizer: &PimPrincipal,
    store: &mut ICalendar,
    messages: &[Message],
    ops: &mut Vec<PimOp>,
    now: DateTime<Utc>,
) -> Result<bool, ApiError> {
    let mut answered = false;
    for m in messages
        .iter()
        .filter(|m| m.method == Method::Request && !m.quiet)
    {
        let Recipient::Local(room) = dir.resolve(&m.to) else {
            continue;
        };
        let Some((uid, component)) = identity(&m.cal) else {
            continue;
        };
        if room.kind == UserType::Individual {
            continue;
        }
        let Some(calendar) = state.db.pim_calendar_for(room.id, &component).await? else {
            continue;
        };
        let Ok(copy) = copy_of(state, dir, room, organizer, &uid).await? else {
            continue;
        };
        let Some(received) = itip::receive(copy.as_ref().map(|(_, _, c)| c), m) else {
            continue;
        };
        let is_room = dir.is(room.id);
        let window = now..now + itip::answer_horizon(&received);
        let taken = busy_of(state, dir, room, &window, Some(&uid)).await?;
        let floating = floating_of(calendar.timezone.as_deref());
        let answer = itip::auto_answer(&received, &is_room, &taken, &window, &floating);
        let Ok((_, Some(reply))) = itip::attend(&received, answer, &is_room, now) else {
            continue;
        };
        answered |= itip::apply_reply(store, &reply.cal, &is_room).changed;
        ops.push(inbox(organizer, &reply, &component));
    }
    Ok(answered)
}

/// The busy time a principal shows to scheduling: its opaque calendars that
/// take events, never the inbox. Objects with UID `skip` do not count.
// ponytail: reads every object of those calendars per call. Keep busy
// periods in a table if principals grow large calendars.
pub(crate) async fn busy_of(
    state: &AppState,
    dir: &Directory,
    p: &PimPrincipal,
    range: &TimeRange,
    skip: Option<&str>,
) -> Result<Vec<Period>, ApiError> {
    let mut calendars = Vec::new();
    for c in state.db.pim_collections(p.id, PimKind::Calendar).await? {
        if c.slug == INBOX || c.transparent || !c.components.split(',').any(|x| x == "VEVENT") {
            continue;
        }
        let objects = state.db.pim_objects_with_data(c.id).await?;
        calendars.push((floating_of(c.timezone.as_deref()), objects));
    }
    let (dir, id, range, skip) = (dir.clone(), p.id, range.clone(), skip.map(str::to_string));
    blocking(move || -> Result<_, ApiError> {
        let me = dir.is(id);
        let mut busy = Vec::new();
        for (floating, objects) in calendars {
            for (o, data) in objects {
                if skip.as_deref().is_some_and(|u| u == o.uid) {
                    continue;
                }
                if let Ok(cal) = ICalendar::parse(String::from_utf8_lossy(&data).as_ref()) {
                    busy.extend(freebusy::busy(&cal, &range, &floating, Some(&me)));
                }
            }
        }
        Ok(freebusy::merge(busy))
    })
    .await
}

fn floating_of(timezone: Option<&str>) -> Zone {
    timezone.and_then(zone::from_vtimezone).unwrap_or(Zone::Utc)
}

/// Whether every instance of the calendar object `body` ended before `now`.
/// A component without a start, or a rule without COUNT that runs until
/// about now or later, never ends.
pub(crate) fn ended(body: &[u8], timezone: Option<&str>, now: DateTime<Utc>) -> bool {
    let Ok(cal) = ICalendar::parse(String::from_utf8_lossy(body).as_ref()) else {
        return false;
    };
    // A day of slack covers an UNTIL in a zone or floating.
    let soon = now.naive_utc() - chrono::TimeDelta::days(1);
    let open = cal.components.iter().any(|c| {
        let item = matches!(
            c.component_type,
            ICalendarComponentType::VEvent
                | ICalendarComponentType::VTodo
                | ICalendarComponentType::VJournal
        );
        let endless = c.properties(&ICalendarProperty::Rrule).any(|e| {
            matches!(e.values.first(), Some(ICalendarValue::RecurrenceRule(r))
                if r.count.is_none() && r.until.as_ref().and_then(|u| u.to_date_time())
                    .is_none_or(|u| u.date_time >= soon))
        });
        item && (endless || !c.has_property(&ICalendarProperty::Dtstart))
    });
    if open {
        return false;
    }
    let x = expand::expand(&cal, now..DateTime::<Utc>::MAX_UTC, floating_of(timezone));
    x.instances.is_empty() && !x.truncated
}

/// The answers to a free-busy request to an outbox (RFC 6638, 5.2): per
/// recipient its address, the REQUEST-STATUS and the VFREEBUSY reply.
pub(crate) async fn free_busy(
    state: &AppState,
    dir: &Directory,
    req: &freebusy::Request,
) -> Result<Vec<(String, &'static str, Option<String>)>, ApiError> {
    let now = Utc::now();
    let mut out = Vec::new();
    let mut known: HashMap<i64, Vec<Period>> = HashMap::new();
    for (i, to) in req.attendees.iter().enumerate() {
        let (status, data) = match dir.resolve(to) {
            _ if i >= MAX_FREE_BUSY_ATTENDEES => ("5.1;Service unavailable", None),
            // A disabled account still gets messages, but shows no busy time.
            Recipient::Local(p) if !p.active => ("3.7;Invalid calendar user", None),
            Recipient::Local(p) => {
                if let Entry::Vacant(e) = known.entry(p.id) {
                    e.insert(busy_of(state, dir, p, &req.range, None).await?);
                }
                (
                    "2.0;Success",
                    Some(freebusy::reply(&known[&p.id], req, to, now)),
                )
            }
            Recipient::Unknown => ("3.7;Invalid calendar user", None),
            Recipient::External => ("5.2;Invalid calendar service", None),
        };
        out.push((to.clone(), status, data));
    }
    Ok(out)
}

/// Statuses of the deliveries in one batch: principal, body, quiet.
type Sent = Vec<(i64, Arc<ICalendar>, bool, Option<&'static str>)>;

/// A REQUEST or CANCEL from `sender` into the recipient's calendar and
/// inbox. Returns the delivery status, `None` for the sender itself. A
/// principal named by two addresses that see the same message gets it once.
async fn deliver(
    state: &AppState,
    dir: &Directory,
    sender: &PimPrincipal,
    m: &Message,
    ops: &mut Vec<PimOp>,
    sent: &mut Sent,
) -> Result<Option<&'static str>, ApiError> {
    let p = match dir.resolve(&m.to) {
        Recipient::Local(p) if p.id == sender.id => return Ok(None),
        Recipient::Local(p) => p,
        Recipient::Unknown => return Ok(Some(INVALID_USER)),
        Recipient::External => return Ok(Some(NO_ROUTE)),
    };
    // A forced send must not reuse a quiet one.
    if let Some((.., status)) = sent
        .iter()
        .find(|(id, c, quiet, _)| *id == p.id && Arc::ptr_eq(c, &m.cal) && *quiet == m.quiet)
    {
        return Ok(*status);
    }
    let status = deliver_to(state, dir, sender, p, m, ops).await?;
    sent.push((p.id, m.cal.clone(), m.quiet, status));
    Ok(status)
}

async fn deliver_to(
    state: &AppState,
    dir: &Directory,
    sender: &PimPrincipal,
    p: &PimPrincipal,
    m: &Message,
    ops: &mut Vec<PimOp>,
) -> Result<Option<&'static str>, ApiError> {
    ensure(state, p).await?;
    let Some((uid, component)) = identity(&m.cal) else {
        return Ok(Some(REFUSED));
    };
    let Ok(copy) = copy_of(state, dir, p, sender, &uid).await? else {
        return Ok(Some(NO_AUTHORITY));
    };
    if let Some(next) = itip::receive(copy.as_ref().map(|(_, _, c)| c), m) {
        let data = render::write(&next).into_bytes();
        let etag = etag_of(&data);
        let (collection_id, name, schedule_tag) = match copy {
            // Only the others' answers changed: the attendee's pending edit
            // may still go through (RFC 6638, 3.2.10).
            Some((id, obj, _)) => (
                id,
                obj.name,
                if m.quiet {
                    obj.schedule_tag
                } else {
                    Some(etag.clone())
                },
            ),
            None => match state.db.pim_calendar_for(p.id, &component).await? {
                Some(c) => (
                    c.id,
                    object_name(&uid, PimKind::Calendar),
                    Some(etag.clone()),
                ),
                None => return Ok(Some(REFUSED)),
            },
        };
        ops.push(PimOp::Put {
            collection_id,
            obj: PimObject {
                name,
                uid,
                component: component.clone(),
                etag,
                schedule_tag,
                ..Default::default()
            },
            data,
        });
    }
    if !m.quiet {
        ops.push(inbox(p, m, &component));
    }
    Ok(Some(DELIVERED))
}

/// An attendee's REPLY: applied to the organizer's object, passed on to the
/// other attendees, and left in the organizer's inbox. Returns the delivery
/// status for the attendee's copy.
async fn reply_to(
    state: &AppState,
    dir: &Directory,
    attendee: &PimPrincipal,
    m: &Message,
    ops: &mut Vec<PimOp>,
) -> Result<&'static str, ApiError> {
    let organizer = match dir.resolve(&m.to) {
        Recipient::Local(p) => p,
        Recipient::Unknown => return Ok(INVALID_USER),
        Recipient::External => return Ok(NO_ROUTE),
    };
    let Some((uid, component)) = identity(&m.cal) else {
        return Ok(REFUSED);
    };
    // RFC 6638, 4.2: a reply to an object the organizer no longer has is
    // ignored.
    let Some((collection_id, obj, data)) = state.db.pim_find_uid(organizer.id, &uid).await? else {
        return Ok(NO_ROUTE);
    };
    let Some(before) = render::parse(&String::from_utf8_lossy(&data)) else {
        return Ok(NO_ROUTE);
    };
    // A UID alone proves nothing: only the organizer's own object takes it.
    if !matches!(
        itip::role(&before, &dir.is(organizer.id)),
        Ok(Role::Organizer)
    ) {
        return Ok(NO_AUTHORITY);
    }
    let replier = dir.is(attendee.id);
    if !matches!(itip::role(&before, &replier), Ok(Role::Attendee)) {
        return Ok(NO_AUTHORITY);
    }
    let mut after = before.clone();
    let applied = itip::apply_reply(&mut after, &m.cal, &replier);
    if applied.changed {
        let data = render::write(&after).into_bytes();
        ops.push(PimOp::Put {
            collection_id,
            obj: PimObject {
                etag: etag_of(&data),
                ..obj
            },
            data,
        });
        // The others learn the new answer without a new Schedule-Tag.
        let organizes = dir.is(organizer.id);
        let mut sent = Sent::new();
        for mut other in itip::messages(Some(&before), Some(&after), &organizes, &[], Utc::now()) {
            if other.method == Method::Request && !replier(&other.to) {
                other.quiet = true;
                deliver(state, dir, organizer, &other, ops, &mut sent).await?;
            }
        }
    }
    ops.push(inbox(organizer, m, &component));
    Ok(match applied.dropped {
        0 => DELIVERED,
        _ => UNDELIVERED,
    })
}

/// Whether `copy` is `p`'s attendee copy of a meeting `organizer` runs. A
/// message may change only that: anyone can pick any UID.
fn attends(dir: &Directory, copy: &ICalendar, p: &PimPrincipal, organizer: &PimPrincipal) -> bool {
    matches!(itip::role(copy, &dir.is(p.id)), Ok(Role::Attendee))
        && itip::organizer(copy).is_some_and(dir.is(organizer.id))
}

/// `p`'s copy of the meeting with `uid` and where it is stored. `Err` if
/// `p` holds that UID in an object the message may not touch.
async fn copy_of(
    state: &AppState,
    dir: &Directory,
    p: &PimPrincipal,
    organizer: &PimPrincipal,
    uid: &str,
) -> Result<Result<Option<(i64, PimObject, ICalendar)>, ()>, ApiError> {
    let Some((id, obj, data)) = state.db.pim_find_uid(p.id, uid).await? else {
        return Ok(Ok(None));
    };
    Ok(match render::parse(&String::from_utf8_lossy(&data)) {
        Some(c) if attends(dir, &c, p, organizer) => Ok(Some((id, obj, c))),
        _ => Err(()),
    })
}

async fn ensure(state: &AppState, p: &PimPrincipal) -> Result<(), ApiError> {
    match p.kind {
        UserType::Individual => state.db.pim_ensure_defaults(p.id).await?,
        _ => state.db.pim_ensure_inbox(p.id).await?,
    }
    Ok(())
}

fn inbox(p: &PimPrincipal, m: &Message, component: &str) -> PimOp {
    let data = render::write(&m.cal).into_bytes();
    let stamp = Utc::now().timestamp_nanos_opt().unwrap_or_default();
    let seed = format!(
        "{}\n{}\n{stamp}\n{:?}",
        m.to,
        String::from_utf8_lossy(&data),
        m.method
    );
    let name = format!("{}.ics", &crate::hex(&Sha256::digest(seed))[..32]);
    PimOp::Inbox {
        principal_id: p.id,
        obj: PimObject {
            // Inbox messages share UIDs, and the store keeps UIDs unique.
            uid: name.clone(),
            name,
            component: component.to_string(),
            etag: etag_of(&data),
            ..Default::default()
        },
        data,
    }
}

/// UID and component type of a scheduling message or object.
fn identity(cal: &ICalendar) -> Option<(String, String)> {
    let c = cal.components.iter().find(|c| {
        matches!(
            c.component_type,
            ICalendarComponentType::VEvent
                | ICalendarComponentType::VTodo
                | ICalendarComponentType::VJournal
        )
    })?;
    Some((c.uid()?.to_string(), c.component_type.as_str().to_string()))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_rule_running_on_has_not_ended_and_costs_nothing() {
        let body = |rule: &str| {
            format!(
                "BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//t//EN\r\nBEGIN:VEVENT\r\nUID:x\r\n\
                 DTSTAMP:20200101T000000Z\r\nDTSTART:20200101T100000Z\r\n{rule}END:VEVENT\r\nEND:VCALENDAR\r\n"
            )
        };
        let now = Utc::now();
        let started = std::time::Instant::now();
        let on = body("RRULE:FREQ=MINUTELY;UNTIL=99991231T000000Z\r\n");
        assert!(!ended(on.as_bytes(), None, now));
        assert!(started.elapsed() < std::time::Duration::from_millis(200));
        let over = body("RRULE:FREQ=DAILY;UNTIL=20200110T000000Z\r\n");
        assert!(ended(over.as_bytes(), None, now));
    }
}