feat(dynamic_routing): analytics improvement using separate postgres table (#6723)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

crates/api_models/src/routing.rs

@@ -740,7 +740,6 @@ pub struct ToggleDynamicRoutingPath {
 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, ToSchema)]
 pub struct EliminationRoutingConfig {
     pub params: Option<Vec<DynamicRoutingConfigParams>>,
-    // pub labels: Option<Vec<String>>,
     pub elimination_analyser_config: Option<EliminationAnalyserConfig>,
 }
 

crates/common_enums/src/enums.rs

@@ -20,7 +20,9 @@ pub mod diesel_exports {
         DbMandateStatus as MandateStatus, DbPaymentMethodIssuerCode as PaymentMethodIssuerCode,
         DbPaymentType as PaymentType, DbRefundStatus as RefundStatus,
         DbRequestIncrementalAuthorization as RequestIncrementalAuthorization,
-        DbScaExemptionType as ScaExemptionType, DbWebhookDeliveryAttempt as WebhookDeliveryAttempt,
+        DbScaExemptionType as ScaExemptionType,
+        DbSuccessBasedRoutingConclusiveState as SuccessBasedRoutingConclusiveState,
+        DbWebhookDeliveryAttempt as WebhookDeliveryAttempt,
     };
 }
 
@@ -3283,10 +3285,20 @@ pub enum DeleteStatus {
 }
 
 #[derive(
-    Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize, strum::Display, Hash,
+    Clone,
+    Copy,
+    Debug,
+    Eq,
+    PartialEq,
+    serde::Deserialize,
+    serde::Serialize,
+    strum::Display,
+    Hash,
+    strum::EnumString,
 )]
 #[serde(rename_all = "snake_case")]
 #[strum(serialize_all = "snake_case")]
+#[router_derive::diesel_enum(storage_type = "db_enum")]
 pub enum SuccessBasedRoutingConclusiveState {
     // pc: payment connector
     // sc: success based routing outcome/first connector

crates/diesel_models/src/dynamic_routing_stats.rs

@@ -0,0 +1,41 @@
+use diesel::{Insertable, Queryable, Selectable};
+
+use crate::schema::dynamic_routing_stats;
+
+#[derive(Clone, Debug, Eq, Insertable, PartialEq)]
+#[diesel(table_name = dynamic_routing_stats)]
+pub struct DynamicRoutingStatsNew {
+    pub payment_id: common_utils::id_type::PaymentId,
+    pub attempt_id: String,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub profile_id: common_utils::id_type::ProfileId,
+    pub amount: common_utils::types::MinorUnit,
+    pub success_based_routing_connector: String,
+    pub payment_connector: String,
+    pub currency: Option<common_enums::Currency>,
+    pub payment_method: Option<common_enums::PaymentMethod>,
+    pub capture_method: Option<common_enums::CaptureMethod>,
+    pub authentication_type: Option<common_enums::AuthenticationType>,
+    pub payment_status: common_enums::AttemptStatus,
+    pub conclusive_classification: common_enums::SuccessBasedRoutingConclusiveState,
+    pub created_at: time::PrimitiveDateTime,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, Queryable, Selectable, Insertable)]
+#[diesel(table_name = dynamic_routing_stats, primary_key(payment_id), check_for_backend(diesel::pg::Pg))]
+pub struct DynamicRoutingStats {
+    pub payment_id: common_utils::id_type::PaymentId,
+    pub attempt_id: String,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub profile_id: common_utils::id_type::ProfileId,
+    pub amount: common_utils::types::MinorUnit,
+    pub success_based_routing_connector: String,
+    pub payment_connector: String,
+    pub currency: Option<common_enums::Currency>,
+    pub payment_method: Option<common_enums::PaymentMethod>,
+    pub capture_method: Option<common_enums::CaptureMethod>,
+    pub authentication_type: Option<common_enums::AuthenticationType>,
+    pub payment_status: common_enums::AttemptStatus,
+    pub conclusive_classification: common_enums::SuccessBasedRoutingConclusiveState,
+    pub created_at: time::PrimitiveDateTime,
+}

crates/diesel_models/src/enums.rs

