ahri_tre_app/

configured_datastore_create.rs

//! Privileged Datastore-creation workflow above configuration, Secret, and adapter capabilities.

use ahri_tre_lake::{
    AwsWorkloadIdentitySource, DatasetCsvSeed, DuckLakeAdapter, DuckLakeCredentialKind,
    DuckLakeOpenedCatalog, FilesystemCatalogTls, FilesystemLakeSessionConfig,
    ObjectLakeSessionConfig, ObjectStorageAuthentication, ObjectStorageLocation, ObjectStorageTls,
    S3UrlStyle, ScratchAttempt, ScratchAttemptId, TrustedScratch,
};
use ahri_tre_pgmeta::{
    PgDatastoreCreationConnection, PgDatastoreCreationSpec, PgDatastoreCreationStart,
    PgDatastoreCredentialRotationError, PgDatastoreValidationSpec, PgPendingDatastoreCreation,
    apply_datastore_credential_rotation, begin_datastore_creation, find_existing_datastore_binding,
    inspect_datastore_identity_bindings, mark_datastore_identity_failed,
    mark_datastore_identity_ready, read_datastore_identity_bindings,
    rollback_datastore_credential_rotation, validate_existing_datastore_postgresql,
};
use ahri_tre_runtime::{
    AdministrationFailureCode, AdministrationPeerIdentity, AdministrationRequest,
    AdministrationRequestHandler, AdministrationResponse, AzureManagedIdentity,
    ConfiguredAwsWorkloadIdentitySource, ConfiguredS3UrlStyle, ConfiguredSecret,
    DatastoreConfigurationId, DatastoreCreationPlan, LakeStoragePlan, ManagedSecretOperationError,
    PostgresqlBindingInspectionTarget, ResolvedOperationSecret, SessionTls,
    StorageAuthenticationPlan, StorageTlsPlan, TrustedRuntimeState,
};
use ahri_tre_secrets::internal::{
    ManagedSecretReferenceAuthorities, ManagedSecretReferenceAuthority,
    ManagedSecretReferenceInspectionError, ManagedSecretReferenceStatus, ManagedSecretRemovalProof,
    ManagedSecretRemovalProofErrorKind, StagedManagedSecret,
};
use ahri_tre_secrets::{
    ManagedSecretErrorKind, ManagedSecretMetadata, ManagedSecretReference, SecretMaterial,
};
use ahri_tre_types::{
    DatastoreCreationFailureCode, DatastoreIdentityBinding, DatastoreLakeLocation,
    DatastoreLifecycleState, EncryptionMode, LakeCatalogCredentialMode, NcName,
    NewDatastoreIdentityBinding, StudyId,
};
use chrono::{DateTime, Utc};
use uuid::Uuid;

const DEMO_DATASTORE: &str = "demo";
const DEMO_STUDY_ID: &str = "00000000-0000-4000-8000-000000000015";

/// Safe metadata for the binding-owned credential created during the saga.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreatedCatalogCredential {
    version: u64,
    created_at: DateTime<Utc>,
}

impl CreatedCatalogCredential {
    pub fn new(version: u64, created_at: DateTime<Utc>) -> Self {
        Self {
            version,
            created_at,
        }
    }

    pub fn version(&self) -> u64 {
        self.version
    }

    pub fn created_at(&self) -> DateTime<Utc> {
        self.created_at
    }
}

impl From<ManagedSecretMetadata> for CreatedCatalogCredential {
    fn from(metadata: ManagedSecretMetadata) -> Self {
        Self::new(metadata.version().get(), metadata.created_at())
    }
}
/// Protected stage carried only by the Datastore credential owner workflow.
pub struct StagedCatalogCredential {
    inner: StagedManagedSecret,
}

impl StagedCatalogCredential {
    pub fn version(&self) -> u64 {
        self.inner.metadata().version().get()
    }

    pub fn material(&self) -> &SecretMaterial {
        self.inner.material()
    }
}

impl std::fmt::Debug for StagedCatalogCredential {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("StagedCatalogCredential")
            .field("metadata", self.inner.metadata())
            .finish()
    }
}

/// One storage Secret paired with the exact configured reference that authorized it.
pub struct ResolvedCreationStorageSecret {
    configured: ConfiguredSecret,
    resolved: ResolvedOperationSecret,
}

impl ResolvedCreationStorageSecret {
    pub fn new(configured: ConfiguredSecret, resolved: ResolvedOperationSecret) -> Self {
        Self {
            configured,
            resolved,
        }
    }

    pub fn configured(&self) -> &ConfiguredSecret {
        &self.configured
    }

    pub fn resolved(&self) -> &ResolvedOperationSecret {
        &self.resolved
    }
}

impl std::fmt::Debug for ResolvedCreationStorageSecret {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ResolvedCreationStorageSecret")
            .field("configured", &"<configured secret reference>")
            .field("resolved", &self.resolved)
            .finish()
    }
}

/// Trusted authority retained by the handler. Clients never receive these capabilities.
pub trait DatastoreCreationAuthority: Send + Sync {
    fn deployment_id(&self) -> Result<Uuid, AdministrationFailureCode> {
        Err(AdministrationFailureCode::ConfigurationUnavailable)
    }

    fn plan(
        &self,
        datastore: &DatastoreConfigurationId,
    ) -> Result<DatastoreCreationPlan, AdministrationFailureCode>;

    fn resolve_administrator(
        &self,
        plan: &DatastoreCreationPlan,
    ) -> Result<ResolvedOperationSecret, AdministrationFailureCode>;

    fn plan_for_existing(
        &self,
        datastore: &DatastoreConfigurationId,
        datastore_id: uuid::Uuid,
    ) -> Result<DatastoreCreationPlan, AdministrationFailureCode>;

    fn resolve_existing_catalog_credential(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
    ) -> Result<ResolvedOperationSecret, AdministrationFailureCode>;

    fn resolve_storage(
        &self,
        plan: &DatastoreCreationPlan,
    ) -> Result<Vec<ResolvedCreationStorageSecret>, AdministrationFailureCode>;

    fn create_catalog_credential(
        &self,
        plan: &DatastoreCreationPlan,
        material: &SecretMaterial,
    ) -> Result<CreatedCatalogCredential, AdministrationFailureCode>;

    fn verify_catalog_credential(
        &self,
        plan: &DatastoreCreationPlan,
        material: &SecretMaterial,
        credential: &CreatedCatalogCredential,
    ) -> Result<(), AdministrationFailureCode>;

    fn stage_catalog_credential_replacement(
        &self,
        _plan: &DatastoreCreationPlan,
        _binding: &DatastoreIdentityBinding,
        _material: &SecretMaterial,
    ) -> Result<StagedCatalogCredential, AdministrationFailureCode> {
        Err(AdministrationFailureCode::CredentialSetupFailed)
    }

    fn activate_catalog_credential_replacement(
        &self,
        _plan: &DatastoreCreationPlan,
        _staged: &StagedCatalogCredential,
    ) -> Result<CreatedCatalogCredential, AdministrationFailureCode> {
        Err(AdministrationFailureCode::CredentialSetupFailed)
    }

    fn discard_catalog_credential_replacement(
        &self,
        _plan: &DatastoreCreationPlan,
        _staged: &StagedCatalogCredential,
    ) -> Result<(), AdministrationFailureCode> {
        Err(AdministrationFailureCode::CredentialSetupFailed)
    }

    fn retain_datastore_binding_reference(
        &self,
        _plan: &DatastoreCreationPlan,
        _reference: &ManagedSecretReference,
    ) -> Result<(), AdministrationFailureCode> {
        Err(AdministrationFailureCode::CredentialSetupFailed)
    }

    fn reconcilable_datastore_binding_reference_ids(
        &self,
        _reference: &ManagedSecretReference,
    ) -> Result<Vec<Uuid>, ManagedSecretReferenceInspectionError> {
        Ok(Vec::new())
    }

    fn release_absent_datastore_binding_reference(
        &self,
        _datastore_id: Uuid,
        _reference: &ManagedSecretReference,
    ) -> Result<(), ManagedSecretReferenceInspectionError> {
        Err(ManagedSecretReferenceInspectionError::unavailable())
    }

    fn postgresql_binding_inspection_targets(
        &self,
    ) -> Result<Vec<PostgresqlBindingInspectionTarget>, AdministrationFailureCode> {
        Err(AdministrationFailureCode::ConfigurationUnavailable)
    }

    fn resolve_postgresql_binding_inspection_administrator(
        &self,
        _target: &PostgresqlBindingInspectionTarget,
    ) -> Result<ResolvedOperationSecret, AdministrationFailureCode> {
        Err(AdministrationFailureCode::CredentialSetupFailed)
    }

    fn configured_datastore_ids(
        &self,
    ) -> Result<Vec<DatastoreConfigurationId>, AdministrationFailureCode> {
        Err(AdministrationFailureCode::ConfigurationUnavailable)
    }

    fn application_reference_status(
        &self,
        _reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        Err(ManagedSecretReferenceInspectionError::unavailable())
    }

    fn session_reference_status(
        &self,
        _reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        Err(ManagedSecretReferenceInspectionError::unavailable())
    }

    fn datastore_reference_status(
        &self,
        _reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        Err(ManagedSecretReferenceInspectionError::unavailable())
    }

    fn authentication_reference_status(
        &self,
        _reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        Err(ManagedSecretReferenceInspectionError::unavailable())
    }

    fn remove_managed_secret(
        &self,
        _reference: &ManagedSecretReference,
        _proof: ManagedSecretRemovalProof,
    ) -> Result<ManagedSecretMetadata, AdministrationFailureCode> {
        Err(AdministrationFailureCode::CredentialSetupFailed)
    }
}

impl DatastoreCreationAuthority for TrustedRuntimeState {
    fn deployment_id(&self) -> Result<Uuid, AdministrationFailureCode> {
        Ok(TrustedRuntimeState::deployment_id(self))
    }

    fn plan(
        &self,
        datastore: &DatastoreConfigurationId,
    ) -> Result<DatastoreCreationPlan, AdministrationFailureCode> {
        self.datastore_creation_plan(datastore)
            .map_err(|_| AdministrationFailureCode::ConfigurationUnavailable)
    }

    fn resolve_administrator(
        &self,
        plan: &DatastoreCreationPlan,
    ) -> Result<ResolvedOperationSecret, AdministrationFailureCode> {
        self.resolve_datastore_administrator(plan)
            .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
    }

    fn plan_for_existing(
        &self,
        datastore: &DatastoreConfigurationId,
        datastore_id: uuid::Uuid,
    ) -> Result<DatastoreCreationPlan, AdministrationFailureCode> {
        self.datastore_creation_plan_with_id(datastore, datastore_id)
            .map_err(|_| AdministrationFailureCode::ConfigurationUnavailable)
    }

    fn resolve_existing_catalog_credential(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
    ) -> Result<ResolvedOperationSecret, AdministrationFailureCode> {
        TrustedRuntimeState::resolve_existing_datastore_catalog_credential(self, plan, binding)
            .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
    }

    fn resolve_storage(
        &self,
        plan: &DatastoreCreationPlan,
    ) -> Result<Vec<ResolvedCreationStorageSecret>, AdministrationFailureCode> {
        plan.storage_secrets()
            .into_iter()
            .map(|configured| {
                self.resolve_datastore_creation_storage(plan, configured)
                    .map(|resolved| {
                        ResolvedCreationStorageSecret::new(configured.clone(), resolved)
                    })
                    .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
            })
            .collect()
    }

    fn create_catalog_credential(
        &self,
        plan: &DatastoreCreationPlan,
        material: &SecretMaterial,
    ) -> Result<CreatedCatalogCredential, AdministrationFailureCode> {
        TrustedRuntimeState::create_datastore_catalog_credential(self, plan, material)
            .map(Into::into)
            .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
    }

    fn verify_catalog_credential(
        &self,
        plan: &DatastoreCreationPlan,
        material: &SecretMaterial,
        credential: &CreatedCatalogCredential,
    ) -> Result<(), AdministrationFailureCode> {
        TrustedRuntimeState::verify_datastore_catalog_credential(
            self,
            plan,
            material,
            credential.version(),
        )
        .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
    }

    fn stage_catalog_credential_replacement(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
        material: &SecretMaterial,
    ) -> Result<StagedCatalogCredential, AdministrationFailureCode> {
        self.stage_datastore_catalog_credential_replacement(plan, binding, material)
            .map(|inner| StagedCatalogCredential { inner })
            .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
    }

    fn activate_catalog_credential_replacement(
        &self,
        plan: &DatastoreCreationPlan,
        staged: &StagedCatalogCredential,
    ) -> Result<CreatedCatalogCredential, AdministrationFailureCode> {
        self.activate_datastore_catalog_credential_replacement(plan, &staged.inner)
            .map(Into::into)
            .map_err(|error| match error {
                ManagedSecretOperationError::Resolve(error)
                    if error.kind() == ManagedSecretErrorKind::CommitUncertain =>
                {
                    AdministrationFailureCode::CredentialStateUncertain
                }
                _ => AdministrationFailureCode::CredentialSetupFailed,
            })
    }

    fn discard_catalog_credential_replacement(
        &self,
        plan: &DatastoreCreationPlan,
        staged: &StagedCatalogCredential,
    ) -> Result<(), AdministrationFailureCode> {
        self.discard_datastore_catalog_credential_replacement(plan, &staged.inner)
            .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
    }

    fn retain_datastore_binding_reference(
        &self,
        plan: &DatastoreCreationPlan,
        reference: &ManagedSecretReference,
    ) -> Result<(), AdministrationFailureCode> {
        TrustedRuntimeState::retain_datastore_binding_reference(self, plan, reference)
            .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
    }

