ahri_tre_app/service/

metadata_admission.rs

use super::*;

// Reuses the same registration rules with owned Session repositories or with
// repositories borrowed for the duration of one adapter-owned admission.
impl<'repository> ScopedAppService<'repository> {
    pub fn new(repositories: ScopedAppRepositories<'repository>) -> Self {
        Self {
            semantic_repository: None,
            datastore_id: None,
            registrations: repositories.registrations,
            domains: repositories.domains,
            studies: repositories.studies,
            study_domains: repositories.study_domains,
            study_governance: repositories.study_governance,
            assets: repositories.assets,
            variables: repositories.variables,
            vocabularies: repositories.vocabularies,
            entities: repositories.entities,
            transformations: repositories.transformations,
            tags: repositories.tags,
        }
    }

    pub async fn get_asset(
        &self,
        request: GetAssetRequest,
    ) -> Result<AssetVersionCatalog, AppError> {
        let asset = match (request.asset_id, request.study_id, request.asset_name) {
            (Some(asset_id), None, None) => self
                .assets
                .get_asset_by_id(asset_id)
                .await
                .map_err(|error| core_error("lookup asset by id", error))?
                .ok_or_else(|| AppError::Validation(format!("asset not found: {}", asset_id.0)))?,
            (None, Some(study_id), Some(asset_name)) => self
                .assets
                .get_asset_by_name(study_id, asset_name.as_str())
                .await
                .map_err(|error| core_error("lookup asset by study and name", error))?
                .ok_or_else(|| {
                    AppError::Validation(format!(
                        "asset not found in study {}: {}",
                        study_id.0,
                        asset_name.as_str()
                    ))
                })?,
            (None, None, None) => {
                return Err(AppError::Validation(
                    "asset lookup requires an asset_id or study_id plus asset_name".to_string(),
                ));
            }
            _ => {
                return Err(AppError::Validation(
                    "asset lookup accepts either asset_id or study_id plus asset_name".to_string(),
                ));
            }
        };

        self.asset_version_catalog(asset).await
    }

    pub(super) async fn asset_version_catalog(
        &self,
        asset: ahri_tre_types::AssetRecord,
    ) -> Result<AssetVersionCatalog, AppError> {
        let versions = self
            .assets
            .list_asset_versions(asset.asset_id)
            .await
            .map_err(|error| core_error("list asset versions", error))?;
        let latest_version = self
            .latest_catalogue_version(&asset, &versions)
            .await
            .map_err(|error| core_error("resolve latest asset version", error))?;

        Ok(AssetVersionCatalog {
            asset,
            versions,
            latest_version,
        })
    }

    pub(super) async fn latest_catalogue_version(
        &self,
        asset: &ahri_tre_types::AssetRecord,
        versions: &[ahri_tre_types::AssetVersionRecord],
    ) -> Result<Option<ahri_tre_types::AssetVersionRecord>, CoreError> {
        // Final withdrawal retains authorized history but has no usable latest.
        // Other missing/multiple markers remain catalogue inconsistencies.
        if asset.asset_type == ahri_tre_types::AssetType::Dataset
            && !versions.is_empty()
            && versions
                .iter()
                .all(|version| version.is_latest != Some(true))
        {
            let withdrawals = self
                .assets
                .list_dataset_version_withdrawals(asset.asset_id)
                .await?;
            if versions.iter().all(|version| {
                withdrawals
                    .iter()
                    .any(|withdrawal| withdrawal.dataset_id == version.version_id)
            }) {
                return Ok(None);
            }
        }
        ahri_tre_core::latest_asset_version(versions)
    }

