Legal holds, step 5: held accounts can't be destroyed

Destroying a held account removes the login, as offboarding needs, but
keeps its data as a deleted account with no expiry, whether or not
undelete keeps accounts; its addresses stay reserved and its holds name
it from then on. Destroy-now refuses it, and its DestroyAccount task
defers itself while it's held or its time hasn't come. Holds placed or
released later freeze or free kept accounts in the same settle pass,
with 30 days' grace after the last release (LH-8, LH-10).

crates/common/src/hold.rs

@@ -15,7 +15,14 @@ use inbuxa_features::{
     hold::{self, HELD_UNTIL, Hold, Keeping, Member, is_held_until},
     undelete::records,
 };
-use registry::schema::{prelude::ObjectType, structs::ArchivedItem};
+use inbuxa_features::undelete::data::{self as undelete_data, KeptAccount};
+use registry::{
+    pickle::PickledStream,
+    schema::{
+        prelude::{ObjectInner, ObjectType},
+        structs::ArchivedItem,
+    },
+};
 use store::{registry::RegistryQuery, write::now};
 use trc::AddContext;
 use types::id::Id;
@@ -29,6 +36,21 @@ const RELEASE_GRACE: u64 = 30 * 86_400;
 pub struct Settled {
     pub frozen: usize,
     pub released: usize,
+    /// Deleted accounts kept by a hold, or let go by a release (LH-8, LH-10).
+    pub accounts_frozen: usize,
+    pub accounts_released: usize,
+}
+
+/// A kept account as it was when deleted, for a hold's scope: its record
+/// still names its domain, groups and tenant.
+pub fn kept_member(account_id: u32, kept: &KeptAccount) -> Member {
+    PickledStream::new(&kept.record)
+        .and_then(|mut stream| ObjectInner::unpickle(ObjectType::Account, &mut stream))
+        .and_then(|inner| Member::of(account_id, &inner))
+        .unwrap_or(Member {
+            account: account_id,
+            ..Default::default()
+        })
 }
 
 impl Server {
@@ -114,9 +136,41 @@ impl Server {
                 settled.released += 1;
             }
         }
+
+        // LH-8, LH-10: deleted accounts kept by undelete follow the holds
+        // too. Their DestroyAccount task defers itself while they're kept.
+        let retention = inbuxa_features::undelete::settings::retention(registry)
+            .await?
+            .accounts;
+        for (account_id, mut kept) in undelete_data::kept_accounts(data).await? {
+            let covered = !hold::covering(data, &kept_member(account_id, &kept)).await?.is_empty();
+            let held = is_held_until(kept.kept_until);
+            let until = if covered && !held {
+                settled.accounts_frozen += 1;
+                HELD_UNTIL
+            } else if !covered && held {
+                settled.accounts_released += 1;
+                (kept.deleted_at + retention.unwrap_or(0)).max(now + RELEASE_GRACE)
+            } else {
+                continue;
+            };
+            kept.kept_until = until;
+            let mut batch = store::write::BatchBuilder::new();
+            undelete_data::set_kept_account(&mut batch, account_id, &kept)?;
+            data.write(batch.build_all())
+                .await
+                .caused_by(trc::location!())?;
+        }
         Ok(settled)
     }
 
+    /// LH-8: whether a hold covers a deleted account undelete keeps.
+    pub async fn is_kept_held(&self, account_id: u32, kept: &KeptAccount) -> trc::Result<bool> {
+        Ok(!hold::covering(self.store(), &kept_member(account_id, kept))
+            .await?
+            .is_empty())
+    }
+
     /// Every account an active hold covers now. Empty, without looking at
     /// accounts, when nothing is held.
     pub async fn held_accounts(&self) -> trc::Result<ahash::AHashSet<u32>> {

crates/features/src/hold/mod.rs

@@ -484,6 +484,19 @@ pub async fn keep_moved(data: &Store, before: &Member, after: &Member) -> trc::R
     Ok(())
 }
 
