ahri_tre_pgmeta/

datastore_creation.rs

//! PostgreSQL-owned mechanics for starting a configured Datastore creation saga.

use ahri_tre_secrets::SecretMaterial;
use ahri_tre_types::{DatastoreIdentityBinding, NewDatastoreIdentityBinding};

use crate::{
    DirectPostgresConnectConfig, PgMetadataAdapter, PgMetadataConnection,
    datastore_identity_table_exists, insert_datastore_identity_binding,
    read_datastore_identity_bindings,
};

/// Explicit PostgreSQL authority used for one configured creation attempt.
#[derive(Debug, Clone)]
pub struct PgDatastoreCreationConnection {
    host: String,
    port: u16,
    bootstrap_database: String,
    administrator_username: String,
    root_certificates: Vec<String>,
}

impl PgDatastoreCreationConnection {
    pub fn new(
        host: impl Into<String>,
        port: u16,
        bootstrap_database: impl Into<String>,
        administrator_username: impl Into<String>,
        root_certificates: Vec<String>,
    ) -> Self {
        Self {
            host: host.into(),
            port,
            bootstrap_database: bootstrap_database.into(),
            administrator_username: administrator_username.into(),
            root_certificates,
        }
    }

    /// Opens the configured maintenance credential behind a cleanup-only API.
    pub fn dataset_maintenance(
        &self,
        database: &str,
        credential: &SecretMaterial,
    ) -> Result<crate::PgDatasetMaintenance, crate::PgMetaError> {
        credential
            .expose(|bytes| {
                let password =
                    std::str::from_utf8(bytes).map_err(|_| crate::PgMetaError::Decode {
                        field: "maintenance_credential",
                        value: String::new(),
                        message: "maintenance credential is unavailable".into(),
                    })?;
                self.adapter(database, password).connect()
            })
            .and_then(crate::PgDatasetMaintenance::new)
    }

    fn adapter(&self, database: &str, password: &str) -> PgMetadataAdapter {
        self.adapter_for(database, &self.administrator_username, password)
    }

    fn adapter_for(&self, database: &str, username: &str, password: &str) -> PgMetadataAdapter {
        let config = DirectPostgresConnectConfig {
            host: self.host.clone(),
            port: self.port,
            dbname: database.to_string(),
            username: username.to_string(),
            password: Some(password.to_string()),
            sslmode: Some("verify-full".to_string()),
            connect_timeout_secs: Some(10),
        };
        if self.root_certificates.is_empty() {
            PgMetadataAdapter::direct(config)
        } else {
            PgMetadataAdapter::direct_with_ca(config, self.root_certificates.clone())
        }
    }
}

/// PostgreSQL names and initial binding required for a new Datastore.
#[derive(Debug, Clone)]
pub struct PgDatastoreCreationSpec {
    target_database: String,
    target_owner: String,
    catalog_database: String,
    catalog_schema: String,
    catalog_role: String,
    binding: NewDatastoreIdentityBinding,
    governance: crate::governance::GovernanceAppointment,
}

impl PgDatastoreCreationSpec {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        target_database: impl Into<String>,
        target_owner: impl Into<String>,
        catalog_database: impl Into<String>,
        catalog_schema: impl Into<String>,
        catalog_role: impl Into<String>,
        binding: NewDatastoreIdentityBinding,
        governance: crate::governance::GovernanceAppointment,
    ) -> Self {
        Self {
            target_database: target_database.into(),
            target_owner: target_owner.into(),
            catalog_database: catalog_database.into(),
            catalog_schema: catalog_schema.into(),
            catalog_role: catalog_role.into(),
            binding,
            governance,
        }
    }

    fn identifiers(&self) -> [&str; 5] {
        [
            &self.target_database,
            &self.target_owner,
            &self.catalog_database,
            &self.catalog_schema,
            &self.catalog_role,
        ]
    }
}

/// Expected PostgreSQL ownership and privilege boundary for one retained Datastore.
#[derive(Debug, Clone)]
pub struct PgDatastoreValidationSpec {
    target_database: String,
    target_owner: String,
    catalog_database: String,
    catalog_schema: String,
    catalog_role: String,
}

impl PgDatastoreValidationSpec {
    pub fn new(
        target_database: impl Into<String>,
        target_owner: impl Into<String>,
        catalog_database: impl Into<String>,
        catalog_schema: impl Into<String>,
        catalog_role: impl Into<String>,
    ) -> Self {
        Self {
            target_database: target_database.into(),
            target_owner: target_owner.into(),
            catalog_database: catalog_database.into(),
            catalog_schema: catalog_schema.into(),
            catalog_role: catalog_role.into(),
        }
    }

