diff --git a/src/state/messages/events/agent.ts b/src/state/messages/events/agent.ts index 422672853f..fdd06bb5d2 100644 --- a/src/state/messages/events/agent.ts +++ b/src/state/messages/events/agent.ts @@ -22,7 +22,11 @@ export class MessagesEventBus { private agent: BskyAgent private __tempFromUserDid: string - private emitter = new EventEmitter() + private emitter = new EventEmitter<{ + events: [ChatBskyConvoGetLog.OutputSchema['logs']] + connect: undefined + error: [MessagesEventBusError] | undefined + }>() private status: MessagesEventBusStatus = MessagesEventBusStatus.Uninitialized private error: MessagesEventBusError | undefined @@ -45,6 +49,8 @@ export class MessagesEventBus { this.requestPollInterval = this.requestPollInterval.bind(this) this.trail = this.trail.bind(this) this.trailConvo = this.trailConvo.bind(this) + this.onConnect = this.onConnect.bind(this) + this.onError = this.onError.bind(this) } private commit() { @@ -81,6 +87,8 @@ export class MessagesEventBus { requestPollInterval: this.requestPollInterval, trail: this.trail, trailConvo: this.trailConvo, + onConnect: this.onConnect, + onError: this.onError, } } case MessagesEventBusStatus.Ready: { @@ -91,6 +99,8 @@ export class MessagesEventBus { requestPollInterval: this.requestPollInterval, trail: this.trail, trailConvo: this.trailConvo, + onConnect: this.onConnect, + onError: this.onError, } } case MessagesEventBusStatus.Suspended: { @@ -101,6 +111,8 @@ export class MessagesEventBus { requestPollInterval: this.requestPollInterval, trail: this.trail, trailConvo: this.trailConvo, + onConnect: this.onConnect, + onError: this.onError, } } case MessagesEventBusStatus.Error: { @@ -116,6 +128,8 @@ export class MessagesEventBus { requestPollInterval: this.requestPollInterval, trail: this.trail, trailConvo: this.trailConvo, + onConnect: this.onConnect, + onError: this.onError, } } default: { @@ -126,6 +140,8 @@ export class MessagesEventBus { requestPollInterval: this.requestPollInterval, trail: this.trail, trailConvo: this.trailConvo, + onConnect: this.onConnect, + onError: this.onError, } } } @@ -150,11 +166,13 @@ export class MessagesEventBus { case MessagesEventBusDispatchEvent.Ready: { this.status = MessagesEventBusStatus.Ready this.resetPoll() + this.emitter.emit('connect') break } case MessagesEventBusDispatchEvent.Background: { this.status = MessagesEventBusStatus.Backgrounded this.resetPoll() + this.emitter.emit('connect') break } case MessagesEventBusDispatchEvent.Suspend: { @@ -164,6 +182,7 @@ export class MessagesEventBus { case MessagesEventBusDispatchEvent.Error: { this.status = MessagesEventBusStatus.Error this.error = action.payload + this.emitter.emit('error', action.payload) break } } @@ -185,6 +204,7 @@ export class MessagesEventBus { this.status = MessagesEventBusStatus.Error this.error = action.payload this.stopPoll() + this.emitter.emit('error', action.payload) break } } @@ -206,6 +226,7 @@ export class MessagesEventBus { this.status = MessagesEventBusStatus.Error this.error = action.payload this.stopPoll() + this.emitter.emit('error', action.payload) break } } @@ -227,6 +248,7 @@ export class MessagesEventBus { this.status = MessagesEventBusStatus.Error this.error = action.payload this.stopPoll() + this.emitter.emit('error', action.payload) break } } @@ -240,6 +262,7 @@ export class MessagesEventBus { this.error = undefined this.latestRev = undefined this.setup() + this.emitter.emit('connect') break } } @@ -346,6 +369,34 @@ export class MessagesEventBus { } } + onConnect(handler: () => void) { + this.emitter.on('connect', handler) + + if ( + this.status === MessagesEventBusStatus.Ready || + this.status === MessagesEventBusStatus.Backgrounded || + this.status === MessagesEventBusStatus.Suspended + ) { + handler() + } + + return () => { + this.emitter.off('connect', handler) + } + } + + onError(handler: (payload?: MessagesEventBusError) => void) { + this.emitter.on('error', handler) + + if (this.status === MessagesEventBusStatus.Error) { + handler(this.error) + } + + return () => { + this.emitter.off('error', handler) + } + } + private async initializeLatestRev() { logger.debug( `${LOGGER_CONTEXT}: initialize latest rev`, diff --git a/src/state/messages/events/types.ts b/src/state/messages/events/types.ts index 6959b4f069..0ec5a2a12a 100644 --- a/src/state/messages/events/types.ts +++ b/src/state/messages/events/types.ts @@ -61,6 +61,10 @@ export type TrailHandler = ( ) => void export type RequestPollIntervalHandler = (interval: number) => () => void +export type OnConnectHandler = (handler: () => void) => () => void +export type OnDisconnectHandler = ( + handler: (error?: MessagesEventBusError) => void, +) => () => void export type MessagesEventBusState = | { @@ -70,6 +74,8 @@ export type MessagesEventBusState = requestPollInterval: RequestPollIntervalHandler trail: (handler: TrailHandler) => () => void trailConvo: (convoId: string, handler: TrailHandler) => () => void + onConnect: OnConnectHandler + onError: OnDisconnectHandler } | { status: MessagesEventBusStatus.Initializing @@ -78,6 +84,8 @@ export type MessagesEventBusState = requestPollInterval: RequestPollIntervalHandler trail: (handler: TrailHandler) => () => void trailConvo: (convoId: string, handler: TrailHandler) => () => void + onConnect: OnConnectHandler + onError: OnDisconnectHandler } | { status: MessagesEventBusStatus.Ready @@ -86,6 +94,8 @@ export type MessagesEventBusState = requestPollInterval: RequestPollIntervalHandler trail: (handler: TrailHandler) => () => void trailConvo: (convoId: string, handler: TrailHandler) => () => void + onConnect: OnConnectHandler + onError: OnDisconnectHandler } | { status: MessagesEventBusStatus.Backgrounded @@ -94,6 +104,8 @@ export type MessagesEventBusState = requestPollInterval: RequestPollIntervalHandler trail: (handler: TrailHandler) => () => void trailConvo: (convoId: string, handler: TrailHandler) => () => void + onConnect: OnConnectHandler + onError: OnDisconnectHandler } | { status: MessagesEventBusStatus.Suspended @@ -102,6 +114,8 @@ export type MessagesEventBusState = requestPollInterval: RequestPollIntervalHandler trail: (handler: TrailHandler) => () => void trailConvo: (convoId: string, handler: TrailHandler) => () => void + onConnect: OnConnectHandler + onError: OnDisconnectHandler } | { status: MessagesEventBusStatus.Error @@ -110,4 +124,6 @@ export type MessagesEventBusState = requestPollInterval: RequestPollIntervalHandler trail: (handler: TrailHandler) => () => void trailConvo: (convoId: string, handler: TrailHandler) => () => void + onConnect: OnConnectHandler + onError: OnDisconnectHandler }