update interface

src/core/message-reply.ts

@@ -1,5 +1,5 @@
-import type { QueryResultEntry } from '../types/message-types.js';
 import type { Readable } from 'readable-stream';
+import type { GenericMessageSubscription, QueryResultEntry } from '../types/message-types.js';
 
 type Status = {
   code: number
@@ -39,4 +39,9 @@ export type UnionMessageReply = GenericMessageReply & {
    * Mutually exclusive with `data`.
    */
   cursor?: string;
+
+  /**
+   * A subscription object if a subscription was requested.
+   */
+  subscription?: GenericMessageSubscription;
 };

src/dwn.ts

@@ -18,6 +18,7 @@ import { DidResolver } from './did/did-resolver.js';
 import { EventsGetHandler } from './handlers/events-get.js';
 import { EventsQueryHandler } from './handlers/events-query.js';
 import { EventsSubscribeHandler } from './handlers/events-subscribe.js';
+import { EventStreamEmitter } from './event-log/event-stream.js';
 import { Message } from './core/message.js';
 import { messageReplyFromError } from './core/message-reply.js';
 import { MessagesGetHandler } from './handlers/messages-get.js';
@@ -45,10 +46,10 @@ export class Dwn {
   private constructor(config: DwnConfig) {
     this.didResolver = config.didResolver!;
     this.tenantGate = config.tenantGate!;
+    this.eventStream = config.eventStream!;
     this.messageStore = config.messageStore;
     this.dataStore = config.dataStore;
     this.eventLog = config.eventLog;
-    this.eventStream = config.eventStream;
 
     this.methodHandlers = {
       [DwnInterfaceName.Events + DwnMethodName.Get]: new EventsGetHandler(
@@ -136,6 +137,7 @@ export class Dwn {
   public static async create(config: DwnConfig): Promise<Dwn> {
     config.didResolver ??= new DidResolver();
     config.tenantGate ??= new AllowAllTenantGate();
+    config.eventStream ??= new EventStreamEmitter({ messageStore: config.messageStore, didResolver: config.didResolver });
 
     const dwn = new Dwn(config);
     await dwn.open();
@@ -242,10 +244,10 @@ export class Dwn {
  */
 export type DwnConfig = {
   didResolver?: DidResolver;
+  eventStream?: EventStream;
   tenantGate?: TenantGate;
 
   messageStore: MessageStore;
   dataStore: DataStore;
   eventLog: EventLog;
-  eventStream: EventStream;
 };

src/event-log/event-stream.ts

@@ -2,7 +2,7 @@ import type { DidResolver } from '../did/did-resolver.js';
 import type { GenericMessage } from '../types/message-types.js';
 import type { MessageStore } from '../types/message-store.js';
 import type { EventsSubscribeMessage, EventSubscription } from '../types/event-types.js';
-import type { EventStream, Subscription } from '../types/event-stream.js';
+import type { EventStream, EventStreamSubscription } from '../types/event-stream.js';
 import type { Filter, KeyValues } from '../types/query-types.js';
 import type { RecordsSubscribeMessage, RecordsSubscription } from '../types/records-types.js';
 
@@ -30,7 +30,7 @@ export class EventStreamEmitter implements EventStream {
   private reauthorizationTTL: number;
 
   private isOpen: boolean = false;
-  private subscriptions: Map<string, Subscription> = new Map();
+  private subscriptions: Map<string, EventStreamSubscription> = new Map();
 
   constructor(config: EventStreamConfig) {
     this.didResolver = config.didResolver;

src/index.ts

@@ -10,7 +10,7 @@ export type { MessagesGetMessage, MessagesGetReply } from './types/messages-type
 export type { PermissionConditions, PermissionScope, PermissionsGrantDescriptor } from './types/permissions-grant-descriptor.js';
 export type { PermissionsGrantMessage, PermissionsRequestDescriptor, PermissionsRequestMessage, PermissionsRevokeDescriptor, PermissionsRevokeMessage } from './types/permissions-types.js';
 export type { ProtocolsConfigureDescriptor, ProtocolDefinition, ProtocolTypes, ProtocolRuleSet, ProtocolsQueryFilter, ProtocolsConfigureMessage, ProtocolsQueryMessage, ProtocolsQueryReply } from './types/protocols-types.js';
-export type { EncryptionProperty, RecordsDeleteMessage, RecordsQueryMessage, RecordsQueryReply, RecordsQueryReplyEntry, RecordsReadReply, RecordsSubscribeDescriptor, RecordsSubscribeMessage, RecordsWriteDescriptor, RecordsWriteMessage } from './types/records-types.js';
+export type { EncryptionProperty, RecordsDeleteMessage, RecordsQueryMessage, RecordsQueryReply, RecordsQueryReplyEntry, RecordsReadReply, RecordsSubscribeDescriptor, RecordsSubscribeMessage, RecordsSubscription, RecordsWriteDescriptor, RecordsWriteMessage } from './types/records-types.js';
 export { authenticate } from './core/auth.js';
 export { AllowAllTenantGate, TenantGate } from './core/tenant-gate.js';
 export { Cid } from './utils/cid.js';

src/types/event-stream.ts

@@ -13,7 +13,7 @@ export interface EventStream {
   close(): Promise<void>;
 }
 
-export interface Subscription {
+export interface EventStreamSubscription {
   id: string;
   listener: (tenant: string, message: GenericMessage, ...indexes: KeyValues[]) => void;
   on: (handler: EventHandler) => { off: () => void };

src/types/message-types.ts

@@ -67,6 +67,14 @@ export type QueryResultEntry = GenericMessage & {
   encodedData?: string;
 };
 
+export type GenericMessageHandler = (message: GenericMessage) => void;
+
+export type GenericMessageSubscription = {
+  id: string;
+  on: (handler: GenericMessageHandler) => { off: () => void };
+  close: () => Promise<void>;
+};
+
 /**
  * Pagination Options for querying messages.
  *