import { IEvent } from '@irontitan/tardis' ; import { EventRepository } from './EventRepository' ; import { IEventEntity } from '../../interfaces/IEventEntity' ; import { Collection , ObjectId , ClientSession } from 'mongodb' ; import { IEntityConstructor } from '../../interfaces/IEntityConstructor' ; import { IPaginatedQueryResult } from '../../interfaces/IPaginatedQueryResult' ; interface IDatabaseDocument { state : any ; events : IEvent < any > [ ] ; } export abstract class MongodbEventRepository < TEntity extends IEventEntity > extends EventRepository < TEntity > { protected _collection : Collection ; constructor ( collection , Entity ) { super ( Entity ) ; this . _collection = collection ; } param { Function } fn param { ClientSession } session returns { * } protected async _tryRunningWithSession ( fn , session ) { session . startTransaction ( ) ; try { const result = await fn ( ) ; session . commitTransaction ( ) ; return result ; } catch ( err ) { session . abortTransaction ( ) ; throw err ; } } param { TEntity } entity private async _create ( entity ) { await this . _collection . insertOne ( { _id : new ObjectId ( entity . id ) , events : entity . pendingEvents , state : entity . state } ) ; return entity . confirmEvents ( ) ; } param { TEntity } entity async save ( entity , force = false ) { const { state , events , pendingEvents } = entity ; const document = await this . findById ( entity . id ) ; if ( ! document ) return this . _create ( entity ) ; const operations = force ? { $set : { state , events } } : { $set : { state } , $push : { events : { $each : pendingEvents } } } ; await this . _collection . updateOne ( { _id : new ObjectId ( entity . id ) } , operations ) ; return entity . confirmEvents ( ) ; } param { { [ key : string ] : any } } query param { number } page param { number } size param { { [ field : string ] : 1 | - 1 } } sort returns { Promise } async _runPaginatedQuery ( query , page , size , sort = { } ) { const skip = ( Number ( page ) - 1 ) * Number ( size ) ; const limit = Number ( size ) ; const total = await this . _collection . countDocuments ( query ) ; if ( total === 0 ) return { documents : [ ] , count : 0 , range : { from : 0 , to : 0 } , total } ; const documents = await this . _collection . find ( query , { skip , limit , projection : { events : 1 } , sort } ) . toArray ( ) ; const count = documents . length ; const range = { from : skip , to : skip + count } ; return { documents , count , range , total } ; } param { TEntity [ ] } entities async bulkUpdate ( entities , session ? ) { const operations = entities . filter ( ( entity ) => entity . pendingEvents . length > 0 ) . map ( entity => { return { updateOne : { filter : { _id : entity . id } , update : { $set : { state : entity . state } , $push : { events : { $each : entity . pendingEvents } } } } } ; } ) ; await this . _collection . bulkWrite ( operations , { ordered : true , session } ) ; return entities . map ( ( entity ) => entity . confirmEvents ( ) ) ; } param { TEntity [ ] } entities async bulkInsert ( entities , session ? ) { const operations = entities . filter ( ( entity ) => entity . pendingEvents . length > 0 ) . map ( entity => { return { insertOne : { document : { _id : entity . id , state : entity . state , events : entity . pendingEvents } } } ; } ) ; await this . _collection . bulkWrite ( operations , { ordered : true , session } ) ; return entities . map ( ( entity ) => entity . confirmEvents ( ) ) ; } param { string | ObjectId } id async findById ( id ) { if ( ! ObjectId . isValid ( id ) ) return null ; const document = await this . _collection . findOne ( { _id : new ObjectId ( id ) } , { projection : { state : 1 , events : 1 } } ) ; if ( ! document ) return null ; return new this . _Entity ( ) . setPersistedEvents ( document . events ) ; } param { Object } query protected async existBy ( query ) { const documentExists = await this . _collection . find ( query ) . limit ( 1 ) . count ( ) ; return Boolean ( documentExists ) ; } param { ClientSession } session returns { { bulkUpdate : Function } } withSession ( session ) { return { bulkUpdate : ( entities ) => this . _tryRunningWithSession ( ( ) => this . bulkUpdate ( entities , session ) , session ) , bulkInsert : ( entities ) => this . _tryRunningWithSession ( ( ) => this . bulkInsert ( entities , session ) , session ) , } ; } }