aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/client/DocServer.ts4
-rw-r--r--src/client/northstar/dash-nodes/HistogramBinPrimitiveCollection.ts4
-rw-r--r--src/client/northstar/dash-nodes/HistogramBoxPrimitives.tsx4
-rw-r--r--src/client/views/nodes/CollectionFreeFormDocumentView.tsx6
-rw-r--r--src/server/IDatabase.ts24
-rw-r--r--src/server/MemoryDatabase.ts99
-rw-r--r--src/server/RouteManager.ts5
-rw-r--r--src/server/Session/session.ts686
-rw-r--r--src/server/Websocket/Websocket.ts10
-rw-r--r--src/server/authentication/models/current_user_utils.ts2
-rw-r--r--src/server/database.ts19
-rw-r--r--src/server/index.ts16
-rw-r--r--src/server/server_Initialization.ts4
13 files changed, 856 insertions, 27 deletions
diff --git a/src/client/DocServer.ts b/src/client/DocServer.ts
index 12fed3e46..d793b56af 100644
--- a/src/client/DocServer.ts
+++ b/src/client/DocServer.ts
@@ -272,7 +272,7 @@ export namespace DocServer {
const fieldMap: { [id: string]: RefField } = {};
const proms: Promise<void>[] = [];
for (const field of fields) {
- if (field !== undefined) {
+ if (field !== undefined && field !== null) {
// deserialize
const prom = SerializationHelper.Deserialize(field).then(deserialized => {
fieldMap[field.id] = deserialized;
@@ -439,4 +439,4 @@ export namespace DocServer {
function respondToDelete(ids: string | string[]) {
_respondToDelete(ids);
}
-} \ No newline at end of file
+}
diff --git a/src/client/northstar/dash-nodes/HistogramBinPrimitiveCollection.ts b/src/client/northstar/dash-nodes/HistogramBinPrimitiveCollection.ts
index 3e9145a1b..6b36ffc9e 100644
--- a/src/client/northstar/dash-nodes/HistogramBinPrimitiveCollection.ts
+++ b/src/client/northstar/dash-nodes/HistogramBinPrimitiveCollection.ts
@@ -3,7 +3,7 @@ import { AttributeTransformationModel } from "../../northstar/core/attribute/Att
import { ChartType } from '../../northstar/model/binRanges/VisualBinRange';
import { AggregateFunction, Bin, Brush, DoubleValueAggregateResult, HistogramResult, MarginAggregateParameters, MarginAggregateResult } from "../../northstar/model/idea/idea";
import { ModelHelpers } from "../../northstar/model/ModelHelpers";
-import { LABColor } from '../../northstar/utils/LABcolor';
+import { LABColor } from '../../northstar/utils/LABColor';
import { PIXIRectangle } from "../../northstar/utils/MathUtil";
import { StyleConstants } from "../../northstar/utils/StyleContants";
import { HistogramBox } from "./HistogramBox";
@@ -237,4 +237,4 @@ export class HistogramBinPrimitiveCollection {
// }
return StyleConstants.HIGHLIGHT_COLOR;
}
-} \ No newline at end of file
+}
diff --git a/src/client/northstar/dash-nodes/HistogramBoxPrimitives.tsx b/src/client/northstar/dash-nodes/HistogramBoxPrimitives.tsx
index 5a16b3782..66d91cc1d 100644
--- a/src/client/northstar/dash-nodes/HistogramBoxPrimitives.tsx
+++ b/src/client/northstar/dash-nodes/HistogramBoxPrimitives.tsx
@@ -5,7 +5,7 @@ import { Utils as DashUtils, emptyFunction } from '../../../Utils';
import { FilterModel } from "../../northstar/core/filter/FilterModel";
import { ModelHelpers } from "../../northstar/model/ModelHelpers";
import { ArrayUtil } from "../../northstar/utils/ArrayUtil";
-import { LABColor } from '../../northstar/utils/LABcolor';
+import { LABColor } from '../../northstar/utils/LABColor';
import { PIXIRectangle } from "../../northstar/utils/MathUtil";
import { StyleConstants } from "../../northstar/utils/StyleContants";
import { HistogramBinPrimitiveCollection, HistogramBinPrimitive } from "./HistogramBinPrimitiveCollection";
@@ -119,4 +119,4 @@ export class HistogramBoxPrimitives extends React.Component<HistogramPrimitivesP
</svg>
</div>;
}
-} \ No newline at end of file
+}
diff --git a/src/client/views/nodes/CollectionFreeFormDocumentView.tsx b/src/client/views/nodes/CollectionFreeFormDocumentView.tsx
index 2e598a14d..f79496ab7 100644
--- a/src/client/views/nodes/CollectionFreeFormDocumentView.tsx
+++ b/src/client/views/nodes/CollectionFreeFormDocumentView.tsx
@@ -1,4 +1,4 @@
-import { random } from "animejs";
+import anime from "animejs";
import { computed, IReactionDisposer, observable, reaction, trace } from "mobx";
import { observer } from "mobx-react";
import { Doc, HeightSym, WidthSym } from "../../../new_fields/Doc";
@@ -26,7 +26,7 @@ export interface CollectionFreeFormDocumentViewProps extends DocumentViewProps {
export class CollectionFreeFormDocumentView extends DocComponent<CollectionFreeFormDocumentViewProps, PositionDocument>(PositionDocument) {
_disposer: IReactionDisposer | undefined = undefined;
get displayName() { return "CollectionFreeFormDocumentView(" + this.props.Document.title + ")"; } // this makes mobx trace() statements more descriptive
- get transform() { return `scale(${this.props.ContentScaling()}) translate(${this.X}px, ${this.Y}px) rotate(${random(-1, 1) * this.props.jitterRotation}deg)`; }
+ get transform() { return `scale(${this.props.ContentScaling()}) translate(${this.X}px, ${this.Y}px) rotate(${anime.random(-1, 1) * this.props.jitterRotation}deg)`; }
get X() { return this._animPos !== undefined ? this._animPos[0] : this.renderScriptDim ? this.renderScriptDim.x : this.props.x !== undefined ? this.props.x : this.dataProvider ? this.dataProvider.x : (this.Document.x || 0); }
get Y() { return this._animPos !== undefined ? this._animPos[1] : this.renderScriptDim ? this.renderScriptDim.y : this.props.y !== undefined ? this.props.y : this.dataProvider ? this.dataProvider.y : (this.Document.y || 0); }
get width() { return this.renderScriptDim ? this.renderScriptDim.width : this.props.width !== undefined ? this.props.width : this.props.dataProvider && this.dataProvider ? this.dataProvider.width : this.layoutDoc[WidthSym](); }
@@ -124,4 +124,4 @@ export class CollectionFreeFormDocumentView extends DocComponent<CollectionFreeF
/> */}
</div>;
}
-} \ No newline at end of file
+}
diff --git a/src/server/IDatabase.ts b/src/server/IDatabase.ts
new file mode 100644
index 000000000..6a63df485
--- /dev/null
+++ b/src/server/IDatabase.ts
@@ -0,0 +1,24 @@
+import * as mongodb from 'mongodb';
+import { Transferable } from './Message';
+
+export const DocumentsCollection = 'documents';
+export const NewDocumentsCollection = 'newDocuments';
+export interface IDatabase {
+ update(id: string, value: any, callback: (err: mongodb.MongoError, res: mongodb.UpdateWriteOpResult) => void, upsert?: boolean, collectionName?: string): Promise<void>;
+ updateMany(query: any, update: any, collectionName?: string): Promise<mongodb.WriteOpResult>;
+
+ replace(id: string, value: any, callback: (err: mongodb.MongoError, res: mongodb.UpdateWriteOpResult) => void, upsert?: boolean, collectionName?: string): void;
+
+ delete(query: any, collectionName?: string): Promise<mongodb.DeleteWriteOpResultObject>;
+ delete(id: string, collectionName?: string): Promise<mongodb.DeleteWriteOpResultObject>;
+
+ deleteAll(collectionName?: string, persist?: boolean): Promise<any>;
+
+ insert(value: any, collectionName?: string): Promise<void>;
+
+ getDocument(id: string, fn: (result?: Transferable) => void, collectionName?: string): void;
+ getDocuments(ids: string[], fn: (result: Transferable[]) => void, collectionName?: string): void;
+ visit(ids: string[], fn: (result: any) => string[] | Promise<string[]>, collectionName?: string): Promise<void>;
+
+ query(query: { [key: string]: any }, projection?: { [key: string]: 0 | 1 }, collectionName?: string): Promise<mongodb.Cursor>;
+}
diff --git a/src/server/MemoryDatabase.ts b/src/server/MemoryDatabase.ts
new file mode 100644
index 000000000..e7396babf
--- /dev/null
+++ b/src/server/MemoryDatabase.ts
@@ -0,0 +1,99 @@
+import { IDatabase, DocumentsCollection, NewDocumentsCollection } from './IDatabase';
+import { Transferable } from './Message';
+import * as mongodb from 'mongodb';
+
+export class MemoryDatabase implements IDatabase {
+
+ private db: { [collectionName: string]: { [id: string]: any } } = {};
+
+ private getCollection(collectionName: string) {
+ let collection = this.db[collectionName];
+ if (collection) {
+ return collection;
+ } else {
+ return this.db[collectionName] = {};
+ }
+ }
+
+ public update(id: string, value: any, callback: (err: mongodb.MongoError, res: mongodb.UpdateWriteOpResult) => void, _upsert?: boolean, collectionName = DocumentsCollection): Promise<void> {
+ const collection = this.getCollection(collectionName);
+ if ("$set" in value) {
+ let currentVal = collection[id] ?? (collection[id] = {});
+ const val = value["$set"];
+ for (const key in val) {
+ const keys = key.split(".");
+ for (let i = 0; i < keys.length - 1; i++) {
+ const k = keys[i];
+ if (typeof currentVal[k] === "object") {
+ currentVal = currentVal[k];
+ } else {
+ currentVal[k] = {};
+ currentVal = currentVal[k];
+ }
+ }
+ currentVal[keys[keys.length - 1]] = val[key];
+ }
+ } else {
+ collection[id] = value;
+ }
+ callback(null as any, {} as any);
+ return Promise.resolve(undefined);
+ }
+
+ public updateMany(query: any, update: any, collectionName = NewDocumentsCollection): Promise<mongodb.WriteOpResult> {
+ throw new Error("Can't updateMany a MemoryDatabase");
+ }
+
+ public replace(id: string, value: any, callback: (err: mongodb.MongoError, res: mongodb.UpdateWriteOpResult) => void, upsert?: boolean, collectionName = DocumentsCollection): void {
+ this.update(id, value, callback, upsert, collectionName);
+ }
+
+ public delete(query: any, collectionName?: string): Promise<mongodb.DeleteWriteOpResultObject>;
+ public delete(id: string, collectionName?: string): Promise<mongodb.DeleteWriteOpResultObject>;
+ public delete(id: any, collectionName = DocumentsCollection): Promise<mongodb.DeleteWriteOpResultObject> {
+ const i = id.id ?? id;
+ delete this.getCollection(collectionName)[i];
+
+ return Promise.resolve({} as any);
+ }
+
+ public deleteAll(collectionName = DocumentsCollection, _persist = true): Promise<any> {
+ delete this.db[collectionName];
+ return Promise.resolve();
+ }
+
+ public insert(value: any, collectionName = DocumentsCollection): Promise<void> {
+ const id = value.id;
+ this.getCollection(collectionName)[id] = value;
+ return Promise.resolve();
+ }
+
+ public getDocument(id: string, fn: (result?: Transferable) => void, collectionName = NewDocumentsCollection): void {
+ fn(this.getCollection(collectionName)[id]);
+ }
+ public getDocuments(ids: string[], fn: (result: Transferable[]) => void, collectionName = DocumentsCollection): void {
+ fn(ids.map(id => this.getCollection(collectionName)[id]));
+ }
+
+ public async visit(ids: string[], fn: (result: any) => string[] | Promise<string[]>, collectionName = NewDocumentsCollection): Promise<void> {
+ const visited = new Set<string>();
+ while (ids.length) {
+ const count = Math.min(ids.length, 1000);
+ const index = ids.length - count;
+ const fetchIds = ids.splice(index, count).filter(id => !visited.has(id));
+ if (!fetchIds.length) {
+ continue;
+ }
+ const docs = await new Promise<{ [key: string]: any }[]>(res => this.getDocuments(fetchIds, res, collectionName));
+ for (const doc of docs) {
+ const id = doc.id;
+ visited.add(id);
+ ids.push(...(await fn(doc)));
+ }
+ }
+ }
+
+ public query(): Promise<mongodb.Cursor> {
+ throw new Error("Can't query a MemoryDatabase");
+ }
+}
diff --git a/src/server/RouteManager.ts b/src/server/RouteManager.ts
index b146d7b72..a7ee405a7 100644
--- a/src/server/RouteManager.ts
+++ b/src/server/RouteManager.ts
@@ -86,7 +86,10 @@ export default class RouteManager {
const { method, subscription, secureHandler: onValidation, publicHandler: onUnauthenticated, errorHandler: onError } = initializer;
const isRelease = this._isRelease;
const supervised = async (req: express.Request, res: express.Response) => {
- const { user, originalUrl: target } = req;
+ let { user, originalUrl: target } = req;
+ if (process.env.DB === "MEM" && !user) {
+ user = { id: "guest", email: "", userDocumentId: "guestDocId" };
+ }
const core = { req, res, isRelease };
const tryExecute = async (toExecute: (args: any) => any | Promise<any>, args: any) => {
try {
diff --git a/src/server/Session/session.ts b/src/server/Session/session.ts
new file mode 100644
index 000000000..ec3d46ac1
--- /dev/null
+++ b/src/server/Session/session.ts
@@ -0,0 +1,686 @@
+import { red, cyan, green, yellow, magenta, blue, white, Color, grey, gray, black } from "colors";
+import { on, fork, setupMaster, Worker, isMaster, isWorker } from "cluster";
+import { get } from "request-promise";
+import { Utils } from "../../Utils";
+import Repl, { ReplAction } from "../repl";
+import { readFileSync } from "fs";
+import { validate, ValidationError } from "jsonschema";
+import { configurationSchema } from "./session_config_schema";
+import { exec, ExecOptions } from "child_process";
+
+/**
+ * This namespace relies on NodeJS's cluster module, which allows a parent (master) process to share
+ * code with its children (workers). A simple `isMaster` flag indicates who is trying to access
+ * the code, and thus determines the functionality that actually gets invoked (checked by the caller, not internally).
+ *
+ * Think of the master thread as a factory, and the workers as the helpers that actually run the server.
+ *
+ * So, when we run `npm start`, given the appropriate check, initializeMaster() is called in the parent process
+ * This will spawn off its own child process (by default, mirrors the execution path of its parent),
+ * in which initializeWorker() is invoked.
+ */
+export namespace Session {
+
+ type ColorLabel = "yellow" | "red" | "cyan" | "green" | "blue" | "magenta" | "grey" | "gray" | "white" | "black";
+ const colorMapping: Map<ColorLabel, Color> = new Map([
+ ["yellow", yellow],
+ ["red", red],
+ ["cyan", cyan],
+ ["green", green],
+ ["blue", blue],
+ ["magenta", magenta],
+ ["grey", grey],
+ ["gray", gray],
+ ["white", white],
+ ["black", black]
+ ]);
+
+ export abstract class AppliedSessionAgent {
+
+ // the following two methods allow the developer to create a custom
+ // session and use the built in customization options for each thread
+ protected abstract async launchMonitor(): Promise<Session.Monitor>;
+ protected abstract async launchServerWorker(): Promise<Session.ServerWorker>;
+
+ private launched = false;
+
+ public killSession = (reason: string, graceful = true, errorCode = 0) => {
+ const target = isMaster ? this.sessionMonitor : this.serverWorker;
+ target.killSession(reason, graceful, errorCode);
+ }
+
+ private sessionMonitorRef: Session.Monitor | undefined;
+ public get sessionMonitor(): Session.Monitor {
+ if (!isMaster) {
+ this.serverWorker.sendMonitorAction("kill", {
+ graceful: false,
+ reason: "Cannot access the session monitor directly from the server worker thread.",
+ errorCode: 1
+ });
+ throw new Error();
+ }
+ return this.sessionMonitorRef!;
+ }
+
+ private serverWorkerRef: Session.ServerWorker | undefined;
+ public get serverWorker(): Session.ServerWorker {
+ if (isMaster) {
+ throw new Error("Cannot access the server worker directly from the session monitor thread");
+ }
+ return this.serverWorkerRef!;
+ }
+
+ public async launch(): Promise<void> {
+ if (!this.launched) {
+ this.launched = true;
+ if (isMaster) {
+ this.sessionMonitorRef = await this.launchMonitor();
+ } else {
+ this.serverWorkerRef = await this.launchServerWorker();
+ }
+ } else {
+ throw new Error("Cannot launch a session thread more than once per process.");
+ }
+ }
+
+ }
+
+ interface Identifier {
+ text: string;
+ color: ColorLabel;
+ }
+
+ interface Identifiers {
+ master: Identifier;
+ worker: Identifier;
+ exec: Identifier;
+ }
+
+ interface Configuration {
+ showServerOutput: boolean;
+ identifiers: Identifiers;
+ ports: { [description: string]: number };
+ polling: {
+ route: string;
+ intervalSeconds: number;
+ failureTolerance: number;
+ };
+ }
+
+ const defaultConfig: Configuration = {
+ showServerOutput: false,
+ identifiers: {
+ master: {
+ text: "__monitor__",
+ color: "yellow"
+ },
+ worker: {
+ text: "__server__",
+ color: "magenta"
+ },
+ exec: {
+ text: "__exec__",
+ color: "green"
+ }
+ },
+ ports: { server: 3000 },
+ polling: {
+ route: "/",
+ intervalSeconds: 30,
+ failureTolerance: 0
+ }
+ };
+
+ export type ExitHandler = (reason: Error | boolean) => void | Promise<void>;
+
+ export namespace Monitor {
+
+ export interface NotifierHooks {
+ key?: (key: string) => (boolean | Promise<boolean>);
+ crash?: (error: Error) => (boolean | Promise<boolean>);
+ }
+
+ export interface Action {
+ message: string;
+ args: any;
+ }
+
+ export type ServerMessageHandler = (action: Action) => void | Promise<void>;
+
+ }
+
+ /**
+ * Validates and reads the configuration file, accordingly builds a child process factory
+ * and spawns off an initial process that will respawn as predecessors die.
+ */
+ export class Monitor {
+
+ private static count = 0;
+ private exitHandlers: ExitHandler[] = [];
+ private readonly notifiers: Monitor.NotifierHooks | undefined;
+ private readonly config: Configuration;
+ private onMessage: { [message: string]: Monitor.ServerMessageHandler[] | undefined } = {};
+ private activeWorker: Worker | undefined;
+ private key: string | undefined;
+ private repl: Repl;
+
+ public static Create(notifiers?: Monitor.NotifierHooks) {
+ if (isWorker) {
+ process.send?.({
+ action: {
+ message: "kill",
+ args: {
+ reason: "cannot create a monitor on the worker process.",
+ graceful: false,
+ errorCode: 1
+ }
+ }
+ });
+ process.exit(1);
+ } else if (++Monitor.count > 1) {
+ console.error(red("cannot create more than one monitor."));
+ process.exit(1);
+ } else {
+ return new Monitor(notifiers);
+ }
+ }
+
+ /**
+ * Kill this session and its active child
+ * server process, either gracefully (may wait
+ * indefinitely, but at least allows active networking
+ * requests to complete) or immediately.
+ */
+ public killSession = async (reason: string, graceful = true, errorCode = 0) => {
+ this.mainLog(cyan(`exiting session ${graceful ? "clean" : "immediate"}ly`));
+ this.mainLog(`session exit reason: ${(red(reason))}`);
+ await this.executeExitHandlers(true);
+ this.killActiveWorker(graceful, true);
+ process.exit(errorCode);
+ }
+
+ /**
+ * Execute the list of functions registered to be called
+ * whenever the process exits.
+ */
+ public addExitHandler = (handler: ExitHandler) => this.exitHandlers.push(handler);
+
+ /**
+ * Extend the default repl by adding in custom commands
+ * that can invoke application logic external to this module
+ */
+ public addReplCommand = (basename: string, argPatterns: (RegExp | string)[], action: ReplAction) => {
+ this.repl.registerCommand(basename, argPatterns, action);
+ }
+
+ public exec = (command: string, options?: ExecOptions) => {
+ return new Promise<void>(resolve => {
+ exec(command, { ...options, encoding: "utf8" }, (error, stdout, stderr) => {
+ if (error) {
+ this.execLog(red(`unable to execute ${white(command)}`));
+ error.message.split("\n").forEach(line => line.length && this.execLog(red(`(error) ${line}`)));
+ } else {
+ let outLines: string[], errorLines: string[];
+ if ((outLines = stdout.split("\n").filter(line => line.length)).length) {
+ outLines.forEach(line => line.length && this.execLog(cyan(`(stdout) ${line}`)));
+ }
+ if ((errorLines = stderr.split("\n").filter(line => line.length)).length) {
+ errorLines.forEach(line => line.length && this.execLog(yellow(`(stderr) ${line}`)));
+ }
+ }
+ resolve();
+ });
+ });
+ }
+
+ /**
+ * Add a listener at this message. When the monitor process
+ * receives a message, it will invoke all registered functions.
+ */
+ public addServerMessageListener = (message: string, handler: Monitor.ServerMessageHandler) => {
+ const handlers = this.onMessage[message];
+ if (handlers) {
+ handlers.push(handler);
+ } else {
+ this.onMessage[message] = [handler];
+ }
+ }
+
+ /**
+ * Unregister a given listener at this message.
+ */
+ public removeServerMessageListener = (message: string, handler: Monitor.ServerMessageHandler) => {
+ const handlers = this.onMessage[message];
+ if (handlers) {
+ const index = handlers.indexOf(handler);
+ if (index > -1) {
+ handlers.splice(index, 1);
+ }
+ }
+ }
+
+ /**
+ * Unregister all listeners at this message.
+ */
+ public clearServerMessageListeners = (message: string) => this.onMessage[message] = undefined;
+
+ private constructor(notifiers?: Monitor.NotifierHooks) {
+ this.notifiers = notifiers;
+
+ console.log(this.timestamp(), cyan("initializing session..."));
+
+ this.config = this.loadAndValidateConfiguration();
+
+ this.initializeSessionKey();
+ // determines whether or not we see the compilation / initialization / runtime output of each child server process
+ const output = this.config.showServerOutput ? "inherit" : "ignore";
+ setupMaster({ stdio: ["ignore", output, output, "ipc"] });
+
+ // handle exceptions in the master thread - there shouldn't be many of these
+ // the IPC (inter process communication) channel closed exception can't seem
+ // to be caught in a try catch, and is inconsequential, so it is ignored
+ process.on("uncaughtException", ({ message, stack }): void => {
+ if (message !== "Channel closed") {
+ this.mainLog(red(message));
+ if (stack) {
+ this.mainLog(`uncaught exception\n${red(stack)}`);
+ }
+ }
+ });
+
+ // a helpful cluster event called on the master thread each time a child process exits
+ on("exit", ({ process: { pid } }, code, signal) => {
+ const prompt = `server worker with process id ${pid} has exited with code ${code}${signal === null ? "" : `, having encountered signal ${signal}`}.`;
+ this.mainLog(cyan(prompt));
+ // to make this a robust, continuous session, every time a child process dies, we immediately spawn a new one
+ this.spawn();
+ });
+
+ this.repl = this.initializeRepl();
+ this.spawn();
+ }
+
+ /**
+ * Generates a blue UTC string associated with the time
+ * of invocation.
+ */
+ private timestamp = () => blue(`[${new Date().toUTCString()}]`);
+
+ /**
+ * A formatted, identified and timestamped log in color
+ */
+ public mainLog = (...optionalParams: any[]) => {
+ console.log(this.timestamp(), this.config.identifiers.master.text, ...optionalParams);
+ }
+
+ /**
+ * A formatted, identified and timestamped log in color for non-
+ */
+ private execLog = (...optionalParams: any[]) => {
+ console.log(this.timestamp(), this.config.identifiers.exec.text, ...optionalParams);
+ }
+
+ /**
+ * If the caller has indicated an interest
+ * in being notified of this feature, creates
+ * a GUID for this session that can, for example,
+ * be used as authentication for killing the server
+ * (checked externally).
+ */
+ private initializeSessionKey = async (): Promise<void> => {
+ if (this.notifiers?.key) {
+ this.key = Utils.GenerateGuid();
+ const success = await this.notifiers.key(this.key);
+ const statement = success ? green("distributed session key to recipients") : red("distribution of session key failed");
+ this.mainLog(statement);
+ }
+ }
+
+ /**
+ * At any arbitrary layer of nesting within the configuration objects, any single value that
+ * is not specified by the configuration is given the default counterpart. If, within an object,
+ * one peer is given by configuration and two are not, the one is preserved while the two are given
+ * the default value.
+ * @returns the composition of all of the assigned objects, much like Object.assign(), but with more
+ * granularity in the overwriting of nested objects
+ */
+ private preciseAssign = (target: any, ...sources: any[]): any => {
+ for (const source of sources) {
+ this.preciseAssignHelper(target, source);
+ }
+ return target;
+ }
+
+ private preciseAssignHelper = (target: any, source: any) => {
+ Array.from(new Set([...Object.keys(target), ...Object.keys(source)])).map(property => {
+ let targetValue: any, sourceValue: any;
+ if (sourceValue = source[property]) {
+ if (typeof sourceValue === "object" && typeof (targetValue = target[property]) === "object") {
+ this.preciseAssignHelper(targetValue, sourceValue);
+ } else {
+ target[property] = sourceValue;
+ }
+ }
+ });
+ }
+
+ /**
+ * Reads in configuration .json file only once, in the master thread
+ * and pass down any variables the pertinent to the child processes as environment variables.
+ */
+ private loadAndValidateConfiguration = (): Configuration => {
+ let config: Configuration;
+ try {
+ console.log(this.timestamp(), cyan("validating configuration..."));
+ config = JSON.parse(readFileSync('./session.config.json', 'utf8'));
+ const options = {
+ throwError: true,
+ allowUnknownAttributes: false
+ };
+ // ensure all necessary and no excess information is specified by the configuration file
+ validate(config, configurationSchema, options);
+ config = this.preciseAssign({}, defaultConfig, config);
+ } catch (error) {
+ if (error instanceof ValidationError) {
+ console.log(red("\nSession configuration failed."));
+ console.log("The given session.config.json configuration file is invalid.");
+ console.log(`${error.instance}: ${error.stack}`);
+ process.exit(0);
+ } else if (error.code === "ENOENT" && error.path === "./session.config.json") {
+ console.log(cyan("Loading default session parameters..."));
+ console.log("Consider including a session.config.json configuration file in your project root for customization.");
+ config = this.preciseAssign({}, defaultConfig);
+ } else {
+ console.log(red("\nSession configuration failed."));
+ console.log("The following unknown error occurred during configuration.");
+ console.log(error.stack);
+ process.exit(0);
+ }
+ } finally {
+ const { identifiers } = config!;
+ Object.keys(identifiers).forEach(key => {
+ const resolved = key as keyof Identifiers;
+ const { text, color } = identifiers[resolved];
+ identifiers[resolved].text = (colorMapping.get(color) || white)(`${text}:`);
+ });
+ return config!;
+ }
+ }
+
+ /**
+ * Builds the repl that allows the following commands to be typed into stdin of the master thread.
+ */
+ private initializeRepl = (): Repl => {
+ const repl = new Repl({ identifier: () => `${this.timestamp()} ${this.config.identifiers.master.text}` });
+ const boolean = /true|false/;
+ const number = /\d+/;
+ const letters = /[a-zA-Z]+/;
+ repl.registerCommand("exit", [/clean|force/], args => this.killSession("manual exit requested by repl", args[0] === "clean", 0));
+ repl.registerCommand("restart", [/clean|force/], args => this.killActiveWorker(args[0] === "clean"));
+ repl.registerCommand("set", [letters, "port", number, boolean], args => this.setPort(args[0], Number(args[2]), args[3] === "true"));
+ repl.registerCommand("set", [/polling/, number, boolean], args => {
+ const newPollingIntervalSeconds = Math.floor(Number(args[2]));
+ if (newPollingIntervalSeconds < 0) {
+ this.mainLog(red("the polling interval must be a non-negative integer"));
+ } else {
+ if (newPollingIntervalSeconds !== this.config.polling.intervalSeconds) {
+ this.config.polling.intervalSeconds = newPollingIntervalSeconds;
+ if (args[3] === "true") {
+ this.activeWorker?.send({ newPollingIntervalSeconds });
+ }
+ }
+ }
+ });
+ return repl;
+ }
+
+ private executeExitHandlers = async (reason: Error | boolean) => Promise.all(this.exitHandlers.map(handler => handler(reason)));
+
+ /**
+ * Attempts to kill the active worker gracefully, unless otherwise specified.
+ */
+ private killActiveWorker = (graceful = true, isSessionEnd = false): void => {
+ if (this.activeWorker && !this.activeWorker.isDead()) {
+ if (graceful) {
+ this.activeWorker.send({ manualExit: { isSessionEnd } });
+ } else {
+ this.activeWorker.process.kill();
+ }
+ }
+ }
+
+ /**
+ * Allows the caller to set the port at which the target (be it the server,
+ * the websocket, some other custom port) is listening. If an immediate restart
+ * is specified, this monitor will kill the active child and re-launch the server
+ * at the port. Otherwise, the updated port won't be used until / unless the child
+ * dies on its own and triggers a restart.
+ */
+ private setPort = (port: "server" | "socket" | string, value: number, immediateRestart: boolean): void => {
+ if (value > 1023 && value < 65536) {
+ this.config.ports[port] = value;
+ if (immediateRestart) {
+ this.killActiveWorker();
+ }
+ } else {
+ this.mainLog(red(`${port} is an invalid port number`));
+ }
+ }
+
+ /**
+ * Kills the current active worker and proceeds to spawn a new worker,
+ * feeding in configuration information as environment variables.
+ */
+ private spawn = (): void => {
+ const {
+ polling: {
+ route,
+ failureTolerance,
+ intervalSeconds
+ },
+ ports
+ } = this.config;
+ this.killActiveWorker();
+ this.activeWorker = fork({
+ pollingRoute: route,
+ pollingFailureTolerance: failureTolerance,
+ serverPort: ports.server,
+ socketPort: ports.socket,
+ pollingIntervalSeconds: intervalSeconds,
+ session_key: this.key,
+ DB: process.env.DB
+ });
+ this.mainLog(cyan(`spawned new server worker with process id ${this.activeWorker.process.pid}`));
+ // an IPC message handler that executes actions on the master thread when prompted by the active worker
+ this.activeWorker.on("message", async ({ lifecycle, action }) => {
+ if (action) {
+ const { message, args } = action as Monitor.Action;
+ console.log(this.timestamp(), `${this.config.identifiers.worker.text} action requested (${cyan(message)})`);
+ switch (message) {
+ case "kill":
+ const { reason, graceful, errorCode } = args;
+ this.killSession(reason, graceful, errorCode);
+ break;
+ case "notify_crash":
+ if (this.notifiers?.crash) {
+ const { error } = args;
+ const success = await this.notifiers.crash(error);
+ const statement = success ? green("distributed crash notification to recipients") : red("distribution of crash notification failed");
+ this.mainLog(statement);
+ }
+ break;
+ case "set_port":
+ const { port, value, immediateRestart } = args;
+ this.setPort(port, value, immediateRestart);
+ break;
+ }
+ const handlers = this.onMessage[message];
+ if (handlers) {
+ handlers.forEach(handler => handler({ message, args }));
+ }
+ }
+ if (lifecycle) {
+ console.log(this.timestamp(), `${this.config.identifiers.worker.text} lifecycle phase (${lifecycle})`);
+ }
+ });
+ }
+
+ }
+
+ /**
+ * Effectively, each worker repairs the connection to the server by reintroducing a consistent state
+ * if its predecessor has died. It itself also polls the server heartbeat, and exits with a notification
+ * email if the server encounters an uncaught exception or if the server cannot be reached.
+ */
+ export class ServerWorker {
+
+ private static count = 0;
+ private shouldServerBeResponsive = false;
+ private exitHandlers: ExitHandler[] = [];
+ private pollingFailureCount = 0;
+ private pollingIntervalSeconds: number;
+ private pollingFailureTolerance: number;
+ private pollTarget: string;
+ private serverPort: number;
+
+ public static Create(work: Function) {
+ if (isMaster) {
+ console.error(red("cannot create a worker on the monitor process."));
+ process.exit(1);
+ } else if (++ServerWorker.count > 1) {
+ process.send?.({
+ action: {
+ message: "kill", args: {
+ reason: "cannot create more than one worker on a given worker process.",
+ graceful: false,
+ errorCode: 1
+ }
+ }
+ });
+ process.exit(1);
+ } else {
+ return new ServerWorker(work);
+ }
+ }
+
+ /**
+ * Allows developers to invoke application specific logic
+ * by hooking into the exiting of the server process.
+ */
+ public addExitHandler = (handler: ExitHandler) => this.exitHandlers.push(handler);
+
+ /**
+ * Kill the session monitor (parent process) from this
+ * server worker (child process). This will also kill
+ * this process (child process).
+ */
+ public killSession = (reason: string, graceful = true, errorCode = 0) => this.sendMonitorAction("kill", { reason, graceful, errorCode });
+
+ /**
+ * A convenience wrapper to tell the session monitor (parent process)
+ * to carry out the action with the specified message and arguments.
+ */
+ public sendMonitorAction = (message: string, args?: any) => process.send!({ action: { message, args } });
+
+ private constructor(work: Function) {
+ this.lifecycleNotification(green(`initializing process... ${white(`[${process.execPath} ${process.execArgv.join(" ")}]`)}`));
+
+ const { pollingRoute, serverPort, pollingIntervalSeconds, pollingFailureTolerance } = process.env;
+ this.serverPort = Number(serverPort);
+ this.pollingIntervalSeconds = Number(pollingIntervalSeconds);
+ this.pollingFailureTolerance = Number(pollingFailureTolerance);
+ this.pollTarget = `http://localhost:${serverPort}${pollingRoute}`;
+
+ this.configureProcess();
+ work();
+ this.pollServer();
+ }
+
+ /**
+ * Set up message and uncaught exception handlers for this
+ * server process.
+ */
+ private configureProcess = () => {
+ // updates the local values of variables to the those sent from master
+ process.on("message", async ({ newPollingIntervalSeconds, manualExit }) => {
+ if (newPollingIntervalSeconds !== undefined) {
+ this.pollingIntervalSeconds = newPollingIntervalSeconds;
+ }
+ if (manualExit !== undefined) {
+ const { isSessionEnd } = manualExit;
+ await this.executeExitHandlers(isSessionEnd);
+ process.exit(0);
+ }
+ });
+
+ // one reason to exit, as the process might be in an inconsistent state after such an exception
+ process.on('uncaughtException', this.proactiveUnplannedExit);
+ process.on('unhandledRejection', reason => {
+ const appropriateError = reason instanceof Error ? reason : new Error(`unhandled rejection: ${reason}`);
+ this.proactiveUnplannedExit(appropriateError);
+ });
+ }
+
+ /**
+ * Execute the list of functions registered to be called
+ * whenever the process exits.
+ */
+ private executeExitHandlers = async (reason: Error | boolean) => Promise.all(this.exitHandlers.map(handler => handler(reason)));
+
+ /**
+ * Notify master thread (which will log update in the console) of initialization via IPC.
+ */
+ public lifecycleNotification = (event: string) => process.send?.({ lifecycle: event });
+
+ /**
+ * Called whenever the process has a reason to terminate, either through an uncaught exception
+ * in the process (potentially inconsistent state) or the server cannot be reached.
+ */
+ private proactiveUnplannedExit = async (error: Error): Promise<void> => {
+ this.shouldServerBeResponsive = false;
+ // communicates via IPC to the master thread that it should dispatch a crash notification email
+ this.sendMonitorAction("notify_crash", { error });
+ await this.executeExitHandlers(error);
+ // notify master thread (which will log update in the console) of crash event via IPC
+ this.lifecycleNotification(red(`crash event detected @ ${new Date().toUTCString()}`));
+ this.lifecycleNotification(red(error.message));
+ process.exit(1);
+ }
+
+ /**
+ * This monitors the health of the server by submitting a get request to whatever port / route specified
+ * by the configuration every n seconds, where n is also given by the configuration.
+ */
+ private pollServer = async (): Promise<void> => {
+ await new Promise<void>(resolve => {
+ setTimeout(async () => {
+ try {
+ await get(this.pollTarget);
+ if (!this.shouldServerBeResponsive) {
+ // notify monitor thread that the server is up and running
+ this.lifecycleNotification(green(`listening on ${this.serverPort}...`));
+ }
+ this.shouldServerBeResponsive = true;
+ } catch (error) {
+ // if we expect the server to be unavailable, i.e. during compilation,
+ // the listening variable is false, activeExit will return early and the child
+ // process will continue
+ if (this.shouldServerBeResponsive) {
+ if (++this.pollingFailureCount > this.pollingFailureTolerance) {
+ this.proactiveUnplannedExit(error);
+ } else {
+ this.lifecycleNotification(yellow(`the server has encountered ${this.pollingFailureCount} of ${this.pollingFailureTolerance} tolerable failures`));
+ }
+ }
+ } finally {
+ resolve();
+ }
+ }, 1000 * this.pollingIntervalSeconds);
+ });
+ // controlled, asynchronous infinite recursion achieves a persistent poll that does not submit a new request until the previous has completed
+ this.pollServer();
+ }
+
+ }
+
+}
diff --git a/src/server/Websocket/Websocket.ts b/src/server/Websocket/Websocket.ts
index 578147d60..6dda6956e 100644
--- a/src/server/Websocket/Websocket.ts
+++ b/src/server/Websocket/Websocket.ts
@@ -28,7 +28,7 @@ export namespace WebSocket {
function initialize(isRelease: boolean) {
const endpoint = io();
- endpoint.on("connection", function (socket: Socket) {
+ endpoint.on("connection", function(socket: Socket) {
_socket = socket;
socket.use((_packet, next) => {
@@ -83,7 +83,9 @@ export namespace WebSocket {
export async function deleteFields() {
await Database.Instance.deleteAll();
- await Search.clear();
+ if (process.env.DISABLE_SEARCH !== "true") {
+ await Search.clear();
+ }
await Database.Instance.deleteAll('newDocuments');
}
@@ -92,7 +94,9 @@ export namespace WebSocket {
await Database.Instance.deleteAll('newDocuments');
await Database.Instance.deleteAll('sessions');
await Database.Instance.deleteAll('users');
- await Search.clear();
+ if (process.env.DISABLE_SEARCH !== "true") {
+ await Search.clear();
+ }
}
function barReceived(socket: SocketIO.Socket, userEmail: string) {
diff --git a/src/server/authentication/models/current_user_utils.ts b/src/server/authentication/models/current_user_utils.ts
index 220c37e2b..36d4cd2f2 100644
--- a/src/server/authentication/models/current_user_utils.ts
+++ b/src/server/authentication/models/current_user_utils.ts
@@ -322,4 +322,4 @@ export class CurrentUserUtils {
};
return recurs([] as Attribute[], schema ? schema.rootAttributeGroup : undefined);
}
-} \ No newline at end of file
+}
diff --git a/src/server/database.ts b/src/server/database.ts
index 6e0771c11..83ce865c6 100644
--- a/src/server/database.ts
+++ b/src/server/database.ts
@@ -5,6 +5,8 @@ import { Utils, emptyFunction } from '../Utils';
import { DashUploadUtils } from './DashUploadUtils';
import { Credentials } from 'google-auth-library';
import { GoogleApiServerUtils } from './apis/google/GoogleApiServerUtils';
+import { IDatabase } from './IDatabase';
+import { MemoryDatabase } from './MemoryDatabase';
import * as mongoose from 'mongoose';
export namespace Database {
@@ -44,7 +46,7 @@ export namespace Database {
}
}
- class Database {
+ class Database implements IDatabase {
public static DocumentsCollection = 'documents';
private MongoClient = mongodb.MongoClient;
private currentWrites: { [id: string]: Promise<void> } = {};
@@ -215,7 +217,7 @@ export namespace Database {
if (!fetchIds.length) {
continue;
}
- const docs = await new Promise<{ [key: string]: any }[]>(res => Instance.getDocuments(fetchIds, res, "newDocuments"));
+ const docs = await new Promise<{ [key: string]: any }[]>(res => this.getDocuments(fetchIds, res, collectionName));
for (const doc of docs) {
const id = doc.id;
visited.add(id);
@@ -262,7 +264,16 @@ export namespace Database {
}
}
- export const Instance = new Database();
+ function getDatabase() {
+ switch (process.env.DB) {
+ case "MEM":
+ return new MemoryDatabase();
+ default:
+ return new Database();
+ }
+ }
+
+ export const Instance: IDatabase = getDatabase();
export namespace Auxiliary {
@@ -331,4 +342,4 @@ export namespace Database {
}
-} \ No newline at end of file
+}
diff --git a/src/server/index.ts b/src/server/index.ts
index 0cce0dc54..ef8ed9700 100644
--- a/src/server/index.ts
+++ b/src/server/index.ts
@@ -5,7 +5,7 @@ import * as path from 'path';
import { Database } from './database';
import { DashUploadUtils } from './DashUploadUtils';
import RouteSubscriber from './RouteSubscriber';
-import initializeServer from './server_initialization';
+import initializeServer from './server_Initialization';
import RouteManager, { Method, _success, _permission_denied, _error, _invalid, PublicHandler } from './RouteManager';
import * as qs from 'query-string';
import UtilManager from './ApiManagers/UtilManager';
@@ -41,11 +41,13 @@ async function preliminaryFunctions() {
await GoogleCredentialsLoader.loadCredentials();
GoogleApiServerUtils.processProjectCredentials();
await DashUploadUtils.buildFileDirectories();
- await log_execution({
- startMessage: "attempting to initialize mongodb connection",
- endMessage: "connection outcome determined",
- action: Database.tryInitializeConnection
- });
+ if (process.env.DB !== "MEM") {
+ await log_execution({
+ startMessage: "attempting to initialize mongodb connection",
+ endMessage: "connection outcome determined",
+ action: Database.tryInitializeConnection
+ });
+ }
}
/**
@@ -142,4 +144,4 @@ if (process.env.RELEASE) {
(sessionAgent = new DashSessionAgent()).launch();
} else {
launchServer();
-} \ No newline at end of file
+}
diff --git a/src/server/server_Initialization.ts b/src/server/server_Initialization.ts
index cbe070293..9f67c1dda 100644
--- a/src/server/server_Initialization.ts
+++ b/src/server/server_Initialization.ts
@@ -86,7 +86,7 @@ function buildWithMiddleware(server: express.Express) {
resave: true,
cookie: { maxAge: week },
saveUninitialized: true,
- store: new MongoStore({ url: Database.url })
+ store: process.env.DB === "MEM" ? new session.MemoryStore() : new MongoStore({ url: Database.url })
}),
flash(),
expressFlash(),
@@ -152,4 +152,4 @@ function registerCorsProxy(server: express.Express) {
});
}).pipe(res);
});
-} \ No newline at end of file
+}
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2025-10-06 17:28:10 +0000