    fn identifiers(&self) -> [&str; 5] {
        [
            &self.target_database,
            &self.target_owner,
            &self.catalog_database,
            &self.catalog_schema,
            &self.catalog_role,
        ]
    }
}

/// PostgreSQL capabilities retained while a creation saga is in progress.
pub struct PgPendingDatastoreCreation {
    _administrator: PgMetadataConnection,
    store: PgMetadataConnection,
}

impl std::fmt::Debug for PgPendingDatastoreCreation {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PgPendingDatastoreCreation")
            .field("administrator_lock_held", &true)
            .field("store", &self.store)
            .finish_non_exhaustive()
    }
}

impl PgPendingDatastoreCreation {
    pub fn store_mut(&mut self) -> &mut PgMetadataConnection {
        &mut self.store
    }
}

/// Closed start result that distinguishes failures with a durable binding.
#[derive(Debug)]
pub enum PgDatastoreCreationStart {
    Started(PgPendingDatastoreCreation),
    Conflict,
    FailedBeforeBinding,
    RollbackFailed,
    FailedAfterBinding(PgPendingDatastoreCreation),
}

/// Closed adapter failure for retained-state inspection and reconciliation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PgDatastoreReconciliationError;

/// Finds the singleton identity binding in an existing target database without
/// adopting an unbound or ambiguously bound database.
pub fn find_existing_datastore_binding(
    connection: &PgDatastoreCreationConnection,
    administrator_password: &SecretMaterial,
    target_database: &str,
) -> Result<Option<DatastoreIdentityBinding>, PgDatastoreReconciliationError> {
    administrator_password.expose(|bytes| {
        let password = std::str::from_utf8(bytes).map_err(|_| PgDatastoreReconciliationError)?;
        let mut administrator = connection
            .adapter(&connection.bootstrap_database, password)
            .connect()
            .map_err(|_| PgDatastoreReconciliationError)?;
        let limit = administrator
            .client()
            .query_one(
                "SELECT current_setting('max_identifier_length')::integer",
                &[],
            )
            .map_err(|_| PgDatastoreReconciliationError)?
            .get::<_, i32>(0);
        if limit <= 0 || !identifiers_fit([target_database], limit as usize) {
            return Err(PgDatastoreReconciliationError);
        }
        let exists = administrator
            .client()
            .query_one(
                "SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname = $1)",
                &[&target_database],
            )
            .map_err(|_| PgDatastoreReconciliationError)?
            .get::<_, bool>(0);
        if !exists {
            return Ok(None);
        }
        let mut target = connection
            .adapter(target_database, password)
            .connect()
            .map_err(|_| PgDatastoreReconciliationError)?;
        let bindings = read_datastore_identity_bindings(&mut target)
            .map_err(|_| PgDatastoreReconciliationError)?;
        let [binding] = bindings.as_slice() else {
            return Err(PgDatastoreReconciliationError);
        };
        Ok(Some(binding.clone()))
    })
}

/// Inspects every connectable database on one configured PostgreSQL server.
/// Any database that cannot be inspected makes absence unprovable.
pub fn inspect_datastore_identity_bindings(
    connection: &PgDatastoreCreationConnection,
    administrator_password: &SecretMaterial,
) -> Result<Vec<DatastoreIdentityBinding>, PgDatastoreReconciliationError> {
    administrator_password.expose(|bytes| {
        let password = std::str::from_utf8(bytes).map_err(|_| PgDatastoreReconciliationError)?;
        let mut administrator = connection
            .adapter(&connection.bootstrap_database, password)
            .connect()
            .map_err(|_| PgDatastoreReconciliationError)?;
        let databases = administrator
            .client()
            .query(
                "SELECT datname FROM pg_database WHERE NOT datistemplate ORDER BY datname",
                &[],
            )
            .map_err(|_| PgDatastoreReconciliationError)?;
        let mut bindings = Vec::new();
        for row in databases {
            let database = row.get::<_, String>(0);
            let mut inspected = connection
                .adapter(&database, password)
                .connect()
                .map_err(|_| PgDatastoreReconciliationError)?;
            if datastore_identity_table_exists(&mut inspected)
                .map_err(|_| PgDatastoreReconciliationError)?
            {
                bindings.extend(
                    read_datastore_identity_bindings(&mut inspected)
                        .map_err(|_| PgDatastoreReconciliationError)?,
                );
            }
        }
        Ok(bindings)
    })
}

