### Complete Effectum Example: Job Runner Setup

Source: https://context7.com/dimfeld/effectum/llms.txt

Defines a job runner for a 'counter' job, enabling auto-heartbeat. This is part of a larger example demonstrating Effectum's typical usage.

```rust
#[derive(Debug)]
struct JobContext {
    counter: AtomicI64,
}

#[derive(Serialize, Deserialize)]
struct CounterPayload {
    add: i64,
}

async fn counter_job(
    job: RunningJob,
    context: Arc<JobContext>,
) -> Result<String, Error> {
    let payload: CounterPayload = job.json_payload()?;
    let new_value = context.counter.fetch_add(payload.add, Ordering::Relaxed) + payload.add;
    Ok(format!("Counter is now {}", new_value))
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    // Initialize queue
    let queue = Queue::new(Path::new("app-jobs.db")).await?;

    // Shared context for all job runners
    let context = Arc::new(JobContext {
        counter: AtomicI64::new(0),
    });

    // Define job types
    let counter_runner = JobRunner::builder("counter", counter_job)
        .autoheartbeat(true)
        .build();
```

--------------------------------

### Complete Effectum Example: Worker and Job Submission

Source: https://context7.com/dimfeld/effectum/llms.txt

Sets up an Effectum worker with defined job runners and concurrency limits, then submits multiple jobs to the queue with specified run times.

```rust
    // Start worker
    let worker = Worker::builder(&queue, context.clone())
        .jobs(vec![counter_runner])
        .min_concurrency(3)
        .max_concurrency(10)
        .build()
        .await?;

    // Submit jobs
    for i in 0..10 {
        let job_id = Job::builder("counter")
            .json_payload(&CounterPayload { add: i })? 
            .run_at(OffsetDateTime::now_utc() + Duration::from_millis(i as u64 * 100))
            .add_to(&queue)
            .await?;

        println!("Submitted job {}: {}", i, job_id);
    }

    // Wait for jobs to complete
    tokio::time::sleep(Duration::from_secs(2)).await;

    // Check final counter value
    println!("Final counter: {}", context.counter.load(Ordering::Relaxed));

    // Query queue status
    let counts = queue.num_active_jobs().await?;
    println!("Remaining: {} pending, {} running", counts.pending, counts.running);

    // Graceful shutdown
    worker.unregister(Some(Duration::from_secs(30))).await?;
    queue.close(Duration::from_secs(30)).await?;

    Ok(())
}
```

--------------------------------

### Add Recurring Jobs with Cron Schedule

Source: https://context7.com/dimfeld/effectum/llms.txt

Schedules a recurring job using a cron expression. This example sets the job to run daily at 3 AM and will execute immediately upon addition, then follow the schedule.

```rust
    // Recurring job with cron schedule (every day at 3 AM)
    let report_job = Job::builder("generate_report")
        .json_payload(&"weekly-summary")?
        .priority(5)
        .build();

    queue.add_recurring_job(
        "morning-report".to_string(),
        RecurringJobSchedule::from_cron_string("0 0 3 * * *".to_string())?,
        report_job,
        true, // Run immediately, then follow schedule
    ).await?;
```

--------------------------------

### List Recurring Jobs by Prefix

Source: https://context7.com/dimfeld/effectum/llms.txt

Fetches a list of recurring jobs whose names start with a specified prefix. This is useful for managing groups of related jobs.

```rust
    // List recurring jobs by prefix
    let jobs = queue.list_recurring_jobs_with_prefix("daily-").await?;
    println!("Daily jobs: {:?}", jobs);
```

--------------------------------

### Implement Checkpointing and Heartbeats for Long-Running Jobs

Source: https://context7.com/dimfeld/effectum/llms.txt

Long-running jobs can save progress via checkpoints and extend their timeout with heartbeats. This example shows how to use `JobRunner` to handle state and manual heartbeats.

