ahri_tre_app/service/

semantic.rs

//! Authenticated semantic catalogue workflows. Public references are decoded at
//! the retained Session boundary; repositories remain the visibility authority.
use super::*;
use ahri_tre_protocol::{
    dictionary, model,
    refs::{ObjectKind, ObjectRef},
};

impl AppService {
    pub(crate) fn semantic_integer_id(
        &self,
        reference: ObjectRef,
        kind: ObjectKind,
        scope: &str,
    ) -> Result<i64, AppError> {
        let id = self.catalogue_reference_id(reference, kind)?;
        ahri_tre_protocol::refs::decode_scoped_integer_ref(scope, id)
            .filter(|id| *id > 0)
            .ok_or_else(|| AppError::Validation("Invalid semantic reference".into()))
    }

    pub(super) async fn catalogue_variable_selector(
        &self,
        selector: dictionary::VariableSelector,
    ) -> Result<VariableSelector, AppError> {
        let (selector, constraint) = match selector {
            dictionary::VariableSelector::Id { variable, domain } => (
                VariableSelector::Id {
                    variable_id: ahri_tre_types::VariableId(self.semantic_integer_id(
                        variable,
                        ObjectKind::Variable,
                        "variable",
                    )?),
                },
                domain,
            ),
            dictionary::VariableSelector::Name { domain, name } => (
                VariableSelector::Name {
                    domain: self.catalogue_domain_selector(domain)?,
                    name: name.into(),
                },
                None,
            ),
        };
        if let Some(domain) = constraint {
            let domain = self
                .resolve_domain_selector(
                    self.catalogue_domain_selector(domain)?,
                    "semantic Domain constraint",
                )
                .await?;
            let target = self
                .resolve_variable_selector_optional(selector.clone())
                .await?
                .ok_or_else(|| {
                    AppError::NotFound("Semantic definition was not found or is not visible".into())
                })?;
            if target.domain_id != domain.domain_id {
                return Err(AppError::Conflict(
                    "Semantic definition and Domain selectors disagree".into(),
                ));
            }
        }
        Ok(selector)
    }

    pub(super) async fn catalogue_vocabulary_selector(
        &self,
        selector: dictionary::VocabularySelector,
    ) -> Result<VocabularySelector, AppError> {
        let (selector, constraint) = match selector {
            dictionary::VocabularySelector::Id { vocabulary, domain } => (
                VocabularySelector::Id {
                    vocabulary_id: ahri_tre_types::VocabularyId(self.semantic_integer_id(
                        vocabulary,
                        ObjectKind::Vocabulary,
                        "vocabulary",
                    )?),
                },
                domain,
            ),
            dictionary::VocabularySelector::Name { domain, name } => (
                VocabularySelector::Name {
                    domain: self.catalogue_domain_selector(domain)?,
                    name: name.into(),
                },
                None,
            ),
        };
        if let Some(domain) = constraint {
            let domain = self
                .resolve_domain_selector(
                    self.catalogue_domain_selector(domain)?,
                    "semantic Domain constraint",
                )
                .await?;
            let target = self
                .resolve_vocabulary_selector_optional(selector.clone())
                .await?
                .ok_or_else(|| {
                    AppError::NotFound("Semantic definition was not found or is not visible".into())
                })?;
            if target.domain_id != domain.domain_id {
                return Err(AppError::Conflict(
                    "Semantic definition and Domain selectors disagree".into(),
                ));
            }
        }
        Ok(selector)
    }

    pub(super) async fn catalogue_entity_selector(
        &self,
        selector: model::EntitySelector,
    ) -> Result<EntitySelector, AppError> {
        let (selector, constraint) = match selector {
            model::EntitySelector::Id { entity, domain } => (
                EntitySelector::Id {
                    entity_id: ahri_tre_types::EntityId(self.semantic_integer_id(
                        entity,
                        ObjectKind::Entity,
                        "entity",
                    )?),
                },
                domain,
            ),
            model::EntitySelector::Name { domain, name } => (
                EntitySelector::Name {
                    domain: self.catalogue_domain_selector(domain)?,
                    name: name.into(),
                },
                None,
            ),
        };
        if let Some(domain) = constraint {
            let domain = self
                .resolve_domain_selector(
                    self.catalogue_domain_selector(domain)?,
                    "semantic Domain constraint",
                )
                .await?;
            let target = self
                .resolve_entity_selector_optional(selector.clone())
                .await?
                .ok_or_else(|| {
                    AppError::NotFound("Semantic definition was not found or is not visible".into())
                })?;
            if target.domain_id != domain.domain_id {
                return Err(AppError::Conflict(
                    "Semantic definition and Domain selectors disagree".into(),
                ));
            }
        }
        Ok(selector)
    }