    fn reconcilable_datastore_binding_reference_ids(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<Vec<Uuid>, ManagedSecretReferenceInspectionError> {
        TrustedRuntimeState::reconcilable_datastore_binding_reference_ids(self, reference)
            .map_err(|_| ManagedSecretReferenceInspectionError::unavailable())
    }

    fn release_absent_datastore_binding_reference(
        &self,
        datastore_id: Uuid,
        reference: &ManagedSecretReference,
    ) -> Result<(), ManagedSecretReferenceInspectionError> {
        TrustedRuntimeState::release_absent_datastore_binding_reference(
            self,
            datastore_id,
            reference,
        )
        .map_err(|_| ManagedSecretReferenceInspectionError::unavailable())
    }

    fn postgresql_binding_inspection_targets(
        &self,
    ) -> Result<Vec<PostgresqlBindingInspectionTarget>, AdministrationFailureCode> {
        TrustedRuntimeState::postgresql_binding_inspection_targets(self)
            .map_err(|_| AdministrationFailureCode::ConfigurationUnavailable)
    }

    fn resolve_postgresql_binding_inspection_administrator(
        &self,
        target: &PostgresqlBindingInspectionTarget,
    ) -> Result<ResolvedOperationSecret, AdministrationFailureCode> {
        TrustedRuntimeState::resolve_postgresql_binding_inspection_administrator(self, target)
            .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
    }

    fn configured_datastore_ids(
        &self,
    ) -> Result<Vec<DatastoreConfigurationId>, AdministrationFailureCode> {
        Ok(TrustedRuntimeState::configured_datastore_ids(self))
    }

    fn application_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        TrustedRuntimeState::application_reference_status(self, reference)
    }

    fn session_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        TrustedRuntimeState::session_reference_status(self, reference)
    }

    fn datastore_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        TrustedRuntimeState::datastore_reference_status(self, reference)
    }

    fn authentication_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        TrustedRuntimeState::authentication_reference_status(self, reference)
    }

    fn remove_managed_secret(
        &self,
        reference: &ManagedSecretReference,
        proof: ManagedSecretRemovalProof,
    ) -> Result<ManagedSecretMetadata, AdministrationFailureCode> {
        TrustedRuntimeState::remove_managed_secret(self, reference, proof)
            .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
    }
}

impl<A> DatastoreCreationAuthority for std::sync::Arc<A>
where
    A: DatastoreCreationAuthority,
{
    fn deployment_id(&self) -> Result<Uuid, AdministrationFailureCode> {
        (**self).deployment_id()
    }

    fn plan(
        &self,
        datastore: &DatastoreConfigurationId,
    ) -> Result<DatastoreCreationPlan, AdministrationFailureCode> {
        (**self).plan(datastore)
    }

    fn resolve_administrator(
        &self,
        plan: &DatastoreCreationPlan,
    ) -> Result<ResolvedOperationSecret, AdministrationFailureCode> {
        (**self).resolve_administrator(plan)
    }

    fn plan_for_existing(
        &self,
        datastore: &DatastoreConfigurationId,
        datastore_id: uuid::Uuid,
    ) -> Result<DatastoreCreationPlan, AdministrationFailureCode> {
        (**self).plan_for_existing(datastore, datastore_id)
    }

    fn resolve_existing_catalog_credential(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
    ) -> Result<ResolvedOperationSecret, AdministrationFailureCode> {
        (**self).resolve_existing_catalog_credential(plan, binding)
    }

    fn resolve_storage(
        &self,
        plan: &DatastoreCreationPlan,
    ) -> Result<Vec<ResolvedCreationStorageSecret>, AdministrationFailureCode> {
        (**self).resolve_storage(plan)
    }

    fn create_catalog_credential(
        &self,
        plan: &DatastoreCreationPlan,
        material: &SecretMaterial,
    ) -> Result<CreatedCatalogCredential, AdministrationFailureCode> {
        (**self).create_catalog_credential(plan, material)
    }

    fn verify_catalog_credential(
        &self,
        plan: &DatastoreCreationPlan,
        material: &SecretMaterial,
        credential: &CreatedCatalogCredential,
    ) -> Result<(), AdministrationFailureCode> {
        (**self).verify_catalog_credential(plan, material, credential)
    }

    fn stage_catalog_credential_replacement(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
        material: &SecretMaterial,
    ) -> Result<StagedCatalogCredential, AdministrationFailureCode> {
        (**self).stage_catalog_credential_replacement(plan, binding, material)
    }

    fn activate_catalog_credential_replacement(
        &self,
        plan: &DatastoreCreationPlan,
        staged: &StagedCatalogCredential,
    ) -> Result<CreatedCatalogCredential, AdministrationFailureCode> {
        (**self).activate_catalog_credential_replacement(plan, staged)
    }

    fn discard_catalog_credential_replacement(
        &self,
        plan: &DatastoreCreationPlan,
        staged: &StagedCatalogCredential,
    ) -> Result<(), AdministrationFailureCode> {
        (**self).discard_catalog_credential_replacement(plan, staged)
    }

    fn retain_datastore_binding_reference(
        &self,
        plan: &DatastoreCreationPlan,
        reference: &ManagedSecretReference,
    ) -> Result<(), AdministrationFailureCode> {
        (**self).retain_datastore_binding_reference(plan, reference)
    }

    fn reconcilable_datastore_binding_reference_ids(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<Vec<Uuid>, ManagedSecretReferenceInspectionError> {
        (**self).reconcilable_datastore_binding_reference_ids(reference)
    }

    fn release_absent_datastore_binding_reference(
        &self,
        datastore_id: Uuid,
        reference: &ManagedSecretReference,
    ) -> Result<(), ManagedSecretReferenceInspectionError> {
        (**self).release_absent_datastore_binding_reference(datastore_id, reference)
    }

    fn postgresql_binding_inspection_targets(
        &self,
    ) -> Result<Vec<PostgresqlBindingInspectionTarget>, AdministrationFailureCode> {
        (**self).postgresql_binding_inspection_targets()
    }

    fn resolve_postgresql_binding_inspection_administrator(
        &self,
        target: &PostgresqlBindingInspectionTarget,
    ) -> Result<ResolvedOperationSecret, AdministrationFailureCode> {
        (**self).resolve_postgresql_binding_inspection_administrator(target)
    }

    fn configured_datastore_ids(
        &self,
    ) -> Result<Vec<DatastoreConfigurationId>, AdministrationFailureCode> {
        (**self).configured_datastore_ids()
    }

    fn application_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        (**self).application_reference_status(reference)
    }

    fn session_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        (**self).session_reference_status(reference)
    }

    fn datastore_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        (**self).datastore_reference_status(reference)
    }

    fn authentication_reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        (**self).authentication_reference_status(reference)
    }

    fn remove_managed_secret(
        &self,
        reference: &ManagedSecretReference,
        proof: ManagedSecretRemovalProof,
    ) -> Result<ManagedSecretMetadata, AdministrationFailureCode> {
        (**self).remove_managed_secret(reference, proof)
    }
}

/// Adapter-owned phases of the cross-store creation saga.
pub trait DatastoreCreationProvisioner: Send + Sync {
    type Pending;

    /// Creates PostgreSQL resources and persists the singleton `creating` binding.
    fn begin(
        &self,
        plan: &DatastoreCreationPlan,
        _administrator: &ResolvedOperationSecret,
        catalog_credential: &SecretMaterial,
    ) -> DatastoreCreationStart<Self::Pending>;

    fn initialize_lake(
        &self,
        pending: &mut Self::Pending,
        plan: &DatastoreCreationPlan,
        catalog_credential: &SecretMaterial,
        storage: &[ResolvedCreationStorageSecret],
    ) -> Result<(), DatastoreCreationFailure>;

    fn verify(
        &self,
        pending: &mut Self::Pending,
        plan: &DatastoreCreationPlan,
    ) -> Result<(), DatastoreCreationFailure>;

    fn mark_ready(
        &self,
        pending: &mut Self::Pending,
        credential: &CreatedCatalogCredential,
    ) -> Result<(), DatastoreCreationFailure>;

    fn mark_failed(
        &self,
        pending: &mut Self::Pending,
        failure: &DatastoreCreationFailure,
    ) -> Result<(), DatastoreCreationFailure>;

    fn find_existing(
        &self,
        _plan: &DatastoreCreationPlan,
        _administrator: &ResolvedOperationSecret,
    ) -> Result<Option<DatastoreIdentityBinding>, DatastoreCreationFailure> {
        Ok(None)
    }

    fn inspect_all_bindings(
        &self,
        _target: &PostgresqlBindingInspectionTarget,
        _administrator: &ResolvedOperationSecret,
    ) -> Result<Vec<DatastoreIdentityBinding>, DatastoreCreationFailure> {
        Err(verification_failure(()))
    }

    fn verify_existing(
        &self,
        _plan: &DatastoreCreationPlan,
        _binding: &DatastoreIdentityBinding,
        _administrator: &ResolvedOperationSecret,
        _catalog_credential: &ResolvedOperationSecret,
        _storage: &[ResolvedCreationStorageSecret],
    ) -> Result<(), DatastoreCreationFailure> {
        Err(verification_failure(()))
    }

    fn plan_governance(
        &self,
        _plan: &DatastoreCreationPlan,
        _binding: &DatastoreIdentityBinding,
        _administrator: &ResolvedOperationSecret,
        _catalog: &ResolvedOperationSecret,
        _storage: &[ResolvedCreationStorageSecret],
    ) -> Result<ahri_tre_pgmeta::governance::GovernanceUpgradePlan, DatastoreCreationFailure> {
        Err(verification_failure(()))
    }

    fn upgrade_governance(
        &self,
        _plan: &DatastoreCreationPlan,
        _binding: &DatastoreIdentityBinding,
        _administrator: &ResolvedOperationSecret,
        _catalog_credential: &ResolvedOperationSecret,
        _storage: &[ResolvedCreationStorageSecret],
        _batch: u32,
    ) -> Result<GovernanceUpgradeProgress, DatastoreCreationFailure> {
        Err(verification_failure(()))
    }

    fn seed_demo(
        &self,
        _plan: &DatastoreCreationPlan,
        _binding: &DatastoreIdentityBinding,
        _administrator: &ResolvedOperationSecret,
        _catalog_credential: &ResolvedOperationSecret,
        _storage: &[ResolvedCreationStorageSecret],
    ) -> Result<(), DatastoreCreationFailure> {
        Err(verification_failure(()))
    }

    fn apply_catalog_credential_rotation(
        &self,
        _plan: &DatastoreCreationPlan,
        _binding: &DatastoreIdentityBinding,
        _administrator: &ResolvedOperationSecret,
        _prior: &ResolvedOperationSecret,
        _staged: &StagedCatalogCredential,
        _rotated_at: DateTime<Utc>,
    ) -> Result<DatastoreIdentityBinding, DatastoreCreationFailure> {
        Err(verification_failure(()))
    }

    fn rollback_catalog_credential_rotation(
        &self,
        _plan: &DatastoreCreationPlan,
        _binding: &DatastoreIdentityBinding,
        _administrator: &ResolvedOperationSecret,
        _prior: &ResolvedOperationSecret,
    ) -> Result<(), DatastoreCreationFailure> {
        Err(verification_failure(()))
    }
}

/// Production PostgreSQL and DuckLake adapter implementation of the creation saga.
#[derive(Debug, Default, Clone, Copy)]
pub struct ConfiguredDatastoreCreationProvisioner;

pub struct PendingConfiguredDatastoreCreation {
    postgres: PgPendingDatastoreCreation,
    lake: Option<DuckLakeOpenedCatalog>,
    scratch: Option<ScratchAttempt>,
}

impl std::fmt::Debug for PendingConfiguredDatastoreCreation {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PendingConfiguredDatastoreCreation")
            .field("postgres", &self.postgres)
            .field("lake_initialized", &self.lake.is_some())
            .finish_non_exhaustive()
    }
}

#[derive(Debug)]
pub enum DatastoreCreationStart<P> {
    Started(P),
    Conflict,
    Failed(DatastoreCreationFailure),
    FailedWithBinding {
        pending: P,
        failure: DatastoreCreationFailure,
    },
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DatastoreCreationFailure {
    code: DatastoreCreationFailureCode,
    safe_summary: &'static str,
}

impl DatastoreCreationFailure {
    pub fn new(code: DatastoreCreationFailureCode, safe_summary: &'static str) -> Self {
        debug_assert!(safe_summary.len() <= 512);
        Self { code, safe_summary }
    }

    pub fn code(&self) -> DatastoreCreationFailureCode {
        self.code
    }

    pub fn safe_summary(&self) -> &'static str {
        self.safe_summary
    }

    fn administration_code(&self) -> AdministrationFailureCode {
        match self.code {
            DatastoreCreationFailureCode::PostgresqlSetupFailed => {
                AdministrationFailureCode::PostgresqlSetupFailed
            }
            DatastoreCreationFailureCode::CredentialSetupFailed => {
                AdministrationFailureCode::CredentialSetupFailed
            }
            DatastoreCreationFailureCode::LakeSetupFailed => {
                AdministrationFailureCode::LakeSetupFailed
            }
            DatastoreCreationFailureCode::VerificationFailed => {
                AdministrationFailureCode::VerificationFailed
            }
            DatastoreCreationFailureCode::InternalFailed => {
                AdministrationFailureCode::InternalFailed
            }
            DatastoreCreationFailureCode::CredentialStateUncertain => {
                AdministrationFailureCode::CredentialStateUncertain
            }
        }
    }
}

impl DatastoreCreationProvisioner for ConfiguredDatastoreCreationProvisioner {
    type Pending = PendingConfiguredDatastoreCreation;