/// Verifies retained database/schema ownership, identifier limits, role
/// attributes, memberships, approved service-function ownership, and PUBLIC
/// privilege revocations.
pub fn validate_existing_datastore_postgresql(
    connection: &PgDatastoreCreationConnection,
    administrator_password: &SecretMaterial,
    spec: &PgDatastoreValidationSpec,
) -> Result<(), PgDatastoreReconciliationError> {
    administrator_password.expose(|bytes| {
        let password = std::str::from_utf8(bytes).map_err(|_| PgDatastoreReconciliationError)?;
        let mut administrator = connection
            .adapter(&connection.bootstrap_database, password)
            .connect()
            .map_err(|_| PgDatastoreReconciliationError)?;
        let limit = administrator
            .client()
            .query_one(
                "SELECT current_setting('max_identifier_length')::integer",
                &[],
            )
            .map_err(|_| PgDatastoreReconciliationError)?
            .get::<_, i32>(0);
        if limit <= 0 || !identifiers_fit(spec.identifiers(), limit as usize) {
            return Err(PgDatastoreReconciliationError);
        }
        let databases = administrator
            .client()
            .query(
                "SELECT d.datname, r.rolname, EXISTS (SELECT 1 FROM aclexplode(COALESCE(d.datacl, acldefault('d', d.datdba))) a WHERE a.grantee = 0) AS public_privilege FROM pg_database d JOIN pg_roles r ON r.oid = d.datdba WHERE d.datname IN ($1, $2) ORDER BY d.datname",
                &[&spec.target_database, &spec.catalog_database],
            )
            .map_err(|_| PgDatastoreReconciliationError)?;
        if databases.len() != 2
            || databases.iter().any(|row| row.get::<_, bool>(2))
            || !databases.iter().any(|row| {
                row.get::<_, String>(0) == spec.target_database
                    && row.get::<_, String>(1) == spec.target_owner
            })
            || !databases.iter().any(|row| {
                row.get::<_, String>(0) == spec.catalog_database
                    && row.get::<_, String>(1) == spec.catalog_role
            })
        {
            return Err(PgDatastoreReconciliationError);
        }
        let roles = administrator
            .client()
            .query(
                "SELECT rolname, rolcanlogin, rolsuper, rolcreatedb, rolcreaterole, rolinherit, rolreplication, rolbypassrls, EXISTS (SELECT 1 FROM pg_auth_members m WHERE m.roleid = r.oid) AS has_members, EXISTS (SELECT 1 FROM pg_auth_members m WHERE m.member = r.oid) AS is_member FROM pg_roles r WHERE rolname IN ($1, $2)",
                &[&spec.target_owner, &spec.catalog_role],
            )
            .map_err(|_| PgDatastoreReconciliationError)?;
        if roles.len() != 2
            || roles.iter().any(|row| {
                row.get::<_, bool>(2)
                    || row.get::<_, bool>(3)
                    || row.get::<_, bool>(4)
                    || row.get::<_, bool>(5)
                    || row.get::<_, bool>(6)
                    || row.get::<_, bool>(7)
            })
            || !roles.iter().any(|row| {
                row.get::<_, String>(0) == spec.target_owner
                    && !row.get::<_, bool>(1)
                    && role_has_permitted_memberships(
                        DatastoreRoleMembershipPolicy::TargetOwner,
                        row.get::<_, bool>(8),
                        row.get::<_, bool>(9),
                    )
            })
            || !roles.iter().any(|row| {
                row.get::<_, String>(0) == spec.catalog_role
                    && row.get::<_, bool>(1)
                    && role_has_permitted_memberships(
                        DatastoreRoleMembershipPolicy::IsolatedCatalog,
                        row.get::<_, bool>(8),
                        row.get::<_, bool>(9),
                    )
            })
        {
            return Err(PgDatastoreReconciliationError);
        }
        let mut target = connection
            .adapter(&spec.target_database, password)
            .connect()
            .map_err(|_| PgDatastoreReconciliationError)?;
        let metadata_objects = target
            .client()
            .query_one(
                "SELECT COUNT(*)::bigint, COALESCE(bool_and(r.rolname = $1), false), NOT EXISTS (SELECT 1 FROM pg_class c2 JOIN pg_namespace n2 ON n2.oid = c2.relnamespace CROSS JOIN LATERAL aclexplode(c2.relacl) a WHERE n2.nspname = 'public' AND a.grantee = 0), NOT EXISTS (SELECT 1 FROM pg_proc p JOIN pg_namespace n3 ON n3.oid = p.pronamespace JOIN pg_roles r3 ON r3.oid = p.proowner WHERE n3.nspname = 'public' AND r3.rolname <> $1 AND NOT (p.oid = to_regprocedure('public.tre_access_request_study_exists(uuid)') AND r3.rolname = 'tre_access_request_lookup' AND NOT r3.rolcanlogin AND NOT r3.rolsuper AND NOT r3.rolcreatedb AND NOT r3.rolcreaterole AND NOT r3.rolinherit AND NOT r3.rolreplication AND r3.rolbypassrls AND NOT EXISTS (SELECT 1 FROM pg_auth_members m WHERE m.roleid = r3.oid OR m.member = r3.oid))), NOT EXISTS (SELECT 1 FROM pg_type t JOIN pg_namespace n4 ON n4.oid = t.typnamespace JOIN pg_roles r4 ON r4.oid = t.typowner WHERE n4.nspname = 'public' AND r4.rolname <> $1) FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace JOIN pg_roles r ON r.oid = c.relowner WHERE n.nspname = 'public'",
                &[&spec.target_owner],
            )
            .map_err(|_| PgDatastoreReconciliationError)?;
        if metadata_objects.get::<_, i64>(0) == 0
            || !metadata_objects.get::<_, bool>(1)
            || !metadata_objects.get::<_, bool>(2)
            || !metadata_objects.get::<_, bool>(3)
            || !metadata_objects.get::<_, bool>(4)
        {
            return Err(PgDatastoreReconciliationError);
        }
        let mut catalog = connection
            .adapter(&spec.catalog_database, password)
            .connect()
            .map_err(|_| PgDatastoreReconciliationError)?;
        let schema = catalog
            .client()
            .query(
                "SELECT n.nspname, r.rolname, EXISTS (SELECT 1 FROM aclexplode(COALESCE(n.nspacl, acldefault('n', n.nspowner))) a WHERE a.grantee = 0) AS public_privilege FROM pg_namespace n JOIN pg_roles r ON r.oid = n.nspowner WHERE n.nspname IN ($1, 'public')",
                &[&spec.catalog_schema],
            )
            .map_err(|_| PgDatastoreReconciliationError)?;
        if schema.len() != 2
            || schema.iter().any(|row| row.get::<_, bool>(2))
            || !schema.iter().any(|row| {
                row.get::<_, String>(0) == spec.catalog_schema
                    && row.get::<_, String>(1) == spec.catalog_role
            })
            || !schema.iter().any(|row| {
                row.get::<_, String>(0) == "public"
                    && row.get::<_, String>(1) == "pg_database_owner"
            })
        {
            return Err(PgDatastoreReconciliationError);
        }
        Ok(())
    })
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DatastoreRoleMembershipPolicy {
    TargetOwner,
    IsolatedCatalog,
}

fn role_has_permitted_memberships(
    policy: DatastoreRoleMembershipPolicy,
    has_members: bool,
    is_member: bool,
) -> bool {
    // A datastore owner may be granted to an entry group, but must not inherit
    // another role. The credential-bearing catalog role stays isolated both ways.
    !is_member && (policy == DatastoreRoleMembershipPolicy::TargetOwner || !has_members)
}

/// Creates the target database and immediately records its `creating` binding,
/// then performs the remaining PostgreSQL side effects under the same advisory
/// lock. Failures after the binding is durable return the pending capability so
/// the caller can atomically mark the saga `failed`.
pub fn begin_datastore_creation(
    connection: &PgDatastoreCreationConnection,
    administrator_password: &SecretMaterial,
    catalog_password: &SecretMaterial,
    spec: PgDatastoreCreationSpec,
) -> PgDatastoreCreationStart {
    administrator_password.expose(|administrator_bytes| {
        let Ok(administrator_password) = std::str::from_utf8(administrator_bytes) else {
            return PgDatastoreCreationStart::FailedBeforeBinding;
        };
        let administrator =
            connection.adapter(&connection.bootstrap_database, administrator_password);
        let target = connection.adapter(&spec.target_database, administrator_password);
        let catalog = connection.adapter(&spec.catalog_database, administrator_password);
        catalog_password.expose(|catalog_bytes| {
            let Ok(catalog_password) = std::str::from_utf8(catalog_bytes) else {
                return PgDatastoreCreationStart::FailedBeforeBinding;
            };
            begin_datastore_creation_with_adapters(
                &administrator,
                &target,
                &catalog,
                spec,
                catalog_password,
            )
        })
    })
}

fn begin_datastore_creation_with_adapters(
    administrator: &PgMetadataAdapter,
    target: &PgMetadataAdapter,
    catalog: &PgMetadataAdapter,
    spec: PgDatastoreCreationSpec,
    catalog_password: &str,
) -> PgDatastoreCreationStart {
    let Ok(mut admin) = administrator.connect() else {
        return PgDatastoreCreationStart::FailedBeforeBinding;
    };
    let lock_key = format!("ahri-tre:create:{}", spec.target_database);
    let Ok(lock_row) = admin.client().query_one(
        "SELECT pg_try_advisory_lock(hashtextextended($1, 0))",
        &[&lock_key],
    ) else {
        return PgDatastoreCreationStart::FailedBeforeBinding;
    };
    if !lock_row.get::<_, bool>(0) {
        return PgDatastoreCreationStart::Conflict;
    }

    let Ok(limit_row) = admin.client().query_one(
        "SELECT current_setting('max_identifier_length')::integer",
        &[],
    ) else {
        return PgDatastoreCreationStart::FailedBeforeBinding;
    };
    let limit = limit_row.get::<_, i32>(0);
    if limit <= 0 || !identifiers_fit(spec.identifiers(), limit as usize) {
        return PgDatastoreCreationStart::FailedBeforeBinding;
    }

    let Ok(database_row) = admin.client().query_one(
        "SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname IN ($1, $2))",
        &[&spec.target_database, &spec.catalog_database],
    ) else {
        return PgDatastoreCreationStart::FailedBeforeBinding;
    };
    let Ok(role_row) = admin.client().query_one(
        "SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname IN ($1, $2))",
        &[&spec.target_owner, &spec.catalog_role],
    ) else {
        return PgDatastoreCreationStart::FailedBeforeBinding;
    };
    if database_row.get::<_, bool>(0) || role_row.get::<_, bool>(0) {
        return PgDatastoreCreationStart::Conflict;
    }

    if admin
        .client()
        .batch_execute(&format!(
            "CREATE ROLE {} NOLOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT",
            sql_identifier(&spec.target_owner),
        ))
        .is_err()
    {
        return PgDatastoreCreationStart::FailedBeforeBinding;
    }

    if admin
        .client()
        .batch_execute(&format!(
            "CREATE DATABASE {} OWNER {}",
            sql_identifier(&spec.target_database),
            sql_identifier(&spec.target_owner),
        ))
        .is_err()
    {
        let _ = admin
            .client()
            .batch_execute(&format!("DROP ROLE {}", sql_identifier(&spec.target_owner),));
        return PgDatastoreCreationStart::FailedBeforeBinding;
    }

    let prepared_store = (|| {
        let mut store = target.connect().map_err(|_| ())?;
        store
            .client()
            .batch_execute(&format!(
                "REVOKE ALL ON DATABASE {} FROM PUBLIC; SET ROLE {}",
                sql_identifier(&spec.target_database),
                sql_identifier(&spec.target_owner),
            ))
            .map_err(|_| ())?;
        store.bootstrap_schema().map_err(|_| ())?;
        insert_datastore_identity_binding(&mut store, &spec.binding).map_err(|_| ())?;
        crate::governance::GovernanceUpgrade::prepare(&mut store, &spec.governance)
            .map_err(|_| ())?;
        Ok::<_, ()>(store)
    })();
    let Ok(store) = prepared_store else {
        // This database was proven absent and created by this locked attempt.
        // Removing it here is rollback, not adoption or force-reset behavior.
        let rollback = admin.client().batch_execute(&format!(
            "DROP DATABASE {} WITH (FORCE); DROP ROLE {}",
            sql_identifier(&spec.target_database),
            sql_identifier(&spec.target_owner),
        ));
        return if rollback.is_ok() {
            PgDatastoreCreationStart::FailedBeforeBinding
        } else {
            PgDatastoreCreationStart::RollbackFailed
        };
    };

    let mut pending = PgPendingDatastoreCreation {
        _administrator: admin,
        store,
    };
    let post_binding_result = (|| {
        pending
            ._administrator
            .client()
            .batch_execute(&format!(
                "CREATE ROLE {} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT PASSWORD {}",
                sql_identifier(&spec.catalog_role),
                sql_literal(catalog_password),
            ))
            .map_err(|_| ())?;
        pending
            ._administrator
            .client()
            .batch_execute(&format!(
                "CREATE DATABASE {} OWNER {}",
                sql_identifier(&spec.catalog_database),
                sql_identifier(&spec.catalog_role),
            ))
            .map_err(|_| ())?;

        let mut catalog = catalog.connect().map_err(|_| ())?;
        catalog
            .client()
            .batch_execute(&format!(
                "REVOKE ALL ON DATABASE {} FROM PUBLIC; CREATE SCHEMA {} AUTHORIZATION {}; REVOKE ALL ON SCHEMA public FROM PUBLIC;",
                sql_identifier(&spec.catalog_database),
                sql_identifier(&spec.catalog_schema),
                sql_identifier(&spec.catalog_role),
            ))
            .map_err(|_| ())
    })();

    if post_binding_result.is_err() {
        PgDatastoreCreationStart::FailedAfterBinding(pending)
    } else {
        PgDatastoreCreationStart::Started(pending)
    }
}

fn identifiers_fit<'a>(identifiers: impl IntoIterator<Item = &'a str>, limit: usize) -> bool {
    identifiers
        .into_iter()
        .all(|identifier| !identifier.is_empty() && identifier.len() <= limit)
}

