Masked email: x:MaskedEmail/changes and a state in /get (fork additions)

The fork's per-account change log answers /changes, collapsing a mask
created and destroyed in the window. /changes on a registry type needs that
type's get permission.

crates/features/src/masked_email/data.rs

@@ -196,15 +196,14 @@ fn change_key(account_id: u32, change_id: u64) -> ValueKey<ValueClass> {
     key(KIND_CHANGE, &rest)
 }
 
-/// The account's changes after `since`, oldest first, and the latest change
-/// id (`since` when there are none).
+/// The account's changes after `since`, oldest first, each with its change
+/// id.
 pub async fn changes_since(
     data: &Store,
     account_id: u32,
     since: u64,
-) -> trc::Result<(Vec<(Id, Change)>, u64)> {
+) -> trc::Result<Vec<(u64, Id, Change)>> {
     let mut changes = Vec::new();
-    let mut latest = since;
     data.iterate(
         IterateParams::new(
             change_key(account_id, since.saturating_add(1)),
@@ -213,13 +212,13 @@ pub async fn changes_since(
         .ascending(),
         |key, value| {
             if key.len() >= 8 && value.len() == 9 {
-                latest = latest.max(u64::from_be_bytes(key[key.len() - 8..].try_into().unwrap()));
                 let change = match value[8] {
                     0 => Change::Created,
                     1 => Change::Updated,
                     _ => Change::Destroyed,
                 };
                 changes.push((
+                    u64::from_be_bytes(key[key.len() - 8..].try_into().unwrap()),
                     Id::new(u64::from_be_bytes(value[0..8].try_into().unwrap())),
                     change,
                 ));
@@ -229,7 +228,7 @@ pub async fn changes_since(
     )
     .await
     .caused_by(trc::location!())?;
-    Ok((changes, latest))
+    Ok(changes)
 }
 
 /// The account's latest change id, 0 when there's none.

crates/features/src/masked_email/ops.rs

@@ -346,6 +346,55 @@ pub async fn sends_as(
         .is_some_and(|mask| mask.object.account_id.document_id() == account_id))
 }
 
+/// Changes to an account's masks since a state, for `/changes`.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct Changes {
+    pub created: Vec<Id>,
+    pub updated: Vec<Id>,
+    pub destroyed: Vec<Id>,
+    /// The state after the changes returned.
+    pub new_state: u64,
+    pub has_more: bool,
+}
+
+/// The changes since `since`, at most `max` masks' worth. A mask created and
+/// destroyed within the window isn't reported; one created and changed is
+/// reported as created.
+pub fn collapse(since: u64, entries: &[(u64, Id, Change)], max: usize) -> Changes {
+    use ahash::AHashMap;
+    let mut state: AHashMap<Id, (bool, bool)> = AHashMap::new(); // (created, destroyed)
+    let mut order = Vec::new();
+    let mut result = Changes {
+        new_state: since,
+        ..Default::default()
+    };
+    for (change_id, id, change) in entries {
+        if !state.contains_key(id) {
+            if order.len() == max {
+                result.has_more = true;
+                break;
+            }
+            order.push(*id);
+        }
+        let entry = state.entry(*id).or_insert((false, false));
+        match change {
+            Change::Created => entry.0 = true,
+            Change::Destroyed => entry.1 = true,
+            Change::Updated => {}
+        }
+        result.new_state = *change_id;
+    }
+    for id in order {
+        match state[&id] {
+            (true, true) => {}
+            (true, false) => result.created.push(id),
+            (false, true) => result.destroyed.push(id),
+            (false, false) => result.updated.push(id),
+        }
+    }
+    result
+}
+
 /// What an address is, as far as masks go.
 #[derive(Debug)]
 pub enum Lookup {
@@ -406,6 +455,30 @@ pub fn with_header(address: &str, message: &[u8]) -> Vec<u8> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn changes_collapse() {
+        let (a, b, c) = (Id::new(1), Id::new(2), Id::new(3));
+        let entries = [
+            (10, a, Change::Created),
+            (11, a, Change::Updated),
+            (12, b, Change::Updated),
+            (13, c, Change::Created),
+            (14, c, Change::Destroyed),
+            (15, b, Change::Destroyed),
+        ];
+        let all = collapse(9, &entries, 100);
+        assert_eq!(all.created, vec![a]);
+        assert_eq!(all.destroyed, vec![b]);
+        assert!(all.updated.is_empty());
+        assert_eq!(all.new_state, 15);
+        assert!(!all.has_more);
+
+        let first = collapse(9, &entries, 1);
+        assert_eq!(first.created, vec![a]);
+        assert_eq!(first.new_state, 11);
+        assert!(first.has_more);
+    }
+
     #[test]
     fn header_goes_first() {
         let out = with_header("[email protected]", b"Subject: hi\r\n\r\nbody");

crates/jmap-proto/src/references/eval.rs

@@ -128,6 +128,10 @@ impl Response<'_> {
                         ChangesResponseMethod::ShareNotification(response) => {
                             response.eval_jptr(path, &mut results)
                         }
+                        // inbuxa: x:MaskedEmail/changes
+                        ChangesResponseMethod::Registry(response) => {
+                            response.eval_jptr(path, &mut results)
+                        }
                     },
                     ResponseMethod::Query(response) => response.eval_jptr(path, &mut results),
                     ResponseMethod::QueryChanges(response) => {

crates/jmap-proto/src/request/method.rs

@@ -361,7 +361,12 @@ impl MethodName {
                 "get" => MethodFunction::Get,
                 "set" => MethodFunction::Set,
                 "query" => MethodFunction::Query,
+                "changes" => MethodFunction::Changes,
             )?;
+            // inbuxa: only masked email has /changes (a fork addition)
+            if fnc == MethodFunction::Changes && obj != ObjectType::MaskedEmail {
+                return None;
+            }
 
             if obj.flags() & OBJ_SINGLETON == 0 || fnc != MethodFunction::Query {
                 (MethodObject::Registry(obj), fnc).into()

crates/jmap-proto/src/response/mod.rs

@@ -137,6 +137,8 @@ pub enum ChangesResponseMethod {
     CalendarEvent(Box<ChangesResponse<CalendarEvent>>),
     CalendarEventNotification(Box<ChangesResponse<CalendarEventNotification>>),
     ShareNotification(Box<ChangesResponse<ShareNotification>>),
+    // inbuxa: x:MaskedEmail/changes
+    Registry(Box<ChangesResponse<crate::object::registry::Registry>>),
 }
 
 #[derive(Debug, serde::Serialize)]

crates/jmap/src/api/auth.rs

@@ -247,8 +247,9 @@ impl JmapAuthorization for AccessToken {
                 | MethodObject::PushSubscription
                 | MethodObject::SearchSnippet
                 | MethodObject::VacationResponse
-                | MethodObject::SieveScript
-                | MethodObject::Registry(_) => Permission::JmapEmailChanges,
+                | MethodObject::SieveScript => Permission::JmapEmailChanges,
+                // inbuxa: x:MaskedEmail/changes reads what /get reads
+                MethodObject::Registry(object_type) => object_type.get_permission(),
             },
             RequestMethod::Copy(m) => match &m {
                 CopyRequestMethod::Email(_) => Permission::JmapEmailCopy,

crates/jmap/src/api/request.rs

@@ -592,10 +592,15 @@ impl RequestHandler for Server {
             RequestMethod::Changes(mut req) => {
                 resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
 
+                // inbuxa: x:MaskedEmail/changes
+                if matches!(method_name.obj, MethodObject::Registry(_)) {
+                    crate::inbuxa::masked_email::changes(self, access_token, *req).await?
+                } else {
                     self.changes(*req, method_name.obj, access_token)
                         .await?
                         .into_method_response()
                 }
+            }
             RequestMethod::Copy(req) => match req {
                 CopyRequestMethod::Email(mut req) => {
                     resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;

crates/jmap/src/inbuxa/masked_email.rs

@@ -350,3 +350,67 @@ pub async fn query(mut req: RegistryQueryResponse<'_>) -> trc::Result<QueryRespo
     }
     Ok(response)
 }
+
+/// The state of an account's masks, for `/get` and `/changes` (a fork
+/// addition: upstream's `/get` has none).
+pub async fn state(server: &Server, account_id: u32) -> trc::Result<JmapState> {
+    let latest =
+        inbuxa_features::masked_email::data::latest_change(&server.core.storage.data, account_id)
+            .await?;
+    Ok(if latest == 0 {
+        JmapState::Initial
+    } else {
+        JmapState::Exact(latest)
+    })
+}
+
+/// `x:MaskedEmail/changes` (a fork addition).
+pub async fn changes(
+    server: &Server,
+    access_token: &AccessToken,
+    request: jmap_proto::method::changes::ChangesRequest,
+) -> trc::Result<jmap_proto::response::ResponseMethod<'static>> {
+    use jmap_proto::{
+        method::changes::ChangesResponse,
+        response::{ChangesResponseMethod, ResponseMethod},
+    };
+
+    let account_id = request.account_id.document_id();
+    assert_can_manage(server, access_token, account_id).await?;
+    let since = match &request.since_state {
+        JmapState::Initial => 0,
+        JmapState::Exact(change_id) => *change_id,
+        JmapState::Intermediate(_) => {
+            return Err(trc::JmapEvent::CannotCalculateChanges.into_err());
+        }
+    };
+    let max = request
+        .max_changes
+        .filter(|max| *max != 0)
+        .unwrap_or(usize::MAX)
+        .min(server.core.jmap.changes_max_results);
+    let entries = inbuxa_features::masked_email::data::changes_since(
+        &server.core.storage.data,
+        account_id,
+        since,
+    )
+    .await?;
+    let changes = ops::collapse(since, &entries, max);
+
+    Ok(ResponseMethod::Changes(ChangesResponseMethod::Registry(
+        Box::new(ChangesResponse {
+            account_id: request.account_id,
+            old_state: request.since_state,
+            new_state: if changes.new_state == 0 {
+                JmapState::Initial
+            } else {
+                JmapState::Exact(changes.new_state)
+            },
+            has_more_changes: changes.has_more,
+            created: changes.created,
+            updated: changes.updated,
+            destroyed: changes.destroyed,
+            updated_properties: None,
+        }),
+    )))
+}

crates/jmap/src/registry/get.rs

@@ -357,6 +357,12 @@ impl RegistryGet for Server {
                     get.insert(id, object);
                 }
 
+                // inbuxa: a state for masked email (a fork addition)
+                if object_type == ObjectType::MaskedEmail {
+                    get.response.state =
+                        Some(crate::inbuxa::masked_email::state(self, get.account_id).await?);
+                }
+
                 Ok(get.into_response())
             }
             ObjectType::QueuedMessage => {