ahri_tre_pgmeta/

governance.rs

//! Guarded Asset classification and operator-owned preserving conversion.
//! Both direct and libpq Sessions use the same transaction/function boundary.
use crate::{MetadataExecutor, PgMetaError, schema::SchemaStatusRow};
use ahri_tre_types::AssetClassification;
pub use ahri_tre_types::{DisclosureIntent, DisclosureSnapshot};
use uuid::Uuid;

const STRUCTURE: &str = include_str!("governance_schema.sql");

/// Narrow governance maintenance over administrator authority already resolved
/// for the Session. No configuration documents or Secret stores are reopened.
#[derive(Clone)]
pub struct GovernanceMaintenance {
    repository: crate::PgMetadataRepository<'static>,
}
impl crate::PgDatasetMaintenance {
    pub fn governance(&self) -> GovernanceMaintenance {
        GovernanceMaintenance {
            repository: self.repository.clone(),
        }
    }
}
impl GovernanceMaintenance {
    pub fn bind_dataset(
        &self,
        location: &ahri_tre_types::GovernedDatasetLocation,
    ) -> Result<(), ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "bind immutable Dataset",
            |db| bind_dataset_location(db, location),
            |db| bind_dataset_location(db, location),
        )
    }
    pub fn binding(&self) -> Result<GovernanceLedgerBinding, ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "Governance binding",
            ledger_binding,
            ledger_binding,
        )
    }
    pub fn activate_identity(
        &self,
        issuer: &str,
        subject: &str,
        principal: &str,
    ) -> Result<(), ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "activate Governance identity",
            |db| activate_identity(db, issuer, subject, principal),
            |db| activate_identity(db, issuer, subject, principal),
        )
    }
    pub fn pending(&self, limit: u32) -> Result<Vec<Uuid>, ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "pending Governance evidence",
            |db| pending_evidence(db, limit),
            |db| pending_evidence(db, limit),
        )
    }
    pub fn project(
        &self,
        event: Uuid,
        append: impl FnMut(Uuid, &str) -> Result<(), ()>,
    ) -> Result<bool, ahri_tre_core::CoreError> {
        let append = std::cell::RefCell::new(append);
        self.repository.with_session_connection(
            "project Governance evidence",
            |db| project_evidence(db, event, |id, text| (append.borrow_mut())(id, text)),
            |db| project_evidence(db, event, |id, text| (append.borrow_mut())(id, text)),
        )
    }
    pub fn unfinished(
        &self,
        limit: u32,
    ) -> Result<Vec<UnfinishedDisclosure>, ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "unfinished disclosures",
            |db| unfinished_disclosures(db, limit),
            |db| unfinished_disclosures(db, limit),
        )
    }
    pub fn abandon(&self, receipt: &UnfinishedDisclosure) -> Result<(), ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "abandon disclosure",
            |db| abandon_disclosure(db, receipt),
            |db| abandon_disclosure(db, receipt),
        )
    }
}
fn activate_identity<E: MetadataExecutor>(
    db: &mut E,
    issuer: &str,
    subject: &str,
    principal: &str,
) -> Result<(), PgMetaError> {
    db.query_one(
        "activate Governance identity",
        "SELECT tre_governance.activate_identity($1,$2,$3)",
        &[&issuer, &subject, &principal],
    )?;
    Ok(())
}

pub(crate) fn validate_derivation<E: MetadataExecutor>(
    db: &mut E,
    inputs: &[Uuid],
    risk: ahri_tre_types::AssetRisk,
    inherit: bool,
) -> Result<(), PgMetaError> {
    db.query_one(
        "validate derivation",
        "SELECT public.tre_validate_derivation($1,$2,$3)",
        &[&inputs, &risk.as_str(), &inherit],
    )?;
    Ok(())
}

