Fix: 'column XMIN does not exist' exception;fix: live queries not honouring query parameters; improve live queris' dedupe boundary to: SQL text + query params + transaction context; chore: internal refactors and other fixes

Author: ox-harris

README.md

@@ -361,7 +361,7 @@ It lets you declare remote data as local tables (views):
 await db.query(`
   CREATE VIEW users AS
   SELECT * FROM users
-  WITH (replication_origin = '/api/db');
+  WITH (replication_origin = 'postgres:/api/db');
 `);
 ```
 
@@ -383,7 +383,7 @@ With just an extra keyword, you get **automatic sync between local and remote st
 await db.query(`
   CREATE REALTIME VIEW users AS
   SELECT * FROM users
-  WITH (replication_origin = '/api/db');
+  WITH (replication_origin = 'postgres:/api/db');
 `);
 ```
 

…/clients/abstracts/MainstreamDBClient.js src/clients/abstracts/MainstreamClient.jssrc/clients/abstracts/MainstreamDBClient.js renamed to src/clients/abstracts/MainstreamClient.js src/clients/abstracts/MainstreamDBClient.js renamed to src/clients/abstracts/MainstreamClient.js

@@ -1,26 +1,23 @@
 import { LinkedQLClient } from './LinkedQLClient.js';
 import { RealtimeClient } from '../../proc/realtime/RealtimeClient.js';
 import { MainstreamSchemaInference } from './MainstreamSchemaInference.js';
-import { MainstreamWalEngine } from './MainstreamWalEngine.js';
 import { SQLParser } from '../../lang/SQLParser.js';
 import { registry } from '../../lang/registry.js';
 import { normalizeQueryArgs } from './util.js';
 import { Result } from '../Result.js';
 
-export class MainstreamDBClient extends LinkedQLClient {
+export class MainstreamClient extends LinkedQLClient {
 
-    // Standard getters: parsers, resolver, wal
+    // Standard getters: parsers, resolver
 
     #parser;
-    #wal;
     #live;
 
     get parser() { return this.#parser; }
     get resolver() {
         return super.resolveGetResolver(() =>
-            new MainstreamSchemaInference({ client: this, dialect: this.dialect }));
+            new MainstreamSchemaInference({ mainstreamClient: this, dialect: this.dialect }));
     }
-    get wal() { return this.#wal; }
     get live() { return this.#live; }
 
     // Internal
@@ -33,64 +30,54 @@ export class MainstreamDBClient extends LinkedQLClient {
         super(options);
 
         this.#parser = new SQLParser({ dialect: this.dialect });
-        this.#wal = new MainstreamWalEngine({
-            client: this,
-            drainMode: 'drain',
-            lifecycleHook: async (status) => {
-                await this.setCapability({ realtime: !!status });
-            }
-        });
 
         this.#realtimeClient = new RealtimeClient(this);
         this.#live = {
             forget: async (id) => await this.#realtimeClient.forget(id),
         };
     }
 
-    async disconnect() {
-        await this.#wal.close({ destroy: true });
-        await super.disconnect();
-    }
-
     // ------------
 
-    async transaction(cb) {
+    async begin(options = {}) {
+        if (options.parentTx) {
+            throw new Error(`Nested transactions are not supported on mainstream databasew for now`);
+        }
+        return await this._begin(options);
+    }
+
+    async transaction(cb, options = {}) {
         if (typeof cb !== 'function') {
             throw new TypeError('transaction(cb): cb must be a function');
         }
-        if (typeof this._beginTransaction !== 'function'
-            || typeof this._commitTransaction !== 'function'
-            || typeof this._rollbackTransaction !== 'function') {
-            throw new Error('Transaction not supported by this client implementation');
-        }
-
-        const tx = await this._beginTransaction();
+        
+        const tx = await this.begin(options);
 
         try {
             const result = await cb(tx);
-            await this._commitTransaction(tx);
+            await tx.commit();
             return result;
         } catch (e) {
-            await this._rollbackTransaction(tx);
+            await tx.rollback();
             throw e;
         }
     }
 
     // ------------
 
     async query(...args) {
-        const [_query, options] = normalizeQueryArgs(...args);
+        const [_query, { tx: inputTx, ...options }] = normalizeQueryArgs(...args);
         const query = await this.#parser.parse(_query, options);
 
-        const resolveQuery = async (query, tx = null, ifHasSugars = false) => {
+        const resolveQuery = async (query, tx, ifHasSugars = false) => {
             const schemaInference = this.resolver;
             return await schemaInference.resolveQuery(query, { tx, ifHasSugars });
         };
 
         // Realtime query?
         if (options.live) {
-            const resolvedQuery = await resolveQuery(query);
-            return await this.#realtimeClient.query(resolvedQuery, options);
+            const resolvedQuery = await resolveQuery(query, inputTx);
+            return await this.#realtimeClient.query(resolvedQuery, { ...options, tx: inputTx });
         }
 
         let result;
@@ -108,22 +95,25 @@ export class MainstreamDBClient extends LinkedQLClient {
                         { ...options, tx }
                     );
                 }
-            });
+            }, { parentTx: inputTx });
         } else {
-            result = await this._query(await resolveQuery(query, null, true), options);
+            result = await this._query(
+                await resolveQuery(query, inputTx, true),
+                { ...options, tx: inputTx }
+            );
         }
 
         // The result instance
         return new Result({ rows: result.rows, rowCount: result.rowCount });
     }
 
     async stream(...args) {
-        const [_query, options] = normalizeQueryArgs(...args);
+        const [_query, { tx: inputTx, ...options }] = normalizeQueryArgs(...args);
         const query = await this.#parser.parse(_query, options);
 
         const schemaInference = this.resolver;
-        const resolvedQuery = await schemaInference.resolveQuery(query, { ...options, ifHasSugars: true });
+        const resolvedQuery = await schemaInference.resolveQuery(query, { ...options, tx: inputTx, ifHasSugars: true });
 
-        return await this._stream(resolvedQuery, options);
+        return await this._stream(resolvedQuery, { ...options, tx: inputTx });
     }
 }

