ahri_tre_pgmeta/

repository.rs

use crate::{
    MetadataExecutor, MetadataTransaction, PgAssetReadModel, PgAssetVersionDuoRestrictionReadModel,
    PgAssetVersionEntityReadModel, PgAssetVersionReadModel,
    PgAssetVersionRelationInstanceReadModel, PgCreateAsset, PgCreateAssetVersion,
    PgCreateAssetVersionDuoRestriction, PgCreateDataFile, PgCreateDomain, PgCreateEntity,
    PgCreateEntityRelation, PgCreateStudy, PgCreateStudyDuoRestriction, PgCreateTransformation,
    PgCreateVariable, PgCreateVocabulary, PgCreateVocabularyItem, PgCreateVocabularyMapping,
    PgDataFileReadModel, PgDatasetReadModel, PgDatasetVariableReadModel,
    PgDatasetVersionEntityReadModel, PgDatasetVersionRelationInstanceReadModel, PgDomainReadModel,
    PgDuoQualifiers, PgDuoRestrictionReadModel, PgEntityInstanceReadModel, PgEntityReadModel,
    PgEntityRelationReadModel, PgMetaError, PgMetadataConnection, PgRelationInstanceReadModel,
    PgStudyAccessReadModel, PgStudyCustodianReadModel, PgStudyEntityInstanceReadModel,
    PgStudyReadModel, PgStudyRelationInstanceReadModel, PgTagReadModel, PgTagTarget,
    PgTransformationInputReadModel, PgTransformationOutputReadModel, PgTransformationReadModel,
    PgValueTypeReadModel, PgVariableReadModel, PgVocabularyItemReadModel, PgVocabularyReadModel,
    PgWriteEntityInstance, PgWriteRelationInstance, PgWriteStudyEntityInstance,
    PgWriteStudyRelationInstance, PgWriteTagTarget,
};
use ahri_tre_core::{
    AssetRepository, CatalogueRegistrationRepository, CoreError, DataFileSearchCursorAnchor,
    DataFileSearchPredicate, DataFileSearchQuery, DataFileSearchRecord,
    DataFileSearchStudySelector, DataFileSearchTextField, DatasetSearchCursorAnchor,
    DatasetSearchDomainSelector, DatasetSearchPredicate, DatasetSearchQuery, DatasetSearchRecord,
    DatasetSearchStudySelector, DatasetSearchTextField, DomainRepository, EntityRepository,
    EntitySearchCursorAnchor, EntitySearchDomainSelector, EntitySearchPredicate, EntitySearchQuery,
    EntitySearchRecord, EntitySearchTextField, RelationSearchCursorAnchor,
    RelationSearchDomainSelector, RelationSearchEntitySelector, RelationSearchPredicate,
    RelationSearchQuery, RelationSearchRecord, RelationSearchTextField, SearchPage,
    StudyDomainRepository, StudyGovernanceRepository, StudyRepository, StudySearchCursorAnchor,
    StudySearchDomainSelector, StudySearchPredicate, StudySearchQuery, StudySearchTextField,
    TagAttachmentTarget, TagRepository, TransformationRepository, VariableRepository,
    VariableSearchCursorAnchor, VariableSearchDomainSelector, VariableSearchPredicate,
    VariableSearchQuery, VariableSearchRecord, VariableSearchTextField,
    VariableSearchVocabularySelector, VocabularyRepository, predicate_set_matches,
    search_tags_match_selector, search_text_matches,
};
use ahri_tre_libpq_oauth::{LibpqOAuthConnection, LibpqOAuthRow};
use ahri_tre_types::{
    AssetId, AssetRecord, AssetType, AssetVersionDuoRestrictionId,
    AssetVersionDuoRestrictionRecord, AssetVersionEntityRecord, AssetVersionRecord,
    AssetVersionRelationInstanceRecord, DataFileRecord, DatasetRecord, DatasetVersionEntityRecord,
    DatasetVersionRelationInstanceRecord, DatasetVersionWithdrawalRecord, DomainId, DomainRecord,
    EntityId, EntityInstanceRecord, EntityInstanceWriteRecord, EntityRecord, EntityRelationId,
    EntityRelationRecord, KeyRole, NcName, NewAssetRecord, NewAssetVersionDuoRestrictionRecord,
    NewAssetVersionRecord, NewDatasetVersionWithdrawalRecord, NewDomainRecord,
    NewEntityInstanceRecord, NewEntityRecord, NewEntityRelationRecord, NewRelationInstanceRecord,
    NewStudyDuoRestrictionRecord, NewStudyRecord, NewTransformationRecord, NewVariableRecord,
    NewVocabularyItemRecord, NewVocabularyMappingRecord, NewVocabularyRecord,
    RelationInstanceRecord, RelationInstanceWriteRecord, Sha256Hex, StudyAccessGrantRecord,
    StudyCustodianLinkRecord, StudyDomainLinkRecord, StudyDuoRestrictionId,
    StudyDuoRestrictionRecord, StudyEntityInstanceRecord, StudyEntityInstanceWriteRecord, StudyId,
    StudyRecord, StudyRelationInstanceRecord, StudyRelationInstanceWriteRecord, StudyTypeId, TagId,
    TagRecord, TransformationId, TransformationInputLinkRecord, TransformationOutputLinkRecord,
    TransformationRecord, TransformationType, ValueTypeId, ValueTypeRecord, VariableId,
    VariableRecord, VersionId, VocabularyId, VocabularyItemId, VocabularyItemRecord,
    VocabularyMappingRecord, VocabularyRecord,
};
use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};

const OAUTH_ADD_STUDY_DELEGATE_CUSTODIAN_SQL: &str = "
        SELECT public.tre_add_study_delegate_custodian($1, $2)
        ";
const OAUTH_TRANSFER_PRIMARY_CUSTODIANSHIP_SQL: &str = "
        SELECT public.tre_transfer_study_primary_custodianship($1, $2)
        ";
const OAUTH_REMOVE_STUDY_DELEGATE_CUSTODIAN_SQL: &str = "
        SELECT public.tre_remove_study_delegate_custodian($1, $2)
        ";
const OAUTH_LIST_DATASET_VARIABLES_SQL: &str = "
        SELECT dv.dataset_id::text, dv.variable_id::text, dv.row_role::text
          FROM public.dataset_variables AS dv
         WHERE dv.dataset_id = $1
         ORDER BY dv.variable_id
        ";
const DATASET_VERSION_WITHDRAWALS_RELATION: &str = "public.dataset_version_withdrawals";
const DATASET_VERSION_WITHDRAWALS_EXISTS_FIELD: &str = "dataset_version_withdrawals_exists";

/// Core repository implementation backed by PostgreSQL metadata tables.
#[derive(Debug, Clone)]
pub struct PgMetadataRepository<'connection> {
    pub(super) observation: Option<ahri_tre_observability::CorrelationContext>,
    pub(super) connection: PgMetadataRepositoryConnection<'connection>,
}

#[derive(Debug, Clone)]
pub(super) enum PgMetadataRepositoryConnection<'connection> {
    Direct(Arc<Mutex<PgMetadataConnection>>),
    OAuth(Arc<Mutex<LibpqOAuthConnection>>),
    ScopedDirect(Arc<Mutex<&'connection mut PgMetadataConnection>>),
    ScopedOAuth(Arc<Mutex<&'connection mut LibpqOAuthConnection>>),
}

impl PgMetadataRepository<'_> {
    /// Creates a repository that owns a direct PostgreSQL metadata connection.
    pub fn new(connection: PgMetadataConnection) -> Self {
        Self {
            observation: None,
            connection: PgMetadataRepositoryConnection::Direct(Arc::new(Mutex::new(connection))),
        }
    }

    /// Creates a repository from a shared direct PostgreSQL metadata connection.
    pub fn from_shared(connection: Arc<Mutex<PgMetadataConnection>>) -> Self {
        Self {
            observation: None,
            connection: PgMetadataRepositoryConnection::Direct(connection),
        }
    }

    /// Creates a repository from a shared OAuth-authenticated metadata connection.
    pub fn from_oauth_shared(connection: Arc<Mutex<LibpqOAuthConnection>>) -> Self {
        Self {
            observation: None,
            connection: PgMetadataRepositoryConnection::OAuth(connection),
        }
    }

    /// Scope observation to a newly constructed repository, never the retained connection.
    pub fn with_correlation(mut self, context: ahri_tre_observability::CorrelationContext) -> Self {
        self.observation = Some(context);
        self
    }

    pub(super) fn with_session_connection<T>(
        &self,
        operation: &'static str,
        direct: impl FnOnce(&mut PgMetadataConnection) -> Result<T, PgMetaError>,
        oauth: impl FnOnce(&mut LibpqOAuthConnection) -> Result<T, PgMetaError>,
    ) -> Result<T, CoreError> {
        match &self.connection {
            PgMetadataRepositoryConnection::Direct(connection) => {
                let mut connection = connection.lock().map_err(|_| {
                    CoreError::Infrastructure(format!(
                        "metadata connection lock was poisoned while trying to {operation}"
                    ))
                })?;
                direct(&mut connection).map_err(|error| pgmeta_error(operation, error))
            }
            PgMetadataRepositoryConnection::OAuth(connection) => {
                let mut connection = connection.lock().map_err(|_| {
                    CoreError::Infrastructure(format!(
                        "OAuth metadata connection lock was poisoned while trying to {operation}"
                    ))
                })?;
                oauth(&mut connection).map_err(|error| pgmeta_error(operation, error))
            }
            PgMetadataRepositoryConnection::ScopedDirect(connection) => {
                let mut connection = connection.lock().map_err(|_| {
                    CoreError::Infrastructure(
                        "Dataset admission connection lock was poisoned".into(),
                    )
                })?;
                direct(&mut connection).map_err(|error| pgmeta_error(operation, error))
            }
            PgMetadataRepositoryConnection::ScopedOAuth(connection) => {
                let mut connection = connection.lock().map_err(|_| {
                    CoreError::Infrastructure(
                        "Dataset admission connection lock was poisoned".into(),
                    )
                })?;
                oauth(&mut connection).map_err(|error| pgmeta_error(operation, error))
            }
        }
    }

    async fn create_transformation_from_input(
        &self,
        input: PgCreateTransformation,
    ) -> Result<TransformationRecord, CoreError> {
        let transformation_id = self.with_session_connection(
            "create transformation",
            |connection| connection.create_transformation(input.clone()),
            |connection| oauth_create_transformation(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created transformation",
            |connection| connection.get_transformation(transformation_id),
            |connection| oauth_get_transformation(connection, transformation_id),
        )?
        .map(transformation_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created transformation {:?} was not readable",
                transformation_id
            ))
        })
    }
}

#[async_trait::async_trait]
impl DomainRepository for PgMetadataRepository<'_> {
    async fn list_domains(&self) -> Result<Vec<DomainRecord>, CoreError> {
        use ahri_tre_observability::{FailureCategory, Measurement, Outcome, Stage};
        let span = self
            .observation
            .map(|context| context.span(Stage::MetadataQuery));
        let category = std::cell::Cell::new(None);
        let observe = |error: PgMetaError| {
            category.set(Some(error.operational_category()));
            error
        };
        let result: Result<Vec<_>, CoreError> = self
            .with_session_connection(
                "list domains",
                |connection| connection.list_domains().map_err(&observe),
                |connection| oauth_list_domains(connection).map_err(&observe),
            )
            .and_then(|rows| rows.into_iter().map(domain_record).collect());
        if let Some(span) = span {
            let (outcome, failure) = match (&result, category.get()) {
                (Ok(_), _) => (Outcome::Success, None),
                (
                    Err(_),
                    Some(
                        category @ (FailureCategory::Authentication
                        | FailureCategory::Authorization),
                    ),
                ) => (Outcome::Rejected, Some(category)),
                (Err(_), Some(category)) => (Outcome::Unavailable, Some(category)),
                (Err(_), None) => (
                    Outcome::InternalFailure,
                    Some(FailureCategory::Compatibility),
                ),
            };
            let evidence = result
                .as_ref()
                .ok()
                .map(|rows| Measurement::ReturnedRecords(rows.len() as u64));
            span.finish_with_evidence(outcome, failure, evidence.as_slice());
        }
        result
    }

    async fn list_domains_by_name(&self, name: &str) -> Result<Vec<DomainRecord>, CoreError> {
        self.with_session_connection(
            "list domains by name",
            |connection| connection.list_domains_by_name(name),
            |connection| oauth_list_domains_by_name(connection, name),
        )?
        .into_iter()
        .map(domain_record)
        .collect()
    }

    async fn get_domain_by_id(
        &self,
        domain_id: DomainId,
    ) -> Result<Option<DomainRecord>, CoreError> {
        self.with_session_connection(
            "get domain by id",
            |connection| connection.get_domain(domain_id),
            |connection| oauth_get_domain(connection, domain_id),
        )?
        .map(domain_record)
        .transpose()
    }

    async fn get_domain_by_name(&self, name: &str) -> Result<Option<DomainRecord>, CoreError> {
        self.with_session_connection(
            "get domain by name",
            |connection| connection.get_domain_by_name(name),
            |connection| oauth_get_domain_by_name(connection, name),
        )?
        .map(domain_record)
        .transpose()
    }

    async fn create_domain(&self, domain: &NewDomainRecord) -> Result<DomainRecord, CoreError> {
        let input = PgCreateDomain {
            domain_id: None,
            name: domain.name.as_str().to_string(),
            uri: domain.uri.clone(),
            description: domain.description.clone(),
        };
        let domain_id = self.with_session_connection(
            "create domain",
            |connection| connection.create_domain(input.clone()),
            |connection| oauth_create_domain(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created domain",
            |connection| connection.get_domain(domain_id),
            |connection| oauth_get_domain(connection, domain_id),
        )?
        .map(domain_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!("created domain {:?} was not readable", domain_id))
        })
    }

    async fn save_domain(&self, domain: &DomainRecord) -> Result<DomainRecord, CoreError> {
        let input = PgCreateDomain {
            domain_id: Some(domain.domain_id),
            name: domain.name.as_str().to_string(),
            uri: domain.uri.clone(),
            description: domain.description.clone(),
        };
        let domain_id = self.with_session_connection(
            "create domain",
            |connection| connection.create_domain(input.clone()),
            |connection| oauth_create_domain(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created domain",
            |connection| connection.get_domain(domain_id),
            |connection| oauth_get_domain(connection, domain_id),
        )?
        .map(domain_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!("created domain {:?} was not readable", domain_id))
        })
    }

    async fn delete_domain_record(&self, domain_id: DomainId) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete domain record",
            |connection| delete_semantic_record(connection, "domains", "domain_id", domain_id.0),
            |connection| delete_semantic_record(connection, "domains", "domain_id", domain_id.0),
        )
    }
}

#[async_trait::async_trait]
impl CatalogueRegistrationRepository for PgMetadataRepository<'_> {
    async fn create_domain_registration(
        &self,
        domain: &NewDomainRecord,
        normalized_tags: &[String],
    ) -> Result<DomainRecord, CoreError> {
        let domain = domain.clone();
        let tags = normalized_tags.to_vec();
        self.with_session_connection(
            "create Domain registration",
            |connection| direct_create_domain_registration(connection, &domain, &tags),
            |connection| oauth_create_domain_registration(connection, &domain, &tags),
        )
    }

    async fn create_study_registration(
        &self,
        study: &NewStudyRecord,
        domain_ids: &[DomainId],
        normalized_tags: &[String],
    ) -> Result<StudyRecord, CoreError> {
        let study = study.clone();
        let domain_ids = domain_ids.to_vec();
        let tags = normalized_tags.to_vec();
        self.with_session_connection(
            "create Study registration",
            |connection| direct_create_study_registration(connection, &study, &domain_ids, &tags),
            |connection| oauth_create_study_registration(connection, &study, &domain_ids, &tags),
        )
    }

    async fn save_domain_registration(
        &self,
        domain: &DomainRecord,
        normalized_tags: &[String],
    ) -> Result<DomainRecord, CoreError> {
        let domain = domain.clone();
        let tags = normalized_tags.to_vec();
        self.with_session_connection(
            "save Domain registration",
            |connection| direct_save_domain_registration(connection, &domain, &tags),
            |connection| oauth_save_domain_registration(connection, &domain, &tags),
        )
    }

    async fn save_study_registration(
        &self,
        study: &StudyRecord,
        domain_ids: &[DomainId],
        normalized_tags: &[String],
    ) -> Result<StudyRecord, CoreError> {
        let study = study.clone();
        let domain_ids = domain_ids.to_vec();
        let tags = normalized_tags.to_vec();
        self.with_session_connection(
            "save Study registration",
            |connection| direct_save_study_registration(connection, &study, &domain_ids, &tags),
            |connection| oauth_save_study_registration(connection, &study, &domain_ids, &tags),
        )
    }
}

#[async_trait::async_trait]
impl StudyRepository for PgMetadataRepository<'_> {
    async fn study_type_exists(&self, id: StudyTypeId) -> Result<bool, CoreError> {
        self.with_session_connection(
            "lookup Study type",
            |connection| {
                Ok(connection
                    .list_study_types()?
                    .iter()
                    .any(|row| row.study_type_id == id))
            },
            |connection| {
                Ok(MetadataExecutor::query_optional(
                    connection,
                    "lookup Study type",
                    "SELECT study_type_id::text FROM public.study_types WHERE study_type_id = $1",
                    &[&metadata_i32(id.0, "study_type_id")?],
                )?
                .is_some())
            },
        )
    }

    async fn list_studies(&self) -> Result<Vec<StudyRecord>, CoreError> {
        self.with_session_connection(
            "list studies",
            |connection| connection.list_studies(),
            oauth_list_studies,
        )?
        .into_iter()
        .map(study_record)
        .collect()
    }

    async fn get_study_by_id(&self, study_id: StudyId) -> Result<Option<StudyRecord>, CoreError> {
        self.with_session_connection(
            "get study by id",
            |connection| connection.get_study(study_id),
            |connection| oauth_get_study(connection, study_id),
        )?
        .map(study_record)
        .transpose()
    }

    async fn get_study_by_name(&self, name: &str) -> Result<Option<StudyRecord>, CoreError> {
        self.with_session_connection(
            "get study by name",
            |connection| connection.get_study_by_name(name),
            |connection| oauth_get_study_by_name(connection, name),
        )?
        .map(study_record)
        .transpose()
    }

    async fn search_studies(
        &self,
        query: &StudySearchQuery,
    ) -> Result<SearchPage<StudyRecord, StudySearchCursorAnchor>, CoreError> {
        let studies = <Self as StudyRepository>::list_studies(self).await?;
        let mut matched = Vec::new();
        for study in studies {
            let domains =
                <Self as StudyDomainRepository>::list_domains_for_study(self, study.study_id)
                    .await?;
            let tags = <Self as TagRepository>::list_tags_for_target(
                self,
                TagAttachmentTarget::Study(study.study_id),
            )
            .await?;
            if study_matches_search_query(&study, &domains, &tags, query) {
                matched.push(study);
            }
        }

        matched.sort_by(study_search_sort_key);
        if let Some(after) = query.after.as_ref() {
            matched.retain(|study| study_sort_tuple(study) > study_cursor_tuple(after));
        }

        let limit = usize::from(query.limit);
        let has_more = matched.len() > limit;
        if has_more {
            matched.truncate(limit);
        }
        let next_cursor = if has_more {
            matched.last().map(|study| StudySearchCursorAnchor {
                name: study.name.as_str().to_string(),
                study_id: study.study_id,
            })
        } else {
            None
        };

        Ok(SearchPage {
            items: matched,
            next_cursor,
        })
    }

    async fn create_study(&self, study: &NewStudyRecord) -> Result<StudyRecord, CoreError> {
        let input = PgCreateStudy {
            study_id: None,
            name: study.name.as_str().to_string(),
            description: study.description.clone(),
            documentation: study.documentation.clone(),
            agent_instructions: study.agent_instructions.clone(),
            external_id: study.external_id.clone(),
            study_type_id: study.study_type_id,
            date_created: study.date_created,
            created_by: study.created_by.clone(),
        };
        let study_id = self.with_session_connection(
            "create study",
            |connection| connection.create_study(input.clone()),
            |connection| oauth_create_study(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created study",
            |connection| connection.get_study(study_id),
            |connection| oauth_get_study(connection, study_id),
        )?
        .map(study_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!("created study {:?} was not readable", study_id))
        })
    }

    async fn save_study(&self, study: &StudyRecord) -> Result<StudyRecord, CoreError> {
        let input = PgCreateStudy {
            study_id: Some(study.study_id),
            name: study.name.as_str().to_string(),
            description: study.description.clone(),
            documentation: study.documentation.clone(),
            agent_instructions: study.agent_instructions.clone(),
            external_id: study.external_id.clone(),
            study_type_id: study.study_type_id,
            date_created: study.date_created,
            created_by: study.created_by.clone(),
        };
        let study_id = self.with_session_connection(
            "create study",
            |connection| connection.create_study(input.clone()),
            |connection| oauth_create_study(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created study",
            |connection| connection.get_study(study_id),
            |connection| oauth_get_study(connection, study_id),
        )?
        .map(study_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!("created study {:?} was not readable", study_id))
        })
    }

    async fn delete_study_record(&self, study_id: StudyId) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete study record",
            |connection| delete_study_metadata_record(connection, study_id),
            |connection| delete_study_metadata_record(connection, study_id),
        )
    }
}

#[async_trait::async_trait]
impl StudyDomainRepository for PgMetadataRepository<'_> {
    async fn link_study_domain(
        &self,
        study_id: StudyId,
        domain_id: DomainId,
    ) -> Result<StudyDomainLinkRecord, CoreError> {
        self.with_session_connection(
            "link study domain",
            |connection| connection.link_study_domain(study_id, domain_id),
            |connection| oauth_link_study_domain(connection, study_id, domain_id),
        )?;
        Ok(StudyDomainLinkRecord {
            study_id,
            domain_id,
        })
    }

    async fn list_domains_for_study(
        &self,
        study_id: StudyId,
    ) -> Result<Vec<DomainRecord>, CoreError> {
        self.with_session_connection(
            "list study domains",
            |connection| connection.list_study_domains(study_id),
            |connection| oauth_list_study_domains(connection, study_id),
        )?
        .into_iter()
        .map(domain_record)
        .collect()
    }
}

#[async_trait::async_trait]
impl StudyGovernanceRepository for PgMetadataRepository<'_> {
    async fn list_access_grants(
        &self,
        study_id: StudyId,
    ) -> Result<Vec<StudyAccessGrantRecord>, CoreError> {
        self.with_session_connection(
            "list study access",
            |connection| connection.list_study_access(study_id),
            |connection| oauth_list_study_access(connection, study_id),
        )?
        .into_iter()
        .map(study_access_record)
        .collect()
    }

    async fn grant_study_access(
        &self,
        study_id: StudyId,
        target_user_id: &str,
    ) -> Result<StudyAccessGrantRecord, CoreError> {
        let normalized_target_user_id = target_user_id.trim().to_string();
        self.with_session_connection(
            "grant study access",
            |connection| connection.grant_study_access(study_id, &normalized_target_user_id),
            |connection| oauth_grant_study_access(connection, study_id, &normalized_target_user_id),
        )?;
        Ok(StudyAccessGrantRecord {
            study_id,
            user_id: normalized_target_user_id,
            date_granted: None,
            granted_by: None,
        })
    }

    async fn save_access_grant(
        &self,
        grant: &StudyAccessGrantRecord,
    ) -> Result<StudyAccessGrantRecord, CoreError> {
        self.grant_study_access(grant.study_id, &grant.user_id)
            .await
    }

    async fn revoke_access_grant(
        &self,
        study_id: StudyId,
        principal: &str,
    ) -> Result<(), CoreError> {
        self.revoke_study_access(study_id, principal).await
    }

    async fn revoke_study_access(
        &self,
        study_id: StudyId,
        target_user_id: &str,
    ) -> Result<(), CoreError> {
        let normalized_target_user_id = target_user_id.trim().to_string();
        self.with_session_connection(
            "revoke study access",
            |connection| connection.revoke_study_access(study_id, &normalized_target_user_id),
            |connection| {
                oauth_revoke_study_access(connection, study_id, &normalized_target_user_id)
            },
        )
    }

    async fn list_custodians(
        &self,
        study_id: StudyId,
    ) -> Result<Vec<StudyCustodianLinkRecord>, CoreError> {
        self.with_session_connection(
            "list study custodians",
            |connection| connection.list_study_custodians(study_id),
            |connection| oauth_list_study_custodians(connection, study_id),
        )?
        .into_iter()
        .map(study_custodian_record)
        .collect()
    }

    async fn add_delegate_custodian(
        &self,
        study_id: StudyId,
        target_user_id: &str,
    ) -> Result<StudyCustodianLinkRecord, CoreError> {
        let normalized_target_user_id = target_user_id.trim().to_string();
        self.with_session_connection(
            "add study delegate custodian",
            |connection| {
                connection.add_study_delegate_custodian(study_id, &normalized_target_user_id)
            },
            |connection| {
                oauth_add_study_delegate_custodian(connection, study_id, &normalized_target_user_id)
            },
        )?;
        Ok(StudyCustodianLinkRecord {
            study_id,
            user_id: normalized_target_user_id,
            is_primary: false,
            date_granted: None,
            granted_by: None,
        })
    }

    async fn transfer_primary_custodianship(
        &self,
        study_id: StudyId,
        target_user_id: &str,
    ) -> Result<StudyCustodianLinkRecord, CoreError> {
        let normalized_target_user_id = target_user_id.trim().to_string();
        self.with_session_connection(
            "transfer study primary custodianship",
            |connection| {
                connection
                    .transfer_study_primary_custodianship(study_id, &normalized_target_user_id)
            },
            |connection| {
                oauth_transfer_study_primary_custodianship(
                    connection,
                    study_id,
                    &normalized_target_user_id,
                )
            },
        )?;
        Ok(StudyCustodianLinkRecord {
            study_id,
            user_id: normalized_target_user_id,
            is_primary: true,
            date_granted: None,
            granted_by: None,
        })
    }

    async fn remove_delegate_custodian(
        &self,
        study_id: StudyId,
        target_user_id: &str,
    ) -> Result<(), CoreError> {
        let normalized_target_user_id = target_user_id.trim().to_string();
        self.with_session_connection(
            "remove study delegate custodian",
            |connection| {
                connection.remove_study_delegate_custodian(study_id, &normalized_target_user_id)
            },
            |connection| {
                oauth_remove_study_delegate_custodian(
                    connection,
                    study_id,
                    &normalized_target_user_id,
                )
            },
        )
    }

    async fn save_custodian(
        &self,
        custodian: &StudyCustodianLinkRecord,
    ) -> Result<StudyCustodianLinkRecord, CoreError> {
        if !custodian.is_primary
            && custodian.date_granted.is_none()
            && custodian.granted_by.is_none()
        {
            return self
                .add_delegate_custodian(custodian.study_id, &custodian.user_id)
                .await;
        }

        self.with_session_connection(
            "save study custodian",
            |connection| {
                connection.add_study_custodian_with_metadata(
                    custodian.study_id,
                    &custodian.user_id,
                    custodian.is_primary,
                    custodian.date_granted,
                    custodian.granted_by.as_deref(),
                )
            },
            |connection| oauth_save_study_custodian(connection, custodian),
        )?;
        Ok(custodian.clone())
    }

    async fn list_study_duo_restrictions(
        &self,
        study_id: StudyId,
    ) -> Result<Vec<StudyDuoRestrictionRecord>, CoreError> {
        self.with_session_connection(
            "list study DUO restrictions",
            |connection| connection.list_study_duo_restrictions(study_id),
            |connection| oauth_list_study_duo_restrictions(connection, study_id),
        )?
        .into_iter()
        .map(study_duo_restriction_record)
        .collect()
    }

    async fn create_study_duo_restriction(
        &self,
        restriction: &NewStudyDuoRestrictionRecord,
    ) -> Result<StudyDuoRestrictionRecord, CoreError> {
        let input = pg_create_study_duo_restriction(None, restriction);
        let restriction_id = self.with_session_connection(
            "create study DUO restriction",
            |connection| connection.create_study_duo_restriction(input.clone()),
            |connection| oauth_create_study_duo_restriction(connection, input.clone()),
        )?;
        Ok(StudyDuoRestrictionRecord {
            study_duo_restriction_id: StudyDuoRestrictionId(restriction_id),
            study_id: restriction.study_id,
            duo_id: restriction.duo_id.clone(),
            qualifier_ontology_namespace: restriction.qualifier_ontology_namespace.clone(),
            qualifier_ontology_id: restriction.qualifier_ontology_id.clone(),
            qualifier_text: restriction.qualifier_text.clone(),
            qualifier_date: restriction.qualifier_date,
            qualifier_integer: restriction.qualifier_integer,
        })
    }

    async fn save_study_duo_restriction(
        &self,
        restriction: &StudyDuoRestrictionRecord,
    ) -> Result<StudyDuoRestrictionRecord, CoreError> {
        let input = pg_create_study_duo_restriction(
            Some(restriction.study_duo_restriction_id),
            &NewStudyDuoRestrictionRecord {
                study_id: restriction.study_id,
                duo_id: restriction.duo_id.clone(),
                qualifier_ontology_namespace: restriction.qualifier_ontology_namespace.clone(),
                qualifier_ontology_id: restriction.qualifier_ontology_id.clone(),
                qualifier_text: restriction.qualifier_text.clone(),
                qualifier_date: restriction.qualifier_date,
                qualifier_integer: restriction.qualifier_integer,
            },
        );
        self.with_session_connection(
            "save study DUO restriction",
            |connection| connection.create_study_duo_restriction(input.clone()),
            |connection| oauth_create_study_duo_restriction(connection, input.clone()),
        )?;
        Ok(restriction.clone())
    }

    async fn replace_study_duo_restrictions(
        &self,
        study_id: StudyId,
        restrictions: &[NewStudyDuoRestrictionRecord],
    ) -> Result<Vec<StudyDuoRestrictionRecord>, CoreError> {
        let restrictions = self.with_session_connection(
            "replace study DUO restrictions",
            |connection| {
                replace_study_duo_restrictions_with_executor(connection, study_id, restrictions)?;
                connection.list_study_duo_restrictions(study_id)
            },
            |connection| {
                replace_study_duo_restrictions_with_executor(connection, study_id, restrictions)?;
                oauth_list_study_duo_restrictions(connection, study_id)
            },
        )?;
        restrictions
            .into_iter()
            .map(study_duo_restriction_record)
            .collect()
    }
}