    pub(super) async fn write_dataset_metadata_in_admission(
        &self,
        prepared: &PreparedDatasetVersion,
        dataset: ahri_tre_types::DatasetRecord,
        transformation: &ahri_tre_types::NewTransformationRecord,
        input_version_ids: &[ahri_tre_types::VersionId],
        registration: DatasetMetadataRegistration<'_>,
    ) -> Result<
        (
            AssetVersionCatalog,
            ahri_tre_types::DatasetRecord,
            ahri_tre_types::TransformationLineageRecord,
            Vec<RegisteredDatasetVariable>,
        ),
        AppError,
    > {
        if let Some(source) = prepared.ordinary_upload_source {
            if input_version_ids != [source] {
                return Err(AppError::Validation(
                    "Uploaded source provenance changed".into(),
                ));
            }
            self.semantic_repository
                .as_ref()
                .ok_or_else(|| AppError::Infrastructure("Upload admission is unavailable".into()))?
                .validate_uploaded_table_source(
                    source,
                    prepared.asset.study_id,
                    prepared.classification.risk,
                )
                .map_err(|error| core_error("validate uploaded source classification", error))?;
        } else if !input_version_ids.is_empty() {
            self.assets
                .validate_dataset_derivation(
                    input_version_ids,
                    prepared.classification.risk,
                    prepared.inherit_risk,
                )
                .await
                .map_err(|error| core_error("validate derivation classification", error))?;
        }
        if !prepared.asset_persisted {
            self.assets
                .save_asset(&prepared.asset)
                .await
                .map_err(|error| core_error("save dataset asset", error))?;
        }
        self.attach_tags(
            TagAttachmentTarget::Asset(prepared.asset.asset_id),
            prepared.asset_tags.clone(),
        )
        .await?;

        if !prepared.version_persisted {
            self.assets
                .admit_asset_version(&prepared.version, &prepared.classification)
                .await
                .map_err(|error| core_error("save dataset asset version", error))?;
        }
        self.attach_tags(
            TagAttachmentTarget::AssetVersion(prepared.version.version_id),
            prepared.version_tags.clone(),
        )
        .await?;
        let dataset = self
            .assets
            .save_dataset(&dataset)
            .await
            .map_err(|error| core_error("save dataset record", error))?;
        if let Some(repository) = &self.semantic_repository {
            let mut sources = input_version_ids.to_vec();
            sources.push(prepared.version.version_id);
            sources.sort_by_key(|id| id.0);
            sources.dedup();
            repository
                .begin_semantic_derivation(&sources)
                .map_err(|error| core_error("retain Dataset dictionary sources", error))?;
        }
        if let Some((domain_id, forms)) = registration.redcap_dictionary {
            self.register_redcap_dictionary_variables(
                domain_id,
                forms,
                registration.force_metadata_updates,
            )
            .await?;
        }
        let registered_variables = self
            .register_dataset_variables_internal(
                dataset.dataset_id,
                first_registration_domain(&registration.variable_registrations),
                registration.variable_registrations,
                registration.loaded_column_names,
                registration.force_metadata_updates,
            )
            .await?;

        if let Some(repository) = &self.semantic_repository {
            for variable in &registered_variables {
                if let Some(vocabulary) = &variable.vocabulary {
                    repository
                        .retain_vocabulary_contribution(vocabulary.vocabulary_id)
                        .map_err(|error| core_error("retain reused dictionary sources", error))?;
                }
            }
        }

        let policy = DefaultProvenancePolicy;
        let transformation = enrich_transformation_source(transformation);
        let outputs = [prepared.version.version_id];
        let lineage = if input_version_ids.is_empty() {
            ahri_tre_core::record_ingest_provenance(
                self.transformations.as_ref(),
                &policy,
                &transformation,
                &outputs,
            )
            .await
            .map_err(|error| core_error("record dataset ingest provenance", error))?
        } else {
            ahri_tre_core::transform_assets(
                self.transformations.as_ref(),
                &policy,
                &transformation,
                input_version_ids,
                &outputs,
            )
            .await
            .map_err(|error| core_error("record dataset transform provenance", error))?
        };
        self.assets
            .mark_asset_version_latest(prepared.asset.asset_id, prepared.version.version_id)
            .await
            .map_err(|error| core_error("mark completed dataset version latest", error))?;
        let catalog = self
            .get_asset(GetAssetRequest {
                asset_id: Some(prepared.asset.asset_id),
                study_id: None,
                asset_name: None,
            })
            .await?;

        Ok((catalog, dataset, lineage, registered_variables))
    }

    async fn register_redcap_dictionary_variables(
        &self,
        domain_id: ahri_tre_types::DomainId,
        forms: &[RedcapForm],
        force_metadata_updates: bool,
    ) -> Result<Vec<String>, AppError> {
        let mut warnings = Vec::new();
        for form in forms {
            for field in &form.fields {
                let (registration, mut field_warnings) =
                    redcap_variable_registration(domain_id, field)?;
                warnings.append(&mut field_warnings);
                let DatasetVariableRegistrationInput::New(registration) = registration else {
                    unreachable!("REDCap registration always creates new metadata inputs");
                };
                let vocabulary = self
                    .register_or_validate_new_vocabulary(
                        domain_id,
                        registration.variable.vocabulary_id,
                        registration.vocabulary,
                        &registration.vocabulary_items,
                        force_metadata_updates,
                    )
                    .await?;
                let mut variable = registration.variable;
                if let Some(vocabulary) = vocabulary {
                    variable.vocabulary_id = Some(vocabulary.vocabulary_id);
                }
                self.register_or_validate_new_variable(
                    domain_id,
                    &variable,
                    force_metadata_updates,
                )
                .await?;
            }
        }
        Ok(warnings)
    }