@@ -20,9 +20,11 @@ pub mod diesel_exports {
         DbRefundStatus as RefundStatus, DbRefundType as RefundType,
         DbRequestIncrementalAuthorization as RequestIncrementalAuthorization,
         DbRoleScope as RoleScope, DbRoutingAlgorithmKind as RoutingAlgorithmKind,
-        DbScaExemptionType as ScaExemptionType, DbTotpStatus as TotpStatus,
-        DbTransactionType as TransactionType, DbUserRoleVersion as UserRoleVersion,
-        DbUserStatus as UserStatus, DbWebhookDeliveryAttempt as WebhookDeliveryAttempt,
+        DbScaExemptionType as ScaExemptionType,
+        DbSuccessBasedRoutingConclusiveState as SuccessBasedRoutingConclusiveState,
+        DbTotpStatus as TotpStatus, DbTransactionType as TransactionType,
+        DbUserRoleVersion as UserRoleVersion, DbUserStatus as UserStatus,
+        DbWebhookDeliveryAttempt as WebhookDeliveryAttempt,
     };
 }
 pub use common_enums::*;

crates/diesel_models/src/lib.rs

@@ -12,6 +12,7 @@ pub mod blocklist;
 pub mod blocklist_fingerprint;
 pub mod customers;
 pub mod dispute;
+pub mod dynamic_routing_stats;
 pub mod enums;
 pub mod ephemeral_key;
 pub mod errors;

crates/diesel_models/src/query.rs

@@ -13,6 +13,7 @@ pub mod blocklist_fingerprint;
 pub mod customers;
 pub mod dashboard_metadata;
 pub mod dispute;
+pub mod dynamic_routing_stats;
 pub mod events;
 pub mod file;
 pub mod fraud_check;

crates/diesel_models/src/query/dynamic_routing_stats.rs

@@ -0,0 +1,11 @@
+use super::generics;
+use crate::{
+    dynamic_routing_stats::{DynamicRoutingStats, DynamicRoutingStatsNew},
+    PgPooledConn, StorageResult,
+};
+
+impl DynamicRoutingStatsNew {
+    pub async fn insert(self, conn: &PgPooledConn) -> StorageResult<DynamicRoutingStats> {
+        generics::generic_insert(conn, self).await
+    }
+}

crates/diesel_models/src/schema.rs

@@ -389,6 +389,35 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    dynamic_routing_stats (attempt_id, merchant_id) {
+        #[max_length = 64]
+        payment_id -> Varchar,
+        #[max_length = 64]
+        attempt_id -> Varchar,
+        #[max_length = 64]
+        merchant_id -> Varchar,
+        #[max_length = 64]
+        profile_id -> Varchar,
+        amount -> Int8,
+        #[max_length = 64]
+        success_based_routing_connector -> Varchar,
+        #[max_length = 64]
+        payment_connector -> Varchar,
+        currency -> Nullable<Currency>,
+        #[max_length = 64]
+        payment_method -> Nullable<Varchar>,
+        capture_method -> Nullable<CaptureMethod>,
+        authentication_type -> Nullable<AuthenticationType>,
+        payment_status -> AttemptStatus,
+        conclusive_classification -> SuccessBasedRoutingConclusiveState,
+        created_at -> Timestamp,
+    }
+}
+
 diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
@@ -1414,6 +1443,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     customers,
     dashboard_metadata,
     dispute,
+    dynamic_routing_stats,
     events,
     file_metadata,
     fraud_check,

crates/diesel_models/src/schema_v2.rs

@@ -401,6 +401,35 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    dynamic_routing_stats (attempt_id, merchant_id) {
+        #[max_length = 64]
+        payment_id -> Varchar,
+        #[max_length = 64]
+        attempt_id -> Varchar,
+        #[max_length = 64]
+        merchant_id -> Varchar,
+        #[max_length = 64]
+        profile_id -> Varchar,
+        amount -> Int8,
+        #[max_length = 64]
+        success_based_routing_connector -> Varchar,
+        #[max_length = 64]
+        payment_connector -> Varchar,
+        currency -> Nullable<Currency>,
+        #[max_length = 64]
+        payment_method -> Nullable<Varchar>,
+        capture_method -> Nullable<CaptureMethod>,
+        authentication_type -> Nullable<AuthenticationType>,
+        payment_status -> AttemptStatus,
+        conclusive_classification -> SuccessBasedRoutingConclusiveState,
+        created_at -> Timestamp,
+    }
+}
+
 diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
@@ -1362,6 +1391,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     customers,
     dashboard_metadata,
     dispute,
