ahri_tre_lake/

adapter.rs

use std::fs::{self, OpenOptions};
use std::io::{Read, Write};
#[cfg(unix)]
use std::os::unix::fs::OpenOptionsExt;
use std::path::PathBuf;

use ahri_tre_secrets::{MAX_SECRET_MATERIAL_BYTES, SecretMaterial};
use ahri_tre_security::{REDACTED_VALUE, legacy_object_storage_environment_variable};
use ahri_tre_types::{DataStoreProfile, EncryptionMode};
use duckdb::OptionalExt;

use crate::LakeCatalogObservations;
use crate::alias::LAKE_ALIAS;
use crate::error::LakeError;

/// Adapter rooted at a TRE Lake location.
#[derive(Debug, Clone)]
pub struct DuckLakeAdapter {
    pub lake_root: String,
}

/// Explicit filesystem and PostgreSQL catalog inputs for one existing Lake.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FilesystemLakeSessionConfig {
    filesystem_base: String,
    datastore_prefix: String,
    catalog_host: String,
    catalog_port: u16,
    catalog_database: String,
    catalog_schema: String,
    catalog_role: String,
    tls: FilesystemCatalogTls,
    encryption_mode: EncryptionMode,
}

/// Explicit PostgreSQL trust used by DuckLake's catalog connection.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FilesystemCatalogTls {
    System,
    CustomCa { certificates: Vec<String> },
}

/// Canonical object-storage namespace owned by one Datastore binding.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ObjectStorageLocation {
    AwsS3 {
        region: String,
        bucket: String,
        prefix: String,
    },
    S3Compatible {
        endpoint: String,
        url_style: S3UrlStyle,
        bucket: String,
        prefix: String,
    },
    AzureBlob {
        account_endpoint: String,
        container: String,
        prefix: String,
    },
}

impl ObjectStorageLocation {
    pub fn aws_s3_for_datastore(
        region: impl Into<String>,
        bucket: impl Into<String>,
        base_prefix: Option<&str>,
        datastore_prefix: &str,
    ) -> Result<Self, LakeError> {
        Ok(Self::AwsS3 {
            region: region.into(),
            bucket: bucket.into(),
            prefix: datastore_object_prefix(base_prefix, datastore_prefix)?,
        })
    }

    pub fn s3_compatible_for_datastore(
        endpoint: impl Into<String>,
        url_style: S3UrlStyle,
        bucket: impl Into<String>,
        base_prefix: Option<&str>,
        datastore_prefix: &str,
    ) -> Result<Self, LakeError> {
        Ok(Self::S3Compatible {
            endpoint: endpoint.into(),
            url_style,
            bucket: bucket.into(),
            prefix: datastore_object_prefix(base_prefix, datastore_prefix)?,
        })
    }

    pub fn azure_blob_for_datastore(
        account_endpoint: impl Into<String>,
        container: impl Into<String>,
        base_prefix: Option<&str>,
        datastore_prefix: &str,
    ) -> Result<Self, LakeError> {
        Ok(Self::AzureBlob {
            account_endpoint: account_endpoint.into(),
            container: container.into(),
            prefix: datastore_object_prefix(base_prefix, datastore_prefix)?,
        })
    }

    pub fn canonical_data_path(&self) -> String {
        match self {
            Self::AwsS3 { bucket, prefix, .. } | Self::S3Compatible { bucket, prefix, .. } => {
                format!("s3://{bucket}/{prefix}/")
            }
            Self::AzureBlob {
                account_endpoint,
                container,
                prefix,
            } => {
                let authority = account_endpoint
                    .strip_prefix("https://")
                    .unwrap_or(account_endpoint);
                format!("az://{authority}/{container}/{prefix}/")
            }
        }
    }

    pub fn service_authority(&self) -> &str {
        match self {
            Self::AwsS3 { region, .. } => region,
            Self::S3Compatible { endpoint, .. } => endpoint,
            Self::AzureBlob {
                account_endpoint, ..
            } => account_endpoint,
        }
    }

    fn validate(&self) -> Result<(), LakeError> {
        let (namespace, prefix) = match self {
            Self::AwsS3 {
                region,
                bucket,
                prefix,
            } => {
                if region.trim().is_empty() {
                    return Err(LakeError::InvalidObjectStorageIdentity);
                }
                (bucket, prefix)
            }
            Self::S3Compatible {
                endpoint,
                bucket,
                prefix,
                ..
            } => {
                if !endpoint.starts_with("https://") || endpoint.ends_with('/') {
                    return Err(LakeError::InvalidObjectStorageIdentity);
                }
                (bucket, prefix)
            }
            Self::AzureBlob {
                account_endpoint,
                container,
                prefix,
            } => {
                if !account_endpoint.starts_with("https://") || account_endpoint.ends_with('/') {
                    return Err(LakeError::InvalidObjectStorageIdentity);
                }
                (container, prefix)
            }
        };
        if namespace.trim().is_empty() || !canonical_datastore_prefix(prefix) {
            return Err(LakeError::InvalidObjectStorageIdentity);
        }
        Ok(())
    }
}

fn datastore_object_prefix(
    base_prefix: Option<&str>,
    datastore_prefix: &str,
) -> Result<String, LakeError> {
    if !canonical_datastore_prefix(datastore_prefix) {
        return Err(LakeError::InvalidDatastorePrefix);
    }
    let base = base_prefix.map(str::trim).unwrap_or("").trim_matches('/');
    if base.is_empty() {
        Ok(datastore_prefix.to_string())
    } else {
        Ok(format!("{base}/{datastore_prefix}"))
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum S3UrlStyle {
    Path,
    VirtualHosted,
}

/// The single AWS platform metadata source authorized by configuration.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AwsWorkloadIdentitySource {
    EcsTask,
    Ec2Instance,
}

/// Explicit object-storage TLS trust. Verification cannot be disabled.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ObjectStorageTls {
    System,
    CustomCa { certificates: Vec<String> },
}

/// Explicit catalog and object namespace inputs for one existing Lake.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ObjectLakeSessionConfig {
    location: ObjectStorageLocation,
    catalog_host: String,
    catalog_port: u16,
    catalog_database: String,
    catalog_schema: String,
    catalog_role: String,
    catalog_tls: FilesystemCatalogTls,
    storage_tls: ObjectStorageTls,
    encryption_mode: EncryptionMode,
}

impl ObjectLakeSessionConfig {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        location: ObjectStorageLocation,
        catalog_host: impl Into<String>,
        catalog_port: u16,
        catalog_database: impl Into<String>,
        catalog_schema: impl Into<String>,
        catalog_role: impl Into<String>,
        catalog_tls: FilesystemCatalogTls,
        storage_tls: ObjectStorageTls,
        encryption_mode: EncryptionMode,
    ) -> Result<Self, LakeError> {
        location.validate()?;
        let config = Self {
            location,
            catalog_host: catalog_host.into(),
            catalog_port,
            catalog_database: catalog_database.into(),
            catalog_schema: catalog_schema.into(),
            catalog_role: catalog_role.into(),
            catalog_tls,
            storage_tls,
            encryption_mode,
        };
        if config.catalog_host.trim().is_empty()
            || config.catalog_port == 0
            || config.catalog_database.trim().is_empty()
            || config.catalog_schema.trim().is_empty()
            || config.catalog_role.trim().is_empty()
        {
            return Err(LakeError::IncompleteCatalogIdentity);
        }
        Ok(config)
    }

    pub fn location(&self) -> &ObjectStorageLocation {
        &self.location
    }

    fn legacy_profile(&self) -> DataStoreProfile {
        DataStoreProfile {
            server: self.catalog_host.clone(),
            port: self.catalog_port,
            dbname: self.catalog_database.clone(),
            sslmode: Some("verify-full".to_string()),
            lake: ahri_tre_types::LakeAttachProfile {
                lake_data: self.location.canonical_data_path(),
                lake_db: self.catalog_database.clone(),
                catalog_schema: Some(self.catalog_schema.clone()),
                auth: ahri_tre_types::LakeAuthProfile {
                    lake_user: self.catalog_role.clone(),
                    encryption_mode: self.encryption_mode,
                },
            },
        }
    }
}

/// Connection-scoped storage authentication. Secret material has no formatting
/// or serialization authority and is consumed only while constructing a DuckDB
/// temporary secret.
#[derive(Clone, Copy)]
pub enum ObjectStorageAuthentication<'a> {
    S3Static {
        access_key_id: &'a SecretMaterial,
        secret_access_key: &'a SecretMaterial,
    },
    AwsWorkloadIdentity {
        source: AwsWorkloadIdentitySource,
    },
    AzureManagedIdentity {
        client_id: Option<uuid::Uuid>,
    },
    AzureServicePrincipal {
        tenant_id: uuid::Uuid,
        client_id: uuid::Uuid,
        client_secret: &'a SecretMaterial,
    },
}

impl std::fmt::Debug for ObjectStorageAuthentication<'_> {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::S3Static { .. } => formatter.write_str("S3Static(<redacted>)"),
            Self::AwsWorkloadIdentity { source } => formatter
                .debug_struct("AwsWorkloadIdentity")
                .field("source", source)
                .finish(),
            Self::AzureManagedIdentity { client_id } => formatter
                .debug_struct("AzureManagedIdentity")
                .field("client_id", client_id)
                .finish(),
            Self::AzureServicePrincipal {
                tenant_id,
                client_id,
                ..
            } => formatter
                .debug_struct("AzureServicePrincipal")
                .field("tenant_id", tenant_id)
                .field("client_id", client_id)
                .field("client_secret", &REDACTED_VALUE)
                .finish(),
        }
    }
}