    pub(super) async fn register_dataset_variables_internal(
        &self,
        dataset_id: ahri_tre_types::VersionId,
        request_domain_id: Option<ahri_tre_types::DomainId>,
        variable_registrations: Vec<DatasetVariableRegistrationInput>,
        loaded_column_names: Option<&[String]>,
        force_metadata_updates: bool,
    ) -> Result<Vec<RegisteredDatasetVariable>, AppError> {
        if variable_registrations.is_empty() {
            return Ok(Vec::new());
        }
        let domain_id = request_domain_id.ok_or_else(|| {
            AppError::Validation("variable registration requires a domain_id".to_string())
        })?;
        let loaded_column_names = loaded_column_names.map(|columns| {
            columns
                .iter()
                .map(|column| column.to_ascii_lowercase())
                .collect::<std::collections::BTreeSet<_>>()
        });

        let mut registered = Vec::with_capacity(variable_registrations.len());
        for registration in variable_registrations {
            let (variable_name, validation_column_name, variable_domain_id, variable_key_role) =
                match &registration {
                    DatasetVariableRegistrationInput::Explicit(registration) => (
                        registration.variable.name.as_str(),
                        registration.variable.name.as_str(),
                        registration.variable.domain_id,
                        registration.variable.key_role,
                    ),
                    DatasetVariableRegistrationInput::New(registration) => (
                        registration.variable.name.as_str(),
                        registration
                            .source_column_name
                            .as_deref()
                            .unwrap_or_else(|| registration.variable.name.as_str()),
                        registration.variable.domain_id,
                        registration.variable.key_role,
                    ),
                };
            if variable_domain_id != domain_id {
                return Err(AppError::Validation(format!(
                    "variable {} belongs to domain {}, not {}",
                    variable_name, variable_domain_id.0, domain_id.0
                )));
            }
            if let Some(column_names) = &loaded_column_names
                && !column_names.contains(&validation_column_name.to_ascii_lowercase())
            {
                return Err(AppError::Validation(format!(
                    "dataset column not found for variable {}",
                    variable_name
                )));
            }
            validate_metadata_key_role("variable key_role", variable_key_role)?;
            let row_role = match &registration {
                DatasetVariableRegistrationInput::Explicit(registration) => {
                    registration.row_role.unwrap_or(variable_key_role)
                }
                DatasetVariableRegistrationInput::New(registration) => {
                    registration.row_role.unwrap_or(variable_key_role)
                }
            };
            validate_metadata_key_role("dataset row_role", row_role)?;

            let (variable, vocabulary) = match registration {
                DatasetVariableRegistrationInput::Explicit(registration) => {
                    let vocabulary = self
                        .register_or_validate_vocabulary(
                            domain_id,
                            registration.variable.vocabulary_id,
                            registration.vocabulary,
                            &registration.vocabulary_items,
                            force_metadata_updates,
                        )
                        .await?;
                    let mut variable_record = registration.variable;
                    if let Some(vocabulary) = &vocabulary {
                        variable_record.vocabulary_id = Some(vocabulary.vocabulary_id);
                    }
                    let variable = self
                        .register_or_validate_variable(
                            domain_id,
                            &variable_record,
                            force_metadata_updates,
                        )
                        .await?;
                    (variable, vocabulary)
                }
                DatasetVariableRegistrationInput::New(registration) => {
                    let vocabulary = self
                        .register_or_validate_new_vocabulary(
                            domain_id,
                            registration.variable.vocabulary_id,
                            registration.vocabulary,
                            &registration.vocabulary_items,
                            force_metadata_updates,
                        )
                        .await?;
                    let mut variable_record = registration.variable;
                    if let Some(vocabulary) = &vocabulary {
                        variable_record.vocabulary_id = Some(vocabulary.vocabulary_id);
                    }
                    let variable = self
                        .register_or_validate_new_variable(
                            domain_id,
                            &variable_record,
                            force_metadata_updates,
                        )
                        .await?;
                    (variable, vocabulary)
                }
            };
            let vocabulary_items = match &vocabulary {
                Some(vocabulary) => self
                    .vocabularies
                    .list_vocabulary_items(vocabulary.vocabulary_id)
                    .await
                    .map_err(|error| core_error("list registered vocabulary items", error))?,
                None => Vec::new(),
            };
            let link = self
                .variables
                .link_dataset_variable(dataset_id, variable.variable_id, row_role)
                .await
                .map_err(|error| core_error("link dataset variable", error))?;
            registered.push(RegisteredDatasetVariable {
                variable,
                vocabulary,
                vocabulary_items,
                link,
            });
        }
        Ok(registered)
    }

    pub(super) async fn resolve_value_type(
        &self,
        value_type_id: Option<ahri_tre_types::ValueTypeId>,
        value_type_name: Option<&str>,
    ) -> Result<ahri_tre_types::ValueTypeRecord, AppError> {
        let value_types = self
            .variables
            .list_value_types()
            .await
            .map_err(|error| core_error("list value types", error))?;
        let by_id = value_type_id.map(|id| {
            value_types
                .iter()
                .find(|candidate| candidate.value_type_id == id)
                .cloned()
                .ok_or_else(|| AppError::Validation(format!("value_type_id {} not found", id.0)))
        });
        let by_name = value_type_name.map(|name| {
            value_types
                .iter()
                .find(|candidate| candidate.value_type.eq_ignore_ascii_case(name))
                .cloned()
                .ok_or_else(|| AppError::Validation(format!("value_type {name} not found")))
        });
        match (by_id, by_name) {
            (Some(by_id), Some(by_name)) => {
                let by_id = by_id?;
                let by_name = by_name?;
                if by_id.value_type_id != by_name.value_type_id {
                    return Err(AppError::Validation(format!(
                        "value_type_id {} is {}, not {}",
                        by_id.value_type_id.0, by_id.value_type, by_name.value_type
                    )));
                }
                Ok(by_id)
            }
            (Some(by_id), None) => by_id,
            (None, Some(by_name)) => by_name,
            (None, None) => Err(AppError::Validation(
                "variable registration requires value_type_id or value_type".to_string(),
            )),
        }
    }