```rust
use std::sync::Arc;
use effectum::{JobRunner, RunningJob, Error};
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone)]
struct Context;

#[derive(Serialize, Deserialize)]
struct ProcessingState {
    items_processed: usize,
    total_items: usize,
    last_item_id: Option<u64>,
}

let batch_processor = JobRunner::builder("batch_process",
    |job: RunningJob, _ctx: Arc<Context>| async move {
        // Resume from checkpoint if this is a retry
        let mut state: ProcessingState = job.json_payload()?;

        let items_to_process: Vec<u64> = vec![1, 2, 3, 4, 5]; // Fetch from DB

        for (i, item_id) in items_to_process.iter().enumerate() {
            // Process item...
            state.items_processed += 1;
            state.last_item_id = Some(*item_id);

            // Checkpoint every 10 items - if job fails, it resumes from here
            if i % 10 == 0 {
                job.checkpoint_json(&state).await?;
            }

            // Manual heartbeat to prevent timeout on slow items
            if i % 5 == 0 {
                let new_expiry = job.heartbeat().await?;
                println!("Extended timeout to {}", new_expiry);
            }
        }

        Ok::<_, Error>(format!("Processed {} items", state.items_processed))
    })
    .autoheartbeat(false) // Using manual heartbeats
    .build();

```

--------------------------------

### Get Recurring Job Information

Source: https://context7.com/dimfeld/effectum/llms.txt

Retrieves detailed information about a specific recurring job, including its schedule, last run status, and next scheduled run.

```rust
    // Get recurring job status
    let info = queue.get_recurring_job_info("daily-cleanup".to_string()).await?;
    println!("Schedule: {:?}", info.schedule);
    if let Some(last) = info.last_run {
        println!("Last run: {:?} - {:?}", last.started_at, last.state);
    }
    if let Some((id, time)) = info.next_run {
        println!("Next run: {} at {}", id, time);
    }
```

--------------------------------

### Define and run jobs with Effectum

Source: https://github.com/dimfeld/effectum/blob/master/effectum/README.md

Demonstrates setting up a queue, defining a job runner, and submitting a job with a payload.

```rust
use effectum::{Error, Job, JobState, JobRunner, RunningJob, Queue, Worker};

#[derive(Debug)]
pub struct JobContext {
   // database pool or other things here
}

#[derive(Serialize, Deserialize)]
struct RemindMePayload {
  email: String,
  message: String,
}

async fn remind_me_job(job: RunningJob, context: Arc<JobContext>) -> Result<(), Error> {
    let payload: RemindMePayload = job.json_payload()?;
    // do something with the job
    Ok(())
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Error> {
  // Create a queue
  let queue = Queue::new(&PathBuf::from("effectum.db")).await?;

  // Define a type job for the queue.
  let a_job = JobRunner::builder("remind_me", remind_me_job).build();

  let context = Arc::new(JobContext{
    // database pool or other things here
  });

  // Create a worker to run jobs.
  let worker = Worker::builder(&queue, context)
    .max_concurrency(10)
    .jobs([a_job])
    .build();

  // Submit a job to the queue.
  let job_id = Job::builder("remind_me")
    .run_at(time::OffsetDateTime::now_utc() + std::time::Duration::from_secs(3600))
    .json_payload(&RemindMePayload {
        email: "me@example.com".to_string(),
        message: "Time to go!".to_string()
    })?
    .add_to(&queue)
    .await?;

  // See what's happening with the job.
  let status = queue.get_job_status(job_id).await?;
  assert_eq!(status.state, JobState::Pending);

  // Do other stuff...

  Ok(())
}
```

--------------------------------

### Create Effectum Queue with Default Options

Source: https://context7.com/dimfeld/effectum/llms.txt

Initializes a new Effectum queue using default settings, storing job data in the specified SQLite database file. Ensure the path is valid and accessible.

```rust
use std::path::Path;
use effectum::{Queue, QueueOptions, JobRecoveryBehavior};

#[tokio::main]
async fn main() -> Result<(), effectum::Error> {
    // Simple queue creation with default options
    let queue = Queue::new(Path::new("jobs.db")).await?;

    // ... rest of the code ...
    Ok(())
}
```

