v0.3.3-pre.004

crates/ksp-store-postgres-lib/src/error.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/error.rs
-// version: 4
+// version: 5
 
 /// Stable KSP error code reserved for PostgreSQL retention transitions that require unsupported physical compaction.
 pub const ERROR_CODE_POSTGRES_RETENTION_COMPACTION_UNSUPPORTED: ksp_store_api::ErrorCode =
@@ -17,16 +17,22 @@ pub enum PostgresBackendErrorKind {
     PoolTimeout,
     /// A lightweight PostgreSQL health/readiness probe failed without exposing server text or SQL.
     HealthFailed,
+    /// PostgreSQL returned stored RAW data that cannot be represented by the stable Store API contract.
+    DataInvalid,
     /// PostgreSQL migration/bootstrap execution failed without exposing server text or SQL.
     MigrationFailed,
     /// Applied PostgreSQL migration history diverges from the embedded immutable KSP history.
     MigrationMismatch,
+    /// A PostgreSQL RAW read statement failed without exposing server text, SQL or bind values.
+    ReadFailed,
     /// The database schema history contains a migration newer than this runtime understands.
     SchemaNewer,
     /// Explicit backend shutdown did not drain inside the supplied deadline.
     ShutdownTimeout,
     /// Verified TLS configuration or negotiation could not be established.
     TlsFailed,
+    /// A network-scoped RAW operation targeted a network different from the backend binding.
+    WrongNetwork,
 }
 
 /// Redacted PostgreSQL backend error carrying only a safe classification and static lifecycle phase.

crates/ksp-store-postgres-lib/src/lib.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/lib.rs
-// version: 8
+// version: 9
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -11,8 +11,9 @@
 //! pool, explicit Rustls TLS policy, private KSP migration/bootstrap engine and
 //! safe lightweight health/readiness probe. `0.3.3-pre.003-fix.001` splits
 //! migrations into versioned physical resources and verifies the effective
-//! PostgreSQL schema contract before readiness; business capability
-//! implementations remain deferred to later prereleases.
+//! PostgreSQL schema contract before readiness. `0.3.3-pre.004` adds exact
+//! backend-private RAW transaction/observation/retention read mapping without
+//! exposing PostgreSQL rows or SQL through the public bridge.
 //!
 //! This crate depends on `ksp-store-api` and never on `ksp-store-lib`. The
 //! common facade consumes only this crate's narrow backend bridge and never
@@ -22,6 +23,7 @@ mod constants;
 mod error;
 mod health;
 mod migration;
+mod raw_transaction;
 mod runtime;
 mod schema;
 
@@ -50,6 +52,14 @@ pub(crate) use self::health::probe_health;
 pub(crate) use self::migration::bootstrap;
 /// Current embedded migration version consumed by the private health probe.
 pub(crate) use self::migration::current_migration_version;
+/// Private RAW transaction reader consumed by the physical backend runtime.
+pub(crate) use self::raw_transaction::get_raw_transaction;
+/// Private RAW transaction observation reader consumed by the physical backend runtime.
+pub(crate) use self::raw_transaction::get_raw_transaction_observation;
+/// Private RAW transaction retention-state reader consumed by the physical backend runtime.
+pub(crate) use self::raw_transaction::get_raw_transaction_retention_state;
+/// Private RAW transaction tombstone reader consumed by the physical backend runtime.
+pub(crate) use self::raw_transaction::get_raw_transaction_tombstone;
 /// Private Deadpool error mapper shared with the health probe.
 pub(crate) use self::runtime::map_pool_error;
 /// Private Deadpool status projector shared with the health probe.

crates/ksp-store-postgres-lib/src/raw_transaction.rs