+/// LH-8: names `account_id` in every hold that reaches it, so a deleted
+/// account, no longer in any domain or tenant, stays held.
+pub async fn pin_account(data: &Store, member: &Member) -> trc::Result<()> {
+    for mut hold in covering(data, member).await? {
+        if !hold.scope.accounts.contains(&member.account) {
+            hold.scope.accounts.push(member.account);
+            hold.scope.accounts.sort_unstable();
+            update(data, &hold).await?;
+        }
+    }
+    Ok(())
+}
+
 /// Replaces a hold that `check_update` allowed.
 pub async fn update(data: &Store, hold: &Hold) -> trc::Result<()> {
     let mut batch = BatchBuilder::new();

crates/jmap/src/inbuxa/deleted_account.rs

@@ -124,13 +124,29 @@ pub async fn reserved(
 /// of upstream's immediate destruction. Returns the other accounts whose
 /// access changed, or `None` when nothing is kept.
 pub async fn keep(server: &Server, id: Id, account: &Account) -> trc::Result<Option<Vec<u32>>> {
-    let Some(period) = retention(server.registry()).await?.accounts else {
-        return Ok(None);
-    };
     let account_id = id.document_id();
-    let deleted_at = now();
-    let kept_until = deleted_at + period;
     let inner = ObjectInner::Account(account.clone());
+    // inbuxa: LH-8: a held account's data stays, with no expiry, whether or
+    // not undelete keeps accounts; its holds name it from now on
+    let member = inbuxa_features::hold::Member::of(account_id, &inner);
+    let held = match &member {
+        Some(member) => {
+            !inbuxa_features::hold::covering(server.store(), member)
+                .await?
+                .is_empty()
+        }
+        None => false,
+    };
+    let period = retention(server.registry()).await?.accounts;
+    let deleted_at = now();
+    let kept_until = match (held, period) {
+        (true, _) => inbuxa_features::hold::HELD_UNTIL,
+        (false, Some(period)) => deleted_at + period,
+        (false, None) => return Ok(None),
+    };
+    if held && let Some(member) = &member {
+        inbuxa_features::hold::pin_account(server.store(), member).await?;
+    }
     let addresses = addresses_of(server, &inner)
         .await?
         .into_iter()
@@ -324,6 +340,18 @@ pub async fn set(
 
     for id in will_destroy {
         match data::kept_account(data, id.document_id()).await? {
+            // inbuxa: LH-8: a held account's data can't be destroyed
+            Some(kept)
+                if may_reach(access_token, &kept, Permission::SysAccountDestroy)
+                    && (inbuxa_features::hold::is_held_until(kept.kept_until)
+                        || server.is_kept_held(id.document_id(), &kept).await?) =>
+            {
+                response.not_destroyed.append(
+                    id,
+                    SetError::forbidden()
+                        .with_description("A legal hold applies to this account, so its data stays."),
+                );
+            }
             Some(kept) if may_reach(access_token, &kept, Permission::SysAccountDestroy) => {
                 destroy_now(server, id, &kept).await?;
                 response.destroyed.push(id);

crates/services/src/task_manager/destroy_account.rs

@@ -55,6 +55,23 @@ impl DestroyAccountTask for Server {
 async fn destroy_account(server: &Server, task: &TaskDestroyAccount) -> trc::Result<TaskResult> {
     let account_id = task.account_id.document_id();
 
+    // inbuxa: LH-8, LH-10: a kept account waits for its time, and a held one
+    // for its release; "destroy now" clears the kept record first
+    if let Some(kept) =
+        inbuxa_features::undelete::data::kept_account(&server.core.storage.data, account_id).await?
+    {
+        let now = store::write::now();
+        let held = inbuxa_features::hold::is_held_until(kept.kept_until)
+            || server.is_kept_held(account_id, &kept).await?;
+        if held || kept.kept_until > now {
+            let retry = if held { now + 86_400 } else { kept.kept_until };
+            return Ok(TaskResult::deferred(
+                Some(retry),
+                "The account is still kept: a legal hold applies, or its time hasn't come.",
+            ));
+        }
+    }
+
     // Destroy public keys and masked emails
     for object in [ObjectType::PublicKey, ObjectType::MaskedEmail] {
         let mut batch = BatchBuilder::new();