Remove UI dependance on NBI

2019-02-07 00:18:19 -08:00
1 parent 520cc43972
commit d46d133878
@@ -1,340 +1,184 @@
 "use strict";

-import http from "http";
-import https from "https";
-import url from "url";
-import * as config from "../config";
 import * as db from "./db";
-import * as cache from "../cache";
-import * as mongodbFunctions from "./mongodb-functions";
+import { del } from "../cache";
+import { ObjectID } from "mongodb";
+import { exec } from "child_process";
+import {
+  insertTasks,
+  watchTask,
+  connectionRequest,
+  deleteDevice
+} from "../api-functions";

-export function deleteResource(resource, id) {
+function deleteFault(id) {
+  const deviceId = id.split(":", 1)[0];
+  const channel = id.slice(deviceId.length + 1);
  return new Promise((resolve, reject) => {
-    if (resource === "config") {
-      db.deleteConfig(id)
-        .then(() => {
-          cache.del("presets_hash", err => {
-            if (err) return void reject(err);
-            resolve();
-          });
-        })
-        .catch(reject);
-      return;
-    }
-
-    const resourceUrl =
-      resource === "virtualParameters" ? "virtual_parameters" : resource;
-    const options = url.parse(
-      `http://127.0.0.1:${config.get(
-        "NBI_PORT"
-      )}/${resourceUrl}/${encodeURIComponent(id)}`
-    );
-
-    options.method = "DELETE";
-
-    const _http = options.protocol === "https:" ? https : http;
-
-    _http
-      .request(options, res => {
-        res.resume();
-        if (res.statusCode === 200) resolve(true);
-        else if (res.statusCode === 404) resolve(false);
-        else reject(new Error(`Unexpected status code ${res.statusCode}`));
-      })
-      .end();
-  });
-}
-
-function postTask(task, connectionRequest, callback) {
-  const options = url.parse(
-    `http://127.0.0.1:${config.get("NBI_PORT")}/devices/${encodeURIComponent(
-      task.device
-    )}/tasks${connectionRequest ? "?connection_request" : ""}`
-  );
-  options.method = "POST";
-  const _http = options.protocol === "https:" ? https : http;
-
-  _http
-    .request(options, res => {
-      if (res.statusCode !== 200 && res.statusCode !== 202) {
-        callback(new Error(`Unexpected status code ${res.statusCode}`));
-        res.resume();
-        return;
-      }
-      const chunks = [];
-      let bytes = 0;
-      res.on("data", chunk => {
-        chunks.push(chunk);
-        bytes += chunk.length;
-      });
-
-      res.on("end", () => {
-        const buf = Buffer.allocUnsafe(bytes);
-        let o = 0;
-        for (const c of chunks) {
-          c.copy(buf, o, 0, c.length);
-          o += c.length;
-        }
-        try {
-          const t = JSON.parse(buf);
-          let connectionRequestStatus;
-          if (connectionRequest) {
-            connectionRequestStatus = "OK";
-            if (
-              res.statusCode === 202 &&
-              res.statusMessage !== "Task queued but not processed" &&
-              res.statusMessage !== "Task faulted"
-            )
-              connectionRequestStatus = res.statusMessage;
-          }
-          callback(null, t, connectionRequestStatus);
-        } catch (err) {
-          callback(err);
-        }
-      });
-    })
-    .end(JSON.stringify(task));
-}
-
-export function postTasks(deviceId, tasks) {
-  return new Promise((resolve, reject) => {
-    let connectionRequestStatus;
-    const promises = [];
-    for (const [idx, task] of tasks.entries()) {
-      task.device = deviceId;
-      promises.push(
-        new Promise((res, rej) => {
-          const conReq = idx === tasks.length - 1;
-          delete task._id;
-          task.expiry = 5;
-          postTask(task, conReq, (err, t, crs) => {
-            if (err) return void rej(err);
-            if (conReq) connectionRequestStatus = crs;
-            res({ _id: t._id, status: "pending" });
-          });
-        })
-      );
-    }
-
-    Promise.all(promises)
-      .then(statuses => {
-        if (connectionRequestStatus !== "OK") {
-          return void resolve({
-            connectionRequest: connectionRequestStatus,
-            tasks: statuses
-          });
-        }
-
-        const promises2 = [];
-        for (const s of statuses) {
-          promises2.push(db.query("tasks", ["=", ["PARAM", "_id"], s._id]));
-          promises2.push(
-            db.query("faults", [
-              "=",
-              ["PARAM", "_id"],
-              `${deviceId}:task_${s._id}`
-            ])
-          );
-        }
-
-        Promise.all(promises2)
-          .then(res => {
-            for (const [i, r] of statuses.entries()) {
-              if (res[i * 2].length === 0) {
-                r.status = "done";
-              } else if (res[i * 2 + 1].length === 1) {
-                r.status = "fault";
-                r.fault = res[i * 2 + 1][0];
-              }
-              deleteResource("tasks", r._id);
-            }
-
-            resolve({
-              connectionRequest: connectionRequestStatus,
-              tasks: statuses
-            });
-          })
-          .catch(reject);
+    Promise.all([
+      db.deleteFault(id),
+      channel.startsWith("task_")
+        ? db.deleteTask(new ObjectID(channel.slice(5)))
+        : null
+    ])
+      .then(() => {
+        del(`${deviceId}_tasks_faults_operations`, err => {
+          err ? reject(err) : resolve();
+        });
      })
      .catch(reject);
  });
 }

-export function updateTags(deviceId, tags) {
-  return Promise.all(
-    Object.entries(tags).map(
-      ([tag, onOff]) =>
-        new Promise((resolve, reject) => {
-          const options = url.parse(
-            `http://127.0.0.1:${config.get(
-              "NBI_PORT"
-            )}/devices/${encodeURIComponent(
-              deviceId
-            )}/tags/${encodeURIComponent(tag)}`
-          );
-          if (onOff) options.method = "POST";
-          else options.method = "DELETE";
-          const _http = options.protocol === "https:" ? https : http;
+export function deleteResource(resource, id) {
+  return new Promise((resolve, reject) => {
+    let promise;

-          _http
-            .request(options, res => {
-              res.resume();
-              if (res.statusCode === 200) resolve();
-              else
-                reject(new Error(`Unexpected status code ${res.statusCode}`));
-            })
-            .end();
-        })
-    )
-  );
+    switch (resource) {
+      case "devices":
+        promise = new Promise((res, rej) => {
+          deleteDevice(id, err => {
+            if (err) return void rej(err);
+            res();
+          });
+        });
+        break;
+
+      case "files":
+        promise = db.deleteFile(id);
+        break;
+
+      case "faults":
+        promise = deleteFault(id);
+        break;
+
+      case "provisions":
+        promise = db.deleteProvision(id);
+        break;
+
+      case "presets":
+        promise = db.deletePreset(id);
+        break;
+
+      case "virtualParameters":
+        promise = db.deleteVirtualParameter(id);
+        break;
+
+      case "config":
+        promise = db.deleteConfig(id);
+        break;
+    }
+
+    promise
+      .then(() => {
+        del("presets_hash", err => {
+          if (err) return void reject(err);
+          resolve();
+        });
+      })
+      .catch(reject);
+  });
+}
+
+export function postTasks(deviceId, tasks, timeout) {
+  return new Promise((resolve, reject) => {
+    for (const task of tasks) task.device = deviceId;
+
+    insertTasks(tasks, err => {
+      if (err) return void reject(err);
+      const statuses = tasks.map(t => {
+        return { _id: t._id, status: "pending" };
+      });
+      del(`${deviceId}_tasks_faults_operations`, err => {
+        if (err) return void reject(err);
+        connectionRequest(deviceId, err => {
+          if (err) {
+            return void resolve({
+              connectionRequest: err.message,
+              tasks: statuses
+            });
+          }
+
+          const sample = tasks[tasks.length - 1];
+
+          // Waiting for session to finish or timeout
+          watchTask(deviceId, sample._id, timeout, err => {
+            if (err) return void reject(err);
+
+            const promises2 = [];
+            for (const s of statuses) {
+              promises2.push(db.query("tasks", ["=", ["PARAM", "_id"], s._id]));
+              promises2.push(
+                db.query("faults", [
+                  "=",
+                  ["PARAM", "_id"],
+                  `${deviceId}:task_${s._id}`
+                ])
+              );
+            }
+
+            Promise.all(promises2)
+              .then(res => {
+                for (const [i, r] of statuses.entries()) {
+                  if (res[i * 2].length === 0) {
+                    r.status = "done";
+                  } else if (res[i * 2 + 1].length === 1) {
+                    r.status = "fault";
+                    r.fault = res[i * 2 + 1][0];
+                  }
+                  db.deleteTask(r._id);
+                }
+
+                resolve({ connectionRequest: "OK", tasks: statuses });
+              })
+              .catch(reject);
+          });
+        });
+      });
+    });
+  });
 }

 export function ping(host) {
  return new Promise((resolve, reject) => {
-    const options = url.parse(
-      `http://127.0.0.1:${config.get("NBI_PORT")}/ping/${encodeURIComponent(
-        host
-      )}`
-    );
+    exec(`ping -w 1 -i 0.2 -c 3 ${host}`, (err, stdout) => {
+      if (err) return void reject(err);

-    const _http = options.protocol === "https:" ? https : http;
-    _http
-      .request(options, res => {
-        if (res.statusCode === 404) {
-          res.resume();
-          return void resolve({});
-        }
+      const m = stdout.match(
+        /(\d) packets transmitted, (\d) received, ([\d.%]+) packet loss[^]*([\d.]+)\/([\d.]+)\/([\d.]+)\/([\d.]+)/
+      );
+      if (!m) return void reject(new Error("Could not parse ping response"));

-        if (res.statusCode !== 200) {
-          res.resume();
-          return void reject(
-            new Error(`Unexpected status code ${res.statusCode}`)
-          );
-        }
-
-        const chunks = [];
-        let bytes = 0;
-        res.on("data", chunk => {
-          chunks.push(chunk);
-          bytes += chunk.length;
-        });
-
-        res.on("end", () => {
-          const buf = Buffer.allocUnsafe(bytes);
-          let o = 0;
-          for (const c of chunks) {
-            c.copy(buf, o, 0, c.length);
-            o += c.length;
-          }
-          const m = buf
-            .toString()
-            .match(
-              /(\d) packets transmitted, (\d) received, ([\d.%]+) packet loss[^]*([\d.]+)\/([\d.]+)\/([\d.]+)\/([\d.]+)/
-            );
-          if (!m)
-            return void reject(new Error("Could not parse ping response"));
-
-          resolve({
-            packetsTransmitted: +m[1],
-            packetsReceived: +m[2],
-            packetLoss: m[3],
-            min: +m[4],
-            avg: +m[5],
-            max: +m[6],
-            mdev: +m[7]
-          });
-        });
-      })
-      .end();
+      resolve({
+        packetsTransmitted: +m[1],
+        packetsReceived: +m[2],
+        packetLoss: m[3],
+        min: +m[4],
+        avg: +m[5],
+        max: +m[6],
+        mdev: +m[7]
+      });
+    });
  });
 }

 export function putResource(resource, id, data) {
  return new Promise((resolve, reject) => {
-    if (resource === "config") {
-      db.putConfig(id, data)
-        .then(() => {
-          cache.del("presets_hash", err => {
-            if (err) return void reject(err);
-            resolve();
-          });
-        })
-        .catch(reject);
-      return;
-    }
+    let promise;

-    const resourceUrl =
-      resource === "virtualParameters" ? "virtual_parameters" : resource;
-    const options = url.parse(
-      `http://127.0.0.1:${config.get(
-        "NBI_PORT"
-      )}/${resourceUrl}/${encodeURIComponent(id)}`
-    );
+    if (resource === "presets") promise = db.putPreset(id, data);
+    else if (resource === "provisions") promise = db.putProvision(id, data);
+    else if (resource === "virtualParameters")
+      promise = db.putVirtualParameter(id, data);
+    else if (resource === "config") promise = db.putConfig(id, data);

-    options.method = "PUT";
-
-    const _http = options.protocol === "https:" ? https : http;
-
-    if (resource === "presets") data = mongodbFunctions.preProcessPreset(data);
-
-    if (resource === "provisions")
-      data = mongodbFunctions.preProcessProvision(data);
-
-    if (resource === "virtualParameters")
-      data = mongodbFunctions.preProcessVirtualParameters(data);
-
-    const body = typeof data !== "string" ? JSON.stringify(data) : data;
-    _http
-      .request(options, res => {
-        const chunks = [];
-        let bytes = 0;
-        res.on("data", chunk => {
-          chunks.push(chunk);
-          bytes += chunk.length;
-        });
-
-        res.on("end", () => {
-          if (res.statusCode === 200) {
-            return void resolve();
-          } else if (res.statusCode !== 400) {
-            return void reject(
-              new Error(`Unexpected status code ${res.statusCode}`)
-            );
-          }
-          const buf = Buffer.allocUnsafe(bytes);
-          let o = 0;
-          for (const c of chunks) {
-            c.copy(buf, o, 0, c.length);
-            o += c.length;
-          }
-          resolve(buf.toString());
+    promise
+      .then(() => {
+        del("presets_hash", err => {
+          if (err) return void reject(err);
+          resolve();
        });
      })
-      .end(body);
-  });
-}
-
-export function putFile(id, metadata, fileStream) {
-  return new Promise((resolve, reject) => {
-    const options = url.parse(
-      `http://127.0.0.1:${config.get("NBI_PORT")}/files/${encodeURIComponent(
-        id
-      )}`
-    );
-
-    options.method = "PUT";
-    options.headers = Object.assign({}, metadata);
-    const _http = options.protocol === "https:" ? https : http;
-    const req = _http.request(options, res => {
-      res.resume();
-      if (res.statusCode === 201) {
-        return void resolve();
-      } else {
-        return void reject(
-          new Error(`Unexpected status code ${res.statusCode}`)
-        );
-      }
-    });
-    fileStream.pipe(req);
+      .catch(reject);
  });
 }
@@ -7,6 +7,7 @@ import * as apiFunctions from "./api-functions";
 import { evaluate, and, extractParams } from "../common/expression";
 import { parse } from "../common/expression-parser";
 import * as logger from "../logger";
+import { getConfig } from "../local-cache";

 const router = new Router();
 export default router;
@@ -400,15 +401,7 @@ router.put("/files/:id", async (ctx, next) => {
    return void (ctx.status = 403);
  }

-  const err = await apiFunctions.putFile(id, metadata, ctx.req);
-
-  if (err) {
-    log.message += " failed";
-    logger.accessWarn(log);
-    ctx.body = err;
-    return void (ctx.status = 400);
-  }
-
+  await db.putFile(id, metadata, ctx.req);
  log.metadata = metadata;
  logger.accessInfo(log);

@@ -439,7 +432,17 @@ router.post("/devices/:id/tasks", async (ctx, next) => {
    }
  }

-  const res = await apiFunctions.postTasks(ctx.params.id, ctx.request.body);
+  // TODO Pass context and timestamp
+  const onlineThreshold = getConfig(
+    ctx.state.configSnapshot,
+    "cwmp.deviceOnlineThreshold"
+  );
+
+  const res = await apiFunctions.postTasks(
+    ctx.params.id,
+    ctx.request.body,
+    onlineThreshold
+  );

  log.tasks = res.tasks.map(t => t._id).join(",");

@@ -472,7 +475,7 @@ router.post("/devices/:id/tags", async (ctx, next) => {
    return void (ctx.status = 403);
  }

-  await apiFunctions.updateTags(ctx.params.id, ctx.request.body);
+  await db.updateDeviceTags(ctx.params.id, ctx.request.body);

  logger.accessInfo(log);

@@ -480,5 +483,11 @@ router.post("/devices/:id/tags", async (ctx, next) => {
 });

 router.get("/ping/:host", async ctx => {
-  ctx.body = await apiFunctions.ping(ctx.params.host);
+  try {
+    ctx.body = await apiFunctions.ping(ctx.params.host);
+  } catch (err) {
+    ctx.status = 404;
+    ctx.response.headers["Cache-Control"] = "no-cache";
+    ctx.body = `${err.name}: ${err.message}`;
+  }
 });
@@ -1,9 +1,10 @@
 "use strict";

-import { MongoClient } from "mongodb";
+import { MongoClient, GridFSBucket } from "mongodb";
 import * as config from "../config";
 import * as mongodbFunctions from "./mongodb-functions";
 import * as expression from "../common/expression";
+import { preProcessPreset } from "./mongodb-functions";

 const CACHE_TTL = 300000;

@@ -187,14 +188,23 @@ export function count(resource, filter) {
  });
 }

-export function deleteConfig(id) {
+export function updateDeviceTags(deviceId, tags) {
  return new Promise((resolve, reject) => {
+    const add = [];
+    const pull = [];
+
+    for (const [tag, onOff] of Object.entries(tags)) {
+      if (onOff) add.push(tag);
+      else pull.push(tag);
+    }
    getClient()
      .then(client => {
-        const collection = client
-          .db()
-          .collection(RESOURCE_COLLECTION["config"] || "config");
-        collection.deleteOne({ _id: id }, err => {
+        const collection = client.db().collection("devices");
+
+        const object = {};
+        if (add && add.length) object["$addToSet"] = { _tags: { $each: add } };
+        if (pull && pull.length) object["$pullAll"] = { _tags: pull };
+        collection.updateOne({ _id: deviceId }, object, err => {
          if (err) return void reject(err);
          resolve();
        });
@@ -203,13 +213,13 @@ export function deleteConfig(id) {
  });
 }

-export function putConfig(id, object) {
+function putResource(resource, id, object) {
  return new Promise((resolve, reject) => {
    getClient()
      .then(client => {
        const collection = client
          .db()
-          .collection(RESOURCE_COLLECTION["config"] || "config");
+          .collection(RESOURCE_COLLECTION[resource] || resource);
        collection.replaceOne({ _id: id }, object, { upsert: true }, err => {
          if (err) return void reject(err);
          resolve();
@@ -219,6 +229,94 @@ export function putConfig(id, object) {
  });
 }

+function deleteResource(resource, id) {
+  return new Promise((resolve, reject) => {
+    getClient()
+      .then(client => {
+        const collection = client
+          .db()
+          .collection(RESOURCE_COLLECTION[resource] || resource);
+        collection.deleteOne({ _id: id }, err => {
+          if (err) return void reject(err);
+          resolve();
+        });
+      })
+      .catch(reject);
+  });
+}
+
+export function putPreset(id, object) {
+  object = preProcessPreset(object);
+  return putResource("presets", id, object);
+}
+
+export function deletePreset(id) {
+  return deleteResource("presets", id);
+}
+
+export function putProvision(id, object) {
+  return putResource("provisions", id, object);
+}
+
+export function deleteProvision(id) {
+  return deleteResource("provisions", id);
+}
+
+export function putVirtualParameter(id, object) {
+  return putResource("virtualParameters", id, object);
+}
+
+export function deleteVirtualParameter(id) {
+  return deleteResource("virtualParameters", id);
+}
+
+export function putConfig(id, object) {
+  return putResource("config", id, object);
+}
+
+export function deleteConfig(id) {
+  return deleteResource("config", id);
+}
+
+export function putFile(filename, metadata, contentStream) {
+  return new Promise((resolve, reject) => {
+    getClient()
+      .then(client => {
+        const bucket = new GridFSBucket(client.db());
+        const uploadStream = bucket.openUploadStreamWithId(filename, filename, {
+          metadata: metadata
+        });
+        uploadStream.on("error", reject);
+        contentStream.on("error", reject);
+        uploadStream.on("finish", resolve);
+        contentStream.pipe(uploadStream);
+      })
+      .catch(reject);
+  });
+}
+
+export function deleteFile(filename) {
+  return new Promise((resolve, reject) => {
+    getClient()
+      .then(client => {
+        const bucket = new GridFSBucket(client.db());
+        bucket.delete(filename, err => {
+          if (err) return void reject(err);
+          resolve();
+        });
+      })
+      .catch(reject);
+  });
+}
+
+export function deleteFault(id) {
+  return deleteResource("faults", id);
+}
+
+export function deleteTask(id) {
+  return deleteResource("tasks", id);
+}
+
 export function disconnect() {
  if (_clientPromise) {
    _clientPromise.then(client => {
@@ -572,11 +572,3 @@ export function preProcessPreset(data) {
  preset.configurations = [configuration];
  return preset;
 }
-
-export function preProcessProvision(data) {
-  return data.script;
-}
-
-export function preProcessVirtualParameters(data) {
-  return data.script;
-}