impl crate::PgMetadataRepository<'_> {
    /// Pre-connection destination authority; final admission rechecks current state.
    pub fn authorize_acquisition_destination(
        &self,
        study: ahri_tre_types::StudyId,
        classification: &ahri_tre_types::IngestClassification,
    ) -> Result<(), ahri_tre_core::CoreError> {
        self.with_session_connection(
            "authorize acquisition destination",
            |db| acquisition_destination(db, study.0, classification),
            |db| acquisition_destination(db, study.0, classification),
        )
    }
    /// Authorizes the one freshly ingested Datafile pinned by the App upload
    /// workflow. This does not confer trusted or arbitrary derivation authority.
    pub fn validate_uploaded_table_source(
        &self,
        source: ahri_tre_types::VersionId,
        study: ahri_tre_types::StudyId,
        risk: ahri_tre_types::AssetRisk,
    ) -> Result<(), ahri_tre_core::CoreError> {
        self.with_session_connection(
            "validate uploaded table source",
            |db| uploaded_table_source(db, source.0, study.0, risk),
            |db| uploaded_table_source(db, source.0, study.0, risk),
        )
    }
}
fn acquisition_destination<E: MetadataExecutor>(
    db: &mut E,
    study: Uuid,
    classification: &ahri_tre_types::IngestClassification,
) -> Result<(), PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let row = db.query_one("authorize acquisition destination",
        "SELECT (EXISTS (SELECT 1 FROM public.studies WHERE study_id=$1) AND public.tre_user_has_study_access($1) AND ($2 <> 'low' OR (public.tre_user_can_administer_study_access($1) AND length(btrim($3)) > 0)))::text",
        &[&study, &classification.risk.as_str(), &classification.justification])?;
    if row.required_text(0, "permitted")? != "true" {
        return Err(invalid("Acquisition destination is unavailable"));
    }
    Ok(())
}
fn uploaded_table_source<E: MetadataExecutor>(
    db: &mut E,
    source: Uuid,
    study: Uuid,
    risk: ahri_tre_types::AssetRisk,
) -> Result<(), PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    // The existing function locks and authorizes input identities. HIGH here is
    // its conservative read-authority rule, not the output's classification.
    validate_derivation(db, &[source], ahri_tre_types::AssetRisk::High, false)?;
    let row = db.query_one("verify uploaded table identity",
        "SELECT count(*)::text FROM public.datafiles d JOIN public.asset_versions v ON v.version_id=d.datafile_id JOIN public.assets a USING(asset_id) WHERE v.version_id=$1 AND a.study_id=$2 AND public.tre_asset_classification(a.asset_id)->>'risk'=$3",
        &[&source, &study, &risk.as_str()])?;
    if number(&row, 0)? != 1 {
        return Err(invalid("uploaded source scope or classification changed"));
    }
    Ok(())
}

/// Explicit appointment. Never inferred from the first Study creator.
#[derive(Debug, Clone)]
pub struct GovernanceAppointment {
    issuers: Vec<String>,
    subjects: Vec<String>,
}

impl GovernanceAppointment {
    pub fn new(issuer: &str, subject: &str) -> Result<Self, PgMetaError> {
        Self::from_identities([(issuer, subject)])
    }
    pub fn from_identities<'a>(
        identities: impl IntoIterator<Item = (&'a str, &'a str)>,
    ) -> Result<Self, PgMetaError> {
        let mut pairs = Vec::new();
        for (issuer, subject) in identities {
            if !issuer.starts_with("https://")
                || issuer.len() > 2048
                || issuer.bytes().any(|b| b.is_ascii_control())
                || subject.trim().is_empty()
                || subject.len() > 255
                || !subject.is_ascii()
                || subject.bytes().any(|b| b.is_ascii_control())
                || pairs.contains(&(issuer, subject))
            {
                return Err(invalid("governance appointment is invalid"));
            }
            pairs.push((issuer, subject));
        }
        if pairs.is_empty() || pairs.len() > 32 {
            return Err(invalid("explicit governance appointments required"));
        }
        let (issuers, subjects) = pairs
            .into_iter()
            .map(|(i, s)| (i.to_owned(), s.to_owned()))
            .unzip();
        Ok(Self { issuers, subjects })
    }
}

pub struct GovernanceUpgrade;

#[derive(Debug)]
pub struct GovernanceLedgerBinding {
    pub study_id: Uuid,
    pub asset_id: Uuid,
    pub version_id: Uuid,
}

/// Operator-only binding; the application uses this to open the system ledger
/// through the Lake adapter. It contains no physical location or credential.
pub fn ledger_binding<E: MetadataExecutor>(
    db: &mut E,
) -> Result<GovernanceLedgerBinding, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let row = db.query_one("governance ledger binding", "SELECT study_id::text,ledger_asset_id::text,ledger_version_id::text FROM tre_governance.state", &[])?;
    let id = |index| {
        row.required_text(index, "governance identity")?
            .parse()
            .map_err(|_| invalid("invalid governance identity"))
    };
    Ok(GovernanceLedgerBinding {
        study_id: id(0)?,
        asset_id: id(1)?,
        version_id: id(2)?,
    })
}

#[derive(Debug)]
pub struct GovernanceUpgradePlan {
    pub unclassified_assets: u64,
    pub immutable_versions: u64,
    pub prepared: bool,
}

