Reduce complexity in PostgreSQL observation manager and source type resolver

Source/Drizzle/for_PostgreSQLObservationManager/when_acquiring/with_trigger_variants.ts

@@ -0,0 +1,57 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+import { database, Listener, managerFor, table } from '../given/a_manager.js';
+
+describe('when acquiring with an incompatible PostgreSQL trigger', () => {
+    for (const [reason, changes] of [
+        ['disabled', { tgenabled: 'D' }],
+        ['events', { tgtype: 61 }],
+        ['events', { no_predicate: false }],
+        ['events', { attributes: '1' }],
+        ['events', { tgisinternal: true }],
+        ['events', { constraint_oid: '2' }],
+        ['function version', { proname: 'old_notify' }],
+        ['function version', { function_schema: 'other' }],
+        ['function version', { pronargs: 1 }],
+        ['function version', { returns_trigger: false }],
+        ['function version', { lanname: 'sql' }],
+        ['channel', { tgnargs: 2 }],
+        ['channel', { arguments: '' }]
+    ] as [string, Record<string, unknown>][]) {
+        it(`should reject ${Object.keys(changes)[0]} as ${reason}`, async () => {
+            const listener = new Listener();
+            const query = listener.query.bind(listener);
+            listener.query = async statement => {
+                const result = await query(statement);
+                return statement.includes('pg_trigger') ? { rows: [{ ...result.rows[0], ...changes }] } : result;
+            };
+            const manager = managerFor(() => listener);
+            const lease = manager.acquire('tenant', database, table, () => {}, () => {});
+            try {
+                await lease.ready.should.be.rejectedWith(`PostgreSQL observation trigger invalid for 'app.tasks': ${reason}; apply postgresqlChangeTrigger(...) through an application migration`);
+                listener.closeCount.should.equal(1);
+            } finally { lease.release(); await manager[Symbol.asyncDispose](); }
+        });
+    }
+
+    it('should accept an always enabled trigger and release its routing only once', async () => {
+        const listener = new Listener();
+        const query = listener.query.bind(listener);
+        listener.query = async statement => {
+            const result = await query(statement);
+            return statement.includes('pg_trigger') ? { rows: [{ ...result.rows[0], tgenabled: 'A' }] } : result;
+        };
+        const manager = managerFor(() => listener);
+        const changes: string[] = [];
+        const first = manager.acquire('tenant', database, table, () => changes.push('first'), () => {});
+        const second = manager.acquire('tenant', database, table, () => changes.push('second'), () => {});
+        try {
+            await Promise.all([first.ready, second.ready]);
+            first.release();
+            first.release();
+            listener.notification?.('arc_changes', 'app.tasks');
+            changes.should.deep.equal(['second']);
+            listener.closeCount.should.equal(0);
+        } finally { first.release(); second.release(); await manager[Symbol.asyncDispose](); }
+    });
+});

Source/Drizzle/for_PostgreSQLObservationManager/when_recovering/with_listener_database_failures.ts

@@ -0,0 +1,50 @@
+// Copyright (c) Cratis. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+import { vi } from 'vitest';
+import { DrizzleObservation } from '../../DrizzleObservation.js';
+import { PostgreSQLObservationManager } from '../../PostgreSQLObservationManager.js';
+import { database, Listener, table } from '../given/a_manager.js';
+
+describe('when verifying the listener database during PostgreSQL recovery', () => {
+    it('should spend a retry on a query failure and catch up on the next listener', async () => {
+        const listeners = [new Listener(), new Listener(), new Listener()];
+        const query = listeners[1]!.query.bind(listeners[1]);
+        listeners[1]!.query = async statement => {
+            if (statement.includes('current_database')) throw new Error('connection unavailable');
+            return query(statement);
+        };
+        let connections = 0;
+        const manager = new PostgreSQLObservationManager({ mode: DrizzleObservation.PostgreSQLNotify,
+            listener: () => listeners[connections++]! }, 60_000, 100, [1, 1]);
+        const changes: boolean[] = [];
+        const failures: Error[] = [];
+        const lease = manager.acquire('tenant', database, table, forced => changes.push(forced ?? false), error => failures.push(error));
+        try {
+            await lease.ready;
+            listeners[0]!.disconnect?.();
+            await vi.waitFor(() => changes.should.deep.equal([true]));
+            connections.should.equal(3);
+            failures.should.have.lengthOf(0);
+            listeners[1]!.closeCount.should.equal(1);
+        } finally { lease.release(); await manager[Symbol.asyncDispose](); }
+    });
+
+    it('should fail the lease definitively when the listener database differs', async () => {
+        const first = new Listener();
+        const next = new Listener();
+        const query = next.query.bind(next);
+        next.query = async statement => statement.includes('current_database') ? { rows: [{ database: 'other' }] } : query(statement);
+        let connections = 0;
+        const manager = new PostgreSQLObservationManager({ mode: DrizzleObservation.PostgreSQLNotify,
+            listener: () => ++connections === 1 ? first : next }, 60_000, 100, [1]);
+        const failures: Error[] = [];
+        const lease = manager.acquire('tenant', database, table, () => {}, error => failures.push(error));
+        try {
+            await lease.ready;
+            first.disconnect?.();
+            await vi.waitFor(() => failures.should.have.lengthOf(1));
+            failures[0]!.message.should.equal("PostgreSQL observation database/table mapping differs between reader and listener for tenant 'tenant'");
+            connections.should.equal(2);
+        } finally { lease.release(); await manager[Symbol.asyncDispose](); }
+    });
+});