ahri_tre_app/

configured_datastore.rs

use std::sync::{Arc, Mutex};

use ahri_tre_lake::{
    AwsWorkloadIdentitySource, DuckLakeAdapter, FilesystemCatalogTls, FilesystemLakeSessionConfig,
    ObjectLakeSessionConfig, ObjectStorageAuthentication, ObjectStorageLocation, ObjectStorageTls,
    S3UrlStyle, ScratchAttemptId, TrustedScratch,
};
use ahri_tre_libpq_oauth::{
    ExplicitLibpqOAuthConnectConfig, LibpqOAuthConnectFailureKind, LibpqOAuthConnector,
    LibpqOAuthTls,
};
use ahri_tre_orcid::OAuthSession;
use ahri_tre_pgmeta::{
    datastore_identity_table_exists_oauth, read_datastore_identity_bindings_oauth,
};
use ahri_tre_runtime::{
    AzureManagedIdentity, ConfiguredAwsWorkloadIdentitySource, ConfiguredS3UrlStyle,
    ConfiguredSecret, DataStoreRuntime, DatastoreSessionPlan, DuckLakeConnection, LakeStoragePlan,
    PgStoreConnection, ResolvedOperationSecret, ResolvedSecretVersion, SessionAuthenticationRecord,
    SessionTls, StorageAuthenticationPlan, StorageTlsPlan,
};
use ahri_tre_secrets::{ManagedSecretReference, ResolvedManagedSecret};
use ahri_tre_session::SessionReferenceIntent;
use ahri_tre_types::{
    DataStoreProfile, DatastoreIdentityBinding, DatastoreLakeLocation, DatastoreLifecycleState,
    EncryptionMode, LakeAttachProfile, LakeAuthProfile,
};

use crate::{AppError, DataStoreSession, StoreSessionConnection};

/// Already-resolved authority required to open one configured Datastore.
pub struct OpenConfiguredDatastoreRequest {
    pub plan: DatastoreSessionPlan,
    pub authentication: OAuthSession,
    pub session_reference: ConfiguredSessionReference,
}

/// Failure opening a configured Datastore for an authenticated Session.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ConfiguredDatastoreOpenError {
    /// PostgreSQL rejected the Session's OAuth authentication.
    #[error("configured PostgreSQL rejected OAuth authentication")]
    AuthenticationRejected,
    /// A non-authentication application failure prevented the open.
    #[error(transparent)]
    Application(#[from] AppError),
}

impl ConfiguredDatastoreOpenError {
    fn into_app_error(self) -> AppError {
        match self {
            Self::AuthenticationRejected => AppError::Infrastructure(
                "configured PostgreSQL open failed: OAuth authentication rejected".to_string(),
            ),
            Self::Application(error) => error,
        }
    }
}

pub struct ConfiguredSessionReference {
    session_id: uuid::Uuid,
    release_capability: Option<uuid::Uuid>,
    authentication: SessionAuthenticationRecord,
}

/// Opaque public identifier paired with one process-ephemeral live Session.
pub struct LiveSessionIdentifier(uuid::Uuid);

impl LiveSessionIdentifier {
    pub fn session_id(&self) -> uuid::Uuid {
        self.0
    }
}

impl ConfiguredSessionReference {
    /// Retains the original live Session identity while refreshing its handles
    /// for newly authorized work. The caller must still own that live Session.
    pub fn retained_live(
        authentication: SessionAuthenticationRecord,
        session_id: uuid::Uuid,
    ) -> Self {
        Self {
            session_id,
            release_capability: None,
            authentication,
        }
    }

    pub fn new(
        intent: SessionReferenceIntent,
        authentication: SessionAuthenticationRecord,
    ) -> Result<Self, AppError> {
        let journal_identity = intent.journal().identity();
        let authenticated_identity = authentication.identity();
        if journal_identity.issuer() != authenticated_identity.issuer()
            || journal_identity.client_id() != authenticated_identity.client_id()
            || journal_identity.subject() != authenticated_identity.subject()
        {
            return Err(AppError::Validation(
                "Session reference intent belongs to a different authenticated identity".into(),
            ));
        }
        Ok(Self {
            session_id: intent.journal().reference_id(),
            release_capability: Some(intent.journal().release_capability()),
            authentication,
        })
    }

    /// Creates a process-ephemeral live Session reference. Unlike a persisted
    /// client Session journal, it retains no durable Managed-secret holder.
    pub fn live(authentication: SessionAuthenticationRecord) -> (Self, LiveSessionIdentifier) {
        let session_id = uuid::Uuid::new_v4();
        (
            Self {
                session_id,
                release_capability: None,
                authentication,
            },
            LiveSessionIdentifier(session_id),
        )
    }

    fn session_id(&self) -> uuid::Uuid {
        self.session_id
    }

    fn release_capability(&self) -> uuid::Uuid {
        self.release_capability
            .expect("durable Session reference has a release capability")
    }

    fn retains_snapshot(&self) -> bool {
        self.release_capability.is_some()
    }
}

/// Trusted operation capability that resolves only the binding-owned catalog credential.
pub trait DatastoreSecretResolver {
    /// Authorizes one cleanup operation using immutable configuration and the
    /// existing Secret resolver. The adapter exposes no materialization API.
    fn dataset_maintenance(
        &self,
        _plan: &DatastoreSessionPlan,
        _binding: &DatastoreIdentityBinding,
    ) -> Result<ahri_tre_pgmeta::PgDatasetMaintenance, AppError> {
        Err(AppError::Infrastructure(
            "Dataset recovery authority is unavailable".into(),
        ))
    }

    fn resolve_catalog_credential(
        &self,
        binding: &DatastoreIdentityBinding,
        reference: &ManagedSecretReference,
    ) -> Result<ResolvedManagedSecret, AppError>;

    fn resolve_storage_credential(
        &self,
        _plan: &DatastoreSessionPlan,
        _configured: &ConfiguredSecret,
    ) -> Result<ResolvedOperationSecret, AppError> {
        Err(AppError::Validation(
            "object-storage credential resolution is unavailable".into(),
        ))
    }

    fn retain_session_reference_snapshot(
        &self,
        _reference: &ConfiguredSessionReference,
        _resolved_versions: &[ResolvedSecretVersion],
    ) -> Result<(), AppError> {
        Err(AppError::Validation(
            "configured Datastore Session reference retention is unavailable".into(),
        ))
    }
}

