EventsSubscribe

[LiranCohen]

This PR introduces an EventStream interface and an implementation based on EventEmitter which emits events for any interfaces which are saved as a part of the DWN's message store. (ie. not Query/Read/Subscribe messages).

We also introduce an EventsSubscribe interface that follows the same authorization model as EventsGet and EventsQuery, which only allows access to the tenant owner.

In a subsequent PR we will introduce RecordsSubscribe, as well as some additional enhancements.

[codesandbox]

Review or Edit in CodeSandbox

Open the branch in Web Editor • VS Code • Insiders

Open Preview

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (30ba7d5) 98.84% compared to head (19101e7) 98.86%.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #658      +/-   ##
==========================================
+ Coverage   98.84%   98.86%   +0.02%     
==========================================
  Files          71       75       +4     
  Lines        9313     9656     +343     
  Branches     1393     1422      +29     
==========================================
+ Hits         9205     9546     +341     
- Misses        100      102       +2     
  Partials        8        8              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[thehenrytsai]

After reading the code a bit more, maybe I am overlooking something, but it seems that the design of EventStream can be simplified, specifically around the subscribe() method:

subscribe(id: string, listener: EventListener) or if absolutely necessary:
subscribe(id: string, subscriptionHandler: SubscriptionHandler)

might get the job done.

I think ideally EventStream should not know (or care) about who is interested in its events and what they do with them. This way the EventsSubscribeHandler and RecordsSubscribeHandler can construct the event handler however they see fit, and just tell EventStream: "Hey, I am interested in your events, send them to me".

Let me know what you think, happy to jump in a call tomorrow!

[LiranCohen]

I think ideally EventStream should not know (or care) about who is interested in its events and what they do with them. This way the EventsSubscribeHandler and RecordsSubscribeHandler can construct the event handler however they see fit, and just tell EventStream: "Hey, I am interested in your events, send them to me".

I think I understand what you're getting at and it kind of makes sense. Admittedly I designed EventStream first and the handler portions as an after-thought when thinking about more specific cases.

Would definitely like to jump on a call and chat just to give some additional context and make sure we are on the same page.

[LiranCohen]

@thehenrytsai I've got much of the refactor in place and have been thinking about what we spoke about wrt the handler function.

Would like your input on some things before I go with them as they're touching many tests.

In order to support the change to treat handler: GenericMessageHandler similarly to how dataStream: Readable is treated I propose we change the processMessage signature as follows:

Currently:

  public async processMessage(tenant: string, rawMessage: GenericMessage, dataStream?: Readable): Promise<UnionMessageReply>;

Proposed:

export type MessageOptions = {
  dataStream?: Readable;
  handler?: GenericMessageHandler
};

public async processMessage(tenant: string, rawMessage: GenericMessage, options?: MessageOptions): Promise<UnionMessageReply>;

If you like this approach, I think it would be worthwhile to update that signature in it's own PR first, with only dataStream being a part of the MessageOptions, as this method is touching nearly every single test file we have.

[LiranCohen]

@thehenrytsai This is ready for review again. I still need to write the scenario tests for all the specific filters, but I wanted to get the review going for the general refactor.

I do think that in a later PR we will introduce the unsubscribe functionality we spoke about vs the .close() method, but I left it like that for now to make the reviews easier.

The logic is simplified a lot further than it was in many places. When I add RecordsSubscribe some things will change and bring in more complexity but we can review those more independently.

[thehenrytsai]

🐐 🔥 🥇 🚀