#[async_trait::async_trait]
impl AssetRepository for PgMetadataRepository<'_> {
    async fn asset_classification(
        &self,
        asset: AssetId,
    ) -> Result<Option<ahri_tre_types::AssetClassification>, CoreError> {
        self.with_session_connection(
            "read Asset classification",
            |db| crate::governance::classification(db, asset.0),
            |db| crate::governance::classification(db, asset.0),
        )
    }
    async fn validate_dataset_derivation(
        &self,
        inputs: &[VersionId],
        risk: ahri_tre_types::AssetRisk,
        inherit: bool,
    ) -> Result<(), CoreError> {
        let inputs = inputs.iter().map(|id| id.0).collect::<Vec<_>>();
        self.with_session_connection(
            "validate Dataset derivation",
            |db| crate::governance::validate_derivation(db, &inputs, risk, inherit),
            |db| crate::governance::validate_derivation(db, &inputs, risk, inherit),
        )
    }
    async fn list_assets_for_study(
        &self,
        study_id: StudyId,
    ) -> Result<Vec<AssetRecord>, CoreError> {
        self.with_session_connection(
            "list assets for study",
            |connection| connection.list_assets_for_study(study_id),
            |connection| oauth_list_assets_for_study(connection, study_id),
        )?
        .into_iter()
        .map(asset_record)
        .collect()
    }

    async fn get_asset_by_id(&self, asset_id: AssetId) -> Result<Option<AssetRecord>, CoreError> {
        self.with_session_connection(
            "get asset by id",
            |connection| connection.get_asset(asset_id),
            |connection| oauth_get_asset(connection, asset_id),
        )?
        .map(asset_record)
        .transpose()
    }

    async fn get_asset_by_name(
        &self,
        study_id: StudyId,
        name: &str,
    ) -> Result<Option<AssetRecord>, CoreError> {
        self.with_session_connection(
            "get asset by name",
            |connection| connection.get_asset_by_name(study_id, name),
            |connection| oauth_get_asset_by_name(connection, study_id, name),
        )?
        .map(asset_record)
        .transpose()
    }

    async fn create_asset(&self, asset: &NewAssetRecord) -> Result<AssetRecord, CoreError> {
        let input = PgCreateAsset {
            asset_id: None,
            study_id: asset.study_id,
            name: asset.name.as_str().to_string(),
            description: asset.description.clone(),
            agent_instructions: asset.agent_instructions.clone(),
            asset_type: asset.asset_type,
            date_created: asset.date_created,
            created_by: asset.created_by.clone(),
        };
        let asset_id = self.with_session_connection(
            "create asset",
            |connection| connection.create_asset(input.clone()),
            |connection| oauth_create_asset(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created asset",
            |connection| connection.get_asset(asset_id),
            |connection| oauth_get_asset(connection, asset_id),
        )?
        .map(asset_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!("created asset {:?} was not readable", asset_id))
        })
    }

    async fn save_asset(&self, asset: &AssetRecord) -> Result<AssetRecord, CoreError> {
        let input = PgCreateAsset {
            asset_id: Some(asset.asset_id),
            study_id: asset.study_id,
            name: asset.name.as_str().to_string(),
            description: asset.description.clone(),
            agent_instructions: asset.agent_instructions.clone(),
            asset_type: asset.asset_type,
            date_created: asset.date_created,
            created_by: asset.created_by.clone(),
        };
        let asset_id = self.with_session_connection(
            "create asset",
            |connection| connection.create_asset(input.clone()),
            |connection| oauth_create_asset(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created asset",
            |connection| connection.get_asset(asset_id),
            |connection| oauth_get_asset(connection, asset_id),
        )?
        .map(asset_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!("created asset {:?} was not readable", asset_id))
        })
    }

    async fn delete_asset_record(&self, asset_id: AssetId) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete asset record",
            |connection| delete_asset_metadata_record(connection, asset_id),
            |connection| delete_asset_metadata_record(connection, asset_id),
        )
    }

    async fn list_asset_versions(
        &self,
        asset_id: AssetId,
    ) -> Result<Vec<AssetVersionRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list asset versions",
                |connection| connection.list_asset_versions(asset_id),
                |connection| oauth_list_asset_versions(connection, asset_id),
            )?
            .into_iter()
            .map(asset_version_record)
            .collect())
    }

    async fn get_asset_version(
        &self,
        version_id: VersionId,
    ) -> Result<Option<AssetVersionRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "get asset version",
                |connection| connection.get_asset_version(version_id),
                |connection| oauth_get_asset_version(connection, version_id),
            )?
            .map(asset_version_record))
    }

    async fn delete_datafile_version_metadata(
        &self,
        version_id: VersionId,
        latest_promotion: Option<(AssetId, Option<VersionId>)>,
    ) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete Datafile version metadata",
            |connection| delete_version_metadata(connection, version_id, latest_promotion),
            |connection| delete_version_metadata(connection, version_id, latest_promotion),
        )
    }

    async fn delete_asset_version_record(&self, version_id: VersionId) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete asset version record",
            |connection| delete_asset_version_metadata_record(connection, version_id),
            |connection| delete_asset_version_metadata_record(connection, version_id),
        )
    }

    async fn create_asset_version(
        &self,
        version: &NewAssetVersionRecord,
    ) -> Result<AssetVersionRecord, CoreError> {
        let input = PgCreateAssetVersion {
            classification: version.classification.clone(),
            version_id: None,
            asset_id: version.asset_id,
            major: Some(version.major),
            minor: Some(version.minor),
            patch: Some(version.patch),
            version_note: version.version_note.clone(),
            is_latest: version.is_latest,
            doi: version.doi.clone(),
            created_at: version.created_at,
            created_by: version.created_by.clone(),
        };
        let version_id = self.with_session_connection(
            "create asset version",
            |connection| connection.create_asset_version(input.clone()),
            |connection| oauth_create_asset_version(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created asset version",
            |connection| connection.get_asset_version(version_id),
            |connection| oauth_get_asset_version(connection, version_id),
        )?
        .map(asset_version_record)
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created asset version {:?} was not readable",
                version_id
            ))
        })
    }

    async fn save_asset_version(
        &self,
        _version: &AssetVersionRecord,
    ) -> Result<AssetVersionRecord, CoreError> {
        Err(CoreError::Validation(
            "Explicit ingest classification is required".into(),
        ))
    }

    async fn admit_asset_version(
        &self,
        version: &AssetVersionRecord,
        classification: &ahri_tre_types::IngestClassification,
    ) -> Result<AssetVersionRecord, CoreError> {
        let input = PgCreateAssetVersion {
            classification: classification.clone(),
            version_id: Some(version.version_id),
            asset_id: version.asset_id,
            major: Some(version.major),
            minor: Some(version.minor),
            patch: Some(version.patch),
            version_note: version.version_note.clone(),
            is_latest: version.is_latest,
            doi: version.doi.clone(),
            created_at: version.created_at,
            created_by: version.created_by.clone(),
        };
        let version_id = self.with_session_connection(
            "create asset version",
            |connection| connection.create_asset_version(input.clone()),
            |connection| oauth_create_asset_version(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created asset version",
            |connection| connection.get_asset_version(version_id),
            |connection| oauth_get_asset_version(connection, version_id),
        )?
        .map(asset_version_record)
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created asset version {:?} was not readable",
                version_id
            ))
        })
    }

    async fn mark_asset_version_latest(
        &self,
        asset_id: AssetId,
        version_id: VersionId,
    ) -> Result<AssetVersionRecord, CoreError> {
        self.with_session_connection(
            "mark asset version latest",
            |connection| connection.with_transaction("promote admitted Dataset", |transaction| {
                transaction.execute_command(
                    "clear latest asset version",
                    "UPDATE public.asset_versions SET is_latest = FALSE WHERE asset_id = $1 AND is_latest = TRUE",
                    &[&asset_id.0],
                )?;
                transaction.query_one(
                    "mark asset version latest",
                    "UPDATE public.asset_versions SET is_latest = TRUE WHERE asset_id = $1 AND version_id = $2 RETURNING version_id",
                    &[&asset_id.0, &version_id.0],
                )?;
                Ok(())
            }),
            |connection| oauth_mark_asset_version_latest(connection, asset_id, version_id),
        )?;

        self.get_asset_version(version_id).await?.ok_or_else(|| {
            CoreError::NotFound(format!(
                "marked asset version {:?} was not readable",
                version_id
            ))
        })
    }

    async fn list_asset_version_duo_restrictions(
        &self,
        version_id: VersionId,
    ) -> Result<Vec<AssetVersionDuoRestrictionRecord>, CoreError> {
        self.with_session_connection(
            "list asset version DUO restrictions",
            |connection| connection.list_asset_version_duo_restrictions(version_id),
            |connection| oauth_list_asset_version_duo_restrictions(connection, version_id),
        )?
        .into_iter()
        .map(asset_version_duo_restriction_record)
        .collect()
    }

    async fn create_asset_version_duo_restriction(
        &self,
        restriction: &NewAssetVersionDuoRestrictionRecord,
    ) -> Result<AssetVersionDuoRestrictionRecord, CoreError> {
        let input = pg_create_asset_version_duo_restriction(None, restriction);
        let restriction_id = self.with_session_connection(
            "create asset version DUO restriction",
            |connection| connection.create_asset_version_duo_restriction(input.clone()),
            |connection| oauth_create_asset_version_duo_restriction(connection, input.clone()),
        )?;
        Ok(AssetVersionDuoRestrictionRecord {
            asset_version_duo_restriction_id: AssetVersionDuoRestrictionId(restriction_id),
            version_id: restriction.version_id,
            duo_id: restriction.duo_id.clone(),
            qualifier_ontology_namespace: restriction.qualifier_ontology_namespace.clone(),
            qualifier_ontology_id: restriction.qualifier_ontology_id.clone(),
            qualifier_text: restriction.qualifier_text.clone(),
            qualifier_date: restriction.qualifier_date,
            qualifier_integer: restriction.qualifier_integer,
        })
    }

    async fn save_asset_version_duo_restriction(
        &self,
        restriction: &AssetVersionDuoRestrictionRecord,
    ) -> Result<AssetVersionDuoRestrictionRecord, CoreError> {
        let input = pg_create_asset_version_duo_restriction(
            Some(restriction.asset_version_duo_restriction_id),
            &NewAssetVersionDuoRestrictionRecord {
                version_id: restriction.version_id,
                duo_id: restriction.duo_id.clone(),
                qualifier_ontology_namespace: restriction.qualifier_ontology_namespace.clone(),
                qualifier_ontology_id: restriction.qualifier_ontology_id.clone(),
                qualifier_text: restriction.qualifier_text.clone(),
                qualifier_date: restriction.qualifier_date,
                qualifier_integer: restriction.qualifier_integer,
            },
        );
        self.with_session_connection(
            "save asset version DUO restriction",
            |connection| connection.create_asset_version_duo_restriction(input.clone()),
            |connection| oauth_create_asset_version_duo_restriction(connection, input.clone()),
        )?;
        Ok(restriction.clone())
    }

    async fn replace_asset_version_duo_restrictions(
        &self,
        version_id: VersionId,
        restrictions: &[NewAssetVersionDuoRestrictionRecord],
    ) -> Result<Vec<AssetVersionDuoRestrictionRecord>, CoreError> {
        let restrictions = self.with_session_connection(
            "replace asset version DUO restrictions",
            |connection| {
                replace_asset_version_duo_restrictions_with_executor(
                    connection,
                    version_id,
                    restrictions,
                )?;
                connection.list_asset_version_duo_restrictions(version_id)
            },
            |connection| {
                replace_asset_version_duo_restrictions_with_executor(
                    connection,
                    version_id,
                    restrictions,
                )?;
                oauth_list_asset_version_duo_restrictions(connection, version_id)
            },
        )?;
        restrictions
            .into_iter()
            .map(asset_version_duo_restriction_record)
            .collect()
    }

    async fn clear_asset_version_duo_restrictions(
        &self,
        version_id: VersionId,
    ) -> Result<Vec<AssetVersionDuoRestrictionRecord>, CoreError> {
        let restrictions = self.with_session_connection(
            "clear asset version DUO restrictions",
            |connection| {
                clear_asset_version_duo_restrictions_with_executor(connection, version_id)?;
                connection.list_asset_version_duo_restrictions(version_id)
            },
            |connection| {
                clear_asset_version_duo_restrictions_with_executor(connection, version_id)?;
                oauth_list_asset_version_duo_restrictions(connection, version_id)
            },
        )?;
        restrictions
            .into_iter()
            .map(asset_version_duo_restriction_record)
            .collect()
    }

    async fn list_datasets_for_asset(
        &self,
        asset_id: AssetId,
    ) -> Result<Vec<DatasetRecord>, CoreError> {
        self.with_session_connection(
            "list datasets for asset",
            |connection| connection.list_datasets_for_asset(asset_id),
            |connection| oauth_list_datasets_for_asset(connection, asset_id),
        )?
        .into_iter()
        .map(dataset_record)
        .collect()
    }

    async fn search_datasets(
        &self,
        query: &DatasetSearchQuery,
    ) -> Result<SearchPage<DatasetSearchRecord, DatasetSearchCursorAnchor>, CoreError> {
        let studies = <Self as StudyRepository>::list_studies(self).await?;
        let mut matched = Vec::new();
        for study in studies {
            let domains =
                <Self as StudyDomainRepository>::list_domains_for_study(self, study.study_id)
                    .await?;
            for asset in <Self as AssetRepository>::list_assets_for_study(self, study.study_id)
                .await?
                .into_iter()
                .filter(|asset| asset.asset_type == AssetType::Dataset)
            {
                let versions =
                    <Self as AssetRepository>::list_asset_versions(self, asset.asset_id).await?;
                let dataset_ids =
                    <Self as AssetRepository>::list_datasets_for_asset(self, asset.asset_id)
                        .await?
                        .into_iter()
                        .map(|dataset| dataset.dataset_id.0)
                        .collect::<Vec<_>>();
                let withdrawn_ids = <Self as AssetRepository>::list_dataset_version_withdrawals(
                    self,
                    asset.asset_id,
                )
                .await?
                .into_iter()
                .map(|withdrawal| withdrawal.dataset_id.0)
                .collect::<Vec<_>>();
                let Some(latest_version) =
                    latest_searchable_dataset_version(&versions, &dataset_ids, &withdrawn_ids)
                else {
                    continue;
                };
                let tags = <Self as TagRepository>::list_tags_for_target(
                    self,
                    TagAttachmentTarget::AssetVersion(latest_version.version_id),
                )
                .await?;
                if dataset_matches_search_query(
                    &study,
                    &asset,
                    &latest_version,
                    &domains,
                    &tags,
                    query,
                ) {
                    matched.push(DatasetSearchRecord {
                        study_id: study.study_id,
                        asset,
                        latest_version,
                    });
                }
            }
        }

        matched.sort_by(dataset_search_sort_key);
        if let Some(after) = query.after.as_ref() {
            matched.retain(|dataset| dataset_sort_tuple(dataset) > dataset_cursor_tuple(after));
        }

        let limit = usize::from(query.limit);
        let has_more = matched.len() > limit;
        if has_more {
            matched.truncate(limit);
        }
        let next_cursor = if has_more {
            matched.last().map(|dataset| DatasetSearchCursorAnchor {
                name: dataset.asset.name.as_str().to_string(),
                study_id: dataset.study_id,
                asset_id: dataset.asset.asset_id,
            })
        } else {
            None
        };

        Ok(SearchPage {
            items: matched,
            next_cursor,
        })
    }

    async fn search_datafiles(
        &self,
        query: &DataFileSearchQuery,
    ) -> Result<SearchPage<DataFileSearchRecord, DataFileSearchCursorAnchor>, CoreError> {
        let studies = <Self as StudyRepository>::list_studies(self).await?;
        let mut matched = Vec::new();
        for study in studies {
            for asset in <Self as AssetRepository>::list_assets_for_study(self, study.study_id)
                .await?
                .into_iter()
                .filter(|asset| asset.asset_type == AssetType::File)
            {
                let versions =
                    <Self as AssetRepository>::list_asset_versions(self, asset.asset_id).await?;
                let datafiles =
                    <Self as AssetRepository>::list_datafiles_for_asset(self, asset.asset_id)
                        .await?;
                let Some((latest_version, datafile)) =
                    latest_searchable_datafile_version(&versions, &datafiles)
                else {
                    continue;
                };
                let tags = <Self as TagRepository>::list_tags_for_target(
                    self,
                    TagAttachmentTarget::AssetVersion(latest_version.version_id),
                )
                .await?;
                if datafile_matches_search_query(&study, &asset, &datafile, &tags, query) {
                    matched.push(DataFileSearchRecord {
                        study_id: study.study_id,
                        asset,
                        latest_version,
                        datafile,
                    });
                }
            }
        }

        matched.sort_by(datafile_search_sort_key);
        if let Some(after) = query.after.as_ref() {
            matched.retain(|datafile| datafile_sort_tuple(datafile) > datafile_cursor_tuple(after));
        }

        let limit = usize::from(query.limit);
        let has_more = matched.len() > limit;
        if has_more {
            matched.truncate(limit);
        }
        let next_cursor = if has_more {
            matched.last().map(|datafile| DataFileSearchCursorAnchor {
                name: datafile.asset.name.as_str().to_string(),
                study_id: datafile.study_id,
                asset_id: datafile.asset.asset_id,
            })
        } else {
            None
        };

        Ok(SearchPage {
            items: matched,
            next_cursor,
        })
    }

    async fn get_dataset(&self, dataset_id: VersionId) -> Result<Option<DatasetRecord>, CoreError> {
        self.with_session_connection(
            "get dataset",
            |connection| connection.get_dataset(dataset_id),
            |connection| oauth_get_dataset(connection, dataset_id),
        )?
        .map(dataset_record)
        .transpose()
    }

    async fn save_dataset(&self, dataset: &DatasetRecord) -> Result<DatasetRecord, CoreError> {
        self.with_session_connection(
            "create dataset",
            |connection| connection.create_dataset(dataset.dataset_id),
            |connection| oauth_create_dataset(connection, dataset.dataset_id),
        )?;

        self.with_session_connection(
            "read created dataset",
            |connection| connection.get_dataset(dataset.dataset_id),
            |connection| oauth_get_dataset(connection, dataset.dataset_id),
        )?
        .map(dataset_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created dataset {:?} was not readable",
                dataset.dataset_id
            ))
        })
    }

    async fn get_dataset_version_withdrawal(
        &self,
        dataset_id: VersionId,
    ) -> Result<Option<DatasetVersionWithdrawalRecord>, CoreError> {
        self.with_session_connection(
            "get dataset version withdrawal",
            |connection| {
                if !pg_dataset_version_withdrawals_table_exists(connection)? {
                    return Ok(None);
                }

                let row = connection
                    .client()
                    .query_opt(
                        "
                        SELECT dataset_id, asset_id, withdrawn_at, withdrawn_by, reason,
                               tombstone_state, provenance, drop_lake_table
                          FROM public.dataset_version_withdrawals
                         WHERE dataset_id = $1
                         LIMIT 1
                        ",
                        &[&dataset_id.0],
                    )
                    .map_err(PgMetaError::Query)?;
                Ok(row.map(|row| DatasetVersionWithdrawalRecord {
                    dataset_id: VersionId(row.get("dataset_id")),
                    asset_id: AssetId(row.get("asset_id")),
                    withdrawn_at: Some(DateTime::from_naive_utc_and_offset(
                        row.get::<_, NaiveDateTime>("withdrawn_at"),
                        Utc,
                    )),
                    withdrawn_by: row.get("withdrawn_by"),
                    reason: row.get("reason"),
                    tombstone_state: row.get("tombstone_state"),
                    provenance: row.get("provenance"),
                    drop_lake_table: row.get("drop_lake_table"),
                }))
            },
            |connection| oauth_get_dataset_version_withdrawal(connection, dataset_id),
        )
    }

    async fn list_dataset_version_withdrawals(
        &self,
        asset_id: AssetId,
    ) -> Result<Vec<DatasetVersionWithdrawalRecord>, CoreError> {
        self.with_session_connection(
            "list dataset version withdrawals",
            |connection| {
                if !pg_dataset_version_withdrawals_table_exists(connection)? {
                    return Ok(Vec::new());
                }

                let rows = connection
                    .client()
                    .query(
                        "
                        SELECT dataset_id, asset_id, withdrawn_at, withdrawn_by, reason,
                               tombstone_state, provenance, drop_lake_table
                          FROM public.dataset_version_withdrawals
                         WHERE asset_id = $1
                         ORDER BY withdrawn_at, dataset_id
                        ",
                        &[&asset_id.0],
                    )
                    .map_err(PgMetaError::Query)?;
                Ok(rows
                    .into_iter()
                    .map(|row| DatasetVersionWithdrawalRecord {
                        dataset_id: VersionId(row.get("dataset_id")),
                        asset_id: AssetId(row.get("asset_id")),
                        withdrawn_at: Some(DateTime::from_naive_utc_and_offset(
                            row.get::<_, NaiveDateTime>("withdrawn_at"),
                            Utc,
                        )),
                        withdrawn_by: row.get("withdrawn_by"),
                        reason: row.get("reason"),
                        tombstone_state: row.get("tombstone_state"),
                        provenance: row.get("provenance"),
                        drop_lake_table: row.get("drop_lake_table"),
                    })
                    .collect())
            },
            |connection| oauth_list_dataset_version_withdrawals(connection, asset_id),
        )
    }

    async fn save_dataset_version_withdrawal(
        &self,
        withdrawal: &NewDatasetVersionWithdrawalRecord,
    ) -> Result<DatasetVersionWithdrawalRecord, CoreError> {
        self.with_session_connection(
            "save dataset version withdrawal",
            |connection| {
                connection
                    .client()
                    .execute(
                        "
                        INSERT INTO public.dataset_version_withdrawals (
                            dataset_id, asset_id, withdrawn_by, reason, tombstone_state,
                            provenance, drop_lake_table
                        )
                        VALUES ($1, $2, COALESCE($3, CURRENT_USER), $4, $5, $6, $7)
                        ON CONFLICT (dataset_id) DO NOTHING
                        ",
                        &[
                            &withdrawal.dataset_id.0,
                            &withdrawal.asset_id.0,
                            &withdrawal.withdrawn_by,
                            &withdrawal.reason,
                            &withdrawal.tombstone_state,
                            &withdrawal.provenance,
                            &withdrawal.drop_lake_table,
                        ],
                    )
                    .map_err(|source| PgMetaError::Write {
                        operation: "save dataset version withdrawal",
                        source,
                    })?;
                Ok(())
            },
            |connection| oauth_save_dataset_version_withdrawal(connection, withdrawal),
        )?;

        self.get_dataset_version_withdrawal(withdrawal.dataset_id)
            .await?
            .ok_or_else(|| {
                CoreError::NotFound(format!(
                    "dataset version withdrawal was not readable: {}",
                    withdrawal.dataset_id.0
                ))
            })
    }

    async fn deactivate_dataset_version_links(
        &self,
        dataset_id: VersionId,
    ) -> Result<(), CoreError> {
        self.with_session_connection(
            "deactivate dataset version links",
            |connection| {
                let mut transaction = connection.client().transaction().map_err(PgMetaError::Transaction)?;
                for (operation, sql) in [
                    (
                        "delete dataset version relation links",
                        "DELETE FROM public.dataset_version_relation_instances WHERE version_id = $1",
                    ),
                    (
                        "delete dataset version entity links",
                        "DELETE FROM public.dataset_version_entities WHERE version_id = $1",
                    ),
                    (
                        "delete asset version relation links",
                        "DELETE FROM public.asset_version_relation_instances WHERE version_id = $1",
                    ),
                    (
                        "delete asset version entity links",
                        "DELETE FROM public.asset_version_entities WHERE version_id = $1",
                    ),
                    (
                        "delete dataset variable links",
                        "DELETE FROM public.dataset_variables WHERE dataset_id = $1",
                    ),
                ] {
                    transaction.execute(sql, &[&dataset_id.0]).map_err(|source| PgMetaError::Write {
                        operation,
                        source,
                    })?;
                }
                transaction.commit().map_err(PgMetaError::Transaction)?;
                Ok(())
            },
            |connection| oauth_deactivate_dataset_version_links(connection, dataset_id),
        )
    }

    async fn clear_asset_version_latest(
        &self,
        asset_id: AssetId,
        version_id: VersionId,
    ) -> Result<AssetVersionRecord, CoreError> {
        self.with_session_connection(
            "clear asset version latest",
            |connection| {
                connection
                    .client()
                    .query_one(
                        "UPDATE public.asset_versions SET is_latest = FALSE WHERE asset_id = $1 AND version_id = $2 RETURNING version_id",
                        &[&asset_id.0, &version_id.0],
                    )
                    .map_err(|source| PgMetaError::Write {
                        operation: "clear asset version latest",
                        source,
                    })?;
                Ok(())
            },
            |connection| oauth_clear_asset_version_latest(connection, asset_id, version_id),
        )?;

        self.get_asset_version(version_id).await?.ok_or_else(|| {
            CoreError::NotFound(format!(
                "cleared asset version {:?} was not readable",
                version_id
            ))
        })
    }

    async fn list_datafiles_for_asset(
        &self,
        asset_id: AssetId,
    ) -> Result<Vec<DataFileRecord>, CoreError> {
        self.with_session_connection(
            "list datafiles for asset",
            |connection| connection.list_datafiles_for_asset(asset_id),
            |connection| oauth_list_datafiles_for_asset(connection, asset_id),
        )?
        .into_iter()
        .map(datafile_record)
        .collect()
    }

    async fn get_datafile(
        &self,
        datafile_id: VersionId,
    ) -> Result<Option<DataFileRecord>, CoreError> {
        self.with_session_connection(
            "get datafile",
            |connection| connection.get_datafile(datafile_id),
            |connection| oauth_get_datafile(connection, datafile_id),
        )?
        .map(datafile_record)
        .transpose()
    }

    async fn save_datafile(&self, datafile: &DataFileRecord) -> Result<DataFileRecord, CoreError> {
        ahri_tre_core::validate_datafile_record(datafile)?;
        let input = PgCreateDataFile {
            datafile_id: datafile.datafile_id,
            size_bytes: datafile
                .size_bytes
                .map(i64::try_from)
                .transpose()
                .map_err(|_| {
                    CoreError::Validation("datafile size exceeds PostgreSQL BIGINT".to_string())
                })?,
            compressed: datafile.compressed,
            encrypted: datafile.encrypted,
            compression_algorithm: datafile.compression_algorithm.clone(),
            encryption_algorithm: datafile.encryption_algorithm.clone(),
            encryption_key: datafile.encryption_key.clone(),
            storage_uri: datafile.storage_uri.clone(),
            edam_format: datafile.edam_format.clone(),
            digest: datafile.digest.as_str().to_string(),
        };
        self.with_session_connection(
            "create datafile",
            |connection| connection.create_datafile(input.clone()),
            |connection| oauth_create_datafile(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created datafile",
            |connection| connection.get_datafile(datafile.datafile_id),
            |connection| oauth_get_datafile(connection, datafile.datafile_id),
        )?
        .map(datafile_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created datafile {:?} was not readable",
                datafile.datafile_id
            ))
        })
    }

    async fn delete_datafile_record(&self, datafile_id: VersionId) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete datafile record",
            |connection| delete_datafile_metadata_record(connection, datafile_id),
            |connection| delete_datafile_metadata_record(connection, datafile_id),
        )
    }
}

#[async_trait::async_trait]
impl VariableRepository for PgMetadataRepository<'_> {
    async fn list_value_types(&self) -> Result<Vec<ValueTypeRecord>, CoreError> {
        self.with_session_connection(
            "list value types",
            |connection| connection.list_value_types(),
            oauth_list_value_types,
        )?
        .into_iter()
        .map(value_type_record)
        .collect()
    }

    async fn list_domain_variables(
        &self,
        domain_id: DomainId,
    ) -> Result<Vec<VariableRecord>, CoreError> {
        self.with_session_connection(
            "list domain variables",
            |connection| connection.list_variables_for_domain(domain_id),
            |connection| oauth_list_variables_for_domain(connection, domain_id),
        )?
        .into_iter()
        .map(variable_record)
        .collect()
    }

    async fn get_variable(
        &self,
        variable_id: VariableId,
    ) -> Result<Option<VariableRecord>, CoreError> {
        self.with_session_connection(
            "get variable",
            |connection| connection.get_variable(variable_id),
            |connection| oauth_get_variable(connection, variable_id),
        )?
        .map(variable_record)
        .transpose()
    }

    async fn create_variable(
        &self,
        variable: &NewVariableRecord,
    ) -> Result<VariableRecord, CoreError> {
        let input = PgCreateVariable {
            variable_id: None,
            domain_id: variable.domain_id,
            name: variable.name.as_str().to_string(),
            value_type_id: variable.value_type_id,
            value_format: variable.value_format.clone(),
            vocabulary_id: variable.vocabulary_id,
            key_role: variable.key_role,
            description: variable.description.clone(),
            note: variable.note.clone(),
            ontology_namespace: variable.ontology_namespace.clone(),
            ontology_class: variable.ontology_class.clone(),
        };
        let variable_id = self.with_session_connection(
            "create variable",
            |connection| connection.create_variable(input.clone()),
            |connection| oauth_create_variable(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created variable",
            |connection| connection.list_variables_for_domain(variable.domain_id),
            |connection| oauth_list_variables_for_domain(connection, variable.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.variable_id == variable_id)
        .map(variable_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created variable {:?} was not readable",
                variable_id
            ))
        })
    }

    async fn save_variable(&self, variable: &VariableRecord) -> Result<VariableRecord, CoreError> {
        let input = PgCreateVariable {
            variable_id: Some(variable.variable_id),
            domain_id: variable.domain_id,
            name: variable.name.as_str().to_string(),
            value_type_id: variable.value_type_id,
            value_format: variable.value_format.clone(),
            vocabulary_id: variable.vocabulary_id,
            key_role: variable.key_role,
            description: variable.description.clone(),
            note: variable.note.clone(),
            ontology_namespace: variable.ontology_namespace.clone(),
            ontology_class: variable.ontology_class.clone(),
        };
        let variable_id = self.with_session_connection(
            "create variable",
            |connection| connection.create_variable(input.clone()),
            |connection| oauth_create_variable(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created variable",
            |connection| connection.list_variables_for_domain(variable.domain_id),
            |connection| oauth_list_variables_for_domain(connection, variable.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.variable_id == variable_id)
        .map(variable_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created variable {:?} was not readable",
                variable_id
            ))
        })
    }

    async fn update_variable_metadata(
        &self,
        variable: &VariableRecord,
    ) -> Result<VariableRecord, CoreError> {
        let input = PgCreateVariable {
            variable_id: Some(variable.variable_id),
            domain_id: variable.domain_id,
            name: variable.name.as_str().to_string(),
            value_type_id: variable.value_type_id,
            value_format: variable.value_format.clone(),
            vocabulary_id: variable.vocabulary_id,
            key_role: variable.key_role,
            description: variable.description.clone(),
            note: variable.note.clone(),
            ontology_namespace: variable.ontology_namespace.clone(),
            ontology_class: variable.ontology_class.clone(),
        };
        self.with_session_connection(
            "update variable",
            |connection| connection.update_variable(input.clone()),
            |connection| oauth_update_variable(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read updated variable",
            |connection| connection.list_variables_for_domain(variable.domain_id),
            |connection| oauth_list_variables_for_domain(connection, variable.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.variable_id == variable.variable_id)
        .map(variable_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "updated variable {:?} was not readable",
                variable.variable_id
            ))
        })
    }

    async fn link_dataset_variable(
        &self,
        dataset_id: VersionId,
        variable_id: VariableId,
        row_role: KeyRole,
    ) -> Result<ahri_tre_types::DatasetVariableLinkRecord, CoreError> {
        self.with_session_connection(
            "link dataset variable",
            |connection| connection.link_dataset_variable(dataset_id, variable_id, row_role),
            |connection| oauth_link_dataset_variable(connection, dataset_id, variable_id, row_role),
        )?;
        Ok(ahri_tre_types::DatasetVariableLinkRecord {
            dataset_id,
            variable_id,
            row_role,
        })
    }

    async fn list_dataset_variables(
        &self,
        dataset_id: VersionId,
    ) -> Result<Vec<ahri_tre_types::DatasetVariableLinkRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list dataset variables",
                |connection| connection.list_dataset_variables(dataset_id),
                |connection| oauth_list_dataset_variables(connection, dataset_id),
            )?
            .into_iter()
            .map(dataset_variable_link_record)
            .collect())
    }

    async fn list_dataset_variables_for_variable(
        &self,
        variable_id: VariableId,
    ) -> Result<Vec<ahri_tre_types::DatasetVariableLinkRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list dataset variables for variable",
                |connection| connection.list_dataset_variables_for_variable(variable_id),
                |connection| oauth_list_dataset_variables_for_variable(connection, variable_id),
            )?
            .into_iter()
            .map(dataset_variable_link_record)
            .collect())
    }

    async fn delete_variable_record(&self, variable_id: VariableId) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete variable record",
            |connection| {
                delete_semantic_record(connection, "variables", "variable_id", variable_id.0)
            },
            |connection| {
                delete_semantic_record(connection, "variables", "variable_id", variable_id.0)
            },
        )
    }

    async fn search_variables(
        &self,
        query: &VariableSearchQuery,
    ) -> Result<SearchPage<VariableSearchRecord, VariableSearchCursorAnchor>, CoreError> {
        let domains = self.list_domains().await?;
        let value_types = self.list_value_types().await?;
        let mut matched = Vec::new();
        for domain in domains {
            let vocabularies = self.list_vocabularies(domain.domain_id).await?;
            for variable in self.list_domain_variables(domain.domain_id).await? {
                let value_type = value_types
                    .iter()
                    .find(|value_type| value_type.value_type_id == variable.value_type_id);
                let vocabulary = variable.vocabulary_id.and_then(|vocabulary_id| {
                    vocabularies
                        .iter()
                        .find(|vocabulary| vocabulary.vocabulary_id == vocabulary_id)
                });
                let tags = self
                    .list_tags_for_target(TagAttachmentTarget::Variable(variable.variable_id))
                    .await?;
                if variable_matches_search_query(
                    query, &domain, &variable, value_type, vocabulary, &tags,
                ) {
                    matched.push(VariableSearchRecord { variable });
                }
            }
        }

        matched.sort_by(variable_search_sort_key);
        let start = query
            .after
            .as_ref()
            .and_then(|anchor| {
                matched.iter().position(|variable| {
                    variable_sort_tuple(variable) > variable_cursor_tuple(anchor)
                })
            })
            .unwrap_or(0);
        let limit = usize::from(query.limit);
        let mut items = matched
            .into_iter()
            .skip(start)
            .take(limit + 1)
            .collect::<Vec<_>>();
        let next_cursor = if items.len() > limit {
            items.truncate(limit);
            items.last().map(|variable| VariableSearchCursorAnchor {
                name: variable.variable.name.as_str().to_string(),
                domain_id: variable.variable.domain_id,
                variable_id: variable.variable.variable_id,
            })
        } else {
            None
        };
        Ok(SearchPage { items, next_cursor })
    }
}

#[async_trait::async_trait]
impl VocabularyRepository for PgMetadataRepository<'_> {
    async fn list_vocabularies(
        &self,
        domain_id: DomainId,
    ) -> Result<Vec<VocabularyRecord>, CoreError> {
        self.with_session_connection(
            "list vocabularies",
            |connection| connection.list_vocabularies_for_domain(domain_id),
            |connection| oauth_list_vocabularies_for_domain(connection, domain_id),
        )?
        .into_iter()
        .map(vocabulary_record)
        .collect()
    }

    async fn get_vocabulary(
        &self,
        vocabulary_id: VocabularyId,
    ) -> Result<Option<VocabularyRecord>, CoreError> {
        self.with_session_connection(
            "get vocabulary",
            |connection| connection.get_vocabulary(vocabulary_id),
            |connection| oauth_get_vocabulary(connection, vocabulary_id),
        )?
        .map(vocabulary_record)
        .transpose()
    }

    async fn list_vocabulary_items(
        &self,
        vocabulary_id: VocabularyId,
    ) -> Result<Vec<VocabularyItemRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list vocabulary items",
                |connection| connection.list_vocabulary_items(vocabulary_id),
                |connection| oauth_list_vocabulary_items(connection, vocabulary_id),
            )?
            .into_iter()
            .map(vocabulary_item_record)
            .collect())
    }

    async fn get_vocabulary_item(
        &self,
        vocabulary_item_id: VocabularyItemId,
    ) -> Result<Option<VocabularyItemRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "get vocabulary item",
                |connection| connection.get_vocabulary_item(vocabulary_item_id),
                |connection| oauth_get_vocabulary_item(connection, vocabulary_item_id),
            )?
            .map(vocabulary_item_record))
    }

    async fn create_vocabulary(
        &self,
        vocabulary: &NewVocabularyRecord,
    ) -> Result<VocabularyRecord, CoreError> {
        let input = PgCreateVocabulary {
            vocabulary_id: None,
            domain_id: vocabulary.domain_id,
            name: vocabulary.name.as_str().to_string(),
            description: vocabulary.description.clone(),
        };
        let vocabulary_id = self.with_session_connection(
            "create vocabulary",
            |connection| connection.create_vocabulary(input.clone()),
            |connection| oauth_create_vocabulary(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created vocabulary",
            |connection| connection.list_vocabularies_for_domain(vocabulary.domain_id),
            |connection| oauth_list_vocabularies_for_domain(connection, vocabulary.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.vocabulary_id == vocabulary_id)
        .map(vocabulary_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created vocabulary {:?} was not readable",
                vocabulary_id
            ))
        })
    }

    async fn save_vocabulary(
        &self,
        vocabulary: &VocabularyRecord,
    ) -> Result<VocabularyRecord, CoreError> {
        let input = PgCreateVocabulary {
            vocabulary_id: Some(vocabulary.vocabulary_id),
            domain_id: vocabulary.domain_id,
            name: vocabulary.name.as_str().to_string(),
            description: vocabulary.description.clone(),
        };
        let vocabulary_id = self.with_session_connection(
            "create vocabulary",
            |connection| connection.create_vocabulary(input.clone()),
            |connection| oauth_create_vocabulary(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read created vocabulary",
            |connection| connection.list_vocabularies_for_domain(vocabulary.domain_id),
            |connection| oauth_list_vocabularies_for_domain(connection, vocabulary.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.vocabulary_id == vocabulary_id)
        .map(vocabulary_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created vocabulary {:?} was not readable",
                vocabulary_id
            ))
        })
    }

    async fn update_vocabulary(
        &self,
        vocabulary: &VocabularyRecord,
    ) -> Result<VocabularyRecord, CoreError> {
        let input = PgCreateVocabulary {
            vocabulary_id: Some(vocabulary.vocabulary_id),
            domain_id: vocabulary.domain_id,
            name: vocabulary.name.as_str().to_string(),
            description: vocabulary.description.clone(),
        };
        self.with_session_connection(
            "update vocabulary",
            |connection| connection.update_vocabulary(input.clone()),
            |connection| oauth_update_vocabulary(connection, input.clone()),
        )?;

        self.with_session_connection(
            "read updated vocabulary",
            |connection| connection.list_vocabularies_for_domain(vocabulary.domain_id),
            |connection| oauth_list_vocabularies_for_domain(connection, vocabulary.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.vocabulary_id == vocabulary.vocabulary_id)
        .map(vocabulary_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "updated vocabulary {:?} was not readable",
                vocabulary.vocabulary_id
            ))
        })
    }

    async fn create_vocabulary_items(
        &self,
        vocabulary_id: VocabularyId,
        items: &[NewVocabularyItemRecord],
    ) -> Result<Vec<VocabularyItemRecord>, CoreError> {
        self.with_session_connection(
            "create vocabulary items",
            |connection| {
                let mut created = Vec::with_capacity(items.len());
                for item in items {
                    let vocabulary_item_id =
                        connection.create_vocabulary_item(PgCreateVocabularyItem {
                            vocabulary_item_id: None,
                            vocabulary_id,
                            value: item.value,
                            code: item.code.clone(),
                            description: item.description.clone(),
                        })?;
                    created.push(VocabularyItemRecord {
                        vocabulary_item_id,
                        vocabulary_id,
                        value: item.value,
                        code: item.code.clone(),
                        description: item.description.clone(),
                    });
                }
                Ok(created)
            },
            |connection| oauth_create_vocabulary_items(connection, vocabulary_id, items),
        )
    }

    async fn save_vocabulary_items(&self, items: &[VocabularyItemRecord]) -> Result<(), CoreError> {
        self.with_session_connection(
            "create vocabulary items",
            |connection| {
                for item in items {
                    connection.create_vocabulary_item(PgCreateVocabularyItem {
                        vocabulary_item_id: Some(item.vocabulary_item_id),
                        vocabulary_id: item.vocabulary_id,
                        value: item.value,
                        code: item.code.clone(),
                        description: item.description.clone(),
                    })?;
                }
                Ok(())
            },
            |connection| oauth_save_vocabulary_items(connection, items),
        )
    }

    async fn update_vocabulary_items(
        &self,
        items: &[VocabularyItemRecord],
    ) -> Result<(), CoreError> {
        self.with_session_connection(
            "update vocabulary items",
            |connection| {
                for item in items {
                    connection.update_vocabulary_item(PgCreateVocabularyItem {
                        vocabulary_item_id: Some(item.vocabulary_item_id),
                        vocabulary_id: item.vocabulary_id,
                        value: item.value,
                        code: item.code.clone(),
                        description: item.description.clone(),
                    })?;
                }
                Ok(())
            },
            |connection| oauth_update_vocabulary_items(connection, items),
        )
    }

    async fn list_vocabulary_mappings(&self) -> Result<Vec<VocabularyMappingRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list vocabulary mappings",
                |connection| connection.list_vocabulary_mappings(),
                oauth_list_vocabulary_mappings,
            )?
            .into_iter()
            .map(vocabulary_mapping_record)
            .collect())
    }

    async fn save_vocabulary_mapping(
        &self,
        mapping: &VocabularyMappingRecord,
    ) -> Result<VocabularyMappingRecord, CoreError> {
        let input = PgCreateVocabularyMapping {
            vocabulary_mapping_id: Some(mapping.vocabulary_mapping_id),
            from_vocabulary_item: mapping.from_vocabulary_item,
            to_vocabulary_item: mapping.to_vocabulary_item,
        };
        let mapping_id = self.with_session_connection(
            "create vocabulary mapping",
            |connection| connection.create_vocabulary_mapping(input.clone()),
            |connection| oauth_create_vocabulary_mapping(connection, input.clone()),
        )?;
        Ok(VocabularyMappingRecord {
            vocabulary_mapping_id: mapping_id,
            from_vocabulary_item: mapping.from_vocabulary_item,
            to_vocabulary_item: mapping.to_vocabulary_item,
        })
    }

    async fn create_vocabulary_mapping(
        &self,
        mapping: &NewVocabularyMappingRecord,
    ) -> Result<VocabularyMappingRecord, CoreError> {
        let input = PgCreateVocabularyMapping {
            vocabulary_mapping_id: None,
            from_vocabulary_item: mapping.from_vocabulary_item,
            to_vocabulary_item: mapping.to_vocabulary_item,
        };
        let mapping_id = self.with_session_connection(
            "create vocabulary mapping",
            |connection| connection.create_vocabulary_mapping(input.clone()),
            |connection| oauth_create_vocabulary_mapping(connection, input.clone()),
        )?;
        Ok(VocabularyMappingRecord {
            vocabulary_mapping_id: mapping_id,
            from_vocabulary_item: mapping.from_vocabulary_item,
            to_vocabulary_item: mapping.to_vocabulary_item,
        })
    }

    async fn delete_vocabulary_record(
        &self,
        vocabulary_id: VocabularyId,
    ) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete vocabulary record",
            |connection| {
                delete_semantic_record(connection, "vocabularies", "vocabulary_id", vocabulary_id.0)
            },
            |connection| {
                delete_semantic_record(connection, "vocabularies", "vocabulary_id", vocabulary_id.0)
            },
        )
    }
}

