Refactor db and local-cache code
Separated cwmp and ui code into distinct files to enhance clarity and maintainability with varied internal data representations.
This commit is contained in:
Zaid Humaimidi committed
1 parent
493f1d2f37
commit
c3954c277f
31 files changed
+1542
-1530
No files matched your search
+729
@@ -0,0 +1,729 @@
|
||||
import { decodeTag, encodeTag, escapeRegExp } from "../util";
|
||||
import {
|
||||
DeviceData,
|
||||
Attributes,
|
||||
SessionFault,
|
||||
Task,
|
||||
Operation,
|
||||
} from "../types";
|
||||
import Path from "../common/path";
|
||||
import { collections } from "../db/db";
|
||||
import { optimizeProjection } from "../db/util";
|
||||
import * as MongoTypes from "../db/types";
|
||||
import { ObjectId } from "mongodb";
|
||||
|
||||
const INVALID_PATH_SUFFIX = "__invalid";
|
||||
|
||||
function compareAccessLists(list1: string[], list2: string[]): boolean {
|
||||
if (list1.length !== list2.length) return false;
|
||||
for (const [i, v] of list1.entries()) if (v !== list2[i]) return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
export async function fetchDevice(
|
||||
id: string,
|
||||
timestamp: number
|
||||
): Promise<[Path, number, Attributes?][]> {
|
||||
const res: [Path, number, Attributes?][] = [
|
||||
[
|
||||
Path.parse("Events"),
|
||||
timestamp,
|
||||
{ object: [timestamp, 1], writable: [timestamp, 0] },
|
||||
],
|
||||
[
|
||||
Path.parse("DeviceID"),
|
||||
timestamp,
|
||||
{ object: [timestamp, 1], writable: [timestamp, 0] },
|
||||
],
|
||||
];
|
||||
|
||||
const device = await collections.devices.findOne({ _id: id });
|
||||
if (!device) return null;
|
||||
|
||||
function storeParams(
|
||||
obj,
|
||||
path: string,
|
||||
pathLength: number,
|
||||
ts: number
|
||||
): void {
|
||||
if (obj["_timestamp"]) obj["_timestamp"] = +obj["_timestamp"];
|
||||
if (obj["_attributesTimestamp"])
|
||||
obj["_attributesTimestamp"] = +obj["_attributesTimestamp"];
|
||||
|
||||
const attrs: Attributes = {};
|
||||
let t = obj["_timestamp"] || 1;
|
||||
if (ts > t) t = ts;
|
||||
|
||||
if (obj["_value"] != null) {
|
||||
attrs.value = [obj["_timestamp"] || 1, [obj["_value"], obj["_type"]]];
|
||||
if (obj["_type"] === "xsd:dateTime" && obj["_value"] instanceof Date)
|
||||
attrs.value[1][0] = +attrs.value[1][0];
|
||||
|
||||
obj["_object"] = false;
|
||||
}
|
||||
if (obj["_writable"] != null)
|
||||
attrs.writable = [ts || 1, obj["_writable"] ? 1 : 0];
|
||||
|
||||
if (obj["_object"] != null) attrs.object = [t, obj["_object"] ? 1 : 0];
|
||||
|
||||
if (obj["_notification"] != null) {
|
||||
attrs.notification = [
|
||||
obj["_attributesTimestamp"] || 1,
|
||||
obj["_notification"],
|
||||
];
|
||||
}
|
||||
|
||||
if (obj["_accessList"] != null)
|
||||
attrs.accessList = [obj["_attributesTimestamp"] || 1, obj["_accessList"]];
|
||||
|
||||
try {
|
||||
res.push([Path.parse(path), t, attrs]);
|
||||
} catch (err) {
|
||||
// The path parser is now more strict so we might be in a situation where
|
||||
// the database contains invalid paths from before this change So here we
|
||||
// encode the invalid characters.
|
||||
const splits = path.split(".");
|
||||
splits[splits.length - 1] =
|
||||
encodeTag(splits[splits.length - 1]) + INVALID_PATH_SUFFIX;
|
||||
path = splits.join(".");
|
||||
res.push([Path.parse(path), t, attrs]);
|
||||
return;
|
||||
}
|
||||
|
||||
for (const [k, v] of Object.entries(obj)) {
|
||||
if (!k.startsWith("_")) {
|
||||
obj["_object"] = true;
|
||||
storeParams(v, `${path}.${k}`, pathLength + 1, obj["_timestamp"]);
|
||||
}
|
||||
}
|
||||
|
||||
if (obj["_object"] && obj["_timestamp"])
|
||||
res.push([Path.parse(path + ".*"), obj["_timestamp"]]);
|
||||
}
|
||||
|
||||
const ts: number = +device["_timestamp"] || 0;
|
||||
if (ts) res.push([Path.parse("*"), ts]);
|
||||
|
||||
for (const [k, v] of Object.entries(device)) {
|
||||
switch (k) {
|
||||
case "_lastInform":
|
||||
res.push([
|
||||
Path.parse("Events.Inform"),
|
||||
+v,
|
||||
{
|
||||
object: [+v, 0],
|
||||
writable: [+v, 0],
|
||||
value: [+v, [+v, "xsd:dateTime"]],
|
||||
},
|
||||
]);
|
||||
break;
|
||||
case "_lastBoot":
|
||||
res.push([
|
||||
Path.parse("Events.1_BOOT"),
|
||||
+v,
|
||||
{
|
||||
object: [+v, 0],
|
||||
writable: [+v, 0],
|
||||
value: [+v, [+v, "xsd:dateTime"]],
|
||||
},
|
||||
]);
|
||||
break;
|
||||
case "_lastBootstrap":
|
||||
res.push([
|
||||
Path.parse("Events.0_BOOTSTRAP"),
|
||||
+v,
|
||||
{
|
||||
object: [+v, 0],
|
||||
writable: [+v, 0],
|
||||
value: [+v, [+v, "xsd:dateTime"]],
|
||||
},
|
||||
]);
|
||||
break;
|
||||
case "_registered":
|
||||
// Use current timestamp for registered event attribute timestamps
|
||||
res.push([
|
||||
Path.parse("Events.Registered"),
|
||||
timestamp,
|
||||
{
|
||||
object: [timestamp, 0],
|
||||
writable: [timestamp, 0],
|
||||
value: [timestamp, [+v, "xsd:dateTime"]],
|
||||
},
|
||||
]);
|
||||
break;
|
||||
case "_id":
|
||||
res.push([
|
||||
Path.parse("DeviceID.ID"),
|
||||
timestamp,
|
||||
{
|
||||
object: [timestamp, 0],
|
||||
writable: [timestamp, 0],
|
||||
value: [timestamp, [v as string, "xsd:string"]],
|
||||
},
|
||||
]);
|
||||
break;
|
||||
case "_tags":
|
||||
if ((v as string[]).length) {
|
||||
res.push([
|
||||
Path.parse("Tags"),
|
||||
timestamp,
|
||||
{ object: [timestamp, 1], writable: [timestamp, 0] },
|
||||
]);
|
||||
}
|
||||
|
||||
for (const t of v as string[]) {
|
||||
res.push([
|
||||
Path.parse("Tags." + encodeTag(t)),
|
||||
timestamp,
|
||||
{
|
||||
object: [timestamp, 0],
|
||||
writable: [timestamp, 1],
|
||||
value: [timestamp, [true, "xsd:boolean"]],
|
||||
},
|
||||
]);
|
||||
}
|
||||
break;
|
||||
case "_deviceId":
|
||||
if (v["_Manufacturer"] != null) {
|
||||
res.push([
|
||||
Path.parse("DeviceID.Manufacturer"),
|
||||
timestamp,
|
||||
{
|
||||
object: [timestamp, 0],
|
||||
writable: [timestamp, 0],
|
||||
value: [timestamp, [v["_Manufacturer"], "xsd:string"]],
|
||||
},
|
||||
]);
|
||||
}
|
||||
|
||||
if (v["_OUI"] != null) {
|
||||
res.push([
|
||||
Path.parse("DeviceID.OUI"),
|
||||
timestamp,
|
||||
{
|
||||
object: [timestamp, 0],
|
||||
writable: [timestamp, 0],
|
||||
value: [timestamp, [v["_OUI"], "xsd:string"]],
|
||||
},
|
||||
]);
|
||||
}
|
||||
|
||||
if (v["_ProductClass"] != null) {
|
||||
res.push([
|
||||
Path.parse("DeviceID.ProductClass"),
|
||||
timestamp,
|
||||
{
|
||||
object: [timestamp, 0],
|
||||
writable: [timestamp, 0],
|
||||
value: [timestamp, [v["_ProductClass"], "xsd:string"]],
|
||||
},
|
||||
]);
|
||||
}
|
||||
|
||||
if (v["_SerialNumber"] != null) {
|
||||
res.push([
|
||||
Path.parse("DeviceID.SerialNumber"),
|
||||
timestamp,
|
||||
{
|
||||
object: [timestamp, 0],
|
||||
writable: [timestamp, 0],
|
||||
value: [timestamp, [v["_SerialNumber"], "xsd:string"]],
|
||||
},
|
||||
]);
|
||||
}
|
||||
break;
|
||||
default:
|
||||
if (!k.startsWith("_")) storeParams(v, k, 1, ts);
|
||||
}
|
||||
}
|
||||
return res;
|
||||
}
|
||||
|
||||
export async function saveDevice(
|
||||
deviceId: string,
|
||||
deviceData: DeviceData,
|
||||
isNew: boolean,
|
||||
sessionTimestamp: number
|
||||
): Promise<void> {
|
||||
const update = { $set: {}, $unset: {}, $addToSet: {}, $pull: {} };
|
||||
|
||||
for (const diff of deviceData.timestamps.diff()) {
|
||||
if (diff[0].wildcard !== 1 << (diff[0].length - 1)) continue;
|
||||
|
||||
if (
|
||||
diff[0].segments[0] === "Events" ||
|
||||
diff[0].segments[0] === "DeviceID" ||
|
||||
diff[0].segments[0] === "Tags"
|
||||
)
|
||||
continue;
|
||||
|
||||
const parent = deviceData.paths.get(diff[0].slice(0, -1));
|
||||
|
||||
// Param timestamps may be greater than session timestamp to track revisions
|
||||
if (diff[2] > sessionTimestamp) diff[2] = sessionTimestamp;
|
||||
|
||||
if (diff[2] == null && diff[1] != null) {
|
||||
update["$unset"][
|
||||
parent.length ? parent.toString() + "._timestamp" : "_timestamp"
|
||||
] = 1;
|
||||
} else {
|
||||
if (parent && (!parent.length || deviceData.attributes.has(parent))) {
|
||||
update["$set"][
|
||||
parent.length ? parent.toString() + "._timestamp" : "_timestamp"
|
||||
] = new Date(diff[2]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (const diff of deviceData.attributes.diff()) {
|
||||
const path = diff[0];
|
||||
const value1 = (((diff[1] || {}).value || [])[1] || [])[0];
|
||||
const value2 = (((diff[2] || {}).value || [])[1] || [])[0];
|
||||
const valueType1 = (((diff[1] || {}).value || [])[1] || [])[1];
|
||||
const valueType2 = (((diff[2] || {}).value || [])[1] || [])[1];
|
||||
const valueTimestamp1 = ((diff[1] || {}).value || [])[0];
|
||||
const valueTimestamp2 = ((diff[2] || {}).value || [])[0];
|
||||
const object1 = ((diff[1] || {}).object || [])[1];
|
||||
const object2 = ((diff[2] || {}).object || [])[1];
|
||||
const writable2 = ((diff[2] || {}).writable || [])[1];
|
||||
const writable1 = ((diff[1] || {}).writable || [])[1];
|
||||
const attributesTimestamp1 = ((diff[1] || {}).notification || [])[0];
|
||||
const attributesTimestamp2 = ((diff[2] || {}).notification || [])[0];
|
||||
const notification1 = ((diff[1] || {}).notification || [])[1];
|
||||
const notification2 = ((diff[2] || {}).notification || [])[1];
|
||||
const accessList1 = ((diff[1] || {}).accessList || [])[1];
|
||||
const accessList2 = ((diff[2] || {}).accessList || [])[1];
|
||||
|
||||
switch (path.segments[0]) {
|
||||
case "Events":
|
||||
if (path.length === 2 && value2 !== value1) {
|
||||
if (!diff[2]) {
|
||||
switch (path.segments[1]) {
|
||||
case "Inform":
|
||||
update["$unset"]["_lastInform"] = 1;
|
||||
break;
|
||||
case "1_BOOT":
|
||||
update["$unset"]["_lastBoot"] = 1;
|
||||
break;
|
||||
case "0_BOOTSTRAP":
|
||||
update["$unset"]["_lastBootstrap"] = 1;
|
||||
break;
|
||||
case "Registered":
|
||||
update["$unset"]["_registered"] = 1;
|
||||
}
|
||||
} else {
|
||||
const t = new Date(diff[2].value[1][0] as number);
|
||||
switch (path.segments[1]) {
|
||||
case "Inform":
|
||||
update["$set"]["_lastInform"] = t;
|
||||
break;
|
||||
case "1_BOOT":
|
||||
update["$set"]["_lastBoot"] = t;
|
||||
break;
|
||||
case "0_BOOTSTRAP":
|
||||
update["$set"]["_lastBootstrap"] = t;
|
||||
break;
|
||||
case "Registered":
|
||||
update["$set"]["_registered"] = t;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
break;
|
||||
case "DeviceID":
|
||||
if (value2 !== value1) {
|
||||
const v = diff[2].value[1][0];
|
||||
switch (path.segments[1]) {
|
||||
case "ID":
|
||||
update["$set"]["_id"] = v;
|
||||
break;
|
||||
case "Manufacturer":
|
||||
update["$set"]["_deviceId._Manufacturer"] = v;
|
||||
break;
|
||||
case "OUI":
|
||||
update["$set"]["_deviceId._OUI"] = v;
|
||||
break;
|
||||
case "ProductClass":
|
||||
update["$set"]["_deviceId._ProductClass"] = v;
|
||||
break;
|
||||
case "SerialNumber":
|
||||
update["$set"]["_deviceId._SerialNumber"] = v;
|
||||
}
|
||||
}
|
||||
break;
|
||||
case "Tags":
|
||||
if (value2 !== value1) {
|
||||
if (value2 != null) {
|
||||
if (!update["$addToSet"]["_tags"])
|
||||
update["$addToSet"]["_tags"] = { $each: [] };
|
||||
update["$addToSet"]["_tags"]["$each"].push(
|
||||
decodeTag(path.segments[1] as string)
|
||||
);
|
||||
} else {
|
||||
if (!update["$pull"]["_tags"]) {
|
||||
update["$pull"]["_tags"] = {
|
||||
$in: [],
|
||||
};
|
||||
}
|
||||
update["$pull"]["_tags"]["$in"].push(
|
||||
decodeTag(path.segments[1] as string)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
break;
|
||||
default:
|
||||
if (!diff[2]) {
|
||||
let pathStr = path.toString();
|
||||
// Paths with that suffix are encoded and need to be decoded
|
||||
if (pathStr.endsWith(INVALID_PATH_SUFFIX)) {
|
||||
const splits = pathStr.split(".");
|
||||
splits[splits.length - 1] = decodeTag(
|
||||
splits[splits.length - 1].slice(0, 0 - INVALID_PATH_SUFFIX.length)
|
||||
);
|
||||
pathStr = splits.join(".");
|
||||
}
|
||||
update["$unset"][pathStr] = 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const attrName of Object.keys(diff[2])) {
|
||||
// Param timestamps may be greater than session timestamp to track revisions
|
||||
if (diff[2][attrName][0] > sessionTimestamp)
|
||||
diff[2][attrName][0] = sessionTimestamp;
|
||||
|
||||
if (diff[2][attrName][1] != null) {
|
||||
switch (attrName) {
|
||||
case "value":
|
||||
if (value2 !== value1) {
|
||||
if (
|
||||
valueType2 === "xsd:dateTime" &&
|
||||
Number.isInteger(value2 as number)
|
||||
) {
|
||||
update["$set"][path.toString() + "._value"] = new Date(
|
||||
value2 as number
|
||||
);
|
||||
} else {
|
||||
update["$set"][path.toString() + "._value"] = value2;
|
||||
}
|
||||
}
|
||||
|
||||
if (valueType2 !== valueType1)
|
||||
update["$set"][path.toString() + "._type"] = valueType2;
|
||||
|
||||
if (valueTimestamp2 !== valueTimestamp1) {
|
||||
update["$set"][path.toString() + "._timestamp"] = new Date(
|
||||
valueTimestamp2
|
||||
);
|
||||
}
|
||||
|
||||
break;
|
||||
case "object":
|
||||
if (!diff[1]?.object || object2 !== object1) {
|
||||
update["$set"][
|
||||
path.length ? path.toString() + "._object" : "_object"
|
||||
] = !!object2;
|
||||
}
|
||||
|
||||
break;
|
||||
case "writable":
|
||||
if (!diff[1]?.writable || writable2 !== writable1) {
|
||||
update["$set"][
|
||||
path.length ? path.toString() + "._writable" : "_writable"
|
||||
] = !!writable2;
|
||||
}
|
||||
|
||||
break;
|
||||
case "notification":
|
||||
if (
|
||||
!diff[1] ||
|
||||
!diff[1].notification ||
|
||||
notification2 !== notification1
|
||||
) {
|
||||
update["$set"][
|
||||
path.length
|
||||
? path.toString() + "._notification"
|
||||
: "_notification"
|
||||
] = notification2;
|
||||
}
|
||||
|
||||
if (attributesTimestamp2 !== attributesTimestamp1) {
|
||||
update["$set"][path.toString() + "._attributesTimestamp"] =
|
||||
new Date(attributesTimestamp2);
|
||||
}
|
||||
|
||||
break;
|
||||
case "accessList":
|
||||
if (
|
||||
!diff[1] ||
|
||||
!diff[1].accessList ||
|
||||
!compareAccessLists(accessList2, accessList1)
|
||||
) {
|
||||
update["$set"][
|
||||
path.length
|
||||
? path.toString() + "._accessList"
|
||||
: "_accessList"
|
||||
] = accessList2;
|
||||
}
|
||||
|
||||
if (attributesTimestamp2 !== attributesTimestamp1) {
|
||||
update["$set"][path.toString() + "._attributesTimestamp"] =
|
||||
new Date(attributesTimestamp2);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (diff[1]) {
|
||||
for (const attrName of Object.keys(diff[1])) {
|
||||
if (
|
||||
diff[1][attrName][1] != null &&
|
||||
diff[2]?.[attrName]?.[1] == null
|
||||
) {
|
||||
const p = path.length ? path.toString() + "." : "";
|
||||
update["$unset"][`${p}_${attrName}`] = 1;
|
||||
if (attrName === "value") {
|
||||
update["$unset"][p + "_type"] = 1;
|
||||
update["$unset"][p + "_timestamp"] = 1;
|
||||
} else if (attrName === "notification") {
|
||||
if (accessList2 == null)
|
||||
update["$unset"][`${p}_attributesTimestamp`] = 1;
|
||||
} else if (attrName === "accessList") {
|
||||
if (notification2 == null)
|
||||
update["$unset"][`${p}_attributesTimestamp`] = 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
update["$unset"] = optimizeProjection(update["$unset"]);
|
||||
|
||||
// Remove overlap possibly caused by parameters changing from objects
|
||||
// to regular parameters or vice versa. Reason being that _timestamp
|
||||
// represents two different things depending on whether the parameter
|
||||
// is an object or not.
|
||||
for (const k of Object.keys(update["$unset"]))
|
||||
if (update["$set"][k] != null) delete update["$unset"][k];
|
||||
|
||||
// Remove empty keys
|
||||
for (const [k, v] of Object.entries(update)) {
|
||||
if (k === "$addToSet") {
|
||||
for (const [kk, vv] of Object.entries(v))
|
||||
if (!vv["$each"].length) delete v[kk];
|
||||
} else if (k === "$pull") {
|
||||
for (const [kk, vv] of Object.entries(v))
|
||||
if (!vv["$in"].length) delete v[kk];
|
||||
}
|
||||
if (!Object.keys(v).length) delete update[k];
|
||||
}
|
||||
|
||||
if (!Object.keys(update).length) return;
|
||||
|
||||
// Mongo doesn't allow $addToSet and $pull at the same time
|
||||
let update2;
|
||||
if (update["$addToSet"] && update["$pull"]) {
|
||||
update2 = { $pull: update["$pull"] };
|
||||
delete update["$pull"];
|
||||
}
|
||||
|
||||
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 collections.devices.updateOne({ _id: deviceId }, update2);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
export async function getFaults(
|
||||
deviceId: string
|
||||
): Promise<{ [channel: string]: SessionFault }> {
|
||||
const res = await collections.faults
|
||||
.find({ _id: { $regex: `^${escapeRegExp(deviceId)}\\:` } })
|
||||
.toArray();
|
||||
|
||||
const faults: { [channel: string]: SessionFault } = {};
|
||||
for (const r of res) {
|
||||
const channel = r._id.slice(deviceId.length + 1);
|
||||
const fault: SessionFault = {
|
||||
code: r.code,
|
||||
message: r.message,
|
||||
...(r.detail && { detail: r.detail }),
|
||||
timestamp: +r.timestamp,
|
||||
provisions: JSON.parse(r.provisions),
|
||||
retries: r.retries,
|
||||
...(r.expiry && { expiry: +r.expiry }),
|
||||
};
|
||||
faults[channel] = fault;
|
||||
}
|
||||
|
||||
return faults;
|
||||
}
|
||||
|
||||
export async function saveFault(
|
||||
deviceId: string,
|
||||
channel: string,
|
||||
fault: SessionFault
|
||||
): Promise<void> {
|
||||
const id = `${deviceId}:${channel}`;
|
||||
const f: MongoTypes.Fault = {
|
||||
_id: id,
|
||||
device: deviceId,
|
||||
channel: channel,
|
||||
timestamp: new Date(fault.timestamp),
|
||||
code: fault.code,
|
||||
message: fault.message,
|
||||
...(fault.detail && { detail: fault.detail }),
|
||||
retries: fault.retries,
|
||||
...(fault.expiry && { expiry: new Date(fault.expiry) }),
|
||||
provisions: JSON.stringify(fault.provisions),
|
||||
};
|
||||
await collections.faults.replaceOne({ _id: id }, f, { upsert: true });
|
||||
}
|
||||
|
||||
export async function deleteFault(
|
||||
deviceId: string,
|
||||
channel: string
|
||||
): Promise<void> {
|
||||
await collections.faults.deleteOne({ _id: `${deviceId}:${channel}` });
|
||||
}
|
||||
|
||||
export async function getDueTasks(
|
||||
deviceId: string,
|
||||
timestamp: number
|
||||
): Promise<[Task[], number]> {
|
||||
const cur = collections.tasks
|
||||
.find({ device: deviceId })
|
||||
.sort({ timestamp: 1 });
|
||||
const tasks = [] as Task[];
|
||||
|
||||
for await (const t of cur) {
|
||||
if (+t.timestamp >= timestamp) return [tasks, +t.timestamp];
|
||||
const task: Task = {
|
||||
_id: t._id.toString(),
|
||||
name: t.name,
|
||||
...(t.timestamp && { timestamp: +t.timestamp }),
|
||||
...(t.expiry && { expiry: +t.expiry }),
|
||||
...(t.name === "getParameterValues" && {
|
||||
parameterNames: t.parameterNames,
|
||||
}),
|
||||
...(t.name === "setParameterValues" && {
|
||||
parameterValues: t.parameterValues,
|
||||
}),
|
||||
...(t.name === "refreshObject" && {
|
||||
objectName: t.objectName,
|
||||
}),
|
||||
...(t.name === "download" && {
|
||||
fileType: t.fileType,
|
||||
fileName: t.fileName,
|
||||
targetFileName: t.targetFileName,
|
||||
}),
|
||||
...(t.name === "addObject" && {
|
||||
objectName: t.objectName,
|
||||
parameterValues: t.parameterValues,
|
||||
}),
|
||||
...(t.name === "deleteObject" && {
|
||||
objectName: t.objectName,
|
||||
}),
|
||||
...(t.name === "provisions" && {
|
||||
provisions: t.provisions,
|
||||
}),
|
||||
};
|
||||
|
||||
tasks.push(task);
|
||||
|
||||
// For API compatibility
|
||||
if (task.name === "download" && t["file"]) {
|
||||
let q;
|
||||
if (ObjectId.isValid(t["file"]))
|
||||
q = { _id: { $in: [t["file"], new ObjectId(t["file"])] } };
|
||||
else q = { _id: t["file"] };
|
||||
|
||||
const res = await collections.files.find(q).toArray();
|
||||
|
||||
if (res[0]) {
|
||||
if (!task.fileType) task.fileType = res[0].metadata.fileType;
|
||||
|
||||
if (!task.fileName)
|
||||
task.fileName = res[0].filename || res[0]._id.toString();
|
||||
}
|
||||
}
|
||||
}
|
||||
return [tasks, null];
|
||||
}
|
||||
|
||||
export async function clearTasks(
|
||||
deviceId: string,
|
||||
taskIds: string[]
|
||||
): Promise<void> {
|
||||
await collections.tasks.deleteMany({
|
||||
_id: { $in: taskIds.map((id) => new ObjectId(id)) },
|
||||
});
|
||||
}
|
||||
|
||||
export async function getOperations(
|
||||
deviceId: string
|
||||
): Promise<{ [commandKey: string]: Operation }> {
|
||||
const res = await collections.operations
|
||||
.find({ _id: { $regex: `^${escapeRegExp(deviceId)}\\:` } })
|
||||
.toArray();
|
||||
|
||||
const operations: { [commandKey: string]: Operation } = {};
|
||||
for (const r of res) {
|
||||
const commandKey = r._id.slice(deviceId.length + 1);
|
||||
// Workaround for a bug in v1.2.1 where operation object is saved without deserialization
|
||||
if (typeof r.provisions !== "string") {
|
||||
delete r._id;
|
||||
operations[commandKey] = r as unknown as Operation;
|
||||
continue;
|
||||
}
|
||||
const operation: Operation = {
|
||||
name: r.name,
|
||||
timestamp: +r.timestamp,
|
||||
channels:
|
||||
typeof r.channels === "string" ? JSON.parse(r.channels) : r.channels,
|
||||
retries: JSON.parse(r.retries),
|
||||
provisions: JSON.parse(r.provisions),
|
||||
...(r.args && { args: JSON.parse(r.args) }),
|
||||
};
|
||||
operations[commandKey] = operation;
|
||||
}
|
||||
return operations;
|
||||
}
|
||||
|
||||
export async function saveOperation(
|
||||
deviceId: string,
|
||||
commandKey: string,
|
||||
operation: Operation
|
||||
): Promise<void> {
|
||||
const id = `${deviceId}:${commandKey}`;
|
||||
const o: MongoTypes.Operation = {
|
||||
_id: id,
|
||||
name: operation.name,
|
||||
timestamp: new Date(operation.timestamp),
|
||||
channels: JSON.stringify(operation.channels),
|
||||
provisions: JSON.stringify(operation.provisions),
|
||||
retries: JSON.stringify(operation.retries),
|
||||
args: JSON.stringify(operation.args),
|
||||
};
|
||||
await collections.operations.replaceOne({ _id: id }, o, {
|
||||
upsert: true,
|
||||
});
|
||||
}
|
||||
|
||||
export async function deleteOperation(
|
||||
deviceId: string,
|
||||
commandKey: string
|
||||
): Promise<void> {
|
||||
await collections.operations.deleteOne({ _id: `${deviceId}:${commandKey}` });
|
||||
}
|
||||
@@ -0,0 +1,359 @@
|
||||
import * as vm from "vm";
|
||||
import * as crypto from "crypto";
|
||||
import * as config from "../config";
|
||||
import { collections } from "../db/db";
|
||||
import { convertOldPrecondition } from "../db/util";
|
||||
import * as logger from "../logger";
|
||||
import * as scheduling from "../scheduling";
|
||||
import * as expression from "../common/expression/util";
|
||||
import { parse } from "../common/expression/parser";
|
||||
import {
|
||||
Preset,
|
||||
Expression,
|
||||
Provisions,
|
||||
VirtualParameters,
|
||||
Files,
|
||||
Config,
|
||||
} from "../types";
|
||||
import { LocalCache } from "../local-cache";
|
||||
|
||||
interface Snapshot {
|
||||
presets: Preset[];
|
||||
provisions: Provisions;
|
||||
virtualParameters: VirtualParameters;
|
||||
files: Files;
|
||||
config: Config;
|
||||
}
|
||||
|
||||
function flattenObject<T extends Record<keyof T, unknown>>(
|
||||
src: T,
|
||||
prefix = "",
|
||||
dst = {} as T
|
||||
): T {
|
||||
const t: Record<string, string> = {};
|
||||
t["hi"] = "hi";
|
||||
for (const k of Object.keys(src)) {
|
||||
const v = src[k];
|
||||
if (typeof v === "object" && !Array.isArray(v))
|
||||
flattenObject(v as T, `${prefix}${k}.`, dst);
|
||||
else dst[`${prefix}${k}`] = v;
|
||||
}
|
||||
return dst;
|
||||
}
|
||||
|
||||
async function fetchPresets(): Promise<[string, Preset[]]> {
|
||||
const res = await collections.presets.find().toArray();
|
||||
let objects = await collections.objects.find().toArray();
|
||||
res.sort((a, b) => (a._id > b._id ? 1 : -1));
|
||||
objects.sort((a, b) => (a._id > b._id ? 1 : -1));
|
||||
const h = crypto
|
||||
.createHash("md5")
|
||||
.update(JSON.stringify(res))
|
||||
.update(JSON.stringify(objects))
|
||||
.digest("hex");
|
||||
|
||||
objects = objects.map((obj) => {
|
||||
// Flatten object
|
||||
obj = flattenObject(obj);
|
||||
|
||||
// If no keys are defined, consider all parameters as keys to keep the
|
||||
// same behavior from v1.0
|
||||
if (!obj["_keys"]?.length)
|
||||
obj["_keys"] = Object.keys(obj).filter((k) => !k.startsWith("_"));
|
||||
|
||||
return obj;
|
||||
});
|
||||
|
||||
res.sort((a, b) => {
|
||||
if (a["weight"] === b["weight"])
|
||||
return a["_id"] > b["_id"] ? 1 : a["_id"] < b["_id"] ? -1 : 0;
|
||||
else return a["weight"] - b["weight"];
|
||||
});
|
||||
|
||||
const presets = [] as Preset[];
|
||||
for (const preset of res) {
|
||||
let schedule: { md5: string; duration: number; schedule: any } = null;
|
||||
if (preset["schedule"]) {
|
||||
const parts = preset["schedule"].trim().split(/\s+/);
|
||||
schedule = {
|
||||
md5: crypto.createHash("md5").update(preset["schedule"]).digest("hex"),
|
||||
duration: null,
|
||||
schedule: null,
|
||||
};
|
||||
|
||||
try {
|
||||
schedule.duration = +parts.shift() * 1000;
|
||||
schedule.schedule = scheduling.parseCron(parts.join(" "));
|
||||
} catch (err) {
|
||||
logger.warn({
|
||||
message: "Invalid preset schedule",
|
||||
preset: preset["_id"],
|
||||
schedule: preset["schedule"],
|
||||
});
|
||||
schedule.schedule = false;
|
||||
}
|
||||
}
|
||||
|
||||
const events = preset["events"] || {};
|
||||
let precondition = true as Expression;
|
||||
if (preset["precondition"]) {
|
||||
try {
|
||||
precondition = parse(preset["precondition"]);
|
||||
} catch (error) {
|
||||
precondition = convertOldPrecondition(
|
||||
JSON.parse(preset["precondition"])
|
||||
);
|
||||
}
|
||||
|
||||
// Simplify expression
|
||||
precondition = expression.evaluate(precondition);
|
||||
}
|
||||
|
||||
const _provisions: Preset["provisions"] = [];
|
||||
|
||||
// Generate provisions from the old configuration format
|
||||
for (const c of preset.configurations) {
|
||||
switch (c.type) {
|
||||
case "age":
|
||||
_provisions.push(["refresh", c.name, +c.age]);
|
||||
break;
|
||||
|
||||
case "value":
|
||||
_provisions.push(["value", c.name, c.value]);
|
||||
break;
|
||||
|
||||
case "add_tag":
|
||||
_provisions.push(["tag", c.tag, true]);
|
||||
break;
|
||||
|
||||
case "delete_tag":
|
||||
_provisions.push(["tag", c.tag, false]);
|
||||
break;
|
||||
|
||||
case "provision":
|
||||
_provisions.push([c.name, ...(c.args || [])]);
|
||||
break;
|
||||
|
||||
case "add_object":
|
||||
for (const obj of objects) {
|
||||
if (obj["_id"] === c.object) {
|
||||
const alias = obj["_keys"]
|
||||
.map((k) => `${k}:${JSON.stringify(obj[k])}`)
|
||||
.join(",");
|
||||
const p = `${c.name}.[${alias}]`;
|
||||
_provisions.push(["instances", p, 1]);
|
||||
|
||||
for (const k in obj) {
|
||||
if (!k.startsWith("_") && !(obj["_keys"].indexOf(k) !== -1))
|
||||
_provisions.push(["value", `${p}.${k}`, obj[k]]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
break;
|
||||
|
||||
case "delete_object":
|
||||
for (const obj of objects) {
|
||||
if (obj["_id"] === c.object) {
|
||||
const alias = obj["_keys"]
|
||||
.map((k) => `${k}:${JSON.stringify(obj[k])}`)
|
||||
.join(",");
|
||||
const p = `${c.name}.[${alias}]`;
|
||||
_provisions.push(["instances", p, 0]);
|
||||
}
|
||||
}
|
||||
|
||||
break;
|
||||
|
||||
default: {
|
||||
const exhaustiveCheck: never = c;
|
||||
throw new Error(
|
||||
`Unknown configuration type ${exhaustiveCheck["type"]}`
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
presets.push({
|
||||
name: preset["_id"],
|
||||
channel: (preset["channel"] as string) || "default",
|
||||
schedule: schedule,
|
||||
events: events,
|
||||
precondition: precondition,
|
||||
provisions: _provisions,
|
||||
});
|
||||
}
|
||||
|
||||
return [h, presets];
|
||||
}
|
||||
|
||||
async function fetchProvisions(): Promise<[string, Provisions]> {
|
||||
const res = await collections.provisions.find().toArray();
|
||||
res.sort((a, b) => (a._id > b._id ? 1 : -1));
|
||||
const h = crypto.createHash("md5").update(JSON.stringify(res)).digest("hex");
|
||||
|
||||
const provisions = {};
|
||||
for (const r of res) {
|
||||
provisions[r._id] = {};
|
||||
provisions[r._id].md5 = crypto
|
||||
.createHash("md5")
|
||||
.update(r.script)
|
||||
.digest("hex");
|
||||
provisions[r._id].script = new vm.Script(
|
||||
`"use strict";(function(){\n${r.script}\n})();`,
|
||||
{ filename: r._id, lineOffset: -1, timeout: 50 }
|
||||
);
|
||||
}
|
||||
|
||||
return [h, provisions];
|
||||
}
|
||||
|
||||
async function fetchVirtualParameters(): Promise<[string, VirtualParameters]> {
|
||||
const res = await collections.virtualParameters.find().toArray();
|
||||
res.sort((a, b) => (a._id > b._id ? 1 : -1));
|
||||
const h = crypto.createHash("md5").update(JSON.stringify(res)).digest("hex");
|
||||
|
||||
const virtualParameters = {};
|
||||
for (const r of res) {
|
||||
virtualParameters[r._id] = {};
|
||||
virtualParameters[r._id].md5 = crypto
|
||||
.createHash("md5")
|
||||
.update(r.script)
|
||||
.digest("hex");
|
||||
virtualParameters[r._id].script = new vm.Script(
|
||||
`"use strict";(function(){\n${r.script}\n})();`,
|
||||
{ filename: r._id, lineOffset: -1, timeout: 50 }
|
||||
);
|
||||
}
|
||||
|
||||
return [h, virtualParameters];
|
||||
}
|
||||
|
||||
async function fetchFiles(): Promise<[string, Files]> {
|
||||
const res = await collections.files.find().toArray();
|
||||
res.sort((a, b) => (a._id > b._id ? 1 : -1));
|
||||
const h = crypto.createHash("md5").update(JSON.stringify(res)).digest("hex");
|
||||
const files = {};
|
||||
|
||||
for (const r of res) {
|
||||
const id = r.filename || r._id.toString();
|
||||
files[id] = {};
|
||||
files[id].length = r.length;
|
||||
}
|
||||
|
||||
return [h, files];
|
||||
}
|
||||
|
||||
async function fetchConfig(): Promise<[string, Config]> {
|
||||
const conf = await collections.config.find().toArray();
|
||||
conf.sort((a, b) => (a._id > b._id ? 1 : -1));
|
||||
const h = crypto.createHash("md5").update(JSON.stringify(conf)).digest("hex");
|
||||
|
||||
const _config = {};
|
||||
|
||||
for (const c of conf) {
|
||||
// Evaluate expressions to simplify them
|
||||
_config[c._id] = expression.evaluate(parse(c.value));
|
||||
}
|
||||
|
||||
return [h, _config];
|
||||
}
|
||||
|
||||
const localCache = new LocalCache<Snapshot>("cwmp-local-cache-hash", refresh);
|
||||
|
||||
async function refresh(): Promise<[string, Snapshot]> {
|
||||
const res = await Promise.all([
|
||||
fetchPresets(),
|
||||
fetchProvisions(),
|
||||
fetchVirtualParameters(),
|
||||
fetchFiles(),
|
||||
fetchConfig(),
|
||||
]);
|
||||
|
||||
const h = crypto.createHash("md5");
|
||||
for (const r of res) h.update(r[0]);
|
||||
|
||||
const snapshot = {
|
||||
presets: res[0][1],
|
||||
provisions: res[1][1],
|
||||
virtualParameters: res[2][1],
|
||||
files: res[3][1],
|
||||
config: res[4][1],
|
||||
};
|
||||
|
||||
return [h.digest("hex"), snapshot];
|
||||
}
|
||||
|
||||
export async function getRevision(): Promise<string> {
|
||||
return await localCache.getRevision();
|
||||
}
|
||||
|
||||
export function getPresets(revision: string): Preset[] {
|
||||
return localCache.get(revision).presets;
|
||||
}
|
||||
|
||||
export function getProvisions(revision: string): Provisions {
|
||||
return localCache.get(revision).provisions;
|
||||
}
|
||||
|
||||
export function getVirtualParameters(revision: string): VirtualParameters {
|
||||
return localCache.get(revision).virtualParameters;
|
||||
}
|
||||
|
||||
export function getFiles(revision: string): Files {
|
||||
return localCache.get(revision).files;
|
||||
}
|
||||
|
||||
export function getConfig(
|
||||
snapshotKey: string,
|
||||
key: string,
|
||||
context: Record<string, unknown>,
|
||||
now: number,
|
||||
cb?: (e: Expression) => Expression
|
||||
): string | number | boolean | null {
|
||||
const snapshot = localCache.get(snapshotKey);
|
||||
if (!snapshot) throw new Error("Cache snapshot does not exist");
|
||||
|
||||
const oldOpts = {
|
||||
"cwmp.downloadTimeout": "DOWNLOAD_TIMEOUT",
|
||||
"cwmp.debug": "DEBUG",
|
||||
"cwmp.retryDelay": "RETRY_DELAY",
|
||||
"cwmp.sessionTimeout": "SESSION_TIMEOUT",
|
||||
"cwmp.connectionRequestTimeout": "CONNECTION_REQUEST_TIMEOUT",
|
||||
"cwmp.gpnNextLevel": "GPN_NEXT_LEVEL",
|
||||
"cwmp.gpvBatchSize": "GPV_BATCH_SIZE",
|
||||
"cwmp.cookiesPath": "COOKIES_PATH",
|
||||
"cwmp.datetimeMilliseconds": "DATETIME_MILLISECONDS",
|
||||
"cwmp.booleanLiteral": "BOOLEAN_LITERAL",
|
||||
"cwmp.connectionRequestAllowBasicAuth":
|
||||
"CONNECTION_REQUEST_ALLOW_BASIC_AUTH",
|
||||
"cwmp.maxCommitIterations": "MAX_COMMIT_ITERATIONS",
|
||||
"cwmp.deviceOnlineThreshold": "DEVICE_ONLINE_THRESHOLD",
|
||||
"cwmp.udpConnectionRequestPort": "UDP_CONNECTION_REQUEST_PORT",
|
||||
};
|
||||
|
||||
if (!(key in snapshot.config)) {
|
||||
if (key in oldOpts) {
|
||||
let id;
|
||||
if (context?.["id"]) {
|
||||
id = context["id"];
|
||||
} else if (cb) {
|
||||
id = cb(["PARAM", "DeviceID.ID"]);
|
||||
if (Array.isArray(id)) id = null;
|
||||
}
|
||||
return config.get(oldOpts[key], id);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
const v = expression.evaluate(snapshot.config[key], context, now, cb);
|
||||
return Array.isArray(v) ? null : v;
|
||||
}
|
||||
|
||||
export function getConfigExpression(
|
||||
snapshotKey: string,
|
||||
key: string
|
||||
): Expression {
|
||||
const snapshot = localCache.get(snapshotKey);
|
||||
return snapshot.config[key];
|
||||
}
|
||||
Reference in new issue
Block a user