/// Disclosure-safe failure from the PostgreSQL-owned credential authority.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PgDatastoreCredentialRotationError {
    /// The prior password and binding were authoritatively restored.
    Restored,
    /// PostgreSQL may require either the prior or staged password.
    StateUncertain,
}

#[allow(clippy::too_many_arguments)]
pub fn apply_datastore_credential_rotation(
    connection: &PgDatastoreCreationConnection,
    administrator_password: &SecretMaterial,
    prior_password: &SecretMaterial,
    staged_password: &SecretMaterial,
    target_database: &str,
    catalog_database: &str,
    catalog_role: &str,
    binding: &DatastoreIdentityBinding,
    new_version: i32,
    rotated_at: chrono::DateTime<chrono::Utc>,
) -> Result<DatastoreIdentityBinding, PgDatastoreCredentialRotationError> {
    administrator_password.expose(|administrator_bytes| {
        let administrator_password = std::str::from_utf8(administrator_bytes)
            .map_err(|_| PgDatastoreCredentialRotationError::Restored)?;
        prior_password.expose(|prior_bytes| {
            let prior_password = std::str::from_utf8(prior_bytes)
                .map_err(|_| PgDatastoreCredentialRotationError::Restored)?;
            staged_password.expose(|staged_bytes| {
                let staged_password = std::str::from_utf8(staged_bytes)
                    .map_err(|_| PgDatastoreCredentialRotationError::Restored)?;
                let mut administrator = rotation_administrator(
                    connection,
                    administrator_password,
                    binding.datastore_id,
                )?;
                set_role_password(&mut administrator, catalog_role, staged_password)?;
                let applied = verify_role_password(
                    connection,
                    catalog_database,
                    catalog_role,
                    staged_password,
                )
                .and_then(|()| {
                    write_rotation_binding(
                        connection,
                        administrator_password,
                        target_database,
                        binding,
                        new_version,
                        rotated_at,
                    )
                });
                match applied {
                    Ok(binding) => Ok(binding),
                    Err(_) => {
                        let restored_password =
                            set_role_password(&mut administrator, catalog_role, prior_password)
                                .and_then(|()| {
                                    verify_role_password(
                                        connection,
                                        catalog_database,
                                        catalog_role,
                                        prior_password,
                                    )
                                });
                        let restored_binding = restore_rotation_binding(
                            connection,
                            administrator_password,
                            target_database,
                            binding,
                        );
                        if restored_password.is_ok() && restored_binding.is_ok() {
                            Err(PgDatastoreCredentialRotationError::Restored)
                        } else {
                            Err(PgDatastoreCredentialRotationError::StateUncertain)
                        }
                    }
                }
            })
        })
    })
}

