Issue:
Bug: Add invoice table in hyperswitch





Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
<img width="539" height="305" alt="Screenshot 2025-09-18 at 22 35 30" src="https://github.com/user-attachments/assets/6ef6a861-7bb8-459e-9ac2-99791ec94c21" />


## Description
This PR introduces the **`invoice` table** along with supporting models, queries, schema definitions, and storage interfaces.  
The goal is to enable **subscription billing and payment tracking** by storing invoices linked to merchants, customers, subscriptions, and connectors.

### Key Features
- **New `Invoice` table** with fields for subscription, connector, merchant, customer, amount, currency, status, and metadata.
- Added **`InvoiceStatus` enum** to represent invoice lifecycle stages (e.g., `PendingPayment`, `PaymentSucceeded`, `InvoicePaid`, etc.).
- Implemented **Diesel models**: `InvoiceNew`, `Invoice`, `InvoiceUpdate`.
- Added **storage query methods**:
  - Insert new invoice
  - Find invoice by ID
  - Find invoice by merchant + invoice ID
  - Update invoice entry
- Integrated with **router storage interface** (`InvoiceInterface`).
- Added **migrations**:
  - `up.sql`: creates `invoice` table and indexes (`payment_intent_id`, `subscription_id`, `merchant_id + customer_id`).
  - `down.sql`: drops indexes and table.

---

## Motivation and Context
- Provides a foundation for **subscription billing system** in Hyperswitch.
- Supports **payment reconciliation and invoice lifecycle tracking**.
- Enables robust query and update flows for invoices tied to subscriptions.

---

## Schema Details
**Table: `invoice`**
- `id` (PK)
- `subscription_id`
- `connector_subscription_id`
- `merchant_id`
- `profile_id`
- `merchant_connector_id`
- `payment_intent_id` (unique)
- `payment_method_id`
- `customer_id`
- `amount`
- `currency`
- `status`
- `provider_name`
- `metadata`
- `created_at`
- `modified_at`

**Indexes**
- `idx_payment_intent_id`
- `idx_subscription_id`
- `idx_merchant_customer`

---


## 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)?
-->
This can't be tested as this is only creation of tables and their respective storage models.

## 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:

This can't be tested as this is only creation of tables and their respective storage models.

Patch:
diff --git a/crates/common_enums/src/connector_enums.rs b/crates/common_enums/src/connector_enums.rs
index 9991ba8a27e..89d2ed7d89c 100644
--- a/crates/common_enums/src/connector_enums.rs
+++ b/crates/common_enums/src/connector_enums.rs
@@ -875,3 +875,16 @@ impl TryFrom<Connector> for RoutableConnectors {
         }
     }
 }
