Issue:
Bug: Handle incoming webhooks for invoice_created event





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 adds support for the invoice-generated event type for the incoming webhook for Chargebee and creates a process tracker entry that will trigger a S2S call and invoice sync.

### 1. __Invoice Generated Webhook Support__

- Implements handling for Chargebee's `invoice_generated` webhook event
- Creates process tracker entries for subscription workflow management
- Added  SubscriptionId to  ObjectReferenceId for subscription webhook
- Created a get_mca_from_subscription_id function to fetch the MerchantConnectorAccount from the SubscriptionId

### 2. __Subscription Workflow Integration__

- Adds subscription workflow tracking data
- Implements process tracker integration for subscription events

### 3. __Database Changes__

- Insert details into the invoice table


### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->

Insert into subscription table
```
INSERT INTO public.subscription (
    id,
    status,
    billing_processor,
    payment_method_id,
    merchant_connector_id,
    client_secret,
    connector_subscription_id,
    merchant_id,
    customer_id,
    metadata,
    created_at,
    modified_at,
    profile_id,
    merchant_reference_id
) VALUES (
    '169vD0Uxi8JB746e',
    'active',
    'chargebee',
    'pm_98765',
    'mca_YH3XxYGMv6IbKD0icBie',
    'secret_subs_1231abcd',
    '169vD0Uxi8JB746e',
    'merchant_1758710612',
    'gaurav_test',
    '{"plan":"basic","trial":true}',
    NOW(),
    NOW(),
    'pro_RkLiGcSuaRjmxA5suSfV',
    'ref_subs_1231abcd'
);
```
incoming webhook for chargebee
```
curl --location 'http://localhost:8080/webhooks/merchant_1758710612/mca_YH3XxYGMv6IbKD0icBie' \
--header 'api-key: dev_5LNrrMNqyyFle8hvMssLob9pFq6OIEh0Vl3DqZttFXiDe7wrrScUePEYrIiYfSKT' \
--header 'Content-Type: application/json' \
--header 'authorization: Basic aHlwZXJzd2l0Y2g6aHlwZXJzd2l0Y2g=' \
--data-raw '{
    "api_version": "v2",
    "content": {
        "invoice": {
            "adjustment_credit_notes": [],
            "amount_adjusted": 0,
            "amount_due": 14100,
            "amount_paid": 0,
            "amount_to_collect": 14100,
            "applied_credits": [],
            "base_currency_code": "INR",
            "channel": "web",
            "credits_applied": 0,
            "currency_code": "INR",
            "customer_id": "gaurav_test",
            "date": 1758711043,
            "deleted": false,
            "due_date": 1758711043,
            "dunning_attempts": [],
            "exchange_rate": 1.0,
            "first_invoice": false,
            "generated_at": 1758711043,
            "has_advance_charges": false,
            "id": "3",
            "is_gifted": false,
            "issued_credit_notes": [],
            "line_items": [
                {
                    "amount": 14100,
                    "customer_id": "gaurav_test",
                    "date_from": 1758711043,
                    "date_to": 1761303043,
                    "description": "Enterprise Suite Monthly",
                    "discount_amount": 0,
                    "entity_id": "cbdemo_enterprise-suite-monthly",
                    "entity_type": "plan_item_price",
                    "id": "li_169vD0Uxi8JBp46g",
                    "is_taxed": false,
                    "item_level_discount_amount": 0,
                    "object": "line_item",
                    "pricing_model": "flat_fee",
                    "quantity": 1,
                    "subscription_id": "169vD0Uxi8JB746e",
                    "tax_amount": 0,
                    "tax_exempt_reason": "tax_not_configured",
                    "unit_amount": 14100
                }
            ],
            "linked_orders": [],
            "linked_payments": [],
            "net_term_days": 0,
            "new_sales_amount": 14100,
            "object": "invoice",
            "price_type": "tax_exclusive",
            "recurring": true,
            "reference_transactions": [],
            "resource_version": 1758711043846,
            "round_off_amount": 0,
            "site_details_at_creation": {
                "timezone": "Asia/Calcutta"
            },
            "status": "payment_due",
            "sub_total": 14100,
            "subscription_id": "169vD0Uxi8JB746e",
            "tax": 0,
            "term_finalized": true,
            "total": 14100,
            "updated_at": 1758711043,
            "write_off_amount": 0
        }
    },
    "event_type": "invoice_generated",
    "id": "ev_169vD0Uxi8JDf46i",
    "object": "event",
    "occurred_at": 1758711043,
    "source": "admin_console",
    "user": "ankit.gupta.001@juspay.in",
    "webhook_status": "scheduled",
    "webhooks": [
        {
            "id": "whv2_169mYOUxi3nvkZ8v",
            "object": "webhook",
            "webhook_status": "scheduled"
        }
    ]
}'
```

response

200 ok 

<img width="2551" height="555" alt="image" src="https://github.com/user-attachments/assets/29c9f075-cbb6-4a3d-ace0-cee8a884644a" />

db entry into invoice table

<img width="578" height="449" alt="image" src="https://github.com/user-attachments/assets/23614dcf-70fc-480b-9922-0d5645e1d333" />

