Bug: feat(opensearch): add additional filters and create sessionizer indexes for local



Global search has two components:

- Free flow search, wherein search takes place across all fields available to search upon
- Search on a specific field based on the field filter

Additional filters should be added for global search to search on a specific field.
A list of values can be provided for every filter field which will be added to the query in order to extract responses from opensearch / elasticsearch.

Existing filters (fields):
- payment_method
- currency
- status
- customer_email
- search_tags

New filters to be added (fields):
- connector
- payment_method_type
- card_network
- card_last_4
- payment_id
diff --git a/config/config.example.toml b/config/config.example.toml
index c0358d91a26..8f04ba49831 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -728,6 +728,10 @@ payment_attempts = "hyperswitch-payment-attempt-events"
 payment_intents = "hyperswitch-payment-intent-events"
 refunds = "hyperswitch-refund-events"
 disputes = "hyperswitch-dispute-events"
+sessionizer_payment_attempts = "sessionizer-payment-attempt-events"
+sessionizer_payment_intents = "sessionizer-payment-intent-events"
+sessionizer_refunds = "sessionizer-refund-events"
+sessionizer_disputes = "sessionizer-dispute-events"
 
 [saved_payment_methods]
 sdk_eligible_payment_methods = "card"
diff --git a/config/deployments/env_specific.toml b/config/deployments/env_specific.toml
index 5dce5be9c96..cfefe9a130f 100644
--- a/config/deployments/env_specific.toml
+++ b/config/deployments/env_specific.toml
@@ -252,6 +252,10 @@ payment_attempts = "hyperswitch-payment-attempt-events"
 payment_intents = "hyperswitch-payment-intent-events"
 refunds = "hyperswitch-refund-events"
 disputes = "hyperswitch-dispute-events"
+sessionizer_payment_attempts = "sessionizer-payment-attempt-events"
+sessionizer_payment_intents = "sessionizer-payment-intent-events"
+sessionizer_refunds = "sessionizer-refund-events"
+sessionizer_disputes = "sessionizer-dispute-events"
 
 # Configuration for the Key Manager Service
 [key_manager]
diff --git a/config/development.toml b/config/development.toml
index 0153c6ef102..29cd2dd048d 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -732,6 +732,10 @@ payment_attempts = "hyperswitch-payment-attempt-events"
 payment_intents = "hyperswitch-payment-intent-events"
 refunds = "hyperswitch-refund-events"
 disputes = "hyperswitch-dispute-events"
+sessionizer_payment_attempts = "sessionizer-payment-attempt-events"
+sessionizer_payment_intents = "sessionizer-payment-intent-events"
+sessionizer_refunds = "sessionizer-refund-events"
+sessionizer_disputes = "sessionizer-dispute-events"
 
 [saved_payment_methods]
 sdk_eligible_payment_methods = "card"
diff --git a/config/docker_compose.toml b/config/docker_compose.toml
index 5ffe5df138f..006b017de0b 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -556,6 +556,10 @@ payment_attempts = "hyperswitch-payment-attempt-events"
 payment_intents = "hyperswitch-payment-intent-events"
 refunds = "hyperswitch-refund-events"
 disputes = "hyperswitch-dispute-events"
+sessionizer_payment_attempts = "sessionizer-payment-attempt-events"
+sessionizer_payment_intents = "sessionizer-payment-intent-events"
+sessionizer_refunds = "sessionizer-refund-events"
+sessionizer_disputes = "sessionizer-dispute-events"
 
 [saved_payment_methods]
 sdk_eligible_payment_methods = "card"
diff --git a/config/vector.yaml b/config/vector.yaml
index 54ff25cab5a..3f0709ae03c 100644
--- a/config/vector.yaml
+++ b/config/vector.yaml
@@ -18,6 +18,15 @@ sources:
     decoding:
       codec: json
 
+  sessionized_kafka_tx_events:
+    type: kafka
+    bootstrap_servers: kafka0:29092
+    group_id: sessionizer
+    topics:
+      - ^sessionizer
+    decoding:
+      codec: json
+
   app_logs:
     type: docker_logs
     include_labels:
@@ -35,10 +44,19 @@ sources:
     encoding: json
 
 transforms:
+  events_create_ts:
+    inputs:
+    - kafka_tx_events
+    source: |-
+      .timestamp = from_unix_timestamp(.created_at, unit: "seconds") ?? now()
+      ."@timestamp" = from_unix_timestamp(.created_at, unit: "seconds") ?? now()
+    type: remap
+
   plus_1_events:
     type: filter
     inputs:
