Recipes

These are the patterns that show up in the example apps (a todo app, a URL shortener, and a chat app, each built three times on the Rust, Node, and Python bindings). The code is trimmed to the Forge calls so the shape is easy to lift into your own handlers.

Password auth

Email and password auth is about thirty lines of Forge calls per handler: hash on the way in, verify on the way out, and issue sessions you can revoke. There’s no separate users table to wire up first: kv stores the user record and auth owns the hashing and the sessions.

Know the boundary

This is the “email and password, done right” baseline. OAuth, magic links, 2FA, and password reset are their own product surface; reach for a dedicated auth provider once you need them rather than growing this into one.

Sign-up does four things: throttle by email, hash the password, claim the email atomically, and open a session. Writing the email key with “set if not present” makes it its own uniqueness check, so there’s no read-then-write race.

Node

// throttle by email, fail open so a limiter outage doesn't lock people out
const limit = await forge.rateLimitCheck("auth", email, 20, 60, true);
if (!limit.allowed) return c.json({ error: "slow down" }, 429);

const user = { id: randomUUID(), email, passwordHash: await forge.hashPassword(password) };
const body = JSON.stringify(user);

// the email key is its own uniqueness guard: false means already taken
const created = await forge.kvSet(`user:email:${email}`, body, null, true);
if (!created) return c.json({ error: "email already registered" }, 409);
await forge.kvSet(`user:id:${user.id}`, body);

const token = await forge.createSession(user.id, 86_400, 30 * 86_400);

Python

# throttle by email, fail open so a limiter outage doesn't lock people out
limit = await forge.rate_limit_check("auth", email, 20, 60.0, True)
if not limit.allowed:
    raise HTTPException(429, "slow down")

user = {"id": uuid4().hex, "email": email,
        "password_hash": await forge.hash_password(password)}
body = json.dumps(user)

# the email key is its own uniqueness guard: False means already taken
created = await forge.kv_set(f"user:email:{email}", body, None, True)
if not created:
    raise HTTPException(409, "email already registered")
await forge.kv_set(f"user:id:{user['id']}", body)

token = await forge.create_session(user["id"], 86_400, 30 * 86_400)

Rust

use std::time::Duration;
use forgelib::{Bytes, Limit, FailMode, SetOpts, SetMode, SessionOpts};

// throttle by email, fail open so a limiter outage doesn't lock people out
let limit = forge.ratelimit()
    .check_with("auth", &email, Limit::per_duration(20, Duration::from_secs(60)), FailMode::Open)
    .await?;
if !limit.allowed { return too_many_requests(); }

let user = User { id: new_id(), email: email.clone(),
    password_hash: forge.auth().hash_password(&password).await?.as_str().to_string() };
let body = Bytes::from(serde_json::to_vec(&user)?);

// the email key is its own uniqueness guard: false means already taken
let created = forge.kv()
    .set(&format!("user:email:{email}"), body.clone(),
         SetOpts::new().with_mode(SetMode::IfNotExists)).await?;
if !created { return conflict("email already registered"); }
forge.kv().set(&format!("user:id:{}", user.id), body, SetOpts::new()).await?;

let token = forge.auth().create_session(&user.id, SessionOpts::new()
    .with_idle_timeout(Duration::from_secs(86_400))
    .with_absolute_timeout(Duration::from_secs(30 * 86_400))).await?;

Logging in runs the same throttle, verifies the password against the stored hash, then opens a fresh session. Keep the rate-limit check ahead of verifyPassword so a stuffing run can’t burn CPU on hashing.

Node

const limit = await forge.rateLimitCheck("auth", email, 20, 60, true);
if (!limit.allowed) return c.json({ error: "slow down" }, 429);

const stored = await forge.kvGet(`user:email:${email}`);
if (!stored) return c.json({ error: "invalid email or password" }, 401);
const user = JSON.parse(stored);

// the throttle runs before verifyPassword, so credential stuffing can't burn CPU on hashing
if (!(await forge.verifyPassword(password, user.passwordHash)))
  return c.json({ error: "invalid email or password" }, 401);

const token = await forge.createSession(user.id, 86_400, 30 * 86_400);

Python