impl FilesystemLakeSessionConfig {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        filesystem_base: impl Into<String>,
        datastore_prefix: impl Into<String>,
        catalog_host: impl Into<String>,
        catalog_port: u16,
        catalog_database: impl Into<String>,
        catalog_schema: impl Into<String>,
        catalog_role: impl Into<String>,
        tls: FilesystemCatalogTls,
        encryption_mode: EncryptionMode,
    ) -> Result<Self, LakeError> {
        let config = Self {
            filesystem_base: filesystem_base.into(),
            datastore_prefix: datastore_prefix.into(),
            catalog_host: catalog_host.into(),
            catalog_port,
            catalog_database: catalog_database.into(),
            catalog_schema: catalog_schema.into(),
            catalog_role: catalog_role.into(),
            tls,
            encryption_mode,
        };
        normalized_root(&config.filesystem_base)?;
        if !canonical_datastore_prefix(&config.datastore_prefix) {
            return Err(LakeError::InvalidDatastorePrefix);
        }
        if config.catalog_host.trim().is_empty()
            || config.catalog_port == 0
            || config.catalog_database.trim().is_empty()
            || config.catalog_schema.trim().is_empty()
            || config.catalog_role.trim().is_empty()
        {
            return Err(LakeError::IncompleteCatalogIdentity);
        }
        Ok(config)
    }

    /// Validates a datastore-owned canonical path and derives its relative prefix.
    #[allow(clippy::too_many_arguments)]
    pub fn from_binding(
        filesystem_base: impl Into<String>,
        canonical_data_path: impl Into<String>,
        catalog_host: impl Into<String>,
        catalog_port: u16,
        catalog_database: impl Into<String>,
        catalog_schema: impl Into<String>,
        catalog_role: impl Into<String>,
        tls: FilesystemCatalogTls,
        encryption_mode: EncryptionMode,
    ) -> Result<Self, LakeError> {
        let filesystem_base = filesystem_base.into();
        let canonical_data_path = canonical_data_path.into();
        let base = normalized_root(&filesystem_base)?;
        let data = normalized_root(&canonical_data_path)?;
        let prefix = if base == "/" {
            data.strip_prefix('/')
        } else {
            data.strip_prefix(base)
                .and_then(|suffix| suffix.strip_prefix('/'))
        }
        .filter(|prefix| canonical_datastore_prefix(prefix))
        .ok_or(LakeError::InvalidDatastorePrefix)?;
        Self::new(
            filesystem_base,
            prefix,
            catalog_host,
            catalog_port,
            catalog_database,
            catalog_schema,
            catalog_role,
            tls,
            encryption_mode,
        )
    }

    pub fn filesystem_base(&self) -> &str {
        &self.filesystem_base
    }

    pub fn datastore_prefix(&self) -> &str {
        &self.datastore_prefix
    }

    pub fn canonical_data_path(&self) -> String {
        if self.filesystem_base == "/" {
            format!("/{}", self.datastore_prefix)
        } else {
            format!(
                "{}/{}",
                self.filesystem_base.trim_end_matches('/'),
                self.datastore_prefix
            )
        }
    }

    fn legacy_profile(&self) -> DataStoreProfile {
        DataStoreProfile {
            server: self.catalog_host.clone(),
            port: self.catalog_port,
            dbname: self.catalog_database.clone(),
            sslmode: Some("verify-full".to_string()),
            lake: ahri_tre_types::LakeAttachProfile {
                lake_data: self.canonical_data_path(),
                lake_db: self.catalog_database.clone(),
                catalog_schema: Some(self.catalog_schema.clone()),
                auth: ahri_tre_types::LakeAuthProfile {
                    lake_user: self.catalog_role.clone(),
                    encryption_mode: self.encryption_mode,
                },
            },
        }
    }
}

/// Local directories used by DuckDB and DuckLake for a TRE lake.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DuckLakeLocalLayout {
    pub data_path: String,
    pub staging_path: String,
    pub tmp_path: String,
    pub test_runs_path: String,
}

/// Rendered DuckLake `ATTACH` statement and its non-secret description.
#[derive(Clone, PartialEq, Eq)]
pub struct DuckLakeAttachPlan {
    pub(crate) metadata_schema: String,
    pub alias: String,
    pub attach_description: String,
    pub(crate) pre_attach_sql: Vec<String>,
    pub(crate) attach_sql: String,
    pub data_path: String,
    pub credential_kind: DuckLakeCredentialKind,
    pub create_if_not_exists: bool,
    pub automatic_migration: bool,
    pub configured_encryption_mode: EncryptionMode,
}

impl std::fmt::Debug for DuckLakeAttachPlan {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DuckLakeAttachPlan")
            .field("alias", &self.alias)
            .field("attach_description", &self.attach_description)
            .field("pre_attach_sql", &REDACTED_VALUE)
            .field("attach_sql", &REDACTED_VALUE)
            .field("data_path", &self.data_path)
            .field("credential_kind", &self.credential_kind)
            .field("create_if_not_exists", &self.create_if_not_exists)
            .field("automatic_migration", &self.automatic_migration)
            .field(
                "configured_encryption_mode",
                &self.configured_encryption_mode,
            )
            .finish()
    }
}

/// Open DuckDB connection with the TRE DuckLake catalog attached.
pub struct DuckLakeOpenedCatalog {
    pub connection: duckdb::Connection,
    pub attach_plan: DuckLakeAttachPlan,
    pub local_layout: DuckLakeLocalLayout,
    pub health: DuckLakeHealth,
    pub tls_guard: Option<DuckLakeTlsGuard>,
}

impl std::fmt::Debug for DuckLakeOpenedCatalog {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("DuckLakeOpenedCatalog")
            .field("attach_plan", &self.attach_plan)
            .field("local_layout", &self.local_layout)
            .field("health", &self.health)
            .finish_non_exhaustive()
    }
}

pub struct DuckLakeTlsGuard {
    paths: Vec<PathBuf>,
}

impl std::fmt::Debug for DuckLakeTlsGuard {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("DuckLakeTlsGuard(<adapter-owned>)")
    }
}

impl Drop for DuckLakeTlsGuard {
    fn drop(&mut self) {
        for path in &self.paths {
            let _ = fs::remove_file(path);
        }
    }
}

impl DuckLakeTlsGuard {
    fn create(certificates: &[String]) -> Result<Self, LakeError> {
        if certificates.is_empty() {
            return Err(LakeError::IncompleteCatalogIdentity);
        }
        let path = std::env::temp_dir().join(format!(
            "ahri-tre-ducklake-ca-{}.pem",
            uuid::Uuid::new_v4().simple()
        ));
        let mut options = OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        options.mode(0o600);
        let mut file = options.open(&path)?;
        let projected = Self { paths: vec![path] };
        for certificate in certificates {
            file.write_all(certificate.as_bytes())?;
            if !certificate.ends_with('\n') {
                file.write_all(b"\n")?;
            }
        }
        file.sync_all()?;
        Ok(projected)
    }

    fn path(&self) -> &std::path::Path {
        &self.paths[0]
    }

    fn combine(mut self, mut other: Self) -> Self {
        self.paths.append(&mut other.paths);
        self
    }
}

/// Basic health and catalog-state information for an attached DuckLake catalog.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DuckLakeHealth {
    pub alias: String,
    pub catalog_type: String,
    pub extension_version: String,
    pub data_path: String,
    pub detected_encryption_mode: EncryptionMode,
    pub snapshot_count: i64,
    pub table_count: i64,
}

/// Credential material used to connect DuckLake to its PostgreSQL catalog.
#[derive(Clone, PartialEq, Eq)]
enum DuckLakeCatalogCredential {
    SessionSecret {
        secret_name: String,
        password: String,
    },
}

impl std::fmt::Debug for DuckLakeCatalogCredential {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::SessionSecret { secret_name, .. } => f
                .debug_struct("SessionSecret")
                .field("secret_name", secret_name)
                .field("password", &REDACTED_VALUE)
                .finish(),
        }
    }
}

/// Non-secret credential category recorded in attach plans.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DuckLakeCredentialKind {
    SessionSecret,
}

/// Options controlling whether DuckLake creates or migrates catalog metadata.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DuckLakeOpenOptions {
    pub create_if_not_exists: bool,
    pub automatic_migration: bool,
    pub apply_configured_encryption: bool,
}

impl Default for DuckLakeOpenOptions {
    fn default() -> Self {
        Self {
            create_if_not_exists: true,
            automatic_migration: false,
            apply_configured_encryption: true,
        }
    }
}

impl DuckLakeOpenOptions {
    /// Options for first-time catalog initialization.
    pub const fn initialize() -> Self {
        Self {
            create_if_not_exists: true,
            automatic_migration: false,
            apply_configured_encryption: true,
        }
    }

    /// Options for opening an existing catalog without creating or migrating it.
    pub const fn attach_existing() -> Self {
        Self {
            create_if_not_exists: false,
            automatic_migration: false,
            apply_configured_encryption: false,
        }
    }

    /// Enables or disables DuckLake automatic migration for this open.
    pub const fn with_automatic_migration(mut self, automatic_migration: bool) -> Self {
        self.automatic_migration = automatic_migration;
        self
    }
}

impl DuckLakeCatalogCredential {
    /// Builds a session-local DuckDB secret credential.
    fn session_secret(secret_name: impl Into<String>, password: impl Into<String>) -> Self {
        Self::SessionSecret {
            secret_name: secret_name.into(),
            password: password.into(),
        }
    }

    fn kind(&self) -> DuckLakeCredentialKind {
        match self {
            Self::SessionSecret { .. } => DuckLakeCredentialKind::SessionSecret,
        }
    }

