import { Store , CreateDocumentParams , ReadDocumentParams , UpdateDocumentParams , DeleteDocumentParams , FindDocumentsParams , CountDocumentsParams , MigrateCollectionParams , MigrateCollectionResult , Document , Expression , Path , Operand } from '@layr/store' ; import type { StorableComponent , Query , SortDescriptor , SortDirection , Operator } from '@layr/storable' ; import { ensureComponentInstance } from '@layr/component' ; import { MongoClient , Db , Collection , Filter , FindOptions } from 'mongodb' ; import { Microbatcher , Operation } from 'microbatcher' ; import { hasOwnProperty , assertIsObjectLike } from 'core-helpers' ; import isEmpty from 'lodash/isEmpty' ; import mapKeys from 'lodash/mapKeys' ; import mapValues from 'lodash/mapValues' ; import escapeRegExp from 'lodash/escapeRegExp' ; import groupBy from 'lodash/groupBy' ; import debugModule from 'debug' ; const debug = debugModule ( 'layr:mongodb-store' ) ; const MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME = '_id' ; const MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_INDEX_NAME = '_id_' ; primaryIdentifier attribute primaryIdentifier attribute export class MongoDBStore extends Store { private _connectionString : string ; private _poolSize : number ; param connectionString param options . poolSize returns example category constructor ( connectionString , options = { } ) { if ( typeof connectionString !== 'string' ) { throw new Error ( `Expected a 'connectionString' to create a MongoDBStore, but received a value of type '${ typeof connectionString }'` ) ; } if ( connectionString . length === 0 ) { throw new Error ( `Expected a 'connectionString' to create a MongoDBStore, but received an empty string` ) ; } const { poolSize = 1 , ... otherOptions } = options ; super ( otherOptions ) ; this . _connectionString = connectionString ; this . _poolSize = poolSize ; } getURL ( ) { return this . _connectionString ; } category category async connect ( ) { await this . _connectClient ( ) ; } category async disconnect ( ) { await this . _disconnectClient ( ) ; } async createDocument ( { collectionName , document } ) { const collection = await this . _getCollection ( collectionName ) ; try { const { acknowledged } = await debugCall ( async ( ) => { const { acknowledged } = await collection . insertOne ( document ) ; return { acknowledged } ; } , 'db.%s.insertOne(%o)' , collectionName , document ) ; return acknowledged ; } catch ( error ) { if ( error . name === 'MongoServerError' && error . code === 11000 ) { const matches = error . message . match ( / index: (.*) dup key/ ) ; if ( matches === null ) { throw error ; } const indexName = matches [ 1 ] ; if ( indexName === MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_INDEX_NAME ) { return false ; } throw Object . assign ( new Error ( `A duplicate key error occurred while creating a MongoDB document (collection: '${ collectionName }', index: '${ indexName }')` ) , { code : 'DUPLICATE_KEY_ERROR' , collectionName , indexName } ) ; } throw error ; } } async readDocument ( { collectionName , identifierDescriptor , projection } ) { const collection = await this . _getCollection ( collectionName ) ; const query = identifierDescriptor ; const options = { projection } ; const document = await debugCall ( async ( ) => await batchableFindOne ( collection , query , options ) , 'db.%s.batchableFindOne(%o, %o)' , collectionName , query , options ) ; if ( document === null ) { return undefined ; } return document ; } async updateDocument ( { collectionName , identifierDescriptor , documentPatch } ) { const collection = await this . _getCollection ( collectionName ) ; const filter = identifierDescriptor ; const { matchedCount } = await debugCall ( async ( ) => { try { const { matchedCount , modifiedCount } = await collection . updateOne ( filter , documentPatch ) ; return { matchedCount , modifiedCount } ; } catch ( error ) { if ( error . name === 'MongoServerError' && error . code === 11000 ) { const matches = error . message . match ( / index: (.*) dup key/ ) ; if ( matches === null ) { throw error ; } const indexName = matches [ 1 ] ; throw Object . assign ( new Error ( `A duplicate key error occurred while updating a MongoDB document (collection: '${ collectionName }', index: '${ indexName }')` ) , { code : 'DUPLICATE_KEY_ERROR' , collectionName , indexName } ) ; } throw error ; } } , 'db.%s.updateOne(%o, %o)' , collectionName , filter , documentPatch ) ; return matchedCount === 1 ; } async deleteDocument ( { collectionName , identifierDescriptor } ) { const collection = await this . _getCollection ( collectionName ) ; const filter = identifierDescriptor ; const { deletedCount } = await debugCall ( async ( ) => { const { deletedCount } = await collection . deleteOne ( filter ) ; return { deletedCount } ; } , 'db.%s.deleteOne(%o)' , collectionName , filter ) ; return deletedCount === 1 ; } async findDocuments ( { collectionName , expressions , projection , sort , skip , limit } ) { const collection = await this . _getCollection ( collectionName ) ; const mongoQuery = buildMongoQuery ( expressions ) ; const mongoSort = buildMongoSort ( sort ) ; const options = { projection } ; const documents = await debugCall ( async ( ) => { const cursor = collection . find ( mongoQuery , options ) ; if ( mongoSort !== undefined ) { cursor . sort ( mongoSort ) ; } if ( skip !== undefined ) { cursor . skip ( skip ) ; } if ( limit !== undefined ) { cursor . limit ( limit ) ; } const documents = await cursor . toArray ( ) ; return documents ; } , 'db.%s.find(%o, %o)' , collectionName , mongoQuery , options ) ; return documents ; } async countDocuments ( { collectionName , expressions } ) { const collection = await this . _getCollection ( collectionName ) ; const query = buildMongoQuery ( expressions ) ; const documentsCount = await debugCall ( async ( ) => { const documentsCount = await collection . countDocuments ( query ) ; return documentsCount ; } , 'db.%s.countDocuments(%o)' , collectionName , query ) ; return documentsCount ; } toDocument < Value > ( storable , value ) { let document = super . toDocument ( storable , value ) ; if ( typeof document === 'object' ) { const primaryIdentifierAttributeName = ensureComponentInstance ( storable ) . getPrimaryIdentifierAttribute ( ) . getName ( ) ; document = mapKeys ( document as any , ( _ , name ) => name === primaryIdentifierAttributeName ? MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME : name ) as Value ; } return document ; } fromDocument ( storable , document ) { let serializedStorable = super . fromDocument ( storable , document ) ; if ( typeof serializedStorable === 'object' ) { const primaryIdentifierAttributeName = ensureComponentInstance ( storable ) . getPrimaryIdentifierAttribute ( ) . getName ( ) ; serializedStorable = mapKeys ( serializedStorable , ( _ , name ) => name === MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME ? primaryIdentifierAttributeName : name ) ; } return serializedStorable ; } category async migrateCollection ( { collectionName , collectionSchema , silent = false } ) { const result = { name : collectionName , createdIndexes : [ ] , droppedIndexes : [ ] } ; const database = await this . _getDatabase ( ) ; let collection ; let collectionHasBeenCreated ; const collections = await database . listCollections ( { name : collectionName } , { nameOnly : true } ) . toArray ( ) ; if ( collections . length === 0 ) { if ( ! silent ) { console . log ( `Creating collection: '${ collectionName }'` ) ; } collection = await database . createCollection ( collectionName ) ; collectionHasBeenCreated = true ; } else { collection = database . collection ( collectionName ) ; collectionHasBeenCreated = false ; } const existingIndexNames = ( await collection . indexes ( ) ) . map ( ( index ) => index . name ) ; const indexesToEnsure = [ ] ; for ( const index of collectionSchema . indexes ) { let indexName = '' ; let indexSpec = { } ; for ( let [ name , direction ] of Object . entries ( index . attributes ) ) { const directionString = direction . toLowerCase ( ) ; if ( indexName !== '' ) { indexName += ' + ' ; } indexName += name ; if ( directionString === 'desc' ) { indexName += ' (desc)' ; } indexSpec [ name ] = directionString === 'desc' ? - 1 : 1 ; } if ( index . isUnique ) { indexName += ' [unique]' ; } if ( indexName === `${ MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME } [unique]` ) { indexName = MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_INDEX_NAME ; } indexesToEnsure . push ( { name : indexName , spec : indexSpec , isUnique : index . isUnique } ) ; } const indexesToCreate = indexesToEnsure . filter ( ( index ) => ! existingIndexNames . includes ( index . name ) ) ; const indexNamesToDrop = existingIndexNames . filter ( ( name ) => name !== MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_INDEX_NAME && ! indexesToEnsure . some ( ( index ) => index . name === name ) ) ; if ( indexesToCreate . length !== 0 || indexNamesToDrop . length !== 0 ) { if ( ! collectionHasBeenCreated && ! silent ) { console . log ( `Migrating collection: '${ collectionName }'` ) ; } } for ( const name of indexNamesToDrop ) { if ( ! silent ) { console . log ( `- Dropping index: '${ name }'` ) ; } await collection . dropIndex ( name ) ; result . droppedIndexes . push ( name ) ; } for ( const index of indexesToCreate ) { if ( ! silent ) { console . log ( `- Creating index: '${ index . name }'` ) ; } await collection . createIndex ( index . spec , { name : index . name , unique : index . isUnique } ) ; result . createdIndexes . push ( index . name ) ; } return result ; } private _client : MongoClient | undefined ; private async _getClient ( ) { await this . _connectClient ( ) ; return this . _client ! ; } private _connectClientPromise : Promise < void > | undefined ; private _connectClient ( ) { if ( this . _connectClientPromise !== undefined ) { return this . _connectClientPromise ; } this . _connectClientPromise = ( async ( ) => { try { if ( this . _client === undefined ) { debug ( `Connecting to MongoDB Server (connectionString: ${ this . _connectionString })...` ) ; this . _client = await MongoClient . connect ( this . _connectionString , { maxPoolSize : this . _poolSize } ) ; debug ( `Connected to MongoDB Server (connectionString: ${ this . _connectionString })` ) ; } } finally { this . _connectClientPromise = undefined ; } } ) ( ) ; return this . _connectClientPromise ; } private async _disconnectClient ( ) { if ( this . _connectClientPromise !== undefined ) { try { await this . _connectClientPromise ; } catch { } } const client = this . _client ; if ( client !== undefined ) { this . _client = undefined ; this . _db = undefined ; debug ( `Disconnecting from MongoDB Server (connectionString: ${ this . _connectionString })...` ) ; await client . close ( ) ; debug ( `Disconnected from MongoDB Server (connectionString: ${ this . _connectionString })` ) ; } } private _db : Db | undefined ; private async _getDatabase ( ) { if ( ! this . _db ) { const client = await this . _getClient ( ) ; this . _db = client . db ( ) ; } return this . _db ; } private _collections : { [ name ] : Collection ; } | undefined ; private async _getCollection ( name ) { if ( this . _collections === undefined ) { this . _collections = Object . create ( null ) ; } if ( this . _collections ! [ name ] === undefined ) { const database = await this . _getDatabase ( ) ; this . _collections ! [ name ] = database . collection ( name ) ; } return this . _collections ! [ name ] ; } } function buildMongoQuery ( expressions ) { const query = { } ; for ( const [ path , operator , value ] of expressions ) { let subquery ; if ( path !== '' ) { subquery = query [ path ] ; if ( subquery === undefined ) { subquery = { } ; query [ path ] = subquery ; } } else { subquery = query ; } const [ actualOperator , actualValue ] = handleOperator ( operator , value , { path } ) ; subquery [ actualOperator ] = actualValue ; } return query ; } function handleOperator ( operator , value , { path } ) { if ( operator === '$equal' ) { return [ '$eq' , value ] ; } if ( operator === '$notEqual' ) { return [ '$ne' , value ] ; } if ( operator === '$greaterThan' ) { return [ '$gt' , value ] ; } if ( operator === '$greaterThanOrEqual' ) { return [ '$gte' , value ] ; } if ( operator === '$lessThan' ) { return [ '$lt' , value ] ; } if ( operator === '$lessThanOrEqual' ) { return [ '$lte' , value ] ; } if ( operator === '$in' ) { return [ '$in' , value ] ; } if ( operator === '$includes' ) { return [ '$regex' , escapeRegExp ( value as string ) ] ; } if ( operator === '$startsWith' ) { return [ '$regex' , `^${ escapeRegExp ( value as string ) }` ] ; } if ( operator === '$endsWith' ) { return [ '$regex' , `${ escapeRegExp ( value as string ) }$` ] ; } if ( operator === '$matches' ) { return [ '$regex' , value ] ; } if ( operator === '$some' ) { const subexpressions = value as Expression [ ] ; const subquery = buildMongoQuery ( subexpressions ) ; return [ '$elemMatch' , subquery ] ; } if ( operator === '$every' ) { const subexpressions = value as Expression [ ] ; const subquery = buildMongoQuery ( subexpressions ) ; return [ '$not' , { $elemMatch : { $not : subquery } } ] ; } if ( operator === '$length' ) { return [ '$size' , value ] ; } if ( operator === '$not' ) { const subexpressions = value as Expression [ ] ; const subquery = buildMongoQuery ( subexpressions ) ; return [ '$not' , subquery ] ; } if ( operator === '$and' ) { const andSubexpressions = value as Expression [ ] [ ] ; const andSubqueries = andSubexpressions . map ( ( subexpressions ) => buildMongoQuery ( subexpressions ) ) ; return [ '$and' , andSubqueries ] ; } if ( operator === '$or' ) { const orSubexpressions = value as Expression [ ] [ ] ; const orSubqueries = orSubexpressions . map ( ( subexpressions ) => buildMongoQuery ( subexpressions ) ) ; return [ '$or' , orSubqueries ] ; } if ( operator === '$nor' ) { const norSubexpressions = value as Expression [ ] [ ] ; const norSubqueries = norSubexpressions . map ( ( subexpressions ) => buildMongoQuery ( subexpressions ) ) ; return [ '$nor' , norSubqueries ] ; } throw new Error ( `A query contains an operator that is not supported (operator: '${ operator }', path: '${ path }')` ) ; } function buildMongoSort ( sort ) { if ( sort === undefined || isEmpty ( sort ) ) { return undefined ; } return mapValues ( sort , ( direction ) => direction . toLowerCase ( ) === 'desc' ? - 1 : 1 ) ; } async function debugCall < Result > ( func , message , ... params ) { let result ; let error ; try { result = await func ( ) ; } catch ( err ) { error = err ; } if ( error !== undefined ) { debug ( `${ message } => Error` , ... params ) ; throw error ; } debug ( `${ message } => %o` , ... params , result ) ; return result as Result ; } const findOneBatcher = Symbol ( 'batcher' ) ; interface FindOneOperation extends Operation { params : [ filter : Filter < any > , options : FindOptions < any > ] ; } function batchableFindOne ( collection , query , options ) { assertIsObjectLike ( query ) ; if ( collection [ findOneBatcher ] === undefined ) { collection [ findOneBatcher ] = new Microbatcher ( function ( operations ) { const operationGroups = groupBy ( operations , ( { params : [ query , options ] } ) => { if ( hasOwnProperty ( query , MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME ) && Object . keys ( query ) . length === 1 ) { query = { [ MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME ] : '___???___' } ; } return JSON . stringify ( [ query , options ] ) ; } ) ; for ( const operations of Object . values ( operationGroups ) ) { if ( operations . length > 1 ) { const ids = operations . map ( ( operation ) => operation . params [ 0 ] [ MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME ] ) ; const options = operations [ 0 ] . params [ 1 ] ; collection . find ( { [ MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME ] : { $in : ids } } , options ) . toArray ( ) . then ( ( documents ) => { for ( const { params : [ query ] , resolve } of operations ) { const document = documents . find ( ( document ) => document [ MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME ] === query [ MONGODB_PRIMARY_IDENTIFIER_ATTRIBUTE_NAME ] ) ; resolve ( document !== undefined ? document : null ) ; } } , ( error ) => { for ( const { reject } of operations ) { reject ( error ) ; } } ) ; } else { const { params : [ query , options ] , resolve , reject } = operations [ 0 ] ; collection . findOne ( query , options ) . then ( resolve , reject ) ; } } } ) ; } return collection [ findOneBatcher ] ! . batch ( query , options ) ; }