From 25981247724a2b2b9c3f945769d246ebcd64f9ca Mon Sep 17 00:00:00 2001 From: Jake Jarvis Date: Sun, 4 Oct 2026 15:27:15 -0400 Subject: [PATCH] fix(imports): keep rewatches and added dates; honor late cancellation Co-Authored-By: Claude Sonnet 5.5 --- packages/core/src/imports/processor.ts | 104 ++++++++--- packages/core/test/import-processor.test.ts | 180 ++++++++++++++++++++ packages/db/src/queries/imports.ts | 58 ++++++- 3 files changed, 315 insertions(+), 27 deletions(-) diff --git a/packages/core/src/imports/processor.ts b/packages/core/src/imports/processor.ts index fcf1a13..183a1ce 100644 --- a/packages/core/src/imports/processor.ts +++ b/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 { 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 { 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 { } } + // 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 { `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, { diff --git a/packages/core/test/import-processor.test.ts b/packages/core/test/import-processor.test.ts index a9f38b1..d2deb9b 100644 --- a/packages/core/test/import-processor.test.ts +++ b/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(); + } + }); +}); diff --git a/packages/db/src/queries/imports.ts b/packages/db/src/queries/imports.ts index 05c6f82..56384b1 100644 --- a/packages/db/src/queries/imports.ts +++ b/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); }