refactor: enforce layered architecture by extracting all DB queries into @sofa/db/queries/*

- Create 18 new query modules under `packages/db/src/queries/` covering every domain (availability, cache, colors, credits, cron, discovery, image-cache, imports, integrations, lists, metadata, person, settings, system-health, thumbhash, title, tracking, webhooks)
- Remove all direct `db` client imports and raw Drizzle expressions from `@sofa/core` services and `apps/server` procedures/cron; replace with named query functions
- Extract cron DB helpers (`startCronRun`, `completeCronRun`, `failCronRun`, `getLibraryTitleIds`, `getStaleLibraryTitles`, etc.) into `packages/core/src/cron.ts`
- Extract integration DB helpers into `packages/core/src/integrations.ts`
- Add `drizzle-orm` as a direct dependency of `@sofa/db` and remove it from `@sofa/core`
- Update `AGENTS.md` to document the strict layered architecture rule

apps/server/src/cron.ts

@@ -3,6 +3,20 @@ import { Cron } from "croner";
 import { refreshAvailability } from "@sofa/core/availability";
 import { createBackup, ensureBackupDir, pruneBackups } from "@sofa/core/backup";
 import { refreshCredits, syncCastProfileThumbHashes } from "@sofa/core/credits";
+import {
+  completeCronRun,
+  failCronRun,
+  getCastEntryForTitle,
+  getLibraryTitleIds,
+  getReturningTvShows,
+  getStaleAvailabilityTitles,
+  getStaleLibraryTitles,
+  getStaleNonLibraryTitlesForRefresh,
+  getThumbhashBackfillTitleIds,
+  getTitleByIdForCron,
+  getTitleIdsWithStaleSeasons,
+  startCronRun,
+} from "@sofa/core/cron";
 import {
   cacheEpisodeStills,
   cacheImagesForTitle,
@@ -20,18 +34,6 @@ import { getSetting } from "@sofa/core/settings";
 import { performTelemetryReport } from "@sofa/core/telemetry";
 import { generateTitleBackdropThumbHash, generateTitlePosterThumbHash } from "@sofa/core/thumbhash";
 import { performUpdateCheck } from "@sofa/core/update-check";
-import { db } from "@sofa/db/client";
-import { and, eq, inArray, isNotNull, lt, or, sql } from "@sofa/db/helpers";
-import {
-  availabilityOffers,
-  cronRuns,
-  episodes,
-  persons,
-  seasons,
-  titleCast,
-  titles,
-  userTitleStatus,
-} from "@sofa/db/schema";
 import { createLogger } from "@sofa/logger";
 import { getTvDetails } from "@sofa/tmdb/client";
 
@@ -57,30 +59,15 @@ function schedule(name: string, cron: string, handler: () => Promise<void>) {
     new Cron(cron, { name, protect: true }, async () => {
       log.info(`Running job: ${name}`);
       const startMs = performance.now();
-      const run = db
-        .insert(cronRuns)
-        .values({ jobName: name, status: "running", startedAt: new Date() })
-        .returning()
-        .get();
+      const run = startCronRun(name);
       try {
         await handler();
         const durationMs = Math.round(performance.now() - startMs);
-        db.update(cronRuns)
-          .set({ status: "success", finishedAt: new Date(), durationMs })
-          .where(eq(cronRuns.id, run.id))
-          .run();
+        completeCronRun(run.id, durationMs);
         log.info(`Completed job: ${name} (${durationMs}ms)`);
       } catch (err) {
         const durationMs = Math.round(performance.now() - startMs);
-        db.update(cronRuns)
-          .set({
-            status: "error",
-            finishedAt: new Date(),
-            durationMs,
-            errorMessage: err instanceof Error ? err.message : String(err),
-          })
-          .where(eq(cronRuns.id, run.id))
-          .run();
+        failCronRun(run.id, durationMs, err);
         log.error(`Job ${name} failed:`, err);
       }
     }),
@@ -108,71 +95,6 @@ export async function triggerJob(name: string): Promise<boolean> {
   return true;
 }
 
-function getLibraryTitleIds(): string[] {
-  const rows = db
-    .select({ titleId: userTitleStatus.titleId })
-    .from(userTitleStatus)
-    .groupBy(userTitleStatus.titleId)
-    .all();
-  return rows.map((r) => r.titleId);
-}
-
-function getThumbhashBackfillTitleIds(): string[] {
-  const titleIds = new Set(getLibraryTitleIds());
-
-  const addIds = (ids: string[]) => {
-    for (const id of ids) titleIds.add(id);
-  };
-
-  addIds(
-    db
-      .select({ id: titles.id })
-      .from(titles)
-      .where(
-        or(
-          and(isNotNull(titles.posterPath), sql`${titles.posterThumbHash} IS NULL`),
-          and(isNotNull(titles.backdropPath), sql`${titles.backdropThumbHash} IS NULL`),
-        ),
-      )
-      .all()
-      .map((row) => row.id),
-  );
-
-  addIds(
-    db
-      .select({ titleId: seasons.titleId })
-      .from(seasons)
-      .where(and(isNotNull(seasons.posterPath), sql`${seasons.posterThumbHash} IS NULL`))
-      .groupBy(seasons.titleId)
-      .all()
-      .map((row) => row.titleId),
-  );
-
-  addIds(
-    db
-      .select({ titleId: seasons.titleId })
-      .from(episodes)
-      .innerJoin(seasons, eq(episodes.seasonId, seasons.id))
-      .where(and(isNotNull(episodes.stillPath), sql`${episodes.stillThumbHash} IS NULL`))
-      .groupBy(seasons.titleId)
-      .all()
-      .map((row) => row.titleId),
-  );
-
-  addIds(
-    db
-      .select({ titleId: titleCast.titleId })
-      .from(titleCast)
-      .innerJoin(persons, eq(titleCast.personId, persons.id))
-      .where(and(isNotNull(persons.profilePath), sql`${persons.profileThumbHash} IS NULL`))
-      .groupBy(titleCast.titleId)
-      .all()
-      .map((row) => row.titleId),
-  );
-
-  return [...titleIds];
-}
-
 // Refresh titles where lastFetchedAt is stale
 async function nightlyRefreshLibrary() {
   const libraryIds = getLibraryTitleIds();
@@ -181,11 +103,7 @@ async function nightlyRefreshLibrary() {
   const nonLibraryStale = new Date(Date.now() - 30 * DAY);
 
   // Library titles: 7 days
-  const staleLibrary = db
-    .select({ id: titles.id })
-    .from(titles)
-    .where(and(inArray(titles.id, libraryIds), lt(titles.lastFetchedAt, libraryStale)))
-    .all();
+  const staleLibrary = getStaleLibraryTitles(libraryIds, libraryStale);
 
   for (const { id } of staleLibrary) {
     await refreshTitle(id);
@@ -193,12 +111,7 @@ async function nightlyRefreshLibrary() {
   }
 
   // Non-library titles: 30 days
-  const nonLibrary = db
-    .select()
-    .from(titles)
-    .where(and(isNotNull(titles.lastFetchedAt), lt(titles.lastFetchedAt, nonLibraryStale)))
-    .limit(50)
-    .all();
+  const nonLibrary = getStaleNonLibraryTitlesForRefresh(nonLibraryStale, 50);
 
   for (const t of nonLibrary) {
     if (!libraryIds.includes(t.id)) {
@@ -214,33 +127,10 @@ async function refreshAvailabilityJob() {
   log.debug(`Checking availability for ${libraryIds.length} library titles`);
   const stale = new Date(Date.now() - DAY);
 
-  const titlesWithOffers = new Set(
-    db
-      .select({ titleId: availabilityOffers.titleId })
-      .from(availabilityOffers)
-      .where(inArray(availabilityOffers.titleId, libraryIds))
-      .groupBy(availabilityOffers.titleId)
-      .all()
-      .map((r) => r.titleId),
-  );
-
-  const titlesWithStaleOffers = new Set(
-    db
-      .select({ titleId: availabilityOffers.titleId })
-      .from(availabilityOffers)
-      .where(
-        and(
-          inArray(availabilityOffers.titleId, libraryIds),
-          lt(availabilityOffers.lastFetchedAt, stale),
-        ),
-      )
-      .groupBy(availabilityOffers.titleId)
-      .all()
-      .map((r) => r.titleId),
-  );
+  const { withOffers, withStaleOffers } = getStaleAvailabilityTitles(libraryIds, stale);
 
   for (const titleId of libraryIds) {
-    if (titlesWithStaleOffers.has(titleId) || !titlesWithOffers.has(titleId)) {
+    if (withStaleOffers.has(titleId) || !withOffers.has(titleId)) {
       await refreshAvailability(titleId);
       await Bun.sleep(RATE_LIMIT_MS);
     }
@@ -258,35 +148,14 @@ async function refreshRecommendationsJob() {
 }
 
 async function refreshTvChildrenJob() {
-  const returningStatuses = ["Returning Series", "In Production"];
   const stale = new Date(Date.now() - 7 * DAY);
 
-  const tvShows = db
-    .select()
-    .from(titles)
-    .where(
-      and(
-        eq(titles.type, "tv"),
-        isNotNull(titles.lastFetchedAt),
-        or(...returningStatuses.map((s) => eq(titles.status, s))),
-      ),
-    )
-    .all();
+  const tvShows = getReturningTvShows();
 
   log.debug(`Checking ${tvShows.length} returning TV shows for stale episodes`);
 
   const tvIds = tvShows.map((s) => s.id);
-  const titlesWithStaleSeasons = new Set(
-    tvIds.length > 0
-      ? db
-          .select({ titleId: seasons.titleId })
-          .from(seasons)
-          .where(and(inArray(seasons.titleId, tvIds), lt(seasons.lastFetchedAt, stale)))
-          .groupBy(seasons.titleId)
-          .all()
-          .map((r) => r.titleId)
-      : [],
-  );
+  const titlesWithStaleSeasons = getTitleIdsWithStaleSeasons(tvIds, stale);
 
   for (const show of tvShows) {
     if (titlesWithStaleSeasons.has(show.id)) {
@@ -304,7 +173,7 @@ async function cacheImagesJob() {
 
   for (const titleId of titleIds) {
     try {
-      const title = db.select().from(titles).where(eq(titles.id, titleId)).get();
+      const title = getTitleByIdForCron(titleId);
       if (!title) continue;
 
       // Phase 1: warm the image cache so thumbhash generation can read from disk
@@ -347,12 +216,7 @@ async function refreshCreditsJob() {
   const stale = new Date(Date.now() - 30 * DAY);
 
   for (const titleId of libraryIds) {
-    const castEntry = db
-      .select()
-      .from(titleCast)
-      .where(eq(titleCast.titleId, titleId))
-      .limit(1)
-      .get();
+    const castEntry = getCastEntryForTitle(titleId);
 
     const needsRefresh = !castEntry || (castEntry.lastFetchedAt && castEntry.lastFetchedAt < stale);
 

apps/server/src/orpc/procedures/imports.ts

@@ -4,15 +4,15 @@ import { AppErrorCode } from "@sofa/api/errors";
 import type { ParseResult } from "@sofa/core/imports";
 import {
   countUnresolved,
+  getActiveImportJobForUser,
+  insertImportJob,
   parseLetterboxdExport,
   parseSimklPayload,
   parseTraktPayload,
   processImportJob,
   readImportJob,
+  updateImportJobProgress,
 } from "@sofa/core/imports";
-import { db } from "@sofa/db/client";
-import { and, eq, inArray } from "@sofa/db/helpers";
-import { importJobs } from "@sofa/db/schema";
 import { createLogger } from "@sofa/logger";
 
 import { os } from "../context";
@@ -100,34 +100,18 @@ export const createJob = os.imports.createJob.use(authed).handler(async ({ input
 
   // Prevent concurrent imports per user.
   // Auto-cancel stale *pending* jobs (server crashed before worker started).
-  // Running jobs are never auto-cancelled — there's no heartbeat to
-  // distinguish active work from a dead worker, and killing a healthy
-  // long-running import is worse than making the user manually cancel.
   const PENDING_STALE_MS = 5 * 60 * 1000; // 5 minutes
   const now = Date.now();
-  const existing = db
-    .select()
-    .from(importJobs)
-    .where(
-      and(
-        eq(importJobs.userId, context.user.id),
-        inArray(importJobs.status, ["pending", "running"]),
-      ),
-    )
-    .get();
+  const existing = getActiveImportJobForUser(context.user.id);
   if (existing) {
     const isPending = existing.status === "pending";
     const isStale = isPending && now - existing.createdAt.getTime() > PENDING_STALE_MS;
     if (isStale) {
-      // Mark as cancelled so the worker loop also stops if it starts late
-      db.update(importJobs)
-        .set({
-          status: "cancelled",
-          finishedAt: new Date(),
-          currentMessage: "Import timed out (stale job auto-cancelled)",
-        })
-        .where(eq(importJobs.id, existing.id))
-        .run();
+      updateImportJobProgress(existing.id, {
+        status: "cancelled",
+        finishedAt: new Date(),
+        currentMessage: "Import timed out (stale job auto-cancelled)",
+      });
       log.warn(`Auto-cancelled stale import job ${existing.id}`);
     } else {
       throw new ORPCError("CONFLICT", {
@@ -137,20 +121,16 @@ export const createJob = os.imports.createJob.use(authed).handler(async ({ input
     }
   }
 
-  const job = db
-    .insert(importJobs)
-    .values({
-      userId: context.user.id,
-      source: data.source,
-      status: "pending",
-      payload: JSON.stringify(data),
-      importWatches: options.importWatches,
-      importWatchlist: options.importWatchlist,
-      importRatings: options.importRatings,
-      createdAt: new Date(),
-    })
-    .returning()
-    .get();
+  const job = insertImportJob({
+    userId: context.user.id,
+    source: data.source,
+    status: "pending",
+    payload: JSON.stringify(data),
+    importWatches: options.importWatches,
+    importWatchlist: options.importWatchlist,
+    importRatings: options.importRatings,
+    createdAt: new Date(),
+  });
 
   // Fire-and-forget processing
   processImportJob(job.id).catch((err) => {
@@ -172,7 +152,7 @@ export const cancelJob = os.imports.cancelJob.use(authed).handler(({ input, cont
       data: { code: AppErrorCode.IMPORT_CANNOT_CANCEL },
     });
   }
-  db.update(importJobs).set({ status: "cancelled" }).where(eq(importJobs.id, input.id)).run();
+  updateImportJobProgress(input.id, { status: "cancelled" });
   return readImportJob(input.id);
 });
 

apps/server/src/orpc/procedures/integrations.ts

@@ -1,132 +1,35 @@
 import { ORPCError } from "@orpc/server";
 
 import { AppErrorCode } from "@sofa/api/errors";
-import { db } from "@sofa/db/client";
-import { and, desc, eq } from "@sofa/db/helpers";
-import { integrationEvents, integrations } from "@sofa/db/schema";
+import {
+  createOrUpdateIntegration,
+  deleteIntegration as coreDeleteIntegration,
+  listUserIntegrations,
+  regenerateToken as coreRegenerateToken,
+  serializeIntegration,
+} from "@sofa/core/integrations";
 
 import { os } from "../context";
 import { authed } from "../middleware";
 
-const LIST_PROVIDERS = new Set(["sonarr", "radarr"]);
-
-function integrationTypeFor(provider: string): "webhook" | "list" {
-  return LIST_PROVIDERS.has(provider) ? "list" : "webhook";
-}
-
-function generateToken() {
-  return Buffer.from(crypto.getRandomValues(new Uint8Array(32))).toString("hex");
-}
-
-function serializeIntegration(row: {
-  id: string;
-  provider: string;
-  type: "webhook" | "list";
-  token: string;
-  enabled: boolean;
-  lastEventAt: Date | null;
-  createdAt: Date;
-}) {
-  return {
-    ...row,
-    lastEventAt: row.lastEventAt?.toISOString() ?? null,
-    createdAt: row.createdAt.toISOString(),
-  };
-}
-
 export const list = os.integrations.list.use(authed).handler(({ context }) => {
-  const userIntegrations = db
-    .select()
-    .from(integrations)
-    .where(eq(integrations.userId, context.user.id))
-    .all();
-
-  const eventsByIntegration = new Map<string, (typeof integrationEvents.$inferSelect)[]>();
-  for (const integration of userIntegrations) {
-    const events = db
-      .select()
-      .from(integrationEvents)
-      .where(eq(integrationEvents.integrationId, integration.id))
-      .orderBy(desc(integrationEvents.receivedAt))
-      .limit(10)
-      .all();
-    eventsByIntegration.set(integration.id, events);
-  }
-
-  const result = userIntegrations.map((integration) => {
-    const events = eventsByIntegration.get(integration.id) ?? [];
-    return Object.assign(serializeIntegration(integration), {
-      recentEvents: events.map((e) => ({
-        id: e.id,
-        eventType: e.eventType,
-        mediaType: e.mediaType,
-        mediaTitle: e.mediaTitle,
-        status: e.status,
-        receivedAt: e.receivedAt.toISOString(),
-      })),
-    });
-  });
-
-  return { integrations: result };
+  return listUserIntegrations(context.user.id);
 });
 
 export const create = os.integrations.create.use(authed).handler(({ input, context }) => {
-  const existing = db
-    .select()
-    .from(integrations)
-    .where(and(eq(integrations.userId, context.user.id), eq(integrations.provider, input.provider)))
-    .get();
-
-  if (existing) {
-    if (input.enabled !== undefined) {
-      const row = db
-        .update(integrations)
-        .set({ enabled: input.enabled })
-        .where(eq(integrations.id, existing.id))
-        .returning()
-        .get();
-      return serializeIntegration(row);
-    }
-    return serializeIntegration(existing);
-  }
-
-  const row = db
-    .insert(integrations)
-    .values({
-      userId: context.user.id,
-      provider: input.provider,
-      type: integrationTypeFor(input.provider),
-      token: generateToken(),
-      enabled: input.enabled ?? true,
-      createdAt: new Date(),
-    })
-    .returning()
-    .get();
-
-  return serializeIntegration(row);
+  return createOrUpdateIntegration(context.user.id, input.provider, input.enabled);
 });
 
 export const deleteIntegration = os.integrations.delete
   .use(authed)
   .handler(({ input, context }) => {
-    db.delete(integrations)
-      .where(
-        and(eq(integrations.userId, context.user.id), eq(integrations.provider, input.provider)),
-      )
-      .run();
+    coreDeleteIntegration(context.user.id, input.provider);
   });
 
 export const regenerateToken = os.integrations.regenerateToken
   .use(authed)
   .handler(({ input, context }) => {
-    const row = db
-      .update(integrations)
-      .set({ token: generateToken() })
-      .where(
-        and(eq(integrations.userId, context.user.id), eq(integrations.provider, input.provider)),
-      )
-      .returning()
-      .get();
+    const row = coreRegenerateToken(context.user.id, input.provider);
 
     if (!row) {
       throw new ORPCError("NOT_FOUND", {

apps/server/src/orpc/procedures/seasons.ts

@@ -1,17 +1,10 @@
-import { logEpisodeWatchBatch, unwatchSeason } from "@sofa/core/tracking";
-import { db } from "@sofa/db/client";
-import { eq } from "@sofa/db/helpers";
-import { episodes } from "@sofa/db/schema";
+import { unwatchSeason, watchSeason } from "@sofa/core/tracking";
 
 import { os } from "../context";
 import { authed } from "../middleware";
 
 export const watch = os.seasons.watch.use(authed).handler(({ input, context }) => {
-  const seasonEps = db.select().from(episodes).where(eq(episodes.seasonId, input.id)).all();
-  logEpisodeWatchBatch(
-    context.user.id,
-    seasonEps.map((ep) => ep.id),
-  );
+  watchSeason(context.user.id, input.id);
 });
 
 export const unwatch = os.seasons.unwatch.use(authed).handler(({ input, context }) => {

apps/server/src/orpc/procedures/titles.ts

@@ -8,13 +8,11 @@ import {
   getUserTitleInfo,
   logMovieWatch,
   markAllEpisodesWatched,
+  quickAddTitle,
   rateTitleStars,
   removeTitleStatus,
   setTitleStatus,
 } from "@sofa/core/tracking";
-import { db } from "@sofa/db/client";
-import { and, eq } from "@sofa/db/helpers";
-import { titles, userTitleStatus } from "@sofa/db/schema";
 
 import { os } from "../context";
 import { authed } from "../middleware";
@@ -65,31 +63,16 @@ export const recommendations = os.titles.recommendations
   });
 
 export const quickAdd = os.titles.quickAdd.use(authed).handler(async ({ input, context }) => {
-  // Look up the title (it exists as a shell from browse/search import)
-  const title = db
-    .select({ id: titles.id, tmdbId: titles.tmdbId, type: titles.type })
-    .from(titles)
-    .where(eq(titles.id, input.id))
-    .get();
-  if (!title) {
+  const result = quickAddTitle(context.user.id, input.id);
+  if (!result) {
     throw new ORPCError("NOT_FOUND", {
       message: "Title not found",
       data: { code: AppErrorCode.TITLE_NOT_FOUND },
     });
   }
 
-  // Trigger full TMDB import if still a shell
-  getOrFetchTitleByTmdbId(title.tmdbId, title.type as "movie" | "tv").catch(() => {});
+  // Trigger full TMDB import if still a shell (fire-and-forget)
+  getOrFetchTitleByTmdbId(result.tmdbId, result.type as "movie" | "tv").catch(() => {});
 
-  const existing = db
-    .select()
-    .from(userTitleStatus)
-    .where(and(eq(userTitleStatus.userId, context.user.id), eq(userTitleStatus.titleId, title.id)))
-    .get();
-
-  if (!existing) {
-    setTitleStatus(context.user.id, title.id, "watchlist");
-  }
-
-  return { id: title.id, alreadyAdded: !!existing };
+  return { id: result.id, alreadyAdded: result.alreadyAdded };
 });

apps/server/src/routes/webhooks.ts

@@ -1,5 +1,6 @@
 import { Hono } from "hono";
 
+import { findIntegrationByToken } from "@sofa/core/integrations";
 import type { WebhookEvent } from "@sofa/core/webhooks";
 import {
   parseEmbyPayload,
@@ -7,9 +8,6 @@ import {
   parsePlexPayload,
   processWebhook,
 } from "@sofa/core/webhooks";
-import { db } from "@sofa/db/client";
-import { eq } from "@sofa/db/helpers";
-import { integrations } from "@sofa/db/schema";
 import { createLogger } from "@sofa/logger";
 
 const log = createLogger("webhooks");
@@ -20,7 +18,7 @@ app.post("/:token", async (c) => {
   const token = c.req.param("token");
 
   // Look up connection by token — this IS the auth
-  const connection = db.select().from(integrations).where(eq(integrations.token, token)).get();
+  const connection = findIntegrationByToken(token);
 
   if (!connection || !connection.enabled) {
     // Always return 200 to avoid retry storms from media servers