8814828651

Committed 24 Apr 2024 10:09AM UTC coverage: 65.467% (-0.04%) from 65.502%

Build # 8814828651

Build Type

Pull #199

github

Committed by

web-flow

Commit Message

Merge 8ca1dbba5 into 11cc07cfd

Pull Request Pull Request #199: fix: optin to websockets for the mediator live mode as an experiment,…

Run Details

717 of 1268 branches covered (56.55%)

Branch coverage included in aggregate %.

1 of 3 new or added lines in 2 files covered. (33.33%)

38 existing lines in 2 files now uncovered.

1776 of 2540 relevant lines covered (69.92%)

12.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

28.21

/src/prism-agent/connectionsManager/ConnectionsManager.ts

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";
import { DIDPair } from "../../domain/models/DIDPair";
import { AgentError } from "../../domain/models/Errors";
import { AgentMessageEvents } from "../Agent.MessageEvents";
import { CancellableTask } from "../helpers/Task";
import {
  AgentCredentials,
  AgentMessageEvents as AgentMessageEventsClass,
  AgentOptions,
  ConnectionsManager as ConnectionsManagerClass,
  ListenerKey,
  MediatorHandler,
} from "../types";
import { ProtocolType } from "../protocols/ProtocolTypes";
import { RevocationNotification } from "../protocols/revocation/RevocationNotfiication";
import { IssueCredential } from "../protocols/issueCredential/IssueCredential";


/**
 * ConnectionsManager is responsible of establishing didcomm connection and
 * mediation process with other mediators through didcomm and is also
 * responsible of managing the task to periodically fetch messages from the mediator once connection is established
 *
 * @class ConnectionsManager
 * @typedef {ConnectionsManager}
 */
export class ConnectionsManager implements ConnectionsManagerClass {
  /**
   * An array with cancellable tasks, mainly used to store one or multiple didcomm
   * connections in storage implementation at the same time. All of them can be cancelled
   * despite they run asyncronously when the Edge agent stops
   *
   * @public
   * @type {CancellableTask<any>[]}
   */
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  public cancellables: CancellableTask<any>[] = [];
  /**
   * Cancellable task used to listen for new messages, stopping the Agent should also stop this
   *  from running and destroy the instance of the task until agent is started again
   *
   * @public
   * @type {?CancellableTask<void>}
   */
  public cancellable?: CancellableTask<void>;

  /**
   * A list of public facing events which will notify the user interface when specific things happen,
   * for now when new messages arrive or didcomm connections are established in order to make UI more reactive
   *
   * @public
   * @type {AgentMessageEventsClass}
   */
  public events: AgentMessageEventsClass;

  /**
   * Creates an instance of ConnectionsManager.
   *
   * @constructor
   * @param {Castor} castor
   * @param {Mercury} mercury
   * @param {Pluto} pluto
   * @param {MediatorHandler} mediationHandler
   * @param {DIDPair[]} [pairings=[]]
   */
  constructor(
    public castor: Castor,
    public mercury: Mercury,
    public pluto: Pluto,
    public agentCredentials: AgentCredentials,
    public mediationHandler: MediatorHandler,
    public pairings: DIDPair[] = [],
    private 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
   *
   * @async
   * @returns {Promise<void>}
   */
  async startMediator(): Promise<void> {
    const mediationHandler =
      await this.mediationHandler.bootRegisteredMediator();

    if (!mediationHandler) {
      throw new AgentError.NoMediatorAvailableError();
    }
  }

  /**
   * Stops all the running events
   */
  stopAllEvents(): void {
    while (this.cancellables.length > 0) {
      const [task] = this.cancellables.splice(0, 1);
      if (task) {
        task.cancel();
      }
    }
  }

  /**
   * Asyncronously wait for a message response just by waiting for new messages that match the specified ID
   *
   * @async
   * @param {string} id
   * @returns {Promise<Message | undefined>}
   */
  async awaitMessageResponse(id: string): Promise<Message | undefined> {
    console.log("Deprecated, use agent.addListener('THREAD-{{Your thread || messageId}}', fn), this method does not support live-mode.");
    const messages = await this.mediationHandler.pickupUnreadMessages(10);
    return messages.find(x => x.message.thid === id)?.message;
  }

  /**
   * Asyncronously process unread messages that are received by either http or websockets didcomm transport
   * This method replaces awaitMessages()
   * @param messages 
   */
  async processMessages(unreadMessages: {
    attachmentId: string;
    message: Message;
  }[]): Promise<void> {
    if (!this.mediationHandler.mediator) {
      throw new AgentError.NoMediatorAvailableError();
    }

    if (unreadMessages.length) {
      const received = unreadMessages.filter(x => x.message.direction === MessageDirection.RECEIVED);
      const messages = received.map(x => x.message);
      const messageIds = received.map(x => x.attachmentId);

      if (messages.length > 0) {
        await this.pluto.storeMessages(messages);
      }

      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;

        const matchingMessages = allMessages.filter(({ thid, piuri }) =>
          thid === threadId &&
          piuri === ProtocolType.DidcommIssueCredential
        );

        if (matchingMessages.length > 0) {
          for (let message of matchingMessages) {
            const issueMessage = IssueCredential.fromMessage(message);
            const credential = await this.agentCredentials.processIssuedCredentialMessage(
              issueMessage
            );
            await this.pluto.revokeCredential(credential);
            this.events.emit(ListenerKey.REVOKE, credential);
          }
        }
      }

      if (messageIds.length) {
        await this.mediationHandler.registerMessagesAsRead(messageIds);
      }

      this.events.emit(ListenerKey.MESSAGE, messages);
    }
  }

