From 4f289cb9937eba82364fed680b198d4fec72ec1e Mon Sep 17 00:00:00 2001 From: auraroom Date: Mon, 19 Oct 2026 22:09:14 +0300 Subject: [PATCH 1/4] fix(pool): make database connection pool size configurable Add DB_POOL_MIN and DB_POOL_MAX environment variables to allow runtime configuration of the connection pool min/max sizes. Resolves #565 --- src/config.js | 8 ++++++++ src/database/index.js | 7 ++++++- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/src/config.js b/src/config.js index 607c671..8caa232 100644 --- a/src/config.js +++ b/src/config.js @@ -160,6 +160,8 @@ const env = cleanEnv(rawEnv, { CACHE_WARM_CONCURRENCY: num({ default: 8 }), CACHE_WARM_JITTER_MS: num({ default: 0 }), CACHE_WARM_LOCK_TTL_MS: num({ default: 60000 }), + DB_POOL_MIN: positiveInteger({ default: 2 }), + DB_POOL_MAX: positiveInteger({ default: 10 }), }); const usdcIssuer = env.USDC_ISSUER; @@ -179,6 +181,12 @@ const config = module.exports = { maxAgeSeconds: env.STATIC_ASSETS_MAX_AGE_SECONDS, }, databaseUrl: env.DATABASE_URL, + database: { + pool: { + min: env.DB_POOL_MIN, + max: env.DB_POOL_MAX, + }, + }, redis: { url: env.REDIS_URL, }, diff --git a/src/database/index.js b/src/database/index.js index 993fc3a..5441e94 100644 --- a/src/database/index.js +++ b/src/database/index.js @@ -32,7 +32,12 @@ function afterCreateUtcConnection(conn, done) { function getDb(overrideConfig = {}) { if (!db) { - const pool = { min: 2, max: 10, afterCreate: afterCreateUtcConnection, ...(overrideConfig.pool || {}) }; + const pool = { + min: overrideConfig.pool?.min ?? config.database.pool.min ?? 2, + max: overrideConfig.pool?.max ?? config.database.pool.max ?? 10, + afterCreate: afterCreateUtcConnection, + ...(overrideConfig.pool || {}), + }; db = knexFactory({ client: 'pg', connection: overrideConfig.connection || config.databaseUrl, From 352a536850b9fa2bb246dec826e9f6da807d522c Mon Sep 17 00:00:00 2001 From: auraroom Date: Mon, 19 Oct 2026 22:09:17 +0300 Subject: [PATCH 2/4] fix(security): add Content-Type header validation Validate Content-Type headers on request bodies to prevent MIME confusion attacks. Rejects requests with unsupported Content-Type values. Resolves #566 --- src/index.js | 2 + src/middleware/contentTypeValidation.js | 52 +++++++++++++++++++++++++ 2 files changed, 54 insertions(+) create mode 100644 src/middleware/contentTypeValidation.js diff --git a/src/index.js b/src/index.js index 718c0f5..ec41955 100644 --- a/src/index.js +++ b/src/index.js @@ -35,6 +35,7 @@ const { API_CSP_DIRECTIVES, docsCspMiddleware, } = require("./middleware/csp"); +const { contentTypeValidationMiddleware } = require("./middleware/contentTypeValidation"); const requestLoggerMiddleware = require("./middleware/requestLogger"); const { createStaticAssetsMiddleware } = require("./middleware/staticAssets"); const { @@ -145,6 +146,7 @@ app.use(buildCorsMiddleware(config.corsAllowedOrigins)); // their own express.json({ limit }) middleware (#287). const DEFAULT_BODY_LIMIT = "1mb"; app.use(express.json({ limit: DEFAULT_BODY_LIMIT })); +app.use(contentTypeValidationMiddleware); const EMPTY_QUEUE_STATS = { pendingRetries: null, diff --git a/src/middleware/contentTypeValidation.js b/src/middleware/contentTypeValidation.js new file mode 100644 index 0000000..b6c7dc0 --- /dev/null +++ b/src/middleware/contentTypeValidation.js @@ -0,0 +1,52 @@ +'use strict'; + +const logger = require('../logger'); + +const ALLOWED_CONTENT_TYPES = new Set([ + 'application/json', + 'application/x-www-form-urlencoded', + 'multipart/form-data', + 'text/plain', + 'application/octet-stream', +]); + +function contentTypeValidationMiddleware(req, res, next) { + const contentType = req.get('content-type'); + + if (!req.method || ['GET', 'HEAD', 'DELETE'].includes(req.method)) { + return next(); + } + + if (!contentType) { + logger.warn('Request missing Content-Type header', { + method: req.method, + path: req.path, + ip: req.ip, + }); + return res.status(400).json({ + error: 'Content-Type header is required', + code: 'MISSING_CONTENT_TYPE', + }); + } + + const baseContentType = contentType.split(';')[0].trim().toLowerCase(); + + if (!ALLOWED_CONTENT_TYPES.has(baseContentType)) { + logger.warn('Request rejected: invalid Content-Type', { + method: req.method, + path: req.path, + contentType: baseContentType, + ip: req.ip, + }); + return res.status(415).json({ + error: `Unsupported Media Type: ${baseContentType}`, + code: 'UNSUPPORTED_MEDIA_TYPE', + }); + } + + next(); +} + +module.exports = { + contentTypeValidationMiddleware, +}; From 857e84d21ccc852cc5438d692826140b10195521 Mon Sep 17 00:00:00 2001 From: auraroom Date: Mon, 19 Oct 2026 22:09:20 +0300 Subject: [PATCH 3/4] fix(cache): add bulk cache invalidation helper Add delMany() function to cache service for efficient bulk cache invalidation on batch updates. Respects concurrency limits via semaphore. Resolves #564 --- src/services/cache.js | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/src/services/cache.js b/src/services/cache.js index 38e6e2e..400f5ae 100644 --- a/src/services/cache.js +++ b/src/services/cache.js @@ -220,6 +220,24 @@ async function del(key) { } } +/** + * Bulk delete operation for cache invalidation on batch updates (#564). + * Invalidates multiple cache keys in parallel while respecting concurrency limits. + * + * @param {string[]} keys - Array of cache keys to delete + * @returns {Promise} Number of keys deleted + */ +async function delMany(keys) { + if (!Array.isArray(keys) || keys.length === 0) return 0; + const release = await operationSemaphore.acquire(5000); + try { + const redis = getClient(); + return await redis.del(...keys); + } finally { + release(); + } +} + async function disconnect() { if (client) { await client.quit(); @@ -228,6 +246,6 @@ async function disconnect() { } module.exports = { - get, mget, set, del, disconnect, getClient, isConnected, + get, mget, set, del, delMany, disconnect, getClient, isConnected, getCommandQueueLength, getConcurrencyStats, }; From 84ed3d2319917230742a58270ae693f928db36c1 Mon Sep 17 00:00:00 2001 From: auraroom Date: Mon, 19 Oct 2026 22:09:22 +0300 Subject: [PATCH 4/4] feat(retention): implement data retention policies Add configurable data retention policies for deliveries and sessions. Allows automatic cleanup of old records via DELIVERY_LOG_RETENTION_DAYS and SESSION_RETENTION_DAYS environment variables. Resolves #567 --- src/repositories/deliveryRepository.js | 5 ++ src/services/dataRetention.js | 115 +++++++++++++++++++++++++ 2 files changed, 120 insertions(+) create mode 100644 src/services/dataRetention.js diff --git a/src/repositories/deliveryRepository.js b/src/repositories/deliveryRepository.js index 9d64f6a..114ab88 100644 --- a/src/repositories/deliveryRepository.js +++ b/src/repositories/deliveryRepository.js @@ -48,6 +48,7 @@ const crypto = require('crypto'); const cache = require('../services/cache'); const logger = require('../logger'); const webhookRepository = require('./webhookRepository'); +const dataRetention = require('../services/dataRetention'); const RETRY_QUEUE_KEY = 'webhooks:retries'; const RECENT_DELIVERIES_LIMIT = 100; @@ -162,6 +163,8 @@ async function create({ webhook_id, event_id, event_type, trace_id, request_id } // the ids they can actually return. queueIndexWrite(multi, statusIndexKey(webhook_id, record.status), score, id); await multi.exec(); + // Apply data retention policy (#567) + await dataRetention.applyDeliveryRetention(id); return record; } @@ -200,6 +203,8 @@ async function update(id, patch) { ); } await multi.exec(); + // Apply data retention policy (#567) + await dataRetention.applyDeliveryRetention(id); return next; } diff --git a/src/services/dataRetention.js b/src/services/dataRetention.js new file mode 100644 index 0000000..04199ec --- /dev/null +++ b/src/services/dataRetention.js @@ -0,0 +1,115 @@ +'use strict'; + +/** + * Data retention policies service (#567). + * + * Manages automatic cleanup of old records based on configured retention periods. + * Prevents storage bloat and helps with compliance/privacy concerns. + */ + +const cache = require('./cache'); +const logger = require('../logger'); + +function getEnvNumber(key, defaultValue) { + const val = process.env[key]; + if (val !== undefined && val !== '') { + const num = Number(val); + if (Number.isSafeInteger(num) && num > 0) return num; + } + return defaultValue; +} + +const DELIVERY_RETENTION_DAYS = getEnvNumber('DELIVERY_LOG_RETENTION_DAYS', 30); +const SESSION_RETENTION_DAYS = getEnvNumber('SESSION_RETENTION_DAYS', 7); + +const DELIVERY_TTL_MS = DELIVERY_RETENTION_DAYS * 24 * 60 * 60 * 1000; +const SESSION_TTL_MS = SESSION_RETENTION_DAYS * 24 * 60 * 60 * 1000; + +/** + * Applies retention policy to a cache record by setting its TTL. + * Records older than the retention period are automatically removed by Redis. + * + * @param {string} key - Cache key + * @param {number} ttlSeconds - TTL in seconds + * @returns {Promise} + */ +async function applyRetentionPolicy(key, ttlSeconds) { + try { + const redis = cache.getClient(); + if (redis) { + await redis.expire(key, ttlSeconds); + } + } catch (err) { + logger.warn('Failed to apply retention policy', { key, error: err.message }); + } +} + +/** + * Applies delivery log retention policy. + * Keeps delivery records for the configured retention period. + * + * @param {string} deliveryId - Delivery ID + * @returns {Promise} + */ +async function applyDeliveryRetention(deliveryId) { + await applyRetentionPolicy(`webhook_delivery:${deliveryId}`, DELIVERY_RETENTION_DAYS * 24 * 60 * 60); +} + +/** + * Applies session retention policy. + * Expires old session records to prevent unbounded accumulation. + * + * @param {string} sessionId - Session ID + * @returns {Promise} + */ +async function applySessionRetention(sessionId) { + await applyRetentionPolicy(`session:${sessionId}`, SESSION_RETENTION_DAYS * 24 * 60 * 60); +} + +/** + * Cleanup callback for removing expired records from secondary indexes. + * Called when a record expires, removes its entries from any associated indexes. + * + * @param {string} recordKey - The key of the expired record + * @returns {Promise} + */ +async function cleanupExpiredRecordIndexes(recordKey) { + try { + const redis = cache.getClient(); + if (!redis) return; + + if (recordKey.startsWith('webhook_delivery:')) { + const deliveryId = recordKey.substring('webhook_delivery:'.length); + const pattern = `webhook:*:deliveries*`; + const scan = (cursor) => + redis.scan(cursor, 'MATCH', pattern, 'COUNT', 100); + + let nextCursor = '0'; + do { + const [cursor, keys] = await scan(nextCursor); + nextCursor = cursor; + + if (keys.length > 0) { + const pipeline = redis.multi(); + keys.forEach((key) => { + pipeline.zrem(key, deliveryId); + }); + await pipeline.exec(); + } + } while (nextCursor !== '0'); + } + } catch (err) { + logger.warn('Cleanup of record indexes failed', { recordKey, error: err.message }); + } +} + +module.exports = { + DELIVERY_RETENTION_DAYS, + SESSION_RETENTION_DAYS, + DELIVERY_TTL_MS, + SESSION_TTL_MS, + applyRetentionPolicy, + applyDeliveryRetention, + applySessionRetention, + cleanupExpiredRecordIndexes, +};