ahri_tre_pgmeta/

operation.rs

use crate::dataset_admission::ReceiptRow;
use crate::{MetadataExecutor, PgMetaError, PgMetadataRepository};
use ahri_tre_core::{CoreError, OperationRepository, StoredOperation};
use ahri_tre_observability::{FailureCategory, LifecycleSpan, Outcome, Phase};
use ahri_tre_protocol::operation::OperationStatus;
use uuid::Uuid;

impl PgMetadataRepository<'static> {
    /// Recheck current Study authority before another governed stage. The
    /// operation record is correlation evidence and never supplies the grant.
    pub fn operation_stage_authorized(&self, id: Uuid) -> Result<bool, CoreError> {
        self.with_session_connection(
            "authorize operation stage",
            |c| stage_authorized(c, id),
            |c| stage_authorized(c, id),
        )
    }

    /// A fresh user-authenticated lifecycle connection. The application exposes
    /// only inspection and cancellation; it shares no mutex with the producer.
    pub fn operation_control(
        connection: ahri_tre_libpq_oauth::LibpqOAuthConnection,
    ) -> Result<Self, CoreError> {
        Ok(Self::from_oauth_shared(std::sync::Arc::new(
            std::sync::Mutex::new(connection),
        )))
    }
}

impl OperationRepository for PgMetadataRepository<'_> {
    fn create_operation(
        &self,
        attempt_id: Uuid,
        record: &StoredOperation,
    ) -> Result<(), CoreError> {
        let summary = &record.detail.summary;
        if summary.status != OperationStatus::Pending
            || !summary.operation_kind.is_supported()
            || record.detail.events.items.len() != 1
            || record.detail.events.items[0].event_sequence != 1
            || record.result.is_some()
        {
            return Err(unavailable());
        }
        let id = operation_id(record)?;
        let body = serde_json::to_string(record).map_err(|_| unavailable())?;
        self.with_session_connection(
            "accept operation",
            |c| create(c, id, attempt_id, &body),
            |c| create(c, id, attempt_id, &body),
        )
        .map_err(|_| unavailable())
    }
    fn get_operation(&self, id: Uuid) -> Result<Option<StoredOperation>, CoreError> {
        self.with_session_connection("inspect operation", |c| read(c, id), |c| read(c, id))
            .map_err(|_| unavailable())
    }
    fn operation_history(
        &self,
        query: &ahri_tre_core::OperationHistoryQuery,
    ) -> Result<Vec<ahri_tre_core::OperationHistoryEntry>, CoreError> {
        self.with_session_connection(
            "browse operation history",
            |c| history(c, query),
            |c| history(c, query),
        )
        .map_err(|_| unavailable())
    }
    fn transition_operation(
        &self,
        expected: OperationStatus,
        sequence: u64,
        record: &StoredOperation,
    ) -> Result<(), CoreError> {
        if !(expected.can_transition_to(record.detail.summary.status)
            || (expected == record.detail.summary.status && !expected.is_terminal()))
            || record.detail.events.items.last().map(|e| e.event_sequence) != Some(sequence + 1)
        {
            return Err(unavailable());
        }
        let id = operation_id(record)?;
        let body = serde_json::to_string(record).map_err(|_| unavailable())?;
        let expected = serde_json::to_value(expected).map_err(|_| unavailable())?;
        let expected = expected.as_str().ok_or_else(unavailable)?;
        let sequence = i64::try_from(sequence).map_err(|_| unavailable())?;
        let changed = self
            .with_session_connection(
                "transition operation",
                |c| transition(c, id, expected, sequence, &body),
                |c| transition(c, id, expected, sequence, &body),
            )
            .map_err(|_| unavailable())?;
        if changed != 1 {
            return Err(CoreError::Conflict("Operation state changed".into()));
        }
        Ok(())
    }
}
fn stage_authorized<E: MetadataExecutor>(c: &mut E, id: Uuid) -> Result<bool, PgMetaError>
where
    E::Row: ReceiptRow,
{
    let row = c.query_optional("authorize operation stage", "SELECT public.tre_user_has_study_access((record#>>'{scope,study_id}')::uuid)::text FROM public.operations WHERE operation_id=$1 AND actor=CURRENT_USER", &[&id])?;
    Ok(row.is_some_and(|row| row.text(0).is_ok_and(|value| value == "true")))
}