    fn validate(&self) -> Result<(), LakeError> {
        let (kind, value) = match self {
            Self::SessionSecret {
                secret_name,
                password,
            } => {
                if secret_name.trim().is_empty() {
                    return Err(LakeError::EmptyCredential {
                        kind: "session_secret",
                    });
                }
                ("session_secret_password", password)
            }
        };

        if value.trim().is_empty() {
            return Err(LakeError::EmptyCredential { kind });
        }

        Ok(())
    }
}

impl DuckLakeAdapter {
    /// Validates object-storage provider and ambient-authority inputs without
    /// rendering Secret-bearing SQL.
    pub fn validate_object_storage_inputs<'a>(
        config: &ObjectLakeSessionConfig,
        storage_authentication: ObjectStorageAuthentication<'a>,
        ambient_environment_names: impl IntoIterator<Item = impl AsRef<str>>,
    ) -> Result<(), LakeError> {
        let ambient_environment_names = ambient_environment_names
            .into_iter()
            .map(|name| name.as_ref().to_string())
            .collect::<Vec<_>>();
        reject_ambient_storage_authority(
            ambient_environment_names.iter().map(String::as_str),
            &storage_authentication,
        )?;
        validate_storage_authentication(config.location(), &storage_authentication)
    }

    /// Opens an existing object-backed Lake with explicit catalog and storage
    /// credentials and a private trusted-scratch attempt.
    pub fn open_existing_object_storage_catalog(
        config: &ObjectLakeSessionConfig,
        catalog_credential: &SecretMaterial,
        storage_authentication: ObjectStorageAuthentication<'_>,
        scratch: &crate::ScratchAttempt,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        Self::open_object_storage_catalog(
            config,
            catalog_credential,
            storage_authentication,
            scratch,
            DuckLakeOpenOptions::attach_existing(),
            &mut LakeCatalogObservations::default(),
        )
    }

    /// Observes the existing attach; it never probes or retries independently.
    pub fn open_existing_object_storage_catalog_observed(
        context: Option<&ahri_tre_observability::CorrelationContext>,
        config: &ObjectLakeSessionConfig,
        catalog_credential: &SecretMaterial,
        storage_authentication: ObjectStorageAuthentication<'_>,
        scratch: &crate::ScratchAttempt,
        observations: &mut LakeCatalogObservations,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        let span = context.map(|context| context.span(ahri_tre_observability::Stage::Lake));
        let result = Self::open_object_storage_catalog(
            config,
            catalog_credential,
            storage_authentication,
            scratch,
            DuckLakeOpenOptions::attach_existing(),
            observations,
        );
        finish_lake_open(span, &result);
        result
    }

    /// Initializes a new object-backed Lake after proving its configured prefix is empty.
    pub fn initialize_object_storage_catalog(
        config: &ObjectLakeSessionConfig,
        catalog_credential: &SecretMaterial,
        storage_authentication: ObjectStorageAuthentication<'_>,
        scratch: &crate::ScratchAttempt,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        Self::open_object_storage_catalog(
            config,
            catalog_credential,
            storage_authentication,
            scratch,
            DuckLakeOpenOptions::initialize(),
            &mut LakeCatalogObservations::default(),
        )
    }

    fn open_object_storage_catalog(
        config: &ObjectLakeSessionConfig,
        catalog_credential: &SecretMaterial,
        storage_authentication: ObjectStorageAuthentication<'_>,
        scratch: &crate::ScratchAttempt,
        options: DuckLakeOpenOptions,
        observations: &mut LakeCatalogObservations,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        reject_ambient_storage_authority(
            std::env::vars_os().map(|(name, _)| name.to_string_lossy().into_owned()),
            &storage_authentication,
        )?;
        validate_storage_authentication(config.location(), &storage_authentication)?;

        let catalog_guard = match &config.catalog_tls {
            FilesystemCatalogTls::System => None,
            FilesystemCatalogTls::CustomCa { certificates } => {
                Some(DuckLakeTlsGuard::create(certificates)?)
            }
        };
        let storage_guard = match &config.storage_tls {
            ObjectStorageTls::System => None,
            ObjectStorageTls::CustomCa { certificates } => {
                Some(DuckLakeTlsGuard::create(certificates)?)
            }
        };
        let catalog_root = catalog_guard
            .as_ref()
            .map(|guard| guard.path().to_string_lossy().into_owned())
            .unwrap_or_else(|| "system".to_string());

        let profile = config.legacy_profile();
        let adapter = Self::new(config.location.canonical_data_path());
        let mut plan = catalog_credential.expose(|bytes| {
            let password =
                std::str::from_utf8(bytes).map_err(|_| LakeError::InvalidCredentialEncoding)?;
            let mut plan = adapter.build_attach_plan_with_credential_and_root(
                &profile,
                DuckLakeCatalogCredential::session_secret("ahri_tre_ducklake_catalog", password),
                options,
                Some(&catalog_root),
            )?;
            plan.data_path = config.location().canonical_data_path();
            plan.pre_attach_sql.insert(
                0,
                storage_secret_sql(config.location(), storage_authentication)?,
            );
            if let Some(guard) = storage_guard.as_ref() {
                plan.pre_attach_sql.insert(
                    0,
                    format!(
                        "SET ca_cert_file = {}; SET enable_server_cert_verification = true",
                        sql_literal(&guard.path().to_string_lossy())
                    ),
                );
            }
            Ok::<_, LakeError>(plan)
        })?;

        let connection = duckdb::Connection::open_in_memory()?;
        connection.execute_batch(&format!(
            "SET temp_directory = {}",
            sql_literal(&scratch.path().to_string_lossy())
        ))?;
        connection.execute_batch(
            "INSTALL postgres; LOAD postgres; INSTALL ducklake; LOAD ducklake; INSTALL httpfs; LOAD httpfs;",
        )?;
        if matches!(config.location(), ObjectStorageLocation::AzureBlob { .. }) {
            connection.execute_batch("INSTALL azure; LOAD azure;")?;
        }
        for statement in &plan.pre_attach_sql {
            connection
                .execute_batch(statement)
                .map_err(|_| LakeError::CredentialProjectionFailed)?;
        }
        if options.create_if_not_exists {
            let pattern = object_namespace_glob(config.location());
            let count: i64 = connection.query_row(
                &format!("SELECT count(*) FROM glob({})", sql_literal(&pattern)),
                [],
                |row| row.get(0),
            )?;
            if count != 0 {
                return Err(LakeError::NamespaceNotEmpty);
            }
        }
        observations.attach(&connection, &plan.attach_sql, &config.catalog_role)?;
        #[cfg(feature = "acceptance-catalog-barrier")]
        crate::catalog_observations::after_attach_barrier()?;
        connection.execute_batch(&format!("USE {LAKE_ALIAS};"))?;
        let health = observations.health_check(&adapter, &connection)?;
        if health.data_path != config.location().canonical_data_path() {
            return Err(LakeError::PersistedDataPathMismatch);
        }
        let tls_guard = match (catalog_guard, storage_guard) {
            (Some(first), Some(second)) => Some(first.combine(second)),
            (Some(guard), None) | (None, Some(guard)) => Some(guard),
            (None, None) => None,
        };
        plan.data_path = config.location().canonical_data_path();
        Ok(DuckLakeOpenedCatalog {
            connection,
            attach_plan: plan,
            local_layout: DuckLakeLocalLayout {
                data_path: config.location().canonical_data_path(),
                staging_path: String::new(),
                tmp_path: String::new(),
                test_runs_path: String::new(),
            },
            health,
            tls_guard,
        })
    }

    /// Builds an existing-catalog attach plan from explicit filesystem and
    /// credential inputs. Lake location composition remains inside this adapter.
    pub fn build_filesystem_attach_plan(
        config: &FilesystemLakeSessionConfig,
        credential: &SecretMaterial,
    ) -> Result<DuckLakeAttachPlan, LakeError> {
        if matches!(config.tls, FilesystemCatalogTls::CustomCa { .. }) {
            return Err(LakeError::CustomTlsRequiresOpen);
        }
        let profile = config.legacy_profile();
        let adapter = Self::new(config.canonical_data_path());
        credential.expose(|bytes| {
            let password =
                std::str::from_utf8(bytes).map_err(|_| LakeError::InvalidCredentialEncoding)?;
            adapter.build_attach_plan_with_credential_and_root(
                &profile,
                DuckLakeCatalogCredential::session_secret("ahri_tre_ducklake_catalog", password),
                DuckLakeOpenOptions::attach_existing(),
                Some("system"),
            )
        })
    }

    /// Opens an existing filesystem Lake from explicit, non-ambient inputs.
    pub fn open_existing_filesystem_catalog(
        config: &FilesystemLakeSessionConfig,
        credential: &SecretMaterial,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        Self::open_existing_filesystem_catalog_inner(
            config,
            credential,
            None,
            &mut LakeCatalogObservations::default(),
        )
    }

    fn open_existing_filesystem_catalog_inner(
        config: &FilesystemLakeSessionConfig,
        credential: &SecretMaterial,
        scratch: Option<&std::path::Path>,
        observations: &mut LakeCatalogObservations,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        let profile = config.legacy_profile();
        let adapter = Self::new(config.canonical_data_path());
        let root_certificate = match &config.tls {
            FilesystemCatalogTls::System => None,
            FilesystemCatalogTls::CustomCa { certificates } => {
                Some(DuckLakeTlsGuard::create(certificates)?)
            }
        };
        let root = match &root_certificate {
            Some(certificate) => certificate.path().to_string_lossy().into_owned(),
            None => "system".to_string(),
        };
        credential.expose(|bytes| {
            let password =
                std::str::from_utf8(bytes).map_err(|_| LakeError::InvalidCredentialEncoding)?;
            let mut opened = adapter.open_attached_catalog_with_credential_and_root(
                &profile,
                DuckLakeCatalogCredential::session_secret("ahri_tre_ducklake_catalog", password),
                DuckLakeOpenOptions::attach_existing(),
                Some(&root),
                scratch,
                observations,
            )?;
            opened.tls_guard = root_certificate;
            Ok(opened)
        })
    }

    /// Opens an existing filesystem Lake while directing all DuckDB temporary
    /// work into one private trusted-scratch attempt rather than the Lake.
    pub fn open_existing_filesystem_catalog_with_scratch(
        config: &FilesystemLakeSessionConfig,
        credential: &SecretMaterial,
        scratch: &crate::ScratchAttempt,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        let mut opened = Self::open_existing_filesystem_catalog_inner(
            config,
            credential,
            Some(scratch.path()),
            &mut LakeCatalogObservations::default(),
        )?;
        opened.local_layout.staging_path.clear();
        opened.local_layout.tmp_path.clear();
        opened.local_layout.test_runs_path.clear();
        Ok(opened)
    }

    /// Records the time authenticated catalog evidence was actually obtained.
    pub fn open_existing_filesystem_catalog_observed(
        context: Option<&ahri_tre_observability::CorrelationContext>,
        config: &FilesystemLakeSessionConfig,
        credential: &SecretMaterial,
        scratch: &crate::ScratchAttempt,
        observations: &mut LakeCatalogObservations,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        let span = context.map(|context| context.span(ahri_tre_observability::Stage::Lake));
        let result = Self::open_existing_filesystem_catalog_inner(
            config,
            credential,
            Some(scratch.path()),
            observations,
        )
        .map(|mut opened| {
            opened.local_layout.staging_path.clear();
            opened.local_layout.tmp_path.clear();
            opened.local_layout.test_runs_path.clear();
            opened
        });
        finish_lake_open(span, &result);
        result
    }

    /// Initializes a new filesystem Lake in a previously absent namespace.
    pub fn initialize_filesystem_catalog_with_scratch(
        config: &FilesystemLakeSessionConfig,
        credential: &SecretMaterial,
        scratch: &crate::ScratchAttempt,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        match fs::symlink_metadata(config.canonical_data_path()) {
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Ok(_) => return Err(LakeError::NamespaceNotEmpty),
            Err(error) => return Err(error.into()),
        }
        let profile = config.legacy_profile();
        let adapter = Self::new(config.canonical_data_path());
        let root_certificate = match &config.tls {
            FilesystemCatalogTls::System => None,
            FilesystemCatalogTls::CustomCa { certificates } => {
                Some(DuckLakeTlsGuard::create(certificates)?)
            }
        };
        let root = root_certificate
            .as_ref()
            .map(|certificate| certificate.path().to_string_lossy().into_owned())
            .unwrap_or_else(|| "system".to_string());
        let mut opened = credential.expose(|bytes| {
            let password =
                std::str::from_utf8(bytes).map_err(|_| LakeError::InvalidCredentialEncoding)?;
            adapter.open_attached_catalog_with_credential_and_root(
                &profile,
                DuckLakeCatalogCredential::session_secret("ahri_tre_ducklake_catalog", password),
                DuckLakeOpenOptions::initialize(),
                Some(&root),
                Some(scratch.path()),
                &mut LakeCatalogObservations::default(),
            )
        })?;
        opened.tls_guard = root_certificate;
        opened.local_layout.staging_path.clear();
        opened.local_layout.tmp_path.clear();
        opened.local_layout.test_runs_path.clear();
        Ok(opened)
    }

    /// Creates an adapter for the supplied lake root.
    pub fn new(lake_root: impl Into<String>) -> Self {
        Self {
            lake_root: lake_root.into(),
        }
    }

    /// Returns the deterministic local layout below the lake root.
    ///
    /// # Errors
    ///
    /// Returns [`LakeError::EmptyLakeRoot`] if the configured root is blank.
    pub fn local_layout(&self) -> Result<DuckLakeLocalLayout, LakeError> {
        let root = normalized_root(&self.lake_root)?;

        Ok(DuckLakeLocalLayout {
            data_path: root.to_string(),
            staging_path: format!("{root}/__tre_duckdb_stage"),
            tmp_path: format!("{root}/__tre_duckdb_stage"),
            test_runs_path: format!("{root}/test-runs"),
        })
    }

    fn build_attach_plan_with_credential_and_root(
        &self,
        profile: &DataStoreProfile,
        credential: DuckLakeCatalogCredential,
        options: DuckLakeOpenOptions,
        root_certificate: Option<&str>,
    ) -> Result<DuckLakeAttachPlan, LakeError> {
        credential.validate()?;

        let layout = self.local_layout()?;
        let data_path = normalized_root(&layout.data_path)?.to_string();
        let pre_attach_sql = ducklake_pre_attach_sql(profile, &credential, false);
        let conninfo = ducklake_postgres_conninfo(profile, &credential, root_certificate, false);
        let conninfo_description =
            ducklake_postgres_conninfo(profile, &credential, root_certificate, true);
        let mut attach_options = self.attach_options(profile, &data_path, options)?;
        let DuckLakeCatalogCredential::SessionSecret { secret_name, .. } = &credential;
        attach_options.push(format!(
            "META_SECRET {}",
            sql_literal(&session_secret_name(secret_name))
        ));

        let attach_sql = format!(
            "ATTACH {} AS {LAKE_ALIAS} ({})",
            sql_literal(&conninfo),
            attach_options.join(", ")
        );
        let attach_description = format!(
            "ATTACH {} AS {LAKE_ALIAS} ({})",
            sql_literal(&conninfo_description),
            attach_options.join(", ")
        );

        Ok(DuckLakeAttachPlan {
            metadata_schema: profile
                .lake
                .catalog_schema
                .clone()
                .unwrap_or_else(|| "main".into()),
            alias: LAKE_ALIAS.to_string(),
            attach_description,
            pre_attach_sql,
            attach_sql,
            data_path,
            credential_kind: credential.kind(),
            create_if_not_exists: options.create_if_not_exists,
            automatic_migration: options.automatic_migration,
            configured_encryption_mode: profile.lake.auth.encryption_mode,
        })
    }

    fn open_attached_catalog_with_credential_and_root(
        &self,
        profile: &DataStoreProfile,
        credential: DuckLakeCatalogCredential,
        options: DuckLakeOpenOptions,
        root_certificate: Option<&str>,
        temp_directory: Option<&std::path::Path>,
        observations: &mut LakeCatalogObservations,
    ) -> Result<DuckLakeOpenedCatalog, LakeError> {
        let local_layout = self.prepare_local_layout(options)?;
        let (connection, attach_plan) = self.attach_catalog(
            profile,
            credential,
            options,
            root_certificate,
            temp_directory,
            observations,
        )?;
        #[cfg(feature = "acceptance-catalog-barrier")]
        crate::catalog_observations::after_attach_barrier()?;
        let health = observations.health_check(self, &connection)?;
        if !options.create_if_not_exists
            && normalize_ducklake_data_path(&health.data_path)
                != normalize_ducklake_data_path(&local_layout.data_path)
        {
            return Err(LakeError::PersistedDataPathMismatch);
        }

        Ok(DuckLakeOpenedCatalog {
            connection,
            attach_plan,
            local_layout,
            health,
            tls_guard: None,
        })
    }

    /// Observe an explicitly authorized health read on the retained connection.
    pub fn health_check_observed(
        &self,
        connection: &duckdb::Connection,
        observations: &mut LakeCatalogObservations,
    ) -> Result<DuckLakeHealth, LakeError> {
        observations.health_check(self, connection)
    }

    /// Reads DuckLake settings and lightweight catalog counts.
    ///
    /// # Errors
    ///
    /// Returns an error if DuckDB metadata queries fail or DuckLake reports an
    /// unexpected encryption setting.
    pub fn health_check(
        &self,
        connection: &duckdb::Connection,
    ) -> Result<DuckLakeHealth, LakeError> {
        let settings = ducklake_settings(connection)?;
        let detected_encryption_mode = detected_encryption_mode(connection, &settings.data_path)?;
        let snapshot_count = connection.query_row(
            &format!(
                "SELECT count(*) FROM ducklake_snapshots({})",
                sql_literal(LAKE_ALIAS)
            ),
            [],
            |row| row.get(0),
        )?;
        let table_count = connection.query_row(
            &format!(
                "SELECT count(*) FROM ducklake_table_info({})",
                sql_literal(LAKE_ALIAS)
            ),
            [],
            |row| row.get(0),
        )?;

        Ok(DuckLakeHealth {
            alias: LAKE_ALIAS.to_string(),
            catalog_type: settings.catalog_type,
            extension_version: settings.extension_version,
            data_path: settings.data_path,
            detected_encryption_mode,
            snapshot_count,
            table_count,
        })
    }
}

