v0.3.3-pre.008

2026-08-30 14:10:10 +02:00
parent 1f0b202135
commit 84ab2b3651
21 changed files with 816 additions and 62 deletions
--- a/crates/ksp-store-lib/src/store.rs
+++ b/crates/ksp-store-lib/src/store.rs
@@ -1,5 +1,5 @@
 // file: crates/ksp-store-lib/src/store.rs
-// version: 5
+// version: 6

 /// Opaque common Store runtime facade.
 ///
@@ -116,6 +116,261 @@ impl std::fmt::Debug for Store {
    }
 }

+impl ksp_store_api::RawTransactionRead for Store {
+    fn get_raw_transaction<'a>(
+        &'a self,
+        reference: &'a ksp_store_api::RawTransactionReference,
+    ) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawTransaction>>> {
+        let network_check = validate_operation_network(&self.network, reference.network(), self.backend_kind);
+        if let std::result::Result::Err(error) = network_check {
+            return std::boxed::Box::pin(async move {
+                return std::result::Result::Err(error);
+            });
+        }
+        return std::boxed::Box::pin(async move {
+            #[cfg(feature = "postgres")]
+            {
+                return match &self.runtime {
+                    StoreRuntime::Postgres(backend) => {
+                        let result = backend.get_raw_transaction(reference).await;
+                        result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
+                    },
+                };
+            }
+            #[cfg(not(feature = "postgres"))]
+            {
+                let _ = reference;
+                return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
+            }
+        });
+    }
+
+    fn list_raw_transactions<'a>(
+        &'a self,
+        query: &'a ksp_store_api::RawTransactionQuery,
+    ) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawPage<ksp_store_api::RawTransactionReference>>> {
+        let network_check = validate_operation_network(&self.network, query.network(), self.backend_kind);
+        if let std::result::Result::Err(error) = network_check {
+            return std::boxed::Box::pin(async move {
+                return std::result::Result::Err(error);
+            });
+        }
+        return std::boxed::Box::pin(async move {
+            #[cfg(feature = "postgres")]
+            {
+                return match &self.runtime {
+                    StoreRuntime::Postgres(backend) => {
+                        let result = backend.list_raw_transactions(query).await;
+                        result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
+                    },
+                };
+            }
+            #[cfg(not(feature = "postgres"))]
+            {
+                let _ = query;
+                return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
+            }
+        });
+    }
+}
+
+impl ksp_store_api::RawTransactionWrite for Store {
+    fn persist_raw_transaction_acquisition<'a>(
+        &'a self,
+        transaction: ksp_store_api::RawTransaction,
+        observation: ksp_store_api::RawTransactionObservation,
+        mode: ksp_store_api::RawTransactionAcquisitionMode,
+    ) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawAcquisitionWriteOutcome>> {
+        let transaction_network = validate_operation_network(&self.network, transaction.reference().network(), self.backend_kind);
+        if let std::result::Result::Err(error) = transaction_network {
+            return std::boxed::Box::pin(async move {
+                return std::result::Result::Err(error);
+            });
+        }
+        let observation_network = validate_operation_network(&self.network, observation.transaction().network(), self.backend_kind);
+        if let std::result::Result::Err(error) = observation_network {
+            return std::boxed::Box::pin(async move {
+                return std::result::Result::Err(error);
+            });
+        }
+        return std::boxed::Box::pin(async move {
+            #[cfg(feature = "postgres")]
+            {
+                return match &self.runtime {
+                    StoreRuntime::Postgres(backend) => {
+                        let result = backend.persist_raw_transaction_acquisition(transaction, observation, mode).await;
+                        result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
+                    },
+                };
+            }
+            #[cfg(not(feature = "postgres"))]
+            {
+                let _ = transaction;
+                let _ = observation;
+                let _ = mode;
+                return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
+            }
+        });
+    }
+}
+
+impl ksp_store_api::RawTransactionObservationRead for Store {
+    fn get_raw_transaction_observation<'a>(
+        &'a self,
+        observation_key: &'a ksp_store_api::RawObservationKey,
+    ) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawTransactionObservation>>> {
+        return std::boxed::Box::pin(async move {
+            #[cfg(feature = "postgres")]
+            {
+                return match &self.runtime {
+                    StoreRuntime::Postgres(backend) => {
+                        let result = backend.get_raw_transaction_observation(observation_key).await;
+                        result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
+                    },
+                };
+            }
+            #[cfg(not(feature = "postgres"))]
+            {
+                let _ = observation_key;
+                return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
+            }
+        });
+    }
+}
+
+impl ksp_store_api::RawTransactionObservationWrite for Store {
+    fn record_raw_transaction_observation<'a>(
+        &'a self,
+        observation: ksp_store_api::RawTransactionObservation,
+    ) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawObservationWriteOutcome>> {
+        let network_check = validate_operation_network(&self.network, observation.transaction().network(), self.backend_kind);
+        if let std::result::Result::Err(error) = network_check {
+            return std::boxed::Box::pin(async move {
+                return std::result::Result::Err(error);
+            });
+        }
+        return std::boxed::Box::pin(async move {
+            #[cfg(feature = "postgres")]
+            {
+                return match &self.runtime {
+                    StoreRuntime::Postgres(backend) => {
+                        let result = backend.record_raw_transaction_observation(observation).await;
+                        result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
+                    },
+                };
+            }
+            #[cfg(not(feature = "postgres"))]
+            {
+                let _ = observation;
+                return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
+            }
+        });
+    }
+}
+
+impl ksp_store_api::RawTransactionRetentionRead for Store {
+    fn get_raw_transaction_retention_state<'a>(
+        &'a self,
+        reference: &'a ksp_store_api::RawTransactionReference,
+    ) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawRetentionState>>> {
+        let network_check = validate_operation_network(&self.network, reference.network(), self.backend_kind);
+        if let std::result::Result::Err(error) = network_check {
+            return std::boxed::Box::pin(async move {
+                return std::result::Result::Err(error);
+            });
+        }
+        return std::boxed::Box::pin(async move {
+            #[cfg(feature = "postgres")]
+            {
+                return match &self.runtime {
+                    StoreRuntime::Postgres(backend) => {
+                        let result = backend.get_raw_transaction_retention_state(reference).await;
+                        result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
+                    },
+                };
+            }
+            #[cfg(not(feature = "postgres"))]
+            {
+                let _ = reference;
+                return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
+            }
+        });
+    }
+
+    fn get_raw_transaction_tombstone<'a>(
+        &'a self,
+        reference: &'a ksp_store_api::RawTransactionReference,
+    ) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<std::option::Option<ksp_store_api::RawTransactionTombstone>>> {
+        let network_check = validate_operation_network(&self.network, reference.network(), self.backend_kind);
+        if let std::result::Result::Err(error) = network_check {
+            return std::boxed::Box::pin(async move {
+                return std::result::Result::Err(error);
+            });
+        }
+        return std::boxed::Box::pin(async move {
+            #[cfg(feature = "postgres")]
+            {
+                return match &self.runtime {
+                    StoreRuntime::Postgres(backend) => {
+                        let result = backend.get_raw_transaction_tombstone(reference).await;
+                        result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
+                    },
+                };
+            }
+            #[cfg(not(feature = "postgres"))]
+            {
+                let _ = reference;
+                return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
+            }
+        });
+    }
+}
+
+impl ksp_store_api::RawTransactionRetentionWrite for Store {
+    fn transition_raw_transaction_retention<'a>(
+        &'a self,
+        transition: ksp_store_api::RawTransactionRetentionTransition,
+    ) -> ksp_store_api::StoreApiFuture<'a, ksp_store_api::Result<ksp_store_api::RawRetentionWriteOutcome>> {
+        let network_check = validate_operation_network(&self.network, transition.reference().network(), self.backend_kind);
+        if let std::result::Result::Err(error) = network_check {
+            return std::boxed::Box::pin(async move {
+                return std::result::Result::Err(error);
+            });
+        }
+        return std::boxed::Box::pin(async move {
+            #[cfg(feature = "postgres")]
+            {
+                return match &self.runtime {
+                    StoreRuntime::Postgres(backend) => {
+                        let result = backend.transition_raw_transaction_retention(transition).await;
+                        result.map_err(|error| return map_postgres_error(error, self.backend_kind, self.network.as_str()))
+                    },
+                };
+            }
+            #[cfg(not(feature = "postgres"))]
+            {
+                let _ = transition;
+                return std::result::Result::Err(unavailable_runtime_error(self.backend_kind));
+            }
+        });
+    }
+}
+
+fn validate_operation_network(
+    store_network: &ksp_store_api::RawNetworkId,
+    operation_network: &ksp_store_api::RawNetworkId,
+    backend_kind: crate::StoreBackendKind,
+) -> ksp_store_api::Result<()> {
+    if store_network != operation_network {
+        return std::result::Result::Err(
+            ksp_store_api::Error::new(crate::ERROR_CODE_WRONG_NETWORK, "Store operation targeted a different logical network")
+                .with_context("backend", backend_kind.code())
+                .with_context("network", store_network.as_str()),
+        );
+    }
+    return std::result::Result::Ok(());
+}
+
 #[cfg(feature = "postgres")]
 enum StoreRuntime {
    Postgres(ksp_store_postgres_lib::PostgresBackend),
@@ -176,7 +431,7 @@ async fn open_postgres(
 #[cfg(feature = "postgres")]
 fn map_postgres_error(error: ksp_store_postgres_lib::PostgresBackendError, backend_kind: crate::StoreBackendKind, network: &str) -> ksp_store_api::Error {
    let code = postgres_error_code(error.kind());
-    return ksp_store_api::Error::new(code, "PostgreSQL Store backend lifecycle operation failed")
+    return ksp_store_api::Error::new(code, "PostgreSQL Store backend operation failed")
        .with_context("backend", backend_kind.code())
        .with_context("network", network)
        .with_context("phase", error.phase());
@@ -215,13 +470,22 @@ fn postgres_error_code(kind: ksp_store_postgres_lib::PostgresBackendErrorKind) -
    return match kind {
        ksp_store_postgres_lib::PostgresBackendErrorKind::ConfigInvalid => crate::ERROR_CODE_POSTGRES_CONFIG_INVALID,
        ksp_store_postgres_lib::PostgresBackendErrorKind::ConnectFailed => crate::ERROR_CODE_POSTGRES_CONNECT_FAILED,
+        ksp_store_postgres_lib::PostgresBackendErrorKind::Conflict => ksp_store_api::ERROR_CODE_RAW_CONFLICT,
+        ksp_store_postgres_lib::PostgresBackendErrorKind::DataInvalid => crate::ERROR_CODE_POSTGRES_DATA_INVALID,
        ksp_store_postgres_lib::PostgresBackendErrorKind::HealthFailed => crate::ERROR_CODE_POSTGRES_HEALTH_FAILED,
        ksp_store_postgres_lib::PostgresBackendErrorKind::PoolTimeout => crate::ERROR_CODE_POSTGRES_POOL_TIMEOUT,
        ksp_store_postgres_lib::PostgresBackendErrorKind::MigrationFailed => crate::ERROR_CODE_POSTGRES_MIGRATION_FAILED,
        ksp_store_postgres_lib::PostgresBackendErrorKind::MigrationMismatch => crate::ERROR_CODE_POSTGRES_MIGRATION_MISMATCH,
+        ksp_store_postgres_lib::PostgresBackendErrorKind::PageLimitUnsupported => crate::ERROR_CODE_POSTGRES_PAGE_LIMIT_UNSUPPORTED,
+        ksp_store_postgres_lib::PostgresBackendErrorKind::QueryInvalid => ksp_store_api::ERROR_CODE_RAW_QUERY_INVALID,
+        ksp_store_postgres_lib::PostgresBackendErrorKind::ReadFailed => crate::ERROR_CODE_POSTGRES_READ_FAILED,
+        ksp_store_postgres_lib::PostgresBackendErrorKind::ReferenceNotFound => crate::ERROR_CODE_RAW_REFERENCE_NOT_FOUND,
+        ksp_store_postgres_lib::PostgresBackendErrorKind::RetentionCompactionUnsupported => crate::ERROR_CODE_POSTGRES_RETENTION_COMPACTION_UNSUPPORTED,
        ksp_store_postgres_lib::PostgresBackendErrorKind::SchemaNewer => crate::ERROR_CODE_POSTGRES_SCHEMA_NEWER,
        ksp_store_postgres_lib::PostgresBackendErrorKind::ShutdownTimeout => crate::ERROR_CODE_SHUTDOWN_TIMEOUT,
        ksp_store_postgres_lib::PostgresBackendErrorKind::TlsFailed => crate::ERROR_CODE_POSTGRES_TLS_FAILED,
+        ksp_store_postgres_lib::PostgresBackendErrorKind::WriteFailed => crate::ERROR_CODE_POSTGRES_WRITE_FAILED,
+        ksp_store_postgres_lib::PostgresBackendErrorKind::WrongNetwork => crate::ERROR_CODE_WRONG_NETWORK,
        _ => crate::ERROR_CODE_BACKEND_OPEN_FAILED,
    };
 }