    fn begin(
        &self,
        plan: &DatastoreCreationPlan,
        administrator: &ResolvedOperationSecret,
        catalog_credential: &SecretMaterial,
    ) -> DatastoreCreationStart<Self::Pending> {
        let (lake_path, lake_location) = match creation_lake_location(plan) {
            Ok(location) => location,
            Err(()) => return DatastoreCreationStart::Failed(postgresql_failure()),
        };
        let binding = match NewDatastoreIdentityBinding::creating_managed(
            plan.datastore_id(),
            plan.postgresql().database(),
            plan.catalog_database(),
            plan.catalog_schema(),
            lake_path,
            plan.storage_policy_id(),
            lake_location,
            EncryptionMode::ServerManaged,
            plan.catalog_role(),
            plan.catalog_secret_reference().as_str(),
            Some(env!("CARGO_PKG_VERSION").to_string()),
        ) {
            Ok(binding) => binding,
            Err(_) => return DatastoreCreationStart::Failed(postgresql_failure()),
        };
        let connection = creation_postgres_connection(plan);
        let Ok(governance) = ahri_tre_pgmeta::governance::GovernanceAppointment::from_identities(
            plan.governance_custodians()
                .iter()
                .map(|a| (a.issuer(), a.subject())),
        ) else {
            return DatastoreCreationStart::Failed(verification_failure(()));
        };
        let spec = PgDatastoreCreationSpec::new(
            plan.postgresql().database(),
            plan.metadata_role(),
            plan.catalog_database(),
            plan.catalog_schema(),
            plan.catalog_role(),
            binding,
            governance,
        );
        let started = begin_datastore_creation(
            &connection,
            administrator.material(),
            catalog_credential,
            spec,
        );
        match started {
            PgDatastoreCreationStart::Started(postgres) => {
                DatastoreCreationStart::Started(PendingConfiguredDatastoreCreation {
                    postgres,
                    lake: None,
                    scratch: None,
                })
            }
            PgDatastoreCreationStart::Conflict => DatastoreCreationStart::Conflict,
            PgDatastoreCreationStart::FailedBeforeBinding => {
                DatastoreCreationStart::Failed(postgresql_failure())
            }
            PgDatastoreCreationStart::RollbackFailed => {
                DatastoreCreationStart::Failed(internal_failure())
            }
            PgDatastoreCreationStart::FailedAfterBinding(postgres) => {
                DatastoreCreationStart::FailedWithBinding {
                    pending: PendingConfiguredDatastoreCreation {
                        postgres,
                        lake: None,
                        scratch: None,
                    },
                    failure: postgresql_failure(),
                }
            }
        }
    }

    fn initialize_lake(
        &self,
        pending: &mut Self::Pending,
        plan: &DatastoreCreationPlan,
        catalog_credential: &SecretMaterial,
        storage: &[ResolvedCreationStorageSecret],
    ) -> Result<(), DatastoreCreationFailure> {
        let catalog_tls = catalog_tls(plan.postgresql().tls());
        let (opened, scratch) = match plan.lake().storage() {
            LakeStoragePlan::Filesystem { base_path } => {
                let config = FilesystemLakeSessionConfig::new(
                    base_path,
                    plan.datastore_prefix(),
                    plan.postgresql().host(),
                    plan.postgresql().port(),
                    plan.catalog_database(),
                    plan.catalog_schema(),
                    plan.catalog_role(),
                    catalog_tls,
                    EncryptionMode::ServerManaged,
                )
                .map_err(lake_failure)?;
                let scratch = create_scratch(plan, [config.canonical_data_path()])?;
                let opened = DuckLakeAdapter::initialize_filesystem_catalog_with_scratch(
                    &config,
                    catalog_credential,
                    &scratch,
                )
                .map_err(lake_failure)?;
                (opened, scratch)
            }
            _ => {
                let location = object_creation_location(plan).map_err(lake_failure)?;
                let config = ObjectLakeSessionConfig::new(
                    location,
                    plan.postgresql().host(),
                    plan.postgresql().port(),
                    plan.catalog_database(),
                    plan.catalog_schema(),
                    plan.catalog_role(),
                    catalog_tls,
                    object_storage_tls(plan.lake().storage()),
                    EncryptionMode::ServerManaged,
                )
                .map_err(lake_failure)?;
                let scratch = create_scratch(plan, std::iter::empty::<String>())?;
                let opened =
                    initialize_object_lake(&config, catalog_credential, plan, storage, &scratch)?;
                (opened, scratch)
            }
        };
        pending.lake = Some(opened);
        pending.scratch = Some(scratch);
        Ok(())
    }

    fn verify(
        &self,
        pending: &mut Self::Pending,
        plan: &DatastoreCreationPlan,
    ) -> Result<(), DatastoreCreationFailure> {
        let bindings = read_datastore_identity_bindings(pending.postgres.store_mut())
            .map_err(verification_failure)?;
        let [binding] = bindings.as_slice() else {
            return Err(verification_failure(()));
        };
        let lake = pending
            .lake
            .as_ref()
            .ok_or_else(|| verification_failure(()))?;
        let (expected_path, expected_location) =
            creation_lake_location(plan).map_err(verification_failure)?;
        if binding.datastore_id != plan.datastore_id()
            || binding.datastore_name != plan.postgresql().database()
            || binding.ducklake_catalog_database != plan.catalog_database()
            || binding.ducklake_catalog_schema.as_deref() != Some(plan.catalog_schema())
            || binding.storage_policy_id.as_deref() != Some(plan.storage_policy_id())
            || binding.lake_location.as_ref() != Some(&expected_location)
            || binding.lake_catalog_role_name.as_deref() != Some(plan.catalog_role())
            || binding.managed_secret_ref.as_deref()
                != Some(plan.catalog_secret_reference().as_str())
            || binding.lake_catalog_credential_mode != LakeCatalogCredentialMode::ManagedLocal
            || binding.credential_version != 1
            || !binding.ducklake_encryption
            || binding.lifecycle_state != DatastoreLifecycleState::Creating
            || binding.binding_fingerprint != binding.computed_binding_fingerprint()
            || binding.lake_path != expected_path.as_str()
            || !equivalent_ducklake_data_paths(&binding.lake_path, &lake.health.data_path)
            || lake.attach_plan.data_path != expected_path.as_str()
            || lake.attach_plan.credential_kind != DuckLakeCredentialKind::SessionSecret
            || lake.attach_plan.configured_encryption_mode != EncryptionMode::ServerManaged
            || lake.health.detected_encryption_mode != EncryptionMode::ServerManaged
            || !lake.attach_plan.create_if_not_exists
            || lake.attach_plan.automatic_migration
        {
            return Err(verification_failure(()));
        }
        Ok(())
    }

    fn mark_ready(
        &self,
        pending: &mut Self::Pending,
        credential: &CreatedCatalogCredential,
    ) -> Result<(), DatastoreCreationFailure> {
        if credential.version() != 1 {
            return Err(verification_failure(()));
        }
        let bindings = read_datastore_identity_bindings(pending.postgres.store_mut())
            .map_err(verification_failure)?;
        let [binding] = bindings.as_slice() else {
            return Err(verification_failure(()));
        };
        let ledger = ahri_tre_pgmeta::governance::ledger_binding(pending.postgres.store_mut())
            .map_err(verification_failure)?;
        let lake = pending
            .lake
            .as_mut()
            .ok_or_else(|| verification_failure(()))?;
        ahri_tre_lake::GovernanceLedger::provision(
            &mut lake.connection,
            ahri_tre_types::StudyId(ledger.study_id),
        )
        .map_err(verification_failure)?;
        crate::disclosure::reconcile_governance_evidence(
            pending.postgres.store_mut(),
            &mut lake.connection,
            1000,
        )
        .map_err(verification_failure)?;
        mark_datastore_identity_ready(
            pending.postgres.store_mut(),
            binding.datastore_id,
            credential.created_at(),
        )
        .map_err(verification_failure)?;
        Ok(())
    }

    fn mark_failed(
        &self,
        pending: &mut Self::Pending,
        failure: &DatastoreCreationFailure,
    ) -> Result<(), DatastoreCreationFailure> {
        let bindings = read_datastore_identity_bindings(pending.postgres.store_mut())
            .map_err(|_| internal_failure())?;
        let [binding] = bindings.as_slice() else {
            return Err(internal_failure());
        };
        mark_datastore_identity_failed(
            pending.postgres.store_mut(),
            binding.datastore_id,
            failure.code(),
            Some(failure.safe_summary()),
        )
        .map_err(|_| internal_failure())?;
        Ok(())
    }

    fn find_existing(
        &self,
        plan: &DatastoreCreationPlan,
        administrator: &ResolvedOperationSecret,
    ) -> Result<Option<DatastoreIdentityBinding>, DatastoreCreationFailure> {
        find_existing_datastore_binding(
            &creation_postgres_connection(plan),
            administrator.material(),
            plan.postgresql().database(),
        )
        .map_err(verification_failure)
    }

    fn inspect_all_bindings(
        &self,
        target: &PostgresqlBindingInspectionTarget,
        administrator: &ResolvedOperationSecret,
    ) -> Result<Vec<DatastoreIdentityBinding>, DatastoreCreationFailure> {
        let connection = PgDatastoreCreationConnection::new(
            target.host(),
            target.port(),
            target.bootstrap_database(),
            target.administrator_username(),
            target.root_certificates().to_vec(),
        );
        inspect_datastore_identity_bindings(&connection, administrator.material())
            .map_err(|_| postgresql_failure())
    }

    fn verify_existing(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
        administrator: &ResolvedOperationSecret,
        catalog_credential: &ResolvedOperationSecret,
        storage: &[ResolvedCreationStorageSecret],
    ) -> Result<(), DatastoreCreationFailure> {
        let (expected_path, expected_location) =
            creation_lake_location(plan).map_err(verification_failure)?;
        if binding.datastore_id != plan.datastore_id()
            || binding.datastore_name != plan.postgresql().database()
            || binding.ducklake_catalog_database != plan.catalog_database()
            || binding.ducklake_catalog_schema.as_deref() != Some(plan.catalog_schema())
            || binding.storage_policy_id.as_deref() != Some(plan.storage_policy_id())
            || binding.lake_location.as_ref() != Some(&expected_location)
            || binding.lake_catalog_role_name.as_deref() != Some(plan.catalog_role())
            || binding.managed_secret_ref.as_deref()
                != Some(plan.catalog_secret_reference().as_str())
            || binding.lake_catalog_credential_mode != LakeCatalogCredentialMode::ManagedLocal
            || binding.credential_version
                != i32::try_from(catalog_credential.version().managed_version().unwrap_or(0))
                    .map_err(verification_failure)?
            || !binding.ducklake_encryption
            || binding.lifecycle_state != DatastoreLifecycleState::Ready
            || binding.failure_code.is_some()
            || binding.binding_fingerprint != binding.computed_binding_fingerprint()
            || binding.lake_path != expected_path
        {
            return Err(verification_failure(()));
        }
        let postgres_spec = PgDatastoreValidationSpec::new(
            plan.postgresql().database(),
            plan.metadata_role(),
            plan.catalog_database(),
            plan.catalog_schema(),
            plan.catalog_role(),
        );
        validate_existing_datastore_postgresql(
            &creation_postgres_connection(plan),
            administrator.material(),
            &postgres_spec,
        )
        .map_err(verification_failure)?;

        let scratch = create_scratch(plan, std::iter::once(expected_path.clone()))?;
        let opened = open_existing_lake(plan, catalog_credential.material(), storage, &scratch)?;
        if !equivalent_ducklake_data_paths(&opened.health.data_path, &expected_path)
            || opened.attach_plan.data_path != expected_path
            || opened.attach_plan.create_if_not_exists
            || opened.attach_plan.configured_encryption_mode != EncryptionMode::ServerManaged
            || opened.health.detected_encryption_mode != EncryptionMode::ServerManaged
        {
            return Err(verification_failure(()));
        }
        Ok(())
    }

    fn plan_governance(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
        administrator: &ResolvedOperationSecret,
        catalog: &ResolvedOperationSecret,
        storage: &[ResolvedCreationStorageSecret],
    ) -> Result<ahri_tre_pgmeta::governance::GovernanceUpgradePlan, DatastoreCreationFailure> {
        self.verify_existing(plan, binding, administrator, catalog, storage)?;
        creation_postgres_connection(plan)
            .dataset_maintenance(plan.postgresql().database(), administrator.material())
            .map_err(verification_failure)?
            .governance()
            .upgrade_plan()
            .map_err(verification_failure)
    }

    fn upgrade_governance(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
        administrator: &ResolvedOperationSecret,
        catalog_credential: &ResolvedOperationSecret,
        storage: &[ResolvedCreationStorageSecret],
        batch: u32,
    ) -> Result<GovernanceUpgradeProgress, DatastoreCreationFailure> {
        ConfiguredDatastoreCreationProvisioner::upgrade_governance(
            self,
            plan,
            binding,
            administrator,
            catalog_credential,
            storage,
            batch,
        )
    }

    fn seed_demo(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
        administrator: &ResolvedOperationSecret,
        catalog_credential: &ResolvedOperationSecret,
        storage: &[ResolvedCreationStorageSecret],
    ) -> Result<(), DatastoreCreationFailure> {
        let (expected_path, _) = creation_lake_location(plan).map_err(verification_failure)?;
        let scratch = create_scratch(plan, std::iter::once(expected_path.clone()))?;
        let executor_root = plan
            .dataset_executor_root()
            .ok_or_else(|| verification_failure(()))?;
        let executor = ahri_tre_lake::DatasetExecutor::acquire(
            std::path::Path::new(executor_root),
            binding.datastore_id,
        )
        .map_err(verification_failure)?;
        let maintenance = creation_postgres_connection(plan)
            .dataset_maintenance(plan.postgresql().database(), administrator.material())
            .map_err(verification_failure)?;
        let mut exclusion = maintenance
            .authorize_namespace_reset(executor)
            .map_err(verification_failure)?;
        let opened = open_existing_lake(plan, catalog_credential.material(), storage, &scratch)?;
        if binding.datastore_name != plan.postgresql().database() {
            return Err(verification_failure(()));
        }

        let study_id = StudyId(Uuid::parse_str(DEMO_STUDY_ID).map_err(verification_failure)?);
        let adapter = DuckLakeAdapter::new(expected_path);
        let seeds = reviewed_demo_dataset_seeds().map_err(verification_failure)?;
        adapter
            .replace_server_managed_dataset_tables_from_csv(
                &opened,
                &scratch,
                study_id,
                &seeds,
                &mut exclusion,
            )
            .map(|_| ())
            .map_err(lake_failure)
    }