impl GovernanceUpgrade {
    pub fn plan<E: MetadataExecutor>(db: &mut E) -> Result<GovernanceUpgradePlan, PgMetaError>
    where
        E::Row: SchemaStatusRow,
    {
        let status = crate::datastore_schema_status(db)?;
        if !matches!(
            status.status,
            crate::DatastoreSchemaCompatibility::Current
                | crate::DatastoreSchemaCompatibility::Pending
        ) {
            return Err(invalid("unsupported metadata baseline"));
        }
        let row = db.query_one(
            "governance structure",
            "SELECT to_regclass('tre_governance.classifications')::text",
            &[],
        )?;
        let prepared = row.optional_text(0).is_some();
        let sql = if prepared {
            "SELECT count(*)::text, (SELECT count(*)::text FROM public.asset_versions) FROM public.assets a WHERE NOT EXISTS (SELECT 1 FROM tre_governance.classifications c WHERE c.asset_id=a.asset_id)"
        } else {
            "SELECT count(*)::text, (SELECT count(*)::text FROM public.asset_versions) FROM public.assets"
        };
        let row = db.query_one("governance inventory", sql, &[])?;
        Ok(GovernanceUpgradePlan {
            unclassified_assets: number(&row, 0)?,
            immutable_versions: number(&row, 1)?,
            prepared,
        })
    }

    pub fn prepare<E: MetadataExecutor>(
        db: &mut E,
        appointment: &GovernanceAppointment,
    ) -> Result<(), PgMetaError>
    where
        E::Row: SchemaStatusRow,
    {
        Self::plan(db)?;
        db.with_transaction("prepare governance", |tx| {
            tx.query_one(
                "serialize upgrade",
                "SELECT pg_advisory_xact_lock(190921,9)",
                &[],
            )?;
            // FORCE RLS normally applies even to the metadata owner. The
            // privileged operator's bounded DDL transaction temporarily removes
            // FORCE only on that owner's tables, with AccessExclusive locks.
            // Restore every original flag before commit; no session ever observes
            // a committed policy gap, including on an error/rollback.
            let privileged = tx.query_one("verify upgrade operator", "SELECT (rolsuper OR rolbypassrls)::text FROM pg_catalog.pg_roles WHERE rolname=SESSION_USER", &[])?;
            if privileged.required_text(0,"operator authority")? != "true" { return Err(invalid("privileged upgrade operator required")); }
            let forced = tx.query_many("owned forced policies", "SELECT n.nspname::text,c.relname::text FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='public' AND c.relowner=CURRENT_USER::regrole AND c.relforcerowsecurity ORDER BY c.oid", &[])?;
            let relations = forced.iter().map(|row| Ok(format!("\"{}\".\"{}\"", row.required_text(0,"schema")?.replace('"',"\"\""), row.required_text(1,"relation")?.replace('"',"\"\"")))).collect::<Result<Vec<_>, PgMetaError>>()?;
            for relation in &relations { tx.execute_command("lock owner migration table", &format!("ALTER TABLE {relation} NO FORCE ROW LEVEL SECURITY"), &[])?; }
            tx.execute_command("governance structure", STRUCTURE, &[])?;
            let owner=tx.query_one("retain metadata owner","SELECT CURRENT_USER::text",&[])?.required_text(0,"metadata owner")?;
            tx.execute_command("operator-owned migration steps","SET LOCAL ROLE NONE",&[])?;
            tx.execute_command("preserve isolated lookup and remove inherited ownership",include_str!("governance_operator.sql"),&[])?;
            tx.execute_command("restore metadata owner",&format!("SET LOCAL ROLE \"{}\"",owner.replace('"',"\"\"")),&[])?;

            tx.execute_command(
                "disclosure structure",
                include_str!("disclosure_schema.sql"),
                &[],
            )?;
            tx.execute_command("semantic provenance", include_str!("semantic_provenance.sql"), &[])?;
            tx.execute_command("semantic instance readbacks", include_str!("semantic_records.sql"), &[])?;
            tx.query_one(
                "appoint governance",
                "SELECT tre_governance.provision_custodians($1,$2)",
                &[&appointment.issuers, &appointment.subjects],
            )?;
            tx.execute_command(
                "governance policy guards",
                include_str!("governance_policy.sql"),
                &[],
            )?;
            tx.execute_command("governed ingest admission",include_str!("governance_ingest.sql"),&[])?;
            tx.execute_command("record governance structure migration","INSERT INTO public.ahri_tre_schema_migrations(version,description,scope) VALUES($1,'Guarded governance and disclosure admission','metadata') ON CONFLICT(version) DO NOTHING",&[&crate::schema::GOVERNANCE_SCHEMA_VERSION])?;
            tx.execute_command("record semantic provenance migration", "INSERT INTO public.ahri_tre_schema_migrations(version,description,scope) VALUES($1,'Durable semantic field provenance','metadata') ON CONFLICT(version) DO NOTHING", &[&crate::schema::SEMANTIC_PROVENANCE_SCHEMA_VERSION])?;
            tx.execute_command("record semantic instances migration", "INSERT INTO public.ahri_tre_schema_migrations(version,description,scope) VALUES($1,'Retained instance and link membership histories','metadata') ON CONFLICT(version) DO NOTHING", &[&crate::schema::SEMANTIC_INSTANCES_SCHEMA_VERSION])?;
            tx.execute_command("record Governance custodians migration", "INSERT INTO public.ahri_tre_schema_migrations(version,description,scope) VALUES($1,'Multiple explicit Governance appointments','metadata') ON CONFLICT(version) DO NOTHING", &[&crate::CURRENT_DATASTORE_SCHEMA_VERSION])?;
            for relation in relations { tx.execute_command("restore forced policy", &format!("ALTER TABLE {relation} FORCE ROW LEVEL SECURITY"), &[])?; }
            Ok(())
        })
    }

