Orchestrator

Orchestrator — Microservice Integration Patterns

Building a distributed Rust backend requires specialized "glue" code to connect web frameworks, background workers, and databases securely. This skill provides the architectural patterns needed to stitch these components together.

Quick Reference: Which Guide Do I Need?

• Fast secret lookups, Argon2id hashing, timing-attack prevention -> Read references/crypto-auth.md
• TOML catalog parsing, Serde, Clap CLI subcommands -> Read references/config-dynamic.md
• Streaming events from Workers to API via Redis Pub/Sub -> Read references/distributed-sse.md
• External RPC SSRF protection for reqwest/Hyper -> Read references/ssrf-protection.md
• ULID task IDs stored as 16-byte SQL values -> Read references/ulid-sqlx.md
• Presigned PUT size limits via S3 bucket policy -> Read references/s3-bucket-policy.md

Core Patterns at a Glance

1. Secure Crypto Auth & Lookups

When dealing with secret identifiers (like a 20-digit Space ID), never store them in plaintext, and never use slow hashes (like Argon2id) for database SELECT lookups. Instead, use a two-step approach:

1. HMAC-SHA256 Fast Index: Hash the ID with a server-side pepper, truncate to 16 bytes, and use this for the database lookup (WHERE idx = $1). 2. Argon2id Verification: Once the row is found, perform a constant-time verification against the stored Argon2id hash.

2. Dynamic Configurations & CLI

Avoid hard-coded endpoints and prices. Parse a reeve.toml catalog into memory at startup using serde and toml. Use clap subcommands to allow a single binary to act as an API server, a background worker, or a migration tool based on the arguments provided.

3. Distributed Server-Sent Events (SSE)

In a microservice topology, the API container holding the user's HTTP request is not the same container executing the background task.

• Worker: Publishes progress to a Redis channel (task:<id>:events).
• API: Acquires a dedicated redis::aio::PubSub connection, subscribes to the channel, and yields axum::response::sse::Event items back to the user via async_stream.

4. External RPC SSRF Protection

External reqwest clients must reject private, loopback, link-local, and unspecified IPs after DNS resolution unless an IP is explicitly allow-listed. Use the current reqwest::ClientBuilder::dns_resolver or a guarded hyper_util::client::legacy::connect::HttpConnector resolver; do not rely on URL string checks.

5. ULID Task IDs and SQLx

Use ulid::Ulid, convert to uuid::Uuid with ulid.into(), and store the 16-byte representation. For BYTEA, bind uuid.as_bytes().as_slice(); for SQL UUID, enable SQLx's uuid feature and bind Uuid directly.

6. S3 Presigned PUT Size Enforcement

Do not try to add content-length-range to presigned PUT URLs. Apply an S3-compatible bucket policy with s3:content-length numeric deny rules scoped to the upload prefixes.

Dependency Matrix (Refresh Before Use)

Known Issues — Orchestrator Skill

Issues discovered through practical compilation testing against actual crate versions (May 2026).

1. hmac v0.13 requires KeyInit trait import (FIXED)

The Hmac::<Sha256>::new_from_slice() call fails without importing KeyInit:

// WRONG:
use hmac::{Hmac, Mac};
let mut mac = Hmac::<Sha256>::new_from_slice(key)?;

// CORRECT:
use hmac::{Hmac, Mac, KeyInit};
let mut mac = Hmac::<Sha256>::new_from_slice(key)?;

The KeyInit trait must be in scope for new_from_slice to be available on Hmac<D>.

2. argon2 v0.6.0-rc.8 API Changes (FIXED)

hash_password signature changed

hash_password no longer takes a separate salt parameter. Salt is now generated independently and embedded in the hash string:

// WRONG (skill's old code):
argon2.hash_password(secret.as_bytes(), &salt)?

// CORRECT (argon2 0.6.x):
argon2.hash_password(secret.as_bytes())?

rand_core not re-exported from argon2::password_hash

Use rand_core directly as a dependency (v0.9):

// WRONG:
use argon2::password_hash::rand_core::OsRng;

// CORRECT:
use rand_core::OsRng;

verify_password takes &str not parsed PasswordHash

// WRONG:
argon2.verify_password(secret.as_bytes(), &parsed_hash)

// CORRECT:
argon2.verify_password(secret.as_bytes(), stored_hash_str)

Updated dependency:

argon2 = "0.6.0-rc.8"
rand_core = "0.9"  # required for OsRng

3. hyper-util dns Feature

