feat: job crate dummy job

Cargo.toml

@@ -8,6 +8,7 @@ edition = "2021"
 fail-on-warnings = []
 
 [dependencies]
+job_crate = { package = "job", version = "0.1.12" }
 es-entity = "0.9.0"
 sqlx-ledger = { version = "0.11.5", features = ["otel"] }
 

bats/dummy_job.bats

@@ -0,0 +1,51 @@
+#!/usr/bin/env bats
+
+load "helpers"
+
+setup_file() {
+  restart_bitcoin_stack
+  reset_pg
+  bitcoind_init
+  start_daemon
+  bria_init
+}
+
+teardown_file() {
+  stop_daemon
+}
+
+@test "dummy_job: Verify job exists in database" {
+  # Wait a few seconds for daemon to initialize
+  sleep 10
+
+  # Query the jobs table to check if dummy job exists
+  job_count=$(${DOCKER_ENGINE} exec "${COMPOSE_PROJECT_NAME}-postgres-1" psql $PG_CON -t -c "SELECT COUNT(*) FROM jobs WHERE job_type = 'dummy';")
+
+  # Trim whitespace
+  job_count=$(echo $job_count | xargs)
+
+  if [[ $job_count -lt 1 ]]; then
+    echo "Dummy job not found in database"
+    echo "Available jobs:"
+    ${DOCKER_ENGINE} exec "${COMPOSE_PROJECT_NAME}-postgres-1" psql $PG_CON -c "SELECT job_type, created_at FROM jobs;"
+    exit 1
+  fi
+
+  echo "Found $job_count dummy job(s) in database"
+}
+
+@test "dummy_job: Verify job has execution scheduled" {
+  # Check if there are any scheduled executions for the dummy job
+  exec_count=$(${DOCKER_ENGINE} exec "${COMPOSE_PROJECT_NAME}-postgres-1" psql $PG_CON -t -c "SELECT COUNT(*) FROM job_executions je JOIN jobs j ON je.id = j.id WHERE j.job_type = 'dummy';")
+
+  # Trim whitespace  
+  exec_count=$(echo $exec_count | xargs)
+
+  if [[ $exec_count -lt 1 ]]; then
+    echo "No executions scheduled for dummy job"
+    exit 1
+  fi
+
+  echo "Found $exec_count scheduled execution(s) for dummy job"
+}
+

migrations/20250904065521_job_setup.sql

@@ -0,0 +1,56 @@
+CREATE TABLE jobs (
+  id UUID PRIMARY KEY,
+  unique_per_type BOOLEAN NOT NULL,
+  job_type VARCHAR NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+CREATE UNIQUE INDEX idx_unique_job_type ON jobs (job_type) WHERE unique_per_type = TRUE;
+
+CREATE TABLE job_events (
+  id UUID NOT NULL REFERENCES jobs(id),
+  sequence INT NOT NULL,
+  event_type VARCHAR NOT NULL,
+  event JSONB NOT NULL,
+  context JSONB DEFAULT NULL,
+  recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  UNIQUE(id, sequence)
+);
+
+CREATE TYPE JobExecutionState AS ENUM ('pending', 'running');
+
+CREATE TABLE job_executions (
+  id UUID REFERENCES jobs(id) NOT NULL UNIQUE,
+  job_type VARCHAR NOT NULL,
+  attempt_index INT NOT NULL DEFAULT 1,
+  state JobExecutionState NOT NULL DEFAULT 'pending',
+  execution_state_json JSONB,
+  execute_at TIMESTAMPTZ,
+  alive_at TIMESTAMPTZ NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL
+);
+
+CREATE OR REPLACE FUNCTION notify_job_execution_insert() RETURNS TRIGGER AS $$
+BEGIN
+  PERFORM pg_notify('job_execution', '');
+  RETURN NULL;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION notify_job_execution_update() RETURNS TRIGGER AS $$
+BEGIN
+  IF NEW.execute_at IS DISTINCT FROM OLD.execute_at THEN
+    PERFORM pg_notify('job_execution', '');
+  END IF;
+  RETURN NULL;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE TRIGGER job_executions_notify_insert_trigger
+AFTER INSERT ON job_executions
+FOR EACH STATEMENT
+EXECUTE FUNCTION notify_job_execution_insert();
+
+CREATE TRIGGER job_executions_notify_update_trigger
+AFTER UPDATE ON job_executions
+FOR EACH STATEMENT
+EXECUTE FUNCTION notify_job_execution_update();

src/api/server/convert.rs

@@ -718,6 +718,9 @@ impl From<ApplicationError> for tonic::Status {
             ApplicationError::CouldNotParseAddress(_) => {
                 tonic::Status::invalid_argument(err.to_string())
             }
+            ApplicationError::JobCrateJobError(_) => {
+                tonic::Status::internal(err.to_string())
+            }
             _ => tonic::Status::internal(err.to_string()),
         }
     }

src/app/error.rs

@@ -1,5 +1,6 @@
 use chacha20poly1305;
 use thiserror::Error;
