fix(imports): keep rewatches and added dates; honor late cancellation

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

packages/core/src/imports/processor.ts

@@ -2,10 +2,13 @@ import { ORPCError } from "@orpc/server";
 
 import { type ImportJob, NormalizedImportSchema } from "@sofa/api/schemas";
 import {
+  backdateTitleStatusAddedAt,
   getImportJob,
   getImportJobStatus,
   hasEpisodeWatch,
+  hasEpisodeWatchBetween,
   hasMovieWatch,
+  hasMovieWatchBetween,
   hasRating,
   getTitleStatusValue,
   updateImportJobProgress,
@@ -20,6 +23,26 @@ import { resolveMovieTmdbId, resolveShowTmdbId } from "./resolve";
 
 const log = createLogger("imports");
 
+/** Two plays of the same item this close together are treated as the same play
+ * (e.g. a Plex webhook watch and the same play scrobbled to Trakt). */
+const TIMESTAMP_DEDUPE_WINDOW_MS = 3 * 60 * 60 * 1000;
+/** Date-only entries (Letterboxd diary) match any watch within ±36h of UTC
+ * midnight on that date, which covers every timezone. */
+const DATE_ONLY_DEDUPE_WINDOW_MS = 36 * 60 * 60 * 1000;
+
+function importedWatchTime(item: {
+  watchedAt?: string;
+  watchedOn?: string;
+}): { at: Date; windowMs: number } | null {
+  const candidate = item.watchedAt
+    ? { at: new Date(item.watchedAt), windowMs: TIMESTAMP_DEDUPE_WINDOW_MS }
+    : item.watchedOn
+      ? { at: new Date(item.watchedOn), windowMs: DATE_ONLY_DEDUPE_WINDOW_MS }
+      : null;
+  if (!candidate || Number.isNaN(candidate.at.getTime())) return null;
+  return candidate;
+}
+
 function safeParseJsonArray(value: string | null): string[] {
   if (!value) return [];
   try {
@@ -79,17 +102,21 @@ async function processMovie(
     return;
   }
 
-  if (hasMovieWatch(userId, title.id)) {
+  const time = importedWatchTime(movie);
+  const isDuplicate = time
+    ? hasMovieWatchBetween(
+        userId,
+        title.id,
+        new Date(time.at.getTime() - time.windowMs),
+        new Date(time.at.getTime() + time.windowMs),
+      )
+    : hasMovieWatch(userId, title.id); // undated: any existing watch counts
+  if (isDuplicate) {
     result.skipped++;
     return;
   }
 
-  const watchedAt = movie.watchedAt
-    ? new Date(movie.watchedAt)
-    : movie.watchedOn
-      ? new Date(movie.watchedOn)
-      : undefined;
-  logMovieWatch(userId, title.id, "import", watchedAt);
+  logMovieWatch(userId, title.id, "import", time?.at);
   result.imported++;
 }
 
@@ -142,17 +169,21 @@ async function processEpisode(
     return;
   }
 
-  if (hasEpisodeWatch(userId, episode.id)) {
+  const time = importedWatchTime(ep);
+  const isDuplicate = time
+    ? hasEpisodeWatchBetween(
+        userId,
+        episode.id,
+        new Date(time.at.getTime() - time.windowMs),
+        new Date(time.at.getTime() + time.windowMs),
+      )
+    : hasEpisodeWatch(userId, episode.id); // undated: any existing watch counts
+  if (isDuplicate) {
     result.skipped++;
     return;
   }
 
-  const watchedAt = ep.watchedAt
-    ? new Date(ep.watchedAt)
-    : ep.watchedOn
-      ? new Date(ep.watchedOn)
-      : undefined;
-  logEpisodeWatch(userId, episode.id, "import", watchedAt);
+  logEpisodeWatch(userId, episode.id, "import", time?.at);
   result.imported++;
 }
 
@@ -200,14 +231,18 @@ async function processWatchlistItem(
         ? "watchlist"
         : requested;
   const currentStatus = getTitleStatusValue(userId, title.id);
+  const parsedAddedAt = item.addedAt ? new Date(item.addedAt) : undefined;
+  const addedAt =
+    parsedAddedAt && !Number.isNaN(parsedAddedAt.getTime()) ? parsedAddedAt : undefined;
 
   if (currentStatus && STATUS_RANK[currentStatus] >= STATUS_RANK[targetStatus]) {
+    if (addedAt) backdateTitleStatusAddedAt(userId, title.id, addedAt);
     result.skipped++;
     return;
   }
 
-  const addedAt = item.addedAt ? new Date(item.addedAt) : undefined;
   setTitleStatus(userId, title.id, targetStatus, "import", addedAt);
+  if (addedAt) backdateTitleStatusAddedAt(userId, title.id, addedAt);
   result.imported++;
 }
 
@@ -332,6 +367,20 @@ export async function processImportJob(jobId: string): Promise<void> {
     warnings: [],
   };
 
+  const finishCancelled = (processed: number) => {
+    updateImportJobProgress(jobId, {
+      finishedAt: new Date(),
+      processedItems: processed,
+      importedCount: result.imported,
+      skippedCount: result.skipped,
+      failedCount: result.failed,
+      errors: JSON.stringify(result.errors),
+      warnings: JSON.stringify(result.warnings),
+      currentMessage: "Import cancelled",
+    });
+    log.info(`Import job ${jobId} cancelled by user`);
+  };
+
   try {
     // Build item list based on options
     const items: { type: string; index: number }[] = [];
@@ -398,17 +447,7 @@ export async function processImportJob(jobId: string): Promise<void> {
       if (i % progressInterval === 0) {
         const currentStatus = getImportJobStatus(jobId);
         if (currentStatus?.status === "cancelled") {
-          updateImportJobProgress(jobId, {
-            finishedAt: new Date(),
-            processedItems: i,
-            importedCount: result.imported,
-            skippedCount: result.skipped,
-            failedCount: result.failed,
-            errors: JSON.stringify(result.errors),
-            warnings: JSON.stringify(result.warnings),
-            currentMessage: "Import cancelled",
-          });
-          log.info(`Import job ${jobId} cancelled by user`);
+          finishCancelled(i);
           return;
         }
       }
@@ -466,6 +505,13 @@ export async function processImportJob(jobId: string): Promise<void> {
       }
     }
 
+    // A cancel may have arrived after the last periodic check. This check and the
+    // success write below are synchronous (no await), so they cannot race.
+    if (getImportJobStatus(jobId)?.status === "cancelled") {
+      finishCancelled(total);
+      return;
+    }
+
     // Success
     updateImportJobProgress(jobId, {
       status: "success",
@@ -483,6 +529,12 @@ export async function processImportJob(jobId: string): Promise<void> {
       `Import job ${jobId} complete: ${result.imported} imported, ${result.skipped} skipped, ${result.failed} failed`,
     );
   } catch (err) {
+    // A cancelled job must not be rewritten to "error"
+    if (getImportJobStatus(jobId)?.status === "cancelled") {
+      finishCancelled(result.imported + result.skipped + result.failed);
+      return;
+    }
+
     // Fatal error
     result.errors.push(`Fatal: ${err instanceof Error ? err.message : String(err)}`);
     updateImportJobProgress(jobId, {

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

@@ -553,3 +553,183 @@ describe("processImportJob — failed resolution", () => {
     }
   });
 });
+
+// ── Rewatches, added dates, late cancellation ───────────────────────
+
+describe("processImportJob — rewatches and added dates", () => {
+  const movieWatches = (userId: string) =>
+    testDb.select().from(userMovieWatches).where(eq(userMovieWatches.userId, userId)).all();
+
+  const emptyPayload = (source: NormalizedImport["source"]): NormalizedImport => ({
+    source,
+    movies: [],
+    episodes: [],
+    watchlist: [],
+    ratings: [],
+  });
+
+  const twoPlays: NormalizedImport = {
+    ...emptyPayload("trakt"),
+    movies: [
+      { tmdbId: 550, title: "Fight Club", watchedAt: "2024-01-01T20:00:00Z" },
+      { tmdbId: 550, title: "Fight Club", watchedAt: "2024-06-01T20:00:00Z" },
+    ],
+  };
+
+  test("keeps rewatches with distinct timestamps", async () => {
+    const userId = insertUser();
+    insertMovieTitle("movie-1", 550, "Fight Club");
+
+    const jobId = createJob(userId, twoPlays);
+    await processImportJob(jobId);
+
+    const job = readImportJob(jobId);
+    expect(job.importedCount).toBe(2);
+    expect(movieWatches(userId)).toHaveLength(2);
+  });
+
+  test("skips a play already recorded at the same time", async () => {
+    const userId = insertUser();
+    insertMovieTitle("movie-1", 550, "Fight Club");
+    insertMovieWatch(userId, "movie-1", new Date("2024-01-01T20:30:00Z"));
+
+    const payload: NormalizedImport = {
+      ...emptyPayload("trakt"),
+      movies: [{ tmdbId: 550, title: "Fight Club", watchedAt: "2024-01-01T20:00:00Z" }],
+    };
+    const jobId = createJob(userId, payload);
+    await processImportJob(jobId);
+
+    const job = readImportJob(jobId);
+    expect(job.skippedCount).toBe(1);
+    expect(job.importedCount).toBe(0);
+    expect(movieWatches(userId)).toHaveLength(1);
+  });
+
+  test("re-running the same import is idempotent", async () => {
+    const userId = insertUser();
+    insertMovieTitle("movie-1", 550, "Fight Club");
+
+    const firstJobId = createJob(userId, twoPlays);
+    await processImportJob(firstJobId);
+    expect(readImportJob(firstJobId).importedCount).toBe(2);
+
+    const secondJobId = createJob(userId, twoPlays);
+    await processImportJob(secondJobId);
+
+    const second = readImportJob(secondJobId);
+    expect(second.skippedCount).toBe(2);
+    expect(second.importedCount).toBe(0);
+    expect(movieWatches(userId)).toHaveLength(2);
+  });
+
+  test("keeps episode rewatches", async () => {
+    const userId = insertUser();
+    insertTvShowWithFetchedAt("tv-1", 1396);
+
+    const payload: NormalizedImport = {
+      ...emptyPayload("trakt"),
+      episodes: [
+        {
+          showTmdbId: 1396,
+          showTitle: "Test Show",
+          seasonNumber: 1,
+          episodeNumber: 1,
+          watchedAt: "2024-01-01T20:00:00Z",
+        },
+        {
+          showTmdbId: 1396,
+          showTitle: "Test Show",
+          seasonNumber: 1,
+          episodeNumber: 1,
+          watchedAt: "2024-02-01T20:00:00Z",
+        },
+      ],
+    };
+    const jobId = createJob(userId, payload);
+    await processImportJob(jobId);
+
+    expect(readImportJob(jobId).importedCount).toBe(2);
+    const watches = testDb
+      .select()
+      .from(userEpisodeWatches)
+      .where(eq(userEpisodeWatches.userId, userId))
+      .all();
+    expect(watches).toHaveLength(2);
+  });
+
+  test("date-only diary entries on different dates are both kept", async () => {
+    const userId = insertUser();
+    insertMovieTitle("movie-1", 550, "Fight Club");
+
+    const payload: NormalizedImport = {
+      ...emptyPayload("letterboxd"),
+      movies: [
+        { tmdbId: 550, title: "Fight Club", watchedOn: "2024-01-01" },
+        { tmdbId: 550, title: "Fight Club", watchedOn: "2024-01-10" },
+      ],
+    };
+    const jobId = createJob(userId, payload);
+    await processImportJob(jobId);
+
+    expect(readImportJob(jobId).importedCount).toBe(2);
+    expect(movieWatches(userId)).toHaveLength(2);
+  });
+
+  test("library item backdates addedAt after a watch created the row", async () => {
+    const userId = insertUser();
+    insertMovieTitle("movie-1", 550, "Fight Club");
+
+    const payload: NormalizedImport = {
+      ...emptyPayload("trakt"),
+      movies: [{ tmdbId: 550, title: "Fight Club", watchedAt: "2024-06-01T20:00:00Z" }],
+      watchlist: [
+        {
+          tmdbId: 550,
+          title: "Fight Club",
+          type: "movie",
+          status: "completed",
+          addedAt: "2023-01-01T00:00:00Z",
+        },
+      ],
+    };
+    const jobId = createJob(userId, payload);
+    await processImportJob(jobId);
+
+    const row = testDb
+      .select()
+      .from(userTitleStatus)
+      .where(eq(userTitleStatus.userId, userId))
+      .get();
+    expect(row).toBeDefined();
+    expect(row?.addedAt).toEqual(new Date("2023-01-01T00:00:00Z"));
+  });
+
+  test("cancel during the last item is not overwritten by success", async () => {
+    const userId = insertUser();
+    insertMovieTitle("movie-1", 550, "Fight Club");
+
+    const payload: NormalizedImport = {
+      ...emptyPayload("letterboxd"),
+      movies: [{ imdbId: "tt0137523", title: "Fight Club" }],
+    };
+
+    let jobId = "";
+    const findSpy = vi.spyOn(tmdbClient, "findByExternalId").mockImplementation(async () => {
+      testDb.update(importJobs).set({ status: "cancelled" }).where(eq(importJobs.id, jobId)).run();
+      return { movie_results: [{ id: 550 }] } as never;
+    });
+
+    try {
+      jobId = createJob(userId, payload);
+      await processImportJob(jobId);
+
+      expect(findSpy).toHaveBeenCalled();
+      const job = testDb.select().from(importJobs).where(eq(importJobs.id, jobId)).get();
+      expect(job?.status).toBe("cancelled");
+      expect(job?.finishedAt).not.toBeNull();
+    } finally {
+      findSpy.mockRestore();
+    }
+  });
+});

packages/db/src/queries/imports.ts

@@ -1,4 +1,4 @@
-import { and, eq, inArray } from "drizzle-orm";
+import { and, eq, gt, gte, inArray, lte } from "drizzle-orm";
 
 import { db } from "../client";
 import {
@@ -29,6 +29,62 @@ export function hasEpisodeWatch(userId: string, episodeId: string): boolean {
   return !!existing;
 }
 
+export function hasMovieWatchBetween(
+  userId: string,
+  titleId: string,
+  from: Date,
+  to: Date,
+): boolean {
+  const existing = db
+    .select({ id: userMovieWatches.id })
+    .from(userMovieWatches)
+    .where(
+      and(
+        eq(userMovieWatches.userId, userId),
+        eq(userMovieWatches.titleId, titleId),
+        gte(userMovieWatches.watchedAt, from),
+        lte(userMovieWatches.watchedAt, to),
+      ),
+    )
+    .get();
+  return !!existing;
+}
+
+export function hasEpisodeWatchBetween(
+  userId: string,
+  episodeId: string,
+  from: Date,
+  to: Date,
+): boolean {
+  const existing = db
+    .select({ id: userEpisodeWatches.id })
+    .from(userEpisodeWatches)
+    .where(
+      and(
+        eq(userEpisodeWatches.userId, userId),
+        eq(userEpisodeWatches.episodeId, episodeId),
+        gte(userEpisodeWatches.watchedAt, from),
+        lte(userEpisodeWatches.watchedAt, to),
+      ),
+    )
+    .get();
+  return !!existing;
+}
+
+/** Move a library row's addedAt earlier (never later). Used by imports. */
+export function backdateTitleStatusAddedAt(userId: string, titleId: string, addedAt: Date): void {
+  db.update(userTitleStatus)
+    .set({ addedAt })
+    .where(
+      and(
+        eq(userTitleStatus.userId, userId),
+        eq(userTitleStatus.titleId, titleId),
+        gt(userTitleStatus.addedAt, addedAt),
+      ),
+    )
+    .run();
+}
+
 export function hasTitleStatus(userId: string, titleId: string): boolean {
   return !!getTitleStatusValue(userId, titleId);
 }