fn operation_id(record: &StoredOperation) -> Result<Uuid, CoreError> {
    let reference = record.detail.summary.operation;
    if reference.kind != ahri_tre_protocol::refs::ObjectKind::Operation
        || reference.datastore_id.as_uuid() != record.scope.datastore_id
    {
        return Err(unavailable());
    }
    Ok(reference.id.as_uuid())
}

fn unavailable() -> CoreError {
    CoreError::Infrastructure("Operation persistence is unavailable".into())
}
fn create(
    c: &mut impl MetadataExecutor,
    id: Uuid,
    attempt: Uuid,
    body: &str,
) -> Result<(), PgMetaError> {
    let changed=c.execute_command(
        "accept operation",
        "INSERT INTO public.operations(operation_id, attempt_id, created_at, record) SELECT $1,a.attempt_id,($3::text::jsonb#>>'{detail,summary,timestamps,created_at}')::timestamptz,$3::text::jsonb FROM public.dataset_output_attempts a JOIN public.dataset_output_reservations r USING(attempt_id) WHERE a.attempt_id=$2 AND a.actor=CURRENT_USER AND NOT a.admitted AND a.study_id=($3::text::jsonb->'scope'->>'study_id')::uuid",
        &[&id, &attempt, &body],
    )?;
    if changed != 1 {
        return Err(PgMetaError::Decode {
            field: "operation",
            value: "<unavailable>".into(),
            message: "acceptance does not own its output".into(),
        });
    }
    Ok(())
}
fn read<E: MetadataExecutor>(c: &mut E, id: Uuid) -> Result<Option<StoredOperation>, PgMetaError>
where
    E::Row: ReceiptRow,
{
    c.query_optional(
        "inspect operation",
        "SELECT record::text FROM public.operations WHERE operation_id=$1 AND actor=CURRENT_USER",
        &[&id],
    )?
    .map(|r| decode_operation_record(r.text(0)?))
    .transpose()
}
/// Stored m0002 results predate canonical public lineage. Decode their private
/// record using the retained Datastore scope; never guess it from a caller.
fn decode_operation_record(text: &str) -> Result<StoredOperation, PgMetaError> {
    use ahri_tre_protocol::{
        PublicUuid,
        refs::{ObjectKind, ObjectRef},
    };
    let mut value: serde_json::Value = serde_json::from_str(text).map_err(|_| recovery_error())?;
    if value
        .pointer("/result/data/lineage/transformation/transformation_id")
        .is_some_and(serde_json::Value::is_i64)
    {
        let datastore: uuid::Uuid = serde_json::from_value(
            value
                .pointer("/scope/datastore_id")
                .cloned()
                .ok_or_else(recovery_error)?,
        )
        .map_err(|_| recovery_error())?;
        let lineage: ahri_tre_types::TransformationLineageRecord = serde_json::from_value(
            value
                .pointer("/result/data/lineage")
                .cloned()
                .ok_or_else(recovery_error)?,
        )
        .map_err(|_| recovery_error())?;
        let version_ref = |id: ahri_tre_types::VersionId| ObjectRef {
            datastore_id: PublicUuid::from_uuid(datastore),
            kind: ObjectKind::AssetVersion,
            id: PublicUuid::from_uuid(id.0),
        };
        let projected = ahri_tre_protocol::transformation::TransformationSummary {
            transformation: ahri_tre_protocol::transformation::TransformationDetail {
                transformation_id: ObjectRef {
                    datastore_id: PublicUuid::from_uuid(datastore),
                    kind: ObjectKind::Transformation,
                    id: ahri_tre_protocol::refs::encode_scoped_integer_ref(
                        "transformation",
                        lineage.transformation.transformation_id.0,
                    ),
                },
                transformation_type: lineage.transformation.transformation_type,
                description: lineage.transformation.description,
                date_created: lineage.transformation.date_created,
                created_by: lineage.transformation.created_by,
            },
            inputs: lineage
                .inputs
                .into_iter()
                .map(|i| version_ref(i.version_id))
                .collect(),
            outputs: lineage
                .outputs
                .into_iter()
                .map(|i| version_ref(i.version_id))
                .collect(),
        };
        value["result"]["data"]["lineage"] =
            serde_json::to_value(projected).map_err(|_| recovery_error())?;
    }
    serde_json::from_value(value).map_err(|_| recovery_error())
}

