associate a publication with a source

api/migrations/20240828090309_create_publication.sql

@@ -2,5 +2,6 @@ create table
     public.publications (
         id bigint generated always as identity primary key,
         tenant_id bigint references public.tenants(id) not null,
+        source_id bigint references public.sources(id) not null,
         config jsonb not null
     );

api/src/db/publications.rs

@@ -8,22 +8,25 @@ pub struct PublicationConfig {
 pub struct Publication {
     pub id: i64,
     pub tenant_id: i64,
+    pub source_id: i64,
     pub config: serde_json::Value,
 }
 
 pub async fn create_publication(
     pool: &PgPool,
     tenant_id: i64,
+    source_id: i64,
     config: &PublicationConfig,
 ) -> Result<i64, sqlx::Error> {
     let config = serde_json::to_value(config).expect("failed to serialize config");
     let record = sqlx::query!(
         r#"
-        insert into publications (tenant_id, config)
-        values ($1, $2)
+        insert into publications (tenant_id, source_id, config)
+        values ($1, $2, $3)
         returning id
         "#,
         tenant_id,
+        source_id,
         config
     )
     .fetch_one(pool)
@@ -39,7 +42,7 @@ pub async fn read_publication(
 ) -> Result<Option<Publication>, sqlx::Error> {
     let record = sqlx::query!(
         r#"
-        select id, tenant_id, config
+        select id, tenant_id, source_id, config
         from publications
         where tenant_id = $1 and id = $2
         "#,
@@ -52,6 +55,7 @@ pub async fn read_publication(
     Ok(record.map(|r| Publication {
         id: r.id,
         tenant_id: r.tenant_id,
+        source_id: r.source_id,
         config: r.config,
     }))
 }
@@ -60,16 +64,18 @@ pub async fn update_publication(
     pool: &PgPool,
     tenant_id: i64,
     publication_id: i64,
+    source_id: i64,
     config: &PublicationConfig,
 ) -> Result<Option<i64>, sqlx::Error> {
     let config = serde_json::to_value(config).expect("failed to serialize config");
     let record = sqlx::query!(
         r#"
         update publications
-        set config = $1
-        where tenant_id = $2 and id = $3
+        set source_id = $1, config = $2
+        where tenant_id = $3 and id = $4
         returning id
         "#,
+        source_id,
         config,
         tenant_id,
         publication_id
@@ -106,7 +112,7 @@ pub async fn read_all_publications(
 ) -> Result<Vec<Publication>, sqlx::Error> {
     let mut record = sqlx::query!(
         r#"
-        select id, tenant_id, config
+        select id, tenant_id, source_id, config
         from publications
         where tenant_id = $1
         "#,
@@ -120,6 +126,7 @@ pub async fn read_all_publications(
         .map(|r| Publication {
             id: r.id,
             tenant_id: r.tenant_id,
+            source_id: r.source_id,
             config: r.config,
         })
         .collect())

api/src/routes/publications.rs

@@ -45,6 +45,7 @@ impl ResponseError for PublicationError {
 
 #[derive(Deserialize)]
 struct PostPublicationRequest {
+    pub source_id: i64,
     pub config: PublicationConfig,
 }
 
@@ -57,6 +58,7 @@ struct PostPublicationResponse {
 struct GetPublicationResponse {
     id: i64,
     tenant_id: i64,
+    source_id: i64,
     config: PublicationConfig,
 }
 
@@ -83,8 +85,9 @@ pub async fn create_publication(
 ) -> Result<impl Responder, PublicationError> {
     let publication = publication.0;
     let tenant_id = extract_tenant_id(&req)?;
+    let source_id = publication.source_id;
     let config = publication.config;
-    let id = db::publications::create_publication(&pool, tenant_id, &config).await?;
+    let id = db::publications::create_publication(&pool, tenant_id, source_id, &config).await?;
     let response = PostPublicationResponse { id };
     Ok(Json(response))
 }
@@ -104,6 +107,7 @@ pub async fn read_publication(
             Ok::<GetPublicationResponse, serde_json::Error>(GetPublicationResponse {
                 id: s.id,
                 tenant_id: s.tenant_id,
+                source_id: s.source_id,
                 config,
             })
         })
@@ -121,8 +125,9 @@ pub async fn update_publication(
 ) -> Result<impl Responder, PublicationError> {
     let tenant_id = extract_tenant_id(&req)?;
     let publication_id = publication_id.into_inner();
+    let source_id = publication.source_id;
     let config = &publication.config;
-    db::publications::update_publication(&pool, tenant_id, publication_id, config)
+    db::publications::update_publication(&pool, tenant_id, publication_id, source_id, config)
         .await?
         .ok_or(PublicationError::NotFound(publication_id))?;
     Ok(HttpResponse::Ok().finish())
@@ -154,6 +159,7 @@ pub async fn read_all_publications(
         let sink = GetPublicationResponse {
             id: publication.id,
             tenant_id: publication.tenant_id,
+            source_id: publication.source_id,
             config,
         };
         publications.push(sink);

api/tests/api/publications.rs

@@ -3,6 +3,7 @@ use reqwest::StatusCode;
 
 use crate::{
     sinks::create_sink,
+    sources::create_source,
     tenants::create_tenant,
     test_app::{
         spawn_app, CreatePublicationRequest, CreatePublicationResponse, PublicationResponse,
@@ -25,9 +26,10 @@ fn updated_publication_config() -> PublicationConfig {
 pub async fn create_publication_with_config(
     app: &TestApp,
     tenant_id: i64,
+    source_id: i64,
     config: PublicationConfig,
 ) -> i64 {
-    let publication = CreatePublicationRequest { config };
+    let publication = CreatePublicationRequest { source_id, config };
     let response = app.create_publication(tenant_id, &publication).await;
     let response: CreatePublicationResponse = response
         .json()
@@ -41,9 +43,11 @@ async fn publication_can_be_created() {
     // Arrange
     let app = spawn_app().await;
     let tenant_id = create_tenant(&app).await;
+    let source_id = create_source(&app, tenant_id).await;
 
     // Act
     let publication = CreatePublicationRequest {
+        source_id,
         config: new_publication_config(),
     };
     let response = app.create_publication(tenant_id, &publication).await;
@@ -62,8 +66,10 @@ async fn an_existing_publication_can_be_read() {
     // Arrange
     let app = spawn_app().await;
     let tenant_id = create_tenant(&app).await;
+    let source_id = create_source(&app, tenant_id).await;
     let sink_id = create_sink(&app, tenant_id).await;
     let publication = CreatePublicationRequest {
+        source_id,
         config: new_publication_config(),
     };
     let response = app.create_publication(tenant_id, &publication).await;
@@ -84,6 +90,7 @@ async fn an_existing_publication_can_be_read() {
         .expect("failed to deserialize response");
     assert_eq!(response.id, sink_id);
     assert_eq!(response.tenant_id, tenant_id);
+    assert_eq!(response.source_id, source_id);
     assert_eq!(response.config, publication.config);
 }
 
@@ -105,8 +112,10 @@ async fn an_existing_publication_can_be_updated() {
     // Arrange
     let app = spawn_app().await;
     let tenant_id = create_tenant(&app).await;
+    let source_id = create_source(&app, tenant_id).await;
 
     let publication = CreatePublicationRequest {
+        source_id,
         config: new_publication_config(),
     };
     let response = app.create_publication(tenant_id, &publication).await;
@@ -117,7 +126,9 @@ async fn an_existing_publication_can_be_updated() {
     let publication_id = response.id;
 
     // Act
+    let updated_source_id = create_source(&app, tenant_id).await;
     let updated_config = UpdatePublicationRequest {
+        source_id: updated_source_id,
         config: updated_publication_config(),
     };
     let response = app
@@ -133,6 +144,7 @@ async fn an_existing_publication_can_be_updated() {
         .expect("failed to deserialize response");
     assert_eq!(response.id, publication_id);
     assert_eq!(response.tenant_id, tenant_id);
+    assert_eq!(response.source_id, updated_source_id);
     assert_eq!(response.config, updated_config.config);
 }
 
@@ -141,9 +153,11 @@ async fn an_non_existing_publication_cant_be_updated() {
     // Arrange
     let app = spawn_app().await;
     let tenant_id = create_tenant(&app).await;
+    let source_id = create_source(&app, tenant_id).await;
 
     // Act
     let updated_config = UpdatePublicationRequest {
+        source_id,
         config: updated_publication_config(),
     };
     let response = app.update_publication(tenant_id, 42, &updated_config).await;
@@ -157,8 +171,10 @@ async fn an_existing_publication_can_be_deleted() {
     // Arrange
     let app = spawn_app().await;
     let tenant_id = create_tenant(&app).await;
+    let source_id = create_source(&app, tenant_id).await;
 
     let publication = CreatePublicationRequest {
+        source_id,
         config: new_publication_config(),
     };
     let response = app.create_publication(tenant_id, &publication).await;
@@ -195,10 +211,12 @@ async fn all_publications_can_be_read() {
     // Arrange
     let app = spawn_app().await;
     let tenant_id = create_tenant(&app).await;
+    let source_id = create_source(&app, tenant_id).await;
     let publication1_id =
-        create_publication_with_config(&app, tenant_id, new_publication_config()).await;
+        create_publication_with_config(&app, tenant_id, source_id, new_publication_config()).await;
     let publication2_id =
-        create_publication_with_config(&app, tenant_id, updated_publication_config()).await;
+        create_publication_with_config(&app, tenant_id, source_id, updated_publication_config())
+            .await;
 
     // Act
     let response = app.read_all_publications(tenant_id).await;