#[async_trait::async_trait]
impl EntityRepository for PgMetadataRepository<'_> {
    async fn list_entities(&self, domain_id: DomainId) -> Result<Vec<EntityRecord>, CoreError> {
        self.with_session_connection(
            "list entities",
            |connection| connection.list_entities_for_domain(domain_id),
            |connection| oauth_list_entities_for_domain(connection, domain_id),
        )?
        .into_iter()
        .map(entity_record)
        .collect()
    }

    async fn list_entity_relations(
        &self,
        domain_id: DomainId,
    ) -> Result<Vec<EntityRelationRecord>, CoreError> {
        self.with_session_connection(
            "list entity relations",
            |connection| connection.list_entity_relations_for_domain(domain_id),
            |connection| oauth_list_entity_relations_for_domain(connection, domain_id),
        )?
        .into_iter()
        .map(entity_relation_record)
        .collect()
    }

    async fn search_entities(
        &self,
        query: &EntitySearchQuery,
    ) -> Result<SearchPage<EntitySearchRecord, EntitySearchCursorAnchor>, CoreError> {
        let domains = self.list_domains().await?;
        let mut matched = Vec::new();
        for domain in domains {
            for entity in self.list_entities(domain.domain_id).await? {
                let tags = self
                    .list_tags_for_target(TagAttachmentTarget::Entity(entity.entity_id))
                    .await?;
                if entity_matches_search_query(query, &domain, &entity, &tags) {
                    matched.push(EntitySearchRecord { entity });
                }
            }
        }

        matched.sort_by(entity_search_sort_key);
        let start = query
            .after
            .as_ref()
            .and_then(|anchor| {
                matched
                    .iter()
                    .position(|entity| entity_sort_tuple(entity) > entity_cursor_tuple(anchor))
            })
            .unwrap_or(0);
        let limit = usize::from(query.limit);
        let mut items = matched
            .into_iter()
            .skip(start)
            .take(limit + 1)
            .collect::<Vec<_>>();
        let next_cursor = if items.len() > limit {
            items.truncate(limit);
            items.last().map(|entity| EntitySearchCursorAnchor {
                name: entity.entity.name.as_str().to_string(),
                domain_id: entity.entity.domain_id,
                entity_id: entity.entity.entity_id,
            })
        } else {
            None
        };
        Ok(SearchPage { items, next_cursor })
    }

    async fn search_entity_relations(
        &self,
        query: &RelationSearchQuery,
    ) -> Result<SearchPage<RelationSearchRecord, RelationSearchCursorAnchor>, CoreError> {
        let domains = self.list_domains().await?;
        let mut matched = Vec::new();
        for domain in domains {
            let entities = self.list_entities(domain.domain_id).await?;
            let entities_by_id = entities
                .iter()
                .map(|entity| (entity.entity_id.0, entity.clone()))
                .collect::<BTreeMap<_, _>>();
            for relation in self.list_entity_relations(domain.domain_id).await? {
                let Some(source_entity) =
                    entities_by_id.get(&relation.subject_entity_id.0).cloned()
                else {
                    continue;
                };
                let Some(target_entity) = entities_by_id.get(&relation.object_entity_id.0).cloned()
                else {
                    continue;
                };
                let tags = self
                    .list_tags_for_target(TagAttachmentTarget::EntityRelation(
                        relation.entity_relation_id,
                    ))
                    .await?;
                if relation_matches_search_query(
                    query,
                    &domain,
                    &relation,
                    &source_entity,
                    &target_entity,
                    &tags,
                ) {
                    matched.push(RelationSearchRecord {
                        relation,
                        source_entity,
                        target_entity,
                    });
                }
            }
        }

        matched.sort_by(relation_search_sort_key);
        let start = query
            .after
            .as_ref()
            .and_then(|anchor| {
                matched.iter().position(|relation| {
                    relation_sort_tuple(relation) > relation_cursor_tuple(anchor)
                })
            })
            .unwrap_or(0);
        let limit = usize::from(query.limit);
        let mut items = matched
            .into_iter()
            .skip(start)
            .take(limit + 1)
            .collect::<Vec<_>>();
        let next_cursor = if items.len() > limit {
            items.truncate(limit);
            items.last().map(|relation| RelationSearchCursorAnchor {
                name: relation.relation.name.as_str().to_string(),
                domain_id: relation.relation.domain_id,
                relation_id: relation.relation.entity_relation_id,
            })
        } else {
            None
        };
        Ok(SearchPage { items, next_cursor })
    }

    async fn create_entity(&self, entity: &NewEntityRecord) -> Result<EntityRecord, CoreError> {
        let input = PgCreateEntity {
            entity_id: None,
            domain_id: entity.domain_id,
            uuid: None,
            name: entity.name.as_str().to_string(),
            description: entity.description.clone(),
            agent_instructions: entity.agent_instructions.clone(),
            ontology_namespace: entity.ontology_namespace.clone(),
            ontology_class: entity.ontology_class.clone(),
        };
        let entity_id = self.with_session_connection(
            "create entity",
            |connection| connection.create_entity(input.clone()),
            |connection| oauth_create_entity(connection, input.clone()),
        )?;
        self.with_session_connection(
            "read created entity",
            |connection| connection.list_entities_for_domain(entity.domain_id),
            |connection| oauth_list_entities_for_domain(connection, entity.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.entity_id == entity_id)
        .map(entity_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!("created entity {:?} was not readable", entity_id))
        })
    }

    async fn save_entity(&self, entity: &EntityRecord) -> Result<EntityRecord, CoreError> {
        let input = PgCreateEntity {
            entity_id: Some(entity.entity_id),
            domain_id: entity.domain_id,
            uuid: entity.uuid,
            name: entity.name.as_str().to_string(),
            description: entity.description.clone(),
            agent_instructions: entity.agent_instructions.clone(),
            ontology_namespace: entity.ontology_namespace.clone(),
            ontology_class: entity.ontology_class.clone(),
        };
        let entity_id = self.with_session_connection(
            "save entity",
            |connection| connection.create_entity(input.clone()),
            |connection| oauth_create_entity(connection, input.clone()),
        )?;
        self.with_session_connection(
            "read saved entity",
            |connection| connection.list_entities_for_domain(entity.domain_id),
            |connection| oauth_list_entities_for_domain(connection, entity.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.entity_id == entity_id)
        .map(entity_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!("saved entity {:?} was not readable", entity_id))
        })
    }

    async fn create_entity_relation(
        &self,
        relation: &NewEntityRelationRecord,
    ) -> Result<EntityRelationRecord, CoreError> {
        let input = PgCreateEntityRelation {
            entity_relation_id: None,
            subject_entity_id: relation.subject_entity_id,
            object_entity_id: relation.object_entity_id,
            domain_id: relation.domain_id,
            uuid: None,
            name: relation.name.as_str().to_string(),
            description: relation.description.clone(),
            agent_instructions: relation.agent_instructions.clone(),
            ontology_namespace: relation.ontology_namespace.clone(),
            ontology_class: relation.ontology_class.clone(),
        };
        let relation_id = self.with_session_connection(
            "create entity relation",
            |connection| connection.create_entity_relation(input.clone()),
            |connection| oauth_create_entity_relation(connection, input.clone()),
        )?;
        self.with_session_connection(
            "read created entity relation",
            |connection| connection.list_entity_relations_for_domain(relation.domain_id),
            |connection| oauth_list_entity_relations_for_domain(connection, relation.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.entity_relation_id == relation_id)
        .map(entity_relation_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "created entity relation {:?} was not readable",
                relation_id
            ))
        })
    }

    async fn save_entity_relation(
        &self,
        relation: &EntityRelationRecord,
    ) -> Result<EntityRelationRecord, CoreError> {
        let input = PgCreateEntityRelation {
            entity_relation_id: Some(relation.entity_relation_id),
            subject_entity_id: relation.subject_entity_id,
            object_entity_id: relation.object_entity_id,
            domain_id: relation.domain_id,
            uuid: relation.uuid,
            name: relation.name.as_str().to_string(),
            description: relation.description.clone(),
            agent_instructions: relation.agent_instructions.clone(),
            ontology_namespace: relation.ontology_namespace.clone(),
            ontology_class: relation.ontology_class.clone(),
        };
        let relation_id = self.with_session_connection(
            "save entity relation",
            |connection| connection.create_entity_relation(input.clone()),
            |connection| oauth_create_entity_relation(connection, input.clone()),
        )?;
        self.with_session_connection(
            "read saved entity relation",
            |connection| connection.list_entity_relations_for_domain(relation.domain_id),
            |connection| oauth_list_entity_relations_for_domain(connection, relation.domain_id),
        )?
        .into_iter()
        .find(|read_model| read_model.entity_relation_id == relation_id)
        .map(entity_relation_record)
        .transpose()?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "saved entity relation {:?} was not readable",
                relation_id
            ))
        })
    }

    async fn delete_entity_record(&self, entity_id: EntityId) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete entity record",
            |connection| delete_semantic_record(connection, "entities", "entity_id", entity_id.0),
            |connection| delete_semantic_record(connection, "entities", "entity_id", entity_id.0),
        )
    }

    async fn delete_entity_relation_record(
        &self,
        relation_id: EntityRelationId,
    ) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete entity relation record",
            |connection| {
                delete_semantic_record(
                    connection,
                    "entityrelations",
                    "entityrelation_id",
                    relation_id.0,
                )
            },
            |connection| {
                delete_semantic_record(
                    connection,
                    "entityrelations",
                    "entityrelation_id",
                    relation_id.0,
                )
            },
        )
    }

    async fn get_entity_instance(
        &self,
        instance_id: i64,
    ) -> Result<Option<EntityInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "get entity instance",
                |connection| connection.get_entity_instance(instance_id),
                |connection| oauth_get_entity_instance(connection, instance_id),
            )?
            .map(entity_instance_record))
    }

    async fn get_entity_instance_by_uuid(
        &self,
        uuid: uuid::Uuid,
    ) -> Result<Option<EntityInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "get entity instance by uuid",
                |connection| connection.get_entity_instance_by_uuid(uuid),
                |connection| oauth_get_entity_instance_by_uuid(connection, uuid),
            )?
            .map(entity_instance_record))
    }

    async fn list_entity_instances(
        &self,
        entity_id: EntityId,
    ) -> Result<Vec<EntityInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list entity instances",
                |connection| connection.list_entity_instances_for_entity(entity_id),
                |connection| oauth_list_entity_instances_for_entity(connection, entity_id),
            )?
            .into_iter()
            .map(entity_instance_record)
            .collect())
    }

    async fn save_entity_instance(
        &self,
        instance: &EntityInstanceWriteRecord,
    ) -> Result<EntityInstanceRecord, CoreError> {
        let input = PgWriteEntityInstance {
            instance_id: instance.instance_id,
            uuid: instance.uuid,
            entity_id: instance.entity_id,
            label: instance.label.clone(),
            note: instance.note.clone(),
            transformation_id: instance.transformation_id,
        };
        let instance_id = self.with_session_connection(
            "save entity instance",
            |connection| connection.save_entity_instance(input.clone()),
            |connection| oauth_save_entity_instance(connection, input.clone()),
        )?;
        self.get_entity_instance(instance_id).await?.ok_or_else(|| {
            CoreError::NotFound(format!(
                "saved entity instance {instance_id} was not readable"
            ))
        })
    }

    async fn create_entity_instance(
        &self,
        instance: &NewEntityInstanceRecord,
    ) -> Result<EntityInstanceRecord, CoreError> {
        let input = PgWriteEntityInstance {
            instance_id: None,
            uuid: None,
            entity_id: instance.entity_id,
            label: instance.label.clone(),
            note: instance.note.clone(),
            transformation_id: instance.transformation_id,
        };
        let instance_id = self.with_session_connection(
            "create entity instance",
            |connection| connection.save_entity_instance(input.clone()),
            |connection| oauth_save_entity_instance(connection, input.clone()),
        )?;
        self.get_entity_instance(instance_id).await?.ok_or_else(|| {
            CoreError::NotFound(format!(
                "created entity instance {instance_id} was not readable"
            ))
        })
    }

    async fn get_study_entity_instance(
        &self,
        study_id: StudyId,
        entity_id: EntityId,
        external_id: &str,
    ) -> Result<Option<StudyEntityInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "get study entity instance",
                |connection| connection.get_study_entity_instance(study_id, entity_id, external_id),
                |connection| {
                    oauth_get_study_entity_instance(connection, study_id, entity_id, external_id)
                },
            )?
            .map(study_entity_instance_record))
    }

    async fn save_study_entity_instance(
        &self,
        mapping: &StudyEntityInstanceWriteRecord,
    ) -> Result<StudyEntityInstanceRecord, CoreError> {
        let input = PgWriteStudyEntityInstance {
            study_id: mapping.study_id,
            entity_instance_id: mapping.entity_instance_id,
            entity_id: mapping.entity_id,
            external_id: mapping.external_id.clone(),
            transformation_id: mapping.transformation_id,
        };
        self.with_session_connection(
            "save study entity instance",
            |connection| connection.save_study_entity_instance(input.clone()),
            |connection| oauth_save_study_entity_instance(connection, input.clone()),
        )?;
        self.get_study_entity_instance(mapping.study_id, mapping.entity_id, &mapping.external_id)
            .await?
            .ok_or_else(|| {
                CoreError::NotFound(format!(
                    "saved study entity mapping {:?}/{} was not readable",
                    mapping.study_id, mapping.external_id
                ))
            })
    }

    async fn list_study_entity_instances(
        &self,
        study_id: StudyId,
    ) -> Result<Vec<StudyEntityInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list study entity instances",
                |connection| connection.list_study_entity_instances(study_id),
                |connection| oauth_list_study_entity_instances(connection, study_id),
            )?
            .into_iter()
            .map(study_entity_instance_record)
            .collect())
    }

    async fn get_relation_instance(
        &self,
        relation_instance_id: i64,
    ) -> Result<Option<RelationInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "get relation instance",
                |connection| connection.get_relation_instance(relation_instance_id),
                |connection| oauth_get_relation_instance(connection, relation_instance_id),
            )?
            .map(relation_instance_record))
    }

    async fn get_relation_instance_by_uuid(
        &self,
        uuid: uuid::Uuid,
    ) -> Result<Option<RelationInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "get relation instance by uuid",
                |connection| connection.get_relation_instance_by_uuid(uuid),
                |connection| oauth_get_relation_instance_by_uuid(connection, uuid),
            )?
            .map(relation_instance_record))
    }

    async fn list_relation_instances(
        &self,
        relation_id: EntityRelationId,
    ) -> Result<Vec<RelationInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list relation instances",
                |connection| connection.list_relation_instances_for_relation(relation_id),
                |connection| oauth_list_relation_instances_for_relation(connection, relation_id),
            )?
            .into_iter()
            .map(relation_instance_record)
            .collect())
    }

    async fn save_relation_instance(
        &self,
        relation: &RelationInstanceWriteRecord,
    ) -> Result<RelationInstanceRecord, CoreError> {
        let input = PgWriteRelationInstance {
            relation_instance_id: relation.relation_instance_id,
            uuid: relation.uuid,
            entity_relation_id: relation.entity_relation_id,
            entity_instance_id_1: relation.entity_instance_id_1,
            entity_instance_id_2: relation.entity_instance_id_2,
            valid_from: relation.valid_from,
            valid_to: relation.valid_to,
            note: relation.note.clone(),
            transformation_id: relation.transformation_id,
        };
        let relation_instance_id = self.with_session_connection(
            "save relation instance",
            |connection| connection.save_relation_instance(input.clone()),
            |connection| oauth_save_relation_instance(connection, input.clone()),
        )?;
        self.get_relation_instance(relation_instance_id)
            .await?
            .ok_or_else(|| {
                CoreError::NotFound(format!(
                    "saved relation instance {relation_instance_id} was not readable"
                ))
            })
    }

    async fn create_relation_instance(
        &self,
        relation: &NewRelationInstanceRecord,
    ) -> Result<RelationInstanceRecord, CoreError> {
        let input = PgWriteRelationInstance {
            relation_instance_id: None,
            uuid: None,
            entity_relation_id: relation.entity_relation_id,
            entity_instance_id_1: relation.entity_instance_id_1,
            entity_instance_id_2: relation.entity_instance_id_2,
            valid_from: relation.valid_from,
            valid_to: relation.valid_to,
            note: relation.note.clone(),
            transformation_id: relation.transformation_id,
        };
        let relation_instance_id = self.with_session_connection(
            "create relation instance",
            |connection| connection.save_relation_instance(input.clone()),
            |connection| oauth_save_relation_instance(connection, input.clone()),
        )?;
        self.get_relation_instance(relation_instance_id)
            .await?
            .ok_or_else(|| {
                CoreError::NotFound(format!(
                    "created relation instance {relation_instance_id} was not readable"
                ))
            })
    }

    async fn get_study_relation_instance(
        &self,
        study_id: StudyId,
        relation_id: EntityRelationId,
        external_id: &str,
    ) -> Result<Option<StudyRelationInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "get study relation instance",
                |connection| {
                    connection.get_study_relation_instance(study_id, relation_id, external_id)
                },
                |connection| {
                    oauth_get_study_relation_instance(
                        connection,
                        study_id,
                        relation_id,
                        external_id,
                    )
                },
            )?
            .map(study_relation_instance_record))
    }

    async fn save_study_relation_instance(
        &self,
        mapping: &StudyRelationInstanceWriteRecord,
    ) -> Result<StudyRelationInstanceRecord, CoreError> {
        let input = PgWriteStudyRelationInstance {
            study_id: mapping.study_id,
            relation_instance_id: mapping.relation_instance_id,
            entity_relation_id: mapping.entity_relation_id,
            external_id: mapping.external_id.clone(),
            transformation_id: mapping.transformation_id,
        };
        self.with_session_connection(
            "save study relation instance",
            |connection| connection.save_study_relation_instance(input.clone()),
            |connection| oauth_save_study_relation_instance(connection, input.clone()),
        )?;
        self.get_study_relation_instance(
            mapping.study_id,
            mapping.entity_relation_id,
            &mapping.external_id,
        )
        .await?
        .ok_or_else(|| {
            CoreError::NotFound(format!(
                "saved study relation mapping {:?}/{} was not readable",
                mapping.study_id, mapping.external_id
            ))
        })
    }

    async fn list_study_relation_instances(
        &self,
        study_id: StudyId,
    ) -> Result<Vec<StudyRelationInstanceRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list study relation instances",
                |connection| connection.list_study_relation_instances(study_id),
                |connection| oauth_list_study_relation_instances(connection, study_id),
            )?
            .into_iter()
            .map(study_relation_instance_record)
            .collect())
    }

    async fn link_asset_version_entity(
        &self,
        link: &AssetVersionEntityRecord,
    ) -> Result<AssetVersionEntityRecord, CoreError> {
        self.with_session_connection(
            "link asset version entity",
            |connection| {
                connection.link_asset_version_entity(
                    link.version_id,
                    link.entity_instance_id,
                    link.transformation_id,
                )
            },
            |connection| oauth_link_asset_version_entity(connection, link),
        )?;
        Ok(link.clone())
    }

    async fn list_asset_version_entities(
        &self,
        version_id: VersionId,
    ) -> Result<Vec<AssetVersionEntityRecord>, CoreError> {
        let links = self.with_session_connection(
            "list asset version entities",
            |connection| connection.list_asset_version_entities(version_id),
            |connection| oauth_list_asset_version_entities(connection, version_id),
        )?;
        Ok(links.into_iter().map(asset_version_entity_record).collect())
    }

    async fn list_asset_version_entities_for_instance(
        &self,
        entity_instance_id: i64,
    ) -> Result<Vec<AssetVersionEntityRecord>, CoreError> {
        let links = self.with_session_connection(
            "list asset version entities for instance",
            |connection| connection.list_asset_version_entities_for_instance(entity_instance_id),
            |connection| {
                oauth_list_asset_version_entities_for_instance(connection, entity_instance_id)
            },
        )?;
        Ok(links.into_iter().map(asset_version_entity_record).collect())
    }

    async fn link_dataset_version_entity(
        &self,
        link: &DatasetVersionEntityRecord,
    ) -> Result<DatasetVersionEntityRecord, CoreError> {
        self.with_session_connection(
            "link dataset version entity",
            |connection| {
                connection.add_dataset_version_entity(
                    link.version_id,
                    link.entity_instance_id,
                    link.entity_variable_id,
                    link.transformation_id,
                )
            },
            |connection| oauth_link_dataset_version_entity(connection, link),
        )?;
        Ok(link.clone())
    }

    async fn list_dataset_version_entities(
        &self,
        version_id: VersionId,
    ) -> Result<Vec<DatasetVersionEntityRecord>, CoreError> {
        let links = self.with_session_connection(
            "list dataset version entities",
            |connection| connection.list_dataset_version_entities(version_id),
            |connection| oauth_list_dataset_version_entities(connection, version_id),
        )?;
        Ok(links
            .into_iter()
            .map(dataset_version_entity_record)
            .collect())
    }

    async fn list_dataset_version_entities_for_instance(
        &self,
        entity_instance_id: i64,
    ) -> Result<Vec<DatasetVersionEntityRecord>, CoreError> {
        let links = self.with_session_connection(
            "list dataset version entities for instance",
            |connection| connection.list_dataset_version_entities_for_instance(entity_instance_id),
            |connection| {
                oauth_list_dataset_version_entities_for_instance(connection, entity_instance_id)
            },
        )?;
        Ok(links
            .into_iter()
            .map(dataset_version_entity_record)
            .collect())
    }

    async fn link_asset_version_relation_instance(
        &self,
        link: &AssetVersionRelationInstanceRecord,
    ) -> Result<AssetVersionRelationInstanceRecord, CoreError> {
        self.with_session_connection(
            "link asset version relation instance",
            |connection| {
                connection.link_asset_version_relation_instance(
                    link.version_id,
                    link.relation_instance_id,
                    link.transformation_id,
                )
            },
            |connection| oauth_link_asset_version_relation_instance(connection, link),
        )?;
        Ok(link.clone())
    }

    async fn list_asset_version_relation_instances(
        &self,
        version_id: VersionId,
    ) -> Result<Vec<AssetVersionRelationInstanceRecord>, CoreError> {
        let links = self.with_session_connection(
            "list asset version relation instances",
            |connection| connection.list_asset_version_relation_instances(version_id),
            |connection| oauth_list_asset_version_relation_instances(connection, version_id),
        )?;
        Ok(links
            .into_iter()
            .map(asset_version_relation_instance_record)
            .collect())
    }

    async fn list_asset_version_relation_instances_for_instance(
        &self,
        relation_instance_id: i64,
    ) -> Result<Vec<AssetVersionRelationInstanceRecord>, CoreError> {
        let links = self.with_session_connection(
            "list asset version relation instances for instance",
            |connection| {
                connection.list_asset_version_relation_instances_for_instance(relation_instance_id)
            },
            |connection| {
                oauth_list_asset_version_relation_instances_for_instance(
                    connection,
                    relation_instance_id,
                )
            },
        )?;
        Ok(links
            .into_iter()
            .map(asset_version_relation_instance_record)
            .collect())
    }

    async fn link_dataset_version_relation_instance(
        &self,
        link: &DatasetVersionRelationInstanceRecord,
    ) -> Result<DatasetVersionRelationInstanceRecord, CoreError> {
        self.with_session_connection(
            "link dataset version relation instance",
            |connection| {
                connection.add_dataset_version_relation(
                    link.version_id,
                    link.relation_instance_id,
                    link.subject_variable_id,
                    link.object_variable_id,
                    link.relation_variable_id,
                    link.transformation_id,
                )
            },
            |connection| oauth_link_dataset_version_relation_instance(connection, link),
        )?;
        Ok(link.clone())
    }

    async fn list_dataset_version_relation_instances(
        &self,
        version_id: VersionId,
    ) -> Result<Vec<DatasetVersionRelationInstanceRecord>, CoreError> {
        let links = self.with_session_connection(
            "list dataset version relation instances",
            |connection| connection.list_dataset_version_relation_instances(version_id),
            |connection| oauth_list_dataset_version_relation_instances(connection, version_id),
        )?;
        Ok(links
            .into_iter()
            .map(dataset_version_relation_instance_record)
            .collect())
    }

    async fn list_dataset_version_relation_instances_for_instance(
        &self,
        relation_instance_id: i64,
    ) -> Result<Vec<DatasetVersionRelationInstanceRecord>, CoreError> {
        let links = self.with_session_connection(
            "list dataset version relation instances for instance",
            |connection| {
                connection
                    .list_dataset_version_relation_instances_for_instance(relation_instance_id)
            },
            |connection| {
                oauth_list_dataset_version_relation_instances_for_instance(
                    connection,
                    relation_instance_id,
                )
            },
        )?;
        Ok(links
            .into_iter()
            .map(dataset_version_relation_instance_record)
            .collect())
    }

    async fn link_dataset_version_entities(
        &self,
        links: &[DatasetVersionEntityRecord],
    ) -> Result<(), CoreError> {
        let link_inputs: Vec<_> = links
            .iter()
            .map(|link| {
                (
                    link.version_id,
                    link.entity_instance_id,
                    link.entity_variable_id,
                    link.transformation_id,
                )
            })
            .collect();
        self.with_session_connection(
            "bulk link dataset version entities",
            |connection| connection.link_dataset_version_entities(&link_inputs),
            |connection| oauth_link_dataset_version_entities(connection, links),
        )
    }

    async fn link_dataset_version_relation_instances(
        &self,
        links: &[DatasetVersionRelationInstanceRecord],
    ) -> Result<(), CoreError> {
        let link_inputs: Vec<_> = links
            .iter()
            .map(|link| {
                (
                    link.version_id,
                    link.relation_instance_id,
                    link.subject_variable_id,
                    link.object_variable_id,
                    link.relation_variable_id,
                    link.transformation_id,
                )
            })
            .collect();
        self.with_session_connection(
            "bulk link dataset version relation instances",
            |connection| connection.link_dataset_version_relation_instances(&link_inputs),
            |connection| oauth_link_dataset_version_relation_instances(connection, links),
        )
    }
}

#[async_trait::async_trait]
impl TransformationRepository for PgMetadataRepository<'_> {
    async fn create_transformation(
        &self,
        transformation: &NewTransformationRecord,
    ) -> Result<TransformationRecord, CoreError> {
        let input = PgCreateTransformation {
            transformation_id: None,
            transformation_type: transformation.transformation_type,
            description: transformation.description.clone(),
            repository_url: transformation.repository_url.clone(),
            commit_hash: transformation.commit_hash.clone(),
            file_path: transformation.file_path.clone(),
            date_created: transformation.date_created,
            created_by: transformation.created_by.clone(),
        };
        self.create_transformation_from_input(input).await
    }

    async fn save_transformation(
        &self,
        transformation: &TransformationRecord,
    ) -> Result<TransformationRecord, CoreError> {
        let input = PgCreateTransformation {
            transformation_id: Some(transformation.transformation_id),
            transformation_type: transformation.transformation_type,
            description: transformation.description.clone(),
            repository_url: transformation.repository_url.clone(),
            commit_hash: transformation.commit_hash.clone(),
            file_path: transformation.file_path.clone(),
            date_created: transformation.date_created,
            created_by: transformation.created_by.clone(),
        };
        self.create_transformation_from_input(input).await
    }

    async fn add_inputs(
        &self,
        transformation_id: TransformationId,
        input_ids: &[VersionId],
    ) -> Result<(), CoreError> {
        self.with_session_connection(
            "link transformation inputs",
            |connection| {
                for version_id in input_ids {
                    connection.link_transformation_input(transformation_id, *version_id)?;
                }
                Ok(())
            },
            |connection| oauth_add_transformation_inputs(connection, transformation_id, input_ids),
        )
    }

    async fn add_outputs(
        &self,
        transformation_id: TransformationId,
        output_ids: &[VersionId],
    ) -> Result<(), CoreError> {
        self.with_session_connection(
            "link transformation outputs",
            |connection| {
                for version_id in output_ids {
                    connection.link_transformation_output(transformation_id, *version_id)?;
                }
                Ok(())
            },
            |connection| {
                oauth_add_transformation_outputs(connection, transformation_id, output_ids)
            },
        )
    }

    async fn delete_transformation(
        &self,
        transformation_id: TransformationId,
    ) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete transformation",
            |connection| connection.delete_transformation(transformation_id),
            |connection| oauth_delete_transformation(connection, transformation_id),
        )
    }

    async fn list_inputs(
        &self,
        transformation_id: TransformationId,
    ) -> Result<Vec<TransformationInputLinkRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list transformation inputs",
                |connection| connection.list_transformation_inputs(transformation_id),
                |connection| oauth_list_transformation_inputs(connection, transformation_id),
            )?
            .into_iter()
            .map(transformation_input_record)
            .collect())
    }

    async fn list_outputs(
        &self,
        transformation_id: TransformationId,
    ) -> Result<Vec<TransformationOutputLinkRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list transformation outputs",
                |connection| connection.list_transformation_outputs(transformation_id),
                |connection| oauth_list_transformation_outputs(connection, transformation_id),
            )?
            .into_iter()
            .map(transformation_output_record)
            .collect())
    }

    async fn list_transformations_producing(
        &self,
        version_id: VersionId,
    ) -> Result<Vec<TransformationRecord>, CoreError> {
        self.with_session_connection(
            "list transformations producing version",
            |connection| connection.list_transformations_producing_version(version_id),
            |connection| oauth_list_transformations_producing_version(connection, version_id),
        )?
        .into_iter()
        .map(transformation_record)
        .collect()
    }

    async fn list_transformations(&self) -> Result<Vec<TransformationRecord>, CoreError> {
        self.with_session_connection(
            "list transformations",
            |connection| connection.list_transformations(),
            oauth_list_transformations,
        )?
        .into_iter()
        .map(transformation_record)
        .collect()
    }
}

