import assert from "assert" ; import { Storage , StorageListOptions , StorageListResult , StoredKey , StoredKeyMeta , StoredValue , StoredValueMeta , millisToSeconds , viewToArray , } from "@miniflare/shared" ; import { listFilterMatch , listPaginate } from "@miniflare/storage-memory" ; import { Commands , Pipeline } from "ioredis" ; internal export function _bufferFromArray ( value ) { return Buffer . from ( value . buffer , value . byteOffset , value . byteLength ) ; } export class RedisStorage extends Storage { readonly #redis : Commands ; constructor ( redis , protected readonly namespace ) { super ( ) ; this . #redis = redis ; } readonly #key = ( key ) : string => `${ this . namespace }:value:${ key }` ; readonly #metaKey = ( key ) : string => `${ this . namespace }:meta:${ key }` ; protected throwPipelineErrors ( pipelineRes ) { for ( const [ error ] of pipelineRes ) if ( error ) throw error ; } async has ( key ) { return ( await this . #redis . exists ( this . #key ( key ) ) ) > 0 ; } async hasMany ( keys ) { if ( keys . length === 0 ) return 0 ; return await this . #redis . exists ( ... keys . map ( this . #key ) ) ; } get < Meta = unknown > ( key , skipMetadata ? ) ; get ( key , skipMetadata ) ; async get < Meta > ( key , skipMetadata ? ) { if ( skipMetadata ) { const value = await this . #redis . getBuffer ( this . #key ( key ) ) ; return value === null ? undefined : { value : viewToArray ( value ) } ; } const pipelineRes = await this . #redis . pipeline ( ) . getBuffer ( this . #key ( key ) ) . get ( this . #metaKey ( key ) ) . pttl ( this . #key ( key ) ) . exec ( ) ; assert . strictEqual ( pipelineRes . length , 3 ) ; this . throwPipelineErrors ( pipelineRes ) ; const value = pipelineRes [ 0 ] [ 1 ] ; const meta = pipelineRes [ 1 ] [ 1 ] ; const ttl = pipelineRes [ 2 ] [ 1 ] ; if ( value === null ) return undefined ; return { value : viewToArray ( value ) , metadata : meta ? JSON . parse ( meta ) : undefined , expiration : ttl >= 0 ? millisToSeconds ( Date . now ( ) + ttl ) : undefined , } ; } getMany < Meta = unknown > ( keys , skipMetadata ? ) ; getMany ( keys , skipMetadata ) ; async getMany < Meta = unknown > ( keys , skipMetadata ? ) { if ( keys . length === 0 ) return [ ] ; if ( skipMetadata ) { const values = await this . #redis . mgetBuffer ( ... keys . map ( this . #key ) ) ; return values . map ( ( value ) => value === null ? undefined : { value : viewToArray ( value ) } ) ; } const redisKeys = keys . map ( this . #key ) ; const redisMetaKeys = keys . map ( this . #metaKey ) ; let pipeline = this . #redis . pipeline ( ) . mgetBuffer ( ... redisKeys ) . mget ( ... redisMetaKeys ) ; for ( const redisKey of redisKeys ) pipeline = pipeline . pttl ( redisKey ) ; const pipelineRes = await pipeline . exec ( ) ; assert . strictEqual ( pipelineRes . length , 2 + keys . length ) ; this . throwPipelineErrors ( pipelineRes ) ; const values = pipelineRes [ 0 ] [ 1 ] ; const metas = pipelineRes [ 1 ] [ 1 ] ; assert . strictEqual ( values . length , keys . length ) ; assert . strictEqual ( metas . length , keys . length ) ; const now = Date . now ( ) ; const res = new Array ( keys . length ) ; for ( let i = 0 ; i < keys . length ; i ++ ) { const value = values [ i ] ; const meta = metas [ i ] ; const ttl = pipelineRes [ 2 + i ] [ 1 ] ; if ( value === null ) { res [ i ] = undefined ; } else { res [ i ] = { value : viewToArray ( value ) , metadata : meta ? JSON . parse ( meta ) : undefined , expiration : ttl >= 0 ? millisToSeconds ( now + ttl ) : undefined , } ; } } return res ; } async put < Meta = unknown > ( key , value ) { await this . putMany ( [ [ key , value ] ] ) ; } async putMany < Meta = unknown > ( data ) { let pipeline = this . #redis . pipeline ( ) ; const now = Date . now ( ) ; for ( const [ key , { value , expiration , metadata } ] of data ) { const redisKey = this . #key ( key ) ; const buffer = _bufferFromArray ( value ) ; const ttl = expiration === undefined ? undefined : expiration * 1000 - now ; if ( ttl === undefined ) { pipeline = pipeline . set ( redisKey , buffer ) ; } else { pipeline = pipeline . set ( redisKey , buffer , "PX" , ttl ) ; } if ( metadata ) { const redisMetaKey = this . #metaKey ( key ) ; const json = JSON . stringify ( metadata ) ; if ( ttl === undefined ) { pipeline = pipeline . set ( redisMetaKey , json ) ; } else { pipeline = pipeline . set ( redisMetaKey , json , "PX" , ttl ) ; } } } const pipelineRes = await pipeline . exec ( ) ; this . throwPipelineErrors ( pipelineRes ) ; } async delete ( key ) { const deleted = await this . #redis . del ( this . #key ( key ) , this . #metaKey ( key ) ) ; return deleted > 0 ; } async deleteMany ( keys ) { if ( keys . length === 0 ) return 0 ; const pipeline = this . #redis . pipeline ( ) . del ( ... keys . map ( this . #key ) ) . del ( ... keys . map ( this . #metaKey ) ) ; const pipelineRes = await pipeline . exec ( ) ; assert . strictEqual ( pipelineRes . length , 2 ) ; this . throwPipelineErrors ( pipelineRes ) ; return pipelineRes [ 0 ] [ 1 ] ; } list < Meta = unknown > ( options ? , skipMetadata ? ) ; list ( options , skipMetadata ) ; async list < Meta > ( options ? , skipMetadata ? ) { const redisKeyPrefix = this . #key ( "" ) ; const keys = await new Promise < StoredKeyMeta < Meta > [ ] > ( ( resolve ) => { const keys = [ ] ; const stream = this . #redis . scanStream ( { match : `${ redisKeyPrefix }${ options ?. prefix ?? "" }*` , } ) ; stream . on ( "data" , ( page ) => { for ( const key of page ) { const name = key . substring ( redisKeyPrefix . length ) ; if ( listFilterMatch ( options , name ) ) keys . push ( { name } ) ; } } ) ; stream . on ( "end" , ( ) => resolve ( keys ) ) ; } ) ; const res = listPaginate ( options , keys ) ; if ( skipMetadata || res . keys . length === 0 ) return res ; const redisMetaKeys = res . keys . map ( ( { name } ) => this . #metaKey ( name ) ) ; let pipeline = this . #redis . pipeline ( ) . mget ( ... redisMetaKeys ) ; for ( const key of res . keys ) pipeline = pipeline . pttl ( this . #key ( key . name ) ) ; const pipelineRes = await pipeline . exec ( ) ; assert . strictEqual ( pipelineRes . length , 1 + res . keys . length ) ; this . throwPipelineErrors ( pipelineRes ) ; const metas = pipelineRes [ 0 ] [ 1 ] ; assert . strictEqual ( metas . length , res . keys . length ) ; const now = Date . now ( ) ; for ( let i = 0 ; i < res . keys . length ; i ++ ) { const meta = metas [ i ] ; const ttl = pipelineRes [ 1 + i ] [ 1 ] ; res . keys [ i ] . metadata = meta ? JSON . parse ( meta ) : undefined ; res . keys [ i ] . expiration = ttl >= 0 ? millisToSeconds ( now + ttl ) : undefined ; } return res ; } }