feat(global)!: INFRA-791 KV storage adapter with keyPrefix (#5719)

* feat(global): INFRA-791 add script for redis key migration & custom redis adapter

* fix(condo): INFRA-791 move redis rename script execution to the migration side

* docs(global): INFRA-791 add migration guide for major versions

* refactor(global): INFRA-791 extract keyPrefix from package name

* refactor(keystone): INFRA-791 add ability to set redis prefix via env variable

* fix(keystone): INFRA-791 make full copy of env inside redis test

* fix(keystone): INFRA-791 add more validations for keyPrefix

* fix(global): INFRA-791 persist redis migration inside migration & make prefix script more configurable

* test(keystone): INFRA-791 add test for redis adapter fallback options

* fix(keystone): INFRA-791 remove ability to set custom key via env

* fix(condo): INFRA-791 rewrite rename logic

* chore(condo): INFRA-791 rebase with origin & remove redis migration

* fix(keystone): INFRA-791 use custom property for bull redis prefix

* refactor(global): INFRA-791 remove redisIndexes from prepare script

* fix(global): INFRA-791 properly set bull prefix

* feat(global): INFRA-791 add option to use valkey cluster

* fix(global): INFRA-791 run valkey for open-source part of ci

* fix(global): INFRA-791 add ability to setup single node valkey

* fix(keystone): INFRA-791 rename default redis specific envs

* test(global): INFRA-791 legacy support

* chore(global): INFRA-791 temp change ci target 2 current branch

* fix(global): INFRA-791 config os related ci checks

* fix(global): INFRA-791 prepare script issue

* fix(global): INFRA-791 use global key-value adapter for session storage

* fix(global): INFRA-791 wrap kv initial value with string

* fix(global): INFRA-791 push valkey to own registry for tests

* fix(global): INFRA-791 ci tag command typo

* fix(global): INFRA-791 remove image push

* fix(global): INFRA-791 add special option to setup kv cluster env

* fix(global): INFRA-791 rollback ci target branch

* chore(global): INFRA-791 rollback condo tests script setup

* test(keystone): INFRA-791 ignore cluster specific tests

* fix(condo): INFRA-791 hashtag replaceOrganizationEmployeeRole for cluster support

* chore(global): INFRA-791 rebase with origin

* fix(global): INFRA-791 oidc redis adapter cluster support

* chore(global): INFRA-791 cleanup PR #1

* chore(global): INFRA-791 rollback bin/prepare cluster options

* fix(keystone): INFRA-791 rename redis util usages after rebase

* fix(global): INFRA-791 rollbacks & final review adjustments

* test(keystone): INFRA-791 basic kv adapter tests

---------

Co-authored-by: Dmitry Koviazin <[email protected]>

Author: sitozzz

Diff for: .github/workflows/_nodejs.condo.core.tests.yml

@@ -43,7 +43,7 @@ jobs:
         env:
           REGISTRY: ${{ secrets.DOCKER_REGISTRY }}/doma/utils/
         run: |
-          docker-compose up -d redis postgresdb-master postgresdb-replica
+          docker-compose up -d redis postgresdb
 
       - name: run tests
         run: |

Diff for: .github/workflows/nodejs.condo.ci.yml

@@ -532,4 +532,4 @@ jobs:
           npm i -g turbo
           yarn install --immutable
       - name: Test webhooks utils
-        run: yarn workspace @open-condo/webhooks jest
+        run: yarn workspace @open-condo/webhooks jest

Diff for: README.md

@@ -159,6 +159,10 @@ And then start it using:
 yarn workspace @app/condo worker
 ```
 
+## Major version migration guide
+
+Check [migration.md](docs/migration.md)
+
 ## Developing
 
 Check [developing.md](docs/develop.md)

Diff for: apps/condo/domains/miniapp/schema/B2BAccessToken.test.js

@@ -5,10 +5,9 @@
 const { faker } = require('@faker-js/faker')
 const dayjs = require('dayjs')
 const { gql } = require('graphql-tag')
-const IORedis = require('ioredis')
 const pick = require('lodash/pick')
 
-const conf = require('@open-condo/config')
+const { getKVClient } = require('@open-condo/keystone/kv')
 const {
     makeLoggedInAdminClient, makeClient, expectValuesOfCommonFields,
     expectToThrowAuthenticationErrorToObj, expectToThrowAuthenticationErrorToObjects,
@@ -411,7 +410,7 @@ describe('B2BAccessToken', () => {
     describe('Real-life cases', () => {
 
         test('Can not update deleted access token to store it in redis', async () => {
-            const redisClient = new IORedis(conf.REDIS_URL)
+            const redisClient = getKVClient()
             const [createdToken] = await createTestB2BAccessTokenAdmin(admin, b2bAppContext, scopedRightSet)
             expect(createdToken).toBeDefined()
             expect(createdToken).toHaveProperty('sessionId')
@@ -465,9 +464,9 @@ describe('B2BAccessToken', () => {
             const [accessTokenByAdmin] = await B2BAccessTokenReadonlyAdmin.getAll(admin, { sessionId: accessToken.sessionId })
             expect(accessTokenByAdmin.id).toEqual(accessToken.id)
         })
-        
+
         test('SessionId is encrypted', async () => {
-            const redisClient = new IORedis(conf.REDIS_URL)
+            const redisClient = getKVClient()
             const [createdToken] = await createTestB2BAccessTokenAdmin(admin, b2bAppContext, scopedRightSet)
             expect(createdToken).toHaveProperty('sessionId')
             const session = await redisClient.get(`sess:${createdToken.sessionId}`)
@@ -487,7 +486,7 @@ describe('B2BAccessToken', () => {
                 await B2BAccessToken.getOne(client, { id: createdToken.id })
             }, ['objs', 0, 'token'])
         })
-        
+
         describe('Token', () => {
 
             let scopedRightSet
@@ -512,7 +511,7 @@ describe('B2BAccessToken', () => {
                 await updateTestB2BAppAccessRightSet(support, globalRightSet.id, { deletedAt: null, canReadOrganizations: true })
                 await updateTestB2BAppAccessRightSet(support, scopedRightSet.id, { deletedAt: null, canReadOrganizations: true })
             })
-            
+
             describe('Authentication', () => {
 
                 test('Can\'t logout with token', async () => {
@@ -560,7 +559,7 @@ describe('B2BAccessToken', () => {
                 })
 
             })
-            
+
             test('Can access only connected organization', async () => {
                 const [accessToken] = await createTestB2BAccessToken(admin, b2bAppContext, scopedRightSet)
                 const anonymous = await makeClient()
@@ -596,7 +595,7 @@ describe('B2BAccessToken', () => {
                 const repeatOrganizations = await Organization.getAll(anonymous, {})
                 expect(repeatOrganizations).toHaveLength(0)
             })
-            
+
             describe('Deleting related objects', () => {
 
                 beforeEach(async () => {
@@ -619,7 +618,7 @@ describe('B2BAccessToken', () => {
                         await Organization.getAll(anonymous, {})
                     })
                 })
-            
+
                 test('Deleting B2BAppContext leads to session removal', async () => {
                     const [accessToken] = await createTestB2BAccessToken(admin, b2bAppContext, scopedRightSet)
                     const anonymous = await makeClient()

Diff for: apps/condo/domains/organization/schema/ReplaceOrganizationEmployeeRoleService.js

@@ -145,8 +145,8 @@ const ReplaceOrganizationEmployeeRoleService = new GQLCustomSchema('ReplaceOrgan
                 let lock
 
                 try {
-                    const oldRoleLockKey = `replaceOrganizationEmployeeRole:${oldRoleId}`
-                    const newRoleLockKey = `replaceOrganizationEmployeeRole:${newRoleId}`
+                    const oldRoleLockKey = `{replaceOrganizationEmployeeRole}:${oldRoleId}`
+                    const newRoleLockKey = `{replaceOrganizationEmployeeRole}:${newRoleId}`
                     lock = await rLock.acquire([oldRoleLockKey, newRoleLockKey], LOCK_DURATION_IN_SEC, {
                         retryCount: 0,
                     })

Diff for: apps/condo/domains/user/oidc/adapter/RedisAdapter.js

@@ -91,6 +91,9 @@ class RedisAdapter {
             ? { payload: JSON.stringify(payload) } : JSON.stringify(payload)
 
         const multi = this.redis.multi()
+        const multiGrant = this.redis.multi()
+        const multiUser = this.redis.multi()
+        const multiUid = this.redis.multi()
         multi[CONSUMABLE.has(this.name) ? 'hmset' : 'set'](key, store)
 
         if (expiresIn) {
@@ -99,28 +102,29 @@ class RedisAdapter {
 
         if (GRANTABLE.has(this.name) && payload.grantId) {
             const grantKey = grantKeyFor(payload.grantId)
-            multi.rpush(grantKey, key)
+            multiGrant.rpush(grantKey, key)
+
             // if you're seeing grant key lists growing out of acceptable proportions consider using LTRIM
             // here to trim the list to an appropriate length
             const ttl = await this.redis.ttl(grantKey)
             if (expiresIn > ttl) {
-                multi.expire(grantKey, expiresIn)
+                multiGrant.expire(grantKey, expiresIn)
             }
         }
 
         if (payload.userCode) {
             const userCodeKey = userCodeKeyFor(payload.userCode)
-            multi.set(userCodeKey, id)
-            multi.expire(userCodeKey, expiresIn)
+            multiUser.set(userCodeKey, id)
+            multiUser.expire(userCodeKey, expiresIn)
         }
 
         if (payload.uid) {
             const uidKey = uidKeyFor(payload.uid)
-            multi.set(uidKey, id)
-            multi.expire(uidKey, expiresIn)
+            multiUid.set(uidKey, id)
+            multiUid.expire(uidKey, expiresIn)
         }
 
-        await multi.exec()
+        await Promise.all([multi.exec(), multiGrant.exec(), multiUser.exec(), multiUid.exec()])
     }
 
     async revokeByGrantId (grantId) { // eslint-disable-line class-methods-use-this

Diff for: docs/migration.md

@@ -189,4 +189,4 @@ SET data_version 2
    will be running at the same time there's risk, that some information about tasks will be lost / corrupted. 
    **So make sure to kill old apps first and start new ones right after**.
    2. New apps (3.x) must point to new redis instance, so make sure to change `KV_URL` / `REDIS_URL` accordingly
-7) Check everything working good. After that you can down your old Redis instance 
+7) Check everything working good. After that you can down your old Redis instance 

Diff for: package.json

@@ -67,7 +67,7 @@
   "dependencies": {
     "@sentry/nextjs": "^6.19.7",
     "@sentry/node": "^7.102.1",
-    "bull": "^4.8.4",
+    "bull": "^4.16.5",
     "bull-repl": "^0.27.2",
     "commitlint-plugin-function-rules": "^1.3.2",
     "dd-trace": "4.46.0"

Diff for: packages/keystone/kv.js

@@ -1,3 +1,7 @@
+const fs = require('fs')
+const path = require('path')
+const { fileURLToPath } = require('url')
+
 const IORedis = require('ioredis')
 
 const conf = require('@open-condo/config')
@@ -7,15 +11,48 @@ const { getLogger } = require('./logging')
 const KV_CLIENTS = {}
 const logger = getLogger('kv')
 
+const getKVPrefix = () => {
+    const toPath = urlOrPath => urlOrPath instanceof URL ? fileURLToPath(urlOrPath) : urlOrPath
+
+    const cwd = process.cwd()
+    let stopAt = 'apps'
+    let directory = path.resolve(toPath(cwd) ?? '')
+    const { root } = path.parse(directory)
+    stopAt = path.resolve(directory, toPath(stopAt) ?? root)
+    let packageJsonPath
+
+    while (directory && directory !== stopAt && directory !== root) {
+        packageJsonPath = path.isAbsolute('package.json') ? 'package.json' : path.join(directory, 'package.json')
+
+        try {
+            const stats = fs.statSync(packageJsonPath, { throwIfNoEntry: false })
+            if (stats?.isFile()) {
+                break
+            }
+        } catch (e) { console.error(e) }
+
+        directory = path.dirname(directory)
+    }
+
+    return require(packageJsonPath).name.split('/')
+        .pop()
+        .replace(/:/g, '')
+        .replace(/-/g, '_').toLowerCase() + ':'
+}
+
+const PREFIX = getKVPrefix()
+
 /**
  * If you really need to use key-value storage client then you should use this function!
  *
  * @param {string} name -- name of key-value storage client or the task purpose (we can use a different KV_URL for each name)
  * @param {string} purpose -- regular / subscriber key-value storage client mode (read details: https://github.com/luin/ioredis#pubsub); you can also use it if you need a two kv client with different settings; For example, the Bull is required three different kv clients
- * @param {Object} opts -- key-value storage config to customize some client options; But please don't use it!
+ * @param {Object} opts -- client config such as internal API for storages and custom @open-condo specific ones
+ * @param {Object} opts.kvOptions -- key-value storage config to customize some client options; But please don't use it!
+ * @param {boolean} opts.ignorePrefix -- special option for disable prepending app specific prefix to all key operations. Using this flag in the enabled state should only be done if you understand how the cluster works, as well as eliminating incompatibility with third-party libraries by setting your own prefix
  * @return {import('ioredis')}
  */
-function getKVClient (name = 'default', purpose = 'regular', opts = {}) {
+function getKVClient (name = 'default', purpose = 'regular', opts = { kvOptions: {}, ignorePrefix: false }) {
     const clientKey = name + ':' + purpose
 
     logger.info({ msg: 'getKVClient', clientKey, opts })
@@ -29,14 +66,21 @@ function getKVClient (name = 'default', purpose = 'regular', opts = {}) {
         const kvUrl = conf[kvEnvName] || conf[redisEnvName] || conf.KV_URL || conf.REDIS_URL
 
         if (!kvUrl) throw new Error(`No KV_URL or REDIS_URL env! You need to set ${kvEnvName} / KV_URL / ${redisEnvName} / REDIS_URL env`)
+
         // BUILD STEP! OR SOME CASE WITH REDIS_URL=undefined
         if (kvUrl === 'undefined') return undefined
-        const client = new IORedis(kvUrl, { connectionName: clientKey, ...opts })
+
+        const clientOptions = { connectionName: clientKey, ...opts.kvOptions }
+        if (!opts.ignorePrefix) clientOptions['keyPrefix'] = PREFIX
+
+        const client = new IORedis(kvUrl, clientOptions)
+
         client.on('connect', () => logger.info({ msg: 'connect', clientKey }))
         client.on('close', () => logger.info({ msg: 'close', clientKey }))
         client.on('reconnecting', (waitTime) => logger.info({ msg: 'reconnecting', clientKey, waitTime }))
         client.on('error', (error) => logger.error({ msg: 'error', clientKey, error }))
         client.on('end', () => logger.error({ msg: 'end', clientKey }))
+
         KV_CLIENTS[clientKey] = client
     }
 
@@ -45,4 +89,5 @@ function getKVClient (name = 'default', purpose = 'regular', opts = {}) {
 
 module.exports = {
     getKVClient,
+    getKVPrefix,
 }

Diff for: packages/keystone/kv.spec.js

@@ -0,0 +1,75 @@
+const Redis = require('ioredis')
+
+const { getKVPrefix } = require('./kv')
+
+describe('Key value adapter', () => {
+    const OLD_ENV = JSON.parse(JSON.stringify(process.env))
+    let client
+    let nonPrefixedClient
+    let moduleName
+
+    beforeAll(() => {
+        process.env.REDIS_URL = 'redis://127.0.0.1:6379'
+        nonPrefixedClient = new Redis(process.env.REDIS_URL)
+        moduleName = require(process.cwd() + '/package.json').name.split('/').pop() + ':'
+
+        jest.resetModules()
+
+        const { getKVClient } = require('./kv')
+        client = getKVClient('test')
+    })
+
+    afterAll(async () => {
+        await client.flushdb()
+        await client.disconnect()
+        await nonPrefixedClient.disconnect()
+    })
+
+    afterAll(async () => {
+        process.env = { ...OLD_ENV }
+    })
+
+    test('prefix should be the name of root package json', () => {
+        expect(getKVPrefix()).toEqual(moduleName)
+    })
+
+    test('key-value keyPrefix should be module specific', () => {
+        expect(client.options.keyPrefix).toMatch(moduleName)
+    })
+
+    test('default key-value client set all keys with prefix', async () => {
+        await client.set('test1', 'result1')
+        const result = await client.get('test1')
+        expect(result).toMatch('result1')
+
+        const result1 = await nonPrefixedClient.get('test1')
+        expect(result1).toBeNull()
+
+        const result2 = await nonPrefixedClient.get(`${moduleName}test1`)
+        expect(result2).toMatch(result)
+    })
+
+    test('pipeline/multi operations should work as expected', async () => {
+        const [[incrError, incrValue], [ttlError, ttlValue]] = await client
+            .multi()
+            .incrby('incrTest', 1)
+            .ttl('incrTest')
+            .exec()
+
+        expect(incrError).toBeNull()
+        expect(ttlError).toBeNull()
+        expect(incrValue).toEqual(1)
+        expect(ttlValue).toEqual(-1)
+    })
+
+    test('should work with oidc adapter', async () => {
+        await client.rpush('testList', 1)
+        const range = await client.lrange('testList', 0, -1)
+        expect(range).toEqual(expect.arrayContaining(['1']))
+        const multi = client.multi()
+        await multi.rpush('testList', 2).expire('testList', 0).exec()
+
+        const expiredKey = await client.keys('testList')
+        expect(expiredKey).toHaveLength(0)
+    })
+})