#[async_trait::async_trait]
impl TagRepository for PgMetadataRepository<'_> {
    async fn list_tags(&self) -> Result<Vec<TagRecord>, CoreError> {
        Ok(self
            .with_session_connection(
                "list tags",
                |connection| connection.list_tags(),
                oauth_list_tags,
            )?
            .into_iter()
            .map(tag_record)
            .collect())
    }

    async fn upsert_tag(&self, normalized_name: &str) -> Result<TagRecord, CoreError> {
        let tag_id = self.with_session_connection(
            "upsert tag",
            |connection| connection.upsert_tag(normalized_name),
            |connection| oauth_upsert_tag(connection, normalized_name),
        )?;
        Ok(TagRecord {
            tag_id: TagId(tag_id),
            name: normalized_name.to_string(),
        })
    }

    async fn attach_tag(
        &self,
        tag_id: TagId,
        target: TagAttachmentTarget,
    ) -> Result<(), CoreError> {
        let direct_target = pg_write_tag_target(target.clone());
        let oauth_target = pg_tag_target(target);
        self.with_session_connection(
            "attach tag",
            |connection| connection.attach_tag(direct_target, tag_id.0),
            |connection| oauth_attach_tag(connection, oauth_target, tag_id.0),
        )
    }

    async fn detach_tag(
        &self,
        tag_id: TagId,
        target: TagAttachmentTarget,
    ) -> Result<(), CoreError> {
        let direct_target = pg_write_tag_target(target.clone());
        let oauth_target = pg_tag_target(target);
        self.with_session_connection(
            "detach tag",
            |connection| connection.detach_tag(direct_target, tag_id.0),
            |connection| oauth_detach_tag(connection, oauth_target, tag_id.0),
        )
    }

    async fn delete_tag_if_unused(&self, tag_id: TagId) -> Result<bool, CoreError> {
        self.with_session_connection(
            "delete unused tag",
            |connection| connection.delete_tag_if_unused(tag_id.0),
            |connection| oauth_delete_tag_if_unused(connection, tag_id.0),
        )
    }

    async fn list_tags_for_target(
        &self,
        target: TagAttachmentTarget,
    ) -> Result<Vec<TagRecord>, CoreError> {
        let target = pg_tag_target(target);
        Ok(self
            .with_session_connection(
                "list tags for target",
                |connection| connection.list_tag_records_for_target(target),
                |connection| oauth_list_tags_for_target(connection, target),
            )?
            .into_iter()
            .map(tag_record)
            .collect())
    }
}

fn oauth_list_assets_for_study(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
) -> Result<Vec<PgAssetReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list assets for study",
        "
        SELECT asset_id::text, study_id::text, name, description, agent_instructions,
               asset_type::text, date_created::text, created_by
          FROM public.assets
         WHERE study_id = $1
         ORDER BY name
        ",
        &[&study_id.0],
    )?;
    rows.iter().map(oauth_asset_from_row).collect()
}

fn oauth_get_asset(
    connection: &mut LibpqOAuthConnection,
    asset_id: AssetId,
) -> Result<Option<PgAssetReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get asset by id",
        "
        SELECT asset_id::text, study_id::text, name, description, agent_instructions,
               asset_type::text, date_created::text, created_by
          FROM public.assets
         WHERE asset_id = $1
         LIMIT 1
        ",
        &[&asset_id.0],
    )?;
    row.as_ref().map(oauth_asset_from_row).transpose()
}

fn oauth_get_asset_by_name(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
    name: &str,
) -> Result<Option<PgAssetReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get asset by name",
        "
        SELECT asset_id::text, study_id::text, name, description, agent_instructions,
               asset_type::text, date_created::text, created_by
          FROM public.assets
         WHERE study_id = $1 AND name = $2
         LIMIT 1
        ",
        &[&study_id.0, &name],
    )?;
    row.as_ref().map(oauth_asset_from_row).transpose()
}

fn oauth_create_asset(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateAsset,
) -> Result<AssetId, PgMetaError> {
    let asset_type = asset_type_sql(input.asset_type);
    let row = match input.asset_id {
        Some(asset_id) => MetadataExecutor::query_one(
            connection,
            "create asset",
            "
            INSERT INTO public.assets (
                asset_id, study_id, name, description, agent_instructions, asset_type,
                date_created, created_by
            )
            VALUES ($1, $2, $3, $4, $5, CAST($6 AS text)::public.asset_type_enum,
                    COALESCE($7, CURRENT_TIMESTAMP), COALESCE($8, CURRENT_USER))
            RETURNING asset_id::text
            ",
            &[
                &asset_id.0,
                &input.study_id.0,
                &input.name,
                &input.description,
                &input.agent_instructions,
                &asset_type,
                &input.date_created,
                &input.created_by,
            ],
        )?,
        None => MetadataExecutor::query_one(
            connection,
            "create asset",
            "
            INSERT INTO public.assets (
                study_id, name, description, agent_instructions, asset_type,
                date_created, created_by
            )
            VALUES ($1, $2, $3, $4, CAST($5 AS text)::public.asset_type_enum,
                    COALESCE($6, CURRENT_TIMESTAMP), COALESCE($7, CURRENT_USER))
            RETURNING asset_id::text
            ",
            &[
                &input.study_id.0,
                &input.name,
                &input.description,
                &input.agent_instructions,
                &asset_type,
                &input.date_created,
                &input.created_by,
            ],
        )?,
    };
    Ok(AssetId(oauth_uuid(&row, 0, "asset_id")?))
}

fn oauth_list_asset_versions(
    connection: &mut LibpqOAuthConnection,
    asset_id: AssetId,
) -> Result<Vec<PgAssetVersionReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list asset versions",
        "
        SELECT version_id::text, asset_id::text, major::text, minor::text, patch::text,
               version_label, version_note, is_latest::text, doi, created_at::text, created_by
          FROM public.asset_versions
         WHERE asset_id = $1
         ORDER BY major, minor, patch
        ",
        &[&asset_id.0],
    )?;
    rows.iter().map(oauth_asset_version_from_row).collect()
}

fn oauth_get_asset_version(
    connection: &mut LibpqOAuthConnection,
    version_id: VersionId,
) -> Result<Option<PgAssetVersionReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get asset version",
        "
        SELECT version_id::text, asset_id::text, major::text, minor::text, patch::text,
               version_label, version_note, is_latest::text, doi, created_at::text, created_by
          FROM public.asset_versions
         WHERE version_id = $1
         LIMIT 1
        ",
        &[&version_id.0],
    )?;
    row.as_ref().map(oauth_asset_version_from_row).transpose()
}

fn oauth_create_asset_version(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateAssetVersion,
) -> Result<VersionId, PgMetaError> {
    MetadataExecutor::with_transaction(connection, "create asset version", |transaction| {
        oauth_insert_asset_version(transaction, input)
    })
}

fn oauth_insert_asset_version(
    transaction: &mut dyn MetadataTransaction<Row = LibpqOAuthRow>,
    input: PgCreateAssetVersion,
) -> Result<VersionId, PgMetaError> {
    transaction.query_one(
        "validate ingest classification",
        "SELECT public.tre_validate_ingest($1,$2,$3)",
        &[
            &input.asset_id.0,
            &input.classification.risk.as_str(),
            &input.classification.justification,
        ],
    )?;
    if input.is_latest.unwrap_or(true) {
        transaction.execute_command(
            "clear latest asset version",
            "UPDATE public.asset_versions SET is_latest = FALSE WHERE asset_id = $1 AND is_latest = TRUE",
            &[&input.asset_id.0],
        )?;
    }

    let row = match input.version_id {
        Some(version_id) => transaction.query_one(
            "create asset version",
            "
                INSERT INTO public.asset_versions (
                    version_id, asset_id, major, minor, patch, version_note, is_latest, doi,
                    created_at, created_by
                )
                VALUES ($1, $2, COALESCE($3, 1), COALESCE($4, 0), COALESCE($5, 0),
                        COALESCE($6, 'Original version'), COALESCE($7, TRUE), $8,
                        COALESCE($9, CURRENT_TIMESTAMP), COALESCE($10, CURRENT_USER))
                RETURNING version_id::text
                ",
            &[
                &version_id.0,
                &input.asset_id.0,
                &input.major,
                &input.minor,
                &input.patch,
                &input.version_note,
                &input.is_latest,
                &input.doi,
                &input.created_at,
                &input.created_by,
            ],
        )?,
        None => transaction.query_one(
            "create asset version",
            "
                INSERT INTO public.asset_versions (
                    asset_id, major, minor, patch, version_note, is_latest, doi,
                    created_at, created_by
                )
                VALUES ($1, COALESCE($2, 1), COALESCE($3, 0), COALESCE($4, 0),
                        COALESCE($5, 'Original version'), COALESCE($6, TRUE), $7,
                        COALESCE($8, CURRENT_TIMESTAMP), COALESCE($9, CURRENT_USER))
                RETURNING version_id::text
                ",
            &[
                &input.asset_id.0,
                &input.major,
                &input.minor,
                &input.patch,
                &input.version_note,
                &input.is_latest,
                &input.doi,
                &input.created_at,
                &input.created_by,
            ],
        )?,
    };
    Ok(VersionId(oauth_uuid(&row, 0, "version_id")?))
}

fn oauth_mark_asset_version_latest(
    connection: &mut LibpqOAuthConnection,
    asset_id: AssetId,
    version_id: VersionId,
) -> Result<(), PgMetaError> {
    MetadataExecutor::with_transaction(connection, "mark asset version latest", |transaction| {
        transaction.execute_command(
            "clear latest asset version",
            "UPDATE public.asset_versions SET is_latest = FALSE WHERE asset_id = $1 AND is_latest = TRUE",
            &[&asset_id.0],
        )?;
        transaction.query_one(
            "mark asset version latest",
            "UPDATE public.asset_versions SET is_latest = TRUE WHERE asset_id = $1 AND version_id = $2 RETURNING version_id::text",
            &[&asset_id.0, &version_id.0],
        )?;
        Ok(())
    })
}

fn oauth_list_asset_version_duo_restrictions(
    connection: &mut LibpqOAuthConnection,
    version_id: VersionId,
) -> Result<Vec<PgAssetVersionDuoRestrictionReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list asset version DUO restrictions",
        "
        SELECT avdr.asset_version_duo_restriction_id::text, avdr.version_id::text, avdr.duo_id,
               dc.duo_code, dc.label,
               avdr.qualifier_ontology_namespace, avdr.qualifier_ontology_id,
               avdr.qualifier_text, avdr.qualifier_date::text, avdr.qualifier_integer::text
          FROM public.asset_version_duo_restrictions avdr
          JOIN public.duo_codes dc ON dc.duo_id = avdr.duo_id
         WHERE avdr.version_id = $1
         ORDER BY avdr.asset_version_duo_restriction_id
        ",
        &[&version_id.0],
    )?;
    rows.iter().map(oauth_asset_version_duo_from_row).collect()
}

fn oauth_create_asset_version_duo_restriction(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateAssetVersionDuoRestriction,
) -> Result<i64, PgMetaError> {
    let restriction_id = optional_metadata_i32(
        input.asset_version_duo_restriction_id.map(|value| value.0),
        "asset_version_duo_restriction_id",
    )?;
    let row = match restriction_id {
        Some(restriction_id) => MetadataExecutor::query_one(
            connection,
            "create asset version DUO restriction",
            "
            INSERT INTO public.asset_version_duo_restrictions (
                asset_version_duo_restriction_id, version_id, duo_id,
                qualifier_ontology_namespace, qualifier_ontology_id,
                qualifier_text, qualifier_date, qualifier_integer
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
            RETURNING asset_version_duo_restriction_id::text
            ",
            &[
                &restriction_id,
                &input.version_id.0,
                &input.duo_id,
                &input.qualifiers.qualifier_ontology_namespace,
                &input.qualifiers.qualifier_ontology_id,
                &input.qualifiers.qualifier_text,
                &input.qualifiers.qualifier_date,
                &input.qualifiers.qualifier_integer,
            ],
        )?,
        None => MetadataExecutor::query_one(
            connection,
            "create asset version DUO restriction",
            "
            INSERT INTO public.asset_version_duo_restrictions (
                version_id, duo_id, qualifier_ontology_namespace, qualifier_ontology_id,
                qualifier_text, qualifier_date, qualifier_integer
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7)
            RETURNING asset_version_duo_restriction_id::text
            ",
            &[
                &input.version_id.0,
                &input.duo_id,
                &input.qualifiers.qualifier_ontology_namespace,
                &input.qualifiers.qualifier_ontology_id,
                &input.qualifiers.qualifier_text,
                &input.qualifiers.qualifier_date,
                &input.qualifiers.qualifier_integer,
            ],
        )?,
    };
    oauth_i64(&row, 0, "asset_version_duo_restriction_id")
}

fn oauth_list_datasets_for_asset(
    connection: &mut LibpqOAuthConnection,
    asset_id: AssetId,
) -> Result<Vec<PgDatasetReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list datasets for asset",
        "
        SELECT d.dataset_id::text
          FROM public.datasets d
          JOIN public.asset_versions av ON av.version_id = d.dataset_id
         WHERE av.asset_id = $1
         ORDER BY av.major, av.minor, av.patch
        ",
        &[&asset_id.0],
    )?;
    rows.iter().map(oauth_dataset_from_row).collect()
}

fn oauth_get_dataset(
    connection: &mut LibpqOAuthConnection,
    dataset_id: VersionId,
) -> Result<Option<PgDatasetReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get dataset",
        "
        SELECT dataset_id::text
          FROM public.datasets
         WHERE dataset_id = $1
         LIMIT 1
        ",
        &[&dataset_id.0],
    )?;
    row.as_ref().map(oauth_dataset_from_row).transpose()
}

fn oauth_create_dataset(
    connection: &mut LibpqOAuthConnection,
    dataset_id: VersionId,
) -> Result<(), PgMetaError> {
    MetadataExecutor::execute_command(
        connection,
        "create dataset",
        "INSERT INTO public.datasets (dataset_id) VALUES ($1)",
        &[&dataset_id.0],
    )?;
    Ok(())
}

fn oauth_get_dataset_version_withdrawal(
    connection: &mut LibpqOAuthConnection,
    dataset_id: VersionId,
) -> Result<Option<DatasetVersionWithdrawalRecord>, PgMetaError> {
    if !oauth_dataset_version_withdrawals_table_exists(connection)? {
        return Ok(None);
    }

    let row = MetadataExecutor::query_optional(
        connection,
        "get dataset version withdrawal",
        "
        SELECT dataset_id::text, asset_id::text, withdrawn_at::text, withdrawn_by, reason,
               tombstone_state, provenance, drop_lake_table::text
          FROM public.dataset_version_withdrawals
         WHERE dataset_id = $1
         LIMIT 1
        ",
        &[&dataset_id.0],
    )?;
    row.as_ref()
        .map(oauth_dataset_version_withdrawal_from_row)
        .transpose()
}

fn oauth_list_dataset_version_withdrawals(
    connection: &mut LibpqOAuthConnection,
    asset_id: AssetId,
) -> Result<Vec<DatasetVersionWithdrawalRecord>, PgMetaError> {
    if !oauth_dataset_version_withdrawals_table_exists(connection)? {
        return Ok(Vec::new());
    }

    let rows = MetadataExecutor::query_many(
        connection,
        "list dataset version withdrawals",
        "
        SELECT dataset_id::text, asset_id::text, withdrawn_at::text, withdrawn_by, reason,
               tombstone_state, provenance, drop_lake_table::text
          FROM public.dataset_version_withdrawals
         WHERE asset_id = $1
         ORDER BY withdrawn_at, dataset_id
        ",
        &[&asset_id.0],
    )?;
    rows.iter()
        .map(oauth_dataset_version_withdrawal_from_row)
        .collect()
}

fn pg_dataset_version_withdrawals_table_exists(
    connection: &mut PgMetadataConnection,
) -> Result<bool, PgMetaError> {
    let row = MetadataExecutor::query_one(
        connection,
        "check dataset version withdrawals table",
        "SELECT to_regclass($1) IS NOT NULL",
        &[&DATASET_VERSION_WITHDRAWALS_RELATION],
    )?;
    Ok(row.get(0))
}

fn oauth_dataset_version_withdrawals_table_exists(
    connection: &mut LibpqOAuthConnection,
) -> Result<bool, PgMetaError> {
    let row = MetadataExecutor::query_one(
        connection,
        "check dataset version withdrawals table",
        "SELECT to_regclass($1) IS NOT NULL",
        &[&DATASET_VERSION_WITHDRAWALS_RELATION],
    )?;
    oauth_bool(&row, 0, DATASET_VERSION_WITHDRAWALS_EXISTS_FIELD)
}

fn oauth_save_dataset_version_withdrawal(
    connection: &mut LibpqOAuthConnection,
    withdrawal: &NewDatasetVersionWithdrawalRecord,
) -> Result<(), PgMetaError> {
    MetadataExecutor::execute_command(
        connection,
        "save dataset version withdrawal",
        "
        INSERT INTO public.dataset_version_withdrawals (
            dataset_id, asset_id, withdrawn_by, reason, tombstone_state, provenance,
            drop_lake_table
        )
        VALUES ($1, $2, COALESCE($3, CURRENT_USER), $4, $5, $6, $7)
        ON CONFLICT (dataset_id) DO NOTHING
        ",
        &[
            &withdrawal.dataset_id.0,
            &withdrawal.asset_id.0,
            &withdrawal.withdrawn_by,
            &withdrawal.reason,
            &withdrawal.tombstone_state,
            &withdrawal.provenance,
            &withdrawal.drop_lake_table,
        ],
    )?;
    Ok(())
}

fn oauth_deactivate_dataset_version_links(
    connection: &mut LibpqOAuthConnection,
    dataset_id: VersionId,
) -> Result<(), PgMetaError> {
    MetadataExecutor::with_transaction(
        connection,
        "deactivate dataset version links",
        |transaction| {
            for (operation, sql) in [
                (
                    "delete dataset version relation links",
                    "DELETE FROM public.dataset_version_relation_instances WHERE version_id = $1",
                ),
                (
                    "delete dataset version entity links",
                    "DELETE FROM public.dataset_version_entities WHERE version_id = $1",
                ),
                (
                    "delete asset version relation links",
                    "DELETE FROM public.asset_version_relation_instances WHERE version_id = $1",
                ),
                (
                    "delete asset version entity links",
                    "DELETE FROM public.asset_version_entities WHERE version_id = $1",
                ),
                (
                    "delete dataset variable links",
                    "DELETE FROM public.dataset_variables WHERE dataset_id = $1",
                ),
            ] {
                transaction.execute_command(operation, sql, &[&dataset_id.0])?;
            }
            Ok(())
        },
    )
}

fn delete_asset_version_metadata_record<E>(
    connection: &mut E,
    version_id: VersionId,
) -> Result<bool, PgMetaError>
where
    E: MetadataExecutor,
{
    delete_version_metadata(connection, version_id, None)
}

fn delete_version_metadata<E: MetadataExecutor>(
    connection: &mut E,
    version_id: VersionId,
    latest_promotion: Option<(AssetId, Option<VersionId>)>,
) -> Result<bool, PgMetaError> {
    connection.with_transaction("delete asset version record", |transaction| {
        if let Some((asset_id, promoted)) = latest_promotion {
            transaction.execute_command(
                "clear deleted latest Datafile version",
                "UPDATE public.asset_versions SET is_latest = FALSE WHERE asset_id = $1 AND is_latest = TRUE",
                &[&asset_id.0],
            )?;
            if let Some(promoted) = promoted {
                transaction.query_one(
                    "promote previous Datafile version",
                    "UPDATE public.asset_versions SET is_latest = TRUE WHERE asset_id = $1 AND version_id = $2 RETURNING version_id::text",
                    &[&asset_id.0, &promoted.0],
                )?;
            }
        }
        for (operation, sql) in [
            (
                "delete dataset version relation links",
                "DELETE FROM public.dataset_version_relation_instances WHERE version_id = $1",
            ),
            (
                "delete dataset version entity links",
                "DELETE FROM public.dataset_version_entities WHERE version_id = $1",
            ),
            (
                "delete asset version relation links",
                "DELETE FROM public.asset_version_relation_instances WHERE version_id = $1",
            ),
            (
                "delete asset version entity links",
                "DELETE FROM public.asset_version_entities WHERE version_id = $1",
            ),
            (
                "delete dataset variable links",
                "DELETE FROM public.dataset_variables WHERE dataset_id = $1",
            ),
        ] {
            transaction.execute_command(operation, sql, &[&version_id.0])?;
        }
        let deleted = transaction.execute_command(
            "delete asset version record",
            "DELETE FROM public.asset_versions WHERE version_id = $1",
            &[&version_id.0],
        )?;
        Ok(deleted > 0)
    })
}

fn delete_datafile_metadata_record<E>(
    connection: &mut E,
    datafile_id: VersionId,
) -> Result<bool, PgMetaError>
where
    E: MetadataExecutor,
{
    let deleted = connection.execute_command(
        "delete datafile record",
        "DELETE FROM public.datafiles WHERE datafile_id = $1",
        &[&datafile_id.0],
    )?;
    Ok(deleted > 0)
}

fn delete_asset_metadata_record<E>(
    connection: &mut E,
    asset_id: AssetId,
) -> Result<bool, PgMetaError>
where
    E: MetadataExecutor,
{
    connection.with_transaction("delete asset record", |transaction| {
        for (operation, sql) in [
            (
                "delete dataset version relation links for asset",
                "DELETE FROM public.dataset_version_relation_instances WHERE version_id IN (SELECT version_id FROM public.asset_versions WHERE asset_id = $1)",
            ),
            (
                "delete dataset version entity links for asset",
                "DELETE FROM public.dataset_version_entities WHERE version_id IN (SELECT version_id FROM public.asset_versions WHERE asset_id = $1)",
            ),
            (
                "delete asset version relation links for asset",
                "DELETE FROM public.asset_version_relation_instances WHERE version_id IN (SELECT version_id FROM public.asset_versions WHERE asset_id = $1)",
            ),
            (
                "delete asset version entity links for asset",
                "DELETE FROM public.asset_version_entities WHERE version_id IN (SELECT version_id FROM public.asset_versions WHERE asset_id = $1)",
            ),
            (
                "delete dataset variable links for asset",
                "DELETE FROM public.dataset_variables WHERE dataset_id IN (SELECT version_id FROM public.asset_versions WHERE asset_id = $1)",
            ),
        ] {
            transaction.execute_command(operation, sql, &[&asset_id.0])?;
        }
        let deleted = transaction.execute_command(
            "delete asset record",
            "DELETE FROM public.assets WHERE asset_id = $1",
            &[&asset_id.0],
        )?;
        Ok(deleted > 0)
    })
}

fn delete_study_metadata_record<E>(
    connection: &mut E,
    study_id: StudyId,
) -> Result<bool, PgMetaError>
where
    E: MetadataExecutor,
{
    connection.with_transaction("delete study record", |transaction| {
        for (operation, sql) in [
            (
                "delete dataset version relation links for study",
                "DELETE FROM public.dataset_version_relation_instances WHERE version_id IN (SELECT av.version_id FROM public.asset_versions av JOIN public.assets a ON a.asset_id = av.asset_id WHERE a.study_id = $1)",
            ),
            (
                "delete dataset version entity links for study",
                "DELETE FROM public.dataset_version_entities WHERE version_id IN (SELECT av.version_id FROM public.asset_versions av JOIN public.assets a ON a.asset_id = av.asset_id WHERE a.study_id = $1)",
            ),
            (
                "delete asset version relation links for study",
                "DELETE FROM public.asset_version_relation_instances WHERE version_id IN (SELECT av.version_id FROM public.asset_versions av JOIN public.assets a ON a.asset_id = av.asset_id WHERE a.study_id = $1)",
            ),
            (
                "delete asset version entity links for study",
                "DELETE FROM public.asset_version_entities WHERE version_id IN (SELECT av.version_id FROM public.asset_versions av JOIN public.assets a ON a.asset_id = av.asset_id WHERE a.study_id = $1)",
            ),
            (
                "delete dataset variable links for study",
                "DELETE FROM public.dataset_variables WHERE dataset_id IN (SELECT av.version_id FROM public.asset_versions av JOIN public.assets a ON a.asset_id = av.asset_id WHERE a.study_id = $1)",
            ),
            (
                "delete assets for study",
                "DELETE FROM public.assets WHERE study_id = $1",
            ),
        ] {
            transaction.execute_command(operation, sql, &[&study_id.0])?;
        }
        let deleted = transaction.execute_command(
            "delete study record",
            "DELETE FROM public.studies WHERE study_id = $1",
            &[&study_id.0],
        )?;
        Ok(deleted > 0)
    })
}

fn delete_semantic_record<E>(
    connection: &mut E,
    table: &'static str,
    id_column: &'static str,
    id: i64,
) -> Result<bool, PgMetaError>
where
    E: MetadataExecutor,
{
    let id = i32::try_from(id).map_err(|_| PgMetaError::Decode {
        field: id_column,
        value: id.to_string(),
        message: "identifier does not fit into PostgreSQL INTEGER".to_string(),
    })?;
    let sql = format!("DELETE FROM public.{table} WHERE {id_column} = $1");
    let deleted = connection.execute_command("delete semantic catalog row", &sql, &[&id])?;
    Ok(deleted > 0)
}

fn oauth_clear_asset_version_latest(
    connection: &mut LibpqOAuthConnection,
    asset_id: AssetId,
    version_id: VersionId,
) -> Result<(), PgMetaError> {
    MetadataExecutor::query_one(
        connection,
        "clear asset version latest",
        "UPDATE public.asset_versions SET is_latest = FALSE WHERE asset_id = $1 AND version_id = $2 RETURNING version_id::text",
        &[&asset_id.0, &version_id.0],
    )?;
    Ok(())
}

fn oauth_list_datafiles_for_asset(
    connection: &mut LibpqOAuthConnection,
    asset_id: AssetId,
) -> Result<Vec<PgDataFileReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list datafiles for asset",
        "
        SELECT df.datafile_id::text, df.size_bytes::text, df.compressed::text, df.encrypted::text,
               df.compression_algorithm, df.encryption_algorithm,
               encode(df.encryption_key, 'hex'), df.storage_uri, df.edam_format, df.digest
          FROM public.datafiles df
          JOIN public.asset_versions av ON av.version_id = df.datafile_id
         WHERE av.asset_id = $1
         ORDER BY av.major, av.minor, av.patch
        ",
        &[&asset_id.0],
    )?;
    rows.iter().map(oauth_datafile_from_row).collect()
}

fn oauth_get_datafile(
    connection: &mut LibpqOAuthConnection,
    datafile_id: VersionId,
) -> Result<Option<PgDataFileReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get datafile",
        "
        SELECT datafile_id::text, size_bytes::text, compressed::text, encrypted::text,
               compression_algorithm, encryption_algorithm,
               encode(encryption_key, 'hex'), storage_uri, edam_format, digest
          FROM public.datafiles
         WHERE datafile_id = $1
         LIMIT 1
        ",
        &[&datafile_id.0],
    )?;
    row.as_ref().map(oauth_datafile_from_row).transpose()
}

fn oauth_create_datafile(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateDataFile,
) -> Result<(), PgMetaError> {
    MetadataExecutor::execute_command(
        connection,
        "create datafile",
        "
        INSERT INTO public.datafiles (
            datafile_id, size_bytes, compressed, encrypted, compression_algorithm,
            encryption_algorithm, encryption_key, storage_uri, edam_format, digest
        )
        VALUES (
            $1, $2, COALESCE($3, FALSE), COALESCE($4, FALSE),
            COALESCE($5, 'zstd'), COALESCE($6, 'AES-256-CBC with PBKDF2'),
            $7, $8, $9, $10
        )
        ",
        &[
            &input.datafile_id.0,
            &input.size_bytes,
            &input.compressed,
            &input.encrypted,
            &input.compression_algorithm,
            &input.encryption_algorithm,
            &input.encryption_key,
            &input.storage_uri,
            &input.edam_format,
            &input.digest,
        ],
    )?;
    Ok(())
}

fn oauth_list_value_types(
    connection: &mut LibpqOAuthConnection,
) -> Result<Vec<PgValueTypeReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list value types",
        "
        SELECT value_type_id::text, value_type, description
          FROM public.value_types
         ORDER BY value_type_id
        ",
        &[],
    )?;
    rows.iter().map(oauth_value_type_from_row).collect()
}

fn oauth_list_variables_for_domain(
    connection: &mut LibpqOAuthConnection,
    domain_id: DomainId,
) -> Result<Vec<PgVariableReadModel>, PgMetaError> {
    let domain_id = metadata_i32(domain_id.0, "domain_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list domain variables",
        "
        SELECT v.variable_id::text, v.domain_id::text, v.name, v.value_type_id::text,
               vt.value_type, v.value_format, v.vocabulary_id::text,
               v.keyrole::text, v.description, v.note,
               v.ontology_namespace, v.ontology_class
          FROM public.variables v
          JOIN public.value_types vt ON vt.value_type_id = v.value_type_id
         WHERE v.domain_id = $1
         ORDER BY v.name
        ",
        &[&domain_id],
    )?;
    rows.iter().map(oauth_variable_from_row).collect()
}

fn oauth_get_variable(
    connection: &mut LibpqOAuthConnection,
    variable_id: VariableId,
) -> Result<Option<PgVariableReadModel>, PgMetaError> {
    let variable_id = metadata_i32(variable_id.0, "variable_id")?;
    let row = MetadataExecutor::query_optional(
        connection,
        "get variable",
        "
        SELECT v.variable_id::text, v.domain_id::text, v.name, v.value_type_id::text,
               vt.value_type, v.value_format, v.vocabulary_id::text,
               v.keyrole::text, v.description, v.note,
               v.ontology_namespace, v.ontology_class
          FROM public.variables v
          JOIN public.value_types vt ON vt.value_type_id = v.value_type_id
         WHERE v.variable_id = $1
         LIMIT 1
        ",
        &[&variable_id],
    )?;
    row.as_ref().map(oauth_variable_from_row).transpose()
}

fn oauth_create_variable(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateVariable,
) -> Result<VariableId, PgMetaError> {
    let key_role = key_role_sql(input.key_role)?;
    let domain_id = metadata_i32(input.domain_id.0, "domain_id")?;
    let value_type_id = metadata_i32(input.value_type_id.0, "value_type_id")?;
    let vocabulary_id = optional_metadata_i32(input.vocabulary_id.map(|id| id.0), "vocabulary_id")?;
    let row = match input.variable_id {
        Some(variable_id) => {
            let variable_id = metadata_i32(variable_id.0, "variable_id")?;
            MetadataExecutor::query_one(
                connection,
                "create variable",
                "
                INSERT INTO public.variables (
                    variable_id, domain_id, name, value_type_id, value_format, vocabulary_id,
                    keyrole, description, note, ontology_namespace, ontology_class
                )
                VALUES ($1, $2, $3, $4, $5, $6, CAST($7 AS text)::public.variable_keyrole_enum,
                        $8, $9, $10, $11)
                RETURNING variable_id::text
                ",
                &[
                    &variable_id,
                    &domain_id,
                    &input.name,
                    &value_type_id,
                    &input.value_format,
                    &vocabulary_id,
                    &key_role,
                    &input.description,
                    &input.note,
                    &input.ontology_namespace,
                    &input.ontology_class,
                ],
            )?
        }
        None => MetadataExecutor::query_one(
            connection,
            "create variable",
            "
            INSERT INTO public.variables (
                domain_id, name, value_type_id, value_format, vocabulary_id,
                keyrole, description, note, ontology_namespace, ontology_class
            )
            VALUES ($1, $2, $3, $4, $5, CAST($6 AS text)::public.variable_keyrole_enum,
                    $7, $8, $9, $10)
            RETURNING variable_id::text
            ",
            &[
                &domain_id,
                &input.name,
                &value_type_id,
                &input.value_format,
                &vocabulary_id,
                &key_role,
                &input.description,
                &input.note,
                &input.ontology_namespace,
                &input.ontology_class,
            ],
        )?,
    };
    Ok(VariableId(oauth_i64(&row, 0, "variable_id")?))
}

fn oauth_update_variable(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateVariable,
) -> Result<(), PgMetaError> {
    let variable_id = input.variable_id.ok_or_else(|| PgMetaError::Decode {
        field: "variable_id",
        value: "NULL".to_string(),
        message: "variable update requires a variable_id".to_string(),
    })?;
    let variable_id = metadata_i32(variable_id.0, "variable_id")?;
    let domain_id = metadata_i32(input.domain_id.0, "domain_id")?;
    let value_type_id = metadata_i32(input.value_type_id.0, "value_type_id")?;
    let vocabulary_id = optional_metadata_i32(input.vocabulary_id.map(|id| id.0), "vocabulary_id")?;
    let key_role = key_role_sql(input.key_role)?;
    MetadataExecutor::execute_command(
        connection,
        "update variable",
        "
        UPDATE public.variables
           SET domain_id = $2,
               name = $3,
               value_type_id = $4,
               value_format = $5,
               vocabulary_id = $6,
               keyrole = CAST($7 AS text)::public.variable_keyrole_enum,
               description = $8,
               note = $9,
               ontology_namespace = $10,
               ontology_class = $11
         WHERE variable_id = $1
        ",
        &[
            &variable_id,
            &domain_id,
            &input.name,
            &value_type_id,
            &input.value_format,
            &vocabulary_id,
            &key_role,
            &input.description,
            &input.note,
            &input.ontology_namespace,
            &input.ontology_class,
        ],
    )?;
    Ok(())
}

fn oauth_link_dataset_variable(
    connection: &mut LibpqOAuthConnection,
    dataset_id: VersionId,
    variable_id: VariableId,
    row_role: KeyRole,
) -> Result<(), PgMetaError> {
    let variable_id = metadata_i32(variable_id.0, "variable_id")?;
    let row_role = key_role_sql(row_role)?;
    MetadataExecutor::execute_command(
        connection,
        "link dataset variable",
        "
        INSERT INTO public.dataset_variables (dataset_id, variable_id, row_role)
        VALUES ($1, $2, CAST($3 AS text)::public.variable_keyrole_enum)
        ON CONFLICT (dataset_id, variable_id) DO UPDATE
          SET row_role = EXCLUDED.row_role
        ",
        &[&dataset_id.0, &variable_id, &row_role],
    )?;
    Ok(())
}

fn oauth_list_dataset_variables(
    connection: &mut LibpqOAuthConnection,
    dataset_id: VersionId,
) -> Result<Vec<PgDatasetVariableReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list dataset variables",
        OAUTH_LIST_DATASET_VARIABLES_SQL,
        &[&dataset_id.0],
    )?;
    rows.iter().map(oauth_dataset_variable_from_row).collect()
}

fn oauth_list_dataset_variables_for_variable(
    connection: &mut LibpqOAuthConnection,
    variable_id: VariableId,
) -> Result<Vec<PgDatasetVariableReadModel>, PgMetaError> {
    let variable_id = metadata_i32(variable_id.0, "variable_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list dataset variables for variable",
        "
        SELECT dataset_id::text, variable_id::text, row_role::text
          FROM public.dataset_variables
         WHERE variable_id = $1
         ORDER BY dataset_id
        ",
        &[&variable_id],
    )?;
    rows.iter().map(oauth_dataset_variable_from_row).collect()
}

fn oauth_list_dataset_version_entities(
    connection: &mut LibpqOAuthConnection,
    version_id: VersionId,
) -> Result<Vec<PgDatasetVersionEntityReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list dataset version entities",
        "
        SELECT version_id::text, entity_instance_id::text, entity_variable_id::text,
               transformation_id::text
          FROM public.dataset_version_entities
         WHERE version_id = $1
         ORDER BY entity_instance_id
        ",
        &[&version_id.0],
    )?;
    rows.iter()
        .map(oauth_dataset_version_entity_from_row)
        .collect()
}

fn oauth_list_dataset_version_entities_for_instance(
    connection: &mut LibpqOAuthConnection,
    entity_instance_id: i64,
) -> Result<Vec<PgDatasetVersionEntityReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list dataset version entities for instance",
        "
        SELECT version_id::text, entity_instance_id::text, entity_variable_id::text,
               transformation_id::text
          FROM public.dataset_version_entities
         WHERE entity_instance_id = $1
         ORDER BY version_id, entity_variable_id
        ",
        &[&entity_instance_id],
    )?;
    rows.iter()
        .map(oauth_dataset_version_entity_from_row)
        .collect()
}

