v0.3.8-pre.009

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/raw_transaction.rs
-// version: 5
+// version: 6
 
 pub(crate) mod cursor;
 
@@ -12,6 +12,8 @@ const GET_TRANSACTION_SQL: &str = "SELECT transaction_row.signature, transaction
 const INSERT_ARCHIVE_PAYLOAD_SQL: &str = "INSERT INTO ksp_raw_transaction_archive_payloads (signature, payload) VALUES ($1, $2)";
 const INSERT_OBSERVATION_SQL: &str = "INSERT INTO ksp_raw_transaction_observations (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) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) ON CONFLICT (observation_key) DO NOTHING RETURNING observation_key";
 const INSERT_TRANSACTION_SQL: &str = "INSERT INTO ksp_raw_transactions (signature, slot, block_time_unix_millis, format_id, format_version, content_hash, payload, retention_state) VALUES ($1, $2::TEXT::NUMERIC, $3, $4, $5, $6, $7, 'full') ON CONFLICT (signature) DO NOTHING RETURNING signature";
+const INSPECT_OBSERVATIONS_ASC_SQL: &str = "WITH filtered_count AS (SELECT COUNT(*)::TEXT AS filtered_count_text FROM ksp_raw_transaction_observations WHERE ($1::BYTEA IS NULL OR transaction_signature = $1)), counts AS (SELECT filtered_count_text, CASE WHEN $1::BYTEA IS NULL THEN filtered_count_text ELSE (SELECT COUNT(*)::TEXT FROM ksp_raw_transaction_observations) END AS total_count_text FROM filtered_count) SELECT counts.total_count_text, counts.filtered_count_text, page.observation_key IS NOT NULL AS page_present, page.observation_key, page.transaction_signature, page.provider, page.protocol, page.acquisition_method, page.origin, page.received_at_unix_millis, page.capture_session_id, page.commitment, page.endpoint_id, page.filter_id, page.observed_at_unix_millis, page.source_payload_hash, page.source_payload_size_bytes FROM counts LEFT JOIN LATERAL (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 ($1::BYTEA IS NULL OR transaction_signature = $1) ORDER BY received_at_unix_millis ASC, observation_key ASC LIMIT $2 OFFSET $3) AS page ON TRUE";
+const INSPECT_OBSERVATIONS_DESC_SQL: &str = "WITH filtered_count AS (SELECT COUNT(*)::TEXT AS filtered_count_text FROM ksp_raw_transaction_observations WHERE ($1::BYTEA IS NULL OR transaction_signature = $1)), counts AS (SELECT filtered_count_text, CASE WHEN $1::BYTEA IS NULL THEN filtered_count_text ELSE (SELECT COUNT(*)::TEXT FROM ksp_raw_transaction_observations) END AS total_count_text FROM filtered_count) SELECT counts.total_count_text, counts.filtered_count_text, page.observation_key IS NOT NULL AS page_present, page.observation_key, page.transaction_signature, page.provider, page.protocol, page.acquisition_method, page.origin, page.received_at_unix_millis, page.capture_session_id, page.commitment, page.endpoint_id, page.filter_id, page.observed_at_unix_millis, page.source_payload_hash, page.source_payload_size_bytes FROM counts LEFT JOIN LATERAL (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 ($1::BYTEA IS NULL OR transaction_signature = $1) ORDER BY received_at_unix_millis DESC, observation_key DESC LIMIT $2 OFFSET $3) AS page ON TRUE";
 const INSPECT_TRANSACTIONS_ASC_SQL: &str = "WITH filtered_count AS (SELECT COUNT(*)::TEXT AS filtered_count_text FROM ksp_raw_transactions WHERE ($1::TEXT IS NULL OR slot >= $1::TEXT::NUMERIC) AND ($2::TEXT IS NULL OR slot <= $2::TEXT::NUMERIC)), counts AS (SELECT filtered_count_text, CASE WHEN $1::TEXT IS NULL AND $2::TEXT IS NULL THEN filtered_count_text ELSE (SELECT COUNT(*)::TEXT FROM ksp_raw_transactions) END AS total_count_text FROM filtered_count) SELECT counts.total_count_text, counts.filtered_count_text, page.signature IS NOT NULL AS page_present, page.signature, page.slot_text, page.block_time_unix_millis, page.format_id, page.format_version, page.content_hash, page.retention_state, page.payload_size_bytes, page.hot_payload_present, page.archive_payload_present FROM counts LEFT JOIN LATERAL (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.retention_state, CASE WHEN transaction_row.retention_state = 'full' THEN OCTET_LENGTH(transaction_row.payload)::BIGINT WHEN transaction_row.retention_state = 'archived' THEN OCTET_LENGTH(archive_row.payload)::BIGINT ELSE NULL END AS payload_size_bytes, transaction_row.payload IS NOT NULL AS hot_payload_present, archive_row.payload IS NOT NULL AS archive_payload_present 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 ($1::TEXT IS NULL OR transaction_row.slot >= $1::TEXT::NUMERIC) AND ($2::TEXT IS NULL OR transaction_row.slot <= $2::TEXT::NUMERIC) ORDER BY transaction_row.slot ASC, transaction_row.signature ASC LIMIT $3 OFFSET $4) AS page ON TRUE";
 const INSPECT_TRANSACTIONS_DESC_SQL: &str = "WITH filtered_count AS (SELECT COUNT(*)::TEXT AS filtered_count_text FROM ksp_raw_transactions WHERE ($1::TEXT IS NULL OR slot >= $1::TEXT::NUMERIC) AND ($2::TEXT IS NULL OR slot <= $2::TEXT::NUMERIC)), counts AS (SELECT filtered_count_text, CASE WHEN $1::TEXT IS NULL AND $2::TEXT IS NULL THEN filtered_count_text ELSE (SELECT COUNT(*)::TEXT FROM ksp_raw_transactions) END AS total_count_text FROM filtered_count) SELECT counts.total_count_text, counts.filtered_count_text, page.signature IS NOT NULL AS page_present, page.signature, page.slot_text, page.block_time_unix_millis, page.format_id, page.format_version, page.content_hash, page.retention_state, page.payload_size_bytes, page.hot_payload_present, page.archive_payload_present FROM counts LEFT JOIN LATERAL (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.retention_state, CASE WHEN transaction_row.retention_state = 'full' THEN OCTET_LENGTH(transaction_row.payload)::BIGINT WHEN transaction_row.retention_state = 'archived' THEN OCTET_LENGTH(archive_row.payload)::BIGINT ELSE NULL END AS payload_size_bytes, transaction_row.payload IS NOT NULL AS hot_payload_present, archive_row.payload IS NOT NULL AS archive_payload_present 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 ($1::TEXT IS NULL OR transaction_row.slot >= $1::TEXT::NUMERIC) AND ($2::TEXT IS NULL OR transaction_row.slot <= $2::TEXT::NUMERIC) ORDER BY transaction_row.slot DESC, transaction_row.signature DESC LIMIT $3 OFFSET $4) AS page ON TRUE";
 const LIST_TRANSACTIONS_ASC_SQL: &str = "SELECT signature, slot::text AS slot_text FROM ksp_raw_transactions WHERE retention_state <> 'purged' AND ($1::TEXT IS NULL OR slot >= $1::TEXT::NUMERIC) AND ($2::TEXT IS NULL OR slot <= $2::TEXT::NUMERIC) AND ($3::TEXT IS NULL OR (slot, signature) > ($3::TEXT::NUMERIC, $4::BYTEA)) ORDER BY slot ASC, signature ASC LIMIT $5";
