Issue:
Bug: Get Plans



**Request to Subscription Povider to get plans**
/api/v2/items?limit=5&type[is]=plan&item_family_id[is]=cbdemo_omnisupport-solutions'

**Response from Subscription Provider**
```
{
    "list": [
        {
            "item": {
                "id": "cbdemo_enterprise-suite",
                "name": "Enterprise Suite",
                "external_name": "Enterprise Suite",
                "description": "High-end customer support suite with enterprise-grade solutions.",
                "status": "active",
                "resource_version": 1754897340100,
                "updated_at": 1754897340,
                "item_family_id": "cbdemo_omnisupport-solutions",
                "type": "plan",
                "is_shippable": false,
                "is_giftable": false,
                "enabled_for_checkout": true,
                "enabled_in_portal": true,
                "item_applicability": "all",
                "metered": false,
                "channel": "web",
                "bundle_items": [],
                "deleted": false,
                "object": "item"
            }
        },
        {
            "item": {
                "id": "cbdemo_business-suite",
                "name": "Business Suite",
                "external_name": "Business Suite",
                "description": "Advanced customer support plan with premium features.",
                "status": "active",
                "resource_version": 1754897339947,
                "updated_at": 1754897339,
                "item_family_id": "cbdemo_omnisupport-solutions",
                "type": "plan",
                "is_shippable": false,
                "is_giftable": false,
                "enabled_for_checkout": true,
                "enabled_in_portal": true,
                "item_applicability": "all",
                "metered": false,
                "channel": "web",
                "bundle_items": [],
                "deleted": false,
                "object": "item"
            }
        },
        {
            "item": {
                "id": "cbdemo_professional-suite",
                "name": "Professional Suite",
                "external_name": "Professional Suite",
                "description": "Comprehensive customer support plan with extended capabilities.",
                "status": "active",
                "resource_version": 1754897339843,
                "updated_at": 1754897339,
                "item_family_id": "cbdemo_omnisupport-solutions",
                "type": "plan",
                "is_shippable": false,
                "is_giftable": false,
                "enabled_for_checkout": true,
                "enabled_in_portal": true,
                "item_applicability": "all",
                "metered": false,
                "channel": "web",
                "bundle_items": [],
                "deleted": false,
                "object": "item"
            }
        },
        {
            "item": {
                "id": "cbdemo_essential-support",
                "name": "Essential Support",
                "external_name": "Essential Support",
                "description": "Basic customer support plan with core features.",
                "status": "active",
                "resource_version": 1754897339723,
                "updated_at": 1754897339,
                "item_family_id": "cbdemo_omnisupport-solutions",
                "type": "plan",
                "is_shippable": false,
                "is_giftable": false,
                "enabled_for_checkout": true,
                "enabled_in_portal": true,
                "item_applicability": "all",
                "metered": false,
                "channel": "web",
                "bundle_items": [],
                "deleted": false,
                "object": "item"
            }
        }
    ]
}
```


Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
This PR includes connector integration for the get subscription plans API for Chargebee

### 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).
-->
Closes #9053 

## 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)?
-->
PR can be tested once the API handlers for this endpoint are implemented

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

PR can be tested once the API handlers for this endpoint are implemented

Patch:
diff --git a/crates/api_models/src/lib.rs b/crates/api_models/src/lib.rs
index 3b343459e9e..c991f3c048b 100644
--- a/crates/api_models/src/lib.rs
+++ b/crates/api_models/src/lib.rs
@@ -42,6 +42,7 @@ pub mod recon;
 pub mod refunds;
 pub mod relay;
 pub mod routing;
+pub mod subscription;
 pub mod surcharge_decision_configs;
 pub mod three_ds_decision_rule;
 #[cfg(feature = "tokenization_v2")]
diff --git a/crates/api_models/src/payments.rs b/crates/api_models/src/payments.rs
index 4789ec15dfb..e8a9cb29d9f 100644
--- a/crates/api_models/src/payments.rs
+++ b/crates/api_models/src/payments.rs
@@ -897,6 +897,13 @@ impl AmountDetailsUpdate {
         self.tax_on_surcharge
     }
 }
+
+#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
+pub struct BillingConnectorDetails {
+    pub processor_mca: id_type::MerchantConnectorAccountId,
+    pub subscription_id: String,
+    pub invoice_id: String,
+}
 #[cfg(feature = "v1")]
 #[derive(
     Default,
@@ -1255,6 +1262,8 @@ pub struct PaymentsRequest {
 
     /// Allow partial authorization for this payment
     pub enable_partial_authorization: Option<bool>,
+
+    pub billing_processor_details: Option<BillingConnectorDetails>,
 }
 
 #[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize, ToSchema)]
@@ -7234,7 +7243,12 @@ pub struct ApplepaySessionRequest {
     pub initiative: String,
     pub initiative_context: String,
 }
-
+// #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+// pub struct BillingConnectorDetails {
+//     pub connector: String,
+//     pub subscription_id: String,
+//     pub invoice_id: String,
+// }
 /// Some connectors like Apple Pay, Airwallex and Noon might require some additional information, find specific details in the child attributes below.
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
 pub struct ConnectorMetadata {
diff --git a/crates/api_models/src/process_tracker.rs b/crates/api_models/src/process_tracker.rs
index 9854933734f..97ff63c65a7 100644
--- a/crates/api_models/src/process_tracker.rs
+++ b/crates/api_models/src/process_tracker.rs
@@ -1,2 +1,3 @@
 #[cfg(feature = "v2")]
 pub mod revenue_recovery;
+pub mod subscription;
diff --git a/crates/api_models/src/process_tracker/subscription.rs b/crates/api_models/src/process_tracker/subscription.rs
new file mode 100644
index 00000000000..0393be4e052
--- /dev/null
+++ b/crates/api_models/src/process_tracker/subscription.rs
@@ -0,0 +1,16 @@
+use common_utils::id_type;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SubscriptionWorkflowTrackingData {
+    pub merchant_id: id_type::MerchantId,
+    pub profile_id: id_type::ProfileId,
+    pub payment_method_id: String,
+    pub subscription_id: Option<String>,
+    pub invoice_id: String,
+    pub amount: common_utils::types::MinorUnit,
+    pub currency: common_enums::Currency,
+    pub customer_id: Option<id_type::CustomerId>,
+    pub connector_name: String,
+    pub billing_connector_mca_id: id_type::MerchantConnectorAccountId,
+}
diff --git a/crates/api_models/src/subscription.rs b/crates/api_models/src/subscription.rs
new file mode 100644
index 00000000000..ce6cc9ff7b8
--- /dev/null
+++ b/crates/api_models/src/subscription.rs
@@ -0,0 +1,108 @@
+use common_utils::{events::ApiEventMetric, pii};
+
+use crate::{
+    customers::{CustomerRequest, CustomerResponse},
+    payments::CustomerDetailsResponse,
+};
+
+pub const SUBSCRIPTION_ID_PREFIX: &str = "sub";
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct CreateSubscriptionRequest {
+    pub subscription_id: Option<String>,
+    pub plan_id: Option<String>,
+    pub coupon_code: Option<String>,
+    pub mca_id: Option<String>,
+    pub confirm: bool,
+    pub customer_id: Option<common_utils::id_type::CustomerId>,
+    pub customer: Option<CustomerRequest>,
+}
+
+impl CreateSubscriptionRequest {
+    pub fn get_customer_id(&self) -> Option<&common_utils::id_type::CustomerId> {
+        self.customer_id
+            .as_ref()
+            .or_else(|| self.customer.as_ref()?.customer_id.as_ref())
+    }
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct CreateSubscriptionResponse {
+    pub subscription: Subscription,
+    pub client_secret: Option<String>,
+    pub merchant_id: String,
+    pub mca_id: Option<String>,
+    pub coupon_code: Option<String>,
+    pub customer: Option<CustomerDetailsResponse>,
+    pub invoice: Option<Invoice>,
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct Subscription {
+    pub id: String,
+    pub status: SubscriptionStatus,
+    pub plan_id: Option<String>,
+}
+
+#[derive(Debug, Clone, serde::Serialize, strum::EnumString, strum::Display)]
+pub enum SubscriptionStatus {
+    Created,
+    Active,
+    InActive,
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct Invoice {
+    pub id: String,
+    pub total: u64,
+}
+
+impl Subscription {
+    pub fn new(id: impl Into<String>, status: SubscriptionStatus, plan_id: Option<String>) -> Self {
+        Self {
+            id: id.into(),
+            status,
+            plan_id,
+        }
+    }
+}
+
+impl Invoice {
+    pub fn new(id: impl Into<String>, total: u64) -> Self {
+        Self {
+            id: id.into(),
+            total,
+        }
+    }
+}
+impl CreateSubscriptionResponse {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        subscription: Subscription,
+        merchant_id: impl Into<String>,
+        mca_id: Option<String>,
+    ) -> Self {
+        Self {
+            subscription,
+            client_secret: None,
+            merchant_id: merchant_id.into(),
+            mca_id,
+            coupon_code: None,
+            customer: None,
+            invoice: None,
+        }
+    }
+}
+
+pub fn map_customer_resp_to_details(r: &CustomerResponse) -> CustomerDetailsResponse {
+    CustomerDetailsResponse {
+        id: Some(r.customer_id.clone()),
+        name: r.name.as_ref().map(|n| n.clone().into_inner()),
+        email: r.email.as_ref().map(|e| pii::Email::from(e.clone())),
+        phone: r.phone.as_ref().map(|p| p.clone().into_inner()),
+        phone_country_code: r.phone_country_code.clone(),
+    }
+}
+
+impl ApiEventMetric for CreateSubscriptionResponse {}
+impl ApiEventMetric for CreateSubscriptionRequest {}
diff --git a/crates/api_models/src/webhooks.rs b/crates/api_models/src/webhooks.rs
index bb58a0dc440..6b42778f564 100644
--- a/crates/api_models/src/webhooks.rs
+++ b/crates/api_models/src/webhooks.rs
@@ -66,6 +66,8 @@ pub enum IncomingWebhookEvent {
     RecoveryPaymentPending,
     #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
     RecoveryInvoiceCancel,
+    #[cfg(feature = "v1")]
+    InvoiceGenerated,
 }
 
 impl IncomingWebhookEvent {
@@ -297,6 +299,8 @@ impl From<IncomingWebhookEvent> for WebhookFlow {
             | IncomingWebhookEvent::RecoveryPaymentFailure
             | IncomingWebhookEvent::RecoveryPaymentPending
             | IncomingWebhookEvent::RecoveryPaymentSuccess => Self::Recovery,
+            #[cfg(feature = "v1")]
+            IncomingWebhookEvent::InvoiceGenerated => Self::Subscription,
         }
     }
 }
diff --git a/crates/common_enums/src/enums.rs b/crates/common_enums/src/enums.rs
index 67c5efb00a6..124af626717 100644
--- a/crates/common_enums/src/enums.rs
+++ b/crates/common_enums/src/enums.rs
@@ -7777,6 +7777,7 @@ pub enum Resource {
     RunRecon,
     ReconConfig,
     RevenueRecovery,
+    Subscription,
     InternalConnector,
     Theme,
 }
@@ -8733,6 +8734,7 @@ pub enum ProcessTrackerRunner {
     PassiveRecoveryWorkflow,
     ProcessDisputeWorkflow,
     DisputeListWorkflow,
+    SubscriptionsWorkflow,
 }
 
 #[derive(Debug)]
diff --git a/crates/diesel_models/src/lib.rs b/crates/diesel_models/src/lib.rs
index 246807a041a..caf979b2c5f 100644
--- a/crates/diesel_models/src/lib.rs
+++ b/crates/diesel_models/src/lib.rs
@@ -44,6 +44,7 @@ pub mod relay;
 pub mod reverse_lookup;
 pub mod role;
 pub mod routing_algorithm;
+pub mod subscription;
 pub mod types;
 pub mod unified_translations;
 
diff --git a/crates/diesel_models/src/payment_intent.rs b/crates/diesel_models/src/payment_intent.rs
index a152d4085df..791f5b5debe 100644
--- a/crates/diesel_models/src/payment_intent.rs
+++ b/crates/diesel_models/src/payment_intent.rs
@@ -178,6 +178,7 @@ pub struct PaymentIntent {
     pub duty_amount: Option<MinorUnit>,
     pub order_date: Option<PrimitiveDateTime>,
     pub enable_partial_authorization: Option<bool>,
+    pub billing_processor_details: Option<serde_json::Value>,
 }
 
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize, diesel::AsExpression, PartialEq)]
@@ -467,6 +468,7 @@ pub struct PaymentIntentNew {
     pub shipping_amount_tax: Option<MinorUnit>,
     pub duty_amount: Option<MinorUnit>,
     pub enable_partial_authorization: Option<bool>,
+    pub billing_processor_details: Option<serde_json::Value>,
 }
 
 #[cfg(feature = "v2")]
diff --git a/crates/diesel_models/src/query.rs b/crates/diesel_models/src/query.rs
index 7c8831b3811..17eae2427e7 100644
--- a/crates/diesel_models/src/query.rs
+++ b/crates/diesel_models/src/query.rs
@@ -39,6 +39,7 @@ pub mod relay;
 pub mod reverse_lookup;
 pub mod role;
 pub mod routing_algorithm;
+pub mod subscription;
 #[cfg(feature = "tokenization_v2")]
 pub mod tokenization;
 pub mod unified_translations;
diff --git a/crates/diesel_models/src/query/subscription.rs b/crates/diesel_models/src/query/subscription.rs
new file mode 100644
index 00000000000..a8289a58d44
--- /dev/null
+++ b/crates/diesel_models/src/query/subscription.rs
@@ -0,0 +1,94 @@
+use async_bb8_diesel::AsyncRunQueryDsl;
+use diesel::{
+    associations::HasTable, debug_query, pg::Pg, BoolExpressionMethods, ExpressionMethods, QueryDsl,
+};
+use error_stack::report;
+use error_stack::ResultExt;
+
+use super::generics;
+use crate::{
+    errors,
+    schema::subscription::dsl,
+    subscription::{Subscription, SubscriptionNew, SubscriptionUpdate},
+    PgPooledConn, StorageResult,
+};
+
+impl SubscriptionNew {
+    pub async fn insert(self, conn: &PgPooledConn) -> StorageResult<Subscription> {
+        generics::generic_insert(conn, self).await
+    }
+}
+
+impl Subscription {
+    pub async fn find_by_merchant_id_subscription_id(
+        conn: &PgPooledConn,
+        merchant_id: &common_utils::id_type::MerchantId,
+        subscription_id: String,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::merchant_id
+                .eq(merchant_id.to_owned())
+                .and(dsl::subscription_id.eq(subscription_id.to_owned())),
+        )
+        .await
+    }
+
+    pub async fn find_payment_method_ids_by_billing_connector_subscription_id(
+        conn: &PgPooledConn,
+        subscription_id: &str,
+    ) -> StorageResult<Vec<String>> {
+        let query = <Self as HasTable>::table()
+            .select(dsl::payment_method_id)
+            .filter(dsl::subscription_id.eq(subscription_id.to_owned()))
+            .filter(dsl::payment_method_id.is_not_null())
+            .order(dsl::created_at.desc())
+            .into_boxed();
+
+        router_env::logger::debug!(query = %debug_query::<Pg, _>(&query).to_string());
+
+        generics::db_metrics::track_database_call::<<Self as HasTable>::Table, _, _>(
+            query.load_async::<Option<String>>(conn),
+            generics::db_metrics::DatabaseOperation::Filter,
+        )
+        .await
+        .change_context(errors::DatabaseError::Others)
+        .attach_printable("Failed to find payment method IDs by billing connector subscription ID")
+        .map(|results| results.into_iter().filter_map(|x| x).collect())
+    }
+
+    pub async fn update_subscription_entry(
+        conn: &PgPooledConn,
+        merchant_id: &common_utils::id_type::MerchantId,
+        subscription_id: String,
+        subscription_update: SubscriptionUpdate,
+    ) -> StorageResult<Self> {
+        generics::generic_update_with_results::<
+            <Self as HasTable>::Table,
+            SubscriptionUpdate,
+            _,
+            _,
+        >(
+            conn,
+            dsl::subscription_id
+                .eq(subscription_id.to_owned())
+                .and(dsl::merchant_id.eq(merchant_id.to_owned())),
+            subscription_update,
+        )
+        .await?
+        .first()
+        .cloned()
+        .ok_or_else(|| {
+            report!(errors::DatabaseError::NotFound)
+                .attach_printable("Error while updating subscription entry")
+        })
+    }
+
+    // pub async fn find_subscription_by_id(conn: &PgPooledConn, id: String) -> StorageResult<Self> {
+    //     generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+    //         conn,
+    //         dsl::id.eq(id.to_owned()),
+    //     )
+    //     .await
+    // }
+}
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index 08675d904de..7881facea27 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -1086,6 +1086,7 @@ diesel::table! {
         duty_amount -> Nullable<Int8>,
         order_date -> Nullable<Timestamp>,
         enable_partial_authorization -> Nullable<Bool>,
+        billing_processor_details -> Nullable<Jsonb>,
     }
 }
 
@@ -1465,6 +1466,34 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    subscription (id) {
+        id -> Int4,
+        #[max_length = 128]
+        subscription_id -> Varchar,
+        #[max_length = 128]
+        status -> Varchar,
+        #[max_length = 128]
+        billing_processor -> Nullable<Varchar>,
+        #[max_length = 128]
+        payment_method_id -> Nullable<Varchar>,
+        #[max_length = 128]
+        mca_id -> Nullable<Varchar>,
+        #[max_length = 128]
+        client_secret -> Nullable<Varchar>,
+        #[max_length = 64]
+        merchant_id -> Varchar,
+        #[max_length = 64]
+        customer_id -> Varchar,
+        metadata -> Nullable<Jsonb>,
+        created_at -> Timestamp,
+        modified_at -> Timestamp,
+    }
+}
+
 diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
@@ -1650,6 +1679,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     reverse_lookup,
     roles,
     routing_algorithm,
+    subscription,
     themes,
     unified_translations,
     user_authentication_methods,
diff --git a/crates/diesel_models/src/schema_v2.rs b/crates/diesel_models/src/schema_v2.rs
index ea3980d942a..8c0cf89b070 100644
--- a/crates/diesel_models/src/schema_v2.rs
+++ b/crates/diesel_models/src/schema_v2.rs
@@ -1400,6 +1400,32 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    subscription (id) {
+        id -> Int4,
+        #[max_length = 128]
+        subscription_id -> Varchar,
+        #[max_length = 128]
+        billing_processor -> Nullable<Varchar>,
+        #[max_length = 128]
+        payment_method_id -> Nullable<Varchar>,
+        #[max_length = 128]
+        mca_id -> Nullable<Varchar>,
+        #[max_length = 128]
+        client_secret -> Nullable<Varchar>,
+        #[max_length = 64]
+        merchant_id -> Varchar,
+        #[max_length = 64]
+        customer_id -> Varchar,
+        metadata -> Nullable<Jsonb>,
+        created_at -> Timestamp,
+        modified_at -> Timestamp,
+    }
+}
+
 diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
@@ -1605,6 +1631,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     reverse_lookup,
     roles,
     routing_algorithm,
+    subscription,
     themes,
     tokenization,
     unified_translations,
diff --git a/crates/diesel_models/src/subscription.rs b/crates/diesel_models/src/subscription.rs
new file mode 100644
index 00000000000..0a0be40be8d
--- /dev/null
+++ b/crates/diesel_models/src/subscription.rs
@@ -0,0 +1,99 @@
+use common_utils::{generate_id_with_default_len, pii::SecretSerdeValue};
+use diesel::{AsChangeset, Identifiable, Insertable, Queryable, Selectable};
+use serde::{Deserialize, Serialize};
+
+use crate::schema::subscription;
+
+#[derive(Clone, Debug, Eq, Insertable, PartialEq, Serialize, Deserialize)]
+#[diesel(table_name = subscription)]
+pub struct SubscriptionNew {
+    subscription_id: String,
+    status: String,
+    billing_processor: Option<String>,
+    payment_method_id: Option<String>,
+    mca_id: Option<String>,
+    client_secret: Option<String>,
+    merchant_id: common_utils::id_type::MerchantId,
+    customer_id: common_utils::id_type::CustomerId,
+    metadata: Option<SecretSerdeValue>,
+    created_at: time::PrimitiveDateTime,
+    modified_at: time::PrimitiveDateTime,
+}
+
+#[derive(
+    Clone, Debug, Eq, PartialEq, Identifiable, Queryable, Selectable, Deserialize, Serialize,
+)]
+#[diesel(table_name = subscription, primary_key(id), check_for_backend(diesel::pg::Pg))]
+pub struct Subscription {
+    #[serde(skip_serializing, skip_deserializing)]
+    pub id: i32,
+    pub subscription_id: String,
+    pub status: String,
+    pub billing_processor: Option<String>,
+    pub payment_method_id: Option<String>,
+    pub mca_id: Option<String>,
+    pub client_secret: Option<String>,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub customer_id: common_utils::id_type::CustomerId,
+    pub metadata: Option<serde_json::Value>,
+    pub created_at: time::PrimitiveDateTime,
+    pub modified_at: time::PrimitiveDateTime,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, AsChangeset, router_derive::DebugAsDisplay, Deserialize)]
+#[diesel(table_name = subscription)]
+pub struct SubscriptionUpdate {
+    pub payment_method_id: Option<String>,
+    pub status: Option<String>,
+    pub modified_at: time::PrimitiveDateTime,
+}
+
+impl SubscriptionNew {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        subscription_id: String,
+        status: String,
+        billing_processor: Option<String>,
+        payment_method_id: Option<String>,
+        mca_id: Option<String>,
+        client_secret: Option<String>,
+        merchant_id: common_utils::id_type::MerchantId,
+        customer_id: common_utils::id_type::CustomerId,
+        metadata: Option<SecretSerdeValue>,
+    ) -> Self {
+        let now = common_utils::date_time::now();
+        Self {
+            subscription_id,
+            status,
+            billing_processor,
+            payment_method_id,
+            mca_id,
+            client_secret,
+            merchant_id,
+            customer_id,
+            metadata,
+            created_at: now,
+            modified_at: now,
+        }
+    }
+
+    pub fn generate_and_set_client_secret(&mut self) -> Option<String> {
+        let client_secret = Some(generate_id_with_default_len(&format!(
+            "{}_secret",
+            self.subscription_id
+        )));
+
+        self.client_secret = self.client_secret.clone();
+        client_secret
+    }
+}
+
+impl SubscriptionUpdate {
+    pub fn new(payment_method_id: Option<String>, status: Option<String>) -> Self {
+        Self {
+            payment_method_id,
+            status,
+            modified_at: common_utils::date_time::now(),
+        }
+    }
+}
diff --git a/crates/hyperswitch_connectors/src/connectors/chargebee.rs b/crates/hyperswitch_connectors/src/connectors/chargebee.rs
index 19ec99fc1d0..35d8044fcd0 100644
--- a/crates/hyperswitch_connectors/src/connectors/chargebee.rs
+++ b/crates/hyperswitch_connectors/src/connectors/chargebee.rs
@@ -9,32 +9,43 @@ use common_utils::{
     request::{Method, Request, RequestBuilder, RequestContent},
     types::{AmountConvertor, MinorUnit, MinorUnitForConnector},
 };
