import * as core from "@fluidframework/server-services-core" ; import { AggregationCursor , Collection , MongoClient , MongoClientOptions } from "mongodb" ; import { Lumberjack } from "@fluidframework/server-services-telemetry" ; const MaxFetchSize = 2000 ; export class MongoCollection < T > implements core . ICollection < T > { constructor ( private readonly collection ) { } public aggregate ( group , options ? ) { const pipeline = [ ] ; pipeline . $group = group ; return this . collection . aggregate ( pipeline , options ) ; } public find ( query , sort , limit = MaxFetchSize ) { return this . collection . find ( query ) . sort ( sort ) . limit ( limit ) . toArray ( ) ; } public findOne ( query ) { return this . collection . findOne ( query ) ; } public findAll ( ) { return this . collection . find ( { } ) . toArray ( ) ; } public async update ( filter , set , addToSet ) { return this . updateCore ( filter , set , addToSet , false ) ; } public async updateMany ( filter , set , addToSet ) { return this . updateManyCore ( filter , set , addToSet , false ) ; } public async upsert ( filter , set , addToSet ) { return this . updateCore ( filter , set , addToSet , true ) ; } public async distinct ( key , query ) { return this . collection . distinct ( key , query ) ; } public async deleteOne ( filter ) { return this . collection . deleteOne ( filter ) ; } public async deleteMany ( filter ) { return this . collection . deleteMany ( filter ) ; } public async insertOne ( value ) { const result = await this . collection . insertOne ( value ) ; return result . insertedId ; } public async insertMany ( values , ordered ) { await this . collection . insertMany ( values , { ordered : false } ) ; } public async createIndex ( index , unique ) { try { const indexName = await this . collection . createIndex ( index , { unique } ) ; Lumberjack . info ( `Created index ${ indexName }` ) ; } catch ( error ) { Lumberjack . error ( `Index creation failed` , error ) ; } } public async createTTLIndex ( index , expireAfterSeconds ? ) { await this . collection . createIndex ( index , { expireAfterSeconds } ) ; } public async findOrCreate ( query , value ) { const result = await this . collection . findOneAndUpdate ( query , { $setOnInsert : value } , { returnOriginal : true , upsert : true } ) ; if ( result . value ) { return { value : result . value , existing : true } ; } else { return { value , existing : false } ; } } private async updateCore ( filter , set , addToSet , upsert ) { const update = { } ; if ( set ) { update . $set = set ; } if ( addToSet ) { update . $addToSet = addToSet ; } const options = { upsert } ; await this . collection . updateOne ( filter , update , options ) ; } private async updateManyCore ( filter , set , addToSet , upsert ) { const update = { } ; if ( set ) { update . $set = set ; } if ( addToSet ) { update . $addToSet = addToSet ; } const options = { upsert } ; await this . collection . updateMany ( filter , update , options ) ; } } export class MongoDb implements core . IDb { constructor ( private readonly client ) { } public close ( ) { return this . client . close ( ) ; } public on ( event , listener ) { this . client . on ( event , listener ) ; } public collection < T > ( name ) { const collection = this . client . db ( "admin" ) . collection < T > ( name ) ; return new MongoCollection < T > ( collection ) ; } } export class MongoDbFactory implements core . IDbFactory { constructor ( private readonly endpoint , private readonly bufferMaxEntries ? ) { } public async connect ( ) { const options = { autoReconnect : true , bufferMaxEntries : this . bufferMaxEntries ?? 50 , keepAlive : true , keepAliveInitialDelay : 180000 , reconnectInterval : 1000 , reconnectTries : 100 , socketTimeoutMS : 120000 , useNewUrlParser : true } ; const connection = await MongoClient . connect ( this . endpoint , options ) ; return new MongoDb ( connection ) ; } }