perf(imports): poll job progress without the payload and drop it when done

readImportJob now selects only progress columns instead of the multi-MB
payload, processImportJob empties the payload once a job reaches a terminal
status, and server startup clears payloads of previously finished jobs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

apps/server/src/index.ts

@@ -8,7 +8,7 @@ import { ensureImageDirs, imageCacheEnabled } from "@sofa/core/image-cache";
 import { registerJobScheduleProvider } from "@sofa/core/system-health";
 import { closeDatabase, isDatabaseAccessBlocked } from "@sofa/db/client";
 import { runMigrations } from "@sofa/db/migrate";
-import { recoverStaleImportJobs } from "@sofa/db/queries/imports";
+import { clearFinishedImportPayloads, recoverStaleImportJobs } from "@sofa/db/queries/imports";
 import { seedPlatforms } from "@sofa/db/seed-platforms";
 import { createLogger } from "@sofa/logger";
 
@@ -46,6 +46,12 @@ if (recoveredJobs > 0) {
   log.warn(`Recovered ${recoveredJobs} stale import job(s) from previous shutdown`);
 }
 
+// Payloads of finished imports are never read again; reclaim the space (and keep backups small).
+const clearedPayloads = clearFinishedImportPayloads();
+if (clearedPayloads > 0) {
+  log.info(`Cleared stored payloads of ${clearedPayloads} finished import job(s)`);
+}
+
 // Wire up job schedule provider for system health
 registerJobScheduleProvider(getJobSchedules);
 

packages/core/src/imports/processor.ts