limit = await forge.rate_limit_check("auth", email, 20, 60.0, True)
if not limit.allowed:
    raise HTTPException(429, "slow down")

stored = await forge.kv_get(f"user:email:{email}")
if not stored:
    raise HTTPException(401, "invalid email or password")
user = json.loads(stored)

# the throttle runs before verify_password, so credential stuffing can't burn CPU on hashing
if not await forge.verify_password(password, user["password_hash"]):
    raise HTTPException(401, "invalid email or password")

token = await forge.create_session(user["id"], 86_400, 30 * 86_400)

Rust

use forgelib::PhcString;

let limit = forge.ratelimit()
    .check_with("auth", &email, Limit::per_duration(20, Duration::from_secs(60)), FailMode::Open)
    .await?;
if !limit.allowed { return too_many_requests(); }

let Some(bytes) = forge.kv().get(&format!("user:email:{email}")).await? else {
    return invalid_login();
};
let user: User = serde_json::from_slice(&bytes)?;

// the throttle runs before verify_password, so credential stuffing can't burn CPU on hashing
if !forge.auth().verify_password(&password, &PhcString::new(user.password_hash.clone())).await? {
    return invalid_login();
}

let token = forge.auth().create_session(&user.id, SessionOpts::new()
    .with_idle_timeout(Duration::from_secs(86_400))
    .with_absolute_timeout(Duration::from_secs(30 * 86_400))).await?;

Every protected route resolves the caller’s token to a user (or 401s), and logout revokes the session so the token stops working immediately.

Node

// middleware: resolve the bearer token (or session cookie) to a user id, else 401
const userId = await forge.validateSession(bearerToken(c.req.header("authorization")));
if (!userId) return c.json({ error: "authentication required" }, 401);

// logout: revoke the session so the token stops resolving
await forge.revokeSession(bearerToken(c.req.header("authorization")));

Python

# middleware: resolve the bearer token (or session cookie) to a user id, else 401
user_id = await forge.validate_session(bearer)
if not user_id:
    raise HTTPException(401, "authentication required")

# logout: revoke the session so the token stops resolving
await forge.revoke_session(bearer)

Rust

// middleware: resolve the bearer token (or session cookie) to a session, else 401
let Some(session) = forge.auth().validate_session(bearer).await? else {
    return unauthorized();
};
let user_id = session.user_id;

// logout: revoke the session so the token stops resolving
forge.auth().revoke_session(bearer).await?;

Offload slow work to a queue

Keep the request fast by doing only the cheap part inline and handing the rest to a queue. The URL shortener bumps a click counter on the redirect and queues everything else, so the user gets their redirect immediately.

Node

// in the redirect handler: cheap now, the rest later
await forge.kvIncr(`clicks:${slug}`, 1);
await forge.queue<{ slug: string }>("clicks").enqueue({ slug }, { maxAttempts: 3 });

Python

# in the redirect handler: cheap now, the rest later
await forge.kv_incr(f"clicks:{slug}", 1)
await forge.queue("clicks").enqueue({"slug": slug}, max_attempts=3)

Rust

use forgelib::{Bytes, EnqueueOpts};
use serde_json::json;

// in the redirect handler: cheap now, the rest later
forge.kv().incr(&format!("clicks:{slug}"), 1).await?;
forge.queue().enqueue("clicks", Bytes::from(json!({ "slug": slug }).to_string()),
    EnqueueOpts::new().with_max_attempts(3)).await?;

A worker leases jobs, does the work, and acks on success or nacks on failure so the job is retried. Run one of these per queue (see Operations for a managed version).

Node

await forge.worker<{ slug: string }>("clicks", async (job) => {
  await handle(forge, job.payload);
});

Python

await forge.worker("clicks", lambda job: handle(forge, job.payload))

Rust

use forgelib::{DequeueOpts, NackOpts};

loop {
    let Some(job) = forge.queue().dequeue("clicks", DequeueOpts::new()).await? else { continue };
    match handle(&forge, &job).await {
        Ok(()) => { forge.queue().ack(&job).await?; }
        Err(_) => { forge.queue().nack(&job, NackOpts::default()).await?; }
    }
}

Push live updates to the browser

