OpenServer
/
RealtimeDB-OLD
Archived
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Adding HTTP Query Endpoint and refining some things

This commit is contained in:
Fabian Stamm 2019-11-15 16:36:42 +01:00
parent 4cee0048f5
commit d2621fdd3c
6 changed files with 163 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
src/connection.ts
View File

@ -1,43 +1,12 @@
import * as WebSocket from "ws";
import { Server, IncomingMessage } from "http";
import { DatabaseManager, IQuery, ITypedQuery } 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, IQuery, ITypedQuery } from "./database/database";
import { CollectionQuery, DocumentQuery } 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,7 +19,6 @@ 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());
const query = new URL(req.url, "http://localhost").searchParams;
@ -92,7 +60,6 @@ export class ConnectionManager {
}
}
const stored = new Map<string, Query>();
const answer = (id: string, data: any, error: boolean = false) => {
socket.send(JSON.stringify({ ns: "message", data: { id, error, data } }));
}
@ -139,9 +106,10 @@ export class ConnectionManager {
socket.on("close", () => {
Logging.log(`${session.id} has disconnected!`);
Logging.debug("Clearing stored:", stored);
stored.forEach(query => (query as DocumentQuery | CollectionQuery).unsubscribe());
stored.clear();
session.queries.forEach((query: DocumentQuery | CollectionQuery) => {
query.unsubscribe();
})
session.queries.clear();
socket.removeAllListeners();
})
}

19
src/database/database.ts
View File

