Refactor DB code to use promises

build/build.ts

@@ -61,6 +61,7 @@ const builtins = [
   "url",
   "readline",
   "stream",
+  "bson", // not a built-in but not a direct dependency either
 ];
 
 function rmDirSync(dirPath): void {

lib/api-functions.ts

@@ -17,7 +17,7 @@
  * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-import * as db from "./db";
+import { collections } from "./db";
 import * as common from "./common";
 import * as cache from "./cache";
 import {
@@ -38,7 +38,7 @@ export async function connectionRequest(
   device?: Record<string, { value?: [boolean | number | string, string] }>
 ): Promise<string> {
   if (!device) {
-    const res = await db.devicesCollection.findOne({ _id: deviceId });
+    const res = await collections.devices.findOne({ _id: deviceId });
     if (!res) throw new Error("No such device");
     device = flattenDevice(res);
   }
@@ -184,7 +184,7 @@ export async function awaitSessionStart(
   timeout: number
 ): Promise<boolean> {
   const now = Date.now();
-  const device = await db.devicesCollection.findOne(
+  const device = await collections.devices.findOne(
     { _id: deviceId },
     { projection: { _lastInform: 1 } }
   );
@@ -322,26 +322,27 @@ export async function insertTasks(tasks: any[]): Promise<Task[]> {
   for (const task of tasks) {
     sanitizeTask(task);
     if (task.uniqueKey) {
-      await db.tasksCollection.deleteOne({
+      await collections.tasks.deleteOne({
         device: task.device,
         uniqueKey: task.uniqueKey,
       });
     }
   }
-  await db.tasksCollection.insertMany(tasks);
+  await collections.tasks.insertMany(tasks);
+  for (const task of tasks) task._id = task._id.toString();
   return tasks;
 }
 
 export async function deleteDevice(deviceId: string): Promise<void> {
   await Promise.all([
-    db.tasksCollection.deleteMany({ device: deviceId }),
+    collections.tasks.deleteMany({ device: deviceId }),
-    db.devicesCollection.deleteOne({ _id: deviceId }),
+    collections.devices.deleteOne({ _id: deviceId }),
-    db.faultsCollection.deleteMany({
+    collections.faults.deleteMany({
       _id: {
         $regex: `^${common.escapeRegExp(deviceId)}\\:`,
       },
     }),
-    db.operationsCollection.deleteMany({
+    collections.operations.deleteMany({
       _id: {
         $regex: `^${common.escapeRegExp(deviceId)}\\:`,
       },

lib/cwmp.ts

@@ -1473,10 +1473,15 @@ async function listenerAsync(
 
   const chunks: Buffer[] = [];
   try {
+    let readableEnded = false;
+    stream.on("end", () => {
+      readableEnded = true;
+    });
     for await (const chunk of stream) chunks.push(chunk);
     // In Node versions prior to 15, the stream will not emit an error if the
     // connection is closed before the stream is finished.
-    if (!stream.readableEnded) throw new Error("Connection closed");
+    // For Node 12.9+ we can just use stream.readableEnded
+    if (!readableEnded) throw new Error("Connection closed");
   } catch (err) {
     return;
   }

lib/db.ts

@@ -32,18 +32,20 @@ import {
 import Path from "./common/path";
 import * as MongoTypes from "./mongodb-types";
 
-export let tasksCollection: Collection<MongoTypes.Task>,
+export const collections = {
-  devicesCollection: Collection,
+  devices: null as Collection<MongoTypes.Device>,
-  presetsCollection: Collection,
+  presets: null as Collection<MongoTypes.Preset>,
-  objectsCollection: Collection,
+  objects: null as Collection<MongoTypes.Object>,
-  provisionsCollection: Collection,
+  provisions: null as Collection<MongoTypes.Provision>,
-  virtualParametersCollection: Collection,
+  virtualParameters: null as Collection<MongoTypes.VirtualParameter>,
-  faultsCollection: Collection<MongoTypes.Fault>,
+  faults: null as Collection<MongoTypes.Fault>,
-  filesCollection: Collection,
+  tasks: null as Collection<MongoTypes.Task>,
-  operationsCollection: Collection<MongoTypes.Operation>,
+  files: null as Collection<MongoTypes.File>,
-  permissionsCollection: Collection,
+  operations: null as Collection<MongoTypes.Operation>,
-  usersCollection: Collection,
+  permissions: null as Collection<MongoTypes.Permission>,
-  configCollection: Collection<MongoTypes.Config>;
+  users: null as Collection<MongoTypes.User>,
+  config: null as Collection<MongoTypes.Config>,
+};
 
 let clientPromise: Promise<MongoClient>;
 
@@ -70,20 +72,20 @@ export async function connect(): Promise<void> {
 }
 
 onConnect(async (db) => {
-  tasksCollection = db.collection("tasks");
+  collections.tasks = db.collection("tasks");
-  await tasksCollection.createIndex({ device: 1, timestamp: 1 });
+  await collections.tasks.createIndex({ device: 1, timestamp: 1 });
 
-  devicesCollection = db.collection("devices");
+  collections.devices = db.collection("devices");
-  presetsCollection = db.collection("presets");
+  collections.presets = db.collection("presets");
-  objectsCollection = db.collection("objects");
+  collections.objects = db.collection("objects");
-  filesCollection = db.collection("fs.files");
+  collections.files = db.collection("fs.files");
-  provisionsCollection = db.collection("provisions");
+  collections.provisions = db.collection("provisions");
-  virtualParametersCollection = db.collection("virtualParameters");
+  collections.virtualParameters = db.collection("virtualParameters");
-  faultsCollection = db.collection("faults");
+  collections.faults = db.collection("faults");
-  operationsCollection = db.collection("operations");
+  collections.operations = db.collection("operations");
-  permissionsCollection = db.collection("permissions");
+  collections.permissions = db.collection("permissions");
-  usersCollection = db.collection("users");
+  collections.users = db.collection("users");
-  configCollection = db.collection("config");
+  collections.config = db.collection("config");
 });
 
 export async function disconnect(): Promise<void> {
@@ -130,7 +132,7 @@ export async function fetchDevice(
     ],
   ];
 
-  const device = await devicesCollection.findOne({ _id: id });
+  const device = await collections.devices.findOne({ _id: id });
   if (!device) return null;
 
   function storeParams(
@@ -194,7 +196,7 @@ export async function fetchDevice(
       res.push([Path.parse(path + ".*"), obj["_timestamp"]]);
   }
 
-  const ts: number = device["_timestamp"] || 0;
+  const ts: number = +device["_timestamp"] || 0;
   if (ts) res.push([Path.parse("*"), ts]);
 
   for (const [k, v] of Object.entries(device)) {
@@ -621,15 +623,19 @@ export async function saveDevice(
     delete update["$pull"];
   }
 
-  const result = await devicesCollection.updateOne({ _id: deviceId }, update, {
+  const result = await collections.devices.updateOne(
+    { _id: deviceId },
+    update,
+    {
       upsert: isNew,
-  });
+    }
+  );
 
   if (!result.matchedCount && !result.upsertedCount)
     throw new Error(`Device ${deviceId} not found in database`);
 
   if (update2) {
-    await devicesCollection.updateOne({ _id: deviceId }, update2);
+    await collections.devices.updateOne({ _id: deviceId }, update2);
     return;
   }
 }
@@ -637,7 +643,7 @@ export async function saveDevice(
 export async function getFaults(
   deviceId: string
 ): Promise<{ [channel: string]: SessionFault }> {
-  const res = await faultsCollection
+  const res = await collections.faults
     .find({ _id: { $regex: `^${escapeRegExp(deviceId)}\\:` } })
     .toArray();
 
@@ -677,21 +683,23 @@ export async function saveFault(
     ...(fault.expiry && { expiry: new Date(fault.expiry) }),
     provisions: JSON.stringify(fault.provisions),
   };
-  await faultsCollection.replaceOne({ _id: id }, f, { upsert: true });
+  await collections.faults.replaceOne({ _id: id }, f, { upsert: true });
 }
 
 export async function deleteFault(
   deviceId: string,
   channel: string
 ): Promise<void> {
-  await faultsCollection.deleteOne({ _id: `${deviceId}:${channel}` });
+  await collections.faults.deleteOne({ _id: `${deviceId}:${channel}` });
 }
 
 export async function getDueTasks(
   deviceId: string,
   timestamp: number
 ): Promise<[Task[], number]> {
-  const cur = tasksCollection.find({ device: deviceId }).sort({ timestamp: 1 });
+  const cur = collections.tasks
+    .find({ device: deviceId })
+    .sort({ timestamp: 1 });
   const tasks = [] as Task[];
 
   for await (const t of cur) {
@@ -736,7 +744,7 @@ export async function getDueTasks(
         q = { _id: { $in: [t["file"], new ObjectId(t["file"])] } };
       else q = { _id: t["file"] };
 
-      const res = await filesCollection.find(q).toArray();
+      const res = await collections.files.find(q).toArray();
 
       if (res[0]) {
         if (!task.fileType) task.fileType = res[0].metadata.fileType;
@@ -753,7 +761,7 @@ export async function clearTasks(
   deviceId: string,
   taskIds: string[]
 ): Promise<void> {
-  await tasksCollection.deleteMany({
+  await collections.tasks.deleteMany({
     _id: { $in: taskIds.map((id) => new ObjectId(id)) },
   });
 }
@@ -761,7 +769,7 @@ export async function clearTasks(
 export async function getOperations(
   deviceId: string
 ): Promise<{ [commandKey: string]: Operation }> {
-  const res = await operationsCollection
+  const res = await collections.operations
     .find({ _id: { $regex: `^${escapeRegExp(deviceId)}\\:` } })
     .toArray();
 
@@ -803,7 +811,7 @@ export async function saveOperation(
     retries: JSON.stringify(operation.retries),
     args: JSON.stringify(operation.args),
   };
-  await operationsCollection.replaceOne({ _id: id }, o, {
+  await collections.operations.replaceOne({ _id: id }, o, {
     upsert: true,
   });
 }
@@ -812,33 +820,33 @@ export async function deleteOperation(
   deviceId: string,
   commandKey: string
 ): Promise<void> {
-  await operationsCollection.deleteOne({ _id: `${deviceId}:${commandKey}` });
+  await collections.operations.deleteOne({ _id: `${deviceId}:${commandKey}` });
 }
 
 export async function getPresets(): Promise<Record<string, any>[]> {
-  return presetsCollection.find().toArray();
+  return collections.presets.find().toArray();
 }
 
 export async function getObjects(): Promise<Record<string, any>[]> {
-  return objectsCollection.find().toArray();
+  return collections.objects.find().toArray();
 }
 
 export async function getProvisions(): Promise<Record<string, any>[]> {
-  return provisionsCollection.find().toArray();
+  return collections.provisions.find().toArray();
 }
 
 export async function getVirtualParameters(): Promise<Record<string, any>[]> {
-  return virtualParametersCollection.find().toArray();
+  return collections.virtualParameters.find().toArray();
 }
 
 export function getFiles(): Promise<Record<string, any>[]> {
-  return filesCollection.find().toArray();
+  return collections.files.find().toArray();
 }
 
 export async function getConfig(): Promise<
   { id: string; value: Expression }[]
 > {
-  const res = await configCollection.find().toArray();
+  const res = await collections.config.find().toArray();
   return res.map((c) => ({
     id: c["_id"],
     value: parse(c["value"]),
@@ -854,7 +862,7 @@ interface Permission {
 }
 
 export async function getPermissions(): Promise<Permission[]> {
-  return permissionsCollection.find().toArray() as unknown as Permission[];
+  return collections.permissions.find().toArray() as unknown as Permission[];
 }
 
 interface User {
@@ -865,5 +873,5 @@ interface User {
 }
 
 export async function getUsers(): Promise<User[]> {
-  return usersCollection.find().toArray() as unknown as User[];
+  return collections.users.find().toArray() as unknown as User[];
 }

lib/fs.ts

@@ -20,17 +20,15 @@
 import * as url from "url";
 import { IncomingMessage, ServerResponse } from "http";
 import { PassThrough, pipeline, Readable } from "stream";
-import { Collection, GridFSBucket } from "mongodb";
+import { GridFSBucket } from "mongodb";
-import { onConnect } from "./db";
+import { onConnect, collections } from "./db";
 import * as logger from "./logger";
 import { getRequestOrigin } from "./forwarded";
 import memoize from "./common/memoize";
 
-let filesCollection: Collection;
 let filesBucket: GridFSBucket;
 
 onConnect(async (db) => {
-  filesCollection = db.collection("fs.files");
   filesBucket = new GridFSBucket(db);
 });
 
@@ -79,7 +77,7 @@ export async function listener(
     remoteAddress: getRequestOrigin(request).remoteAddress,
   };
 
-  const file = await filesCollection.findOne({ _id: filename });
+  const file = await collections.files.findOne({ _id: filename });
 
   if (!file) {
     response.writeHead(404);

lib/mongodb-types.ts

@@ -25,6 +25,7 @@ interface TaskBase {
   timestamp?: Date;
   expiry?: Date;
   name: string;
+  device: string;
 }
 
 interface TaskGetParameterValues extends TaskBase {
@@ -105,3 +106,61 @@ export interface Cache {
   timestamp: Date;
   expire: Date;
 }
+
+export interface Device {
+  _id: string;
+  _lastInform: Date;
+  _registered: Date;
+  _tags?: string[];
+  _timestamp?: Date;
+}
+
+export interface Preset {
+  _id: string;
+  weight: number;
+  channels: string;
+  provisions: string;
+  events: Record<string, boolean>;
+  configurations: { type: string; name: string; args: Expression[] }[];
+}
+
+export interface Object {
+  _id: string;
+}
+
+export interface Provision {
+  _id: string;
+  script: string;
+}
+
+export interface VirtualParameter {
+  _id: string;
+  script: string;
+}
+
+export interface File {
+  _id: string;
+  length: number;
+  filename: string;
+  metadata?: {
+    fileType?: string;
+    oui?: string;
+    productClass?: string;
+    version?: string;
+  };
+}
+
+export interface Permission {
+  _id: string;
+  role: string;
+  resource: string;
+  access: 1 | 2 | 3;
+  validate: string;
+}
+
+export interface User {
+  _id: string;
+  password: string;
+  roles: string;
+  salt: string;
+}

lib/nbi.ts

@@ -17,11 +17,10 @@
  * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-import * as url from "url";
 import { Collection, GridFSBucket, ObjectId } from "mongodb";
 import * as vm from "vm";
 import * as config from "./config";
-import { onConnect, optimizeProjection } from "./db";
+import { onConnect, optimizeProjection, collections } from "./db";
 import * as query from "./query";
 import * as apiFunctions from "./api-functions";
 import { IncomingMessage, ServerResponse } from "http";
@@ -30,7 +29,7 @@ import { version as VERSION } from "../package.json";
 import { ping } from "./ping";
 import * as logger from "./logger";
 import { flattenDevice } from "./mongodb-functions";
-import { getSocketEndpoints } from "./server";
+import { getRequestOrigin } from "./forwarded";
 
 const DEVICE_TASKS_REGEX = /^\/devices\/([a-zA-Z0-9\-_%]+)\/tasks\/?$/;
 const TASKS_REGEX = /^\/tasks\/([a-zA-Z0-9\-_%]+)(\/[a-zA-Z_]*)?$/;
@@ -47,74 +46,74 @@ const VIRTUAL_PARAMETERS_REGEX =
   /^\/virtual_parameters\/([a-zA-Z0-9\-_%]+)\/?$/;
 const FAULTS_REGEX = /^\/faults\/([a-zA-Z0-9\-_%:]+)\/?$/;
 
-const collections: Record<string, Collection> = {
-  tasks: null as Collection,
-  devices: null as Collection,
-  presets: null as Collection,
-  objects: null as Collection,
-  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)) {
-    if (k === "files") collections[k] = db.collection("fs.files");
-    else collections[k] = db.collection(k);
-  }
   filesBucket = new GridFSBucket(db);
 });
 
-function throwError(err: Error, httpResponse?: ServerResponse): never {
+async function getBody(request: IncomingMessage): Promise<Buffer> {
-  if (httpResponse) {
+  const chunks: Buffer[] = [];
-    httpResponse.writeHead(500, { Connection: "close" });
+  let readableEnded = false;
-    httpResponse.end(`${err.name}: ${err.message}`);
+  request.on("end", () => {
-  }
+    readableEnded = true;
-  throw err;
+  });
+  for await (const chunk of request) chunks.push(chunk);
+  // In Node versions prior to 15, the stream will not emit an error if the
+  // connection is closed before the stream is finished.
+  // For Node 12.9+ we can just use stream.readableEnded
+  if (!readableEnded) throw new Error("Connection closed");
+  return Buffer.concat(chunks);
 }
 
 export function listener(
   request: IncomingMessage,
   response: ServerResponse
 ): void {
-  const chunks = [];
-  let bytes = 0;
   response.setHeader("GenieACS-Version", VERSION);
 
-  request.addListener("data", (chunk): void => {
+  const origin = getRequestOrigin(request);
-    chunks.push(chunk);
+  const url = new URL(
-    bytes += chunk.length;
+    request.url,
+    (origin.encrypted ? "https://" : "http://") + origin.host
+  );
+
+  let readableEnded = false;
+  // For Node 12.9+ we can just use stream.readableEnded
+  request.on("end", () => {
+    readableEnded = true;
   });
 
-  function getBody(): Buffer {
+  getBody(request)
-    // Write all chunks into a Buffer
+    .then((body) => {
-    const body = Buffer.allocUnsafe(bytes);
-    let offset = 0;
-    for (const chunk of chunks) {
-      chunk.copy(body, offset, 0, chunk.length);
-      offset += chunk.length;
-    }
-    return body;
-  }
-
-  request.addListener("end", () => {
-    const body = getBody();
-    const urlParts = url.parse(request.url, true);
-    const socketEndpoints = getSocketEndpoints(request.socket);
-
       logger.accessInfo(
-      Object.assign({}, urlParts.query, {
+        Object.assign({}, Object.fromEntries(url.searchParams), {
-        remoteAddress: socketEndpoints.remoteAddress,
+          remoteAddress: origin.remoteAddress,
-        message: `${request.method} ${urlParts.pathname}`,
+          message: `${request.method} ${url.pathname}`,
         })
       );
+      return handler(request, response, url, body);
+    })
+    .catch((err) => {
+      // Ignore incomplete requests
+      if (!readableEnded) return;
 
-    if (PRESETS_REGEX.test(urlParts.pathname)) {
+      if (!response.headersSent) {
-      const presetName = decodeURIComponent(
+        response.writeHead(500, { Connection: "close" });
-        PRESETS_REGEX.exec(urlParts.pathname)[1]
+        response.end(`${err.name}: ${err.message}`);
-      );
+      }
+      throw err;
+    });
+}
+
+async function handler(
+  request: IncomingMessage,
+  response: ServerResponse,
+  url: URL,
+  body: Buffer
+): Promise<void> {
+  if (PRESETS_REGEX.test(url.pathname)) {
+    const presetName = decodeURIComponent(PRESETS_REGEX.exec(url.pathname)[1]);
     if (request.method === "PUT") {
       let preset;
       try {
@@ -125,50 +124,23 @@ export function listener(
         return;
       }
       preset._id = presetName;
-        collections.presets.replaceOne(
+      await collections.presets.replaceOne({ _id: presetName }, preset, {
-          { _id: presetName },
+        upsert: true,
-          preset,
+      });
-          { upsert: true },
+      await cache.del("presets_hash");
-          (err) => {
-            if (err) return void throwError(err, response);
-
-            cache
-              .del("presets_hash")
-              .then(() => {
       response.writeHead(200);
       response.end();
-              })
-              .catch((err) => {
-                setTimeout(() => {
-                  throwError(err, response);
-                });
-              });
-          }
-        );
     } else if (request.method === "DELETE") {
-        collections.presets.deleteOne({ _id: presetName }, (err) => {
+      await collections.presets.deleteOne({ _id: presetName });
-          if (err) return void throwError(err, response);
+      await cache.del("presets_hash");
-
-          cache
-            .del("presets_hash")
-            .then(() => {
       response.writeHead(200);
       response.end();
-            })
-            .catch((err) => {
-              setTimeout(() => {
-                throwError(err, response);
-              });
-            });
-        });
     } else {
       response.writeHead(405, { Allow: "PUT, DELETE" });
       response.end("405 Method Not Allowed");
     }
-    } else if (OBJECTS_REGEX.test(urlParts.pathname)) {
+  } else if (OBJECTS_REGEX.test(url.pathname)) {
-      const objectName = decodeURIComponent(
+    const objectName = decodeURIComponent(OBJECTS_REGEX.exec(url.pathname)[1]);
-        OBJECTS_REGEX.exec(urlParts.pathname)[1]
-      );
     if (request.method === "PUT") {
       let object;
       try {
@@ -179,49 +151,24 @@ export function listener(
         return;
       }
       object._id = objectName;
-        collections.objects.replaceOne(
+      await collections.objects.replaceOne({ _id: objectName }, object, {
-          { _id: objectName },
+        upsert: true,
-          object,
+      });
-          { upsert: true },
+      await cache.del("presets_hash");
-          (err) => {
-            if (err) return void throwError(err, response);
-
-            cache
-              .del("presets_hash")
-              .then(() => {
       response.writeHead(200);
       response.end();
-              })
-              .catch((err) => {
-                setTimeout(() => {
-                  throwError(err, response);
-                });
-              });
-          }
-        );
     } else if (request.method === "DELETE") {
-        collections.objects.deleteOne({ _id: objectName }, (err) => {
+      await collections.objects.deleteOne({ _id: objectName });
-          if (err) return void throwError(err, response);
+      await cache.del("presets_hash");
-
-          cache
-            .del("presets_hash")
-            .then(() => {
       response.writeHead(200);
       response.end();
-            })
-            .catch((err) => {
-              setTimeout(() => {
-                throwError(err, response);
-              });
-            });
-        });
     } else {
       response.writeHead(405, { Allow: "PUT, DELETE" });
       response.end("405 Method Not Allowed");
     }
-    } else if (PROVISIONS_REGEX.test(urlParts.pathname)) {
+  } else if (PROVISIONS_REGEX.test(url.pathname)) {
     const provisionName = decodeURIComponent(
-        PROVISIONS_REGEX.exec(urlParts.pathname)[1]
+      PROVISIONS_REGEX.exec(url.pathname)[1]
     );
     if (request.method === "PUT") {
       const object = {
@@ -237,49 +184,24 @@ export function listener(
         return;
       }
 
-        collections.provisions.replaceOne(
+      await collections.provisions.replaceOne({ _id: provisionName }, object, {
-          { _id: provisionName },
+        upsert: true,
-          object,
+      });
-          { upsert: true },
+      await cache.del("presets_hash");
-          (err) => {
-            if (err) return void throwError(err, response);
-
-            cache
-              .del("presets_hash")
-              .then(() => {
       response.writeHead(200);
       response.end();
-              })
-              .catch((err) => {
-                setTimeout(() => {
-                  throwError(err, response);
-                });
-              });
-          }
-        );
     } else if (request.method === "DELETE") {
-        collections.provisions.deleteOne({ _id: provisionName }, (err) => {
+      await collections.provisions.deleteOne({ _id: provisionName });
-          if (err) return void throwError(err, response);
+      await cache.del("presets_hash");
-
-          cache
-            .del("presets_hash")
-            .then(() => {
       response.writeHead(200);
       response.end();
-            })
-            .catch((err) => {
-              setTimeout(() => {
-                throwError(err, response);
-              });
-            });
-        });
     } else {
       response.writeHead(405, { Allow: "PUT, DELETE" });
       response.end("405 Method Not Allowed");
     }
-    } else if (VIRTUAL_PARAMETERS_REGEX.test(urlParts.pathname)) {
+  } else if (VIRTUAL_PARAMETERS_REGEX.test(url.pathname)) {
     const virtualParameterName = decodeURIComponent(
-        VIRTUAL_PARAMETERS_REGEX.exec(urlParts.pathname)[1]
+      VIRTUAL_PARAMETERS_REGEX.exec(url.pathname)[1]
     );
     if (request.method === "PUT") {
       const object = {
@@ -295,127 +217,73 @@ export function listener(
         return;
       }
 
-        collections.virtualParameters.replaceOne(
+      await collections.virtualParameters.replaceOne(
         { _id: virtualParameterName },
         object,
-          { upsert: true },
+        { upsert: true }
-          (err) => {
+      );
-            if (err) return void throwError(err, response);
+      await cache.del("presets_hash");
-
-            cache
-              .del("presets_hash")
-              .then(() => {
       response.writeHead(200);
       response.end();
-              })
-              .catch((err) => {
-                setTimeout(() => {
-                  throwError(err, response);
-                });
-              });
-          }
-        );
     } else if (request.method === "DELETE") {
-        collections.virtualParameters.deleteOne(
+      await collections.virtualParameters.deleteOne({
-          { _id: virtualParameterName },
+        _id: virtualParameterName,
-          (err) => {
+      });
-            if (err) return void throwError(err, response);
+      await cache.del("presets_hash");
-
-            cache
-              .del("presets_hash")
-              .then(() => {
       response.writeHead(200);
       response.end();
-              })
-              .catch((err) => {
-                setTimeout(() => {
-                  throwError(err, response);
-                });
-              });
-          }
-        );
     } else {
       response.writeHead(405, { Allow: "PUT, DELETE" });
       response.end("405 Method Not Allowed");
     }
-    } else if (TAGS_REGEX.test(urlParts.pathname)) {
+  } else if (TAGS_REGEX.test(url.pathname)) {
-      const r = TAGS_REGEX.exec(urlParts.pathname);
+    const r = TAGS_REGEX.exec(url.pathname);
     const deviceId = decodeURIComponent(r[1]);
     const tag = decodeURIComponent(r[2]);
     if (request.method === "POST") {
-        collections.devices.updateOne(
+      await collections.devices.updateOne(
         { _id: deviceId },
-          { $addToSet: { _tags: tag } },
+        { $addToSet: { _tags: tag } }
-          (err) => {
+      );
-            if (err) return void throwError(err, response);
       response.writeHead(200);
       response.end();
-          }
-        );
     } else if (request.method === "DELETE") {
-        collections.devices.updateOne(
+      await collections.devices.updateOne(
         { _id: deviceId },
-          { $pull: { _tags: tag } },
+        { $pull: { _tags: tag } }
-          (err) => {
+      );
-            if (err) return void throwError(err, response);
-
       response.writeHead(200);
       response.end();
-          }
-        );
     } else {
       response.writeHead(405, { Allow: "POST, DELETE" });
       response.end("405 Method Not Allowed");
     }
-    } else if (FAULTS_REGEX.test(urlParts.pathname)) {
+  } else if (FAULTS_REGEX.test(url.pathname)) {
     if (request.method === "DELETE") {
-        const faultId = decodeURIComponent(
+      const faultId = decodeURIComponent(FAULTS_REGEX.exec(url.pathname)[1]);
-          FAULTS_REGEX.exec(urlParts.pathname)[1]
-        );
       const deviceId = faultId.split(":", 1)[0];
       const channel = faultId.slice(deviceId.length + 1);
-        collections.faults.deleteOne({ _id: faultId }, (err) => {
+      await collections.faults.deleteOne({ _id: faultId });
-          if (err) return void throwError(err, response);
-
       if (channel.startsWith("task_")) {
         const objId = new ObjectId(channel.slice(5));
-            return void collections.tasks.deleteOne({ _id: objId }, (err) => {
+        await collections.tasks.deleteOne({ _id: objId });
-              if (err) return void throwError(err, response);
+        await cache.del(`${deviceId}_tasks_faults_operations`);
-
-              cache
-                .del(`${deviceId}_tasks_faults_operations`)
-                .then(() => {
         response.writeHead(200);
         response.end();
-                })
+        return;
-                .catch((err) => {
-                  setTimeout(() => {
-                    throwError(err, response);
-                  });
-                });
-            });
       }
 
-          cache
+      await cache.del(`${deviceId}_tasks_faults_operations`);
-            .del(`${deviceId}_tasks_faults_operations`)
-            .then(() => {
       response.writeHead(200);
       response.end();
-            })
-            .catch((err) => {
-              setTimeout(() => {
-                throwError(err, response);
-              });
-            });
-        });
     } else {
       response.writeHead(405, { Allow: "DELETE" });
       response.end("405 Method Not Allowed");
     }
-    } else if (DEVICE_TASKS_REGEX.test(urlParts.pathname)) {
+  } else if (DEVICE_TASKS_REGEX.test(url.pathname)) {
     if (request.method === "POST") {
       const deviceId = decodeURIComponent(
-          DEVICE_TASKS_REGEX.exec(urlParts.pathname)[1]
+        DEVICE_TASKS_REGEX.exec(url.pathname)[1]
       );
       if (body.length) {
         let task;
@@ -427,9 +295,9 @@ export function listener(
           return;
         }
         task.device = deviceId;
-          collections.devices
+        const dev = await collections.devices.findOne({
-            .findOne({ _id: deviceId })
+          _id: deviceId,
-            .then(async (dev) => {
+        });
         if (!dev) {
           response.writeHead(404);
           response.end("No such device");
@@ -451,7 +319,7 @@ export function listener(
           30000
         );
         await cache.del(`${deviceId}_tasks_faults_operations`);
-              if (urlParts.query.connection_request == null) {
+        if (url.searchParams.has("connection_request")) {
           if (socketTimeout) request.socket.setTimeout(socketTimeout);
           response.writeHead(202, {
             "Content-Type": "application/json",
@@ -469,10 +337,7 @@ export function listener(
           return;
         }
 
-              const status = await apiFunctions.connectionRequest(
+        const status = await apiFunctions.connectionRequest(deviceId, device);
-                deviceId,
-                device
-              );
 
         if (status) {
           if (socketTimeout) request.socket.setTimeout(socketTimeout);
@@ -484,8 +349,8 @@ export function listener(
         }
 
         const onlineThreshold =
-                (urlParts.query.timeout &&
+          (url.searchParams.has("timeout") &&
-                  parseInt(urlParts.query.timeout as string)) ||
+            parseInt(url.searchParams.get("timeout"))) ||
           (config.get("DEVICE_ONLINE_THRESHOLD", deviceId) as number);
 
         const sessionStarted = await apiFunctions.awaitSessionStart(
@@ -547,23 +412,17 @@ export function listener(
           });
           response.end(JSON.stringify(task));
         }
-            })
+      } else if (url.searchParams.has("connection_request")) {
-            .catch((err) => {
-              setTimeout(() => {
-                throwError(err, response);
-              });
-            });
-        } else if (urlParts.query.connection_request != null) {
         // No task, send connection request only
-          collections.devices
+        const dev = await collections.devices.findOne({
-            .findOne({ _id: deviceId })
+          _id: deviceId,
-            .then(async (dev) => {
+        });
         if (!dev) {
           response.writeHead(404);
           response.end("No such device");
           return;
         }
-              return apiFunctions.connectionRequest(deviceId).then((status) => {
+        const status = await apiFunctions.connectionRequest(deviceId);
         if (status) {
           response.writeHead(504, status);
           response.end(status);
@@ -571,13 +430,6 @@ export function listener(
         }
         response.writeHead(200);
         response.end();
-              });
-            })
-            .catch((err) => {
-              setTimeout(() => {
-                throwError(err, response);
-              });
-            });
       } else {
         response.writeHead(400);
         response.end();
@@ -586,17 +438,16 @@ export function listener(
       response.writeHead(405, { Allow: "POST" });
       response.end("405 Method Not Allowed");
     }
-    } else if (TASKS_REGEX.test(urlParts.pathname)) {
+  } else if (TASKS_REGEX.test(url.pathname)) {
-      const r = TASKS_REGEX.exec(urlParts.pathname);
+    const r = TASKS_REGEX.exec(url.pathname);
     const taskId = decodeURIComponent(r[1]);
     const action = r[2];
     if (!action || action === "/") {
       if (request.method === "DELETE") {
-          collections.tasks.findOne(
+        const task = await collections.tasks.findOne(
           { _id: new ObjectId(taskId) },
-            { projection: { device: 1 } },
+          { projection: { device: 1 } }
-            (err, task) => {
+        );
-              if (err) return void throwError(err, response);
 
         if (!task) {
           response.writeHead(404);
@@ -605,66 +456,34 @@ export function listener(
         }
 
         const deviceId = task.device;
-              collections.tasks.deleteOne(
+        await collections.tasks.deleteOne({ _id: new ObjectId(taskId) });
-                { _id: new ObjectId(taskId) },
-                (err) => {
-                  if (err) return void throwError(err, response);
 
-                  collections.faults.deleteOne(
+        await collections.faults.deleteOne({
-                    { _id: `${deviceId}:task_${taskId}` },
+          _id: `${deviceId}:task_${taskId}`,
-                    (err) => {
+        });
-                      if (err) return void throwError(err, response);
 
-                      cache
+        await cache.del(`${deviceId}_tasks_faults_operations`);
-                        .del(`${deviceId}_tasks_faults_operations`)
-                        .then(() => {
         response.writeHead(200);
         response.end();
-                        })
-                        .catch((err) => {
-                          setTimeout(() => {
-                            throwError(err, response);
-                          });
-                        });
-                    }
-                  );
-                }
-              );
-            }
-          );
       } else {
         response.writeHead(405, { Allow: "PUT DELETE" });
         response.end("405 Method Not Allowed");
       }
     } else if (action === "/retry") {
       if (request.method === "POST") {
-          collections.tasks.findOne(
+        const task = await collections.tasks.findOne(
           { _id: new ObjectId(taskId) },
-            { projection: { device: 1 } },
+          { projection: { device: 1 } }
-            (err, task) => {
+        );
-              if (err) return void throwError(err, response);
 
         const deviceId = task.device;
-              collections.faults.deleteOne(
+        await collections.faults.deleteOne({
-                { _id: `${deviceId}:task_${taskId}` },
+          _id: `${deviceId}:task_${taskId}`,
-                (err) => {
+        });
-                  if (err) return void throwError(err, response);
 
-                  cache
+        await cache.del(`${deviceId}_tasks_faults_operations`);
-                    .del(`${deviceId}_tasks_faults_operations`)
-                    .then(() => {
         response.writeHead(200);
         response.end();
-                    })
-                    .catch((err) => {
-                      setTimeout(() => {
-                        throwError(err, response);
-                      });
-                    });
-                }
-              );
-            }
-          );
       } else {
         response.writeHead(405, { Allow: "POST" });
         response.end("405 Method Not Allowed");
@@ -673,10 +492,8 @@ export function listener(
       response.writeHead(404);
       response.end();
     }
-    } else if (FILES_REGEX.test(urlParts.pathname)) {
+  } else if (FILES_REGEX.test(url.pathname)) {
-      const filename = decodeURIComponent(
+    const filename = decodeURIComponent(FILES_REGEX.exec(url.pathname)[1]);
-        FILES_REGEX.exec(urlParts.pathname)[1]
-      );
     if (request.method === "PUT") {
       const metadata = {
         fileType: request.headers.filetype,
@@ -684,7 +501,13 @@ export function listener(
         productClass: request.headers.productclass,
         version: request.headers.version,
       };
-        filesBucket.delete(filename as unknown as ObjectId, () => {
+      try {
+        await filesBucket.delete(filename as unknown as ObjectId);
+      } catch (err) {
+        // Ignore error if file doesn't exist
+      }
+
+      return new Promise((resolve, reject) => {
         const uploadStream = filesBucket.openUploadStreamWithId(
           filename as unknown as ObjectId,
           filename,
@@ -693,35 +516,34 @@ export function listener(
           }
         );
 
-          uploadStream.on("error", (err) => {
+        uploadStream.on("error", reject);
-            throwError(err, response);
-          });
 
         uploadStream.end(body, () => {
           response.writeHead(201);
           response.end();
+          resolve();
         });
       });
     } else if (request.method === "DELETE") {
-        filesBucket.delete(filename as unknown as ObjectId, (err) => {
+      try {
-          if (err) {
+        await filesBucket.delete(filename as unknown as ObjectId);
+      } catch (err) {
         if (err.message.startsWith("FileNotFound")) {
           response.writeHead(404);
           response.end("404 Not Found");
           return;
         }
-            return void throwError(err, response);
+        throw err;
       }
-
       response.writeHead(200);
       response.end();
-        });
     } else {
       response.writeHead(405, { Allow: "PUT, DELETE" });
       response.end("405 Method Not Allowed");
     }
-    } else if (PING_REGEX.test(urlParts.pathname)) {
+  } else if (PING_REGEX.test(url.pathname)) {
-      const host = decodeURIComponent(PING_REGEX.exec(urlParts.pathname)[1]);
+    const host = decodeURIComponent(PING_REGEX.exec(url.pathname)[1]);
+    return new Promise((resolve) => {
       ping(host, (err, res, stdout) => {
         if (err) {
           if (!res) {
@@ -739,8 +561,10 @@ export function listener(
           "Cache-Control": "no-cache",
         });
         response.end(stdout);
+        resolve();
       });
-    } else if (DELETE_DEVICE_REGEX.test(urlParts.pathname)) {
+    });
+  } else if (DELETE_DEVICE_REGEX.test(url.pathname)) {
     if (request.method !== "DELETE") {
       response.writeHead(405, { Allow: "DELETE" });
       response.end("405 Method Not Allowed");
@@ -748,21 +572,13 @@ export function listener(
     }
 
     const deviceId = decodeURIComponent(
-        DELETE_DEVICE_REGEX.exec(urlParts.pathname)[1]
+      DELETE_DEVICE_REGEX.exec(url.pathname)[1]
     );
-      apiFunctions
+    await apiFunctions.deleteDevice(deviceId);
-        .deleteDevice(deviceId)
-        .then(() => {
     response.writeHead(200);
     response.end();
-        })
+  } else if (QUERY_REGEX.test(url.pathname)) {
-        .catch((err) => {
+    let collectionName = QUERY_REGEX.exec(url.pathname)[1];
-          setTimeout(() => {
-            throwError(err, response);
-          });
-        });
-    } else if (QUERY_REGEX.test(urlParts.pathname)) {
-      let collectionName = QUERY_REGEX.exec(urlParts.pathname)[1];
 
     // Convert to camel case
     let i = collectionName.indexOf("_");
@@ -780,7 +596,7 @@ export function listener(
       return;
     }
 
-      const collection = collections[collectionName];
+    const collection = collections[collectionName] as Collection<unknown>;
     if (!collection) {
       response.writeHead(404);
       response.end("404 Not Found");
@@ -788,9 +604,9 @@ export function listener(
     }
 
     let q = {};
-      if (urlParts.query.query) {
+    if (url.searchParams.has("query")) {
       try {
-          q = JSON.parse(urlParts.query.query as string);
+        q = JSON.parse(url.searchParams.get("query") as string);
       } catch (err) {
         response.writeHead(400);
         response.end(`${err.name}: ${err.message}`);
@@ -817,19 +633,19 @@ export function listener(
     }
 
     let projection = null;
-      if (urlParts.query.projection) {
+    if (url.searchParams.has("projection")) {
       projection = {};
-        for (const p of (urlParts.query.projection as string).split(","))
+      for (const p of (url.searchParams.get("projection") as string).split(","))
         projection[p.trim()] = 1;
       projection = optimizeProjection(projection);
     }
 
     const cur = collection.find(q, { projection: projection });
 
-      if (urlParts.query.sort) {
+    if (url.searchParams.has("sort")) {
       let s;
       try {
-          s = JSON.parse(urlParts.query.sort as string);
+        s = JSON.parse(url.searchParams.get("sort") as string);
       } catch (err) {
         response.writeHead(400);
         response.end(`${err.name}: ${err.message}`);
@@ -845,8 +661,7 @@ export function listener(
       cur.sort(sort);
     }
 
-      collection.countDocuments(q, (err, total) => {
+    const total = await collection.countDocuments(q);
-        if (err) return void throwError(err);
 
     response.writeHead(200, {
       "Content-Type": "application/json",
@@ -858,28 +673,21 @@ export function listener(
       return;
     }
 
-        if (urlParts.query.skip)
+    if (url.searchParams.has("skip"))
-          cur.skip(parseInt(urlParts.query.skip as string));
+      cur.skip(parseInt(url.searchParams.get("skip") as string));
 
-        if (urlParts.query.limit)
+    if (url.searchParams.has("limit"))
-          cur.limit(parseInt(urlParts.query.limit as string));
+      cur.limit(parseInt(url.searchParams.get("limit") as string));
 
     response.write("[\n");
     i = 0;
-        cur.forEach(
+    for await (const item of cur) {
-          (item) => {
       if (i++) response.write(",\n");
       response.write(JSON.stringify(item));
-          },
-          (err) => {
-            if (err) return void throwError(err);
-            response.end("\n]");
     }
-        );
+    response.end("\n]");
-      });
   } else {
     response.writeHead(404);
     response.end("404 Not Found");
   }
-  });
 }

lib/ui/api-functions.ts

@@ -145,19 +145,19 @@ export async function postTasks(
     };
   }
 
-  const promises = [];
+  const promises: Promise<number>[] = [];
   for (const s of statuses) {
-    promises.push(db.query("tasks", ["=", ["PARAM", "_id"], s._id]));
+    promises.push(db.count("tasks", ["=", ["PARAM", "_id"], s._id]));
     promises.push(
-      db.query("faults", ["=", ["PARAM", "_id"], `${deviceId}:task_${s._id}`])
+      db.count("faults", ["=", ["PARAM", "_id"], `${deviceId}:task_${s._id}`])
     );
   }
 
   const res = await Promise.all(promises);
   for (const [i, r] of statuses.entries()) {
-    if (res[i * 2].length === 0) {
+    if (!res[i * 2]) {
       r.status = "done";
-    } else if (res[i * 2 + 1].length === 1) {
+    } else if (res[i * 2 + 1]) {
       r.status = "fault";
       r["fault"] = res[i * 2 + 1][0];
     }

lib/ui/api.ts

@@ -221,16 +221,11 @@ for (const [resource, flags] of Object.entries(resources)) {
 
     let c = 0;
     ctx.body.write("[\n");
-    db.query(resource, filter, options, (obj) => {
+    for await (const obj of db.query(resource, filter, options))
       ctx.body.write((c++ ? "," : "") + JSON.stringify(obj) + "\n");
-    })
+
-      .then(() => {
     ctx.body.end("]");
     logger.accessInfo(log);
-      })
-      .catch((err) => {
-        ctx.body.emit("error", err);
-      });
   });
 
   // CSV download
@@ -289,7 +284,7 @@ for (const [resource, flags] of Object.entries(resources)) {
     ctx.body.write(
       Object.keys(columns).map((k) => `"${k.replace(/"/, '""')}"`) + "\n"
     );
-    db.query(resource, filter, options, (obj) => {
+    for await (const obj of db.query(resource, filter, options)) {
       const arr = Object.values(columns).map((exp) => {
         const v = evaluate(exp, obj, null, (e) => {
           if (Array.isArray(e)) {
@@ -331,14 +326,9 @@ for (const [resource, flags] of Object.entries(resources)) {
         return v;
       });
       ctx.body.write(arr.join(",") + "\n");
-    })
+    }
-      .then(() => {
     ctx.body.end();
     logger.accessInfo(log);
-      })
-      .catch((err) => {
-        ctx.body.emit("error", err);
-      });
   });
 
   router.head(`/${resource}/:id`, async (ctx) => {
@@ -359,9 +349,9 @@ for (const [resource, flags] of Object.entries(resources)) {
       return void (ctx.status = 403);
     }
 
-    const res = await db.query(resource, filter);
+    const count = await db.count(resource, filter);
 
-    if (!res.length) return void (ctx.status = 404);
+    if (!count) return void (ctx.status = 404);
 
     logger.accessInfo(log);
     ctx.body = "";
@@ -385,12 +375,12 @@ for (const [resource, flags] of Object.entries(resources)) {
       return void (ctx.status = 403);
     }
 
-    const res = await db.query(resource, filter);
+    const { value: res } = await db.query(resource, filter).next();
 
-    if (!res.length) return void (ctx.status = 404);
+    if (!res) return void (ctx.status = 404);
 
     logger.accessInfo(log);
-    ctx.body = res[0];
+    ctx.body = res;
   });
 
   if (flags & RESOURCE_DELETE) {
@@ -411,10 +401,10 @@ for (const [resource, flags] of Object.entries(resources)) {
         logUnauthorizedWarning(log);
         return void (ctx.status = 403);
       }
-      const res = await db.query(resource, filter);
+      const { value: res } = await db.query(resource, filter).next();
-      if (!res.length) return void (ctx.status = 404);
+      if (!res) return void (ctx.status = 404);
 
-      const validate = authorizer.getValidator(resource, res[0]);
+      const validate = authorizer.getValidator(resource, res);
       if (!validate("delete")) {
         logUnauthorizedWarning(log);
         return void (ctx.status = 403);
@@ -560,9 +550,8 @@ router.post("/devices/:id/tasks", async (ctx) => {
     logUnauthorizedWarning(log);
     return void (ctx.status = 403);
   }
-  const devices = await db.query("devices", filter);
+  const { value: device } = await db.query("devices", filter).next();
-  if (!devices.length) return void (ctx.status = 404);
+  if (!device) return void (ctx.status = 404);
-  const device = devices[0];
 
   const validate = authorizer.getValidator("devices", device);
   for (const t of ctx.request.body) {
@@ -636,10 +625,10 @@ router.post("/devices/:id/tags", async (ctx) => {
     logUnauthorizedWarning(log);
     return void (ctx.status = 403);
   }
-  const res = await db.query("devices", filter);
+  const { value: res } = await db.query("devices", filter).next();
-  if (!res.length) return void (ctx.status = 404);
+  if (!res) return void (ctx.status = 404);
 
-  const validate = authorizer.getValidator("devices", res[0]);
+  const validate = authorizer.getValidator("devices", res);
   if (!validate("tags", ctx.request.body)) {
     logUnauthorizedWarning(log);
     return void (ctx.status = 403);
@@ -707,12 +696,12 @@ router.put("/users/:id/password", async (ctx) => {
     ["PARAM", RESOURCE_IDS.users],
     username,
   ]);
-  const res = await db.query("users", filter);
+  const { value: res } = await db.query("users", filter).next();
-  if (!res.length) return void (ctx.status = 404);
+  if (!res) return void (ctx.status = 404);
 
   const newPassword = ctx.request.body.newPassword;
   if (ctx.state.user) {
-    const validate = authorizer.getValidator("users", res[0]);
+    const validate = authorizer.getValidator("users", res);
     if (!validate("password", { password: newPassword })) {
       logUnauthorizedWarning(log);
       return void (ctx.status = 403);

lib/ui/db.ts

@@ -17,56 +17,36 @@
  * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-import { Db, GridFSBucket, ObjectId } from "mongodb";
+import { Collection, GridFSBucket, ObjectId } from "mongodb";
 import { Script } from "vm";
-import { onConnect, optimizeProjection } from "../db";
+import { collections, onConnect, optimizeProjection } from "../db";
 import * as mongodbFunctions from "../mongodb-functions";
 import { evaluate } from "../common/expression/util";
 import { QueryOptions, Expression } from "../types";
 import { Readable } from "stream";
 
-const RESOURCE_COLLECTION = {
+let filesBucket: GridFSBucket;
-  files: "fs.files",
-};
 
-let db: Db;
+onConnect(async (db) => {
-
+  filesBucket = new GridFSBucket(db);
-onConnect(async (_db) => {
-  db = _db;
 });
 
-export function query(
+export async function* query(
   resource: string,
   filter: Expression,
   options?: QueryOptions
-): Promise<any[]>;
+): AsyncGenerator<any, void, undefined> {
-export function query(
-  resource: string,
-  filter: Expression,
-  options: QueryOptions,
-  callback: (doc: any) => void
-): Promise<void>;
-export function query(
-  resource: string,
-  filter: Expression,
-  options?: QueryOptions,
-  callback?: (doc: any) => void
-): Promise<void | any[]> {
   options = options || {};
   filter = evaluate(filter, null, Date.now());
   const q = mongodbFunctions.toMongoQuery(filter, resource);
-  if (!q) return Promise.resolve([]);
+  if (!q) return;
 
-  return new Promise((resolve, reject) => {
+  const collection = collections[resource] as Collection<any>;
-    const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
   const cursor = collection.find(q);
   if (options.projection) {
     let projection = options.projection;
-      if (resource === "devices") {
+    if (resource === "devices")
-        projection = mongodbFunctions.processDeviceProjection(
+      projection = mongodbFunctions.processDeviceProjection(options.projection);
-          options.projection
-        );
-      }
 
     if (resource === "presets") projection.configurations = 1;
     projection = optimizeProjection(projection);
@@ -91,47 +71,19 @@ export function query(
     cursor.sort(s);
   }
 
-    if (!callback) {
+  for await (let doc of cursor) {
-      cursor.toArray((err, docs) => {
+    if (resource === "devices") doc = mongodbFunctions.flattenDevice(doc);
-        if (err) return reject(err);
+    else if (resource === "faults") doc = mongodbFunctions.flattenFault(doc);
-        if (resource === "devices")
+    else if (resource === "tasks") doc = mongodbFunctions.flattenTask(doc);
-          return resolve(docs.map((d) => mongodbFunctions.flattenDevice(d)));
+    else if (resource === "presets") doc = mongodbFunctions.flattenPreset(doc);
-        else if (resource === "faults")
+    else if (resource === "files") doc = mongodbFunctions.flattenFile(doc);
-          return resolve(docs.map((d) => mongodbFunctions.flattenFault(d)));
+
-        else if (resource === "tasks")
+    yield doc;
-          return resolve(docs.map((d) => mongodbFunctions.flattenTask(d)));
-        else if (resource === "presets")
-          return resolve(docs.map((d) => mongodbFunctions.flattenPreset(d)));
-        else if (resource === "files")
-          return resolve(docs.map((d) => mongodbFunctions.flattenFile(d)));
-        return resolve(docs);
-      });
-    } else {
-      cursor.forEach(
-        (doc) => {
-          if (resource === "devices")
-            callback(mongodbFunctions.flattenDevice(doc));
-          else if (resource === "faults")
-            callback(mongodbFunctions.flattenFault(doc));
-          else if (resource === "tasks")
-            callback(mongodbFunctions.flattenTask(doc));
-          else if (resource === "presets")
-            callback(mongodbFunctions.flattenPreset(doc));
-          else if (resource === "files")
-            callback(mongodbFunctions.flattenFile(doc));
-          else callback(doc);
-        },
-        (err) => {
-          if (err) reject(err);
-          else resolve();
   }
-      );
-    }
-  });
 }
 
 export function count(resource: string, filter: Expression): Promise<number> {
-  const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
+  const collection = collections[resource] as Collection<unknown>;
   filter = evaluate(filter, null, Date.now());
   const q = mongodbFunctions.toMongoQuery(filter, resource);
   if (!q) return Promise.resolve(0);
@@ -150,37 +102,25 @@ export async function updateDeviceTags(
     if (onOff) add.push(tag);
     else pull.push(tag);
   }
-
-  const collection = db.collection("devices");
   const object = {};
 
   if (add?.length) object["$addToSet"] = { _tags: { $each: add } };
   if (pull?.length) object["$pullAll"] = { _tags: pull };
 
-  await collection.updateOne({ _id: deviceId }, object);
+  await collections.devices.updateOne({ _id: deviceId }, object);
 }
 
-function putResource(resource, id, object): Promise<void> {
+async function putResource(resource, id, object): Promise<void> {
-  return new Promise((resolve, reject) => {
+  const collection = collections[resource] as Collection<unknown>;
-    const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
+  await collection.replaceOne({ _id: id }, object, { upsert: true });
-    collection.replaceOne({ _id: id }, object, { upsert: true }, (err) => {
-      if (err) return void reject(err);
-      resolve();
-    });
-  });
 }
 
-function deleteResource(
+async function deleteResource(
   resource: string,
   id: string | ObjectId
 ): Promise<void> {
-  return new Promise((resolve, reject) => {
+  const collection = collections[resource] as Collection<unknown>;
-    const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
+  await collection.deleteOne({ _id: id });
-    collection.deleteOne({ _id: id }, (err) => {
-      if (err) return void reject(err);
-      resolve();
-    });
-  });
 }
 
 export function putPreset(
@@ -267,23 +207,16 @@ export function deletePermission(id: string): Promise<void> {
   return deleteResource("permissions", id);
 }
 
-export function putUser(
+export async function putUser(
   id: string,
   object: Record<string, unknown>
 ): Promise<void> {
-  return new Promise((resolve, reject) => {
-    const collection = db.collection("users");
   // update instead of replace to keep the password if not set by user
-    collection.updateOne(
+  await collections.users.updateOne(
     { _id: id },
     { $set: object },
-      { upsert: true },
+    { upsert: true }
-      (err) => {
-        if (err) return void reject(err);
-        resolve();
-      }
   );
-  });
 }
 
 export function deleteUser(id: string): Promise<void> {
@@ -291,8 +224,7 @@ export function deleteUser(id: string): Promise<void> {
 }
 
 export function downloadFile(filename: string): Readable {
-  const bucket = new GridFSBucket(db);
+  return filesBucket.openDownloadStreamByName(filename);
-  return bucket.openDownloadStreamByName(filename);
 }
 
 export function putFile(
@@ -301,8 +233,7 @@ export function putFile(
   contentStream: Readable
 ): Promise<void> {
   return new Promise((resolve, reject) => {
-    const bucket = new GridFSBucket(db);
+    const uploadStream = filesBucket.openUploadStreamWithId(
-    const uploadStream = bucket.openUploadStreamWithId(
       filename as unknown as ObjectId,
       filename,
       {
@@ -310,8 +241,15 @@ export function putFile(
       }
     );
 
+    let readableEnded = false;
+    contentStream.on("end", () => {
+      readableEnded = true;
+    });
     contentStream.on("close", () => {
-      if (!contentStream.readableEnded)
+      // In Node versions prior to 15, the stream will not emit an error if the
+      // connection is closed before the stream is finished.
+      // For Node 12.9+ we can just use stream.readableEnded
+      if (!readableEnded)
         uploadStream.destroy(new Error("Stream closed prematurely"));
     });
 
@@ -325,14 +263,8 @@ export function putFile(
   });
 }
 
-export function deleteFile(filename: string): Promise<void> {
+export async function deleteFile(filename: string): Promise<void> {
-  return new Promise((resolve, reject) => {
+  await filesBucket.delete(filename as any);
-    const bucket = new GridFSBucket(db);
-    bucket.delete(filename as any, (err) => {
-      if (err) return void reject(err);
-      resolve();
-    });
-  });
 }
 
 export function deleteFault(id: string): Promise<void> {

package.json

@@ -30,7 +30,7 @@
     "koa-jwt": "^4.0.3",
     "koa-router": "^12.0.0",
     "koa-static": "^5.0.0",
-    "mongodb": "^4.9.0",
+    "mongodb": "^4.16.0",
     "parsimmon": "^1.18.1",
     "seedrandom": "^3.0.5"
   },