    pub(super) async fn catalogue_relation_selector(
        &self,
        selector: model::RelationSelector,
    ) -> Result<EntityRelationSelector, AppError> {
        let (selector, constraint) = match selector {
            model::RelationSelector::Id { relation, domain } => (
                EntityRelationSelector::Id {
                    entity_relation_id: ahri_tre_types::EntityRelationId(
                        self.semantic_integer_id(relation, ObjectKind::Relation, "relation")?,
                    ),
                },
                domain,
            ),
            model::RelationSelector::Name { domain, name } => (
                EntityRelationSelector::Name {
                    domain: self.catalogue_domain_selector(domain)?,
                    name: name.into(),
                },
                None,
            ),
        };
        if let Some(domain) = constraint {
            let domain = self
                .resolve_domain_selector(
                    self.catalogue_domain_selector(domain)?,
                    "semantic Domain constraint",
                )
                .await?;
            let target = self
                .resolve_entity_relation_selector_optional(selector.clone())
                .await?
                .ok_or_else(|| {
                    AppError::NotFound("Semantic definition was not found or is not visible".into())
                })?;
            if target.domain_id != domain.domain_id {
                return Err(AppError::Conflict(
                    "Semantic definition and Domain selectors disagree".into(),
                ));
            }
        }
        Ok(selector)
    }

    pub async fn list_catalogue_variables(
        &self,
        request: dictionary::ListVariablesRequest,
    ) -> Result<dictionary::VariableListResponse, AppError> {
        let values = self
            .list_variables(ListVariablesRequest {
                domain: request
                    .domain
                    .map(|v| self.catalogue_domain_selector(v))
                    .transpose()?,
                study: request
                    .study
                    .map(|v| self.catalogue_study_selector(v))
                    .transpose()?,
            })
            .await?;
        let mut variables = Vec::with_capacity(values.len());
        for value in values {
            variables.push(self.variable_response(value).await?.variable.summary);
        }
        Ok(dictionary::VariableListResponse {
            variables,
            warnings: None,
        })
    }

    pub async fn get_catalogue_variable(
        &self,
        request: dictionary::GetVariableRequest,
    ) -> Result<dictionary::VariableDetailResponse, AppError> {
        let variable = self
            .get_variable(GetVariableRequest {
                variable: self.catalogue_variable_selector(request.variable).await?,
            })
            .await?
            .ok_or_else(|| AppError::NotFound("Variable was not found or is not visible".into()))?;
        self.variable_response(variable).await
    }

    pub async fn add_catalogue_variable(
        &self,
        request: dictionary::AddVariableRequest,
    ) -> Result<dictionary::VariableDetailResponse, AppError> {
        let variable = self
            .add_variable(AddVariableRequest {
                domain: self.catalogue_domain_selector(request.domain)?,
                name: request.name.into(),
                value_type: request.value_type,
                value_format: request.value_format,
                key_role: request.key_role,
                description: request.description,
                ontology_namespace: request.ontology_namespace,
                ontology_class: request.ontology_class,
                vocabulary: request.vocabulary.map(Into::into),
                vocabulary_items: request.vocabulary_items,
            })
            .await?;
        self.variable_response(variable).await
    }

    pub async fn update_catalogue_variable(
        &self,
        request: dictionary::UpdateVariableRequest,
    ) -> Result<dictionary::VariableDetailResponse, AppError> {
        let selector = self.catalogue_variable_selector(request.variable).await?;
        let existing = self
            .resolve_variable_selector_optional(selector)
            .await?
            .ok_or_else(|| AppError::NotFound("Variable was not found or is not visible".into()))?;
        let variable = self
            .update_variable(UpdateVariableRequest {
                variable: VariableSelector::Id {
                    variable_id: existing.variable_id,
                },
                value_type: request.value_type,
                value_format: request.value_format,
                key_role: request.key_role,
                description: request.description,
                ontology_namespace: request.ontology_namespace,
                ontology_class: request.ontology_class,
                vocabulary: request.vocabulary.map(Into::into),
            })
            .await?;
        self.variable_response(variable).await
    }

