feat(router): add organization_id in authentication table and add it in authentication events (#7168)

crates/analytics/docs/clickhouse/scripts/authentications.sql

@@ -35,6 +35,7 @@ CREATE TABLE authentication_queue (
     `ds_trans_id` Nullable(String),
     `directory_server_id` Nullable(String),
     `acquirer_country_code` Nullable(String),
+    `organization_id` String,
     `sign_flag` Int8
 ) ENGINE = Kafka SETTINGS kafka_broker_list = 'kafka0:29092',
 kafka_topic_list = 'hyperswitch-authentication-events',
@@ -80,6 +81,7 @@ CREATE TABLE authentications (
     `ds_trans_id` Nullable(String),
     `directory_server_id` Nullable(String),
     `acquirer_country_code` Nullable(String),
+    `organization_id` String,
     `sign_flag` Int8,
     INDEX authenticationConnectorIndex authentication_connector TYPE bloom_filter GRANULARITY 1,
     INDEX transStatusIndex trans_status TYPE bloom_filter GRANULARITY 1,
@@ -127,6 +129,7 @@ CREATE MATERIALIZED VIEW authentication_mv TO authentications (
     `ds_trans_id` Nullable(String),
     `directory_server_id` Nullable(String),
     `acquirer_country_code` Nullable(String),
+    `organization_id` String,
     `sign_flag` Int8
 ) AS
 SELECT
@@ -167,6 +170,7 @@ SELECT
     ds_trans_id,
     directory_server_id,
     acquirer_country_code,
+    organization_id,
     sign_flag
 FROM
     authentication_queue

crates/diesel_models/src/authentication.rs

@@ -48,6 +48,7 @@ pub struct Authentication {
     pub directory_server_id: Option<String>,
     pub acquirer_country_code: Option<String>,
     pub service_details: Option<serde_json::Value>,
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 impl Authentication {
@@ -96,6 +97,7 @@ pub struct AuthenticationNew {
     pub directory_server_id: Option<String>,
     pub acquirer_country_code: Option<String>,
     pub service_details: Option<serde_json::Value>,
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 #[derive(Debug)]

crates/diesel_models/src/schema.rs

@@ -121,6 +121,8 @@ diesel::table! {
         #[max_length = 64]
         acquirer_country_code -> Nullable<Varchar>,
         service_details -> Nullable<Jsonb>,
+        #[max_length = 32]
+        organization_id -> Varchar,
     }
 }
 

crates/diesel_models/src/schema_v2.rs

@@ -122,6 +122,8 @@ diesel::table! {
         #[max_length = 64]
         acquirer_country_code -> Nullable<Varchar>,
         service_details -> Nullable<Jsonb>,
+        #[max_length = 32]
+        organization_id -> Varchar,
     }
 }
 

crates/router/src/core/authentication.rs

@@ -124,6 +124,7 @@ pub async fn perform_post_authentication(
     }
 }
 
+#[allow(clippy::too_many_arguments)]
 pub async fn perform_pre_authentication(
     state: &SessionState,
     key_store: &domain::MerchantKeyStore,
@@ -132,6 +133,7 @@ pub async fn perform_pre_authentication(
     business_profile: &domain::Profile,
     acquirer_details: Option<types::AcquirerDetails>,
     payment_id: Option<common_utils::id_type::PaymentId>,
+    organization_id: common_utils::id_type::OrganizationId,
 ) -> CustomResult<storage::Authentication, ApiErrorResponse> {
     let (authentication_connector, three_ds_connector_account) =
         utils::get_authentication_connector_data(state, key_store, business_profile).await?;
@@ -147,6 +149,7 @@ pub async fn perform_pre_authentication(
             .get_mca_id()
             .ok_or(ApiErrorResponse::InternalServerError)
             .attach_printable("Error while finding mca_id from merchant_connector_account")?,
+        organization_id,
     )
     .await?;
 

crates/router/src/core/authentication/utils.rs

@@ -174,6 +174,7 @@ impl ForeignFrom<common_enums::AuthenticationStatus> for common_enums::AttemptSt
     }
 }
 
+#[allow(clippy::too_many_arguments)]
 pub async fn create_new_authentication(
     state: &SessionState,
     merchant_id: common_utils::id_type::MerchantId,
@@ -182,6 +183,7 @@ pub async fn create_new_authentication(
     profile_id: common_utils::id_type::ProfileId,
     payment_id: Option<common_utils::id_type::PaymentId>,
     merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
+    organization_id: common_utils::id_type::OrganizationId,
 ) -> RouterResult<storage::Authentication> {
     let authentication_id =
         common_utils::generate_id_with_default_len(consts::AUTHENTICATION_ID_PREFIX);
@@ -220,6 +222,7 @@ pub async fn create_new_authentication(
         directory_server_id: None,
         acquirer_country_code: None,
         service_details: None,
+        organization_id,
     };
     state
         .store

crates/router/src/core/payments/operations/payment_confirm.rs

@@ -983,6 +983,7 @@ impl<F: Clone + Send + Sync> Domain<F, api::PaymentsRequest, PaymentData<F>> for
                     business_profile,
                     Some(acquirer_details),
                     Some(payment_data.payment_attempt.payment_id.clone()),
+                    payment_data.payment_attempt.organization_id.clone(),
                 )
                 .await?;
                 if authentication.is_separate_authn_required()
@@ -1175,6 +1176,7 @@ impl<F: Clone + Send + Sync> Domain<F, api::PaymentsRequest, PaymentData<F>> for
                     &authentication_id,
                     payment_data.service_details.clone(),
                     authentication_status,
+                    payment_data.payment_attempt.organization_id.clone(),
                 )
                 .await?;
             },
@@ -1197,6 +1199,7 @@ impl<F: Clone + Send + Sync> Domain<F, api::PaymentsRequest, PaymentData<F>> for
                         .get_mca_id()
                         .ok_or(errors::ApiErrorResponse::InternalServerError)
                         .attach_printable("Error while finding mca_id from merchant_connector_account")?,
+                    payment_data.payment_attempt.organization_id.clone(),
                 )
                 .await?;
 

