diff --git a/src/App.native.tsx b/src/App.native.tsx
index 9fa82e9cdb..08ef35bcf5 100644
--- a/src/App.native.tsx
+++ b/src/App.native.tsx
@@ -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}>
-
-
- {/* LabelDefsProvider MUST come before ModerationOptsProvider */}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+ {/* LabelDefsProvider MUST come before ModerationOptsProvider */}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/App.web.tsx b/src/App.web.tsx
index 9c2b34a788..9dfbfbe528 100644
--- a/src/App.web.tsx
+++ b/src/App.web.tsx
@@ -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}>
-
- {/* LabelDefsProvider MUST come before ModerationOptsProvider */}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+ {/* LabelDefsProvider MUST come before ModerationOptsProvider */}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
@@ -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) {
diff --git a/src/state/messages/events/agent.ts b/src/state/messages/events/agent.ts
new file mode 100644
index 0000000000..0e2921588c
--- /dev/null
+++ b/src/state/messages/events/agent.ts
@@ -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)
+ }
+ }
+}
diff --git a/src/state/messages/events/index.tsx b/src/state/messages/events/index.tsx
new file mode 100644
index 0000000000..34a6c881d5
--- /dev/null
+++ b/src/state/messages/events/index.tsx
@@ -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(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 (
+
+ {children}
+
+ )
+}
diff --git a/src/state/messages/events/types.ts b/src/state/messages/events/types.ts
new file mode 100644
index 0000000000..7c2d85cead
--- /dev/null
+++ b/src/state/messages/events/types.ts
@@ -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
+ }