fn object_namespace_glob(location: &ObjectStorageLocation) -> String {
    format!("{}**", location.canonical_data_path())
}

fn reject_ambient_storage_authority(
    names: impl IntoIterator<Item = impl AsRef<str>>,
    authentication: &ObjectStorageAuthentication<'_>,
) -> Result<(), LakeError> {
    let names = names
        .into_iter()
        .map(|name| name.as_ref().to_string())
        .collect::<Vec<_>>();
    if let Some(name) = legacy_object_storage_environment_variable(|candidate| {
        names.iter().any(|name| name == candidate)
    }) {
        return Err(LakeError::AmbientStorageAuthority {
            name: name.to_string(),
        });
    }
    match authentication {
        ObjectStorageAuthentication::AwsWorkloadIdentity {
            source: AwsWorkloadIdentitySource::EcsTask,
        } if !names
            .iter()
            .any(|name| name == "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI") =>
        {
            Err(LakeError::MissingWorkloadIdentitySource { kind: "ecs_task" })
        }
        ObjectStorageAuthentication::AwsWorkloadIdentity {
            source: AwsWorkloadIdentitySource::Ec2Instance,
        } if names
            .iter()
            .any(|name| name == "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI") =>
        {
            Err(LakeError::AmbientStorageAuthority {
                name: "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI".into(),
            })
        }
        ObjectStorageAuthentication::S3Static { .. }
        | ObjectStorageAuthentication::AzureManagedIdentity { .. }
        | ObjectStorageAuthentication::AzureServicePrincipal { .. }
            if names
                .iter()
                .any(|name| name == "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI") =>
        {
            Err(LakeError::AmbientStorageAuthority {
                name: "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI".into(),
            })
        }
        _ => Ok(()),
    }
}

