From a4cb012ce8c9979e30e5a8ed02f3ba0564f5ec13 Mon Sep 17 00:00:00 2001
From: dswbx <dennis.senn@gmx.ch>
Date: Fri, 7 Mar 2025 19:40:58 +0100
Subject: [PATCH] added batching for postgres

---
 app/__test__/data/specs/connection/pg.spec.ts |  7 ++++-
 app/src/core/utils/numbers.ts                 |  8 +++++
 app/src/data/connection/Connection.ts         |  1 +
 .../connection/postgres/PostgresConnection.ts | 31 +++++++++++++++++--
 .../connection/sqlite/SqliteConnection.ts     |  4 +++
 app/src/data/entities/query/Repository.ts     |  6 ++--
 app/src/data/schema/SchemaManager.ts          |  1 +
 7 files changed, 53 insertions(+), 5 deletions(-)

diff --git a/app/__test__/data/specs/connection/pg.spec.ts b/app/__test__/data/specs/connection/pg.spec.ts
index 2f0fbbc..015272d 100644
--- a/app/__test__/data/specs/connection/pg.spec.ts
+++ b/app/__test__/data/specs/connection/pg.spec.ts
@@ -21,7 +21,7 @@ describe.skipIf(ALL_TESTS)("postgres", () => {
          plugins: [new ParseJSONResultsPlugin()],
       });
 
-      console.log(await introspector.getSchema());
+      console.log(await introspector.getSchemaSpec());
    });
 
    test("builds", async () => {
@@ -50,6 +50,11 @@ describe.skipIf(ALL_TESTS)("postgres", () => {
 
       await app.build({ sync: true });
 
+      /*await app.em
+         .mutator("posts")
+         .insertMany([{ title: "hello world" }, { title: "hello world 2" }]);*/
+
       expect(app.version()).toBeDefined();
+      console.log(await app.em.repo("posts").findMany());
    });
 });
diff --git a/app/src/core/utils/numbers.ts b/app/src/core/utils/numbers.ts
index 1435f68..33394f6 100644
--- a/app/src/core/utils/numbers.ts
+++ b/app/src/core/utils/numbers.ts
@@ -3,3 +3,11 @@ export function clampNumber(value: number, min: number, max: number): number {
    const upper = Math.max(min, max);
    return Math.max(lower, Math.min(value, upper));
 }
+
+export function ensureInt(value?: string | number | null | undefined): number {
+   if (value === undefined || value === null) {
+      return 0;
+   }
+
+   return typeof value === "number" ? value : Number.parseInt(value, 10);
+}
diff --git a/app/src/data/connection/Connection.ts b/app/src/data/connection/Connection.ts
index c1ea50a..5757511 100644
--- a/app/src/data/connection/Connection.ts
+++ b/app/src/data/connection/Connection.ts
@@ -151,4 +151,5 @@ export abstract class Connection<DB = any> {
    }
 
    abstract getFieldSchema(spec: FieldSpec, strict?: boolean): SchemaResponse;
+   abstract close(): Promise<void>;
 }
diff --git a/app/src/data/connection/postgres/PostgresConnection.ts b/app/src/data/connection/postgres/PostgresConnection.ts
index 812ebee..0479369 100644
--- a/app/src/data/connection/postgres/PostgresConnection.ts
+++ b/app/src/data/connection/postgres/PostgresConnection.ts
@@ -8,7 +8,12 @@ import {
 } from "kysely";
 import pg from "pg";
 import { PostgresIntrospector } from "./PostgresIntrospector";
-import { type FieldSpec, type SchemaResponse, Connection } from "data/connection/Connection";
+import {
+   type FieldSpec,
+   type SchemaResponse,
+   Connection,
+   type QB,
+} from "data/connection/Connection";
 
 export type PostgresConnectionConfig = pg.PoolConfig;
 
@@ -21,16 +26,20 @@ class CustomPostgresDialect extends PostgresDialect {
 }
 
 export class PostgresConnection extends Connection {
+   private pool: pg.Pool;
+
    constructor(config: PostgresConnectionConfig) {
+      const pool = new pg.Pool(config);
       const kysely = new Kysely({
          dialect: new CustomPostgresDialect({
-            pool: new pg.Pool(config),
+            pool,
          }),
          plugins,
          //log: ["query", "error"],
       });
 
       super(kysely, {}, plugins);
+      this.pool = pool;
    }
 
    override supportsIndices(): boolean {
@@ -73,4 +82,22 @@ export class PostgresConnection extends Connection {
          },
       ];
    }
+
+   override supportsBatching(): boolean {
+      return true;
+   }
+
+   override async close(): Promise<void> {
+      await this.pool.end();
+   }
+
+   protected override async batch<Queries extends QB[]>(
+      queries: [...Queries],
+   ): Promise<{
+      [K in keyof Queries]: Awaited<ReturnType<Queries[K]["execute"]>>;
+   }> {
+      return this.kysely.transaction().execute(async (trx) => {
+         return Promise.all(queries.map((q) => trx.executeQuery(q).then((r) => r.rows)));
+      }) as any;
+   }
 }
diff --git a/app/src/data/connection/sqlite/SqliteConnection.ts b/app/src/data/connection/sqlite/SqliteConnection.ts
index ea24320..9ef9091 100644
--- a/app/src/data/connection/sqlite/SqliteConnection.ts
+++ b/app/src/data/connection/sqlite/SqliteConnection.ts
@@ -47,4 +47,8 @@ export class SqliteConnection extends Connection {
          },
       ] as const;
    }
+
+   override async close(): Promise<void> {
+      // no-op
+   }
 }
diff --git a/app/src/data/entities/query/Repository.ts b/app/src/data/entities/query/Repository.ts
index 4734ddf..47657f8 100644
--- a/app/src/data/entities/query/Repository.ts
+++ b/app/src/data/entities/query/Repository.ts
@@ -14,6 +14,7 @@ import {
    WithBuilder,
 } from "../index";
 import { JoinBuilder } from "./JoinBuilder";
+import { ensureInt } from "core/utils";
 
 export type RepositoryQB = SelectQueryBuilder<any, any, any>;
 
@@ -225,8 +226,9 @@ export class Repository<TBD extends object = DefaultDB, TB extends keyof TBD = a
             data,
             meta: {
                ...payload.meta,
-               total: _total[0]?.total ?? 0,
-               count: _count[0]?.count ?? 0, // @todo: better graceful method
+               // parsing is important since pg returns string
+               total: ensureInt(_total[0]?.total),
+               count: ensureInt(_count[0]?.count),
                items: result.length,
                time,
             },
diff --git a/app/src/data/schema/SchemaManager.ts b/app/src/data/schema/SchemaManager.ts
index d0cf2b2..a8d9df0 100644
--- a/app/src/data/schema/SchemaManager.ts
+++ b/app/src/data/schema/SchemaManager.ts
@@ -329,6 +329,7 @@ export class SchemaManager {
          if (local_updates === 0) continue;
 
          // iterate through built qbs
+         // @todo: run in batches
          for (const qb of qbs) {
             const { sql, parameters } = qb.compile();
             statements.push({ sql, parameters });