fn oauth_list_dataset_version_relation_instances(
    connection: &mut LibpqOAuthConnection,
    version_id: VersionId,
) -> Result<Vec<PgDatasetVersionRelationInstanceReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list dataset version relation instances",
        "
        SELECT version_id::text, relation_instance_id::text, subject_variable_id::text,
               object_variable_id::text, relation_variable_id::text, transformation_id::text
          FROM public.dataset_version_relation_instances
         WHERE version_id = $1
         ORDER BY relation_instance_id
        ",
        &[&version_id.0],
    )?;
    rows.iter()
        .map(oauth_dataset_version_relation_instance_from_row)
        .collect()
}

fn oauth_list_dataset_version_relation_instances_for_instance(
    connection: &mut LibpqOAuthConnection,
    relation_instance_id: i64,
) -> Result<Vec<PgDatasetVersionRelationInstanceReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list dataset version relation instances for instance",
        "
        SELECT version_id::text, relation_instance_id::text, subject_variable_id::text,
               object_variable_id::text, relation_variable_id::text, transformation_id::text
          FROM public.dataset_version_relation_instances
         WHERE relation_instance_id = $1
         ORDER BY version_id
        ",
        &[&relation_instance_id],
    )?;
    rows.iter()
        .map(oauth_dataset_version_relation_instance_from_row)
        .collect()
}

fn oauth_list_vocabularies_for_domain(
    connection: &mut LibpqOAuthConnection,
    domain_id: DomainId,
) -> Result<Vec<PgVocabularyReadModel>, PgMetaError> {
    let domain_id = metadata_i32(domain_id.0, "domain_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list vocabularies",
        "
        SELECT vocabulary_id::text, domain_id::text, name, description
          FROM public.vocabularies
         WHERE domain_id = $1
         ORDER BY name
        ",
        &[&domain_id],
    )?;
    rows.iter().map(oauth_vocabulary_from_row).collect()
}

fn oauth_get_vocabulary(
    connection: &mut LibpqOAuthConnection,
    vocabulary_id: VocabularyId,
) -> Result<Option<PgVocabularyReadModel>, PgMetaError> {
    let vocabulary_id = metadata_i32(vocabulary_id.0, "vocabulary_id")?;
    let row = MetadataExecutor::query_optional(
        connection,
        "get vocabulary",
        "
        SELECT vocabulary_id::text, domain_id::text, name, description
          FROM public.vocabularies
         WHERE vocabulary_id = $1
         LIMIT 1
        ",
        &[&vocabulary_id],
    )?;
    row.as_ref().map(oauth_vocabulary_from_row).transpose()
}

fn oauth_create_vocabulary(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateVocabulary,
) -> Result<VocabularyId, PgMetaError> {
    let domain_id = metadata_i32(input.domain_id.0, "domain_id")?;
    let row = match input.vocabulary_id {
        Some(vocabulary_id) => {
            let vocabulary_id = metadata_i32(vocabulary_id.0, "vocabulary_id")?;
            MetadataExecutor::query_one(
                connection,
                "create vocabulary",
                "
                INSERT INTO public.vocabularies (vocabulary_id, domain_id, name, description)
                VALUES ($1, $2, $3, $4)
                RETURNING vocabulary_id::text
                ",
                &[&vocabulary_id, &domain_id, &input.name, &input.description],
            )?
        }
        None => MetadataExecutor::query_one(
            connection,
            "create vocabulary",
            "
            INSERT INTO public.vocabularies (domain_id, name, description)
            VALUES ($1, $2, $3)
            RETURNING vocabulary_id::text
            ",
            &[&domain_id, &input.name, &input.description],
        )?,
    };
    Ok(VocabularyId(oauth_i64(&row, 0, "vocabulary_id")?))
}

fn oauth_update_vocabulary(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateVocabulary,
) -> Result<(), PgMetaError> {
    let vocabulary_id = input.vocabulary_id.ok_or_else(|| PgMetaError::Decode {
        field: "vocabulary_id",
        value: "NULL".to_string(),
        message: "vocabulary update requires a vocabulary_id".to_string(),
    })?;
    let vocabulary_id = metadata_i32(vocabulary_id.0, "vocabulary_id")?;
    let domain_id = metadata_i32(input.domain_id.0, "domain_id")?;
    MetadataExecutor::execute_command(
        connection,
        "update vocabulary",
        "
        UPDATE public.vocabularies
           SET domain_id = $2, name = $3, description = $4
         WHERE vocabulary_id = $1
        ",
        &[&vocabulary_id, &domain_id, &input.name, &input.description],
    )?;
    Ok(())
}

fn oauth_list_vocabulary_items(
    connection: &mut LibpqOAuthConnection,
    vocabulary_id: VocabularyId,
) -> Result<Vec<PgVocabularyItemReadModel>, PgMetaError> {
    let vocabulary_id = metadata_i32(vocabulary_id.0, "vocabulary_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list vocabulary items",
        "
        SELECT vocabulary_item_id::text, vocabulary_id::text, value::text, code, description
          FROM public.vocabulary_items
         WHERE vocabulary_id = $1
         ORDER BY value, code
        ",
        &[&vocabulary_id],
    )?;
    rows.iter().map(oauth_vocabulary_item_from_row).collect()
}

fn oauth_get_vocabulary_item(
    connection: &mut LibpqOAuthConnection,
    vocabulary_item_id: VocabularyItemId,
) -> Result<Option<PgVocabularyItemReadModel>, PgMetaError> {
    let vocabulary_item_id = metadata_i32(vocabulary_item_id.0, "vocabulary_item_id")?;
    let row = MetadataExecutor::query_optional(
        connection,
        "get vocabulary item",
        "
        SELECT vocabulary_item_id::text, vocabulary_id::text, value::text, code, description
          FROM public.vocabulary_items
         WHERE vocabulary_item_id = $1
         LIMIT 1
        ",
        &[&vocabulary_item_id],
    )?;
    row.as_ref().map(oauth_vocabulary_item_from_row).transpose()
}

fn oauth_save_vocabulary_items(
    connection: &mut LibpqOAuthConnection,
    items: &[VocabularyItemRecord],
) -> Result<(), PgMetaError> {
    connection.with_transaction("create vocabulary items", |transaction| {
        for item in items {
            let vocabulary_item_id = metadata_i32(item.vocabulary_item_id.0, "vocabulary_item_id")?;
            let vocabulary_id = metadata_i32(item.vocabulary_id.0, "vocabulary_id")?;
            transaction.execute_command(
                "create vocabulary item",
                "
                INSERT INTO public.vocabulary_items (
                    vocabulary_item_id, vocabulary_id, value, code, description
                )
                VALUES ($1, $2, $3, $4, $5)
                ",
                &[
                    &vocabulary_item_id,
                    &vocabulary_id,
                    &item.value,
                    &item.code,
                    &item.description,
                ],
            )?;
        }
        Ok(())
    })
}

fn oauth_create_vocabulary_items(
    connection: &mut LibpqOAuthConnection,
    vocabulary_id: VocabularyId,
    items: &[NewVocabularyItemRecord],
) -> Result<Vec<VocabularyItemRecord>, PgMetaError> {
    connection.with_transaction("create vocabulary items", |transaction| {
        let vocabulary_id_i32 = metadata_i32(vocabulary_id.0, "vocabulary_id")?;
        let mut created = Vec::with_capacity(items.len());
        for item in items {
            let row = transaction.query_one(
                "create vocabulary item",
                "
                INSERT INTO public.vocabulary_items (vocabulary_id, value, code, description)
                VALUES ($1, $2, $3, $4)
                RETURNING vocabulary_item_id::text
                ",
                &[
                    &vocabulary_id_i32,
                    &item.value,
                    &item.code,
                    &item.description,
                ],
            )?;
            created.push(VocabularyItemRecord {
                vocabulary_item_id: VocabularyItemId(oauth_i64(&row, 0, "vocabulary_item_id")?),
                vocabulary_id,
                value: item.value,
                code: item.code.clone(),
                description: item.description.clone(),
            });
        }
        Ok(created)
    })
}

fn oauth_update_vocabulary_items(
    connection: &mut LibpqOAuthConnection,
    items: &[VocabularyItemRecord],
) -> Result<(), PgMetaError> {
    connection.with_transaction("update vocabulary items", |transaction| {
        for item in items {
            let vocabulary_item_id = metadata_i32(item.vocabulary_item_id.0, "vocabulary_item_id")?;
            let vocabulary_id = metadata_i32(item.vocabulary_id.0, "vocabulary_id")?;
            transaction.execute_command(
                "update vocabulary item",
                "
                UPDATE public.vocabulary_items
                   SET vocabulary_id = $2, value = $3, code = $4, description = $5
                 WHERE vocabulary_item_id = $1
                ",
                &[
                    &vocabulary_item_id,
                    &vocabulary_id,
                    &item.value,
                    &item.code,
                    &item.description,
                ],
            )?;
        }
        Ok(())
    })
}

fn oauth_list_vocabulary_mappings(
    connection: &mut LibpqOAuthConnection,
) -> Result<Vec<crate::PgVocabularyMappingReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list vocabulary mappings",
        "
        SELECT vocabulary_mapping_id::text, from_vocabulary_item::text, to_vocabulary_item::text
          FROM public.vocabulary_mapping
         ORDER BY vocabulary_mapping_id
        ",
        &[],
    )?;
    rows.iter().map(oauth_vocabulary_mapping_from_row).collect()
}

fn oauth_create_vocabulary_mapping(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateVocabularyMapping,
) -> Result<i64, PgMetaError> {
    let from_vocabulary_item = metadata_i32(input.from_vocabulary_item.0, "from_vocabulary_item")?;
    let to_vocabulary_item = metadata_i32(input.to_vocabulary_item.0, "to_vocabulary_item")?;
    let row = match input.vocabulary_mapping_id {
        Some(vocabulary_mapping_id) => {
            let vocabulary_mapping_id =
                metadata_i32(vocabulary_mapping_id, "vocabulary_mapping_id")?;
            MetadataExecutor::query_one(
                connection,
                "create vocabulary mapping",
                "
                INSERT INTO public.vocabulary_mapping (
                    vocabulary_mapping_id, from_vocabulary_item, to_vocabulary_item
                )
                VALUES ($1, $2, $3)
                RETURNING vocabulary_mapping_id::text
                ",
                &[
                    &vocabulary_mapping_id,
                    &from_vocabulary_item,
                    &to_vocabulary_item,
                ],
            )?
        }
        None => MetadataExecutor::query_one(
            connection,
            "create vocabulary mapping",
            "
            INSERT INTO public.vocabulary_mapping (from_vocabulary_item, to_vocabulary_item)
            VALUES ($1, $2)
            RETURNING vocabulary_mapping_id::text
            ",
            &[&from_vocabulary_item, &to_vocabulary_item],
        )?,
    };
    oauth_i64(&row, 0, "vocabulary_mapping_id")
}

fn oauth_list_entities_for_domain(
    connection: &mut LibpqOAuthConnection,
    domain_id: DomainId,
) -> Result<Vec<PgEntityReadModel>, PgMetaError> {
    let domain_id = metadata_i32(domain_id.0, "domain_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list entities",
        "
        SELECT entity_id::text, domain_id::text, uuid::text, name, description,
               agent_instructions, ontology_namespace, ontology_class
          FROM public.entities
         WHERE domain_id = $1
         ORDER BY name
        ",
        &[&domain_id],
    )?;
    rows.iter().map(oauth_entity_from_row).collect()
}

fn oauth_create_entity(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateEntity,
) -> Result<EntityId, PgMetaError> {
    let domain_id = metadata_i32(input.domain_id.0, "domain_id")?;
    let row = match input.entity_id {
        Some(entity_id) => {
            let entity_id = metadata_i32(entity_id.0, "entity_id")?;
            MetadataExecutor::query_one(
                connection,
                "create entity",
                "
                INSERT INTO public.entities (
                    entity_id, domain_id, uuid, name, description,
                    agent_instructions, ontology_namespace, ontology_class
                )
                VALUES ($1, $2, COALESCE($3, uuidv7()), $4, $5, $6, $7, $8)
                RETURNING entity_id::text
                ",
                &[
                    &entity_id,
                    &domain_id,
                    &input.uuid,
                    &input.name,
                    &input.description,
                    &input.agent_instructions,
                    &input.ontology_namespace,
                    &input.ontology_class,
                ],
            )?
        }
        None => MetadataExecutor::query_one(
            connection,
            "create entity",
            "
            INSERT INTO public.entities (
                domain_id, uuid, name, description, agent_instructions,
                ontology_namespace, ontology_class
            )
            VALUES ($1, COALESCE($2, uuidv7()), $3, $4, $5, $6, $7)
            RETURNING entity_id::text
            ",
            &[
                &domain_id,
                &input.uuid,
                &input.name,
                &input.description,
                &input.agent_instructions,
                &input.ontology_namespace,
                &input.ontology_class,
            ],
        )?,
    };
    Ok(EntityId(oauth_i64(&row, 0, "entity_id")?))
}

fn oauth_list_entity_relations_for_domain(
    connection: &mut LibpqOAuthConnection,
    domain_id: DomainId,
) -> Result<Vec<PgEntityRelationReadModel>, PgMetaError> {
    let domain_id = metadata_i32(domain_id.0, "domain_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list entity relations",
        "
        SELECT entityrelation_id::text, subject_entity_id::text, object_entity_id::text,
               domain_id::text, uuid::text, name, description, agent_instructions,
               ontology_namespace, ontology_class
          FROM public.entityrelations
         WHERE domain_id = $1
         ORDER BY name
        ",
        &[&domain_id],
    )?;
    rows.iter().map(oauth_entity_relation_from_row).collect()
}

fn oauth_create_entity_relation(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateEntityRelation,
) -> Result<EntityRelationId, PgMetaError> {
    let subject_entity_id = metadata_i32(input.subject_entity_id.0, "subject_entity_id")?;
    let object_entity_id = metadata_i32(input.object_entity_id.0, "object_entity_id")?;
    let domain_id = metadata_i32(input.domain_id.0, "domain_id")?;
    let row = match input.entity_relation_id {
        Some(entity_relation_id) => {
            let entity_relation_id = metadata_i32(entity_relation_id.0, "entityrelation_id")?;
            MetadataExecutor::query_one(
                connection,
                "create entity relation",
                "
                INSERT INTO public.entityrelations (
                    entityrelation_id, subject_entity_id, object_entity_id,
                    domain_id, uuid, name, description, agent_instructions,
                    ontology_namespace, ontology_class
                )
                VALUES ($1, $2, $3, $4, COALESCE($5, uuidv7()), $6, $7, $8, $9, $10)
                RETURNING entityrelation_id::text
                ",
                &[
                    &entity_relation_id,
                    &subject_entity_id,
                    &object_entity_id,
                    &domain_id,
                    &input.uuid,
                    &input.name,
                    &input.description,
                    &input.agent_instructions,
                    &input.ontology_namespace,
                    &input.ontology_class,
                ],
            )?
        }
        None => MetadataExecutor::query_one(
            connection,
            "create entity relation",
            "
            INSERT INTO public.entityrelations (
                subject_entity_id, object_entity_id, domain_id, uuid, name, description,
                agent_instructions, ontology_namespace, ontology_class
            )
            VALUES ($1, $2, $3, COALESCE($4, uuidv7()), $5, $6, $7, $8, $9)
            RETURNING entityrelation_id::text
            ",
            &[
                &subject_entity_id,
                &object_entity_id,
                &domain_id,
                &input.uuid,
                &input.name,
                &input.description,
                &input.agent_instructions,
                &input.ontology_namespace,
                &input.ontology_class,
            ],
        )?,
    };
    Ok(EntityRelationId(oauth_i64(&row, 0, "entityrelation_id")?))
}

fn oauth_get_entity_instance(
    connection: &mut LibpqOAuthConnection,
    instance_id: i64,
) -> Result<Option<PgEntityInstanceReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get entity instance",
        "
        SELECT instance_id::text, uuid::text, entity_id::text, label, note, transformation_id::text
          FROM public.entity_instances
         WHERE instance_id = $1
         LIMIT 1
        ",
        &[&instance_id],
    )?;
    row.as_ref().map(oauth_entity_instance_from_row).transpose()
}

fn oauth_get_entity_instance_by_uuid(
    connection: &mut LibpqOAuthConnection,
    uuid: uuid::Uuid,
) -> Result<Option<PgEntityInstanceReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get entity instance by uuid",
        "
        SELECT instance_id::text, uuid::text, entity_id::text, label, note, transformation_id::text
          FROM public.entity_instances
         WHERE uuid = $1
         LIMIT 1
        ",
        &[&uuid],
    )?;
    row.as_ref().map(oauth_entity_instance_from_row).transpose()
}

fn oauth_list_entity_instances_for_entity(
    connection: &mut LibpqOAuthConnection,
    entity_id: EntityId,
) -> Result<Vec<PgEntityInstanceReadModel>, PgMetaError> {
    let entity_id = metadata_i32(entity_id.0, "entity_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list entity instances",
        "
        SELECT instance_id::text, uuid::text, entity_id::text, label, note, transformation_id::text
          FROM public.entity_instances
         WHERE entity_id = $1
         ORDER BY instance_id
        ",
        &[&entity_id],
    )?;
    rows.iter().map(oauth_entity_instance_from_row).collect()
}

fn oauth_save_entity_instance(
    connection: &mut LibpqOAuthConnection,
    input: PgWriteEntityInstance,
) -> Result<i64, PgMetaError> {
    let entity_id = metadata_i32(input.entity_id.0, "entity_id")?;
    let transformation_id = metadata_i32(input.transformation_id.0, "transformation_id")?;
    let row = match input.instance_id {
        Some(instance_id) => MetadataExecutor::query_one(
            connection,
            "save entity instance",
            "
            UPDATE public.entity_instances
               SET entity_id = $2,
                   label = $3,
                   note = $4,
                   transformation_id = $5
             WHERE instance_id = $1
               AND ($6::uuid IS NULL OR uuid = $6)
            RETURNING instance_id::text
            ",
            &[
                &instance_id,
                &entity_id,
                &input.label,
                &input.note,
                &transformation_id,
                &input.uuid,
            ],
        )?,
        None => MetadataExecutor::query_one(
            connection,
            "save entity instance",
            "
            INSERT INTO public.entity_instances (uuid, entity_id, label, note, transformation_id)
            VALUES (COALESCE($1, uuidv7()), $2, $3, $4, $5)
            RETURNING instance_id::text
            ",
            &[
                &input.uuid,
                &entity_id,
                &input.label,
                &input.note,
                &transformation_id,
            ],
        )?,
    };
    oauth_i64(&row, 0, "instance_id")
}

fn oauth_get_study_entity_instance(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
    entity_id: EntityId,
    external_id: &str,
) -> Result<Option<PgStudyEntityInstanceReadModel>, PgMetaError> {
    let entity_id = metadata_i32(entity_id.0, "entity_id")?;
    let row = MetadataExecutor::query_optional(
        connection,
        "get study entity instance",
        "
        SELECT study_id::text, entity_instance_id::text, entity_id::text,
               external_id, transformation_id::text
          FROM public.study_entity_instances
         WHERE study_id = $1 AND entity_id = $2 AND external_id = $3
         LIMIT 1
        ",
        &[&study_id.0, &entity_id, &external_id],
    )?;
    row.as_ref()
        .map(oauth_study_entity_instance_from_row)
        .transpose()
}

fn oauth_list_study_entity_instances(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
) -> Result<Vec<PgStudyEntityInstanceReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list study entity instances",
        "
        SELECT study_id::text, entity_instance_id::text, entity_id::text,
               external_id, transformation_id::text
          FROM public.study_entity_instances
         WHERE study_id = $1
         ORDER BY entity_instance_id
        ",
        &[&study_id.0],
    )?;
    rows.iter()
        .map(oauth_study_entity_instance_from_row)
        .collect()
}

fn oauth_save_study_entity_instance(
    connection: &mut LibpqOAuthConnection,
    input: PgWriteStudyEntityInstance,
) -> Result<(), PgMetaError> {
    let entity_id = metadata_i32(input.entity_id.0, "entity_id")?;
    let transformation_id = metadata_i32(input.transformation_id.0, "transformation_id")?;
    MetadataExecutor::execute_command(
        connection,
        "save study entity instance",
        "
        INSERT INTO public.study_entity_instances (
            study_id, entity_instance_id, entity_id, external_id, transformation_id
        )
        VALUES ($1, $2, $3, $4, $5)
        ON CONFLICT (study_id, entity_instance_id) DO UPDATE
          SET entity_id = EXCLUDED.entity_id,
              external_id = EXCLUDED.external_id,
              transformation_id = EXCLUDED.transformation_id
        ",
        &[
            &input.study_id.0,
            &input.entity_instance_id,
            &entity_id,
            &input.external_id,
            &transformation_id,
        ],
    )?;
    Ok(())
}

fn oauth_get_relation_instance(
    connection: &mut LibpqOAuthConnection,
    relation_instance_id: i64,
) -> Result<Option<PgRelationInstanceReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get relation instance",
        "
        SELECT relation_instance_id::text, uuid::text, entityrelation_id::text,
               entity_instance_id_1::text, entity_instance_id_2::text,
               valid_from::text, valid_to::text, note, transformation_id::text
          FROM public.relation_instances
         WHERE relation_instance_id = $1
         LIMIT 1
        ",
        &[&relation_instance_id],
    )?;
    row.as_ref()
        .map(oauth_relation_instance_from_row)
        .transpose()
}

fn oauth_get_relation_instance_by_uuid(
    connection: &mut LibpqOAuthConnection,
    uuid: uuid::Uuid,
) -> Result<Option<PgRelationInstanceReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get relation instance by uuid",
        "
        SELECT relation_instance_id::text, uuid::text, entityrelation_id::text,
               entity_instance_id_1::text, entity_instance_id_2::text,
               valid_from::text, valid_to::text, note, transformation_id::text
          FROM public.relation_instances
         WHERE uuid = $1
         LIMIT 1
        ",
        &[&uuid],
    )?;
    row.as_ref()
        .map(oauth_relation_instance_from_row)
        .transpose()
}

fn oauth_list_relation_instances_for_relation(
    connection: &mut LibpqOAuthConnection,
    relation_id: EntityRelationId,
) -> Result<Vec<PgRelationInstanceReadModel>, PgMetaError> {
    let relation_id = metadata_i32(relation_id.0, "entityrelation_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list relation instances",
        "
        SELECT relation_instance_id::text, uuid::text, entityrelation_id::text,
               entity_instance_id_1::text, entity_instance_id_2::text,
               valid_from::text, valid_to::text, note, transformation_id::text
          FROM public.relation_instances
         WHERE entityrelation_id = $1
         ORDER BY relation_instance_id
        ",
        &[&relation_id],
    )?;
    rows.iter().map(oauth_relation_instance_from_row).collect()
}

fn oauth_save_relation_instance(
    connection: &mut LibpqOAuthConnection,
    input: PgWriteRelationInstance,
) -> Result<i64, PgMetaError> {
    let entity_relation_id = metadata_i32(input.entity_relation_id.0, "entityrelation_id")?;
    let transformation_id = metadata_i32(input.transformation_id.0, "transformation_id")?;
    let row = match input.relation_instance_id {
        Some(relation_instance_id) => MetadataExecutor::query_one(
            connection,
            "save relation instance",
            "
            UPDATE public.relation_instances
               SET entityrelation_id = $2,
                   entity_instance_id_1 = $3,
                   entity_instance_id_2 = $4,
                   valid_from = $5,
                   valid_to = $6,
                   note = $7,
                   transformation_id = $8
             WHERE relation_instance_id = $1
               AND ($9::uuid IS NULL OR uuid = $9)
            RETURNING relation_instance_id::text
            ",
            &[
                &relation_instance_id,
                &entity_relation_id,
                &input.entity_instance_id_1,
                &input.entity_instance_id_2,
                &input.valid_from,
                &input.valid_to,
                &input.note,
                &transformation_id,
                &input.uuid,
            ],
        )?,
        None => MetadataExecutor::query_one(
            connection,
            "save relation instance",
            "
            INSERT INTO public.relation_instances (
                uuid, entityrelation_id, entity_instance_id_1, entity_instance_id_2,
                valid_from, valid_to, note, transformation_id
            )
            VALUES (COALESCE($1, uuidv7()), $2, $3, $4, $5, $6, $7, $8)
            RETURNING relation_instance_id::text
            ",
            &[
                &input.uuid,
                &entity_relation_id,
                &input.entity_instance_id_1,
                &input.entity_instance_id_2,
                &input.valid_from,
                &input.valid_to,
                &input.note,
                &transformation_id,
            ],
        )?,
    };
    oauth_i64(&row, 0, "relation_instance_id")
}

fn oauth_get_study_relation_instance(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
    relation_id: EntityRelationId,
    external_id: &str,
) -> Result<Option<PgStudyRelationInstanceReadModel>, PgMetaError> {
    let relation_id = metadata_i32(relation_id.0, "entityrelation_id")?;
    let row = MetadataExecutor::query_optional(
        connection,
        "get study relation instance",
        "
        SELECT study_id::text, relation_instance_id::text, entityrelation_id::text,
               external_id, transformation_id::text
          FROM public.study_relation_instances
         WHERE study_id = $1 AND entityrelation_id = $2 AND external_id = $3
         LIMIT 1
        ",
        &[&study_id.0, &relation_id, &external_id],
    )?;
    row.as_ref()
        .map(oauth_study_relation_instance_from_row)
        .transpose()
}

fn oauth_list_study_relation_instances(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
) -> Result<Vec<PgStudyRelationInstanceReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list study relation instances",
        "
        SELECT study_id::text, relation_instance_id::text, entityrelation_id::text,
               external_id, transformation_id::text
          FROM public.study_relation_instances
         WHERE study_id = $1
         ORDER BY relation_instance_id
        ",
        &[&study_id.0],
    )?;
    rows.iter()
        .map(oauth_study_relation_instance_from_row)
        .collect()
}

fn oauth_save_study_relation_instance(
    connection: &mut LibpqOAuthConnection,
    input: PgWriteStudyRelationInstance,
) -> Result<(), PgMetaError> {
    let entity_relation_id = metadata_i32(input.entity_relation_id.0, "entityrelation_id")?;
    let transformation_id = metadata_i32(input.transformation_id.0, "transformation_id")?;
    MetadataExecutor::execute_command(
        connection,
        "save study relation instance",
        "
        INSERT INTO public.study_relation_instances (
            study_id, relation_instance_id, entityrelation_id,
            external_id, transformation_id
        )
        VALUES ($1, $2, $3, $4, $5)
        ON CONFLICT (study_id, relation_instance_id) DO UPDATE
          SET entityrelation_id = EXCLUDED.entityrelation_id,
              external_id = EXCLUDED.external_id,
              transformation_id = EXCLUDED.transformation_id
        ",
        &[
            &input.study_id.0,
            &input.relation_instance_id,
            &entity_relation_id,
            &input.external_id,
            &transformation_id,
        ],
    )?;
    Ok(())
}

fn oauth_link_asset_version_entity(
    connection: &mut LibpqOAuthConnection,
    link: &AssetVersionEntityRecord,
) -> Result<(), PgMetaError> {
    let transformation_id = metadata_i32(link.transformation_id.0, "transformation_id")?;
    MetadataExecutor::execute_command(
        connection,
        "link asset version entity",
        "
        INSERT INTO public.asset_version_entities (
            version_id, entity_instance_id, transformation_id
        )
        VALUES ($1, $2, $3)
        ON CONFLICT (version_id, entity_instance_id) DO UPDATE
          SET transformation_id = EXCLUDED.transformation_id
        ",
        &[
            &link.version_id.0,
            &link.entity_instance_id,
            &transformation_id,
        ],
    )?;
    Ok(())
}

fn oauth_list_asset_version_entities(
    connection: &mut LibpqOAuthConnection,
    version_id: VersionId,
) -> Result<Vec<PgAssetVersionEntityReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list asset version entities",
        "
        SELECT version_id::text, entity_instance_id::text, transformation_id::text
          FROM public.asset_version_entities
         WHERE version_id = $1
         ORDER BY entity_instance_id
        ",
        &[&version_id.0],
    )?;
    rows.iter()
        .map(oauth_asset_version_entity_from_row)
        .collect()
}

fn oauth_list_asset_version_entities_for_instance(
    connection: &mut LibpqOAuthConnection,
    entity_instance_id: i64,
) -> Result<Vec<PgAssetVersionEntityReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list asset version entities for instance",
        "
        SELECT version_id::text, entity_instance_id::text, transformation_id::text
          FROM public.asset_version_entities
         WHERE entity_instance_id = $1
         ORDER BY version_id
        ",
        &[&entity_instance_id],
    )?;
    rows.iter()
        .map(oauth_asset_version_entity_from_row)
        .collect()
}

fn oauth_link_dataset_version_entity(
    connection: &mut LibpqOAuthConnection,
    link: &DatasetVersionEntityRecord,
) -> Result<(), PgMetaError> {
    connection.with_transaction("link dataset version entity", |transaction| {
        oauth_link_dataset_version_entity_in_transaction(transaction, link)
    })
}

fn oauth_link_dataset_version_entities(
    connection: &mut LibpqOAuthConnection,
    links: &[DatasetVersionEntityRecord],
) -> Result<(), PgMetaError> {
    connection.with_transaction("bulk link dataset version entities", |transaction| {
        for link in links {
            oauth_link_dataset_version_entity_in_transaction(transaction, link)?;
        }
        Ok(())
    })
}

fn oauth_link_dataset_version_entity_in_transaction(
    transaction: &mut dyn MetadataTransaction<Row = LibpqOAuthRow>,
    link: &DatasetVersionEntityRecord,
) -> Result<(), PgMetaError> {
    let entity_variable_id = metadata_i32(link.entity_variable_id.0, "entity_variable_id")?;
    let transformation_id = metadata_i32(link.transformation_id.0, "transformation_id")?;
    transaction.execute_command(
        "link asset version entity",
        "
        INSERT INTO public.asset_version_entities (
            version_id, entity_instance_id, transformation_id
        )
        VALUES ($1, $2, $3)
        ON CONFLICT (version_id, entity_instance_id) DO UPDATE
          SET transformation_id = EXCLUDED.transformation_id
        ",
        &[
            &link.version_id.0,
            &link.entity_instance_id,
            &transformation_id,
        ],
    )?;
    transaction.execute_command(
        "link dataset version entity",
        "
        INSERT INTO public.dataset_version_entities (
            version_id, entity_instance_id, entity_variable_id, transformation_id
        )
        VALUES ($1, $2, $3, $4)
        ON CONFLICT (version_id, entity_instance_id) DO UPDATE
          SET entity_variable_id = EXCLUDED.entity_variable_id,
              transformation_id = EXCLUDED.transformation_id
        ",
        &[
            &link.version_id.0,
            &link.entity_instance_id,
            &entity_variable_id,
            &transformation_id,
        ],
    )?;
    Ok(())
}

fn oauth_link_asset_version_relation_instance(
    connection: &mut LibpqOAuthConnection,
    link: &AssetVersionRelationInstanceRecord,
) -> Result<(), PgMetaError> {
    let transformation_id = metadata_i32(link.transformation_id.0, "transformation_id")?;
    MetadataExecutor::execute_command(
        connection,
        "link asset version relation instance",
        "
        INSERT INTO public.asset_version_relation_instances (
            version_id, relation_instance_id, transformation_id
        )
        VALUES ($1, $2, $3)
        ON CONFLICT (version_id, relation_instance_id) DO UPDATE
          SET transformation_id = EXCLUDED.transformation_id
        ",
        &[
            &link.version_id.0,
            &link.relation_instance_id,
            &transformation_id,
        ],
    )?;
    Ok(())
}

fn oauth_list_asset_version_relation_instances(
    connection: &mut LibpqOAuthConnection,
    version_id: VersionId,
) -> Result<Vec<PgAssetVersionRelationInstanceReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list asset version relation instances",
        "
        SELECT version_id::text, relation_instance_id::text, transformation_id::text
          FROM public.asset_version_relation_instances
         WHERE version_id = $1
         ORDER BY relation_instance_id
        ",
        &[&version_id.0],
    )?;
    rows.iter()
        .map(oauth_asset_version_relation_instance_from_row)
        .collect()
}

fn oauth_list_asset_version_relation_instances_for_instance(
    connection: &mut LibpqOAuthConnection,
    relation_instance_id: i64,
) -> Result<Vec<PgAssetVersionRelationInstanceReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list asset version relation instances for instance",
        "
        SELECT version_id::text, relation_instance_id::text, transformation_id::text
          FROM public.asset_version_relation_instances
         WHERE relation_instance_id = $1
         ORDER BY version_id
        ",
        &[&relation_instance_id],
    )?;
    rows.iter()
        .map(oauth_asset_version_relation_instance_from_row)
        .collect()
}

fn oauth_link_dataset_version_relation_instance(
    connection: &mut LibpqOAuthConnection,
    link: &DatasetVersionRelationInstanceRecord,
) -> Result<(), PgMetaError> {
    connection.with_transaction("link dataset version relation instance", |transaction| {
        oauth_link_dataset_version_relation_in_transaction(transaction, link)
    })
}

fn oauth_link_dataset_version_relation_instances(
    connection: &mut LibpqOAuthConnection,
    links: &[DatasetVersionRelationInstanceRecord],
) -> Result<(), PgMetaError> {
    connection.with_transaction(
        "bulk link dataset version relation instances",
        |transaction| {
            for link in links {
                oauth_link_dataset_version_relation_in_transaction(transaction, link)?;
            }
            Ok(())
        },
    )
}

fn oauth_link_dataset_version_relation_in_transaction(
    transaction: &mut dyn MetadataTransaction<Row = LibpqOAuthRow>,
    link: &DatasetVersionRelationInstanceRecord,
) -> Result<(), PgMetaError> {
    let subject_variable_id = metadata_i32(link.subject_variable_id.0, "subject_variable_id")?;
    let object_variable_id = metadata_i32(link.object_variable_id.0, "object_variable_id")?;
    let relation_variable_id = optional_metadata_i32(
        link.relation_variable_id.map(|id| id.0),
        "relation_variable_id",
    )?;
    let transformation_id = metadata_i32(link.transformation_id.0, "transformation_id")?;
    transaction.execute_command(
        "link asset version relation",
        "
        INSERT INTO public.asset_version_relation_instances (
            version_id, relation_instance_id, transformation_id
        )
        VALUES ($1, $2, $3)
        ON CONFLICT (version_id, relation_instance_id) DO UPDATE
          SET transformation_id = EXCLUDED.transformation_id
        ",
        &[
            &link.version_id.0,
            &link.relation_instance_id,
            &transformation_id,
        ],
    )?;
    transaction.execute_command(
        "link dataset version relation",
        "
        INSERT INTO public.dataset_version_relation_instances (
            version_id, relation_instance_id, subject_variable_id,
            object_variable_id, relation_variable_id, transformation_id
        )
        VALUES ($1, $2, $3, $4, $5, $6)
        ON CONFLICT (version_id, relation_instance_id) DO UPDATE
          SET subject_variable_id = EXCLUDED.subject_variable_id,
              object_variable_id = EXCLUDED.object_variable_id,
              relation_variable_id = EXCLUDED.relation_variable_id,
              transformation_id = EXCLUDED.transformation_id
        ",
        &[
            &link.version_id.0,
            &link.relation_instance_id,
            &subject_variable_id,
            &object_variable_id,
            &relation_variable_id,
            &transformation_id,
        ],
    )?;
    Ok(())
}

