// Kysely 0.29 on Deno: DIALECT=pg (node-postgres Pool) or DIALECT=postgresjs (kysely-postgres-js). import { Kysely, PostgresDialect, sql } from "kysely"; import { FileMigrationProvider, Migrator } from "kysely/migration"; import { PostgresJSDialect } from "kysely-postgres-js"; import pg from "pg"; import postgres from "postgres"; import { promises as fs } from "node:fs"; import * as path from "node:path"; const url = Deno.env.get("PG_URL") ?? "postgresql://postgres:spike@127.0.0.1:55435/postgres"; const dialectName = Deno.env.get("DIALECT") ?? "pg"; const schema = Deno.env.get("SCHEMA") ?? `kysely_${dialectName}`; export function createDb(): Kysely { const dialect = dialectName === "postgresjs" ? new PostgresJSDialect({ postgres: postgres(url, { max: 4, onnotice: () => {} }) }) : new PostgresDialect({ pool: new pg.Pool({ connectionString: url, max: 4 }) }); return new Kysely({ dialect }); } if (import.meta.main) { const root = createDb(); await sql`DROP SCHEMA IF EXISTS ${sql.id(schema)} CASCADE`.execute(root); await sql`CREATE SCHEMA ${sql.id(schema)}`.execute(root); const db = root.withSchema(schema); const migrator = new Migrator({ db: root, migrationTableSchema: schema, provider: new FileMigrationProvider({ fs, path, migrationFolder: Deno.env.get("MIGRATIONS_DIR") ?? new URL("./migrations", import.meta.url).pathname }), }); // Migrations use unqualified names: run them with search_path pointing at the dialect's schema. await sql`SET search_path TO ${sql.id(schema)}`.execute(root); const r = await migrator.migrateToLatest(); console.log(dialectName, "migrate:", JSON.stringify(r.results?.map((x) => `${x.migrationName}:${x.status}`)), "error:", r.error ?? null); if (Deno.env.get("MIGRATIONS_DIR")) { await root.destroy(); Deno.exit(r.error ? 1 : 0); } await db.insertInto("items").values([{ title: "Dune", meta: JSON.stringify({ q: "1080p" }) }, { title: "Arrival", meta: null }]).execute(); const rows = await db.selectFrom("items").select(["id", "title", "meta", "created_at"]).orderBy("id").execute(); console.log(dialectName, "rows:", rows.map((r) => `${typeof r.id}:${r.id}/${r.title}/${JSON.stringify(r.meta)}/${r.created_at instanceof Date}`).join(" ")); // transaction + rollback await db.transaction().execute(async (tx) => { await tx.insertInto("items").values({ title: "tx-ok" }).execute(); }).catch(() => {}); await db.transaction().execute(async (tx) => { await tx.insertInto("items").values({ title: "tx-rollback" }).execute(); throw new Error("rollback"); }).catch((e) => console.log(dialectName, "tx error ->", e.message)); const titles = (await db.selectFrom("items").select("title").execute()).map((r) => r.title); console.log(dialectName, "titles after tx:", titles.join(",")); // concurrency: 50 parallel queries on pool max=4 const t0 = performance.now(); await Promise.all(Array.from({ length: 50 }, (_, i) => sql`SELECT pg_sleep(0.01), ${i}::int AS i`.execute(root))); console.log(dialectName, `50 parallel queries on pool=4: ${(performance.now() - t0).toFixed(0)}ms`); await root.destroy(); console.log(dialectName, "destroyed cleanly"); }