Database, eventbus, pluginloader fixes

src/backend/Admin.ts

@@ -5,15 +5,13 @@ import { promises as fs, mkdirSync } from "fs"
 import { RPCServer } from 'rpclibrary/js/src/Backend'
 import { AdminConf, TableDefiniton } from './Types';
 import { RPCConfigLoader } from './RPCConfigLoader';
-
+import { RPCPluginLoader } from './PluginLoader';
 import * as Path from 'path'
-
 import Knex = require('knex');
 import http = require('http');
 import express = require('express');
 import { TableDefinitionExporter } from './Interfaces';
 import { FrontworkEventBus } from './Eventbus';
-import { Plugin } from './Plugin';
 
 const logger = getLogger("admin", 'debug') 
 
@@ -22,14 +20,12 @@ implements TableDefinitionExporter {
 
     private express
     private httpServer
+    private pluginLoader:RPCPluginLoader = new RPCPluginLoader(this)
     private eventBus:FrontworkEventBus = new FrontworkEventBus(this)
-    private plugins: Plugin[] = []
     config: RPCConfigLoader<AdminConf>
     knex:Knex
 
-    constructor(){
-        this.eventBus = new FrontworkEventBus(this)
-    }
+    constructor(){}
 
     async start(){
         this.initConfig()
@@ -74,7 +70,8 @@ implements TableDefinitionExporter {
     private startWebsocket(){
         new RPCServer(20000, [
             this.config,
-            ...this.plugins
+            this.pluginLoader,
+            this.eventBus
         ])
     }
 
@@ -134,7 +131,7 @@ implements TableDefinitionExporter {
         logger.info("Webserver stopped")
     }
 
-    public async makeKnex():Promise<Knex>{
+    async makeKnex():Promise<Knex>{
         const conf:Knex.Config = this.config.getConfigKey("dbConf")
 
         logger.debug("Making new knex:", conf)        
@@ -157,7 +154,7 @@ implements TableDefinitionExporter {
     getTableDefinitions(): TableDefiniton[]{
         return [
             this.eventBus,
-            ...this.plugins
+            ...this.pluginLoader.getPlugins()
         ].flatMap(exporter => exporter.getTableDefinitions())
     }
 }

src/backend/Eventbus.ts

@@ -1,8 +1,10 @@
 import { RPCExporter } from "rpclibrary/js/src/Interfaces";
-import { SubscriptionResponse, ErrorResponse } from "rpclibrary/js/src/Types";
-import { makeSubResponse } from "rpclibrary/js/src/Utils";
+import { SubscriptionResponse, ErrorResponse, SuccessResponse } from "rpclibrary/js/src/Types";
 import { FrontworkAdmin } from "./Admin";
 import { TableDefinitionExporter } from "./Interfaces";
+import { getLogger } from 'frontblock-generic/Types';
+
+import * as uuid from 'uuid/v4'
 
 export type NotificationSeverity = 'Info' | 'Important' | 'Error'
 
@@ -22,25 +24,63 @@ export type EventbusIfc = {
     }
 } 
 
+const logger = getLogger("Eventbus", 'debug') 
+
 export class FrontworkEventBus 
 implements RPCExporter<EventbusIfc, "Eventbus">, TableDefinitionExporter {
     name = "Eventbus" as "Eventbus" 
-    
+    private subscriptions : { [uid in string]:Function } = {}
+
     constructor(private admin: FrontworkAdmin){
-        this.admin
+    }
+
+    async subscribeNotifications(callback) : Promise<SubscriptionResponse>{
+        const uid = uuid()
+        this.subscriptions[uid] = callback
+        return { result: 'Success', uuid: uid }
+    }
+
+    async getNotificationLog() : Promise<Notification[]>{
+        try{
+            return await this.admin.knex.select('*').from('notifications')
+        }catch(e){
+            logger.error(e)
+            throw e
+        }
+    }
+
+    async unsubscribeNotifications(uid:string) : Promise<SuccessResponse | ErrorResponse>{
+        if(!this.subscriptions[uid]) return { result: 'Error', message: "Unknown subscription" }
+        delete this.subscriptions[uid]
+        return { result: 'Success' }
+    }
+
+    async pushNotification(notification: Notification){
+        if(!notification.time) notification.time = Date.now()
+        logger.debug("inserting into notifications", notification)
+        try{
+            await this.admin.knex('notifications').insert(notification)
+        }catch(e){
+            logger.error(e)
+            throw e
+        }
+ 
+        Object.values(this.subscriptions).forEach(callback => {
+            callback(notification)
+        })       
     }
 
     exportRPCs(){
         return [{
             name: 'getNotificationLog' as 'getNotificationLog',
-            call: async () => []
+            call: async () => await this.getNotificationLog()
         },{
             name: 'pushNotification' as 'pushNotification',
-            call: async () => {}
+            call: async (notification: Notification) => { return await this.pushNotification(notification) }
         },{
             name: 'subscribeNotificaitons' as 'subscribeNotifications',
             hook: async (callback: Function) => {
-                return makeSubResponse({})
+                return await this.subscribeNotifications(callback)
             }
         }]
     }

src/backend/PluginLoader.ts

@@ -5,7 +5,7 @@ import { Plugin } from "./Plugin";
 import { FrontworkAdmin } from "./Admin";
 
 class PluginLoader {
-    private runningPlugins: Plugin[]
+    private runningPlugins: Plugin[] = []
     private pluginUpdaters:{[name in string]:Git.Updater} = {}
 
     constructor(private admin:FrontworkAdmin){}
@@ -33,9 +33,9 @@ class PluginLoader {
         //logger.warn("Cloning fb-dist/"+name+".git into ./plugins/"+name+" ..."+(force?" USING FORCE!":""))
         this.pluginUpdaters[name] = new Git.Updater({
             schema: 'https',
-            localPath: './dist',
+            localPath: './plugins/'+name,
             remoteHost: 'www.versioncontrol.me',
-            remotePath: 'frontwork-distribution',
+            remotePath: 'frontblock-distribution',
             repoName: name
         })
         const status = await this.pluginUpdaters[name].cloneRepo(force)
@@ -63,7 +63,7 @@ class PluginLoader {
 
         let evalstr = "../plugins/"+name+"/Plugin"
         const pluginClass = await eval('require')(evalstr)
-        const pluginObj = new pluginClass.default(this)
+        const pluginObj = new pluginClass.default(this.admin)
         try{
             if(pluginObj.start)
                 await pluginObj.start()
@@ -137,7 +137,7 @@ export type PluginLoaderIfc = {
     }
 }
 
-class RPCPluginLoader 
+export class RPCPluginLoader 
 extends PluginLoader
 implements RPCExporter<PluginLoaderIfc, "PluginLoader">{