feat(scheduler): add task segregation based on application source (#10505)

Author: kashif-m

config/config.example.toml

@@ -2,6 +2,11 @@
 # all the available configuration options, and is intended to be used
 # solely as a reference. Please copy this file to create a config.
 
+# This is used to identify the running application instance
+# Check common_enums::ApplicationSource for available options
+# For more details - https://github.com/juspay/hyperswitch/issues/10489
+application_source = "main"
+
 # Server configuration
 [server]
 port = 8080
@@ -416,6 +421,7 @@ cards = [
 # It defines the streams/queues name and configuration as well as event selection variables
 [scheduler]
 stream = "SCHEDULER_STREAM"
+cug_stream = "CUG_SCHEDULER_STREAM"
 graceful_shutdown_interval = 60000 # Specifies how much time to wait while re-attempting shutdown for a service (in milliseconds)
 loop_interval = 5000               # Specifies how much time to wait before starting the defined behaviour of producer or consumer (in milliseconds)
 

config/development.toml

@@ -1,3 +1,5 @@
+application_source = "main"
+
 [log.file]
 enabled = false
 
@@ -382,6 +384,7 @@ zsl.base_url = "https://api.sitoffalb.net/"
 
 [scheduler]
 stream = "SCHEDULER_STREAM"
+cug_stream = "CUG_SCHEDULER_STREAM"
 
 [scheduler.consumer]
 disabled = false
@@ -1444,4 +1447,4 @@ token = "123456" # Superposition token
 org_id = "localorg" # Organization ID in Superposition
 workspace_id = "dev" # Workspace ID in Superposition
 polling_interval = 15 # Polling interval in seconds for configuration updates
-# request_timeout =   # Request timeout in seconds for Superposition API calls (optional, default: none)
+# request_timeout =   # Request timeout in seconds for Superposition API calls (optional, default: none)

crates/common_enums/src/enums.rs

@@ -9966,6 +9966,28 @@ pub enum ProcessTrackerRunner {
     InvoiceSyncflow,
 }
 
+#[derive(
+    Clone,
+    Copy,
+    Debug,
+    Default,
+    Eq,
+    PartialEq,
+    serde::Deserialize,
+    serde::Serialize,
+    strum::Display,
+    strum::EnumString,
+    ToSchema,
+)]
+#[router_derive::diesel_enum(storage_type = "text")]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum ApplicationSource {
+    #[default]
+    Main,
+    Cug,
+}
+
 #[derive(Debug)]
 pub enum CryptoPadding {
     PKCS7,

crates/diesel_models/src/process_tracker.rs

@@ -5,7 +5,10 @@ use error_stack::ResultExt;
 use serde::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
 
-use crate::{enums as storage_enums, errors, schema::process_tracker, StorageResult};
+use crate::{
+    enums as storage_enums, enums::ApplicationSource, errors, schema::process_tracker,
+    StorageResult,
+};
 
 #[derive(
     Clone,
@@ -40,6 +43,7 @@ pub struct ProcessTracker {
     #[serde(with = "common_utils::custom_serde::iso8601")]
     pub updated_at: PrimitiveDateTime,
     pub version: ApiVersion,
+    pub application_source: Option<ApplicationSource>,
 }
 
 impl ProcessTracker {
@@ -66,6 +70,7 @@ pub struct ProcessTrackerNew {
     pub created_at: PrimitiveDateTime,
     pub updated_at: PrimitiveDateTime,
     pub version: ApiVersion,
+    pub application_source: Option<ApplicationSource>,
 }
 
 impl ProcessTrackerNew {
@@ -79,6 +84,7 @@ impl ProcessTrackerNew {
         retry_count: Option<i32>,
         schedule_time: PrimitiveDateTime,
         api_version: ApiVersion,
+        application_source: ApplicationSource,
     ) -> StorageResult<Self>
     where
         T: Serialize + std::fmt::Debug,
@@ -102,6 +108,7 @@ impl ProcessTrackerNew {
             created_at: current_time,
             updated_at: current_time,
             version: api_version,
+            application_source: Some(application_source),
         })
     }
 }

crates/diesel_models/src/schema.rs

@@ -1426,6 +1426,8 @@ diesel::table! {
         created_at -> Timestamp,
         updated_at -> Timestamp,
         version -> ApiVersion,
+        #[max_length = 64]
+        application_source -> Nullable<Varchar>,
     }
 }
 

crates/diesel_models/src/schema_v2.rs

@@ -1371,6 +1371,8 @@ diesel::table! {
         created_at -> Timestamp,
         updated_at -> Timestamp,
         version -> ApiVersion,
+        #[max_length = 64]
+        application_source -> Nullable<Varchar>,
     }
 }
 

crates/router/src/configs/secrets_transformers.rs