-#[cfg(feature = "v1")]
-use error_stack::report;
+
 use error_stack::ResultExt;
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
 use hyperswitch_domain_models::{
-    revenue_recovery, router_flow_types::revenue_recovery::RecoveryRecordBack,
+    router_flow_types::revenue_recovery::RecoveryRecordBack,
     router_request_types::revenue_recovery::RevenueRecoveryRecordBackRequest,
     router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
     types::RevenueRecoveryRecordBackRouterData,
 };
+#[cfg(feature = "v2")]
+use hyperswitch_domain_models::{
+    router_data_v2::{flow_common_types::RevenueRecoveryRecordBackData, RouterDataV2},
+};
+#[cfg(all(feature = "v2", feature = "revenue_recovery"))]
+use hyperswitch_domain_models::revenue_recovery;
+
 use hyperswitch_domain_models::{
     router_data::{AccessToken, ConnectorAuthType, ErrorResponse, RouterData},
     router_flow_types::{
         access_token_auth::AccessTokenAuth,
         payments::{Authorize, Capture, PSync, PaymentMethodToken, Session, SetupMandate, Void},
         refunds::{Execute, RSync},
+        subscriptions::GetSubscriptionPlans,
     },
     router_request_types::{
-        AccessTokenRequestData, PaymentMethodTokenizationData, PaymentsAuthorizeData,
-        PaymentsCancelData, PaymentsCaptureData, PaymentsSessionData, PaymentsSyncData,
-        RefundsData, SetupMandateRequestData,
+        subscriptions::GetSubscriptionPlansRequest, AccessTokenRequestData,
+        PaymentMethodTokenizationData, PaymentsAuthorizeData, PaymentsCancelData,
+        PaymentsCaptureData, PaymentsSessionData, PaymentsSyncData, RefundsData,
+        SetupMandateRequestData,
+    },
+    router_response_types::{
+        subscriptions::GetSubscriptionPlansResponse, ConnectorInfo, PaymentsResponseData,
+        RefundsResponseData,
     },
-    router_response_types::{ConnectorInfo, PaymentsResponseData, RefundsResponseData},
     types::{
-        PaymentsAuthorizeRouterData, PaymentsCaptureRouterData, PaymentsSyncRouterData,
-        RefundSyncRouterData, RefundsRouterData,
+        GetSubscriptionPlansRouterData, PaymentsAuthorizeRouterData, PaymentsCaptureRouterData,
+        PaymentsSyncRouterData, RefundSyncRouterData, RefundsRouterData,
     },
 };
 use hyperswitch_interfaces::{
@@ -48,10 +59,15 @@ use hyperswitch_interfaces::{
     types::{self, Response},
     webhooks,
 };
+#[cfg(feature = "v2")]
+use hyperswitch_interfaces::connector_integration_v2::ConnectorIntegrationV2;
 use masking::{Mask, PeekInterface, Secret};
 use transformers as chargebee;
 
-use crate::{constants::headers, types::ResponseRouterData, utils};
+use crate::{
+    connectors::chargebee::transformers::ChargebeeListPlansResponse, constants::headers,
+    types::ResponseRouterData, utils,
+};
 
 #[derive(Clone)]
 pub struct Chargebee {
@@ -80,6 +96,396 @@ impl api::RefundSync for Chargebee {}
 impl api::PaymentToken for Chargebee {}
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
 impl api::revenue_recovery::RevenueRecoveryRecordBack for Chargebee {}
+// #[cfg(feature = "v1")]
+// impl api::subscriptions::Subscriptions for Chargebee {}
+#[cfg(feature = "v1")]
+impl api::subscriptions::SubscriptionRecordBack for Chargebee {}
+
+// impl api::revenue_recovery::SubscriptionsRecordBack for Chargebee {}
+
+#[cfg(feature = "v1")]
+impl
+    ConnectorIntegration<
+        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+        hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+    > for Chargebee
+{
+    fn get_headers(
+        &self,
+        req: &hyperswitch_domain_models::types::SubscriptionRecordBackRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<Vec<(String, masking::Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+    fn get_url(
+        &self,
+        req: &hyperswitch_domain_models::types::SubscriptionRecordBackRouterData,
+        connectors: &Connectors,
+    ) -> 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("$", "hyperswitch-juspay-test"); // have to fix this
+        let invoice_id = req
+            .request
+            .merchant_reference_id
+            .to_string();
+        Ok(format!("{url}v2/invoices/{invoice_id}/record_payment"))
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_request_body(
+        &self,
+        req: &hyperswitch_domain_models::types::SubscriptionRecordBackRouterData,
+        _connectors: &Connectors,
+    ) -> CustomResult<RequestContent, errors::ConnectorError> {
+        let amount = utils::convert_amount(
+            self.amount_converter,
+            req.request.amount,
+            req.request.currency,
+        )?;
+        let connector_router_data = chargebee::ChargebeeRouterData::from((amount, req));
+        let connector_req =
+            chargebee::ChargebeeRecordPaymentRequest::try_from(&connector_router_data)?;
+        Ok(RequestContent::FormUrlEncoded(Box::new(connector_req)))
+    }
+
+    fn build_request(
+        &self,
+        req: &hyperswitch_domain_models::types::SubscriptionRecordBackRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<Option<Request>, errors::ConnectorError> {
+        Ok(Some(
+            RequestBuilder::new()
+                .method(Method::Post)
+                .url(&types::SubscriptionRecordBackType::get_url(
+                    self, req, connectors,
+                )?)
+                .attach_default_headers()
+                .headers(types::SubscriptionRecordBackType::get_headers(
+                    self, req, connectors,
+                )?)
+                .set_body(types::SubscriptionRecordBackType::get_request_body(
+                    self, req, connectors,
+                )?)
+                .build(),
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        data: &hyperswitch_domain_models::types::SubscriptionRecordBackRouterData,
+        event_builder: Option<&mut ConnectorEvent>,
+        res: Response,
+    ) -> CustomResult<hyperswitch_domain_models::types::SubscriptionRecordBackRouterData, errors::ConnectorError> {
+        let response: chargebee::ChargebeeRecordbackResponse = res
+            .response
+            .parse_struct("chargebee ChargebeeRecordbackResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_response_body(&response));
+        router_env::logger::info!(connector_response=?response);
+        RouterData::try_from(ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+
+    fn get_error_response(
+        &self,
+        res: Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res, event_builder)
+    }          
+}
+#[cfg(all(feature = "v2", feature = "v1"))]
+impl
+    ConnectorIntegrationV2<
+        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+        hyperswitch_domain_models::router_data_v2::flow_common_types::RevenueRecoveryRecordBackData,
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+        hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+    > for Chargebee
+{
+    fn get_headers(
+        &self,
+        _req: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::RevenueRecoveryRecordBackData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+        >,
+    ) -> CustomResult<Vec<(String, masking::Maskable<String>)>, errors::ConnectorError> {
+        todo!()
+    }
+
+    fn get_url(
+        &self,
+        _req: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::RevenueRecoveryRecordBackData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+        >,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        todo!()
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        todo!()
+    }
+
+    fn get_request_body(
+        &self,
+        _req: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::RevenueRecoveryRecordBackData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+        >,
+    ) -> CustomResult<Option<RequestContent>, errors::ConnectorError> {
+        todo!()
+    }
+
+    fn build_request_v2(
+        &self,
+        _req: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::RevenueRecoveryRecordBackData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+        >,
+    ) -> CustomResult<Option<Request>, errors::ConnectorError> {
+        todo!()
+    }
+
+    fn handle_response_v2(
+        &self,
+        _data: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::RevenueRecoveryRecordBackData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+        >,
+        _event_builder: Option<&mut ConnectorEvent>,
+        _res: types::Response,
+    ) -> CustomResult<
+        RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::RevenueRecoveryRecordBackData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+        >,
+        errors::ConnectorError,
+    > {
+        todo!()
+    }
+
+    fn get_error_response_v2(
+        &self,
+        _res: types::Response,
+        _event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<ErrorResponse, errors::ConnectorError> {
+        todo!()
+    }
+}
+
+
+#[cfg(feature = "v1")]
+impl api::subscriptions::SubscriptionCreate for Chargebee {}
+
+#[cfg(feature = "v1")]
+impl
+    ConnectorIntegration<
+        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+        hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse,
+    > for Chargebee
+{
+    fn get_headers(
+        &self,
+        req: &hyperswitch_domain_models::types::SubscriptionCreateRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<Vec<(String, masking::Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_url(
+        &self,
+        req: &hyperswitch_domain_models::types::SubscriptionCreateRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        let url = self
+            .base_url(connectors)
+            .to_string()
+            .replace("$", "hyperswitch-juspay-test"); // have to fix this
+        let customer_id = &req.request.customer_id;
+        Ok(format!("{url}v2/customers/{customer_id}/subscription_for_items"))
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_request_body(
+        &self,
+        req: &hyperswitch_domain_models::types::SubscriptionCreateRouterData,
+        _connectors: &Connectors,
+    ) -> CustomResult<RequestContent, errors::ConnectorError> {
+        let connector_router_data = chargebee::ChargebeeRouterData::from((MinorUnit::new(0), req));
+        let connector_req =
+            chargebee::ChargebeeSubscriptionCreateRequest::try_from(&connector_router_data)?;
+        Ok(RequestContent::FormUrlEncoded(Box::new(connector_req)))
+    }
+
+    fn build_request(
+        &self,
+        req: &hyperswitch_domain_models::types::SubscriptionCreateRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<Option<Request>, errors::ConnectorError> {
+        Ok(Some(
+            RequestBuilder::new()
+                .method(Method::Post)
+                .url(&types::SubscriptionCreateType::get_url(
+                    self, req, connectors,
+                )?)
+                .attach_default_headers()
+                .headers(types::SubscriptionCreateType::get_headers(
+                    self, req, connectors,
+                )?)
+                .set_body(types::SubscriptionCreateType::get_request_body(
+                    self, req, connectors,
+                )?)
+                .build(),
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        data: &hyperswitch_domain_models::types::SubscriptionCreateRouterData,
+        event_builder: Option<&mut ConnectorEvent>,
+        res: Response,
+    ) -> CustomResult<hyperswitch_domain_models::types::SubscriptionCreateRouterData, errors::ConnectorError> {
+        let response: chargebee::ChargebeeSubscriptionCreateResponse = res
+            .response
+            .parse_struct("chargebee ChargebeeSubscriptionCreateResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_response_body(&response));
+        router_env::logger::info!(connector_response=?response);
+        RouterData::try_from(ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+
+    fn get_error_response(
+        &self,
+        res: Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res, event_builder)
+    }
+}
+
+#[cfg(all(feature = "v2", feature = "v1"))]
+impl
+    ConnectorIntegrationV2<
+        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+        hyperswitch_domain_models::router_data_v2::flow_common_types::SubscriptionCreateData,
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+        hyperswitch_domain_models::router_response_types::revenue_recovery::SubscriptionCreateResponse,
+    > for Chargebee
+{
+    fn get_headers(
+        &self,
+        _req: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::SubscriptionCreateData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::SubscriptionCreateResponse,
+        >,
+    ) -> CustomResult<Vec<(String, masking::Maskable<String>)>, errors::ConnectorError> {
+        todo!()
+    }
+
+    fn get_url(
+        &self,
+        _req: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::SubscriptionCreateData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::SubscriptionCreateResponse,
+        >,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        todo!()
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        todo!()
+    }
+
+    fn get_request_body(
+        &self,
+        _req: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::SubscriptionCreateData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::SubscriptionCreateResponse,
+        >,
+    ) -> CustomResult<Option<RequestContent>, errors::ConnectorError> {
+        todo!()
+    }
+
+    fn build_request_v2(
+        &self,
+        _req: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::SubscriptionCreateData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::SubscriptionCreateResponse,
+        >,
+    ) -> CustomResult<Option<Request>, errors::ConnectorError> {
+        todo!()
+    }
+
+    fn handle_response_v2(
+        &self,
+        _data: &RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::SubscriptionCreateData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::SubscriptionCreateResponse,
+        >,
+        _event_builder: Option<&mut ConnectorEvent>,
+        _res: types::Response,
+    ) -> CustomResult<
+        RouterDataV2<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+            hyperswitch_domain_models::router_data_v2::flow_common_types::SubscriptionCreateData,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+            hyperswitch_domain_models::router_response_types::revenue_recovery::SubscriptionCreateResponse,
+        >,
+        errors::ConnectorError,
+    > {
+        todo!()
+    }
+
+    fn get_error_response_v2(
+        &self,
+        _res: types::Response,
+        _event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<ErrorResponse, errors::ConnectorError> {
+        todo!()
+    }
+}
 
 impl ConnectorIntegration<PaymentMethodToken, PaymentMethodTokenizationData, PaymentsResponseData>
     for Chargebee
@@ -664,6 +1070,100 @@ impl
     }
 }
 
+fn get_chargebee_plans_query_params(
+    _req: &RouterData<
+        GetSubscriptionPlans,
+        GetSubscriptionPlansRequest,
+        GetSubscriptionPlansResponse,
+    >,
+) -> CustomResult<String, errors::ConnectorError> {
+    let limit = 10; // hardcoded limit param
+    let item_type = "plan"; // hardcoded filter param
+    let param = format!("?limit={}&type[is]={}", limit, item_type);
+    Ok(param)
+}
+
+impl
+    ConnectorIntegration<
+        GetSubscriptionPlans,
+        GetSubscriptionPlansRequest,
+        GetSubscriptionPlansResponse,
+    > for Chargebee
+{
+    fn get_headers(
+        &self,
+        req: &GetSubscriptionPlansRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<Vec<(String, masking::Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_url(
+        &self,
+        req: &GetSubscriptionPlansRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<String, errors::ConnectorError> {
+        let query_params = get_chargebee_plans_query_params(req)?;
+        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());
+        Ok(format!("{url}v2/items{query_params}"))
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn build_request(
+        &self,
+        req: &GetSubscriptionPlansRouterData,
+        connectors: &Connectors,
+    ) -> CustomResult<Option<Request>, errors::ConnectorError> {
+        Ok(Some(
+            RequestBuilder::new()
+                .method(Method::Get)
+                .url(&types::GetSubscriptionPlansType::get_url(
+                    self, req, connectors,
+                )?)
+                .attach_default_headers()
+                .headers(types::GetSubscriptionPlansType::get_headers(
+                    self, req, connectors,
+                )?)
+                .build(),
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        data: &GetSubscriptionPlansRouterData,
+        event_builder: Option<&mut ConnectorEvent>,
+        res: Response,
+    ) -> CustomResult<GetSubscriptionPlansRouterData, errors::ConnectorError> {
+        let response: ChargebeeListPlansResponse = res
+            .response
+            .parse_struct("chargebee ChargebeeListPlansResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        event_builder.map(|i| i.set_response_body(&response));
+        router_env::logger::info!(connector_response=?response);
+        RouterData::try_from(ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+
+    fn get_error_response(
+        &self,
+        res: Response,
+        event_builder: Option<&mut ConnectorEvent>,
+    ) -> CustomResult<ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res, event_builder)
+    }
+}
+
 #[async_trait::async_trait]
 impl webhooks::IncomingWebhook for Chargebee {
     fn get_webhook_source_verification_signature(
@@ -717,7 +1217,6 @@ impl webhooks::IncomingWebhook for Chargebee {
         Ok(signature_auth == secret_auth)
     }
 
-    #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
     fn get_webhook_object_reference_id(
         &self,
         request: &webhooks::IncomingWebhookRequestDetails<'_>,
@@ -725,18 +1224,28 @@ impl webhooks::IncomingWebhook for Chargebee {
         let webhook =
             chargebee::ChargebeeInvoiceBody::get_invoice_webhook_data_from_body(request.body)
                 .change_context(errors::ConnectorError::WebhookReferenceIdNotFound)?;
-        Ok(api_models::webhooks::ObjectReferenceId::InvoiceId(
-            api_models::webhooks::InvoiceIdType::ConnectorInvoiceId(webhook.content.invoice.id),
-        ))
-    }
-    #[cfg(any(feature = "v1", not(all(feature = "revenue_recovery", feature = "v2"))))]
-    fn get_webhook_object_reference_id(
-        &self,
-        _request: &webhooks::IncomingWebhookRequestDetails<'_>,
-    ) -> CustomResult<api_models::webhooks::ObjectReferenceId, errors::ConnectorError> {
-        Err(report!(errors::ConnectorError::WebhooksNotImplemented))
+        
+        #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
+        {
+            Ok(api_models::webhooks::ObjectReferenceId::InvoiceId(
+                api_models::webhooks::InvoiceIdType::ConnectorInvoiceId(webhook.content.invoice.id),
+            ))
+        }
+        
+        #[cfg(any(feature = "v1", not(all(feature = "revenue_recovery", feature = "v2"))))]
+        {
+            // For v1 or non-revenue recovery, use subscription ID as object reference
+            // This allows the webhook to be processed without errors
+            let subscription_id = webhook.content.invoice.subscription_id
+                .unwrap_or_else(|| webhook.content.invoice.id.clone());
+            Ok(api_models::webhooks::ObjectReferenceId::PaymentId(
+                api_models::payments::PaymentIdType::PaymentIntentId(
+                    common_utils::id_type::PaymentId::wrap(subscription_id)
+                        .change_context(errors::ConnectorError::WebhookReferenceIdNotFound)?
+                )
+            ))
+        }
     }
-    #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
     fn get_webhook_event_type(
         &self,
         request: &webhooks::IncomingWebhookRequestDetails<'_>,
@@ -747,13 +1256,6 @@ impl webhooks::IncomingWebhook for Chargebee {
         let event = api_models::webhooks::IncomingWebhookEvent::from(webhook.event_type);
         Ok(event)
     }
-    #[cfg(any(feature = "v1", not(all(feature = "revenue_recovery", feature = "v2"))))]
-    fn get_webhook_event_type(
-        &self,
-        _request: &webhooks::IncomingWebhookRequestDetails<'_>,
-    ) -> CustomResult<api_models::webhooks::IncomingWebhookEvent, errors::ConnectorError> {
-        Err(report!(errors::ConnectorError::WebhooksNotImplemented))
-    }
 
     fn get_webhook_resource_object(
         &self,
diff --git a/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs b/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
index 95f105c67b9..a7a13ea1c1e 100644
--- a/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
+++ b/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
@@ -1,6 +1,3 @@
-#[cfg(all(feature = "revenue_recovery", feature = "v2"))]
-use std::str::FromStr;
-
 use common_enums::enums;
 use common_utils::{errors::CustomResult, ext_traits::ByteSliceExt, pii, types::MinorUnit};
 use error_stack::ResultExt;
@@ -13,16 +10,19 @@ use hyperswitch_domain_models::{
         refunds::{Execute, RSync},
         RecoveryRecordBack,
     },
+    router_request_types::subscriptions::SubscriptionsRecordBackRequest,
     router_request_types::{revenue_recovery::RevenueRecoveryRecordBackRequest, ResponseId},
     router_response_types::{
-        revenue_recovery::RevenueRecoveryRecordBackResponse, PaymentsResponseData,
-        RefundsResponseData,
+        revenue_recovery::RevenueRecoveryRecordBackResponse,
+        subscriptions::GetSubscriptionPlansResponse, PaymentsResponseData, RefundsResponseData,
     },
     types::{PaymentsAuthorizeRouterData, RefundsRouterData, RevenueRecoveryRecordBackRouterData},
 };
 use hyperswitch_interfaces::errors;
-use masking::Secret;
+use masking::{ExposeInterface, Secret};
 use serde::{Deserialize, Serialize};
+#[cfg(all(feature = "revenue_recovery", feature = "v2"))]
+use std::str::FromStr;
 use time::PrimitiveDateTime;
 
 use crate::{
@@ -169,6 +169,143 @@ impl<F, T> TryFrom<ResponseRouterData<F, ChargebeePaymentsResponse, T, PaymentsR
     }
 }
 
+// SubscriptionCreate structures
+#[derive(Debug, Serialize)]
+pub struct ChargebeeSubscriptionCreateRequest {
+    #[serde(rename = "subscription_items[item_price_id][0]")]
+    pub item_price_id: String,
+    #[serde(rename = "id")]
+    pub subscription_id: String,
+    #[serde(rename = "subscription_items[quantity][0]")]
+    pub quantity: Option<u32>,
+    #[serde(rename = "billing_address[line1]")]
+    pub billing_address_line1: Option<String>,
+    #[serde(rename = "billing_address[city]")]
+    pub billing_address_city: Option<String>,
+    #[serde(rename = "billing_address[state]")]
+    pub billing_address_state: Option<String>,
+    #[serde(rename = "billing_address[zip]")]
+    pub billing_address_zip: Option<String>,
+    #[serde(rename = "billing_address[country]")]
+    pub billing_address_country: Option<String>,
+    #[serde(rename = "auto_collection")]
+    pub auto_collection: String,
+}
+
+#[cfg(feature = "v1")]
+impl TryFrom<&ChargebeeRouterData<&hyperswitch_domain_models::types::SubscriptionCreateRouterData>>
+    for ChargebeeSubscriptionCreateRequest
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: &ChargebeeRouterData<&hyperswitch_domain_models::types::SubscriptionCreateRouterData>,
+    ) -> Result<Self, Self::Error> {
+        let req = &item.router_data.request;
+
+        // Get the first subscription item (assuming at least one exists)
+        let first_item =
+            req.subscription_items
+                .first()
+                .ok_or(errors::ConnectorError::MissingRequiredField {
+                    field_name: "subscription_items",
+                })?;
+
+        Ok(Self {
+            item_price_id: first_item.item_price_id.clone(),
+            subscription_id: req.subscription_id.clone(),
+            quantity: first_item.quantity,
+            billing_address_line1: req
+                .billing_address
+                .address
+                .as_ref()
+                .and_then(|addr| addr.line1.as_ref().map(|line1| line1.clone().expose())),
+            billing_address_city: req
+                .billing_address
+                .address
+                .as_ref()
+                .and_then(|addr| addr.city.clone()),
+            billing_address_state: req
+                .billing_address
+                .address
+                .as_ref()
+                .and_then(|addr| addr.state.as_ref().map(|state| state.clone().expose())),
+            billing_address_zip: req
+                .billing_address
+                .address
+                .as_ref()
+                .and_then(|addr| addr.zip.as_ref().map(|zip| zip.clone().expose())),
+            billing_address_country: req
+                .billing_address
+                .address
+                .as_ref()
+                .and_then(|addr| addr.country.map(|country| country.to_string())),
+            auto_collection: req.auto_collection.clone(),
+        })
+    }
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct ChargebeeSubscriptionCreateResponse {
+    pub subscription: ChargebeeSubscriptionDetails,
+    pub invoice: ChargebeeInvoiceDetails,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct ChargebeeInvoiceDetails {
+    pub id: String,
+    pub subscription_id: String,
+    pub status: String,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct ChargebeeSubscriptionDetails {
+    pub id: String,
+    pub status: String,
+    pub customer_id: String,
+    pub currency_code: enums::Currency,
+    pub total_dues: Option<MinorUnit>,
+    #[serde(default, with = "common_utils::custom_serde::timestamp::option")]
+    pub next_billing_at: Option<PrimitiveDateTime>,
+    #[serde(default, with = "common_utils::custom_serde::timestamp::option")]
+    pub created_at: Option<PrimitiveDateTime>,
+}
+
+#[cfg(feature = "v1")]
+impl TryFrom<
+    ResponseRouterData<
+        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+        ChargebeeSubscriptionCreateResponse,
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+        hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse,
+    >,
+> for hyperswitch_domain_models::types::SubscriptionCreateRouterData
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: ResponseRouterData<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+            ChargebeeSubscriptionCreateResponse,
+            hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+            hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse,
+        >,
+    ) -> Result<Self, Self::Error> {
+        let subscription = &item.response.subscription;
+        let invoice = &item.response.invoice;
+        Ok(Self {
+            response: Ok(hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse {
+                subscription_id: subscription.id.clone(),
+                invoice_id: invoice.id.clone(),
+                status: subscription.status.clone(),
+                customer_id: subscription.customer_id.clone(),
+                currency_code: subscription.currency_code,
+                total_amount: subscription.total_dues.unwrap_or(MinorUnit::new(0)),
+                next_billing_at: subscription.next_billing_at,
+                created_at: subscription.created_at,
+            }),
+            ..item.data
+        })
+    }
+}
 //TODO: Fill the struct with respective fields
 // REFUND :
 // Type definition for RefundRequest
@@ -291,6 +428,7 @@ pub enum ChargebeeEventType {
     PaymentSucceeded,
     PaymentFailed,
     InvoiceDeleted,
+    InvoiceGenerated,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -301,6 +439,11 @@ pub struct ChargebeeInvoiceData {
     pub currency_code: enums::Currency,
     pub billing_address: Option<ChargebeeInvoiceBillingAddress>,
     pub linked_payments: Option<Vec<ChargebeeInvoicePayments>>,
+    // New fields for invoice_generated webhook
+    pub status: Option<String>,
+    pub customer_id: Option<common_utils::id_type::CustomerId>,
+    pub subscription_id: Option<String>,
+    pub first_invoice: Option<bool>,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -415,6 +558,36 @@ impl ChargebeeInvoiceBody {
     }
 }
 
+// Structure to extract MIT payment data from invoice_generated webhook
+#[derive(Debug, Clone)]
+pub struct ChargebeeMitPaymentData {
+    pub invoice_id: String,
+    pub amount_due: MinorUnit,
+    pub currency_code: enums::Currency,
+    pub status: String,
+    pub customer_id: Option<common_utils::id_type::CustomerId>,
+    pub subscription_id: Option<String>,
+    pub first_invoice: bool,
+}
+
+impl TryFrom<ChargebeeInvoiceBody> for ChargebeeMitPaymentData {
+    type Error = error_stack::Report<errors::ConnectorError>;
+
+    fn try_from(webhook_body: ChargebeeInvoiceBody) -> Result<Self, Self::Error> {
+        let invoice = webhook_body.content.invoice;
+
+        Ok(Self {
+            invoice_id: invoice.id,
+            amount_due: invoice.total,
+            currency_code: invoice.currency_code,
+            status: invoice.status.unwrap_or_else(|| "payment_due".to_string()),
+            customer_id: invoice.customer_id,
+            subscription_id: invoice.subscription_id,
+            first_invoice: invoice.first_invoice.unwrap_or(false),
+        })
+    }
+}
+
 pub struct ChargebeeMandateDetails {
     pub customer_id: String,
     pub mandate_id: String,
@@ -575,6 +748,18 @@ impl From<ChargebeeEventType> for api_models::webhooks::IncomingWebhookEvent {
             ChargebeeEventType::PaymentSucceeded => Self::RecoveryPaymentSuccess,
             ChargebeeEventType::PaymentFailed => Self::RecoveryPaymentFailure,
             ChargebeeEventType::InvoiceDeleted => Self::RecoveryInvoiceCancel,
+            ChargebeeEventType::InvoiceGenerated => todo!(),
+        }
+    }
+}
+#[cfg(feature = "v1")]
+impl From<ChargebeeEventType> for api_models::webhooks::IncomingWebhookEvent {
+    fn from(event: ChargebeeEventType) -> Self {
+        match event {
+            ChargebeeEventType::PaymentSucceeded => Self::PaymentIntentSuccess,
+            ChargebeeEventType::PaymentFailed => Self::PaymentIntentFailure,
+            ChargebeeEventType::InvoiceDeleted => Self::EventNotSupported,
+            ChargebeeEventType::InvoiceGenerated => Self::InvoiceGenerated,
         }
     }
 }
@@ -734,6 +919,71 @@ impl TryFrom<enums::AttemptStatus> for ChargebeeRecordStatus {
         }
     }
 }
+#[cfg(feature = "v1")]
+impl
+    TryFrom<
+        &ChargebeeRouterData<&hyperswitch_domain_models::types::SubscriptionRecordBackRouterData>,
+    > for ChargebeeRecordPaymentRequest
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: &ChargebeeRouterData<
+            &hyperswitch_domain_models::types::SubscriptionRecordBackRouterData,
+        >,
+    ) -> Result<Self, Self::Error> {
+        let req = &item.router_data.request;
+        Ok(Self {
+            amount: req.amount,
+            payment_method: ChargebeeRecordPaymentMethod::Other,
+            connector_payment_id: req
+                .connector_transaction_id
+                .as_ref()
+                .map(|connector_payment_id| connector_payment_id.get_id().to_string()),
+            status: ChargebeeRecordStatus::try_from(req.attempt_status)?,
+        })
+    }
+}
+
+#[cfg(feature = "v1")]
+impl TryFrom<enums::AttemptStatus> for ChargebeeRecordStatus {
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(status: enums::AttemptStatus) -> Result<Self, Self::Error> {
+        match status {
+            enums::AttemptStatus::Charged
+            | enums::AttemptStatus::PartialCharged
+            | enums::AttemptStatus::PartialChargedAndChargeable => Ok(Self::Success),
+            enums::AttemptStatus::Failure
+            | enums::AttemptStatus::CaptureFailed
+            | enums::AttemptStatus::RouterDeclined => Ok(Self::Failure),
+            enums::AttemptStatus::AuthenticationFailed
+            | enums::AttemptStatus::Started
+            | enums::AttemptStatus::AuthenticationPending
+            | enums::AttemptStatus::AuthenticationSuccessful
+            | enums::AttemptStatus::Authorized
+            | enums::AttemptStatus::PartiallyAuthorized
+            | enums::AttemptStatus::AuthorizationFailed
+            | enums::AttemptStatus::Authorizing
+            | enums::AttemptStatus::CodInitiated
+            | enums::AttemptStatus::Voided
+            | enums::AttemptStatus::VoidedPostCharge
+            | enums::AttemptStatus::VoidInitiated
+            | enums::AttemptStatus::CaptureInitiated
+            | enums::AttemptStatus::VoidFailed
+            | enums::AttemptStatus::AutoRefunded
+            | enums::AttemptStatus::Unresolved
+            | enums::AttemptStatus::Pending
+            | enums::AttemptStatus::PaymentMethodAwaited
+            | enums::AttemptStatus::ConfirmationAwaited
+            | enums::AttemptStatus::DeviceDataCollectionPending
+            | enums::AttemptStatus::IntegrityFailure
+            | enums::AttemptStatus::Expired => Err(errors::ConnectorError::NotSupported {
+                message: "Record back flow is only supported for terminal status".to_string(),
+                connector: "chargebee",
+            }
+            .into()),
+        }
+    }
+}
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
 pub struct ChargebeeRecordbackResponse {
@@ -773,3 +1023,79 @@ impl
         })
     }
 }
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ChargebeeListPlansResponse {
+    pub list: Vec<ChargebeeItemWrapper>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ChargebeeItemWrapper {
+    pub item: ChargebeeItem,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ChargebeeItem {
+    pub id: String,
+    pub name: String,
+    #[serde(rename = "type")]
+    pub plan_type: String, // to check if new enum is required for this
+    pub is_giftable: bool,
+    pub enabled_for_checkout: bool,
+    pub enabled_in_portal: bool,
+    pub metered: bool,
+    pub deleted: bool,
+    pub description: Option<String>,
+}
+
+impl<F, T>
+    TryFrom<ResponseRouterData<F, ChargebeeListPlansResponse, T, GetSubscriptionPlansResponse>>
+    for RouterData<F, T, GetSubscriptionPlansResponse>
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: ResponseRouterData<F, ChargebeeListPlansResponse, T, GetSubscriptionPlansResponse>,
+    ) -> Result<Self, Self::Error> {
+        let plans = item
+            .response
+            .list
+            .into_iter()
+            .map(|wrapper| {
+                hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionPlans {
+                    subscription_provider_plan_id: wrapper.item.id,
+                    name: wrapper.item.name,
+                    description: wrapper.item.description,
+                }
+            })
+            .collect();
+        Ok(Self {
+            response: Ok(GetSubscriptionPlansResponse { list: plans }),
+#[cfg(feature = "v1")]
+impl
+    TryFrom<
+        ResponseRouterData<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+            ChargebeeRecordbackResponse,
+            SubscriptionsRecordBackRequest,
+            RevenueRecoveryRecordBackResponse,
+        >,
+    > for hyperswitch_domain_models::types::SubscriptionRecordBackRouterData
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: ResponseRouterData<
+            hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+            ChargebeeRecordbackResponse,
+            SubscriptionsRecordBackRequest,
+            RevenueRecoveryRecordBackResponse,
+        >,
+    ) -> Result<Self, Self::Error> {
+        let merchant_reference_id = item.response.invoice.id;
+        Ok(Self {
+            response: Ok(RevenueRecoveryRecordBackResponse {
+                merchant_reference_id,
+            }),
+            ..item.data
+        })
+    }
+}
diff --git a/crates/hyperswitch_connectors/src/connectors/recurly.rs b/crates/hyperswitch_connectors/src/connectors/recurly.rs
index 35b736b844e..baa438b2da6 100644
--- a/crates/hyperswitch_connectors/src/connectors/recurly.rs
+++ b/crates/hyperswitch_connectors/src/connectors/recurly.rs
@@ -10,8 +10,11 @@ use error_stack::ResultExt;
 use hyperswitch_domain_models::{
     router_data::{ConnectorAuthType, ErrorResponse},
     router_data_v2::UasFlowData,
-    router_flow_types::unified_authentication_service::{
-        Authenticate, AuthenticationConfirmation, PostAuthenticate, PreAuthenticate,
+    router_flow_types::{
+        unified_authentication_service::{
+            Authenticate, AuthenticationConfirmation, PostAuthenticate, PreAuthenticate,
+        },
+        SubscriptionRecordBack,
     },
     router_request_types::unified_authentication_service::{
         UasAuthenticationRequestData, UasAuthenticationResponseData, UasConfirmationRequestData,
@@ -20,12 +23,17 @@ use hyperswitch_domain_models::{
 };
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
 use hyperswitch_domain_models::{
-    router_data_v2::flow_common_types as recovery_flow_common_types,
     router_flow_types::revenue_recovery as recovery_router_flows,
     router_request_types::revenue_recovery as recovery_request_types,
-    router_response_types::revenue_recovery as recovery_response_types,
     types as recovery_router_data_types,
 };
+use hyperswitch_domain_models::{
+    router_flow_types::subscriptions::SubscriptionCreate,
+    router_data_v2::flow_common_types as recovery_flow_common_types,
+    router_request_types::subscriptions as subscriptions_request_types,
+    router_response_types::subscriptions as subscriptions_response_types,
+    router_response_types::revenue_recovery as recovery_response_types,
+};
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
 use hyperswitch_interfaces::types;
 use hyperswitch_interfaces::{
@@ -131,6 +139,27 @@ impl
 }
 
 impl api::revenue_recovery_v2::RevenueRecoveryV2 for Recurly {}
+impl api::subscriptions_v2::SubscriptionsV2 for Recurly {}
+impl api::subscriptions_v2::SubscriptionsRecordBackV2 for Recurly {}
+impl api::subscriptions_v2::SubscriptionsCreateV2 for Recurly {}
+impl
+    ConnectorIntegrationV2<
+        SubscriptionRecordBack,
+        recovery_flow_common_types::RevenueRecoveryRecordBackData,
+        subscriptions_request_types::SubscriptionsRecordBackRequest,
+        recovery_response_types::RevenueRecoveryRecordBackResponse,
+    > for Recurly
+{
+}
+impl 
+    ConnectorIntegrationV2<
+        SubscriptionCreate,
+        recovery_flow_common_types::SubscriptionCreateData,
+        subscriptions_request_types::SubscriptionCreateRequest,
+        subscriptions_response_types::SubscriptionCreateResponse,
+    > for Recurly
+{
+}
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
 impl api::revenue_recovery_v2::RevenueRecoveryRecordBackV2 for Recurly {}
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
diff --git a/crates/hyperswitch_connectors/src/default_implementations.rs b/crates/hyperswitch_connectors/src/default_implementations.rs
index 8b87e0c098a..f2fa0652afb 100644
--- a/crates/hyperswitch_connectors/src/default_implementations.rs
+++ b/crates/hyperswitch_connectors/src/default_implementations.rs
@@ -10,6 +10,13 @@ use common_utils::errors::CustomResult;
 use hyperswitch_domain_models::router_flow_types::{
     BillingConnectorInvoiceSync, BillingConnectorPaymentsSync, RecoveryRecordBack,
 };
+
+use hyperswitch_domain_models::router_flow_types::{
+    SubscriptionRecordBack as SubscriptionRecordBackFlow,
+    SubscriptionCreate as SubscriptionCreateFlow,
+};
+use hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest;
+use hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse;
 #[cfg(feature = "dummy_connector")]
 use hyperswitch_domain_models::router_request_types::authentication::{
     ConnectorAuthenticationRequestData, ConnectorPostAuthenticationRequestData, PreAuthNRequestData,
@@ -19,11 +26,14 @@ use hyperswitch_domain_models::router_request_types::revenue_recovery::{
     BillingConnectorInvoiceSyncRequest, BillingConnectorPaymentsSyncRequest,
     RevenueRecoveryRecordBackRequest,
 };
+
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
 use hyperswitch_domain_models::router_response_types::revenue_recovery::{
-    BillingConnectorInvoiceSyncResponse, BillingConnectorPaymentsSyncResponse,
-    RevenueRecoveryRecordBackResponse,
+    BillingConnectorInvoiceSyncResponse, BillingConnectorPaymentsSyncResponse, RevenueRecoveryRecordBackResponse,
 };
+#[cfg(feature = "v1")]
+use hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse;
+
 use hyperswitch_domain_models::{
     router_data::AccessTokenAuthenticationResponse,
     router_flow_types::{
@@ -45,6 +55,7 @@ use hyperswitch_domain_models::{
         ExternalVaultProxy, ExternalVaultRetrieveFlow, PostAuthenticate, PreAuthenticate,
     },
     router_request_types::{
+        subscriptions::SubscriptionsRecordBackRequest,
         authentication,
         unified_authentication_service::{
             UasAuthenticationRequestData, UasAuthenticationResponseData,
@@ -124,6 +135,7 @@ use hyperswitch_interfaces::{
             PaymentsPreProcessing, TaxCalculation,
         },
         revenue_recovery::RevenueRecovery,
+        subscriptions::{Subscriptions,SubscriptionRecordBack,SubscriptionCreate},
         vault::{
             ExternalVault, ExternalVaultCreate, ExternalVaultDelete, ExternalVaultInsert,
             ExternalVaultRetrieve,
@@ -6333,6 +6345,141 @@ macro_rules! default_imp_for_revenue_recovery {
     };
 }
 
+macro_rules! default_imp_for_subscriptions {
+    ($($path:ident::$connector:ident),*) => {
+        $(  impl Subscriptions for $path::$connector {}
+        )*
+    };
+}
+
+default_imp_for_subscriptions!(
+    connectors::Vgs,
+    connectors::Aci,
+    connectors::Adyen,
+    connectors::Adyenplatform,
+    connectors::Affirm,
+    connectors::Airwallex,
+    connectors::Amazonpay,
+    connectors::Archipel,
+    connectors::Authipay,
+    connectors::Authorizedotnet,
+    connectors::Bambora,
+    connectors::Bamboraapac,
+    connectors::Bankofamerica,
+    connectors::Barclaycard,
+    connectors::Billwerk,
+    connectors::Bluesnap,
+    connectors::Bitpay,
+    connectors::Blackhawknetwork,
+    connectors::Bluecode,
+    connectors::Braintree,
+    connectors::Boku,
+    connectors::Breadpay,
+    connectors::Cashtocode,
+    connectors::Celero,
+    connectors::Chargebee,
+    connectors::Checkbook,
+    connectors::Checkout,
+    connectors::Coinbase,
+    connectors::Coingate,
+    connectors::Cryptopay,
+    connectors::CtpMastercard,
+    connectors::Custombilling,
+    connectors::Cybersource,
+    connectors::Datatrans,
+    connectors::Deutschebank,
+    connectors::Digitalvirgo,
+    connectors::Dlocal,
+    connectors::Dwolla,
+    connectors::Ebanx,
+    connectors::Elavon,
+    connectors::Facilitapay,
+    connectors::Fiserv,
+    connectors::Fiservemea,
+    connectors::Fiuu,
+    connectors::Flexiti,
+    connectors::Forte,
+    connectors::Getnet,
+    connectors::Globalpay,
+    connectors::Globepay,
+    connectors::Gocardless,
+    connectors::Gpayments,
+    connectors::Hipay,
+    connectors::Helcim,
+    connectors::HyperswitchVault,
+    connectors::Hyperwallet,
+    connectors::Iatapay,
+    connectors::Inespay,
+    connectors::Itaubank,
+    connectors::Jpmorgan,
+    connectors::Juspaythreedsserver,
+    connectors::Katapult,
+    connectors::Klarna,
+    connectors::Netcetera,
+    connectors::Nmi,
+    connectors::Nomupay,
+    connectors::Noon,
+    connectors::Nordea,
+    connectors::Novalnet,
+    connectors::Nexinets,
+    connectors::Nexixpay,
+    connectors::Nuvei,
+    connectors::Opayo,
+    connectors::Opennode,
+    connectors::Payeezy,
+    connectors::Payload,
+    connectors::Paystack,
+    connectors::Paytm,
+    connectors::Payu,
+    connectors::Phonepe,
+    connectors::Paypal,
+    connectors::Powertranz,
+    connectors::Prophetpay,
+    connectors::Mifinity,
+    connectors::Mollie,
+    connectors::Moneris,
+    connectors::Mpgs,
+    connectors::Multisafepay,
+    connectors::Paybox,
+    connectors::Payme,
+    connectors::Payone,
+    connectors::Placetopay,
+    connectors::Plaid,
+    connectors::Rapyd,
+    connectors::Razorpay,
+    connectors::Recurly,
+    connectors::Redsys,
+    connectors::Riskified,
+    connectors::Santander,
+    connectors::Shift4,
+    connectors::Sift,
+    connectors::Silverflow,
+    connectors::Signifyd,
+    connectors::Stax,
+    connectors::Stripe,
+    connectors::Square,
+    connectors::Stripebilling,
+    connectors::Taxjar,
+    connectors::Threedsecureio,
+    connectors::Thunes,
+    connectors::Tokenio,
+    connectors::Trustpay,
+    connectors::Trustpayments,
+    connectors::Tsys,
+    connectors::UnifiedAuthenticationService,
+    connectors::Wise,
+    connectors::Worldline,
+    connectors::Worldpay,
+    connectors::Worldpayvantiv,
+    connectors::Worldpayxml,
+    connectors::Wellsfargo,
+    connectors::Wellsfargopayout,
+    connectors::Volt,
+    connectors::Xendit,
+    connectors::Zen,
+    connectors::Zsl
+);
+
 default_imp_for_revenue_recovery!(
     connectors::Vgs,
     connectors::Aci,
@@ -6894,6 +7041,148 @@ default_imp_for_billing_connector_invoice_sync!(
     connectors::Zsl
 );
 
+#[cfg(feature = "v1")]
+macro_rules! default_imp_for_subscription_record_back {
+    ($($path:ident::$connector:ident),*) => {
+        $( impl SubscriptionRecordBack for $path::$connector {}
+            impl
+            ConnectorIntegration<
+            SubscriptionRecordBackFlow,
+            SubscriptionsRecordBackRequest,
+            RevenueRecoveryRecordBackResponse
+            > for $path::$connector
+            {}
+        )*
+    };
+}
+
+#[cfg(feature = "v1")]
+default_imp_for_subscription_record_back!(
+    connectors::Vgs,
+    connectors::Aci,
+    connectors::Adyen,
+    connectors::Adyenplatform,
+    connectors::Affirm,
+    connectors::Airwallex,
+    connectors::Amazonpay,
+    connectors::Archipel,
+    connectors::Authipay,
+    connectors::Authorizedotnet,
+    connectors::Bambora,
+    connectors::Bamboraapac,
+    connectors::Bankofamerica,
+    connectors::Barclaycard,
+    connectors::Billwerk,
+    connectors::Bluesnap,
+    connectors::Bitpay,
+    connectors::Blackhawknetwork,
+    connectors::Bluecode,
+    connectors::Braintree,
+    connectors::Boku,
+    connectors::Breadpay,
+    connectors::Cashtocode,
+    connectors::Celero,
+    connectors::Checkbook,
+    connectors::Checkout,
+    connectors::Coinbase,
+    connectors::Coingate,
+    connectors::Cryptopay,
+    connectors::CtpMastercard,
+    connectors::Custombilling,
+    connectors::Cybersource,
+    connectors::Datatrans,
+    connectors::Deutschebank,
+    connectors::Digitalvirgo,
+    connectors::Dlocal,
+    connectors::Dwolla,
+    connectors::Ebanx,
+    connectors::Elavon,
+    connectors::Facilitapay,
+    connectors::Fiserv,
+    connectors::Fiservemea,
+    connectors::Fiuu,
+    connectors::Flexiti,
+    connectors::Forte,
+    connectors::Getnet,
+    connectors::Globalpay,
+    connectors::Globepay,
+    connectors::Gocardless,
+    connectors::Gpayments,
+    connectors::Helcim,
+    connectors::Hipay,
+    connectors::HyperswitchVault,
+    connectors::Hyperwallet,
+    connectors::Iatapay,
+    connectors::Inespay,
+    connectors::Itaubank,
+    connectors::Jpmorgan,
+    connectors::Juspaythreedsserver,
+    connectors::Katapult,
+    connectors::Klarna,
+    connectors::Netcetera,
+    connectors::Nmi,
+    connectors::Nomupay,
+    connectors::Noon,
+    connectors::Nordea,
+    connectors::Novalnet,
+    connectors::Nexinets,
+    connectors::Nexixpay,
+    connectors::Nuvei,
+    connectors::Opayo,
+    connectors::Opennode,
+    connectors::Payeezy,
+    connectors::Payload,
+    connectors::Paystack,
+    connectors::Paytm,
+    connectors::Payu,
+    connectors::Phonepe,
+    connectors::Paypal,
+    connectors::Powertranz,
+    connectors::Prophetpay,
+    connectors::Mifinity,
+    connectors::Mollie,
+    connectors::Moneris,
+    connectors::Mpgs,
+    connectors::Multisafepay,
+    connectors::Paybox,
+    connectors::Payme,
+    connectors::Payone,
+    connectors::Placetopay,
+    connectors::Plaid,
+    connectors::Rapyd,
+    connectors::Razorpay,
+    connectors::Recurly,
+    connectors::Redsys,
+    connectors::Riskified,
+    connectors::Santander,
+    connectors::Shift4,
+    connectors::Sift,
+    connectors::Silverflow,
+    connectors::Signifyd,
+    connectors::Stripe,
+    connectors::Stripebilling,
+    connectors::Stax,
+    connectors::Square,
+    connectors::Taxjar,
+    connectors::Threedsecureio,
+    connectors::Thunes,
+    connectors::Tokenio,
+    connectors::Trustpay,
+    connectors::Trustpayments,
+    connectors::Tsys,
+    connectors::UnifiedAuthenticationService,
+    connectors::Wise,
+    connectors::Worldline,
+    connectors::Worldpay,
+    connectors::Worldpayvantiv,
+    connectors::Worldpayxml,
+    connectors::Wellsfargo,
+    connectors::Wellsfargopayout,
+    connectors::Volt,
+    connectors::Xendit,
+    connectors::Zen,
+    connectors::Zsl
+);
 macro_rules! default_imp_for_external_vault {
     ($($path:ident::$connector:ident),*) => {
         $(
@@ -7888,6 +8177,148 @@ default_imp_for_external_vault_proxy_payments_create!(
     connectors::Zsl,
     connectors::CtpMastercard
 );
+#[cfg(feature = "v1")]
+macro_rules! default_imp_for_subscription_create {
+    ($($path:ident::$connector:ident),*) => {
+        $( impl SubscriptionCreate for $path::$connector {}
+            impl
+            ConnectorIntegration<
+            SubscriptionCreateFlow,
+            SubscriptionCreateRequest,
+            SubscriptionCreateResponse,
+            > for $path::$connector
+            {}
+        )*
+    };
+}
+
+#[cfg(feature = "v1")]
+default_imp_for_subscription_create!(
+    connectors::Vgs,
+    connectors::Aci,
+    connectors::Adyen,
+    connectors::Adyenplatform,
+    connectors::Affirm,
+    connectors::Airwallex,
+    connectors::Amazonpay,
+    connectors::Archipel,
+    connectors::Authipay,
+    connectors::Authorizedotnet,
+    connectors::Bambora,
+    connectors::Bamboraapac,
+    connectors::Bankofamerica,
+    connectors::Barclaycard,
+    connectors::Billwerk,
+    connectors::Bluesnap,
+    connectors::Bitpay,
+    connectors::Blackhawknetwork,
+    connectors::Bluecode,
+    connectors::Braintree,
+    connectors::Boku,
+    connectors::Breadpay,
+    connectors::Cashtocode,
+    connectors::Celero,
+    connectors::Checkbook,
+    connectors::Checkout,
+    connectors::Coinbase,
+    connectors::Coingate,
+    connectors::Cryptopay,
+    connectors::CtpMastercard,
+    connectors::Custombilling,
+    connectors::Cybersource,
+    connectors::Datatrans,
+    connectors::Deutschebank,
+    connectors::Digitalvirgo,
+    connectors::Dlocal,
+    connectors::Dwolla,
+    connectors::Ebanx,
+    connectors::Elavon,
+    connectors::Facilitapay,
+    connectors::Fiserv,
+    connectors::Fiservemea,
+    connectors::Fiuu,
+    connectors::Flexiti,
+    connectors::Forte,
+    connectors::Getnet,
+    connectors::Globalpay,
+    connectors::Globepay,
+    connectors::Gocardless,
+    connectors::Gpayments,
+    connectors::Helcim,
+    connectors::Hipay,
+    connectors::HyperswitchVault,
+    connectors::Hyperwallet,
+    connectors::Iatapay,
+    connectors::Inespay,
+    connectors::Itaubank,
+    connectors::Jpmorgan,
+    connectors::Juspaythreedsserver,
+    connectors::Katapult,
+    connectors::Klarna,
+    connectors::Netcetera,
+    connectors::Nmi,
+    connectors::Nomupay,
+    connectors::Noon,
+    connectors::Nordea,
+    connectors::Novalnet,
+    connectors::Nexinets,
+    connectors::Nexixpay,
+    connectors::Nuvei,
+    connectors::Opayo,
+    connectors::Opennode,
+    connectors::Payeezy,
+    connectors::Payload,
+    connectors::Paystack,
+    connectors::Paytm,
+    connectors::Payu,
+    connectors::Phonepe,
+    connectors::Paypal,
+    connectors::Powertranz,
+    connectors::Prophetpay,
+    connectors::Mifinity,
+    connectors::Mollie,
+    connectors::Moneris,
+    connectors::Mpgs,
+    connectors::Multisafepay,
+    connectors::Paybox,
+    connectors::Payme,
+    connectors::Payone,
+    connectors::Placetopay,
+    connectors::Plaid,
+    connectors::Rapyd,
+    connectors::Razorpay,
+    connectors::Recurly,
+    connectors::Redsys,
+    connectors::Riskified,
+    connectors::Santander,
+    connectors::Shift4,
+    connectors::Sift,
+    connectors::Silverflow,
+    connectors::Signifyd,
+    connectors::Stripe,
+    connectors::Stripebilling,
+    connectors::Stax,
+    connectors::Square,
+    connectors::Taxjar,
+    connectors::Threedsecureio,
+    connectors::Thunes,
+    connectors::Tokenio,
+    connectors::Trustpay,
+    connectors::Trustpayments,
+    connectors::Tsys,
+    connectors::UnifiedAuthenticationService,
+    connectors::Wise,
+    connectors::Worldline,
+    connectors::Worldpay,
+    connectors::Worldpayvantiv,
+    connectors::Worldpayxml,
+    connectors::Wellsfargo,
+    connectors::Wellsfargopayout,
+    connectors::Volt,
+    connectors::Xendit,
+    connectors::Zen,
+    connectors::Zsl
+);
 
 #[cfg(feature = "dummy_connector")]
 impl<const T: u8> PaymentsCompleteAuthorize for connectors::DummyConnector<T> {}
@@ -8333,6 +8764,36 @@ impl<const T: u8>
 #[cfg(feature = "dummy_connector")]
 impl<const T: u8> RevenueRecovery for connectors::DummyConnector<T> {}
 
+#[cfg(feature = "dummy_connector")]
+impl<const T: u8> SubscriptionRecordBack for connectors::DummyConnector<T> {}
+
+#[cfg(feature = "dummy_connector")]
+impl<const T: u8>
+    ConnectorIntegration<
+        SubscriptionRecordBackFlow,
+        SubscriptionsRecordBackRequest,
+        RevenueRecoveryRecordBackResponse,
+    > for connectors::DummyConnector<T>
+{
+}
+
+#[cfg(feature = "dummy_connector")]
+impl<const T: u8> SubscriptionCreate for connectors::DummyConnector<T> {}
+
+#[cfg(feature = "dummy_connector")]
+impl<const T: u8>
+    ConnectorIntegration<
+        SubscriptionCreateFlow,
+        SubscriptionCreateRequest,
+        SubscriptionCreateResponse,
+    > for connectors::DummyConnector<T>
+{
+}
+
+
+#[cfg(feature = "dummy_connector")]
+impl<const T: u8> Subscriptions for connectors::DummyConnector<T> {}
+
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
 #[cfg(feature = "dummy_connector")]
 impl<const T: u8> api::revenue_recovery::BillingConnectorPaymentsSyncIntegration
diff --git a/crates/hyperswitch_domain_models/src/payment_methods.rs b/crates/hyperswitch_domain_models/src/payment_methods.rs
index 0fa3d5b4e07..f94ee594fd1 100644
--- a/crates/hyperswitch_domain_models/src/payment_methods.rs
+++ b/crates/hyperswitch_domain_models/src/payment_methods.rs
@@ -863,6 +863,12 @@ pub trait PaymentMethodInterface {
         merchant_id: &id_type::MerchantId,
         payment_method_id: &str,
     ) -> CustomResult<PaymentMethod, Self::Error>;
+
+    #[cfg(feature = "v1")]
+    async fn find_payment_method_ids_by_billing_connector_subscription_id(
+        &self,
+        subscription_id: &str,
+    ) -> CustomResult<Vec<String>, Self::Error>;
 }
 
 #[cfg(feature = "v2")]
diff --git a/crates/hyperswitch_domain_models/src/payments.rs b/crates/hyperswitch_domain_models/src/payments.rs
index 58a859acbca..b69119f46f6 100644
--- a/crates/hyperswitch_domain_models/src/payments.rs
+++ b/crates/hyperswitch_domain_models/src/payments.rs
@@ -123,6 +123,7 @@ pub struct PaymentIntent {
     pub shipping_amount_tax: Option<MinorUnit>,
     pub duty_amount: Option<MinorUnit>,
     pub enable_partial_authorization: Option<bool>,
+    pub billing_processor_details: Option<api_models::payments::BillingConnectorDetails>,
 }
 
 impl PaymentIntent {
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
index 8dbdd1761bf..f9969604119 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
@@ -2105,6 +2105,9 @@ impl behaviour::Conversion for PaymentIntent {
             shipping_amount_tax: self.shipping_amount_tax,
             duty_amount: self.duty_amount,
             enable_partial_authorization: self.enable_partial_authorization,
+            billing_processor_details: self
+                .billing_processor_details
+                .and_then(|details| serde_json::to_value(details).ok()),
         })
     }
 
@@ -2213,6 +2216,9 @@ impl behaviour::Conversion for PaymentIntent {
                 duty_amount: storage_model.duty_amount,
                 order_date: storage_model.order_date,
                 enable_partial_authorization: storage_model.enable_partial_authorization,
+                billing_processor_details: storage_model
+                    .billing_processor_details
+                    .and_then(|details| serde_json::from_value(details).ok()),
             })
         }
         .await
@@ -2293,6 +2299,9 @@ impl behaviour::Conversion for PaymentIntent {
             shipping_amount_tax: self.shipping_amount_tax,
             duty_amount: self.duty_amount,
             enable_partial_authorization: self.enable_partial_authorization,
+            billing_processor_details: self
+                .billing_processor_details
+                .and_then(|details| serde_json::to_value(details).ok()),
         })
     }
 }
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 686912c6dca..0e0bb848eb1 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
@@ -150,6 +150,9 @@ pub struct FilesFlowData {
 #[derive(Debug, Clone)]
 pub struct RevenueRecoveryRecordBackData;
 
+#[derive(Debug, Clone)]
+pub struct SubscriptionCreateData;
+
 #[derive(Debug, Clone)]
 pub struct UasFlowData {
     pub authenticate_by: String,
diff --git a/crates/hyperswitch_domain_models/src/router_flow_types.rs b/crates/hyperswitch_domain_models/src/router_flow_types.rs
index 887babd5fca..5778148c8e5 100644
--- a/crates/hyperswitch_domain_models/src/router_flow_types.rs
+++ b/crates/hyperswitch_domain_models/src/router_flow_types.rs
@@ -8,6 +8,8 @@ pub mod payments;
 pub mod payouts;
 pub mod refunds;
 pub mod revenue_recovery;
+// #[cfg(feature = "v1")]
+pub mod subscriptions;
 pub mod unified_authentication_service;
 pub mod vault;
 pub mod webhooks;
@@ -20,6 +22,8 @@ pub use payments::*;
 pub use payouts::*;
 pub use refunds::*;
 pub use revenue_recovery::*;
+// #[cfg(feature = "v1")]
+pub use subscriptions::*;
 pub use unified_authentication_service::*;
 pub use vault::*;
 pub use webhooks::*;
diff --git a/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs b/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs
new file mode 100644
index 00000000000..2c147a3d4ae
--- /dev/null
+++ b/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs
@@ -0,0 +1,9 @@
+#[derive(Debug, Clone)]
+pub struct GetSubscriptionPlans;
+//! Subscriptions flow types for V1
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionRecordBack;
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionCreate;
diff --git a/crates/hyperswitch_domain_models/src/router_request_types.rs b/crates/hyperswitch_domain_models/src/router_request_types.rs
index a9a842dcfd5..fe61b589b0e 100644
--- a/crates/hyperswitch_domain_models/src/router_request_types.rs
+++ b/crates/hyperswitch_domain_models/src/router_request_types.rs
@@ -1,7 +1,9 @@
 pub mod authentication;
 pub mod fraud_check;
 pub mod revenue_recovery;
+pub mod subscriptions;
 pub mod unified_authentication_service;
+pub mod subscriptions;
 use api_models::payments::{AdditionalPaymentData, RequestSurchargeDetails};
 use common_types::payments as common_payments_types;
 use common_utils::{consts, errors, ext_traits::OptionExt, id_type, pii, types::MinorUnit};
diff --git a/crates/hyperswitch_domain_models/src/router_request_types/subscriptions.rs b/crates/hyperswitch_domain_models/src/router_request_types/subscriptions.rs
new file mode 100644
index 00000000000..d84782b52f1
--- /dev/null
+++ b/crates/hyperswitch_domain_models/src/router_request_types/subscriptions.rs
@@ -0,0 +1,33 @@
+#[derive(Debug, Clone)]
+pub struct GetSubscriptionPlansRequest;
+use api_models::payments::Address;
+use common_enums::enums;
+
+use crate::connector_endpoints;
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionsRecordBackRequest {
+    pub merchant_reference_id: String,
+    pub amount: common_utils::types::MinorUnit,
+    pub currency: enums::Currency,
+    pub payment_method_type: Option<common_enums::PaymentMethodType>,
+    pub attempt_status: common_enums::AttemptStatus,
+    pub connector_transaction_id: Option<common_utils::types::ConnectorTransactionId>,
+    pub connector_params: connector_endpoints::ConnectorParams,
+}
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionItem {
+    pub item_price_id: String,
+    pub quantity: Option<u32>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionCreateRequest {
+    pub customer_id: String,
+    pub subscription_id: String,
+    pub subscription_items: Vec<SubscriptionItem>,
+    pub billing_address: Address,
+    pub auto_collection: String,
+    pub connector_params: connector_endpoints::ConnectorParams,
+}
diff --git a/crates/hyperswitch_domain_models/src/router_response_types.rs b/crates/hyperswitch_domain_models/src/router_response_types.rs
index 17c268fb60c..cc766a00ba8 100644
--- a/crates/hyperswitch_domain_models/src/router_response_types.rs
+++ b/crates/hyperswitch_domain_models/src/router_response_types.rs
@@ -1,6 +1,7 @@
 pub mod disputes;
 pub mod fraud_check;
 pub mod revenue_recovery;
+pub mod subscriptions;
 use std::collections::HashMap;
 
 use common_utils::{pii, request::Method, types::MinorUnit};
diff --git a/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs b/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
new file mode 100644
index 00000000000..a6bfcfe68fe
--- /dev/null
+++ b/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
@@ -0,0 +1,25 @@
+#[derive(Debug, Clone)]
+pub struct GetSubscriptionPlansResponse {
+    pub list: Vec<SubscriptionPlans>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionPlans {
+    pub subscription_provider_plan_id: String,
+    pub name: String,
+    pub description: Option<String>,
+use common_enums::enums;
+use common_utils::types::MinorUnit;
+use time::PrimitiveDateTime;
+
+#[derive(Debug, Clone)]
+pub struct SubscriptionCreateResponse {
+    pub subscription_id: String,
+    pub invoice_id: String,
+    pub status: String,
+    pub customer_id: String,
+    pub currency_code: enums::Currency,
+    pub total_amount: MinorUnit,
+    pub next_billing_at: Option<PrimitiveDateTime>,
+    pub created_at: Option<PrimitiveDateTime>,
+}
diff --git a/crates/hyperswitch_domain_models/src/types.rs b/crates/hyperswitch_domain_models/src/types.rs
index bcca0ba12a9..43d200d68ee 100644
--- a/crates/hyperswitch_domain_models/src/types.rs
+++ b/crates/hyperswitch_domain_models/src/types.rs
@@ -4,19 +4,22 @@ use crate::{
     router_data::{AccessToken, AccessTokenAuthenticationResponse, RouterData},
     router_data_v2::{self, RouterDataV2},
     router_flow_types::{
-        mandate_revoke::MandateRevoke, revenue_recovery::RecoveryRecordBack, AccessTokenAuth,
-        AccessTokenAuthentication, Authenticate, AuthenticationConfirmation, Authorize,
-        AuthorizeSessionToken, BillingConnectorInvoiceSync, BillingConnectorPaymentsSync,
-        CalculateTax, Capture, CompleteAuthorize, CreateConnectorCustomer, CreateOrder, Execute,
-        ExternalVaultProxy, IncrementalAuthorization, PSync, PaymentMethodToken, PostAuthenticate,
-        PostCaptureVoid, PostSessionTokens, PreAuthenticate, PreProcessing, RSync,
-        SdkSessionUpdate, Session, SetupMandate, UpdateMetadata, VerifyWebhookSource, Void,
+        mandate_revoke::MandateRevoke, revenue_recovery::RecoveryRecordBack,
+        subscriptions::GetSubscriptionPlans, AccessTokenAuth, AccessTokenAuthentication,
+        Authenticate, AuthenticationConfirmation, Authorize, AuthorizeSessionToken,
+        BillingConnectorInvoiceSync, BillingConnectorPaymentsSync, CalculateTax, Capture,
+        CompleteAuthorize, CreateConnectorCustomer, CreateOrder, Execute, ExternalVaultProxy,
+        IncrementalAuthorization, PSync, PaymentMethodToken, PostAuthenticate, PostCaptureVoid,
+        PostSessionTokens, PreAuthenticate, PreProcessing, RSync, SdkSessionUpdate, Session,
+        SetupMandate, UpdateMetadata, VerifyWebhookSource, Void,
     },
     router_request_types::{
         revenue_recovery::{
             BillingConnectorInvoiceSyncRequest, BillingConnectorPaymentsSyncRequest,
             RevenueRecoveryRecordBackRequest,
         },
+        subscriptions::GetSubscriptionPlansRequest,
+        subscriptions::{SubscriptionsRecordBackRequest, SubscriptionCreateRequest},
         unified_authentication_service::{
             UasAuthenticationRequestData, UasAuthenticationResponseData,
             UasConfirmationRequestData, UasPostAuthenticationRequestData,
@@ -37,6 +40,8 @@ use crate::{
             BillingConnectorInvoiceSyncResponse, BillingConnectorPaymentsSyncResponse,
             RevenueRecoveryRecordBackResponse,
         },
+        subscriptions::GetSubscriptionPlansResponse,
+        subscriptions::SubscriptionCreateResponse,
         MandateRevokeResponseData, PaymentsResponseData, RefundsResponseData,
         TaxCalculationResponseData, VaultResponseData, VerifyWebhookSourceResponseData,
     },
@@ -44,6 +49,9 @@ use crate::{
 #[cfg(feature = "payouts")]
 pub use crate::{router_request_types::PayoutsData, router_response_types::PayoutsResponseData};
 
+#[cfg(feature = "v1")]
+use crate::router_flow_types::subscriptions::{SubscriptionRecordBack, SubscriptionCreate};
+
 pub type PaymentsAuthorizeRouterData =
     RouterData<Authorize, PaymentsAuthorizeData, PaymentsResponseData>;
 pub type ExternalVaultProxyPaymentsRouterData =
@@ -120,6 +128,9 @@ pub type RevenueRecoveryRecordBackRouterData = RouterData<
     RevenueRecoveryRecordBackResponse,
 >;
 
+pub type GetSubscriptionPlansRouterData =
+    RouterData<GetSubscriptionPlans, GetSubscriptionPlansRequest, GetSubscriptionPlansResponse>;
+
 pub type UasAuthenticationRouterData =
     RouterData<Authenticate, UasAuthenticationRequestData, UasAuthenticationResponseData>;
 
@@ -171,3 +182,17 @@ pub type ExternalVaultProxyPaymentsRouterDataV2 = RouterDataV2<
     ExternalVaultProxyPaymentsData,
     PaymentsResponseData,
 >;
+
+#[cfg(feature = "v1")]
+pub type SubscriptionRecordBackRouterData = RouterData<
+    SubscriptionRecordBack,
+    SubscriptionsRecordBackRequest,
+    RevenueRecoveryRecordBackResponse,
+>;
+
+#[cfg(feature = "v1")]
+pub type SubscriptionCreateRouterData = RouterData<
+    SubscriptionCreate,
+    SubscriptionCreateRequest,
+    SubscriptionCreateResponse,
+>;
diff --git a/crates/hyperswitch_interfaces/src/api.rs b/crates/hyperswitch_interfaces/src/api.rs
index c4d427a8f9a..013473ed7f7 100644
--- a/crates/hyperswitch_interfaces/src/api.rs
+++ b/crates/hyperswitch_interfaces/src/api.rs
@@ -22,6 +22,8 @@ pub mod refunds;
 pub mod refunds_v2;
 pub mod revenue_recovery;
 pub mod revenue_recovery_v2;
+pub mod subscriptions_v2;
+pub mod subscriptions;
 pub mod vault;
 pub mod vault_v2;
 
@@ -104,6 +106,7 @@ pub trait Connector:
     + UnifiedAuthenticationService
     + revenue_recovery::RevenueRecovery
     + ExternalVault
+    + subscriptions::Subscriptions
 {
 }
 
@@ -126,7 +129,8 @@ impl<
             + TaxCalculation
             + UnifiedAuthenticationService
             + revenue_recovery::RevenueRecovery
-            + ExternalVault,
+            + ExternalVault
+            + subscriptions::Subscriptions,
     > Connector for T
 {
 }
diff --git a/crates/hyperswitch_interfaces/src/api/revenue_recovery.rs b/crates/hyperswitch_interfaces/src/api/revenue_recovery.rs
index 053d3cdde35..2041dbd73df 100644
--- a/crates/hyperswitch_interfaces/src/api/revenue_recovery.rs
+++ b/crates/hyperswitch_interfaces/src/api/revenue_recovery.rs
@@ -13,9 +13,8 @@ use hyperswitch_domain_models::{
         RevenueRecoveryRecordBackResponse,
     },
 };
-
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
-use super::ConnectorCommon;
+ use crate::api::ConnectorCommon;
 use super::ConnectorIntegration;
 
 /// trait RevenueRecovery
@@ -28,6 +27,7 @@ pub trait RevenueRecovery:
 {
 }
 
+
 /// trait BillingConnectorPaymentsSyncIntegration
 pub trait BillingConnectorPaymentsSyncIntegration:
     ConnectorIntegration<
@@ -48,6 +48,8 @@ pub trait RevenueRecoveryRecordBack:
 {
 }
 
+
+
 /// trait BillingConnectorInvoiceSyncIntegration
 pub trait BillingConnectorInvoiceSyncIntegration:
     ConnectorIntegration<
diff --git a/crates/hyperswitch_interfaces/src/api/subscriptions.rs b/crates/hyperswitch_interfaces/src/api/subscriptions.rs
new file mode 100644
index 00000000000..ce9c3e7ab0a
--- /dev/null
+++ b/crates/hyperswitch_interfaces/src/api/subscriptions.rs
@@ -0,0 +1,57 @@
+//! Subscriptions Interface for V1
+
+use hyperswitch_domain_models::{
+    router_flow_types::subscriptions::{
+        SubscriptionRecordBack as SubscriptionRecordBackFlow,
+        SubscriptionCreate as SubscriptionCreateFlow,
+    },
+    router_request_types::subscriptions::{SubscriptionsRecordBackRequest, SubscriptionCreateRequest},
+    router_response_types::{
+        revenue_recovery::RevenueRecoveryRecordBackResponse,
+        subscriptions::SubscriptionCreateResponse,
+    },
+};
+
+use super::{ConnectorCommon, ConnectorIntegration};
+
+#[cfg(feature = "v1")]
+/// trait SubscriptionRecordBack for V1
+pub trait SubscriptionRecordBack:
+    ConnectorIntegration<
+        SubscriptionRecordBackFlow,
+        SubscriptionsRecordBackRequest,
+        RevenueRecoveryRecordBackResponse,
+    >
+{
+}
+
+#[cfg(feature = "v1")]
+/// trait SubscriptionCreate for V1
+pub trait SubscriptionCreate:
+    ConnectorIntegration<
+        SubscriptionCreateFlow,
+        SubscriptionCreateRequest,
+        SubscriptionCreateResponse,
+    >
+{
+}
+
+/// trait Subscriptions 
+#[cfg(feature = "v1")]
+pub trait Subscriptions:
+    ConnectorCommon
+    + SubscriptionRecordBack
+    + SubscriptionCreate
+{
+}
+
+#[cfg(not(feature = "v1"))]
+/// trait SubscriptionRecordBack (disabled when not V1)
+pub trait SubscriptionRecordBack {}
+
+/// trait Subscriptions (disabled when not V1)
+#[cfg(not(feature = "v1"))]
+pub trait Subscriptions {}
+
+#[cfg(not(feature = "v1"))]
+pub trait SubscriptionCreate {}
diff --git a/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs b/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs
new file mode 100644
index 00000000000..fdad445be07
--- /dev/null
+++ b/crates/hyperswitch_interfaces/src/api/subscriptions_v2.rs
@@ -0,0 +1,43 @@
+//! SubscriptionsV2
+use hyperswitch_domain_models::{
+    router_data_v2::flow_common_types::{
+        RevenueRecoveryRecordBackData, SubscriptionCreateData,
+    },
+    router_flow_types::{
+        SubscriptionCreate, SubscriptionRecordBack,
+    },
+    router_request_types::subscriptions::{
+        SubscriptionCreateRequest, SubscriptionsRecordBackRequest,
+    },
+    router_response_types::{
+        revenue_recovery::RevenueRecoveryRecordBackResponse,
+        subscriptions::SubscriptionCreateResponse,
+    },
+};
+
+use crate::connector_integration_v2::ConnectorIntegrationV2;
+
+/// trait SubscriptionsV2
+pub trait SubscriptionsV2: SubscriptionsRecordBackV2 + SubscriptionsCreateV2 {}
+
+/// trait SubscriptionsRecordBackV2
+pub trait SubscriptionsRecordBackV2:
+    ConnectorIntegrationV2<
+    SubscriptionRecordBack,
+    RevenueRecoveryRecordBackData,
+    SubscriptionsRecordBackRequest,
+    RevenueRecoveryRecordBackResponse,
+>
+{
+}
+
+/// trait SubscriptionsCreateV2
+pub trait SubscriptionsCreateV2:
+    ConnectorIntegrationV2<
+    SubscriptionCreate,
+    SubscriptionCreateData,
+    SubscriptionCreateRequest,
+    SubscriptionCreateResponse,
+>
+{
+}
diff --git a/crates/hyperswitch_interfaces/src/connector_integration_v2.rs b/crates/hyperswitch_interfaces/src/connector_integration_v2.rs
index aa4bc72f610..73df8a76611 100644
--- a/crates/hyperswitch_interfaces/src/connector_integration_v2.rs
+++ b/crates/hyperswitch_interfaces/src/connector_integration_v2.rs
@@ -35,6 +35,7 @@ pub trait ConnectorV2:
     + api::UnifiedAuthenticationServiceV2
     + api::revenue_recovery_v2::RevenueRecoveryV2
     + api::ExternalVaultV2
+    + api::subscriptions_v2::SubscriptionsV2
 {
 }
 impl<
@@ -55,7 +56,8 @@ impl<
             + api::authentication_v2::ExternalAuthenticationV2
             + api::UnifiedAuthenticationServiceV2
             + api::revenue_recovery_v2::RevenueRecoveryV2
-            + api::ExternalVaultV2,
+            + api::ExternalVaultV2
+            + api::subscriptions_v2::SubscriptionsV2,
     > ConnectorV2 for T
 {
 }
diff --git a/crates/hyperswitch_interfaces/src/conversion_impls.rs b/crates/hyperswitch_interfaces/src/conversion_impls.rs
index 23fb9e76bcd..b9fe040f97c 100644
--- a/crates/hyperswitch_interfaces/src/conversion_impls.rs
+++ b/crates/hyperswitch_interfaces/src/conversion_impls.rs
@@ -12,7 +12,7 @@ use hyperswitch_domain_models::{
             AccessTokenFlowData, AuthenticationTokenFlowData, BillingConnectorInvoiceSyncFlowData,
             BillingConnectorPaymentsSyncFlowData, DisputesFlowData, ExternalAuthenticationFlowData,
             ExternalVaultProxyFlowData, FilesFlowData, MandateRevokeFlowData, PaymentFlowData,
-            RefundFlowData, RevenueRecoveryRecordBackData, UasFlowData, VaultConnectorFlowData,
+            RefundFlowData, RevenueRecoveryRecordBackData, SubscriptionCreateData, UasFlowData, VaultConnectorFlowData,
             WebhookSourceVerifyData,
         },
         RouterDataV2,
@@ -798,6 +798,46 @@ impl<T, Req: Clone, Resp: Clone> RouterDataConversion<T, Req, Resp>
     }
 }
 
+impl<T, Req: Clone, Resp: Clone> RouterDataConversion<T, Req, Resp>
+    for SubscriptionCreateData
+{
+    fn from_old_router_data(
+        old_router_data: &RouterData<T, Req, Resp>,
+    ) -> CustomResult<RouterDataV2<T, Self, Req, Resp>, ConnectorError>
+    where
+        Self: Sized,
+    {
+        let resource_common_data = Self {};
+        Ok(RouterDataV2 {
+            flow: std::marker::PhantomData,
+            tenant_id: old_router_data.tenant_id.clone(),
+            resource_common_data,
+            connector_auth_type: old_router_data.connector_auth_type.clone(),
+            request: old_router_data.request.clone(),
+            response: old_router_data.response.clone(),
+        })
+    }
+
+    fn to_old_router_data(
+        new_router_data: RouterDataV2<T, Self, Req, Resp>,
+    ) -> CustomResult<RouterData<T, Req, Resp>, ConnectorError>
+    where
+        Self: Sized,
+    {
+        let router_data = get_default_router_data(
+            new_router_data.tenant_id.clone(),
+            "subscription_create",
+            new_router_data.request,
+            new_router_data.response,
+        );
+        Ok(RouterData {
+            connector_auth_type: new_router_data.connector_auth_type.clone(),
+            ..router_data
+        })
+    }
+}
+
+
 impl<T, Req: Clone, Resp: Clone> RouterDataConversion<T, Req, Resp> for UasFlowData {
     fn from_old_router_data(
         old_router_data: &RouterData<T, Req, Resp>,
diff --git a/crates/hyperswitch_interfaces/src/types.rs b/crates/hyperswitch_interfaces/src/types.rs
index eed89fe00ba..b74eb51d297 100644
--- a/crates/hyperswitch_interfaces/src/types.rs
+++ b/crates/hyperswitch_interfaces/src/types.rs
@@ -16,6 +16,7 @@ use hyperswitch_domain_models::{
         },
         refunds::{Execute, RSync},
         revenue_recovery::{BillingConnectorPaymentsSync, RecoveryRecordBack},
+        subscriptions::GetSubscriptionPlans,
         unified_authentication_service::{
             Authenticate, AuthenticationConfirmation, PostAuthenticate, PreAuthenticate,
         },
@@ -31,6 +32,8 @@ use hyperswitch_domain_models::{
             BillingConnectorInvoiceSyncRequest, BillingConnectorPaymentsSyncRequest,
             RevenueRecoveryRecordBackRequest,
         },
+        subscriptions::GetSubscriptionPlansRequest,
+        subscriptions::{SubscriptionsRecordBackRequest, SubscriptionCreateRequest},
         unified_authentication_service::{
             UasAuthenticationRequestData, UasAuthenticationResponseData,
             UasConfirmationRequestData, UasPostAuthenticationRequestData,
@@ -53,6 +56,7 @@ use hyperswitch_domain_models::{
             BillingConnectorInvoiceSyncResponse, BillingConnectorPaymentsSyncResponse,
             RevenueRecoveryRecordBackResponse,
         },
+        subscriptions::GetSubscriptionPlansResponse,
         AcceptDisputeResponse, DefendDisputeResponse, DisputeSyncResponse, FetchDisputesResponse,
         MandateRevokeResponseData, PaymentsResponseData, RefundsResponseData, RetrieveFileResponse,
         SubmitEvidenceResponse, TaxCalculationResponseData, UploadFileResponse, VaultResponseData,
@@ -271,6 +275,22 @@ pub type RevenueRecoveryRecordBackType = dyn ConnectorIntegration<
     RevenueRecoveryRecordBackResponse,
 >;
 
+#[cfg(feature = "v1")]
+/// Type alias for `ConnectorIntegration<SubscriptionRecordBack, SubscriptionsRecordBackRequest, RevenueRecoveryRecordBackResponse>`
+pub type SubscriptionRecordBackType = dyn ConnectorIntegration<
+    hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+    SubscriptionsRecordBackRequest,
+    RevenueRecoveryRecordBackResponse,
+>;
+
+#[cfg(feature = "v1")]
+/// Type alias for `ConnectorIntegration<SubscriptionCreate, SubscriptionCreateRequest, SubscriptionCreateResponse>`
+pub type SubscriptionCreateType = dyn ConnectorIntegration<
+    hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+    SubscriptionCreateRequest,
+    hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse,
+>;
+
 /// Type alias for `ConnectorIntegration<BillingConnectorPaymentsSync, BillingConnectorPaymentsSyncRequest, BillingConnectorPaymentsSyncResponse>`
 pub type BillingConnectorPaymentsSyncType = dyn ConnectorIntegration<
     BillingConnectorPaymentsSync,
@@ -309,6 +329,29 @@ pub type BillingConnectorInvoiceSyncTypeV2 = dyn ConnectorIntegrationV2<
     BillingConnectorInvoiceSyncResponse,
 >;
 
+/// Type alias for `ConnectorIntegration<GetSubscriptionPlans, GetSubscriptionPlansRequest, GetSubscriptionPlansResponse>`
+pub type GetSubscriptionPlansType = dyn ConnectorIntegration<
+    GetSubscriptionPlans,
+    GetSubscriptionPlansRequest,
+    GetSubscriptionPlansResponse,
+#[cfg(feature = "v1")]
+/// Type alias for `ConnectorIntegrationV2<SubscriptionRecordBack, SubscriptionCreateData, SubscriptionsRecordBackRequest, RevenueRecoveryRecordBackResponse>`
+pub type SubscriptionRecordBackTypeV2 = dyn ConnectorIntegrationV2<
+    hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+    flow_common_types::SubscriptionCreateData,
+    SubscriptionsRecordBackRequest,
+    RevenueRecoveryRecordBackResponse,
+>;
+
+#[cfg(feature = "v1")]
+/// Type alias for `ConnectorIntegrationV2<SubscriptionCreate, SubscriptionCreateData, SubscriptionCreateRequest, SubscriptionCreateResponse>`
+pub type SubscriptionCreateTypeV2 = dyn ConnectorIntegrationV2<
+    hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+    flow_common_types::SubscriptionCreateData,
+    SubscriptionCreateRequest,
+    hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse,
+>;
+
 /// Type alias for `ConnectorIntegration<ExternalVaultInsertFlow, VaultRequestData, VaultResponseData>`
 pub type ExternalVaultInsertType =
     dyn ConnectorIntegration<ExternalVaultInsertFlow, VaultRequestData, VaultResponseData>;
diff --git a/crates/router/src/bin/scheduler.rs b/crates/router/src/bin/scheduler.rs
index 492c07e5ab9..b5948c79b5a 100644
--- a/crates/router/src/bin/scheduler.rs
+++ b/crates/router/src/bin/scheduler.rs
@@ -330,6 +330,9 @@ impl ProcessTrackerWorkflows<routes::SessionState> for WorkflowRunner {
                 storage::ProcessTrackerRunner::PassiveRecoveryWorkflow => {
                     Ok(Box::new(workflows::revenue_recovery::ExecutePcrWorkflow))
                 }
+                storage::ProcessTrackerRunner::SubscriptionsWorkflow => Ok(Box::new(
+                    workflows::subscription::ExecuteSubscriptionWorkflow,
+                )),
             }
         };
 
diff --git a/crates/router/src/core.rs b/crates/router/src/core.rs
index 0fe3e57f872..fbfa0e6c56c 100644
--- a/crates/router/src/core.rs
+++ b/crates/router/src/core.rs
@@ -49,6 +49,8 @@ pub mod refunds_v2;
 #[cfg(feature = "v1")]
 pub mod debit_routing;
 pub mod routing;
+#[cfg(feature = "v1")]
+pub mod subscription;
 pub mod surcharge_decision_config;
 pub mod three_ds_decision_rule;
 #[cfg(feature = "olap")]
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index d4bf2c455d0..f21026c64b2 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -98,7 +98,7 @@ use crate::{
     },
 };
 #[cfg(feature = "v2")]
-use crate::{core::admin as core_admin, headers, types::ConnectorAuthType};
+use crate::{core::admin as core_admin, headers};
 #[cfg(feature = "v1")]
 use crate::{
     core::payment_methods::cards::create_encrypted_data, types::storage::CustomerUpdate::Update,
@@ -3899,6 +3899,7 @@ mod tests {
             shipping_amount_tax: None,
             duty_amount: None,
             enable_partial_authorization: None,
+            billing_processor_details: None,
         };
         let req_cs = Some("1".to_string());
         assert!(authenticate_client_secret(req_cs.as_ref(), &payment_intent).is_ok());
@@ -3983,6 +3984,7 @@ mod tests {
             shipping_amount_tax: None,
             duty_amount: None,
             enable_partial_authorization: None,
+            billing_processor_details: None,
         };
         let req_cs = Some("1".to_string());
         assert!(authenticate_client_secret(req_cs.as_ref(), &payment_intent,).is_err())
@@ -4065,6 +4067,7 @@ mod tests {
             shipping_amount_tax: None,
             duty_amount: None,
             enable_partial_authorization: None,
+            billing_processor_details: None,
         };
         let req_cs = Some("1".to_string());
         assert!(authenticate_client_secret(req_cs.as_ref(), &payment_intent).is_err())
@@ -7675,3 +7678,233 @@ pub async fn get_merchant_connector_account_v2(
         .attach_printable("merchant_connector_id is not provided"),
     }
 }
+#[cfg(feature = "v1")]
+pub fn create_subscription_router_data<F, Req, Res>(
+    state: &SessionState,
+    merchant_id: id_type::MerchantId,
+    customer_id: Option<id_type::CustomerId>,
+    connector_name: String,
+    auth_type: hyperswitch_domain_models::router_data::ConnectorAuthType,
+    request: Req,
+    payment_id: Option<id_type::PaymentId>,
+) -> CustomResult<RouterData<F, Req, Res>, errors::ApiErrorResponse>
+where
+    F: Clone,
+{
+    // Placeholder for future implementation
+
+    Ok(RouterData {
+        flow: std::marker::PhantomData,
+        merchant_id,
+        customer_id,
+        connector_customer: None,
+        connector: connector_name,
+        payment_id: payment_id
+            .map(|id| id.get_string_repr().to_owned())
+            .unwrap_or("DefaultPaymentId".to_string()),
+        tenant_id: state.tenant.tenant_id.clone(),
+        attempt_id: "Subscriptions attempt".to_owned(),
+        status: common_enums::AttemptStatus::default(),
+        payment_method: common_enums::PaymentMethod::default(),
+        connector_auth_type: auth_type,
+        description: None,
+        address: hyperswitch_domain_models::payment_address::PaymentAddress::default(),
+        auth_type: common_enums::AuthenticationType::default(),
+        connector_meta_data: None,
+        connector_wallets_details: None,
+        amount_captured: None,
+        minor_amount_captured: None,
+        access_token: None,
+        session_token: None,
+        reference_id: None,
+        payment_method_token: None,
+        recurring_mandate_payment_data: None,
+        preprocessing_id: None,
+        payment_method_balance: None,
+        connector_api_version: None,
+        request,
+        response: Err(ErrorResponse::default()),
+        connector_request_reference_id: "Notjing".to_owned(),
+        #[cfg(feature = "payouts")]
+        payout_method_data: None,
+        #[cfg(feature = "payouts")]
+        quote_id: None,
+        test_mode: None,
+        connector_http_status_code: None,
+        external_latency: None,
+        apple_pay_flow: None,
+        frm_metadata: None,
+        dispute_id: None,
+        refund_id: None,
+        payment_method_status: None,
+        connector_response: None,
+        integrity_check: Ok(()),
+        additional_merchant_data: None,
+        header_payload: None,
+        connector_mandate_request_reference_id: None,
+        authentication_id: None,
+        psd2_sca_exemption_type: None,
+        raw_connector_response: None,
+        is_payment_id_from_merchant: None,
+        l2_l3_data: None,
+        minor_amount_capturable: None,
+    })
+}
+#[cfg(feature = "v1")]
+pub async fn perform_billing_processor_record_back<F, D>(
+    state: &SessionState,
+    payment_data: &mut D,
+    key_store: &domain::MerchantKeyStore,
+) -> CustomResult<(), errors::ApiErrorResponse>
+where
+    F: Clone,
+    D: payments::OperationSessionGetters<F> + payments::OperationSessionSetters<F> + Send,
+{
+    logger::info!("perform_billing_processor_record_back");
+    let billing_processor_detail = payment_data
+        .get_payment_intent()
+        .billing_processor_details
+        .clone();
+    let attempt_status = payment_data.get_payment_attempt().status;
+    logger::info!("attempt_status is {:?}", attempt_status);
+
+    if billing_processor_detail.is_none() || !attempt_status.is_success() {
+        return Ok(());
+    }
+
+    let billing_processor_detail =
+        billing_processor_detail.ok_or(errors::ApiErrorResponse::GenericNotFoundError {
+            message: "billing_processor_detail not found in payment".to_string(),
+        })?;
+
+    let merchant_id = &payment_data.get_payment_intent().merchant_id;
+    let customer_id = payment_data
+        .get_payment_intent()
+        .customer_id
+        .clone()
+        .ok_or(errors::ApiErrorResponse::MissingRequiredField {
+            field_name: "customer_id",
+        })?;
+
+    let db = &*state.store;
+
+    // Fetch Subscriptions record from DB
+    db.find_by_merchant_id_subscription_id(
+        merchant_id,
+        billing_processor_detail.subscription_id.clone(),
+    )
+    .await
+    .change_context(errors::ApiErrorResponse::GenericNotFoundError {
+        message: format!(
+            "subscription not found for id: {}",
+            &billing_processor_detail.subscription_id
+        ),
+    })?;
+
+    // Fetch billing processor mca
+    let mca_id = billing_processor_detail.processor_mca.to_owned();
+    let billing_processor_mca = db
+        .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+            &state.into(),
+            &payment_data.get_payment_intent().merchant_id,
+            &mca_id,
+            key_store,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+            id: mca_id.get_string_repr().to_string(),
+        })?;
+
+    // Record back to billing processor
+
+    let auth_type = MerchantConnectorAccountType::DbVal(Box::new(billing_processor_mca.clone()))
+        .get_connector_account_details()
+        .parse_value("ConnectorAuthType")
+        .change_context(errors::ApiErrorResponse::InternalServerError)?;
+
+    let connector = &billing_processor_mca.connector_name;
+
+    let connector_data = api::ConnectorData::get_connector_by_name(
+        &state.conf.connectors,
+        &billing_processor_mca.connector_name,
+        api::GetToken::Connector,
+        Some(billing_processor_mca.get_id()),
+    )
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("invalid connector name received in billing merchant connector account")?;
+
+    let connector_enum = common_enums::connector_enums::Connector::from_str(connector.as_str())
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Cannot find connector from the connector_name")?;
+
+    let connector_params =
+        hyperswitch_domain_models::connector_endpoints::Connectors::get_connector_params(
+            &state.conf.connectors,
+            connector_enum,
+        )
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable(format!(
+            "cannot find connector params for this connector {connector} in this flow",
+        ))?;
+
+    let connector_integration: services::BoxedRevenueRecoveryRecordBackInterface<
+                        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+                        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+                        hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+                    > = connector_data.connector.get_connector_integration();
+
+    // let connector_integration_for_create_subscription: services::BoxedSubscriptionConnectorIntegrationInterface<
+    //                     hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+    //                     hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+    //                     hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse,
+    //                 > = connector_data.connector.get_connector_integration();
+
+    let request = hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest {
+        merchant_reference_id: billing_processor_detail.invoice_id.clone(),
+        amount: payment_data.get_payment_attempt().get_total_amount(),
+        currency: payment_data
+            .get_payment_intent()
+            .currency
+            .unwrap_or(common_enums::Currency::USD),
+        payment_method_type: payment_data.get_payment_attempt().payment_method_type,
+
+        attempt_status: payment_data.get_payment_attempt().status,
+        connector_transaction_id: payment_data
+            .get_payment_attempt()
+            .connector_transaction_id
+            .clone()
+            .map(|id| common_utils::types::ConnectorTransactionId::TxnId(id)),
+        connector_params,
+    };
+
+    let router_data = create_subscription_router_data::<
+        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionRecordBack,
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionsRecordBackRequest,
+        hyperswitch_domain_models::router_response_types::revenue_recovery::RevenueRecoveryRecordBackResponse,
+    >(
+        state,
+        merchant_id.clone(),
+        Some(customer_id),
+        connector.clone(),
+        auth_type,
+        request,
+        Some(payment_data
+            .get_payment_intent()
+            .payment_id
+            .to_owned())
+    )?;
+
+    services::execute_connector_processing_step(
+        state,
+        connector_integration,
+        &router_data,
+        common_enums::CallConnectorAction::Trigger,
+        None,
+        None,
+    )
+    .await
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed while handling response of record back to billing connector")?;
+
+    Ok(())
+}
diff --git a/crates/router/src/core/payments/operations/payment_create.rs b/crates/router/src/core/payments/operations/payment_create.rs
index 79bfbebf499..09f09a98837 100644
--- a/crates/router/src/core/payments/operations/payment_create.rs
+++ b/crates/router/src/core/payments/operations/payment_create.rs
@@ -1630,6 +1630,7 @@ impl PaymentCreate {
             tax_status: request.tax_status,
             shipping_amount_tax: request.shipping_amount_tax,
             enable_partial_authorization: request.enable_partial_authorization,
+            billing_processor_details: request.billing_processor_details.clone(),
         })
     }
 
diff --git a/crates/router/src/core/payments/operations/payment_response.rs b/crates/router/src/core/payments/operations/payment_response.rs
index a4ef1bdac82..51dceb30674 100644
--- a/crates/router/src/core/payments/operations/payment_response.rs
+++ b/crates/router/src/core/payments/operations/payment_response.rs
@@ -277,6 +277,46 @@ impl<F: Send + Clone> PostUpdateTracker<F, PaymentData<F>, types::PaymentsAuthor
             } else {
                 None
             };
+
+            if let Some(ref pm_id) = payment_method_id {
+                let billing_processor_detail = payment_data
+                    .payment_intent
+                    .billing_processor_details
+                    .clone();
+
+                if let Some(details) = billing_processor_detail {
+                    let merchant_id = &payment_data.payment_intent.merchant_id;
+                    // Update subscription record with the payment method id
+
+                    let subscription_record = state
+                        .store
+                        .find_by_merchant_id_subscription_id(
+                            merchant_id,
+                            details.subscription_id.clone(),
+                        )
+                        .await
+                        .change_context(errors::ApiErrorResponse::GenericNotFoundError {
+                            message: format!(
+                                "subscription not found for id: {}",
+                                &details.subscription_id
+                            ),
+                        })?;
+
+                    let udpate = storage::SubscriptionUpdate::new(Some(pm_id.clone()), None);
+
+                    state
+                        .store
+                        .update_subscription_entry(
+                            &subscription_record.merchant_id,
+                            subscription_record.subscription_id.clone(),
+                            udpate,
+                        )
+                        .await
+                        .change_context(errors::ApiErrorResponse::InternalServerError)
+                        .attach_printable("Failed to update subscription with payment method")?;
+                }
+            }
+
             payment_data.payment_attempt.payment_method_id = payment_method_id;
             payment_data.payment_attempt.connector_mandate_detail = connector_mandate_reference_id
                 .clone()
diff --git a/crates/router/src/core/payouts/helpers.rs b/crates/router/src/core/payouts/helpers.rs
index 8f104451be5..f3d10887de3 100644
--- a/crates/router/src/core/payouts/helpers.rs
+++ b/crates/router/src/core/payouts/helpers.rs
@@ -756,7 +756,7 @@ pub(super) async fn get_or_create_customer_details(
 }
 
 #[cfg(feature = "v1")]
-pub(super) async fn get_or_create_customer_details(
+pub async fn get_or_create_customer_details(
     state: &SessionState,
     customer_details: &CustomerDetails,
     merchant_context: &domain::MerchantContext,
diff --git a/crates/router/src/core/subscription.rs b/crates/router/src/core/subscription.rs
new file mode 100644
index 00000000000..f2ae8dd9c7f
--- /dev/null
+++ b/crates/router/src/core/subscription.rs
@@ -0,0 +1,371 @@
+pub mod utils;
+use super::errors::{self, RouterResponse};
+use crate::{
+    core::{payments as payments_core, payouts::helpers},
+    routes::SessionState,
+    services::{api as service_api, logger},
+    types::api as api_types,
+};
+use api_models::enums as api_enums;
+use api_models::subscription::{
+    self as subscription_types, CreateSubscriptionResponse, Subscription, SubscriptionStatus,
+    SUBSCRIPTION_ID_PREFIX,
+};
+use common_types::payments::CustomerAcceptance;
+use common_utils::ext_traits::ValueExt;
+use common_utils::{events::ApiEventMetric, generate_id_with_default_len, id_type};
+use diesel_models::subscription::{SubscriptionNew, SubscriptionUpdate};
+use error_stack::ResultExt;
+use hyperswitch_domain_models::{api::ApplicationResponse, merchant_context::MerchantContext};
+use payment_methods::helpers::StorageErrorExt;
+use std::{num::NonZeroI64, str::FromStr};
+use utils::{get_customer_details_from_request, get_or_create_customer};
+
+pub async fn create_subscription(
+    state: SessionState,
+    merchant_context: MerchantContext,
+    request: subscription_types::CreateSubscriptionRequest,
+) -> RouterResponse<subscription_types::CreateSubscriptionResponse> {
+    let store = state.store.clone();
+    let db = store.as_ref();
+    let id = request.subscription_id.clone().unwrap_or(generate_id_with_default_len(SUBSCRIPTION_ID_PREFIX));
+    let subscription_details = Subscription::new(&id, SubscriptionStatus::Created, None);
+    let mut response = subscription_types::CreateSubscriptionResponse::new(
+        subscription_details,
+        merchant_context
+            .get_merchant_account()
+            .get_id()
+            .get_string_repr(),
+        request.mca_id.clone(),
+    );
+
+    let customer = get_customer_details_from_request(request.clone());
+    let customer_id = if customer.customer_id.is_some()
+        || customer.name.is_some()
+        || customer.email.is_some()
+        || customer.phone.is_some()
+        || customer.phone_country_code.is_some()
+    {
+        let customer = get_or_create_customer(state, request.customer, merchant_context.clone())
+            .await
+            .map_err(|e| e.change_context(errors::ApiErrorResponse::CustomerNotFound))
+            .attach_printable("subscriptions: unable to process customer")?;
+
+        let customer_table_response = match &customer {
+            ApplicationResponse::Json(inner) => {
+                Some(subscription_types::map_customer_resp_to_details(inner))
+            }
+            _ => None,
+        };
+        response.customer = customer_table_response;
+        response
+            .customer
+            .as_ref()
+            .and_then(|customer| customer.id.clone())
+    } else {
+        request.customer_id.clone()
+    }
+    .ok_or(errors::ApiErrorResponse::CustomerNotFound)
+    .attach_printable("subscriptions: unable to create a customer")?;
+
+    // If provided we can strore plan_id, coupon_code etc as metadata
+    let mut subscription = SubscriptionNew::new(
+        id,
+        SubscriptionStatus::Created.to_string(),
+        None,
+        None,
+        request.mca_id,
+        None,
+        merchant_context.get_merchant_account().get_id().clone(),
+        customer_id,
+        None,
+    );
+    response.client_secret = subscription.generate_and_set_client_secret();
+    db.insert_subscription_entry(subscription)
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::ResourceIdNotFound)
+        .attach_printable("subscriptions: unable to insert subscription entry to database")?;
+
+    Ok(ApplicationResponse::Json(response))
+}
+
+pub async fn confirm_subscription(
+    state: SessionState,
+    merchant_context: MerchantContext,
+    authentication_profile_id: Option<common_utils::id_type::ProfileId>,
+    request: ConfirmSubscriptionRequest,
+    subscription_id: String,
+) -> RouterResponse<ConfirmSubscriptionResponse> {
+    let db = state.store.as_ref();
+    // Fetch subscription from DB
+    let mercahnt_account = merchant_context.get_merchant_account();
+    let key_store = merchant_context.get_merchant_key_store();
+    let subscription = state
+        .store
+        .find_by_merchant_id_subscription_id(mercahnt_account.get_id(), subscription_id.clone())
+        .await
+        .change_context(errors::ApiErrorResponse::GenericNotFoundError {
+            message: format!("subscription not found for id: {}", subscription_id),
+        })?;
+
+    logger::debug!("fetched_subscription: {:?}", subscription);
+
+    let mca_id = subscription
+        .mca_id
+        .as_ref()
+        .map(|id| id_type::MerchantConnectorAccountId::wrap(id.clone()))
+        .transpose()
+        .change_context(errors::ApiErrorResponse::InternalServerError)?
+        .ok_or(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+            id: "No mca_id associated with this subscription".to_string(),
+        })?;
+
+    let billing_processor_mca = db
+        .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+            &(&state).into(),
+            &mercahnt_account.get_id(),
+            &mca_id,
+            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 connector_data = api_types::ConnectorData::get_connector_by_name(
+        &state.conf.connectors,
+        &connector_name,
+        api_types::GetToken::Connector,
+        Some(billing_processor_mca.get_id()),
+    )
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .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("Cannot find connector from the connector_name")?;
+
+    let connector_params =
+        hyperswitch_domain_models::connector_endpoints::Connectors::get_connector_params(
+            &state.conf.connectors,
+            connector_enum,
+        )
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable(format!(
+            "cannot find connector params for this connector {connector_name} in this flow",
+        ))?;
+
+    let connector_integration: service_api::BoxedSubscriptionConnectorIntegrationInterface<
+        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+        hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse,
+    > = connector_data.connector.get_connector_integration();
+
+    // Create Subscription at billing processor
+    let subscription_item =
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionItem {
+            item_price_id: request.item_price_id.ok_or(
+                errors::ApiErrorResponse::InvalidRequestData {
+                    message: "item_price_id is required".to_string(),
+                },
+            )?,
+            quantity: Some(1),
+        };
+
+    let conn_request =
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest {
+            customer_id: subscription.customer_id.get_string_repr().to_string(),
+            subscription_id: subscription.subscription_id.clone(),
+            subscription_items: vec![subscription_item],
+            billing_address: request.billing_address.clone().ok_or(
+                errors::ApiErrorResponse::InvalidRequestData {
+                    message: "Billing address is required".to_string(),
+                },
+            )?,
+            auto_collection: "off".to_string(),
+            connector_params,
+        };
+
+    logger::debug!("conn_request_customer: {:?}", conn_request.customer_id);
+
+    let auth_type = payments_core::helpers::MerchantConnectorAccountType::DbVal(Box::new(
+        billing_processor_mca.clone(),
+    ))
+    .get_connector_account_details()
+    .parse_value("ConnectorAuthType")
+    .change_context(errors::ApiErrorResponse::InternalServerError)?;
+
+    let router_data = payments_core::helpers::create_subscription_router_data::<
+        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+        hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse,
+    >(
+        &state,
+        subscription.merchant_id.to_owned(),
+        Some(subscription.customer_id.to_owned()),
+        connector_name,
+        auth_type,
+        conn_request,
+        None,
+    )?;
+
+    let response = service_api::execute_connector_processing_step::<
+        hyperswitch_domain_models::router_flow_types::subscriptions::SubscriptionCreate,
+        _,
+        hyperswitch_domain_models::router_request_types::subscriptions::SubscriptionCreateRequest,
+        hyperswitch_domain_models::router_response_types::subscriptions::SubscriptionCreateResponse,
+    >(
+        &state,
+        connector_integration,
+        &router_data,
+        common_enums::CallConnectorAction::Trigger,
+        None,
+        None,
+    )
+    .await
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed while handling response of subscription_create")?;
+
+    let connector_resp = response.response.map_err(|err| {
+        crate::logger::error!(?err);
+        errors::ApiErrorResponse::InternalServerError
+    })?;
+
+    crate::logger::debug!("connector_resp: {:?}", connector_resp);
+
+    // Form Payments Request with billing processor details
+    let billing_connector_details = api_models::payments::BillingConnectorDetails {
+        processor_mca: mca_id.to_owned(),
+        subscription_id: connector_resp.subscription_id.clone(),
+        invoice_id: connector_resp.invoice_id.clone(),
+    };
+
+    let mut payment_request = api_types::PaymentsRequest {
+        amount: Some(api_types::Amount::Value(
+            NonZeroI64::new(request.amount).unwrap(), // fix this
+        )),
+        currency: Some(request.currency),
+        customer_id: Some(subscription.customer_id.to_owned()),
+        merchant_id: Some(subscription.merchant_id.to_owned()),
+        billing_processor_details: Some(billing_connector_details),
+        confirm: Some(true),
+        setup_future_usage: request.payment_data.setup_future_usage,
+        payment_method: Some(request.payment_data.payment_method),
+        payment_method_type: request.payment_data.payment_method_type,
+        payment_method_data: Some(request.payment_data.payment_method_data),
+        customer_acceptance: request.payment_data.customer_acceptance,
+        ..Default::default()
+    };
+
+    if let Err(err) = crate::routes::payments::get_or_generate_payment_id(&mut payment_request) {
+        return Err(err.into());
+    }
+
+    // Call Payments Core
+    let payment_response = payments_core::payments_core::<
+        api_types::Authorize,
+        api_types::PaymentsResponse,
+        _,
+        _,
+        _,
+        payments_core::PaymentData<api_types::Authorize>,
+    >(
+        state.clone(),
+        state.get_req_state(),
+        merchant_context,
+        authentication_profile_id,
+        payments_core::PaymentCreate,
+        payment_request,
+        service_api::AuthFlow::Merchant,
+        payments_core::CallConnectorAction::Trigger,
+        None,
+        hyperswitch_domain_models::payments::HeaderPayload::with_source(
+            common_enums::PaymentSource::Webhook,
+        ),
+    )
+    .await;
+
+    // fix this error handling
+    let payment_res = match payment_response {
+        Ok(service_api::ApplicationResponse::JsonWithHeaders((pi, _))) => Ok(pi),
+        Ok(_) => Err(errors::ApiErrorResponse::InternalServerError),
+        Err(error) => {
+            crate::logger::error!(?error);
+            Err(errors::ApiErrorResponse::InternalServerError)
+        }
+    }?;
+    // Semd back response
+    let update = SubscriptionUpdate::new(None, Some(SubscriptionStatus::Active.to_string()));
+    db.update_subscription_entry(
+        &subscription.merchant_id,
+        subscription.subscription_id.clone(),
+        update,
+    )
+    .await
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("subscriptions: unable to update subscription entry to database")?;
+
+    let response = ConfirmSubscriptionResponse {
+        subscription: Subscription::new(
+            subscription.subscription_id.clone(),
+            SubscriptionStatus::Active,
+            None,
+        ), // ?!?
+        customer_id: Some(subscription.customer_id.to_owned()),
+        invoice: None,
+        payment: payment_res,
+    };
+
+    Ok(service_api::ApplicationResponse::Json(response))
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct Invoice {
+    pub id: String,
+    pub total: u64,
+}
+
+impl Invoice {
+    pub fn new(id: impl Into<String>, total: u64) -> Self {
+        Self {
+            id: id.into(),
+            total,
+        }
+    }
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct PaymentData {
+    pub payment_method: api_enums::PaymentMethod,
+    pub payment_method_type: Option<api_enums::PaymentMethodType>,
+    pub payment_method_data: api_models::payments::PaymentMethodDataRequest,
+    pub setup_future_usage: Option<api_enums::FutureUsage>,
+    pub customer_acceptance: Option<CustomerAcceptance>,
+}
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct ConfirmSubscriptionRequest {
+    // pub client_secret: Option<String>,
+    pub amount: i64,
+    pub currency: api_enums::Currency,
+    pub plan_id: Option<String>,
+    pub item_price_id: Option<String>,
+    pub coupon_code: Option<String>,
+    pub customer: Option<api_models::payments::CustomerDetails>,
+    pub billing_address: Option<api_models::payments::Address>,
+    pub payment_data: PaymentData,
+}
+
+impl ApiEventMetric for ConfirmSubscriptionRequest {}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct ConfirmSubscriptionResponse {
+    pub subscription: Subscription,
+    pub payment: api_models::payments::PaymentsResponse,
+    pub customer_id: Option<common_utils::id_type::CustomerId>,
+    pub invoice: Option<Invoice>,
+}
+
+impl ApiEventMetric for ConfirmSubscriptionResponse {}
diff --git a/crates/router/src/core/subscription/utils.rs b/crates/router/src/core/subscription/utils.rs
new file mode 100644
index 00000000000..b97f49b58fc
--- /dev/null
+++ b/crates/router/src/core/subscription/utils.rs
@@ -0,0 +1,73 @@
+use api_models::{customers::CustomerRequest, subscription::CreateSubscriptionRequest};
+use common_utils::id_type::GenerateId;
+use error_stack::ResultExt;
+use hyperswitch_domain_models::{
+    api::ApplicationResponse, merchant_context::MerchantContext,
+    router_request_types::CustomerDetails,
+};
+
+use crate::{
+    core::customers::create_customer,
+    db::{errors, StorageInterface},
+    routes::SessionState,
+    types::{api::CustomerResponse, transformers::ForeignInto},
+};
+
+pub async fn get_or_create_customer(
+    state: SessionState,
+    customer_request: Option<CustomerRequest>,
+    merchant_context: MerchantContext,
+) -> errors::CustomerResponse<CustomerResponse> {
+    let db: &dyn StorageInterface = &*state.store;
+
+    // Create customer_id if not passed in request
+    let customer_id = customer_request
+        .as_ref()
+        .and_then(|c| c.customer_id.clone())
+        .unwrap_or_else(common_utils::id_type::CustomerId::generate);
+
+    let merchant_id = merchant_context.get_merchant_account().get_id();
+    let key_manager_state = &(&state).into();
+
+    match db
+        .find_customer_optional_by_customer_id_merchant_id(
+            key_manager_state,
+            &customer_id,
+            merchant_id,
+            merchant_context.get_merchant_key_store(),
+            merchant_context.get_merchant_account().storage_scheme,
+        )
+        .await
+        .change_context(errors::CustomersErrorResponse::InternalServerError)
+        .attach_printable("subscription: unable to perform db read query")?
+    {
+        // Customer found
+        Some(customer) => {
+            let api_customer: CustomerResponse =
+                (customer, None::<api_models::payments::AddressDetails>).foreign_into();
+            Ok(ApplicationResponse::Json(api_customer))
+        }
+
+        // Customer not found
+        None => Ok(create_customer(
+            state,
+            merchant_context,
+            customer_request.ok_or(errors::CustomersErrorResponse::CustomerNotFound)?,
+            None,
+        )
+        .await?),
+    }
+}
+
+pub fn get_customer_details_from_request(request: CreateSubscriptionRequest) -> CustomerDetails {
+    let customer_id = request.get_customer_id().map(ToOwned::to_owned);
+    let customer = request.customer.as_ref();
+    CustomerDetails {
+        customer_id,
+        name: customer.and_then(|cus| cus.name.clone()),
+        email: customer.and_then(|cus| cus.email.clone()),
+        phone: customer.and_then(|cus| cus.phone.clone()),
+        phone_country_code: customer.and_then(|cus| cus.phone_country_code.clone()),
+        tax_registration_id: customer.and_then(|cus| cus.tax_registration_id.clone()),
+    }
+}
diff --git a/crates/router/src/core/webhooks/incoming.rs b/crates/router/src/core/webhooks/incoming.rs
index 4d38ce2ed02..0b935aaa390 100644
--- a/crates/router/src/core/webhooks/incoming.rs
+++ b/crates/router/src/core/webhooks/incoming.rs
@@ -4,6 +4,7 @@ use actix_web::FromRequest;
 #[cfg(feature = "payouts")]
 use api_models::payouts as payout_models;
 use api_models::webhooks::{self, WebhookResponseTracker};
+pub use common_enums::enums::ProcessTrackerRunner;
 use common_utils::{
     errors::ReportSwitchExt,
     events::ApiEventsType,
@@ -252,7 +253,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 .decode_webhook_body(
                     &request_details,
                     merchant_context.get_merchant_account().get_id(),
-                    merchant_connector_account
+                    merchant_connector_account.clone()
                         .and_then(|mca| mca.connector_webhook_details.clone()),
                     &connector_name,
                 )
@@ -317,6 +318,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 &webhook_processing_result.transform_data,
                 &final_request_details,
                 is_relay_webhook,
+                merchant_connector_account.unwrap().merchant_connector_id,
             )
             .await;
 
@@ -499,6 +501,7 @@ async fn process_webhook_business_logic(
     webhook_transform_data: &Option<Box<unified_connector_service::WebhookTransformData>>,
     request_details: &IncomingWebhookRequestDetails<'_>,
     is_relay_webhook: bool,
+    billing_connector_mca_id: common_utils::id_type::MerchantConnectorAccountId,
 ) -> errors::RouterResult<WebhookResponseTracker> {
     let object_ref_id = connector
         .get_webhook_object_reference_id(request_details)
@@ -789,8 +792,21 @@ async fn process_webhook_business_logic(
             .await
             .attach_printable("Incoming webhook flow for payouts failed"),
 
-            _ => Err(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Unsupported Flow Type received in incoming webhooks"),
+            api::WebhookFlow::Subscription => Box::pin(subscription_incoming_webhook_flow(
+                state.clone(),
+                req_state,
+                merchant_context.clone(),
+                business_profile,
+                webhook_details,
+                source_verified,
+                &connector,
+                &request_details,
+                event_type,
+                billing_connector_mca_id,
+            ))
+            .await
+            .attach_printable("Incoming webhook flow for subscription failed"),
+
         }
     };
 
@@ -2505,3 +2521,132 @@ fn insert_mandate_details(
     )?;
     Ok(connector_mandate_details)
 }
+
+#[allow(clippy::too_many_arguments)]
+#[instrument(skip_all)]
+async fn subscription_incoming_webhook_flow(
+    state: SessionState,
+    _req_state: ReqState,
+    merchant_context: domain::MerchantContext,
+    business_profile: domain::Profile,
+    webhook_details: api::IncomingWebhookDetails,
+    source_verified: bool,
+    connector: &ConnectorEnum,
+    _request_details: &IncomingWebhookRequestDetails<'_>,
+    event_type: webhooks::IncomingWebhookEvent,
+    billing_connector_mca_id: common_utils::id_type::MerchantConnectorAccountId,
+) -> CustomResult<WebhookResponseTracker, errors::ApiErrorResponse> {
+    // Only process invoice_generated events for MIT payments
+    if event_type != webhooks::IncomingWebhookEvent::InvoiceGenerated {
+        return Ok(WebhookResponseTracker::NoEffect);
+    }
+
+    if !source_verified {
+        logger::error!("Webhook source verification failed for subscription webhook flow");
+        return Err(report!(
+            errors::ApiErrorResponse::WebhookAuthenticationFailed
+        ));
+    }
+
+    // Parse the webhook body to extract MIT payment data
+    let mit_payment_data = match connector.id() {
+        "chargebee" => {
+            use hyperswitch_connectors::connectors::chargebee::transformers::{
+                ChargebeeInvoiceBody, ChargebeeMitPaymentData,
+            };
+
+            let webhook_body = ChargebeeInvoiceBody::get_invoice_webhook_data_from_body(
+                &webhook_details.resource_object,
+            )
+            .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
+            .attach_printable("Failed to parse Chargebee invoice webhook body")?;
+
+            ChargebeeMitPaymentData::try_from(webhook_body)
+                .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
+                .attach_printable("Failed to extract MIT payment data from Chargebee webhook")?
+        }
+        _ => {
+            return Err(errors::ApiErrorResponse::WebhookProcessingFailure)
+                .attach_printable("Subscription webhook flow not supported for this connector");
+        }
+    };
+
+    logger::info!(
+        invoice_id = %mit_payment_data.invoice_id,
+        amount_due = %mit_payment_data.amount_due.get_amount_as_i64(),
+        currency = %mit_payment_data.currency_code,
+        status = %mit_payment_data.status,
+        subscription_id = ?mit_payment_data.subscription_id,
+        first_invoice = %mit_payment_data.first_invoice,
+        "Received invoice_generated webhook for MIT payment"
+    );
+
+    if mit_payment_data.first_invoice {
+        return Ok(WebhookResponseTracker::NoEffect)
+    }
+
+    // For now, we need a payment_method_id to create the subscription workflow
+    // TODO: Implement proper payment method retrieval from subscription/customer data
+
+    let payment_method_id = state
+        .store
+        .find_payment_method_ids_by_billing_connector_subscription_id(
+            mit_payment_data
+                .subscription_id
+                .as_ref()
+                .ok_or(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Missing subscription_id in MIT payment data")?,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to find payment method IDs by subscription ID")?
+        .first()
+        .ok_or(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("No payment method found for subscription ID")?
+        .clone();
+
+
+    logger::info!("Payment method ID found: {}", payment_method_id);
+
+    // Create tracking data for subscription MIT payment
+    let tracking_data =
+        api_models::process_tracker::subscription::SubscriptionWorkflowTrackingData {
+            merchant_id: merchant_context.get_merchant_account().get_id().clone(),
+            profile_id: business_profile.get_id().clone(),
+            payment_method_id,
+            subscription_id: mit_payment_data.subscription_id,
+            invoice_id: mit_payment_data.invoice_id.clone(),
+            amount: mit_payment_data.amount_due,
+            currency: mit_payment_data.currency_code,
+            customer_id: mit_payment_data.customer_id,
+            connector_name: connector.id().to_string(),
+            billing_connector_mca_id: billing_connector_mca_id.clone(),
+        };
+
+    // Create process tracker entry for subscription MIT payment
+    let process_tracker_entry = diesel_models::ProcessTrackerNew {
+        id: generate_id(consts::ID_LENGTH, "proc"),
+        name: Some("SUBSCRIPTION_MIT_PAYMENT".to_string()),
+        tag: vec!["SUBSCRIPTION".to_string()],
+        runner: Some(ProcessTrackerRunner::SubscriptionsWorkflow.to_string()),
+        retry_count: 0,
+        schedule_time: Some(common_utils::date_time::now()),
+        rule: String::new(),
+        tracking_data: serde_json::to_value(&tracking_data)
+            .change_context(errors::ApiErrorResponse::InternalServerError)?,
+        business_status: "Pending".to_string(),
+        status: diesel_models::enums::ProcessTrackerStatus::New,
+        event: vec![],
+        created_at: common_utils::date_time::now(),
+        updated_at: common_utils::date_time::now(),
+        version: common_types::consts::API_VERSION,
+    };
+
+    state
+        .store
+        .insert_process(process_tracker_entry)
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)?;
+
+    Ok(WebhookResponseTracker::NoEffect)
+}
diff --git a/crates/router/src/db.rs b/crates/router/src/db.rs
index 59c31d4b2b8..f675e316aef 100644
--- a/crates/router/src/db.rs
+++ b/crates/router/src/db.rs
@@ -34,6 +34,7 @@ pub mod relay;
 pub mod reverse_lookup;
 pub mod role;
 pub mod routing_algorithm;
+pub mod subscription;
 pub mod unified_translations;
 pub mod user;
 pub mod user_authentication_method;
@@ -143,6 +144,7 @@ pub trait StorageInterface:
     + payment_method_session::PaymentMethodsSessionInterface
     + tokenization::TokenizationInterface
     + callback_mapper::CallbackMapperInterface
+    + subscription::SubscriptionInterface
     + 'static
 {
     fn get_scheduler_db(&self) -> Box<dyn scheduler::SchedulerInterface>;
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 9eaef17b9c5..0bb7f9c2da8 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -2340,6 +2340,16 @@ impl PaymentMethodInterface for KafkaStore {
             .find_payment_method_by_fingerprint_id(state, key_store, fingerprint_id)
             .await
     }
+
+    #[cfg(feature = "v1")]
+    async fn find_payment_method_ids_by_billing_connector_subscription_id(
+        &self,
+        subscription_id: &str,
+    ) -> CustomResult<Vec<String>, errors::StorageError> {
+        self.diesel_store
+            .find_payment_method_ids_by_billing_connector_subscription_id(subscription_id)
+            .await
+    }
 }
 
 #[cfg(not(feature = "payouts"))]
diff --git a/crates/router/src/db/subscription.rs b/crates/router/src/db/subscription.rs
new file mode 100644
index 00000000000..368e254a445
--- /dev/null
+++ b/crates/router/src/db/subscription.rs
@@ -0,0 +1,170 @@
+use error_stack::report;
+use router_env::{instrument, tracing};
+use storage_impl::MockDb;
+
+use super::Store;
+use crate::{
+    connection,
+    core::errors::{self, CustomResult},
+    db::kafka_store::KafkaStore,
+    types::storage,
+};
+
+#[async_trait::async_trait]
+pub trait SubscriptionInterface {
+    async fn insert_subscription_entry(
+        &self,
+        subscription_new: storage::subscription::SubscriptionNew,
+    ) -> CustomResult<storage::Subscription, errors::StorageError>;
+
+    async fn find_by_merchant_id_subscription_id(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+        subscription_id: String,
+    ) -> CustomResult<storage::Subscription, errors::StorageError>;
+
+    async fn update_subscription_entry(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+        subscription_id: String,
+        data: storage::SubscriptionUpdate,
+    ) -> CustomResult<storage::Subscription, errors::StorageError>;
+
+    // async fn find_subscription_by_id(
+    //     &self,
+    //     id: String,
+    // ) -> CustomResult<storage::Subscription, errors::StorageError>;
+}
+
+#[async_trait::async_trait]
+impl SubscriptionInterface for Store {
+    #[instrument(skip_all)]
+    async fn insert_subscription_entry(
+        &self,
+        subscription_new: storage::subscription::SubscriptionNew,
+    ) -> CustomResult<storage::Subscription, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        subscription_new
+            .insert(&conn)
+            .await
+            .map_err(|error| report!(errors::StorageError::from(error)))
+    }
+
+    #[instrument(skip_all)]
+    async fn find_by_merchant_id_subscription_id(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+        subscription_id: String,
+    ) -> CustomResult<storage::Subscription, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::Subscription::find_by_merchant_id_subscription_id(
+            &conn,
+            merchant_id,
+            subscription_id,
+        )
+        .await
+        .map_err(|error| report!(errors::StorageError::from(error)))
+    }
+
+    #[instrument(skip_all)]
+    async fn update_subscription_entry(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+        subscription_id: String,
+        data: storage::SubscriptionUpdate,
+    ) -> CustomResult<storage::Subscription, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::Subscription::update_subscription_entry(&conn, merchant_id, subscription_id, data)
+            .await
+            .map_err(|error| report!(errors::StorageError::from(error)))
+    }
+
+    // #[instrument(skip_all)]
+    // async fn find_subscription_by_id(
+    //     &self,
+    //     id: String,
+    // ) -> CustomResult<storage::Subscription, errors::StorageError> {
+    //     let conn = connection::pg_connection_write(self).await?;
+    //     storage::Subscription::find_subscription_by_id(&conn, id)
+    //         .await
+    //         .map_err(|error| report!(errors::StorageError::from(error)))
+    // }
+}
+
+#[async_trait::async_trait]
+impl SubscriptionInterface for MockDb {
+    #[instrument(skip_all)]
+    async fn insert_subscription_entry(
+        &self,
+        _subscription_new: storage::subscription::SubscriptionNew,
+    ) -> CustomResult<storage::Subscription, errors::StorageError> {
+        Err(errors::StorageError::MockDbError)?
+    }
+
+    async fn find_by_merchant_id_subscription_id(
+        &self,
+        _merchant_id: &common_utils::id_type::MerchantId,
+        _subscription_id: String,
+    ) -> CustomResult<storage::Subscription, errors::StorageError> {
+        Err(errors::StorageError::MockDbError)?
+    }
+
+    async fn update_subscription_entry(
+        &self,
+        _merchant_id: &common_utils::id_type::MerchantId,
+        _subscription_id: String,
+        _data: storage::SubscriptionUpdate,
+    ) -> CustomResult<storage::Subscription, errors::StorageError> {
+        Err(errors::StorageError::MockDbError)?
+    }
+
+    // async fn find_subscription_by_id(
+    //     &self,
+    //     _id: String,
+    // ) -> CustomResult<storage::Subscription, errors::StorageError> {
+    //     Err(errors::StorageError::MockDbError)?
+    // }
+}
+
+#[async_trait::async_trait]
+impl SubscriptionInterface for KafkaStore {
+    #[instrument(skip_all)]
+    async fn insert_subscription_entry(
+        &self,
+        subscription_new: storage::subscription::SubscriptionNew,
+    ) -> CustomResult<storage::Subscription, errors::StorageError> {
+        self.diesel_store
+            .insert_subscription_entry(subscription_new)
+            .await
+    }
+
+    #[instrument(skip_all)]
+    async fn find_by_merchant_id_subscription_id(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+        subscription_id: String,
+    ) -> CustomResult<storage::Subscription, errors::StorageError> {
+        self.diesel_store
+            .find_by_merchant_id_subscription_id(merchant_id, subscription_id)
+            .await
+    }
+
+    #[instrument(skip_all)]
+    async fn update_subscription_entry(
+        &self,
+        merchant_id: &common_utils::id_type::MerchantId,
+        subscription_id: String,
+        data: storage::SubscriptionUpdate,
+    ) -> CustomResult<storage::Subscription, errors::StorageError> {
+        self.diesel_store
+            .update_subscription_entry(merchant_id, subscription_id, data)
+            .await
+    }
+
+    // async fn find_subscription_by_id(
+    //     &self,
+    //     id: String,
+    // ) -> CustomResult<storage::Subscription, errors::StorageError> {
+    //     self.diesel_store.find_subscription_by_id(id).await
+    // }
+}
diff --git a/crates/router/src/lib.rs b/crates/router/src/lib.rs
index 00df2908f44..43cd29e2f72 100644
--- a/crates/router/src/lib.rs
+++ b/crates/router/src/lib.rs
@@ -209,6 +209,7 @@ pub fn mk_app(
                 .service(routes::Files::server(state.clone()))
                 .service(routes::Disputes::server(state.clone()))
                 .service(routes::Blocklist::server(state.clone()))
+                .service(routes::Subscription::server(state.clone()))
                 .service(routes::Gsm::server(state.clone()))
                 .service(routes::ApplePayCertificatesMigration::server(state.clone()))
                 .service(routes::PaymentLink::server(state.clone()))
diff --git a/crates/router/src/routes.rs b/crates/router/src/routes.rs
index b8fde79e2a1..f2af4631917 100644
--- a/crates/router/src/routes.rs
+++ b/crates/router/src/routes.rs
@@ -50,6 +50,8 @@ pub mod recon;
 pub mod refunds;
 #[cfg(feature = "olap")]
 pub mod routing;
+#[cfg(feature = "v1")]
+pub mod subscription;
 pub mod three_ds_decision_rule;
 pub mod tokenization;
 #[cfg(feature = "olap")]
@@ -96,7 +98,7 @@ pub use self::app::{
     User, UserDeprecated, Webhooks,
 };
 #[cfg(feature = "olap")]
-pub use self::app::{Blocklist, Organization, Routing, Verify, WebhookEvents};
+pub use self::app::{Blocklist, Organization, Routing, Subscription, Verify, WebhookEvents};
 #[cfg(feature = "payouts")]
 pub use self::app::{PayoutLink, Payouts};
 #[cfg(all(feature = "stripe", feature = "v1"))]
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 690a4337351..0fab52a40cf 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -65,7 +65,9 @@ use super::{
     profiles, relay, user, user_role,
 };
 #[cfg(feature = "v1")]
-use super::{apple_pay_certificates_migration, blocklist, payment_link, webhook_events};
+use super::{
+    apple_pay_certificates_migration, blocklist, payment_link, subscription, webhook_events,
+};
 #[cfg(any(feature = "olap", feature = "oltp"))]
 use super::{configs::*, customers, payments};
 #[cfg(all(any(feature = "olap", feature = "oltp"), feature = "v1"))]
@@ -1155,6 +1157,29 @@ impl Routing {
     }
 }
 
+#[cfg(feature = "olap")]
+pub struct Subscription;
+
+#[cfg(all(feature = "olap", feature = "v1"))]
+impl Subscription {
+    pub fn server(state: AppState) -> Scope {
+        web::scope("/subscription")
+            .app_data(web::Data::new(state.clone()))
+            .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)
+                    },
+                )),
+            )
+    }
+}
+
 pub struct Customers;
 
 #[cfg(all(feature = "v2", any(feature = "olap", feature = "oltp")))]
diff --git a/crates/router/src/routes/lock_utils.rs b/crates/router/src/routes/lock_utils.rs
index 27b66e5bb54..fb8c6c8a935 100644
--- a/crates/router/src/routes/lock_utils.rs
+++ b/crates/router/src/routes/lock_utils.rs
@@ -26,6 +26,7 @@ pub enum ApiIdentifier {
     ApiKeys,
     PaymentLink,
     Routing,
+    Subscription,
     Blocklist,
     Forex,
     RustLockerMigration,
@@ -88,6 +89,8 @@ impl From<Flow> for ApiIdentifier {
             | Flow::DecisionEngineDecideGatewayCall
             | Flow::DecisionEngineGatewayFeedbackCall => Self::Routing,
 
+            Flow::CreateSubscription => 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
new file mode 100644
index 00000000000..b2edfb4ed85
--- /dev/null
+++ b/crates/router/src/routes/subscription.rs
@@ -0,0 +1,94 @@
+//! Analysis for usage of Subscription in Payment flows
+//!
+//! Functions that are used to perform the api level configuration and retrieval
+//! of various types under Subscriptions.
+
+use actix_web::{web, HttpRequest, Responder};
+use api_models::subscription as subscription_types;
+use router_env::{
+    tracing::{self, instrument},
+    Flow,
+};
+
+use crate::{
+    core::{api_locking, subscription},
+    routes::AppState,
+    services::{api as oss_api, authentication as auth, authorization::permissions::Permission},
+    types::domain,
+};
+
+#[cfg(all(feature = "olap", feature = "v1"))]
+#[instrument(skip_all)]
+pub async fn create_subscription(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    json_payload: web::Json<subscription_types::CreateSubscriptionRequest>,
+) -> impl Responder {
+    let flow = Flow::CreateSubscription;
+    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::create_subscription(state, merchant_context, payload.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
+}
+
+#[cfg(all(feature = "olap", feature = "v1"))]
+#[instrument(skip_all)]
+pub async fn confirm_subscription(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    subscription_id: web::Path<String>,
+    json_payload: web::Json<subscription::ConfirmSubscriptionRequest>,
+) -> impl Responder {
+    let flow = Flow::RoutingCreateConfig;
+    let subscription_id = subscription_id.into_inner();
+    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,
+                auth.profile_id,
+                payload.clone(),
+                subscription_id.clone(),
+            )
+        },
+        auth::auth_type(
+            &auth::HeaderAuth(auth::ApiKeyAuth {
+                is_connected_allowed: false,
+                is_platform_allowed: false,
+            }),
+            &auth::JWTAuth {
+                permission: Permission::ProfileRoutingWrite,
+            },
+            req.headers(),
+        ),
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index fa8e840fdea..ec0e4e43b0c 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -127,6 +127,8 @@ pub type BoxedBillingConnectorPaymentsSyncIntegrationInterface<T, Req, Res> =
 pub type BoxedVaultConnectorIntegrationInterface<T, Req, Res> =
     BoxedConnectorIntegrationInterface<T, common_types::VaultConnectorFlowData, Req, Res>;
 
+pub type BoxedSubscriptionConnectorIntegrationInterface<T, Req, Res> =
+    BoxedConnectorIntegrationInterface<T, common_types::SubscriptionCreateData, Req, Res>;
 /// Handle UCS webhook response processing
 fn handle_ucs_response<T, Req, Resp>(
     router_data: types::RouterData<T, Req, Resp>,
diff --git a/crates/router/src/services/authorization/permissions.rs b/crates/router/src/services/authorization/permissions.rs
index 3860d343b68..6fb4b09cd5d 100644
--- a/crates/router/src/services/authorization/permissions.rs
+++ b/crates/router/src/services/authorization/permissions.rs
@@ -43,6 +43,10 @@ generate_permissions! {
             scopes: [Read, Write],
             entities: [Profile, Merchant]
         },
+        Subscription: {
+            scopes: [Read, Write],
+            entities: [Profile, Merchant]
+        },
         ThreeDsDecisionManager: {
             scopes: [Read, Write],
             entities: [Merchant, Profile]
@@ -123,6 +127,7 @@ pub fn get_resource_name(resource: Resource, entity_type: EntityType) -> Option<
             Some("Payment Processors, Payout Processors, Fraud & Risk Managers")
         }
         (Resource::Routing, _) => Some("Routing"),
+        (Resource::Subscription, _) => Some("Subscription"),
         (Resource::RevenueRecovery, _) => Some("Revenue Recovery"),
         (Resource::ThreeDsDecisionManager, _) => Some("3DS Decision Manager"),
         (Resource::SurchargeDecisionManager, _) => Some("Surcharge Decision Manager"),
diff --git a/crates/router/src/types/storage.rs b/crates/router/src/types/storage.rs
index 70a0a344150..5d4c4a8bb29 100644
--- a/crates/router/src/types/storage.rs
+++ b/crates/router/src/types/storage.rs
@@ -41,6 +41,7 @@ pub mod revenue_recovery_redis_operation;
 pub mod reverse_lookup;
 pub mod role;
 pub mod routing_algorithm;
+pub mod subscription;
 pub mod unified_translations;
 pub mod user;
 pub mod user_authentication_method;
@@ -77,5 +78,5 @@ pub use self::{
     generic_link::*, gsm::*, locker_mock_up::*, mandate::*, merchant_account::*,
     merchant_connector_account::*, merchant_key_store::*, payment_link::*, payment_method::*,
     process_tracker::*, refund::*, reverse_lookup::*, role::*, routing_algorithm::*,
-    unified_translations::*, user::*, user_authentication_method::*, user_role::*,
+    subscription::*, unified_translations::*, user::*, user_authentication_method::*, user_role::*,
 };
diff --git a/crates/router/src/types/storage/subscription.rs b/crates/router/src/types/storage/subscription.rs
new file mode 100644
index 00000000000..973c2283494
--- /dev/null
+++ b/crates/router/src/types/storage/subscription.rs
@@ -0,0 +1 @@
+pub use diesel_models::subscription::{Subscription, SubscriptionNew, SubscriptionUpdate};
diff --git a/crates/router/src/utils/user/sample_data.rs b/crates/router/src/utils/user/sample_data.rs
index 5c8d79e0f04..9ffa01045ef 100644
--- a/crates/router/src/utils/user/sample_data.rs
+++ b/crates/router/src/utils/user/sample_data.rs
@@ -297,6 +297,7 @@ pub async fn generate_sample_data(
             tax_status: None,
             shipping_amount_tax: None,
             enable_partial_authorization: None,
+            billing_processor_details: None,
         };
         let (connector_transaction_id, processor_transaction_data) =
             ConnectorTransactionId::form_id_and_data(attempt_id.clone());
diff --git a/crates/router/src/workflows.rs b/crates/router/src/workflows.rs
index 2c2410f7449..283b72d086d 100644
--- a/crates/router/src/workflows.rs
+++ b/crates/router/src/workflows.rs
@@ -15,3 +15,5 @@ pub mod revenue_recovery;
 pub mod process_dispute;
 
 pub mod dispute_list;
+
+pub mod subscription;
diff --git a/crates/router/src/workflows/payment_sync.rs b/crates/router/src/workflows/payment_sync.rs
index 52a72a2d020..f16554d7896 100644
--- a/crates/router/src/workflows/payment_sync.rs
+++ b/crates/router/src/workflows/payment_sync.rs
@@ -1,12 +1,19 @@
+#[cfg(feature = "v1")]
+use crate::core::payments::helpers::{
+    perform_billing_processor_record_back,
+};
 use common_utils::ext_traits::{OptionExt, StringExt, ValueExt};
 use diesel_models::process_tracker::business_status;
 use error_stack::ResultExt;
+
+
 use router_env::logger;
 use scheduler::{
     consumer::{self, types::process_data, workflows::ProcessTrackerWorkflow},
     errors as sch_errors, utils as scheduler_utils,
 };
 
+
 use crate::{
     consts,
     core::{
@@ -112,7 +119,10 @@ impl ProcessTrackerWorkflow<SessionState> for PaymentsSyncWorkflow {
                     .store
                     .as_scheduler()
                     .finish_process_with_business_status(process, business_status::COMPLETED_BY_PT)
-                    .await?
+                    .await?;
+
+                // call to subsription connector
+                perform_billing_processor_record_back(state, &mut payment_data, &key_store).await?;
             }
             _ => {
                 let connector = payment_data
diff --git a/crates/router/src/workflows/subscription.rs b/crates/router/src/workflows/subscription.rs
new file mode 100644
index 00000000000..fcb7c784e38
--- /dev/null
+++ b/crates/router/src/workflows/subscription.rs
@@ -0,0 +1,206 @@
+use api_models::payments::BillingConnectorDetails;
+use async_trait::async_trait;
+use common_utils::ext_traits::ValueExt;
+use diesel_models::process_tracker::business_status;
+use error_stack::ResultExt;
+use router_env::logger;
+use scheduler::{consumer::workflows::ProcessTrackerWorkflow, errors};
+
+#[cfg(feature = "v1")]
+use crate::routes::payments::get_or_generate_payment_id;
+use crate::{
+    core::{errors::RecoveryError::ProcessTrackerFailure, payments},
+    routes::SessionState,
+    services,
+    types::{api as api_types, domain, storage},
+};
+pub struct ExecuteSubscriptionWorkflow;
+
+#[async_trait]
+impl ProcessTrackerWorkflow<SessionState> for ExecuteSubscriptionWorkflow {
+    #[cfg(feature = "v1")]
+    async fn execute_workflow<'a>(
+        &'a self,
+        state: &'a SessionState,
+        process: storage::ProcessTracker,
+    ) -> Result<(), errors::ProcessTrackerError> {
+        let tracking_data = process
+            .tracking_data
+            .clone()
+            .parse_value::<api_models::process_tracker::subscription::SubscriptionWorkflowTrackingData>(
+                "SubscriptionWorkflowTrackingData",
+            )?;
+
+        match process.name.as_deref() {
+            Some("SUBSCRIPTION_MIT_PAYMENT") => {
+                Box::pin(perform_subscription_mit_payment(
+                    state,
+                    &process,
+                    &tracking_data,
+                ))
+                .await
+            }
+            _ => Err(errors::ProcessTrackerError::JobNotFound),
+        }
+    }
+    #[cfg(feature = "v2")]
+    async fn execute_workflow<'a>(
+        &'a self,
+        state: &'a SessionState,
+        process: storage::ProcessTracker,
+    ) -> Result<(), errors::ProcessTrackerError> {
+        todo!()
+    }
+}
+#[cfg(feature = "v1")]
+async fn perform_subscription_mit_payment(
+    state: &SessionState,
+    process: &storage::ProcessTracker,
+    tracking_data: &api_models::process_tracker::subscription::SubscriptionWorkflowTrackingData,
+) -> Result<(), errors::ProcessTrackerError> {
+    // Extract merchant context
+    let key_manager_state = &state.into();
+    let key_store = state
+        .store
+        .get_merchant_key_store_by_merchant_id(
+            key_manager_state,
+            &tracking_data.merchant_id,
+            &state.store.get_master_key().to_vec().into(),
+        )
+        .await?;
+
+    let merchant_account = state
+        .store
+        .find_merchant_account_by_merchant_id(
+            key_manager_state,
+            &tracking_data.merchant_id,
+            &key_store,
+        )
+        .await?;
+
+    let profile = state
+        .store
+        .find_business_profile_by_profile_id(
+            key_manager_state,
+            &key_store,
+            &tracking_data.profile_id,
+        )
+        .await?;
+
+    let merchant_context = domain::MerchantContext::NormalMerchant(Box::new(domain::Context(
+        merchant_account,
+        key_store,
+    )));
+
+    let profile_id = profile.get_id().clone();
+
+    let billing_connector_details = BillingConnectorDetails {
+        processor_mca: tracking_data.billing_connector_mca_id.clone(),
+        subscription_id: tracking_data.subscription_id.clone().ok_or_else(|| {
+            errors::ProcessTrackerError::SerializationFailed
+        })?,
+        invoice_id: tracking_data.invoice_id.clone(),
+    };
+
+    logger::debug!(
+        "Executing subscription MIT payment for process: {:?}, tracking_data: {:?}",
+        process.id,
+        tracking_data
+    );
+
+
+    // Create MIT payment request with the determined payment_method_id
+    let mut payment_request = api_types::PaymentsRequest {
+        amount: Some(api_types::Amount::from(tracking_data.amount)),
+        currency: Some(tracking_data.currency),
+        customer_id: tracking_data.customer_id.clone(),
+        recurring_details: Some(api_models::mandates::RecurringDetails::PaymentMethodId(
+            tracking_data.payment_method_id.clone(),
+        )),
+        merchant_id: Some(tracking_data.merchant_id.clone()),
+        billing_processor_details: Some(billing_connector_details),
+        confirm: Some(true),
+        off_session: Some(true),
+        ..Default::default()
+    };
+
+    logger::debug!(
+        "payment_request for subscription MIT payment: {:?}, process_id: {:?}, tracking_data: {:?}",
+        payment_request,
+        process.id,
+        payment_request
+    );
+
+    if let Err(err) = get_or_generate_payment_id(&mut payment_request) {
+        return Err(err.into());
+    }
+
+    // Execute MIT payment
+    let payment_response = payments::payments_core::<
+        api_types::Authorize,
+        api_types::PaymentsResponse,
+        _,
+        _,
+        _,
+        payments::PaymentData<api_types::Authorize>,
+    >(
+        state.clone(),
+        state.get_req_state(),
+        merchant_context,
+        Some(profile_id),
+        payments::PaymentCreate,
+        payment_request,
+        services::api::AuthFlow::Merchant,
+        payments::CallConnectorAction::Trigger,
+        None,
+        hyperswitch_domain_models::payments::HeaderPayload::with_source(
+            common_enums::PaymentSource::Webhook,
+        ),
+    )
+    .await;
+
+    let payment_res = match payment_response {
+        Ok(services::ApplicationResponse::JsonWithHeaders((pi, _))) => Ok(pi),
+        Ok(_) => Err(errors::ProcessTrackerError::FlowExecutionError {
+            flow: "SUBSCRIPTION_MIT_PAYMENT",
+        }),
+        Err(error) => {
+            logger::error!(?error);
+            Err(errors::ProcessTrackerError::FlowExecutionError {
+                flow: "SUBSCRIPTION_MIT_PAYMENT",
+            })
+        }
+    }?;
+
+    if payment_res.status == common_enums::IntentStatus::Succeeded {
+        // Update the process tracker with the payment response
+        let updated_process = storage::ProcessTracker {
+            id: process.id.clone(),
+            status: common_enums::ProcessTrackerStatus::Finish,
+            ..process.clone()
+        };
+
+        state
+            .store
+            .as_scheduler()
+            .finish_process_with_business_status(
+                updated_process.clone(),
+                business_status::EXECUTE_WORKFLOW_COMPLETE,
+            )
+            .await
+            .change_context(ProcessTrackerFailure)
+            .attach_printable("Failed to update the process tracker")?;
+    } else {
+        // Handle payment failure - log the payment status and return appropriate error
+        logger::error!(
+            "Payment failed for subscription MIT payment. Payment ID: {:?}, Status: {:?}",
+            payment_res.payment_id,
+            payment_res.status
+        );
+        return Err(errors::ProcessTrackerError::FlowExecutionError {
+            flow: "SUBSCRIPTION_MIT_PAYMENT",
+        });
+    }
+
+    Ok(())
+}
diff --git a/crates/router_env/src/logger/types.rs b/crates/router_env/src/logger/types.rs
index b758d98f498..56cdd049b2f 100644
--- a/crates/router_env/src/logger/types.rs
+++ b/crates/router_env/src/logger/types.rs
@@ -261,6 +261,8 @@ pub enum Flow {
     RoutingUpdateDefaultConfig,
     /// Routing delete config
     RoutingDeleteConfig,
+    /// Subscription create flow,
+    CreateSubscription,
     /// Create dynamic routing
     CreateDynamicRoutingConfig,
     /// Toggle dynamic routing
diff --git a/crates/storage_impl/src/mock_db.rs b/crates/storage_impl/src/mock_db.rs
index 884d9747943..260bb33a5e2 100644
--- a/crates/storage_impl/src/mock_db.rs
+++ b/crates/storage_impl/src/mock_db.rs
@@ -65,6 +65,7 @@ pub struct MockDb {
     pub user_authentication_methods:
         Arc<Mutex<Vec<store::user_authentication_method::UserAuthenticationMethod>>>,
     pub themes: Arc<Mutex<Vec<store::user::theme::Theme>>>,
+    pub subscriptions: Arc<Mutex<Vec<store::subscription::Subscription>>>,
 }
 
 impl MockDb {
@@ -113,6 +114,7 @@ impl MockDb {
             user_key_store: Default::default(),
             user_authentication_methods: Default::default(),
             themes: Default::default(),
+            subscriptions: Default::default(),
         })
     }
 
diff --git a/crates/storage_impl/src/payment_method.rs b/crates/storage_impl/src/payment_method.rs
index 760f09e8eb0..d1707cb8296 100644
--- a/crates/storage_impl/src/payment_method.rs
+++ b/crates/storage_impl/src/payment_method.rs
@@ -370,6 +370,16 @@ impl<T: DatabaseStore> PaymentMethodInterface for KVRouterStore<T> {
             .await
     }
 
+    #[cfg(feature = "v1")]
+    async fn find_payment_method_ids_by_billing_connector_subscription_id(
+        &self,
+        subscription_id: &str,
+    ) -> CustomResult<Vec<String>, errors::StorageError> {
+        self.router_store
+            .find_payment_method_ids_by_billing_connector_subscription_id(subscription_id)
+            .await
+    }
+
     // Soft delete, Check if KV stuff is needed here
     #[cfg(feature = "v2")]
     async fn delete_payment_method(
@@ -666,6 +676,23 @@ impl<T: DatabaseStore> PaymentMethodInterface for RouterStore<T> {
         .await
     }
 
+    #[cfg(feature = "v1")]
+    async fn find_payment_method_ids_by_billing_connector_subscription_id(
+        &self,
+        subscription_id: &str,
+    ) -> CustomResult<Vec<String>, errors::StorageError> {
+        let conn = pg_connection_read(self).await?;
+        diesel_models::subscription::Subscription::find_payment_method_ids_by_billing_connector_subscription_id(
+            &conn,
+            subscription_id,
+        )
+        .await
+        .map_err(|error| {
+            let new_err = diesel_error_to_data_error(*error.current_context());
+            error.change_context(new_err)
+        })
+    }
+
     #[cfg(feature = "v2")]
     async fn delete_payment_method(
         &self,
@@ -923,6 +950,24 @@ impl PaymentMethodInterface for MockDb {
         }
     }
 
+    #[cfg(feature = "v1")]
+    async fn find_payment_method_ids_by_billing_connector_subscription_id(
+        &self,
+        subscription_id: &str,
+    ) -> CustomResult<Vec<String>, errors::StorageError> {
+        // let subscriptions = self.subscriptions.lock().await;
+        // let payment_method_ids: Vec<String> = subscriptions
+        //     .iter()
+        //     .filter(|sub| {
+        //         sub.subscription_id
+        //             .as_ref()
+        //             .map_or(false, |id| id == subscription_id)
+        //     })
+        //     .filter_map(|sub| sub.payment_method_id.clone())
+        //     .collect();
+        Ok(vec![])
+    }
+
     async fn update_payment_method(
         &self,
         state: &KeyManagerState,
diff --git a/migrations/2025-08-21-110802_add_subcription_table/down.sql b/migrations/2025-08-21-110802_add_subcription_table/down.sql
new file mode 100644
index 00000000000..0979103e48c
--- /dev/null
+++ b/migrations/2025-08-21-110802_add_subcription_table/down.sql
@@ -0,0 +1 @@
+DROP table subscription;
diff --git a/migrations/2025-08-21-110802_add_subcription_table/up.sql b/migrations/2025-08-21-110802_add_subcription_table/up.sql
new file mode 100644
index 00000000000..94614bf9448
--- /dev/null
+++ b/migrations/2025-08-21-110802_add_subcription_table/up.sql
@@ -0,0 +1,16 @@
+CREATE TABLE subscription (
+  id SERIAL PRIMARY KEY,
+  subscription_id VARCHAR(128) NOT NULL,
+  status VARCHAR(128) NOT NULL,
+  billing_processor VARCHAR(128),
+  payment_method_id VARCHAR(128),
+  mca_id VARCHAR(128),
+  client_secret VARCHAR(128),
+  merchant_id VARCHAR(64) NOT NULL,
+  customer_id VARCHAR(64) NOT NULL,
+  metadata JSONB,
+  created_at TIMESTAMP NOT NULL,
+  modified_at TIMESTAMP NOT NULL
+);
+
+CREATE UNIQUE INDEX merchant_subscription_unique_index ON subscription (merchant_id, subscription_id);
diff --git a/migrations/2025-08-26-085230_add_billing_processor_in_connector_type/down.sql b/migrations/2025-08-26-085230_add_billing_processor_in_connector_type/down.sql
new file mode 100644
index 00000000000..d0b08278125
--- /dev/null
+++ b/migrations/2025-08-26-085230_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-08-26-085230_add_billing_processor_in_connector_type/up.sql b/migrations/2025-08-26-085230_add_billing_processor_in_connector_type/up.sql
new file mode 100644
index 00000000000..0201e3753f1
--- /dev/null
+++ b/migrations/2025-08-26-085230_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
diff --git a/migrations/2025-08-26-085518_add_billing_processor_details_in_payment_intent/down.sql b/migrations/2025-08-26-085518_add_billing_processor_details_in_payment_intent/down.sql
new file mode 100644
index 00000000000..0c5528d1072
--- /dev/null
+++ b/migrations/2025-08-26-085518_add_billing_processor_details_in_payment_intent/down.sql
@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE payment_intent DROP COLUMN IF EXISTS billing_processor_details jsonb;
\ No newline at end of file
diff --git a/migrations/2025-08-26-085518_add_billing_processor_details_in_payment_intent/up.sql b/migrations/2025-08-26-085518_add_billing_processor_details_in_payment_intent/up.sql
new file mode 100644
index 00000000000..6efa6e47164
--- /dev/null
+++ b/migrations/2025-08-26-085518_add_billing_processor_details_in_payment_intent/up.sql
@@ -0,0 +1,2 @@
+-- Your SQL goes here
+ALTER TABLE payment_intent ADD COLUMN IF NOT EXISTS billing_processor_details jsonb;
\ No newline at end of file