impl DatastoreSecretResolver for ahri_tre_runtime::TrustedRuntimeState {
    fn dataset_maintenance(
        &self,
        plan: &DatastoreSessionPlan,
        binding: &DatastoreIdentityBinding,
    ) -> Result<ahri_tre_pgmeta::PgDatasetMaintenance, AppError> {
        let configured = plan.datastore_id().parse().map_err(|_| {
            AppError::Infrastructure("Dataset maintenance configuration is unavailable".into())
        })?;
        let maintenance = self
            .datastore_creation_plan_with_id(&configured, binding.datastore_id)
            .map_err(|e| infrastructure("select Dataset maintenance configuration", e))?;
        if maintenance.postgresql().database() != plan.postgresql().database()
            || maintenance.postgresql().host() != plan.postgresql().host()
            || maintenance.postgresql().port() != plan.postgresql().port()
        {
            return Err(AppError::Infrastructure(
                "Dataset maintenance target does not match the Session".into(),
            ));
        }
        let credential = self
            .resolve_datastore_administrator(&maintenance)
            .map_err(|e| infrastructure("authorize Dataset maintenance", e))?;
        let certificates = match maintenance.postgresql().tls() {
            SessionTls::System => Vec::new(),
            SessionTls::CustomCa { certificates } => certificates.clone(),
        };
        ahri_tre_pgmeta::PgDatastoreCreationConnection::new(
            maintenance.postgresql().host(),
            maintenance.postgresql().port(),
            maintenance.postgresql().bootstrap_database(),
            maintenance.administrator().username(),
            certificates,
        )
        .dataset_maintenance(maintenance.postgresql().database(), credential.material())
        .map_err(|e| infrastructure("open Dataset maintenance capability", e))
    }

    fn resolve_catalog_credential(
        &self,
        binding: &DatastoreIdentityBinding,
        reference: &ManagedSecretReference,
    ) -> Result<ResolvedManagedSecret, AppError> {
        self.resolve_datastore_managed_for_new_operation(binding, reference)
            .map_err(|error| infrastructure("resolve Datastore catalog credential", error))
    }

    fn resolve_storage_credential(
        &self,
        plan: &DatastoreSessionPlan,
        configured: &ConfiguredSecret,
    ) -> Result<ResolvedOperationSecret, AppError> {
        self.resolve_storage_for_new_operation(plan, configured)
            .map_err(|error| infrastructure("resolve object-storage credential", error))
    }
    fn retain_session_reference_snapshot(
        &self,
        reference: &ConfiguredSessionReference,
        resolved_versions: &[ResolvedSecretVersion],
    ) -> Result<(), AppError> {
        self.retain_session_reference_snapshot(
            reference.session_id(),
            reference.release_capability(),
            &reference.authentication,
            resolved_versions,
        )
        .map_err(|error| infrastructure("retain Session Secret references", error))
    }
}

impl std::fmt::Debug for OpenConfiguredDatastoreRequest {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("OpenConfiguredDatastoreRequest")
            .field("plan", &self.plan)
            .field("authentication", &"OAuthSession(<redacted>)")
            .finish()
    }
}

/// Opens one configured Datastore from resolved runtime authority without
/// requiring an otherwise unrelated application repository set.
pub fn open_configured_datastore<R: DatastoreSecretResolver>(
    request: OpenConfiguredDatastoreRequest,
    resolver: &R,
) -> Result<DataStoreSession, AppError> {
    open_configured_datastore_for_session(request, resolver)
        .map_err(ConfiguredDatastoreOpenError::into_app_error)
}

/// Authenticates PostgreSQL and validates its binding without opening Lake.
fn open_configured_metadata(
    request: &OpenConfiguredDatastoreRequest,
) -> Result<
    (
        ahri_tre_libpq_oauth::LibpqOAuthConnection,
        DatastoreIdentityBinding,
    ),
    ConfiguredDatastoreOpenError,
> {
    open_configured_metadata_observed(request, None)
}

fn open_configured_metadata_observed(
    request: &OpenConfiguredDatastoreRequest,
    context: Option<&ahri_tre_observability::CorrelationContext>,
) -> Result<
    (
        ahri_tre_libpq_oauth::LibpqOAuthConnection,
        DatastoreIdentityBinding,
    ),
    ConfiguredDatastoreOpenError,
> {
    validate_authentication_identity(
        request.session_reference.authentication.identity(),
        &request.authentication,
    )?;
    validate_authentication(&request.plan, &request.authentication)?;
    let tls = match request.plan.postgresql().tls() {
        SessionTls::System => LibpqOAuthTls::System,
        SessionTls::CustomCa { certificates } => LibpqOAuthTls::CustomCa {
            certificates: certificates.clone(),
        },
    };
    let postgresql = request.plan.postgresql();
    let authentication = request.plan.authentication();
    let config = ExplicitLibpqOAuthConnectConfig {
        host: postgresql.host().to_string(),
        routing_address: postgresql.routing_address().map(str::to_string),
        port: postgresql.port(),
        database: postgresql.database().to_string(),
        tls,
        oauth_issuer: authentication.issuer().to_string(),
        oauth_client_id: authentication.client_id().to_string(),
        oauth_scope: Some(authentication.scopes().join(" ")),
    };
    let mut store = match context {
        Some(context) => LibpqOAuthConnector.open_explicit_session_observed(
            context,
            &request.authentication,
            &config,
        ),
        None => LibpqOAuthConnector.open_explicit_session(&request.authentication, &config),
    }
    .map_err(configured_postgresql_open_error)?;
    store
        .health_check()
        .map_err(|error| infrastructure("configured PostgreSQL health check failed", error))?;
    let binding = resolve_binding(&mut store, &request.plan)?;
    Ok((store, binding))
}

/// Opens only PostgreSQL under current user authority; never attaches Lake or resolves its Secrets.
pub fn open_operation_control(
    request: OpenConfiguredDatastoreRequest,
    deployment_id: uuid::Uuid,
) -> Result<crate::service::OperationControl, ConfiguredDatastoreOpenError> {
    let (store, binding) = open_configured_metadata(&request)?;
    let repository = ahri_tre_pgmeta::PgMetadataRepository::operation_control(store)
        .map_err(|_| AppError::Infrastructure("Operation inspection is unavailable".into()))?;
    Ok(crate::service::OperationControl::new(
        repository,
        deployment_id,
        binding.datastore_id,
    ))
}

