wip: removing EventBus

[ci skip]

Author: tegefaulkes

src/PolykeyAgent.ts

@@ -31,11 +31,11 @@ import Discovery from './discovery/Discovery';
 import SessionManager from './sessions/SessionManager';
 import IdentitiesManager from './identities/IdentitiesManager';
 import { providers } from './identities';
-import { EventBus, captureRejectionSymbol } from './events';
 import config from './config';
 import * as errors from './errors';
 import * as utils from './utils';
 import * as keysUtils from './keys/utils';
+import * as keysEvents from './keys/events';
 import * as nodesUtils from './nodes/utils';
 import * as nodesEvents from './nodes/events';
 import * as workersUtils from './workers/utils';
@@ -99,15 +99,6 @@ interface PolykeyAgent extends CreateDestroyStartStop {}
   new errors.ErrorPolykeyAgentDestroyed(),
 )
 class PolykeyAgent {
-  /**
-   * Event symbols
-   * These represent event topics
-   */
-  public static readonly eventSymbols = {
-    [CertManager.name]: Symbol(CertManager.name),
-  } as {
-    readonly CertManager: unique symbol;
-  };
 
   /**
    * Create the Polykey Agent.
@@ -240,9 +231,6 @@ class PolykeyAgent {
     const dbPath = path.join(statePath, config.paths.dbBase);
     const keysPath = path.join(statePath, config.paths.keysBase);
     const vaultsPath = path.join(statePath, config.paths.vaultsBase);
-    const events = new EventBus({
-      captureRejections: true,
-    });
     let pkAgentProm: PromiseDeconstructed<PolykeyAgent> | undefined;
     try {
       status =
@@ -322,8 +310,6 @@ class PolykeyAgent {
           taskManager,
           certDuration: optionsDefaulted.keys.certDuration,
           certRenewLeadTime: optionsDefaulted.keys.certRenewLeadTime,
-          changeCallback: async (data) =>
-            events.emitAsync(PolykeyAgent.eventSymbols.CertManager, data),
           logger: logger.getChild(CertManager.name),
           fresh,
         }));
@@ -574,7 +560,6 @@ class PolykeyAgent {
       rpcServerClient,
       webSocketServerClient,
       rpcServerAgent,
-      events,
       fs,
       logger,
     });
@@ -614,7 +599,6 @@ class PolykeyAgent {
   public readonly vaultManager: VaultManager;
   public readonly notificationsManager: NotificationsManager;
   public readonly sessionManager: SessionManager;
-  public readonly events: EventBus;
   public readonly fs: FileSystem;
   public readonly logger: Logger;
   public readonly rpcServerClient: RPCServer;
@@ -649,7 +633,6 @@ class PolykeyAgent {
     rpcServerClient,
     webSocketServerClient,
     rpcServerAgent,
-    events,
     fs,
     logger,
   }: {
@@ -674,7 +657,6 @@ class PolykeyAgent {
     rpcServerClient: RPCServer;
     webSocketServerClient: WebSocketServer;
     rpcServerAgent: RPCServer;
-    events: EventBus;
     fs: FileSystem;
     logger: Logger;
   }) {
@@ -700,7 +682,6 @@ class PolykeyAgent {
     this.rpcServerClient = rpcServerClient;
     this.webSocketServerClient = webSocketServerClient;
     this.rpcServerAgent = rpcServerAgent;
-    this.events = events;
     this.fs = fs;
   }
 
@@ -727,43 +708,26 @@ class PolykeyAgent {
     });
     try {
       this.logger.info(`Starting ${this.constructor.name}`);
-      // Set up error handling for event handlers
-      this.events[captureRejectionSymbol] = (err, event: symbol) => {
-        let msg = `EventBus error for ${event.toString()}`;
-        if (err instanceof errors.ErrorPolykey) {
-          msg += `: ${err.name}: ${err.description}`;
-          if (err.message !== '') {
-            msg += ` - ${err.message}`;
-          }
-        } else {
-          msg += `: ${err.name}`;
-          if (err.message !== '') {
-            msg += `: ${err.message}`;
-          }
-        }
-        this.logger.error(msg);
-        throw err;
-      };
       // Register event handlers
-      this.events.on(
-        PolykeyAgent.eventSymbols.CertManager,
-        async (data: CertManagerChangeData) => {
-          this.logger.info(`${KeyRing.name} change propagating`);
-          await this.status.updateStatusLive({
-            nodeId: data.nodeId,
-          });
-          await this.nodeManager.resetBuckets();
-          // Update the sigchain
-          await this.sigchain.onKeyRingChange();
-          const tlsConfig: TLSConfig = {
-            keyPrivatePem: keysUtils.privateKeyToPEM(data.keyPair.privateKey),
-            certChainPem: await this.certManager.getCertPEMsChainPEM(),
-          };
-          this.webSocketServerClient.setTlsConfig(tlsConfig);
-          this.nodeConnectionManager.updateTlsConfig(tlsConfig);
-          this.logger.info(`${KeyRing.name} change propagated`);
-        },
-      );
+      // FIXME: we need to handle the EventCertManagerCertChanged event to update the status
+      const handleCertChange = async (evt: keysEvents.EventsCertManagerCertChange) => {
+        const data = evt.detail
+        this.logger.info(`${KeyRing.name} change propagating`);
+        await this.status.updateStatusLive({
+          nodeId: data.nodeId,
+        });
+        await this.nodeManager.resetBuckets();
+        // Update the sigchain
+        await this.sigchain.onKeyRingChange();
+        const tlsConfig: TLSConfig = {
+          keyPrivatePem: keysUtils.privateKeyToPEM(data.keyPair.privateKey),
+          certChainPem: await this.certManager.getCertPEMsChainPEM(),
+        };
+        this.webSocketServerClient.setTlsConfig(tlsConfig);
+        this.nodeConnectionManager.updateTlsConfig(tlsConfig);
+        this.logger.info(`${KeyRing.name} change propagated`);
+      }
+      this.certManager.addEventListener(keysEvents.EventsCertManagerCertChange.name, handleCertChange);
       await this.status.start({ pid: process.pid });
       await this.schema.start({ fresh });
       // Starting modules
@@ -854,7 +818,7 @@ class PolykeyAgent {
       this.logger.warn(
         `Failed Starting ${this.constructor.name} with ${e.message}`,
       );
-      this.events.removeAllListeners();
+      this.certManager.removeEventListener(keysEvents.EventsCertManagerCertChange.name, handleCertChange);
       await this.status?.beginStop({ pid: process.pid });
       await this.taskManager?.stopProcessing();
       await this.taskManager?.stopTasks();
@@ -892,7 +856,7 @@ class PolykeyAgent {
    */
   public async stop() {
     this.logger.info(`Stopping ${this.constructor.name}`);
-    this.events.removeAllListeners();
+    this.certManager.removeEventListener(keysEvents.EventsCertManagerCertChange.name, handleCertChange);
     await this.status.beginStop({ pid: process.pid });
     await this.taskManager.stopProcessing();
     await this.taskManager.stopTasks();

src/keys/CertManager.ts

@@ -4,7 +4,6 @@ import type {
   PrivateKey,
   Certificate,
   CertificateASN1,
-  CertManagerChangeData,
   CertificatePEM,
   KeyPair,
   RecoveryCode,
@@ -24,7 +23,9 @@ import {
 import { Lock } from '@matrixai/async-locks';
 import * as keysUtils from './utils';
 import * as keysErrors from './errors';
+import * as events from './events';
 import * as ids from '../ids';
+import config from '../config';
 
 /**
  * This signal reason indicates we want to stop the renewal
@@ -37,6 +38,14 @@ interface CertManager extends CreateDestroyStartStop {}
 @CreateDestroyStartStop(
   new keysErrors.ErrorCertManagerRunning(),
   new keysErrors.ErrorCertManagerDestroyed(),
+  {
+    eventStart: events.EventsCertManagerStart,
+    eventStarted: events.EventsCertManagerStarted,
+    eventStop: events.EventsCertManagerStop,
+    eventStopped: events.EventsCertManagerStopped,
+    eventDestroy: events.EventsCertManagerDestroy,
+    eventDestroyed: events.EventsCertManagerDestroyed,
+  },
 )
 class CertManager {
   /**
@@ -57,9 +66,8 @@ class CertManager {
     db,
     keyRing,
     taskManager,
-    certDuration = 31536000,
-    certRenewLeadTime = 86400,
-    changeCallback,
+    certDuration = config.defaultsUser.certDuration,
+    certRenewLeadTime = config.defaultsUser.certRenewLeadTime,
     workerManager,
     logger = new Logger(this.name),
     subjectAttrsExtra,
@@ -72,7 +80,6 @@ class CertManager {
     taskManager: TaskManager;
     certDuration?: number;
     certRenewLeadTime?: number;
-    changeCallback?: (data: CertManagerChangeData) => any;
     workerManager?: PolykeyWorkerManagerInterface;
     logger?: Logger;
     subjectAttrsExtra?: Array<{ [key: string]: Array<string> }>;
@@ -88,7 +95,6 @@ class CertManager {
       taskManager,
       certDuration,
       certRenewLeadTime,
-      changeCallback,
       workerManager,
       logger,
     });
@@ -112,7 +118,6 @@ class CertManager {
   protected db: DB;
   protected keyRing: KeyRing;
   protected taskManager: TaskManager;
-  protected changeCallback?: (data: CertManagerChangeData) => any;
   protected workerManager?: PolykeyWorkerManagerInterface;
   protected generateCertId: () => CertId;
   protected dbPath: LevelPath = [this.constructor.name];
@@ -147,7 +152,6 @@ class CertManager {
     taskManager,
     certDuration,
     certRenewLeadTime,
-    changeCallback,
     workerManager,
     logger,
   }: {
@@ -156,7 +160,6 @@ class CertManager {
     taskManager: TaskManager;
     certDuration: number;
     certRenewLeadTime: number;
-    changeCallback?: (data: CertManagerChangeData) => any;
     workerManager?: PolykeyWorkerManagerInterface;
     logger: Logger;
   }) {
@@ -166,7 +169,6 @@ class CertManager {
     this.taskManager = taskManager;
     this.certDuration = certDuration;
     this.certRenewLeadTime = certRenewLeadTime;
-    this.changeCallback = changeCallback;
     this.workerManager = workerManager;
   }
 
@@ -452,14 +454,14 @@ class CertManager {
       if (this.tasksRunning) {
         await this.setupRenewCurrentCertTask(now);
       }
-      if (this.changeCallback != null) {
-        await this.changeCallback({
+      this.dispatchEvent(new events.EventsCertManagerCertChange({
+        detail: {
           nodeId: this.keyRing.getNodeId(),
           keyPair: this.keyRing.keyPair,
           cert: certNew,
           recoveryCode: recoveryCodeNew!,
-        });
-      }
+        },
+      }));
       this.logger.info('Renewed certificate chain with new key pair');
     });
     return certNew!;
@@ -516,14 +518,14 @@ class CertManager {
       if (this.tasksRunning) {
         await this.setupRenewCurrentCertTask(now);
       }
-      if (this.changeCallback != null) {
-        await this.changeCallback({
+      this.dispatchEvent(new events.EventsCertManagerCertChange({
+        detail: {
           nodeId: this.keyRing.getNodeId(),
           keyPair: this.keyRing.keyPair,
           cert: certNew,
           recoveryCode: undefined,
-        });
-      }
+        },
+      }));
       this.logger.info('Renewed certificate chain with current key pair');
     });
     return certNew!;
@@ -587,14 +589,14 @@ class CertManager {
       if (this.tasksRunning) {
         await this.setupRenewCurrentCertTask(now);
       }
-      if (this.changeCallback != null) {
-        await this.changeCallback({
+      this.dispatchEvent(new events.EventsCertManagerCertChange({
+        detail: {
           nodeId: this.keyRing.getNodeId(),
           keyPair: this.keyRing.keyPair,
           cert: certNew!,
           recoveryCode: recoveryCodeNew!,
-        });
-      }
+        },
+      }));
       this.logger.info('Resetted certificate chain with new key pair');
     });
     return certNew!;
@@ -646,13 +648,14 @@ class CertManager {
       if (this.tasksRunning) {
         await this.setupRenewCurrentCertTask(now);
       }
-      if (this.changeCallback != null) {
-        await this.changeCallback({
+      this.dispatchEvent(new events.EventsCertManagerCertChange({
+        detail: {
           nodeId: this.keyRing.getNodeId(),
           keyPair: this.keyRing.keyPair,
           cert: certNew,
-        });
-      }
+          recoveryCode: undefined,
+        },
+      }));
       this.logger.info('Resetted certificate chain with current key pair');
     });
     return certNew!;

src/keys/events.ts

@@ -0,0 +1,32 @@
+import { AbstractEvent } from '@matrixai/events';
+import { CertManagerChangeData } from "./types";
+
+abstract class EventsKeys<T = null> extends AbstractEvent<T> {}
+
+abstract class EventsCertManager<T = null> extends EventsKeys<T> {}
+
+class EventsCertManagerStart extends EventsCertManager {}
+
+class EventsCertManagerStarted extends EventsCertManager {}
+
+class EventsCertManagerStop extends EventsCertManager {}
+
+class EventsCertManagerStopped extends EventsCertManager {}
+
+class EventsCertManagerDestroy extends EventsCertManager {}
+
+class EventsCertManagerDestroyed extends EventsCertManager {}
+
+class EventsCertManagerCertChange extends EventsCertManager<CertManagerChangeData> {}
+
+export {
+  EventsKeys,
+  EventsCertManager,
+  EventsCertManagerStart,
+  EventsCertManagerStarted,
+  EventsCertManagerStop,
+  EventsCertManagerStopped,
+  EventsCertManagerDestroy,
+  EventsCertManagerDestroyed,
+  EventsCertManagerCertChange,
+}