vcon-server 0.4.0

//! Where the server's data is kept: each tenant's catalog of rows and its
//! sealed objects, the registry of tenants and keys, and the places
//! tenants are made in.
//!
//! These are traits because Store and Tenants are written once over more
//! than one backend: the local data directory now, and a PostgreSQL
//! catalog and S3-compatible objects next. Store and Tenants hold every
//! rule and call these for each statement they need; a backend implements
//! statements, never a rule, so the rules cannot drift between backends.
//!
//! The local backend, SQLCipher catalog files and directories of object
//! files, is in `local`.

use std::io::Write;
use std::path::Path;
use std::sync::Arc;

use time::OffsetDateTime;
use uuid::Uuid;

use crate::audit::{Action, AuditError, Entry};
use crate::clock::Clock;
use crate::envelope::Key;
use crate::ids::{FileId, KeyId, TenantId, VersionId};
use crate::store::{OpenError, Store, StoreError};
use crate::tenants::TenantError;

#[cfg(test)]
mod conformance;
pub mod local;
// Moving objects into a bucket, so only a build with S3 has it.
#[cfg(feature = "s3")]
pub(crate) mod migrate;
pub(crate) mod pack;
#[cfg(feature = "s3")]
pub(crate) mod s3;

/// A tenant's catalog row for a uuid, as stored.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VconRow {
    pub current: Option<VersionId>,
    pub deleted_at: Option<i64>,
}

/// A version row's columns that reading it needs.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VersionRow {
    /// The form's name as stored: unsigned, signed or encrypted.
    pub form: String,
    pub gzip: bool,
    pub wrapped: Vec<u8>,
    /// Where its sealed blob is when it is packed; None: in its object.
    pub place: Option<Packed>,
}

/// A packed version's row in its pack file, and the file as its catalog
/// row records it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Packed {
    pub file: FileId,
    /// The UTC store date of the file's rows, as its name writes it.
    pub date: String,
    /// The file's length in bytes.
    pub bytes: u64,
    pub place: pack::Place,
}

/// A version row to add.
#[derive(Debug, Clone, Copy)]
pub struct NewVersion<'a> {
    pub version: VersionId,
    pub uuid: Uuid,
    pub form: &'static str,
    pub gzip: bool,
    pub size: i64,
    pub stored_at: i64,
    pub key: KeyId,
    pub wrapped: &'a [u8],
}

/// A hidden version as the purge rule reads it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct HiddenRow {
    pub uuid: Uuid,
    pub stored_at: i64,
    pub held: bool,
    pub deleted: bool,
}

/// A kept version as a legal listing reads it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct KeptRow {
    pub version: VersionId,
    pub stored_at: i64,
    pub hidden_at: Option<i64>,
}

/// A live uuid as the list reads it: its current version, the uuid's first
/// store time, and the current version's store time if its row is there.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ListedRow {
    pub uuid: Uuid,
    pub version: VersionId,
    pub stored_at: i64,
    pub updated_at: Option<i64>,
}

/// What a page of the list asks the catalog for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PageQuery {
    /// Rows strictly before this (stored_at, uuid) in the list's order.
    pub before: Option<(i64, Uuid)>,
    pub since: Option<i64>,
    pub until: Option<i64>,
    /// How many rows at most.
    pub rows: i64,
}

/// A version row as `check` and `examine` read it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ExaminedRow {
    pub version: VersionId,
    pub uuid: Uuid,
    pub stored_at: i64,
    pub hidden_at: Option<i64>,
    pub held: bool,
}

/// A kept version as the export reads it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExportedRow {
    pub version: VersionId,
    pub uuid: Uuid,
    pub form: String,
    pub gzip: bool,
    pub size: i64,
    pub stored_at: i64,
    /// Its uuid's current version, and the uuid not deleted.
    pub live: bool,
    pub wrapped: Vec<u8>,
    /// Where its sealed blob is when it is packed; None: in its object.
    pub place: Option<Packed>,
}

/// A version still in its own object, as a pack run plans it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct UnpackedRow {
    pub version: VersionId,
    pub stored_at: i64,
    /// The body's size in bytes, as received.
    pub size: i64,
}

/// A recorded pack file, as its row says.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PackFileRow {
    pub date: String,
    pub bytes: u64,
    /// When no place named it any more; None while one may.
    pub unnamed_at: Option<i64>,
}

