Adding metadata to application CRUD group

server/run-tests.sh

@@ -7,7 +7,7 @@ TEST_COMMAND="cargo test --all --all-features --all-targets"
 # Common variables:
 export DATABASE_URL="postgresql://postgres:postgres@localhost:5432/postgres"
 export SVIX_JWT_SECRET="test value"
-export SVIX_LOG_LEVEL="trace"
+export SVIX_LOG_LEVEL="info"
 
 echo "*********** RUN 1 ***********"
 SVIX_QUEUE_TYPE="redis" \

server/svix-server/migrations/20221104170041_add_applicationmetadata.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+DROP TABLE IF EXISTS applicationmetadata;

server/svix-server/migrations/20221104170041_add_applicationmetadata.up.sql

@@ -0,0 +1,10 @@
+-- Add up migration script here
+CREATE TABLE applicationmetadata (
+    created_at timestamp with time zone NOT NULL,
+    updated_at timestamp with time zone NOT NULL,
+    id character varying NOT NULL COLLATE pg_catalog."C",
+    data jsonb NOT NULL
+);
+
+ALTER TABLE ONLY applicationmetadata
+    ADD CONSTRAINT applicationmetadata_pkey PRIMARY KEY (id);

server/svix-server/src/core/permissions.rs

@@ -7,19 +7,30 @@ use sea_orm::DatabaseConnection;
 
 use crate::{
     ctx,
-    db::models::application,
+    db::models::{application, applicationmetadata},
     error::{Error, HttpError, Result},
 };
 
 use super::{
-    security::{permissions_from_bearer, AccessLevel},
-    types::{ApplicationIdOrUid, OrganizationId},
+    security::{permissions_from_bearer, AccessLevel, Permissions},
+    types::{ApplicationId, ApplicationIdOrUid, OrganizationId},
 };
 
 pub struct Organization {
     pub org_id: OrganizationId,
 }
 
+impl Permissions {
+    fn check_app_is_permitted(&self, app_id: &ApplicationId) -> Result<()> {
+        if let Some(ref permitted_app_id) = self.app_id() {
+            if permitted_app_id != app_id {
+                return Err(HttpError::not_found(None, None).into());
+            }
+        }
+        Ok(())
+    }
+}
+
 #[async_trait]
 impl<B> FromRequest<B> for Organization
 where
@@ -63,11 +74,7 @@ where
         )?
         .ok_or_else(|| HttpError::not_found(None, None))?;
 
-        if let Some(permitted_app_id) = permissions.app_id() {
-            if permitted_app_id != app.id {
-                return Err(HttpError::not_found(None, None).into());
-            }
-        }
+        permissions.check_app_is_permitted(&app.id)?;
 
         Ok(Self { app })
     }
@@ -101,6 +108,36 @@ where
     }
 }
 
+pub struct ApplicationWithMetadata {
+    pub app: application::Model,
+    pub metadata: applicationmetadata::Model,
+}
+
+#[async_trait]
+impl<B> FromRequest<B> for ApplicationWithMetadata
+where
+    B: Send,
+{
+    type Rejection = Error;
+
+    async fn from_request(req: &mut RequestParts<B>) -> Result<Self> {
+        let permissions = permissions_from_bearer(req).await?;
+
+        let Path(ApplicationPathParams { app_id }) =
+            ctx!(Path::<ApplicationPathParams>::from_request(req).await)?;
+        let Extension(ref db) = ctx!(Extension::<DatabaseConnection>::from_request(req).await)?;
+        let (app, metadata) = ctx!(
+            application::Model::fetch_with_metadata(db, permissions.org_id(), app_id.to_owned())
+                .await
+        )?
+        .ok_or_else(|| HttpError::not_found(None, None))?;
+
+        permissions.check_app_is_permitted(&app.id)?;
+
+        Ok(Self { app, metadata })
+    }
+}
+
 #[derive(serde::Deserialize)]
 struct ApplicationPathParams {
     app_id: ApplicationIdOrUid,

server/svix-server/src/core/types/metadata.rs

@@ -0,0 +1,39 @@
+use std::collections::HashMap;
+
+use crate::json_wrapper;
+use serde::{Deserialize, Serialize};
+
+pub const MAX_METADATA_SIZE: usize = 4096;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Default)]
+pub struct Metadata(HashMap<String, String>);
+
+json_wrapper!(Metadata);
+
+impl Metadata {
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+}
+
+impl<'de> Deserialize<'de> for Metadata {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let inner: Option<HashMap<String, String>> = Deserialize::deserialize(deserializer)?;
+        let metadata = inner.unwrap_or_default(); // coerce `null` to `{}`
+
+        let size = serde_json::to_string(&metadata)
+            .map(|blob| blob.len())
+            .map_err(|_| serde::de::Error::custom("metadata is not valid json"))?;
+
+        if size > MAX_METADATA_SIZE {
+            return Err(serde::de::Error::custom(format!(
+                "metadata must be less than or equal to {MAX_METADATA_SIZE} bytes"
+            )));
+        }
+
+        Ok(Self(metadata))
+    }
+}

