import { Database } from "./database"; import Encoder, { DataTypes } from "@hibas123/binary-encoder"; import { resNull } from "../storage"; import { Bytes } from "leveldown"; import { LevelUpChain } from "levelup"; import shortid = require("shortid"); import Logging from "@hibas123/nodelogging"; enum FieldTypes { OBJECT, VALUE } interface IField { type: FieldTypes; // fields?: string[]; value?: any } const FieldEncoder = new Encoder({ type: { index: 1, type: DataTypes.UINT8 }, // fields: { // index: 2, // type: DataTypes.STRING, // array: true // }, value: { index: 3, type: DataTypes.AUTO, allowNull: true } }) export default class Query { constructor(private database: Database, private path: string[]) { if (path.length > 10) { throw new Error("Path is to long. Path is only allowed to be 10 Layers deep!"); } if (path.find(segment => segment.indexOf("/") >= 0)) { throw new Error("Path cannot contain '/'!"); } } private pathToKey(path?: string[]) { return "/" + (path || this.path).join("/"); } private getField(path: string[]): Promise { return this.database.level.get(this.pathToKey(path), { asBuffer: true }).then((res: Buffer) => FieldEncoder.decode(res)).catch(resNull); } private getFields(path: string[]) { let p = this.pathToKey(path); if (!p.endsWith("/")) p += "/"; let t = Buffer.from(p); let gt = Buffer.alloc(t.length + 1); gt.set(t); gt[t.length] = 0; let lt = Buffer.alloc(t.length + 1); lt.set(t); lt[t.length] = 0xFF; return new Promise((yes, no) => { let keys = []; const stream = this.database.level.createKeyStream({ gt: Buffer.from(p), lt: Buffer.from(lt) }) stream.on("data", key => keys.push(key.toString())); stream.on("end", () => yes(keys)); stream.on("error", no); }); } async get() { const lock = await this.database.locks.lock(this.path); try { const getData = async (path: string[]) => { let obj = await this.getField(path); if (!obj) return null; else { if (obj.type === FieldTypes.VALUE) { return obj.value; } else { let res = {}; let fields = await this.getFields(this.path); let a = fields.map(field => field.split("/").filter(e => e !== "")).sort((a, b) => a.length - b.length).map(async path => { let field = await this.getField(path); Logging.debug("Path:", path, "Field:", field); let shortened = path.slice(this.path.length); let t = res; for (let section of shortened.slice(0, -1)) { t = t[section]; } if (field.type === FieldTypes.OBJECT) { t[path[path.length - 1]] = {}; } else { t[path[path.length - 1]] = field.value; } }) await Promise.all(a); return res; } } } return await getData(this.path); } finally { lock(); } } async push(value: any) { let id = shortid.generate(); let q = new Query(this.database, [...this.path, id]); await q.set(value); return id; } async set(value: any) { const lock = await this.database.locks.lock(this.path); let batch = this.database.level.batch(); try { let field = await this.getField(this.path); if (field) { await this.delete(batch); } else { for (let i = 0; i < this.path.length; i++) { let subpath = this.path.slice(0, i); let field = await this.getField(subpath); if (!field) { batch.put(this.pathToKey(subpath), FieldEncoder.encode({ type: FieldTypes.OBJECT })); } else if (field.type !== FieldTypes.OBJECT) { throw new Error("Parent elements not all Object. Cannot set value!"); } } } const saveValue = (path: string[], value: any) => { Logging.debug("Save Value:", path, value); if (typeof value === "object") { //TODO: Handle case array! // Field type array? batch.put(this.pathToKey(path), FieldEncoder.encode({ type: FieldTypes.OBJECT })) for (let field in value) { saveValue([...path, field], value[field]); } } else { batch.put(this.pathToKey(path), FieldEncoder.encode({ type: FieldTypes.VALUE, value })); } } saveValue(this.path, value); await batch.write(); } catch (err) { if (batch.length > 0) batch.clear(); throw err; } finally { lock(); } } async delete(batch?: LevelUpChain) { let lock = batch ? undefined : await this.database.locks.lock(this.path); const commit = batch ? false : true; if (!batch) batch = this.database.level.batch(); try { let field = await this.getField(this.path); if (field) { let fields = await this.getFields(this.path) fields.forEach(field => batch.del(field)); batch.del(this.pathToKey(this.path)); } if (commit) await batch.write(); } catch (err) { if (batch.length > 0) batch.clear() } finally { if (lock) lock() } } async subscribe() { } async unsubscribe() { } }