/// Opens a Datastore while preserving OAuth authentication rejection.
pub fn open_configured_datastore_for_session<R: DatastoreSecretResolver>(
    request: OpenConfiguredDatastoreRequest,
    resolver: &R,
) -> Result<DataStoreSession, ConfiguredDatastoreOpenError> {
    open_configured_datastore_for_session_observed(None, request, resolver)
}

/// Times the app workflow using its already authorized inputs and resolver.
pub fn open_configured_datastore_for_session_observed<R: DatastoreSecretResolver>(
    context: Option<&ahri_tre_observability::CorrelationContext>,
    request: OpenConfiguredDatastoreRequest,
    resolver: &R,
) -> Result<DataStoreSession, ConfiguredDatastoreOpenError> {
    open_configured_datastore_with_observations(
        context,
        request,
        resolver,
        crate::SessionObservations::default(),
        ahri_tre_protocol::diagnostics::DiagnosticObservationOrigin::SessionOpen,
    )
}

/// Runs the existing open workflow while retaining its actual adapter evidence.
pub fn open_configured_datastore_with_observations<R: DatastoreSecretResolver>(
    context: Option<&ahri_tre_observability::CorrelationContext>,
    request: OpenConfiguredDatastoreRequest,
    resolver: &R,
    observations: crate::SessionObservations,
    origin: ahri_tre_protocol::diagnostics::DiagnosticObservationOrigin,
) -> Result<DataStoreSession, ConfiguredDatastoreOpenError> {
    use ahri_tre_observability::{FailureCategory, Outcome, Stage};
    let span = context.map(|context| context.span(Stage::Application));
    let child = span.as_ref().map(|span| span.context());
    let result = open_configured_session(child.as_ref(), request, resolver, observations, origin);
    if let Some(span) = span {
        let (outcome, category) = match &result {
            Ok(_) => (Outcome::Success, None),
            Err(ConfiguredDatastoreOpenError::AuthenticationRejected) => {
                (Outcome::Rejected, Some(FailureCategory::Authentication))
            }
            Err(ConfiguredDatastoreOpenError::Application(AppError::Validation(_))) => {
                (Outcome::Rejected, Some(FailureCategory::Compatibility))
            }
            Err(_) => (Outcome::Unavailable, Some(FailureCategory::Dependency)),
        };
        span.finish(outcome, category);
    }
    result
}

fn open_configured_session<R: DatastoreSecretResolver>(
    context: Option<&ahri_tre_observability::CorrelationContext>,
    request: OpenConfiguredDatastoreRequest,
    resolver: &R,
    observations: crate::SessionObservations,
    origin: ahri_tre_protocol::diagnostics::DiagnosticObservationOrigin,
) -> Result<DataStoreSession, ConfiguredDatastoreOpenError> {
    let metadata = open_configured_metadata_observed(&request, context);
    observations.record(
        ahri_tre_protocol::diagnostics::DiagnosticComponent::Metadata,
        origin,
        metadata.is_ok(),
    );
    let (store, binding) = metadata?;
    let postgresql = request.plan.postgresql();
    let sslmode = "verify-full";
    let lake_tls = match postgresql.tls() {
        SessionTls::System => FilesystemCatalogTls::System,
        SessionTls::CustomCa { certificates } => FilesystemCatalogTls::CustomCa {
            certificates: certificates.clone(),
        },
    };

    let catalog_schema = binding.ducklake_catalog_schema.as_deref().ok_or_else(|| {
        AppError::Validation("persisted Datastore has no DuckLake catalog schema".to_string())
    })?;
    let catalog_role = binding.lake_catalog_role_name.as_deref().ok_or_else(|| {
        AppError::Validation("persisted Datastore has no DuckLake catalog role".to_string())
    })?;
    let managed_reference = binding
        .managed_secret_ref
        .as_deref()
        .ok_or_else(|| AppError::Validation("Datastore has no Managed catalog credential".into()))?
        .parse::<ManagedSecretReference>()
        .map_err(|_| {
            AppError::Validation("Datastore catalog credential reference is invalid".into())
        })?;
    let expected_reference = format!(
        "managed://datastore/{}/ducklake-password",
        binding.datastore_id
    );
    if managed_reference.as_secret_reference().as_str() != expected_reference {
        return Err(AppError::Validation(
            "Datastore catalog credential reference is not binding-owned".into(),
        )
        .into());
    }
    let resolved = resolver.resolve_catalog_credential(&binding, &managed_reference)?;
    if resolved.metadata().reference() != &managed_reference
        || u64::try_from(binding.credential_version).ok()
            != Some(resolved.metadata().version().get())
    {
        return Err(AppError::Validation(
            "resolved DuckLake credential does not match the persisted Datastore identity".into(),
        )
        .into());
    }
    let lake_credential_version =
        ResolvedSecretVersion::managed(managed_reference, resolved.metadata().version().get());
    let mut catalog_observations = ahri_tre_lake::LakeCatalogObservations::new(context);
    let lake = open_configured_lake(
        context,
        resolver,
        &request.plan,
        &binding,
        catalog_schema,
        catalog_role,
        lake_tls,
        resolved.material(),
        &mut catalog_observations,
    );
    observations.record_catalog(&catalog_observations, origin);
    observations.record(
        ahri_tre_protocol::diagnostics::DiagnosticComponent::Lake,
        origin,
        lake.is_ok(),
    );
    let (mut opened, scratch_attempt, mut storage_versions, canonical_data_path) = lake?;
    storage_versions.insert(0, lake_credential_version);

    let mut dataset_executor = request.plan.dataset_executor_root().and_then(|root| {
        ahri_tre_lake::DatasetExecutor::acquire(std::path::Path::new(root), binding.datastore_id)
            .ok()
    });

    let mut dataset_maintenance = None;
    if let (Some(executor), Some(scratch)) = (&dataset_executor, &scratch_attempt) {
        match resolver.dataset_maintenance(&request.plan, &binding) {
            Ok(maintenance) => {
                let lake = DuckLakeAdapter::new(&canonical_data_path);
                if reconcile_dataset_attempts(
                    &maintenance,
                    executor,
                    scratch,
                    &mut opened.connection,
                    &lake,
                )
                .is_err()
                {
                    dataset_executor = None;
                } else {
                    dataset_maintenance = Some(maintenance);
                }
            }
            Err(_) => dataset_executor = None,
        }
    }

    let profile = DataStoreProfile {
        server: postgresql.host().to_string(),
        port: postgresql.port(),
        dbname: postgresql.database().to_string(),
        sslmode: Some(sslmode.to_string()),
        lake: LakeAttachProfile {
            lake_data: canonical_data_path,
            lake_db: binding.ducklake_catalog_database.clone(),
            catalog_schema: Some(catalog_schema.to_string()),
            auth: LakeAuthProfile {
                lake_user: catalog_role.to_string(),
                encryption_mode: if binding.ducklake_encryption {
                    EncryptionMode::ServerManaged
                } else {
                    EncryptionMode::None
                },
            },
        },
    };
    let capture = ahri_tre_lake::DisclosureCaptureAuthority::from_opened(&mut opened).ok();
    let mut runtime = DataStoreRuntime::new(
        profile,
        PgStoreConnection {
            connection_description: store.connection_description().to_string(),
        },
        DuckLakeConnection {
            alias: opened.attach_plan.alias,
            data_path: opened.local_layout.data_path,
            staging_path: opened.local_layout.staging_path,
            tmp_path: opened.local_layout.tmp_path,
            test_runs_path: opened.local_layout.test_runs_path,
            catalog_database: binding.ducklake_catalog_database.clone(),
            catalog_schema: Some(catalog_schema.to_string()),
            attach_description: opened.attach_plan.attach_description,
            detected_encryption_mode: opened.health.detected_encryption_mode,
            automatic_migration: opened.attach_plan.automatic_migration,
            snapshot_count: opened.health.snapshot_count,
            table_count: opened.health.table_count,
            catalog_type: opened.health.catalog_type,
            extension_version: opened.health.extension_version,
        },
    )
    .with_identity_binding(binding)
    .with_session_provenance(request.plan.provenance().clone(), storage_versions);
    runtime.disclosure = request.plan.disclosure().clone();
    if request.session_reference.retains_snapshot() {
        resolver.retain_session_reference_snapshot(
            &request.session_reference,
            &runtime.resolved_secret_versions,
        )?;
    }
    let session = DataStoreSession::new(
        runtime,
        StoreSessionConnection::OAuth(Arc::new(Mutex::new(store))),
        opened.connection,
    )
    .with_authenticated_actor(request.authentication.role.clone())
    .with_disclosure_context(
        request.session_reference.session_id,
        request.session_reference.authentication.clone(),
    )
    .with_observations(observations)
    .with_lake_tls_guard(opened.tls_guard)
    .with_scratch_attempt(scratch_attempt)
    .with_dataset_maintenance(dataset_maintenance);
    let mut session = match dataset_executor {
        Some(executor) => session.with_dataset_executor(executor),
        None => session,
    };
    session.disclosure_capture = capture;
    // Older pending evidence must not impose a Datastore-wide admission pause.
    // A new disclosure still persists and verifies its own required evidence.
    let _ = session.reconcile_disclosures();
    Ok(session)
}

