Merge remote-tracking branch 'upstream/master' into merge-glitch

14 files changed, 221 insertions, 57 deletions

diff --git a/app/controllers/api/base_controller.rb b/app/controllers/api/base_controller.rb
index 818819a3f..399d11766 100644
--- a/app/controllers/api/base_controller.rb
+++ b/app/controllers/api/base_controller.rb
@@ -103,7 +103,7 @@ class Api::BaseController < ApplicationController
     elsif !current_user.functional?
       render json: { error: 'Your login is currently disabled' }, status: 403
     else
-      set_user_activity
+      update_user_sign_in
     end
   end
 
diff --git a/app/controllers/auth/sessions_controller.rb b/app/controllers/auth/sessions_controller.rb
index 1cf6a0a59..548832b21 100644
--- a/app/controllers/auth/sessions_controller.rb
+++ b/app/controllers/auth/sessions_controller.rb
@@ -7,6 +7,7 @@ class Auth::SessionsController < Devise::SessionsController
 
   skip_before_action :require_no_authentication, only: [:create]
   skip_before_action :require_functional!
+  skip_before_action :update_user_sign_in
 
   prepend_before_action :set_pack
 
@@ -26,6 +27,7 @@ class Auth::SessionsController < Devise::SessionsController
 
   def create
     super do |resource|
+      resource.update_sign_in!(request, new_sign_in: true)
       remember_me(resource)
       flash.delete(:notice)
     end
@@ -59,7 +61,7 @@ class Auth::SessionsController < Devise::SessionsController
 
   def find_user
     if session[:attempt_user_id]
-      User.find(session[:attempt_user_id])
+      User.find_by(id: session[:attempt_user_id])
     else
       user   = User.authenticate_with_ldap(user_params) if Devise.ldap_authentication
       user ||= User.authenticate_with_pam(user_params) if Devise.pam_authentication
@@ -92,6 +94,7 @@ class Auth::SessionsController < Devise::SessionsController
 
   def require_no_authentication
     super
+
     # Delete flash message that isn't entirely useful and may be confusing in
     # most cases because /web doesn't display/clear flash messages.
     flash.delete(:alert) if flash[:alert] == I18n.t('devise.failure.already_authenticated')
@@ -113,13 +116,30 @@ class Auth::SessionsController < Devise::SessionsController
 
   def home_paths(resource)
     paths = [about_path]
+
     if single_user_mode? && resource.is_a?(User)
       paths << short_account_path(username: resource.account)
     end
+
     paths
   end
 
   def continue_after?
     truthy_param?(:continue)
   end
+
+  def restart_session
+    clear_attempt_from_session
+    redirect_to new_user_session_path, alert: I18n.t('devise.failure.timeout')
+  end
+
+  def set_attempt_session(user)
+    session[:attempt_user_id]         = user.id
+    session[:attempt_user_updated_at] = user.updated_at.to_s
+  end
+
+  def clear_attempt_from_session
+    session.delete(:attempt_user_id)
+    session.delete(:attempt_user_updated_at)
+  end
 end
diff --git a/app/controllers/concerns/sign_in_token_authentication_concern.rb b/app/controllers/concerns/sign_in_token_authentication_concern.rb
index f5178930b..51ebcb115 100644
--- a/app/controllers/concerns/sign_in_token_authentication_concern.rb
+++ b/app/controllers/concerns/sign_in_token_authentication_concern.rb
@@ -18,7 +18,9 @@ module SignInTokenAuthenticationConcern
   def authenticate_with_sign_in_token
     user = self.resource = find_user
 
-    if user_params[:sign_in_token_attempt].present? && session[:attempt_user_id]
+    if user.present? && session[:attempt_user_id].present? && session[:attempt_user_updated_at] != user.updated_at.to_s
+      restart_session
+    elsif user_params.key?(:sign_in_token_attempt) && session[:attempt_user_id]
       authenticate_with_sign_in_token_attempt(user)
     elsif user.present? && user.external_or_valid_password?(user_params[:password])
       prompt_for_sign_in_token(user)
@@ -27,7 +29,7 @@ module SignInTokenAuthenticationConcern
 
   def authenticate_with_sign_in_token_attempt(user)
     if valid_sign_in_token_attempt?(user)