+
+// Enum representing different status an invoice can have.
+#[derive(Debug, Clone, PartialEq, Eq, strum::Display, strum::EnumString)]
+pub enum InvoiceStatus {
+    InvoiceCreated,
+    PaymentPending,
+    PaymentPendingTimeout,
+    PaymentSucceeded,
+    PaymentFailed,
+    PaymentCanceled,
+    InvoicePaid,
+    ManualReview,
+}
diff --git a/crates/diesel_models/src/invoice.rs b/crates/diesel_models/src/invoice.rs
new file mode 100644
index 00000000000..24534629307
--- /dev/null
+++ b/crates/diesel_models/src/invoice.rs
@@ -0,0 +1,108 @@
+use common_enums::connector_enums::{Connector, InvoiceStatus};
+use common_utils::{pii::SecretSerdeValue, types::MinorUnit};
+use diesel::{AsChangeset, Identifiable, Insertable, Queryable, Selectable};
+use serde::{Deserialize, Serialize};
+
+use crate::schema::invoice;
+
+#[derive(Clone, Debug, Eq, Insertable, PartialEq, Serialize, Deserialize)]
+#[diesel(table_name = invoice, check_for_backend(diesel::pg::Pg))]
+pub struct InvoiceNew {
+    pub id: String,
+    pub subscription_id: String,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub profile_id: common_utils::id_type::ProfileId,
+    pub merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
+    pub payment_intent_id: Option<common_utils::id_type::PaymentId>,
+    pub payment_method_id: Option<String>,
+    pub customer_id: common_utils::id_type::CustomerId,
+    pub amount: MinorUnit,
+    pub currency: String,
+    pub status: String,
+    pub provider_name: Connector,
+    pub metadata: Option<SecretSerdeValue>,
+    pub created_at: time::PrimitiveDateTime,
+    pub modified_at: time::PrimitiveDateTime,
+}
+
+#[derive(
+    Clone, Debug, Eq, PartialEq, Identifiable, Queryable, Selectable, Deserialize, Serialize,
+)]
+#[diesel(
+    table_name = invoice,
+    primary_key(id),
+    check_for_backend(diesel::pg::Pg)
+)]
+pub struct Invoice {
+    id: String,
+    subscription_id: String,
+    merchant_id: common_utils::id_type::MerchantId,
+    profile_id: common_utils::id_type::ProfileId,
+    merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
+    payment_intent_id: Option<common_utils::id_type::PaymentId>,
+    payment_method_id: Option<String>,
+    customer_id: common_utils::id_type::CustomerId,
+    amount: MinorUnit,
+    currency: String,
+    status: String,
+    provider_name: Connector,
+    metadata: Option<SecretSerdeValue>,
+    created_at: time::PrimitiveDateTime,
+    modified_at: time::PrimitiveDateTime,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, AsChangeset, Deserialize)]
+#[diesel(table_name = invoice)]
+pub struct InvoiceUpdate {
+    pub status: Option<String>,
+    pub payment_method_id: Option<String>,
+    pub modified_at: time::PrimitiveDateTime,
+}
+
+impl InvoiceNew {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        id: String,
+        subscription_id: String,
+        merchant_id: common_utils::id_type::MerchantId,
+        profile_id: common_utils::id_type::ProfileId,
+        merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
+        payment_intent_id: Option<common_utils::id_type::PaymentId>,
+        payment_method_id: Option<String>,
+        customer_id: common_utils::id_type::CustomerId,
+        amount: MinorUnit,
+        currency: String,
+        status: InvoiceStatus,
+        provider_name: Connector,
+        metadata: Option<SecretSerdeValue>,
+    ) -> Self {
+        let now = common_utils::date_time::now();
+        Self {
+            id,
+            subscription_id,
+            merchant_id,
+            profile_id,
+            merchant_connector_id,
+            payment_intent_id,
+            payment_method_id,
+            customer_id,
+            amount,
+            currency,
+            status: status.to_string(),
+            provider_name,
+            metadata,
+            created_at: now,
+            modified_at: now,
+        }
+    }
+}
+
+impl InvoiceUpdate {
+    pub fn new(payment_method_id: Option<String>, status: Option<InvoiceStatus>) -> Self {
+        Self {
+            payment_method_id,
+            status: status.map(|status| status.to_string()),
+            modified_at: common_utils::date_time::now(),
+        }
+    }
+}
diff --git a/crates/diesel_models/src/lib.rs b/crates/diesel_models/src/lib.rs
index b715134f926..d29eeac41cb 100644
--- a/crates/diesel_models/src/lib.rs
+++ b/crates/diesel_models/src/lib.rs
@@ -24,6 +24,7 @@ pub mod fraud_check;
 pub mod generic_link;
 pub mod gsm;
 pub mod hyperswitch_ai_interaction;
+pub mod invoice;
 #[cfg(feature = "kv_store")]
 pub mod kv;
 pub mod locker_mock_up;
diff --git a/crates/diesel_models/src/query.rs b/crates/diesel_models/src/query.rs
index 2dc6c18a1a1..9bf6adc40b2 100644
--- a/crates/diesel_models/src/query.rs
+++ b/crates/diesel_models/src/query.rs
@@ -22,6 +22,7 @@ pub mod generic_link;
 pub mod generics;
 pub mod gsm;
 pub mod hyperswitch_ai_interaction;
+pub mod invoice;
 pub mod locker_mock_up;
 pub mod mandate;
 pub mod merchant_account;
