conversation view works

src/components/chat_view/chat_view.js

@@ -1,7 +1,6 @@
 import { get, maxBy, minBy, sortBy, throttle } from 'lodash'
-import { mapState as mapPiniaState } from 'pinia'
+import { mapState } from 'pinia'
 import { nextTick } from 'vue'
-import { mapState } from 'vuex'
 
 import ChatMessageList from 'src/components/chat_message_list/chat_message_list.vue'
 import ChatTitle from 'src/components/chat_title/chat_title.vue'
@@ -20,6 +19,7 @@ import { useInterfaceStore } from 'src/stores/interface.js'
 import { useMergedConfigStore } from 'src/stores/merged_config.js'
 import { useOAuthStore } from 'src/stores/oauth.js'
 import { useStatusesStore } from 'src/stores/statuses.js'
+import { useStreamingStore } from 'src/stores/streaming.js'
 import { useUsersStore } from 'src/stores/users.js'
 
 import {
@@ -87,6 +87,8 @@ const Chat = {
 
       // Internal network stuff
       fetcher: null,
+      socket: null,
+      streaming: false,
       errorLoadingChat: false,
       messageRetriers: {},
       idempotencyKeyIndex: {},
@@ -94,7 +96,8 @@ const Chat = {
   },
   created() {
     if (this.testMode) return
-    this.startFetching()
+    this.activate()
+    this.attachSocket()
   },
   mounted() {
     window.addEventListener('resize', this.handleResize)
@@ -120,6 +123,9 @@ const Chat = {
         this.handleVisibilityChange,
         false,
       )
+
+    if (this.testMode) return
+    this.deactivate()
   },
   computed: {
     conversationId() {
@@ -159,17 +165,14 @@ const Chat = {
       if (this.isConversation) return false // Unsupported
       return (
         this.mergedConfig.useStreamingApi &&
-        this.mastoUserSocketStatus === WSConnectionStatus.JOINED
+        useStreamingStore().state === WSConnectionStatus.JOINED
       )
     },
-    ...mapPiniaState(useInterfaceStore, {
+    ...mapState(useInterfaceStore, {
       mobileLayout: (store) => store.layoutType === 'mobile',
     }),
-    ...mapPiniaState(useMergedConfigStore, ['mergedConfig']),
+    ...mapState(useMergedConfigStore, ['mergedConfig']),
-    ...mapPiniaState(useUsersStore, ['currentUser']),
+    ...mapState(useUsersStore, ['currentUser']),
-    ...mapState({
-      mastoUserSocketStatus: (state) => state.api.mastoUserSocketStatus,
-    }),
   },
   watch: {
     messages(old, neu) {
@@ -228,16 +231,85 @@ const Chat = {
         return
       }
 
-      this.clear()
+      this.deactivate()
-      this.startFetching()
+      this.activate()
-    },
-    mastoUserSocketStatus(newValue) {
-      if (newValue === WSConnectionStatus.JOINED) {
-        this.fetchChat({ isFirstFetch: true })
-      }
     },
   },
   methods: {
+    async activate() {
+      if (!this.isConversation) {
+        try {
+          const result = await getOrCreateChat({
+            accountId: this.chatUserId,
+            credentials: useOAuthStore().token,
+          })
+          useUsersStore().addNewUsers(result)
+          const { data } = result
+          data.account = useUsersStore().findUser(data.account.id)
+          this.chat = data
+        } catch (e) {
+          console.error('Error creating or getting a chat', e)
+          this.errorLoadingChat = true
+        }
+      }
+
+      if (this.isConversation || this.chat) {
+        this.$nextTick(() => {
+          this.scrollDown({ forceRead: true })
+        })
+        this.startFetching('Chat activated', true)
+      }
+    },
+    deactivate() {
+      this.clear()
+      if (!this.streaming) {
+        this.stopFetching()
+      }
+    },
+    attachSocket() {
+      const et = new EventTarget()
+      const socket = { et }
+
+      et.addEventListener('update', this.onStreamMessage)
+      et.addEventListener('open', this.onStreamConnect)
+      et.addEventListener('close', this.onStreamDisconnect)
+
+      useStreamingStore().addSubscriber(socket)
+      this.socket = socket
+    },
+    detachSocket() {
+      const { et } = this.socket
+
+      et.removeEventListener('update', this.onStreamMessage)
+      et.removeEventListener('open', this.onStreamConnect)
+      et.removeEventListener('close', this.onStreamDisconnect)
+
+      useStreamingStore().removeSubscriber(this.socket)
+    },
+
+    // Poll & Push
+    onStreamConnect() {
+      this.streaming = true
+      this.stopFetching('Socket connected')
+    },
+    onStreamDisconnect(closeEvent) {
+      this.streaming = false
+      this.startFetching('Socket disconnected')
+    },
+    startFetching(reason, isFirstFetch) {
+      console.debug('[Chat View] Started fetching', 'Reason:', reason)
+      this.fetcher = promiseInterval(
+        () => this.fetchChat({ fetchLatest: true }),
+        5000,
+      )
+      this.fetchChat({ isFirstFetch })
+    },
+    stopFetching(reason) {
+      console.debug('[Chat View] Stopped fetching', 'Reason:', reason)
+      this.fetcher.stop()
+      this.fetcher = null
+    },
+
     // Actions
     async readChat() {
       if (this.conversationId) return // Unsupported
@@ -261,18 +333,8 @@ const Chat = {
       this.lastReadMessageId = this.maxId
       this.newMessageCount = 0
     },
-    scrollDown(options = {}) {
+
-      const { behavior = 'auto', forceRead = false } = options
+    // Clears
-      this.$nextTick(() => {
-        window.scrollTo({
-          top: document.documentElement.scrollHeight,
-          behavior,
-        })
-      })
-      if (forceRead) {
-        this.readChat()
-      }
-    },
     cullOlder() {
       const maxIndex = this.messages.length
       const minIndex = maxIndex - 50
@@ -368,36 +430,12 @@ const Chat = {
         })
       }
     },
-    async startFetching() {
+    onStreamMessage({ data }) {
-      if (!this.isConversation) {
+      const messages = data.filter(
-        try {
+        ({ statusnet_conversation_id }) =>
-          const result = await getOrCreateChat({
+          statusnet_conversation_id === this.conversationId,
-            accountId: this.chatUserId,
-            credentials: useOAuthStore().token,
-          })
-          useUsersStore().addNewUsers(result)
-          const { data } = result
-          data.account = useUsersStore().findUser(data.account.id)
-          this.chat = data
-        } catch (e) {
-          console.error('Error creating or getting a chat', e)
-          this.errorLoadingChat = true
-        }
-      }
-
-      if (this.isConversation || this.chat) {
-        this.$nextTick(() => {
-          this.scrollDown({ forceRead: true })
-        })
-        this.doStartFetching()
-      }
-    },
-    doStartFetching() {
-      this.fetcher = promiseInterval(
-        () => this.fetchChat({ fetchLatest: true }),
-        5000,
       )
-      this.fetchChat({ isFirstFetch: true })
+      this.addMessages({ messages })
     },
     addMessages({ messages: newMessages }) {
       for (let i = 0; i < newMessages.length; i++) {
@@ -441,9 +479,6 @@ const Chat = {
         }
       }
     },
-    goBack() {
-      this.$router.back()
-    },
 
     // Optimistic posting (chats only)
     async sendMessage({ status, media, idempotencyKey }) {
@@ -621,6 +656,23 @@ const Chat = {
       })
     },
 
+    // Misc
+    scrollDown(options = {}) {
+      const { behavior = 'auto', forceRead = false } = options
+      this.$nextTick(() => {
+        window.scrollTo({
+          top: document.documentElement.scrollHeight,
+          behavior,
+        })
+      })
+      if (forceRead) {
+        this.readChat()
+      }
+    },
+    goBack() {
+      this.$router.back()
+    },
+
     // Ugly
     // TODO move to ChatMessage
     async deleteChatMessage({ chatId, messageId }) {

src/components/conversation/conversation.js

@@ -451,7 +451,7 @@ const conversation = {
           credentials: useOAuthStore().token,
         })
           .then(({ data: status }) => {
-            this.$store.dispatch('addNewStatuses', { statuses: [status] })
+            useStatusesStore().addNewStatuses({ statuses: [status] })
             this.fetchConversation()
           })
           .catch((error) => {

src/stores/streaming.js

@@ -86,7 +86,9 @@ export const useStreamingStore = defineStore('streaming', {
 
       this.subscribers.add(subscriber)
       if (this.state === WSConnectionStatus.JOINED) {
-        this.socket.subscribe(...this.getSubArgs(stream))
+        if (stream) {
+          this.socket.subscribe(...this.getSubArgs(stream))
+        }
         et.dispatchEvent(new StreamStateEvent('open'))
       }
     },
@@ -171,6 +173,7 @@ export const useStreamingStore = defineStore('streaming', {
           case 'delete':
             return [data.id]
           default:
+            console.log('UNKNOWN', eventName, eventStream, data)
             return data
         }
       })()