server/svix-server/src/core/types.rs → server/svix-server/src/core/types/mod.rs

@@ -17,6 +17,8 @@ use validator::{Validate, ValidationErrors};
 
 use crate::{err_generic, v1::utils::validation_error};
 
+pub mod metadata;
+
 use super::cryptography::{AsymmetricKey, Encryption};
 
 const ALL_ERROR: &str = "__all__";

server/svix-server/src/db/mod.rs

@@ -134,3 +134,16 @@ pub async fn wipe_org(cfg: &Configuration, org_id: OrganizationId) {
             )
         });
 }
+
+#[macro_export]
+/// Runs an async closure inside of a DB Transaction. The closure should return an [`error::Result<T>`]. If the closure returns an error for any reason, the transaction is rolled back.
+macro_rules! transaction {
+    ($db:expr, $do:expr) => {
+        $crate::ctx!(
+            sea_orm::TransactionTrait::transaction::<_, _, $crate::error::Error>($db, |txn| {
+                std::boxed::Box::pin({ $do(txn) })
+            })
+            .await
+        )
+    };
+}

server/svix-server/src/db/models/application.rs

@@ -4,9 +4,13 @@
 use crate::core::types::{
     ApplicationId, ApplicationIdOrUid, ApplicationUid, BaseId, OrganizationId,
 };
+use crate::{ctx, error};
 use chrono::Utc;
 use sea_orm::ActiveValue::Set;
 use sea_orm::{entity::prelude::*, Condition};
+use sea_orm::{ConnectionTrait, QueryOrder, QuerySelect};
+
+use super::applicationmetadata;
 
 #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
 #[sea_orm(table_name = "application")]
@@ -28,6 +32,8 @@ pub enum Relation {
     Endpoint,
     #[sea_orm(has_many = "super::message::Entity")]
     Message,
+    #[sea_orm(has_one = "super::applicationmetadata::Entity")]
+    Metadata,
 }
 
 impl Related<super::endpoint::Entity> for Entity {
@@ -42,22 +48,91 @@ impl Related<super::message::Entity> for Entity {
     }
 }
 