fn validate_storage_authentication(
    location: &ObjectStorageLocation,
    authentication: &ObjectStorageAuthentication<'_>,
) -> Result<(), LakeError> {
    let compatible = matches!(
        (location, authentication),
        (
            ObjectStorageLocation::AwsS3 { .. },
            ObjectStorageAuthentication::S3Static { .. }
                | ObjectStorageAuthentication::AwsWorkloadIdentity { .. }
        ) | (
            ObjectStorageLocation::S3Compatible { .. },
            ObjectStorageAuthentication::S3Static { .. }
        ) | (
            ObjectStorageLocation::AzureBlob { .. },
            ObjectStorageAuthentication::AzureManagedIdentity { .. }
                | ObjectStorageAuthentication::AzureServicePrincipal { .. }
        )
    );
    if compatible {
        Ok(())
    } else {
        Err(LakeError::IncompatibleStorageAuthentication)
    }
}

fn storage_secret_sql(
    location: &ObjectStorageLocation,
    authentication: ObjectStorageAuthentication<'_>,
) -> Result<String, LakeError> {
    let scope = location.canonical_data_path();
    let fields = match authentication {
        ObjectStorageAuthentication::S3Static {
            access_key_id,
            secret_access_key,
        } => access_key_id.expose(|access_bytes| {
            secret_access_key.expose(|secret_bytes| -> Result<Vec<String>, LakeError> {
                let access = std::str::from_utf8(access_bytes)
                    .map_err(|_| LakeError::InvalidCredentialEncoding)?;
                let secret = std::str::from_utf8(secret_bytes)
                    .map_err(|_| LakeError::InvalidCredentialEncoding)?;
                let mut fields = vec![
                    "TYPE s3".to_string(),
                    "PROVIDER config".to_string(),
                    format!("KEY_ID {}", sql_literal(access)),
                    format!("SECRET {}", sql_literal(secret)),
                ];
                match location {
                    ObjectStorageLocation::AwsS3 { region, .. } => {
                        fields.push(format!("REGION {}", sql_literal(region)));
                    }
                    ObjectStorageLocation::S3Compatible {
                        endpoint,
                        url_style,
                        ..
                    } => {
                        let endpoint = endpoint.strip_prefix("https://").unwrap_or(endpoint);
                        fields.push(format!("ENDPOINT {}", sql_literal(endpoint)));
                        fields.push(format!(
                            "URL_STYLE {}",
                            sql_literal(match url_style {
                                S3UrlStyle::Path => "path",
                                S3UrlStyle::VirtualHosted => "vhost",
                            })
                        ));
                        fields.push("USE_SSL true".to_string());
                        fields.push("VERIFY_SSL true".to_string());
                    }
                    ObjectStorageLocation::AzureBlob { .. } => {
                        return Err(LakeError::IncompatibleStorageAuthentication);
                    }
                }
                Ok(fields)
            })
        })?,
        ObjectStorageAuthentication::AwsWorkloadIdentity {
            source: AwsWorkloadIdentitySource::Ec2Instance,
        } => {
            let ObjectStorageLocation::AwsS3 { region, .. } = location else {
                return Err(LakeError::IncompatibleStorageAuthentication);
            };
            aws_session_credential_fields(resolve_ec2_instance_credentials()?, region)?
        }
        ObjectStorageAuthentication::AwsWorkloadIdentity {
            source: AwsWorkloadIdentitySource::EcsTask,
        } => {
            let ObjectStorageLocation::AwsS3 { region, .. } = location else {
                return Err(LakeError::IncompatibleStorageAuthentication);
            };
            aws_session_credential_fields(resolve_ecs_task_credentials()?, region)?
        }
        ObjectStorageAuthentication::AzureManagedIdentity { client_id } => {
            let ObjectStorageLocation::AzureBlob {
                account_endpoint, ..
            } = location
            else {
                return Err(LakeError::IncompatibleStorageAuthentication);
            };
            let account_name = azure_account_name(account_endpoint)?;
            let mut fields = vec![
                "TYPE azure".to_string(),
                "PROVIDER managed_identity".to_string(),
                format!("ACCOUNT_NAME {}", sql_literal(account_name)),
            ];
            if let Some(client_id) = client_id {
                fields.push(format!("CLIENT_ID {}", sql_literal(&client_id.to_string())));
            }
            fields
        }
        ObjectStorageAuthentication::AzureServicePrincipal {
            tenant_id,
            client_id,
            client_secret,
        } => {
            let ObjectStorageLocation::AzureBlob {
                account_endpoint, ..
            } = location
            else {
                return Err(LakeError::IncompatibleStorageAuthentication);
            };
            let account_name = azure_account_name(account_endpoint)?;
            client_secret.expose(|bytes| -> Result<Vec<String>, LakeError> {
                let secret =
                    std::str::from_utf8(bytes).map_err(|_| LakeError::InvalidCredentialEncoding)?;
                Ok(vec![
                    "TYPE azure".to_string(),
                    "PROVIDER service_principal".to_string(),
                    format!("TENANT_ID {}", sql_literal(&tenant_id.to_string())),
                    format!("CLIENT_ID {}", sql_literal(&client_id.to_string())),
                    format!("CLIENT_SECRET {}", sql_literal(secret)),
                    format!("ACCOUNT_NAME {}", sql_literal(account_name)),
                ])
            })?
        }
    };
    let mut fields = fields;
    fields.push(format!("SCOPE {}", sql_literal(&scope)));
    Ok(format!(
        "CREATE OR REPLACE SECRET ahri_tre_lake_storage ({})",
        fields.join(", ")
    ))
}

struct AwsSessionCredentials {
    access_key_id: SecretMaterial,
    secret_access_key: SecretMaterial,
    session_token: SecretMaterial,
}

#[derive(serde::Deserialize)]
#[serde(rename_all = "PascalCase")]
struct AwsCredentialResponse {
    access_key_id: String,
    secret_access_key: String,
    token: String,
}

fn aws_session_credential_fields(
    credentials: AwsSessionCredentials,
    region: &str,
) -> Result<Vec<String>, LakeError> {
    credentials.access_key_id.expose(|access_bytes| {
        credentials.secret_access_key.expose(|secret_bytes| {
            credentials.session_token.expose(|token_bytes| {
                let access = std::str::from_utf8(access_bytes)
                    .map_err(|_| LakeError::WorkloadIdentityUnavailable)?;
                let secret = std::str::from_utf8(secret_bytes)
                    .map_err(|_| LakeError::WorkloadIdentityUnavailable)?;
                let token = std::str::from_utf8(token_bytes)
                    .map_err(|_| LakeError::WorkloadIdentityUnavailable)?;
                Ok(vec![
                    "TYPE s3".to_string(),
                    "PROVIDER config".to_string(),
                    format!("KEY_ID {}", sql_literal(access)),
                    format!("SECRET {}", sql_literal(secret)),
                    format!("SESSION_TOKEN {}", sql_literal(token)),
                    format!("REGION {}", sql_literal(region)),
                ])
            })
        })
    })
}

fn workload_identity_client() -> Result<reqwest::blocking::Client, LakeError> {
    reqwest::blocking::Client::builder()
        .no_proxy()
        .redirect(reqwest::redirect::Policy::none())
        .timeout(std::time::Duration::from_secs(2))
        .build()
        .map_err(|_| LakeError::WorkloadIdentityUnavailable)
}

fn read_bounded_workload_response(
    response: reqwest::blocking::Response,
    limit: usize,
) -> Result<Vec<u8>, LakeError> {
    if !response.status().is_success() {
        return Err(LakeError::WorkloadIdentityUnavailable);
    }
    let mut encoded = Vec::new();
    response
        .take((limit + 1) as u64)
        .read_to_end(&mut encoded)
        .map_err(|_| LakeError::WorkloadIdentityUnavailable)?;
    if encoded.len() > limit {
        return Err(LakeError::WorkloadIdentityUnavailable);
    }
    Ok(encoded)
}