    pub(super) async fn value_type_for_id(
        &self,
        value_type_id: ahri_tre_types::ValueTypeId,
    ) -> Result<ahri_tre_types::ValueTypeRecord, AppError> {
        self.resolve_value_type(Some(value_type_id), None).await
    }

    pub(super) async fn register_or_validate_vocabulary(
        &self,
        domain_id: ahri_tre_types::DomainId,
        variable_vocabulary_id: Option<ahri_tre_types::VocabularyId>,
        vocabulary: Option<ahri_tre_types::VocabularyRecord>,
        vocabulary_items: &[ahri_tre_types::VocabularyItemRecord],
        force_metadata_updates: bool,
    ) -> Result<Option<ahri_tre_types::VocabularyRecord>, AppError> {
        let Some(vocabulary_id) = vocabulary
            .as_ref()
            .map(|v| v.vocabulary_id)
            .or(variable_vocabulary_id)
        else {
            if !vocabulary_items.is_empty() {
                return Err(AppError::Validation(
                    "vocabulary items require a variable vocabulary_id or vocabulary record"
                        .to_string(),
                ));
            }
            return Ok(None);
        };

        if let Some(variable_vocabulary_id) = variable_vocabulary_id
            && variable_vocabulary_id != vocabulary_id
        {
            return Err(AppError::Validation(format!(
                "variable vocabulary_id {} does not match vocabulary record {}",
                variable_vocabulary_id.0, vocabulary_id.0
            )));
        }

        let vocabulary = match vocabulary {
            Some(vocabulary) => {
                if vocabulary.domain_id != domain_id {
                    return Err(AppError::Validation(format!(
                        "vocabulary {} belongs to domain {}, not {}",
                        vocabulary.name.as_str(),
                        vocabulary.domain_id.0,
                        domain_id.0
                    )));
                }
                let domain_vocabularies = self
                    .vocabularies
                    .list_vocabularies(domain_id)
                    .await
                    .map_err(|error| core_error("list domain vocabularies", error))?;
                if let Some(existing) = domain_vocabularies.into_iter().find(|candidate| {
                    candidate.vocabulary_id == vocabulary.vocabulary_id
                        || candidate.name == vocabulary.name
                }) {
                    if existing.name != vocabulary.name
                        && existing.vocabulary_id == vocabulary.vocabulary_id
                    {
                        return Err(AppError::Validation(format!(
                            "vocabulary {} is named {}, not {}",
                            vocabulary.vocabulary_id.0,
                            existing.name.as_str(),
                            vocabulary.name.as_str()
                        )));
                    }
                    if force_metadata_updates && existing.description != vocabulary.description {
                        let mut updated = vocabulary.clone();
                        updated.vocabulary_id = existing.vocabulary_id;
                        self.vocabularies
                            .update_vocabulary(&updated)
                            .await
                            .map_err(|error| core_error("update variable vocabulary", error))?
                    } else {
                        existing
                    }
                } else {
                    self.vocabularies
                        .save_vocabulary(&vocabulary)
                        .await
                        .map_err(|error| core_error("save variable vocabulary", error))?
                }
            }
            None => self
                .vocabularies
                .list_vocabularies(domain_id)
                .await
                .map_err(|error| core_error("list domain vocabularies", error))?
                .into_iter()
                .find(|candidate| candidate.vocabulary_id == vocabulary_id)
                .ok_or_else(|| {
                    AppError::Validation(format!(
                        "vocabulary {} not found in domain {}",
                        vocabulary_id.0, domain_id.0
                    ))
                })?,
        };

        let vocabulary_id = vocabulary.vocabulary_id;
        if vocabulary.domain_id != domain_id {
            return Err(AppError::Validation(format!(
                "vocabulary {} belongs to domain {}, not {}",
                vocabulary.vocabulary_id.0, vocabulary.domain_id.0, domain_id.0
            )));
        }
        if !vocabulary_items.is_empty() {
            let normalized_items: Vec<_> = vocabulary_items
                .iter()
                .map(|item| {
                    let mut item = item.clone();
                    item.vocabulary_id = vocabulary_id;
                    item
                })
                .collect();
            let existing_items = self
                .vocabularies
                .list_vocabulary_items(vocabulary_id)
                .await
                .map_err(|error| core_error("list vocabulary items", error))?;
            let mut seen_items: Vec<_> = existing_items
                .iter()
                .map(SeenVocabularyItem::from_record)
                .collect();
            let mut items_to_save = Vec::new();
            let mut items_to_update = Vec::new();
            for item in &normalized_items {
                let candidate = vocabulary_item_candidate(
                    Some(item.vocabulary_item_id),
                    item.value,
                    item.code.as_str(),
                    item.description.as_deref(),
                );
                match classify_vocabulary_item(&seen_items, &candidate) {
                    VocabularyItemReconciliation::ExactMatch => {}
                    VocabularyItemReconciliation::Conflict(existing_index) => {
                        let existing = &seen_items[existing_index];
                        if force_metadata_updates {
                            let mut updated = item.clone();
                            updated.vocabulary_item_id = existing
                                .vocabulary_item_id
                                .expect("persisted vocabulary item should have an id");
                            seen_items[existing_index] = SeenVocabularyItem::from_record(&updated);
                            items_to_update.push(updated);
                        } else {
                            return Err(vocabulary_item_conflict_error(
                                Some(item.vocabulary_item_id),
                                existing,
                                &candidate,
                            ));
                        }
                    }
                    VocabularyItemReconciliation::Missing => {
                        seen_items.push(SeenVocabularyItem::from_record(item));
                        items_to_save.push(item.clone());
                    }
                }
            }
            if !items_to_save.is_empty() {
                self.vocabularies
                    .save_vocabulary_items(&items_to_save)
                    .await
                    .map_err(|error| core_error("save vocabulary items", error))?;
            }
            if !items_to_update.is_empty() {
                self.vocabularies
                    .update_vocabulary_items(&items_to_update)
                    .await
                    .map_err(|error| core_error("update vocabulary items", error))?;
            }
        }
        Ok(Some(vocabulary))
    }

