runner_q 0.6.4

//! Runner-Q: A pluggable activity queue and worker system for Rust
//!
//! This crate provides a robust, scalable activity queue system with pluggable storage backends:
//!
//! ## Features
//!
//! - **Priority-based activity processing** with Critical, High, Normal, and Low priority levels
//! - **Activity scheduling** with precise timestamp-based scheduling for future execution
//! - **Intelligent retry mechanism** with exponential backoff for failed activities
//! - **Dead letter queue** handling for activities that exceed retry limits
//! - **Concurrent activity processing** with configurable worker pools
//! - **Graceful shutdown** handling with proper cleanup
//! - **Activity orchestration** enabling activities to execute other activities
//! - **Comprehensive error handling** with retryable and non-retryable error types
//! - **Activity metadata** support for context and tracking
//! - **Pluggable storage backends** - PostgreSQL (built-in), or bring your own (e.g. `runner_q_redis` for Redis)
//! - **Worker-level activity type filtering** - Isolate workloads by restricting engines to specific types
//! - **Queue statistics** and monitoring capabilities
//! - **Web-based observability console** for real-time monitoring
//!
//! ## Storage Backends
//!
//! Runner-Q supports multiple storage backends through the [`Storage`] trait:
//!
//! | Backend | Feature Flag | Status | Description |
//! |---------|--------------|--------|-------------|
//! | PostgreSQL | `postgres` (default) | Stable | Permanent persistence, `FOR UPDATE SKIP LOCKED` |
//! | Redis | N/A (separate crate) | Optional | Use the `runner_q_redis` crate for Redis/Valkey |
//!
//! ### Using a Backend
//!
//! ```rust,ignore
//! use runner_q::{WorkerEngine, storage::PostgresBackend};
//! use std::sync::Arc;
//!
//! // With PostgreSQL (built-in)
//! #[cfg(feature = "postgres")]
//! {
//!     use runner_q::storage::PostgresBackend;
//!     let backend = Arc::new(
//!         PostgresBackend::new("postgres://localhost/mydb", "my_queue").await?
//!     );
//!     let engine = WorkerEngine::builder()
//!         .backend(backend)
//!         .build()
//!         .await?;
//! }
//! ```
//!
//! # Example
//!
//! ```rust,no_run
//! use runner_q::{WorkerEngine, ActivityPriority, ActivityHandler, ActivityContext, ActivityHandlerResult, ActivityError, storage::PostgresBackend};
//! use std::sync::Arc;
//! use async_trait::async_trait;
//! use serde_json::json;
//! use serde::{Serialize, Deserialize};
//! use std::time::Duration;
//!
//! // Define activity types
//! #[derive(Debug, Clone)]
//! enum MyActivityType {
//!     SendEmail,
//!     ProcessPayment,
//! }
//!
//! impl std::fmt::Display for MyActivityType {
//!     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
//!         match self {
//!             MyActivityType::SendEmail => write!(f, "send_email"),
//!             MyActivityType::ProcessPayment => write!(f, "process_payment"),
//!         }
//!     }
//! }
//!
//! // Implement activity handler
//! pub struct SendEmailActivity;
//!
//! #[async_trait]
//! impl ActivityHandler for SendEmailActivity {
//!     async fn handle(&self, payload: serde_json::Value, context: ActivityContext) -> ActivityHandlerResult {
//!         // Parse the email data - use ? operator for clean error handling
//!         let email_data: serde_json::Map<String, serde_json::Value> = payload
//!             .as_object()
//!             .ok_or_else(|| ActivityError::NonRetry("Invalid payload format".to_string()))?
//!             .clone();
//!
//!         let to = email_data.get("to")
//!             .and_then(|v| v.as_str())
//!             .ok_or_else(|| ActivityError::NonRetry("Missing 'to' field".to_string()))?;
//!
//!         // Simulate sending email
//!         println!("Sending email to: {}", to);
//!
//!         // Return success with result data
//!         Ok(Some(serde_json::json!({
//!             "message": format!("Email sent to {}", to),
//!             "status": "delivered"
//!         })))
//!     }
//!
//!     fn activity_type(&self) -> String {
//!         MyActivityType::SendEmail.to_string()
//!     }
//! }
//!
//! #[derive(Debug, Serialize, Deserialize)]
//! pub struct EmailResult {
//!     message: String,
//!     status: String,
//! }
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     // Builder pattern: provide a backend (e.g. PostgresBackend::new(...))
//!     let backend = PostgresBackend::new("postgres://localhost/mydb", "my_app").await?;
//!     let engine = WorkerEngine::builder()
//!         .backend(Arc::new(backend))
//!         .queue_name("my_app")
//!         .max_workers(8)
//!         .schedule_poll_interval(Duration::from_secs(30))
//!         .build()
//!         .await?;
//!
//!     // Register activity handler
//!     let send_email_activity = SendEmailActivity;
//!     engine.register_activity(MyActivityType::SendEmail.to_string(), Arc::new(send_email_activity));
//!
//!     // Get activity executor for fluent activity execution
//!     let activity_executor = engine.get_activity_executor();
//!
//!     // Improved API: Fluent activity execution
//!     let future = activity_executor
//!         .activity("send_email")
//!         .payload(json!({"to": "user@example.com", "subject": "Welcome!"}))
//!         .max_retries(5)
//!         .timeout(Duration::from_secs(600))
//!         .execute()
//!         .await?;
//!
//!     // Schedule an activity for future execution (10 seconds from now)
//!     let scheduled_future = activity_executor
//!         .activity("send_email")
//!         .payload(json!({"to": "user@example.com", "subject": "Reminder"}))
//!         .max_retries(3)
//!         .timeout(Duration::from_secs(300))
//!         .delay(Duration::from_secs(10))
//!         .execute()
//!         .await?;
//!
//!     // Execute an activity with default options
//!     let future2 = activity_executor
//!         .activity("send_email")
//!         .payload(json!({"to": "admin@example.com"}))
//!         .execute()
//!         .await?;
//!
//!     // Spawn a task to handle the result
//!     tokio::spawn(async move {
//!         if let Ok(result) = future.get_result().await {
//!             match result {
//!                 None => {}
//!                 Some(data) => {
//!                     let email_result: EmailResult = serde_json::from_value(data).unwrap();
//!                     println!("Email result: {:?}", email_result);
//!                 }
//!             }
//!         }
//!     });
//!
//!     // Start the worker engine (this will run indefinitely)
//!     engine.start().await?;
//!
//!     Ok(())
//! }
//! ```
//!
//! ## Activity Orchestration
//!
//! Activities can execute other activities using the `ActivityExecutor` available in the
//! `ActivityContext`. This enables building complex workflows and activity orchestration patterns.
//!
//! ```rust,no_run
//! use runner_q::{ActivityHandler, ActivityContext, ActivityHandlerResult, ActivityOption, ActivityPriority, ActivityError};
//! use async_trait::async_trait;
//!
//! pub struct OrderProcessingActivity;
//!
//! #[async_trait]
//! impl ActivityHandler for OrderProcessingActivity {
//!     async fn handle(&self, payload: serde_json::Value, context: ActivityContext) -> ActivityHandlerResult {
//!         let order_id = payload["order_id"]
//!             .as_str()
//!             .ok_or_else(|| ActivityError::NonRetry("Missing order_id".to_string()))?;
//!
//!         // Step 1: Validate payment
//!         let _payment_future = context.activity_executor
//!             .activity("validate_payment")
//!             .payload(serde_json::json!({"order_id": order_id}))
//!             .priority(ActivityPriority::High)
//!             .max_retries(3)
//!             .timeout(std::time::Duration::from_secs(120))
//!             .execute()
//!             .await.map_err(|e| ActivityError::Retry(format!("Failed to enqueue payment validation: {}", e)))?;
//!
//!         // Step 2: Update inventory
//!         let _inventory_future = context.activity_executor
//!             .activity("update_inventory")
//!             .payload(serde_json::json!({"order_id": order_id}))
//!             .execute()
//!             .await.map_err(|e| ActivityError::Retry(format!("Failed to enqueue inventory update: {}", e)))?;
//!
//!         // Step 3: Schedule delivery notification for later
//!         context.activity_executor
//!             .activity("send_delivery_notification")
//!             .payload(serde_json::json!({"order_id": order_id, "customer_email": payload["customer_email"]}))
//!             .max_retries(5)
//!             .timeout(std::time::Duration::from_secs(300))
//!             .delay(std::time::Duration::from_secs(3600)) // 1 hour
//!             .execute()
//!             .await.map_err(|e| ActivityError::Retry(format!("Failed to schedule notification: {}", e)))?;
//!
//!         Ok(Some(serde_json::json!({
//!             "order_id": order_id,
//!             "status": "processing",
//!             "steps_initiated": ["payment_validation", "inventory_update", "delivery_notification"]
//!         })))
//!     }
//!
//!     fn activity_type(&self) -> String {
//!         "process_order".to_string()
//!     }
//! }
//! ```

