8823694925

Committed 24 Apr 2024 09:42PM UTC coverage: 66.649% (-0.2%) from 66.807%

Build # 8823694925

Build Type

Pull #200

github

Committed by

web-flow

Commit Message

Merge 0ac49f5f4 into 5fe7f33b4

Pull Request Pull Request #200: fix: e2e issues with latest websocket changes

Run Details

728 of 1272 branches covered (57.23%)

Branch coverage included in aggregate %.

4 of 15 new or added lines in 3 files covered. (26.67%)

4 existing lines in 2 files now uncovered.

1810 of 2536 relevant lines covered (71.37%)

12.69 hits per line

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

45.54

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

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[] = [],
    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
   *
   * @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://")
    )
    );
    if (hasWebsocket && this.withWebsocketsExperiment) {
      this.cancellable = new CancellableTask(async (signal) => {
        this.mediationHandler.listenUnreadMessages(
          signal,
          hasWebsocket.serviceEndpoint.uri,
          (messages) => this.processMessages(messages)
        );
      });
    } else {
      const timeInterval = Math.max(iterationPeriod, 5) * 1000;
      this.cancellable = new CancellableTask(async () => {
        const unreadMessages = await this.mediationHandler.pickupUnreadMessages(10);
        await this.processMessages(unreadMessages);
      }, timeInterval);
    }

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

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

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