diff --git a/src/services/airdrops.js b/src/services/airdrops.js index 2ebca52..9889720 100644 --- a/src/services/airdrops.js +++ b/src/services/airdrops.js @@ -60,7 +60,7 @@ async function create(data) { const redis = cache.getClient(); await cache.set(airdropKey(id), airdrop); - await redis.sadd(IDS_KEY, id); + await redis.zadd(IDS_KEY, Date.now(), id); if (recipients.length > 0) { await redis.lpush(recipientsKey(id), ...recipients.map((r) => JSON.stringify(r))); @@ -70,11 +70,11 @@ async function create(data) { } /** - * Pages through the full airdrop ID set via SSCAN instead of SMEMBERS. Used + * Pages through the full airdrop ID sorted set via ZSCAN instead of ZREVRANGE. Used * by the expiry reconciliation job (#88), which needs to visit every - * airdrop every cycle: SMEMBERS returns the whole set in one blocking call + * airdrop every cycle: ZREVRANGE with 0 -1 returns the whole set in one call * and would need it all held in memory at once, which doesn't scale as the - * set grows. SSCAN pages incrementally with a small, bounded cursor cost per + * set grows. ZSCAN pages incrementally with a small, bounded cursor cost per * call. `list()` above is unchanged — this is a separate, job-internal * scanning path, not a replacement for the paginated HTTP listing endpoint. */ @@ -82,8 +82,9 @@ async function* scanIds(batchSize = config.airdrops.expiryScanBatchSize) { const redis = cache.getClient(); let cursor = '0'; do { - const [nextCursor, batch] = await redis.sscan(IDS_KEY, cursor, 'COUNT', batchSize); + const [nextCursor, batchWithScores] = await redis.zscan(IDS_KEY, cursor, 'COUNT', batchSize); cursor = nextCursor; + const batch = batchWithScores.filter((_, index) => index % 2 === 0); if (batch.length > 0) { yield batch; } @@ -134,10 +135,10 @@ async function markExpired(id, currentLedger) { async function list(page = 1, limit = 20) { const redis = cache.getClient(); - const ids = await redis.smembers(IDS_KEY); + const total = await redis.zcard(IDS_KEY); const start = (page - 1) * limit; - const end = start + limit; - const paginatedIds = ids.slice(start, end); + const end = start + limit - 1; + const paginatedIds = await redis.zrevrange(IDS_KEY, start, end); const airdrops = await Promise.all(paginatedIds.map((id) => cache.get(airdropKey(id)))); return { @@ -145,8 +146,8 @@ async function list(page = 1, limit = 20) { pagination: { page, limit, - total: ids.length, - total_pages: Math.ceil(ids.length / limit), + total, + total_pages: Math.ceil(total / limit), }, }; } @@ -179,7 +180,7 @@ async function remove(id) { await cache.del(airdropKey(id)); await cache.del(recipientsKey(id)); - await redis.srem(IDS_KEY, id); + await redis.zrem(IDS_KEY, id); return existing; } diff --git a/src/services/alerts.js b/src/services/alerts.js index e1ebaef..5ba9320 100644 --- a/src/services/alerts.js +++ b/src/services/alerts.js @@ -51,23 +51,22 @@ async function create(data) { const redis = cache.getClient(); await cache.set(alertKey(id), alert); - await redis.sadd(IDS_KEY, id); + await redis.zadd(IDS_KEY, Date.now(), id); return alert; } async function list() { const redis = cache.getClient(); - const ids = await redis.smembers(IDS_KEY); + const ids = await redis.zrevrange(IDS_KEY, 0, -1); const alerts = await Promise.all(ids.map((id) => cache.get(alertKey(id)))); return alerts.filter(Boolean); } async function listPaginated({ offset = 0, limit = 20 } = {}) { const redis = cache.getClient(); - const ids = await redis.smembers(IDS_KEY); - const total = ids.length; - const paginatedIds = ids.slice(offset, offset + limit); + const total = await redis.zcard(IDS_KEY); + const paginatedIds = await redis.zrevrange(IDS_KEY, offset, offset + limit - 1); const alerts = await Promise.all( paginatedIds.map((id) => cache.get(alertKey(id))) ); @@ -82,7 +81,7 @@ async function remove(id) { const existing = await cache.get(alertKey(id)); if (!existing) return null; await cache.del(alertKey(id)); - await redis.srem(IDS_KEY, id); + await redis.zrem(IDS_KEY, id); return existing; } @@ -103,7 +102,7 @@ async function fire(alert, priceUsd) { async function evaluateForAsset(asset, priceUsd) { const redis = cache.getClient(); - const ids = await redis.smembers(IDS_KEY); + const ids = await redis.zrevrange(IDS_KEY, 0, -1); for (const id of ids) { const alert = await cache.get(alertKey(id)); diff --git a/src/services/apiKeys.js b/src/services/apiKeys.js index 1c68819..eaa1fa0 100644 --- a/src/services/apiKeys.js +++ b/src/services/apiKeys.js @@ -38,7 +38,7 @@ async function getKey(id) { async function listKeys() { const redis = cache.getClient(); - const ids = await redis.smembers(IDS_KEY); + const ids = await redis.zrevrange(IDS_KEY, 0, -1); const records = await Promise.all(ids.map((id) => getKey(id))); return records.filter(Boolean).map(sanitize); } @@ -60,7 +60,7 @@ async function createKey({ label, scopes = ['default'] }) { const redis = cache.getClient(); await cache.set(keyPath(record.id), record); await cache.set(hashPath(hashed), record.id); - await redis.sadd(IDS_KEY, record.id); + await redis.zadd(IDS_KEY, Date.now(), record.id); return { api_key: apiKey, @@ -75,7 +75,7 @@ async function revokeKey(id) { const redis = cache.getClient(); await cache.del(keyPath(id)); await cache.del(hashPath(record.key_hash)); - await redis.srem(IDS_KEY, id); + await redis.zrem(IDS_KEY, id); return sanitize(record); } diff --git a/test/airdrops-service.test.js b/test/airdrops-service.test.js index 1761516..d2fa9ca 100644 --- a/test/airdrops-service.test.js +++ b/test/airdrops-service.test.js @@ -2,6 +2,7 @@ const mockStore = new Map(); const mockSets = new Map(); +const mockZSets = new Map(); const mockLists = new Map(); // Faithfully mirrors MARK_EXPIRED_SCRIPT's condition/write logic in JS, @@ -22,6 +23,14 @@ function mockMarkExpiredEval(store, key, currentLedger, nowIso) { return JSON.stringify(updated); } +function getSortedZSetMembers(key) { + const z = mockZSets.get(key); + if (!z) return []; + return [...z.entries()] + .sort((a, b) => b[1] - a[1]) + .map(([member]) => member); +} + const mockRedis = { smembers: jest.fn(async (key) => [...(mockSets.get(key) || [])]), sadd: jest.fn(async (key, val) => { @@ -41,6 +50,31 @@ const mockRedis = { const nextCursor = start + count >= members.length ? '0' : String(start + count); return [nextCursor, batch]; }), + zadd: jest.fn(async (key, score, member) => { + if (!mockZSets.has(key)) mockZSets.set(key, new Map()); + mockZSets.get(key).set(member, Number(score)); + }), + zrem: jest.fn(async (key, ...members) => { + const z = mockZSets.get(key); + if (!z) return; + for (const m of members) z.delete(m); + }), + zrevrange: jest.fn(async (key, start, stop) => { + const sorted = getSortedZSetMembers(key); + const end = stop === -1 ? sorted.length : stop + 1; + return sorted.slice(start, end); + }), + zcard: jest.fn(async (key) => (mockZSets.get(key)?.size || 0)), + zscan: jest.fn(async (key, cursor, _countKeyword, count) => { + const entries = [...(mockZSets.get(key)?.entries() || [])]; + const batchWithScores = []; + const start = Number(cursor); + for (let i = start; i < start + count && i < entries.length; i++) { + batchWithScores.push(entries[i][0], entries[i][1]); + } + const nextCursor = start + count >= entries.length ? '0' : String(start + count); + return [nextCursor, batchWithScores]; + }), llen: jest.fn(async (key) => (mockLists.get(key) || []).length), lpush: jest.fn(async (key, ...vals) => { if (!mockLists.has(key)) mockLists.set(key, []); @@ -108,6 +142,7 @@ const airdropsService = require('../src/services/airdrops'); beforeEach(() => { mockStore.clear(); mockSets.clear(); + mockZSets.clear(); mockLists.clear(); }); diff --git a/test/alerts.test.js b/test/alerts.test.js index 5196cee..3bd9a07 100644 --- a/test/alerts.test.js +++ b/test/alerts.test.js @@ -2,11 +2,35 @@ const mockStore = new Map(); const mockSets = new Map(); +const mockZSets = new Map(); + +function getSortedZSetMembers(key) { + const z = mockZSets.get(key); + if (!z) return []; + return [...z.entries()] + .sort((a, b) => b[1] - a[1]) + .map(([member]) => member); +} const mockRedis = { smembers: jest.fn(async (key) => [...(mockSets.get(key) || [])]), sadd: jest.fn(async (key, val) => { if (!mockSets.has(key)) mockSets.set(key, new Set()); mockSets.get(key).add(val); }), srem: jest.fn(async (key, val) => { mockSets.get(key)?.delete(val); }), + zadd: jest.fn(async (key, score, member) => { + if (!mockZSets.has(key)) mockZSets.set(key, new Map()); + mockZSets.get(key).set(member, Number(score)); + }), + zrem: jest.fn(async (key, ...members) => { + const z = mockZSets.get(key); + if (!z) return; + for (const m of members) z.delete(m); + }), + zrevrange: jest.fn(async (key, start, stop) => { + const sorted = getSortedZSetMembers(key); + const end = stop === -1 ? sorted.length : stop + 1; + return sorted.slice(start, end); + }), + zcard: jest.fn(async (key) => (mockZSets.get(key)?.size || 0)), }; jest.mock('../src/services/cache', () => ({ @@ -35,6 +59,7 @@ const cache = require('../src/services/cache'); beforeEach(() => { mockStore.clear(); mockSets.clear(); + mockZSets.clear(); mockWebhookDeliver.mockClear(); cache.get.mockClear(); cache.set.mockClear();