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

590 lines
17 KiB

  1. const os = require('os');
  2. const throng = require('throng');
  3. const dotenv = require('dotenv');
  4. const express = require('express');
  5. const http = require('http');
  6. const redis = require('redis');
  7. const pg = require('pg');
  8. const log = require('npmlog');
  9. const url = require('url');
  10. const WebSocket = require('uws');
  11. const uuid = require('uuid');
  12. const env = process.env.NODE_ENV || 'development';
  13. dotenv.config({
  14. path: env === 'production' ? '.env.production' : '.env',
  15. });
  16. log.level = process.env.LOG_LEVEL || 'verbose';
  17. const dbUrlToConfig = (dbUrl) => {
  18. if (!dbUrl) {
  19. return {};
  20. }
  21. const params = url.parse(dbUrl);
  22. const config = {};
  23. if (params.auth) {
  24. [config.user, config.password] = params.auth.split(':');
  25. }
  26. if (params.hostname) {
  27. config.host = params.hostname;
  28. }
  29. if (params.port) {
  30. config.port = params.port;
  31. }
  32. if (params.pathname) {
  33. config.database = params.pathname.split('/')[1];
  34. }
  35. const ssl = params.query && params.query.ssl;
  36. if (ssl) {
  37. config.ssl = ssl === 'true' || ssl === '1';
  38. }
  39. return config;
  40. };
  41. const redisUrlToClient = (defaultConfig, redisUrl) => {
  42. const config = defaultConfig;
  43. if (!redisUrl) {
  44. return redis.createClient(config);
  45. }
  46. if (redisUrl.startsWith('unix://')) {
  47. return redis.createClient(redisUrl.slice(7), config);
  48. }
  49. return redis.createClient(Object.assign(config, {
  50. url: redisUrl,
  51. }));
  52. };
  53. const numWorkers = +process.env.STREAMING_CLUSTER_NUM || (env === 'development' ? 1 : Math.max(os.cpus().length - 1, 1));
  54. const startMaster = () => {
  55. log.info(`Starting streaming API server master with ${numWorkers} workers`);
  56. };
  57. const startWorker = (workerId) => {
  58. log.info(`Starting worker ${workerId}`);
  59. const pgConfigs = {
  60. development: {
  61. user: process.env.DB_USER || pg.defaults.user,
  62. password: process.env.DB_PASS || pg.defaults.password,
  63. database: process.env.DB_NAME || 'mastodon_development',
  64. host: process.env.DB_HOST || pg.defaults.host,
  65. port: process.env.DB_PORT || pg.defaults.port,
  66. max: 10,
  67. },
  68. production: {
  69. user: process.env.DB_USER || 'mastodon',
  70. password: process.env.DB_PASS || '',
  71. database: process.env.DB_NAME || 'mastodon_production',
  72. host: process.env.DB_HOST || 'localhost',
  73. port: process.env.DB_PORT || 5432,
  74. max: 10,
  75. },
  76. };
  77. const app = express();
  78. app.set('trusted proxy', process.env.TRUSTED_PROXY_IP || 'loopback,uniquelocal');
  79. const pgPool = new pg.Pool(Object.assign(pgConfigs[env], dbUrlToConfig(process.env.DATABASE_URL)));
  80. const server = http.createServer(app);
  81. const redisNamespace = process.env.REDIS_NAMESPACE || null;
  82. const redisParams = {
  83. host: process.env.REDIS_HOST || '127.0.0.1',
  84. port: process.env.REDIS_PORT || 6379,
  85. db: process.env.REDIS_DB || 0,
  86. password: process.env.REDIS_PASSWORD,
  87. };
  88. if (redisNamespace) {
  89. redisParams.namespace = redisNamespace;
  90. }
  91. const redisPrefix = redisNamespace ? `${redisNamespace}:` : '';
  92. const redisSubscribeClient = redisUrlToClient(redisParams, process.env.REDIS_URL);
  93. const redisClient = redisUrlToClient(redisParams, process.env.REDIS_URL);
  94. const subs = {};
  95. redisSubscribeClient.on('message', (channel, message) => {
  96. const callbacks = subs[channel];
  97. log.silly(`New message on channel ${channel}`);
  98. if (!callbacks) {
  99. return;
  100. }
  101. callbacks.forEach(callback => callback(message));
  102. });
  103. const subscriptionHeartbeat = (channel) => {
  104. const interval = 6*60;
  105. const tellSubscribed = () => {
  106. redisClient.set(`${redisPrefix}subscribed:${channel}`, '1', 'EX', interval*3);
  107. };
  108. tellSubscribed();
  109. const heartbeat = setInterval(tellSubscribed, interval*1000);
  110. return () => {
  111. clearInterval(heartbeat);
  112. };
  113. };
  114. const subscribe = (channel, callback) => {
  115. log.silly(`Adding listener for ${channel}`);
  116. subs[channel] = subs[channel] || [];
  117. if (subs[channel].length === 0) {
  118. log.verbose(`Subscribe ${channel}`);
  119. redisSubscribeClient.subscribe(channel);
  120. }
  121. subs[channel].push(callback);
  122. };
  123. const unsubscribe = (channel, callback) => {
  124. log.silly(`Removing listener for ${channel}`);
  125. subs[channel] = subs[channel].filter(item => item !== callback);
  126. if (subs[channel].length === 0) {
  127. log.verbose(`Unsubscribe ${channel}`);
  128. redisSubscribeClient.unsubscribe(channel);
  129. }
  130. };
  131. const allowCrossDomain = (req, res, next) => {
  132. res.header('Access-Control-Allow-Origin', '*');
  133. res.header('Access-Control-Allow-Headers', 'Authorization, Accept, Cache-Control');
  134. res.header('Access-Control-Allow-Methods', 'GET, OPTIONS');
  135. next();
  136. };
  137. const setRequestId = (req, res, next) => {
  138. req.requestId = uuid.v4();
  139. res.header('X-Request-Id', req.requestId);
  140. next();
  141. };
  142. const setRemoteAddress = (req, res, next) => {
  143. req.remoteAddress = req.connection.remoteAddress;
  144. next();
  145. };
  146. const accountFromToken = (token, req, next) => {
  147. pgPool.connect((err, client, done) => {
  148. if (err) {
  149. next(err);
  150. return;
  151. }
  152. 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) => {
  153. done();
  154. if (err) {
  155. next(err);
  156. return;
  157. }
  158. if (result.rows.length === 0) {
  159. err = new Error('Invalid access token');
  160. err.statusCode = 401;
  161. next(err);
  162. return;
  163. }
  164. req.accountId = result.rows[0].account_id;
  165. req.filteredLanguages = result.rows[0].filtered_languages;
  166. next();
  167. });
  168. });
  169. };
  170. const accountFromRequest = (req, next, required = true) => {
  171. const authorization = req.headers.authorization;
  172. const location = url.parse(req.url, true);
  173. const accessToken = location.query.access_token;
  174. if (!authorization && !accessToken) {
  175. if (required) {
  176. const err = new Error('Missing access token');
  177. err.statusCode = 401;
  178. next(err);
  179. return;
  180. } else {
  181. next();
  182. return;
  183. }
  184. }
  185. const token = authorization ? authorization.replace(/^Bearer /, '') : accessToken;
  186. accountFromToken(token, req, next);
  187. };
  188. const PUBLIC_STREAMS = [
  189. 'public',
  190. 'public:local',
  191. 'hashtag',
  192. 'hashtag:local',
  193. ];
  194. const wsVerifyClient = (info, cb) => {
  195. const location = url.parse(info.req.url, true);
  196. const authRequired = !PUBLIC_STREAMS.some(stream => stream === location.query.stream);
  197. accountFromRequest(info.req, err => {
  198. if (!err) {
  199. cb(true, undefined, undefined);
  200. } else {
  201. log.error(info.req.requestId, err.toString());
  202. cb(false, 401, 'Unauthorized');
  203. }
  204. }, authRequired);
  205. };
  206. const PUBLIC_ENDPOINTS = [
  207. '/api/v1/streaming/public',
  208. '/api/v1/streaming/public/local',
  209. '/api/v1/streaming/hashtag',
  210. '/api/v1/streaming/hashtag/local',
  211. ];
  212. const authenticationMiddleware = (req, res, next) => {
  213. if (req.method === 'OPTIONS') {
  214. next();
  215. return;
  216. }
  217. const authRequired = !PUBLIC_ENDPOINTS.some(endpoint => endpoint === req.path);
  218. accountFromRequest(req, next, authRequired);
  219. };
  220. const errorMiddleware = (err, req, res, {}) => {
  221. log.error(req.requestId, err.toString());
  222. res.writeHead(err.statusCode || 500, { 'Content-Type': 'application/json' });
  223. res.end(JSON.stringify({ error: err.statusCode ? err.toString() : 'An unexpected error occurred' }));
  224. };
  225. const placeholders = (arr, shift = 0) => arr.map((_, i) => `$${i + 1 + shift}`).join(', ');
  226. const authorizeListAccess = (id, req, next) => {
  227. pgPool.connect((err, client, done) => {
  228. if (err) {
  229. next(false);
  230. return;
  231. }
  232. client.query('SELECT id, account_id FROM lists WHERE id = $1 LIMIT 1', [id], (err, result) => {
  233. done();
  234. if (err || result.rows.length === 0 || result.rows[0].account_id !== req.accountId) {
  235. next(false);
  236. return;
  237. }
  238. next(true);
  239. });
  240. });
  241. };
  242. const streamFrom = (id, req, output, attachCloseHandler, needsFiltering = false, notificationOnly = false) => {
  243. const accountId = req.accountId || req.remoteAddress;
  244. const streamType = notificationOnly ? ' (notification)' : '';
  245. log.verbose(req.requestId, `Starting stream from ${id} for ${accountId}${streamType}`);
  246. const listener = message => {
  247. const { event, payload, queued_at } = JSON.parse(message);
  248. const transmit = () => {
  249. const now = new Date().getTime();
  250. const delta = now - queued_at;
  251. const encodedPayload = typeof payload === 'object' ? JSON.stringify(payload) : payload;
  252. log.silly(req.requestId, `Transmitting for ${accountId}: ${event} ${encodedPayload} Delay: ${delta}ms`);
  253. output(event, encodedPayload);
  254. };
  255. if (notificationOnly && event !== 'notification') {
  256. return;
  257. }
  258. // Only messages that may require filtering are statuses, since notifications
  259. // are already personalized and deletes do not matter
  260. if (!needsFiltering || event !== 'update') {
  261. transmit();
  262. return;
  263. }
  264. const unpackedPayload = payload;
  265. const targetAccountIds = [unpackedPayload.account.id].concat(unpackedPayload.mentions.map(item => item.id));
  266. const accountDomain = unpackedPayload.account.acct.split('@')[1];
  267. if (Array.isArray(req.filteredLanguages) && req.filteredLanguages.indexOf(unpackedPayload.language) !== -1) {
  268. log.silly(req.requestId, `Message ${unpackedPayload.id} filtered by language (${unpackedPayload.language})`);
  269. return;
  270. }
  271. // When the account is not logged in, it is not necessary to confirm the block or mute
  272. if (!req.accountId) {
  273. transmit();
  274. return;
  275. }
  276. pgPool.connect((err, client, done) => {
  277. if (err) {
  278. log.error(err);
  279. return;
  280. }
  281. const queries = [
  282. 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)),
  283. ];
  284. if (accountDomain) {
  285. queries.push(client.query('SELECT 1 FROM account_domain_blocks WHERE account_id = $1 AND domain = $2', [req.accountId, accountDomain]));
  286. }
  287. Promise.all(queries).then(values => {
  288. done();
  289. if (values[0].rows.length > 0 || (values.length > 1 && values[1].rows.length > 0)) {
  290. return;
  291. }
  292. transmit();
  293. }).catch(err => {
  294. done();
  295. log.error(err);
  296. });
  297. });
  298. };
  299. subscribe(`${redisPrefix}${id}`, listener);
  300. attachCloseHandler(`${redisPrefix}${id}`, listener);
  301. };
  302. // Setup stream output to HTTP
  303. const streamToHttp = (req, res) => {
  304. const accountId = req.accountId || req.remoteAddress;
  305. res.setHeader('Content-Type', 'text/event-stream');
  306. res.setHeader('Transfer-Encoding', 'chunked');
  307. const heartbeat = setInterval(() => res.write(':thump\n'), 15000);
  308. req.on('close', () => {
  309. log.verbose(req.requestId, `Ending stream for ${accountId}`);
  310. clearInterval(heartbeat);
  311. });
  312. return (event, payload) => {
  313. res.write(`event: ${event}\n`);
  314. res.write(`data: ${payload}\n\n`);
  315. };
  316. };
  317. // Setup stream end for HTTP
  318. const streamHttpEnd = (req, closeHandler = false) => (id, listener) => {
  319. req.on('close', () => {
  320. unsubscribe(id, listener);
  321. if (closeHandler) {
  322. closeHandler();
  323. }
  324. });
  325. };
  326. // Setup stream output to WebSockets
  327. const streamToWs = (req, ws) => (event, payload) => {
  328. if (ws.readyState !== ws.OPEN) {
  329. log.error(req.requestId, 'Tried writing to closed socket');
  330. return;
  331. }
  332. ws.send(JSON.stringify({ event, payload }));
  333. };
  334. // Setup stream end for WebSockets
  335. const streamWsEnd = (req, ws, closeHandler = false) => (id, listener) => {
  336. const accountId = req.accountId || req.remoteAddress;
  337. ws.on('close', () => {
  338. log.verbose(req.requestId, `Ending stream for ${accountId}`);
  339. unsubscribe(id, listener);
  340. if (closeHandler) {
  341. closeHandler();
  342. }
  343. });
  344. ws.on('error', () => {
  345. log.verbose(req.requestId, `Ending stream for ${accountId}`);
  346. unsubscribe(id, listener);
  347. if (closeHandler) {
  348. closeHandler();
  349. }
  350. });
  351. };
  352. app.use(setRequestId);
  353. app.use(setRemoteAddress);
  354. app.use(allowCrossDomain);
  355. app.use(authenticationMiddleware);
  356. app.use(errorMiddleware);
  357. app.get('/api/v1/streaming/user', (req, res) => {
  358. const channel = `timeline:${req.accountId}`;
  359. streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req, subscriptionHeartbeat(channel)));
  360. });
  361. app.get('/api/v1/streaming/user/notification', (req, res) => {
  362. streamFrom(`timeline:${req.accountId}`, req, streamToHttp(req, res), streamHttpEnd(req), false, true);
  363. });
  364. app.get('/api/v1/streaming/public', (req, res) => {
  365. streamFrom('timeline:public', req, streamToHttp(req, res), streamHttpEnd(req), true);
  366. });
  367. app.get('/api/v1/streaming/public/local', (req, res) => {
  368. streamFrom('timeline:public:local', req, streamToHttp(req, res), streamHttpEnd(req), true);
  369. });
  370. app.get('/api/v1/streaming/direct', (req, res) => {
  371. streamFrom(`timeline:direct:${req.accountId}`, req, streamToHttp(req, res), streamHttpEnd(req), true);
  372. });
  373. app.get('/api/v1/streaming/hashtag', (req, res) => {
  374. streamFrom(`timeline:hashtag:${req.query.tag.toLowerCase()}`, req, streamToHttp(req, res), streamHttpEnd(req), true);
  375. });
  376. app.get('/api/v1/streaming/hashtag/local', (req, res) => {
  377. streamFrom(`timeline:hashtag:${req.query.tag.toLowerCase()}:local`, req, streamToHttp(req, res), streamHttpEnd(req), true);
  378. });
  379. app.get('/api/v1/streaming/list', (req, res) => {
  380. const listId = req.query.list;
  381. authorizeListAccess(listId, req, authorized => {
  382. if (!authorized) {
  383. res.writeHead(404, { 'Content-Type': 'application/json' });
  384. res.end(JSON.stringify({ error: 'Not found' }));
  385. return;
  386. }
  387. const channel = `timeline:list:${listId}`;
  388. streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req, subscriptionHeartbeat(channel)));
  389. });
  390. });
  391. const wss = new WebSocket.Server({ server, verifyClient: wsVerifyClient });
  392. wss.on('connection', ws => {
  393. const req = ws.upgradeReq;
  394. const location = url.parse(req.url, true);
  395. req.requestId = uuid.v4();
  396. req.remoteAddress = ws._socket.remoteAddress;
  397. ws.isAlive = true;
  398. ws.on('pong', () => {
  399. ws.isAlive = true;
  400. });
  401. switch(location.query.stream) {
  402. case 'user':
  403. const channel = `timeline:${req.accountId}`;
  404. streamFrom(channel, req, streamToWs(req, ws), streamWsEnd(req, ws, subscriptionHeartbeat(channel)));
  405. break;
  406. case 'user:notification':
  407. streamFrom(`timeline:${req.accountId}`, req, streamToWs(req, ws), streamWsEnd(req, ws), false, true);
  408. break;
  409. case 'public':
  410. streamFrom('timeline:public', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  411. break;
  412. case 'public:local':
  413. streamFrom('timeline:public:local', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  414. break;
  415. case 'direct':
  416. streamFrom(`timeline:direct:${req.accountId}`, req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  417. break;
  418. case 'hashtag':
  419. streamFrom(`timeline:hashtag:${location.query.tag.toLowerCase()}`, req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  420. break;
  421. case 'hashtag:local':
  422. streamFrom(`timeline:hashtag:${location.query.tag.toLowerCase()}:local`, req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  423. break;
  424. case 'list':
  425. const listId = location.query.list;
  426. authorizeListAccess(listId, req, authorized => {
  427. if (!authorized) {
  428. ws.close();
  429. return;
  430. }
  431. const channel = `timeline:list:${listId}`;
  432. streamFrom(channel, req, streamToWs(req, ws), streamWsEnd(req, ws, subscriptionHeartbeat(channel)));
  433. });
  434. break;
  435. default:
  436. ws.close();
  437. }
  438. });
  439. setInterval(() => {
  440. wss.clients.forEach(ws => {
  441. if (ws.isAlive === false) {
  442. ws.terminate();
  443. return;
  444. }
  445. ws.isAlive = false;
  446. ws.ping('', false, true);
  447. });
  448. }, 30000);
  449. server.listen(process.env.PORT || 4000, process.env.BIND || '0.0.0.0', () => {
  450. log.info(`Worker ${workerId} now listening on ${server.address().address}:${server.address().port}`);
  451. });
  452. const onExit = () => {
  453. log.info(`Worker ${workerId} exiting, bye bye`);
  454. server.close();
  455. process.exit(0);
  456. };
  457. const onError = (err) => {
  458. log.error(err);
  459. server.close();
  460. process.exit(0);
  461. };
  462. process.on('SIGINT', onExit);
  463. process.on('SIGTERM', onExit);
  464. process.on('exit', onExit);
  465. process.on('uncaughtException', onError);
  466. };
  467. throng({
  468. workers: numWorkers,
  469. lifetime: Infinity,
  470. start: startWorker,
  471. master: startMaster,
  472. });