@@ -48,6 +50,26 @@ struct RawInspectionDbRow {
     total_count_text: std::string::String,
 }
 
+struct RawObservationInspectionDbRow {
+    acquisition_method: std::option::Option<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>,
+    filtered_count_text: std::string::String,
+    observation_key: std::option::Option<std::vec::Vec<u8>>,
+    observed_at_unix_millis: std::option::Option<i64>,
+    origin: std::option::Option<std::string::String>,
+    page_present: bool,
+    protocol: std::option::Option<std::string::String>,
+    provider: std::option::Option<std::string::String>,
+    received_at_unix_millis: std::option::Option<i64>,
+    source_payload_hash: std::option::Option<std::vec::Vec<u8>>,
+    source_payload_size_bytes: std::option::Option<i64>,
+    total_count_text: std::string::String,
+    transaction_signature: std::option::Option<std::vec::Vec<u8>>,
+}
+
 struct RawObservationDbRow {
     acquisition_method: std::string::String,
     capture_session_id: std::option::Option<std::string::String>,
@@ -307,6 +329,116 @@ pub(crate) async fn inspect_raw_transactions(
     };
 }
 
+/// Inspects one safe random-access RAW transaction-observation window with exact counts.
+pub(crate) async fn inspect_raw_transaction_observations(
+    pool: &deadpool_postgres::Pool,
+    network: &ksp_store_api::RawNetworkId,
+    query: &ksp_store_api::RawTransactionObservationInspectionQuery,
+) -> std::result::Result<ksp_store_api::RawInspectionPage<ksp_store_api::RawTransactionObservationSummary>, crate::PostgresBackendError> {
+    if query.network() != network {
+        return std::result::Result::Err(crate::PostgresBackendError::new(
+            crate::PostgresBackendErrorKind::WrongNetwork,
+            "raw_transaction_observation_inspection_network",
+        ));
+    }
+    let sql = match query.direction() {
+        ksp_store_api::RawSortDirection::Ascending => INSPECT_OBSERVATIONS_ASC_SQL,
+        ksp_store_api::RawSortDirection::Descending => INSPECT_OBSERVATIONS_DESC_SQL,
+        _ => {
+            return std::result::Result::Err(crate::PostgresBackendError::new(
+                crate::PostgresBackendErrorKind::QueryInvalid,
+                "raw_transaction_observation_inspection_direction",
+            ));
+        },
+    };
+    let (sql_limit, sql_offset) = match raw_transaction_inspection_sql_window(query.page()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let signature = query.transaction().map(|reference| return reference.signature().as_bytes().to_vec());
+    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 rows_result = client.query(sql, &[&signature, &sql_limit, &sql_offset]).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_observation_inspection_query",
+            ));
+        },
+    };
+    if rows.is_empty() {
+        return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_cardinality"));
+    }
+    let row_count = rows.len();
+    let mut total_items = std::option::Option::None;
+    let mut filtered_items = std::option::Option::None;
+    let mut items = std::vec::Vec::new();
+    for row in rows {
+        let physical = match raw_observation_inspection_db_row(&row) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let row_total = match decode_u64_decimal(physical.total_count_text.as_str(), "raw_transaction_observation_inspection_total") {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let row_filtered = match decode_u64_decimal(physical.filtered_count_text.as_str(), "raw_transaction_observation_inspection_filtered") {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        match (total_items, filtered_items) {
+            (std::option::Option::None, std::option::Option::None) => {
+                total_items = std::option::Option::Some(row_total);
+                filtered_items = std::option::Option::Some(row_filtered);
+            },
+            (std::option::Option::Some(total), std::option::Option::Some(filtered)) if total == row_total && filtered == row_filtered => {},
+            _ => return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_counts")),
+        }
+        if physical.page_present {
+            let observation = match decode_raw_observation_inspection_row(network, physical) {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            if let std::option::Option::Some(reference) = query.transaction() {
+                if observation.transaction() != reference {
+                    return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_reference"));
+                }
+            }
+            items.push(ksp_store_api::RawTransactionObservationSummary::new(
+                observation.observation_key(),
+                observation.transaction().clone(),
+                observation.provenance().clone(),
+            ));
+        } else if row_count != 1 || !raw_observation_inspection_empty_page_is_clean(&physical) {
+            return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_page"));
+        }
+    }
+    let total_items = match total_items {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_total")),
+    };
+    let filtered_items = match filtered_items {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_filtered")),
+    };
+    let item_count = match u64::try_from(items.len()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_page")),
+    };
+    if item_count > query.page().limit().get() {
+        return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_page"));
+    }
+    return match ksp_store_api::RawInspectionPage::try_new(items, total_items, filtered_items) {
+        std::result::Result::Ok(value) => std::result::Result::Ok(value),
+        std::result::Result::Err(_) => std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_page")),
+    };
+}
+
 /// Reads one RAW transaction observation from the physical PostgreSQL backend.
 pub(crate) async fn get_raw_transaction_observation(
     pool: &deadpool_postgres::Pool,
@@ -1401,6 +1533,116 @@ fn raw_transaction_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawT
     });
 }
 
