Crate qstash_rs 

qstash-rs

A hand-written Rust SDK for Upstash QStash.

Scope

Current crate focus:

• publish and batch publish
• messages and bulk cancellation
• logs
• dead letter queue (DLQ)
• queues and enqueue
• schedules
• URL groups
• signing keys
• inbound signature verification via Receiver

Flow-control management endpoints and product-specific convenience integrations are intentionally out of scope for now.

Install

cargo add qstash-rs

Quick start

Publish JSON

use qstash_rs::{Client, Destination, PublishRequest};

#[tokio::main]
async fn main() -> Result<(), qstash_rs::Error> {
    let client = Client::new(std::env::var("QSTASH_TOKEN").expect("missing QSTASH_TOKEN"))?;

    let response = client
        .publish(
            PublishRequest::builder(Destination::url("https://example.com/tasks")?)
                .json_body(&serde_json::json!({ "hello": "world" }))?
                .retries(3)
                .label("demo")
                .build(),
        )
        .await?;

    println!("published: {:?}", response);
    Ok(())
}

Batch publish

use qstash_rs::{BatchRequest, Client, Destination, PublishRequest};

#[tokio::main]
async fn main() -> Result<(), qstash_rs::Error> {
    let client = Client::new(std::env::var("QSTASH_TOKEN").expect("missing QSTASH_TOKEN"))?;

    let batch = [
        BatchRequest::new(
            PublishRequest::builder(Destination::url("https://example.com/first")?)
                .body("first")
                .build(),
        ),
        BatchRequest::new(
            PublishRequest::builder(Destination::url_group("workers"))
                .body("second")
                .build(),
        )
        .queue_name("critical"),
    ];

    let response = client.batch(batch).await?;
    println!("batched: {:?}", response);
    Ok(())
}

Queue operations

use qstash_rs::{Client, Destination, PublishRequest, QueueUpsertRequest};

#[tokio::main]
async fn main() -> Result<(), qstash_rs::Error> {
    let client = Client::new(std::env::var("QSTASH_TOKEN").expect("missing QSTASH_TOKEN"))?;

    client
        .queues()
        .upsert(QueueUpsertRequest {
            queue_name: "emails".into(),
            parallelism: 1,
        })
        .await?;

    client
        .queues()
        .enqueue(
            "emails",
            PublishRequest::builder(Destination::url("https://example.com/email")?)
                .json_body(&serde_json::json!({ "kind": "welcome" }))?
                .build(),
        )
        .await?;

    Ok(())
}

Schedule a recurring delivery

use qstash_rs::{Client, Destination, ScheduleRequest};

#[tokio::main]
async fn main() -> Result<(), qstash_rs::Error> {
    let client = Client::new(std::env::var("QSTASH_TOKEN").expect("missing QSTASH_TOKEN"))?;

    let schedule = client
        .schedules()
        .create(
            ScheduleRequest::builder(
                Destination::url("https://example.com/report")?,
                "*/15 * * * *",
            )
            .json_body(&serde_json::json!({ "report": "daily" }))?
            .queue_name("reports")
            .build(),
        )
        .await?;

    println!("schedule: {}", schedule.schedule_id);
    Ok(())
}

Verify inbound signatures

use qstash_rs::{Receiver, ReceiverConfig, VerifyRequest};

fn verify(signature: &str, body: &[u8], url: &str) -> Result<(), qstash_rs::Error> {
    let receiver = Receiver::new(ReceiverConfig {
        current_signing_key: std::env::var("QSTASH_CURRENT_SIGNING_KEY")
            .expect("missing QSTASH_CURRENT_SIGNING_KEY"),
        next_signing_key: std::env::var("QSTASH_NEXT_SIGNING_KEY")
            .expect("missing QSTASH_NEXT_SIGNING_KEY"),
    });

    receiver.verify(VerifyRequest {
        signature,
        body,
        url: Some(url),
        clock_tolerance_seconds: 0,
    })
}

Testing

Default test suite is deterministic and mock-server based:

cargo test
cargo clippy --all-targets --all-features -- -D warnings

Optional live smoke tests:

QSTASH_TOKEN=... QSTASH_RUN_LIVE_TESTS=1 cargo test live_ -- --nocapture

Re-exports

pub use crate::client::BatchRequest;

pub use crate::client::CancelResult;

pub use crate::client::Client;

pub use crate::client::ClientBuilder;

pub use crate::client::Destination;

pub use crate::client::DlqApi;

pub use crate::client::DlqDeleteResult;

pub use crate::client::DlqFilter;

pub use crate::client::DlqMessage;

pub use crate::client::DlqPage;

pub use crate::client::Endpoint;

pub use crate::client::LogEntry;

pub use crate::client::LogFilter;

pub use crate::client::LogState;

pub use crate::client::LogsApi;

pub use crate::client::LogsPage;

pub use crate::client::Message;

pub use crate::client::MessageFilter;

pub use crate::client::MessagesApi;

pub use crate::client::PublishRequest;

pub use crate::client::PublishRequestBuilder;

pub use crate::client::PublishResponse;

pub use crate::client::PublishedMessage;

pub use crate::client::Queue;

pub use crate::client::QueueUpsertRequest;

pub use crate::client::QueuesApi;

pub use crate::client::Redaction;

pub use crate::client::Schedule;

pub use crate::client::ScheduleCreateResponse;

pub use crate::client::ScheduleRequest;

pub use crate::client::ScheduleRequestBuilder;

pub use crate::client::SchedulesApi;

pub use crate::client::SigningKeys;

pub use crate::client::SigningKeysApi;

pub use crate::client::UrlGroup;

pub use crate::client::UrlGroupsApi;

pub use crate::receiver::Receiver;

pub use crate::receiver::ReceiverConfig;

pub use crate::receiver::VerifyRequest;

Modules

client

Async QStash client.

receiver

Enums

Error

Errors returned by the QStash client and receiver.

Type Aliases

Result

Result alias used throughout the crate.