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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MzFywujuH9TiuXbr9vjoX8

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;
+}

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()]);
+}

apps/api/src/server.ts

@@ -1,67 +1,13 @@
-import cookie from '@fastify/cookie';
+import { buildApp } from './app';
-import cors from '@fastify/cors';
+import { closeConnections } from './lib/lifecycle';
-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';
 
-const app = Fastify({ logger: true, trustProxy: true });
+const app = await buildApp();
-
-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);
 
 for (const signal of ['SIGINT', 'SIGTERM']) {
   process.on(signal, async () => {
     await app.close();
-    await sql.end();
+    await closeConnections();
     process.exit(0);
   });
 }

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',

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