rodrigo/genieacs
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use a single MongoDB client

Browse Source
This commit is contained in:
Zaid Abdulla committed 2020-11-18 00:01:41 -08:00
1 parent bd55d6f512
commit a9b0e3a1f7
9 files changed
+100 -190
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

bin/genieacs-cwmp.ts
+2 -3
View File
@@ -24,7 +24,6 @@ import * as server from "../lib/server";
import * as cwmp from "../lib/cwmp"; import * as cwmp from "../lib/cwmp";
import * as db from "../lib/db"; import * as db from "../lib/db";
import * as extensions from "../lib/extensions"; import * as extensions from "../lib/extensions";
import * as cache from "../lib/cache";
import { version as VERSION } from "../package.json"; import { version as VERSION } from "../package.json";
logger.init("cwmp", VERSION); logger.init("cwmp", VERSION);
@@ -36,7 +35,6 @@ function exitWorkerGracefully(): void {
setTimeout(exitWorkerUngracefully, 5000).unref(); setTimeout(exitWorkerUngracefully, 5000).unref();
Promise.all([ Promise.all([
db.disconnect(), db.disconnect(),
cache.disconnect(),
extensions.killAll(), extensions.killAll(),
cluster.worker.disconnect(), cluster.worker.disconnect(),
]).catch(exitWorkerUngracefully); ]).catch(exitWorkerUngracefully);
@@ -99,7 +97,8 @@ if (!cluster.worker) {
server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully);
}); });
const initPromise = Promise.all([db.connect(), cache.connect()]) const initPromise = db
.connect()
.then(() => { .then(() => {
server.start(options, cwmp.listener); server.start(options, cwmp.listener);
}) })
bin/genieacs-fs.ts
+5 -7
View File
@@ -23,7 +23,6 @@ import * as cluster from "../lib/cluster";
import * as server from "../lib/server"; import * as server from "../lib/server";
import { listener } from "../lib/fs"; import { listener } from "../lib/fs";
import * as db from "../lib/db"; import * as db from "../lib/db";
import * as cache from "../lib/cache";
import { version as VERSION } from "../package.json"; import { version as VERSION } from "../package.json";
logger.init("fs", VERSION); logger.init("fs", VERSION);
@@ -33,11 +32,9 @@ const SERVICE_PORT = config.get("FS_PORT") as number;
function exitWorkerGracefully(): void { function exitWorkerGracefully(): void {
setTimeout(exitWorkerUngracefully, 5000).unref(); setTimeout(exitWorkerUngracefully, 5000).unref();
Promise.all([ Promise.all([db.disconnect(), cluster.worker.disconnect()]).catch(
db.disconnect(), exitWorkerUngracefully
cache.disconnect(), );
cluster.worker.disconnect(),
]).catch(exitWorkerUngracefully);
} }
function exitWorkerUngracefully(): void { function exitWorkerUngracefully(): void {
@@ -100,7 +97,8 @@ if (!cluster.worker) {
listener(req, res); listener(req, res);
}; };
const initPromise = Promise.all([db.connect(), cache.connect()]) const initPromise = db
.connect()
.then(() => { .then(() => {
server.start(options, _listener); server.start(options, _listener);
}) })
bin/genieacs-nbi.ts
+2 -3
View File
@@ -24,7 +24,6 @@ import * as server from "../lib/server";
import { listener } from "../lib/nbi"; import { listener } from "../lib/nbi";
import * as db from "../lib/db"; import * as db from "../lib/db";
import * as extensions from "../lib/extensions"; import * as extensions from "../lib/extensions";
import * as cache from "../lib/cache";
import { version as VERSION } from "../package.json"; import { version as VERSION } from "../package.json";
logger.init("nbi", VERSION); logger.init("nbi", VERSION);
@@ -36,7 +35,6 @@ function exitWorkerGracefully(): void {
setTimeout(exitWorkerUngracefully, 5000).unref(); setTimeout(exitWorkerUngracefully, 5000).unref();
Promise.all([ Promise.all([
db.disconnect(), db.disconnect(),
cache.disconnect(),
extensions.killAll(), extensions.killAll(),
cluster.worker.disconnect(), cluster.worker.disconnect(),
]).catch(exitWorkerUngracefully); ]).catch(exitWorkerUngracefully);
@@ -104,7 +102,8 @@ if (!cluster.worker) {
listener(req, res); listener(req, res);
}; };
const initPromise = Promise.all([db.connect(), cache.connect()]) const initPromise = db
.connect()
.then(() => { .then(() => {
server.start(options, _listener); server.start(options, _listener);
}) })
bin/genieacs-ui.ts
+3 -6
View File
@@ -23,9 +23,7 @@ import * as cluster from "../lib/cluster";
import * as server from "../lib/server"; import * as server from "../lib/server";
import { listener } from "../lib/ui"; import { listener } from "../lib/ui";
import * as extensions from "../lib/extensions"; import * as extensions from "../lib/extensions";
import * as db from "../lib/ui/db"; import * as db from "../lib/db";
import * as db2 from "../lib/db";
import * as cache from "../lib/cache";
import { version as VERSION } from "../package.json"; import { version as VERSION } from "../package.json";
logger.init("ui", VERSION); logger.init("ui", VERSION);
@@ -37,8 +35,6 @@ function exitWorkerGracefully(): void {
setTimeout(exitWorkerUngracefully, 5000).unref(); setTimeout(exitWorkerUngracefully, 5000).unref();
Promise.all([ Promise.all([
db.disconnect(), db.disconnect(),
db2.disconnect(),
cache.disconnect(),
extensions.killAll(), extensions.killAll(),
cluster.worker.disconnect(), cluster.worker.disconnect(),
]).catch(exitWorkerUngracefully); ]).catch(exitWorkerUngracefully);
@@ -106,7 +102,8 @@ if (!cluster.worker) {
listener(req, res); listener(req, res);
}; };
const initPromise = Promise.all([db2.connect(), cache.connect()]) const initPromise = db
.connect()
.then(() => { .then(() => {
server.start(options, _listener); server.start(options, _listener);
}) })
lib/cache.ts
+4 -14
View File
@@ -17,32 +17,22 @@
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>. * along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/ */
import { MongoClient, Collection } from "mongodb"; import { Collection } from "mongodb";
import { onConnect } from "./db";
import * as config from "./config"; import * as config from "./config";
const MAX_CACHE_TTL = +config.get("MAX_CACHE_TTL"); const MAX_CACHE_TTL = +config.get("MAX_CACHE_TTL");
let clientPromise: Promise<MongoClient>;
let mongoCollection: Collection; let mongoCollection: Collection;
let mongoTimeOffset = 0; let mongoTimeOffset = 0;
export async function connect(): Promise<void> { onConnect(async (db) => {
const MONGODB_CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL");
clientPromise = MongoClient.connect(MONGODB_CONNECTION_URL, {
useNewUrlParser: true,
useUnifiedTopology: true,
});
const db = (await clientPromise).db();
mongoCollection = db.collection("cache"); mongoCollection = db.collection("cache");
await mongoCollection.createIndex({ expire: 1 }, { expireAfterSeconds: 0 }); await mongoCollection.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
const now = Date.now(); const now = Date.now();
const res = await db.command({ hostInfo: 1 }); const res = await db.command({ hostInfo: 1 });
mongoTimeOffset = res.system.currentTime.getTime() - now; mongoTimeOffset = res.system.currentTime.getTime() - now;
} });
export async function disconnect(): Promise<void> {
if (clientPromise) await (await clientPromise).close();
}
export async function get(key: string | string[]): Promise<any> { export async function get(key: string | string[]): Promise<any> {
const expire = new Date(Date.now() - mongoTimeOffset); const expire = new Date(Date.now() - mongoTimeOffset);
lib/db.ts
+12 -4
View File
@@ -17,7 +17,7 @@
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>. * along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/ */
import { MongoClient, ObjectID, Collection } from "mongodb"; import { MongoClient, ObjectID, Collection, Db } from "mongodb";
import { get } from "./config"; import { get } from "./config";
import { encodeTag, escapeRegExp } from "./common"; import { encodeTag, escapeRegExp } from "./common";
import { parse } from "./common/expression-parser"; import { parse } from "./common/expression-parser";
@@ -45,7 +45,6 @@ export let tasksCollection: Collection,
configCollection: Collection; configCollection: Collection;
let clientPromise: Promise<MongoClient>; let clientPromise: Promise<MongoClient>;
export let client: MongoClient;
function compareAccessLists(list1: string[], list2: string[]): boolean { function compareAccessLists(list1: string[], list2: string[]): boolean {
if (list1.length !== list2.length) return false; if (list1.length !== list2.length) return false;
@@ -53,15 +52,24 @@ function compareAccessLists(list1: string[], list2: string[]): boolean {
return true; return true;
} }
const onConnectCallbacks: ((db: Db) => Promise<void>)[] = [];
export function onConnect(callback: (db: Db) => Promise<void>): void {
onConnectCallbacks.push(callback);
}
export async function connect(): Promise<void> { export async function connect(): Promise<void> {
clientPromise = MongoClient.connect("" + get("MONGODB_CONNECTION_URL"), { clientPromise = MongoClient.connect("" + get("MONGODB_CONNECTION_URL"), {
useNewUrlParser: true, useNewUrlParser: true,
useUnifiedTopology: true, useUnifiedTopology: true,
}); });
client = await clientPromise; const client = await clientPromise;
const db = client.db(); const db = client.db();
await Promise.all(onConnectCallbacks.map((c) => c(db)));
}
onConnect(async (db) => {
tasksCollection = db.collection("tasks"); tasksCollection = db.collection("tasks");
await tasksCollection.createIndex({ device: 1, timestamp: 1 }); await tasksCollection.createIndex({ device: 1, timestamp: 1 });
@@ -76,7 +84,7 @@ export async function connect(): Promise<void> {
permissionsCollection = db.collection("permissions"); permissionsCollection = db.collection("permissions");
usersCollection = db.collection("users"); usersCollection = db.collection("users");
configCollection = db.collection("config"); configCollection = db.collection("config");
} });
export async function disconnect(): Promise<void> { export async function disconnect(): Promise<void> {
if (clientPromise) await (await clientPromise).close(); if (clientPromise) await (await clientPromise).close();
lib/fs.ts
+12 -5
View File
@@ -20,11 +20,19 @@
import * as url from "url"; import * as url from "url";
import * as querystring from "querystring"; import * as querystring from "querystring";
import { IncomingMessage, ServerResponse } from "http"; import { IncomingMessage, ServerResponse } from "http";
import { GridFSBucket } from "mongodb"; import { Collection, GridFSBucket } from "mongodb";
import * as db from "./db"; import { onConnect } from "./db";
import * as logger from "./logger"; import * as logger from "./logger";
import { getRequestOrigin } from "./forwarded"; 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( export function listener(
request: IncomingMessage, request: IncomingMessage,
response: ServerResponse response: ServerResponse
@@ -39,7 +47,7 @@ export function listener(
remoteAddress: getRequestOrigin(request).remoteAddress, remoteAddress: getRequestOrigin(request).remoteAddress,
}; };
db.filesCollection.findOne({ _id: filename }, (err, file) => { filesCollection.findOne({ _id: filename }, (err, file) => {
if (err) throw err; if (err) throw err;
if (!file) { if (!file) {
@@ -55,8 +63,7 @@ export function listener(
"Content-Length": file.length, "Content-Length": file.length,
}); });
const bucket = new GridFSBucket(db.client.db()); const downloadStream = filesBucket.openDownloadStreamByName(filename);
const downloadStream = bucket.openDownloadStreamByName(filename);
downloadStream.pipe(response); downloadStream.pipe(response);
logger.accessInfo(log); logger.accessInfo(log);
lib/nbi.ts
+46 -31
View File
@@ -18,11 +18,11 @@
*/ */
import * as url from "url"; import * as url from "url";
import * as mongodb from "mongodb"; import { Collection, GridFSBucket, ObjectId } from "mongodb";
import * as querystring from "querystring"; import * as querystring from "querystring";
import * as vm from "vm"; import * as vm from "vm";
import * as config from "./config"; import * as config from "./config";
import * as db from "./db"; import { onConnect } from "./db";
import * as query from "./query"; import * as query from "./query";
import * as apiFunctions from "./api-functions"; import * as apiFunctions from "./api-functions";
import { IncomingMessage, ServerResponse } from "http"; 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 VIRTUAL_PARAMETERS_REGEX = /^\/virtual_parameters\/([a-zA-Z0-9\-_%]+)\/?$/;
const FAULTS_REGEX = /^\/faults\/([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 { function throwError(err: Error, httpResponse?: ServerResponse): never {
if (httpResponse) { if (httpResponse) {
httpResponse.writeHead(500, { Connection: "close" }); httpResponse.writeHead(500, { Connection: "close" });
@@ -85,7 +102,7 @@ export function listener(
if (request.method === "PUT") { if (request.method === "PUT") {
const preset = JSON.parse(body.toString()); const preset = JSON.parse(body.toString());
preset._id = presetName; preset._id = presetName;
db.presetsCollection.replaceOne( collections.presets.replaceOne(
{ _id: presetName }, { _id: presetName },
preset, preset,
{ upsert: true }, { upsert: true },
@@ -106,7 +123,7 @@ export function listener(
} }
); );
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
db.presetsCollection.deleteOne({ _id: presetName }, (err) => { collections.presets.deleteOne({ _id: presetName }, (err) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache cache
@@ -132,7 +149,7 @@ export function listener(
if (request.method === "PUT") { if (request.method === "PUT") {
const object = JSON.parse(body.toString()); const object = JSON.parse(body.toString());
object._id = objectName; object._id = objectName;
db.objectsCollection.replaceOne( collections.objects.replaceOne(
{ _id: objectName }, { _id: objectName },
object, object,
{ upsert: true }, { upsert: true },
@@ -153,7 +170,7 @@ export function listener(
} }
); );
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
db.objectsCollection.deleteOne({ _id: objectName }, (err) => { collections.objects.deleteOne({ _id: objectName }, (err) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache cache
@@ -190,7 +207,7 @@ export function listener(
return; return;
} }
db.provisionsCollection.replaceOne( collections.provisions.replaceOne(
{ _id: provisionName }, { _id: provisionName },
object, object,
{ upsert: true }, { upsert: true },
@@ -211,7 +228,7 @@ export function listener(
} }
); );
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
db.provisionsCollection.deleteOne({ _id: provisionName }, (err) => { collections.provisions.deleteOne({ _id: provisionName }, (err) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache cache
@@ -248,7 +265,7 @@ export function listener(
return; return;
} }
db.virtualParametersCollection.replaceOne( collections.virtualParameters.replaceOne(
{ _id: virtualParameterName }, { _id: virtualParameterName },
object, object,
{ upsert: true }, { upsert: true },
@@ -269,7 +286,7 @@ export function listener(
} }
); );
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
db.virtualParametersCollection.deleteOne( collections.virtualParameters.deleteOne(
{ _id: virtualParameterName }, { _id: virtualParameterName },
(err) => { (err) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
@@ -296,7 +313,7 @@ export function listener(
const deviceId = querystring.unescape(r[1]); const deviceId = querystring.unescape(r[1]);
const tag = querystring.unescape(r[2]); const tag = querystring.unescape(r[2]);
if (request.method === "POST") { if (request.method === "POST") {
db.devicesCollection.updateOne( collections.devices.updateOne(
{ _id: deviceId }, { _id: deviceId },
{ $addToSet: { _tags: tag } }, { $addToSet: { _tags: tag } },
(err) => { (err) => {
@@ -306,7 +323,7 @@ export function listener(
} }
); );
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
db.devicesCollection.updateOne( collections.devices.updateOne(
{ _id: deviceId }, { _id: deviceId },
{ $pull: { _tags: tag } }, { $pull: { _tags: tag } },
(err) => { (err) => {
@@ -327,12 +344,12 @@ export function listener(
); );
const deviceId = faultId.split(":", 1)[0]; const deviceId = faultId.split(":", 1)[0];
const channel = faultId.slice(deviceId.length + 1); 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 (err) return void throwError(err, response);
if (channel.startsWith("task_")) { if (channel.startsWith("task_")) {
const objId = new mongodb.ObjectID(channel.slice(5)); const objId = new ObjectId(channel.slice(5));
return void db.tasksCollection.deleteOne({ _id: objId }, (err) => { return void collections.tasks.deleteOne({ _id: objId }, (err) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache cache
@@ -404,7 +421,7 @@ export function listener(
); );
response.end(JSON.stringify(task)); response.end(JSON.stringify(task));
} else if (status === "fault") { } else if (status === "fault") {
db.tasksCollection.findOne( collections.tasks.findOne(
{ _id: task._id }, { _id: task._id },
(err, task2) => { (err, task2) => {
if (err) if (err)
@@ -479,8 +496,8 @@ export function listener(
const action = r[2]; const action = r[2];
if (!action || action === "/") { if (!action || action === "/") {
if (request.method === "DELETE") { if (request.method === "DELETE") {
db.tasksCollection.findOne( collections.tasks.findOne(
{ _id: new mongodb.ObjectID(taskId) }, { _id: new ObjectId(taskId) },
{ projection: { device: 1 } }, { projection: { device: 1 } },
(err, task) => { (err, task) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
@@ -492,12 +509,12 @@ export function listener(
} }
const deviceId = task.device; const deviceId = task.device;
db.tasksCollection.deleteOne( collections.tasks.deleteOne(
{ _id: new mongodb.ObjectID(taskId) }, { _id: new ObjectId(taskId) },
(err) => { (err) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
db.faultsCollection.deleteOne( collections.faults.deleteOne(
{ _id: `${deviceId}:task_${taskId}` }, { _id: `${deviceId}:task_${taskId}` },
(err) => { (err) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
@@ -525,14 +542,14 @@ export function listener(
} }
} else if (action === "/retry") { } else if (action === "/retry") {
if (request.method === "POST") { if (request.method === "POST") {
db.tasksCollection.findOne( collections.tasks.findOne(
{ _id: new mongodb.ObjectID(taskId) }, { _id: new ObjectId(taskId) },
{ projection: { device: 1 } }, { projection: { device: 1 } },
(err, task) => { (err, task) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
const deviceId = task.device; const deviceId = task.device;
db.faultsCollection.deleteOne( collections.faults.deleteOne(
{ _id: `${deviceId}:task_${taskId}` }, { _id: `${deviceId}:task_${taskId}` },
(err) => { (err) => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
@@ -571,9 +588,8 @@ export function listener(
productClass: request.headers.productclass, productClass: request.headers.productclass,
version: request.headers.version, version: request.headers.version,
}; };
const bucket = new mongodb.GridFSBucket(db.client.db()); filesBucket.delete((filename as unknown) as ObjectId, () => {
bucket.delete((filename as unknown) as mongodb.ObjectId, () => { const uploadStream = filesBucket.openUploadStreamWithId(
const uploadStream = bucket.openUploadStreamWithId(
filename, filename,
filename, filename,
{ {
@@ -591,8 +607,7 @@ export function listener(
}); });
}); });
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
const bucket = new mongodb.GridFSBucket(db.client.db()); filesBucket.delete((filename as unknown) as ObjectId, (err) => {
bucket.delete((filename as unknown) as mongodb.ObjectId, (err) => {
if (err) { if (err) {
if (err.message.startsWith("FileNotFound")) { if (err.message.startsWith("FileNotFound")) {
response.writeHead(404); response.writeHead(404);
@@ -669,7 +684,7 @@ export function listener(
return; return;
} }
const collection = db[`${collectionName}Collection`]; const collection = collections[collectionName];
if (!collection) { if (!collection) {
response.writeHead(404); response.writeHead(404);
response.end("404 Not Found"); response.end("404 Not Found");
@@ -693,7 +708,7 @@ export function listener(
break; break;
case "tasks": case "tasks":
q = query.sanitizeQueryTypes(q, { 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), timestamp: (v) => new Date(v as number),
retries: Number, retries: Number,
}); });
lib/ui/db.ts
+14 -117
View File
@@ -17,85 +17,24 @@
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>. * along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/ */
import { MongoClient, GridFSBucket, ObjectID } from "mongodb"; import { Db, GridFSBucket, ObjectID } from "mongodb";
import { Script } from "vm"; import { Script } from "vm";
import * as config from "../config"; import { onConnect } from "../db";
import * as mongodbFunctions from "../mongodb-functions"; import * as mongodbFunctions from "../mongodb-functions";
import * as expression from "../common/expression"; import * as expression from "../common/expression";
import { QueryOptions, Expression } from "../types"; import { QueryOptions, Expression } from "../types";
import { Readable } from "stream"; import { Readable } from "stream";
import { minimize } from "../common/boolean-expression"; import { minimize } from "../common/boolean-expression";
const CACHE_TTL = 300000;
let clientPromise: Promise<MongoClient>;
const RESOURCE_COLLECTION = { const RESOURCE_COLLECTION = {
files: "fs.files", files: "fs.files",
}; };
function ensureIndexes(client): void { let db: Db;
client
.db()
.collection("cache")
.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
}
function getClient(): Promise<MongoClient> { onConnect(async (_db) => {
if (!clientPromise) { db = _db;
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<T>(
key: string,
valueGetter: () => Promise<T>,
ttl: number
): Promise<T> {
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);
});
}
export function query( export function query(
resource: string, resource: string,
@@ -128,11 +67,7 @@ export function query(
} }
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
getClient() const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
.then((client) => {
const collection = client
.db()
.collection(RESOURCE_COLLECTION[resource] || resource);
const cursor = collection.find(q); const cursor = collection.find(q);
if (options.projection) { if (options.projection) {
let projection = options.projection; let projection = options.projection;
@@ -182,8 +117,7 @@ export function query(
} else { } else {
cursor.forEach( cursor.forEach(
(doc) => { (doc) => {
if (resource === "devices") if (resource === "devices") doc = mongodbFunctions.flattenDevice(doc);
doc = mongodbFunctions.flattenDevice(doc);
else if (resource === "faults") else if (resource === "faults")
doc = mongodbFunctions.flattenFault(doc); doc = mongodbFunctions.flattenFault(doc);
else if (resource === "tasks") else if (resource === "tasks")
@@ -200,8 +134,6 @@ export function query(
} }
); );
} }
})
.catch(reject);
}); });
} }
@@ -223,17 +155,11 @@ export function count(resource: string, filter: Expression): Promise<number> {
} }
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
getClient() const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
.then((client) => {
const collection = client
.db()
.collection(RESOURCE_COLLECTION[resource] || resource);
collection.find(q).count((err, c) => { collection.find(q).count((err, c) => {
if (err) reject(err); if (err) reject(err);
else resolve(c); else resolve(c);
}); });
})
.catch(reject);
}); });
} }
@@ -250,8 +176,7 @@ export async function updateDeviceTags(
else pull.push(tag); else pull.push(tag);
} }
const client = await getClient(); const collection = db.collection("devices");
const collection = client.db().collection("devices");
const object = {}; const object = {};
if (add && add.length) object["$addToSet"] = { _tags: { $each: add } }; if (add && add.length) object["$addToSet"] = { _tags: { $each: add } };
@@ -262,17 +187,11 @@ export async function updateDeviceTags(
function putResource(resource, id, object): Promise<void> { function putResource(resource, id, object): Promise<void> {
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
getClient() const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
.then((client) => {
const collection = client
.db()
.collection(RESOURCE_COLLECTION[resource] || resource);
collection.replaceOne({ _id: id }, object, { upsert: true }, (err) => { collection.replaceOne({ _id: id }, object, { upsert: true }, (err) => {
if (err) return void reject(err); if (err) return void reject(err);
resolve(); resolve();
}); });
})
.catch(reject);
}); });
} }
@@ -281,17 +200,11 @@ function deleteResource(
id: string | ObjectID id: string | ObjectID
): Promise<void> { ): Promise<void> {
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
getClient() const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
.then((client) => {
const collection = client
.db()
.collection(RESOURCE_COLLECTION[resource] || resource);
collection.deleteOne({ _id: id }, (err) => { collection.deleteOne({ _id: id }, (err) => {
if (err) return void reject(err); if (err) return void reject(err);
resolve(); resolve();
}); });
})
.catch(reject);
}); });
} }
@@ -368,9 +281,7 @@ export function putUser(
object: Record<string, unknown> object: Record<string, unknown>
): Promise<void> { ): Promise<void> {
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
getClient() const collection = db.collection("users");
.then((client) => {
const collection = client.db().collection("users");
// update instead of replace to keep the password if not set by user // update instead of replace to keep the password if not set by user
collection.updateOne( collection.updateOne(
{ _id: id }, { _id: id },
@@ -381,8 +292,6 @@ export function putUser(
resolve(); resolve();
} }
); );
})
.catch(reject);
}); });
} }
@@ -396,9 +305,7 @@ export function putFile(
contentStream: Readable contentStream: Readable
): Promise<void> { ): Promise<void> {
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
getClient() const bucket = new GridFSBucket(db);
.then((client) => {
const bucket = new GridFSBucket(client.db());
const uploadStream = bucket.openUploadStreamWithId(filename, filename, { const uploadStream = bucket.openUploadStreamWithId(filename, filename, {
metadata: metadata, metadata: metadata,
}); });
@@ -406,22 +313,16 @@ export function putFile(
contentStream.on("error", reject); contentStream.on("error", reject);
uploadStream.on("finish", resolve); uploadStream.on("finish", resolve);
contentStream.pipe(uploadStream); contentStream.pipe(uploadStream);
})
.catch(reject);
}); });
} }
export function deleteFile(filename: string): Promise<void> { export function deleteFile(filename: string): Promise<void> {
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
getClient() const bucket = new GridFSBucket(db);
.then((client) => {
const bucket = new GridFSBucket(client.db());
bucket.delete(filename as any, (err) => { bucket.delete(filename as any, (err) => {
if (err) return void reject(err); if (err) return void reject(err);
resolve(); resolve();
}); });
})
.catch(reject);
}); });
} }
@@ -432,7 +333,3 @@ export function deleteFault(id: string): Promise<void> {
export function deleteTask(id: ObjectID): Promise<void> { export function deleteTask(id: ObjectID): Promise<void> {
return deleteResource("tasks", id); return deleteResource("tasks", id);
} }
export async function disconnect(): Promise<void> {
if (clientPromise) await (await clientPromise).close();
}