/// A new pack file's row.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct NewPackFile<'a> {
    pub file: FileId,
    pub date: &'a str,
    pub rows: i64,
    pub bytes: i64,
    pub created_at: i64,
}

/// A data key to rewrap.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WrappedRow {
    pub version: VersionId,
    pub uuid: Uuid,
    pub wrapped: Vec<u8>,
}

/// How a commit that did not succeed ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[must_use]
pub enum CommitError {
    /// Nothing of the transaction was kept.
    RolledBack,
    /// The transaction may or may not have been kept.
    Unknown,
}

impl std::fmt::Display for CommitError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            CommitError::RolledBack => "the transaction was rolled back",
            CommitError::Unknown => "the transaction may or may not have been kept",
        })
    }
}

impl std::error::Error for CommitError {}

/// One tenant's catalog.
pub trait Catalog: Send + Sync {
    /// A transaction that writes: one at a time per tenant, so a tenant's
    /// writes queue here.
    fn immediate(&self) -> Result<Box<dyn CatalogTx + '_>, StoreError>;
    /// A transaction that only reads, seeing one consistent state.
    fn snapshot(&self) -> Result<Box<dyn CatalogTx + '_>, StoreError>;
    /// Start-up recovery of the catalog itself (local: the WAL is
    /// checkpointed with truncation).
    fn recover(&self) -> Result<(), StoreError>;
    /// What follows purges (local: the WAL is checkpointed with truncation,
    /// asking without waiting and waiting between tries).
    fn after_purge(&self) -> Result<(), StoreError>;
    /// Copies the catalog, as it is in one transaction, into a local
    /// catalog file at `to` under the same catalog key, and returns the
    /// version ids that copy holds, each with its place in that copy.
    fn backup_into(&self, to: &Path) -> Result<Vec<(VersionId, Option<Packed>)>, StoreError>;
    /// Makes the copy `backup_into` wrote at `copy` say that every version
    /// is in its own object: no place, no pack file, no purge left to
    /// finish.
    fn unpack_copy(&self, copy: &Path) -> Result<(), StoreError>;
    /// Writes every audit record after the seq `after` as JSON lines,
    /// oldest first.
    fn export(&self, after: i64, out: &mut dyn Write) -> Result<(), AuditError>;
    /// The local connection, for tests that read or break the catalog
    /// directly.
    /// Makes the `nth` commit from now fail as `how` says: kept and then
    /// reported unknown, or rolled back.
    #[cfg(test)]
    fn fail_commit(&self, nth: usize, how: CommitError);
    /// For tests: `busy` is sent a message each time a checkpoint after a
    /// purge finds a reader in its way and is about to wait.
    #[cfg(test)]
    fn on_checkpoint_wait(&self, busy: std::sync::mpsc::Sender<()>);
    #[cfg(test)]
    fn local_conn(&self) -> std::sync::MutexGuard<'_, rusqlite::Connection>;
}

