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.

604 lines
18 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.chosen_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.chosenLanguages = result.rows[0].chosen_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:media',
  191. 'public:local',
  192. 'public:local:media',
  193. 'hashtag',
  194. 'hashtag:local',
  195. ];
  196. const wsVerifyClient = (info, cb) => {
  197. const location = url.parse(info.req.url, true);
  198. const authRequired = !PUBLIC_STREAMS.some(stream => stream === location.query.stream);
  199. accountFromRequest(info.req, err => {
  200. if (!err) {
  201. cb(true, undefined, undefined);
  202. } else {
  203. log.error(info.req.requestId, err.toString());
  204. cb(false, 401, 'Unauthorized');
  205. }
  206. }, authRequired);
  207. };
  208. const PUBLIC_ENDPOINTS = [
  209. '/api/v1/streaming/public',
  210. '/api/v1/streaming/public/local',
  211. '/api/v1/streaming/hashtag',
  212. '/api/v1/streaming/hashtag/local',
  213. ];
  214. const authenticationMiddleware = (req, res, next) => {
  215. if (req.method === 'OPTIONS') {
  216. next();
  217. return;
  218. }
  219. const authRequired = !PUBLIC_ENDPOINTS.some(endpoint => endpoint === req.path);
  220. accountFromRequest(req, next, authRequired);
  221. };
  222. const errorMiddleware = (err, req, res, {}) => {
  223. log.error(req.requestId, err.toString());
  224. res.writeHead(err.statusCode || 500, { 'Content-Type': 'application/json' });
  225. res.end(JSON.stringify({ error: err.statusCode ? err.toString() : 'An unexpected error occurred' }));
  226. };
  227. const placeholders = (arr, shift = 0) => arr.map((_, i) => `$${i + 1 + shift}`).join(', ');
  228. const authorizeListAccess = (id, req, next) => {
  229. pgPool.connect((err, client, done) => {
  230. if (err) {
  231. next(false);
  232. return;
  233. }
  234. client.query('SELECT id, account_id FROM lists WHERE id = $1 LIMIT 1', [id], (err, result) => {
  235. done();
  236. if (err || result.rows.length === 0 || result.rows[0].account_id !== req.accountId) {
  237. next(false);
  238. return;
  239. }
  240. next(true);
  241. });
  242. });
  243. };
  244. const streamFrom = (id, req, output, attachCloseHandler, needsFiltering = false, notificationOnly = false) => {
  245. const accountId = req.accountId || req.remoteAddress;
  246. const streamType = notificationOnly ? ' (notification)' : '';
  247. log.verbose(req.requestId, `Starting stream from ${id} for ${accountId}${streamType}`);
  248. const listener = message => {
  249. const { event, payload, queued_at } = JSON.parse(message);
  250. const transmit = () => {
  251. const now = new Date().getTime();
  252. const delta = now - queued_at;
  253. const encodedPayload = typeof payload === 'object' ? JSON.stringify(payload) : payload;
  254. log.silly(req.requestId, `Transmitting for ${accountId}: ${event} ${encodedPayload} Delay: ${delta}ms`);
  255. output(event, encodedPayload);
  256. };
  257. if (notificationOnly && event !== 'notification') {
  258. return;
  259. }
  260. // Only messages that may require filtering are statuses, since notifications
  261. // are already personalized and deletes do not matter
  262. if (!needsFiltering || event !== 'update') {
  263. transmit();
  264. return;
  265. }
  266. const unpackedPayload = payload;
  267. const targetAccountIds = [unpackedPayload.account.id].concat(unpackedPayload.mentions.map(item => item.id));
  268. const accountDomain = unpackedPayload.account.acct.split('@')[1];
  269. if (Array.isArray(req.chosenLanguages) && unpackedPayload.language !== null && req.chosenLanguages.indexOf(unpackedPayload.language) === -1) {
  270. log.silly(req.requestId, `Message ${unpackedPayload.id} filtered by language (${unpackedPayload.language})`);
  271. return;
  272. }
  273. // When the account is not logged in, it is not necessary to confirm the block or mute
  274. if (!req.accountId) {
  275. transmit();
  276. return;
  277. }
  278. pgPool.connect((err, client, done) => {
  279. if (err) {
  280. log.error(err);
  281. return;
  282. }
  283. const queries = [
  284. 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)),
  285. ];
  286. if (accountDomain) {
  287. queries.push(client.query('SELECT 1 FROM account_domain_blocks WHERE account_id = $1 AND domain = $2', [req.accountId, accountDomain]));
  288. }
  289. Promise.all(queries).then(values => {
  290. done();
  291. if (values[0].rows.length > 0 || (values.length > 1 && values[1].rows.length > 0)) {
  292. return;
  293. }
  294. transmit();
  295. }).catch(err => {
  296. done();
  297. log.error(err);
  298. });
  299. });
  300. };
  301. subscribe(`${redisPrefix}${id}`, listener);
  302. attachCloseHandler(`${redisPrefix}${id}`, listener);
  303. };
  304. // Setup stream output to HTTP
  305. const streamToHttp = (req, res) => {
  306. const accountId = req.accountId || req.remoteAddress;
  307. res.setHeader('Content-Type', 'text/event-stream');
  308. res.setHeader('Transfer-Encoding', 'chunked');
  309. const heartbeat = setInterval(() => res.write(':thump\n'), 15000);
  310. req.on('close', () => {
  311. log.verbose(req.requestId, `Ending stream for ${accountId}`);
  312. clearInterval(heartbeat);
  313. });
  314. return (event, payload) => {
  315. res.write(`event: ${event}\n`);
  316. res.write(`data: ${payload}\n\n`);
  317. };
  318. };
  319. // Setup stream end for HTTP
  320. const streamHttpEnd = (req, closeHandler = false) => (id, listener) => {
  321. req.on('close', () => {
  322. unsubscribe(id, listener);
  323. if (closeHandler) {
  324. closeHandler();
  325. }
  326. });
  327. };
  328. // Setup stream output to WebSockets
  329. const streamToWs = (req, ws) => (event, payload) => {
  330. if (ws.readyState !== ws.OPEN) {
  331. log.error(req.requestId, 'Tried writing to closed socket');
  332. return;
  333. }
  334. ws.send(JSON.stringify({ event, payload }));
  335. };
  336. // Setup stream end for WebSockets
  337. const streamWsEnd = (req, ws, closeHandler = false) => (id, listener) => {
  338. const accountId = req.accountId || req.remoteAddress;
  339. ws.on('close', () => {
  340. log.verbose(req.requestId, `Ending stream for ${accountId}`);
  341. unsubscribe(id, listener);
  342. if (closeHandler) {
  343. closeHandler();
  344. }
  345. });
  346. ws.on('error', () => {
  347. log.verbose(req.requestId, `Ending stream for ${accountId}`);
  348. unsubscribe(id, listener);
  349. if (closeHandler) {
  350. closeHandler();
  351. }
  352. });
  353. };
  354. app.use(setRequestId);
  355. app.use(setRemoteAddress);
  356. app.use(allowCrossDomain);
  357. app.use(authenticationMiddleware);
  358. app.use(errorMiddleware);
  359. app.get('/api/v1/streaming/user', (req, res) => {
  360. const channel = `timeline:${req.accountId}`;
  361. streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req, subscriptionHeartbeat(channel)));
  362. });
  363. app.get('/api/v1/streaming/user/notification', (req, res) => {
  364. streamFrom(`timeline:${req.accountId}`, req, streamToHttp(req, res), streamHttpEnd(req), false, true);
  365. });
  366. app.get('/api/v1/streaming/public', (req, res) => {
  367. const onlyMedia = req.query.only_media === '1' || req.query.only_media === 'true';
  368. const channel = onlyMedia ? 'timeline:public:media' : 'timeline:public';
  369. streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req), true);
  370. });
  371. app.get('/api/v1/streaming/public/local', (req, res) => {
  372. const onlyMedia = req.query.only_media === '1' || req.query.only_media === 'true';
  373. const channel = onlyMedia ? 'timeline:public:local:media' : 'timeline:public:local';
  374. streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req), true);
  375. });
  376. app.get('/api/v1/streaming/direct', (req, res) => {
  377. streamFrom(`timeline:direct:${req.accountId}`, req, streamToHttp(req, res), streamHttpEnd(req), true);
  378. });
  379. app.get('/api/v1/streaming/hashtag', (req, res) => {
  380. streamFrom(`timeline:hashtag:${req.query.tag.toLowerCase()}`, req, streamToHttp(req, res), streamHttpEnd(req), true);
  381. });
  382. app.get('/api/v1/streaming/hashtag/local', (req, res) => {
  383. streamFrom(`timeline:hashtag:${req.query.tag.toLowerCase()}:local`, req, streamToHttp(req, res), streamHttpEnd(req), true);
  384. });
  385. app.get('/api/v1/streaming/list', (req, res) => {
  386. const listId = req.query.list;
  387. authorizeListAccess(listId, req, authorized => {
  388. if (!authorized) {
  389. res.writeHead(404, { 'Content-Type': 'application/json' });
  390. res.end(JSON.stringify({ error: 'Not found' }));
  391. return;
  392. }
  393. const channel = `timeline:list:${listId}`;
  394. streamFrom(channel, req, streamToHttp(req, res), streamHttpEnd(req, subscriptionHeartbeat(channel)));
  395. });
  396. });
  397. const wss = new WebSocket.Server({ server, verifyClient: wsVerifyClient });
  398. wss.on('connection', ws => {
  399. const req = ws.upgradeReq;
  400. const location = url.parse(req.url, true);
  401. req.requestId = uuid.v4();
  402. req.remoteAddress = ws._socket.remoteAddress;
  403. ws.isAlive = true;
  404. ws.on('pong', () => {
  405. ws.isAlive = true;
  406. });
  407. switch(location.query.stream) {
  408. case 'user':
  409. const channel = `timeline:${req.accountId}`;
  410. streamFrom(channel, req, streamToWs(req, ws), streamWsEnd(req, ws, subscriptionHeartbeat(channel)));
  411. break;
  412. case 'user:notification':
  413. streamFrom(`timeline:${req.accountId}`, req, streamToWs(req, ws), streamWsEnd(req, ws), false, true);
  414. break;
  415. case 'public':
  416. streamFrom('timeline:public', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  417. break;
  418. case 'public:local':
  419. streamFrom('timeline:public:local', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  420. break;
  421. case 'public:media':
  422. streamFrom('timeline:public:media', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  423. break;
  424. case 'public:local:media':
  425. streamFrom('timeline:public:local:media', req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  426. break;
  427. case 'direct':
  428. streamFrom(`timeline:direct:${req.accountId}`, req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  429. break;
  430. case 'hashtag':
  431. streamFrom(`timeline:hashtag:${location.query.tag.toLowerCase()}`, req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  432. break;
  433. case 'hashtag:local':
  434. streamFrom(`timeline:hashtag:${location.query.tag.toLowerCase()}:local`, req, streamToWs(req, ws), streamWsEnd(req, ws), true);
  435. break;
  436. case 'list':
  437. const listId = location.query.list;
  438. authorizeListAccess(listId, req, authorized => {
  439. if (!authorized) {
  440. ws.close();
  441. return;
  442. }
  443. const channel = `timeline:list:${listId}`;
  444. streamFrom(channel, req, streamToWs(req, ws), streamWsEnd(req, ws, subscriptionHeartbeat(channel)));
  445. });
  446. break;
  447. default:
  448. ws.close();
  449. }
  450. });
  451. setInterval(() => {
  452. wss.clients.forEach(ws => {
  453. if (ws.isAlive === false) {
  454. ws.terminate();
  455. return;
  456. }
  457. ws.isAlive = false;
  458. ws.ping('', false, true);
  459. });
  460. }, 30000);
  461. server.listen(process.env.PORT || 4000, process.env.BIND || '0.0.0.0', () => {
  462. log.info(`Worker ${workerId} now listening on ${server.address().address}:${server.address().port}`);
  463. });
  464. const onExit = () => {
  465. log.info(`Worker ${workerId} exiting, bye bye`);
  466. server.close();
  467. process.exit(0);
  468. };
  469. const onError = (err) => {
  470. log.error(err);
  471. server.close();
  472. process.exit(0);
  473. };
  474. process.on('SIGINT', onExit);
  475. process.on('SIGTERM', onExit);
  476. process.on('exit', onExit);
  477. process.on('uncaughtException', onError);
  478. };
  479. throng({
  480. workers: numWorkers,
  481. lifetime: Infinity,
  482. start: startWorker,
  483. master: startMaster,
  484. });