    pub(super) async fn register_or_validate_new_vocabulary(
        &self,
        domain_id: ahri_tre_types::DomainId,
        variable_vocabulary_id: Option<ahri_tre_types::VocabularyId>,
        vocabulary: Option<ahri_tre_types::NewVocabularyRecord>,
        vocabulary_items: &[ahri_tre_types::NewVocabularyItemRecord],
        force_metadata_updates: bool,
    ) -> Result<Option<ahri_tre_types::VocabularyRecord>, AppError> {
        let Some(vocabulary) = vocabulary else {
            let Some(vocabulary_id) = variable_vocabulary_id else {
                if !vocabulary_items.is_empty() {
                    return Err(AppError::Validation(
                        "vocabulary items require a variable vocabulary_id or vocabulary record"
                            .to_string(),
                    ));
                }
                return Ok(None);
            };
            return self
                .vocabularies
                .list_vocabularies(domain_id)
                .await
                .map_err(|error| core_error("list domain vocabularies", error))?
                .into_iter()
                .find(|candidate| candidate.vocabulary_id == vocabulary_id)
                .ok_or_else(|| {
                    AppError::Validation(format!(
                        "vocabulary {} not found in domain {}",
                        vocabulary_id.0, domain_id.0
                    ))
                })
                .map(Some);
        };

        if vocabulary.domain_id != domain_id {
            return Err(AppError::Validation(format!(
                "vocabulary {} belongs to domain {}, not {}",
                vocabulary.name.as_str(),
                vocabulary.domain_id.0,
                domain_id.0
            )));
        }
        let domain_vocabularies = self
            .vocabularies
            .list_vocabularies(domain_id)
            .await
            .map_err(|error| core_error("list domain vocabularies", error))?;
        let vocabulary = if let Some(existing) = domain_vocabularies
            .into_iter()
            .find(|candidate| candidate.name == vocabulary.name)
        {
            if force_metadata_updates && existing.description != vocabulary.description {
                let updated = ahri_tre_types::VocabularyRecord {
                    vocabulary_id: existing.vocabulary_id,
                    domain_id: vocabulary.domain_id,
                    name: vocabulary.name.clone(),
                    description: vocabulary.description.clone(),
                };
                self.vocabularies
                    .update_vocabulary(&updated)
                    .await
                    .map_err(|error| core_error("update variable vocabulary", error))?
            } else {
                existing
            }
        } else {
            self.vocabularies
                .create_vocabulary(&vocabulary)
                .await
                .map_err(|error| core_error("create variable vocabulary", error))?
        };

        if let Some(variable_vocabulary_id) = variable_vocabulary_id
            && variable_vocabulary_id != vocabulary.vocabulary_id
        {
            return Err(AppError::Validation(format!(
                "variable vocabulary_id {} does not match vocabulary record {}",
                variable_vocabulary_id.0, vocabulary.vocabulary_id.0
            )));
        }

        self.register_new_vocabulary_items(
            vocabulary.vocabulary_id,
            vocabulary_items,
            force_metadata_updates,
        )
        .await?;
        Ok(Some(vocabulary))
    }