    async fn variable_response(
        &self,
        variable: RegisteredVariable,
    ) -> Result<dictionary::VariableDetailResponse, AppError> {
        let tags = self
            .semantic_target_tags(TagAttachmentTarget::Variable(variable.variable.variable_id))
            .await?;
        let items = match variable.vocabulary.as_ref() {
            Some(vocabulary) => self.declared_vocabulary_items(vocabulary.vocabulary_id),
            None => Some(Vec::new()),
        };
        let mut variable =
            crate::projections::semantic_variable(self.catalogue_datastore_id()?, variable);
        if let Some(items) = items {
            variable.vocabulary_items = items.into_iter().map(schema_item).collect();
            variable.vocabulary_items_withheld = false;
            if let Some(vocabulary) = variable.summary.vocabulary.as_mut() {
                vocabulary.item_count = Some(variable.vocabulary_items.len());
            }
        }
        variable.summary.tags = tags;
        Ok(dictionary::VariableDetailResponse {
            variable,
            warnings: None,
        })
    }

    async fn semantic_target_tags(
        &self,
        target: TagAttachmentTarget,
    ) -> Result<Vec<ahri_tre_protocol::refs::TagSummary>, AppError> {
        let datastore_id = self.catalogue_datastore_id()?;
        Ok(self
            .get_tags(GetTagsRequest { target })
            .await?
            .tags
            .into_iter()
            .map(|tag| crate::projections::tag_summary(datastore_id, tag))
            .collect())
    }

    pub async fn list_catalogue_vocabularies(
        &self,
        request: dictionary::ListVocabulariesRequest,
    ) -> Result<dictionary::VocabularyListResponse, AppError> {
        let values = self
            .list_vocabularies(ListVocabulariesRequest {
                domain: request
                    .domain
                    .map(|v| self.catalogue_domain_selector(v))
                    .transpose()?,
                study: request
                    .study
                    .map(|v| self.catalogue_study_selector(v))
                    .transpose()?,
            })
            .await?;
        Ok(dictionary::VocabularyListResponse {
            vocabularies: values
                .into_iter()
                .map(|v| self.vocabulary_response(v).map(|v| v.vocabulary.summary))
                .collect::<Result<_, _>>()?,
            warnings: None,
        })
    }

    pub async fn get_catalogue_vocabulary(
        &self,
        request: dictionary::GetVocabularyRequest,
    ) -> Result<dictionary::VocabularyDetailResponse, AppError> {
        let vocabulary = self
            .get_vocabulary(GetVocabularyRequest {
                vocabulary: self
                    .catalogue_vocabulary_selector(request.vocabulary)
                    .await?,
            })
            .await?
            .ok_or_else(|| {
                AppError::NotFound("Vocabulary was not found or is not visible".into())
            })?;
        self.vocabulary_response(vocabulary)
    }

    pub async fn add_catalogue_vocabulary(
        &self,
        request: dictionary::AddVocabularyRequest,
    ) -> Result<dictionary::VocabularyDetailResponse, AppError> {
        let domain = self
            .resolve_domain_selector(
                self.catalogue_domain_selector(request.domain)?,
                "vocabulary Domain",
            )
            .await?;
        let declaration = RegisterVocabularyRequest {
            domain_id: domain.domain_id,
            vocabulary_id: None,
            name: parse_vocabulary_name(request.name.into())?,
            description: request.description,
            items: parse_vocabulary_item_requests(&request.items)?,
            force_metadata_updates: false,
        };
        let repository = self.semantic_repository.as_ref().ok_or_else(|| {
            AppError::Infrastructure("Dictionary provenance is unavailable".into())
        })?;
        let vocabulary = repository
            .with_semantic_declarations(|repository| {
                let repositories =
                    crate::session::ScopedSessionMetadataRepositories::from_repository(Arc::new(
                        repository,
                    ));
                let service = ScopedAppService::new(repositories.into());
                futures::FutureExt::now_or_never(service.register_vocabulary(declaration))
                    .ok_or_else(|| {
                        CoreError::Infrastructure(
                            "Dictionary declaration unexpectedly suspended".into(),
                        )
                    })?
                    .map_err(semantic_mutation_error)
            })
            .map_err(|error| core_error("declare vocabulary", error))?;
        self.vocabulary_response(vocabulary)
    }

