status-im/js-waku
2
0
Fork 0
mirror of https://github.com/status-im/js-waku.git synced 2025-02-24 10:58:18 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #1196 from waku-org/chore/update-proto-definition

Browse Source
This commit is contained in:
fryorcraken.eth 2023-02-27 14:12:09 +11:00 committed by GitHub
commit 7a3c9a8df2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
32 changed files with 276 additions and 303 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
packages/core/src/lib/filter/filter_rpc.ts
View File

@ -8,16 +8,16 @@ export type ContentFilter = {
/**
* FilterRPC represents a message conforming to the Waku Filter protocol
*/
export class FilterRPC {
public constructor(public proto: proto.FilterRPC) {}
export class FilterRpc {
public constructor(public proto: proto.FilterRpc) {}
static createRequest(
topic: string,
contentFilters: ContentFilter[],
requestId?: string,
subscribe = true
): FilterRPC {
return new FilterRPC({
): FilterRpc {
return new FilterRpc({
requestId: requestId || uuid(),
request: {
subscribe,
@ -31,11 +31,11 @@ export class FilterRPC {
/**
*
* @param bytes Uint8Array of bytes from a FilterRPC message
* @returns FilterRPC
* @returns FilterRpc
*/
static decode(bytes: Uint8Array): FilterRPC {
const res = proto.FilterRPC.decode(bytes);
return new FilterRPC(res);
static decode(bytes: Uint8Array): FilterRpc {
const res = proto.FilterRpc.decode(bytes);
return new FilterRpc(res);
}
/**
@ -43,14 +43,14 @@ export class FilterRPC {
* @returns Uint8Array
*/
encode(): Uint8Array {
return proto.FilterRPC.encode(this.proto);
return proto.FilterRpc.encode(this.proto);
}
get push(): proto.MessagePush | undefined {
return this.proto.push;
}
get requestId(): string | undefined {
get requestId(): string {
return this.proto.requestId;
}
}

12
packages/core/src/lib/filter/index.ts
View File

@ -21,7 +21,7 @@ import { DefaultPubSubTopic } from "../constants.js";
import { groupByContentTopic } from "../group_by.js";
import { toProtoMessage } from "../to_proto_message.js";
import { ContentFilter, FilterRPC } from "./filter_rpc.js";
import { ContentFilter, FilterRpc } from "./filter_rpc.js";
export { ContentFilter };
@ -75,7 +75,7 @@ class Filter extends BaseProtocol implements IFilter {
const contentFilters = contentTopics.map((contentTopic) => ({
contentTopic,
}));
const request = FilterRPC.createRequest(
const request = FilterRpc.createRequest(
pubSubTopic,
contentFilters,
undefined,
@ -83,10 +83,6 @@ class Filter extends BaseProtocol implements IFilter {
);
const requestId = request.requestId;
if (!requestId)
throw new Error(
"Internal error: createRequest expected to set `requestId`"
);
const peer = await this.getPeer(opts?.peerId);
const stream = await this.newStream(peer);
@ -128,7 +124,7 @@ class Filter extends BaseProtocol implements IFilter {
try {
pipe(streamData.stream, lp.decode(), async (source) => {
for await (const bytes of source) {
const res = FilterRPC.decode(bytes.slice());
const res = FilterRpc.decode(bytes.slice());
if (res.requestId && res.push?.messages?.length) {
await this.pushMessages(res.requestId, res.push.messages);
}
@ -228,7 +224,7 @@ class Filter extends BaseProtocol implements IFilter {
requestId: string,
peer: Peer
): Promise<void> {
const unsubscribeRequest = FilterRPC.createRequest(
const unsubscribeRequest = FilterRpc.createRequest(
topic,
contentFilters,
requestId,

6
packages/core/src/lib/light_push/index.ts
View File

@ -18,7 +18,7 @@ import { Uint8ArrayList } from "uint8arraylist";
import { BaseProtocol } from "../base_protocol.js";
import { DefaultPubSubTopic } from "../constants.js";
import { PushRPC } from "./push_rpc.js";
import { PushRpc } from "./push_rpc.js";
const log = debug("waku:light-push");
@ -54,7 +54,7 @@ class LightPush extends BaseProtocol implements ILightPush {
log("Failed to encode to protoMessage, aborting push");
return { recipients };
}
const query = PushRPC.createRequest(protoMessage, pubSubTopic);
const query = PushRpc.createRequest(protoMessage, pubSubTopic);
const res = await pipe(
[query.encode()],
lp.encode(),
@ -68,7 +68,7 @@ class LightPush extends BaseProtocol implements ILightPush {
bytes.append(chunk);
});
const response = PushRPC.decode(bytes).response;
const response = PushRpc.decode(bytes).response;
if (!response) {
log("No response in PushRPC");

18
packages/core/src/lib/light_push/push_rpc.ts
View File

@ -2,30 +2,30 @@ import { proto_lightpush as proto } from "@waku/proto";
import type { Uint8ArrayList } from "uint8arraylist";
import { v4 as uuid } from "uuid";
export class PushRPC {
public constructor(public proto: proto.PushRPC) {}
export class PushRpc {
public constructor(public proto: proto.PushRpc) {}
static createRequest(
message: proto.WakuMessage,
pubSubTopic: string
): PushRPC {
return new PushRPC({
): PushRpc {
return new PushRpc({
requestId: uuid(),
request: {
message: message,
pubSubTopic: pubSubTopic,
pubsubTopic: pubSubTopic,
},
response: undefined,
});
}
static decode(bytes: Uint8ArrayList): PushRPC {
const res = proto.PushRPC.decode(bytes);
return new PushRPC(res);
static decode(bytes: Uint8ArrayList): PushRpc {
const res = proto.PushRpc.decode(bytes);
return new PushRpc(res);
}
encode(): Uint8Array {
return proto.PushRPC.encode(this.proto);
return proto.PushRpc.encode(this.proto);
}
get query(): proto.PushRequest | undefined {

6
packages/core/src/lib/message/topic_only_message.ts
View File

@ -9,7 +9,7 @@ import debug from "debug";
const log = debug("waku:message:topic-only");
export class TopicOnlyMessage implements IDecodedMessage {
public payload: undefined;
public payload: Uint8Array = new Uint8Array();
public rateLimitProof: undefined;
public timestamp: undefined;
public ephemeral: undefined;
@ -17,7 +17,7 @@ export class TopicOnlyMessage implements IDecodedMessage {
constructor(private proto: ProtoTopicOnlyMessage) {}
get contentTopic(): string {
return this.proto.contentTopic ?? "";
return this.proto.contentTopic;
}
}
@ -29,7 +29,7 @@ export class TopicOnlyDecoder implements IDecoder<TopicOnlyMessage> {
log("Message decoded", protoMessage);
return Promise.resolve({
contentTopic: protoMessage.contentTopic,
payload: undefined,
payload: new Uint8Array(),
rateLimitProof: undefined,
timestamp: undefined,
version: undefined,

35
packages/core/src/lib/message/version_0.ts
View File

@ -19,22 +19,15 @@ export { proto };
export class DecodedMessage implements IDecodedMessage {
constructor(protected proto: proto.WakuMessage) {}
get _rawPayload(): Uint8Array | undefined {
if (this.proto.payload) {
return new Uint8Array(this.proto.payload);
}
return;
}
get ephemeral(): boolean {
return Boolean(this.proto.ephemeral);
}
get payload(): Uint8Array | undefined {
return this._rawPayload;
get payload(): Uint8Array {
return this.proto.payload;
}
get contentTopic(): string | undefined {
get contentTopic(): string {
return this.proto.contentTopic;
}
@ -51,18 +44,15 @@ export class DecodedMessage implements IDecodedMessage {
const timestamp = this.proto.timestamp / OneMillion;
return new Date(Number(timestamp));
}
if (this.proto.timestampDeprecated) {
return new Date(this.proto.timestampDeprecated * 1000);
}
return;
} catch (e) {
return;
}
return;
}
get version(): number {
// https://github.com/status-im/js-waku/issues/921
// https://rfc.vac.dev/spec/14/
// > If omitted, the value SHOULD be interpreted as version 0.
return this.proto.version ?? 0;
}
@ -115,8 +105,8 @@ export class Decoder implements IDecoder<DecodedMessage> {
const protoMessage = proto.WakuMessage.decode(bytes);
log("Message decoded", protoMessage);
return Promise.resolve({
payload: protoMessage.payload ?? undefined,
contentTopic: protoMessage.contentTopic ?? undefined,
payload: protoMessage.payload,
contentTopic: protoMessage.contentTopic,
version: protoMessage.version ?? undefined,
timestamp: protoMessage.timestamp ?? undefined,
rateLimitProof: protoMessage.rateLimitProof ?? undefined,
@ -127,12 +117,9 @@ export class Decoder implements IDecoder<DecodedMessage> {
async fromProtoObj(
proto: IProtoMessage
): Promise<DecodedMessage | undefined> {
// https://github.com/status-im/js-waku/issues/921
if (proto.version === undefined) {
proto.version = 0;
}
if (proto.version !== Version) {
// https://rfc.vac.dev/spec/14/
// > If omitted, the value SHOULD be interpreted as version 0.
if (proto.version ?? 0 !== Version) {
log(
"Failed to decode due to incorrect version, expected:",
Version,

24
packages/core/src/lib/store/history_rpc.ts
View File

@ -19,8 +19,8 @@ export interface Params {
cursor?: proto.Index;
}
export class HistoryRPC {
private constructor(public readonly proto: proto.HistoryRPC) {}
export class HistoryRpc {
private constructor(public readonly proto: proto.HistoryRpc) {}
get query(): proto.HistoryQuery | undefined {
return this.proto.query;
@ -33,7 +33,7 @@ export class HistoryRPC {
/**
* Create History Query.
*/
static createQuery(params: Params): HistoryRPC {
static createQuery(params: Params): HistoryRpc {
const contentFilters = params.contentTopics.map((contentTopic) => {
return { contentTopic };
});
@ -56,10 +56,10 @@ export class HistoryRPC {
// milliseconds 10^-3 to nanoseconds 10^-9
endTime = BigInt(params.endTime.valueOf()) * OneMillion;
}
return new HistoryRPC({
return new HistoryRpc({
requestId: uuid(),
query: {
pubSubTopic: params.pubSubTopic,
pubsubTopic: params.pubSubTopic,
contentFilters,
pagingInfo,
startTime,
@ -69,13 +69,13 @@ export class HistoryRPC {
});
}
decode(bytes: Uint8ArrayList): HistoryRPC {
const res = proto.HistoryRPC.decode(bytes);
return new HistoryRPC(res);
decode(bytes: Uint8ArrayList): HistoryRpc {
const res = proto.HistoryRpc.decode(bytes);
return new HistoryRpc(res);
}
encode(): Uint8Array {
return proto.HistoryRPC.encode(this.proto);
return proto.HistoryRpc.encode(this.proto);
}
}
@ -84,10 +84,10 @@ function directionToProto(
): proto.PagingInfo.Direction {
switch (pageDirection) {
case PageDirection.BACKWARD:
return proto.PagingInfo.Direction.DIRECTION_BACKWARD_UNSPECIFIED;
return proto.PagingInfo.Direction.BACKWARD;
case PageDirection.FORWARD:
return proto.PagingInfo.Direction.DIRECTION_FORWARD;
return proto.PagingInfo.Direction.FORWARD;
default:
return proto.PagingInfo.Direction.DIRECTION_BACKWARD_UNSPECIFIED;
return proto.PagingInfo.Direction.BACKWARD;
}
}

14
packages/core/src/lib/store/index.ts
View File

@ -6,7 +6,6 @@ import {
Cursor,
IDecodedMessage,
IDecoder,
Index,
IStore,
ProtocolCreateOptions,
} from "@waku/interfaces";
@ -22,7 +21,7 @@ import { BaseProtocol } from "../base_protocol.js";
import { DefaultPubSubTopic } from "../constants.js";
import { toProtoMessage } from "../to_proto_message.js";
import { HistoryRPC, PageDirection, Params } from "./history_rpc.js";
import { HistoryRpc, PageDirection, Params } from "./history_rpc.js";
import HistoryError = proto.HistoryResponse.HistoryError;
@ -262,7 +261,7 @@ async function* paginate<T extends IDecodedMessage>(
while (true) {
queryOpts.cursor = currentCursor;
const historyRpcQuery = HistoryRPC.createQuery(queryOpts);
const historyRpcQuery = HistoryRpc.createQuery(queryOpts);
log(
"Querying store peer",
@ -294,10 +293,7 @@ async function* paginate<T extends IDecodedMessage>(
const response = reply.response as proto.HistoryResponse;
if (
response.error &&
response.error !== HistoryError.ERROR_NONE_UNSPECIFIED
) {
if (response.error && response.error !== HistoryError.NONE) {
throw "History response contains an Error: " + response.error;
}
@ -353,7 +349,7 @@ export function isDefined<T>(msg: T | undefined): msg is T {
export async function createCursor(
message: IDecodedMessage,
pubsubTopic: string = DefaultPubSubTopic
): Promise<Index> {
): Promise<Cursor> {
if (
!message ||
!message.timestamp ||
@ -373,7 +369,7 @@ export async function createCursor(
digest,
pubsubTopic,
senderTime: messageTime,
receivedTime: messageTime,
receiverTime: messageTime,
};
}

1
packages/core/src/lib/to_proto_message.spec.ts
View File

@ -6,6 +6,7 @@ import { toProtoMessage } from "./to_proto_message.js";
describe("to proto message", () => {
it("Fields are not dropped", () => {
const wire: WakuMessageProto = {
payload: new Uint8Array(),
contentTopic: "foo",
};

4
packages/core/src/lib/to_proto_message.ts
View File

@ -2,8 +2,8 @@ import { IProtoMessage } from "@waku/interfaces";
import { WakuMessage as WakuMessageProto } from "@waku/proto";
const EmptyMessage: IProtoMessage = {
payload: undefined,
contentTopic: undefined,
payload: new Uint8Array(),
contentTopic: "",
version: undefined,
timestamp: undefined,
rateLimitProof: undefined,

10
packages/interfaces/src/message.ts
View File

@ -13,8 +13,8 @@ export interface IRateLimitProof {
* Field types matches the protobuf type over the wire
*/
export interface IProtoMessage {
payload: Uint8Array | undefined;
contentTopic: string | undefined;
payload: Uint8Array;
contentTopic: string;
version: number | undefined;
timestamp: bigint | undefined;
rateLimitProof: IRateLimitProof | undefined;
@ -25,7 +25,7 @@ export interface IProtoMessage {
* Interface for messages to encode and send.
*/
export interface IMessage {
payload?: Uint8Array;
payload: Uint8Array;
timestamp?: Date;
rateLimitProof?: IRateLimitProof;
}
@ -48,8 +48,8 @@ export interface IEncoder {
}
export interface IDecodedMessage {
payload: Uint8Array | undefined;
contentTopic: string | undefined;
payload: Uint8Array;
contentTopic: string;
timestamp: Date | undefined;
rateLimitProof: IRateLimitProof | undefined;
ephemeral: boolean | undefined;

6
packages/interfaces/src/protocols.ts
View File

@ -2,7 +2,7 @@ import type { PeerId } from "@libp2p/interface-peer-id";
import type { Peer, PeerStore } from "@libp2p/interface-peer-store";
import type { Libp2pOptions } from "libp2p";
import type { IMessage } from "./message.js";
import type { IDecodedMessage } from "./message.js";
export enum Protocols {
Relay = "relay",
@ -58,7 +58,9 @@ export type ProtocolOptions = {
peerId?: PeerId;
};
export type Callback<T extends IMessage> = (msg: T) => void | Promise<void>;
export type Callback<T extends IDecodedMessage> = (
msg: T
) => void | Promise<void>;
export interface SendResult {
recipients: PeerId[];

19
packages/interfaces/src/store.ts
View File

@ -11,19 +11,13 @@ export interface TimeFilter {
endTime: Date;
}
export interface Index {
digest?: Uint8Array;
receivedTime?: bigint;
senderTime?: bigint;
pubsubTopic?: string;
export interface Cursor {
digest: Uint8Array;
receiverTime: bigint;
senderTime: bigint;
pubsubTopic: string;
}
export type Cursor = {
digest?: Uint8Array;
senderTime?: bigint;
pubsubTopic?: string;
};
export type StoreQueryOptions = {
/**
* The direction in which pages are retrieved:
@ -45,7 +39,8 @@ export type StoreQueryOptions = {
*/
timeFilter?: TimeFilter;
/**
* Cursor as an index to start a query from.
* Cursor as an index to start a query from. Must be generated from a Waku
* Message.
*/
cursor?: Cursor;
} & ProtocolOptions;

8
packages/message-encryption/src/ecies.ts
View File

@ -45,10 +45,6 @@ export class Encoder implements IEncoder {
async toProtoObj(message: IMessage): Promise<IProtoMessage | undefined> {
const timestamp = message.timestamp ?? new Date();
if (!message.payload) {
log("No payload to encrypt, skipping: ", message);
return;
}
const preparedPayload = await preCipher(message.payload, this.sigPrivKey);
const payload = await encryptAsymmetric(preparedPayload, this.publicKey);
@ -113,10 +109,6 @@ export class Decoder extends DecoderV0 implements IDecoder<DecodedMessage> {
}
let payload;
if (!cipherPayload) {
log(`No payload to decrypt for contentTopic ${this.contentTopic}`);
return;
}
try {
payload = await decryptAsymmetric(cipherPayload, this.privateKey);

8
packages/message-encryption/src/symmetric.ts
View File

@ -44,10 +44,6 @@ export class Encoder implements IEncoder {
async toProtoObj(message: IMessage): Promise<IProtoMessage | undefined> {
const timestamp = message.timestamp ?? new Date();
if (!message.payload) {
log("No payload to encrypt, skipping: ", message);
return;
}
const preparedPayload = await preCipher(message.payload, this.sigPrivKey);
const payload = await encryptSymmetric(preparedPayload, this.symKey);
@ -112,10 +108,6 @@ export class Decoder extends DecoderV0 implements IDecoder<DecodedMessage> {
}
let payload;
if (!cipherPayload) {
log(`No payload to decrypt for contentTopic ${this.contentTopic}`);
return;
}
try {
payload = await decryptSymmetric(cipherPayload, this.symKey);

17
packages/proto/src/lib/filter.proto
View File

@ -1,23 +1,26 @@
// 12/WAKU2-FILTER rfc: https://rfc.vac.dev/spec/12/
// Protocol identifier: /vac/waku/filter/2.0.0-beta1
syntax = "proto3";
import "message.proto";
message FilterRequest {
optional bool subscribe = 1;
optional string topic = 2;
repeated ContentFilter content_filters = 3;
message ContentFilter {
optional string content_topic = 1;
string content_topic = 1;
}
bool subscribe = 1;
string topic = 2;
repeated ContentFilter content_filters = 3;
}
message MessagePush {
repeated WakuMessage messages = 1;
}
message FilterRPC {
optional string request_id = 1;
message FilterRpc {
string request_id = 1;
optional FilterRequest request = 2;
optional MessagePush push = 3;
}

66
packages/proto/src/lib/filter.ts
View File

@ -9,14 +9,14 @@ import type { Codec } from "protons-runtime";
import type { Uint8ArrayList } from "uint8arraylist";
export interface FilterRequest {
subscribe?: boolean;
topic?: string;
subscribe: boolean;
topic: string;
contentFilters: FilterRequest.ContentFilter[];
}
export namespace FilterRequest {
export interface ContentFilter {
contentTopic?: string;
contentTopic: string;
}
export namespace ContentFilter {
@ -30,7 +30,7 @@ export namespace FilterRequest {
w.fork();
}
if (obj.contentTopic != null) {
if (obj.contentTopic != null && obj.contentTopic !== "") {
w.uint32(10);
w.string(obj.contentTopic);
}
@ -40,7 +40,9 @@ export namespace FilterRequest {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
contentTopic: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -84,12 +86,12 @@ export namespace FilterRequest {
w.fork();
}
if (obj.subscribe != null) {
if (obj.subscribe != null && obj.subscribe !== false) {
w.uint32(8);
w.bool(obj.subscribe);
}
if (obj.topic != null) {
if (obj.topic != null && obj.topic !== "") {
w.uint32(18);
w.string(obj.topic);
}
@ -107,6 +109,8 @@ export namespace FilterRequest {
},
(reader, length) => {
const obj: any = {
subscribe: false,
topic: "",
contentFilters: [],
};
@ -218,24 +222,24 @@ export namespace MessagePush {
};
}
export interface FilterRPC {
requestId?: string;
export interface FilterRpc {
requestId: string;
request?: FilterRequest;
push?: MessagePush;
}
export namespace FilterRPC {
let _codec: Codec<FilterRPC>;
export namespace FilterRpc {
let _codec: Codec<FilterRpc>;
export const codec = (): Codec<FilterRPC> => {
export const codec = (): Codec<FilterRpc> => {
if (_codec == null) {
_codec = message<FilterRPC>(
_codec = message<FilterRpc>(
(obj, w, opts = {}) => {
if (opts.lengthDelimited !== false) {
w.fork();
}
if (obj.requestId != null) {
if (obj.requestId != null && obj.requestId !== "") {
w.uint32(10);
w.string(obj.requestId);
}
@ -255,7 +259,9 @@ export namespace FilterRPC {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
requestId: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -289,12 +295,12 @@ export namespace FilterRPC {
return _codec;
};
export const encode = (obj: Partial<FilterRPC>): Uint8Array => {
return encodeMessage(obj, FilterRPC.codec());
export const encode = (obj: Partial<FilterRpc>): Uint8Array => {
return encodeMessage(obj, FilterRpc.codec());
};
export const decode = (buf: Uint8Array | Uint8ArrayList): FilterRPC => {
return decodeMessage(buf, FilterRPC.codec());
export const decode = (buf: Uint8Array | Uint8ArrayList): FilterRpc => {
return decodeMessage(buf, FilterRpc.codec());
};
}
@ -420,10 +426,9 @@ export namespace RateLimitProof {
}
export interface WakuMessage {
payload?: Uint8Array;
contentTopic?: string;
payload: Uint8Array;
contentTopic: string;
version?: number;
timestampDeprecated?: number;
timestamp?: bigint;
rateLimitProof?: RateLimitProof;
ephemeral?: boolean;
@ -440,12 +445,12 @@ export namespace WakuMessage {
w.fork();
}
if (obj.payload != null) {
if (obj.payload != null && obj.payload.byteLength > 0) {
w.uint32(10);
w.bytes(obj.payload);
}
if (obj.contentTopic != null) {
if (obj.contentTopic != null && obj.contentTopic !== "") {
w.uint32(18);
w.string(obj.contentTopic);
}
@ -455,11 +460,6 @@ export namespace WakuMessage {
w.uint32(obj.version);
}
if (obj.timestampDeprecated != null) {
w.uint32(33);
w.double(obj.timestampDeprecated);
}
if (obj.timestamp != null) {
w.uint32(80);
w.sint64(obj.timestamp);
@ -480,7 +480,10 @@ export namespace WakuMessage {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
payload: new Uint8Array(0),
contentTopic: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -497,9 +500,6 @@ export namespace WakuMessage {
case 3:
obj.version = reader.uint32();
break;
case 4:
obj.timestampDeprecated = reader.double();
break;
case 10:
obj.timestamp = reader.sint64();
break;

13
packages/proto/src/lib/light_push.proto
View File

@ -1,19 +1,22 @@
// 19/WAKU2-LIGHTPUSH rfc: https://rfc.vac.dev/spec/19/
// Protocol identifier: /vac/waku/lightpush/2.0.0-beta1
syntax = "proto3";
import "message.proto";
message PushRequest {
optional string pub_sub_topic = 1;
optional WakuMessage message = 2;
string pubsub_topic = 1;
WakuMessage message = 2;
}
message PushResponse {
optional bool is_success = 1;
bool is_success = 1;
optional string info = 2;
}
message PushRPC {
optional string request_id = 1;
message PushRpc {
string request_id = 1;
optional PushRequest request = 2;
optional PushResponse response = 3;
}

68
packages/proto/src/lib/light_push.ts
View File

@ -9,7 +9,7 @@ import type { Codec } from "protons-runtime";
import type { Uint8ArrayList } from "uint8arraylist";
export interface PushRequest {
pubSubTopic?: string;
pubsubTopic: string;
message?: WakuMessage;
}
@ -24,9 +24,9 @@ export namespace PushRequest {
w.fork();
}
if (obj.pubSubTopic != null) {
if (obj.pubsubTopic != null && obj.pubsubTopic !== "") {
w.uint32(10);
w.string(obj.pubSubTopic);
w.string(obj.pubsubTopic);
}
if (obj.message != null) {
@ -39,7 +39,9 @@ export namespace PushRequest {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
pubsubTopic: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -48,7 +50,7 @@ export namespace PushRequest {
switch (tag >>> 3) {
case 1:
obj.pubSubTopic = reader.string();
obj.pubsubTopic = reader.string();
break;
case 2:
obj.message = WakuMessage.codec().decode(
@ -80,7 +82,7 @@ export namespace PushRequest {
}
export interface PushResponse {
isSuccess?: boolean;
isSuccess: boolean;
info?: string;
}
@ -95,7 +97,7 @@ export namespace PushResponse {
w.fork();
}
if (obj.isSuccess != null) {
if (obj.isSuccess != null && obj.isSuccess !== false) {
w.uint32(8);
w.bool(obj.isSuccess);
}
@ -110,7 +112,9 @@ export namespace PushResponse {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
isSuccess: false,
};
const end = length == null ? reader.len : reader.pos + length;
@ -147,24 +151,24 @@ export namespace PushResponse {
};
}
export interface PushRPC {
requestId?: string;
export interface PushRpc {
requestId: string;
request?: PushRequest;
response?: PushResponse;
}
export namespace PushRPC {
let _codec: Codec<PushRPC>;
export namespace PushRpc {
let _codec: Codec<PushRpc>;
export const codec = (): Codec<PushRPC> => {
export const codec = (): Codec<PushRpc> => {
if (_codec == null) {
_codec = message<PushRPC>(
_codec = message<PushRpc>(
(obj, w, opts = {}) => {
if (opts.lengthDelimited !== false) {
w.fork();
}
if (obj.requestId != null) {
if (obj.requestId != null && obj.requestId !== "") {
w.uint32(10);
w.string(obj.requestId);
}
@ -184,7 +188,9 @@ export namespace PushRPC {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
requestId: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -221,12 +227,12 @@ export namespace PushRPC {
return _codec;
};
export const encode = (obj: Partial<PushRPC>): Uint8Array => {
return encodeMessage(obj, PushRPC.codec());
export const encode = (obj: Partial<PushRpc>): Uint8Array => {
return encodeMessage(obj, PushRpc.codec());
};
export const decode = (buf: Uint8Array | Uint8ArrayList): PushRPC => {
return decodeMessage(buf, PushRPC.codec());
export const decode = (buf: Uint8Array | Uint8ArrayList): PushRpc => {
return decodeMessage(buf, PushRpc.codec());
};
}
@ -352,10 +358,9 @@ export namespace RateLimitProof {
}
export interface WakuMessage {
payload?: Uint8Array;
contentTopic?: string;
payload: Uint8Array;
contentTopic: string;
version?: number;
timestampDeprecated?: number;
timestamp?: bigint;
rateLimitProof?: RateLimitProof;
ephemeral?: boolean;
@ -372,12 +377,12 @@ export namespace WakuMessage {
w.fork();
}
if (obj.payload != null) {
if (obj.payload != null && obj.payload.byteLength > 0) {
w.uint32(10);
w.bytes(obj.payload);
}
if (obj.contentTopic != null) {
if (obj.contentTopic != null && obj.contentTopic !== "") {
w.uint32(18);
w.string(obj.contentTopic);
}
@ -387,11 +392,6 @@ export namespace WakuMessage {
w.uint32(obj.version);
}
if (obj.timestampDeprecated != null) {
w.uint32(33);
w.double(obj.timestampDeprecated);
}
if (obj.timestamp != null) {
w.uint32(80);
w.sint64(obj.timestamp);
@ -412,7 +412,10 @@ export namespace WakuMessage {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
payload: new Uint8Array(0),
contentTopic: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -429,9 +432,6 @@ export namespace WakuMessage {
case 3:
obj.version = reader.uint32();
break;
case 4:
obj.timestampDeprecated = reader.double();
break;
case 10:
obj.timestamp = reader.sint64();
break;

12
packages/proto/src/lib/message.proto
View File

@ -1,3 +1,5 @@
// 14/WAKU2-MESSAGE rfc: https://rfc.vac.dev/spec/14/
syntax = "proto3";
message RateLimitProof {
@ -7,16 +9,14 @@ message RateLimitProof {
bytes share_x = 4;
bytes share_y = 5;
bytes nullifier = 6;
bytes rlnIdentifier = 7;
bytes rln_identifier = 7;
}
message WakuMessage {
optional bytes payload = 1;
optional string content_topic = 2;
bytes payload = 1;
string content_topic = 2;
optional uint32 version = 3;
optional double timestamp_deprecated = 4;
optional sint64 timestamp = 10;
optional sint64 timestamp = 10;
optional RateLimitProof rate_limit_proof = 21;
optional bool ephemeral = 31;
}

22
packages/proto/src/lib/message.ts
View File

@ -130,10 +130,9 @@ export namespace RateLimitProof {
}
export interface WakuMessage {
payload?: Uint8Array;
contentTopic?: string;
payload: Uint8Array;
contentTopic: string;
version?: number;
timestampDeprecated?: number;
timestamp?: bigint;
rateLimitProof?: RateLimitProof;
ephemeral?: boolean;
@ -150,12 +149,12 @@ export namespace WakuMessage {
w.fork();
}
if (obj.payload != null) {
if (obj.payload != null && obj.payload.byteLength > 0) {
w.uint32(10);
w.bytes(obj.payload);
}
if (obj.contentTopic != null) {
if (obj.contentTopic != null && obj.contentTopic !== "") {
w.uint32(18);
w.string(obj.contentTopic);
}
@ -165,11 +164,6 @@ export namespace WakuMessage {
w.uint32(obj.version);
}
if (obj.timestampDeprecated != null) {
w.uint32(33);
w.double(obj.timestampDeprecated);
}
if (obj.timestamp != null) {
w.uint32(80);
w.sint64(obj.timestamp);
@ -190,7 +184,10 @@ export namespace WakuMessage {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
payload: new Uint8Array(0),
contentTopic: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -207,9 +204,6 @@ export namespace WakuMessage {
case 3:
obj.version = reader.uint32();
break;
case 4:
obj.timestampDeprecated = reader.double();
break;
case 10:
obj.timestamp = reader.sint64();
break;

31
packages/proto/src/lib/store.proto
View File

@ -1,30 +1,34 @@
// 13/WAKU2-STORE rfc: https://rfc.vac.dev/spec/13/
// Protocol identifier: /vac/waku/store/2.0.0-beta4
syntax = "proto3";
import "message.proto";
message Index {
optional bytes digest = 1;
optional sint64 received_time = 2;
optional sint64 sender_time = 3;
optional string pubsub_topic = 4;
bytes digest = 1;
sint64 receiver_time = 2;
sint64 sender_time = 3;
string pubsub_topic = 4;
}
message PagingInfo {
optional uint64 page_size = 1;
optional Index cursor = 2;
enum Direction {
DIRECTION_BACKWARD_UNSPECIFIED = 0;
DIRECTION_FORWARD = 1;
BACKWARD = 0;
FORWARD = 1;
}
optional Direction direction = 3;
}
message ContentFilter {
optional string content_topic = 1;
string content_topic = 1;
}
message HistoryQuery {
optional string pub_sub_topic = 2;
// The first field is reserved for future use
optional string pubsub_topic = 2;
repeated ContentFilter content_filters = 3;
optional PagingInfo paging_info = 4;
optional sint64 start_time = 5;
@ -32,17 +36,18 @@ message HistoryQuery {
}
message HistoryResponse {
// The first field is reserved for future use
repeated WakuMessage messages = 2;
optional PagingInfo paging_info = 3;
enum HistoryError {
ERROR_NONE_UNSPECIFIED = 0;
ERROR_INVALID_CURSOR = 1;
NONE = 0;
INVALID_CURSOR = 1;
}
optional HistoryError error = 4;
HistoryError error = 4;
}
message HistoryRPC {
optional string request_id = 1;
message HistoryRpc {
string request_id = 1;
optional HistoryQuery query = 2;
optional HistoryResponse response = 3;
}

112
packages/proto/src/lib/store.ts
View File

@ -14,10 +14,10 @@ import type { Codec } from "protons-runtime";
import type { Uint8ArrayList } from "uint8arraylist";
export interface Index {
digest?: Uint8Array;
receivedTime?: bigint;
senderTime?: bigint;
pubsubTopic?: string;
digest: Uint8Array;
receiverTime: bigint;
senderTime: bigint;
pubsubTopic: string;
}
export namespace Index {
@ -31,22 +31,22 @@ export namespace Index {
w.fork();
}
if (obj.digest != null) {
if (obj.digest != null && obj.digest.byteLength > 0) {
w.uint32(10);
w.bytes(obj.digest);
}
if (obj.receivedTime != null) {
if (obj.receiverTime != null && obj.receiverTime !== 0n) {
w.uint32(16);
w.sint64(obj.receivedTime);
w.sint64(obj.receiverTime);
}
if (obj.senderTime != null) {
if (obj.senderTime != null && obj.senderTime !== 0n) {
w.uint32(24);
w.sint64(obj.senderTime);
}
if (obj.pubsubTopic != null) {
if (obj.pubsubTopic != null && obj.pubsubTopic !== "") {
w.uint32(34);
w.string(obj.pubsubTopic);
}
@ -56,7 +56,12 @@ export namespace Index {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
digest: new Uint8Array(0),
receiverTime: 0n,
senderTime: 0n,
pubsubTopic: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -68,7 +73,7 @@ export namespace Index {
obj.digest = reader.bytes();
break;
case 2:
obj.receivedTime = reader.sint64();
obj.receiverTime = reader.sint64();
break;
case 3:
obj.senderTime = reader.sint64();
@ -107,13 +112,13 @@ export interface PagingInfo {
export namespace PagingInfo {
export enum Direction {
DIRECTION_BACKWARD_UNSPECIFIED = "DIRECTION_BACKWARD_UNSPECIFIED",
DIRECTION_FORWARD = "DIRECTION_FORWARD",
BACKWARD = "BACKWARD",
FORWARD = "FORWARD",
}
enum __DirectionValues {
DIRECTION_BACKWARD_UNSPECIFIED = 0,
DIRECTION_FORWARD = 1,
BACKWARD = 0,
FORWARD = 1,
}
export namespace Direction {
@ -193,7 +198,7 @@ export namespace PagingInfo {
}
export interface ContentFilter {
contentTopic?: string;
contentTopic: string;
}
export namespace ContentFilter {
@ -207,7 +212,7 @@ export namespace ContentFilter {
w.fork();
}
if (obj.contentTopic != null) {
if (obj.contentTopic != null && obj.contentTopic !== "") {
w.uint32(10);
w.string(obj.contentTopic);
}
@ -217,7 +222,9 @@ export namespace ContentFilter {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
contentTopic: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -252,7 +259,7 @@ export namespace ContentFilter {
}
export interface HistoryQuery {
pubSubTopic?: string;
pubsubTopic?: string;
contentFilters: ContentFilter[];
pagingInfo?: PagingInfo;
startTime?: bigint;
@ -270,9 +277,9 @@ export namespace HistoryQuery {
w.fork();
}
if (obj.pubSubTopic != null) {
if (obj.pubsubTopic != null) {
w.uint32(18);
w.string(obj.pubSubTopic);
w.string(obj.pubsubTopic);
}
if (obj.contentFilters != null) {
@ -313,7 +320,7 @@ export namespace HistoryQuery {
switch (tag >>> 3) {
case 2:
obj.pubSubTopic = reader.string();
obj.pubsubTopic = reader.string();
break;
case 3:
obj.contentFilters.push(
@ -358,18 +365,18 @@ export namespace HistoryQuery {
export interface HistoryResponse {
messages: WakuMessage[];
pagingInfo?: PagingInfo;
error?: HistoryResponse.HistoryError;
error: HistoryResponse.HistoryError;
}
export namespace HistoryResponse {
export enum HistoryError {
ERROR_NONE_UNSPECIFIED = "ERROR_NONE_UNSPECIFIED",
ERROR_INVALID_CURSOR = "ERROR_INVALID_CURSOR",
NONE = "NONE",
INVALID_CURSOR = "INVALID_CURSOR",
}
enum __HistoryErrorValues {
ERROR_NONE_UNSPECIFIED = 0,
ERROR_INVALID_CURSOR = 1,
NONE = 0,
INVALID_CURSOR = 1,
}
export namespace HistoryError {
@ -400,7 +407,7 @@ export namespace HistoryResponse {
PagingInfo.codec().encode(obj.pagingInfo, w);
}
if (obj.error != null) {
if (obj.error != null && __HistoryErrorValues[obj.error] !== 0) {
w.uint32(32);
HistoryResponse.HistoryError.codec().encode(obj.error, w);
}
@ -412,6 +419,7 @@ export namespace HistoryResponse {
(reader, length) => {
const obj: any = {
messages: [],
error: HistoryError.NONE,
};
const end = length == null ? reader.len : reader.pos + length;
@ -457,24 +465,24 @@ export namespace HistoryResponse {
};
}
export interface HistoryRPC {
requestId?: string;
export interface HistoryRpc {
requestId: string;
query?: HistoryQuery;
response?: HistoryResponse;
}
export namespace HistoryRPC {
let _codec: Codec<HistoryRPC>;
export namespace HistoryRpc {
let _codec: Codec<HistoryRpc>;
export const codec = (): Codec<HistoryRPC> => {
export const codec = (): Codec<HistoryRpc> => {
if (_codec == null) {
_codec = message<HistoryRPC>(
_codec = message<HistoryRpc>(
(obj, w, opts = {}) => {
if (opts.lengthDelimited !== false) {
w.fork();
}
if (obj.requestId != null) {
if (obj.requestId != null && obj.requestId !== "") {
w.uint32(10);
w.string(obj.requestId);
}
@ -494,7 +502,9 @@ export namespace HistoryRPC {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
requestId: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -531,12 +541,12 @@ export namespace HistoryRPC {
return _codec;
};
export const encode = (obj: Partial<HistoryRPC>): Uint8Array => {
return encodeMessage(obj, HistoryRPC.codec());
export const encode = (obj: Partial<HistoryRpc>): Uint8Array => {
return encodeMessage(obj, HistoryRpc.codec());
};
export const decode = (buf: Uint8Array | Uint8ArrayList): HistoryRPC => {
return decodeMessage(buf, HistoryRPC.codec());
export const decode = (buf: Uint8Array | Uint8ArrayList): HistoryRpc => {
return decodeMessage(buf, HistoryRpc.codec());
};
}
@ -662,10 +672,9 @@ export namespace RateLimitProof {
}
export interface WakuMessage {
payload?: Uint8Array;
contentTopic?: string;
payload: Uint8Array;
contentTopic: string;
version?: number;
timestampDeprecated?: number;
timestamp?: bigint;
rateLimitProof?: RateLimitProof;
ephemeral?: boolean;
@ -682,12 +691,12 @@ export namespace WakuMessage {
w.fork();
}
if (obj.payload != null) {
if (obj.payload != null && obj.payload.byteLength > 0) {
w.uint32(10);
w.bytes(obj.payload);
}
if (obj.contentTopic != null) {
if (obj.contentTopic != null && obj.contentTopic !== "") {
w.uint32(18);
w.string(obj.contentTopic);
}
@ -697,11 +706,6 @@ export namespace WakuMessage {
w.uint32(obj.version);
}
if (obj.timestampDeprecated != null) {
w.uint32(33);
w.double(obj.timestampDeprecated);
}
if (obj.timestamp != null) {
w.uint32(80);
w.sint64(obj.timestamp);
@ -722,7 +726,10 @@ export namespace WakuMessage {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
payload: new Uint8Array(0),
contentTopic: "",
};
const end = length == null ? reader.len : reader.pos + length;
@ -739,9 +746,6 @@ export namespace WakuMessage {
case 3:
obj.version = reader.uint32();
break;
case 4:
obj.timestampDeprecated = reader.double();
break;
case 10:
obj.timestamp = reader.sint64();
break;

2
packages/proto/src/lib/topic_only_message.proto
View File

@ -1,5 +1,5 @@
syntax = "proto3";
message TopicOnlyMessage {
optional string content_topic = 2;
string content_topic = 2;
}

8
packages/proto/src/lib/topic_only_message.ts
View File

@ -9,7 +9,7 @@ import type { Codec } from "protons-runtime";
import type { Uint8ArrayList } from "uint8arraylist";
export interface TopicOnlyMessage {
contentTopic?: string;
contentTopic: string;
}
export namespace TopicOnlyMessage {
@ -23,7 +23,7 @@ export namespace TopicOnlyMessage {
w.fork();
}
if (obj.contentTopic != null) {
if (obj.contentTopic != null && obj.contentTopic !== "") {
w.uint32(18);
w.string(obj.contentTopic);
}
@ -33,7 +33,9 @@ export namespace TopicOnlyMessage {
}
},
(reader, length) => {
const obj: any = {};
const obj: any = {
contentTopic: "",
};
const end = length == null ? reader.len : reader.pos + length;

3
packages/tests/package.json
View File

@ -33,7 +33,8 @@
"privacy"
],
"scripts": {
"build": "tsc",
"build": "run-s build:**",
"build:esm": "tsc",
"fix": "run-s fix:*",
"fix:prettier": "prettier . --write",
"fix:lint": "eslint src tests --ext .ts --ext .cjs --fix",

12
packages/tests/tests/ephemeral.node.spec.ts
View File

@ -192,10 +192,10 @@ describe("Waku Message Ephemeral field", () => {
}
const normalMsg = messages.find(
(msg) => bytesToUtf8(msg.payload!) === normalTxt
(msg) => bytesToUtf8(msg.payload) === normalTxt
);
const ephemeralMsg = messages.find(
(msg) => bytesToUtf8(msg.payload!) === ephemeralTxt
(msg) => bytesToUtf8(msg.payload) === ephemeralTxt
);
expect(normalMsg).to.not.be.undefined;
@ -241,10 +241,10 @@ describe("Waku Message Ephemeral field", () => {
}
const normalMsg = messages.find(
(msg) => bytesToUtf8(msg.payload!) === normalTxt
(msg) => bytesToUtf8(msg.payload) === normalTxt
);
const ephemeralMsg = messages.find(
(msg) => bytesToUtf8(msg.payload!) === ephemeralTxt
(msg) => bytesToUtf8(msg.payload) === ephemeralTxt
);
expect(normalMsg).to.not.be.undefined;
@ -291,10 +291,10 @@ describe("Waku Message Ephemeral field", () => {
}
const normalMsg = messages.find(
(msg) => bytesToUtf8(msg.payload!) === normalTxt
(msg) => bytesToUtf8(msg.payload) === normalTxt
);
const ephemeralMsg = messages.find(
(msg) => bytesToUtf8(msg.payload!) === ephemeralTxt
(msg) => bytesToUtf8(msg.payload) === ephemeralTxt
);
expect(normalMsg).to.not.be.undefined;

2
packages/tests/tests/filter.node.spec.ts
View File

@ -52,7 +52,7 @@ describe("Waku Filter", () => {
log("Got a message");
messageCount++;
expect(msg.contentTopic).to.eq(TestContentTopic);
expect(bytesToUtf8(msg.payload!)).to.eq(messageText);
expect(bytesToUtf8(msg.payload)).to.eq(messageText);
};
await waku.filter.subscribe([TestDecoder], callback);

2
packages/tests/tests/light_push.node.spec.ts
View File

@ -106,6 +106,6 @@ describe("Waku Light Push [node only]", () => {
}
expect(msgs[0].contentTopic).to.equal(TestContentTopic);
expect(base64ToUtf8(msgs[0].payload!)).to.equal(messageText);
expect(base64ToUtf8(msgs[0].payload)).to.equal(messageText);
});
});

6
packages/tests/tests/relay.node.spec.ts
View File

@ -130,7 +130,7 @@ describe("Waku Relay [node only]", () => {
const receivedMsg = await receivedMsgPromise;
expect(receivedMsg.contentTopic).to.eq(TestContentTopic);
expect(bytesToUtf8(receivedMsg.payload!)).to.eq(messageText);
expect(bytesToUtf8(receivedMsg.payload)).to.eq(messageText);
expect(receivedMsg.timestamp?.valueOf()).to.eq(
messageTimestamp.valueOf()
);
@ -173,10 +173,10 @@ describe("Waku Relay [node only]", () => {
}
expect(fooMessages[0].contentTopic).to.eq(fooContentTopic);
expect(bytesToUtf8(fooMessages[0].payload!)).to.eq(fooMessageText);
expect(bytesToUtf8(fooMessages[0].payload)).to.eq(fooMessageText);
expect(barMessages[0].contentTopic).to.eq(barContentTopic);
expect(bytesToUtf8(barMessages[0].payload!)).to.eq(barMessageText);
expect(bytesToUtf8(barMessages[0].payload)).to.eq(barMessageText);
expect(fooMessages.length).to.eq(1);
expect(barMessages.length).to.eq(1);

10
packages/tests/tests/store.node.spec.ts
View File

@ -90,7 +90,7 @@ describe("Waku Store", () => {
expect(messages?.length).eq(totalMsgs);
const result = messages?.findIndex((msg) => {
return msg.payload![0]! === 0;
return msg.payload[0]! === 0;
});
expect(result).to.not.eq(-1);
});
@ -173,8 +173,8 @@ describe("Waku Store", () => {
expect(messages.length).be.eq(totalMsgs);
expect(bytesToUtf8(testMessage.payload!)).to.be.eq(
bytesToUtf8(messages[cursorIndex + 1].payload!)
expect(bytesToUtf8(testMessage.payload)).to.be.eq(
bytesToUtf8(messages[cursorIndex + 1].payload)
);
});
@ -214,7 +214,7 @@ describe("Waku Store", () => {
expect(messages?.length).eq(totalMsgs);
const result = messages?.findIndex((msg) => {
return msg.payload![0]! === 0;
return msg.payload[0]! === 0;
});
expect(result).to.not.eq(-1);
});
@ -294,7 +294,7 @@ describe("Waku Store", () => {
);
expect(messages?.length).eq(totalMsgs);
const payloads = messages.map((msg) => msg.payload![0]!);
const payloads = messages.map((msg) => msg.payload[0]!);
expect(payloads).to.deep.eq(Array.from(Array(totalMsgs).keys()));
});

2
packages/tests/tests/waku.node.spec.ts
View File

@ -187,7 +187,7 @@ describe("Decryption Keys", () => {
const receivedMsg = await receivedMsgPromise;
expect(receivedMsg.contentTopic).to.eq(TestContentTopic);
expect(bytesToUtf8(receivedMsg.payload!)).to.eq(messageText);
expect(bytesToUtf8(receivedMsg.payload)).to.eq(messageText);
expect(receivedMsg.timestamp?.valueOf()).to.eq(messageTimestamp.valueOf());
});
});