+use job_crate::error::JobError as JobCrateJobError;
 
 use crate::{
     address::error::AddressError,
@@ -81,6 +82,8 @@ pub enum ApplicationError {
     CouldNotDecryptKey(chacha20poly1305::Error),
     #[error("AddressError - Could not parse the address: {0}")]
     CouldNotParseAddress(#[from] bitcoin::AddressError),
+    #[error("JobCrateError - Sqlx: {0}")]
+    JobCrateJobError(#[from] JobCrateJobError),
 }
 
 impl From<chacha20poly1305::Error> for ApplicationError {

src/app/mod.rs

@@ -8,6 +8,7 @@ use std::collections::HashMap;
 
 pub use config::*;
 use error::*;
+use job_crate::{Jobs, JobSvcConfig};
 
 use crate::{
     account::balance::AccountBalanceSummary,
@@ -86,6 +87,13 @@ impl App {
             fees_client.clone(),
         )
         .await?;
+
+        let mut jobs = Jobs::init(JobSvcConfig::builder().pool(pool.clone()).build().expect("Couldn't build JobSvcConfig")).await?;
+
+        job::spawn_dummy(&jobs).await?;
+
+        jobs.start_poll().await?;
+
         Self::spawn_sync_all_wallets(pool.clone(), config.jobs.sync_all_wallets_delay).await?;
         Self::spawn_process_all_payout_queues(
             pool.clone(),
@@ -97,6 +105,7 @@ impl App {
             config.jobs.respawn_all_outbox_handlers_delay,
         )
         .await?;
+
         let app = Self {
             outbox,
             profiles: Profiles::new(&pool),

src/job/dummy.rs

@@ -0,0 +1,34 @@
+use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
+use job_crate::*;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DummyJobConfig;
+
+impl JobConfig for DummyJobConfig {
+    type Initializer = DummyJobInit;
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DummyJobInit;
+impl JobInitializer for DummyJobInit {
+    fn job_type() -> JobType {
+        JobType::new("dummy")
+    }
+
+    fn init(&self, _job: &Job) -> Result<Box<dyn JobRunner>, Box<dyn std::error::Error>> {
+        Ok(Box::new(DummyJobRunner))
+    }
+}
+
+struct DummyJobRunner;
+
+#[async_trait]
+impl JobRunner for DummyJobRunner {
+    async fn run(&self, _current_job: CurrentJob) -> Result<JobCompletion, Box<dyn std::error::Error>> {
+        tracing::info!("Dummy job running!");
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        tracing::info!("Dummy job completed successfully");
+        Ok(JobCompletion::Complete)
+    }
+}

src/job/error.rs

@@ -58,6 +58,8 @@ pub enum JobError {
     PsbtMissingInSigningSessions,
     #[error("JobError - psbt::Error: {0}")]
     PsbtError(#[from] psbt::Error),
+    #[error("JobCrateError: {0}")]
+    JobCrateError(#[from] job_crate::error::JobError),
 }
 
 impl JobExecutionError for JobError {}

src/job/mod.rs

@@ -6,6 +6,7 @@ mod executor;
 mod populate_outbox;
 mod sync_wallet;
 
+pub mod dummy;
 pub mod error;
 pub mod process_payout_queue;
 
@@ -16,13 +17,12 @@ use tracing::instrument;
 use uuid::{uuid, Uuid};
 
 use crate::{
-    account::*, address::Addresses, app::BlockchainConfig, batch::*, fees::FeesClient,
-    ledger::Ledger, outbox::*, payout::*, payout_queue::*, primitives::*, signing_session::*,
-    utxo::Utxos, wallet::*, xpub::*,
+    account::*, address::Addresses, app::BlockchainConfig, batch::*, fees::FeesClient, ledger::Ledger, outbox::*, payout::*, payout_queue::*, primitives::*, signing_session::*, utxo::Utxos, wallet::*, xpub::*
 };
 use batch_broadcasting::BatchBroadcastingData;
 use batch_signing::BatchSigningData;
 use batch_wallet_accounting::BatchWalletAccountingData;
+use dummy::{DummyJobConfig, DummyJobInit};
 use error::JobError;
 pub use executor::JobExecutionError;
 use executor::JobExecutor;
@@ -611,6 +611,25 @@ pub async fn spawn_respawn_all_outbox_handlers(
     }
 }
 
+#[instrument(name = "job.spawn_dummy", skip_all, fields(error, error.level, error.message), err)]
+pub async fn spawn_dummy(
+    jobs: &job_crate::Jobs
+) -> Result<(), JobError> {
+    tracing::info!("Attempting to spawn dummy job");
+    let job_config = DummyJobConfig;
+    match jobs.add_initializer_and_spawn_unique(DummyJobInit, job_config).await {
+        Ok(_) => {
+            tracing::info!("Successfully spawned dummy job");
+            Ok(())
+        }
+        Err(e) => {
+            tracing::error!("Failed to spawn dummy job: {}", e);
+            crate::tracing::insert_error_fields(tracing::Level::ERROR, &e);
+            Err(e.into())
+        }
+    }
+}
+
 fn schedule_payout_queue_channel_arg(payout_queue_id: PayoutQueueId) -> String {
     format!("payout_queue_id:{payout_queue_id}")
 }