    fn apply_catalog_credential_rotation(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
        administrator: &ResolvedOperationSecret,
        prior: &ResolvedOperationSecret,
        staged: &StagedCatalogCredential,
        rotated_at: DateTime<Utc>,
    ) -> Result<DatastoreIdentityBinding, DatastoreCreationFailure> {
        let version = i32::try_from(staged.version()).map_err(|_| postgresql_failure())?;
        apply_datastore_credential_rotation(
            &creation_postgres_connection(plan),
            administrator.material(),
            prior.material(),
            staged.material(),
            plan.postgresql().database(),
            plan.catalog_database(),
            plan.catalog_role(),
            binding,
            version,
            rotated_at,
        )
        .map_err(|error| match error {
            PgDatastoreCredentialRotationError::Restored => postgresql_failure(),
            PgDatastoreCredentialRotationError::StateUncertain => credential_state_uncertain(),
        })
    }

    fn rollback_catalog_credential_rotation(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
        administrator: &ResolvedOperationSecret,
        prior: &ResolvedOperationSecret,
    ) -> Result<(), DatastoreCreationFailure> {
        rollback_datastore_credential_rotation(
            &creation_postgres_connection(plan),
            administrator.material(),
            prior.material(),
            plan.postgresql().database(),
            plan.catalog_role(),
            plan.catalog_database(),
            binding,
        )
        .map_err(|error| match error {
            PgDatastoreCredentialRotationError::Restored => postgresql_failure(),
            PgDatastoreCredentialRotationError::StateUncertain => credential_state_uncertain(),
        })
    }
}

fn creation_postgres_connection(plan: &DatastoreCreationPlan) -> PgDatastoreCreationConnection {
    let root_certificates = match plan.postgresql().tls() {
        SessionTls::System => Vec::new(),
        SessionTls::CustomCa { certificates } => certificates.clone(),
    };
    PgDatastoreCreationConnection::new(
        plan.postgresql().host(),
        plan.postgresql().port(),
        plan.postgresql().bootstrap_database(),
        plan.administrator().username(),
        root_certificates,
    )
}

fn postgresql_failure() -> DatastoreCreationFailure {
    DatastoreCreationFailure::new(
        DatastoreCreationFailureCode::PostgresqlSetupFailed,
        "PostgreSQL Datastore initialization did not complete",
    )
}

fn lake_failure(_error: impl std::fmt::Debug) -> DatastoreCreationFailure {
    DatastoreCreationFailure::new(
        DatastoreCreationFailureCode::LakeSetupFailed,
        "Lake initialization did not complete",
    )
}

fn verification_failure(_error: impl std::fmt::Debug) -> DatastoreCreationFailure {
    DatastoreCreationFailure::new(
        DatastoreCreationFailureCode::VerificationFailed,
        "Cross-store verification did not complete",
    )
}

fn internal_failure() -> DatastoreCreationFailure {
    DatastoreCreationFailure::new(
        DatastoreCreationFailureCode::InternalFailed,
        "Datastore creation state could not be persisted",
    )
}

fn credential_state_uncertain() -> DatastoreCreationFailure {
    DatastoreCreationFailure::new(
        DatastoreCreationFailureCode::CredentialStateUncertain,
        "Credential owner state requires reconciliation",
    )
}

fn creation_lake_location(
    plan: &DatastoreCreationPlan,
) -> Result<(String, DatastoreLakeLocation), ()> {
    match plan.lake().storage() {
        LakeStoragePlan::Filesystem { base_path } => {
            let config = FilesystemLakeSessionConfig::new(
                base_path,
                plan.datastore_prefix(),
                plan.postgresql().host(),
                plan.postgresql().port(),
                plan.catalog_database(),
                plan.catalog_schema(),
                plan.catalog_role(),
                catalog_tls(plan.postgresql().tls()),
                EncryptionMode::None,
            )
            .map_err(|_| ())?;
            Ok((
                config.canonical_data_path(),
                DatastoreLakeLocation::Filesystem {
                    base_path: base_path.clone(),
                    directory: plan.datastore_prefix().to_string(),
                },
            ))
        }
        storage => {
            let location = object_creation_location(plan).map_err(|_| ())?;
            let lake_path = location.canonical_data_path();
            let persisted = match (storage, location) {
                (
                    LakeStoragePlan::AwsS3 { .. },
                    ObjectStorageLocation::AwsS3 {
                        region,
                        bucket,
                        prefix,
                    },
                ) => DatastoreLakeLocation::AwsS3 {
                    region,
                    bucket,
                    prefix,
                },
                (
                    LakeStoragePlan::S3Compatible { .. },
                    ObjectStorageLocation::S3Compatible {
                        endpoint,
                        url_style,
                        bucket,
                        prefix,
                    },
                ) => DatastoreLakeLocation::S3Compatible {
                    endpoint,
                    url_style: match url_style {
                        S3UrlStyle::Path => "path",
                        S3UrlStyle::VirtualHosted => "virtual_hosted",
                    }
                    .to_string(),
                    bucket,
                    prefix,
                },
                (
                    LakeStoragePlan::AzureBlob { .. },
                    ObjectStorageLocation::AzureBlob {
                        account_endpoint,
                        container,
                        prefix,
                    },
                ) => DatastoreLakeLocation::AzureBlob {
                    account_endpoint,
                    container,
                    prefix,
                },
                _ => return Err(()),
            };
            Ok((lake_path, persisted))
        }
    }
}

fn object_creation_location(
    plan: &DatastoreCreationPlan,
) -> Result<ObjectStorageLocation, ahri_tre_lake::LakeError> {
    let location = match plan.lake().storage() {
        LakeStoragePlan::AwsS3 {
            region,
            bucket,
            base_prefix,
            ..
        } => ObjectStorageLocation::aws_s3_for_datastore(
            region,
            bucket,
            base_prefix.as_deref(),
            plan.datastore_prefix(),
        )?,
        LakeStoragePlan::S3Compatible {
            endpoint,
            url_style,
            bucket,
            base_prefix,
            ..
        } => ObjectStorageLocation::s3_compatible_for_datastore(
            endpoint,
            match url_style {
                ConfiguredS3UrlStyle::Path => S3UrlStyle::Path,
                ConfiguredS3UrlStyle::VirtualHosted => S3UrlStyle::VirtualHosted,
            },
            bucket,
            base_prefix.as_deref(),
            plan.datastore_prefix(),
        )?,
        LakeStoragePlan::AzureBlob {
            account_endpoint,
            container,
            base_prefix,
            ..
        } => ObjectStorageLocation::azure_blob_for_datastore(
            account_endpoint,
            container,
            base_prefix.as_deref(),
            plan.datastore_prefix(),
        )?,
        LakeStoragePlan::Filesystem { .. } => {
            return Err(ahri_tre_lake::LakeError::InvalidObjectStorageIdentity);
        }
    };
    Ok(location)
}

fn create_scratch(
    plan: &DatastoreCreationPlan,
    lake_paths: impl IntoIterator<Item = String>,
) -> Result<ScratchAttempt, DatastoreCreationFailure> {
    let paths = lake_paths
        .into_iter()
        .map(std::path::PathBuf::from)
        .collect::<Vec<_>>();
    let scratch = TrustedScratch::open(
        std::path::Path::new(plan.configuration_scratch_root()),
        paths.iter().map(std::path::PathBuf::as_path),
    )
    .map_err(lake_failure)?;
    let id = uuid::Uuid::new_v4().simple().to_string();
    scratch
        .create_attempt(ScratchAttemptId::new(&id).map_err(lake_failure)?)
        .map_err(lake_failure)
}

fn initialize_object_lake(
    config: &ObjectLakeSessionConfig,
    catalog_credential: &SecretMaterial,
    plan: &DatastoreCreationPlan,
    storage: &[ResolvedCreationStorageSecret],
    scratch: &ScratchAttempt,
) -> Result<DuckLakeOpenedCatalog, DatastoreCreationFailure> {
    let resolved = |configured: &ConfiguredSecret| {
        storage
            .iter()
            .find(|secret| secret.configured() == configured)
            .map(ResolvedCreationStorageSecret::resolved)
            .ok_or_else(|| lake_failure(()))
    };
    let authentication = plan
        .lake()
        .storage()
        .authentication()
        .ok_or_else(|| lake_failure(()))?;
    match authentication {
        StorageAuthenticationPlan::S3Static {
            access_key_id,
            secret_access_key,
        } => {
            let access = resolved(access_key_id)?;
            let secret = resolved(secret_access_key)?;
            DuckLakeAdapter::initialize_object_storage_catalog(
                config,
                catalog_credential,
                ObjectStorageAuthentication::S3Static {
                    access_key_id: access.material(),
                    secret_access_key: secret.material(),
                },
                scratch,
            )
        }
        StorageAuthenticationPlan::AwsWorkloadIdentity { source } => {
            let source = match source {
                ConfiguredAwsWorkloadIdentitySource::EcsTask => AwsWorkloadIdentitySource::EcsTask,
                ConfiguredAwsWorkloadIdentitySource::Ec2Instance => {
                    AwsWorkloadIdentitySource::Ec2Instance
                }
            };
            DuckLakeAdapter::initialize_object_storage_catalog(
                config,
                catalog_credential,
                ObjectStorageAuthentication::AwsWorkloadIdentity { source },
                scratch,
            )
        }
        StorageAuthenticationPlan::AzureManagedIdentity { identity } => {
            let client_id = match identity {
                AzureManagedIdentity::SystemAssigned => None,
                AzureManagedIdentity::UserAssigned { client_id } => Some(*client_id),
            };
            DuckLakeAdapter::initialize_object_storage_catalog(
                config,
                catalog_credential,
                ObjectStorageAuthentication::AzureManagedIdentity { client_id },
                scratch,
            )
        }
        StorageAuthenticationPlan::AzureServicePrincipal {
            tenant_id,
            client_id,
            client_secret,
        } => {
            let secret = resolved(client_secret)?;
            DuckLakeAdapter::initialize_object_storage_catalog(
                config,
                catalog_credential,
                ObjectStorageAuthentication::AzureServicePrincipal {
                    tenant_id: *tenant_id,
                    client_id: *client_id,
                    client_secret: secret.material(),
                },
                scratch,
            )
        }
    }
    .map_err(lake_failure)
}

fn open_existing_lake(
    plan: &DatastoreCreationPlan,
    catalog_credential: &SecretMaterial,
    storage: &[ResolvedCreationStorageSecret],
    scratch: &ScratchAttempt,
) -> Result<DuckLakeOpenedCatalog, DatastoreCreationFailure> {
    let catalog_tls = catalog_tls(plan.postgresql().tls());
    match plan.lake().storage() {
        LakeStoragePlan::Filesystem { base_path } => {
            let config = FilesystemLakeSessionConfig::new(
                base_path,
                plan.datastore_prefix(),
                plan.postgresql().host(),
                plan.postgresql().port(),
                plan.catalog_database(),
                plan.catalog_schema(),
                plan.catalog_role(),
                catalog_tls,
                EncryptionMode::ServerManaged,
            )
            .map_err(lake_failure)?;
            DuckLakeAdapter::open_existing_filesystem_catalog_with_scratch(
                &config,
                catalog_credential,
                scratch,
            )
            .map_err(lake_failure)
        }
        _ => {
            let location = object_creation_location(plan).map_err(lake_failure)?;
            let config = ObjectLakeSessionConfig::new(
                location,
                plan.postgresql().host(),
                plan.postgresql().port(),
                plan.catalog_database(),
                plan.catalog_schema(),
                plan.catalog_role(),
                catalog_tls,
                object_storage_tls(plan.lake().storage()),
                EncryptionMode::ServerManaged,
            )
            .map_err(lake_failure)?;
            open_existing_object_lake(&config, catalog_credential, plan, storage, scratch)
        }
    }
}

fn open_existing_object_lake(
    config: &ObjectLakeSessionConfig,
    catalog_credential: &SecretMaterial,
    plan: &DatastoreCreationPlan,
    storage: &[ResolvedCreationStorageSecret],
    scratch: &ScratchAttempt,
) -> Result<DuckLakeOpenedCatalog, DatastoreCreationFailure> {
    let resolved = |configured: &ConfiguredSecret| {
        storage
            .iter()
            .find(|secret| secret.configured() == configured)
            .map(ResolvedCreationStorageSecret::resolved)
            .ok_or_else(|| lake_failure(()))
    };
    let authentication = plan
        .lake()
        .storage()
        .authentication()
        .ok_or_else(|| lake_failure(()))?;
    match authentication {
        StorageAuthenticationPlan::S3Static {
            access_key_id,
            secret_access_key,
        } => {
            let access = resolved(access_key_id)?;
            let secret = resolved(secret_access_key)?;
            DuckLakeAdapter::open_existing_object_storage_catalog(
                config,
                catalog_credential,
                ObjectStorageAuthentication::S3Static {
                    access_key_id: access.material(),
                    secret_access_key: secret.material(),
                },
                scratch,
            )
        }
        StorageAuthenticationPlan::AwsWorkloadIdentity { source } => {
            let source = match source {
                ConfiguredAwsWorkloadIdentitySource::EcsTask => AwsWorkloadIdentitySource::EcsTask,
                ConfiguredAwsWorkloadIdentitySource::Ec2Instance => {
                    AwsWorkloadIdentitySource::Ec2Instance
                }
            };
            DuckLakeAdapter::open_existing_object_storage_catalog(
                config,
                catalog_credential,
                ObjectStorageAuthentication::AwsWorkloadIdentity { source },
                scratch,
            )
        }
        StorageAuthenticationPlan::AzureManagedIdentity { identity } => {
            let client_id = match identity {
                AzureManagedIdentity::SystemAssigned => None,
                AzureManagedIdentity::UserAssigned { client_id } => Some(*client_id),
            };
            DuckLakeAdapter::open_existing_object_storage_catalog(
                config,
                catalog_credential,
                ObjectStorageAuthentication::AzureManagedIdentity { client_id },
                scratch,
            )
        }
        StorageAuthenticationPlan::AzureServicePrincipal {
            tenant_id,
            client_id,
            client_secret,
        } => {
            let secret = resolved(client_secret)?;
            DuckLakeAdapter::open_existing_object_storage_catalog(
                config,
                catalog_credential,
                ObjectStorageAuthentication::AzureServicePrincipal {
                    tenant_id: *tenant_id,
                    client_id: *client_id,
                    client_secret: secret.material(),
                },
                scratch,
            )
        }
    }
    .map_err(lake_failure)
}