fn transition(
    c: &mut impl MetadataExecutor,
    id: Uuid,
    expected: &str,
    sequence: i64,
    body: &str,
) -> Result<u64, PgMetaError> {
    c.execute_command("transition operation", "UPDATE public.operations SET record=$4::text::jsonb, event_sequence=event_sequence+1 WHERE operation_id=$1 AND actor=CURRENT_USER AND record->'detail'->'summary'->>'status'=$2 AND event_sequence=$3 AND record->'scope'=$4::text::jsonb->'scope' AND record#>'{detail,summary,started_by_request_id}'=$4::text::jsonb#>'{detail,summary,started_by_request_id}' AND record#>'{detail,summary,timestamps,created_at}'=$4::text::jsonb#>'{detail,summary,timestamps,created_at}' AND record#>'{detail,events,items}'=($4::text::jsonb#>'{detail,events,items}') - $3::int AND record->'detail'->'summary'->>'status' NOT IN ('completed','failed','cancelled')", &[&id,&expected,&sequence,&body])
}

/// Operation recovery is restricted to stopped private attempts under the same
/// Dataset executor exclusion used by Lake cleanup. It can only fail unfinished
/// records; committed completion/result evidence is never rewritten.
impl crate::PgDatasetMaintenance {
    pub fn verify_operation_recovery(&self) -> Result<(), CoreError> {
        self.repository.with_session_connection(
            "verify operation recovery",
            verify_recovery,
            verify_recovery,
        )
    }

    pub fn reconcile_operations(
        &self,
        executor: std::sync::Arc<dyn ahri_tre_core::DatasetExecutorLease>,
        interrupt: impl Fn(&mut StoredOperation),
    ) -> Result<(), CoreError> {
        self.verify_operation_recovery()?;
        let candidates = self.repository.with_session_connection(
            "inspect operation recovery",
            recovery_candidates,
            recovery_candidates,
        )?;
        for (attempt, owner, operation_id) in candidates {
            if !executor.can_recover(owner, attempt) {
                continue;
            }
            let Ok(_exclusion) = executor.clone().begin_attempt(attempt) else {
                continue;
            };
            let recovery =
                LifecycleSpan::start_operation(Phase::Recovery, None, Some(operation_id));
            let result = self.repository.with_session_connection(
                "reconcile operation",
                |c| reconcile(c, attempt, owner, &interrupt),
                |c| reconcile(c, attempt, owner, &interrupt),
            );
            match &result {
                Ok(true) => {
                    recovery.finish(Outcome::Interrupted, Some(FailureCategory::Interrupted))
                }
                Ok(false) => recovery.finish(Outcome::Skipped, None),
                Err(_) => recovery.finish(Outcome::Unavailable, Some(FailureCategory::Metadata)),
            }
            result?;
        }
        self.prune_expired_operations()?;
        Ok(())
    }

    /// A bounded lifecycle pass. The only cascade is into expired key bindings;
    /// Dataset ownership, private attempts, governed Assets and provenance are
    /// independent tables and remain available to trusted reconciliation.
    pub fn prune_expired_operations(&self) -> Result<u64, CoreError> {
        self.verify_operation_recovery()?;
        self.repository.with_session_connection(
            "prune expired operations",
            prune_expired,
            prune_expired,
        )
    }
}

