Export what a legal hold keeps as a ZIP (LH-12)

inbuxa:HoldExport/set takes a hold, optionally some of the accounts it
covers, and a reason; the collection runs in the background and get
says when it's ready. The ZIP has, per account, mail as .eml under its
folders, calendars as .ics, contacts as .vcf, files as stored, and the
archived items the hold keeps under archived/; a manifest.csv gives each
entry's account, kind, folder, date, whether it was archived, size and
SHA-256, and manifest.sha256 hashes the manifest. Accounts the hold
doesn't cover are left out, and items outside its date range are too:
live mail by arrival, events by start, and archived items the same way,
so an export doesn't carry deleted items that only another hold keeps.

The finished file is a blob of whoever started the export, so only they
download it, and it lasts as long as any upload (uploadTtl). Exports
are records under the hold (SUBSPACE_INBUXA H/e): never changed or
destroyed, each with its status, counts, size and checksum. Starting
one needs sysLegalHoldExport, an active hold and a reason, and is
recorded in the audit log like the audit log's own export.

The build is in memory and capped at 2 GB; bigger holds fail with a
message saying so, and are split by picking accounts.

Tested: unit tests for safe ZIP names and the manifest and its hash;
the legal_hold system test, on RocksDB, PostgreSQL and MySQL, exports a
hold end to end (live and archived mail, the manifest's hash, an asked-
for account the hold doesn't cover left out) and checks the refusals
(no reason, a user without the permission, a released hold) and the
audit record; and by hand from the console on a local server. Not
covered by a test: the archived-item date range with two holds of
different ranges over one account.

crates/features/src/hold/mod.rs

@@ -108,6 +108,45 @@ impl Keeping {
 const FEATURE: u8 = b'H';
 const KIND_HOLD: u8 = b'h';
 const KIND_ORIGINAL: u8 = b'o';
+const KIND_EXPORT: u8 = b'e';
+
+/// How far a hold export has got (LH-12).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum ExportStatus {
+    Running,
+    Ready,
+    Failed,
+}
+
+/// A collection of what a hold keeps, as a ZIP (LH-12).
+#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Export {
+    pub id: u32,
+    pub hold_id: u32,
+    /// The accounts asked for; empty for every account the hold covers.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub accounts: Vec<u32>,
+    pub reason: String,
+    pub created_at: u64,
+    pub created_by: String,
+    /// Whose blob the ZIP is, so only they download it.
+    pub created_by_id: u32,
+    pub status: ExportStatus,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub finished_at: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub blob_id: Option<String>,
+    #[serde(default)]
+    pub size: u64,
+    #[serde(default)]
+    pub items: u64,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sha256: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+}
 
 /// How many times creating a hold retries when another node took its id.
 const CREATE_ATTEMPTS: usize = 5;
@@ -402,6 +441,70 @@ pub async fn set_original_deadline(data: &Store, item_id: u64, until: Option<u64
         .map(|_| ())
 }
 
+fn export_class(id: u32) -> ValueClass {
+    let mut key = Vec::with_capacity(6);
+    key.push(FEATURE);
+    key.push(KIND_EXPORT);
+    key.extend_from_slice(&id.to_be_bytes());
+    ValueClass::Any(AnyClass {
+        subspace: SUBSPACE_INBUXA,
+        key,
+    })
+}
+
+/// Every hold export, oldest first.
+pub async fn exports(data: &Store) -> trc::Result<Vec<Export>> {
+    let mut exports = Vec::new();
+    data.iterate(
+        IterateParams::new(ValueKey::from(export_class(0)), ValueKey::from(export_class(u32::MAX))),
+        |_, value| {
+            if let Ok(Json(export)) = Json::<Export>::deserialize(value) {
+                exports.push(export);
+            }
+            Ok(true)
+        },
+    )
+    .await
+    .caused_by(trc::location!())?;
+    Ok(exports)
+}
+
+/// Writes a new export under the next free id, which it returns.
+pub async fn create_export(data: &Store, export: &Export) -> trc::Result<u32> {
+    let mut attempt = 0;
+    loop {
+        attempt += 1;
+        let id = exports(data).await?.iter().map(|e| e.id).max().unwrap_or(0) + 1;
+        let stored = Export {
+            id,
+            ..export.clone()
+        };
+        let mut batch = BatchBuilder::new();
+        batch.assert_value(export_class(id), AssertValue::None);
+        batch.set(export_class(id), Json(&stored).serialize()?);
+        match data.write(batch.build_all()).await {
+            Ok(_) => return Ok(id),
+            Err(err)
+                if attempt < CREATE_ATTEMPTS
+                    && matches!(
+                        err.as_ref(),
+                        trc::EventType::Store(trc::StoreEvent::AssertValueFailed)
+                    ) => {}
+            Err(err) => return Err(err.caused_by(trc::location!())),
+        }
+    }
+}
+
+/// Saves an export's progress.
+pub async fn update_export(data: &Store, export: &Export) -> trc::Result<()> {
+    let mut batch = BatchBuilder::new();
+    batch.set(export_class(export.id), Json(export).serialize()?);
+    data.write(batch.build_all())
+        .await
+        .caused_by(trc::location!())
+        .map(|_| ())
+}
+
 /// One hold, released or not.
 pub async fn get(data: &Store, id: u32) -> trc::Result<Option<Hold>> {
     Ok(data