Merge pull request #1653 from ClearlyClaire/glitch-soc/merge-upstream

Merge upstream changes

Dockerfile

@@ -56,8 +56,8 @@ RUN npm install -g npm@latest && \
 COPY Gemfile* package.json yarn.lock /opt/mastodon/
 
 RUN cd /opt/mastodon && \
-  bundle config set deployment 'true' && \
+  bundle config set --local deployment 'true' && \
-  bundle config set without 'development test' && \
+  bundle config set --local without 'development test' && \
   bundle config set silence_root_warning true && \
 	bundle install -j"$(nproc)" && \
 	yarn install --pure-lockfile

Gemfile

@@ -18,7 +18,7 @@ gem 'makara', '~> 0.5'
 gem 'pghero', '~> 2.8'
 gem 'dotenv-rails', '~> 2.7'
 
-gem 'aws-sdk-s3', '~> 1.107', require: false
+gem 'aws-sdk-s3', '~> 1.109', require: false
 gem 'fog-core', '<= 2.1.0'
 gem 'fog-openstack', '~> 0.3', require: false
 gem 'kt-paperclip', '~> 7.0'
@@ -102,7 +102,7 @@ gem 'rdf-normalize', '~> 0.4'
 gem 'redcarpet', '~> 3.5'
 
 group :development, :test do
-  gem 'fabrication', '~> 2.22'
+  gem 'fabrication', '~> 2.23'
   gem 'fuubar', '~> 2.5'
   gem 'i18n-tasks', '~> 0.9', require: false
   gem 'pry-byebug', '~> 3.9'
@@ -131,13 +131,13 @@ group :development do
   gem 'annotate', '~> 3.1'
   gem 'better_errors', '~> 2.9'
   gem 'binding_of_caller', '~> 1.0'
-  gem 'bullet', '~> 6.1'
+  gem 'bullet', '~> 7.0'
   gem 'letter_opener', '~> 1.7'
   gem 'letter_opener_web', '~> 2.0'
   gem 'memory_profiler'
   gem 'rubocop', '~> 1.23', require: false
   gem 'rubocop-rails', '~> 2.12', require: false
-  gem 'brakeman', '~> 5.1', require: false
+  gem 'brakeman', '~> 5.2', require: false
   gem 'bundler-audit', '~> 0.9', require: false
 
   gem 'capistrano', '~> 3.16'

Gemfile.lock

@@ -79,16 +79,16 @@ GEM
       encryptor (~> 3.0.0)
     awrence (1.1.1)
     aws-eventstream (1.2.0)
-    aws-partitions (1.534.0)
+    aws-partitions (1.539.0)
-    aws-sdk-core (3.123.0)
+    aws-sdk-core (3.124.0)
       aws-eventstream (~> 1, >= 1.0.2)
       aws-partitions (~> 1, >= 1.525.0)
       aws-sigv4 (~> 1.1)
       jmespath (~> 1.0)
-    aws-sdk-kms (1.51.0)
+    aws-sdk-kms (1.52.0)
       aws-sdk-core (~> 3, >= 3.122.0)
       aws-sigv4 (~> 1.1)
-    aws-sdk-s3 (1.107.0)
+    aws-sdk-s3 (1.109.0)
       aws-sdk-core (~> 3, >= 3.122.0)
       aws-sdk-kms (~> 1)
       aws-sigv4 (~> 1.4)
@@ -106,13 +106,13 @@ GEM
       ffi (~> 1.14)
     bootsnap (1.9.3)
       msgpack (~> 1.0)
-    brakeman (5.1.2)
+    brakeman (5.2.0)
     browser (4.2.0)
     brpoplpush-redis_script (0.1.2)
       concurrent-ruby (~> 1.0, >= 1.0.5)
       redis (>= 1.0, <= 5.0)
     builder (3.2.4)
-    bullet (6.1.5)
+    bullet (7.0.0)
       activesupport (>= 3.0.0)
       uniform_notifier (~> 1.11)
     bundler-audit (0.9.0.1)
@@ -168,7 +168,7 @@ GEM
     css_parser (1.7.1)
       addressable
     debug_inspector (1.0.0)
