Improvements in collection storage. Breaks compatibility with previous versions!

This commit is contained in:
Fabian Stamm 2019-11-04 17:52:18 +01:00
parent 9f944549b6
commit aa2a846031
8 changed files with 131 additions and 56 deletions

15
package-lock.json generated
View File

@ -195,6 +195,15 @@
"integrity": "sha512-FwI9gX75FgVBJ7ywgnq/P7tw+/o1GUbtP0KzbtusLigAOgIgNISRK0ZPl4qertvXSIE8YbsVJueQ90cDt9YYyw==", "integrity": "sha512-FwI9gX75FgVBJ7ywgnq/P7tw+/o1GUbtP0KzbtusLigAOgIgNISRK0ZPl4qertvXSIE8YbsVJueQ90cDt9YYyw==",
"dev": true "dev": true
}, },
"@types/nanoid": {
"version": "2.1.0",
"resolved": "https://registry.npmjs.org/@types/nanoid/-/nanoid-2.1.0.tgz",
"integrity": "sha512-xdkn/oRTA0GSNPLIKZgHWqDTWZsVrieKomxJBOQUK9YDD+zfSgmwD5t4WJYra5S7XyhTw7tfvwznW+pFexaepQ==",
"dev": true,
"requires": {
"@types/node": "*"
}
},
"@types/node": { "@types/node": {
"version": "12.12.5", "version": "12.12.5",
"resolved": "https://registry.npmjs.org/@types/node/-/node-12.12.5.tgz", "resolved": "https://registry.npmjs.org/@types/node/-/node-12.12.5.tgz",
@ -2645,9 +2654,9 @@
"optional": true "optional": true
}, },
"nanoid": { "nanoid": {
"version": "2.1.1", "version": "2.1.6",
"resolved": "https://registry.npmjs.org/nanoid/-/nanoid-2.1.1.tgz", "resolved": "https://registry.npmjs.org/nanoid/-/nanoid-2.1.6.tgz",
"integrity": "sha512-0YbJdaL4JFoejIOoawgLcYValFGJ2iyUuVDIWL3g8Es87SSOWFbWdRUMV3VMSiyPs3SQ3QxCIxFX00q5DLkMCw==" "integrity": "sha512-2NDzpiuEy3+H0AVtdt8LoFi7PnqkOnIzYmJQp7xsEU6VexLluHQwKREuiz57XaQC5006seIadPrIZJhyS2n7aw=="
}, },
"nanomatch": { "nanomatch": {
"version": "1.2.13", "version": "1.2.13",

View File

@ -21,8 +21,8 @@
"@types/koa-router": "^7.0.42", "@types/koa-router": "^7.0.42",
"@types/leveldown": "^4.0.1", "@types/leveldown": "^4.0.1",
"@types/levelup": "^3.1.1", "@types/levelup": "^3.1.1",
"@types/nanoid": "^2.1.0",
"@types/node": "^12.12.5", "@types/node": "^12.12.5",
"@types/shortid": "0.0.29",
"@types/ws": "^6.0.3", "@types/ws": "^6.0.3",
"concurrently": "^5.0.0", "concurrently": "^5.0.0",
"nodemon": "^1.19.4", "nodemon": "^1.19.4",
@ -40,7 +40,7 @@
"koa-router": "^7.4.0", "koa-router": "^7.4.0",
"leveldown": "^5.4.1", "leveldown": "^5.4.1",
"levelup": "^4.3.2", "levelup": "^4.3.2",
"shortid": "^2.2.15", "nanoid": "^2.1.6",
"what-the-pack": "^2.0.3", "what-the-pack": "^2.0.3",
"ws": "^7.2.0" "ws": "^7.2.0"
} }

View File

