feat: add d1 support

README.md

@@ -342,6 +342,46 @@ assert(res.ok);
 assert.strictEqual(await res.text(), "Hello, World!");
 ```
 
+## D1 Databases
+
+### Enabling D1 databases
+As D1 databases are in alpha, you'll need to enable the `d1` feature on the `worker` crate.
+
+```toml
+worker = { version = "x.y.z", features = ["d1"] }
+```
+
+### Example usage
+```rust
+use worker::*;
+
+#[derive(Deserialize)]
+struct Thing {
+	thing_id: String,
+	desc: String,
+	num: u32,
+}
+
+#[event(fetch, respond_with_errors)]
+pub async fn main(request: Request,	env: Env, _ctx: Context) -> Result<Response> {
+	Router::new()
+		.get_async("/:id", |_, ctx| async move {
+			let id = ctx.param("id").unwrap()?;
+			let d1 = ctx.env.d1("things-db")?;
+			let statement = d1.prepare("SELECT * FROM things WHERE thing_id = ?1");
+			let query = statement.bind(&[id])?;
+			let result = query.first::<Thing>(None).await?;
+			match result {
+				Some(thing) => Response::from_json(&thing),
+				None => Response::error("Not found", 404),
+			}
+		})
+		.run(request, env)
+		.await
+}
+```
+
+
 # Notes and FAQ
 
 It is exciting to see how much is possible with a framework like this, by expanding the options

tsconfig.json

@@ -11,7 +11,7 @@
     // "disableReferencedProjectLoad": true,             /* Reduce the number of projects loaded automatically by TypeScript. */
 
     /* Language and Environment */
-    "target": "es2016",                                  /* Set the JavaScript language version for emitted JavaScript and include compatible library declarations. */
+    "target": "es2022",                                  /* Set the JavaScript language version for emitted JavaScript and include compatible library declarations. */
     // "lib": [],                                        /* Specify a set of bundled library declaration files that describe the target runtime environment. */
     // "jsx": "preserve",                                /* Specify what JSX code is generated. */
     // "experimentalDecorators": true,                   /* Enable experimental support for legacy experimental decorators. */
@@ -25,9 +25,9 @@
     // "moduleDetection": "auto",                        /* Control what method is used to detect module-format JS files. */
 
     /* Modules */
-    "module": "commonjs",                                /* Specify what module code is generated. */
+    "module": "ES2022",                                /* Specify what module code is generated. */
     // "rootDir": "./",                                  /* Specify the root folder within your source files. */
-    // "moduleResolution": "node10",                     /* Specify how TypeScript looks up a file from a given module specifier. */
+    "moduleResolution": "nodenext",                     /* Specify how TypeScript looks up a file from a given module specifier. */
     // "baseUrl": "./",                                  /* Specify the base directory to resolve non-relative module names. */
     // "paths": {},                                      /* Specify a set of entries that re-map imports to additional lookup locations. */
     // "rootDirs": [],                                   /* Allow multiple folders to be treated as one when resolving modules. */

worker-sandbox/.gitignore

@@ -0,0 +1 @@
+.wrangler

worker-sandbox/Cargo.toml

@@ -26,7 +26,7 @@ http = "0.2.9"
 regex = "1.8.4"
 serde = { version = "1.0.164", features = ["derive"] }
 serde_json = "1.0.96"
-worker = { path = "../worker", version = "0.0.17", features= ["queue"] }
+worker = { path = "../worker", version = "0.0.17", features= ["queue", "d1"] }
 futures-channel = "0.3.28"
 futures-util = { version = "0.3.28", default-features = false }
 rand = "0.8.5"

worker-sandbox/src/d1.rs

@@ -0,0 +1,106 @@
+use serde::Deserialize;
+use worker::*;
+
+use crate::SomeSharedData;
+
+#[derive(Deserialize)]
+struct Person {
+    id: u32,
+    name: String,
+    age: u32,
+}
+
+pub async fn prepared_statement(
+    _req: Request,
+    ctx: RouteContext<SomeSharedData>,
+) -> Result<Response> {
+    let db = ctx.env.d1("DB")?;
+    let stmt = worker::query!(&db, "SELECT * FROM people WHERE name = ?", "Ryan Upton")?;
+
+    // All rows
+    let results = stmt.all().await?;
+    let people = results.results::<Person>()?;
+
+    assert!(results.success());
+    assert_eq!(results.error(), None);
+    assert_eq!(people.len(), 1);
+    assert_eq!(people[0].name, "Ryan Upton");
+    assert_eq!(people[0].age, 21);
+    assert_eq!(people[0].id, 6);
+
+    // All columns of the first rows
+    let person = stmt.first::<Person>(None).await?.unwrap();
+    assert_eq!(person.name, "Ryan Upton");
+    assert_eq!(person.age, 21);
+
+    // The name of the first row
+    let name = stmt.first::<String>(Some("name")).await?.unwrap();
+    assert_eq!(name, "Ryan Upton");
+
+    // All of the rows as column arrays of raw JSON values.
+    let rows = stmt.raw::<serde_json::Value>().await?;
+    assert_eq!(rows.len(), 1);
+    let columns = &rows[0];
+
+    assert_eq!(columns[0].as_u64(), Some(6));
+    assert_eq!(columns[1].as_str(), Some("Ryan Upton"));
+    assert_eq!(columns[2].as_u64(), Some(21));
+
+    Response::ok("ok")
+}
+
+pub async fn batch(_req: Request, ctx: RouteContext<SomeSharedData>) -> Result<Response> {
+    let db = ctx.env.d1("DB")?;
+    let mut results = db
+        .batch(vec![
+            worker::query!(&db, "SELECT * FROM people WHERE id < 4"),
+            worker::query!(&db, "SELECT * FROM people WHERE id > 4"),
+        ])
+        .await?
+        .into_iter();
+
+    let first_results = results.next().unwrap().results::<Person>()?;
+    assert_eq!(first_results.len(), 3);
+    assert_eq!(first_results[0].id, 1);
+    assert_eq!(first_results[1].id, 2);
+    assert_eq!(first_results[2].id, 3);
+
+    let second_results = results.next().unwrap().results::<Person>()?;
+    assert_eq!(second_results.len(), 2);
+    assert_eq!(second_results[0].id, 5);
+    assert_eq!(second_results[1].id, 6);
+
+    Response::ok("ok")
+}
+
+pub async fn exec(mut req: Request, ctx: RouteContext<SomeSharedData>) -> Result<Response> {
+    let db = ctx.env.d1("DB")?;
+    let result = db
+        .exec(req.text().await?.as_ref())
+        .await
+        .expect("doesn't exist");
+
+    Response::ok(result.count().unwrap_or_default().to_string())
+}
+
+pub async fn dump(_req: Request, ctx: RouteContext<SomeSharedData>) -> Result<Response> {
+    let db = ctx.env.d1("DB")?;
+    let bytes = db.dump().await?;
+    Response::from_bytes(bytes)
+}
+
+pub async fn error(_req: Request, ctx: RouteContext<SomeSharedData>) -> Result<Response> {
+    let db = ctx.env.d1("DB")?;
+    let error = db
+        .exec("THIS IS NOT VALID SQL")
+        .await
+        .expect_err("did not get error");
+
+    if let Error::D1(error) = error {
+        assert_eq!(error.cause(), "Error in line 1: THIS IS NOT VALID SQL: ERROR 9009: SQL prepare error: near \"THIS\": syntax error in THIS IS NOT VALID SQL at offset 0")
+    } else {
+        panic!("expected D1 error");
+    }
+
+    Response::ok("")
+}

worker-sandbox/src/lib.rs

@@ -15,6 +15,7 @@ use worker::*;
 
 mod alarm;
 mod counter;
+mod d1;
 mod r2;
 mod test;
 mod utils;
@@ -497,7 +498,7 @@ pub async fn main(req: Request, env: Env, _ctx: worker::Context) -> Result<Respo
                 Either::Left((res, cancelled_fut)) => {
                     // Ensure that the cancelled future returns an AbortError.
                     match cancelled_fut.await {
-                        Err(e) if e.to_string().starts_with("AbortError") => { /* Yay! It worked, let's do nothing to celebrate */},
+                        Err(e) if e.to_string().contains("AbortError") => { /* Yay! It worked, let's do nothing to celebrate */},
                         Err(e) => panic!("Fetch errored with a different error than expected: {:#?}", e),
                         Ok(text) => panic!("Fetch unexpectedly succeeded: {}", text)
                     }
@@ -698,6 +699,11 @@ pub async fn main(req: Request, env: Env, _ctx: worker::Context) -> Result<Respo
             let messages: Vec<QueueBody> = guard.clone();
             Response::from_json(&messages)
         })
+        .get_async("/d1/prepared", d1::prepared_statement)
+        .get_async("/d1/batch", d1::batch)
+        .get_async("/d1/dump", d1::dump)
+        .post_async("/d1/exec", d1::exec)
+        .get_async("/d1/error", d1::error)
         .get_async("/r2/list-empty", r2::list_empty)
         .get_async("/r2/list", r2::list)
         .get_async("/r2/get-empty", r2::get_empty)

worker-sandbox/tests/d1.spec.ts

@@ -0,0 +1,70 @@
+import { describe, test, expect, beforeAll } from "vitest";
+
+const hasLocalDevServer = await fetch("http://localhost:8787/request")
+  .then((resp) => resp.ok)
+  .catch(() => false);
+
+async function exec(query: string): Promise<number> {
+  const resp = await fetch("http://localhost:8787/d1/exec", {
+    method: "POST",
+    body: query.split("\n").join(""),
+  });
+
+  const body = await resp.text();
+  expect(resp.status).toBe(200);
+  return Number(body);
+}
+
+describe.skipIf(!hasLocalDevServer)("d1", () => {
+  test("create table", async () => {
+    const query = `CREATE TABLE IF NOT EXISTS uniqueTable (
+      id INTEGER PRIMARY KEY,
+      name TEXT NOT NULL,
+      age INTEGER NOT NULL
+    );`;
+
+    expect(await exec(query)).toBe(1);
+  });
+
+  test("insert data", async () => {
+    let query = `CREATE TABLE IF NOT EXISTS people (
+      id INTEGER PRIMARY KEY,
+      name TEXT NOT NULL,
+      age INTEGER NOT NULL
+    );`;
+
+    expect(await exec(query)).toBe(1);
+
+    query = `INSERT OR IGNORE INTO people
+    (id, name, age)
+    VALUES
+    (1, 'Freddie Pearce', 26),
+    (2, 'Wynne Ogley', 67),
+    (3, 'Dorian Fischer', 19),
+    (4, 'John Smith', 92),
+    (5, 'Magaret Willamson', 54),
+    (6, 'Ryan Upton', 21);`;
+
+    expect(await exec(query)).toBe(1);
+  });
+
+  test("prepared statement", async () => {
+    const resp = await fetch("http://localhost:8787/d1/prepared");
+    expect(resp.status).toBe(200);
+  });
+
+  test("batch", async () => {
+    const resp = await fetch("http://localhost:8787/d1/batch");
+    expect(resp.status).toBe(200);
+  });
+
+  test("dump", async () => {
+    const resp = await fetch("http://localhost:8787/d1/dump");
+    expect(resp.status).toBe(200);
+  });
+
+  test("dump", async () => {
+    const resp = await fetch("http://localhost:8787/d1/error");
+    expect(resp.status).toBe(200);
+  });
+});

worker-sandbox/wrangler.toml

@@ -21,6 +21,12 @@ remote-service = "./remote-service"
 [durable_objects]
 bindings = [{ name = "COUNTER", class_name = "Counter" }, { name = "ALARM", class_name = "AlarmObject" }]
 
+[[d1_databases]]
+binding = 'DB'
+database_name = 'my_db'
+database_id = '.'
+preview_database_id = '.'
+
 [[queues.consumers]]
  queue = "my_queue"
 

worker-sys/Cargo.toml

@@ -41,4 +41,5 @@ features = [
 ]
 
 [features]
+d1 = []
 queue = []

worker-sys/src/types.rs

@@ -1,4 +1,6 @@
 mod context;
+#[cfg(feature = "d1")]
+mod d1;
 mod durable_object;
 mod dynamic_dispatcher;
 mod fetcher;
@@ -13,6 +15,8 @@ mod tls_client_auth;
 mod websocket_pair;
 
 pub use context::*;
+#[cfg(feature = "d1")]
+pub use d1::*;
 pub use durable_object::*;
 pub use dynamic_dispatcher::*;
 pub use fetcher::*;

worker-sys/src/types/d1.rs

@@ -0,0 +1,75 @@
+use ::js_sys::Object;
+use wasm_bindgen::prelude::*;
+
+use js_sys::{Array, Promise};
+
+#[wasm_bindgen]
+extern "C" {
+    #[derive(Debug, Clone)]
+    pub type D1Result;
+
+    #[wasm_bindgen(structural, method, getter, js_name=results)]
+    pub fn results(this: &D1Result) -> Option<Array>;
+
+    #[wasm_bindgen(structural, method, getter, js_name=success)]
+    pub fn success(this: &D1Result) -> bool;
+
+    #[wasm_bindgen(structural, method, getter, js_name=error)]
+    pub fn error(this: &D1Result) -> Option<String>;
+
+    #[wasm_bindgen(structural, method, getter, js_name=meta)]
+    pub fn meta(this: &D1Result) -> Object;
+}
+
+#[wasm_bindgen]
+extern "C" {
+    #[derive(Debug, Clone)]
+    pub type D1ExecResult;
+
+    #[wasm_bindgen(structural, method, getter, js_name=count)]
+    pub fn count(this: &D1ExecResult) -> Option<u32>;
+
+    #[wasm_bindgen(structural, method, getter, js_name=duration)]
+    pub fn duration(this: &D1ExecResult) -> Option<f64>;
+}
+
+#[wasm_bindgen]
+extern "C" {
+    #[wasm_bindgen(extends=::js_sys::Object, js_name=D1Database)]
+    #[derive(Debug, Clone, PartialEq, Eq)]
+    pub type D1Database;
+
+    #[wasm_bindgen(structural, method, js_class=D1Database, js_name=prepare)]
+    pub fn prepare(this: &D1Database, query: &str) -> D1PreparedStatement;
+
+    #[wasm_bindgen(structural, method, js_class=D1Database, js_name=dump)]
+    pub fn dump(this: &D1Database) -> Promise;
+
+    #[wasm_bindgen(structural, method, js_class=D1Database, js_name=batch)]
+    pub fn batch(this: &D1Database, statements: Array) -> Promise;
+
+    #[wasm_bindgen(structural, method, js_class=D1Database, js_name=exec)]
+    pub fn exec(this: &D1Database, query: &str) -> Promise;
+}
+
+#[wasm_bindgen]
+extern "C" {
+    #[wasm_bindgen(extends=::js_sys::Object, js_name=D1PreparedStatement)]
+    #[derive(Debug, Clone, PartialEq, Eq)]
+    pub type D1PreparedStatement;
+
+    #[wasm_bindgen(structural, method, catch, variadic, js_class=D1PreparedStatement, js_name=bind)]
+    pub fn bind(this: &D1PreparedStatement, values: Array) -> Result<D1PreparedStatement, JsValue>;
+
+    #[wasm_bindgen(structural, method, js_class=D1PreparedStatement, js_name=first)]
+    pub fn first(this: &D1PreparedStatement, col_name: Option<&str>) -> Promise;
+
+    #[wasm_bindgen(structural, method, js_class=D1PreparedStatement, js_name=run)]
+    pub fn run(this: &D1PreparedStatement) -> Promise;
+
+    #[wasm_bindgen(structural, method, js_class=D1PreparedStatement, js_name=all)]
+    pub fn all(this: &D1PreparedStatement) -> Promise;
+
+    #[wasm_bindgen(structural, method, js_class=D1PreparedStatement, js_name=raw)]
+    pub fn raw(this: &D1PreparedStatement) -> Promise;
+}