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);