Bug: refactor(kafka_message): NanoSecond precision for consolidated logs



Due to millisecond level precision earlier , sessionizer was getting updates in same millisecond due to which updates were not getting picked up. Need to change this behaviour to  NanoSecond precision
diff --git a/crates/router/src/services/kafka/dispute_event.rs b/crates/router/src/services/kafka/dispute_event.rs
index 92327c044d7..45e7ff2c798 100644
--- a/crates/router/src/services/kafka/dispute_event.rs
+++ b/crates/router/src/services/kafka/dispute_event.rs
@@ -20,15 +20,15 @@ pub struct KafkaDisputeEvent<'a> {
     pub connector_dispute_id: &'a String,
     pub connector_reason: Option<&'a String>,
     pub connector_reason_code: Option<&'a String>,
-    #[serde(default, with = "time::serde::timestamp::milliseconds::option")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds::option")]
     pub challenge_required_by: Option<OffsetDateTime>,
-    #[serde(default, with = "time::serde::timestamp::milliseconds::option")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds::option")]
     pub connector_created_at: Option<OffsetDateTime>,
-    #[serde(default, with = "time::serde::timestamp::milliseconds::option")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds::option")]
     pub connector_updated_at: Option<OffsetDateTime>,
-    #[serde(default, with = "time::serde::timestamp::milliseconds")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds")]
     pub created_at: OffsetDateTime,
-    #[serde(default, with = "time::serde::timestamp::milliseconds")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds")]
     pub modified_at: OffsetDateTime,
     pub connector: &'a String,
     pub evidence: &'a Secret<serde_json::Value>,
diff --git a/crates/router/src/services/kafka/payment_attempt_event.rs b/crates/router/src/services/kafka/payment_attempt_event.rs
index 177b211ae89..ec67dc75953 100644
--- a/crates/router/src/services/kafka/payment_attempt_event.rs
+++ b/crates/router/src/services/kafka/payment_attempt_event.rs
@@ -25,15 +25,15 @@ pub struct KafkaPaymentAttemptEvent<'a> {
     pub payment_method: Option<storage_enums::PaymentMethod>,
     pub connector_transaction_id: Option<&'a String>,
     pub capture_method: Option<storage_enums::CaptureMethod>,
-    #[serde(default, with = "time::serde::timestamp::milliseconds::option")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds::option")]
     pub capture_on: Option<OffsetDateTime>,
     pub confirm: bool,
     pub authentication_type: Option<storage_enums::AuthenticationType>,
-    #[serde(with = "time::serde::timestamp::milliseconds")]
+    #[serde(with = "time::serde::timestamp::nanoseconds")]
     pub created_at: OffsetDateTime,
-    #[serde(with = "time::serde::timestamp::milliseconds")]
+    #[serde(with = "time::serde::timestamp::nanoseconds")]
     pub modified_at: OffsetDateTime,
-    #[serde(default, with = "time::serde::timestamp::milliseconds::option")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds::option")]
     pub last_synced: Option<OffsetDateTime>,
     pub cancellation_reason: Option<&'a String>,
     pub amount_to_capture: Option<MinorUnit>,
diff --git a/crates/router/src/services/kafka/payment_intent_event.rs b/crates/router/src/services/kafka/payment_intent_event.rs
index 7509f711620..195b8679d31 100644
--- a/crates/router/src/services/kafka/payment_intent_event.rs
+++ b/crates/router/src/services/kafka/payment_intent_event.rs
@@ -22,11 +22,11 @@ pub struct KafkaPaymentIntentEvent<'a> {
     pub connector_id: Option<&'a String>,
     pub statement_descriptor_name: Option<&'a String>,
     pub statement_descriptor_suffix: Option<&'a String>,
-    #[serde(with = "time::serde::timestamp::milliseconds")]
+    #[serde(with = "time::serde::timestamp::nanoseconds")]
     pub created_at: OffsetDateTime,
-    #[serde(with = "time::serde::timestamp::milliseconds")]
+    #[serde(with = "time::serde::timestamp::nanoseconds")]
     pub modified_at: OffsetDateTime,
-    #[serde(default, with = "time::serde::timestamp::milliseconds::option")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds::option")]
     pub last_synced: Option<OffsetDateTime>,
     pub setup_future_usage: Option<storage_enums::FutureUsage>,
     pub off_session: Option<bool>,
@@ -60,11 +60,11 @@ pub struct KafkaPaymentIntentEvent<'a> {
     pub return_url: Option<&'a String>,
     pub metadata: Option<String>,
     pub statement_descriptor: Option<&'a String>,
-    #[serde(with = "time::serde::timestamp::milliseconds")]
+    #[serde(with = "time::serde::timestamp::nanoseconds")]
     pub created_at: OffsetDateTime,
-    #[serde(with = "time::serde::timestamp::milliseconds")]
+    #[serde(with = "time::serde::timestamp::nanoseconds")]
     pub modified_at: OffsetDateTime,
-    #[serde(default, with = "time::serde::timestamp::milliseconds::option")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds::option")]
     pub last_synced: Option<OffsetDateTime>,
     pub setup_future_usage: Option<storage_enums::FutureUsage>,
     pub off_session: Option<bool>,
diff --git a/crates/router/src/services/kafka/refund_event.rs b/crates/router/src/services/kafka/refund_event.rs
index 74278944b04..b9b3db17b58 100644
--- a/crates/router/src/services/kafka/refund_event.rs
+++ b/crates/router/src/services/kafka/refund_event.rs
@@ -24,9 +24,9 @@ pub struct KafkaRefundEvent<'a> {
     pub sent_to_gateway: &'a bool,
     pub refund_error_message: Option<&'a String>,
     pub refund_arn: Option<&'a String>,
-    #[serde(default, with = "time::serde::timestamp::milliseconds")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds")]
     pub created_at: OffsetDateTime,
-    #[serde(default, with = "time::serde::timestamp::milliseconds")]
+    #[serde(default, with = "time::serde::timestamp::nanoseconds")]
     pub modified_at: OffsetDateTime,
     pub description: Option<&'a String>,
     pub attempt_id: &'a String,

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [x] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
Adding nanosecond level precision to kafka messages for sessionizer

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
So due to millisecond level precision earlier , sessionizer was getting updates in same millisecond due to which updates we not getting picked up

![Screenshot](https://github.com/user-attachments/assets/b1bad009-86db-45d5-8cc5-30b28a8880ae)

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?

Post deployment, the timestamp columns would have nano-second based epoch value
![Screenshot](https://github.com/user-attachments/assets/46a3641d-92ae-4cbb-8af6-9e147f8c529d)

-->
Via Kafka UI

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [ ] I formatted the code `cargo +nightly fmt --all`
- [ ] I addressed lints thrown by `cargo clippy`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
