Merge pull request #5 from pratham-ez/feat/bento_kafka_to_clickhouse

Feat/bento kafka to clickhouse

Author: nkmishra1997

internal/custom-kafka-to-flexprice-clickhouse.yaml

@@ -53,7 +53,16 @@ pipeline:
         root.environment_id = env("FLEXPRICE_ENVIRONMENT_ID").string().or("")
         root.external_customer_id = this.orgId.string().or("")
         root.source = this.targetItemId.string().or("")
-        root.timestamp = this.createdAt.string().or("")
+        
+        # Timestamp: parse ISO 8601 with timezone, then format for ClickHouse
+        # 1. Check if createdAt exists and is valid
+        # 2. Parse it as timestamp to preserve timezone info
+        # 3. Format to ClickHouse DateTime64 format: "2006-01-02 15:04:05.999999"
+        root.timestamp = if this.createdAt != null && this.createdAt.string().or("") != "" && this.createdAt.string() != "null" {
+          this.createdAt.ts_parse("2006-01-02T15:04:05.999Z07:00").ts_format("2006-01-02 15:04:05.999999")
+        } else {
+          timestamp_unix().ts_format("2006-01-02 15:04:05.999999")
+        }
 
         # ---- Build event_name SAFELY ----
         root.event_name = ""