// file: crates/ksp-store-postgres-lib/src/raw_transaction.rs // version: 14 pub(crate) mod cursor; const DELETE_ARCHIVE_PAYLOAD_SQL: &str = "DELETE FROM ksp_raw_transaction_archive_payloads WHERE signature = $1"; const GET_ARCHIVE_PAYLOAD_SQL: &str = "SELECT payload FROM ksp_raw_transaction_archive_payloads WHERE signature = $1"; const GET_FIRST_TRANSACTION_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 transaction_signature = $1 ORDER BY received_at_unix_millis ASC, observation_key ASC LIMIT 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"; 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_CONFLICT_SQL: &str = "INSERT INTO ksp_raw_transaction_conflicts (transaction_signature, status, revision, canonical_variant_id, incoming_variant_id, latest_relation, latest_reason_code, created_at_unix_millis, updated_at_unix_millis) VALUES ($1, 'open', 1, $2::TEXT::NUMERIC, $3::TEXT::NUMERIC, $4, $5, $6, $6)"; 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 INSERT_TRANSACTION_VARIANT_LINK_SQL: &str = "INSERT INTO ksp_raw_transaction_observation_variants (observation_key, transaction_signature, variant_id, linked_at_unix_millis) VALUES ($1, $2, $3::TEXT::NUMERIC, $4) RETURNING observation_key"; const INSERT_TRANSACTION_VARIANT_SELECTOR_SQL: &str = "INSERT INTO ksp_raw_transaction_canonical_selectors (transaction_signature, canonical_variant_id, canonical_revision, updated_at_unix_millis) VALUES ($1, $2::TEXT::NUMERIC, 1, $3)"; const INSERT_TRANSACTION_VARIANT_SQL: &str = "INSERT INTO ksp_raw_transaction_variants (transaction_signature, variant_id, origin_kind, slot, block_time_unix_millis, format_id, format_version, content_hash, payload, retention_state, created_at_unix_millis) VALUES ($1, $2::TEXT::NUMERIC, 'native', $3::TEXT::NUMERIC, $4, $5, $6, $7, $8, $9, $10)"; 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"; const LIST_TRANSACTIONS_DESC_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 DESC, signature DESC LIMIT $5"; const LOCK_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 FOR UPDATE"; const LOCK_RETENTION_TRANSACTION_SQL: &str = "SELECT block_time_unix_millis, payload, retention_state FROM ksp_raw_transactions WHERE signature = $1 FOR UPDATE"; const LOCK_TRANSACTION_CONFLICT_SQL: &str = "SELECT status, revision::TEXT AS revision_text, canonical_variant_id::TEXT AS canonical_variant_id_text, incoming_variant_id::TEXT AS incoming_variant_id_text, latest_relation, latest_reason_code, created_at_unix_millis, updated_at_unix_millis FROM ksp_raw_transaction_conflicts WHERE transaction_signature = $1 FOR UPDATE"; const LOCK_TRANSACTION_SQL: &str = "SELECT signature, slot::text AS slot_text, block_time_unix_millis, format_id, format_version, content_hash, payload, retention_state, NULL::BYTEA AS archive_payload FROM ksp_raw_transactions WHERE signature = $1 FOR UPDATE"; const RAW_TRANSACTION_CONTENT_CONFLICT_META_FIELDS: [&str; 15] = [ "err", "status", "fee", "preBalances", "postBalances", "innerInstructions", "logMessages", "preTokenBalances", "postTokenBalances", "rewards", "loadedAddresses", "returnData", "computeUnitsConsumed", "costUnits", "accounts", ]; const RAW_TRANSACTION_CONTENT_CONFLICT_PAYLOAD_FIELDS: [&str; 4] = ["transaction", "meta", "version", "transactionIndex"]; const REHYDRATE_TRANSACTION_SQL: &str = "UPDATE ksp_raw_transactions SET block_time_unix_millis = $2, payload = $3, retention_state = 'full' WHERE signature = $1"; const SELECT_CANONICAL_VARIANT_ID_SQL: &str = "SELECT canonical_variant_id::TEXT AS canonical_variant_id_text FROM ksp_raw_transaction_canonical_selectors WHERE transaction_signature = $1"; const SELECT_CANONICAL_VARIANT_SELECTOR_SQL: &str = "SELECT canonical_variant_id::TEXT AS canonical_variant_id_text, canonical_revision::TEXT AS canonical_revision_text FROM ksp_raw_transaction_canonical_selectors WHERE transaction_signature = $1"; const SELECT_EXACT_NATIVE_VARIANT_SQL: &str = "SELECT variant_id::TEXT AS variant_id_text FROM ksp_raw_transaction_variants WHERE transaction_signature = $1 AND origin_kind = 'native' AND slot = $2::TEXT::NUMERIC AND block_time_unix_millis IS NOT DISTINCT FROM $3 AND format_id = $4 AND format_version = $5 AND content_hash = $6 AND payload = $7 AND retention_state = 'full' ORDER BY variant_id ASC LIMIT 1"; const SELECT_MAX_VARIANT_ID_SQL: &str = "SELECT COALESCE(MAX(variant_id), 0)::TEXT AS variant_id_text FROM ksp_raw_transaction_variants WHERE transaction_signature = $1"; const SELECT_TRANSACTION_VARIANT_LINK_SQL: &str = "SELECT transaction_signature, variant_id::TEXT AS variant_id_text FROM ksp_raw_transaction_observation_variants WHERE observation_key = $1"; const UPDATE_ARCHIVED_TRANSACTION_SQL: &str = "UPDATE ksp_raw_transactions SET payload = NULL, retention_state = 'archived' WHERE signature = $1 AND retention_state = 'full'"; const UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL: &str = "UPDATE ksp_raw_transactions SET slot = $2::TEXT::NUMERIC, block_time_unix_millis = $3, format_id = $4, format_version = $5, content_hash = $6, payload = $7, retention_state = 'full' WHERE signature = $1"; const UPDATE_PURGED_TRANSACTION_SQL: &str = "UPDATE ksp_raw_transactions SET block_time_unix_millis = NULL, payload = NULL, retention_state = 'purged' WHERE signature = $1 AND retention_state = 'archived'"; const UPDATE_REHYDRATED_TRANSACTION_VARIANT_SQL: &str = "UPDATE ksp_raw_transaction_variants SET block_time_unix_millis = $3, payload = $4, retention_state = 'full' WHERE transaction_signature = $1 AND variant_id = $2::TEXT::NUMERIC"; const UPDATE_TRANSACTION_CONFLICT_SQL: &str = "UPDATE ksp_raw_transaction_conflicts SET status = 'open', revision = $2::TEXT::NUMERIC, canonical_variant_id = $3::TEXT::NUMERIC, incoming_variant_id = $4::TEXT::NUMERIC, latest_relation = $5, latest_reason_code = $6, updated_at_unix_millis = $7 WHERE transaction_signature = $1 AND revision = $8::TEXT::NUMERIC"; const UPDATE_TRANSACTION_VARIANT_SELECTOR_SQL: &str = "UPDATE ksp_raw_transaction_canonical_selectors SET canonical_variant_id = $4::TEXT::NUMERIC, canonical_revision = $5::TEXT::NUMERIC, updated_at_unix_millis = $6 WHERE transaction_signature = $1 AND canonical_variant_id = $2::TEXT::NUMERIC AND canonical_revision = $3::TEXT::NUMERIC"; const UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL: &str = "UPDATE ksp_raw_transaction_variants SET payload = $8, retention_state = 'full' WHERE transaction_signature = $1 AND variant_id = $2::TEXT::NUMERIC AND slot = $3::TEXT::NUMERIC AND block_time_unix_millis IS NOT DISTINCT FROM $4 AND format_id = $5 AND format_version = $6 AND content_hash = $7 AND ((retention_state = 'archived' AND payload IS NULL) OR (retention_state = 'full' AND payload = $8))"; struct RawLogMessagesContentConflictDiagnostic { available: bool, first_mismatch_index: std::string::String, incoming_count: usize, incoming_first_kind: &'static str, incoming_first_length: usize, incoming_has_truncation_marker: bool, incoming_prefix_of_stored: bool, incoming_state: &'static str, stored_count: usize, stored_first_kind: &'static str, stored_first_length: usize, stored_has_truncation_marker: bool, stored_prefix_of_incoming: bool, stored_state: &'static str, } struct RawPayloadContentConflictDiagnostic { available: bool, log_messages: RawLogMessagesContentConflictDiagnostic, meta_mismatch: bool, meta_mismatch_fields: std::string::String, other_mismatch: bool, transaction_index_mismatch: bool, transaction_mismatch: bool, version_mismatch: bool, } struct RawTransactionContentConflictDiagnostic { block_time_mismatch: bool, content_hash_mismatch: bool, format_id_mismatch: bool, format_version_mismatch: bool, log_messages: RawLogMessagesContentConflictDiagnostic, meta_mismatch_fields: std::string::String, payload_bytes_mismatch: bool, payload_diagnostic_available: bool, payload_meta_mismatch: bool, payload_other_mismatch: bool, payload_transaction_index_mismatch: bool, payload_transaction_mismatch: bool, payload_version_mismatch: bool, slot_mismatch: bool, } struct RawListDbRow { signature: std::vec::Vec, slot_text: std::string::String, } struct RawInspectionDbRow { archive_payload_present: std::option::Option, block_time_unix_millis: std::option::Option, content_hash: std::option::Option>, filtered_count_text: std::string::String, format_id: std::option::Option, format_version: std::option::Option, hot_payload_present: std::option::Option, page_present: bool, payload_size_bytes: std::option::Option, retention_state: std::option::Option, signature: std::option::Option>, slot_text: std::option::Option, total_count_text: std::string::String, } struct RawObservationInspectionDbRow { acquisition_method: std::option::Option, capture_session_id: std::option::Option, commitment: std::option::Option, endpoint_id: std::option::Option, filter_id: std::option::Option, filtered_count_text: std::string::String, observation_key: std::option::Option>, observed_at_unix_millis: std::option::Option, origin: std::option::Option, page_present: bool, protocol: std::option::Option, provider: std::option::Option, received_at_unix_millis: std::option::Option, source_payload_hash: std::option::Option>, source_payload_size_bytes: std::option::Option, total_count_text: std::string::String, transaction_signature: std::option::Option>, } struct RawObservationDbRow { acquisition_method: std::string::String, capture_session_id: std::option::Option, commitment: std::option::Option, endpoint_id: std::option::Option, filter_id: std::option::Option, observation_key: std::vec::Vec, observed_at_unix_millis: std::option::Option, origin: std::string::String, protocol: std::string::String, provider: std::string::String, received_at_unix_millis: i64, source_payload_hash: std::option::Option>, source_payload_size_bytes: std::option::Option, transaction_signature: std::vec::Vec, } struct RawRetentionDbRow { block_time_unix_millis: std::option::Option, payload: std::option::Option>, retention_state: std::string::String, } #[derive(Clone)] struct RawTransactionDbRow { archive_payload: std::option::Option>, block_time_unix_millis: std::option::Option, content_hash: std::vec::Vec, format_id: std::string::String, format_version: i64, payload: std::option::Option>, retention_state: std::string::String, signature: std::vec::Vec, slot_text: std::string::String, } struct RawTombstoneDbRow { block_time_unix_millis: std::option::Option, content_hash: std::vec::Vec, format_id: std::string::String, format_version: i64, retention_state: std::string::String, signature: std::vec::Vec, 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, 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); } /// Lists deterministic canonical RAW transaction references using PostgreSQL keyset pagination. pub(crate) async fn list_raw_transactions( pool: &deadpool_postgres::Pool, network: &ksp_store_api::RawNetworkId, query: &ksp_store_api::RawTransactionQuery, ) -> std::result::Result, crate::PostgresBackendError> { if query.network() != network { return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::WrongNetwork, "raw_transaction_list_network")); } let (requested_usize, sql_limit) = match crate::raw_transaction_physical_page_limit(query.page().limit().get()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let decoded_cursor = match query.page().cursor() { std::option::Option::Some(value) => match crate::decode_raw_transaction_cursor(query, value) { std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), std::result::Result::Err(error) => return std::result::Result::Err(error), }, std::option::Option::None => std::option::Option::None, }; 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 slots = query.slots(); let start_text = slots.start_inclusive().map(|value| return value.to_string()); let end_text = slots.end_inclusive().map(|value| return value.to_string()); let cursor_slot_text = decoded_cursor.as_ref().map(|value| return value.last_slot.to_string()); let cursor_signature = decoded_cursor.as_ref().map(|value| return value.last_signature.to_vec()); let sql = match query.direction() { ksp_store_api::RawSortDirection::Ascending => LIST_TRANSACTIONS_ASC_SQL, ksp_store_api::RawSortDirection::Descending => LIST_TRANSACTIONS_DESC_SQL, _ => { return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::QueryInvalid, "raw_transaction_list_direction")); }, }; let rows_result = client.query(sql, &[&start_text, &end_text, &cursor_slot_text, &cursor_signature, &sql_limit]).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_list_query")); }, }; let mut decoded = std::vec::Vec::with_capacity(rows.len()); for row in rows { let physical = match raw_list_db_row(&row) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let item = match decode_raw_list_row(network, physical) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; decoded.push(item); } let has_more = decoded.len() > requested_usize; if has_more { decoded.truncate(requested_usize); } let next_cursor = if has_more { let last = match decoded.last() { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(data_invalid("raw_transaction_list_page")), }; match crate::encode_raw_transaction_cursor(query, last.0, &last.1.signature()) { std::result::Result::Ok(value) => std::option::Option::Some(value), std::result::Result::Err(error) => return std::result::Result::Err(error), } } else { std::option::Option::None }; let items = decoded.into_iter().map(|value| return value.1).collect(); return std::result::Result::Ok(ksp_store_api::RawPage::new(items, next_cursor)); } /// Inspects one payload-free random-access RAW transaction window with exact counts. pub(crate) async fn inspect_raw_transactions( pool: &deadpool_postgres::Pool, network: &ksp_store_api::RawNetworkId, query: &ksp_store_api::RawTransactionInspectionQuery, ) -> std::result::Result, crate::PostgresBackendError> { if query.network() != network { return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::WrongNetwork, "raw_transaction_inspection_network")); } let sql = match query.direction() { ksp_store_api::RawSortDirection::Ascending => INSPECT_TRANSACTIONS_ASC_SQL, ksp_store_api::RawSortDirection::Descending => INSPECT_TRANSACTIONS_DESC_SQL, _ => { return std::result::Result::Err(crate::PostgresBackendError::new( crate::PostgresBackendErrorKind::QueryInvalid, "raw_transaction_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 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 slots = query.slots(); let start_text = slots.start_inclusive().map(|value| return value.to_string()); let end_text = slots.end_inclusive().map(|value| return value.to_string()); let rows_result = client.query(sql, &[&start_text, &end_text, &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_inspection_query")); }, }; if rows.is_empty() { return std::result::Result::Err(data_invalid("raw_transaction_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_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_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_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_inspection_counts")), } if physical.page_present { let summary = match decode_raw_inspection_summary(network, physical) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; items.push(summary); } else if row_count != 1 || !raw_inspection_empty_page_is_clean(&physical) { return std::result::Result::Err(data_invalid("raw_transaction_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_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_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_inspection_page")), }; if item_count > query.page().limit().get() { return std::result::Result::Err(data_invalid("raw_transaction_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_inspection_page")), }; } /// 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, 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() && 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, network: &ksp_store_api::RawNetworkId, observation_key: &ksp_store_api::RawObservationKey, ) -> std::result::Result, 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, 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, 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)); } /// Persists one canonical RAW transaction and one observation atomically. pub(crate) async fn persist_raw_transaction_acquisition( pool: &deadpool_postgres::Pool, network: &ksp_store_api::RawNetworkId, raw_transaction: ksp_store_api::RawTransaction, observation: ksp_store_api::RawTransactionObservation, mode: ksp_store_api::RawTransactionAcquisitionMode, ) -> std::result::Result { let input_result = ensure_acquisition_inputs(network, &raw_transaction, &observation); if let std::result::Result::Err(error) = input_result { return std::result::Result::Err(error); } let linked_at = match raw_variant_timestamp(observation.provenance().received_at(), "raw_acquisition_variant_timestamp") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let client_result = pool.get().await; let mut 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 sql_transaction_result = client.transaction().await; let sql_transaction = match sql_transaction_result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_acquisition_begin")), }; let insert_result = insert_canonical_transaction(&sql_transaction, &raw_transaction).await; let inserted = match insert_result { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let locked_result = load_locked_transaction_row(&sql_transaction, raw_transaction.reference()).await; let locked = match locked_result { std::result::Result::Ok(std::option::Option::Some(value)) => value, std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Err(data_invalid("raw_acquisition_conflict_missing")), std::result::Result::Err(error) => return std::result::Result::Err(error), }; let canonical_variant_result = ensure_v003_transaction_variant_bootstrap(&sql_transaction, &locked, linked_at).await; let canonical_variant_id = match canonical_variant_result { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let (entity_outcome, observed_variant_id, transaction_variant_outcome, canonical_promotion, quarantined_conflict) = if inserted { ( ksp_store_api::RawEntityWriteOutcome::Inserted, canonical_variant_id, ksp_store_api::RawTransactionVariantWriteOutcome::InsertedCanonical, std::option::Option::None, std::option::Option::None, ) } else { let comparison = match compare_existing_transaction(network, locked.clone(), &raw_transaction) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; match comparison { ExistingTransactionMatch::Active => ( ksp_store_api::RawEntityWriteOutcome::AlreadyPresent, canonical_variant_id, ksp_store_api::RawTransactionVariantWriteOutcome::ObservedExact, std::option::Option::None, std::option::Option::None, ), ExistingTransactionMatch::ActiveCompatibleLessComplete => { let variant_id = match persist_or_reuse_native_transaction_variant(&sql_transaction, &raw_transaction, linked_at).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; ( ksp_store_api::RawEntityWriteOutcome::AlreadyPresent, variant_id, ksp_store_api::RawTransactionVariantWriteOutcome::ObservedCompatibleLessComplete, std::option::Option::None, std::option::Option::None, ) }, ExistingTransactionMatch::ActiveCompatibleMoreComplete => { let variant_id = match persist_or_reuse_native_transaction_variant(&sql_transaction, &raw_transaction, linked_at).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let promotion = match promote_more_complete_transaction_variant(&sql_transaction, &locked, &raw_transaction, canonical_variant_id, variant_id, linked_at) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; ( ksp_store_api::RawEntityWriteOutcome::AlreadyPresent, variant_id, ksp_store_api::RawTransactionVariantWriteOutcome::PromotedCompatibleMoreComplete, std::option::Option::Some(promotion), std::option::Option::None, ) }, ExistingTransactionMatch::ActiveConflict(variant_comparison) => { let variant_id = match persist_or_reuse_native_transaction_variant(&sql_transaction, &raw_transaction, linked_at).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let conflict_transition = match open_or_update_raw_transaction_conflict( &sql_transaction, raw_transaction.reference(), canonical_variant_id, variant_id, variant_comparison, linked_at, ) .await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; log_raw_transaction_content_conflict_provenance(&sql_transaction, network, &observation).await; ( ksp_store_api::RawEntityWriteOutcome::AlreadyPresent, variant_id, ksp_store_api::RawTransactionVariantWriteOutcome::QuarantinedConflict, std::option::Option::None, std::option::Option::Some(conflict_transition), ) }, ExistingTransactionMatch::Purged => { if mode == ksp_store_api::RawTransactionAcquisitionMode::ForceRehydrate { let rehydrate_result = rehydrate_transaction(&sql_transaction, &raw_transaction).await; if let std::result::Result::Err(error) = rehydrate_result { return std::result::Result::Err(error); } let variant_result = rehydrate_canonical_transaction_variant(&sql_transaction, &raw_transaction, canonical_variant_id).await; if let std::result::Result::Err(error) = variant_result { return std::result::Result::Err(error); } ( ksp_store_api::RawEntityWriteOutcome::Rehydrated, canonical_variant_id, ksp_store_api::RawTransactionVariantWriteOutcome::Rehydrated, std::option::Option::None, std::option::Option::None, ) } else { let commit_result = sql_transaction.commit().await; if commit_result.is_err() { return std::result::Result::Err(write_failed("raw_acquisition_commit")); } return std::result::Result::Ok(ksp_store_api::RawAcquisitionWriteOutcome::with_transaction_variant( ksp_store_api::RawEntityWriteOutcome::SkippedPurged, ksp_store_api::RawObservationWriteOutcome::NotRecorded, ksp_store_api::RawTransactionVariantWriteOutcome::SkippedPurged, )); } }, } }; let observation_result = persist_observation_row(&sql_transaction, network, &observation).await; let observation_outcome = match observation_result { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let link_result = persist_transaction_variant_link(&sql_transaction, &observation, observed_variant_id, observation_outcome, linked_at).await; if let std::result::Result::Err(error) = link_result { return std::result::Result::Err(error); } let commit_result = sql_transaction.commit().await; if commit_result.is_err() { return std::result::Result::Err(write_failed("raw_acquisition_commit")); } if let std::option::Option::Some(transition) = canonical_promotion { log_raw_transaction_canonical_promotion_committed(network, raw_transaction.slot(), transition); } if let std::option::Option::Some(transition) = quarantined_conflict { let stored = decode_raw_transaction_row(network, locked.clone()); if let std::result::Result::Ok(std::option::Option::Some(stored)) = stored { log_raw_transaction_content_conflict(network, &stored, &raw_transaction); } log_raw_transaction_conflict_quarantined_committed(network, raw_transaction.slot(), transition); } return std::result::Result::Ok(ksp_store_api::RawAcquisitionWriteOutcome::with_transaction_variant( entity_outcome, observation_outcome, transaction_variant_outcome, )); } /// Persists one additional observation for an already known RAW transaction. pub(crate) async fn record_raw_transaction_observation( pool: &deadpool_postgres::Pool, network: &ksp_store_api::RawNetworkId, observation: ksp_store_api::RawTransactionObservation, ) -> std::result::Result { let network_result = ensure_network(network, observation.transaction(), "raw_observation_write_network"); if let std::result::Result::Err(error) = network_result { return std::result::Result::Err(error); } let linked_at = match raw_variant_timestamp(observation.provenance().received_at(), "raw_observation_variant_timestamp") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let client_result = pool.get().await; let mut 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 sql_transaction_result = client.transaction().await; let sql_transaction = match sql_transaction_result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_observation_begin")), }; let locked_result = load_locked_transaction_row(&sql_transaction, observation.transaction()).await; let locked = match locked_result { std::result::Result::Ok(std::option::Option::Some(value)) => value, std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Err(reference_not_found("raw_observation_transaction")), std::result::Result::Err(error) => return std::result::Result::Err(error), }; let state = match decode_retention_state(locked.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 { let commit_result = sql_transaction.commit().await; if commit_result.is_err() { return std::result::Result::Err(write_failed("raw_observation_commit")); } return std::result::Result::Ok(ksp_store_api::RawObservationWriteOutcome::NotRecorded); } let canonical_variant_result = ensure_v003_transaction_variant_bootstrap(&sql_transaction, &locked, linked_at).await; let canonical_variant_id = match canonical_variant_result { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let observation_result = persist_observation_row(&sql_transaction, network, &observation).await; let outcome = match observation_result { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let link_result = persist_transaction_variant_link(&sql_transaction, &observation, canonical_variant_id, outcome, linked_at).await; if let std::result::Result::Err(error) = link_result { return std::result::Result::Err(error); } let commit_result = sql_transaction.commit().await; if commit_result.is_err() { return std::result::Result::Err(write_failed("raw_observation_commit")); } return std::result::Result::Ok(outcome); } /// Applies one atomic compare-and-transition RAW transaction retention mutation. pub(crate) async fn transition_raw_transaction_retention( pool: &deadpool_postgres::Pool, network: &ksp_store_api::RawNetworkId, transition: ksp_store_api::RawTransactionRetentionTransition, ) -> std::result::Result { let input_result = ensure_retention_transition_inputs(network, &transition); if let std::result::Result::Err(error) = input_result { return std::result::Result::Err(error); } let client_result = pool.get().await; let mut 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 sql_transaction_result = client.transaction().await; let sql_transaction = match sql_transaction_result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_retention_begin")), }; let signature = transition.reference().signature(); let signature_bytes: &[u8] = signature.as_bytes(); let row_result = sql_transaction.query_opt(LOCK_RETENTION_TRANSACTION_SQL, &[&signature_bytes]).await; let row = match row_result { std::result::Result::Ok(std::option::Option::Some(value)) => value, std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Err(reference_not_found("raw_retention_transition_reference")), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_retention_lock_transaction")), }; let physical = match raw_retention_db_row(&row) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let current = match decode_retention_state(physical.retention_state.as_str()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let shape_result = validate_retention_shape(&sql_transaction, signature_bytes, &physical, current).await; if let std::result::Result::Err(error) = shape_result { return std::result::Result::Err(error); } let decision = match retention_transition_decision(current, transition.expected(), transition.target()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; match decision { RetentionTransitionDecision::AlreadyAtTarget => { return commit_retention_outcome(sql_transaction, ksp_store_api::RawRetentionWriteOutcome::AlreadyAtTarget).await; }, RetentionTransitionDecision::ExpectedStateMismatch => { return commit_retention_outcome(sql_transaction, ksp_store_api::RawRetentionWriteOutcome::ExpectedStateMismatch).await; }, RetentionTransitionDecision::Archive => { let payload = match physical.payload.as_ref() { std::option::Option::Some(value) => value.as_slice(), std::option::Option::None => return std::result::Result::Err(data_invalid("raw_retention_full_payload")), }; let archive_result = archive_full_transaction(&sql_transaction, signature_bytes, payload).await; if let std::result::Result::Err(error) = archive_result { return std::result::Result::Err(error); } }, RetentionTransitionDecision::Purge => { let purge_result = purge_archived_transaction(&sql_transaction, signature_bytes).await; if let std::result::Result::Err(error) = purge_result { return std::result::Result::Err(error); } }, } return commit_retention_outcome(sql_transaction, ksp_store_api::RawRetentionWriteOutcome::Applied).await; } #[derive(Clone, Copy, Debug, Eq, PartialEq)] struct RawCanonicalPromotionTransition { from_revision: u64, from_variant_id: u64, to_revision: u64, to_variant_id: u64, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] struct RawTransactionConflictTransition { canonical_variant_id: u64, incoming_variant_id: u64, reason: ksp_store_api::RawTransactionVariantRelationReason, relation: ksp_store_api::RawTransactionVariantRelation, revision: u64, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum ExistingTransactionMatch { Active, ActiveCompatibleLessComplete, ActiveCompatibleMoreComplete, ActiveConflict(ksp_store_api::RawTransactionVariantComparison), Purged, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum RetentionTransitionDecision { AlreadyAtTarget, Archive, ExpectedStateMismatch, Purge, } async fn archive_full_transaction( sql_transaction: &deadpool_postgres::Transaction<'_>, signature_bytes: &[u8], payload: &[u8], ) -> std::result::Result<(), crate::PostgresBackendError> { let insert_result = sql_transaction.execute(INSERT_ARCHIVE_PAYLOAD_SQL, &[&signature_bytes, &payload]).await; match insert_result { std::result::Result::Ok(1) => {}, std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_retention_archive_insert_cardinality")), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_retention_archive_insert")), } let update_result = sql_transaction.execute(UPDATE_ARCHIVED_TRANSACTION_SQL, &[&signature_bytes]).await; return match update_result { std::result::Result::Ok(1) => std::result::Result::Ok(()), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_retention_archive_update_cardinality")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_retention_archive_update")), }; } async fn commit_retention_outcome( sql_transaction: deadpool_postgres::Transaction<'_>, outcome: ksp_store_api::RawRetentionWriteOutcome, ) -> std::result::Result { let commit_result = sql_transaction.commit().await; if commit_result.is_err() { return std::result::Result::Err(write_failed("raw_retention_commit")); } return std::result::Result::Ok(outcome); } async fn load_retention_archive_payload( sql_transaction: &deadpool_postgres::Transaction<'_>, signature_bytes: &[u8], ) -> std::result::Result>, crate::PostgresBackendError> { let row_result = sql_transaction.query_opt(GET_ARCHIVE_PAYLOAD_SQL, &[&signature_bytes]).await; let row = match row_result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_retention_archive_query")), }; let archive_row = match row { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Ok(std::option::Option::None), }; let payload_result = archive_row.try_get::<_, std::vec::Vec>("payload"); return match payload_result { std::result::Result::Ok(value) => std::result::Result::Ok(std::option::Option::Some(value)), std::result::Result::Err(_) => std::result::Result::Err(data_invalid("raw_retention_archive_decode")), }; } async fn purge_archived_transaction( sql_transaction: &deadpool_postgres::Transaction<'_>, signature_bytes: &[u8], ) -> std::result::Result<(), crate::PostgresBackendError> { let delete_result = sql_transaction.execute(DELETE_ARCHIVE_PAYLOAD_SQL, &[&signature_bytes]).await; match delete_result { std::result::Result::Ok(1) => {}, std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_retention_purge_archive_cardinality")), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_retention_purge_archive")), } let update_result = sql_transaction.execute(UPDATE_PURGED_TRANSACTION_SQL, &[&signature_bytes]).await; return match update_result { std::result::Result::Ok(1) => std::result::Result::Ok(()), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_retention_purge_update_cardinality")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_retention_purge_update")), }; } fn raw_retention_db_row(row: &tokio_postgres::Row) -> std::result::Result { let block_time_unix_millis = match row.try_get::<_, std::option::Option>("block_time_unix_millis") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_retention_transition_decode")), }; let payload = match row.try_get::<_, std::option::Option>>("payload") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_retention_transition_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_retention_transition_decode")), }; return std::result::Result::Ok(RawRetentionDbRow { block_time_unix_millis, payload, retention_state }); } fn retention_transition_decision( current: ksp_store_api::RawRetentionState, expected: ksp_store_api::RawRetentionState, target: ksp_store_api::RawRetentionState, ) -> std::result::Result { if current == target { return std::result::Result::Ok(RetentionTransitionDecision::AlreadyAtTarget); } if current != expected { return std::result::Result::Ok(RetentionTransitionDecision::ExpectedStateMismatch); } return match (expected, target) { (ksp_store_api::RawRetentionState::Full, ksp_store_api::RawRetentionState::Archived) => std::result::Result::Ok(RetentionTransitionDecision::Archive), (ksp_store_api::RawRetentionState::Archived, ksp_store_api::RawRetentionState::Purged) => std::result::Result::Ok(RetentionTransitionDecision::Purge), (ksp_store_api::RawRetentionState::Full, ksp_store_api::RawRetentionState::Compacted) | (ksp_store_api::RawRetentionState::Compacted, ksp_store_api::RawRetentionState::Archived) => { std::result::Result::Err(retention_compaction_unsupported("raw_retention_compaction")) }, _ => std::result::Result::Err(data_invalid("raw_retention_transition_unreachable")), }; } async fn validate_retention_shape( sql_transaction: &deadpool_postgres::Transaction<'_>, signature_bytes: &[u8], row: &RawRetentionDbRow, state: ksp_store_api::RawRetentionState, ) -> std::result::Result<(), crate::PostgresBackendError> { let archive_payload = match load_retention_archive_payload(sql_transaction, signature_bytes).await { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; return match state { ksp_store_api::RawRetentionState::Full => { let payload = match row.payload.as_ref() { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(data_invalid("raw_retention_full_shape")), }; let payload_result = validate_retained_payload_bytes(payload.as_slice(), "raw_retention_full_payload"); if let std::result::Result::Err(error) = payload_result { return std::result::Result::Err(error); } if archive_payload.is_some() { return std::result::Result::Err(data_invalid("raw_retention_full_archive_residue")); } std::result::Result::Ok(()) }, ksp_store_api::RawRetentionState::Archived => { if row.payload.is_some() { return std::result::Result::Err(data_invalid("raw_retention_archived_hot_payload")); } let payload = match archive_payload.as_ref() { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(data_invalid("raw_retention_archived_payload_missing")), }; validate_retained_payload_bytes(payload.as_slice(), "raw_retention_archived_payload") }, ksp_store_api::RawRetentionState::Purged => { if row.payload.is_some() || archive_payload.is_some() || row.block_time_unix_millis.is_some() { return std::result::Result::Err(data_invalid("raw_retention_purged_shape")); } std::result::Result::Ok(()) }, ksp_store_api::RawRetentionState::Compacted => std::result::Result::Err(retention_compaction_unsupported("raw_retention_compaction")), _ => std::result::Result::Err(data_invalid("raw_retention_state")), }; } fn validate_retained_payload_bytes(payload: &[u8], phase: &'static str) -> std::result::Result<(), crate::PostgresBackendError> { if payload.is_empty() || payload.len() > ksp_store_api::MAX_RAW_PAYLOAD_BYTES { return std::result::Result::Err(data_invalid(phase)); } return std::result::Result::Ok(()); } async fn ensure_v003_transaction_variant_bootstrap( sql_transaction: &deadpool_postgres::Transaction<'_>, row: &RawTransactionDbRow, created_at_unix_millis: i64, ) -> std::result::Result { let signature_bytes: &[u8] = row.signature.as_slice(); let selector_result = sql_transaction.query_opt(SELECT_CANONICAL_VARIANT_ID_SQL, &[&signature_bytes]).await; match selector_result { std::result::Result::Ok(std::option::Option::Some(selector_row)) => { let variant_id_text = match selector_row.try_get::<_, std::string::String>("canonical_variant_id_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_selector_id")), }; return decode_u64_decimal(variant_id_text.as_str(), "raw_variant_selector_id"); }, std::result::Result::Ok(std::option::Option::None) => {}, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_selector_query")), } 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 variant_payload: std::option::Option<&[u8]> = match retention_state { ksp_store_api::RawRetentionState::Full => { if row.archive_payload.is_some() { return std::result::Result::Err(data_invalid("raw_variant_bootstrap_full_archive")); } match row.payload.as_ref() { std::option::Option::Some(value) => std::option::Option::Some(value.as_slice()), std::option::Option::None => return std::result::Result::Err(data_invalid("raw_variant_bootstrap_full_payload")), } }, ksp_store_api::RawRetentionState::Archived => { if row.payload.is_some() || row.archive_payload.is_none() { return std::result::Result::Err(data_invalid("raw_variant_bootstrap_archived_shape")); } std::option::Option::None }, ksp_store_api::RawRetentionState::Purged => { if row.payload.is_some() || row.archive_payload.is_some() || row.block_time_unix_millis.is_some() { return std::result::Result::Err(data_invalid("raw_variant_bootstrap_purged_shape")); } std::option::Option::None }, _ => return std::result::Result::Err(data_invalid("raw_variant_bootstrap_retention_state")), }; let variant_id_text = "1"; let content_hash_bytes: &[u8] = row.content_hash.as_slice(); let insert_result = sql_transaction .execute( INSERT_TRANSACTION_VARIANT_SQL, &[ &signature_bytes, &variant_id_text, &row.slot_text.as_str(), &row.block_time_unix_millis, &row.format_id.as_str(), &row.format_version, &content_hash_bytes, &variant_payload, &row.retention_state.as_str(), &created_at_unix_millis, ], ) .await; match insert_result { std::result::Result::Ok(1) => {}, std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_variant_bootstrap_insert_count")), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_bootstrap_insert")), } let selector_insert_result = sql_transaction.execute(INSERT_TRANSACTION_VARIANT_SELECTOR_SQL, &[&signature_bytes, &variant_id_text, &created_at_unix_millis]).await; return match selector_insert_result { std::result::Result::Ok(1) => std::result::Result::Ok(1), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_variant_selector_insert_count")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_variant_selector_insert")), }; } async fn persist_or_reuse_native_transaction_variant( sql_transaction: &deadpool_postgres::Transaction<'_>, raw_transaction: &ksp_store_api::RawTransaction, created_at_unix_millis: i64, ) -> std::result::Result { let signature = raw_transaction.reference().signature(); let signature_bytes: &[u8] = signature.as_bytes(); let slot_text = raw_transaction.slot().to_string(); let block_time = match raw_transaction.block_time() { std::option::Option::Some(value) => match i64::try_from(value.unix_millis()) { std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_block_time")), }, std::option::Option::None => std::option::Option::None, }; let format_version = i64::from(raw_transaction.payload().format_version()); let content_hash = raw_transaction.payload().content_hash(); let content_hash_bytes: &[u8] = content_hash.as_bytes(); let payload_bytes = raw_transaction.payload().bytes(); let exact_result = sql_transaction .query_opt( SELECT_EXACT_NATIVE_VARIANT_SQL, &[ &signature_bytes, &slot_text.as_str(), &block_time, &raw_transaction.payload().format_id().as_str(), &format_version, &content_hash_bytes, &payload_bytes, ], ) .await; match exact_result { std::result::Result::Ok(std::option::Option::Some(row)) => { let variant_id_text = match row.try_get::<_, std::string::String>("variant_id_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_exact_id")), }; return decode_u64_decimal(variant_id_text.as_str(), "raw_variant_exact_id"); }, std::result::Result::Ok(std::option::Option::None) => {}, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_exact_query")), } let max_result = sql_transaction.query_one(SELECT_MAX_VARIANT_ID_SQL, &[&signature_bytes]).await; let max_row = match max_result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_max_query")), }; let max_variant_id_text = match max_row.try_get::<_, std::string::String>("variant_id_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_max_id")), }; let variant_id = match next_transaction_variant_id(max_variant_id_text.as_str()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let variant_id_text = variant_id.to_string(); let variant_payload = std::option::Option::Some(payload_bytes); let insert_result = sql_transaction .execute( INSERT_TRANSACTION_VARIANT_SQL, &[ &signature_bytes, &variant_id_text.as_str(), &slot_text.as_str(), &block_time, &raw_transaction.payload().format_id().as_str(), &format_version, &content_hash_bytes, &variant_payload, &"full", &created_at_unix_millis, ], ) .await; return match insert_result { std::result::Result::Ok(1) => std::result::Result::Ok(variant_id), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_variant_insert_count")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_variant_insert")), }; } async fn open_or_update_raw_transaction_conflict( sql_transaction: &deadpool_postgres::Transaction<'_>, reference: &ksp_store_api::RawTransactionReference, canonical_variant_id: u64, incoming_variant_id: u64, comparison: ksp_store_api::RawTransactionVariantComparison, observed_at_unix_millis: i64, ) -> std::result::Result { if canonical_variant_id == incoming_variant_id { return std::result::Result::Err(data_invalid("raw_conflict_variant_identity")); } if !matches!(comparison.relation(), ksp_store_api::RawTransactionVariantRelation::Conflict | ksp_store_api::RawTransactionVariantRelation::Incomparable) { return std::result::Result::Err(data_invalid("raw_conflict_relation")); } let signature = reference.signature(); let signature_bytes: &[u8] = signature.as_bytes(); let relation_code = comparison.relation().code(); let reason_code = comparison.reason().code(); let row_result = sql_transaction.query_opt(LOCK_TRANSACTION_CONFLICT_SQL, &[&signature_bytes]).await; let row = match row_result { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_conflict_lock")), }; let canonical_variant_id_text = canonical_variant_id.to_string(); let incoming_variant_id_text = incoming_variant_id.to_string(); if row.is_none() { let insert_result = sql_transaction .execute( INSERT_TRANSACTION_CONFLICT_SQL, &[ &signature_bytes, &canonical_variant_id_text.as_str(), &incoming_variant_id_text.as_str(), &relation_code, &reason_code, &observed_at_unix_millis, ], ) .await; return match insert_result { std::result::Result::Ok(1) => std::result::Result::Ok(RawTransactionConflictTransition { canonical_variant_id, incoming_variant_id, reason: comparison.reason(), relation: comparison.relation(), revision: 1, }), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_conflict_insert_count")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_conflict_insert")), }; } let row = match row { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(data_invalid("raw_conflict_lock_state")), }; let status = match row.try_get::<_, std::string::String>("status") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_conflict_status")), }; if status != ksp_store_api::RawTransactionConflictStatus::Open.code() && status != ksp_store_api::RawTransactionConflictStatus::Resolved.code() { return std::result::Result::Err(data_invalid("raw_conflict_status")); } let revision_text = match row.try_get::<_, std::string::String>("revision_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_conflict_revision")), }; let revision = match decode_u64_decimal(revision_text.as_str(), "raw_conflict_revision") { std::result::Result::Ok(value) if value > 0 => value, std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_conflict_revision")), std::result::Result::Err(error) => return std::result::Result::Err(error), }; let existing_canonical_text = match row.try_get::<_, std::string::String>("canonical_variant_id_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_conflict_canonical_variant_id")), }; let existing_canonical = match decode_u64_decimal(existing_canonical_text.as_str(), "raw_conflict_canonical_variant_id") { std::result::Result::Ok(value) if value > 0 => value, std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_conflict_canonical_variant_id")), std::result::Result::Err(error) => return std::result::Result::Err(error), }; let existing_incoming_text = match row.try_get::<_, std::string::String>("incoming_variant_id_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_conflict_incoming_variant_id")), }; let existing_incoming = match decode_u64_decimal(existing_incoming_text.as_str(), "raw_conflict_incoming_variant_id") { std::result::Result::Ok(value) if value > 0 => value, std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_conflict_incoming_variant_id")), std::result::Result::Err(error) => return std::result::Result::Err(error), }; let existing_relation = match row.try_get::<_, std::string::String>("latest_relation") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_conflict_relation")), }; let existing_reason = match row.try_get::<_, std::string::String>("latest_reason_code") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_conflict_reason")), }; if status == ksp_store_api::RawTransactionConflictStatus::Open.code() && existing_canonical == canonical_variant_id && existing_incoming == incoming_variant_id && existing_relation == relation_code && existing_reason == reason_code { return std::result::Result::Ok(RawTransactionConflictTransition { canonical_variant_id, incoming_variant_id, reason: comparison.reason(), relation: comparison.relation(), revision, }); } let updated_at = match row.try_get::<_, i64>("updated_at_unix_millis") { std::result::Result::Ok(value) => std::cmp::max(value, observed_at_unix_millis), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_conflict_updated_at")), }; let next_revision = match revision.checked_add(1) { std::option::Option::Some(value) if value != 0 => value, _ => return std::result::Result::Err(data_invalid("raw_conflict_revision_exhausted")), }; let next_revision_text = next_revision.to_string(); let update_result = sql_transaction .execute( UPDATE_TRANSACTION_CONFLICT_SQL, &[ &signature_bytes, &next_revision_text.as_str(), &canonical_variant_id_text.as_str(), &incoming_variant_id_text.as_str(), &relation_code, &reason_code, &updated_at, &revision_text.as_str(), ], ) .await; return match update_result { std::result::Result::Ok(1) => std::result::Result::Ok(RawTransactionConflictTransition { canonical_variant_id, incoming_variant_id, reason: comparison.reason(), relation: comparison.relation(), revision: next_revision, }), std::result::Result::Ok(0) => std::result::Result::Err(data_invalid("raw_conflict_stale_revision")), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_conflict_update_count")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_conflict_update")), }; } async fn promote_more_complete_transaction_variant( sql_transaction: &deadpool_postgres::Transaction<'_>, locked: &RawTransactionDbRow, incoming: &ksp_store_api::RawTransaction, expected_canonical_variant_id: u64, promoted_variant_id: u64, updated_at_unix_millis: i64, ) -> std::result::Result { if promoted_variant_id == expected_canonical_variant_id { return std::result::Result::Err(data_invalid("raw_variant_promotion_identity")); } let signature = incoming.reference().signature(); let signature_bytes: &[u8] = signature.as_bytes(); if locked.signature.as_slice() != signature_bytes { return std::result::Result::Err(data_invalid("raw_variant_promotion_reference")); } let selector_result = sql_transaction.query_opt(SELECT_CANONICAL_VARIANT_SELECTOR_SQL, &[&signature_bytes]).await; let selector_row = match selector_result { std::result::Result::Ok(std::option::Option::Some(value)) => value, std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Err(data_invalid("raw_variant_promotion_selector_missing")), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_promotion_selector_query")), }; let canonical_variant_id_text = match selector_row.try_get::<_, std::string::String>("canonical_variant_id_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_selector_id")), }; let current_canonical_variant_id = match decode_u64_decimal(canonical_variant_id_text.as_str(), "raw_variant_promotion_selector_id") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if current_canonical_variant_id != expected_canonical_variant_id { return std::result::Result::Err(data_invalid("raw_variant_promotion_selector_changed")); } let canonical_revision_text = match selector_row.try_get::<_, std::string::String>("canonical_revision_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_revision")), }; let current_revision = match decode_u64_decimal(canonical_revision_text.as_str(), "raw_variant_promotion_revision") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let next_revision = match next_transaction_canonical_revision(current_revision) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let preserve_result = preserve_previous_canonical_variant(sql_transaction, locked, expected_canonical_variant_id).await; if let std::result::Result::Err(error) = preserve_result { return std::result::Result::Err(error); } let archive_cleanup_result = sql_transaction.execute(DELETE_ARCHIVE_PAYLOAD_SQL, &[&signature_bytes]).await; match archive_cleanup_result { std::result::Result::Ok(count) if count <= 1 => {}, std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_archive_cleanup_count")), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_promotion_archive_cleanup")), } let slot_text = incoming.slot().to_string(); let block_time = match incoming.block_time() { std::option::Option::Some(value) => match i64::try_from(value.unix_millis()) { std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_block_time")), }, std::option::Option::None => std::option::Option::None, }; let format_version = i64::from(incoming.payload().format_version()); let content_hash = incoming.payload().content_hash(); let content_hash_bytes: &[u8] = content_hash.as_bytes(); let payload_bytes = incoming.payload().bytes(); let projection_result = sql_transaction .execute( UPDATE_CANONICAL_TRANSACTION_PROJECTION_SQL, &[ &signature_bytes, &slot_text.as_str(), &block_time, &incoming.payload().format_id().as_str(), &format_version, &content_hash_bytes, &payload_bytes, ], ) .await; match projection_result { std::result::Result::Ok(1) => {}, std::result::Result::Ok(_) => return std::result::Result::Err(data_invalid("raw_variant_promotion_projection_count")), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_promotion_projection")), } let expected_variant_id_text = expected_canonical_variant_id.to_string(); let expected_revision_text = current_revision.to_string(); let promoted_variant_id_text = promoted_variant_id.to_string(); let next_revision_text = next_revision.to_string(); let selector_update_result = sql_transaction .execute( UPDATE_TRANSACTION_VARIANT_SELECTOR_SQL, &[ &signature_bytes, &expected_variant_id_text.as_str(), &expected_revision_text.as_str(), &promoted_variant_id_text.as_str(), &next_revision_text.as_str(), &updated_at_unix_millis, ], ) .await; return match selector_update_result { std::result::Result::Ok(1) => std::result::Result::Ok(RawCanonicalPromotionTransition { from_revision: current_revision, from_variant_id: expected_canonical_variant_id, to_revision: next_revision, to_variant_id: promoted_variant_id, }), std::result::Result::Ok(0) => std::result::Result::Err(data_invalid("raw_variant_promotion_stale_selector")), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_variant_promotion_selector_count")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_variant_promotion_selector_update")), }; } async fn preserve_previous_canonical_variant( sql_transaction: &deadpool_postgres::Transaction<'_>, locked: &RawTransactionDbRow, canonical_variant_id: u64, ) -> std::result::Result<(), crate::PostgresBackendError> { let state = match decode_retention_state(locked.retention_state.as_str()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let preserved_payload = if state == ksp_store_api::RawRetentionState::Full { if locked.archive_payload.is_some() { return std::result::Result::Err(data_invalid("raw_variant_promotion_full_shape")); } match locked.payload.as_ref() { std::option::Option::Some(value) => value.as_slice(), std::option::Option::None => return std::result::Result::Err(data_invalid("raw_variant_promotion_full_payload")), } } else if state == ksp_store_api::RawRetentionState::Archived { if locked.payload.is_some() { return std::result::Result::Err(data_invalid("raw_variant_promotion_archived_shape")); } match locked.archive_payload.as_ref() { std::option::Option::Some(value) => value.as_slice(), std::option::Option::None => return std::result::Result::Err(data_invalid("raw_variant_promotion_archived_payload")), } } else { return std::result::Result::Err(data_invalid("raw_variant_promotion_retention_state")); }; let signature_bytes: &[u8] = locked.signature.as_slice(); let canonical_variant_id_text = canonical_variant_id.to_string(); let content_hash_bytes: &[u8] = locked.content_hash.as_slice(); let update_result = sql_transaction .execute( UPDATE_TRANSACTION_VARIANT_TO_FULL_SQL, &[ &signature_bytes, &canonical_variant_id_text.as_str(), &locked.slot_text.as_str(), &locked.block_time_unix_millis, &locked.format_id.as_str(), &locked.format_version, &content_hash_bytes, &preserved_payload, ], ) .await; return match update_result { std::result::Result::Ok(1) => std::result::Result::Ok(()), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_variant_promotion_preserve_count")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_variant_promotion_preserve")), }; } fn next_transaction_canonical_revision(current: u64) -> std::result::Result { return match current.checked_add(1) { std::option::Option::Some(value) if value != 0 => std::result::Result::Ok(value), _ => std::result::Result::Err(data_invalid("raw_variant_canonical_revision_exhausted")), }; } async fn persist_transaction_variant_link( sql_transaction: &deadpool_postgres::Transaction<'_>, observation: &ksp_store_api::RawTransactionObservation, variant_id: u64, observation_outcome: ksp_store_api::RawObservationWriteOutcome, linked_at_unix_millis: i64, ) -> std::result::Result<(), crate::PostgresBackendError> { let observation_key = observation.observation_key(); let observation_key_bytes: &[u8] = observation_key.as_bytes(); let signature = observation.transaction().signature(); let signature_bytes: &[u8] = signature.as_bytes(); let variant_id_text = variant_id.to_string(); if observation_outcome == ksp_store_api::RawObservationWriteOutcome::Inserted { let insert_result = sql_transaction .query_opt(INSERT_TRANSACTION_VARIANT_LINK_SQL, &[&observation_key_bytes, &signature_bytes, &variant_id_text.as_str(), &linked_at_unix_millis]) .await; return match insert_result { std::result::Result::Ok(std::option::Option::Some(_)) => std::result::Result::Ok(()), std::result::Result::Ok(std::option::Option::None) => std::result::Result::Err(data_invalid("raw_variant_link_insert_count")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_variant_link_insert")), }; } if observation_outcome != ksp_store_api::RawObservationWriteOutcome::AlreadyPresent { return std::result::Result::Err(data_invalid("raw_variant_link_observation_outcome")); } let existing_result = sql_transaction.query_opt(SELECT_TRANSACTION_VARIANT_LINK_SQL, &[&observation_key_bytes]).await; let existing = match existing_result { std::result::Result::Ok(std::option::Option::Some(value)) => value, std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Ok(()), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_variant_link_query")), }; let existing_signature = match existing.try_get::<_, std::vec::Vec>("transaction_signature") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_link_signature")), }; let existing_variant_id_text = match existing.try_get::<_, std::string::String>("variant_id_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_link_id")), }; let existing_variant_id = match decode_u64_decimal(existing_variant_id_text.as_str(), "raw_variant_link_id") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; if existing_signature.as_slice() != signature_bytes || existing_variant_id != variant_id { return std::result::Result::Err(conflict("raw_variant_link_conflict")); } return std::result::Result::Ok(()); } async fn rehydrate_canonical_transaction_variant( sql_transaction: &deadpool_postgres::Transaction<'_>, raw_transaction: &ksp_store_api::RawTransaction, canonical_variant_id: u64, ) -> std::result::Result<(), crate::PostgresBackendError> { let signature = raw_transaction.reference().signature(); let signature_bytes: &[u8] = signature.as_bytes(); let variant_id_text = canonical_variant_id.to_string(); let block_time = match raw_transaction.block_time() { std::option::Option::Some(value) => match i64::try_from(value.unix_millis()) { std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_variant_rehydrate_block_time")), }, std::option::Option::None => std::option::Option::None, }; let payload_bytes = raw_transaction.payload().bytes(); let update_result = sql_transaction .execute(UPDATE_REHYDRATED_TRANSACTION_VARIANT_SQL, &[&signature_bytes, &variant_id_text.as_str(), &block_time, &payload_bytes]) .await; return match update_result { std::result::Result::Ok(1) => std::result::Result::Ok(()), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_variant_rehydrate_count")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_variant_rehydrate")), }; } fn next_transaction_variant_id(current_max: &str) -> std::result::Result { let current = match decode_u64_decimal(current_max, "raw_variant_max_id") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; return match current.checked_add(1) { std::option::Option::Some(value) if value != 0 => std::result::Result::Ok(value), _ => std::result::Result::Err(data_invalid("raw_variant_id_exhausted")), }; } fn raw_variant_timestamp(timestamp: ksp_store_api::RawTimestamp, phase: &'static str) -> std::result::Result { return match i64::try_from(timestamp.unix_millis()) { std::result::Result::Ok(value) => std::result::Result::Ok(value), std::result::Result::Err(_) => std::result::Result::Err(data_invalid(phase)), }; } async fn insert_canonical_transaction( sql_transaction: &deadpool_postgres::Transaction<'_>, raw_transaction: &ksp_store_api::RawTransaction, ) -> std::result::Result { let signature = raw_transaction.reference().signature(); let signature_bytes: &[u8] = signature.as_bytes(); let slot_text = raw_transaction.slot().to_string(); let block_time = match raw_transaction.block_time() { std::option::Option::Some(value) => match i64::try_from(value.unix_millis()) { std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_acquisition_block_time")), }, std::option::Option::None => std::option::Option::None, }; let format_version = i64::from(raw_transaction.payload().format_version()); let content_hash = raw_transaction.payload().content_hash(); let content_hash_bytes: &[u8] = content_hash.as_bytes(); let payload_bytes = raw_transaction.payload().bytes(); let row_result = sql_transaction .query_opt( INSERT_TRANSACTION_SQL, &[ &signature_bytes, &slot_text.as_str(), &block_time, &raw_transaction.payload().format_id().as_str(), &format_version, &content_hash_bytes, &payload_bytes, ], ) .await; return match row_result { std::result::Result::Ok(std::option::Option::Some(_)) => std::result::Result::Ok(true), std::result::Result::Ok(std::option::Option::None) => std::result::Result::Ok(false), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_acquisition_insert_transaction")), }; } async fn load_locked_transaction_row( sql_transaction: &deadpool_postgres::Transaction<'_>, reference: &ksp_store_api::RawTransactionReference, ) -> std::result::Result, crate::PostgresBackendError> { let signature = reference.signature(); let signature_bytes: &[u8] = signature.as_bytes(); let row_result = sql_transaction.query_opt(LOCK_TRANSACTION_SQL, &[&signature_bytes]).await; let row = match row_result { std::result::Result::Ok(std::option::Option::Some(value)) => value, std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Ok(std::option::Option::None), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_acquisition_lock_transaction")), }; let mut physical = match raw_transaction_db_row(&row) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let archive_result = sql_transaction.query_opt(GET_ARCHIVE_PAYLOAD_SQL, &[&signature_bytes]).await; physical.archive_payload = match archive_result { std::result::Result::Ok(std::option::Option::Some(archive_row)) => match archive_row.try_get::<_, std::vec::Vec>("payload") { std::result::Result::Ok(value) => std::option::Option::Some(value), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_acquisition_archive_decode")), }, std::result::Result::Ok(std::option::Option::None) => std::option::Option::None, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_acquisition_archive_query")), }; return std::result::Result::Ok(std::option::Option::Some(physical)); } fn compare_existing_transaction( network: &ksp_store_api::RawNetworkId, row: RawTransactionDbRow, incoming: &ksp_store_api::RawTransaction, ) -> std::result::Result { 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 { if row.payload.is_some() || row.archive_payload.is_some() || row.block_time_unix_millis.is_some() { return std::result::Result::Err(data_invalid("raw_acquisition_purged_shape")); } 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_acquisition_purged_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_acquisition_purged_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 matches = reference.eq(incoming.reference()) && slot == incoming.slot() && format_id.as_str() == incoming.payload().format_id().as_str() && format_version == incoming.payload().format_version() && content_hash == incoming.payload().content_hash(); if matches { return std::result::Result::Ok(ExistingTransactionMatch::Purged); } return std::result::Result::Err(conflict("raw_acquisition_purged_conflict")); } let stored_result = decode_raw_transaction_row(network, row); let stored = match stored_result { std::result::Result::Ok(std::option::Option::Some(value)) => value, std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Err(data_invalid("raw_acquisition_active_shape")), std::result::Result::Err(error) => return std::result::Result::Err(error), }; let comparison = match ksp_store_api::compare_raw_transaction_variants(&stored, incoming) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_acquisition_variant_comparison")), }; return match comparison.relation() { ksp_store_api::RawTransactionVariantRelation::Exact => std::result::Result::Ok(ExistingTransactionMatch::Active), ksp_store_api::RawTransactionVariantRelation::CompatibleLessComplete => { log_raw_transaction_compatible_truncated_log_messages(network, &stored, incoming); std::result::Result::Ok(ExistingTransactionMatch::ActiveCompatibleLessComplete) }, ksp_store_api::RawTransactionVariantRelation::CompatibleMoreComplete => std::result::Result::Ok(ExistingTransactionMatch::ActiveCompatibleMoreComplete), ksp_store_api::RawTransactionVariantRelation::Conflict | ksp_store_api::RawTransactionVariantRelation::Incomparable => { std::result::Result::Ok(ExistingTransactionMatch::ActiveConflict(comparison)) }, _ => std::result::Result::Err(data_invalid("raw_acquisition_variant_relation")), }; } fn raw_transaction_content_conflict_diagnostic( stored: &ksp_store_api::RawTransaction, incoming: &ksp_store_api::RawTransaction, ) -> RawTransactionContentConflictDiagnostic { let payload = raw_payload_content_conflict_diagnostic(stored.payload().bytes(), incoming.payload().bytes()); return RawTransactionContentConflictDiagnostic { block_time_mismatch: stored.block_time() != incoming.block_time(), content_hash_mismatch: stored.payload().content_hash() != incoming.payload().content_hash(), format_id_mismatch: stored.payload().format_id() != incoming.payload().format_id(), format_version_mismatch: stored.payload().format_version() != incoming.payload().format_version(), log_messages: payload.log_messages, meta_mismatch_fields: payload.meta_mismatch_fields, payload_bytes_mismatch: stored.payload().bytes() != incoming.payload().bytes(), payload_diagnostic_available: payload.available, payload_meta_mismatch: payload.meta_mismatch, payload_other_mismatch: payload.other_mismatch, payload_transaction_index_mismatch: payload.transaction_index_mismatch, payload_transaction_mismatch: payload.transaction_mismatch, payload_version_mismatch: payload.version_mismatch, slot_mismatch: stored.slot() != incoming.slot(), }; } fn raw_payload_content_conflict_diagnostic(stored: &[u8], incoming: &[u8]) -> RawPayloadContentConflictDiagnostic { let stored = serde_json::from_slice::(stored); let incoming = serde_json::from_slice::(incoming); let (stored, incoming) = match (stored, incoming) { (std::result::Result::Ok(serde_json::Value::Object(stored)), std::result::Result::Ok(serde_json::Value::Object(incoming))) => (stored, incoming), _ => { return RawPayloadContentConflictDiagnostic { available: false, log_messages: raw_log_messages_content_conflict_diagnostic(std::option::Option::None, std::option::Option::None), meta_mismatch: false, meta_mismatch_fields: "unavailable".to_owned(), other_mismatch: false, transaction_index_mismatch: false, transaction_mismatch: false, version_mismatch: false, }; }, }; let transaction_mismatch = stored.get("transaction") != incoming.get("transaction"); let meta_mismatch = stored.get("meta") != incoming.get("meta"); let version_mismatch = stored.get("version") != incoming.get("version"); let transaction_index_mismatch = stored.get("transactionIndex") != incoming.get("transactionIndex"); let other_mismatch = json_object_other_fields_mismatch(&stored, &incoming, RAW_TRANSACTION_CONTENT_CONFLICT_PAYLOAD_FIELDS.as_slice()); let meta_mismatch_fields = raw_meta_content_conflict_fields(stored.get("meta"), incoming.get("meta")); let log_messages = raw_log_messages_content_conflict_diagnostic(stored.get("meta"), incoming.get("meta")); return RawPayloadContentConflictDiagnostic { available: true, log_messages, meta_mismatch, meta_mismatch_fields, other_mismatch, transaction_index_mismatch, transaction_mismatch, version_mismatch, }; } fn raw_log_messages_content_conflict_diagnostic( stored_meta: std::option::Option<&serde_json::Value>, incoming_meta: std::option::Option<&serde_json::Value>, ) -> RawLogMessagesContentConflictDiagnostic { let stored = raw_log_messages_value(stored_meta); let incoming = raw_log_messages_value(incoming_meta); let stored_state = raw_log_messages_state(stored); let incoming_state = raw_log_messages_state(incoming); let (stored_values, incoming_values) = match (stored, incoming) { (std::option::Option::Some(serde_json::Value::Array(stored)), std::option::Option::Some(serde_json::Value::Array(incoming))) => { (stored.as_slice(), incoming.as_slice()) }, _ => { return RawLogMessagesContentConflictDiagnostic { available: false, first_mismatch_index: "unavailable".to_owned(), incoming_count: raw_log_messages_count(incoming), incoming_first_kind: "unavailable", incoming_first_length: 0, incoming_has_truncation_marker: raw_log_messages_has_truncation_marker(incoming), incoming_prefix_of_stored: false, incoming_state, stored_count: raw_log_messages_count(stored), stored_first_kind: "unavailable", stored_first_length: 0, stored_has_truncation_marker: raw_log_messages_has_truncation_marker(stored), stored_prefix_of_incoming: false, stored_state, }; }, }; let shared_len = stored_values.len().min(incoming_values.len()); let first_mismatch = stored_values.iter().zip(incoming_values.iter()).position(|(stored, incoming)| return stored != incoming).or_else(|| { if stored_values.len() == incoming_values.len() { return std::option::Option::None; } return std::option::Option::Some(shared_len); }); let (stored_first_kind, stored_first_length) = raw_log_message_kind_and_length(first_mismatch.and_then(|index| return stored_values.get(index))); let (incoming_first_kind, incoming_first_length) = raw_log_message_kind_and_length(first_mismatch.and_then(|index| return incoming_values.get(index))); let first_mismatch_index = match first_mismatch { std::option::Option::Some(value) => value.to_string(), std::option::Option::None => "none".to_owned(), }; return RawLogMessagesContentConflictDiagnostic { available: true, first_mismatch_index, incoming_count: incoming_values.len(), incoming_first_kind, incoming_first_length, incoming_has_truncation_marker: raw_log_message_array_has_truncation_marker(incoming_values), incoming_prefix_of_stored: raw_json_array_is_prefix(incoming_values, stored_values), incoming_state, stored_count: stored_values.len(), stored_first_kind, stored_first_length, stored_has_truncation_marker: raw_log_message_array_has_truncation_marker(stored_values), stored_prefix_of_incoming: raw_json_array_is_prefix(stored_values, incoming_values), stored_state, }; } fn raw_log_messages_value(meta: std::option::Option<&serde_json::Value>) -> std::option::Option<&serde_json::Value> { return match meta { std::option::Option::Some(serde_json::Value::Object(meta)) => meta.get("logMessages"), _ => std::option::Option::None, }; } fn raw_log_messages_state(value: std::option::Option<&serde_json::Value>) -> &'static str { return match value { std::option::Option::None => "missing", std::option::Option::Some(serde_json::Value::Null) => "null", std::option::Option::Some(serde_json::Value::Array(_)) => "array", std::option::Option::Some(_) => "other", }; } fn raw_log_messages_count(value: std::option::Option<&serde_json::Value>) -> usize { return match value { std::option::Option::Some(serde_json::Value::Array(values)) => values.len(), _ => 0, }; } fn raw_json_array_is_prefix(prefix: &[serde_json::Value], values: &[serde_json::Value]) -> bool { if prefix.len() > values.len() { return false; } return prefix.iter().zip(values.iter()).all(|(left, right)| return left == right); } fn raw_log_message_kind_and_length(value: std::option::Option<&serde_json::Value>) -> (&'static str, usize) { let line = match value { std::option::Option::Some(serde_json::Value::String(value)) => value.as_str(), std::option::Option::Some(_) => return ("non_string", 0), std::option::Option::None => return ("missing", 0), }; let kind = if line.starts_with("Log truncated") || line.starts_with("log truncated") { "log_truncated" } else if line.starts_with("Program log:") { "program_log" } else if line.starts_with("Program data:") { "program_data" } else if line.starts_with("Program ") && line.contains(" invoke [") { "program_invoke" } else if line.starts_with("Program ") && line.ends_with(" success") { "program_success" } else if line.starts_with("Program ") && line.contains(" failed:") { "program_failed" } else if line.starts_with("Program ") && line.contains(" consumed ") && line.contains(" compute units") { "compute_units" } else { "other" }; return (kind, line.len()); } fn raw_log_messages_has_truncation_marker(value: std::option::Option<&serde_json::Value>) -> bool { return match value { std::option::Option::Some(serde_json::Value::Array(values)) => raw_log_message_array_has_truncation_marker(values.as_slice()), _ => false, }; } fn raw_log_message_array_has_truncation_marker(values: &[serde_json::Value]) -> bool { return values.iter().any(|value| { return match value { serde_json::Value::String(line) => line.starts_with("Log truncated") || line.starts_with("log truncated"), _ => false, }; }); } fn raw_meta_content_conflict_fields(stored: std::option::Option<&serde_json::Value>, incoming: std::option::Option<&serde_json::Value>) -> std::string::String { if stored == incoming { return "none".to_owned(); } let (stored, incoming) = match (stored, incoming) { (std::option::Option::Some(serde_json::Value::Object(stored)), std::option::Option::Some(serde_json::Value::Object(incoming))) => (stored, incoming), _ => return "shape".to_owned(), }; let mut fields = std::vec::Vec::new(); for field in RAW_TRANSACTION_CONTENT_CONFLICT_META_FIELDS { if stored.get(field) != incoming.get(field) { fields.push(field); } } if json_object_other_fields_mismatch(stored, incoming, RAW_TRANSACTION_CONTENT_CONFLICT_META_FIELDS.as_slice()) { fields.push("other"); } if fields.is_empty() { return "shape".to_owned(); } return fields.join(","); } fn json_object_other_fields_mismatch( stored: &serde_json::Map, incoming: &serde_json::Map, known_fields: &[&str], ) -> bool { let stored_other = stored.iter().filter(|(key, _)| return !known_fields.contains(&key.as_str())).collect::>(); let incoming_other = incoming.iter().filter(|(key, _)| return !known_fields.contains(&key.as_str())).collect::>(); return stored_other != incoming_other; } fn log_raw_transaction_compatible_truncated_log_messages( network: &ksp_store_api::RawNetworkId, stored: &ksp_store_api::RawTransaction, incoming: &ksp_store_api::RawTransaction, ) { let diagnostic = raw_transaction_content_conflict_diagnostic(stored, incoming); ksp_logging_lib::debug!( target: crate::TRACING_TARGET, domain = "store.raw_transaction.compatible_partial", network = network.as_str(), slot = stored.slot(), stored_log_messages_count = diagnostic.log_messages.stored_count, incoming_log_messages_count = diagnostic.log_messages.incoming_count, incoming_log_messages_has_truncation_marker = diagnostic.log_messages.incoming_has_truncation_marker, "PostgreSQL Store accepted compatible truncated RAW transaction logs without replacing canonical content" ); return; } fn log_raw_transaction_canonical_promotion_committed(network: &ksp_store_api::RawNetworkId, slot: u64, transition: RawCanonicalPromotionTransition) { ksp_logging_lib::info!( target: crate::TRACING_TARGET, domain = "store.raw_transaction.canonical_promotion", network = network.as_str(), slot = slot, from_variant_id = transition.from_variant_id, to_variant_id = transition.to_variant_id, from_revision = transition.from_revision, to_revision = transition.to_revision, "PostgreSQL Store committed a proved RAW transaction canonical promotion" ); return; } fn log_raw_transaction_conflict_quarantined_committed(network: &ksp_store_api::RawNetworkId, slot: u64, transition: RawTransactionConflictTransition) { ksp_logging_lib::warn!( target: crate::TRACING_TARGET, domain = "store.raw_transaction.quarantined_conflict", network = network.as_str(), slot = slot, canonical_variant_id = transition.canonical_variant_id, incoming_variant_id = transition.incoming_variant_id, conflict_revision = transition.revision, relation = transition.relation.code(), reason = transition.reason.code(), "PostgreSQL Store committed a durable RAW transaction conflict case" ); return; } fn log_raw_transaction_content_conflict( network: &ksp_store_api::RawNetworkId, stored: &ksp_store_api::RawTransaction, incoming: &ksp_store_api::RawTransaction, ) { let diagnostic = raw_transaction_content_conflict_diagnostic(stored, incoming); ksp_logging_lib::warn!( target: crate::TRACING_TARGET, domain = "store.raw_transaction.content_conflict", network = network.as_str(), stored_slot = stored.slot(), incoming_slot = incoming.slot(), slot_mismatch = diagnostic.slot_mismatch, block_time_mismatch = diagnostic.block_time_mismatch, format_id_mismatch = diagnostic.format_id_mismatch, format_version_mismatch = diagnostic.format_version_mismatch, content_hash_mismatch = diagnostic.content_hash_mismatch, payload_bytes_mismatch = diagnostic.payload_bytes_mismatch, payload_diagnostic_available = diagnostic.payload_diagnostic_available, transaction_mismatch = diagnostic.payload_transaction_mismatch, meta_mismatch = diagnostic.payload_meta_mismatch, version_mismatch = diagnostic.payload_version_mismatch, transaction_index_mismatch = diagnostic.payload_transaction_index_mismatch, payload_other_mismatch = diagnostic.payload_other_mismatch, meta_mismatch_fields = diagnostic.meta_mismatch_fields.as_str(), log_messages_diagnostic_available = diagnostic.log_messages.available, stored_log_messages_state = diagnostic.log_messages.stored_state, incoming_log_messages_state = diagnostic.log_messages.incoming_state, stored_log_messages_count = diagnostic.log_messages.stored_count, incoming_log_messages_count = diagnostic.log_messages.incoming_count, log_messages_first_mismatch_index = diagnostic.log_messages.first_mismatch_index.as_str(), stored_log_messages_first_kind = diagnostic.log_messages.stored_first_kind, incoming_log_messages_first_kind = diagnostic.log_messages.incoming_first_kind, stored_log_messages_first_length = diagnostic.log_messages.stored_first_length, incoming_log_messages_first_length = diagnostic.log_messages.incoming_first_length, stored_log_messages_prefix_of_incoming = diagnostic.log_messages.stored_prefix_of_incoming, incoming_log_messages_prefix_of_stored = diagnostic.log_messages.incoming_prefix_of_stored, stored_log_messages_has_truncation_marker = diagnostic.log_messages.stored_has_truncation_marker, incoming_log_messages_has_truncation_marker = diagnostic.log_messages.incoming_has_truncation_marker, "PostgreSQL Store classified divergent canonical RAW transaction content" ); return; } async fn log_raw_transaction_content_conflict_provenance( sql_transaction: &deadpool_postgres::Transaction<'_>, network: &ksp_store_api::RawNetworkId, incoming_observation: &ksp_store_api::RawTransactionObservation, ) { let signature = incoming_observation.transaction().signature(); let signature_bytes: &[u8] = signature.as_bytes(); let stored_observation = match sql_transaction.query_opt(GET_FIRST_TRANSACTION_OBSERVATION_SQL, &[&signature_bytes]).await { std::result::Result::Ok(std::option::Option::Some(row)) => match raw_observation_db_row(&row) { std::result::Result::Ok(row) => decode_raw_observation_row(network, row).ok(), std::result::Result::Err(_) => std::option::Option::None, }, std::result::Result::Ok(std::option::Option::None) | std::result::Result::Err(_) => std::option::Option::None, }; let incoming = incoming_observation.provenance(); let incoming_endpoint = incoming.endpoint_id().map(|value| return value.as_str()).unwrap_or("none"); let incoming_commitment = incoming.commitment().map(|value| return value.as_str()).unwrap_or("none"); let stored_available = stored_observation.is_some(); let (stored_provider, stored_protocol, stored_method, stored_endpoint, stored_commitment) = match stored_observation.as_ref() { std::option::Option::Some(observation) => { let provenance = observation.provenance(); ( provenance.provider().as_str(), provenance.protocol().as_str(), provenance.acquisition_method().as_str(), provenance.endpoint_id().map(|value| return value.as_str()).unwrap_or("none"), provenance.commitment().map(|value| return value.as_str()).unwrap_or("none"), ) }, std::option::Option::None => ("unknown", "unknown", "unknown", "unknown", "unknown"), }; ksp_logging_lib::warn!( target: crate::TRACING_TARGET, domain = "store.raw_transaction.content_conflict", network = network.as_str(), stored_observation_available = stored_available, stored_provider = stored_provider, stored_protocol = stored_protocol, stored_acquisition_method = stored_method, stored_endpoint_id = stored_endpoint, stored_commitment = stored_commitment, incoming_provider = incoming.provider().as_str(), incoming_protocol = incoming.protocol().as_str(), incoming_acquisition_method = incoming.acquisition_method().as_str(), incoming_endpoint_id = incoming_endpoint, incoming_commitment = incoming_commitment, "PostgreSQL Store classified canonical RAW transaction conflict provenance" ); return; } async fn rehydrate_transaction( sql_transaction: &deadpool_postgres::Transaction<'_>, raw_transaction: &ksp_store_api::RawTransaction, ) -> std::result::Result<(), crate::PostgresBackendError> { let signature = raw_transaction.reference().signature(); let signature_bytes: &[u8] = signature.as_bytes(); let block_time = match raw_transaction.block_time() { std::option::Option::Some(value) => match i64::try_from(value.unix_millis()) { std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_rehydrate_block_time")), }, std::option::Option::None => std::option::Option::None, }; let payload_bytes = raw_transaction.payload().bytes(); let update_result = sql_transaction.execute(REHYDRATE_TRANSACTION_SQL, &[&signature_bytes, &block_time, &payload_bytes]).await; return match update_result { std::result::Result::Ok(1) => std::result::Result::Ok(()), std::result::Result::Ok(_) => std::result::Result::Err(data_invalid("raw_rehydrate_cardinality")), std::result::Result::Err(_) => std::result::Result::Err(write_failed("raw_rehydrate_update")), }; } async fn persist_observation_row( sql_transaction: &deadpool_postgres::Transaction<'_>, network: &ksp_store_api::RawNetworkId, observation: &ksp_store_api::RawTransactionObservation, ) -> std::result::Result { let provenance = observation.provenance(); let observation_key = observation.observation_key(); let observation_key_bytes: &[u8] = observation_key.as_bytes(); let signature = observation.transaction().signature(); let signature_bytes: &[u8] = signature.as_bytes(); let origin = match encode_origin(provenance.origin()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let received_at = match i64::try_from(provenance.received_at().unix_millis()) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_observation_received_at_encode")), }; let observed_at = match provenance.observed_at() { std::option::Option::Some(value) => match i64::try_from(value.unix_millis()) { std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_observation_observed_at_encode")), }, std::option::Option::None => std::option::Option::None, }; let source_payload_size = match provenance.source_payload_size_bytes() { std::option::Option::Some(value) => match i64::try_from(value) { std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_observation_source_size_encode")), }, std::option::Option::None => std::option::Option::None, }; let capture_session_id = provenance.capture_session_id().map(|value| return value.as_str()); let commitment = provenance.commitment().map(|value| return value.as_str()); let endpoint_id = provenance.endpoint_id().map(|value| return value.as_str()); let filter_id = provenance.filter_id().map(|value| return value.as_str()); let source_payload_hash = provenance.source_payload_hash(); let source_payload_hash_bytes: std::option::Option<&[u8]> = source_payload_hash.as_ref().map(|value| return &value.as_bytes()[..]); let insert_result = sql_transaction .query_opt( INSERT_OBSERVATION_SQL, &[ &observation_key_bytes, &signature_bytes, &provenance.provider().as_str(), &provenance.protocol().as_str(), &provenance.acquisition_method().as_str(), &origin, &received_at, &capture_session_id, &commitment, &endpoint_id, &filter_id, &observed_at, &source_payload_hash_bytes, &source_payload_size, ], ) .await; let inserted = match insert_result { std::result::Result::Ok(std::option::Option::Some(_)) => true, std::result::Result::Ok(std::option::Option::None) => false, std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_observation_insert")), }; if inserted { return std::result::Result::Ok(ksp_store_api::RawObservationWriteOutcome::Inserted); } let existing_result = sql_transaction.query_opt(LOCK_OBSERVATION_SQL, &[&observation_key_bytes]).await; let existing_row = match existing_result { std::result::Result::Ok(std::option::Option::Some(value)) => value, std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Err(data_invalid("raw_observation_conflict_missing")), std::result::Result::Err(_) => return std::result::Result::Err(write_failed("raw_observation_conflict_query")), }; let physical = match raw_observation_db_row(&existing_row) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let stored = 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 stored.eq(observation) { return std::result::Result::Ok(ksp_store_api::RawObservationWriteOutcome::AlreadyPresent); } return std::result::Result::Err(conflict("raw_observation_content_conflict")); } fn ensure_acquisition_inputs( network: &ksp_store_api::RawNetworkId, raw_transaction: &ksp_store_api::RawTransaction, observation: &ksp_store_api::RawTransactionObservation, ) -> std::result::Result<(), crate::PostgresBackendError> { let transaction_network_result = ensure_network(network, raw_transaction.reference(), "raw_acquisition_transaction_network"); if let std::result::Result::Err(error) = transaction_network_result { return std::result::Result::Err(error); } let observation_network_result = ensure_network(network, observation.transaction(), "raw_acquisition_observation_network"); if let std::result::Result::Err(error) = observation_network_result { return std::result::Result::Err(error); } if observation.transaction() != raw_transaction.reference() { return std::result::Result::Err(conflict("raw_acquisition_reference_mismatch")); } return std::result::Result::Ok(()); } fn ensure_retention_transition_inputs( network: &ksp_store_api::RawNetworkId, transition: &ksp_store_api::RawTransactionRetentionTransition, ) -> std::result::Result<(), crate::PostgresBackendError> { let network_result = ensure_network(network, transition.reference(), "raw_retention_transition_network"); if let std::result::Result::Err(error) = network_result { return std::result::Result::Err(error); } if transition.expected() == ksp_store_api::RawRetentionState::Compacted || transition.target() == ksp_store_api::RawRetentionState::Compacted { return std::result::Result::Err(retention_compaction_unsupported("raw_retention_compaction")); } return std::result::Result::Ok(()); } fn encode_origin(origin: ksp_store_api::RawAcquisitionOrigin) -> std::result::Result<&'static str, crate::PostgresBackendError> { return match origin { ksp_store_api::RawAcquisitionOrigin::Backfill => std::result::Result::Ok("backfill"), ksp_store_api::RawAcquisitionOrigin::Import => std::result::Result::Ok("import"), ksp_store_api::RawAcquisitionOrigin::Live => std::result::Result::Ok("live"), ksp_store_api::RawAcquisitionOrigin::Repair => std::result::Result::Ok("repair"), ksp_store_api::RawAcquisitionOrigin::Replay => std::result::Result::Ok("replay"), _ => std::result::Result::Err(data_invalid("raw_observation_origin_encode")), }; } fn conflict(phase: &'static str) -> crate::PostgresBackendError { return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::Conflict, phase); } fn retention_compaction_unsupported(phase: &'static str) -> crate::PostgresBackendError { return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::RetentionCompactionUnsupported, phase); } fn reference_not_found(phase: &'static str) -> crate::PostgresBackendError { return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::ReferenceNotFound, phase); } fn write_failed(phase: &'static str) -> crate::PostgresBackendError { return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::WriteFailed, phase); } fn raw_inspection_empty_page_is_clean(row: &RawInspectionDbRow) -> bool { return !row.page_present && row.archive_payload_present.is_none() && row.block_time_unix_millis.is_none() && row.content_hash.is_none() && row.format_id.is_none() && row.format_version.is_none() && row.hot_payload_present.is_none() && row.payload_size_bytes.is_none() && row.retention_state.is_none() && row.signature.is_none() && row.slot_text.is_none(); } fn raw_transaction_inspection_sql_window(page: ksp_store_api::RawInspectionPageRequest) -> std::result::Result<(i64, i64), crate::PostgresBackendError> { let sql_limit = match i64::try_from(page.limit().get()) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => { return std::result::Result::Err(crate::PostgresBackendError::new( crate::PostgresBackendErrorKind::PageLimitUnsupported, "raw_transaction_inspection_limit", )); }, }; let sql_offset = match i64::try_from(page.offset()) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => { return std::result::Result::Err(crate::PostgresBackendError::new( crate::PostgresBackendErrorKind::QueryInvalid, "raw_transaction_inspection_offset", )); }, }; return std::result::Result::Ok((sql_limit, sql_offset)); } fn raw_inspection_db_row(row: &tokio_postgres::Row) -> std::result::Result { let archive_payload_present = match row.try_get::<_, std::option::Option>("archive_payload_present") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_inspection_decode")), }; let block_time_unix_millis = match row.try_get::<_, std::option::Option>("block_time_unix_millis") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_inspection_decode")), }; let content_hash = match row.try_get::<_, std::option::Option>>("content_hash") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_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_inspection_decode")), }; let format_id = match row.try_get::<_, std::option::Option>("format_id") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_inspection_decode")), }; let format_version = match row.try_get::<_, std::option::Option>("format_version") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_inspection_decode")), }; let hot_payload_present = match row.try_get::<_, std::option::Option>("hot_payload_present") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_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_inspection_decode")), }; let payload_size_bytes = match row.try_get::<_, std::option::Option>("payload_size_bytes") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_inspection_decode")), }; let retention_state = match row.try_get::<_, std::option::Option>("retention_state") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_inspection_decode")), }; let signature = match row.try_get::<_, std::option::Option>>("signature") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_inspection_decode")), }; let slot_text = match row.try_get::<_, std::option::Option>("slot_text") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_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_inspection_decode")), }; return std::result::Result::Ok(RawInspectionDbRow { archive_payload_present, block_time_unix_millis, content_hash, filtered_count_text, format_id, format_version, hot_payload_present, page_present, payload_size_bytes, retention_state, signature, slot_text, total_count_text, }); } fn decode_raw_inspection_summary( network: &ksp_store_api::RawNetworkId, row: RawInspectionDbRow, ) -> std::result::Result { if !row.page_present { return std::result::Result::Err(data_invalid("raw_transaction_inspection_page")); } let signature = match inspection_required(row.signature, "raw_transaction_inspection_signature") { std::result::Result::Ok(value) => match fixed_bytes::<64>(value) { std::result::Result::Ok(bytes) => ksp_store_api::RawTransactionSignature::new(bytes), std::result::Result::Err(error) => return std::result::Result::Err(error), }, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let slot_text = match inspection_required(row.slot_text, "raw_transaction_inspection_slot") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let slot = match decode_u64_decimal(slot_text.as_str(), "raw_transaction_inspection_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_inspection_block_time") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let format_id_raw = match inspection_required(row.format_id, "raw_transaction_inspection_format_id") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let format_id = match decode_format_id(format_id_raw) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let format_version_raw = match inspection_required(row.format_version, "raw_transaction_inspection_format_version") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let format_version = match decode_u32_i64(format_version_raw, "raw_transaction_inspection_format_version") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let content_hash_raw = match inspection_required(row.content_hash, "raw_transaction_inspection_content_hash") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let content_hash = match fixed_bytes::<32>(content_hash_raw) { std::result::Result::Ok(value) => ksp_store_api::RawContentHash::new(value), std::result::Result::Err(error) => return std::result::Result::Err(error), }; let retention_raw = match inspection_required(row.retention_state, "raw_transaction_inspection_retention") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let retention_state = match decode_retention_state(retention_raw.as_str()) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let hot_payload_present = match inspection_required(row.hot_payload_present, "raw_transaction_inspection_hot_payload") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let archive_payload_present = match inspection_required(row.archive_payload_present, "raw_transaction_inspection_archive_payload") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let payload_size_bytes = match row.payload_size_bytes { std::option::Option::Some(value) => match u64::try_from(value) { std::result::Result::Ok(decoded) => std::option::Option::Some(decoded), std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_inspection_payload_size")), }, std::option::Option::None => std::option::Option::None, }; match retention_state { ksp_store_api::RawRetentionState::Full if !hot_payload_present || archive_payload_present => { return std::result::Result::Err(data_invalid("raw_transaction_inspection_full_shape")); }, ksp_store_api::RawRetentionState::Archived if hot_payload_present || !archive_payload_present => { return std::result::Result::Err(data_invalid("raw_transaction_inspection_archived_shape")); }, ksp_store_api::RawRetentionState::Purged if hot_payload_present || archive_payload_present || block_time.is_some() => { return std::result::Result::Err(data_invalid("raw_transaction_inspection_purged_shape")); }, ksp_store_api::RawRetentionState::Full | ksp_store_api::RawRetentionState::Archived | ksp_store_api::RawRetentionState::Purged => {}, _ => return std::result::Result::Err(data_invalid("raw_transaction_inspection_retention")), } let reference = ksp_store_api::RawTransactionReference::new(network.clone(), signature); return match ksp_store_api::RawTransactionSummary::try_new( reference, slot, block_time, format_id, format_version, content_hash, payload_size_bytes, retention_state, ) { std::result::Result::Ok(value) => std::result::Result::Ok(value), std::result::Result::Err(_) => std::result::Result::Err(data_invalid("raw_transaction_inspection_summary")), }; } fn inspection_required(value: std::option::Option, phase: &'static str) -> std::result::Result { return match value { std::option::Option::Some(inner) => std::result::Result::Ok(inner), std::option::Option::None => std::result::Result::Err(data_invalid(phase)), }; } fn raw_list_db_row(row: &tokio_postgres::Row) -> std::result::Result { let signature = match row.try_get::<_, std::vec::Vec>("signature") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(data_invalid("raw_transaction_list_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_list_decode")), }; return std::result::Result::Ok(RawListDbRow { signature, slot_text }); } fn raw_transaction_db_row(row: &tokio_postgres::Row) -> std::result::Result { let signature = match row.try_get::<_, std::vec::Vec>("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>("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>("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>>("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>>("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_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 { 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), 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), source_payload_size_bytes: get_optional!("source_payload_size_bytes", i64), total_count_text, transaction_signature: get_optional!("transaction_signature", std::vec::Vec), }); } fn decode_raw_observation_inspection_row( network: &ksp_store_api::RawNetworkId, row: RawObservationInspectionDbRow, ) -> std::result::Result { 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 { 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), commitment: required!("commitment", std::option::Option), endpoint_id: required!("endpoint_id", std::option::Option), filter_id: required!("filter_id", std::option::Option), observation_key: required!("observation_key", std::vec::Vec), observed_at_unix_millis: required!("observed_at_unix_millis", std::option::Option), 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>), source_payload_size_bytes: required!("source_payload_size_bytes", std::option::Option), transaction_signature: required!("transaction_signature", std::vec::Vec), }); } fn raw_tombstone_db_row(row: &tokio_postgres::Row) -> std::result::Result { let signature = match row.try_get::<_, std::vec::Vec>("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>("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>("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_list_row( network: &ksp_store_api::RawNetworkId, row: RawListDbRow, ) -> std::result::Result<(u64, ksp_store_api::RawTransactionReference), 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_list_slot") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; let reference = ksp_store_api::RawTransactionReference::new(network.clone(), signature); return std::result::Result::Ok((slot, reference)); } fn decode_raw_transaction_row( network: &ksp_store_api::RawNetworkId, row: RawTransactionDbRow, ) -> std::result::Result, 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 { 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, 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 { 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 { 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 { 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 { 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, phase: &'static str, ) -> std::result::Result, 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 { 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 { 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 { return match value.parse::() { 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 { 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(value: std::vec::Vec) -> 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;