diff --git a/tests/unit/handlers/pools.handlers.test.js b/tests/unit/handlers/pools.handlers.test.js new file mode 100644 index 0000000..1b1e6dd --- /dev/null +++ b/tests/unit/handlers/pools.handlers.test.js @@ -0,0 +1,137 @@ +'use strict' + +const test = require('brittle') +const { + getPools, + flattenPoolStats, + calculatePoolsSummary +} = require('../../../workers/lib/server/handlers/pools.handlers') + +test('getPools - happy path', async (t) => { + const mockCtx = { + conf: { + orks: [{ rpcPublicKey: 'key1' }] + }, + net_r0: { + jRequest: async (key, method, payload) => { + if (method === 'getWrkExtData') { + return [{ + ts: '1770000000000', + stats: [ + { poolType: 'f2pool', username: 'user1', hashrate: 100000, worker_count: 5, balance: 0.5, timestamp: 1770000000000 }, + { poolType: 'ocean', username: 'user2', hashrate: 200000, worker_count: 10, balance: 1.2, timestamp: 1770000000000 } + ] + }] + } + return [] + } + } + } + + const mockReq = { query: {} } + const result = await getPools(mockCtx, mockReq, {}) + t.ok(result.pools, 'should return pools array') + t.ok(result.summary, 'should return summary') + t.ok(Array.isArray(result.pools), 'pools should be array') + t.is(result.pools.length, 2, 'should have 2 pools') + t.is(result.summary.poolCount, 2, 'summary should count 2 pools') + t.pass() +}) + +test('getPools - with filter', async (t) => { + const mockCtx = { + conf: { orks: [{ rpcPublicKey: 'key1' }] }, + net_r0: { + jRequest: async () => [{ + ts: '1770000000000', + stats: [ + { poolType: 'f2pool', username: 'user1', hashrate: 100000, worker_count: 5, balance: 0.5 }, + { poolType: 'ocean', username: 'user2', hashrate: 200000, worker_count: 10, balance: 0 } + ] + }] + } + } + + const mockReq = { query: { query: '{"pool":"f2pool"}' } } + const result = await getPools(mockCtx, mockReq, {}) + t.ok(result.pools, 'should return filtered pools') + t.is(result.pools.length, 1, 'should have 1 pool after filter') + t.is(result.pools[0].pool, 'f2pool', 'should match filter') + t.pass() +}) + +test('getPools - empty ork results', async (t) => { + const mockCtx = { + conf: { orks: [{ rpcPublicKey: 'key1' }] }, + net_r0: { jRequest: async () => ([]) } + } + + const result = await getPools(mockCtx, { query: {} }, {}) + t.ok(result.pools, 'should return pools array') + t.is(result.pools.length, 0, 'pools should be empty') + t.is(result.summary.poolCount, 0, 'pool count should be 0') + t.pass() +}) + +test('flattenPoolStats - extracts pools from ext-data stats array', (t) => { + const results = [ + [{ + ts: '1770000000000', + stats: [ + { poolType: 'f2pool', username: 'user1', hashrate: 100000, worker_count: 5, balance: 0.5 }, + { poolType: 'ocean', username: 'user2', hashrate: 200000, worker_count: 10, balance: 1.2 } + ] + }] + ] + const pools = flattenPoolStats(results) + t.is(pools.length, 2, 'should extract 2 pools') + t.is(pools[0].pool, 'f2pool', 'should have correct pool type') + t.is(pools[0].account, 'user1', 'should have correct account') + t.is(pools[0].hashrate, 100000, 'should have correct hashrate') + t.is(pools[1].pool, 'ocean', 'should have correct pool type') + t.pass() +}) + +test('flattenPoolStats - deduplicates pools across orks', (t) => { + const results = [ + [{ ts: '1770000000000', stats: [{ poolType: 'f2pool', username: 'user1', hashrate: 100 }] }], + [{ ts: '1770000000000', stats: [{ poolType: 'f2pool', username: 'user1', hashrate: 200 }] }] + ] + const pools = flattenPoolStats(results) + t.is(pools.length, 1, 'should deduplicate by poolType:username') + t.pass() +}) + +test('flattenPoolStats - handles error results', (t) => { + const results = [{ error: 'timeout' }] + const pools = flattenPoolStats(results) + t.is(pools.length, 0, 'should be empty') + t.pass() +}) + +test('flattenPoolStats - handles non-array input', (t) => { + const pools = flattenPoolStats(null) + t.is(pools.length, 0, 'should return empty array') + t.pass() +}) + +test('calculatePoolsSummary - calculates totals', (t) => { + const pools = [ + { hashrate: 100, workerCount: 5, balance: 50000 }, + { hashrate: 200, workerCount: 10, balance: 30000 } + ] + + const summary = calculatePoolsSummary(pools) + t.is(summary.poolCount, 2, 'should count pools') + t.is(summary.totalHashrate, 300, 'should sum hashrate') + t.is(summary.totalWorkers, 15, 'should sum workers') + t.is(summary.totalBalance, 80000, 'should sum balance') + t.pass() +}) + +test('calculatePoolsSummary - handles empty pools', (t) => { + const summary = calculatePoolsSummary([]) + t.is(summary.poolCount, 0, 'should be zero') + t.is(summary.totalHashrate, 0, 'should be zero') + t.pass() +}) diff --git a/tests/unit/routes/pools.routes.test.js b/tests/unit/routes/pools.routes.test.js new file mode 100644 index 0000000..0e77076 --- /dev/null +++ b/tests/unit/routes/pools.routes.test.js @@ -0,0 +1,39 @@ +'use strict' + +const test = require('brittle') +const { testModuleStructure, testHandlerFunctions, testOnRequestFunctions } = require('../helpers/routeTestHelpers') +const { createRoutesForTest } = require('../helpers/mockHelpers') + +const ROUTES_PATH = '../../../workers/lib/server/routes/pools.routes.js' + +test('pools routes - module structure', (t) => { + testModuleStructure(t, ROUTES_PATH, 'pools') + t.pass() +}) + +test('pools routes - route definitions', (t) => { + const routes = createRoutesForTest(ROUTES_PATH) + const routeUrls = routes.map(route => route.url) + t.ok(routeUrls.includes('/auth/pools'), 'should have pools route') + t.pass() +}) + +test('pools routes - HTTP methods', (t) => { + const routes = createRoutesForTest(ROUTES_PATH) + routes.forEach(route => { + t.is(route.method, 'GET', `route ${route.url} should be GET`) + }) + t.pass() +}) + +test('pools routes - handler functions', (t) => { + const routes = createRoutesForTest(ROUTES_PATH) + testHandlerFunctions(t, routes, 'pools') + t.pass() +}) + +test('pools routes - onRequest functions', (t) => { + const routes = createRoutesForTest(ROUTES_PATH) + testOnRequestFunctions(t, routes, 'pools') + t.pass() +}) diff --git a/workers/lib/constants.js b/workers/lib/constants.js index f1f407e..95530ab 100644 --- a/workers/lib/constants.js +++ b/workers/lib/constants.js @@ -108,7 +108,10 @@ const ENDPOINTS = { THING_CONFIG: '/auth/thing-config', // WebSocket endpoint - WEBSOCKET: '/ws' + WEBSOCKET: '/ws', + + // Pools endpoints + POOLS: '/auth/pools' } const HTTP_METHODS = { @@ -183,6 +186,16 @@ const STATUS_CODES = { INTERNAL_SERVER_ERROR: 500 } +const RPC_METHODS = { + LIST_THINGS: 'listThings', + GET_WRK_EXT_DATA: 'getWrkExtData' +} + +const MINERPOOL_EXT_DATA_KEYS = { + TRANSACTIONS: 'transactions', + STATS: 'stats' +} + const RPC_TIMEOUT = 15000 const RPC_CONCURRENCY_LIMIT = 2 @@ -202,5 +215,7 @@ module.exports = { STATUS_CODES, RPC_TIMEOUT, RPC_CONCURRENCY_LIMIT, - USER_SETTINGS_TYPE + USER_SETTINGS_TYPE, + RPC_METHODS, + MINERPOOL_EXT_DATA_KEYS } diff --git a/workers/lib/server/handlers/pools.handlers.js b/workers/lib/server/handlers/pools.handlers.js new file mode 100644 index 0000000..593c170 --- /dev/null +++ b/workers/lib/server/handlers/pools.handlers.js @@ -0,0 +1,97 @@ +'use strict' + +const mingo = require('mingo') +const { + RPC_METHODS, + MINERPOOL_EXT_DATA_KEYS +} = require('../../constants') +const { + requestRpcMapLimit, + parseJsonQueryParam +} = require('../../utils') + +async function getPools (ctx, req) { + const filter = req.query.query ? parseJsonQueryParam(req.query.query, 'ERR_QUERY_INVALID_JSON') : null + const sort = req.query.sort ? parseJsonQueryParam(req.query.sort, 'ERR_SORT_INVALID_JSON') : null + const fields = req.query.fields ? parseJsonQueryParam(req.query.fields, 'ERR_FIELDS_INVALID_JSON') : null + + const statsResults = await requestRpcMapLimit(ctx, RPC_METHODS.GET_WRK_EXT_DATA, { + type: 'minerpool', + query: { key: MINERPOOL_EXT_DATA_KEYS.STATS } + }) + + const pools = flattenPoolStats(statsResults) + + const query = new mingo.Query(filter || {}) + let cursor = query.find(pools, fields || {}) + if (sort) cursor = cursor.sort(sort) + const result = cursor.all() + + const summary = calculatePoolsSummary(pools) + + return { pools: result, summary } +} + +function flattenPoolStats (results) { + const pools = [] + const seen = new Set() + if (!Array.isArray(results)) return pools + + for (const orkResult of results) { + if (!orkResult || orkResult.error) continue + const items = Array.isArray(orkResult) ? orkResult : (orkResult.data || orkResult.result || []) + if (!Array.isArray(items)) continue + + for (const item of items) { + if (!item) continue + const stats = item.stats || item.data || [] + if (!Array.isArray(stats)) continue + + for (const stat of stats) { + if (!stat) continue + const poolKey = `${stat.poolType}:${stat.username}` + if (seen.has(poolKey)) continue + seen.add(poolKey) + + pools.push({ + name: stat.username || stat.poolType, + pool: stat.poolType, + account: stat.username, + status: 'active', + hashrate: stat.hashrate || 0, + hashrate1h: stat.hashrate_1h || 0, + hashrate24h: stat.hashrate_24h || 0, + workerCount: stat.worker_count || 0, + activeWorkerCount: stat.active_workers_count || 0, + balance: stat.balance || 0, + unsettled: stat.unsettled || 0, + revenue24h: stat.revenue_24h || stat.estimated_today_income || 0, + yearlyBalances: stat.yearlyBalances || [], + lastUpdated: stat.timestamp || null + }) + } + } + } + + return pools +} + +function calculatePoolsSummary (pools) { + const totals = pools.reduce((acc, pool) => { + acc.totalHashrate += pool.hashrate || 0 + acc.totalWorkers += pool.workerCount || pool.worker_count || 0 + acc.totalBalance += pool.balance || 0 + return acc + }, { totalHashrate: 0, totalWorkers: 0, totalBalance: 0 }) + + return { + poolCount: pools.length, + ...totals + } +} + +module.exports = { + getPools, + flattenPoolStats, + calculatePoolsSummary +} diff --git a/workers/lib/server/index.js b/workers/lib/server/index.js index ac884da..9ce3cab 100644 --- a/workers/lib/server/index.js +++ b/workers/lib/server/index.js @@ -8,6 +8,7 @@ const globalRoutes = require('./routes/global.routes') const thingsRoutes = require('./routes/things.routes') const settingsRoutes = require('./routes/settings.routes') const wsRoutes = require('./routes/ws.routes') +const poolsRoutes = require('./routes/pools.routes') /** * Collect all routes into a flat array for server injection. @@ -22,7 +23,8 @@ function routes (ctx) { ...thingsRoutes(ctx), ...usersRoutes(ctx), ...settingsRoutes(ctx), - ...wsRoutes(ctx) + ...wsRoutes(ctx), + ...poolsRoutes(ctx) ] } diff --git a/workers/lib/server/routes/pools.routes.js b/workers/lib/server/routes/pools.routes.js new file mode 100644 index 0000000..64b982c --- /dev/null +++ b/workers/lib/server/routes/pools.routes.js @@ -0,0 +1,35 @@ +'use strict' + +const { + ENDPOINTS, + HTTP_METHODS +} = require('../../constants') +const { + getPools +} = require('../handlers/pools.handlers') +const { createCachedAuthRoute } = require('../lib/routeHelpers') + +module.exports = (ctx) => { + const schemas = require('../schemas/pools.schemas.js') + + return [ + { + method: HTTP_METHODS.GET, + url: ENDPOINTS.POOLS, + schema: { + querystring: schemas.query.pools + }, + ...createCachedAuthRoute( + ctx, + (req) => [ + 'pools', + req.query.query, + req.query.sort, + req.query.fields + ], + ENDPOINTS.POOLS, + getPools + ) + } + ] +} diff --git a/workers/lib/server/schemas/pools.schemas.js b/workers/lib/server/schemas/pools.schemas.js new file mode 100644 index 0000000..4dae736 --- /dev/null +++ b/workers/lib/server/schemas/pools.schemas.js @@ -0,0 +1,17 @@ +'use strict' + +const schemas = { + query: { + pools: { + type: 'object', + properties: { + query: { type: 'string' }, + sort: { type: 'string' }, + fields: { type: 'string' }, + overwriteCache: { type: 'boolean' } + } + } + } +} + +module.exports = schemas diff --git a/workers/lib/utils.js b/workers/lib/utils.js index 886ae5f..180b8a7 100644 --- a/workers/lib/utils.js +++ b/workers/lib/utils.js @@ -128,6 +128,14 @@ const requestRpcMapLimit = async (ctx, method, payload) => { }) } +const runParallel = (tasks) => + new Promise((resolve, reject) => { + async.parallel(tasks, (err, results) => { + if (err) reject(err) + else resolve(results) + }) + }) + module.exports = { dateNowSec, extractIps, @@ -137,5 +145,6 @@ module.exports = { getAuthTokenFromHeaders, parseJsonQueryParam, requestRpcEachLimit, - requestRpcMapLimit + requestRpcMapLimit, + runParallel }