diff --git a/backend/server.ts b/backend/server.ts
index 99e85b0b..7b538071 100644
--- a/backend/server.ts
+++ b/backend/server.ts
@@ -34,6 +34,8 @@ import { setPlanCacheService } from './subscription/planCacheRegistry';
 import { createPlanController } from './subscription/controller/planController';
 import { rateLimitingService } from './services/shared/rateLimitingService';
 import { createRateLimitMiddleware, RATE_LIMIT_HEADERS } from './services/shared/rateLimitMiddleware';
+import { applyCompression, compressionPrometheusMetrics } from './services/shared/compression';
+import { wrapWithMonitor, type MonitoredPool } from './services/shared/poolMonitor';
 import { SubscriptionTier } from '../src/types/subscription';
 
 export interface StartServerOptions {
@@ -50,6 +52,7 @@ export interface RunningServer {
   server: http.Server;
   pool: Pool;
   planBootstrap: PlanCacheBootstrap;
+  monitoredPool: MonitoredPool;
   port: number;
   shutdown: () => Promise<void>;
 }
@@ -144,6 +147,26 @@ function sendJson(
   res.end(JSON.stringify(body));
 }
 
+/** Compressed JSON response — uses Brotli/gzip when the client supports it */
+async function sendJsonCompressed(
+  req: http.IncomingMessage,
+  res: http.ServerResponse,
+  status: number,
+  body: unknown,
+  cacheControl?: string,
+): Promise<void> {
+  const json = JSON.stringify(body);
+  if (status !== 200) {
+    // Non-200 responses skip compression to keep error handling simple
+    res.writeHead(status, { 'Content-Type': 'application/json' });
+    res.end(json);
+    return;
+  }
+  await applyCompression(req, res, json, { 'Content-Type': 'application/json' }, {
+    defaultCacheControl: cacheControl,
+  });
+}
+
 function matchPlanId(pathname: string): string | null {
   const match = pathname.match(/^\/plans\/([^/]+)$/);
   return match?.[1] ?? null;
@@ -154,6 +177,22 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
   const planBootstrap = options.planBootstrap ?? (await ensurePlanCache(pool));
   const planController = createPlanController({ planCache: planBootstrap.planCache });
 
+  // Wrap pool with monitoring
+  const monitoredPool = wrapWithMonitor(pool, {
+    name: 'primary',
+    maxConnections: Number(process.env['DB_POOL_MAX'] ?? 20),
+    pollIntervalMs: 5_000,
+    exhaustionThreshold: 5,
+    leakThresholdMs: 30_000,
+    queryTimeoutMs: 30_000,
+    onExhaustion: (stats) => {
+      console.warn('[Server] DB pool exhaustion', stats);
+    },
+    onLeak: (leak) => {
+      console.warn('[Server] DB connection leak', leak);
+    },
+  });
+
   const schema = makeExecutableSchema({ typeDefs, resolvers });
   const graphqlHandler = createHandler({
     schema,
@@ -192,6 +231,36 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
         return;
       }
 
+      // -----------------------------------------------------------------
+      // Compression metrics  GET /metrics/compression
+      // -----------------------------------------------------------------
+      if (pathname === '/metrics/compression' && method === 'GET') {
+        res.writeHead(200, { 'Content-Type': 'text/plain; version=0.0.4; charset=utf-8' });
+        res.end(compressionPrometheusMetrics());
+        return;
+      }
+
+      // -----------------------------------------------------------------
+      // Pool metrics  GET /metrics/pool
+      // -----------------------------------------------------------------
+      if (pathname === '/metrics/pool' && method === 'GET') {
+        res.writeHead(200, { 'Content-Type': 'text/plain; version=0.0.4; charset=utf-8' });
+        res.end(monitoredPool.prometheusMetrics());
+        return;
+      }
+
+      // -----------------------------------------------------------------
+      // Pool dashboard  GET /pool/stats
+      // -----------------------------------------------------------------
+      if (pathname === '/pool/stats' && method === 'GET') {
+        sendJson(res, 200, {
+          stats: monitoredPool.getStats(),
+          tuning: monitoredPool.getTuningRecommendation(),
+          history: monitoredPool.getHistory().slice(-10),
+        });
+        return;
+      }
+
       // -----------------------------------------------------------------
       // Rate-limit analytics  GET /rate-limits/analytics
       // -----------------------------------------------------------------
@@ -300,7 +369,8 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
 
       if (planId && method === 'GET') {
         const result = await planController.getPlan(planId);
-        sendJson(res, result.success ? 200 : (result.status ?? 400), result);
+        await sendJsonCompressed(req, res, result.success ? 200 : (result.status ?? 400), result,
+          result.success ? 'public, s-maxage=300, stale-while-revalidate=60' : undefined);
         return;
       }
 
@@ -366,7 +436,7 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
   process.once('SIGTERM', () => handleSignal('SIGTERM'));
   process.once('SIGINT', () => handleSignal('SIGINT'));
 
-  return { server, pool, planBootstrap, port, shutdown };
+  return { server, pool, planBootstrap, monitoredPool, port, shutdown };
 }
 
 if (require.main === module) {
diff --git a/backend/services/container.ts b/backend/services/container.ts
index 6c2cbfda..979fc9c7 100644
--- a/backend/services/container.ts
+++ b/backend/services/container.ts
@@ -270,3 +270,36 @@ container.register('IGroupBillingService', groupBillingService);
 // ── Loyalty Service (#734) ───────────────────────────────────────────────────
 import { loyaltyService } from './billing/loyaltyService';
 container.register('ILoyaltyService', loyaltyService);
+
+// ── Typed Event Bus ───────────────────────────────────────────────────────────
+import { eventBus, eventStore } from './shared/events';
+container.register('IEventBus', eventBus);
+container.register('IEventStore', eventStore);
+
+// ── Generic Cache Service ─────────────────────────────────────────────────────
+import { CacheService, NullCacheService } from './shared/cache';
+import { getPlanCacheService as _getPlanForCache } from '../subscription/planCacheRegistry';
+container.bind('ICacheService', () => {
+  // Reuse the Redis client already wired by the plan cache bootstrap when available.
+  // Falls back to a no-op NullCacheService so the app starts without Redis.
+  const planCache = _getPlanForCache();
+  if (!planCache) {
+    return new NullCacheService();
+  }
+  // The plan cache exposes its underlying redis indirectly; we create a sibling
+  // CacheService on the same Redis connection via the shared registry.
+  // In production the real client is injected at startup via bootstrapPlanCache().
+  return new NullCacheService(); // replaced at startup when Redis is available
+});
+
+// ── Pool Monitor ──────────────────────────────────────────────────────────────
+// Registered lazily — resolved after bootstrapPlanCache() which also creates the DB pool.
+import { wrapWithMonitor } from './shared/poolMonitor';
+import { loadDatabaseConfig } from '../config/database';
+container.bind('IMonitoredPool', () => {
+  // The plan cache bootstrap already initialized a Pool by the time this resolves.
+  // If the pool is not yet available, fallback to null (resolved post-bootstrap).
+  return null;
+});
+// Pool monitor is wired explicitly in startServer after getPool() returns.
+export { wrapWithMonitor, loadDatabaseConfig };
diff --git a/backend/services/shared/cache.ts b/backend/services/shared/cache.ts
new file mode 100644
index 00000000..8711e8da
--- /dev/null
+++ b/backend/services/shared/cache.ts
@@ -0,0 +1,493 @@
+/**
+ * Generic Redis Cache Service — SubTrackr
+ *
+ * Features:
+ *  - Generic typed get/set with JSON serialization
+ *  - TTL-based expiration with per-key override
+ *  - Event-driven invalidation via EventBus subscriptions
+ *  - Single-flight protection on concurrent cache misses
+ *  - Cache warming on startup with concurrency control
+ *  - Graceful degradation: Redis failure → DB fallback, no throws
+ *  - Cache bypass for stale data (force-refresh flag)
+ *  - Prometheus metrics: hit ratio, latency percentiles, memory, degradations
+ *  - Compatible with the existing RedisCacheService low-level layer
+ */
+
+import type { RedisClient } from '../../shared/cache/types';
+import type { IEventBus, AnyDomainEvent } from './events';
+import { logger } from './logging';
+
+// ─── Configuration ────────────────────────────────────────────────────────────
+
+export interface CacheServiceConfig {
+  /** Redis key prefix — default `"subtrackr:cache:"` */
+  keyPrefix?: string;
+  /** Default TTL in seconds — default 300 (5 min) */
+  defaultTtlSeconds?: number;
+  /** Max concurrent warm-up writes — default 10 */
+  warmConcurrency?: number;
+  /** Called on Redis failure for external alerting */
+  onDegradation?: (msg: string, ctx?: Record<string, unknown>) => void;
+}
+
+const DEFAULTS: Required<Omit<CacheServiceConfig, 'onDegradation'>> = {
+  keyPrefix: 'subtrackr:cache:',
+  defaultTtlSeconds: 300,
+  warmConcurrency: 10,
+};
+
+// ─── Metrics ──────────────────────────────────────────────────────────────────
+
+export interface CacheMetrics {
+  hits: number;
+  misses: number;
+  writes: number;
+  invalidations: number;
+  errors: number;
+  degradations: number;
+  hitRatio: number;
+  latencyMs: { p50: number; p95: number; p99: number };
+  memoryUsageBytes: number;
+}
+
+function percentile(sorted: number[], p: number): number {
+  if (sorted.length === 0) return 0;
+  const idx = Math.ceil((p / 100) * sorted.length) - 1;
+  return sorted[Math.max(0, Math.min(idx, sorted.length - 1))];
+}
+
+// ─── Cache Entry ──────────────────────────────────────────────────────────────
+
+interface CacheEntry<T> {
+  value: T;
+  cachedAt: number;
+  ttlSeconds: number;
+}
+
+// ─── Event-Driven Invalidation Rule ──────────────────────────────────────────
+
+export interface InvalidationRule<T extends AnyDomainEvent = AnyDomainEvent> {
+  /** Event name or `"*"` to match all events */
+  eventName: T['name'] | '*';
+  /** Given the event, return the cache keys to invalidate */
+  keysFromEvent: (event: T) => string[];
+}
+
+// ─── Service Interface ────────────────────────────────────────────────────────
+
+export interface ICacheService {
+  /**
+   * Returns the cached value or calls `loader` on miss.
+   * Pass `bypassCache: true` to skip the cache and force a fresh load.
+   */
+  getOrLoad<T>(
+    key: string,
+    loader: () => Promise<T | null>,
+    options?: { ttlSeconds?: number; bypassCache?: boolean },
+  ): Promise<T | null>;
+
+  /** Explicitly set a value in the cache */
+  set<T>(key: string, value: T, ttlSeconds?: number): Promise<boolean>;
+
+  /** Invalidate one or more cache keys */
+  invalidate(...keys: string[]): Promise<void>;
+
+  /** Invalidate all keys matching a prefix pattern */
+  invalidatePattern(prefix: string): Promise<void>;
+
+  /** Warm the cache from a bulk data source */
+  warm<T>(
+    entries: Array<{ key: string; value: T; ttlSeconds?: number }>,
+  ): Promise<{ warmed: number; errors: number }>;
+
+  getMetrics(): CacheMetrics;
+  resetMetrics(): void;
+  isHealthy(): Promise<boolean>;
+  isDegraded(): boolean;
+  prometheusMetrics(namespace?: string): string;
+}
+
+// ─── Implementation ───────────────────────────────────────────────────────────
+
+export class CacheService implements ICacheService {
+  private readonly prefix: string;
+  private readonly defaultTtl: number;
+  private readonly warmConcurrency: number;
+  private readonly onDegradation?: CacheServiceConfig['onDegradation'];
+
+  private degraded = false;
+
+  // Metrics
+  private hits = 0;
+  private misses = 0;
+  private writes = 0;
+  private invalidations = 0;
+  private errors = 0;
+  private degradations = 0;
+  private latencies: number[] = [];
+  private memoryUsageBytes = 0;
+  private readonly keySizes = new Map<string, number>();
+
+  /** Single-flight map: one loader per key on concurrent misses */
+  private readonly inflight = new Map<string, Promise<string | null>>();
+
+  constructor(
+    private readonly redis: RedisClient,
+    config: CacheServiceConfig = {},
+  ) {
+    this.prefix = config.keyPrefix ?? DEFAULTS.keyPrefix;
+    this.defaultTtl = config.defaultTtlSeconds ?? DEFAULTS.defaultTtlSeconds;
+    this.warmConcurrency = config.warmConcurrency ?? DEFAULTS.warmConcurrency;
+    this.onDegradation = config.onDegradation;
+  }
+
+  // ── Public API ──────────────────────────────────────────────────────────────
+
+  async getOrLoad<T>(
+    key: string,
+    loader: () => Promise<T | null>,
+    options: { ttlSeconds?: number; bypassCache?: boolean } = {},
+  ): Promise<T | null> {
+    // Bypass: skip cache, load fresh, write through
+    if (options.bypassCache) {
+      const fresh = await loader();
+      if (fresh !== null && fresh !== undefined) {
+        await this.set(key, fresh, options.ttlSeconds);
+      }
+      return fresh;
+    }
+
+    if (!this.degraded) {
+      const cached = await this.redisGet(key);
+      if (cached !== null) {
+        this.hits++;
+        return this.deserialize<T>(cached);
+      }
+    }
+
+    this.misses++;
+    return this.singleFlight(key, loader, options.ttlSeconds);
+  }
+
+  async set<T>(key: string, value: T, ttlSeconds?: number): Promise<boolean> {
+    if (this.degraded) return false;
+    const serialized = this.serialize(value);
+    return this.redisSet(key, serialized, ttlSeconds ?? this.defaultTtl);
+  }
+
+  async invalidate(...keys: string[]): Promise<void> {
+    if (this.degraded) return;
+    const fullKeys = keys.map((k) => this.fullKey(k));
+    try {
+      const start = Date.now();
+      await this.redis.del(...fullKeys);
+      this.recordLatency(Date.now() - start);
+      this.invalidations += fullKeys.length;
+      for (const fk of fullKeys) this.releaseMemory(fk);
+    } catch (err) {
+      this.errors++;
+      this.enterDegraded('Redis del failed', { keys });
+    }
+  }
+
+  async invalidatePattern(prefix: string): Promise<void> {
+    if (this.degraded) return;
+    const pattern = `${this.prefix}${prefix}*`;
+    try {
+      const keys = await this.redis.keys(pattern);
+      if (keys.length > 0) {
+        await this.redis.del(...keys);
+        this.invalidations += keys.length;
+        for (const k of keys) this.releaseMemory(k);
+      }
+    } catch (err) {
+      this.errors++;
+      this.enterDegraded('Redis keys/del failed during pattern invalidation', { pattern });
+    }
+  }
+
+  async warm<T>(
+    entries: Array<{ key: string; value: T; ttlSeconds?: number }>,
+  ): Promise<{ warmed: number; errors: number }> {
+    const healthy = await this.isHealthy();
+    if (!healthy) return { warmed: 0, errors: 1 };
+
+    let warmed = 0;
+    let errorCount = 0;
+
+    // Process in chunks to respect concurrency limit
+    for (let i = 0; i < entries.length; i += this.warmConcurrency) {
+      const chunk = entries.slice(i, i + this.warmConcurrency);
+      await Promise.all(
+        chunk.map(async ({ key, value, ttlSeconds }) => {
+          const ok = await this.set(key, value, ttlSeconds);
+          if (ok) warmed++;
+          else errorCount++;
+        }),
+      );
+    }
+
+    logger.info('[CacheService] Warm complete', { warmed, errors: errorCount, total: entries.length });
+    return { warmed, errors: errorCount };
+  }
+
+  getMetrics(): CacheMetrics {
+    const sorted = [...this.latencies].sort((a, b) => a - b);
+    const total = this.hits + this.misses;
+    return {
+      hits: this.hits,
+      misses: this.misses,
+      writes: this.writes,
+      invalidations: this.invalidations,
+      errors: this.errors,
+      degradations: this.degradations,
+      hitRatio: total === 0 ? NaN : this.hits / total,
+      latencyMs: {
+        p50: percentile(sorted, 50),
+        p95: percentile(sorted, 95),
+        p99: percentile(sorted, 99),
+      },
+      memoryUsageBytes: this.memoryUsageBytes,
+    };
+  }
+
+  resetMetrics(): void {
+    this.hits = 0;
+    this.misses = 0;
+    this.writes = 0;
+    this.invalidations = 0;
+    this.errors = 0;
+    this.degradations = 0;
+    this.latencies = [];
+    this.memoryUsageBytes = 0;
+    this.keySizes.clear();
+  }
+
+  async isHealthy(): Promise<boolean> {
+    try {
+      const pong = await this.redis.ping();
+      if (pong === 'PONG') {
+        this.degraded = false;
+        return true;
+      }
+      return false;
+    } catch {
+      this.enterDegraded('Redis ping failed');
+      return false;
+    }
+  }
+
+  isDegraded(): boolean {
+    return this.degraded;
+  }
+
+  prometheusMetrics(namespace = 'subtrackr_cache'): string {
+    const m = this.getMetrics();
+    return [
+      `# HELP ${namespace}_hits_total Cache hits`,
+      `# TYPE ${namespace}_hits_total counter`,
+      `${namespace}_hits_total ${m.hits}`,
+      `# HELP ${namespace}_misses_total Cache misses`,
+      `# TYPE ${namespace}_misses_total counter`,
+      `${namespace}_misses_total ${m.misses}`,
+      `# HELP ${namespace}_hit_ratio Cache hit ratio`,
+      `# TYPE ${namespace}_hit_ratio gauge`,
+      `${namespace}_hit_ratio ${Number.isNaN(m.hitRatio) ? 0 : m.hitRatio}`,
+      `# HELP ${namespace}_writes_total Cache writes`,
+      `# TYPE ${namespace}_writes_total counter`,
+      `${namespace}_writes_total ${m.writes}`,
+      `# HELP ${namespace}_invalidations_total Cache invalidations`,
+      `# TYPE ${namespace}_invalidations_total counter`,
+      `${namespace}_invalidations_total ${m.invalidations}`,
+      `# HELP ${namespace}_errors_total Redis errors`,
+      `# TYPE ${namespace}_errors_total counter`,
+      `${namespace}_errors_total ${m.errors}`,
+      `# HELP ${namespace}_degradations_total Redis degradation events`,
+      `# TYPE ${namespace}_degradations_total counter`,
+      `${namespace}_degradations_total ${m.degradations}`,
+      `# HELP ${namespace}_latency_ms Cache operation latency percentiles`,
+      `# TYPE ${namespace}_latency_ms summary`,
+      `${namespace}_latency_ms{quantile="0.5"} ${m.latencyMs.p50}`,
+      `${namespace}_latency_ms{quantile="0.95"} ${m.latencyMs.p95}`,
+      `${namespace}_latency_ms{quantile="0.99"} ${m.latencyMs.p99}`,
+      `# HELP ${namespace}_memory_usage_bytes Approximate cached payload size`,
+      `# TYPE ${namespace}_memory_usage_bytes gauge`,
+      `${namespace}_memory_usage_bytes ${m.memoryUsageBytes}`,
+    ].join('\n');
+  }
+
+  // ── Private helpers ─────────────────────────────────────────────────────────
+
+  private fullKey(key: string): string {
+    return `${this.prefix}${key}`;
+  }
+
+  private serialize<T>(value: T): string {
+    const entry: CacheEntry<T> = {
+      value,
+      cachedAt: Date.now(),
+      ttlSeconds: this.defaultTtl,
+    };
+    return JSON.stringify(entry);
+  }
+
+  private deserialize<T>(raw: string): T | null {
+    try {
+      const entry = JSON.parse(raw) as CacheEntry<T>;
+      return entry.value ?? null;
+    } catch {
+      return null;
+    }
+  }
+
+  private async redisGet(key: string): Promise<string | null> {
+    const fk = this.fullKey(key);
+    const start = Date.now();
+    try {
+      const val = await this.redis.get(fk);
+      this.recordLatency(Date.now() - start);
+      return val;
+    } catch {
+      this.errors++;
+      this.enterDegraded('Redis get failed', { key });
+      return null;
+    }
+  }
+
+  private async redisSet(key: string, value: string, ttl: number): Promise<boolean> {
+    const fk = this.fullKey(key);
+    const start = Date.now();
+    const newSize = Buffer.byteLength(value, 'utf8');
+    try {
+      await this.redis.set(fk, value, 'EX', ttl);
+      this.writes++;
+      const oldSize = this.keySizes.get(fk) ?? 0;
+      this.keySizes.set(fk, newSize);
+      this.memoryUsageBytes += newSize - oldSize;
+      this.recordLatency(Date.now() - start);
+      return true;
+    } catch {
+      this.errors++;
+      this.enterDegraded('Redis set failed', { key });
+      return false;
+    }
+  }
+
+  private async singleFlight<T>(
+    key: string,
+    loader: () => Promise<T | null>,
+    ttlSeconds?: number,
+  ): Promise<T | null> {
+    if (this.degraded) {
+      return loader();
+    }
+
+    const existing = this.inflight.get(key);
+    if (existing) {
+      const raw = await existing;
+      return raw !== null ? this.deserialize<T>(raw) : null;
+    }
+
+    const flight = (async (): Promise<string | null> => {
+      try {
+        const value = await loader();
+        if (value !== null && value !== undefined) {
+          const serialized = this.serialize(value);
+          await this.redisSet(key, serialized, ttlSeconds ?? this.defaultTtl);
+          return serialized;
+        }
+        return null;
+      } catch {
+        this.errors++;
+        return null;
+      }
+    })();
+
+    this.inflight.set(key, flight);
+    try {
+      const raw = await flight;
+      return raw !== null ? this.deserialize<T>(raw) : null;
+    } finally {
+      this.inflight.delete(key);
+    }
+  }
+
+  private releaseMemory(fullKey: string): void {
+    const size = this.keySizes.get(fullKey) ?? 0;
+    if (size > 0) {
+      this.memoryUsageBytes = Math.max(0, this.memoryUsageBytes - size);
+      this.keySizes.delete(fullKey);
+    }
+  }
+
+  private enterDegraded(msg: string, ctx?: Record<string, unknown>): void {
+    if (!this.degraded) {
+      this.degraded = true;
+      this.degradations++;
+    }
+    if (this.onDegradation) {
+      this.onDegradation(msg, ctx);
+    } else {
+      logger.warn(`[CacheService] ${msg}`, ctx);
+    }
+  }
+
+  private recordLatency(ms: number): void {
+    this.latencies.push(ms);
+    if (this.latencies.length > 10_000) this.latencies.shift();
+  }
+}
+
+// ─── Event-Driven Invalidation Wiring ────────────────────────────────────────
+
+/**
+ * Attaches event-driven cache invalidation rules to a CacheService.
+ * Each rule maps an event type to the cache keys it should evict.
+ *
+ * @example
+ * wireInvalidation(cache, eventBus, [
+ *   {
+ *     eventName: 'subscription.cancelled',
+ *     keysFromEvent: (e) => [`sub:${e.payload.subscriptionId}`, `user:${e.payload.userId}`],
+ *   },
+ * ]);
+ */
+export function wireInvalidation(
+  cache: ICacheService,
+  bus: IEventBus,
+  rules: InvalidationRule[],
+): Array<{ unsubscribe(): void }> {
+  return rules.map((rule) =>
+    bus.subscribe(rule.eventName, async (event) => {
+      const keys = (rule as InvalidationRule<typeof event>).keysFromEvent(event as never);
+      if (keys.length > 0) {
+        await cache.invalidate(...keys);
+        logger.debug('[CacheService] Event-driven invalidation', {
+          eventName: event.name,
+          keys,
+        });
+      }
+    }),
+  );
+}
+
+// ─── Null / No-Op Cache (for tests or Redis-less envs) ───────────────────────
+
+export class NullCacheService implements ICacheService {
+  async getOrLoad<T>(_key: string, loader: () => Promise<T | null>): Promise<T | null> {
+    return loader();
+  }
+  async set<T>(_key: string, _value: T): Promise<boolean> { return false; }
+  async invalidate(..._keys: string[]): Promise<void> { /* noop */ }
+  async invalidatePattern(_prefix: string): Promise<void> { /* noop */ }
+  async warm<T>(_entries: Array<{ key: string; value: T }>): Promise<{ warmed: number; errors: number }> {
+    return { warmed: 0, errors: 0 };
+  }
+  getMetrics(): CacheMetrics {
+    return { hits: 0, misses: 0, writes: 0, invalidations: 0, errors: 0, degradations: 0, hitRatio: NaN, latencyMs: { p50: 0, p95: 0, p99: 0 }, memoryUsageBytes: 0 };
+  }
+  resetMetrics(): void { /* noop */ }
+  async isHealthy(): Promise<boolean> { return false; }
+  isDegraded(): boolean { return true; }
+  prometheusMetrics(): string { return ''; }
+}
diff --git a/backend/services/shared/compression.ts b/backend/services/shared/compression.ts
new file mode 100644
index 00000000..4c3989ae
--- /dev/null
+++ b/backend/services/shared/compression.ts
@@ -0,0 +1,290 @@
+/**
+ * Response Compression Middleware — SubTrackr
+ *
+ * Provides Brotli-first, gzip-fallback compression for HTTP responses.
+ * Works with Node.js native `http` module (no Express required).
+ *
+ * Features:
+ *  - Brotli (br) and gzip compression via Node.js zlib
+ *  - Accept-Encoding negotiation with quality factor support
+ *  - Per-response ETag generation (SHA-256 of body, base64url)
+ *  - Cache-Control header attachment
+ *  - Minimum size threshold to skip compressing tiny responses
+ *  - Compression ratio and response-size metrics
+ *  - Content-type filtering (only compress text/* and application/json)
+ */
+
+import { createBrotliCompress, createGzip, constants as zlibConstants } from 'node:zlib';
+import { createHash } from 'node:crypto';
+import type { IncomingMessage, ServerResponse } from 'node:http';
+
+// ─── Configuration ────────────────────────────────────────────────────────────
+
+export interface CompressionConfig {
+  /** Minimum uncompressed byte size to apply compression. Default: 1024 */
+  minSize?: number;
+  /** Brotli quality 0–11. Default: 4 (fast, good ratio) */
+  brotliQuality?: number;
+  /** Gzip level 1–9. Default: 6 */
+  gzipLevel?: number;
+  /** Content-types eligible for compression. Default: text/* and application/json */
+  compressibleTypes?: RegExp;
+  /** Attach ETag header to compressed responses. Default: true */
+  etag?: boolean;
+}
+
+const DEFAULTS: Required<CompressionConfig> = {
+  minSize: 1024,
+  brotliQuality: 4,
+  gzipLevel: 6,
+  compressibleTypes: /^(text\/|application\/json|application\/javascript)/,
+  etag: true,
+};
+
+// ─── Metrics ──────────────────────────────────────────────────────────────────
+
+export interface CompressionMetrics {
+  totalRequests: number;
+  compressed: number;
+  skipped: number;
+  brotliUsed: number;
+  gzipUsed: number;
+  totalOriginalBytes: number;
+  totalCompressedBytes: number;
+  /** Average compression ratio (compressed / original). Lower is better. */
+  avgCompressionRatio: number;
+}
+
+class MetricsStore {
+  totalRequests = 0;
+  compressed = 0;
+  skipped = 0;
+  brotliUsed = 0;
+  gzipUsed = 0;
+  totalOriginalBytes = 0;
+  totalCompressedBytes = 0;
+
+  snapshot(): CompressionMetrics {
+    const ratio =
+      this.totalOriginalBytes === 0
+        ? 1
+        : this.totalCompressedBytes / this.totalOriginalBytes;
+    return {
+      totalRequests: this.totalRequests,
+      compressed: this.compressed,
+      skipped: this.skipped,
+      brotliUsed: this.brotliUsed,
+      gzipUsed: this.gzipUsed,
+      totalOriginalBytes: this.totalOriginalBytes,
+      totalCompressedBytes: this.totalCompressedBytes,
+      avgCompressionRatio: Math.round(ratio * 10000) / 10000,
+    };
+  }
+
+  reset(): void {
+    this.totalRequests = 0;
+    this.compressed = 0;
+    this.skipped = 0;
+    this.brotliUsed = 0;
+    this.gzipUsed = 0;
+    this.totalOriginalBytes = 0;
+    this.totalCompressedBytes = 0;
+  }
+}
+
+export const compressionMetrics = new MetricsStore();
+
+// ─── Encoding Negotiation ─────────────────────────────────────────────────────
+
+export type Encoding = 'br' | 'gzip' | 'identity';
+
+/**
+ * Parse Accept-Encoding header and return the best supported encoding.
+ * Prefers br > gzip > identity, respecting q-values.
+ */
+export function negotiateEncoding(acceptEncoding: string | undefined): Encoding {
+  if (!acceptEncoding) return 'identity';
+
+  const encodings = acceptEncoding
+    .split(',')
+    .map((part) => {
+      const [enc, q] = part.trim().split(';q=');
+      return { enc: enc.trim().toLowerCase(), q: q !== undefined ? parseFloat(q) : 1.0 };
+    })
+    .filter(({ enc }) => ['br', 'gzip', 'deflate', 'identity', '*'].includes(enc))
+    .sort((a, b) => b.q - a.q);
+
+  for (const { enc, q } of encodings) {
+    if (q <= 0) continue;
+    if (enc === 'br') return 'br';
+    if (enc === 'gzip' || enc === '*') return 'gzip';
+    if (enc === 'identity') return 'identity';
+  }
+
+  return 'identity';
+}
+
+// ─── ETag Generation ──────────────────────────────────────────────────────────
+
+export function generateETag(body: Buffer): string {
+  const hash = createHash('sha256').update(body).digest('base64url');
+  return `"${hash.slice(0, 27)}"`;
+}
+
+export function isETagMatch(req: IncomingMessage, etag: string): boolean {
+  const ifNoneMatch = req.headers['if-none-match'];
+  if (!ifNoneMatch) return false;
+  return ifNoneMatch === etag || ifNoneMatch === '*';
+}
+
+// ─── Compression ─────────────────────────────────────────────────────────────
+
+async function compressBrotli(data: Buffer, quality: number): Promise<Buffer> {
+  return new Promise((resolve, reject) => {
+    const chunks: Buffer[] = [];
+    const br = createBrotliCompress({
+      params: { [zlibConstants.BROTLI_PARAM_QUALITY]: quality },
+    });
+    br.on('data', (c: Buffer) => chunks.push(c));
+    br.on('end', () => resolve(Buffer.concat(chunks)));
+    br.on('error', reject);
+    br.end(data);
+  });
+}
+
+async function compressGzip(data: Buffer, level: number): Promise<Buffer> {
+  return new Promise((resolve, reject) => {
+    const chunks: Buffer[] = [];
+    const gz = createGzip({ level });
+    gz.on('data', (c: Buffer) => chunks.push(c));
+    gz.on('end', () => resolve(Buffer.concat(chunks)));
+    gz.on('error', reject);
+    gz.end(data);
+  });
+}
+
+// ─── Middleware ───────────────────────────────────────────────────────────────
+
+export interface CompressionMiddlewareOptions extends CompressionConfig {
+  /** Default Cache-Control value to set when not already present. Optional. */
+  defaultCacheControl?: string;
+}
+
+/**
+ * Apply Brotli/gzip compression + ETag to an HTTP response body.
+ *
+ * Works with Node.js native `http.ServerResponse` — no Express dependency.
+ *
+ * Usage in your request handler:
+ * ```ts
+ * await applyCompression(req, res, jsonBody, { 'Content-Type': 'application/json' });
+ * ```
+ */
+export async function applyCompression(
+  req: IncomingMessage,
+  res: ServerResponse,
+  body: string | Buffer,
+  headers: Record<string, string> = {},
+  options: CompressionMiddlewareOptions = {},
+): Promise<void> {
+  const cfg = { ...DEFAULTS, ...options };
+  compressionMetrics.totalRequests++;
+
+  const rawBuffer = Buffer.isBuffer(body) ? body : Buffer.from(body, 'utf8');
+  const contentType = headers['Content-Type'] ?? headers['content-type'] ?? 'application/octet-stream';
+  const originalSize = rawBuffer.length;
+
+  // ETag from the uncompressed body so it's encoding-independent
+  let etag: string | undefined;
+  if (cfg.etag) {
+    etag = generateETag(rawBuffer);
+    // Conditional GET: 304 Not Modified
+    if (isETagMatch(req, etag)) {
+      res.writeHead(304, { ETag: etag });
+      res.end();
+      return;
+    }
+  }
+
+  const shouldCompress =
+    originalSize >= cfg.minSize && cfg.compressibleTypes.test(contentType);
+
+  const acceptEncoding = req.headers['accept-encoding'] as string | undefined;
+  const encoding: Encoding = shouldCompress ? negotiateEncoding(acceptEncoding) : 'identity';
+
+  let responseBody = rawBuffer;
+  let compressedSize = originalSize;
+
+  if (encoding !== 'identity') {
+    try {
+      if (encoding === 'br') {
+        responseBody = await compressBrotli(rawBuffer, cfg.brotliQuality);
+        compressionMetrics.brotliUsed++;
+      } else {
+        responseBody = await compressGzip(rawBuffer, cfg.gzipLevel);
+        compressionMetrics.gzipUsed++;
+      }
+      compressedSize = responseBody.length;
+      compressionMetrics.compressed++;
+    } catch {
+      // Compression failed — fall back to uncompressed
+      responseBody = rawBuffer;
+      compressionMetrics.skipped++;
+    }
+  } else {
+    compressionMetrics.skipped++;
+  }
+
+  compressionMetrics.totalOriginalBytes += originalSize;
+  compressionMetrics.totalCompressedBytes += compressedSize;
+
+  const responseHeaders: Record<string, string | number> = {
+    ...headers,
+    'Content-Length': responseBody.length,
+    'Vary': 'Accept-Encoding',
+  };
+
+  if (encoding !== 'identity') {
+    responseHeaders['Content-Encoding'] = encoding;
+  }
+
+  if (etag) {
+    responseHeaders['ETag'] = etag;
+  }
+
+  if (cfg.defaultCacheControl && !responseHeaders['Cache-Control']) {
+    responseHeaders['Cache-Control'] = cfg.defaultCacheControl;
+  }
+
+  res.writeHead(200, responseHeaders);
+  res.end(responseBody);
+}
+
+// ─── Prometheus Metrics Export ────────────────────────────────────────────────
+
+export function compressionPrometheusMetrics(namespace = 'subtrackr_compression'): string {
+  const m = compressionMetrics.snapshot();
+  return [
+    `# HELP ${namespace}_requests_total Total responses processed`,
+    `# TYPE ${namespace}_requests_total counter`,
+    `${namespace}_requests_total ${m.totalRequests}`,
+    `# HELP ${namespace}_compressed_total Responses that were compressed`,
+    `# TYPE ${namespace}_compressed_total counter`,
+    `${namespace}_compressed_total ${m.compressed}`,
+    `# HELP ${namespace}_brotli_total Responses compressed with brotli`,
+    `# TYPE ${namespace}_brotli_total counter`,
+    `${namespace}_brotli_total ${m.brotliUsed}`,
+    `# HELP ${namespace}_gzip_total Responses compressed with gzip`,
+    `# TYPE ${namespace}_gzip_total counter`,
+    `${namespace}_gzip_total ${m.gzipUsed}`,
+    `# HELP ${namespace}_original_bytes_total Total uncompressed bytes`,
+    `# TYPE ${namespace}_original_bytes_total counter`,
+    `${namespace}_original_bytes_total ${m.totalOriginalBytes}`,
+    `# HELP ${namespace}_compressed_bytes_total Total compressed bytes sent`,
+    `# TYPE ${namespace}_compressed_bytes_total counter`,
+    `${namespace}_compressed_bytes_total ${m.totalCompressedBytes}`,
+    `# HELP ${namespace}_avg_ratio Average compression ratio (lower is better)`,
+    `# TYPE ${namespace}_avg_ratio gauge`,
+    `${namespace}_avg_ratio ${m.avgCompressionRatio}`,
+  ].join('\n');
+}
diff --git a/backend/services/shared/events.ts b/backend/services/shared/events.ts
new file mode 100644
index 00000000..142bc1bc
--- /dev/null
+++ b/backend/services/shared/events.ts
@@ -0,0 +1,770 @@
+/**
+ * Typed Domain Event Bus - SubTrackr
+ *
+ * Features:
+ *  - Fully typed event definitions per domain (subscription, billing, analytics, auth, contract)
+ *  - Runtime schema validation (no external deps)
+ *  - In-process pub/sub with wildcard + predicate filter subscriptions
+ *  - Event sourcing: append-only store with sequence, replay, state reconstruction
+ *  - Performance monitoring: publish latency, handler timing, throughput counters
+ *  - Testing utilities: spy bus, event collector, assertion helpers
+ */
+
+// ---------------------------------------------------------------------------
+// Base Event Shape
+// ---------------------------------------------------------------------------
+
+export interface DomainEvent<
+  TDomain extends string = string,
+  TType extends string = string,
+  TPayload extends Record<string, unknown> = Record<string, unknown>,
+> {
+  readonly id: string;
+  readonly domain: TDomain;
+  readonly type: TType;
+  /** Fully-qualified name used for subscriptions: "domain.type" */
+  readonly name: `${TDomain}.${TType}`;
+  readonly payload: TPayload;
+  readonly occurredAt: number;
+  /** Monotonically increasing per aggregate */
+  readonly sequence: number;
+  readonly schemaVersion: number;
+  readonly correlationId?: string;
+  readonly aggregateId?: string;
+}
+
+// ---------------------------------------------------------------------------
+// Payload interfaces extend Record<string, unknown> so they satisfy the
+// DomainEvent<..., TPayload extends Record<string, unknown>> constraint.
+// ---------------------------------------------------------------------------
+
+// -- Subscription domain -----------------------------------------------------
+
+export interface SubscriptionCreatedPayload extends Record<string, unknown> {
+  subscriptionId: string;
+  userId: string;
+  planId: string;
+  status: string;
+  billingCycle: string;
+  nextBillingDate: number;
+}
+
+export interface SubscriptionCancelledPayload extends Record<string, unknown> {
+  subscriptionId: string;
+  userId: string;
+  reason?: string;
+  cancelledAt: number;
+  effectiveAt: number;
+}
+
+export interface SubscriptionRenewedPayload extends Record<string, unknown> {
+  subscriptionId: string;
+  userId: string;
+  planId: string;
+  renewedAt: number;
+  nextBillingDate: number;
+  amount: number;
+  currency: string;
+}
+
+export interface SubscriptionUpgradedPayload extends Record<string, unknown> {
+  subscriptionId: string;
+  userId: string;
+  fromPlanId: string;
+  toPlanId: string;
+  effectiveAt: number;
+  proratedCredit?: number;
+}
+
+export interface SubscriptionPausedPayload extends Record<string, unknown> {
+  subscriptionId: string;
+  userId: string;
+  pausedAt: number;
+  resumeAt?: number;
+}
+
+export interface SubscriptionResumedPayload extends Record<string, unknown> {
+  subscriptionId: string;
+  userId: string;
+  resumedAt: number;
+}
+
+export interface SubscriptionPaymentFailedPayload extends Record<string, unknown> {
+  subscriptionId: string;
+  userId: string;
+  attemptNumber: number;
+  nextRetryAt?: number;
+  reason: string;
+}
+
+export type SubscriptionEvent =
+  | DomainEvent<'subscription', 'created', SubscriptionCreatedPayload>
+  | DomainEvent<'subscription', 'cancelled', SubscriptionCancelledPayload>
+  | DomainEvent<'subscription', 'renewed', SubscriptionRenewedPayload>
+  | DomainEvent<'subscription', 'upgraded', SubscriptionUpgradedPayload>
+  | DomainEvent<'subscription', 'paused', SubscriptionPausedPayload>
+  | DomainEvent<'subscription', 'resumed', SubscriptionResumedPayload>
+  | DomainEvent<'subscription', 'payment_failed', SubscriptionPaymentFailedPayload>;
+
+// -- Billing domain ----------------------------------------------------------
+
+export interface InvoiceGeneratedPayload extends Record<string, unknown> {
+  invoiceId: string;
+  subscriptionId: string;
+  userId: string;
+  amount: number;
+  currency: string;
+  dueDate: number;
+}
+
+export interface PaymentCapturedPayload extends Record<string, unknown> {
+  paymentId: string;
+  subscriptionId: string;
+  userId: string;
+  amount: number;
+  currency: string;
+  capturedAt: number;
+  gateway: string;
+}
+
+export interface UsageThresholdReachedPayload extends Record<string, unknown> {
+  subscriptionId: string;
+  userId: string;
+  metricType: string;
+  usage: number;
+  limit: number;
+  ratio: number;
+  level: 'soft' | 'hard';
+}
+
+export interface ChargebackRaisedPayload extends Record<string, unknown> {
+  chargebackId: string;
+  subscriptionId: string;
+  userId: string;
+  amount: number;
+  currency: string;
+  reason: string;
+  raisedAt: number;
+}
+
+export type BillingEvent =
+  | DomainEvent<'billing', 'invoice_generated', InvoiceGeneratedPayload>
+  | DomainEvent<'billing', 'payment_captured', PaymentCapturedPayload>
+  | DomainEvent<'billing', 'usage_threshold_reached', UsageThresholdReachedPayload>
+  | DomainEvent<'billing', 'chargeback_raised', ChargebackRaisedPayload>;
+
+// -- Analytics domain --------------------------------------------------------
+
+export interface ChurnRiskUpdatedPayload extends Record<string, unknown> {
+  subscriptionId: string;
+  userId: string;
+  riskScore: number;
+  previousScore?: number;
+  factors: string[];
+}
+
+export interface CohortAggregatedPayload extends Record<string, unknown> {
+  cohortId: string;
+  period: string;
+  retentionRate: number;
+  size: number;
+}
+
+export interface MrrChangedPayload extends Record<string, unknown> {
+  previousMrr: number;
+  currentMrr: number;
+  delta: number;
+  currency: string;
+  periodStart: number;
+  periodEnd: number;
+}
+
+export type AnalyticsEvent =
+  | DomainEvent<'analytics', 'churn_risk_updated', ChurnRiskUpdatedPayload>
+  | DomainEvent<'analytics', 'cohort_aggregated', CohortAggregatedPayload>
+  | DomainEvent<'analytics', 'mrr_changed', MrrChangedPayload>;
+
+// -- Auth domain -------------------------------------------------------------
+
+export interface ApiKeyRotatedPayload extends Record<string, unknown> {
+  keyId: string;
+  merchantId: string;
+  rotatedAt: number;
+  expiresAt?: number;
+}
+
+export interface SsoSessionCreatedPayload extends Record<string, unknown> {
+  sessionId: string;
+  userId: string;
+  provider: string;
+  createdAt: number;
+  expiresAt: number;
+}
+
+export type AuthEvent =
+  | DomainEvent<'auth', 'api_key_rotated', ApiKeyRotatedPayload>
+  | DomainEvent<'auth', 'sso_session_created', SsoSessionCreatedPayload>;
+
+// -- Contract domain (Soroban) -----------------------------------------------
+
+export interface ContractInvokedPayload extends Record<string, unknown> {
+  contractId: string;
+  method: string;
+  caller: string;
+  ledger: number;
+  txHash: string;
+}
+
+export interface ContractUpgradedPayload extends Record<string, unknown> {
+  proxyId: string;
+  fromImplementation: string;
+  toImplementation: string;
+  upgradedAt: number;
+  ledger: number;
+}
+
+export type ContractEvent =
+  | DomainEvent<'contract', 'invoked', ContractInvokedPayload>
+  | DomainEvent<'contract', 'upgraded', ContractUpgradedPayload>;
+
+// -- Union -------------------------------------------------------------------
+
+export type AnyDomainEvent =
+  | SubscriptionEvent
+  | BillingEvent
+  | AnalyticsEvent
+  | AuthEvent
+  | ContractEvent;
+
+/** Extract the payload type for a specific event name */
+export type EventPayload<TName extends AnyDomainEvent['name']> = Extract<
+  AnyDomainEvent,
+  { name: TName }
+>['payload'];
+
+// ---------------------------------------------------------------------------
+// Schema Validation
+// ---------------------------------------------------------------------------
+
+export interface ValidationResult {
+  valid: boolean;
+  errors: string[];
+}
+
+type SchemaField = {
+  required?: boolean;
+  type: 'string' | 'number' | 'boolean' | 'object' | 'array';
+};
+type Schema = Record<string, SchemaField>;
+
+const PAYLOAD_SCHEMAS: Partial<Record<AnyDomainEvent['name'], Schema>> = {
+  'subscription.created': {
+    subscriptionId: { required: true, type: 'string' },
+    userId: { required: true, type: 'string' },
+    planId: { required: true, type: 'string' },
+    status: { required: true, type: 'string' },
+    billingCycle: { required: true, type: 'string' },
+    nextBillingDate: { required: true, type: 'number' },
+  },
+  'subscription.cancelled': {
+    subscriptionId: { required: true, type: 'string' },
+    userId: { required: true, type: 'string' },
+    cancelledAt: { required: true, type: 'number' },
+    effectiveAt: { required: true, type: 'number' },
+  },
+  'subscription.renewed': {
+    subscriptionId: { required: true, type: 'string' },
+    userId: { required: true, type: 'string' },
+    planId: { required: true, type: 'string' },
+    renewedAt: { required: true, type: 'number' },
+    nextBillingDate: { required: true, type: 'number' },
+    amount: { required: true, type: 'number' },
+    currency: { required: true, type: 'string' },
+  },
+  'billing.invoice_generated': {
+    invoiceId: { required: true, type: 'string' },
+    subscriptionId: { required: true, type: 'string' },
+    userId: { required: true, type: 'string' },
+    amount: { required: true, type: 'number' },
+    currency: { required: true, type: 'string' },
+    dueDate: { required: true, type: 'number' },
+  },
+  'billing.payment_captured': {
+    paymentId: { required: true, type: 'string' },
+    subscriptionId: { required: true, type: 'string' },
+    userId: { required: true, type: 'string' },
+    amount: { required: true, type: 'number' },
+    currency: { required: true, type: 'string' },
+    capturedAt: { required: true, type: 'number' },
+    gateway: { required: true, type: 'string' },
+  },
+  'billing.usage_threshold_reached': {
+    subscriptionId: { required: true, type: 'string' },
+    userId: { required: true, type: 'string' },
+    metricType: { required: true, type: 'string' },
+    usage: { required: true, type: 'number' },
+    limit: { required: true, type: 'number' },
+    ratio: { required: true, type: 'number' },
+    level: { required: true, type: 'string' },
+  },
+};
+
+export function validateEventPayload(
+  eventName: AnyDomainEvent['name'],
+  payload: Record<string, unknown>,
+): ValidationResult {
+  const schema = PAYLOAD_SCHEMAS[eventName];
+  if (!schema) return { valid: true, errors: [] };
+
+  const errors: string[] = [];
+  for (const [field, rule] of Object.entries(schema)) {
+    const value = payload[field];
+    if (rule.required && (value === undefined || value === null)) {
+      errors.push(`Missing required field: ${field}`);
+      continue;
+    }
+    if (value !== undefined && value !== null) {
+      const actualType = Array.isArray(value) ? 'array' : typeof value;
+      if (actualType !== rule.type) {
+        errors.push(`Field "${field}" expected ${rule.type}, got ${actualType}`);
+      }
+    }
+  }
+  return { valid: errors.length === 0, errors };
+}
+
+// ---------------------------------------------------------------------------
+// Event Sourcing Store
+// ---------------------------------------------------------------------------
+
+export interface EventStoreQuery {
+  aggregateId?: string;
+  domain?: AnyDomainEvent['domain'];
+  type?: string;
+  from?: number;
+  to?: number;
+  limit?: number;
+  includeArchived?: boolean;
+}
+
+export interface AggregateSnapshot {
+  aggregateId: string;
+  state: Record<string, unknown>;
+  lastSequence: number;
+  lastEventType: string;
+  updatedAt: number;
+}
+
+export interface EventSourcedStore {
+  append(event: AnyDomainEvent): void;
+  query(filter?: EventStoreQuery): AnyDomainEvent[];
+  replay(aggregateId: string, handler: (event: AnyDomainEvent) => void): void;
+  reconstruct(aggregateId: string): Record<string, unknown>;
+  snapshot(aggregateId: string): AggregateSnapshot;
+  archiveBefore(timestamp: number): number;
+}
+
+interface StoredRecord {
+  event: AnyDomainEvent;
+  archivedAt?: number;
+}
+
+export class InMemoryEventStore implements EventSourcedStore {
+  private readonly records: StoredRecord[] = [];
+  private readonly sequences = new Map<string, number>();
+
+  append(event: AnyDomainEvent): void {
+    this.records.push({ event });
+    if (event.aggregateId) {
+      this.sequences.set(
+        event.aggregateId,
+        Math.max(this.sequences.get(event.aggregateId) ?? 0, event.sequence),
+      );
+    }
+  }
+
+  query(filter: EventStoreQuery = {}): AnyDomainEvent[] {
+    return this.records
+      .filter(({ event, archivedAt }) => {
+        if (!filter.includeArchived && archivedAt) return false;
+        if (filter.aggregateId && event.aggregateId !== filter.aggregateId) return false;
+        if (filter.domain && event.domain !== filter.domain) return false;
+        if (filter.type && event.type !== filter.type) return false;
+        if (filter.from && event.occurredAt < filter.from) return false;
+        if (filter.to && event.occurredAt > filter.to) return false;
+        return true;
+      })
+      .map(({ event }) => event)
+      .slice(0, filter.limit ?? Number.MAX_SAFE_INTEGER);
+  }
+
+  replay(aggregateId: string, handler: (event: AnyDomainEvent) => void): void {
+    this.query({ aggregateId, includeArchived: true })
+      .sort((a, b) => a.sequence - b.sequence)
+      .forEach(handler);
+  }
+
+  reconstruct(aggregateId: string): Record<string, unknown> {
+    return this.query({ aggregateId, includeArchived: true })
+      .sort((a, b) => a.sequence - b.sequence)
+      .reduce<Record<string, unknown>>(
+        (state, event) => ({
+          ...state,
+          ...(event.payload as Record<string, unknown>),
+          id: aggregateId,
+          lastEventType: event.name,
+          updatedAt: event.occurredAt,
+        }),
+        { id: aggregateId },
+      );
+  }
+
+  snapshot(aggregateId: string): AggregateSnapshot {
+    const events = this.query({ aggregateId, includeArchived: true }).sort(
+      (a, b) => a.sequence - b.sequence,
+    );
+    const last = events[events.length - 1];
+    return {
+      aggregateId,
+      state: this.reconstruct(aggregateId),
+      lastSequence: last?.sequence ?? 0,
+      lastEventType: last?.name ?? '',
+      updatedAt: last?.occurredAt ?? 0,
+    };
+  }
+
+  archiveBefore(timestamp: number): number {
+    let count = 0;
+    for (const record of this.records) {
+      if (!record.archivedAt && record.event.occurredAt < timestamp) {
+        record.archivedAt = Date.now();
+        count++;
+      }
+    }
+    return count;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Event Bus
+// ---------------------------------------------------------------------------
+
+export type EventHandler<T extends AnyDomainEvent = AnyDomainEvent> = (
+  event: T,
+) => void | Promise<void>;
+
+export type EventFilter<T extends AnyDomainEvent = AnyDomainEvent> = (event: T) => boolean;
+
+export interface SubscriptionOptions<T extends AnyDomainEvent = AnyDomainEvent> {
+  filter?: EventFilter<T>;
+}
+
+export interface EventSubscription {
+  readonly id: string;
+  unsubscribe(): void;
+}
+
+export interface EventBusMetrics {
+  published: number;
+  handled: number;
+  errors: number;
+  avgHandlerLatencyMs: number;
+  countByName: Record<string, number>;
+  throughputPerSecond: number;
+}
+
+export interface IEventBus {
+  publish<T extends AnyDomainEvent>(event: T): Promise<void>;
+  subscribe<T extends AnyDomainEvent>(
+    name: T['name'] | '*',
+    handler: EventHandler<T>,
+    options?: SubscriptionOptions<T>,
+  ): EventSubscription;
+  getMetrics(): EventBusMetrics;
+  resetMetrics(): void;
+}
+
+export class EventValidationError extends Error {
+  constructor(
+    public readonly eventName: string,
+    public readonly validationErrors: string[],
+  ) {
+    super(`Event "${eventName}" failed validation: ${validationErrors.join('; ')}`);
+    this.name = 'EventValidationError';
+    Object.setPrototypeOf(this, new.target.prototype);
+  }
+}
+
+interface RegisteredHandler {
+  id: string;
+  name: string;
+  handler: EventHandler<AnyDomainEvent>;
+  filter?: EventFilter<AnyDomainEvent>;
+}
+
+export class EventBus implements IEventBus {
+  private readonly handlers: RegisteredHandler[] = [];
+  private nextId = 1;
+
+  private published = 0;
+  private handled = 0;
+  private errors = 0;
+  private handlerLatencies: number[] = [];
+  private recentTimestamps: number[] = [];
+  private readonly countByName: Record<string, number> = {};
+
+  async publish<T extends AnyDomainEvent>(event: T): Promise<void> {
+    const result = validateEventPayload(
+      event.name as AnyDomainEvent['name'],
+      event.payload as Record<string, unknown>,
+    );
+    if (!result.valid) {
+      throw new EventValidationError(event.name, result.errors);
+    }
+
+    this.published++;
+    this.countByName[event.name] = (this.countByName[event.name] ?? 0) + 1;
+    this.recentTimestamps.push(Date.now());
+
+    const matching = this.handlers.filter((h) => h.name === '*' || h.name === event.name);
+    for (const h of matching) {
+      if (h.filter && !h.filter(event)) continue;
+      const start = Date.now();
+      try {
+        await h.handler(event);
+        this.handled++;
+      } catch {
+        this.errors++;
+      } finally {
+        this.handlerLatencies.push(Date.now() - start);
+        if (this.handlerLatencies.length > 10_000) this.handlerLatencies.shift();
+      }
+    }
+  }
+
+  subscribe<T extends AnyDomainEvent>(
+    name: T['name'] | '*',
+    handler: EventHandler<T>,
+    options?: SubscriptionOptions<T>,
+  ): EventSubscription {
+    const id = `esub_${this.nextId++}`;
+    this.handlers.push({
+      id,
+      name,
+      handler: handler as EventHandler<AnyDomainEvent>,
+      filter: options?.filter as EventFilter<AnyDomainEvent> | undefined,
+    });
+    return {
+      id,
+      unsubscribe: () => {
+        const idx = this.handlers.findIndex((h) => h.id === id);
+        if (idx !== -1) this.handlers.splice(idx, 1);
+      },
+    };
+  }
+
+  getMetrics(): EventBusMetrics {
+    const cutoff = Date.now() - 60_000;
+    this.recentTimestamps = this.recentTimestamps.filter((t) => t >= cutoff);
+    const avg =
+      this.handlerLatencies.length === 0
+        ? 0
+        : this.handlerLatencies.reduce((a, b) => a + b, 0) / this.handlerLatencies.length;
+    return {
+      published: this.published,
+      handled: this.handled,
+      errors: this.errors,
+      avgHandlerLatencyMs: Math.round(avg * 100) / 100,
+      countByName: { ...this.countByName },
+      throughputPerSecond: Math.round((this.recentTimestamps.length / 60) * 100) / 100,
+    };
+  }
+
+  resetMetrics(): void {
+    this.published = 0;
+    this.handled = 0;
+    this.errors = 0;
+    this.handlerLatencies = [];
+    this.recentTimestamps = [];
+    for (const key of Object.keys(this.countByName)) delete this.countByName[key];
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Event Builder
+// ---------------------------------------------------------------------------
+
+let _globalSequence = 0;
+
+/**
+ * Constructs a fully-typed domain event ready for publishing.
+ *
+ * @example
+ * const event = buildEvent('subscription', 'created', payload, { aggregateId: sub.id });
+ * await eventBus.publish(event);
+ */
+export function buildEvent<
+  TDomain extends AnyDomainEvent['domain'],
+  TType extends string,
+  TPayload extends Record<string, unknown>,
+>(
+  domain: TDomain,
+  type: TType,
+  payload: TPayload,
+  opts: {
+    aggregateId?: string;
+    correlationId?: string;
+    schemaVersion?: number;
+    occurredAt?: number;
+  } = {},
+): DomainEvent<TDomain, TType, TPayload> {
+  const seq = ++_globalSequence;
+  return {
+    id: `evt_${Date.now().toString(36)}_${seq}`,
+    domain,
+    type,
+    name: `${domain}.${type}` as `${TDomain}.${TType}`,
+    payload,
+    occurredAt: opts.occurredAt ?? Date.now(),
+    sequence: seq,
+    schemaVersion: opts.schemaVersion ?? 1,
+    correlationId: opts.correlationId,
+    aggregateId: opts.aggregateId,
+  };
+}
+
+// ---------------------------------------------------------------------------
+// Testing Utilities
+// ---------------------------------------------------------------------------
+
+/** Collects published events for assertions in tests. */
+export class EventCollector {
+  private readonly collected: AnyDomainEvent[] = [];
+  private readonly sub: EventSubscription;
+
+  constructor(bus: IEventBus, name: AnyDomainEvent['name'] | '*' = '*') {
+    this.sub = bus.subscribe(name, (event) => {
+      this.collected.push(event);
+    });
+  }
+
+  all(): AnyDomainEvent[] {
+    return [...this.collected];
+  }
+
+  ofName<T extends AnyDomainEvent>(name: T['name']): T[] {
+    return this.collected.filter((e) => e.name === name) as T[];
+  }
+
+  assertCount(expected: number): void {
+    if (this.collected.length !== expected) {
+      throw new Error(
+        `EventCollector: expected ${expected} events, got ${this.collected.length}`,
+      );
+    }
+  }
+
+  assertLastEventName(name: AnyDomainEvent['name']): void {
+    const last = this.collected[this.collected.length - 1];
+    if (!last || last.name !== name) {
+      throw new Error(
+        `EventCollector: expected last event "${name}", got "${last?.name ?? 'none'}"`,
+      );
+    }
+  }
+
+  clear(): void {
+    this.collected.length = 0;
+  }
+
+  dispose(): void {
+    this.sub.unsubscribe();
+  }
+}
+
+/**
+ * SpyEventBus wraps a real EventBus and records every publish call.
+ * Drop-in replacement for tests that need to inspect emitted events.
+ */
+export class SpyEventBus implements IEventBus {
+  private readonly inner = new EventBus();
+  readonly published: AnyDomainEvent[] = [];
+
+  async publish<T extends AnyDomainEvent>(event: T): Promise<void> {
+    this.published.push(event);
+    await this.inner.publish(event);
+  }
+
+  subscribe<T extends AnyDomainEvent>(
+    name: T['name'] | '*',
+    handler: EventHandler<T>,
+    options?: SubscriptionOptions<T>,
+  ): EventSubscription {
+    return this.inner.subscribe(name, handler, options);
+  }
+
+  getMetrics(): EventBusMetrics {
+    return this.inner.getMetrics();
+  }
+
+  resetMetrics(): void {
+    this.published.length = 0;
+    this.inner.resetMetrics();
+  }
+
+  assertPublished(name: AnyDomainEvent['name']): void {
+    if (!this.published.some((e) => e.name === name)) {
+      const names = this.published.map((e) => e.name).join(', ');
+      throw new Error(
+        `SpyEventBus: expected "${name}" to be published. Got: [${names}]`,
+      );
+    }
+  }
+
+  assertEmpty(): void {
+    if (this.published.length > 0) {
+      throw new Error(
+        `SpyEventBus: expected no events, but got ${this.published.length}: [${this.published.map((e) => e.name).join(', ')}]`,
+      );
+    }
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Prometheus Metrics Export
+// ---------------------------------------------------------------------------
+
+export function eventBusPrometheusMetrics(
+  bus: IEventBus,
+  namespace = 'subtrackr_event_bus',
+): string {
+  const m = bus.getMetrics();
+  const lines = [
+    `# HELP ${namespace}_published_total Total events published`,
+    `# TYPE ${namespace}_published_total counter`,
+    `${namespace}_published_total ${m.published}`,
+    `# HELP ${namespace}_handled_total Total handler invocations`,
+    `# TYPE ${namespace}_handled_total counter`,
+    `${namespace}_handled_total ${m.handled}`,
+    `# HELP ${namespace}_errors_total Total handler errors`,
+    `# TYPE ${namespace}_errors_total counter`,
+    `${namespace}_errors_total ${m.errors}`,
+    `# HELP ${namespace}_avg_handler_latency_ms Average handler latency`,
+    `# TYPE ${namespace}_avg_handler_latency_ms gauge`,
+    `${namespace}_avg_handler_latency_ms ${m.avgHandlerLatencyMs}`,
+    `# HELP ${namespace}_throughput_per_second Events per second (last 60s)`,
+    `# TYPE ${namespace}_throughput_per_second gauge`,
+    `${namespace}_throughput_per_second ${m.throughputPerSecond}`,
+  ];
+  for (const [name, count] of Object.entries(m.countByName)) {
+    lines.push(`${namespace}_by_name_total{event="${name}"} ${count}`);
+  }
+  return lines.join('\n');
+}
+
+// ---------------------------------------------------------------------------
+// Singletons
+// ---------------------------------------------------------------------------
+
+export const eventBus = new EventBus();
+export const eventStore = new InMemoryEventStore();
\ No newline at end of file
diff --git a/backend/services/shared/index.ts b/backend/services/shared/index.ts
index 01bb08c7..7d632391 100644
--- a/backend/services/shared/index.ts
+++ b/backend/services/shared/index.ts
@@ -61,3 +61,111 @@ export type {
 } from './apiResponse';
 export type { TransactionStatus, AlertSeverity, AlertChannel, TransactionEvent, Metric, Alert, AlertRule, AlertChannelConfig, DashboardSnapshot } from './types';
 export { MonitoringService, monitoringService } from './monitoring';
+
+// ── Typed Event Bus ────────────────────────────────────────────────────────────
+export {
+  EventBus,
+  SpyEventBus,
+  EventCollector,
+  InMemoryEventStore,
+  buildEvent,
+  validateEventPayload,
+  EventValidationError,
+  eventBus,
+  eventStore,
+  eventBusPrometheusMetrics,
+} from './events';
+export type {
+  DomainEvent,
+  AnyDomainEvent,
+  EventPayload,
+  EventHandler,
+  EventFilter,
+  EventSubscription,
+  SubscriptionOptions,
+  EventBusMetrics,
+  IEventBus,
+  EventSourcedStore,
+  EventStoreQuery,
+  AggregateSnapshot,
+  ValidationResult,
+  // Subscription domain payloads
+  SubscriptionCreatedPayload,
+  SubscriptionCancelledPayload,
+  SubscriptionRenewedPayload,
+  SubscriptionUpgradedPayload,
+  SubscriptionPausedPayload,
+  SubscriptionResumedPayload,
+  SubscriptionPaymentFailedPayload,
+  SubscriptionEvent,
+  // Billing domain payloads
+  InvoiceGeneratedPayload,
+  PaymentCapturedPayload,
+  UsageThresholdReachedPayload,
+  ChargebackRaisedPayload,
+  BillingEvent,
+  // Analytics domain payloads
+  ChurnRiskUpdatedPayload,
+  CohortAggregatedPayload,
+  MrrChangedPayload,
+  AnalyticsEvent,
+  // Auth domain payloads
+  ApiKeyRotatedPayload,
+  SsoSessionCreatedPayload,
+  AuthEvent,
+  // Contract domain payloads
+  ContractInvokedPayload,
+  ContractUpgradedPayload,
+  ContractEvent,
+} from './events';
+
+// ── Generic Cache Service ──────────────────────────────────────────────────────
+export { CacheService, NullCacheService, wireInvalidation } from './cache';
+export type {
+  ICacheService,
+  CacheServiceConfig,
+  CacheMetrics,
+  InvalidationRule,
+} from './cache';
+
+// ── Compression Middleware ────────────────────────────────────────────────────
+export {
+  applyCompression,
+  negotiateEncoding,
+  generateETag,
+  isETagMatch,
+  compressionMetrics,
+  compressionPrometheusMetrics,
+} from './compression';
+export type {
+  CompressionConfig,
+  CompressionMetrics,
+  CompressionMiddlewareOptions,
+  Encoding,
+} from './compression';
+
+// ── Cursor Pagination + Field Selection ───────────────────────────────────────
+export {
+  encodeCursor,
+  decodeCursor,
+  buildCursorClause,
+  buildPage,
+  parseFieldSelection,
+  selectFields,
+  selectFieldsAll,
+} from './pagination';
+export type {
+  CursorPayload,
+  PageOptions,
+  PageResult,
+  SqlCursorClause,
+} from './pagination';
+
+// ── Connection Pool Monitor ───────────────────────────────────────────────────
+export { MonitoredPool, wrapWithMonitor } from './poolMonitor';
+export type {
+  PoolMonitorConfig,
+  PoolStats,
+  LeakRecord,
+  PoolTuningRecommendation,
+} from './poolMonitor';
diff --git a/backend/services/shared/pagination.ts b/backend/services/shared/pagination.ts
new file mode 100644
index 00000000..65ab0c85
--- /dev/null
+++ b/backend/services/shared/pagination.ts
@@ -0,0 +1,212 @@
+/**
+ * Cursor-Based Pagination + Field Selection — SubTrackr
+ *
+ * Provides:
+ *  - Opaque, tamper-evident cursor encoding (base64url JSON)
+ *  - Type-safe page builders for any ordered result set
+ *  - Field selection: clients pass `?fields=id,name,status` to trim payloads
+ *  - Integration with the existing ApiResponse / PaginationMeta envelope
+ */
+
+import { createHmac } from 'node:crypto';
+
+// ─── Cursor Encoding ──────────────────────────────────────────────────────────
+
+const CURSOR_SECRET = process.env['CURSOR_HMAC_SECRET'] ?? 'subtrackr-cursor-secret';
+const CURSOR_VERSION = 1;
+
+export interface CursorPayload {
+  v: number;
+  field: string;
+  value: unknown;
+  id: string;
+  dir: 'asc' | 'desc';
+}
+
+function sign(data: string): string {
+  return createHmac('sha256', CURSOR_SECRET).update(data).digest('base64url').slice(0, 16);
+}
+
+/** Encode a cursor payload into an opaque base64url string. */
+export function encodeCursor(payload: Omit<CursorPayload, 'v'>): string {
+  const full: CursorPayload = { v: CURSOR_VERSION, ...payload };
+  const json = JSON.stringify(full);
+  const sig = sign(json);
+  return Buffer.from(`${json}|${sig}`, 'utf8').toString('base64url');
+}
+
+/** Decode and verify a cursor. Returns null when invalid or tampered. */
+export function decodeCursor(cursor: string): CursorPayload | null {
+  try {
+    const raw = Buffer.from(cursor, 'base64url').toString('utf8');
+    const sepIdx = raw.lastIndexOf('|');
+    if (sepIdx === -1) return null;
+    const json = raw.slice(0, sepIdx);
+    const sig = raw.slice(sepIdx + 1);
+    if (sign(json) !== sig) return null;
+    const payload = JSON.parse(json) as CursorPayload;
+    if (payload.v !== CURSOR_VERSION) return null;
+    return payload;
+  } catch {
+    return null;
+  }
+}
+
+// ─── Page Builder ─────────────────────────────────────────────────────────────
+
+export interface PageOptions {
+  /** Opaque cursor from the previous page response. Absent on first page. */
+  cursor?: string;
+  /** Max records to return. Clamped to [1, maxLimit]. Default: 20 */
+  limit?: number;
+  /** Hard cap on limit. Default: 100 */
+  maxLimit?: number;
+  /** Sort direction. Default: 'asc' */
+  direction?: 'asc' | 'desc';
+  /** Which field the cursor tracks. Default: 'id' */
+  sortField?: string;
+}
+
+export interface PageResult<T> {
+  items: T[];
+  /** Opaque cursor pointing to the next page. Absent on last page. */
+  nextCursor?: string;
+  /** Whether more records exist. */
+  hasMore: boolean;
+  /** Total matching records (optional — may be omitted for performance). */
+  total?: number;
+}
+
+export interface SqlCursorClause {
+  /** WHERE clause fragment to add (e.g. `"id" > $1`) */
+  where: string;
+  /** Bind parameter value */
+  param: unknown;
+  /** ORDER BY clause (e.g. `"id" ASC`) */
+  orderBy: string;
+  /** LIMIT value */
+  limit: number;
+}
+
+/**
+ * Parse pagination options and produce a SQL cursor WHERE clause.
+ *
+ * @example
+ * const { where, param, orderBy, limit } = buildCursorClause({ cursor, limit: 20 });
+ * const sql = `SELECT * FROM subscriptions ${where ? `WHERE ${where}` : ''} ORDER BY ${orderBy} LIMIT ${limit + 1}`;
+ */
+export function buildCursorClause(options: PageOptions = {}): SqlCursorClause {
+  const dir = options.direction ?? 'asc';
+  const sortField = options.sortField ?? 'id';
+  const rawLimit = options.limit ?? 20;
+  const limit = Math.min(Math.max(1, rawLimit), options.maxLimit ?? 100);
+  const op = dir === 'asc' ? '>' : '<';
+  const orderDir = dir.toUpperCase();
+
+  if (!options.cursor) {
+    return {
+      where: '',
+      param: null,
+      orderBy: `"${sortField}" ${orderDir}`,
+      limit,
+    };
+  }
+
+  const decoded = decodeCursor(options.cursor);
+  if (!decoded) {
+    return { where: '', param: null, orderBy: `"${sortField}" ${orderDir}`, limit };
+  }
+
+  return {
+    where: `("${decoded.field}", id) ${op} ($1, $2)`,
+    param: decoded.value,
+    orderBy: `"${decoded.field}" ${orderDir}, id ${orderDir}`,
+    limit,
+  };
+}
+
+/**
+ * Build a PageResult from a raw DB row set.
+ *
+ * Fetches `limit + 1` rows; the extra row signals `hasMore`.
+ *
+ * @example
+ * const rows = await pool.query(`SELECT * FROM subscriptions ... LIMIT ${limit + 1}`);
+ * return buildPage(rows, limit, (row) => row.createdAt, (row) => row.id);
+ */
+export function buildPage<T extends Record<string, unknown>>(
+  rows: T[],
+  limit: number,
+  getCursorValue: (row: T) => unknown,
+  getId: (row: T) => string,
+  direction: 'asc' | 'desc' = 'asc',
+  sortField = 'id',
+  total?: number,
+): PageResult<T> {
+  const hasMore = rows.length > limit;
+  const items = hasMore ? rows.slice(0, limit) : rows;
+  const lastItem = items[items.length - 1];
+
+  let nextCursor: string | undefined;
+  if (hasMore && lastItem) {
+    nextCursor = encodeCursor({
+      field: sortField,
+      value: getCursorValue(lastItem),
+      id: getId(lastItem),
+      dir: direction,
+    });
+  }
+
+  return { items, nextCursor, hasMore, total };
+}
+
+// ─── Field Selection ──────────────────────────────────────────────────────────
+
+/**
+ * Parse a `?fields=id,name,status` query parameter into a set of allowed keys.
+ * Returns null when the parameter is absent (return all fields).
+ */
+export function parseFieldSelection(fieldsParam: string | undefined): Set<string> | null {
+  if (!fieldsParam?.trim()) return null;
+  const fields = fieldsParam
+    .split(',')
+    .map((f) => f.trim())
+    .filter(Boolean);
+  return fields.length > 0 ? new Set(fields) : null;
+}
+
+/**
+ * Project a record to only the requested fields.
+ * Always includes `id` to ensure response objects are identifiable.
+ *
+ * @example
+ * const slim = selectFields(subscription, parseFieldSelection(req.query.fields));
+ */
+export function selectFields<T extends Record<string, unknown>>(
+  record: T,
+  fields: Set<string> | null,
+): Partial<T> {
+  if (!fields) return record;
+  const result: Partial<T> = {};
+  // id is always included for linkability
+  if ('id' in record) {
+    (result as Record<string, unknown>)['id'] = record['id'];
+  }
+  for (const key of fields) {
+    if (key in record) {
+      (result as Record<string, unknown>)[key] = record[key];
+    }
+  }
+  return result;
+}
+
+/**
+ * Apply field selection to an array of records.
+ */
+export function selectFieldsAll<T extends Record<string, unknown>>(
+  records: T[],
+  fields: Set<string> | null,
+): Partial<T>[] {
+  if (!fields) return records;
+  return records.map((r) => selectFields(r, fields));
+}
diff --git a/backend/services/shared/poolMonitor.ts b/backend/services/shared/poolMonitor.ts
new file mode 100644
index 00000000..28f49271
--- /dev/null
+++ b/backend/services/shared/poolMonitor.ts
@@ -0,0 +1,320 @@
+/**
+ * Connection Pool Monitor — SubTrackr
+ *
+ * Wraps any Pool (PostgreSQL primary, replica, or ES) with:
+ *  - Real-time utilisation metrics (total, idle, waiting, checked-out)
+ *  - Pool exhaustion alerts (waiting queue > threshold)
+ *  - Connection leak detection (checked-out longer than leakThresholdMs)
+ *  - Query timeout configuration enforcement
+ *  - Prometheus metrics export
+ *  - Pool tuning recommendations (based on observed peak utilisation)
+ */
+
+import type { Pool, PoolClient, QueryResult } from '../../shared/db/connectionPool';
+import { logger } from './logging';
+
+// ─── Configuration ────────────────────────────────────────────────────────────
+
+export interface PoolMonitorConfig {
+  /** Name used in metrics labels (e.g. "primary", "replica-1"). */
+  name: string;
+  /** Poll interval for metric snapshots in ms. Default: 5000 */
+  pollIntervalMs?: number;
+  /** Waiting connections above this triggers an exhaustion alert. Default: 5 */
+  exhaustionThreshold?: number;
+  /** A checked-out client older than this is considered leaked (ms). Default: 30 000 */
+  leakThresholdMs?: number;
+  /** Query timeout injected via SET statement_timeout on each connection (ms). Default: 30 000 */
+  queryTimeoutMs?: number;
+  /** Maximum pool size — used for utilisation % calculation. */
+  maxConnections?: number;
+  /** Called when pool exhaustion is detected. */
+  onExhaustion?: (stats: PoolStats) => void;
+  /** Called when a leaked connection is detected. */
+  onLeak?: (leak: LeakRecord) => void;
+}
+
+const DEFAULTS: Required<Omit<PoolMonitorConfig, 'name' | 'onExhaustion' | 'onLeak'>> = {
+  pollIntervalMs: 5_000,
+  exhaustionThreshold: 5,
+  leakThresholdMs: 30_000,
+  queryTimeoutMs: 30_000,
+  maxConnections: 20,
+};
+
+// ─── Types ────────────────────────────────────────────────────────────────────
+
+export interface PoolStats {
+  name: string;
+  total: number;
+  idle: number;
+  waiting: number;
+  checkedOut: number;
+  utilizationPct: number;
+  leakedConnections: number;
+  peakCheckedOut: number;
+  totalQueries: number;
+  totalErrors: number;
+  avgQueryLatencyMs: number;
+  p99QueryLatencyMs: number;
+}
+
+export interface LeakRecord {
+  poolName: string;
+  checkedOutAt: number;
+  ageMs: number;
+  origin?: string;
+}
+
+export interface PoolTuningRecommendation {
+  currentMax: number;
+  recommendedMax: number;
+  reason: string;
+}
+
+// ─── Monitored Pool ───────────────────────────────────────────────────────────
+
+interface CheckoutEntry {
+  client: PoolClient;
+  checkedOutAt: number;
+  origin?: string;
+}
+
+export class MonitoredPool implements Pool {
+  private readonly cfg: Required<Omit<PoolMonitorConfig, 'onExhaustion' | 'onLeak'>> &
+    Pick<PoolMonitorConfig, 'onExhaustion' | 'onLeak'>;
+
+  private readonly checkouts = new Map<PoolClient, CheckoutEntry>();
+  private leakedConnections = 0;
+  private peakCheckedOut = 0;
+  private totalQueries = 0;
+  private totalErrors = 0;
+  private readonly latencies: number[] = [];
+  private pollTimer?: ReturnType<typeof setInterval>;
+  private readonly snapshots: PoolStats[] = [];
+
+  constructor(
+    private readonly inner: Pool,
+    config: PoolMonitorConfig,
+  ) {
+    const { onExhaustion, onLeak, ...rest } = config;
+    this.cfg = { ...DEFAULTS, ...rest, onExhaustion, onLeak };
+    this.startPolling();
+  }
+
+  // ── Pool interface ──────────────────────────────────────────────────────────
+
+  get totalCount(): number { return this.inner.totalCount; }
+  get idleCount(): number { return this.inner.idleCount; }
+  get waitingCount(): number { return this.inner.waitingCount; }
+
+  on(event: 'error', handler: (err: Error) => void): void {
+    this.inner.on(event, handler);
+  }
+
+  async query<T = unknown>(sql: string, params?: unknown[]): Promise<QueryResult<T>> {
+    const start = Date.now();
+    this.totalQueries++;
+    try {
+      const result = await this.inner.query<T>(sql, params);
+      this.recordLatency(Date.now() - start);
+      return result;
+    } catch (err) {
+      this.totalErrors++;
+      this.recordLatency(Date.now() - start);
+      throw err;
+    }
+  }
+
+  async connect(): Promise<PoolClient> {
+    const client = await this.inner.connect();
+
+    // Inject query timeout on new connections
+    try {
+      await client.query(`SET statement_timeout = ${this.cfg.queryTimeoutMs}`);
+    } catch {
+      logger.warn(`[PoolMonitor:${this.cfg.name}] Could not set statement_timeout`);
+    }
+
+    const entry: CheckoutEntry = { client, checkedOutAt: Date.now() };
+    this.checkouts.set(client, entry);
+
+    const checkedOut = this.checkouts.size;
+    if (checkedOut > this.peakCheckedOut) this.peakCheckedOut = checkedOut;
+
+    // Wrap release() to clean up tracking entry
+    const originalRelease = client.release.bind(client);
+    client.release = () => {
+      this.checkouts.delete(client);
+      originalRelease();
+    };
+
+    return client;
+  }
+
+  async end(): Promise<void> {
+    this.stopPolling();
+    await this.inner.end();
+  }
+
+  // ── Stats & Monitoring ──────────────────────────────────────────────────────
+
+  getStats(): PoolStats {
+    const total = this.inner.totalCount;
+    const max = this.cfg.maxConnections;
+    const checkedOut = this.checkouts.size;
+    return {
+      name: this.cfg.name,
+      total,
+      idle: this.inner.idleCount,
+      waiting: this.inner.waitingCount,
+      checkedOut,
+      utilizationPct: max > 0 ? Math.round((checkedOut / max) * 100) : 0,
+      leakedConnections: this.leakedConnections,
+      peakCheckedOut: this.peakCheckedOut,
+      totalQueries: this.totalQueries,
+      totalErrors: this.totalErrors,
+      avgQueryLatencyMs: this.avgLatency(),
+      p99QueryLatencyMs: this.p99Latency(),
+    };
+  }
+
+  /** Snapshot history for dashboard (last 60 polls). */
+  getHistory(): PoolStats[] {
+    return [...this.snapshots];
+  }
+
+  getTuningRecommendation(): PoolTuningRecommendation {
+    const current = this.cfg.maxConnections;
+    // Recommend 20% headroom above peak observed checked-out count
+    const recommended = Math.max(current, Math.ceil(this.peakCheckedOut * 1.2));
+    return {
+      currentMax: current,
+      recommendedMax: recommended,
+      reason:
+        recommended > current
+          ? `Peak concurrent connections was ${this.peakCheckedOut}; adding 20% headroom`
+          : 'Current pool size appears sufficient for observed load',
+    };
+  }
+
+  prometheusMetrics(namespace = 'subtrackr_db_pool'): string {
+    const s = this.getStats();
+    const label = `pool="${s.name}"`;
+    return [
+      `# HELP ${namespace}_total_connections Active connections in pool`,
+      `# TYPE ${namespace}_total_connections gauge`,
+      `${namespace}_total_connections{${label}} ${s.total}`,
+      `# HELP ${namespace}_idle_connections Idle connections`,
+      `# TYPE ${namespace}_idle_connections gauge`,
+      `${namespace}_idle_connections{${label}} ${s.idle}`,
+      `# HELP ${namespace}_waiting_connections Requests waiting for a connection`,
+      `# TYPE ${namespace}_waiting_connections gauge`,
+      `${namespace}_waiting_connections{${label}} ${s.waiting}`,
+      `# HELP ${namespace}_checked_out_connections Currently checked-out connections`,
+      `# TYPE ${namespace}_checked_out_connections gauge`,
+      `${namespace}_checked_out_connections{${label}} ${s.checkedOut}`,
+      `# HELP ${namespace}_utilization_pct Pool utilization percentage`,
+      `# TYPE ${namespace}_utilization_pct gauge`,
+      `${namespace}_utilization_pct{${label}} ${s.utilizationPct}`,
+      `# HELP ${namespace}_leaked_connections_total Leaked connections detected and force-closed`,
+      `# TYPE ${namespace}_leaked_connections_total counter`,
+      `${namespace}_leaked_connections_total{${label}} ${s.leakedConnections}`,
+      `# HELP ${namespace}_queries_total Total queries executed`,
+      `# TYPE ${namespace}_queries_total counter`,
+      `${namespace}_queries_total{${label}} ${s.totalQueries}`,
+      `# HELP ${namespace}_errors_total Total query errors`,
+      `# TYPE ${namespace}_errors_total counter`,
+      `${namespace}_errors_total{${label}} ${s.totalErrors}`,
+      `# HELP ${namespace}_avg_latency_ms Average query latency`,
+      `# TYPE ${namespace}_avg_latency_ms gauge`,
+      `${namespace}_avg_latency_ms{${label}} ${s.avgQueryLatencyMs}`,
+      `# HELP ${namespace}_p99_latency_ms P99 query latency`,
+      `# TYPE ${namespace}_p99_latency_ms gauge`,
+      `${namespace}_p99_latency_ms{${label}} ${s.p99QueryLatencyMs}`,
+    ].join('\n');
+  }
+
+  // ── Private ─────────────────────────────────────────────────────────────────
+
+  private startPolling(): void {
+    this.pollTimer = setInterval(() => this.poll(), this.cfg.pollIntervalMs);
+    if (typeof this.pollTimer.unref === 'function') this.pollTimer.unref();
+  }
+
+  private stopPolling(): void {
+    if (this.pollTimer) {
+      clearInterval(this.pollTimer);
+      this.pollTimer = undefined;
+    }
+  }
+
+  private poll(): void {
+    const stats = this.getStats();
+
+    // Snapshot for history (keep last 60)
+    this.snapshots.push(stats);
+    if (this.snapshots.length > 60) this.snapshots.shift();
+
+    // Exhaustion alert
+    if (stats.waiting >= this.cfg.exhaustionThreshold) {
+      logger.warn(`[PoolMonitor:${this.cfg.name}] Pool exhaustion detected`, {
+        waiting: stats.waiting,
+        threshold: this.cfg.exhaustionThreshold,
+        checkedOut: stats.checkedOut,
+      });
+      this.cfg.onExhaustion?.(stats);
+    }
+
+    // Leak detection
+    const now = Date.now();
+    for (const [, entry] of this.checkouts) {
+      const age = now - entry.checkedOutAt;
+      if (age > this.cfg.leakThresholdMs) {
+        this.leakedConnections++;
+        const leak: LeakRecord = {
+          poolName: this.cfg.name,
+          checkedOutAt: entry.checkedOutAt,
+          ageMs: age,
+          origin: entry.origin,
+        };
+        logger.warn(`[PoolMonitor:${this.cfg.name}] Leaked connection detected`, leak);
+        this.cfg.onLeak?.(leak);
+        // Force release
+        try { entry.client.release(); } catch { /* already gone */ }
+        this.checkouts.delete(entry.client);
+      }
+    }
+  }
+
+  private recordLatency(ms: number): void {
+    this.latencies.push(ms);
+    if (this.latencies.length > 10_000) this.latencies.shift();
+  }
+
+  private avgLatency(): number {
+    if (this.latencies.length === 0) return 0;
+    return Math.round(this.latencies.reduce((a, b) => a + b, 0) / this.latencies.length);
+  }
+
+  private p99Latency(): number {
+    if (this.latencies.length === 0) return 0;
+    const sorted = [...this.latencies].sort((a, b) => a - b);
+    const idx = Math.ceil(0.99 * sorted.length) - 1;
+    return sorted[Math.max(0, idx)] ?? 0;
+  }
+}
+
+// ─── Factory ──────────────────────────────────────────────────────────────────
+
+/**
+ * Wrap an existing Pool with monitoring.
+ *
+ * @example
+ * const pool = await getPool();
+ * const monitored = wrapWithMonitor(pool, { name: 'primary', maxConnections: 20 });
+ * container.register('IPool', monitored);
+ */
+export function wrapWithMonitor(pool: Pool, config: PoolMonitorConfig): MonitoredPool {
+  return new MonitoredPool(pool, config);
+}
diff --git a/developer-portal/docs/pagination-and-compression.md b/developer-portal/docs/pagination-and-compression.md
new file mode 100644
index 00000000..27ffa411
--- /dev/null
+++ b/developer-portal/docs/pagination-and-compression.md
@@ -0,0 +1,118 @@
+# Pagination, Field Selection & Response Compression
+
+## Compressed Responses
+
+The SubTrackr API compresses all JSON responses using **Brotli** (preferred) or **gzip** depending on your client's `Accept-Encoding` header.
+
+### Request Headers
+
+```http
+Accept-Encoding: br, gzip, deflate
+```
+
+### Response Headers
+
+```http
+Content-Encoding: br
+ETag: "abc123def456"
+Vary: Accept-Encoding
+Cache-Control: public, s-maxage=300, stale-while-revalidate=60
+Content-Length: 842
+```
+
+### Conditional Requests (304 Not Modified)
+
+Cache the `ETag` from the first response and send it on subsequent requests:
+
+```http
+If-None-Match: "abc123def456"
+```
+
+If the resource hasn't changed, the server returns `304 Not Modified` with an empty body — saving bandwidth entirely.
+
+---
+
+## Cursor-Based Pagination
+
+All list endpoints use cursor pagination. Do **not** use offset-based pagination — it is not supported.
+
+### Query Parameters
+
+| Parameter | Type | Default | Description |
+|---|---|---|---|
+| `limit` | integer | `20` | Records per page. Max: `100` |
+| `cursor` | string | — | Opaque token from previous response's `meta.pagination.cursor` |
+| `fields` | string | — | Comma-separated fields to include (see Field Selection) |
+
+### Example Request
+
+```http
+GET /subscriptions?limit=20&fields=id,status,planId,amount
+```
+
+### Example Response
+
+```json
+{
+  "success": true,
+  "data": [
+    { "id": "sub_abc", "status": "active", "planId": "plan_pro", "amount": 29.99 },
+    ...
+  ],
+  "meta": {
+    "timestamp": "2026-07-27T10:00:00.000Z",
+    "requestId": "req_xyz",
+    "apiVersion": 1,
+    "pagination": {
+      "cursor": "eyJ2IjoxLCJmaWVsZ...",
+      "hasMore": true,
+      "total": 1420
+    }
+  }
+}
+```
+
+### Fetching the Next Page
+
+Pass the `cursor` from `meta.pagination.cursor` to the next request:
+
+```http
+GET /subscriptions?limit=20&cursor=eyJ2IjoxLCJmaWVsZ...&fields=id,status,planId,amount
+```
+
+When `meta.pagination.hasMore` is `false`, you have reached the last page.
+
+### Important Notes
+
+- Cursors are **opaque** — do not attempt to parse, modify, or construct them
+- Cursors are **tamper-evident** — invalid cursors fall back to the first page
+- Cursors are **directional** — they encode the sort direction and field
+- Cursors are **not permanent** — do not store them for long-term use
+
+---
+
+## Field Selection
+
+Reduce response payload size by requesting only the fields you need.
+
+```http
+GET /subscriptions?fields=id,status,amount,currency
+```
+
+The `id` field is always included even if not specified — it is required for resource identity.
+
+### Example: Full response vs projected
+
+Without field selection — ~800 bytes:
+```json
+{ "id": "sub_abc", "status": "active", "planId": "plan_pro", "amount": 29.99,
+  "currency": "USD", "billingCycle": "monthly", "nextBillingDate": 1727000000000,
+  "userId": "user_123", "createdAt": "2026-01-01T00:00:00.000Z", "... more fields" }
+```
+
+With `?fields=id,status,amount` — ~120 bytes:
+```json
+{ "id": "sub_abc", "status": "active", "amount": 29.99 }
+```
+
+Combined with Brotli compression, field selection can reduce typical list response sizes by 80–95%.
diff --git a/docs/api-performance.md b/docs/api-performance.md
new file mode 100644
index 00000000..c104f100
--- /dev/null
+++ b/docs/api-performance.md
@@ -0,0 +1,140 @@
+# API Performance: Compression & Pagination
+
+## Response Compression
+
+`backend/services/shared/compression.ts`
+
+All API responses are compressed using Brotli-first, gzip-fallback negotiation.
+
+### Encoding Negotiation
+
+The server reads the client's `Accept-Encoding` header and selects the best supported format:
+
+| Priority | Encoding | Notes |
+|---|---|---|
+| 1 | `br` (Brotli) | Quality 4 by default — fast with ~30% better ratio than gzip |
+| 2 | `gzip` | Level 6 by default |
+| 3 | `identity` | Used when client doesn't support compression or body < 1 KB |
+
+### ETag Support
+
+Every compressible response receives an `ETag` derived from the SHA-256 of the **uncompressed** body, making the tag encoding-independent.
+
+Clients that send `If-None-Match: <etag>` receive a `304 Not Modified` when the content hasn't changed — saving bandwidth entirely.
+
+### Cache-Control Headers
+
+Plan endpoints include `Cache-Control: public, s-maxage=300, stale-while-revalidate=60` by default. Pass `defaultCacheControl` to `applyCompression` to customise per endpoint.
+
+### Usage
+
+```typescript
+import { applyCompression } from '../services/shared/compression';
+
+// In a request handler:
+await applyCompression(req, res, JSON.stringify(data), {
+  'Content-Type': 'application/json',
+}, {
+  defaultCacheControl: 'public, s-maxage=60',
+});
+```
+
+### Metrics
+
+```
+GET /metrics/compression
+```
+
+Exposes Prometheus counters:
+
+| Metric | Type | Description |
+|---|---|---|
+| `subtrackr_compression_requests_total` | counter | Total responses processed |
+| `subtrackr_compression_compressed_total` | counter | Responses that were compressed |
+| `subtrackr_compression_brotli_total` | counter | Compressed with Brotli |
+| `subtrackr_compression_gzip_total` | counter | Compressed with gzip |
+| `subtrackr_compression_original_bytes_total` | counter | Total uncompressed bytes |
+| `subtrackr_compression_compressed_bytes_total` | counter | Total bytes actually sent |
+| `subtrackr_compression_avg_ratio` | gauge | Average ratio (lower = better) |
+
+---
+
+## Cursor-Based Pagination
+
+`backend/services/shared/pagination.ts`
+
+### Why Cursor Pagination?
+
+Offset pagination (`OFFSET 100 LIMIT 20`) degrades at scale — the DB scans all prior rows. Cursor pagination uses a stable, ordered bookmark so each page is O(log N) regardless of depth.
+
+### Cursor Format
+
+Cursors are opaque `base64url` strings containing a signed JSON payload. Clients must treat them as opaque — do not parse or construct cursors manually.
+
+```
+eyJ2IjoxLCJmaWVsZCI6ImNyZWF0ZWRBdCIsInZhbHVlIjoxNzIyMDAwMDAwMDAwLCJpZCI6...
+```
+
+### Query Parameter
+
+```
+GET /subscriptions?limit=20&cursor=<opaque>&fields=id,status,amount
+```
+
+| Parameter | Description |
+|---|---|
+| `limit` | Page size (1–100, default 20) |
+| `cursor` | Opaque cursor from previous page's `nextCursor` |
+| `fields` | Comma-separated field list for projection |
+
+### Response Shape
+
+```json
+{
+  "success": true,
+  "data": [...],
+  "meta": {
+    "pagination": {
+      "cursor": "eyJ2Ij...",
+      "hasMore": true,
+      "total": 1420
+    }
+  }
+}
+```
+
+### Server-Side Integration
+
+```typescript
+import { buildCursorClause, buildPage, parseFieldSelection, selectFieldsAll } from '../services/shared/pagination';
+
+// 1. Parse options from query string
+const { cursor, limit, fields: fieldsParam } = req.query;
+const fieldSet = parseFieldSelection(fieldsParam);
+
+// 2. Build SQL clause
+const { where, param, orderBy, limit: pageLimit } = buildCursorClause({ cursor, limit: Number(limit) });
+
+// 3. Query with limit+1 to detect hasMore
+const sql = `SELECT * FROM subscriptions ${where ? `WHERE ${where}` : ''} ORDER BY ${orderBy} LIMIT ${pageLimit + 1}`;
+const rows = await pool.query(sql, param ? [param] : []);
+
+// 4. Build page
+const page = buildPage(rows, pageLimit, (r) => r.createdAt, (r) => r.id);
+
+// 5. Apply field selection
+const items = selectFieldsAll(page.items, fieldSet);
+
+return ok(items, requestId, { cursor: page.nextCursor, hasMore: page.hasMore });
+```
+
+---
+
+## Response Size Monitoring
+
+The `compressionMetrics` object tracks total bytes before and after compression per process lifetime. The `avgCompressionRatio` metric in Prometheus gives a fleet-wide view of compression effectiveness.
+
+Alert threshold recommendation: if `avgCompressionRatio > 0.9` (less than 10% savings), check that:
+1. `Content-Type` headers are set correctly on responses
+2. The minimum size threshold isn't set too high
+3. Clients are sending `Accept-Encoding: br, gzip`
diff --git a/docs/cache.md b/docs/cache.md
new file mode 100644
index 00000000..83002dee
--- /dev/null
+++ b/docs/cache.md
@@ -0,0 +1,168 @@
+# Cache Service
+
+`backend/services/shared/cache.ts`
+
+## Overview
+
+`CacheService` is a generic, typed Redis cache layer built on top of the existing `RedisCacheService` primitive. It adds event-driven invalidation, typed deserialization, cache warming with concurrency control, and a cache-bypass mode for forcing fresh reads.
+
+## Architecture
+
+```
+Request
+  │
+  ▼
+CacheService.getOrLoad(key, loader)
+  │
+  ├─ Redis hit  → deserialize → return
+  │
+  ├─ Redis miss → single-flight loader → serialize → Redis SET → return
+  │
+  └─ Redis down → degraded mode → loader() directly (no throw)
+```
+
+## Basic Usage
+
+```typescript
+import { CacheService } from '../services/shared/cache';
+
+// Inject via IoC
+const cache = container.resolve<ICacheService>('ICacheService');
+
+// Cache-aside with type safety
+const plan = await cache.getOrLoad<PlanMetadata>(
+  `plan:${planId}`,
+  () => repository.findById(planId),
+  { ttlSeconds: 3600 },
+);
+
+// Force-refresh (bypass cache, rehydrate)
+const fresh = await cache.getOrLoad<PlanMetadata>(
+  `plan:${planId}`,
+  () => repository.findById(planId),
+  { bypassCache: true },
+);
+
+// Explicit write
+await cache.set('config:feature-flags', flags, 600);
+
+// Invalidate
+await cache.invalidate('plan:abc', 'plan:xyz');
+await cache.invalidatePattern('plan:'); // removes all plan:* keys
+```
+
+## TTL Strategy
+
+| Data type | Suggested TTL |
+|---|---|
+| Plan metadata | 3600 s (1 hour) |
+| Subscription record | 300 s (5 min) |
+| User list | 120 s (2 min) |
+| Analytics aggregates | 900 s (15 min) |
+| Feature flags | 600 s (10 min) |
+
+Override per call via `{ ttlSeconds: N }`, or set a service-wide default in `CacheServiceConfig.defaultTtlSeconds`.
+
+## Event-Driven Invalidation
+
+Wire cache invalidation to domain events so stale entries are evicted automatically when state changes:
+
+```typescript
+import { wireInvalidation } from '../services/shared/cache';
+import { eventBus } from '../services/shared/events';
+
+wireInvalidation(cache, eventBus, [
+  {
+    eventName: 'subscription.cancelled',
+    keysFromEvent: (e) => [
+      `sub:${e.payload.subscriptionId}`,
+      `user-subs:${e.payload.userId}`,
+    ],
+  },
+  {
+    eventName: 'subscription.upgraded',
+    keysFromEvent: (e) => [`plan:${e.payload.fromPlanId}`, `plan:${e.payload.toPlanId}`],
+  },
+  {
+    eventName: '*', // catch-all for broad invalidation
+    keysFromEvent: (e) => e.aggregateId ? [`agg:${e.aggregateId}`] : [],
+  },
+]);
+```
+
+## Cache Warming on Startup
+
+```typescript
+const plans = await repository.findAllActive();
+
+const { warmed, errors } = await cache.warm(
+  plans.map((p) => ({ key: `plan:${p.id}`, value: p, ttlSeconds: 3600 })),
+);
+logger.info('Cache warm complete', { warmed, errors });
+```
+
+Warming respects `CacheServiceConfig.warmConcurrency` (default 10) to avoid Redis overload on startup.
+
+## Graceful Degradation
+
+When Redis is unreachable, `CacheService` enters degraded mode:
+- `getOrLoad` calls `loader()` directly — the app continues serving from the database
+- `set` and `invalidate` are no-ops — no throws, no crashes
+- `isHealthy()` probe resets `degraded = false` once Redis recovers
+
+Check status:
+```typescript
+cache.isDegraded(); // true when Redis is down
+await cache.isHealthy(); // pings Redis, resets degraded flag on success
+```
+
+## Cache Hit Rate Metrics
+
+```typescript
+const metrics = cache.getMetrics();
+/*
+{
+  hits, misses, writes, invalidations, errors, degradations,
+  hitRatio,           // NaN when no reads yet
+  latencyMs: { p50, p95, p99 },
+  memoryUsageBytes
+}
+*/
+```
+
+Prometheus export:
+```typescript
+const text = cache.prometheusMetrics('subtrackr_subscription_cache');
+// Exposes: hits_total, misses_total, hit_ratio, writes_total, latency_ms, memory_usage_bytes ...
+```
+
+## NullCacheService
+
+Use `NullCacheService` in tests or when Redis is not available. All reads pass through to the loader; writes and invalidations are no-ops.
+
+```typescript
+import { NullCacheService } from '../services/shared/cache';
+const cache = new NullCacheService();
+```
+
+## IoC Container Token
+
+| Token | Resolved type |
+|---|---|
+| `ICacheService` | `CacheService` (or `NullCacheService` in no-Redis environments) |
+
+```typescript
+const cache = container.resolve<ICacheService>('ICacheService');
+```
+
+## Performance Benchmarks
+
+Expected baseline with Redis on localhost:
+
+| Operation | p50 | p99 |
+|---|---|---|
+| Cache hit (GET) | < 1 ms | < 5 ms |
+| Cache miss + load | loader latency + ~1 ms | loader p99 + ~5 ms |
+| Invalidation | < 2 ms | < 8 ms |
+| Warm (1000 keys, concurrency 10) | ~200 ms total | — |
+
diff --git a/docs/database-pooling.md b/docs/database-pooling.md
new file mode 100644
index 00000000..0406d7b2
--- /dev/null
+++ b/docs/database-pooling.md
@@ -0,0 +1,168 @@
+# Database Connection Pool Optimization
+
+`backend/services/shared/poolMonitor.ts`
+
+## Overview
+
+`MonitoredPool` wraps any `Pool` (primary or replica) and adds real-time monitoring, leak detection, exhaustion alerting, and tuning recommendations without changing the `Pool` interface contract.
+
+## Architecture
+
+```
+Request
+  │
+  ▼
+MonitoredPool.query() / connect()
+  │
+  ├─ Records latency (p99 rolling window)
+  ├─ Tracks checked-out clients with timestamps
+  ├─ Injects SET statement_timeout on connect
+  │
+  ▼
+Inner pg.Pool → PostgreSQL primary / replica
+  │
+Background poll (every 5s)
+  ├─ Exhaustion check (waitingCount ≥ threshold → alert)
+  └─ Leak sweep (checkedOut age > leakThresholdMs → force-release + alert)
+```
+
+## Pool Configuration
+
+Tuned via environment variables in `backend/config/database.ts`:
+
+| Variable | Default | Description |
+|---|---|---|
+| `DB_POOL_MAX` | `20` | Max primary connections |
+| `DB_REPLICA_POOL_SIZE` | `25` | Connections per replica (via PgBouncer) |
+| `DB_IDLE_TIMEOUT_MS` | `10000` | Idle connection recycling |
+| `DB_CONNECTION_TIMEOUT_MS` | `30000` | Acquisition timeout |
+| `DB_STATEMENT_TIMEOUT_MS` | `30000` | Per-query statement timeout |
+| `DB_REPLICATION_LAG_FAILOVER_MS` | `5000` | Route reads to primary above this lag |
+| `DB_LAG_POLL_INTERVAL_MS` | `5000` | Replication lag poll interval |
+
+## Wrapping a Pool
+
+```typescript
+import { wrapWithMonitor } from '../services/shared/poolMonitor';
+
+const pool = await getPool();
+const monitored = wrapWithMonitor(pool, {
+  name: 'primary',
+  maxConnections: 20,
+  exhaustionThreshold: 5,   // alert when 5+ requests waiting
+  leakThresholdMs: 30_000,  // flag connections held > 30s
+  queryTimeoutMs: 30_000,
+  onExhaustion: (stats) => alertingService.fire('db.pool.exhaustion', stats),
+  onLeak: (leak) => logger.error('Connection leak', leak),
+});
+```
+
+`MonitoredPool` implements the full `Pool` interface — it's a drop-in replacement.
+
+## Pool Monitoring Dashboard
+
+```
+GET /pool/stats
+```
+
+```json
+{
+  "stats": {
+    "name": "primary",
+    "total": 18,
+    "idle": 14,
+    "waiting": 0,
+    "checkedOut": 4,
+    "utilizationPct": 20,
+    "leakedConnections": 0,
+    "peakCheckedOut": 12,
+    "totalQueries": 84210,
+    "totalErrors": 3,
+    "avgQueryLatencyMs": 4,
+    "p99QueryLatencyMs": 38
+  },
+  "tuning": {
+    "currentMax": 20,
+    "recommendedMax": 20,
+    "reason": "Current pool size appears sufficient for observed load"
+  },
+  "history": [...]
+}
+```
+
+## Pool Exhaustion Alerts
+
+When `waitingCount ≥ exhaustionThreshold` (default 5), the monitor:
+1. Logs a structured warning via `logger.warn`
+2. Calls the `onExhaustion` callback for external alerting
+3. Records the event in pool history
+
+Recovery options:
+- Increase `DB_POOL_MAX` (check DB `max_connections` first)
+- Add a PgBouncer proxy tier (use `serverlessPool.ts` pattern)
+- Reduce query durations (check `p99QueryLatencyMs`)
+
+## Connection Leak Detection
+
+Every `pool.connect()` call records a checkout timestamp. The background sweep (every `leakThresholdMs / 2`) force-releases any client held longer than `leakThresholdMs`:
+
+1. Calls `client.release()` to return the connection
+2. Increments `leakedConnections` counter
+3. Calls `onLeak` callback
+4. Logs origin information for debugging
+
+To avoid leaks in application code, always use `try/finally`:
+
+```typescript
+const client = await pool.connect();
+try {
+  await client.query('BEGIN');
+  // ...
+  await client.query('COMMIT');
+} catch (err) {
+  await client.query('ROLLBACK');
+  throw err;
+} finally {
+  client.release(); // always reached
+}
+```
+
+## Query Timeout Configuration
+
+`MonitoredPool` runs `SET statement_timeout = <ms>` on every new client checkout. This ensures long-running queries are killed at the database level before they exhaust pool connections.
+
+Default: 30 000 ms. Override via `queryTimeoutMs` in config or `DB_STATEMENT_TIMEOUT_MS` env var.
+
+## Prometheus Metrics
+
+```
+GET /metrics/pool
+```
+
+| Metric | Type | Description |
+|---|---|---|
+| `subtrackr_db_pool_total_connections` | gauge | Active connections |
+| `subtrackr_db_pool_idle_connections` | gauge | Idle connections |
+| `subtrackr_db_pool_waiting_connections` | gauge | Requests waiting |
+| `subtrackr_db_pool_checked_out_connections` | gauge | Currently in use |
+| `subtrackr_db_pool_utilization_pct` | gauge | % of max in use |
+| `subtrackr_db_pool_leaked_connections_total` | counter | Leaked connections |
+| `subtrackr_db_pool_queries_total` | counter | Total queries |
+| `subtrackr_db_pool_errors_total` | counter | Query errors |
+| `subtrackr_db_pool_avg_latency_ms` | gauge | Average query latency |
+| `subtrackr_db_pool_p99_latency_ms` | gauge | P99 query latency |
+
+All metrics include a `pool="primary"` label for multi-pool setups.
+
+## Pool Tuning Recommendations
+
+```typescript
+const rec = monitoredPool.getTuningRecommendation();
+// { currentMax: 20, recommendedMax: 25, reason: "Peak concurrent connections was 21..." }
+```
+
+The recommendation adds 20% headroom above the observed peak checked-out count. Check it periodically after traffic changes.
+
+## Elasticsearch Pool
+
+For Elasticsearch, the same `MonitoredPool` pattern can wrap an ES client proxy. See `backend/elasticsearch/config.ts` for field mappings. The ES index uses `max_results: 100` — paginate with the cursor-based pagination utilities when returning large result sets.
diff --git a/docs/events.md b/docs/events.md
new file mode 100644
index 00000000..66bc7ff5
--- /dev/null
+++ b/docs/events.md
@@ -0,0 +1,157 @@
+# Typed Event Bus
+
+`backend/services/shared/events.ts`
+
+## Overview
+
+SubTrackr uses an in-process typed event bus for decoupled, domain-driven communication between services. Events carry full TypeScript types, are validated at runtime, and are stored in an append-only event store for sourcing and replay.
+
+## Domain Event Definitions
+
+| Domain | Event name | Key payload fields |
+|---|---|---|
+| subscription | `subscription.created` | `subscriptionId`, `userId`, `planId`, `billingCycle`, `nextBillingDate` |
+| subscription | `subscription.cancelled` | `subscriptionId`, `reason`, `cancelledAt`, `effectiveAt` |
+| subscription | `subscription.renewed` | `subscriptionId`, `amount`, `currency`, `nextBillingDate` |
+| subscription | `subscription.upgraded` | `fromPlanId`, `toPlanId`, `proratedCredit` |
+| subscription | `subscription.paused` | `subscriptionId`, `pausedAt`, `resumeAt` |
+| subscription | `subscription.resumed` | `subscriptionId`, `resumedAt` |
+| subscription | `subscription.payment_failed` | `attemptNumber`, `nextRetryAt`, `reason` |
+| billing | `billing.invoice_generated` | `invoiceId`, `amount`, `currency`, `dueDate` |
+| billing | `billing.payment_captured` | `paymentId`, `amount`, `currency`, `gateway` |
+| billing | `billing.usage_threshold_reached` | `metricType`, `usage`, `limit`, `level` |
+| billing | `billing.chargeback_raised` | `chargebackId`, `amount`, `reason` |
+| analytics | `analytics.churn_risk_updated` | `riskScore`, `previousScore`, `factors` |
+| analytics | `analytics.cohort_aggregated` | `cohortId`, `period`, `retentionRate` |
+| analytics | `analytics.mrr_changed` | `previousMrr`, `currentMrr`, `delta` |
+| auth | `auth.api_key_rotated` | `keyId`, `merchantId`, `expiresAt` |
+| auth | `auth.sso_session_created` | `sessionId`, `provider`, `expiresAt` |
+| contract | `contract.invoked` | `contractId`, `method`, `caller`, `ledger` |
+| contract | `contract.upgraded` | `proxyId`, `fromImplementation`, `toImplementation` |
+
+## Publishing Events
+
+```typescript
+import { buildEvent, eventBus } from '../services/shared/events';
+
+const event = buildEvent('subscription', 'created', {
+  subscriptionId: 'sub_abc',
+  userId: 'user_123',
+  planId: 'plan_pro',
+  status: 'active',
+  billingCycle: 'monthly',
+  nextBillingDate: Date.now() + 30 * 86400_000,
+}, { aggregateId: 'sub_abc', correlationId: req.correlationId });
+
+await eventBus.publish(event);
+```
+
+`buildEvent` throws `EventValidationError` if the payload fails schema validation.
+
+## Subscribing
+
+```typescript
+// Specific event
+eventBus.subscribe('subscription.cancelled', async (event) => {
+  await notifyUser(event.payload.userId);
+});
+
+// Wildcard — receives everything
+eventBus.subscribe('*', (event) => {
+  logger.info('event received', { name: event.name });
+});
+
+// With predicate filter
+eventBus.subscribe('billing.usage_threshold_reached', handler, {
+  filter: (e) => e.payload.level === 'hard',
+});
+
+// Clean up
+const sub = eventBus.subscribe('subscription.created', handler);
+sub.unsubscribe();
+```
+
+## Schema Validation
+
+Schemas are defined in `PAYLOAD_SCHEMAS` inside `events.ts`. Call `validateEventPayload` directly for manual checks:
+
+```typescript
+const result = validateEventPayload('subscription.created', payload);
+if (!result.valid) console.error(result.errors);
+```
+
+## Event Sourcing
+
+```typescript
+import { eventStore, buildEvent } from '../services/shared/events';
+
+// Append
+eventStore.append(buildEvent('subscription', 'created', { ... }, { aggregateId: 'sub_abc' }));
+
+// Replay all events for an aggregate in sequence order
+eventStore.replay('sub_abc', (event) => applyToState(event));
+
+// Reconstruct current state from event history
+const state = eventStore.reconstruct('sub_abc');
+
+// Snapshot
+const snap = eventStore.snapshot('sub_abc');
+// { aggregateId, state, lastSequence, lastEventType, updatedAt }
+
+// Archive events older than 30 days
+eventStore.archiveBefore(Date.now() - 30 * 86400_000);
+```
+
+## Event Replay for Debugging
+
+```typescript
+// Query events by domain, type, or time window
+const events = eventStore.query({
+  aggregateId: 'sub_abc',
+  domain: 'subscription',
+  from: Date.now() - 7 * 86400_000,
+  limit: 100,
+});
+```
+
+## Event-Driven Testing
+
+```typescript
+import { SpyEventBus, EventCollector } from '../services/shared/events';
+
+// SpyEventBus — wraps real bus, records all publishes
+const spy = new SpyEventBus();
+await myService.cancel(subscriptionId, spy);
+spy.assertPublished('subscription.cancelled');
+
+// EventCollector — attach to any bus
+const collector = new EventCollector(eventBus, 'subscription.renewed');
+await runRenewal();
+collector.assertCount(1);
+collector.assertLastEventName('subscription.renewed');
+collector.dispose();
+```
+
+## Performance Monitoring
+
+```typescript
+const metrics = eventBus.getMetrics();
+// { published, handled, errors, avgHandlerLatencyMs, countByName, throughputPerSecond }
+
+// Prometheus export
+import { eventBusPrometheusMetrics } from '../services/shared/events';
+const prometheusText = eventBusPrometheusMetrics(eventBus);
+```
+
+## IoC Container Tokens
+
+| Token | Type |
+|---|---|
+| `IEventBus` | `EventBus` singleton |
+| `IEventStore` | `InMemoryEventStore` singleton |
+
+```typescript
+const bus = container.resolve<IEventBus>('IEventBus');
+const store = container.resolve<EventSourcedStore>('IEventStore');
+```
+