--------------------------------

### Submit Jobs to Queue

Source: https://context7.com/dimfeld/effectum/llms.txt

Demonstrates submitting individual jobs with scheduling, priority, and retry configurations, as well as atomic batch submission.

```rust
use std::time::Duration;
use effectum::{Queue, Job, Retries};
use serde_json::json;
use time::OffsetDateTime;

async fn submit_jobs(queue: &Queue) -> Result<(), effectum::Error> {
    // Simple job that runs immediately
    let job_id = Job::builder("send_email")
        .json_payload(&json!({
            "to": "user@example.com",
            "subject": "Welcome!",
            "body": "Thanks for signing up."
        }))?
        .add_to(queue)
        .await?;

    // Scheduled job with high priority
    let future_time = OffsetDateTime::now_utc() + Duration::from_secs(3600);
    let scheduled_job_id = Job::builder("generate_report")
        .json_payload(&json!({"report_type": "daily"}))?
        .run_at(future_time)
        .priority(10) // Higher priority runs first
        .name("daily-report-2024") // Optional searchable name
        .add_to(queue)
        .await?;

    // Job with custom retry configuration
    let robust_job_id = Job::builder("sync_external_api")
        .json_payload(&"https://api.example.com")?
        .retries(Retries {
            max_retries: 5,
            backoff_initial_interval: Duration::from_secs(30),
            backoff_multiplier: 2.0,
            backoff_randomization: 0.3,
        })
        .timeout(Duration::from_secs(120))
        .weight(3) // Counts as 3 against concurrency limit
        .add_to(queue)
        .await?;

    // Submit multiple jobs atomically
    let job_ids = queue.add_jobs(vec![
        Job::builder("task_a").build(),
        Job::builder("task_b").build(),
        Job::builder("task_c").build(),
    ]).await?;

    println!("Submitted jobs: {:?}", job_ids);
    Ok(())
}
```

--------------------------------

### Create Effectum Queue with Custom Recovery Behavior

Source: https://context7.com/dimfeld/effectum/llms.txt

Configures a new Effectum queue with custom options, specifically setting the job recovery behavior to fail and retry with backoff for jobs that were running during an unexpected shutdown. This enhances resilience for long-running tasks.

```rust
use std::path::Path;
use effectum::{Queue, QueueOptions, JobRecoveryBehavior};

#[tokio::main]
async fn main() -> Result<(), effectum::Error> {
    // ... previous code ...

    // Queue with custom options for handling jobs left running after unexpected shutdown
    let queue = Queue::builder(Path::new("jobs.db"))
        .job_recovery_behavior(JobRecoveryBehavior::FailAndRetryWithBackoff)
        .build()
        .await?;

    // ... rest of the code ...
    Ok(())
}
```

--------------------------------

### Define Job Runner with Closure and Context

Source: https://context7.com/dimfeld/effectum/llms.txt

Defines a job runner named 'send_email' using a closure. It accepts a `RunningJob` and a shared `AppContext`. The job payload is expected to be JSON serializable into an `EmailPayload` struct. Requires `serde` for serialization/deserialization.

```rust
use std::sync::Arc;
use effectum::{JobRunner, RunningJob, Error};
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone)]
pub struct AppContext {
    pub db_pool: String, // Your database pool or other shared resources
}

#[derive(Serialize, Deserialize)]
struct EmailPayload {
    to: String,
    subject: String,
    body: String,
}

// Define a job runner using a closure
let send_email_job = JobRunner::builder("send_email",
    |job: RunningJob, ctx: Arc<AppContext>| async move {
        let payload: EmailPayload = job.json_payload()?;

        // Simulate sending email
        println!("Sending email to {} via {}", payload.to, ctx.db_pool);

        // Return success info (can be any Serialize type)
        Ok::<_, Error>(format!("Email sent to {}", payload.to))
    })
    .autoheartbeat(true) // Automatically extend timeout while running
    .build();
```

--------------------------------

### Define Job Runner with Async Function