## 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:


Insert into subscription table
```
INSERT INTO public.subscription (
    id,
    status,
    billing_processor,
    payment_method_id,
    merchant_connector_id,
    client_secret,
    connector_subscription_id,
    merchant_id,
    customer_id,
    metadata,
    created_at,
    modified_at,
    profile_id,
    merchant_reference_id
) VALUES (
    '169vD0Uxi8JB746e',
    'active',
    'chargebee',
    'pm_98765',
    'mca_YH3XxYGMv6IbKD0icBie',
    'secret_subs_1231abcd',
    '169vD0Uxi8JB746e',
    'merchant_1758710612',
    'gaurav_test',
    '{"plan":"basic","trial":true}',
    NOW(),
    NOW(),
    'pro_RkLiGcSuaRjmxA5suSfV',
    'ref_subs_1231abcd'
);
```
incoming webhook for chargebee
```
curl --location 'http://localhost:8080/webhooks/merchant_1758710612/mca_YH3XxYGMv6IbKD0icBie' \
--header 'api-key: dev_5LNrrMNqyyFle8hvMssLob9pFq6OIEh0Vl3DqZttFXiDe7wrrScUePEYrIiYfSKT' \
--header 'Content-Type: application/json' \
--header 'authorization: Basic aHlwZXJzd2l0Y2g6aHlwZXJzd2l0Y2g=' \
--data-raw '{
    "api_version": "v2",
    "content": {
        "invoice": {
            "adjustment_credit_notes": [],
            "amount_adjusted": 0,
            "amount_due": 14100,
            "amount_paid": 0,
            "amount_to_collect": 14100,
            "applied_credits": [],
            "base_currency_code": "INR",
            "channel": "web",
            "credits_applied": 0,
            "currency_code": "INR",
            "customer_id": "gaurav_test",
            "date": 1758711043,
            "deleted": false,
            "due_date": 1758711043,
            "dunning_attempts": [],
            "exchange_rate": 1.0,
            "first_invoice": false,
            "generated_at": 1758711043,
            "has_advance_charges": false,
            "id": "3",
            "is_gifted": false,
            "issued_credit_notes": [],
            "line_items": [
                {
                    "amount": 14100,
                    "customer_id": "gaurav_test",
                    "date_from": 1758711043,
                    "date_to": 1761303043,
                    "description": "Enterprise Suite Monthly",
                    "discount_amount": 0,
                    "entity_id": "cbdemo_enterprise-suite-monthly",
                    "entity_type": "plan_item_price",
                    "id": "li_169vD0Uxi8JBp46g",
                    "is_taxed": false,
                    "item_level_discount_amount": 0,
                    "object": "line_item",
                    "pricing_model": "flat_fee",
                    "quantity": 1,
                    "subscription_id": "169vD0Uxi8JB746e",
                    "tax_amount": 0,
                    "tax_exempt_reason": "tax_not_configured",
                    "unit_amount": 14100
                }
            ],
            "linked_orders": [],
            "linked_payments": [],
            "net_term_days": 0,
            "new_sales_amount": 14100,
            "object": "invoice",
            "price_type": "tax_exclusive",
            "recurring": true,
            "reference_transactions": [],
            "resource_version": 1758711043846,
            "round_off_amount": 0,
            "site_details_at_creation": {
                "timezone": "Asia/Calcutta"
            },
            "status": "payment_due",
            "sub_total": 14100,
            "subscription_id": "169vD0Uxi8JB746e",
            "tax": 0,
            "term_finalized": true,
            "total": 14100,
            "updated_at": 1758711043,
            "write_off_amount": 0
        }
    },
    "event_type": "invoice_generated",
    "id": "ev_169vD0Uxi8JDf46i",
    "object": "event",
    "occurred_at": 1758711043,
    "source": "admin_console",
    "user": "ankit.gupta.001@juspay.in",
    "webhook_status": "scheduled",
    "webhooks": [
        {
            "id": "whv2_169mYOUxi3nvkZ8v",
            "object": "webhook",
            "webhook_status": "scheduled"
        }
    ]
}'
```

response

200 ok 

<img width="2551" height="555" alt="image" src="https://github.com/user-attachments/assets/29c9f075-cbb6-4a3d-ace0-cee8a884644a" />

db entry into invoice table

<img width="578" height="449" alt="image" src="https://github.com/user-attachments/assets/23614dcf-70fc-480b-9922-0d5645e1d333" />

Patch:
diff --git a/crates/api_models/src/webhooks.rs b/crates/api_models/src/webhooks.rs
index 5885f7fb98c..23212cb69f1 100644
--- a/crates/api_models/src/webhooks.rs
+++ b/crates/api_models/src/webhooks.rs
@@ -67,6 +67,7 @@ pub enum IncomingWebhookEvent {
     #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
     RecoveryInvoiceCancel,
     SetupWebhook,
+    InvoiceGenerated,
 }
 
 impl IncomingWebhookEvent {
@@ -300,6 +301,7 @@ impl From<IncomingWebhookEvent> for WebhookFlow {
             | IncomingWebhookEvent::RecoveryPaymentPending
             | IncomingWebhookEvent::RecoveryPaymentSuccess => Self::Recovery,
             IncomingWebhookEvent::SetupWebhook => Self::Setup,
+            IncomingWebhookEvent::InvoiceGenerated => Self::Subscription,
         }
     }
 }