diff --git a/crates/diesel_models/src/query/invoice.rs b/crates/diesel_models/src/query/invoice.rs
new file mode 100644
index 00000000000..10df8620461
--- /dev/null
+++ b/crates/diesel_models/src/query/invoice.rs
@@ -0,0 +1,41 @@
+use diesel::{associations::HasTable, ExpressionMethods};
+
+use super::generics;
+use crate::{
+    invoice::{Invoice, InvoiceNew, InvoiceUpdate},
+    schema::invoice::dsl,
+    PgPooledConn, StorageResult,
+};
+
+impl InvoiceNew {
+    pub async fn insert(self, conn: &PgPooledConn) -> StorageResult<Invoice> {
+        generics::generic_insert(conn, self).await
+    }
+}
+
+impl Invoice {
+    pub async fn find_invoice_by_id_invoice_id(
+        conn: &PgPooledConn,
+        id: String,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::id.eq(id.to_owned()),
+        )
+        .await
+    }
+
+    pub async fn update_invoice_entry(
+        conn: &PgPooledConn,
+        id: String,
+        invoice_update: InvoiceUpdate,
+    ) -> StorageResult<Self> {
+        generics::generic_update_with_unique_predicate_get_result::<
+            <Self as HasTable>::Table,
+            _,
+            _,
+            _,
+        >(conn, dsl::id.eq(id.to_owned()), invoice_update)
+        .await
+    }
+}
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index 36345ef7570..5581856b56d 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -715,6 +715,40 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    invoice (id) {
+        #[max_length = 64]
+        id -> Varchar,
+        #[max_length = 128]
+        subscription_id -> Varchar,
+        #[max_length = 64]
+        merchant_id -> Varchar,
+        #[max_length = 64]
+        profile_id -> Varchar,
+        #[max_length = 128]
+        merchant_connector_id -> Varchar,
+        #[max_length = 64]
+        payment_intent_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        payment_method_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        customer_id -> Varchar,
+        amount -> Int8,
+        #[max_length = 3]
+        currency -> Varchar,
+        #[max_length = 64]
+        status -> Varchar,
+        #[max_length = 128]
+        provider_name -> Varchar,
+        metadata -> Nullable<Jsonb>,
+        created_at -> Timestamp,
+        modified_at -> Timestamp,
+    }
+}
+
 diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
@@ -1730,6 +1764,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     hyperswitch_ai_interaction,
     hyperswitch_ai_interaction_default,
     incremental_authorization,
+    invoice,
     locker_mock_up,
     mandate,
     merchant_account,
diff --git a/crates/diesel_models/src/schema_v2.rs b/crates/diesel_models/src/schema_v2.rs
index f05b4e460c3..11c23b0169a 100644
--- a/crates/diesel_models/src/schema_v2.rs
+++ b/crates/diesel_models/src/schema_v2.rs
@@ -729,6 +729,40 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::enums::diesel_exports::*;
+
+    invoice (id) {
+        #[max_length = 64]
+        id -> Varchar,
+        #[max_length = 128]
+        subscription_id -> Varchar,
+        #[max_length = 64]
+        merchant_id -> Varchar,
+        #[max_length = 64]
+        profile_id -> Varchar,
+        #[max_length = 128]
+        merchant_connector_id -> Varchar,
+        #[max_length = 64]
+        payment_intent_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        payment_method_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        customer_id -> Varchar,
+        amount -> Int8,
+        #[max_length = 3]
+        currency -> Varchar,
+        #[max_length = 64]
+        status -> Varchar,
+        #[max_length = 128]
+        provider_name -> Varchar,
+        metadata -> Nullable<Jsonb>,
+        created_at -> Timestamp,
+        modified_at -> Timestamp,
+    }
+}
+
 diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
@@ -1685,6 +1719,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     hyperswitch_ai_interaction,
     hyperswitch_ai_interaction_default,
     incremental_authorization,
+    invoice,
     locker_mock_up,
     mandate,
     merchant_account,
diff --git a/crates/router/src/db.rs b/crates/router/src/db.rs
index ab3b7cba577..e43c12da150 100644
--- a/crates/router/src/db.rs
+++ b/crates/router/src/db.rs
@@ -21,6 +21,7 @@ pub mod generic_link;
 pub mod gsm;
 pub mod health_check;
 pub mod hyperswitch_ai_interaction;