/// Startup maintenance for a configured Datastore, before the listener schedules
/// user work. Missing/unavailable Datastores are retried on explicit Session open.
/// No Session, user identity, Current study, or output admission is created here.
pub fn reconcile_configured_dataset_operations(
    runtime: &ahri_tre_runtime::TrustedRuntimeState,
    plan: &DatastoreSessionPlan,
) -> Result<bool, AppError> {
    let fail = || AppError::Infrastructure("Configured Dataset recovery is unavailable".into());
    let configured = plan.datastore_id().parse().map_err(|_| fail())?;
    let creation = runtime
        .datastore_creation_plan(&configured)
        .map_err(|_| fail())?;
    let credential = runtime
        .resolve_datastore_administrator(&creation)
        .map_err(|_| fail())?;
    let certificates = match creation.postgresql().tls() {
        SessionTls::System => Vec::new(),
        SessionTls::CustomCa { certificates } => certificates.clone(),
    };
    let connection = ahri_tre_pgmeta::PgDatastoreCreationConnection::new(
        creation.postgresql().host(),
        creation.postgresql().port(),
        creation.postgresql().bootstrap_database(),
        creation.administrator().username(),
        certificates,
    );
    let Some(binding) = ahri_tre_pgmeta::find_existing_datastore_binding(
        &connection,
        credential.material(),
        creation.postgresql().database(),
    )
    .map_err(|_| fail())?
    else {
        return Ok(false);
    };
    validate_binding(&binding, plan)?;
    let executor = ahri_tre_lake::DatasetExecutor::acquire(
        std::path::Path::new(plan.dataset_executor_root().ok_or_else(fail)?),
        binding.datastore_id,
    )
    .map_err(|_| fail())?;
    let maintenance = connection
        .dataset_maintenance(creation.postgresql().database(), credential.material())
        .map_err(|_| fail())?;
    // Record interruptions even if Lake recovery is temporarily unavailable.
    maintenance
        .reconcile_operations(executor.clone(), super::service::interrupt_operation)
        .map_err(|_| fail())?;
    let reference = format!(
        "managed://datastore/{}/ducklake-password",
        binding.datastore_id
    )
    .parse()
    .map_err(|_| fail())?;
    let catalog = runtime.resolve_catalog_credential(&binding, &reference)?;
    if catalog.metadata().version().get()
        != u64::try_from(binding.credential_version).map_err(|_| fail())?
    {
        return Err(fail());
    }
    let tls = match plan.postgresql().tls() {
        SessionTls::System => FilesystemCatalogTls::System,
        SessionTls::CustomCa { certificates } => FilesystemCatalogTls::CustomCa {
            certificates: certificates.clone(),
        },
    };
    let (mut lake, scratch, _, path) = open_configured_lake(
        None,
        runtime,
        plan,
        &binding,
        binding
            .ducklake_catalog_schema
            .as_deref()
            .ok_or_else(fail)?,
        binding.lake_catalog_role_name.as_deref().ok_or_else(fail)?,
        tls,
        catalog.material(),
        &mut ahri_tre_lake::LakeCatalogObservations::default(),
    )?;
    reconcile_dataset_attempts(
        &maintenance,
        &executor,
        &scratch.ok_or_else(fail)?,
        &mut lake.connection,
        &DuckLakeAdapter::new(&path),
    )
    .map(|()| true)
}

