diff --git a/crates/sdk-axum/src/api.rs b/crates/sdk-axum/src/api.rs index 1b5617d..dee27c5 100644 --- a/crates/sdk-axum/src/api.rs +++ b/crates/sdk-axum/src/api.rs @@ -19,6 +19,7 @@ use dataplane_sdk::{ core::{ db::tx::TransactionalContext, model::{ + control_plane::ControlPlane, data_flow::DataFlowState, messages::{ DataFlowPrepareMessage, DataFlowResumeMessage, DataFlowStartMessage, @@ -37,24 +38,26 @@ use crate::error::SignalingResult; pub async fn start_flow( State(sdk): State>, Extension(participant): Extension, + Extension(control_plane): Extension, Json(msg): Json, ) -> SignalingResult> where C: TransactionalContext, { - let response = sdk.start(&participant.id, msg).await?; + let response = sdk.start(&participant.id, &control_plane.id, msg).await?; Ok(Json(response)) } pub async fn prepare_flow( State(sdk): State>, Extension(participant): Extension, + Extension(control_plane): Extension, Json(msg): Json, ) -> SignalingResult<(StatusCode, Json)> where C: TransactionalContext, { - let response = sdk.prepare(&participant.id, msg).await?; + let response = sdk.prepare(&participant.id, &control_plane.id, msg).await?; let status = match response.state { DataFlowState::Preparing => StatusCode::OK, diff --git a/crates/sdk-axum/src/api_tests.rs b/crates/sdk-axum/src/api_tests.rs index 78c40f9..eaa5d2e 100644 --- a/crates/sdk-axum/src/api_tests.rs +++ b/crates/sdk-axum/src/api_tests.rs @@ -19,12 +19,13 @@ use dataplane_sdk::sdk; use dataplane_sdk::{ core::{ db::{ + control_plane::ControlPlaneRepo, data_flow::DataFlowRepo, tx::{Transaction, TransactionalContext}, }, error::DbResult, handler::DataFlowHandler, - model::data_flow::DataFlow, + model::{control_plane::ControlPlane, data_flow::DataFlow}, }, sdk::DataPlaneSdk, }; @@ -75,6 +76,21 @@ mock! { } } +mock! { + ControlPlaneRepo{} + + #[async_trait] + impl ControlPlaneRepo for ControlPlaneRepo { + type Transaction = MockTx; + async fn create(&self, tx: &mut MockTx, control_plane: &ControlPlane) -> DbResult<()>; + async fn fetch_by_id( + &self, + tx: &mut MockTx, + control_plane_id: &str, + ) -> DbResult>; + } +} + mock! { Handler {} @@ -123,15 +139,24 @@ struct TestCtx { base_url: String, app: Router>, participant_context: ParticipantContext, + control_plane: ControlPlane, } impl TestCtx { pub fn app(&self, sdk: DataPlaneSdk) -> Router { let app = self.app.clone().with_state(sdk); app.layer(axum::Extension(self.participant_context.clone())) + .layer(axum::Extension(self.control_plane.clone())) } } +fn control_plane() -> ControlPlane { + ControlPlane::builder() + .id("control_plane_id") + .url("http://localhost/callback") + .build() +} + fn single_ctx() -> TestCtx { TestCtx { base_url: "/api/v1".to_string(), @@ -139,6 +164,7 @@ fn single_ctx() -> TestCtx { participant_context: ParticipantContext::builder() .id("example-participant") .build(), + control_plane: control_plane(), } } @@ -149,6 +175,7 @@ fn multi_participant_ctx() -> TestCtx { participant_context: ParticipantContext::builder() .id("example-participant") .build(), + control_plane: control_plane(), } } @@ -159,6 +186,7 @@ fn context() -> (MockTxContext, MockRepo, MockHandler) { fn sdk(ctx: MockTxContext, repo: MockRepo, handler: MockHandler) -> DataPlaneSdk { DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap() @@ -216,9 +244,8 @@ mod start { .build(), ) .agreement_id("agreement_id") - .transfer_type("transfer_type") + .profile("transfer_type") .dataspace_context("dataspace_context") - .callback_address("callback_address") .message_id("message_id") .counter_party_id("counter_party_id") .build(); @@ -284,8 +311,8 @@ mod terminate { .agreement_id("agreement_id") .dataspace_context("dataspace_context") .participant_id("participant_id") - .callback_address("callback_address") - .transfer_type("transfer_type") + .control_plane_id("control_plane_id") + .profile("transfer_type") .kind(DataFlowType::Provider) .build(), )) @@ -365,8 +392,8 @@ mod suspend { .agreement_id("agreement_id") .dataspace_context("dataspace_context") .participant_id("participant_id") - .callback_address("callback_address") - .transfer_type("transfer_type") + .control_plane_id("control_plane_id") + .profile("transfer_type") .kind(DataFlowType::Provider) .build(), )) @@ -451,9 +478,8 @@ mod prepare { .participant_id("counter_party_id") .process_id("process_id") .agreement_id("agreement_id") - .transfer_type("transfer_type") + .profile("transfer_type") .dataspace_context("dataspace_context") - .callback_address("callback_address") .message_id("message_id") .counter_party_id("counter_party_id") .build(); @@ -523,8 +549,8 @@ mod started { .agreement_id("agreement_id") .dataspace_context("dataspace_context") .participant_id("participant_id") - .callback_address("callback_address") - .transfer_type("transfer_type") + .control_plane_id("control_plane_id") + .profile("transfer_type") .kind(DataFlowType::Provider) .build(), )) @@ -593,8 +619,8 @@ mod completed { .agreement_id("agreement_id") .dataspace_context("dataspace_context") .participant_id("participant_id") - .callback_address("callback_address") - .transfer_type("transfer_type") + .control_plane_id("control_plane_id") + .profile("transfer_type") .kind(DataFlowType::Provider) .build(), )) @@ -661,8 +687,8 @@ mod flow_status { .agreement_id("agreement_id") .dataspace_context("dataspace_context") .participant_id("participant_id") - .callback_address("callback_address") - .transfer_type("transfer_type") + .control_plane_id("control_plane_id") + .profile("transfer_type") .kind(DataFlowType::Provider) .build(), )) @@ -737,8 +763,8 @@ mod resume { .agreement_id("agreement_id") .dataspace_context("dataspace_context") .participant_id("participant_id") - .callback_address("callback_address") - .transfer_type("transfer_type") + .control_plane_id("control_plane_id") + .profile("transfer_type") .kind(DataFlowType::Provider) .build(), )) diff --git a/crates/sdk-axum/src/error.rs b/crates/sdk-axum/src/error.rs index 0cc88fe..02bb524 100644 --- a/crates/sdk-axum/src/error.rs +++ b/crates/sdk-axum/src/error.rs @@ -43,6 +43,7 @@ impl IntoResponse for SignalingError { (StatusCode::CONFLICT, err.to_string()) } SignalingError::Sdk(e) => { + dbg!(&e); error!("SDK error: {:#}", e); (StatusCode::INTERNAL_SERVER_ERROR, "SDK error".to_owned()) } diff --git a/crates/sdk-postgres/migrations/20260218164640_create_data_flows.sql b/crates/sdk-postgres/migrations/20260218164640_create_data_flows.sql index 7de62f6..8a140ce 100644 --- a/crates/sdk-postgres/migrations/20260218164640_create_data_flows.sql +++ b/crates/sdk-postgres/migrations/20260218164640_create_data_flows.sql @@ -5,13 +5,13 @@ CREATE TYPE data_flow_type AS ENUM ('consumer','provider'); CREATE TABLE IF NOT EXISTS data_flows ( id TEXT PRIMARY KEY, participant_context_id TEXT NOT NULL, - transfer_type TEXT NOT NULL, + profile TEXT NOT NULL, agreement_id TEXT NOT NULL, dataset_id TEXT NOT NULL, participant_id TEXT NOT NULL, dataspace_context TEXT NOT NULL, counter_party_id TEXT NOT NULL, - callback_address TEXT NOT NULL, + control_plane_id TEXT NOT NULL, state data_flow_state NOT NULL, type data_flow_type NOT NULL, data_address JSONB , diff --git a/crates/sdk-postgres/migrations/20260218170000_create_control_planes.sql b/crates/sdk-postgres/migrations/20260218170000_create_control_planes.sql new file mode 100644 index 0000000..3adb3bf --- /dev/null +++ b/crates/sdk-postgres/migrations/20260218170000_create_control_planes.sql @@ -0,0 +1,4 @@ +CREATE TABLE IF NOT EXISTS control_planes ( + id TEXT PRIMARY KEY, + url TEXT NOT NULL +) diff --git a/crates/sdk-postgres/src/control_plane.rs b/crates/sdk-postgres/src/control_plane.rs new file mode 100644 index 0000000..87de511 --- /dev/null +++ b/crates/sdk-postgres/src/control_plane.rs @@ -0,0 +1,16 @@ +// Copyright (c) 2026 Metaform Systems, Inc +// +// This program and the accompanying materials are made available under the +// terms of the Apache License, Version 2.0 which is available at +// https://www.apache.org/licenses/LICENSE-2.0 +// +// SPDX-License-Identifier: Apache-2.0 +// +// Contributors: +// Metaform Systems, Inc. - initial API and implementation +// + +mod model; +mod repo; + +pub use repo::PgControlPlaneRepo; diff --git a/crates/sdk-postgres/src/control_plane/model.rs b/crates/sdk-postgres/src/control_plane/model.rs new file mode 100644 index 0000000..d034c1f --- /dev/null +++ b/crates/sdk-postgres/src/control_plane/model.rs @@ -0,0 +1,28 @@ +// Copyright (c) 2026 Metaform Systems, Inc +// +// This program and the accompanying materials are made available under the +// terms of the Apache License, Version 2.0 which is available at +// https://www.apache.org/licenses/LICENSE-2.0 +// +// SPDX-License-Identifier: Apache-2.0 +// +// Contributors: +// Metaform Systems, Inc. - initial API and implementation +// + +use sqlx::FromRow; + +#[derive(Clone, Debug, FromRow, PartialEq)] +pub struct ControlPlane { + pub id: String, + pub url: String, +} + +impl From for dataplane_sdk::core::model::control_plane::ControlPlane { + fn from(control_plane: ControlPlane) -> Self { + Self::builder() + .id(control_plane.id) + .url(control_plane.url) + .build() + } +} diff --git a/crates/sdk-postgres/src/control_plane/repo.rs b/crates/sdk-postgres/src/control_plane/repo.rs new file mode 100644 index 0000000..4a06664 --- /dev/null +++ b/crates/sdk-postgres/src/control_plane/repo.rs @@ -0,0 +1,85 @@ +// Copyright (c) 2026 Metaform Systems, Inc +// +// This program and the accompanying materials are made available under the +// terms of the Apache License, Version 2.0 which is available at +// https://www.apache.org/licenses/LICENSE-2.0 +// +// SPDX-License-Identifier: Apache-2.0 +// +// Contributors: +// Metaform Systems, Inc. - initial API and implementation +// + +use sqlx::Error; + +use dataplane_sdk::core::{ + db::control_plane::ControlPlaneRepo, + error::{DbError, DbResult}, + model::control_plane::ControlPlane, +}; + +use crate::{PgTransaction, control_plane::model::ControlPlane as DbControlPlane}; + +#[derive(Default)] +pub struct PgControlPlaneRepo; + +#[async_trait::async_trait] +impl ControlPlaneRepo for PgControlPlaneRepo { + type Transaction = PgTransaction; + + async fn create( + &self, + tx: &mut Self::Transaction, + control_plane: &ControlPlane, + ) -> DbResult<()> { + let result = sqlx::query( + r#" + INSERT INTO control_planes (id, url) + VALUES ($1, $2) + "#, + ) + .bind(&control_plane.id) + .bind(&control_plane.url) + .execute(&mut *tx.0) + .await; + + match result { + Ok(_) => Ok(()), + Err(Error::Database(db)) if db.is_unique_violation() => Err(DbError::AlreadyExists( + format!("Control plane with id {} already exists", control_plane.id), + )), + Err(err) => Err(DbError::Generic(Box::new(err))), + } + } + + async fn fetch_by_id( + &self, + tx: &mut Self::Transaction, + control_plane_id: &str, + ) -> DbResult> { + Ok(sqlx::query_as::<_, DbControlPlane>( + r#" + SELECT * FROM control_planes where id = $1 + "#, + ) + .bind(control_plane_id) + .fetch_optional(&mut *tx.0) + .await + .map_err(|err| DbError::Generic(Box::new(err)))? + .map(|control_plane| control_plane.into())) + } +} + +impl PgControlPlaneRepo { + pub async fn migrate(&self, tx: &mut PgTransaction) -> DbResult<()> { + let mut migrator = sqlx::migrate!("./migrations"); + migrator.set_ignore_missing(true); + + migrator + .run(&mut *tx.0) + .await + .map_err(|err| DbError::Generic(Box::new(err)))?; + + Ok(()) + } +} diff --git a/crates/sdk-postgres/src/data_flow/model.rs b/crates/sdk-postgres/src/data_flow/model.rs index 6978d14..99bf2cb 100644 --- a/crates/sdk-postgres/src/data_flow/model.rs +++ b/crates/sdk-postgres/src/data_flow/model.rs @@ -25,14 +25,14 @@ pub struct DataFlow { pub participant_context_id: String, pub counter_party_id: String, pub state: DataFlowState, - pub transfer_type: String, + pub profile: String, #[sqlx(rename = "type")] pub kind: DataFlowType, pub agreement_id: String, pub dataset_id: String, pub dataspace_context: String, pub participant_id: String, - pub callback_address: String, + pub control_plane_id: String, pub suspension_reason: Option, pub termination_reason: Option, #[builder(default)] @@ -81,8 +81,8 @@ impl From for dataplane_sdk::core::model::data_flow::DataFlow { .dataset_id(flow.dataset_id) .dataspace_context(flow.dataspace_context) .participant_id(flow.participant_id) - .callback_address(flow.callback_address) - .transfer_type(flow.transfer_type) + .control_plane_id(flow.control_plane_id) + .profile(flow.profile) .kind(flow.kind.into()) .created_at(flow.created_at) .updated_at(flow.updated_at) diff --git a/crates/sdk-postgres/src/data_flow/repo.rs b/crates/sdk-postgres/src/data_flow/repo.rs index 6d3eb86..0576541 100644 --- a/crates/sdk-postgres/src/data_flow/repo.rs +++ b/crates/sdk-postgres/src/data_flow/repo.rs @@ -33,7 +33,7 @@ impl DataFlowRepo for PgDataFlowRepo { async fn create(&self, tx: &mut Self::Transaction, flow: &DataFlow) -> DbResult<()> { let result = sqlx::query( r#" - INSERT INTO data_flows (id, participant_id, dataspace_context, participant_context_id, counter_party_id, dataset_id, agreement_id, state, transfer_type, type, data_address, callback_address, labels, metadata, created_at, updated_at) + INSERT INTO data_flows (id, participant_id, dataspace_context, participant_context_id, counter_party_id, dataset_id, agreement_id, state, profile, type, data_address, control_plane_id, labels, metadata, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) "#, ) @@ -45,10 +45,10 @@ impl DataFlowRepo for PgDataFlowRepo { .bind(&flow.dataset_id) .bind(&flow.agreement_id) .bind(DataFlowState::from(flow.state.clone())) - .bind(&flow.transfer_type) + .bind(&flow.profile) .bind(DataFlowType::from(flow.kind.clone())) .bind(flow.data_address.clone().map(Json)) - .bind(&flow.callback_address) + .bind(&flow.control_plane_id) .bind(Json(flow.labels.clone())) .bind(Json(flow.metadata.clone())) .bind(flow.created_at) diff --git a/crates/sdk-postgres/src/lib.rs b/crates/sdk-postgres/src/lib.rs index 4d09e8b..30deac9 100644 --- a/crates/sdk-postgres/src/lib.rs +++ b/crates/sdk-postgres/src/lib.rs @@ -10,8 +10,10 @@ // Metaform Systems, Inc. - initial API and implementation // +mod control_plane; mod data_flow; mod tx; +pub use control_plane::PgControlPlaneRepo; pub use data_flow::PgDataFlowRepo; pub use tx::{PgContext, PgTransaction}; diff --git a/crates/sdk-tck-tests/tests/dps.tck.properties b/crates/sdk-tck-tests/tests/dps.tck.properties index 491954e..646d941 100644 --- a/crates/sdk-tck-tests/tests/dps.tck.properties +++ b/crates/sdk-tck-tests/tests/dps.tck.properties @@ -4,41 +4,42 @@ dataspacetck.test.package=org.eclipse.dataspacetck.dps.verification.dataplane +dataspacetck.filters.tags.exclude=http-profile dataspacetck.dps.dataplane.url=http://host.docker.internal:8282/api/v1 # pull provider side -DP_P_PULL_01_01_TRANSFERTYPE=http_pull_sync -DP_P_PULL_01_02_TRANSFERTYPE=http_pull_sync -DP_P_PULL_02_01_TRANSFERTYPE=http_pull_sync -DP_P_PULL_02_02_TRANSFERTYPE=http_pull_sync -DP_P_PULL_04_01_TRANSFERTYPE=http_pull_async +DP_P_PULL_01_01_PROFILE=http_pull_sync +DP_P_PULL_01_02_PROFILE=http_pull_sync +DP_P_PULL_02_01_PROFILE=http_pull_sync +DP_P_PULL_02_02_PROFILE=http_pull_sync +DP_P_PULL_04_01_PROFILE=http_pull_async DP_P_PULL_04_01_AGREEMENTID=initiating-started # pull consumer side -DP_C_PULL_01_01_TRANSFERTYPE=http_pull_sync -DP_C_PULL_01_02_TRANSFERTYPE=http_pull_sync -DP_C_PULL_02_01_TRANSFERTYPE=http_pull_sync -DP_C_PULL_02_02_TRANSFERTYPE=http_pull_sync -DP_C_PULL_03_01_TRANSFERTYPE=http_pull_sync +DP_C_PULL_01_01_PROFILE=http_pull_sync +DP_C_PULL_01_02_PROFILE=http_pull_sync +DP_C_PULL_02_01_PROFILE=http_pull_sync +DP_C_PULL_02_02_PROFILE=http_pull_sync +DP_C_PULL_03_01_PROFILE=http_pull_sync DP_C_PULL_03_01_AGREEMENTID=prepared-completed -DP_C_PULL_04_01_TRANSFERTYPE=http_pull_async +DP_C_PULL_04_01_PROFILE=http_pull_async DP_C_PULL_04_01_AGREEMENTID=initiating-prepared # push consumer side -DP_C_PUSH_01_01_TRANSFERTYPE=http_push_sync -DP_C_PUSH_01_02_TRANSFERTYPE=http_push_sync -DP_C_PUSH_02_01_TRANSFERTYPE=http_push_sync -DP_C_PUSH_02_02_TRANSFERTYPE=http_push_sync -DP_C_PUSH_03_01_TRANSFERTYPE=http_push_sync -DP_C_PUSH_04_01_TRANSFERTYPE=http_push_async +DP_C_PUSH_01_01_PROFILE=http_push_sync +DP_C_PUSH_01_02_PROFILE=http_push_sync +DP_C_PUSH_02_01_PROFILE=http_push_sync +DP_C_PUSH_02_02_PROFILE=http_push_sync +DP_C_PUSH_03_01_PROFILE=http_push_sync +DP_C_PUSH_04_01_PROFILE=http_push_async DP_C_PUSH_04_01_AGREEMENTID=initiating-prepared # push provider side -DP_P_PUSH_01_01_TRANSFERTYPE=http_push_sync -DP_P_PUSH_01_02_TRANSFERTYPE=http_push_sync -DP_P_PUSH_02_01_TRANSFERTYPE=http_push_sync -DP_P_PUSH_02_02_TRANSFERTYPE=http_push_sync -DP_P_PUSH_03_01_TRANSFERTYPE=http_push_sync +DP_P_PUSH_01_01_PROFILE=http_push_sync +DP_P_PUSH_01_02_PROFILE=http_push_sync +DP_P_PUSH_02_01_PROFILE=http_push_sync +DP_P_PUSH_02_02_PROFILE=http_push_sync +DP_P_PUSH_03_01_PROFILE=http_push_sync DP_P_PUSH_03_01_AGREEMENTID=initiating-completed -DP_P_PUSH_04_01_TRANSFERTYPE=http_push_async +DP_P_PUSH_04_01_PROFILE=http_push_async DP_P_PUSH_04_01_AGREEMENTID=initiating-started diff --git a/crates/sdk-tck-tests/tests/tck_tests.rs b/crates/sdk-tck-tests/tests/tck_tests.rs index 9b1a780..5a8bdd1 100644 --- a/crates/sdk-tck-tests/tests/tck_tests.rs +++ b/crates/sdk-tck-tests/tests/tck_tests.rs @@ -86,11 +86,11 @@ mod tck_tests { .with(env_filter()) .with(tracing_subscriber::fmt::layer()) .init(); - let (ctx, repo, _container) = util::setup_postgres_container().await; + let (ctx, repo, control_plane_repo, _container) = util::setup_postgres_container().await; let (tx, rx) = tokio::sync::mpsc::channel(100); - let sdk = util::sdk(ctx, repo, TckTestHandler::new(tx)).await; + let sdk = util::sdk(ctx, repo, control_plane_repo, TckTestHandler::new(tx)).await; handle_notifications(sdk.clone(), rx); @@ -177,12 +177,12 @@ impl DataFlowHandler for TckTestHandler { ) -> HandlerResult { self.fire_notification(flow).await; self.handlers - .get(&flow.transfer_type) + .get(&flow.profile) .map(|action| action(flow)) .ok_or_else(|| { HandlerError::NotSupported(format!( - "No action defined for transfer type: {}", - flow.transfer_type + "No action defined for profile: {}", + flow.profile )) })? } @@ -194,12 +194,12 @@ impl DataFlowHandler for TckTestHandler { ) -> HandlerResult { self.fire_notification(flow).await; self.handlers - .get(&flow.transfer_type) + .get(&flow.profile) .map(|action| action(flow)) .ok_or_else(|| { HandlerError::NotSupported(format!( - "No action defined for transfer type: {}", - flow.transfer_type + "No action defined for profile: {}", + flow.profile )) })? } diff --git a/crates/sdk-tck-tests/tests/util.rs b/crates/sdk-tck-tests/tests/util.rs index e62a517..d24c485 100644 --- a/crates/sdk-tck-tests/tests/util.rs +++ b/crates/sdk-tck-tests/tests/util.rs @@ -21,16 +21,26 @@ use axum::{Extension, Router}; use dataplane_sdk::{ core::{ db::{ + control_plane::ControlPlaneRepo, data_flow::DataFlowRepo, tx::{Transaction, TransactionalContext}, }, handler::DataFlowHandler, - model::participant::ParticipantContext, + model::{control_plane::ControlPlane, participant::ParticipantContext}, }, sdk::DataPlaneSdk, }; use dataplane_sdk_axum::router::router; -use dataplane_sdk_postgres::{PgContext, PgDataFlowRepo}; +use dataplane_sdk_postgres::{PgContext, PgControlPlaneRepo, PgDataFlowRepo}; + +/// Well-known id of the control plane the TCK references in its start/prepare +/// messages via `controlPlaneId`. The data plane seeds a matching `ControlPlane` +/// record whose URL is where control-plane notification callbacks are sent. +/// +/// NOTE: the id and URL below must match what the `dps-tck` runtime sends and +/// expects; align these with the TCK image before relying on the callback flow. +pub const CONTROL_PLANE_ID: &str = "tck-control-plane"; +pub const CONTROL_PLANE_URL: &str = "http://localhost:8083"; use futures::FutureExt; use futures::future::BoxFuture; use regex::Regex; @@ -46,6 +56,7 @@ use tracing::{Level, info}; pub async fn setup_postgres_container() -> ( PgContext, PgDataFlowRepo, + PgControlPlaneRepo, testcontainers::ContainerAsync, ) { let container = Postgres::default().start().await.unwrap(); @@ -67,27 +78,29 @@ pub async fn setup_postgres_container() -> ( .await .unwrap_or_else(|_| panic!("PostgreSQL launch failed")); - (ctx.0, ctx.1, container) + (ctx.0, ctx.1, ctx.2, container) } -async fn setup_pg(url: &str) -> anyhow::Result<(PgContext, PgDataFlowRepo)> { +async fn setup_pg(url: &str) -> anyhow::Result<(PgContext, PgDataFlowRepo, PgControlPlaneRepo)> { let ctx = PgContext::connect(url).await?; let mut tx = ctx.begin().await?; let repo = PgDataFlowRepo; + let control_plane_repo = PgControlPlaneRepo; repo.migrate(&mut tx).await?; + control_plane_repo.migrate(&mut tx).await?; tx.commit().await?; - Ok((ctx, repo)) + Ok((ctx, repo, control_plane_repo)) } pub async fn setup_tck_container( reporter: TckTestReporter, ) -> testcontainers::ContainerAsync { let path = Path::new("tests/dps.tck.properties"); - GenericImage::new("eclipsedataspacetck/dps-tck-runtime", "1.1.2") + GenericImage::new("eclipsedataspacetck/dps-tck-runtime", "1.2.0") .with_exposed_port(8083.tcp()) .with_wait_for(WaitFor::message_on_stdout("Test run complete")) .with_mapped_port(8083, ContainerPort::Tcp(8083)) @@ -110,7 +123,13 @@ where let addr = SocketAddr::from_str(&format!("0.0.0.0:{port}")).expect("Invalid socket address"); let p_context = ParticipantContext::builder().id("tck-participant").build(); - let router = router().layer(Extension(p_context)); + let control_plane = ControlPlane::builder() + .id(CONTROL_PLANE_ID) + .url(CONTROL_PLANE_URL) + .build(); + let router = router() + .layer(Extension(p_context)) + .layer(Extension(control_plane)); launch_server("Signaling API", router, sdk.clone(), addr).await; } @@ -182,15 +201,33 @@ pub async fn wait_for_server(socket: std::net::SocketAddr) { } } -pub async fn sdk(ctx: C, repo: R, handler: H) -> DataPlaneSdk +pub async fn sdk( + ctx: C, + repo: R, + control_plane_repo: CP, + handler: H, +) -> DataPlaneSdk where C: TransactionalContext + 'static, C::Transaction: Send, R: DataFlowRepo + 'static, + CP: ControlPlaneRepo + 'static, H: DataFlowHandler + 'static, { + let control_plane = ControlPlane::builder() + .id(CONTROL_PLANE_ID) + .url(CONTROL_PLANE_URL) + .build(); + let mut tx = ctx.begin().await.expect("Failed to begin transaction"); + control_plane_repo + .create(&mut tx, &control_plane) + .await + .expect("Failed to seed control plane"); + tx.commit().await.expect("Failed to commit control plane"); + DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(control_plane_repo) .with_handler(handler) .build() .unwrap() diff --git a/crates/sdk/src/core/db.rs b/crates/sdk/src/core/db.rs index 0118981..ca8eca4 100644 --- a/crates/sdk/src/core/db.rs +++ b/crates/sdk/src/core/db.rs @@ -10,6 +10,7 @@ // Metaform Systems, Inc. - initial API and implementation // +pub mod control_plane; pub mod data_flow; pub mod memory; pub mod tx; diff --git a/crates/sdk/src/core/db/control_plane.rs b/crates/sdk/src/core/db/control_plane.rs new file mode 100644 index 0000000..92131e1 --- /dev/null +++ b/crates/sdk/src/core/db/control_plane.rs @@ -0,0 +1,35 @@ +// Copyright (c) 2026 Metaform Systems, Inc +// +// This program and the accompanying materials are made available under the +// terms of the Apache License, Version 2.0 which is available at +// https://www.apache.org/licenses/LICENSE-2.0 +// +// SPDX-License-Identifier: Apache-2.0 +// +// Contributors: +// Metaform Systems, Inc. - initial API and implementation +// + +use crate::core::{error::DbResult, model::control_plane::ControlPlane}; +pub mod memory; + +#[cfg(test)] +use crate::core::db::tx::MockTransaction; + +#[async_trait::async_trait] +#[cfg_attr(test, mockall::automock(type Transaction = MockTransaction;))] +pub trait ControlPlaneRepo: Send + Sync { + type Transaction; + + async fn create( + &self, + tx: &mut Self::Transaction, + control_plane: &ControlPlane, + ) -> DbResult<()>; + + async fn fetch_by_id( + &self, + tx: &mut Self::Transaction, + control_plane_id: &str, + ) -> DbResult>; +} diff --git a/crates/sdk/src/core/db/control_plane/memory.rs b/crates/sdk/src/core/db/control_plane/memory.rs new file mode 100644 index 0000000..8955c2f --- /dev/null +++ b/crates/sdk/src/core/db/control_plane/memory.rs @@ -0,0 +1,31 @@ +use crate::core::{ + db::memory::{MemoryRepo, MemoryTransaction}, + error::DbResult, + model::control_plane::ControlPlane, +}; + +use super::ControlPlaneRepo; + +#[derive(Default, Clone)] +pub struct MemoryControlPlaneRepo(MemoryRepo); + +#[async_trait::async_trait] +impl ControlPlaneRepo for MemoryControlPlaneRepo { + type Transaction = MemoryTransaction; + + async fn create( + &self, + _tx: &mut Self::Transaction, + control_plane: &ControlPlane, + ) -> DbResult<()> { + self.0.create(&control_plane.id, control_plane).await + } + + async fn fetch_by_id( + &self, + _tx: &mut Self::Transaction, + control_plane_id: &str, + ) -> DbResult> { + self.0.fetch_by_id(control_plane_id).await + } +} diff --git a/crates/sdk/src/core/db/test_suite.rs b/crates/sdk/src/core/db/test_suite.rs index 3422c2b..66cd2df 100644 --- a/crates/sdk/src/core/db/test_suite.rs +++ b/crates/sdk/src/core/db/test_suite.rs @@ -53,8 +53,8 @@ pub fn create_data_flow(id: &str) -> DataFlow { .dataset_id("dataset_id") .dataspace_context("dataspace_context") .participant_id("participant_id") - .callback_address("callback_address") - .transfer_type("transfer_type") + .control_plane_id("control_plane_id") + .profile("profile") .created_at(chrono::Utc::now().trunc_subsecs(6)) .updated_at(chrono::Utc::now().trunc_subsecs(6)) .kind(DataFlowType::Provider) diff --git a/crates/sdk/src/core/model.rs b/crates/sdk/src/core/model.rs index 171fc3b..38f6279 100644 --- a/crates/sdk/src/core/model.rs +++ b/crates/sdk/src/core/model.rs @@ -10,6 +10,7 @@ // Metaform Systems, Inc. - initial API and implementation // +pub mod control_plane; pub mod data_address; pub mod data_flow; pub mod messages; diff --git a/crates/sdk/src/core/model/control_plane.rs b/crates/sdk/src/core/model/control_plane.rs new file mode 100644 index 0000000..e8c159a --- /dev/null +++ b/crates/sdk/src/core/model/control_plane.rs @@ -0,0 +1,23 @@ +// Copyright (c) 2026 Metaform Systems, Inc +// +// This program and the accompanying materials are made available under the +// terms of the Apache License, Version 2.0 which is available at +// https://www.apache.org/licenses/LICENSE-2.0 +// +// SPDX-License-Identifier: Apache-2.0 +// +// Contributors: +// Metaform Systems, Inc. - initial API and implementation +// + +use bon::Builder; + +/// A control plane that a data flow reports back to. The `url` is the base +/// address used to build control-plane notification callbacks. Data flows +/// reference a control plane by its `id` (see `DataFlow::control_plane_id`). +#[derive(Debug, Clone, PartialEq, Builder)] +#[builder(on(String, into))] +pub struct ControlPlane { + pub id: String, + pub url: String, +} diff --git a/crates/sdk/src/core/model/data_flow.rs b/crates/sdk/src/core/model/data_flow.rs index eed006c..b21a83b 100644 --- a/crates/sdk/src/core/model/data_flow.rs +++ b/crates/sdk/src/core/model/data_flow.rs @@ -26,14 +26,14 @@ pub struct DataFlow { pub id: String, #[builder(default = DataFlowState::Initiating)] pub state: DataFlowState, - pub transfer_type: String, + pub profile: String, pub kind: DataFlowType, pub agreement_id: String, pub dataset_id: String, pub dataspace_context: String, pub participant_id: String, pub counter_party_id: String, - pub callback_address: String, + pub control_plane_id: String, pub participant_context_id: String, pub suspension_reason: Option, pub termination_reason: Option, diff --git a/crates/sdk/src/core/model/messages.rs b/crates/sdk/src/core/model/messages.rs index d1390bc..67e6579 100644 --- a/crates/sdk/src/core/model/messages.rs +++ b/crates/sdk/src/core/model/messages.rs @@ -29,8 +29,7 @@ pub struct DataFlowStartMessage { pub process_id: String, pub agreement_id: String, pub dataset_id: String, - pub callback_address: String, - pub transfer_type: String, + pub profile: String, pub data_address: Option, #[builder(default)] #[serde(default)] @@ -58,8 +57,7 @@ pub struct DataFlowPrepareMessage { pub process_id: String, pub agreement_id: String, pub dataset_id: String, - pub callback_address: String, - pub transfer_type: String, + pub profile: String, #[builder(default)] #[serde(default)] pub labels: Vec, diff --git a/crates/sdk/src/sdk.rs b/crates/sdk/src/sdk.rs index 416847c..57ac72d 100644 --- a/crates/sdk/src/sdk.rs +++ b/crates/sdk/src/sdk.rs @@ -12,7 +12,7 @@ use std::{ops::Deref, sync::Arc}; use crate::core::{ - db::{data_flow::DataFlowRepo, tx::TransactionalContext}, + db::{control_plane::ControlPlaneRepo, data_flow::DataFlowRepo, tx::TransactionalContext}, handler::DataFlowHandler, }; @@ -40,12 +40,14 @@ where pub(crate) fn new( ctx: C, repo: Box>, + control_plane_repo: Box>, handler: Box>, client: reqwest::Client, ) -> Self { Self(Arc::new(internal::DataPlaneSdkInternal { ctx, repo, + control_plane_repo, handler, client, })) @@ -69,6 +71,7 @@ where { ctx: C, repo: Option>>, + control_plane_repo: Option>>, handler: Option>>, client: Option, } @@ -81,6 +84,7 @@ where Self { ctx, repo: None, + control_plane_repo: None, handler: None, client: None, } @@ -94,6 +98,14 @@ where self } + pub fn with_control_plane_repo( + mut self, + control_plane_repo: impl ControlPlaneRepo + 'static, + ) -> Self { + self.control_plane_repo = Some(Box::new(control_plane_repo)); + self + } + pub fn with_handler( mut self, handler: impl DataFlowHandler + 'static, @@ -112,10 +124,20 @@ where pub fn build(self) -> Result, String> { let repo = self.repo.ok_or("DataFlowRepo is not set")?; + let control_plane_repo = self + .control_plane_repo + .ok_or("ControlPlaneRepo is not set")?; + let handler = self.handler.ok_or("DataFlowHandler is not set")?; let client = self.client.unwrap_or_default(); - Ok(DataPlaneSdk::new(self.ctx, repo, handler, client)) + Ok(DataPlaneSdk::new( + self.ctx, + repo, + control_plane_repo, + handler, + client, + )) } } diff --git a/crates/sdk/src/sdk/internal.rs b/crates/sdk/src/sdk/internal.rs index 051e9a2..f69b0cf 100644 --- a/crates/sdk/src/sdk/internal.rs +++ b/crates/sdk/src/sdk/internal.rs @@ -13,6 +13,7 @@ use crate::{ core::{ db::{ + control_plane::ControlPlaneRepo, data_flow::DataFlowRepo, tx::{Transaction, TransactionalContext}, }, @@ -37,6 +38,7 @@ where { pub(crate) ctx: C, pub(crate) repo: Box>, + pub(crate) control_plane_repo: Box>, pub(crate) handler: Box>, pub(crate) client: reqwest::Client, } @@ -48,6 +50,7 @@ where pub async fn start( &self, participant_context_id: &str, + control_plane_id: &str, req: DataFlowStartMessage, ) -> SdkResult { let mut flow = DataFlow::builder() @@ -61,9 +64,9 @@ where .dataspace_context(req.dataspace_context) .dataset_id(req.dataset_id) .agreement_id(req.agreement_id) - .callback_address(req.callback_address) + .control_plane_id(control_plane_id) .labels(req.labels) - .transfer_type(req.transfer_type) + .profile(req.profile) .kind(DataFlowType::Provider) .build(); @@ -95,6 +98,7 @@ where pub async fn prepare( &self, participant_context_id: &str, + control_plane_id: &str, req: DataFlowPrepareMessage, ) -> SdkResult { let mut flow = DataFlow::builder() @@ -107,9 +111,9 @@ where .dataspace_context(req.dataspace_context) .dataset_id(req.dataset_id) .agreement_id(req.agreement_id) - .callback_address(req.callback_address) + .control_plane_id(control_plane_id) .labels(req.labels) - .transfer_type(req.transfer_type) + .profile(req.profile) .kind(DataFlowType::Consumer) .build(); @@ -282,8 +286,9 @@ where } /// Notifies the control plane that the data flow has been prepared, by - /// POSTing to the flow's `callback_address`. See the Data Plane Signalling - /// "Control Plane Endpoint" section. + /// POSTing to the URL of the flow's referenced control plane (resolved from + /// `control_plane_id`). See the Data Plane Signalling "Control Plane + /// Endpoint" section. pub async fn notify_prepared( &self, ctx: &str, @@ -366,6 +371,12 @@ where .await? .ok_or_else(|| DbError::NotFound(flow_id.to_string()))?; + let control_plane = self + .control_plane_repo + .fetch_by_id(&mut tx, &flow.control_plane_id) + .await? + .ok_or_else(|| DbError::NotFound(flow.control_plane_id.clone()))?; + op(&mut flow)?; let msg = DataFlowStatusMessage::builder() @@ -377,7 +388,7 @@ where let url = format!( "{}/transfers/{}/dataflow/{}", - flow.callback_address.trim_end_matches('/'), + control_plane.url.trim_end_matches('/'), flow.id, operation ); diff --git a/crates/sdk/src/sdk_test.rs b/crates/sdk/src/sdk_test.rs index 38c51eb..65701da 100644 --- a/crates/sdk/src/sdk_test.rs +++ b/crates/sdk/src/sdk_test.rs @@ -62,11 +62,15 @@ mod prepare { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.prepare("participant", prepare_message()).await.unwrap(); + let response = sdk + .prepare("participant", "control-plane", prepare_message()) + .await + .unwrap(); assert!(response.data_address.is_none()); } @@ -87,11 +91,14 @@ mod prepare { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.prepare("participant", prepare_message()).await; + let response = sdk + .prepare("participant", "control-plane", prepare_message()) + .await; assert!(matches!(response, Err(SdkError::Repo(DbError::Generic(_))))); } @@ -122,11 +129,14 @@ mod prepare { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.prepare("participant", prepare_message()).await; + let response = sdk + .prepare("participant", "control-plane", prepare_message()) + .await; assert!(matches!( response, @@ -160,11 +170,14 @@ mod prepare { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.prepare("participant", prepare_message()).await; + let response = sdk + .prepare("participant", "control-plane", prepare_message()) + .await; assert!(matches!( response, @@ -182,11 +195,14 @@ mod prepare { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.prepare("participant", prepare_message()).await; + let response = sdk + .prepare("participant", "control-plane", prepare_message()) + .await; assert!(matches!( response, @@ -237,11 +253,15 @@ mod start { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.start("participant", start_message()).await.unwrap(); + let response = sdk + .start("participant", "control-plane", start_message()) + .await + .unwrap(); assert!(response.data_address.is_none()); } @@ -262,11 +282,14 @@ mod start { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.start("participant", start_message()).await; + let response = sdk + .start("participant", "control-plane", start_message()) + .await; assert!(matches!(response, Err(SdkError::Repo(DbError::Generic(_))))); } @@ -297,11 +320,14 @@ mod start { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.start("participant", start_message()).await; + let response = sdk + .start("participant", "control-plane", start_message()) + .await; assert!(matches!( response, @@ -335,11 +361,14 @@ mod start { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.start("participant", start_message()).await; + let response = sdk + .start("participant", "control-plane", start_message()) + .await; assert!(matches!( response, @@ -357,11 +386,14 @@ mod start { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); - let response = sdk.start("participant", start_message()).await; + let response = sdk + .start("participant", "control-plane", start_message()) + .await; assert!(matches!( response, @@ -406,6 +438,7 @@ mod terminate { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -437,6 +470,7 @@ mod terminate { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -486,6 +520,7 @@ mod suspend { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -517,6 +552,7 @@ mod suspend { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -569,6 +605,7 @@ mod started { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -598,6 +635,7 @@ mod started { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -645,6 +683,7 @@ mod completed { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -672,6 +711,7 @@ mod completed { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -734,6 +774,7 @@ mod resume { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -764,6 +805,7 @@ mod resume { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -802,6 +844,7 @@ mod status { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -828,6 +871,7 @@ mod status { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -851,26 +895,33 @@ mod notify { use crate::{ core::{ - db::tx::{MockTransaction, MockTransactionalContext}, + db::{ + control_plane::MockControlPlaneRepo, + tx::{MockTransaction, MockTransactionalContext}, + }, error::DbError, - model::data_flow::{DataFlow, DataFlowState}, + model::{ + control_plane::ControlPlane, + data_flow::{DataFlow, DataFlowState}, + }, }, error::SdkError, sdk::DataPlaneSdk, sdk_test::{context, flow}, }; - fn flow_with(callback: &str, state: DataFlowState) -> DataFlow { + fn flow_with(state: DataFlowState) -> DataFlow { let mut f = flow(); - f.callback_address = callback.to_string(); f.state = state; f } - /// Configures the mock context so that `begin`/`commit` succeed and - /// `fetch_by_id` returns the supplied flow. - fn with_flow(f: DataFlow) -> DataPlaneSdk { + /// Configures the mock context so that `begin`/`commit` succeed, + /// `fetch_by_id` returns the supplied flow, and the referenced control + /// plane resolves to `url` (the mock notification server). + fn with_flow(f: DataFlow, url: &str) -> DataPlaneSdk { let (mut ctx, mut repo, handler) = context(); + let mut cp_repo = MockControlPlaneRepo::new(); ctx.expect_begin().returning(|| { let mut tx = MockTransaction::new(); @@ -879,14 +930,25 @@ mod notify { Box::pin(future::ready(Ok(tx))) }); + let control_plane_id = f.control_plane_id.clone(); repo.expect_fetch_by_id() .returning(move |_, _| Box::pin(future::ready(Ok(Some(f.clone()))))); repo.expect_update() .returning(|_, _| Box::pin(future::ready(Ok(())))); + let url = url.to_string(); + cp_repo.expect_fetch_by_id().returning(move |_, _| { + let cp = ControlPlane::builder() + .id(control_plane_id.clone()) + .url(url.clone()) + .build(); + Box::pin(future::ready(Ok(Some(cp)))) + }); + DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(cp_repo) .with_handler(handler) .build() .unwrap() @@ -904,7 +966,7 @@ mod notify { .mount(&server) .await; - let sdk = with_flow(flow_with(&server.uri(), DataFlowState::Started)); + let sdk = with_flow(flow_with(DataFlowState::Started), &server.uri()); let response = sdk.notify_started("participant", "flow-id", None).await; @@ -921,7 +983,7 @@ mod notify { .mount(&server) .await; - let sdk = with_flow(flow_with(&server.uri(), DataFlowState::Prepared)); + let sdk = with_flow(flow_with(DataFlowState::Prepared), &server.uri()); assert!( sdk.notify_prepared("participant", "flow-id", None) @@ -940,7 +1002,7 @@ mod notify { .mount(&server) .await; - let sdk = with_flow(flow_with(&server.uri(), DataFlowState::Completed)); + let sdk = with_flow(flow_with(DataFlowState::Completed), &server.uri()); assert!(sdk.notify_completed("participant", "flow-id").await.is_ok()); } @@ -956,7 +1018,7 @@ mod notify { .mount(&server) .await; - let sdk = with_flow(flow_with(&server.uri(), DataFlowState::Started)); + let sdk = with_flow(flow_with(DataFlowState::Started), &server.uri()); let response = sdk .notify_errored( @@ -977,7 +1039,7 @@ mod notify { .mount(&server) .await; - let sdk = with_flow(flow_with(&server.uri(), DataFlowState::Started)); + let sdk = with_flow(flow_with(DataFlowState::Started), &server.uri()); let response = sdk.notify_started("participant", "flow-id", None).await; @@ -1003,6 +1065,7 @@ mod notify { let sdk = DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(crate::core::db::control_plane::MockControlPlaneRepo::new()) .with_handler(handler) .build() .unwrap(); @@ -1028,8 +1091,7 @@ fn start_message() -> DataFlowStartMessage { .build(), ) .agreement_id("agreement") - .callback_address("callback") - .transfer_type("transfer-type") + .profile("transfer-type") .counter_party_id("counter-party") .dataspace_context("dataspace-context") .message_id("message-id") @@ -1042,8 +1104,7 @@ fn prepare_message() -> DataFlowPrepareMessage { .participant_id("counter-party") .dataset_id("dataset") .agreement_id("agreement") - .callback_address("callback") - .transfer_type("transfer-type") + .profile("transfer-type") .counter_party_id("counter-party") .dataspace_context("dataspace-context") .message_id("message-id") @@ -1064,8 +1125,8 @@ fn flow() -> DataFlow { ) .dataset_id("dataset") .agreement_id("agreement") - .callback_address("callback") - .transfer_type("transfer-type") + .control_plane_id("control-plane") + .profile("transfer-type") .dataspace_context("dataspace-context") .labels(vec![]) .participant_id("participant") diff --git a/examples/example-common/src/controlplane.rs b/examples/example-common/src/controlplane.rs index 2b688f7..70cc508 100644 --- a/examples/example-common/src/controlplane.rs +++ b/examples/example-common/src/controlplane.rs @@ -17,6 +17,11 @@ use dataplane_sdk::core::model::{ }; use uuid::Uuid; +/// Well-known id of the control plane the data plane associates flows with. The +/// data plane seeds a matching `ControlPlane` record and layers it as an Axum +/// extension (see the launcher); the id is not carried on the wire messages. +pub const CONTROL_PLANE_ID: &str = "control-plane-simulator"; + #[derive(Builder)] #[builder(on(String, into))] pub struct ControlPlaneSimulator { @@ -44,11 +49,10 @@ impl ControlPlaneSimulator { .process_id(req.process_id) .agreement_id(req.agreement_id) .maybe_data_address(req.data_address) - .callback_address(format!("{}/callback", self.provider)) .participant_id("control-plane-simulator") .counter_party_id("counter-party") .dataspace_context("example-dataspace") - .transfer_type("example-transfer-type") + .profile("example-transfer-type") .message_id(Uuid::new_v4().to_string()) .build(); @@ -75,11 +79,10 @@ impl ControlPlaneSimulator { .process_id(req.process_id) .agreement_id(req.agreement_id) .maybe_data_address(req.data_address) - .callback_address(format!("{}/callback", self.consumer)) .participant_id("control-plane-simulator") .counter_party_id("counter-party") .dataspace_context("example-dataspace") - .transfer_type("example-transfer-type") + .profile("example-transfer-type") .message_id(Uuid::new_v4().to_string()) .build(); @@ -132,11 +135,10 @@ impl ControlPlaneSimulator { .process_id(req.process_id) .agreement_id(req.agreement_id) .maybe_data_address(req.data_address) - .callback_address(format!("{}/callback", base_url)) .participant_id("control-plane-simulator") .counter_party_id("counter-party") .dataspace_context("example-dataspace") - .transfer_type("example-transfer-type") + .profile("example-transfer-type") .message_id(Uuid::new_v4().to_string()) .build(); diff --git a/examples/example-common/src/signaling.rs b/examples/example-common/src/signaling.rs index 8663aed..09dc4f9 100644 --- a/examples/example-common/src/signaling.rs +++ b/examples/example-common/src/signaling.rs @@ -14,7 +14,10 @@ use std::{net::SocketAddr, str::FromStr, sync::Arc}; use axum::Extension; use dataplane_sdk::{ - core::{db::tx::TransactionalContext, model::participant::ParticipantContext}, + core::{ + db::tx::TransactionalContext, + model::{control_plane::ControlPlane, participant::ParticipantContext}, + }, sdk::DataPlaneSdk, }; use dataplane_sdk_axum::router::router; @@ -22,8 +25,12 @@ use tokio::sync::Barrier; use crate::util::launch_server; -pub async fn start_signaling(port: u16, sdk: DataPlaneSdk, barrier: Arc) -where +pub async fn start_signaling( + port: u16, + sdk: DataPlaneSdk, + control_plane: ControlPlane, + barrier: Arc, +) where C: TransactionalContext + 'static, C::Transaction: Send, { @@ -32,7 +39,9 @@ where let p_context = ParticipantContext::builder() .id("example-participant") .build(); - let router = router().layer(Extension(p_context)); + let router = router() + .layer(Extension(p_context)) + .layer(Extension(control_plane)); launch_server("Signaling API", router, sdk.clone(), addr, barrier).await; } diff --git a/examples/sync-pull-dataplane/src/launcher.rs b/examples/sync-pull-dataplane/src/launcher.rs index ea0cc24..5043b3d 100644 --- a/examples/sync-pull-dataplane/src/launcher.rs +++ b/examples/sync-pull-dataplane/src/launcher.rs @@ -19,72 +19,103 @@ use crate::tokens::manager::TokenManager; use crate::tokens::repo::TokenRepo; use crate::tokens::repo::memory::MemoryTokenRepo; use crate::tokens::repo::postgres::PgTokenRepo; +use dataplane_sdk::core::db::control_plane::ControlPlaneRepo; +use dataplane_sdk::core::db::control_plane::memory::MemoryControlPlaneRepo; use dataplane_sdk::core::db::data_flow::DataFlowRepo; use dataplane_sdk::core::db::data_flow::memory::MemoryDataFlowRepo; use dataplane_sdk::core::db::memory::MemoryContext; use dataplane_sdk::core::db::tx::{Transaction, TransactionalContext}; +use dataplane_sdk::core::model::control_plane::ControlPlane; use dataplane_sdk::sdk::DataPlaneSdk; -use dataplane_sdk_postgres::{PgContext, PgDataFlowRepo}; +use dataplane_sdk_postgres::{PgContext, PgControlPlaneRepo, PgDataFlowRepo}; +use example_common::controlplane::CONTROL_PLANE_ID; use example_common::signaling::start_signaling; use tokio::sync::Barrier; pub async fn start_dataplane(cfg: DataPlaneConfig) -> anyhow::Result<()> { match &cfg.db { crate::config::Db::Memory => { - let (ctx, repo, token_repo) = setup_memory().await?; - internal_launch(&cfg, ctx, repo, token_repo).await + let (ctx, repo, control_plane_repo, token_repo) = setup_memory().await?; + internal_launch(&cfg, ctx, repo, control_plane_repo, token_repo).await } crate::config::Db::Postgres { url } => { - let (ctx, repo, token_repo) = setup_pg(url).await?; - internal_launch(&cfg, ctx, repo, token_repo).await + let (ctx, repo, control_plane_repo, token_repo) = setup_pg(url).await?; + internal_launch(&cfg, ctx, repo, control_plane_repo, token_repo).await } } } -async fn setup_memory() -> anyhow::Result<(MemoryContext, MemoryDataFlowRepo, MemoryTokenRepo)> { +async fn setup_memory() -> anyhow::Result<( + MemoryContext, + MemoryDataFlowRepo, + MemoryControlPlaneRepo, + MemoryTokenRepo, +)> { let ctx = MemoryContext; let repo = MemoryDataFlowRepo::default(); + let control_plane_repo = MemoryControlPlaneRepo::default(); let token_repo = MemoryTokenRepo::default(); - Ok((ctx, repo, token_repo)) + Ok((ctx, repo, control_plane_repo, token_repo)) } -async fn setup_pg(url: &str) -> anyhow::Result<(PgContext, PgDataFlowRepo, PgTokenRepo)> { +async fn setup_pg( + url: &str, +) -> anyhow::Result<(PgContext, PgDataFlowRepo, PgControlPlaneRepo, PgTokenRepo)> { let ctx = PgContext::connect(url).await?; let mut tx = ctx.begin().await?; let repo = PgDataFlowRepo; + let control_plane_repo = PgControlPlaneRepo; let token_repo = PgTokenRepo; repo.migrate(&mut tx).await?; + control_plane_repo.migrate(&mut tx).await?; token_repo.migrate(&mut tx).await?; tx.commit().await?; - Ok((ctx, repo, token_repo)) + Ok((ctx, repo, control_plane_repo, token_repo)) } -async fn internal_launch( +async fn internal_launch( cfg: &DataPlaneConfig, ctx: C, flows: R, + control_planes: CP, tokens: T, ) -> anyhow::Result<()> where C: TransactionalContext + 'static, C::Transaction: Send, R: DataFlowRepo + 'static, + CP: ControlPlaneRepo + 'static, T: TokenRepo + 'static, { + let control_plane = ControlPlane::builder() + .id(CONTROL_PLANE_ID) + .url(format!( + "http://localhost:{}/api/v1/callback", + cfg.signaling.port + )) + .build(); + seed_control_plane(&ctx, &control_planes, &control_plane).await?; + let token_manager = Arc::new(create_token_manager(cfg, tokens).await?); let handler = TokenHandler::new(token_manager.clone()); - let sdk = sdk(ctx, flows, handler).await; + let sdk = sdk(ctx, flows, control_planes, handler).await; let barrier = Arc::new(Barrier::new(4)); - start_signaling(cfg.signaling.port, sdk.clone(), barrier.clone()).await; + start_signaling( + cfg.signaling.port, + sdk.clone(), + control_plane.clone(), + barrier.clone(), + ) + .await; start_public_api( cfg.public_api.port, @@ -107,19 +138,44 @@ where Ok(()) } -async fn sdk(ctx: C, repo: R, handler: TokenHandler) -> DataPlaneSdk +async fn sdk( + ctx: C, + repo: R, + control_plane_repo: CP, + handler: TokenHandler, +) -> DataPlaneSdk where C: TransactionalContext + 'static, C::Transaction: Send, R: DataFlowRepo + 'static, + CP: ControlPlaneRepo + 'static, { DataPlaneSdk::builder(ctx) .with_repo(repo) + .with_control_plane_repo(control_plane_repo) .with_handler(handler) .build() .unwrap() } +/// Registers the control plane that data flows report back to, so that +/// notification callbacks can later resolve its URL by id. +async fn seed_control_plane( + ctx: &C, + control_planes: &CP, + control_plane: &ControlPlane, +) -> anyhow::Result<()> +where + C: TransactionalContext, + CP: ControlPlaneRepo, +{ + let mut tx = ctx.begin().await?; + control_planes.create(&mut tx, control_plane).await?; + tx.commit().await?; + + Ok(()) +} + async fn create_token_manager< T: TransactionalContext, R: TokenRepo + 'static,