The hyper-util crate does NOT have a dns feature flag. The SSRF protection pattern described in the skill that uses hyper-util with dns feature cannot work as documented. Use hickory-dns or trust-dns-resolver instead for custom DNS resolution.

orchestrator-skills

A specialized "glue" skill for building dynamic, secure, and distributed Rust microservices. This skill bridges the gap between individual component libraries (Axum, Apalis, SQLx) by providing production-grade integration patterns for distributed architectures.

Overview

Modern Rust backends—especially those using zero-account designs, dynamic pricing catalogs, and separated API/Worker containers—require specific integration patterns to function securely and reliably.

This skill provides those exact patterns:

• Crypto Auth: Fast $O(1)$ database lookups without timing leaks, using HMAC-SHA256 indexes and Argon2id.
• Dynamic Config: Parsing complex, nested TOML manifests into application state, and building robust CLI routers for multi-mode execution (Monolith vs. Microservice).
• Distributed SSE: Bridging background workers to user-facing HTTP streams using Redis Pub/Sub and Axum.
• SSRF Protection: Guarding external reqwest/Hyper RPC clients with connector-level IP filtering.
• ULID SQLx Mapping: Using ULID task IDs as 16-byte database values without string storage.
• S3 Bucket Policies: Enforcing presigned PUT upload size limits with prefix-scoped bucket policies.

Installation

npx skills add melonask/orchestrator-skills

File Structure

orchestrator/
├── SKILL.md                          # Core overview and routing
└── references/
    ├── crypto-auth.md                # HMAC-SHA256 indexing, Argon2id, Constant-Time checks
    ├── config-dynamic.md             # TOML parsing, Serde, Clap CLI routing
    ├── distributed-sse.md            # Redis Pub/Sub to Axum Server-Sent Events (SSE)
    ├── ssrf-protection.md            # Reqwest/Hyper external RPC IP filtering
    ├── ulid-sqlx.md                  # ULID to uuid::Uuid and SQLx 16-byte storage
    └── s3-bucket-policy.md           # Presigned PUT size limits via bucket policy

License

Provided as-is for development with LLM assistants.

Config & Dynamic CLI Routing

Production orchestrators rely heavily on declarative TOML manifests to avoid hardcoding prices/endpoints, and extensive CLI tooling for operational management.

Dependencies

[dependencies]
toml = "1.1.2+spec-1.1.0"
serde = { version = "1.0.228", features = ["derive"] }
clap = { version = "4.6.1", features = ["derive"] }

Pattern 1: Nested TOML Parsing (The Catalog)

Parse complex TOML structures directly into Axum application state. This powers dynamic pricing arrays and catalog capabilities.

use serde::Deserialize;
use std::collections::HashMap;

// The struct representation of reeve.toml
#[derive(Deserialize, Debug, Clone)]
pub struct ReeveConfig {
    pub assets: HashMap<String, AssetConfig>,
    pub tasks: HashMap<String, TaskConfig>,
}

#[derive(Deserialize, Debug, Clone)]
pub struct AssetConfig {
    pub network: String,
    pub contract: String,
    pub decimals: u8,
}

#[derive(Deserialize, Debug, Clone)]
pub struct TaskConfig {
    pub title: String,
    pub input_schema: String,
    // The [[tasks.X.accepts]] array maps to Vec<AcceptsConfig>
    #[serde(default)]
    pub accepts: Vec<AcceptsConfig>,
}

#[derive(Deserialize, Debug, Clone)]
pub struct AcceptsConfig {
    pub asset: String,
    pub amount: Option<String>,
    pub amount_usd: Option<String>,
}

// Loading the configuration:
pub fn load_config() -> ReeveConfig {
    let toml_str = std::fs::read_to_string("reeve.toml").expect("Failed to read reeve.toml");
    toml::from_str(&toml_str).expect("Failed to parse TOML")
}

Pattern 2: Clap CLI Subcommands

Use clap for operational binaries. The Microservices/Monolith pattern reduces to simply switching which Command is invoked at startup.

use clap::{Parser, Subcommand};

#[derive(Parser)]
#[command(name = "reeve-cli", version, about = "Ops and migrations CLI")]
pub struct Cli {
    #[command(subcommand)]
    pub command: Commands,
}

