Feat/issues 962 964 968 941

backend/src/index.ts

@@ -31,6 +31,7 @@ import { schedulerService } from './services/scheduler';
 import { reminderEngine } from './services/reminder-engine';
 import { notificationPreferenceService } from './services/notification-preference-service';
 import subscriptionRoutes from './routes/subscriptions';
+import subscriptionShareRoutes from './routes/subscription-shares';
 import riskScoreRoutes from './routes/risk-score';
 import simulationRoutes from './routes/simulation';
 import merchantRoutes from './routes/merchants';
@@ -72,6 +73,7 @@ import { startChannelSettlementJob } from './jobs/channel-settlement-job';
 import { startJobAlertMonitor, stopJobAlertMonitor } from './jobs/job-alert-monitor';
 import giftCardLedgerRoutes from './routes/gift-card-ledger';
 import notificationDeadLetterRoutes from './routes/notification-dead-letter';
+import renewalDeadLetterRoutes from './routes/renewal-dead-letter';
 import telegramWebhookRoutes from './routes/telegram-webhook';
 import { telegramCommandService } from './services/telegram-command-service';
 import calendarRouter from './routes/calendar';
@@ -171,6 +173,7 @@ app.get('/api/docs.json', (_req, res) => {
 
 // API Routes
 app.use('/api/keys', apiKeysRoutes);
+app.use('/api/subscriptions', subscriptionShareRoutes);
 app.use('/api/subscriptions', subscriptionRoutes);
 app.use('/api/risk-score', riskScoreRoutes);
 app.use('/api/simulation', simulationRoutes);
@@ -198,6 +201,7 @@ app.use('/api/wallet', walletRoutes);
 app.use('/api/key-rotation', keyRotationRoutes);
 app.use('/api/privacy', privacyRoutes);
 app.use('/api/notifications/dead-letter', notificationDeadLetterRoutes);
+app.use('/api/renewals/dead-letter', renewalDeadLetterRoutes);
 app.use('/api/exchange-rates', createExchangeRatesRouter(exchangeRateService));
 app.use('/api/gift-card-ledger', giftCardLedgerRoutes);
 app.use('/api/payments', authenticate, paymentsRoutes);
@@ -311,6 +315,26 @@ app.get('/api/admin/metrics/failed-items', createAdminLimiter(), adminAuth, asyn
   }
 });
 
