initial commit pt. 2.

Author: igalklebanov

src/connection.ts

@@ -0,0 +1,93 @@
+import type {CompiledQuery, DatabaseConnection, QueryResult, TransactionSettings} from 'kysely'
+import type {Sql} from 'postgres'
+
+import {PostgresJSDialectError} from './errors.js'
+import type {PostgresJSDialectConfig} from './types.js'
+import {freeze} from './utils.js'
+
+export class PostgresJSConnection implements DatabaseConnection {
+  readonly #config: PostgresJSDialectConfig
+  #sql: Sql
+  #transaction?: Sql
+
+  constructor(config: PostgresJSDialectConfig, sql: Sql) {
+    this.#config = freeze({...config})
+    this.#sql = sql
+  }
+
+  async beginTransaction(settings: TransactionSettings): Promise<void> {
+    if (this.#transaction) {
+      throw new PostgresJSDialectError('transaction already begun!')
+    }
+
+    const {isolationLevel} = settings
+
+    this.#transaction = this.#config.postgres({...this.#config.options, max: 1})
+
+    const statement = `start transaction${isolationLevel ? ` ${isolationLevel}` : ''}`
+
+    await this.#transaction.unsafe(statement)
+  }
+
+  async commitTransaction(): Promise<void> {
+    if (!this.#transaction) {
+      throw new PostgresJSDialectError('no transaction to commit!')
+    }
+
+    await this.#transaction`commit`
+
+    this.#releaseTransaction()
+  }
+
+  async executeQuery<R>(compiledQuery: CompiledQuery<unknown>): Promise<QueryResult<R>> {
+    const result = await this.#resolveExecutor().unsafe<R[]>(compiledQuery.sql, 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}
+  }
+
+  async rollbackTransaction(): Promise<void> {
+    if (!this.#transaction) {
+      throw new PostgresJSDialectError('no transaction to rollback!')
+    }
+
+    await this.#transaction`rollback`
+
+    this.#releaseTransaction()
+  }
+
+  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.#resolveExecutor()
+      .unsafe<R[]>(compiledQuery.sql, compiledQuery.parameters.slice() as any)
+      .cursor(chunkSize)
+
+    for await (const rows of cursor) {
+      yield {rows}
+    }
+  }
+
+  #resolveExecutor(): Sql {
+    return this.#transaction || this.#sql
+  }
+
+  #releaseTransaction(): void {
+    if (this.#transaction) {
+      this.#transaction.end()
+      this.#transaction = undefined
+    }
+  }
+}

src/dialect.ts

@@ -0,0 +1,39 @@
+import {
+  PostgresAdapter,
+  PostgresIntrospector,
+  PostgresQueryCompiler,
+  type DatabaseIntrospector,
+  type Dialect,
+  type DialectAdapter,
+  type Driver,
+  type Kysely,
+  type QueryCompiler,
+} from 'kysely'
+
+import {PostgresJSDriver} from './driver.js'
+import type {PostgresJSDialectConfig} from './types.js'
+import {freeze} from './utils.js'
+
+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)
+  }
+
+  createIntrospector(db: Kysely<any>): DatabaseIntrospector {
+    return new PostgresIntrospector(db)
+  }
+
+  createQueryCompiler(): QueryCompiler {
+    return new PostgresQueryCompiler()
+  }
+}

src/driver.ts

@@ -0,0 +1,44 @@
+import type {Driver, TransactionSettings} from 'kysely'
+import type {Sql} from 'postgres'
+
+import {PostgresJSConnection} from './connection.js'
+import type {PostgresJSDialectConfig} from './types.js'
+import {freeze} from './utils.js'
+
+export class PostgresJSDriver implements Driver {
+  readonly #config: PostgresJSDialectConfig
+  readonly #sql: Sql
+
+  constructor(config: PostgresJSDialectConfig) {
+    this.#config = freeze({...config})
+    this.#sql = this.#config.postgres(this.#config.options)
+  }
+
+  async init(): Promise<void> {
+    // noop
+  }
+
+  async acquireConnection(): Promise<PostgresJSConnection> {
+    return new PostgresJSConnection(this.#config, this.#sql)
+  }
+
+  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()
+  }
+
+  async releaseConnection(_: PostgresJSConnection): Promise<void> {
+    // noop
+  }
+
+  async destroy(): Promise<void> {
+    await this.#sql.end()
+  }
+}

src/errors.ts

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

src/index.ts

@@ -0,0 +1,5 @@
+export * from './connection.js'
+export * from './dialect.js'
+export * from './driver.js'
+export * from './errors.js'
+export type * from './types.js'

src/types.ts

@@ -0,0 +1,7 @@
+import type postgres from 'postgres'
+import type {Options} from 'postgres'
+
+export interface PostgresJSDialectConfig {
+  options: Options<any>
+  postgres: typeof postgres
+}

src/utils.ts

@@ -0,0 +1,3 @@
+export function freeze<T>(obj: T): Readonly<T> {
+  return Object.freeze(obj)
+}