Bug: fix(opensearch): show search results only if user has access to the index



/search and /search/{domain} are currently checking for analytics ACL before hitting opensearch and returning results.

we have to change this to check access permissions per index 

payments attempts/intents => PaymentRead | PaymentWrite
refunds => RefundRead | RefundWrite
disputes => DisputeRead | DisputeWrite

can modify GetSearchResponse to include status 

```
enum SearchStatus = Success | Failure | NoAccess

pub struct GetSearchResponse {
    pub count: u64,
    pub index: SearchIndex,
    pub hits: Vec<Value>,
    pub status: SearchStatus
}
```
diff --git a/crates/analytics/src/opensearch.rs b/crates/analytics/src/opensearch.rs
index e8f87aaef2e..b8c582df1bf 100644
--- a/crates/analytics/src/opensearch.rs
+++ b/crates/analytics/src/opensearch.rs
@@ -20,7 +20,6 @@ use opensearch::{
 };
 use serde_json::{json, Value};
 use storage_impl::errors::ApplicationError;
-use strum::IntoEnumIterator;
 
 use super::{health_check::HealthCheck, query::QueryResult, types::QueryExecutionError};
 use crate::query::QueryBuildingError;
@@ -78,6 +77,10 @@ pub enum OpenSearchError {
     QueryBuildingError,
     #[error("Opensearch deserialisation error")]
     DeserialisationError,
+    #[error("Opensearch index access not present error: {0:?}")]
+    IndexAccessNotPermittedError(SearchIndex),
+    #[error("Opensearch unknown error")]
+    UnknownError,
 }
 
 impl ErrorSwitch<OpenSearchError> for QueryBuildingError {
@@ -97,28 +100,39 @@ impl ErrorSwitch<ApiErrorResponse> for OpenSearchError {
             )),
             Self::ResponseNotOK(response) => ApiErrorResponse::InternalServerError(ApiError::new(
                 "IR",
-                0,
+                1,
                 format!("Something went wrong {}", response),
                 None,
             )),
             Self::ResponseError => ApiErrorResponse::InternalServerError(ApiError::new(
                 "IR",
-                0,
+                2,
                 "Something went wrong",
                 None,
             )),
             Self::QueryBuildingError => ApiErrorResponse::InternalServerError(ApiError::new(
                 "IR",
-                0,
+                3,
                 "Query building error",
                 None,
             )),
             Self::DeserialisationError => ApiErrorResponse::InternalServerError(ApiError::new(
                 "IR",
-                0,
+                4,
                 "Deserialisation error",
                 None,
             )),
+            Self::IndexAccessNotPermittedError(index) => {
+                ApiErrorResponse::ForbiddenCommonResource(ApiError::new(
+                    "IR",
+                    5,
+                    format!("Index access not permitted: {index:?}"),
+                    None,
+                ))
+            }
+            Self::UnknownError => {
+                ApiErrorResponse::InternalServerError(ApiError::new("IR", 6, "Unknown error", None))
+            }
         }
     }
 }
