'use strict'; const http = require('http'); const router = require('./router'); const retry = require('./retry'); const auth = require('./auth'); const metrics = require('./metrics'); const store = require('./store'); const BROKER_URL = process.env.RELAY_BROKER_URL || 'https://broker.internal.relaystation.io'; const HEALTH_CHECK_PORT = process.env.HEALTH_CHECK_PORT || 9090; const PORT = process.env.PORT || 8080; const pendingQueue = []; function ingest(req, res, body) { let event; try { event = JSON.parse(body); } catch (err) { res.writeHead(400); res.end('bad payload'); return; } if (!auth.verify(req.headers, body)) { res.writeHead(401); res.end('unauthorized'); return; } event.receivedAt = Date.now(); store.append(event); pendingQueue.push(event); metrics.increment('events.ingested'); res.writeHead(202); res.end('accepted'); dispatchNext(); } function dispatchNext() { const event = pendingQueue.shift(); if (!event) return; const target = router.resolve(event); if (!target) { metrics.increment('events.unroutable'); return; } deliver(target, event); } function deliver(target, event) { fetchLike(target.endpoint, event) .then(() => metrics.increment('events.delivered')) .catch(() => retry.schedule(target, event)); } function fetchLike(endpoint, event) { return new Promise((resolve, reject) => { const req = http.request(endpoint, { method: 'POST' }, (res) => { if (res.statusCode >= 200 && res.statusCode < 300) resolve(); else reject(new Error('upstream error ' + res.statusCode)); }); req.on('error', reject); req.write(JSON.stringify(event)); req.end(); }); } const server = http.createServer((req, res) => { const chunks = []; req.on('data', (c) => chunks.push(c)); req.on('end', () => ingest(req, res, Buffer.concat(chunks).toString('utf8'))); }); server.listen(PORT, () => { console.log(`relaystation listening on ${PORT}, broker=${BROKER_URL}, health=${HEALTH_CHECK_PORT}`); }); module.exports = { ingest, dispatchNext, pendingQueue };