    /// Bounded resumable Q1 conversion. Row locking and insert-only state mean a
    /// retry cannot overwrite an intervening custodian classification.
    pub fn classify_existing<E: MetadataExecutor>(
        db: &mut E,
        batch: u32,
    ) -> Result<u64, PgMetaError>
    where
        E::Row: SchemaStatusRow,
    {
        if batch == 0 || batch > 1000 {
            return Err(invalid("invalid migration batch"));
        }
        let row = db.query_one(
            "classify existing Assets",
            "SELECT tre_governance.convert_existing($1)::text",
            &[&(batch as i32)],
        )?;
        number(&row, 0)
    }
}

pub fn classification<E: MetadataExecutor>(
    db: &mut E,
    asset: Uuid,
) -> Result<Option<AssetClassification>, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let row = db.query_one(
        "read Asset classification",
        "SELECT public.tre_asset_classification($1)::text",
        &[&asset],
    )?;
    row.optional_text(0)
        .map(|text| {
            serde_json::from_str(&text).map_err(|_| invalid("invalid classification evidence"))
        })
        .transpose()
}

fn number<R: SchemaStatusRow>(row: &R, index: usize) -> Result<u64, PgMetaError> {
    row.required_text(index, "count")?
        .parse()
        .map_err(|_| invalid("invalid governance count"))
}

/// The database rechecks current owning-Study custodianship and revision while
/// holding the same locks used for admission. This is an all-version decision.
pub fn reclassify<E: MetadataExecutor>(
    db: &mut E,
    asset: Uuid,
    expected_revision: u64,
    risk: ahri_tre_types::AssetRisk,
    justification: &str,
) -> Result<AssetClassification, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let revision = i64::try_from(expected_revision).map_err(|_| invalid("invalid revision"))?;
    let row = db.query_one(
        "reclassify Asset",
        "SELECT public.tre_reclassify_asset($1,$2,$3,$4)::text",
        &[&asset, &revision, &risk.as_str(), &justification],
    )?;
    serde_json::from_str(&row.required_text(0, "classification")?)
        .map_err(|_| invalid("invalid classification evidence"))
}

fn invalid(message: &str) -> PgMetaError {
    PgMetaError::Decode {
        field: "governance",
        value: "<redacted>".into(),
        message: message.into(),
    }
}

pub fn admit_disclosure<E: MetadataExecutor>(
    db: &mut E,
    intent: &DisclosureIntent,
) -> Result<DisclosureSnapshot, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let encoded =
        serde_json::to_string(intent).map_err(|_| invalid("invalid disclosure intent"))?;
    let row = db.query_one(
        "admit disclosure",
        "SELECT public.tre_admit_disclosure($1::text::jsonb)::text",
        &[&encoded],
    )?;
    serde_json::from_str(&row.required_text(0, "disclosure snapshot")?)
        .map_err(|_| invalid("invalid disclosure snapshot"))
}