fn resolve_ec2_instance_credentials() -> Result<AwsSessionCredentials, LakeError> {
    const IMDS: &str = "http://169.254.169.254";
    const TOKEN_HEADER: &str = "x-aws-ec2-metadata-token";
    let client = workload_identity_client()?;
    let token = read_bounded_workload_response(
        client
            .put(format!("{IMDS}/latest/api/token"))
            .header("x-aws-ec2-metadata-token-ttl-seconds", "60")
            .send()
            .map_err(|_| LakeError::WorkloadIdentityUnavailable)?,
        4096,
    )?;
    if token.is_empty() {
        return Err(LakeError::WorkloadIdentityUnavailable);
    }
    let token = reqwest::header::HeaderValue::from_bytes(&token)
        .map_err(|_| LakeError::WorkloadIdentityUnavailable)?;
    let role = read_bounded_workload_response(
        client
            .get(format!("{IMDS}/latest/meta-data/iam/security-credentials/"))
            .header(TOKEN_HEADER, token.clone())
            .send()
            .map_err(|_| LakeError::WorkloadIdentityUnavailable)?,
        128,
    )?;
    let role = validated_imds_role_name(&role)?;
    let encoded = read_bounded_workload_response(
        client
            .get(format!(
                "{IMDS}/latest/meta-data/iam/security-credentials/{role}"
            ))
            .header(TOKEN_HEADER, token)
            .send()
            .map_err(|_| LakeError::WorkloadIdentityUnavailable)?,
        MAX_SECRET_MATERIAL_BYTES,
    )?;
    aws_credentials_from_response(&encoded)
}

fn validated_imds_role_name(encoded: &[u8]) -> Result<&str, LakeError> {
    let role = std::str::from_utf8(encoded)
        .map_err(|_| LakeError::WorkloadIdentityUnavailable)?
        .trim_end_matches(['\r', '\n']);
    if role.is_empty()
        || role.len() > 64
        || !role.bytes().all(|byte| {
            byte.is_ascii_alphanumeric()
                || matches!(byte, b'_' | b'+' | b'=' | b',' | b'.' | b'@' | b'-')
        })
    {
        return Err(LakeError::WorkloadIdentityUnavailable);
    }
    Ok(role)
}

fn resolve_ecs_task_credentials() -> Result<AwsSessionCredentials, LakeError> {
    let relative = std::env::var("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI")
        .map_err(|_| LakeError::MissingWorkloadIdentitySource { kind: "ecs_task" })?;
    if relative.len() <= "/v2/credentials/".len()
        || relative.len() > 2048
        || !relative.starts_with("/v2/credentials/")
        || relative.contains("//")
        || relative.contains('\n')
        || relative.contains('\r')
        || relative.contains('\\')
        || relative.contains('?')
        || relative.contains('#')
        || relative
            .split('/')
            .any(|component| component == "." || component == "..")
    {
        return Err(LakeError::WorkloadIdentityUnavailable);
    }
    let response = workload_identity_client()?
        .get(format!("http://169.254.170.2{relative}"))
        .send()
        .map_err(|_| LakeError::WorkloadIdentityUnavailable)?;
    let encoded = read_bounded_workload_response(response, MAX_SECRET_MATERIAL_BYTES)?;
    aws_credentials_from_response(&encoded)
}

fn aws_credentials_from_response(encoded: &[u8]) -> Result<AwsSessionCredentials, LakeError> {
    let response: AwsCredentialResponse =
        serde_json::from_slice(encoded).map_err(|_| LakeError::WorkloadIdentityUnavailable)?;
    if response.access_key_id.is_empty()
        || response.secret_access_key.is_empty()
        || response.token.is_empty()
    {
        return Err(LakeError::WorkloadIdentityUnavailable);
    }
    Ok(AwsSessionCredentials {
        access_key_id: SecretMaterial::try_from(response.access_key_id.into_bytes())
            .map_err(|_| LakeError::WorkloadIdentityUnavailable)?,
        secret_access_key: SecretMaterial::try_from(response.secret_access_key.into_bytes())
            .map_err(|_| LakeError::WorkloadIdentityUnavailable)?,
        session_token: SecretMaterial::try_from(response.token.into_bytes())
            .map_err(|_| LakeError::WorkloadIdentityUnavailable)?,
    })
}

fn azure_account_name(endpoint: &str) -> Result<&str, LakeError> {
    endpoint
        .strip_prefix("https://")
        .and_then(|authority| authority.split('.').next())
        .filter(|value| !value.is_empty())
        .ok_or(LakeError::InvalidObjectStorageIdentity)
}

fn canonical_datastore_prefix(prefix: &str) -> bool {
    !prefix.is_empty()
        && !prefix.starts_with('/')
        && !prefix.ends_with('/')
        && !prefix.contains("//")
        && !prefix.contains('\\')
        && prefix
            .split('/')
            .all(|component| !component.is_empty() && component != "." && component != "..")
        && !prefix.chars().any(char::is_control)
}

impl DuckLakeAdapter {
    fn prepare_local_layout(
        &self,
        options: DuckLakeOpenOptions,
    ) -> Result<DuckLakeLocalLayout, LakeError> {
        let layout = self.local_layout()?;
        if options.create_if_not_exists {
            fs::create_dir_all(&layout.data_path)?;
            fs::create_dir_all(&layout.staging_path)?;
            fs::create_dir_all(&layout.tmp_path)?;
            fs::create_dir_all(&layout.test_runs_path)?;
        } else {
            fs::metadata(&layout.data_path)?;
        }
        Ok(layout)
    }

    fn attach_catalog(
        &self,
        profile: &DataStoreProfile,
        credential: DuckLakeCatalogCredential,
        options: DuckLakeOpenOptions,
        root_certificate: Option<&str>,
        temp_directory: Option<&std::path::Path>,
        observations: &mut LakeCatalogObservations,
    ) -> Result<(duckdb::Connection, DuckLakeAttachPlan), LakeError> {
        let attach_plan = self.build_attach_plan_with_credential_and_root(
            profile,
            credential,
            options,
            root_certificate,
        )?;
        let conn = duckdb::Connection::open_in_memory()?;
        if let Some(temp_directory) = temp_directory {
            conn.execute_batch(&format!(
                "SET temp_directory = {}",
                sql_literal(&temp_directory.to_string_lossy())
            ))?;
        }
        conn.execute_batch("INSTALL postgres;\nLOAD postgres;\nINSTALL ducklake;\nLOAD ducklake;")?;
        for statement in &attach_plan.pre_attach_sql {
            conn.execute_batch(statement)
                .map_err(|_| LakeError::CredentialProjectionFailed)?;
        }
        observations.attach(&conn, &attach_plan.attach_sql, &profile.lake.auth.lake_user)?;
        conn.execute_batch(&format!("USE {LAKE_ALIAS};"))?;
        Ok((conn, attach_plan))
    }

    fn attach_options(
        &self,
        profile: &DataStoreProfile,
        data_path: &str,
        options: DuckLakeOpenOptions,
    ) -> Result<Vec<String>, LakeError> {
        let mut attach_options = vec![
            format!(
                "CREATE_IF_NOT_EXISTS {}",
                sql_bool(options.create_if_not_exists)
            ),
            format!(
                "AUTOMATIC_MIGRATION {}",
                sql_bool(options.automatic_migration)
            ),
        ];
        if options.create_if_not_exists {
            attach_options.insert(0, format!("DATA_PATH {}", sql_literal(data_path)));
        }

        if let Some(metadata_schema) = profile.lake.catalog_schema.as_deref() {
            attach_options.push(format!("METADATA_SCHEMA {}", sql_literal(metadata_schema)));
        }

        if options.apply_configured_encryption {
            match profile.lake.auth.encryption_mode {
                EncryptionMode::None => {}
                EncryptionMode::ServerManaged => attach_options.push("ENCRYPTED".to_string()),
                EncryptionMode::ClientManaged => {
                    return Err(LakeError::UnsupportedEncryptionMode(
                        EncryptionMode::ClientManaged,
                    ));
                }
            }
        }

        Ok(attach_options)
    }
}

