nicolium: experimental timeline: support streaming

Signed-off-by: nicole mikołajczyk <git@mkljczk.pl>

packages/nicolium/src/actions/timelines.ts

@@ -64,8 +64,6 @@ const processTimelineUpdate =
       return;
     }
 
-    importEntities({ statuses: [status] });
-
     if (shouldSkipQueue) {
       dispatch(updateTimeline(timeline, status.id));
     } else {

packages/nicolium/src/api/hooks/streaming/use-user-stream.ts

@@ -11,6 +11,7 @@ import { updateConversations } from '@/queries/conversations/use-conversations';
 import { queryKeys } from '@/queries/keys';
 import { useProcessStreamNotification } from '@/queries/notifications/use-notifications';
 import { useSettings } from '@/stores/settings';
+import { useTimelinesStore } from '@/stores/timelines';
 import { getUnreadChatsCount, updateChatListItem } from '@/utils/chats';
 import { play, soundCache } from '@/utils/sounds';
 
@@ -110,14 +111,19 @@ const useUserStream = () => {
 
   const listener = useCallback((event: StreamingEvent) => {
     switch (event.event) {
-      case 'update':
+      case 'update': {
+        const timelineId = getTimelineFromStream(event.stream);
         dispatch(processTimelineUpdate(getTimelineFromStream(event.stream), event.payload));
+        importEntities({ statuses: [event.payload] });
+        useTimelinesStore.getState().actions.receiveStreamingStatus(timelineId, event.payload);
         break;
+      }
       case 'status.update':
         importEntities({ statuses: [event.payload] });
         break;
       case 'delete':
         dispatch(deleteFromTimelines(event.payload));
+        useTimelinesStore.getState().actions.deleteStatus(event.payload);
         break;
       case 'notification':
         processStreamNotification(event.payload);

packages/nicolium/src/stores/timelines.ts

@@ -46,6 +46,8 @@ interface State {
       hasMore: boolean,
       initialFetch: boolean,
     ) => void;
+    receiveStreamingStatus: (timelineId: string, status: Status) => void;
+    deleteStatus: (statusId: string) => void;
     setLoading: (timelineId: string, isFetching: boolean) => void;
     dequeueEntries: (timelineId: string) => void;
   };
@@ -131,6 +133,41 @@ const useTimelinesStore = create<State>()(
           timeline.isFetching = false;
           state.timelines[timelineId] = timeline;
         }),
+      receiveStreamingStatus: (timelineId, status) => {
+        set((state) => {
+          const timeline = state.timelines[timelineId];
+          if (!timeline) return;
+
+          if (timeline.entries.some((entry) => entry.type === 'status' && entry.id === status.id))
+            return;
+
+          timeline.queuedEntries.unshift({
+            type: 'status',
+            id: status.id,
+            rebloggedBy: [],
+          });
+          timeline.queuedCount += 1;
+        });
+      },
+      deleteStatus: (statusId) => {
+        set((state) => {
+          for (const timeline of Object.values(state.timelines)) {
+            const entryIndex = timeline.entries.findIndex(
+              (entry) => entry.type === 'status' && entry.id === statusId,
+            );
+            if (entryIndex !== -1) {
+              timeline.entries.splice(entryIndex, 1);
+            }
+            const queuedEntryIndex = timeline.queuedEntries.findIndex(
+              (entry) => entry.type === 'status' && entry.id === statusId,
+            );
+            if (queuedEntryIndex !== -1) {
+              timeline.queuedEntries.splice(queuedEntryIndex, 1);
+              timeline.queuedCount = Math.max(timeline.queuedCount - 1, 0);
+            }
+          }
+        });
+      },
       setLoading: (timelineId, isFetching) =>
         set((state) => {
           const timeline = state.timelines[timelineId];