The worker that drains the clicks queue publishes the new total on a topic. An SSE route subscribes to that topic and forwards each message to the connected client, then drops the subscription when the client disconnects.

Node

// in the worker handler, after reading the new count
await forge.topic<{ slug: string; clicks: number }>(`clicks:${slug}`)
  .publish({ slug, clicks: total });

// in the SSE route (Hono)
return streamSSE(c, async (stream) => {
  const events = await forge.topic<{ slug: string; clicks: number }>(`clicks:${slug}`)
    .subscribe();
  stream.onAbort(() => { void events.return?.(); });
  for await (const event of events) {
    await stream.writeSSE({ data: JSON.stringify(event) });
  }
});

Python

# in the worker handler, after reading the new count
await forge.topic(f"clicks:{slug}").publish({"slug": slug, "clicks": total})

# in the SSE route (FastAPI)
async def live(slug: str):
    topic = forge.topic(f"clicks:{slug}")
    async def events():
        async for event in topic.subscribe():
            yield f"data: {json.dumps(event)}\n\n"
    return StreamingResponse(events(), media_type="text/event-stream")

Rust

use futures_util::StreamExt;

// in the worker handler, after reading the new count
forge.pubsub().publish(&format!("clicks:{slug}"),
    Bytes::from(json!({ "slug": slug, "clicks": total }).to_string())).await?;

// in the SSE route
let mut sub = forge.pubsub().subscribe(&format!("clicks:{slug}")).await?;
while let Some(Ok(payload)) = sub.next().await {
    // write `payload` to the response as an SSE `data:` line
}

Note

Pub/sub only reaches connected clients, which is exactly right for a live counter. The durable number lives in the key/value counter; the topic just nudges open dashboards to refresh.

Store and serve a file

Put generated or uploaded bytes under a key with a content type, read them back to serve them, or hand the client a presigned link so it can fetch the file directly.

Node

await forge.blobPut(`qr/${slug}.svg`, svg, "image/svg+xml");

const stored = await forge.blobGet(`qr/${slug}.svg`);   // string | null

const url = await forge.blobPresignDownload(`qr/${slug}.svg`, 3600);

Python

await forge.blob_put(f"qr/{slug}.svg", svg, "image/svg+xml")  # svg is bytes

stored = await forge.blob_get(f"qr/{slug}.svg")   # bytes | None

url = await forge.blob_presign_download(f"qr/{slug}.svg", 3600)

Rust

use std::time::Duration;
use forgelib::{Bytes, PutOpts};

forge.blob().put(&format!("qr/{slug}.svg"), Bytes::from(svg),
    PutOpts::new().with_content_type("image/svg+xml")).await?;

let svg = forge.blob().get(&format!("qr/{slug}.svg")).await?;   // Option<Bytes>

let url = forge.blob()
    .presign_download(&format!("qr/{slug}.svg"), Duration::from_secs(3600)).await?;

Presigned links need a signing_secret set in forge.toml. For uploads, hand back a presigned upload URL capped at a max_bytes you choose and let the client PUT straight to it.

Run work later

Schedule a one-shot for delayed work, like expiring a link at its TTL, or a cron for recurring jobs. Both land on a normal queue, so the same workers drain them.

Node

// expire this link in an hour
await forge.scheduleAt(Date.now() + 3600_000, "expire", JSON.stringify({ slug }));

// a daily report
await forge.scheduleCron("nightly-report", "0 0 * * *", "reports", "{}");

Python

# expire this link in an hour
await forge.schedule_at(now_ms + 3_600_000, "expire", json.dumps({"slug": slug}))

# a daily report
await forge.schedule_cron("nightly-report", "0 0 * * *", "reports", "{}")

Rust

use std::time::{Duration, SystemTime};
use forgelib::ScheduleOpts;

// expire this link in an hour
let when = SystemTime::now() + Duration::from_secs(3600);
forge.schedule().at(when, "expire", payload.into(), ScheduleOpts::new()).await?;

// a daily report
forge.schedule().cron("nightly-report", "0 0 * * *", "reports",
    b"{}".to_vec().into(), ScheduleOpts::new()).await?;

Schedules only fire when the scheduler is ticked. The same loop that ticks it is a good place to run Forge’s maintenance sweep. See Operations for the loop these apps run on every replica.