fn oauth_create_transformation(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateTransformation,
) -> Result<TransformationId, PgMetaError> {
    let transformation_type = transformation_type_sql(input.transformation_type);
    let row = match input.transformation_id {
        Some(transformation_id) => {
            let transformation_id = metadata_i32(transformation_id.0, "transformation_id")?;
            MetadataExecutor::query_one(
                connection,
                "create transformation",
                "
                INSERT INTO public.transformations (
                    transformation_id, transformation_type, description,
                    repository_url, commit_hash, file_path, date_created, created_by
                )
                VALUES ($1, CAST($2 AS text)::public.transformation_type_enum, $3, $4, $5, $6,
                        COALESCE($7, CURRENT_TIMESTAMP), COALESCE($8, CURRENT_USER))
                RETURNING transformation_id::text
                ",
                &[
                    &transformation_id,
                    &transformation_type,
                    &input.description,
                    &input.repository_url,
                    &input.commit_hash,
                    &input.file_path,
                    &input.date_created,
                    &input.created_by,
                ],
            )?
        }
        None => MetadataExecutor::query_one(
            connection,
            "create transformation",
            "
            INSERT INTO public.transformations (
                transformation_type, description, repository_url, commit_hash, file_path,
                date_created, created_by
            )
            VALUES (CAST($1 AS text)::public.transformation_type_enum, $2, $3, $4, $5,
                    COALESCE($6, CURRENT_TIMESTAMP), COALESCE($7, CURRENT_USER))
            RETURNING transformation_id::text
            ",
            &[
                &transformation_type,
                &input.description,
                &input.repository_url,
                &input.commit_hash,
                &input.file_path,
                &input.date_created,
                &input.created_by,
            ],
        )?,
    };
    Ok(TransformationId(oauth_i64(&row, 0, "transformation_id")?))
}

fn oauth_get_transformation(
    connection: &mut LibpqOAuthConnection,
    transformation_id: TransformationId,
) -> Result<Option<PgTransformationReadModel>, PgMetaError> {
    let transformation_id = metadata_i32(transformation_id.0, "transformation_id")?;
    let row = MetadataExecutor::query_optional(
        connection,
        "get transformation",
        "
        SELECT transformation_id::text, transformation_type::text,
               description, repository_url, commit_hash, file_path,
               date_created::text, created_by
          FROM public.transformations
         WHERE transformation_id = $1
         LIMIT 1
        ",
        &[&transformation_id],
    )?;
    row.as_ref().map(oauth_transformation_from_row).transpose()
}

fn oauth_add_transformation_inputs(
    connection: &mut LibpqOAuthConnection,
    transformation_id: TransformationId,
    input_ids: &[VersionId],
) -> Result<(), PgMetaError> {
    let transformation_id = metadata_i32(transformation_id.0, "transformation_id")?;
    connection.with_transaction("link transformation inputs", |transaction| {
        for version_id in input_ids {
            transaction.execute_command(
                "link transformation input",
                "
                INSERT INTO public.transformation_inputs (transformation_id, version_id)
                VALUES ($1, $2)
                ON CONFLICT DO NOTHING
                ",
                &[&transformation_id, &version_id.0],
            )?;
        }
        Ok(())
    })
}

fn oauth_delete_transformation(
    connection: &mut LibpqOAuthConnection,
    transformation_id: TransformationId,
) -> Result<bool, PgMetaError> {
    let transformation_id = metadata_i32(transformation_id.0, "transformation_id")?;
    let deleted = MetadataExecutor::execute_command(
        connection,
        "delete transformation",
        "DELETE FROM public.transformations WHERE transformation_id = $1",
        &[&transformation_id],
    )?;
    Ok(deleted > 0)
}

fn oauth_add_transformation_outputs(
    connection: &mut LibpqOAuthConnection,
    transformation_id: TransformationId,
    output_ids: &[VersionId],
) -> Result<(), PgMetaError> {
    let transformation_id = metadata_i32(transformation_id.0, "transformation_id")?;
    connection.with_transaction("link transformation outputs", |transaction| {
        for version_id in output_ids {
            transaction.execute_command(
                "link transformation output",
                "
                INSERT INTO public.transformation_outputs (transformation_id, version_id)
                VALUES ($1, $2)
                ON CONFLICT DO NOTHING
                ",
                &[&transformation_id, &version_id.0],
            )?;
        }
        Ok(())
    })
}

fn oauth_list_transformation_inputs(
    connection: &mut LibpqOAuthConnection,
    transformation_id: TransformationId,
) -> Result<Vec<PgTransformationInputReadModel>, PgMetaError> {
    let transformation_id = metadata_i32(transformation_id.0, "transformation_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list transformation inputs",
        "
        SELECT transformation_id::text, version_id::text
          FROM public.transformation_inputs
         WHERE transformation_id = $1
         ORDER BY version_id
        ",
        &[&transformation_id],
    )?;
    rows.iter()
        .map(oauth_transformation_input_from_row)
        .collect()
}

fn oauth_list_transformation_outputs(
    connection: &mut LibpqOAuthConnection,
    transformation_id: TransformationId,
) -> Result<Vec<PgTransformationOutputReadModel>, PgMetaError> {
    let transformation_id = metadata_i32(transformation_id.0, "transformation_id")?;
    let rows = MetadataExecutor::query_many(
        connection,
        "list transformation outputs",
        "
        SELECT transformation_id::text, version_id::text
          FROM public.transformation_outputs
         WHERE transformation_id = $1
         ORDER BY version_id
        ",
        &[&transformation_id],
    )?;
    rows.iter()
        .map(oauth_transformation_output_from_row)
        .collect()
}

fn oauth_list_transformations_producing_version(
    connection: &mut LibpqOAuthConnection,
    version_id: VersionId,
) -> Result<Vec<PgTransformationReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list transformations producing version",
        "
        SELECT t.transformation_id::text, t.transformation_type::text,
               t.description, t.repository_url, t.commit_hash, t.file_path,
               t.date_created::text, t.created_by
          FROM public.transformations t
          JOIN public.transformation_outputs o
            ON o.transformation_id = t.transformation_id
         WHERE o.version_id = $1
         ORDER BY t.transformation_id
        ",
        &[&version_id.0],
    )?;
    rows.iter().map(oauth_transformation_from_row).collect()
}

fn oauth_list_transformations(
    connection: &mut LibpqOAuthConnection,
) -> Result<Vec<PgTransformationReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list transformations",
        "
        SELECT transformation_id::text, transformation_type::text,
               description, repository_url, commit_hash, file_path,
               date_created::text, created_by
          FROM public.transformations
         ORDER BY transformation_id
        ",
        &[],
    )?;
    rows.iter().map(oauth_transformation_from_row).collect()
}

fn oauth_list_tags(
    connection: &mut LibpqOAuthConnection,
) -> Result<Vec<PgTagReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list tags",
        "
        SELECT tag_id::text, name
          FROM public.tags
         ORDER BY name
        ",
        &[],
    )?;
    rows.iter().map(oauth_tag_from_row).collect()
}

fn oauth_upsert_tag(
    connection: &mut LibpqOAuthConnection,
    normalized_name: &str,
) -> Result<i64, PgMetaError> {
    let row = MetadataExecutor::query_one(
        connection,
        "upsert tag",
        "
        INSERT INTO public.tags (name)
        VALUES ($1)
        ON CONFLICT (name) DO UPDATE SET name = EXCLUDED.name
        RETURNING tag_id::text
        ",
        &[&normalized_name],
    )?;
    oauth_i64(&row, 0, "tag_id")
}

fn oauth_attach_tag(
    connection: &mut LibpqOAuthConnection,
    target: PgTagTarget,
    tag_id: i64,
) -> Result<(), PgMetaError> {
    let tag_id = metadata_i32(tag_id, "tag_id")?;
    let (mapping_table, key_column, key_value) = oauth_tag_target_mapping(target);
    let sql = format!(
        "
        INSERT INTO public.{mapping_table} ({key_column}, tag_id)
        VALUES ($1, $2)
        ON CONFLICT DO NOTHING
        "
    );
    oauth_execute_tag_link(connection, "attach tag", &sql, key_value, tag_id)
}

fn oauth_detach_tag(
    connection: &mut LibpqOAuthConnection,
    target: PgTagTarget,
    tag_id: i64,
) -> Result<(), PgMetaError> {
    let tag_id = metadata_i32(tag_id, "tag_id")?;
    let (mapping_table, key_column, key_value) = oauth_tag_target_mapping(target);
    let sql = format!(
        "
        DELETE FROM public.{mapping_table}
         WHERE {key_column} = $1
           AND tag_id = $2
        "
    );
    oauth_execute_tag_link(connection, "detach tag", &sql, key_value, tag_id)
}

fn oauth_delete_tag_if_unused(
    connection: &mut LibpqOAuthConnection,
    tag_id: i64,
) -> Result<bool, PgMetaError> {
    let tag_id = metadata_i32(tag_id, "tag_id")?;
    let deleted = MetadataExecutor::execute_command(
        connection,
        "delete unused tag",
        "
        DELETE FROM public.tags t
         WHERE t.tag_id = $1
           AND NOT EXISTS (SELECT 1 FROM public.study_tags WHERE tag_id = t.tag_id)
           AND NOT EXISTS (SELECT 1 FROM public.domain_tags WHERE tag_id = t.tag_id)
           AND NOT EXISTS (SELECT 1 FROM public.variable_tags WHERE tag_id = t.tag_id)
           AND NOT EXISTS (SELECT 1 FROM public.asset_tags WHERE tag_id = t.tag_id)
           AND NOT EXISTS (SELECT 1 FROM public.asset_version_tags WHERE tag_id = t.tag_id)
           AND NOT EXISTS (SELECT 1 FROM public.entity_tags WHERE tag_id = t.tag_id)
           AND NOT EXISTS (SELECT 1 FROM public.entityrelation_tags WHERE tag_id = t.tag_id)
        ",
        &[&tag_id],
    )?;
    Ok(deleted == 1)
}

fn oauth_execute_tag_link(
    connection: &mut LibpqOAuthConnection,
    operation: &'static str,
    sql: &str,
    key_value: OauthTagKey,
    tag_id: i32,
) -> Result<(), PgMetaError> {
    match key_value {
        OauthTagKey::Uuid(value) => {
            MetadataExecutor::execute_command(connection, operation, sql, &[&value, &tag_id])?;
        }
        OauthTagKey::Int(value) => {
            let value = metadata_i32(value, "tag target id")?;
            MetadataExecutor::execute_command(connection, operation, sql, &[&value, &tag_id])?;
        }
    }
    Ok(())
}

fn oauth_list_tags_for_target(
    connection: &mut LibpqOAuthConnection,
    target: PgTagTarget,
) -> Result<Vec<PgTagReadModel>, PgMetaError> {
    let (mapping_table, key_column, key_value) = oauth_tag_target_mapping(target);
    let sql = format!(
        "
        SELECT t.tag_id::text, t.name
          FROM public.tags t
          JOIN public.{mapping_table} m ON m.tag_id = t.tag_id
         WHERE m.{key_column} = $1
         ORDER BY t.name
        "
    );
    let rows = match key_value {
        OauthTagKey::Uuid(value) => {
            MetadataExecutor::query_many(connection, "list tags for target", &sql, &[&value])?
        }
        OauthTagKey::Int(value) => {
            let value = metadata_i32(value, "tag target id")?;
            MetadataExecutor::query_many(connection, "list tags for target", &sql, &[&value])?
        }
    };
    rows.iter().map(oauth_tag_from_row).collect()
}

#[derive(Debug, Clone, Copy)]
enum OauthTagKey {
    Uuid(uuid::Uuid),
    Int(i64),
}

fn oauth_tag_target_mapping(target: PgTagTarget) -> (&'static str, &'static str, OauthTagKey) {
    match target {
        PgTagTarget::Study(id) => ("study_tags", "study_id", OauthTagKey::Uuid(id.0)),
        PgTagTarget::Domain(id) => ("domain_tags", "domain_id", OauthTagKey::Int(id.0)),
        PgTagTarget::Variable(id) => ("variable_tags", "variable_id", OauthTagKey::Int(id.0)),
        PgTagTarget::Asset(id) => ("asset_tags", "asset_id", OauthTagKey::Uuid(id.0)),
        PgTagTarget::AssetVersion(id) => {
            ("asset_version_tags", "version_id", OauthTagKey::Uuid(id.0))
        }
        PgTagTarget::Entity(id) => ("entity_tags", "entity_id", OauthTagKey::Int(id.0)),
        PgTagTarget::EntityRelation(id) => (
            "entityrelation_tags",
            "entityrelation_id",
            OauthTagKey::Int(id.0),
        ),
    }
}

fn oauth_list_domains(
    connection: &mut LibpqOAuthConnection,
) -> Result<Vec<PgDomainReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list domains",
        "
        SELECT domain_id::text, name, uri, description
          FROM public.domains
         ORDER BY name
        ",
        &[],
    )?;
    rows.iter().map(oauth_domain_from_row).collect()
}

fn oauth_list_domains_by_name(
    connection: &mut LibpqOAuthConnection,
    name: &str,
) -> Result<Vec<PgDomainReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list domains by name",
        "SELECT domain_id::text, name, uri, description FROM public.domains WHERE name = $1 ORDER BY domain_id",
        &[&name],
    )?;
    rows.iter().map(oauth_domain_from_row).collect()
}

fn oauth_get_domain(
    connection: &mut LibpqOAuthConnection,
    domain_id: DomainId,
) -> Result<Option<PgDomainReadModel>, PgMetaError> {
    let domain_id = metadata_i32(domain_id.0, "domain_id")?;
    let row = MetadataExecutor::query_optional(
        connection,
        "get domain by id",
        "
        SELECT domain_id::text, name, uri, description
          FROM public.domains
         WHERE domain_id = $1
         LIMIT 1
        ",
        &[&domain_id],
    )?;
    row.as_ref().map(oauth_domain_from_row).transpose()
}

fn oauth_get_domain_by_name(
    connection: &mut LibpqOAuthConnection,
    name: &str,
) -> Result<Option<PgDomainReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get domain by name",
        "
        SELECT domain_id::text, name, uri, description
          FROM public.domains
         WHERE name = $1
         LIMIT 1
        ",
        &[&name],
    )?;
    row.as_ref().map(oauth_domain_from_row).transpose()
}

fn attach_registered_domain_tags<Row>(
    transaction: &mut dyn MetadataTransaction<Row = Row>,
    domain_id: i32,
    normalized_tags: &[String],
) -> Result<(), PgMetaError> {
    for tag in normalized_tags {
        transaction.execute_command(
            "attach registered Domain tag",
            "
            WITH registered_tag AS (
                INSERT INTO public.tags (name)
                VALUES ($1)
                ON CONFLICT (name) DO UPDATE SET name = EXCLUDED.name
                RETURNING tag_id
            )
            INSERT INTO public.domain_tags (domain_id, tag_id)
            SELECT $2, tag_id FROM registered_tag
            ON CONFLICT DO NOTHING
            ",
            &[&tag, &domain_id],
        )?;
    }
    Ok(())
}

fn attach_registered_study_metadata<Row>(
    transaction: &mut dyn MetadataTransaction<Row = Row>,
    study_id: StudyId,
    domain_ids: &[DomainId],
    normalized_tags: &[String],
) -> Result<(), PgMetaError> {
    for tag in normalized_tags {
        transaction.execute_command(
            "attach registered Study tag",
            "
            WITH registered_tag AS (
                INSERT INTO public.tags (name)
                VALUES ($1)
                ON CONFLICT (name) DO UPDATE SET name = EXCLUDED.name
                RETURNING tag_id
            )
            INSERT INTO public.study_tags (study_id, tag_id)
            SELECT $2, tag_id FROM registered_tag
            ON CONFLICT DO NOTHING
            ",
            &[&tag, &study_id.0],
        )?;
    }
    for domain_id in domain_ids {
        let domain_id = metadata_i32(domain_id.0, "domain_id")?;
        transaction.execute_command(
            "link registered Study Domain",
            "
            INSERT INTO public.study_domains (study_id, domain_id)
            VALUES ($1, $2)
            ON CONFLICT DO NOTHING
            ",
            &[&study_id.0, &domain_id],
        )?;
    }
    Ok(())
}

fn direct_create_domain_registration(
    connection: &mut PgMetadataConnection,
    input: &NewDomainRecord,
    normalized_tags: &[String],
) -> Result<DomainRecord, PgMetaError> {
    MetadataExecutor::with_transaction(connection, "create Domain registration", |transaction| {
        let row = transaction.query_one(
            "create registered Domain",
            "
            INSERT INTO public.domains (name, uri, description)
            VALUES ($1, $2, $3)
            RETURNING domain_id
            ",
            &[&input.name.as_str(), &input.uri, &input.description],
        )?;
        let domain_id = DomainId(i64::from(row.get::<_, i32>("domain_id")));
        let domain_id_i32 = metadata_i32(domain_id.0, "domain_id")?;
        attach_registered_domain_tags(transaction, domain_id_i32, normalized_tags)?;
        Ok(DomainRecord {
            domain_id,
            name: input.name.clone(),
            description: input.description.clone(),
            uri: input.uri.clone(),
        })
    })
}

fn oauth_create_domain_registration(
    connection: &mut LibpqOAuthConnection,
    input: &NewDomainRecord,
    normalized_tags: &[String],
) -> Result<DomainRecord, PgMetaError> {
    MetadataExecutor::with_transaction(connection, "create Domain registration", |transaction| {
        let row = transaction.query_one(
            "create registered Domain",
            "
            INSERT INTO public.domains (name, uri, description)
            VALUES ($1, $2, $3)
            RETURNING domain_id::text
            ",
            &[&input.name.as_str(), &input.uri, &input.description],
        )?;
        let domain_id = DomainId(oauth_i64(&row, 0, "domain_id")?);
        let domain_id_i32 = metadata_i32(domain_id.0, "domain_id")?;
        attach_registered_domain_tags(transaction, domain_id_i32, normalized_tags)?;
        Ok(DomainRecord {
            domain_id,
            name: input.name.clone(),
            description: input.description.clone(),
            uri: input.uri.clone(),
        })
    })
}

fn direct_save_domain_registration(
    connection: &mut PgMetadataConnection,
    input: &DomainRecord,
    normalized_tags: &[String],
) -> Result<DomainRecord, PgMetaError> {
    let domain_id = metadata_i32(input.domain_id.0, "domain_id")?;
    MetadataExecutor::with_transaction(connection, "save Domain registration", |transaction| {
        transaction.query_one(
            "save registered Domain",
            "
            INSERT INTO public.domains (domain_id, name, uri, description)
            VALUES ($1, $2, $3, $4)
            RETURNING domain_id
            ",
            &[
                &domain_id,
                &input.name.as_str(),
                &input.uri,
                &input.description,
            ],
        )?;
        attach_registered_domain_tags(transaction, domain_id, normalized_tags)?;
        Ok(input.clone())
    })
}

fn oauth_save_domain_registration(
    connection: &mut LibpqOAuthConnection,
    input: &DomainRecord,
    normalized_tags: &[String],
) -> Result<DomainRecord, PgMetaError> {
    let domain_id = metadata_i32(input.domain_id.0, "domain_id")?;
    MetadataExecutor::with_transaction(connection, "save Domain registration", |transaction| {
        transaction.query_one(
            "save registered Domain",
            "
            INSERT INTO public.domains (domain_id, name, uri, description)
            VALUES ($1, $2, $3, $4)
            RETURNING domain_id::text
            ",
            &[
                &domain_id,
                &input.name.as_str(),
                &input.uri,
                &input.description,
            ],
        )?;
        attach_registered_domain_tags(transaction, domain_id, normalized_tags)?;
        Ok(input.clone())
    })
}

fn oauth_create_domain(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateDomain,
) -> Result<DomainId, PgMetaError> {
    let row = match input.domain_id {
        Some(domain_id) => {
            let domain_id = metadata_i32(domain_id.0, "domain_id")?;
            MetadataExecutor::query_one(
                connection,
                "create domain",
                "
                INSERT INTO public.domains (domain_id, name, uri, description)
                VALUES ($1, $2, $3, $4)
                RETURNING domain_id::text
                ",
                &[&domain_id, &input.name, &input.uri, &input.description],
            )?
        }
        None => MetadataExecutor::query_one(
            connection,
            "create domain",
            "
            INSERT INTO public.domains (name, uri, description)
            VALUES ($1, $2, $3)
            RETURNING domain_id::text
            ",
            &[&input.name, &input.uri, &input.description],
        )?,
    };
    Ok(DomainId(oauth_i64(&row, 0, "domain_id")?))
}

fn direct_create_study_registration(
    connection: &mut PgMetadataConnection,
    input: &NewStudyRecord,
    domain_ids: &[DomainId],
    normalized_tags: &[String],
) -> Result<StudyRecord, PgMetaError> {
    let study_type_id = optional_metadata_i32(input.study_type_id.map(|id| id.0), "study_type_id")?;
    let date_created = input.date_created.map(|value| value.naive_utc());
    MetadataExecutor::with_transaction(connection, "create Study registration", |transaction| {
        let row = transaction.query_one(
            "create registered Study",
            "
            INSERT INTO public.studies (
                name, description, documentation, agent_instructions,
                external_id, study_type_id, date_created, created_by
            )
            VALUES ($1, $2, $3, $4, $5, $6,
                    COALESCE($7, CURRENT_TIMESTAMP AT TIME ZONE 'UTC'),
                    COALESCE($8, CURRENT_USER))
            RETURNING study_id, date_created, created_by
            ",
            &[
                &input.name.as_str(),
                &input.description,
                &input.documentation,
                &input.agent_instructions,
                &input.external_id,
                &study_type_id,
                &date_created,
                &input.created_by,
            ],
        )?;
        let study_id = StudyId(row.get("study_id"));
        attach_registered_study_metadata(transaction, study_id, domain_ids, normalized_tags)?;
        Ok(StudyRecord {
            study_id,
            name: input.name.clone(),
            description: input.description.clone(),
            documentation: input.documentation.clone(),
            agent_instructions: input.agent_instructions.clone(),
            external_id: input.external_id.clone(),
            study_type_id: input.study_type_id,
            date_created: row
                .get::<_, Option<NaiveDateTime>>("date_created")
                .map(|value| DateTime::from_naive_utc_and_offset(value, Utc)),
            created_by: row.get("created_by"),
        })
    })
}

fn oauth_create_study_registration(
    connection: &mut LibpqOAuthConnection,
    input: &NewStudyRecord,
    domain_ids: &[DomainId],
    normalized_tags: &[String],
) -> Result<StudyRecord, PgMetaError> {
    let study_type_id = optional_metadata_i32(input.study_type_id.map(|id| id.0), "study_type_id")?;
    let date_created = input.date_created.map(|value| value.naive_utc());
    MetadataExecutor::with_transaction(connection, "create Study registration", |transaction| {
        let row = transaction.query_one(
            "create registered Study",
            "
            INSERT INTO public.studies (
                name, description, documentation, agent_instructions,
                external_id, study_type_id, date_created, created_by
            )
            VALUES ($1, $2, $3, $4, $5, $6,
                    COALESCE($7, CURRENT_TIMESTAMP AT TIME ZONE 'UTC'),
                    COALESCE($8, CURRENT_USER))
            RETURNING study_id::text, date_created::text, created_by
            ",
            &[
                &input.name.as_str(),
                &input.description,
                &input.documentation,
                &input.agent_instructions,
                &input.external_id,
                &study_type_id,
                &date_created,
                &input.created_by,
            ],
        )?;
        let study_id = StudyId(oauth_uuid(&row, 0, "study_id")?);
        attach_registered_study_metadata(transaction, study_id, domain_ids, normalized_tags)?;
        Ok(StudyRecord {
            study_id,
            name: input.name.clone(),
            description: input.description.clone(),
            documentation: input.documentation.clone(),
            agent_instructions: input.agent_instructions.clone(),
            external_id: input.external_id.clone(),
            study_type_id: input.study_type_id,
            date_created: oauth_optional_timestamp(&row, 1, "date_created")?,
            created_by: oauth_optional_string(&row, 2),
        })
    })
}

fn direct_save_study_registration(
    connection: &mut PgMetadataConnection,
    input: &StudyRecord,
    domain_ids: &[DomainId],
    normalized_tags: &[String],
) -> Result<StudyRecord, PgMetaError> {
    let study_type_id = optional_metadata_i32(input.study_type_id.map(|id| id.0), "study_type_id")?;
    let date_created = input.date_created.map(|value| value.naive_utc());
    MetadataExecutor::with_transaction(connection, "save Study registration", |transaction| {
        let row = transaction.query_one(
            "save registered Study",
            "
            INSERT INTO public.studies (
                study_id, name, description, documentation, agent_instructions,
                external_id, study_type_id, date_created, created_by
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7,
                    COALESCE($8, CURRENT_TIMESTAMP AT TIME ZONE 'UTC'),
                    COALESCE($9, CURRENT_USER))
            RETURNING date_created, created_by
            ",
            &[
                &input.study_id.0,
                &input.name.as_str(),
                &input.description,
                &input.documentation,
                &input.agent_instructions,
                &input.external_id,
                &study_type_id,
                &date_created,
                &input.created_by,
            ],
        )?;
        attach_registered_study_metadata(transaction, input.study_id, domain_ids, normalized_tags)?;
        let mut study = input.clone();
        study.date_created = row
            .get::<_, Option<NaiveDateTime>>("date_created")
            .map(|value| DateTime::from_naive_utc_and_offset(value, Utc));
        study.created_by = row.get("created_by");
        Ok(study)
    })
}

fn oauth_save_study_registration(
    connection: &mut LibpqOAuthConnection,
    input: &StudyRecord,
    domain_ids: &[DomainId],
    normalized_tags: &[String],
) -> Result<StudyRecord, PgMetaError> {
    let study_type_id = optional_metadata_i32(input.study_type_id.map(|id| id.0), "study_type_id")?;
    let date_created = input.date_created.map(|value| value.naive_utc());
    MetadataExecutor::with_transaction(connection, "save Study registration", |transaction| {
        let row = transaction.query_one(
            "save registered Study",
            "
            INSERT INTO public.studies (
                study_id, name, description, documentation, agent_instructions,
                external_id, study_type_id, date_created, created_by
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7,
                    COALESCE($8, CURRENT_TIMESTAMP AT TIME ZONE 'UTC'),
                    COALESCE($9, CURRENT_USER))
            RETURNING date_created::text, created_by
            ",
            &[
                &input.study_id.0,
                &input.name.as_str(),
                &input.description,
                &input.documentation,
                &input.agent_instructions,
                &input.external_id,
                &study_type_id,
                &date_created,
                &input.created_by,
            ],
        )?;
        attach_registered_study_metadata(transaction, input.study_id, domain_ids, normalized_tags)?;
        let mut study = input.clone();
        study.date_created = oauth_optional_timestamp(&row, 0, "date_created")?;
        study.created_by = oauth_optional_string(&row, 1);
        Ok(study)
    })
}

fn oauth_list_studies(
    connection: &mut LibpqOAuthConnection,
) -> Result<Vec<PgStudyReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list studies",
        "
        SELECT study_id::text, name, description, documentation, agent_instructions,
               external_id, study_type_id::text, date_created::text, created_by
          FROM public.studies
         ORDER BY name
        ",
        &[],
    )?;
    rows.iter().map(oauth_study_from_row).collect()
}

fn oauth_get_study(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
) -> Result<Option<PgStudyReadModel>, PgMetaError> {
    let study_id = study_id.0;
    let row = MetadataExecutor::query_optional(
        connection,
        "get study by id",
        "
        SELECT study_id::text, name, description, documentation, agent_instructions,
               external_id, study_type_id::text, date_created::text, created_by
          FROM public.studies
         WHERE study_id = $1
         LIMIT 1
        ",
        &[&study_id],
    )?;
    row.as_ref().map(oauth_study_from_row).transpose()
}

fn oauth_get_study_by_name(
    connection: &mut LibpqOAuthConnection,
    name: &str,
) -> Result<Option<PgStudyReadModel>, PgMetaError> {
    let row = MetadataExecutor::query_optional(
        connection,
        "get study by name",
        "
        SELECT study_id::text, name, description, documentation, agent_instructions,
               external_id, study_type_id::text, date_created::text, created_by
          FROM public.studies
         WHERE name = $1
         LIMIT 1
        ",
        &[&name],
    )?;
    row.as_ref().map(oauth_study_from_row).transpose()
}

fn oauth_create_study(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateStudy,
) -> Result<StudyId, PgMetaError> {
    let study_type_id = optional_metadata_i32(input.study_type_id.map(|id| id.0), "study_type_id")?;
    let date_created = input.date_created.map(|value| value.naive_utc());
    let row = match input.study_id {
        Some(study_id) => {
            let study_id = study_id.0;
            MetadataExecutor::query_one(
                connection,
                "create study",
                "
                INSERT INTO public.studies (
                    study_id, name, description, documentation, agent_instructions,
                    external_id, study_type_id, date_created, created_by
                )
                VALUES ($1, $2, $3, $4, $5, $6, $7, COALESCE($8, CURRENT_TIMESTAMP),
                        COALESCE($9, CURRENT_USER))
                RETURNING study_id::text
                ",
                &[
                    &study_id,
                    &input.name,
                    &input.description,
                    &input.documentation,
                    &input.agent_instructions,
                    &input.external_id,
                    &study_type_id,
                    &date_created,
                    &input.created_by,
                ],
            )?
        }
        None => MetadataExecutor::query_one(
            connection,
            "create study",
            "
            INSERT INTO public.studies (
                name, description, documentation, agent_instructions,
                external_id, study_type_id, date_created, created_by
            )
            VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, CURRENT_TIMESTAMP),
                    COALESCE($8, CURRENT_USER))
            RETURNING study_id::text
            ",
            &[
                &input.name,
                &input.description,
                &input.documentation,
                &input.agent_instructions,
                &input.external_id,
                &study_type_id,
                &date_created,
                &input.created_by,
            ],
        )?,
    };
    Ok(StudyId(oauth_uuid(&row, 0, "study_id")?))
}

fn oauth_link_study_domain(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
    domain_id: DomainId,
) -> Result<(), PgMetaError> {
    let study_id = study_id.0;
    let domain_id = metadata_i32(domain_id.0, "domain_id")?;
    MetadataExecutor::execute_command(
        connection,
        "link study domain",
        "
        INSERT INTO public.study_domains (study_id, domain_id)
        VALUES ($1, $2)
        ON CONFLICT DO NOTHING
        ",
        &[&study_id, &domain_id],
    )?;
    Ok(())
}

fn oauth_list_study_domains(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
) -> Result<Vec<PgDomainReadModel>, PgMetaError> {
    let study_id = study_id.0;
    let rows = MetadataExecutor::query_many(
        connection,
        "list study domains",
        "
        SELECT d.domain_id::text, d.name, d.uri, d.description
          FROM public.domains d
          JOIN public.study_domains sd ON sd.domain_id = d.domain_id
         WHERE sd.study_id = $1
         ORDER BY d.name
        ",
        &[&study_id],
    )?;
    rows.iter().map(oauth_domain_from_row).collect()
}

fn oauth_list_study_access(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
) -> Result<Vec<PgStudyAccessReadModel>, PgMetaError> {
    let study_id = study_id.0;
    let rows = MetadataExecutor::query_many(
        connection,
        "list study access",
        "
        SELECT study_id::text, user_id, date_granted::text, granted_by
          FROM public.study_access
         WHERE study_id = $1
         ORDER BY user_id
        ",
        &[&study_id],
    )?;
    rows.iter().map(oauth_study_access_from_row).collect()
}

fn oauth_grant_study_access(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
    target_user_id: &str,
) -> Result<(), PgMetaError> {
    let study_id = study_id.0;
    MetadataExecutor::query_many(
        connection,
        "grant study access",
        "
        SELECT public.tre_grant_study_access($1, $2)
        ",
        &[&study_id, &target_user_id],
    )?;
    Ok(())
}

fn oauth_add_study_delegate_custodian(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
    target_user_id: &str,
) -> Result<(), PgMetaError> {
    let study_id = study_id.0;
    MetadataExecutor::query_many(
        connection,
        "add study delegate custodian",
        OAUTH_ADD_STUDY_DELEGATE_CUSTODIAN_SQL,
        &[&study_id, &target_user_id],
    )?;
    Ok(())
}

fn oauth_transfer_study_primary_custodianship(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
    target_user_id: &str,
) -> Result<(), PgMetaError> {
    let study_id = study_id.0;
    MetadataExecutor::query_many(
        connection,
        "transfer study primary custodianship",
        OAUTH_TRANSFER_PRIMARY_CUSTODIANSHIP_SQL,
        &[&study_id, &target_user_id],
    )?;
    Ok(())
}

fn oauth_remove_study_delegate_custodian(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
    target_user_id: &str,
) -> Result<(), PgMetaError> {
    let study_id = study_id.0;
    MetadataExecutor::query_many(
        connection,
        "remove study delegate custodian",
        OAUTH_REMOVE_STUDY_DELEGATE_CUSTODIAN_SQL,
        &[&study_id, &target_user_id],
    )?;
    Ok(())
}

fn oauth_revoke_study_access(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
    target_user_id: &str,
) -> Result<(), PgMetaError> {
    let study_id = study_id.0;
    MetadataExecutor::query_many(
        connection,
        "revoke study access",
        "
        SELECT public.tre_revoke_study_access($1, $2)
        ",
        &[&study_id, &target_user_id],
    )?;
    Ok(())
}

fn oauth_list_study_custodians(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
) -> Result<Vec<PgStudyCustodianReadModel>, PgMetaError> {
    let study_id = study_id.0;
    let rows = MetadataExecutor::query_many(
        connection,
        "list study custodians",
        "
        SELECT study_id::text, user_id, is_primary::text, date_granted::text, granted_by
          FROM public.study_custodians
         WHERE study_id = $1
         ORDER BY is_primary DESC, user_id
        ",
        &[&study_id],
    )?;
    rows.iter().map(oauth_study_custodian_from_row).collect()
}