+    dynamic_routing_stats,
     events,
     file_metadata,
     fraud_check,

crates/router/src/core/routing/helpers.rs

@@ -12,6 +12,8 @@ use api_models::routing as routing_types;
 use common_utils::ext_traits::ValueExt;
 use common_utils::{ext_traits::Encode, id_type, types::keymanager::KeyManagerState};
 use diesel_models::configs;
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+use diesel_models::dynamic_routing_stats::DynamicRoutingStatsNew;
 #[cfg(feature = "v1")]
 use diesel_models::routing_algorithm;
 use error_stack::ResultExt;
@@ -751,6 +753,23 @@ pub async fn push_metrics_with_update_window_for_success_based_routing(
             first_success_based_connector.to_string(),
         );
 
+        let dynamic_routing_stats = DynamicRoutingStatsNew {
+            payment_id: payment_attempt.payment_id.to_owned(),
+            attempt_id: payment_attempt.attempt_id.clone(),
+            merchant_id: payment_attempt.merchant_id.to_owned(),
+            profile_id: payment_attempt.profile_id.to_owned(),
+            amount: payment_attempt.get_total_amount(),
+            success_based_routing_connector: first_success_based_connector.to_string(),
+            payment_connector: payment_connector.to_string(),
+            currency: payment_attempt.currency,
+            payment_method: payment_attempt.payment_method,
+            capture_method: payment_attempt.capture_method,
+            authentication_type: payment_attempt.authentication_type,
+            payment_status: payment_attempt.status,
+            conclusive_classification: outcome,
+            created_at: common_utils::date_time::now(),
+        };
+
         core_metrics::DYNAMIC_SUCCESS_BASED_ROUTING.add(
             &metrics::CONTEXT,
             1,
@@ -812,6 +831,13 @@ pub async fn push_metrics_with_update_window_for_success_based_routing(
         );
         logger::debug!("successfully pushed success_based_routing metrics");
 
+        state
+            .store
+            .insert_dynamic_routing_stat_entry(dynamic_routing_stats)
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to push dynamic routing stats to db")?;
+
         client
             .update_success_rate(
                 tenant_business_profile_id,

crates/router/src/db.rs

@@ -12,6 +12,7 @@ pub mod configs;
 pub mod customers;
 pub mod dashboard_metadata;
 pub mod dispute;
+pub mod dynamic_routing_stats;
 pub mod ephemeral_key;
 pub mod events;
 pub mod file;
@@ -107,6 +108,7 @@ pub trait StorageInterface:
     + payment_method::PaymentMethodInterface
     + blocklist::BlocklistInterface
     + blocklist_fingerprint::BlocklistFingerprintInterface
+    + dynamic_routing_stats::DynamicRoutingStatsInterface
     + scheduler::SchedulerInterface
     + PayoutAttemptInterface
     + PayoutsInterface

crates/router/src/db/dynamic_routing_stats.rs

@@ -0,0 +1,58 @@
+use error_stack::report;
+use router_env::{instrument, tracing};
+use storage_impl::MockDb;
+
+use super::Store;
+use crate::{
+    connection,
+    core::errors::{self, CustomResult},
+    db::kafka_store::KafkaStore,
+    types::storage,
+};
+
+#[async_trait::async_trait]
+pub trait DynamicRoutingStatsInterface {
+    async fn insert_dynamic_routing_stat_entry(
+        &self,
+        dynamic_routing_stat_new: storage::DynamicRoutingStatsNew,
+    ) -> CustomResult<storage::DynamicRoutingStats, errors::StorageError>;
+}
+
+#[async_trait::async_trait]
+impl DynamicRoutingStatsInterface for Store {
+    #[instrument(skip_all)]
+    async fn insert_dynamic_routing_stat_entry(
+        &self,
+        dynamic_routing_stat: storage::DynamicRoutingStatsNew,
+    ) -> CustomResult<storage::DynamicRoutingStats, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        dynamic_routing_stat
+            .insert(&conn)
+            .await
+            .map_err(|error| report!(errors::StorageError::from(error)))
+    }
+}
+
+#[async_trait::async_trait]
+impl DynamicRoutingStatsInterface for MockDb {
+    #[instrument(skip_all)]
+    async fn insert_dynamic_routing_stat_entry(
+        &self,
+        _dynamic_routing_stat: storage::DynamicRoutingStatsNew,
+    ) -> CustomResult<storage::DynamicRoutingStats, errors::StorageError> {
+        Err(errors::StorageError::MockDbError)?
+    }
+}
+
+#[async_trait::async_trait]
+impl DynamicRoutingStatsInterface for KafkaStore {
+    #[instrument(skip_all)]
+    async fn insert_dynamic_routing_stat_entry(
+        &self,
+        dynamic_routing_stat: storage::DynamicRoutingStatsNew,
+    ) -> CustomResult<storage::DynamicRoutingStats, errors::StorageError> {
+        self.diesel_store
+            .insert_dynamic_routing_stat_entry(dynamic_routing_stat)
+            .await
+    }
+}

crates/router/src/types/storage.rs

@@ -12,6 +12,7 @@ pub mod configs;
 pub mod customers;
 pub mod dashboard_metadata;
 pub mod dispute;
+pub mod dynamic_routing_stats;
 pub mod enums;
 pub mod ephemeral_key;
 pub mod events;
@@ -63,12 +64,12 @@ pub use scheduler::db::process_tracker;
 pub use self::{
     address::*, api_keys::*, authentication::*, authorization::*, blocklist::*,
     blocklist_fingerprint::*, blocklist_lookup::*, business_profile::*, capture::*, cards_info::*,
-    configs::*, customers::*, dashboard_metadata::*, dispute::*, ephemeral_key::*, events::*,
-    file::*, fraud_check::*, generic_link::*, gsm::*, locker_mock_up::*, mandate::*,
-    merchant_account::*, merchant_connector_account::*, merchant_key_store::*, payment_link::*,
-    payment_method::*, process_tracker::*, refund::*, reverse_lookup::*, role::*,
-    routing_algorithm::*, unified_translations::*, user::*, user_authentication_method::*,
-    user_role::*,
+    configs::*, customers::*, dashboard_metadata::*, dispute::*, dynamic_routing_stats::*,
+    ephemeral_key::*, events::*, file::*, fraud_check::*, generic_link::*, gsm::*,
+    locker_mock_up::*, mandate::*, merchant_account::*, merchant_connector_account::*,
+    merchant_key_store::*, payment_link::*, payment_method::*, process_tracker::*, refund::*,
+    reverse_lookup::*, role::*, routing_algorithm::*, unified_translations::*, user::*,
+    user_authentication_method::*, user_role::*,
 };
 use crate::types::api::routing;
 

crates/router/src/types/storage/dynamic_routing_stats.rs

@@ -0,0 +1 @@
+pub use diesel_models::dynamic_routing_stats::{DynamicRoutingStats, DynamicRoutingStatsNew};

migrations/2024-12-02-095127_add_new_table_dynamic_routing_stats/down.sql

@@ -0,0 +1,3 @@
+-- This file should undo anything in `up.sql`
+DROP TABLE IF EXISTS dynamic_routing_stats;
+DROP TYPE IF EXISTS "SuccessBasedRoutingConclusiveState";

migrations/2024-12-02-095127_add_new_table_dynamic_routing_stats/up.sql

@@ -0,0 +1,26 @@
+--- Your SQL goes here
+CREATE TYPE "SuccessBasedRoutingConclusiveState" AS ENUM(
+  'true_positive',
+  'false_positive',
+  'true_negative',
+  'false_negative'
+);
+
+CREATE TABLE IF NOT EXISTS dynamic_routing_stats (
+    payment_id VARCHAR(64) NOT NULL,
+    attempt_id VARCHAR(64) NOT NULL,
+    merchant_id VARCHAR(64) NOT NULL,
+    profile_id VARCHAR(64) NOT NULL,
+    amount BIGINT NOT NULL,
+    success_based_routing_connector VARCHAR(64) NOT NULL,
+    payment_connector VARCHAR(64) NOT NULL,
+    currency "Currency",
+    payment_method VARCHAR(64),
+    capture_method "CaptureMethod",
+    authentication_type "AuthenticationType",
+    payment_status "AttemptStatus" NOT NULL,
+    conclusive_classification "SuccessBasedRoutingConclusiveState" NOT NULL,
+    created_at TIMESTAMP NOT NULL,
+    PRIMARY KEY(attempt_id, merchant_id)
+);
+CREATE INDEX profile_id_index ON dynamic_routing_stats (profile_id);