Source: https://context7.com/dimfeld/effectum/llms.txt

Defines a job runner named 'process_order' using a standalone async function. It accepts a `RunningJob` and a shared `AppContext`. The job payload is expected to be a `u64` representing an order ID. This runner demonstrates manual heartbeat calls for long-running tasks.

```rust
use std::sync::Arc;
use effectum::{JobRunner, RunningJob, Error};
use serde::{Deserialize, Serialize};

// Assume AppContext and EmailPayload are defined as above
#[derive(Debug, Clone)]
pub struct AppContext {
    pub db_pool: String, // Your database pool or other shared resources
}

#[derive(Serialize, Deserialize)]
struct EmailPayload {
    to: String,
    subject: String,
    body: String,
}

// Or define as a standalone async function
async fn process_order(job: RunningJob, ctx: Arc<AppContext>) -> Result<String, Error> {
    let order_id: u64 = job.json_payload()?;

    // Long-running job can send manual heartbeats
    job.heartbeat().await?;

    // Process the order...
    Ok(format!("Order {} processed", order_id))
}

let order_job = JobRunner::builder("process_order", process_order)
    .format_failures_with_debug(true) // Use Debug formatting for errors
    .build();
```

--------------------------------

### Create Workers with Job Runners

Source: https://context7.com/dimfeld/effectum/llms.txt

Configures workers to process specific job types with defined concurrency limits. Workers must be kept alive to continue processing jobs.

```rust
use std::sync::Arc;
use effectum::{Queue, Worker, JobRunner, JobRegistry, RunningJob, Error};

#[derive(Debug, Clone)]
struct MyContext {
    counter: Arc<std::sync::atomic::AtomicI64>,
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let queue = Queue::new(std::path::Path::new("jobs.db")).await?;
    let context = MyContext {
        counter: Arc::new(std::sync::atomic::AtomicI64::new(0)),
    };

    // Define job runners
    let job_a = JobRunner::builder("job_a", |_job: RunningJob, _ctx: MyContext| async {
        Ok::<_, String>("done")
    }).build();

    let job_b = JobRunner::builder("job_b", |_job: RunningJob, _ctx: MyContext| async {
        Ok::<_, String>("done")
    }).build();

    // Create worker with job list directly
    let worker = Worker::builder(&queue, context.clone())
        .jobs(vec![job_a.clone(), job_b.clone()])
        .min_concurrency(5)  // Fetch more jobs when running count drops below this
        .max_concurrency(10) // Maximum concurrent jobs
        .build()
        .await?;

    // Alternative: Use a JobRegistry and limit to specific job types
    let mut registry = JobRegistry::new([&job_a, &job_b]);

    let worker2 = Worker::builder(&queue, context)
        .registry(&registry)
        .limit_job_types(&["job_a"]) // Only run job_a
        .max_concurrency(5)
        .build()
        .await?;

    // Worker must be kept alive; dropping it stops processing
    // Gracefully unregister with optional timeout
    worker.unregister(Some(std::time::Duration::from_secs(30))).await?;

    Ok(())
}
```

--------------------------------

### Manage Job Status and Operations in Effectum

Source: https://context7.com/dimfeld/effectum/llms.txt

Query job status, update pending jobs, or cancel them. This function demonstrates how to interact with the Effectum queue for various job management tasks.

