feat(condo): INFRA-202 multiple database sources

apps/condo/domains/analytics/utils/serverSchema/analytics.helper.js

@@ -3,6 +3,7 @@ const isoWeek = require('dayjs/plugin/isoWeek')
 const get = require('lodash/get')
 const groupBy = require('lodash/groupBy')
 
+const { getDatabaseAdapter } = require('@open-condo/keystone/databaseAdapters/utils')
 const { getSchemaCtx } = require('@open-condo/keystone/schema')
 
 const { GqlToKnexBaseAdapter } = require('@condo/domains/common/utils/serverSchema/GqlToKnexBaseAdapter')
@@ -47,7 +48,7 @@ class TicketGqlToKnexAdapter extends GqlToKnexBaseAdapter {
     async loadData () {
         this.result = null
         const { keystone } = await getSchemaCtx(this.domainName)
-        const knex = keystone.adapter.knex
+        const { knex } = getDatabaseAdapter(keystone)
 
         this.whereIn = Object.fromEntries(this.whereIn)
         // create whereIn structure [['property_id', 'user_id'], [['some_property_id', 'some_user_id'], ...]]

apps/condo/domains/analytics/utils/services/dataLoaders/incident.js

@@ -1,5 +1,6 @@
 const { get, omit, find } = require('lodash')
 
+const { getDatabaseAdapter } = require('@open-condo/keystone/databaseAdapters/utils')
 const { getSchemaCtx } = require('@open-condo/keystone/schema')
 
 const { AbstractDataLoader } = require('@condo/domains/analytics/utils/services/dataLoaders/AbstractDataLoader')
@@ -14,8 +15,7 @@ class IncidentPropertyGqlKnexLoader extends GqlToKnexBaseAdapter {
 
     async loadData () {
         const { keystone } = await getSchemaCtx(this.domainName)
-        const knex = keystone.adapter.knex
-
+        const { knex } = getDatabaseAdapter(keystone)
 
         const propertyFilter = get(find(this.where, 'property', {}), 'property.id_in', [])
         const incidentFilter = get(find(this.where, 'incident', {}), 'incident.id_in', [])

apps/condo/domains/analytics/utils/services/dataLoaders/payment.js

@@ -1,6 +1,7 @@
 const Big = require('big.js')
 const { get, pick, isEmpty } = require('lodash')
 
+const { getDatabaseAdapter } = require('@open-condo/keystone/databaseAdapters/utils')
 const { getSchemaCtx } = require('@open-condo/keystone/schema')
 
 const { PAYMENT_DONE_STATUS, PAYMENT_WITHDRAWN_STATUS } = require('@condo/domains/acquiring/constants/payment')
@@ -16,7 +17,7 @@ class BillingResidentKnexLoader extends GqlToKnexBaseAdapter {
 
     async loadData () {
         const { keystone } = await getSchemaCtx(this.domainName)
-        const knex = keystone.adapter.knex
+        const { knex } = getDatabaseAdapter(keystone)
 
         const propertyIds = get(this.where, 'property.id_in', [])
 
@@ -55,7 +56,7 @@ class PaymentGqlKnexLoader extends GqlToKnexBaseAdapter {
     async loadData () {
         this.result = null
         const { keystone } = await getSchemaCtx(this.domainName)
-        const knex = keystone.adapter.knex
+        const { knex } = getDatabaseAdapter(keystone)
 
         this.extendAggregationWithFilter(this.aggregateBy)
 

apps/condo/domains/analytics/utils/services/dataLoaders/receipt.js

@@ -2,6 +2,7 @@ const Big = require('big.js')
 const dayjs = require('dayjs')
 const { get, omit } = require('lodash')
 
+const { getDatabaseAdapter } = require('@open-condo/keystone/databaseAdapters/utils')
 const { getSchemaCtx } = require('@open-condo/keystone/schema')
 
 const { AbstractDataLoader } = require('@condo/domains/analytics/utils/services/dataLoaders/AbstractDataLoader')
@@ -23,7 +24,7 @@ class ReceiptGqlKnexLoader extends GqlToKnexBaseAdapter {
         this.result = null
 
         const { keystone } = await getSchemaCtx(this.domainName)
-        const knex = keystone.adapter.knex
+        const { knex } = getDatabaseAdapter(keystone)
 
         this.whereIn = {}
 

apps/condo/domains/analytics/utils/services/dataLoaders/resident.js

@@ -1,5 +1,6 @@
 const { get, isEmpty } = require('lodash')
 
+const { getDatabaseAdapter } = require('@open-condo/keystone/databaseAdapters/utils')
 const { getSchemaCtx } = require('@open-condo/keystone/schema')
 
 const { AbstractDataLoader } = require('@condo/domains/analytics/utils/services/dataLoaders/AbstractDataLoader')
@@ -15,7 +16,7 @@ class ResidentGqlKnexLoader extends GqlToKnexBaseAdapter {
 
     async loadData () {
         const { keystone } = await getSchemaCtx(this.domainName)
-        const knex = keystone.adapter.knex
+        const { knex } = getDatabaseAdapter(keystone)
 
         this.whereIn = {}
 

apps/condo/domains/analytics/utils/services/dataLoaders/ticket.js

@@ -2,6 +2,7 @@ const dayjs = require('dayjs')
 const { get, isEmpty, find } = require('lodash')
 
 const conf = require('@open-condo/config')
+const { getDatabaseAdapter } = require('@open-condo/keystone/databaseAdapters/utils')
 const { getSchemaCtx } = require('@open-condo/keystone/schema')
 const { extractReqLocale } = require('@open-condo/locales/extractReqLocale')
 const { i18n } = require('@open-condo/locales/loader')
@@ -236,7 +237,7 @@ class TicketQualityControlGqlLoader extends GqlToKnexBaseAdapter {
         this.result = null
 
         const { keystone } = await getSchemaCtx(this.domainName)
-        const knex = keystone.adapter.knex
+        const { knex } = getDatabaseAdapter(keystone)
 
         this.extendAggregationWithFilter(this.aggregateBy)
 

apps/condo/domains/common/utils/serverSchema/index.js

@@ -1,6 +1,7 @@
 const { getItems } = require('@keystonejs/server-side-graphql-client')
 const { isFunction } = require('lodash')
 
+const { getDatabaseAdapter } = require('@open-condo/keystone/databaseAdapters/utils')
 const { getSchemaCtx } = require('@open-condo/keystone/schema')
 const GLOBAL_QUERY_LIMIT = 1000
 
@@ -93,7 +94,7 @@ class GqlWithKnexLoadList {
     async initContext () {
         const { keystone: modelAdapter } = await getSchemaCtx(this.listKey)
         this.keystone = modelAdapter
-        this.knex = modelAdapter.adapter.knex
+        this.knex = getDatabaseAdapter(modelAdapter).knex
     }
 
     // Takes rawAggregate SQL function and apply it on all objects with id from ids

bin/prepare.js

@@ -24,6 +24,7 @@ const CERT_FILE = path.join(__filename, '..', '.ssl', 'localhost.pem')
 
 program.option('-f, --filter <names...>', 'Filters apps by name')
 program.option('--https', 'Uses https for local running')
+program.option('-r, --replication <names...>', 'Enables replica adapter to interact with multiple databases')
 program.description(`Prepares applications from the /apps directory for local running 
 by creating separate databases for them 
 and running their local bin/prepare.js scripts.
@@ -35,7 +36,7 @@ function logWithIndent (message, indent = 1) {
 
 async function prepare () {
     program.parse()
-    const { https, filter } = program.opts()
+    const { https, filter, replication } = program.opts()
 
     // Step 1. Sanity checks
     logWithIndent('Running sanity checks')
@@ -108,6 +109,12 @@ async function prepare () {
                 SPORT: String(app.sport),
                 SERVER_URL: app.serviceUrl,
             }
+
+            if (replication && app in replication) {
+                env.DATABASE_URL = `custom:{"default":{"read":"postgresql://postgres:postgres@127.0.0.1:5433/${app.pgName}","write":"postgresql://postgres:postgres@127.0.0.1:5432/${app.pgName}"}}` // NOSONAR used only for test purposes
+                env.DATABASE_MAPPING = '[{"match":"*","query":"default","command":"default"}]'
+            }
+
             await prepareAppEnv(app.name, env)
             logWithIndent('Running migration script', 2)
             const migrateResult = await runAppPackageJsonScript(app.name, 'migrate')

packages/keystone/adapterCache.js

@@ -52,6 +52,8 @@
 const { get, cloneDeep, floor, isEqual } = require('lodash')
 const LRUCache = require('lru-cache')
 
+const { getListAdapters } = require('@open-condo/keystone/databaseAdapters/utils')
+
 const { getExecutionContext } = require('./executionContext')
 const { getLogger } = require('./logging')
 const Metrics = require('./metrics')
@@ -251,12 +253,9 @@ class AdapterCache {
  * @returns {Promise<void>}
  */
 async function patchKeystoneWithAdapterCache (keystone, cacheAPI) {
-    const keystoneAdapter = keystone.adapter
-
     const cache = cacheAPI.cache
     const excludedLists = cacheAPI.excludedLists
-
-    const listAdapters = Object.values(keystoneAdapter.listAdapters)
+    const listAdapters = Object.values(getListAdapters(keystone))
 
     // Step 1: Preprocess lists.
     const relations = {}               // list -> [{list, path, many}]
@@ -512,4 +511,4 @@ function queryIsComplex (query, list, relations) {
 
 module.exports = {
     AdapterCache,
-}
+}

packages/keystone/databaseAdapters/ReplicaKnexAdapter.js

@@ -0,0 +1,117 @@
+const { KnexAdapter } = require('@keystonejs/adapter-knex')
+const { knex } = require('knex')
+const { get, omit } = require('lodash')
+
+const { getLogger } = require('@open-condo/keystone/logging')
+
+const logger = getLogger('replicaKnexAdapter')
+
+const IMMUTABLE_OPERATIONS = ['select', 'show']
+
+/**
+ *
+ */
+class ReplicaKnexAdapter extends KnexAdapter {
+    constructor (props) {
+        super(omit(props, ['connection']))
+        this.readConnection = get(props, ['connection', 'read'])
+        this.writeConnection = get(props, ['connection', 'write'])
+    }
+
+    async _connect () {
+        this.knex = knex({
+            client: 'postgres',
+            pool: { min: 0, max: 1 },
+            connection: this.writeConnection,
+        })
+
+        this.knexWrite = knex({
+            client: 'postgres',
+            pool: { min: 0, max: 3 },
+            connection: this.writeConnection,
+        })
+
+        this.knexRead = knex({
+            client: 'postgres',
+            pool: { min: 0, max: 3 },
+            connection: this.readConnection,
+        })
+
+        this.knex.context.transaction = (...props) => {
+            return this.knexWrite.context.transaction(...props)
+        }
+
+        const checkUseMasterSingle = (object) => {
+            // if object.method equals "insert", "delete" or "update" then use master endpoint
+            if (object.method !== undefined) {
+                return !IMMUTABLE_OPERATIONS.includes(object.method)
+            }
+            // try to parse sql
+            return !IMMUTABLE_OPERATIONS.some(sub => object.sql.includes(sub))
+        }
+
+        const checkUseMasterMultiple = (array) => array.some(checkUseMasterSingle)
+
+        //override runner method
+        this.knex.client.runner = (builder) => {
+            if (builder._queryContext && builder._queryContext.useMaster === true) {
+                return this.knexWrite.client.runner(builder)
+            } else if (
+                builder._queryContext &&
+                builder._queryContext.useMaster === false
+            ) {
+                return this.knexRead.client.runner(builder)
+            }
+
+            let sql
+            let useMaster = true
+            try {
+                sql = builder.toSQL()
+                useMaster = Array.isArray(sql)
+                    ? checkUseMasterMultiple(sql)
+                    : checkUseMasterSingle(sql)
+            } catch (error) {
+                logger.error({
+                    err: error,
+                    msg: 'catch error at connection determination',
+                })
+                // swallow this, it will be thrown properly in a second when Knex internally runs it
+            }
+
+            return useMaster
+                ? this.knexWrite.client.runner(builder)
+                : this.knexRead.client.runner(builder)
+        }
+
+        const masterConnectionResult = await this.knexWrite.raw('select 1+1 as result').catch(result => ({ error: result.error || result }))
+        const readConnectionResult = await this.knexRead.raw('select 1+1 as result').catch(result => ({ error: result.error || result }))
+
+        if (masterConnectionResult.error || readConnectionResult.error) {
+            if (masterConnectionResult.error) {
+                logger.error({
+                    err: masterConnectionResult.error,
+                    msg: 'Could not connect to master database',
+                })
+                throw masterConnectionResult.error
+            }
+
+            logger.error({
+                err: readConnectionResult.error,
+                msg: 'Could not connect to replica database',
+            })
+            throw readConnectionResult.error
+        }
+
+        return true
+    }
+
+    disconnect () {
+        this.knexWrite.destroy()
+        this.knexRead.destroy()
+        this.knex.destroy()
+    }
+}
+
+module.exports = {
+    ReplicaKnexAdapter,
+}

packages/keystone/databaseAdapters/ScalableDatabaseAdapter.js

@@ -1,17 +1,20 @@
 const { KnexAdapter } = require('@keystonejs/adapter-knex')
 const { BaseKeystoneAdapter } = require('@keystonejs/keystone')
-const { find } = require('lodash/collection')
-const { mapValues } = require('lodash/object')
+const { find, mapValues } = require('lodash')
 
 const { FakeDatabaseAdapter } = require('./FakeDatabaseAdapter')
+const { ReplicaKnexAdapter } = require('./ReplicaKnexAdapter')
 const { parseDatabaseUrl, parseDatabaseMapping, matchDatabase } = require('./utils')
 
+
 function initDatabaseAdapters (databases) {
     return mapValues(databases, initDatabaseAdapter)
 }
 
 function initDatabaseAdapter (databaseUrl) {
-    if (databaseUrl.startsWith('postgresql:')) {
+    if (typeof databaseUrl === 'object') {
+        return new ReplicaKnexAdapter({ connection: databaseUrl })
+    } else if (databaseUrl.startsWith('postgresql:')) {
         return new KnexAdapter({ knexOptions: { connection: databaseUrl } })
     } else if (databaseUrl.startsWith('fake:')) {
         // NOTE: case for testing!
@@ -35,15 +38,15 @@ function initListAdapter (rootAdapter, parentAdapter, key, adapterConfig) {
 
 class ScalableDatabaseAdapter extends BaseKeystoneAdapter {
     static PUBLIC_API = [
-        'name', 'config',  // base keystone adapter props
-        'newListAdapter', 'getListAdapterByKey', 'connect', '_connect', 'checkDatabaseVersion', 'postConnect', 'disconnect',  // keystone interface props
-        '__databaseAdapters', '__listMappingRule', '__listMappingAdapters', '__listToDatabase',  // own private props
-        '__kmigratorKnexAdapters',  // kmigrator hacks for backward compatibility
+        'name', 'config', // base keystone adapter props
+        'newListAdapter', 'getListAdapterByKey', 'connect', '_connect', 'checkDatabaseVersion', 'postConnect', 'disconnect', // keystone interface props
+        '__databaseAdapters', '__listMappingRule', '__listMappingAdapters', '__listToDatabase', // own private props
+        '__kmigratorKnexAdapters', // kmigrator hacks for backward compatibility
     ]
 
     constructor (opts = {}) {
         if (!opts.url || !opts.url.startsWith('custom:')) throw new Error('ScalableDatabaseAdapter({ url }) wrong url format!')
-        if (!opts.mapping || !opts.mapping.startsWith('[')) throw new Error('ScalableDatabaseAdapter({ mapping }) wrong url format!')
+        if (!opts.mapping || !opts.mapping.startsWith('[')) throw new Error(`ScalableDatabaseAdapter({ mapping }) wrong url format! ${opts.mapping}`)
 
         const databases = parseDatabaseUrl(opts.url)
         if (!databases) throw new Error('ScalableDatabaseAdapter({ url }) wrong url format!')