import setupLogger from '@/config/logger' ; import { IExplorer } from '@/services/datastores/base/datastore' ; import Cursor from '@/services/datastores/dynomite/lib/Cursor' ; import { MAX_KEY_STRING_SIZE_CHARS } from '@/services/datastores/dynomite/lib/dynomite-constants' ; import DynomiteCluster from '@/services/datastores/dynomite/lib/DynomiteCluster' ; import types from '@/services/datastores/dynomite/lib/DynomiteTypes' ; import { FieldOperationsNotSupportedForTypeError , KeyNotFoundError , KeyTooLargeError , } from '@/services/datastores/dynomite/lib/errors' ; import { getKeyCountFromInfo , parseInfoString , } from '@/services/datastores/dynomite/lib/utils/redis-info-utils' ; import { scan } from '@/services/datastores/dynomite/lib/utils/scan-utils' ; import { IClientCursor , IDynoCollectionKeyValue , IDynomiteInfo , IKeyValue , IScanResult , } from '@/services/datastores/dynomite/typings/dynomite' ; import { IClusterDefinition } from '@/typings/typings' ; import { deleteHashKeys , deleteListItems , deleteSetMembers , deleteZsetMembers , } from './collections' ; import { ValueType } from 'ioredis' ; const logger = setupLogger ( module ) ; const EMPTY_RESULT = { cursor : { complete : true } , keys : [ ] , count : 0 , } as IScanResult ; class export default class DynomiteExplorer implements IExplorer { private dynomiteCluster : DynomiteCluster ; constructor ( readonly cluster ) { this . dynomiteCluster = new DynomiteCluster ( cluster ) ; } param type param key param fieldValues public async addFields ( type , key , fieldValues ) { const conn = await this . dynomiteCluster . getConnection ( ) ; switch ( type ) { case types . string : throw new Error ( 'Cannot add fields to strings. Please create a key of a type that supports fields.' ) ; case types . list : return conn . rpush ( key , fieldValues . map ( ( r ) => r . value ) ) ; case types . set : return conn . sadd ( key , fieldValues . map ( ( r ) => r . value ) ) ; case types . hash : const keyValuePairs = new Array < string > ( ) ; fieldValues . forEach ( ( r ) => { if ( r . type === 'hash' ) { keyValuePairs . push ( r . key , r . value ) ; } } ) ; if ( keyValuePairs . length === 0 ) { throw Error ( 'No hash key/value pairs provided' ) ; } return conn . hmset ( key , keyValuePairs ) ; case types . zset : const zsetItems = new Array < string > ( ) ; fieldValues . forEach ( ( r ) => { if ( r . type === 'zset' ) { zsetItems . push ( r . score ) ; zsetItems . push ( r . value ) ; } } ) ; if ( zsetItems . length === 0 ) { throw Error ( 'No zset score/value pairs provided' ) ; } return conn . zadd ( key , zsetItems as any ) ; case types . none : throw new KeyNotFoundError ( key ) ; default : throw new Error ( `Unsupported key type: ${ type }` ) ; } } param key param fields public async deleteFields ( key , fieldValues ) { const conn = await this . dynomiteCluster . getConnection ( ) ; const keyType = await conn . type ( key ) ; switch ( keyType ) { case types . string : { throw new Error ( 'Cannot delete field from string types. Please use deleteKey() method instead.' ) ; } case types . list : return deleteListItems ( conn , key , fieldValues ) ; case types . set : return deleteSetMembers ( conn , key , fieldValues ) ; case types . hash : return deleteHashKeys ( conn , key , fieldValues ) ; case types . zset : return deleteZsetMembers ( conn , key , fieldValues ) ; case types . none : throw new KeyNotFoundError ( key ) ; default : throw new Error ( `Unsupported key type: ${ keyType }` ) ; } } param { String } key returns { Promise . < Object > } public async deleteKey ( key ) { logger . info ( 'DynomiteExplorer:deleteKey()' ) ; const conn = await this . dynomiteCluster . getConnection ( ) ; const result = await conn . del ( key ) ; if ( result === 0 ) { throw new KeyNotFoundError ( key ) ; } return { count : result } ; } returns public async getInfo ( ) { logger . info ( 'DynomiteExplorer:getInfo()' ) ; const results = ( await this . dynomiteCluster . executeCommandInSingleZone ( ( conn ) => conn . info ( ) as any ) ) as string [ ] ; return results . map ( ( result ) => parseInfoString ( result ) ) ; } /**
     * Returns the total number of keys available.
     */ public async getKeyCount ( ) { const results = ( await this . dynomiteCluster . executeCommandInSingleZone ( ( conn ) => conn . info ( ) as any ) ) as string [ ] ; return results . reduce ( ( previous , current ) => previous + getKeyCountFromInfo ( current ) , 0 ) ; } param { String } key returns { Promise . < Object > } public async getValue ( key ) { const conn = await this . dynomiteCluster . getConnection ( ) ; const type = await conn . type ( key ) ; let value ; switch ( type ) { case types . string : const len = await conn . strlen ( key ) ; if ( len > MAX_KEY_STRING_SIZE_CHARS ) { throw new KeyTooLargeError ( key , len ) ; } value = await conn . get ( key ) ; break ; case types . list : value = await conn . lrange ( key , 0 , - 1 ) ; break ; case types . set : value = await conn . smembers ( key ) ; break ; case types . hash : value = await conn . hgetall ( key ) ; break ; case types . zset : value = await conn . zrange ( key , 0 , - 1 , 'WITHSCORES' ) ; break ; case types . none : throw new KeyNotFoundError ( key ) ; default : throw new Error ( `Unsupported key type: ${ type }` ) ; } const ttl = await conn . ttl ( key ) ; return { name : key , value , type , ttl } ; } param key param fieldValues public async updateFields ( key , fieldValues ) { const conn = await this . dynomiteCluster . getConnection ( ) ; const type = await conn . type ( key ) ; const supportedKeyTypes = [ types . list , types . hash , types . zset ] ; if ( ! supportedKeyTypes . includes ( type ) ) { throw new FieldOperationsNotSupportedForTypeError ( key , type , supportedKeyTypes ) ; } const fieldPairs = [ ] as string [ ] ; switch ( type ) { case types . list : { const promises = fieldValues . map ( ( r ) => { if ( r . type === 'list' ) { return conn . lset ( key , r . index , r . value ) ; } return Promise . resolve ( 'OK' ) ; } ) ; await Promise . all ( promises ) ; break ; } case types . hash : fieldValues . forEach ( ( r ) => { if ( r . type === 'hash' ) { fieldPairs . push ( r . key ) ; fieldPairs . push ( r . value ) ; } } ) ; await conn . hmset ( key , fieldPairs ) ; break ; case types . zset : fieldValues . forEach ( ( r ) => { if ( r . type === 'zset' ) { fieldPairs . push ( r . score ) ; fieldPairs . push ( r . value ) ; } } ) ; await conn . zadd ( key , fieldPairs as any ) ; break ; } logger . info ( 'field updated successfully' ) ; return true ; } param key param value returns public async setValue ( key , value ) { const conn = await this . dynomiteCluster . getConnection ( ) ; return conn . set ( key , value ) ; } param key param ttl public async setExpiration ( key , ttl ) { const conn = await this . dynomiteCluster . getConnection ( ) ; if ( ttl === undefined ) { return conn . persist ( key ) ; } return conn . expire ( key , ttl ) ; } param cursor param match param count param pageSize public async getKeys ( cursorObj , match , count , pageSize ) { if ( match . indexOf ( '*' ) < 0 ) { try { await this . getValue ( match ) ; return { cursor : { complete : true } , keys : [ match ] , count : 1 , } ; } catch ( err ) { if ( err instanceof KeyNotFoundError ) { return EMPTY_RESULT ; } throw err ; } } else { let cursor ; if ( cursorObj ) { cursor = Cursor . fromClientCursor ( cursorObj ) ; } else { const ringMembers = this . dynomiteCluster . getFirstRingMembers ( ) ; logger . debug ( `no scan cursor provided. setting up cursor: ${ JSON . stringify ( ringMembers ) }` ) ; cursor = new Cursor ( ringMembers ) ; } const result = await scan ( this . dynomiteCluster , cursor , match , count , pageSize ) ; return result ; } } public async shutdown ( ) { this . dynomiteCluster . disconnect ( ) ; } }