pub fn rollback_datastore_credential_rotation(
    connection: &PgDatastoreCreationConnection,
    administrator_password: &SecretMaterial,
    prior_password: &SecretMaterial,
    target_database: &str,
    catalog_role: &str,
    catalog_database: &str,
    binding: &DatastoreIdentityBinding,
) -> Result<(), PgDatastoreCredentialRotationError> {
    administrator_password.expose(|administrator_bytes| {
        let administrator_password = std::str::from_utf8(administrator_bytes)
            .map_err(|_| PgDatastoreCredentialRotationError::StateUncertain)?;
        prior_password.expose(|prior_bytes| {
            let prior_password = std::str::from_utf8(prior_bytes)
                .map_err(|_| PgDatastoreCredentialRotationError::StateUncertain)?;
            let mut administrator =
                rotation_administrator(connection, administrator_password, binding.datastore_id)?;
            set_role_password(&mut administrator, catalog_role, prior_password)
                .and_then(|()| {
                    verify_role_password(connection, catalog_database, catalog_role, prior_password)
                })
                .and_then(|()| {
                    restore_rotation_binding(
                        connection,
                        administrator_password,
                        target_database,
                        binding,
                    )
                    .map(|_| ())
                })
                .map_err(|_| PgDatastoreCredentialRotationError::StateUncertain)
        })
    })
}