#[derive(Subcommand)]
pub enum Commands {
    /// Check configuration validity
    Config {
        #[arg(long, default_value = "reeve.toml")]
        file: String,
    },
    /// Run database or infrastructure migrations
    Migrate {
        resource: String, // e.g., "db", "mq", "s3"
        #[arg(long)]
        from: String,
        #[arg(long)]
        to: String,
        #[arg(long)]
        batch: Option<usize>,
        #[arg(long, action)]
        dry_run: bool,
    },
    /// Start the API server
    Serve {
        #[arg(long, default_value = "0.0.0.0:8080")]
        bind: String,
    },
    /// Start the background worker
    Work {
        #[arg(long)]
        queue: String,
    }
}

// In main.rs:
// let cli = Cli::parse();
// match cli.command {
//     Commands::Migrate { resource, from, to, batch, dry_run } => { ... }
//     Commands::Serve { bind } => { ... }
//     _ => { ... }
// }

Crypto Auth — Secure Identifiers & Hashing

In high-performance, passwordless systems (like reeve), you often need to verify a secret (like a 20-digit Space ID) quickly while protecting against timing attacks and database leaks.

Dependencies

[dependencies]
argon2 = "0.6.0-rc.8" # Password hashing
hmac = "0.13.0"        # Message authentication
sha2 = "0.11.0"        # SHA-256 hash function
subtle = "2.6.1"       # Constant-time comparisons
rand_core = "0.9"      # Cryptographically secure RNG

Pattern 1: Fast Lookup Index (HMAC-SHA256)

Argon2id is too slow to use for searching a database (e.g., SELECT * FROM users WHERE hash = ?). Instead, create a fast, deterministic lookup index using an environment-level pepper. Truncate the HMAC result to 16 bytes to store it optimally as a standard BYTEA / BLOB (UUID size).

use hmac::{Hmac, Mac, KeyInit};
use sha2::Sha256;

/// Create a fast, indexed lookup value (16 bytes)
pub fn create_lookup_index(pepper: &[u8], space_id: &str) -> [u8; 16] {
    let mut mac = Hmac::<Sha256>::new_from_slice(pepper)
        .expect("HMAC can take key of any size");

    mac.update(space_id.as_bytes());
    let result = mac.finalize().into_bytes();

    // Truncate to 16 bytes for optimal DB storage
    let mut index =[0u8; 16];
    index.copy_from_slice(&result[..16]);
    index
}

// SQLx Usage: query!("SELECT hash FROM space WHERE idx = $1", &index)

Pattern 2: Secure Hashing (Argon2id)

Once the row is found via the fast index, verify the actual secret using Argon2id. Configure memory and time costs carefully — API endpoints should use lower memory costs (e.g., 15MB) to prevent memory-exhaustion Denial of Service (DoS) attacks from concurrent requests.

use argon2::{
    password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, SaltString},
    Argon2, Params,
};

/// Hash the Space ID for storage
pub fn hash_secret(secret: &str) -> String {
    use rand_core::OsRng;
    let salt = SaltString::generate(&mut OsRng);

    // For API limits: 15MB memory, 2 iterations, 1 parallelism
    let params = Params::new(15360, 2, 1, None).unwrap();
    let argon2 = Argon2::new(argon2::Algorithm::Argon2id, argon2::Version::V0x13, params);

    argon2.hash_password(secret.as_bytes())
        .unwrap()
        .to_string() // Store this string in the database
}

/// Verify the submitted secret against the hash
pub fn verify_secret(secret: &str, stored_hash: &str) -> bool {
    let parsed_hash = match PasswordHash::new(stored_hash) {
        Ok(hash) => hash,
        Err(_) => return false,
    };

    Argon2::default().verify_password(secret.as_bytes(), stored_hash).is_ok()
}

Pattern 3: Constant-Time Comparisons

Whenever comparing sensitive bytes directly (like verifying Webhook payloads or custom signatures), ALWAYS use the subtle crate to prevent timing attacks. The standard == operator short-circuits and leaks timing data.

use subtle::ConstantTimeEq;

pub fn is_equal_secure(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false; // Length leaks are generally accepted, byte content leaks are not
    }

    // ct_eq() returns a Choice. Unwrap it to a boolean securely.
    a.ct_eq(b).unwrap_u8() == 1
}

Distributed SSE (Redis Pub/Sub to Axum)

In a microservice architecture, the container executing a background task (e.g., an Apalis Worker processing an image) is almost never the same container holding the user's open HTTP request.

To stream progress events to the user, the Worker publishes messages to Redis, and the API subscribes to Redis and translates those messages into HTTP Server-Sent Events (SSE).

Dependencies