fn catalog_tls(tls: &SessionTls) -> FilesystemCatalogTls {
    match tls {
        SessionTls::System => FilesystemCatalogTls::System,
        SessionTls::CustomCa { certificates } => FilesystemCatalogTls::CustomCa {
            certificates: certificates.clone(),
        },
    }
}

fn object_storage_tls(storage: &LakeStoragePlan) -> ObjectStorageTls {
    match storage {
        LakeStoragePlan::S3Compatible { tls, .. } => match tls {
            StorageTlsPlan::System => ObjectStorageTls::System,
            StorageTlsPlan::CustomCa { certificates } => ObjectStorageTls::CustomCa {
                certificates: certificates.clone(),
            },
        },
        _ => ObjectStorageTls::System,
    }
}

enum RemovalReferenceAuthority<'a, A, P> {
    Application(&'a A),
    DatastoreBindings {
        authority: &'a A,
        provisioner: &'a P,
    },
    SessionRecords(&'a A),
    AuthenticationArtifacts(&'a A),
}

impl<A, P> ManagedSecretReferenceAuthority for RemovalReferenceAuthority<'_, A, P>
where
    A: DatastoreCreationAuthority,
    P: DatastoreCreationProvisioner,
{
    fn reference_status(
        &self,
        reference: &ManagedSecretReference,
    ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
        match self {
            Self::Application(authority) => authority.application_reference_status(reference),
            Self::SessionRecords(authority) => authority.session_reference_status(reference),
            Self::AuthenticationArtifacts(authority) => {
                authority.authentication_reference_status(reference)
            }
            Self::DatastoreBindings {
                authority,
                provisioner,
            } => {
                let durable_status = authority.datastore_reference_status(reference)?;
                let targets = authority
                    .postgresql_binding_inspection_targets()
                    .map_err(|_| ManagedSecretReferenceInspectionError::unavailable())?;
                let mut bindings = Vec::new();
                for target in targets {
                    let administrator = authority
                        .resolve_postgresql_binding_inspection_administrator(&target)
                        .map_err(|_| ManagedSecretReferenceInspectionError::unavailable())?;
                    bindings.extend(
                        provisioner
                            .inspect_all_bindings(&target, &administrator)
                            .map_err(|_| ManagedSecretReferenceInspectionError::unavailable())?,
                    );
                }
                if bindings.iter().any(|binding| {
                    binding.managed_secret_ref.as_deref()
                        == Some(reference.as_secret_reference().as_str())
                }) {
                    return Ok(ManagedSecretReferenceStatus::Referenced);
                }
                if durable_status == ManagedSecretReferenceStatus::Referenced {
                    for datastore_id in
                        authority.reconcilable_datastore_binding_reference_ids(reference)?
                    {
                        if !bindings
                            .iter()
                            .any(|binding| binding.datastore_id == datastore_id)
                        {
                            authority.release_absent_datastore_binding_reference(
                                datastore_id,
                                reference,
                            )?;
                        }
                    }
                }
                authority.datastore_reference_status(reference)
            }
        }
    }
}

/// Authenticated handler that turns logical create intent into one durable saga.
pub struct PrivilegedDatastoreCreationHandler<A, P> {
    authority: A,
    provisioner: P,
}

impl<A, P> PrivilegedDatastoreCreationHandler<A, P> {
    pub fn new(authority: A, provisioner: P) -> Self {
        Self {
            authority,
            provisioner,
        }
    }
}

impl<A, P> AdministrationRequestHandler for PrivilegedDatastoreCreationHandler<A, P>
where
    A: DatastoreCreationAuthority,
    P: DatastoreCreationProvisioner,
{
    fn handle(
        &self,
        peer: AdministrationPeerIdentity,
        request: AdministrationRequest,
    ) -> AdministrationResponse {
        match request {
            AdministrationRequest::DatastoreCreate { datastore } => self.create(peer, &datastore),
            AdministrationRequest::DatastoreReconcile { datastore } => {
                self.reconcile(peer, &datastore)
            }
            AdministrationRequest::DatastoreUpgradeGovernance {
                datastore,
                batch,
                plan_only,
            } => self.upgrade_governance(peer, &datastore, batch, plan_only),
            AdministrationRequest::DatastoreSeedDemo { datastore } => {
                self.seed_demo(peer, &datastore)
            }
            AdministrationRequest::DatastoreRotateCredential { datastore } => {
                self.rotate_credential(peer, &datastore)
            }
            AdministrationRequest::ManagedSecretRemove {
                deployment_id,
                reference,
            } => self.remove_secret(peer, deployment_id, &reference),
        }
    }
}

impl<A, P> PrivilegedDatastoreCreationHandler<A, P>
where
    A: DatastoreCreationAuthority,
    P: DatastoreCreationProvisioner,
{
    fn remove_secret(
        &self,
        peer: AdministrationPeerIdentity,
        deployment_id: Uuid,
        reference: &ManagedSecretReference,
    ) -> AdministrationResponse {
        let failed = |code: AdministrationFailureCode| {
            audit_secret_outcome(peer, reference, code.as_str());
            AdministrationResponse::managed_secret_removal_failed(reference.as_str(), code)
        };
        if self.authority.deployment_id() != Ok(deployment_id) {
            return failed(AdministrationFailureCode::ConfigurationUnavailable);
        }
        let application: RemovalReferenceAuthority<'_, A, P> =
            RemovalReferenceAuthority::Application(&self.authority);
        let datastores = RemovalReferenceAuthority::DatastoreBindings {
            authority: &self.authority,
            provisioner: &self.provisioner,
        };
        let sessions: RemovalReferenceAuthority<'_, A, P> =
            RemovalReferenceAuthority::SessionRecords(&self.authority);
        let authentication: RemovalReferenceAuthority<'_, A, P> =
            RemovalReferenceAuthority::AuthenticationArtifacts(&self.authority);
        let proof = match ManagedSecretReferenceAuthorities::new(
            &application,
            &datastores,
            &sessions,
            &authentication,
        )
        .prove_unreferenced(reference)
        {
            Ok(proof) => proof,
            Err(error) => {
                return failed(match error.kind() {
                    ManagedSecretRemovalProofErrorKind::Referenced => {
                        AdministrationFailureCode::ReferencePresent
                    }
                    ManagedSecretRemovalProofErrorKind::InspectionUnavailable => {
                        AdministrationFailureCode::ReferenceInspectionUnavailable
                    }
                });
            }
        };
        let removed = match self.authority.remove_managed_secret(reference, proof) {
            Ok(metadata) => metadata,
            Err(_) => return failed(AdministrationFailureCode::SecretRemovalFailed),
        };
        audit_secret_outcome(peer, reference, "removed");
        AdministrationResponse::managed_secret_removed(reference.as_str(), removed.version().get())
    }
    fn upgrade_governance(
        &self,
        peer: AdministrationPeerIdentity,
        datastore: &DatastoreConfigurationId,
        batch: u32,
        plan_only: bool,
    ) -> AdministrationResponse {
        let failed = |code: AdministrationFailureCode| {
            audit_outcome(
                peer,
                "datastore_upgrade_governance",
                datastore,
                code.as_str(),
            );
            AdministrationResponse::datastore_failed(datastore.as_str(), code)
        };
        if batch == 0 || batch > 1000 {
            return failed(AdministrationFailureCode::ConfigurationUnavailable);
        }
        let run = || -> Result<_, AdministrationFailureCode> {
            let initial = self.authority.plan(datastore)?;
            let administrator = self.authority.resolve_administrator(&initial)?;
            let binding = self
                .provisioner
                .find_existing(&initial, &administrator)
                .map_err(|e| e.administration_code())?
                .ok_or(AdministrationFailureCode::VerificationFailed)?;
            let plan = self
                .authority
                .plan_for_existing(datastore, binding.datastore_id)?;
            let catalog = self
                .authority
                .resolve_existing_catalog_credential(&plan, &binding)?;
            let storage = self.authority.resolve_storage(&plan)?;
            if plan_only {
                if plan.governance_custodians().is_empty() {
                    return Err(AdministrationFailureCode::ConfigurationUnavailable);
                }
                let p = self
                    .provisioner
                    .plan_governance(&plan, &binding, &administrator, &catalog, &storage)
                    .map_err(|e| e.administration_code())?;
                return Ok(AdministrationResponse::GovernancePlanned {
                    datastore: datastore.as_str().into(),
                    datastore_id: binding.datastore_id,
                    prepared: p.prepared,
                    unclassified_assets: p.unclassified_assets,
                    immutable_versions: p.immutable_versions,
                    custodians: plan.governance_custodians().to_vec(),
                });
            }
            let progress = self
                .provisioner
                .upgrade_governance(&plan, &binding, &administrator, &catalog, &storage, batch)
                .map_err(|e| e.administration_code())?;
            Ok(AdministrationResponse::GovernanceUpgraded {
                datastore: datastore.as_str().into(),
                datastore_id: binding.datastore_id,
                converted_assets: progress.converted_assets,
                remaining_unclassified_assets: progress.remaining_unclassified_assets,
                bound_datasets: progress.bound_datasets,
                bound_datafiles: progress.bound_datafiles,
                remaining_dataset_bindings: progress.remaining_dataset_bindings,
                remaining_datafile_bindings: progress.remaining_datafile_bindings,
                remaining_evidence: progress.remaining_evidence,
            })
        };
        match run() {
            Ok(response) => {
                audit_outcome(
                    peer,
                    "datastore_upgrade_governance",
                    datastore,
                    if plan_only {
                        "planned"
                    } else {
                        "batch_complete"
                    },
                );
                response
            }
            Err(code) => failed(code),
        }
    }

    fn seed_demo(
        &self,
        peer: AdministrationPeerIdentity,
        datastore: &DatastoreConfigurationId,
    ) -> AdministrationResponse {
        let failed = |code: AdministrationFailureCode| {
            audit_outcome(peer, "datastore_seed_demo", datastore, code.as_str());
            AdministrationResponse::datastore_failed(datastore.as_str(), code)
        };
        if datastore.as_str() != DEMO_DATASTORE {
            return failed(AdministrationFailureCode::ConfigurationUnavailable);
        }
        if let response @ AdministrationResponse::Failed { .. } = self.reconcile(peer, datastore) {
            return response;
        }
        let initial_plan = match self.authority.plan(datastore) {
            Ok(plan) => plan,
            Err(code) => return failed(code),
        };
        let administrator = match self.authority.resolve_administrator(&initial_plan) {
            Ok(secret) => secret,
            Err(code) => return failed(code),
        };
        let binding = match self
            .provisioner
            .find_existing(&initial_plan, &administrator)
        {
            Ok(Some(binding)) => binding,
            Ok(None) => return failed(AdministrationFailureCode::VerificationFailed),
            Err(error) => return failed(error.administration_code()),
        };
        let plan = match self
            .authority
            .plan_for_existing(datastore, binding.datastore_id)
        {
            Ok(plan) => plan,
            Err(code) => return failed(code),
        };
        let catalog_credential = match self
            .authority
            .resolve_existing_catalog_credential(&plan, &binding)
        {
            Ok(secret) => secret,
            Err(code) => return failed(code),
        };
        let storage = match self.authority.resolve_storage(&plan) {
            Ok(storage) => storage,
            Err(code) => return failed(code),
        };
        if let Err(error) = self.provisioner.seed_demo(
            &plan,
            &binding,
            &administrator,
            &catalog_credential,
            &storage,
        ) {
            return failed(error.administration_code());
        }
        audit_outcome(peer, "datastore_seed_demo", datastore, "ready");
        AdministrationResponse::datastore_ready(datastore.as_str(), binding.datastore_id)
    }

    fn reconcile(
        &self,
        peer: AdministrationPeerIdentity,
        datastore: &DatastoreConfigurationId,
    ) -> AdministrationResponse {
        let failed = |code: AdministrationFailureCode| {
            audit_outcome(peer, "datastore_reconcile", datastore, code.as_str());
            AdministrationResponse::datastore_failed(datastore.as_str(), code)
        };
        let initial_plan = match self.authority.plan(datastore) {
            Ok(plan) => plan,
            Err(code) => return failed(code),
        };
        let administrator = match self.authority.resolve_administrator(&initial_plan) {
            Ok(secret) => secret,
            Err(code) => return failed(code),
        };
        let existing = match self
            .provisioner
            .find_existing(&initial_plan, &administrator)
        {
            Ok(existing) => existing,
            Err(error) => return failed(error.administration_code()),
        };
        let Some(binding) = existing else {
            return self.create(peer, datastore);
        };
        let plan = match self
            .authority
            .plan_for_existing(datastore, binding.datastore_id)
        {
            Ok(plan) => plan,
            Err(code) => return failed(code),
        };
        let catalog_credential = match self
            .authority
            .resolve_existing_catalog_credential(&plan, &binding)
        {
            Ok(secret) => secret,
            Err(code) => return failed(code),
        };
        let storage = match self.authority.resolve_storage(&plan) {
            Ok(storage) => storage,
            Err(code) => return failed(code),
        };
        if let Err(error) = self.provisioner.verify_existing(
            &plan,
            &binding,
            &administrator,
            &catalog_credential,
            &storage,
        ) {
            return failed(error.administration_code());
        }
        if let Err(code) = self
            .authority
            .retain_datastore_binding_reference(&plan, plan.catalog_secret_reference())
        {
            return failed(code);
        }
        audit_outcome(peer, "datastore_reconcile", datastore, "ready");
        AdministrationResponse::datastore_ready(datastore.as_str(), binding.datastore_id)
    }

    fn rotate_credential(
        &self,
        peer: AdministrationPeerIdentity,
        datastore: &DatastoreConfigurationId,
    ) -> AdministrationResponse {
        let failed = |code: AdministrationFailureCode| {
            audit_outcome(
                peer,
                "datastore_rotate_credential",
                datastore,
                code.as_str(),
            );
            AdministrationResponse::datastore_failed(datastore.as_str(), code)
        };
        let initial_plan = match self.authority.plan(datastore) {
            Ok(plan) => plan,
            Err(code) => return failed(code),
        };
        let administrator = match self.authority.resolve_administrator(&initial_plan) {
            Ok(secret) => secret,
            Err(code) => return failed(code),
        };
        let binding = match self
            .provisioner
            .find_existing(&initial_plan, &administrator)
        {
            Ok(Some(binding)) if binding.lifecycle_state == DatastoreLifecycleState::Ready => {
                binding
            }
            Ok(_) => return failed(AdministrationFailureCode::VerificationFailed),
            Err(error) => return failed(error.administration_code()),
        };
        let plan = match self
            .authority
            .plan_for_existing(datastore, binding.datastore_id)
        {
            Ok(plan) => plan,
            Err(code) => return failed(code),
        };
        if let Err(code) = self
            .authority
            .retain_datastore_binding_reference(&plan, plan.catalog_secret_reference())
        {
            return failed(code);
        }
        let prior = match self
            .authority
            .resolve_existing_catalog_credential(&plan, &binding)
        {
            Ok(secret) => secret,
            Err(code) => return failed(code),
        };
        let generated = match generate_catalog_credential() {
            Ok(secret) => secret,
            Err(()) => return failed(AdministrationFailureCode::InternalFailed),
        };
        let staged = match self
            .authority
            .stage_catalog_credential_replacement(&plan, &binding, &generated)
        {
            Ok(staged) => staged,
            Err(code) => return failed(code),
        };
        let rotated_at = Utc::now();
        let externally_applied = match self.provisioner.apply_catalog_credential_rotation(
            &plan,
            &binding,
            &administrator,
            &prior,
            &staged,
            rotated_at,
        ) {
            Ok(binding) => binding,
            Err(error) => {
                if error.administration_code()
                    == AdministrationFailureCode::CredentialStateUncertain
                {
                    return failed(AdministrationFailureCode::CredentialStateUncertain);
                }
                if self
                    .authority
                    .discard_catalog_credential_replacement(&plan, &staged)
                    .is_err()
                {
                    return failed(AdministrationFailureCode::InternalFailed);
                }
                return failed(error.administration_code());
            }
        };
        if i32::try_from(staged.version()).ok() != Some(externally_applied.credential_version) {
            let rollback = self.provisioner.rollback_catalog_credential_rotation(
                &plan,
                &binding,
                &administrator,
                &prior,
            );
            if rollback.is_err() {
                return failed(AdministrationFailureCode::CredentialStateUncertain);
            }
            return failed(
                if self
                    .authority
                    .discard_catalog_credential_replacement(&plan, &staged)
                    .is_ok()
                {
                    AdministrationFailureCode::VerificationFailed
                } else {
                    AdministrationFailureCode::InternalFailed
                },
            );
        }
        if let Err(code) = self
            .authority
            .activate_catalog_credential_replacement(&plan, &staged)
        {
            if code == AdministrationFailureCode::CredentialStateUncertain {
                return failed(code);
            }
            let rollback = self.provisioner.rollback_catalog_credential_rotation(
                &plan,
                &binding,
                &administrator,
                &prior,
            );
            let discard = self
                .authority
                .discard_catalog_credential_replacement(&plan, &staged);
            return failed(if rollback.is_ok() && discard.is_ok() {
                code
            } else {
                AdministrationFailureCode::CredentialStateUncertain
            });
        }
        audit_outcome(peer, "datastore_rotate_credential", datastore, "ready");
        AdministrationResponse::datastore_ready(datastore.as_str(), binding.datastore_id)
    }

    fn create(
        &self,
        peer: AdministrationPeerIdentity,
        datastore: &DatastoreConfigurationId,
    ) -> AdministrationResponse {
        let failed = |code: AdministrationFailureCode| {
            audit_outcome(peer, "datastore_create", datastore, code.as_str());
            AdministrationResponse::datastore_failed(datastore.as_str(), code)
        };
        let plan = match self.authority.plan(datastore) {
            Ok(plan) => plan,
            Err(code) => return failed(code),
        };
        let administrator = match self.authority.resolve_administrator(&plan) {
            Ok(secret) => secret,
            Err(code) => return failed(code),
        };
        let storage = match self.authority.resolve_storage(&plan) {
            Ok(storage) => storage,
            Err(code) => return failed(code),
        };
        let catalog_credential = match generate_catalog_credential() {
            Ok(secret) => secret,
            Err(()) => return failed(AdministrationFailureCode::InternalFailed),
        };

        let mut pending = match self
            .provisioner
            .begin(&plan, &administrator, &catalog_credential)
        {
            DatastoreCreationStart::Started(pending) => pending,
            DatastoreCreationStart::Conflict => {
                return failed(AdministrationFailureCode::Conflict);
            }
            DatastoreCreationStart::Failed(error) => {
                return failed(error.administration_code());
            }
            DatastoreCreationStart::FailedWithBinding {
                mut pending,
                failure,
            } => {
                if self
                    .provisioner
                    .mark_failed(&mut pending, &failure)
                    .is_err()
                {
                    return failed(AdministrationFailureCode::InternalFailed);
                }
                return failed(failure.administration_code());
            }
        };
        let created = match self
            .authority
            .create_catalog_credential(&plan, &catalog_credential)
        {
            Ok(metadata) => metadata,
            Err(_) => {
                let error = DatastoreCreationFailure::new(
                    DatastoreCreationFailureCode::CredentialSetupFailed,
                    "Managed catalog credential was not activated",
                );
                if self.provisioner.mark_failed(&mut pending, &error).is_err() {
                    return failed(AdministrationFailureCode::InternalFailed);
                }
                return failed(error.administration_code());
            }
        };

        if self
            .authority
            .verify_catalog_credential(&plan, &catalog_credential, &created)
            .is_err()
        {
            let error = DatastoreCreationFailure::new(
                DatastoreCreationFailureCode::CredentialSetupFailed,
                "Managed catalog credential verification did not complete",
            );
            if self.provisioner.mark_failed(&mut pending, &error).is_err() {
                return failed(AdministrationFailureCode::InternalFailed);
            }
            return failed(error.administration_code());
        }

        if let Err(error) =
            self.provisioner
                .initialize_lake(&mut pending, &plan, &catalog_credential, &storage)
        {
            if self.provisioner.mark_failed(&mut pending, &error).is_err() {
                return failed(AdministrationFailureCode::InternalFailed);
            }
            return failed(error.administration_code());
        }
        if let Err(error) = self.provisioner.verify(&mut pending, &plan) {
            if self.provisioner.mark_failed(&mut pending, &error).is_err() {
                return failed(AdministrationFailureCode::InternalFailed);
            }
            return failed(error.administration_code());
        }
        if let Err(code) = self
            .authority
            .retain_datastore_binding_reference(&plan, plan.catalog_secret_reference())
        {
            let error = DatastoreCreationFailure::new(
                DatastoreCreationFailureCode::CredentialSetupFailed,
                "durable Datastore credential ownership could not be recorded",
            );
            if self.provisioner.mark_failed(&mut pending, &error).is_err() {
                return failed(AdministrationFailureCode::InternalFailed);
            }
            return failed(code);
        }
        if let Err(error) = self.provisioner.mark_ready(&mut pending, &created) {
            if self.provisioner.mark_failed(&mut pending, &error).is_err() {
                return failed(AdministrationFailureCode::InternalFailed);
            }
            return failed(error.administration_code());
        }

        audit_outcome(peer, "datastore_create", datastore, "ready");
        AdministrationResponse::datastore_ready(datastore.as_str(), plan.datastore_id())
    }
}

fn reviewed_demo_dataset_seeds() -> Result<Vec<DatasetCsvSeed<'static>>, ()> {
    const FIXTURES: [(&str, &[u8]); 12] = [
        ("entities", include_bytes!("../../../data/entities.csv")),
        ("episodes", include_bytes!("../../../data/episodes.csv")),
        (
            "household_map",
            include_bytes!("../../../data/household_map.csv"),
        ),
        (
            "householdmemberships",
            include_bytes!("../../../data/householdmemberships.csv"),
        ),
        (
            "householdresidences",
            include_bytes!("../../../data/householdresidences.csv"),
        ),
        ("households", include_bytes!("../../../data/households.csv")),
        (
            "individual_map",
            include_bytes!("../../../data/individual_map.csv"),
        ),
        (
            "individualresidences",
            include_bytes!("../../../data/individualresidences.csv"),
        ),
        (
            "individuals",
            include_bytes!("../../../data/individuals.csv"),
        ),
        (
            "location_map",
            include_bytes!("../../../data/location_map.csv"),
        ),
        ("locations", include_bytes!("../../../data/locations.csv")),
        (
            "transformations",
            include_bytes!("../../../data/transformations.csv"),
        ),
    ];

    FIXTURES
        .into_iter()
        .map(|(name, contents)| {
            Ok(DatasetCsvSeed {
                dataset_name: NcName::parse(name).map_err(|_| ())?,
                contents,
            })
        })
        .collect()
}

fn audit_secret_outcome(
    peer: AdministrationPeerIdentity,
    reference: &ManagedSecretReference,
    outcome: &'static str,
) {
    tracing::info!(
        peer_uid = peer.uid(),
        peer_gid = peer.gid(),
        peer_pid = peer.pid(),
        command = "managed_secret_remove",
        reference = reference.as_str(),
        outcome,
        "privileged administration command completed"
    );
}
fn audit_outcome(
    peer: AdministrationPeerIdentity,
    command: &'static str,
    datastore: &DatastoreConfigurationId,
    outcome: &'static str,
) {
    tracing::info!(
        peer_uid = peer.uid(),
        peer_gid = peer.gid(),
        peer_pid = peer.pid(),
        command,
        datastore = datastore.as_str(),
        outcome,
        "privileged administration command completed"
    );
}

fn generate_catalog_credential() -> Result<SecretMaterial, ()> {
    let encoded = format!(
        "tre-{}-{}-{}",
        uuid::Uuid::new_v4().simple(),
        uuid::Uuid::new_v4().simple(),
        uuid::Uuid::new_v4().simple()
    );
    SecretMaterial::try_from(encoded.into_bytes()).map_err(|_| ())
}

fn equivalent_ducklake_data_paths(left: &str, right: &str) -> bool {
    left == right || left.trim_end_matches('/') == right.trim_end_matches('/')
}

/// One resumable operator batch. Safe metadata remains available while any of
/// these prerequisites is pending; ordinary Session opens cannot invoke it.
#[derive(Debug)]
pub struct GovernanceUpgradeProgress {
    pub converted_assets: u64,
    pub remaining_unclassified_assets: u64,
    pub bound_datasets: usize,
    pub bound_datafiles: usize,
    pub remaining_datafile_bindings: bool,
    pub remaining_dataset_bindings: bool,
    pub remaining_evidence: bool,
}
impl ConfiguredDatastoreCreationProvisioner {
    pub fn upgrade_governance(
        &self,
        plan: &DatastoreCreationPlan,
        binding: &DatastoreIdentityBinding,
        administrator: &ResolvedOperationSecret,
        catalog_credential: &ResolvedOperationSecret,
        storage: &[ResolvedCreationStorageSecret],
        batch: u32,
    ) -> Result<GovernanceUpgradeProgress, DatastoreCreationFailure> {
        if batch == 0 || batch > 1000 {
            return Err(verification_failure(()));
        }
        self.verify_existing(plan, binding, administrator, catalog_credential, storage)?;
        let coordinator = plan
            .dataset_executor_root()
            .ok_or_else(|| verification_failure(()))?;
        let executor = ahri_tre_lake::DatasetExecutor::acquire(
            std::path::Path::new(coordinator),
            binding.datastore_id,
        )
        .map_err(verification_failure)?;
        let maintenance = creation_postgres_connection(plan)
            .dataset_maintenance(plan.postgresql().database(), administrator.material())
            .map_err(verification_failure)?;
        let _exclusion = maintenance
            .authorize_namespace_reset(executor)
            .map_err(verification_failure)?;

        let appointment = ahri_tre_pgmeta::governance::GovernanceAppointment::from_identities(
            plan.governance_custodians()
                .iter()
                .map(|a| (a.issuer(), a.subject())),
        )
        .map_err(verification_failure)?;
        let operator = creation_postgres_connection(plan)
            .upgrade_governance(
                plan.postgresql().database(),
                plan.metadata_role(),
                administrator.material(),
                &appointment,
            )
            .map_err(verification_failure)?;
        let (path, _) = creation_lake_location(plan).map_err(verification_failure)?;
        let scratch = create_scratch(plan, std::iter::once(path))?;
        let mut lake = open_existing_lake(plan, catalog_credential.material(), storage, &scratch)?;
        let ledger = operator.binding().map_err(verification_failure)?;
        ahri_tre_lake::GovernanceLedger::provision(&mut lake.connection, StudyId(ledger.study_id))
            .map_err(verification_failure)?;
        let converted_assets = operator
            .convert_existing(batch)
            .map_err(verification_failure)?;
        let origins = operator
            .unbound_legacy_datasets(batch)
            .map_err(verification_failure)?;
        let bound_datasets = origins.len();
        for origin in origins {
            let (table_uuid, lake_snapshot) = ahri_tre_lake::dataset_physical_identity(
                &lake.connection,
                plan.catalog_schema(),
                StudyId(origin.study_id),
                origin.name.as_str(),
                origin.major,
                origin.minor,
                origin.patch,
            )
            .map_err(verification_failure)?;
            operator
                .bind_dataset(&ahri_tre_types::GovernedDatasetLocation {
                    version_id: origin.version_id,
                    asset_id: origin.asset_id,
                    study_id: origin.study_id,
                    name: origin.name,
                    major: origin.major,
                    minor: origin.minor,
                    patch: origin.patch,
                    table_uuid,
                    lake_snapshot,
                })
                .map_err(verification_failure)?;
        }
        let files = operator
            .unbound_legacy_datafiles(batch)
            .map_err(verification_failure)?;
        let bound_datafiles = files.len();
        for origin in files {
            let attestation = ahri_tre_lake::attest_datafile(
                &lake.attach_plan.data_path,
                StudyId(origin.study_id),
                ahri_tre_types::AssetId(origin.asset_id),
                &origin.datafile,
            )
            .map_err(verification_failure)?;
            operator
                .bind_datafile(&attestation)
                .map_err(verification_failure)?;
        }
        for event in operator.pending(batch).map_err(verification_failure)? {
            crate::disclosure::project_session_evidence(&operator, &mut lake.connection, event)
                .map_err(verification_failure)?;
        }
        Ok(GovernanceUpgradeProgress {
            converted_assets,
            remaining_unclassified_assets: operator
                .upgrade_plan()
                .map_err(verification_failure)?
                .unclassified_assets,
            bound_datasets,
            bound_datafiles,
            remaining_datafile_bindings: !operator
                .unbound_legacy_datafiles(1)
                .map_err(verification_failure)?
                .is_empty(),
            remaining_dataset_bindings: !operator
                .unbound_legacy_datasets(1)
                .map_err(verification_failure)?
                .is_empty(),
            remaining_evidence: !operator
                .pending(1)
                .map_err(verification_failure)?
                .is_empty(),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::{equivalent_ducklake_data_paths, reviewed_demo_dataset_seeds};

    #[test]
    fn ducklake_persisted_trailing_separator_is_equivalent() {
        assert!(equivalent_ducklake_data_paths(
            "/var/lib/ahri-tre/lake/development/",
            "/var/lib/ahri-tre/lake/development"
        ));
        assert!(!equivalent_ducklake_data_paths(
            "/var/lib/ahri-tre/lake/development-other",
            "/var/lib/ahri-tre/lake/development"
        ));
    }

    #[test]
    fn trusted_demo_fixtures_are_the_exact_reviewed_allowlist() {
        let fixtures = reviewed_demo_dataset_seeds().unwrap();
        assert_eq!(
            fixtures
                .iter()
                .map(|fixture| fixture.dataset_name.as_str())
                .collect::<Vec<_>>(),
            vec![
                "entities",
                "episodes",
                "household_map",
                "householdmemberships",
                "householdresidences",
                "households",
                "individual_map",
                "individualresidences",
                "individuals",
                "location_map",
                "locations",
                "transformations",
            ]
        );
        assert!(fixtures.iter().all(|fixture| !fixture.contents.is_empty()));
    }
}
#[cfg(test)]
mod ticket19_tests {
    use super::{
        CreatedCatalogCredential, DatastoreCreationAuthority, DatastoreCreationFailure,
        DatastoreCreationProvisioner, DatastoreCreationStart, PrivilegedDatastoreCreationHandler,
        ResolvedCreationStorageSecret, StagedCatalogCredential,
    };
    use age::secrecy::ExposeSecret;
    use ahri_tre_config::{
        ConfigurationAuthority, DatastoreConfigurationId, EffectiveConfigurationTarget,
    };
    use ahri_tre_runtime::{
        AdministrationFailureCode, AdministrationPeerIdentity, AdministrationRequest,
        AdministrationRequestHandler, AdministrationResponse, DatastoreCreationPlan,
        ResolvedOperationSecret, ResolvedSecretVersion,
    };
    use ahri_tre_secrets::internal::{
        ManagedSecretOwnerBackend, ManagedSecretReferenceInspectionError,
        ManagedSecretReferenceStatus, ManagedSecretRemovalProof, ManagedSecretTrustedBackend,
    };
    use ahri_tre_secrets::{
        LocalManagedSecretStore, ManagedSecretBackend, ManagedSecretErrorKind,
        ManagedSecretIdentity, ManagedSecretMetadata, ManagedSecretReference, SecretMaterial,
    };
    use ahri_tre_types::{
        DatastoreCreationFailureCode, DatastoreIdentityBinding, DatastoreLifecycleState,
        LakeCatalogCredentialMode,
    };
    use chrono::Utc;
    use std::fs;
    use std::os::unix::fs::PermissionsExt;
    use std::path::PathBuf;
    use std::str::FromStr;
    use std::sync::{Arc, Mutex};
    use uuid::Uuid;

    struct RotationAuthority {
        plan: DatastoreCreationPlan,
        store: Arc<LocalManagedSecretStore>,
        events: Arc<Mutex<Vec<&'static str>>>,
        application_referenced: bool,
    }

    impl DatastoreCreationAuthority for RotationAuthority {
        fn deployment_id(&self) -> Result<Uuid, AdministrationFailureCode> {
            Ok(Uuid::from_u128(1))
        }

        fn retain_datastore_binding_reference(
            &self,
            _plan: &DatastoreCreationPlan,
            _reference: &ManagedSecretReference,
        ) -> Result<(), AdministrationFailureCode> {
            Ok(())
        }

        fn plan(
            &self,
            _datastore: &DatastoreConfigurationId,
        ) -> Result<DatastoreCreationPlan, AdministrationFailureCode> {
            Ok(self.plan.clone())
        }

        fn resolve_administrator(
            &self,
            _plan: &DatastoreCreationPlan,
        ) -> Result<ResolvedOperationSecret, AdministrationFailureCode> {
            Ok(resolved_secret(
                "managed://postgresql/runtime-password",
                b"administrator",
                1,
            ))
        }

        fn plan_for_existing(
            &self,
            _datastore: &DatastoreConfigurationId,
            _datastore_id: Uuid,
        ) -> Result<DatastoreCreationPlan, AdministrationFailureCode> {
            Ok(self.plan.clone())
        }

        fn resolve_existing_catalog_credential(
            &self,
            plan: &DatastoreCreationPlan,
            binding: &DatastoreIdentityBinding,
        ) -> Result<ResolvedOperationSecret, AdministrationFailureCode> {
            self.events.lock().unwrap().push("resolve-prior");
            Ok(resolved_secret(
                plan.catalog_secret_reference().as_str(),
                b"prior-catalog-credential",
                u64::try_from(binding.credential_version).unwrap(),
            ))
        }

        fn resolve_storage(
            &self,
            _plan: &DatastoreCreationPlan,
        ) -> Result<Vec<ResolvedCreationStorageSecret>, AdministrationFailureCode> {
            Ok(Vec::new())
        }

        fn create_catalog_credential(
            &self,
            _plan: &DatastoreCreationPlan,
            _material: &SecretMaterial,
        ) -> Result<CreatedCatalogCredential, AdministrationFailureCode> {
            unreachable!("rotation never creates a second logical reference")
        }

        fn verify_catalog_credential(
            &self,
            _plan: &DatastoreCreationPlan,
            _material: &SecretMaterial,
            _credential: &CreatedCatalogCredential,
        ) -> Result<(), AdministrationFailureCode> {
            unreachable!("rotation verifies through the owning PostgreSQL adapter")
        }

        fn stage_catalog_credential_replacement(
            &self,
            plan: &DatastoreCreationPlan,
            _binding: &DatastoreIdentityBinding,
            material: &SecretMaterial,
        ) -> Result<StagedCatalogCredential, AdministrationFailureCode> {
            self.events.lock().unwrap().push("stage");
            let active = self
                .store
                .lookup_active_metadata(plan.catalog_secret_reference())
                .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)?;
            self.store
                .stage_owner_replacement(
                    plan.catalog_secret_reference(),
                    material,
                    active.version(),
                )
                .map(|inner| StagedCatalogCredential { inner })
                .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
        }

        fn activate_catalog_credential_replacement(
            &self,
            _plan: &DatastoreCreationPlan,
            staged: &StagedCatalogCredential,
        ) -> Result<CreatedCatalogCredential, AdministrationFailureCode> {
            self.events.lock().unwrap().push("activate");
            self.store
                .activate_owner_replacement(&staged.inner)
                .map(Into::into)
                .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
        }

        fn discard_catalog_credential_replacement(
            &self,
            _plan: &DatastoreCreationPlan,
            staged: &StagedCatalogCredential,
        ) -> Result<(), AdministrationFailureCode> {
            self.events.lock().unwrap().push("discard");
            self.store
                .discard_owner_replacement(&staged.inner)
                .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
        }

        fn configured_datastore_ids(
            &self,
        ) -> Result<Vec<DatastoreConfigurationId>, AdministrationFailureCode> {
            Ok(Vec::new())
        }

        fn postgresql_binding_inspection_targets(
            &self,
        ) -> Result<
            Vec<ahri_tre_runtime::PostgresqlBindingInspectionTarget>,
            AdministrationFailureCode,
        > {
            Ok(Vec::new())
        }

        fn application_reference_status(
            &self,
            _reference: &ManagedSecretReference,
        ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
            self.events.lock().unwrap().push("application-inspect");
            Ok(if self.application_referenced {
                ManagedSecretReferenceStatus::Referenced
            } else {
                ManagedSecretReferenceStatus::Unreferenced
            })
        }

        fn session_reference_status(
            &self,
            _reference: &ManagedSecretReference,
        ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
            Ok(ManagedSecretReferenceStatus::Unreferenced)
        }

        fn datastore_reference_status(
            &self,
            reference: &ManagedSecretReference,
        ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
            self.store
                .trusted_held_reference_status(reference)
                .map_err(|_| ManagedSecretReferenceInspectionError::unavailable())
        }

        fn authentication_reference_status(
            &self,
            _reference: &ManagedSecretReference,
        ) -> Result<ManagedSecretReferenceStatus, ManagedSecretReferenceInspectionError> {
            Ok(ManagedSecretReferenceStatus::Unreferenced)
        }

        fn remove_managed_secret(
            &self,
            reference: &ManagedSecretReference,
            proof: ManagedSecretRemovalProof,
        ) -> Result<ManagedSecretMetadata, AdministrationFailureCode> {
            self.store
                .remove_with_repository_evidence(reference, proof)
                .map_err(|_| AdministrationFailureCode::CredentialSetupFailed)
        }
    }

    struct RotationProvisioner {
        binding: DatastoreIdentityBinding,
        events: Arc<Mutex<Vec<&'static str>>>,
        fail_apply: bool,
        uncertain_apply: bool,
    }

    impl DatastoreCreationProvisioner for RotationProvisioner {
        type Pending = ();

        fn begin(
            &self,
            _plan: &DatastoreCreationPlan,
            _administrator: &ResolvedOperationSecret,
            _catalog_credential: &SecretMaterial,
        ) -> DatastoreCreationStart<Self::Pending> {
            unreachable!("rotation does not run Datastore creation")
        }

        fn initialize_lake(
            &self,
            _pending: &mut Self::Pending,
            _plan: &DatastoreCreationPlan,
            _catalog_credential: &SecretMaterial,
            _storage: &[ResolvedCreationStorageSecret],
        ) -> Result<(), DatastoreCreationFailure> {
            unreachable!("rotation does not initialize the lake")
        }

        fn verify(
            &self,
            _pending: &mut Self::Pending,
            _plan: &DatastoreCreationPlan,
        ) -> Result<(), DatastoreCreationFailure> {
            unreachable!("rotation verifies in its owner-specific phase")
        }

        fn mark_ready(
            &self,
            _pending: &mut Self::Pending,
            _credential: &CreatedCatalogCredential,
        ) -> Result<(), DatastoreCreationFailure> {
            unreachable!("rotation updates the retained ready binding")
        }

        fn mark_failed(
            &self,
            _pending: &mut Self::Pending,
            _failure: &DatastoreCreationFailure,
        ) -> Result<(), DatastoreCreationFailure> {
            unreachable!("rotation compensates instead of failing the Datastore")
        }

        fn find_existing(
            &self,
            _plan: &DatastoreCreationPlan,
            _administrator: &ResolvedOperationSecret,
        ) -> Result<Option<DatastoreIdentityBinding>, DatastoreCreationFailure> {
            Ok(Some(self.binding.clone()))
        }

        fn apply_catalog_credential_rotation(
            &self,
            _plan: &DatastoreCreationPlan,
            _binding: &DatastoreIdentityBinding,
            _administrator: &ResolvedOperationSecret,
            _prior: &ResolvedOperationSecret,
            staged: &StagedCatalogCredential,
            rotated_at: chrono::DateTime<Utc>,
        ) -> Result<DatastoreIdentityBinding, DatastoreCreationFailure> {
            self.events.lock().unwrap().push("apply-and-verify");
            if self.uncertain_apply {
                return Err(DatastoreCreationFailure::new(
                    DatastoreCreationFailureCode::CredentialStateUncertain,
                    "injected uncertain owner state",
                ));
            }
            if self.fail_apply {
                return Err(DatastoreCreationFailure::new(
                    DatastoreCreationFailureCode::PostgresqlSetupFailed,
                    "injected owner apply failure",
                ));
            }
            let mut binding = self.binding.clone();
            binding.credential_version = i32::try_from(staged.version()).unwrap();
            binding.credential_last_rotated_at = Some(rotated_at);
            binding.binding_fingerprint = binding.computed_binding_fingerprint();
            Ok(binding)
        }

        fn rollback_catalog_credential_rotation(
            &self,
            _plan: &DatastoreCreationPlan,
            _binding: &DatastoreIdentityBinding,
            _administrator: &ResolvedOperationSecret,
            _prior: &ResolvedOperationSecret,
        ) -> Result<(), DatastoreCreationFailure> {
            self.events.lock().unwrap().push("rollback");
            Ok(())
        }
    }

    #[test]
    fn owner_rotation_activates_only_after_external_apply_and_verification() {
        let fixture = ManagedStoreFixture::new("activate");
        let store = Arc::new(fixture.initialize_store());
        let plan = rotation_plan();
        let reference = plan.catalog_secret_reference().clone();
        store
            .create(
                &reference,
                &SecretMaterial::try_from(b"prior-catalog-credential".to_vec()).unwrap(),
            )
            .unwrap();
        let events = Arc::new(Mutex::new(Vec::new()));
        let binding = ready_binding(plan.datastore_id(), &reference);
        let handler = PrivilegedDatastoreCreationHandler::new(
            RotationAuthority {
                plan,
                store: Arc::clone(&store),
                events: Arc::clone(&events),
                application_referenced: false,
            },
            RotationProvisioner {
                binding: binding.clone(),
                events: Arc::clone(&events),
                fail_apply: false,
                uncertain_apply: false,
            },
        );

        let response = handler.handle(peer(), rotate_request());

        assert_eq!(
            response,
            AdministrationResponse::datastore_ready("research", binding.datastore_id)
        );
        assert_eq!(
            events.lock().unwrap().as_slice(),
            ["resolve-prior", "stage", "apply-and-verify", "activate"]
        );
        assert_eq!(
            store
                .resolve(&reference)
                .unwrap()
                .metadata()
                .version()
                .get(),
            2
        );
        assert!(
            store
                .resume_owner_replacement(&reference)
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn failed_owner_apply_discards_stage_and_preserves_prior_active_version() {
        let fixture = ManagedStoreFixture::new("compensate");
        let store = Arc::new(fixture.initialize_store());
        let plan = rotation_plan();
        let reference = plan.catalog_secret_reference().clone();
        store
            .create(
                &reference,
                &SecretMaterial::try_from(b"prior-catalog-credential".to_vec()).unwrap(),
            )
            .unwrap();
        let events = Arc::new(Mutex::new(Vec::new()));
        let binding = ready_binding(plan.datastore_id(), &reference);
        let handler = PrivilegedDatastoreCreationHandler::new(
            RotationAuthority {
                plan,
                store: Arc::clone(&store),
                events: Arc::clone(&events),
                application_referenced: false,
            },
            RotationProvisioner {
                binding,
                events: Arc::clone(&events),
                fail_apply: true,
                uncertain_apply: false,
            },
        );

        let response = handler.handle(peer(), rotate_request());

        assert_eq!(
            response,
            AdministrationResponse::datastore_failed(
                "research",
                AdministrationFailureCode::PostgresqlSetupFailed,
            )
        );
        assert_eq!(
            events.lock().unwrap().as_slice(),
            ["resolve-prior", "stage", "apply-and-verify", "discard"]
        );
        assert_eq!(
            store
                .resolve(&reference)
                .unwrap()
                .metadata()
                .version()
                .get(),
            1
        );
        assert!(
            store
                .resume_owner_replacement(&reference)
                .unwrap()
                .is_none()
        );
    }
    #[test]
    fn uncertain_owner_apply_retains_stage_for_authoritative_reconciliation() {
        let fixture = ManagedStoreFixture::new("uncertain");
        let store = Arc::new(fixture.initialize_store());
        let plan = rotation_plan();
        let reference = plan.catalog_secret_reference().clone();
        store
            .create(
                &reference,
                &SecretMaterial::try_from(b"prior-catalog-credential".to_vec()).unwrap(),
            )
            .unwrap();
        let events = Arc::new(Mutex::new(Vec::new()));
        let binding = ready_binding(plan.datastore_id(), &reference);
        let handler = PrivilegedDatastoreCreationHandler::new(
            RotationAuthority {
                plan,
                store: Arc::clone(&store),
                events: Arc::clone(&events),
                application_referenced: false,
            },
            RotationProvisioner {
                binding,
                events: Arc::clone(&events),
                fail_apply: false,
                uncertain_apply: true,
            },
        );

        let response = handler.handle(peer(), rotate_request());

        assert_eq!(
            response,
            AdministrationResponse::datastore_failed(
                "research",
                AdministrationFailureCode::CredentialStateUncertain,
            )
        );
        assert_eq!(
            events.lock().unwrap().as_slice(),
            ["resolve-prior", "stage", "apply-and-verify"]
        );
        assert_eq!(
            store
                .resolve(&reference)
                .unwrap()
                .metadata()
                .version()
                .get(),
            1
        );
        assert!(
            store
                .resume_owner_replacement(&reference)
                .unwrap()
                .is_some()
        );
    }

    #[test]
    fn durable_datastore_marker_blocks_removal_after_configuration_entry_is_gone() {
        let fixture = ManagedStoreFixture::new("stale-datastore-binding");
        let store = Arc::new(fixture.initialize_store());
        let plan = rotation_plan();
        let reference =
            ManagedSecretReference::from_str("managed://datastore/orphan/credential").unwrap();
        store
            .create(
                &reference,
                &SecretMaterial::try_from(b"orphan-binding-credential".to_vec()).unwrap(),
            )
            .unwrap();
        store
            .replace_trusted_reference_holder(
                &format!("datastore:{}", Uuid::new_v4()),
                std::slice::from_ref(&reference),
            )
            .unwrap();
        let binding = ready_binding(plan.datastore_id(), plan.catalog_secret_reference());
        let handler = PrivilegedDatastoreCreationHandler::new(
            RotationAuthority {
                plan,
                store: Arc::clone(&store),
                events: Arc::new(Mutex::new(Vec::new())),
                application_referenced: false,
            },
            RotationProvisioner {
                binding,
                events: Arc::new(Mutex::new(Vec::new())),
                fail_apply: false,
                uncertain_apply: false,
            },
        );

        let response = handler.handle(
            peer(),
            AdministrationRequest::ManagedSecretRemove {
                reference: reference.clone(),
                deployment_id: Uuid::from_u128(1),
            },
        );

        assert_eq!(
            response,
            AdministrationResponse::managed_secret_removal_failed(
                reference.as_str(),
                AdministrationFailureCode::ReferencePresent,
            )
        );
        assert_eq!(
            store
                .resolve(&reference)
                .unwrap()
                .metadata()
                .version()
                .get(),
            1
        );
    }

    #[test]
    fn removal_rejects_deployment_mismatch_before_reference_inspection() {
        let fixture = ManagedStoreFixture::new("deployment-mismatch");
        let store = Arc::new(fixture.initialize_store());
        let plan = rotation_plan();
        let reference =
            ManagedSecretReference::from_str("managed://operator/removal-candidate").unwrap();
        store
            .create(
                &reference,
                &SecretMaterial::try_from(b"removal-candidate".to_vec()).unwrap(),
            )
            .unwrap();
        let events = Arc::new(Mutex::new(Vec::new()));
        let binding = ready_binding(plan.datastore_id(), plan.catalog_secret_reference());
        let handler = PrivilegedDatastoreCreationHandler::new(
            RotationAuthority {
                plan,
                store: Arc::clone(&store),
                events: Arc::clone(&events),
                application_referenced: false,
            },
            RotationProvisioner {
                binding,
                events: Arc::clone(&events),
                fail_apply: false,
                uncertain_apply: false,
            },
        );

        let response = handler.handle(
            peer(),
            AdministrationRequest::ManagedSecretRemove {
                reference: reference.clone(),
                deployment_id: Uuid::from_u128(2),
            },
        );

        assert_eq!(
            response,
            AdministrationResponse::managed_secret_removal_failed(
                reference.as_str(),
                AdministrationFailureCode::ConfigurationUnavailable,
            )
        );
        assert!(events.lock().unwrap().is_empty());
        assert_eq!(
            store
                .resolve(&reference)
                .unwrap()
                .metadata()
                .version()
                .get(),
            1
        );
    }

    #[test]
    fn removal_requires_all_authorities_to_report_unreferenced() {
        let fixture = ManagedStoreFixture::new("removal-proof");
        let store = Arc::new(fixture.initialize_store());
        let plan = rotation_plan();
        let reference =
            ManagedSecretReference::from_str("managed://operator/removal-candidate").unwrap();
        store
            .create(
                &reference,
                &SecretMaterial::try_from(b"removal-candidate".to_vec()).unwrap(),
            )
            .unwrap();
        let binding = ready_binding(plan.datastore_id(), plan.catalog_secret_reference());
        let handler = PrivilegedDatastoreCreationHandler::new(
            RotationAuthority {
                plan,
                store: Arc::clone(&store),
                events: Arc::new(Mutex::new(Vec::new())),
                application_referenced: true,
            },
            RotationProvisioner {
                binding,
                events: Arc::new(Mutex::new(Vec::new())),
                fail_apply: false,
                uncertain_apply: false,
            },
        );

        let response = handler.handle(
            peer(),
            AdministrationRequest::ManagedSecretRemove {
                reference: reference.clone(),
                deployment_id: Uuid::from_u128(1),
            },
        );

        assert_eq!(
            response,
            AdministrationResponse::managed_secret_removal_failed(
                reference.as_str(),
                AdministrationFailureCode::ReferencePresent,
            )
        );
        assert_eq!(
            store
                .resolve(&reference)
                .unwrap()
                .metadata()
                .version()
                .get(),
            1
        );
    }

    #[test]
    fn unreferenced_removal_is_durable_and_tombstones_the_reference() {
        let fixture = ManagedStoreFixture::new("remove");
        let store = Arc::new(fixture.initialize_store());
        let plan = rotation_plan();
        let reference =
            ManagedSecretReference::from_str("managed://operator/removal-candidate").unwrap();
        store
            .create(
                &reference,
                &SecretMaterial::try_from(b"removal-candidate".to_vec()).unwrap(),
            )
            .unwrap();
        let binding = ready_binding(plan.datastore_id(), plan.catalog_secret_reference());
        let handler = PrivilegedDatastoreCreationHandler::new(
            RotationAuthority {
                plan,
                store: Arc::clone(&store),
                events: Arc::new(Mutex::new(Vec::new())),
                application_referenced: false,
            },
            RotationProvisioner {
                binding,
                events: Arc::new(Mutex::new(Vec::new())),
                fail_apply: false,
                uncertain_apply: false,
            },
        );

        let response = handler.handle(
            peer(),
            AdministrationRequest::ManagedSecretRemove {
                reference: reference.clone(),
                deployment_id: Uuid::from_u128(1),
            },
        );

        assert_eq!(
            response,
            AdministrationResponse::managed_secret_removed(reference.as_str(), 1)
        );
        assert_eq!(
            store.resolve(&reference).unwrap_err().kind(),
            ManagedSecretErrorKind::Missing
        );
        assert_eq!(
            store
                .create(
                    &reference,
                    &SecretMaterial::try_from(b"cannot-recreate".to_vec()).unwrap(),
                )
                .unwrap_err()
                .kind(),
            ManagedSecretErrorKind::VersionConflict
        );
    }

    fn rotation_plan() -> DatastoreCreationPlan {
        let document = fs::read_to_string(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../examples/config/application-v1.toml"
        ))
        .unwrap();
        let datastore = DatastoreConfigurationId::from_str("research").unwrap();
        let effective = ConfigurationAuthority::parse_application(&document)
            .unwrap()
            .resolve_effective(EffectiveConfigurationTarget::DatastoreCreate { datastore })
            .unwrap();
        DatastoreCreationPlan::try_from((
            &effective,
            Uuid::parse_str("12345678-9abc-4def-8123-456789abcdef").unwrap(),
        ))
        .unwrap()
    }

    fn ready_binding(
        datastore_id: Uuid,
        reference: &ManagedSecretReference,
    ) -> DatastoreIdentityBinding {
        let mut binding = DatastoreIdentityBinding {
            datastore_id,
            datastore_name: "Research".to_string(),
            ducklake_catalog_database: "ducklake_catalog".to_string(),
            ducklake_catalog_schema: Some("catalog".to_string()),
            lake_path: "/srv/ahri-tre/lake/datasets".to_string(),
            storage_policy_id: Some("filesystem".to_string()),
            lake_location: None,
            ducklake_encryption: true,
            lake_catalog_credential_mode: LakeCatalogCredentialMode::ManagedLocal,
            lake_catalog_role_name: Some("catalog_owner".to_string()),
            managed_secret_ref: Some(reference.as_str().to_string()),
            credential_version: 1,
            credential_last_rotated_at: Some(Utc::now()),
            created_at: Utc::now(),
            creating_tool_version: Some(env!("CARGO_PKG_VERSION").to_string()),
            binding_fingerprint: String::new(),
            lifecycle_state: DatastoreLifecycleState::Ready,
            failure_code: None,
            failure_summary: None,
        };
        binding.binding_fingerprint = binding.computed_binding_fingerprint();
        binding
    }

    fn resolved_secret(reference: &str, material: &[u8], version: u64) -> ResolvedOperationSecret {
        ResolvedOperationSecret::new(
            SecretMaterial::try_from(material.to_vec()).unwrap(),
            ResolvedSecretVersion::managed(
                ManagedSecretReference::from_str(reference).unwrap(),
                version,
            ),
        )
    }

    fn peer() -> AdministrationPeerIdentity {
        AdministrationPeerIdentity::for_test(1000, 1000, Some(42))
    }

    fn rotate_request() -> AdministrationRequest {
        AdministrationRequest::DatastoreRotateCredential {
            datastore: DatastoreConfigurationId::from_str("research").unwrap(),
        }
    }

    struct ManagedStoreFixture {
        root: PathBuf,
        store_root: PathBuf,
        deployment_id: Uuid,
        identity: String,
    }

    impl ManagedStoreFixture {
        fn new(label: &str) -> Self {
            let root = std::env::temp_dir().join(format!(
                "ahri-tre-app-ticket19-{label}-{}-{}",
                std::process::id(),
                Uuid::new_v4()
            ));
            fs::create_dir(&root).unwrap();
            fs::set_permissions(&root, fs::Permissions::from_mode(0o700)).unwrap();
            let identity = age::x25519::Identity::generate()
                .to_string()
                .expose_secret()
                .to_string();
            Self {
                store_root: root.join("secrets"),
                root,
                deployment_id: Uuid::new_v4(),
                identity,
            }
        }

        fn initialize_store(&self) -> LocalManagedSecretStore {
            LocalManagedSecretStore::initialize_at_for_test(
                &self.store_root,
                self.deployment_id,
                self.parsed_identity(),
            )
            .unwrap()
        }

        fn parsed_identity(&self) -> ManagedSecretIdentity {
            ManagedSecretIdentity::from_material(
                &SecretMaterial::try_from(self.identity.as_bytes().to_vec()).unwrap(),
            )
            .unwrap()
        }
    }

    impl Drop for ManagedStoreFixture {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.root);
        }
    }
}