    pub(super) async fn register_new_vocabulary_items(
        &self,
        vocabulary_id: ahri_tre_types::VocabularyId,
        vocabulary_items: &[ahri_tre_types::NewVocabularyItemRecord],
        force_metadata_updates: bool,
    ) -> Result<(), AppError> {
        if vocabulary_items.is_empty() {
            return Ok(());
        }
        let existing_items = self
            .vocabularies
            .list_vocabulary_items(vocabulary_id)
            .await
            .map_err(|error| core_error("list vocabulary items", error))?;
        let mut seen_items: Vec<_> = existing_items
            .iter()
            .map(SeenVocabularyItem::from_record)
            .collect();
        let mut items_to_create = Vec::new();
        let mut items_to_update = Vec::new();
        for item in vocabulary_items {
            let candidate = vocabulary_item_candidate(
                None,
                item.value,
                item.code.as_str(),
                item.description.as_deref(),
            );
            match classify_vocabulary_item(&seen_items, &candidate) {
                VocabularyItemReconciliation::ExactMatch => {}
                VocabularyItemReconciliation::Conflict(existing_index) => {
                    let existing = &seen_items[existing_index];
                    if force_metadata_updates {
                        if let Some(vocabulary_item_id) = existing.vocabulary_item_id {
                            let updated = build_vocabulary_item_record(
                                vocabulary_item_id,
                                vocabulary_id,
                                item.value,
                                item.code.clone(),
                                item.description.clone(),
                            );
                            seen_items[existing_index] = SeenVocabularyItem::from_record(&updated);
                            items_to_update.push(updated);
                        } else if let Some(pending_index) = existing.pending_create_index {
                            items_to_create[pending_index] = item.clone();
                            seen_items[existing_index] =
                                SeenVocabularyItem::from_pending_create(pending_index, item);
                        }
                    } else {
                        return Err(vocabulary_item_conflict_error(
                            existing.vocabulary_item_id,
                            existing,
                            &candidate,
                        ));
                    }
                }
                VocabularyItemReconciliation::Missing => {
                    items_to_create.push(item.clone());
                    let pending_index = items_to_create.len() - 1;
                    seen_items.push(SeenVocabularyItem::from_pending_create(pending_index, item));
                }
            }
        }
        if !items_to_create.is_empty() {
            self.vocabularies
                .create_vocabulary_items(vocabulary_id, &items_to_create)
                .await
                .map_err(|error| core_error("create vocabulary items", error))?;
        }
        if !items_to_update.is_empty() {
            self.vocabularies
                .update_vocabulary_items(&items_to_update)
                .await
                .map_err(|error| core_error("update vocabulary items", error))?;
        }
        Ok(())
    }

    pub(super) async fn register_or_validate_variable(
        &self,
        domain_id: ahri_tre_types::DomainId,
        variable: &ahri_tre_types::VariableRecord,
        force_metadata_updates: bool,
    ) -> Result<ahri_tre_types::VariableRecord, AppError> {
        let variables = self
            .variables
            .list_domain_variables(domain_id)
            .await
            .map_err(|error| core_error("list domain variables", error))?;
        if let Some(existing) = variables
            .iter()
            .find(|candidate| candidate.variable_id == variable.variable_id)
        {
            if existing.name != variable.name {
                return Err(AppError::Validation(format!(
                    "variable {} is named {}, not {}",
                    variable.variable_id.0,
                    existing.name.as_str(),
                    variable.name.as_str()
                )));
            }
            if existing.vocabulary_id != variable.vocabulary_id && !force_metadata_updates {
                return Err(AppError::Validation(format!(
                    "variable {} vocabulary_id does not match existing variable; set force_metadata_updates to update metadata",
                    variable.variable_id.0
                )));
            }
            return self
                .merge_or_update_variable(existing, variable, force_metadata_updates)
                .await;
        }
        if let Some(existing) = variables
            .iter()
            .find(|candidate| candidate.name == variable.name)
        {
            if existing.value_type_id != variable.value_type_id && !force_metadata_updates {
                return Err(AppError::Validation(format!(
                    "variable {} already exists in domain {} with incompatible value type; set force_metadata_updates to update metadata",
                    variable.name.as_str(),
                    domain_id.0
                )));
            }
            return self
                .merge_or_update_variable(existing, variable, force_metadata_updates)
                .await;
        }
        self.variables
            .save_variable(variable)
            .await
            .map_err(|error| core_error("save dataset variable", error))
    }