pub mod activity;
pub mod config;
pub mod observability;
pub mod queue;
pub mod runner;
pub mod storage;

// Re-export main types for easy access
pub use crate::config::WorkerConfig;
#[cfg(feature = "postgres")]
pub use crate::observability::QueueInspector;
#[cfg(all(feature = "axum-ui", feature = "postgres"))]
pub use crate::observability::{observability_api, runnerq_ui};
pub use crate::observability::{
    ActivityEvent, ActivityEventType, ActivitySnapshot, DeadLetterRecord, QueueStats, CONSOLE_HTML,
};
pub use crate::runner::error::WorkerError;
pub use crate::runner::runner::{
    ActivityBuilder, ActivityExecutor, MetricsSink, WorkerEngine, WorkerEngineBuilder,
};
pub use activity::activity::{
    ActivityContext, ActivityFuture, ActivityHandler, ActivityHandlerResult, ActivityPriority,
    ActivityStatus, OnDuplicate,
};
pub use activity::error::{ActivityError, RetryableError};

// Re-export storage types for custom backend implementations
pub use crate::storage::{InspectionStorage, QueueStorage, Storage, StorageError};

// Re-export PostgresBackend when the postgres feature is enabled
#[cfg(feature = "postgres")]
pub use crate::storage::PostgresBackend;