import { EventStore , AggregateCommit , AggregateEvent } from './EventStore' ; import { ExtractEventTypes } from './utilityTypes' ; export function defineAggregateRoot < TKeyProp extends string , TState , TEventFactories extends Record < string , ( ... args : any [ ] ) => any > , TAggregateType extends string > ( config ) { return new AggregateRoot < TState , ExtractEventTypes < typeof config . events > , Record < typeof config . keyProps [ number ] , string | number > , TEventFactories , TAggregateType > ( config ) ; } typeParam typeParam typeParam export class AggregateRoot < TState , TEvent extends AggregateEvent , TKeyProps extends Record < string , string | number > , TEventFactories , TAggregateType extends string > { constructor ( public readonly config ) { if ( ! config . keyPropSeparator ) this . config . keyPropSeparator = '.' ; } /**
     * Get aggregate root instance data
     *
     * ```typescript
     * Account.get('mykey')
     * Account.get({customerId: '1011', departmentId: '101})
     * ```
     */ async get ( key , options = { } , data ? ) { const keyString = typeof key === 'string' ? key : this . keyFromProps ( key ) ; const getCommitsOptions = { } ; if ( 'version' in options ) { getCommitsOptions . maxVersion = options . version ; } else if ( 'time' in options ) { getCommitsOptions . maxTime = options . time ; } let { state = undefined , version = 0 , timestamp = 0 } = data || { } ; let existingSnapshotIsInvalid = false ; if ( this . config . snapshots && options . useSnapshots !== false ) { const snapshot = await this . config . snapshots . read ( [ this . config . type , keyString ] ) ; if ( snapshot ) { let snapshotIsUsable = false ; if ( snapshot . compatibilityVersion !== this . config . snapshots . compatibilityVersion ) { existingSnapshotIsInvalid = true ; } else if ( 'version' in options && options . version ) { snapshotIsUsable = options . version >= snapshot . version ; } else if ( 'time' in options && options . time ) { snapshotIsUsable = options . time >= snapshot . timestamp ; } else { snapshotIsUsable = true ; } if ( snapshotIsUsable ) { state = snapshot . state ; version = snapshot . version ; timestamp = snapshot . timestamp ; } } } getCommitsOptions . minVersion = version + 1 ; for await ( const commit of this . getCommits ( keyString , getCommitsOptions ) ) { version = commit . aggregateVersion ; timestamp = commit . timestamp ; for ( const event of commit . events ) { state = this . config . state ( state , commit , event ) ; } } if ( ! version || ! state ) { return ; } if ( this . config . snapshots && options . useSnapshots !== false && state && ( existingSnapshotIsInvalid || version % this . config . snapshots . frequency === 0 ) ) { const snapshotData = { compatibilityVersion : this . config . snapshots . compatibilityVersion , state , timestamp , version , } ; await this . config . snapshots . write ( [ this . config . type , keyString ] , snapshotData ) ; } return { key : keyString , version , timestamp , state } ; } /**
     * Commit aggregate events to the store
     * TODO include info about VersionConflictError
     */ async commit ( key , version , events ) { const aggregateKey = typeof key === 'string' ? key : this . keyFromProps ( key ) ; const commitData = { aggregateType : this . config . type , aggregateKey , aggregateVersion : version , events , timestamp : Date . now ( ) , } ; const commit = { ... commitData , chronologicalKey : this . config . store . chronologicalKey ( commitData ) , } ; return await this . config . store . commit < AggregateCommit < TEvent , TAggregateType > > ( commit ) ; } /**
     * Retrieve aggregate commits for the given key
     */ async * getCommits ( key , fetchOptions ? ) { const keyString = typeof key === 'string' ? key : this . keyFromProps ( key ) ; for await ( const commits of this . config . store . queryAggregateCommits < AggregateCommit < TEvent , TAggregateType > > ( this . config . type , keyString , fetchOptions ) ) { for ( const commit of commits ) yield commit ; } } /**
     * Scan through aggregate roots
     */ async * scanInstances ( ) { let state = undefined ; let key = undefined ; let version = 0 ; let timestamp = 0 ; for await ( const commits of this . config . store . scanAggregateCommitsGroupedByKey < AggregateCommit < TEvent , TAggregateType > > ( this . config . type ) ) { for ( const commit of commits ) { if ( key && key !== commit . aggregateKey ) { if ( state ) yield { key , version , timestamp , state } as AggregateInstanceData < TState > ; state = undefined ; } version = commit . aggregateVersion ; timestamp = commit . timestamp ; key = commit . aggregateKey ; for ( const event of commit . events ) { state = this . config . state ( state , commit , event ) ; } } if ( version ) { yield { key , version , timestamp , state } as AggregateInstanceData < TState > ; } } } public get events ( ) { return this . config . events ; } public propsFromKey ( key ) { const values = key . split ( this . config . keyPropSeparator ! ) ; return this . config . keyProps . reduce ( ( r , keyProp ) => ( { ... r , [ keyProp ] : values . shift ( ) ?. toString ( ) , } ) , { } as TKeyProps ) ; } public keyFromProps ( props ) { return this . config . keyProps . map ( keyProp => props [ keyProp ] . toString ( ) ) . join ( this . config . keyPropSeparator ) ; } } export interface AggregateInstanceData < TState > { state : TState ; version : number ; timestamp : number ; key : string ; } export interface SnapshotData < TState > { state : TState ; version : number ; timestamp : number ; compatibilityVersion : string ; }