Compare commits
14 Commits
v2.0.0-bet
...
v2.0.0-bet
Author | SHA1 | Date | |
---|---|---|---|
1434036b42 | |||
88b0cb68d8 | |||
904b986e22 | |||
2ac9def153 | |||
d2621fdd3c | |||
4cee0048f5 | |||
3432ea5e23 | |||
0175822699 | |||
72e5c0dedd | |||
d48bf46231 | |||
e287890ca1 | |||
3e0dc06521 | |||
50268d05c5 | |||
10f3b4fa50 |
21
.drone.yml
Normal file
21
.drone.yml
Normal file
@ -0,0 +1,21 @@
|
||||
kind: pipeline
|
||||
type: docker
|
||||
name: default
|
||||
|
||||
steps:
|
||||
- name: Build with node
|
||||
image: node:12
|
||||
commands:
|
||||
- npm install
|
||||
- npm run build
|
||||
- name: Publish to docker
|
||||
image: plugins/docker
|
||||
settings:
|
||||
username:
|
||||
from_secret: docker_username
|
||||
password:
|
||||
from_secret: docker_password
|
||||
auto_tag: true
|
||||
repo: hibas123.azurecr.io/realtimedb
|
||||
registry: hibas123.azurecr.io
|
||||
debug: true
|
@ -1,3 +1,5 @@
|
||||
[*]
|
||||
charset = utf-8
|
||||
indent_size = 3
|
||||
indent_style = space
|
||||
insert_final_newline = true
|
2444
package-lock.json
generated
2444
package-lock.json
generated
File diff suppressed because it is too large
Load Diff
36
package.json
36
package.json
@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@hibas123/realtimedb",
|
||||
"version": "2.0.0-beta.5",
|
||||
"version": "2.0.0-beta.9",
|
||||
"description": "",
|
||||
"main": "lib/index.js",
|
||||
"private": true,
|
||||
@ -17,31 +17,31 @@
|
||||
"license": "ISC",
|
||||
"devDependencies": {
|
||||
"@types/dotenv": "^8.2.0",
|
||||
"@types/jsonwebtoken": "^8.3.5",
|
||||
"@types/koa": "^2.0.51",
|
||||
"@types/koa-router": "^7.0.42",
|
||||
"@types/leveldown": "^4.0.1",
|
||||
"@types/levelup": "^3.1.1",
|
||||
"@types/jsonwebtoken": "^8.3.8",
|
||||
"@types/koa": "^2.11.2",
|
||||
"@types/koa-router": "^7.4.0",
|
||||
"@types/leveldown": "^4.0.2",
|
||||
"@types/levelup": "^4.3.0",
|
||||
"@types/nanoid": "^2.1.0",
|
||||
"@types/node": "^12.12.5",
|
||||
"@types/ws": "^6.0.3",
|
||||
"concurrently": "^5.0.0",
|
||||
"nodemon": "^1.19.4",
|
||||
"typescript": "^3.6.4"
|
||||
"@types/node": "^13.9.3",
|
||||
"@types/ws": "^7.2.3",
|
||||
"concurrently": "^5.1.0",
|
||||
"nodemon": "^2.0.2",
|
||||
"typescript": "^3.8.3"
|
||||
},
|
||||
"dependencies": {
|
||||
"@hibas123/nodelogging": "^2.1.1",
|
||||
"@hibas123/utils": "^2.1.1",
|
||||
"@hibas123/nodelogging": "^2.1.5",
|
||||
"@hibas123/utils": "^2.2.3",
|
||||
"dotenv": "^8.2.0",
|
||||
"handlebars": "^4.5.1",
|
||||
"handlebars": "^4.7.3",
|
||||
"jsonwebtoken": "^8.5.1",
|
||||
"koa": "^2.11.0",
|
||||
"koa-body": "^4.1.1",
|
||||
"koa-router": "^7.4.0",
|
||||
"leveldown": "^5.4.1",
|
||||
"koa-router": "^8.0.8",
|
||||
"leveldown": "^5.5.1",
|
||||
"levelup": "^4.3.2",
|
||||
"nanoid": "^2.1.6",
|
||||
"nanoid": "^2.1.11",
|
||||
"what-the-pack": "^2.0.3",
|
||||
"ws": "^7.2.0"
|
||||
"ws": "^7.2.3"
|
||||
}
|
||||
}
|
@ -1,43 +1,12 @@
|
||||
import * as WebSocket from "ws";
|
||||
import { Server, IncomingMessage } from "http";
|
||||
import { DatabaseManager } from "./database/database";
|
||||
import Logging from "@hibas123/nodelogging";
|
||||
import { Query, CollectionQuery, DocumentQuery } from "./database/query";
|
||||
import { IncomingMessage, Server } from "http";
|
||||
import * as WebSocket from "ws";
|
||||
import { DatabaseManager } from "./database/database";
|
||||
import { CollectionQuery, DocumentQuery, IQuery, ITypedQuery } from "./database/query";
|
||||
import Session from "./database/session";
|
||||
import { verifyJWT } from "./helper/jwt";
|
||||
import nanoid = require("nanoid");
|
||||
|
||||
import * as JWT from "jsonwebtoken";
|
||||
|
||||
async function verifyJWT(token: string, publicKey: string) {
|
||||
return new Promise<any | undefined>((yes) => {
|
||||
JWT.verify(token, publicKey, (err, decoded) => {
|
||||
if (err)
|
||||
yes(undefined);
|
||||
else
|
||||
yes(decoded);
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
const StoreSym = Symbol("store");
|
||||
function StoreQuery(result?: any) {
|
||||
return {
|
||||
[StoreSym]: true,
|
||||
result
|
||||
}
|
||||
}
|
||||
|
||||
function DeleteQuery(result?: any) {
|
||||
return {
|
||||
[StoreSym]: false,
|
||||
result
|
||||
}
|
||||
}
|
||||
|
||||
import { URLSearchParams } from "url";
|
||||
|
||||
type QueryTypes = "keys" | "get" | "set" | "update" | "delete" | "push" | "subscribe" | "unsubscribe";
|
||||
|
||||
export class ConnectionManager {
|
||||
static server: WebSocket.Server;
|
||||
|
||||
@ -50,10 +19,9 @@ export class ConnectionManager {
|
||||
Logging.log("New Connection:");
|
||||
const sendError = (error: string) => socket.send(JSON.stringify({ ns: "error_msg", data: error }));
|
||||
|
||||
|
||||
const session = new Session(nanoid());
|
||||
|
||||
let query = new URLSearchParams(req.url.split("?").pop());
|
||||
const query = new URL(req.url, "http://localhost").searchParams;
|
||||
|
||||
const database = query.get("database");
|
||||
const db = DatabaseManager.getDatabase(database);
|
||||
@ -92,53 +60,42 @@ export class ConnectionManager {
|
||||
}
|
||||
}
|
||||
|
||||
const stored = new Map<string, Query>();
|
||||
const answer = (id: string, data: any, error: boolean = false) => {
|
||||
if (error)
|
||||
Logging.error(error as any);
|
||||
socket.send(JSON.stringify({ ns: "message", data: { id, error, data } }));
|
||||
}
|
||||
|
||||
const handler = new Map<string, ((data: any) => void)>();
|
||||
type QueryData = { id: string, type: QueryTypes, path: string[], data: any, options: any };
|
||||
|
||||
handler.set("query", async ({ id, type, path, data }: QueryData) => {
|
||||
//TODO: Handle case with no id, type, path
|
||||
Logging.debug(`Request with id '${id}' from type '${type}' and path '${path.join("/")}' with data`, data)
|
||||
handler.set("v2", async ({ id, query }) => db.run(Array.isArray(query) ? query : [query], session)
|
||||
.then(res => answer(id, res))
|
||||
.catch(err => answer(id, undefined, err))
|
||||
);
|
||||
|
||||
try {
|
||||
if (!db)
|
||||
throw new Error("Database not found!");
|
||||
else {
|
||||
let isDoc = path.length % 2 === 0;
|
||||
let handler = isDoc ? documentHandler.get(type) : collectionHandler.get(type);
|
||||
// handler.set("bulk", async ({ id, query }) => db.run(query, session)
|
||||
// .then(res => answer(id, res))
|
||||
// .catch(err => answer(id, undefined, err))
|
||||
// );
|
||||
|
||||
if (!handler && session.root)
|
||||
handler = rootHandler.get(type);
|
||||
|
||||
if (!handler)
|
||||
throw new Error("Invalid Request!");
|
||||
|
||||
let query = db.getQuery(path || [], session.sessionid, isDoc ? "document" : "collection");
|
||||
let res = await handler({
|
||||
id,
|
||||
data,
|
||||
socket,
|
||||
query: query as any // We know it is the right one
|
||||
const SnapshotMap = new Map<string, string>();
|
||||
handler.set("snapshot", async ({ id, query }: { id: string, query: ITypedQuery<"snapshot"> }) => {
|
||||
db.snapshot(query, session, (data => {
|
||||
socket.send(JSON.stringify({
|
||||
ns: "snapshot", data: { id, data }
|
||||
}));
|
||||
})).then(s => {
|
||||
answer(id, s.snaphot);
|
||||
SnapshotMap.set(id, s.id);
|
||||
}).catch(err => answer(id, undefined, err));
|
||||
})
|
||||
|
||||
if (res && typeof res === "object" && res[StoreSym] !== undefined) {
|
||||
if (res[StoreSym])
|
||||
stored.set(id, query);
|
||||
else
|
||||
stored.delete(id);
|
||||
|
||||
res = res.result;
|
||||
}
|
||||
answer(id, res);
|
||||
}
|
||||
} catch (err) {
|
||||
// Logging.error(err);
|
||||
Logging.debug("Sending error:", err);
|
||||
answer(id, err.message, true);
|
||||
handler.set("unsubscribe", async ({ id }) => {
|
||||
let i = SnapshotMap.get(id);
|
||||
if (i) {
|
||||
db.unsubscribe(i, session);
|
||||
SnapshotMap.delete(i);
|
||||
}
|
||||
})
|
||||
|
||||
@ -156,102 +113,10 @@ export class ConnectionManager {
|
||||
})
|
||||
|
||||
socket.on("close", () => {
|
||||
Logging.log(`${session.sessionid} has disconnected!`);
|
||||
Logging.debug("Clearing stored:", stored);
|
||||
stored.forEach(query => (query as DocumentQuery | CollectionQuery).unsubscribe());
|
||||
stored.clear();
|
||||
Logging.log(`${session.id} has disconnected!`);
|
||||
session.subscriptions.forEach(unsubscribe => unsubscribe());
|
||||
session.subscriptions.clear();
|
||||
socket.removeAllListeners();
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type QueryHandler<T extends Query> = (api: {
|
||||
id: string;
|
||||
query: T;
|
||||
// storedQuery(id: string): T | undefined;
|
||||
socket: WebSocket;
|
||||
data: any;
|
||||
}) => any;
|
||||
|
||||
const NoPermissionError = new Error("No permisison!");
|
||||
|
||||
const rootHandler = new Map<string, QueryHandler<Query>>();
|
||||
rootHandler.set("collections", ({ query }) => {
|
||||
return CollectionQuery.fromQuery(query).collections();
|
||||
})
|
||||
|
||||
rootHandler.set("delete-collection", ({ query }) => {
|
||||
return CollectionQuery.fromQuery(query).deleteCollection();
|
||||
})
|
||||
|
||||
|
||||
const documentHandler = new Map<string, QueryHandler<DocumentQuery>>();
|
||||
documentHandler.set("get", ({ query }) => {
|
||||
return query.get();
|
||||
})
|
||||
|
||||
documentHandler.set("set", ({ query, data }) => {
|
||||
return query.set(data, {});
|
||||
})
|
||||
|
||||
documentHandler.set("update", ({ query, data }) => {
|
||||
return query.update(data);
|
||||
})
|
||||
|
||||
documentHandler.set("delete", ({ query }) => {
|
||||
return query.delete();
|
||||
})
|
||||
|
||||
documentHandler.set("snapshot", async ({ query, data, id, socket }) => {
|
||||
let res = await query.snapshot((data) => {
|
||||
socket.send(JSON.stringify({
|
||||
ns: "snapshot", data: { id, data }
|
||||
}));
|
||||
});
|
||||
|
||||
return StoreQuery(res);
|
||||
})
|
||||
|
||||
documentHandler.set("unsubscribe", async ({ query }) => {
|
||||
query.unsubscribe();
|
||||
return DeleteQuery(true);
|
||||
})
|
||||
|
||||
|
||||
const collectionHandler = new Map<string, QueryHandler<CollectionQuery>>();
|
||||
collectionHandler.set("keys", ({ query }) => {
|
||||
return query.keys();
|
||||
})
|
||||
|
||||
|
||||
collectionHandler.set("add", ({ query, data }) => {
|
||||
return query.add(data);
|
||||
})
|
||||
|
||||
collectionHandler.set("get", ({ query, data }) => {
|
||||
if (data.where)
|
||||
query.where = data.where;
|
||||
if (data.limit)
|
||||
query.limit = data.limit;
|
||||
return query.get();
|
||||
})
|
||||
|
||||
collectionHandler.set("snapshot", async ({ query, id, socket, data }) => {
|
||||
if (data.where)
|
||||
query.where = data.where;
|
||||
if (data.limit)
|
||||
query.limit = data.limit;
|
||||
|
||||
let res = await query.snapshot((data) => {
|
||||
socket.send(JSON.stringify({
|
||||
ns: "snapshot", data: { id, data }
|
||||
}));
|
||||
});
|
||||
|
||||
return StoreQuery(res);
|
||||
})
|
||||
|
||||
collectionHandler.set("unsubscribe", async ({ query }) => {
|
||||
query.unsubscribe();
|
||||
return DeleteQuery(true);
|
||||
})
|
@ -1,9 +1,18 @@
|
||||
import { Rules } from "./rules";
|
||||
import Settings from "../settings";
|
||||
import getLevelDB, { LevelDB, deleteLevelDB } from "../storage";
|
||||
import getLevelDB, { LevelDB, deleteLevelDB, resNull } from "../storage";
|
||||
import DocumentLock from "./lock";
|
||||
import { DocumentQuery, CollectionQuery, Query } from "./query";
|
||||
import { DocumentQuery, CollectionQuery, Query, QueryError, ITypedQuery, IQuery } from "./query";
|
||||
import Logging from "@hibas123/nodelogging";
|
||||
import Session from "./session";
|
||||
import nanoid = require("nanoid/generate");
|
||||
import { Observable } from "@hibas123/utils";
|
||||
|
||||
const ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
|
||||
|
||||
// interface ITransaction {
|
||||
// queries: ITypedQuery<IWriteQueries>[];
|
||||
// }
|
||||
|
||||
export class DatabaseManager {
|
||||
static databases = new Map<string, Database>();
|
||||
@ -46,12 +55,17 @@ export type ChangeTypes = "added" | "modified" | "deleted";
|
||||
export type Change = {
|
||||
data: any;
|
||||
document: string;
|
||||
collection: string;
|
||||
type: ChangeTypes;
|
||||
sender: string;
|
||||
}
|
||||
|
||||
|
||||
export class Database {
|
||||
public static getKey(collectionid: string, documentid?: string) {
|
||||
return `${collectionid || ""}/${documentid || ""}`;
|
||||
}
|
||||
|
||||
private level = getLevelDB(this.name);
|
||||
|
||||
get data() {
|
||||
@ -64,10 +78,15 @@ export class Database {
|
||||
|
||||
|
||||
public rules: Rules;
|
||||
public locks = new DocumentLock()
|
||||
private locks = new DocumentLock()
|
||||
public collectionLocks = new DocumentLock()
|
||||
|
||||
public changes = new Map<string, Set<(change: Change) => void>>();
|
||||
public changeListener = new Map<string, Set<(change: Change[]) => void>>();
|
||||
public collectionChangeListener = new Observable<{
|
||||
key: string;
|
||||
id: string;
|
||||
type: "create" | "delete"
|
||||
}>();
|
||||
|
||||
toJSON() {
|
||||
return {
|
||||
@ -104,14 +123,201 @@ export class Database {
|
||||
this.publickey = key;
|
||||
}
|
||||
|
||||
public async resolve(path: string[], create = false): Promise<{ collection: string, document: string, collectionKey: string }> {
|
||||
path = [...path]; // Create modifiable copy
|
||||
let collectionID: string = undefined;
|
||||
let documentKey = path.length % 2 === 0 ? path.pop() : undefined;
|
||||
let key = path.join("/");
|
||||
|
||||
getQuery(path: string[], sender: string, type: "document" | "collection" | "any") {
|
||||
if (type === "document")
|
||||
return new DocumentQuery(this, path, sender);
|
||||
else if (type === "collection")
|
||||
return new CollectionQuery(this, path, sender);
|
||||
const lock = await this.collectionLocks.lock(key);
|
||||
|
||||
try {
|
||||
collectionID = await this.collections.get(key).then(r => r.toString()).catch(resNull);
|
||||
if (!collectionID && create) {
|
||||
collectionID = nanoid(ALPHABET, 32);
|
||||
await this.collections.put(key, collectionID);
|
||||
setImmediate(() => {
|
||||
this.collectionChangeListener.send({
|
||||
id: collectionID,
|
||||
key,
|
||||
type: "create"
|
||||
})
|
||||
})
|
||||
}
|
||||
} finally {
|
||||
lock();
|
||||
}
|
||||
|
||||
return {
|
||||
collection: collectionID,
|
||||
document: documentKey,
|
||||
collectionKey: key
|
||||
};
|
||||
}
|
||||
|
||||
private sendChanges(changes: Change[]) {
|
||||
let col = new Map<string, Map<string, Change[]>>();
|
||||
changes.forEach(change => {
|
||||
let e = col.get(change.collection);
|
||||
if (!e) {
|
||||
e = new Map()
|
||||
col.set(change.collection, e);
|
||||
}
|
||||
|
||||
let d = e.get(change.document);
|
||||
if (!d) {
|
||||
d = [];
|
||||
e.set(change.document, d);
|
||||
}
|
||||
|
||||
d.push(change);
|
||||
})
|
||||
|
||||
setImmediate(() => {
|
||||
for (let [collection, documents] of col.entries()) {
|
||||
let collectionChanges = [];
|
||||
for (let [document, documentChanges] of documents.entries()) {
|
||||
let s = this.changeListener.get(Database.getKey(collection, document));
|
||||
if (s)
|
||||
s.forEach(e => setImmediate(() => e(documentChanges)));
|
||||
|
||||
collectionChanges.push(...documentChanges);
|
||||
}
|
||||
let s = this.changeListener.get(Database.getKey(collection))
|
||||
if (s)
|
||||
s.forEach(e => setImmediate(() => e(collectionChanges)))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
private validate(query: ITypedQuery<any>) {
|
||||
const inv = new QueryError("Malformed query!");
|
||||
if (!query || typeof query !== "object")
|
||||
throw inv;
|
||||
|
||||
if (!query.type)
|
||||
throw inv;
|
||||
|
||||
if (!query.path)
|
||||
throw inv;
|
||||
}
|
||||
|
||||
async run(queries: IQuery[], session: Session) {
|
||||
let resolve: { path: string[], create: boolean, resolved?: [string, string, string] }[] = [];
|
||||
|
||||
const addToResolve = (path: string[], create?: boolean) => {
|
||||
let entry = resolve.find(e => { //TODO: Find may be slow...
|
||||
if (e.path.length !== path.length)
|
||||
return false;
|
||||
for (let i = 0; i < e.path.length; i++) {
|
||||
if (e.path[i] !== path[i])
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
})
|
||||
|
||||
if (!entry) {
|
||||
entry = {
|
||||
path,
|
||||
create
|
||||
}
|
||||
resolve.push(entry);
|
||||
}
|
||||
|
||||
entry.create = entry.create || create;
|
||||
|
||||
return entry;
|
||||
}
|
||||
|
||||
const isBatch = queries.length > 1;
|
||||
let parsed = queries.map(rawQuery => {
|
||||
this.validate(rawQuery);
|
||||
const isCollection = rawQuery.path.length % 2 === 1;
|
||||
|
||||
let query = isCollection
|
||||
? new CollectionQuery(this, session, rawQuery)
|
||||
: new DocumentQuery(this, session, rawQuery);
|
||||
|
||||
if (isBatch && !query.batchCompatible)
|
||||
throw new Error("There are queries that are not batch compatible!");
|
||||
|
||||
let path = addToResolve(rawQuery.path, query.createCollection);
|
||||
if (query.additionalLock)
|
||||
addToResolve(query.additionalLock);
|
||||
|
||||
return {
|
||||
path,
|
||||
query
|
||||
};
|
||||
});
|
||||
|
||||
resolve = resolve.sort((a, b) => a.path.length - b.path.length);
|
||||
|
||||
let locks: (() => void)[] = [];
|
||||
for (let e of resolve) {
|
||||
let { collection, document, collectionKey } = await this.resolve(e.path, e.create);
|
||||
e.resolved = [collection, document, collectionKey];
|
||||
|
||||
locks.push(
|
||||
await this.locks.lock(collection, document)
|
||||
);
|
||||
}
|
||||
|
||||
let result = [];
|
||||
try {
|
||||
let batch = this.data.batch();
|
||||
let changes: Change[] = [];
|
||||
for (let e of parsed) {
|
||||
result.push(
|
||||
await e.query.run(e.path.resolved[0], e.path.resolved[1], batch, e.path.resolved[2])
|
||||
);
|
||||
changes.push(...e.query.changes);
|
||||
}
|
||||
if (batch.length > 0)
|
||||
await batch.write();
|
||||
|
||||
this.sendChanges(changes);
|
||||
} finally {
|
||||
locks.forEach(lock => lock());
|
||||
}
|
||||
|
||||
if (isBatch)
|
||||
return result;
|
||||
else
|
||||
return new Query(this, path, sender);
|
||||
return result[0]
|
||||
}
|
||||
|
||||
async snapshot(rawQuery: ITypedQuery<"snapshot">, session: Session, onchange: (change: any) => void) {
|
||||
Logging.debug("Snaphot request:", rawQuery.path);
|
||||
this.validate(rawQuery);
|
||||
|
||||
if (rawQuery.type !== "snapshot")
|
||||
throw new Error("Invalid query type!");
|
||||
|
||||
const isCollection = rawQuery.path.length % 2 === 1;
|
||||
let query = isCollection
|
||||
? new CollectionQuery(this, session, rawQuery, true)
|
||||
: new DocumentQuery(this, session, rawQuery, true);
|
||||
|
||||
const {
|
||||
unsubscribe,
|
||||
value
|
||||
} = await query.snapshot(onchange);
|
||||
|
||||
const id = nanoid(ALPHABET, 16);
|
||||
session.subscriptions.set(id, unsubscribe);
|
||||
return {
|
||||
id,
|
||||
snaphot: value
|
||||
};
|
||||
}
|
||||
|
||||
async unsubscribe(id: string, session: Session) {
|
||||
let query = session.subscriptions.get(id);
|
||||
if (query) {
|
||||
query();
|
||||
session.subscriptions.delete(id);
|
||||
}
|
||||
}
|
||||
|
||||
async stop() {
|
||||
|
@ -8,6 +8,7 @@ export default class DocumentLock {
|
||||
}
|
||||
|
||||
async lock(collection: string = "", document: string = "") {
|
||||
//TODO: Check collection locks
|
||||
let key = collection + "/" + document;
|
||||
let l = this.locks.get(key);
|
||||
if (l)
|
||||
|
@ -3,91 +3,238 @@ import { resNull } from "../storage";
|
||||
import nanoid = require("nanoid/generate");
|
||||
import Logging from "@hibas123/nodelogging";
|
||||
import * as MSGPack from "what-the-pack";
|
||||
import Session from "./session";
|
||||
import { LevelUpChain } from "levelup";
|
||||
|
||||
export type IWriteQueries = "set" | "update" | "delete" | "add";
|
||||
export type ICollectionQueries =
|
||||
| "get"
|
||||
| "add"
|
||||
| "keys"
|
||||
| "delete-collection"
|
||||
| "list";
|
||||
export type IDocumentQueries = "get" | "set" | "update" | "delete";
|
||||
|
||||
export interface ITypedQuery<T> {
|
||||
path: string[];
|
||||
type: T;
|
||||
data?: any;
|
||||
options?: any;
|
||||
}
|
||||
|
||||
export type IQuery = ITypedQuery<
|
||||
ICollectionQueries | IDocumentQueries | "snapshot"
|
||||
>;
|
||||
|
||||
export const MP = MSGPack.initialize(2 ** 20);
|
||||
|
||||
const ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
|
||||
const ALPHABET =
|
||||
"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
|
||||
|
||||
const { encode, decode } = MP;
|
||||
|
||||
interface ISubscribeOptions {
|
||||
existing: boolean;
|
||||
type Runner = (
|
||||
collection: string,
|
||||
document: string,
|
||||
batch: LevelUpChain,
|
||||
collectionKey: string
|
||||
) => any;
|
||||
|
||||
interface IPreparedQuery {
|
||||
createCollection: boolean;
|
||||
needDocument: boolean;
|
||||
batchCompatible: boolean;
|
||||
runner: Runner;
|
||||
permission: "write" | "read";
|
||||
additionalLock?: string[];
|
||||
}
|
||||
|
||||
export class Query {
|
||||
interface DocRes {
|
||||
id: string;
|
||||
data: any;
|
||||
}
|
||||
|
||||
export abstract class Query {
|
||||
/**
|
||||
* Returns true if the path only contains valid characters and false if it doesn't
|
||||
* @param path Path to be checked
|
||||
*/
|
||||
private validatePath(path: string[]) {
|
||||
return path.every(e => (e.match(/[^a-zA-Z0-9_\-\<\>]/g) || []).length === 0);
|
||||
return path.every(
|
||||
e => (e.match(/[^a-zA-Z0-9_\-\<\>]/g) || []).length === 0
|
||||
);
|
||||
}
|
||||
|
||||
constructor(protected database: Database, protected path: string[], protected sender: string) {
|
||||
if (path.length > 10) {
|
||||
throw new Error("Path is to long. Path is only allowed to be 10 Layers deep!");
|
||||
public changes: Change[] = [];
|
||||
|
||||
public readonly createCollection: boolean;
|
||||
public readonly needDocument: boolean;
|
||||
public readonly batchCompatible: boolean;
|
||||
public readonly additionalLock?: string[];
|
||||
public readonly permission: string;
|
||||
private readonly _runner: Runner;
|
||||
|
||||
constructor(
|
||||
protected database: Database,
|
||||
protected session: Session,
|
||||
protected query: IQuery,
|
||||
snapshot = false
|
||||
) {
|
||||
if (query.path.length > 10) {
|
||||
throw new QueryError(
|
||||
"Path is to long. Path is only allowed to be 10 Layers deep!"
|
||||
);
|
||||
}
|
||||
if (!this.validatePath(path)) {
|
||||
throw new Error("Path can only contain a-z A-Z 0-9 '-' '-' '<' and '>' ");
|
||||
if (!this.validatePath(query.path)) {
|
||||
throw new QueryError(
|
||||
"Path can only contain a-z A-Z 0-9 '-' '-' '<' and '>' "
|
||||
);
|
||||
}
|
||||
|
||||
if (!snapshot) {
|
||||
let data = this.prepare(query);
|
||||
this.createCollection = data.createCollection;
|
||||
this.needDocument = data.needDocument;
|
||||
this.batchCompatible = data.batchCompatible;
|
||||
this.additionalLock = data.additionalLock;
|
||||
this._runner = data.runner;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
protected async resolve(path: string[], create = false): Promise<{ collection: string, document: string, collectionKey: string }> {
|
||||
path = [...path]; // Create modifiable copy
|
||||
let collectionID: string = undefined;
|
||||
let documentKey = path.length % 2 === 0 ? path.pop() : undefined;
|
||||
let key = path.join("/");
|
||||
|
||||
const lock = await this.database.collectionLocks.lock(key);
|
||||
|
||||
try {
|
||||
collectionID = await this.database.collections.get(key).then(r => r.toString()).catch(resNull);
|
||||
if (!collectionID && create) {
|
||||
collectionID = nanoid(ALPHABET, 32);
|
||||
await this.database.collections.put(key, collectionID);
|
||||
}
|
||||
} finally {
|
||||
lock();
|
||||
}
|
||||
|
||||
return {
|
||||
collection: collectionID,
|
||||
document: documentKey,
|
||||
collectionKey: key
|
||||
};
|
||||
}
|
||||
|
||||
protected getKey(collection: string, document?: string) {
|
||||
return `${collection || ""}/${document || ""}`;
|
||||
}
|
||||
protected abstract prepare(query: IQuery): IPreparedQuery;
|
||||
|
||||
protected getDoc(collection: string, document: string) {
|
||||
return this.database.data
|
||||
.get(this.getKey(collection, document), { asBuffer: true })
|
||||
.then(res => decode<any>(res as Buffer)).catch(resNull);
|
||||
.get(Database.getKey(collection, document), { asBuffer: true })
|
||||
.then(res => decode<any>(res as Buffer))
|
||||
.catch(resNull);
|
||||
}
|
||||
|
||||
protected sendChange(collection: string, document: string, type: ChangeTypes, data: any) {
|
||||
protected sendChange(
|
||||
collection: string,
|
||||
document: string,
|
||||
type: ChangeTypes,
|
||||
data: any
|
||||
) {
|
||||
let change: Change = {
|
||||
type,
|
||||
document,
|
||||
collection,
|
||||
data,
|
||||
sender: this.sender
|
||||
sender: this.session.id
|
||||
};
|
||||
|
||||
this.changes.push(change);
|
||||
}
|
||||
|
||||
let s = this.database.changes.get(this.getKey(collection, document))
|
||||
|
||||
if (s)
|
||||
s.forEach(e => setImmediate(() => e(change)))
|
||||
s = this.database.changes.get(this.getKey(collection))
|
||||
if (s)
|
||||
s.forEach(e => setImmediate(() => e(change)))
|
||||
|
||||
protected static getConstructorParams(
|
||||
query: Query
|
||||
): [Database, Session, IQuery] {
|
||||
return [query.database, query.session, query.query];
|
||||
}
|
||||
|
||||
protected static getConstructorParams(query: Query): [Database, string[], string] {
|
||||
return [query.database, query.path, query.sender];
|
||||
protected abstract checkChange(change: Change): boolean;
|
||||
protected abstract firstSend(
|
||||
collection: string,
|
||||
document: string
|
||||
): Promise<any>;
|
||||
|
||||
public run(
|
||||
collection: string,
|
||||
document: string,
|
||||
batch: LevelUpChain,
|
||||
collectionKey: string
|
||||
) {
|
||||
let perm = this.database.rules.hasPermission(
|
||||
this.query.path,
|
||||
this.session
|
||||
);
|
||||
if (this.permission === "read" && !perm.read) {
|
||||
throw new QueryError("No permission!");
|
||||
} else if (this.permission === "write" && !perm.write) {
|
||||
throw new QueryError("No permission!");
|
||||
}
|
||||
return this._runner.call(
|
||||
this,
|
||||
collection,
|
||||
document,
|
||||
batch,
|
||||
collectionKey
|
||||
);
|
||||
}
|
||||
|
||||
public async snapshot(
|
||||
onChange: (change: (DocRes & { type: ChangeTypes })[]) => void
|
||||
) {
|
||||
let perm = this.database.rules.hasPermission(
|
||||
this.query.path,
|
||||
this.session
|
||||
);
|
||||
if (this.permission === "read" && !perm.read) {
|
||||
throw new QueryError("No permission!");
|
||||
}
|
||||
|
||||
const receivedChanges = (changes: Change[]) => {
|
||||
let res = changes
|
||||
.filter(change => this.checkChange(change))
|
||||
.map(change => {
|
||||
return {
|
||||
id: change.document,
|
||||
data: change.data,
|
||||
type: change.type
|
||||
};
|
||||
});
|
||||
if (res.length > 0) onChange(res);
|
||||
};
|
||||
|
||||
const unsub = this.database.collectionChangeListener.subscribe(change => {
|
||||
if (change.key === collectionKey) {
|
||||
if (change.type === "create") addSubscriber(change.id);
|
||||
else removeSubscriber(); // Send delete for all elements (Don't know how to do this...)
|
||||
}
|
||||
});
|
||||
|
||||
let { collection, document, collectionKey } = await this.database.resolve(
|
||||
this.query.path
|
||||
);
|
||||
let oldKey: string = undefined;
|
||||
|
||||
const removeSubscriber = () => {
|
||||
if (!oldKey) return;
|
||||
let s = this.database.changeListener.get(oldKey);
|
||||
if (s) {
|
||||
s.delete(receivedChanges);
|
||||
if (s.size <= 0) this.database.changeListener.delete(oldKey);
|
||||
}
|
||||
oldKey = undefined;
|
||||
};
|
||||
|
||||
const addSubscriber = () => {
|
||||
let key = Database.getKey(collection, document);
|
||||
if (oldKey !== key) {
|
||||
if (oldKey !== undefined) removeSubscriber();
|
||||
|
||||
let s = this.database.changeListener.get(key);
|
||||
if (!s) {
|
||||
s = new Set();
|
||||
this.database.changeListener.set(key, s);
|
||||
}
|
||||
|
||||
s.add(receivedChanges);
|
||||
}
|
||||
};
|
||||
|
||||
if (collection) {
|
||||
addSubscriber();
|
||||
}
|
||||
|
||||
return {
|
||||
unsubscribe: () => {
|
||||
unsub();
|
||||
removeSubscriber();
|
||||
},
|
||||
value: await this.firstSend(collection, document)
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@ -95,62 +242,79 @@ interface UpdateData {
|
||||
[path: string]: {
|
||||
type: "value" | "timestamp" | "increment" | "push";
|
||||
value: any;
|
||||
}
|
||||
};
|
||||
}
|
||||
export class DocumentQuery extends Query {
|
||||
constructor(database: Database, path: string[], sender: string) {
|
||||
super(database, path, sender);
|
||||
this.onChange = this.onChange.bind(this);
|
||||
prepare(query: IQuery): IPreparedQuery {
|
||||
let type = query.type as IDocumentQueries;
|
||||
switch (type) {
|
||||
case "get":
|
||||
return {
|
||||
batchCompatible: false,
|
||||
createCollection: false,
|
||||
needDocument: false,
|
||||
permission: "read",
|
||||
runner: this.get
|
||||
};
|
||||
case "set":
|
||||
return {
|
||||
batchCompatible: true,
|
||||
createCollection: true,
|
||||
needDocument: true,
|
||||
permission: "write",
|
||||
runner: this.set
|
||||
};
|
||||
case "update":
|
||||
return {
|
||||
batchCompatible: true,
|
||||
createCollection: true,
|
||||
needDocument: true,
|
||||
permission: "write",
|
||||
runner: this.update
|
||||
};
|
||||
case "delete":
|
||||
return {
|
||||
batchCompatible: true,
|
||||
createCollection: false,
|
||||
needDocument: true,
|
||||
permission: "write",
|
||||
runner: this.delete
|
||||
};
|
||||
default:
|
||||
throw new Error("Invalid query type: " + type);
|
||||
}
|
||||
}
|
||||
|
||||
public async get() {
|
||||
let { collection, document } = await this.resolve(this.path);
|
||||
|
||||
private async get(collection: string, document: string) {
|
||||
if (!collection || !document) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return this.getDoc(collection, document)
|
||||
return this.getDoc(collection, document);
|
||||
}
|
||||
|
||||
public async set(data: any, { merge = false }) {
|
||||
if (data === null)
|
||||
return this.delete();
|
||||
let { collection, document } = await this.resolve(this.path, true);
|
||||
if (!collection) {
|
||||
throw new Error("There must be a collection!")
|
||||
private async set(
|
||||
collection: string,
|
||||
document: string,
|
||||
batch?: LevelUpChain
|
||||
) {
|
||||
const { data, options } = this.query;
|
||||
if (data === null) return this.delete(collection, document, batch);
|
||||
|
||||
let isNew = !(await this.getDoc(collection, document));
|
||||
batch.put(Database.getKey(collection, document), encode(data));
|
||||
this.sendChange(collection, document, isNew ? "added" : "modified", data);
|
||||
}
|
||||
|
||||
if (!document) {
|
||||
throw new Error("There must be a document key!")
|
||||
}
|
||||
private async update(
|
||||
collection: string,
|
||||
document: string,
|
||||
batch?: LevelUpChain
|
||||
) {
|
||||
const updateData: UpdateData = this.query.data;
|
||||
|
||||
const lock = await this.database.locks.lock(collection, document);
|
||||
|
||||
let isNew = !(await this.getDoc(collection, document))
|
||||
|
||||
return this.database.data
|
||||
.put(this.getKey(collection, document), encode(data))
|
||||
.then(() => this.sendChange(collection, document, isNew ? "added" : "modified", data))
|
||||
.finally(() => lock())
|
||||
}
|
||||
|
||||
public async update(updateData: UpdateData) {
|
||||
let { collection, document } = await this.resolve(this.path, true);
|
||||
if (!collection) {
|
||||
throw new Error("There must be a collection!")
|
||||
}
|
||||
|
||||
if (!document) {
|
||||
throw new Error("There must be a document key!")
|
||||
}
|
||||
|
||||
// Logging.debug(updateData);
|
||||
|
||||
const lock = await this.database.locks.lock(collection, document);
|
||||
try {
|
||||
let data = await this.getDoc(collection, document);
|
||||
let isNew = false
|
||||
let isNew = false;
|
||||
if (!data) {
|
||||
isNew = true;
|
||||
data = {};
|
||||
@ -162,15 +326,12 @@ export class DocumentQuery extends Query {
|
||||
let parts = path.split(".");
|
||||
while (parts.length > 1) {
|
||||
let seg = parts.shift();
|
||||
if (!data[seg])
|
||||
data[seg] = {}
|
||||
if (!data[seg]) data[seg] = {};
|
||||
d = data[seg];
|
||||
}
|
||||
|
||||
const last = parts[0];
|
||||
|
||||
// Logging.debug(parts, last, d)
|
||||
|
||||
switch (toUpdate.type) {
|
||||
case "value":
|
||||
d[last] = toUpdate.value;
|
||||
@ -179,7 +340,7 @@ export class DocumentQuery extends Query {
|
||||
if (d[last] === undefined || d[last] === null)
|
||||
d[last] = toUpdate.value;
|
||||
else if (typeof d[last] !== "number") {
|
||||
throw new Error("Field is no number!");
|
||||
throw new QueryError("Field is no number!");
|
||||
} else {
|
||||
d[last] += toUpdate.value;
|
||||
}
|
||||
@ -193,90 +354,46 @@ export class DocumentQuery extends Query {
|
||||
else if (Array.isArray(d[last])) {
|
||||
d[last].push(toUpdate.value);
|
||||
} else {
|
||||
throw new Error("Field is not array!");
|
||||
throw new QueryError("Field is not array!");
|
||||
}
|
||||
break;
|
||||
default:
|
||||
throw new Error("Invalid update type: " + toUpdate.type);
|
||||
throw new QueryError("Invalid update type: " + toUpdate.type);
|
||||
}
|
||||
}
|
||||
|
||||
this.database.data
|
||||
.put(this.getKey(collection, document), encode(data))
|
||||
.then(() => this.sendChange(collection, document, isNew ? "added" : "modified", data))
|
||||
} finally {
|
||||
lock();
|
||||
}
|
||||
//TODO: Implement
|
||||
if (batch) {
|
||||
batch.put(Database.getKey(collection, document), encode(data));
|
||||
} else {
|
||||
await this.database.data.put(
|
||||
Database.getKey(collection, document),
|
||||
encode(data)
|
||||
);
|
||||
}
|
||||
|
||||
public async delete() {
|
||||
let { collection, document } = await this.resolve(this.path);
|
||||
|
||||
if (!collection) {
|
||||
throw new Error("There must be a collection!")
|
||||
this.sendChange(collection, document, isNew ? "added" : "modified", data);
|
||||
}
|
||||
|
||||
if (!document) {
|
||||
throw new Error("There must be a document key!")
|
||||
private async delete(
|
||||
collection: string,
|
||||
document: string,
|
||||
batch?: LevelUpChain
|
||||
) {
|
||||
if (batch) {
|
||||
batch.del(Database.getKey(collection, document));
|
||||
} else {
|
||||
await this.database.data.del(Database.getKey(collection, document));
|
||||
}
|
||||
|
||||
const lock = await this.database.locks.lock(collection, document);
|
||||
|
||||
return await this.database.data
|
||||
.del(`${collection}/${document}`)
|
||||
.then(() => this.sendChange(collection, document, "deleted", null))
|
||||
.finally(() => lock())
|
||||
this.sendChange(collection, document, "deleted", null);
|
||||
}
|
||||
|
||||
|
||||
|
||||
private subscription: {
|
||||
key: string,
|
||||
onChange: (change: DocRes & { type: ChangeTypes }) => void
|
||||
};
|
||||
|
||||
async snapshot(onChange: (change: DocRes & { type: ChangeTypes }) => void) {
|
||||
if (this.subscription)
|
||||
throw new Error("This query is already subscribed!");
|
||||
let { collection, document } = await this.resolve(this.path);
|
||||
|
||||
let data = await this.getDoc(collection, document);
|
||||
let key = this.getKey(collection, document);
|
||||
this.subscription = {
|
||||
key,
|
||||
onChange
|
||||
}
|
||||
let s = this.database.changes.get(key);
|
||||
if (!s) {
|
||||
s = new Set();
|
||||
this.database.changes.set(key, s);
|
||||
checkChange(change: Change) {
|
||||
return true;
|
||||
}
|
||||
|
||||
s.add(this.onChange);
|
||||
|
||||
return data;
|
||||
}
|
||||
|
||||
onChange(change: Change) {
|
||||
// if(change.sender === this.sender)
|
||||
// return
|
||||
this.subscription.onChange({
|
||||
id: change.document,
|
||||
data: change.data,
|
||||
type: change.type
|
||||
})
|
||||
}
|
||||
|
||||
unsubscribe() {
|
||||
if (!this.subscription)
|
||||
return;
|
||||
let s = this.database.changes.get(this.subscription.key);
|
||||
s.delete(this.onChange);
|
||||
if (s.size <= 0)
|
||||
this.database.changes.delete(this.subscription.key);
|
||||
|
||||
this.subscription = undefined;
|
||||
firstSend(collection: string, document: string) {
|
||||
return this.get(collection, document);
|
||||
}
|
||||
|
||||
public static fromQuery(query: Query) {
|
||||
@ -286,138 +403,216 @@ export class DocumentQuery extends Query {
|
||||
|
||||
type FieldPath = string;
|
||||
type WhereFilterOp =
|
||||
| '<'
|
||||
| '<='
|
||||
| '=='
|
||||
| '>='
|
||||
| '>'
|
||||
| 'array-contains'
|
||||
| 'in'
|
||||
| 'array-contains-any';
|
||||
| "<"
|
||||
| "<="
|
||||
| "=="
|
||||
| ">="
|
||||
| ">"
|
||||
| "array-contains"
|
||||
| "in"
|
||||
| "array-contains-any";
|
||||
|
||||
interface IQueryWhere {
|
||||
fieldPath: FieldPath,
|
||||
opStr: WhereFilterOp,
|
||||
value: any
|
||||
interface IQueryWhereVerbose {
|
||||
fieldPath: FieldPath;
|
||||
opStr: WhereFilterOp;
|
||||
value: any;
|
||||
}
|
||||
|
||||
type IQueryWhereArray = [FieldPath, WhereFilterOp, any];
|
||||
|
||||
interface DocRes {
|
||||
id: string;
|
||||
data: any;
|
||||
}
|
||||
type IQueryWhere = IQueryWhereArray | IQueryWhereVerbose;
|
||||
|
||||
export class CollectionQuery extends Query {
|
||||
constructor(database: Database, path: string[], sender: string) {
|
||||
super(database, path, sender);
|
||||
this.onChange = this.onChange.bind(this);
|
||||
private _addId: string;
|
||||
|
||||
prepare(query): IPreparedQuery {
|
||||
switch (query.type as ICollectionQueries) {
|
||||
case "add":
|
||||
this._addId = nanoid(ALPHABET, 32);
|
||||
return {
|
||||
batchCompatible: true,
|
||||
createCollection: true,
|
||||
needDocument: false,
|
||||
runner: this.add,
|
||||
permission: "write",
|
||||
additionalLock: [...query.path, this._addId]
|
||||
};
|
||||
case "get":
|
||||
const limit = (query.options || {}).limit;
|
||||
if (limit) this.limit = limit;
|
||||
const where = (query.options || {}).where;
|
||||
if (where) this.where = where;
|
||||
|
||||
return {
|
||||
batchCompatible: false,
|
||||
createCollection: false,
|
||||
needDocument: false,
|
||||
permission: "read",
|
||||
runner: this.get
|
||||
};
|
||||
case "keys":
|
||||
return {
|
||||
batchCompatible: false,
|
||||
createCollection: false,
|
||||
needDocument: false,
|
||||
permission: "read",
|
||||
runner: this.keys
|
||||
};
|
||||
case "list":
|
||||
return {
|
||||
batchCompatible: false,
|
||||
createCollection: false,
|
||||
needDocument: false,
|
||||
permission: "read",
|
||||
runner: this.keys
|
||||
};
|
||||
case "delete-collection":
|
||||
return {
|
||||
batchCompatible: false,
|
||||
createCollection: false,
|
||||
needDocument: false,
|
||||
permission: "write",
|
||||
runner: this.deleteCollection
|
||||
};
|
||||
// run = () => q.deleteCollection();
|
||||
// break;
|
||||
default:
|
||||
throw new Error("Invalid query!");
|
||||
}
|
||||
}
|
||||
|
||||
private _where: IQueryWhereArray[] = [];
|
||||
public set where(value: IQueryWhere[]) {
|
||||
const invalidWhere = new QueryError("Invalid Where");
|
||||
if (!Array.isArray(value)) throw invalidWhere;
|
||||
let c = [];
|
||||
this._where = value.map(cond => {
|
||||
Logging.debug("Query Condition", cond);
|
||||
if (Array.isArray(cond)) {
|
||||
if (cond.length !== 3) throw invalidWhere;
|
||||
return cond;
|
||||
} else {
|
||||
if (
|
||||
cond &&
|
||||
typeof cond === "object" &&
|
||||
"fieldPath" in cond &&
|
||||
"opStr" in cond &&
|
||||
"value" in cond
|
||||
) {
|
||||
return [cond.fieldPath, cond.opStr, cond.value];
|
||||
} else {
|
||||
throw invalidWhere;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public where: IQueryWhere[] = [];
|
||||
public limit: number = -1;
|
||||
|
||||
public async add(value: any) {
|
||||
let id = nanoid(ALPHABET, 32);
|
||||
let q = new DocumentQuery(this.database, [...this.path, id], this.sender);
|
||||
await q.set(value, {});
|
||||
return id;
|
||||
public async add(
|
||||
collection: string,
|
||||
document: string,
|
||||
batch: LevelUpChain,
|
||||
collectionKey: string
|
||||
) {
|
||||
let q = new DocumentQuery(this.database, this.session, {
|
||||
type: "set",
|
||||
path: this.additionalLock,
|
||||
data: this.query.data,
|
||||
options: this.query.options
|
||||
});
|
||||
await q.run(collection, this._addId, batch, collectionKey);
|
||||
return this._addId;
|
||||
}
|
||||
|
||||
private getStreamOptions(collection: string) {
|
||||
let gt = Buffer.from(this.getKey(collection) + " ");
|
||||
let gt = Buffer.from(Database.getKey(collection) + " ");
|
||||
gt[gt.length - 1] = 0;
|
||||
|
||||
let lt = Buffer.alloc(gt.length);
|
||||
lt.set(gt);
|
||||
lt[gt.length - 1] = 0xFF;
|
||||
lt[gt.length - 1] = 0xff;
|
||||
|
||||
return {
|
||||
gt,
|
||||
lt
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
public async keys() {
|
||||
let { collection, document } = await this.resolve(this.path);
|
||||
if (document)
|
||||
throw new Error("Keys only works on collections!");
|
||||
if (!collection)
|
||||
throw new Error("There must be a collection");
|
||||
public async keys(collection: string) {
|
||||
if (!collection) return [];
|
||||
|
||||
return new Promise<string[]>((yes, no) => {
|
||||
let keys = [];
|
||||
const stream = this.database.data.createKeyStream({
|
||||
...this.getStreamOptions(collection),
|
||||
keyAsBuffer: false
|
||||
})
|
||||
});
|
||||
stream.on("data", (key: string) => {
|
||||
let s = key.split("/", 2);
|
||||
if (s.length > 1)
|
||||
keys.push(s[1]);
|
||||
if (s.length > 1) keys.push(s[1]);
|
||||
});
|
||||
stream.on("end", () => yes(keys));
|
||||
stream.on("error", no);
|
||||
});
|
||||
}
|
||||
|
||||
private getFieldValue(data: any, path: FieldPath) {
|
||||
private _getFieldValue(data: any, path: FieldPath) {
|
||||
let parts = path.split(".");
|
||||
let d = data;
|
||||
while (parts.length > 0) {
|
||||
let seg = parts.shift();
|
||||
|
||||
d = data[seg];
|
||||
if (d === undefined || d === null)
|
||||
break; // Undefined/Null has no other fields!
|
||||
if (d === undefined || d === null) break; // Undefined/Null has no other fields!
|
||||
}
|
||||
return d;
|
||||
}
|
||||
|
||||
private fitsWhere(data: any): boolean {
|
||||
if (this.where.length > 0) {
|
||||
return this.where.every(where => {
|
||||
let val = this.getFieldValue(data, where.fieldPath);
|
||||
switch (where.opStr) {
|
||||
private _fitsWhere(data: any): boolean {
|
||||
if (this._where.length > 0) {
|
||||
return this._where.every(([fieldPath, opStr, value]) => {
|
||||
let val = this._getFieldValue(data, fieldPath);
|
||||
switch (opStr) {
|
||||
case "<":
|
||||
return val < where.value;
|
||||
return val < value;
|
||||
case "<=":
|
||||
return val <= where.value;
|
||||
return val <= value;
|
||||
case "==":
|
||||
return val == where.value;
|
||||
return val == value;
|
||||
case ">=":
|
||||
return val >= where.value;
|
||||
return val >= value;
|
||||
case ">":
|
||||
return val > where.value;
|
||||
return val > value;
|
||||
case "array-contains":
|
||||
if (Array.isArray(val)) {
|
||||
return val.some(e => e === where.value);
|
||||
return val.some(e => e === value);
|
||||
}
|
||||
|
||||
break;
|
||||
return false;
|
||||
// case "array-contains-any":
|
||||
// case "in":
|
||||
|
||||
default:
|
||||
throw new Error("Invalid where operation " + where.opStr);
|
||||
case "in":
|
||||
if (typeof val === "object") {
|
||||
return value in val;
|
||||
}
|
||||
})
|
||||
return false;
|
||||
default:
|
||||
throw new QueryError("Invalid where operation " + opStr);
|
||||
}
|
||||
});
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
async get() {
|
||||
let { collection, document } = await this.resolve(this.path);
|
||||
if (document)
|
||||
throw new Error("Keys only works on collections!");
|
||||
if (!collection)
|
||||
throw new Error("There must be a collection");
|
||||
async get(collection: string) {
|
||||
if (!collection) return [];
|
||||
|
||||
return new Promise<DocRes[]>((yes, no) => {
|
||||
const stream = this.database.data.iterator({
|
||||
...this.getStreamOptions(collection),
|
||||
keyAsBuffer: false,
|
||||
valueAsBuffer: true
|
||||
})
|
||||
});
|
||||
|
||||
let values: DocRes[] = [];
|
||||
|
||||
@ -425,29 +620,26 @@ export class CollectionQuery extends Query {
|
||||
if (err) {
|
||||
no(err);
|
||||
stream.end(err => Logging.error(err));
|
||||
}
|
||||
else {
|
||||
} else {
|
||||
if (!key && !value) {
|
||||
// END
|
||||
Logging.debug("Checked all!")
|
||||
Logging.debug("Checked all!");
|
||||
yes(values);
|
||||
} else {
|
||||
let s = key.split("/", 2);
|
||||
if (s.length <= 1)
|
||||
return;
|
||||
if (s.length <= 1) return;
|
||||
|
||||
const id = s[1];
|
||||
|
||||
let data = decode(value);
|
||||
if (this.fitsWhere(data)) {
|
||||
if (this._fitsWhere(data)) {
|
||||
if (this.limit < 0 || values.length < this.limit) {
|
||||
values.push({
|
||||
id,
|
||||
data
|
||||
});
|
||||
}
|
||||
else {
|
||||
stream.end((err) => err ? no(err) : yes(values))
|
||||
} else {
|
||||
stream.end(err => (err ? no(err) : yes(values)));
|
||||
return;
|
||||
}
|
||||
}
|
||||
@ -455,99 +647,62 @@ export class CollectionQuery extends Query {
|
||||
stream.next(onValue);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
stream.next(onValue)
|
||||
})
|
||||
}
|
||||
|
||||
private subscription: {
|
||||
key: string,
|
||||
onChange: (change: (DocRes & { type: ChangeTypes })[]) => void
|
||||
};
|
||||
|
||||
async snapshot(onChange: (change: (DocRes & { type: ChangeTypes })[]) => void) {
|
||||
if (this.subscription)
|
||||
throw new Error("This query is already subscribed!");
|
||||
let { collection, document } = await this.resolve(this.path);
|
||||
|
||||
let data = await this.get();
|
||||
|
||||
let key = this.getKey(collection, document);
|
||||
this.subscription = {
|
||||
key,
|
||||
onChange
|
||||
}
|
||||
let s = this.database.changes.get(key);
|
||||
if (!s) {
|
||||
s = new Set();
|
||||
this.database.changes.set(key, s);
|
||||
stream.next(onValue);
|
||||
});
|
||||
}
|
||||
|
||||
s.add(this.onChange);
|
||||
|
||||
return data;
|
||||
checkChange(change: Change) {
|
||||
return this._fitsWhere(change.data);
|
||||
}
|
||||
|
||||
onChange(change: Change) {
|
||||
// if(change.sender === this.sender)
|
||||
// return
|
||||
|
||||
if (this.fitsWhere(change.data)) {
|
||||
this.subscription.onChange([{
|
||||
id: change.document,
|
||||
data: change.data,
|
||||
type: change.type
|
||||
}])
|
||||
firstSend(collection: string) {
|
||||
return this.get(collection);
|
||||
}
|
||||
}
|
||||
|
||||
unsubscribe() {
|
||||
if (!this.subscription)
|
||||
return;
|
||||
let s = this.database.changes.get(this.subscription.key);
|
||||
s.delete(this.onChange);
|
||||
if (s.size <= 0)
|
||||
this.database.changes.delete(this.subscription.key);
|
||||
|
||||
this.subscription = undefined;
|
||||
}
|
||||
|
||||
|
||||
public async collections() {
|
||||
if (!this.session.root) throw new QueryError("No Permission!");
|
||||
|
||||
return new Promise<string[]>((yes, no) => {
|
||||
let keys = [];
|
||||
const stream = this.database.data.createKeyStream({ keyAsBuffer: false })
|
||||
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");
|
||||
}
|
||||
public async deleteCollection(
|
||||
collection: string,
|
||||
document: string,
|
||||
_b: LevelUpChain,
|
||||
collectionKey: string
|
||||
) {
|
||||
if (!this.session.root) throw new QueryError("No Permission!");
|
||||
|
||||
//TODO: Lock whole collection!
|
||||
|
||||
let batch = this.database.data.batch();
|
||||
try {
|
||||
if (collection) {
|
||||
let documents = await this.keys();
|
||||
let documents = await this.keys(collection);
|
||||
// Logging.debug("To delete:", documents)
|
||||
for (let document of documents) {
|
||||
batch.del(this.getKey(collection, document));
|
||||
batch.del(Database.getKey(collection, document));
|
||||
}
|
||||
await batch.write();
|
||||
batch = undefined;
|
||||
await this.database.collections.del(collectionKey);
|
||||
this.database.collectionChangeListener.send({
|
||||
id: collection,
|
||||
key: collectionKey,
|
||||
type: "delete"
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
if (batch)
|
||||
batch.clear();
|
||||
if (batch) batch.clear();
|
||||
}
|
||||
}
|
||||
|
||||
@ -555,3 +710,9 @@ export class CollectionQuery extends Query {
|
||||
return new CollectionQuery(...Query.getConstructorParams(query));
|
||||
}
|
||||
}
|
||||
|
||||
export class QueryError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
|
@ -2,13 +2,15 @@ import Session from "./session";
|
||||
import Logging from "@hibas123/nodelogging";
|
||||
|
||||
interface IRule<T> {
|
||||
".write"?: T
|
||||
".read"?: T
|
||||
".write"?: T;
|
||||
".read"?: T;
|
||||
}
|
||||
|
||||
type IRuleConfig<T> = {
|
||||
type IRuleConfig<T> =
|
||||
| IRule<T>
|
||||
| {
|
||||
[segment: string]: IRuleConfig<T>;
|
||||
} | IRule<T>;
|
||||
};
|
||||
|
||||
type IRuleRaw = IRuleConfig<string>;
|
||||
type IRuleParsed = IRuleConfig<boolean>;
|
||||
@ -17,17 +19,16 @@ const resolve = (value: any) => {
|
||||
if (value === true) {
|
||||
return true;
|
||||
} else if (typeof value === "string") {
|
||||
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
|
||||
export class Rules {
|
||||
rules: IRuleParsed;
|
||||
constructor(private config: string) {
|
||||
let parsed: IRuleRaw = JSON.parse(config);
|
||||
|
||||
const analyze = (raw: IRuleRaw) => {
|
||||
const analyse = (raw: IRuleRaw) => {
|
||||
let r: IRuleParsed = {};
|
||||
|
||||
if (raw[".read"]) {
|
||||
@ -47,18 +48,25 @@ export class Rules {
|
||||
}
|
||||
|
||||
for (let segment in raw) {
|
||||
if (segment.startsWith("."))
|
||||
continue;
|
||||
if (segment.startsWith(".")) continue;
|
||||
|
||||
r[segment] = analyze(raw[segment]);
|
||||
r[segment] = analyse(raw[segment]);
|
||||
}
|
||||
return r;
|
||||
};
|
||||
|
||||
this.rules = analyse(parsed);
|
||||
}
|
||||
|
||||
this.rules = analyze(parsed);
|
||||
}
|
||||
|
||||
hasPermission(path: string[], session: Session): { read: boolean, write: boolean } {
|
||||
hasPermission(
|
||||
path: string[],
|
||||
session: Session
|
||||
): { read: boolean; write: boolean } {
|
||||
if (session.root)
|
||||
return {
|
||||
read: true,
|
||||
write: true
|
||||
};
|
||||
let read = this.rules[".read"] || false;
|
||||
let write = this.rules[".write"] || false;
|
||||
|
||||
@ -77,22 +85,21 @@ export class Rules {
|
||||
.find(e => {
|
||||
switch (e) {
|
||||
case "$uid":
|
||||
if (segment === session.uid)
|
||||
return true;
|
||||
if (segment === session.uid) return true;
|
||||
break;
|
||||
}
|
||||
return false;
|
||||
})
|
||||
});
|
||||
|
||||
rules = (k ? rules[k] : undefined) || rules[segment] || rules["*"];
|
||||
|
||||
if (rules) {
|
||||
if (rules[".read"]) {
|
||||
read = rules[".read"]
|
||||
read = rules[".read"];
|
||||
}
|
||||
|
||||
if (rules[".write"]) {
|
||||
read = rules[".write"]
|
||||
read = rules[".write"];
|
||||
}
|
||||
} else {
|
||||
break;
|
||||
@ -102,7 +109,7 @@ export class Rules {
|
||||
return {
|
||||
read: read as boolean,
|
||||
write: write as boolean
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
toJSON() {
|
||||
|
@ -1,8 +1,11 @@
|
||||
|
||||
export default class Session {
|
||||
constructor(private _sessionid: string) { }
|
||||
get sessionid() {
|
||||
get id() {
|
||||
return this._sessionid;
|
||||
}
|
||||
root: boolean = false;
|
||||
uid: string = undefined;
|
||||
|
||||
subscriptions = new Map<string, (() => void)>();
|
||||
}
|
12
src/helper/jwt.ts
Normal file
12
src/helper/jwt.ts
Normal file
@ -0,0 +1,12 @@
|
||||
import * as JWT from "jsonwebtoken";
|
||||
|
||||
export async function verifyJWT(token: string, publicKey: string) {
|
||||
return new Promise<any | undefined>((yes) => {
|
||||
JWT.verify(token, publicKey, (err, decoded) => {
|
||||
if (err)
|
||||
yes(undefined);
|
||||
else
|
||||
yes(decoded);
|
||||
})
|
||||
})
|
||||
}
|
@ -5,9 +5,14 @@ import { DatabaseManager } from "./database/database";
|
||||
import { createServer } from "http";
|
||||
import { ConnectionManager } from "./connection";
|
||||
import { LoggingTypes } from "@hibas123/logging";
|
||||
import { readFileSync } from "fs";
|
||||
|
||||
Logging.logLevel = config.dev ? LoggingTypes.Debug : LoggingTypes.Log;
|
||||
|
||||
const version = JSON.parse(readFileSync("./package.json").toString()).version;
|
||||
|
||||
Logging.log("Starting Database version:", version);
|
||||
|
||||
DatabaseManager.init().then(() => {
|
||||
const http = createServer(Web.callback());
|
||||
ConnectionManager.bind(http);
|
||||
|
@ -398,6 +398,12 @@ export class NoPermissionError extends HttpError {
|
||||
}
|
||||
}
|
||||
|
||||
export class UnauthorizedError extends HttpError {
|
||||
constructor(message: string) {
|
||||
super(message, HttpStatusCode.UNAUTHORIZED)
|
||||
}
|
||||
}
|
||||
|
||||
export class BadRequestError extends HttpError {
|
||||
constructor(message: string) {
|
||||
super(message, HttpStatusCode.BAD_REQUEST)
|
||||
|
@ -1,5 +1,62 @@
|
||||
import * as Router from "koa-router";
|
||||
import AdminRoute from "./admin";
|
||||
import { DatabaseManager } from "../../database/database";
|
||||
import {
|
||||
NotFoundError,
|
||||
NoPermissionError,
|
||||
BadRequestError
|
||||
} from "../helper/errors";
|
||||
import Logging from "@hibas123/nodelogging";
|
||||
import Session from "../../database/session";
|
||||
import nanoid = require("nanoid");
|
||||
import { verifyJWT } from "../../helper/jwt";
|
||||
import { QueryError } from "../../database/query";
|
||||
const V1 = new Router({ prefix: "/v1" });
|
||||
|
||||
V1.use("/admin", AdminRoute.routes(), AdminRoute.allowedMethods());
|
||||
|
||||
V1.post("/db/:database/query", async ctx => {
|
||||
const { database } = ctx.params;
|
||||
const { accesskey, authkey, rootkey } = ctx.query;
|
||||
|
||||
const query = ctx.request.body;
|
||||
if (!query) {
|
||||
throw new BadRequestError("Query not defined!");
|
||||
}
|
||||
|
||||
const session = new Session(nanoid());
|
||||
const db = DatabaseManager.getDatabase(database);
|
||||
if (!db) {
|
||||
throw new NotFoundError("Database not found!");
|
||||
}
|
||||
|
||||
if (db.accesskey) {
|
||||
if (!accesskey || accesskey !== db.accesskey) {
|
||||
throw new NoPermissionError("Invalid Access Key");
|
||||
}
|
||||
}
|
||||
|
||||
if (authkey && db.publickey) {
|
||||
let res = await verifyJWT(authkey, db.publickey);
|
||||
if (!res || !res.uid) {
|
||||
throw new BadRequestError("Invalid JWT");
|
||||
} else {
|
||||
session.uid = res.uid;
|
||||
}
|
||||
}
|
||||
|
||||
if (rootkey && db.rootkey) {
|
||||
if (rootkey === db.rootkey) {
|
||||
session.root = true;
|
||||
Logging.warning(`Somebody logged into ${database} via rootkey`);
|
||||
}
|
||||
}
|
||||
|
||||
ctx.body = await db.run([query], session).catch(err => {
|
||||
if (err instanceof QueryError) {
|
||||
throw new BadRequestError(err.message);
|
||||
}
|
||||
throw err;
|
||||
});
|
||||
});
|
||||
export default V1;
|
Reference in New Issue
Block a user