@ -4,7 +4,7 @@ import { DatabaseManager } from "./database/database";
import Logging from "@hibas123/logging"; import Logging from "@hibas123/logging";
import Query from "./database/query"; import Query from "./database/query";
import Session from "./database/session"; import Session from "./database/session";
import shortid = require("shortid"); import nanoid = require("nanoid");
import * as JWT from "jsonwebtoken"; import * as JWT from "jsonwebtoken";
@ -51,7 +51,7 @@ export class ConnectionManager {
const sendError = (error: string) => socket.send(JSON.stringify({ ns: "error_msg", data: error })); const sendError = (error: string) => socket.send(JSON.stringify({ ns: "error_msg", data: error }));
const session = new Session(shortid()); const session = new Session(nanoid());
let query = new URLSearchParams(req.url.split("?").pop()); let query = new URLSearchParams(req.url.split("?").pop());
@ -104,6 +104,16 @@ export class ConnectionManager {
const noperm = new Error("No permisison!"); const noperm = new Error("No permisison!");
if (session.root) {
queryHandler.set("collections", (query, perm, data) => {
return query.getCollections();
})
queryHandler.set("delete-collection", (query, perm, data) => {
return query.deleteCollection();
})
}
queryHandler.set("keys", (query, perm, data) => { queryHandler.set("keys", (query, perm, data) => {
if (!perm.read) if (!perm.read)
throw noperm; throw noperm;
@ -128,6 +138,12 @@ export class ConnectionManager {
return query.update(data); return query.update(data);
}) })
queryHandler.set("push", (query, perm, data) => {
if (!perm.write)
throw noperm;
return query.push(data);
})
queryHandler.set("delete", (query, perm, data) => { queryHandler.set("delete", (query, perm, data) => {
if (!perm.write) if (!perm.write)
throw noperm; throw noperm;
@ -138,7 +154,7 @@ export class ConnectionManager {
if (!perm.read) if (!perm.read)
throw noperm; throw noperm;
let subscriptionID = shortid.generate(); let subscriptionID = nanoid();
query.subscribe((data) => { query.subscribe((data) => {
socket.send(JSON.stringify({ ns: "event", data: { id: subscriptionID, data } })); socket.send(JSON.stringify({ ns: "event", data: { id: subscriptionID, data } }));
@ -156,8 +172,6 @@ export class ConnectionManager {
//TODO: Handle case with no id, type, path //TODO: Handle case with no id, type, path
Logging.debug(`Request with id '${id}' from type '${type}' and path '${path}' with data`, data) Logging.debug(`Request with id '${id}' from type '${type}' and path '${path}' with data`, data)
try { try {
if (!db) if (!db)
throw new Error("Database not found!"); throw new Error("Database not found!");
@ -170,7 +184,7 @@ export class ConnectionManager {
const perms = db.rules.hasPermission(path, session); const perms = db.rules.hasPermission(path, session);
let res = await handler(query, perms, data); let res = await handler(query, perms, data);
if (res[StoreSym] !== undefined) { if (res && typeof res === "object" && res[StoreSym] !== undefined) {
if (res[StoreSym]) if (res[StoreSym])
stored.set(id, query); stored.set(id, query);
else else
@ -180,7 +194,8 @@ export class ConnectionManager {
} }
} }
} catch (err) { } catch (err) {
Logging.error(err); // Logging.error(err);
Logging.debug("Sending error:", err);
answer(id, err.message, true); answer(id, err.message, true);
} }
}) })

View File

@ -54,12 +54,17 @@ export class Database {
private level = getLevelDB(this.name); private level = getLevelDB(this.name);
get data() { get data() {
return this.level; return this.level.data;
}
get collections() {
return this.level.collection;
} }
public rules: Rules; public rules: Rules;
public locks = new DocumentLock() public locks = new DocumentLock()
public collectionLocks = new DocumentLock()
public changes = new Map<string, Set<(change: Change) => void>>(); public changes = new Map<string, Set<(change: Change) => void>>();
@ -89,8 +94,8 @@ export class Database {
} }
async setRootKey(key: string) { async setRootKey(key: string) {
await Settings.setDatabaseAccessKey(this.name, key); await Settings.setDatabaseRootKey(this.name, key);
this.accesskey = key; this.rootkey = key;
} }
async setPublicKey(key: string) { async setPublicKey(key: string) {

View File

@ -3,6 +3,10 @@ export type Release = { release: () => void };
export default class DocumentLock { export default class DocumentLock {
private locks = new Map<string, (() => void)[]>(); private locks = new Map<string, (() => void)[]>();
getLocks() {
return Array.from(this.locks.keys())
}
async lock(collection: string = "", document: string = "") { async lock(collection: string = "", document: string = "") {
let key = collection + "/" + document; let key = collection + "/" + document;
let l = this.locks.get(key); let l = this.locks.get(key);

View File

@ -1,11 +1,13 @@
import { Database, Change, ChangeTypes } from "./database"; import { Database, Change, ChangeTypes } from "./database";
import { resNull } from "../storage"; import { resNull } from "../storage";
import shortid = require("shortid"); import nanoid = require("nanoid/generate");
import Logging from "@hibas123/nodelogging"; import Logging from "@hibas123/nodelogging";
import * as MSGPack from "what-the-pack"; import * as MSGPack from "what-the-pack";
export const MP = MSGPack.initialize(2 ** 20); export const MP = MSGPack.initialize(2 ** 20);
const ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
const { encode, decode } = MP; const { encode, decode } = MP;
interface ISubscribeOptions { interface ISubscribeOptions {
@ -33,39 +35,28 @@ export default class Query {
} }
private async resolve(path: string[], create = false): Promise<{ collection: string, document: string }> { private async resolve(path: string[], create = false): Promise<{ collection: string, document: string, collectionKey: string }> {
path = [...path]; // Create modifiable copy path = [...path]; // Create modifiable copy
let collectionID: string = undefined; let collectionID: string = undefined;
let documentKey = undefined; let documentKey = path.length % 2 === 0 ? path.pop() : undefined;
while (path.length > 0) { let key = path.join("/");
let collectionName = path.shift();
let key = `${collectionID || ""}/${documentKey || ""}/${collectionName}`;
let lock = await this.database.locks.lock(collectionID, documentKey);
try {
collectionID = await this.database.data.get(key).then(r => r.toString()).catch(resNull);
if (!collectionID) { const lock = await this.database.collectionLocks.lock(key);
if (create) {
collectionID = shortid.generate();
await this.database.data.put(key, Buffer.from(collectionID));
} else {
return { collection: null, document: null };
}
}
try {
if (path.length > 0) collectionID = await this.database.collections.get(key).then(r => r.toString()).catch(resNull);
documentKey = path.shift(); if (!collectionID && create) {
else collectionID = nanoid(ALPHABET, 32);
documentKey = undefined; await this.database.collections.put(key, collectionID);
} finally {
lock();
} }
} finally {
lock();
} }
return { return {
collection: collectionID, collection: collectionID,
document: documentKey document: documentKey,
collectionKey: key
}; };
} }
@ -123,12 +114,12 @@ export default class Query {
let keys = []; let keys = [];
const stream = this.database.data.createKeyStream({ const stream = this.database.data.createKeyStream({
gt, gt,
lt lt,
keyAsBuffer: false
}) })
stream.on("data", (key: string | Buffer) => { stream.on("data", (key: string) => {
key = key.toString(); let s = key.split("/", 2);
let s = key.split("/"); if (s.length > 1)
if (s.length === 2)
keys.push(s[1]); keys.push(s[1]);
}); });
stream.on("end", () => yes(keys)); stream.on("end", () => yes(keys));
@ -159,7 +150,7 @@ export default class Query {
} }
public async push(value: any) { public async push(value: any) {
let id = shortid.generate(); let id = nanoid(ALPHABET, 32);
let q = new Query(this.database, [...this.path, id], this.sender); let q = new Query(this.database, [...this.path, id], this.sender);
await q.set(value, {}); await q.set(value, {});
return id; return id;
@ -189,7 +180,43 @@ export default class Query {
//TODO: Implement //TODO: Implement
} }
subscription: { public async getCollections() {
return new Promise<string[]>((yes, no) => {
let keys = [];
const stream = this.database.data.createKeyStream({ keyAsBuffer: false })
stream.on("data", (key: string) => keys.push(key.split("/")));
stream.on("end", () => yes(keys));
stream.on("error", no);
});
}
public async deleteCollection() {
const { collection, document, collectionKey } = await this.resolve(this.path);
if (document) {
throw new Error("There can be no document defined on this operation");
}
let batch = this.database.data.batch();
try {
if (collection) {
let keys = await this.keys();
for (let key in keys) {
batch.del(key);
}
await batch.write();
batch = undefined;
await this.database.collections.del(collectionKey);
}
} finally {
if (batch)
batch.clear();
}
}
private subscription: {
key: string; key: string;
type: Set<ChangeTypes>; type: Set<ChangeTypes>;
send: (data: Omit<Change, "sender">) => void; send: (data: Omit<Change, "sender">) => void;
@ -213,6 +240,8 @@ export default class Query {
type: new Set(type || []) type: new Set(type || [])
}; };
Logging.debug("Existing?", options.existing)
if (options.existing) { if (options.existing) {
if (document) { if (document) {
send({ send({

View File

@ -11,7 +11,7 @@ interface IDatabaseConfig {
class SettingComponent { class SettingComponent {
db = getLevelDB("_server"); db = getLevelDB("_server").data;
databaseLock = new Lock(); databaseLock = new Lock();
constructor() { } constructor() { }

View File

@ -9,8 +9,9 @@ import LevelDown, { LevelDown as LD } from "leveldown";
import { AbstractIterator } from "abstract-leveldown"; import { AbstractIterator } from "abstract-leveldown";
export type LevelDB = LU<LD, AbstractIterator<any, any>>; export type LevelDB = LU<LD, AbstractIterator<any, any>>;
export type DBSet = { data: LevelDB, collection: LevelDB };
const databases = new Map<string, LevelDB>(); const databases = new Map<string, DBSet>();
export function resNull(err): null { export function resNull(err): null {
if (!err.notFound) if (!err.notFound)
@ -38,19 +39,31 @@ export async function deleteLevelDB(name: string) {
return; return;
let db = databases.get(name); let db = databases.get(name);
if (db && !db.isClosed()) if (db) {
await db.close() if (db.data.isOpen())
await db.data.close()
if (db.collection.isOpen())
await db.collection.close()
}
//TODO make sure, that name doesn't make it possible to delete all databases :) //TODO make sure, that name doesn't make it possible to delete all databases :)
await rmRecursice("./databases/" + name); await rmRecursice("./databases/" + name);
} }
export default function getLevelDB(name: string): LevelDB { export default function getLevelDB(name: string): DBSet {
let db = databases.get(name); let db = databases.get(name);
if (!db || db.isClosed()) { if (!db) {
db = LevelUp(LevelDown("./databases/" + name)); if (!fs.existsSync("./databases/" + name)) {
databases.set(name, db); fs.mkdirSync("./databases/" + name);
}
} }
db = {
data: db && db.data.isOpen() ? db.data : LevelUp(LevelDown("./databases/" + name + "/data")),
collection: db && db.collection.isOpen() ? db.collection : LevelUp(LevelDown("./databases/" + name + "/collection"))
}
databases.set(name, db);
return db; return db;
} }