Add global event bus for messages logs

This commit is contained in:
Eric Bailey
2024-05-07 14:18:37 -05:00
parent c6d30f671a
commit 38262e6ab0
5 changed files with 608 additions and 41 deletions
+22 -19
View File
@@ -16,6 +16,7 @@ import {useQueryClient} from '@tanstack/react-query'
import {Provider as StatsigProvider} from '#/lib/statsig/statsig'
import {logger} from '#/logger'
import {MessagesEventBusProvider} from '#/state/messages/events'
import {init as initPersistedState} from '#/state/persisted'
import {Provider as LabelDefsProvider} from '#/state/preferences/label-defs'
import {Provider as ModerationOptsProvider} from '#/state/preferences/moderation-opts'
@@ -95,25 +96,27 @@ function InnerApp() {
// Resets the entire tree below when it changes:
key={currentAccount?.did}>
<QueryProvider currentDid={currentAccount?.did}>
<PushNotificationsListener>
<StatsigProvider>
{/* LabelDefsProvider MUST come before ModerationOptsProvider */}
<LabelDefsProvider>
<ModerationOptsProvider>
<LoggedOutViewProvider>
<SelectedFeedProvider>
<UnreadNotifsProvider>
<GestureHandlerRootView style={s.h100pct}>
<TestCtrls />
<Shell />
</GestureHandlerRootView>
</UnreadNotifsProvider>
</SelectedFeedProvider>
</LoggedOutViewProvider>
</ModerationOptsProvider>
</LabelDefsProvider>
</StatsigProvider>
</PushNotificationsListener>
<MessagesEventBusProvider>
<PushNotificationsListener>
<StatsigProvider>
{/* LabelDefsProvider MUST come before ModerationOptsProvider */}
<LabelDefsProvider>
<ModerationOptsProvider>
<LoggedOutViewProvider>
<SelectedFeedProvider>
<UnreadNotifsProvider>
<GestureHandlerRootView style={s.h100pct}>
<TestCtrls />
<Shell />
</GestureHandlerRootView>
</UnreadNotifsProvider>
</SelectedFeedProvider>
</LoggedOutViewProvider>
</ModerationOptsProvider>
</LabelDefsProvider>
</StatsigProvider>
</PushNotificationsListener>
</MessagesEventBusProvider>
</QueryProvider>
</React.Fragment>
</RootSiblingParent>
+20 -22
View File
@@ -9,6 +9,7 @@ import {useLingui} from '@lingui/react'
import {Provider as StatsigProvider} from '#/lib/statsig/statsig'
import {logger} from '#/logger'
import {MessagesEventBusProvider} from '#/state/messages/events'
import {init as initPersistedState} from '#/state/persisted'
import {Provider as LabelDefsProvider} from '#/state/preferences/label-defs'
import {Provider as ModerationOptsProvider} from '#/state/preferences/moderation-opts'
@@ -83,22 +84,24 @@ function InnerApp() {
// Resets the entire tree below when it changes:
key={currentAccount?.did}>
<QueryProvider currentDid={currentAccount?.did}>
<StatsigProvider>
{/* LabelDefsProvider MUST come before ModerationOptsProvider */}
<LabelDefsProvider>
<ModerationOptsProvider>
<LoggedOutViewProvider>
<SelectedFeedProvider>
<UnreadNotifsProvider>
<SafeAreaProvider>
<Shell />
</SafeAreaProvider>
</UnreadNotifsProvider>
</SelectedFeedProvider>
</LoggedOutViewProvider>
</ModerationOptsProvider>
</LabelDefsProvider>
</StatsigProvider>
<MessagesEventBusProvider>
<StatsigProvider>
{/* LabelDefsProvider MUST come before ModerationOptsProvider */}
<LabelDefsProvider>
<ModerationOptsProvider>
<LoggedOutViewProvider>
<SelectedFeedProvider>
<UnreadNotifsProvider>
<SafeAreaProvider>
<Shell />
</SafeAreaProvider>
</UnreadNotifsProvider>
</SelectedFeedProvider>
</LoggedOutViewProvider>
</ModerationOptsProvider>
</LabelDefsProvider>
</StatsigProvider>
</MessagesEventBusProvider>
</QueryProvider>
</React.Fragment>
<ToastContainer />
@@ -112,12 +115,7 @@ function App() {
const [isReady, setReady] = useState(false)
React.useEffect(() => {
initPersistedState().then(() => {
setReady(true)
const preloadElement = document.getElementById('preload')
preloadElement?.remove()
})
initPersistedState().then(() => setReady(true))
}, [])
if (!isReady) {
+412
View File
@@ -0,0 +1,412 @@
import {BskyAgent, ChatBskyConvoGetLog} from '@atproto-labs/api'
import EventEmitter from 'eventemitter3'
import {nanoid} from 'nanoid/non-secure'
import {logger} from '#/logger'
import {
MessagesEventBusDispatch,
MessagesEventBusDispatchEvent,
MessagesEventBusError,
MessagesEventBusErrorCode,
MessagesEventBusParams,
MessagesEventBusState,
MessagesEventBusStatus,
} from '#/state/messages/events/types'
const LOGGER_CONTEXT = 'MessagesEventBus'
const ACTIVE_POLL_INTERVAL = 3e3
export class MessagesEventBus {
private id: string
private agent: BskyAgent
private __tempFromUserDid: string
private emitter = new EventEmitter()
private status: MessagesEventBusStatus = MessagesEventBusStatus.Uninitialized
private pollInterval = ACTIVE_POLL_INTERVAL
private error: MessagesEventBusError | undefined
private latestRev: string | undefined = undefined
private nextPoll: NodeJS.Timeout | undefined
snapshot: MessagesEventBusState | undefined
constructor(params: MessagesEventBusParams) {
this.id = nanoid(3)
this.agent = params.agent
this.__tempFromUserDid = params.__tempFromUserDid
this.subscribe = this.subscribe.bind(this)
this.getSnapshot = this.getSnapshot.bind(this)
this.init = this.init.bind(this)
this.suspend = this.suspend.bind(this)
this.resume = this.resume.bind(this)
this.setPollInterval = this.setPollInterval.bind(this)
this.trail = this.trail.bind(this)
}
private commit() {
this.snapshot = undefined
this.subscribers.forEach(subscriber => subscriber())
}
private subscribers: (() => void)[] = []
subscribe(subscriber: () => void) {
if (this.subscribers.length === 0) this.init()
this.subscribers.push(subscriber)
return () => {
this.subscribers = this.subscribers.filter(s => s !== subscriber)
if (this.subscribers.length === 0) this.suspend()
}
}
getSnapshot(): MessagesEventBusState {
if (!this.snapshot) this.snapshot = this.generateSnapshot()
// logger.debug(`${LOGGER_CONTEXT}: snapshotted`, {}, logger.DebugContext.convo)
return this.snapshot
}
private generateSnapshot(): MessagesEventBusState {
switch (this.status) {
case MessagesEventBusStatus.Initializing: {
return {
status: MessagesEventBusStatus.Initializing,
error: undefined,
setPollInterval: this.setPollInterval,
trail: this.trail,
}
}
case MessagesEventBusStatus.Ready: {
return {
status: this.status,
error: undefined,
setPollInterval: this.setPollInterval,
trail: this.trail,
}
}
case MessagesEventBusStatus.Suspended: {
return {
status: this.status,
error: undefined,
setPollInterval: this.setPollInterval,
trail: this.trail,
}
}
case MessagesEventBusStatus.Error: {
return {
status: MessagesEventBusStatus.Error,
error: this.error || {
code: MessagesEventBusErrorCode.Unknown,
retry: () => {
this.init()
},
},
setPollInterval: this.setPollInterval,
trail: this.trail,
}
}
default: {
return {
status: MessagesEventBusStatus.Uninitialized,
error: undefined,
setPollInterval: this.setPollInterval,
trail: this.trail,
}
}
}
}
dispatch(action: MessagesEventBusDispatch) {
const prevStatus = this.status
switch (this.status) {
case MessagesEventBusStatus.Uninitialized: {
switch (action.event) {
case MessagesEventBusDispatchEvent.Init: {
this.status = MessagesEventBusStatus.Initializing
this.setup()
break
}
}
break
}
case MessagesEventBusStatus.Initializing: {
switch (action.event) {
case MessagesEventBusDispatchEvent.Ready: {
this.status = MessagesEventBusStatus.Ready
this.pollInterval = ACTIVE_POLL_INTERVAL
this.restartPoll()
break
}
case MessagesEventBusDispatchEvent.Error: {
this.status = MessagesEventBusStatus.Error
this.error = action.payload
break
}
}
break
}
case MessagesEventBusStatus.Ready: {
switch (action.event) {
case MessagesEventBusDispatchEvent.Suspend: {
this.status = MessagesEventBusStatus.Suspended
this.cancelNextPoll()
break
}
case MessagesEventBusDispatchEvent.Error: {
this.status = MessagesEventBusStatus.Error
this.error = action.payload
this.cancelNextPoll()
break
}
}
break
}
case MessagesEventBusStatus.Suspended: {
switch (action.event) {
case MessagesEventBusDispatchEvent.Resume: {
this.status = MessagesEventBusStatus.Ready
this.pollInterval = ACTIVE_POLL_INTERVAL
this.restartPoll()
break
}
case MessagesEventBusDispatchEvent.Error: {
this.status = MessagesEventBusStatus.Error
this.error = action.payload
this.cancelNextPoll()
break
}
}
break
}
case MessagesEventBusStatus.Error: {
switch (action.event) {
case MessagesEventBusDispatchEvent.Init: {
this.status = MessagesEventBusStatus.Initializing
this.error = undefined
this.latestRev = undefined
this.setup()
break
}
}
break
}
default:
break
}
logger.debug(
`${LOGGER_CONTEXT}: dispatch '${action.event}'`,
{
id: this.id,
prev: prevStatus,
next: this.status,
},
logger.DebugContext.convo,
)
this.commit()
}
private async setup() {
logger.debug(`${LOGGER_CONTEXT}: setup`, {}, logger.DebugContext.convo)
try {
await this.initializeLatestRev()
// await new Promise(y => setTimeout(y, 2000))
// throw new Error('UNCOMMENT TO TEST INIT FAILURE')
this.dispatch({event: MessagesEventBusDispatchEvent.Ready})
} catch (e: any) {
logger.error(`${LOGGER_CONTEXT}: setup failed`)
this.dispatch({
event: MessagesEventBusDispatchEvent.Error,
payload: {
exception: e,
code: MessagesEventBusErrorCode.InitFailed,
retry: () => {
this.init()
},
},
})
}
}
init() {
logger.debug(`${LOGGER_CONTEXT}: init`, {}, logger.DebugContext.convo)
this.dispatch({event: MessagesEventBusDispatchEvent.Init})
}
suspend() {
logger.debug(`${LOGGER_CONTEXT}: suspend`, {}, logger.DebugContext.convo)
this.dispatch({event: MessagesEventBusDispatchEvent.Suspend})
}
resume() {
logger.debug(`${LOGGER_CONTEXT}: resume`, {}, logger.DebugContext.convo)
this.dispatch({event: MessagesEventBusDispatchEvent.Resume})
}
setPollInterval(interval: number) {
this.pollInterval = interval
this.restartPoll()
}
trail(handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void) {
this.emitter.on('events', handler)
return () => {
this.emitter.off('events', handler)
}
}
private async initializeLatestRev() {
logger.debug(
`${LOGGER_CONTEXT}: initialize latest rev`,
{},
logger.DebugContext.convo,
)
// throw new Error('UNCOMMENT TO TEST INIT FAILURE')
const response = await this.agent.api.chat.bsky.convo.listConvos(
{
limit: 1,
},
{
headers: {
Authorization: this.__tempFromUserDid,
},
},
)
const {convos} = response.data
for (const convo of convos) {
// set to latest rev
if (convo.rev > (this.latestRev = this.latestRev || convo.rev)) {
this.latestRev = convo.rev
}
}
}
private restartPoll() {
logger.debug(
`${LOGGER_CONTEXT}: restart poll`,
{},
logger.DebugContext.convo,
)
this.cancelNextPoll()
this.pollLatestEvents()
}
private cancelNextPoll() {
logger.debug(
`${LOGGER_CONTEXT}: cancel next poll`,
{},
logger.DebugContext.convo,
)
if (this.nextPoll) clearTimeout(this.nextPoll)
}
private pollLatestEvents() {
/*
* Uncomment to view poll events
*/
logger.debug(`${LOGGER_CONTEXT}: poll`, {}, logger.DebugContext.convo)
this.nextPoll = setTimeout(() => {
this.pollLatestEvents()
}, this.pollInterval)
this.fetchLatestEvents()
.then(({events}) => {
this.processLatestEvents(events)
})
.catch(e => {
logger.error(`${LOGGER_CONTEXT}: poll events failed`)
this.dispatch({
event: MessagesEventBusDispatchEvent.Error,
payload: {
exception: e,
code: MessagesEventBusErrorCode.PollFailed,
retry: () => {
this.init()
},
},
})
})
}
private pendingFetchLatestEvents:
| Promise<{
events: ChatBskyConvoGetLog.OutputSchema['logs']
}>
| undefined
async fetchLatestEvents() {
if (this.pendingFetchLatestEvents) return this.pendingFetchLatestEvents
this.pendingFetchLatestEvents = new Promise<{
events: ChatBskyConvoGetLog.OutputSchema['logs']
}>(async (resolve, reject) => {
try {
// throw new Error('UNCOMMENT TO TEST POLL FAILURE')
const response = await this.agent.api.chat.bsky.convo.getLog(
{
cursor: this.latestRev,
},
{
headers: {
Authorization: this.__tempFromUserDid,
},
},
)
const {logs} = response.data
resolve({events: logs})
} catch (e) {
reject(e)
} finally {
this.pendingFetchLatestEvents = undefined
}
})
return this.pendingFetchLatestEvents
}
private processLatestEvents(
events: ChatBskyConvoGetLog.OutputSchema['logs'],
) {
let needsEmit = false
let batch: ChatBskyConvoGetLog.OutputSchema['logs'] = []
for (const ev of events) {
/*
* If there's a rev, we should handle it. If there's not a rev, we don't
* know what it is.
*/
if (typeof ev.rev === 'string') {
/*
* We only care about new events
*/
if (ev.rev > (this.latestRev = this.latestRev || ev.rev)) {
/*
* Update rev regardless of if it's a ev type we care about or not
*/
this.latestRev = ev.rev
needsEmit = true
batch.push(ev)
}
}
}
if (needsEmit) {
this.emitter.emit('events', batch)
}
}
}
+60
View File
@@ -0,0 +1,60 @@
import React from 'react'
import {AppState} from 'react-native'
import {BskyAgent} from '@atproto-labs/api'
import {MessagesEventBus} from '#/state/messages/events/agent'
import {MessagesEventBusState} from '#/state/messages/events/types'
import {useAgent} from '#/state/session'
import {useDmServiceUrlStorage} from '#/screens/Messages/Temp/useDmServiceUrlStorage'
const MessagesEventBusContext =
React.createContext<MessagesEventBusState | null>(null)
export function useMessagesEventBus() {
const ctx = React.useContext(MessagesEventBusContext)
if (!ctx) {
throw new Error('useChat must be used within a ChatProvider')
}
return ctx
}
export function MessagesEventBusProvider({
children,
}: {
children: React.ReactNode
}) {
const {serviceUrl} = useDmServiceUrlStorage()
const {getAgent} = useAgent()
const [bus] = React.useState(
() =>
new MessagesEventBus({
agent: new BskyAgent({
service: serviceUrl,
}),
__tempFromUserDid: getAgent().session?.did!,
}),
)
const service = React.useSyncExternalStore(bus.subscribe, bus.getSnapshot)
React.useEffect(() => {
const handleAppStateChange = (nextAppState: string) => {
if (nextAppState === 'active') {
bus.resume()
} else {
bus.suspend()
}
}
const sub = AppState.addEventListener('change', handleAppStateChange)
return () => {
sub.remove()
}
}, [bus])
return (
<MessagesEventBusContext.Provider value={service}>
{children}
</MessagesEventBusContext.Provider>
)
}
+94
View File
@@ -0,0 +1,94 @@
import {BskyAgent, ChatBskyConvoGetLog} from '@atproto-labs/api'
export type MessagesEventBusParams = {
agent: BskyAgent
__tempFromUserDid: string
}
export enum MessagesEventBusStatus {
Uninitialized = 'uninitialized',
Initializing = 'initializing',
Ready = 'ready',
Error = 'error',
Suspended = 'suspended',
}
export enum MessagesEventBusDispatchEvent {
Init = 'init',
Ready = 'ready',
Error = 'error',
Suspend = 'suspend',
Resume = 'resume',
}
export enum MessagesEventBusErrorCode {
Unknown = 'unknown',
InitFailed = 'initFailed',
PollFailed = 'pollFailed',
}
export type MessagesEventBusError = {
code: MessagesEventBusErrorCode
exception?: Error
retry: () => void
}
export type MessagesEventBusDispatch =
| {
event: MessagesEventBusDispatchEvent.Init
}
| {
event: MessagesEventBusDispatchEvent.Ready
}
| {
event: MessagesEventBusDispatchEvent.Suspend
}
| {
event: MessagesEventBusDispatchEvent.Resume
}
| {
event: MessagesEventBusDispatchEvent.Error
payload: MessagesEventBusError
}
export type MessagesEventBusState =
| {
status: MessagesEventBusStatus.Uninitialized
error: undefined
setPollInterval: (interval: number) => void
trail: (
handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
) => () => void
}
| {
status: MessagesEventBusStatus.Initializing
error: undefined
setPollInterval: (interval: number) => void
trail: (
handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
) => () => void
}
| {
status: MessagesEventBusStatus.Ready
error: undefined
setPollInterval: (interval: number) => void
trail: (
handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
) => () => void
}
| {
status: MessagesEventBusStatus.Suspended
error: undefined
setPollInterval: (interval: number) => void
trail: (
handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
) => () => void
}
| {
status: MessagesEventBusStatus.Error
error: MessagesEventBusError
setPollInterval: (interval: number) => void
trail: (
handler: (events: ChatBskyConvoGetLog.OutputSchema['logs']) => void,
) => () => void
}