/// The statements Store runs in a catalog transaction. Dropping one
/// without `commit` rolls it back.
pub trait CatalogTx {
    /// Whether the uuid has a tombstone: it was purged.
    fn tombstoned(&mut self, uuid: Uuid) -> Result<bool, StoreError>;
    fn vcon_row(&mut self, uuid: Uuid) -> Result<Option<VconRow>, StoreError>;
    fn version_row(&mut self, version: VersionId) -> Result<Option<VersionRow>, StoreError>;
    /// Whether `version` is a kept version of `uuid`.
    fn has_version(&mut self, version: VersionId, uuid: Uuid) -> Result<bool, StoreError>;
    fn insert_version(&mut self, row: &NewVersion) -> Result<(), StoreError>;
    fn insert_vcon(
        &mut self,
        uuid: Uuid,
        current: VersionId,
        stored_at: i64,
    ) -> Result<(), StoreError>;
    fn set_current(&mut self, uuid: Uuid, current: VersionId) -> Result<(), StoreError>;
    fn hide_version(&mut self, version: VersionId, at: i64) -> Result<(), StoreError>;
    /// Marks the uuid deleted at `at` with no current version, and hides
    /// every version of it not hidden yet.
    fn mark_deleted(&mut self, uuid: Uuid, at: i64) -> Result<(), StoreError>;
    /// Wraps counted against a tenant key.
    fn wraps(&mut self, key: KeyId) -> Result<u64, StoreError>;
    fn add_wrap(&mut self, key: KeyId) -> Result<(), StoreError>;
    /// Every version still in its own object, by store time and then
    /// version id.
    fn unpacked(&mut self) -> Result<Vec<UnpackedRow>, StoreError>;
    /// Whether `version` is kept and still in its own object.
    fn is_unpacked(&mut self, version: VersionId) -> Result<bool, StoreError>;
    fn insert_pack_file(&mut self, row: &NewPackFile) -> Result<(), StoreError>;
    /// Moves `version`'s place to `place` in `file`.
    fn set_place(
        &mut self,
        version: VersionId,
        file: FileId,
        place: pack::Place,
    ) -> Result<(), StoreError>;
    /// Every pack file the catalog records.
    fn pack_files(&mut self) -> Result<Vec<FileId>, StoreError>;
    /// A recorded pack file's row; None if it has none.
    fn pack_file(&mut self, file: FileId) -> Result<Option<PackFileRow>, StoreError>;
    /// Every kept version packed in `file`, with its store time and place,
    /// by row group and row.
    fn placed_in(&mut self, file: FileId) -> Result<Vec<(VersionId, i64, Packed)>, StoreError>;
    /// Lists `file` for a purge's rewrite; one listed already stays listed.
    fn add_purge_pending(&mut self, file: FileId) -> Result<(), StoreError>;
    fn purge_pending(&mut self) -> Result<Vec<FileId>, StoreError>;
    fn delete_purge_pending(&mut self, file: FileId) -> Result<(), StoreError>;
    /// Marks `file` as no longer named by any place, at `at`.
    fn unname_pack_file(&mut self, file: FileId, at: i64) -> Result<(), StoreError>;
    /// Every recorded file marked unnamed.
    fn unnamed_files(&mut self) -> Result<Vec<FileId>, StoreError>;
    fn delete_pack_file(&mut self, file: FileId) -> Result<(), StoreError>;
    /// Takes the next manifest number, never one given before.
    fn allocate_manifest(&mut self) -> Result<i64, StoreError>;
    /// The manifest numbers taken and not yet marked written, in order.
    fn unpublished_manifests(&mut self) -> Result<Vec<i64>, StoreError>;
    fn mark_published(&mut self, seq: i64, at: i64) -> Result<(), StoreError>;
    /// Every manifest marked written, with when, by number.
    fn published_manifests(&mut self) -> Result<Vec<(i64, i64)>, StoreError>;
    fn delete_manifest(&mut self, seq: i64) -> Result<(), StoreError>;
    /// Every pack file the catalog names (not unnamed), with its date, its
    /// rows and its length, by date and then file id.
    fn named_files(&mut self) -> Result<Vec<(FileId, PackFileRow, i64)>, StoreError>;
    /// The versions a read returns: current, of a uuid not deleted, in
    /// version order.
    fn live_versions(&mut self) -> Result<Vec<VersionId>, StoreError>;
    /// Every version id with its place, in version order; a place that is
    /// not whole is `Damaged` for that version alone.
    #[expect(clippy::type_complexity, reason = "one result per version")]
    fn places(
        &mut self,
    ) -> Result<Vec<(VersionId, Result<Option<Packed>, StoreError>)>, StoreError>;
    fn count_versions(&mut self) -> Result<u64, StoreError>;
    /// The versions whose row names a pack file.
    fn count_packed(&mut self) -> Result<u64, StoreError>;
    fn count_held(&mut self) -> Result<u64, StoreError>;
    /// The hidden versions, of one uuid or of all.
    fn hidden_versions(&mut self, uuid: Option<Uuid>) -> Result<Vec<VersionId>, StoreError>;
    /// A hidden version with its uuid's hold and deletion; None if it is
    /// gone or not hidden.
    fn hidden_row(&mut self, version: VersionId) -> Result<Option<HiddenRow>, StoreError>;
    fn delete_version(&mut self, version: VersionId) -> Result<(), StoreError>;
    fn versions_of(&mut self, uuid: Uuid) -> Result<u64, StoreError>;
    /// Removes the uuid's row and adds its tombstone.
    fn tombstone(&mut self, uuid: Uuid, at: i64) -> Result<(), StoreError>;
    /// The latest kept version of a uuid, by store time and then by the
    /// order the rows were added.
    fn latest(&mut self, uuid: Uuid) -> Result<Option<VersionId>, StoreError>;
    /// Every kept version of a uuid, oldest first, in the same order.
    fn kept(&mut self, uuid: Uuid) -> Result<Vec<KeptRow>, StoreError>;
    fn unhide(&mut self, version: VersionId) -> Result<(), StoreError>;
    /// Makes `version` current and clears the uuid's deletion.
    fn undelete(&mut self, uuid: Uuid, version: VersionId) -> Result<(), StoreError>;
    fn set_hold(&mut self, uuid: Uuid, on: bool) -> Result<(), StoreError>;
    fn page(&mut self, query: &PageQuery) -> Result<Vec<ListedRow>, StoreError>;
    /// Up to `limit` data keys wrapped under `key`, in version order.
    fn wrapped_under(&mut self, key: KeyId, limit: usize) -> Result<Vec<WrappedRow>, StoreError>;
    fn set_wrapped(
        &mut self,
        version: VersionId,
        key: KeyId,
        wrapped: &[u8],
    ) -> Result<(), StoreError>;
    fn count_not_under(&mut self, key: KeyId) -> Result<u64, StoreError>;
    /// Every version row with its uuid's hold, in version order.
    fn examined(&mut self) -> Result<Vec<ExaminedRow>, StoreError>;
    /// Every kept version stored at or after `since` and before `until`,
    /// by store time and then version id.
    fn exported(
        &mut self,
        since: Option<i64>,
        until: Option<i64>,
    ) -> Result<Vec<ExportedRow>, StoreError>;
    /// The times of every successful purge recorded, as stored.
    fn purge_times(&mut self) -> Result<Vec<String>, StoreError>;
    /// Starts a nested part that `rollback_part` undoes and `release_part`
    /// keeps.
    fn begin_part(&mut self) -> Result<(), StoreError>;
    fn release_part(&mut self) -> Result<(), StoreError>;
    fn rollback_part(&mut self) -> Result<(), StoreError>;
    /// Appends an audit record.
    fn record(&mut self, entry: &Entry, now: OffsetDateTime) -> Result<(), AuditError>;
    /// Adds one to a counted audit record.
    fn count(&mut self, entry: &Entry, now: OffsetDateTime) -> Result<(), AuditError>;
    /// Writes at most `limit` audit records after the seq `after` as JSON
    /// lines, oldest first; None is no limit.
    fn audit_lines(
        &mut self,
        after: i64,
        limit: Option<u32>,
        out: &mut dyn Write,
    ) -> Result<(), AuditError>;
    fn commit(self: Box<Self>) -> Result<(), CommitError>;
}

