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.

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