Add streaming updates for profiles in web UI

app/javascript/mastodon/actions/streaming.js

@@ -39,7 +39,7 @@ const randomUpTo = max =>
  * @param {Object} options
  * @param {function(Function, Function): Promise<void>} [options.fallback]
  * @param {function(): void} [options.fillGaps]
- * @param {function(object): boolean} [options.accept]
+ * @param {function(import("mastodon/api_types/statuses").ApiStatusJSON): boolean} [options.accept]
  * @returns {function(): void}
  */
 export const connectTimelineStream = (timelineId, channelName, params = {}, options = {}) => {
@@ -192,3 +192,32 @@ export const connectDirectStream = () =>
  */
 export const connectListStream = listId =>
   connectTimelineStream(`list:${listId}`, 'list', { list: listId }, { fillGaps: () => fillListTimelineGaps(listId) });
+
+/**
+ * @param {string} accountId
+ * @param {Object} options
+ * @param {boolean} [options.withReplies]
+ * @param {string} [options.tagged]
+ * @param {boolean} [options.onlyMedia]
+ * @returns {function(): void}
+ */
+export const connectProfileStream = (accountId, { withReplies, tagged, onlyMedia }) =>
+  connectTimelineStream(`account:${accountId}${onlyMedia ? ':media' : ''}${withReplies ? ':with_replies' : ''}${tagged ? `:${tagged}` : ''}`, 'profile', { account_id: accountId }, {
+    accept (status) {
+      let passThrough = true;
+
+      if (!withReplies) {
+        passThrough = passThrough && (status.in_reply_to_id === null || status.in_reply_to_account_id === status.account.id);
+      }
+
+      if (tagged) {
+        passThrough = passThrough && status.tags.some(tag => tag.name === tagged);
+      }
+
+      if (onlyMedia) {
+        passThrough = passThrough && status.media_attachments.length > 0;
+      }
+
+      return passThrough;
+    },
+  });

app/javascript/mastodon/features/account_gallery/index.jsx

@@ -8,11 +8,13 @@ import { connect } from 'react-redux';
 
 import { lookupAccount, fetchAccount } from 'mastodon/actions/accounts';
 import { openModal } from 'mastodon/actions/modal';
+import { connectProfileStream } from 'mastodon/actions/streaming';
 import { ColumnBackButton } from 'mastodon/components/column_back_button';
 import { LoadMore } from 'mastodon/components/load_more';
 import { LoadingIndicator } from 'mastodon/components/loading_indicator';
 import ScrollContainer from 'mastodon/containers/scroll_container';
 import BundleColumnError from 'mastodon/features/ui/components/bundle_column_error';
+import { identityContextPropShape, withIdentity } from 'mastodon/identity_context';
 import { normalizeForLookup } from 'mastodon/reducers/accounts_map';
 import { getAccountGallery } from 'mastodon/selectors';
 
@@ -80,6 +82,7 @@ class AccountGallery extends ImmutablePureComponent {
     blockedBy: PropTypes.bool,
     suspended: PropTypes.bool,
     multiColumn: PropTypes.bool,
+    identity: identityContextPropShape,
   };
 
   state = {
@@ -88,9 +91,20 @@ class AccountGallery extends ImmutablePureComponent {
 
   _load () {
     const { accountId, isAccount, dispatch } = this.props;
+    const { signedIn } = this.props.identity;
 
     if (!isAccount) dispatch(fetchAccount(accountId));
+
     dispatch(expandAccountMediaTimeline(accountId));
+
+    if (this.disconnect) {
+      this.disconnect();
+      this.disconnect = null;
+    }
+
+    if (signedIn) {
+      this.disconnect = dispatch(connectProfileStream(accountId, { onlyMedia: true }));
+    }
   }
 
   componentDidMount () {
@@ -103,6 +117,13 @@ class AccountGallery extends ImmutablePureComponent {
     }
   }
 
+  componentWillUnmount () {
+    if (this.disconnect) {
+      this.disconnect();
+      this.disconnect = null;
+    }
+  }
+
   componentDidUpdate (prevProps) {
     const { params: { acct }, accountId, dispatch } = this.props;
 
@@ -238,4 +259,4 @@ class AccountGallery extends ImmutablePureComponent {
 
 }
 
-export default connect(mapStateToProps)(AccountGallery);
+export default withIdentity(connect(mapStateToProps)(AccountGallery));

app/javascript/mastodon/features/account_timeline/index.jsx

@@ -7,21 +7,21 @@ import ImmutablePropTypes from 'react-immutable-proptypes';
 import ImmutablePureComponent from 'react-immutable-pure-component';
 import { connect } from 'react-redux';
 
+import { lookupAccount, fetchAccount } from 'mastodon/actions/accounts';
+import { fetchFeaturedTags } from 'mastodon/actions/featured_tags';
+import { connectProfileStream } from 'mastodon/actions/streaming';
+import { expandAccountFeaturedTimeline, expandAccountTimeline } from 'mastodon/actions/timelines';
+import { ColumnBackButton } from 'mastodon/components/column_back_button';
+import { LoadingIndicator } from 'mastodon/components/loading_indicator';
+import StatusList from 'mastodon/components/status_list';
 import { TimelineHint } from 'mastodon/components/timeline_hint';
 import BundleColumnError from 'mastodon/features/ui/components/bundle_column_error';
-import { me } from 'mastodon/initial_state';
+import Column from 'mastodon/features/ui/components/column';
+import { identityContextPropShape, withIdentity } from 'mastodon/identity_context';
 import { normalizeForLookup } from 'mastodon/reducers/accounts_map';
 import { getAccountHidden } from 'mastodon/selectors/accounts';
 import { useAppSelector } from 'mastodon/store';
 
-import { lookupAccount, fetchAccount } from '../../actions/accounts';
-import { fetchFeaturedTags } from '../../actions/featured_tags';
-import { expandAccountFeaturedTimeline, expandAccountTimeline, connectTimeline, disconnectTimeline } from '../../actions/timelines';
-import { ColumnBackButton } from '../../components/column_back_button';
-import { LoadingIndicator } from '../../components/loading_indicator';
-import StatusList from '../../components/status_list';
-import Column from '../ui/components/column';
-
 import { AccountHeader } from './components/account_header';
 import { LimitedAccountHint } from './components/limited_account_hint';
 
@@ -100,10 +100,12 @@ class AccountTimeline extends ImmutablePureComponent {
     remote: PropTypes.bool,
     remoteUrl: PropTypes.string,
     multiColumn: PropTypes.bool,
+    identity: identityContextPropShape,
   };
 
   _load () {
     const { accountId, withReplies, params: { tagged }, dispatch } = this.props;
+    const { signedIn } = this.props.identity;
 
     dispatch(fetchAccount(accountId));
 
@@ -114,8 +116,13 @@ class AccountTimeline extends ImmutablePureComponent {
     dispatch(fetchFeaturedTags(accountId));
     dispatch(expandAccountTimeline(accountId, { withReplies, tagged }));
 
-    if (accountId === me) {
+    if (this.disconnect) {
-      dispatch(connectTimeline(`account:${me}`));
+      this.disconnect();
+      this.disconnect = null;
+    }
+
+    if (signedIn) {
+      this.disconnect = dispatch(connectProfileStream(accountId, { withReplies, tagged }));
     }
   }
 
@@ -142,17 +149,12 @@ class AccountTimeline extends ImmutablePureComponent {
       }
       dispatch(expandAccountTimeline(accountId, { withReplies, tagged }));
     }
-
-    if (prevProps.accountId === me && accountId !== me) {
-      dispatch(disconnectTimeline({ timeline: `account:${me}` }));
-    }
   }
 
   componentWillUnmount () {
-    const { dispatch, accountId } = this.props;
+    if (this.disconnect) {
-
+      this.disconnect();
-    if (accountId === me) {
+      this.disconnect = null;
-      dispatch(disconnectTimeline({ timeline: `account:${me}` }));
     }
   }
 
@@ -218,4 +220,4 @@ class AccountTimeline extends ImmutablePureComponent {
 
 }
 
-export default connect(mapStateToProps)(AccountTimeline);
+export default withIdentity(connect(mapStateToProps)(AccountTimeline));

app/services/fan_out_on_write_service.rb

@@ -18,6 +18,7 @@ class FanOutOnWriteService < BaseService
     warm_payload_cache!
 
     fan_out_to_local_recipients!
+    fan_out_to_profile_streams! if distributable?
     fan_out_to_public_recipients! if broadcastable?
     fan_out_to_public_streams! if broadcastable?
   end
@@ -145,6 +146,10 @@ class FanOutOnWriteService < BaseService
     end
   end
 
+  def fan_out_to_profile_streams!
+    redis.publish("timeline:profile:#{@status.account_id}:public", anonymous_payload)
+  end
+
   def deliver_to_conversation!
     AccountConversation.add_status(@account, @status) unless update?
   end
@@ -168,6 +173,10 @@ class FanOutOnWriteService < BaseService
     @options[:update]
   end
 
+  def distributable?
+    @status.distributable?
+  end
+
   def broadcastable?
     @status.public_visibility? && !@status.reblog? && !@account.silenced?
   end

streaming/index.js

@@ -87,6 +87,7 @@ const PUBLIC_CHANNELS = [
   'public:remote:media',
   'hashtag',
   'hashtag:local',
+  'profile',
 ];
 
 // Used for priming the counters/gauges for the various metrics that are
@@ -420,6 +421,8 @@ const startServer = async () => {
       return 'direct';
     case '/api/v1/streaming/list':
       return 'list';
+    case '/api/v1/streaming/profile':
+      return 'profile';
     default:
       return undefined;
     }
@@ -593,6 +596,21 @@ const startServer = async () => {
     }
   };
 
+  /**
+   * @param {string} targetAccountId
+   * @param {http.IncomingMessage & ResolvedAccount} req
+   * @returns {Promise.<void>}
+   */
+  const authorizeProfileAccess = async (targetAccountId, req) => {
+    const { accountId } = req;
+
+    const result = await pgPool.query('SELECT 1 FROM blocks WHERE account_id = $1 AND target_account_id = $2 LIMIT 1', [targetAccountId, accountId]);
+
+    if (result.rows.length > 0) {
+      throw new AuthenticationError('Forbidden');
+    }
+  };
+
   /**
    * @param {string[]} channelIds
    * @param {http.IncomingMessage & ResolvedAccount} req
@@ -972,6 +990,7 @@ const startServer = async () => {
    * @property {string} [tag]
    * @property {string} [list]
    * @property {string} [only_media]
+   * @property {string} [account_id]
    */
 
   /**
@@ -1096,6 +1115,22 @@ const startServer = async () => {
         reject(new AuthenticationError('Not authorized to stream this list'));
       });
 
+      break;
+    case 'profile':
+      if (!params.account_id) {
+        reject(new RequestError('Missing account id parameter'));
+        return;
+      }
+
+      authorizeProfileAccess(params.account_id, req).then(() => {
+        resolve({
+          channelIds: [`timeline:profile:${params.account_id}:public`],
+          options: { needsFiltering: true },
+        });
+      }).catch(() => {
+        reject(new AuthenticationError('Not authorized to stream this profile'));
+      });
+
       break;
     default:
       reject(new RequestError('Unknown stream type'));