Merge commit from fork

* Streaming: Ensure disabled users cannot connect to streaming

* Streaming: Disconnect when the user is disabled

---------

Co-authored-by: Emelia Smith <ThisIsMissEm@users.noreply.github.com>

app/models/user.rb

@@ -197,6 +197,10 @@ class User < ApplicationRecord
 
   def disable!
     update!(disabled: true)
+
+    # This terminates all connections for the given account with the streaming
+    # server:
+    redis.publish("timeline:system:#{account.id}", Oj.dump(event: :kill))
   end
 
   def enable!

spec/models/user_spec.rb

@@ -476,12 +476,15 @@ RSpec.describe User do
 
     let(:current_sign_in_at) { Time.zone.now }
 
-    before do
-      user.disable!
-    end
-
     it 'disables user' do
+      allow(redis).to receive(:publish)
+
+      user.disable!
+
       expect(user).to have_attributes(disabled: true)
+
+      expect(redis)
+        .to have_received(:publish).with("timeline:system:#{user.account.id}", Oj.dump(event: :kill)).once
     end
   end
 

spec/system/streaming/streaming_spec.rb

@@ -74,4 +74,28 @@ RSpec.describe 'Streaming', :inline_jobs, :streaming do
       expect(streaming_client.open?).to be(false)
     end
   end
+
+  context 'with a disabled user account' do
+    before do
+      user.disable!
+    end
+
+    it 'receives an 401 unauthorized error when trying to connect' do
+      streaming_client.connect
+
+      expect(streaming_client.status).to eq(401)
+      expect(streaming_client.open?).to be(false)
+    end
+  end
+
+  context 'when the user account is disabled whilst connected' do
+    it 'terminates the connection for the user' do
+      streaming_client.connect
+
+      user.disable!
+
+      expect(streaming_client.wait_for(:closed).code).to be(1000)
+      expect(streaming_client.open?).to be(false)
+    end
+  end
 end

streaming/index.js

@@ -355,7 +355,7 @@ const startServer = async () => {
    * @returns {Promise<ResolvedAccount>}
    */
   const accountFromToken = async (token, req) => {
-    const result = await pgPool.query('SELECT oauth_access_tokens.id, oauth_access_tokens.resource_owner_id, users.account_id, users.chosen_languages, oauth_access_tokens.scopes FROM oauth_access_tokens INNER JOIN users ON oauth_access_tokens.resource_owner_id = users.id WHERE oauth_access_tokens.token = $1 AND oauth_access_tokens.revoked_at IS NULL LIMIT 1', [token]);
+    const result = await pgPool.query('SELECT oauth_access_tokens.id, oauth_access_tokens.resource_owner_id, users.account_id, users.chosen_languages, oauth_access_tokens.scopes FROM oauth_access_tokens INNER JOIN users ON oauth_access_tokens.resource_owner_id = users.id WHERE oauth_access_tokens.token = $1 AND oauth_access_tokens.revoked_at IS NULL AND users.disabled IS FALSE LIMIT 1', [token]);
 
     if (result.rows.length === 0) {
       throw new AuthenticationError('Invalid access token');