ahri_tre_pgmeta/

semantic.rs

//! Durable semantic field origins. Missing provenance is never reconstructed
//! from a shared Domain, a current Dataset link, or a Transformation label.
use crate::{MetadataExecutor, PgMetaError, PgMetadataRepository, schema::SchemaStatusRow};
use ahri_tre_core::CoreError;
use ahri_tre_core::VocabularyRepository;
use ahri_tre_types::VocabularyId;
use futures::FutureExt;
use serde::{Deserialize, Serialize};

pub type SemanticVocabulary = (
    ahri_tre_types::VocabularyRecord,
    Vec<ahri_tre_types::VocabularyItemRecord>,
);

/// A frozen metadata result and its admission decision. Only the application
/// can turn this into delivery after acknowledging the protected ledger event.
pub struct CapturedVocabulary {
    snapshot: ahri_tre_types::DisclosureSnapshot,
    vocabulary: ahri_tre_types::VocabularyRecord,
    items: Vec<ahri_tre_types::VocabularyItemRecord>,
}

impl CapturedVocabulary {
    pub fn into_parts(
        self,
    ) -> (
        ahri_tre_types::DisclosureSnapshot,
        ahri_tre_types::VocabularyRecord,
        Vec<ahri_tre_types::VocabularyItemRecord>,
    ) {
        (self.snapshot, self.vocabulary, self.items)
    }
}

/// Provenance is descriptive evidence, not a disclosure permission.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum SemanticOrigin {
    Unknown,
    Declared,
    Derived { inputs: Vec<uuid::Uuid> },
}

#[derive(Debug, Clone)]
pub enum SemanticTarget {
    Vocabulary(VocabularyId),
    EntityInstance(i64),
}

