# liteq

High-performance job queue library for Rust using Redis with auto-retry, connection pooling, dependency injection, and multi-consumer fair distribution.

## Features

- **🚀 High Performance**: ZSET optimization for scheduled jobs, 50% reduction in network roundtrips
- **⚡ Auto-Scaling**: Multi-consumer fair distribution with auto-registration and heartbeat
- **🔄 Auto-Retry**: Automatic retry with exponential backoff when Redis connection fails
- **🛡️ Resilient**: RabbitMQ-style connection supervisor - continues working even when Redis is temporarily down
- **💉 Dependency Injection**: Built-in support for injecting shared state into job handlers via `.data()`
- **📊 Monitoring**: Health checks, job counts, and queue statistics
- **📝 Structured Logging**: Uses `tracing` for structured, production-ready logging
- **🔒 Type Safe**: Full Rust type safety with generics
- **⚙️ Async/Await**: Built on Tokio for efficient async processing
- **🎨 Builder Pattern**: Fluent API for configuring queues and workers

## Installation

Add to your `Cargo.toml`:

```toml
[dependencies]
liteq = "1.0"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tracing = "0.1"
tracing-subscriber = "0.3"
chrono = "0.4"
```

## Quick Start

### Producer (Send Jobs with ETA Scheduling)

```rust
use liteq::{Job, JobQueue, QueueConfig, RedisConfig};
use chrono::Utc;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let queue = JobQueue::new(
        QueueConfig::new("orders"),
        RedisConfig::new("redis://127.0.0.1:6379")
    ).await?;

    // Regular job (process immediately)
    let job = Job::new(
        serde_json::json!({"order_id": 123, "item": "widget"}),
        "orders"
    );
    queue.enqueue(job).await?;

    // Scheduled job (process in 2 hours)
    let scheduled_job = Job::new(
        serde_json::json!({"order_id": 124, "item": "widget"}),
        "orders"
    ).with_eta(Utc::now() + chrono::Duration::hours(2));
    queue.enqueue(scheduled_job).await?;

    Ok(())
}
```

### Consumer (Process Jobs with Multi-Instance Support)

```rust
use liteq::{JobResult, SubscriberRegistry};
use std::sync::Arc;

#[derive(Clone)]
struct AppData {
    db_url: String,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut registry = SubscriberRegistry::new();
    let app_data = Arc::new(AppData {
        db_url: "postgresql://localhost:5432/mydb".to_string(),
    });

    // Auto-registers and enables fair distribution!
    registry.register("orders", handle_orders)
        .with_data(app_data)
        .with_pool_size(20)
        .with_concurrency(5)
        .build();

    registry.run().await?;
    Ok(())
}

fn handle_orders(data: Vec<u8>, app_data: Arc<AppData>) -> JobResult<()> {
    // Deserialize payload directly
    let order: serde_json::Value = serde_json::from_slice(&data)?;
    println!("Processing order: {}", order);

    Ok(())
}
```

**Run Multiple Instances for Fair Distribution:**
```bash
# Terminal 1
cargo run --example queue_consumer

# Terminal 2
cargo run --example queue_consumer

# Terminal 3
cargo run --example queue_producer
```

Jobs are automatically distributed 50:50 between instances! 🎯

## Multi-Consumer Fair Distribution

**Run multiple consumer instances for automatic fair job distribution!**

Each consumer instance auto-registers with Redis and gets a unique ID. Jobs are distributed using **modulo arithmetic** for perfect round-robin:

```bash
# Terminal 1
cargo run --example queue_consumer

# Terminal 2
cargo run --example queue_consumer
```

**Features:**
- ✅ **Auto-Registration**: Zero configuration needed
- ✅ **Fair Distribution**: Perfect 50:50 (or 33:33:33, etc) split
- ✅ **Heartbeat**: 30s TTL, auto-failure detection
- ✅ **Auto-Recovery**: Consumer crash → jobs redistributed automatically
- ✅ **No Starvation**: Every instance gets equal job share

**How It Works:**
```
Job 1 → Consumer 1 (1 % 2 = 1)
Job 2 → Consumer 0 (2 % 2 = 0)
Job 3 → Consumer 1 (3 % 2 = 1)
Job 4 → Consumer 0 (4 % 2 = 0)
Perfect round-robin!
```

For details, see [MULTI_CONSUMER_FLOW.md](MULTI_CONSUMER_FLOW.md)

## ETA Scheduling with ZSET Optimization

**Schedule jobs to run at specific times with zero performance penalty!**

```rust
use chrono::Utc;

// Process this job in 2 hours
let job = Job::new(task, "orders")
    .with_eta(Utc::now() + chrono::Duration::hours(2));
queue.enqueue(job).await?;
```

**Benefits:**
- ✅ **50% Reduction** in network roundtrips for scheduled jobs
- ✅ **40% Reduction** in CPU usage (no redundant JSON parsing)
- ✅ **No Job Bouncing**: Jobs stay in Redis until ready
- ✅ **Priority Queue**: Ready scheduled jobs processed before regular jobs