@@ -3,8 +3,10 @@ import { ORPCError } from "@orpc/server";
 import { type ImportJob, NormalizedImportSchema } from "@sofa/api/schemas";
 import {
   backdateTitleStatusAddedAt,
+  clearFinishedImportPayloads,
   getImportJob,
   getImportJobStatus,
+  getImportJobSummary,
   hasEpisodeWatch,
   hasEpisodeWatchBetween,
   hasMovieWatch,
@@ -297,7 +299,7 @@ async function processRating(
 // ─── Read Job Helper ─────────────────────────────────────────────────
 
 export function readImportJob(jobId: string, userId?: string): ImportJob {
-  const row = getImportJob(jobId);
+  const row = getImportJobSummary(jobId);
 
   if (!row) {
     throw new ORPCError("NOT_FOUND", { message: `Import job ${jobId} not found` });
@@ -328,6 +330,18 @@ export function readImportJob(jobId: string, userId?: string): ImportJob {
 // ─── Job Processor ───────────────────────────────────────────────────
 
 export async function processImportJob(jobId: string): Promise<void> {
+  try {
+    await runImportJob(jobId);
+  } finally {
+    try {
+      clearFinishedImportPayloads(jobId);
+    } catch (err) {
+      log.warn(`Failed to clear payload for import job ${jobId}:`, err);
+    }
+  }
+}
+
+async function runImportJob(jobId: string): Promise<void> {
   const row = getImportJob(jobId);
 
   if (!row) {

packages/core/test/import-processor.test.ts

@@ -4,6 +4,7 @@ import { fileURLToPath } from "node:url";
 
 import { beforeEach, describe, expect, test, vi } from "vitest";
 
+import { clearFinishedImportPayloads } from "@sofa/db/queries/imports";
 import * as maintenance from "@sofa/db/queries/maintenance";
 import {
   importJobs,
@@ -772,3 +773,109 @@ describe("processImportJob — rewatches and added dates", () => {
     }
   });
 });
+
+// ── Payload retention ───────────────────────────────────────────────
+
+const blankPayload = (source: NormalizedImport["source"]): NormalizedImport => ({
+  source,
+  movies: [],
+  episodes: [],
+  watchlist: [],
+  ratings: [],
+});
+
+function readPayload(jobId: string) {
+  return testDb
+    .select({ payload: importJobs.payload })
+    .from(importJobs)
+    .where(eq(importJobs.id, jobId))
+    .get()?.payload;
+}
+
+describe("processImportJob — payload retention", () => {
+  test("empties the payload after a successful import", async () => {
+    const userId = insertUser();
+    insertMovieTitle("movie-1", 550, "Fight Club");
+
+    const payload: NormalizedImport = {
+      ...blankPayload("trakt"),
+      movies: [{ tmdbId: 550, title: "Fight Club", year: 1999, watchedAt: "2024-06-15T20:00:00Z" }],
+    };
+
+    const jobId = createJob(userId, payload);
+    expect(readPayload(jobId)).not.toBe("");
+
+    await processImportJob(jobId);
+
+    expect(readImportJob(jobId).status).toBe("success");
+    expect(readPayload(jobId)).toBe("");
+  });
+
+  test("empties the payload of a job that fails on invalid JSON", async () => {
+    const userId = insertUser();
+    const jobId = "job-invalid-json";
+    testDb
+      .insert(importJobs)
+      .values({
+        id: jobId,
+        userId,
+        source: "trakt",
+        status: "pending",
+        payload: "{not json",
+        importWatches: true,
+        importWatchlist: true,
+        importRatings: true,
+        createdAt: new Date(),
+      })
+      .run();
+
+    await processImportJob(jobId);
+
+    const job = testDb.select().from(importJobs).where(eq(importJobs.id, jobId)).get();
+    expect(job?.status).toBe("error");
+    expect(job?.payload).toBe("");
+  });
+
+  test("clearFinishedImportPayloads only empties finished jobs", () => {
+    // Only one active (pending/running) job is allowed per user
+    const pendingId = createJob(insertUser("user-pending"), blankPayload("trakt"));
+    const successId = createJob(insertUser("user-success"), blankPayload("trakt"));
+    testDb.update(importJobs).set({ status: "success" }).where(eq(importJobs.id, successId)).run();
+
+    expect(clearFinishedImportPayloads()).toBe(1);
+
+    expect(readPayload(pendingId)).not.toBe("");
+    expect(readPayload(successId)).toBe("");
+    expect(clearFinishedImportPayloads()).toBe(0);
+  });
+
+  test("readImportJob still returns progress fields after the payload is cleared", () => {
+    const userId = insertUser();
+    const jobId = createJob(userId, blankPayload("trakt"));
+    testDb
+      .update(importJobs)
+      .set({
+        status: "error",
+        totalItems: 5,
+        processedItems: 3,
+        importedCount: 2,
+        skippedCount: 1,
+        failedCount: 0,
+        errors: JSON.stringify(["boom"]),
+        finishedAt: new Date(),
+      })
+      .where(eq(importJobs.id, jobId))
+      .run();
+    clearFinishedImportPayloads(jobId);
+    expect(readPayload(jobId)).toBe("");
+
+    const job = readImportJob(jobId, userId);
+    expect(job.status).toBe("error");
+    expect(job.totalItems).toBe(5);
+    expect(job.processedItems).toBe(3);
+    expect(job.importedCount).toBe(2);
+    expect(job.skippedCount).toBe(1);
+    expect(job.errors).toEqual(["boom"]);
+    expect(job.finishedAt).not.toBeNull();
+  });
+});

packages/db/src/queries/imports.ts

@@ -1,4 +1,4 @@
-import { and, eq, gt, gte, inArray, lte } from "drizzle-orm";
+import { and, eq, gt, gte, inArray, lte, ne } from "drizzle-orm";
 
 import { db } from "../client";
 import {
@@ -116,6 +116,44 @@ export function getImportJob(jobId: string) {
   return db.select().from(importJobs).where(eq(importJobs.id, jobId)).get();
 }
 
+/** A job without its payload (which can be several MB) — for progress polling. */
+export function getImportJobSummary(jobId: string) {
+  return db
+    .select({
+      id: importJobs.id,
+      userId: importJobs.userId,
+      source: importJobs.source,
+      status: importJobs.status,
+      totalItems: importJobs.totalItems,
+      processedItems: importJobs.processedItems,
+      importedCount: importJobs.importedCount,
+      skippedCount: importJobs.skippedCount,
+      failedCount: importJobs.failedCount,
+      currentMessage: importJobs.currentMessage,
+      errors: importJobs.errors,
+      warnings: importJobs.warnings,
+      createdAt: importJobs.createdAt,
+      startedAt: importJobs.startedAt,
+      finishedAt: importJobs.finishedAt,
+    })
+    .from(importJobs)
+    .where(eq(importJobs.id, jobId))
+    .get();
+}
+
+const FINISHED_STATUSES = ["success", "error", "cancelled"] as const;
+
+/** Empty the payload of finished jobs; only the processor reads it. Returns rows changed. */
+export function clearFinishedImportPayloads(jobId?: string): number {
+  const finished = inArray(importJobs.status, [...FINISHED_STATUSES]);
+  return db
+    .update(importJobs)
+    .set({ payload: "" })
+    .where(and(finished, ne(importJobs.payload, ""), jobId ? eq(importJobs.id, jobId) : undefined))
+    .returning({ id: importJobs.id })
+    .all().length;
+}
+
 export function updateImportJobProgress(
   jobId: string,
   values: Partial<typeof importJobs.$inferInsert>,