Ensure messagecontent expiration is set (#1943)

Fixes #1933

Author: jaymell

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

@@ -35,13 +35,13 @@ impl Related<super::message::Entity> for Entity {
 impl ActiveModelBehavior for ActiveModel {}
 
 impl ActiveModel {
-    pub fn new(msg_id: MessageId, payload: Vec<u8>) -> Self {
+    pub fn new(msg_id: MessageId, payload: Vec<u8>, expiration: DateTimeWithTimeZone) -> Self {
         let timestamp = Utc::now();
         Self {
             id: Set(msg_id),
             created_at: Set(timestamp.into()),
             payload: Set(payload),
-            ..ActiveModelTrait::default()
+            expiration: Set(expiration),
         }
     }
 }

server/svix-server/src/v1/endpoints/message.rs

@@ -382,7 +382,8 @@ pub(crate) async fn create_message_inner(
         .transaction(|txn| {
             async move {
                 let msg = msg.insert(txn).await.map_err(http_error_on_conflict)?;
-                let msg_content = messagecontent::ActiveModel::new(msg.id.clone(), payload);
+                let msg_content =
+                    messagecontent::ActiveModel::new(msg.id.clone(), payload, msg.expiration);
                 let msg_content = msg_content.insert(txn).await?;
                 Ok((msg, msg_content))
             }

server/svix-server/tests/it/e2e_message.rs

@@ -423,6 +423,45 @@ async fn test_payload_retention_period() {
     assert!(content.is_none());
 }
 
+#[tokio::test]
+async fn test_payload_retention_period_messagecontent() {
+    let (client, _jh) = start_svix_server().await;
+    dotenvy::dotenv().ok();
+    let cfg = svix_server::cfg::load().expect("Error loading configuration");
+    let pool = svix_server::db::init_db(&cfg).await;
+
+    let app_id = create_test_app(&client, "test-content-expiration-period")
+        .await
+        .unwrap()
+        .id;
+
+    let custom_retention_period = 5;
+    let msg: MessageOut = client
+        .post(
+            &format!("api/v1/app/{app_id}/msg/"),
+            json!({
+                "eventType": "test.event",
+                "payload": { "test": "value" },
+                "payloadRetentionPeriod": custom_retention_period
+            }),
+            StatusCode::ACCEPTED,
+        )
+        .await
+        .unwrap();
+    let msg_id = msg.id.clone();
+
+    let content: messagecontent::Model = messagecontent::Entity::find_by_id(msg_id.clone())
+        .one(&pool)
+        .await
+        .unwrap()
+        .unwrap();
+
+    let expected = Utc::now() + Duration::days(custom_retention_period) + Duration::hours(1);
+    let actual: chrono::DateTime<Utc> = content.expiration.into();
+
+    assert!(actual < expected);
+}
+
 #[tokio::test]
 async fn test_expunge_message_payload() {
     let (client, _jh) = start_svix_server().await;