@@ -341,6 +343,7 @@ pub enum ObjectReferenceId {
     PayoutId(PayoutIdType),
     #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
     InvoiceId(InvoiceIdType),
+    SubscriptionId(common_utils::id_type::SubscriptionId),
 }
 
 #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
@@ -388,7 +391,12 @@ impl ObjectReferenceId {
                 common_utils::errors::ValidationError::IncorrectValueProvided {
                     field_name: "PaymentId is required but received InvoiceId",
                 },
-            )
+            ),
+            Self::SubscriptionId(_) => Err(
+                common_utils::errors::ValidationError::IncorrectValueProvided {
+                    field_name: "PaymentId is required but received SubscriptionId",
+                },
+            ),
         }
     }
 }
diff --git a/crates/hyperswitch_connectors/src/connectors/chargebee.rs b/crates/hyperswitch_connectors/src/connectors/chargebee.rs
index 167c2a337ba..43773022596 100644
--- a/crates/hyperswitch_connectors/src/connectors/chargebee.rs
+++ b/crates/hyperswitch_connectors/src/connectors/chargebee.rs
@@ -9,8 +9,6 @@ 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_data_v2::RouterDataV2};
@@ -1319,17 +1317,25 @@ impl webhooks::IncomingWebhook for Chargebee {
             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),
+            api_models::webhooks::InvoiceIdType::ConnectorInvoiceId(
+                webhook.content.invoice.id.get_string_repr().to_string(),
+            ),
         ))
     }
     #[cfg(any(feature = "v1", not(all(feature = "revenue_recovery", feature = "v2"))))]
     fn get_webhook_object_reference_id(
         &self,
-        _request: &webhooks::IncomingWebhookRequestDetails<'_>,
+        request: &webhooks::IncomingWebhookRequestDetails<'_>,
     ) -> CustomResult<api_models::webhooks::ObjectReferenceId, errors::ConnectorError> {
-        Err(report!(errors::ConnectorError::WebhooksNotImplemented))
+        let webhook =
+            chargebee::ChargebeeInvoiceBody::get_invoice_webhook_data_from_body(request.body)
+                .change_context(errors::ConnectorError::WebhookReferenceIdNotFound)?;
+
+        let subscription_id = webhook.content.invoice.subscription_id;
+        Ok(api_models::webhooks::ObjectReferenceId::SubscriptionId(
+            subscription_id,
+        ))
     }