+pub mod invoice;
 pub mod kafka_store;
 pub mod locker_mock_up;
 pub mod mandate;
@@ -147,6 +148,7 @@ pub trait StorageInterface:
     + tokenization::TokenizationInterface
     + callback_mapper::CallbackMapperInterface
     + subscription::SubscriptionInterface
+    + invoice::InvoiceInterface
     + 'static
 {
     fn get_scheduler_db(&self) -> Box<dyn scheduler::SchedulerInterface>;
diff --git a/crates/router/src/db/invoice.rs b/crates/router/src/db/invoice.rs
new file mode 100644
index 00000000000..850f1c52730
--- /dev/null
+++ b/crates/router/src/db/invoice.rs
@@ -0,0 +1,126 @@
+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 InvoiceInterface {
+    async fn insert_invoice_entry(
+        &self,
+        invoice_new: storage::invoice::InvoiceNew,
+    ) -> CustomResult<storage::Invoice, errors::StorageError>;
+
+    async fn find_invoice_by_invoice_id(
+        &self,
+        invoice_id: String,
+    ) -> CustomResult<storage::Invoice, errors::StorageError>;
+
+    async fn update_invoice_entry(
+        &self,
+        invoice_id: String,
+        data: storage::invoice::InvoiceUpdate,
+    ) -> CustomResult<storage::Invoice, errors::StorageError>;
+}
+
+#[async_trait::async_trait]
+impl InvoiceInterface for Store {
+    #[instrument(skip_all)]
+    async fn insert_invoice_entry(
+        &self,
+        invoice_new: storage::invoice::InvoiceNew,
+    ) -> CustomResult<storage::Invoice, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        invoice_new
+            .insert(&conn)
+            .await
+            .map_err(|error| report!(errors::StorageError::from(error)))
+    }
+
+    #[instrument(skip_all)]
+    async fn find_invoice_by_invoice_id(
+        &self,
+        invoice_id: String,
+    ) -> CustomResult<storage::Invoice, errors::StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+        storage::Invoice::find_invoice_by_id_invoice_id(&conn, invoice_id)
+            .await
+            .map_err(|error| report!(errors::StorageError::from(error)))
+    }
+
+    #[instrument(skip_all)]
+    async fn update_invoice_entry(
+        &self,
+        invoice_id: String,
+        data: storage::invoice::InvoiceUpdate,
+    ) -> CustomResult<storage::Invoice, errors::StorageError> {
+        let conn = connection::pg_connection_write(self).await?;
+        storage::Invoice::update_invoice_entry(&conn, invoice_id, data)
+            .await
+            .map_err(|error| report!(errors::StorageError::from(error)))
+    }
+}
+
+#[async_trait::async_trait]
+impl InvoiceInterface for MockDb {
+    #[instrument(skip_all)]
+    async fn insert_invoice_entry(
+        &self,
+        _invoice_new: storage::invoice::InvoiceNew,
+    ) -> CustomResult<storage::Invoice, errors::StorageError> {
+        Err(errors::StorageError::MockDbError)?
+    }
+
+    async fn find_invoice_by_invoice_id(
+        &self,
+        _invoice_id: String,
+    ) -> CustomResult<storage::Invoice, errors::StorageError> {
+        Err(errors::StorageError::MockDbError)?
+    }
+
+    async fn update_invoice_entry(
+        &self,
+        _invoice_id: String,
+        _data: storage::invoice::InvoiceUpdate,
+    ) -> CustomResult<storage::Invoice, errors::StorageError> {
+        Err(errors::StorageError::MockDbError)?
+    }
+}
+
+#[async_trait::async_trait]
+impl InvoiceInterface for KafkaStore {
+    #[instrument(skip_all)]
+    async fn insert_invoice_entry(
+        &self,
+        invoice_new: storage::invoice::InvoiceNew,
+    ) -> CustomResult<storage::Invoice, errors::StorageError> {
+        self.diesel_store.insert_invoice_entry(invoice_new).await
+    }
+
+    #[instrument(skip_all)]
+    async fn find_invoice_by_invoice_id(
+        &self,
+        invoice_id: String,
+    ) -> CustomResult<storage::Invoice, errors::StorageError> {
+        self.diesel_store
+            .find_invoice_by_invoice_id(invoice_id)
+            .await
+    }
+
+    #[instrument(skip_all)]
+    async fn update_invoice_entry(
+        &self,
+        invoice_id: String,
+        data: storage::invoice::InvoiceUpdate,
+    ) -> CustomResult<storage::Invoice, errors::StorageError> {
+        self.diesel_store
+            .update_invoice_entry(invoice_id, data)
+            .await
+    }
+}
diff --git a/crates/router/src/types/storage.rs b/crates/router/src/types/storage.rs
index e9c26dd30c2..bc5fa611d3e 100644
--- a/crates/router/src/types/storage.rs
+++ b/crates/router/src/types/storage.rs
@@ -22,6 +22,7 @@ pub mod fraud_check;
 pub mod generic_link;
 pub mod gsm;
 pub mod hyperswitch_ai_interaction;
