"use strict"; const http = require("http"); const https = require("https"); const url = require("url"); const querystring = require("querystring"); const config = require("./config"); const device = require("./device"); const filterParser = require("../common/filter-parser"); const Filter = require("../common/filter"); const commaAscii = ",".charCodeAt(0); const newLineAscii = "\n".charCodeAt(0); function unpackTimestamps(filter) { return filterParser.map(filter, exp => { if (["=", "<>", ">=", "<=", "<", ">"].includes(exp[0])) if (typeof exp[2] === "number") { let alt = exp.slice(); alt[2] = new Date(alt[2]).toJSON(); return ["OR", exp, alt]; } }); } function filterToMongoQuery(filter, negate = false, res = {}) { const op = filter[0]; if ((!negate && op === "AND") || (negate && op === "OR")) { res["$and"] = res["$and"] || []; for (let i = 1; i < filter.length; ++i) res["$and"].push(filterToMongoQuery(filter[i], negate)); } else if ((!negate && op === "OR") || (negate && op === "AND")) { res["$or"] = res["$or"] || []; for (let i = 1; i < filter.length; ++i) res["$or"].push(filterToMongoQuery(filter[i], negate)); } else if (op === "NOT") { filterToMongoQuery(filter[1], !negate, res); } else if (op === "=") { let p = (res[filter[1]] = res[filter[1]] || {}); if (negate) p["$ne"] = filter[2]; else p["$eq"] = filter[2]; } else if (op === "<>") { let p = (res[filter[1]] = res[filter[1]] || {}); if (negate) { p["$eq"] = filter[2]; } else { p["$ne"] = filter[2]; p["$exists"] = true; } } else if (op === ">") { let p = (res[filter[1]] = res[filter[1]] || {}); if (negate) p = p["$not"] = p["$not"] || {}; p["$gt"] = filter[2]; } else if (op === ">=") { let p = (res[filter[1]] = res[filter[1]] || {}); if (negate) p = p["$not"] = p["$not"] || {}; p["$gte"] = filter[2]; } else if (op === "<") { let p = (res[filter[1]] = res[filter[1]] || {}); if (negate) p = p["$not"] = p["$not"] || {}; p["$lt"] = filter[2]; } else if (op === "<=") { let p = (res[filter[1]] = res[filter[1]] || {}); if (negate) p = p["$not"] = p["$not"] || {}; p["$lte"] = filter[2]; } else if (op === "IS NULL") { res[filter[1]] = { $exists: negate }; } else if (op === "IS NOT NULL") { res[filter[1]] = { $exists: !negate }; } else { throw new Error(`Unrecognized operator ${op}`); } return res; } function count(resource, filter, limit) { return new Promise((resolve, reject) => { let qs = {}; if (filter && filter.ast) { filter = filter.evaluateExpressions(); let ast = unpackTimestamps(filter.ast); let q = filterToMongoQuery(ast); if (resource === "devices") q = device.transposeQuery(q); qs.query = JSON.stringify(q); } if (limit) qs.limit = limit; let options = url.parse( `${config.get("server.nbi")}${resource}?${querystring.stringify(qs)}` ); options.method = "HEAD"; let _http = options.protocol === "https:" ? https : http; _http .request(options, res => { if (res.statusCode !== 200) { reject(new Error(`Unexpected status code ${res.statusCode}`)); res.resume(); return; } resolve(+res.headers["total"]); }) .end(); }); } function query(resource, filter, limit, skip, projection, callback) { return new Promise((resolve, reject) => { let ret; if (!callback) ret = []; let qs = {}; if (filter && filter.ast) { filter = filter.evaluateExpressions(); let ast = unpackTimestamps(filter.ast); let q = filterToMongoQuery(ast); if (resource === "devices") q = device.transposeQuery(q); qs.query = JSON.stringify(q); } if (limit) qs.limit = limit; if (skip) qs.skip = skip; if (projection) qs.projection = projection; qs.sort = JSON.stringify({ _id: 1 }); let options = url.parse( `${config.get("server.nbi")}${resource}?${querystring.stringify(qs)}` ); let _http = options.protocol === "https:" ? https : http; _http.get(options, res => { if (res.statusCode !== 200) { reject(new Error(`Unexpected status code ${res.statusCode}`)); res.resume(); return; } let chunks = []; let bytes = 0; res.on("data", chunk => { let i; while ((i = chunk.indexOf(newLineAscii)) !== -1) { let buf; if (chunk[chunk.length - 2] === commaAscii) { i -= 1; buf = new Buffer(bytes + i); chunk.copy(buf, buf.length - i, 0, i); chunk = chunk.slice(i + 2); } else { buf = new Buffer(bytes + i); chunk.copy(buf, buf.length - i, 0, i); chunk = chunk.slice(i + 1); } let o = 0; for (let c of chunks) { c.copy(buf, o, 0, c.length); o += c.length; } if (buf.length > 1) { let obj = JSON.parse(buf); if (resource === "devices") obj = device.transpose(obj); if (ret) ret.push(obj); else callback(obj); } bytes = 0; chunks = []; } bytes += chunk.length; chunks.push(chunk); }); res.on("end", () => { let buf = new Buffer(bytes); let o = 0; for (let c of chunks) { c.copy(buf, o, 0, c.length); o += c.length; } if (buf.length > 1) { let obj = JSON.parse(buf); if (resource === "devices") obj = device.transpose(obj); if (ret) ret.push(obj); else callback(obj); } resolve(ret); }); res.on("aborted", () => { res.removeAllListeners("end"); reject(new Error("Timeout")); }); }); }); } function deleteResource(resource, id) { return new Promise((resolve, reject) => { let options = url.parse( `${config.get("server.nbi")}${resource}/${encodeURIComponent(id)}` ); options.method = "DELETE"; let _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) { let options = url.parse( `${config.get("server.nbi")}devices/${task.device}/tasks${ connectionRequest ? "?connection_request" : "" }` ); options.method = "POST"; let _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", () => { let buf = new Buffer(bytes); let o = 0; for (let c of chunks) { c.copy(buf, o, 0, c.length); o += c.length; } try { let 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)); } function postTasks(deviceId, tasks) { return new Promise((resolve, reject) => { let connectionRequestStatus; let promises = []; for (let [idx, task] of tasks.entries()) { task.device = deviceId; promises.push( new Promise((res, rej) => { let conReq = idx === tasks.length - 1; delete task._id; task.expiry = 5; postTask(task, conReq, (err, t, crs) => { if (err) return rej(err); if (conReq) connectionRequestStatus = crs; res({ _id: t._id, status: "pending" }); }); }) ); } Promise.all(promises) .then(statuses => { if (connectionRequestStatus !== "OK") return resolve({ connectionRequest: connectionRequestStatus, tasks: statuses }); let promises2 = []; for (let s of statuses) { promises2.push(query("tasks", new Filter(["=", "_id", s._id]))); promises2.push( query( "faults", new Filter(["=", "_id", `${deviceId}:task_${s._id}`]) ) ); } Promise.all(promises2) .then(res => { for (let [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); }) .catch(reject); }); } function updateTags(deviceId, tags) { return Promise.all( Object.entries(tags).map( ([tag, onOff]) => new Promise((resolve, reject) => { const options = url.parse( `${config.get( "server.nbi" )}devices/${deviceId}/tags/${encodeURIComponent(tag)}` ); if (onOff) options.method = "POST"; else options.method = "DELETE"; let _http = options.protocol === "https:" ? https : http; _http .request(options, res => { res.resume(); if (res.statusCode === 200) resolve(); else reject(new Error(`Unexpected status code ${res.statusCode}`)); }) .end(); }) ) ); } exports.query = query; exports.count = count; exports.postTasks = postTasks; exports.deleteResource = deleteResource; exports.updateTags = updateTags; exports.filterToMongoQuery = filterToMongoQuery;