@@ -0,0 +1,660 @@
+// file: crates/ksp-store-postgres-lib/src/raw_transaction.rs
+// version: 1
+
+const GET_OBSERVATION_SQL: &str = "SELECT observation_key, transaction_signature, provider, protocol, acquisition_method, origin, received_at_unix_millis, capture_session_id, commitment, endpoint_id, filter_id, observed_at_unix_millis, source_payload_hash, source_payload_size_bytes FROM ksp_raw_transaction_observations WHERE observation_key = $1";
+const GET_RETENTION_SQL: &str = "SELECT retention_state FROM ksp_raw_transactions WHERE signature = $1";
+const GET_TOMBSTONE_SQL: &str = "SELECT signature, slot::text AS slot_text, block_time_unix_millis, format_id, format_version, content_hash, retention_state FROM ksp_raw_transactions WHERE signature = $1";
+const GET_TRANSACTION_SQL: &str = "SELECT transaction_row.signature, transaction_row.slot::text AS slot_text, transaction_row.block_time_unix_millis, transaction_row.format_id, transaction_row.format_version, transaction_row.content_hash, transaction_row.payload, transaction_row.retention_state, archive_row.payload AS archive_payload FROM ksp_raw_transactions AS transaction_row LEFT JOIN ksp_raw_transaction_archive_payloads AS archive_row ON archive_row.signature = transaction_row.signature WHERE transaction_row.signature = $1";
+
+struct RawObservationDbRow {
+    acquisition_method: std::string::String,
+    capture_session_id: std::option::Option<std::string::String>,
+    commitment: std::option::Option<std::string::String>,
+    endpoint_id: std::option::Option<std::string::String>,
+    filter_id: std::option::Option<std::string::String>,
+    observation_key: std::vec::Vec<u8>,
+    observed_at_unix_millis: std::option::Option<i64>,
+    origin: std::string::String,
+    protocol: std::string::String,
+    provider: std::string::String,
+    received_at_unix_millis: i64,
+    source_payload_hash: std::option::Option<std::vec::Vec<u8>>,
+    source_payload_size_bytes: std::option::Option<i64>,
+    transaction_signature: std::vec::Vec<u8>,
+}
+
+struct RawTransactionDbRow {
+    archive_payload: std::option::Option<std::vec::Vec<u8>>,
+    block_time_unix_millis: std::option::Option<i64>,
+    content_hash: std::vec::Vec<u8>,
+    format_id: std::string::String,
+    format_version: i64,
+    payload: std::option::Option<std::vec::Vec<u8>>,
+    retention_state: std::string::String,
+    signature: std::vec::Vec<u8>,
+    slot_text: std::string::String,
+}
+
+struct RawTombstoneDbRow {
+    block_time_unix_millis: std::option::Option<i64>,
+    content_hash: std::vec::Vec<u8>,
+    format_id: std::string::String,
+    format_version: i64,
+    retention_state: std::string::String,
+    signature: std::vec::Vec<u8>,
+    slot_text: std::string::String,
+}
+
+/// Reads one canonical RAW transaction from the physical PostgreSQL backend.
+pub(crate) async fn get_raw_transaction(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    reference: &ksp_store_api::RawTransactionReference,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawTransaction>, crate::PostgresBackendError> {
+    let network_result = ensure_network(network, reference, "raw_transaction_network");
+    if let std::result::Result::Err(error) = network_result {
+        return std::result::Result::Err(error);
+    }
+    let client_result = pool.get().await;
+    let client = match client_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(crate::map_pool_error(error)),
+    };
+    let signature = reference.signature();
+    let signature_bytes: &[u8] = signature.as_bytes();
+    let rows_result = client.query(GET_TRANSACTION_SQL, &[&signature_bytes]).await;
+    let rows = match rows_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ReadFailed, "raw_transaction_query"));
+        },
+    };
+    if rows.is_empty() {
+        return std::result::Result::Ok(std::option::Option::None);
+    }
+    if rows.len() != 1 {
+        return std::result::Result::Err(data_invalid("raw_transaction_cardinality"));
+    }
+    let row = match rows.first() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(data_invalid("raw_transaction_cardinality")),
+    };
+    let physical = match raw_transaction_db_row(row) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    return decode_raw_transaction_row(network, physical);
+}
+
+/// Reads one RAW transaction observation from the physical PostgreSQL backend.
+pub(crate) async fn get_raw_transaction_observation(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    observation_key: &ksp_store_api::RawObservationKey,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawTransactionObservation>, crate::PostgresBackendError> {
+    let client_result = pool.get().await;
+    let client = match client_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(crate::map_pool_error(error)),
+    };
+    let observation_key_bytes: &[u8] = observation_key.as_bytes();
+    let rows_result = client.query(GET_OBSERVATION_SQL, &[&observation_key_bytes]).await;
+    let rows = match rows_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ReadFailed, "raw_observation_query"));
+        },
+    };
+    if rows.is_empty() {
+        return std::result::Result::Ok(std::option::Option::None);
+    }
+    if rows.len() != 1 {
+        return std::result::Result::Err(data_invalid("raw_observation_cardinality"));
+    }
+    let row = match rows.first() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(data_invalid("raw_observation_cardinality")),
+    };
+    let physical = match raw_observation_db_row(row) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let decoded = match decode_raw_observation_row(network, physical) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    if decoded.observation_key() != *observation_key {
+        return std::result::Result::Err(data_invalid("raw_observation_identity"));
+    }
+    return std::result::Result::Ok(std::option::Option::Some(decoded));
+}
+
+/// Reads one RAW transaction retention state from the physical PostgreSQL backend.
+pub(crate) async fn get_raw_transaction_retention_state(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    reference: &ksp_store_api::RawTransactionReference,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawRetentionState>, crate::PostgresBackendError> {
+    let network_result = ensure_network(network, reference, "raw_retention_network");
+    if let std::result::Result::Err(error) = network_result {
+        return std::result::Result::Err(error);
+    }
+    let client_result = pool.get().await;
+    let client = match client_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(crate::map_pool_error(error)),
+    };
+    let signature = reference.signature();
+    let signature_bytes: &[u8] = signature.as_bytes();
+    let rows_result = client.query(GET_RETENTION_SQL, &[&signature_bytes]).await;
+    let rows = match rows_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ReadFailed, "raw_retention_query"));
+        },
+    };
+    if rows.is_empty() {
+        return std::result::Result::Ok(std::option::Option::None);
+    }
+    if rows.len() != 1 {
+        return std::result::Result::Err(data_invalid("raw_retention_cardinality"));
+    }
+    let row = match rows.first() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(data_invalid("raw_retention_cardinality")),
+    };
+    let state_result = row.try_get::<_, std::string::String>("retention_state");
+    let state = match state_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_retention_decode")),
+    };
+    let decoded = match decode_retention_state(state.as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    return std::result::Result::Ok(std::option::Option::Some(decoded));
+}
+
+/// Reads one minimal RAW transaction tombstone from the physical PostgreSQL backend.
+pub(crate) async fn get_raw_transaction_tombstone(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    reference: &ksp_store_api::RawTransactionReference,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawTransactionTombstone>, crate::PostgresBackendError> {
+    let network_result = ensure_network(network, reference, "raw_tombstone_network");
+    if let std::result::Result::Err(error) = network_result {
+        return std::result::Result::Err(error);
+    }
+    let client_result = pool.get().await;
+    let client = match client_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(crate::map_pool_error(error)),
+    };
+    let signature = reference.signature();
+    let signature_bytes: &[u8] = signature.as_bytes();
+    let rows_result = client.query(GET_TOMBSTONE_SQL, &[&signature_bytes]).await;
+    let rows = match rows_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ReadFailed, "raw_tombstone_query"));
+        },
+    };
+    if rows.is_empty() {
+        return std::result::Result::Ok(std::option::Option::None);
+    }
+    if rows.len() != 1 {
+        return std::result::Result::Err(data_invalid("raw_tombstone_cardinality"));
+    }
+    let row = match rows.first() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(data_invalid("raw_tombstone_cardinality")),
+    };
+    let physical = match raw_tombstone_db_row(row) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let decoded = match decode_raw_tombstone_row(network, physical) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let tombstone = match decoded {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
+    };
+    if tombstone.reference() != reference {
+        return std::result::Result::Err(data_invalid("raw_tombstone_identity"));
+    }
+    return std::result::Result::Ok(std::option::Option::Some(tombstone));
+}
+
+fn raw_transaction_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawTransactionDbRow, crate::PostgresBackendError> {
+    let signature = match row.try_get::<_, std::vec::Vec<u8>>("signature") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_decode")),
+    };
+    let slot_text = match row.try_get::<_, std::string::String>("slot_text") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_decode")),
+    };
+    let block_time_unix_millis = match row.try_get::<_, std::option::Option<i64>>("block_time_unix_millis") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_decode")),
+    };
+    let format_id = match row.try_get::<_, std::string::String>("format_id") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_decode")),
+    };
+    let format_version = match row.try_get::<_, i64>("format_version") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_decode")),
+    };
+    let content_hash = match row.try_get::<_, std::vec::Vec<u8>>("content_hash") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_decode")),
+    };
+    let payload = match row.try_get::<_, std::option::Option<std::vec::Vec<u8>>>("payload") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_decode")),
+    };
+    let retention_state = match row.try_get::<_, std::string::String>("retention_state") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_decode")),
+    };
+    let archive_payload = match row.try_get::<_, std::option::Option<std::vec::Vec<u8>>>("archive_payload") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_decode")),
+    };
+    return std::result::Result::Ok(RawTransactionDbRow {
+        archive_payload,
+        block_time_unix_millis,
+        content_hash,
+        format_id,
+        format_version,
+        payload,
+        retention_state,
+        signature,
+        slot_text,
+    });
+}
+
+fn raw_observation_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawObservationDbRow, crate::PostgresBackendError> {
+    macro_rules! required {
+        ($name:literal, $ty:ty) => {
+            match row.try_get::<_, $ty>($name) {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_observation_decode")),
+            }
+        };
+    }
+    return std::result::Result::Ok(RawObservationDbRow {
+        acquisition_method: required!("acquisition_method", std::string::String),
+        capture_session_id: required!("capture_session_id", std::option::Option<std::string::String>),
+        commitment: required!("commitment", std::option::Option<std::string::String>),
+        endpoint_id: required!("endpoint_id", std::option::Option<std::string::String>),
+        filter_id: required!("filter_id", std::option::Option<std::string::String>),
+        observation_key: required!("observation_key", std::vec::Vec<u8>),
+        observed_at_unix_millis: required!("observed_at_unix_millis", std::option::Option<i64>),
+        origin: required!("origin", std::string::String),
+        protocol: required!("protocol", std::string::String),
+        provider: required!("provider", std::string::String),
+        received_at_unix_millis: required!("received_at_unix_millis", i64),
+        source_payload_hash: required!("source_payload_hash", std::option::Option<std::vec::Vec<u8>>),
+        source_payload_size_bytes: required!("source_payload_size_bytes", std::option::Option<i64>),
+        transaction_signature: required!("transaction_signature", std::vec::Vec<u8>),
+    });
+}
+
+fn raw_tombstone_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawTombstoneDbRow, crate::PostgresBackendError> {
+    let signature = match row.try_get::<_, std::vec::Vec<u8>>("signature") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_tombstone_decode")),
+    };
+    let slot_text = match row.try_get::<_, std::string::String>("slot_text") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_tombstone_decode")),
+    };
+    let block_time_unix_millis = match row.try_get::<_, std::option::Option<i64>>("block_time_unix_millis") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_tombstone_decode")),
+    };
+    let format_id = match row.try_get::<_, std::string::String>("format_id") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_tombstone_decode")),
+    };
+    let format_version = match row.try_get::<_, i64>("format_version") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_tombstone_decode")),
+    };
+    let content_hash = match row.try_get::<_, std::vec::Vec<u8>>("content_hash") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_tombstone_decode")),
+    };
+    let retention_state = match row.try_get::<_, std::string::String>("retention_state") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_tombstone_decode")),
+    };
+    return std::result::Result::Ok(RawTombstoneDbRow {
+        block_time_unix_millis,
+        content_hash,
+        format_id,
+        format_version,
+        retention_state,
+        signature,
+        slot_text,
+    });
+}
+
+fn decode_raw_transaction_row(
+    network: &ksp_store_api::RawNetworkId,
+    row: RawTransactionDbRow,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawTransaction>, crate::PostgresBackendError> {
+    let signature = match fixed_bytes::<64>(row.signature) {
+        std::result::Result::Ok(value) => ksp_store_api::RawTransactionSignature::new(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let slot = match decode_u64_decimal(row.slot_text.as_str(), "raw_transaction_slot") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let block_time = match decode_optional_timestamp(row.block_time_unix_millis, "raw_transaction_block_time") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let format_id = match decode_format_id(row.format_id) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let format_version = match decode_u32_i64(row.format_version, "raw_transaction_format_version") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let content_hash = match fixed_bytes::<32>(row.content_hash) {
+        std::result::Result::Ok(value) => ksp_store_api::RawContentHash::new(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let retention_state = match decode_retention_state(row.retention_state.as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let payload_bytes = match retention_state {
+        ksp_store_api::RawRetentionState::Full => {
+            if row.archive_payload.is_some() {
+                return std::result::Result::Err(data_invalid("raw_transaction_full_archive"));
+            }
+            match row.payload {
+                std::option::Option::Some(value) => value,
+                std::option::Option::None => return std::result::Result::Err(data_invalid("raw_transaction_full_payload")),
+            }
+        },
+        ksp_store_api::RawRetentionState::Archived => {
+            if row.payload.is_some() {
+                return std::result::Result::Err(data_invalid("raw_transaction_archived_hot_payload"));
+            }
+            match row.archive_payload {
+                std::option::Option::Some(value) => value,
+                std::option::Option::None => return std::result::Result::Err(data_invalid("raw_transaction_archive_payload")),
+            }
+        },
+        ksp_store_api::RawRetentionState::Purged => {
+            if row.payload.is_some() || row.archive_payload.is_some() || block_time.is_some() {
+                return std::result::Result::Err(data_invalid("raw_transaction_purged_shape"));
+            }
+            return std::result::Result::Ok(std::option::Option::None);
+        },
+        _ => return std::result::Result::Err(data_invalid("raw_transaction_retention_state")),
+    };
+    let payload = match ksp_store_api::RawPayload::try_new(format_id, format_version, payload_bytes.into_boxed_slice(), content_hash) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_payload")),
+    };
+    let reference = ksp_store_api::RawTransactionReference::new(network.clone(), signature);
+    return std::result::Result::Ok(std::option::Option::Some(ksp_store_api::RawTransaction::new(reference, slot, block_time, payload)));
+}
+
+fn decode_raw_observation_row(
+    network: &ksp_store_api::RawNetworkId,
+    row: RawObservationDbRow,
+) -> std::result::Result<ksp_store_api::RawTransactionObservation, crate::PostgresBackendError> {
+    let observation_key = match fixed_bytes::<32>(row.observation_key) {
+        std::result::Result::Ok(value) => ksp_store_api::RawObservationKey::new(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let signature = match fixed_bytes::<64>(row.transaction_signature) {
+        std::result::Result::Ok(value) => ksp_store_api::RawTransactionSignature::new(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let provider = match decode_provenance_code(row.provider) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let protocol = match decode_provenance_code(row.protocol) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let acquisition_method = match decode_provenance_code(row.acquisition_method) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let origin = match decode_origin(row.origin.as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let received_at = match decode_timestamp_i64(row.received_at_unix_millis, "raw_observation_received_at") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let mut provenance = ksp_store_api::RawAcquisitionProvenance::new(provider, protocol, acquisition_method, origin, received_at);
+    provenance = match row.capture_session_id {
+        std::option::Option::Some(value) => match decode_provenance_code(value) {
+            std::result::Result::Ok(code) => provenance.with_capture_session_id(code),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.commitment {
+        std::option::Option::Some(value) => match decode_provenance_code(value) {
+            std::result::Result::Ok(code) => provenance.with_commitment(code),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.endpoint_id {
+        std::option::Option::Some(value) => match decode_provenance_code(value) {
+            std::result::Result::Ok(code) => provenance.with_endpoint_id(code),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.filter_id {
+        std::option::Option::Some(value) => match decode_provenance_code(value) {
+            std::result::Result::Ok(code) => provenance.with_filter_id(code),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.observed_at_unix_millis {
+        std::option::Option::Some(value) => {
+            let timestamp = match decode_timestamp_i64(value, "raw_observation_observed_at") {
+                std::result::Result::Ok(decoded) => decoded,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            match provenance.try_with_observed_at(timestamp) {
+                std::result::Result::Ok(updated) => updated,
+                std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_observation_time_order")),
+            }
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.source_payload_hash {
+        std::option::Option::Some(value) => {
+            let hash = match fixed_bytes::<32>(value) {
+                std::result::Result::Ok(decoded) => ksp_store_api::RawContentHash::new(decoded),
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            provenance.with_source_payload_hash(hash)
+        },
+        std::option::Option::None => provenance,
+    };
+    provenance = match row.source_payload_size_bytes {
+        std::option::Option::Some(value) => {
+            let size = match decode_u64_i64(value, "raw_observation_source_size") {
+                std::result::Result::Ok(decoded) => decoded,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            match provenance.try_with_source_payload_size_bytes(size) {
+                std::result::Result::Ok(updated) => updated,
+                std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_observation_source_size")),
+            }
+        },
+        std::option::Option::None => provenance,
+    };
+    let transaction = ksp_store_api::RawTransactionReference::new(network.clone(), signature);
+    return std::result::Result::Ok(ksp_store_api::RawTransactionObservation::new(observation_key, transaction, provenance));
+}
+
+fn decode_raw_tombstone_row(
+    network: &ksp_store_api::RawNetworkId,
+    row: RawTombstoneDbRow,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawTransactionTombstone>, crate::PostgresBackendError> {
+    let state = match decode_retention_state(row.retention_state.as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    if state != ksp_store_api::RawRetentionState::Purged {
+        return std::result::Result::Ok(std::option::Option::None);
+    }
+    if row.block_time_unix_millis.is_some() {
+        return std::result::Result::Err(data_invalid("raw_tombstone_block_time"));
+    }
+    let signature = match fixed_bytes::<64>(row.signature) {
+        std::result::Result::Ok(value) => ksp_store_api::RawTransactionSignature::new(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let slot = match decode_u64_decimal(row.slot_text.as_str(), "raw_tombstone_slot") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let format_id = match decode_format_id(row.format_id) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let format_version = match decode_u32_i64(row.format_version, "raw_tombstone_format_version") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let content_hash = match fixed_bytes::<32>(row.content_hash) {
+        std::result::Result::Ok(value) => ksp_store_api::RawContentHash::new(value),
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let reference = ksp_store_api::RawTransactionReference::new(network.clone(), signature);
+    let tombstone = match ksp_store_api::RawTransactionTombstone::try_new(reference, slot, format_id, format_version, content_hash) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_tombstone_model")),
+    };
+    return std::result::Result::Ok(std::option::Option::Some(tombstone));
+}
+
+fn decode_format_id(value: std::string::String) -> std::result::Result<ksp_store_api::RawFormatId, crate::PostgresBackendError> {
+    return match ksp_store_api::RawFormatId::new(value) {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid("raw_format_id")),
+    };
+}
+
+fn decode_origin(value: &str) -> std::result::Result<ksp_store_api::RawAcquisitionOrigin, crate::PostgresBackendError> {
+    return match value {
+        "backfill" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Backfill),
+        "import" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Import),
+        "live" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Live),
+        "repair" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Repair),
+        "replay" => std::result::Result::Ok(ksp_store_api::RawAcquisitionOrigin::Replay),
+        _ => std::result::Result::Err(data_invalid("raw_observation_origin")),
+    };
+}
+
+fn decode_provenance_code(value: std::string::String) -> std::result::Result<ksp_store_api::RawProvenanceCode, crate::PostgresBackendError> {
+    return match ksp_store_api::RawProvenanceCode::new(value) {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid("raw_provenance_code")),
+    };
+}
+
+fn decode_retention_state(value: &str) -> std::result::Result<ksp_store_api::RawRetentionState, crate::PostgresBackendError> {
+    return match value {
+        "full" => std::result::Result::Ok(ksp_store_api::RawRetentionState::Full),
+        "archived" => std::result::Result::Ok(ksp_store_api::RawRetentionState::Archived),
+        "purged" => std::result::Result::Ok(ksp_store_api::RawRetentionState::Purged),
+        _ => std::result::Result::Err(data_invalid("raw_retention_state")),
+    };
+}
+
+fn decode_optional_timestamp(
+    value: std::option::Option<i64>,
+    phase: &'static str,
+) -> std::result::Result<std::option::Option<ksp_store_api::RawTimestamp>, crate::PostgresBackendError> {
+    return match value {
+        std::option::Option::Some(inner) => match decode_timestamp_i64(inner, phase) {
+            std::result::Result::Ok(decoded) => std::result::Result::Ok(std::option::Option::Some(decoded)),
+            std::result::Result::Err(error) => std::result::Result::Err(error),
+        },
+        std::option::Option::None => std::result::Result::Ok(std::option::Option::None),
+    };
+}
+
+fn decode_timestamp_i64(value: i64, phase: &'static str) -> std::result::Result<ksp_store_api::RawTimestamp, crate::PostgresBackendError> {
+    let unsigned = match u64::try_from(value) {
+        std::result::Result::Ok(decoded) => decoded,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid(phase)),
+    };
+    return match ksp_store_api::RawTimestamp::from_unix_millis(unsigned) {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid(phase)),
+    };
+}
+
+fn decode_u32_i64(value: i64, phase: &'static str) -> std::result::Result<u32, crate::PostgresBackendError> {
+    return match u32::try_from(value) {
+        std::result::Result::Ok(decoded) if decoded > 0 => std::result::Result::Ok(decoded),
+        _ => std::result::Result::Err(data_invalid(phase)),
+    };
+}
+
+fn decode_u64_decimal(value: &str, phase: &'static str) -> std::result::Result<u64, crate::PostgresBackendError> {
+    return match value.parse::<u64>() {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid(phase)),
+    };
+}
+
+fn decode_u64_i64(value: i64, phase: &'static str) -> std::result::Result<u64, crate::PostgresBackendError> {
+    return match u64::try_from(value) {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid(phase)),
+    };
+}
+
+fn fixed_bytes<const N: usize>(value: std::vec::Vec<u8>) -> std::result::Result<[u8; N], crate::PostgresBackendError> {
+    return match <[u8; N]>::try_from(value.as_slice()) {
+        std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid("raw_fixed_bytes")),
+    };
+}
+
+fn ensure_network(
+    network: &ksp_store_api::RawNetworkId,
+    reference: &ksp_store_api::RawTransactionReference,
+    phase: &'static str,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    if reference.network() != network {
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::WrongNetwork, phase));
+    }
+    return std::result::Result::Ok(());
+}
+
+fn data_invalid(phase: &'static str) -> crate::PostgresBackendError {
+    return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::DataInvalid, phase);
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/raw_transaction.rs"]
+mod tests;

crates/ksp-store-postgres-lib/src/runtime.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/runtime.rs
-// version: 5
+// version: 6
 
 const APPLICATION_NAME: &str = "ksp-store";
 const MAX_CONNECTION_URI_BYTES: usize = 4_096;
@@ -332,6 +332,38 @@ impl PostgresBackend {
         return crate::probe_health(&self.pool).await;
     }
 
+    /// Reads one canonical RAW transaction without exposing physical PostgreSQL row types.
+    pub async fn get_raw_transaction(
+        &self,
+        reference: &ksp_store_api::RawTransactionReference,
+    ) -> std::result::Result<std::option::Option<ksp_store_api::RawTransaction>, crate::PostgresBackendError> {
+        return crate::get_raw_transaction(&self.pool, &self.network, reference).await;
+    }
+
+    /// Reads one persisted RAW transaction observation by producer-owned idempotence key.
+    pub async fn get_raw_transaction_observation(
+        &self,
+        observation_key: &ksp_store_api::RawObservationKey,
+    ) -> std::result::Result<std::option::Option<ksp_store_api::RawTransactionObservation>, crate::PostgresBackendError> {
+        return crate::get_raw_transaction_observation(&self.pool, &self.network, observation_key).await;
+    }
+
+    /// Reads the retention state of one canonical RAW transaction identity.
+    pub async fn get_raw_transaction_retention_state(
+        &self,
+        reference: &ksp_store_api::RawTransactionReference,
+    ) -> std::result::Result<std::option::Option<ksp_store_api::RawRetentionState>, crate::PostgresBackendError> {
+        return crate::get_raw_transaction_retention_state(&self.pool, &self.network, reference).await;
+    }
+
+    /// Reads the minimal durable tombstone only when one RAW transaction is purged.
+    pub async fn get_raw_transaction_tombstone(
+        &self,
+        reference: &ksp_store_api::RawTransactionReference,
+    ) -> std::result::Result<std::option::Option<ksp_store_api::RawTransactionTombstone>, crate::PostgresBackendError> {
+        return crate::get_raw_transaction_tombstone(&self.pool, &self.network, reference).await;
+    }
+
     /// Explicitly closes the pool and waits for all owned pooled objects to drain inside the supplied bound.
     pub async fn close(self, timeout: std::time::Duration) -> std::result::Result<(), crate::PostgresBackendError> {
         self.pool.close();