-    devise (4.8.0)
+    devise (4.8.1)
       bcrypt (~> 3.0)
       orm_adapter (~> 0.1)
       railties (>= 4.1.0)
@@ -184,8 +184,8 @@ GEM
       devise (>= 4.0.0)
       rpam2 (~> 4.0)
     diff-lcs (1.4.4)
-    discard (1.2.0)
+    discard (1.2.1)
-      activerecord (>= 4.2, < 7)
+      activerecord (>= 4.2, < 8)
     docile (1.3.4)
     domain_name (0.5.20190701)
       unf (>= 0.0.5, < 1.0.0)
@@ -211,7 +211,7 @@ GEM
     et-orbi (1.2.4)
       tzinfo
     excon (0.76.0)
-    fabrication (2.22.0)
+    fabrication (2.23.1)
     faker (2.19.0)
       i18n (>= 1.6, < 2)
     faraday (1.8.0)
@@ -234,7 +234,7 @@ GEM
     faraday-patron (1.0.0)
     faraday-rack (1.0.0)
     fast_blank (1.0.1)
-    fastimage (2.2.5)
+    fastimage (2.2.6)
     ffi (1.15.4)
     ffi-compiler (1.0.1)
       ffi (>= 1.0.0)
@@ -355,7 +355,7 @@ GEM
       activesupport (>= 4)
       railties (>= 4)
       request_store (~> 1.0)
-    loofah (2.12.0)
+    loofah (2.13.0)
       crass (~> 1.0.2)
       nokogiri (>= 1.5.9)
     mail (2.7.1)
@@ -376,7 +376,7 @@ GEM
     mime-types-data (3.2021.1115)
     mini_mime (1.1.2)
     mini_portile2 (2.6.1)
-    minitest (5.14.4)
+    minitest (5.15.0)
     msgpack (1.4.2)
     multi_json (1.15.0)
     multipart-post (2.1.1)
@@ -393,7 +393,7 @@ GEM
       concurrent-ruby (~> 1.0, >= 1.0.2)
       sidekiq (>= 3.5)
       statsd-ruby (~> 1.4, >= 1.4.0)
-    oj (3.13.9)
+    oj (3.13.10)
     omniauth (1.9.1)
       hashie (>= 3.4.6)
       rack (>= 1.6.2, < 3)
@@ -418,7 +418,7 @@ GEM
     pastel (0.8.0)
       tty-color (~> 0.5)
     pg (1.2.3)
-    pghero (2.8.1)
+    pghero (2.8.2)
       activerecord (>= 5)
     pkg-config (1.4.6)
     posix-spawn (0.3.15)
@@ -580,7 +580,7 @@ GEM
       sidekiq (>= 3)
       thwait
       tilt (>= 1.4.0)
-    sidekiq-unique-jobs (7.1.8)
+    sidekiq-unique-jobs (7.1.12)
       brpoplpush-redis_script (> 0.1.1, <= 2.0.0)
       concurrent-ruby (~> 1.0, >= 1.0.5)
       sidekiq (>= 5.0, < 8.0)
@@ -599,7 +599,7 @@ GEM
     sprockets (3.7.2)
       concurrent-ruby (~> 1.0)
       rack (> 1, < 3)
-    sprockets-rails (3.4.1)
+    sprockets-rails (3.4.2)
       actionpack (>= 5.2)
       activesupport (>= 5.2)
       sprockets (>= 3.0.0)
@@ -609,7 +609,7 @@ GEM
     stackprof (0.2.17)
     statsd-ruby (1.5.0)
     stoplight (2.2.1)
-    strong_migrations (0.7.8)
+    strong_migrations (0.7.9)
       activerecord (>= 5)
     temple (0.8.2)
     terminal-table (3.0.2)
@@ -686,14 +686,14 @@ DEPENDENCIES
   active_record_query_trace (~> 1.8)
   addressable (~> 2.8)
   annotate (~> 3.1)
-  aws-sdk-s3 (~> 1.107)
+  aws-sdk-s3 (~> 1.109)
   better_errors (~> 2.9)
   binding_of_caller (~> 1.0)
   blurhash (~> 0.1)
   bootsnap (~> 1.9.2)