/// Why an object call failed.
#[derive(Debug)]
#[must_use]
pub enum ObjectError {
    /// No object has the name.
    Missing,
    /// The storage failed; the error says how.
    Io(std::io::Error),
}

impl std::fmt::Display for ObjectError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ObjectError::Missing => f.write_str("no object has the name"),
            ObjectError::Io(e) => write!(f, "the object storage failed: {e}"),
        }
    }
}

impl std::error::Error for ObjectError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            ObjectError::Missing => None,
            ObjectError::Io(e) => Some(e),
        }
    }
}

/// One tenant's sealed objects, named by version id.
pub trait Objects: Send + Sync {
    /// Writes a new object; it is durable once this returns Ok. On an
    /// error nothing of it is left that `list` would call a version.
    fn put(&self, version: VersionId, blob: &[u8]) -> Result<(), ObjectError>;
    fn get(&self, version: VersionId) -> Result<Vec<u8>, ObjectError>;
    /// Removes an object. One already absent is `Missing`, or Ok where the
    /// storage cannot tell (S3 answers success either way); callers treat
    /// both as removed.
    fn remove(&self, version: VersionId) -> Result<(), ObjectError>;
    /// Every name held, sorted, leftovers included.
    fn list(&self) -> Result<Vec<String>, ObjectError>;
    /// Removes a name `list` returned.
    fn remove_name(&self, name: &str) -> Result<(), ObjectError>;
    /// Makes removals durable.
    fn sync(&self) -> Result<(), ObjectError>;
}