/// Reconcile only stopped, durably recorded attempts. One blocked cleanup keeps
/// its reservation; it does not prevent independent output identities proceeding.
pub(crate) fn reconcile_dataset_attempts(
    maintenance: &ahri_tre_pgmeta::PgDatasetMaintenance,
    executor: &Arc<ahri_tre_lake::DatasetExecutor>,
    scratch: &ahri_tre_lake::ScratchAttempt,
    connection: &mut duckdb::Connection,
    lake: &DuckLakeAdapter,
) -> Result<(), AppError> {
    maintenance
        .reconcile_operations(executor.clone(), super::service::interrupt_operation)
        .map_err(|e| infrastructure("reconcile interrupted operations", e))?;
    let attempts = maintenance
        .recoverable_attempts(executor.clone())
        .map_err(|e| infrastructure("inspect Dataset recovery", e))?;
    for attempt in attempts {
        let cleanup = || -> Result<(), AppError> {
            let mut recovery = maintenance
                .claim(executor.clone(), attempt.attempt_id)
                .map_err(|e| infrastructure("claim Dataset recovery", e))?;
            scratch
                .cleanup_dataset_attempt(&mut recovery)
                .map_err(|e| infrastructure("clean interrupted Dataset source", e))?;
            if recovery.requires_output_cleanup() {
                lake.cleanup_reserved_dataset_output(connection, &mut recovery)
                    .map_err(|e| infrastructure("clean interrupted Dataset output", e))?;
            }
            recovery
                .release_after_cleanup()
                .map_err(|e| infrastructure("release interrupted Dataset output", e))
        };
        // Ownership remains durable after any failed or concurrently claimed cleanup.
        let recovery = ahri_tre_observability::LifecycleSpan::start_operation(
            ahri_tre_observability::Phase::RecoveryCleanup,
            None,
            attempt.operation_id,
        );
        let result = cleanup();
        recovery.complete(&result, ahri_tre_observability::FailureCategory::Cleanup);
    }
    Ok(())
}

fn validate_authentication(
    plan: &DatastoreSessionPlan,
    session: &OAuthSession,
) -> Result<(), AppError> {
    if session.issuer != plan.authentication().issuer()
        || session.client_id != plan.authentication().client_id()
    {
        return Err(AppError::Validation(
            "Datastore Session credential does not match the selected authentication policy"
                .to_string(),
        ));
    }
    if session.expires_at <= chrono::Utc::now() {
        return Err(AppError::Validation(
            "Datastore Session credential has expired".to_string(),
        ));
    }
    Ok(())
}

fn validate_authentication_identity(
    identity: &ahri_tre_runtime::AuthenticationIdentityBinding,
    session: &OAuthSession,
) -> Result<(), AppError> {
    if session.issuer != identity.issuer()
        || session.client_id != identity.client_id()
        || session.subject != identity.subject()
    {
        return Err(AppError::Validation(
            "Datastore Session credential does not match the persisted Session identity"
                .to_string(),
        ));
    }
    Ok(())
}

fn reap_session_scratch(scratch: &TrustedScratch) {
    if scratch.reap_abandoned_sessions().is_err() {
        tracing::warn!("Abandoned Session scratch cleanup requires a later retry");
    }
}

#[allow(clippy::too_many_arguments)]
fn open_configured_lake<R: DatastoreSecretResolver>(
    context: Option<&ahri_tre_observability::CorrelationContext>,
    resolver: &R,
    plan: &DatastoreSessionPlan,
    binding: &DatastoreIdentityBinding,
    catalog_schema: &str,
    catalog_role: &str,
    catalog_tls: FilesystemCatalogTls,
    catalog_credential: &ahri_tre_secrets::SecretMaterial,
    catalog_observations: &mut ahri_tre_lake::LakeCatalogObservations,
) -> Result<
    (
        ahri_tre_lake::DuckLakeOpenedCatalog,
        Option<ahri_tre_lake::ScratchAttempt>,
        Vec<ResolvedSecretVersion>,
        String,
    ),
    AppError,
