diff --git a/backend/.env.example b/backend/.env.example index c1339c5..bed4f2b 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -22,10 +22,18 @@ DATABASE_URL=./data/spanglish.db # Note: running more than one instance requires DB_TYPE=postgres. SQLite is a # single local file and cannot be shared safely across instances. -# Redis connection URL. When set, the cache, rate limiter, pub/sub (real-time -# payment events), distributed locks, and the email hourly cap are shared across -# all instances. When unset, each instance uses in-memory equivalents. +# Redis connection URL. When set, the cache, rate limiter, login lockout, +# pub/sub (real-time payment events), distributed locks, and the email hourly +# cap are shared across all instances. When unset, each instance uses in-memory +# equivalents. +# +# In production always set a password (requirepass on the server) and put it in +# the URL. Use the rediss:// scheme for TLS (handled natively by the client), +# and an optional /N path to select a DB index when sharing a Redis instance +# with other applications. # REDIS_URL=redis://localhost:6379 +# REDIS_URL=redis://:your-redis-password@redis.internal:6379 +# REDIS_URL=rediss://:your-redis-password@redis.example.com:6380/1 # Optional S3-compatible object storage for media uploads (e.g. Garage, MinIO, # AWS S3). When S3_ENDPOINT and S3_BUCKET are set, uploads go to the bucket and diff --git a/backend/package.json b/backend/package.json index a045340..c0c948d 100644 --- a/backend/package.json +++ b/backend/package.json @@ -5,6 +5,7 @@ "scripts": { "dev": "tsx watch src/index.ts", "build": "tsc", + "test": "vitest run", "start": "NODE_ENV=production node dist/index.js", "db:generate": "drizzle-kit generate", "db:migrate": "tsx src/db/migrate.ts", @@ -42,7 +43,9 @@ "@types/pg": "^8.11.6", "@types/qrcode": "^1.5.6", "drizzle-kit": "^0.22.8", + "ioredis-mock": "^8.13.1", "tsx": "^4.15.7", - "typescript": "^5.5.2" + "typescript": "^5.5.2", + "vitest": "^4.1.10" } } diff --git a/backend/src/index.ts b/backend/src/index.ts index d7773c9..2c8ed7c 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -24,10 +24,11 @@ import legalPagesRoutes from './routes/legal-pages.js'; import legalSettingsRoutes from './routes/legal-settings.js'; import faqRoutes from './routes/faq.js'; import emailService from './lib/email.js'; -import { initEmailQueue } from './lib/emailQueue.js'; -import { startBookingCleanup } from './lib/bookingCleanup.js'; -import { startHoldSweep } from './lib/holdSweep.js'; -import { startEventEndSweep } from './lib/eventEndSweep.js'; +import { initEmailQueue, stopQueue } from './lib/emailQueue.js'; +import { startBookingCleanup, stopBookingCleanup } from './lib/bookingCleanup.js'; +import { startHoldSweep, stopHoldSweep } from './lib/holdSweep.js'; +import { startEventEndSweep, stopEventEndSweep } from './lib/eventEndSweep.js'; +import { closeRedis } from './lib/redis.js'; import { getLock } from './lib/stores/lock.js'; import { describeBackends, describeRedis, logSelectedBackends } from './lib/backends.js'; @@ -1943,8 +1944,11 @@ startEventEndSweep(); // Initialize email templates on startup. // Guarded by a distributed lock so that, when running multiple replicas, only // one instance seeds/updates templates per boot instead of all of them racing. +// onUnavailable 'run': at boot the Redis connection may not be ready yet, and +// seeding is upsert-idempotent, so racing replicas are safe — never skipping +// beats never seeding on a first boot during a Redis blip. getLock() - .withLock('seed-templates', 30_000, () => emailService.seedDefaultTemplates()) + .withLock('seed-templates', 30_000, () => emailService.seedDefaultTemplates(), { onUnavailable: 'run' }) .then((result) => { if (result === null) { console.log('[Email] Template seeding skipped (another instance holds the lock)'); @@ -1961,7 +1965,43 @@ console.log(`📋 OpenAPI spec at http://localhost:${port}/openapi.json`); // Log which backend (memory/redis, local/s3) each subsystem selected. logSelectedBackends(); -serve({ +const server = serve({ fetch: app.fetch, port, }); + +// Graceful shutdown: stop the periodic jobs, stop accepting connections, then +// close Redis and exit. Open SSE payment streams hold sockets forever, so +// server.close() alone never completes — force-close remaining connections +// after a short grace period, with a hard exit as the final backstop. +let shuttingDown = false; +function shutdown(signal: string): void { + if (shuttingDown) return; + shuttingDown = true; + console.log(`[shutdown] ${signal} received, draining...`); + + stopBookingCleanup(); + stopHoldSweep(); + stopEventEndSweep(); + stopQueue(); + + server.close(() => { + console.log('[shutdown] server closed, closing redis'); + closeRedis().finally(() => process.exit(0)); + }); + + const forceClose = setTimeout(() => { + console.warn('[shutdown] force-closing remaining connections (SSE streams)'); + (server as any).closeAllConnections?.(); + }, 5_000); + forceClose.unref(); + + const forceExit = setTimeout(() => { + console.warn('[shutdown] drain timed out, forcing exit'); + process.exit(1); + }, 10_000); + forceExit.unref(); +} + +process.on('SIGTERM', () => shutdown('SIGTERM')); +process.on('SIGINT', () => shutdown('SIGINT')); diff --git a/backend/src/lib/backends.ts b/backend/src/lib/backends.ts index 63196a9..c2eec12 100644 --- a/backend/src/lib/backends.ts +++ b/backend/src/lib/backends.ts @@ -1,11 +1,12 @@ // Reports which backend each scalable subsystem is using, for the health // endpoint and startup logging. -import { isRedisEnabled, isRedisHealthy } from './redis.js'; +import { isRedisEnabled, isRedisHealthy, getRedisHealthDetail } from './redis.js'; import { getRateLimiter } from './stores/rateLimiter.js'; import { getPubSub } from './stores/pubsub.js'; import { getCache } from './stores/cache.js'; import { getLock } from './stores/lock.js'; +import { getLoginLockout } from './stores/loginLockout.js'; import { getStorage } from './storage.js'; export function describeBackends() { @@ -14,12 +15,13 @@ export function describeBackends() { rateLimiter: getRateLimiter().backend, pubsub: getPubSub().backend, lock: getLock().backend, + loginLockout: getLoginLockout().backend, storage: getStorage().backend, }; } export function describeRedis() { - return { enabled: isRedisEnabled(), healthy: isRedisHealthy() }; + return { enabled: isRedisEnabled(), healthy: isRedisHealthy(), ...getRedisHealthDetail() }; } /** Log one line per subsystem at startup so the active backend is obvious. */ @@ -32,5 +34,6 @@ export function logSelectedBackends(): void { console.log(` rate limiter: ${b.rateLimiter}`); console.log(` pub/sub: ${b.pubsub}`); console.log(` lock: ${b.lock}`); + console.log(` login lockout:${b.loginLockout}`); console.log(` storage: ${b.storage}`); } diff --git a/backend/src/lib/redis.ts b/backend/src/lib/redis.ts index 4c53bca..ccdca88 100644 --- a/backend/src/lib/redis.ts +++ b/backend/src/lib/redis.ts @@ -4,7 +4,15 @@ // before with in-memory backends. When set, this module owns a single shared // command connection plus a dedicated subscriber connection (a connection in // subscribe mode cannot run normal commands), with auto-reconnect, capped -// backoff, and a health flag that callers and the health endpoint can read. +// backoff, an active PING probe, and a health flag that callers and the health +// endpoint can read. +// +// TLS: use a rediss:// URL — ioredis enables TLS from the scheme. A /N path +// selects a DB index (e.g. redis://host:6379/1) when sharing an instance. +// We deliberately do not use ioredis's keyPrefix option: all keys are already +// namespaced per subsystem (cache:, rl:, lock:, lockout:), and keyPrefix has a +// pub/sub asymmetry (SUBSCRIBE channels get prefixed, PUBLISH channels do not) +// that would silently break the payment SSE channel. import Redis from 'ioredis'; @@ -12,6 +20,14 @@ let client: Redis | null = null; let subscriber: Redis | null = null; let healthy = false; let initialized = false; +let pingTimer: ReturnType | null = null; +let lastPingOkAt: string | null = null; +let lastPingMs: number | null = null; + +// Debounce repeated error logs during a sustained outage: transitions are +// always logged, repeated per-retry errors at most once per LOG_EVERY_MS. +const ERROR_LOG_EVERY_MS = 30_000; +let lastErrorLogAt = 0; /** Whether Redis is configured via REDIS_URL. */ export function isRedisEnabled(): boolean { @@ -23,14 +39,41 @@ export function isRedisHealthy(): boolean { return isRedisEnabled() && healthy; } +/** Detail for the health endpoint: when the last successful PING happened. */ +export function getRedisHealthDetail(): { lastPingOkAt: string | null; lastPingMs: number | null } { + return { lastPingOkAt, lastPingMs }; +} + +function setHealthy(next: boolean, context: string): void { + if (next !== healthy) { + if (next) { + console.log(`[redis] healthy (${context})`); + } else { + console.warn(`[redis] unhealthy (${context})`); + } + } + healthy = next; +} + +function logErrorDebounced(label: string, err: unknown): void { + const now = Date.now(); + if (now - lastErrorLogAt >= ERROR_LOG_EVERY_MS) { + lastErrorLogAt = now; + console.error(`[redis] (${label}) error:`, (err as any)?.message || err); + } +} + function buildClient(label: string): Redis { const url = process.env.REDIS_URL as string; const instance = new Redis(url, { // Keep the process responsive: fail fast on a per-command basis and let the - // callers degrade to their in-memory fallback rather than hanging. + // callers degrade to their in-memory fallback rather than hanging. Do not + // queue commands while disconnected — with fail-open callers everywhere a + // growing offline queue would only add latency and memory pressure. maxRetriesPerRequest: 1, enableOfflineQueue: false, lazyConnect: false, + connectTimeout: 5000, retryStrategy(times) { // Capped exponential backoff for reconnects: 200ms, 400ms ... max 5s. const delay = Math.min(times * 200, 5000); @@ -42,30 +85,53 @@ function buildClient(label: string): Redis { console.log(`[redis] (${label}) connecting`); }); instance.on('ready', () => { - healthy = true; - console.log(`[redis] (${label}) ready`); + setHealthy(true, `${label} ready`); }); instance.on('error', (err) => { - healthy = false; - console.error(`[redis] (${label}) error:`, err?.message || err); + setHealthy(false, `${label} error`); + logErrorDebounced(label, err); }); instance.on('reconnecting', () => { - healthy = false; - console.warn(`[redis] (${label}) reconnecting`); + setHealthy(false, `${label} reconnecting`); }); instance.on('end', () => { - healthy = false; - console.warn(`[redis] (${label}) connection closed`); + setHealthy(false, `${label} connection closed`); }); return instance; } +// Actively probe the command connection. The event-driven flag alone misses a +// silently hung connection; a periodic PING with a hard timeout catches it. +async function pingOnce(): Promise { + if (!client) return; + const started = Date.now(); + try { + await Promise.race([ + client.ping(), + new Promise((_, reject) => { + const t = setTimeout(() => reject(new Error('ping timeout')), 2000); + (t as any).unref?.(); + }), + ]); + lastPingMs = Date.now() - started; + lastPingOkAt = new Date().toISOString(); + setHealthy(true, 'ping ok'); + } catch (err) { + setHealthy(false, 'ping failed'); + logErrorDebounced('ping', err); + } +} + function ensureInit(): void { if (initialized || !isRedisEnabled()) return; initialized = true; client = buildClient('commands'); subscriber = buildClient('subscriber'); + pingTimer = setInterval(() => { + void pingOnce(); + }, 10_000); + (pingTimer as any).unref?.(); } /** Shared command connection, or null when Redis is not configured. */ @@ -82,6 +148,10 @@ export function getSubscriber(): Redis | null { /** Close connections (used for graceful shutdown). */ export async function closeRedis(): Promise { + if (pingTimer) { + clearInterval(pingTimer); + pingTimer = null; + } const tasks: Promise[] = []; if (client) tasks.push(client.quit().catch(() => undefined)); if (subscriber) tasks.push(subscriber.quit().catch(() => undefined)); diff --git a/backend/src/lib/stores/lock.test.ts b/backend/src/lib/stores/lock.test.ts new file mode 100644 index 0000000..1c5b64e --- /dev/null +++ b/backend/src/lib/stores/lock.test.ts @@ -0,0 +1,98 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import RedisMock from 'ioredis-mock'; + +const mocks = vi.hoisted(() => ({ redis: null as any })); + +vi.mock('../redis.js', () => ({ + isRedisEnabled: () => true, + getRedis: () => mocks.redis, +})); + +import { MemoryLock, RedisLock, LockUnavailableError } from './lock.js'; + +describe('MemoryLock', () => { + it('grants the lock, blocks contenders, and frees on release', async () => { + const lock = new MemoryLock(); + const token = await lock.acquire('a', 60_000); + expect(token).toBeTruthy(); + expect(await lock.acquire('a', 60_000)).toBeNull(); + await lock.release('a', token!); + expect(await lock.acquire('a', 60_000)).toBeTruthy(); + }); + + it('ignores release with the wrong token', async () => { + const lock = new MemoryLock(); + const token = await lock.acquire('a', 60_000); + await lock.release('a', 'not-the-token'); + expect(await lock.acquire('a', 60_000)).toBeNull(); + await lock.release('a', token!); + }); + + it('withLock runs fn while held and returns null under contention', async () => { + const lock = new MemoryLock(); + const held = await lock.acquire('a', 60_000); + expect(await lock.withLock('a', 60_000, async () => 'ran')).toBeNull(); + await lock.release('a', held!); + expect(await lock.withLock('a', 60_000, async () => 'ran')).toBe('ran'); + // Released after withLock completes. + expect(await lock.acquire('a', 60_000)).toBeTruthy(); + }); +}); + +describe('RedisLock', () => { + beforeEach(async () => { + mocks.redis = new RedisMock(); + // ioredis-mock shares data between instances by connection string. + await mocks.redis.flushall(); + }); + + it('grants the lock and blocks contenders', async () => { + const lock = new RedisLock(); + const token = await lock.acquire('a', 60_000); + expect(token).toBeTruthy(); + expect(await lock.acquire('a', 60_000)).toBeNull(); + }); + + it('release is compare-and-delete: wrong token does not free the lock', async () => { + const lock = new RedisLock(); + const token = await lock.acquire('a', 60_000); + await lock.release('a', 'not-the-token'); + expect(await lock.acquire('a', 60_000)).toBeNull(); + await lock.release('a', token!); + expect(await lock.acquire('a', 60_000)).toBeTruthy(); + }); + + it('throws LockUnavailableError when the client is not initialized', async () => { + mocks.redis = null; + const lock = new RedisLock(); + await expect(lock.acquire('a', 60_000)).rejects.toBeInstanceOf(LockUnavailableError); + }); + + it('throws LockUnavailableError when the backend errors', async () => { + mocks.redis = { set: () => Promise.reject(new Error('connection refused')) }; + const lock = new RedisLock(); + await expect(lock.acquire('a', 60_000)).rejects.toBeInstanceOf(LockUnavailableError); + }); + + it('withLock skips (returns null, fn not called) when unavailable by default', async () => { + mocks.redis = null; + const lock = new RedisLock(); + const fn = vi.fn(async () => 'ran'); + expect(await lock.withLock('a', 60_000, fn)).toBeNull(); + expect(fn).not.toHaveBeenCalled(); + }); + + it('withLock runs fn without the lock when onUnavailable is "run"', async () => { + mocks.redis = null; + const lock = new RedisLock(); + const fn = vi.fn(async () => 'ran'); + expect(await lock.withLock('a', 60_000, fn, { onUnavailable: 'run' })).toBe('ran'); + expect(fn).toHaveBeenCalledOnce(); + }); + + it('withLock returns fn result and releases the lock afterwards', async () => { + const lock = new RedisLock(); + expect(await lock.withLock('a', 60_000, async () => 42)).toBe(42); + expect(await lock.acquire('a', 60_000)).toBeTruthy(); + }); +}); diff --git a/backend/src/lib/stores/lock.ts b/backend/src/lib/stores/lock.ts index 10950ae..e673031 100644 --- a/backend/src/lib/stores/lock.ts +++ b/backend/src/lib/stores/lock.ts @@ -5,22 +5,45 @@ // // Use acquire/release for long-lived ownership (e.g. a background poller) and // withLock for a one-shot critical section. Selection is based on REDIS_URL. +// +// There is deliberately no auto-renewal/watchdog: every guarded section (sweep +// jobs, template seeding) is a handful of status-conditional bulk UPDATEs that +// finish far below the lock TTL, and a rare TTL overrun only risks one +// idempotent overlapping run. Keep TTLs generous instead of adding renewal. import { randomUUID } from 'crypto'; import { getRedis, isRedisEnabled } from '../redis.js'; +// Thrown when Redis is configured but the lock backend cannot be reached. +// This is distinct from contention (acquire resolves null): the caller must +// decide whether its critical section is safe to run without mutual exclusion. +export class LockUnavailableError extends Error { + constructor(message: string) { + super(message); + this.name = 'LockUnavailableError'; + } +} + +export interface WithLockOptions { + // What withLock should do when the lock backend is unavailable (not mere + // contention): 'skip' (default) returns null as if the lock were held + // elsewhere; 'run' executes fn without mutual exclusion. + onUnavailable?: 'skip' | 'run'; +} + export interface Lock { readonly backend: 'memory' | 'redis'; // Returns a token when the lock was acquired, or null when already held. + // Throws LockUnavailableError when the backend is configured but erroring. acquire(key: string, ttlMs: number): Promise; release(key: string, token: string): Promise; // Runs fn while holding the lock; returns fn's result, or null if not acquired. - withLock(key: string, ttlMs: number, fn: () => Promise): Promise; + withLock(key: string, ttlMs: number, fn: () => Promise, opts?: WithLockOptions): Promise; } // ==================== Memory implementation ==================== -class MemoryLock implements Lock { +export class MemoryLock implements Lock { readonly backend = 'memory' as const; private held = new Map(); @@ -58,23 +81,23 @@ class MemoryLock implements Lock { const RELEASE_SCRIPT = 'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end'; -class RedisLock implements Lock { +export class RedisLock implements Lock { readonly backend = 'redis' as const; async acquire(key: string, ttlMs: number): Promise { const redis = getRedis(); if (!redis) { - // Redis configured but unavailable: do not block critical sections. - return randomUUID(); + throw new LockUnavailableError('redis client not initialized'); } const token = randomUUID(); try { const result = await redis.set(`lock:${key}`, token, 'PX', ttlMs, 'NX'); return result === 'OK' ? token : null; } catch (err: any) { - console.error('[lock] redis acquire error, proceeding without lock:', err?.message || err); - // Fail open so a Redis outage does not deadlock startup or jobs. - return randomUUID(); + // Do NOT fabricate a token here: during an outage every replica would + // "acquire" every lock and run the guarded sections concurrently. Let the + // caller choose between skipping the run and running unlocked. + throw new LockUnavailableError(err?.message || String(err)); } } @@ -88,8 +111,19 @@ class RedisLock implements Lock { } } - async withLock(key: string, ttlMs: number, fn: () => Promise): Promise { - const token = await this.acquire(key, ttlMs); + async withLock(key: string, ttlMs: number, fn: () => Promise, opts?: WithLockOptions): Promise { + let token: string | null; + try { + token = await this.acquire(key, ttlMs); + } catch (err) { + if (!(err instanceof LockUnavailableError)) throw err; + if (opts?.onUnavailable === 'run') { + console.warn(`[lock] backend unavailable, running "${key}" WITHOUT mutual exclusion:`, err.message); + return fn(); + } + console.warn(`[lock] backend unavailable, skipping "${key}" this run:`, err.message); + return null; + } if (!token) return null; try { return await fn(); diff --git a/backend/src/lib/stores/loginLockout.test.ts b/backend/src/lib/stores/loginLockout.test.ts new file mode 100644 index 0000000..c3a02ef --- /dev/null +++ b/backend/src/lib/stores/loginLockout.test.ts @@ -0,0 +1,125 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import RedisMock from 'ioredis-mock'; + +const mocks = vi.hoisted(() => ({ redis: null as any })); + +vi.mock('../redis.js', () => ({ + isRedisEnabled: () => true, + getRedis: () => mocks.redis, +})); + +import { + MemoryLoginLockout, + RedisLoginLockout, + MAX_LOGIN_ATTEMPTS, +} from './loginLockout.js'; + +describe('MemoryLoginLockout', () => { + it('locks after MAX_LOGIN_ATTEMPTS failures with retryAfter', async () => { + const lockout = new MemoryLoginLockout(); + for (let i = 0; i < MAX_LOGIN_ATTEMPTS - 1; i++) { + await lockout.recordFailure('user@example.com'); + expect((await lockout.isLocked('user@example.com')).locked).toBe(false); + } + await lockout.recordFailure('user@example.com'); + const status = await lockout.isLocked('user@example.com'); + expect(status.locked).toBe(true); + expect(status.retryAfter).toBeGreaterThan(0); + }); + + it('clear removes the lockout', async () => { + const lockout = new MemoryLoginLockout(); + for (let i = 0; i < MAX_LOGIN_ATTEMPTS; i++) { + await lockout.recordFailure('user@example.com'); + } + await lockout.clear('user@example.com'); + expect((await lockout.isLocked('user@example.com')).locked).toBe(false); + }); + + it('treats emails case-insensitively', async () => { + const lockout = new MemoryLoginLockout(); + for (let i = 0; i < MAX_LOGIN_ATTEMPTS; i++) { + await lockout.recordFailure('User@Example.com'); + } + expect((await lockout.isLocked('user@example.com')).locked).toBe(true); + }); + + it('expires after the lockout window', async () => { + vi.useFakeTimers(); + try { + const lockout = new MemoryLoginLockout(); + for (let i = 0; i < MAX_LOGIN_ATTEMPTS; i++) { + await lockout.recordFailure('user@example.com'); + } + expect((await lockout.isLocked('user@example.com')).locked).toBe(true); + vi.advanceTimersByTime(16 * 60 * 1000); + expect((await lockout.isLocked('user@example.com')).locked).toBe(false); + } finally { + vi.useRealTimers(); + } + }); +}); + +describe('RedisLoginLockout', () => { + beforeEach(async () => { + mocks.redis = new RedisMock(); + // ioredis-mock shares data between instances by connection string. + await mocks.redis.flushall(); + }); + + it('locks after MAX_LOGIN_ATTEMPTS failures shared via redis', async () => { + const lockout = new RedisLoginLockout(); + for (let i = 0; i < MAX_LOGIN_ATTEMPTS - 1; i++) { + await lockout.recordFailure('user@example.com'); + expect((await lockout.isLocked('user@example.com')).locked).toBe(false); + } + await lockout.recordFailure('user@example.com'); + const status = await lockout.isLocked('user@example.com'); + expect(status.locked).toBe(true); + expect(status.retryAfter).toBeGreaterThan(0); + }); + + it('sets the lockout window TTL on the first failure', async () => { + const lockout = new RedisLoginLockout(); + await lockout.recordFailure('user@example.com'); + const ttl = await mocks.redis.pttl('lockout:user@example.com'); + expect(ttl).toBeGreaterThan(0); + }); + + it('clear removes the lockout', async () => { + const lockout = new RedisLoginLockout(); + for (let i = 0; i < MAX_LOGIN_ATTEMPTS; i++) { + await lockout.recordFailure('user@example.com'); + } + await lockout.clear('user@example.com'); + expect((await lockout.isLocked('user@example.com')).locked).toBe(false); + }); + + it('treats emails case-insensitively', async () => { + const lockout = new RedisLoginLockout(); + for (let i = 0; i < MAX_LOGIN_ATTEMPTS; i++) { + await lockout.recordFailure('User@Example.com'); + } + expect((await lockout.isLocked('user@example.com')).locked).toBe(true); + }); + + it('fails open when the client is not initialized', async () => { + mocks.redis = null; + const lockout = new RedisLoginLockout(); + await expect(lockout.recordFailure('user@example.com')).resolves.toBeUndefined(); + expect((await lockout.isLocked('user@example.com')).locked).toBe(false); + }); + + it('fails open when the backend errors', async () => { + mocks.redis = { + get: () => Promise.reject(new Error('connection refused')), + pttl: () => Promise.reject(new Error('connection refused')), + del: () => Promise.reject(new Error('connection refused')), + lockoutRecordFailure: () => Promise.reject(new Error('connection refused')), + }; + const lockout = new RedisLoginLockout(); + await expect(lockout.recordFailure('user@example.com')).resolves.toBeUndefined(); + expect((await lockout.isLocked('user@example.com')).locked).toBe(false); + await expect(lockout.clear('user@example.com')).resolves.toBeUndefined(); + }); +}); diff --git a/backend/src/lib/stores/loginLockout.ts b/backend/src/lib/stores/loginLockout.ts new file mode 100644 index 0000000..c821078 --- /dev/null +++ b/backend/src/lib/stores/loginLockout.ts @@ -0,0 +1,164 @@ +// Per-email login lockout abstraction with two implementations: +// - memory: per-process Map (the original routes/auth.ts behavior) +// - redis: shared counter across all instances so the lockout cannot be +// bypassed by round-robining replicas +// +// Semantics: recordFailure starts a window on the first failure; once the +// failure count reaches the max the email is locked until the window expires; +// clear removes the counter on successful login. Selection is based on +// REDIS_URL. The redis implementation fails open (never locked, failures not +// recorded) — the per-IP auth rate limit remains as a backstop during a +// Redis outage. + +import type Redis from 'ioredis'; +import { getRedis, isRedisEnabled } from '../redis.js'; + +export const MAX_LOGIN_ATTEMPTS = 5; +export const LOCKOUT_DURATION_MS = 15 * 60 * 1000; // 15 minutes + +export interface LockoutStatus { + locked: boolean; + // Seconds until the lockout lifts; only set when locked. + retryAfter?: number; +} + +export interface LoginLockout { + readonly backend: 'memory' | 'redis'; + isLocked(email: string): Promise; + recordFailure(email: string): Promise; + clear(email: string): Promise; +} + +// Emails are compared case-insensitively so "User@x.com" and "user@x.com" +// share one counter. +function normalize(email: string): string { + return email.trim().toLowerCase(); +} + +// ==================== Memory implementation ==================== + +export class MemoryLoginLockout implements LoginLockout { + readonly backend = 'memory' as const; + private attempts = new Map(); + + constructor() { + // Periodically drop expired entries so the Map does not grow unbounded. + const cleanup = setInterval(() => { + const now = Date.now(); + for (const [key, entry] of this.attempts) { + if (now > entry.resetAt) this.attempts.delete(key); + } + }, 60_000); + (cleanup as any).unref?.(); + } + + async isLocked(email: string): Promise { + const entry = this.attempts.get(normalize(email)); + const now = Date.now(); + if (!entry || now > entry.resetAt) return { locked: false }; + if (entry.count >= MAX_LOGIN_ATTEMPTS) { + return { locked: true, retryAfter: Math.ceil((entry.resetAt - now) / 1000) }; + } + return { locked: false }; + } + + async recordFailure(email: string): Promise { + const key = normalize(email); + const now = Date.now(); + const entry = this.attempts.get(key); + if (!entry || now > entry.resetAt) { + this.attempts.set(key, { count: 1, resetAt: now + LOCKOUT_DURATION_MS }); + return; + } + entry.count++; + } + + async clear(email: string): Promise { + this.attempts.delete(normalize(email)); + } +} + +// ==================== Redis implementation ==================== + +// Same atomic INCR+PEXPIRE shape as the rate limiter's consume script: the +// window starts at the first failure and any TTL-less counter is repaired. +const RECORD_FAILURE_SCRIPT = ` +local count = redis.call('INCR', KEYS[1]) +if count == 1 then + redis.call('PEXPIRE', KEYS[1], ARGV[1]) +end +if redis.call('PTTL', KEYS[1]) < 0 then + redis.call('PEXPIRE', KEYS[1], ARGV[1]) +end +return count +`; + +type RedisWithLockout = Redis & { + lockoutRecordFailure(key: string, windowMs: number): Promise; +}; + +function withLockoutCommand(redis: Redis): RedisWithLockout { + if (typeof (redis as any).lockoutRecordFailure !== 'function') { + redis.defineCommand('lockoutRecordFailure', { numberOfKeys: 1, lua: RECORD_FAILURE_SCRIPT }); + } + return redis as RedisWithLockout; +} + +export class RedisLoginLockout implements LoginLockout { + readonly backend = 'redis' as const; + + private key(email: string): string { + return `lockout:${normalize(email)}`; + } + + async isLocked(email: string): Promise { + const redis = getRedis(); + if (!redis) return { locked: false }; + try { + const [count, ttl] = await Promise.all([ + redis.get(this.key(email)), + redis.pttl(this.key(email)), + ]); + if (count !== null && parseInt(count, 10) >= MAX_LOGIN_ATTEMPTS) { + const retryAfter = ttl > 0 ? Math.ceil(ttl / 1000) : Math.ceil(LOCKOUT_DURATION_MS / 1000); + return { locked: true, retryAfter }; + } + return { locked: false }; + } catch (err: any) { + // Fail open: the per-IP auth rate limit still applies. + console.error('[loginLockout] redis error, treating as unlocked:', err?.message || err); + return { locked: false }; + } + } + + async recordFailure(email: string): Promise { + const redis = getRedis(); + if (!redis) return; + try { + await withLockoutCommand(redis).lockoutRecordFailure(this.key(email), LOCKOUT_DURATION_MS); + } catch (err: any) { + console.error('[loginLockout] redis error recording failure:', err?.message || err); + } + } + + async clear(email: string): Promise { + const redis = getRedis(); + if (!redis) return; + try { + await redis.del(this.key(email)); + } catch (err: any) { + console.error('[loginLockout] redis error clearing failures:', err?.message || err); + } + } +} + +// ==================== Selection ==================== + +let instance: LoginLockout | null = null; + +export function getLoginLockout(): LoginLockout { + if (!instance) { + instance = isRedisEnabled() ? new RedisLoginLockout() : new MemoryLoginLockout(); + } + return instance; +} diff --git a/backend/src/lib/stores/rateLimiter.test.ts b/backend/src/lib/stores/rateLimiter.test.ts new file mode 100644 index 0000000..c53b612 --- /dev/null +++ b/backend/src/lib/stores/rateLimiter.test.ts @@ -0,0 +1,84 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import RedisMock from 'ioredis-mock'; + +const mocks = vi.hoisted(() => ({ redis: null as any })); + +vi.mock('../redis.js', () => ({ + isRedisEnabled: () => true, + getRedis: () => mocks.redis, +})); + +import { MemoryRateLimiter, RedisRateLimiter } from './rateLimiter.js'; + +describe('MemoryRateLimiter', () => { + it('allows up to max within the window, then blocks with retryAfter', async () => { + const limiter = new MemoryRateLimiter(); + for (let i = 0; i < 3; i++) { + expect((await limiter.consume('k', 3, 60_000)).allowed).toBe(true); + } + const blocked = await limiter.consume('k', 3, 60_000); + expect(blocked.allowed).toBe(false); + expect(blocked.retryAfter).toBeGreaterThan(0); + expect(blocked.retryAfter).toBeLessThanOrEqual(60); + }); + + it('resets after the window elapses', async () => { + vi.useFakeTimers(); + try { + const limiter = new MemoryRateLimiter(); + expect((await limiter.consume('k', 1, 1_000)).allowed).toBe(true); + expect((await limiter.consume('k', 1, 1_000)).allowed).toBe(false); + vi.advanceTimersByTime(1_500); + expect((await limiter.consume('k', 1, 1_000)).allowed).toBe(true); + } finally { + vi.useRealTimers(); + } + }); +}); + +describe('RedisRateLimiter', () => { + beforeEach(async () => { + mocks.redis = new RedisMock(); + // ioredis-mock shares data between instances by connection string. + await mocks.redis.flushall(); + }); + + it('sets the window TTL atomically on the first hit', async () => { + const limiter = new RedisRateLimiter(); + expect((await limiter.consume('k', 5, 60_000)).allowed).toBe(true); + const ttl = await mocks.redis.pttl('rl:k'); + expect(ttl).toBeGreaterThan(0); + expect(ttl).toBeLessThanOrEqual(60_000); + }); + + it('blocks over the limit with a sane retryAfter', async () => { + const limiter = new RedisRateLimiter(); + for (let i = 0; i < 2; i++) { + expect((await limiter.consume('k', 2, 60_000)).allowed).toBe(true); + } + const blocked = await limiter.consume('k', 2, 60_000); + expect(blocked.allowed).toBe(false); + expect(blocked.retryAfter).toBeGreaterThan(0); + expect(blocked.retryAfter).toBeLessThanOrEqual(60); + }); + + it('self-heals a counter stranded without a TTL', async () => { + await mocks.redis.set('rl:k', '3'); + expect(await mocks.redis.pttl('rl:k')).toBeLessThan(0); + const limiter = new RedisRateLimiter(); + await limiter.consume('k', 10, 60_000); + expect(await mocks.redis.pttl('rl:k')).toBeGreaterThan(0); + }); + + it('fails open when the backend errors', async () => { + mocks.redis = { rlConsume: () => Promise.reject(new Error('connection refused')) }; + const limiter = new RedisRateLimiter(); + expect((await limiter.consume('k', 1, 60_000)).allowed).toBe(true); + }); + + it('fails open when the client is not initialized', async () => { + mocks.redis = null; + const limiter = new RedisRateLimiter(); + expect((await limiter.consume('k', 1, 60_000)).allowed).toBe(true); + }); +}); diff --git a/backend/src/lib/stores/rateLimiter.ts b/backend/src/lib/stores/rateLimiter.ts index 6cf9f34..8e9ef21 100644 --- a/backend/src/lib/stores/rateLimiter.ts +++ b/backend/src/lib/stores/rateLimiter.ts @@ -1,10 +1,13 @@ // Rate limiter abstraction with two implementations: // - memory: per-process fixed window (the original behavior) -// - redis: shared fixed window across all instances (INCR + PEXPIRE) +// - redis: shared fixed window across all instances, implemented as a single +// Lua script so INCR and PEXPIRE are atomic (a crash between separate calls +// would otherwise strand a counter with no expiry) // // Selection happens once based on REDIS_URL. On any Redis error the limiter // fails open (allows the request) so a Redis blip never takes the API down. +import type Redis from 'ioredis'; import { getRedis, isRedisEnabled } from '../redis.js'; export interface RateLimitResult { @@ -24,7 +27,7 @@ interface Bucket { resetAt: number; } -class MemoryRateLimiter implements RateLimiter { +export class MemoryRateLimiter implements RateLimiter { readonly backend = 'memory' as const; private buckets = new Map(); @@ -58,22 +61,44 @@ class MemoryRateLimiter implements RateLimiter { // ==================== Redis implementation ==================== -class RedisRateLimiter implements RateLimiter { +// Atomically increments the window counter, sets the expiry on the first hit, +// and repairs any counter left without a TTL (self-heals keys stranded by the +// pre-Lua implementation or a lost PEXPIRE). Returns {count, ttlMs}. +export const CONSUME_SCRIPT = ` +local count = redis.call('INCR', KEYS[1]) +if count == 1 then + redis.call('PEXPIRE', KEYS[1], ARGV[1]) +end +local ttl = redis.call('PTTL', KEYS[1]) +if ttl < 0 then + redis.call('PEXPIRE', KEYS[1], ARGV[1]) + ttl = tonumber(ARGV[1]) +end +return {count, ttl} +`; + +type RedisWithConsume = Redis & { + rlConsume(key: string, windowMs: number): Promise<[number, number]>; +}; + +function withConsumeCommand(redis: Redis): RedisWithConsume { + if (typeof (redis as any).rlConsume !== 'function') { + // ioredis caches the script SHA and transparently handles NOSCRIPT. + redis.defineCommand('rlConsume', { numberOfKeys: 1, lua: CONSUME_SCRIPT }); + } + return redis as RedisWithConsume; +} + +export class RedisRateLimiter implements RateLimiter { readonly backend = 'redis' as const; async consume(key: string, max: number, windowMs: number): Promise { const redis = getRedis(); if (!redis) return { allowed: true }; - const redisKey = `rl:${key}`; try { - const count = await redis.incr(redisKey); - if (count === 1) { - // First hit in this window: set the expiry that defines the window. - await redis.pexpire(redisKey, windowMs); - } + const [count, ttl] = await withConsumeCommand(redis).rlConsume(`rl:${key}`, windowMs); if (count > max) { - const ttl = await redis.pttl(redisKey); const retryAfter = ttl > 0 ? Math.ceil(ttl / 1000) : Math.ceil(windowMs / 1000); return { allowed: false, retryAfter }; } diff --git a/backend/src/routes/auth.ts b/backend/src/routes/auth.ts index 8e65ee0..2528602 100644 --- a/backend/src/routes/auth.ts +++ b/backend/src/routes/auth.ts @@ -21,6 +21,7 @@ import { import { generateId, getNow, toDbBool } from '../lib/utils.js'; import { sendEmail } from '../lib/email.js'; import { rateLimitMiddleware } from '../lib/rateLimit.js'; +import { getLoginLockout } from '../lib/stores/loginLockout.js'; // Per-IP rate limit for sensitive auth endpoints (registration, login, and all // email-dispatching flows) to curb credential stuffing and email flooding. @@ -36,42 +37,6 @@ type AuthUser = User & { const auth = new Hono(); -// Rate limiting store (in production, use Redis) -const loginAttempts = new Map(); -const MAX_LOGIN_ATTEMPTS = 5; -const LOCKOUT_DURATION = 15 * 60 * 1000; // 15 minutes - -function checkRateLimit(email: string): { allowed: boolean; retryAfter?: number } { - const now = Date.now(); - const attempts = loginAttempts.get(email); - - if (!attempts) { - return { allowed: true }; - } - - if (now > attempts.resetAt) { - loginAttempts.delete(email); - return { allowed: true }; - } - - if (attempts.count >= MAX_LOGIN_ATTEMPTS) { - return { allowed: false, retryAfter: Math.ceil((attempts.resetAt - now) / 1000) }; - } - - return { allowed: true }; -} - -function recordFailedAttempt(email: string): void { - const now = Date.now(); - const attempts = loginAttempts.get(email) || { count: 0, resetAt: now + LOCKOUT_DURATION }; - attempts.count++; - loginAttempts.set(email, attempts); -} - -function clearFailedAttempts(email: string): void { - loginAttempts.delete(email); -} - const registerSchema = z.object({ email: z.string().email(), password: z.string().min(10, 'Password must be at least 10 characters'), @@ -188,12 +153,12 @@ auth.post('/register', authRateLimit, zValidator('json', registerSchema), async auth.post('/login', authRateLimit, zValidator('json', loginSchema), async (c) => { const data = c.req.valid('json'); - // Check rate limit - const rateLimit = checkRateLimit(data.email); - if (!rateLimit.allowed) { - return c.json({ + // Per-email lockout (shared across instances when Redis is configured). + const lockout = await getLoginLockout().isLocked(data.email); + if (lockout.locked) { + return c.json({ error: 'Too many login attempts. Please try again later.', - retryAfter: rateLimit.retryAfter + retryAfter: lockout.retryAfter }, 429); } @@ -201,7 +166,7 @@ auth.post('/login', authRateLimit, zValidator('json', loginSchema), async (c) => (db as any).select().from(users).where(eq((users as any).email, data.email)) ); if (!user) { - recordFailedAttempt(data.email); + await getLoginLockout().recordFailure(data.email); return c.json({ error: 'Invalid credentials' }, 401); } @@ -223,12 +188,12 @@ auth.post('/login', authRateLimit, zValidator('json', loginSchema), async (c) => const validPassword = await verifyPassword(data.password, user.password); if (!validPassword) { - recordFailedAttempt(data.email); + await getLoginLockout().recordFailure(data.email); return c.json({ error: 'Invalid credentials' }, 401); } // Clear failed attempts on successful login - clearFailedAttempts(data.email); + await getLoginLockout().clear(data.email); // Transparently upgrade legacy bcrypt hashes to argon2 now that we have the // plaintext and have verified it. Best-effort: a failure here must not block diff --git a/backend/src/routes/lnbits.ts b/backend/src/routes/lnbits.ts index af34907..46efe29 100644 --- a/backend/src/routes/lnbits.ts +++ b/backend/src/routes/lnbits.ts @@ -12,7 +12,7 @@ import { } from '../lib/lnbits.js'; import emailService from '../lib/email.js'; import { getPubSub } from '../lib/stores/pubsub.js'; -import { getLock } from '../lib/stores/lock.js'; +import { getLock, LockUnavailableError } from '../lib/stores/lock.js'; const lnbitsRouter = new Hono(); @@ -106,12 +106,26 @@ async function startBackgroundChecker(ticketId: string, paymentHash: string, exp const expiryMs = expirySeconds * 1000; - const lockToken = await getLock().acquire(`checker:${ticketId}`, expiryMs); - if (!lockToken) { + let lockToken: string | null = null; + let lockUnavailable = false; + try { + lockToken = await getLock().acquire(`checker:${ticketId}`, expiryMs); + } catch (err) { + if (!(err instanceof LockUnavailableError)) throw err; + // Fail open: in webhook-less deployments this poller is the only way a + // Lightning payment gets confirmed, so a Redis outage must not stop it. + // Duplicate pollers across replicas are harmless because + // handlePaymentComplete only acts on rows it actually transitions. + console.warn(`[lnbits] lock backend unavailable, polling ticket ${ticketId} without lock:`, err.message); + lockUnavailable = true; + } + if (!lockToken && !lockUnavailable) { // Another instance is already polling this ticket. return; } - checkerLockTokens.set(ticketId, lockToken); + if (lockToken) { + checkerLockTokens.set(ticketId, lockToken); + } const startTime = Date.now(); let checkCount = 0; @@ -270,25 +284,35 @@ async function handlePaymentComplete(ticketId: string, paymentHash: string) { } // Confirm all tickets in the booking (idempotent: only flip pending -> confirmed) + let transitioned = 0; for (const ticket of ticketsToConfirm) { - await (db as any) + const result: any = await (db as any) .update(tickets) .set({ status: 'confirmed' }) .where(and(eq((tickets as any).id, ticket.id), eq((tickets as any).status, 'pending'))); - + transitioned += result?.changes ?? result?.rowCount ?? 0; + await (db as any) .update(payments) - .set({ + .set({ status: 'paid', reference: paymentHash, paidAt: now, updatedAt: now, }) .where(and(eq((payments as any).ticketId, ticket.id), eq((payments as any).status, 'pending'))); - + console.log(`Ticket ${ticket.id} confirmed via Lightning payment (hash: ${paymentHash})`); } - + + // Only the caller that actually flipped rows sends the emails. The webhook + // and the background poller (and pollers on multiple replicas during a Redis + // outage) can all land here; without this guard they would each email. + if (transitioned === 0) { + console.log(`Ticket ${ticketId} was already confirmed by a concurrent caller, skipping emails`); + return; + } + // Get primary payment for sending receipt const payment = await dbGet( (db as any) @@ -296,7 +320,7 @@ async function handlePaymentComplete(ticketId: string, paymentHash: string) { .from(payments) .where(eq((payments as any).ticketId, ticketId)) ); - + // Send confirmation emails asynchronously // For multi-ticket bookings, send email with all ticket info Promise.all([ @@ -372,8 +396,7 @@ lnbitsRouter.get('/stream/:ticketId', async (c) => { } }, 15000); - // Clean up on disconnect - stream.onAbort(() => { + const cleanup = () => { clearInterval(heartbeat); const connections = activeConnections.get(ticketId); if (connections) { @@ -388,11 +411,28 @@ lnbitsRouter.get('/stream/:ticketId', async (c) => { } } } - }); - - // Keep the stream open - while (true) { - await stream.sleep(30000); + }; + + // Clean up on disconnect + stream.onAbort(cleanup); + + // Keep the stream open, and every 15s fall back to reading the ticket + // status from the DB. Pub/sub is best-effort: a message published while the + // subscriber connection was reconnecting is lost, and without this check + // the client would wait forever on a payment that already confirmed. + try { + while (true) { + await stream.sleep(15000); + const current = await dbGet( + (db as any).select().from(tickets).where(eq((tickets as any).id, ticketId)) + ); + if (current?.status === 'confirmed') { + await sendEvent({ type: 'paid', ticketId }); + return; + } + } + } finally { + cleanup(); } }); }); diff --git a/backend/tsconfig.json b/backend/tsconfig.json index 83e9234..d483629 100644 --- a/backend/tsconfig.json +++ b/backend/tsconfig.json @@ -12,5 +12,5 @@ "resolveJsonModule": true }, "include": ["src/**/*"], - "exclude": ["node_modules", "dist"] + "exclude": ["node_modules", "dist", "src/**/*.test.ts"] } diff --git a/backend/vitest.config.ts b/backend/vitest.config.ts new file mode 100644 index 0000000..6ec74ee --- /dev/null +++ b/backend/vitest.config.ts @@ -0,0 +1,7 @@ +import { defineConfig } from 'vitest/config'; + +export default defineConfig({ + test: { + include: ['src/**/*.test.ts'], + }, +}); diff --git a/deploy/docker-compose.scale.yml b/deploy/docker-compose.scale.yml index ed19405..1d89db8 100644 --- a/deploy/docker-compose.scale.yml +++ b/deploy/docker-compose.scale.yml @@ -28,7 +28,20 @@ services: redis: image: redis:7-alpine - command: ["redis-server", "--appendonly", "yes"] + # noeviction is deliberate: rate-limit, lock, and lockout keys must never + # be evicted for correctness. The cache workload is a single short-TTL key, + # so memory pressure is negligible; if caching ever grows, revisit this or + # move the cache to its own DB index. + command: + [ + "redis-server", + "--appendonly", "yes", + "--requirepass", "${REDIS_PASSWORD:-change-me-redis-password}", + "--maxmemory", "256mb", + "--maxmemory-policy", "noeviction", + ] + environment: + REDISCLI_AUTH: "${REDIS_PASSWORD:-change-me-redis-password}" volumes: - redisdata:/data healthcheck: @@ -46,7 +59,7 @@ services: DB_TYPE: postgres DATABASE_URL: postgresql://spanglish:spanglish@postgres:5432/spanglish DB_POOL_MAX: "15" - REDIS_URL: redis://redis:6379 + REDIS_URL: "redis://:${REDIS_PASSWORD:-change-me-redis-password}@redis:6379" JWT_SECRET: change-me-to-a-strong-secret FRONTEND_URL: http://localhost:8080 # Optional S3-compatible storage so uploads are shared across replicas.