[dependencies]
redis = { version = "1.2.1", features = ["tokio-comp"] }
axum = "0.8.9"
tokio-stream = "0.1.18"
futures-util = "0.3.32"
async-stream = "0.3.6"

The Pattern

1. Worker (Publisher)

When the background worker completes a step, it publishes a JSON payload to a Redis channel specific to the Task ID. Multiplexed Redis connections are perfectly safe to share for PUBLISH operations.

use redis::AsyncCommands;

async fn worker_process(redis_url: &str, task_id: &str) {
    let client = redis::Client::open(redis_url).unwrap();
    let mut conn = client.get_multiplexed_tokio_connection().await.unwrap();

    let channel = format!("evt.task.{}", task_id);
    let payload = r#"{"progress": 50, "status": "running"}"#;

    // Publish the event to the channel
    let _: () = conn.publish(channel, payload).await.unwrap();
}

2. API (Subscriber & SSE Stream)

The Axum handler MUST acquire a dedicated Pub/Sub connection (a shared multiplexed connection will panic if you attempt to call SUBSCRIBE on it). It wraps the Redis stream into an axum::response::sse::Event and yields it.

use axum::{extract::Path, response::sse::{Event, Sse, KeepAlive}};
use futures_util::StreamExt;
use std::convert::Infallible;
use std::time::Duration;

async fn task_events_handler(Path(task_id): Path<String>)
    -> Sse<impl futures_util::stream::Stream<Item = Result<Event, Infallible>>>
{
    let client = redis::Client::open("redis://127.0.0.1/").unwrap();

    let stream = async_stream::stream! {
        // 1. Get a dedicated connection for Pub/Sub
        let mut pubsub = client.get_async_pubsub().await.unwrap();

        // 2. Subscribe to the specific task's channel
        let channel = format!("evt.task.{}", task_id);
        pubsub.subscribe(&channel).await.unwrap();

        let mut msg_stream = pubsub.on_message();

        // 3. Yield events as they arrive from Redis
        while let Some(msg) = msg_stream.next().await {
            let payload: String = msg.get_payload().unwrap();

            yield Ok(Event::default().data(payload.clone()));

            // Optional: Break the stream if the task reports completion
            if payload.contains(r#""status":"done""#) {
                break;
            }
        }
    };

    // Keep the HTTP connection alive with pings every 15 seconds
    Sse::new(stream).keep_alive(
        KeepAlive::new()
            .interval(Duration::from_secs(15))
            .text("ping")
    )
}

S3 Bucket Policies - Presigned PUT Size Limits

Presigned PUT URLs do not carry content-length-range conditions. Enforce upload sizes in the object store policy for the exact prefixes that presigned PUTs may write to, and keep the URL signing code focused on method, key, expiry, and required headers.

Dependencies

[dependencies]
aws-config = "1.8.16"
aws-sdk-s3 = "1.131.0"
serde_json = "1.0.149"

Pattern

Apply or update a bucket policy with explicit prefix resources and numeric s3:content-length denies. This avoids trying to embed POST-only conditions into a PUT presign.

use aws_sdk_s3::Client;
use serde_json::json;

pub async fn apply_upload_size_policy(
    s3: &Client,
    bucket: &str,
    prefix: &str,
    min_bytes: u64,
    max_bytes: u64,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let normalized_prefix = prefix.trim_matches('/');
    let resource = format!("arn:aws:s3:::{bucket}/{normalized_prefix}/*");

    let policy = json!({
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "DenyUploadsBelowMinimumSize",
                "Effect": "Deny",
                "Principal": "*",
                "Action": "s3:PutObject",
                "Resource": resource,
                "Condition": {
                    "NumericLessThan": {
                        "s3:content-length": min_bytes
                    }
                }
            },
            {
                "Sid": "DenyUploadsAboveMaximumSize",
                "Effect": "Deny",
                "Principal": "*",
                "Action": "s3:PutObject",
                "Resource": resource,
                "Condition": {
                    "NumericGreaterThan": {
                        "s3:content-length": max_bytes
                    }
                }
            }
        ]
    });

    s3.put_bucket_policy()
        .bucket(bucket)
        .policy(policy.to_string())
        .send()
        .await?;

    Ok(())
}

Presigned PUT Usage

use aws_sdk_s3::presigning::PresigningConfig;
use std::time::Duration;

pub async fn presign_upload(
    s3: &aws_sdk_s3::Client,
    bucket: &str,
    key: &str,
    content_type: &str,
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
    let request = s3
        .put_object()
        .bucket(bucket)
        .key(key)
        .content_type(content_type)
        .presigned(PresigningConfig::expires_in(Duration::from_secs(900))?)
        .await?;

    Ok(request.uri().to_string())
}

