Implement inline projection execution in RDBMS PostgreSQL via PLV8 extenstion

.eslintrc.js

@@ -323,6 +323,7 @@ module.exports = {
           'pid',
           'pingreq',
           'pingresp',
+          'plv8',
           'png',
           'polyfill',
           'polyfills',

.github/workflows/pr-dev.yml

@@ -93,7 +93,7 @@ jobs:
     if: startsWith(github.head_ref, 'feature/') || startsWith(github.head_ref, 'hotfix/') || startsWith(github.head_ref, 'fix/')
     services:
       postgres:
-        image: postgres:10
+        image: clkao/postgres-plv8:10-2
         env:
           POSTGRES_PASSWORD: postgres
         ports:

package.json

@@ -56,8 +56,8 @@
     "@babel/runtime": "7.9.6",
     "@internal/helpers": "0.30.4",
     "@testing-library/dom": "7.21.1",
-    "@typescript-eslint/eslint-plugin": "4.22.1",
-    "@typescript-eslint/parser": "4.22.1",
+    "@typescript-eslint/eslint-plugin": "4.26.0",
+    "@typescript-eslint/parser": "4.26.0",
     "babel-eslint": "10.1.0",
     "babel-jest": "26.0.1",
     "babel-plugin-transform-inline-environment-variables": "0.4.3",

packages/runtime/adapters/readmodel-adapters/readmodel-base/src/index.ts

@@ -12,14 +12,23 @@ import wrapOperation from './wrap-operation'
 import { CreateAdapterMethod } from './types'
 export * from './types'
 
-const createAdapter = _createAdapter.bind(null, {
+const baseAdapterImports = {
   PathToolkit,
   makeSplitNestedPath,
   withPerformanceTracer,
   wrapConnect,
   wrapDisconnect,
   wrapDispose,
   wrapOperation,
-}) as CreateAdapterMethod
+}
+
+const createAdapter = _createAdapter.bind(
+  null,
+  baseAdapterImports
+) as CreateAdapterMethod
+
+const splitNestedPath = makeSplitNestedPath(baseAdapterImports)
 
 export default createAdapter
+
+export { splitNestedPath }

packages/runtime/adapters/readmodel-adapters/readmodel-base/src/types.ts

@@ -283,16 +283,22 @@ export type AdapterOperations<AdapterPool extends CommonAdapterPool> = {
     pool: AdapterPool,
     readModelName: string,
     eventTypes: Array<ReadModelEvent['type']> | null,
-    aggregateIds: Array<ReadModelEvent['aggregateId']> | null
+    aggregateIds: Array<ReadModelEvent['aggregateId']> | null,
+    readModelSource?: string
   ): Promise<void>
 
-  unsubscribe(pool: AdapterPool, readModelName: string): Promise<void>
+  unsubscribe(
+    pool: AdapterPool,
+    readModelName: string,
+    readModelSource?: string
+  ): Promise<void>
 
   resubscribe(
     pool: AdapterPool,
     readModelName: string,
     eventTypes: Array<ReadModelEvent['type']> | null,
-    aggregateIds: Array<ReadModelEvent['aggregateId']> | null
+    aggregateIds: Array<ReadModelEvent['aggregateId']> | null,
+    readModelSource?: string
   ): Promise<void>
 
   resume(
@@ -588,6 +594,30 @@ export type IfEquals<T, U, Y = unknown, N = never> = (<G>() => G extends T
 
 export type IsTypeLike<T, B> = IfEquals<Extract<T, B>, T>
 
+export type MatchTypeConditional<
+  M extends any,
+  V extends Array<[any, any]>,
+  D = never
+> = V extends [[infer A, infer B], ...infer T]
+  ? T extends Array<[any, any]>
+    ? IfEquals<M, A, true, false> extends true
+      ? B
+      : MatchTypeConditional<M, T, D>
+    : D
+  : D
+
+export type MatchTypeConditionalLike<
+  M extends any,
+  V extends Array<[any, any]>,
+  D = never
+> = V extends [[infer A, infer B], ...infer T]
+  ? T extends Array<[any, any]>
+    ? IfEquals<Extract<M, A>, M, true, false> extends true
+      ? B
+      : MatchTypeConditionalLike<M, T, D>
+    : D
+  : D
+
 export type ExtractNewable<F extends NewableLike> = F extends new (
   ...args: infer Args
 ) => infer Result
@@ -605,36 +635,17 @@ export type MakeNewableFunction<F extends FunctionLike> = F extends (
   ...args: infer Args
 ) => infer Result
   ? T extends object
-    ? IfEquals<
+    ? MatchTypeConditionalLike<
         Result,
-        T,
-        new (...args: Args) => T,
-        IfEquals<
-          Result,
-          null,
-          new (...args: Args) => T,
-          IfEquals<
-            Result,
-            undefined,
-            new (...args: Args) => T,
-            IfEquals<
-              Result,
-              boolean,
-              new (...args: Args) => T,
-              IfEquals<
-                Result,
-                string,
-                new (...args: Args) => T,
-                IfEquals<
-                  Result,
-                  number,
-                  new (...args: Args) => T,
-                  IfEquals<Result, void, new (...args: Args) => T, never>
-                >
-              >
-            >
-          >
-        >
+        [
+          [T, new (...args: Args) => T],
+          [null, new (...args: Args) => T],
+          [undefined, new (...args: Args) => T],
+          [boolean, new (...args: Args) => T],
+          [number, new (...args: Args) => T],
+          [string, new (...args: Args) => T],
+          [void, new (...args: Args) => T]
+        ]
       >
     : never
   : never

packages/runtime/adapters/readmodel-adapters/readmodel-postgresql-serverless/src/build.ts

@@ -529,7 +529,8 @@ const build: ExternalMethods['build'] = async (
 
   const now = Date.now()
 
-  const hasSendTime = typeof metricData.sendTime === 'number'
+  const hasSendTime =
+    metricData.sendTime != null && metricData.sendTime.constructor === Number
 
   const groupMonitoring =
     monitoring != null
@@ -539,16 +540,14 @@ const build: ExternalMethods['build'] = async (
       : null
 
   if (hasSendTime) {
-    void [monitoring, groupMonitoring].forEach((innerMonitoring) => {
-      if (innerMonitoring == null) {
-        return
-      }
-
-      innerMonitoring.time('EventDelivery', metricData.sendTime)
-      innerMonitoring.timeEnd('EventDelivery', now)
+    for (const innerMonitoring of [monitoring, groupMonitoring]) {
+      if (innerMonitoring != null) {
+        innerMonitoring.time('EventDelivery', metricData.sendTime)
+        innerMonitoring.timeEnd('EventDelivery', now)
 
-      innerMonitoring.time('EventApply', metricData.sendTime)
-    })
+        innerMonitoring.time('EventApply', metricData.sendTime)
+      }
+    }
   }
 
   const inlineLedgerExecuteStatement: typeof ledgerStatement = Object.assign(
@@ -687,24 +686,25 @@ const build: ExternalMethods['build'] = async (
   } finally {
     basePool.activePassthrough = false
 
-    void [monitoring, groupMonitoring].forEach((innerMonitoring) => {
-      if (innerMonitoring == null) {
-        return
-      }
-
-      if (hasSendTime) {
-        innerMonitoring.timeEnd('EventApply')
-      }
+    for (const innerMonitoring of [monitoring, groupMonitoring]) {
+      if (innerMonitoring != null) {
+        if (hasSendTime) {
+          innerMonitoring.timeEnd('EventApply')
+        }
 
-      innerMonitoring.duration('EventBatchLoad', metricData.eventBatchLoadTime)
+        innerMonitoring.duration(
+          'EventBatchLoad',
+          metricData.eventBatchLoadTime
+        )
 
-      innerMonitoring.duration(
-        'EventProjectionApply',
-        metricData.pureProjectionApplyTime
-      )
+        innerMonitoring.duration(
+          'EventProjectionApply',
+          metricData.pureProjectionApplyTime
+        )
 
-      innerMonitoring.duration('Ledger', metricData.pureLedgerTime)
-    })
+        innerMonitoring.duration('Ledger', metricData.pureLedgerTime)
+      }
+    }
   }
 }
 

packages/runtime/adapters/readmodel-adapters/readmodel-postgresql-serverless/src/types.ts

@@ -8,13 +8,13 @@ import type {
   StoreApi,
   PerformanceTracerLike,
   SplitNestedPathMethod,
+  MatchTypeConditional,
   JsonMap,
   SearchCondition,
   UpdateCondition,
   ObjectFixedKeys,
   OmitObject,
   JsonPrimitive,
-  IfEquals,
 } from '@resolve-js/readmodel-base'
 
 import type RDSDataService from 'aws-sdk/clients/rdsdataservice'
@@ -44,30 +44,25 @@ export type InlineLedgerExecuteTransactionMethodParameters<
 > = [
   pool: AdapterPool,
   method: MethodName,
-  ...args: IfEquals<
+  ...args: MatchTypeConditional<
     MethodName,
-    'begin',
-    [],
-    IfEquals<
-      MethodName,
-      'commit',
-      [transactionId: string],
-      IfEquals<MethodName, 'rollback', [transactionId: string], never>
-    >
+    [
+      ['begin', []],
+      ['commit', [transactionId: string]],
+      ['rollback', [transactionId: string]]
+    ]
   >
 ]
+
 export type InlineLedgerExecuteTransactionMethodReturnType<
   MethodName extends InlineLedgerExecuteTransactionMethodNames
-> = IfEquals<
+> = MatchTypeConditional<
   MethodName,
-  'begin',
-  string,
-  IfEquals<
-    MethodName,
-    'commit',
-    null | undefined,
-    IfEquals<MethodName, 'rollback', null | undefined, never>
-  >
+  [
+    ['begin', string],
+    ['commit', null | undefined],
+    ['rollback', null | undefined]
+  ]
 >
 
 export type InlineLedgerExecuteTransactionMethod = <