chore(notifications): boilerplate to start job registry

core/notifications/Cargo.toml

@@ -15,6 +15,7 @@ path = "src/bin/write_sdl.rs"
 fail-on-warnings = []
 
 [dependencies]
+job-executor = { path = "../../lib/job-executor-rs" }
 tracing = { path = "../../lib/tracing-rs" }
 es-entity = { path = "../../lib/es-entity-rs" }
 
@@ -34,6 +35,7 @@ thiserror = { workspace = true }
 chrono = { workspace = true }
 futures = { workspace = true }
 sqlx = { workspace = true }
+sqlxmq = { workspace = true }
 mongodb = { workspace = true }
 rand = { workspace = true }
 uuid = { workspace = true }

core/notifications/src/app/error.rs

@@ -1,11 +1,16 @@
 use thiserror::Error;
 
-use crate::{executor::error::*, user_notification_settings::error::*};
+use crate::{
+    executor::error::ExecutorError, job::error::JobError,
+    user_notification_settings::error::UserNotificationSettingsError,
+};
 
 #[derive(Error, Debug)]
 pub enum ApplicationError {
     #[error("{0}")]
     UserNotificationSettingsError(#[from] UserNotificationSettingsError),
     #[error("{0}")]
-    Novu(#[from] ExecutorError),
+    JobError(#[from] JobError),
+    #[error("{0}")]
+    ExecutorError(#[from] ExecutorError),
 }

core/notifications/src/app/mod.rs

@@ -2,9 +2,14 @@ mod config;
 pub mod error;
 
 use sqlx::{Pool, Postgres};
+use sqlxmq::JobRunnerHandle;
 use tracing::instrument;
 
-use crate::{executor::*, notification_event::*, primitives::*, user_notification_settings::*};
+use std::sync::Arc;
+
+use crate::{
+    executor::*, job, notification_event::*, primitives::*, user_notification_settings::*,
+};
 
 pub use config::*;
 use error::*;
@@ -15,17 +20,20 @@ pub struct NotificationsApp {
     settings: UserNotificationSettingsRepo,
     executor: Executor,
     _pool: Pool<Postgres>,
+    _runner: Arc<JobRunnerHandle>,
 }
 
 impl NotificationsApp {
     pub async fn init(pool: Pool<Postgres>, config: AppConfig) -> Result<Self, ApplicationError> {
         let settings = UserNotificationSettingsRepo::new(&pool);
         let executor = Executor::init(config.executor.clone(), settings.clone()).await?;
+        let runner = job::start_job_runner(&pool).await?;
         Ok(Self {
             _config: config,
             _pool: pool,
             executor,
             settings,
+            _runner: Arc::new(runner),
         })
     }
 

core/notifications/src/job/error.rs

@@ -0,0 +1,7 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum JobError {
+    #[error("JobError - Sqlx: {0}")]
+    Sqlx(#[from] sqlx::Error),
+}

core/notifications/src/job/mod.rs

@@ -0,0 +1,22 @@
+pub mod error;
+
+use sqlxmq::{job, CurrentJob, JobBuilder, JobRegistry, JobRunnerHandle};
+
+use error::JobError;
+
+pub async fn start_job_runner(pool: &sqlx::PgPool) -> Result<JobRunnerHandle, JobError> {
+    let mut registry = JobRegistry::new(&[
+        // sync_all_wallets,
+        // sync_wallet,
+        // process_all_payout_queues,
+        // schedule_process_payout_queue,
+        // process_payout_queue,
+        // batch_wallet_accounting,
+        // batch_signing,
+        // batch_broadcasting,
+        // respawn_all_outbox_handlers,
+        // populate_outbox,
+    ]);
+
+    Ok(registry.runner(pool).set_keep_alive(false).run().await?)
+}

core/notifications/src/lib.rs

@@ -5,6 +5,7 @@ rust_i18n::i18n!("locales", fallback = "en");
 
 mod app;
 mod data_import;
+mod job;
 mod messages;
 
 pub mod cli;