  /**
   * Asyncronously add a didPair (didcomm connection) into pluto
   *
   * @async
   * @param {DIDPair} paired
   * @returns {Promise<void>}
   */
  async addConnection(paired: DIDPair): Promise<void> {
    // TODO: what is the desired functionality here?
    // can/should we re-use connections?
    // seems always false as `paired.name` is always unique so will never match?
    if (this.findIndex(paired) !== -1) {
      return;
    }

    const storeDIDPairTask = new CancellableTask<DIDPair>(async () => {
      await this.pluto.storeDIDPair(paired.host, paired.receiver, paired.name);
      this.events.emit(ListenerKey.CONNECTION, paired);
      return paired;
    });

    this.cancellables.push(storeDIDPairTask);

    storeDIDPairTask.callback((pair: DIDPair) => this.pairings.push(pair));
  }

  /**
   * Find the specified did pair connection index
   *
   * @param {DIDPair} pair
   * @returns {*}
   */
  findIndex(pair: DIDPair) {
    return this.pairings.findIndex(
      (pairing) =>
        pair.host === pairing.host &&
        pair.name === pairing.name &&
        pair.receiver === pairing.receiver
    );
  }

  /**
   * Remove a didPair or a didcomm connection, this does not mean the mediator will do
   * this but just means the connection will be removed from the current storage
   *
   * @async
   * @param {DIDPair} pair
   * @returns {Promise<void>}
   */
  async removeConnection(pair: DIDPair): Promise<void> {
    const pairIndex = this.findIndex(pair);
    if (pairIndex !== -1) {
      this.pairings.splice(pairIndex, 1);
    }
  }

  /**
   * Asyncronously establish mediator with a mediator by providing the Host DID
   *
   * @async
   * @param {DID} hostDID
   * @returns {Promise<void>}
   */
  async registerMediator(hostDID: DID): Promise<void> {
    await this.mediationHandler.achieveMediation(hostDID);
  }

  /**
   * Asyncronously store a message and send it as didcomm message through the mercury implementation
   *
   * @async
   * @param {Message} message
   * @returns {Promise<Message | undefined>}
   */
  async sendMessage(message: Message): Promise<Message | undefined> {
    await this.pluto.storeMessage(message);
    return this.mercury.sendMessageParseMessage(message);
  }

