From a9b0e3a1f7e8e3a9f0ade71d987ba7119e70fb5b Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Tue, 17 Nov 2020 23:51:45 -0800 Subject: [PATCH] Use a single MongoDB client --- bin/genieacs-cwmp.ts | 5 +- bin/genieacs-fs.ts | 12 +- bin/genieacs-nbi.ts | 5 +- bin/genieacs-ui.ts | 9 +- lib/cache.ts | 18 +-- lib/db.ts | 16 ++- lib/fs.ts | 17 ++- lib/nbi.ts | 77 +++++----- lib/ui/db.ts | 325 +++++++++++++++---------------------------- 9 files changed, 197 insertions(+), 287 deletions(-) diff --git a/bin/genieacs-cwmp.ts b/bin/genieacs-cwmp.ts index a271d77..6e01d95 100755 --- a/bin/genieacs-cwmp.ts +++ b/bin/genieacs-cwmp.ts @@ -24,7 +24,6 @@ import * as server from "../lib/server"; import * as cwmp from "../lib/cwmp"; import * as db from "../lib/db"; import * as extensions from "../lib/extensions"; -import * as cache from "../lib/cache"; import { version as VERSION } from "../package.json"; logger.init("cwmp", VERSION); @@ -36,7 +35,6 @@ function exitWorkerGracefully(): void { setTimeout(exitWorkerUngracefully, 5000).unref(); Promise.all([ db.disconnect(), - cache.disconnect(), extensions.killAll(), cluster.worker.disconnect(), ]).catch(exitWorkerUngracefully); @@ -99,7 +97,8 @@ if (!cluster.worker) { server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); - const initPromise = Promise.all([db.connect(), cache.connect()]) + const initPromise = db + .connect() .then(() => { server.start(options, cwmp.listener); }) diff --git a/bin/genieacs-fs.ts b/bin/genieacs-fs.ts index b6d3146..c794fa6 100755 --- a/bin/genieacs-fs.ts +++ b/bin/genieacs-fs.ts @@ -23,7 +23,6 @@ import * as cluster from "../lib/cluster"; import * as server from "../lib/server"; import { listener } from "../lib/fs"; import * as db from "../lib/db"; -import * as cache from "../lib/cache"; import { version as VERSION } from "../package.json"; logger.init("fs", VERSION); @@ -33,11 +32,9 @@ const SERVICE_PORT = config.get("FS_PORT") as number; function exitWorkerGracefully(): void { setTimeout(exitWorkerUngracefully, 5000).unref(); - Promise.all([ - db.disconnect(), - cache.disconnect(), - cluster.worker.disconnect(), - ]).catch(exitWorkerUngracefully); + Promise.all([db.disconnect(), cluster.worker.disconnect()]).catch( + exitWorkerUngracefully + ); } function exitWorkerUngracefully(): void { @@ -100,7 +97,8 @@ if (!cluster.worker) { listener(req, res); }; - const initPromise = Promise.all([db.connect(), cache.connect()]) + const initPromise = db + .connect() .then(() => { server.start(options, _listener); }) diff --git a/bin/genieacs-nbi.ts b/bin/genieacs-nbi.ts index 4b695bf..623dd50 100755 --- a/bin/genieacs-nbi.ts +++ b/bin/genieacs-nbi.ts @@ -24,7 +24,6 @@ import * as server from "../lib/server"; import { listener } from "../lib/nbi"; import * as db from "../lib/db"; import * as extensions from "../lib/extensions"; -import * as cache from "../lib/cache"; import { version as VERSION } from "../package.json"; logger.init("nbi", VERSION); @@ -36,7 +35,6 @@ function exitWorkerGracefully(): void { setTimeout(exitWorkerUngracefully, 5000).unref(); Promise.all([ db.disconnect(), - cache.disconnect(), extensions.killAll(), cluster.worker.disconnect(), ]).catch(exitWorkerUngracefully); @@ -104,7 +102,8 @@ if (!cluster.worker) { listener(req, res); }; - const initPromise = Promise.all([db.connect(), cache.connect()]) + const initPromise = db + .connect() .then(() => { server.start(options, _listener); }) diff --git a/bin/genieacs-ui.ts b/bin/genieacs-ui.ts index 57c9b27..2ab4f7d 100755 --- a/bin/genieacs-ui.ts +++ b/bin/genieacs-ui.ts @@ -23,9 +23,7 @@ import * as cluster from "../lib/cluster"; import * as server from "../lib/server"; import { listener } from "../lib/ui"; import * as extensions from "../lib/extensions"; -import * as db from "../lib/ui/db"; -import * as db2 from "../lib/db"; -import * as cache from "../lib/cache"; +import * as db from "../lib/db"; import { version as VERSION } from "../package.json"; logger.init("ui", VERSION); @@ -37,8 +35,6 @@ function exitWorkerGracefully(): void { setTimeout(exitWorkerUngracefully, 5000).unref(); Promise.all([ db.disconnect(), - db2.disconnect(), - cache.disconnect(), extensions.killAll(), cluster.worker.disconnect(), ]).catch(exitWorkerUngracefully); @@ -106,7 +102,8 @@ if (!cluster.worker) { listener(req, res); }; - const initPromise = Promise.all([db2.connect(), cache.connect()]) + const initPromise = db + .connect() .then(() => { server.start(options, _listener); }) diff --git a/lib/cache.ts b/lib/cache.ts index 0973ab5..15ba41f 100644 --- a/lib/cache.ts +++ b/lib/cache.ts @@ -17,32 +17,22 @@ * along with GenieACS. If not, see . */ -import { MongoClient, Collection } from "mongodb"; +import { Collection } from "mongodb"; +import { onConnect } from "./db"; import * as config from "./config"; const MAX_CACHE_TTL = +config.get("MAX_CACHE_TTL"); -let clientPromise: Promise; let mongoCollection: Collection; let mongoTimeOffset = 0; -export async function connect(): Promise { - const MONGODB_CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL"); - clientPromise = MongoClient.connect(MONGODB_CONNECTION_URL, { - useNewUrlParser: true, - useUnifiedTopology: true, - }); - const db = (await clientPromise).db(); +onConnect(async (db) => { mongoCollection = db.collection("cache"); await mongoCollection.createIndex({ expire: 1 }, { expireAfterSeconds: 0 }); const now = Date.now(); const res = await db.command({ hostInfo: 1 }); mongoTimeOffset = res.system.currentTime.getTime() - now; -} - -export async function disconnect(): Promise { - if (clientPromise) await (await clientPromise).close(); -} +}); export async function get(key: string | string[]): Promise { const expire = new Date(Date.now() - mongoTimeOffset); diff --git a/lib/db.ts b/lib/db.ts index c15607b..c87b9cf 100644 --- a/lib/db.ts +++ b/lib/db.ts @@ -17,7 +17,7 @@ * along with GenieACS. If not, see . */ -import { MongoClient, ObjectID, Collection } from "mongodb"; +import { MongoClient, ObjectID, Collection, Db } from "mongodb"; import { get } from "./config"; import { encodeTag, escapeRegExp } from "./common"; import { parse } from "./common/expression-parser"; @@ -45,7 +45,6 @@ export let tasksCollection: Collection, configCollection: Collection; let clientPromise: Promise; -export let client: MongoClient; function compareAccessLists(list1: string[], list2: string[]): boolean { if (list1.length !== list2.length) return false; @@ -53,15 +52,24 @@ function compareAccessLists(list1: string[], list2: string[]): boolean { return true; } +const onConnectCallbacks: ((db: Db) => Promise)[] = []; + +export function onConnect(callback: (db: Db) => Promise): void { + onConnectCallbacks.push(callback); +} + export async function connect(): Promise { clientPromise = MongoClient.connect("" + get("MONGODB_CONNECTION_URL"), { useNewUrlParser: true, useUnifiedTopology: true, }); - client = await clientPromise; + const client = await clientPromise; const db = client.db(); + await Promise.all(onConnectCallbacks.map((c) => c(db))); +} +onConnect(async (db) => { tasksCollection = db.collection("tasks"); await tasksCollection.createIndex({ device: 1, timestamp: 1 }); @@ -76,7 +84,7 @@ export async function connect(): Promise { permissionsCollection = db.collection("permissions"); usersCollection = db.collection("users"); configCollection = db.collection("config"); -} +}); export async function disconnect(): Promise { if (clientPromise) await (await clientPromise).close(); diff --git a/lib/fs.ts b/lib/fs.ts index 7eec066..1d9d38f 100644 --- a/lib/fs.ts +++ b/lib/fs.ts @@ -20,11 +20,19 @@ import * as url from "url"; import * as querystring from "querystring"; import { IncomingMessage, ServerResponse } from "http"; -import { GridFSBucket } from "mongodb"; -import * as db from "./db"; +import { Collection, GridFSBucket } from "mongodb"; +import { onConnect } from "./db"; import * as logger from "./logger"; import { getRequestOrigin } from "./forwarded"; +let filesCollection: Collection; +let filesBucket: GridFSBucket; + +onConnect(async (db) => { + filesCollection = db.collection("fs.files"); + filesBucket = new GridFSBucket(db); +}); + export function listener( request: IncomingMessage, response: ServerResponse @@ -39,7 +47,7 @@ export function listener( remoteAddress: getRequestOrigin(request).remoteAddress, }; - db.filesCollection.findOne({ _id: filename }, (err, file) => { + filesCollection.findOne({ _id: filename }, (err, file) => { if (err) throw err; if (!file) { @@ -55,8 +63,7 @@ export function listener( "Content-Length": file.length, }); - const bucket = new GridFSBucket(db.client.db()); - const downloadStream = bucket.openDownloadStreamByName(filename); + const downloadStream = filesBucket.openDownloadStreamByName(filename); downloadStream.pipe(response); logger.accessInfo(log); diff --git a/lib/nbi.ts b/lib/nbi.ts index 2a86d31..11188da 100644 --- a/lib/nbi.ts +++ b/lib/nbi.ts @@ -18,11 +18,11 @@ */ import * as url from "url"; -import * as mongodb from "mongodb"; +import { Collection, GridFSBucket, ObjectId } from "mongodb"; import * as querystring from "querystring"; import * as vm from "vm"; import * as config from "./config"; -import * as db from "./db"; +import { onConnect } from "./db"; import * as query from "./query"; import * as apiFunctions from "./api-functions"; import { IncomingMessage, ServerResponse } from "http"; @@ -43,6 +43,23 @@ const PROVISIONS_REGEX = /^\/provisions\/([a-zA-Z0-9\-_%]+)\/?$/; const VIRTUAL_PARAMETERS_REGEX = /^\/virtual_parameters\/([a-zA-Z0-9\-_%]+)\/?$/; const FAULTS_REGEX = /^\/faults\/([a-zA-Z0-9\-_%:]+)\/?$/; +const collections = { + tasks: null as Collection, + devices: null as Collection, + presets: null as Collection, + objects: null as Collection, + "fs.files": null as Collection, + provisions: null as Collection, + virtualParameters: null as Collection, + faults: null as Collection, +}; +let filesBucket: GridFSBucket; + +onConnect(async (db) => { + for (const k of Object.keys(collections)) collections[k] = db.collection(k); + filesBucket = new GridFSBucket(db); +}); + function throwError(err: Error, httpResponse?: ServerResponse): never { if (httpResponse) { httpResponse.writeHead(500, { Connection: "close" }); @@ -85,7 +102,7 @@ export function listener( if (request.method === "PUT") { const preset = JSON.parse(body.toString()); preset._id = presetName; - db.presetsCollection.replaceOne( + collections.presets.replaceOne( { _id: presetName }, preset, { upsert: true }, @@ -106,7 +123,7 @@ export function listener( } ); } else if (request.method === "DELETE") { - db.presetsCollection.deleteOne({ _id: presetName }, (err) => { + collections.presets.deleteOne({ _id: presetName }, (err) => { if (err) return void throwError(err, response); cache @@ -132,7 +149,7 @@ export function listener( if (request.method === "PUT") { const object = JSON.parse(body.toString()); object._id = objectName; - db.objectsCollection.replaceOne( + collections.objects.replaceOne( { _id: objectName }, object, { upsert: true }, @@ -153,7 +170,7 @@ export function listener( } ); } else if (request.method === "DELETE") { - db.objectsCollection.deleteOne({ _id: objectName }, (err) => { + collections.objects.deleteOne({ _id: objectName }, (err) => { if (err) return void throwError(err, response); cache @@ -190,7 +207,7 @@ export function listener( return; } - db.provisionsCollection.replaceOne( + collections.provisions.replaceOne( { _id: provisionName }, object, { upsert: true }, @@ -211,7 +228,7 @@ export function listener( } ); } else if (request.method === "DELETE") { - db.provisionsCollection.deleteOne({ _id: provisionName }, (err) => { + collections.provisions.deleteOne({ _id: provisionName }, (err) => { if (err) return void throwError(err, response); cache @@ -248,7 +265,7 @@ export function listener( return; } - db.virtualParametersCollection.replaceOne( + collections.virtualParameters.replaceOne( { _id: virtualParameterName }, object, { upsert: true }, @@ -269,7 +286,7 @@ export function listener( } ); } else if (request.method === "DELETE") { - db.virtualParametersCollection.deleteOne( + collections.virtualParameters.deleteOne( { _id: virtualParameterName }, (err) => { if (err) return void throwError(err, response); @@ -296,7 +313,7 @@ export function listener( const deviceId = querystring.unescape(r[1]); const tag = querystring.unescape(r[2]); if (request.method === "POST") { - db.devicesCollection.updateOne( + collections.devices.updateOne( { _id: deviceId }, { $addToSet: { _tags: tag } }, (err) => { @@ -306,7 +323,7 @@ export function listener( } ); } else if (request.method === "DELETE") { - db.devicesCollection.updateOne( + collections.devices.updateOne( { _id: deviceId }, { $pull: { _tags: tag } }, (err) => { @@ -327,12 +344,12 @@ export function listener( ); const deviceId = faultId.split(":", 1)[0]; const channel = faultId.slice(deviceId.length + 1); - db.faultsCollection.deleteOne({ _id: faultId }, (err) => { + collections.faults.deleteOne({ _id: faultId }, (err) => { if (err) return void throwError(err, response); if (channel.startsWith("task_")) { - const objId = new mongodb.ObjectID(channel.slice(5)); - return void db.tasksCollection.deleteOne({ _id: objId }, (err) => { + const objId = new ObjectId(channel.slice(5)); + return void collections.tasks.deleteOne({ _id: objId }, (err) => { if (err) return void throwError(err, response); cache @@ -404,7 +421,7 @@ export function listener( ); response.end(JSON.stringify(task)); } else if (status === "fault") { - db.tasksCollection.findOne( + collections.tasks.findOne( { _id: task._id }, (err, task2) => { if (err) @@ -479,8 +496,8 @@ export function listener( const action = r[2]; if (!action || action === "/") { if (request.method === "DELETE") { - db.tasksCollection.findOne( - { _id: new mongodb.ObjectID(taskId) }, + collections.tasks.findOne( + { _id: new ObjectId(taskId) }, { projection: { device: 1 } }, (err, task) => { if (err) return void throwError(err, response); @@ -492,12 +509,12 @@ export function listener( } const deviceId = task.device; - db.tasksCollection.deleteOne( - { _id: new mongodb.ObjectID(taskId) }, + collections.tasks.deleteOne( + { _id: new ObjectId(taskId) }, (err) => { if (err) return void throwError(err, response); - db.faultsCollection.deleteOne( + collections.faults.deleteOne( { _id: `${deviceId}:task_${taskId}` }, (err) => { if (err) return void throwError(err, response); @@ -525,14 +542,14 @@ export function listener( } } else if (action === "/retry") { if (request.method === "POST") { - db.tasksCollection.findOne( - { _id: new mongodb.ObjectID(taskId) }, + collections.tasks.findOne( + { _id: new ObjectId(taskId) }, { projection: { device: 1 } }, (err, task) => { if (err) return void throwError(err, response); const deviceId = task.device; - db.faultsCollection.deleteOne( + collections.faults.deleteOne( { _id: `${deviceId}:task_${taskId}` }, (err) => { if (err) return void throwError(err, response); @@ -571,9 +588,8 @@ export function listener( productClass: request.headers.productclass, version: request.headers.version, }; - const bucket = new mongodb.GridFSBucket(db.client.db()); - bucket.delete((filename as unknown) as mongodb.ObjectId, () => { - const uploadStream = bucket.openUploadStreamWithId( + filesBucket.delete((filename as unknown) as ObjectId, () => { + const uploadStream = filesBucket.openUploadStreamWithId( filename, filename, { @@ -591,8 +607,7 @@ export function listener( }); }); } else if (request.method === "DELETE") { - const bucket = new mongodb.GridFSBucket(db.client.db()); - bucket.delete((filename as unknown) as mongodb.ObjectId, (err) => { + filesBucket.delete((filename as unknown) as ObjectId, (err) => { if (err) { if (err.message.startsWith("FileNotFound")) { response.writeHead(404); @@ -669,7 +684,7 @@ export function listener( return; } - const collection = db[`${collectionName}Collection`]; + const collection = collections[collectionName]; if (!collection) { response.writeHead(404); response.end("404 Not Found"); @@ -693,7 +708,7 @@ export function listener( break; case "tasks": q = query.sanitizeQueryTypes(q, { - _id: (v) => new mongodb.ObjectID(v as string), + _id: (v) => new ObjectId(v as string), timestamp: (v) => new Date(v as number), retries: Number, }); diff --git a/lib/ui/db.ts b/lib/ui/db.ts index e32525e..d452367 100644 --- a/lib/ui/db.ts +++ b/lib/ui/db.ts @@ -17,85 +17,24 @@ * along with GenieACS. If not, see . */ -import { MongoClient, GridFSBucket, ObjectID } from "mongodb"; +import { Db, GridFSBucket, ObjectID } from "mongodb"; import { Script } from "vm"; -import * as config from "../config"; +import { onConnect } from "../db"; import * as mongodbFunctions from "../mongodb-functions"; import * as expression from "../common/expression"; import { QueryOptions, Expression } from "../types"; import { Readable } from "stream"; import { minimize } from "../common/boolean-expression"; -const CACHE_TTL = 300000; - -let clientPromise: Promise; - const RESOURCE_COLLECTION = { files: "fs.files", }; -function ensureIndexes(client): void { - client - .db() - .collection("cache") - .createIndex({ expire: 1 }, { expireAfterSeconds: 0 }); -} +let db: Db; -function getClient(): Promise { - if (!clientPromise) { - clientPromise = new Promise((resolve, reject) => { - const CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL"); - MongoClient.connect( - CONNECTION_URL, - { useNewUrlParser: true, useUnifiedTopology: true }, - (err, client) => { - if (err) return void reject(err); - ensureIndexes(client); - resolve(client); - } - ); - }); - } - - return clientPromise; -} - -export function cache( - key: string, - valueGetter: () => Promise, - ttl: number -): Promise { - return new Promise((resolve, reject) => { - getClient() - .then((client) => { - const collection = client.db().collection("cache"); - collection.findOne({ _id: key }, (err, doc) => { - if (err) return void reject(err); - if (doc != null) return void resolve(JSON.parse(doc.value) as T); - valueGetter() - .then((res) => { - const expire = Date.now() + (ttl || CACHE_TTL); - const cacheDoc = { - _id: key, - value: JSON.stringify(res), - expire: new Date(expire), - }; - collection.updateOne( - { _id: key }, - { $set: cacheDoc }, - { upsert: true }, - (err) => { - if (err) reject(err); - else resolve(res); - } - ); - }) - .catch(reject); - }); - }) - .catch(reject); - }); -} +onConnect(async (_db) => { + db = _db; +}); export function query( resource: string, @@ -128,80 +67,73 @@ export function query( } return new Promise((resolve, reject) => { - getClient() - .then((client) => { - const collection = client - .db() - .collection(RESOURCE_COLLECTION[resource] || resource); - const cursor = collection.find(q); - if (options.projection) { - let projection = options.projection; - if (resource === "devices") { - projection = mongodbFunctions.processDeviceProjection( - options.projection - ); - } + const collection = db.collection(RESOURCE_COLLECTION[resource] || resource); + const cursor = collection.find(q); + if (options.projection) { + let projection = options.projection; + if (resource === "devices") { + projection = mongodbFunctions.processDeviceProjection( + options.projection + ); + } - if (resource === "presets") projection.configurations = 1; - cursor.project(projection); + if (resource === "presets") projection.configurations = 1; + cursor.project(projection); + } + + if (resource === "users") cursor.project({ password: 0, salt: 0 }); + + if (options.skip) cursor.skip(options.skip); + if (options.limit) cursor.limit(options.limit); + + if (options.sort) { + let s = Object.entries(options.sort) + .sort((a, b) => Math.abs(b[1]) - Math.abs(a[1])) + .reduce( + (obj, [k, v]) => + Object.assign(obj, { [k]: Math.min(Math.max(v, -1), 1) }), + {} + ); + + if (resource === "devices") s = mongodbFunctions.processDeviceSort(s); + cursor.sort(s); + } + + if (!callback) { + cursor.toArray((err, docs) => { + if (err) return reject(err); + if (resource === "devices") + docs = docs.map((d) => mongodbFunctions.flattenDevice(d)); + else if (resource === "faults") + docs = docs.map((d) => mongodbFunctions.flattenFault(d)); + else if (resource === "tasks") + docs = docs.map((d) => mongodbFunctions.flattenTask(d)); + else if (resource === "presets") + docs = docs.map((d) => mongodbFunctions.flattenPreset(d)); + else if (resource === "files") + docs = docs.map((d) => mongodbFunctions.flattenFile(d)); + return resolve(docs); + }); + } else { + cursor.forEach( + (doc) => { + if (resource === "devices") doc = mongodbFunctions.flattenDevice(doc); + else if (resource === "faults") + doc = mongodbFunctions.flattenFault(doc); + else if (resource === "tasks") + doc = mongodbFunctions.flattenTask(doc); + else if (resource === "presets") + doc = mongodbFunctions.flattenPreset(doc); + else if (resource === "files") + doc = mongodbFunctions.flattenFile(doc); + callback(doc); + }, + (err) => { + if (err) reject(err); + else resolve(); } - - if (resource === "users") cursor.project({ password: 0, salt: 0 }); - - if (options.skip) cursor.skip(options.skip); - if (options.limit) cursor.limit(options.limit); - - if (options.sort) { - let s = Object.entries(options.sort) - .sort((a, b) => Math.abs(b[1]) - Math.abs(a[1])) - .reduce( - (obj, [k, v]) => - Object.assign(obj, { [k]: Math.min(Math.max(v, -1), 1) }), - {} - ); - - if (resource === "devices") s = mongodbFunctions.processDeviceSort(s); - cursor.sort(s); - } - - if (!callback) { - cursor.toArray((err, docs) => { - if (err) return reject(err); - if (resource === "devices") - docs = docs.map((d) => mongodbFunctions.flattenDevice(d)); - else if (resource === "faults") - docs = docs.map((d) => mongodbFunctions.flattenFault(d)); - else if (resource === "tasks") - docs = docs.map((d) => mongodbFunctions.flattenTask(d)); - else if (resource === "presets") - docs = docs.map((d) => mongodbFunctions.flattenPreset(d)); - else if (resource === "files") - docs = docs.map((d) => mongodbFunctions.flattenFile(d)); - return resolve(docs); - }); - } else { - cursor.forEach( - (doc) => { - if (resource === "devices") - doc = mongodbFunctions.flattenDevice(doc); - else if (resource === "faults") - doc = mongodbFunctions.flattenFault(doc); - else if (resource === "tasks") - doc = mongodbFunctions.flattenTask(doc); - else if (resource === "presets") - doc = mongodbFunctions.flattenPreset(doc); - else if (resource === "files") - doc = mongodbFunctions.flattenFile(doc); - callback(doc); - }, - (err) => { - if (err) reject(err); - else resolve(); - } - ); - } - }) - .catch(reject); + ); + } }); } @@ -223,17 +155,11 @@ export function count(resource: string, filter: Expression): Promise { } return new Promise((resolve, reject) => { - getClient() - .then((client) => { - const collection = client - .db() - .collection(RESOURCE_COLLECTION[resource] || resource); - collection.find(q).count((err, c) => { - if (err) reject(err); - else resolve(c); - }); - }) - .catch(reject); + const collection = db.collection(RESOURCE_COLLECTION[resource] || resource); + collection.find(q).count((err, c) => { + if (err) reject(err); + else resolve(c); + }); }); } @@ -250,8 +176,7 @@ export async function updateDeviceTags( else pull.push(tag); } - const client = await getClient(); - const collection = client.db().collection("devices"); + const collection = db.collection("devices"); const object = {}; if (add && add.length) object["$addToSet"] = { _tags: { $each: add } }; @@ -262,17 +187,11 @@ export async function updateDeviceTags( function putResource(resource, id, object): Promise { return new Promise((resolve, reject) => { - getClient() - .then((client) => { - const collection = client - .db() - .collection(RESOURCE_COLLECTION[resource] || resource); - collection.replaceOne({ _id: id }, object, { upsert: true }, (err) => { - if (err) return void reject(err); - resolve(); - }); - }) - .catch(reject); + const collection = db.collection(RESOURCE_COLLECTION[resource] || resource); + collection.replaceOne({ _id: id }, object, { upsert: true }, (err) => { + if (err) return void reject(err); + resolve(); + }); }); } @@ -281,17 +200,11 @@ function deleteResource( id: string | ObjectID ): Promise { return new Promise((resolve, reject) => { - getClient() - .then((client) => { - const collection = client - .db() - .collection(RESOURCE_COLLECTION[resource] || resource); - collection.deleteOne({ _id: id }, (err) => { - if (err) return void reject(err); - resolve(); - }); - }) - .catch(reject); + const collection = db.collection(RESOURCE_COLLECTION[resource] || resource); + collection.deleteOne({ _id: id }, (err) => { + if (err) return void reject(err); + resolve(); + }); }); } @@ -368,21 +281,17 @@ export function putUser( object: Record ): Promise { return new Promise((resolve, reject) => { - getClient() - .then((client) => { - const collection = client.db().collection("users"); - // update instead of replace to keep the password if not set by user - collection.updateOne( - { _id: id }, - { $set: object }, - { upsert: true }, - (err) => { - if (err) return void reject(err); - resolve(); - } - ); - }) - .catch(reject); + const collection = db.collection("users"); + // update instead of replace to keep the password if not set by user + collection.updateOne( + { _id: id }, + { $set: object }, + { upsert: true }, + (err) => { + if (err) return void reject(err); + resolve(); + } + ); }); } @@ -396,32 +305,24 @@ export function putFile( contentStream: Readable ): Promise { return new Promise((resolve, reject) => { - getClient() - .then((client) => { - const bucket = new GridFSBucket(client.db()); - const uploadStream = bucket.openUploadStreamWithId(filename, filename, { - metadata: metadata, - }); - uploadStream.on("error", reject); - contentStream.on("error", reject); - uploadStream.on("finish", resolve); - contentStream.pipe(uploadStream); - }) - .catch(reject); + const bucket = new GridFSBucket(db); + const uploadStream = bucket.openUploadStreamWithId(filename, filename, { + metadata: metadata, + }); + uploadStream.on("error", reject); + contentStream.on("error", reject); + uploadStream.on("finish", resolve); + contentStream.pipe(uploadStream); }); } export function deleteFile(filename: string): Promise { return new Promise((resolve, reject) => { - getClient() - .then((client) => { - const bucket = new GridFSBucket(client.db()); - bucket.delete(filename as any, (err) => { - if (err) return void reject(err); - resolve(); - }); - }) - .catch(reject); + const bucket = new GridFSBucket(db); + bucket.delete(filename as any, (err) => { + if (err) return void reject(err); + resolve(); + }); }); } @@ -432,7 +333,3 @@ export function deleteFault(id: string): Promise { export function deleteTask(id: ObjectID): Promise { return deleteResource("tasks", id); } - -export async function disconnect(): Promise { - if (clientPromise) await (await clientPromise).close(); -}