/// One tenant's pack files, each named by `pack::name` under the tenant's
/// own directory. A file never changes once written.
pub trait Packs: Send + Sync + std::fmt::Debug {
    /// `len` bytes of the file `name` from byte `start`; fewer if the file
    /// ends sooner. `Missing` if no file has the name.
    fn read(&self, name: &str, start: u64, len: u64) -> Result<Vec<u8>, ObjectError>;
    /// The file's length; None if no file has the name.
    fn size(&self, name: &str) -> Result<Option<u64>, ObjectError>;
    /// Writes a new file, durable once this returns, only if no file has
    /// the name (`If-None-Match: *` on S3). A name already taken is
    /// `Ok(false)` and writes nothing.
    fn put_new(&self, name: &str, bytes: &[u8]) -> Result<bool, ObjectError>;
    /// Removes a file; one already gone is removed too.
    fn remove(&self, name: &str) -> Result<(), ObjectError>;
    /// Every file name under the tenant's `parquet/`, as `pack::name`
    /// writes them and any other, sorted.
    fn list(&self) -> Result<Vec<String>, ObjectError>;
}

/// A tenant's row in the registry, as stored.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TenantRow {
    pub id: TenantId,
    pub name: String,
    /// The period as `Retention` writes it; None is the server default.
    pub retention: Option<String>,
    /// The metadata level as `Metadata` writes it; None is `none`.
    pub metadata: Option<String>,
    pub created_at: i64,
    pub tenant_key_id: KeyId,
    pub wrapped: Vec<u8>,
    pub pending: Option<Vec<u8>>,
    pub key_since: i64,
}

/// An access key's row in the registry, as stored.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct KeyRow {
    pub id: KeyId,
    pub tenant: Option<TenantId>,
    pub token_hash: Vec<u8>,
    /// The scopes as `Scopes` writes them.
    pub scopes: String,
    pub label: String,
    pub created_at: i64,
    pub revoked_at: Option<i64>,
}

/// The registry: tenants, access keys, master-key use and the admin log.
pub trait Registry: Send + Sync {
    /// A transaction that writes: one at a time.
    fn immediate(&self) -> Result<Box<dyn RegistryTx + '_>, TenantError>;
    /// A transaction that only reads, seeing one consistent state.
    fn snapshot(&self) -> Result<Box<dyn RegistryTx + '_>, TenantError>;
    /// Writes every admin record after the seq `after` as JSON lines,
    /// oldest first.
    fn export(&self, after: i64, out: &mut dyn Write) -> Result<(), AuditError>;
    /// Copies the registry, as it is in one transaction, into a local
    /// registry file at `to` under the same key, and returns the tenants
    /// that copy lists, by id.
    fn backup_into(&self, to: &Path) -> Result<Vec<TenantId>, TenantError>;
    /// Makes the `nth` commit from now fail as `how` says.
    #[cfg(test)]
    fn fail_commit(&self, nth: usize, how: CommitError);
    /// The local connection, for tests that read or break the registry
    /// directly.
    #[cfg(test)]
    fn local_conn(&self) -> std::sync::MutexGuard<'_, rusqlite::Connection>;
}