-      session.delete(:attempt_user_id)
+      clear_attempt_from_session
       remember_me(user)
       sign_in(user)
     else
@@ -42,11 +44,11 @@ module SignInTokenAuthenticationConcern
       UserMailer.sign_in_token(user, request.remote_ip, request.user_agent, Time.now.utc.to_s).deliver_later!
     end
 
-    set_locale do
-      session[:attempt_user_id] = user.id
-      use_pack 'auth'
-      @body_classes = 'lighter'
-      render :sign_in_token
-    end
+    set_attempt_session(user)
+    use_pack 'auth'
+
+    @body_classes = 'lighter'
+
+    set_locale { render :sign_in_token }
   end
 end
diff --git a/app/controllers/concerns/two_factor_authentication_concern.rb b/app/controllers/concerns/two_factor_authentication_concern.rb
index 6b043a804..4800db348 100644
--- a/app/controllers/concerns/two_factor_authentication_concern.rb
+++ b/app/controllers/concerns/two_factor_authentication_concern.rb
@@ -37,9 +37,11 @@ module TwoFactorAuthenticationConcern
   def authenticate_with_two_factor
     user = self.resource = find_user
 
-    if user.webauthn_enabled? && user_params[:credential].present? && session[:attempt_user_id]
+    if user.present? && session[:attempt_user_id].present? && session[:attempt_user_updated_at] != user.updated_at.to_s
+      restart_session
+    elsif user.webauthn_enabled? && user_params.key?(:credential) && session[:attempt_user_id]
       authenticate_with_two_factor_via_webauthn(user)
-    elsif user_params[:otp_attempt].present? && session[:attempt_user_id]
+    elsif user_params.key?(:otp_attempt) && session[:attempt_user_id]
       authenticate_with_two_factor_via_otp(user)
     elsif user.present? && user.external_or_valid_password?(user_params[:password])
       prompt_for_two_factor(user)
@@ -50,7 +52,7 @@ module TwoFactorAuthenticationConcern
     webauthn_credential = WebAuthn::Credential.from_get(user_params[:credential])
 
     if valid_webauthn_credential?(user, webauthn_credential)
-      session.delete(:attempt_user_id)
+      clear_attempt_from_session
       remember_me(user)
       sign_in(user)
       render json: { redirect_path: root_path }, status: :ok
@@ -61,7 +63,7 @@ module TwoFactorAuthenticationConcern
 
   def authenticate_with_two_factor_via_otp(user)
     if valid_otp_attempt?(user)
-      session.delete(:attempt_user_id)
+      clear_attempt_from_session
       remember_me(user)
       sign_in(user)
     else
@@ -71,17 +73,20 @@ module TwoFactorAuthenticationConcern
   end
 
   def prompt_for_two_factor(user)
-    set_locale do
-      session[:attempt_user_id] = user.id
-      use_pack 'auth'
-      @body_classes = 'lighter'
-      @webauthn_enabled = user.webauthn_enabled?
-      @scheme_type = if user.webauthn_enabled? && user_params[:otp_attempt].blank?
-                       'webauthn'
-                     else
-                       'totp'
-                     end
-      render :two_factor
+    set_attempt_session(user)
+
+    use_pack 'auth'
+
+    @body_classes     = 'lighter'
+    @webauthn_enabled = user.webauthn_enabled?
+    @scheme_type      = begin
+      if user.webauthn_enabled? && user_params[:otp_attempt].blank?
+        'webauthn'
+      else
+        'totp'
+      end
     end
+
+    set_locale { render :two_factor }
   end
 end
diff --git a/app/controllers/concerns/user_tracking_concern.rb b/app/controllers/concerns/user_tracking_concern.rb
index be10705fc..efda37fae 100644
--- a/app/controllers/concerns/user_tracking_concern.rb
+++ b/app/controllers/concerns/user_tracking_concern.rb
@@ -6,14 +6,13 @@ module UserTrackingConcern
   UPDATE_SIGN_IN_HOURS = 24
 
   included do
-    before_action :set_user_activity
+    before_action :update_user_sign_in
   end
 
   private
 
-  def set_user_activity
-    return unless user_needs_sign_in_update?
-    current_user.update_tracked_fields!(request)
+  def update_user_sign_in
+    current_user.update_sign_in!(request) if user_needs_sign_in_update?
   end
 
   def user_needs_sign_in_update?
