Use Kysely with DaloyJS

import type { Generated, Insertable, Selectable } from "kysely";

export interface UserTable {
  id: Generated<string>; // uuid, DEFAULT gen_random_uuid()
  email: string;
  name: string | null;
  created_at: Generated<Date>; // timestamptz, DEFAULT now()
}

export interface Database {
  users: UserTable;
}

export type UserRow = Selectable<UserTable>;
export type NewUser = Insertable<UserTable>;

import { Kysely, PostgresDialect } from "kysely";
import { Pool } from "pg";
import type { App } from "@daloyjs/core";
import type { Database } from "./types.ts";

// One pool per process. Size `max` so (max x instances) stays under the
// database's connection limit.
const pool = new Pool({ connectionString: process.env.DATABASE_URL!, max: 10 });
export const db = new Kysely<Database>({ dialect: new PostgresDialect({ pool }) });

// Call with the root app, not through app.register(): a decoration made
// inside a registered plugin is visible only to that plugin's own routes.
export function kyselyPlugin(app: App) {
  app.decorate("db", db);
  app.onClose(async () => {
    await db.destroy(); // ends the pg pool
  });
}

// same module as the plugin above
declare module "@daloyjs/core" {
  interface AppState {
    db: typeof db;
  }
}

import { z } from "zod";
import { DatabaseError } from "pg";
import { App, ConflictError, NotFoundError } from "@daloyjs/core";
import { serve } from "@daloyjs/core/node";
import { kyselyPlugin } from "./db/kysely.ts";
import type { UserRow } from "./db/types.ts";

const app = new App();
kyselyPlugin(app);

// Dates go over the wire as ISO strings: z.coerce.date() has no
// representation in the generated OpenAPI document.
const UserSchema = z.object({
  id: z.uuid(),
  email: z.email(),
  name: z.string().nullable(),
  createdAt: z.iso.datetime(),
});

const toUser = (row: UserRow) => ({
  id: row.id,
  email: row.email,
  name: row.name,
  createdAt: row.created_at.toISOString(),
});

app.get(
  "/users/:id",
  {
    operationId: "getUser",
    request: { params: z.object({ id: z.uuid() }) },
    responses: {
      200: { description: "Found", body: UserSchema },
      404: { description: "Not found" },
    },
  },
  async ({ params, state }) => {
    const user = await state.db
      .selectFrom("users")
      .selectAll()
      .where("id", "=", params.id)
      .executeTakeFirst();
    if (!user) throw new NotFoundError("User not found");
    return { status: 200, body: toUser(user) };
  },
);

app.post(
  "/users",
  {
    operationId: "createUser",
    request: { body: z.object({ email: z.email(), name: z.string().max(200).optional() }) },
    responses: {
      201: { description: "Created", body: UserSchema },
      409: { description: "Email already in use" },
    },
  },
  async ({ body, state }) => {
    try {
      // Pick columns explicitly: never spread a request body into values().
      const created = await state.db
        .insertInto("users")
        .values({ email: body.email, name: body.name ?? null })
        .returningAll()
        .executeTakeFirstOrThrow();
      return { status: 201, body: toUser(created) };
    } catch (err) {
      if (err instanceof DatabaseError && err.code === "23505") {
        throw new ConflictError("Email already in use");
      }
      throw err;
    }
  },
);

app.delete(
  "/users/:id",
  {
    operationId: "deleteUser",
    request: { params: z.object({ id: z.uuid() }) },
    responses: {
      204: { description: "Deleted" },
      404: { description: "Not found" },
    },
  },
  async ({ params, state }) => {
    const result = await state.db
      .deleteFrom("users")
      .where("id", "=", params.id)
      .executeTakeFirst();
    if (result.numDeletedRows === 0n) throw new NotFoundError("User not found");
    return { status: 204, body: undefined };
  },
);

await app.ready();
serve(app, { port: 3000 });

import { sql } from "kysely";

handler: async ({ body, state }) => {
  const order = await state.db.transaction().execute(async (trx) => {
    const created = await trx
      .insertInto("orders")
      .values({ sku: body.sku, qty: body.qty })
      .returningAll()
      .executeTakeFirstOrThrow();
    await trx
      .updateTable("inventory")
      .set({ stock: sql<number>`stock - ${body.qty}` })
      .where("sku", "=", body.sku)
      .execute();
    return created;
  });
  return { status: 201, body: { id: order.id, sku: order.sku, qty: order.qty } };
}

import { sql } from "kysely";

// SAFE: ${params.email} is sent as a bound parameter ($1).
const rows = await sql<{ id: string }>`
  SELECT id FROM users WHERE lower(email) = lower(${params.email})
`.execute(state.db);

// SAFE: identifiers from an allowlist you own, values still bound.
const SORTABLE = { created: "created_at", email: "email" } as const;
const column = SORTABLE[query.sort]; // query.sort: z.enum(["created", "email"])
await state.db.selectFrom("users").selectAll().orderBy(sql.ref(column), "desc").execute();

// DANGEROUS: sql.raw() and sql.lit() splice text into the query.
await sql.raw(`SELECT * FROM users WHERE email = '${params.email}'`).execute(state.db);

import { sql } from "kysely";
import type { Kysely } from "kysely";

