Bug: Subscription confirm API




diff --git a/crates/api_models/src/subscription.rs b/crates/api_models/src/subscription.rs
index 7dff21205a2..4ed14a9c64e 100644
--- a/crates/api_models/src/subscription.rs
+++ b/crates/api_models/src/subscription.rs
@@ -1,7 +1,13 @@
-use common_utils::events::ApiEventMetric;
+use common_types::payments::CustomerAcceptance;
+use common_utils::{errors::ValidationError, events::ApiEventMetric, types::MinorUnit};
 use masking::Secret;
 use utoipa::ToSchema;
 
+use crate::{
+    enums as api_enums,
+    payments::{Address, PaymentMethodDataRequest},
+};
+
 // use crate::{
 //     customers::{CustomerRequest, CustomerResponse},
 //     payments::CustomerDetailsResponse,
@@ -63,7 +69,14 @@ pub struct CreateSubscriptionResponse {
 ///
 /// - `Created`: Subscription was created but not yet activated.
 /// - `Active`: Subscription is currently active.
-/// - `InActive`: Subscription is inactive (e.g., cancelled or expired).
+/// - `InActive`: Subscription is inactive.
+/// - `Pending`: Subscription is pending activation.
+/// - `Trial`: Subscription is in a trial period.
+/// - `Paused`: Subscription is paused.
+/// - `Unpaid`: Subscription is unpaid.
+/// - `Onetime`: Subscription is a one-time payment.
+/// - `Cancelled`: Subscription has been cancelled.
+/// - `Failed`: Subscription has failed.
 #[derive(Debug, Clone, serde::Serialize, strum::EnumString, strum::Display, ToSchema)]
 pub enum SubscriptionStatus {
     /// Subscription is active.
@@ -74,6 +87,18 @@ pub enum SubscriptionStatus {
     InActive,
     /// Subscription is in pending state.
     Pending,
+    /// Subscription is in trial state.
+    Trial,
+    /// Subscription is paused.
+    Paused,
+    /// Subscription is unpaid.
+    Unpaid,
+    /// Subscription is a one-time payment.
+    Onetime,
+    /// Subscription is cancelled.
+    Cancelled,
+    /// Subscription has failed.
+    Failed,
 }
 
 impl CreateSubscriptionResponse {
@@ -107,3 +132,141 @@ impl CreateSubscriptionResponse {
 
 impl ApiEventMetric for CreateSubscriptionResponse {}
 impl ApiEventMetric for CreateSubscriptionRequest {}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+pub struct PaymentDetails {
+    pub payment_method: api_enums::PaymentMethod,
+    pub payment_method_type: Option<api_enums::PaymentMethodType>,
+    pub payment_method_data: PaymentMethodDataRequest,
+    pub setup_future_usage: Option<api_enums::FutureUsage>,
+    pub customer_acceptance: Option<CustomerAcceptance>,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+pub struct PaymentResponseData {
+    pub payment_id: common_utils::id_type::PaymentId,
+    pub status: api_enums::IntentStatus,
+    pub amount: MinorUnit,
+    pub currency: api_enums::Currency,
+    pub connector: Option<String>,
+}
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+pub struct ConfirmSubscriptionRequest {
+    /// Client secret for SDK based interaction.
+    pub client_secret: Option<String>,
+
+    /// Amount to be charged for the invoice.
+    pub amount: MinorUnit,
+
+    /// Currency for the amount.
+    pub currency: api_enums::Currency,
+
+    /// Identifier for the associated plan_id.
+    pub plan_id: Option<String>,
+
+    /// Identifier for the associated item_price_id for the subscription.
+    pub item_price_id: Option<String>,
+
+    /// Idenctifier for the coupon code for the subscription.
+    pub coupon_code: Option<String>,
+
+    /// Identifier for customer.
+    pub customer_id: common_utils::id_type::CustomerId,
+
+    /// Billing address for the subscription.
+    pub billing_address: Option<Address>,
+
+    /// Payment details for the invoice.
+    pub payment_details: PaymentDetails,
+}
+
+impl ConfirmSubscriptionRequest {
+    pub fn get_item_price_id(&self) -> Result<String, error_stack::Report<ValidationError>> {
+        self.item_price_id.clone().ok_or(error_stack::report!(
+            ValidationError::MissingRequiredField {
+                field_name: "item_price_id".to_string()
+            }
+        ))
+    }
+
+    pub fn get_billing_address(&self) -> Result<Address, error_stack::Report<ValidationError>> {
+        self.billing_address.clone().ok_or(error_stack::report!(
+            ValidationError::MissingRequiredField {
+                field_name: "billing_address".to_string()
+            }
+        ))
+    }
+}
+
+impl ApiEventMetric for ConfirmSubscriptionRequest {}
+
+#[derive(Debug, Clone, serde::Serialize, ToSchema)]
+pub struct ConfirmSubscriptionResponse {
+    /// Unique identifier for the subscription.
+    pub id: common_utils::id_type::SubscriptionId,
+
+    /// Merchant specific Unique identifier.
+    pub merchant_reference_id: Option<String>,
+
+    /// Current status of the subscription.
+    pub status: SubscriptionStatus,
+
+    /// Identifier for the associated subscription plan.
+    pub plan_id: Option<String>,
+
+    /// Identifier for the associated item_price_id for the subscription.
+    pub price_id: Option<String>,
+
+    /// Optional coupon code applied to this subscription.
+    pub coupon: Option<String>,
+
+    /// Associated profile ID.
+    pub profile_id: common_utils::id_type::ProfileId,
+
+    /// Payment details for the invoice.
+    pub payment: Option<PaymentResponseData>,
+
+    /// Customer ID associated with this subscription.
+    pub customer_id: Option<common_utils::id_type::CustomerId>,
+
+    /// Invoice Details for the subscription.
+    pub invoice: Option<Invoice>,
+}
+
+#[derive(Debug, Clone, serde::Serialize, ToSchema)]
+pub struct Invoice {
+    /// Unique identifier for the invoice.
+    pub id: common_utils::id_type::InvoiceId,
+
+    /// Unique identifier for the subscription.
+    pub subscription_id: common_utils::id_type::SubscriptionId,
+
+    /// Identifier for the merchant.
+    pub merchant_id: common_utils::id_type::MerchantId,
+
+    /// Identifier for the profile.
+    pub profile_id: common_utils::id_type::ProfileId,
+
+    /// Identifier for the merchant connector account.
+    pub merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
+
+    /// Identifier for the Payment.
+    pub payment_intent_id: Option<common_utils::id_type::PaymentId>,
+
+    /// Identifier for the Payment method.
+    pub payment_method_id: Option<String>,
+
+    /// Identifier for the Customer.
+    pub customer_id: common_utils::id_type::CustomerId,
+
+    /// Invoice amount.
+    pub amount: MinorUnit,
+
+    /// Currency for the invoice payment.
+    pub currency: api_enums::Currency,
+
+    /// Status of the invoice.
+    pub status: String,
+}
+
+impl ApiEventMetric for ConfirmSubscriptionResponse {}
diff --git a/crates/diesel_models/src/invoice.rs b/crates/diesel_models/src/invoice.rs
index 57024d0730a..8ffd54c288d 100644
--- a/crates/diesel_models/src/invoice.rs
+++ b/crates/diesel_models/src/invoice.rs
@@ -34,21 +34,21 @@ pub struct InvoiceNew {
     check_for_backend(diesel::pg::Pg)
 )]
 pub struct Invoice {
-    id: common_utils::id_type::InvoiceId,
-    subscription_id: common_utils::id_type::SubscriptionId,
-    merchant_id: common_utils::id_type::MerchantId,
-    profile_id: common_utils::id_type::ProfileId,
-    merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
-    payment_intent_id: Option<common_utils::id_type::PaymentId>,
-    payment_method_id: Option<String>,
-    customer_id: common_utils::id_type::CustomerId,
-    amount: MinorUnit,
-    currency: String,
-    status: String,
-    provider_name: Connector,
-    metadata: Option<SecretSerdeValue>,
-    created_at: time::PrimitiveDateTime,
-    modified_at: time::PrimitiveDateTime,
+    pub id: common_utils::id_type::InvoiceId,
+    pub subscription_id: common_utils::id_type::SubscriptionId,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub profile_id: common_utils::id_type::ProfileId,
+    pub merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
+    pub payment_intent_id: Option<common_utils::id_type::PaymentId>,
+    pub payment_method_id: Option<String>,
+    pub customer_id: common_utils::id_type::CustomerId,
+    pub amount: MinorUnit,
+    pub currency: String,
+    pub status: String,
+    pub provider_name: Connector,
+    pub metadata: Option<SecretSerdeValue>,
+    pub created_at: time::PrimitiveDateTime,
+    pub modified_at: time::PrimitiveDateTime,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq, AsChangeset, Deserialize)]
diff --git a/crates/hyperswitch_connectors/src/connectors/chargebee.rs b/crates/hyperswitch_connectors/src/connectors/chargebee.rs
index b761dc9ebbf..167c2a337ba 100644
--- a/crates/hyperswitch_connectors/src/connectors/chargebee.rs
+++ b/crates/hyperswitch_connectors/src/connectors/chargebee.rs
@@ -16,7 +16,7 @@ use error_stack::ResultExt;
 use hyperswitch_domain_models::{revenue_recovery, router_data_v2::RouterDataV2};
 use hyperswitch_domain_models::{
     router_data::{AccessToken, ConnectorAuthType, ErrorResponse, RouterData},
-    router_data_v2::flow_common_types::SubscriptionCreateData,
+    router_data_v2::flow_common_types::{SubscriptionCreateData, SubscriptionCustomerData},
     router_flow_types::{
         access_token_auth::AccessTokenAuth,
         payments::{Authorize, Capture, PSync, PaymentMethodToken, Session, SetupMandate, Void},
@@ -131,13 +131,28 @@ impl ConnectorIntegration<SubscriptionCreate, SubscriptionCreateRequest, Subscri
     ) -> CustomResult<String, errors::ConnectorError> {
         let metadata: chargebee::ChargebeeMetadata =
             utils::to_connector_meta_from_secret(req.connector_meta_data.clone())?;
-        let url = self
-            .base_url(connectors)
-            .to_string()
-            .replace("{{merchant_endpoint_prefix}}", metadata.site.peek());
+
+        let site = metadata.site.peek();
+
+        let mut base = self.base_url(connectors).to_string();
+
+        base = base.replace("{{merchant_endpoint_prefix}}", site);
+        base = base.replace("$", site);
+
+        if base.contains("{{merchant_endpoint_prefix}}") || base.contains('$') {
+            return Err(errors::ConnectorError::InvalidConnectorConfig {
+                config: "Chargebee base_url has an unresolved placeholder (expected `$` or `{{merchant_endpoint_prefix}}`).",
+            }
+            .into());
+        }
+
+        if !base.ends_with('/') {
+            base.push('/');
+        }
+
         let customer_id = &req.request.customer_id.get_string_repr().to_string();
         Ok(format!(
-            "{url}v2/customers/{customer_id}/subscription_for_items"
+            "{base}v2/customers/{customer_id}/subscription_for_items"
         ))
     }
 
@@ -217,6 +232,17 @@ impl
     // Not Implemented (R)
 }
 
+impl
+    ConnectorIntegrationV2<
+        CreateConnectorCustomer,
+        SubscriptionCustomerData,
+        ConnectorCustomerData,
+        PaymentsResponseData,
+    > for Chargebee
+{
+    // Not Implemented (R)
+}
+
 impl ConnectorIntegration<PaymentMethodToken, PaymentMethodTokenizationData, PaymentsResponseData>
     for Chargebee
 {
diff --git a/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs b/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
index 572cb082a45..953dfdca788 100644
--- a/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
+++ b/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
@@ -1039,7 +1039,20 @@ pub struct ChargebeeCustomerCreateRequest {
     #[serde(rename = "first_name")]
     pub name: Option<Secret<String>>,
     pub email: Option<Email>,
-    pub billing_address: Option<api_models::payments::AddressDetails>,
+    #[serde(rename = "billing_address[first_name]")]
+    pub billing_address_first_name: Option<Secret<String>>,
+    #[serde(rename = "billing_address[last_name]")]
+    pub billing_address_last_name: Option<Secret<String>>,
+    #[serde(rename = "billing_address[line1]")]
+    pub billing_address_line1: Option<Secret<String>>,
+    #[serde(rename = "billing_address[city]")]
+    pub billing_address_city: Option<String>,
+    #[serde(rename = "billing_address[state]")]
+    pub billing_address_state: Option<Secret<String>>,
+    #[serde(rename = "billing_address[zip]")]
+    pub billing_address_zip: Option<Secret<String>>,
+    #[serde(rename = "billing_address[country]")]
+    pub billing_address_country: Option<String>,
 }
 
 impl TryFrom<&ChargebeeRouterData<&hyperswitch_domain_models::types::ConnectorCustomerRouterData>>
@@ -1062,7 +1075,34 @@ impl TryFrom<&ChargebeeRouterData<&hyperswitch_domain_models::types::ConnectorCu
                 .clone(),
             name: req.name.clone(),
             email: req.email.clone(),
-            billing_address: req.billing_address.clone(),
+            billing_address_first_name: req
+                .billing_address
+                .as_ref()
+                .and_then(|address| address.first_name.clone()),
+            billing_address_last_name: req
+                .billing_address
+                .as_ref()
+                .and_then(|address| address.last_name.clone()),
+            billing_address_line1: req
+                .billing_address
+                .as_ref()
+                .and_then(|addr| addr.line1.clone()),
+            billing_address_city: req
+                .billing_address
+                .as_ref()
+                .and_then(|addr| addr.city.clone()),
+            billing_address_country: req
+                .billing_address
+                .as_ref()
+                .and_then(|addr| addr.country.map(|country| country.to_string())),
+            billing_address_state: req
+                .billing_address
+                .as_ref()
+                .and_then(|addr| addr.state.clone()),
+            billing_address_zip: req
+                .billing_address
+                .as_ref()
+                .and_then(|addr| addr.zip.clone()),
         })
     }
 }