-    #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
     fn get_webhook_event_type(
         &self,
         request: &webhooks::IncomingWebhookRequestDetails<'_>,
@@ -1340,13 +1346,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,
@@ -1375,6 +1374,36 @@ impl webhooks::IncomingWebhook for Chargebee {
             transformers::ChargebeeInvoiceBody::get_invoice_webhook_data_from_body(request.body)?;
         revenue_recovery::RevenueRecoveryInvoiceData::try_from(webhook)
     }
+
+    fn get_subscription_mit_payment_data(
+        &self,
+        request: &webhooks::IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<
+        hyperswitch_domain_models::router_flow_types::SubscriptionMitPaymentData,
+        errors::ConnectorError,
+    > {
+        let webhook_body =
+            transformers::ChargebeeInvoiceBody::get_invoice_webhook_data_from_body(request.body)
+                .change_context(errors::ConnectorError::WebhookBodyDecodingFailed)
+                .attach_printable("Failed to parse Chargebee invoice webhook body")?;
+
+        let chargebee_mit_data = transformers::ChargebeeMitPaymentData::try_from(webhook_body)
+            .change_context(errors::ConnectorError::WebhookBodyDecodingFailed)
+            .attach_printable("Failed to extract MIT payment data from Chargebee webhook")?;
+
+        // Convert Chargebee-specific data to generic domain model
+        Ok(
+            hyperswitch_domain_models::router_flow_types::SubscriptionMitPaymentData {
+                invoice_id: chargebee_mit_data.invoice_id,
+                amount_due: chargebee_mit_data.amount_due,
+                currency_code: chargebee_mit_data.currency_code,
+                status: chargebee_mit_data.status.map(|s| s.into()),
+                customer_id: chargebee_mit_data.customer_id,
+                subscription_id: chargebee_mit_data.subscription_id,
+                first_invoice: chargebee_mit_data.first_invoice,
+            },
+        )
+    }
 }
 
 static CHARGEBEE_CONNECTOR_INFO: ConnectorInfo = ConnectorInfo {
diff --git a/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs b/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
index 64d9431c195..4c7b135f1a2 100644
--- a/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
+++ b/crates/hyperswitch_connectors/src/connectors/chargebee/transformers.rs
@@ -5,7 +5,7 @@ use common_enums::{connector_enums, enums};
 use common_utils::{
     errors::CustomResult,
     ext_traits::ByteSliceExt,
-    id_type::{CustomerId, SubscriptionId},
+    id_type::{CustomerId, InvoiceId, SubscriptionId},
     pii::{self, Email},
     types::MinorUnit,
 };
@@ -461,17 +461,21 @@ pub enum ChargebeeEventType {
     PaymentSucceeded,
     PaymentFailed,
     InvoiceDeleted,
+    InvoiceGenerated,
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct ChargebeeInvoiceData {
     // invoice id
-    pub id: String,
+    pub id: InvoiceId,
     pub total: MinorUnit,
     pub currency_code: enums::Currency,
     pub status: Option<ChargebeeInvoiceStatus>,
     pub billing_address: Option<ChargebeeInvoiceBillingAddress>,
     pub linked_payments: Option<Vec<ChargebeeInvoicePayments>>,
+    pub customer_id: CustomerId,
+    pub subscription_id: SubscriptionId,
+    pub first_invoice: Option<bool>,
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
@@ -585,7 +589,35 @@ impl ChargebeeInvoiceBody {
         Ok(webhook_body)
     }
 }
+// Structure to extract MIT payment data from invoice_generated webhook
+#[derive(Debug, Clone)]
+pub struct ChargebeeMitPaymentData {
+    pub invoice_id: InvoiceId,
+    pub amount_due: MinorUnit,
+    pub currency_code: enums::Currency,
+    pub status: Option<ChargebeeInvoiceStatus>,
+    pub customer_id: CustomerId,
+    pub subscription_id: SubscriptionId,
+    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,
+            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,
@@ -620,9 +652,10 @@ impl TryFrom<ChargebeeWebhookBody> for revenue_recovery::RevenueRecoveryAttemptD
     fn try_from(item: ChargebeeWebhookBody) -> Result<Self, Self::Error> {
         let amount = item.content.transaction.amount;
         let currency = item.content.transaction.currency_code.to_owned();
-        let merchant_reference_id =
-            common_utils::id_type::PaymentReferenceId::from_str(&item.content.invoice.id)
-                .change_context(errors::ConnectorError::WebhookBodyDecodingFailed)?;
+        let merchant_reference_id = common_utils::id_type::PaymentReferenceId::from_str(
+            item.content.invoice.id.get_string_repr(),
+        )
+        .change_context(errors::ConnectorError::WebhookBodyDecodingFailed)?;
         let connector_transaction_id = item
             .content
             .transaction
@@ -746,6 +779,19 @@ impl From<ChargebeeEventType> for api_models::webhooks::IncomingWebhookEvent {
             ChargebeeEventType::PaymentSucceeded => Self::RecoveryPaymentSuccess,
             ChargebeeEventType::PaymentFailed => Self::RecoveryPaymentFailure,
             ChargebeeEventType::InvoiceDeleted => Self::RecoveryInvoiceCancel,
+            ChargebeeEventType::InvoiceGenerated => Self::InvoiceGenerated,
+        }
+    }
+}
+
+#[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,
         }
     }
 }