+fn raw_observation_inspection_empty_page_is_clean(row: &RawObservationInspectionDbRow) -> bool {
+    return row.acquisition_method.is_none()
+        && row.capture_session_id.is_none()
+        && row.commitment.is_none()
+        && row.endpoint_id.is_none()
+        && row.filter_id.is_none()
+        && row.observation_key.is_none()
+        && row.observed_at_unix_millis.is_none()
+        && row.origin.is_none()
+        && row.protocol.is_none()
+        && row.provider.is_none()
+        && row.received_at_unix_millis.is_none()
+        && row.source_payload_hash.is_none()
+        && row.source_payload_size_bytes.is_none()
+        && row.transaction_signature.is_none();
+}
+
+fn raw_observation_inspection_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawObservationInspectionDbRow, crate::PostgresBackendError> {
+    macro_rules! get_optional {
+        ($name:literal, $ty:ty) => {
+            match row.try_get::<_, std::option::Option<$ty>>($name) {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_decode")),
+            }
+        };
+    }
+    let filtered_count_text = match row.try_get::<_, std::string::String>("filtered_count_text") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_decode")),
+    };
+    let page_present = match row.try_get::<_, bool>("page_present") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_decode")),
+    };
+    let total_count_text = match row.try_get::<_, std::string::String>("total_count_text") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_observation_inspection_decode")),
+    };
+    return std::result::Result::Ok(RawObservationInspectionDbRow {
+        acquisition_method: get_optional!("acquisition_method", std::string::String),
+        capture_session_id: get_optional!("capture_session_id", std::string::String),
+        commitment: get_optional!("commitment", std::string::String),
+        endpoint_id: get_optional!("endpoint_id", std::string::String),
+        filter_id: get_optional!("filter_id", std::string::String),
+        filtered_count_text,
+        observation_key: get_optional!("observation_key", std::vec::Vec<u8>),
+        observed_at_unix_millis: get_optional!("observed_at_unix_millis", i64),
+        origin: get_optional!("origin", std::string::String),
+        page_present,
+        protocol: get_optional!("protocol", std::string::String),
+        provider: get_optional!("provider", std::string::String),
+        received_at_unix_millis: get_optional!("received_at_unix_millis", i64),
+        source_payload_hash: get_optional!("source_payload_hash", std::vec::Vec<u8>),
+        source_payload_size_bytes: get_optional!("source_payload_size_bytes", i64),
+        total_count_text,
+        transaction_signature: get_optional!("transaction_signature", std::vec::Vec<u8>),
+    });
+}
+
+fn decode_raw_observation_inspection_row(
+    network: &ksp_store_api::RawNetworkId,
+    row: RawObservationInspectionDbRow,
+) -> std::result::Result<ksp_store_api::RawTransactionObservation, crate::PostgresBackendError> {
+    let observation_key = match inspection_required(row.observation_key, "raw_transaction_observation_inspection_shape") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let received_at_unix_millis = match inspection_required(row.received_at_unix_millis, "raw_transaction_observation_inspection_shape") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let transaction_signature = match inspection_required(row.transaction_signature, "raw_transaction_observation_inspection_shape") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let acquisition_method = match inspection_required(row.acquisition_method, "raw_transaction_observation_inspection_shape") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let origin = match inspection_required(row.origin, "raw_transaction_observation_inspection_shape") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let protocol = match inspection_required(row.protocol, "raw_transaction_observation_inspection_shape") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let provider = match inspection_required(row.provider, "raw_transaction_observation_inspection_shape") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let physical = RawObservationDbRow {
+        acquisition_method,
+        capture_session_id: row.capture_session_id,
+        commitment: row.commitment,
+        endpoint_id: row.endpoint_id,
+        filter_id: row.filter_id,
+        observation_key,
+        observed_at_unix_millis: row.observed_at_unix_millis,
+        origin,
+        protocol,
+        provider,
+        received_at_unix_millis,
+        source_payload_hash: row.source_payload_hash,
+        source_payload_size_bytes: row.source_payload_size_bytes,
+        transaction_signature,
+    };
+    return decode_raw_observation_row(network, physical);
+}
+
 fn raw_observation_db_row(row: &tokio_postgres::Row) -> std::result::Result<RawObservationDbRow, crate::PostgresBackendError> {
     macro_rules! required {
         ($name:literal, $ty:ty) => {