fix: optin to websockets for the mediator live mode as an experiment,…

.vscode/launch.json

@@ -28,8 +28,10 @@
         "${fileBasenameNoExtension}",
         "--colors",
         "--workerThreads",
+        "--detectOpenHandles",
         "--maxWorkers",
-        "1"
+        "1",
+        "./tests"
       ],
       "skipFiles": [
         "${workspaceRoot}/../../node_modules/**/*",

src/prism-agent/Agent.ts

@@ -11,6 +11,7 @@ import {
   AgentCredentials as AgentCredentialsClass,
   AgentDIDHigherFunctions as AgentDIDHigherFunctionsClass,
   AgentInvitations as AgentInvitationsClass,
+  AgentOptions,
   EventCallback,
   InvitationType,
   ListenerKey,
@@ -86,8 +87,10 @@ export default class Agent
     public readonly mediationHandler: MediatorHandler,
     public readonly connectionManager: ConnectionsManager,
     public readonly seed: Domain.Seed = apollo.createRandomSeed().seed,
-    public readonly api: Domain.Api = new ApiImpl()
+    public readonly api: Domain.Api = new ApiImpl(),
+    options?: AgentOptions
   ) {
+
     this.pollux = new Pollux(castor);
     this.agentCredentials = new AgentCredentials(
       apollo,
@@ -104,7 +107,8 @@ export default class Agent
         pluto,
         this.agentCredentials,
         mediationHandler,
-        []
+        [],
+        options
       );
 
 
@@ -147,6 +151,7 @@ export default class Agent
     castor?: Domain.Castor;
     mercury?: Domain.Mercury;
     seed?: Domain.Seed;
+    options?: AgentOptions
   }): Agent {
     const mediatorDID = Domain.DID.from(params.mediatorDID);
     const pluto = params.pluto;
@@ -170,7 +175,15 @@ export default class Agent
       pollux,
       seed
     );
-    const manager = new ConnectionsManager(castor, mercury, pluto, agentCredentials, handler);
+    const manager = new ConnectionsManager(
+      castor,
+      mercury,
+      pluto,
+      agentCredentials,
+      handler,
+      [],
+      params.options
+    );
 
     const agent = new Agent(
       apollo,
@@ -180,7 +193,8 @@ export default class Agent
       handler,
       manager,
       seed,
-      api
+      api,
+      params.options
     );
 
     return agent;
@@ -217,7 +231,8 @@ export default class Agent
     mercury: Domain.Mercury,
     connectionManager: ConnectionsManager,
     seed?: Domain.Seed,
-    api?: Domain.Api
+    api?: Domain.Api,
+    options?: AgentOptions
   ) {
     return new Agent(
       apollo,
@@ -227,7 +242,8 @@ export default class Agent
       connectionManager.mediationHandler,
       connectionManager,
       seed ? seed : apollo.createRandomSeed().seed,
-      api ? api : new ApiImpl()
+      api ? api : new ApiImpl(),
+      options
     );
   }
 

src/prism-agent/connectionsManager/ConnectionsManager.ts

@@ -1,5 +1,4 @@
-import { uuid } from "@stablelib/uuid";
 import { DID, Message, MessageDirection, Pollux } from "../../domain";
 import { Castor } from "../../domain/buildingBlocks/Castor";
 import { Mercury } from "../../domain/buildingBlocks/Mercury";
 import { Pluto } from "../../domain/buildingBlocks/Pluto";
@@ -10,6 +9,7 @@
 import {
   AgentCredentials,
   AgentMessageEvents as AgentMessageEventsClass,
+  AgentOptions,
   ConnectionsManager as ConnectionsManagerClass,
   ListenerKey,
   MediatorHandler,
@@ -72,11 +72,16 @@
     public pluto: Pluto,
     public agentCredentials: AgentCredentials,
     public mediationHandler: MediatorHandler,
-    public pairings: DIDPair[] = []
+    public pairings: DIDPair[] = [],
+    public options?: AgentOptions
   ) {
     this.events = new AgentMessageEvents();
   }
 