fn prune_expired(c: &mut impl MetadataExecutor) -> Result<u64, PgMetaError> {
    c.execute_command("prune expired operations", r#"
        WITH expired AS (
            SELECT o.operation_id FROM public.operations o
            WHERE o.record#>>'{detail,summary,status}' IN ('completed','failed','cancelled')
              AND public.operation_now() >= (o.record#>>'{detail,summary,retention,operation_expires_at}')::timestamptz
              AND NOT EXISTS (
                  SELECT 1 FROM public.operation_idempotency k
                  WHERE k.operation_id=o.operation_id
                    AND public.operation_now() < k.accepted_at + INTERVAL '24 hours'
              )
            ORDER BY o.created_at, o.operation_id
            LIMIT 100 FOR UPDATE OF o SKIP LOCKED
        )
        DELETE FROM public.operations o USING expired e WHERE o.operation_id=e.operation_id
    "#, &[])
}

fn verify_recovery<E: MetadataExecutor>(c: &mut E) -> Result<(), PgMetaError>
where
    E::Row: ReceiptRow,
{
    let row = c.query_optional("verify operation recovery", "SELECT (count(*)=2 AND bool_and(r.rolsuper OR (has_table_privilege(c.oid, 'SELECT') AND (c.relname <> 'operations' OR (has_table_privilege(c.oid, 'UPDATE') AND has_table_privilege(c.oid, 'DELETE'))) AND (r.rolbypassrls OR (NOT c.relforcerowsecurity AND pg_has_role(c.relowner, 'USAGE'))))))::text FROM pg_class c CROSS JOIN pg_roles r WHERE r.rolname = CURRENT_USER AND c.oid IN (to_regclass('public.operations'), to_regclass('public.operation_idempotency'))", &[])?;
    if row.is_some_and(|r| r.text(0).is_ok_and(|v| v == "true")) {
        Ok(())
    } else {
        Err(recovery_error())
    }
}

fn recovery_candidates<E: MetadataExecutor>(
    c: &mut E,
) -> Result<Vec<(Uuid, ahri_tre_core::DatasetExecutorIdentity, Uuid)>, PgMetaError>
where
    E::Row: ReceiptRow,
{
    c.query_many("inspect operation recovery", "SELECT a.attempt_id::text, a.coordinator_id::text, a.generation_id::text, o.operation_id::text FROM public.dataset_output_attempts a JOIN public.operations o USING(attempt_id) WHERE o.record#>>'{detail,summary,status}' NOT IN ('completed','failed','cancelled') OR (NOT a.admitted AND NOT EXISTS (SELECT 1 FROM public.dataset_output_reservations r WHERE r.attempt_id=a.attempt_id)) ORDER BY a.created_at, a.attempt_id", &[])?.into_iter().map(|r| Ok((
        r.text(0)?.parse().map_err(|_| recovery_error())?,
        ahri_tre_core::DatasetExecutorIdentity {
            coordinator_id: r.text(1)?.parse().map_err(|_| recovery_error())?,
            generation_id: r.text(2)?.parse().map_err(|_| recovery_error())?,
        },
        r.text(3)?.parse().map_err(|_| recovery_error())?
    ))).collect()
}

fn reconcile<E: MetadataExecutor>(
    c: &mut E,
    attempt: Uuid,
    owner: ahri_tre_core::DatasetExecutorIdentity,
    interrupt: &impl Fn(&mut StoredOperation),
) -> Result<bool, PgMetaError>
where
    E::Row: ReceiptRow,
{
    c.with_transaction("reconcile stopped operation", |t| {
        // Order after any in-flight final admission before inspecting its receipt.
        t.query_optional("lock operation reservation", "SELECT attempt_id FROM public.dataset_output_reservations WHERE attempt_id=$1 FOR UPDATE", &[&attempt])?;
        let row = t.query_optional("inspect stopped operation", "SELECT o.record::text FROM public.operations o JOIN public.dataset_output_attempts a USING(attempt_id) WHERE a.attempt_id=$1 AND a.coordinator_id=$2 AND a.generation_id=$3 AND NOT a.admitted AND NOT EXISTS (SELECT 1 FROM public.dataset_admission_receipts d WHERE d.version_id=a.version_id AND d.admitted) FOR UPDATE OF o, a", &[&attempt, &owner.coordinator_id, &owner.generation_id])?;
        let Some(row) = row else { return Ok(false); };
        let mut record: StoredOperation = decode_operation_record(row.text(0)?)?;
        if record.detail.summary.status.is_terminal() {
            t.execute_command("remove reconciled clean attempt", "DELETE FROM public.dataset_output_attempts a WHERE a.attempt_id=$1 AND NOT EXISTS (SELECT 1 FROM public.dataset_output_reservations r WHERE r.attempt_id=a.attempt_id)", &[&attempt])?;
            return Ok(false);
        }
        let before = record.clone();
        interrupt(&mut record);
        // This capability cannot change identity, result, prior events, or claim success.
        let sequence = before.detail.events.items.len();
        if record.scope != before.scope || record.result != before.result
            || record.detail.summary.status != OperationStatus::Failed
            || record.detail.summary.operation != before.detail.summary.operation
            || record.detail.summary.started_by_request_id != before.detail.summary.started_by_request_id
            || record.detail.summary.timestamps.created_at != before.detail.summary.timestamps.created_at
            || record.detail.events.items.len() != sequence + 1
            || record.detail.events.items[..sequence] != before.detail.events.items
            || record.detail.events.items[sequence].event_sequence != sequence as u64 + 1 {
            return Err(recovery_error());
        }
        let body = serde_json::to_string(&record).map_err(|_| recovery_error())?;
        t.execute_command("interrupt stopped operation", "UPDATE public.operations SET record=$2::text::jsonb, event_sequence=event_sequence+1 WHERE attempt_id=$1", &[&attempt, &body])?;
        // A normal cleanup may have released the reservation before its user
        // connection lost the terminal response. Keep evidence until this point.
        t.execute_command("remove reconciled clean attempt", "DELETE FROM public.dataset_output_attempts a WHERE a.attempt_id=$1 AND NOT EXISTS (SELECT 1 FROM public.dataset_output_reservations r WHERE r.attempt_id=a.attempt_id)", &[&attempt])?;
        Ok(true)
    })
}
fn recovery_error() -> PgMetaError {
    PgMetaError::Decode {
        field: "operation",
        value: "<unavailable>".into(),
        message: "Operation recovery is unavailable".into(),
    }
}

/// Serializes only admission for one actor-scoped key on a dedicated user
/// connection. Dropping it after committed acceptance lets retries inspect the
/// operation without waiting for Lake execution or occupying a worker permit.
pub struct PgOperationKey {
    repository: PgMetadataRepository<'static>,
    scope_digest: String,
    intent_digest: String,
}

impl PgMetadataRepository<'static> {
    pub fn lock_operation_key(
        &self,
        scope_digest: String,
        intent_digest: String,
    ) -> Result<PgOperationKey, CoreError> {
        self.with_session_connection(
            "serialize operation admission",
            |c| key_lock(c, &scope_digest, true),
            |c| key_lock(c, &scope_digest, true),
        )?;
        Ok(PgOperationKey {
            repository: self.clone(),
            scope_digest,
            intent_digest,
        })
    }
}

impl PgOperationKey {
    /// An absent deadline protects active operations indefinitely. Terminal
    /// bindings last through both the 24-hour minimum and operation retention.
    pub fn retained(&self) -> Result<Option<(StoredOperation, bool)>, CoreError> {
        self.repository.with_session_connection(
            "recover idempotent operation",
            |c| retained_key(c, &self.scope_digest, &self.intent_digest),
            |c| retained_key(c, &self.scope_digest, &self.intent_digest),
        )
    }

    /// Must join the reservation/operation transaction. An expired binding can
    /// be replaced only after the new request has acquired output ownership.
    pub fn bind(
        &self,
        transaction: &PgMetadataRepository<'_>,
        operation: Uuid,
    ) -> Result<(), CoreError> {
        transaction.with_session_connection(
            "bind operation key",
            |c| bind_key(c, &self.scope_digest, &self.intent_digest, operation),
            |c| bind_key(c, &self.scope_digest, &self.intent_digest, operation),
        )
    }
}

impl Drop for PgOperationKey {
    fn drop(&mut self) {
        let _ = self.repository.with_session_connection(
            "release operation admission",
            |c| key_lock(c, &self.scope_digest, false),
            |c| key_lock(c, &self.scope_digest, false),
        );
    }
}

fn key_lock<E: MetadataExecutor>(
    c: &mut E,
    digest: &str,
    acquire: bool,
) -> Result<(), PgMetaError> {
    c.query_optional(
        "serialize operation admission",
        if acquire {
            "SELECT pg_advisory_lock(hashtextextended(CURRENT_USER || $1, 606))"
        } else {
            "SELECT pg_advisory_unlock(hashtextextended(CURRENT_USER || $1, 606))"
        },
        &[&digest],
    )?;
    Ok(())
}

fn retained_key<E: MetadataExecutor>(
    c: &mut E,
    scope: &str,
    intent: &str,
) -> Result<Option<(StoredOperation, bool)>, PgMetaError>
where
    E::Row: ReceiptRow,
{
    c.query_optional("recover idempotent operation", "SELECT o.record::text, (k.intent_digest=$2)::text FROM public.operation_idempotency k JOIN public.operations o USING(operation_id) WHERE k.actor=CURRENT_USER AND o.actor=CURRENT_USER AND k.scope_digest=$1 AND (o.record#>>'{detail,summary,status}' NOT IN ('completed','failed','cancelled') OR o.record#>>'{detail,summary,retention,operation_expires_at}' IS NULL OR public.operation_now() < GREATEST(k.accepted_at + INTERVAL '24 hours', (o.record#>>'{detail,summary,retention,operation_expires_at}')::timestamptz))", &[&scope, &intent])?.map(|row| {
        let record = decode_operation_record(row.text(0)?)?;
        Ok((record, row.text(1)? == "true"))
    }).transpose()
}

fn bind_key(
    c: &mut impl MetadataExecutor,
    scope: &str,
    intent: &str,
    operation: Uuid,
) -> Result<(), PgMetaError> {
    let changed = c.execute_command("bind operation key", "INSERT INTO public.operation_idempotency(scope_digest,intent_digest,operation_id) VALUES($1,$2,$3) ON CONFLICT(actor,scope_digest) DO UPDATE SET intent_digest=EXCLUDED.intent_digest, operation_id=EXCLUDED.operation_id, accepted_at=public.operation_now() WHERE EXISTS (SELECT 1 FROM public.operations o WHERE o.operation_id=operation_idempotency.operation_id AND o.record#>>'{detail,summary,status}' IN ('completed','failed','cancelled') AND o.record#>>'{detail,summary,retention,operation_expires_at}' IS NOT NULL AND public.operation_now() >= GREATEST(operation_idempotency.accepted_at + INTERVAL '24 hours', (o.record#>>'{detail,summary,retention,operation_expires_at}')::timestamptz))", &[&scope, &intent, &operation])?;
    if changed != 1 {
        return Err(recovery_error());
    }
    Ok(())
}

impl PgMetadataRepository<'_> {
    /// Evaluated inside output acceptance after ownership has been acquired.
    pub fn operation_precondition_satisfied(
        &self,
        scope: &ahri_tre_core::OperationResourceScope,
        precondition: &ahri_tre_protocol::mutation::Precondition,
    ) -> Result<bool, CoreError> {
        self.with_session_connection(
            "validate operation precondition",
            |c| check_precondition(c, scope, precondition),
            |c| check_precondition(c, scope, precondition),
        )
    }
}