src/clients/abstracts/MainstreamSchemaInference.js

@@ -1,21 +1,20 @@
-import { SchemaInference as BaseSchemaInference } from '../../lang/SchemaInference.js';
+import { SchemaInference } from '../../lang/SchemaInference.js';
 import { normalizeRelationSelectorArg, parseRelationSelectors } from './util.js';
 import { registry } from '../../lang/registry.js';
 
-export class MainstreamSchemaInference extends BaseSchemaInference {
+export class MainstreamSchemaInference extends SchemaInference {
 
-    #client;
-    get client() { return this.#client; }
+    #mainstreamClient;
 
-    constructor({ client, ...options }) {
+    constructor({ mainstreamClient, ...options }) {
         super(options);
-        this.#client = client;
+        this.#mainstreamClient = mainstreamClient;
     }
 
     async showCreate(selector, { structured = false, tx = null } = {}) {
         selector = normalizeRelationSelectorArg(selector);
         const sql = this.#composeShowCreateSQL(selector);
-        const result = await this.#client._query(sql, { tx }); // Must be _query() not query()
+        const result = await this.#mainstreamClient._query(sql, { tx }); // Must be _query() not query()
         return await this.#formatShowCreateResult(result.rows, structured);
     }
 

…clients/abstracts/MainstreamWalEngine.js src/clients/abstracts/MainstreamWal.jssrc/clients/abstracts/MainstreamWalEngine.js renamed to src/clients/abstracts/MainstreamWal.js src/clients/abstracts/MainstreamWalEngine.js renamed to src/clients/abstracts/MainstreamWal.js

@@ -1,14 +1,23 @@
-import { WalEngine as BaseWalEngine } from '../../proc/timeline/WalEngine.js';
+import { LinkedQlWal } from '../../proc/timeline/LinkedQlWal.js';
 import { ConflictError } from '../../flashql/errors/ConflictError.js';
 
-export class MainstreamWalEngine extends BaseWalEngine {
+export class MainstreamWal extends LinkedQlWal {
 
-    #client;
-    get client() { return this.#client; }
+    #mainstreamClient;
 
-    constructor({ client, ...options }) {
-        super(options);
-        this.#client = client;
+    constructor({ mainstreamClient, ...options }) {
+        super({
+            ...options,
+            drainMode: 'drain',
+            lifecycleHook: async (status) => {
+                if (status) {
+                    await this._setupRealtime();
+                } else {
+                    await this._teardownRealtime();
+                }
+            }
+        });
+        this.#mainstreamClient = mainstreamClient;
     }
 
     _quoteIdent(name) {
@@ -42,8 +51,8 @@ export class MainstreamWalEngine extends BaseWalEngine {
         if (mvccKey) {
             if (!event.mvccTag)
                 throw new TypeError(`Missing event.mvccTag for the specified mvccKey ${mvccKey}`);
-            const mvccExpr = this.#client.dialect === 'postgres' && mvccKey.toUpperCase() === 'XMIN'
-                ? 'CAST(CAST(xmin AS TEXT) AS BIGINT)'
+            const mvccExpr = this.#mainstreamClient.dialect === 'postgres' && mvccKey.toUpperCase() === 'XMIN'
+                ? 'CAST(xmin AS TEXT)'
                 : this._quoteIdent(mvccKey);
             sql += ` AND ${mvccExpr} = ${this._serializeValue(event.mvccTag)}`;
         }
@@ -82,15 +91,15 @@ export class MainstreamWalEngine extends BaseWalEngine {
 
                 if (!sql) continue;
 
-                const result = await this.#client._query(sql, { tx });
+                const result = await this.#mainstreamClient._query(sql, { tx });
                 if ((op === 'update' || op === 'delete') && result?.rowCount === 0) {
                     throw new ConflictError(`[${this._quoteQualifiedRelation(relation)}] Origin row version no longer matches the expected version`);
                 }
             }
         };
 
-        if (typeof this.#client.transaction === 'function') {
-            return await this.#client.transaction(async (tx) => {
+        if (typeof this.#mainstreamClient.transaction === 'function') {
+            return await this.#mainstreamClient.transaction(async (tx) => {
                 await applyCommit(tx);
             });
         }