/// Keep the metadata policy/version locks until the Lake adapter has captured
/// immutable input capabilities. A failed capture rolls the decision back.
pub fn admit_disclosure_with_inputs<E: MetadataExecutor, T>(
    db: &mut E,
    intent: &DisclosureIntent,
    capture: impl FnOnce(
        &DisclosureSnapshot,
        Vec<ahri_tre_types::GovernedDatasetLocation>,
    ) -> Result<T, ()>,
) -> Result<(DisclosureSnapshot, T), PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    admit_disclosure_with_semantic_inputs(db, intent, None, capture)
}

pub fn admit_disclosure_with_semantic_inputs<E: MetadataExecutor, T>(
    db: &mut E,
    intent: &DisclosureIntent,
    semantic: Option<&crate::semantic::SemanticState>,
    capture: impl FnOnce(
        &DisclosureSnapshot,
        Vec<ahri_tre_types::GovernedDatasetLocation>,
    ) -> Result<T, ()>,
) -> Result<(DisclosureSnapshot, T), PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let encoded =
        serde_json::to_string(intent).map_err(|_| invalid("invalid disclosure intent"))?;
    db.with_transaction("admit and pin disclosure", |tx| {
        if let Some(state) = semantic {
            crate::semantic::verify_semantic_state(tx, state)?;
        }
        let row = tx.query_one(
            "admit disclosure",
            "SELECT public.tre_admit_disclosure($1::text::jsonb)::text",
            &[&encoded],
        )?;
        let snapshot = serde_json::from_str(&row.required_text(0, "snapshot")?)
            .map_err(|_| invalid("invalid disclosure snapshot"))?;
        let origins = tx.query_one(
            "resolve admitted origins",
            "SELECT public.tre_disclosure_locations($1)::text",
            &[&intent.admission_id],
        )?;
        let origins = serde_json::from_str(&origins.required_text(0, "input origins")?)
            .map_err(|_| invalid("invalid input origins"))?;
        let inputs =
            capture(&snapshot, origins).map_err(|_| invalid("immutable inputs unavailable"))?;
        Ok((snapshot, inputs))
    })
}

pub fn start_disclosure<E: MetadataExecutor>(db: &mut E, id: Uuid) -> Result<Uuid, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let row = db.query_one(
        "start disclosure",
        "SELECT public.tre_start_disclosure($1)::text",
        &[&id],
    )?;
    row.required_text(0, "delivery evidence")?
        .parse()
        .map_err(|_| invalid("invalid event identity"))
}

pub fn finish_disclosure<E: MetadataExecutor>(
    db: &mut E,
    id: Uuid,
    outcome: ahri_tre_types::DisclosureOutcome,
    bytes: Option<u64>,
    rows: Option<u64>,
) -> Result<Uuid, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let bytes = bytes
        .map(i64::try_from)
        .transpose()
        .map_err(|_| invalid("invalid byte count"))?;
    let rows = rows
        .map(i64::try_from)
        .transpose()
        .map_err(|_| invalid("invalid row count"))?;
    let row = db.query_one(
        "finish disclosure",
        "SELECT public.tre_finish_disclosure($1,$2,$3,$4)::text",
        &[&id, &outcome.as_str(), &bytes, &rows],
    )?;
    row.required_text(0, "terminal evidence")?
        .parse()
        .map_err(|_| invalid("invalid event identity"))
}

/// Bounded operator recovery inventory. Pending older disclosure completions
/// are independent of the evidence required by a new admission.
pub fn pending_evidence<E: MetadataExecutor>(
    db: &mut E,
    limit: u32,
) -> Result<Vec<Uuid>, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    if limit == 0 || limit > 1000 {
        return Err(invalid("invalid evidence batch"));
    }
    db.query_many("pending governance evidence", "SELECT event_id::text FROM tre_governance.events WHERE NOT ledger_acknowledged ORDER BY occurred_at,event_id LIMIT $1", &[&(limit as i64)])?
        .iter().map(|row| row.required_text(0,"event identity")?.parse().map_err(|_| invalid("invalid event identity"))).collect()
}

pub struct UnfinishedDisclosure {
    pub admission_id: Uuid,
    pub coordinator_id: Uuid,
    pub generation_id: Uuid,
}

