Full example

A walkthrough of examples/cloudflare-worker — D1, Queues, R2, and WebSocket together.

The repository ships a complete example at examples/cloudflare-worker: a task API with D1, asynchronous Queue calls, R2 object streaming, a WebSocket echo endpoint, and end-to-end unit + integration tests. This page walks through the main pieces.

Project layout

examples/cloudflare-worker/
  src/
    lib.rs          # pub mod app; assets; demo (private); boards; orgs; tasks; users (public)
    entry.rs         # wasm-only glue: #[event(fetch)] and #[event(queue)]
    app.rs           # builds the Rocket, body limits, .manage(env)
    tasks/           # the D1-backed task API (the example's focus)
    assets/          # R2 upload/download routes
    demo/            # /, /echo, /stream, /ws/echo — streaming and WebSocket proofs
    boards/ orgs/ users/  # extra Nebula entities, for relationships
  migrations/
    0001_init.sql
  tests/
    integration.sh
    perf.sh
  wrangler.jsonc
  package.json

Data model

The tasks table (from migrations/0001_init.sql):

CREATE TABLE tasks (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  title TEXT NOT NULL,
  done INTEGER NOT NULL DEFAULT 0,
  created_at TEXT NOT NULL DEFAULT (datetime('now'))
);

A companion task_events table is written only by the queue consumer, never by the HTTP routes — proving the asynchronous round trip.

TaskRow (the Nebula entity) maps the raw D1 row (with done: i32) and converts via From<TaskRow> for Task into the public Task (with done: bool) served as JSON — that's the "D1 has no boolean type" quirk mentioned in Nebula ORM.

Routes exposed

Route	Description
`GET /`	greeting text
`POST /echo`	streamed body echo
`GET /stream`	3 chunks with real 400ms pauses (streaming proof)
`GET /ws/echo`	WebSocket echo
`GET /tasks`, `GET /tasks/<id>`	list/fetch tasks
`POST /tasks`	create a task (publishes an event to the queue)
`POST /tasks/<id>/complete`	complete a task (publishes an event to the queue)
`PUT /assets/<key..>`, `GET /assets/<key..>`	upload/download via R2

D1: a CRUD route

#[get("/tasks/<id>")]
pub async fn get_task(id: i32, db: D1<DB>) -> ApiResult<Json<Task>> {
    let row = TaskRow::select()
        .where_(TaskRow::ID.eq(id))
        .to_statement()
        .fetch_optional_d1::<TaskRow>(&db)
        .await
        .map_err(ApiError::from)?
        .ok_or(ApiError::NotFound)?;

    Ok(Json(Task::from(row)))
}

No route calls env.d1(...) manually — the D1<DB> guard already hands over the connection, and Nebula's query builder assembles the SQL.

Queue: producer and consumer

Producer side, inside POST /tasks:

let row = TaskRow::insert()
    .set(TaskRow::TITLE, title)
    .returning(TASK_COLUMNS.iter().copied())
    .to_statement()
    .fetch_one_d1::<TaskRow>(&db)
    .await
    .map_err(ApiError::from)?;

let task: Task = row.into();
publish_task_event(&queue, task.id, TaskEventKind::Created).await?;

Consumer side, in src/entry.rs (uses the worker crate directly, without Nebula, because it runs outside the context of a Rocket request):

#[event(queue)]
pub async fn queue(batch: MessageBatch<TaskEvent>, env: Env, _ctx: Context) -> Result<()> {
    let db = env.d1("DB")?;
    for message in batch.messages()? {
        let event = message.into_body();
        db.prepare(RECORD_TASK_EVENT_QUERY)
            .bind(&[JsValue::from(event.task_id), JsValue::from(event.kind.as_str())])?
            .run()
            .await?;
    }
    batch.ack_all();
    Ok(())
}

R2: upload and download