/// The statements Tenants runs in a registry transaction. Dropping one
/// without `commit` rolls it back.
pub trait RegistryTx {
    fn tenant(&mut self, id: TenantId) -> Result<Option<TenantRow>, TenantError>;
    /// Every tenant, by name.
    fn tenants(&mut self) -> Result<Vec<TenantRow>, TenantError>;
    fn name_taken(&mut self, name: &str) -> Result<bool, TenantError>;
    fn insert_tenant(&mut self, row: &TenantRow) -> Result<(), TenantError>;
    /// Sets a tenant's retention; false if no tenant has the id.
    fn set_retention(&mut self, id: TenantId, retention: Option<&str>)
    -> Result<bool, TenantError>;
    /// Sets a tenant's metadata level; false if no tenant has the id.
    fn set_metadata(&mut self, id: TenantId, level: Option<&str>) -> Result<bool, TenantError>;
    /// Removes a tenant's row and revokes its access keys at `at`.
    fn delete_tenant(&mut self, id: TenantId, at: i64) -> Result<(), TenantError>;
    fn set_pending(&mut self, id: TenantId, pending: &[u8]) -> Result<(), TenantError>;
    /// Makes the pending key current, as `key` since `at`.
    fn promote_pending(&mut self, id: TenantId, key: KeyId, at: i64) -> Result<(), TenantError>;
    /// Wraps counted against a master key.
    fn master_wraps(&mut self, key: KeyId) -> Result<u64, TenantError>;
    fn add_master_wrap(&mut self, key: KeyId, at: i64) -> Result<(), TenantError>;
    /// When the master key's use began, if it is recorded.
    fn master_since(&mut self, key: KeyId) -> Result<Option<i64>, TenantError>;
    fn key(&mut self, id: KeyId) -> Result<Option<KeyRow>, TenantError>;
    /// Every access key, or one tenant's, by creation time then id.
    fn keys(&mut self, tenant: Option<TenantId>) -> Result<Vec<KeyRow>, TenantError>;
    fn insert_key(&mut self, row: &KeyRow) -> Result<(), TenantError>;
    /// Revokes a key at `at` unless it is revoked already.
    fn revoke_key(&mut self, id: KeyId, at: i64) -> Result<(), TenantError>;
    /// The times of the admin records of `action` with `status`, as
    /// stored.
    fn record_times(&mut self, action: Action, status: u16) -> Result<Vec<String>, TenantError>;
    fn record(&mut self, entry: &Entry, now: OffsetDateTime) -> Result<(), AuditError>;
    fn count(&mut self, entry: &Entry, now: OffsetDateTime) -> Result<(), AuditError>;
    fn audit_lines(
        &mut self,
        after: i64,
        limit: Option<u32>,
        out: &mut dyn Write,
    ) -> Result<(), AuditError>;
    fn commit(self: Box<Self>) -> Result<(), CommitError>;
}

/// Where each tenant's catalog and objects are made, opened and removed.
pub trait TenantSpace: Send + Sync {
    /// Makes a new tenant's catalog and objects, durably.
    fn make(&self, id: TenantId, key: Key, clock: Arc<dyn Clock>) -> Result<(), OpenError>;
    /// Opens a tenant's Store; with `recover`, runs start-up recovery.
    fn open(
        &self,
        id: TenantId,
        key: Key,
        recover: bool,
        clock: Arc<dyn Clock>,
    ) -> Result<Store, OpenError>;
    /// Every tenant that has a catalog or objects here, by id.
    fn present(&self) -> Result<Vec<TenantId>, TenantError>;
    /// Removes a tenant's catalog and objects.
    fn remove(&self, id: TenantId) -> Result<(), TenantError>;
    /// Which catalog key the tenant's catalog is under now, `old` or
    /// `new`; under neither is an error.
    fn catalog_under(&self, id: TenantId, old: &Key, new: &Key) -> Result<Under, TenantError>;
    /// Moves the tenant's catalog from the catalog key `old` to `new`.
    fn reseal(&self, id: TenantId, old: &Key, new: &Key) -> Result<(), TenantError>;
}

/// The key a tenant catalog is under.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Under {
    Old,
    New,
}

/// Where each tenant's sealed objects are kept, as the configuration says.
#[derive(Debug, Clone, Default)]
pub(crate) enum ObjectPlace {
    /// Files in each tenant's directory.
    #[default]
    Local,
    /// An S3-compatible bucket, under each tenant's own name.
    #[cfg(feature = "s3")]
    S3(Arc<s3::S3>),
}

/// The object place `storage` names, reached and checked: for a bucket,
/// the secret key is read from its file, the client made, and the bucket's
/// versioning read, which also shows that the bucket can be reached.
///
/// # Errors
///
/// A `StorageError` that says which setting or step failed.
pub(crate) fn objects_for(
    storage: &crate::config::Storage,
    #[cfg_attr(
        not(feature = "s3"),
        expect(
            unused_variables,
            reason = "only a bucket is bound to a data directory"
        )
    )]
    data_dir: &Path,
) -> Result<ObjectPlace, crate::config::StorageError> {
    match &storage.objects {
        crate::config::ObjectStore::Local => Ok(ObjectPlace::Local),
        #[cfg(feature = "s3")]
        crate::config::ObjectStore::S3(settings) => {
            s3::place(settings, storage.parquet.is_some(), data_dir)
        }
        #[cfg(not(feature = "s3"))]
        crate::config::ObjectStore::S3(_) => Err(crate::config::StorageError(
            "storage.objects is \"s3\", but this build has no S3 support (cargo feature s3)"
                .to_owned(),
        )),
    }
}