import { AggregateCommit } from './EventStore' ; /**
 * Yields batches of commits that can be safely processed in parallel.
 *
 * This means sequential processing of commits for a given aggregateType and aggregateKey
 * as well as respecting the dependency decisions via 'isDependent' option.
 */ export async function * parallelizableCommitBatches ( commits , params ) { const { maxBacklogSize = 1000 , maxBatchSize = 100 , aborted : abortedPromise } = params ; let aborted = false ; abortedPromise ?. then ( ( ) => ( aborted = true ) ) ; let backlog = new Set ( ) ; let batch = new Set ( ) ; const commitCanBePutInBatch = ( commit ) => { for ( const commitInBatch of batch ) { if ( ( commit . aggregateType === commitInBatch . aggregateType && commit . aggregateKey === commitInBatch . aggregateKey ) || ( params . isDependent && params . isDependent ( commit , commitInBatch ) ) ) { return false ; } } return true ; } ; let commitIterator = commits ; while ( ! aborted ) { /**
         * Move as many backlog commits as possible to the batch
         */ for ( const backlogCommit of backlog ) { if ( commitCanBePutInBatch ( backlogCommit ) ) { batch . add ( backlogCommit ) ; backlog . delete ( backlogCommit ) ; } if ( batch . size === maxBatchSize ) break ; } /**
         * Fill batch as much as possible with new commits
         */ let exhaustedCommitIterator = false ; if ( batch . size < maxBatchSize ) { exhaustedCommitIterator = true ; for await ( const commit of commitIterator ) { if ( aborted ) return ; if ( commit ) { if ( commitCanBePutInBatch ( commit ) ) { batch . add ( commit ) ; } else { backlog . add ( commit ) ; } } if ( ( ! commit && batch . size ) || batch . size === maxBatchSize || backlog . size === maxBacklogSize ) { exhaustedCommitIterator = false ; break ; } } } /**
         * Yield batch and progress cursor, we either reached max batch size or commit iterator was exhausted
         */ if ( batch . size ) { let progressCursor ; if ( backlog . size ) { const earliestCursorInBacklog = Array . from ( backlog ) . shift ( ) ! . chronologicalKey ; for ( const commitInBatch of batch ) { if ( commitInBatch . chronologicalKey >= earliestCursorInBacklog ) break ; progressCursor = commitInBatch . chronologicalKey ; } } else { if ( batch . size ) { progressCursor = Array . from ( batch ) . pop ( ) ! . chronologicalKey ; } } yield { commits : Array . from ( batch ) , progressCursor } ; batch = new Set ( ) ; } if ( exhaustedCommitIterator ) { if ( backlog . size ) { commitIterator = backlog ; backlog = new Set ( ) ; } else { return ; } } } }