Update MongoDB driver
This commit is contained in:
1 parent
cc885e7d04
commit
f4692faed6
12 files changed
+132
-156
No files matched your search
+3
-3
@@ -78,9 +78,9 @@ export async function acquireLock(
|
||||
value: token,
|
||||
expire: new Date(now + ttl + CLOCK_SKEW_TOLERANCE),
|
||||
},
|
||||
$currentDate: { timestamp: true },
|
||||
$currentDate: { timestamp: (true as unknown) as Date },
|
||||
},
|
||||
{ upsert: true, returnOriginal: false }
|
||||
{ upsert: true, returnDocument: "after" }
|
||||
);
|
||||
const v = r.value;
|
||||
if (Math.abs(v["timestamp"].getTime() - now) > CLOCK_SKEW_TOLERANCE)
|
||||
@@ -114,5 +114,5 @@ export async function releaseLock(
|
||||
_id: lockName,
|
||||
value: token,
|
||||
});
|
||||
if (res["result"]["n"] !== 1) throw new Error("Lock expired");
|
||||
if (res.deletedCount !== 1) throw new Error("Lock expired");
|
||||
}
|
||||
@@ -17,7 +17,7 @@
|
||||
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import { MongoClient, ObjectID, Collection, Db } from "mongodb";
|
||||
import { MongoClient, ObjectId, Collection, Db } from "mongodb";
|
||||
import { get } from "./config";
|
||||
import { decodeTag, encodeTag, escapeRegExp } from "./common";
|
||||
import { parse } from "./common/expression-parser";
|
||||
@@ -61,10 +61,7 @@ export function onConnect(callback: (db: Db) => Promise<void>): void {
|
||||
}
|
||||
|
||||
export async function connect(): Promise<void> {
|
||||
clientPromise = MongoClient.connect("" + get("MONGODB_CONNECTION_URL"), {
|
||||
useNewUrlParser: true,
|
||||
useUnifiedTopology: true,
|
||||
});
|
||||
clientPromise = MongoClient.connect("" + get("MONGODB_CONNECTION_URL"));
|
||||
|
||||
const client = await clientPromise;
|
||||
const db = client.db();
|
||||
@@ -623,7 +620,7 @@ export async function saveDevice(
|
||||
upsert: isNew,
|
||||
});
|
||||
|
||||
if (result.result.n !== 1)
|
||||
if (!result.matchedCount && !result.upsertedCount)
|
||||
throw new Error(`Device ${deviceId} not found in database`);
|
||||
|
||||
if (update2) {
|
||||
@@ -647,7 +644,7 @@ export async function getFaults(
|
||||
delete r.device;
|
||||
r.timestamp = +r.timestamp;
|
||||
r.provisions = JSON.parse(r.provisions);
|
||||
faults[channel] = r;
|
||||
faults[channel] = r as SessionFault;
|
||||
}
|
||||
|
||||
return faults;
|
||||
@@ -688,13 +685,13 @@ export async function getDueTasks(
|
||||
if (task.timestamp >= timestamp) return [tasks, +task.timestamp];
|
||||
task._id = String(task._id);
|
||||
|
||||
tasks.push(task);
|
||||
tasks.push(task as Task);
|
||||
|
||||
// For API compatibility
|
||||
if (task.name === "download" && task.file) {
|
||||
let q;
|
||||
if (ObjectID.isValid(task.file))
|
||||
q = { _id: { $in: [task.file, new ObjectID(task.file)] } };
|
||||
if (ObjectId.isValid(task.file))
|
||||
q = { _id: { $in: [task.file, new ObjectId(task.file)] } };
|
||||
else q = { _id: task.file };
|
||||
|
||||
const res = await filesCollection.find(q).toArray();
|
||||
@@ -715,7 +712,7 @@ export async function clearTasks(
|
||||
taskIds: string[]
|
||||
): Promise<void> {
|
||||
await tasksCollection.deleteMany({
|
||||
_id: { $in: taskIds.map((id) => new ObjectID(id)) },
|
||||
_id: { $in: taskIds.map((id) => new ObjectId(id)) },
|
||||
});
|
||||
}
|
||||
|
||||
@@ -732,14 +729,14 @@ export async function getOperations(
|
||||
delete r._id;
|
||||
// Workaround for a bug in v1.2.1 where operation object is saved without deserialization
|
||||
if (typeof r.provisions !== "string") {
|
||||
operations[commandKey] = r;
|
||||
operations[commandKey] = r as Operation;
|
||||
continue;
|
||||
}
|
||||
r.timestamp = +r.timestamp;
|
||||
if (r.args) r.args = JSON.parse(r.args);
|
||||
r.provisions = JSON.parse(r.provisions);
|
||||
r.retries = JSON.parse(r.retries);
|
||||
operations[commandKey] = r;
|
||||
operations[commandKey] = r as Operation;
|
||||
}
|
||||
return operations;
|
||||
}
|
||||
@@ -807,7 +804,7 @@ interface Permission {
|
||||
}
|
||||
|
||||
export async function getPermissions(): Promise<Permission[]> {
|
||||
return permissionsCollection.find().toArray();
|
||||
return (permissionsCollection.find().toArray() as unknown) as Permission[];
|
||||
}
|
||||
|
||||
interface User {
|
||||
@@ -818,5 +815,5 @@ interface User {
|
||||
}
|
||||
|
||||
export async function getUsers(): Promise<User[]> {
|
||||
return usersCollection.find().toArray();
|
||||
return (usersCollection.find().toArray() as unknown) as User[];
|
||||
}
|
||||
@@ -35,7 +35,7 @@ onConnect(async (db) => {
|
||||
});
|
||||
|
||||
const getFile = memoize(
|
||||
(md5: string, size: number, filename: string): Promise<Buffer> => {
|
||||
(uploadDate: number, size: number, filename: string): Promise<Buffer> => {
|
||||
return new Promise((resolve, reject) => {
|
||||
const buffer = Buffer.allocUnsafe(size);
|
||||
let i = 0;
|
||||
@@ -82,10 +82,14 @@ export async function listener(
|
||||
return;
|
||||
}
|
||||
|
||||
const buffer = await getFile(file.md5, file.length, filename);
|
||||
const buffer = await getFile(
|
||||
file["uploadDate"].getTime(),
|
||||
file.length,
|
||||
filename
|
||||
);
|
||||
|
||||
response.writeHead(200, {
|
||||
"Content-Type": file.contentType || "application/octet-stream",
|
||||
"Content-Type": "application/octet-stream",
|
||||
"Content-Length": file.length,
|
||||
});
|
||||
|
||||
|
||||
@@ -325,8 +325,6 @@ async function fetchFiles(): Promise<Files> {
|
||||
const id = r["filename"] || r["_id"].toString();
|
||||
files[id] = {};
|
||||
files[id].length = r["length"];
|
||||
files[id].md5 = r["md5"];
|
||||
files[id].contentType = r["contentType"];
|
||||
}
|
||||
|
||||
return files;
|
||||
|
||||
@@ -17,7 +17,7 @@
|
||||
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import { ObjectID } from "mongodb";
|
||||
import { ObjectId } from "mongodb";
|
||||
import { map, parse, stringify, parseList } from "./common/expression-parser";
|
||||
import { likePatternToRegExp } from "./common/expression";
|
||||
import { Expression, Fault, Task } from "./types";
|
||||
@@ -90,7 +90,7 @@ export function processTasksFilter(filter: Expression): Expression {
|
||||
if (!isArray(exp)) return exp;
|
||||
if (["=", "<>", ">", ">=", "<", "<="].includes(exp[0])) {
|
||||
const e = exp.slice();
|
||||
if (e[1][0] === "PARAM" && e[1][1] === "_id") e[2] = new ObjectID(e[2]);
|
||||
if (e[1][0] === "PARAM" && e[1][1] === "_id") e[2] = new ObjectId(e[2]);
|
||||
else if (e[1][0] === "PARAM" && e[1][1] === "timestamp")
|
||||
e[2] = new Date(e[2]);
|
||||
else if (e[1][0] === "PARAM" && e[1][1] === "expiry")
|
||||
|
||||
+9
-9
@@ -45,7 +45,7 @@ const PROVISIONS_REGEX = /^\/provisions\/([a-zA-Z0-9\-_%]+)\/?$/;
|
||||
const VIRTUAL_PARAMETERS_REGEX = /^\/virtual_parameters\/([a-zA-Z0-9\-_%]+)\/?$/;
|
||||
const FAULTS_REGEX = /^\/faults\/([a-zA-Z0-9\-_%:]+)\/?$/;
|
||||
|
||||
const collections = {
|
||||
const collections: Record<string, Collection> = {
|
||||
tasks: null as Collection,
|
||||
devices: null as Collection,
|
||||
presets: null as Collection,
|
||||
@@ -662,7 +662,7 @@ export function listener(
|
||||
};
|
||||
filesBucket.delete((filename as unknown) as ObjectId, () => {
|
||||
const uploadStream = filesBucket.openUploadStreamWithId(
|
||||
filename,
|
||||
(filename as unknown) as ObjectId,
|
||||
filename,
|
||||
{
|
||||
metadata: metadata,
|
||||
@@ -813,13 +813,7 @@ export function listener(
|
||||
cur.sort(sort);
|
||||
}
|
||||
|
||||
if (urlParts.query.skip)
|
||||
cur.skip(parseInt(urlParts.query.skip as string));
|
||||
|
||||
if (urlParts.query.limit)
|
||||
cur.limit(parseInt(urlParts.query.limit as string));
|
||||
|
||||
cur.count(false, (err, total) => {
|
||||
cur.count((err, total) => {
|
||||
if (err) return void throwError(err);
|
||||
|
||||
response.writeHead(200, {
|
||||
@@ -832,6 +826,12 @@ export function listener(
|
||||
return;
|
||||
}
|
||||
|
||||
if (urlParts.query.skip)
|
||||
cur.skip(parseInt(urlParts.query.skip as string));
|
||||
|
||||
if (urlParts.query.limit)
|
||||
cur.limit(parseInt(urlParts.query.limit as string));
|
||||
|
||||
response.write("[\n");
|
||||
i = 0;
|
||||
cur.forEach(
|
||||
|
||||
+1
-1
@@ -477,7 +477,7 @@ export interface VirtualParameters {
|
||||
}
|
||||
|
||||
export interface Files {
|
||||
[name: string]: { length: number; md5: string; contentType: string };
|
||||
[name: string]: { length: number };
|
||||
}
|
||||
|
||||
export interface Users {
|
||||
|
||||
@@ -17,7 +17,7 @@
|
||||
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import { ObjectID } from "mongodb";
|
||||
import { ObjectId } from "mongodb";
|
||||
import * as db from "./db";
|
||||
import { del } from "../cache";
|
||||
import { getUsers } from "../local-cache";
|
||||
@@ -38,7 +38,7 @@ async function deleteFault(id): Promise<void> {
|
||||
await Promise.all([
|
||||
db.deleteFault(id),
|
||||
channel.startsWith("task_")
|
||||
? db.deleteTask(new ObjectID(channel.slice(5)))
|
||||
? db.deleteTask(new ObjectId(channel.slice(5)))
|
||||
: null,
|
||||
]);
|
||||
|
||||
@@ -108,7 +108,7 @@ export async function postTasks(
|
||||
const lastInform = Date.now();
|
||||
const notInSession = await awaitSessionEnd(deviceId, 30000);
|
||||
if (!notInSession) {
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectId(t._id))));
|
||||
return {
|
||||
connectionRequest: "Session took too long to complete",
|
||||
tasks: statuses,
|
||||
@@ -120,7 +120,7 @@ export async function postTasks(
|
||||
const status = await connectionRequest(deviceId, device);
|
||||
|
||||
if (status) {
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectId(t._id))));
|
||||
return {
|
||||
connectionRequest: status,
|
||||
tasks: statuses,
|
||||
@@ -129,7 +129,7 @@ export async function postTasks(
|
||||
|
||||
const sessionStarted = await awaitSessionStart(deviceId, lastInform, timeout);
|
||||
if (!sessionStarted) {
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectId(t._id))));
|
||||
return {
|
||||
connectionRequest: "No contact from CPE",
|
||||
tasks: statuses,
|
||||
@@ -138,7 +138,7 @@ export async function postTasks(
|
||||
|
||||
const sessionEnded = await awaitSessionEnd(deviceId, 120000);
|
||||
if (!sessionEnded) {
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectId(t._id))));
|
||||
return {
|
||||
connectionRequest: "Session took too long to complete",
|
||||
tasks: statuses,
|
||||
@@ -163,7 +163,7 @@ export async function postTasks(
|
||||
}
|
||||
}
|
||||
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
|
||||
await Promise.all(statuses.map((t) => db.deleteTask(new ObjectId(t._id))));
|
||||
|
||||
return { connectionRequest: "OK", tasks: statuses };
|
||||
}
|
||||
|
||||
+12
-8
@@ -17,13 +17,13 @@
|
||||
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import { Db, GridFSBucket, ObjectID } from "mongodb";
|
||||
import { Db, GridFSBucket, ObjectId } from "mongodb";
|
||||
import { Script } from "vm";
|
||||
import { onConnect } from "../db";
|
||||
import * as mongodbFunctions from "../mongodb-functions";
|
||||
import * as expression from "../common/expression";
|
||||
import { QueryOptions, Expression } from "../types";
|
||||
import { Readable } from "stream";
|
||||
import { Readable, Writable } from "stream";
|
||||
import { minimize } from "../common/boolean-expression";
|
||||
|
||||
const RESOURCE_COLLECTION = {
|
||||
@@ -208,7 +208,7 @@ function putResource(resource, id, object): Promise<void> {
|
||||
|
||||
function deleteResource(
|
||||
resource: string,
|
||||
id: string | ObjectID
|
||||
id: string | ObjectId
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
|
||||
@@ -338,13 +338,17 @@ export function putFile(
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const bucket = new GridFSBucket(db);
|
||||
const uploadStream = bucket.openUploadStreamWithId(filename, filename, {
|
||||
metadata: metadata,
|
||||
});
|
||||
const uploadStream = bucket.openUploadStreamWithId(
|
||||
(filename as unknown) as ObjectId,
|
||||
filename,
|
||||
{
|
||||
metadata: metadata,
|
||||
}
|
||||
);
|
||||
uploadStream.on("error", reject);
|
||||
contentStream.on("error", reject);
|
||||
uploadStream.on("finish", resolve);
|
||||
contentStream.pipe(uploadStream);
|
||||
contentStream.pipe(uploadStream as Writable);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -362,6 +366,6 @@ export function deleteFault(id: string): Promise<void> {
|
||||
return deleteResource("faults", id);
|
||||
}
|
||||
|
||||
export function deleteTask(id: ObjectID): Promise<void> {
|
||||
export function deleteTask(id: ObjectId): Promise<void> {
|
||||
return deleteResource("tasks", id);
|
||||
}
|
||||
Reference in new issue
Block a user