diff --git a/README.md b/README.md index bb0048e..d4bd5b5 100644 --- a/README.md +++ b/README.md @@ -437,6 +437,7 @@ Express tip: capture the raw body via `express.json({ verify: (req, _res, buf) = - **Retryable**: network errors, HTTP 5xx, 408, 429. - **Not retried**: HTTP 4xx (except 408/429). These are marked `failed` immediately so a misconfigured consumer cannot be retried into the ground. - Each delivery is logged in `webhook_deliveries` (Redis-backed today, drop-in PG migration documented in `src/repositories/deliveryRepository.js`). +- **Safe for multiple replicas**: `webhookRetryWorker` claims due retries via `deliveryRepository.popDueRetries`, which uses a single atomic Redis Lua script (`ZRANGEBYSCORE` + `ZREM` in one round trip) rather than two separate calls. Running N instances of this backend against the same Redis is safe - each due retry is claimed by exactly one instance, so a delivery is never dispatched twice for the same retry. The worker's in-process `running` flag only guards against a single process overlapping with itself; cross-replica safety comes from the atomic claim, not from that flag. ### Storage model diff --git a/src/config.js b/src/config.js index e3fbf23..f4f8002 100644 --- a/src/config.js +++ b/src/config.js @@ -1,258 +1,155 @@ -require('dotenv').config(); - -const { cleanEnv, makeValidator, num, port, str, url } = require('envalid'); - -const stellarAddress = makeValidator((input) => { - if (!/^G[A-Z0-9]{55}$/.test(input)) { - throw new Error('must be a valid Stellar public key'); - } - return input; -}); - -const databaseDevDefault = - process.env.NODE_ENV === 'test' - ? 'postgres://localhost/smartdrop_test' - : 'postgres://localhost/smartdrop'; - -const rawEnv = { - ...process.env, - NODE_ENV: process.env.NODE_ENV || 'development', -}; - -const env = cleanEnv(rawEnv, { - NODE_ENV: str({ - default: 'development', - choices: ['development', 'test', 'production'], - }), - PORT: port({ default: 3000 }), - REDIS_URL: url({ devDefault: 'redis://localhost:6379' }), - DATABASE_URL: url({ devDefault: databaseDevDefault }), - STELLAR_HORIZON_URL: url({ default: 'https://horizon.stellar.org' }), - USDC_ISSUER: stellarAddress({ - default: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335AX2OBFLDTQLNUEHRGPTM6RIA', - }), - COINGECKO_API_KEY: str({ default: '' }), - COINMARKETCAP_API_KEY: str({ default: '' }), - ADMIN_API_KEY: str({ default: '' }), - PRICE_CACHE_TTL_SECONDS: num({ default: 60 }), - PRICE_REFRESH_INTERVAL_SECONDS: num({ default: 30 }), - PRICE_STALE_THRESHOLD_MINUTES: num({ default: 5 }), - PRICE_ANOMALY_THRESHOLD_PCT: num({ default: 20 }), - LOG_LEVEL: str({ - default: 'info', - choices: ['debug', 'info', 'warn', 'error'], - }), -}); - -const usdcIssuer = env.USDC_ISSUER; - -module.exports = { - nodeEnv: env.NODE_ENV, - port: env.PORT, - databaseUrl: env.DATABASE_URL, - redis: { - url: env.REDIS_URL, - }, - stellar: { - horizonUrl: env.STELLAR_HORIZON_URL, - usdcIssuer, - }, - coingecko: { - apiKey: env.COINGECKO_API_KEY, - baseUrl: 'https://api.coingecko.com/api/v3', - }, - coinmarketcap: { - apiKey: env.COINMARKETCAP_API_KEY, - baseUrl: 'https://pro-api.coinmarketcap.com/v1', - assetIssuerMap: { - XLM: { symbol: 'XLM' }, - [`USDC:${usdcIssuer}`]: { id: 3408 }, - }, - }, - price: { - cacheTtl: env.PRICE_CACHE_TTL_SECONDS, - refreshInterval: env.PRICE_REFRESH_INTERVAL_SECONDS, - staleThresholdMinutes: env.PRICE_STALE_THRESHOLD_MINUTES, - anomalyThresholdPercent: env.PRICE_ANOMALY_THRESHOLD_PCT, - }, - auth: { - adminApiKey: env.ADMIN_API_KEY, - }, - corsAllowedOrigins: (process.env.CORS_ALLOWED_ORIGINS || 'http://localhost:3000,http://localhost:3001') - .split(',') - .map((o) => o.trim()) - .filter(Boolean), - rateLimit: { - windowMs: parseInt(process.env.RATE_LIMIT_WINDOW_MS, 10) || 60000, - max: parseInt(process.env.RATE_LIMIT_MAX, 10) || 100, - }, - priceRateLimit: { - windowSeconds: parseInt(process.env.PRICE_RATELIMIT_WINDOW, 10) || 60, - max: parseInt(process.env.PRICE_RATELIMIT_MAX, 10) || 30, - }, - webhooks: { - maxAttempts: parseInt(process.env.WEBHOOK_MAX_ATTEMPTS, 10) || 3, - retryBaseMs: parseInt(process.env.WEBHOOK_RETRY_BASE_MS, 10) || 30000, - retryFactor: parseFloat(process.env.WEBHOOK_RETRY_FACTOR) || 2, - timeoutMs: parseInt(process.env.WEBHOOK_TIMEOUT_MS, 10) || 5000, - retryPollMs: parseInt(process.env.WEBHOOK_RETRY_POLL_MS, 10) || 5000, - retryBatchSize: parseInt(process.env.WEBHOOK_RETRY_BATCH, 10) || 25, - rateLimit: { - windowSeconds: parseInt(process.env.WEBHOOK_RATELIMIT_WINDOW, 10) || 60, - max: parseInt(process.env.WEBHOOK_RATELIMIT_MAX, 10) || 60, - }, - testRateLimit: { - windowSeconds: parseInt(process.env.WEBHOOK_TEST_RATELIMIT_WINDOW, 10) || 60, - max: parseInt(process.env.WEBHOOK_TEST_RATELIMIT_MAX, 10) || 5, - }, - }, -}; -require('dotenv').config(); - -const { cleanEnv, makeValidator, num, port, str, url } = require('envalid'); - -const stellarAddress = makeValidator((input) => { - if (!/^G[A-Z0-9]{55}$/.test(input)) { - throw new Error('must be a valid Stellar public key'); - } - return input; -}); - -const positiveInteger = makeValidator((input) => { - const value = Number(input); - if (!Number.isSafeInteger(value) || value <= 0) { - throw new Error('must be a positive integer'); - } - return value; -}); - -const databaseDevDefault = - process.env.NODE_ENV === 'test' - ? 'postgres://localhost/smartdrop_test' - : 'postgres://localhost/smartdrop'; - -const rawEnv = { - ...process.env, - NODE_ENV: process.env.NODE_ENV || 'development', -}; - -const env = cleanEnv(rawEnv, { - NODE_ENV: str({ - default: 'development', - choices: ['development', 'test', 'production'], - }), - PORT: port({ default: 3000 }), - REDIS_URL: url({ devDefault: 'redis://localhost:6379' }), - DATABASE_URL: url({ devDefault: databaseDevDefault }), - STELLAR_HORIZON_URL: url({ default: 'https://horizon.stellar.org' }), - USDC_ISSUER: stellarAddress({ - default: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335AX2OBFLDTQLNUEHRGPTM6RIA', - }), - COINGECKO_API_KEY: str({ default: '' }), - COINMARKETCAP_API_KEY: str({ default: '' }), - ADMIN_API_KEY: str({ default: '' }), - AIRDROP_CSV_MAX_BYTES: positiveInteger({ default: 5 * 1024 * 1024 }), - AIRDROP_JSON_MAX_BYTES: positiveInteger({ default: 2 * 1024 * 1024 }), - AIRDROP_RATELIMIT_WINDOW: positiveInteger({ default: 60 }), - AIRDROP_RATELIMIT_MAX: positiveInteger({ default: 10 }), - PRICE_CACHE_TTL_SECONDS: num({ default: 60 }), - PRICE_REFRESH_INTERVAL_SECONDS: num({ default: 30 }), - PRICE_STALE_THRESHOLD_MINUTES: num({ default: 5 }), - PRICE_ANOMALY_THRESHOLD_PCT: num({ default: 20 }), - PRICE_SOURCE_CIRCUIT_COOLDOWN_MS: num({ default: 15 * 60 * 1000 }), - PRICE_SOURCE_CIRCUIT_REMINDER_MS: num({ default: 5 * 60 * 1000 }), - AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS: num({ default: 60 }), - AIRDROP_LEDGER_CACHE_TTL_MS: num({ default: 5000 }), - AIRDROP_EXPIRY_SCAN_BATCH_SIZE: num({ default: 100 }), - LOG_LEVEL: str({ - default: 'info', - choices: ['debug', 'info', 'warn', 'error'], - }), -}); - -const usdcIssuer = env.USDC_ISSUER; - -module.exports = { - nodeEnv: env.NODE_ENV, - port: env.PORT, - databaseUrl: env.DATABASE_URL, - redis: { - url: env.REDIS_URL, - }, - stellar: { - horizonUrl: env.STELLAR_HORIZON_URL, - usdcIssuer, - }, - coingecko: { - apiKey: env.COINGECKO_API_KEY, - baseUrl: 'https://api.coingecko.com/api/v3', - }, - coinmarketcap: { - apiKey: env.COINMARKETCAP_API_KEY, - baseUrl: 'https://pro-api.coinmarketcap.com/v1', - assetIssuerMap: { - XLM: { symbol: 'XLM' }, - [`USDC:${usdcIssuer}`]: { id: 3408 }, - }, - }, - price: { - cacheTtl: env.PRICE_CACHE_TTL_SECONDS, - refreshInterval: env.PRICE_REFRESH_INTERVAL_SECONDS, - staleThresholdMinutes: env.PRICE_STALE_THRESHOLD_MINUTES, - anomalyThresholdPercent: env.PRICE_ANOMALY_THRESHOLD_PCT, - }, - priceSources: { - // How long a source's circuit stays open after a nonRetryable (e.g. 401) - // failure before it's attempted again. - circuitCooldownMs: env.PRICE_SOURCE_CIRCUIT_COOLDOWN_MS, - // Minimum gap between repeated "circuit open, skipping" log lines while - // the circuit stays open, so a misconfigured key doesn't spam one log - // line per fetch cycle for the entire cooldown window. - circuitReminderIntervalMs: env.PRICE_SOURCE_CIRCUIT_REMINDER_MS, - }, - airdrops: { - // How often the expiry reconciliation job scans non-terminal airdrops - // against the live Horizon ledger sequence. - expiryCheckIntervalSeconds: env.AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS, - // getCurrentLedger() is a live Horizon call with no caching; a job that - // polls frequently should reuse the same ledger sequence for this long - // rather than hitting Horizon once per airdrop per cycle. - ledgerCacheTtlMs: env.AIRDROP_LEDGER_CACHE_TTL_MS, - // SSCAN batch size used when scanning the full airdrop ID set — keeps - // each Redis round-trip small instead of loading the whole set (SMEMBERS) - // into memory at once. - expiryScanBatchSize: env.AIRDROP_EXPIRY_SCAN_BATCH_SIZE, - }, - auth: { - adminApiKey: env.ADMIN_API_KEY, - }, - airdrops: { - csvMaxBytes: env.AIRDROP_CSV_MAX_BYTES, - jsonMaxBytes: env.AIRDROP_JSON_MAX_BYTES, - maxRecipients: 10000, - rateLimit: { - windowSeconds: env.AIRDROP_RATELIMIT_WINDOW, - max: env.AIRDROP_RATELIMIT_MAX, - }, - }, - corsAllowedOrigins: (process.env.CORS_ALLOWED_ORIGINS || 'http://localhost:3000,http://localhost:3001') - .split(',') - .map((o) => o.trim()) - .filter(Boolean), - webhooks: { - maxAttempts: parseInt(process.env.WEBHOOK_MAX_ATTEMPTS, 10) || 3, - retryBaseMs: parseInt(process.env.WEBHOOK_RETRY_BASE_MS, 10) || 30000, - retryFactor: parseFloat(process.env.WEBHOOK_RETRY_FACTOR) || 2, - timeoutMs: parseInt(process.env.WEBHOOK_TIMEOUT_MS, 10) || 5000, - retryPollMs: parseInt(process.env.WEBHOOK_RETRY_POLL_MS, 10) || 5000, - retryBatchSize: parseInt(process.env.WEBHOOK_RETRY_BATCH, 10) || 25, - rateLimit: { - windowSeconds: parseInt(process.env.WEBHOOK_RATELIMIT_WINDOW, 10) || 60, - max: parseInt(process.env.WEBHOOK_RATELIMIT_MAX, 10) || 60, - }, - testRateLimit: { - windowSeconds: parseInt(process.env.WEBHOOK_TEST_RATELIMIT_WINDOW, 10) || 60, - max: parseInt(process.env.WEBHOOK_TEST_RATELIMIT_MAX, 10) || 5, - }, - }, -}; +require('dotenv').config(); + +const { cleanEnv, makeValidator, num, port, str, url } = require('envalid'); + +const stellarAddress = makeValidator((input) => { + if (!/^G[A-Z0-9]{55}$/.test(input)) { + throw new Error('must be a valid Stellar public key'); + } + return input; +}); + +const positiveInteger = makeValidator((input) => { + const value = Number(input); + if (!Number.isSafeInteger(value) || value <= 0) { + throw new Error('must be a positive integer'); + } + return value; +}); + +const databaseDevDefault = + process.env.NODE_ENV === 'test' + ? 'postgres://localhost/smartdrop_test' + : 'postgres://localhost/smartdrop'; + +const rawEnv = { + ...process.env, + NODE_ENV: process.env.NODE_ENV || 'development', +}; + +const env = cleanEnv(rawEnv, { + NODE_ENV: str({ + default: 'development', + choices: ['development', 'test', 'production'], + }), + PORT: port({ default: 3000 }), + REDIS_URL: url({ devDefault: 'redis://localhost:6379' }), + DATABASE_URL: url({ devDefault: databaseDevDefault }), + STELLAR_HORIZON_URL: url({ default: 'https://horizon.stellar.org' }), + USDC_ISSUER: stellarAddress({ + default: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335AX2OBFLDTQLNUEHRGPTM6RIA', + }), + COINGECKO_API_KEY: str({ default: '' }), + COINMARKETCAP_API_KEY: str({ default: '' }), + ADMIN_API_KEY: str({ default: '' }), + AIRDROP_CSV_MAX_BYTES: positiveInteger({ default: 5 * 1024 * 1024 }), + AIRDROP_JSON_MAX_BYTES: positiveInteger({ default: 2 * 1024 * 1024 }), + AIRDROP_RATELIMIT_WINDOW: positiveInteger({ default: 60 }), + AIRDROP_RATELIMIT_MAX: positiveInteger({ default: 10 }), + PRICE_CACHE_TTL_SECONDS: num({ default: 60 }), + PRICE_REFRESH_INTERVAL_SECONDS: num({ default: 30 }), + PRICE_STALE_THRESHOLD_MINUTES: num({ default: 5 }), + PRICE_ANOMALY_THRESHOLD_PCT: num({ default: 20 }), + PRICE_SOURCE_CIRCUIT_COOLDOWN_MS: num({ default: 15 * 60 * 1000 }), + PRICE_SOURCE_CIRCUIT_REMINDER_MS: num({ default: 5 * 60 * 1000 }), + AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS: num({ default: 60 }), + AIRDROP_LEDGER_CACHE_TTL_MS: num({ default: 5000 }), + AIRDROP_EXPIRY_SCAN_BATCH_SIZE: num({ default: 100 }), + LOG_LEVEL: str({ + default: 'info', + choices: ['debug', 'info', 'warn', 'error'], + }), +}); + +const usdcIssuer = env.USDC_ISSUER; + +module.exports = { + nodeEnv: env.NODE_ENV, + port: env.PORT, + databaseUrl: env.DATABASE_URL, + redis: { + url: env.REDIS_URL, + }, + stellar: { + horizonUrl: env.STELLAR_HORIZON_URL, + usdcIssuer, + }, + coingecko: { + apiKey: env.COINGECKO_API_KEY, + baseUrl: 'https://api.coingecko.com/api/v3', + }, + coinmarketcap: { + apiKey: env.COINMARKETCAP_API_KEY, + baseUrl: 'https://pro-api.coinmarketcap.com/v1', + assetIssuerMap: { + XLM: { symbol: 'XLM' }, + [`USDC:${usdcIssuer}`]: { id: 3408 }, + }, + }, + price: { + cacheTtl: env.PRICE_CACHE_TTL_SECONDS, + refreshInterval: env.PRICE_REFRESH_INTERVAL_SECONDS, + staleThresholdMinutes: env.PRICE_STALE_THRESHOLD_MINUTES, + anomalyThresholdPercent: env.PRICE_ANOMALY_THRESHOLD_PCT, + }, + priceSources: { + // How long a source's circuit stays open after a nonRetryable (e.g. 401) + // failure before it's attempted again. + circuitCooldownMs: env.PRICE_SOURCE_CIRCUIT_COOLDOWN_MS, + // Minimum gap between repeated "circuit open, skipping" log lines while + // the circuit stays open, so a misconfigured key doesn't spam one log + // line per fetch cycle for the entire cooldown window. + circuitReminderIntervalMs: env.PRICE_SOURCE_CIRCUIT_REMINDER_MS, + }, + airdrops: { + // How often the expiry reconciliation job scans non-terminal airdrops + // against the live Horizon ledger sequence. + expiryCheckIntervalSeconds: env.AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS, + // getCurrentLedger() is a live Horizon call with no caching; a job that + // polls frequently should reuse the same ledger sequence for this long + // rather than hitting Horizon once per airdrop per cycle. + ledgerCacheTtlMs: env.AIRDROP_LEDGER_CACHE_TTL_MS, + // SSCAN batch size used when scanning the full airdrop ID set — keeps + // each Redis round-trip small instead of loading the whole set (SMEMBERS) + // into memory at once. + expiryScanBatchSize: env.AIRDROP_EXPIRY_SCAN_BATCH_SIZE, + csvMaxBytes: env.AIRDROP_CSV_MAX_BYTES, + jsonMaxBytes: env.AIRDROP_JSON_MAX_BYTES, + maxRecipients: 10000, + rateLimit: { + windowSeconds: env.AIRDROP_RATELIMIT_WINDOW, + max: env.AIRDROP_RATELIMIT_MAX, + }, + }, + auth: { + adminApiKey: env.ADMIN_API_KEY, + }, + corsAllowedOrigins: (process.env.CORS_ALLOWED_ORIGINS || 'http://localhost:3000,http://localhost:3001') + .split(',') + .map((o) => o.trim()) + .filter(Boolean), + rateLimit: { + windowMs: parseInt(process.env.RATE_LIMIT_WINDOW_MS, 10) || 60000, + max: parseInt(process.env.RATE_LIMIT_MAX, 10) || 100, + }, + priceRateLimit: { + windowSeconds: parseInt(process.env.PRICE_RATELIMIT_WINDOW, 10) || 60, + max: parseInt(process.env.PRICE_RATELIMIT_MAX, 10) || 30, + }, + webhooks: { + maxAttempts: parseInt(process.env.WEBHOOK_MAX_ATTEMPTS, 10) || 3, + retryBaseMs: parseInt(process.env.WEBHOOK_RETRY_BASE_MS, 10) || 30000, + retryFactor: parseFloat(process.env.WEBHOOK_RETRY_FACTOR) || 2, + timeoutMs: parseInt(process.env.WEBHOOK_TIMEOUT_MS, 10) || 5000, + retryPollMs: parseInt(process.env.WEBHOOK_RETRY_POLL_MS, 10) || 5000, + retryBatchSize: parseInt(process.env.WEBHOOK_RETRY_BATCH, 10) || 25, + rateLimit: { + windowSeconds: parseInt(process.env.WEBHOOK_RATELIMIT_WINDOW, 10) || 60, + max: parseInt(process.env.WEBHOOK_RATELIMIT_MAX, 10) || 60, + }, + testRateLimit: { + windowSeconds: parseInt(process.env.WEBHOOK_TEST_RATELIMIT_WINDOW, 10) || 60, + max: parseInt(process.env.WEBHOOK_TEST_RATELIMIT_MAX, 10) || 5, + }, + }, +}; diff --git a/src/index.js b/src/index.js index 0f76499..37252f7 100644 --- a/src/index.js +++ b/src/index.js @@ -1,220 +1,138 @@ -'use strict'; - -const express = require('express'); -const helmet = require('helmet'); -const config = require('./config'); -const logger = require('./logger'); -const cache = require('./services/cache'); -const priceRefreshJob = require('./jobs/priceRefresh'); -const webhookRetryWorker = require('./jobs/webhookRetryWorker'); -const buildCorsMiddleware = require('./middleware/cors'); -const buildRateLimit = require('./middleware/rateLimit'); -const { requestIdMiddleware } = require('./middleware/requestId'); -const { requireApiKey } = require('./middleware/auth'); -const { errorHandler, notFoundHandler } = require('./middleware/errorHandler'); -const pricesRouter = require('./routes/prices'); -const alertsRouter = require('./routes/alerts'); -const keysRouter = require('./routes/keys'); -const webhooksRouter = require('./routes/webhooks'); -const airdropsRouter = require('./routes/airdrops'); -const apiDocsRouter = require('./routes/apiDocs'); - -const priceWebSocket = require('./ws/priceWebSocket'); - -const app = express(); -let server; - -app.use(requestIdMiddleware); -app.use(helmet()); -app.use(buildCorsMiddleware(config.corsAllowedOrigins)); -app.use(express.json()); - -app.get('/health', (req, res) => { - const redisConnected = cache.isConnected(); - res.json({ - status: 'ok', - timestamp: new Date().toISOString(), - redis_connected: redisConnected, - redis_unavailable: !redisConnected, - }); -}); - -const globalApiLimit = buildRateLimit({ - windowSeconds: Math.floor(config.rateLimit.windowMs / 1000), - max: config.rateLimit.max, - keyPrefix: 'api', -}); - -app.use('/api/v1', globalApiLimit); -app.use('/api/v1', pricesRouter); -app.use('/api/v1', keysRouter); -app.use('/api/v1/alerts', requireApiKey()); -app.use('/api/v1', alertsRouter); -app.use('/api/v1', webhooksRouter); -app.use('/api/v1', airdropsRouter); -app.use('/api-docs', globalApiLimit); -app.use('/api-docs', apiDocsRouter); - -app.use(notFoundHandler); -app.use(errorHandler); - -function shutdown(signal) { - return async () => { - logger.info(`${signal} received, shutting down`); - priceRefreshJob.stop(); - webhookRetryWorker.stop(); - require('./ws/PriceSubscriptionManager').stopHeartbeat(); - if (server) server.close(); - await cache.disconnect(); - process.exit(0); - }; -} - -if (require.main === module) { - server = app.listen(config.port, () => { - logger.info(`SmartDrop backend running on port ${config.port}`); - priceWebSocket.attach(server); - priceRefreshJob.start(); - webhookRetryWorker.start(); - }); - - process.on('SIGTERM', shutdown('SIGTERM')); - process.on('SIGINT', shutdown('SIGINT')); -} - -module.exports = app; -module.exports.app = app; -module.exports.server = server || { - close(callback) { - if (callback) callback(); - }, -}; -'use strict'; - -const express = require('express'); -const helmet = require('helmet'); -const config = require('./config'); -const logger = require('./logger'); -const cache = require('./services/cache'); -const priceOracle = require('./services/priceOracle'); -const priceRefreshJob = require('./jobs/priceRefresh'); -const webhookRetryWorker = require('./jobs/webhookRetryWorker'); -const airdropExpiryJob = require('./jobs/airdropExpiry'); -const buildCorsMiddleware = require('./middleware/cors'); -const { requestIdMiddleware } = require('./middleware/requestId'); -const { requireApiKey } = require('./middleware/auth'); -const { errorHandler, notFoundHandler } = require('./middleware/errorHandler'); -const pricesRouter = require('./routes/prices'); -const alertsRouter = require('./routes/alerts'); -const keysRouter = require('./routes/keys'); -const webhooksRouter = require('./routes/webhooks'); -const airdropsRouter = require('./routes/airdrops'); -const apiDocsRouter = require('./routes/apiDocs'); - -const priceWebSocket = require('./ws/priceWebSocket'); - -const app = express(); -let server; - -app.use(requestIdMiddleware); -app.use(helmet()); -app.use(buildCorsMiddleware(config.corsAllowedOrigins)); -app.use(express.json({ limit: config.airdrops.jsonMaxBytes })); - -app.get('/health', (req, res) => { - const redisConnected = cache.isConnected(); - const priceRefreshHealth = priceRefreshJob.getHealth(); - const webhookWorkerHealth = webhookRetryWorker.getHealth(); - - // Compute overall status: - // unhealthy – Redis is down, or a job is stalled past its grace period - // degraded – a job has not yet run but is still within its startup grace period - // ok – all dependencies healthy - let status = 'ok'; - if (!redisConnected || !priceRefreshHealth.healthy || !webhookWorkerHealth.healthy) { - // Distinguish between "never started" (degraded) vs outright stalled/down (unhealthy) - const jobsDegraded = - (!priceRefreshHealth.healthy && !priceRefreshHealth.stalled) || - (!webhookWorkerHealth.healthy && !webhookWorkerHealth.stalled); - status = (!redisConnected || priceRefreshHealth.stalled || webhookWorkerHealth.stalled) - ? 'unhealthy' - : jobsDegraded ? 'degraded' : 'unhealthy'; - } - - res.json({ - status, - timestamp: new Date().toISOString(), - redis: { - connected: redisConnected, - }, - jobs: { - price_refresh: { - healthy: priceRefreshHealth.healthy, - last_success_at: priceRefreshHealth.lastSuccessAt - ? new Date(priceRefreshHealth.lastSuccessAt).toISOString() - : null, - last_error: priceRefreshHealth.lastError, - stalled: priceRefreshHealth.stalled, - }, - webhook_retry_worker: { - healthy: webhookWorkerHealth.healthy, - last_success_at: webhookWorkerHealth.lastSuccessAt - ? new Date(webhookWorkerHealth.lastSuccessAt).toISOString() - : null, - last_error: webhookWorkerHealth.lastError, - stalled: webhookWorkerHealth.stalled, - }, - }, - database: { - configured: true, - checked: false, - status: 'unused', - }, - price_source_circuits: priceOracle.getSourceCircuitStates(), - }); -}); - -app.use('/api/v1', pricesRouter); -app.use('/api/v1', keysRouter); -app.use('/api/v1/alerts', requireApiKey()); -app.use('/api/v1', alertsRouter); -app.use('/api/v1', webhooksRouter); -app.use('/api/v1', airdropsRouter); -app.use('/api-docs', apiDocsRouter); - -app.use(notFoundHandler); -app.use(errorHandler); - -function shutdown(signal) { - return async () => { - logger.info(`${signal} received, shutting down`); - priceRefreshJob.stop(); - webhookRetryWorker.stop(); - airdropExpiryJob.stop(); - require('./ws/PriceSubscriptionManager').stopHeartbeat(); - if (server) server.close(); - await cache.disconnect(); - process.exit(0); - }; -} - -if (require.main === module) { - server = app.listen(config.port, () => { - logger.info(`SmartDrop backend running on port ${config.port}`); - priceWebSocket.attach(server); - priceRefreshJob.start(); - webhookRetryWorker.start(); - airdropExpiryJob.start(); - }); - - process.on('SIGTERM', shutdown('SIGTERM')); - process.on('SIGINT', shutdown('SIGINT')); -} - -module.exports = app; -module.exports.app = app; -module.exports.server = server || { - close(callback) { - if (callback) callback(); - }, -}; +'use strict'; + +const express = require('express'); +const helmet = require('helmet'); +const config = require('./config'); +const logger = require('./logger'); +const cache = require('./services/cache'); +const priceOracle = require('./services/priceOracle'); +const priceRefreshJob = require('./jobs/priceRefresh'); +const webhookRetryWorker = require('./jobs/webhookRetryWorker'); +const airdropExpiryJob = require('./jobs/airdropExpiry'); +const buildCorsMiddleware = require('./middleware/cors'); +const buildRateLimit = require('./middleware/rateLimit'); +const { requestIdMiddleware } = require('./middleware/requestId'); +const { requireApiKey } = require('./middleware/auth'); +const { errorHandler, notFoundHandler } = require('./middleware/errorHandler'); +const pricesRouter = require('./routes/prices'); +const alertsRouter = require('./routes/alerts'); +const keysRouter = require('./routes/keys'); +const webhooksRouter = require('./routes/webhooks'); +const airdropsRouter = require('./routes/airdrops'); +const apiDocsRouter = require('./routes/apiDocs'); + +const priceWebSocket = require('./ws/priceWebSocket'); + +const app = express(); +let server; + +app.use(requestIdMiddleware); +app.use(helmet()); +app.use(buildCorsMiddleware(config.corsAllowedOrigins)); +app.use(express.json({ limit: config.airdrops.jsonMaxBytes })); + +app.get('/health', (req, res) => { + const redisConnected = cache.isConnected(); + const priceRefreshHealth = priceRefreshJob.getHealth(); + const webhookWorkerHealth = webhookRetryWorker.getHealth(); + + // Compute overall status: + // unhealthy – Redis is down, or a job is stalled past its grace period + // degraded – a job has not yet run but is still within its startup grace period + // ok – all dependencies healthy + let status = 'ok'; + if (!redisConnected || !priceRefreshHealth.healthy || !webhookWorkerHealth.healthy) { + // Distinguish between "never started" (degraded) vs outright stalled/down (unhealthy) + const jobsDegraded = + (!priceRefreshHealth.healthy && !priceRefreshHealth.stalled) || + (!webhookWorkerHealth.healthy && !webhookWorkerHealth.stalled); + status = (!redisConnected || priceRefreshHealth.stalled || webhookWorkerHealth.stalled) + ? 'unhealthy' + : jobsDegraded ? 'degraded' : 'unhealthy'; + } + + res.json({ + status, + timestamp: new Date().toISOString(), + redis: { + connected: redisConnected, + }, + jobs: { + price_refresh: { + healthy: priceRefreshHealth.healthy, + last_success_at: priceRefreshHealth.lastSuccessAt + ? new Date(priceRefreshHealth.lastSuccessAt).toISOString() + : null, + last_error: priceRefreshHealth.lastError, + stalled: priceRefreshHealth.stalled, + }, + webhook_retry_worker: { + healthy: webhookWorkerHealth.healthy, + last_success_at: webhookWorkerHealth.lastSuccessAt + ? new Date(webhookWorkerHealth.lastSuccessAt).toISOString() + : null, + last_error: webhookWorkerHealth.lastError, + stalled: webhookWorkerHealth.stalled, + }, + }, + database: { + configured: true, + checked: false, + status: 'unused', + }, + price_source_circuits: priceOracle.getSourceCircuitStates(), + }); +}); + +const globalApiLimit = buildRateLimit({ + windowSeconds: Math.floor(config.rateLimit.windowMs / 1000), + max: config.rateLimit.max, + keyPrefix: 'api', +}); + +app.use('/api/v1', globalApiLimit); +app.use('/api/v1', pricesRouter); +app.use('/api/v1', keysRouter); +app.use('/api/v1/alerts', requireApiKey()); +app.use('/api/v1', alertsRouter); +app.use('/api/v1', webhooksRouter); +app.use('/api/v1', airdropsRouter); +app.use('/api-docs', globalApiLimit); +app.use('/api-docs', apiDocsRouter); + +app.use(notFoundHandler); +app.use(errorHandler); + +function shutdown(signal) { + return async () => { + logger.info(`${signal} received, shutting down`); + priceRefreshJob.stop(); + webhookRetryWorker.stop(); + airdropExpiryJob.stop(); + require('./ws/PriceSubscriptionManager').stopHeartbeat(); + if (server) server.close(); + await cache.disconnect(); + process.exit(0); + }; +} + +if (require.main === module) { + server = app.listen(config.port, () => { + logger.info(`SmartDrop backend running on port ${config.port}`); + priceWebSocket.attach(server); + priceRefreshJob.start(); + webhookRetryWorker.start(); + airdropExpiryJob.start(); + }); + + process.on('SIGTERM', shutdown('SIGTERM')); + process.on('SIGINT', shutdown('SIGINT')); +} + +module.exports = app; +module.exports.app = app; +module.exports.server = server || { + close(callback) { + if (callback) callback(); + }, +}; diff --git a/src/repositories/deliveryRepository.js b/src/repositories/deliveryRepository.js index e897f40..77c79f8 100644 --- a/src/repositories/deliveryRepository.js +++ b/src/repositories/deliveryRepository.js @@ -22,6 +22,15 @@ * Indexes that would back the queries below: * (webhook_id, created_at desc) - listing recent deliveries per webhook * (next_retry_at) - retry worker scan + * + * Atomicity: `popDueRetries` claims due retries from the `webhooks:retries` + * sorted set via a single Lua script (ZRANGEBYSCORE + ZREM in one round + * trip), registered on the ioredis client with `defineCommand`. Redis + * executes Lua scripts single-threaded to completion, so N instances of + * this backend calling `popDueRetries` concurrently against the same Redis + * always receive a disjoint set of ids - no delivery is ever claimed by + * more than one instance. This makes `webhookRetryWorker` safe to run on + * multiple replicas without duplicate delivery attempts. */ const crypto = require('crypto'); @@ -30,6 +39,23 @@ const cache = require('../services/cache'); const RETRY_QUEUE_KEY = 'webhooks:retries'; const RECENT_DELIVERIES_LIMIT = 100; +// Atomically claims up to ARGV[2] due members (score <= ARGV[1]) from the +// sorted set at KEYS[1] and removes them in the same round trip, so +// concurrent callers can never be handed overlapping ids. +const POP_DUE_RETRIES_LUA = ` +local ids = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'LIMIT', 0, ARGV[2]) +if #ids > 0 then + redis.call('ZREM', KEYS[1], unpack(ids)) +end +return ids +`; + +function ensurePopDueRetriesCommand(redis) { + if (typeof redis.popDueRetriesAtomic !== 'function') { + redis.defineCommand('popDueRetriesAtomic', { numberOfKeys: 1, lua: POP_DUE_RETRIES_LUA }); + } +} + function key(id) { return `webhook_delivery:${id}`; } @@ -92,10 +118,8 @@ async function scheduleRetry(deliveryId, nextRetryAtMs) { async function popDueRetries(nowMs, max = 25) { const redis = cache.getClient(); - const ids = await redis.zrangebyscore(RETRY_QUEUE_KEY, '-inf', nowMs, 'LIMIT', 0, max); - if (ids.length === 0) return []; - await redis.zrem(RETRY_QUEUE_KEY, ...ids); - return ids; + ensurePopDueRetriesCommand(redis); + return redis.popDueRetriesAtomic(RETRY_QUEUE_KEY, nowMs, max); } async function cancelRetry(deliveryId) { diff --git a/test/airdrops-service.test.js b/test/airdrops-service.test.js index d2fa9ca..99605f4 100644 --- a/test/airdrops-service.test.js +++ b/test/airdrops-service.test.js @@ -203,7 +203,7 @@ describe('airdrops service', () => { }); describe('scanIds (#88)', () => { - test('pages through every ID in the set across multiple SSCAN batches', async () => { + test('pages through every ID in the set across multiple ZSCAN batches', async () => { for (let i = 0; i < 5; i++) { await airdropsService.create({ name: `Airdrop ${i}`, @@ -221,9 +221,9 @@ describe('airdrops service', () => { expect(seen).toHaveLength(5); expect(new Set(seen).size).toBe(5); - // Confirms it actually paged (more than one SSCAN call for 5 items at - // batch size 2), not just a single SMEMBERS-style dump. - expect(mockRedis.sscan.mock.calls.length).toBeGreaterThan(1); + // Confirms it actually paged (more than one ZSCAN call for 5 items at + // batch size 2), not just a single ZREVRANGE-style dump. + expect(mockRedis.zscan.mock.calls.length).toBeGreaterThan(1); }); test('yields nothing for an empty airdrop set', async () => { diff --git a/test/airdrops.test.js b/test/airdrops.test.js index fae6506..d3cc6a6 100644 --- a/test/airdrops.test.js +++ b/test/airdrops.test.js @@ -2,6 +2,7 @@ const mockStore = new Map(); const mockSets = new Map(); +const mockZSets = new Map(); const mockLists = new Map(); const mockCounters = new Map(); @@ -14,6 +15,33 @@ const mockRedis = { 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 z = mockZSets.get(key); + if (!z) return []; + const sorted = [...z.entries()].sort((a, b) => b[1] - a[1]).map(([m]) => m); + 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 += 1) { + 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, []); @@ -87,6 +115,7 @@ beforeAll(() => { beforeEach(() => { mockStore.clear(); mockSets.clear(); + mockZSets.clear(); mockLists.clear(); mockCounters.clear(); cache.get.mockClear(); @@ -95,6 +124,11 @@ beforeEach(() => { mockRedis.smembers.mockClear(); mockRedis.sadd.mockClear(); mockRedis.srem.mockClear(); + mockRedis.zadd.mockClear(); + mockRedis.zrem.mockClear(); + mockRedis.zrevrange.mockClear(); + mockRedis.zcard.mockClear(); + mockRedis.zscan.mockClear(); mockRedis.llen.mockClear(); mockRedis.lpush.mockClear(); mockRedis.rpush.mockClear(); @@ -122,10 +156,6 @@ describe('POST /api/v1/airdrops', () => { { address: validAddress2, amount: 50 }, ], }); - console.log('POST /airdrops response status:', response.status); - console.log('POST /airdrops response body:', response.body); - console.log('mockStore contents after POST:', Array.from(mockStore.entries())); - console.log('mockSets contents after POST:', Array.from(mockSets.entries())); expect(response.status).toBe(201); expect(response.body.id).toMatch(/^drop_/); expect(response.body.name).toBe('Test Airdrop'); @@ -174,11 +204,7 @@ describe('POST /api/v1/airdrops', () => { describe('GET /api/v1/airdrops', () => { test('lists airdrops with pagination', async () => { - console.log('=== Test: lists airdrops with pagination ==='); - console.log('Before first POST: mockStore', Array.from(mockStore.entries())); - console.log('Before first POST: mockSets', Array.from(mockSets.entries())); - - const res1 = await request(app) + await request(app) .post('/api/v1/airdrops') .send({ name: 'Airdrop 1', @@ -187,13 +213,8 @@ describe('GET /api/v1/airdrops', () => { total_amount: 100, expiry_ledger: 123456, }); - console.log('First POST res status:', res1.status); - console.log('First POST res body:', res1.body); - console.log('After first POST: mockStore', Array.from(mockStore.entries())); - console.log('After first POST: mockSets', Array.from(mockSets.entries())); - - const res2 = await request(app) + await request(app) .post('/api/v1/airdrops') .send({ name: 'Airdrop 2', @@ -202,13 +223,8 @@ describe('GET /api/v1/airdrops', () => { total_amount: 200, expiry_ledger: 123457, }); - console.log('Second POST res status:', res2.status); - - console.log('After second POST: mockStore', Array.from(mockStore.entries())); - console.log('After second POST: mockSets', Array.from(mockSets.entries())); const response = await request(app).get('/api/v1/airdrops?page=1&limit=2'); - console.log('GET /airdrops response body:', response.body); expect(response.status).toBe(200); expect(response.body.airdrops).toHaveLength(2); expect(response.body.pagination.total).toBe(2); diff --git a/test/alerts-routes.test.js b/test/alerts-routes.test.js index b96c20b..6f8ae8b 100644 --- a/test/alerts-routes.test.js +++ b/test/alerts-routes.test.js @@ -3,8 +3,27 @@ const adminApiKey = 'a'.repeat(64); process.env.ADMIN_API_KEY = adminApiKey; +const mockZSets = new Map(); + const mockRedis = { smembers: jest.fn(async () => []), + 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 z = mockZSets.get(key); + if (!z) return []; + const sorted = [...z.entries()].sort((a, b) => b[1] - a[1]).map(([m]) => m); + 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', () => ({ diff --git a/test/auth.test.js b/test/auth.test.js index e3815a7..1252885 100644 --- a/test/auth.test.js +++ b/test/auth.test.js @@ -6,6 +6,7 @@ const crypto = require('crypto'); const mockStore = new Map(); const mockSets = new Map(); +const mockZSets = new Map(); const mockRedis = { smembers: jest.fn(async (key) => [...(mockSets.get(key) || [])]), @@ -16,6 +17,22 @@ const mockRedis = { 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 z = mockZSets.get(key); + if (!z) return []; + const sorted = [...z.entries()].sort((a, b) => b[1] - a[1]).map(([m]) => m); + const end = stop === -1 ? sorted.length : stop + 1; + return sorted.slice(start, end); + }), }; jest.mock('../src/services/cache', () => ({ diff --git a/test/config.test.js b/test/config.test.js index 704b613..51be94f 100644 --- a/test/config.test.js +++ b/test/config.test.js @@ -72,6 +72,9 @@ describe('configuration validation', () => { anomalyThresholdPercent: 20, }, airdrops: { + expiryCheckIntervalSeconds: 60, + ledgerCacheTtlMs: 5000, + expiryScanBatchSize: 100, csvMaxBytes: 5 * 1024 * 1024, jsonMaxBytes: 2 * 1024 * 1024, maxRecipients: 10000, diff --git a/test/deliveryRepository.test.js b/test/deliveryRepository.test.js new file mode 100644 index 0000000..55315fb --- /dev/null +++ b/test/deliveryRepository.test.js @@ -0,0 +1,116 @@ +'use strict'; + +const { createCacheMock } = require('./helpers/cacheMock'); + +const mockHelper = createCacheMock(); +const { reset, redis, zsets } = mockHelper; + +jest.mock('../src/services/cache', () => mockHelper.cacheMock); +jest.mock('../src/logger', () => ({ + info: jest.fn(), warn: jest.fn(), error: jest.fn(), debug: jest.fn(), +})); + +const deliveryRepo = require('../src/repositories/deliveryRepository'); + +const RETRY_QUEUE_KEY = 'webhooks:retries'; + +beforeEach(() => reset()); + +function seedDueRetries(count, { dueAt = 1000 } = {}) { + const ids = []; + for (let i = 0; i < count; i += 1) { + const id = `dlv_${String(i).padStart(4, '0')}`; + ids.push(id); + redis.zadd(RETRY_QUEUE_KEY, dueAt + i, id); + } + return ids; +} + +describe('popDueRetries', () => { + test('returns and removes due ids up to max', async () => { + seedDueRetries(5); + const popped = await deliveryRepo.popDueRetries(2000, 3); + expect(popped).toHaveLength(3); + const remaining = zsets.get(RETRY_QUEUE_KEY); + expect(remaining.size).toBe(2); + }); + + test('ignores retries not yet due', async () => { + await deliveryRepo.scheduleRetry('dlv_future', 5000); + const popped = await deliveryRepo.popDueRetries(1000, 25); + expect(popped).toEqual([]); + expect(zsets.get(RETRY_QUEUE_KEY).size).toBe(1); + }); + + test('empty due set returns [] without error', async () => { + expect(await deliveryRepo.popDueRetries(Date.now(), 25)).toEqual([]); + }); + + test('two concurrent callers never receive overlapping ids', async () => { + const seeded = seedDueRetries(50); + const [first, second] = await Promise.all([ + deliveryRepo.popDueRetries(2000, 25), + deliveryRepo.popDueRetries(2000, 25), + ]); + + const overlap = first.filter((id) => second.includes(id)); + expect(overlap).toEqual([]); + + const union = new Set([...first, ...second]); + expect(union.size).toBe(50); + expect([...union].sort()).toEqual([...seeded].sort()); + expect(zsets.get(RETRY_QUEUE_KEY).size).toBe(0); + }); + + test('many concurrent callers still partition the queue with no duplicates', async () => { + const seeded = seedDueRetries(100); + const results = await Promise.all( + Array.from({ length: 4 }, () => deliveryRepo.popDueRetries(2000, 25)), + ); + + const allIds = results.flat(); + expect(allIds).toHaveLength(100); + expect(new Set(allIds).size).toBe(100); + expect([...allIds].sort()).toEqual([...seeded].sort()); + }); + + test('regression: the old read-then-delete pattern double-claims under a race', async () => { + // Demonstrates the bug this fix closes: two round trips to Redis (a + // ZRANGEBYSCORE followed later by a ZREM) let a second caller read the + // same ids before the first caller's ZREM has run. The production code + // no longer does this - popDueRetries now uses a single atomic Lua + // round trip - but this test proves the failure mode it replaces. + seedDueRetries(10); + + async function racyPop(nowMs, max) { + const ids = await redis.zrangebyscore(RETRY_QUEUE_KEY, '-inf', nowMs, 'LIMIT', 0, max); + await new Promise((resolve) => setTimeout(resolve, 10)); + if (ids.length > 0) await redis.zrem(RETRY_QUEUE_KEY, ...ids); + return ids; + } + + const [first, second] = await Promise.all([racyPop(2000, 10), racyPop(2000, 10)]); + const overlap = first.filter((id) => second.includes(id)); + expect(overlap.length).toBeGreaterThan(0); + }); +}); + +describe('cancelRetry / scheduleRetry / listByWebhook (unchanged by the atomic fix)', () => { + test('scheduleRetry adds a member with the given score', async () => { + await deliveryRepo.scheduleRetry('dlv_a', 12345); + expect(zsets.get(RETRY_QUEUE_KEY).get('dlv_a')).toBe(12345); + }); + + test('cancelRetry removes a scheduled retry', async () => { + await deliveryRepo.scheduleRetry('dlv_b', 12345); + await deliveryRepo.cancelRetry('dlv_b'); + expect(zsets.get(RETRY_QUEUE_KEY).has('dlv_b')).toBe(false); + }); + + test('listByWebhook returns all persisted deliveries for that webhook', async () => { + const a = await deliveryRepo.create({ webhook_id: 'wh_1', event_id: 'evt_a', event_type: 'x' }); + const b = await deliveryRepo.create({ webhook_id: 'wh_1', event_id: 'evt_b', event_type: 'x' }); + const list = await deliveryRepo.listByWebhook('wh_1', 10); + expect(list.map((d) => d.id).sort()).toEqual([a.id, b.id].sort()); + }); +}); diff --git a/test/errorHandler.test.js b/test/errorHandler.test.js index 2d9a545..11f002b 100644 --- a/test/errorHandler.test.js +++ b/test/errorHandler.test.js @@ -1,222 +1,122 @@ -const request = require('supertest'); -const express = require('express'); -const AppError = require('../src/errors/AppError'); -const { requestIdMiddleware } = require('../src/middleware/requestId'); -const { errorHandler, notFoundHandler } = require('../src/middleware/errorHandler'); -const buildRateLimit = require('../src/middleware/rateLimit'); -const cache = require('../src/services/cache'); - -jest.mock('../src/logger', () => ({ - info: jest.fn(), warn: jest.fn(), error: jest.fn(), debug: jest.fn(), -})); - -jest.mock('../src/services/cache', () => ({ - getClient: jest.fn(), -})); - -function buildApp(route) { - const app = express(); - app.use(express.json()); - app.use(requestIdMiddleware); - route(app); - app.use(notFoundHandler); - app.use(errorHandler); - return app; -} - -describe('structured error responses', () => { - test.each([ - ['VALIDATION_ERROR', 400], - ['UNAUTHORIZED', 401], - ['NOT_FOUND', 404], - ['UPSTREAM_ERROR', 502], - ['INTERNAL_ERROR', 500], - ])('returns standard shape for %s', async (code, status) => { - const app = buildApp((app) => { - app.get('/boom', (_req, _res, next) => next(new AppError(code, `${code} message`, status, { field: 'x' }))); - }); - - const res = await request(app).get('/boom'); - - expect(res.status).toBe(status); - expect(res.body).toEqual({ - error: { - code, - message: `${code} message`, - details: { field: 'x' }, - request_id: expect.stringMatching(/^req_/), - } - }); - expect(res.headers['x-request-id']).toBe(res.body.error.request_id); - }); - - test('omits stack traces for unhandled errors', async () => { - const app = buildApp((app) => { - app.get('/boom', () => { throw new Error('secret stack details'); }); - }); - - const res = await request(app).get('/boom'); - - expect(res.status).toBe(500); - expect(res.body.error).toEqual({ - code: 'INTERNAL_ERROR', - message: 'An unexpected error occurred', - request_id: expect.stringMatching(/^req_/) - }); - expect(JSON.stringify(res.body)).not.toContain('secret stack details'); - expect(JSON.stringify(res.body)).not.toContain('stack'); - }); - - test('adds request_id to success responses', async () => { - const app = buildApp((app) => { - app.get('/ok', (_req, res) => res.json({ ok: true })); - }); - - const res = await request(app).get('/ok'); - - expect(res.status).toBe(200); - expect(res.body).toEqual({ ok: true, request_id: expect.stringMatching(/^req_/) }); - }); - - test('returns structured 404 for undefined routes', async () => { - const app = buildApp(() => {}); - const res = await request(app).get('/missing'); - expect(res.status).toBe(404); - expect(res.body.error.code).toBe('NOT_FOUND'); - expect(res.body.error.request_id).toMatch(/^req_/); - }); - - test('returns RATE_LIMITED shape', async () => { - cache.getClient.mockReturnValue({ incr: jest.fn().mockResolvedValue(2), expire: jest.fn().mockResolvedValue(1) }); - const app = buildApp((app) => { - app.get('/limited', buildRateLimit({ windowSeconds: 60, max: 1, keyPrefix: 'test' }), (_req, res) => res.json({ ok: true })); - }); - - const res = await request(app).get('/limited'); - expect(res.status).toBe(429); - expect(res.body.error.code).toBe('RATE_LIMITED'); - expect(res.body.error.details).toEqual({ - limit: 1, - window_seconds: 60, - retry_after_seconds: expect.any(Number), - }); - }); -}); -const request = require('supertest'); -const express = require('express'); -const AppError = require('../src/errors/AppError'); -const { requestIdMiddleware } = require('../src/middleware/requestId'); -const { errorHandler, notFoundHandler } = require('../src/middleware/errorHandler'); -const buildRateLimit = require('../src/middleware/rateLimit'); -const cache = require('../src/services/cache'); - -jest.mock('../src/logger', () => ({ - info: jest.fn(), warn: jest.fn(), error: jest.fn(), debug: jest.fn(), -})); - -jest.mock('../src/services/cache', () => ({ - getClient: jest.fn(), -})); - -function buildApp(route) { - const app = express(); - app.use(express.json()); - app.use(requestIdMiddleware); - route(app); - app.use(notFoundHandler); - app.use(errorHandler); - return app; -} - -describe('structured error responses', () => { - test.each([ - ['VALIDATION_ERROR', 400], - ['UNAUTHORIZED', 401], - ['NOT_FOUND', 404], - ['PAYLOAD_TOO_LARGE', 413], - ['UPSTREAM_ERROR', 502], - ['INTERNAL_ERROR', 500], - ])('returns standard shape for %s', async (code, status) => { - const app = buildApp((app) => { - app.get('/boom', (_req, _res, next) => next(new AppError(code, `${code} message`, status, { field: 'x' }))); - }); - - const res = await request(app).get('/boom'); - - expect(res.status).toBe(status); - expect(res.body).toEqual({ - error: { - code, - message: `${code} message`, - details: { field: 'x' }, - request_id: expect.stringMatching(/^req_/), - } - }); - expect(res.headers['x-request-id']).toBe(res.body.error.request_id); - }); - - test('omits stack traces for unhandled errors', async () => { - const app = buildApp((app) => { - app.get('/boom', () => { throw new Error('secret stack details'); }); - }); - - const res = await request(app).get('/boom'); - - expect(res.status).toBe(500); - expect(res.body.error).toEqual({ - code: 'INTERNAL_ERROR', - message: 'An unexpected error occurred', - request_id: expect.stringMatching(/^req_/) - }); - expect(JSON.stringify(res.body)).not.toContain('secret stack details'); - expect(JSON.stringify(res.body)).not.toContain('stack'); - }); - - test('returns a structured 413 when the JSON body limit is exceeded', async () => { - const app = express(); - app.use(requestIdMiddleware); - app.use(express.json({ limit: 10 })); - app.post('/payload', (_req, res) => res.json({ ok: true })); - app.use(errorHandler); - - const res = await request(app).post('/payload').send({ value: 'too large' }); - - expect(res.status).toBe(413); - expect(res.body.error).toEqual({ - code: 'PAYLOAD_TOO_LARGE', - message: 'Request body is too large', - request_id: expect.stringMatching(/^req_/), - }); - }); - - test('adds request_id to success responses', async () => { - const app = buildApp((app) => { - app.get('/ok', (_req, res) => res.json({ ok: true })); - }); - - const res = await request(app).get('/ok'); - - expect(res.status).toBe(200); - expect(res.body).toEqual({ ok: true, request_id: expect.stringMatching(/^req_/) }); - }); - - test('returns structured 404 for undefined routes', async () => { - const app = buildApp(() => {}); - const res = await request(app).get('/missing'); - expect(res.status).toBe(404); - expect(res.body.error.code).toBe('NOT_FOUND'); - expect(res.body.error.request_id).toMatch(/^req_/); - }); - - test('returns RATE_LIMITED shape', async () => { - cache.getClient.mockReturnValue({ incr: jest.fn().mockResolvedValue(2), expire: jest.fn().mockResolvedValue(1) }); - const app = buildApp((app) => { - app.get('/limited', buildRateLimit({ windowSeconds: 60, max: 1, keyPrefix: 'test' }), (_req, res) => res.json({ ok: true })); - }); - - const res = await request(app).get('/limited'); - expect(res.status).toBe(429); - expect(res.body.error.code).toBe('RATE_LIMITED'); - expect(res.body.error.details).toEqual({ limit: 1, window_seconds: 60 }); - }); -}); +const request = require('supertest'); +const express = require('express'); +const AppError = require('../src/errors/AppError'); +const { requestIdMiddleware } = require('../src/middleware/requestId'); +const { errorHandler, notFoundHandler } = require('../src/middleware/errorHandler'); +const buildRateLimit = require('../src/middleware/rateLimit'); +const cache = require('../src/services/cache'); + +jest.mock('../src/logger', () => ({ + info: jest.fn(), warn: jest.fn(), error: jest.fn(), debug: jest.fn(), +})); + +jest.mock('../src/services/cache', () => ({ + getClient: jest.fn(), +})); + +function buildApp(route) { + const app = express(); + app.use(express.json()); + app.use(requestIdMiddleware); + route(app); + app.use(notFoundHandler); + app.use(errorHandler); + return app; +} + +describe('structured error responses', () => { + test.each([ + ['VALIDATION_ERROR', 400], + ['UNAUTHORIZED', 401], + ['NOT_FOUND', 404], + ['PAYLOAD_TOO_LARGE', 413], + ['UPSTREAM_ERROR', 502], + ['INTERNAL_ERROR', 500], + ])('returns standard shape for %s', async (code, status) => { + const app = buildApp((app) => { + app.get('/boom', (_req, _res, next) => next(new AppError(code, `${code} message`, status, { field: 'x' }))); + }); + + const res = await request(app).get('/boom'); + + expect(res.status).toBe(status); + expect(res.body).toEqual({ + error: { + code, + message: `${code} message`, + details: { field: 'x' }, + request_id: expect.stringMatching(/^req_/), + } + }); + expect(res.headers['x-request-id']).toBe(res.body.error.request_id); + }); + + test('omits stack traces for unhandled errors', async () => { + const app = buildApp((app) => { + app.get('/boom', () => { throw new Error('secret stack details'); }); + }); + + const res = await request(app).get('/boom'); + + expect(res.status).toBe(500); + expect(res.body.error).toEqual({ + code: 'INTERNAL_ERROR', + message: 'An unexpected error occurred', + request_id: expect.stringMatching(/^req_/) + }); + expect(JSON.stringify(res.body)).not.toContain('secret stack details'); + expect(JSON.stringify(res.body)).not.toContain('stack'); + }); + + test('returns a structured 413 when the JSON body limit is exceeded', async () => { + const app = express(); + app.use(requestIdMiddleware); + app.use(express.json({ limit: 10 })); + app.post('/payload', (_req, res) => res.json({ ok: true })); + app.use(errorHandler); + + const res = await request(app).post('/payload').send({ value: 'too large' }); + + expect(res.status).toBe(413); + expect(res.body.error).toEqual({ + code: 'PAYLOAD_TOO_LARGE', + message: 'Request body is too large', + request_id: expect.stringMatching(/^req_/), + }); + }); + + test('adds request_id to success responses', async () => { + const app = buildApp((app) => { + app.get('/ok', (_req, res) => res.json({ ok: true })); + }); + + const res = await request(app).get('/ok'); + + expect(res.status).toBe(200); + expect(res.body).toEqual({ ok: true, request_id: expect.stringMatching(/^req_/) }); + }); + + test('returns structured 404 for undefined routes', async () => { + const app = buildApp(() => {}); + const res = await request(app).get('/missing'); + expect(res.status).toBe(404); + expect(res.body.error.code).toBe('NOT_FOUND'); + expect(res.body.error.request_id).toMatch(/^req_/); + }); + + test('returns RATE_LIMITED shape', async () => { + cache.getClient.mockReturnValue({ incr: jest.fn().mockResolvedValue(2), expire: jest.fn().mockResolvedValue(1) }); + const app = buildApp((app) => { + app.get('/limited', buildRateLimit({ windowSeconds: 60, max: 1, keyPrefix: 'test' }), (_req, res) => res.json({ ok: true })); + }); + + const res = await request(app).get('/limited'); + expect(res.status).toBe(429); + expect(res.body.error.code).toBe('RATE_LIMITED'); + expect(res.body.error.details).toEqual({ + limit: 1, + window_seconds: 60, + retry_after_seconds: expect.any(Number), + }); + }); +}); diff --git a/test/helpers/cacheMock.js b/test/helpers/cacheMock.js index e7f3f0c..9f19e6d 100644 --- a/test/helpers/cacheMock.js +++ b/test/helpers/cacheMock.js @@ -68,6 +68,28 @@ function createCacheMock() { return n; }), expire: jest.fn(async () => 1), + // Mimics ioredis#defineCommand for the one custom command this codebase + // registers (see deliveryRepository.js). Real Redis runs the Lua body + // single-threaded to completion, so this mock implementation reads and + // deletes without an intervening `await`, preserving that atomicity + // guarantee for tests. + defineCommand: jest.fn((name, { lua } = {}) => { + if (name === 'popDueRetriesAtomic') { + redis.popDueRetriesAtomic = jest.fn(async (queueKey, maxScore, limit) => { + const z = getZSet(queueKey); + const max = Number(maxScore); + const ids = [...z.entries()] + .filter(([, score]) => score <= max) + .sort((a, b) => a[1] - b[1]) + .slice(0, Number(limit)) + .map(([m]) => m); + ids.forEach((id) => z.delete(id)); + return ids; + }); + return; + } + throw new Error(`cacheMock.defineCommand: unsupported command "${name}" (lua: ${typeof lua})`); + }), }; const cacheMock = {