  /**
   * Asyncronously start fetching new messages
   *
   * @param {number} iterationPeriod
   */
  async startFetchingMessages(iterationPeriod: number): Promise<void> {
    if (this.cancellable || !this.mediationHandler.mediator) {
      return;
    }
    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://")
      ) && this.withWebsocketsExperiment
    );
    if (!hasWebsocket) {
      const timeInterval = Math.max(iterationPeriod, 5) * 1000;
      this.cancellable = new CancellableTask(async () => {
        const unreadMessages = await this.mediationHandler.pickupUnreadMessages(10);
        await this.processMessages(unreadMessages);
      }, timeInterval);
    } else {
      //Connecting to websockets, do not repeat the task
      this.cancellable = new CancellableTask(async (signal) => {
        this.mediationHandler.listenUnreadMessages(
          signal,
          hasWebsocket.serviceEndpoint.uri,
          async (messages) => {
            const unreadMessages = messages.reduce<{
              attachmentId: string;
              message: Message;
            }[]>((unreads, message) => {
              const attachment = message.attachments.at(0);
              if (!attachment) {
                return unreads;
              }
              return [
                ...unreads,
                {
                  message: message,
                  attachmentId: attachment.id
                }
              ];
            }, []);

            await this.processMessages(unreadMessages);
          }
        );
      });
    }

    this.cancellable.then().catch((err) => {
      if (err instanceof Error) {
        if (err.message !== "Task was cancelled") throw err;
      } else throw err;
    });
  }

  /**
   * Asyncronously stop fetching messages
   */
  stopFetchingMessages(): void {
    this.cancellable?.cancel();
    this.cancellable = undefined;
  }
}

