import { Logger } from 'src/core/server' ; import { map } from 'lodash' ; import { Subject , race , from } from 'rxjs' ; import { bufferWhen , filter , bufferCount , flatMap , mapTo , first } from 'rxjs/operators' ; import { either , Result , asOk , asErr , Ok , Err } from './result_type' ; export interface BufferOptions { bufferMaxDuration ? : number ; bufferMaxOperations ? : number ; logger ? : Logger ; } export interface Entity { id : string ; } export interface OperationError < Input , ErrorOutput > { entity : Input ; error : ErrorOutput ; } export type OperationResult < Input , ErrorOutput , Output = Input > = Result < Output , OperationError < Input , ErrorOutput > > ; export type Operation < Input , ErrorOutput , Output = Input > = ( entity : Input ) => Promise < Result < Output , ErrorOutput > > ; export type BulkOperation < Input , ErrorOutput , Output = Input > = ( entities : Input [ ] ) => Promise < Array < OperationResult < Input , ErrorOutput , Output > > > ; const DONT_FLUSH = false ; const FLUSH = true ; export function createBuffer < Input extends Entity , ErrorOutput , Output extends Entity = Input > ( bulkOperation , { bufferMaxDuration = 0 , bufferMaxOperations = Number . MAX_VALUE , logger } = { } ) { const flushBuffer = new Subject < void > ( ) ; const storeUpdateBuffer = new Subject < { entity : Input ; onSuccess : ( entity : Ok < Output > ) => void ; onFailure : ( error : Err < ErrorOutput | Error > ) => void ; } > ( ) ; storeUpdateBuffer . pipe ( bufferWhen ( ( ) => flushBuffer ) , filter ( ( tasks ) => tasks . length > 0 ) ) . subscribe ( ( bufferedEntities ) => { bulkOperation ( map ( bufferedEntities , 'entity' ) ) . then ( ( results ) => { results . forEach ( ( result ) => either ( result , ( entity ) => { either ( pullFirstWhere ( bufferedEntities , ( { entity : { id } } ) => id === entity . id ) , ( { onSuccess } ) => { onSuccess ( asOk ( entity ) ) ; } , ( ) => { if ( logger ) { logger . warn ( `Unhandled successful Bulk Operation result: ${ entity ?. id ? entity . id : entity }` ) ; } } ) ; } , ( { entity , error } ) => { either ( pullFirstWhere ( bufferedEntities , ( { entity : { id } } ) => id === entity . id ) , ( { onFailure } ) => { onFailure ( asErr ( error ) ) ; } , ( ) => { if ( logger ) { logger . warn ( `Unhandled failed Bulk Operation result: ${ entity ?. id ? entity . id : entity }` ) ; } } ) ; } ) ) ; bufferedEntities . forEach ( ( unhandledBufferedEntity ) => { unhandledBufferedEntity . onFailure ( asErr ( new Error ( `Unhandled buffered operation for entity: ${ unhandledBufferedEntity . entity . id }` ) ) ) ; } ) ; } ) . catch ( ( ex ) => { bufferedEntities . forEach ( ( { onFailure } ) => onFailure ( asErr ( ex ) ) ) ; } ) ; } ) ; let countInBuffer = 0 ; const flushAndResetCounter = ( ) => { countInBuffer = 0 ; flushBuffer . next ( ) ; } ; storeUpdateBuffer . pipe ( flatMap ( ( ) => { return ++ countInBuffer === 1 ? race ( [ storeUpdateBuffer . pipe ( bufferCount ( bufferMaxOperations - 1 ) ) , from ( resolveIn ( bufferMaxDuration ) ) , ] ) . pipe ( first ( ) , mapTo ( FLUSH ) ) : from ( [ DONT_FLUSH ] ) ; } ) , filter ( ( shouldFlush ) => shouldFlush ) ) . subscribe ( { next : flushAndResetCounter , error : flushAndResetCounter , } ) ; return async function ( entity ) { return new Promise ( ( resolve , reject ) => { storeUpdateBuffer . next ( { entity , onSuccess : resolve , onFailure : reject } ) ; } ) ; } ; } function resolveIn ( ms ) { return new Promise ( ( resolve ) => { setTimeout ( resolve , ms ) ; } ) ; } function pullFirstWhere < T > ( collection , predicate ) { const indexOfFirstEntity = collection . findIndex ( predicate ) ; return indexOfFirstEntity >= 0 ? asOk ( collection . splice ( indexOfFirstEntity , 1 ) [ 0 ] ) : asErr ( undefined ) ; }