struct DuckLakeSettings {
    catalog_type: String,
    extension_version: String,
    data_path: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct DuckLakeMetadataRelation {
    catalog: String,
    schema: String,
}

impl DuckLakeMetadataRelation {
    fn table_sql(&self, table_name: &str) -> String {
        format!(
            "{}.{}.{}",
            quote_identifier(&self.catalog),
            quote_identifier(&self.schema),
            quote_identifier(table_name)
        )
    }
}

fn ducklake_settings(connection: &duckdb::Connection) -> Result<DuckLakeSettings, LakeError> {
    connection
        .query_row(
            &format!(
                "SELECT catalog_type, extension_version, data_path FROM ducklake_settings({})",
                sql_literal(LAKE_ALIAS)
            ),
            [],
            |row| {
                Ok(DuckLakeSettings {
                    catalog_type: row.get(0)?,
                    extension_version: row.get(1)?,
                    data_path: row.get(2)?,
                })
            },
        )
        .map_err(LakeError::from)
}

fn detected_encryption_mode(
    connection: &duckdb::Connection,
    data_path: &str,
) -> Result<EncryptionMode, LakeError> {
    if let Some(relation) = ducklake_metadata_relation(connection, data_path)?
        && let Some(value) = connection
            .query_row(
                &format!(
                    "SELECT value FROM {} WHERE lower(key) = 'encrypted' LIMIT 1",
                    relation.table_sql("ducklake_metadata")
                ),
                [],
                |row| row.get::<_, String>(0),
            )
            .optional()?
    {
        return parse_encryption_metadata_value("ducklake_metadata.encrypted", value);
    }

    let value = connection
        .query_row(
            &format!(
                "SELECT value FROM {LAKE_ALIAS}.options() WHERE lower(option_name) = 'encrypted' LIMIT 1"
            ),
            [],
            |row| row.get::<_, String>(0),
        )
        .optional()?
        .unwrap_or_else(|| "false".to_string());

    parse_encryption_metadata_value("encrypted", value)
}

fn ducklake_metadata_relation(
    connection: &duckdb::Connection,
    data_path: &str,
) -> Result<Option<DuckLakeMetadataRelation>, LakeError> {
    let metadata_catalog = format!("__ducklake_metadata_{LAKE_ALIAS}");
    let expected_data_path = normalize_ducklake_data_path(data_path);
    let mut statement = connection.prepare(
        "SELECT table_schema
         FROM information_schema.tables
         WHERE table_catalog = ?
           AND table_name = 'ducklake_metadata'
         ORDER BY table_schema",
    )?;
    let schemas = statement
        .query_map([metadata_catalog.as_str()], |row| row.get::<_, String>(0))?
        .collect::<Result<Vec<_>, _>>()?;

    for schema in schemas {
        let metadata_data_path = connection
            .query_row(
                &format!(
                    "SELECT value FROM {}.{}.{} WHERE lower(key) = 'data_path' LIMIT 1",
                    quote_identifier(&metadata_catalog),
                    quote_identifier(&schema),
                    quote_identifier("ducklake_metadata")
                ),
                [],
                |row| row.get::<_, String>(0),
            )
            .optional()?;

        if metadata_data_path
            .as_deref()
            .map(normalize_ducklake_data_path)
            .as_deref()
            == Some(expected_data_path.as_str())
        {
            return Ok(Some(DuckLakeMetadataRelation {
                catalog: metadata_catalog,
                schema,
            }));
        }
    }

    Ok(None)
}

fn parse_encryption_metadata_value(
    key: &'static str,
    value: String,
) -> Result<EncryptionMode, LakeError> {
    match value.trim().to_ascii_lowercase().as_str() {
        "false" => Ok(EncryptionMode::None),
        "true" => Ok(EncryptionMode::ServerManaged),
        _ => Err(LakeError::InvalidMetadataValue { key, value }),
    }
}

fn normalize_ducklake_data_path(value: &str) -> String {
    value.trim().trim_end_matches('/').to_string()
}

fn normalized_root(lake_root: &str) -> Result<&str, LakeError> {
    let root = lake_root.trim().trim_end_matches('/');
    if root.is_empty() {
        return Err(LakeError::EmptyLakeRoot);
    }
    Ok(root)
}

fn ducklake_postgres_conninfo(
    profile: &DataStoreProfile,
    credential: &DuckLakeCatalogCredential,
    root_certificate: Option<&str>,
    redact_secret: bool,
) -> String {
    let mut base_params = vec![
        libpq_param("host", &profile.server),
        libpq_param("port", &profile.port.to_string()),
        libpq_param("dbname", &profile.lake.lake_db),
        libpq_param("user", &profile.lake.auth.lake_user),
    ];
    if let Some(sslmode) = profile.sslmode.as_deref().filter(|value| !value.is_empty()) {
        base_params.push(libpq_param("sslmode", sslmode));
    }
    if let Some(root_certificate) = root_certificate {
        base_params.push(libpq_param(
            "sslrootcert",
            if redact_secret && root_certificate != "system" {
                "<adapter-projected-ca>"
            } else {
                root_certificate
            },
        ));
    }

    let params = match credential {
        DuckLakeCatalogCredential::SessionSecret { .. } => base_params,
    };

    format!("ducklake:postgres:{}", params.join(" "))
}

fn ducklake_pre_attach_sql(
    profile: &DataStoreProfile,
    credential: &DuckLakeCatalogCredential,
    redact_secret: bool,
) -> Vec<String> {
    match credential {
        DuckLakeCatalogCredential::SessionSecret {
            secret_name,
            password,
        } => {
            let password = if redact_secret {
                "<redacted>"
            } else {
                password.as_str()
            };
            vec![format!(
                "CREATE OR REPLACE SECRET {} (TYPE postgres, HOST {}, PORT {}, DATABASE {}, USER {}, PASSWORD {})",
                quote_identifier(&session_secret_name(secret_name)),
                sql_literal(&profile.server),
                profile.port,
                sql_literal(&profile.lake.lake_db),
                sql_literal(&profile.lake.auth.lake_user),
                sql_literal(password),
            )]
        }
    }
}

fn session_secret_name(value: &str) -> String {
    value
        .trim()
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || ch == '_' {
                ch
            } else {
                '_'
            }
        })
        .collect()
}

fn libpq_param(key: &str, value: &str) -> String {
    format!(
        "{}='{}'",
        key,
        value.replace('\\', "\\\\").replace('\'', "\\'")
    )
}

fn sql_literal(value: &str) -> String {
    format!("'{}'", value.replace('\'', "''"))
}

fn quote_identifier(value: &str) -> String {
    format!("\"{}\"", value.replace('"', "\"\""))
}

fn sql_bool(value: bool) -> &'static str {
    if value { "true" } else { "false" }
}

fn finish_lake_open(
    span: Option<ahri_tre_observability::OperationSpan>,
    result: &Result<DuckLakeOpenedCatalog, LakeError>,
) {
    use ahri_tre_observability::{FailureCategory, Outcome};
    if let Some(span) = span {
        span.finish_observed(
            if result.is_ok() {
                Outcome::Success
            } else {
                Outcome::Unavailable
            },
            result.as_ref().err().map(|_| FailureCategory::Adapter),
        );
    }
}

#[cfg(test)]
mod tests {
    use std::{
        fs,
        os::unix::fs::PermissionsExt,
        path::PathBuf,
        time::{SystemTime, UNIX_EPOCH},
    };

    use ahri_tre_secrets::SecretMaterial;
    use ahri_tre_types::{DataStoreProfile, EncryptionMode, LakeAttachProfile, LakeAuthProfile};

    use super::{
        DuckLakeAdapter, DuckLakeCatalogCredential, DuckLakeCredentialKind, DuckLakeOpenOptions,
        FilesystemCatalogTls, FilesystemLakeSessionConfig, ObjectStorageLocation,
        aws_credentials_from_response, object_namespace_glob, validated_imds_role_name,
    };
    use crate::{LAKE_ALIAS, LakeError, ScratchAttemptId, TrustedScratch};

    #[test]
    fn local_layout_expands_canonical_subdirectories() {
        let adapter = DuckLakeAdapter::new("/lake/root/");
        let layout = adapter.local_layout().expect("layout should build");

        assert_eq!(layout.data_path, "/lake/root");
        assert_eq!(layout.staging_path, "/lake/root/__tre_duckdb_stage");
        assert_eq!(layout.tmp_path, "/lake/root/__tre_duckdb_stage");
        assert_eq!(layout.test_runs_path, "/lake/root/test-runs");
    }

    #[test]
    fn local_layout_rejects_empty_root() {
        let adapter = DuckLakeAdapter::new("   ");
        assert!(matches!(
            adapter.local_layout(),
            Err(LakeError::EmptyLakeRoot)
        ));
    }

    #[test]
    fn object_namespace_glob_preserves_the_exact_canonical_prefix() {
        let location = ObjectStorageLocation::aws_s3_for_datastore(
            "af-south-1",
            "research",
            Some("/tenant/"),
            "datastore",
        )
        .unwrap();

        assert_eq!(
            location.canonical_data_path(),
            "s3://research/tenant/datastore/"
        );
        assert_eq!(
            object_namespace_glob(&location),
            "s3://research/tenant/datastore/**"
        );
    }

    #[test]
    fn ecs_task_response_becomes_bounded_protected_session_credentials() {
        let credentials = aws_credentials_from_response(
            br#"{
                "AccessKeyId":"ecs-access-canary",
                "SecretAccessKey":"ecs-secret-canary",
                "Token":"ecs-token-canary",
                "Expiration":"2030-01-01T00:00:00Z"
            }"#,
        )
        .unwrap();