```rust
use effectum::{Queue, Job, JobState, JobUpdate};
use uuid::Uuid;
use time::OffsetDateTime;

async fn manage_jobs(queue: &Queue, job_id: Uuid) -> Result<(), effectum::Error> {
    // Get detailed job status
    let status = queue.get_job_status(job_id).await?;

    println!("Job {} is {:?}", status.id, status.state);
    println!("  Type: {}", status.job_type);
    println!("  Priority: {}", status.priority);
    println!("  Retries: {}/{}", status.current_try.unwrap_or(0), status.max_retries);

    // Check run history
    for run in &status.run_info {
        println!("  Run: success={}, start={}, end={}",
            run.success, run.start, run.end);
    }

    // Update a pending job (fails if running or finished)
    if status.state == JobState::Pending {
        queue.update_job(
            JobUpdate::builder(job_id)
                .run_at(OffsetDateTime::now_utc()) // Run now
                .priority(100) // Increase priority
                .json_payload(&"updated payload")? 
                .build()
        ).await?;
    }

    // Cancel a pending job
    match queue.cancel_job(job_id).await {
        Ok(()) => println!("Job cancelled"),
        Err(effectum::Error::JobRunning) => println!("Cannot cancel: job is running"),
        Err(effectum::Error::JobFinished) => println!("Cannot cancel: job already finished"),
        Err(e) => return Err(e),
    }

    // Query jobs by name
    let reports = queue.get_jobs_by_name("daily-report".to_string(), 10).await?;
    for job in reports {
        println!("Found report job: {} - {:?}", job.id, job.state);
    }

    // Get queue statistics
    let counts = queue.num_active_jobs().await?;
    println!("Queue: {} pending, {} running", counts.pending, counts.running);

    Ok(())
}

```

--------------------------------

### Add Recurring Jobs with Fixed Interval

Source: https://context7.com/dimfeld/effectum/llms.txt

Schedules a recurring job to run at a fixed interval. The job is not run immediately upon addition.

```rust
use std::time::Duration;
use effectum::{Queue, Job, RecurringJobSchedule};

async fn setup_recurring_jobs(queue: &Queue) -> Result<(), effectum::Error> {
    // Recurring job with fixed interval
    let cleanup_job = Job::builder("cleanup_old_data")
        .json_payload(&serde_json::json!({"days_to_keep": 30}))?
        .build();

    queue.add_recurring_job(
        "daily-cleanup".to_string(),
        RecurringJobSchedule::RepeatEvery {
            interval: Duration::from_secs(24 * 60 * 60)
        },
        cleanup_job,
        false, // Don't run immediately on add
    ).await?;
```

--------------------------------

### Gracefully Close Effectum Queue

Source: https://context7.com/dimfeld/effectum/llms.txt

Closes the Effectum queue, allowing a specified duration for any currently running jobs to complete gracefully. This is important for clean shutdowns.

```rust
use std::path::Path;
use effectum::{Queue, QueueOptions, JobRecoveryBehavior};

#[tokio::main]
async fn main() -> Result<(), effectum::Error> {
    // ... previous code ...

    // Gracefully close the queue with a timeout
    queue.close(std::time::Duration::from_secs(30)).await?;

    Ok(())
}
```

--------------------------------

### Upsert Recurring Job

Source: https://context7.com/dimfeld/effectum/llms.txt

Adds a new recurring job or updates an existing one in a single operation. If the job does not exist, it is inserted; otherwise, it is updated.

```rust
    // Upsert: add or update in one call
    queue.upsert_recurring_job(
        "health-check".to_string(),
        RecurringJobSchedule::RepeatEvery {
            interval: Duration::from_secs(60)
        },
        Job::builder("ping").build(),
        true, // Run immediately if inserting new
    ).await?;
```

--------------------------------

### Update an Existing Recurring Job

Source: https://context7.com/dimfeld/effectum/llms.txt

Modifies an existing recurring job's schedule and payload. This is useful for changing the frequency or details of a job that is already scheduled.

```rust
    // Update an existing recurring job
    let updated_job = Job::builder("generate_report")
        .json_payload(&"detailed-weekly-summary")?
        .build();

    queue.update_recurring_job(
        "morning-report".to_string(),
        RecurringJobSchedule::RepeatEvery {
            interval: Duration::from_secs(7 * 24 * 60 * 60)
        },
        updated_job,
    ).await?;
```

--------------------------------

### Delete a Recurring Job

Source: https://context7.com/dimfeld/effectum/llms.txt

Removes a recurring job from the schedule. This operation permanently deletes the job and its associated schedule.

```rust
    // Delete a recurring job
    queue.delete_recurring_job("health-check".to_string()).await?;
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.