diff --git a/app/javascript/flavours/glitch/features/ui/components/favourite_modal.js b/app/javascript/flavours/glitch/features/ui/components/favourite_modal.js
index 176e7c487..ea1d7876e 100644
--- a/app/javascript/flavours/glitch/features/ui/components/favourite_modal.js
+++ b/app/javascript/flavours/glitch/features/ui/components/favourite_modal.js
@@ -7,11 +7,17 @@ import StatusContent from 'flavours/glitch/components/status_content';
 import Avatar from 'flavours/glitch/components/avatar';
 import RelativeTimestamp from 'flavours/glitch/components/relative_timestamp';
 import DisplayName from 'flavours/glitch/components/display_name';
+import AttachmentList from 'flavours/glitch/components/attachment_list';
 import Icon from 'flavours/glitch/components/icon';
 import ImmutablePureComponent from 'react-immutable-pure-component';
+import classNames from 'classnames';
 
 const messages = defineMessages({
   favourite: { id: 'status.favourite', defaultMessage: 'Favourite' },
+  public_short: { id: 'privacy.public.short', defaultMessage: 'Public' },
+  unlisted_short: { id: 'privacy.unlisted.short', defaultMessage: 'Unlisted' },
+  private_short: { id: 'privacy.private.short', defaultMessage: 'Followers-only' },
+  direct_short: { id: 'privacy.direct.short', defaultMessage: 'Direct' },
 });
 
 export default @injectIntl