> {
    let lake_plan = plan.lake();
    let encryption = if binding.ducklake_encryption {
        EncryptionMode::ServerManaged
    } else {
        EncryptionMode::None
    };
    if let LakeStoragePlan::Filesystem { .. } = lake_plan.storage() {
        let config = FilesystemLakeSessionConfig::from_binding(
            lake_plan
                .filesystem_base()
                .expect("filesystem variant has a base"),
            &binding.lake_path,
            plan.postgresql().host(),
            plan.postgresql().port(),
            &binding.ducklake_catalog_database,
            catalog_schema,
            catalog_role,
            catalog_tls,
            encryption,
        )
        .map_err(|error| infrastructure("configured filesystem Lake inputs are invalid", error))?;
        validate_configured_physical_intent(lake_plan, &config, binding)?;
        let data_path = config.canonical_data_path();
        let scratch = TrustedScratch::open(
            std::path::Path::new(plan.configuration_scratch_root()),
            [std::path::Path::new(&data_path)],
        )
        .map_err(|error| infrastructure("trusted scratch validation failed", error))?;
        reap_session_scratch(&scratch);
        let attempt = scratch
            .create_session_attempt(
                ScratchAttemptId::new(&uuid::Uuid::new_v4().simple().to_string())
                    .expect("UUID simple form is a canonical scratch attempt identifier"),
            )
            .map_err(|error| infrastructure("trusted scratch attempt creation failed", error))?;
        let opened = DuckLakeAdapter::open_existing_filesystem_catalog_observed(
            context,
            &config,
            catalog_credential,
            &attempt,
            catalog_observations,
        )
        .map_err(|error| infrastructure("configured filesystem Lake open failed", error))?;
        return Ok((opened, Some(attempt), Vec::new(), data_path));
    }

    let location = verified_object_location(plan, binding)?;
    let storage_tls = match lake_plan.storage() {
        LakeStoragePlan::S3Compatible { tls, .. } => match tls {
            StorageTlsPlan::System => ObjectStorageTls::System,
            StorageTlsPlan::CustomCa { certificates } => ObjectStorageTls::CustomCa {
                certificates: certificates.clone(),
            },
        },
        _ => ObjectStorageTls::System,
    };
    let config = ObjectLakeSessionConfig::new(
        location,
        plan.postgresql().host(),
        plan.postgresql().port(),
        &binding.ducklake_catalog_database,
        catalog_schema,
        catalog_role,
        catalog_tls,
        storage_tls,
        encryption,
    )
    .map_err(|error| infrastructure("configured object-storage Lake inputs are invalid", error))?;
    let scratch = TrustedScratch::open(
        std::path::Path::new(plan.configuration_scratch_root()),
        std::iter::empty::<&std::path::Path>(),
    )
    .map_err(|error| infrastructure("trusted scratch validation failed", error))?;
    reap_session_scratch(&scratch);
    let attempt = scratch
        .create_session_attempt(
            ScratchAttemptId::new(&uuid::Uuid::new_v4().simple().to_string())
                .expect("UUID simple form is a canonical scratch attempt identifier"),
        )
        .map_err(|error| infrastructure("trusted scratch attempt creation failed", error))?;

    let mut versions = Vec::new();
    let opened = match lake_plan.storage().authentication().ok_or_else(|| {
        AppError::Validation("object-storage policy has no authentication capability".into())
    })? {
        StorageAuthenticationPlan::S3Static {
            access_key_id,
            secret_access_key,
        } => {
            let access = resolver.resolve_storage_credential(plan, access_key_id)?;
            let secret = resolver.resolve_storage_credential(plan, secret_access_key)?;
            versions.push(access.version().clone());
            versions.push(secret.version().clone());
            DuckLakeAdapter::open_existing_object_storage_catalog_observed(
                context,
                &config,
                catalog_credential,
                ObjectStorageAuthentication::S3Static {
                    access_key_id: access.material(),
                    secret_access_key: secret.material(),
                },
                &attempt,
                catalog_observations,
            )
        }
        StorageAuthenticationPlan::AwsWorkloadIdentity { source } => {
            let source = match source {
                ConfiguredAwsWorkloadIdentitySource::EcsTask => AwsWorkloadIdentitySource::EcsTask,
                ConfiguredAwsWorkloadIdentitySource::Ec2Instance => {
                    AwsWorkloadIdentitySource::Ec2Instance
                }
            };
            DuckLakeAdapter::open_existing_object_storage_catalog_observed(
                context,
                &config,
                catalog_credential,
                ObjectStorageAuthentication::AwsWorkloadIdentity { source },
                &attempt,
                catalog_observations,
            )
        }
        StorageAuthenticationPlan::AzureManagedIdentity { identity } => {
            let client_id = match identity {
                AzureManagedIdentity::SystemAssigned => None,
                AzureManagedIdentity::UserAssigned { client_id } => Some(*client_id),
            };
            DuckLakeAdapter::open_existing_object_storage_catalog_observed(
                context,
                &config,
                catalog_credential,
                ObjectStorageAuthentication::AzureManagedIdentity { client_id },
                &attempt,
                catalog_observations,
            )
        }
        StorageAuthenticationPlan::AzureServicePrincipal {
            tenant_id,
            client_id,
            client_secret,
        } => {
            let secret = resolver.resolve_storage_credential(plan, client_secret)?;
            versions.push(secret.version().clone());
            DuckLakeAdapter::open_existing_object_storage_catalog_observed(
                context,
                &config,
                catalog_credential,
                ObjectStorageAuthentication::AzureServicePrincipal {
                    tenant_id: *tenant_id,
                    client_id: *client_id,
                    client_secret: secret.material(),
                },
                &attempt,
                catalog_observations,
            )
        }
    }
    .map_err(|error| infrastructure("configured object-storage Lake open failed", error))?;
    let data_path = config.location().canonical_data_path();
    Ok((opened, Some(attempt), versions, data_path))
}

fn verified_object_location(
    plan: &DatastoreSessionPlan,
    binding: &DatastoreIdentityBinding,
) -> Result<ObjectStorageLocation, AppError> {
    if binding.storage_policy_id.as_deref() != Some(plan.storage_policy_id()) {
        return Err(AppError::Validation(
            "persisted object-storage policy identity does not match configuration".into(),
        ));
    }
    let persisted = binding.lake_location.as_ref().ok_or_else(|| {
        AppError::Validation("persisted Datastore has no structured Lake namespace".into())
    })?;
    let configured_prefix = plan.lake().datastore_prefix();
    let (location, base_prefix, actual_prefix) = match (plan.lake().storage(), persisted) {
        (
            LakeStoragePlan::AwsS3 {
                region,
                bucket,
                base_prefix,
                ..
            },
            DatastoreLakeLocation::AwsS3 {
                region: saved_region,
                bucket: saved_bucket,
                prefix,
            },
        ) if region == saved_region && bucket == saved_bucket => (
            ObjectStorageLocation::AwsS3 {
                region: saved_region.clone(),
                bucket: saved_bucket.clone(),
                prefix: prefix.clone(),
            },
            base_prefix.as_deref(),
            prefix.as_str(),
        ),
        (
            LakeStoragePlan::S3Compatible {
                endpoint,
                url_style,
                bucket,
                base_prefix,
                ..
            },
            DatastoreLakeLocation::S3Compatible {
                endpoint: saved_endpoint,
                url_style: saved_style,
                bucket: saved_bucket,
                prefix,
            },
        ) if endpoint == saved_endpoint
            && bucket == saved_bucket
            && saved_style
                == match url_style {
                    ConfiguredS3UrlStyle::Path => "path",
                    ConfiguredS3UrlStyle::VirtualHosted => "virtual_hosted",
                } =>
        {
            (
                ObjectStorageLocation::S3Compatible {
                    endpoint: saved_endpoint.clone(),
                    url_style: match url_style {
                        ConfiguredS3UrlStyle::Path => S3UrlStyle::Path,
                        ConfiguredS3UrlStyle::VirtualHosted => S3UrlStyle::VirtualHosted,
                    },
                    bucket: saved_bucket.clone(),
                    prefix: prefix.clone(),
                },
                base_prefix.as_deref(),
                prefix.as_str(),
            )
        }
        (
            LakeStoragePlan::AzureBlob {
                account_endpoint,
                container,
                base_prefix,
                ..
            },
            DatastoreLakeLocation::AzureBlob {
                account_endpoint: saved_endpoint,
                container: saved_container,
                prefix,
            },
        ) if account_endpoint == saved_endpoint && container == saved_container => (
            ObjectStorageLocation::AzureBlob {
                account_endpoint: saved_endpoint.clone(),
                container: saved_container.clone(),
                prefix: prefix.clone(),
            },
            base_prefix.as_deref(),
            prefix.as_str(),
        ),
        _ => {
            return Err(AppError::Validation(
                "persisted object-storage namespace does not match configuration".into(),
            ));
        }
    };
    if !prefix_matches_intent(
        base_prefix,
        configured_prefix,
        binding.datastore_id,
        actual_prefix,
    ) || binding.lake_path != location.canonical_data_path()
    {
        return Err(AppError::Validation(
            "persisted object-storage namespace or DuckLake data path does not match configuration"
                .into(),
        ));
    }
    Ok(location)
}