diff --git a/crates/hyperswitch_connectors/src/connectors/recurly.rs b/crates/hyperswitch_connectors/src/connectors/recurly.rs
index 738a7026c8d..bfa9a50a99a 100644
--- a/crates/hyperswitch_connectors/src/connectors/recurly.rs
+++ b/crates/hyperswitch_connectors/src/connectors/recurly.rs
@@ -12,7 +12,7 @@ use hyperswitch_domain_models::{
     router_data_v2::{
         flow_common_types::{
             GetSubscriptionEstimateData, GetSubscriptionPlanPricesData, GetSubscriptionPlansData,
-            SubscriptionCreateData,
+            SubscriptionCreateData, SubscriptionCustomerData,
         },
         UasFlowData,
     },
@@ -24,6 +24,7 @@ use hyperswitch_domain_models::{
         unified_authentication_service::{
             Authenticate, AuthenticationConfirmation, PostAuthenticate, PreAuthenticate,
         },
+        CreateConnectorCustomer,
     },
     router_request_types::{
         subscriptions::{
@@ -35,10 +36,14 @@ use hyperswitch_domain_models::{
             UasConfirmationRequestData, UasPostAuthenticationRequestData,
             UasPreAuthenticationRequestData,
         },
+        ConnectorCustomerData,
     },
-    router_response_types::subscriptions::{
-        GetSubscriptionEstimateResponse, GetSubscriptionPlanPricesResponse,
-        GetSubscriptionPlansResponse, SubscriptionCreateResponse,
+    router_response_types::{
+        subscriptions::{
+            GetSubscriptionEstimateResponse, GetSubscriptionPlanPricesResponse,
+            GetSubscriptionPlansResponse, SubscriptionCreateResponse,
+        },
+        PaymentsResponseData,
     },
 };
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
@@ -156,6 +161,7 @@ impl
 impl api::revenue_recovery_v2::RevenueRecoveryV2 for Recurly {}
 impl api::subscriptions_v2::SubscriptionsV2 for Recurly {}
 impl api::subscriptions_v2::GetSubscriptionPlansV2 for Recurly {}
+impl api::subscriptions_v2::SubscriptionConnectorCustomerV2 for Recurly {}
 
 impl
     ConnectorIntegrationV2<
@@ -167,6 +173,16 @@ impl
 {
 }
 
+impl
+    ConnectorIntegrationV2<
+        CreateConnectorCustomer,
+        SubscriptionCustomerData,
+        ConnectorCustomerData,
+        PaymentsResponseData,
+    > for Recurly
+{
+}
+
 impl api::subscriptions_v2::GetSubscriptionPlanPricesV2 for Recurly {}
 
 impl
diff --git a/crates/hyperswitch_domain_models/src/business_profile.rs b/crates/hyperswitch_domain_models/src/business_profile.rs
index 095e2a392c5..42c2398582c 100644
--- a/crates/hyperswitch_domain_models/src/business_profile.rs
+++ b/crates/hyperswitch_domain_models/src/business_profile.rs
@@ -1547,6 +1547,21 @@ impl Profile {
                 Cow::Borrowed(common_types::consts::DEFAULT_PAYOUT_WEBHOOK_TRIGGER_STATUSES)
             })
     }