-      - kafka_tx_events
+      - events_create_ts
+      - sessionized_events_create_ts
     condition: ".sign_flag == 1"
 
   hs_server_logs:
@@ -54,13 +72,13 @@ transforms:
     source: |-
       .message = parse_json!(.message)
 
-  events:
+  sessionized_events_create_ts:
     type: remap
     inputs:
-      - plus_1_events
+    - sessionized_kafka_tx_events
     source: |-
-      .timestamp = from_unix_timestamp!(.created_at, unit: "seconds")
-      ."@timestamp" = from_unix_timestamp(.created_at, unit: "seconds") ?? now()
+      .timestamp = from_unix_timestamp(.created_at, unit: "milliseconds") ?? now()
+      ."@timestamp" = from_unix_timestamp(.created_at, unit: "milliseconds") ?? now()
 
   sdk_transformed:
     type: throttle
@@ -74,7 +92,7 @@ sinks:
   opensearch_events_1:
     type: elasticsearch
     inputs:
-      - events
+      - plus_1_events
     endpoints:
       - "https://opensearch:9200"
     id_key: message_key
@@ -98,7 +116,7 @@ sinks:
   opensearch_events_2:
     type: elasticsearch
     inputs:
-      - events
+      - plus_1_events
     endpoints:
       - "https://opensearch:9200"
     id_key: message_key
@@ -120,6 +138,33 @@ sinks:
       # Add a date suffixed index for better grouping
       index: "vector-{{ .topic }}-%Y-%m-%d"
 
+  opensearch_events_3:
+    type: elasticsearch
+    inputs:
+      - plus_1_events
+    endpoints:
+      - "https://opensearch:9200"
+    id_key: message_key
+    api_version: v7
+    tls:
+      verify_certificate: false
+      verify_hostname: false
+    auth:
+      strategy: basic
+      user: admin
+      password: 0penS3arc#
+    encoding:
+      except_fields:
+        - message_key
+        - offset
+        - partition
+        - topic
+        - clickhouse_database
+        - last_synced
+        - sign_flag
+    bulk:
+      index: "{{ .topic }}"
+
   opensearch_logs:
     type: elasticsearch
     inputs:
@@ -143,6 +188,7 @@ sinks:
     type: loki
     inputs:
       - kafka_tx_events
+      - sessionized_kafka_tx_events
     endpoint: http://loki:3100
     labels:
       source: vector
diff --git a/crates/analytics/src/opensearch.rs b/crates/analytics/src/opensearch.rs
index 3be9688d8f7..a6e6c486ebe 100644
--- a/crates/analytics/src/opensearch.rs
+++ b/crates/analytics/src/opensearch.rs
@@ -42,6 +42,10 @@ pub struct OpenSearchIndexes {
     pub payment_intents: String,
     pub refunds: String,
     pub disputes: String,
+    pub sessionizer_payment_attempts: String,
+    pub sessionizer_payment_intents: String,
+    pub sessionizer_refunds: String,
+    pub sessionizer_disputes: String,
 }
 
 #[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
@@ -81,6 +85,10 @@ impl Default for OpenSearchConfig {
                 payment_intents: "hyperswitch-payment-intent-events".to_string(),
                 refunds: "hyperswitch-refund-events".to_string(),
                 disputes: "hyperswitch-dispute-events".to_string(),
+                sessionizer_payment_attempts: "sessionizer-payment-attempt-events".to_string(),
+                sessionizer_payment_intents: "sessionizer-payment-intent-events".to_string(),
+                sessionizer_refunds: "sessionizer-refund-events".to_string(),
+                sessionizer_disputes: "sessionizer-dispute-events".to_string(),
             },
         }
     }
@@ -219,6 +227,14 @@ impl OpenSearchClient {
             SearchIndex::PaymentIntents => self.indexes.payment_intents.clone(),
             SearchIndex::Refunds => self.indexes.refunds.clone(),
             SearchIndex::Disputes => self.indexes.disputes.clone(),
+            SearchIndex::SessionizerPaymentAttempts => {
+                self.indexes.sessionizer_payment_attempts.clone()
+            }
+            SearchIndex::SessionizerPaymentIntents => {
+                self.indexes.sessionizer_payment_intents.clone()
+            }
+            SearchIndex::SessionizerRefunds => self.indexes.sessionizer_refunds.clone(),
+            SearchIndex::SessionizerDisputes => self.indexes.sessionizer_disputes.clone(),
         }
     }
 
