Recipes

These are the patterns that show up in the canonical examples. Each language has one application example instead of parallel copies of every app. The code here shows all languages where comparison helps, trimmed to the Forge calls so the shape is easy to lift into your own handlers.

Production recipe index

The deployment recipes live beside their operational guarantees: migration jobs, managed workers and graceful shutdown, transactional outbox delivery, liveness and readiness endpoints, and Prometheus export. Queue producers should supply a deterministic job ID for a business operation and make handlers safe to repeat; use the outbox when enqueue must agree with an application-owned SQL transaction. For blob uploads, use the Forge proxy signature when the server must enforce a hard body limit or application-specific authorization, and native S3 presigning when direct provider transfer is acceptable.

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. Password reset, email verification, and magic links are covered by one-time tokens (below); Forge mints and checks the token, your app sends the email. OAuth and 2FA are their own product surface — reach for a dedicated auth provider once you need those 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?;

Password reset is a one-time token plus an email your app sends. Mint a short-lived token scoped to "password-reset", put it in the link, and consume it when the user comes back; the first consume wins and every retry after that returns nothing. Respond identically whether or not the email exists so the endpoint doesn’t confirm accounts.

Node

// request handler: always answer 200; only mint + email when the account exists
const found = await forge.kvGet(`user:email:${email}`);
if (found) {
  const user = JSON.parse(found);
  const token = await forge.createToken(user.id, "password-reset", 900); // 15 min
  await forge.queue("emails").enqueue({ to: email, resetLink: linkFor(token) });
}

// reset handler: consume exactly once, then re-hash and revoke every session
const consumed = await forge.consumeToken(token, "password-reset");
if (!consumed) return c.json({ error: "link expired" }, 400);
await updatePasswordHash(consumed.userId, await forge.hashPassword(newPassword));
await forge.revokeAllSessions(consumed.userId);

Python

# request handler: always answer 200; only mint + email when the account exists
found = await forge.kv_get(f"user:email:{email}")
if found:
    user = json.loads(found)
    token = await forge.create_token(user["id"], "password-reset", 900)  # 15 min
    await forge.queue("emails").enqueue({"to": email, "reset_link": link_for(token)})

# reset handler: consume exactly once, then re-hash and revoke every session
consumed = await forge.consume_token(token, "password-reset")
if not consumed:
    raise HTTPException(400, "link expired")
await update_password_hash(consumed.user_id, await forge.hash_password(new_password))
await forge.revoke_all_sessions(consumed.user_id)

Rust

// request handler: always answer 200; only mint + email when the account exists
if let Some(bytes) = forge.kv().get(&format!("user:email:{email}")).await? {
    let user: User = serde_json::from_slice(&bytes)?;
    let token = forge.auth()
        .create_token(&user.id, "password-reset", Duration::from_secs(900)) // 15 min
        .await?;
    forge.queue().enqueue("emails", email_payload(&email, token.as_str()), EnqueueOpts::default()).await?;
}

// reset handler: consume exactly once, then re-hash and revoke every session
let Some(consumed) = forge.auth().consume_token_with_payload(token, "password-reset").await? else {
    return link_expired();
};
update_password_hash(&consumed.user_id, forge.auth().hash_password(&new_password).await?).await?;
forge.auth().revoke_all_sessions(&consumed.user_id).await?;

The same shape does email verification ("email-verify" purpose, consume on the confirmation link) and magic-link login (consume, then create_session for the returned user id). The purpose string is what stops a reset token from being replayed against the verify endpoint.

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?; }
    }
}

Publish backend events through an application transport

The worker that drains the clicks queue publishes a bounded invalidation hint after the durable count changes. An application-owned SSE route subscribes to that topic and forwards each hint to a connected client, then drops the subscription when the client disconnects. The client refetches the count; it never treats the hint as authoritative state. WebSockets or another transport can use the same backend pattern.

Node

// in the worker handler, after committing the new count
await forge.pubsubPublish(`clicks:${slug}`, encodeInvalidationEvent({
  schema_version: 1, tags: [`links/${slug}`],
  query_keys: [["link", slug]], revision: String(total),
}).toString("utf8"));

// in the SSE route (Hono)
return streamSSE(c, async (stream) => {
  const events = await forge.topic<unknown>(`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 committing the new count
hint = encode_invalidation_event({
    "schema_version": 1, "tags": [f"links/{slug}"],
    "query_keys": [["link", slug]], "revision": str(total),
})
await forge.pubsub_publish(f"clicks:{slug}", hint.decode())

# 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 committing the new count
let hint = InvalidationEvent::new(
    vec![format!("links/{slug}")],
    vec![vec![json!("link"), json!(slug)]],
    Some(total.to_string()),
)?;
forge.pubsub().publish(&format!("clicks:{slug}"), Bytes::from(hint.encode()?)).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. The durable number lives in the key/value counter; the topic only tells open consumers to refresh. See Event delivery for transport limits and client ownership.

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)).url;

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)).url

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?.url;

Presigned links need a signing_secret set in forge.toml. For uploads, hand back the .url from the structured proxy presign result; the application proxy verifies its method, key, expiry, namespace, and max_bytes before accepting the body.

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.