Bug: add retrieve api support for relay



/relay is a api that will be used to just forward the request sent by merchant without performing any application logic on it. This is particular to support the refunds for the payments that was directly performed with the connector (example stripe) with out hyperswitch involved during the payment flow. 

This PR introduces changes to add a retrieve API for relay refunds.

The retrieve API will:

1. Return the relay status from our database when a retrieve call is made.
2. If force_sync = true is passed and the relay status is not in a terminal state in our system, the API will call the connector to fetch the latest status.

Flow
1. Merchant performs payment directly with processor and stores the payment reference id associated with it.
2. Merchant tiggers the refund through hyperswitch using the processor reference id for the payment.
3. Merchant makes retrieve call for the relay status
diff --git a/crates/api_models/src/relay.rs b/crates/api_models/src/relay.rs
index e4bc607fc0d..f54e1471632 100644
--- a/crates/api_models/src/relay.rs
+++ b/crates/api_models/src/relay.rs
@@ -84,9 +84,20 @@ pub struct RelayRetrieveRequest {
     #[serde(default)]
     pub force_sync: bool,
     /// The unique identifier for the Relay
-    pub id: String,
+    pub id: common_utils::id_type::RelayId,
+}
+
+#[derive(Debug, ToSchema, Clone, Deserialize, Serialize)]
+pub struct RelayRetrieveBody {
+    /// The unique identifier for the Relay
+    #[serde(default)]
+    pub force_sync: bool,
 }
 
 impl common_utils::events::ApiEventMetric for RelayRequest {}
 
 impl common_utils::events::ApiEventMetric for RelayResponse {}
+
+impl common_utils::events::ApiEventMetric for RelayRetrieveRequest {}
+
+impl common_utils::events::ApiEventMetric for RelayRetrieveBody {}
diff --git a/crates/diesel_models/src/query/relay.rs b/crates/diesel_models/src/query/relay.rs
index 28ede3cd61a..034446fe6b5 100644
--- a/crates/diesel_models/src/query/relay.rs
+++ b/crates/diesel_models/src/query/relay.rs
@@ -35,4 +35,15 @@ impl Relay {
             result => result,
         }
     }
+
+    pub async fn find_by_id(
+        conn: &PgPooledConn,
+        id: &common_utils::id_type::RelayId,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::id.eq(id.to_owned()),
+        )
+        .await
+    }
 }
diff --git a/crates/diesel_models/src/relay.rs b/crates/diesel_models/src/relay.rs
index 24a6ab8b034..153e06ab17f 100644
--- a/crates/diesel_models/src/relay.rs
+++ b/crates/diesel_models/src/relay.rs
@@ -67,7 +67,7 @@ pub struct RelayNew {
 }
 
 #[derive(Clone, Debug, AsChangeset, router_derive::DebugAsDisplay)]
-#[table_name = "relay"]
+#[diesel(table_name = relay)]
 pub struct RelayUpdateInternal {
     pub connector_reference_id: Option<String>,
     pub status: Option<storage_enums::RelayStatus>,
diff --git a/crates/router/src/core/relay.rs b/crates/router/src/core/relay.rs
index 812709269d8..d25296635ad 100644
--- a/crates/router/src/core/relay.rs
+++ b/crates/router/src/core/relay.rs
@@ -1,5 +1,6 @@
 use api_models::relay as relay_models;
-use common_utils::{self, ext_traits::OptionExt, id_type};
+use common_enums::RelayStatus;
+use common_utils::{self, id_type};
 use error_stack::ResultExt;
 
 use super::errors::{self, ConnectorErrorExt, RouterResponse, RouterResult, StorageErrorExt};
@@ -11,6 +12,7 @@ use crate::{
         api::{self},
         domain,
     },
+    utils::OptionExt,
 };
 
 pub mod utils;
