Bug: Add support to create subscription with plans having trial period




diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index 5d8c7a3f0bd..a2cd9d19041 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -5025,7 +5025,7 @@ pub enum NextActionType {
     RedirectInsidePopup,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum NextActionData {
     /// Contains the url for redirection flow
@@ -5097,7 +5097,7 @@ pub enum NextActionData {
     },
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(tag = "method_key")]
 pub enum IframeData {
     #[serde(rename = "threeDSMethodData")]
@@ -5115,7 +5115,7 @@ pub enum IframeData {
     },
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 pub struct ThreeDsData {
     /// ThreeDS authentication url - to initiate authentication
     pub three_ds_authentication_url: String,
@@ -5131,7 +5131,7 @@ pub struct ThreeDsData {
     pub directory_server_id: Option<String>,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(untagged)]
 pub enum ThreeDsMethodData {
     AcsThreeDsMethodData {
@@ -5148,7 +5148,7 @@ pub enum ThreeDsMethodData {
     },
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 pub enum ThreeDsMethodKey {
     #[serde(rename = "threeDSMethodData")]
     ThreeDsMethodData,
@@ -5156,7 +5156,7 @@ pub enum ThreeDsMethodKey {
     JWT,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 pub struct PollConfigResponse {
     /// Poll Id
     pub poll_id: String,
@@ -7761,7 +7761,7 @@ pub struct GooglePayTokenizationParameters {
     pub stripe_version: Option<Secret<String>>,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(tag = "wallet_name")]
 #[serde(rename_all = "snake_case")]
 pub enum SessionToken {
@@ -7819,7 +7819,7 @@ pub struct HyperswitchVaultSessionDetails {
     pub profile_id: Secret<String>,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct PazeSessionTokenResponse {
     /// Paze Client ID
@@ -7839,7 +7839,7 @@ pub struct PazeSessionTokenResponse {
     pub email_address: Option<Email>,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(untagged)]
 pub enum GpaySessionTokenResponse {
     /// Google pay response involving third party sdk
@@ -7848,7 +7848,7 @@ pub enum GpaySessionTokenResponse {
     GooglePaySession(GooglePaySessionResponse),
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct GooglePayThirdPartySdk {
     /// Identifier for the delayed session response
@@ -7859,7 +7859,7 @@ pub struct GooglePayThirdPartySdk {
     pub sdk_next_action: SdkNextAction,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct GooglePaySessionResponse {
     /// The merchant info
@@ -7884,7 +7884,7 @@ pub struct GooglePaySessionResponse {
     pub secrets: Option<SecretInfoToInitiateSdk>,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct SamsungPaySessionTokenResponse {
     /// Samsung Pay API version
@@ -7908,13 +7908,13 @@ pub struct SamsungPaySessionTokenResponse {
     pub shipping_address_required: bool,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub enum SamsungPayProtocolType {
     Protocol3ds,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct SamsungPayMerchantPaymentInformation {
     /// Merchant name, this will be displayed on the Samsung Pay screen
@@ -7926,7 +7926,7 @@ pub struct SamsungPayMerchantPaymentInformation {
     pub country_code: api_enums::CountryAlpha2,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct SamsungPayAmountDetails {
     #[serde(rename = "option")]
@@ -7941,7 +7941,7 @@ pub struct SamsungPayAmountDetails {
     pub total_amount: StringMajorUnit,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub enum SamsungPayAmountFormat {
     /// Display the total amount only
@@ -7950,14 +7950,14 @@ pub enum SamsungPayAmountFormat {
     FormatTotalEstimatedAmount,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct GpayShippingAddressParameters {
     /// Is shipping phone number required
     pub phone_number_required: bool,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct KlarnaSessionTokenResponse {
     /// The session token for Klarna
@@ -7985,7 +7985,7 @@ pub struct PaypalTransactionInfo {
     pub total_price: StringMajorUnit,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct PaypalSessionTokenResponse {
     /// Name of the connector
@@ -8000,14 +8000,14 @@ pub struct PaypalSessionTokenResponse {
     pub transaction_info: Option<PaypalTransactionInfo>,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct OpenBankingSessionToken {
     /// The session token for OpenBanking Connectors
     pub open_banking_session_token: String,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub struct ApplepaySessionTokenResponse {
     /// Session object for Apple Pay
@@ -8030,7 +8030,7 @@ pub struct ApplepaySessionTokenResponse {
     pub connector_merchant_id: Option<String>,
 }
 
-#[derive(Debug, Eq, PartialEq, serde::Serialize, Clone, ToSchema)]
+#[derive(Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize, Clone, ToSchema)]
 pub struct SdkNextAction {
     /// The type of next action
     pub next_action: NextActionCall,
@@ -8051,7 +8051,7 @@ pub enum NextActionCall {
     AwaitMerchantCallback,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 #[serde(untagged)]
 pub enum ApplePaySessionResponse {
     ///  We get this session response, when third party sdk is involved
@@ -8090,7 +8090,7 @@ pub struct NoThirdPartySdkSessionResponse {
     pub psp_id: String,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 pub struct ThirdPartySdkSessionResponse {
     pub secrets: SecretInfoToInitiateSdk,
 }
@@ -8211,7 +8211,7 @@ pub struct ApplepayErrorResponse {
     pub status_message: String,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 pub struct AmazonPaySessionTokenResponse {
     /// Amazon Pay merchant account identifier
     pub merchant_id: String,
@@ -8235,7 +8235,7 @@ pub struct AmazonPaySessionTokenResponse {
     pub delivery_options: Vec<AmazonPayDeliveryOptions>,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 pub enum AmazonPayPaymentIntent {
     /// Create a Charge Permission to authorize and capture funds at a later time
     Confirm,
@@ -9291,7 +9291,7 @@ pub struct ExtendedCardInfoResponse {
     pub payload: String,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
 pub struct ClickToPaySessionResponse {
     pub dpa_id: String,
     pub dpa_name: String,
@@ -9951,7 +9951,7 @@ pub struct RecordAttemptErrorDetails {
     pub network_error_message: Option<String>,
 }
 
-#[derive(Debug, Clone, Eq, PartialEq, ToSchema)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Deserialize, ToSchema)]
 pub struct NullObject;
 
 impl Serialize for NullObject {
diff --git a/crates/api_models/src/subscription.rs b/crates/api_models/src/subscription.rs
index 45e32a31c66..c7aea80d4f2 100644
--- a/crates/api_models/src/subscription.rs
+++ b/crates/api_models/src/subscription.rs
@@ -6,7 +6,7 @@ use utoipa::ToSchema;
 use crate::{
     enums as api_enums,
     mandates::RecurringDetails,
-    payments::{Address, PaymentMethodDataRequest},
+    payments::{Address, NextActionData, PaymentMethodDataRequest},
 };
 
 /// Request payload for creating a subscription.
@@ -216,6 +216,7 @@ pub struct ConfirmSubscriptionPaymentDetails {
     pub payment_method_type: Option<api_enums::PaymentMethodType>,
     pub payment_method_data: PaymentMethodDataRequest,
     pub customer_acceptance: Option<CustomerAcceptance>,
+    pub payment_type: Option<api_enums::PaymentType>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
@@ -224,6 +225,7 @@ pub struct CreateSubscriptionPaymentDetails {
     pub setup_future_usage: Option<api_enums::FutureUsage>,
     pub capture_method: Option<api_enums::CaptureMethod>,
     pub authentication_type: Option<api_enums::AuthenticationType>,
+    pub payment_type: Option<api_enums::PaymentType>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
@@ -236,6 +238,7 @@ pub struct PaymentDetails {
     pub return_url: Option<common_utils::types::Url>,
     pub capture_method: Option<api_enums::CaptureMethod>,
     pub authentication_type: Option<api_enums::AuthenticationType>,
+    pub payment_type: Option<api_enums::PaymentType>,
 }
 
 // Creating new type for PaymentRequest API call as usage of api_models::PaymentsRequest will result in invalid payment request during serialization
@@ -247,6 +250,7 @@ pub struct CreatePaymentsRequestData {
     pub customer_id: Option<common_utils::id_type::CustomerId>,
     pub billing: Option<Address>,
     pub shipping: Option<Address>,
+    pub profile_id: Option<common_utils::id_type::ProfileId>,
     pub setup_future_usage: Option<api_enums::FutureUsage>,
     pub return_url: Option<common_utils::types::Url>,
     pub capture_method: Option<api_enums::CaptureMethod>,
@@ -257,10 +261,12 @@ pub struct CreatePaymentsRequestData {
 pub struct ConfirmPaymentsRequestData {
     pub billing: Option<Address>,
     pub shipping: Option<Address>,
+    pub profile_id: Option<common_utils::id_type::ProfileId>,
     pub payment_method: api_enums::PaymentMethod,
     pub payment_method_type: Option<api_enums::PaymentMethodType>,
     pub payment_method_data: PaymentMethodDataRequest,
     pub customer_acceptance: Option<CustomerAcceptance>,
+    pub payment_type: Option<api_enums::PaymentType>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, ToSchema)]
@@ -271,6 +277,7 @@ pub struct CreateAndConfirmPaymentsRequestData {
     pub confirm: bool,
     pub billing: Option<Address>,
     pub shipping: Option<Address>,
+    pub profile_id: Option<common_utils::id_type::ProfileId>,
     pub setup_future_usage: Option<api_enums::FutureUsage>,
     pub return_url: Option<common_utils::types::Url>,
     pub capture_method: Option<api_enums::CaptureMethod>,
@@ -279,6 +286,7 @@ pub struct CreateAndConfirmPaymentsRequestData {
     pub payment_method_type: Option<api_enums::PaymentMethodType>,
     pub payment_method_data: Option<PaymentMethodDataRequest>,
     pub customer_acceptance: Option<CustomerAcceptance>,
+    pub payment_type: Option<api_enums::PaymentType>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
@@ -287,13 +295,17 @@ pub struct PaymentResponseData {
     pub status: api_enums::IntentStatus,
     pub amount: MinorUnit,
     pub currency: api_enums::Currency,
+    pub profile_id: Option<common_utils::id_type::ProfileId>,
     pub connector: Option<String>,
     pub payment_method_id: Option<Secret<String>>,
+    pub return_url: Option<common_utils::types::Url>,
+    pub next_action: Option<NextActionData>,
     pub payment_experience: Option<api_enums::PaymentExperience>,
     pub error_code: Option<String>,
     pub error_message: Option<String>,
     pub payment_method_type: Option<api_enums::PaymentMethodType>,
     pub client_secret: Option<Secret<String>>,
+    pub payment_type: Option<api_enums::PaymentType>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
@@ -304,6 +316,7 @@ pub struct CreateMitPaymentRequestData {
     pub customer_id: Option<common_utils::id_type::CustomerId>,
     pub recurring_details: Option<RecurringDetails>,
     pub off_session: Option<bool>,
+    pub profile_id: Option<common_utils::id_type::ProfileId>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
@@ -452,7 +465,7 @@ pub struct Invoice {
     pub currency: api_enums::Currency,
 
     /// Status of the invoice.
-    pub status: String,
+    pub status: common_enums::connector_enums::InvoiceStatus,
 }
 
 impl ApiEventMetric for ConfirmSubscriptionResponse {}
diff --git a/crates/common_enums/src/connector_enums.rs b/crates/common_enums/src/connector_enums.rs
index fe11c685aec..6b42837bc27 100644
--- a/crates/common_enums/src/connector_enums.rs
+++ b/crates/common_enums/src/connector_enums.rs
@@ -900,7 +900,19 @@ impl TryFrom<Connector> for RoutableConnectors {
 }
 
 // Enum representing different status an invoice can have.
-#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, strum::Display, strum::EnumString)]
+#[derive(
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    serde::Deserialize,
+    serde::Serialize,
+    strum::Display,
+    strum::EnumString,
+    ToSchema,
+)]
+#[router_derive::diesel_enum(storage_type = "text")]
+#[serde(rename_all = "snake_case")]
 #[strum(serialize_all = "snake_case")]
 pub enum InvoiceStatus {
     InvoiceCreated,
diff --git a/crates/diesel_models/src/invoice.rs b/crates/diesel_models/src/invoice.rs
index 2b93b0b4a62..cbb3b251d19 100644
--- a/crates/diesel_models/src/invoice.rs
+++ b/crates/diesel_models/src/invoice.rs
@@ -18,12 +18,12 @@ pub struct InvoiceNew {
     pub customer_id: common_utils::id_type::CustomerId,
     pub amount: MinorUnit,
     pub currency: String,
-    pub status: String,
+    pub status: InvoiceStatus,
     pub provider_name: Connector,
     pub metadata: Option<SecretSerdeValue>,
     pub created_at: time::PrimitiveDateTime,
     pub modified_at: time::PrimitiveDateTime,
-    pub connector_invoice_id: Option<String>,
+    pub connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
 }
 
 #[derive(
@@ -45,20 +45,20 @@ pub struct Invoice {
     pub customer_id: common_utils::id_type::CustomerId,
     pub amount: MinorUnit,
     pub currency: String,
-    pub status: String,
+    pub status: InvoiceStatus,
     pub provider_name: Connector,
     pub metadata: Option<SecretSerdeValue>,
     pub created_at: time::PrimitiveDateTime,
     pub modified_at: time::PrimitiveDateTime,
-    pub connector_invoice_id: Option<String>,
+    pub connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq, AsChangeset, Deserialize)]
 #[diesel(table_name = invoice)]
 pub struct InvoiceUpdate {
-    pub status: Option<String>,
+    pub status: Option<InvoiceStatus>,
     pub payment_method_id: Option<String>,
-    pub connector_invoice_id: Option<String>,
+    pub connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
     pub modified_at: time::PrimitiveDateTime,
     pub payment_intent_id: Option<common_utils::id_type::PaymentId>,
 }
@@ -78,7 +78,7 @@ impl InvoiceNew {
         status: InvoiceStatus,
         provider_name: Connector,
         metadata: Option<SecretSerdeValue>,
-        connector_invoice_id: Option<String>,
+        connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
     ) -> Self {
         let id = common_utils::id_type::InvoiceId::generate();
         let now = common_utils::date_time::now();
@@ -93,7 +93,7 @@ impl InvoiceNew {
             customer_id,
             amount,
             currency,
-            status: status.to_string(),
+            status,
             provider_name,
             metadata,
             created_at: now,
@@ -107,12 +107,12 @@ impl InvoiceUpdate {
     pub fn new(
         payment_method_id: Option<String>,
         status: Option<InvoiceStatus>,
-        connector_invoice_id: Option<String>,
+        connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
         payment_intent_id: Option<common_utils::id_type::PaymentId>,
     ) -> Self {
         Self {
             payment_method_id,
-            status: status.map(|status| status.to_string()),
+            status,
             connector_invoice_id,
             payment_intent_id,
             modified_at: common_utils::date_time::now(),
diff --git a/crates/diesel_models/src/query/invoice.rs b/crates/diesel_models/src/query/invoice.rs
index 0166469d340..6e62d9b5b5f 100644
--- a/crates/diesel_models/src/query/invoice.rs
+++ b/crates/diesel_models/src/query/invoice.rs
@@ -1,4 +1,4 @@
-use diesel::{associations::HasTable, ExpressionMethods};
+use diesel::{associations::HasTable, BoolExpressionMethods, ExpressionMethods};
 
 use super::generics;
 use crate::{
@@ -66,4 +66,18 @@ impl Invoice {
             .await
         }
     }
+
+    pub async fn get_invoice_by_subscription_id_connector_invoice_id(
+        conn: &PgPooledConn,
+        subscription_id: String,
+        connector_invoice_id: common_utils::id_type::InvoiceId,
+    ) -> StorageResult<Option<Self>> {
+        generics::generic_find_one_optional::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::subscription_id
+                .eq(subscription_id.to_owned())
+                .and(dsl::connector_invoice_id.eq(connector_invoice_id.to_owned())),
+        )
+        .await
+    }
 }
diff --git a/crates/hyperswitch_domain_models/src/customer.rs b/crates/hyperswitch_domain_models/src/customer.rs
index 720296cdcac..6179e221694 100644
--- a/crates/hyperswitch_domain_models/src/customer.rs
+++ b/crates/hyperswitch_domain_models/src/customer.rs
@@ -6,6 +6,7 @@ use common_utils::{
     date_time,
     encryption::Encryption,
     errors::{CustomResult, ValidationError},
+    ext_traits::ValueExt,
     id_type, pii,
     types::{
         keymanager::{self, KeyManagerState, ToEncryptable},
@@ -90,6 +91,23 @@ impl Customer {
         &self.id
     }
 
+    /// Get the connector customer ID for the specified connector label, if present
+    #[cfg(feature = "v1")]
+    pub fn get_connector_customer_map(
+        &self,
+    ) -> FxHashMap<id_type::MerchantConnectorAccountId, String> {
+        use masking::PeekInterface;
+        if let Some(connector_customer_value) = &self.connector_customer {
+            connector_customer_value
+                .peek()
+                .clone()
+                .parse_value("ConnectorCustomerMap")
+                .unwrap_or_default()
+        } else {
+            FxHashMap::default()
+        }
+    }
+
     /// Get the connector customer ID for the specified connector label, if present
     #[cfg(feature = "v1")]
     pub fn get_connector_customer_id(&self, connector_label: &str) -> Option<&str> {
diff --git a/crates/hyperswitch_domain_models/src/invoice.rs b/crates/hyperswitch_domain_models/src/invoice.rs
index 9457eac56d2..a6b836d3e58 100644
--- a/crates/hyperswitch_domain_models/src/invoice.rs
+++ b/crates/hyperswitch_domain_models/src/invoice.rs
@@ -1,5 +1,3 @@
-use std::str::FromStr;
-
 use common_utils::{
     errors::{CustomResult, ValidationError},
     id_type::GenerateId,
@@ -9,7 +7,6 @@ use common_utils::{
         MinorUnit,
     },
 };
-use error_stack::ResultExt;
 use masking::Secret;
 use utoipa::ToSchema;
 
@@ -27,10 +24,10 @@ pub struct Invoice {
     pub customer_id: common_utils::id_type::CustomerId,
     pub amount: MinorUnit,
     pub currency: String,
-    pub status: String,
+    pub status: common_enums::connector_enums::InvoiceStatus,
     pub provider_name: common_enums::connector_enums::Connector,
     pub metadata: Option<SecretSerdeValue>,
-    pub connector_invoice_id: Option<String>,
+    pub connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
 }
 
 #[async_trait::async_trait]
@@ -89,11 +86,6 @@ impl super::behaviour::Conversion for Invoice {
     }
 
     async fn construct_new(self) -> CustomResult<Self::NewDstType, ValidationError> {
-        let invoice_status = common_enums::connector_enums::InvoiceStatus::from_str(&self.status)
-            .change_context(ValidationError::InvalidValue {
-            message: "Invalid invoice status".to_string(),
-        })?;
-
         Ok(diesel_models::invoice::InvoiceNew::new(
             self.subscription_id,
             self.merchant_id,
@@ -104,7 +96,7 @@ impl super::behaviour::Conversion for Invoice {
             self.customer_id,
             self.amount,
             self.currency.to_string(),
-            invoice_status,
+            self.status,
             self.provider_name,
             None,
             self.connector_invoice_id,
@@ -127,7 +119,7 @@ impl Invoice {
         status: common_enums::connector_enums::InvoiceStatus,
         provider_name: common_enums::connector_enums::Connector,
         metadata: Option<SecretSerdeValue>,
-        connector_invoice_id: Option<String>,
+        connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
     ) -> Self {
         Self {
             id: common_utils::id_type::InvoiceId::generate(),
@@ -140,7 +132,7 @@ impl Invoice {
             customer_id,
             amount,
             currency: currency.to_string(),
-            status: status.to_string(),
+            status,
             provider_name,
             metadata,
             connector_invoice_id,
@@ -179,12 +171,20 @@ pub trait InvoiceInterface {
         key_store: &MerchantKeyStore,
         subscription_id: String,
     ) -> CustomResult<Invoice, Self::Error>;
+
+    async fn find_invoice_by_subscription_id_connector_invoice_id(
+        &self,
+        state: &KeyManagerState,
+        key_store: &MerchantKeyStore,
+        subscription_id: String,
+        connector_invoice_id: common_utils::id_type::InvoiceId,
+    ) -> CustomResult<Option<Invoice>, Self::Error>;
 }
 
 pub struct InvoiceUpdate {
-    pub status: Option<String>,
+    pub status: Option<common_enums::connector_enums::InvoiceStatus>,
     pub payment_method_id: Option<String>,
-    pub connector_invoice_id: Option<String>,
+    pub connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
     pub modified_at: time::PrimitiveDateTime,
     pub payment_intent_id: Option<common_utils::id_type::PaymentId>,
 }
@@ -236,15 +236,15 @@ impl InvoiceUpdate {
     pub fn new(
         payment_method_id: Option<String>,
         status: Option<common_enums::connector_enums::InvoiceStatus>,
-        connector_invoice_id: Option<String>,
+        connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
         payment_intent_id: Option<common_utils::id_type::PaymentId>,
     ) -> Self {
         Self {
             payment_method_id,
-            status: status.map(|status| status.to_string()),
-            connector_invoice_id,
+            status,
             modified_at: common_utils::date_time::now(),
             payment_intent_id,
+            connector_invoice_id,
         }
     }
 }
diff --git a/crates/router/src/core/subscription.rs b/crates/router/src/core/subscription.rs
index 05254a9b77d..e2d376f3f08 100644
--- a/crates/router/src/core/subscription.rs
+++ b/crates/router/src/core/subscription.rs
@@ -35,7 +35,7 @@ pub async fn create_subscription(
         SubscriptionHandler::find_business_profile(&state, &merchant_context, &profile_id)
             .await
             .attach_printable("subscriptions: failed to find business profile")?;
-    let customer =
+    let _customer =
         SubscriptionHandler::find_customer(&state, &merchant_context, &request.customer_id)
             .await
             .attach_printable("subscriptions: failed to find customer")?;
@@ -43,7 +43,6 @@ pub async fn create_subscription(
         &state,
         merchant_context.get_merchant_account(),
         merchant_context.get_merchant_key_store(),
-        Some(customer),
         profile.clone(),
     )
     .await?;
@@ -119,7 +118,6 @@ pub async fn get_subscription_plans(
         &state,
         merchant_context.get_merchant_account(),
         merchant_context.get_merchant_key_store(),
-        None,
         profile.clone(),
     )
     .await?;
@@ -169,7 +167,6 @@ pub async fn create_and_confirm_subscription(
         &state,
         merchant_context.get_merchant_account(),
         merchant_context.get_merchant_key_store(),
-        Some(customer),
         profile.clone(),
     )
     .await?;
@@ -187,9 +184,10 @@ pub async fn create_and_confirm_subscription(
         .attach_printable("subscriptions: failed to create subscription entry")?;
     let invoice_handler = subs_handler.get_invoice_handler(profile.clone());
 
-    let _customer_create_response = billing_handler
+    let customer_create_response = billing_handler
         .create_customer_on_connector(
             &state,
+            customer.clone(),
             request.customer_id.clone(),
             request.billing.clone(),
             request
@@ -199,6 +197,15 @@ pub async fn create_and_confirm_subscription(
                 .and_then(|data| data.payment_method_data),
         )
         .await?;
+    let _customer_updated_response = SubscriptionHandler::update_connector_customer_id_in_customer(
+        &state,
+        &merchant_context,
+        &billing_handler.merchant_connector_id,
+        &customer,
+        customer_create_response,
+    )
+    .await
+    .attach_printable("Failed to update customer with connector customer ID")?;
 
     let subscription_create_response = billing_handler
         .create_subscription_on_connector(
@@ -232,9 +239,7 @@ pub async fn create_and_confirm_subscription(
                 .unwrap_or(connector_enums::InvoiceStatus::InvoiceCreated),
             billing_handler.connector_data.connector_name,
             None,
-            invoice_details
-                .clone()
-                .map(|invoice| invoice.id.get_string_repr().to_string()),
+            invoice_details.clone().map(|invoice| invoice.id),
         )
         .await?;
 
@@ -242,13 +247,7 @@ pub async fn create_and_confirm_subscription(
         .create_invoice_sync_job(
             &state,
             &invoice_entry,
-            invoice_details
-                .ok_or(errors::ApiErrorResponse::MissingRequiredField {
-                    field_name: "invoice_details",
-                })?
-                .id
-                .get_string_repr()
-                .to_string(),
+            invoice_details.clone().map(|details| details.id),
             billing_handler.connector_data.connector_name,
         )
         .await?;
@@ -327,16 +326,16 @@ pub async fn confirm_subscription(
         &state,
         merchant_context.get_merchant_account(),
         merchant_context.get_merchant_key_store(),
-        Some(customer),
         profile.clone(),
     )
     .await?;
     let invoice_handler = subscription_entry.get_invoice_handler(profile);
     let subscription = subscription_entry.subscription.clone();
 
-    let _customer_create_response = billing_handler
+    let customer_create_response = billing_handler
         .create_customer_on_connector(
             &state,
+            customer.clone(),
             subscription.customer_id.clone(),
             request.payment_details.payment_method_data.billing.clone(),
             request
@@ -345,6 +344,15 @@ pub async fn confirm_subscription(
                 .payment_method_data,
         )
         .await?;
+    let _customer_updated_response = SubscriptionHandler::update_connector_customer_id_in_customer(
+        &state,
+        &merchant_context,
+        &billing_handler.merchant_connector_id,
+        &customer,
+        customer_create_response,
+    )
+    .await
+    .attach_printable("Failed to update customer with connector customer ID")?;
 
     let subscription_create_response = billing_handler
         .create_subscription_on_connector(
@@ -366,9 +374,7 @@ pub async fn confirm_subscription(
                 .clone()
                 .and_then(|invoice| invoice.status)
                 .unwrap_or(connector_enums::InvoiceStatus::InvoiceCreated),
-            invoice_details
-                .clone()
-                .map(|invoice| invoice.id.get_string_repr().to_string()),
+            invoice_details.clone().map(|invoice| invoice.id),
         )
         .await?;
 
@@ -376,14 +382,7 @@ pub async fn confirm_subscription(
         .create_invoice_sync_job(
             &state,
             &invoice_entry,
-            invoice_details
-                .clone()
-                .ok_or(errors::ApiErrorResponse::MissingRequiredField {
-                    field_name: "invoice_details",
-                })?
-                .id
-                .get_string_repr()
-                .to_string(),
+            invoice_details.map(|invoice| invoice.id),
             billing_handler.connector_data.connector_name,
         )
         .await?;
@@ -449,7 +448,6 @@ pub async fn get_estimate(
         &state,
         merchant_context.get_merchant_account(),
         merchant_context.get_merchant_key_store(),
-        None,
         profile,
     )
     .await?;
diff --git a/crates/router/src/core/subscription/billing_processor_handler.rs b/crates/router/src/core/subscription/billing_processor_handler.rs
index 910596d66c4..00ca480e5ea 100644
--- a/crates/router/src/core/subscription/billing_processor_handler.rs
+++ b/crates/router/src/core/subscription/billing_processor_handler.rs
@@ -31,7 +31,6 @@ pub struct BillingHandler {
     pub connector_data: api_types::ConnectorData,
     pub connector_params: hyperswitch_domain_models::connector_endpoints::ConnectorParams,
     pub connector_metadata: Option<pii::SecretSerdeValue>,
-    pub customer: Option<hyperswitch_domain_models::customer::Customer>,
     pub merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
 }
 
@@ -41,7 +40,6 @@ impl BillingHandler {
         state: &SessionState,
         merchant_account: &hyperswitch_domain_models::merchant_account::MerchantAccount,
         key_store: &hyperswitch_domain_models::merchant_key_store::MerchantKeyStore,
-        customer: Option<hyperswitch_domain_models::customer::Customer>,
         profile: hyperswitch_domain_models::business_profile::Profile,
     ) -> errors::RouterResult<Self> {
         let merchant_connector_id = profile.get_billing_processor_id()?;
@@ -102,24 +100,22 @@ impl BillingHandler {
             connector_data,
             connector_params,
             connector_metadata: billing_processor_mca.metadata.clone(),
-            customer,
             merchant_connector_id,
         })
     }
     pub async fn create_customer_on_connector(
         &self,
         state: &SessionState,
+        customer: hyperswitch_domain_models::customer::Customer,
         customer_id: common_utils::id_type::CustomerId,
         billing_address: Option<api_models::payments::Address>,
         payment_method_data: Option<api_models::payments::PaymentMethodData>,
-    ) -> errors::RouterResult<ConnectorCustomerResponseData> {
-        let customer =
-            self.customer
-                .as_ref()
-                .ok_or(errors::ApiErrorResponse::MissingRequiredField {
-                    field_name: "customer",
-                })?;
-
+    ) -> errors::RouterResult<Option<ConnectorCustomerResponseData>> {
+        let connector_customer_map = customer.get_connector_customer_map();
+        if connector_customer_map.contains_key(&self.merchant_connector_id) {
+            // Customer already exists on the connector, no need to create again
+            return Ok(None);
+        }
         let customer_req = ConnectorCustomerData {
             email: customer.email.clone().map(pii::Email::from),
             payment_method_data: payment_method_data.clone().map(|pmd| pmd.into()),
@@ -156,7 +152,7 @@ impl BillingHandler {
         match response {
             Ok(response_data) => match response_data {
                 PaymentsResponseData::ConnectorCustomerResponse(customer_response) => {
-                    Ok(customer_response)
+                    Ok(Some(customer_response))
                 }
                 _ => Err(errors::ApiErrorResponse::SubscriptionError {
                     operation: "Subscription Customer Create".to_string(),
@@ -233,7 +229,7 @@ impl BillingHandler {
     pub async fn record_back_to_billing_processor(
         &self,
         state: &SessionState,
-        invoice_id: String,
+        invoice_id: common_utils::id_type::InvoiceId,
         payment_id: common_utils::id_type::PaymentId,
         payment_status: common_enums::AttemptStatus,
         amount: common_utils::types::MinorUnit,
@@ -245,10 +241,12 @@ impl BillingHandler {
             currency,
             payment_method_type,
             attempt_status: payment_status,
-            merchant_reference_id: common_utils::id_type::PaymentReferenceId::from_str(&invoice_id)
-                .change_context(errors::ApiErrorResponse::InvalidDataValue {
-                    field_name: "invoice_id",
-                })?,
+            merchant_reference_id: common_utils::id_type::PaymentReferenceId::from_str(
+                invoice_id.get_string_repr(),
+            )
+            .change_context(errors::ApiErrorResponse::InvalidDataValue {
+                field_name: "invoice_id",
+            })?,
             connector_params: self.connector_params.clone(),
             connector_transaction_id: Some(common_utils::types::ConnectorTransactionId::TxnId(
                 payment_id.get_string_repr().to_string(),
@@ -392,7 +390,6 @@ impl BillingHandler {
                 connector_integration,
             )
             .await?;
-
         match response {
             Ok(resp) => Ok(resp),
             Err(err) => Err(errors::ApiErrorResponse::ExternalConnectorError {
diff --git a/crates/router/src/core/subscription/invoice_handler.rs b/crates/router/src/core/subscription/invoice_handler.rs
index b383f7a4650..bcdc1b74779 100644
--- a/crates/router/src/core/subscription/invoice_handler.rs
+++ b/crates/router/src/core/subscription/invoice_handler.rs
@@ -10,9 +10,7 @@ use masking::{PeekInterface, Secret};
 
 use super::errors;
 use crate::{
-    core::{errors::utils::StorageErrorExt, subscription::payments_api_client},
-    routes::SessionState,
-    types::storage as storage_types,
+    core::subscription::payments_api_client, routes::SessionState, types::storage as storage_types,
     workflows::invoice_sync as invoice_sync_workflow,
 };
 
@@ -20,6 +18,7 @@ pub struct InvoiceHandler {
     pub subscription: hyperswitch_domain_models::subscription::Subscription,
     pub merchant_account: hyperswitch_domain_models::merchant_account::MerchantAccount,
     pub profile: hyperswitch_domain_models::business_profile::Profile,
+    pub merchant_key_store: hyperswitch_domain_models::merchant_key_store::MerchantKeyStore,
 }
 
 #[allow(clippy::todo)]
@@ -28,11 +27,13 @@ impl InvoiceHandler {
         subscription: hyperswitch_domain_models::subscription::Subscription,
         merchant_account: hyperswitch_domain_models::merchant_account::MerchantAccount,
         profile: hyperswitch_domain_models::business_profile::Profile,
+        merchant_key_store: hyperswitch_domain_models::merchant_key_store::MerchantKeyStore,
     ) -> Self {
         Self {
             subscription,
             merchant_account,
             profile,
+            merchant_key_store,
         }
     }
     #[allow(clippy::too_many_arguments)]
@@ -46,7 +47,7 @@ impl InvoiceHandler {
         status: connector_enums::InvoiceStatus,
         provider_name: connector_enums::Connector,
         metadata: Option<pii::SecretSerdeValue>,
-        connector_invoice_id: Option<String>,
+        connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
     ) -> errors::RouterResult<hyperswitch_domain_models::invoice::Invoice> {
         let invoice_new = hyperswitch_domain_models::invoice::Invoice::to_invoice(
             self.subscription.id.to_owned(),
@@ -65,18 +66,9 @@ impl InvoiceHandler {
         );
 
         let key_manager_state = &(state).into();
-        let merchant_key_store = state
-            .store
-            .get_merchant_key_store_by_merchant_id(
-                key_manager_state,
-                self.merchant_account.get_id(),
-                &state.store.get_master_key().to_vec().into(),
-            )
-            .await
-            .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
         let invoice = state
             .store
-            .insert_invoice_entry(key_manager_state, &merchant_key_store, invoice_new)
+            .insert_invoice_entry(key_manager_state, &self.merchant_key_store, invoice_new)
             .await
             .change_context(errors::ApiErrorResponse::SubscriptionError {
                 operation: "Create Invoice".to_string(),
@@ -93,7 +85,7 @@ impl InvoiceHandler {
         payment_method_id: Option<Secret<String>>,
         payment_intent_id: Option<common_utils::id_type::PaymentId>,
         status: connector_enums::InvoiceStatus,
-        connector_invoice_id: Option<String>,
+        connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
     ) -> errors::RouterResult<hyperswitch_domain_models::invoice::Invoice> {
         let update_invoice = hyperswitch_domain_models::invoice::InvoiceUpdate::new(
             payment_method_id.as_ref().map(|id| id.peek()).cloned(),
@@ -102,20 +94,11 @@ impl InvoiceHandler {
             payment_intent_id,
         );
         let key_manager_state = &(state).into();
-        let merchant_key_store = state
-            .store
-            .get_merchant_key_store_by_merchant_id(
-                key_manager_state,
-                self.merchant_account.get_id(),
-                &state.store.get_master_key().to_vec().into(),
-            )
-            .await
-            .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
         state
             .store
             .update_invoice_entry(
                 key_manager_state,
-                &merchant_key_store,
+                &self.merchant_key_store,
                 invoice_id.get_string_repr().to_string(),
                 update_invoice,
             )
@@ -151,6 +134,7 @@ impl InvoiceHandler {
             customer_id: Some(self.subscription.customer_id.clone()),
             billing: request.billing.clone(),
             shipping: request.shipping.clone(),
+            profile_id: Some(self.profile.get_id().clone()),
             setup_future_usage: payment_details.setup_future_usage,
             return_url: Some(payment_details.return_url.clone()),
             capture_method: payment_details.capture_method,
@@ -194,6 +178,7 @@ impl InvoiceHandler {
             customer_id: Some(self.subscription.customer_id.clone()),
             billing: request.billing.clone(),
             shipping: request.shipping.clone(),
+            profile_id: Some(self.profile.get_id().clone()),
             setup_future_usage: payment_details.setup_future_usage,
             return_url: payment_details.return_url.clone(),
             capture_method: payment_details.capture_method,
@@ -202,6 +187,7 @@ impl InvoiceHandler {
             payment_method_type: payment_details.payment_method_type,
             payment_method_data: payment_details.payment_method_data.clone(),
             customer_acceptance: payment_details.customer_acceptance.clone(),
+            payment_type: payment_details.payment_type,
         };
         payments_api_client::PaymentsApiClient::create_and_confirm_payment(
             state,
@@ -222,10 +208,12 @@ impl InvoiceHandler {
         let cit_payment_request = subscription_types::ConfirmPaymentsRequestData {
             billing: request.payment_details.payment_method_data.billing.clone(),
             shipping: request.payment_details.shipping.clone(),
+            profile_id: Some(self.profile.get_id().clone()),
             payment_method: payment_details.payment_method,
             payment_method_type: payment_details.payment_method_type,
             payment_method_data: payment_details.payment_method_data.clone(),
             customer_acceptance: payment_details.customer_acceptance.clone(),
+            payment_type: payment_details.payment_type,
         };
         payments_api_client::PaymentsApiClient::confirm_payment(
             state,
@@ -242,20 +230,11 @@ impl InvoiceHandler {
         state: &SessionState,
     ) -> errors::RouterResult<hyperswitch_domain_models::invoice::Invoice> {
         let key_manager_state = &(state).into();
-        let merchant_key_store = state
-            .store
-            .get_merchant_key_store_by_merchant_id(
-                key_manager_state,
-                self.merchant_account.get_id(),
-                &state.store.get_master_key().to_vec().into(),
-            )
-            .await
-            .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
         state
             .store
             .get_latest_invoice_for_subscription(
                 key_manager_state,
-                &merchant_key_store,
+                &self.merchant_key_store,
                 self.subscription.id.get_string_repr().to_string(),
             )
             .await
@@ -271,20 +250,11 @@ impl InvoiceHandler {
         invoice_id: common_utils::id_type::InvoiceId,
     ) -> errors::RouterResult<hyperswitch_domain_models::invoice::Invoice> {
         let key_manager_state = &(state).into();
-        let merchant_key_store = state
-            .store
-            .get_merchant_key_store_by_merchant_id(
-                key_manager_state,
-                self.merchant_account.get_id(),
-                &state.store.get_master_key().to_vec().into(),
-            )
-            .await
-            .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
         state
             .store
             .find_invoice_by_invoice_id(
                 key_manager_state,
-                &merchant_key_store,
+                &self.merchant_key_store,
                 invoice_id.get_string_repr().to_string(),
             )
             .await
@@ -294,11 +264,33 @@ impl InvoiceHandler {
             .attach_printable("invoices: unable to get invoice by id from database")
     }
 
+    pub async fn find_invoice_by_subscription_id_connector_invoice_id(
+        &self,
+        state: &SessionState,
+        subscription_id: common_utils::id_type::SubscriptionId,
+        connector_invoice_id: common_utils::id_type::InvoiceId,
+    ) -> errors::RouterResult<Option<hyperswitch_domain_models::invoice::Invoice>> {
+        let key_manager_state = &(state).into();
+        state
+            .store
+            .find_invoice_by_subscription_id_connector_invoice_id(
+                key_manager_state,
+                &self.merchant_key_store,
+                subscription_id.get_string_repr().to_string(),
+                connector_invoice_id,
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::SubscriptionError {
+                operation: "Get Invoice by Subscription ID and Connector Invoice ID".to_string(),
+            })
+            .attach_printable("invoices: unable to get invoice by subscription id and connector invoice id from database")
+    }
+
     pub async fn create_invoice_sync_job(
         &self,
         state: &SessionState,
         invoice: &hyperswitch_domain_models::invoice::Invoice,
-        connector_invoice_id: String,
+        connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
         connector_name: connector_enums::Connector,
     ) -> errors::RouterResult<()> {
         let request = storage_types::invoice_sync::InvoiceSyncRequest::new(
@@ -333,6 +325,7 @@ impl InvoiceHandler {
                 payment_method_id.to_owned(),
             )),
             off_session: Some(true),
+            profile_id: Some(self.profile.get_id().clone()),
         };
 
         payments_api_client::PaymentsApiClient::create_mit_payment(
diff --git a/crates/router/src/core/subscription/payments_api_client.rs b/crates/router/src/core/subscription/payments_api_client.rs
index e530a5843c8..b688e3bc339 100644
--- a/crates/router/src/core/subscription/payments_api_client.rs
+++ b/crates/router/src/core/subscription/payments_api_client.rs
@@ -36,6 +36,10 @@ impl PaymentsApiClient {
                         .clone(),
                 ),
             ),
+            (
+                headers::X_TENANT_ID.to_string(),
+                masking::Maskable::Normal(state.tenant.tenant_id.get_string_repr().to_string()),
+            ),
             (
                 headers::X_MERCHANT_ID.to_string(),
                 masking::Maskable::Normal(merchant_id.to_string()),
diff --git a/crates/router/src/core/subscription/subscription_handler.rs b/crates/router/src/core/subscription/subscription_handler.rs
index 76799c21a74..1f29f38073e 100644
--- a/crates/router/src/core/subscription/subscription_handler.rs
+++ b/crates/router/src/core/subscription/subscription_handler.rs
@@ -9,14 +9,17 @@ use common_utils::{consts, ext_traits::OptionExt};
 use error_stack::ResultExt;
 use hyperswitch_domain_models::{
     merchant_context::MerchantContext,
-    router_response_types::subscriptions as subscription_response_types,
+    router_response_types::{self, subscriptions as subscription_response_types},
     subscription::{Subscription, SubscriptionStatus},
 };
 use masking::Secret;
 
 use super::errors;
 use crate::{
-    core::{errors::StorageErrorExt, subscription::invoice_handler::InvoiceHandler},
+    core::{
+        errors::StorageErrorExt, payments as payments_core,
+        subscription::invoice_handler::InvoiceHandler,
+    },
     db::CustomResult,
     routes::SessionState,
     types::{domain, transformers::ForeignTryFrom},
@@ -109,6 +112,64 @@ impl<'a> SubscriptionHandler<'a> {
             .change_context(errors::ApiErrorResponse::CustomerNotFound)
             .attach_printable("subscriptions: unable to fetch customer from database")
     }
+    pub async fn update_connector_customer_id_in_customer(
+        state: &SessionState,
+        merchant_context: &MerchantContext,
+        merchant_connector_id: &common_utils::id_type::MerchantConnectorAccountId,
+        customer: &hyperswitch_domain_models::customer::Customer,
+        customer_create_response: Option<router_response_types::ConnectorCustomerResponseData>,
+    ) -> errors::RouterResult<hyperswitch_domain_models::customer::Customer> {
+        match customer_create_response {
+            Some(customer_response) => {
+                match payments_core::customers::update_connector_customer_in_customers(
+                    merchant_connector_id.get_string_repr(),
+                    Some(customer),
+                    Some(customer_response.connector_customer_id),
+                )
+                .await
+                {
+                    Some(customer_update) => Self::update_customer(
+                        state,
+                        merchant_context,
+                        customer.clone(),
+                        customer_update,
+                    )
+                    .await
+                    .attach_printable("Failed to update customer with connector customer ID"),
+                    None => Ok(customer.clone()),
+                }
+            }
+            None => Ok(customer.clone()),
+        }
+    }
+
+    pub async fn update_customer(
+        state: &SessionState,
+        merchant_context: &MerchantContext,
+        customer: hyperswitch_domain_models::customer::Customer,
+        customer_update: domain::CustomerUpdate,
+    ) -> errors::RouterResult<hyperswitch_domain_models::customer::Customer> {
+        let key_manager_state = &(state).into();
+        let merchant_key_store = merchant_context.get_merchant_key_store();
+        let merchant_id = merchant_context.get_merchant_account().get_id();
+        let db = state.store.as_ref();
+
+        let updated_customer = db
+            .update_customer_by_customer_id_merchant_id(
+                key_manager_state,
+                customer.customer_id.clone(),
+                merchant_id.clone(),
+                customer,
+                customer_update,
+                merchant_key_store,
+                merchant_context.get_merchant_account().storage_scheme,
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("subscriptions: unable to update customer entry in database")?;
+
+        Ok(updated_customer)
+    }
 
     /// Helper function to find business profile.
     pub async fn find_business_profile(
@@ -304,6 +365,11 @@ impl SubscriptionWithHandler<'_> {
             subscription: self.subscription.clone(),
             merchant_account: self.merchant_account.clone(),
             profile,
+            merchant_key_store: self
+                .handler
+                .merchant_context
+                .get_merchant_key_store()
+                .clone(),
         }
     }
     pub async fn get_mca(
diff --git a/crates/router/src/core/webhooks/incoming.rs b/crates/router/src/core/webhooks/incoming.rs
index ce183002834..62db07c8a3f 100644
--- a/crates/router/src/core/webhooks/incoming.rs
+++ b/crates/router/src/core/webhooks/incoming.rs
@@ -2600,10 +2600,6 @@ async fn subscription_incoming_webhook_flow(
         .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
         .attach_printable("Failed to extract MIT payment data from subscription webhook")?;
 
-    if mit_payment_data.first_invoice {
-        return Ok(WebhookResponseTracker::NoEffect);
-    }
-
     let profile_id = business_profile.get_id().clone();
 
     let profile =
@@ -2618,11 +2614,29 @@ async fn subscription_incoming_webhook_flow(
     let subscription_id = mit_payment_data.subscription_id.clone();
 
     let subscription_with_handler = handler
-        .find_subscription(subscription_id)
+        .find_subscription(subscription_id.clone())
         .await
         .attach_printable("subscriptions: failed to get subscription entry in get_subscription")?;
 
     let invoice_handler = subscription_with_handler.get_invoice_handler(profile.clone());
+    let invoice = invoice_handler
+        .find_invoice_by_subscription_id_connector_invoice_id(
+            &state,
+            subscription_id,
+            mit_payment_data.invoice_id.clone(),
+        )
+        .await
+        .attach_printable(
+            "subscriptions: failed to get invoice by subscription id and connector invoice id",
+        )?;
+    if let Some(invoice) = invoice {
+        // During CIT payment we would have already created invoice entry with status as PaymentPending or Paid.
+        // So we skip incoming webhook for the already processed invoice
+        if invoice.status != InvoiceStatus::InvoiceCreated {
+            logger::info!("Invoice is already being processed, skipping MIT payment creation");
+            return Ok(WebhookResponseTracker::NoEffect);
+        }
+    }
 
     let payment_method_id = subscription_with_handler
         .subscription
@@ -2635,17 +2649,36 @@ async fn subscription_incoming_webhook_flow(
 
     logger::info!("Payment method ID found: {}", payment_method_id);
 
+    let payment_id = generate_id(consts::ID_LENGTH, "pay");
+    let payment_id = common_utils::id_type::PaymentId::wrap(payment_id).change_context(
+        errors::ApiErrorResponse::InvalidDataValue {
+            field_name: "payment_id",
+        },
+    )?;
+
+    // Multiple MIT payments for the same invoice_generated event is avoided by having the unique constraint on (subscription_id, connector_invoice_id) in the invoices table
     let invoice_entry = invoice_handler
         .create_invoice_entry(
             &state,
             billing_connector_mca_id.clone(),
-            None,
+            Some(payment_id),
             mit_payment_data.amount_due,
             mit_payment_data.currency_code,
             InvoiceStatus::PaymentPending,
             connector,
             None,
-            None,
+            Some(mit_payment_data.invoice_id.clone()),
+        )
+        .await?;
+
+    // Create a sync job for the invoice with generated payment_id before initiating MIT payment creation.
+    // This ensures that if payment creation call fails, the sync job can still retrieve the payment status
+    invoice_handler
+        .create_invoice_sync_job(
+            &state,
+            &invoice_entry,
+            Some(mit_payment_data.invoice_id.clone()),
+            connector,
         )
         .await?;
 
@@ -2658,23 +2691,14 @@ async fn subscription_incoming_webhook_flow(
         )
         .await?;
 
-    let updated_invoice = invoice_handler
+    let _updated_invoice = invoice_handler
         .update_invoice(
             &state,
             invoice_entry.id.clone(),
             payment_response.payment_method_id.clone(),
             Some(payment_response.payment_id.clone()),
             InvoiceStatus::from(payment_response.status),
-            Some(mit_payment_data.invoice_id.get_string_repr().to_string()),
-        )
-        .await?;
-
-    invoice_handler
-        .create_invoice_sync_job(
-            &state,
-            &updated_invoice,
-            mit_payment_data.invoice_id.get_string_repr().to_string(),
-            connector,
+            Some(mit_payment_data.invoice_id.clone()),
         )
         .await?;
 
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 84799f91cf5..08e7b18678b 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -4393,6 +4393,24 @@ impl InvoiceInterface for KafkaStore {
             .get_latest_invoice_for_subscription(state, key_store, subscription_id)
             .await
     }
+
+    #[instrument(skip_all)]
+    async fn find_invoice_by_subscription_id_connector_invoice_id(
+        &self,
+        state: &KeyManagerState,
+        key_store: &hyperswitch_domain_models::merchant_key_store::MerchantKeyStore,
+        subscription_id: String,
+        connector_invoice_id: id_type::InvoiceId,
+    ) -> CustomResult<Option<DomainInvoice>, errors::StorageError> {
+        self.diesel_store
+            .find_invoice_by_subscription_id_connector_invoice_id(
+                state,
+                key_store,
+                subscription_id,
+                connector_invoice_id,
+            )
+            .await
+    }
 }
 
 #[async_trait::async_trait]
diff --git a/crates/router/src/types/storage/invoice_sync.rs b/crates/router/src/types/storage/invoice_sync.rs
index 01bbcf17d92..93bd983765b 100644
--- a/crates/router/src/types/storage/invoice_sync.rs
+++ b/crates/router/src/types/storage/invoice_sync.rs
@@ -7,7 +7,8 @@ pub struct InvoiceSyncTrackingData {
     pub merchant_id: id_type::MerchantId,
     pub profile_id: id_type::ProfileId,
     pub customer_id: id_type::CustomerId,
-    pub connector_invoice_id: String,
+    // connector_invoice_id is optional because in some cases (Trial/Future), the invoice might not have been created in the connector yet.
+    pub connector_invoice_id: Option<id_type::InvoiceId>,
     pub connector_name: api_enums::Connector, // The connector to which the invoice belongs
 }
 
@@ -18,7 +19,7 @@ pub struct InvoiceSyncRequest {
     pub merchant_id: id_type::MerchantId,
     pub profile_id: id_type::ProfileId,
     pub customer_id: id_type::CustomerId,
-    pub connector_invoice_id: String,
+    pub connector_invoice_id: Option<id_type::InvoiceId>,
     pub connector_name: api_enums::Connector,
 }
 
@@ -44,7 +45,7 @@ impl InvoiceSyncRequest {
         merchant_id: id_type::MerchantId,
         profile_id: id_type::ProfileId,
         customer_id: id_type::CustomerId,
-        connector_invoice_id: String,
+        connector_invoice_id: Option<id_type::InvoiceId>,
         connector_name: api_enums::Connector,
     ) -> Self {
         Self {
@@ -67,7 +68,7 @@ impl InvoiceSyncTrackingData {
         merchant_id: id_type::MerchantId,
         profile_id: id_type::ProfileId,
         customer_id: id_type::CustomerId,
-        connector_invoice_id: String,
+        connector_invoice_id: Option<id_type::InvoiceId>,
         connector_name: api_enums::Connector,
     ) -> Self {
         Self {
diff --git a/crates/router/src/workflows/invoice_sync.rs b/crates/router/src/workflows/invoice_sync.rs
index d72e59904f1..72c3a11071c 100644
--- a/crates/router/src/workflows/invoice_sync.rs
+++ b/crates/router/src/workflows/invoice_sync.rs
@@ -162,11 +162,31 @@ impl<'a> InvoiceSyncHandler<'a> {
         Ok(payments_response)
     }
 
+    pub async fn perform_billing_processor_record_back_if_possible(
+        &self,
+        payment_response: subscription_types::PaymentResponseData,
+        payment_status: common_enums::AttemptStatus,
+        connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
+        invoice_sync_status: storage::invoice_sync::InvoiceSyncPaymentStatus,
+    ) -> CustomResult<(), router_errors::ApiErrorResponse> {
+        if let Some(connector_invoice_id) = connector_invoice_id {
+            Box::pin(self.perform_billing_processor_record_back(
+                payment_response,
+                payment_status,
+                connector_invoice_id,
+                invoice_sync_status,
+            ))
+            .await
+            .attach_printable("Failed to record back to billing processor")?;
+        }
+        Ok(())
+    }
+
     pub async fn perform_billing_processor_record_back(
         &self,
         payment_response: subscription_types::PaymentResponseData,
         payment_status: common_enums::AttemptStatus,
-        connector_invoice_id: String,
+        connector_invoice_id: common_utils::id_type::InvoiceId,
         invoice_sync_status: storage::invoice_sync::InvoiceSyncPaymentStatus,
     ) -> CustomResult<(), router_errors::ApiErrorResponse> {
         logger::info!("perform_billing_processor_record_back");
@@ -175,7 +195,6 @@ impl<'a> InvoiceSyncHandler<'a> {
             self.state,
             &self.merchant_account,
             &self.key_store,
-            Some(self.customer.clone()),
             self.profile.clone(),
         )
         .await
@@ -185,6 +204,7 @@ impl<'a> InvoiceSyncHandler<'a> {
             self.subscription.clone(),
             self.merchant_account.clone(),
             self.profile.clone(),
+            self.key_store.clone(),
         );
 
         // TODO: Handle retries here on failure
@@ -220,20 +240,20 @@ impl<'a> InvoiceSyncHandler<'a> {
         &self,
         process: storage::ProcessTracker,
         payment_response: subscription_types::PaymentResponseData,
-        connector_invoice_id: String,
+        connector_invoice_id: Option<common_utils::id_type::InvoiceId>,
     ) -> CustomResult<(), router_errors::ApiErrorResponse> {
         let invoice_sync_status =
             storage::invoice_sync::InvoiceSyncPaymentStatus::from(payment_response.status);
         match invoice_sync_status {
             storage::invoice_sync::InvoiceSyncPaymentStatus::PaymentSucceeded => {
-                Box::pin(self.perform_billing_processor_record_back(
-                    payment_response,
+                Box::pin(self.perform_billing_processor_record_back_if_possible(
+                    payment_response.clone(),
                     common_enums::AttemptStatus::Charged,
                     connector_invoice_id,
-                    invoice_sync_status,
+                    invoice_sync_status.clone(),
                 ))
                 .await
-                .attach_printable("Failed to record back to billing processor")?;
+                .attach_printable("Failed to record back success status to billing processor")?;
 
                 self.finish_process_with_business_status(&process, business_status::COMPLETED_BY_PT)
                     .await
@@ -256,14 +276,14 @@ impl<'a> InvoiceSyncHandler<'a> {
                 .attach_printable("Failed to update process tracker status")
             }
             storage::invoice_sync::InvoiceSyncPaymentStatus::PaymentFailed => {
-                Box::pin(self.perform_billing_processor_record_back(
-                    payment_response,
-                    common_enums::AttemptStatus::Failure,
+                Box::pin(self.perform_billing_processor_record_back_if_possible(
+                    payment_response.clone(),
+                    common_enums::AttemptStatus::Charged,
                     connector_invoice_id,
-                    invoice_sync_status,
+                    invoice_sync_status.clone(),
                 ))
                 .await
-                .attach_printable("Failed to record back to billing processor")?;
+                .attach_printable("Failed to record back failure status to billing processor")?;
 
                 self.finish_process_with_business_status(&process, business_status::COMPLETED_BY_PT)
                     .await
diff --git a/crates/storage_impl/src/invoice.rs b/crates/storage_impl/src/invoice.rs
index 95330239167..0a3a93d2b90 100644
--- a/crates/storage_impl/src/invoice.rs
+++ b/crates/storage_impl/src/invoice.rs
@@ -100,6 +100,27 @@ impl<T: DatabaseStore> InvoiceInterface for RouterStore<T> {
                 subscription_id
             ))))
     }
+
+    #[instrument(skip_all)]
+    async fn find_invoice_by_subscription_id_connector_invoice_id(
+        &self,
+        state: &KeyManagerState,
+        key_store: &MerchantKeyStore,
+        subscription_id: String,
+        connector_invoice_id: common_utils::id_type::InvoiceId,
+    ) -> CustomResult<Option<DomainInvoice>, StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+        self.find_optional_resource(
+            state,
+            key_store,
+            Invoice::get_invoice_by_subscription_id_connector_invoice_id(
+                &conn,
+                subscription_id,
+                connector_invoice_id,
+            ),
+        )
+        .await
+    }
 }
 
 #[async_trait::async_trait]
@@ -154,6 +175,24 @@ impl<T: DatabaseStore> InvoiceInterface for KVRouterStore<T> {
             .get_latest_invoice_for_subscription(state, key_store, subscription_id)
             .await
     }
+
+    #[instrument(skip_all)]
+    async fn find_invoice_by_subscription_id_connector_invoice_id(
+        &self,
+        state: &KeyManagerState,
+        key_store: &MerchantKeyStore,
+        subscription_id: String,
+        connector_invoice_id: common_utils::id_type::InvoiceId,
+    ) -> CustomResult<Option<DomainInvoice>, StorageError> {
+        self.router_store
+            .find_invoice_by_subscription_id_connector_invoice_id(
+                state,
+                key_store,
+                subscription_id,
+                connector_invoice_id,
+            )
+            .await
+    }
 }
 
 #[async_trait::async_trait]
@@ -197,4 +236,14 @@ impl InvoiceInterface for MockDb {
     ) -> CustomResult<DomainInvoice, StorageError> {
         Err(StorageError::MockDbError)?
     }
+
+    async fn find_invoice_by_subscription_id_connector_invoice_id(
+        &self,
+        _state: &KeyManagerState,
+        _key_store: &MerchantKeyStore,
+        _subscription_id: String,
+        _connector_invoice_id: common_utils::id_type::InvoiceId,
+    ) -> CustomResult<Option<DomainInvoice>, StorageError> {
+        Err(StorageError::MockDbError)?
+    }
 }
diff --git a/migrations/2025-10-09-171834_invoice_subscription_id_connector_invoice_id_unique_index/down.sql b/migrations/2025-10-09-171834_invoice_subscription_id_connector_invoice_id_unique_index/down.sql
new file mode 100644
index 00000000000..9972bff1772
--- /dev/null
+++ b/migrations/2025-10-09-171834_invoice_subscription_id_connector_invoice_id_unique_index/down.sql
@@ -0,0 +1,3 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE invoice DROP CONSTRAINT IF EXISTS invoice_subscription_id_connector_invoice_id_unique_index;
+DROP INDEX IF EXISTS invoice_subscription_id_connector_invoice_id_unique_index;
\ No newline at end of file
diff --git a/migrations/2025-10-09-171834_invoice_subscription_id_connector_invoice_id_unique_index/up.sql b/migrations/2025-10-09-171834_invoice_subscription_id_connector_invoice_id_unique_index/up.sql
new file mode 100644
index 00000000000..f455ec85e55
--- /dev/null
+++ b/migrations/2025-10-09-171834_invoice_subscription_id_connector_invoice_id_unique_index/up.sql
@@ -0,0 +1,2 @@
+-- Your SQL goes here
+ALTER TABLE invoice ADD CONSTRAINT invoice_subscription_id_connector_invoice_id_unique_index UNIQUE (subscription_id, connector_invoice_id);
\ No newline at end of file

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

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

## Description
<!-- Describe your changes in detail -->
1. Add support to create a subscription with trial plans.
When this type of subscription is created, a payment of $0 will be made to validate the payment method and save it for future use. The incoming webhook for invoice-generated event should work similarly to the MIT flow.

2. Add support to make 3DS payments in subscriptions

### 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)?
-->
1. Create Merchant Account, create API key, create payment processor, create billing processor
2. Create Customer
Request
```
curl --location 'http://localhost:8080/customers' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_4IM0uzCEoKQ9lkSaTUEceuErgHTSnZhcc0CsjiSc1P2jVv5W4p9nYumBWwnJLqLb' \
--header 'X-Tenant-Id: public' \
--data-raw '{
    "email": "guest@example.com",
    "name": "John Doe",
    "phone": "999999999",
    "phone_country_code": "+65",
    "description": "First customer",
    "metadata": {
        "udf1": "value1",
        "new_customer": "true",
        "login_date": "2019-09-10T10:11:12Z"
    },
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "IN",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "123456789",
            "country_code": "+1"
        }
    }
}'
```
Response
```
{
    "customer_id": "cus_PqvEsNGKq1kLRdX7HTHJ",
    "name": "John Doe",
    "email": "guest@example.com",
    "phone": "999999999",
    "phone_country_code": "+65",
    "description": "First customer",
    "address": null,
    "created_at": "2025-10-08T06:35:33.524Z",
    "metadata": {
        "udf1": "value1",
        "new_customer": "true",
        "login_date": "2019-09-10T10:11:12Z"
    },
    "default_payment_method_id": null,
    "tax_registration_id": null
}
```
3. Create and Confirm subscription
Request
```
curl --location 'http://localhost:8080/subscriptions' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'X-Profile-Id: pro_BSjmSsJE4oqVfDdiJViI' \
--header 'api-key: dev_4IM0uzCEoKQ9lkSaTUEceuErgHTSnZhcc0CsjiSc1P2jVv5W4p9nYumBWwnJLqLb' \
--header 'X-Tenant-Id: public' \
--data '{
    "item_price_id": "standard-plan-yearly",
    "customer_id": "cus_PqvEsNGKq1kLRdX7HTHJ",
    "description": "Hello this is description",
    "merchant_reference_id": "mer_ref_1759906164",
    "shipping": {
        "address": {
            "state": "zsaasdas",
            "city": "Banglore",
            "country": "US",
            "line1": "sdsdfsdf",
            "line2": "hsgdbhd",
            "line3": "alsksoe",
            "zip": "571201",
            "first_name": "joseph",
            "last_name": "doe"
        },
        "phone": {
            "number": "123456789",
            "country_code": "+1"
        }
    },
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "123456789",
            "country_code": "+1"
        }
    },
    "payment_details": {
        "payment_type": "setup_mandate",
        "payment_method": "card",
        "payment_method_type": "credit",
        "payment_method_data": {
            "card": {
                "card_number": "4111111111111111",
                "card_exp_month": "03",
                "card_exp_year": "2030",
                "card_holder_name": "CLBRW dffdg",
                "card_cvc": "737"
            }
        },
        "authentication_type": "no_three_ds",
        "setup_future_usage": "off_session",
        "capture_method": "automatic",
        "return_url": "https://google.com",
        "customer_acceptance": {
            "acceptance_type": "online",
            "accepted_at": "1963-05-03T04:07:52.723Z",
            "online": {
                "ip_address": "127.0.0.1",
                "user_agent": "amet irure esse"
            }
        }
    }
}'
```
Response
```
{
    "id": "sub_ZEqweuW4DlSc5BuM6zhW",
    "merchant_reference_id": "mer_ref_1759906160",
    "status": "trial",
    "plan_id": null,
    "price_id": null,
    "coupon": null,
    "profile_id": "pro_BSjmSsJE4oqVfDdiJViI",
    "payment": {
        "payment_id": "pay_1B9LjwEWwAfHb4bf0rcS",
        "status": "succeeded",
        "amount": 0,
        "currency": "USD",
        "connector": "stripe",
        "payment_method_id": "pm_kLqHMaoTQwDD5wRFj4mf",
        "payment_experience": null,
        "error_code": null,
        "error_message": null,
        "payment_method_type": "credit",
        "client_secret": "pay_1B9LjwEWwAfHb4bf0rcS_secret_a5IfTsNMNyh2hXHmmIEq",
        "payment_type": null
    },
    "customer_id": "cus_PqvEsNGKq1kLRdX7HTHJ",
    "invoice": {
        "id": "invoice_gO4K4XTvSoBJygwXBmsf",
        "subscription_id": "sub_ZEqweuW4DlSc5BuM6zhW",
        "merchant_id": "merchant_1759818834",
        "profile_id": "pro_BSjmSsJE4oqVfDdiJViI",
        "merchant_connector_id": "mca_4C0kfzkr1veBQWDVKnUj",
        "payment_intent_id": "pay_1B9LjwEWwAfHb4bf0rcS",
        "payment_method_id": null,
        "customer_id": "cus_PqvEsNGKq1kLRdX7HTHJ",
        "amount": 0,
        "currency": "USD",
        "status": "invoice_created"
    },
    "billing_processor_subscription_id": "sub_ZEqweuW4DlSc5BuM6zhW"
}
```
<img width="1309" height="107" alt="Screenshot 2025-10-08 at 12 20 19 PM" src="https://github.com/user-attachments/assets/feaf1e79-3ed4-441c-8401-767b0a76ac00" />
Time travel to the next billing date, an invoice will be generated with payment due statue and webhook will be received 
<img width="847" height="200" alt="Screenshot 2025-10-08 at 12 22 26 PM" src="https://github.com/user-attachments/assets/e5fe8953-a3a2-4773-a83c-060bb7385b94" />


4. Time travel to the next billing date and accept the incoming webhook for `invoice_generated` event, now the invoice should get marked as paid

Request
```
curl --location 'http://localhost:8080/webhooks/merchant_1759818834/mca_4C0kfzkr1veBQWDVKnUj' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'X-Profile-Id: pro_BSjmSsJE4oqVfDdiJViI' \
--header 'api-key: dev_4IM0uzCEoKQ9lkSaTUEceuErgHTSnZhcc0CsjiSc1P2jVv5W4p9nYumBWwnJLqLb' \
--header 'X-tenant-id: public' \
--header 'Authorization: Basic aHlwZXJzd2l0Y2g6aHlwZXJzd2l0Y2g=' \
--data '{
    "api_version": "v2",
    "content": {
        "invoice": {
            "adjustment_credit_notes": [],
            "amount_adjusted": 0,
            "amount_due": 12000,
            "amount_paid": 0,
            "amount_to_collect": 12000,
            "applied_credits": [],
            "base_currency_code": "INR",
            "billing_address": {
                "city": "San Fransico",
                "country": "US",
                "first_name": "joseph",
                "last_name": "Doe",
                "line1": "1467",
                "object": "billing_address",
                "state": "California",
                "state_code": "CA",
                "validation_status": "not_validated",
                "zip": "94122"
            },
            "channel": "web",
            "credits_applied": 0,
            "currency_code": "USD",
            "customer_id": "cus_PqvEsNGKq1kLRdX7HTHJ",
            "date": 1745303361,
            "deleted": false,
            "due_date": 1745303361,
            "dunning_attempts": [],
            "exchange_rate": 0.011742719,
            "first_invoice": true,
            "generated_at": 1745303361,
            "has_advance_charges": false,
            "id": "8",
            "is_gifted": false,
            "issued_credit_notes": [],
            "line_items": [
                {
                    "amount": 12000,
                    "customer_id": "cus_PqvEsNGKq1kLRdX7HTHJ",
                    "date_from": 1745303361,
                    "date_to": 1776839361,
                    "description": "Standard Plan - Yearly",
                    "discount_amount": 0,
                    "entity_id": "standard-plan-yearly",
                    "entity_type": "plan_item_price",
                    "id": "li_169lH1Uz11LJr2CNi",
                    "is_taxed": false,
                    "item_level_discount_amount": 0,
                    "object": "line_item",
                    "pricing_model": "flat_fee",
                    "quantity": 1,
                    "subscription_id": "sub_ZEqweuW4DlSc5BuM6zhW",
                    "tax_amount": 0,
                    "tax_exempt_reason": "tax_not_configured",
                    "unit_amount": 12000
                }
            ],
            "linked_orders": [],
            "linked_payments": [],
            "net_term_days": 0,
            "new_sales_amount": 12000,
            "object": "invoice",
            "price_type": "tax_exclusive",
            "recurring": true,
            "reference_transactions": [],
            "resource_version": 1745303362952,
            "round_off_amount": 0,
            "site_details_at_creation": {
                "timezone": "Asia/Calcutta"
            },
            "status": "payment_due",
            "sub_total": 12000,
            "subscription_id": "sub_ZEqweuW4DlSc5BuM6zhW",
            "tax": 0,
            "term_finalized": true,
            "total": 12000,
            "updated_at": 1745303362,
            "write_off_amount": 0
        }
    },
    "event_type": "invoice_generated",
    "id": "ev_169lH1Uz11LL82CNk",
    "object": "event",
    "occurred_at": 1745303363,
    "source": "scheduled_job",
    "webhook_status": "scheduled",
    "webhooks": [
        {
            "id": "whv2_169wFTUz0zZE6S87",
            "object": "webhook",
            "webhook_status": "scheduled"
        }
    ]
}'
```
Response
```
200 OK
```
<img width="820" height="153" alt="Screenshot 2025-10-08 at 2 49 12 PM" src="https://github.com/user-attachments/assets/fc43b38d-03eb-4015-b8d0-0bbc4e07e2bf" />


5. Create Customer and create subscription with auth_type: "three_ds" to create 3DS payment
Request
```
curl --location 'http://localhost:8080/subscriptions' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'X-Profile-Id: pro_Z5PRkNfUsNYHHGgKDmSc' \
--header 'api-key: dev_TVnuWX7gAfpjrVv4Ex5OHjwTZ7lUKY5LNy8MQnvnSbj19ys5nbEDME4kejTBZ0tg' \
--header 'X-tenant-id: public' \
--data '{
    "item_price_id": "standard-plan-USD-Monthly",
    "customer_id": "cus_Pp7MTQL9p1ovyoKB2QtS",
    "description": "Hello this is description",
    "merchant_reference_id": "mer_ref_1760016459",
    "shipping": {
        "address": {
            "state": "zsaasdas",
            "city": "Banglore",
            "country": "US",
            "line1": "sdsdfsdf",
            "line2": "hsgdbhd",
            "line3": "alsksoe",
            "zip": "571201",
            "first_name": "joseph",
            "last_name": "doe"
        },
        "phone": {
            "number": "123456789",
            "country_code": "+1"
        }
    },
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "123456789",
            "country_code": "+1"
        }
    },
    "payment_details": {
        "payment_type": "setup_mandate",
        "payment_method": "card",
        "payment_method_type": "credit",
        "payment_method_data": {
            "card": {
                "card_number": "4111111111111111",
                "card_exp_month": "03",
                "card_exp_year": "2030",
                "card_holder_name": "CLBRW dffdg",
                "card_cvc": "737"
            }
        },
        "authentication_type": "three_ds",
        "setup_future_usage": "off_session",
        "capture_method": "automatic",
        "return_url": "https://google.com",
        "customer_acceptance": {
            "acceptance_type": "online",
            "accepted_at": "1963-05-03T04:07:52.723Z",
            "online": {
                "ip_address": "127.0.0.1",
                "user_agent": "amet irure esse"
            }
        }
    }
}'
```
Response
```
{
    "id": "sub_XSgS745pisVSvcXEWxiE",
    "merchant_reference_id": "mer_ref_1760016450",
    "status": "active",
    "plan_id": null,
    "price_id": null,
    "coupon": null,
    "profile_id": "pro_Z5PRkNfUsNYHHGgKDmSc",
    "payment": {
        "payment_id": "pay_pwKrT9HW6vmSukp56Oo0",
        "status": "requires_customer_action",
        "amount": 1200,
        "currency": "USD",
        "profile_id": "pro_Z5PRkNfUsNYHHGgKDmSc",
        "connector": "stripe",
        "payment_method_id": "pm_2i4KXYKijSeX0u4PymkJ",
        "return_url": "https://google.com/",
        "next_action": {
            "type": "redirect_to_url",
            "redirect_to_url": "http://localhost:8080/payments/redirect/pay_pwKrT9HW6vmSukp56Oo0/merchant_1760009370/pay_pwKrT9HW6vmSukp56Oo0_1"
        },
        "payment_experience": null,
        "error_code": null,
        "error_message": null,
        "payment_method_type": "credit",
        "client_secret": "pay_pwKrT9HW6vmSukp56Oo0_secret_Z8ZehHMFc45D1KliHX0g",
        "payment_type": null
    },
    "customer_id": "cus_Pp7MTQL9p1ovyoKB2QtS",
    "invoice": {
        "id": "invoice_iVKvCnVwg9aotKN2tcwn",
        "subscription_id": "sub_XSgS745pisVSvcXEWxiE",
        "merchant_id": "merchant_1760009370",
        "profile_id": "pro_Z5PRkNfUsNYHHGgKDmSc",
        "merchant_connector_id": "mca_hbJ2wahF6dp7ACUW72eO",
        "payment_intent_id": "pay_pwKrT9HW6vmSukp56Oo0",
        "payment_method_id": null,
        "customer_id": "cus_Pp7MTQL9p1ovyoKB2QtS",
        "amount": 1200,
        "currency": "USD",
        "status": "payment_pending"
    },
    "billing_processor_subscription_id": "sub_XSgS745pisVSvcXEWxiE"
}
```
<img width="1195" height="542" alt="Screenshot 2025-10-09 at 6 59 39 PM" src="https://github.com/user-attachments/assets/24457ec6-f816-4956-8d88-b59302f9e654" />

Created Unique index on subscription_id_connector_invoice_id in invoice table to avoid multiple MIT payments

```
➜  ~ seq 1 4 | xargs -n1 -P5 curl -w "\nStatus: %{http_code}\n" --location 'http://localhost.proxyman.io:8080/webhooks/merchant_1760030871/mca_Qxsaj3wPHpCw2BfVLfcp' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'X-Profile-Id: pro_fRxGSwMSfrfb19ttnUJt' \
--header 'api-key: dev_oioLdMN4eQuEHr6e9H3WWllhWgJMwkhYhZgiEP2nc2whZ7tPtC8xx4pCHKt3d8C7' \
--header 'X-tenant-id: public' \
--header 'Authorization: Basic aHlwZXJzd2l0Y2g6aHlwZXJzd2l0Y2g=' \
--data '{"api_version":"v2","content":{"invoice":{"adjustment_credit_notes":[],"amount_adjusted":0,"amount_due":12000,"amount_paid":0,"amount_to_collect":12000,"applied_credits":[],"base_currency_code":"INR","billing_address":{"city":"San Fransico","country":"US","first_name":"joseph","last_name":"Doe","line1":"1467","object":"billing_address","state":"California","state_code":"CA","validation_status":"not_validated","zip":"94122"},"channel":"web","credits_applied":0,"currency_code":"USD","customer_id":"cus_4dPaSdR6AvhvVRGsw2j3","date":1745428099,"deleted":false,"due_date":1745428099,"dunning_attempts":[],"exchange_rate":0.011732607,"first_invoice":true,"generated_at":1745428099,"has_advance_charges":false,"id":"30","is_gifted":false,"issued_credit_notes":[],"line_items":[{"amount":12000,"customer_id":"cus_4dPaSdR6AvhvVRGsw2j3","date_from":1745428099,"date_to":1776964099,"description":"Standard Plan - Yearly","discount_amount":0,"entity_id":"standard-plan-yearly","entity_type":"plan_item_price","id":"li_AzZtuxUz9SYOs8pX8","is_taxed":false,"item_level_discount_amount":0,"object":"line_item","pricing_model":"flat_fee","quantity":1,"subscription_id":"sub_Q1awfb6GdSXDf3QXcmQr","tax_amount":0,"tax_exempt_reason":"tax_not_configured","unit_amount":12000}],"linked_orders":[],"linked_payments":[],"net_term_days":0,"new_sales_amount":12000,"object":"invoice","price_type":"tax_exclusive","recurring":true,"reference_transactions":[],"resource_version":1745428100772,"round_off_amount":0,"site_details_at_creation":{"timezone":"Asia/Calcutta"},"status":"payment_due","sub_total":12000,"subscription_id":"sub_Q1awfb6GdSXDf3QXcmQr","tax":0,"term_finalized":true,"total":12000,"updated_at":1745428100,"write_off_amount":0}},"event_type":"invoice_generated","id":"ev_AzZtuxUz9SYQ08pXA","object":"event","occurred_at":1745428100,"source":"scheduled_job","webhook_status":"scheduled","webhooks":[{"id":"whv2_169wFTUz0zZE6S87","object":"webhook","webhook_status":"scheduled"}]}'
{"error":{"type":"invalid_request","message":"Subscription operation: Create Invoice failed with connector","code":"CE_09"}}{"error":{"type":"invalid_request","message":"Subscription operation: Create Invoice failed with connector","code":"CE_09"}}{"error":{"type":"invalid_request","message":"Subscription operation: Create Invoice failed with connector","code":"CE_09"}}


Status: 400
Status: 400
Status: 400
curl: (7) Failed to connect to 0.0.0.1 port 80 after 0 ms: Couldn't connect to server

Status: 000
curl: (7) Failed to connect to 0.0.0.3 port 80 after 0 ms: Couldn't connect to server

Status: 000
curl: (7) Failed to connect to 0.0.0.4 port 80 after 0 ms: Couldn't connect to server

Status: 000

Status: 200
curl: (7) Failed to connect to 0.0.0.2 port 80 after 0 ms: Couldn't connect to server

Status: 000
```
Here there is only one 200 and rest of the parallel request are 400



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

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