impl PgMetadataRepository<'_> {
    /// Captures every field and membership dependency under the same locks as
    /// admission. The caller's suggested inputs are replaced by stored origins.
    pub fn capture_semantic_vocabulary(
        &self,
        intent: &ahri_tre_types::DisclosureIntent,
        target: VocabularyId,
    ) -> Result<CapturedVocabulary, CoreError> {
        let (snapshot, mut values) = self.capture_semantic_vocabularies(intent, &[target])?;
        let (vocabulary, items) = values.pop().ok_or_else(unavailable)?;
        Ok(CapturedVocabulary {
            snapshot,
            vocabulary,
            items,
        })
    }

    /// A single policy decision closes over all values in a nested response.
    pub fn capture_semantic_vocabularies(
        &self,
        intent: &ahri_tre_types::DisclosureIntent,
        targets: &[VocabularyId],
    ) -> Result<(ahri_tre_types::DisclosureSnapshot, Vec<SemanticVocabulary>), CoreError> {
        if intent.lifetime_seconds == 0
            || intent.lifetime_seconds > 600
            || intent.max_bytes == 0
            || intent.max_rows == 0
            || targets.is_empty()
            || targets.len() > 128
        {
            return Err(unavailable());
        }
        self.with_dataset_acceptance(|repository| {
            let mut inputs = std::collections::BTreeSet::new();
            for target in targets {
                repository.with_session_connection(
                    "capture semantic fields",
                    |db| lock_vocabulary(db, *target, intent),
                    |db| lock_vocabulary(db, *target, intent),
                )?;
                let SemanticOrigin::Derived { inputs: sources } =
                    repository.semantic_origin(&SemanticTarget::Vocabulary(*target))?
                else {
                    return Err(unavailable());
                };
                inputs.extend(sources);
            }
            let mut intent = intent.clone();
            intent.inputs = inputs.into_iter().collect();
            intent.latest_inputs.clear();
            let snapshot = repository.with_session_connection(
                "admit semantic content",
                |db| crate::governance::admit_disclosure(db, &intent),
                |db| crate::governance::admit_disclosure(db, &intent),
            )?;
            let mut values = Vec::new();
            let mut rows = 0;
            for target in targets {
                let vocabulary = repository
                    .get_vocabulary(*target)
                    .now_or_never()
                    .ok_or_else(unavailable)??
                    .ok_or_else(unavailable)?;
                let items = repository
                    .list_vocabulary_items(*target)
                    .now_or_never()
                    .ok_or_else(unavailable)??;
                rows += items.len() as u64;
                if rows > intent.max_rows {
                    return Err(unavailable());
                }
                values.push((vocabulary, items));
            }
            Ok((snapshot, values))
        })
    }

    /// Bounded non-content schema, captured atomically with its declaration proof.
    /// Unknown or derived dictionaries have no catalogue item projection.
    pub fn catalogue_vocabulary_items(
        &self,
        target: VocabularyId,
    ) -> Result<Option<Vec<ahri_tre_types::VocabularyItemRecord>>, CoreError> {
        self.with_dataset_acceptance(|repository| {
            if repository.semantic_origin(&SemanticTarget::Vocabulary(target))?
                != SemanticOrigin::Declared
            {
                return Ok(None);
            }
            repository
                .with_session_connection(
                    "capture declared dictionary schema",
                    |db| declared_vocabulary_items(db, target),
                    |db| declared_vocabulary_items(db, target),
                )
                .map(Some)
        })
    }

    /// Record a dependency even if equality checks reused every existing value.
    pub fn retain_vocabulary_contribution(&self, target: VocabularyId) -> Result<(), CoreError> {
        self.with_session_connection(
            "retain vocabulary contribution",
            |db| contribution(db, target),
            |db| contribution(db, target),
        )
    }

    pub fn with_semantic_derivation<T>(
        &self,
        inputs: &[ahri_tre_types::VersionId],
        write: impl FnOnce(PgMetadataRepository<'_>) -> Result<T, CoreError>,
    ) -> Result<T, CoreError> {
        self.with_dataset_acceptance(|repository| {
            repository.begin_semantic_derivation(inputs)?;
            write(repository)
        })
    }

    /// Joins an adapter-owned mutation transaction. This cannot mark a later
    /// independent write or turn source provenance into disclosure permission.
    pub fn begin_semantic_derivation(
        &self,
        inputs: &[ahri_tre_types::VersionId],
    ) -> Result<(), CoreError> {
        use crate::repository::PgMetadataRepositoryConnection;
        if !matches!(
            &self.connection,
            PgMetadataRepositoryConnection::ScopedDirect(_)
                | PgMetadataRepositoryConnection::ScopedOAuth(_)
        ) {
            return Err(CoreError::Conflict(
                "Semantic provenance requires a mutation transaction".into(),
            ));
        }
        let inputs = inputs.iter().map(|id| id.0).collect::<Vec<_>>();
        self.with_session_connection(
            "retain semantic sources",
            |db| begin_derivation(db, &inputs),
            |db| begin_derivation(db, &inputs),
        )
    }

    /// Only explicit dictionary declarations use this boundary. Inference and
    /// instance data must retain immutable source versions instead.
    pub fn with_semantic_declarations<T>(
        &self,
        write: impl FnOnce(PgMetadataRepository<'_>) -> Result<T, CoreError>,
    ) -> Result<T, CoreError> {
        self.with_dataset_acceptance(|repository| {
            repository.with_session_connection(
                "declare dictionary input",
                begin_declaration,
                begin_declaration,
            )?;
            write(repository)
        })
    }

    pub fn semantic_origin(&self, target: &SemanticTarget) -> Result<SemanticOrigin, CoreError> {
        let (statement, id) = match target {
            SemanticTarget::Vocabulary(id) => {
                ("SELECT public.tre_vocabulary_origin($1)::text", id.0)
            }
            SemanticTarget::EntityInstance(id) => {
                ("SELECT public.tre_entity_instance_origin($1)::text", *id)
            }
        };
        self.with_session_connection(
            "read semantic provenance",
            |db| read_origin(db, statement, id),
            |db| read_origin(db, statement, id),
        )
    }
}

fn unavailable() -> CoreError {
    CoreError::Infrastructure("Semantic provenance or admission is unavailable".into())
}

fn lock_vocabulary<E: MetadataExecutor>(
    db: &mut E,
    target: VocabularyId,
    intent: &ahri_tre_types::DisclosureIntent,
) -> Result<(), PgMetaError> {
    let milliseconds = i64::from(intent.lifetime_seconds) * 1000;
    let rows = i64::try_from(intent.max_rows).unwrap_or(i64::MAX);
    let bytes = intent.max_bytes.min(1024 * 1024) as i64;
    db.query_one(
        "bound semantic capture deadline",
        "SELECT set_config('statement_timeout',$1,TRUE)",
        &[&milliseconds.to_string()],
    )?;
    db.query_one(
        "capture bounded semantic fields",
        "SELECT public.tre_lock_semantic_vocabulary($1,$2,$3,$4)",
        &[&target.0, &milliseconds, &rows, &bytes],
    )?;
    Ok(())
}

fn begin_derivation<E: MetadataExecutor>(
    db: &mut E,
    inputs: &[uuid::Uuid],
) -> Result<(), PgMetaError> {
    db.query_one(
        "retain semantic sources",
        "SELECT public.tre_semantic_derivation($1)",
        &[&inputs],
    )?;
    Ok(())
}

fn begin_declaration<E: MetadataExecutor>(db: &mut E) -> Result<(), PgMetaError> {
    db.query_one(
        "declare dictionary input",
        "SELECT public.tre_semantic_declaration()",
        &[],
    )?;
    Ok(())
}

fn read_origin<E: MetadataExecutor>(
    db: &mut E,
    statement: &str,
    id: i64,
) -> Result<SemanticOrigin, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let row = db.query_one("read semantic provenance", statement, &[&id])?;
    serde_json::from_str(&row.required_text(0, "semantic origin")?).map_err(|_| {
        PgMetaError::Decode {
            field: "semantic origin",
            value: "<redacted>".into(),
            message: "invalid semantic provenance".into(),
        }
    })
}