+
+    pub fn get_billing_processor_id(
+        &self,
+    ) -> CustomResult<
+        common_utils::id_type::MerchantConnectorAccountId,
+        api_error_response::ApiErrorResponse,
+    > {
+        self.billing_processor_id
+            .to_owned()
+            .ok_or(error_stack::report!(
+                api_error_response::ApiErrorResponse::MissingRequiredField {
+                    field_name: "billing_processor_id"
+                }
+            ))
+    }
 }
 
 #[cfg(feature = "v2")]
diff --git a/crates/hyperswitch_domain_models/src/errors/api_error_response.rs b/crates/hyperswitch_domain_models/src/errors/api_error_response.rs
index 830b90fc763..f096ba78632 100644
--- a/crates/hyperswitch_domain_models/src/errors/api_error_response.rs
+++ b/crates/hyperswitch_domain_models/src/errors/api_error_response.rs
@@ -321,6 +321,8 @@ pub enum ApiErrorResponse {
     },
     #[error(error_type = ErrorType::ObjectNotFound, code = "HE_02", message = "Tokenization record not found for the given token_id {id}")]
     TokenizationRecordNotFound { id: String },
+    #[error(error_type = ErrorType::ConnectorError, code = "CE_00", message = "Subscription operation: {operation} failed with connector")]
+    SubscriptionError { operation: String },
 }
 
 #[derive(Clone)]
@@ -710,6 +712,9 @@ impl ErrorSwitch<api_models::errors::types::ApiErrorResponse> for ApiErrorRespon
             Self::TokenizationRecordNotFound{ id } => {
                 AER::NotFound(ApiError::new("HE", 2, format!("Tokenization record not found for the given token_id '{id}' "), None))
             }
+            Self::SubscriptionError { operation } => {
+                AER::BadRequest(ApiError::new("CE", 9, format!("Subscription operation: {operation} failed with connector"), None))
+            }
         }
     }
 }