@@ -27,11 +29,7 @@ pub async fn relay(
     let merchant_id = merchant_account.get_id();
     let connector_id = &req.connector_id;
 
-    let profile_id_from_auth_layer = profile_id_optional
-        .get_required_value("ProfileId")
-        .change_context(errors::ApiErrorResponse::MissingRequiredField {
-            field_name: "profile id",
-        })?;
+    let profile_id_from_auth_layer = profile_id_optional.get_required_value("ProfileId")?;
 
     let profile = db
         .find_business_profile_by_merchant_id_profile_id(
@@ -175,3 +173,158 @@ pub fn validate_relay_refund_data(
     }
     Ok(())
 }
+
+pub async fn relay_retrieve(
+    state: SessionState,
+    merchant_account: domain::MerchantAccount,
+    profile_id_optional: Option<id_type::ProfileId>,
+    key_store: domain::MerchantKeyStore,
+    req: relay_models::RelayRetrieveRequest,
+) -> RouterResponse<relay_models::RelayResponse> {
+    let db = state.store.as_ref();
+    let key_manager_state = &(&state).into();
+    let merchant_id = merchant_account.get_id();
+    let relay_id = &req.id;
+
+    let profile_id_from_auth_layer = profile_id_optional.get_required_value("ProfileId")?;
+
+    db.find_business_profile_by_merchant_id_profile_id(
+        key_manager_state,
+        &key_store,
+        merchant_id,
+        &profile_id_from_auth_layer,
+    )
+    .await
+    .change_context(errors::ApiErrorResponse::ProfileNotFound {
+        id: profile_id_from_auth_layer.get_string_repr().to_owned(),
+    })?;
+
+    let relay_record_result = db
+        .find_relay_by_id(key_manager_state, &key_store, relay_id)
+        .await;
+
+    let relay_record = match relay_record_result {
+        Err(error) => {
+            if error.current_context().is_db_not_found() {
+                Err(error).change_context(errors::ApiErrorResponse::GenericNotFoundError {
+                    message: "relay not found".to_string(),
+                })?
+            } else {
+                Err(error)
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("error while fetch relay record")?
+            }
+        }
+        Ok(relay) => relay,
+    };
+
+    #[cfg(feature = "v1")]
+    let connector_account = db
+        .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+            key_manager_state,
+            merchant_id,
+            &relay_record.connector_id,
+            &key_store,
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+            id: relay_record.connector_id.get_string_repr().to_string(),
+        })?;
+
+    #[cfg(feature = "v2")]
+    let connector_account = db
+        .find_merchant_connector_account_by_id(
+            key_manager_state,
+            &relay_record.connector_id,
+            &key_store,
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+            id: relay_record.connector_id.get_string_repr().to_string(),
+        })?;
+
+    let relay_response = match relay_record.relay_type {
+        common_enums::RelayType::Refund => {
+            if should_call_connector_for_relay_refund_status(&relay_record, req.force_sync) {
+                let relay_response = sync_relay_refund_with_gateway(
+                    &state,
+                    &merchant_account,
+                    &relay_record,
+                    connector_account,
+                )
+                .await?;
+
+                db.update_relay(key_manager_state, &key_store, relay_record, relay_response)
+                    .await
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("Failed to update the relay record")?
+            } else {
+                relay_record
+            }
+        }
+    };
+
+    let response = relay_models::RelayResponse::from(relay_response);
+
+    Ok(hyperswitch_domain_models::api::ApplicationResponse::Json(
+        response,
+    ))
+}
+
+fn should_call_connector_for_relay_refund_status(
+    relay: &hyperswitch_domain_models::relay::Relay,
+    force_sync: bool,
+) -> bool {
+    // This allows refund sync at connector level if force_sync is enabled, or
+    // check if the refund is in terminal state
+    !matches!(relay.status, RelayStatus::Failure | RelayStatus::Success) && force_sync
+}
+
+pub async fn sync_relay_refund_with_gateway(
+    state: &SessionState,
+    merchant_account: &domain::MerchantAccount,
+    relay_record: &hyperswitch_domain_models::relay::Relay,
+    connector_account: domain::MerchantConnectorAccount,
+) -> RouterResult<hyperswitch_domain_models::relay::RelayUpdate> {
+    let connector_id = &relay_record.connector_id;
+    let merchant_id = merchant_account.get_id();
+
+    let connector_data: api::ConnectorData = api::ConnectorData::get_connector_by_name(
+        &state.conf.connectors,
+        &connector_account.connector_name,
+        api::GetToken::Connector,
+        Some(connector_id.clone()),
+    )
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed to get the connector")?;
+
+    let router_data = utils::construct_relay_refund_router_data(
+        state,
+        &connector_account.connector_name,
+        merchant_id,
+        &connector_account,
+        relay_record,
+    )
+    .await?;
+
+    let connector_integration: services::BoxedRefundConnectorIntegrationInterface<
+        api::RSync,
+        hyperswitch_domain_models::router_request_types::RefundsData,
+        hyperswitch_domain_models::router_response_types::RefundsResponseData,
+    > = connector_data.connector.get_connector_integration();
+
+    let router_data_res = services::execute_connector_processing_step(
+        state,
+        connector_integration,
+        &router_data,
+        payments::CallConnectorAction::Trigger,
+        None,
+    )
+    .await
+    .to_refund_failed_response()?;
+
+    let relay_response =
+        hyperswitch_domain_models::relay::RelayUpdate::from(router_data_res.response);
+
+    Ok(relay_response)
+}
diff --git a/crates/router/src/db/relay.rs b/crates/router/src/db/relay.rs
index e869165aa33..46259679c55 100644
--- a/crates/router/src/db/relay.rs
+++ b/crates/router/src/db/relay.rs
@@ -28,6 +28,13 @@ pub trait RelayInterface {
         current_state: hyperswitch_domain_models::relay::Relay,
         relay_update: hyperswitch_domain_models::relay::RelayUpdate,
     ) -> CustomResult<hyperswitch_domain_models::relay::Relay, errors::StorageError>;