fn oauth_save_study_custodian(
    connection: &mut LibpqOAuthConnection,
    custodian: &StudyCustodianLinkRecord,
) -> Result<(), PgMetaError> {
    let study_id = custodian.study_id.0;
    let date_granted = custodian.date_granted.map(|value| value.naive_utc());
    let granted_by = custodian.granted_by.as_deref();
    MetadataExecutor::with_transaction(connection, "save study custodian", |transaction| {
        if custodian.is_primary {
            transaction.execute_command(
                "unset primary study custodian",
                "UPDATE public.study_custodians SET is_primary = FALSE WHERE study_id = $1",
                &[&study_id],
            )?;
        }
        transaction.execute_command(
            "add study custodian",
            "
            INSERT INTO public.study_custodians (
                study_id, user_id, is_primary, date_granted, granted_by
            )
            VALUES ($1, $2, $3, COALESCE($4, CURRENT_TIMESTAMP), COALESCE($5, CURRENT_USER))
            ON CONFLICT (study_id, user_id) DO UPDATE SET
                is_primary = EXCLUDED.is_primary
            ",
            &[
                &study_id,
                &custodian.user_id,
                &custodian.is_primary,
                &date_granted,
                &granted_by,
            ],
        )?;
        Ok(())
    })
}

fn replace_study_duo_restrictions_with_executor<E>(
    executor: &mut E,
    study_id: StudyId,
    restrictions: &[NewStudyDuoRestrictionRecord],
) -> Result<(), PgMetaError>
where
    E: MetadataExecutor,
{
    executor.with_transaction("replace study DUO restrictions", |transaction| {
        transaction.query_one(
            "authorize study DUO replacement",
            "SELECT public.tre_assert_study_duo_management($1)",
            &[&study_id.0],
        )?;
        transaction.execute_command(
            "delete existing study DUO restrictions",
            "DELETE FROM public.study_duo_restrictions WHERE study_id = $1",
            &[&study_id.0],
        )?;

        for restriction in restrictions {
            let inserted = transaction.query_optional(
                "insert replacement study DUO restriction",
                "
                INSERT INTO public.study_duo_restrictions (
                    study_id, duo_id, qualifier_ontology_namespace, qualifier_ontology_id,
                    qualifier_text, qualifier_date, qualifier_integer
                )
                SELECT $1, dc.duo_id, $3, $4, $5, $6, $7
                  FROM public.duo_codes dc
                 WHERE dc.duo_id = $2 OR dc.duo_code = $2
                RETURNING study_duo_restriction_id
                ",
                &[
                    &study_id.0,
                    &restriction.duo_id,
                    &restriction.qualifier_ontology_namespace,
                    &restriction.qualifier_ontology_id,
                    &restriction.qualifier_text,
                    &restriction.qualifier_date,
                    &restriction.qualifier_integer,
                ],
            )?;
            if inserted.is_none() {
                return Err(PgMetaError::Decode {
                    field: "duo_id",
                    value: restriction.duo_id.clone(),
                    message: "DUO code was not found".to_string(),
                });
            }
        }

        Ok(())
    })
}

fn replace_asset_version_duo_restrictions_with_executor<E>(
    executor: &mut E,
    version_id: VersionId,
    restrictions: &[NewAssetVersionDuoRestrictionRecord],
) -> Result<(), PgMetaError>
where
    E: MetadataExecutor,
{
    executor.with_transaction("replace asset version DUO restrictions", |transaction| {
        transaction.execute_command(
            "delete existing asset version DUO restrictions",
            "DELETE FROM public.asset_version_duo_restrictions WHERE version_id = $1",
            &[&version_id.0],
        )?;

        for restriction in restrictions {
            let inserted = transaction.query_optional(
                "insert replacement asset version DUO restriction",
                "
                INSERT INTO public.asset_version_duo_restrictions (
                    version_id, duo_id, qualifier_ontology_namespace, qualifier_ontology_id,
                    qualifier_text, qualifier_date, qualifier_integer
                )
                SELECT $1, dc.duo_id, $3, $4, $5, $6, $7
                  FROM public.duo_codes dc
                 WHERE dc.duo_id = $2 OR dc.duo_code = $2
                RETURNING asset_version_duo_restriction_id
                ",
                &[
                    &version_id.0,
                    &restriction.duo_id,
                    &restriction.qualifier_ontology_namespace,
                    &restriction.qualifier_ontology_id,
                    &restriction.qualifier_text,
                    &restriction.qualifier_date,
                    &restriction.qualifier_integer,
                ],
            )?;
            if inserted.is_none() {
                return Err(PgMetaError::Decode {
                    field: "duo_id",
                    value: restriction.duo_id.clone(),
                    message: "DUO code was not found".to_string(),
                });
            }
        }

        Ok(())
    })
}

fn clear_asset_version_duo_restrictions_with_executor<E>(
    executor: &mut E,
    version_id: VersionId,
) -> Result<(), PgMetaError>
where
    E: MetadataExecutor,
{
    executor.with_transaction("clear asset version DUO restrictions", |transaction| {
        transaction.execute_command(
            "delete existing asset version DUO restrictions",
            "DELETE FROM public.asset_version_duo_restrictions WHERE version_id = $1",
            &[&version_id.0],
        )?;
        Ok(())
    })
}

fn oauth_list_study_duo_restrictions(
    connection: &mut LibpqOAuthConnection,
    study_id: StudyId,
) -> Result<Vec<PgDuoRestrictionReadModel>, PgMetaError> {
    let rows = MetadataExecutor::query_many(
        connection,
        "list study DUO restrictions",
        "
        SELECT sdr.study_duo_restriction_id::text, sdr.study_id::text, sdr.duo_id,
               dc.duo_code, dc.label,
               sdr.qualifier_ontology_namespace, sdr.qualifier_ontology_id,
               sdr.qualifier_text, sdr.qualifier_date::text, sdr.qualifier_integer::text
          FROM public.study_duo_restrictions sdr
          JOIN public.duo_codes dc ON dc.duo_id = sdr.duo_id
         WHERE sdr.study_id = $1
         ORDER BY sdr.study_duo_restriction_id
        ",
        &[&study_id.0],
    )?;
    rows.iter().map(oauth_study_duo_from_row).collect()
}

fn oauth_create_study_duo_restriction(
    connection: &mut LibpqOAuthConnection,
    input: PgCreateStudyDuoRestriction,
) -> Result<i64, PgMetaError> {
    let restriction_id = optional_metadata_i32(
        input.study_duo_restriction_id.map(|value| value.0),
        "study_duo_restriction_id",
    )?;
    let row = match restriction_id {
        Some(restriction_id) => MetadataExecutor::query_one(
            connection,
            "create study DUO restriction",
            "
            INSERT INTO public.study_duo_restrictions (
                study_duo_restriction_id, study_id, duo_id,
                qualifier_ontology_namespace, qualifier_ontology_id,
                qualifier_text, qualifier_date, qualifier_integer
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
            RETURNING study_duo_restriction_id::text
            ",
            &[
                &restriction_id,
                &input.study_id.0,
                &input.duo_id,
                &input.qualifiers.qualifier_ontology_namespace,
                &input.qualifiers.qualifier_ontology_id,
                &input.qualifiers.qualifier_text,
                &input.qualifiers.qualifier_date,
                &input.qualifiers.qualifier_integer,
            ],
        )?,
        None => MetadataExecutor::query_one(
            connection,
            "create study DUO restriction",
            "
            INSERT INTO public.study_duo_restrictions (
                study_id, duo_id, qualifier_ontology_namespace, qualifier_ontology_id,
                qualifier_text, qualifier_date, qualifier_integer
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7)
            RETURNING study_duo_restriction_id::text
            ",
            &[
                &input.study_id.0,
                &input.duo_id,
                &input.qualifiers.qualifier_ontology_namespace,
                &input.qualifiers.qualifier_ontology_id,
                &input.qualifiers.qualifier_text,
                &input.qualifiers.qualifier_date,
                &input.qualifiers.qualifier_integer,
            ],
        )?,
    };
    oauth_i64(&row, 0, "study_duo_restriction_id")
}

fn oauth_domain_from_row(row: &LibpqOAuthRow) -> Result<PgDomainReadModel, PgMetaError> {
    Ok(PgDomainReadModel {
        domain_id: DomainId(oauth_i64(row, 0, "domain_id")?),
        name: oauth_string(row, 1, "name")?,
        uri: oauth_optional_string(row, 2),
        description: oauth_optional_string(row, 3),
    })
}

fn oauth_study_from_row(row: &LibpqOAuthRow) -> Result<PgStudyReadModel, PgMetaError> {
    Ok(PgStudyReadModel {
        study_id: StudyId(oauth_uuid(row, 0, "study_id")?),
        name: oauth_string(row, 1, "name")?,
        description: oauth_optional_string(row, 2),
        documentation: oauth_optional_string(row, 3),
        agent_instructions: oauth_optional_string(row, 4),
        external_id: oauth_optional_string(row, 5),
        study_type_id: oauth_optional_i64(row, 6, "study_type_id")?.map(StudyTypeId),
        date_created: oauth_optional_timestamp(row, 7, "date_created")?,
        created_by: oauth_optional_string(row, 8),
    })
}

fn oauth_study_access_from_row(row: &LibpqOAuthRow) -> Result<PgStudyAccessReadModel, PgMetaError> {
    Ok(PgStudyAccessReadModel {
        study_id: StudyId(oauth_uuid(row, 0, "study_id")?),
        user_id: oauth_string(row, 1, "user_id")?,
        date_granted: oauth_optional_timestamp(row, 2, "date_granted")?,
        granted_by: oauth_optional_string(row, 3),
    })
}

fn oauth_study_custodian_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgStudyCustodianReadModel, PgMetaError> {
    Ok(PgStudyCustodianReadModel {
        study_id: StudyId(oauth_uuid(row, 0, "study_id")?),
        user_id: oauth_string(row, 1, "user_id")?,
        is_primary: oauth_bool(row, 2, "is_primary")?,
        date_granted: oauth_optional_timestamp(row, 3, "date_granted")?,
        granted_by: oauth_optional_string(row, 4),
    })
}

fn oauth_study_duo_from_row(row: &LibpqOAuthRow) -> Result<PgDuoRestrictionReadModel, PgMetaError> {
    Ok(PgDuoRestrictionReadModel {
        restriction_id: oauth_i64(row, 0, "study_duo_restriction_id")?,
        study_id: StudyId(oauth_uuid(row, 1, "study_id")?),
        duo_id: oauth_string(row, 2, "duo_id")?,
        duo_code: oauth_string(row, 3, "duo_code")?,
        label: oauth_string(row, 4, "label")?,
        qualifier_ontology_namespace: oauth_optional_string(row, 5),
        qualifier_ontology_id: oauth_optional_string(row, 6),
        qualifier_text: oauth_optional_string(row, 7),
        qualifier_date: oauth_optional_date(row, 8, "qualifier_date")?,
        qualifier_integer: oauth_optional_i64(row, 9, "qualifier_integer")?
            .map(|value| metadata_i32(value, "qualifier_integer"))
            .transpose()?,
    })
}

fn oauth_asset_from_row(row: &LibpqOAuthRow) -> Result<PgAssetReadModel, PgMetaError> {
    Ok(PgAssetReadModel {
        asset_id: AssetId(oauth_uuid(row, 0, "asset_id")?),
        study_id: StudyId(oauth_uuid(row, 1, "study_id")?),
        name: oauth_string(row, 2, "name")?,
        description: oauth_optional_string(row, 3),
        agent_instructions: oauth_optional_string(row, 4),
        asset_type: parse_asset_type_text(oauth_string(row, 5, "asset_type")?)?,
        date_created: oauth_optional_timestamp(row, 6, "date_created")?,
        created_by: oauth_optional_string(row, 7),
    })
}

fn oauth_asset_version_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgAssetVersionReadModel, PgMetaError> {
    Ok(PgAssetVersionReadModel {
        version_id: VersionId(oauth_uuid(row, 0, "version_id")?),
        asset_id: AssetId(oauth_uuid(row, 1, "asset_id")?),
        major: oauth_i32(row, 2, "major")?,
        minor: oauth_i32(row, 3, "minor")?,
        patch: oauth_i32(row, 4, "patch")?,
        version_label: oauth_optional_string(row, 5),
        version_note: oauth_optional_string(row, 6),
        is_latest: oauth_optional_bool(row, 7, "is_latest")?,
        doi: oauth_optional_string(row, 8),
        created_at: oauth_optional_timestamp(row, 9, "created_at")?,
        created_by: oauth_optional_string(row, 10),
    })
}

fn oauth_asset_version_duo_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgAssetVersionDuoRestrictionReadModel, PgMetaError> {
    Ok(PgAssetVersionDuoRestrictionReadModel {
        restriction_id: oauth_i64(row, 0, "asset_version_duo_restriction_id")?,
        version_id: VersionId(oauth_uuid(row, 1, "version_id")?),
        duo_id: oauth_string(row, 2, "duo_id")?,
        duo_code: oauth_string(row, 3, "duo_code")?,
        label: oauth_string(row, 4, "label")?,
        qualifier_ontology_namespace: oauth_optional_string(row, 5),
        qualifier_ontology_id: oauth_optional_string(row, 6),
        qualifier_text: oauth_optional_string(row, 7),
        qualifier_date: oauth_optional_date(row, 8, "qualifier_date")?,
        qualifier_integer: oauth_optional_i64(row, 9, "qualifier_integer")?
            .map(|value| metadata_i32(value, "qualifier_integer"))
            .transpose()?,
    })
}

fn oauth_dataset_from_row(row: &LibpqOAuthRow) -> Result<PgDatasetReadModel, PgMetaError> {
    Ok(PgDatasetReadModel {
        dataset_id: VersionId(oauth_uuid(row, 0, "dataset_id")?),
    })
}

fn oauth_dataset_version_withdrawal_from_row(
    row: &LibpqOAuthRow,
) -> Result<DatasetVersionWithdrawalRecord, PgMetaError> {
    Ok(DatasetVersionWithdrawalRecord {
        dataset_id: VersionId(oauth_uuid(row, 0, "dataset_id")?),
        asset_id: AssetId(oauth_uuid(row, 1, "asset_id")?),
        withdrawn_at: oauth_optional_timestamp(row, 2, "withdrawn_at")?,
        withdrawn_by: oauth_optional_string(row, 3),
        reason: oauth_string(row, 4, "reason")?,
        tombstone_state: oauth_string(row, 5, "tombstone_state")?,
        provenance: oauth_optional_string(row, 6),
        drop_lake_table: oauth_optional_bool(row, 7, "drop_lake_table")?.unwrap_or(false),
    })
}

fn oauth_datafile_from_row(row: &LibpqOAuthRow) -> Result<PgDataFileReadModel, PgMetaError> {
    let size_bytes = oauth_optional_string(row, 1);
    Ok(PgDataFileReadModel {
        datafile_id: VersionId(oauth_uuid(row, 0, "datafile_id")?),
        size_bytes: size_bytes
            .as_deref()
            .map(|value| value.parse::<i64>())
            .transpose()
            .map_err(|error| PgMetaError::Decode {
                field: "size_bytes",
                value: size_bytes.unwrap_or_default(),
                message: error.to_string(),
            })?,
        compressed: oauth_optional_bool(row, 2, "compressed")?,
        encrypted: oauth_optional_bool(row, 3, "encrypted")?,
        compression_algorithm: oauth_optional_string(row, 4),
        encryption_algorithm: oauth_optional_string(row, 5),
        encryption_key: oauth_optional_hex(row, 6, "encryption_key")?,
        storage_uri: oauth_string(row, 7, "storage_uri")?,
        edam_format: oauth_string(row, 8, "edam_format")?,
        digest: oauth_string(row, 9, "digest")?,
    })
}

fn oauth_value_type_from_row(row: &LibpqOAuthRow) -> Result<PgValueTypeReadModel, PgMetaError> {
    Ok(PgValueTypeReadModel {
        value_type_id: ValueTypeId(oauth_i64(row, 0, "value_type_id")?),
        value_type: oauth_string(row, 1, "value_type")?,
        description: oauth_optional_string(row, 2),
    })
}

fn oauth_variable_from_row(row: &LibpqOAuthRow) -> Result<PgVariableReadModel, PgMetaError> {
    Ok(PgVariableReadModel {
        variable_id: VariableId(oauth_i64(row, 0, "variable_id")?),
        domain_id: DomainId(oauth_i64(row, 1, "domain_id")?),
        name: oauth_string(row, 2, "name")?,
        value_type_id: ValueTypeId(oauth_i64(row, 3, "value_type_id")?),
        value_type: oauth_string(row, 4, "value_type")?,
        value_format: oauth_optional_string(row, 5),
        vocabulary_id: oauth_optional_i64(row, 6, "vocabulary_id")?.map(VocabularyId),
        key_role: parse_key_role_text(oauth_string(row, 7, "keyrole")?)?,
        description: oauth_optional_string(row, 8),
        note: oauth_optional_string(row, 9),
        ontology_namespace: oauth_optional_string(row, 10),
        ontology_class: oauth_optional_string(row, 11),
    })
}

fn oauth_dataset_variable_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgDatasetVariableReadModel, PgMetaError> {
    Ok(PgDatasetVariableReadModel {
        dataset_id: VersionId(oauth_uuid(row, 0, "dataset_id")?),
        variable_id: VariableId(oauth_i64(row, 1, "variable_id")?),
        row_role: parse_key_role_text(oauth_string(row, 2, "row_role")?)?,
    })
}

fn oauth_dataset_version_entity_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgDatasetVersionEntityReadModel, PgMetaError> {
    Ok(PgDatasetVersionEntityReadModel {
        version_id: VersionId(oauth_uuid(row, 0, "version_id")?),
        entity_instance_id: oauth_i64(row, 1, "entity_instance_id")?,
        entity_variable_id: VariableId(oauth_i64(row, 2, "entity_variable_id")?),
        transformation_id: TransformationId(oauth_i64(row, 3, "transformation_id")?),
    })
}

fn oauth_asset_version_entity_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgAssetVersionEntityReadModel, PgMetaError> {
    Ok(PgAssetVersionEntityReadModel {
        version_id: VersionId(oauth_uuid(row, 0, "version_id")?),
        entity_instance_id: oauth_i64(row, 1, "entity_instance_id")?,
        transformation_id: TransformationId(oauth_i64(row, 2, "transformation_id")?),
    })
}

fn oauth_dataset_version_relation_instance_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgDatasetVersionRelationInstanceReadModel, PgMetaError> {
    Ok(PgDatasetVersionRelationInstanceReadModel {
        version_id: VersionId(oauth_uuid(row, 0, "version_id")?),
        relation_instance_id: oauth_i64(row, 1, "relation_instance_id")?,
        subject_variable_id: VariableId(oauth_i64(row, 2, "subject_variable_id")?),
        object_variable_id: VariableId(oauth_i64(row, 3, "object_variable_id")?),
        relation_variable_id: oauth_optional_i64(row, 4, "relation_variable_id")?.map(VariableId),
        transformation_id: TransformationId(oauth_i64(row, 5, "transformation_id")?),
    })
}

fn oauth_asset_version_relation_instance_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgAssetVersionRelationInstanceReadModel, PgMetaError> {
    Ok(PgAssetVersionRelationInstanceReadModel {
        version_id: VersionId(oauth_uuid(row, 0, "version_id")?),
        relation_instance_id: oauth_i64(row, 1, "relation_instance_id")?,
        transformation_id: TransformationId(oauth_i64(row, 2, "transformation_id")?),
    })
}

fn oauth_vocabulary_from_row(row: &LibpqOAuthRow) -> Result<PgVocabularyReadModel, PgMetaError> {
    Ok(PgVocabularyReadModel {
        vocabulary_id: VocabularyId(oauth_i64(row, 0, "vocabulary_id")?),
        domain_id: DomainId(oauth_i64(row, 1, "domain_id")?),
        name: oauth_string(row, 2, "name")?,
        description: oauth_optional_string(row, 3),
    })
}

fn oauth_vocabulary_item_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgVocabularyItemReadModel, PgMetaError> {
    Ok(PgVocabularyItemReadModel {
        vocabulary_item_id: VocabularyItemId(oauth_i64(row, 0, "vocabulary_item_id")?),
        vocabulary_id: VocabularyId(oauth_i64(row, 1, "vocabulary_id")?),
        value: oauth_i32(row, 2, "value")?,
        code: oauth_string(row, 3, "code")?,
        description: oauth_optional_string(row, 4),
    })
}

fn oauth_vocabulary_mapping_from_row(
    row: &LibpqOAuthRow,
) -> Result<crate::PgVocabularyMappingReadModel, PgMetaError> {
    Ok(crate::PgVocabularyMappingReadModel {
        vocabulary_mapping_id: oauth_i64(row, 0, "vocabulary_mapping_id")?,
        from_vocabulary_item: VocabularyItemId(oauth_i64(row, 1, "from_vocabulary_item")?),
        to_vocabulary_item: VocabularyItemId(oauth_i64(row, 2, "to_vocabulary_item")?),
    })
}

fn oauth_entity_from_row(row: &LibpqOAuthRow) -> Result<PgEntityReadModel, PgMetaError> {
    Ok(PgEntityReadModel {
        entity_id: EntityId(oauth_i64(row, 0, "entity_id")?),
        domain_id: DomainId(oauth_i64(row, 1, "domain_id")?),
        uuid: oauth_optional_uuid(row, 2, "uuid")?,
        name: oauth_string(row, 3, "name")?,
        description: oauth_optional_string(row, 4),
        agent_instructions: oauth_optional_string(row, 5),
        ontology_namespace: oauth_optional_string(row, 6),
        ontology_class: oauth_optional_string(row, 7),
    })
}

fn oauth_entity_relation_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgEntityRelationReadModel, PgMetaError> {
    Ok(PgEntityRelationReadModel {
        entity_relation_id: EntityRelationId(oauth_i64(row, 0, "entityrelation_id")?),
        subject_entity_id: EntityId(oauth_i64(row, 1, "subject_entity_id")?),
        object_entity_id: EntityId(oauth_i64(row, 2, "object_entity_id")?),
        domain_id: DomainId(oauth_i64(row, 3, "domain_id")?),
        uuid: oauth_optional_uuid(row, 4, "uuid")?,
        name: oauth_string(row, 5, "name")?,
        description: oauth_optional_string(row, 6),
        agent_instructions: oauth_optional_string(row, 7),
        ontology_namespace: oauth_optional_string(row, 8),
        ontology_class: oauth_optional_string(row, 9),
    })
}

fn oauth_entity_instance_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgEntityInstanceReadModel, PgMetaError> {
    Ok(PgEntityInstanceReadModel {
        instance_id: oauth_i64(row, 0, "instance_id")?,
        uuid: oauth_uuid(row, 1, "uuid")?,
        entity_id: EntityId(oauth_i64(row, 2, "entity_id")?),
        label: oauth_optional_string(row, 3),
        note: oauth_optional_string(row, 4),
        transformation_id: TransformationId(oauth_i64(row, 5, "transformation_id")?),
    })
}

fn oauth_study_entity_instance_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgStudyEntityInstanceReadModel, PgMetaError> {
    Ok(PgStudyEntityInstanceReadModel {
        study_id: StudyId(oauth_uuid(row, 0, "study_id")?),
        entity_instance_id: oauth_i64(row, 1, "entity_instance_id")?,
        entity_id: EntityId(oauth_i64(row, 2, "entity_id")?),
        external_id: oauth_string(row, 3, "external_id")?,
        transformation_id: TransformationId(oauth_i64(row, 4, "transformation_id")?),
    })
}

fn oauth_relation_instance_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgRelationInstanceReadModel, PgMetaError> {
    Ok(PgRelationInstanceReadModel {
        relation_instance_id: oauth_i64(row, 0, "relation_instance_id")?,
        uuid: oauth_uuid(row, 1, "uuid")?,
        entity_relation_id: EntityRelationId(oauth_i64(row, 2, "entityrelation_id")?),
        entity_instance_id_1: oauth_i64(row, 3, "entity_instance_id_1")?,
        entity_instance_id_2: oauth_i64(row, 4, "entity_instance_id_2")?,
        valid_from: oauth_optional_date(row, 5, "valid_from")?,
        valid_to: oauth_optional_date(row, 6, "valid_to")?,
        note: oauth_optional_string(row, 7),
        transformation_id: TransformationId(oauth_i64(row, 8, "transformation_id")?),
    })
}

fn oauth_study_relation_instance_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgStudyRelationInstanceReadModel, PgMetaError> {
    Ok(PgStudyRelationInstanceReadModel {
        study_id: StudyId(oauth_uuid(row, 0, "study_id")?),
        relation_instance_id: oauth_i64(row, 1, "relation_instance_id")?,
        entity_relation_id: EntityRelationId(oauth_i64(row, 2, "entityrelation_id")?),
        external_id: oauth_string(row, 3, "external_id")?,
        transformation_id: TransformationId(oauth_i64(row, 4, "transformation_id")?),
    })
}

fn oauth_transformation_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgTransformationReadModel, PgMetaError> {
    Ok(PgTransformationReadModel {
        transformation_id: TransformationId(oauth_i64(row, 0, "transformation_id")?),
        transformation_type: parse_transformation_type_text(oauth_string(
            row,
            1,
            "transformation_type",
        )?)?,
        description: oauth_string(row, 2, "description")?,
        repository_url: oauth_optional_string(row, 3),
        commit_hash: oauth_optional_string(row, 4),
        file_path: oauth_optional_string(row, 5),
        date_created: oauth_optional_timestamp(row, 6, "date_created")?,
        created_by: oauth_optional_string(row, 7),
    })
}

fn oauth_transformation_input_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgTransformationInputReadModel, PgMetaError> {
    Ok(PgTransformationInputReadModel {
        transformation_id: TransformationId(oauth_i64(row, 0, "transformation_id")?),
        version_id: VersionId(oauth_uuid(row, 1, "version_id")?),
    })
}

fn oauth_transformation_output_from_row(
    row: &LibpqOAuthRow,
) -> Result<PgTransformationOutputReadModel, PgMetaError> {
    Ok(PgTransformationOutputReadModel {
        transformation_id: TransformationId(oauth_i64(row, 0, "transformation_id")?),
        version_id: VersionId(oauth_uuid(row, 1, "version_id")?),
    })
}

fn oauth_tag_from_row(row: &LibpqOAuthRow) -> Result<PgTagReadModel, PgMetaError> {
    Ok(PgTagReadModel {
        tag_id: oauth_i64(row, 0, "tag_id")?,
        name: oauth_string(row, 1, "name")?,
    })
}

fn metadata_i32(value: i64, field: &'static str) -> Result<i32, PgMetaError> {
    i32::try_from(value).map_err(|_| PgMetaError::Decode {
        field,
        value: value.to_string(),
        message: "value does not fit into PostgreSQL int4".to_string(),
    })
}

fn optional_metadata_i32(
    value: Option<i64>,
    field: &'static str,
) -> Result<Option<i32>, PgMetaError> {
    value.map(|value| metadata_i32(value, field)).transpose()
}

fn oauth_string(
    row: &LibpqOAuthRow,
    index: usize,
    field: &'static str,
) -> Result<String, PgMetaError> {
    oauth_required(row, index, field).map(ToOwned::to_owned)
}

fn oauth_optional_string(row: &LibpqOAuthRow, index: usize) -> Option<String> {
    row.get(index).map(ToOwned::to_owned)
}

fn oauth_i64(row: &LibpqOAuthRow, index: usize, field: &'static str) -> Result<i64, PgMetaError> {
    let value = oauth_required(row, index, field)?;
    value.parse::<i64>().map_err(|error| PgMetaError::Decode {
        field,
        value: value.to_string(),
        message: error.to_string(),
    })
}

fn oauth_i32(row: &LibpqOAuthRow, index: usize, field: &'static str) -> Result<i32, PgMetaError> {
    let value = oauth_i64(row, index, field)?;
    metadata_i32(value, field)
}

fn oauth_optional_i64(
    row: &LibpqOAuthRow,
    index: usize,
    field: &'static str,
) -> Result<Option<i64>, PgMetaError> {
    row.get(index)
        .map(|value| {
            value.parse::<i64>().map_err(|error| PgMetaError::Decode {
                field,
                value: value.to_string(),
                message: error.to_string(),
            })
        })
        .transpose()
}

fn oauth_uuid(
    row: &LibpqOAuthRow,
    index: usize,
    field: &'static str,
) -> Result<uuid::Uuid, PgMetaError> {
    let value = oauth_required(row, index, field)?;
    uuid::Uuid::parse_str(value).map_err(|error| PgMetaError::Decode {
        field,
        value: value.to_string(),
        message: error.to_string(),
    })
}

fn oauth_optional_uuid(
    row: &LibpqOAuthRow,
    index: usize,
    field: &'static str,
) -> Result<Option<uuid::Uuid>, PgMetaError> {
    row.get(index)
        .map(|value| {
            uuid::Uuid::parse_str(value).map_err(|error| PgMetaError::Decode {
                field,
                value: value.to_string(),
                message: error.to_string(),
            })
        })
        .transpose()
}

fn oauth_bool(row: &LibpqOAuthRow, index: usize, field: &'static str) -> Result<bool, PgMetaError> {
    let value = oauth_required(row, index, field)?;
    match value.trim().to_ascii_lowercase().as_str() {
        "t" | "true" | "1" => Ok(true),
        "f" | "false" | "0" => Ok(false),
        _ => Err(PgMetaError::Decode {
            field,
            value: value.to_string(),
            message: "expected PostgreSQL boolean text".to_string(),
        }),
    }
}

fn oauth_optional_bool(
    row: &LibpqOAuthRow,
    index: usize,
    field: &'static str,
) -> Result<Option<bool>, PgMetaError> {
    row.get(index)
        .map(|value| parse_oauth_bool(value, field))
        .transpose()
}

fn parse_oauth_bool(value: &str, field: &'static str) -> Result<bool, PgMetaError> {
    match value.trim().to_ascii_lowercase().as_str() {
        "t" | "true" | "1" => Ok(true),
        "f" | "false" | "0" => Ok(false),
        _ => Err(PgMetaError::Decode {
            field,
            value: value.to_string(),
            message: "expected PostgreSQL boolean text".to_string(),
        }),
    }
}

fn oauth_optional_timestamp(
    row: &LibpqOAuthRow,
    index: usize,
    field: &'static str,
) -> Result<Option<DateTime<Utc>>, PgMetaError> {
    row.get(index)
        .map(|value| {
            NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S%.f")
                .map(|value| DateTime::from_naive_utc_and_offset(value, Utc))
                .map_err(|error| PgMetaError::Decode {
                    field,
                    value: value.to_string(),
                    message: error.to_string(),
                })
        })
        .transpose()
}

fn oauth_optional_date(
    row: &LibpqOAuthRow,
    index: usize,
    field: &'static str,
) -> Result<Option<NaiveDate>, PgMetaError> {
    row.get(index)
        .map(|value| {
            NaiveDate::parse_from_str(value, "%Y-%m-%d").map_err(|error| PgMetaError::Decode {
                field,
                value: value.to_string(),
                message: error.to_string(),
            })
        })
        .transpose()
}

fn oauth_optional_hex(
    row: &LibpqOAuthRow,
    index: usize,
    field: &'static str,
) -> Result<Option<Vec<u8>>, PgMetaError> {
    row.get(index)
        .map(|value| parse_hex(value, field))
        .transpose()
}

fn parse_hex(value: &str, field: &'static str) -> Result<Vec<u8>, PgMetaError> {
    if !value.len().is_multiple_of(2) {
        return Err(PgMetaError::Decode {
            field,
            value: value.to_string(),
            message: "hex bytea output had odd length".to_string(),
        });
    }
    let mut bytes = Vec::with_capacity(value.len() / 2);
    for chunk in value.as_bytes().chunks_exact(2) {
        let chunk = std::str::from_utf8(chunk).map_err(|error| PgMetaError::Decode {
            field,
            value: value.to_string(),
            message: error.to_string(),
        })?;
        let byte = u8::from_str_radix(chunk, 16).map_err(|error| PgMetaError::Decode {
            field,
            value: value.to_string(),
            message: error.to_string(),
        })?;
        bytes.push(byte);
    }
    Ok(bytes)
}

fn asset_type_sql(value: AssetType) -> &'static str {
    match value {
        AssetType::Dataset => "dataset",
        AssetType::File => "file",
    }
}

fn key_role_sql(value: KeyRole) -> Result<&'static str, PgMetaError> {
    match value {
        KeyRole::None => Ok("none"),
        KeyRole::Record => Ok("record"),
        KeyRole::External => Ok("external"),
    }
}

fn transformation_type_sql(value: TransformationType) -> &'static str {
    match value {
        TransformationType::Ingest => "ingest",
        TransformationType::Transform => "transform",
        TransformationType::Entity => "entity",
        TransformationType::Export => "export",
        TransformationType::Repository => "repository",
    }
}

fn parse_asset_type_text(value: String) -> Result<AssetType, PgMetaError> {
    match value.as_str() {
        "dataset" => Ok(AssetType::Dataset),
        "file" => Ok(AssetType::File),
        _ => Err(PgMetaError::Decode {
            field: "asset_type",
            value,
            message: "expected dataset or file".to_string(),
        }),
    }
}

fn parse_key_role_text(value: String) -> Result<KeyRole, PgMetaError> {
    match value.as_str() {
        "none" => Ok(KeyRole::None),
        "record" => Ok(KeyRole::Record),
        "external" => Ok(KeyRole::External),
        _ => Err(PgMetaError::Decode {
            field: "keyrole",
            value,
            message: "expected none, record, or external".to_string(),
        }),
    }
}

fn parse_transformation_type_text(value: String) -> Result<TransformationType, PgMetaError> {
    match value.as_str() {
        "ingest" => Ok(TransformationType::Ingest),
        "transform" => Ok(TransformationType::Transform),
        "entity" => Ok(TransformationType::Entity),
        "export" => Ok(TransformationType::Export),
        "repository" => Ok(TransformationType::Repository),
        _ => Err(PgMetaError::Decode {
            field: "transformation_type",
            value,
            message: "expected ingest, transform, entity, export, or repository".to_string(),
        }),
    }
}

fn oauth_required<'a>(
    row: &'a LibpqOAuthRow,
    index: usize,
    field: &'static str,
) -> Result<&'a str, PgMetaError> {
    row.get(index).ok_or_else(|| PgMetaError::Decode {
        field,
        value: "NULL".to_string(),
        message: "required metadata field was null or missing".to_string(),
    })
}

fn study_matches_search_query(
    study: &StudyRecord,
    domains: &[DomainRecord],
    tags: &[TagRecord],
    query: &StudySearchQuery,
) -> bool {
    let matches = |predicate: &StudySearchPredicate| -> bool {
        match predicate {
            StudySearchPredicate::Text { field, mode, value } => {
                let candidate = match field {
                    StudySearchTextField::Name => Some(study.name.as_str()),
                    StudySearchTextField::Title => study.documentation.as_deref(),
                    StudySearchTextField::Description => study.description.as_deref(),
                };
                candidate.is_some_and(|candidate| search_text_matches(candidate, *mode, value))
            }
            StudySearchPredicate::Domain(selector) => domains.iter().any(|domain| match selector {
                StudySearchDomainSelector::Id(domain_id) => domain.domain_id == *domain_id,
                StudySearchDomainSelector::Name(name) => domain.name.as_str() == name,
            }),
            StudySearchPredicate::Tag(selector) => search_tags_match_selector(tags, selector),
        }
    };

    predicate_set_matches(query.kind, &query.predicates, matches)
}

fn study_search_sort_key(left: &StudyRecord, right: &StudyRecord) -> std::cmp::Ordering {
    study_sort_tuple(left).cmp(&study_sort_tuple(right))
}