fn prefix_matches_intent(
    base: Option<&str>,
    datastore: Option<&str>,
    datastore_id: uuid::Uuid,
    actual: &str,
) -> bool {
    let generated;
    let datastore = match datastore {
        Some(configured) => configured,
        None => {
            generated = format!("datastores/{}", datastore_id.simple());
            &generated
        }
    };
    match base {
        Some(base) => actual == format!("{base}/{datastore}"),
        None => actual == datastore,
    }
}

fn resolve_binding(
    store: &mut ahri_tre_libpq_oauth::LibpqOAuthConnection,
    plan: &DatastoreSessionPlan,
) -> Result<DatastoreIdentityBinding, AppError> {
    if !datastore_identity_table_exists_oauth(store)
        .map_err(|error| infrastructure("check Datastore identity table", error))?
    {
        return Err(AppError::Validation(
            "configured Datastore has no identity binding".to_string(),
        ));
    }
    let bindings = read_datastore_identity_bindings_oauth(store)
        .map_err(|error| infrastructure("read Datastore identity binding", error))?;
    let [binding] = bindings.as_slice() else {
        return Err(AppError::Validation(
            "configured Datastore must have exactly one identity binding".to_string(),
        ));
    };
    validate_binding(binding, plan)?;
    Ok(binding.clone())
}

fn validate_binding(
    binding: &DatastoreIdentityBinding,
    plan: &DatastoreSessionPlan,
) -> Result<(), AppError> {
    if binding.lifecycle_state != DatastoreLifecycleState::Ready {
        return Err(AppError::Validation(
            "configured Datastore identity binding is not ready".to_string(),
        ));
    }
    if binding.datastore_name != plan.postgresql().database() {
        return Err(AppError::Validation(
            "configured Datastore database does not match the persisted Datastore identity".into(),
        ));
    }
    if binding.binding_fingerprint != binding.computed_binding_fingerprint() {
        return Err(AppError::Validation(
            "persisted Datastore identity fingerprint is invalid".to_string(),
        ));
    }
    Ok(())
}

fn validate_configured_physical_intent(
    configured: &ahri_tre_runtime::FilesystemLakeSessionPlan,
    resolved: &FilesystemLakeSessionConfig,
    binding: &DatastoreIdentityBinding,
) -> Result<(), AppError> {
    let mismatched = configured
        .datastore_prefix()
        .is_some_and(|value| value != resolved.datastore_prefix())
        || configured
            .catalog_database()
            .is_some_and(|value| value != binding.ducklake_catalog_database)
        || configured
            .catalog_schema()
            .is_some_and(|value| binding.ducklake_catalog_schema.as_deref() != Some(value))
        || configured
            .catalog_role()
            .is_some_and(|value| binding.lake_catalog_role_name.as_deref() != Some(value));
    if mismatched {
        Err(AppError::Validation(
            "configured physical Datastore intent does not match its persisted binding".into(),
        ))
    } else {
        Ok(())
    }
}

fn infrastructure(context: &str, error: impl std::fmt::Display) -> AppError {
    AppError::Infrastructure(format!("{context}: {error}"))
}

fn configured_postgresql_open_error(
    error: ahri_tre_libpq_oauth::LibpqOAuthError,
) -> ConfiguredDatastoreOpenError {
    match error.connect_failure_kind() {
        LibpqOAuthConnectFailureKind::AuthenticationRejected => {
            ConfiguredDatastoreOpenError::AuthenticationRejected
        }
        _ => ConfiguredDatastoreOpenError::Application(infrastructure(
            "configured PostgreSQL open failed",
            error,
        )),
    }
}

#[cfg(test)]
mod tests {
    use ahri_tre_config::{ConfigurationAuthority, EffectiveConfigurationTarget};
    use ahri_tre_orcid::TokenSource;
    use ahri_tre_runtime::DatastoreSessionPlan;
    use ahri_tre_types::{
        DatastoreIdentityBinding, DatastoreLakeLocation, DatastoreLifecycleState,
        LakeCatalogCredentialMode,
    };
    use chrono::{Duration, Utc};

    #[test]
    fn configured_postgresql_open_preserves_oauth_rejection_as_authorization() {
        let rejected = configured_postgresql_open_error(
            ahri_tre_libpq_oauth::LibpqOAuthError::AuthenticationRejected {
                token_source: "id_token",
            },
        );
        assert!(matches!(
            rejected,
            ConfiguredDatastoreOpenError::AuthenticationRejected
        ));

        let unavailable = configured_postgresql_open_error(
            ahri_tre_libpq_oauth::LibpqOAuthError::ConnectFailed {
                token_source: "id_token",
                message: "connection refused".to_string(),
            },
        );
        assert!(matches!(
            unavailable,
            ConfiguredDatastoreOpenError::Application(AppError::Infrastructure(_))
        ));
    }
    use secrecy::SecretString;

    use super::{
        AppError, ConfiguredDatastoreOpenError, ConfiguredSessionReference,
        configured_postgresql_open_error, prefix_matches_intent, validate_authentication,
        validate_authentication_identity, verified_object_location,
    };