Rules

• Use bucket policy for PUT size enforcement; content-length-range is a presigned POST policy condition, not a PUT URL parameter.
• Scope each policy statement to the upload prefix, for example incoming/tasks/*, not the whole bucket.
• Merge these statements into any existing bucket policy; put_bucket_policy replaces the full policy document.
• Keep application-side validation too; bucket policy is the final server-side guardrail.
• Confirm the deployed S3-compatible server supports s3:content-length policy conditions before relying on this for enforcement.

SSRF Protection - External RPC Clients

External RPC clients must reject loopback, link-local, private, and otherwise non-public IP targets unless the target is explicitly allow-listed. Do this at the connector/resolver boundary so redirects and DNS answers are checked immediately before TCP connect.

Dependencies

[dependencies]
reqwest = { version = "0.13.3", default-features = false, features = ["rustls", "json"] }
tokio = { version = "1.52.1", features = ["net"] }
ipnet = "2.12.0"

For lower-level Hyper clients, current Hyper uses hyper-util:

[dependencies]
hyper-util = { version = "0.1.20", features = ["client", "client-legacy", "tokio"] }
tower = "0.5.3"
ipnet = "2.12.0"

Pattern 1: Reqwest DNS Resolver Guard

Use reqwest::ClientBuilder::dns_resolver for current reqwest. Do not hand-roll URL string checks; they miss redirects, alternate DNS answers, IPv6, and direct IP hosts.

use reqwest::dns::{Addrs, Name, Resolve, Resolving};
use ipnet::IpNet;
use std::{collections::HashSet, net::{IpAddr, SocketAddr}, sync::LazyLock};

static BLOCKED_NETS: LazyLock<Vec<IpNet>> = LazyLock::new(|| {
    [
        "0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8",
        "169.254.0.0/16", "172.16.0.0/12", "192.0.0.0/24", "192.0.2.0/24",
        "192.168.0.0/16", "198.18.0.0/15", "198.51.100.0/24", "203.0.113.0/24",
        "224.0.0.0/4", "240.0.0.0/4", "::/128", "::1/128", "::ffff:0:0/96",
        "64:ff9b:1::/48", "100::/64", "2001:db8::/32", "fc00::/7", "fe80::/10",
        "ff00::/8",
    ]
    .into_iter()
    .map(|cidr| cidr.parse().expect("valid blocked CIDR"))
    .collect()
});

#[derive(Clone, Default)]
pub struct SsrfSafeResolver {
    allow_list: HashSet<IpAddr>,
}

impl SsrfSafeResolver {
    pub fn new(allow_list: impl IntoIterator<Item = IpAddr>) -> Self {
        Self { allow_list: allow_list.into_iter().collect() }
    }
}

impl Resolve for SsrfSafeResolver {
    fn resolve(&self, name: Name) -> Resolving {
        let host = name.as_str().to_owned();
        let allow_list = self.allow_list.clone();

        Box::pin(async move {
            let addrs: Vec<SocketAddr> = tokio::net::lookup_host((host.as_str(), 0))
                .await?
                .filter(|addr| is_allowed_external_ip(addr.ip(), &allow_list))
                .collect();

            if addrs.is_empty() {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::PermissionDenied,
                    "blocked by SSRF IP policy",
                ).into());
            }

            Ok(Box::new(addrs.into_iter()) as Addrs)
        })
    }
}

fn is_allowed_external_ip(ip: IpAddr, allow_list: &HashSet<IpAddr>) -> bool {
    if allow_list.contains(&ip) {
        return true;
    }

    !BLOCKED_NETS.iter().any(|net| net.contains(&ip))
}

pub fn external_rpc_client(allow_list: HashSet<IpAddr>) -> reqwest::Result<reqwest::Client> {
    reqwest::Client::builder()
        .https_only(true)
        .no_proxy()
        .dns_resolver(SsrfSafeResolver::new(allow_list))
        .build()
}

Pattern 2: Hyper HttpConnector Guard

If the codebase uses Hyper directly, wrap the resolver passed to hyper_util::client::legacy::connect::HttpConnector::new_with_resolver. The old hyper::client::HttpConnector path is stale for current Hyper. This snippet uses the same is_allowed_external_ip helper shown above.

use hyper_util::client::legacy::connect::{dns::Name, dns::GaiResolver, HttpConnector};
use std::{collections::HashSet, future::Future, net::{IpAddr, SocketAddr}, task::{Context, Poll}};
use tower::Service;

#[derive(Clone)]
pub struct FilteringResolver {
    inner: GaiResolver,
    allow_list: HashSet<IpAddr>,
}

impl Service<Name> for FilteringResolver {
    type Response = std::vec::IntoIter<SocketAddr>;
    type Error = Box<dyn std::error::Error + Send + Sync>;
    type Future = std::pin::Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx).map_err(Into::into)
    }

    fn call(&mut self, name: Name) -> Self::Future {
        let mut inner = self.inner.clone();
        let allow_list = self.allow_list.clone();

        Box::pin(async move {
            let addrs = inner.call(name).await?;
            let filtered: Vec<_> = addrs
                .filter(|addr| is_allowed_external_ip(addr.ip(), &allow_list))
                .collect();

            if filtered.is_empty() {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::PermissionDenied,
                    "blocked by SSRF IP policy",
                ).into());
            }

            Ok(filtered.into_iter())
        })
    }
}

pub fn guarded_http_connector(allow_list: HashSet<IpAddr>) -> HttpConnector<FilteringResolver> {
    let resolver = FilteringResolver { inner: GaiResolver::new(), allow_list };
    let mut connector = HttpConnector::new_with_resolver(resolver);
    connector.enforce_http(false); // allow HTTPS schemes to pass to the TLS connector layer
    connector
}

Rules

• Always filter resolved SocketAddr values, not just request host strings.
• Disable ambient proxies with no_proxy() for external RPC clients unless proxy egress is separately controlled.
• Keep allow-lists explicit and narrow; never allow-list private ranges wholesale.
• Re-run this check on every DNS resolution so DNS rebinding cannot reuse a stale approval.

ULID & SQLx - 16-Byte Task IDs

Task IDs use ULID for sortable, random identifiers. Do not store ULIDs as strings unless the schema explicitly requires human-readable IDs; store the 16 raw bytes.

Dependencies

[dependencies]
ulid = { version = "1.2.1", features = ["uuid"] }
uuid = "1.23.1"
sqlx = { version = "0.9.0-alpha.1", features = ["postgres", "runtime-tokio", "uuid"] }

Pattern

Use ulid::Ulid, convert to uuid::Uuid with into(), then bind the 16 bytes for a BYTEA column. If the database column type is UUID instead of BYTEA, bind the Uuid directly and let SQLx's uuid feature encode it.

use sqlx::Row;
use ulid::Ulid;
use uuid::Uuid;

pub struct TaskRow {
    pub id: Uuid,
    pub status: String,
}

pub fn new_task_id() -> Uuid {
    let ulid = Ulid::new();
    ulid.into()
}

pub async fn insert_task(pool: &sqlx::PgPool) -> Result<Uuid, sqlx::Error> {
    let task_id = new_task_id();

    sqlx::query(r#"INSERT INTO tasks (id) VALUES ($1)"#)
        .bind(task_id.as_bytes().as_slice()) // BYTEA, exactly 16 bytes
        .execute(pool)
        .await?;

    Ok(task_id)
}

pub async fn load_task(pool: &sqlx::PgPool, task_id: Uuid) -> Result<Option<TaskRow>, sqlx::Error> {
    let Some(row) = sqlx::query(r#"SELECT id, status FROM tasks WHERE id = $1"#)
        .bind(task_id.as_bytes().as_slice())
        .fetch_optional(pool)
        .await?
    else {
        return Ok(None);
    };

    let id_bytes: Vec<u8> = row.try_get("id")?;
    let id = Uuid::from_slice(&id_bytes).map_err(|error| sqlx::Error::Decode(Box::new(error)))?;
    let status = row.try_get("status")?;

    Ok(Some(TaskRow { id, status }))
}

For a PostgreSQL UUID column, bind the Uuid value instead:

sqlx::query(r#"INSERT INTO tasks (id) VALUES ($1)"#)
    .bind(task_id)
    .execute(pool)
    .await?;

Rules

• Generate IDs with Ulid::new(); do not use random strings or database sequences for task IDs.
• Convert with let id: uuid::Uuid = ulid.into();; do not parse through a string.
• For BYTEA, bind id.as_bytes().as_slice() and keep a CHECK (octet_length(id) = 16) constraint.
• For UUID, enable SQLx's uuid feature and bind Uuid directly.
• Use sqlx::query in reusable snippets; query! requires a database connection or prepared offline metadata at compile time.