    pub(super) async fn register_or_validate_new_variable(
        &self,
        domain_id: ahri_tre_types::DomainId,
        variable: &ahri_tre_types::NewVariableRecord,
        force_metadata_updates: bool,
    ) -> Result<ahri_tre_types::VariableRecord, AppError> {
        let variables = self
            .variables
            .list_domain_variables(domain_id)
            .await
            .map_err(|error| core_error("list domain variables", error))?;
        if let Some(existing) = variables
            .iter()
            .find(|candidate| candidate.name == variable.name)
        {
            let incoming = ahri_tre_types::VariableRecord {
                variable_id: existing.variable_id,
                domain_id: variable.domain_id,
                name: variable.name.clone(),
                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(),
            };
            if existing.value_type_id != incoming.value_type_id && !force_metadata_updates {
                return Err(AppError::Validation(format!(
                    "variable {} already exists in domain {} with incompatible value type; set force_metadata_updates to update metadata",
                    variable.name.as_str(),
                    domain_id.0
                )));
            }
            return self
                .merge_or_update_variable(existing, &incoming, force_metadata_updates)
                .await;
        }
        self.variables
            .create_variable(variable)
            .await
            .map_err(|error| core_error("create dataset variable", error))
    }

    pub(super) async fn merge_or_update_variable(
        &self,
        existing: &ahri_tre_types::VariableRecord,
        incoming: &ahri_tre_types::VariableRecord,
        force_metadata_updates: bool,
    ) -> Result<ahri_tre_types::VariableRecord, AppError> {
        let mut merged = incoming.clone();
        merged.variable_id = existing.variable_id;
        let value_type_changed = existing.value_type_id != incoming.value_type_id;
        let value_format_changed = existing.value_format != incoming.value_format;
        let storage_affecting_change = value_format_changed
            || self
                .storage_affecting_value_type_change(existing.value_type_id, incoming.value_type_id)
                .await?;
        if value_type_changed || value_format_changed {
            if !force_metadata_updates {
                return Err(AppError::Validation(format!(
                    "variable {} value type or format differs; set force_metadata_updates to update metadata",
                    existing.name.as_str()
                )));
            }
            if storage_affecting_change {
                let links = self
                    .variables
                    .list_dataset_variables_for_variable(existing.variable_id)
                    .await
                    .map_err(|error| core_error("list variable dataset links", error))?;
                if !links.is_empty() {
                    return Err(AppError::Validation(format!(
                        "variable {} is linked to {} dataset(s); changing value type or format requires a storage-aware migration workflow",
                        existing.name.as_str(),
                        links.len()
                    )));
                }
            }
        }
        if !force_metadata_updates {
            if existing.vocabulary_id != incoming.vocabulary_id {
                return Err(AppError::Validation(format!(
                    "variable {} vocabulary differs; set force_metadata_updates to update metadata",
                    existing.name.as_str()
                )));
            }
            return Ok(existing.clone());
        }
        self.variables
            .update_variable_metadata(&merged)
            .await
            .map_err(|error| core_error("update dataset variable", error))
    }

    pub(super) async fn storage_affecting_value_type_change(
        &self,
        existing: ahri_tre_types::ValueTypeId,
        incoming: ahri_tre_types::ValueTypeId,
    ) -> Result<bool, AppError> {
        if existing == incoming {
            return Ok(false);
        }
        let existing = self.value_type_for_id(existing).await?;
        let incoming = self.value_type_for_id(incoming).await?;
        Ok(!integer_to_enumeration_value_type_names(
            &existing.value_type,
            &incoming.value_type,
        ))
    }

    pub(super) async fn attach_tags(
        &self,
        target: TagAttachmentTarget,
        tags: Vec<String>,
    ) -> Result<(), AppError> {
        for name in tags {
            let tag = self
                .tags
                .create_tag(&ahri_tre_types::NewTagRecord { name })
                .await
                .map_err(|error| core_error("create tag", error))?;
            self.tags
                .attach_tag(tag.tag_id, target.clone())
                .await
                .map_err(|error| core_error("attach tag", error))?;
        }
        Ok(())
    }
}

