import { EventEmitter } from 'events' ; import { RecordAlreadyExistsError , RecordNotFoundError } from '../engine/error/' ; import { TransientBundle } from './TransientBundle' ; import type { CRUDEngine } from '../engine/' ; import type { ExpiredBundle } from './ExpiredBundle' ; enum TOPIC { EXPIRED = 'expired' } export interface TransientStore { on ( event , listener ) : this ; } export class TransientStore extends EventEmitter { public static readonly TOPIC = TOPIC ; private readonly bundles : Record < string , TransientBundle > = { } ; private tableName = '' ; constructor ( private readonly engine ) { super ( ) ; } public async delete ( primaryKey ) { const cacheKey = this . constructCacheKey ( primaryKey ) ; await Promise . all ( [ this . deleteFromStore ( primaryKey ) , this . deleteFromCache ( cacheKey ) ] ) ; return cacheKey ; } public deleteFromCache ( cacheKey ) { const timeoutID = this . bundles [ cacheKey ] ?. timeoutID ; if ( timeoutID ) { clearTimeout ( timeoutID as number ) ; } delete this . bundles [ cacheKey ] ; return cacheKey ; } public async get ( primaryKey ) { try { const cachedBundle = await this . getFromCache ( primaryKey ) ; return cachedBundle !== undefined ? cachedBundle : await this . getFromStore ( primaryKey ) ; } catch ( error ) { if ( error instanceof RecordNotFoundError ) { return undefined ; } throw error ; } } public async init ( tableName ) { this . tableName = tableName ; const cacheKeys = [ ] ; const primaryKeys = await this . engine . readAllPrimaryKeys ( this . tableName ) ; const readBundles = primaryKeys . map ( primaryKey => { const cacheKey = this . constructCacheKey ( primaryKey ) ; cacheKeys . push ( cacheKey ) ; return this . engine . read < TransientBundle > ( this . tableName , primaryKey ) ; } ) ; const bundles = await Promise . all ( readBundles ) ; for ( const index in bundles ) { const bundle = bundles [ index ] ; const cacheKey = cacheKeys [ index ] ; await this . startTimer ( cacheKey ) ; this . bundles [ cacheKey ] = bundle ; } return bundles ; } param primaryKey param record param ttl returns public async set < T > ( primaryKey , record , ttl ) { const bundle = this . createTransientBundle ( record , ttl ) ; const cachedBundle = await this . getFromCache ( primaryKey ) ; if ( cachedBundle ) { const message = `Record with primary key "${ primaryKey }" already exists in table "${ this . tableName }" of database "${ this . engine . storeName }".` ; throw new RecordAlreadyExistsError ( message ) ; } else { const cacheKey = await this . save ( primaryKey , bundle ) ; const transientBundle = await this . startTimer ( cacheKey ) ; return this . saveInCache ( cacheKey , transientBundle ) ; } } param primaryKey returns private constructCacheKey ( primaryKey ) { return `${ this . engine . storeName }@${ this . tableName }@${ primaryKey }` ; } private constructPrimaryKey ( cacheKey ) { return cacheKey . replace ( `${ this . engine . storeName }@${ this . tableName }@` , '' ) ; } private createTransientBundle < T > ( record , ttl ) { return { expires : Date . now ( ) + ttl , payload : record , } ; } private deleteFromStore ( primaryKey ) { return this . engine . delete ( this . tableName , primaryKey ) ; } private async expireBundle ( cacheKey ) { const expiredBundle = { cacheKey : cacheKey , payload : this . bundles [ cacheKey ] . payload , primaryKey : this . constructPrimaryKey ( cacheKey ) , } ; await this . delete ( expiredBundle . primaryKey ) ; return expiredBundle ; } private getFromCache ( primaryKey ) { const cacheBundle = this . bundles [ this . constructCacheKey ( primaryKey ) ] ; return Promise . resolve ( cacheBundle ) ; } private getFromStore ( primaryKey ) { return this . engine . read ( this . tableName , primaryKey ) ; } private async save < TransientBundle > ( primaryKey , bundle ) { const cacheKey = this . constructCacheKey ( primaryKey ) ; await Promise . all ( [ this . saveInStore ( primaryKey , bundle ) , this . saveInCache ( cacheKey , bundle ) ] ) ; return cacheKey ; } private saveInCache < TransientBundle > ( cacheKey , bundle ) { return ( this . bundles [ cacheKey ] = bundle as any ) ; } private saveInStore < TransientBundle > ( primaryKey , bundle ) { return this . engine . create ( this . tableName , primaryKey , bundle ) ; } private async startTimer ( cacheKey ) { const primaryKey = this . constructPrimaryKey ( cacheKey ) ; let bundle = await this . get ( primaryKey ) ; if ( ! bundle ) { bundle = new TransientBundle ( ) ; bundle . expires = 0 ; bundle . payload = undefined ; } const { expires , timeoutID } = bundle ; const timespan = expires - Date . now ( ) ; if ( expires <= 0 ) { await this . expireBundle ( cacheKey ) ; } else if ( ! timeoutID ) { bundle . timeoutID = setTimeout ( async ( ) => { const expiredBundle = await this . expireBundle ( cacheKey ) ; this . emit ( TransientStore . TOPIC . EXPIRED , expiredBundle ) ; } , timespan ) ; } return bundle ; } }