crates/router/src/core/unified_authentication_service.rs

@@ -410,6 +410,7 @@ pub async fn create_new_authentication(
     authentication_id: &str,
     service_details: Option<payments::CtpServiceDetails>,
     authentication_status: common_enums::AuthenticationStatus,
+    organization_id: common_utils::id_type::OrganizationId,
 ) -> RouterResult<Authentication> {
     let service_details_value = service_details
         .map(serde_json::to_value)
@@ -453,6 +454,7 @@ pub async fn create_new_authentication(
         directory_server_id: None,
         acquirer_country_code: None,
         service_details: service_details_value,
+        organization_id,
     };
     state
         .store

crates/router/src/db/authentication.rs

@@ -151,6 +151,7 @@ impl AuthenticationInterface for MockDb {
             directory_server_id: authentication.directory_server_id,
             acquirer_country_code: authentication.acquirer_country_code,
             service_details: authentication.service_details,
+            organization_id: authentication.organization_id,
         };
         authentications.push(authentication.clone());
         Ok(authentication)

crates/router/src/services/kafka/authentication.rs

@@ -41,6 +41,7 @@ pub struct KafkaAuthentication<'a> {
     pub ds_trans_id: Option<&'a String>,
     pub directory_server_id: Option<&'a String>,
     pub acquirer_country_code: Option<&'a String>,
+    pub organization_id: &'a common_utils::id_type::OrganizationId,
 }
 
 impl<'a> KafkaAuthentication<'a> {
@@ -82,6 +83,7 @@ impl<'a> KafkaAuthentication<'a> {
             ds_trans_id: authentication.ds_trans_id.as_ref(),
             directory_server_id: authentication.directory_server_id.as_ref(),
             acquirer_country_code: authentication.acquirer_country_code.as_ref(),
+            organization_id: &authentication.organization_id,
         }
     }
 }

crates/router/src/services/kafka/authentication_event.rs

@@ -42,6 +42,7 @@ pub struct KafkaAuthenticationEvent<'a> {
     pub ds_trans_id: Option<&'a String>,
     pub directory_server_id: Option<&'a String>,
     pub acquirer_country_code: Option<&'a String>,
+    pub organization_id: &'a common_utils::id_type::OrganizationId,
 }
 
 impl<'a> KafkaAuthenticationEvent<'a> {
@@ -83,6 +84,7 @@ impl<'a> KafkaAuthenticationEvent<'a> {
             ds_trans_id: authentication.ds_trans_id.as_ref(),
             directory_server_id: authentication.directory_server_id.as_ref(),
             acquirer_country_code: authentication.acquirer_country_code.as_ref(),
+            organization_id: &authentication.organization_id,
         }
     }
 }

migrations/2025-01-30-111507_add_organization_id_in_authentication/down.sql

@@ -0,0 +1,3 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE authentication
+    DROP COLUMN IF EXISTS organization_id;

migrations/2025-01-30-111507_add_organization_id_in_authentication/up.sql

@@ -0,0 +1,3 @@
+-- Your SQL goes here
+ALTER TABLE authentication
+    ADD COLUMN IF NOT EXISTS organization_id VARCHAR(32) NOT NULL DEFAULT 'default_org';