From 95cd740e519f3c17b9d7d1a31a5731b7e8137f33 Mon Sep 17 00:00:00 2001 From: chris Date: Sat, 3 Oct 2026 11:28:17 -0400 Subject: [PATCH] Separate app construction from server startup buildApp() returns the configured Fastify instance without listening, so tests can drive it with inject(). Shutdown of database, Redis and queue connections now lives in one place shared by the server and worker. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MzFywujuH9TiuXbr9vjoX8 --- apps/api/src/app.ts | 64 +++++++++++++++++++++++++++++++++ apps/api/src/lib/lifecycle.ts | 8 +++++ apps/api/src/server.ts | 62 +++----------------------------- apps/api/src/services/queues.ts | 5 +++ apps/api/src/worker.ts | 5 +-- 5 files changed, 84 insertions(+), 60 deletions(-) create mode 100644 apps/api/src/app.ts create mode 100644 apps/api/src/lib/lifecycle.ts diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts new file mode 100644 index 0000000..151e2c0 --- /dev/null +++ b/apps/api/src/app.ts @@ -0,0 +1,64 @@ +import cookie from '@fastify/cookie'; +import cors from '@fastify/cors'; +import rateLimit from '@fastify/rate-limit'; +import websocket from '@fastify/websocket'; +import Fastify from 'fastify'; +import { ZodError } from 'zod'; +import { sql } from './db'; +import { env } from './env'; +import { loadUser } from './lib/auth'; +import { HttpError } from './lib/http'; +import { redis } from './lib/redis'; +import { authRoutes } from './routes/auth'; +import { connectionRoutes } from './routes/connections'; +import { meRoutes } from './routes/me'; +import { realtimeRoutes } from './routes/realtime'; +import { trackRoutes } from './routes/tracks'; +import { userRoutes } from './routes/users'; +import { webhookRoutes } from './routes/webhooks'; + +/** Builds the API without listening, so tests can drive it with app.inject(). */ +export async function buildApp({ logger = true, rateLimits = true }: { logger?: boolean; rateLimits?: boolean } = {}) { + const app = Fastify({ logger, trustProxy: true }); + + await app.register(cors, { + origin: [env.WEB_URL], + credentials: true, + methods: ['GET', 'POST', 'PUT', 'PATCH', 'DELETE'], + }); + await app.register(cookie); + // Rate limits are per-route (see `config.rateLimit`); the Redis store makes them global across instances. + if (rateLimits) await app.register(rateLimit, { global: false, redis }); + await app.register(websocket); + + app.decorateRequest('user', null); + app.decorateRequest('sessionId', null); + app.addHook('onRequest', loadUser); + + app.setErrorHandler((err, req, reply) => { + if (err instanceof ZodError) { + return reply.status(400).send({ error: err.issues[0]?.message ?? 'Invalid input', issues: err.issues }); + } + if (err instanceof HttpError) return reply.status(err.status).send({ error: err.message }); + const status = (err as { statusCode?: number }).statusCode; + if (status && status < 500) return reply.status(status).send({ error: (err as Error).message }); + req.log.error(err); + return reply.status(500).send({ error: 'Internal server error' }); + }); + + app.get('/health', async () => { + await Promise.all([sql`select 1`, redis.ping()]); + return { ok: true }; + }); + + await app.register(authRoutes); + await app.register(meRoutes); + await app.register(userRoutes); + await app.register(trackRoutes); + await app.register(connectionRoutes); + await app.register(realtimeRoutes); + // registered as its own plugin so its permissive body parsers stay scoped to webhook routes + await app.register(webhookRoutes); + + return app; +} diff --git a/apps/api/src/lib/lifecycle.ts b/apps/api/src/lib/lifecycle.ts new file mode 100644 index 0000000..eed9b66 --- /dev/null +++ b/apps/api/src/lib/lifecycle.ts @@ -0,0 +1,8 @@ +import { sql } from '../db'; +import { closeQueues } from '../services/queues'; +import { redis } from './redis'; + +/** Close every long-lived connection opened at import time (graceful shutdown, tests). */ +export async function closeConnections() { + await Promise.allSettled([sql.end({ timeout: 5 }), closeQueues(), redis.quit()]); +} diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index fc46284..8c25362 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -1,67 +1,13 @@ -import cookie from '@fastify/cookie'; -import cors from '@fastify/cors'; -import rateLimit from '@fastify/rate-limit'; -import websocket from '@fastify/websocket'; -import Fastify from 'fastify'; -import { ZodError } from 'zod'; -import { sql } from './db'; +import { buildApp } from './app'; +import { closeConnections } from './lib/lifecycle'; import { env } from './env'; -import { loadUser } from './lib/auth'; -import { HttpError } from './lib/http'; -import { redis } from './lib/redis'; -import { authRoutes } from './routes/auth'; -import { connectionRoutes } from './routes/connections'; -import { meRoutes } from './routes/me'; -import { realtimeRoutes } from './routes/realtime'; -import { trackRoutes } from './routes/tracks'; -import { userRoutes } from './routes/users'; -import { webhookRoutes } from './routes/webhooks'; -const app = Fastify({ logger: true, trustProxy: true }); - -await app.register(cors, { - origin: [env.WEB_URL], - credentials: true, - methods: ['GET', 'POST', 'PUT', 'PATCH', 'DELETE'], -}); -await app.register(cookie); -// Rate limits are per-route (see `config.rateLimit`); the Redis store makes them global across instances. -await app.register(rateLimit, { global: false, redis }); -await app.register(websocket); - -app.decorateRequest('user', null); -app.decorateRequest('sessionId', null); -app.addHook('onRequest', loadUser); - -app.setErrorHandler((err, req, reply) => { - if (err instanceof ZodError) { - return reply.status(400).send({ error: err.issues[0]?.message ?? 'Invalid input', issues: err.issues }); - } - if (err instanceof HttpError) return reply.status(err.status).send({ error: err.message }); - const status = (err as { statusCode?: number }).statusCode; - if (status && status < 500) return reply.status(status).send({ error: (err as Error).message }); - req.log.error(err); - return reply.status(500).send({ error: 'Internal server error' }); -}); - -app.get('/health', async () => { - await Promise.all([sql`select 1`, redis.ping()]); - return { ok: true }; -}); - -await app.register(authRoutes); -await app.register(meRoutes); -await app.register(userRoutes); -await app.register(trackRoutes); -await app.register(connectionRoutes); -await app.register(realtimeRoutes); -// registered as its own plugin so its permissive body parsers stay scoped to webhook routes -await app.register(webhookRoutes); +const app = await buildApp(); for (const signal of ['SIGINT', 'SIGTERM']) { process.on(signal, async () => { await app.close(); - await sql.end(); + await closeConnections(); process.exit(0); }); } diff --git a/apps/api/src/services/queues.ts b/apps/api/src/services/queues.ts index 425fc0a..ac122d2 100644 --- a/apps/api/src/services/queues.ts +++ b/apps/api/src/services/queues.ts @@ -9,6 +9,11 @@ export const RESOLVE_QUEUE = 'resolve-track'; export const pollQueue = new Queue<{ connectionId: string }>(POLL_QUEUE, { connection }); export const resolveQueue = new Queue<{ trackId: string }>(RESOLVE_QUEUE, { connection }); +export async function closeQueues() { + await Promise.all([pollQueue.close(), resolveQueue.close()]); + await connection.quit(); +} + export function enqueueResolve(trackId: string) { return resolveQueue.add( 'resolve', diff --git a/apps/api/src/worker.ts b/apps/api/src/worker.ts index ccaaa71..8aeaec1 100644 --- a/apps/api/src/worker.ts +++ b/apps/api/src/worker.ts @@ -1,7 +1,8 @@ import { Worker } from 'bullmq'; import { and, eq, inArray } from 'drizzle-orm'; -import { db, schema, sql } from './db'; +import { db, schema } from './db'; import { entitlementsFor } from './lib/entitlements'; +import { closeConnections } from './lib/lifecycle'; import { createRedis } from './lib/redis'; import { POLLED_PROVIDERS, pollingProviders } from './providers'; import { ProviderAuthError, ProviderRateLimitError } from './providers/types'; @@ -114,7 +115,7 @@ async function shutdown() { clearInterval(tickTimer); clearInterval(reconcileTimer); await Promise.all([pollWorker.close(), resolveWorker.close()]); - await sql.end(); + await closeConnections(); process.exit(0); } process.on('SIGTERM', shutdown);