+impl Related<super::applicationmetadata::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Metadata.def()
+    }
+}
+
+impl Model {
+    pub async fn fetch_or_create_metadata(
+        &self,
+        db: &impl ConnectionTrait,
+    ) -> error::Result<applicationmetadata::ActiveModel> {
+        let query = applicationmetadata::Entity::secure_find(self.id.clone());
+        let metadata = ctx!(query.one(db).await)?
+            .map(applicationmetadata::ActiveModel::from)
+            .unwrap_or_else(|| applicationmetadata::ActiveModel::new(self.id.clone(), None));
+
+        Ok(metadata)
+    }
+
+    pub async fn fetch_many_with_metadata(
+        db: &DatabaseConnection,
+        org_id: OrganizationId,
+        limit: u64,
+        after_id: impl Into<Option<ApplicationId>>,
+    ) -> error::Result<impl Iterator<Item = (Self, applicationmetadata::Model)>> {
+        let mut query = Entity::secure_find(org_id)
+            .order_by_asc(Column::Id)
+            .limit(limit);
+
+        if let Some(id) = after_id.into() {
+            query = query.filter(Column::Id.gt(id))
+        }
+
+        let results = ctx!(
+            query
+                .find_also_related(applicationmetadata::Entity)
+                .all(db)
+                .await
+        )?;
+
+        Ok(results.into_iter().map(|(app, metadata)| {
+            let metadata =
+                metadata.unwrap_or_else(|| applicationmetadata::Model::new(app.id.clone()));
+            (app, metadata)
+        }))
+    }
+
+    pub async fn fetch_with_metadata(
+        db: &DatabaseConnection,
+        org_id: OrganizationId,
+        id_or_uid: ApplicationIdOrUid,
+    ) -> error::Result<Option<(Self, applicationmetadata::Model)>> {
+        let result = ctx!(
+            Entity::secure_find_by_id_or_uid(org_id, id_or_uid)
+                .find_also_related(applicationmetadata::Entity)
+                .one(db)
+                .await
+        )?;
+        Ok(result.map(|(app, metadata)| {
+            let metadata =
+                metadata.unwrap_or_else(|| applicationmetadata::Model::new(app.id.clone()));
+            (app, metadata)
+        }))
+    }
+}
+
 impl ActiveModelBehavior for ActiveModel {
-    fn new() -> Self {
+    fn before_save(mut self, _insert: bool) -> Result<Self, DbErr> {
+        self.updated_at = Set(Utc::now().into());
+        Ok(self)
+    }
+}
+
+impl ActiveModel {
+    pub fn new(org_id: OrganizationId) -> Self {
         let timestamp = Utc::now();
         Self {
             id: Set(ApplicationId::new(timestamp.into(), None)),
+            org_id: Set(org_id),
             created_at: Set(timestamp.into()),
             updated_at: Set(timestamp.into()),
             deleted: Set(false),
             ..ActiveModelTrait::default()
         }
     }
-
-    fn before_save(mut self, _insert: bool) -> Result<Self, DbErr> {
-        self.updated_at = Set(Utc::now().into());
-        Ok(self)
-    }
 }
 
 impl Entity {

server/svix-server/src/db/models/applicationmetadata.rs

@@ -0,0 +1,100 @@
+// SPDX-FileCopyrightText: © 2022 Svix Authors
+// SPDX-License-Identifier: MIT
+use crate::core::types::ApplicationId;
+
+use crate::core::types::metadata::Metadata;
+use crate::{ctx, error};
+use chrono::Utc;
+use sea_orm::entity::prelude::*;
+use sea_orm::sea_query::OnConflict;
+use sea_orm::ActiveValue::Set;
+use sea_orm::{ConnectionTrait, TryIntoModel};
+
+#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
+#[sea_orm(table_name = "applicationmetadata")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub id: ApplicationId,
+    pub created_at: DateTimeWithTimeZone,
+    pub updated_at: DateTimeWithTimeZone,
+    pub data: Metadata,
+}
+
+impl Model {
+    pub fn metadata(self) -> Metadata {
+        self.data
+    }
+
+    pub fn new(app_id: ApplicationId) -> Self {
+        ActiveModel::new(app_id, None)
+            .try_into_model()
+            .expect("ActiveModel::create(...) should have set all fields")
+    }
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::application::Entity",
+        from = "Column::Id",
+        to = "super::application::Column::Id",
+        on_update = "NoAction",
+        on_delete = "Restrict"
+    )]
+    Application,
+}
+
+impl Related<super::application::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Application.def()
+    }
+}
+
+impl ActiveModelBehavior for ActiveModel {
+    fn before_save(mut self, _insert: bool) -> Result<Self, DbErr> {
+        self.updated_at = Set(Utc::now().into());
+        Ok(self)
+    }
+}
+
+impl ActiveModel {
+    pub fn new(app_id: ApplicationId, metadata: impl Into<Option<Metadata>>) -> Self {
+        let id = Set(app_id);
+        let data = Set(metadata.into().unwrap_or_default());
+        let timestamp = Utc::now();
+        Self {
+            id,
+            data,
+            created_at: Set(timestamp.into()),
+            updated_at: Set(timestamp.into()),
+        }
+    }
+
+    /// Upserts the record if it's new or updated, AND data is nonempty. Otherwise the record is
+    /// ignored or destroyed as appropriate.
+    pub async fn upsert_or_delete(self, db: &impl ConnectionTrait) -> error::Result<Model> {
+        let data = self.data.clone().take().unwrap_or_default();
+
+        if data.is_empty() {
+            let model = ctx!(self.clone().try_into_model())?;
+            ctx!(self.delete(db).await)?;
+            return Ok(model);
+        }
+
+        ctx!(Entity::upsert(self).exec_with_returning(db).await)
+    }
+}
+
+impl Entity {
+    pub fn secure_find(app_id: ApplicationId) -> sea_orm::Select<Entity> {
+        Self::find().filter(Column::Id.eq(app_id))
+    }
+
+    pub fn upsert(am: ActiveModel) -> sea_orm::Insert<ActiveModel> {
+        Self::insert(am).on_conflict(
+            OnConflict::column(Column::Id)
+                .update_columns([Column::Data, Column::UpdatedAt])
+                .to_owned(),
+        )
+    }
+}