@@ -754,9 +800,10 @@ impl From<ChargebeeEventType> for api_models::webhooks::IncomingWebhookEvent {
 impl TryFrom<ChargebeeInvoiceBody> for revenue_recovery::RevenueRecoveryInvoiceData {
     type Error = error_stack::Report<errors::ConnectorError>;
     fn try_from(item: ChargebeeInvoiceBody) -> Result<Self, Self::Error> {
-        let merchant_reference_id =
-            common_utils::id_type::PaymentReferenceId::from_str(&item.content.invoice.id)
-                .change_context(errors::ConnectorError::WebhookBodyDecodingFailed)?;
+        let merchant_reference_id = common_utils::id_type::PaymentReferenceId::from_str(
+            item.content.invoice.id.get_string_repr(),
+        )
+        .change_context(errors::ConnectorError::WebhookBodyDecodingFailed)?;
 
         // The retry count will never exceed u16 limit in a billing connector. It can have maximum of 12 in case of charge bee so its ok to suppress this
         #[allow(clippy::as_conversions)]
diff --git a/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs b/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs
index ac4e8388956..fcb94c9c5ab 100644
--- a/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs
+++ b/crates/hyperswitch_domain_models/src/router_flow_types/subscriptions.rs
@@ -1,3 +1,4 @@
+use common_enums::connector_enums::InvoiceStatus;
 #[derive(Debug, Clone)]
 pub struct SubscriptionCreate;
 #[derive(Debug, Clone)]
@@ -8,3 +9,15 @@ pub struct GetSubscriptionPlanPrices;
 
 #[derive(Debug, Clone)]
 pub struct GetSubscriptionEstimate;
+
+/// Generic structure for subscription MIT (Merchant Initiated Transaction) payment data
+#[derive(Debug, Clone)]
+pub struct SubscriptionMitPaymentData {
+    pub invoice_id: common_utils::id_type::InvoiceId,
+    pub amount_due: common_utils::types::MinorUnit,
+    pub currency_code: common_enums::enums::Currency,
+    pub status: Option<InvoiceStatus>,
+    pub customer_id: common_utils::id_type::CustomerId,
+    pub subscription_id: common_utils::id_type::SubscriptionId,
+    pub first_invoice: bool,
+}
diff --git a/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs b/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
index e60b7783cb5..3da78e63beb 100644
--- a/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
+++ b/crates/hyperswitch_domain_models/src/router_response_types/subscriptions.rs
@@ -16,7 +16,7 @@ pub struct SubscriptionCreateResponse {
 
 #[derive(Debug, Clone, serde::Serialize)]
 pub struct SubscriptionInvoiceData {
-    pub id: String,
+    pub id: id_type::InvoiceId,
     pub total: MinorUnit,
     pub currency_code: Currency,
     pub status: Option<common_enums::connector_enums::InvoiceStatus>,
diff --git a/crates/hyperswitch_interfaces/src/connector_integration_interface.rs b/crates/hyperswitch_interfaces/src/connector_integration_interface.rs
index 47ead156e2d..a876258f39f 100644
--- a/crates/hyperswitch_interfaces/src/connector_integration_interface.rs
+++ b/crates/hyperswitch_interfaces/src/connector_integration_interface.rs
@@ -416,6 +416,18 @@ impl IncomingWebhook for ConnectorEnum {
             Self::New(connector) => connector.get_revenue_recovery_attempt_details(request),
         }
     }
+    fn get_subscription_mit_payment_data(
+        &self,
+        request: &IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<
+        hyperswitch_domain_models::router_flow_types::SubscriptionMitPaymentData,
+        errors::ConnectorError,
+    > {
+        match self {
+            Self::Old(connector) => connector.get_subscription_mit_payment_data(request),
+            Self::New(connector) => connector.get_subscription_mit_payment_data(request),
+        }
+    }
 }
 
 impl ConnectorRedirectResponse for ConnectorEnum {
diff --git a/crates/hyperswitch_interfaces/src/webhooks.rs b/crates/hyperswitch_interfaces/src/webhooks.rs
index 3c7e3c04c30..7204330d6d5 100644
--- a/crates/hyperswitch_interfaces/src/webhooks.rs
+++ b/crates/hyperswitch_interfaces/src/webhooks.rs
@@ -301,4 +301,18 @@ pub trait IncomingWebhook: ConnectorCommon + Sync {
         )
         .into())
     }
+
+    /// get subscription MIT payment data from webhook
+    fn get_subscription_mit_payment_data(
+        &self,
+        _request: &IncomingWebhookRequestDetails<'_>,
+    ) -> CustomResult<
+        hyperswitch_domain_models::router_flow_types::SubscriptionMitPaymentData,
+        errors::ConnectorError,
+    > {
+        Err(errors::ConnectorError::NotImplemented(
+            "get_subscription_mit_payment_data method".to_string(),
+        )
+        .into())
+    }
 }
diff --git a/crates/router/src/core/subscription.rs b/crates/router/src/core/subscription.rs
index 71e32c43bf6..5c53a60c5d7 100644
--- a/crates/router/src/core/subscription.rs
+++ b/crates/router/src/core/subscription.rs
@@ -42,7 +42,7 @@ pub async fn create_subscription(
     let billing_handler =
         BillingHandler::create(&state, &merchant_context, customer, profile.clone()).await?;
 
-    let subscription_handler = SubscriptionHandler::new(&state, &merchant_context, profile);
+    let subscription_handler = SubscriptionHandler::new(&state, &merchant_context);
     let mut subscription = subscription_handler
         .create_subscription_entry(
             subscription_id,
@@ -50,10 +50,11 @@ pub async fn create_subscription(
             billing_handler.connector_data.connector_name,
             billing_handler.merchant_connector_id.clone(),
             request.merchant_reference_id.clone(),
+            &profile.clone(),
         )
         .await
         .attach_printable("subscriptions: failed to create subscription entry")?;
-    let invoice_handler = subscription.get_invoice_handler();
+    let invoice_handler = subscription.get_invoice_handler(profile.clone());
     let payment = invoice_handler
         .create_payment_with_confirm_false(subscription.handler.state, &request)
         .await
@@ -107,7 +108,7 @@ pub async fn create_and_confirm_subscription(
 
     let billing_handler =
         BillingHandler::create(&state, &merchant_context, customer, profile.clone()).await?;
-    let subscription_handler = SubscriptionHandler::new(&state, &merchant_context, profile.clone());
+    let subscription_handler = SubscriptionHandler::new(&state, &merchant_context);
     let mut subs_handler = subscription_handler
         .create_subscription_entry(
             subscription_id.clone(),
@@ -115,10 +116,11 @@ pub async fn create_and_confirm_subscription(
             billing_handler.connector_data.connector_name,
             billing_handler.merchant_connector_id.clone(),
             request.merchant_reference_id.clone(),
+            &profile.clone(),
         )
         .await
         .attach_printable("subscriptions: failed to create subscription entry")?;
-    let invoice_handler = subs_handler.get_invoice_handler();
+    let invoice_handler = subs_handler.get_invoice_handler(profile.clone());
 
     let _customer_create_response = billing_handler
         .create_customer_on_connector(
@@ -210,9 +212,9 @@ pub async fn confirm_subscription(
             .await
             .attach_printable("subscriptions: failed to find customer")?;
 
-    let handler = SubscriptionHandler::new(&state, &merchant_context, profile.clone());
+    let handler = SubscriptionHandler::new(&state, &merchant_context);
     let mut subscription_entry = handler.find_subscription(subscription_id).await?;
-    let invoice_handler = subscription_entry.get_invoice_handler();
+    let invoice_handler = subscription_entry.get_invoice_handler(profile.clone());
     let invoice = invoice_handler
         .get_latest_invoice(&state)
         .await
@@ -230,8 +232,8 @@ pub async fn confirm_subscription(
         .await?;
 
     let billing_handler =
-        BillingHandler::create(&state, &merchant_context, customer, profile).await?;
-    let invoice_handler = subscription_entry.get_invoice_handler();
+        BillingHandler::create(&state, &merchant_context, customer, profile.clone()).await?;
+    let invoice_handler = subscription_entry.get_invoice_handler(profile);
     let subscription = subscription_entry.subscription.clone();
 
     let _customer_create_response = billing_handler
@@ -296,13 +298,13 @@ pub async fn get_subscription(
     profile_id: common_utils::id_type::ProfileId,
     subscription_id: common_utils::id_type::SubscriptionId,
 ) -> RouterResponse<SubscriptionResponse> {
-    let profile =
+    let _profile =
         SubscriptionHandler::find_business_profile(&state, &merchant_context, &profile_id)
             .await
             .attach_printable(
                 "subscriptions: failed to find business profile in get_subscription",
             )?;
-    let handler = SubscriptionHandler::new(&state, &merchant_context, profile);
+    let handler = SubscriptionHandler::new(&state, &merchant_context);
     let subscription = handler
         .find_subscription(subscription_id)
         .await
diff --git a/crates/router/src/core/subscription/subscription_handler.rs b/crates/router/src/core/subscription/subscription_handler.rs
index 25259a47696..24c08313b82 100644
--- a/crates/router/src/core/subscription/subscription_handler.rs
+++ b/crates/router/src/core/subscription/subscription_handler.rs
@@ -14,24 +14,23 @@ use hyperswitch_domain_models::{
 use masking::Secret;
 
 use super::errors;
-use crate::{core::subscription::invoice_handler::InvoiceHandler, routes::SessionState};
+use crate::{
+    core::{errors::StorageErrorExt, subscription::invoice_handler::InvoiceHandler},
+    db::CustomResult,
+    routes::SessionState,
+    types::domain,
+};
 
 pub struct SubscriptionHandler<'a> {
     pub state: &'a SessionState,
     pub merchant_context: &'a MerchantContext,
-    pub profile: hyperswitch_domain_models::business_profile::Profile,
 }
 
 impl<'a> SubscriptionHandler<'a> {
-    pub fn new(
-        state: &'a SessionState,
-        merchant_context: &'a MerchantContext,
-        profile: hyperswitch_domain_models::business_profile::Profile,
-    ) -> Self {
+    pub fn new(state: &'a SessionState, merchant_context: &'a MerchantContext) -> Self {
         Self {
             state,
             merchant_context,
-            profile,
         }
     }
 
@@ -43,6 +42,7 @@ impl<'a> SubscriptionHandler<'a> {
         billing_processor: connector_enums::Connector,
         merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
         merchant_reference_id: Option<String>,
+        profile: &hyperswitch_domain_models::business_profile::Profile,
     ) -> errors::RouterResult<SubscriptionWithHandler<'_>> {
         let store = self.state.store.clone();
         let db = store.as_ref();
@@ -61,7 +61,7 @@ impl<'a> SubscriptionHandler<'a> {
                 .clone(),
             customer_id.clone(),
             None,
-            self.profile.get_id().clone(),
+            profile.get_id().clone(),
             merchant_reference_id,
         );
 
@@ -76,7 +76,6 @@ impl<'a> SubscriptionHandler<'a> {
         Ok(SubscriptionWithHandler {
             handler: self,
             subscription: new_subscription,
-            profile: self.profile.clone(),
             merchant_account: self.merchant_context.get_merchant_account().clone(),
         })
     }
@@ -145,7 +144,6 @@ impl<'a> SubscriptionHandler<'a> {
         Ok(SubscriptionWithHandler {
             handler: self,
             subscription,
-            profile: self.profile.clone(),
             merchant_account: self.merchant_context.get_merchant_account().clone(),
         })
     }
@@ -153,7 +151,6 @@ impl<'a> SubscriptionHandler<'a> {
 pub struct SubscriptionWithHandler<'a> {
     pub handler: &'a SubscriptionHandler<'a>,
     pub subscription: diesel_models::subscription::Subscription,
-    pub profile: hyperswitch_domain_models::business_profile::Profile,
     pub merchant_account: hyperswitch_domain_models::merchant_account::MerchantAccount,
 }
 
@@ -237,11 +234,83 @@ impl SubscriptionWithHandler<'_> {
         Ok(())
     }
 
-    pub fn get_invoice_handler(&self) -> InvoiceHandler {
+    pub fn get_invoice_handler(
+        &self,
+        profile: hyperswitch_domain_models::business_profile::Profile,
+    ) -> InvoiceHandler {
         InvoiceHandler {
             subscription: self.subscription.clone(),
             merchant_account: self.merchant_account.clone(),
-            profile: self.profile.clone(),
+            profile,
+        }
+    }
+    pub async fn get_mca(
+        &mut self,
+        connector_name: &str,
+    ) -> CustomResult<domain::MerchantConnectorAccount, errors::ApiErrorResponse> {
+        let db = self.handler.state.store.as_ref();
+        let key_manager_state = &(self.handler.state).into();
+
+        match &self.subscription.merchant_connector_id {
+            Some(merchant_connector_id) => {
+                #[cfg(feature = "v1")]
+                {
+                    db.find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+                        key_manager_state,
+                        self.handler
+                            .merchant_context
+                            .get_merchant_account()
+                            .get_id(),
+                        merchant_connector_id,
+                        self.handler.merchant_context.get_merchant_key_store(),
+                    )
+                    .await
+                    .to_not_found_response(
+                        errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+                            id: merchant_connector_id.get_string_repr().to_string(),
+                        },
+                    )
+                }
+                #[cfg(feature = "v2")]
+                {
+                    //get mca using id
+                    let _ = key_manager_state;
+                    let _ = connector_name;
+                    let _ = merchant_context.get_merchant_key_store();
+                    let _ = subscription.profile_id;
+                    todo!()
+                }
+            }
+            None => {
+                // Fallback to profile-based lookup when merchant_connector_id is not set
+                #[cfg(feature = "v1")]
+                {
+                    db.find_merchant_connector_account_by_profile_id_connector_name(
+                        key_manager_state,
+                        &self.subscription.profile_id,
+                        connector_name,
+                        self.handler.merchant_context.get_merchant_key_store(),
+                    )
+                    .await
+                    .to_not_found_response(
+                        errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+                            id: format!(
+                                "profile_id {} and connector_name {connector_name}",
+                                self.subscription.profile_id.get_string_repr()
+                            ),
+                        },
+                    )
+                }
+                #[cfg(feature = "v2")]
+                {
+                    //get mca using id
+                    let _ = key_manager_state;
+                    let _ = connector_name;
+                    let _ = self.handler.merchant_context.get_merchant_key_store();
+                    let _ = self.subscription.profile_id;
+                    todo!()
+                }
+            }
         }
     }
 }
diff --git a/crates/router/src/core/webhooks/incoming.rs b/crates/router/src/core/webhooks/incoming.rs
index d1203cd1a34..f207c243d26 100644
--- a/crates/router/src/core/webhooks/incoming.rs
+++ b/crates/router/src/core/webhooks/incoming.rs
@@ -3,7 +3,11 @@ use std::{str::FromStr, time::Instant};
 use actix_web::FromRequest;
 #[cfg(feature = "payouts")]
 use api_models::payouts as payout_models;
-use api_models::webhooks::{self, WebhookResponseTracker};
+use api_models::{
+    enums::Connector,
+    webhooks::{self, WebhookResponseTracker},
+};
+pub use common_enums::{connector_enums::InvoiceStatus, enums::ProcessTrackerRunner};
 use common_utils::{
     errors::ReportSwitchExt,
     events::ApiEventsType,
@@ -30,7 +34,9 @@ use crate::{
         errors::{self, ConnectorErrorExt, CustomResult, RouterResponse, StorageErrorExt},
         metrics, payment_methods,
         payments::{self, tokenization},
-        refunds, relay, unified_connector_service, utils as core_utils,
+        refunds, relay,
+        subscription::subscription_handler::SubscriptionHandler,
+        unified_connector_service, utils as core_utils,
         webhooks::{network_tokenization_incoming, utils::construct_webhook_router_data},
     },
     db::StorageInterface,
@@ -253,6 +259,7 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                     &request_details,
                     merchant_context.get_merchant_account().get_id(),
                     merchant_connector_account
+                        .clone()
                         .and_then(|mca| mca.connector_webhook_details.clone()),
                     &connector_name,
                 )
@@ -342,6 +349,11 @@ async fn incoming_webhooks_core<W: types::OutgoingWebhookType>(
                 &webhook_processing_result.transform_data,
                 &final_request_details,
                 is_relay_webhook,
+                merchant_connector_account
+                    .ok_or(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+                        id: connector_name_or_mca_id.to_string(),
+                    })?
+                    .merchant_connector_id,
             )
             .await;
 
@@ -524,12 +536,13 @@ 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)
         .switch()
         .attach_printable("Could not find object reference id in incoming webhook body")?;
-    let connector_enum = api_models::enums::Connector::from_str(connector_name)
+    let connector_enum = Connector::from_str(connector_name)
         .change_context(errors::ApiErrorResponse::InvalidDataValue {
             field_name: "connector",
         })
@@ -814,6 +827,21 @@ async fn process_webhook_business_logic(
             .await
             .attach_printable("Incoming webhook flow for payouts failed"),
 
+            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"),
+
             _ => Err(errors::ApiErrorResponse::InternalServerError)
                 .attach_printable("Unsupported Flow Type received in incoming webhooks"),
         }
@@ -845,7 +873,7 @@ fn handle_incoming_webhook_error(
     logger::error!(?error, "Incoming webhook flow failed");
 
     // fetch the connector enum from the connector name
-    let connector_enum = api_models::connector_enums::Connector::from_str(connector_name)
+    let connector_enum = Connector::from_str(connector_name)
         .change_context(errors::ApiErrorResponse::InvalidDataValue {
             field_name: "connector",
         })
@@ -2533,3 +2561,92 @@ 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_enum: &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
+        ));
+    }
+
+    let connector_name = connector_enum.id().to_string();
+
+    let connector = Connector::from_str(&connector_name)
+        .change_context(errors::ConnectorError::InvalidConnectorName)
+        .change_context(errors::ApiErrorResponse::IncorrectConnectorNameGiven)
+        .attach_printable_lazy(|| format!("unable to parse connector name {connector_name}"))?;
+
+    let mit_payment_data = connector_enum
+        .get_subscription_mit_payment_data(request_details)
+        .change_context(errors::ApiErrorResponse::WebhookProcessingFailure)
+        .attach_printable("Failed to extract MIT payment data from subscription webhook")?;
+
+    if mit_payment_data.first_invoice {
+        return Ok(WebhookResponseTracker::NoEffect);
+    }
+
+    let profile_id = business_profile.get_id().clone();
+
+    let profile =
+        SubscriptionHandler::find_business_profile(&state, &merchant_context, &profile_id)
+            .await
+            .attach_printable(
+                "subscriptions: failed to find business profile in get_subscription",
+            )?;
+
+    let handler = SubscriptionHandler::new(&state, &merchant_context);
+
+    let subscription_id = mit_payment_data.subscription_id.clone();
+
+    let subscription_with_handler = handler
+        .find_subscription(subscription_id)
+        .await
+        .attach_printable("subscriptions: failed to get subscription entry in get_subscription")?;
+
+    let invoice_handler = subscription_with_handler.get_invoice_handler(profile);
+
+    let payment_method_id = subscription_with_handler
+        .subscription
+        .payment_method_id
+        .clone()
+        .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
+            message: "No payment method found for subscription".to_string(),
+        })
+        .attach_printable("No payment method found for subscription")?;
+
+    logger::info!("Payment method ID found: {}", payment_method_id);
+
+    let _invoice_new = invoice_handler
+        .create_invoice_entry(
+            &state,
+            billing_connector_mca_id.clone(),
+            None,
+            mit_payment_data.amount_due,
+            mit_payment_data.currency_code,
+            InvoiceStatus::PaymentPending,
+            connector,
+            None,
+        )
+        .await?;
+
+    Ok(WebhookResponseTracker::NoEffect)
+}
diff --git a/crates/router/src/core/webhooks/recovery_incoming.rs b/crates/router/src/core/webhooks/recovery_incoming.rs
index fab407e4161..4a8ce622255 100644
--- a/crates/router/src/core/webhooks/recovery_incoming.rs
+++ b/crates/router/src/core/webhooks/recovery_incoming.rs
@@ -1456,6 +1456,7 @@ impl RecoveryAction {
             | webhooks::IncomingWebhookEvent::PayoutCreated
             | webhooks::IncomingWebhookEvent::PayoutExpired
             | webhooks::IncomingWebhookEvent::PayoutReversed
+            | webhooks::IncomingWebhookEvent::InvoiceGenerated
             | webhooks::IncomingWebhookEvent::SetupWebhook => {
                 common_types::payments::RecoveryAction::InvalidAction
             }
diff --git a/crates/router/src/utils.rs b/crates/router/src/utils.rs
index 71f1ed404b2..123d723e566 100644
--- a/crates/router/src/utils.rs
+++ b/crates/router/src/utils.rs
@@ -44,6 +44,8 @@ use serde_json::Value;
 use tracing_futures::Instrument;
 
 pub use self::ext_traits::{OptionExt, ValidateCall};
+#[cfg(feature = "v1")]
+use crate::core::subscription::subscription_handler::SubscriptionHandler;
 use crate::{
     consts,
     core::{
@@ -459,7 +461,6 @@ pub async fn get_mca_from_payment_intent(
         }
     }
 }
-
 #[cfg(feature = "payouts")]
 pub async fn get_mca_from_payout_attempt(
     state: &SessionState,
@@ -639,6 +640,22 @@ pub async fn get_mca_from_object_reference_id(
                 )
                 .await
             }
+            webhooks::ObjectReferenceId::SubscriptionId(subscription_id_type) => {
+                #[cfg(feature = "v1")]
+                {
+                    let subscription_handler = SubscriptionHandler::new(state, merchant_context);
+                    let mut subscription_with_handler = subscription_handler
+                        .find_subscription(subscription_id_type)
+                        .await?;
+
+                    subscription_with_handler.get_mca(connector_name).await
+                }
+                #[cfg(feature = "v2")]
+                {
+                    let _db = db;
+                    todo!()
+                }
+            }
             #[cfg(feature = "payouts")]
             webhooks::ObjectReferenceId::PayoutId(payout_id_type) => {
                 get_mca_from_payout_attempt(state, merchant_context, payout_id_type, connector_name)