fn rotation_administrator(
    connection: &PgDatastoreCreationConnection,
    administrator_password: &str,
    datastore_id: uuid::Uuid,
) -> Result<PgMetadataConnection, PgDatastoreCredentialRotationError> {
    let mut administrator = connection
        .adapter(&connection.bootstrap_database, administrator_password)
        .connect()
        .map_err(|_| PgDatastoreCredentialRotationError::Restored)?;
    let key = format!("ahri-tre:credential-rotation:{datastore_id}");
    let row = administrator
        .client()
        .query_one(
            "SELECT pg_try_advisory_lock(hashtextextended($1, 0))",
            &[&key],
        )
        .map_err(|_| PgDatastoreCredentialRotationError::Restored)?;
    if row.get::<_, bool>(0) {
        Ok(administrator)
    } else {
        Err(PgDatastoreCredentialRotationError::Restored)
    }
}

fn set_role_password(
    administrator: &mut PgMetadataConnection,
    role: &str,
    password: &str,
) -> Result<(), PgDatastoreCredentialRotationError> {
    classify_password_mutation_result(administrator.client().batch_execute(&format!(
        "ALTER ROLE {} PASSWORD {}",
        sql_identifier(role),
        sql_literal(password)
    )))
}

fn classify_password_mutation_result<T, E>(
    result: Result<T, E>,
) -> Result<T, PgDatastoreCredentialRotationError> {
    result.map_err(|_| PgDatastoreCredentialRotationError::StateUncertain)
}