+
+    async fn find_relay_by_id(
+        &self,
+        key_manager_state: &KeyManagerState,
+        merchant_key_store: &domain::MerchantKeyStore,
+        relay_id: &common_utils::id_type::RelayId,
+    ) -> CustomResult<hyperswitch_domain_models::relay::Relay, errors::StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -79,6 +86,25 @@ impl RelayInterface for Store {
             .await
             .change_context(errors::StorageError::DecryptionError)
     }
+
+    async fn find_relay_by_id(
+        &self,
+        key_manager_state: &KeyManagerState,
+        merchant_key_store: &domain::MerchantKeyStore,
+        relay_id: &common_utils::id_type::RelayId,
+    ) -> CustomResult<hyperswitch_domain_models::relay::Relay, errors::StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+        diesel_models::relay::Relay::find_by_id(&conn, relay_id)
+            .await
+            .map_err(|error| report!(errors::StorageError::from(error)))?
+            .convert(
+                key_manager_state,
+                merchant_key_store.key.get_inner(),
+                merchant_key_store.merchant_id.clone().into(),
+            )
+            .await
+            .change_context(errors::StorageError::DecryptionError)
+    }
 }
 
 #[async_trait::async_trait]
@@ -101,6 +127,15 @@ impl RelayInterface for MockDb {
     ) -> CustomResult<hyperswitch_domain_models::relay::Relay, errors::StorageError> {
         Err(errors::StorageError::MockDbError)?
     }
+
+    async fn find_relay_by_id(
+        &self,
+        _key_manager_state: &KeyManagerState,
+        _merchant_key_store: &domain::MerchantKeyStore,
+        _relay_id: &common_utils::id_type::RelayId,
+    ) -> CustomResult<hyperswitch_domain_models::relay::Relay, errors::StorageError> {
+        Err(errors::StorageError::MockDbError)?
+    }
 }
 
 #[async_trait::async_trait]