@@ -179,18 +193,16 @@ impl OpenSearchClient {
         query_builder: OpenSearchQueryBuilder,
     ) -> CustomResult<Response, OpenSearchError> {
         match query_builder.query_type {
-            OpenSearchQuery::Msearch => {
-                let search_indexes = SearchIndex::iter();
-
+            OpenSearchQuery::Msearch(ref indexes) => {
                 let payload = query_builder
-                    .construct_payload(search_indexes.clone().collect())
+                    .construct_payload(indexes)
                     .change_context(OpenSearchError::QueryBuildingError)?;
 
-                let payload_with_indexes = payload.into_iter().zip(search_indexes).fold(
+                let payload_with_indexes = payload.into_iter().zip(indexes).fold(
                     Vec::new(),
                     |mut payload_with_indexes, (index_hit, index)| {
                         payload_with_indexes.push(
-                            json!({"index": self.search_index_to_opensearch_index(index)}).into(),
+                            json!({"index": self.search_index_to_opensearch_index(*index)}).into(),
                         );
                         payload_with_indexes.push(JsonBody::new(index_hit.clone()));
                         payload_with_indexes
@@ -207,7 +219,7 @@ impl OpenSearchClient {
             OpenSearchQuery::Search(index) => {
                 let payload = query_builder
                     .clone()
-                    .construct_payload(vec![index])
+                    .construct_payload(&[index])
                     .change_context(OpenSearchError::QueryBuildingError)?;
 
                 let final_payload = payload.first().unwrap_or(&Value::Null);
@@ -349,7 +361,7 @@ pub struct OpenSearchHealth {
 
 #[derive(Debug, Clone)]
 pub enum OpenSearchQuery {
-    Msearch,
+    Msearch(Vec<SearchIndex>),
     Search(SearchIndex),
 }
 
@@ -384,7 +396,7 @@ impl OpenSearchQueryBuilder {
         Ok(())
     }
 
-    pub fn construct_payload(&self, indexes: Vec<SearchIndex>) -> QueryResult<Vec<Value>> {
+    pub fn construct_payload(&self, indexes: &[SearchIndex]) -> QueryResult<Vec<Value>> {
         let mut query =
             vec![json!({"multi_match": {"type": "phrase", "query": self.query, "lenient": true}})];
 
diff --git a/crates/analytics/src/search.rs b/crates/analytics/src/search.rs
index 8810dc1e3a1..373ca3acbb6 100644
--- a/crates/analytics/src/search.rs
+++ b/crates/analytics/src/search.rs
@@ -1,11 +1,10 @@
 use api_models::analytics::search::{
     GetGlobalSearchRequest, GetSearchRequestWithIndex, GetSearchResponse, OpenMsearchOutput,
-    OpensearchOutput, SearchIndex,
+    OpensearchOutput, SearchIndex, SearchStatus,
 };
 use common_utils::errors::{CustomResult, ReportSwitchExt};
 use error_stack::ResultExt;
 use router_env::tracing;
-use strum::IntoEnumIterator;
 
 use crate::opensearch::{
     OpenSearchClient, OpenSearchError, OpenSearchQuery, OpenSearchQueryBuilder,
@@ -15,8 +14,10 @@ pub async fn msearch_results(
     client: &OpenSearchClient,
     req: GetGlobalSearchRequest,
     merchant_id: &String,
+    indexes: Vec<SearchIndex>,
 ) -> CustomResult<Vec<GetSearchResponse>, OpenSearchError> {
-    let mut query_builder = OpenSearchQueryBuilder::new(OpenSearchQuery::Msearch, req.query);
+    let mut query_builder =
+        OpenSearchQueryBuilder::new(OpenSearchQuery::Msearch(indexes.clone()), req.query);
 
     query_builder
         .add_filter_clause("merchant_id".to_string(), merchant_id.to_string())
@@ -40,29 +41,19 @@ pub async fn msearch_results(
     Ok(response_body
         .responses
         .into_iter()
-        .zip(SearchIndex::iter())
+        .zip(indexes)
         .map(|(index_hit, index)| match index_hit {
-            OpensearchOutput::Success(success) => {
-                if success.status == 200 {
-                    GetSearchResponse {
-                        count: success.hits.total.value,
-                        index,
-                        hits: success
-                            .hits
-                            .hits
-                            .into_iter()
-                            .map(|hit| hit.source)
-                            .collect(),
-                    }
-                } else {
-                    tracing::error!("Unexpected status code: {}", success.status,);
-                    GetSearchResponse {
-                        count: 0,
-                        index,
-                        hits: Vec::new(),
-                    }
-                }
-            }
+            OpensearchOutput::Success(success) => GetSearchResponse {
+                count: success.hits.total.value,
+                index,
+                hits: success
+                    .hits
+                    .hits
+                    .into_iter()
+                    .map(|hit| hit.source)
+                    .collect(),
+                status: SearchStatus::Success,
+            },
             OpensearchOutput::Error(error) => {
                 tracing::error!(
                     index = ?index,
@@ -73,6 +64,7 @@ pub async fn msearch_results(
                     count: 0,
                     index,
                     hits: Vec::new(),
+                    status: SearchStatus::Failure,
                 }
             }
         })
@@ -113,27 +105,17 @@ pub async fn search_results(
     let response_body: OpensearchOutput = response_text;
 
     match response_body {
-        OpensearchOutput::Success(success) => {
-            if success.status == 200 {
-                Ok(GetSearchResponse {
-                    count: success.hits.total.value,
-                    index: req.index,
-                    hits: success
-                        .hits
-                        .hits
-                        .into_iter()
-                        .map(|hit| hit.source)
-                        .collect(),
-                })
-            } else {
-                tracing::error!("Unexpected status code: {}", success.status);
-                Ok(GetSearchResponse {
-                    count: 0,
-                    index: req.index,
-                    hits: Vec::new(),
-                })
-            }
-        }
+        OpensearchOutput::Success(success) => Ok(GetSearchResponse {
+            count: success.hits.total.value,
+            index: req.index,
+            hits: success
+                .hits
+                .hits
+                .into_iter()
+                .map(|hit| hit.source)
+                .collect(),
+            status: SearchStatus::Success,
+        }),
         OpensearchOutput::Error(error) => {
             tracing::error!(
                 index = ?req.index,
@@ -144,6 +126,7 @@ pub async fn search_results(
                 count: 0,
                 index: req.index,
                 hits: Vec::new(),
+                status: SearchStatus::Failure,
             })
         }
     }
diff --git a/crates/api_models/src/analytics/search.rs b/crates/api_models/src/analytics/search.rs
index c29bb0e9f71..a01349d987f 100644
--- a/crates/api_models/src/analytics/search.rs
+++ b/crates/api_models/src/analytics/search.rs
@@ -30,7 +30,9 @@ pub struct GetSearchRequestWithIndex {
     pub search_req: GetSearchRequest,
 }
 
-#[derive(Debug, strum::EnumIter, Clone, serde::Deserialize, serde::Serialize, Copy)]
+#[derive(
+    Debug, strum::EnumIter, Clone, serde::Deserialize, serde::Serialize, Copy, Eq, PartialEq,
+)]
 #[serde(rename_all = "snake_case")]
 pub enum SearchIndex {
     PaymentAttempts,
@@ -39,17 +41,26 @@ pub enum SearchIndex {
     Disputes,
 }
 
+#[derive(Debug, strum::EnumIter, Clone, serde::Deserialize, serde::Serialize, Copy)]
+pub enum SearchStatus {
+    Success,
+    Failure,
+}
+
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GetSearchResponse {
     pub count: u64,
     pub index: SearchIndex,
     pub hits: Vec<Value>,
+    pub status: SearchStatus,
 }
 
 #[derive(Debug, serde::Deserialize)]
 pub struct OpenMsearchOutput {
+    #[serde(default)]
     pub responses: Vec<OpensearchOutput>,
+    pub error: Option<OpensearchErrorDetails>,
 }
 
 #[derive(Debug, serde::Deserialize)]
@@ -74,7 +85,6 @@ pub struct OpensearchErrorDetails {
 
 #[derive(Debug, serde::Deserialize)]
 pub struct OpensearchSuccess {
-    pub status: u16,
     pub hits: OpensearchHits,
 }
 
diff --git a/crates/router/src/analytics.rs b/crates/router/src/analytics.rs
index b471448368f..988870e8c06 100644
--- a/crates/router/src/analytics.rs
+++ b/crates/router/src/analytics.rs
@@ -4,7 +4,7 @@ pub mod routes {
     use actix_web::{web, Responder, Scope};
     use analytics::{
         api_event::api_events_core, connector_events::connector_events_core,
-        errors::AnalyticsError, lambda_utils::invoke_lambda,
+        errors::AnalyticsError, lambda_utils::invoke_lambda, opensearch::OpenSearchError,
         outgoing_webhook_event::outgoing_webhook_events_core, sdk_events::sdk_events_core,
         AnalyticsFlow,
     };
@@ -20,13 +20,14 @@ pub mod routes {
     use error_stack::ResultExt;
 
     use crate::{
-        core::api_locking,
+        consts::opensearch::OPENSEARCH_INDEX_PERMISSIONS,
+        core::{api_locking, errors::user::UserErrors},
         db::user::UserInterface,
         routes::AppState,
         services::{
             api,
-            authentication::{self as auth, AuthenticationData},
-            authorization::permissions::Permission,
+            authentication::{self as auth, AuthenticationData, UserFromToken},
+            authorization::{permissions::Permission, roles::RoleInfo},
             ApplicationResponse,
         },
         types::domain::UserEmail,
@@ -653,11 +654,25 @@ pub mod routes {
             state.clone(),
             &req,
             json_payload.into_inner(),
-            |state, auth: AuthenticationData, req, _| async move {
+            |state, auth: UserFromToken, req, _| async move {
+                let role_id = auth.role_id;
+                let role_info =
+                    RoleInfo::from_role_id(&state, &role_id, &auth.merchant_id, &auth.org_id)
+                        .await
+                        .change_context(UserErrors::InternalServerError)
+                        .change_context(OpenSearchError::UnknownError)?;
+                let permissions = role_info.get_permissions_set();
+                let accessible_indexes: Vec<_> = OPENSEARCH_INDEX_PERMISSIONS
+                    .iter()
+                    .filter(|(_, perm)| perm.iter().any(|p| permissions.contains(p)))
+                    .map(|(i, _)| *i)
+                    .collect();
+
                 analytics::search::msearch_results(
                     &state.opensearch_client,
                     req,
-                    &auth.merchant_account.merchant_id,
+                    &auth.merchant_id,
+                    accessible_indexes,
                 )
                 .await
                 .map(ApplicationResponse::Json)
@@ -674,24 +689,33 @@ pub mod routes {
         json_payload: web::Json<GetSearchRequest>,
         index: web::Path<SearchIndex>,
     ) -> impl Responder {
+        let index = index.into_inner();
         let flow = AnalyticsFlow::GetSearchResults;
         let indexed_req = GetSearchRequestWithIndex {
             search_req: json_payload.into_inner(),
-            index: index.into_inner(),
+            index,
         };
         Box::pin(api::server_wrap(
             flow,
             state.clone(),
             &req,
             indexed_req,
-            |state, auth: AuthenticationData, req, _| async move {
-                analytics::search::search_results(
-                    &state.opensearch_client,
-                    req,
-                    &auth.merchant_account.merchant_id,
-                )
-                .await
-                .map(ApplicationResponse::Json)
+            |state, auth: UserFromToken, req, _| async move {
+                let role_id = auth.role_id;
+                let role_info =
+                    RoleInfo::from_role_id(&state, &role_id, &auth.merchant_id, &auth.org_id)
+                        .await
+                        .change_context(UserErrors::InternalServerError)
+                        .change_context(OpenSearchError::UnknownError)?;
+                let permissions = role_info.get_permissions_set();
+                let _ = OPENSEARCH_INDEX_PERMISSIONS
+                    .iter()
+                    .filter(|(ind, _)| *ind == index)
+                    .find(|i| i.1.iter().any(|p| permissions.contains(p)))
+                    .ok_or(OpenSearchError::IndexAccessNotPermittedError(index))?;
+                analytics::search::search_results(&state.opensearch_client, req, &auth.merchant_id)
+                    .await
+                    .map(ApplicationResponse::Json)
             },
             &auth::JWTAuth(Permission::Analytics),
             api_locking::LockAction::NotApplicable,
diff --git a/crates/router/src/consts.rs b/crates/router/src/consts.rs
index 73ff7c428e4..bffc5241339 100644
--- a/crates/router/src/consts.rs
+++ b/crates/router/src/consts.rs
@@ -1,3 +1,4 @@
+pub mod opensearch;
 #[cfg(feature = "olap")]
 pub mod user;
 pub mod user_role;
diff --git a/crates/router/src/consts/opensearch.rs b/crates/router/src/consts/opensearch.rs
new file mode 100644
index 00000000000..fc1c071439a
--- /dev/null
+++ b/crates/router/src/consts/opensearch.rs
@@ -0,0 +1,22 @@
+use api_models::analytics::search::SearchIndex;
+
+use crate::services::authorization::permissions::Permission;
+
+pub const OPENSEARCH_INDEX_PERMISSIONS: &[(SearchIndex, &[Permission])] = &[
+    (
+        SearchIndex::PaymentAttempts,
+        &[Permission::PaymentRead, Permission::PaymentWrite],
+    ),
+    (
+        SearchIndex::PaymentIntents,
+        &[Permission::PaymentRead, Permission::PaymentWrite],
+    ),
+    (
+        SearchIndex::Refunds,
+        &[Permission::RefundRead, Permission::RefundWrite],
+    ),
+    (
+        SearchIndex::Disputes,
+        &[Permission::DisputeRead, Permission::DisputeWrite],
+    ),
+];
diff --git a/crates/router/src/events.rs b/crates/router/src/events.rs
index 632989c4806..1bdd9016c49 100644
--- a/crates/router/src/events.rs
+++ b/crates/router/src/events.rs
@@ -83,7 +83,7 @@ impl EventsConfig {
 impl EventsHandler {
     pub fn log_event<T: KafkaMessage>(&self, event: &T) {
         match self {
-            Self::Kafka(kafka) => kafka.log_event(event).map_or((), |e| {
+            Self::Kafka(kafka) => kafka.log_event(event).unwrap_or_else(|e| {
                 logger::error!("Failed to log event: {:?}", e);
             }),
             Self::Logs(logger) => logger.log_event(event),

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

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

## Description
<!-- Describe your changes in detail -->
/search and /search/{domain} are currently checking for analytics ACL before hitting opensearch and returning results.

Additionally we have to check access permissions per index:

- payment attempts/intents => PaymentRead | PaymentWrite
- refunds => RefundRead | RefundWrite
- disputes => DisputeRead | DisputeWrite

### 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).
-->

Checking access permissions to an index for a user is necessary to show results appropriately.

## 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)?
-->

- Open the dashboard locally and hit the get_global_search_results API through the global search feature (command+k).
- Search using a query (eg: USD), to get the results corresponding to respective indexes (if user has access permission)
- Evaluate and verify the results

![Screenshot 2024-06-24 at 2 09 09 PM](https://github.com/juspay/hyperswitch/assets/83278309/9a86540a-4e76-44db-981f-175f1520b59e)

<img width="1683" alt="Screenshot 2024-06-25 at 11 26 25 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/4a50c14f-59ca-48e0-ab10-81a5208445ba">

![Screenshot 2024-06-24 at 2 09 42 PM](https://github.com/juspay/hyperswitch/assets/83278309/44ec721a-2c38-4703-b0d2-442ac85121a7)

<img width="1706" alt="Screenshot 2024-06-25 at 11 26 44 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/5123df9f-89bb-4433-85ab-b25988dd317c">


Alternatively, it can also be verified through Postman:
- Hit the /search and /search/{domain} API's to retrieve the data
<img width="1243" alt="Screenshot 2024-06-24 at 1 45 52 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/b00a552d-128a-4fc1-bef2-a449aa6eef0c">
<img width="1234" alt="Screenshot 2024-06-24 at 2 16 30 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/ef741c5f-d27f-435d-a86e-3c47c980b112">

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

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