    fn vocabulary_response(
        &self,
        vocabulary: RegisteredVocabulary,
    ) -> Result<dictionary::VocabularyDetailResponse, AppError> {
        let items = self.declared_vocabulary_items(vocabulary.vocabulary.vocabulary_id);
        let mut vocabulary =
            crate::projections::semantic_vocabulary(self.catalogue_datastore_id()?, vocabulary);
        if let Some(items) = items {
            vocabulary.items = items.into_iter().map(schema_item).collect();
            vocabulary.summary.item_count = Some(vocabulary.items.len());
            vocabulary.items_withheld = false;
        }
        Ok(dictionary::VocabularyDetailResponse {
            vocabulary,
            warnings: None,
        })
    }

    pub(super) fn declared_vocabulary_items(
        &self,
        id: ahri_tre_types::VocabularyId,
    ) -> Option<Vec<ahri_tre_types::VocabularyItemRecord>> {
        // A missing proof, exhausted schema bound, or concurrent derivation
        // leaves identity available with an explicit withheld marker.
        self.semantic_repository
            .as_ref()?
            .catalogue_vocabulary_items(id)
            .ok()
            .flatten()
    }

    pub async fn list_catalogue_entities(
        &self,
        request: model::ListEntitiesRequest,
    ) -> Result<model::EntityListResponse, AppError> {
        let records = self
            .list_entities(ListEntitiesRequest {
                domain: self.catalogue_domain_selector(request.domain)?,
            })
            .await?;
        let mut values = Vec::with_capacity(records.len());
        for record in records {
            values.push(self.entity_response(record).await?.entity.summary);
        }
        Ok(model::EntityListResponse {
            entities: values,
            next_cursor: None,
            warnings: None,
        })
    }
    pub async fn get_catalogue_entity(
        &self,
        request: model::GetEntityRequest,
    ) -> Result<model::EntityDetailResponse, AppError> {
        let record = self
            .get_entity(GetEntityRequest {
                entity: self.catalogue_entity_selector(request.entity).await?,
            })
            .await?
            .ok_or_else(|| AppError::NotFound("Entity was not found or is not visible".into()))?;
        self.entity_response(record).await
    }
    pub async fn add_catalogue_entity(
        &self,
        request: model::AddEntityRequest,
    ) -> Result<model::EntityDetailResponse, AppError> {
        let record = self
            .add_entity(AddEntityRequest {
                domain: self.catalogue_domain_selector(request.domain)?,
                name: request.name.into(),
                description: request.description,
                agent_instructions: request.agent_instructions,
                ontology_namespace: request.ontology_namespace,
                ontology_class: request.ontology_class,
            })
            .await?;
        self.entity_response(record).await
    }
    async fn entity_response(
        &self,
        record: ahri_tre_types::EntityRecord,
    ) -> Result<model::EntityDetailResponse, AppError> {
        let tags = self
            .semantic_target_tags(TagAttachmentTarget::Entity(record.entity_id))
            .await?;
        let mut entity =
            crate::projections::semantic_entity(self.catalogue_datastore_id()?, record);
        entity.summary.tags = tags;
        Ok(model::EntityDetailResponse {
            entity,
            warnings: None,
        })
    }

    pub async fn list_catalogue_relations(
        &self,
        request: model::ListRelationsRequest,
    ) -> Result<model::RelationListResponse, AppError> {
        let records = self
            .list_entity_relations(ListEntityRelationsRequest {
                domain: self.catalogue_domain_selector(request.domain)?,
            })
            .await?;
        let mut values = Vec::with_capacity(records.len());
        for record in records {
            values.push(self.relation_response(record).await?.relation.summary);
        }
        Ok(model::RelationListResponse {
            relations: values,
            next_cursor: None,
            warnings: None,
        })
    }
    pub async fn get_catalogue_relation(
        &self,
        request: model::GetRelationRequest,
    ) -> Result<model::RelationDetailResponse, AppError> {
        let record = self
            .get_entity_relation(GetEntityRelationRequest {
                relation: self.catalogue_relation_selector(request.relation).await?,
            })
            .await?
            .ok_or_else(|| AppError::NotFound("Relation was not found or is not visible".into()))?;
        self.relation_response(record).await
    }
    pub async fn add_catalogue_relation(
        &self,
        request: model::AddRelationRequest,
    ) -> Result<model::RelationDetailResponse, AppError> {
        let record = self
            .add_entity_relation(AddEntityRelationRequest {
                domain: self.catalogue_domain_selector(request.domain)?,
                name: request.name.into(),
                description: request.description,
                agent_instructions: request.agent_instructions,
                ontology_namespace: request.ontology_namespace,
                ontology_class: request.ontology_class,
                subject: request.subject.into(),
                object: request.object.into(),
            })
            .await?;
        self.relation_response(record).await
    }
    async fn relation_response(
        &self,
        record: ahri_tre_types::EntityRelationRecord,
    ) -> Result<model::RelationDetailResponse, AppError> {
        let mut names = BTreeMap::new();
        for id in [record.subject_entity_id, record.object_entity_id] {
            if let Some(entity) = self
                .get_entity(GetEntityRequest {
                    entity: EntitySelector::Id { entity_id: id },
                })
                .await?
            {
                names.insert(id.0, entity.name.as_str().to_string());
            }
        }
        let tags = self
            .semantic_target_tags(TagAttachmentTarget::EntityRelation(
                record.entity_relation_id,
            ))
            .await?;
        let mut relation =
            crate::projections::semantic_relation(self.catalogue_datastore_id()?, record, &names);
        relation.summary.tags = tags;
        Ok(model::RelationDetailResponse {
            relation,
            warnings: None,
        })
    }

