Refactor DB code to use promises

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 }),
-    db.devicesCollection.deleteOne({ _id: deviceId }),
-    db.faultsCollection.deleteMany({
+    collections.tasks.deleteMany({ device: deviceId }),
+    collections.devices.deleteOne({ _id: deviceId }),
+    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>,
-  devicesCollection: Collection,
-  presetsCollection: Collection,
-  objectsCollection: Collection,
-  provisionsCollection: Collection,
-  virtualParametersCollection: Collection,
-  faultsCollection: Collection<MongoTypes.Fault>,
-  filesCollection: Collection,
-  operationsCollection: Collection<MongoTypes.Operation>,
-  permissionsCollection: Collection,
-  usersCollection: Collection,
-  configCollection: Collection<MongoTypes.Config>;
+export const collections = {
+  devices: null as Collection<MongoTypes.Device>,
+  presets: null as Collection<MongoTypes.Preset>,
+  objects: null as Collection<MongoTypes.Object>,
+  provisions: null as Collection<MongoTypes.Provision>,
+  virtualParameters: null as Collection<MongoTypes.VirtualParameter>,
+  faults: null as Collection<MongoTypes.Fault>,
+  tasks: null as Collection<MongoTypes.Task>,
+  files: null as Collection<MongoTypes.File>,
+  operations: null as Collection<MongoTypes.Operation>,
+  permissions: null as Collection<MongoTypes.Permission>,
+  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");
-  await tasksCollection.createIndex({ device: 1, timestamp: 1 });
+  collections.tasks = db.collection("tasks");
+  await collections.tasks.createIndex({ device: 1, timestamp: 1 });
 
-  devicesCollection = db.collection("devices");
-  presetsCollection = db.collection("presets");
-  objectsCollection = db.collection("objects");
-  filesCollection = db.collection("fs.files");
-  provisionsCollection = db.collection("provisions");
-  virtualParametersCollection = db.collection("virtualParameters");
-  faultsCollection = db.collection("faults");
-  operationsCollection = db.collection("operations");
-  permissionsCollection = db.collection("permissions");
-  usersCollection = db.collection("users");
-  configCollection = db.collection("config");
+  collections.devices = db.collection("devices");
+  collections.presets = db.collection("presets");
+  collections.objects = db.collection("objects");
+  collections.files = db.collection("fs.files");
+  collections.provisions = db.collection("provisions");
+  collections.virtualParameters = db.collection("virtualParameters");
+  collections.faults = db.collection("faults");
+  collections.operations = db.collection("operations");
+  collections.permissions = db.collection("permissions");
+  collections.users = db.collection("users");
+  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, {
-    upsert: isNew,
-  });
+  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 { onConnect } from "./db";
+import { GridFSBucket } from "mongodb";
+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/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);
-      });
+
+    ctx.body.end("]");
+    logger.accessInfo(log);
   });
 
   // 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);
-      });
+    }
+    ctx.body.end();
+    logger.accessInfo(log);
   });
 
   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);
-      if (!res.length) return void (ctx.status = 404);
+      const { value: res } = await db.query(resource, filter).next();
+      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);
-  if (!devices.length) return void (ctx.status = 404);
-  const device = devices[0];
+  const { value: device } = await db.query("devices", filter).next();
+  if (!device) return void (ctx.status = 404);
 
   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);
-  if (!res.length) return void (ctx.status = 404);
+  const { value: res } = await db.query("devices", filter).next();
+  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);
-  if (!res.length) return void (ctx.status = 404);
+  const { value: res } = await db.query("users", filter).next();
+  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,121 +17,73 @@
  * 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 = {
-  files: "fs.files",
-};
+let filesBucket: GridFSBucket;
 
-let db: Db;
-
-onConnect(async (_db) => {
-  db = _db;
+onConnect(async (db) => {
+  filesBucket = new GridFSBucket(db);
 });
 
-export function query(
+export async function* query(
   resource: string,
   filter: Expression,
   options?: QueryOptions
-): Promise<any[]>;
-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[]> {
+): AsyncGenerator<any, void, undefined> {
   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 = 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 = collections[resource] as Collection<any>;
+  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;
-      projection = optimizeProjection(projection);
-      cursor.project(projection);
-    }
+    if (resource === "presets") projection.configurations = 1;
+    projection = optimizeProjection(projection);
+    cursor.project(projection);
+  }
 
-    if (resource === "users") cursor.project({ password: 0, salt: 0 });
+  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.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")
-          return resolve(docs.map((d) => mongodbFunctions.flattenDevice(d)));
-        else if (resource === "faults")
-          return resolve(docs.map((d) => mongodbFunctions.flattenFault(d)));
-        else if (resource === "tasks")
-          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();
-        }
+  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);
+  }
+
+  for await (let doc of cursor) {
+    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);
+
+    yield doc;
+  }
 }
 
 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> {
-  return new Promise((resolve, reject) => {
-    const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
-    collection.replaceOne({ _id: id }, object, { upsert: true }, (err) => {
-      if (err) return void reject(err);
-      resolve();
-    });
-  });
+async function putResource(resource, id, object): Promise<void> {
+  const collection = collections[resource] as Collection<unknown>;
+  await collection.replaceOne({ _id: id }, object, { upsert: true });
 }
 
-function deleteResource(
+async function deleteResource(
   resource: string,
   id: string | ObjectId
 ): Promise<void> {
-  return new Promise((resolve, reject) => {
-    const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
-    collection.deleteOne({ _id: id }, (err) => {
-      if (err) return void reject(err);
-      resolve();
-    });
-  });
+  const collection = collections[resource] as Collection<unknown>;
+  await collection.deleteOne({ _id: id });
 }
 
 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(
-      { _id: id },
-      { $set: object },
-      { upsert: true },
-      (err) => {
-        if (err) return void reject(err);
-        resolve();
-      }
-    );
-  });
+  // update instead of replace to keep the password if not set by user
+  await collections.users.updateOne(
+    { _id: id },
+    { $set: object },
+    { upsert: true }
+  );
 }
 
 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 bucket.openDownloadStreamByName(filename);
+  return filesBucket.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 = bucket.openUploadStreamWithId(
+    const uploadStream = filesBucket.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> {
-  return new Promise((resolve, reject) => {
-    const bucket = new GridFSBucket(db);
-    bucket.delete(filename as any, (err) => {
-      if (err) return void reject(err);
-      resolve();
-    });
-  });
+export async function deleteFile(filename: string): Promise<void> {
+  await filesBucket.delete(filename as any);
 }
 
 export function deleteFault(id: string): Promise<void> {