diff --git a/stellar-payment-platform/horizonListener.js b/stellar-payment-platform/horizonListener.js index 00ef7d4..dc8a5be 100644 --- a/stellar-payment-platform/horizonListener.js +++ b/stellar-payment-platform/horizonListener.js @@ -37,6 +37,20 @@ const POLL_INTERVAL_MS = parseInt(process.env.POLL_INTERVAL_MS, 10) || 60000; // // --------------------------------------------------------------------------- const horizon = new Horizon.Server(HORIZON_URL); +// Attempt to connect to the API server's Socket.IO endpoint so detected +// payments can be forwarded to connected clients in real-time. The listener +// runs as a separate process; it connects as a Socket.IO client and emits +// 'payment' events which the server will route to the appropriate room. +try { + const ioClient = require('socket.io-client'); + const SOCKET_SERVER = process.env.SOCKET_SERVER_URL || `http://localhost:${process.env.PORT || 5000}`; + global.__socketClient = ioClient(SOCKET_SERVER, { reconnection: true }); + global.__socketClient.on('connect', () => logger.info('[SocketClient] connected to server')); + global.__socketClient.on('connect_error', (err) => logger.error('[SocketClient] connect_error', err)); +} catch (err) { + logger.warn('Socket.IO client not available; real-time notifications disabled', err?.message || err); +} + // Track active streams so we can clean up on shutdown const activeStreams = new Map(); @@ -87,6 +101,20 @@ const watchAccount = (accountId) => { onmessage: (payment) => { if (payment.type === 'payment' || payment.type_i === 1) { logger.info(formatPayment(payment, accountId)); + + // If a Socket.IO server is available, emit the payment event so + // connected clients listening for this account receive a real-time + // notification. The horizon listener connects as a client to the + // API server and emits a 'payment' event with the address and payload. + try { + if (typeof global.__socketClient !== 'undefined' && global.__socketClient && global.__socketClient.connected) { + global.__socketClient.emit('payment', { address: accountId, payment }); + } + } catch (err) { + logger.error('Failed to emit payment over socket client', err); + } + } +<<<<<<< HEAD dispatchPaymentWebhooks({ prisma, poolGetFn: poolGet, @@ -99,6 +127,8 @@ const watchAccount = (accountId) => { ), ); } +======= +>>>>>>> 4095f79 (feat(websockets): add Socket.IO server and horizon listener forwarding (fix #430)) }, onerror: (error) => { logger.error( diff --git a/stellar-payment-platform/package-lock.json b/stellar-payment-platform/package-lock.json index 4b9136a..5286b78 100644 --- a/stellar-payment-platform/package-lock.json +++ b/stellar-payment-platform/package-lock.json @@ -28,6 +28,8 @@ "prom-client": "^15.1.3", "rate-limit-redis": "^4.2.0", "redis": "^4.7.0", + "socket.io": "^4.8.0", + "socket.io-client": "^4.8.0", "sqlite3": "^5.1.7", "uuid": "^9.0.1", "winston": "^3.19.0", @@ -2006,6 +2008,12 @@ "text-hex": "1.0.x" } }, + "node_modules/@socket.io/component-emitter": { + "version": "3.1.2", + "resolved": "https://registry.npmjs.org/@socket.io/component-emitter/-/component-emitter-3.1.2.tgz", + "integrity": "sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA==", + "license": "MIT" + }, "node_modules/@standard-schema/spec": { "version": "1.1.0", "resolved": "https://registry.npmjs.org/@standard-schema/spec/-/spec-1.1.0.tgz", @@ -2113,6 +2121,15 @@ "@babel/types": "^7.28.2" } }, + "node_modules/@types/cors": { + "version": "2.8.19", + "resolved": "https://registry.npmjs.org/@types/cors/-/cors-2.8.19.tgz", + "integrity": "sha512-mFNylyeyqN93lfe/9CSxOGREz8cpzAhH+E93xJ4xWQf62V8sQ/24reV2nyzUWM6H6Xji+GGHpkbLe7pVoUEskg==", + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@types/estree": { "version": "1.0.9", "resolved": "https://registry.npmjs.org/@types/estree/-/estree-1.0.9.tgz", @@ -2160,7 +2177,6 @@ "version": "26.0.1", "resolved": "https://registry.npmjs.org/@types/node/-/node-26.0.1.tgz", "integrity": "sha512-fc3KiUoBt6kie0N9bIW3E47vZsuaMf0PM2AaUpLCLT0s/LvX1nxAim6Fc049cNxODPpGm6qRAuUOB86SkRuPQw==", - "dev": true, "license": "MIT", "dependencies": { "undici-types": "~8.3.0" @@ -2179,6 +2195,15 @@ "integrity": "sha512-6WaYesThRMCl19iryMYP7/x2OVgCtbIVflDGFpWnb9irXI3UjYE4AzmYuiUKY1AJstGijoY+MgUszMgRxIYTYw==", "license": "MIT" }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@types/yargs": { "version": "17.0.35", "resolved": "https://registry.npmjs.org/@types/yargs/-/yargs-17.0.35.tgz", @@ -2622,6 +2647,15 @@ ], "license": "MIT" }, + "node_modules/base64id": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/base64id/-/base64id-2.0.0.tgz", + "integrity": "sha512-lGe34o6EHj9y3Kts9R4ZYs/Gr+6N7MCaMlIFA3F1R2O5/m7K06AxfSeO5530PEERE6/WyEg3lsuyw4GHlPZHog==", + "license": "MIT", + "engines": { + "node": "^4.5.0 || >= 5.9" + } + }, "node_modules/baseline-browser-mapping": { "version": "2.10.40", "resolved": "https://registry.npmjs.org/baseline-browser-mapping/-/baseline-browser-mapping-2.10.40.tgz", @@ -3897,6 +3931,95 @@ "once": "^1.4.0" } }, + "node_modules/engine.io": { + "version": "6.6.9", + "resolved": "https://registry.npmjs.org/engine.io/-/engine.io-6.6.9.tgz", + "integrity": "sha512-clKkw4C7nJ22mGgoVcCg6V/W/TxdNyIOTr89k2ONZu81qqkddPFDF0LXcbAwhzPD8DjkiRCjzuiO6Y+fkpD4vg==", + "license": "MIT", + "dependencies": { + "@types/cors": "^2.8.12", + "@types/node": ">=10.0.0", + "@types/ws": "^8.5.12", + "accepts": "~1.3.4", + "base64id": "2.0.0", + "cookie": "~0.7.2", + "cors": "~2.8.5", + "debug": "~4.4.1", + "engine.io-parser": "~5.2.1", + "ws": "~8.21.0" + }, + "engines": { + "node": ">=10.2.0" + } + }, + "node_modules/engine.io-client": { + "version": "6.6.6", + "resolved": "https://registry.npmjs.org/engine.io-client/-/engine.io-client-6.6.6.tgz", + "integrity": "sha512-iY6QdftLQ9pyiPoX082bpf/u1UewnOaJrtJIF9T0++QB34lZrj0uP+Q/bj8AlUsAxqhnkTV2BS8SBZSxOmoV5Q==", + "license": "MIT", + "dependencies": { + "@socket.io/component-emitter": "~3.1.0", + "debug": "~4.4.1", + "engine.io-parser": "~5.2.1", + "ws": "~8.21.0", + "xmlhttprequest-ssl": "~2.1.1" + } + }, + "node_modules/engine.io-client/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/engine.io-client/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, + "node_modules/engine.io-parser": { + "version": "5.2.3", + "resolved": "https://registry.npmjs.org/engine.io-parser/-/engine.io-parser-5.2.3.tgz", + "integrity": "sha512-HqD3yTBfnBxIrbnM1DoD6Pcq8NECnh8d4As1Qgh0z5Gg3jRRIqijury0CL3ghu/edArpUYiYqQiDUQBIs4np3Q==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + } + }, + "node_modules/engine.io/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/engine.io/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, "node_modules/env-paths": { "version": "2.2.1", "resolved": "https://registry.npmjs.org/env-paths/-/env-paths-2.2.1.tgz", @@ -8179,6 +8302,154 @@ "url": "https://github.com/sponsors/cyyynthia" } }, + "node_modules/socket.io": { + "version": "4.8.3", + "resolved": "https://registry.npmjs.org/socket.io/-/socket.io-4.8.3.tgz", + "integrity": "sha512-2Dd78bqzzjE6KPkD5fHZmDAKRNe3J15q+YHDrIsy9WEkqttc7GY+kT9OBLSMaPbQaEd0x1BjcmtMtXkfpc+T5A==", + "license": "MIT", + "dependencies": { + "accepts": "~1.3.4", + "base64id": "~2.0.0", + "cors": "~2.8.5", + "debug": "~4.4.1", + "engine.io": "~6.6.0", + "socket.io-adapter": "~2.5.2", + "socket.io-parser": "~4.2.4" + }, + "engines": { + "node": ">=10.2.0" + } + }, + "node_modules/socket.io-adapter": { + "version": "2.5.8", + "resolved": "https://registry.npmjs.org/socket.io-adapter/-/socket.io-adapter-2.5.8.tgz", + "integrity": "sha512-6Oy52pbg+kvdCVvjcN+FnY7BvxZ7cIHNScbvztT/It5d0vbwoJoVZmF2gjJmnV0/4WlXRfG15zc45ySk9Ah8bw==", + "license": "MIT", + "dependencies": { + "debug": "~4.4.1", + "ws": "~8.21.0" + } + }, + "node_modules/socket.io-adapter/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/socket.io-adapter/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, + "node_modules/socket.io-client": { + "version": "4.8.3", + "resolved": "https://registry.npmjs.org/socket.io-client/-/socket.io-client-4.8.3.tgz", + "integrity": "sha512-uP0bpjWrjQmUt5DTHq9RuoCBdFJF10cdX9X+a368j/Ft0wmaVgxlrjvK3kjvgCODOMMOz9lcaRzxmso0bTWZ/g==", + "license": "MIT", + "dependencies": { + "@socket.io/component-emitter": "~3.1.0", + "debug": "~4.4.1", + "engine.io-client": "~6.6.1", + "socket.io-parser": "~4.2.4" + }, + "engines": { + "node": ">=10.0.0" + } + }, + "node_modules/socket.io-client/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/socket.io-client/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, + "node_modules/socket.io-parser": { + "version": "4.2.7", + "resolved": "https://registry.npmjs.org/socket.io-parser/-/socket.io-parser-4.2.7.tgz", + "integrity": "sha512-IH/iSeO9T6gz1KkFleGDWkG9N3dl4jXVYUtMhIqH10Md0ttMer8nUNWiP1DKuNrybD2xBrixLJdCC9J6ECoYkg==", + "license": "MIT", + "dependencies": { + "@socket.io/component-emitter": "~3.1.0", + "debug": "~4.4.1" + }, + "engines": { + "node": ">=10.0.0" + } + }, + "node_modules/socket.io-parser/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/socket.io-parser/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, + "node_modules/socket.io/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/socket.io/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, "node_modules/socks": { "version": "2.8.9", "resolved": "https://registry.npmjs.org/socks/-/socks-2.8.9.tgz", @@ -8803,7 +9074,6 @@ "version": "8.3.0", "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-8.3.0.tgz", "integrity": "sha512-j375ScV60dom+YkPFIfTLcOiPxkN/buHz5GobjLhixFuANaNs3C9l4GmrWqejgXWJ7BbJcFYpTEUkS1Ge8bpZQ==", - "dev": true, "license": "MIT" }, "node_modules/unicode-properties": { @@ -9131,6 +9401,35 @@ "node": "^12.13.0 || ^14.15.0 || >=16.0.0" } }, + "node_modules/ws": { + "version": "8.21.1", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz", + "integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + }, + "node_modules/xmlhttprequest-ssl": { + "version": "2.1.2", + "resolved": "https://registry.npmjs.org/xmlhttprequest-ssl/-/xmlhttprequest-ssl-2.1.2.tgz", + "integrity": "sha512-TEU+nJVUUnA4CYJFLvK5X9AOeH4KvDvhIfm0vV1GaQRtchnG0hgK5p8hw/xjv8cunWYCsiPCSDzObPyhEwq3KQ==", + "engines": { + "node": ">=0.4.0" + } + }, "node_modules/xss": { "version": "1.0.15", "resolved": "https://registry.npmjs.org/xss/-/xss-1.0.15.tgz", diff --git a/stellar-payment-platform/package.json b/stellar-payment-platform/package.json index 0034d4b..560d47a 100644 --- a/stellar-payment-platform/package.json +++ b/stellar-payment-platform/package.json @@ -45,7 +45,12 @@ "winston": "^3.19.0", "winston-daily-rotate-file": "^5.0.0", "xss": "^1.0.15", - "zod": "^4.4.3" +<<<<<<< HEAD + "zod": "^4.4.3", +======= +>>>>>>> 4095f79 (feat(websockets): add Socket.IO server and horizon listener forwarding (fix #430)) + "socket.io": "^4.8.0", + "socket.io-client": "^4.8.0" }, "devDependencies": { "@prisma/client": "^6.19.3", diff --git a/stellar-payment-platform/server.js b/stellar-payment-platform/server.js index d1837c5..e40d723 100644 --- a/stellar-payment-platform/server.js +++ b/stellar-payment-platform/server.js @@ -895,7 +895,21 @@ const gracefulShutdown = (server, prismaClient, signal) => { if (require.main === module) { - const server = app.listen(PORT, '0.0.0.0', () => { + const http = require('http'); + const { initSocketServer } = require('./src/socketManager'); + + const server = http.createServer(app); + + // Initialize Socket.IO on the HTTP server so the same port serves both + // the API and WebSocket connections. + try { + initSocketServer(server); + logger.info('Socket.IO initialized'); + } catch (err) { + logger.error('Failed to initialize Socket.IO', err); + } + + server.listen(PORT, '0.0.0.0', () => { logger.info(`Server successfully initialized on port ${PORT}`); }); diff --git a/stellar-payment-platform/src/socketManager.js b/stellar-payment-platform/src/socketManager.js new file mode 100644 index 0000000..26a9c65 --- /dev/null +++ b/stellar-payment-platform/src/socketManager.js @@ -0,0 +1,75 @@ +const { Server } = require('socket.io'); +const { logger } = require('./logger'); + +let io = null; + +/** + * Initialize Socket.IO on the passed http.Server instance. + * Maintains a simple room-per-address mapping: clients call `authenticate` with + * { address } and are added to a room named by that address. Emitted payments + * are broadcast to that room. + */ +function initSocketServer(httpServer, corsOptions = {}) { + if (io) return io; + io = new Server(httpServer, { + cors: Object.assign({ origin: true, methods: ['GET', 'POST'] }, corsOptions), + }); + + io.on('connection', (socket) => { + logger.info(`Socket connected: ${socket.id}`); + + // The optional ack lets a client wait until it is actually subscribed. + // Without it a client that emits `authenticate` and immediately expects + // events can miss any payment that arrives before the room join lands. + socket.on('authenticate', async (payload, ack) => { + let subscribed = false; + try { + const address = payload && typeof payload.address === 'string' ? payload.address : null; + if (address) { + // socket.join can be async in some adapters; await to be safe. + await socket.join(address); + socket.address = address; + subscribed = true; + logger.info(`Socket ${socket.id} joined room for ${address}`); + } + } catch (err) { + logger.error('Socket authenticate error', err); + } + if (typeof ack === 'function') { + try { ack({ subscribed }); } catch (e) { /* ignore ack errors */ } + } + }); + + socket.on('payment', (payload) => { + try { + const addr = payload && typeof payload.address === 'string' ? payload.address : null; + const payment = payload && payload.payment ? payload.payment : payload; + if (addr) { + io.to(addr).emit('payment', payment); + } + } catch (err) { + logger.error('Error handling payment emit', err); + } + }); + + socket.on('disconnect', () => { + logger.info(`Socket disconnected: ${socket.id}`); + }); + }); + + return io; +} + +function emitToAddress(address, event, data) { + if (!io) return; + io.to(address).emit(event, data); +} + +function closeSocketServer() { + if (io) { + io.close(); + io = null; + } +} + +module.exports = { initSocketServer, emitToAddress, closeSocketServer }; diff --git a/stellar-payment-platform/tests/socket.test.js b/stellar-payment-platform/tests/socket.test.js new file mode 100644 index 0000000..42817b2 --- /dev/null +++ b/stellar-payment-platform/tests/socket.test.js @@ -0,0 +1,109 @@ +jest.setTimeout(20000); +jest.mock('../src/logger', () => ({ logger: { info: jest.fn(), warn: jest.fn(), error: jest.fn() } })); + +const http = require('http'); +const ioClient = require('socket.io-client'); +const { initSocketServer } = require('../src/socketManager'); + +let server; +let clientSocket; +let publisherSocket; +let port; + +function startSocketServer(done) { + // Create a minimal express app for the socket server so we don't pull in + // heavyweight dependencies (like @stellar/stellar-sdk) during tests. + const express = require('express'); + const app = express(); + app.get('/health', (_req, res) => res.json({ ok: true })); + + const httpServer = http.createServer(app); + initSocketServer(httpServer); + httpServer.listen(0, '127.0.0.1', () => { + port = httpServer.address().port; + server = httpServer; + done(); + }); +} + +function closeSocketServer() { + if (server && server.listening) server.close(); +} + +describe('Socket.IO real-time notification system', () => { + beforeAll((done) => { + startSocketServer(() => { + clientSocket = ioClient(`http://localhost:${port}`, { transports: ['websocket'] }); + clientSocket.on('connect', () => done()); + }); + }); + + afterAll((done) => { + // Ensure client sockets are disconnected before closing the server so + // server.close's callback fires promptly. + try { + if (clientSocket && clientSocket.connected) clientSocket.disconnect(); + if (publisherSocket && publisherSocket.connected) publisherSocket.disconnect(); + } catch (e) { /* ignore */ } + + if (server && server.listening) { + server.close(() => done()); + return; + } + done(); + }); + + afterEach(() => { + if (publisherSocket && publisherSocket.connected) publisherSocket.disconnect(); + publisherSocket = null; + }); + + test('should authenticate and join the address room and receive payment events', (done) => { + const testAddress = 'GAPUQZH3WZUXHEMUGZN5ZYU4D4GHCFEMOGUINU6MF345GBD2QXNYYIEQ'; + + clientSocket.on('payment', (paymentData) => { + try { + expect(paymentData).toEqual({ amount: '100', asset: 'XLM' }); + done(); + } catch (err) { + done(err); + } + }); + + // The ack fires after the server has joined the room, so the publish below + // cannot outrun the subscription. + clientSocket.emit('authenticate', { address: testAddress }, () => { + publisherSocket = ioClient(`http://localhost:${port}`, { transports: ['websocket'] }); + publisherSocket.on('connect', () => { + publisherSocket.emit('payment', { + address: testAddress, + payment: { amount: '100', asset: 'XLM' }, + }); + // Disconnecting here would close the transport before the packet is + // flushed and drop the event; afterEach tears the socket down instead. + }); + }); + }); + + test('does not deliver payments for an address the client did not subscribe to', (done) => { + const subscribed = 'GAPUQZH3WZUXHEMUGZN5ZYU4D4GHCFEMOGUINU6MF345GBD2QXNYYIEQ'; + const other = 'GBDQD3WTQ6W2VQ2W4V74UZ5WYF6B72GZ6EHD7I3L3WYH357Y4K5H3E4W'; + + clientSocket.on('payment', () => { + done(new Error('received a payment addressed to another account')); + }); + + clientSocket.emit('authenticate', { address: subscribed }, () => { + publisherSocket = ioClient(`http://localhost:${port}`, { transports: ['websocket'] }); + publisherSocket.on('connect', () => { + publisherSocket.emit('payment', { + address: other, + payment: { amount: '100', asset: 'XLM' }, + }); + // Round-trip through the same socket: once this ack returns the server + // has already handled the payment above, so nothing is still in flight. + publisherSocket.emit('authenticate', { address: other }, () => done()); + }); + }); + }); +});