add publication id to pipeline

supabase · Aug 29, 2024 · a8126e9 · a8126e9
1 parent 63767c8
commit a8126e9
Show file tree

Hide file tree

Showing 13 changed files with 101 additions and 28 deletions.
diff --git a/...257832d249a1fbe3369f587c75f616e88c2d.json → ...f8004c5440da61595a1481eb93f3f3076a7c.json b/...257832d249a1fbe3369f587c75f616e88c2d.json → ...f8004c5440da61595a1481eb93f3f3076a7c.json
diff --git a/...db97dfc5bea7c315e68287dc478bfa5cbb98.json → ...e8de0098154c541dd59b0c3180facc5e60a6.json b/...db97dfc5bea7c315e68287dc478bfa5cbb98.json → ...e8de0098154c541dd59b0c3180facc5e60a6.json
diff --git a/...12e115e805db2cd2cf934e192824f52177a3.json → ...72e23f1315be11cd01a0a84ae327223d63b1.json b/...12e115e805db2cd2cf934e192824f52177a3.json → ...72e23f1315be11cd01a0a84ae327223d63b1.json
diff --git a/...01a7f31df72aae63dafc0d182fbdcc624de4.json → ...a321afeff6d4aaa053c5b0a281d38dbfebeb.json b/...01a7f31df72aae63dafc0d182fbdcc624de4.json → ...a321afeff6d4aaa053c5b0a281d38dbfebeb.json
diff --git a/...ons/20240828090309_create_publication.sql → ...ns/20240822082905_create_publications.sql b/...ons/20240828090309_create_publication.sql → ...ns/20240822082905_create_publications.sql
diff --git a/...tions/20240822082905_create_pipelines.sql → ...tions/20240828090309_create_pipelines.sql b/...tions/20240822082905_create_pipelines.sql → ...tions/20240828090309_create_pipelines.sql
@@ -4,5 +4,6 @@ create table
         tenant_id bigint references public.tenants(id) not null,
         source_id bigint references public.sources(id) not null,
         sink_id bigint references public.sinks(id) not null,
+        publication_id bigint references public.publications(id) not null,
         config jsonb not null
     );
diff --git a/api/src/db/pipelines.rs b/api/src/db/pipelines.rs
@@ -19,6 +19,7 @@ pub struct Pipeline {
     pub tenant_id: i64,
     pub source_id: i64,
     pub sink_id: i64,
+    pub publication_id: i64,
     pub config: serde_json::Value,
 }
 
@@ -27,18 +28,20 @@ pub async fn create_pipeline(
     tenant_id: i64,
     source_id: i64,
     sink_id: i64,
+    publication_id: i64,
     config: &PipelineConfig,
 ) -> Result<i64, sqlx::Error> {
     let config = serde_json::to_value(config).expect("failed to serialize config");
     let record = sqlx::query!(
         r#"
-        insert into pipelines (tenant_id, source_id, sink_id, config)
-        values ($1, $2, $3, $4)
+        insert into pipelines (tenant_id, source_id, sink_id, publication_id, config)
+        values ($1, $2, $3, $4, $5)
         returning id
         "#,
         tenant_id,
         source_id,
         sink_id,
+        publication_id,
         config
     )
     .fetch_one(pool)
@@ -54,7 +57,7 @@ pub async fn read_pipeline(
 ) -> Result<Option<Pipeline>, sqlx::Error> {
     let record = sqlx::query!(
         r#"
-        select id, tenant_id, source_id, sink_id, config
+        select id, tenant_id, source_id, sink_id, publication_id, config
         from pipelines
         where tenant_id = $1 and id = $2
         "#,
@@ -69,6 +72,7 @@ pub async fn read_pipeline(
         tenant_id: r.tenant_id,
         source_id: r.source_id,
         sink_id: r.sink_id,
+        publication_id: r.publication_id,
         config: r.config,
     }))
 }
@@ -79,18 +83,20 @@ pub async fn update_pipeline(
     pipeline_id: i64,
     source_id: i64,
     sink_id: i64,
+    publication_id: i64,
     config: &PipelineConfig,
 ) -> Result<Option<i64>, sqlx::Error> {
     let config = serde_json::to_value(config).expect("failed to serialize config");
     let record = sqlx::query!(
         r#"
         update pipelines
-        set source_id = $1, sink_id = $2, config = $3
-        where tenant_id = $4 and id = $5
+        set source_id = $1, sink_id = $2, publication_id = $3, config = $4
+        where tenant_id = $5 and id = $6
         returning id
         "#,
         source_id,
         sink_id,
+        publication_id,
         config,
         tenant_id,
         pipeline_id
@@ -127,7 +133,7 @@ pub async fn read_all_pipelines(
 ) -> Result<Vec<Pipeline>, sqlx::Error> {
     let mut record = sqlx::query!(
         r#"
-        select id, tenant_id, source_id, sink_id, config
+        select id, tenant_id, source_id, sink_id, publication_id, config
         from pipelines
         where tenant_id = $1
         "#,
@@ -143,6 +149,7 @@ pub async fn read_all_pipelines(
             tenant_id: r.tenant_id,
             source_id: r.source_id,
             sink_id: r.sink_id,
+            publication_id: r.publication_id,
             config: r.config,
         })
         .collect())

diff --git a/api/src/routes/pipelines.rs b/api/src/routes/pipelines.rs
@@ -47,6 +47,7 @@ impl ResponseError for PipelineError {
 struct PostPipelineRequest {
     pub source_id: i64,
     pub sink_id: i64,
+    pub publication_id: i64,
     pub config: PipelineConfig,
 }
 
@@ -61,6 +62,7 @@ struct GetPipelineResponse {
     tenant_id: i64,
     source_id: i64,
     sink_id: i64,
+    publication_id: i64,
     config: PipelineConfig,
 }
 
@@ -93,6 +95,7 @@ pub async fn create_pipeline(
         tenant_id,
         pipeline.source_id,
         pipeline.sink_id,
+        pipeline.publication_id,
         &config,
     )
     .await?;
@@ -117,6 +120,7 @@ pub async fn read_pipeline(
                 tenant_id: s.tenant_id,
                 source_id: s.source_id,
                 sink_id: s.sink_id,
+                publication_id: s.publication_id,
                 config,
             })
         })
@@ -141,6 +145,7 @@ pub async fn update_pipeline(
         pipeline_id,
         pipeline.source_id,
         pipeline.sink_id,
+        pipeline.publication_id,
         config,
     )
     .await?
@@ -176,6 +181,7 @@ pub async fn read_all_pipelines(
             tenant_id: pipeline.tenant_id,
             source_id: pipeline.source_id,
             sink_id: pipeline.sink_id,
+            publication_id: pipeline.publication_id,
             config,
         };
         pipelines.push(sink);