fn verify_role_password(
    connection: &PgDatastoreCreationConnection,
    database: &str,
    role: &str,
    password: &str,
) -> Result<(), PgDatastoreCredentialRotationError> {
    let mut verified = connection
        .adapter_for(database, role, password)
        .connect()
        .map_err(|_| PgDatastoreCredentialRotationError::Restored)?;
    let row = verified
        .client()
        .query_one("SELECT current_user", &[])
        .map_err(|_| PgDatastoreCredentialRotationError::Restored)?;
    if row.get::<_, String>(0) == role {
        Ok(())
    } else {
        Err(PgDatastoreCredentialRotationError::Restored)
    }
}

fn write_rotation_binding(
    connection: &PgDatastoreCreationConnection,
    administrator_password: &str,
    target_database: &str,
    binding: &DatastoreIdentityBinding,
    version: i32,
    rotated_at: chrono::DateTime<chrono::Utc>,
) -> Result<DatastoreIdentityBinding, PgDatastoreCredentialRotationError> {
    let mut expected = binding.clone();
    expected.credential_version = version;
    expected.credential_last_rotated_at = Some(rotated_at);
    let reference = expected
        .managed_secret_ref
        .clone()
        .ok_or(PgDatastoreCredentialRotationError::Restored)?;
    let fingerprint = expected.computed_binding_fingerprint();
    let mut target = connection
        .adapter(target_database, administrator_password)
        .connect()
        .map_err(|_| PgDatastoreCredentialRotationError::Restored)?;
    crate::update_datastore_identity_credential(
        &mut target,
        binding.datastore_id,
        &reference,
        version,
        rotated_at,
        &fingerprint,
    )
    .map_err(|_| PgDatastoreCredentialRotationError::Restored)
}

