Kysely & Deno postgres

_seeder.ts

import { hash } from "https://deno.land/x/bcrypt@v0.4.1/mod.ts";
import { Kysely } from "https://esm.sh/kysely@0.22.0";

export async function seed(db: Kysely<unknown>) {
  const userAdmin = await db.selectFrom("users" as never).where(
    "email",
    "=",
    "admin@example.net" as never,
  ).executeTakeFirst();
  if (!userAdmin) {
    console.log("Create admin user");
    await db.insertInto("users" as never)
      .values({
        email: "admin@example.net",
        emailVerified: new Date(),
        password: await hash("password"),
        avatarUrl:
          "https://www.gravatar.com/avatar/00000000000000000000000000000000",
      })
      .execute();
  }
}

export default seed;

db.ts

import { Kysely } from "https://esm.sh/kysely@0.22.0";
import { createKyPg, sql } from "./kypg.ts";

export interface AnyDB {
  [k: string]: any;
}

declare let globalThis: { kyPgGlobal: Kysely<AnyDB> };
let db: Kysely<AnyDB>;

if (!globalThis.kyPgGlobal) {
  const dsn = Deno.env.get("DATABASE_URL"),
    poolSize = parseInt(Deno.env.get("DATABASE_POOL_SIZE") || "2");
  if (!dsn) throw new Error("Invalid Database Configuration");
  db = globalThis.kyPgGlobal = createKyPg(dsn, {
    poolSize,
    //log: Deno.env.get("MODE") !== "production" ? ["query", "error"] : undefined,
  });
} else {
  db = globalThis.kyPgGlobal;
}

addEventListener("unload", async () => {
  await db.destroy();
});

export { db, sql };

kypg.ts

import {
  CompiledQuery,
  DatabaseConnection,
  DatabaseIntrospector,
  Dialect,
  DialectAdapter,
  Driver,
  Kysely,
  KyselyPlugin,
  LogConfig,
  Migration,
  MigrationProvider,
  Migrator,
  PostgresAdapter,
  PostgresIntrospector,
  PostgresQueryCompiler,
  QueryCompiler,
  QueryResult,
  sql,
  TransactionSettings,
} from "https://esm.sh/kysely@0.22.0";
import { Pool, PoolClient } from "https://deno.land/x/postgres@v0.17.0/mod.ts";

export { sql };

export function createKyPgMigrator<TDB = unknown>(
  db: Kysely<TDB>,
  migrationFolder: URL,
) {
  return new Migrator({
    db,
    provider: new DenoFileMigrationProvider(migrationFolder),
  });
}

export function createKyPg<TDB = unknown>(
  connectionString: string,
  opts?: {
    poolSize?: number;
    log?: LogConfig;
    plugins?: KyselyPlugin[];
  },
) {
  return new Kysely<TDB>({
    dialect: new DenoProstgresDialect<TDB>({
      pool: new Pool(connectionString, opts?.poolSize || 2),
    }),
    plugins: opts?.plugins,
    log: opts?.log,
  });
}

const PRIVATE_RELEASE_METHOD = Symbol();

interface DenoProstgresDialectConfig {
  pool: Pool | (() => Pool | Promise<Pool>);
}

class DenoProstgresDialect<TDB = unknown> implements Dialect {
  constructor(private config: DenoProstgresDialectConfig) {}
  createDriver(): Driver {
    return new DenoPostgresDriver(this.config);
  }
  createQueryCompiler(): QueryCompiler {
    return new PostgresQueryCompiler();
  }
  createAdapter(): DialectAdapter {
    return new PostgresAdapter();
  }
  createIntrospector(db: Kysely<TDB>): DatabaseIntrospector {
    return new PostgresIntrospector(db);
  }
}

class DenoPostgresDriver implements Driver {
  #pool?: Pool;
  #connections?: WeakMap<PoolClient, DenoPostgresDatabaseConnection>;

  constructor(private config: DenoProstgresDialectConfig) {}
  async init(): Promise<void> {
    this.#pool = typeof this.config.pool === "function"
      ? await this.config.pool()
      : this.config.pool;
  }
  async acquireConnection(): Promise<DenoPostgresDatabaseConnection> {
    const client = await this.#pool!.connect();
    let connection = this.#connections?.get(client);
    if (!connection) {
      connection = new DenoPostgresDatabaseConnection(client);
      this.#connections?.set(client, connection);
    }
    return connection;
  }
  async beginTransaction(
    connection: DenoPostgresDatabaseConnection,
    settings: TransactionSettings,
  ): Promise<void> {
    await connection.executeQuery(CompiledQuery.raw("begin"));
  }
  async commitTransaction(
    connection: DenoPostgresDatabaseConnection,
  ): Promise<void> {
    await connection.executeQuery(CompiledQuery.raw("commit"));
  }
  async rollbackTransaction(
    connection: DenoPostgresDatabaseConnection,
  ): Promise<void> {
    await connection.executeQuery(CompiledQuery.raw("rollback"));
  }
  async releaseConnection(
    connection: DenoPostgresDatabaseConnection,
  ): Promise<void> {
    connection[PRIVATE_RELEASE_METHOD]();
  }
  async destroy(): Promise<void> {
    if (this.#pool) {
      const pool = this.#pool;
      this.#pool = undefined;
      await pool.end();
    }
  }
}