fn check_precondition<E: MetadataExecutor>(
    c: &mut E,
    scope: &ahri_tre_core::OperationResourceScope,
    precondition: &ahri_tre_protocol::mutation::Precondition,
) -> Result<bool, PgMetaError>
where
    E::Row: ReceiptRow,
{
    use ahri_tre_protocol::mutation::Precondition;
    let exists = |c: &mut E| -> Result<bool, PgMetaError> {
        Ok(c.query_optional("inspect Dataset precondition", "SELECT asset_id FROM public.assets WHERE study_id=$1 AND name=$2 AND asset_type='dataset'", &[&scope.study_id.0, &scope.dataset_name])?.is_some())
    };
    match precondition {
        Precondition::Exists { target } if target == "dataset" => exists(c),
        Precondition::DoesNotExist { target } if target == "dataset" => Ok(!exists(c)?),
        Precondition::Exists { target } if target == "source" => Ok(c.query_optional("inspect source precondition", "SELECT datafile_id FROM public.datafiles WHERE datafile_id=$1", &[&scope.source_version_id.0])?.is_some()),
        Precondition::ResourceVersion { target, version } if target == "source" => Ok(c.query_optional("inspect source version precondition", "SELECT version_id FROM public.asset_versions WHERE version_id=$1 AND major::text || '.' || minor::text || '.' || patch::text=$2", &[&scope.source_version_id.0, &version.trim_start_matches('v')])?.is_some()),
        _ => Err(recovery_error()),
    }
}

