From 652a9fabb72829590d7ea85b645f9dc2494337b5 Mon Sep 17 00:00:00 2001 From: Henry Jameson Date: Thu, 13 Aug 2026 01:12:48 +0300 Subject: [PATCH] initial streaming re-support --- src/api/websocket.js | 12 +- src/boot/routes.js | 2 +- src/components/timeline/timeline.js | 12 +- src/components/user_profile/user_profile.js | 13 +- src/modules/api.js | 150 -------------- .../timeline_fetcher.service.js | 56 +++--- src/stores/notifications.js | 4 +- src/stores/statuses.js | 28 ++- src/stores/streaming.js | 189 ++++++++++++++++++ src/stores/timelines.js | 133 +++++++++--- src/stores/users.js | 73 ++----- 11 files changed, 391 insertions(+), 281 deletions(-) create mode 100644 src/stores/streaming.js diff --git a/src/api/websocket.js b/src/api/websocket.js index ae2e5358f..863e609f5 100644 --- a/src/api/websocket.js +++ b/src/api/websocket.js @@ -126,14 +126,14 @@ export const handleMastoWS = ( const { data } = wsEvent if (!data) return const parsedEvent = JSON.parse(data) - const { event, payload } = parsedEvent + const { event, stream, payload } = parsedEvent if ( MASTODON_STREAMING_EVENTS.has(event) || PLEROMA_STREAMING_EVENTS.has(event) ) { // MastoBE and PleromaBE both send payload for delete as a PLAIN string if (event === 'delete') { - return { event, id: payload } + return { event, stream, id: payload } } const data = payload ? JSON.parse(payload) : null if (event === 'pleroma:respond') { @@ -150,13 +150,13 @@ export const handleMastoWS = ( } return null } else if (event === 'update') { - return { event, status: parseStatus(data) } + return { event, stream, status: parseStatus(data) } } else if (event === 'status.update') { - return { event, status: parseStatus(data) } + return { event, stream, status: parseStatus(data) } } else if (event === 'notification') { - return { event, notification: parseNotification(data) } + return { event, stream, notification: parseNotification(data) } } else if (event === 'pleroma:chat_update') { - return { event, chatUpdate: parseChat(data) } + return { event, stream, chatUpdate: parseChat(data) } } } else { console.warn('Unknown event', wsEvent) diff --git a/src/boot/routes.js b/src/boot/routes.js index 8f8335c9c..4ea6f39e6 100644 --- a/src/boot/routes.js +++ b/src/boot/routes.js @@ -238,7 +238,7 @@ export default (store) => { path: '/lists/:id', component: Timeline, props: (route) => ({ - timelineRef: { name: 'lists', argument: route.params.id }, + timelineRef: { name: 'list', argument: route.params.id }, }), }, { diff --git a/src/components/timeline/timeline.js b/src/components/timeline/timeline.js index a9433111d..a71097782 100644 --- a/src/components/timeline/timeline.js +++ b/src/components/timeline/timeline.js @@ -58,7 +58,7 @@ const Timeline = { filteredVisibleStatuses() { return [...this.timeline.visibleStatusesIds.keys()] .map((id) => this.timeline.statuses.get(id)) - .filter(({ pinned }) => this.skipPinned ? !pinned : true) + .filter(({ pinned }) => (this.skipPinned ? !pinned : true)) }, newStatusCount() { return this.timeline.newStatusCount @@ -132,6 +132,7 @@ const Timeline = { setTimeout(this.determineVisibleStatuses, 250) }, unmounted() { + this.timelineChange(null, this.timelineRef) window.removeEventListener('scroll', this.handleScroll) window.removeEventListener('keydown', this.handleShortKey) if (document.hidden !== undefined) @@ -143,15 +144,16 @@ const Timeline = { }, methods: { timelineChange(newTimeline, oldTimeline) { - // TODO this might not be necessary if we optimize mergeOrAdd const sameName = newTimeline?.name === oldTimeline?.name const sameArgument = newTimeline?.argument === oldTimeline?.argument if (sameName && sameArgument) return - if (oldTimeline && oldTimeline.name !== 'friends') { - useTimelinesStore().clearTimeline(oldTimeline.name) + if (oldTimeline) { + useTimelinesStore().deactivate(oldTimeline.name) + } + if (newTimeline) { + useTimelinesStore().activate(newTimeline.name, newTimeline.argument) } - useTimelinesStore().startFetchingTimeline(newTimeline.name, newTimeline.argument) }, stopBlockingClicks: debounce(function () { this.blockingClicks = false diff --git a/src/components/user_profile/user_profile.js b/src/components/user_profile/user_profile.js index db96926d9..109c0bbbc 100644 --- a/src/components/user_profile/user_profile.js +++ b/src/components/user_profile/user_profile.js @@ -79,12 +79,14 @@ const UserProfile = { return useMergedConfigStore().mergedConfig.compactProfiles }, friends() { - return [...useUsersStore().relationshipsLists.friends.get(this.user).keys()] - .map((id) => useUsersStore().findUser(id)) + return [ + ...useUsersStore().relationshipsLists.friends.get(this.user).keys(), + ].map((id) => useUsersStore().findUser(id)) }, followers() { - return [...useUsersStore().relationshipsLists.followers.get(this.user).keys()] - .map((id) => useUsersStore().findUser(id)) + return [ + ...useUsersStore().relationshipsLists.followers.get(this.user).keys(), + ].map((id) => useUsersStore().findUser(id)) }, }, methods: { @@ -93,7 +95,8 @@ const UserProfile = { }, fetchUsers(group) { return () => - useUsersStore()['fetch' + group](this.userId) + useUsersStore() + ['fetch' + group](this.userId) .then((result) => ({ items: result })) }, load(userNameOrId) { diff --git a/src/modules/api.js b/src/modules/api.js index ca7bba29c..3fddb819f 100644 --- a/src/modules/api.js +++ b/src/modules/api.js @@ -90,156 +90,6 @@ const api = { return dispatch('stopMastoUserSocket') }, - // MastoAPI 'User' sockets - startMastoUserSocket(store) { - return new Promise((resolve, reject) => { - try { - const { state, commit, dispatch, rootState } = store - const timelineData = rootState.statuses.timelines.friends - - const credentials = useOAuthStore().token - const url = getMastodonSocketURI({ credentials }) - - state.mastoUserSocket = ProcessedWS({ - url, - id: 'Unified', - credentials, - }) - - state.mastoUserSocket.addEventListener( - 'pleroma:authenticated', - () => { - state.mastoUserSocket.subscribe('user') - }, - ) - state.mastoUserSocket.addEventListener( - 'message', - ({ detail: message }) => { - if (!message) return // pings - if (message.event === 'notification') { - useNotificationsStore().addNewNotifications({ - timestamp: Date.now(), - data: message.notification, - }) - } else if (message.event === 'update') { - useStatusesStore().addNewStatuses({ - timestamp: Date.now(), - statuses: [message.status], - userId: false, - showImmediately: timelineData.visibleStatuses.length === 0, - timeline: 'friends', - }) - } else if (message.event === 'status.update') { - useStatusesStore().addNewStatuses({ - timestamp: Date.now(), - statuses: [message.status], - userId: false, - showImmediately: - message.status.id in timelineData.visibleStatusesObject, - timeline: 'friends', - }) - } else if (message.event === 'delete') { - dispatch('deleteStatusById', message.id) - } else if (message.event === 'pleroma:chat_update') { - // The setTimeout wrapper is a temporary band-aid to avoid duplicates for the user's own messages when doing optimistic sending. - // The cause of the duplicates is the WS event arriving earlier than the HTTP response. - // This setTimeout wrapper can be removed once the commit `8e41baff` is in the stable Pleroma release. - // (`8e41baff` adds the idempotency key to the chat message entity, which PleromaFE uses when it's available, and it makes this artificial delay unnecessary). - setTimeout(() => { - dispatch('addChatMessages', { - chatId: message.chatUpdate.id, - messages: [message.chatUpdate.lastMessage], - }) - dispatch('updateChat', { chat: message.chatUpdate }) - maybeShowChatNotification(store, message.chatUpdate) - }, 100) - } - }, - ) - state.mastoUserSocket.addEventListener('open', () => { - // Do not show notification when we just opened up the page - if ( - state.mastoUserSocketStatus !== - WSConnectionStatus.STARTING_INITIAL - ) { - useInterfaceStore().pushGlobalNotice({ - level: 'success', - messageKey: 'timeline.socket_reconnected', - timeout: 5000, - }) - } - // Stop polling if we were errored or disabled - if ( - new Set([ - WSConnectionStatus.ERROR, - WSConnectionStatus.DISABLED, - ]).has(state.mastoUserSocketStatus) - ) { - dispatch('stopFetchingTimeline', { timeline: 'friends' }) - dispatch('stopFetchingNotifications') - useChatsStore().stopFetchingChats() - } - commit('resetRetryMultiplier') - commit('setMastoUserSocketStatus', WSConnectionStatus.JOINED) - }) - state.mastoUserSocket.addEventListener( - 'error', - ({ detail: error }) => { - console.error('Error in MastoAPI websocket:', error) - // TODO is this needed? - dispatch('clearOpenedChats') - }, - ) - state.mastoUserSocket.addEventListener( - 'close', - ({ detail: closeEvent }) => { - const ignoreCodes = new Set([ - 1000, // Normal (intended) closure - 1001, // Going away - ]) - const { code } = closeEvent - if (ignoreCodes.has(code)) { - console.debug( - `Not restarting socket becasue of closure code ${code} is in ignore list`, - ) - commit('setMastoUserSocketStatus', WSConnectionStatus.CLOSED) - } else { - console.warn( - `MastoAPI websocket disconnected, restarting. CloseEvent code: ${code}`, - ) - setTimeout(() => { - dispatch('startMastoUserSocket') - }, retryTimeout(state.retryMultiplier)) - commit('incrementRetryMultiplier') - if (state.mastoUserSocketStatus !== WSConnectionStatus.ERROR) { - dispatch('startFetchingTimeline', { timeline: 'friends' }) - dispatch('startFetchingNotifications') - useChatsStore().startFetchingChats() - useInterfaceStore().pushGlobalNotice({ - level: 'error', - messageKey: 'timeline.socket_broke', - messageArgs: [code], - timeout: 5000, - }) - } - commit('setMastoUserSocketStatus', WSConnectionStatus.ERROR) - } - dispatch('clearOpenedChats') - }, - ) - resolve() - } catch (e) { - reject(e) - } - }) - }, - stopMastoUserSocket({ state, dispatch }) { - dispatch('startFetchingTimeline', { timeline: 'friends' }) - dispatch('startFetchingNotifications') - useChatsStore().startFetchingChats() - state.mastoUserSocket.close() - }, - // Notifications startFetchingNotifications(store) { if (store.state.fetchers.notifications) return diff --git a/src/services/timeline_fetcher/timeline_fetcher.service.js b/src/services/timeline_fetcher/timeline_fetcher.service.js index 331d0df7d..dfb202b4d 100644 --- a/src/services/timeline_fetcher/timeline_fetcher.service.js +++ b/src/services/timeline_fetcher/timeline_fetcher.service.js @@ -18,16 +18,10 @@ const REPLY_VISIBILITY_TIMELINES = new Set([ 'bubble', ]) -const fetchAndUpdate = ({ - timeline, - argument, - credentials, -}, { - maxId, - sinceId, - older = false, - showImmediately = false, -}) => { +const fetchAndUpdate = ( + { timeline, argument, credentials }, + { maxId, sinceId, older = false, showImmediately = false }, +) => { timeline.loading = true const { hideMutedPosts, replyVisibility } = useMergedConfigStore().mergedConfig @@ -71,15 +65,11 @@ const fetchAndUpdate = ({ .addNewStatuses({ statuses, timestamp }) .filter(Boolean) - useTimelinesStore().addStatusesToTimeline( - timeline.name, - argument, - { - statuses, - showImmediately, - pagination, - } - ) + useTimelinesStore().addStatusesToTimeline(timeline.name, argument, { + statuses, + showImmediately, + pagination, + }) return { statuses, pagination } }) .catch((error) => { @@ -101,7 +91,7 @@ const fetchAndUpdate = ({ const timelineFetcher = (timeline, argument, credentials) => { const state = { - interval: null + interval: null, } const boundFetchAndUpdate = ({ @@ -109,22 +99,26 @@ const timelineFetcher = (timeline, argument, credentials) => { maxId, sinceId, older, - } = {}) => fetchAndUpdate({ - timeline, - argument, - credentials, - }, { - maxId, - sinceId, - older, - showImmediately, - }) + } = {}) => + fetchAndUpdate( + { + timeline, + argument, + credentials, + }, + { + maxId, + sinceId, + older, + showImmediately, + }, + ) const startFetching = () => { if (state.interval) throw new Error('Interval already exists!') boundFetchAndUpdate({ - showImmediately: timeline.visibleStatusesIds.size === 0 + showImmediately: timeline.visibleStatusesIds.size === 0, }) state.interval = promiseInterval(boundFetchAndUpdate, 10000) diff --git a/src/stores/notifications.js b/src/stores/notifications.js index 034538dab..d9c388dcf 100644 --- a/src/stores/notifications.js +++ b/src/stores/notifications.js @@ -77,7 +77,9 @@ export const useNotificationsStore = defineStore('notifications', { useUsersStore().addNewUsers({ timestamp, - data: validNotifications.map((notification) => notification.from_profile), + data: validNotifications.map( + (notification) => notification.from_profile, + ), }) const statusNotifications = validNotifications.filter( diff --git a/src/stores/statuses.js b/src/stores/statuses.js index 699c2cf7c..a7f17c15c 100644 --- a/src/stores/statuses.js +++ b/src/stores/statuses.js @@ -4,6 +4,7 @@ import { useInstanceCapabilitiesStore } from 'src/stores/instance_capabilities.j import { useInterfaceStore } from 'src/stores/interface.js' import { useOAuthStore } from 'src/stores/oauth.js' import { useUsersStore } from 'src/stores/users.js' +import { useStreamingStore } from 'src/stores/streaming.js' import { fetchEmojiReactions, @@ -78,7 +79,30 @@ const getLatestScrobble = (user) => { export const useStatusesStore = defineStore('statuses', { state: defaultState, actions: { - addNewStatuses({ statuses, user = {}, userId, timestamp }) { + // Init + attachSocket() { + const et = new EventTarget() + const handleStatusMessage = ({ data, timestamp }) => { + console.log('STATUSES', data) + this.addNewStatuses({ statuses: [data.status], timestamp }) + } + + et.addEventListener('update', ({ detail: message }) => { + handleStatusMessage(message) + }) + + et.addEventListener('status.update', ({ detail: message }) => { + handleStatusMessage(message) + }) + + et.addEventListener('delete', ({ detail: message }) => { + console.log('DELETE', message) + this.deleteStatus(message.data) + }) + + useStreamingStore().addSubscriber({ et }) + }, + addNewStatuses({ statuses, timestamp }) { // Sanity check if (!Array.isArray(statuses)) { throw new TypeError("Statuses aren't an array!") @@ -132,7 +156,7 @@ export const useStatusesStore = defineStore('statuses', { if (status) { // This is our favorite, so the relevant bit. - if (favorite.user.id === user.id) { + if (favorite.user.id === useUsersStore().currentUser?.id) { status.favorited = true } else { status.fave_num += 1 diff --git a/src/stores/streaming.js b/src/stores/streaming.js new file mode 100644 index 000000000..0843980f2 --- /dev/null +++ b/src/stores/streaming.js @@ -0,0 +1,189 @@ +import { defineStore } from 'pinia' + +import { useOAuthStore } from 'src/stores/oauth.js' + +import { + getMastodonSocketURI, + ProcessedWS, + WSConnectionStatus, +} from 'src/api/websocket.js' + +const ARGUMENT_MAP = { + tag: 'tag', + list: 'list', +} + +export const TIMELINE_STREAM_MAP = { + friends: 'user', + public: 'public', + tag: 'hashtag', + list: 'list', + dms: 'direct', +} + +const retryTimeout = (multiplier) => 1000 * multiplier + +export const useStreamingStore = defineStore('streaming', { + state: () => ({ + socket: null, + error: null, + state: null, + retryMultiplier: 1, + subscribers: new Set(), + subscriptions: new Map(), + globalSubscriptions: new Set(), + }), + actions: { + addSubscriber(subscriber) { + const { stream, et } = subscriber + + if (stream) { + if (!this.subscriptions.has(stream.name)) { + this.subscriptions.set(stream.name, new Map()) + } + + const streamSubs = this.subscriptions.get(stream.name) + + if (streamSubs.has(stream.argument)) { + throw new Error('Subscription already exists!') + } + + streamSubs.set(stream.argument, subscriber) + } else { + this.globalSubscriptions.add(subscriber) + } + + this.subscribers.add(subscriber) + if (this.state === WSConnectionStatus.JOINED) { + this.socket.subscribe(...this.getSubArgs(stream)) + subscriber.et.dispatchEvent(new CustomEvent('open')) + } + }, + removeSubscriber(subscriber) { + const { stream, et } = subscriber + + this.subscribers.delete(subscriber) + this.subscriptions.get(stream.name).delete(stream.argument) + + if (this.state === WSConnectionStatus.JOINED) { + this.socket.unsubscribe(...this.getSubArgs(stream)) + } + }, + initSocket(initial) { + this.state = initial + ? WSConnectionStatus.STARTING_INITIAL + : WSConnectionStatus.STARTING + + const credentials = useOAuthStore().token + const url = getMastodonSocketURI({ credentials }) + + this.socket = ProcessedWS({ + url, + id: 'Unified', + credentials, + }) + + this.socket.addEventListener('pleroma:authenticated', this.onAuth) + this.socket.addEventListener('open', this.onOpen) + this.socket.addEventListener('close', this.onClose) + this.socket.addEventListener('message', this.onMessage) + this.socket.addEventListener('error', this.onError) + }, + stopSocket() { + this.socket.close() + }, + + getSubArgs(stream) { + const argumentKey = ARGUMENT_MAP[stream.name] + const args = argumentKey + ? { + [argumentKey]: stream.argument, + } + : null + + return [stream.name, args] + }, + onAuth() { + this.subscribers.forEach(({ stream, et }) => { + et.dispatchEvent(new CustomEvent('authenticated')) + + if (stream) { + this.socket.subscribe(...this.getSubArgs(stream)) + } + }) + this.state = WSConnectionStatus.JOINED + }, + onOpen() { + this.subscribers.forEach(({ stream, et }) => { + et.dispatchEvent(new CustomEvent('open')) + }) + }, + onMessage({ detail: message }) { + if (!message) return // pings + const timestamp = Date.now() + const { event: eventName, stream: eventStream, ...data } = message + const [streamName, streamArgument] = eventStream ?? [] + + const subscriber = this.subscriptions.get(streamName)?.get(streamArgument) + + const totalSubs = [ + ...this.globalSubscriptions.values(), + subscriber + ].filter(Boolean) + + totalSubs.forEach(({ stream, et }) => { + et.dispatchEvent(new CustomEvent( + eventName, + { + detail: { streamName, streamArgument, data, timestamp }, + } + )) + }) + + console.log('WS', message) + }, + onError({ detail: error }) { + this.subscribers.forEach(({ stream, et }) => { + et.dispatchEvent(new CustomEvent('error', error)) + }) + console.error('Error in MastoAPI websocket:', error) + }, + onClose({ detail: closeEvent }) { + const ignoreCodes = new Set([ + 1000, // Normal (intended) closure + 1001, // Going away + ]) + const { code } = closeEvent + + if (ignoreCodes.has(code)) { + console.debug( + `Not restarting socket becasue of closure code ${code} is in ignore list`, + ) + + this.state = WSConnectionStatus.CLOSED + + this.subscribers.forEach(({ et }) => { + et.dispatchEvent(new CustomEvent('close', closeEvent)) + }) + } else { + console.warn( + `MastoAPI websocket disconnected, restarting. CloseEvent code: ${code}`, + ) + + setTimeout(() => { + this.initSocket() + }, retryTimeout(this.retryMultiplier)) + + this.retryMultiplier += 1 + + if (this.state !== WSConnectionStatus.ERROR) { + this.subscribers.forEach(({ et }) => { + et.dispatchEvent(new CustomEvent('close', closeEvent)) + }) + } + + this.state = WSConnectionStatus.ERROR + } + }, + }, +}) diff --git a/src/stores/timelines.js b/src/stores/timelines.js index d78617a54..e9ad6853a 100644 --- a/src/stores/timelines.js +++ b/src/stores/timelines.js @@ -4,6 +4,8 @@ import { defineStore } from 'pinia' import { useInstanceCapabilitiesStore } from 'src/stores/instance_capabilities.js' import { useOAuthStore } from 'src/stores/oauth.js' import { useUsersStore } from 'src/stores/users.js' +import { useStatusesStore } from 'src/stores/statuses.js' +import { useStreamingStore, TIMELINE_STREAM_MAP } from 'src/stores/streaming.js' import timelineFetcher from 'src/services/timeline_fetcher/timeline_fetcher.service.js' @@ -17,6 +19,7 @@ const emptyTl = (name, argument = null) => { minId: '', minVisibleId: '', loading: false, + streaming: false, flushMarker: 0, fetcher: null, } @@ -27,6 +30,10 @@ const emptyTl = (name, argument = null) => { result[property] = argument } + if (name === 'dms' || name === 'friends') { + result.persistent = true + } + return result } @@ -41,24 +48,28 @@ export const ARGUMENT_MAP = { media: 'userId', } +const TIMELINES = new Set([ + 'mentions', + 'public', + 'user', + 'userPinned', + 'media', + 'favorites', + 'publicAndExternal', + 'friends', + 'tag', + 'dms', + 'bookmarks', + 'list', + 'bubble', + 'quotes', + 'search', +]) + export const defaultState = () => { - return Object.fromEntries([ - 'mentions', - 'public', - 'user', - 'userPinned', - 'media', - 'favorites', - 'publicAndExternal', - 'friends', - 'tag', - 'dms', - 'bookmarks', - 'list', - 'bubble', - 'quotes', - 'search', - ].map((name) => [name, emptyTl(name)])) + return Object.fromEntries( + [...TIMELINES].map((name) => [name, emptyTl(name)]), + ) } //const CUSTOM_SORT = new Set(['bookmarks', 'favorites']) @@ -123,7 +134,10 @@ export const useTimelinesStore = defineStore('timelines', { ) { // Add the mention to the mentions timeline if (timeline !== this.mentions) { - this.addStatusesToTimeline('mentions', null, { statuses, nested: true }) + this.addStatusesToTimeline('mentions', null, { + statuses, + nested: true, + }) } } @@ -135,10 +149,17 @@ export const useTimelinesStore = defineStore('timelines', { }) }, - // Fetchers - startFetchingTimeline(timelineName, argument) { + activatePersistents() { + TIMELINES.forEach(name => { + if (this[name].persistent) { + this.activate(name, undefined, true) + } + }) + }, + + activate(timelineName, argument, persistent) { const timeline = this[timelineName] - if (timeline.fetcher) return + if (timeline.persistent && !persistent) return if ( timelineName === 'favourites' && @@ -153,12 +174,64 @@ export const useTimelinesStore = defineStore('timelines', { useOAuthStore().token, ) + this.startFetchingTimeline(timelineName, argument) + + const streamName = TIMELINE_STREAM_MAP[timelineName] + + if (streamName) { + const et = new EventTarget() + et.addEventListener('open', () => this.onStreamConnect(timelineName, argument)) + et.addEventListener('close', () => this.onStreamDisconnect(timelineName, argument)) + et.addEventListener('update', ({ detail: message }) => this.onStreamMessage(timelineName, argument, message)) + + timeline.socket = { + stream: { + name: streamName, + argument, + }, + et, + } + + useStreamingStore().addSubscriber(timeline.socket) + } + }, + deactivate(timelineName) { + const timeline = this[timelineName] + if (timeline.persistent) return + this.clearTimeline(timelineName) + }, + + onStreamMessage(timeline, argument, event) { + // This relies on statuses store to process this event first + const status = useStatusesStore().allStatuses.get(event.data.status.id) + + this.addStatusesToTimeline(timeline, argument, { + statuses: [status], + }) + }, + + onStreamConnect(timeline) { + console.log('STREAM OK', timeline) + this[timeline].streaming = true + this.stopFetchingTimeline(timeline) + }, + + onStreamDisconnect(timeline, argument) { + console.log('STREAM DED', timeline, argument) + this[timeline].streaming = false + this.startFetchingTimeline(timeline, argument) + }, + + // Fetchers + startFetchingTimeline(timelineName, argument) { + console.log('START FETCHING', timelineName, argument) + const timeline = this[timelineName] timeline.fetcher.startFetching() }, stopFetchingTimeline(timelineName) { + console.log('STOP FETCHING', timelineName) const timeline = this[timelineName] - timeline.fetcher?.stopFetching() - timeline.fetcher = null + timeline.fetcher.stopFetching() }, // Queues & Timeline manip @@ -198,10 +271,16 @@ export const useTimelinesStore = defineStore('timelines', { this.updateTimelineExtremes(timeline, [...timeline.statuses.keys()]) }, - clearTimeline(timeline, excludeUserId = false) { - const userId = excludeUserId ? this[timeline].userId : undefined - this.stopFetchingTimeline(timeline) - this[timeline] = emptyTl(timeline, userId) + clearTimeline(timeline) { + console.log('CLEAR TIMELINE', timeline) + const data = this[timeline] + if (!data.streaming) { + this.stopFetchingTimeline(timeline) + } + if (data.socket) { + useStreamingStore().removeSubscriber(data.socket) + } + this[timeline] = emptyTl(timeline) }, queueFlush(timeline, id) { this[timeline].flushMarker = id diff --git a/src/stores/users.js b/src/stores/users.js index be9b91c71..8de0c60a3 100644 --- a/src/stores/users.js +++ b/src/stores/users.js @@ -23,8 +23,9 @@ import { useMergedConfigStore } from 'src/stores/merged_config.js' import { useNotificationsStore } from 'src/stores/notifications.js' import { useOAuthStore } from 'src/stores/oauth.js' import { useStatusesStore } from 'src/stores/statuses.js' -import { useTimelinesStore } from 'src/stores/timelines.js' import { useSyncConfigStore } from 'src/stores/sync_config.js' +import { useTimelinesStore } from 'src/stores/timelines.js' +import { useStreamingStore } from 'src/stores/streaming.js' import { useUserHighlightStore } from 'src/stores/user_highlight.js' import { revokeToken } from 'src/api/oauth.js' @@ -391,9 +392,7 @@ export const useUsersStore = defineStore('users', { fetchUserRelationship({ id, credentials: useOAuthStore().token, - }).then((result) => - this.updateUserRelationships(result), - ) + }).then((result) => this.updateUserRelationships(result)) } }, fetchUserInLists(id) { @@ -431,7 +430,7 @@ export const useUsersStore = defineStore('users', { const predictedRelationship = this.relationships[id] || { id } this.updateUserRelationships({ optimism: true, - data: [predictedRelationship] + data: [predictedRelationship], }) this.addBlockId(id) @@ -494,7 +493,7 @@ export const useUsersStore = defineStore('users', { predictedRelationship.muting = true this.updateUserRelationships({ optimism: true, - data: [predictedRelationship] + data: [predictedRelationship], }) this.addMuteId(id) @@ -512,7 +511,7 @@ export const useUsersStore = defineStore('users', { predictedRelationship.muting = false this.updateUserRelationships({ optimism: true, - data: [predictedRelationship] + data: [predictedRelationship], }) return unmuteUser({ id }).then(({ data: relationship }) => @@ -524,18 +523,14 @@ export const useUsersStore = defineStore('users', { id, reblogs: false, credentials: useOAuthStore().token, - }).then(({ data: relationship }) => - this.updateUserRelationships(data), - ) + }).then(({ data: relationship }) => this.updateUserRelationships(data)) }, showReblogs(id) { return followUser({ id, reblogs: true, credentials: useOAuthStore().token, - }).then(({ data: relationship }) => - this.updateUserRelationships(data), - ) + }).then(({ data: relationship }) => this.updateUserRelationships(data)) }, muteUsers(data = []) { return Promise.all(data.map((d) => this.muteUser(d))) @@ -600,18 +595,14 @@ export const useUsersStore = defineStore('users', { id, notify: true, credentials: useOAuthStore().token, - }).then(({ data: relationship }) => - this.updateUserRelationships(data), - ) + }).then(({ data: relationship }) => this.updateUserRelationships(data)) }, unsubscribeUser(id) { return followUser({ id, notify: false, credentials: useOAuthStore().token, - }).then(({ data: relationship }) => - this.updateUserRelationships(data), - ) + }).then(({ data: relationship }) => this.updateUserRelationships(data)) }, registerPushNotifications() { const token = this.currentUser.credentials @@ -729,55 +720,31 @@ export const useUsersStore = defineStore('users', { /**/ if (user.token) { + // Shoutbox dispatch('setWsToken', user.token) - - // Initialize the shout socket. dispatch('initializeSocket') } - const startPolling = () => { - // Start getting fresh posts. - useTimelinesStore().startFetchingTimeline('friends') + // DMs and Home + useTimelinesStore().activatePersistents() - // Start fetching notifications - dispatch('startFetchingNotifications') - - if (useInstanceCapabilitiesStore().pleromaChatMessagesAvailable) { - // Start fetching chats - useChatsStore().startFetchingChats() - } + if (useInstanceCapabilitiesStore().pleromaChatMessagesAvailable) { + // Start fetching chats + useChatsStore().startFetchingChats() } useListsStore().startFetching() useBookmarkFoldersStore().startFetching() + useStatusesStore().attachSocket() + //useNotificationsStore().attachSocket() if (user.locked) { dispatch('startFetchingFollowRequests') } + useStreamingStore().initSocket() if (useMergedConfigStore().mergedConfig.useStreamingApi) { - dispatch('fetchTimeline', { - timeline: 'friends', - sinceId: null, - }) - dispatch('fetchNotifications', { sinceId: null }) - dispatch('enableMastoSockets', true) - .catch((error) => { - console.error( - 'Failed initializing MastoAPI Streaming socket', - error, - ) - }) - .then(() => { - dispatch('fetchChats', { latest: true }) - setTimeout( - () => - useNotificationsStore().setNotificationsSilence(false), - 10000, - ) - }) - } else { - startPolling() + useStreamingStore().initSocket() } // Start fetching things that don't need to block the UI