Fixed Issue #306

Author: BYTES-TECHES

indexer/README.md

@@ -11,6 +11,7 @@ A high-performance event indexing service for the Stellar Escrow smart contract
 - **Event Replay**: Replay historical events for catch-up or analysis
 - **Configurable**: Flexible configuration for different networks and contracts
 - **Redis Caching**: Read-through caching, invalidation, monitoring, and warming for hot API paths
+- **Background Jobs**: Redis-backed queue with priorities, scheduling, retries, and monitoring
 
 ## Architecture
 

indexer/src/event_monitor.rs

@@ -10,7 +10,7 @@ use crate::config::StellarConfig;
 use crate::database::Database;
 use crate::error::AppError;
 use crate::fraud_service::FraudDetectionService;
-use crate::job_queue::{JobQueue, types::{Job, JobType}};
+use crate::job_queue::{JobQueue, types::{Job, JobPriority, JobType}};
 use crate::models::{Event, WebSocketMessage};
 use crate::websocket::WebSocketManager;
 
@@ -308,22 +308,48 @@ impl EventMonitor {
             }
         }
 
-        // Enqueue background job
-        let job = Job {
-            job_type: JobType::Event,
-            event_id: event.id.to_string(),
-            trade_id: event
-                .data
-                .get("trade_id")
-                .and_then(|v| v.as_str())
-                .unwrap_or("unknown")
-                .to_string(),
-            payload: event.data.clone(),
+        let trade_id = event
+            .data
+            .get("trade_id")
+            .map(|value| {
+                value
+                    .as_str()
+                    .map(|s| s.to_string())
+                    .unwrap_or_else(|| value.to_string())
+            })
+            .unwrap_or_else(|| "unknown".to_string());
+
+        let event_priority = match event.event_type.as_str() {
+            "dispute_raised" | "dispute_resolved" => JobPriority::Critical,
+            "trade_created" | "trade_funded" | "trade_confirmed" => JobPriority::High,
+            _ => JobPriority::Normal,
         };
+        let event_job = Job::new(
+            JobType::Event,
+            event.id.to_string(),
+            trade_id.clone(),
+            event.data.clone(),
+            event_priority,
+        );
+
+        let notification_job = Job::new(
+            JobType::Notification,
+            event.id.to_string(),
+            trade_id,
+            serde_json::json!({
+                "event_type": event.event_type,
+                "timestamp": event.timestamp,
+                "data": event.data,
+            }),
+            JobPriority::High,
+        );
 
         let mut queue = self.job_queue.lock().await;
-        if let Err(e) = queue.enqueue(job).await {
-            error!("Failed to enqueue job for event {}: {}", event.id, e);
+        if let Err(e) = queue.enqueue(event_job).await {
+            error!("Failed to enqueue event job for event {}: {}", event.id, e);
+        }
+        if let Err(e) = queue.enqueue(notification_job).await {
+            error!("Failed to enqueue notification job for event {}: {}", event.id, e);
         }
     }
 

indexer/src/handlers.rs

@@ -16,6 +16,9 @@ use crate::database::Database;
 use crate::error::AppError;
 use crate::fraud_service::FraudDetectionService;
 use crate::health::HealthState;
+use crate::job_queue::types::{Job, JobPriority, JobType};
+use crate::job_queue::worker::JobWorker;
+use crate::job_queue::JobQueue;
 use crate::webhook_service::WebhookService;
 use crate::monitoring_service::{MonitoringService, dashboard};
 use crate::models::{
@@ -61,6 +64,9 @@ pub async fn api_index() -> Json<serde_json::Value> {
             "cache_stats":     "GET  /cache/stats",
             "cache_invalidate":"POST /cache/invalidate",
             "cache_warm":      "POST /cache/warm",
+            "jobs_stats":      "GET  /jobs/stats",
+            "jobs_enqueue":    "POST /jobs/enqueue",
+            "jobs_schedule":   "POST /jobs/schedule",
             "fraud_review":    "POST /fraud/review",
             "notif_prefs_get": "GET  /notifications/preferences/:address",
             "notif_prefs_put": "PUT  /notifications/preferences/:address",
@@ -425,6 +431,8 @@ pub async fn update_fraud_review(
 pub struct AppState {
     pub database: Arc<Database>,
     pub stellar_contract_id: String,
+    pub job_queue: Arc<tokio::sync::Mutex<JobQueue>>,
+    pub job_worker: Arc<JobWorker>,
     pub ws_manager: Arc<WebSocketManager>,
     pub health: HealthState,
     pub fraud_service: Arc<FraudDetectionService>,
@@ -645,6 +653,86 @@ pub struct IntegrationLogQuery {
     pub limit: Option<i64>,
 }
 
+#[derive(Deserialize)]
+pub struct EnqueueJobRequest {
+    pub job_type: String,
+    pub event_id: Option<String>,
+    pub trade_id: Option<String>,
+    pub priority: Option<String>,
+    pub payload: Option<serde_json::Value>,
+    pub run_at: Option<i64>,
+    pub max_attempts: Option<u32>,
+}
+
+/// GET /jobs/stats — job queue depths and worker status.
+pub async fn get_job_stats(
+    State(state): State<AppState>,
+) -> Result<Json<serde_json::Value>, AppError> {
+    let snapshot = state.job_worker.snapshot().await?;
+    Ok(Json(serde_json::to_value(&snapshot).unwrap_or_default()))
+}
+
+/// POST /jobs/enqueue — enqueue an immediate background job.
+pub async fn enqueue_job(
+    State(state): State<AppState>,
+    Json(body): Json<EnqueueJobRequest>,
+) -> Result<Json<serde_json::Value>, AppError> {
+    let job_type = parse_job_type(&body.job_type)?;
+    let priority = parse_job_priority(body.priority.as_deref())?;
+    let mut job = Job::new(
+        job_type,
+        body.event_id.unwrap_or_else(|| "manual".to_string()),
+        body.trade_id.unwrap_or_else(|| "manual".to_string()),
+        body.payload.unwrap_or_default(),
+        priority,
+    );
+    if let Some(max_attempts) = body.max_attempts {
+        job = job.with_max_attempts(max_attempts.max(1));
+    }
+
+    let mut queue = state.job_queue.lock().await;
+    queue.enqueue(job.clone()).await?;
+
+    Ok(Json(json!({
+        "status": "queued",
+        "job_id": job.id,
+        "priority": job.priority.as_str(),
+    })))
+}
+
+/// POST /jobs/schedule — enqueue a delayed background job.
+pub async fn schedule_job(
+    State(state): State<AppState>,
+    Json(body): Json<EnqueueJobRequest>,
+) -> Result<Json<serde_json::Value>, AppError> {
+    let run_at = body
+        .run_at
+        .ok_or_else(|| AppError::BadRequest("run_at is required".to_string()))?;
+    let job_type = parse_job_type(&body.job_type)?;
+    let priority = parse_job_priority(body.priority.as_deref())?;
+    let mut job = Job::new(
+        job_type,
+        body.event_id.unwrap_or_else(|| "scheduled".to_string()),
+        body.trade_id.unwrap_or_else(|| "scheduled".to_string()),
+        body.payload.unwrap_or_default(),
+        priority,
+    )
+    .scheduled_at(run_at);
+    if let Some(max_attempts) = body.max_attempts {
+        job = job.with_max_attempts(max_attempts.max(1));
+    }
+
+    let mut queue = state.job_queue.lock().await;
+    queue.enqueue_at(job.clone(), run_at).await?;
+
+    Ok(Json(json!({
+        "status": "scheduled",
+        "job_id": job.id,
+        "run_at": run_at,
+        "priority": job.priority.as_str(),
+    })))
+}
+
 // =============================================================================
 // Performance Monitoring Handlers
 // =============================================================================
@@ -993,3 +1081,23 @@ pub async fn get_prometheus_metrics(
         .body(body)
         .unwrap()
 }
+
+fn parse_job_type(value: &str) -> Result<JobType, AppError> {
+    match value {
+        "event" => Ok(JobType::Event),
+        "notification" => Ok(JobType::Notification),
+        "cache_warm" => Ok(JobType::CacheWarm),
+        "compliance" => Ok(JobType::Compliance),
+        _ => Err(AppError::BadRequest(format!("unsupported job_type '{}'", value))),
+    }
+}
+
+fn parse_job_priority(value: Option<&str>) -> Result<JobPriority, AppError> {
+    match value.unwrap_or("normal") {
+        "critical" => Ok(JobPriority::Critical),
+        "high" => Ok(JobPriority::High),
+        "normal" => Ok(JobPriority::Normal),
+        "low" => Ok(JobPriority::Low),
+        other => Err(AppError::BadRequest(format!("unsupported priority '{}'", other))),
+    }
+}