const env = require("../environment") // ioredis mock is all in memory const Redis = env.isTest() ? require("ioredis-mock") : require("ioredis") const { addDbPrefix, removeDbPrefix, getRedisOptions, SEPARATOR, } = require("./utils") const RETRY_PERIOD_MS = 2000 const STARTUP_TIMEOUT_MS = 5000 const CLUSTERED = false // for testing just generate the client once let CLOSED = false let CLIENT = env.isTest() ? new Redis(getRedisOptions()) : null // if in test always connected let CONNECTED = !!env.isTest() function connectionError(timeout, err) { // manually shut down, ignore errors if (CLOSED) { return } CLIENT.end() CLOSED = true // always clear this on error clearTimeout(timeout) CONNECTED = false console.error("Redis connection failed - " + err) setTimeout(() => { init() }, RETRY_PERIOD_MS) } /** * Inits the system, will error if unable to connect to redis cluster (may take up to 10 seconds) otherwise * will return the ioredis client which will be ready to use. */ function init() { let timeout CLOSED = false // testing uses a single in memory client if (env.isTest() || (CLIENT && CONNECTED)) { return } // start the timer - only allowed 5 seconds to connect timeout = setTimeout(() => { if (!CONNECTED) { connectionError(timeout, "Did not successfully connect in timeout") } }, STARTUP_TIMEOUT_MS) // disconnect any lingering client if (CLIENT) { CLIENT.disconnect() } const { redisProtocolUrl, opts, host, port } = getRedisOptions(CLUSTERED) if (CLUSTERED) { CLIENT = new Redis.Cluster([{ host, port }], opts) } else if (redisProtocolUrl) { CLIENT = new Redis(redisProtocolUrl) } else { CLIENT = new Redis(opts) } // attach handlers CLIENT.on("end", err => { connectionError(timeout, err) }) CLIENT.on("error", err => { connectionError(timeout, err) }) CLIENT.on("connect", () => { clearTimeout(timeout) CONNECTED = true }) } function waitForConnection() { return new Promise(resolve => { if (CLIENT == null) { init() } else if (CONNECTED) { resolve() return } // check if the connection is ready const interval = setInterval(() => { if (CONNECTED) { clearInterval(interval) resolve() } }, 500) }) } /** * Utility function, takes a redis stream and converts it to a promisified response - * this can only be done with redis streams because they will have an end. * @param stream A redis stream, specifically as this type of stream will have an end. * @return {Promise} The final output of the stream */ function promisifyStream(stream) { return new Promise((resolve, reject) => { const outputKeys = new Set() stream.on("data", keys => { keys.forEach(key => { outputKeys.add(key) }) }) stream.on("error", err => { reject(err) }) stream.on("end", async () => { const keysArray = Array.from(outputKeys) try { let getPromises = [] for (let key of keysArray) { getPromises.push(CLIENT.get(key)) } const jsonArray = await Promise.all(getPromises) resolve( keysArray.map(key => ({ key: removeDbPrefix(key), value: JSON.parse(jsonArray.shift()), })) ) } catch (err) { reject(err) } }) }) } class RedisWrapper { constructor(db) { this._db = db } async init() { CLOSED = false init() await waitForConnection() return this } async finish() { CLOSED = true CLIENT.disconnect() } async scan(key = "") { const db = this._db key = `${db}${SEPARATOR}${key}` let stream if (CLUSTERED) { let node = CLIENT.nodes("master") stream = node[0].scanStream({ match: key + "*", count: 100 }) } else { stream = CLIENT.scanStream({ match: key + "*", count: 100 }) } return promisifyStream(stream) } async get(key) { const db = this._db let response = await CLIENT.get(addDbPrefix(db, key)) // overwrite the prefixed key if (response != null && response.key) { response.key = key } // if its not an object just return the response try { return JSON.parse(response) } catch (err) { return response } } async store(key, value, expirySeconds = null) { const db = this._db if (typeof value === "object") { value = JSON.stringify(value) } const prefixedKey = addDbPrefix(db, key) await CLIENT.set(prefixedKey, value) if (expirySeconds) { await CLIENT.expire(prefixedKey, expirySeconds) } } async getTTL(key) { const db = this._db const prefixedKey = addDbPrefix(db, key) return CLIENT.ttl(prefixedKey) } async setExpiry(key, expirySeconds) { const db = this._db const prefixedKey = addDbPrefix(db, key) await CLIENT.expire(prefixedKey, expirySeconds) } async delete(key) { const db = this._db await CLIENT.del(addDbPrefix(db, key)) } async clear() { let items = await this.scan() await Promise.all(items.map(obj => this.delete(obj.key))) } } module.exports = RedisWrapper