**How It Works:**
- Jobs with ETA → Stored in ZSET (score = ETA timestamp)
- Jobs without ETA → Stored in LIST (process immediately)
- Dequeue script checks ZSET first (ready ETA jobs), then LIST

For details, see [PERFORMANCE_IMPROVEMENTS.md](PERFORMANCE_IMPROVEMENTS.md)

## Health Checks & Monitoring

**Monitor queue health and job counts in real-time:**

```rust
// Get job counts
let (regular, scheduled) = queue.get_job_counts().await?;

// Get detailed statistics
let stats = queue.get_queue_stats().await?;
println!("Total pending: {}", stats.total_pending);

// Health check
let health = registry.health_check("orders").await?;
println!("Status: {:?}", health.status);
```

**Run monitoring demo:**
```bash
cargo run --example monitoring_demo orders
```

For details, see [MONITORING.md](MONITORING.md)

## Dependency Injection

Inject shared state into your job handlers:

```rust
registry.register("queue_name", handler)
    .with_data(shared_state)
    .with_pool_size(10)
    .with_concurrency(3)
    .build();
```

The `data` is passed as `Arc<T>` to your handler function, making it easy to share:
- Database connections
- API clients
- Configuration
- Counters/metrics

## Connection Supervision & Auto-Retry

The library automatically handles Redis connection failures with RabbitMQ-style supervision:

**Features:**
- ✅ **Single Retry Loop**: Only 1 retry log (not N×workers)
- ✅ **Worker Coordination**: Workers wait for "ready" signal
- ✅ **Clean Logs**: No retry spam during Redis downtime
- ✅ **Auto-Recovery**: Single reconnection attempt, all workers notified

**Default retry settings:**
- **Max Attempts**: 20 retries
- **Initial Delay**: 500ms
- **Max Delay**: 30 seconds
- **Backoff**: 2x exponential

## Builder Pattern

Configure queues with the fluent builder API:

```rust
SubscriberRegistry::new()
    .with_redis("redis://127.0.0.1:6379")
    .register("queue1", handler1)
        .with_pool_size(20)
        .with_concurrency(5)
        .build()
    .register("queue2", handler2)
        .with_data(shared_data)
        .with_pool_size(10)
        .build();
```

Available builder methods:
- `.with_redis(url)` - Set Redis connection URL
- `.register(queue, handler)` - Register a handler
- `.with_data(data)` - Inject shared state (optional)
- `.with_pool_size(size)` - Set connection pool size
- `.with_concurrency(num)` - Set number of concurrent workers
- `.with_min_idle(num)` - Set minimum idle connections
- `.build()` - Complete registration

## Examples

```bash
# Run producer (send jobs)
cargo run --example queue_producer

# Run consumer (process jobs)
cargo run --example queue_consumer

# Monitor queue stats
cargo run --example monitoring_demo orders
```

**Testing Multiple Instances:**
```bash
# Terminal 1
cargo run --example queue_consumer

# Terminal 2
cargo run --example queue_consumer

# Terminal 3
cargo run --example queue_producer
```

You'll see fair 50:50 distribution! 🎯

## Architecture

- **ZSET**: Scheduled jobs with ETA (score = timestamp)
- **LIST**: Regular jobs (FIFO)
- **Connection Pooling**: Each queue has its own pool with configurable size
- **Connection Supervisor**: RabbitMQ-style supervision per pool
- **Worker Management**: Multiple workers per queue with configurable concurrency
- **Multi-Consumer**: Auto-registration with heartbeat for fair distribution
- **Retry Logic**: Exponential backoff with configurable attempts
- **Structured Logging**: All operations logged with tracing
- **Dependency Injection**: Per-queue shared state via `.data()`

## Benefits

🚀 **High Performance** - ZSET optimization reduces roundtrips by 50%  
⚡ **Auto-Scaling** - Multi-consumer fair distribution with zero config  
🔄 **Resilient** - Survives Redis restarts with automatic reconnection  
📊 **Observable** - Health checks and monitoring with tracing  
🎯 **Flexible** - Configure per-queue pool sizes and worker counts  
🔒 **Type Safe** - Full type safety with generics and Arc<T> for shared state  
💾 **Persistent** - Jobs stored in Redis even if consumer is down  

## Documentation

- [README.md](README.md) - Getting started guide
- [MULTI_CONSUMER_FLOW.md](MULTI_CONSUMER_FLOW.md) - Multi-consumer fair distribution
- [PERFORMANCE_IMPROVEMENTS.md](PERFORMANCE_IMPROVEMENTS.md) - ZSET optimization details
- [MONITORING.md](MONITORING.md) - Health checks and monitoring guide
- [CONNECTION_SUPERVISION.md](CONNECTION_SUPERVISION.md) - Connection supervision details
- [IMPLEMENTATION_COMPLETE.md](IMPLEMENTATION_COMPLETE.md) - Implementation details

## License

MIT

## Contributing

Contributions are welcome! Please feel free to submit a Pull Request.