impl ScopedAppService<'_> {
    pub async fn register_vocabulary(
        &self,
        request: RegisterVocabularyRequest,
    ) -> Result<RegisteredVocabulary, AppError> {
        self.domains
            .get_domain_by_id(request.domain_id)
            .await
            .map_err(|error| core_error("lookup vocabulary domain", error))?
            .ok_or_else(|| {
                AppError::Validation(format!(
                    "vocabulary domain not found: {}",
                    request.domain_id.0
                ))
            })?;

        let domain_vocabularies = self
            .vocabularies
            .list_vocabularies(request.domain_id)
            .await
            .map_err(|error| core_error("list domain vocabularies", error))?;
        let existing = domain_vocabularies.into_iter().find(|candidate| {
            request
                .vocabulary_id
                .is_some_and(|id| candidate.vocabulary_id == id)
                || candidate.name == request.name
        });
        let vocabulary = match (existing, request.vocabulary_id) {
            (Some(existing), _) => {
                if existing.name != request.name {
                    return Err(AppError::Validation(format!(
                        "vocabulary {} is named {}, not {}",
                        existing.vocabulary_id.0,
                        existing.name.as_str(),
                        request.name.as_str()
                    )));
                }
                if request.force_metadata_updates && existing.description != request.description {
                    self.vocabularies
                        .update_vocabulary(&ahri_tre_types::VocabularyRecord {
                            vocabulary_id: existing.vocabulary_id,
                            domain_id: request.domain_id,
                            name: request.name,
                            description: request.description,
                        })
                        .await
                        .map_err(|error| core_error("update vocabulary", error))?
                } else {
                    existing
                }
            }
            (None, Some(vocabulary_id)) => self
                .vocabularies
                .save_vocabulary(&ahri_tre_types::VocabularyRecord {
                    vocabulary_id,
                    domain_id: request.domain_id,
                    name: request.name,
                    description: request.description,
                })
                .await
                .map_err(|error| core_error("save vocabulary", error))?,
            (None, None) => self
                .vocabularies
                .create_vocabulary(&ahri_tre_types::NewVocabularyRecord {
                    domain_id: request.domain_id,
                    name: request.name,
                    description: request.description,
                })
                .await
                .map_err(|error| core_error("create vocabulary", error))?,
        };
        self.register_vocabulary_item_requests(
            vocabulary.vocabulary_id,
            request.items,
            request.force_metadata_updates,
        )
        .await?;
        let items = self
            .vocabularies
            .list_vocabulary_items(vocabulary.vocabulary_id)
            .await
            .map_err(|error| core_error("list registered vocabulary items", error))?;
        Ok(RegisteredVocabulary { vocabulary, items })
    }

    async fn register_vocabulary_item_requests(
        &self,
        vocabulary_id: ahri_tre_types::VocabularyId,
        vocabulary_items: Vec<RegisterVocabularyItemRequest>,
        force_metadata_updates: bool,
    ) -> Result<(), AppError> {
        if vocabulary_items.is_empty() {
            return Ok(());
        }
        let existing_items = self
            .vocabularies
            .list_vocabulary_items(vocabulary_id)
            .await
            .map_err(|error| core_error("list vocabulary items", error))?;
        let mut seen_items: Vec<_> = existing_items
            .iter()
            .map(SeenVocabularyItem::from_record)
            .collect();
        let mut items_to_create = Vec::new();
        let mut items_to_save = Vec::new();
        let mut items_to_update = Vec::new();
        for item in vocabulary_items {
            let candidate = vocabulary_item_candidate(
                item.vocabulary_item_id,
                item.value,
                item.code.as_str(),
                item.description.as_deref(),
            );
            match classify_vocabulary_item(&seen_items, &candidate) {
                VocabularyItemReconciliation::ExactMatch => {}
                VocabularyItemReconciliation::Conflict(existing_index) => {
                    let existing = &seen_items[existing_index];
                    if force_metadata_updates {
                        if let Some(vocabulary_item_id) = existing.vocabulary_item_id {
                            let updated = build_vocabulary_item_record(
                                vocabulary_item_id,
                                vocabulary_id,
                                item.value,
                                item.code,
                                item.description,
                            );
                            seen_items[existing_index] = SeenVocabularyItem::from_record(&updated);
                            items_to_update.push(updated);
                        } else if let Some(pending_index) = existing.pending_create_index {
                            let created = ahri_tre_types::NewVocabularyItemRecord {
                                value: item.value,
                                code: item.code,
                                description: item.description,
                            };
                            items_to_create[pending_index] = created;
                            seen_items[existing_index] = SeenVocabularyItem::from_pending_create(
                                pending_index,
                                &items_to_create[pending_index],
                            );
                        }
                    } else {
                        return Err(vocabulary_item_conflict_error(
                            existing.vocabulary_item_id,
                            existing,
                            &candidate,
                        ));
                    }
                }
                VocabularyItemReconciliation::Missing => {
                    if let Some(vocabulary_item_id) = item.vocabulary_item_id {
                        let saved = build_vocabulary_item_record(
                            vocabulary_item_id,
                            vocabulary_id,
                            item.value,
                            item.code,
                            item.description,
                        );
                        seen_items.push(SeenVocabularyItem::from_record(&saved));
                        items_to_save.push(saved);
                    } else {
                        let created = ahri_tre_types::NewVocabularyItemRecord {
                            value: item.value,
                            code: item.code,
                            description: item.description,
                        };
                        items_to_create.push(created);
                        let pending_index = items_to_create.len() - 1;
                        seen_items.push(SeenVocabularyItem::from_pending_create(
                            pending_index,
                            &items_to_create[pending_index],
                        ));
                    }
                }
            }
        }
        if !items_to_create.is_empty() {
            self.vocabularies
                .create_vocabulary_items(vocabulary_id, &items_to_create)
                .await
                .map_err(|error| core_error("create vocabulary items", error))?;
        }
        if !items_to_save.is_empty() {
            self.vocabularies
                .save_vocabulary_items(&items_to_save)
                .await
                .map_err(|error| core_error("save vocabulary items", error))?;
        }
        if !items_to_update.is_empty() {
            self.vocabularies
                .update_vocabulary_items(&items_to_update)
                .await
                .map_err(|error| core_error("update vocabulary items", error))?;
        }
        Ok(())
    }
}