fn study_sort_tuple(study: &StudyRecord) -> (String, String, uuid::Uuid) {
    (
        study.name.as_str().to_ascii_lowercase(),
        study.name.as_str().to_string(),
        study.study_id.0,
    )
}

fn study_cursor_tuple(anchor: &StudySearchCursorAnchor) -> (String, String, uuid::Uuid) {
    (
        anchor.name.to_ascii_lowercase(),
        anchor.name.clone(),
        anchor.study_id.0,
    )
}

fn latest_searchable_dataset_version(
    versions: &[AssetVersionRecord],
    dataset_ids: &[uuid::Uuid],
    withdrawn_ids: &[uuid::Uuid],
) -> Option<AssetVersionRecord> {
    versions
        .iter()
        .filter(|version| {
            dataset_ids.contains(&version.version_id.0)
                && version.is_latest == Some(true)
                && !withdrawn_ids.contains(&version.version_id.0)
        })
        .max_by_key(|version| (version.major, version.minor, version.patch))
        .cloned()
        .map(|mut version| {
            version.is_latest = Some(true);
            version
        })
}

fn latest_searchable_datafile_version(
    versions: &[AssetVersionRecord],
    datafiles: &[DataFileRecord],
) -> Option<(AssetVersionRecord, DataFileRecord)> {
    versions
        .iter()
        .filter(|version| {
            version.is_latest == Some(true)
                && datafiles
                    .iter()
                    .any(|datafile| datafile.datafile_id == version.version_id)
        })
        .max_by_key(|version| (version.major, version.minor, version.patch))
        .and_then(|version| {
            datafiles
                .iter()
                .find(|datafile| datafile.datafile_id == version.version_id)
                .map(|datafile| {
                    let mut version = version.clone();
                    version.is_latest = Some(true);
                    (version, datafile.clone())
                })
        })
}

fn dataset_matches_search_query(
    study: &StudyRecord,
    asset: &AssetRecord,
    latest_version: &AssetVersionRecord,
    domains: &[DomainRecord],
    tags: &[TagRecord],
    query: &DatasetSearchQuery,
) -> bool {
    let matches = |predicate: &DatasetSearchPredicate| -> bool {
        match predicate {
            DatasetSearchPredicate::Text { field, mode, value } => {
                let candidate = match field {
                    DatasetSearchTextField::Name => Some(asset.name.as_str()),
                    DatasetSearchTextField::Description => asset.description.as_deref(),
                    DatasetSearchTextField::VersionLabel => latest_version.version_label.as_deref(),
                };
                candidate.is_some_and(|candidate| search_text_matches(candidate, *mode, value))
            }
            DatasetSearchPredicate::Study(selector) => match selector {
                DatasetSearchStudySelector::Id(study_id) => study.study_id == *study_id,
                DatasetSearchStudySelector::Name { name, domain } => {
                    study.name.as_str() == name
                        && domain.as_ref().is_none_or(|domain_selector| {
                            domains.iter().any(|domain| match domain_selector {
                                DatasetSearchDomainSelector::Id(domain_id) => {
                                    domain.domain_id == *domain_id
                                }
                                DatasetSearchDomainSelector::Name(domain_name) => {
                                    domain.name.as_str() == domain_name
                                }
                            })
                        })
                }
            },
            DatasetSearchPredicate::Domain(selector) => {
                domains.iter().any(|domain| match selector {
                    DatasetSearchDomainSelector::Id(domain_id) => domain.domain_id == *domain_id,
                    DatasetSearchDomainSelector::Name(name) => domain.name.as_str() == name,
                })
            }
            DatasetSearchPredicate::VersionLabel(label) => latest_version
                .version_label
                .as_deref()
                .is_some_and(|version_label| version_label == label),
            DatasetSearchPredicate::Tag(selector) => search_tags_match_selector(tags, selector),
        }
    };

    predicate_set_matches(query.kind, &query.predicates, matches)
}

fn datafile_matches_search_query(
    study: &StudyRecord,
    asset: &AssetRecord,
    datafile: &DataFileRecord,
    tags: &[TagRecord],
    query: &DataFileSearchQuery,
) -> bool {
    let matches = |predicate: &DataFileSearchPredicate| -> bool {
        match predicate {
            DataFileSearchPredicate::Text { field, mode, value } => {
                let candidate = match field {
                    DataFileSearchTextField::Name => Some(asset.name.as_str()),
                    DataFileSearchTextField::Description => asset.description.as_deref(),
                    DataFileSearchTextField::Format => Some(datafile.edam_format.as_str()),
                };
                candidate.is_some_and(|candidate| search_text_matches(candidate, *mode, value))
            }
            DataFileSearchPredicate::Study(selector) => match selector {
                DataFileSearchStudySelector::Id(study_id) => study.study_id == *study_id,
                DataFileSearchStudySelector::Name { name } => study.name.as_str() == name,
            },
            DataFileSearchPredicate::Format(format) => datafile.edam_format == *format,
            DataFileSearchPredicate::Tag(selector) => search_tags_match_selector(tags, selector),
        }
    };

    predicate_set_matches(query.kind, &query.predicates, matches)
}

fn dataset_search_sort_key(
    left: &DatasetSearchRecord,
    right: &DatasetSearchRecord,
) -> std::cmp::Ordering {
    dataset_sort_tuple(left).cmp(&dataset_sort_tuple(right))
}

fn dataset_sort_tuple(dataset: &DatasetSearchRecord) -> (String, String, uuid::Uuid, uuid::Uuid) {
    (
        dataset.asset.name.as_str().to_ascii_lowercase(),
        dataset.asset.name.as_str().to_string(),
        dataset.study_id.0,
        dataset.asset.asset_id.0,
    )
}

fn dataset_cursor_tuple(
    anchor: &DatasetSearchCursorAnchor,
) -> (String, String, uuid::Uuid, uuid::Uuid) {
    (
        anchor.name.to_ascii_lowercase(),
        anchor.name.clone(),
        anchor.study_id.0,
        anchor.asset_id.0,
    )
}

fn datafile_search_sort_key(
    left: &DataFileSearchRecord,
    right: &DataFileSearchRecord,
) -> std::cmp::Ordering {
    datafile_sort_tuple(left).cmp(&datafile_sort_tuple(right))
}

fn datafile_sort_tuple(
    datafile: &DataFileSearchRecord,
) -> (String, String, uuid::Uuid, uuid::Uuid) {
    (
        datafile.asset.name.as_str().to_ascii_lowercase(),
        datafile.asset.name.as_str().to_string(),
        datafile.study_id.0,
        datafile.asset.asset_id.0,
    )
}

fn datafile_cursor_tuple(
    anchor: &DataFileSearchCursorAnchor,
) -> (String, String, uuid::Uuid, uuid::Uuid) {
    (
        anchor.name.to_ascii_lowercase(),
        anchor.name.clone(),
        anchor.study_id.0,
        anchor.asset_id.0,
    )
}

fn variable_matches_search_query(
    query: &VariableSearchQuery,
    domain: &DomainRecord,
    variable: &VariableRecord,
    value_type: Option<&ValueTypeRecord>,
    vocabulary: Option<&VocabularyRecord>,
    tags: &[TagRecord],
) -> bool {
    let matches = |predicate: &VariableSearchPredicate| -> bool {
        match predicate {
            VariableSearchPredicate::Text { field, mode, value } => {
                let candidate = match field {
                    VariableSearchTextField::Name => Some(variable.name.as_str()),
                    VariableSearchTextField::Description => variable.description.as_deref(),
                    VariableSearchTextField::ValueType => {
                        value_type.map(|value_type| value_type.value_type.as_str())
                    }
                    VariableSearchTextField::VocabularyName => {
                        vocabulary.map(|vocabulary| vocabulary.name.as_str())
                    }
                    VariableSearchTextField::OntologyClass => variable.ontology_class.as_deref(),
                };
                candidate.is_some_and(|candidate| search_text_matches(candidate, *mode, value))
            }
            VariableSearchPredicate::Domain(selector) => match selector {
                VariableSearchDomainSelector::Id(domain_id) => domain.domain_id == *domain_id,
                VariableSearchDomainSelector::Name(name) => domain.name.as_str() == name,
            },
            VariableSearchPredicate::ValueType(expected) => value_type
                .is_some_and(|value_type| value_type.value_type.eq_ignore_ascii_case(expected)),
            VariableSearchPredicate::Vocabulary(selector) => match selector {
                VariableSearchVocabularySelector::Id(vocabulary_id) => {
                    variable.vocabulary_id == Some(*vocabulary_id)
                }
                VariableSearchVocabularySelector::Name {
                    domain: vocabulary_domain,
                    name,
                } => {
                    let Some(vocabulary) = vocabulary else {
                        return false;
                    };
                    let domain_matches = match vocabulary_domain {
                        None => true,
                        Some(VariableSearchDomainSelector::Id(domain_id)) => {
                            vocabulary.domain_id == *domain_id
                        }
                        Some(VariableSearchDomainSelector::Name(domain_name)) => {
                            domain.name.as_str() == domain_name
                        }
                    };
                    domain_matches && vocabulary.name.as_str() == name
                }
            },
            VariableSearchPredicate::KeyRole(key_role) => variable.key_role == *key_role,
            VariableSearchPredicate::Tag(selector) => search_tags_match_selector(tags, selector),
        }
    };
    predicate_set_matches(query.kind, &query.predicates, matches)
}

fn variable_search_sort_key(
    left: &VariableSearchRecord,
    right: &VariableSearchRecord,
) -> std::cmp::Ordering {
    variable_sort_tuple(left).cmp(&variable_sort_tuple(right))
}

fn variable_sort_tuple(variable: &VariableSearchRecord) -> (String, String, i64, i64) {
    (
        variable.variable.name.as_str().to_ascii_lowercase(),
        variable.variable.name.as_str().to_string(),
        variable.variable.domain_id.0,
        variable.variable.variable_id.0,
    )
}

fn variable_cursor_tuple(anchor: &VariableSearchCursorAnchor) -> (String, String, i64, i64) {
    (
        anchor.name.to_ascii_lowercase(),
        anchor.name.clone(),
        anchor.domain_id.0,
        anchor.variable_id.0,
    )
}

fn entity_matches_search_query(
    query: &EntitySearchQuery,
    domain: &DomainRecord,
    entity: &EntityRecord,
    tags: &[TagRecord],
) -> bool {
    let matches = |predicate: &EntitySearchPredicate| -> bool {
        match predicate {
            EntitySearchPredicate::Text { field, mode, value } => {
                let candidate = match field {
                    EntitySearchTextField::Name => Some(entity.name.as_str()),
                    EntitySearchTextField::Description => entity.description.as_deref(),
                    EntitySearchTextField::OntologyClass => entity.ontology_class.as_deref(),
                };
                candidate.is_some_and(|candidate| search_text_matches(candidate, *mode, value))
            }
            EntitySearchPredicate::Domain(selector) => match selector {
                EntitySearchDomainSelector::Id(domain_id) => domain.domain_id == *domain_id,
                EntitySearchDomainSelector::Name(name) => domain.name.as_str() == name,
            },
            EntitySearchPredicate::Tag(selector) => search_tags_match_selector(tags, selector),
        }
    };
    predicate_set_matches(query.kind, &query.predicates, matches)
}

fn entity_search_sort_key(
    left: &EntitySearchRecord,
    right: &EntitySearchRecord,
) -> std::cmp::Ordering {
    entity_sort_tuple(left).cmp(&entity_sort_tuple(right))
}

fn entity_sort_tuple(entity: &EntitySearchRecord) -> (String, String, i64, i64) {
    (
        entity.entity.name.as_str().to_ascii_lowercase(),
        entity.entity.name.as_str().to_string(),
        entity.entity.domain_id.0,
        entity.entity.entity_id.0,
    )
}

fn entity_cursor_tuple(anchor: &EntitySearchCursorAnchor) -> (String, String, i64, i64) {
    (
        anchor.name.to_ascii_lowercase(),
        anchor.name.clone(),
        anchor.domain_id.0,
        anchor.entity_id.0,
    )
}

fn relation_matches_search_query(
    query: &RelationSearchQuery,
    domain: &DomainRecord,
    relation: &EntityRelationRecord,
    source_entity: &EntityRecord,
    target_entity: &EntityRecord,
    tags: &[TagRecord],
) -> bool {
    let matches = |predicate: &RelationSearchPredicate| -> bool {
        match predicate {
            RelationSearchPredicate::Text { field, mode, value } => {
                let candidate = match field {
                    RelationSearchTextField::Name => Some(relation.name.as_str()),
                    RelationSearchTextField::Description => relation.description.as_deref(),
                    RelationSearchTextField::SourceEntityName => Some(source_entity.name.as_str()),
                    RelationSearchTextField::TargetEntityName => Some(target_entity.name.as_str()),
                    RelationSearchTextField::OntologyClass => relation.ontology_class.as_deref(),
                };
                candidate.is_some_and(|candidate| search_text_matches(candidate, *mode, value))
            }
            RelationSearchPredicate::Domain(selector) => {
                relation_search_domain_matches(domain, selector)
            }
            RelationSearchPredicate::SourceEntity(selector) => {
                relation_search_entity_matches(domain, source_entity, selector)
            }
            RelationSearchPredicate::TargetEntity(selector) => {
                relation_search_entity_matches(domain, target_entity, selector)
            }
            RelationSearchPredicate::Tag(selector) => search_tags_match_selector(tags, selector),
        }
    };
    predicate_set_matches(query.kind, &query.predicates, matches)
}

fn relation_search_domain_matches(
    domain: &DomainRecord,
    selector: &RelationSearchDomainSelector,
) -> bool {
    match selector {
        RelationSearchDomainSelector::Id(domain_id) => domain.domain_id == *domain_id,
        RelationSearchDomainSelector::Name(name) => domain.name.as_str() == name,
    }
}

fn relation_search_entity_matches(
    domain: &DomainRecord,
    entity: &EntityRecord,
    selector: &RelationSearchEntitySelector,
) -> bool {
    match selector {
        RelationSearchEntitySelector::Id(entity_id) => entity.entity_id == *entity_id,
        RelationSearchEntitySelector::Name {
            domain: selector_domain,
            name,
        } => {
            entity.name.as_str() == name
                && selector_domain
                    .as_ref()
                    .is_none_or(|selector| relation_search_domain_matches(domain, selector))
        }
    }
}

fn relation_search_sort_key(
    left: &RelationSearchRecord,
    right: &RelationSearchRecord,
) -> std::cmp::Ordering {
    relation_sort_tuple(left).cmp(&relation_sort_tuple(right))
}

fn relation_sort_tuple(relation: &RelationSearchRecord) -> (String, String, i64, i64) {
    (
        relation.relation.name.as_str().to_ascii_lowercase(),
        relation.relation.name.as_str().to_string(),
        relation.relation.domain_id.0,
        relation.relation.entity_relation_id.0,
    )
}

fn relation_cursor_tuple(anchor: &RelationSearchCursorAnchor) -> (String, String, i64, i64) {
    (
        anchor.name.to_ascii_lowercase(),
        anchor.name.clone(),
        anchor.domain_id.0,
        anchor.relation_id.0,
    )
}

fn domain_record(read_model: PgDomainReadModel) -> Result<DomainRecord, CoreError> {
    Ok(DomainRecord {
        domain_id: read_model.domain_id,
        name: nc_name("domains.name", read_model.name)?,
        description: read_model.description,
        uri: read_model.uri,
    })
}

fn study_record(read_model: PgStudyReadModel) -> Result<StudyRecord, CoreError> {
    Ok(StudyRecord {
        study_id: read_model.study_id,
        name: nc_name("studies.name", read_model.name)?,
        description: read_model.description,
        documentation: read_model.documentation,
        agent_instructions: read_model.agent_instructions,
        external_id: read_model.external_id,
        study_type_id: read_model.study_type_id,
        date_created: read_model.date_created,
        created_by: read_model.created_by,
    })
}

fn study_access_record(
    read_model: PgStudyAccessReadModel,
) -> Result<StudyAccessGrantRecord, CoreError> {
    Ok(StudyAccessGrantRecord {
        study_id: read_model.study_id,
        user_id: read_model.user_id,
        date_granted: read_model.date_granted,
        granted_by: read_model.granted_by,
    })
}

fn study_custodian_record(
    read_model: PgStudyCustodianReadModel,
) -> Result<StudyCustodianLinkRecord, CoreError> {
    Ok(StudyCustodianLinkRecord {
        study_id: read_model.study_id,
        user_id: read_model.user_id,
        is_primary: read_model.is_primary,
        date_granted: read_model.date_granted,
        granted_by: read_model.granted_by,
    })
}

fn study_duo_restriction_record(
    read_model: PgDuoRestrictionReadModel,
) -> Result<StudyDuoRestrictionRecord, CoreError> {
    Ok(StudyDuoRestrictionRecord {
        study_duo_restriction_id: StudyDuoRestrictionId(read_model.restriction_id),
        study_id: read_model.study_id,
        duo_id: read_model.duo_id,
        qualifier_ontology_namespace: read_model.qualifier_ontology_namespace,
        qualifier_ontology_id: read_model.qualifier_ontology_id,
        qualifier_text: read_model.qualifier_text,
        qualifier_date: read_model.qualifier_date,
        qualifier_integer: read_model.qualifier_integer,
    })
}

fn asset_record(read_model: PgAssetReadModel) -> Result<AssetRecord, CoreError> {
    Ok(AssetRecord {
        asset_id: read_model.asset_id,
        study_id: read_model.study_id,
        name: nc_name("assets.name", read_model.name)?,
        description: read_model.description,
        agent_instructions: read_model.agent_instructions,
        asset_type: read_model.asset_type,
        date_created: read_model.date_created,
        created_by: read_model.created_by,
    })
}

fn asset_version_record(read_model: PgAssetVersionReadModel) -> AssetVersionRecord {
    AssetVersionRecord {
        version_id: read_model.version_id,
        asset_id: read_model.asset_id,
        major: read_model.major,
        minor: read_model.minor,
        patch: read_model.patch,
        version_label: read_model.version_label,
        version_note: read_model.version_note,
        is_latest: read_model.is_latest,
        doi: read_model.doi,
        created_at: read_model.created_at,
        created_by: read_model.created_by,
    }
}

fn asset_version_duo_restriction_record(
    read_model: PgAssetVersionDuoRestrictionReadModel,
) -> Result<AssetVersionDuoRestrictionRecord, CoreError> {
    Ok(AssetVersionDuoRestrictionRecord {
        asset_version_duo_restriction_id: AssetVersionDuoRestrictionId(read_model.restriction_id),
        version_id: read_model.version_id,
        duo_id: read_model.duo_id,
        qualifier_ontology_namespace: read_model.qualifier_ontology_namespace,
        qualifier_ontology_id: read_model.qualifier_ontology_id,
        qualifier_text: read_model.qualifier_text,
        qualifier_date: read_model.qualifier_date,
        qualifier_integer: read_model.qualifier_integer,
    })
}

fn datafile_record(read_model: PgDataFileReadModel) -> Result<DataFileRecord, CoreError> {
    Ok(DataFileRecord {
        datafile_id: read_model.datafile_id,
        size_bytes: read_model
            .size_bytes
            .map(u64::try_from)
            .transpose()
            .map_err(|_| CoreError::Infrastructure("datafile size is negative".to_string()))?,
        compressed: read_model.compressed,
        encrypted: read_model.encrypted,
        compression_algorithm: read_model.compression_algorithm,
        encryption_algorithm: read_model.encryption_algorithm,
        encryption_key: read_model.encryption_key,
        storage_uri: read_model.storage_uri,
        edam_format: read_model.edam_format,
        digest: Sha256Hex::parse(read_model.digest)
            .map_err(|error| CoreError::Infrastructure(error.to_string()))?,
    })
}

fn dataset_record(read_model: PgDatasetReadModel) -> Result<DatasetRecord, CoreError> {
    Ok(DatasetRecord {
        dataset_id: read_model.dataset_id,
    })
}

fn variable_record(read_model: PgVariableReadModel) -> Result<VariableRecord, CoreError> {
    Ok(VariableRecord {
        variable_id: read_model.variable_id,
        domain_id: read_model.domain_id,
        name: nc_name("variables.name", read_model.name)?,
        value_type_id: read_model.value_type_id,
        value_format: read_model.value_format,
        vocabulary_id: read_model.vocabulary_id,
        key_role: read_model.key_role,
        description: read_model.description,
        note: read_model.note,
        ontology_namespace: read_model.ontology_namespace,
        ontology_class: read_model.ontology_class,
    })
}

fn value_type_record(read_model: PgValueTypeReadModel) -> Result<ValueTypeRecord, CoreError> {
    Ok(ValueTypeRecord {
        value_type_id: read_model.value_type_id,
        value_type: read_model.value_type,
        description: read_model.description,
    })
}

fn vocabulary_record(read_model: PgVocabularyReadModel) -> Result<VocabularyRecord, CoreError> {
    Ok(VocabularyRecord {
        vocabulary_id: read_model.vocabulary_id,
        domain_id: read_model.domain_id,
        name: nc_name("vocabularies.name", read_model.name)?,
        description: read_model.description,
    })
}

fn vocabulary_item_record(read_model: PgVocabularyItemReadModel) -> VocabularyItemRecord {
    VocabularyItemRecord {
        vocabulary_item_id: read_model.vocabulary_item_id,
        vocabulary_id: read_model.vocabulary_id,
        value: read_model.value,
        code: read_model.code,
        description: read_model.description,
    }
}

fn vocabulary_mapping_record(
    read_model: crate::PgVocabularyMappingReadModel,
) -> VocabularyMappingRecord {
    VocabularyMappingRecord {
        vocabulary_mapping_id: read_model.vocabulary_mapping_id,
        from_vocabulary_item: read_model.from_vocabulary_item,
        to_vocabulary_item: read_model.to_vocabulary_item,
    }
}

fn dataset_variable_link_record(
    read_model: PgDatasetVariableReadModel,
) -> ahri_tre_types::DatasetVariableLinkRecord {
    ahri_tre_types::DatasetVariableLinkRecord {
        dataset_id: read_model.dataset_id,
        variable_id: read_model.variable_id,
        row_role: read_model.row_role,
    }
}

fn entity_record(read_model: PgEntityReadModel) -> Result<EntityRecord, CoreError> {
    Ok(EntityRecord {
        entity_id: read_model.entity_id,
        domain_id: read_model.domain_id,
        uuid: read_model.uuid,
        name: nc_name("entities.name", read_model.name)?,
        description: read_model.description,
        agent_instructions: read_model.agent_instructions,
        ontology_namespace: read_model.ontology_namespace,
        ontology_class: read_model.ontology_class,
    })
}

fn entity_relation_record(
    read_model: PgEntityRelationReadModel,
) -> Result<EntityRelationRecord, CoreError> {
    Ok(EntityRelationRecord {
        entity_relation_id: read_model.entity_relation_id,
        subject_entity_id: read_model.subject_entity_id,
        object_entity_id: read_model.object_entity_id,
        domain_id: read_model.domain_id,
        uuid: read_model.uuid,
        name: nc_name("entityrelations.name", read_model.name)?,
        description: read_model.description,
        agent_instructions: read_model.agent_instructions,
        ontology_namespace: read_model.ontology_namespace,
        ontology_class: read_model.ontology_class,
    })
}

fn entity_instance_record(read_model: PgEntityInstanceReadModel) -> EntityInstanceRecord {
    EntityInstanceRecord {
        instance_id: read_model.instance_id,
        uuid: read_model.uuid,
        entity_id: read_model.entity_id,
        label: read_model.label,
        note: read_model.note,
        transformation_id: read_model.transformation_id,
    }
}

fn study_entity_instance_record(
    read_model: PgStudyEntityInstanceReadModel,
) -> StudyEntityInstanceRecord {
    StudyEntityInstanceRecord {
        study_id: read_model.study_id,
        entity_instance_id: read_model.entity_instance_id,
        entity_id: read_model.entity_id,
        external_id: read_model.external_id,
        transformation_id: read_model.transformation_id,
    }
}

fn relation_instance_record(read_model: PgRelationInstanceReadModel) -> RelationInstanceRecord {
    RelationInstanceRecord {
        relation_instance_id: read_model.relation_instance_id,
        uuid: read_model.uuid,
        entity_relation_id: read_model.entity_relation_id,
        entity_instance_id_1: read_model.entity_instance_id_1,
        entity_instance_id_2: read_model.entity_instance_id_2,
        valid_from: read_model.valid_from,
        valid_to: read_model.valid_to,
        note: read_model.note,
        transformation_id: read_model.transformation_id,
    }
}

fn study_relation_instance_record(
    read_model: PgStudyRelationInstanceReadModel,
) -> StudyRelationInstanceRecord {
    StudyRelationInstanceRecord {
        study_id: read_model.study_id,
        relation_instance_id: read_model.relation_instance_id,
        entity_relation_id: read_model.entity_relation_id,
        external_id: read_model.external_id,
        transformation_id: read_model.transformation_id,
    }
}

#[allow(dead_code)]
fn asset_version_entity_record(
    read_model: PgAssetVersionEntityReadModel,
) -> AssetVersionEntityRecord {
    AssetVersionEntityRecord {
        version_id: read_model.version_id,
        entity_instance_id: read_model.entity_instance_id,
        transformation_id: read_model.transformation_id,
    }
}

#[allow(dead_code)]
fn dataset_version_entity_record(
    read_model: PgDatasetVersionEntityReadModel,
) -> DatasetVersionEntityRecord {
    DatasetVersionEntityRecord {
        version_id: read_model.version_id,
        entity_instance_id: read_model.entity_instance_id,
        entity_variable_id: read_model.entity_variable_id,
        transformation_id: read_model.transformation_id,
    }
}

#[allow(dead_code)]
fn asset_version_relation_instance_record(
    read_model: PgAssetVersionRelationInstanceReadModel,
) -> AssetVersionRelationInstanceRecord {
    AssetVersionRelationInstanceRecord {
        version_id: read_model.version_id,
        relation_instance_id: read_model.relation_instance_id,
        transformation_id: read_model.transformation_id,
    }
}

#[allow(dead_code)]
fn dataset_version_relation_instance_record(
    read_model: PgDatasetVersionRelationInstanceReadModel,
) -> DatasetVersionRelationInstanceRecord {
    DatasetVersionRelationInstanceRecord {
        version_id: read_model.version_id,
        relation_instance_id: read_model.relation_instance_id,
        subject_variable_id: read_model.subject_variable_id,
        object_variable_id: read_model.object_variable_id,
        relation_variable_id: read_model.relation_variable_id,
        transformation_id: read_model.transformation_id,
    }
}

fn transformation_record(
    read_model: PgTransformationReadModel,
) -> Result<TransformationRecord, CoreError> {
    Ok(TransformationRecord {
        transformation_id: read_model.transformation_id,
        transformation_type: read_model.transformation_type,
        description: read_model.description,
        repository_url: read_model.repository_url,
        commit_hash: read_model.commit_hash,
        file_path: read_model.file_path,
        date_created: read_model.date_created,
        created_by: read_model.created_by,
    })
}

fn transformation_input_record(
    read_model: PgTransformationInputReadModel,
) -> TransformationInputLinkRecord {
    TransformationInputLinkRecord {
        transformation_id: read_model.transformation_id,
        version_id: read_model.version_id,
    }
}

fn transformation_output_record(
    read_model: PgTransformationOutputReadModel,
) -> TransformationOutputLinkRecord {
    TransformationOutputLinkRecord {
        transformation_id: read_model.transformation_id,
        version_id: read_model.version_id,
    }
}

fn tag_record(read_model: PgTagReadModel) -> TagRecord {
    TagRecord {
        tag_id: TagId(read_model.tag_id),
        name: read_model.name,
    }
}

fn pg_create_study_duo_restriction(
    study_duo_restriction_id: Option<StudyDuoRestrictionId>,
    restriction: &NewStudyDuoRestrictionRecord,
) -> PgCreateStudyDuoRestriction {
    PgCreateStudyDuoRestriction {
        study_duo_restriction_id,
        study_id: restriction.study_id,
        duo_id: restriction.duo_id.clone(),
        qualifiers: PgDuoQualifiers {
            qualifier_ontology_namespace: restriction.qualifier_ontology_namespace.clone(),
            qualifier_ontology_id: restriction.qualifier_ontology_id.clone(),
            qualifier_text: restriction.qualifier_text.clone(),
            qualifier_date: restriction.qualifier_date,
            qualifier_integer: restriction.qualifier_integer,
        },
    }
}

fn pg_create_asset_version_duo_restriction(
    asset_version_duo_restriction_id: Option<AssetVersionDuoRestrictionId>,
    restriction: &NewAssetVersionDuoRestrictionRecord,
) -> PgCreateAssetVersionDuoRestriction {
    PgCreateAssetVersionDuoRestriction {
        asset_version_duo_restriction_id,
        version_id: restriction.version_id,
        duo_id: restriction.duo_id.clone(),
        qualifiers: PgDuoQualifiers {
            qualifier_ontology_namespace: restriction.qualifier_ontology_namespace.clone(),
            qualifier_ontology_id: restriction.qualifier_ontology_id.clone(),
            qualifier_text: restriction.qualifier_text.clone(),
            qualifier_date: restriction.qualifier_date,
            qualifier_integer: restriction.qualifier_integer,
        },
    }
}

fn pg_tag_target(target: TagAttachmentTarget) -> PgTagTarget {
    match target {
        TagAttachmentTarget::Study(id) => PgTagTarget::Study(id),
        TagAttachmentTarget::Domain(id) => PgTagTarget::Domain(id),
        TagAttachmentTarget::Variable(id) => PgTagTarget::Variable(id),
        TagAttachmentTarget::Asset(id) => PgTagTarget::Asset(id),
        TagAttachmentTarget::AssetVersion(id) => PgTagTarget::AssetVersion(id),
        TagAttachmentTarget::Entity(id) => PgTagTarget::Entity(id),
        TagAttachmentTarget::EntityRelation(id) => PgTagTarget::EntityRelation(id),
    }
}

fn pg_write_tag_target(target: TagAttachmentTarget) -> PgWriteTagTarget {
    match target {
        TagAttachmentTarget::Study(id) => PgWriteTagTarget::Study(id),
        TagAttachmentTarget::Domain(id) => PgWriteTagTarget::Domain(id),
        TagAttachmentTarget::Variable(id) => PgWriteTagTarget::Variable(id),
        TagAttachmentTarget::Asset(id) => PgWriteTagTarget::Asset(id),
        TagAttachmentTarget::AssetVersion(id) => PgWriteTagTarget::AssetVersion(id),
        TagAttachmentTarget::Entity(id) => PgWriteTagTarget::Entity(id),
        TagAttachmentTarget::EntityRelation(id) => PgWriteTagTarget::EntityRelation(id),
    }
}

fn nc_name(field: &'static str, value: String) -> Result<NcName, CoreError> {
    NcName::parse(value.clone()).map_err(|error| {
        CoreError::Infrastructure(format!(
            "could not decode PostgreSQL metadata field {field}: {error}"
        ))
    })
}

fn pgmeta_error(operation: &'static str, error: PgMetaError) -> CoreError {
    CoreError::Infrastructure(format!("PostgreSQL metadata {operation} failed: {error}"))
}

/// Retains the connection's upload deadline until work or compensation starts.
pub struct UploadMetadataDeadline {
    repository: PgMetadataRepository<'static>,
    stop: std::sync::mpsc::Sender<()>,
    worker: Option<std::thread::JoinHandle<()>>,
}
impl PgMetadataRepository<'static> {
    pub fn upload_deadline(
        &self,
        deadline: std::time::Instant,
        cancelled: std::sync::Arc<std::sync::atomic::AtomicBool>,
    ) -> Result<UploadMetadataDeadline, CoreError> {
        let millis = deadline
            .saturating_duration_since(std::time::Instant::now())
            .as_millis()
            .max(1)
            .to_string();
        let cancellation = self.with_session_connection(
            "bound upload metadata",
            |connection| {
                connection
                    .client
                    .query_one(
                        "SELECT set_config('statement_timeout',$1,FALSE)",
                        &[&millis],
                    )
                    .map_err(PgMetaError::Query)?;
                Ok(None)
            },
            |connection| {
                let token = connection
                    .cancellation()
                    .map_err(|source| PgMetaError::OAuth {
                        operation: "prepare cancellation",
                        source,
                    })?;
                connection.set_query_deadline(Some(deadline), Some(cancelled.clone()));
                Ok(Some(token))
            },
        )?;
        let (stop, stopped) = std::sync::mpsc::channel();
        let worker = cancellation.map(|mut token| {
            std::thread::spawn(move || {
                while stopped
                    .recv_timeout(std::time::Duration::from_millis(25))
                    .is_err()
                {
                    if std::time::Instant::now() >= deadline
                        || cancelled.load(std::sync::atomic::Ordering::Acquire)
                    {
                        token.cancel();
                        break;
                    }
                }
            })
        });
        Ok(UploadMetadataDeadline {
            repository: self.clone(),
            stop,
            worker,
        })
    }
}
impl Drop for UploadMetadataDeadline {
    fn drop(&mut self) {
        let _ = self.stop.send(());
        if let Some(worker) = self.worker.take() {
            let _ = worker.join();
        }
        let _ = self.repository.with_session_connection(
            "clear upload deadline",
            |connection| {
                connection
                    .client
                    .batch_execute("RESET statement_timeout")
                    .map_err(PgMetaError::Query)
            },
            |connection| {
                connection.set_query_deadline(None, None);
                connection
                    .execute_raw("RESET statement_timeout", &[])
                    .map(|_| ())
                    .map_err(|source| PgMetaError::OAuth {
                        operation: "clear upload deadline",
                        source,
                    })
            },
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn oauth_repository_key_role_contract_round_trips_external() {
        assert_eq!(
            key_role_sql(KeyRole::External).expect("external should encode"),
            "external"
        );
        assert_eq!(
            parse_key_role_text("external".to_string()).expect("external should decode"),
            KeyRole::External
        );
    }

    #[test]
    fn oauth_delegate_custodian_path_uses_named_function() {
        assert!(
            OAUTH_ADD_STUDY_DELEGATE_CUSTODIAN_SQL
                .contains("public.tre_add_study_delegate_custodian($1, $2)")
        );
    }

    #[test]
    fn oauth_primary_custodianship_transfer_path_uses_named_function() {
        assert!(
            OAUTH_TRANSFER_PRIMARY_CUSTODIANSHIP_SQL
                .contains("public.tre_transfer_study_primary_custodianship($1, $2)")
        );
    }

    #[test]
    fn oauth_remove_delegate_custodian_path_uses_named_function() {
        assert!(
            OAUTH_REMOVE_STUDY_DELEGATE_CUSTODIAN_SQL
                .contains("public.tre_remove_study_delegate_custodian($1, $2)")
        );
    }

    #[test]
    fn oauth_dataset_variable_query_orders_the_numeric_source_column() {
        assert_eq!(
            OAUTH_LIST_DATASET_VARIABLES_SQL
                .split_whitespace()
                .collect::<Vec<_>>()
                .join(" "),
            "SELECT dv.dataset_id::text, dv.variable_id::text, dv.row_role::text FROM \
             public.dataset_variables AS dv WHERE dv.dataset_id = $1 ORDER BY dv.variable_id"
        );
    }
}