feat: update postgres client (#9)

Author: danewalters

README.md

@@ -4,28 +4,28 @@
 [![Latest version](https://deno.land/badge/kysely_deno_postgres_dialect/version)](https://deno.land/x/kysely_deno_postgres_dialect)
 
 [Kysely](https://github.com/kysely-org/kysely) dialect for PostgreSQL using the
-[deno-postgres](https://github.com/denodrivers/postgres) client.
+[postgresjs](https://github.com/porsager/postgres) client.
 
 ## 🚀 Getting started
 
 ```typescript
 import {
-  KyselyDenoPostgresDialect,
+  PostgresJSDialect,
   setup,
-} from "https://deno.land/x/kysely_deno_postgres_dialect/mod.ts";
+} from "https://deno.land/x/kysely_postgrs_js_dialect/mod.ts";
+import postgres from "https://deno.land/x/postgresjs@v3.4.4/mod.js";
 
 setup(() => {
-  const dialect = new KyselyDenoPostgresDialect({
-    pool: new Pool(
+  const dialect = new PostgresJSDialect({
+    postgres: postgres(
       {
         database: "postgres",
         hostname: "localhost",
         password: "postgres",
         port: 5432,
         user: "postgres",
+        max: 10,
       },
-      10,
-      true,
     ),
   });
 
@@ -47,44 +47,31 @@ See detail at `./tests/testing/utils_test.ts`.
 
 ```typescript
 import {
-  KyselyDenoPostgresDialect,
-  PostgresConnection,
+  PostgresJSDialect,
   setup,
   wrapTransaction,
 } from "https://deno.land/x/kysely_deno_postgres_dialect/mod.ts";
 import { setupTesting } from "https://deno.land/x/kysely_deno_postgres_dialect/testing.ts";
 
-const connections: PostgresConnection[] = [];
-
 setup(() => {
-  const dialect = new KyselyDenoPostgresDialect({
-    pool: new Pool(
+  const dialect = new PostgresJSDialect({
+    postgres: postgres(
       {
         database: "postgres",
         hostname: "localhost",
         password: "postgres",
         port: 5432,
         user: "postgres",
+        max: 10,
       },
-      10,
-      true,
     ),
-    onCreateConnection: (connection: DatabaseConnection) => {
-      connections.push(connection as PostgresConnection);
-    },
   });
 
   return new Kysely<Database>({
     dialect,
   });
 });
 
-async function endConnections() {
-  for (const connection of connections) {
-    await connection.end();
-  }
-}
-
 // test files
 
 const {
@@ -106,7 +93,6 @@ describe("tests", () => {
   });
   afterEach(async () => {
     // note: fix Leaking resources error
-    await endConnections();
     await afterEachFn();
   });
 

deps.ts

@@ -1,3 +1,7 @@
+export type {
+  ReservedSql,
+  Sql,
+} from "https://deno.land/x/postgresjs@v3.4.4/types/index.d.ts";
 export {
   CompiledQuery,
   type DatabaseConnection,
@@ -13,8 +17,7 @@ export {
   type QueryResult,
   Transaction,
   type TransactionSettings,
-} from "https://esm.sh/kysely@0.27.0";
-export { Pool, PoolClient } from "https://deno.land/x/postgres@v0.17.0/mod.ts";
+} from "https://cdn.jsdelivr.net/npm/kysely@0.27.3/dist/esm/index.js";
 
-export * as kysely from "https://esm.sh/kysely@0.27.0";
-export * as postgres from "https://deno.land/x/postgres@v0.17.0/mod.ts";
+export * as kysely from "https://cdn.jsdelivr.net/npm/kysely@0.27.3/dist/esm/index.js";
+export { default as postgres } from "https://deno.land/x/postgresjs@v3.4.4/mod.js";

mod.ts

@@ -1,44 +1,8 @@
-import {
-  DatabaseIntrospector,
-  Dialect,
-  DialectAdapter,
-  Driver,
-  Kysely,
-  PostgresAdapter,
-  PostgresIntrospector,
-  PostgresQueryCompiler,
-  QueryCompiler,
-} from "./deps.ts";
-
-import { PostgresDriver } from "./src/postgres-driver.ts";
-import { PostgresDialectConfig } from "./src/postgres-dialect-config.ts";
-
-export { type DatabaseConnection, kysely, Pool, postgres } from "./deps.ts";
-export { type PostgresConnection } from "./src/postgres-connection.ts";
-
-export class KyselyDenoPostgresDialect implements Dialect {
-  readonly #config: PostgresDialectConfig;
-
-  constructor(config: PostgresDialectConfig) {
-    this.#config = config;
-  }
-
-  createDriver(): Driver {
-    return new PostgresDriver(this.#config);
-  }
-
-  createQueryCompiler(): QueryCompiler {
-    return new PostgresQueryCompiler();
-  }
-
-  createAdapter(): DialectAdapter {
-    return new PostgresAdapter();
-  }
-
-  // deno-lint-ignore no-explicit-any
-  createIntrospector(db: Kysely<any>): DatabaseIntrospector {
-    return new PostgresIntrospector(db);
-  }
-}
+export { type DatabaseConnection, kysely, postgres } from "./deps.ts";
+export {
+  PostgresJSConnection,
+  PostgresJSDialect,
+  PostgresJSDialectError,
+} from "./src/mod.ts";
 
 export { getDB, setup, wrapTransaction } from "./src/utils.ts";

src/connection.ts

@@ -0,0 +1,80 @@
+import {
+  CompiledQuery,
+  DatabaseConnection,
+  QueryResult,
+  ReservedSql,
+  TransactionSettings,
+} from "../deps.ts";
+import { PostgresJSDialectError } from "./errors.ts";
+
+export class PostgresJSConnection implements DatabaseConnection {
+  #reservedConnection: ReservedSql;
+
+  constructor(reservedConnection: ReservedSql) {
+    this.#reservedConnection = reservedConnection;
+  }
+
+  async beginTransaction(settings: TransactionSettings): Promise<void> {
+    const { isolationLevel } = settings;
+
+    const compiledQuery = CompiledQuery.raw(
+      isolationLevel
+        ? `start transaction isolation level ${isolationLevel}`
+        : "begin",
+    );
+
+    await this.executeQuery(compiledQuery);
+  }
+
+  async commitTransaction(): Promise<void> {
+    await this.executeQuery(CompiledQuery.raw("commit"));
+  }
+
+  async executeQuery<R>(
+    compiledQuery: CompiledQuery<unknown>,
+  ): Promise<QueryResult<R>> {
+    const result = await this.#reservedConnection.unsafe<R[]>(
+      compiledQuery.sql,
+      // deno-lint-ignore no-explicit-any
+      compiledQuery.parameters.slice() as any,
+    );
+
+    const rows = Array.from(result.values());
+
+    if (["INSERT", "UPDATE", "DELETE"].includes(result.command)) {
+      const numAffectedRows = BigInt(result.count);
+
+      return { numAffectedRows, rows };
+    }
+
+    return { rows };
+  }
+
+  releaseConnection(): void {
+    this.#reservedConnection.release();
+
+    this.#reservedConnection = null!;
+  }
+
+  async rollbackTransaction(): Promise<void> {
+    await this.executeQuery(CompiledQuery.raw("rollback"));
+  }
+
+  async *streamQuery<R>(
+    compiledQuery: CompiledQuery<unknown>,
+    chunkSize: number,
+  ): AsyncIterableIterator<QueryResult<R>> {
+    if (!Number.isInteger(chunkSize) || chunkSize <= 0) {
+      throw new PostgresJSDialectError("chunkSize must be a positive integer");
+    }
+
+    const cursor = this.#reservedConnection
+      // deno-lint-ignore no-explicit-any
+      .unsafe<R[]>(compiledQuery.sql, compiledQuery.parameters.slice() as any)
+      .cursor(chunkSize);
+
+    for await (const rows of cursor) {
+      yield { rows };
+    }
+  }
+}

src/dialect.ts

@@ -0,0 +1,39 @@
+import {
+  type DatabaseIntrospector,
+  type Dialect,
+  type DialectAdapter,
+  type Driver,
+  type Kysely,
+  PostgresAdapter,
+  PostgresIntrospector,
+  PostgresQueryCompiler,
+  type QueryCompiler,
+} from "../deps.ts";
+import { PostgresJSDriver } from "./driver.ts";
+import type { PostgresJSDialectConfig } from "./types.ts";
+import { freeze } from "./postgres-utils.ts";
+
+export class PostgresJSDialect implements Dialect {
+  readonly #config: PostgresJSDialectConfig;
+
+  constructor(config: PostgresJSDialectConfig) {
+    this.#config = freeze({ ...config });
+  }
+
+  createAdapter(): DialectAdapter {
+    return new PostgresAdapter();
+  }
+
+  createDriver(): Driver {
+    return new PostgresJSDriver(this.#config);
+  }
+
+  // deno-lint-ignore no-explicit-any
+  createIntrospector(db: Kysely<any>): DatabaseIntrospector {
+    return new PostgresIntrospector(db);
+  }
+
+  createQueryCompiler(): QueryCompiler {
+    return new PostgresQueryCompiler();
+  }
+}

src/driver.ts

@@ -0,0 +1,46 @@
+import type { Driver, TransactionSettings } from "../deps.ts";
+import { PostgresJSConnection } from "./connection.ts";
+import type { PostgresJSDialectConfig } from "./types.ts";
+import { freeze } from "./postgres-utils.ts";
+
+export class PostgresJSDriver implements Driver {
+  readonly #config: PostgresJSDialectConfig;
+
+  constructor(config: PostgresJSDialectConfig) {
+    this.#config = freeze({ ...config });
+  }
+
+  async init(): Promise<void> {
+    // noop
+  }
+
+  async acquireConnection(): Promise<PostgresJSConnection> {
+    const reservedConnection = await this.#config.postgres.reserve();
+
+    return new PostgresJSConnection(reservedConnection);
+  }
+
+  async beginTransaction(
+    connection: PostgresJSConnection,
+    settings: TransactionSettings,
+  ): Promise<void> {
+    await connection.beginTransaction(settings);
+  }
+
+  async commitTransaction(connection: PostgresJSConnection): Promise<void> {
+    await connection.commitTransaction();
+  }
+
+  async rollbackTransaction(connection: PostgresJSConnection): Promise<void> {
+    await connection.rollbackTransaction();
+  }
+
+  // deno-lint-ignore require-await
+  async releaseConnection(connection: PostgresJSConnection): Promise<void> {
+    connection.releaseConnection();
+  }
+
+  async destroy(): Promise<void> {
+    await this.#config.postgres.end();
+  }
+}

src/errors.ts

@@ -0,0 +1,6 @@
+export class PostgresJSDialectError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "PostgresJSDialectError";
+  }
+}

src/mod.ts

@@ -0,0 +1,5 @@
+export * from "./connection.ts";
+export * from "./dialect.ts";
+export * from "./driver.ts";
+export * from "./errors.ts";
+export type * from "./types.ts";