#[put("/assets/<key..>", data = "<body>")]
pub async fn put_asset(key: PathBuf, body: Capped<Vec<u8>>, bucket: R2Bucket<Assets>)
    -> Result<Status, Status>
{
    if !body.is_complete() {
        return Err(Status::PayloadTooLarge);
    }
    bucket.put(asset_key(key), body.value).execute().await
        .map_err(|_| Status::InternalServerError)?;
    Ok(Status::Created)
}

#[get("/assets/<key..>")]
pub async fn get_asset(key: PathBuf, bucket: R2Bucket<Assets>) -> Option<R2Object> {
    R2Object::get(&bucket, asset_key(key)).await.ok().flatten()
}

WebSocket: echo

#[get("/ws/echo")]
pub async fn websocket_echo(ws: WebSocketUpgrade) -> WebSocketResponse {
    ws.accept(|socket| async move {
        let mut events = socket.events()?;
        while let Some(event) = events.next().await {
            match event? {
                WebsocketEvent::Message(message) => {
                    if let Some(text) = message.text() {
                        socket.send_with_str(text)?;
                    } else if let Some(bytes) = message.bytes() {
                        socket.send_with_bytes(bytes)?;
                    }
                }
                WebsocketEvent::Close(_) => break,
            }
        }
        Ok(())
    })
}

Configuring `wrangler.jsonc`

{
  "main": "build/worker/shim.mjs",
  "compatibility_flags": ["nodejs_compat"],
  "build": { "command": "... worker-build --release" },
  "d1_databases": [{
    "binding": "DB",
    "database_name": "comet-cloudflare-worker-example",
    "database_id": "00000000-0000-0000-0000-000000000000",
    "migrations_dir": "migrations"
  }],
  "r2_buckets": [{ "binding": "ASSETS", "bucket_name": "comet-cloudflare-worker-example-assets" }],
  "queues": {
    "producers": [{ "binding": "TASK_EVENTS", "queue": "task-events" }],
    "consumers": [{ "queue": "task-events", "max_batch_size": 10, "max_batch_timeout": 5 }]
  }
}

The binding names (DB, ASSETS, TASK_EVENTS) need to match the BindingName::NAME constants defined in the Rust code:

pub struct DB;
impl BindingName for DB { const NAME: &'static str = "DB"; }

pub struct Assets;
impl BindingName for Assets { const NAME: &'static str = "ASSETS"; }

pub struct TaskEvents;
impl BindingName for TaskEvents { const NAME: &'static str = "TASK_EVENTS"; }

Running the example

npx wrangler d1 create comet-cloudflare-worker-example
npx wrangler queues create task-events
npx wrangler r2 bucket create comet-cloudflare-worker-example-assets
# paste the database_id into wrangler.jsonc, in d1_databases[0]
npx wrangler d1 migrations apply DB --local     # for wrangler dev
npx wrangler d1 migrations apply DB --remote    # for wrangler deploy

cd examples/cloudflare-worker
npm install
npm run dev

Running the tests

# build check for the wasm32 target
rustup target add wasm32-unknown-unknown
npm run check

# unit tests, native, no Node/wasm needed
npm run test

# real integration, via wrangler dev (needs rustup, wasm32, jq, npm)
npm run test:integration

# informational performance test (autocannon) — only fails on connection
# errors or non-2xx responses, not on throughput
npm run test:perf

tests/integration.sh resets local D1/R2 state, applies the migrations, starts wrangler dev on port 8788, and then: hits /, does a small /echo, round-trips a 1MiB body through /echo byte for byte (proving streamed, not buffered, ingestion), checks /stream's time-to-first-byte against total time (proving response streaming), round-trips a 1MiB object through PUT/GET /assets/..., drives a real WebSocket client against /ws/echo, exercises task CRUD (including 400 for a blank title and 404 for a missing id), and then waits 8s (the queue's local max_batch_timeout is 5s) to confirm, via wrangler d1 execute, that both the created and completed events landed in task_events.

Full example

On this page