me
/
glitchier-soc
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'master' into glitch-soc/master

Browse Source 
Conflicts:
- `streaming/index.js`:
  Upstream entirely refactored it.
  Ported our changes to upstream's refactor. Hopefuly.
main
Thibaut Girka 4 years ago
parent 03712799e2 ef057584fd
commit 78fa15d08f
6 changed files with 816 additions and 320 deletions
Show Stats Download Patch File Download Diff File

99
app/javascript/mastodon/actions/streaming.js
Unescape View File

@ -1,3 +1,5 @@
// @ts-check
import { connectStream } from '../stream'; import { connectStream } from '../stream';
import { import {
updateTimeline, updateTimeline,
@ -19,24 +21,59 @@ import { getLocale } from '../locales';
const { messages } = getLocale(); const { messages } = getLocale();
export function connectTimelineStream (timelineId, path, pollingRefresh = null, accept = null) { /**
* @param {number} max
* @return {number}
*/
const randomUpTo = max =>
Math.floor(Math.random() * Math.floor(max));
return connectStream (path, pollingRefresh, (dispatch, getState) => { /**
* @param {string} timelineId
* @param {string} channelName
* @param {Object.<string, string>} params
* @param {Object} options
* @param {function(Function, Function): void} [options.fallback]
* @param {function(object): boolean} [options.accept]
* @return {function(): void}
*/
export const connectTimelineStream = (timelineId, channelName, params = {}, options = {}) =>
connectStream(channelName, params, (dispatch, getState) => {
const locale = getState().getIn(['meta', 'locale']); const locale = getState().getIn(['meta', 'locale']);
let pollingId;
/**
* @param {function(Function, Function): void} fallback
*/
const useFallback = fallback => {
fallback(dispatch, () => {
pollingId = setTimeout(() => useFallback(fallback), 20000 + randomUpTo(20000));
});
};
return { return {
onConnect() { onConnect() {
dispatch(connectTimeline(timelineId)); dispatch(connectTimeline(timelineId));
if (pollingId) {
clearTimeout(pollingId);
pollingId = null;
}
}, },
onDisconnect() { onDisconnect() {
dispatch(disconnectTimeline(timelineId)); dispatch(disconnectTimeline(timelineId));
if (options.fallback) {
pollingId = setTimeout(() => useFallback(options.fallback), randomUpTo(40000));
}
}, },
onReceive (data) { onReceive (data) {
switch(data.event) { switch(data.event) {
case 'update': case 'update':
dispatch(updateTimeline(timelineId, JSON.parse(data.payload), accept)); dispatch(updateTimeline(timelineId, JSON.parse(data.payload), options.accept));
break; break;
case 'delete': case 'delete':
dispatch(deleteFromTimelines(data.payload)); dispatch(deleteFromTimelines(data.payload));
@ -63,17 +100,59 @@ export function connectTimelineStream (timelineId, path, pollingRefresh = null,
}, },
}; };
}); });
}
/**
* @param {Function} dispatch
* @param {function(): void} done
*/
const refreshHomeTimelineAndNotification = (dispatch, done) => { const refreshHomeTimelineAndNotification = (dispatch, done) => {
dispatch(expandHomeTimeline({}, () => dispatch(expandHomeTimeline({}, () =>
dispatch(expandNotifications({}, () => dispatch(expandNotifications({}, () =>
dispatch(fetchAnnouncements(done)))))); dispatch(fetchAnnouncements(done))))));
}; };
export const connectUserStream = () => connectTimelineStream('home', 'user', refreshHomeTimelineAndNotification); /**
export const connectCommunityStream = ({ onlyMedia } = {}) => connectTimelineStream(`community${onlyMedia ? ':media' : ''}`, `public:local${onlyMedia ? ':media' : ''}`); * @return {function(): void}
export const connectPublicStream = ({ onlyMedia, onlyRemote } = {}) => connectTimelineStream(`public${onlyRemote ? ':remote' : ''}${onlyMedia ? ':media' : ''}`, `public${onlyRemote ? ':remote' : ''}${onlyMedia ? ':media' : ''}`); */
export const connectHashtagStream = (id, tag, local, accept) => connectTimelineStream(`hashtag:${id}${local ? ':local' : ''}`, `hashtag${local ? ':local' : ''}&tag=${tag}`, null, accept); export const connectUserStream = () =>
export const connectDirectStream = () => connectTimelineStream('direct', 'direct'); connectTimelineStream('home', 'user', {}, { fallback: refreshHomeTimelineAndNotification });
export const connectListStream = id => connectTimelineStream(`list:${id}`, `list&list=${id}`);
/**
* @param {Object} options
* @param {boolean} [options.onlyMedia]
* @return {function(): void}
*/
export const connectCommunityStream = ({ onlyMedia } = {}) =>
connectTimelineStream(`community${onlyMedia ? ':media' : ''}`, `public:local${onlyMedia ? ':media' : ''}`);
/**
* @param {Object} options
* @param {boolean} [options.onlyMedia]
* @param {boolean} [options.onlyRemote]
* @return {function(): void}
*/
export const connectPublicStream = ({ onlyMedia, onlyRemote } = {}) =>
connectTimelineStream(`public${onlyRemote ? ':remote' : ''}${onlyMedia ? ':media' : ''}`, `public${onlyRemote ? ':remote' : ''}${onlyMedia ? ':media' : ''}`);
/**
* @param {string} columnId
* @param {string} tagName
* @param {boolean} onlyLocal
* @param {function(object): boolean} accept
* @return {function(): void}
*/
export const connectHashtagStream = (columnId, tagName, onlyLocal, accept) =>
connectTimelineStream(`hashtag:${columnId}${onlyLocal ? ':local' : ''}`, `hashtag${onlyLocal ? ':local' : ''}`, { tag: tagName }, { accept });
/**
* @return {function(): void}
*/
export const connectDirectStream = () =>
connectTimelineStream('direct', 'direct');
/**
* @param {string} listId
* @return {function(): void}
*/
export const connectListStream = listId =>
connectTimelineStream(`list:${listId}`, 'list', { list: listId });

266
app/javascript/mastodon/stream.js
Unescape View File

@ -1,87 +1,236 @@
// @ts-check
import WebSocketClient from '@gamestdio/websocket'; import WebSocketClient from '@gamestdio/websocket';
const randomIntUpTo = max => Math.floor(Math.random() * Math.floor(max)); /**
* @type {WebSocketClient | undefined}
*/
let sharedConnection;
const knownEventTypes = [ /**
'update', * @typedef Subscription
'delete', * @property {string} channelName
'notification', * @property {Object.<string, string>} params
'conversation', * @property {function(): void} onConnect
'filters_changed', * @property {function(StreamEvent): void} onReceive
]; * @property {function(): void} onDisconnect
*/
export function connectStream(path, pollingRefresh = null, callbacks = () => ({ onConnect() {}, onDisconnect() {}, onReceive() {} })) { /**
return (dispatch, getState) => { * @typedef StreamEvent
const streamingAPIBaseURL = getState().getIn(['meta', 'streaming_api_base_url']); * @property {string} event
const accessToken = getState().getIn(['meta', 'access_token']); * @property {object} payload
const { onConnect, onDisconnect, onReceive } = callbacks(dispatch, getState); */
let polling = null; /**
* @type {Array.<Subscription>}
*/
const subscriptions = [];
const setupPolling = () => { /**
pollingRefresh(dispatch, () => { * @type {Object.<string, number>}
polling = setTimeout(() => setupPolling(), 20000 + randomIntUpTo(20000)); */
}); const subscriptionCounters = {};
};
const clearPolling = () => { /**
if (polling) { * @param {Subscription} subscription
clearTimeout(polling); */
polling = null; const addSubscription = subscription => {
subscriptions.push(subscription);
};
/**
* @param {Subscription} subscription
*/
const removeSubscription = subscription => {
const index = subscriptions.indexOf(subscription);
if (index !== -1) {
subscriptions.splice(index, 1);
} }
}; };
const subscription = getStream(streamingAPIBaseURL, accessToken, path, { /**
connected () { * @param {Subscription} subscription
if (pollingRefresh) { */
clearPolling(); const subscribe = ({ channelName, params, onConnect }) => {
const key = channelNameWithInlineParams(channelName, params);
subscriptionCounters[key] = subscriptionCounters[key] || 0;
if (subscriptionCounters[key] === 0) {
sharedConnection.send(JSON.stringify({ type: 'subscribe', stream: channelName, ...params }));
} }
subscriptionCounters[key] += 1;
onConnect(); onConnect();
}, };
disconnected () { /**
if (pollingRefresh) { * @param {Subscription} subscription
polling = setTimeout(() => setupPolling(), randomIntUpTo(40000)); */
const unsubscribe = ({ channelName, params, onDisconnect }) => {
const key = channelNameWithInlineParams(channelName, params);
subscriptionCounters[key] = subscriptionCounters[key] || 1;
if (subscriptionCounters[key] === 1 && sharedConnection.readyState === WebSocketClient.OPEN) {
sharedConnection.send(JSON.stringify({ type: 'unsubscribe', stream: channelName, ...params }));
} }
subscriptionCounters[key] -= 1;
onDisconnect(); onDisconnect();
};
const sharedCallbacks = {
connected () {
subscriptions.forEach(subscription => subscribe(subscription));
}, },
received (data) { received (data) {
onReceive(data); const { stream } = data;
subscriptions.filter(({ channelName, params }) => {
const streamChannelName = stream[0];
if (stream.length === 1) {
return channelName === streamChannelName;
}
const streamIdentifier = stream[1];
if (['hashtag', 'hashtag:local'].includes(channelName)) {
return channelName === streamChannelName && params.tag === streamIdentifier;
} else if (channelName === 'list') {
return channelName === streamChannelName && params.list === streamIdentifier;
}
return false;
}).forEach(subscription => {
subscription.onReceive(data);
});
},
disconnected () {
subscriptions.forEach(({ onDisconnect }) => onDisconnect());
}, },
reconnected () { reconnected () {
if (pollingRefresh) { subscriptions.forEach(subscription => subscribe(subscription));
clearPolling(); },
pollingRefresh(dispatch); };
/**
* @param {string} channelName
* @param {Object.<string, string>} params
* @return {string}
*/
const channelNameWithInlineParams = (channelName, params) => {
if (Object.keys(params).length === 0) {
return channelName;
} }
return `${channelName}&${Object.keys(params).map(key => `${key}=${params[key]}`).join('&')}`;
};
/**
* @param {string} channelName
* @param {Object.<string, string>} params
* @param {function(Function, Function): { onConnect: (function(): void), onReceive: (function(StreamEvent): void), onDisconnect: (function(): void) }} callbacks
* @return {function(): void}
*/
export const connectStream = (channelName, params, callbacks) => (dispatch, getState) => {
const streamingAPIBaseURL = getState().getIn(['meta', 'streaming_api_base_url']);
const accessToken = getState().getIn(['meta', 'access_token']);
const { onConnect, onReceive, onDisconnect } = callbacks(dispatch, getState);
// If we cannot use a websockets connection, we must fall back
// to using individual connections for each channel
if (!streamingAPIBaseURL.startsWith('ws')) {
const connection = createConnection(streamingAPIBaseURL, accessToken, channelNameWithInlineParams(channelName, params), {
connected () {
onConnect(); onConnect();
}, },
received (data) {
onReceive(data);
},
disconnected () {
onDisconnect();
},
reconnected () {
onConnect();
},
}); });
const disconnect = () => { return () => {
if (subscription) { connection.close();
subscription.close(); };
} }
clearPolling(); const subscription = {
channelName,
params,
onConnect,
onReceive,
onDisconnect,
}; };
return disconnect; addSubscription(subscription);
// If a connection is open, we can execute the subscription right now. Otherwise,
// because we have already registered it, it will be executed on connect
if (!sharedConnection) {
sharedConnection = /** @type {WebSocketClient} */ (createConnection(streamingAPIBaseURL, accessToken, '', sharedCallbacks));
} else if (sharedConnection.readyState === WebSocketClient.OPEN) {
subscribe(subscription);
}
return () => {
removeSubscription(subscription);
unsubscribe(subscription);
}; };
} };
const KNOWN_EVENT_TYPES = [
'update',
'delete',
'notification',
'conversation',
'filters_changed',
'encrypted_message',
'announcement',
'announcement.delete',
'announcement.reaction',
];
export default function getStream(streamingAPIBaseURL, accessToken, stream, { connected, received, disconnected, reconnected }) { /**
const params = stream.split('&'); * @param {MessageEvent} e
stream = params.shift(); * @param {function(StreamEvent): void} received
*/
const handleEventSourceMessage = (e, received) => {
received({
event: e.type,
payload: e.data,
});
};
/**
* @param {string} streamingAPIBaseURL
* @param {string} accessToken
* @param {string} channelName
* @param {{ connected: Function, received: function(StreamEvent): void, disconnected: Function, reconnected: Function }} callbacks
* @return {WebSocketClient | EventSource}
*/
const createConnection = (streamingAPIBaseURL, accessToken, channelName, { connected, received, disconnected, reconnected }) => {
const params = channelName.split('&');
channelName = params.shift();
if (streamingAPIBaseURL.startsWith('ws')) { if (streamingAPIBaseURL.startsWith('ws')) {
params.unshift(`stream=${stream}`);
const ws = new WebSocketClient(`${streamingAPIBaseURL}/api/v1/streaming/?${params.join('&')}`, accessToken); const ws = new WebSocketClient(`${streamingAPIBaseURL}/api/v1/streaming/?${params.join('&')}`, accessToken);
ws.onopen = connected; ws.onopen = connected;
@ -92,11 +241,19 @@ export default function getStream(streamingAPIBaseURL, accessToken, stream, { co
return ws; return ws;
} }
stream = stream.replace(/:/g, '/'); channelName = channelName.replace(/:/g, '/');
if (channelName.endsWith(':media')) {
channelName = channelName.replace('/media', '');
params.push('only_media=true');
}
params.push(`access_token=${accessToken}`); params.push(`access_token=${accessToken}`);
const es = new EventSource(`${streamingAPIBaseURL}/api/v1/streaming/${stream}?${params.join('&')}`);
const es = new EventSource(`${streamingAPIBaseURL}/api/v1/streaming/${channelName}?${params.join('&')}`);
let firstConnect = true; let firstConnect = true;
es.onopen = () => { es.onopen = () => {
if (firstConnect) { if (firstConnect) {
firstConnect = false; firstConnect = false;
@ -105,15 +262,12 @@ export default function getStream(streamingAPIBaseURL, accessToken, stream, { co
reconnected(); reconnected();
} }
}; };
for (let type of knownEventTypes) {
es.addEventListener(type, (e) => { KNOWN_EVENT_TYPES.forEach(type => {
received({ es.addEventListener(type, e => handleEventSourceMessage(/** @type {MessageEvent} */ (e), received));
event: e.type,
payload: e.data,
});
}); });
}
es.onerror = disconnected; es.onerror = /** @type {function(): void} */ (disconnected);
return es; return es;
}; };

2
app/models/form/custom_emoji_batch.rb
Unescape View File

@ -30,7 +30,7 @@ class Form::CustomEmojiBatch
private private
def custom_emojis def custom_emojis
CustomEmoji.where(id: custom_emoji_ids) @custom_emojis ||= CustomEmoji.where(id: custom_emoji_ids)
end end
def update! def update!

6
config/routes.rb
Unescape View File

@ -173,11 +173,7 @@ Rails.application.routes.draw do
get '/dashboard', to: 'dashboard#index' get '/dashboard', to: 'dashboard#index'
resources :domain_allows, only: [:new, :create, :show, :destroy] resources :domain_allows, only: [:new, :create, :show, :destroy]
resources :domain_blocks, only: [:new, :create, :show, :destroy, :update] do resources :domain_blocks, only: [:new, :create, :show, :destroy, :update, :edit]
member do
get :edit
end
end
resources :email_domain_blocks, only: [:index, :new, :create, :destroy] resources :email_domain_blocks, only: [:index, :new, :create, :destroy]
resources :action_logs, only: [:index] resources :action_logs, only: [:index]

8
lib/mastodon/media_cli.rb
Unescape View File

@ -89,7 +89,7 @@ module Mastodon
path_segments = object.key.split('/') path_segments = object.key.split('/')
path_segments.delete('cache') path_segments.delete('cache')
if path_segments.size != 7 unless [7, 10].include?(path_segments.size)
progress.log(pastel.yellow("Unrecognized file found: #{object.key}")) progress.log(pastel.yellow("Unrecognized file found: #{object.key}"))
next next
end end
@ -133,7 +133,7 @@ module Mastodon
path_segments = key.split(File::SEPARATOR) path_segments = key.split(File::SEPARATOR)
path_segments.delete('cache') path_segments.delete('cache')
if path_segments.size != 7 unless [7, 10].include?(path_segments.size)
progress.log(pastel.yellow("Unrecognized file found: #{key}")) progress.log(pastel.yellow("Unrecognized file found: #{key}"))
next next
end end
@ -258,7 +258,7 @@ module Mastodon
path_segments = path.split('/')[2..-1] path_segments = path.split('/')[2..-1]
path_segments.delete('cache') path_segments.delete('cache')
if path_segments.size != 7 unless [7, 10].include?(path_segments.size)
say('Not a media URL', :red) say('Not a media URL', :red)
exit(1) exit(1)
end end
@ -311,7 +311,7 @@ module Mastodon
segments = object.key.split('/') segments = object.key.split('/')
segments.delete('cache') segments.delete('cache')
next if segments.size != 7 next unless [7, 10].include?(segments.size)
model_name = segments.first.classify model_name = segments.first.classify
record_id = segments[2..-2].join.to_i record_id = segments[2..-2].join.to_i

709
streaming/index.js
Unescape View File

@ -1,3 +1,5 @@
// @ts-check
const os = require('os'); const os = require('os');
const throng = require('throng'); const throng = require('throng');
const dotenv = require('dotenv'); const dotenv = require('dotenv');
@ -12,7 +14,7 @@ const uuid = require('uuid');
const fs = require('fs'); const fs = require('fs');
const env = process.env.NODE_ENV || 'development'; const env = process.env.NODE_ENV || 'development';
const alwaysRequireAuth = process.env.WHITELIST_MODE === 'true' || process.env.AUTHORIZED_FETCH === 'true'; const alwaysRequireAuth = process.env.LIMITED_FEDERATION_MODE === 'true' || process.env.WHITELIST_MODE === 'true' || process.env.AUTHORIZED_FETCH === 'true';
dotenv.config({ dotenv.config({
path: env === 'production' ? '.env.production' : '.env', path: env === 'production' ? '.env.production' : '.env',
@ -20,6 +22,10 @@ dotenv.config({
log.level = process.env.LOG_LEVEL || 'verbose'; log.level = process.env.LOG_LEVEL || 'verbose';
/**
* @param {string} dbUrl
* @return {Object.<string, any>}
*/
const dbUrlToConfig = (dbUrl) => { const dbUrlToConfig = (dbUrl) => {
if (!dbUrl) { if (!dbUrl) {
return {}; return {};
@ -53,6 +59,10 @@ const dbUrlToConfig = (dbUrl) => {
return config; return config;
}; };
/**
* @param {Object.<string, any>} defaultConfig
* @param {string} redisUrl
*/
const redisUrlToClient = (defaultConfig, redisUrl) => { const redisUrlToClient = (defaultConfig, redisUrl) => {
const config = defaultConfig; const config = defaultConfig;
@ -108,6 +118,7 @@ const startWorker = (workerId) => {
} }
const app = express(); const app = express();
app.set('trusted proxy', process.env.TRUSTED_PROXY_IP || 'loopback,uniquelocal'); app.set('trusted proxy', process.env.TRUSTED_PROXY_IP || 'loopback,uniquelocal');
const pgPool = new pg.Pool(Object.assign(pgConfigs[env], dbUrlToConfig(process.env.DATABASE_URL))); const pgPool = new pg.Pool(Object.assign(pgConfigs[env], dbUrlToConfig(process.env.DATABASE_URL)));
@ -130,6 +141,9 @@ const startWorker = (workerId) => {
const redisSubscribeClient = redisUrlToClient(redisParams, process.env.REDIS_URL); const redisSubscribeClient = redisUrlToClient(redisParams, process.env.REDIS_URL);
const redisClient = redisUrlToClient(redisParams, process.env.REDIS_URL); const redisClient = redisUrlToClient(redisParams, process.env.REDIS_URL);
/**
* @type {Object.<string, Array.<function(string): void>>}
*/
const subs = {}; const subs = {};
redisSubscribeClient.on('message', (channel, message) => { redisSubscribeClient.on('message', (channel, message) => {
@ -144,11 +158,11 @@ const startWorker = (workerId) => {
callbacks.forEach(callback => callback(message)); callbacks.forEach(callback => callback(message));
}); });
/**
* @param {string[]} channels
* @return {function(): void}
*/
const subscriptionHeartbeat = channels => { const subscriptionHeartbeat = channels => {
if (!Array.isArray(channels)) {
channels = [channels];
}
const interval = 6 * 60; const interval = 6 * 60;
const tellSubscribed = () => { const tellSubscribed = () => {
@ -164,25 +178,65 @@ const startWorker = (workerId) => {
}; };
}; };
/**
* @param {string} channel
* @param {function(string): void} callback
*/
const subscribe = (channel, callback) => { const subscribe = (channel, callback) => {
log.silly(`Adding listener for ${channel}`); log.silly(`Adding listener for ${channel}`);
subs[channel] = subs[channel] || []; subs[channel] = subs[channel] || [];
if (subs[channel].length === 0) { if (subs[channel].length === 0) {
log.verbose(`Subscribe ${channel}`); log.verbose(`Subscribe ${channel}`);
redisSubscribeClient.subscribe(channel); redisSubscribeClient.subscribe(channel);
} }
subs[channel].push(callback); subs[channel].push(callback);
}; };
/**
* @param {string} channel
* @param {function(string): void} callback
*/
const unsubscribe = (channel, callback) => { const unsubscribe = (channel, callback) => {
log.silly(`Removing listener for ${channel}`); log.silly(`Removing listener for ${channel}`);
if (!subs[channel]) {
return;
}
subs[channel] = subs[channel].filter(item => item !== callback); subs[channel] = subs[channel].filter(item => item !== callback);
if (subs[channel].length === 0) { if (subs[channel].length === 0) {
log.verbose(`Unsubscribe ${channel}`); log.verbose(`Unsubscribe ${channel}`);
redisSubscribeClient.unsubscribe(channel); redisSubscribeClient.unsubscribe(channel);
} }
}; };
const FALSE_VALUES = [
false,
0,
"0",
"f",
"F",
"false",
"FALSE",
"off",
"OFF"
];
/**
* @param {any} value
* @return {boolean}
*/
const isTruthy = value =>
value && !FALSE_VALUES.includes(value);
/**
* @param {any} req
* @param {any} res
* @param {function(Error=): void}
*/
const allowCrossDomain = (req, res, next) => { const allowCrossDomain = (req, res, next) => {
res.header('Access-Control-Allow-Origin', '*'); res.header('Access-Control-Allow-Origin', '*');
res.header('Access-Control-Allow-Headers', 'Authorization, Accept, Cache-Control'); res.header('Access-Control-Allow-Headers', 'Authorization, Accept, Cache-Control');
@ -191,6 +245,11 @@ const startWorker = (workerId) => {
next(); next();
}; };
/**
* @param {any} req
* @param {any} res
* @param {function(Error=): void}
*/
const setRequestId = (req, res, next) => { const setRequestId = (req, res, next) => {
req.requestId = uuid.v4(); req.requestId = uuid.v4();
res.header('X-Request-Id', req.requestId); res.header('X-Request-Id', req.requestId);
@ -198,16 +257,26 @@ const startWorker = (workerId) => {
next(); next();
}; };
/**
* @param {any} req
* @param {any} res
* @param {function(Error=): void}
*/
const setRemoteAddress = (req, res, next) => { const setRemoteAddress = (req, res, next) => {
req.remoteAddress = req.connection.remoteAddress; req.remoteAddress = req.connection.remoteAddress;
next(); next();
}; };
const accountFromToken = (token, allowedScopes, req, next) => { /**
* @param {string} token
* @param {any} req
* @return {Promise.<void>}
*/
const accountFromToken = (token, req) => new Promise((resolve, reject) => {
pgPool.connect((err, client, done) => { pgPool.connect((err, client, done) => {
if (err) { if (err) {
next(err); reject(err);
return; return;
} }
@ -215,39 +284,35 @@ const startWorker = (workerId) => {
done(); done();
if (err) { if (err) {
next(err); reject(err);
return; return;
} }
if (result.rows.length === 0) { if (result.rows.length === 0) {
err = new Error('Invalid access token'); err = new Error('Invalid access token');
err.statusCode = 401; err.status = 401;
next(err);
return;
}
const scopes = result.rows[0].scopes.split(' '); reject(err);
if (allowedScopes.size > 0 && !scopes.some(scope => allowedScopes.includes(scope))) {
err = new Error('Access token does not cover required scopes');
err.statusCode = 401;
next(err);
return; return;
} }
req.scopes = result.rows[0].scopes.split(' ');
req.accountId = result.rows[0].account_id; req.accountId = result.rows[0].account_id;
req.chosenLanguages = result.rows[0].chosen_languages; req.chosenLanguages = result.rows[0].chosen_languages;
req.allowNotifications = scopes.some(scope => ['read', 'read:notifications'].includes(scope)); req.allowNotifications = req.scopes.some(scope => ['read', 'read:notifications'].includes(scope));
req.deviceId = result.rows[0].device_id; req.deviceId = result.rows[0].device_id;
next(); resolve();
});
}); });
}); });
};
const accountFromRequest = (req, next, required = true, allowedScopes = ['read']) => { /**
* @param {any} req
* @param {boolean=} required
* @return {Promise.<void>}
*/
const accountFromRequest = (req, required = true) => new Promise((resolve, reject) => {
const authorization = req.headers.authorization; const authorization = req.headers.authorization;
const location = url.parse(req.url, true); const location = url.parse(req.url, true);
const accessToken = location.query.access_token || req.headers['sec-websocket-protocol']; const accessToken = location.query.access_token || req.headers['sec-websocket-protocol'];
@ -255,22 +320,53 @@ const startWorker = (workerId) => {
if (!authorization && !accessToken) { if (!authorization && !accessToken) {
if (required) { if (required) {
const err = new Error('Missing access token'); const err = new Error('Missing access token');
err.statusCode = 401; err.status = 401;
next(err); reject(err);
return; return;
} else { } else {
next(); resolve();
return; return;
} }
} }
const token = authorization ? authorization.replace(/^Bearer /, '') : accessToken; const token = authorization ? authorization.replace(/^Bearer /, '') : accessToken;
accountFromToken(token, allowedScopes, req, next); resolve(accountFromToken(token, req));
});
/**
* @param {any} req
* @return {string}
*/
const channelNameFromPath = req => {
const { path, query } = req;
const onlyMedia = isTruthy(query.only_media);
const allowLocalOnly = isTruthy(query.allow_local_only);
switch(path) {
case '/api/v1/streaming/user':
return 'user';
case '/api/v1/streaming/user/notification':
return 'user:notification';
case '/api/v1/streaming/public':
return onlyMedia ? 'public:media' : 'public';
case '/api/v1/streaming/public/local':
return onlyMedia ? 'public:local:media' : 'public:local';
case '/api/v1/streaming/public/remote':
return onlyMedia ? 'public:remote:media' : 'public:remote';
case '/api/v1/streaming/hashtag':
return 'hashtag';
case '/api/v1/streaming/hashtag/local':
return 'hashtag:local';
case '/api/v1/streaming/direct':
return 'direct';
case '/api/v1/streaming/list':
return 'list';
}
}; };
const PUBLIC_STREAMS = [ const PUBLIC_CHANNELS = [
'public', 'public',
'public:media', 'public:media',
'public:local', 'public:local',
@ -281,96 +377,149 @@ const startWorker = (workerId) => {
'hashtag:local', 'hashtag:local',
]; ];
const wsVerifyClient = (info, cb) => { /**
const location = url.parse(info.req.url, true); * @param {any} req
const authRequired = alwaysRequireAuth || !PUBLIC_STREAMS.some(stream => stream === location.query.stream); * @param {string} channelName
const allowedScopes = []; * @return {Promise.<void>}
*/
const checkScopes = (req, channelName) => new Promise((resolve, reject) => {
log.silly(req.requestId, `Checking OAuth scopes for ${channelName}`);
// When accessing public channels, no scopes are needed
if (PUBLIC_CHANNELS.includes(channelName)) {
resolve();
return;
}
if (authRequired) { // The `read` scope has the highest priority, if the token has it
allowedScopes.push('read'); // then it can access all streams
if (location.query.stream === 'user:notification') { const requiredScopes = ['read'];
allowedScopes.push('read:notifications');
// When accessing specifically the notifications stream,
// we need a read:notifications, while in all other cases,
// we can allow access with read:statuses. Mind that the
// user stream will not contain notifications unless
// the token has either read or read:notifications scope
// as well, this is handled separately.
if (channelName === 'user:notification') {
requiredScopes.push('read:notifications');
} else { } else {
allowedScopes.push('read:statuses'); requiredScopes.push('read:statuses');
} }
if (requiredScopes.some(requiredScope => req.scopes.includes(requiredScope))) {
resolve();
return;
} }
accountFromRequest(info.req, err => { const err = new Error('Access token does not cover required scopes');
if (!err) { err.status = 401;
cb(true, undefined, undefined);
} else { reject(err);
});
/**
* @param {any} info
* @param {function(boolean, number, string): void} callback
*/
const wsVerifyClient = (info, callback) => {
// When verifying the websockets connection, we no longer pre-emptively
// check OAuth scopes and drop the connection if they're missing. We only
// drop the connection if access without token is not allowed by environment
// variables. OAuth scope checks are moved to the point of subscription
// to a specific stream.
accountFromRequest(info.req, alwaysRequireAuth).then(() => {
callback(true, undefined, undefined);
}).catch(err => {
log.error(info.req.requestId, err.toString()); log.error(info.req.requestId, err.toString());
cb(false, 401, 'Unauthorized'); callback(false, 401, 'Unauthorized');
} });
}, authRequired, allowedScopes);
}; };
const PUBLIC_ENDPOINTS = [ /**
'/api/v1/streaming/public', * @param {any} req
'/api/v1/streaming/public/allow_local_only', * @param {any} res
'/api/v1/streaming/public/local', * @param {function(Error=): void} next
'/api/v1/streaming/public/remote', */
'/api/v1/streaming/hashtag',
'/api/v1/streaming/hashtag/local',
];
const authenticationMiddleware = (req, res, next) => { const authenticationMiddleware = (req, res, next) => {
if (req.method === 'OPTIONS') { if (req.method === 'OPTIONS') {
next(); next();
return; return;
} }
const authRequired = alwaysRequireAuth || !PUBLIC_ENDPOINTS.some(endpoint => endpoint === req.path); accountFromRequest(req, alwaysRequireAuth).then(() => checkScopes(req, channelNameFromPath(req))).then(() => {
const allowedScopes = []; next();
}).catch(err => {
if (authRequired) { next(err);
allowedScopes.push('read'); });
if (req.path === '/api/v1/streaming/user/notification') {
allowedScopes.push('read:notifications');
} else {
allowedScopes.push('read:statuses');
}
}
accountFromRequest(req, next, authRequired, allowedScopes);
}; };
const errorMiddleware = (err, req, res, {}) => { /**
* @param {Error} err
* @param {any} req
* @param {any} res
* @param {function(Error=): void} next
*/
const errorMiddleware = (err, req, res, next) => {
log.error(req.requestId, err.toString()); log.error(req.requestId, err.toString());
res.writeHead(err.statusCode || 500, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: err.statusCode ? err.toString() : 'An unexpected error occurred' })); if (res.headersSent) {
return next(err);
}
res.writeHead(err.status || 500, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: err.status ? err.toString() : 'An unexpected error occurred' }));
}; };
/**
* @param {array}
* @param {number=} shift
* @return {string}
*/
const placeholders = (arr, shift = 0) => arr.map((_, i) => `$${i + 1 + shift}`).join(', '); const placeholders = (arr, shift = 0) => arr.map((_, i) => `$${i + 1 + shift}`).join(', ');
const authorizeListAccess = (id, req, next) => { /**
* @param {string} listId
* @param {any} req
* @return {Promise.<void>}
*/
const authorizeListAccess = (listId, req) => new Promise((resolve, reject) => {
const { accountId } = req;
pgPool.connect((err, client, done) => { pgPool.connect((err, client, done) => {
if (err) { if (err) {
next(false); reject();
return; return;
} }
client.query('SELECT id, account_id FROM lists WHERE id = $1 LIMIT 1', [id], (err, result) => { client.query('SELECT id, account_id FROM lists WHERE id = $1 LIMIT 1', [listId], (err, result) => {
done(); done();
if (err || result.rows.length === 0 || result.rows[0].account_id !== req.accountId) { if (err || result.rows.length === 0 || result.rows[0].account_id !== accountId) {
next(false); reject();
return; return;
} }
next(true); resolve();
});
}); });
}); });
};
/**
* @param {string[]} ids
* @param {any} req
* @param {function(string, string): void} output
* @param {function(string[], function(string): void): void} attachCloseHandler
* @param {boolean=} needsFiltering
* @param {boolean=} notificationOnly
* @param {boolean=} allowLocalOnly
* @return {function(string): void}
*/
const streamFrom = (ids, req, output, attachCloseHandler, needsFiltering = false, notificationOnly = false, allowLocalOnly = false) => { const streamFrom = (ids, req, output, attachCloseHandler, needsFiltering = false, notificationOnly = false, allowLocalOnly = false) => {
const accountId = req.accountId || req.remoteAddress; const accountId = req.accountId || req.remoteAddress;
const streamType = notificationOnly ? ' (notification)' : ''; const streamType = notificationOnly ? ' (notification)' : '';
if (!Array.isArray(ids)) {
ids = [ids];
}
log.verbose(req.requestId, `Starting stream from ${ids.join(', ')} for ${accountId}${streamType}`); log.verbose(req.requestId, `Starting stream from ${ids.join(', ')} for ${accountId}${streamType}`);
const listener = message => { const listener = message => {
@ -454,10 +603,18 @@ const startWorker = (workerId) => {
subscribe(`${redisPrefix}${id}`, listener); subscribe(`${redisPrefix}${id}`, listener);
}); });
if (attachCloseHandler) {
attachCloseHandler(ids.map(id => `${redisPrefix}${id}`), listener); attachCloseHandler(ids.map(id => `${redisPrefix}${id}`), listener);
}
return listener;
}; };
// Setup stream output to HTTP /**
* @param {any} req
* @param {any} res
* @return {function(string, string): void}
*/
const streamToHttp = (req, res) => { const streamToHttp = (req, res) => {
const accountId = req.accountId || req.remoteAddress; const accountId = req.accountId || req.remoteAddress;
@ -480,12 +637,12 @@ const startWorker = (workerId) => {
}; };
}; };
// Setup stream end for HTTP /**
const streamHttpEnd = (req, closeHandler = false) => (ids, listener) => { * @param {any} req
if (!Array.isArray(ids)) { * @param {function(): void} [closeHandler]
ids = [ids]; * @return {function(string[], function(string): void)}
} */
const streamHttpEnd = (req, closeHandler = undefined) => (ids, listener) => {
req.on('close', () => { req.on('close', () => {
ids.forEach(id => { ids.forEach(id => {
unsubscribe(id, listener); unsubscribe(id, listener);
@ -497,37 +654,24 @@ const startWorker = (workerId) => {
}); });
}; };
// Setup stream output to WebSockets /**
const streamToWs = (req, ws) => (event, payload) => { * @param {any} req
* @param {any} ws
* @param {string[]} streamName
* @return {function(string, string): void}
*/
const streamToWs = (req, ws, streamName) => (event, payload) => {
if (ws.readyState !== ws.OPEN) { if (ws.readyState !== ws.OPEN) {
log.error(req.requestId, 'Tried writing to closed socket'); log.error(req.requestId, 'Tried writing to closed socket');
return; return;
} }
ws.send(JSON.stringify({ event, payload })); ws.send(JSON.stringify({ stream: streamName, event, payload }));
};
// Setup stream end for WebSockets
const streamWsEnd = (req, ws, closeHandler = false) => (id, listener) => {
const accountId = req.accountId || req.remoteAddress;
ws.on('close', () => {
log.verbose(req.requestId, `Ending stream for ${accountId}`);
unsubscribe(id, listener);
if (closeHandler) {
closeHandler();
}
});
ws.on('error', () => {
log.verbose(req.requestId, `Ending stream for ${accountId}`);
unsubscribe(id, listener);
if (closeHandler) {
closeHandler();
}
});
}; };
/**
* @param {any} res
*/
const httpNotFound = res => { const httpNotFound = res => {
res.writeHead(404, { 'Content-Type': 'application/json' }); res.writeHead(404, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: 'Not found' })); res.end(JSON.stringify({ error: 'Not found' }));
@ -545,165 +689,281 @@ const startWorker = (workerId) => {
app.use(authenticationMiddleware); app.use(authenticationMiddleware);
app.use(errorMiddleware); app.use(errorMiddleware);
app.get('/api/v1/streaming/user', (req, res) => { app.get('/api/v1/streaming/*', (req, res) => {
const channels = [`timeline:${req.accountId}`]; channelNameToIds(req, channelNameFromPath(req), req.query).then(({ channelIds, options }) => {
const onSend = streamToHttp(req, res);
const onEnd = streamHttpEnd(req, subscriptionHeartbeat(channelIds));
if (req.deviceId) { streamFrom(channelIds, req, onSend, onEnd, options.needsFiltering, options.notificationOnly, options.allowLocalOnly);
channels.push(`timeline:${req.accountId}:${req.deviceId}`); }).catch(err => {
} log.verbose(req.requestId, 'Subscription error:', err.toString());
httpNotFound(res);
});
});
const wss = new WebSocketServer({ server, verifyClient: wsVerifyClient });
/**
* @typedef StreamParams
* @property {string} [tag]
* @property {string} [list]
* @property {string} [only_media]
*/
/**
* @param {any} req
* @param {string} name
* @param {StreamParams} params
* @return {Promise.<{ channelIds: string[], options: { needsFiltering: boolean, notificationOnly: boolean } }>}
*/
const channelNameToIds = (req, name, params) => new Promise((resolve, reject) => {
switch(name) {
case 'user':
resolve({
channelIds: req.deviceId ? [`timeline:${req.accountId}`, `timeline:${req.accountId}:${req.deviceId}`] : [`timeline:${req.accountId}`],
options: { needsFiltering: false, notificationOnly: false, allowLocalOnly: true },
});
streamFrom(channels, req, streamToHttp(req, res), streamHttpEnd(req, subscriptionHeartbeat(channels))); break;
case 'user:notification':
resolve({
channelIds: [`timeline:${req.accountId}`],
options: { needsFiltering: false, notificationOnly: true, allowLocalOnly: true },
}); });
app.get('/api/v1/streaming/user/notification', (req, res) => { break;
streamFrom(`timeline:${req.accountId}`, req, streamToHttp(req, res), streamHttpEnd(req), false, true); case 'public':
resolve({
channelIds: ['timeline:public'],
options: { needsFiltering: true, notificationOnly: false, allowLocalOnly: isTruthy(params.allow_local_only) },
}); });
app.get('/api/v1/streaming/public', (req, res) => { break;
const onlyMedia = req.query.only_media === '1' || req.query.only_media === 'true'; case 'public:allow_local_only':
const channel = onlyMedia ? 'timeline:public:media' : 'timeline:public'; resolve({
channelIds: ['timeline:public'],
options: { needsFiltering: true, notificationOnly: false, allowLocalOnly: true },
});
const allowLocalOnly = req.query.allow_local_only === '1' || req.query.allow_local_only === 'true'; break;
case 'public:local':
resolve({
channelIds: ['timeline:public:local'],
options: { needsFiltering: true, notificationOnly: false, allowLocalOnly: true },
});
streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req), true, false, allowLocalOnly); break;
case 'public:remote':
resolve({
channelIds: ['timeline:public:remote'],
options: { needsFiltering: true, notificationOnly: false, allowLocalOnly: false },
}); });
app.get('/api/v1/streaming/public/local', (req, res) => { break;
const onlyMedia = req.query.only_media === '1' || req.query.only_media === 'true'; case 'public:media':
const channel = onlyMedia ? 'timeline:public:local:media' : 'timeline:public:local'; resolve({
channelIds: ['timeline:public:media'],
options: { needsFiltering: true, notificationOnly: false, allowLocalOnly: isTruthy(query.allow_local_only) },
});
streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req), true, false, true); break;
case 'public:allow_local_only:media':
resolve({
channelIds: ['timeline:public:media'],
options: { needsFiltering: true, notificationsOnly: false, allowLocalOnly: true },
}); });
app.get('/api/v1/streaming/public/remote', (req, res) => { break;
const onlyMedia = req.query.only_media === '1' || req.query.only_media === 'true'; case 'public:local:media':
const channel = onlyMedia ? 'timeline:public:remote:media' : 'timeline:public:remote'; resolve({
channelIds: ['timeline:public:local:media'],
options: { needsFiltering: true, notificationOnly: false, allowLocalOnly: true },
});
streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req), true); break;
case 'public:remote:media':
resolve({
channelIds: ['timeline:public:remote:media'],
options: { needsFiltering: true, notificationOnly: false, allowLocalOnly: false },
}); });
app.get('/api/v1/streaming/direct', (req, res) => { break;
const channel = `timeline:direct:${req.accountId}`; case 'direct':
streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req, subscriptionHeartbeat(channel)), true); resolve({
channelIds: [`timeline:direct:${req.accountId}`],
options: { needsFiltering: false, notificationOnly: false, allowLocalOnly: true },
}); });
app.get('/api/v1/streaming/hashtag', (req, res) => { break;
const { tag } = req.query; case 'hashtag':
if (!params.tag || params.tag.length === 0) {
reject('No tag for stream provided');
} else {
resolve({
channelIds: [`timeline:hashtag:${params.tag.toLowerCase()}`],
options: { needsFiltering: true, notificationOnly: false, allowLocalOnly: true },
});
}
if (!tag || tag.length === 0) { break;
httpNotFound(res); case 'hashtag:local':
return; if (!params.tag || params.tag.length === 0) {
reject('No tag for stream provided');
} else {
resolve({
channelIds: [`timeline:hashtag:${params.tag.toLowerCase()}:local`],
options: { needsFiltering: true, notificationOnly: false, allowLocalOnly: true },
});
} }
streamFrom(`timeline:hashtag:${tag.toLowerCase()}`, req, streamToHttp(req, res), streamHttpEnd(req), true); break;
case 'list':
authorizeListAccess(params.list, req).then(() => {
resolve({
channelIds: [`timeline:list:${params.list}`],
options: { needsFiltering: false, notificationOnly: false, allowLocalOnly: true },
});
}).catch(() => {
reject('Not authorized to stream this list');
}); });
app.get('/api/v1/streaming/hashtag/local', (req, res) => { break;
const { tag } = req.query; default:
reject('Unknown stream type');
}
});
if (!tag || tag.length === 0) { /**
httpNotFound(res); * @param {string} channelName
* @param {StreamParams} params
* @return {string[]}
*/
const streamNameFromChannelName = (channelName, params) => {
if (channelName === 'list') {
return [channelName, params.list];
} else if (['hashtag', 'hashtag:local'].includes(channelName)) {
return [channelName, params.tag];
} else {
return [channelName];
}
};
/**
* @typedef WebSocketSession
* @property {any} socket
* @property {any} request
* @property {Object.<string, { listener: function(string): void, stopHeartbeat: function(): void }>} subscriptions
*/
/**
* @param {WebSocketSession} session
* @param {string} channelName
* @param {StreamParams} params
*/
const subscribeWebsocketToChannel = ({ socket, request, subscriptions }, channelName, params) =>
checkScopes(request, channelName).then(() => channelNameToIds(request, channelName, params)).then(({ channelIds, options }) => {
if (subscriptions[channelIds.join(';')]) {
return; return;
} }
streamFrom(`timeline:hashtag:${tag.toLowerCase()}:local`, req, streamToHttp(req, res), streamHttpEnd(req), true); const onSend = streamToWs(request, socket, streamNameFromChannelName(channelName, params));
const stopHeartbeat = subscriptionHeartbeat(channelIds);
const listener = streamFrom(channelIds, request, onSend, undefined, options.needsFiltering, options.notificationOnly, options.allowLocalOnly);
subscriptions[channelIds.join(';')] = {
listener,
stopHeartbeat,
};
}).catch(err => {
log.verbose(request.requestId, 'Subscription error:', err.toString());
socket.send(JSON.stringify({ error: err.toString() }));
}); });
app.get('/api/v1/streaming/list', (req, res) => { /**
const listId = req.query.list; * @param {WebSocketSession} session
* @param {string} channelName
* @param {StreamParams} params
*/
const unsubscribeWebsocketFromChannel = ({ socket, request, subscriptions }, channelName, params) =>
channelNameToIds(request, channelName, params).then(({ channelIds }) => {
log.verbose(request.requestId, `Ending stream from ${channelIds.join(', ')} for ${request.accountId}`);
authorizeListAccess(listId, req, authorized => { const { listener, stopHeartbeat } = subscriptions[channelIds.join(';')];
if (!authorized) {
httpNotFound(res); if (!listener) {
return; return;
} }
const channel = `timeline:list:${listId}`; channelIds.forEach(channelId => {
streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req, subscriptionHeartbeat(channel))); unsubscribe(`${redisPrefix}${channelId}`, listener);
}); });
stopHeartbeat();
subscriptions[channelIds.join(';')] = undefined;
}).catch(err => {
log.verbose(request.requestId, 'Unsubscription error:', err);
socket.send(JSON.stringify({ error: err.toString() }));
}); });
const wss = new WebSocketServer({ server, verifyClient: wsVerifyClient }); /**
* @param {string|string[]} arrayOrString
* @return {string}
*/
const firstParam = arrayOrString => {
if (Array.isArray(arrayOrString)) {
return arrayOrString[0];
} else {
return arrayOrString;
}
};
wss.on('connection', (ws, req) => { wss.on('connection', (ws, req) => {
const location = url.parse(req.url, true); const location = url.parse(req.url, true);
req.requestId = uuid.v4(); req.requestId = uuid.v4();
req.remoteAddress = ws._socket.remoteAddress; req.remoteAddress = ws._socket.remoteAddress;
let channel; /**
* @type {WebSocketSession}
*/
const session = {
socket: ws,
request: req,
subscriptions: {},
};
switch(location.query.stream) { const onEnd = () => {
case 'user': const keys = Object.keys(session.subscriptions);
channel = [`timeline:${req.accountId}`];
if (req.deviceId) { keys.forEach(channelIds => {
channel.push(`timeline:${req.accountId}:${req.deviceId}`); const { listener, stopHeartbeat } = session.subscriptions[channelIds];
}
streamFrom(channel, req, streamToWs(req, ws), streamWsEnd(req, ws, subscriptionHeartbeat(channel))); channelIds.split(';').forEach(channelId => {
break; unsubscribe(`${redisPrefix}${channelId}`, listener);
case 'user:notification': });
streamFrom(`timeline:${req.accountId}`, req, streamToWs(req, ws), streamWsEnd(req, ws), false, true);
break;
case 'public':
streamFrom('timeline:public', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
break;
case 'public:allow_local_only':
streamFrom('timeline:public', req, streamToWs(req, ws), streamWsEnd(req, ws), true, false, true);
break;
case 'public:local':
streamFrom('timeline:public:local', req, streamToWs(req, ws), streamWsEnd(req, ws), true, false, true);
break;
case 'public:remote':
streamFrom('timeline:public:remote', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
break;
case 'public:media':
streamFrom('timeline:public:media', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
break;
case 'public:allow_local_only:media':
streamFrom('timeline:public:media', req, streamToWs(req, ws), streamWsEnd(req, ws), true, false, true);
break;
case 'public:local:media':
streamFrom('timeline:public:local:media', req, streamToWs(req, ws), streamWsEnd(req, ws), true, false, true);
break;
case 'public:remote:media':
streamFrom('timeline:public:remote:media', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
break;
case 'direct':
channel = `timeline:direct:${req.accountId}`;
streamFrom(channel, req, streamToWs(req, ws), streamWsEnd(req, ws, subscriptionHeartbeat(channel)), true);
break;
case 'hashtag':
if (!location.query.tag || location.query.tag.length === 0) {
ws.close();
return;
}
streamFrom(`timeline:hashtag:${location.query.tag.toLowerCase()}`, req, streamToWs(req, ws), streamWsEnd(req, ws), true); stopHeartbeat();
break; });
case 'hashtag:local': };
if (!location.query.tag || location.query.tag.length === 0) {
ws.close();
return;
}
streamFrom(`timeline:hashtag:${location.query.tag.toLowerCase()}:local`, req, streamToWs(req, ws), streamWsEnd(req, ws), true); ws.on('close', onEnd);
break; ws.on('error', onEnd);
case 'list':
const listId = location.query.list;
authorizeListAccess(listId, req, authorized => { ws.on('message', data => {
if (!authorized) { const { type, stream, ...params } = JSON.parse(data);
ws.close();
return;
}
channel = `timeline:list:${listId}`; if (type === 'subscribe') {
streamFrom(channel, req, streamToWs(req, ws), streamWsEnd(req, ws, subscriptionHeartbeat(channel))); subscribeWebsocketToChannel(session, firstParam(stream), params);
} else if (type === 'unsubscribe') {
unsubscribeWebsocketFromChannel(session, firstParam(stream), params)
} else {
// Unknown action type
}
}); });
break;
default: if (location.query.stream) {
ws.close(); subscribeWebsocketToChannel(session, firstParam(location.query.stream), location.query);
} }
}); });
@ -731,6 +991,10 @@ const startWorker = (workerId) => {
process.on('uncaughtException', onError); process.on('uncaughtException', onError);
}; };
/**
* @param {any} server
* @param {function(string): void} [onSuccess]
*/
const attachServerWithConfig = (server, onSuccess) => { const attachServerWithConfig = (server, onSuccess) => {
if (process.env.SOCKET || process.env.PORT && isNaN(+process.env.PORT)) { if (process.env.SOCKET || process.env.PORT && isNaN(+process.env.PORT)) {
server.listen(process.env.SOCKET || process.env.PORT, () => { server.listen(process.env.SOCKET || process.env.PORT, () => {
@ -748,6 +1012,9 @@ const attachServerWithConfig = (server, onSuccess) => {
} }
}; };
/**
* @param {function(Error=): void} onSuccess
*/
const onPortAvailable = onSuccess => { const onPortAvailable = onSuccess => {
const testServer = http.createServer(); const testServer = http.createServer();

Write Preview
Loading…
Cancel
Save