fn contribution<E: MetadataExecutor>(db: &mut E, target: VocabularyId) -> Result<(), PgMetaError> {
    db.query_one(
        "retain vocabulary contribution",
        "SELECT public.tre_semantic_vocabulary_contribution($1)",
        &[&target.0],
    )?;
    Ok(())
}

/// Captures bounded declaration-only schema under provenance locks. Browser
/// Discovery receives no analyst admission capability or inferred values.
pub fn declared_vocabulary_items<E: MetadataExecutor>(
    db: &mut E,
    target: VocabularyId,
) -> Result<Vec<ahri_tre_types::VocabularyItemRecord>, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let row = db.query_one(
        "capture declared vocabulary",
        "SELECT public.tre_declared_vocabulary_items($1)::text",
        &[&target.0],
    )?;
    serde_json::from_str(&row.required_text(0, "declared vocabulary")?).map_err(|_| {
        PgMetaError::Decode {
            field: "declared vocabulary",
            value: "<redacted>".into(),
            message: "invalid declaration capture".into(),
        }
    })
}

/// Definition and immutable Dataset scope resolved by the application.
#[derive(Debug, Clone)]
pub enum SemanticScope {
    Entity {
        study_id: ahri_tre_types::StudyId,
        entity_id: ahri_tre_types::EntityId,
        version_id: ahri_tre_types::VersionId,
    },
    Relation {
        study_id: ahri_tre_types::StudyId,
        relation_id: ahri_tre_types::EntityRelationId,
        version_id: ahri_tre_types::VersionId,
    },
}

/// Opaque revision captured before resolving mutable semantic definitions.
pub struct SemanticResolution {
    revision: i64,
}
impl SemanticResolution {
    pub(crate) fn revision(&self) -> i64 {
        self.revision
    }
}

#[derive(Clone, Copy)]
pub enum SemanticInstance {
    Entity(i64),
    Relation(i64),
}

/// Retained origin closure; its fields cannot be supplied or edited by adapters.
/// It is evidence about metadata, never a substitute for Session admission.
#[derive(Debug)]
pub struct SemanticState {
    scope: SemanticScope,
    revision: i64,
    inputs: Vec<uuid::Uuid>,
}
impl SemanticState {
    pub fn inputs(&self) -> &[uuid::Uuid] {
        &self.inputs
    }
    pub fn scope(&self) -> &SemanticScope {
        &self.scope
    }
    pub fn revision(&self) -> i64 {
        self.revision
    }
}