+  get withWebsocketsExperiment() {
+    return this.options?.experiments?.liveMode === true
+  }
+
   /**
    * Asyncronously Start the mediator, just checking if we had one stored in Database and
    * setting that one as default during the Agent start
@@ -143,7 +148,7 @@
       const revokeMessages = messages.filter(x => x.piuri === ProtocolType.PrismRevocation);
       const allMessages = await this.pluto.getAllMessages();
 
       for (let message of revokeMessages) {
         const revokeMessage = RevocationNotification.fromMessage(message);
         const threadId = revokeMessage.body.issueCredentialProtocolThreadId;
 
@@ -153,7 +158,7 @@
         );
 
         if (matchingMessages.length > 0) {
           for (let message of matchingMessages) {
             const issueMessage = IssueCredential.fromMessage(message);
             const credential = await this.agentCredentials.processIssuedCredentialMessage(
               issueMessage
@@ -263,8 +268,10 @@
     const currentMediator = this.mediationHandler.mediator.mediatorDID;
     const resolvedMediator = await this.castor.resolveDID(currentMediator.toString());
     const hasWebsocket = resolvedMediator.services.find(({ serviceEndpoint: { uri } }) =>
-      uri.startsWith("ws://") ||
-      uri.startsWith("wss://")
+      (
+        uri.startsWith("ws://") ||
+        uri.startsWith("wss://")
+      ) && this.withWebsocketsExperiment
     );
     if (!hasWebsocket) {
       const timeInterval = Math.max(iterationPeriod, 5) * 1000;

src/prism-agent/helpers/Task.ts

@@ -1,9 +1,5 @@
 type Task<T> = (signal: AbortSignal) => Promise<T>;
 
-
-
-
-
 export class CancellableTask<T> {
   private period?: number;
   private controller: AbortController;

src/prism-agent/types/index.ts

@@ -6,7 +6,7 @@
   Service as DIDDocumentService,
   Signature,
   Credential,
   Pollux,
 } from "../../domain";
 import { DIDPair } from "../../domain/models/DIDPair";
 import { Castor } from "../../domain/buildingBlocks/Castor";
@@ -28,6 +28,13 @@
   PRISM_ONBOARD,
 }
 
+
+export type AgentOptions = {
+  experiments?: {
+    liveMode?: boolean
+  }
+}
+
 export type InvitationType = PrismOnboardingInvitation | OutOfBandInvitation;
 
 export class PrismOnboardingInvitation implements InvitationInterface {
@@ -111,6 +118,7 @@
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   cancellables: CancellableTask<any>[];
 
+  withWebsocketsExperiment: boolean;
   stopAllEvents(): void;
 
   addConnection(paired: DIDPair): Promise<void>;

tests/agent/Agent.ConnectionsManager.test.ts

@@ -0,0 +1,225 @@
+/**
+ * @jest-environment node
+ */
+import chai from "chai";
+import chaiAsPromised from "chai-as-promised";
+import SinonChai from "sinon-chai";
+import { Apollo, BasicMediatorHandler, Castor, ConnectionsManager, MediatorStore, Pluto } from "../../src";
+import { Curve, KeyTypes, Mercury, Service, ServiceEndpoint } from "../../src/domain";
+import { MercuryStub } from "./mocks/MercuryMock";
+import { AgentCredentials } from "../../src/prism-agent/Agent.Credentials";
+import { AgentOptions } from "../../src/prism-agent/types";
+
+chai.use(SinonChai);
+chai.use(chaiAsPromised);
+
+const store: MediatorStore = null as any;
+const mercury: Mercury = new MercuryStub();
+
+const apollo = new Apollo();
+const castor = new Castor(apollo)
+const pluto: Pluto = null as any;
+const agentCredentials: AgentCredentials = null as any;
+
+
+async function createBasicMediationHandler(
+    ConnectionsManager: any,
+    BasicMediatorHandler: any,
+    services: Service[],
+    options?: AgentOptions
+): Promise<
+    {
+        manager: ConnectionsManager,
+        handler: BasicMediatorHandler
+    }
+> {
+
+    const seed = apollo.createRandomSeed().seed;
+    const keypair = apollo.createPrivateKey({
+        type: KeyTypes.EC,
+        curve: Curve.SECP256K1,
+        seed: Buffer.from(seed.value).toString("hex"),
+    });
+    const mediatorDID = await castor.createPrismDID(keypair.publicKey(), services);
+    const handler = new BasicMediatorHandler(
+        mediatorDID,
+        mercury,
+        store
+    );
+    handler.mediator = {
+        hostDID: mediatorDID,
+        routingDID: mediatorDID,
+        mediatorDID: mediatorDID
+    }
+    const manager = new ConnectionsManager(
+        castor,
+        mercury,
+        pluto,
+        agentCredentials,
+        handler,
+        [],
+        options
+    )
+    return {
+        manager,
+        handler
+    }
+}
+
+
+describe("ConnectionsManager tests", () => {
+
+    beforeEach(() => {
+        jest.mock('isows', () => ({
+            WebSocket: jest.fn(() => ({
+                addEventListener: jest.fn(),
+                send: jest.fn(),
+                close: jest.fn(),
+            })),
+        }));
+    })
+
+    afterEach(() => {
+        jest.restoreAllMocks();
+    });
+
+    it("Should use websockets if the mediator's did endpoint uri contains ws or wss and agent options have the opt in", async () => {
+        const services = [
+            new Service(
+                "#didcomm-1",
+                ["DIDCommMessaging"],
+                new ServiceEndpoint("ws://localhost:12346")
+            )
+        ];
+        const ConnectionsManager = jest.requireActual('../../src/prism-agent/connectionsManager/ConnectionsManager').ConnectionsManager;
+        const BasicMediatorHandler = jest.requireMock('../../src/prism-agent/mediator/BasicMediatorHandler').BasicMediatorHandler;
+        const { manager, handler } = await createBasicMediationHandler(
+            ConnectionsManager,
+            BasicMediatorHandler,
+            services,
+            {
+                experiments: {
+                    liveMode: true
+                }
+            }
+        );
+        const listenUnread = jest.spyOn(handler, 'listenUnreadMessages')
+        expect(manager).toHaveProperty('withWebsocketsExperiment', true);
+        await manager.startFetchingMessages(1)
+        expect(listenUnread).toHaveBeenCalled();
+
+        manager.stopFetchingMessages()
+    })
+
+    it("Should not use websockets even if the mediator's did endpoint uri contains ws or wss if the agent options don't have the opt-in", async () => {
+        const services = [
+            new Service(
+                "#didcomm-1",
+                ["DIDCommMessaging"],
+                new ServiceEndpoint("ws://localhost:12346")
+            )
+        ];
+        const ConnectionsManager = jest.requireActual('../../src/prism-agent/connectionsManager/ConnectionsManager').ConnectionsManager;
+        const BasicMediatorHandler = jest.requireMock('../../src/prism-agent/mediator/BasicMediatorHandler').BasicMediatorHandler;
+        const { manager, handler } = await createBasicMediationHandler(
+            ConnectionsManager,
+            BasicMediatorHandler,
+            services
+        );
+        const listenUnread = jest.spyOn(handler, 'listenUnreadMessages')
+        expect(manager).toHaveProperty('withWebsocketsExperiment', false);
+
+        await manager.startFetchingMessages(1)
+
+        expect(listenUnread).not.toHaveBeenCalled()
+        manager.stopFetchingMessages()
+    })
+
+    it("Should not use websockets even if the mediator's did endpoint uri contains ws or wss if the agent options don't have the opt-in 1", async () => {
+        const services = [
+            new Service(
+                "#didcomm-1",
+                ["DIDCommMessaging"],
+                new ServiceEndpoint("ws://localhost:12346")
+            )
+        ];
+        const ConnectionsManager = jest.requireActual('../../src/prism-agent/connectionsManager/ConnectionsManager').ConnectionsManager;
+        const BasicMediatorHandler = jest.requireMock('../../src/prism-agent/mediator/BasicMediatorHandler').BasicMediatorHandler;
+        const { manager, handler } = await createBasicMediationHandler(
+            ConnectionsManager,
+            BasicMediatorHandler,
+            services,
+            {
+                experiments: {
+
+                }
+            }
+        );
+        const listenUnread = jest.spyOn(handler, 'listenUnreadMessages')
+        expect(manager).toHaveProperty('withWebsocketsExperiment', false);
+
+        await manager.startFetchingMessages(1)
+
+        expect(listenUnread).not.toHaveBeenCalled()
+        manager.stopFetchingMessages()
+    })
+
+    it("Should not use websockets even if the mediator's did endpoint uri contains ws or wss if the agent options don't have the opt-in 2", async () => {
+        const services = [
+            new Service(
+                "#didcomm-1",
+                ["DIDCommMessaging"],
+                new ServiceEndpoint("ws://localhost:12346")
+            )
+        ];
+        const ConnectionsManager = jest.requireActual('../../src/prism-agent/connectionsManager/ConnectionsManager').ConnectionsManager;
+        const BasicMediatorHandler = jest.requireMock('../../src/prism-agent/mediator/BasicMediatorHandler').BasicMediatorHandler;
+        const { manager, handler } = await createBasicMediationHandler(
+            ConnectionsManager,
+            BasicMediatorHandler,
+            services,
+            {
+                experiments: {
+                    liveMode: false
+                }
+            }
+        );
+        const listenUnread = jest.spyOn(handler, 'listenUnreadMessages')
+        expect(manager).toHaveProperty('withWebsocketsExperiment', false);
+
+        await manager.startFetchingMessages(1)
+
+        expect(listenUnread).not.toHaveBeenCalled()
+        manager.stopFetchingMessages()
+    })
+
+    it("Should not use websockets if the mediator'd did endpoint uri does not contain ws or wss for more than the agent has opted in", async () => {
+        const services = [
+            new Service(
+                "#didcomm-1",
+                ["DIDCommMessaging"],
+                new ServiceEndpoint("http://localhost:12346")
+            )
+        ];
+        const ConnectionsManager = jest.requireActual('../../src/prism-agent/connectionsManager/ConnectionsManager').ConnectionsManager;
+        const BasicMediatorHandler = jest.requireMock('../../src/prism-agent/mediator/BasicMediatorHandler').BasicMediatorHandler;
+        const { manager, handler } = await createBasicMediationHandler(
+            ConnectionsManager,
+            BasicMediatorHandler,
+            services,
+            {
+                experiments: {
+                    liveMode: true
+                }
+            }
+        );
+        const listenUnread = jest.spyOn(handler, 'listenUnreadMessages')
+        expect(manager).toHaveProperty('withWebsocketsExperiment', true);
+
+        await manager.startFetchingMessages(1)
+
+        expect(listenUnread).not.toHaveBeenCalled()
+        manager.stopFetchingMessages()
+    })
+
+})