feat: neDB persistent datastore

dashboard/.gitignore

@@ -3,3 +3,5 @@ node_modules/
 # Prevent test coverage reports from being added
 .coverage
 .nyc_output
+
+/lib/data

dashboard/bin/index.js

@@ -19,7 +19,7 @@ program
     )
     .option('-d, --daemonize', 'Run the server as a daemon.')
     .option('-t, --test', 'Run the CLI without any actions for unit tests.')
-    .action(() => {
+    .action(async () => {
         if (program.opts().test) {
             return;
         }
@@ -31,7 +31,7 @@ program
             return;
         }
 
-        Server.init();
+        await Server.init();
     });
 
 // Run this script if this is a direct stdin.

dashboard/lib/controllers/batchQueue/index.js

@@ -0,0 +1,38 @@
+class BatchQueue {
+    constructor(queueProcessor) {
+        this.queue = [];
+        this.processing = false;
+        this.queueProcessor = queueProcessor;
+    }
+
+    add(data) {
+        this.queue.push(data);
+        this.processQueue();
+    }
+
+    async processQueue() {
+        if (this.queue.length == 0) {
+            this.processing = false;
+            return;
+        }
+
+        if (this.processing) return;
+
+        this.processing = true;
+        const currentQueue = this.queue;
+        this.queue = [];
+
+        await currentQueue.reduce(async (prevPromise, item, index) => {
+            await prevPromise;
+            await this.queueProcessor(item);
+
+            // at last, mark processing as complete so that new batch is saved
+            if (index == currentQueue.length - 1) {
+                this.processing = false;
+                this.processQueue();
+            }
+        }, Promise.resolve());
+    }
+}
+
+module.exports = BatchQueue;

dashboard/lib/controllers/run.js

@@ -1,83 +1,101 @@
 const Run = require('../models/run');
-const db = require('../store');
+const Event = require('../models/event');
+const RunStat = require('../models/runStat');
+const BatchQueue = require('./batchQueue');
+
+// queues for batching stats and events
+let statsBatch, eventsBatch;
 
 const api = {
     insert: async (data) => {
-        const runs = await db.getTable('runs');
-        await runs.insert(data.id, new Run(data));
-    },
-
-    findOne: async (id) => {
-        const runs = await db.getTable('runs');
-
-        if (!id) throw new TypeError('Invalid id');
-        return runs.findOne(id);
-    },
+        if (!data.id) throw new TypeError('Invalid run data');
 
-    find: async () => {
-        const runs = await db.getTable('runs');
-        return runs.find();
-    },
+        const runDocument = Run.create(data);
+        const run = await runDocument.save();
 
-    clear: async () => {
-        const runs = await db.getTable('runs');
-        return runs.clear();
+        if (!run)
+            throw new Error('Unable to save run data. Incoherent schema.');
     },
 
     pause: async (id) => {
         if (!id) throw new TypeError('Invalid id');
-        const run = await api.findOne(id);
+        const run = await Run.findOne({ _id: id });
 
         if (!run) throw new Error('Run not found.');
-        run.setPaused();
+
+        run.status = 'paused';
+        await run.save();
     },
 
     resume: async (id) => {
         if (!id) throw new TypeError('Invalid id');
-        const run = await api.findOne(id);
+        const run = await Run.findOne({ _id: id });
 
         if (!run) throw new Error('Run not found.');
-        run.setActive();
+
+        run.status = 'active';
+        await run.save();
     },
 
     abort: async (id) => {
         if (!id) throw new TypeError('Invalid id');
-        const run = await api.findOne(id);
+        const run = await Run.findOne({ _id: id });
 
         if (!run) throw new Error('Run not found.');
-        run.setAborted();
+
+        run.status = 'aborted';
+        run.endTime = Date.now();
+        await run.save();
     },
 
     done: async (id) => {
         if (!id) throw new TypeError('Invalid id');
-        const run = await api.findOne(id);
+        const run = await Run.findOne({ _id: id });
 
         if (!run) throw new Error('Run not found.');
-        run.setFinished();
+
+        if (run.status === 'aborted') return;
+
+        run.status = 'finished';
+        run.endTime = Date.now();
+        await run.save();
     },
 
     interrupt: async (id) => {
         if (!id) throw new TypeError('Invalid id');
-        const run = await api.findOne(id);
+        const run = await Run.findOne({ _id: id });
 
         if (!run) throw new Error('Run not found.');
-        run.setInterrupted();
+
+        run.status = 'interrupted';
+        await run.save();
     },
 
-    addEvent: async (data) => {
-        if (!data.id) throw new TypeError('Invalid id');
-        const run = await api.findOne(data.id);
+    saveEvent: async (data) => {
+        if (!data.runId) throw new TypeError('Invalid parent id');
 
-        if (!run) throw new Error('Run not found.');
-        run.addEvent(data);
+        const event = Event.create(data);
+        if (!event) throw new Error('Invalid event type');
+        await event.save();
     },
 
-    addStats: async (data) => {
-        if (!data.id) throw new TypeError('Invalid id');
-        const run = await api.findOne(data.id);
+    saveStats: async (data) => {
+        if (!data.runId) throw new TypeError('Invalid parent id');
 
-        if (!run) throw new Error('Run not found.');
-        run.addRunStats(data);
+        const stat = RunStat.create(data);
+        if (!stat) throw new Error('Invalid stat type');
+
+        await stat.save();
+    },
+
+    addEvent: (data) => {
+        eventsBatch = eventsBatch || new BatchQueue(api.saveEvent);
+        eventsBatch.add(data);
+    },
+
+    addStats: (data) => {
+        statsBatch = statsBatch || new BatchQueue(api.saveStats);
+        statsBatch.add(data);
     },
 };
 

dashboard/lib/models/event.js

@@ -1,14 +1,15 @@
-class Event {
-    constructor(data) {
-        this._init(data);
-    }
+const Document = require('camo').Document;
+
+class Event extends Document {
+    constructor() {
+        super();
 
-    _init(data) {
-        this.type = data.type;
-        this.parentId = data.id;
-        this.time = Date.now();
-        this.args = data.args;
-        this.err = data.err;
+        this.type = String;
+        this.parentId = String;
+        this.time = Number;
+        this.args = [String];
+        this.err = String;
+        this.runId = String;
     }
 }
 

dashboard/lib/models/run.js

@@ -1,86 +1,28 @@
+const Document = require('camo').Document;
 const Event = require('./event');
+const RunStat = require('./runStat');
 
-const RUN_STATUS = {
-    ACTIVE: 'active',
-    PAUSED: 'paused',
-    FINISHED: 'finished',
-    ABORTED: 'aborted',
-    INTERRUPTED: 'interrupted',
-};
+class Run extends Document {
+    constructor() {
+        super();
 
-class Run {
-    constructor(data) {
-        this._init(data);
+        this.command = String;
+        this._id = String;
+        this.startTime = Date;
+        this.endTime = {
+            type: Date,
+            default: 0,
+        };
+        this.status = String;
     }
 
-    _init(data) {
-        this.command = data.command;
-        this.id = data.id;
-
-        this.startTime = data.startTime;
-        this.endTime = 0;
-
-        this.events = [];
-        this.cpuUsage = [];
-        this.memoryUsage = [];
-    }
-
-    // setters
-    setPaused() {
-        this.status = RUN_STATUS.PAUSED;
-    }
-
-    setFinished() {
-        if (this.isAborted()) return;
-
-        this.status = RUN_STATUS.FINISHED;
-        this.endTime = Date.now();
-    }
-
-    setActive() {
-        this.status = RUN_STATUS.ACTIVE;
-    }
-
-    setAborted() {
-        this.status = RUN_STATUS.ABORTED;
-        this.endTime = Date.now();
-    }
-
-    setInterrupted() {
-        this.status = RUN_STATUS.INTERRUPTED;
-    }
-
-    addEvent(data) {
-        if (data.err) this.setInterrupted();
-        this.events.push(new Event(data));
-    }
-
-    addRunStats(data) {
-        data.cpu &&
-            data.memory &&
-            this.cpuUsage.push(data.cpu) &&
-            this.memoryUsage.push(data.memory);
-    }
-
-    // getters
-    isPaused() {
-        return this.status === RUN_STATUS.PAUSED;
-    }
-
-    isFinished() {
-        return this.status === RUN_STATUS.FINISHED;
-    }
-
-    isActive() {
-        return this.status === RUN_STATUS.ACTIVE;
-    }
-
-    isAborted() {
-        return this.status === RUN_STATUS.ABORTED;
+    async populate() {
+        this.stats = await RunStat.find({ runId: this._id });
+        this.events = await Event.find({ runId: this._id }, { sort: 'time' });
     }
 
-    isInterrupted() {
-        return this.status === RUN_STATUS.INTERRUPTED;
+    static collectionName() {
+        return 'runs';
     }
 }
 

dashboard/lib/models/runStat.js

@@ -0,0 +1,13 @@
+const Document = require('camo').Document;
+
+class RunStat extends Document {
+    constructor() {
+        super();
+
+        this.cpu = Number;
+        this.memory = Number;
+        this.runId = String;
+    }
+}
+
+module.exports = RunStat;

dashboard/lib/store/index.js

@@ -1,17 +1,16 @@
-const Table = require('./table');
+const connect = require('camo').connect;
+const paths = require('env-paths')('newman-dashboard');
+const uri = `nedb://${paths.data}`;
 
-const cache = {};
+const init = async () => {
+    try {
+        await connect(uri);
 
-const api = {
-    _createTable: (tableName) => {
-        cache[tableName] = new Table({});
-        return;
-    },
-
-    getTable: async (tableName) => {
-        if (!cache.hasOwnProperty(tableName)) api._createTable(tableName);
-        return cache[tableName];
-    },
+        // cleanup function to terminate db connection
+        return () => process.exit(0);
+    } catch (e) {
+        console.log('Error in connecting to database.');
+    }
 };
 
-module.exports = api;
+module.exports = { init };