Merge branch 'streaming-hooks' into 'develop'

Add hooks for Streaming API

Closes #1469

See merge request soapbox-pub/soapbox!2626

app/soapbox/actions/streaming.ts

@@ -10,18 +10,16 @@ import { connectStream } from '../stream';
 
 import {
   deleteAnnouncement,
-  fetchAnnouncements,
   updateAnnouncements,
   updateReaction as updateAnnouncementsReaction,
 } from './announcements';
 import { updateConversations } from './conversations';
 import { fetchFilters } from './filters';
 import { MARKER_FETCH_SUCCESS } from './markers';
-import { updateNotificationsQueue, expandNotifications } from './notifications';
+import { updateNotificationsQueue } from './notifications';
 import { updateStatus } from './statuses';
 import {
   // deleteFromTimelines,
-  expandHomeTimeline,
   connectTimeline,
   disconnectTimeline,
   processTimelineUpdate,
@@ -73,8 +71,9 @@ const updateChatQuery = (chat: IChat) => {
   queryClient.setQueryData<Chat>(ChatKeys.chat(chat.id), newChat as any);
 };
 
-interface StreamOpts {
+interface TimelineStreamOpts {
   statContext?: IStatContext
+  enabled?: boolean
 }
 
 const connectTimelineStream = (
@@ -82,7 +81,7 @@ const connectTimelineStream = (
   path: string,
   pollingRefresh: ((dispatch: AppDispatch, done?: () => void) => void) | null = null,
   accept: ((status: APIEntity) => boolean) | null = null,
-  opts?: StreamOpts,
+  opts?: TimelineStreamOpts,
 ) => connectStream(path, pollingRefresh, (dispatch: AppDispatch, getState: () => RootState) => {
   const locale = getLocale(getState());
 
@@ -191,49 +190,9 @@ const connectTimelineStream = (
   };
 });
 
-const refreshHomeTimelineAndNotification = (dispatch: AppDispatch, done?: () => void) =>
-  dispatch(expandHomeTimeline({}, () =>
-    dispatch(expandNotifications({}, () =>
-      dispatch(fetchAnnouncements(done))))));
-
-const connectUserStream      = (opts?: StreamOpts) =>
-  connectTimelineStream('home', 'user', refreshHomeTimelineAndNotification, null, opts);
-
-const connectCommunityStream = ({ onlyMedia }: Record<string, any> = {}) =>
-  connectTimelineStream(`community${onlyMedia ? ':media' : ''}`, `public:local${onlyMedia ? ':media' : ''}`);
-
-const connectPublicStream    = ({ onlyMedia }: Record<string, any> = {}) =>
-  connectTimelineStream(`public${onlyMedia ? ':media' : ''}`, `public${onlyMedia ? ':media' : ''}`);
-
-const connectRemoteStream    = (instance: string, { onlyMedia }: Record<string, any> = {}) =>
-  connectTimelineStream(`remote${onlyMedia ? ':media' : ''}:${instance}`, `public:remote${onlyMedia ? ':media' : ''}&instance=${instance}`);
-
-const connectHashtagStream   = (id: string, tag: string, accept: (status: APIEntity) => boolean) =>
-  connectTimelineStream(`hashtag:${id}`, `hashtag&tag=${tag}`, null, accept);
-
-const connectDirectStream    = () =>
-  connectTimelineStream('direct', 'direct');
-
-const connectListStream      = (id: string) =>
-  connectTimelineStream(`list:${id}`, `list&list=${id}`);
-
-const connectGroupStream     = (id: string) =>
-  connectTimelineStream(`group:${id}`, `group&group=${id}`);
-
-const connectNostrStream     = () =>
-  connectTimelineStream('nostr', 'nostr');
-
 export {
   STREAMING_CHAT_UPDATE,
   STREAMING_FOLLOW_RELATIONSHIPS_UPDATE,
   connectTimelineStream,
-  connectUserStream,
-  connectCommunityStream,
-  connectPublicStream,
-  connectRemoteStream,
-  connectHashtagStream,
-  connectDirectStream,
-  connectListStream,
-  connectGroupStream,
-  connectNostrStream,
+  type TimelineStreamOpts,
 };

app/soapbox/api/hooks/index.ts

@@ -43,3 +43,14 @@ export { useSuggestedGroups } from './groups/useSuggestedGroups';
 export { useUnmuteGroup } from './groups/useUnmuteGroup';
 export { useUpdateGroup } from './groups/useUpdateGroup';
 export { useUpdateGroupTag } from './groups/useUpdateGroupTag';
+
+// Streaming
+export { useUserStream } from './streaming/useUserStream';
+export { useCommunityStream } from './streaming/useCommunityStream';
+export { usePublicStream } from './streaming/usePublicStream';
+export { useDirectStream } from './streaming/useDirectStream';
+export { useHashtagStream } from './streaming/useHashtagStream';
+export { useListStream } from './streaming/useListStream';
+export { useGroupStream } from './streaming/useGroupStream';
+export { useRemoteStream } from './streaming/useRemoteStream';
+export { useNostrStream } from './streaming/useNostrStream';

app/soapbox/api/hooks/streaming/useCommunityStream.ts

@@ -0,0 +1,14 @@
+import { useTimelineStream } from './useTimelineStream';
+
+interface UseCommunityStreamOpts {
+  onlyMedia?: boolean
+}
+
+function useCommunityStream({ onlyMedia }: UseCommunityStreamOpts = {}) {
+  return useTimelineStream(
+    `community${onlyMedia ? ':media' : ''}`,
+    `public:local${onlyMedia ? ':media' : ''}`,
+  );
+}
+
+export { useCommunityStream };

app/soapbox/api/hooks/streaming/useDirectStream.ts

@@ -0,0 +1,17 @@
+import { useLoggedIn } from 'soapbox/hooks';
+
+import { useTimelineStream } from './useTimelineStream';
+
+function useDirectStream() {
+  const { isLoggedIn } = useLoggedIn();
+
+  return useTimelineStream(
+    'direct',
+    'direct',
+    null,
+    null,
+    { enabled: isLoggedIn },
+  );
+}
+
+export { useDirectStream };

app/soapbox/api/hooks/streaming/useGroupStream.ts

@@ -0,0 +1,10 @@
+import { useTimelineStream } from './useTimelineStream';
+
+function useGroupStream(groupId: string) {
+  return useTimelineStream(
+    `group:${groupId}`,
+    `group&group=${groupId}`,
+  );
+}
+
+export { useGroupStream };

app/soapbox/api/hooks/streaming/useHashtagStream.ts

@@ -0,0 +1,10 @@
+import { useTimelineStream } from './useTimelineStream';
+
+function useHashtagStream(tag: string) {
+  return useTimelineStream(
+    `hashtag:${tag}`,
+    `hashtag&tag=${tag}`,
+  );
+}
+
+export { useHashtagStream };

app/soapbox/api/hooks/streaming/useListStream.ts

@@ -0,0 +1,17 @@
+import { useLoggedIn } from 'soapbox/hooks';
+
+import { useTimelineStream } from './useTimelineStream';
+
+function useListStream(listId: string) {
+  const { isLoggedIn } = useLoggedIn();
+
+  return useTimelineStream(
+    `list:${listId}`,
+    `list&list=${listId}`,
+    null,
+    null,
+    { enabled: isLoggedIn },
+  );
+}
+
+export { useListStream };

app/soapbox/api/hooks/streaming/useNostrStream.ts

@@ -0,0 +1,20 @@
+import { useFeatures, useLoggedIn } from 'soapbox/hooks';
+
+import { useTimelineStream } from './useTimelineStream';
+
+function useNostrStream() {
+  const features = useFeatures();
+  const { isLoggedIn } = useLoggedIn();
+
+  return useTimelineStream(
+    'nostr',
+    'nostr',
+    null,
+    null,
+    {
+      enabled: isLoggedIn && features.nostrSign && Boolean(window.nostr),
+    },
+  );
+}
+
+export { useNostrStream };

app/soapbox/api/hooks/streaming/usePublicStream.ts

@@ -0,0 +1,14 @@
+import { useTimelineStream } from './useTimelineStream';
+
+interface UsePublicStreamOpts {
+  onlyMedia?: boolean
+}
+
+function usePublicStream({ onlyMedia }: UsePublicStreamOpts = {}) {
+  return useTimelineStream(
+    `public${onlyMedia ? ':media' : ''}`,
+    `public${onlyMedia ? ':media' : ''}`,
+  );
+}
+
+export { usePublicStream };

app/soapbox/api/hooks/streaming/useRemoteStream.ts

@@ -0,0 +1,15 @@
+import { useTimelineStream } from './useTimelineStream';
+
+interface UseRemoteStreamOpts {
+  instance: string
+  onlyMedia?: boolean
+}
+
+function useRemoteStream({ instance, onlyMedia }: UseRemoteStreamOpts) {
+  return useTimelineStream(
+    `remote${onlyMedia ? ':media' : ''}:${instance}`,
+    `public:remote${onlyMedia ? ':media' : ''}&instance=${instance}`,
+  );
+}
+
+export { useRemoteStream };

app/soapbox/api/hooks/streaming/useTimelineStream.ts

@@ -0,0 +1,42 @@
+import { useEffect, useRef } from 'react';
+
+import { connectTimelineStream } from 'soapbox/actions/streaming';
+import { useAppDispatch, useAppSelector, useInstance } from 'soapbox/hooks';
+import { getAccessToken } from 'soapbox/utils/auth';
+
+function useTimelineStream(...args: Parameters<typeof connectTimelineStream>) {
+  // TODO: get rid of streaming.ts and move the actual opts here.
+  const [timelineId, path] = args;
+  const { enabled = true } = args[4] ?? {};
+
+  const dispatch = useAppDispatch();
+  const instance = useInstance();
+  const stream = useRef<(() => void) | null>(null);
+
+  const accessToken = useAppSelector(getAccessToken);
+  const streamingUrl = instance.urls.get('streaming_api');
+
+  const connect = () => {
+    if (enabled && streamingUrl && !stream.current) {
+      stream.current = dispatch(connectTimelineStream(...args));
+    }
+  };
+
+  const disconnect = () => {
+    if (stream.current) {
+      stream.current();
+      stream.current = null;
+    }
+  };
+
+  useEffect(() => {
+    connect();
+    return disconnect;
+  }, [accessToken, streamingUrl, timelineId, path, enabled]);
+
+  return {
+    disconnect,
+  };
+}
+
+export { useTimelineStream };

app/soapbox/api/hooks/streaming/useUserStream.ts

@@ -0,0 +1,31 @@
+import { fetchAnnouncements } from 'soapbox/actions/announcements';
+import { expandNotifications } from 'soapbox/actions/notifications';
+import { expandHomeTimeline } from 'soapbox/actions/timelines';
+import { useStatContext } from 'soapbox/contexts/stat-context';
+import { useLoggedIn } from 'soapbox/hooks';
+
+import { useTimelineStream } from './useTimelineStream';
+
+import type { AppDispatch } from 'soapbox/store';
+
+function useUserStream() {
+  const { isLoggedIn } = useLoggedIn();
+  const statContext = useStatContext();
+
+  return useTimelineStream(
+    'home',
+    'user',
+    refresh,
+    null,
+    { statContext, enabled: isLoggedIn },
+  );
+}
+
+/** Refresh home timeline and notifications. */
+function refresh(dispatch: AppDispatch, done?: () => void) {
+  return dispatch(expandHomeTimeline({}, () =>
+    dispatch(expandNotifications({}, () =>
+      dispatch(fetchAnnouncements(done))))));
+}
+
+export { useUserStream };

app/soapbox/features/community-timeline/index.tsx

@@ -1,8 +1,8 @@
 import React, { useEffect } from 'react';
 import { defineMessages, FormattedMessage, useIntl } from 'react-intl';
 
-import { connectCommunityStream } from 'soapbox/actions/streaming';
 import { expandCommunityTimeline } from 'soapbox/actions/timelines';
+import { useCommunityStream } from 'soapbox/api/hooks';
 import PullToRefresh from 'soapbox/components/pull-to-refresh';
 import { Column } from 'soapbox/components/ui';
 import { useAppSelector, useAppDispatch, useSettings } from 'soapbox/hooks';
@@ -18,7 +18,7 @@ const CommunityTimeline = () => {
   const dispatch = useAppDispatch();
 
   const settings = useSettings();
-  const onlyMedia = settings.getIn(['community', 'other', 'onlyMedia']);
+  const onlyMedia = !!settings.getIn(['community', 'other', 'onlyMedia'], false);
   const next = useAppSelector(state => state.timelines.get('community')?.next);
 
   const timelineId = 'community';
@@ -28,16 +28,13 @@ const CommunityTimeline = () => {
   };
 
   const handleRefresh = () => {
-    return dispatch(expandCommunityTimeline({ onlyMedia } as any));
+    return dispatch(expandCommunityTimeline({ onlyMedia }));
   };
 
-  useEffect(() => {
-    dispatch(expandCommunityTimeline({ onlyMedia } as any));
-    const disconnect = dispatch(connectCommunityStream({ onlyMedia } as any));
+  useCommunityStream({ onlyMedia });
 
-    return () => {
-      disconnect();
-    };
+  useEffect(() => {
+    dispatch(expandCommunityTimeline({ onlyMedia }));
   }, [onlyMedia]);
 
   return (

app/soapbox/features/conversations/index.tsx

@@ -3,7 +3,7 @@ import { defineMessages, useIntl } from 'react-intl';
 
 import { directComposeById } from 'soapbox/actions/compose';
 import { mountConversations, unmountConversations, expandConversations } from 'soapbox/actions/conversations';
-import { connectDirectStream } from 'soapbox/actions/streaming';
+import { useDirectStream } from 'soapbox/api/hooks';
 import AccountSearch from 'soapbox/components/account-search';
 import { Column } from 'soapbox/components/ui';
 import { useAppDispatch } from 'soapbox/hooks';
@@ -19,15 +19,14 @@ const ConversationsTimeline = () => {
   const intl = useIntl();
   const dispatch = useAppDispatch();
 
+  useDirectStream();
+
   useEffect(() => {
     dispatch(mountConversations());
     dispatch(expandConversations());
 
-    const disconnect = dispatch(connectDirectStream());
-
     return () => {
       dispatch(unmountConversations());
-      disconnect();
     };
   }, []);
 

app/soapbox/features/direct-timeline/index.tsx

@@ -2,8 +2,8 @@ import React, { useEffect } from 'react';
 import { defineMessages, FormattedMessage, useIntl } from 'react-intl';
 
 import { directComposeById } from 'soapbox/actions/compose';
-import { connectDirectStream } from 'soapbox/actions/streaming';
 import { expandDirectTimeline } from 'soapbox/actions/timelines';
+import { useDirectStream } from 'soapbox/api/hooks';
 import AccountSearch from 'soapbox/components/account-search';
 import { Column } from 'soapbox/components/ui';
 import { useAppSelector, useAppDispatch } from 'soapbox/hooks';
@@ -20,13 +20,10 @@ const DirectTimeline = () => {
   const dispatch = useAppDispatch();
   const next = useAppSelector(state => state.timelines.get('direct')?.next);
 
+  useDirectStream();
+
   useEffect(() => {
     dispatch(expandDirectTimeline());
-    const disconnect = dispatch(connectDirectStream());
-
-    return (() => {
-      disconnect();
-    });
   }, []);
 
   const handleSuggestion = (accountId: string) => {

app/soapbox/features/group/group-timeline.tsx

@@ -4,9 +4,8 @@ import { FormattedMessage, useIntl } from 'react-intl';
 import { Link } from 'react-router-dom';
 
 import { groupCompose, setGroupTimelineVisible, uploadCompose } from 'soapbox/actions/compose';
-import { connectGroupStream } from 'soapbox/actions/streaming';
 import { expandGroupFeaturedTimeline, expandGroupTimeline } from 'soapbox/actions/timelines';
-import { useGroup } from 'soapbox/api/hooks';
+import { useGroup, useGroupStream } from 'soapbox/api/hooks';
 import { Avatar, HStack, Icon, Stack, Text, Toggle } from 'soapbox/components/ui';
 import ComposeForm from 'soapbox/features/compose/components/compose-form';
 import { useAppDispatch, useAppSelector, useDraggedFiles, useOwnAccount } from 'soapbox/hooks';
@@ -49,16 +48,12 @@ const GroupTimeline: React.FC<IGroupTimeline> = (props) => {
     dispatch(setGroupTimelineVisible(composeId, !groupTimelineVisible));
   };
 
+  useGroupStream(groupId);
+
   useEffect(() => {
     dispatch(expandGroupTimeline(groupId));
     dispatch(expandGroupFeaturedTimeline(groupId));
     dispatch(groupCompose(composeId, groupId));
-
-    const disconnect = dispatch(connectGroupStream(groupId));
-
-    return () => {
-      disconnect();
-    };
   }, [groupId]);
 
   if (!group) {

app/soapbox/features/hashtag-timeline/index.tsx

@@ -1,96 +1,31 @@
-import React, { useEffect, useRef } from 'react';
-import { useIntl, defineMessages, FormattedMessage } from 'react-intl';
+import React, { useEffect } from 'react';
+import { FormattedMessage } from 'react-intl';
 
-import { connectHashtagStream } from 'soapbox/actions/streaming';
 import { fetchHashtag, followHashtag, unfollowHashtag } from 'soapbox/actions/tags';
 import { expandHashtagTimeline, clearTimeline } from 'soapbox/actions/timelines';
+import { useHashtagStream } from 'soapbox/api/hooks';
 import List, { ListItem } from 'soapbox/components/list';
 import { Column, Toggle } from 'soapbox/components/ui';
 import Timeline from 'soapbox/features/ui/components/timeline';
 import { useAppDispatch, useAppSelector, useFeatures } from 'soapbox/hooks';
 
-import type { Tag as TagEntity } from 'soapbox/types/entities';
-
-type Mode = 'any' | 'all' | 'none';
-
-type Tag = { value: string };
-type Tags = { [k in Mode]: Tag[] };
-
-const messages = defineMessages({
-  any: { id: 'hashtag.column_header.tag_mode.any', defaultMessage: 'or {additional}' },
-  all: { id: 'hashtag.column_header.tag_mode.all', defaultMessage: 'and {additional}' },
-  none: { id: 'hashtag.column_header.tag_mode.none', defaultMessage: 'without {additional}' },
-  empty: { id: 'empty_column.hashtag', defaultMessage: 'There is nothing in this hashtag yet.' },
-});
-
 interface IHashtagTimeline {
   params?: {
     id?: string
-    tags?: Tags
   }
 }
 
 export const HashtagTimeline: React.FC<IHashtagTimeline> = ({ params }) => {
-  const intl = useIntl();
   const id = params?.id || '';
-  const tags = params?.tags || { any: [], all: [], none: [] };
-  
+
   const features = useFeatures();
   const dispatch = useAppDispatch();
-  const disconnects = useRef<(() => void)[]>([]);
   const tag = useAppSelector((state) => state.tags.get(id));
   const next = useAppSelector(state => state.timelines.get(`hashtag:${id}`)?.next);
 
-  // Mastodon supports displaying results from multiple hashtags.
-  // https://github.com/mastodon/mastodon/issues/6359
-  const title = (): string => {
-    const title: string[] = [`#${id}`];
-
-    if (additionalFor('any')) {
-      title.push(' ', intl.formatMessage(messages.any, { additional: additionalFor('any') }));
-    }
-
-    if (additionalFor('all')) {
-      title.push(' ', intl.formatMessage(messages.any, { additional: additionalFor('all') }));
-    }
-
-    if (additionalFor('none')) {
-      title.push(' ', intl.formatMessage(messages.any, { additional: additionalFor('none') }));
-    }
-
-    return title.join('');
-  };
-
-  const additionalFor = (mode: Mode) => {
-    if (tags && (tags[mode] || []).length > 0) {
-      return tags[mode].map(tag => tag.value).join('/');
-    } else {
-      return '';
-    }
-  };
-
-  const subscribe = () => {
-    const any  = tags.any.map(tag => tag.value);
-    const all  = tags.all.map(tag => tag.value);
-    const none = tags.none.map(tag => tag.value);
-
-    [id, ...any].map(tag => {
-      disconnects.current.push(dispatch(connectHashtagStream(id, tag, status => {
-        const tags = status.tags.map((tag: TagEntity) => tag.name);
-
-        return all.filter(tag => tags.includes(tag)).length === all.length &&
-               none.filter(tag => tags.includes(tag)).length === 0;
-      })));
-    });
-  };
-
-  const unsubscribe = () => {
-    disconnects.current.map(disconnect => disconnect());
-    disconnects.current = [];
-  };
 
   const handleLoadMore = (maxId: string) => {
-    dispatch(expandHashtagTimeline(id, { url: next, maxId, tags }));
+    dispatch(expandHashtagTimeline(id, { url: next, maxId }));
   };
 
   const handleFollow = () => {
@@ -101,25 +36,20 @@ export const HashtagTimeline: React.FC<IHashtagTimeline> = ({ params }) => {
     }
   };
 
-  useEffect(() => {
-    subscribe();
-    dispatch(expandHashtagTimeline(id, { tags }));
-    dispatch(fetchHashtag(id));
+  useHashtagStream(id);
 
-    return () => {
-      unsubscribe();
-    };
+  useEffect(() => {
+    dispatch(expandHashtagTimeline(id));
+    dispatch(fetchHashtag(id));
   }, []);
 
   useEffect(() => {
-    unsubscribe();
-    subscribe();
     dispatch(clearTimeline(`hashtag:${id}`));
-    dispatch(expandHashtagTimeline(id, { tags }));
+    dispatch(expandHashtagTimeline(id));
   }, [id]);
 
   return (
-    <Column bodyClassName='space-y-3' label={title()} transparent>
+    <Column label={`#${id}`} transparent>
       {features.followHashtags && (
         <List>
           <ListItem
@@ -136,7 +66,7 @@ export const HashtagTimeline: React.FC<IHashtagTimeline> = ({ params }) => {
         scrollKey='hashtag_timeline'
         timelineId={`hashtag:${id}`}
         onLoadMore={handleLoadMore}
-        emptyMessage={intl.formatMessage(messages.empty)}
+        emptyMessage={<FormattedMessage id='empty_column.hashtag' defaultMessage='There is nothing in this hashtag yet.' />}
         divideType='space'
       />
     </Column>

app/soapbox/features/list-timeline/index.tsx

@@ -4,8 +4,8 @@ import { useParams } from 'react-router-dom';
 
 import { fetchList } from 'soapbox/actions/lists';
 import { openModal } from 'soapbox/actions/modals';
-import { connectListStream } from 'soapbox/actions/streaming';
 import { expandListTimeline } from 'soapbox/actions/timelines';
+import { useListStream } from 'soapbox/api/hooks';
 import MissingIndicator from 'soapbox/components/missing-indicator';
 import { Column, Button, Spinner } from 'soapbox/components/ui';
 import { useAppDispatch, useAppSelector } from 'soapbox/hooks';
@@ -19,15 +19,11 @@ const ListTimeline: React.FC = () => {
   const list = useAppSelector((state) => state.lists.get(id));
   const next = useAppSelector(state => state.timelines.get(`list:${id}`)?.next);
 
+  useListStream(id);
+
   useEffect(() => {
     dispatch(fetchList(id));
     dispatch(expandListTimeline(id));
-
-    const disconnect = dispatch(connectListStream(id));
-
-    return () => {
-      disconnect();
-    };
   }, [id]);
 
   const handleLoadMore = (maxId: string) => {

app/soapbox/features/public-timeline/index.tsx

@@ -3,8 +3,8 @@ import { defineMessages, FormattedMessage, useIntl } from 'react-intl';
 import { Link } from 'react-router-dom';
 
 import { changeSetting } from 'soapbox/actions/settings';
-import { connectPublicStream } from 'soapbox/actions/streaming';
 import { expandPublicTimeline } from 'soapbox/actions/timelines';
+import { usePublicStream } from 'soapbox/api/hooks';
 import PullToRefresh from 'soapbox/components/pull-to-refresh';
 import { Accordion, Column } from 'soapbox/components/ui';
 import { useAppSelector, useAppDispatch, useInstance, useSettings } from 'soapbox/hooks';
@@ -23,7 +23,7 @@ const CommunityTimeline = () => {
 
   const instance = useInstance();
   const settings = useSettings();
-  const onlyMedia = settings.getIn(['public', 'other', 'onlyMedia']);
+  const onlyMedia = !!settings.getIn(['public', 'other', 'onlyMedia'], false);
   const next = useAppSelector(state => state.timelines.get('public')?.next);
 
   const timelineId = 'public';
@@ -44,16 +44,13 @@ const CommunityTimeline = () => {
   };
 
   const handleRefresh = () => {
-    return dispatch(expandPublicTimeline({ onlyMedia } as any));
+    return dispatch(expandPublicTimeline({ onlyMedia }));
   };
 
-  useEffect(() => {
-    dispatch(expandPublicTimeline({ onlyMedia } as any));
-    const disconnect = dispatch(connectPublicStream({ onlyMedia }));
+  usePublicStream({ onlyMedia });
 
-    return () => {
-      disconnect();
-    };
+  useEffect(() => {
+    dispatch(expandPublicTimeline({ onlyMedia }));
   }, [onlyMedia]);
 
   return (

app/soapbox/features/remote-timeline/index.tsx

@@ -1,9 +1,9 @@
-import React, { useEffect, useRef } from 'react';
+import React, { useEffect } from 'react';
 import { FormattedMessage } from 'react-intl';
 import { useHistory } from 'react-router-dom';
 
-import { connectRemoteStream } from 'soapbox/actions/streaming';
 import { expandRemoteTimeline } from 'soapbox/actions/timelines';
+import { useRemoteStream } from 'soapbox/api/hooks';
 import IconButton from 'soapbox/components/icon-button';
 import { Column, HStack, Text } from 'soapbox/components/ui';
 import { useAppSelector, useAppDispatch, useSettings } from 'soapbox/hooks';
@@ -26,20 +26,12 @@ const RemoteTimeline: React.FC<IRemoteTimeline> = ({ params }) => {
   const instance = params?.instance as string;
   const settings = useSettings();
 
-  const stream = useRef<any>(null);
-
   const timelineId = 'remote';
   const onlyMedia = !!settings.getIn(['remote', 'other', 'onlyMedia']);
   const next = useAppSelector(state => state.timelines.get('remote')?.next);
 
   const pinned: boolean = (settings.getIn(['remote_timeline', 'pinnedHosts']) as any).includes(instance);
 
-  const disconnect = () => {
-    if (stream.current) {
-      stream.current();
-    }
-  };
-
   const handleCloseClick: React.MouseEventHandler = () => {
     history.push('/timeline/fediverse');
   };
@@ -48,15 +40,10 @@ const RemoteTimeline: React.FC<IRemoteTimeline> = ({ params }) => {
     dispatch(expandRemoteTimeline(instance, { url: next, maxId, onlyMedia }));
   };
 
-  useEffect(() => {
-    disconnect();
-    dispatch(expandRemoteTimeline(instance, { onlyMedia, maxId: undefined }));
-    stream.current = dispatch(connectRemoteStream(instance, { onlyMedia }));
+  useRemoteStream({ instance, onlyMedia });
 
-    return () => {
-      disconnect();
-      stream.current = null;
-    };
+  useEffect(() => {
+    dispatch(expandRemoteTimeline(instance, { onlyMedia, maxId: undefined }));
   }, [onlyMedia]);
 
   return (

app/soapbox/features/ui/index.tsx

@@ -14,16 +14,15 @@ import { openModal } from 'soapbox/actions/modals';
 import { expandNotifications } from 'soapbox/actions/notifications';
 import { register as registerPushNotifications } from 'soapbox/actions/push-notifications';
 import { fetchScheduledStatuses } from 'soapbox/actions/scheduled-statuses';
-import { connectNostrStream, connectUserStream } from 'soapbox/actions/streaming';
 import { fetchSuggestionsForTimeline } from 'soapbox/actions/suggestions';
 import { expandHomeTimeline } from 'soapbox/actions/timelines';
+import { useNostrStream, useUserStream } from 'soapbox/api/hooks';
 import GroupLookupHoc from 'soapbox/components/hoc/group-lookup-hoc';
 import withHoc from 'soapbox/components/hoc/with-hoc';
 import SidebarNavigation from 'soapbox/components/sidebar-navigation';
 import ThumbNavigation from 'soapbox/components/thumb-navigation';
 import { Layout } from 'soapbox/components/ui';
-import { useStatContext } from 'soapbox/contexts/stat-context';
-import { useAppDispatch, useAppSelector, useOwnAccount, useSoapboxConfig, useFeatures, useInstance, useDraggedFiles } from 'soapbox/hooks';
+import { useAppDispatch, useAppSelector, useOwnAccount, useSoapboxConfig, useFeatures, useDraggedFiles } from 'soapbox/hooks';
 import AdminPage from 'soapbox/pages/admin-page';
 import ChatsPage from 'soapbox/pages/chats-page';
 import DefaultPage from 'soapbox/pages/default-page';
@@ -39,7 +38,7 @@ import RemoteInstancePage from 'soapbox/pages/remote-instance-page';
 import SearchPage from 'soapbox/pages/search-page';
 import StatusPage from 'soapbox/pages/status-page';
 import { usePendingPolicy } from 'soapbox/queries/policies';
-import { getAccessToken, getVapidKey } from 'soapbox/utils/auth';
+import { getVapidKey } from 'soapbox/utils/auth';
 import { isStandalone } from 'soapbox/utils/state';
 
 import BackgroundShapes from './components/background-shapes';
@@ -363,21 +362,13 @@ const UI: React.FC<IUI> = ({ children }) => {
   const history = useHistory();
   const dispatch = useAppDispatch();
   const { data: pendingPolicy } = usePendingPolicy();
-  const instance = useInstance();
-  const statContext = useStatContext();
-
-  const userStream = useRef<any>(null);
-  const nostrStream = useRef<any>(null);
   const node = useRef<HTMLDivElement | null>(null);
-
   const me = useAppSelector(state => state.me);
   const { account } = useOwnAccount();
   const features = useFeatures();
   const vapidKey = useAppSelector(state => getVapidKey(state));
 
   const dropdownMenuIsOpen = useAppSelector(state => state.dropdown_menu.isOpen);
-  const accessToken = useAppSelector(state => getAccessToken(state));
-  const streamingUrl = instance.urls.get('streaming_api');
   const standalone = useAppSelector(isStandalone);
 
   const { isDragging } = useDraggedFiles(node);
@@ -390,28 +381,6 @@ const UI: React.FC<IUI> = ({ children }) => {
     }
   };
 
-  const connectStreaming = () => {
-    if (accessToken && streamingUrl) {
-      if (!userStream.current) {
-        userStream.current = dispatch(connectUserStream({ statContext }));
-      }
-      if (!nostrStream.current && features.nostrSign && window.nostr) {
-        nostrStream.current = dispatch(connectNostrStream());
-      }
-    }
-  };
-
-  const disconnectStreaming = () => {
-    if (userStream.current) {
-      userStream.current();
-      userStream.current = null;
-    }
-    if (nostrStream.current) {
-      nostrStream.current();
-      nostrStream.current = null;
-    }
-  };
-
   const handleDragEnter = (e: DragEvent) => e.preventDefault();
   const handleDragLeave = (e: DragEvent) => e.preventDefault();
   const handleDragOver = (e: DragEvent) => e.preventDefault();
@@ -458,10 +427,6 @@ const UI: React.FC<IUI> = ({ children }) => {
     if (window.Notification?.permission === 'default') {
       window.setTimeout(() => Notification.requestPermission(), 120 * 1000);
     }
-
-    return () => {
-      disconnectStreaming();
-    };
   }, []);
 
   useEffect(() => {
@@ -477,9 +442,8 @@ const UI: React.FC<IUI> = ({ children }) => {
     };
   }, []);
 
-  useEffect(() => {
-    connectStreaming();
-  }, [accessToken, streamingUrl]);
+  useUserStream();
+  useNostrStream();
 
   // The user has logged in
   useEffect(() => {

app/soapbox/locales/en.json

@@ -868,9 +868,6 @@
   "groups.search.placeholder": "Search My Groups",
   "groups.suggested.label": "Suggested Groups",
   "groups.tags.title": "Browse Topics",
-  "hashtag.column_header.tag_mode.all": "and {additional}",
-  "hashtag.column_header.tag_mode.any": "or {additional}",
-  "hashtag.column_header.tag_mode.none": "without {additional}",
   "hashtag.follow": "Follow hashtag",
   "header.home.label": "Home",
   "header.login.email.placeholder": "E-mail address",