    pub async fn list_catalogue_transformations(
        &self,
        request: ahri_tre_protocol::transformation::TransformationListRequest,
    ) -> Result<Vec<ahri_tre_protocol::transformation::TransformationSummary>, AppError> {
        let study = self
            .resolve_study_selector(
                self.catalogue_study_selector(request.study)?,
                "Transformation Study",
            )
            .await?;
        let mut values = self.study_transformations(study.study_id).await?;
        // A Transformation can span Studies. Each nested version must itself
        // be visible; visibility of one output cannot authorize other inputs.
        let mut visible = BTreeMap::new();
        for value in &values {
            for version_id in value
                .inputs
                .iter()
                .map(|link| link.version_id)
                .chain(value.outputs.iter().map(|link| link.version_id))
            {
                if visible.contains_key(&version_id.0) {
                    continue;
                }
                let version = self
                    .assets
                    .get_asset_version(version_id)
                    .await
                    .map_err(|error| core_error("resolve provenance version", error))?;
                let allowed = match version {
                    Some(version) => self
                        .assets
                        .get_asset_by_id(version.asset_id)
                        .await
                        .map_err(|error| core_error("authorize provenance Asset", error))?
                        .is_some(),
                    None => false,
                };
                visible.insert(version_id.0, allowed);
            }
        }
        for value in &mut values {
            value
                .inputs
                .retain(|link| visible.get(&link.version_id.0) == Some(&true));
            value
                .outputs
                .retain(|link| visible.get(&link.version_id.0) == Some(&true));
        }
        let datastore_id = self.catalogue_datastore_id()?;
        Ok(values
            .into_iter()
            .map(|value| crate::projections::semantic_transformation(datastore_id, value))
            .collect())
    }

    pub(super) async fn semantic_domains(
        &self,
        domain: Option<DomainSelector>,
        study: Option<StudySelector>,
        label: &str,
    ) -> Result<Vec<DomainRecord>, AppError> {
        match (domain, study) {
            (Some(domain), study) => {
                let domain = self.resolve_domain_selector(domain, label).await?;
                if let Some(study) = study {
                    let study = self.resolve_study_selector(study, label).await?;
                    let domains = self
                        .study_domains
                        .list_domains_for_study(study.study_id)
                        .await
                        .map_err(|e| core_error("list semantic Study domains", e))?;
                    if !domains.iter().any(|d| d.domain_id == domain.domain_id) {
                        return Err(AppError::Conflict(
                            "Domain does not belong to the selected Study".into(),
                        ));
                    }
                }
                Ok(vec![domain])
            }
            (None, Some(study)) => {
                let study = self.resolve_study_selector(study, label).await?;
                self.study_domains
                    .list_domains_for_study(study.study_id)
                    .await
                    .map_err(|e| core_error("list semantic Study domains", e))
            }
            (None, None) => Err(AppError::Validation(
                "Semantic listing requires a Domain or Study".into(),
            )),
        }
    }
}

fn schema_item(item: ahri_tre_types::VocabularyItemRecord) -> dictionary::VocabularyItemSummary {
    dictionary::VocabularyItemSummary {
        value: item.value,
        code: item.code,
        description: item.description,
    }
}

fn semantic_mutation_error(error: AppError) -> CoreError {
    match error {
        AppError::Validation(message) => CoreError::Validation(message),
        AppError::NotFound(message) => CoreError::NotFound(message),
        AppError::Conflict(message) => CoreError::Conflict(message),
        AppError::NotImplemented(message) => CoreError::NotImplemented(message),
        _ => CoreError::Infrastructure("Semantic mutation is unavailable".into()),
    }
}