class DenoPostgresDatabaseConnection implements DatabaseConnection {
  constructor(private poolCLient: PoolClient) {}
  async executeQuery<R>(
    compiledQuery: CompiledQuery,
  ): Promise<QueryResult<R>> {
    try {
      await this.poolCLient.connect();
      const query = await this.poolCLient.queryObject<R>(
        compiledQuery.sql,
        compiledQuery.parameters as unknown[],
      );
      if (
        query.command === "INSERT" || query.command === "UPDATE" ||
        query.command === "DELETE"
      ) {
        const numAffectedRows = BigInt(query.rowCount || 0);
        return { rows: query.rows, insertId: numAffectedRows };
      }
      return { rows: query.rows };
    } catch (e) {
      throw e;
    }
  }
  streamQuery<R>(
    compiledQuery: CompiledQuery,
    chunkSize?: number | undefined,
  ): AsyncIterableIterator<QueryResult<R>> {
    throw new Error("Unsupported.");
  }
  [PRIVATE_RELEASE_METHOD](): void {
    this.poolCLient.release();
  }
}

class DenoFileMigrationProvider implements MigrationProvider {
  constructor(private url: URL) {
    if (!url.href.endsWith("/")) {
      this.url = new URL(`${url.href}/`);
    }
  }
  async getMigrations(): Promise<Record<string, Migration>> {
    const migrations: Record<string, Migration> = {};
    for await (let file of Deno.readDir(this.url)) {
      if (
        file.isFile && file.name.endsWith(".js") ||
        (file.name.endsWith(".ts") && !file.name.endsWith(".d.ts")) ||
        file.name.endsWith(".mjs") ||
        (file.name.endsWith(".mts") && !file.name.endsWith(".d.mts"))
      ) {
        const migration = await import(
          new URL(file.name, this.url).href
        );
        const migrationKey = file.name.substring(0, file.name.lastIndexOf("."));

        // Handle esModuleInterop export's `default` prop...
        if (isMigration(migration?.default)) {
          migrations[migrationKey] = migration.default;
        } else if (isMigration(migration)) {
          migrations[migrationKey] = migration;
        }
      }
    }
    return migrations;
  }
}

function isMigration(obj: unknown): obj is Migration {
  return typeof obj === "object" || typeof obj === "function";
}

migrations.md

How to create migrations

First create migrations path with file schema below

import { Kysely, sql } from "https://esm.sh/kysely@0.22.0";

export async function up(db: Kysely<unknown>) {
  await db.schema
    .createTable("users")
    .ifNotExists()
    .addColumn(
      "id",
      "uuid",
      (col) => col.primaryKey().defaultTo(sql`gen_random_uuid()`),
    )
    .addColumn("email", "varchar(256)", (col) => col.notNull().unique())
    .addColumn("emailVerified", "timestamp")
    .addColumn("password", "varchar(256)")
    .addColumn("avatarUrl", "varchar(256)")
    .execute();
}

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

migrator.ts

import "https://deno.land/std@0.170.0/dotenv/load.ts";
import { bold, gray, green, red, yellow } from "https://deno.land/std@0.170.0/fmt/colors.ts";
import { parse } from "https://deno.land/std@0.170.0/flags/mod.ts";
import { MigrationResultSet } from "https://esm.sh/kysely@0.22.0";
import { db } from "./db.ts";
import { createKyPgMigrator } from "./kypg.ts";
import seed from "./_seeder.ts";

const migrator = createKyPgMigrator(
  db,
  // Relative Path to migrations folder
  new URL("./migrations", import.meta.url),
);

const flags = parse(Deno.args, {
  boolean: ["up", "down", "latest", "seed"],
});

if (flags.up) {
  printResult(await migrator.migrateUp());
} else if (flags.down) {
  printResult(await migrator.migrateDown());
} else if (flags.latest) {
  printResult(await migrator.migrateToLatest());
} else if (flags.seed) {
  await seed(db);
} else {
  console.log("No Migration are executed!");
}

function printResult(mig: MigrationResultSet) {
  if (mig.error) {
    console.error(mig.error);
  }
  mig.results?.forEach((result) => {
    let status: string = "";
    switch (result.status) {
      case "Error":
        status = red(result.status);
        break;
      case "Success":
        status = green(result.status);
        break;
      case "NotExecuted":
        status = gray(result.status);
        break;
    }
    console.log(
      `Migrate ${yellow(result.direction)}: ${bold(result.migrationName)}`,
      status,
    );
  });
}