mirror of
https://github.com/logos-messaging/js-waku.git
synced 2026-01-10 01:33:13 +00:00
Merge 2b07864a25216a74537f39a5a70a31d743cad77a into 5334a7fcc91544d33294beaad9b45e641ecf404d
This commit is contained in:
fryorcraken
GitHub
commit
b6f2026ee1
@ -25,4 +25,8 @@ export { StreamManager } from "./lib/stream_manager/index.js";
|
||||
|
||||
export { MetadataCodec, wakuMetadata } from "./lib/metadata/index.js";
|
||||
|
||||
export { messageHash, messageHashStr } from "./lib/message_hash/index.js";
|
||||
export {
|
||||
messageHash,
|
||||
messageHashStr,
|
||||
deterministicMessageHashing
|
||||
} from "./lib/message_hash/index.js";
|
||||
|
||||
@ -1 +1,5 @@
|
||||
export { messageHash, messageHashStr } from "./message_hash.js";
|
||||
export {
|
||||
messageHash,
|
||||
messageHashStr,
|
||||
deterministicMessageHashing
|
||||
} from "./message_hash.js";
|
||||
|
||||
@ -105,6 +105,7 @@ export interface IEncoder {
|
||||
export interface IDecoder<T extends IDecodedMessage> {
|
||||
contentTopic: string;
|
||||
pubsubTopic: PubsubTopic;
|
||||
routingInfo: IRoutingInfo;
|
||||
fromWireToProtoObj: (bytes: Uint8Array) => Promise<IProtoMessage | undefined>;
|
||||
fromProtoObj: (
|
||||
pubsubTopic: string,
|
||||
|
||||
@ -11,9 +11,10 @@ import type { HealthStatus } from "./health_status.js";
|
||||
import type { Libp2p } from "./libp2p.js";
|
||||
import type { ILightPush } from "./light_push.js";
|
||||
import { IDecodedMessage, IDecoder, IEncoder } from "./message.js";
|
||||
import { ContentTopic } from "./misc.js";
|
||||
import type { Protocols } from "./protocols.js";
|
||||
import type { IRelay } from "./relay.js";
|
||||
import type { ShardId } from "./sharding.js";
|
||||
import type { IRoutingInfo, ShardId } from "./sharding.js";
|
||||
import type { IStore } from "./store.js";
|
||||
|
||||
export type CreateDecoderParams = {
|
||||
@ -27,7 +28,8 @@ export type CreateEncoderParams = CreateDecoderParams & {
|
||||
|
||||
export enum WakuEvent {
|
||||
Connection = "waku:connection",
|
||||
Health = "waku:health"
|
||||
Health = "waku:health",
|
||||
SubscribeError = "subscribe:error"
|
||||
}
|
||||
|
||||
export interface IWakuEvents {
|
||||
@ -52,9 +54,28 @@ export interface IWakuEvents {
|
||||
* });
|
||||
*/
|
||||
[WakuEvent.Health]: CustomEvent<HealthStatus>;
|
||||
|
||||
/**
|
||||
* Emitted when there is an irrecoverable error when subscribing.
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* waku.addEventListener(WakuEvent.SubscribeError, (event) => {
|
||||
* console.log(event.detail);
|
||||
* });
|
||||
*/
|
||||
[WakuEvent.SubscribeError]: CustomEvent<string>;
|
||||
}
|
||||
|
||||
export interface IMessageEmitterEvents {
|
||||
[contentTopic: string]: CustomEvent<{
|
||||
payload: Uint8Array;
|
||||
messageHash: Uint8Array;
|
||||
}>;
|
||||
}
|
||||
|
||||
export type IWakuEventEmitter = TypedEventEmitter<IWakuEvents>;
|
||||
export type IMessageEmitter = TypedEventEmitter<IMessageEmitterEvents>;
|
||||
|
||||
export interface IWaku {
|
||||
libp2p: Libp2p;
|
||||
@ -78,6 +99,20 @@ export interface IWaku {
|
||||
*/
|
||||
events: IWakuEventEmitter;
|
||||
|
||||
/**
|
||||
* Emits messages on their content topic. Messages may be coming from subscriptions
|
||||
* or store queries (TODO). The payload is directly emitted
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* waku.messageEmitter.addEventListener("/some/0/content-topic/proto", (event) => {
|
||||
* const payload: UInt8Array = event.detail
|
||||
* MyDecoder.decode(payload);
|
||||
* });
|
||||
* ```
|
||||
*/
|
||||
messageEmitter: IMessageEmitter;
|
||||
|
||||
/**
|
||||
* Returns a unique identifier for a node on the network.
|
||||
*
|
||||
@ -221,6 +256,8 @@ export interface IWaku {
|
||||
*/
|
||||
createDecoder(params: CreateDecoderParams): IDecoder<IDecodedMessage>;
|
||||
|
||||
createRoutingInfo(contentTopic?: string, shardId?: number): IRoutingInfo;
|
||||
|
||||
/**
|
||||
* Creates an encoder for Waku messages on a specific content topic.
|
||||
*
|
||||
@ -251,6 +288,8 @@ export interface IWaku {
|
||||
*/
|
||||
createEncoder(params: CreateEncoderParams): IEncoder;
|
||||
|
||||
subscribe(contentTopics: ContentTopic[]): void;
|
||||
|
||||
/**
|
||||
* @returns {boolean} `true` if the node was started and `false` otherwise
|
||||
*/
|
||||
|
||||
@ -74,3 +74,32 @@ export async function sign(
|
||||
export function keccak256(input: Uint8Array): Uint8Array {
|
||||
return new Uint8Array(sha3.keccak256.arrayBuffer(input));
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare two public keys, can be used to verify that a given signature matches
|
||||
* expectations.
|
||||
*
|
||||
* @param publicKeyA - The first public key to compare
|
||||
* @param publicKeyB - The second public key to compare
|
||||
* @returns true if the public keys are the same
|
||||
*/
|
||||
export function comparePublicKeys(
|
||||
publicKeyA: Uint8Array | undefined,
|
||||
publicKeyB: Uint8Array | undefined
|
||||
): boolean {
|
||||
if (!publicKeyA || !publicKeyB) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (publicKeyA.length !== publicKeyB.length) {
|
||||
return false;
|
||||
}
|
||||
|
||||
for (let i = 0; i < publicKeyA.length; i++) {
|
||||
if (publicKeyA[i] !== publicKeyB[i]) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
@ -1,11 +1,17 @@
|
||||
import {
|
||||
comparePublicKeys,
|
||||
generatePrivateKey,
|
||||
generateSymmetricKey,
|
||||
getPublicKey
|
||||
} from "./crypto/index.js";
|
||||
import { DecodedMessage } from "./decoded_message.js";
|
||||
|
||||
export { generatePrivateKey, generateSymmetricKey, getPublicKey };
|
||||
export {
|
||||
generatePrivateKey,
|
||||
generateSymmetricKey,
|
||||
getPublicKey,
|
||||
comparePublicKeys
|
||||
};
|
||||
export type { DecodedMessage };
|
||||
|
||||
export * as ecies from "./ecies.js";
|
||||
|
||||
@ -206,3 +206,105 @@ export function createDecoder(
|
||||
): Decoder {
|
||||
return new Decoder(contentTopic, routingInfo, symKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of decrypting a message with AES symmetric encryption.
|
||||
*/
|
||||
export interface SymmetricDecryptionResult {
|
||||
/** The decrypted payload */
|
||||
payload: Uint8Array;
|
||||
/** The signature if the message was signed */
|
||||
signature?: Uint8Array;
|
||||
/** The recovered public key if the message was signed */
|
||||
signaturePublicKey?: Uint8Array;
|
||||
}
|
||||
|
||||
/**
|
||||
* AES symmetric encryption.
|
||||
*
|
||||
*
|
||||
* Follows [26/WAKU2-PAYLOAD](https://rfc.vac.dev/spec/26/) encryption standard.
|
||||
*/
|
||||
export class SymmetricEncryption {
|
||||
/**
|
||||
* Creates an AES Symmetric encryption instance.
|
||||
*
|
||||
* @param symKey - The symmetric key for encryption (32 bytes recommended)
|
||||
* @param sigPrivKey - Optional private key to sign messages before encryption
|
||||
*/
|
||||
public constructor(
|
||||
private symKey: Uint8Array,
|
||||
private sigPrivKey?: Uint8Array
|
||||
) {}
|
||||
|
||||
/**
|
||||
* Encrypts a byte array payload.
|
||||
*
|
||||
* The encryption process:
|
||||
* 1. Optionally signs the payload with the private key
|
||||
* 2. Adds padding to obscure payload size
|
||||
* 3. Encrypts using AES-256-GCM
|
||||
*
|
||||
* @param payload - The data to encrypt
|
||||
* @returns The encrypted payload
|
||||
*/
|
||||
public async encrypt(payload: Uint8Array): Promise<Uint8Array> {
|
||||
const preparedPayload = await preCipher(payload, this.sigPrivKey);
|
||||
return encryptSymmetric(preparedPayload, this.symKey);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* AES symmetric decryption.
|
||||
*
|
||||
* Follows [26/WAKU2-PAYLOAD](https://rfc.vac.dev/spec/26/) encryption standard.
|
||||
*/
|
||||
export class SymmetricDecryption {
|
||||
/**
|
||||
* Creates an AES Symmetric decryption instance.
|
||||
*
|
||||
* @param symKey - The symmetric key for decryption (must match encryption key)
|
||||
*/
|
||||
public constructor(private symKey: Uint8Array) {}
|
||||
|
||||
/**
|
||||
* Decrypts an encrypted byte array payload.
|
||||
*
|
||||
* The decryption process:
|
||||
* 1. Decrypts using AES-256-GCM
|
||||
* 2. Removes padding
|
||||
* 3. Verifies and recovers signature if present
|
||||
*
|
||||
* @param encryptedPayload - The encrypted data (from [[SymmetricEncryption.encrypt]])
|
||||
* @returns Object containing the decrypted payload and signature info, or undefined if decryption fails
|
||||
*/
|
||||
public async decrypt(
|
||||
encryptedPayload: Uint8Array
|
||||
): Promise<SymmetricDecryptionResult | undefined> {
|
||||
try {
|
||||
const decryptedData = await decryptSymmetric(
|
||||
encryptedPayload,
|
||||
this.symKey
|
||||
);
|
||||
|
||||
if (!decryptedData) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const result = postCipher(decryptedData);
|
||||
|
||||
if (!result) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return {
|
||||
payload: result.payload,
|
||||
signature: result.sig?.signature,
|
||||
signaturePublicKey: result.sig?.publicKey
|
||||
};
|
||||
} catch (error) {
|
||||
log.error("Failed to decrypt payload", error);
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
import { IDecodedMessage, ProtocolError } from "@waku/interfaces";
|
||||
import { ProtocolError } from "@waku/interfaces";
|
||||
import type { HistoryEntry, MessageId } from "@waku/sds";
|
||||
|
||||
export const ReliableChannelEvent = {
|
||||
@ -56,8 +56,7 @@ export interface ReliableChannelEvents {
|
||||
possibleAckCount: number;
|
||||
}>;
|
||||
"message-acknowledged": CustomEvent<MessageId>;
|
||||
// TODO probably T extends IDecodedMessage?
|
||||
"message-received": CustomEvent<IDecodedMessage>;
|
||||
"message-received": CustomEvent<Uint8Array>;
|
||||
"irretrievable-message": CustomEvent<HistoryEntry>;
|
||||
"sending-message-irrecoverable-error": CustomEvent<{
|
||||
messageId: MessageId;
|
||||
|
||||
@ -1,15 +1,10 @@
|
||||
import { TypedEventEmitter } from "@libp2p/interface";
|
||||
import { messageHash } from "@waku/core";
|
||||
import {
|
||||
type Callback,
|
||||
type ContentTopic,
|
||||
type IDecodedMessage,
|
||||
type IDecoder,
|
||||
type IEncoder,
|
||||
type IMessage,
|
||||
ISendOptions,
|
||||
type IWaku,
|
||||
LightPushError,
|
||||
LightPushSDKResult,
|
||||
QueryRequestParams
|
||||
} from "@waku/interfaces";
|
||||
import {
|
||||
@ -23,14 +18,9 @@ import {
|
||||
SyncMessage
|
||||
} from "@waku/sds";
|
||||
import { Logger } from "@waku/utils";
|
||||
|
||||
import {
|
||||
QueryOnConnect,
|
||||
QueryOnConnectEvent
|
||||
} from "../query_on_connect/index.js";
|
||||
import { bytesToHex } from "@waku/utils/bytes";
|
||||
|
||||
import { ReliableChannelEvent, ReliableChannelEvents } from "./events.js";
|
||||
import { MissingMessageRetriever } from "./missing_message_retriever.js";
|
||||
import { RetryManager } from "./retry_manager.js";
|
||||
|
||||
const log = new Logger("sdk:reliable-channel");
|
||||
@ -113,6 +103,16 @@ export type ReliableChannelOptions = MessageChannelOptions & {
|
||||
processTaskMinElapseMs?: number;
|
||||
};
|
||||
|
||||
/**
|
||||
* It is best for SDS (e2e reliability) to happen within the encryption layer.
|
||||
* Hence, the consumer need to pass encryption and decryption methods for
|
||||
* outgoing and incoming messages.
|
||||
*/
|
||||
export interface IEncryption {
|
||||
encrypt: (clearPayload: Uint8Array) => Uint8Array | Promise<Uint8Array>;
|
||||
decrypt: (encryptedPayload: Uint8Array) => Uint8Array | Promise<Uint8Array>;
|
||||
}
|
||||
|
||||
/**
|
||||
* An easy-to-use reliable channel that ensures all participants to the channel have eventual message consistency.
|
||||
*
|
||||
@ -122,19 +122,16 @@ export type ReliableChannelOptions = MessageChannelOptions & {
|
||||
* @emits [[ReliableChannelEvents]]
|
||||
*
|
||||
*/
|
||||
export class ReliableChannel<
|
||||
T extends IDecodedMessage
|
||||
> extends TypedEventEmitter<ReliableChannelEvents> {
|
||||
export class ReliableChannel extends TypedEventEmitter<ReliableChannelEvents> {
|
||||
// TODO: this is PoC, we assume that message id is returned, and `undefined` means some error.
|
||||
// Borrowed from https://github.com/waku-org/js-waku/pull/2583/ for now
|
||||
private readonly _send: (
|
||||
encoder: IEncoder,
|
||||
message: IMessage,
|
||||
sendOptions?: ISendOptions
|
||||
) => Promise<LightPushSDKResult>;
|
||||
contentTopic: string,
|
||||
payload: Uint8Array,
|
||||
ephemeral?: boolean
|
||||
) => Promise<Uint8Array | undefined>;
|
||||
|
||||
private readonly _subscribe: (
|
||||
decoders: IDecoder<T> | IDecoder<T>[],
|
||||
callback: Callback<T>
|
||||
) => Promise<boolean>;
|
||||
private readonly _subscribe: (contentTopics: ContentTopic[]) => void;
|
||||
|
||||
private readonly _retrieve?: <T extends IDecodedMessage>(
|
||||
decoders: IDecoder<T>[],
|
||||
@ -147,36 +144,36 @@ export class ReliableChannel<
|
||||
private readonly sweepInBufIntervalMs: number;
|
||||
private processTaskTimeout: ReturnType<typeof setTimeout> | undefined;
|
||||
private readonly retryManager: RetryManager | undefined;
|
||||
private readonly missingMessageRetriever?: MissingMessageRetriever<T>;
|
||||
private readonly queryOnConnect?: QueryOnConnect<T>;
|
||||
// private readonly missingMessageRetriever?: MissingMessageRetriever;
|
||||
// private readonly queryOnConnect?: QueryOnConnect<T>;
|
||||
private readonly processTaskMinElapseMs: number;
|
||||
private _started: boolean;
|
||||
private encryption: IEncryption;
|
||||
|
||||
private constructor(
|
||||
public node: IWaku,
|
||||
public messageChannel: MessageChannel,
|
||||
private encoder: IEncoder,
|
||||
private decoder: IDecoder<T>,
|
||||
private contentTopic: ContentTopic,
|
||||
encryption?: IEncryption,
|
||||
options?: ReliableChannelOptions
|
||||
) {
|
||||
super();
|
||||
if (node.lightPush) {
|
||||
this._send = node.lightPush.send.bind(node.lightPush);
|
||||
// TODO: this is just a PoC
|
||||
// this._send = node.lightPush.send.bind(node.lightPush);
|
||||
} else if (node.relay) {
|
||||
this._send = node.relay.send.bind(node.relay);
|
||||
// this._send = node.relay.send.bind(node.relay);
|
||||
} else {
|
||||
throw "No protocol available to send messages";
|
||||
}
|
||||
|
||||
if (node.filter) {
|
||||
this._subscribe = node.filter.subscribe.bind(node.filter);
|
||||
} else if (node.relay) {
|
||||
// TODO: Why do relay and filter have different interfaces?
|
||||
// this._subscribe = node.relay.subscribeWithUnsubscribe;
|
||||
throw "Not implemented";
|
||||
} else {
|
||||
throw "No protocol available to receive messages";
|
||||
}
|
||||
this._subscribe = node.subscribe.bind(node);
|
||||
|
||||
// If no encryption, just set a pass through without changing the payload to keep the code simpler
|
||||
this.encryption = encryption ?? {
|
||||
encrypt: (p: Uint8Array) => p,
|
||||
decrypt: (p: Uint8Array) => p
|
||||
};
|
||||
|
||||
if (node.store) {
|
||||
this._retrieve = node.store.queryGenerator.bind(node.store);
|
||||
@ -185,13 +182,14 @@ export class ReliableChannel<
|
||||
peerManagerEvents !== undefined &&
|
||||
(options?.queryOnConnect ?? true)
|
||||
) {
|
||||
this.queryOnConnect = new QueryOnConnect(
|
||||
[this.decoder],
|
||||
this.isChannelMessageWithCausalHistory.bind(this),
|
||||
peerManagerEvents,
|
||||
node.events,
|
||||
this._retrieve.bind(this)
|
||||
);
|
||||
// this.queryOnConnect = new QueryOnConnect(
|
||||
// [this.decoder],
|
||||
// this.isChannelMessageWithCausalHistory.bind(this),
|
||||
// peerManagerEvents,
|
||||
// node.events,
|
||||
// this._retrieve.bind(this)
|
||||
// );
|
||||
// TODO: stop using decoder for store
|
||||
}
|
||||
}
|
||||
|
||||
@ -215,14 +213,15 @@ export class ReliableChannel<
|
||||
options?.processTaskMinElapseMs ?? DEFAULT_PROCESS_TASK_MIN_ELAPSE_MS;
|
||||
|
||||
if (this._retrieve) {
|
||||
this.missingMessageRetriever = new MissingMessageRetriever(
|
||||
this.decoder,
|
||||
options?.retrieveFrequencyMs,
|
||||
this._retrieve,
|
||||
async (msg: T) => {
|
||||
await this.processIncomingMessage(msg);
|
||||
}
|
||||
);
|
||||
// this.missingMessageRetriever = new MissingMessageRetriever(
|
||||
// this.decoder,
|
||||
// options?.retrieveFrequencyMs,
|
||||
// this._retrieve,
|
||||
// async (msg: T) => {
|
||||
// await this.processIncomingMessage(msg.payload);
|
||||
// }
|
||||
// );
|
||||
// TODO: stop using decoder for store
|
||||
}
|
||||
|
||||
this._started = false;
|
||||
@ -261,26 +260,26 @@ export class ReliableChannel<
|
||||
* @param decoder A channel operates within a singular encryption layer, hence the same decoder is needed for all messages
|
||||
* @param options
|
||||
*/
|
||||
public static async create<T extends IDecodedMessage>(
|
||||
public static async create(
|
||||
node: IWaku,
|
||||
channelId: ChannelId,
|
||||
senderId: SenderId,
|
||||
encoder: IEncoder,
|
||||
decoder: IDecoder<T>,
|
||||
contentTopic: ContentTopic,
|
||||
encryption?: IEncryption,
|
||||
options?: ReliableChannelOptions
|
||||
): Promise<ReliableChannel<T>> {
|
||||
): Promise<ReliableChannel> {
|
||||
const sdsMessageChannel = new MessageChannel(channelId, senderId, options);
|
||||
const messageChannel = new ReliableChannel(
|
||||
node,
|
||||
sdsMessageChannel,
|
||||
encoder,
|
||||
decoder,
|
||||
contentTopic,
|
||||
encryption,
|
||||
options
|
||||
);
|
||||
|
||||
const autoStart = options?.autoStart ?? true;
|
||||
if (autoStart) {
|
||||
await messageChannel.start();
|
||||
messageChannel.start();
|
||||
}
|
||||
|
||||
return messageChannel;
|
||||
@ -317,51 +316,32 @@ export class ReliableChannel<
|
||||
|
||||
// `payload` wrapped in SDS
|
||||
const sdsPayload = sdsMessage.encode();
|
||||
|
||||
const wakuMessage = {
|
||||
payload: sdsPayload
|
||||
};
|
||||
const encPayload = await this.encryption.encrypt(sdsPayload);
|
||||
|
||||
const messageId = ReliableChannel.getMessageId(messagePayload);
|
||||
|
||||
// TODO: should the encoder give me the message hash?
|
||||
// Encoding now to fail early, used later to get message hash
|
||||
const protoMessage = await this.encoder.toProtoObj(wakuMessage);
|
||||
if (!protoMessage) {
|
||||
this.safeSendEvent("sending-message-irrecoverable-error", {
|
||||
detail: {
|
||||
messageId: messageId,
|
||||
error: "could not encode message"
|
||||
}
|
||||
});
|
||||
return { success: false };
|
||||
}
|
||||
const retrievalHint = messageHash(
|
||||
this.encoder.pubsubTopic,
|
||||
protoMessage
|
||||
);
|
||||
|
||||
this.safeSendEvent("sending-message", {
|
||||
detail: messageId
|
||||
});
|
||||
|
||||
const sendRes = await this._send(this.encoder, wakuMessage);
|
||||
const retrievalHint = await this._send(this.contentTopic, encPayload);
|
||||
|
||||
// If it's a recoverable failure, we will try again to send later
|
||||
// If not, then we should error to the user now
|
||||
for (const { error } of sendRes.failures) {
|
||||
if (IRRECOVERABLE_SENDING_ERRORS.includes(error)) {
|
||||
// Not recoverable, best to return it
|
||||
log.error("Irrecoverable error, cannot send message: ", error);
|
||||
this.safeSendEvent("sending-message-irrecoverable-error", {
|
||||
detail: {
|
||||
messageId,
|
||||
error
|
||||
}
|
||||
});
|
||||
return { success: false, retrievalHint };
|
||||
}
|
||||
}
|
||||
// for (const { error } of sendRes.failures) {
|
||||
// if (IRRECOVERABLE_SENDING_ERRORS.includes(error)) {
|
||||
// // Not recoverable, best to return it
|
||||
// log.error("Irrecoverable error, cannot send message: ", error);
|
||||
// this.safeSendEvent("sending-message-irrecoverable-error", {
|
||||
// detail: {
|
||||
// messageId,
|
||||
// error
|
||||
// }
|
||||
// });
|
||||
// return { success: false, retrievalHint };
|
||||
// }
|
||||
// }
|
||||
// TODO: if failure, process it
|
||||
|
||||
return {
|
||||
success: true,
|
||||
@ -381,26 +361,35 @@ export class ReliableChannel<
|
||||
});
|
||||
}
|
||||
|
||||
private async subscribe(): Promise<boolean> {
|
||||
private subscribe(): void {
|
||||
this.assertStarted();
|
||||
return this._subscribe(this.decoder, async (message: T) => {
|
||||
await this.processIncomingMessage(message);
|
||||
this.node.messageEmitter.addEventListener(this.contentTopic, (event) => {
|
||||
const { payload, messageHash } = event.detail;
|
||||
// messageHash is the retrievalHint
|
||||
void this.processIncomingMessage(payload, messageHash);
|
||||
});
|
||||
|
||||
this._subscribe([this.contentTopic]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Don't forget to call `this.messageChannel.sweepIncomingBuffer();` once done.
|
||||
* @param msg
|
||||
* @private
|
||||
* @param payload
|
||||
*/
|
||||
private async processIncomingMessage<T extends IDecodedMessage>(
|
||||
msg: T
|
||||
private async processIncomingMessage(
|
||||
payload: Uint8Array,
|
||||
retrievalHint: Uint8Array
|
||||
): Promise<void> {
|
||||
// New message arrives, we need to unwrap it first
|
||||
const sdsMessage = SdsMessage.decode(msg.payload);
|
||||
// Decrypt first
|
||||
// TODO: skip on failure
|
||||
const decPayload = await this.encryption.decrypt(payload);
|
||||
|
||||
// Unwrap SDS layer
|
||||
const sdsMessage = SdsMessage.decode(decPayload);
|
||||
|
||||
if (!sdsMessage) {
|
||||
log.error("could not SDS decode message", msg);
|
||||
log.error("could not SDS decode message");
|
||||
return;
|
||||
}
|
||||
|
||||
@ -412,48 +401,33 @@ export class ReliableChannel<
|
||||
return;
|
||||
}
|
||||
|
||||
const retrievalHint = msg.hash;
|
||||
log.info(`processing message ${sdsMessage.messageId}:${msg.hashStr}`);
|
||||
log.info(
|
||||
`processing message ${sdsMessage.messageId}:${bytesToHex(retrievalHint)}`
|
||||
);
|
||||
// SDS Message decoded, let's pass it to the channel so we can learn about
|
||||
// missing messages or the status of previous outgoing messages
|
||||
this.messageChannel.pushIncomingMessage(sdsMessage, retrievalHint);
|
||||
|
||||
this.missingMessageRetriever?.removeMissingMessage(sdsMessage.messageId);
|
||||
// TODO
|
||||
// this.missingMessageRetriever?.removeMissingMessage(sdsMessage.messageId);
|
||||
|
||||
if (sdsMessage.content && sdsMessage.content.length > 0) {
|
||||
// Now, process the message with callback
|
||||
|
||||
// Overrides msg.payload with unwrapped payload
|
||||
// TODO: can we do better?
|
||||
const { payload: _p, ...allButPayload } = msg;
|
||||
const unwrappedMessage = Object.assign(allButPayload, {
|
||||
payload: sdsMessage.content,
|
||||
hash: msg.hash,
|
||||
hashStr: msg.hashStr,
|
||||
version: msg.version,
|
||||
contentTopic: msg.contentTopic,
|
||||
pubsubTopic: msg.pubsubTopic,
|
||||
timestamp: msg.timestamp,
|
||||
rateLimitProof: msg.rateLimitProof,
|
||||
ephemeral: msg.ephemeral,
|
||||
meta: msg.meta
|
||||
});
|
||||
|
||||
this.safeSendEvent("message-received", {
|
||||
detail: unwrappedMessage as unknown as T
|
||||
detail: sdsMessage.content
|
||||
});
|
||||
}
|
||||
|
||||
this.queueProcessTasks();
|
||||
}
|
||||
|
||||
private async processIncomingMessages<T extends IDecodedMessage>(
|
||||
messages: T[]
|
||||
): Promise<void> {
|
||||
for (const message of messages) {
|
||||
await this.processIncomingMessage(message);
|
||||
}
|
||||
}
|
||||
// private async processIncomingMessages<T extends IDecodedMessage>(
|
||||
// messages: T[]
|
||||
// ): Promise<void> {
|
||||
// for (const message of messages) {
|
||||
// await this.processIncomingMessage(message.payload);
|
||||
// }
|
||||
// }
|
||||
|
||||
// TODO: For now we only queue process tasks for incoming messages
|
||||
// As this is where there is most volume
|
||||
@ -472,17 +446,17 @@ export class ReliableChannel<
|
||||
}
|
||||
}
|
||||
|
||||
public async start(): Promise<boolean> {
|
||||
if (this._started) return true;
|
||||
public start(): void {
|
||||
if (this._started) return;
|
||||
this._started = true;
|
||||
this.setupEventListeners();
|
||||
this.restartSync();
|
||||
this.startSweepIncomingBufferLoop();
|
||||
if (this._retrieve) {
|
||||
this.missingMessageRetriever?.start();
|
||||
this.queryOnConnect?.start();
|
||||
}
|
||||
return this.subscribe();
|
||||
// if (this._retrieve) {
|
||||
// this.missingMessageRetriever?.start();
|
||||
// this.queryOnConnect?.start();
|
||||
// }
|
||||
this.subscribe();
|
||||
}
|
||||
|
||||
public stop(): void {
|
||||
@ -490,8 +464,8 @@ export class ReliableChannel<
|
||||
this._started = false;
|
||||
this.stopSync();
|
||||
this.stopSweepIncomingBufferLoop();
|
||||
this.missingMessageRetriever?.stop();
|
||||
this.queryOnConnect?.stop();
|
||||
// this.missingMessageRetriever?.stop();
|
||||
// this.queryOnConnect?.stop();
|
||||
// TODO unsubscribe
|
||||
// TODO unsetMessageListeners
|
||||
}
|
||||
@ -665,27 +639,27 @@ export class ReliableChannel<
|
||||
}
|
||||
);
|
||||
|
||||
this.messageChannel.addEventListener(
|
||||
MessageChannelEvent.InMessageMissing,
|
||||
(event) => {
|
||||
for (const { messageId, retrievalHint } of event.detail) {
|
||||
if (retrievalHint && this.missingMessageRetriever) {
|
||||
this.missingMessageRetriever.addMissingMessage(
|
||||
messageId,
|
||||
retrievalHint
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
);
|
||||
// this.messageChannel.addEventListener(
|
||||
// MessageChannelEvent.InMessageMissing,
|
||||
// (event) => {
|
||||
// for (const { messageId, retrievalHint } of event.detail) {
|
||||
// if (retrievalHint && this.missingMessageRetriever) {
|
||||
// this.missingMessageRetriever.addMissingMessage(
|
||||
// messageId,
|
||||
// retrievalHint
|
||||
// );
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
// );
|
||||
|
||||
if (this.queryOnConnect) {
|
||||
this.queryOnConnect.addEventListener(
|
||||
QueryOnConnectEvent.MessagesRetrieved,
|
||||
(event) => {
|
||||
void this.processIncomingMessages(event.detail);
|
||||
}
|
||||
);
|
||||
}
|
||||
// if (this.queryOnConnect) {
|
||||
// this.queryOnConnect.addEventListener(
|
||||
// QueryOnConnectEvent.MessagesRetrieved,
|
||||
// (event) => {
|
||||
// void this.processIncomingMessages(event.detail);
|
||||
// }
|
||||
// );
|
||||
// }
|
||||
}
|
||||
}
|
||||
|
||||
@ -5,8 +5,14 @@ import {
|
||||
TypedEventEmitter
|
||||
} from "@libp2p/interface";
|
||||
import type { MultiaddrInput } from "@multiformats/multiaddr";
|
||||
import { ConnectionManager, createDecoder, createEncoder } from "@waku/core";
|
||||
import type {
|
||||
import {
|
||||
ConnectionManager,
|
||||
createDecoder,
|
||||
createEncoder,
|
||||
DecodedMessage
|
||||
} from "@waku/core";
|
||||
import {
|
||||
ContentTopic,
|
||||
CreateDecoderParams,
|
||||
CreateEncoderParams,
|
||||
CreateNodeOptions,
|
||||
@ -15,20 +21,22 @@ import type {
|
||||
IEncoder,
|
||||
IFilter,
|
||||
ILightPush,
|
||||
IMessageEmitter,
|
||||
IRelay,
|
||||
IRoutingInfo,
|
||||
IStore,
|
||||
IWaku,
|
||||
IWakuEventEmitter,
|
||||
Libp2p,
|
||||
NetworkConfig
|
||||
NetworkConfig,
|
||||
PubsubTopic
|
||||
} from "@waku/interfaces";
|
||||
import {
|
||||
DefaultNetworkConfig,
|
||||
HealthStatus,
|
||||
Protocols
|
||||
} from "@waku/interfaces";
|
||||
import { createRoutingInfo, Logger } from "@waku/utils";
|
||||
import { createRoutingInfo, Logger, pushOrInitMapSet } from "@waku/utils";
|
||||
|
||||
import { Filter } from "../filter/index.js";
|
||||
import { HealthIndicator } from "../health_indicator/index.js";
|
||||
@ -54,6 +62,7 @@ export class WakuNode implements IWaku {
|
||||
public lightPush?: ILightPush;
|
||||
|
||||
public readonly events: IWakuEventEmitter = new TypedEventEmitter();
|
||||
public readonly messageEmitter: IMessageEmitter = new TypedEventEmitter();
|
||||
|
||||
private readonly networkConfig: NetworkConfig;
|
||||
|
||||
@ -134,6 +143,41 @@ export class WakuNode implements IWaku {
|
||||
);
|
||||
}
|
||||
|
||||
public subscribe(contentTopics: ContentTopic[]): void {
|
||||
// Group decoders by pubsubTopics in case they spread across several shards
|
||||
const ctToDecoders: Map<
|
||||
PubsubTopic,
|
||||
Set<IDecoder<DecodedMessage>>
|
||||
> = new Map();
|
||||
for (const contentTopic of contentTopics) {
|
||||
const decoder = this.createDecoder({ contentTopic });
|
||||
pushOrInitMapSet(ctToDecoders, decoder.pubsubTopic, decoder);
|
||||
}
|
||||
|
||||
if (this.filter) {
|
||||
for (const [_, decoders] of ctToDecoders) {
|
||||
void this.filter
|
||||
.subscribe(
|
||||
Array.from(decoders),
|
||||
this.emitIncomingMessages.bind(this, Array.from(contentTopics))
|
||||
)
|
||||
.then((_) => {
|
||||
// TODO: emit irrecoverable errors
|
||||
})
|
||||
.catch((_) => {
|
||||
// TODO: emit irrecoverable errors
|
||||
});
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.relay) {
|
||||
throw "not implemented";
|
||||
}
|
||||
throw "no subscribe protocol available";
|
||||
}
|
||||
|
||||
public get peerId(): PeerId {
|
||||
return this.libp2p.peerId;
|
||||
}
|
||||
@ -282,10 +326,26 @@ export class WakuNode implements IWaku {
|
||||
});
|
||||
}
|
||||
|
||||
private createRoutingInfo(
|
||||
public createRoutingInfo(
|
||||
contentTopic?: string,
|
||||
shardId?: number
|
||||
): IRoutingInfo {
|
||||
return createRoutingInfo(this.networkConfig, { contentTopic, shardId });
|
||||
}
|
||||
|
||||
private emitIncomingMessages(
|
||||
contentTopics: ContentTopic[],
|
||||
message: {
|
||||
contentTopic: ContentTopic;
|
||||
payload: Uint8Array;
|
||||
}
|
||||
): void {
|
||||
if (contentTopics.includes(message.contentTopic)) {
|
||||
this.messageEmitter.dispatchEvent(
|
||||
new CustomEvent<Uint8Array>(message.contentTopic, {
|
||||
detail: message.payload
|
||||
})
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -124,14 +124,14 @@ export class ServiceNodesFleet {
|
||||
}
|
||||
|
||||
class MultipleNodesMessageCollector {
|
||||
public callback: (msg: IDecodedMessage) => void = () => {};
|
||||
protected messageList: Array<IDecodedMessage> = [];
|
||||
public callback: (msg: Partial<IDecodedMessage>) => void = () => {};
|
||||
protected messageList: Array<Partial<IDecodedMessage>> = [];
|
||||
public constructor(
|
||||
private messageCollectors: MessageCollector[],
|
||||
private relayNodes?: ServiceNode[],
|
||||
private strictChecking: boolean = false
|
||||
) {
|
||||
this.callback = (msg: IDecodedMessage): void => {
|
||||
this.callback = (msg: Partial<IDecodedMessage>): void => {
|
||||
log.info("Got a message");
|
||||
this.messageList.push(msg);
|
||||
};
|
||||
@ -153,7 +153,9 @@ class MultipleNodesMessageCollector {
|
||||
}
|
||||
}
|
||||
|
||||
public getMessage(index: number): MessageRpcResponse | IDecodedMessage {
|
||||
public getMessage(
|
||||
index: number
|
||||
): MessageRpcResponse | Partial<IDecodedMessage> {
|
||||
return this.messageList[index];
|
||||
}
|
||||
|
||||
|
||||
@ -7,10 +7,17 @@ import type {
|
||||
RelayNode
|
||||
} from "@waku/interfaces";
|
||||
import { Protocols } from "@waku/interfaces";
|
||||
import { generateSymmetricKey } from "@waku/message-encryption";
|
||||
import {
|
||||
comparePublicKeys,
|
||||
generatePrivateKey,
|
||||
generateSymmetricKey,
|
||||
getPublicKey
|
||||
} from "@waku/message-encryption";
|
||||
import {
|
||||
createDecoder,
|
||||
createEncoder
|
||||
createEncoder,
|
||||
SymmetricDecryption,
|
||||
SymmetricDecryptionResult
|
||||
} from "@waku/message-encryption/symmetric";
|
||||
import { createRelayNode } from "@waku/relay";
|
||||
import {
|
||||
@ -29,8 +36,11 @@ import {
|
||||
makeLogFileName,
|
||||
NOISE_KEY_1,
|
||||
NOISE_KEY_2,
|
||||
runMultipleNodes,
|
||||
ServiceNode,
|
||||
tearDownNodes
|
||||
ServiceNodesFleet,
|
||||
tearDownNodes,
|
||||
teardownNodesWithRedundancy
|
||||
} from "../src/index.js";
|
||||
|
||||
const TestContentTopic = "/test/1/waku/utf8";
|
||||
@ -291,3 +301,148 @@ describe("User Agent", function () {
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("Waku API", function () {
|
||||
describe("WakuNode.subscribe (light node)", function () {
|
||||
this.timeout(100000);
|
||||
let waku: LightNode;
|
||||
let serviceNodes: ServiceNodesFleet;
|
||||
const messageText = "some message";
|
||||
const messagePayload = utf8ToBytes(messageText);
|
||||
|
||||
beforeEachCustom(this, async () => {
|
||||
[serviceNodes, waku] = await runMultipleNodes(
|
||||
this.ctx,
|
||||
TestRoutingInfo,
|
||||
undefined
|
||||
);
|
||||
});
|
||||
|
||||
afterEachCustom(this, async () => {
|
||||
await teardownNodesWithRedundancy(serviceNodes, waku);
|
||||
});
|
||||
|
||||
it("Subscribe and receive messages on 2 different content topics", async function () {
|
||||
// Subscribe to the first content topic and send a message.
|
||||
waku.messageEmitter.addEventListener(TestContentTopic, (event) => {
|
||||
// TODO: fix the callback type
|
||||
serviceNodes.messageCollector.callback({
|
||||
contentTopic: TestContentTopic,
|
||||
payload: event.detail
|
||||
});
|
||||
});
|
||||
waku.subscribe([TestContentTopic]);
|
||||
|
||||
await waku.lightPush.send(TestEncoder, { payload: messagePayload });
|
||||
expect(await serviceNodes.messageCollector.waitForMessages(1)).to.eq(
|
||||
true,
|
||||
"Waiting for the first message"
|
||||
);
|
||||
serviceNodes.messageCollector.verifyReceivedMessage(0, {
|
||||
expectedMessageText: messageText,
|
||||
expectedContentTopic: TestContentTopic,
|
||||
expectedPubsubTopic: TestRoutingInfo.pubsubTopic
|
||||
});
|
||||
|
||||
// Modify subscription to include a new content topic and send a message.
|
||||
const newMessageText = "Filtering still works!";
|
||||
const newContentTopic = "/test/2/waku-filter/default";
|
||||
const newRoutingInfo = createRoutingInfo(DefaultTestNetworkConfig, {
|
||||
contentTopic: newContentTopic
|
||||
});
|
||||
const newEncoder = createPlainEncoder({
|
||||
contentTopic: newContentTopic,
|
||||
routingInfo: newRoutingInfo
|
||||
});
|
||||
// subscribe to second content topic
|
||||
waku.messageEmitter.addEventListener(newContentTopic, (event) => {
|
||||
// TODO: fix the callback type
|
||||
serviceNodes.messageCollector.callback({
|
||||
contentTopic: TestContentTopic,
|
||||
payload: event.detail
|
||||
});
|
||||
});
|
||||
waku.subscribe([newContentTopic]);
|
||||
|
||||
await waku.lightPush.send(newEncoder, {
|
||||
payload: utf8ToBytes(newMessageText)
|
||||
});
|
||||
expect(await serviceNodes.messageCollector.waitForMessages(2)).to.eq(
|
||||
true,
|
||||
"Waiting for the second message"
|
||||
);
|
||||
serviceNodes.messageCollector.verifyReceivedMessage(1, {
|
||||
expectedContentTopic: newContentTopic,
|
||||
expectedMessageText: newMessageText,
|
||||
expectedPubsubTopic: TestRoutingInfo.pubsubTopic
|
||||
});
|
||||
|
||||
// Send another message on the initial content topic to verify it still works.
|
||||
const thirdMessageText = "Filtering still works on first subscription!";
|
||||
const thirdMessagePayload = { payload: utf8ToBytes(thirdMessageText) };
|
||||
await waku.lightPush.send(TestEncoder, thirdMessagePayload);
|
||||
expect(await serviceNodes.messageCollector.waitForMessages(3)).to.eq(
|
||||
true,
|
||||
"Waiting for the third message"
|
||||
);
|
||||
serviceNodes.messageCollector.verifyReceivedMessage(2, {
|
||||
expectedMessageText: thirdMessageText,
|
||||
expectedContentTopic: TestContentTopic,
|
||||
expectedPubsubTopic: TestRoutingInfo.pubsubTopic
|
||||
});
|
||||
});
|
||||
|
||||
it("Subscribe and receive messages encrypted with AES", async function () {
|
||||
const symKey = generateSymmetricKey();
|
||||
const senderPrivKey = generatePrivateKey();
|
||||
// TODO: For now, still using encoder
|
||||
const newEncoder = createEncoder({
|
||||
contentTopic: TestContentTopic,
|
||||
routingInfo: TestRoutingInfo,
|
||||
symKey,
|
||||
sigPrivKey: senderPrivKey
|
||||
});
|
||||
|
||||
// Setup payload decryption
|
||||
const symDecryption = new SymmetricDecryption(symKey);
|
||||
|
||||
// subscribe to second content topic
|
||||
waku.messageEmitter.addEventListener(TestContentTopic, (event) => {
|
||||
const encryptedPayload = event.detail;
|
||||
void symDecryption
|
||||
.decrypt(encryptedPayload)
|
||||
.then((decryptionResult: SymmetricDecryptionResult | undefined) => {
|
||||
if (!decryptionResult) return;
|
||||
serviceNodes.messageCollector.callback({
|
||||
contentTopic: TestContentTopic,
|
||||
payload: decryptionResult.payload
|
||||
});
|
||||
|
||||
// TODO: probably best to adapt the message collector
|
||||
expect(decryptionResult?.signature).to.not.be.undefined;
|
||||
expect(
|
||||
comparePublicKeys(
|
||||
getPublicKey(senderPrivKey),
|
||||
decryptionResult?.signaturePublicKey
|
||||
)
|
||||
);
|
||||
// usually best to ignore decryption failure
|
||||
});
|
||||
});
|
||||
waku.subscribe([TestContentTopic]);
|
||||
|
||||
await waku.lightPush.send(newEncoder, {
|
||||
payload: utf8ToBytes(messageText)
|
||||
});
|
||||
expect(await serviceNodes.messageCollector.waitForMessages(1)).to.eq(
|
||||
true,
|
||||
"Waiting for the message"
|
||||
);
|
||||
serviceNodes.messageCollector.verifyReceivedMessage(1, {
|
||||
expectedContentTopic: TestContentTopic,
|
||||
expectedMessageText: messageText,
|
||||
expectedPubsubTopic: TestRoutingInfo.pubsubTopic
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user