-  brakeman (~> 5.1)
+  brakeman (~> 5.2)
   browser
-  bullet (~> 6.1)
+  bullet (~> 7.0)
   bundler-audit (~> 0.9)
   capistrano (~> 3.16)
   capistrano-rails (~> 1.6)
@@ -714,7 +714,7 @@ DEPENDENCIES
   doorkeeper (~> 5.5)
   dotenv-rails (~> 2.7)
   ed25519 (~> 1.2)
-  fabrication (~> 2.22)
+  fabrication (~> 2.23)
   faker (~> 2.19)
   fast_blank (~> 1.0)
   fastimage

app/javascript/flavours/glitch/features/hashtag_timeline/components/column_settings.js

@@ -33,8 +33,8 @@ class ColumnSettings extends React.PureComponent {
   tags (mode) {
     let tags = this.props.settings.getIn(['tags', mode]) || [];
 
-    if (tags.toJSON) {
+    if (tags.toJS) {
-      return tags.toJSON();
+      return tags.toJS();
     } else {
       return tags;
     }

app/javascript/mastodon/features/hashtag_timeline/components/column_settings.js

@@ -33,8 +33,8 @@ class ColumnSettings extends React.PureComponent {
   tags (mode) {
     let tags = this.props.settings.getIn(['tags', mode]) || [];
 
-    if (tags.toJSON) {
+    if (tags.toJS) {
-      return tags.toJSON();
+      return tags.toJS();
     } else {
       return tags;
     }

app/models/user.rb

@@ -107,7 +107,7 @@ class User < ApplicationRecord
   scope :inactive, -> { where(arel_table[:current_sign_in_at].lt(ACTIVE_DURATION.ago)) }
   scope :active, -> { confirmed.where(arel_table[:current_sign_in_at].gteq(ACTIVE_DURATION.ago)).joins(:account).where(accounts: { suspended_at: nil }) }
   scope :matches_email, ->(value) { where(arel_table[:email].matches("#{value}%")) }
-  scope :matches_ip, ->(value) { left_joins(:session_activations).where('users.current_sign_in_ip <<= ?', value).or(left_joins(:session_activations).where('users.sign_up_ip <<= ?', value)).or(left_joins(:session_activations).where('users.last_sign_in_ip <<= ?', value)).or(left_joins(:session_activations).where('session_activations.ip <<= ?', value)) }
+  scope :matches_ip, ->(value) { where('current_sign_in_ip <<= ?', value).or(where('users.sign_up_ip <<= ?', value)).or(where('users.last_sign_in_ip <<= ?', value)).or(where(id: SessionActivation.select(:user_id).where('ip <<= ?', value))) }
   scope :emailable, -> { confirmed.enabled.joins(:account).merge(Account.searchable) }
 
   before_validation :sanitize_languages

config/environments/development.rb

@@ -81,7 +81,7 @@ Rails.application.configure do
     Bullet.bullet_logger = true
     Bullet.rails_logger  = false
 
-    Bullet.add_whitelist type: :n_plus_one_query, class_name: 'User', association: :account
+    Bullet.add_safelist type: :n_plus_one_query, class_name: 'User', association: :account
   end
 
   config.x.otp_secret = ENV.fetch('OTP_SECRET', '1fc2b87989afa6351912abeebe31ffc5c476ead9bf8b3d74cbc4a302c7b69a45b40b1bbef3506ddad73e942e15ed5ca4b402bf9a66423626051104f4b5f05109')

config/initializers/sidekiq.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require_relative '../../lib/sidekiq_error_handler'
+
 Sidekiq.configure_server do |config|
   config.redis = REDIS_SIDEKIQ_PARAMS
 

dist/mastodon-sidekiq.service

@@ -47,6 +47,7 @@ SystemCallFilter=~@cpu-emulation @debug @keyring @ipc @mount @obsolete @privileg
 SystemCallFilter=@chown
 SystemCallFilter=pipe
 SystemCallFilter=pipe2
+ReadWritePaths=/home/mastodon/live
 
 [Install]
 WantedBy=multi-user.target

dist/mastodon-streaming.service

@@ -45,6 +45,7 @@ SystemCallArchitectures=native
 SystemCallFilter=~@cpu-emulation @debug @keyring @ipc @memlock @mount @obsolete @privileged @resources @setuid
 SystemCallFilter=pipe
 SystemCallFilter=pipe2
+ReadWritePaths=/home/mastodon/live
 
 [Install]
 WantedBy=multi-user.target

dist/mastodon-web.service

@@ -47,6 +47,7 @@ SystemCallFilter=~@cpu-emulation @debug @keyring @ipc @mount @obsolete @privileg
 SystemCallFilter=@chown
 SystemCallFilter=pipe
 SystemCallFilter=pipe2
+ReadWritePaths=/home/mastodon/live
 
 [Install]
 WantedBy=multi-user.target

lib/sidekiq_error_handler.rb

@@ -17,10 +17,8 @@ class SidekiqErrorHandler
 
   private
 
-  # rubocop:disable Naming/MethodParameterName
+  def limit_backtrace_and_raise(exception)
-  def limit_backtrace_and_raise(e)
+    exception.set_backtrace(exception.backtrace.first(BACKTRACE_LIMIT))
-    e.set_backtrace(e.backtrace.first(BACKTRACE_LIMIT))
+    raise exception
-    raise e
   end
-  # rubocop:enable Naming/MethodParameterName
 end

package.json

@@ -61,13 +61,13 @@
   },
   "private": true,
   "dependencies": {
-    "@babel/core": "^7.16.0",
+    "@babel/core": "^7.16.5",
-    "@babel/plugin-proposal-decorators": "^7.16.4",
+    "@babel/plugin-proposal-decorators": "^7.16.5",
-    "@babel/plugin-transform-react-inline-elements": "^7.16.0",
+    "@babel/plugin-transform-react-inline-elements": "^7.16.5",
-    "@babel/plugin-transform-runtime": "^7.16.4",
+    "@babel/plugin-transform-runtime": "^7.16.5",
-    "@babel/preset-env": "^7.16.4",
+    "@babel/preset-env": "^7.16.5",
-    "@babel/preset-react": "^7.16.0",
+    "@babel/preset-react": "^7.16.5",
-    "@babel/runtime": "^7.16.3",
+    "@babel/runtime": "^7.16.5",
     "@gamestdio/websocket": "^0.3.2",
     "@github/webauthn-json": "^0.5.7",
     "@rails/ujs": "^6.1.4",
@@ -115,7 +115,7 @@
     "marky": "^1.2.2",
     "mini-css-extract-plugin": "^1.6.2",
     "mkdirp": "^1.0.4",
-    "npmlog": "^5.0.1",
+    "npmlog": "^6.0.0",
     "object-assign": "^4.1.1",
     "object-fit-images": "^3.2.3",
     "object.values": "^1.1.5",
@@ -146,7 +146,7 @@
     "react-swipeable-views": "^0.14.0",
     "react-textarea-autosize": "^8.3.3",
     "react-toggle": "^4.1.2",
-    "redis": "^3.1.2",
+    "redis": "^4.0.1",
     "redux": "^4.1.2",
     "redux-immutable": "^4.0.0",
     "redux-thunk": "^2.4.1",
@@ -155,7 +155,7 @@
     "requestidlecallback": "^0.3.0",
     "reselect": "^4.1.5",
     "rimraf": "^3.0.2",
-    "sass": "^1.43.5",
+    "sass": "^1.45.1",
     "sass-loader": "^10.2.0",
     "stacktrace-js": "^2.0.2",
     "stringz": "^2.1.0",
@@ -175,22 +175,22 @@
     "ws": "^8.3.0"
   },
   "devDependencies": {
-    "@testing-library/jest-dom": "^5.16.0",
+    "@testing-library/jest-dom": "^5.16.1",
     "@testing-library/react": "^12.1.2",
     "babel-eslint": "^10.1.0",
-    "babel-jest": "^27.4.0",
+    "babel-jest": "^27.4.5",
     "eslint": "^7.32.0",
     "eslint-plugin-import": "~2.25.3",
     "eslint-plugin-jsx-a11y": "~6.5.1",
-    "eslint-plugin-promise": "~5.1.1",
+    "eslint-plugin-promise": "~6.0.0",
     "eslint-plugin-react": "~7.27.1",
-    "jest": "^27.4.3",
+    "jest": "^27.4.5",
     "raf": "^3.4.1",
     "react-intl-translations-manager": "^5.0.3",
     "react-test-renderer": "^16.14.0",
     "sass-lint": "^1.13.1",
     "webpack-dev-server": "^3.11.3",
-    "yargs": "^17.2.1"
+    "yargs": "^17.3.0"
   },
   "resolutions": {
     "kind-of": "^6.0.3"

streaming/index.js

@@ -63,20 +63,29 @@ const dbUrlToConfig = (dbUrl) => {
  * @param {Object.<string, any>} defaultConfig
  * @param {string} redisUrl
  */
-const redisUrlToClient = (defaultConfig, redisUrl) => {
+const redisUrlToClient = async (defaultConfig, redisUrl) => {
   const config = defaultConfig;
 
+  let client;
+
   if (!redisUrl) {
-    return redis.createClient(config);
+    client = redis.createClient(config);
+  } else if (redisUrl.startsWith('unix://')) {
+    client = redis.createClient(Object.assign(config, {
+      socket: {
+        path: redisUrl.slice(7),
+      },
+    }));
+  } else {
+    client = redis.createClient(Object.assign(config, {
+      url: redisUrl,
+    }));
   }
 
-  if (redisUrl.startsWith('unix://')) {
+  client.on('error', (err) => log.error('Redis Client Error!', err));
-    return redis.createClient(redisUrl.slice(7), config);
+  await client.connect();
-  }
 
-  return redis.createClient(Object.assign(config, {
+  return client;
-    url: redisUrl,
-  }));
 };
 
 const numWorkers = +process.env.STREAMING_CLUSTER_NUM || (env === 'development' ? 1 : Math.max(os.cpus().length - 1, 1));
@@ -102,7 +111,7 @@ const startMaster = () => {
   log.warn(`Starting streaming API server master with ${numWorkers} workers`);
 };
 
-const startWorker = (workerId) => {
+const startWorker = async (workerId) => {
   log.warn(`Starting worker ${workerId}`);
 
   const pgConfigs = {
@@ -127,7 +136,7 @@ const startWorker = (workerId) => {
 
   if (!!process.env.DB_SSLMODE && process.env.DB_SSLMODE !== 'disable') {
     pgConfigs.development.ssl = true;
-    pgConfigs.production.ssl  = true;
+    pgConfigs.production.ssl = true;
   }
 
   const app = express();
@@ -139,9 +148,11 @@ const startWorker = (workerId) => {
   const redisNamespace = process.env.REDIS_NAMESPACE || null;
 
   const redisParams = {
-    host:     process.env.REDIS_HOST     || '127.0.0.1',
+    socket: {
-    port:     process.env.REDIS_PORT     || 6379,
+      host: process.env.REDIS_HOST || '127.0.0.1',
-    db:       process.env.REDIS_DB       || 0,
+      port: process.env.REDIS_PORT || 6379,
+    },
+    database: process.env.REDIS_DB || 0,
     password: process.env.REDIS_PASSWORD || undefined,
   };
 
@@ -151,25 +162,8 @@ const startWorker = (workerId) => {
 
   const redisPrefix = redisNamespace ? `${redisNamespace}:` : '';
 
-  const redisSubscribeClient = redisUrlToClient(redisParams, process.env.REDIS_URL);
+  const redisSubscribeClient = await redisUrlToClient(redisParams, process.env.REDIS_URL);
-  const redisClient = redisUrlToClient(redisParams, process.env.REDIS_URL);
+  const redisClient = await redisUrlToClient(redisParams, process.env.REDIS_URL);
-
-  /**
-   * @type {Object.<string, Array.<function(string): void>>}
-   */
-  const subs = {};
-
-  redisSubscribeClient.on('message', (channel, message) => {
-    const callbacks = subs[channel];
-
-    log.silly(`New message on channel ${channel}`);
-
-    if (!callbacks) {
-      return;
-    }
-
-    callbacks.forEach(callback => callback(message));
-  });
 
   /**
    * @param {string[]} channels
@@ -197,34 +191,16 @@ const startWorker = (workerId) => {
    */
   const subscribe = (channel, callback) => {
     log.silly(`Adding listener for ${channel}`);
-    subs[channel] = subs[channel] || [];
 
-    if (subs[channel].length === 0) {
+    redisSubscribeClient.subscribe(channel, callback);
-      log.verbose(`Subscribe ${channel}`);
-      redisSubscribeClient.subscribe(channel);
-    }
-
-    subs[channel].push(callback);
   };
 
   /**
    * @param {string} channel
-   * @param {function(string): void} callback
    */
-  const unsubscribe = (channel, callback) => {
+  const unsubscribe = (channel) => {
-    log.silly(`Removing listener for ${channel}`);
 
-    if (!subs[channel]) {
+    redisSubscribeClient.unsubscribe(channel);
-      return;
-    }
-
-    subs[channel] = subs[channel].filter(item => item !== callback);
-
-    if (subs[channel].length === 0) {
-      log.verbose(`Unsubscribe ${channel}`);
-      redisSubscribeClient.unsubscribe(channel);
-      delete subs[channel];
-    }
   };
 
   const FALSE_VALUES = [
@@ -366,7 +342,7 @@ const startWorker = (workerId) => {
     const onlyMedia = isTruthy(query.only_media);
     const allowLocalOnly = isTruthy(query.allow_local_only);
 
-    switch(path) {
+    switch (path) {
     case '/api/v1/streaming/user':
       return 'user';
     case '/api/v1/streaming/user/notification':
@@ -497,7 +473,7 @@ const startWorker = (workerId) => {
 
     const listener = createSystemMessageListener(req, {
 
-      onKill () {
+      onKill() {
         res.end();
       },
 
@@ -549,7 +525,7 @@ const startWorker = (workerId) => {
   };
 
   /**
-   * @param {array}
+   * @param {array} arr
    * @param {number=} shift
    * @return {string}
    */
@@ -592,7 +568,7 @@ const startWorker = (workerId) => {
    * @return {function(string): void}
    */
   const streamFrom = (ids, req, output, attachCloseHandler, needsFiltering = false, allowLocalOnly = false) => {
-    const accountId  = req.accountId || req.remoteAddress;
+    const accountId = req.accountId || req.remoteAddress;
 
     log.verbose(req.requestId, `Starting stream from ${ids.join(', ')} for ${accountId}`);
 
@@ -604,8 +580,8 @@ const startWorker = (workerId) => {
       const { event, payload, queued_at } = json;
 
       const transmit = () => {
-        const now            = new Date().getTime();
+        const now = new Date().getTime();
-        const delta          = now - queued_at;
+        const delta = now - queued_at;
         const encodedPayload = typeof payload === 'object' ? JSON.stringify(payload) : payload;
 
         log.silly(req.requestId, `Transmitting for ${accountId}: ${event} ${encodedPayload} Delay: ${delta}ms`);
@@ -625,9 +601,9 @@ const startWorker = (workerId) => {
         return;
       }
 
-      const unpackedPayload  = payload;
+      const unpackedPayload = payload;
       const targetAccountIds = [unpackedPayload.account.id].concat(unpackedPayload.mentions.map(item => item.id));
-      const accountDomain    = unpackedPayload.account.acct.split('@')[1];
+      const accountDomain = unpackedPayload.account.acct.split('@')[1];
 
       if (Array.isArray(req.chosenLanguages) && unpackedPayload.language !== null && req.chosenLanguages.indexOf(unpackedPayload.language) === -1) {
         log.silly(req.requestId, `Message ${unpackedPayload.id} filtered by language (${unpackedPayload.language})`);
@@ -647,7 +623,15 @@ const startWorker = (workerId) => {
         }
 
         const queries = [
-          client.query(`SELECT 1 FROM blocks WHERE (account_id = $1 AND target_account_id IN (${placeholders(targetAccountIds, 2)})) OR (account_id = $2 AND target_account_id = $1) UNION SELECT 1 FROM mutes WHERE account_id = $1 AND target_account_id IN (${placeholders(targetAccountIds, 2)})`, [req.accountId, unpackedPayload.account.id].concat(targetAccountIds)),
+          client.query(`SELECT 1
+                        FROM blocks
+                        WHERE (account_id = $1 AND target_account_id IN (${placeholders(targetAccountIds, 2)}))
+                           OR (account_id = $2 AND target_account_id = $1)
+                        UNION
+                        SELECT 1
+                        FROM mutes
+                        WHERE account_id = $1
+                          AND target_account_id IN (${placeholders(targetAccountIds, 2)})`, [req.accountId, unpackedPayload.account.id].concat(targetAccountIds)),
         ];
 
         if (accountDomain) {
@@ -710,12 +694,12 @@ const startWorker = (workerId) => {
   /**
    * @param {any} req
    * @param {function(): void} [closeHandler]
-   * @return {function(string[], function(string): void)}
+   * @return {function(string[]): void}
    */
-  const streamHttpEnd = (req, closeHandler = undefined) => (ids, listener) => {
+  const streamHttpEnd = (req, closeHandler = undefined) => (ids) => {
     req.on('close', () => {
       ids.forEach(id => {
-        unsubscribe(id, listener);
+        unsubscribe(id);
       });
 
       if (closeHandler) {
@@ -762,7 +746,7 @@ const startWorker = (workerId) => {
   app.get('/api/v1/streaming/*', (req, res) => {
     channelNameToIds(req, channelNameFromPath(req), req.query).then(({ channelIds, options }) => {
       const onSend = streamToHttp(req, res);
-      const onEnd  = streamHttpEnd(req, subscriptionHeartbeat(channelIds));
+      const onEnd = streamHttpEnd(req, subscriptionHeartbeat(channelIds));
 
       streamFrom(channelIds, req, onSend, onEnd, options.needsFiltering, options.allowLocalOnly);
     }).catch(err => {
@@ -805,7 +789,7 @@ const startWorker = (workerId) => {
    * @return {Promise.<{ channelIds: string[], options: { needsFiltering: boolean } }>}
    */
   const channelNameToIds = (req, name, params) => new Promise((resolve, reject) => {
-    switch(name) {
+    switch (name) {
     case 'user':
       resolve({
         channelIds: channelsForUserStream(req),
@@ -949,15 +933,17 @@ const startWorker = (workerId) => {
    * @param {StreamParams} params
    */
   const subscribeWebsocketToChannel = ({ socket, request, subscriptions }, channelName, params) =>
-    checkScopes(request, channelName).then(() => channelNameToIds(request, channelName, params)).then(({ channelIds, options }) => {
+    checkScopes(request, channelName).then(() => channelNameToIds(request, channelName, params)).then(({
+      channelIds,
+      options,
+    }) => {
       if (subscriptions[channelIds.join(';')]) {
         return;
       }
 
-      const onSend        = streamToWs(request, socket, streamNameFromChannelName(channelName, params));
+      const onSend = streamToWs(request, socket, streamNameFromChannelName(channelName, params));
       const stopHeartbeat = subscriptionHeartbeat(channelIds);
-
+      const listener = streamFrom(channelIds, request, onSend, undefined, options.needsFiltering, options.allowLocalOnly);
-      const listener      = streamFrom(channelIds, request, onSend, undefined, options.needsFiltering, options.allowLocalOnly);
 
       subscriptions[channelIds.join(';')] = {
         listener,
@@ -1005,7 +991,7 @@ const startWorker = (workerId) => {
 
     const listener = createSystemMessageListener(request, {
 
-      onKill () {
+      onKill() {
         socket.close();
       },
 
@@ -1015,7 +1001,8 @@ const startWorker = (workerId) => {
 
     subscriptions[systemChannelId] = {
       listener,
-      stopHeartbeat: () => {},
+      stopHeartbeat: () => {
+      },
     };
   };
 
@@ -1034,7 +1021,7 @@ const startWorker = (workerId) => {
   wss.on('connection', (ws, req) => {
     const location = url.parse(req.url, true);
 
-    req.requestId     = uuid.v4();
+    req.requestId = uuid.v4();
     req.remoteAddress = ws._socket.remoteAddress;
 
     ws.isAlive = true;