diff --git a/crates/hyperswitch_domain_models/src/router_data_v2/flow_common_types.rs b/crates/hyperswitch_domain_models/src/router_data_v2/flow_common_types.rs
index 40be24722a5..245d57064b9 100644
--- a/crates/hyperswitch_domain_models/src/router_data_v2/flow_common_types.rs
+++ b/crates/hyperswitch_domain_models/src/router_data_v2/flow_common_types.rs
@@ -151,13 +151,24 @@ pub struct FilesFlowData {
 pub struct InvoiceRecordBackData;
 
 #[derive(Debug, Clone)]
-pub struct SubscriptionCreateData;
+pub struct SubscriptionCustomerData {
+    pub connector_meta_data: Option<pii::SecretSerdeValue>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionCreateData {
+    pub connector_meta_data: Option<pii::SecretSerdeValue>,
+}
 
 #[derive(Debug, Clone)]
-pub struct GetSubscriptionPlansData;
+pub struct GetSubscriptionPlansData {
+    pub connector_meta_data: Option<pii::SecretSerdeValue>,
+}
 
 #[derive(Debug, Clone)]
-pub struct GetSubscriptionPlanPricesData;
+pub struct GetSubscriptionPlanPricesData {
+    pub connector_meta_data: Option<pii::SecretSerdeValue>,
+}
 
 #[derive(Debug, Clone)]
 pub struct GetSubscriptionEstimateData;
diff --git a/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs b/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
index f480765effa..73b80b82075 100644
--- a/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
+++ b/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
@@ -13,7 +13,7 @@ pub struct SubscriptionCreateResponse {
     pub created_at: Option<PrimitiveDateTime>,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Copy)]
 pub enum SubscriptionStatus {
     Pending,
     Trial,
@@ -25,6 +25,22 @@ pub enum SubscriptionStatus {
     Failed,
 }
 
+#[cfg(feature = "v1")]
+impl From<SubscriptionStatus> for api_models::subscription::SubscriptionStatus {
+    fn from(status: SubscriptionStatus) -> Self {
+        match status {
+            SubscriptionStatus::Pending => Self::Pending,
+            SubscriptionStatus::Trial => Self::Trial,
+            SubscriptionStatus::Active => Self::Active,
+            SubscriptionStatus::Paused => Self::Paused,
+            SubscriptionStatus::Unpaid => Self::Unpaid,
+            SubscriptionStatus::Onetime => Self::Onetime,
+            SubscriptionStatus::Cancelled => Self::Cancelled,
+            SubscriptionStatus::Failed => Self::Failed,
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct GetSubscriptionPlansResponse {
     pub list: Vec<SubscriptionPlans>,
diff --git a/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs b/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs
index 09f25918c56..47a8b4484a1 100644
--- a/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs
+++ b/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs
@@ -2,30 +2,35 @@
 use hyperswitch_domain_models::{
     router_data_v2::flow_common_types::{
         GetSubscriptionEstimateData, GetSubscriptionPlanPricesData, GetSubscriptionPlansData,
-        SubscriptionCreateData,
+        SubscriptionCreateData, SubscriptionCustomerData,
     },
-    router_flow_types::subscriptions::{
-        GetSubscriptionEstimate, GetSubscriptionPlanPrices, GetSubscriptionPlans,
-        SubscriptionCreate,
+    router_flow_types::{
+        subscriptions::{GetSubscriptionPlanPrices, GetSubscriptionPlans, SubscriptionCreate},
+        CreateConnectorCustomer, GetSubscriptionEstimate,
     },
-    router_request_types::subscriptions::{
-        GetSubscriptionEstimateRequest, GetSubscriptionPlanPricesRequest,
-        GetSubscriptionPlansRequest, SubscriptionCreateRequest,
+    router_request_types::{
+        subscriptions::{
+            GetSubscriptionEstimateRequest, GetSubscriptionPlanPricesRequest,
+            GetSubscriptionPlansRequest, SubscriptionCreateRequest,
+        },
+        ConnectorCustomerData,
     },
-    router_response_types::subscriptions::{
-        GetSubscriptionEstimateResponse, GetSubscriptionPlanPricesResponse,
-        GetSubscriptionPlansResponse, SubscriptionCreateResponse,
+    router_response_types::{
+        subscriptions::{
+            GetSubscriptionEstimateResponse, GetSubscriptionPlanPricesResponse,
+            GetSubscriptionPlansResponse, SubscriptionCreateResponse,
+        },
+        PaymentsResponseData,
     },
 };
 
-use super::payments_v2::ConnectorCustomerV2;
 use crate::connector_integration_v2::ConnectorIntegrationV2;
 
 /// trait SubscriptionsV2
 pub trait SubscriptionsV2:
     GetSubscriptionPlansV2
     + SubscriptionsCreateV2
-    + ConnectorCustomerV2
+    + SubscriptionConnectorCustomerV2
     + GetSubscriptionPlanPricesV2
     + GetSubscriptionEstimateV2
 {
@@ -64,6 +69,16 @@ pub trait SubscriptionsCreateV2:
 {
 }
 
+/// trait SubscriptionConnectorCustomerV2
+pub trait SubscriptionConnectorCustomerV2:
+    ConnectorIntegrationV2<
+    CreateConnectorCustomer,
+    SubscriptionCustomerData,
+    ConnectorCustomerData,
+    PaymentsResponseData,
+>
+{
+}
 /// trait GetSubscriptionEstimate for V2
 pub trait GetSubscriptionEstimateV2:
     ConnectorIntegrationV2<
diff --git a/crates/hyperswitch_interfaces/src/conversion_impls.rs b/crates/hyperswitch_interfaces/src/conversion_impls.rs
index 3d3219a33f8..ed83c8235f1 100644
--- a/crates/hyperswitch_interfaces/src/conversion_impls.rs
+++ b/crates/hyperswitch_interfaces/src/conversion_impls.rs
@@ -14,7 +14,7 @@ use hyperswitch_domain_models::{
             ExternalVaultProxyFlowData, FilesFlowData, GetSubscriptionPlanPricesData,
             GetSubscriptionPlansData, GiftCardBalanceCheckFlowData, InvoiceRecordBackData,
             MandateRevokeFlowData, PaymentFlowData, RefundFlowData, SubscriptionCreateData,
-            UasFlowData, VaultConnectorFlowData, WebhookSourceVerifyData,
+            SubscriptionCustomerData, UasFlowData, VaultConnectorFlowData, WebhookSourceVerifyData,
         },
         RouterDataV2,
     },
@@ -846,7 +846,9 @@ macro_rules! default_router_data_conversion {
             where
                 Self: Sized,
             {
-                let resource_common_data = Self {};
+                let resource_common_data = Self {
+                    connector_meta_data: old_router_data.connector_meta_data.clone(),
+                };
                 Ok(RouterDataV2 {
                     flow: std::marker::PhantomData,
                     tenant_id: old_router_data.tenant_id.clone(),
@@ -863,16 +865,19 @@ macro_rules! default_router_data_conversion {
             where
                 Self: Sized,
             {
-                let router_data = get_default_router_data(
+                let Self {
+                    connector_meta_data,
+                } = new_router_data.resource_common_data;
+                let mut router_data = get_default_router_data(
                     new_router_data.tenant_id.clone(),
                     stringify!($flow_name),
                     new_router_data.request,
                     new_router_data.response,
                 );
-                Ok(RouterData {
-                    connector_auth_type: new_router_data.connector_auth_type.clone(),
-                    ..router_data
-                })
+                router_data.connector_meta_data = connector_meta_data;
+                router_data.connector_auth_type = new_router_data.connector_auth_type;
+
+                Ok(router_data)
             }
         }
     };
@@ -880,6 +885,7 @@ macro_rules! default_router_data_conversion {
 default_router_data_conversion!(GetSubscriptionPlansData);
 default_router_data_conversion!(GetSubscriptionPlanPricesData);
 default_router_data_conversion!(SubscriptionCreateData);
+default_router_data_conversion!(SubscriptionCustomerData);
 
 impl<T, Req: Clone, Resp: Clone> RouterDataConversion<T, Req, Resp> for UasFlowData {
     fn from_old_router_data(
diff --git a/crates/router/src/compatibility/stripe/errors.rs b/crates/router/src/compatibility/stripe/errors.rs
index a30f8d4c565..a0bec1a5a1b 100644
--- a/crates/router/src/compatibility/stripe/errors.rs
+++ b/crates/router/src/compatibility/stripe/errors.rs
@@ -285,6 +285,8 @@ pub enum StripeErrorCode {
     PlatformUnauthorizedRequest,
     #[error(error_type = StripeErrorType::HyperswitchError, code = "", message = "Profile Acquirer not found")]
     ProfileAcquirerNotFound,
+    #[error(error_type = StripeErrorType::HyperswitchError, code = "Subscription Error", message = "Subscription operation: {operation} failed with connector")]
+    SubscriptionError { operation: String },
     // [#216]: https://github.com/juspay/hyperswitch/issues/216
     // Implement the remaining stripe error codes
 
@@ -697,6 +699,9 @@ impl From<errors::ApiErrorResponse> for StripeErrorCode {
                 object: "tokenization record".to_owned(),
                 id,
             },
+            errors::ApiErrorResponse::SubscriptionError { operation } => {
+                Self::SubscriptionError { operation }
+            }
         }
     }
 }
@@ -781,7 +786,8 @@ impl actix_web::ResponseError for StripeErrorCode {
             | Self::WebhookProcessingError
             | Self::InvalidTenant
             | Self::ExternalVaultFailed
-            | Self::AmountConversionFailed { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            | Self::AmountConversionFailed { .. }
+            | Self::SubscriptionError { .. } => StatusCode::INTERNAL_SERVER_ERROR,
             Self::ReturnUrlUnavailable => StatusCode::SERVICE_UNAVAILABLE,
             Self::ExternalConnectorError { status_code, .. } => {
                 StatusCode::from_u16(*status_code).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR)
diff --git a/crates/router/src/core/subscription.rs b/crates/router/src/core/subscription.rs
index 32c6754e2fe..b5c760ad3ec 100644
--- a/crates/router/src/core/subscription.rs
+++ b/crates/router/src/core/subscription.rs
@@ -1,16 +1,31 @@
 use std::str::FromStr;
 
-use api_models::subscription::{
-    self as subscription_types, CreateSubscriptionResponse, SubscriptionStatus,
+use api_models::{
+    enums as api_enums,
+    subscription::{self as subscription_types, CreateSubscriptionResponse, SubscriptionStatus},
 };
-use common_utils::id_type::GenerateId;
+use common_utils::{ext_traits::ValueExt, id_type::GenerateId, pii};
 use diesel_models::subscription::SubscriptionNew;
 use error_stack::ResultExt;
-use hyperswitch_domain_models::{api::ApplicationResponse, merchant_context::MerchantContext};
+use hyperswitch_domain_models::{
+    api::ApplicationResponse,
+    merchant_context::MerchantContext,
+    router_data_v2::flow_common_types::{SubscriptionCreateData, SubscriptionCustomerData},
+    router_request_types::{subscriptions as subscription_request_types, ConnectorCustomerData},
+    router_response_types::{
+        subscriptions as subscription_response_types, ConnectorCustomerResponseData,
+        PaymentsResponseData,
+    },
+};
 use masking::Secret;
 
 use super::errors::{self, RouterResponse};
-use crate::routes::SessionState;
+use crate::{
+    core::payments as payments_core, routes::SessionState, services, types::api as api_types,
+};
+
+pub const SUBSCRIPTION_CONNECTOR_ID: &str = "DefaultSubscriptionConnectorId";
+pub const SUBSCRIPTION_PAYMENT_ID: &str = "DefaultSubscriptionPaymentId";
 
 pub async fn create_subscription(
     state: SessionState,
@@ -63,3 +78,544 @@ pub async fn create_subscription(
 
     Ok(ApplicationResponse::Json(response))
 }
+
+pub async fn confirm_subscription(
+    state: SessionState,
+    merchant_context: MerchantContext,
+    profile_id: String,
+    request: subscription_types::ConfirmSubscriptionRequest,
+    subscription_id: common_utils::id_type::SubscriptionId,
+) -> RouterResponse<subscription_types::ConfirmSubscriptionResponse> {
+    let profile_id = common_utils::id_type::ProfileId::from_str(&profile_id).change_context(
+        errors::ApiErrorResponse::InvalidDataValue {
+            field_name: "X-Profile-Id",
+        },
+    )?;
+
+    let key_manager_state = &(&state).into();
+    let merchant_key_store = merchant_context.get_merchant_key_store();
+
+    let profile = state
+        .store
+        .find_business_profile_by_profile_id(key_manager_state, merchant_key_store, &profile_id)
+        .await
+        .change_context(errors::ApiErrorResponse::ProfileNotFound {
+            id: profile_id.get_string_repr().to_string(),
+        })?;
+
+    let customer = state
+        .store
+        .find_customer_by_customer_id_merchant_id(
+            key_manager_state,
+            &request.customer_id,
+            merchant_context.get_merchant_account().get_id(),
+            merchant_key_store,
+            merchant_context.get_merchant_account().storage_scheme,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::CustomerNotFound)
+        .attach_printable("subscriptions: unable to fetch customer from database")?;
+
+    let handler = SubscriptionHandler::new(state, merchant_context, request, profile);
+
+    let mut subscription_entry = handler
+        .find_subscription(subscription_id.get_string_repr().to_string())
+        .await?;
+
+    let billing_handler = subscription_entry.get_billing_handler(customer).await?;
+    let invoice_handler = subscription_entry.get_invoice_handler().await?;
+
+    let _customer_create_response = billing_handler
+        .create_customer_on_connector(&handler.state)
+        .await?;
+
+    let subscription_create_response = billing_handler
+        .create_subscription_on_connector(&handler.state)
+        .await?;
+
+    // let payment_response = invoice_handler.create_cit_payment().await?;
+
+    let invoice_entry = invoice_handler
+        .create_invoice_entry(
+            &handler.state,
+            subscription_entry.profile.get_billing_processor_id()?,
+            None,
+            billing_handler.request.amount,
+            billing_handler.request.currency.to_string(),
+            common_enums::connector_enums::InvoiceStatus::InvoiceCreated,
+            billing_handler.connector_data.connector_name,
+            None,
+        )
+        .await?;
+
+    // invoice_entry
+    //     .create_invoice_record_back_job(&payment_response)
+    //     .await?;
+
+    subscription_entry
+        .update_subscription_status(
+            SubscriptionStatus::from(subscription_create_response.status).to_string(),
+        )
+        .await?;
+
+    let response = subscription_entry
+        .generate_response(&invoice_entry, subscription_create_response.status)?;
+
+    Ok(ApplicationResponse::Json(response))
+}
+
+pub struct SubscriptionHandler {
+    state: SessionState,
+    merchant_context: MerchantContext,
+    request: subscription_types::ConfirmSubscriptionRequest,
+    profile: hyperswitch_domain_models::business_profile::Profile,
+}
+
+impl SubscriptionHandler {
+    pub fn new(
+        state: SessionState,
+        merchant_context: MerchantContext,
+        request: subscription_types::ConfirmSubscriptionRequest,
+        profile: hyperswitch_domain_models::business_profile::Profile,
+    ) -> Self {
+        Self {
+            state,
+            merchant_context,
+            request,
+            profile,
+        }
+    }
+    pub async fn find_subscription(
+        &self,
+        subscription_id: String,
+    ) -> errors::RouterResult<SubscriptionWithHandler<'_>> {
+        let subscription = self
+            .state
+            .store
+            .find_by_merchant_id_subscription_id(
+                self.merchant_context.get_merchant_account().get_id(),
+                subscription_id.clone(),
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::GenericNotFoundError {
+                message: format!("subscription not found for id: {subscription_id}"),
+            })?;
+
+        Ok(SubscriptionWithHandler {
+            handler: self,
+            subscription,
+            profile: self.profile.clone(),
+        })
+    }
+}
+pub struct SubscriptionWithHandler<'a> {
+    handler: &'a SubscriptionHandler,
+    subscription: diesel_models::subscription::Subscription,
+    profile: hyperswitch_domain_models::business_profile::Profile,
+}
+
+impl<'a> SubscriptionWithHandler<'a> {
+    fn generate_response(
+        &self,
+        invoice: &diesel_models::invoice::Invoice,
+        // _payment_response: &subscription_types::PaymentResponseData,
+        status: subscription_response_types::SubscriptionStatus,
+    ) -> errors::RouterResult<subscription_types::ConfirmSubscriptionResponse> {
+        Ok(subscription_types::ConfirmSubscriptionResponse {
+            id: self.subscription.id.clone(),
+            merchant_reference_id: self.subscription.merchant_reference_id.clone(),
+            status: SubscriptionStatus::from(status),
+            plan_id: None,
+            profile_id: self.subscription.profile_id.to_owned(),
+            payment: None,
+            customer_id: Some(self.subscription.customer_id.clone()),
+            price_id: None,
+            coupon: None,
+            invoice: Some(subscription_types::Invoice {
+                id: invoice.id.clone(),
+                subscription_id: invoice.subscription_id.clone(),
+                merchant_id: invoice.merchant_id.clone(),
+                profile_id: invoice.profile_id.clone(),
+                merchant_connector_id: invoice.merchant_connector_id.clone(),
+                payment_intent_id: invoice.payment_intent_id.clone(),
+                payment_method_id: invoice.payment_method_id.clone(),
+                customer_id: invoice.customer_id.clone(),
+                amount: invoice.amount,
+                currency: api_enums::Currency::from_str(invoice.currency.as_str())
+                    .change_context(errors::ApiErrorResponse::InvalidDataValue {
+                        field_name: "currency",
+                    })
+                    .attach_printable(format!(
+                        "unable to parse currency name {currency:?}",
+                        currency = invoice.currency
+                    ))?,
+                status: invoice.status.clone(),
+            }),
+        })
+    }
+
+    async fn update_subscription_status(&mut self, status: String) -> errors::RouterResult<()> {
+        let db = self.handler.state.store.as_ref();
+        let updated_subscription = db
+            .update_subscription_entry(
+                self.handler
+                    .merchant_context
+                    .get_merchant_account()
+                    .get_id(),
+                self.subscription.id.get_string_repr().to_string(),
+                diesel_models::subscription::SubscriptionUpdate::new(None, Some(status)),
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::SubscriptionError {
+                operation: "Subscription Update".to_string(),
+            })
+            .attach_printable("subscriptions: unable to update subscription entry in database")?;
+
+        self.subscription = updated_subscription;
+
+        Ok(())
+    }
+
+    pub async fn get_billing_handler(
+        &self,
+        customer: hyperswitch_domain_models::customer::Customer,
+    ) -> errors::RouterResult<BillingHandler> {
+        let mca_id = self.profile.get_billing_processor_id()?;
+
+        let billing_processor_mca = self
+            .handler
+            .state
+            .store
+            .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+                &(&self.handler.state).into(),
+                self.handler
+                    .merchant_context
+                    .get_merchant_account()
+                    .get_id(),
+                &mca_id,
+                self.handler.merchant_context.get_merchant_key_store(),
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+                id: mca_id.get_string_repr().to_string(),
+            })?;
+
+        let connector_name = billing_processor_mca.connector_name.clone();
+
+        let auth_type: hyperswitch_domain_models::router_data::ConnectorAuthType =
+            payments_core::helpers::MerchantConnectorAccountType::DbVal(Box::new(
+                billing_processor_mca.clone(),
+            ))
+            .get_connector_account_details()
+            .parse_value("ConnectorAuthType")
+            .change_context(errors::ApiErrorResponse::InvalidDataFormat {
+                field_name: "connector_account_details".to_string(),
+                expected_format: "auth_type and api_key".to_string(),
+            })?;
+
+        let connector_data = api_types::ConnectorData::get_connector_by_name(
+            &self.handler.state.conf.connectors,
+            &connector_name,
+            api_types::GetToken::Connector,
+            Some(billing_processor_mca.get_id()),
+        )
+        .change_context(errors::ApiErrorResponse::IncorrectConnectorNameGiven)
+        .attach_printable(
+            "invalid connector name received in billing merchant connector account",
+        )?;
+
+        let connector_enum =
+            common_enums::connector_enums::Connector::from_str(connector_name.as_str())
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable(format!("unable to parse connector name {connector_name:?}"))?;
+
+        let connector_params =
+            hyperswitch_domain_models::connector_endpoints::Connectors::get_connector_params(
+                &self.handler.state.conf.connectors,
+                connector_enum,
+            )
+            .change_context(errors::ApiErrorResponse::ConfigNotFound)
+            .attach_printable(format!(
+                "cannot find connector params for this connector {connector_name} in this flow",
+            ))?;
+
+        Ok(BillingHandler {
+            subscription: self.subscription.clone(),
+            auth_type,
+            connector_data,
+            connector_params,
+            request: self.handler.request.clone(),
+            connector_metadata: billing_processor_mca.metadata.clone(),
+            customer,
+        })
+    }
+
+    pub async fn get_invoice_handler(&self) -> errors::RouterResult<InvoiceHandler> {
+        Ok(InvoiceHandler {
+            subscription: self.subscription.clone(),
+        })
+    }
+}
+
+pub struct BillingHandler {
+    subscription: diesel_models::subscription::Subscription,
+    auth_type: hyperswitch_domain_models::router_data::ConnectorAuthType,
+    connector_data: api_types::ConnectorData,
+    connector_params: hyperswitch_domain_models::connector_endpoints::ConnectorParams,
+    connector_metadata: Option<pii::SecretSerdeValue>,
+    customer: hyperswitch_domain_models::customer::Customer,
+    request: subscription_types::ConfirmSubscriptionRequest,
+}
+
+pub struct InvoiceHandler {
+    subscription: diesel_models::subscription::Subscription,
+}
+
+#[allow(clippy::todo)]
+impl InvoiceHandler {
+    #[allow(clippy::too_many_arguments)]
+    pub async fn create_invoice_entry(
+        self,
+        state: &SessionState,
+        merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
+        payment_intent_id: Option<common_utils::id_type::PaymentId>,
+        amount: common_utils::types::MinorUnit,
+        currency: String,
+        status: common_enums::connector_enums::InvoiceStatus,
+        provider_name: common_enums::connector_enums::Connector,
+        metadata: Option<pii::SecretSerdeValue>,
+    ) -> errors::RouterResult<diesel_models::invoice::Invoice> {
+        let invoice_new = diesel_models::invoice::InvoiceNew::new(
+            self.subscription.id.to_owned(),
+            self.subscription.merchant_id.to_owned(),
+            self.subscription.profile_id.to_owned(),
+            merchant_connector_id,
+            payment_intent_id,
+            self.subscription.payment_method_id.clone(),
+            self.subscription.customer_id.to_owned(),
+            amount,
+            currency,
+            status,
+            provider_name,
+            metadata,
+        );
+
+        let invoice = state
+            .store
+            .insert_invoice_entry(invoice_new)
+            .await
+            .change_context(errors::ApiErrorResponse::SubscriptionError {
+                operation: "Subscription Confirm".to_string(),
+            })
+            .attach_printable("invoices: unable to insert invoice entry to database")?;
+
+        Ok(invoice)
+    }
+
+    pub async fn create_cit_payment(
+        &self,
+    ) -> errors::RouterResult<subscription_types::PaymentResponseData> {
+        // Create a CIT payment for the invoice
+        todo!("Create a CIT payment for the invoice")
+    }
+
+    pub async fn create_invoice_record_back_job(
+        &self,
+        // _invoice: &subscription_types::Invoice,
+        _payment_response: &subscription_types::PaymentResponseData,
+    ) -> errors::RouterResult<()> {
+        // Create an invoice job entry based on payment status
+        todo!("Create an invoice job entry based on payment status")
+    }
+}
+
+#[allow(clippy::todo)]
+impl BillingHandler {
+    pub async fn create_customer_on_connector(
+        &self,
+        state: &SessionState,
+    ) -> errors::RouterResult<ConnectorCustomerResponseData> {
+        let customer_req = ConnectorCustomerData {
+            email: self.customer.email.clone().map(pii::Email::from),
+            payment_method_data: self
+                .request
+                .payment_details
+                .payment_method_data
+                .payment_method_data
+                .clone()
+                .map(|pmd| pmd.into()),
+            description: None,
+            phone: None,
+            name: None,
+            preprocessing_id: None,
+            split_payments: None,
+            setup_future_usage: None,
+            customer_acceptance: None,
+            customer_id: Some(self.subscription.customer_id.to_owned()),
+            billing_address: self
+                .request
+                .billing_address
+                .as_ref()
+                .and_then(|add| add.address.clone())
+                .and_then(|addr| addr.into()),
+        };
+        let router_data = self.build_router_data(
+            state,
+            customer_req,
+            SubscriptionCustomerData {
+                connector_meta_data: self.connector_metadata.clone(),
+            },
+        )?;
+        let connector_integration = self.connector_data.connector.get_connector_integration();
+
+        let response = Box::pin(self.call_connector(
+            state,
+            router_data,
+            "create customer on connector",
+            connector_integration,
+        ))
+        .await?;
+
+        match response {
+            Ok(response_data) => match response_data {
+                PaymentsResponseData::ConnectorCustomerResponse(customer_response) => {
+                    Ok(customer_response)
+                }
+                _ => Err(errors::ApiErrorResponse::SubscriptionError {
+                    operation: "Subscription Customer Create".to_string(),
+                }
+                .into()),
+            },
+            Err(err) => Err(errors::ApiErrorResponse::ExternalConnectorError {
+                code: err.code,
+                message: err.message,
+                connector: self.connector_data.connector_name.to_string(),
+                status_code: err.status_code,
+                reason: err.reason,
+            }
+            .into()),
+        }
+    }
+    pub async fn create_subscription_on_connector(
+        &self,
+        state: &SessionState,
+    ) -> errors::RouterResult<subscription_response_types::SubscriptionCreateResponse> {
+        let subscription_item = subscription_request_types::SubscriptionItem {
+            item_price_id: self.request.get_item_price_id().change_context(
+                errors::ApiErrorResponse::MissingRequiredField {
+                    field_name: "item_price_id",
+                },
+            )?,
+            quantity: Some(1),
+        };
+        let subscription_req = subscription_request_types::SubscriptionCreateRequest {
+            subscription_id: self.subscription.id.to_owned(),
+            customer_id: self.subscription.customer_id.to_owned(),
+            subscription_items: vec![subscription_item],
+            billing_address: self.request.get_billing_address().change_context(
+                errors::ApiErrorResponse::MissingRequiredField {
+                    field_name: "billing_address",
+                },
+            )?,
+            auto_collection: subscription_request_types::SubscriptionAutoCollection::Off,
+            connector_params: self.connector_params.clone(),
+        };
+
+        let router_data = self.build_router_data(
+            state,
+            subscription_req,
+            SubscriptionCreateData {
+                connector_meta_data: self.connector_metadata.clone(),
+            },
+        )?;
+        let connector_integration = self.connector_data.connector.get_connector_integration();
+
+        let response = self
+            .call_connector(
+                state,
+                router_data,
+                "create subscription on connector",
+                connector_integration,
+            )
+            .await?;
+
+        match response {
+            Ok(response_data) => Ok(response_data),
+            Err(err) => Err(errors::ApiErrorResponse::ExternalConnectorError {
+                code: err.code,
+                message: err.message,
+                connector: self.connector_data.connector_name.to_string(),
+                status_code: err.status_code,
+                reason: err.reason,
+            }
+            .into()),
+        }
+    }
+
+    async fn call_connector<F, ResourceCommonData, Req, Resp>(
+        &self,
+        state: &SessionState,
+        router_data: hyperswitch_domain_models::router_data_v2::RouterDataV2<
+            F,
+            ResourceCommonData,
+            Req,
+            Resp,
+        >,
+        operation_name: &str,
+        connector_integration: hyperswitch_interfaces::connector_integration_interface::BoxedConnectorIntegrationInterface<F, ResourceCommonData, Req, Resp>,
+    ) -> errors::RouterResult<Result<Resp, hyperswitch_domain_models::router_data::ErrorResponse>>
+    where
+        F: Clone + std::fmt::Debug + 'static,
+        Req: Clone + std::fmt::Debug + 'static,
+        Resp: Clone + std::fmt::Debug + 'static,
+        ResourceCommonData:
+            hyperswitch_interfaces::connector_integration_interface::RouterDataConversion<
+                    F,
+                    Req,
+                    Resp,
+                > + Clone
+                + 'static,
+    {
+        let old_router_data = ResourceCommonData::to_old_router_data(router_data).change_context(
+            errors::ApiErrorResponse::SubscriptionError {
+                operation: { operation_name.to_string() },
+            },
+        )?;
+
+        let router_resp = services::execute_connector_processing_step(
+            state,
+            connector_integration,
+            &old_router_data,
+            payments_core::CallConnectorAction::Trigger,
+            None,
+            None,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::SubscriptionError {
+            operation: operation_name.to_string(),
+        })
+        .attach_printable(format!(
+            "Failed while in subscription operation: {operation_name}"
+        ))?;
+
+        Ok(router_resp.response)
+    }
+
+    fn build_router_data<F, ResourceCommonData, Req, Resp>(
+        &self,
+        state: &SessionState,
+        req: Req,
+        resource_common_data: ResourceCommonData,
+    ) -> errors::RouterResult<
+        hyperswitch_domain_models::router_data_v2::RouterDataV2<F, ResourceCommonData, Req, Resp>,
+    > {
+        Ok(hyperswitch_domain_models::router_data_v2::RouterDataV2 {
+            flow: std::marker::PhantomData,
+            connector_auth_type: self.auth_type.clone(),
+            resource_common_data,
+            tenant_id: state.tenant.tenant_id.clone(),
+            request: req,
+            response: Err(hyperswitch_domain_models::router_data::ErrorResponse::default()),
+        })
+    }
+}
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 1de0029f767..0d7da850f11 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -1174,13 +1174,21 @@ pub struct Subscription;
 #[cfg(all(feature = "oltp", feature = "v1"))]
 impl Subscription {
     pub fn server(state: AppState) -> Scope {
-        web::scope("/subscription/create")
-            .app_data(web::Data::new(state.clone()))
-            .service(web::resource("").route(
+        let route = web::scope("/subscription").app_data(web::Data::new(state.clone()));
+
+        route
+            .service(web::resource("/create").route(
                 web::post().to(|state, req, payload| {
                     subscription::create_subscription(state, req, payload)
                 }),
             ))
+            .service(
+                web::resource("/{subscription_id}/confirm").route(web::post().to(
+                    |state, req, id, payload| {
+                        subscription::confirm_subscription(state, req, id, payload)
+                    },
+                )),
+            )
     }
 }
 
diff --git a/crates/router/src/routes/lock_utils.rs b/crates/router/src/routes/lock_utils.rs
index 992edb61c35..c0c670f3eee 100644
--- a/crates/router/src/routes/lock_utils.rs
+++ b/crates/router/src/routes/lock_utils.rs
@@ -88,7 +88,7 @@ impl From<Flow> for ApiIdentifier {
             | Flow::DecisionEngineDecideGatewayCall
             | Flow::DecisionEngineGatewayFeedbackCall => Self::Routing,
 
-            Flow::CreateSubscription => Self::Subscription,
+            Flow::CreateSubscription | Flow::ConfirmSubscription => Self::Subscription,
 
             Flow::RetrieveForexFlow => Self::Forex,
             Flow::AddToBlocklist => Self::Blocklist,
diff --git a/crates/router/src/routes/subscription.rs b/crates/router/src/routes/subscription.rs
index d4b716f63c0..aacf03392be 100644
--- a/crates/router/src/routes/subscription.rs
+++ b/crates/router/src/routes/subscription.rs
@@ -67,3 +67,55 @@ pub async fn create_subscription(
     ))
     .await
 }
+
+#[cfg(all(feature = "olap", feature = "v1"))]
+#[instrument(skip_all)]
+pub async fn confirm_subscription(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    subscription_id: web::Path<common_utils::id_type::SubscriptionId>,
+    json_payload: web::Json<subscription_types::ConfirmSubscriptionRequest>,
+) -> impl Responder {
+    let flow = Flow::ConfirmSubscription;
+    let subscription_id = subscription_id.into_inner();
+    let profile_id = match req.headers().get(X_PROFILE_ID) {
+        Some(val) => val.to_str().unwrap_or_default().to_string(),
+        None => {
+            return HttpResponse::BadRequest().json(
+                errors::api_error_response::ApiErrorResponse::MissingRequiredField {
+                    field_name: "x-profile-id",
+                },
+            );
+        }
+    };
+    Box::pin(oss_api::server_wrap(
+        flow,
+        state,
+        &req,
+        json_payload.into_inner(),
+        |state, auth: auth::AuthenticationData, payload, _| {
+            let merchant_context = domain::MerchantContext::NormalMerchant(Box::new(
+                domain::Context(auth.merchant_account, auth.key_store),
+            ));
+            subscription::confirm_subscription(
+                state,
+                merchant_context,
+                profile_id.clone(),
+                payload.clone(),
+                subscription_id.clone(),
+            )
+        },
+        auth::auth_type(
+            &auth::HeaderAuth(auth::ApiKeyAuth {
+                is_connected_allowed: false,
+                is_platform_allowed: false,
+            }),
+            &auth::JWTAuth {
+                permission: Permission::ProfileSubscriptionWrite,
+            },
+            req.headers(),
+        ),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
diff --git a/crates/router/tests/connectors/utils.rs b/crates/router/tests/connectors/utils.rs
index d8f5f1a3a3b..b2582d7e3fe 100644
--- a/crates/router/tests/connectors/utils.rs
+++ b/crates/router/tests/connectors/utils.rs
@@ -3,6 +3,7 @@ use std::{fmt::Debug, marker::PhantomData, str::FromStr, sync::Arc, time::Durati
 use async_trait::async_trait;
 use common_utils::{id_type::GenerateId, pii::Email};
 use error_stack::Report;
+use hyperswitch_domain_models::router_data_v2::flow_common_types::PaymentFlowData;
 use masking::Secret;
 use router::{
     configs::settings::Settings,
@@ -117,7 +118,8 @@ pub trait ConnectorActions: Connector {
         payment_data: Option<types::ConnectorCustomerData>,
         payment_info: Option<PaymentInfo>,
     ) -> Result<types::ConnectorCustomerRouterData, Report<ConnectorError>> {
-        let integration = self.get_data().connector.get_connector_integration();
+        let integration: BoxedConnectorIntegrationInterface<_, PaymentFlowData, _, _> =
+            self.get_data().connector.get_connector_integration();
         let request = self.generate_data(
             types::ConnectorCustomerData {
                 ..(payment_data.unwrap_or(CustomerType::default().0))
diff --git a/crates/router_env/src/logger/types.rs b/crates/router_env/src/logger/types.rs
index 50c47489108..1bbc2605f6b 100644
--- a/crates/router_env/src/logger/types.rs
+++ b/crates/router_env/src/logger/types.rs
@@ -265,6 +265,8 @@ pub enum Flow {
     RoutingDeleteConfig,
     /// Subscription create flow,
     CreateSubscription,
+    /// Subscription confirm flow,
+    ConfirmSubscription,
     /// Create dynamic routing
     CreateDynamicRoutingConfig,
     /// Toggle dynamic routing
diff --git a/migrations/2025-09-23-112547_add_billing_processor_in_connector_type/down.sql b/migrations/2025-09-23-112547_add_billing_processor_in_connector_type/down.sql
new file mode 100644
index 00000000000..d0b08278125
--- /dev/null
+++ b/migrations/2025-09-23-112547_add_billing_processor_in_connector_type/down.sql
@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+SELECT(1);
\ No newline at end of file
diff --git a/migrations/2025-09-23-112547_add_billing_processor_in_connector_type/up.sql b/migrations/2025-09-23-112547_add_billing_processor_in_connector_type/up.sql
new file mode 100644
index 00000000000..0201e3753f1
--- /dev/null
+++ b/migrations/2025-09-23-112547_add_billing_processor_in_connector_type/up.sql
@@ -0,0 +1,3 @@
+-- Your SQL goes here
+ALTER TYPE "ConnectorType"
+ADD VALUE 'billing_processor';
\ No newline at end of file

This pull request introduces a new subscription feature to the codebase, adding comprehensive support for subscription management via new API models, endpoints, and supporting logic. The changes include new data models, API routes, core business logic for creating and confirming subscriptions, and integration with customer management. The implementation is gated behind the `v1` and `olap` feature flags.

**New Subscription Feature**

* Added `subscription` module to `api_models`, including request and response types, core data structures (`Subscription`, `Invoice`), and helper functions for mapping customer data.
* Introduced a new `Subscription` resource to the global `Resource` enum for system-wide referencing.

**Core Logic and Utilities**

* Implemented core business logic in `core/subscription.rs` for creating and confirming subscriptions, including customer handling, client secret generation, and placeholder billing processor integration.
* Added utility functions in `core/subscription/utils.rs` for customer retrieval/creation and extracting customer details from subscription requests.
* Updated customer helper function signature in payouts to allow broader usage in subscription flows.

**Database and Model Changes**

* Added client secret generation to the `SubscriptionNew` model in `diesel_models`, supporting secure subscription creation. [[1]](diffhunk://#diff-18a18c6db4052184eddf2b35c376f2eff90d566c06c6d2fbc4aea751852b0f3aR85-R94) [[2]](diffhunk://#diff-18a18c6db4052184eddf2b35c376f2eff90d566c06c6d2fbc4aea751852b0f3aL1-R1)

**API Routing and Integration**

* Registered new subscription endpoints in the API routes and application server, enabling `/subscription` and `/subscription/{subscription_id}/confirm` endpoints for managing subscriptions. [[1]](diffhunk://#diff-a6e49b399ffdee83dd322b3cb8bee170c65fb9c6e2e52c14222ec310f6927c54R1164-R1186) [[2]](diffhunk://#diff-911527e4effff636e27a7c44689ecc4a5c37963f0e9198193eacb52e1488cab5R212) [[3]](diffhunk://#diff-7e1080d750a904d5d9ea4a8041a6b210d2b0c1fa9e45207500824f0dad7fbc51R53-R54) [[4]](diffhunk://#diff-7e1080d750a904d5d9ea4a8041a6b210d2b0c1fa9e45207500824f0dad7fbc51L99-R101) [[5]](diffhunk://#diff-a6e49b399ffdee83dd322b3cb8bee170c65fb9c6e2e52c14222ec310f6927c54L68-R70)
* Added subscription module references to core and feature-gated module lists for proper compilation and integration. [[1]](diffhunk://#diff-77a35e04ebe8b59efecf0b6a986e706e15170eb486b847e8c8c55514c5e55486R45-R46) [[2]](diffhunk://#diff-3f92a5ed20c7decce0129ef9eb838b87bae76de2197e8efa267657c26f590f4bR52-R53)
* Updated API identifier enums to include `Subscription` for locking and routing utilities.] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->


### 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 a customer
```
curl --location 'http://localhost:8080/customers' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_Ske75Nx2J7qtHsP8cc7pFx5k4dccYBedM6UAExaLOdHCkji3uVWSqfmZ0Qz0Tnyj' \
--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"
    }
}'
```
2. create a subscription
```
curl --location 'http://localhost:8080/subscription/create' \
--header 'Content-Type: application/json' \
--header 'X-Profile-Id: pro_X16mQSHipgRuWbTEMkMp' \
--header 'api-key: dev_Ske75Nx2J7qtHsP8cc7pFx5k4dccYBedM6UAExaLOdHCkji3uVWSqfmZ0Qz0Tnyj' \
--data '{
    "customer_id": "cus_NdHhw4wwWyYXSldO9oYE"
}'
```

response - 
```
{"id":"subscription_wBV1G9dhh6EBhTOTXRBA","merchant_reference_id":null,"status":"Created","plan_id":null,"profile_id":"pro_X16mQSHipgRuWbTEMkMp","client_secret":"subscription_wBV1G9dhh6EBhTOTXRBA_secret_6W89VL1bl5C3esgbxpJp","merchant_id":"merchant_1758626894","coupon_code":null,"customer_id":"cus_NdHhw4wwWyYXSldO9oYE"}
```
3. Confirm subscription
```
curl --location 'http://localhost:8080/subscription/subscription_wBV1G9dhh6EBhTOTXRBA/confirm' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'X-Profile-Id: pro_2WzEeiNyj8fSCObXqo36' \
--header 'api-key: dev_Ske75Nx2J7qtHsP8cc7pFx5k4dccYBedM6UAExaLOdHCkji3uVWSqfmZ0Qz0Tnyj' \
--data '{
    "amount": 14100,
    "currency": "INR",
    "item_price_id": "cbdemo_enterprise-suite-INR-Daily",
    "customer_id": "cus_NdHhw4wwWyYXSldO9oYE",
    "billing_address": {
        "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": "8056594427",
            "country_code": "+91"
        }
    },
    "payment_details": {
        "payment_method": "card",
        "payment_method_type": "credit",
        "payment_method_data": {
            "card": {
                "card_number": "4242424242424242",
                "card_exp_month": "10",
                "card_exp_year": "25",
                "card_holder_name": "joseph Doe",
                "card_cvc": "123"
            }
        },
        "setup_future_usage": "off_session",
        "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":"subscription_wBV1G9dhh6EBhTOTXRBA","merchant_reference_id":null,"status":"Active","plan_id":null,"price_id":null,"coupon":null,"profile_id":"pro_X16mQSHipgRuWbTEMkMp","payment":null,"customer_id":"cus_NdHhw4wwWyYXSldO9oYE","invoice":{"id":"invoice_0XANlbhMp2V7wUvWRhhJ","subscription_id":"subscription_wBV1G9dhh6EBhTOTXRBA","merchant_id":"merchant_1758626894","profile_id":"pro_X16mQSHipgRuWbTEMkMp","merchant_connector_id":"mca_eN6JxSK2NkuT0wSYAH5s","payment_intent_id":null,"payment_method_id":null,"customer_id":"cus_NdHhw4wwWyYXSldO9oYE","amount":14100,"currency":"INR","status":"InvoiceCreated"}}
```

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