+app.get('/api/admin/metrics/query-cache', createAdminLimiter(), adminAuth, async (_req, res) => {
+  try {
+    const metrics = await monitoringService.getQueryCacheMetrics();
+    res.json(metrics);
+  } catch (error) {
+    logger.error('Error fetching query cache metrics:', error);
+    res.status(500).json({ error: 'Failed to fetch query cache metrics' });
+  }
+});
+
+app.get('/api/admin/metrics/renewal-locks', createAdminLimiter(), adminAuth, async (_req, res) => {
+  try {
+    const metrics = await monitoringService.getRenewalLockMetrics();
+    res.json(metrics);
+  } catch (error) {
+    logger.error('Error fetching renewal lock metrics:', error);
+    res.status(500).json({ error: 'Failed to fetch renewal lock metrics' });
+  }
+});
+
 app.get('/api/admin/metrics/api-latency', createAdminLimiter(), adminAuth, async (req, res) => {
   try {
     const metrics = await monitoringService.getApiLatencyMetrics();

backend/src/lib/redis-client.ts

@@ -0,0 +1,103 @@
+import { createClient, RedisClientType } from 'redis';
+import logger from '../config/logger';
+import { rateLimitConfig } from '../config/rate-limit';
+
+/**
+ * Shared Redis client for distributed locks, query caching, and metrics.
+ * Reuses the same URL as rate limiting when available, otherwise REDIS_URL.
+ */
+export class SharedRedisClient {
+  private static instance: SharedRedisClient | null = null;
+  private client: RedisClientType | null = null;
+  private isConnected = false;
+  private initPromise: Promise<void> | null = null;
+
+  private constructor() {}
+
+  static getInstance(): SharedRedisClient {
+    if (!SharedRedisClient.instance) {
+      SharedRedisClient.instance = new SharedRedisClient();
+    }
+    return SharedRedisClient.instance;
+  }
+
+  private resolveUrl(): string | undefined {
+    return rateLimitConfig.redis.url || process.env.REDIS_URL;
+  }
+
+  async initialize(): Promise<void> {
+    if (this.isConnected && this.client) {
+      return;
+    }
+    if (this.initPromise) {
+      return this.initPromise;
+    }
+
+    const url = this.resolveUrl();
+    if (!url) {
+      logger.info('[SharedRedisClient] No Redis URL configured');
+      return;
+    }
+
+    this.initPromise = (async () => {
+      try {
+        this.client = createClient({
+          url,
+          socket: {
+            reconnectStrategy: (retries) => Math.min(5000 * Math.pow(2, retries), 30000),
+          },
+        });
+
+        this.client.on('connect', () => {
+          this.isConnected = true;
+          logger.info('[SharedRedisClient] Connected');
+        });
+
+        this.client.on('error', (error) => {
+          this.isConnected = false;
+          logger.error('[SharedRedisClient] Error:', error);
+        });
+
+        this.client.on('disconnect', () => {
+          this.isConnected = false;
+        });
+
+        await this.client.connect();
+      } catch (error) {
+        logger.error('[SharedRedisClient] Failed to initialize:', error);
+        this.client = null;
+        this.isConnected = false;
+        throw error;
+      } finally {
+        this.initPromise = null;
+      }
+    })();
+
+    return this.initPromise;
+  }
+
+  async getClient(): Promise<RedisClientType | null> {
+    if (!this.client || !this.isConnected) {
+      try {
+        await this.initialize();
+      } catch {
+        return null;
+      }
+    }
+    return this.client;
+  }
+
+  isAvailable(): boolean {
+    return this.isConnected && this.client !== null;
+  }
+
+  async shutdown(): Promise<void> {
+    if (this.client) {
+      await this.client.disconnect().catch(() => undefined);
+      this.client = null;
+      this.isConnected = false;
+    }
+  }
+}
+
+export const sharedRedisClient = SharedRedisClient.getInstance();

backend/src/lib/redis-lock.ts

@@ -0,0 +1,134 @@
+import { randomUUID } from 'crypto';
+import logger from '../config/logger';
+import { sharedRedisClient } from './redis-client';
+
+export interface LockAcquireResult {
+  acquired: boolean;
+  lockToken?: string;
+  reason?: 'contention' | 'unavailable';
+}
+
+export interface LockMetricsSnapshot {
+  acquired: number;
+  contention: number;
+  released: number;
+  expired: number;
+}
+
+const METRICS_PREFIX = 'renewal_lock:metrics:';
+
+/**
+ * Redis-based distributed lock using SET NX with TTL.
+ * Default TTL: 5 minutes (300 seconds).
+ */
+export class RedisDistributedLock {
+  private readonly keyPrefix = 'renewal_lock:';
+  private readonly defaultTtlMs: number;
+
+  constructor(defaultTtlMs = 5 * 60 * 1000) {
+    this.defaultTtlMs = defaultTtlMs;
+  }
+
+  private lockKey(subscriptionId: string, cycleId: number): string {
+    return `${this.keyPrefix}${subscriptionId}:${cycleId}`;
+  }
+
+  private async incrementMetric(field: 'acquired' | 'contention' | 'released' | 'expired'): Promise<void> {
+    const client = await sharedRedisClient.getClient();
+    if (!client) return;
+    try {
+      await client.incr(`${METRICS_PREFIX}${field}`);
+    } catch (error) {
+      logger.warn('[RedisDistributedLock] Failed to increment metric', { field, error });
+    }
+  }
+
+  async acquire(
+    subscriptionId: string,
+    cycleId: number,
+    ttlMs: number = this.defaultTtlMs,
+  ): Promise<LockAcquireResult> {
+    const client = await sharedRedisClient.getClient();
+    if (!client) {
+      logger.warn('[RedisDistributedLock] Redis unavailable, lock not acquired');
+      return { acquired: false, reason: 'unavailable' };
+    }
+
+    const token = randomUUID();
+    const key = this.lockKey(subscriptionId, cycleId);
+    const ttlSeconds = Math.max(1, Math.ceil(ttlMs / 1000));
+
+    try {
+      const result = await client.set(key, token, { NX: true, EX: ttlSeconds });
+      if (result === 'OK') {
+        await this.incrementMetric('acquired');
+        logger.info('[RedisDistributedLock] Lock acquired', { subscriptionId, cycleId, ttlSeconds });
+        return { acquired: true, lockToken: token };
+      }
+
+      await this.incrementMetric('contention');
+      logger.warn('[RedisDistributedLock] Lock contention', { subscriptionId, cycleId });
+      return { acquired: false, reason: 'contention' };
+    } catch (error) {
+      logger.error('[RedisDistributedLock] Acquire failed', { subscriptionId, cycleId, error });
+      return { acquired: false, reason: 'unavailable' };
+    }
+  }
+
+  /**
+   * Release lock only if we still hold it (compare token via Lua script).
+   */
+  async release(subscriptionId: string, cycleId: number, lockToken: string): Promise<boolean> {
+    const client = await sharedRedisClient.getClient();
+    if (!client) return false;
+
+    const key = this.lockKey(subscriptionId, cycleId);
+    const script = `
+      if redis.call("get", KEYS[1]) == ARGV[1] then
+        return redis.call("del", KEYS[1])
+      else
+        return 0
+      end
+    `;
+
+    try {
+      const result = await client.eval(script, { keys: [key], arguments: [lockToken] });
+      if (result === 1) {
+        await this.incrementMetric('released');
+        logger.info('[RedisDistributedLock] Lock released', { subscriptionId, cycleId });
+        return true;
+      }
+      await this.incrementMetric('expired');
+      return false;
+    } catch (error) {
+      logger.error('[RedisDistributedLock] Release failed', { subscriptionId, cycleId, error });
+      return false;
+    }
+  }
+
+  async getMetrics(): Promise<LockMetricsSnapshot> {
+    const client = await sharedRedisClient.getClient();
+    if (!client) {
+      return { acquired: 0, contention: 0, released: 0, expired: 0 };
+    }
+
+    const fields: Array<keyof LockMetricsSnapshot> = ['acquired', 'contention', 'released', 'expired'];
+    const values = await Promise.all(
+      fields.map(async (field) => {
+        const val = await client.get(`${METRICS_PREFIX}${field}`);
+        return parseInt(val ?? '0', 10) || 0;
+      }),
+    );
+
+    return {
+      acquired: values[0],
+      contention: values[1],
+      released: values[2],
+      expired: values[3],
+    };
+  }
+}
+
+export const redisDistributedLock = new RedisDistributedLock(
+  parseInt(process.env.RENEWAL_LOCK_TTL_MS ?? '300000', 10) || 5 * 60 * 1000,
+);

backend/src/routes/renewal-dead-letter.ts

@@ -0,0 +1,45 @@
+import { Router, Response } from 'express';
+import { authenticate, AuthenticatedRequest } from '../middleware/auth';
+import { adminAuth } from '../middleware/admin';
+import { renewalDeadLetterService } from '../services/renewal-dead-letter-service';
+import logger from '../config/logger';
+
+const router: Router = Router();
+
+router.use(authenticate);
+
+/**
+ * GET /api/renewals/dead-letter
+ * List renewal dead-letter entries for the authenticated user.
+ */
+router.get('/', async (req: AuthenticatedRequest, res: Response) => {
+  try {
+    const entries = await renewalDeadLetterService.getUserDeadLetters(req.user!.id);
+    res.json({ success: true, data: entries });
+  } catch (error) {
+    logger.error('GET /api/renewals/dead-letter error:', error);
+    res.status(500).json({
+      success: false,
+      error: error instanceof Error ? error.message : 'Failed to fetch renewal dead-letter entries',
+    });
+  }
+});
+
+/**
+ * GET /api/renewals/dead-letter/stats
+ * Admin-only aggregate DLQ statistics.
+ */
+router.get('/stats', adminAuth, async (_req: AuthenticatedRequest, res: Response) => {
+  try {
+    const stats = await renewalDeadLetterService.getDeadLetterStats();
+    res.json({ success: true, data: stats });
+  } catch (error) {
+    logger.error('GET /api/renewals/dead-letter/stats error:', error);
+    res.status(500).json({
+      success: false,
+      error: error instanceof Error ? error.message : 'Failed to fetch renewal DLQ stats',
+    });
+  }
+});
+
+export default router;