@@ -324,6 +340,36 @@ impl OpenSearchIndexes {
             ))
         })?;
 
+        when(
+            self.sessionizer_payment_attempts.is_default_or_empty(),
+            || {
+                Err(ApplicationError::InvalidConfigurationValueError(
+                    "Opensearch Sessionizer Payment Attempts index must not be empty".into(),
+                ))
+            },
+        )?;
+
+        when(
+            self.sessionizer_payment_intents.is_default_or_empty(),
+            || {
+                Err(ApplicationError::InvalidConfigurationValueError(
+                    "Opensearch Sessionizer Payment Intents index must not be empty".into(),
+                ))
+            },
+        )?;
+
+        when(self.sessionizer_refunds.is_default_or_empty(), || {
+            Err(ApplicationError::InvalidConfigurationValueError(
+                "Opensearch Sessionizer Refunds index must not be empty".into(),
+            ))
+        })?;
+
+        when(self.sessionizer_disputes.is_default_or_empty(), || {
+            Err(ApplicationError::InvalidConfigurationValueError(
+                "Opensearch Sessionizer Disputes index must not be empty".into(),
+            ))
+        })?;
+
         Ok(())
     }
 }
diff --git a/crates/analytics/src/search.rs b/crates/analytics/src/search.rs
index c81ff2f416b..f53b07b1232 100644
--- a/crates/analytics/src/search.rs
+++ b/crates/analytics/src/search.rs
@@ -92,6 +92,44 @@ pub async fn msearch_results(
                     .switch()?;
             }
         };
+        if let Some(connector) = filters.connector {
+            if !connector.is_empty() {
+                query_builder
+                    .add_filter_clause("connector.keyword".to_string(), connector.clone())
+                    .switch()?;
+            }
+        };
+        if let Some(payment_method_type) = filters.payment_method_type {
+            if !payment_method_type.is_empty() {
+                query_builder
+                    .add_filter_clause(
+                        "payment_method_type.keyword".to_string(),
+                        payment_method_type.clone(),
+                    )
+                    .switch()?;
+            }
+        };
+        if let Some(card_network) = filters.card_network {
+            if !card_network.is_empty() {
+                query_builder
+                    .add_filter_clause("card_network.keyword".to_string(), card_network.clone())
+                    .switch()?;
+            }
+        };
+        if let Some(card_last_4) = filters.card_last_4 {
+            if !card_last_4.is_empty() {
+                query_builder
+                    .add_filter_clause("card_last_4.keyword".to_string(), card_last_4.clone())
+                    .switch()?;
+            }
+        };
+        if let Some(payment_id) = filters.payment_id {
+            if !payment_id.is_empty() {
+                query_builder
+                    .add_filter_clause("payment_id.keyword".to_string(), payment_id.clone())
+                    .switch()?;
+            }
+        };
     };
 
     if let Some(time_range) = req.time_range {
@@ -217,6 +255,44 @@ pub async fn search_results(
                     .switch()?;
             }
         };