1	import { uuid } from "@stablelib/uuid";
2	import { DID, Message, MessageDirection, Pollux } from "../../domain";
3	import { Castor } from "../../domain/buildingBlocks/Castor";
4	import { Mercury } from "../../domain/buildingBlocks/Mercury";
5	import { Pluto } from "../../domain/buildingBlocks/Pluto";
6	import { DIDPair } from "../../domain/models/DIDPair";
7	import { AgentError } from "../../domain/models/Errors";
8	import { AgentMessageEvents } from "../Agent.MessageEvents";
9	import { CancellableTask } from "../helpers/Task";
10	import {
11	AgentCredentials,
12	AgentMessageEvents as AgentMessageEventsClass,
13	AgentOptions,
14	ConnectionsManager as ConnectionsManagerClass,
15	ListenerKey,
16	MediatorHandler,
17	} from "../types";
18	import { ProtocolType } from "../protocols/ProtocolTypes";
19	import { RevocationNotification } from "../protocols/revocation/RevocationNotfiication";
20	import { IssueCredential } from "../protocols/issueCredential/IssueCredential";
21
22
23	/**
24	* ConnectionsManager is responsible of establishing didcomm connection and
25	* mediation process with other mediators through didcomm and is also
26	* responsible of managing the task to periodically fetch messages from the mediator once connection is established
27	*
28	* @class ConnectionsManager
29	* @typedef {ConnectionsManager}
30	*/
31	export class ConnectionsManager implements ConnectionsManagerClass {
32	/**
33	* An array with cancellable tasks, mainly used to store one or multiple didcomm
34	* connections in storage implementation at the same time. All of them can be cancelled
35	* despite they run asyncronously when the Edge agent stops
36	*
37	* @public
38	* @type {CancellableTask<any>[]}
39	*/
40	// eslint-disable-next-line @typescript-eslint/no-explicit-any
41	public cancellables: CancellableTask<any>[] = [];	30✔
42	/**
43	* Cancellable task used to listen for new messages, stopping the Agent should also stop this
44	* from running and destroy the instance of the task until agent is started again
45	*
46	* @public
47	* @type {?CancellableTask<void>}
48	*/
49	public cancellable?: CancellableTask<void>;
50
51	/**
52	* A list of public facing events which will notify the user interface when specific things happen,
53	* for now when new messages arrive or didcomm connections are established in order to make UI more reactive
54	*
55	* @public
56	* @type {AgentMessageEventsClass}
57	*/
58	public events: AgentMessageEventsClass;
59
60	/**
61	* Creates an instance of ConnectionsManager.
62	*
63	* @constructor
64	* @param {Castor} castor
65	* @param {Mercury} mercury
66	* @param {Pluto} pluto
67	* @param {MediatorHandler} mediationHandler
68	* @param {DIDPair[]} [pairings=[]]
69	*/
70	constructor(
71	public castor: Castor,
72	public mercury: Mercury,
73	public pluto: Pluto,
74	public agentCredentials: AgentCredentials,
75	public mediationHandler: MediatorHandler,
76	public pairings: DIDPair[] = [],	28✔
77	private options?: AgentOptions
78	) {
79	this.events = new AgentMessageEvents();	30✔
80	}
81
82	get withWebsocketsExperiment() {
NEW 83	return this.options?.experiments?.liveMode === true	×
84	}
85
86	/**
87	* Asyncronously Start the mediator, just checking if we had one stored in Database and
88	* setting that one as default during the Agent start
89	*
90	* @async
91	* @returns {Promise<void>}
92	*/
93	async startMediator(): Promise<void> {
94	const mediationHandler =
UNCOV 95	await this.mediationHandler.bootRegisteredMediator();	×
96
97	if (!mediationHandler) {	×
UNCOV 98	throw new AgentError.NoMediatorAvailableError();	×
99	}
100	}
101
102	/**
103	* Stops all the running events
104	*/
105	stopAllEvents(): void {
106	while (this.cancellables.length > 0) {	×
107	const [task] = this.cancellables.splice(0, 1);	×
108	if (task) {	×
UNCOV 109	task.cancel();	×
110	}
111	}
112	}
113
114	/**
115	* Asyncronously wait for a message response just by waiting for new messages that match the specified ID
116	*
117	* @async
118	* @param {string} id
119	* @returns {Promise<Message \| undefined>}
120	*/
121	async awaitMessageResponse(id: string): Promise<Message \| undefined> {
122	console.log("Deprecated, use agent.addListener('THREAD-{{Your thread \|\| messageId}}', fn), this method does not support live-mode.");	×
123	const messages = await this.mediationHandler.pickupUnreadMessages(10);	×
UNCOV 124	return messages.find(x => x.message.thid === id)?.message;	×
125	}
126
127	/**
128	* Asyncronously process unread messages that are received by either http or websockets didcomm transport
129	* This method replaces awaitMessages()
130	* @param messages
131	*/
132	async processMessages(unreadMessages: {
133	attachmentId: string;
134	message: Message;
135	}[]): Promise<void> {
136	if (!this.mediationHandler.mediator) {	1!
UNCOV 137	throw new AgentError.NoMediatorAvailableError();	×
138	}
139
140	if (unreadMessages.length) {	1!
141	const received = unreadMessages.filter(x => x.message.direction === MessageDirection.RECEIVED);	1✔
142	const messages = received.map(x => x.message);	1✔
143	const messageIds = received.map(x => x.attachmentId);	1✔
144
145	if (messages.length > 0) {	1!
146	await this.pluto.storeMessages(messages);	1✔
147	}
148
149	const revokeMessages = messages.filter(x => x.piuri === ProtocolType.PrismRevocation);	1✔
150	const allMessages = await this.pluto.getAllMessages();	1✔
151
152	for (let message of revokeMessages) {	1✔
153	const revokeMessage = RevocationNotification.fromMessage(message);	1✔
154	const threadId = revokeMessage.body.issueCredentialProtocolThreadId;	1✔
155
156	const matchingMessages = allMessages.filter(({ thid, piuri }) =>	1✔
157	thid === threadId &&	2✔
158	piuri === ProtocolType.DidcommIssueCredential
159	);
160
161	if (matchingMessages.length > 0) {	1!
162	for (let message of matchingMessages) {	1✔
163	const issueMessage = IssueCredential.fromMessage(message);	1✔
164	const credential = await this.agentCredentials.processIssuedCredentialMessage(	1✔
165	issueMessage
166	);
167	await this.pluto.revokeCredential(credential);	1✔
168	this.events.emit(ListenerKey.REVOKE, credential);	1✔
169	}
170	}
171	}
172
173	if (messageIds.length) {	1!
174	await this.mediationHandler.registerMessagesAsRead(messageIds);	1✔
175	}
176
177	this.events.emit(ListenerKey.MESSAGE, messages);	1✔
178	}
179	}
180
181	/**
182	* Asyncronously add a didPair (didcomm connection) into pluto
183	*
184	* @async
185	* @param {DIDPair} paired
186	* @returns {Promise<void>}
187	*/
188	async addConnection(paired: DIDPair): Promise<void> {
189	// TODO: what is the desired functionality here?
190	// can/should we re-use connections?
191	// seems always false as `paired.name` is always unique so will never match?
192	if (this.findIndex(paired) !== -1) {	×
UNCOV 193	return;	×
194	}
195
196	const storeDIDPairTask = new CancellableTask<DIDPair>(async () => {	×
197	await this.pluto.storeDIDPair(paired.host, paired.receiver, paired.name);	×
198	this.events.emit(ListenerKey.CONNECTION, paired);	×
UNCOV 199	return paired;	×
200	});
201
UNCOV 202	this.cancellables.push(storeDIDPairTask);	×
203
UNCOV 204	storeDIDPairTask.callback((pair: DIDPair) => this.pairings.push(pair));	×
205	}
206
207	/**
208	* Find the specified did pair connection index
209	*
210	* @param {DIDPair} pair
211	* @returns {*}
212	*/
213	findIndex(pair: DIDPair) {
UNCOV 214	return this.pairings.findIndex(	×
215	(pairing) =>
UNCOV 216	pair.host === pairing.host &&	×
217	pair.name === pairing.name &&
218	pair.receiver === pairing.receiver
219	);
220	}
221
222	/**
223	* Remove a didPair or a didcomm connection, this does not mean the mediator will do
224	* this but just means the connection will be removed from the current storage
225	*
226	* @async
227	* @param {DIDPair} pair
228	* @returns {Promise<void>}
229	*/
230	async removeConnection(pair: DIDPair): Promise<void> {
231	const pairIndex = this.findIndex(pair);	×
232	if (pairIndex !== -1) {	×
UNCOV 233	this.pairings.splice(pairIndex, 1);	×
234	}
235	}
236
237	/**
238	* Asyncronously establish mediator with a mediator by providing the Host DID
239	*
240	* @async
241	* @param {DID} hostDID
242	* @returns {Promise<void>}
243	*/
244	async registerMediator(hostDID: DID): Promise<void> {
UNCOV 245	await this.mediationHandler.achieveMediation(hostDID);	×
246	}
247
248	/**
249	* Asyncronously store a message and send it as didcomm message through the mercury implementation
250	*
251	* @async
252	* @param {Message} message
253	* @returns {Promise<Message \| undefined>}
254	*/
255	async sendMessage(message: Message): Promise<Message \| undefined> {
256	await this.pluto.storeMessage(message);	×
UNCOV 257	return this.mercury.sendMessageParseMessage(message);	×
258	}
259
260	/**
261	* Asyncronously start fetching new messages
262	*
263	* @param {number} iterationPeriod
264	*/
265	async startFetchingMessages(iterationPeriod: number): Promise<void> {
266	if (this.cancellable \|\| !this.mediationHandler.mediator) {	×
UNCOV 267	return;	×
268	}
269	const currentMediator = this.mediationHandler.mediator.mediatorDID;	×
270	const resolvedMediator = await this.castor.resolveDID(currentMediator.toString());	×
271	const hasWebsocket = resolvedMediator.services.find(({ serviceEndpoint: { uri } }) =>	×
NEW 272	(	×
273	uri.startsWith("ws://") \|\|
274	uri.startsWith("wss://")
275	) && this.withWebsocketsExperiment
276	);
277	if (!hasWebsocket) {	×
278	const timeInterval = Math.max(iterationPeriod, 5) * 1000;	×
279	this.cancellable = new CancellableTask(async () => {	×
280	const unreadMessages = await this.mediationHandler.pickupUnreadMessages(10);	×
UNCOV 281	await this.processMessages(unreadMessages);	×
282	}, timeInterval);
283	} else {
284	//Connecting to websockets, do not repeat the task
285	this.cancellable = new CancellableTask(async (signal) => {	×
UNCOV 286	this.mediationHandler.listenUnreadMessages(	×
287	signal,
288	hasWebsocket.serviceEndpoint.uri,
289	async (messages) => {
UNCOV 290	const unreadMessages = messages.reduce<{	×
291	attachmentId: string;
292	message: Message;
293	}[]>((unreads, message) => {
294	const attachment = message.attachments.at(0);	×
295	if (!attachment) {	×
UNCOV 296	return unreads;	×
297	}
UNCOV 298	return [	×
299	...unreads,
300	{
301	message: message,
302	attachmentId: attachment.id
303	}
304	];
305	}, []);
306
UNCOV 307	await this.processMessages(unreadMessages);	×
308	}
309	);
310	});
311	}
312
313	this.cancellable.then().catch((err) => {	×
314	if (err instanceof Error) {	×
315	if (err.message !== "Task was cancelled") throw err;	×
UNCOV 316	} else throw err;	×
317	});
318	}
319
320	/**
321	* Asyncronously stop fetching messages
322	*/
323	stopFetchingMessages(): void {
324	this.cancellable?.cancel();	×
UNCOV 325	this.cancellable = undefined;	×
326	}
327	}

input-output-hk / atala-prism-wallet-sdk-ts / 8814828651

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous