feat: add controlplane and RC3 compliance

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<C>(
     State(sdk): State<DataPlaneSdk<C>>,
     Extension(participant): Extension<ParticipantContext>,
+    Extension(control_plane): Extension<ControlPlane>,
     Json(msg): Json<DataFlowStartMessage>,
 ) -> SignalingResult<Json<DataFlowStatusMessage>>
 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<C>(
     State(sdk): State<DataPlaneSdk<C>>,
     Extension(participant): Extension<ParticipantContext>,
+    Extension(control_plane): Extension<ControlPlane>,
     Json(msg): Json<DataFlowPrepareMessage>,
 ) -> SignalingResult<(StatusCode, Json<DataFlowStatusMessage>)>
 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,

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<Option<ControlPlane>>;
+    }
+}
+
 mock! {
     Handler {}
 
@@ -123,22 +139,32 @@ struct TestCtx {
     base_url: String,
     app: Router<DataPlaneSdk<MockTxContext>>,
     participant_context: ParticipantContext,
+    control_plane: ControlPlane,
 }
 
 impl TestCtx {
     pub fn app(&self, sdk: DataPlaneSdk<MockTxContext>) -> 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(),
         app: router(),
         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<MockTxContext> {
     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(),
             ))

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())
             }

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 ,

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
+)

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;

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<ControlPlane> 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()
+    }
+}

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<Option<ControlPlane>> {
+        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(())
+    }
+}

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<String>,
     pub termination_reason: Option<String>,
     #[builder(default)]
@@ -81,8 +81,8 @@ impl From<DataFlow> 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)