You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
mastodon/streaming/index.js

441 lines
12 KiB
JavaScript

import os from 'os';
import throng from 'throng';
import dotenv from 'dotenv';
import express from 'express';
import http from 'http';
import redis from 'redis';
import pg from 'pg';
import log from 'npmlog';
import url from 'url';
import WebSocket from 'uws';
import uuid from 'uuid';
const env = process.env.NODE_ENV || 'development';
dotenv.config({
path: env === 'production' ? '.env.production' : '.env',
});
log.level = process.env.LOG_LEVEL || 'verbose';
const dbUrlToConfig = (dbUrl) => {
if (!dbUrl) {
return {};
}
const params = url.parse(dbUrl);
const config = {};
if (params.auth) {
[config.user, config.password] = params.auth.split(':');
}
if (params.hostname) {
config.host = params.hostname;
}
if (params.port) {
config.port = params.port;
}
if (params.pathname) {
config.database = params.pathname.split('/')[1];
}
const ssl = params.query && params.query.ssl;
if (ssl) {
config.ssl = ssl === 'true' || ssl === '1';
}
return config;
};
const redisUrlToClient = (defaultConfig, redisUrl) => {
const config = defaultConfig;
if (!redisUrl) {
return redis.createClient(config);
}
if (redisUrl.startsWith('unix://')) {
return redis.createClient(redisUrl.slice(7), config);
}
return redis.createClient(Object.assign(config, {
url: redisUrl,
}));
};
const numWorkers = +process.env.STREAMING_CLUSTER_NUM || (env === 'development' ? 1 : Math.max(os.cpus().length - 1, 1));
const startMaster = () => {
log.info(`Starting streaming API server master with ${numWorkers} workers`);
};
const startWorker = (workerId) => {
log.info(`Starting worker ${workerId}`);
const pgConfigs = {
development: {
database: 'mastodon_development',
max: 10,
},
production: {
user: process.env.DB_USER || 'mastodon',
password: process.env.DB_PASS || '',
database: process.env.DB_NAME || 'mastodon_production',
host: process.env.DB_HOST || 'localhost',
port: process.env.DB_PORT || 5432,
max: 10,
},
};
const app = express();
const pgPool = new pg.Pool(Object.assign(pgConfigs[env], dbUrlToConfig(process.env.DATABASE_URL)));
const server = http.createServer(app);
const redisNamespace = process.env.REDIS_NAMESPACE || null;
const redisParams = {
host: process.env.REDIS_HOST || '127.0.0.1',
port: process.env.REDIS_PORT || 6379,
db: process.env.REDIS_DB || 0,
password: process.env.REDIS_PASSWORD,
};
if (redisNamespace) {
redisParams.namespace = redisNamespace;
}
const redisPrefix = redisNamespace ? `${redisNamespace}:` : '';
const redisClient = redisUrlToClient(redisParams, process.env.REDIS_URL);
const subs = {};
redisClient.on('pmessage', (_, channel, message) => {
const callbacks = subs[channel];
log.silly(`New message on channel ${channel}`);
if (!callbacks) {
return;
}
callbacks.forEach(callback => callback(message));
});
redisClient.psubscribe(`${redisPrefix}timeline:*`);
const subscribe = (channel, callback) => {
log.silly(`Adding listener for ${channel}`);
subs[channel] = subs[channel] || [];
subs[channel].push(callback);
};
const unsubscribe = (channel, callback) => {
log.silly(`Removing listener for ${channel}`);
subs[channel] = subs[channel].filter(item => item !== callback);
};
const allowCrossDomain = (req, res, next) => {
res.header('Access-Control-Allow-Origin', '*');
res.header('Access-Control-Allow-Headers', 'Authorization, Accept, Cache-Control');
res.header('Access-Control-Allow-Methods', 'GET, OPTIONS');
next();
};
const setRequestId = (req, res, next) => {
req.requestId = uuid.v4();
res.header('X-Request-Id', req.requestId);
next();
};
const accountFromToken = (token, req, next) => {
pgPool.connect((err, client, done) => {
if (err) {
next(err);
return;
}
client.query('SELECT oauth_access_tokens.resource_owner_id, users.account_id, users.filtered_languages FROM oauth_access_tokens INNER JOIN users ON oauth_access_tokens.resource_owner_id = users.id WHERE oauth_access_tokens.token = $1 AND oauth_access_tokens.revoked_at IS NULL LIMIT 1', [token], (err, result) => {
done();
if (err) {
next(err);
return;
}
if (result.rows.length === 0) {
err = new Error('Invalid access token');
err.statusCode = 401;
next(err);
return;
}
req.accountId = result.rows[0].account_id;
req.filteredLanguages = result.rows[0].filtered_languages;
next();
});
});
};
const accountFromRequest = (req, next) => {
const authorization = req.headers.authorization;
const location = url.parse(req.url, true);
const accessToken = location.query.access_token;
if (!authorization && !accessToken) {
const err = new Error('Missing access token');
err.statusCode = 401;
next(err);
return;
}
const token = authorization ? authorization.replace(/^Bearer /, '') : accessToken;
accountFromToken(token, req, next);
};
const wsVerifyClient = (info, cb) => {
accountFromRequest(info.req, err => {
if (!err) {
cb(true, undefined, undefined);
} else {
log.error(info.req.requestId, err.toString());
cb(false, 401, 'Unauthorized');
}
});
};
const authenticationMiddleware = (req, res, next) => {
if (req.method === 'OPTIONS') {
next();
return;
}
accountFromRequest(req, next);
};
const errorMiddleware = (err, req, res, next) => {
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' }));
};
const placeholders = (arr, shift = 0) => arr.map((_, i) => `$${i + 1 + shift}`).join(', ');
const streamFrom = (id, req, output, attachCloseHandler, needsFiltering = false) => {
log.verbose(req.requestId, `Starting stream from ${id} for ${req.accountId}`);
const listener = message => {
const { event, payload, queued_at } = JSON.parse(message);
const transmit = () => {
const now = new Date().getTime();
const delta = now - queued_at;
log.silly(req.requestId, `Transmitting for ${req.accountId}: ${event} ${payload} Delay: ${delta}ms`);
output(event, payload);
};
// Only messages that may require filtering are statuses, since notifications
// are already personalized and deletes do not matter
if (needsFiltering && event === 'update') {
pgPool.connect((err, client, done) => {
if (err) {
log.error(err);
return;
}
const unpackedPayload = JSON.parse(payload);
const targetAccountIds = [unpackedPayload.account.id].concat(unpackedPayload.mentions.map(item => item.id));
const accountDomain = unpackedPayload.account.acct.split('@')[1];
if (Array.isArray(req.filteredLanguages) && req.filteredLanguages.includes(unpackedPayload.language)) {
log.silly(req.requestId, `Message ${unpackedPayload.id} filtered by language (${unpackedPayload.language})`);
done();
return;
}
Account domain blocks (#2381) * Add <ostatus:conversation /> tag to Atom input/output Only uses ref attribute (not href) because href would be the alternate link that's always included also. Creates new conversation for every non-reply status. Carries over conversation for every reply. Keeps remote URIs verbatim, generates local URIs on the fly like the rest of them. * Conversation muting - prevents notifications that reference a conversation (including replies, favourites, reblogs) from being created. API endpoints /api/v1/statuses/:id/mute and /api/v1/statuses/:id/unmute Currently no way to tell when a status/conversation is muted, so the web UI only has a "disable notifications" button, doesn't work as a toggle * Display "Dismiss notifications" on all statuses in notifications column, not just own * Add "muted" as a boolean attribute on statuses JSON For now always false on contained reblogs, since it's only relevant for statuses returned from the notifications endpoint, which are not nested Remove "Disable notifications" from detailed status view, since it's only relevant in the notifications column * Up max class length * Remove pending test for conversation mute * Add tests, clean up * Rename to "mute conversation" and "unmute conversation" * Raise validation error when trying to mute/unmute status without conversation * Adding account domain blocks that filter notifications and public timelines * Add tests for domain blocks in notifications, public timelines Filter reblogs of blocked domains from home * Add API for listing and creating account domain blocks * API for creating/deleting domain blocks, tests for Status#ancestors and Status#descendants, filter domain blocks from them * Filter domains in streaming API * Update account_domain_block_spec.rb
8 years ago
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)),
];
Account domain blocks (#2381) * Add <ostatus:conversation /> tag to Atom input/output Only uses ref attribute (not href) because href would be the alternate link that's always included also. Creates new conversation for every non-reply status. Carries over conversation for every reply. Keeps remote URIs verbatim, generates local URIs on the fly like the rest of them. * Conversation muting - prevents notifications that reference a conversation (including replies, favourites, reblogs) from being created. API endpoints /api/v1/statuses/:id/mute and /api/v1/statuses/:id/unmute Currently no way to tell when a status/conversation is muted, so the web UI only has a "disable notifications" button, doesn't work as a toggle * Display "Dismiss notifications" on all statuses in notifications column, not just own * Add "muted" as a boolean attribute on statuses JSON For now always false on contained reblogs, since it's only relevant for statuses returned from the notifications endpoint, which are not nested Remove "Disable notifications" from detailed status view, since it's only relevant in the notifications column * Up max class length * Remove pending test for conversation mute * Add tests, clean up * Rename to "mute conversation" and "unmute conversation" * Raise validation error when trying to mute/unmute status without conversation * Adding account domain blocks that filter notifications and public timelines * Add tests for domain blocks in notifications, public timelines Filter reblogs of blocked domains from home * Add API for listing and creating account domain blocks * API for creating/deleting domain blocks, tests for Status#ancestors and Status#descendants, filter domain blocks from them * Filter domains in streaming API * Update account_domain_block_spec.rb
8 years ago
if (accountDomain) {
queries.push(client.query('SELECT 1 FROM account_domain_blocks WHERE account_id = $1 AND domain = $2', [req.accountId, accountDomain]));
Account domain blocks (#2381) * Add <ostatus:conversation /> tag to Atom input/output Only uses ref attribute (not href) because href would be the alternate link that's always included also. Creates new conversation for every non-reply status. Carries over conversation for every reply. Keeps remote URIs verbatim, generates local URIs on the fly like the rest of them. * Conversation muting - prevents notifications that reference a conversation (including replies, favourites, reblogs) from being created. API endpoints /api/v1/statuses/:id/mute and /api/v1/statuses/:id/unmute Currently no way to tell when a status/conversation is muted, so the web UI only has a "disable notifications" button, doesn't work as a toggle * Display "Dismiss notifications" on all statuses in notifications column, not just own * Add "muted" as a boolean attribute on statuses JSON For now always false on contained reblogs, since it's only relevant for statuses returned from the notifications endpoint, which are not nested Remove "Disable notifications" from detailed status view, since it's only relevant in the notifications column * Up max class length * Remove pending test for conversation mute * Add tests, clean up * Rename to "mute conversation" and "unmute conversation" * Raise validation error when trying to mute/unmute status without conversation * Adding account domain blocks that filter notifications and public timelines * Add tests for domain blocks in notifications, public timelines Filter reblogs of blocked domains from home * Add API for listing and creating account domain blocks * API for creating/deleting domain blocks, tests for Status#ancestors and Status#descendants, filter domain blocks from them * Filter domains in streaming API * Update account_domain_block_spec.rb
8 years ago
}
Account domain blocks (#2381) * Add <ostatus:conversation /> tag to Atom input/output Only uses ref attribute (not href) because href would be the alternate link that's always included also. Creates new conversation for every non-reply status. Carries over conversation for every reply. Keeps remote URIs verbatim, generates local URIs on the fly like the rest of them. * Conversation muting - prevents notifications that reference a conversation (including replies, favourites, reblogs) from being created. API endpoints /api/v1/statuses/:id/mute and /api/v1/statuses/:id/unmute Currently no way to tell when a status/conversation is muted, so the web UI only has a "disable notifications" button, doesn't work as a toggle * Display "Dismiss notifications" on all statuses in notifications column, not just own * Add "muted" as a boolean attribute on statuses JSON For now always false on contained reblogs, since it's only relevant for statuses returned from the notifications endpoint, which are not nested Remove "Disable notifications" from detailed status view, since it's only relevant in the notifications column * Up max class length * Remove pending test for conversation mute * Add tests, clean up * Rename to "mute conversation" and "unmute conversation" * Raise validation error when trying to mute/unmute status without conversation * Adding account domain blocks that filter notifications and public timelines * Add tests for domain blocks in notifications, public timelines Filter reblogs of blocked domains from home * Add API for listing and creating account domain blocks * API for creating/deleting domain blocks, tests for Status#ancestors and Status#descendants, filter domain blocks from them * Filter domains in streaming API * Update account_domain_block_spec.rb
8 years ago
Promise.all(queries).then(values => {
done();
Account domain blocks (#2381) * Add <ostatus:conversation /> tag to Atom input/output Only uses ref attribute (not href) because href would be the alternate link that's always included also. Creates new conversation for every non-reply status. Carries over conversation for every reply. Keeps remote URIs verbatim, generates local URIs on the fly like the rest of them. * Conversation muting - prevents notifications that reference a conversation (including replies, favourites, reblogs) from being created. API endpoints /api/v1/statuses/:id/mute and /api/v1/statuses/:id/unmute Currently no way to tell when a status/conversation is muted, so the web UI only has a "disable notifications" button, doesn't work as a toggle * Display "Dismiss notifications" on all statuses in notifications column, not just own * Add "muted" as a boolean attribute on statuses JSON For now always false on contained reblogs, since it's only relevant for statuses returned from the notifications endpoint, which are not nested Remove "Disable notifications" from detailed status view, since it's only relevant in the notifications column * Up max class length * Remove pending test for conversation mute * Add tests, clean up * Rename to "mute conversation" and "unmute conversation" * Raise validation error when trying to mute/unmute status without conversation * Adding account domain blocks that filter notifications and public timelines * Add tests for domain blocks in notifications, public timelines Filter reblogs of blocked domains from home * Add API for listing and creating account domain blocks * API for creating/deleting domain blocks, tests for Status#ancestors and Status#descendants, filter domain blocks from them * Filter domains in streaming API * Update account_domain_block_spec.rb
8 years ago
if (values[0].rows.length > 0 || (values.length > 1 && values[1].rows.length > 0)) {
return;
}
transmit();
Account domain blocks (#2381) * Add <ostatus:conversation /> tag to Atom input/output Only uses ref attribute (not href) because href would be the alternate link that's always included also. Creates new conversation for every non-reply status. Carries over conversation for every reply. Keeps remote URIs verbatim, generates local URIs on the fly like the rest of them. * Conversation muting - prevents notifications that reference a conversation (including replies, favourites, reblogs) from being created. API endpoints /api/v1/statuses/:id/mute and /api/v1/statuses/:id/unmute Currently no way to tell when a status/conversation is muted, so the web UI only has a "disable notifications" button, doesn't work as a toggle * Display "Dismiss notifications" on all statuses in notifications column, not just own * Add "muted" as a boolean attribute on statuses JSON For now always false on contained reblogs, since it's only relevant for statuses returned from the notifications endpoint, which are not nested Remove "Disable notifications" from detailed status view, since it's only relevant in the notifications column * Up max class length * Remove pending test for conversation mute * Add tests, clean up * Rename to "mute conversation" and "unmute conversation" * Raise validation error when trying to mute/unmute status without conversation * Adding account domain blocks that filter notifications and public timelines * Add tests for domain blocks in notifications, public timelines Filter reblogs of blocked domains from home * Add API for listing and creating account domain blocks * API for creating/deleting domain blocks, tests for Status#ancestors and Status#descendants, filter domain blocks from them * Filter domains in streaming API * Update account_domain_block_spec.rb
8 years ago
}).catch(err => {
done();
log.error(err);
});
});
} else {
transmit();
}
};
subscribe(`${redisPrefix}${id}`, listener);
attachCloseHandler(`${redisPrefix}${id}`, listener);
};
// Setup stream output to HTTP
const streamToHttp = (req, res) => {
res.setHeader('Content-Type', 'text/event-stream');
res.setHeader('Transfer-Encoding', 'chunked');
const heartbeat = setInterval(() => res.write(':thump\n'), 15000);
req.on('close', () => {
log.verbose(req.requestId, `Ending stream for ${req.accountId}`);
clearInterval(heartbeat);
});
return (event, payload) => {
res.write(`event: ${event}\n`);
res.write(`data: ${payload}\n\n`);
};
};
// Setup stream end for HTTP
const streamHttpEnd = req => (id, listener) => {
req.on('close', () => {
unsubscribe(id, listener);
});
};
// Setup stream output to WebSockets
const streamToWs = (req, ws) => (event, payload) => {
if (ws.readyState !== ws.OPEN) {
log.error(req.requestId, 'Tried writing to closed socket');
return;
}
ws.send(JSON.stringify({ event, payload }));
};
// Setup stream end for WebSockets
const streamWsEnd = (req, ws) => (id, listener) => {
ws.on('close', () => {
log.verbose(req.requestId, `Ending stream for ${req.accountId}`);
unsubscribe(id, listener);
});
ws.on('error', e => {
log.verbose(req.requestId, `Ending stream for ${req.accountId}`);
unsubscribe(id, listener);
});
};
app.use(setRequestId);
app.use(allowCrossDomain);
app.use(authenticationMiddleware);
app.use(errorMiddleware);
app.get('/api/v1/streaming/user', (req, res) => {
streamFrom(`timeline:${req.accountId}`, req, streamToHttp(req, res), streamHttpEnd(req));
});
app.get('/api/v1/streaming/public', (req, res) => {
streamFrom('timeline:public', req, streamToHttp(req, res), streamHttpEnd(req), true);
});
app.get('/api/v1/streaming/public/local', (req, res) => {
streamFrom('timeline:public:local', req, streamToHttp(req, res), streamHttpEnd(req), true);
});
app.get('/api/v1/streaming/hashtag', (req, res) => {
streamFrom(`timeline:hashtag:${req.query.tag}`, req, streamToHttp(req, res), streamHttpEnd(req), true);
});
app.get('/api/v1/streaming/hashtag/local', (req, res) => {
streamFrom(`timeline:hashtag:${req.query.tag}:local`, req, streamToHttp(req, res), streamHttpEnd(req), true);
});
const wss = new WebSocket.Server({ server, verifyClient: wsVerifyClient });
wss.on('connection', ws => {
const req = ws.upgradeReq;
const location = url.parse(req.url, true);
req.requestId = uuid.v4();
ws.isAlive = true;
ws.on('pong', () => {
ws.isAlive = true;
});
switch(location.query.stream) {
case 'user':
streamFrom(`timeline:${req.accountId}`, req, streamToWs(req, ws), streamWsEnd(req, ws));
break;
case 'public':
streamFrom('timeline:public', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
break;
case 'public:local':
streamFrom('timeline:public:local', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
break;
case 'hashtag':
streamFrom(`timeline:hashtag:${location.query.tag}`, req, streamToWs(req, ws), streamWsEnd(req, ws), true);
break;
case 'hashtag:local':
streamFrom(`timeline:hashtag:${location.query.tag}:local`, req, streamToWs(req, ws), streamWsEnd(req, ws), true);
break;
default:
ws.close();
}
});
const wsInterval = setInterval(() => {
wss.clients.forEach(ws => {
if (ws.isAlive === false) {
ws.terminate();
return;
}
ws.isAlive = false;
ws.ping('', false, true);
});
}, 30000);
server.listen(process.env.PORT || 4000, () => {
log.info(`Worker ${workerId} now listening on ${server.address().address}:${server.address().port}`);
});
const onExit = () => {
log.info(`Worker ${workerId} exiting, bye bye`);
server.close();
};
const onError = (err) => {
log.error(err);
};
process.on('SIGINT', onExit);
process.on('SIGTERM', onExit);
process.on('exit', onExit);
process.on('error', onError);
};
throng({
workers: numWorkers,
lifetime: Infinity,
start: startWorker,
master: startMaster,
});