impl PgMetadataRepository<'_> {
    pub fn begin_semantic_resolution(&self) -> Result<SemanticResolution, CoreError> {
        self.with_session_connection(
            "begin semantic resolution",
            resolution_revision,
            resolution_revision,
        )
    }

    pub fn retain_instance_contribution(
        &self,
        instance: SemanticInstance,
    ) -> Result<(), CoreError> {
        let (collection, id) = match instance {
            SemanticInstance::Entity(id) => ("entity_instance", id),
            SemanticInstance::Relation(id) => ("relation_instances", id),
        };
        self.with_session_connection(
            "retain instance contribution",
            |db| instance_contribution(db, collection, id),
            |db| instance_contribution(db, collection, id),
        )
    }

    pub fn capture_semantic_state(
        &self,
        resolution: &SemanticResolution,
        scope: &SemanticScope,
        limits: &ahri_tre_types::DisclosureBudgets,
    ) -> Result<SemanticState, CoreError> {
        self.with_dataset_acceptance(|repository| {
            let state = repository.with_session_connection(
                "capture semantic state",
                |db| capture_state(db, scope, limits),
                |db| capture_state(db, scope, limits),
            )?;
            if state.revision != resolution.revision {
                return Err(unavailable());
            }
            Ok(state)
        })
    }
}
fn resolution_revision<E: MetadataExecutor>(db: &mut E) -> Result<SemanticResolution, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let row = db.query_one(
        "begin semantic resolution",
        "SELECT public.tre_semantic_revision()::text",
        &[],
    )?;
    let revision = row
        .required_text(0, "semantic revision")?
        .parse()
        .map_err(|_| PgMetaError::Decode {
            field: "semantic revision",
            value: "<redacted>".into(),
            message: "invalid revision".into(),
        })?;
    Ok(SemanticResolution { revision })
}
fn instance_contribution<E: MetadataExecutor>(
    db: &mut E,
    collection: &str,
    id: i64,
) -> Result<(), PgMetaError> {
    db.query_one(
        "retain instance contribution",
        "SELECT public.tre_semantic_instance_contribution($1,$2)",
        &[&collection, &id],
    )?;
    Ok(())
}

fn capture_state<E: MetadataExecutor>(
    db: &mut E,
    scope: &SemanticScope,
    limits: &ahri_tre_types::DisclosureBudgets,
) -> Result<SemanticState, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let (study, definition, relation, version) = match scope {
        SemanticScope::Entity {
            study_id,
            entity_id,
            version_id,
        } => (study_id.0, entity_id.0, false, version_id.0),
        SemanticScope::Relation {
            study_id,
            relation_id,
            version_id,
        } => (study_id.0, relation_id.0, true, version_id.0),
    };
    if limits.total_seconds == 0 || limits.total_seconds > 600 {
        return Err(PgMetaError::Decode {
            field: "semantic limits",
            value: "<redacted>".into(),
            message: "invalid deadline".into(),
        });
    }
    db.query_one(
        "bound semantic state capture",
        "SELECT set_config('statement_timeout',$1,TRUE)",
        &[&(u64::from(limits.total_seconds) * 1000).to_string()],
    )?;
    let rows = i64::try_from(limits.rows).unwrap_or(i64::MAX);
    let bytes = limits.payload_bytes.min(1024 * 1024) as i64;
    let row = db.query_one(
        "capture semantic state",
        "SELECT public.tre_capture_semantic_state($1,$2,$3,$4,$5,$6)::text",
        &[&study, &definition, &relation, &version, &rows, &bytes],
    )?;
    #[derive(Deserialize)]
    struct Capture {
        revision: i64,
        inputs: Vec<uuid::Uuid>,
    }
    let capture: Capture =
        serde_json::from_str(&row.required_text(0, "semantic state")?).map_err(|_| {
            PgMetaError::Decode {
                field: "semantic state",
                value: "<redacted>".into(),
                message: "invalid semantic capture".into(),
            }
        })?;
    Ok(SemanticState {
        scope: scope.clone(),
        revision: capture.revision,
        inputs: capture.inputs,
    })
}