        credentials
            .access_key_id
            .expose(|bytes| assert_eq!(bytes, b"ecs-access-canary"));
        credentials
            .secret_access_key
            .expose(|bytes| assert_eq!(bytes, b"ecs-secret-canary"));
        credentials
            .session_token
            .expose(|bytes| assert_eq!(bytes, b"ecs-token-canary"));
    }

    #[test]
    fn ec2_role_name_is_a_single_closed_path_component() {
        assert_eq!(
            validated_imds_role_name(b"research-node-role\n").unwrap(),
            "research-node-role"
        );
        for invalid in [
            b"".as_slice(),
            b"../role".as_slice(),
            b"role/other".as_slice(),
            b"role?query".as_slice(),
            b"role\nother".as_slice(),
        ] {
            assert!(matches!(
                validated_imds_role_name(invalid),
                Err(LakeError::WorkloadIdentityUnavailable)
            ));
        }
    }

    #[test]
    fn explicit_filesystem_inputs_keep_path_joining_and_credentials_in_adapter() {
        let config = FilesystemLakeSessionConfig::new(
            "/srv/ahri-tre/lake",
            "research/data",
            "postgres.example.test",
            5432,
            "ducklake_catalog",
            "catalog",
            "catalog_owner",
            FilesystemCatalogTls::System,
            EncryptionMode::ServerManaged,
        )
        .unwrap();
        let credential = SecretMaterial::try_from(b"catalog-secret".to_vec()).unwrap();

        let plan = DuckLakeAdapter::build_filesystem_attach_plan(&config, &credential).unwrap();

        assert_eq!(config.filesystem_base(), "/srv/ahri-tre/lake");
        assert_eq!(config.datastore_prefix(), "research/data");
        assert_eq!(plan.data_path, "/srv/ahri-tre/lake/research/data");
        assert!(plan.attach_sql.contains("sslrootcert"));
        assert!(!plan.attach_description.contains("catalog-secret"));
        assert!(!format!("{plan:?}").contains("catalog-secret"));
    }

    #[test]
    fn datastore_owned_path_is_validated_and_relativized_inside_the_adapter() {
        let config = FilesystemLakeSessionConfig::from_binding(
            "/srv/ahri-tre/lake",
            "/srv/ahri-tre/lake/datastores/abc",
            "postgres.example.test",
            5432,
            "ducklake_catalog",
            "catalog",
            "catalog_owner",
            FilesystemCatalogTls::System,
            EncryptionMode::ServerManaged,
        )
        .unwrap();

        assert_eq!(config.datastore_prefix(), "datastores/abc");
        assert_eq!(
            config.canonical_data_path(),
            "/srv/ahri-tre/lake/datastores/abc"
        );
    }

    #[test]
    fn filesystem_initialization_rejects_an_existing_namespace_before_opening_duckdb() {
        let base = temp_lake_root("occupied_namespace");
        let data_path = base.join("datastores/occupied");
        fs::create_dir_all(&data_path).unwrap();
        let scratch_root = temp_lake_root("occupied_namespace_scratch");
        fs::set_permissions(&scratch_root, fs::Permissions::from_mode(0o700)).unwrap();
        let scratch = TrustedScratch::open(&scratch_root, [data_path.as_path()])
            .unwrap()
            .create_attempt(ScratchAttemptId::new("0123456789abcdef0123456789abcdef").unwrap())
            .unwrap();
        let config = FilesystemLakeSessionConfig::new(
            base.display().to_string(),
            "datastores/occupied",
            "postgres.example.test",
            5432,
            "ducklake_catalog",
            "catalog",
            "catalog_owner",
            FilesystemCatalogTls::System,
            EncryptionMode::None,
        )
        .unwrap();
        let credential = SecretMaterial::try_from(b"catalog-secret".to_vec()).unwrap();

        let error = DuckLakeAdapter::initialize_filesystem_catalog_with_scratch(
            &config,
            &credential,
            &scratch,
        )
        .expect_err("an existing namespace must never be adopted or replaced");

        assert!(matches!(error, LakeError::NamespaceNotEmpty));
        drop(scratch);
        fs::remove_dir_all(base).unwrap();
        fs::remove_dir_all(scratch_root).unwrap();
    }

    #[test]
    fn attach_existing_layout_does_not_create_runtime_subdirectories() {
        let root = temp_lake_root("attach_existing_layout");
        let adapter = DuckLakeAdapter::new(root.display().to_string());

        let layout = adapter
            .prepare_local_layout(DuckLakeOpenOptions::attach_existing())
            .expect("existing data root should be accepted without helper dirs");

        assert_eq!(layout.data_path, root.display().to_string());
        assert!(!root.join("__tre_duckdb_stage").exists());
        assert!(!root.join("test-runs").exists());
        fs::remove_dir_all(root).expect("temp lake root should be removed");
    }

    #[test]
    fn initialize_layout_creates_runtime_subdirectories() {
        let root = temp_lake_root("initialize_layout");
        let adapter = DuckLakeAdapter::new(root.display().to_string());

        adapter
            .prepare_local_layout(DuckLakeOpenOptions::initialize())
            .expect("initialization should prepare helper dirs");

        assert!(root.join("__tre_duckdb_stage").is_dir());
        assert!(root.join("test-runs").is_dir());
        fs::remove_dir_all(root).expect("temp lake root should be removed");
    }

    fn temp_lake_root(name: &str) -> PathBuf {
        let suffix = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system time should be after epoch")
            .as_nanos();
        let root = std::env::temp_dir().join(format!(
            "ahri_tre_lake_{name}_{}_{}",
            std::process::id(),
            suffix
        ));
        fs::create_dir_all(&root).expect("temp lake root should be created");
        root
    }

    #[test]
    fn attach_plan_uses_canonical_data_path_and_migration_options() {
        let adapter = DuckLakeAdapter::new("/lake/root");
        let plan = adapter
            .build_attach_plan_with_credential_and_root(
                &test_profile(EncryptionMode::None),
                DuckLakeCatalogCredential::session_secret("tre_catalog", "catalog-secret"),
                DuckLakeOpenOptions::attach_existing().with_automatic_migration(true),
                None,
            )
            .expect("attach plan should build");

        assert_eq!(plan.alias, LAKE_ALIAS);
        assert_eq!(plan.credential_kind, DuckLakeCredentialKind::SessionSecret);
        assert_eq!(plan.data_path, "/lake/root");
        assert!(!plan.create_if_not_exists);
        assert!(plan.automatic_migration);
        assert_eq!(plan.attach_sql, plan.attach_description);
        assert!(
            plan.attach_description
                .contains("ATTACH 'ducklake:postgres:host=")
        );
        assert!(plan.attach_sql.contains("META_SECRET 'tre_catalog'"));
        assert!(!plan.attach_description.contains("catalog-secret"));
        assert!(plan.attach_description.contains("sslmode=''disable''"));
        assert!(!plan.attach_description.contains("DATA_PATH"));
        assert!(
            plan.attach_description
                .contains("CREATE_IF_NOT_EXISTS false")
        );
        assert!(plan.attach_description.contains("AUTOMATIC_MIGRATION true"));
        assert!(
            plan.attach_description
                .contains("METADATA_SCHEMA 'lake_catalog'")
        );
    }

    #[test]
    fn attach_plan_supports_session_secret_and_redacts_setup_sql() {
        let adapter = DuckLakeAdapter::new("/lake/root");
        let plan = adapter
            .build_attach_plan_with_credential_and_root(
                &test_profile(EncryptionMode::None),
                DuckLakeCatalogCredential::session_secret("tre lake secret", "super-secret"),
                DuckLakeOpenOptions::attach_existing(),
                None,
            )
            .expect("attach plan should build");

        assert_eq!(plan.credential_kind, DuckLakeCredentialKind::SessionSecret);
        assert_eq!(plan.pre_attach_sql.len(), 1);
        assert!(plan.pre_attach_sql[0].contains("CREATE OR REPLACE SECRET \"tre_lake_secret\""));
        assert!(plan.pre_attach_sql[0].contains("PASSWORD 'super-secret'"));
        assert!(
            plan.attach_sql
                .contains("'ducklake:postgres:host=''tre-postgres''")
        );
        assert!(plan.attach_sql.contains("dbname=''tre_lake''"));
        assert!(plan.attach_sql.contains("META_SECRET 'tre_lake_secret'"));
        assert!(
            plan.attach_description
                .contains("'ducklake:postgres:host=''tre-postgres''")
        );
        assert!(
            plan.attach_description
                .contains("META_SECRET 'tre_lake_secret'")
        );
        assert!(!plan.attach_description.contains("super-secret"));
        assert!(!format!("{plan:?}").contains("super-secret"));
        assert!(!format!("{plan:?}").contains("CREATE OR REPLACE SECRET"));
    }

    #[test]
    fn attach_plan_applies_server_managed_encryption_for_initialization() {
        let adapter = DuckLakeAdapter::new("/lake/root");
        let plan = adapter
            .build_attach_plan_with_credential_and_root(
                &test_profile(EncryptionMode::ServerManaged),
                DuckLakeCatalogCredential::session_secret("tre_catalog", "catalog-secret"),
                DuckLakeOpenOptions::initialize(),
                None,
            )
            .expect("attach plan should build");

        assert_eq!(
            plan.configured_encryption_mode,
            EncryptionMode::ServerManaged
        );
        assert!(plan.attach_description.contains("DATA_PATH '/lake/root'"));
        assert!(plan.attach_sql.contains("ENCRYPTED"));
    }

    #[test]
    fn attach_plan_discovers_existing_catalog_encryption_without_encrypted_option() {
        let adapter = DuckLakeAdapter::new("/lake/root");
        let plan = adapter
            .build_attach_plan_with_credential_and_root(
                &test_profile(EncryptionMode::ServerManaged),
                DuckLakeCatalogCredential::session_secret("tre_catalog", "catalog-secret"),
                DuckLakeOpenOptions::attach_existing(),
                None,
            )
            .expect("attach plan should build");

        assert!(!plan.create_if_not_exists);
        assert_eq!(
            plan.configured_encryption_mode,
            EncryptionMode::ServerManaged
        );
        assert!(!plan.attach_sql.contains("ENCRYPTED"));
    }

    #[test]
    fn attach_plan_rejects_client_managed_encryption() {
        let adapter = DuckLakeAdapter::new("/lake/root");
        assert!(matches!(
            adapter.build_attach_plan_with_credential_and_root(
                &test_profile(EncryptionMode::ClientManaged),
                DuckLakeCatalogCredential::session_secret("tre_catalog", "catalog-secret"),
                DuckLakeOpenOptions::initialize(),
                None,
            ),
            Err(LakeError::UnsupportedEncryptionMode(
                EncryptionMode::ClientManaged
            ))
        ));
    }

    #[test]
    fn attach_plan_rejects_empty_credentials() {
        let adapter = DuckLakeAdapter::new("/lake/root");
        assert!(matches!(
            adapter.build_attach_plan_with_credential_and_root(
                &test_profile(EncryptionMode::None),
                DuckLakeCatalogCredential::session_secret("tre_catalog", "  "),
                DuckLakeOpenOptions::default(),
                None,
            ),
            Err(LakeError::EmptyCredential {
                kind: "session_secret_password"
            })
        ));
    }

    fn test_profile(encryption_mode: EncryptionMode) -> DataStoreProfile {
        DataStoreProfile {
            server: "tre-postgres".to_string(),
            port: 5432,
            dbname: "tre".to_string(),
            sslmode: Some("disable".to_string()),
            lake: LakeAttachProfile {
                lake_data: "/lake/root".to_string(),
                lake_db: "tre_lake".to_string(),
                catalog_schema: Some("lake_catalog".to_string()),
                auth: LakeAuthProfile {
                    lake_user: "lake_user".to_string(),
                    encryption_mode,
                },
            },
        }
    }
}