feat: implement issues #990, #991, #994, #997 (#1021)

- Add priority queue background job system (#990)
- Add CDN edge caching service (#991)
- Add WebSocket connection pool with message batching (#994)
- Add read replica router with automatic failover (#997)

Closes #990
Closes #991
Closes #994
Closes #997

Author: susanyusuf

backend/services/shared/cdnService.ts

@@ -0,0 +1,161 @@
+/**
+ * CDN Edge Caching Service — SubTrackr
+ *
+ * Manages edge caching for API responses with TTL, purge, and invalidation.
+ */
+
+export interface CdnConfig {
+  defaultTtlSeconds: number;
+  maxTtlSeconds: number;
+  staleWhileRevalidateSeconds: number;
+  purgeBatchSize: number;
+}
+
+export interface CacheEntry<T = unknown> {
+  key: string;
+  data: T;
+  headers: Record<string, string>;
+  cachedAt: number;
+  ttlMs: number;
+  tags: string[];
+}
+
+export interface PurgeRequest {
+  patterns: string[];
+  tags?: string[];
+}
+
+export interface PurgeResult {
+  purgedCount: number;
+  patterns: string[];
+}
+
+export interface CdnMetrics {
+  hits: number;
+  misses: number;
+  purges: number;
+  hitRate: number;
+}
+
+export class CdnService {
+  private cache = new Map<string, CacheEntry>();
+  private tagIndex = new Map<string, Set<string>>();
+  private config: CdnConfig;
+  private metrics: CdnMetrics = { hits: 0, misses: 0, purges: 0, hitRate: 0 };
+
+  constructor(config: Partial<CdnConfig> = {}) {
+    this.config = {
+      defaultTtlSeconds: config.defaultTtlSeconds ?? 300,
+      maxTtlSeconds: config.maxTtlSeconds ?? 86400,
+      staleWhileRevalidateSeconds: config.staleWhileRevalidateSeconds ?? 60,
+      purgeBatchSize: config.purgeBatchSize ?? 100,
+    };
+  }
+
+  get<T>(key: string): CacheEntry<T> | null {
+    const entry = this.cache.get(key) as CacheEntry<T> | undefined;
+    if (!entry) {
+      this.metrics.misses++;
+      this.updateHitRate();
+      return null;
+    }
+
+    const now = Date.now();
+    if (now > entry.cachedAt + entry.ttlMs) {
+      this.cache.delete(key);
+      this.removeFromTagIndex(key, entry.tags);
+      this.metrics.misses++;
+      this.updateHitRate();
+      return null;
+    }
+
+    this.metrics.hits++;
+    this.updateHitRate();
+    return entry;
+  }
+
+  set<T>(key: string, data: T, options: { ttlSeconds?: number; tags?: string[] } = {}): void {
+    const ttlMs = Math.min(
+      (options.ttlSeconds ?? this.config.defaultTtlSeconds) * 1000,
+      this.config.maxTtlSeconds * 1000,
+    );
+
+    const entry: CacheEntry<T> = {
+      key,
+      data,
+      headers: {
+        'cache-control': `public, max-age=${Math.floor(ttlMs / 1000)}, stale-while-revalidate=${this.config.staleWhileRevalidateSeconds}`,
+        'cdn-cache-status': 'HIT',
+      },
+      cachedAt: Date.now(),
+      ttlMs,
+      tags: options.tags ?? [],
+    };
+
+    this.cache.set(key, entry);
+    for (const tag of entry.tags) {
+      if (!this.tagIndex.has(tag)) this.tagIndex.set(tag, new Set());
+      this.tagIndex.get(tag)!.add(key);
+    }
+  }
+
+  purge(request: PurgeRequest): PurgeResult {
+    let purgedCount = 0;
+    const purgedPatterns: string[] = [];
+
+    for (const pattern of request.patterns) {
+      const regex = new RegExp('^' + pattern.replace(/\*/g, '.*') + '$');
+      for (const [key, entry] of this.cache) {
+        if (regex.test(key)) {
+          this.cache.delete(key);
+          this.removeFromTagIndex(key, entry.tags);
+          purgedCount++;
+        }
+      }
+      purgedPatterns.push(pattern);
+    }
+
+    if (request.tags) {
+      for (const tag of request.tags) {
+        const keys = this.tagIndex.get(tag);
+        if (keys) {
+          for (const key of keys) {
+            this.cache.delete(key);
+            purgedCount++;
+          }
+          this.tagIndex.delete(tag);
+        }
+      }
+    }
+
+    this.metrics.purges++;
+    return { purgedCount, patterns: purgedPatterns };
+  }
+
+  invalidate(key: string): boolean {
+    const entry = this.cache.get(key);
+    if (entry) {
+      this.removeFromTagIndex(key, entry.tags);
+      this.cache.delete(key);
+      return true;
+    }
+    return false;
+  }
+
+  getMetrics(): CdnMetrics {
+    return { ...this.metrics };
+  }
+
+  private removeFromTagIndex(key: string, tags: string[]): void {
+    for (const tag of tags) {
+      this.tagIndex.get(tag)?.delete(key);
+    }
+  }
+
+  private updateHitRate(): void {
+    const total = this.metrics.hits + this.metrics.misses;
+    this.metrics.hitRate = total > 0 ? this.metrics.hits / total : 0;
+  }
+}
+
+export const cdnService = new CdnService();

backend/services/shared/index.ts

@@ -169,3 +169,19 @@ export type {
   LeakRecord,
   PoolTuningRecommendation,
 } from './poolMonitor';
+
+// ── Background Job Queue (#990) ──────────────────────────────────────────────
+export { PriorityQueue, jobQueue } from './jobQueue';
+export type { Job, JobHandler, QueueConfig, QueueMetrics, JobStatus, JobPriority } from './jobQueue';
+
+// ── CDN Edge Caching (#991) ──────────────────────────────────────────────────
+export { CdnService, cdnService } from './cdnService';
+export type { CdnConfig, CacheEntry, PurgeRequest, PurgeResult, CdnMetrics } from './cdnService';
+
+// ── WebSocket Connection Pool (#994) ──────────────────────────────────────────
+export { WsConnectionPool } from './wsConnectionPool';
+export type { WsPoolConfig, WsConnection, WsMessage, WsPoolMetrics } from './wsConnectionPool';
+
+// ── Read Replica Router (#997) ────────────────────────────────────────────────
+export { ReadReplicaRouter } from './readReplicaRouter';
+export type { ReplicaConfig, ReplicaHealth, ReadRouteOptions, QueryRoute } from './readReplicaRouter';

backend/services/shared/jobQueue.ts

@@ -0,0 +1,192 @@
+/**
+ * Background Job Queue — SubTrackr
+ *
+ * Priority-based job queue for background processing (email, billing, analytics).
+ */
+
+export type JobPriority = 'critical' | 'high' | 'medium' | 'low';
+export type JobStatus = 'pending' | 'processing' | 'completed' | 'failed' | 'retrying';
+
+export interface Job {
+  id: string;
+  type: string;
+  payload: Record<string, unknown>;
+  priority: JobPriority;
+  status: JobStatus;
+  attempts: number;
+  maxAttempts: number;
+  createdAt: number;
+  startedAt?: number;
+  completedAt?: number;
+  error?: string;
+  nextRetryAt?: number;
+}
+
+export type JobHandler = (job: Job) => Promise<void>;
+
+export interface QueueConfig {
+  maxConcurrent: number;
+  maxRetries: number;
+  retryDelayMs: number;
+  processIntervalMs: number;
+}
+
+interface QueueMetrics {
+  totalProcessed: number;
+  totalFailed: number;
+  avgProcessingTimeMs: number;
+  currentlyProcessing: number;
+}
+
+const PRIORITY_WEIGHTS: Record<JobPriority, number> = {
+  critical: 100,
+  high: 75,
+  medium: 50,
+  low: 25,
+};
+
+export class PriorityQueue {
+  private queues: Map<JobPriority, Job[]> = new Map();
+  private handlers: Map<string, JobHandler> = new Map();
+  private processing = new Set<string>();
+  private config: QueueConfig;
+  private metrics: QueueMetrics = {
+    totalProcessed: 0,
+    totalFailed: 0,
+    avgProcessingTimeMs: 0,
+    currentlyProcessing: 0,
+  };
+  private timer?: ReturnType<typeof setInterval>;
+
+  constructor(config: Partial<QueueConfig> = {}) {
+    this.config = {
+      maxConcurrent: config.maxConcurrent ?? 5,
+      maxRetries: config.maxRetries ?? 3,
+      retryDelayMs: config.retryDelayMs ?? 1000,
+      processIntervalMs: config.processIntervalMs ?? 100,
+    };
+
+    for (const priority of Object.keys(PRIORITY_WEIGHTS) as JobPriority[]) {
+      this.queues.set(priority, []);
+    }
+  }
+
+  registerHandler(type: string, handler: JobHandler): void {
+    this.handlers.set(type, handler);
+  }
+
+  enqueue(type: string, payload: Record<string, unknown>, priority: JobPriority = 'medium'): Job {
+    const job: Job = {
+      id: `${type}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
+      type,
+      payload,
+      priority,
+      status: 'pending',
+      attempts: 0,
+      maxAttempts: this.config.maxRetries,
+      createdAt: Date.now(),
+    };
+
+    this.queues.get(priority)!.push(job);
+    return job;
+  }
+
+  private dequeue(): Job | undefined {
+    const sortedPriorities = (Object.keys(PRIORITY_WEIGHTS) as JobPriority[]).sort(
+      (a, b) => PRIORITY_WEIGHTS[b] - PRIORITY_WEIGHTS[a],
+    );
+
+    for (const priority of sortedPriorities) {
+      const queue = this.queues.get(priority)!;
+      if (queue.length > 0) {
+        return queue.shift();
+      }
+    }
+    return undefined;
+  }
+
+  async processNext(): Promise<boolean> {
+    if (this.processing.size >= this.config.maxConcurrent) return false;
+
+    const job = this.dequeue();
+    if (!job) return false;
+
+    const handler = this.handlers.get(job.type);
+    if (!handler) {
+      job.status = 'failed';
+      job.error = `No handler registered for job type: ${job.type}`;
+      this.metrics.totalFailed++;
+      return false;
+    }
+
+    job.status = 'processing';
+    job.startedAt = Date.now();
+    job.attempts++;
+    this.processing.add(job.id);
+    this.metrics.currentlyProcessing = this.processing.size;
+
+    try {
+      await handler(job);
+      job.status = 'completed';
+      job.completedAt = Date.now();
+      this.metrics.totalProcessed++;
+
+      const duration = job.completedAt - job.startedAt;
+      this.metrics.avgProcessingTimeMs =
+        (this.metrics.avgProcessingTimeMs * (this.metrics.totalProcessed - 1) + duration) /
+        this.metrics.totalProcessed;
+    } catch (err) {
+      job.error = err instanceof Error ? err.message : String(err);
+
+      if (job.attempts < job.maxAttempts) {
+        job.status = 'retrying';
+        job.nextRetryAt = Date.now() + this.config.retryDelayMs * job.attempts;
+        this.queues.get(job.priority)!.push(job);
+      } else {
+        job.status = 'failed';
+        this.metrics.totalFailed++;
+      }
+    } finally {
+      this.processing.delete(job.id);
+      this.metrics.currentlyProcessing = this.processing.size;
+    }
+
+    return true;
+  }
+
+  start(): void {
+    if (this.timer) return;
+    this.timer = setInterval(() => {
+      void this.processNext();
+    }, this.config.processIntervalMs);
+  }
+
+  stop(): void {
+    if (this.timer) {
+      clearInterval(this.timer);
+      this.timer = undefined;
+    }
+  }
+
+  getMetrics(): QueueMetrics {
+    return { ...this.metrics };
+  }
+
+  getPendingCount(): number {
+    let count = 0;
+    for (const queue of this.queues.values()) {
+      count += queue.length;
+    }
+    return count;
+  }
+
+  getJob(id: string): Job | undefined {
+    for (const queue of this.queues.values()) {
+      const job = queue.find((j) => j.id === id);
+      if (job) return job;
+    }
+    return undefined;
+  }
+}
+
+export const jobQueue = new PriorityQueue();