@@ -54,13 +60,25 @@ class FavouriteModal extends ImmutablePureComponent {
   render () {
     const { status, intl } = this.props;
 
+    const visibilityIconInfo = {
+      'public': { icon: 'globe', text: intl.formatMessage(messages.public_short) },
+      'unlisted': { icon: 'unlock', text: intl.formatMessage(messages.unlisted_short) },
+      'private': { icon: 'lock', text: intl.formatMessage(messages.private_short) },
+      'direct': { icon: 'envelope', text: intl.formatMessage(messages.direct_short) },
+    };
+
+    const visibilityIcon = visibilityIconInfo[status.get('visibility')];
+
     return (
       <div className='modal-root__modal favourite-modal'>
         <div className='favourite-modal__container'>
-          <div className='status light'>
+          <div className={classNames('status', `status-${status.get('visibility')}`, 'light')}>
             <div className='favourite-modal__status-header'>
               <div className='favourite-modal__status-time'>
-                <a href={status.get('url')} className='status__relative-time' target='_blank' rel='noopener'><RelativeTimestamp timestamp={status.get('created_at')} /></a>
+                <a href={status.get('url')} className='status__relative-time' target='_blank' rel='noopener noreferrer'>
+                  <span className='status__visibility-icon'><Icon id={visibilityIcon.icon} title={visibilityIcon.text} /></span>
+                  <RelativeTimestamp timestamp={status.get('created_at')} />
+                </a>
               </div>
 
               <a onClick={this.handleAccountClick} href={status.getIn(['account', 'url'])} className='status__display-name'>
@@ -69,10 +87,18 @@ class FavouriteModal extends ImmutablePureComponent {
                 </div>
 
                 <DisplayName account={status.get('account')} />
+
               </a>
             </div>
 
             <StatusContent status={status} />
+
+            {status.get('media_attachments').size > 0 && (
+              <AttachmentList
+                compact
+                media={status.get('media_attachments')}
+              />
+            )}
           </div>
         </div>
 
diff --git a/app/lib/access_token_extension.rb b/app/lib/access_token_extension.rb
new file mode 100644
index 000000000..3e184e775
--- /dev/null
+++ b/app/lib/access_token_extension.rb
@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+module AccessTokenExtension
+  extend ActiveSupport::Concern
+
+  included do
+    after_commit :push_to_streaming_api
+  end
+
+  def revoke(clock = Time)
+    update(revoked_at: clock.now.utc)
+  end
+
+  def push_to_streaming_api
+    Redis.current.publish("timeline:access_token:#{id}", Oj.dump(event: :kill)) if revoked? || destroyed?
+  end
+end
diff --git a/app/models/session_activation.rb b/app/models/session_activation.rb
index 34d25c83d..b0ce9d112 100644
--- a/app/models/session_activation.rb
+++ b/app/models/session_activation.rb
@@ -70,12 +70,16 @@ class SessionActivation < ApplicationRecord
   end
 
   def assign_access_token
-    superapp = Doorkeeper::Application.find_by(superapp: true)
+    self.access_token = Doorkeeper::AccessToken.create!(access_token_attributes)
+  end
 
-    self.access_token = Doorkeeper::AccessToken.create!(application_id: superapp&.id,
-                                                        resource_owner_id: user_id,
-                                                        scopes: 'read write follow',
-                                                        expires_in: Doorkeeper.configuration.access_token_expires_in,
-                                                        use_refresh_token: Doorkeeper.configuration.refresh_token_enabled?)
+  def access_token_attributes
+    {
+      application_id: Doorkeeper::Application.find_by(superapp: true)&.id,
+      resource_owner_id: user_id,
+      scopes: 'read write follow',
+      expires_in: Doorkeeper.configuration.access_token_expires_in,
+      use_refresh_token: Doorkeeper.configuration.refresh_token_enabled?,
+    }
   end
 end
diff --git a/app/models/user.rb b/app/models/user.rb
index 167bfd605..8d91593ae 100644
--- a/app/models/user.rb
+++ b/app/models/user.rb
@@ -65,7 +65,7 @@ class User < ApplicationRecord
   devise :two_factor_backupable,
          otp_number_of_backup_codes: 10
 
-  devise :registerable, :recoverable, :rememberable, :trackable, :validatable,
+  devise :registerable, :recoverable, :rememberable, :validatable,
          :confirmable
 
   include Omniauthable
@@ -174,6 +174,24 @@ class User < ApplicationRecord
     prepare_new_user! if new_user && approved?
   end
 
+  def update_sign_in!(request, new_sign_in: false)
+    old_current, new_current = current_sign_in_at, Time.now.utc
+    self.last_sign_in_at     = old_current || new_current
+    self.current_sign_in_at  = new_current
+
+    old_current, new_current = current_sign_in_ip, request.remote_ip
+    self.last_sign_in_ip     = old_current || new_current
+    self.current_sign_in_ip  = new_current
+
+    if new_sign_in
+      self.sign_in_count ||= 0
+      self.sign_in_count  += 1
+    end
+
+    save(validate: false) unless new_record?
+    prepare_returning_user!
+  end
+
   def pending?
     !approved?
   end
@@ -205,11 +223,6 @@ class User < ApplicationRecord
     prepare_new_user!
   end
 
-  def update_tracked_fields!(request)
-    super
-    prepare_returning_user!
-  end
-
   def otp_enabled?
     otp_required_for_login
   end
diff --git a/app/serializers/activitypub/note_serializer.rb b/app/serializers/activitypub/note_serializer.rb
index 0b2ab0cfd..0d3fe715d 100644
--- a/app/serializers/activitypub/note_serializer.rb
+++ b/app/serializers/activitypub/note_serializer.rb
@@ -63,10 +63,6 @@ class ActivityPub::NoteSerializer < ActivityPub::Serializer
     object.spoiler_text.presence || (instance_options[:allow_local_only] ? nil : Setting.outgoing_spoilers.presence)
   end
 
-  def sensitive
-    object.sensitive || (!instance_options[:allow_local_only] && Setting.outgoing_spoilers.present?)
-  end
-
   def direct_message
     object.direct_visibility?
   end
@@ -138,7 +134,7 @@ class ActivityPub::NoteSerializer < ActivityPub::Serializer
   end
 
   def sensitive
-    object.account.sensitized? || object.sensitive
+    object.account.sensitized? || object.sensitive || (!instance_options[:allow_local_only] && Setting.outgoing_spoilers.present?)
   end
 
   def virtual_tags
diff --git a/app/workers/account_deletion_worker.rb b/app/workers/account_deletion_worker.rb
index b6016bf8c..81c3b91ad 100644
--- a/app/workers/account_deletion_worker.rb
+++ b/app/workers/account_deletion_worker.rb
@@ -5,7 +5,8 @@ class AccountDeletionWorker
 
   sidekiq_options queue: 'pull'
 
-  def perform(account_id, reserve_username: true)
+  def perform(account_id, options = {})
+    reserve_username = options.with_indifferent_access.fetch(:reserve_username, true)
     DeleteAccountService.new.call(Account.find(account_id), reserve_username: reserve_username, reserve_email: false)
   rescue ActiveRecord::RecordNotFound
     true
diff --git a/config/application.rb b/config/application.rb
index bc372851c..1fb79a4aa 100644
--- a/config/application.rb
+++ b/config/application.rb
@@ -142,6 +142,7 @@ module Mastodon
       Doorkeeper::AuthorizationsController.layout 'modal'
       Doorkeeper::AuthorizedApplicationsController.layout 'admin'
       Doorkeeper::Application.send :include, ApplicationExtension
+      Doorkeeper::AccessToken.send :include, AccessTokenExtension
       Devise::FailureApp.send :include, AbstractController::Callbacks
       Devise::FailureApp.send :include, HttpAcceptLanguage::EasyAccess
       Devise::FailureApp.send :include, Localized
diff --git a/spec/controllers/auth/sessions_controller_spec.rb b/spec/controllers/auth/sessions_controller_spec.rb
index 8ad9e74fc..d3a9a11eb 100644
--- a/spec/controllers/auth/sessions_controller_spec.rb
+++ b/spec/controllers/auth/sessions_controller_spec.rb
@@ -219,7 +219,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
 
         context 'using a valid OTP' do
           before do
-            post :create, params: { user: { otp_attempt: user.current_otp } }, session: { attempt_user_id: user.id }
+            post :create, params: { user: { otp_attempt: user.current_otp } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
           end
 
           it 'redirects to home' do
@@ -234,7 +234,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
         context 'when the server has an decryption error' do
           before do
             allow_any_instance_of(User).to receive(:validate_and_consume_otp!).and_raise(OpenSSL::Cipher::CipherError)
-            post :create, params: { user: { otp_attempt: user.current_otp } }, session: { attempt_user_id: user.id }
+            post :create, params: { user: { otp_attempt: user.current_otp } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
           end
 
           it 'shows a login error' do
@@ -248,7 +248,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
 
         context 'using a valid recovery code' do
           before do
-            post :create, params: { user: { otp_attempt: recovery_codes.first } }, session: { attempt_user_id: user.id }
+            post :create, params: { user: { otp_attempt: recovery_codes.first } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
           end
 
           it 'redirects to home' do
@@ -262,7 +262,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
 
         context 'using an invalid OTP' do
           before do
-            post :create, params: { user: { otp_attempt: 'wrongotp' } }, session: { attempt_user_id: user.id }
+            post :create, params: { user: { otp_attempt: 'wrongotp' } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
           end
 
           it 'shows a login error' do
@@ -334,7 +334,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
           before do
             @controller.session[:webauthn_challenge] = challenge
 
-            post :create, params: { user: { credential: fake_credential } }, session: { attempt_user_id: user.id }
+            post :create, params: { user: { credential: fake_credential } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
           end
 
           it 'instructs the browser to redirect to home' do
@@ -383,7 +383,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
       context 'using a valid sign in token' do
         before do
           user.generate_sign_in_token && user.save
-          post :create, params: { user: { sign_in_token_attempt: user.sign_in_token } }, session: { attempt_user_id: user.id }
+          post :create, params: { user: { sign_in_token_attempt: user.sign_in_token } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
         end
 
         it 'redirects to home' do
@@ -397,7 +397,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
 
       context 'using an invalid sign in token' do
         before do
-          post :create, params: { user: { sign_in_token_attempt: 'wrongotp' } }, session: { attempt_user_id: user.id }
+          post :create, params: { user: { sign_in_token_attempt: 'wrongotp' } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
         end
 
         it 'shows a login error' do
diff --git a/streaming/index.js b/streaming/index.js
index 03861e39a..0cddb36b4 100644
--- a/streaming/index.js
+++ b/streaming/index.js
@@ -294,7 +294,7 @@ const startWorker = (workerId) => {
         return;
       }
 
-      client.query('SELECT oauth_access_tokens.resource_owner_id, users.account_id, users.chosen_languages, oauth_access_tokens.scopes, devices.device_id FROM oauth_access_tokens INNER JOIN users ON oauth_access_tokens.resource_owner_id = users.id LEFT OUTER JOIN devices ON oauth_access_tokens.id = devices.access_token_id WHERE oauth_access_tokens.token = $1 AND oauth_access_tokens.revoked_at IS NULL LIMIT 1', [token], (err, result) => {
+      client.query('SELECT oauth_access_tokens.id, oauth_access_tokens.resource_owner_id, users.account_id, users.chosen_languages, oauth_access_tokens.scopes, devices.device_id FROM oauth_access_tokens INNER JOIN users ON oauth_access_tokens.resource_owner_id = users.id LEFT OUTER JOIN devices ON oauth_access_tokens.id = devices.access_token_id WHERE oauth_access_tokens.token = $1 AND oauth_access_tokens.revoked_at IS NULL LIMIT 1', [token], (err, result) => {
         done();
 
         if (err) {
@@ -310,6 +310,7 @@ const startWorker = (workerId) => {
           return;
         }
 
+        req.accessTokenId = result.rows[0].id;
         req.scopes = result.rows[0].scopes.split(' ');
         req.accountId = result.rows[0].account_id;
         req.chosenLanguages = result.rows[0].chosen_languages;
@@ -452,6 +453,55 @@ const startWorker = (workerId) => {
   };
 
   /**
+   * @typedef SystemMessageHandlers
+   * @property {function(): void} onKill
+   */
+
+  /**
+   * @param {any} req
+   * @param {SystemMessageHandlers} eventHandlers
+   * @return {function(string): void}
+   */
+  const createSystemMessageListener = (req, eventHandlers) => {
+    return message => {
+      const json = parseJSON(message);
+
+      if (!json) return;
+
+      const { event } = json;
+
+      log.silly(req.requestId, `System message for ${req.accountId}: ${event}`);
+
+      if (event === 'kill') {
+        log.verbose(req.requestId, `Closing connection for ${req.accountId} due to expired access token`);
+        eventHandlers.onKill();
+      }
+    }
+  };
+
+  /**
+   * @param {any} req
+   * @param {any} res
+   */
+  const subscribeHttpToSystemChannel = (req, res) => {
+    const systemChannelId = `timeline:access_token:${req.accessTokenId}`;
+
+    const listener = createSystemMessageListener(req, {
+
+      onKill () {
+        res.end();
+      },
+
+    });
+
+    res.on('close', () => {
+      unsubscribe(`${redisPrefix}${systemChannelId}`, listener);
+    });
+
+    subscribe(`${redisPrefix}${systemChannelId}`, listener);
+  };
+
+  /**
    * @param {any} req
    * @param {any} res
    * @param {function(Error=): void} next
@@ -463,6 +513,8 @@ const startWorker = (workerId) => {
     }
 
     accountFromRequest(req, alwaysRequireAuth).then(() => checkScopes(req, channelNameFromPath(req))).then(() => {
+      subscribeHttpToSystemChannel(req, res);
+    }).then(() => {
       next();
     }).catch(err => {
       next(err);
@@ -538,7 +590,9 @@ const startWorker = (workerId) => {
 
     const listener = message => {
       const json = parseJSON(message);
+
       if (!json) return;
+
       const { event, payload, queued_at } = json;
 
       const transmit = () => {
@@ -925,6 +979,28 @@ const startWorker = (workerId) => {
     });
 
   /**
+   * @param {WebSocketSession} session
+   */
+  const subscribeWebsocketToSystemChannel = ({ socket, request, subscriptions }) => {
+    const systemChannelId = `timeline:access_token:${request.accessTokenId}`;
+
+    const listener = createSystemMessageListener(request, {
+
+      onKill () {
+        socket.close();
+      },
+
+    });
+
+    subscribe(`${redisPrefix}${systemChannelId}`, listener);
+
+    subscriptions[systemChannelId] = {
+      listener,
+      stopHeartbeat: () => {},
+    };
+  };
+
+  /**
    * @param {string|string[]} arrayOrString
    * @return {string}
    */
@@ -970,7 +1046,9 @@ const startWorker = (workerId) => {
 
     ws.on('message', data => {
       const json = parseJSON(data);
+
       if (!json) return;
+
       const { type, stream, ...params } = json;
 
       if (type === 'subscribe') {
@@ -982,6 +1060,8 @@ const startWorker = (workerId) => {
       }
     });
 
+    subscribeWebsocketToSystemChannel(session);
+
     if (location.query.stream) {
       subscribeWebsocketToChannel(session, firstParam(location.query.stream), location.query);
     }