+        if let Some(connector) = filters.connector {
+            if !connector.is_empty() {
+                query_builder
+                    .add_filter_clause("connector.keyword".to_string(), connector.clone())
+                    .switch()?;
+            }
+        };
+        if let Some(payment_method_type) = filters.payment_method_type {
+            if !payment_method_type.is_empty() {
+                query_builder
+                    .add_filter_clause(
+                        "payment_method_type.keyword".to_string(),
+                        payment_method_type.clone(),
+                    )
+                    .switch()?;
+            }
+        };
+        if let Some(card_network) = filters.card_network {
+            if !card_network.is_empty() {
+                query_builder
+                    .add_filter_clause("card_network.keyword".to_string(), card_network.clone())
+                    .switch()?;
+            }
+        };
+        if let Some(card_last_4) = filters.card_last_4 {
+            if !card_last_4.is_empty() {
+                query_builder
+                    .add_filter_clause("card_last_4.keyword".to_string(), card_last_4.clone())
+                    .switch()?;
+            }
+        };
+        if let Some(payment_id) = filters.payment_id {
+            if !payment_id.is_empty() {
+                query_builder
+                    .add_filter_clause("payment_id.keyword".to_string(), payment_id.clone())
+                    .switch()?;
+            }
+        };
     };
 
     if let Some(time_range) = search_req.time_range {
diff --git a/crates/api_models/src/analytics/search.rs b/crates/api_models/src/analytics/search.rs
index 24dd0effcbb..a33dd100c79 100644
--- a/crates/api_models/src/analytics/search.rs
+++ b/crates/api_models/src/analytics/search.rs
@@ -9,6 +9,11 @@ pub struct SearchFilters {
     pub status: Option<Vec<String>>,
     pub customer_email: Option<Vec<HashedString<common_utils::pii::EmailStrategy>>>,
     pub search_tags: Option<Vec<HashedString<WithType>>>,
+    pub connector: Option<Vec<String>>,
+    pub payment_method_type: Option<Vec<String>>,
+    pub card_network: Option<Vec<String>>,
+    pub card_last_4: Option<Vec<String>>,
+    pub payment_id: Option<Vec<String>>,
 }
 impl SearchFilters {
     pub fn is_all_none(&self) -> bool {
@@ -17,6 +22,11 @@ impl SearchFilters {
             && self.status.is_none()
             && self.customer_email.is_none()
             && self.search_tags.is_none()
+            && self.connector.is_none()
+            && self.payment_method_type.is_none()
+            && self.card_network.is_none()
+            && self.card_last_4.is_none()
+            && self.payment_id.is_none()
     }
 }
 
@@ -58,6 +68,10 @@ pub enum SearchIndex {
     PaymentIntents,
     Refunds,
     Disputes,
+    SessionizerPaymentAttempts,
+    SessionizerPaymentIntents,
+    SessionizerRefunds,
+    SessionizerDisputes,
 }
 
 #[derive(Debug, strum::EnumIter, Clone, serde::Deserialize, serde::Serialize, Copy)]
diff --git a/crates/router/src/consts/opensearch.rs b/crates/router/src/consts/opensearch.rs
index 277b0e946ba..c9eeec1b343 100644
--- a/crates/router/src/consts/opensearch.rs
+++ b/crates/router/src/consts/opensearch.rs
@@ -1,12 +1,16 @@
 use api_models::analytics::search::SearchIndex;
 
-pub const fn get_search_indexes() -> [SearchIndex; 4] {
+pub const fn get_search_indexes() -> [SearchIndex; 8] {
     [
         SearchIndex::PaymentAttempts,
         SearchIndex::PaymentIntents,
         SearchIndex::Refunds,
         SearchIndex::Disputes,
+        SearchIndex::SessionizerPaymentAttempts,
+        SearchIndex::SessionizerPaymentIntents,
+        SearchIndex::SessionizerRefunds,
+        SearchIndex::SessionizerDisputes,
     ]
 }
 
-pub const SEARCH_INDEXES: [SearchIndex; 4] = get_search_indexes();
+pub const SEARCH_INDEXES: [SearchIndex; 8] = get_search_indexes();

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

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

## Description
<!-- Describe your changes in detail -->
Global search has two components:
- Free flow search, wherein search takes place across all fields available to search upon
- Search on a specific field based on the field filter


In this PR, additional filters have been added for global search to search on a specific field.
A list of values can be provided for every filter field which will be added to the query in order to extract responses from opensearch / elasticsearch.
Existing filters (fields):
- payment_method
- currency
- status
- customer_email
- search_tags

New filters added (fields):
- connector
- payment_method_type
- card_network
- card_last_4
- payment_id

### 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).
-->
Enhance search experience on global search using filters

## 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)?
-->
The following filters can be used to be searched upon:
- connector
- payment_method_type
- card_network
- card_last_4
- payment_id

Hit the following curl for the `/search` API
```bash
curl --location 'http://localhost:8080/analytics/v1/search' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36' \
--header 'api-key: hyperswitch' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyOTI0ODM1MCwib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb192NXNGb0hlODBPZWlVbElvbm9jTSJ9.TQD6wYkiTXlTvPW3-mH8FzEP1d7hXw9SmWgjVMH7flk' \
--header 'sec-ch-ua: "Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '{
    "query": "merchant_1726046328",
    "filters": {
        "payment_method": [
            "card"
        ]
    },
    "timeRange": {
        "startTime": "2024-09-13T00:30:00Z",
        "endTime": "2024-10-18T21:45:00Z"
    }
}'
```
## Checklist
<!-- Put an `x` in the boxes that apply -->

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