@@ -489,6 +489,7 @@ pub(crate) async fn fetch_raw_secrets(
 
     Settings {
         server: conf.server,
+        application_source: conf.application_source,
         chat,
         master_database,
         redis: conf.redis,

crates/router/src/configs/settings.rs

@@ -7,6 +7,7 @@ use std::{
 #[cfg(feature = "olap")]
 use analytics::{opensearch::OpenSearchConfig, ReportConfig};
 use api_models::enums;
+use common_enums;
 use common_utils::{ext_traits::ConfigExt, id_type, types::user::EmailThemeConfig};
 use config::{Environment, File};
 use error_stack::ResultExt;
@@ -73,6 +74,7 @@ pub struct CmdLineConf {
 #[serde(default)]
 pub struct Settings<S: SecretState> {
     pub server: Server,
+    pub application_source: common_enums::ApplicationSource,
     pub proxy: Proxy,
     pub env: Env,
     pub chat: SecretStateContainer<ChatSettings, S>,

crates/router/src/core/api_keys.rs

@@ -171,10 +171,15 @@ pub async fn create_api_key(
         if api_key.expires_at.is_some() {
             let expiry_reminder_days = state.conf.api_keys.get_inner().expiry_reminder_days.clone();
 
-            add_api_key_expiry_task(store, &api_key, expiry_reminder_days)
-                .await
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Failed to insert API key expiry reminder to process tracker")?;
+            add_api_key_expiry_task(
+                store,
+                &api_key,
+                expiry_reminder_days,
+                state.conf.application_source,
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to insert API key expiry reminder to process tracker")?;
         }
     }
 
@@ -193,6 +198,7 @@ pub async fn add_api_key_expiry_task(
     store: &dyn crate::db::StorageInterface,
     api_key: &ApiKey,
     expiry_reminder_days: Vec<u8>,
+    application_source: common_enums::ApplicationSource,
 ) -> Result<(), errors::ProcessTrackerError> {
     let current_time = date_time::now();
 
@@ -231,6 +237,7 @@ pub async fn add_api_key_expiry_task(
         None,
         schedule_time,
         common_types::consts::API_VERSION,
+        application_source,
     )
     .change_context(errors::ApiErrorResponse::InternalServerError)
     .attach_printable("Failed to construct API key expiry process tracker task")?;
@@ -346,12 +353,15 @@ pub async fn update_api_key(
         else if api_key.expires_at.is_some() {
             // Process doesn't exist in process_tracker table, so create new entry with
             // schedule_time based on new expiry set.
-            add_api_key_expiry_task(store, &api_key, expiry_reminder_days)
-                .await
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable(
-                    "Failed to insert API key expiry reminder task to process tracker",
-                )?;
+            add_api_key_expiry_task(
+                store,
+                &api_key,
+                expiry_reminder_days,
+                state.conf.application_source,
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to insert API key expiry reminder task to process tracker")?;
         }
     }
 

crates/router/src/core/disputes.rs

@@ -12,6 +12,8 @@ use router_env::{
 use strum::IntoEnumIterator;
 pub mod transformers;
 
+use common_enums;
+
 use super::{
     errors::{self, ConnectorErrorExt, RouterResponse, StorageErrorExt},
     metrics,
@@ -843,6 +845,7 @@ pub async fn fetch_disputes_from_connector(
                 dispute,
                 merchant_id.clone(),
                 schedule_time,
+                state.conf.application_source,
             )
             .await;
 
@@ -917,6 +920,7 @@ pub async fn add_process_dispute_task_to_pt(
     dispute_payload: &DisputeSyncResponse,
     merchant_id: common_utils::id_type::MerchantId,
     schedule_time: Option<time::PrimitiveDateTime>,
+    application_source: common_enums::ApplicationSource,
 ) -> common_utils::errors::CustomResult<(), errors::StorageError> {
     match schedule_time {
         Some(time) => {
@@ -947,6 +951,7 @@ pub async fn add_process_dispute_task_to_pt(
                 None,
                 time,
                 common_types::consts::API_VERSION,
+                application_source,
             )
             .map_err(errors::StorageError::from)?;
             db.insert_process(process_tracker_entry).await?;
@@ -964,6 +969,7 @@ pub async fn add_dispute_list_task_to_pt(
     merchant_connector_id: common_utils::id_type::MerchantConnectorAccountId,
     profile_id: common_utils::id_type::ProfileId,
     fetch_request: FetchDisputesRequestData,
+    application_source: common_enums::ApplicationSource,
 ) -> common_utils::errors::CustomResult<(), errors::StorageError> {
     TASKS_ADDED_COUNT.add(1, router_env::metric_attributes!(("flow", "dispute_list")));
     let tracking_data = disputes::DisputeListPTData {
@@ -992,6 +998,7 @@ pub async fn add_dispute_list_task_to_pt(
         None,
         fetch_request.created_from,
         common_types::consts::API_VERSION,
+        application_source,
     )
     .map_err(errors::StorageError::from)?;
     db.insert_process(process_tracker_entry).await?;
@@ -1028,6 +1035,7 @@ pub async fn schedule_dispute_sync_task(
         let merchant_id = mca.merchant_id.clone();
         let merchant_connector_id = mca.merchant_connector_id.clone();
         let business_profile_id = business_profile.get_id().clone();
+        let application_source = state.conf.application_source;
 
         tokio::spawn(
             async move {
@@ -1041,6 +1049,7 @@ pub async fn schedule_dispute_sync_task(
                         created_from,
                         created_till,
                     },
+                    application_source,
                 )
                 .await
                 .map_err(|error| {