pub fn unfinished_disclosures<E: MetadataExecutor>(
    db: &mut E,
    limit: u32,
) -> Result<Vec<UnfinishedDisclosure>, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    if limit == 0 || limit > 1000 {
        return Err(invalid("invalid recovery batch"));
    }
    db.query_many("unfinished disclosures","SELECT admission_id::text,snapshot->>'coordinator_id',snapshot->>'worker_generation' FROM tre_governance.disclosures WHERE terminal_evidence IS NULL ORDER BY admission_id LIMIT $1",&[&(limit as i64)])?.iter().map(|row| {
        let id=|index|row.required_text(index,"recovery identity")?.parse().map_err(|_|invalid("invalid recovery identity"));
        Ok(UnfinishedDisclosure {admission_id:id(0)?,coordinator_id:id(1)?,generation_id:id(2)?})
    }).collect()
}

/// The application must prove executor exclusion before calling this operator
/// function. It terminalizes evidence without recreating user authority.
pub fn abandon_disclosure<E: MetadataExecutor>(
    db: &mut E,
    receipt: &UnfinishedDisclosure,
) -> Result<(), PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    db.query_one(
        "recover unknown disclosure",
        "SELECT tre_governance.abandon_disclosure($1,$2,$3)",
        &[
            &receipt.admission_id,
            &receipt.coordinator_id,
            &receipt.generation_id,
        ],
    )?;
    Ok(())
}

/// Serialize one event's projection across workers and both durable stores.
/// The callback must append and read back the identical event in the ledger.
/// A lost acknowledgement leaves an independently retryable outbox event.
pub fn project_evidence<E: MetadataExecutor>(
    db: &mut E,
    event: Uuid,
    append_and_verify: impl FnOnce(Uuid, &str) -> Result<(), ()>,
) -> Result<bool, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    db.with_transaction("project governance evidence",|tx| {
        let row=tx.query_one("lock governance event", "SELECT jsonb_build_object('event_id',event_id,'study_id',study_id,'asset_id',asset_id,'event_kind',event_kind,'actor',actor,'occurred_at',to_char(occurred_at AT TIME ZONE 'UTC','YYYY-MM-DD\"T\"HH24:MI:SS.US\"Z\"'),'evidence',evidence)::text,ledger_acknowledged::text FROM tre_governance.events WHERE event_id=$1 FOR UPDATE", &[&event])?;
        if row.required_text(1,"acknowledgement")?=="true" { return Ok(false); }
        append_and_verify(event,&row.required_text(0,"event")?).map_err(|_|invalid("ledger evidence unavailable"))?;
        tx.execute_command("acknowledge governance event", "UPDATE tre_governance.events SET ledger_acknowledged=TRUE WHERE event_id=$1", &[&event])?;
        Ok(true)
    })
}

/// Current owning-Study custodians and the appointed governance custodian may
/// discover only acknowledged event identities through the ordinary Session.
pub fn scoped_evidence_ids<E: MetadataExecutor>(
    db: &mut E,
    study: Uuid,
    after: chrono::DateTime<chrono::Utc>,
    after_id: Uuid,
    limit: u32,
) -> Result<Vec<Uuid>, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    if limit == 0 || limit > 100 {
        return Err(invalid("invalid evidence page"));
    }
    let timestamp = after.to_rfc3339();
    db.query_many("scoped governance history", "SELECT event_id::text FROM public.tre_governance_evidence_ids($1,$2::text::timestamptz,$3,$4::bigint::integer)", &[&study,&timestamp,&after_id,&(limit as i64)])?.iter()
        .map(|row| row.required_text(0,"event identity")?.parse().map_err(|_|invalid("invalid event identity"))).collect()
}