@@ -132,4 +167,15 @@ impl RelayInterface for KafkaStore {
             )
             .await
     }
+
+    async fn find_relay_by_id(
+        &self,
+        key_manager_state: &KeyManagerState,
+        merchant_key_store: &domain::MerchantKeyStore,
+        relay_id: &common_utils::id_type::RelayId,
+    ) -> CustomResult<hyperswitch_domain_models::relay::Relay, errors::StorageError> {
+        self.diesel_store
+            .find_relay_by_id(key_manager_state, merchant_key_store, relay_id)
+            .await
+    }
 }
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 2decddb806c..21ea9446138 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -598,6 +598,7 @@ impl Relay {
         web::scope("/relay")
             .app_data(web::Data::new(state))
             .service(web::resource("").route(web::post().to(relay::relay)))
+            .service(web::resource("/{relay_id}").route(web::get().to(relay::relay_retrieve)))
     }
 }
 
diff --git a/crates/router/src/routes/lock_utils.rs b/crates/router/src/routes/lock_utils.rs
index 4b43c42f3bc..dd0fbcdc08e 100644
--- a/crates/router/src/routes/lock_utils.rs
+++ b/crates/router/src/routes/lock_utils.rs
@@ -165,7 +165,7 @@ impl From<Flow> for ApiIdentifier {
             | Flow::RefundsFilters
             | Flow::RefundsAggregate
             | Flow::RefundsManualUpdate => Self::Refunds,
-            Flow::Relay => Self::Relay,
+            Flow::Relay | Flow::RelayRetrieve => Self::Relay,
 
             Flow::FrmFulfillment
             | Flow::IncomingWebhookReceive
diff --git a/crates/router/src/routes/relay.rs b/crates/router/src/routes/relay.rs
index 13c92eb19a6..cfc66253d50 100644
--- a/crates/router/src/routes/relay.rs
+++ b/crates/router/src/routes/relay.rs
@@ -38,3 +38,39 @@ pub async fn relay(
     ))
     .await
 }