+pub mod invoice;
 #[cfg(feature = "kv_store")]
 pub mod kv;
 pub mod locker_mock_up;
@@ -76,9 +77,9 @@ pub use self::{
     blocklist_fingerprint::*, blocklist_lookup::*, business_profile::*, callback_mapper::*,
     capture::*, cards_info::*, configs::*, customers::*, dashboard_metadata::*, dispute::*,
     dynamic_routing_stats::*, ephemeral_key::*, events::*, file::*, fraud_check::*,
-    generic_link::*, gsm::*, hyperswitch_ai_interaction::*, locker_mock_up::*, mandate::*,
-    merchant_account::*, merchant_connector_account::*, merchant_key_store::*, payment_link::*,
-    payment_method::*, process_tracker::*, refund::*, reverse_lookup::*, role::*,
+    generic_link::*, gsm::*, hyperswitch_ai_interaction::*, invoice::*, locker_mock_up::*,
+    mandate::*, merchant_account::*, merchant_connector_account::*, merchant_key_store::*,
+    payment_link::*, payment_method::*, process_tracker::*, refund::*, reverse_lookup::*, role::*,
     routing_algorithm::*, subscription::*, unified_translations::*, user::*,
     user_authentication_method::*, user_role::*,
 };
diff --git a/crates/router/src/types/storage/invoice.rs b/crates/router/src/types/storage/invoice.rs
new file mode 100644
index 00000000000..ac16c517efd
--- /dev/null
+++ b/crates/router/src/types/storage/invoice.rs
@@ -0,0 +1 @@
+pub use diesel_models::invoice::{Invoice, InvoiceNew, InvoiceUpdate};
diff --git a/migrations/2025-09-10-101514_add_invoice_table/down.sql b/migrations/2025-09-10-101514_add_invoice_table/down.sql
new file mode 100644
index 00000000000..fa537f820cc
--- /dev/null
+++ b/migrations/2025-09-10-101514_add_invoice_table/down.sql
@@ -0,0 +1,3 @@
+DROP INDEX IF EXISTS idx_subscription_id;
+
+DROP TABLE IF EXISTS invoice;
diff --git a/migrations/2025-09-10-101514_add_invoice_table/up.sql b/migrations/2025-09-10-101514_add_invoice_table/up.sql
new file mode 100644
index 00000000000..4dd8bb3bbf8
--- /dev/null
+++ b/migrations/2025-09-10-101514_add_invoice_table/up.sql
@@ -0,0 +1,19 @@
+CREATE TABLE invoice (
+    id VARCHAR(64) PRIMARY KEY,
+    subscription_id VARCHAR(128) NOT NULL,
+    merchant_id VARCHAR(64) NOT NULL,
+    profile_id VARCHAR(64) NOT NULL,
+    merchant_connector_id VARCHAR(128) NOT NULL,
+    payment_intent_id VARCHAR(64) UNIQUE,
+    payment_method_id VARCHAR(64),
+    customer_id VARCHAR(64) NOT NULL,
+    amount BIGINT NOT NULL,
+    currency VARCHAR(3) NOT NULL,
+    status VARCHAR(64) NOT NULL,
+    provider_name VARCHAR(128) NOT NULL,
+    metadata JSONB,
+    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
+    modified_at TIMESTAMP NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX idx_subscription_id ON invoice (subscription_id);