Adding batch support
This commit is contained in:
Fabian Stamm
@ -1,29 +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, QueryError } from "./query";
|
||||
import { DocumentQuery, CollectionQuery, Query, QueryError, ITypedQuery, IQuery } from "./query";
|
||||
import Logging from "@hibas123/nodelogging";
|
||||
import Session from "./session";
|
||||
import nanoid = require("nanoid");
|
||||
import nanoid = require("nanoid/generate");
|
||||
import { Observable } from "@hibas123/utils";
|
||||
|
||||
type IWriteQueries = "set" | "update" | "delete" | "add";
|
||||
type ICollectionQueries = "get" | "add" | "keys" | "delete-collection" | "list";
|
||||
type IDocumentQueries = "get" | "set" | "update" | "delete";
|
||||
|
||||
export interface ITypedQuery<T> {
|
||||
path: string[];
|
||||
type: T;
|
||||
data?: any;
|
||||
options?: any;
|
||||
}
|
||||
|
||||
interface ITransaction {
|
||||
queries: ITypedQuery<IWriteQueries>[];
|
||||
}
|
||||
|
||||
export type IQuery = ITypedQuery<ICollectionQueries | IDocumentQueries>;
|
||||
const ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
|
||||
|
||||
// interface ITransaction {
|
||||
// queries: ITypedQuery<IWriteQueries>[];
|
||||
// }
|
||||
|
||||
export class DatabaseManager {
|
||||
static databases = new Map<string, Database>();
|
||||
@ -66,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() {
|
||||
@ -84,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 {
|
||||
@ -124,14 +123,71 @@ 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[], session: Session, type: "document" | "collection" | "any") {
|
||||
if (type === "document")
|
||||
return new DocumentQuery(this, path, session);
|
||||
else if (type === "collection")
|
||||
return new CollectionQuery(this, path, session);
|
||||
else
|
||||
return new Query(this, path, session);
|
||||
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>) {
|
||||
@ -146,80 +202,121 @@ export class Database {
|
||||
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);
|
||||
let type = query.type as ICollectionQueries;
|
||||
switch (type) {
|
||||
case "add":
|
||||
return q.add(query.data);
|
||||
case "get":
|
||||
const limit = (query.options || {}).limit;
|
||||
if (limit)
|
||||
q.limit = limit;
|
||||
const where = (query.options || {}).where;
|
||||
if (where)
|
||||
q.where = where;
|
||||
return q.get();
|
||||
case "keys":
|
||||
return q.keys();
|
||||
case "list":
|
||||
return q.collections();
|
||||
case "delete-collection":
|
||||
return q.deleteCollection();
|
||||
default:
|
||||
return Promise.reject(new Error("Invalid query!"));
|
||||
}
|
||||
} else {
|
||||
const q = new DocumentQuery(this, query.path, session);
|
||||
let type = query.type as IDocumentQueries;
|
||||
switch (type) {
|
||||
case "get":
|
||||
return q.get();
|
||||
case "set":
|
||||
return q.set(query.data, query.options || {});
|
||||
case "update":
|
||||
return q.update(query.data);
|
||||
case "delete":
|
||||
return q.delete();
|
||||
default:
|
||||
return Promise.reject(new Error("Invalid query!"));
|
||||
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 result[0]
|
||||
}
|
||||
|
||||
async snapshot(query: ITypedQuery<"snapshot">, session: Session, onchange: (change: any) => void) {
|
||||
this.validate(query);
|
||||
async snapshot(rawQuery: ITypedQuery<"snapshot">, session: Session, onchange: (change: any) => void) {
|
||||
Logging.debug("Snaphot request:", rawQuery.path);
|
||||
this.validate(rawQuery);
|
||||
|
||||
const isCollection = query.path.length % 2 === 1;
|
||||
let q: DocumentQuery | CollectionQuery;
|
||||
if (isCollection) {
|
||||
q = new CollectionQuery(this, query.path, session);
|
||||
const limit = (query.options || {}).limit;
|
||||
if (limit)
|
||||
q.limit = limit;
|
||||
const where = (query.options || {}).where;
|
||||
if (where)
|
||||
q.where = where;
|
||||
} else {
|
||||
q = new DocumentQuery(this, query.path, session);
|
||||
}
|
||||
if (rawQuery.type !== "snapshot")
|
||||
throw new Error("Invalid query type!");
|
||||
|
||||
const id = nanoid(16);
|
||||
session.queries.set(id, q);
|
||||
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: await q.snapshot(onchange)
|
||||
snaphot: value
|
||||
};
|
||||
}
|
||||
|
||||
async unsubscribe(id: string, session: Session) {
|
||||
let query: CollectionQuery | DocumentQuery = session.queries.get(id) as any;
|
||||
let query = session.subscriptions.get(id);
|
||||
if (query) {
|
||||
query.unsubscribe();
|
||||
session.queries.delete(id);
|
||||
query();
|
||||
session.subscriptions.delete(id);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user