fn history<E: MetadataExecutor>(
    c: &mut E,
    query: &ahri_tre_core::OperationHistoryQuery,
) -> Result<Vec<ahri_tre_core::OperationHistoryEntry>, PgMetaError>
where
    E::Row: ReceiptRow,
{
    let lower = query.created_after.map(|value| value.to_rfc3339());
    let before = query.created_before.map(|value| value.to_rfc3339());
    let upper = query.upper_bound.to_rfc3339();
    let after_time = query
        .after
        .as_ref()
        .map(|value| value.created_at.to_rfc3339());
    let after_id = query.after.as_ref().map(|value| value.id);
    let limit = i64::from(query.limit.clamp(1, 500));
    c.query_many("browse operation history", r#"
        SELECT jsonb_build_object('scope', record->'scope', 'summary', record#>'{detail,summary}', 'result_dataset_id', record#>'{result,data,dataset,dataset_id}')::text
        FROM public.operations
        WHERE actor=CURRENT_USER::text COLLATE "default"
          AND ($1::text IS NULL OR created_at > $1::text::timestamptz)
          AND ($2::text IS NULL OR created_at < $2::text::timestamptz)
          AND created_at <= $3::text::timestamptz
          AND ($4::text IS NULL OR (created_at, operation_id) < ($4::text::timestamptz, $5::uuid))
        ORDER BY created_at DESC, operation_id DESC
        LIMIT $6
    "#, &[&lower, &before, &upper, &after_time, &after_id, &limit])?
    .into_iter().map(|row| serde_json::from_str(row.text(0)?).map_err(|_| recovery_error())).collect()
}

impl PgMetadataRepository<'_> {
    /// One authoritative lifecycle instant, also used by key and cleanup queries.
    pub fn operation_now(&self) -> Result<chrono::DateTime<chrono::Utc>, CoreError> {
        self.with_session_connection("read operation clock", operation_now, operation_now)
    }
}
fn operation_now<E: MetadataExecutor>(
    c: &mut E,
) -> Result<chrono::DateTime<chrono::Utc>, PgMetaError>
where
    E::Row: ReceiptRow,
{
    let row = c
        .query_optional(
            "read operation clock",
            r#"SELECT to_char(public.operation_now() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')"#,
            &[],
        )?
        .ok_or_else(recovery_error)?;
    row.text(0)?.parse().map_err(|_| recovery_error())
}