@ -2,7 +2,7 @@ import { Rules } from "./rules";
import Settings from "../settings";
import getLevelDB, { LevelDB, deleteLevelDB } from "../storage";
import DocumentLock from "./lock";
import { DocumentQuery, CollectionQuery, Query } from "./query";
import { DocumentQuery, CollectionQuery, Query, QueryError } from "./query";
import Logging from "@hibas123/nodelogging";
import Session from "./session";
import nanoid = require("nanoid");
@ -134,7 +134,20 @@ export class Database {
return new Query(this, path, session);
}
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(query: IQuery, session: Session) {
this.validate(query);
const isCollection = query.path.length % 2 === 1;
if (isCollection) {
const q = new CollectionQuery(this, query.path, session);
@ -178,6 +191,8 @@ export class Database {
}
async snapshot(query: ITypedQuery<"snapshot">, session: Session, onchange: (change: any) => void) {
this.validate(query);
const isCollection = query.path.length % 2 === 1;
let q: DocumentQuery | CollectionQuery;
if (isCollection) {
@ -208,8 +223,6 @@ export class Database {
}
}
async stop() {
await this.data.close();
}

109
src/database/query.ts
View File

@ -11,10 +11,6 @@ const ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz
const { encode, decode } = MP;
interface ISubscribeOptions {
existing: boolean;
}
export class Query {
/**
* Returns true if the path only contains valid characters and false if it doesn't
@ -26,10 +22,10 @@ export class Query {
constructor(protected database: Database, protected path: string[], protected session: Session) {
if (path.length > 10) {
throw new Error("Path is to long. Path is only allowed to be 10 Layers deep!");
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 '>' ");
throw new QueryError("Path can only contain a-z A-Z 0-9 '-' '-' '<' and '>' ");
}
}
@ -119,11 +115,11 @@ export class DocumentQuery extends Query {
return this.delete();
let { collection, document } = await this.resolve(this.path, true);
if (!collection) {
throw new Error("There must be a collection!")
throw new QueryError("There must be a collection!")
}
if (!document) {
throw new Error("There must be a document key!")
throw new QueryError("There must be a document key!")
}
const lock = await this.database.locks.lock(collection, document);
@ -139,11 +135,11 @@ export class DocumentQuery extends Query {
public async update(updateData: UpdateData) {
let { collection, document } = await this.resolve(this.path, true);
if (!collection) {
throw new Error("There must be a collection!")
throw new QueryError("There must be a collection!")
}
if (!document) {
throw new Error("There must be a document key!")
throw new QueryError("There must be a document key!")
}
// Logging.debug(updateData);
@ -180,7 +176,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;
}
@ -194,11 +190,11 @@ 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);
}
}
@ -215,11 +211,11 @@ export class DocumentQuery extends Query {
let { collection, document } = await this.resolve(this.path);
if (!collection) {
throw new Error("There must be a collection!")
throw new QueryError("There must be a collection!")
}
if (!document) {
throw new Error("There must be a document key!")
throw new QueryError("There must be a document key!")
}
const lock = await this.database.locks.lock(collection, document);
@ -239,7 +235,7 @@ export class DocumentQuery extends Query {
async snapshot(onChange: (change: DocRes & { type: ChangeTypes }) => void) {
if (this.subscription)
throw new Error("This query is already subscribed!");
throw new QueryError("This query is already subscribed!");
let { collection, document } = await this.resolve(this.path);
let data = await this.getDoc(collection, document);
@ -296,12 +292,15 @@ type WhereFilterOp =
| 'in'
| 'array-contains-any';
interface IQueryWhere {
interface IQueryWhereVerbose {
fieldPath: FieldPath,
opStr: WhereFilterOp,
value: any
}
type IQueryWhereArray = [FieldPath, WhereFilterOp, any];
type IQueryWhere = IQueryWhereArray | IQueryWhereVerbose;
interface DocRes {
id: string;
@ -315,7 +314,28 @@ export class CollectionQuery extends Query {
}
public where: IQueryWhere[] = [];
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 limit: number = -1;
public async add(value: any) {
@ -342,9 +362,9 @@ export class CollectionQuery extends Query {
public async keys() {
let { collection, document } = await this.resolve(this.path);
if (document)
throw new Error("Keys only works on collections!");
throw new QueryError("Keys only works on collections!");
if (!collection)
throw new Error("There must be a collection");
return []
return new Promise<string[]>((yes, no) => {
let keys = [];
@ -376,35 +396,34 @@ export class CollectionQuery extends Query {
}
private fitsWhere(data: any): boolean {
if (this.where.length > 0) {
return this.where.every(where => {
let val = this.getFieldValue(data, where.fieldPath);
Logging.debug("Value:", val);
switch (where.opStr) {
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);
}
return false;
// case "array-contains-any":
case "in":
if (typeof val === "object") {
return where.value in val;
return value in val;
}
return false;
default:
throw new Error("Invalid where operation " + where.opStr);
throw new QueryError("Invalid where operation " + opStr);
}
})
}
@ -414,9 +433,10 @@ export class CollectionQuery extends Query {
async get() {
let { collection, document } = await this.resolve(this.path);
if (document)
throw new Error("Keys only works on collections!");
throw new QueryError("Keys only works on collections!");
if (!collection)
throw new Error("There must be a collection");
return [];
return new Promise<DocRes[]>((yes, no) => {
const stream = this.database.data.iterator({
...this.getStreamOptions(collection),
@ -462,10 +482,7 @@ export class CollectionQuery extends Query {
}
}
stream.next(onValue)
}).then(val => {
Logging.debug("Get returns:", val, ((this.where || [])[0] || {}));
return val;
stream.next(onValue);
})
}
@ -476,7 +493,7 @@ export class CollectionQuery extends Query {
async snapshot(onChange: (change: (DocRes & { type: ChangeTypes })[]) => void) {
if (this.subscription)
throw new Error("This query is already subscribed!");
throw new QueryError("This query is already subscribed!");
let { collection, document } = await this.resolve(this.path, true);
let data = await this.get();
@ -524,7 +541,7 @@ export class CollectionQuery extends Query {
public async collections() {
if (!this.session.root)
throw new Error("No Permission!");
throw new QueryError("No Permission!");
return new Promise<string[]>((yes, no) => {
let keys = [];
@ -537,12 +554,12 @@ export class CollectionQuery extends Query {
public async deleteCollection() {
if (!this.session.root)
throw new Error("No Permission!");
throw new QueryError("No Permission!");
const { collection, document, collectionKey } = await this.resolve(this.path);
if (document) {
throw new Error("There can be no document defined on this operation");
throw new QueryError("There can be no document defined on this operation");
}
//TODO: Lock whole collection!
@ -568,4 +585,10 @@ export class CollectionQuery extends Query {
public static fromQuery(query: Query) {
return new CollectionQuery(...Query.getConstructorParams(query));
}
}
export class QueryError extends Error {
constructor(message: string) {
super(message);
}
}

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

6
src/web/helper/errors.ts
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)

54
src/web/v1/index.ts
View File

@ -1,5 +1,59 @@
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("");
}
}
if (authkey && db.publickey) {
let res = await verifyJWT(authkey, db.publickey);
if (!res || !res.uid) {
throw new BadRequestError("Invalid JWT");
return;
} 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;