Added Dumbo - utilities for PostgreSQL handling

That will help to keep the shared code between Pongo and Emmett in one place.

Dumbo will be versioned separately, which should give possibility to use different versions in Pongo and Emmett. And have this peer dependency more stable.

Dumbo is treated for now as internal Pongo and Emmett dependency, so there are no plans (for now) to provide docs for it.

src/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@event-driven-io/pongo-core",
-  "version": "0.2.4",
+  "version": "0.3.0",
   "description": "Pongo - Mongo with strong consistency on top of Postgres",
   "type": "module",
   "engines": {
@@ -98,6 +98,7 @@
     "pg-format": "^1.0.4"
   },
   "workspaces": [
+    "packages/dumbo",
     "packages/pongo"
   ]
 }

src/packages/dumbo/package.json

@@ -0,0 +1,60 @@
+{
+  "name": "@event-driven-io/dumbo",
+  "version": "0.1.0",
+  "description": "Dumbo - tools for dealing with PostgreSQL",
+  "type": "module",
+  "scripts": {
+    "build": "tsup",
+    "build:ts": "tsc",
+    "build:ts:watch": "tsc  --watch",
+    "test": "run-s test:unit test:int test:e2e",
+    "test:unit": "glob -c \"node --import tsx --test\" **/*.unit.spec.ts",
+    "test:int": "glob -c \"node --import tsx --test\" **/*.int.spec.ts",
+    "test:e2e": "glob -c \"node --import tsx --test\" **/*.e2e.spec.ts",
+    "test:watch": "node --import tsx --test --watch",
+    "test:unit:watch": "glob -c \"node --import tsx --test --watch\" **/*.unit.spec.ts",
+    "test:int:watch": "glob -c \"node --import tsx --test --watch\" **/*.int.spec.ts",
+    "test:e2e:watch": "glob -c \"node --import tsx --test --watch\" **/*.e2e.spec.ts"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/event-driven-io/Pongo.git"
+  },
+  "keywords": [
+    "Event Sourcing"
+  ],
+  "author": "Oskar Dudycz",
+  "bugs": {
+    "url": "https://github.com/event-driven-io/Pongo/issues"
+  },
+  "homepage": "https://event-driven-io.github.io/Pongo/",
+  "exports": {
+    ".": {
+      "import": {
+        "types": "./dist/index.d.ts",
+        "default": "./dist/index.js"
+      },
+      "require": {
+        "types": "./dist/index.d.cts",
+        "default": "./dist/index.cjs"
+      }
+    }
+  },
+  "main": "./dist/index.cjs",
+  "module": "./dist/index.js",
+  "types": "./dist/index.d.ts",
+  "files": [
+    "dist"
+  ],
+  "peerDependencies": {
+    "@types/uuid": "^9.0.8",
+    "@types/pg": "^8.11.6",
+    "@types/pg-format": "^1.0.5",
+    "pg": "^8.12.0",
+    "pg-format": "^1.0.4",
+    "uuid": "^9.0.1"
+  },
+  "devDependencies": {
+    "@types/node": "20.11.30"
+  }
+}

src/packages/dumbo/src/connections/client.ts

@@ -0,0 +1,19 @@
+import pg from 'pg';
+import { endPool, getPool } from './pool';
+
+export interface PostgresClient {
+  connect(): Promise<pg.PoolClient>;
+  close(): Promise<void>;
+}
+
+export const postgresClient = (
+  connectionString: string,
+  database?: string,
+): PostgresClient => {
+  const pool = getPool({ connectionString, database });
+
+  return {
+    connect: () => pool.connect(),
+    close: () => endPool(connectionString),
+  };
+};

src/packages/dumbo/src/connections/index.ts

@@ -0,0 +1,2 @@
+export * from './client';
+export * from './pool';

src/packages/dumbo/src/execute/index.ts