/// Only the resolved operator capability may attest an actual Lake table.
pub fn bind_dataset_location<E: MetadataExecutor>(
    db: &mut E,
    location: &ahri_tre_types::GovernedDatasetLocation,
) -> Result<(), PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let encoded = serde_json::to_string(location).map_err(|_| invalid("invalid Dataset origin"))?;
    db.with_transaction("bind Dataset origin", |tx| {
        tx.execute_command("retain immutable origin", "INSERT INTO tre_governance.dataset_locations SELECT * FROM jsonb_populate_record(NULL::tre_governance.dataset_locations,$1::text::jsonb) ON CONFLICT(version_id) DO NOTHING", &[&encoded])?;
        let row = tx.query_one("verify immutable origin", "SELECT ((to_jsonb(l)-'lake_snapshot')=($2::text::jsonb-'lake_snapshot'))::text FROM tre_governance.dataset_locations l WHERE version_id=$1", &[&location.version_id,&encoded])?;
        if row.required_text(0,"origin equality")? != "true" { return Err(invalid("immutable Dataset origin conflicts")); }
        Ok(())
    })
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct LegacyDatasetOrigin {
    pub version_id: Uuid,
    pub asset_id: Uuid,
    pub study_id: Uuid,
    pub name: ahri_tre_types::NcName,
    pub major: i32,
    pub minor: i32,
    pub patch: i32,
}
impl GovernanceMaintenance {
    pub fn convert_existing(&self, batch: u32) -> Result<u64, ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "convert legacy classifications",
            |db| GovernanceUpgrade::classify_existing(db, batch),
            |db| GovernanceUpgrade::classify_existing(db, batch),
        )
    }
    pub fn upgrade_plan(&self) -> Result<GovernanceUpgradePlan, ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "inspect governance conversion",
            GovernanceUpgrade::plan,
            GovernanceUpgrade::plan,
        )
    }
    pub fn unbound_legacy_datasets(
        &self,
        batch: u32,
    ) -> Result<Vec<LegacyDatasetOrigin>, ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "legacy Dataset origins",
            |db| unbound_legacy_datasets(db, batch),
            |db| unbound_legacy_datasets(db, batch),
        )
    }
}
fn unbound_legacy_datasets<E: MetadataExecutor>(
    db: &mut E,
    batch: u32,
) -> Result<Vec<LegacyDatasetOrigin>, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    if batch == 0 || batch > 1000 {
        return Err(invalid("invalid origin batch"));
    }
    db.query_many("legacy Dataset origin batch", "SELECT jsonb_build_object('version_id',v.version_id,'asset_id',a.asset_id,'study_id',a.study_id,'name',a.name,'major',v.major,'minor',v.minor,'patch',v.patch)::text FROM public.assets a JOIN public.asset_versions v USING(asset_id) JOIN public.datasets d ON d.dataset_id=v.version_id JOIN tre_governance.legacy_assets legacy USING(asset_id,study_id) WHERE NOT EXISTS(SELECT 1 FROM tre_governance.dataset_locations l WHERE l.version_id=v.version_id) ORDER BY v.version_id LIMIT $1",&[&(batch as i64)])?.iter().map(|row|serde_json::from_str(&row.required_text(0,"Dataset origin")?).map_err(|_|invalid("invalid Dataset origin"))).collect()
}

/// Computation may consume high inputs, but has no content-delivery capability.
/// The final output admission independently revalidates its forced-high policy.
pub fn capture_derivation_inputs<E: MetadataExecutor, T>(
    db: &mut E,
    inputs: &[Uuid],
    capture: impl FnOnce(Vec<ahri_tre_types::GovernedDatasetLocation>) -> Result<T, ()>,
) -> Result<T, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    db.with_transaction("authorize and capture derivation", |tx| {
        let row = tx.query_one(
            "resolve derivation closure",
            "SELECT public.tre_derivation_locations($1)::text",
            &[&inputs],
        )?;
        let origins = serde_json::from_str(&row.required_text(0, "origins")?)
            .map_err(|_| invalid("invalid immutable origins"))?;
        capture(origins).map_err(|_| invalid("derivation inputs unavailable"))
    })
}

impl GovernanceMaintenance {
    pub fn bind_datafile(
        &self,
        binding: &ahri_tre_types::GovernedDatafileBinding,
    ) -> Result<(), ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "attest Datafile",
            |db| bind_datafile(db, binding),
            |db| bind_datafile(db, binding),
        )
    }
}
pub fn bind_datafile<E: MetadataExecutor>(
    db: &mut E,
    binding: &ahri_tre_types::GovernedDatafileBinding,
) -> Result<(), PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let encoded =
        serde_json::to_string(binding).map_err(|_| invalid("invalid Datafile attestation"))?;
    db.with_transaction("retain Datafile origin",|tx| {
        tx.execute_command("retain Datafile attestation","INSERT INTO tre_governance.datafile_bindings SELECT * FROM jsonb_populate_record(NULL::tre_governance.datafile_bindings,$1::text::jsonb) ON CONFLICT(version_id) DO NOTHING",&[&encoded])?;
        let row=tx.query_one("verify Datafile attestation","SELECT (to_jsonb(b)=$2::text::jsonb)::text FROM tre_governance.datafile_bindings b WHERE version_id=$1",&[&binding.version_id,&encoded])?;
        if row.required_text(0,"attestation equality")?!="true" {return Err(invalid("immutable Datafile origin conflicts"));}
        Ok(())
    })
}
pub fn admit_disclosure_with_datafile<E: MetadataExecutor, T>(
    db: &mut E,
    intent: &DisclosureIntent,
    capture: impl FnOnce(&DisclosureSnapshot, ahri_tre_types::AdmittedDatafileInput) -> Result<T, ()>,
) -> Result<(DisclosureSnapshot, T), PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    let encoded =
        serde_json::to_string(intent).map_err(|_| invalid("invalid disclosure intent"))?;
    db.with_transaction("admit and capture Datafile", |tx| {
        let row = tx.query_one(
            "admit disclosure",
            "SELECT public.tre_admit_disclosure($1::text::jsonb)::text",
            &[&encoded],
        )?;
        let snapshot = serde_json::from_str(&row.required_text(0, "snapshot")?)
            .map_err(|_| invalid("invalid disclosure snapshot"))?;
        let row = tx.query_one(
            "resolve admitted Datafile",
            "SELECT public.tre_disclosure_datafile($1)::text",
            &[&intent.admission_id],
        )?;
        let input = serde_json::from_str(&row.required_text(0, "Datafile input")?)
            .map_err(|_| invalid("Datafile input unavailable"))?;
        let captured = capture(&snapshot, input)
            .map_err(|_| invalid("immutable Datafile capture unavailable"))?;
        Ok((snapshot, captured))
    })
}

