闭社主体 forked from https://github.com/tootsuite/mastodon
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.

341 lines
9.7 KiB

  1. import os from 'os';
  2. import cluster from 'cluster';
  3. import dotenv from 'dotenv'
  4. import express from 'express'
  5. import http from 'http'
  6. import redis from 'redis'
  7. import pg from 'pg'
  8. import log from 'npmlog'
  9. import url from 'url'
  10. import WebSocket from 'ws'
  11. import uuid from 'uuid'
  12. const env = process.env.NODE_ENV || 'development'
  13. dotenv.config({
  14. path: env === 'production' ? '.env.production' : '.env'
  15. })
  16. if (cluster.isMaster) {
  17. // cluster master
  18. const core = +process.env.STREAMING_CLUSTER_NUM || (env === 'development' ? 1 : Math.max(os.cpus().length - 1, 1))
  19. const fork = () => {
  20. const worker = cluster.fork();
  21. worker.on('exit', (code, signal) => {
  22. log.error(`Worker died with exit code ${code}, signal ${signal} received.`);
  23. setTimeout(() => fork(), 0);
  24. });
  25. };
  26. for (let i = 0; i < core; i++) fork();
  27. log.info(`Starting streaming API server master with ${core} workers`)
  28. } else {
  29. // cluster worker
  30. const pgConfigs = {
  31. development: {
  32. database: 'mastodon_development',
  33. host: '/var/run/postgresql',
  34. max: 10
  35. },
  36. production: {
  37. user: process.env.DB_USER || 'mastodon',
  38. password: process.env.DB_PASS || '',
  39. database: process.env.DB_NAME || 'mastodon_production',
  40. host: process.env.DB_HOST || 'localhost',
  41. port: process.env.DB_PORT || 5432,
  42. max: 10
  43. }
  44. }
  45. const app = express()
  46. const pgPool = new pg.Pool(pgConfigs[env])
  47. const server = http.createServer(app)
  48. const wss = new WebSocket.Server({ server })
  49. const redisClient = redis.createClient({
  50. host: process.env.REDIS_HOST || '127.0.0.1',
  51. port: process.env.REDIS_PORT || 6379,
  52. password: process.env.REDIS_PASSWORD
  53. })
  54. const subs = {}
  55. redisClient.on('pmessage', (_, channel, message) => {
  56. const callbacks = subs[channel]
  57. log.silly(`New message on channel ${channel}`)
  58. if (!callbacks) {
  59. return
  60. }
  61. callbacks.forEach(callback => callback(message))
  62. })
  63. redisClient.psubscribe('timeline:*')
  64. const subscribe = (channel, callback) => {
  65. log.silly(`Adding listener for ${channel}`)
  66. subs[channel] = subs[channel] || []
  67. subs[channel].push(callback)
  68. }
  69. const unsubscribe = (channel, callback) => {
  70. log.silly(`Removing listener for ${channel}`)
  71. subs[channel] = subs[channel].filter(item => item !== callback)
  72. }
  73. const allowCrossDomain = (req, res, next) => {
  74. res.header('Access-Control-Allow-Origin', '*')
  75. res.header('Access-Control-Allow-Headers', 'Authorization, Accept, Cache-Control')
  76. res.header('Access-Control-Allow-Methods', 'GET, OPTIONS')
  77. next()
  78. }
  79. const setRequestId = (req, res, next) => {
  80. req.requestId = uuid.v4()
  81. res.header('X-Request-Id', req.requestId)
  82. next()
  83. }
  84. const accountFromToken = (token, req, next) => {
  85. pgPool.connect((err, client, done) => {
  86. if (err) {
  87. next(err)
  88. return
  89. }
  90. client.query('SELECT oauth_access_tokens.resource_owner_id, users.account_id FROM oauth_access_tokens INNER JOIN users ON oauth_access_tokens.resource_owner_id = users.id WHERE oauth_access_tokens.token = $1 LIMIT 1', [token], (err, result) => {
  91. done()
  92. if (err) {
  93. next(err)
  94. return
  95. }
  96. if (result.rows.length === 0) {
  97. err = new Error('Invalid access token')
  98. err.statusCode = 401
  99. next(err)
  100. return
  101. }
  102. req.accountId = result.rows[0].account_id
  103. next()
  104. })
  105. })
  106. }
  107. const authenticationMiddleware = (req, res, next) => {
  108. if (req.method === 'OPTIONS') {
  109. next()
  110. return
  111. }
  112. const authorization = req.get('Authorization')
  113. if (!authorization) {
  114. const err = new Error('Missing access token')
  115. err.statusCode = 401
  116. next(err)
  117. return
  118. }
  119. const token = authorization.replace(/^Bearer /, '')
  120. accountFromToken(token, req, next)
  121. }
  122. const errorMiddleware = (err, req, res, next) => {
  123. log.error(req.requestId, err)
  124. res.writeHead(err.statusCode || 500, { 'Content-Type': 'application/json' })
  125. res.end(JSON.stringify({ error: err.statusCode ? `${err}` : 'An unexpected error occurred' }))
  126. }
  127. const placeholders = (arr, shift = 0) => arr.map((_, i) => `$${i + 1 + shift}`).join(', ');
  128. const streamFrom = (id, req, output, attachCloseHandler, needsFiltering = false) => {
  129. log.verbose(req.requestId, `Starting stream from ${id} for ${req.accountId}`)
  130. const listener = message => {
  131. const { event, payload, queued_at } = JSON.parse(message)
  132. const transmit = () => {
  133. const now = new Date().getTime()
  134. const delta = now - queued_at;
  135. log.silly(req.requestId, `Transmitting for ${req.accountId}: ${event} ${payload} Delay: ${delta}ms`)
  136. output(event, payload)
  137. }
  138. // Only messages that may require filtering are statuses, since notifications
  139. // are already personalized and deletes do not matter
  140. if (needsFiltering && event === 'update') {
  141. pgPool.connect((err, client, done) => {
  142. if (err) {
  143. log.error(err)
  144. return
  145. }
  146. const unpackedPayload = JSON.parse(payload)
  147. const targetAccountIds = [unpackedPayload.account.id].concat(unpackedPayload.mentions.map(item => item.id)).concat(unpackedPayload.reblog ? [unpackedPayload.reblog.account.id] : [])
  148. client.query(`SELECT target_account_id FROM blocks WHERE account_id = $1 AND target_account_id IN (${placeholders(targetAccountIds, 1)}) UNION SELECT target_account_id FROM mutes WHERE account_id = $1 AND target_account_id IN (${placeholders(targetAccountIds, 1)})`, [req.accountId].concat(targetAccountIds), (err, result) => {
  149. done()
  150. if (err) {
  151. log.error(err)
  152. return
  153. }
  154. if (result.rows.length > 0) {
  155. return
  156. }
  157. transmit()
  158. })
  159. })
  160. } else {
  161. transmit()
  162. }
  163. }
  164. subscribe(id, listener)
  165. attachCloseHandler(id, listener)
  166. }
  167. // Setup stream output to HTTP
  168. const streamToHttp = (req, res) => {
  169. res.setHeader('Content-Type', 'text/event-stream')
  170. res.setHeader('Transfer-Encoding', 'chunked')
  171. const heartbeat = setInterval(() => res.write(':thump\n'), 15000)
  172. req.on('close', () => {
  173. log.verbose(req.requestId, `Ending stream for ${req.accountId}`)
  174. clearInterval(heartbeat)
  175. })
  176. return (event, payload) => {
  177. res.write(`event: ${event}\n`)
  178. res.write(`data: ${payload}\n\n`)
  179. }
  180. }
  181. // Setup stream end for HTTP
  182. const streamHttpEnd = req => (id, listener) => {
  183. req.on('close', () => {
  184. unsubscribe(id, listener)
  185. })
  186. }
  187. // Setup stream output to WebSockets
  188. const streamToWs = (req, ws) => {
  189. const heartbeat = setInterval(() => ws.ping(), 15000)
  190. ws.on('close', () => {
  191. log.verbose(req.requestId, `Ending stream for ${req.accountId}`)
  192. clearInterval(heartbeat)
  193. })
  194. return (event, payload) => {
  195. if (ws.readyState !== ws.OPEN) {
  196. log.error(req.requestId, 'Tried writing to closed socket')
  197. return
  198. }
  199. ws.send(JSON.stringify({ event, payload }))
  200. }
  201. }
  202. // Setup stream end for WebSockets
  203. const streamWsEnd = ws => (id, listener) => {
  204. ws.on('close', () => {
  205. unsubscribe(id, listener)
  206. })
  207. ws.on('error', e => {
  208. unsubscribe(id, listener)
  209. })
  210. }
  211. app.use(setRequestId)
  212. app.use(allowCrossDomain)
  213. app.use(authenticationMiddleware)
  214. app.use(errorMiddleware)
  215. app.get('/api/v1/streaming/user', (req, res) => {
  216. streamFrom(`timeline:${req.accountId}`, req, streamToHttp(req, res), streamHttpEnd(req))
  217. })
  218. app.get('/api/v1/streaming/public', (req, res) => {
  219. streamFrom('timeline:public', req, streamToHttp(req, res), streamHttpEnd(req), true)
  220. })
  221. app.get('/api/v1/streaming/public/local', (req, res) => {
  222. streamFrom('timeline:public:local', req, streamToHttp(req, res), streamHttpEnd(req), true)
  223. })
  224. app.get('/api/v1/streaming/hashtag', (req, res) => {
  225. streamFrom(`timeline:hashtag:${req.query.tag}`, req, streamToHttp(req, res), streamHttpEnd(req), true)
  226. })
  227. app.get('/api/v1/streaming/hashtag/local', (req, res) => {
  228. streamFrom(`timeline:hashtag:${req.query.tag}:local`, req, streamToHttp(req, res), streamHttpEnd(req), true)
  229. })
  230. wss.on('connection', ws => {
  231. const location = url.parse(ws.upgradeReq.url, true)
  232. const token = location.query.access_token
  233. const req = { requestId: uuid.v4() }
  234. accountFromToken(token, req, err => {
  235. if (err) {
  236. log.error(req.requestId, err)
  237. ws.close()
  238. return
  239. }
  240. switch(location.query.stream) {
  241. case 'user':
  242. streamFrom(`timeline:${req.accountId}`, req, streamToWs(req, ws), streamWsEnd(ws))
  243. break;
  244. case 'public':
  245. streamFrom('timeline:public', req, streamToWs(req, ws), streamWsEnd(ws), true)
  246. break;
  247. case 'public:local':
  248. streamFrom('timeline:public:local', req, streamToWs(req, ws), streamWsEnd(ws), true)
  249. break;
  250. case 'hashtag':
  251. streamFrom(`timeline:hashtag:${location.query.tag}`, req, streamToWs(req, ws), streamWsEnd(ws), true)
  252. break;
  253. case 'hashtag:local':
  254. streamFrom(`timeline:hashtag:${location.query.tag}:local`, req, streamToWs(req, ws), streamWsEnd(ws), true)
  255. break;
  256. default:
  257. ws.close()
  258. }
  259. })
  260. })
  261. server.listen(process.env.PORT || 4000, () => {
  262. log.level = process.env.LOG_LEVEL || 'verbose'
  263. log.info(`Starting streaming API server worker on ${server.address().address}:${server.address().port}`)
  264. })
  265. process.on('SIGINT', exit)
  266. process.on('SIGTERM', exit)
  267. process.on('exit', exit)
  268. function exit() {
  269. server.close()
  270. }
  271. }