18 Commits

Author SHA1 Message Date
68295c148d Version Bump
Some checks reported errors
continuous-integration/drone/push Build encountered an error
continuous-integration/drone/tag Build is passing
2020-03-24 15:20:19 +01:00
2a62c3d3ac Fixing error
All checks were successful
continuous-integration/drone/tag Build is passing
2020-03-24 15:18:13 +01:00
1434036b42 Enabling rules
Some checks failed
continuous-integration/drone/tag Build is failing
2020-03-24 15:16:21 +01:00
88b0cb68d8 Adding CI
All checks were successful
continuous-integration/drone/push Build is passing
2020-01-18 14:40:35 +01:00
904b986e22 Adding batch support 2019-12-01 03:34:25 +01:00
2ac9def153 Version bump 2019-11-16 18:23:07 +01:00
d2621fdd3c Adding HTTP Query Endpoint and refining some things 2019-11-15 16:36:42 +01:00
4cee0048f5 Fixing wrong URL arguments 2019-11-14 17:37:10 +01:00
3432ea5e23 Version Bump 2019-11-14 17:32:44 +01:00
0175822699 Switching to new Query mechanism 2019-11-14 17:28:07 +01:00
72e5c0dedd Improving version support 2019-11-14 16:40:57 +01:00
d48bf46231 Merge branch 'v2' of https://git.stamm.me/OpenServer/RealtimeDB into v2 2019-11-14 14:26:14 +01:00
e287890ca1 Adding version log on startup 2019-11-14 14:26:06 +01:00
3e0dc06521 Adding version log on startup 2019-11-14 14:24:11 +01:00
50268d05c5 onSnaphot creates Collections 2019-11-14 14:19:11 +01:00
10f3b4fa50 Changing sender to session in Query 2019-11-12 13:02:28 +01:00
e42f0d3991 Beta 5 release 2019-11-10 15:06:34 +01:00
87985def82 Fixing bug with non functional limit 2019-11-10 15:05:30 +01:00
17 changed files with 1295 additions and 2731 deletions

21
.drone.yml Normal file
View 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

View File

@ -1,3 +1,5 @@
[*]
charset = utf-8
indent_size = 3
indent_style = space
insert_final_newline = true

2458
package-lock.json generated

File diff suppressed because it is too large Load Diff

View File

@ -1,6 +1,6 @@
{
"name": "@hibas123/realtimedb",
"version": "2.0.0-beta.4",
"version": "2.0.0-beta.18",
"description": "",
"main": "lib/index.js",
"private": true,
@ -17,32 +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/logging": "^2.1.1",
"@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"
}
}

View File

@ -1,5 +1,6 @@
import Logging from "@hibas123/nodelogging";
import * as dotenv from "dotenv";
import { LoggingTypes } from "@hibas123/logging";
dotenv.config()
@ -17,4 +18,8 @@ const config: IConfig = {
dev: (process.env.DEV || "").toLowerCase() === "true"
}
if (config.dev) {
Logging.logLevel = LoggingTypes.Log;
}
export default config;

View File

@ -1,43 +1,12 @@
import Logging from "@hibas123/nodelogging";
import { IncomingMessage, Server } from "http";
import * as WebSocket from "ws";
import { Server, IncomingMessage } from "http";
import { DatabaseManager } from "./database/database";
import Logging from "@hibas123/logging";
import { Query, CollectionQuery, DocumentQuery } from "./database/query";
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);
})

View File

@ -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() {

View File

@ -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)

File diff suppressed because it is too large Load Diff

View File

@ -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() {

View File

@ -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
View 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);
})
})
}

View File

@ -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);

View File

@ -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)

View File

@ -1,7 +1,7 @@
import * as Handlebars from "handlebars";
import { readFileSync } from "fs";
import config from "../../config";
import Logging from "@hibas123/logging";
import Logging from "@hibas123/nodelogging";
function checkCondition(v1, operator, v2) {

View File

@ -6,7 +6,7 @@ import { BadRequestError, NoPermissionError } from "../helper/errors";
import { DatabaseManager } from "../../database/database";
import { MP } from "../../database/query";
import config from "../../config";
import Logging from "@hibas123/logging";
import Logging from "@hibas123/nodelogging";
import { getView } from "../helper/hb";
const AdminRoute = new Router();

View File

@ -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;