fn restore_rotation_binding(
    connection: &PgDatastoreCreationConnection,
    administrator_password: &str,
    target_database: &str,
    binding: &DatastoreIdentityBinding,
) -> Result<DatastoreIdentityBinding, PgDatastoreCredentialRotationError> {
    let reference = binding
        .managed_secret_ref
        .as_deref()
        .ok_or(PgDatastoreCredentialRotationError::Restored)?;
    let rotated_at = binding
        .credential_last_rotated_at
        .ok_or(PgDatastoreCredentialRotationError::Restored)?;
    let mut target = connection
        .adapter(target_database, administrator_password)
        .connect()
        .map_err(|_| PgDatastoreCredentialRotationError::Restored)?;
    crate::update_datastore_identity_credential(
        &mut target,
        binding.datastore_id,
        reference,
        binding.credential_version,
        rotated_at,
        &binding.binding_fingerprint,
    )
    .map_err(|_| PgDatastoreCredentialRotationError::Restored)
}

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

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

impl PgDatastoreCreationConnection {
    /// Preserving operator upgrade with the original metadata owner. Only an
    /// already resolved administrator credential can create this capability.
    pub fn upgrade_governance(
        &self,
        database: &str,
        owner: &str,
        credential: &SecretMaterial,
        appointment: &crate::governance::GovernanceAppointment,
    ) -> Result<crate::governance::GovernanceMaintenance, crate::PgMetaError> {
        credential.expose(|bytes| {
            let password=std::str::from_utf8(bytes).map_err(|_|crate::PgMetaError::Decode { field:"administrator",value:String::new(),message:"administrator credential unavailable".into() })?;
            let mut connection=self.adapter(database,password).connect()?;
            let actual=connection.client().query_one("SELECT pg_get_userbyid(datdba) FROM pg_database WHERE datname=current_database()",&[]).map_err(crate::PgMetaError::Query)?.get::<_,String>(0);
            if actual!=owner { return Err(crate::PgMetaError::Decode {field:"metadata_owner",value:String::new(),message:"metadata owner mismatch".into()}); }
            connection.client().batch_execute(&format!("SET ROLE {}",sql_identifier(owner))).map_err(crate::PgMetaError::Query)?;
            crate::governance::GovernanceUpgrade::prepare(&mut connection,appointment)?;
            connection.client().batch_execute("RESET ROLE").map_err(crate::PgMetaError::Query)?;
            Ok(crate::PgDatasetMaintenance::new(connection)?.governance())
        })
    }
}

#[cfg(test)]
mod tests {
    use super::{
        DatastoreRoleMembershipPolicy, PgDatastoreCredentialRotationError,
        classify_password_mutation_result, identifiers_fit, role_has_permitted_memberships,
    };

    #[test]
    fn validates_identifiers_by_server_reported_byte_limit() {
        assert!(identifiers_fit(["target", "catalog", "role"], 63));
        assert!(!identifiers_fit([""], 63));
        assert!(!identifiers_fit(["a".repeat(64).as_str()], 63));
        assert!(!identifiers_fit(["é".repeat(32).as_str()], 63));
    }

    #[test]
    fn transport_failure_after_password_mutation_is_state_uncertain() {
        let result = classify_password_mutation_result::<(), _>(Err("connection lost"));

        assert_eq!(
            result,
            Err(PgDatastoreCredentialRotationError::StateUncertain)
        );
    }

    #[test]
    fn datastore_owner_may_be_granted_to_an_entry_group_without_inheriting_another_role() {
        assert!(role_has_permitted_memberships(
            DatastoreRoleMembershipPolicy::TargetOwner,
            true,
            false
        ));
        assert!(!role_has_permitted_memberships(
            DatastoreRoleMembershipPolicy::TargetOwner,
            false,
            true
        ));
    }

    #[test]
    fn catalog_role_must_have_no_memberships_in_either_direction() {
        assert!(role_has_permitted_memberships(
            DatastoreRoleMembershipPolicy::IsolatedCatalog,
            false,
            false
        ));
        assert!(!role_has_permitted_memberships(
            DatastoreRoleMembershipPolicy::IsolatedCatalog,
            true,
            false
        ));
        assert!(!role_has_permitted_memberships(
            DatastoreRoleMembershipPolicy::IsolatedCatalog,
            false,
            true
        ));
    }
}