/g, '\n\n');
- const searchableTextContent = JSDOM.fragment(searchableContent).textContent;
- const searchableAccountContent = JSDOM.fragment([status.account.display_name, status.account.note].join('\n\n')).textContent;
+ if (req.cachedFilters) {
+ const status = payload;
+ // TODO: Calculate searchableContent in Ruby on Rails:
+ // @ts-ignore
+ const searchableContent = [
+ status.spoiler_text || '',
+ status.content,
+ ...(reference_texts || []),
+ ]
+ .concat(
+ status.poll && status.poll.options
+ ? status.poll.options.map((option) => option.title)
+ : [],
+ )
+ .concat(status.media_attachments.map((att) => att.description))
+ .join('\n\n')
+ .replace(/ /g, '\n\n');
+ const searchableTextContent =
+ JSDOM.fragment(searchableContent).textContent;
+ const searchableAccountContent = JSDOM.fragment(
+ [status.account.display_name, status.account.note].join('\n\n'),
+ ).textContent;
- const now = new Date();
- // @ts-ignore
- const filter_results = Object.values(req.cachedFilters).reduce((results, cachedFilter) => {
- // Check the filter hasn't expired before applying:
- if (cachedFilter.expires_at !== null && cachedFilter.expires_at < now) {
- return results;
- }
+ const now = new Date();
+ // @ts-ignore
+ const filter_results = Object.values(req.cachedFilters).reduce(
+ (results, cachedFilter) => {
+ // Check the filter hasn't expired before applying:
+ if (
+ cachedFilter.expires_at !== null &&
+ cachedFilter.expires_at < now
+ ) {
+ return results;
+ }
- if (cachedFilter.filter && cachedFilter.filter.excludeFollows && following) {
- return results;
- }
+ if (
+ cachedFilter.filter &&
+ cachedFilter.filter.excludeFollows &&
+ following
+ ) {
+ return results;
+ }
- if (cachedFilter.filter && cachedFilter.filter.excludeLocalusers && !accountDomain) {
- return results;
- }
+ if (
+ cachedFilter.filter &&
+ cachedFilter.filter.excludeLocalusers &&
+ !accountDomain
+ ) {
+ return results;
+ }
- // Just in-case JSDOM fails to find textContent in searchableContent
- if (!searchableTextContent) {
- return results;
- }
+ // Just in-case JSDOM fails to find textContent in searchableContent
+ if (!searchableTextContent) {
+ return results;
+ }
- const keyword_matches = searchableTextContent.match(cachedFilter.regexp) ||
- ((cachedFilter.withAccountName && searchableAccountContent) ? searchableAccountContent.match(cachedFilter.regexp) : null);
- if (keyword_matches) {
- // results is an Array of FilterResult; status_matches is always
- // null as we only are only applying the keyword-based custom
- // filters, not the status-based custom filters.
- // https://docs.joinmastodon.org/entities/FilterResult/
- results.push({
- filter: cachedFilter.filter,
- keyword_matches,
- status_matches: null
- });
- }
+ const keyword_matches =
+ searchableTextContent.match(cachedFilter.regexp) ||
+ (cachedFilter.withAccountName && searchableAccountContent
+ ? searchableAccountContent.match(cachedFilter.regexp)
+ : null);
+ if (keyword_matches) {
+ // results is an Array of FilterResult; status_matches is always
+ // null as we only are only applying the keyword-based custom
+ // filters, not the status-based custom filters.
+ // https://docs.joinmastodon.org/entities/FilterResult/
+ results.push({
+ filter: cachedFilter.filter,
+ keyword_matches,
+ status_matches: null,
+ });
+ }
- return results;
- }, []);
+ return results;
+ },
+ [],
+ );
- // Send the payload + the FilterResults as the `filtered` property
- // to the streaming connection. To reach this code, the `event` must
- // have been either `update` or `status.update`, meaning the
- // `payload` is a Status entity, which has a `filtered` property:
- //
- // filtered: https://docs.joinmastodon.org/entities/Status/#filtered
- transmit(event, {
- ...payload,
- filtered: filter_results
- });
- } else {
- transmit(event, payload);
- }
- }).catch(err => {
- log.error(err);
- releasePgConnection();
- });
+ // Send the payload + the FilterResults as the `filtered` property
+ // to the streaming connection. To reach this code, the `event` must
+ // have been either `update` or `status.update`, meaning the
+ // `payload` is a Status entity, which has a `filtered` property:
+ //
+ // filtered: https://docs.joinmastodon.org/entities/Status/#filtered
+ transmit(event, {
+ ...payload,
+ filtered: filter_results,
+ });
+ } else {
+ transmit(event, payload);
+ }
+ })
+ .catch((err) => {
+ log.error(err);
+ releasePgConnection();
+ });
});
};
- channelIds.forEach(id => {
+ channelIds.forEach((id) => {
subscribe(id, listener);
});
@@ -914,7 +1081,9 @@ const startServer = async () => {
// In theory we'll always have a channel name, but channelNameFromPath can return undefined:
if (typeof channelName === 'string') {
- metrics.connectedChannels.labels({ type: 'eventsource', channel: channelName }).inc();
+ metrics.connectedChannels
+ .labels({ type: 'eventsource', channel: channelName })
+ .inc();
}
res.setHeader('Content-Type', 'text/event-stream');
@@ -933,7 +1102,9 @@ const startServer = async () => {
metrics.connectedClients.labels({ type: 'eventsource' }).dec();
// In theory we'll always have a channel name, but channelNameFromPath can return undefined:
if (typeof channelName === 'string') {
- metrics.connectedChannels.labels({ type: 'eventsource', channel: channelName }).dec();
+ metrics.connectedChannels
+ .labels({ type: 'eventsource', channel: channelName })
+ .dec();
}
clearInterval(heartbeat);
@@ -951,17 +1122,19 @@ const startServer = async () => {
* @returns {function(string[], SubscriptionListener): void}
*/
- const streamHttpEnd = (req, closeHandler = undefined) => (ids, listener) => {
- req.on('close', () => {
- ids.forEach(id => {
- unsubscribe(id, listener);
- });
+ const streamHttpEnd =
+ (req, closeHandler = undefined) =>
+ (ids, listener) => {
+ req.on('close', () => {
+ ids.forEach((id) => {
+ unsubscribe(id, listener);
+ });
- if (closeHandler) {
- closeHandler();
- }
- });
- };
+ if (closeHandler) {
+ closeHandler();
+ }
+ });
+ };
/**
* @param {http.IncomingMessage} req
@@ -979,7 +1152,7 @@ const startServer = async () => {
ws.send(message, (/** @type {Error|undefined} */ err) => {
if (err) {
- req.log.error({err}, `Failed to send to websocket`);
+ req.log.error({ err }, `Failed to send to websocket`);
}
});
};
@@ -987,7 +1160,7 @@ const startServer = async () => {
/**
* @param {http.ServerResponse} res
*/
- const httpNotFound = res => {
+ const httpNotFound = (res) => {
res.writeHead(404, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: 'Not found' }));
};
@@ -1010,20 +1183,30 @@ const startServer = async () => {
return;
}
- channelNameToIds(req, channelName, req.query).then(({ channelIds, options }) => {
- const onSend = streamToHttp(req, res);
- const onEnd = streamHttpEnd(req, subscriptionHeartbeat(channelIds));
+ channelNameToIds(req, channelName, req.query)
+ .then(({ channelIds, options }) => {
+ const onSend = streamToHttp(req, res);
+ const onEnd = streamHttpEnd(req, subscriptionHeartbeat(channelIds));
- // @ts-ignore
- streamFrom(channelIds, req, req.log, onSend, onEnd, 'eventsource', options.needsFiltering);
- }).catch(err => {
- const {statusCode, errorMessage } = extractErrorStatusAndMessage(err);
+ // @ts-ignore
+ streamFrom(
+ channelIds,
+ req,
+ req.log,
+ onSend,
+ onEnd,
+ 'eventsource',
+ options.needsFiltering,
+ );
+ })
+ .catch((err) => {
+ const { statusCode, errorMessage } = extractErrorStatusAndMessage(err);
- res.log.info({ err }, 'Eventsource subscription error');
+ res.log.info({ err }, 'Eventsource subscription error');
- res.writeHead(statusCode, { 'Content-Type': 'application/json' });
- res.end(JSON.stringify({ error: errorMessage }));
- });
+ res.writeHead(statusCode, { 'Content-Type': 'application/json' });
+ res.end(JSON.stringify({ error: errorMessage }));
+ });
});
/**
@@ -1038,7 +1221,7 @@ const startServer = async () => {
* @param {any} req
* @returns {string[]}
*/
- const channelsForUserStream = req => {
+ const channelsForUserStream = (req) => {
const arr = [`timeline:${req.accountId}`];
if (isInScope(req, ['read', 'read:notifications'])) {
@@ -1054,125 +1237,134 @@ const startServer = async () => {
* @param {StreamParams} params
* @returns {Promise.<{ channelIds: string[], options: { needsFiltering: boolean } }>}
*/
- const channelNameToIds = (req, name, params) => new Promise((resolve, reject) => {
- switch (name) {
- case 'user':
- resolve({
- channelIds: channelsForUserStream(req),
- options: { needsFiltering: false },
- });
+ const channelNameToIds = (req, name, params) =>
+ new Promise((resolve, reject) => {
+ switch (name) {
+ case 'user':
+ resolve({
+ channelIds: channelsForUserStream(req),
+ options: { needsFiltering: false },
+ });
- break;
- case 'user:notification':
- resolve({
- channelIds: [`timeline:${req.accountId}:notifications`],
- options: { needsFiltering: false },
- });
+ break;
+ case 'user:notification':
+ resolve({
+ channelIds: [`timeline:${req.accountId}:notifications`],
+ options: { needsFiltering: false },
+ });
- break;
- case 'public':
- resolve({
- channelIds: ['timeline:public'],
- options: { needsFiltering: true },
- });
+ break;
+ case 'public':
+ resolve({
+ channelIds: ['timeline:public'],
+ options: { needsFiltering: true },
+ });
- break;
- case 'public:local':
- resolve({
- channelIds: ['timeline:public:local'],
- options: { needsFiltering: true },
- });
+ break;
+ case 'public:local':
+ resolve({
+ channelIds: ['timeline:public:local'],
+ options: { needsFiltering: true },
+ });
- break;
- case 'public:remote':
- resolve({
- channelIds: ['timeline:public:remote'],
- options: { needsFiltering: true },
- });
+ break;
+ case 'public:remote':
+ resolve({
+ channelIds: ['timeline:public:remote'],
+ options: { needsFiltering: true },
+ });
- break;
- case 'public:media':
- resolve({
- channelIds: ['timeline:public:media'],
- options: { needsFiltering: true },
- });
+ break;
+ case 'public:media':
+ resolve({
+ channelIds: ['timeline:public:media'],
+ options: { needsFiltering: true },
+ });
- break;
- case 'public:local:media':
- resolve({
- channelIds: ['timeline:public:local:media'],
- options: { needsFiltering: true },
- });
+ break;
+ case 'public:local:media':
+ resolve({
+ channelIds: ['timeline:public:local:media'],
+ options: { needsFiltering: true },
+ });
- break;
- case 'public:remote:media':
- resolve({
- channelIds: ['timeline:public:remote:media'],
- options: { needsFiltering: true },
- });
+ break;
+ case 'public:remote:media':
+ resolve({
+ channelIds: ['timeline:public:remote:media'],
+ options: { needsFiltering: true },
+ });
- break;
- case 'direct':
- resolve({
- channelIds: [`timeline:direct:${req.accountId}`],
- options: { needsFiltering: false },
- });
+ break;
+ case 'direct':
+ resolve({
+ channelIds: [`timeline:direct:${req.accountId}`],
+ options: { needsFiltering: false },
+ });
- break;
- case 'hashtag':
- if (!params.tag) {
- reject(new RequestError('Missing tag name parameter'));
- } else {
- resolve({
- channelIds: [`timeline:hashtag:${normalizeHashtag(params.tag)}`],
- options: { needsFiltering: true },
- });
+ break;
+ case 'hashtag':
+ if (!params.tag) {
+ reject(new RequestError('Missing tag name parameter'));
+ } else {
+ resolve({
+ channelIds: [`timeline:hashtag:${normalizeHashtag(params.tag)}`],
+ options: { needsFiltering: true },
+ });
+ }
+
+ break;
+ case 'hashtag:local':
+ if (!params.tag) {
+ reject(new RequestError('Missing tag name parameter'));
+ } else {
+ resolve({
+ channelIds: [
+ `timeline:hashtag:${normalizeHashtag(params.tag)}:local`,
+ ],
+ options: { needsFiltering: true },
+ });
+ }
+
+ break;
+ case 'list':
+ if (!params.list) {
+ reject(new RequestError('Missing list name parameter'));
+ return;
+ }
+
+ authorizeListAccess(params.list, req)
+ .then(() => {
+ resolve({
+ channelIds: [`timeline:list:${params.list}`],
+ options: { needsFiltering: false },
+ });
+ })
+ .catch(() => {
+ reject(
+ new AuthenticationError('Not authorized to stream this list'),
+ );
+ });
+
+ break;
+ case 'antenna':
+ // @ts-ignore
+ authorizeAntennaAccess(params.antenna, req)
+ .then(() => {
+ resolve({
+ channelIds: [`timeline:antenna:${params.antenna}`],
+ options: { needsFiltering: false },
+ });
+ })
+ .catch(() => {
+ reject('Not authorized to stream this antenna');
+ });
+
+ break;
+ default:
+ reject(new RequestError('Unknown stream type'));
}
-
- break;
- case 'hashtag:local':
- if (!params.tag) {
- reject(new RequestError('Missing tag name parameter'));
- } else {
- resolve({
- channelIds: [`timeline:hashtag:${normalizeHashtag(params.tag)}:local`],
- options: { needsFiltering: true },
- });
- }
-
- break;
- case 'list':
- if (!params.list) {
- reject(new RequestError('Missing list name parameter'));
- return;
- }
-
- authorizeListAccess(params.list, req).then(() => {
- resolve({
- channelIds: [`timeline:list:${params.list}`],
- options: { needsFiltering: false },
- });
- }).catch(() => {
- reject(new AuthenticationError('Not authorized to stream this list'));
- });
-
- break;
- case 'antenna':
- // @ts-ignore
- authorizeAntennaAccess(params.antenna, req).then(() => {
- resolve({
- channelIds: [`timeline:antenna:${params.antenna}`],
- options: { needsFiltering: false },
- });
- }).catch(() => {
- reject('Not authorized to stream this antenna');
- });
-
- break;
- default:
- reject(new RequestError('Unknown stream type'));
- }
- });
+ });
/**
* @param {string} channelName
@@ -1184,7 +1376,10 @@ const startServer = async () => {
return [channelName, params.list];
} else if (channelName === 'antenna' && params.antenna) {
return [channelName, params.antenna];
- } else if (['hashtag', 'hashtag:local'].includes(channelName) && params.tag) {
+ } else if (
+ ['hashtag', 'hashtag:local'].includes(channelName) &&
+ params.tag
+ ) {
return [channelName, params.tag];
} else {
return [channelName];
@@ -1205,46 +1400,69 @@ const startServer = async () => {
* @param {StreamParams} params
* @returns {void}
*/
- const subscribeWebsocketToChannel = ({ websocket, request, logger, subscriptions }, channelName, params) => {
- checkScopes(request, logger, channelName).then(() => channelNameToIds(request, channelName, params)).then(({
- channelIds,
- options,
- }) => {
- if (subscriptions[channelIds.join(';')]) {
- return;
- }
+ const subscribeWebsocketToChannel = (
+ { websocket, request, logger, subscriptions },
+ channelName,
+ params,
+ ) => {
+ checkScopes(request, logger, channelName)
+ .then(() => channelNameToIds(request, channelName, params))
+ .then(({ channelIds, options }) => {
+ if (subscriptions[channelIds.join(';')]) {
+ return;
+ }
- const onSend = streamToWs(request, websocket, streamNameFromChannelName(channelName, params));
- const stopHeartbeat = subscriptionHeartbeat(channelIds);
- const listener = streamFrom(channelIds, request, logger, onSend, undefined, 'websocket', options.needsFiltering);
+ const onSend = streamToWs(
+ request,
+ websocket,
+ streamNameFromChannelName(channelName, params),
+ );
+ const stopHeartbeat = subscriptionHeartbeat(channelIds);
+ const listener = streamFrom(
+ channelIds,
+ request,
+ logger,
+ onSend,
+ undefined,
+ 'websocket',
+ options.needsFiltering,
+ );
- metrics.connectedChannels.labels({ type: 'websocket', channel: channelName }).inc();
+ metrics.connectedChannels
+ .labels({ type: 'websocket', channel: channelName })
+ .inc();
- subscriptions[channelIds.join(';')] = {
- channelName,
- listener,
- stopHeartbeat,
- };
- }).catch(err => {
- const {statusCode, errorMessage } = extractErrorStatusAndMessage(err);
+ subscriptions[channelIds.join(';')] = {
+ channelName,
+ listener,
+ stopHeartbeat,
+ };
+ })
+ .catch((err) => {
+ const { statusCode, errorMessage } = extractErrorStatusAndMessage(err);
- logger.error({ err }, 'Websocket subscription error');
+ logger.error({ err }, 'Websocket subscription error');
- // If we have a socket that is alive and open still, send the error back to the client:
- if (websocket.isAlive && websocket.readyState === websocket.OPEN) {
- websocket.send(JSON.stringify({
- error: errorMessage,
- status: statusCode
- }));
- }
- });
+ // If we have a socket that is alive and open still, send the error back to the client:
+ if (websocket.isAlive && websocket.readyState === websocket.OPEN) {
+ websocket.send(
+ JSON.stringify({
+ error: errorMessage,
+ status: statusCode,
+ }),
+ );
+ }
+ });
};
/**
* @param {WebSocketSession} session
* @param {string[]} channelIds
*/
- const removeSubscription = ({ request, logger, subscriptions }, channelIds) => {
+ const removeSubscription = (
+ { request, logger, subscriptions },
+ channelIds,
+ ) => {
logger.info({ channelIds, accountId: request.accountId }, `Ending stream`);
const subscription = subscriptions[channelIds.join(';')];
@@ -1253,11 +1471,13 @@ const startServer = async () => {
return;
}
- channelIds.forEach(channelId => {
+ channelIds.forEach((channelId) => {
unsubscribe(channelId, subscription.listener);
});
- metrics.connectedChannels.labels({ type: 'websocket', channel: subscription.channelName }).dec();
+ metrics.connectedChannels
+ .labels({ type: 'websocket', channel: subscription.channelName })
+ .dec();
subscription.stopHeartbeat();
delete subscriptions[channelIds.join(';')];
@@ -1272,23 +1492,31 @@ const startServer = async () => {
const unsubscribeWebsocketFromChannel = (session, channelName, params) => {
const { websocket, request, logger } = session;
- channelNameToIds(request, channelName, params).then(({ channelIds }) => {
- removeSubscription(session, channelIds);
- }).catch(err => {
- logger.error({err}, 'Websocket unsubscribe error');
+ channelNameToIds(request, channelName, params)
+ .then(({ channelIds }) => {
+ removeSubscription(session, channelIds);
+ })
+ .catch((err) => {
+ logger.error({ err }, 'Websocket unsubscribe error');
- // If we have a socket that is alive and open still, send the error back to the client:
- if (websocket.isAlive && websocket.readyState === websocket.OPEN) {
- // TODO: Use a better error response here
- websocket.send(JSON.stringify({ error: "Error unsubscribing from channel" }));
- }
- });
+ // If we have a socket that is alive and open still, send the error back to the client:
+ if (websocket.isAlive && websocket.readyState === websocket.OPEN) {
+ // TODO: Use a better error response here
+ websocket.send(
+ JSON.stringify({ error: 'Error unsubscribing from channel' }),
+ );
+ }
+ });
};
/**
* @param {WebSocketSession} session
*/
- const subscribeWebsocketToSystemChannel = ({ websocket, request, subscriptions }) => {
+ const subscribeWebsocketToSystemChannel = ({
+ websocket,
+ request,
+ subscriptions,
+ }) => {
const accessTokenChannelId = `timeline:access_token:${request.accessTokenId}`;
const systemChannelId = `timeline:system:${request.accountId}`;
@@ -1304,18 +1532,18 @@ const startServer = async () => {
subscriptions[accessTokenChannelId] = {
channelName: 'system',
listener,
- stopHeartbeat: () => {
- },
+ stopHeartbeat: () => {},
};
subscriptions[systemChannelId] = {
channelName: 'system',
listener,
- stopHeartbeat: () => {
- },
+ stopHeartbeat: () => {},
};
- metrics.connectedChannels.labels({ type: 'websocket', channel: 'system' }).inc(2);
+ metrics.connectedChannels
+ .labels({ type: 'websocket', channel: 'system' })
+ .inc(2);
};
/**
@@ -1348,7 +1576,7 @@ const startServer = async () => {
ws.on('close', function onWebsocketClose() {
const subscriptions = Object.keys(session.subscriptions);
- subscriptions.forEach(channelIds => {
+ subscriptions.forEach((channelIds) => {
removeSubscription(session, channelIds.split(';'));
});
@@ -1375,7 +1603,10 @@ const startServer = async () => {
ws.on('message', (data, isBinary) => {
if (isBinary) {
log.warn('Received binary data, closing connection');
- ws.close(1003, 'The mastodon streaming server does not support binary messages');
+ ws.close(
+ 1003,
+ 'The mastodon streaming server does not support binary messages',
+ );
return;
}
const message = data.toString('utf8');
@@ -1401,14 +1632,18 @@ const startServer = async () => {
const location = req.url && url.parse(req.url, true);
if (location && location.query.stream) {
- subscribeWebsocketToChannel(session, firstParam(location.query.stream), location.query);
+ subscribeWebsocketToChannel(
+ session,
+ firstParam(location.query.stream),
+ location.query,
+ );
}
}
wss.on('connection', onConnection);
setInterval(() => {
- wss.clients.forEach(ws => {
+ wss.clients.forEach((ws) => {
// @ts-ignore
if (ws.isAlive === false) {
ws.terminate();
@@ -1421,7 +1656,7 @@ const startServer = async () => {
});
}, 30000);
- attachServerWithConfig(server, address => {
+ attachServerWithConfig(server, (address) => {
logger.info(`Streaming API now listening on ${address}`);
});