8433882641

Committed 26 Mar 2024 09:44AM UTC coverage: 65.501% (+1.7%) from 63.779%

Build # 8433882641

Build Type

push

github

Committed by

web-flow

Commit Message

fix: Agent.createNewPrismDID to use derivationPath (#158)

Run Details

716 of 1265 branches covered (56.6%)

Branch coverage included in aggregate %.

46 of 50 new or added lines in 6 files covered. (92.0%)

31 existing lines in 1 file now uncovered.

1775 of 2538 relevant lines covered (69.94%)

12.21 hits per line

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

28.7

/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,
  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[] = []
  ) {
    this.events = new AgentMessageEvents();
  }

  /**
   * 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) {
      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	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>[] = [];	30✔
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[] = []	30✔
76	) {
77	this.events = new AgentMessageEvents();	30✔
78	}
79
80	/**
81	* Asyncronously Start the mediator, just checking if we had one stored in Database and
82	* setting that one as default during the Agent start
83	*
84	* @async
85	* @returns {Promise<void>}
86	*/
87	async startMediator(): Promise<void> {
88	const mediationHandler =
89	await this.mediationHandler.bootRegisteredMediator();	×
90
91	if (!mediationHandler) {	×
92	throw new AgentError.NoMediatorAvailableError();	×
93	}
94	}
95
96	/**
97	* Stops all the running events
98	*/
99	stopAllEvents(): void {
100	while (this.cancellables.length > 0) {	×
101	const [task] = this.cancellables.splice(0, 1);	×
102	if (task) {	×
103	task.cancel();	×
104	}
105	}
106	}
107
108	/**
109	* Asyncronously wait for a message response just by waiting for new messages that match the specified ID
110	*
111	* @async
112	* @param {string} id
113	* @returns {Promise<Message \| undefined>}
114	*/
115	async awaitMessageResponse(id: string): Promise<Message \| undefined> {
116	console.log("Deprecated, use agent.addListener('THREAD-{{Your thread \|\| messageId}}', fn), this method does not support live-mode.");	×
117	const messages = await this.mediationHandler.pickupUnreadMessages(10);	×
118	return messages.find(x => x.message.thid === id)?.message;	×
119	}
120
121	/**
122	* Asyncronously process unread messages that are received by either http or websockets didcomm transport
123	* This method replaces awaitMessages()
124	* @param messages
125	*/
126	async processMessages(unreadMessages: {
127	attachmentId: string;
128	message: Message;
129	}[]): Promise<void> {
130	if (!this.mediationHandler.mediator) {	1!
UNCOV 131	throw new AgentError.NoMediatorAvailableError();	×
132	}
133
134	if (unreadMessages.length) {	1!
135	const received = unreadMessages.filter(x => x.message.direction === MessageDirection.RECEIVED);	1✔
136	const messages = received.map(x => x.message);	1✔
137	const messageIds = received.map(x => x.attachmentId);	1✔
138
139	if (messages.length > 0) {	1!
140	await this.pluto.storeMessages(messages);	1✔
141	}
142
143	const revokeMessages = messages.filter(x => x.piuri === ProtocolType.PrismRevocation);	1✔
144	const allMessages = await this.pluto.getAllMessages();	1✔
145
146	for (let message of revokeMessages) {	1✔
147	const revokeMessage = RevocationNotification.fromMessage(message);	1✔
148	const threadId = revokeMessage.body.issueCredentialProtocolThreadId;	1✔
149
150	const matchingMessages = allMessages.filter(({ thid, piuri }) =>	1✔
151	thid === threadId &&	2✔
152	piuri === ProtocolType.DidcommIssueCredential
153	);
154
155	if (matchingMessages.length > 0) {	1!
156	for (let message of matchingMessages) {	1✔
157	const issueMessage = IssueCredential.fromMessage(message);	1✔
158	const credential = await this.agentCredentials.processIssuedCredentialMessage(	1✔
159	issueMessage
160	);
161	await this.pluto.revokeCredential(credential);	1✔
162	this.events.emit(ListenerKey.REVOKE, credential);	1✔
163	}
164	}
165	}
166
167	if (messageIds.length) {	1!
168	await this.mediationHandler.registerMessagesAsRead(messageIds);	1✔
169	}
170
171	this.events.emit(ListenerKey.MESSAGE, messages);	1✔
172	}
173	}
174
175	/**
176	* Asyncronously add a didPair (didcomm connection) into pluto
177	*
178	* @async
179	* @param {DIDPair} paired
180	* @returns {Promise<void>}
181	*/
182	async addConnection(paired: DIDPair): Promise<void> {
183	// TODO: what is the desired functionality here?
184	// can/should we re-use connections?
185	// seems always false as `paired.name` is always unique so will never match?
UNCOV 186	if (this.findIndex(paired) !== -1) {	×
UNCOV 187	return;	×
188	}
189
UNCOV 190	const storeDIDPairTask = new CancellableTask<DIDPair>(async () => {	×
UNCOV 191	await this.pluto.storeDIDPair(paired.host, paired.receiver, paired.name);	×
UNCOV 192	this.events.emit(ListenerKey.CONNECTION, paired);	×
UNCOV 193	return paired;	×
194	});
195
UNCOV 196	this.cancellables.push(storeDIDPairTask);	×
197
198	storeDIDPairTask.callback((pair: DIDPair) => this.pairings.push(pair));	×
199	}
200
201	/**
202	* Find the specified did pair connection index
203	*
204	* @param {DIDPair} pair
205	* @returns {*}
206	*/
207	findIndex(pair: DIDPair) {
UNCOV 208	return this.pairings.findIndex(	×
209	(pairing) =>
UNCOV 210	pair.host === pairing.host &&	×
211	pair.name === pairing.name &&
212	pair.receiver === pairing.receiver
213	);
214	}
215
216	/**
217	* Remove a didPair or a didcomm connection, this does not mean the mediator will do
218	* this but just means the connection will be removed from the current storage
219	*
220	* @async
221	* @param {DIDPair} pair
222	* @returns {Promise<void>}
223	*/
224	async removeConnection(pair: DIDPair): Promise<void> {
UNCOV 225	const pairIndex = this.findIndex(pair);	×
UNCOV 226	if (pairIndex !== -1) {	×
UNCOV 227	this.pairings.splice(pairIndex, 1);	×
228	}
229	}
230
231	/**
232	* Asyncronously establish mediator with a mediator by providing the Host DID
233	*
234	* @async
235	* @param {DID} hostDID
236	* @returns {Promise<void>}
237	*/
238	async registerMediator(hostDID: DID): Promise<void> {
UNCOV 239	await this.mediationHandler.achieveMediation(hostDID);	×
240	}
241
242	/**
243	* Asyncronously store a message and send it as didcomm message through the mercury implementation
244	*
245	* @async
246	* @param {Message} message
247	* @returns {Promise<Message \| undefined>}
248	*/
249	async sendMessage(message: Message): Promise<Message \| undefined> {
250	await this.pluto.storeMessage(message);	×
UNCOV 251	return this.mercury.sendMessageParseMessage(message);	×
252	}
253
254	/**
255	* Asyncronously start fetching new messages
256	*
257	* @param {number} iterationPeriod
258	*/
259	async startFetchingMessages(iterationPeriod: number): Promise<void> {
UNCOV 260	if (this.cancellable \|\| !this.mediationHandler.mediator) {	×
261	return;	×
262	}
UNCOV 263	const currentMediator = this.mediationHandler.mediator.mediatorDID;	×
UNCOV 264	const resolvedMediator = await this.castor.resolveDID(currentMediator.toString());	×
UNCOV 265	const hasWebsocket = resolvedMediator.services.find(({ serviceEndpoint: { uri } }) =>	×
UNCOV 266	uri.startsWith("ws://") \|\|	×
267	uri.startsWith("wss://")
268	);
UNCOV 269	if (!hasWebsocket) {	×
UNCOV 270	const timeInterval = Math.max(iterationPeriod, 5) * 1000;	×
271	this.cancellable = new CancellableTask(async () => {	×
272	const unreadMessages = await this.mediationHandler.pickupUnreadMessages(10);	×
UNCOV 273	await this.processMessages(unreadMessages);	×
274	}, timeInterval);
275	} else {
276	//Connecting to websockets, do not repeat the task
277	this.cancellable = new CancellableTask(async (signal) => {	×
UNCOV 278	this.mediationHandler.listenUnreadMessages(	×
279	signal,
280	hasWebsocket.serviceEndpoint.uri,
281	async (messages) => {
282	const unreadMessages = messages.reduce<{	×
283	attachmentId: string;
284	message: Message;
285	}[]>((unreads, message) => {
UNCOV 286	const attachment = message.attachments.at(0);	×
UNCOV 287	if (!attachment) {	×
288	return unreads;	×
289	}
UNCOV 290	return [	×
291	...unreads,
292	{
293	message: message,
294	attachmentId: attachment.id
295	}
296	];
297	}, []);
298
299	await this.processMessages(unreadMessages);	×
300	}
301	);
302	});
303	}
304
UNCOV 305	this.cancellable.then().catch((err) => {	×
UNCOV 306	if (err instanceof Error) {	×
UNCOV 307	if (err.message !== "Task was cancelled") throw err;	×
UNCOV 308	} else throw err;	×
309	});
310	}
311
312	/**
313	* Asyncronously stop fetching messages
314	*/
315	stopFetchingMessages(): void {
316	this.cancellable?.cancel();	×
317	this.cancellable = undefined;	×
318	}
319	}

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

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