Merge pull request #595 from auraroom/fix-cache-pool-content-type-retention

fix: cache, pool sizing, content-type validation, and retention

Author: ritaifeoluwa

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,
   },

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,

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,

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,
+};

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;
 }
 

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>} 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,
 };

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<void>}
+ */
+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<void>}
+ */
+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<void>}
+ */
+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<void>}
+ */
+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,
+};