Implement TR-069 Upload method

lib/fs.ts

@@ -20,23 +20,42 @@
 import * as url from "url";
 import * as querystring from "querystring";
 import { IncomingMessage, ServerResponse } from "http";
-import { Collection, GridFSBucket } from "mongodb";
-import { onConnect } from "./db";
+import { Collection, GridFSBucket, ObjectId } from "mongodb";
+import { onConnect, getOperations } from "./db";
 import * as logger from "./logger";
 import { getRequestOrigin } from "./forwarded";
 
 let filesCollection: Collection;
 let filesBucket: GridFSBucket;
+let uploadsBucket: GridFSBucket;
 
 onConnect(async (db) => {
   filesCollection = db.collection("fs.files");
   filesBucket = new GridFSBucket(db);
+  uploadsBucket = new GridFSBucket(db, { bucketName: "uploads" });
 });
 
-export function listener(
+async function canUpload(
+  deviceId: string,
+  fileName: string,
+  timeout = Date.now() + 5000
+): Promise<boolean> {
+  const operations = Object.values(await getOperations(deviceId));
+  for (const operation of operations) {
+    if (operation.name === "Upload" && operation.args.fileName === fileName)
+      return true;
+  }
+
+  if (Date.now() >= timeout) return false;
+  // Need to wait and retry in case upload was initiated before session was closed
+  await new Promise((resolve) => setTimeout(resolve, 1000));
+  return canUpload(deviceId, fileName, timeout);
+}
+
+export async function listener(
   request: IncomingMessage,
   response: ServerResponse
-): void {
+): Promise<void> {
   const urlParts = url.parse(request.url, true);
   if (request.method === "GET") {
     const filename = querystring.unescape(urlParts.pathname.substring(1));
@@ -47,27 +66,58 @@ export function listener(
       remoteAddress: getRequestOrigin(request).remoteAddress,
     };
 
-    filesCollection.findOne({ _id: filename }, (err, file) => {
-      if (err) throw err;
+    const file = await filesCollection.findOne({ _id: filename });
+    if (!file) {
+      response.writeHead(404);
+      response.end();
+      log.message += " not found";
+      logger.accessError(log);
+      return;
+    }
 
-      if (!file) {
-        response.writeHead(404);
-        response.end();
-        log.message += " not found";
-        logger.accessError(log);
-        return;
-      }
-
-      response.writeHead(200, {
-        "Content-Type": file.contentType || "application/octet-stream",
-        "Content-Length": file.length,
-      });
-
-      const downloadStream = filesBucket.openDownloadStreamByName(filename);
-      downloadStream.pipe(response);
-
-      logger.accessInfo(log);
+    response.writeHead(200, {
+      "Content-Type": file.contentType || "application/octet-stream",
+      "Content-Length": file.length,
     });
+
+    const downloadStream = filesBucket.openDownloadStreamByName(filename);
+    downloadStream.pipe(response);
+
+    logger.accessInfo(log);
+  } else if (request.method === "PUT") {
+    const [, deviceId, ...filePath] = urlParts.pathname
+      .split("/")
+      .map(querystring.unescape);
+
+    const fileName = `${deviceId}/${filePath.join("/")}`;
+
+    if (!(await canUpload(deviceId, fileName))) {
+      response.writeHead(403);
+      response.end("403 Forbidden");
+      return;
+    }
+
+    try {
+      await uploadsBucket.delete((fileName as unknown) as ObjectId);
+    } catch (err) {
+      // File not found, do nothing
+    }
+    const uploadStream = uploadsBucket.openUploadStreamWithId(
+      fileName,
+      fileName
+    );
+
+    uploadStream.on("finish", () => {
+      response.writeHead(200);
+      response.end();
+    });
+
+    uploadStream.on("error", (err) => {
+      response.writeHead(500);
+      response.end(err.message);
+    });
+
+    request.pipe(uploadStream);
   } else {
     response.writeHead(405, { Allow: "GET" });
     response.end("405 Method Not Allowed");