// Migrations are frozen in time: type them against Kysely<any>, not against
// today's Database interface.
export async function up(db: Kysely<any>): Promise<void> {
  await db.schema
    .createTable("users")
    .addColumn("id", "uuid", (col) => col.primaryKey().defaultTo(sql`gen_random_uuid()`))
    .addColumn("email", "text", (col) => col.notNull().unique())
    .addColumn("name", "text")
    .addColumn("created_at", "timestamptz", (col) => col.notNull().defaultTo(sql`now()`))
    .execute();

  // Backs the keyset pagination below.
  await db.schema
    .createIndex("users_created_at_id_idx")
    .on("users")
    .columns(["created_at", "id"])
    .execute();
}

export async function down(db: Kysely<any>): Promise<void> {
  await db.schema.dropTable("users").execute();
}

import { promises as fs } from "node:fs";
import path from "node:path";
import { pathToFileURL } from "node:url";
import { Kysely, PostgresDialect } from "kysely";
import { FileMigrationProvider, Migrator } from "kysely/migration";
import { Pool } from "pg";

// A dedicated role with DDL rights; the app's runtime role should not have them.
const db = new Kysely<any>({
  dialect: new PostgresDialect({
    pool: new Pool({ connectionString: process.env.MIGRATION_DATABASE_URL!, max: 1 }),
  }),
});

const migrator = new Migrator({
  db,
  provider: new FileMigrationProvider({
    fs,
    path,
    migrationFolder: path.join(import.meta.dirname, "../migrations"),
    // import() needs a file:// URL for absolute paths on Windows.
    import: (file) => import(pathToFileURL(file).href),
  }),
});

const { error, results } = await migrator.migrateToLatest();
for (const result of results ?? []) {
  console.log(`${result.status}: ${result.migrationName}`);
}
await db.destroy();

if (error) {
  console.error("migration failed", error);
  process.exit(1);
}

node scripts/migrate.ts

import { z } from "zod";
import { BadRequestError, decodeCursor, encodeCursor, paginationQuery } from "@daloyjs/core";

const CursorSchema = z.strictObject({ createdAt: z.iso.datetime(), id: z.uuid() });

function readCursor(cursor: string | undefined) {
  if (!cursor) return undefined;
  const parsed = CursorSchema.safeParse(decodeCursor(cursor));
  if (!parsed.success) throw new BadRequestError("Invalid pagination cursor.");
  return parsed.data;
}

app.get(
  "/users",
  {
    operationId: "listUsers",
    request: { query: paginationQuery({ maxLimit: 100 }) },
    responses: {
      200: {
        description: "One page of users",
        body: z.object({ items: z.array(UserSchema), nextCursor: z.string().nullable() }),
      },
      400: { description: "Invalid cursor" },
    },
  },
  async ({ query, state }) => {
    const after = readCursor(query.cursor);
    let q = state.db
      .selectFrom("users")
      .selectAll()
      .orderBy("created_at", "desc")
      .orderBy("id", "desc")
      .limit(query.limit + 1);
    if (after) {
      // Row-value comparison: (created_at, id) < ($1, $2), served by the index.
      q = q.where((eb) =>
        eb(eb.refTuple("created_at", "id"), "<", eb.tuple(new Date(after.createdAt), after.id)),
      );
    }
    const rows = await q.execute();
    const page = rows.slice(0, query.limit);
    const last = rows.length > query.limit ? page[page.length - 1] : undefined;
    return {
      status: 200,
      body: {
        items: page.map(toUser),
        nextCursor: last
          ? encodeCursor({ createdAt: last.created_at.toISOString(), id: last.id })
          : null,
      },
    };
  },
);

import { neon } from "@neondatabase/serverless";
import { Kysely } from "kysely";
import { NeonDialect } from "kysely-neon";
import type { Database } from "./types.ts";

// Each query is one HTTP request: nothing to pool, nothing to close.
export const db = new Kysely<Database>({
  dialect: new NeonDialect({ neon: neon(process.env.DATABASE_URL!) }),
});

import { env } from "cloudflare:workers";
import { Kysely } from "kysely";
import { D1Dialect } from "kysely-d1";
import { App } from "@daloyjs/core";
import { toFetchHandler } from "@daloyjs/core/cloudflare";
import type { Database } from "./db/types.ts";

// `wrangler types` generates the global Env interface, which types env.DB.
const app = new App();

// A D1 binding is not a socket, so one Kysely instance can serve every
// request. Decorate once, before the routes: calling decorate() inside the
// Worker fetch handler throws on the second request.
app.decorate("db", new Kysely<Database>({ dialect: new D1Dialect({ database: env.DB }) }));

// ...register routes that read state.db

export default toFetchHandler(app);

import { Kysely } from "kysely";
import { LibsqlDialect } from "@libsql/kysely-libsql";
import { PlanetScaleDialect } from "kysely-planetscale";
import type { Database } from "./types.ts";

// Turso / libSQL (the dialect does not support file: URLs).
new Kysely<Database>({
  dialect: new LibsqlDialect({ url: process.env.TURSO_URL!, authToken: process.env.TURSO_TOKEN! }),
});

// PlanetScale over HTTP; accepts the same options as connect() from @planetscale/database.
new Kysely<Database>({ dialect: new PlanetScaleDialect({ url: process.env.DATABASE_URL! }) });