Use JS Date for message timestamps

Signed-off-by: Didier Wenzek <[email protected]>

Author: didier-wenzek

crates/extensions/tedge_flows/src/flow.rs

@@ -404,11 +404,6 @@ impl DateTime {
         DateTime::try_from(OffsetDateTime::now_utc()).unwrap()
     }
 
-    pub fn tick_now(&self, tick_every: std::time::Duration) -> bool {
-        let tick_every_secs = tick_every.as_secs();
-        tick_every_secs != 0 && (self.seconds % tick_every_secs == 0)
-    }
-
     pub fn json(&self) -> Value {
         json!({"seconds": self.seconds, "nanoseconds": self.nanoseconds})
     }
@@ -450,12 +445,6 @@ impl Message {
         }
     }
 
-    #[cfg(test)]
-    pub fn sent_now(mut self) -> Self {
-        self.timestamp = Some(DateTime::now());
-        self
-    }
-
     pub fn json(&self) -> Value {
         if let Some(timestamp) = &self.timestamp {
             json!({"topic": self.topic, "payload": self.payload, "timestamp": timestamp.json()})

crates/extensions/tedge_flows/src/js_script.rs

@@ -181,7 +181,7 @@ mod tests {
         let js = "export function onMessage(msg) { return [msg]; };";
         let (runtime, script) = runtime_with(js).await;
 
-        let input = Message::new("te/main/device///m/", "hello world").sent_now();
+        let input = Message::new("te/main/device///m/", "hello world");
         let output = input.clone();
         assert_eq!(
             script
@@ -197,7 +197,7 @@ mod tests {
         let js = "export function onMessage(msg) { return msg; };";
         let (runtime, script) = runtime_with(js).await;
 
-        let input = Message::new("te/main/device///m/", "hello world").sent_now();
+        let input = Message::new("te/main/device///m/", "hello world");
         let output = input.clone();
         assert_eq!(
             script
@@ -314,6 +314,37 @@ export async function onMessage(message, config) {
         );
     }
 
+    #[tokio::test]
+    async fn using_date() {
+        let js = r#"
+export function onMessage(message, config) {
+    let time = message.timestamp;
+    return {
+        "topic": message.topic,
+        "payload": JSON.stringify({
+            "milliseconds": time.getTime(),
+            "date": time.toUTCString(),
+        })
+    }
+}
+        "#;
+        let (runtime, script) = runtime_with(js).await;
+
+        let datetime = DateTime {
+            seconds: 1763050414,
+            nanoseconds: 0,
+        };
+        let input = Message::new("clock", "");
+        let output = Message::new(
+            "clock",
+            r#"{"milliseconds":1763050414000,"date":"Thu, 13 Nov 2025 16:13:34 GMT"}"#.to_string(),
+        );
+        assert_eq!(
+            script.on_message(&runtime, datetime, &input).await.unwrap(),
+            vec![output]
+        );
+    }
+
     #[tokio::test]
     async fn using_unknown_function() {
         let js = r#"

crates/extensions/tedge_flows/src/js_value.rs

@@ -5,16 +5,18 @@ use rquickjs::Ctx;
 use rquickjs::FromJs;
 use rquickjs::IntoJs;
 use rquickjs::Value;
+use serde_json::json;
 use std::collections::BTreeMap;
 
-/// Akin to serde_json::Value with an extra case for binary data
-#[derive(Clone, Debug)]
+/// Akin to serde_json::Value with extra cases for date and binary data
+#[derive(Clone, Debug, Eq, PartialEq)]
 pub enum JsonValue {
     Null,
     Bool(bool),
     Number(serde_json::Number),
     String(String),
     Bytes(Vec<u8>), // <= This case motivates the use of JsonValue vs serde_json::Value
+    DateTime(DateTime),
     Array(Vec<JsonValue>),
     Object(BTreeMap<String, JsonValue>),
 }
@@ -30,10 +32,6 @@ impl JsonValue {
         JsonValue::String(value.to_string())
     }
 
-    fn number(value: impl Into<serde_json::Number>) -> Self {
-        JsonValue::Number(value.into())
-    }
-
     fn option(value: Option<impl Into<JsonValue>>) -> Self {
         value.map(|v| v.into()).unwrap_or(JsonValue::Null)
     }
@@ -69,10 +67,7 @@ impl From<Message> for JsonValue {
 
 impl From<DateTime> for JsonValue {
     fn from(value: DateTime) -> Self {
-        JsonValue::object([
-            ("seconds", JsonValue::number(value.seconds)),
-            ("nanoseconds", JsonValue::number(value.nanoseconds)),
-        ])
+        JsonValue::DateTime(value)
     }
 }
 
@@ -101,6 +96,7 @@ impl From<JsonValue> for serde_json::Value {
             JsonValue::Number(n) => serde_json::Value::Number(n),
             JsonValue::String(s) => serde_json::Value::String(s),
             JsonValue::Bytes(b) => serde_json::Value::String(format!("0x {b:?}")),
+            JsonValue::DateTime(t) => json!({ "seconds": t.seconds, "nanos": t.nanoseconds }),
             JsonValue::Array(a) => {
                 serde_json::Value::Array(a.into_iter().map(serde_json::Value::from).collect())
             }
@@ -131,15 +127,11 @@ impl TryFrom<BTreeMap<String, JsonValue>> for Message {
                 .into())
             }
         };
-        let timestamp = value
-            .get("timestamp")
-            .map(|t| DateTime::try_from(t.clone()))
-            .transpose()?;
 
         Ok(Message {
             topic: topic.to_owned(),
             payload,
-            timestamp,
+            timestamp: None,
         })
     }
 }
@@ -157,37 +149,6 @@ impl TryFrom<JsonValue> for Message {
     }
 }
 
-impl TryFrom<JsonValue> for DateTime {
-    type Error = FlowError;
-
-    fn try_from(value: JsonValue) -> Result<Self, Self::Error> {
-        let JsonValue::Object(object) = value else {
-            return Err(
-                anyhow::anyhow!("Expect a timestamp object with seconds and nanoseconds").into(),
-            );
-        };
-        DateTime::try_from(object)
-    }
-}
-
-impl TryFrom<BTreeMap<String, JsonValue>> for DateTime {
-    type Error = FlowError;
-
-    fn try_from(value: BTreeMap<String, JsonValue>) -> Result<Self, Self::Error> {
-        let Some(JsonValue::Number(seconds)) = value.get("seconds") else {
-            return Err(anyhow::anyhow!("Missing timestamp seconds").into());
-        };
-        let Some(JsonValue::Number(nanoseconds)) = value.get("nanoseconds") else {
-            return Err(anyhow::anyhow!("Missing timestamp nanoseconds").into());
-        };
-
-        Ok(DateTime {
-            seconds: seconds.as_u64().unwrap_or_default(),
-            nanoseconds: nanoseconds.as_u64().unwrap_or_default() as u32,
-        })
-    }
-}
-
 impl TryFrom<JsonValue> for Vec<Message> {
     type Error = FlowError;
 
@@ -242,6 +203,12 @@ impl<'js> IntoJs<'js> for JsonValueRef<'_> {
                 let bytes = rquickjs::TypedArray::new(ctx.clone(), value.clone())?;
                 Ok(bytes.into_value())
             }
+            JsonValue::DateTime(value) => {
+                let seconds = value.seconds;
+                let milliseconds = value.nanoseconds / 1_000_000;
+                let time: Value<'js> = ctx.eval(format!("new Date({seconds}{milliseconds:03})"))?;
+                Ok(time)
+            }
             JsonValue::Array(values) => {
                 let array = rquickjs::Array::new(ctx.clone())?;
                 for (i, value) in values.iter().enumerate() {

docs/src/references/mappers/flows.md

@@ -79,12 +79,7 @@ type Message = {
   topic: string,
   payload: string,
   raw_payload: Uint8Array,
-  timestamp: Timestamp
-}
-
-type Timestamp = {
-  seconds: number,
-  nanoseconds: number
+  timestamp: Date
 }
 ```
 
@@ -93,7 +88,7 @@ These values are configured by the flow and can be dynamically updated on recept
 
 The `onMessage` function is called for each message to be transformed
   - The arguments passed to the function are:
-    - The message `{ topic: string, payload: string, raw_payload: Uint8Array, timestamp: { seconds: u64, nanoseconds: u32 } }`
+    - The message `{ topic: string, payload: string, raw_payload: Uint8Array, timestamp: Date }`
     - The config as read from the flow config or updated by the script
   - The function is expected to return zero, one or many transformed messages `[{ topic: string, payload: string }]`
   - An exception can be thrown if the input message cannot be transformed.