@@ -0,0 +1,158 @@
+import type pg from 'pg';
+import type { SQL } from '../sql';
+
+export const execute = async <Result = void>(
+  pool: pg.Pool,
+  handle: (client: pg.PoolClient) => Promise<Result>,
+) => {
+  const client = await pool.connect();
+  try {
+    return await handle(client);
+  } finally {
+    client.release();
+  }
+};
+
+export const executeInTransaction = async <Result = void>(
+  pool: pg.Pool,
+  handle: (
+    client: pg.PoolClient,
+  ) => Promise<{ success: boolean; result: Result }>,
+): Promise<Result> =>
+  execute(pool, async (client) => {
+    try {
+      await client.query('BEGIN');
+
+      const { success, result } = await handle(client);
+
+      if (success) await client.query('COMMIT');
+      else await client.query('ROLLBACK');
+
+      return result;
+    } catch (e) {
+      await client.query('ROLLBACK');
+      throw e;
+    }
+  });
+
+export const executeSQL = async <
+  Result extends pg.QueryResultRow = pg.QueryResultRow,
+>(
+  poolOrClient: pg.Pool | pg.PoolClient,
+  sql: SQL,
+): Promise<pg.QueryResult<Result>> =>
+  'totalCount' in poolOrClient
+    ? execute(poolOrClient, (client) => client.query<Result>(sql))
+    : poolOrClient.query<Result>(sql);
+
+export const executeSQLInTransaction = async <
+  Result extends pg.QueryResultRow = pg.QueryResultRow,
+>(
+  pool: pg.Pool,
+  sql: SQL,
+) => {
+  console.log(sql);
+  return executeInTransaction(pool, async (client) => ({
+    success: true,
+    result: await client.query<Result>(sql),
+  }));
+};
+
+export const executeSQLBatchInTransaction = async <
+  Result extends pg.QueryResultRow = pg.QueryResultRow,
+>(
+  pool: pg.Pool,
+  ...sqls: SQL[]
+) =>
+  executeInTransaction(pool, async (client) => {
+    for (const sql of sqls) {
+      await client.query<Result>(sql);
+    }
+
+    return { success: true, result: undefined };
+  });
+
+export const firstOrNull = async <
+  Result extends pg.QueryResultRow = pg.QueryResultRow,
+>(
+  getResult: Promise<pg.QueryResult<Result>>,
+): Promise<Result | null> => {
+  const result = await getResult;
+
+  return result.rows.length > 0 ? result.rows[0] ?? null : null;
+};
+
+export const first = async <
+  Result extends pg.QueryResultRow = pg.QueryResultRow,
+>(
+  getResult: Promise<pg.QueryResult<Result>>,
+): Promise<Result> => {
+  const result = await getResult;
+
+  if (result.rows.length === 0)
+    throw new Error("Query didn't return any result");
+
+  return result.rows[0]!;
+};
+
+export const singleOrNull = async <
+  Result extends pg.QueryResultRow = pg.QueryResultRow,
+>(
+  getResult: Promise<pg.QueryResult<Result>>,
+): Promise<Result | null> => {
+  const result = await getResult;
+
+  if (result.rows.length > 1) throw new Error('Query had more than one result');
+
+  return result.rows.length > 0 ? result.rows[0] ?? null : null;
+};
+
+export const single = async <
+  Result extends pg.QueryResultRow = pg.QueryResultRow,
+>(
+  getResult: Promise<pg.QueryResult<Result>>,
+): Promise<Result> => {
+  const result = await getResult;
+
+  if (result.rows.length === 0)
+    throw new Error("Query didn't return any result");
+
+  if (result.rows.length > 1) throw new Error('Query had more than one result');
+
+  return result.rows[0]!;
+};
+
+export const mapRows = async <
+  Result extends pg.QueryResultRow = pg.QueryResultRow,
+  Mapped = unknown,
+>(
+  getResult: Promise<pg.QueryResult<Result>>,
+  map: (row: Result) => Mapped,
+): Promise<Mapped[]> => {
+  const result = await getResult;
+
+  return result.rows.map(map);
+};
+
+export const toCamelCase = (snakeStr: string): string =>
+  snakeStr.replace(/_([a-z])/g, (g) => g[1]?.toUpperCase() ?? '');
+
+export const mapToCamelCase = <T extends Record<string, unknown>>(
+  obj: T,
+): T => {
+  const newObj: Record<string, unknown> = {};
+  for (const key in obj) {
+    if (Object.prototype.hasOwnProperty.call(obj, key)) {
+      newObj[toCamelCase(key)] = obj[key];
+    }
+  }
+  return newObj as T;
+};
+
+export type ExistsSQLQueryResult = { exists: boolean };
+
+export const exists = async (pool: pg.Pool, sql: SQL): Promise<boolean> => {
+  const result = await single(executeSQL<ExistsSQLQueryResult>(pool, sql));
+
+  return result.exists === true;
+};

src/packages/dumbo/src/index.ts

@@ -0,0 +1,3 @@
+export * from './connections';
+export * from './execute';
+export * from './sql';

src/packages/dumbo/src/sql/schema.ts

@@ -0,0 +1,36 @@
+import pg from 'pg';
+import { sql, type SQL } from '.';
+import { exists } from '../execute';
+export * from './schema';
+
+export const tableExistsSQL = (tableName: string): SQL =>
+  sql(
+    `
+  SELECT EXISTS (
+    SELECT FROM pg_tables
+    WHERE tablename = %L
+  ) AS exists;`,
+    tableName,
+  );
+
+export const tableExists = async (
+  pool: pg.Pool,
+  tableName: string,
+): Promise<boolean> => exists(pool, tableExistsSQL(tableName));
+
+export const functionExistsSQL = (functionName: string): SQL =>
+  sql(
+    `
+      SELECT EXISTS (
+        SELECT FROM pg_proc 
+        WHERE 
+        proname = %L
+      ) AS exists;
+    `,
+    functionName,
+  );
+
+export const functionExists = async (
+  pool: pg.Pool,
+  functionName: string,
+): Promise<boolean> => exists(pool, functionExistsSQL(functionName));

src/packages/dumbo/tsconfig.build.json

@@ -0,0 +1,6 @@
+{
+  "extends": "./tsconfig.json",
+  "compilerOptions": {
+    "composite": false
+  }
+}

src/packages/dumbo/tsconfig.eslint.json

@@ -0,0 +1,3 @@
+{
+  "extends": "./tsconfig.json"
+}

src/packages/dumbo/tsconfig.json

@@ -0,0 +1,11 @@
+{
+  "extends": "../../tsconfig.shared.json",
+  "include": ["./src/**/*"],
+  "compilerOptions": {
+    "composite": true,
+    "outDir": "./dist" /* Redirect output structure to the directory. */,
+    "rootDir": "./src",
+    "paths": {}
+  },
+  "references": []
+}