+
+#[instrument(skip_all, fields(flow = ?Flow::RelayRetrieve))]
+#[cfg(feature = "oltp")]
+pub async fn relay_retrieve(
+    state: web::Data<app::AppState>,
+    path: web::Path<common_utils::id_type::RelayId>,
+    req: actix_web::HttpRequest,
+    query_params: web::Query<api_models::relay::RelayRetrieveBody>,
+) -> impl Responder {
+    let flow = Flow::RelayRetrieve;
+    let relay_retrieve_request = api_models::relay::RelayRetrieveRequest {
+        force_sync: query_params.force_sync,
+        id: path.into_inner(),
+    };
+    Box::pin(api::server_wrap(
+        flow,
+        state,
+        &req,
+        relay_retrieve_request,
+        |state, auth: auth::AuthenticationData, req, _| {
+            relay::relay_retrieve(
+                state,
+                auth.merchant_account,
+                #[cfg(feature = "v1")]
+                auth.profile_id,
+                #[cfg(feature = "v2")]
+                Some(auth.profile.get_id().clone()),
+                auth.key_store,
+                req,
+            )
+        },
+        &auth::HeaderAuth(auth::ApiKeyAuth),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
diff --git a/crates/router_env/src/logger/types.rs b/crates/router_env/src/logger/types.rs
index de2577e6c34..661375aadbf 100644
--- a/crates/router_env/src/logger/types.rs
+++ b/crates/router_env/src/logger/types.rs
@@ -531,6 +531,8 @@ pub enum Flow {
     VolumeSplitOnRoutingType,
     /// Relay flow
     Relay,
+    /// Relay retrieve flow
+    RelayRetrieve,
 }
 
 /// Trait for providing generic behaviour to flow metric

## 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 -->
/relay is a api that will be used to just forward the request sent by merchant without performing any application logic on it. This is particular to support the refunds for the payments that was directly performed with the connector (example stripe) with out hyperswitch involved during the payment flow. 

This PR introduces changes to add a retrieve API for relay refunds.

The retrieve API will:

1. Return the relay status from our database when a retrieve call is made.
2. If force_sync = true is passed and the relay status is not in a terminal state in our system, the API will call the connector to fetch the latest status.

Flow
1. Merchant performs payment directly with processor and stores the payment reference id associated with it.
2. Merchant tiggers the refund through hyperswitch using the processor reference id for the payment.
3. Merchant makes retrieve call for the relay status



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


## 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)?
-->
-> Create a merchant connector account
-> Make a refund request with invalid connector_resource_id
```
curl --location 'http://localhost:8080/relay' \
--header 'Content-Type: application/json' \
--header 'X-Profile-Id: pro_rIikb0gPrzd0vEY8hIiP' \
--header 'api-key: dev_qqMQAuMWCclJSk4uMBeNNSpqAPlsKlBA3blQOwOoq5ogTSFVON3taShABqtqYNA4' \
--data '{
  "connector_id": "mca_8W0CQfwE89J1BJBbpcnt",
  "connector_resource_id": "7349526182906968904951",
  "data": {
      "refund": {
    "amount": 50,
    "currency": "USD"
  }},
  "type": "refund"
}'
```
```
{
    "id": "relay_4lgXCQW0EnqKddZ9CC3b",
    "status": "pending",
    "connector_resource_id": "7349526182906968904951",
    "error": null,
    "connector_reference_id": "7349526431306986204951",
    "connector_id": "mca_8W0CQfwE89J1BJBbpcnt",
    "profile_id": "pro_rIikb0gPrzd0vEY8hIiP",
    "type": "refund",
    "data": {
        "refund": {
            "amount": 50,
            "currency": "USD",
            "reason": null
        }
    }
}
```
-> Relay retrieve
```
curl --location 'http://localhost:8080/relay/relay_4lgXCQW0EnqKddZ9CC3b' \
--header 'Content-Type: application/json' \
--header 'X-Idempotency-Key: <x-idempotency-key>' \
--header 'X-Profile-Id: pro_rIikb0gPrzd0vEY8hIiP' \
--header 'api-key: dev_qqMQAuMWCclJSk4uMBeNNSpqAPlsKlBA3blQOwOoq5ogTSFVON3taShABqtqYNA4' \
--data ''
```
```
{
    "id": "relay_4lgXCQW0EnqKddZ9CC3b",
    "status": "pending",
    "connector_resource_id": "7349526182906968904951",
    "error": null,
    "connector_reference_id": "7349526431306986204951",
    "connector_id": "mca_8W0CQfwE89J1BJBbpcnt",
    "profile_id": "pro_rIikb0gPrzd0vEY8hIiP",
    "type": "refund",
    "data": {
        "refund": {
            "amount": 50,
            "currency": "USD",
            "reason": null
        }
    }
}
```

-> Relay retrieve force_sync = `true`
```
curl --location 'http://localhost:8080/relay/relay_4lgXCQW0EnqKddZ9CC3b?force_sync=true' \
--header 'Content-Type: application/json' \
--header 'X-Idempotency-Key: <x-idempotency-key>' \
--header 'X-Profile-Id: pro_rIikb0gPrzd0vEY8hIiP' \
--header 'api-key: dev_qqMQAuMWCclJSk4uMBeNNSpqAPlsKlBA3blQOwOoq5ogTSFVON3taShABqtqYNA4' \
--data ''
```
```
{
    "id": "relay_4lgXCQW0EnqKddZ9CC3b",
    "status": "success",
    "connector_resource_id": "7349526182906968904951",
    "error": null,
    "connector_reference_id": "7349526431306986204951",
    "connector_id": "mca_8W0CQfwE89J1BJBbpcnt",
    "profile_id": "pro_rIikb0gPrzd0vEY8hIiP",
    "type": "refund",
    "data": {
        "refund": {
            "amount": 50,
            "currency": "USD",
            "reason": null
        }
    }
}
```


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