/// Called in the same adapter-owned transaction as the admission decision.
pub fn verify_semantic_state<E: crate::MetadataTransaction + ?Sized>(
    db: &mut E,
    state: &SemanticState,
) -> Result<(), PgMetaError> {
    db.query_one(
        "retain semantic snapshot",
        "SELECT public.tre_lock_semantic_state($1)",
        &[&state.revision],
    )?;
    Ok(())
}
impl PgMetadataRepository<'_> {
    /// Retains the captured state, verifies durable admission, and commits all
    /// semantic changes and their origins together. Any callback failure rolls back.
    pub fn with_admitted_semantic_mutation<T>(
        &self,
        state: &SemanticState,
        admission: uuid::Uuid,
        write: impl FnOnce(PgMetadataRepository<'_>) -> Result<T, CoreError>,
    ) -> Result<T, CoreError> {
        self.with_dataset_acceptance(|repository| {
            repository.with_session_connection(
                "authorize semantic mutation",
                |db| mutation_context(db, state, admission),
                |db| mutation_context(db, state, admission),
            )?;
            write(repository)
        })
    }
}
fn mutation_context<E: MetadataExecutor>(
    db: &mut E,
    state: &SemanticState,
    admission: uuid::Uuid,
) -> Result<(), PgMetaError> {
    db.query_one(
        "authorize semantic mutation",
        "SELECT public.tre_semantic_mutation_context($1,$2,$3)",
        &[&admission, &state.revision, &state.inputs],
    )?;
    Ok(())
}

pub enum SemanticMappings {
    Entity(Vec<ahri_tre_types::StudyEntityInstanceRecord>),
    Relation(Vec<ahri_tre_types::StudyRelationInstanceRecord>),
}
impl PgMetadataRepository<'_> {
    /// Reads only the mapping scope whose row/byte bounds and revision were
    /// captured. This is available inside the retained adapter transaction.
    pub fn retained_semantic_mappings(
        &self,
        state: &SemanticState,
    ) -> Result<SemanticMappings, CoreError> {
        use crate::repository::PgMetadataRepositoryConnection;
        if !matches!(
            self.connection,
            PgMetadataRepositoryConnection::ScopedDirect(_)
                | PgMetadataRepositoryConnection::ScopedOAuth(_)
        ) {
            return Err(unavailable());
        }
        self.with_session_connection(
            "read retained semantic mappings",
            |db| read_mappings(db, state),
            |db| read_mappings(db, state),
        )
    }
}
fn read_mappings<E: MetadataExecutor>(
    db: &mut E,
    state: &SemanticState,
) -> Result<SemanticMappings, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    db.query_one(
        "retain semantic mapping revision",
        "SELECT public.tre_lock_semantic_state($1)",
        &[&state.revision],
    )?;
    let (study, definition, sql, relation) = match state.scope {
        SemanticScope::Entity {
            study_id,
            entity_id,
            ..
        } => (
            study_id.0,
            entity_id.0,
            "SELECT to_jsonb(m)::text FROM public.study_entity_instances m WHERE study_id=$1 AND entity_id::bigint=$2 ORDER BY entity_instance_id",
            false,
        ),
        SemanticScope::Relation {
            study_id,
            relation_id,
            ..
        } => (
            study_id.0,
            relation_id.0,
            "SELECT (to_jsonb(m)-'entityrelation_id' || jsonb_build_object('entity_relation_id',m.entityrelation_id))::text FROM public.study_relation_instances m WHERE study_id=$1 AND entityrelation_id::bigint=$2 ORDER BY relation_instance_id",
            true,
        ),
    };
    let rows = db.query_many(
        "read retained semantic mappings",
        sql,
        &[&study, &definition],
    )?;
    let mut payloads = Vec::new();
    for row in rows {
        payloads.push(row.required_text(0, "semantic mapping")?);
    }
    let decode_error = |_| PgMetaError::Decode {
        field: "semantic mapping",
        value: "<redacted>".into(),
        message: "invalid stored mapping".into(),
    };
    if relation {
        Ok(SemanticMappings::Relation(
            payloads
                .iter()
                .map(|v| serde_json::from_str(v).map_err(decode_error))
                .collect::<Result<_, _>>()?,
        ))
    } else {
        Ok(SemanticMappings::Entity(
            payloads
                .iter()
                .map(|v| serde_json::from_str(v).map_err(decode_error))
                .collect::<Result<_, _>>()?,
        ))
    }
}