    #[test]
    fn live_server_session_reference_does_not_request_durable_secret_retention() {
        let authentication: ahri_tre_runtime::SessionAuthenticationRecord =
            serde_json::from_value(serde_json::json!({
                "owner": "11111111-1111-4111-8111-111111111111",
                "identity": {
                    "issuer": "https://orcid.org",
                    "client_id": "ahri-tre-example",
                    "subject": "0000-0001-2345-6789"
                },
                "oauth": {
                    "artifact_id": "22222222-2222-4222-8222-222222222222",
                    "kind": "o_auth",
                    "version": 1,
                    "expires_at": "2030-01-01T00:00:00Z"
                }
            }))
            .unwrap();

        let (reference, identifier) = ConfiguredSessionReference::live(authentication);

        assert!(!reference.retains_snapshot());
        assert_eq!(reference.session_id(), identifier.session_id());
        assert!(reference.release_capability.is_none());
    }

    #[test]
    fn configured_open_rejects_expired_material_at_the_authentication_seam() {
        let effective = ConfigurationAuthority::parse_application(include_str!(
            "../../../examples/config/application-v1.toml"
        ))
        .unwrap()
        .resolve_effective(EffectiveConfigurationTarget::ExecutionProfile { explicit: None })
        .unwrap();
        let plan = DatastoreSessionPlan::try_from(&effective).unwrap();
        let authentication = ahri_tre_orcid::OAuthSession {
            id_token: SecretString::from("postgres-token-canary".to_string()),
            issuer: "https://orcid.org".to_string(),
            client_id: "ahri-tre-example".to_string(),
            subject: "subject".to_string(),
            orcid: "0000-0000-0000-0001".to_string(),
            role: "orcid_0000000000000001".to_string(),
            expires_at: Utc::now() - Duration::seconds(1),
            token_source: TokenSource::JupyterHub,
        };

        let rendered = format!("{authentication:?}");
        assert!(!rendered.contains("postgres-token-canary"));
        assert!(
            validate_authentication(&plan, &authentication)
                .expect_err("expired authentication must be rejected")
                .to_string()
                .contains("has expired")
        );
    }

    #[test]
    fn configured_open_rejects_an_oauth_subject_different_from_the_journal_identity() {
        let identity: ahri_tre_runtime::AuthenticationIdentityBinding =
            serde_json::from_value(serde_json::json!({
                "issuer": "https://orcid.org",
                "client_id": "ahri-tre-example",
                "subject": "subject-a"
            }))
            .unwrap();
        let authentication = ahri_tre_orcid::OAuthSession {
            id_token: SecretString::from("postgres-token-canary".to_string()),
            issuer: "https://orcid.org".to_string(),
            client_id: "ahri-tre-example".to_string(),
            subject: "subject-b".to_string(),
            orcid: "0000-0000-0000-0001".to_string(),
            role: "orcid_0000000000000001".to_string(),
            expires_at: Utc::now() + Duration::minutes(5),
            token_source: TokenSource::JupyterHub,
        };

        let error = validate_authentication_identity(&identity, &authentication)
            .expect_err("different OAuth subjects must be rejected before opening");
        assert!(error.to_string().contains("persisted Session identity"));
    }

    #[test]
    fn configured_object_open_requires_the_exact_persisted_service_authority() {
        let document = include_str!("../../../examples/config/application-v1.toml").replace(
            "[storage_policies.filesystem]\nkind = \"filesystem\"\nbase_path = \"/srv/ahri-tre/lake\"",
            "[storage_authentication_policies.object]\n\
             kind = \"s3_static\"\n\
             access_key_id = { uri = \"managed://storage/access-key\" }\n\
             secret_access_key = { uri = \"managed://storage/secret-key\" }\n\n\
             [storage_policies.filesystem]\n\
             kind = \"s3_compatible\"\n\
             endpoint = \"https://rustfs.example.test\"\n\
             url_style = \"path\"\n\
             bucket = \"ahri-research\"\n\
             base_prefix = \"governed\"\n\
             authentication_policy = \"object\"",
        );
        let effective = ConfigurationAuthority::parse_application(&document)
            .unwrap()
            .resolve_effective(EffectiveConfigurationTarget::ExecutionProfile { explicit: None })
            .unwrap();
        let plan = DatastoreSessionPlan::try_from(&effective).unwrap();
        let mut binding = DatastoreIdentityBinding {
            datastore_id: uuid::Uuid::new_v4(),
            datastore_name: "research".into(),
            ducklake_catalog_database: "ducklake_catalog".into(),
            ducklake_catalog_schema: Some("catalog".into()),
            lake_path: "s3://ahri-research/governed/datasets/".into(),
            storage_policy_id: Some("filesystem".into()),
            lake_location: Some(DatastoreLakeLocation::S3Compatible {
                endpoint: "https://rustfs.example.test".into(),
                url_style: "path".into(),
                bucket: "ahri-research".into(),
                prefix: "governed/datasets".into(),
            }),
            ducklake_encryption: false,
            lake_catalog_credential_mode: LakeCatalogCredentialMode::ManagedLocal,
            lake_catalog_role_name: Some("catalog_owner".into()),
            managed_secret_ref: None,
            credential_version: 1,
            credential_last_rotated_at: None,
            created_at: Utc::now(),
            creating_tool_version: None,
            binding_fingerprint: String::new(),
            lifecycle_state: DatastoreLifecycleState::Ready,
            failure_code: None,
            failure_summary: None,
        };
        binding.binding_fingerprint = binding.computed_binding_fingerprint();

        let location = verified_object_location(&plan, &binding).unwrap();
        assert_eq!(
            location.canonical_data_path(),
            "s3://ahri-research/governed/datasets/"
        );

        let Some(DatastoreLakeLocation::S3Compatible { endpoint, .. }) =
            binding.lake_location.as_mut()
        else {
            panic!("test binding must use S3-compatible storage");
        };
        *endpoint = "https://other.example.test".into();
        assert!(
            verified_object_location(&plan, &binding)
                .unwrap_err()
                .to_string()
                .contains("namespace does not match")
        );
    }

    #[test]
    fn omitted_object_prefix_requires_the_uuid_derived_namespace() {
        let datastore_id = uuid::Uuid::parse_str("10000000-0000-4000-8000-000000000001").unwrap();
        assert!(prefix_matches_intent(
            Some("governed"),
            None,
            datastore_id,
            "governed/datastores/10000000000040008000000000000001"
        ));
        assert!(!prefix_matches_intent(
            Some("governed"),
            None,
            datastore_id,
            "governed/arbitrary"
        ));
    }
}