#[derive(serde::Deserialize)]
pub struct LegacyDatafileOrigin {
    pub asset_id: Uuid,
    pub study_id: Uuid,
    pub datafile: ahri_tre_types::DataFileRecord,
}
impl GovernanceMaintenance {
    pub fn unbound_legacy_datafiles(
        &self,
        batch: u32,
    ) -> Result<Vec<LegacyDatafileOrigin>, ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "legacy Datafile origins",
            |db| unbound_legacy_datafiles(db, batch),
            |db| unbound_legacy_datafiles(db, batch),
        )
    }
}
fn unbound_legacy_datafiles<E: MetadataExecutor>(
    db: &mut E,
    batch: u32,
) -> Result<Vec<LegacyDatafileOrigin>, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    if batch == 0 || batch > 1000 {
        return Err(invalid("invalid origin batch"));
    }
    db.query_many("legacy Datafile origin batch", "SELECT jsonb_build_object('asset_id',a.asset_id,'study_id',a.study_id,'datafile',tre_governance.datafile_descriptor(d))::text FROM public.assets a JOIN public.asset_versions v USING(asset_id) JOIN public.datafiles d ON d.datafile_id=v.version_id JOIN tre_governance.legacy_assets legacy USING(asset_id,study_id) WHERE NOT EXISTS(SELECT 1 FROM tre_governance.datafile_bindings b WHERE b.version_id=v.version_id) ORDER BY v.version_id LIMIT $1", &[&(batch as i64)])?.iter().map(|row|serde_json::from_str(&row.required_text(0,"Datafile origin")?).map_err(|_|invalid("invalid Datafile origin"))).collect()
}

impl GovernanceMaintenance {
    /// Only evidence required by this input closure, never unrelated completions.
    pub fn required_evidence(
        &self,
        inputs: &[Uuid],
        batch: u32,
    ) -> Result<Vec<Uuid>, ahri_tre_core::CoreError> {
        self.repository.with_session_connection(
            "required disclosure evidence",
            |db| required_evidence(db, inputs, batch),
            |db| required_evidence(db, inputs, batch),
        )
    }
}
pub fn required_evidence<E: MetadataExecutor>(
    db: &mut E,
    inputs: &[Uuid],
    batch: u32,
) -> Result<Vec<Uuid>, PgMetaError>
where
    E::Row: SchemaStatusRow,
{
    if inputs.is_empty() || inputs.len() > 128 || batch == 0 || batch > 1000 {
        return Err(invalid("invalid prerequisite batch"));
    }
    db.query_many("required disclosure evidence", "SELECT e.event_id::text FROM tre_governance.events e WHERE NOT e.ledger_acknowledged AND (e.event_id IN (SELECT c.evidence_id FROM tre_governance.classifications c JOIN public.asset_versions v USING(asset_id) WHERE v.version_id=ANY($1)) OR (e.event_kind IN ('study_policy_baseline','study_policy_change') AND e.study_id IN (SELECT a.study_id FROM public.assets a JOIN public.asset_versions v USING(asset_id) WHERE v.version_id=ANY($1))) OR (e.event_kind='version_ingest' AND e.evidence->>'version_id'=ANY(SELECT unnest($1::uuid[])::text))) ORDER BY e.occurred_at,e.event_id LIMIT $2", &[&inputs,&(batch as i64)])?.iter().map(|row| row.required_text(0,"event identity")?.parse().map_err(|_|invalid("invalid event identity"))).collect()
}