Deal with missing created_at

hook-api/src/handlers/webhook.rs

@@ -1,6 +1,7 @@
 use std::time::Instant;
 
 use axum::{extract::State, http::StatusCode, Json};
+use chrono::Utc;
 use hook_common::pgqueue::{NewJob, PgQueue};
 use hook_common::webhook::{WebhookJobMetadata, WebhookJobParameters};
 use serde::Serialize;
@@ -32,7 +33,7 @@ fn default_max_attempts() -> u32 {
 
 pub async fn post(
     State(pg_queue): State<PgQueue>,
-    Json(payload): Json<WebhookPostRequestBody>,
+    Json(mut payload): Json<WebhookPostRequestBody>,
 ) -> Result<Json<WebhookPostResponse>, (StatusCode, Json<WebhookPostResponse>)> {
     debug!("received payload: {:?}", payload);
 
@@ -55,6 +56,11 @@ pub async fn post(
             }),
         )
     })?;
+
+    if payload.metadata.created_at.is_none() {
+        payload.metadata.created_at = Some(Utc::now());
+    }
+
     let job = NewJob::new(
         max_attempts,
         payload.metadata,

hook-common/src/kafka_messages/mod.rs

@@ -11,18 +11,53 @@ where
     serializer.serialize_str(&datetime.format("%Y-%m-%d %H:%M:%S").to_string())
 }
 
+pub fn serialize_datetime_option<S>(
+    datetime: &Option<DateTime<Utc>>,
+    serializer: S,
+) -> Result<S::Ok, S::Error>
+where
+    S: Serializer,
+{
+    match datetime {
+        Some(datetime) => serialize_datetime(datetime, serializer),
+        None => serializer.serialize_none(),
+    }
+}
+
 pub fn deserialize_datetime<'de, D>(deserializer: D) -> Result<DateTime<Utc>, D::Error>
 where
     D: Deserializer<'de>,
 {
     let formatted: String = Deserialize::deserialize(deserializer)?;
-    let datetime = match DateTime::parse_from_rfc3339(&formatted) {
-        Ok(d) => d.with_timezone(&Utc),
-        Err(_) => match NaiveDateTime::parse_from_str(&formatted, "%Y-%m-%d %H:%M:%S") {
-            Ok(d) => d.and_utc(),
-            Err(_) => return Err(serde::de::Error::custom("Invalid datetime format")),
-        },
-    };
+    let datetime = DateTime::parse_from_rfc3339(&formatted)
+        .map(|d| d.with_timezone(&Utc))
+        .or_else(|_| {
+            NaiveDateTime::parse_from_str(&formatted, "%Y-%m-%d %H:%M:%S").map(|d| d.and_utc())
+        })
+        .map_err(|e| serde::de::Error::custom(format!("Invalid datetime format: {}", e)))?;
 
     Ok(datetime)
 }
+
+pub fn deserialize_datetime_option<'de, D>(
+    deserializer: D,
+) -> Result<Option<DateTime<Utc>>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let opt = Option::<String>::deserialize(deserializer)?;
+    match opt {
+        Some(formatted) => {
+            let datetime = DateTime::parse_from_rfc3339(&formatted)
+                .map(|d| d.with_timezone(&Utc))
+                .or_else(|_| {
+                    NaiveDateTime::parse_from_str(&formatted, "%Y-%m-%d %H:%M:%S")
+                        .map(|d| d.and_utc())
+                })
+                .map_err(|e| serde::de::Error::custom(format!("Invalid datetime format: {}", e)))?;
+
+            Ok(Some(datetime))
+        }
+        None => Ok(None),
+    }
+}

hook-common/src/webhook.rs

@@ -7,7 +7,7 @@ use chrono::{DateTime, Utc};
 use serde::{de::Visitor, Deserialize, Serialize};
 
 use crate::kafka_messages::app_metrics;
-use crate::kafka_messages::{deserialize_datetime, serialize_datetime};
+use crate::kafka_messages::{deserialize_datetime_option, serialize_datetime_option};
 use crate::pgqueue::PgQueueError;
 
 /// Supported HTTP methods for webhooks.
@@ -138,10 +138,11 @@ pub struct WebhookJobMetadata {
     pub plugin_id: i32,
     pub plugin_config_id: i32,
     #[serde(
-        serialize_with = "serialize_datetime",
-        deserialize_with = "deserialize_datetime"
+        serialize_with = "serialize_datetime_option",
+        deserialize_with = "deserialize_datetime_option",
+        skip_serializing_if = "Option::is_none"
     )]
-    pub created_at: DateTime<Utc>,
+    pub created_at: Option<DateTime<Utc>>,
 }
 
 /// An error originating during a Webhook Job invocation.

hook-worker/src/worker.rs

@@ -265,9 +265,11 @@ async fn process_webhook_job<W: WebhookJob>(
 
     match send_result {
         Ok(_) => {
-            let end_to_end_duration = Utc::now() - webhook_job.metadata().created_at;
-            metrics::histogram!("webhook_jobs_end_to_end_duration_seconds", &labels)
-                .record((end_to_end_duration.num_milliseconds() as f64) / 1_000_f64);
+            if let Some(created_at) = webhook_job.metadata().created_at {
+                let end_to_end_duration = Utc::now() - created_at;
+                metrics::histogram!("webhook_jobs_end_to_end_duration_seconds", &labels)
+                    .record((end_to_end_duration.num_milliseconds() as f64) / 1_000_f64);
+            }
 
             webhook_job
                 .complete()