import * as _ from 'lodash' ; import * as mongodb from 'mongodb' ; import { ConceptDbConcurrentUpdateError , ConceptDbDuplicateKeyError , ConceptDbNotFoundError , ConceptDbUnknownUpdateError , Collection , Context , Query , UpsertOptions } from './db' ; import { getReqIdPendingFilter , releaseLockOp , UpdateOp } from './updateOp' ; export interface PendingDoc { _pending ? : boolean ; _pendingDetails ? : { reqId : string ; type : string ; } ; } export const unsetPendingOp = { $unset : { _pending : '' , _pendingDetails : '' } } ; export type DbDoc < T > = PendingDoc & T ; const MONGODB_DUPLICATE_KEY_ERROR_CODE = 11000 ; /**
 * An implementation of Déjà Vu's Collection,
 * a wrapper around the MongoDb Node.js driver for transaction handling.
 *
 * This uses a `_pending` boolean field as a lock for each document,
 * and stores the request ID and type (e.g. create-foo, update-foo) under
 * the `_pendingDetails` field of the document.
 *
 * More specifically,
 * - Before creating/updating/deleting a document, acquire the lock
 *   to that document, i.e. set `_pending: true`
 * - If the modified count is 0, another tx has the lock, so throw a concurrent
 *   update error.
 *   Else, locking was a success so the scheme sets the `_pendingDetails` field
 *   and applies any relevant updates, depending on which phase of 2PC it is.
 *   Create operations specify that a document is pending create under
 *   `_pendingDetails` which find operations will take care not to return.
 *   Updates only happen on commit so find operations will only return the
 *   current version of docs, without the pending update.
 * - Release the lock, i.e. unset `_pending` and `_pendingDetails`, when the tx
 *   is over (on abort and at the end of the commit) or at the end of a non-tx
 *   operation, if needed (see below).
 * - Non-tx updates still need to follow the locking scheme so that it can throw
 *   a not found error, or a concurrent update error since you can't perform an
 *   update on a document which has another update in progress/is locked by a tx
 *
 * The locking scheme above ensures that once a tx successfully locks on a doc,
 * no other tx or non-tx operation can modify it. Those concurrent operations
 * will fail with a concurrent update error. Reads will still be able to read
 * the unupdated value.
 *
 * The only constraint that can be enforced atomically for inserts
 * are the uniqueness constraints set by the created indices.
 * Concepts would have to check other preconditions separately,
 * without the guarantee that those conditions would not have changed
 * by the time the insert happens. If the concept does not allow certain fields
 * to be mutated, then preconditions that rely on those field values could be
 * checked outside of a transaction. Alternatively, concepts could handle
 * transactions by themselves and not rely on this library. They could also
 * perform operations that involve multiple documents or multiple Collection<T>s
 * in a transaction by using ConceptDb.inTransaction()
 */ export class CollectionWithPendingLocks < T > implements Collection < T > { private readonly _db : mongodb . Db ; private readonly _name : string ; private readonly _collection : mongodb . Collection < DbDoc < T > > ; constructor ( db , name ) { this . _db = db ; this . _name = name ; this . _collection = this . _db . collection ( this . _name ) ; this . _collection . createIndex ( { '_pendingDetails.reqId' : 1 } , { sparse : true } ) ; } aggregate ( pipeline , options ? ) { const isFirstStageAQuery = pipeline . length > 0 && pipeline [ 0 ] [ '$match' ] ; const originalQuery = isFirstStageAQuery ? pipeline [ 0 ] [ '$match' ] : { } ; const queryNotPendingCreate = this . getNotPendingCreateFilter ( originalQuery ) ; let newPipeline ; if ( isFirstStageAQuery ) { pipeline [ 0 ] [ '$match' ] = queryNotPendingCreate ; newPipeline = pipeline ; } else { const newFirstStage = { $match : queryNotPendingCreate } ; newPipeline = [ newFirstStage ] . concat ( pipeline ) ; } return this . _collection . aggregate ( newPipeline , options ) ; } async createIndex ( fieldOrSpec , options ? ) { return this . _collection . createIndex ( fieldOrSpec , options ) ; } async countDocuments ( query = { } , options ? ) { const queryNotPendingCreate = this . getNotPendingCreateFilter ( query ) ; return this . _collection . countDocuments ( queryNotPendingCreate , options ) ; } async deleteMany ( context , filter , options ? ) { switch ( context . reqType ) { case 'vote' : await this . getLockForUpdate ( context , filter , 'delete' , { upsert : false } , true ) ; return true ; case undefined : await this . getLockForUpdate ( context , filter , 'delete' , { upsert : false } , true ) ; const res = await this . _collection . deleteMany ( filter , options ) ; if ( res . deletedCount === 0 ) { await this . releaseLock ( context , filter , true ) ; throw new ConceptDbUnknownUpdateError ( ) ; } return true ; case 'commit' : await this . _collection . deleteMany ( getReqIdPendingFilter ( context , filter ) ) ; break ; case 'abort' : await this . releaseLock ( context , filter , true ) ; break ; } return undefined ; } async deleteOne ( context , filter ) { switch ( context . reqType ) { case 'vote' : await this . getLockForUpdate ( context , filter , 'delete' ) ; return true ; case undefined : await this . getLockForUpdate ( context , filter , 'delete' ) ; const res = await this . _collection . deleteOne ( filter ) ; if ( res . deletedCount === 0 ) { await this . releaseLock ( context , filter ) ; throw new ConceptDbUnknownUpdateError ( ) ; } return true ; case 'commit' : await this . _collection . deleteOne ( getReqIdPendingFilter ( context , filter ) ) ; break ; case 'abort' : await this . releaseLock ( context , filter ) ; break ; } return undefined ; } async find ( query ? ) { return ( await this . findCursor ( query ) ) . toArray ( ) ; } async findCursor ( query ? ) { const queryNotPendingCreate = this . getNotPendingCreateFilter ( query ) ; return await this . _collection . find ( queryNotPendingCreate ) ; } async findOne ( query , options ? ) { const queryNotPendingCreate = this . getNotPendingCreateFilter ( query ) ; const doc = await this . _collection . findOne ( queryNotPendingCreate , options ) ; if ( doc === null ) { throw new ConceptDbNotFoundError ( this . _name ) ; } return doc ; } async findOneAndUpdateWithFn ( context , filter , updateFn , options ? , validationFn ? ) { switch ( context . reqType ) { case 'vote' : case undefined : { const setOnInsertVal = options && options . setOnInsert ; const getLockUpdateObj = setOnInsertVal ? { $setOnInsert : setOnInsertVal } : { } ; await this . getLockAndUpdate ( context , filter , 'update' , getLockUpdateObj , options ) ; const doc = await this . _collection . findOne ( filter ) ; try { validationFn ( doc ) ; } catch ( err ) { await this . abortUpdate ( context , filter ) ; throw err ; } if ( context . reqType === undefined ) { const update = updateFn ( doc ) ; if ( ! update ) { await this . _collection . deleteOne ( filter ) ; } else { await this . _collection . updateOne ( filter , update , options ) ; await this . releaseLock ( context , filter ) ; } } return doc ; } case 'commit' : const doc = await this . _collection . findOne ( filter ) ; const update = updateFn ( doc ) ; if ( ! update ) { await this . _collection . deleteOne ( getReqIdPendingFilter ( context , filter ) ) ; break ; } await this . _collection . updateOne ( getReqIdPendingFilter ( context , filter ) , { ... updateFn ( doc ) , ... unsetPendingOp } as Object ) ; break ; case 'abort' : await this . abortUpdate ( context , filter ) ; break ; } return undefined ; } async insertMany ( context , docs , options ? ) { switch ( context . reqType ) { case 'vote' : _ . each ( docs , ( doc ) => this . setPendingCreate ( context , doc ) ) ; case undefined : await this . _collection . insertMany ( docs , options ) ; return docs ; case 'commit' : await this . _collection . updateMany ( getReqIdPendingFilter ( context ) , unsetPendingOp ) ; break ; case 'abort' : await this . _collection . deleteMany ( getReqIdPendingFilter ( context ) ) ; break ; } return docs ; } async insertOne ( context , doc , options ? ) { switch ( context . reqType ) { case 'vote' : this . setPendingCreate ( context , doc ) ; case undefined : try { const insertResult = await this . _collection . insertOne ( doc , options ) ; if ( insertResult . insertedCount === 0 ) { throw new ConceptDbUnknownUpdateError ( ) ; } } catch ( err ) { if ( err . code === MONGODB_DUPLICATE_KEY_ERROR_CODE ) { throw new ConceptDbDuplicateKeyError ( ) ; } throw err ; } return doc ; case 'commit' : await this . releaseLock ( context , { } ) ; break ; case 'abort' : await this . _collection . deleteOne ( getReqIdPendingFilter ( context , { } ) ) ; break ; } return doc ; } async updateMany ( context , filter , update , options ? ) { switch ( context . reqType ) { case 'vote' : await this . getLockForUpdate ( context , filter , 'update' , this . getUpsertOptions ( update , options ) , true ) ; return true ; case undefined : await this . getLockAndUpdate ( context , filter , 'update' , update , options , true ) ; await this . releaseLock ( context , filter , true ) ; return true ; case 'commit' : await this . _collection . updateMany ( getReqIdPendingFilter ( context , filter ) , { ... update , ... unsetPendingOp } as Object ) ; break ; case 'abort' : await this . abortUpdate ( context , filter , true ) ; break ; } return undefined ; } async updateOne ( context , filter , update , options ? ) { switch ( context . reqType ) { case 'vote' : await this . getLockForUpdate ( context , filter , 'update' , this . getUpsertOptions ( update , options ) ) ; return true ; case undefined : await this . getLockAndUpdate ( context , filter , 'update' , update , options ) ; await this . releaseLock ( context , filter ) ; return true ; case 'commit' : await this . _collection . updateOne ( getReqIdPendingFilter ( context , filter ) , { ... update , ... unsetPendingOp } as Object ) ; break ; case 'abort' : await this . abortUpdate ( context , filter ) ; break ; } return undefined ; } param context param filter param upsert param updateMany return throws throws private async getPendingLock ( context , filter , upsert , setOnInsert , updateMany ) { const setPendingRes = await this . _update ( { filter , update : { $set : { _pending : true } , $setOnInsert : { ... setOnInsert , ... this . getPendingCreateFilter ( context ) } } , upsert } , updateMany ) ; if ( setPendingRes . upsertedId ) { return true ; } if ( setPendingRes . matchedCount === 0 ) { throw new ConceptDbNotFoundError ( this . _name ) ; } if ( setPendingRes . matchedCount !== setPendingRes . modifiedCount ) { if ( setPendingRes . modifiedCount !== 0 ) { const unsetPendingRes = await this . _update ( { filter : Object . assign ( { } , filter , { _pending : true , _pendingDetails : { $exists : false } } ) , update : unsetPendingOp } ) ; if ( unsetPendingRes . modifiedCount !== setPendingRes . modifiedCount ) { } } throw new ConceptDbConcurrentUpdateError ( ) ; } return false ; } private async releaseLock ( context , filter , updateMany = false ) { await this . _update ( releaseLockOp ( context , filter ) , updateMany ) ; } private async _update ( op , updateMany = false ) { if ( updateMany ) { return await this . _collection . updateMany ( op . filter , op . update , { upsert : op . upsert } ) ; } return await this . _collection . updateOne ( op . filter , op . update , { upsert : op . upsert } ) ; } param context param filter param updateType param upsertOptions param updateMany throws throws private async getLockForUpdate ( context , filter , updateType , upsertOptions = { upsert : false } , updateMany = false ) { return await this . getLockAndUpdate ( context , filter , updateType , upsertOptions . setOnInsert ? { $setOnInsert : upsertOptions . setOnInsert } : { } , { upsert : upsertOptions . upsert } , updateMany ) ; } param context param filter param updateType param update param options param updateMany throws throws private async getLockAndUpdate ( context , filter , updateType , update = { } , options ? , updateMany = false ) { const didUpsert = await this . getPendingLock ( context , filter , options && options . upsert , _ . get ( update , '$setOnInsert' , { } ) , updateMany ) ; if ( didUpsert && update [ '$setOnInsert' ] && Object . keys ( update ) . length === 1 ) { return ; } const updateSetOp = didUpsert ? { } : { $set : { ... update [ '$set' ] , _pendingDetails : { reqId : context . reqId , type : `${ updateType }-${ this . _name }` } } } ; if ( _ . isEmpty ( updateSetOp ) && _ . isEmpty ( update ) ) { return ; } const updateResult = await this . _update ( { filter , update : { ... update , ... updateSetOp } as Object } , updateMany ) ; if ( updateResult . matchedCount === 0 || ( updateResult . matchedCount !== updateResult . modifiedCount ) ) { throw new ConceptDbUnknownUpdateError ( ) ; } } private getNotPendingCreateFilter ( filter ) { return Object . assign ( { } , filter , { '_pendingDetails.type' : { $ne : `create-${ this . _name }` } } ) ; } private getPendingCreateFilter ( context , filter = { } ) { return Object . assign ( { } , filter , { _pendingDetails : { reqId : context . reqId , type : `create-${ this . _name }` } } ) ; } param context param doc private setPendingCreate ( context , doc ) { doc . _pending = true ; doc . _pendingDetails = { reqId : context . reqId , type : `create-${ this . _name }` } ; } private async abortUpdate ( context , filter , updateMany = false ) { const deleteResult = await this . _collection . deleteOne ( this . getPendingCreateFilter ( context , filter ) ) ; if ( deleteResult . deletedCount === 0 ) { await this . releaseLock ( context , filter , updateMany ) ; } } private getUpsertOptions ( update , options ) { return { upsert : options && options . upsert , setOnInsert : update [ '$setOnInsert' ] } ; } }