Issue:
Bug: Subscription table and core change





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


### Additional Changes

- [ ] This PR modifies the API contract
- [x] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

## Description

Introduce **Subscriptions** as a first-class entity in the router.  
This PR adds a `subscription` table, Diesel models, storage interfaces, and query helpers to **create**, **fetch**, and **update** subscription entries keyed by `(merchant_id, customer_id, subscription_id)`.

### What’s included

**Database (PostgreSQL)**
- New table: `subscription`
- Columns:
  - `id VARCHAR(128) PRIMARY KEY`
  - `subscription_id VARCHAR(128) NULL`
  - `billing_processor VARCHAR(128) NULL`
  - `payment_method_id VARCHAR(128) NULL`
  - `mca_id VARCHAR(128) NULL`
  - `client_secret VARCHAR(128) NULL`
  - `merchant_id VARCHAR(64) NOT NULL`
  - `customer_id VARCHAR(64) NOT NULL`
  - `metadata JSONB NULL`
  - `created_at TIMESTAMP NOT NULL`
  - `modified_at TIMESTAMP NOT NULL`
- Indexes:
  - `CREATE UNIQUE INDEX merchant_customer_subscription_unique_index ON subscription (merchant_id, customer_id, subscription_id);`

**Migrations**
- `up.sql` creates the table + unique index
- `down.sql` drops the table

**Diesel models & schema**
- `crates/diesel_models/src/schema.rs`: `subscription` table mapping
- `crates/diesel_models/src/subscription.rs`:
  - `SubscriptionNew` (insert)
  - `Subscription` (read)
  - `SubscriptionUpdate` (update: `subscription_id`, `payment_method_id`, `modified_at`)
- Query functions: `crates/diesel_models/src/query/subscription.rs`
  - `SubscriptionNew::insert`
  - `Subscription::find_by_merchant_id_customer_id_subscription_id`
  - `Subscription::update_subscription_entry`

**Router storage interface**
- Trait: `SubscriptionInterface` (`crates/router/src/db/subscription.rs`)
  - `insert_subscription_entry`
  - `find_by_merchant_id_customer_id_subscription_id`
  - `update_subscription_entry`
- Implemented for `Store`, `KafkaStore` (delegates), and `MockDb` (stubbed)

**Wiring**
- Re-exports in `crates/router/src/types/storage.rs` and module plumbing in `db.rs`, `diesel_models/src/lib.rs`, `query.rs`.

---

## Motivation and Context

Subscriptions are often provisioned/managed outside Payments, but Payments needs a lightweight mapping to:
- Track **billing processor** and connector meta (`mca_id`, `client_secret`)
- Attach or rotate **payment_method_id** per subscription
- Support **idempotent** reads keyed by merchant & customer context

This schema enables upstream orchestration to store/retrieve subscription context without coupling to payment flows.

---

## API / Contract / Config Impact

- **DB schema change**: ✅ (new table + unique index)
- **API contract**: ❌ (no external API changes in this PR)
- **Config/env**: ❌

---

### Testing
can't be performed, as in this PR we only have created the tables and not used it anywhere in the upcoming micro-prs, this will be used. however I have used it myself by making code changes, here are the screenshots.

<img width="3456" height="1182" alt="Screenshot 2025-08-26 at 20 31 13" src="https://github.com/user-attachments/assets/320f6b05-74a9-44d2-ad9a-5514431e5ca2" />
<img width="3456" height="998" alt="Screenshot 2025-08-26 at 20 31 38" src="https://github.com/user-attachments/assets/123a55a4-74bd-4aa4-812b-821a0e73caa4" />



## 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:

Patch:
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/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..e25f1617873
--- /dev/null
+++ b/crates/diesel_models/src/query/subscription.rs
@@ -0,0 +1,59 @@
+use diesel::{associations::HasTable, BoolExpressionMethods, ExpressionMethods};
+use error_stack::report;
+
+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 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")
+        })
+    }
+}
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index 08675d904de..0d4564c83c3 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -1465,6 +1465,36 @@ 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 = 128]
+        connector_subscription_id -> 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 +1680,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..b734e2c4bb8 100644
--- a/crates/diesel_models/src/schema_v2.rs
+++ b/crates/diesel_models/src/schema_v2.rs
@@ -1400,6 +1400,36 @@ 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 = 128]
+        connector_subscription_id -> 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 +1635,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..0cabb6a7b44
--- /dev/null
+++ b/crates/diesel_models/src/subscription.rs
@@ -0,0 +1,93 @@
+use common_utils::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>,
+    connector_subscription_id: 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 connector_subscription_id: 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>,
+        connector_subscription_id: 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,
+            connector_subscription_id,
+            merchant_id,
+            customer_id,
+            metadata,
+            created_at: now,
+            modified_at: now,
+        }
+    }
+}
+
+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/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/subscription.rs b/crates/router/src/db/subscription.rs
new file mode 100644
index 00000000000..5242c44299b
--- /dev/null
+++ b/crates/router/src/db/subscription.rs
@@ -0,0 +1,140 @@
+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_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)))
+    }
+}
+
+#[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_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
+    }
+}
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/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..4e5419aefd4
--- /dev/null
+++ b/migrations/2025-08-21-110802_add_subcription_table/up.sql
@@ -0,0 +1,17 @@
+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),
+  connector_subscription_id 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);