fix(transfer): Improve duplicate detection and error handling

zbrox · zbrox · commit 49d5d35cbad3 · 2025-11-02T15:43:14.000+01:00
diff --git a/src/backend/common/vendor/koito/KoitoApiClient.ts b/src/backend/common/vendor/koito/KoitoApiClient.ts
@@ -144,6 +144,37 @@ export class KoitoApiClient extends AbstractApiClient {
         }
     }
 
+    getUserListensWithPagination = async (options: {
+        count?: number;
+        minTs?: number;
+        maxTs?: number;
+    } = {}): Promise<ListensResponse> => {
+        const { count = 100, minTs, maxTs } = options;
+
+        try {
+            const query: any = { count };
+            if (minTs !== undefined) {
+                query.min_ts = minTs;
+            }
+            if (maxTs !== undefined) {
+                query.max_ts = maxTs;
+            }
+
+            const resp = await this.callApi(request
+                .get(`${joinedUrl(this.url.url, '/apis/listenbrainz/1/user', this.config.username, 'listens')}`)
+                .timeout({
+                    response: 15000,
+                    deadline: 30000
+                })
+                .query(query));
+
+            const { body: { payload } } = resp as any;
+            return payload as ListensResponse;
+        } catch (e) {
+            throw e;
+        }
+    }
+
     getRecentlyPlayed = async (maxTracks: number): Promise<PlayObject[]> => {
         try {
             const resp = await this.getUserListens(maxTracks);
diff --git a/src/backend/scrobblers/LastfmScrobbler.ts b/src/backend/scrobblers/LastfmScrobbler.ts
@@ -1,4 +1,5 @@
 import { Logger } from "@foxxmd/logging";
+import dayjs, { Dayjs } from "dayjs";
 import EventEmitter from "events";
 import { NowPlayingResponse, TrackScrobbleResponse, UserGetRecentTracksResponse } from "lastfm-node-client";
 import { PlayObject } from "../../core/Atomic.js";
@@ -96,6 +97,51 @@ export default class LastfmScrobbler extends AbstractScrobbleClient {
         }
     }
 
+    getScrobblesForTimeRange = async (fromDate?: Dayjs, toDate?: Dayjs, limit: number = 1000): Promise<PlayObject[]> => {
+        const allPlays: PlayObject[] = [];
+        let currentPage = 1;
+        const perPage = 200;
+
+        while (allPlays.length < limit) {
+            const resp = await this.api.getRecentTracksWithPagination({
+                page: currentPage,
+                limit: perPage,
+                from: fromDate?.unix(),
+                to: toDate?.unix(),
+            });
+
+            const {
+                recenttracks: {
+                    track: rawTracks = [],
+                    '@attr': pageInfo
+                } = {}
+            } = resp;
+
+            if (rawTracks.length === 0) {
+                break;
+            }
+
+            const plays = rawTracks
+                .filter(t => t.date !== undefined)
+                .map(t => LastfmApiClient.formatPlayObj(t))
+                .filter(p => p.data.playDate && p.data.playDate.isValid()); // Filter out plays with invalid dates
+
+            allPlays.push(...plays);
+
+            if (allPlays.length >= limit) {
+                break;
+            }
+
+            if (pageInfo && currentPage >= parseInt(pageInfo.totalPages, 10)) {
+                break;
+            }
+
+            currentPage++;
+        }
+
+        return allPlays.slice(0, limit);
+    }
+
     cleanSourceSearchTitle = (playObj: PlayObject) => {
         const {
             data: {
diff --git a/src/backend/scrobblers/ListenbrainzScrobbler.ts b/src/backend/scrobblers/ListenbrainzScrobbler.ts
@@ -1,4 +1,5 @@
 import { Logger } from "@foxxmd/logging";
+import dayjs, { Dayjs } from "dayjs";
 import EventEmitter from "events";
 import { PlayObject } from "../../core/Atomic.js";
 import { buildTrackString, capitalize } from "../../core/StringUtils.js";
@@ -67,6 +68,44 @@ export default class ListenbrainzScrobbler extends AbstractScrobbleClient {
         return await this.api.getRecentlyPlayed(limit);
     }
 
+    getScrobblesForTimeRange = async (fromDate?: Dayjs, toDate?: Dayjs, limit: number = 1000): Promise<PlayObject[]> => {
+        const allPlays: PlayObject[] = [];
+        let maxTs = toDate?.unix();
+        const minTs = fromDate?.unix();
+
+        while (allPlays.length < limit) {
+            const batchSize = Math.min(100, limit - allPlays.length);
+            const resp = await this.api.getUserListensWithPagination({
+                count: batchSize,
+                minTs,
+                maxTs,
+            });
+
+            if (!resp.listens || resp.listens.length === 0) {
+                break;
+            }
+
+            const plays = resp.listens
+                .map(l => ListenbrainzApiClient.formatPlayObj(l, {}))
+                .filter(p => p.data.playDate && p.data.playDate.isValid()); // Filter out plays with invalid dates
+
+            allPlays.push(...plays);
+
+            if (plays.length < batchSize) {
+                break;
+            }
+
+            const oldestPlay = plays[plays.length - 1];
+            if (oldestPlay.data.playDate) {
+                maxTs = oldestPlay.data.playDate.unix() - 1;
+            } else {
+                break;
+            }
+        }
+
+        return allPlays;
+    }
+
     alreadyScrobbled = async (playObj: PlayObject, log = false) => (await this.existingScrobble(playObj)) !== undefined
 
     public playToClientPayload(playObj: PlayObject): ListenPayload {
diff --git a/src/backend/transfer/TransferJob.ts b/src/backend/transfer/TransferJob.ts
@@ -197,21 +197,12 @@ export class TransferJob {
             } = resp;
 
             if (pageInfo && !totalPagesKnown) {
-                const totalPages = parseInt(pageInfo.totalPages, 10);
-                const total = parseInt(pageInfo.total, 10);
-
-                // If playCount is specified, cap the total at playCount
-                if (this.playCount) {
-                    this.progress.total = Math.min(this.playCount, total);
-                    // Calculate how many pages we'll actually need
-                    this.progress.totalPages = Math.ceil(this.progress.total / this.PAGE_SIZE);
-                } else {
-                    this.progress.total = total;
-                    this.progress.totalPages = totalPages;
-                }
+                const apiTotal = parseInt(pageInfo.total, 10);
+                this.progress.total = this.playCount ? Math.min(this.playCount, apiTotal) : apiTotal;
+                this.progress.totalPages = Math.ceil(this.progress.total / this.PAGE_SIZE);
 
                 totalPagesKnown = true;
-                this.logger.info(`Total pages in source: ${totalPages}, Total plays in source: ${total}, Will transfer: ${this.progress.total}, Expected pages: ${this.progress.totalPages}`);
+                this.logger.info(`Total plays in source: ${apiTotal}, Will transfer: ${this.progress.total}, Expected pages: ${this.progress.totalPages}`);
             }
 
             if (rawTracks.length === 0) {
@@ -313,6 +304,8 @@ export class TransferJob {
         await this.processPlaysWithSlidingWindow(sortedPlays, client);
     }
 
+    private timeRangeScrobbles: PlayObject[] = [];
+
     private async processPlaysWithSlidingWindow(plays: PlayObject[], client: AbstractScrobbleClient): Promise<void> {
         if (plays.length === 0) {
             return;
@@ -321,11 +314,33 @@ export class TransferJob {
         const oldest = plays[0].data.playDate;
         const newest = plays[plays.length - 1].data.playDate;
 
+        // Fetch scrobbles for the specific time range being processed
         if (oldest && newest) {
-            this.logger.debug(`Refreshing client scrobbles for time range: ${oldest.format()} to ${newest.format()}`);
-            await client.refreshScrobbles(this.SLIDING_WINDOW_SIZE);
+            this.logger.debug(`Fetching client scrobbles for time range: ${oldest.format()} to ${newest.format()}`);
+
+            // Check if client supports time-range fetching
+            if ('getScrobblesForTimeRange' in client && typeof (client as any).getScrobblesForTimeRange === 'function') {
+                try {
+                    // Add a buffer before/after to catch nearby scrobbles
+                    const bufferHours = 24;
+                    const fromDate = oldest.subtract(bufferHours, 'hours');
+                    const toDate = newest.add(bufferHours, 'hours');
+
+                    this.timeRangeScrobbles = await (client as any).getScrobblesForTimeRange(fromDate, toDate, this.SLIDING_WINDOW_SIZE);
+                    this.logger.debug(`Fetched ${this.timeRangeScrobbles.length} scrobbles from ${this.clientName} for duplicate detection`);
+                } catch (e) {
+                    this.logger.error(`Error fetching scrobbles for time range: ${e.message}`);
+                    throw e;
+                }
+            } else {
+                // Fallback to regular refresh (will only work for recent scrobbles)
+                this.logger.warn('Client does not support time-range fetching, falling back to regular refresh (may not detect duplicates for old scrobbles)');
+                await client.refreshScrobbles(this.SLIDING_WINDOW_SIZE);
+                this.timeRangeScrobbles = [];
+            }
         } else {
             await client.refreshScrobbles(this.SLIDING_WINDOW_SIZE);
+            this.timeRangeScrobbles = [];
         }
 
         for (let i = 0; i < plays.length; i++) {
@@ -358,6 +373,37 @@ export class TransferJob {
         this.progress.currentTrack = undefined;
     }
 
+    private isAlreadyScrobbledInTimeRange(play: PlayObject): boolean {
+        if (this.timeRangeScrobbles.length === 0) {
+            return false;
+        }
+
+        const playDate = play.data.playDate;
+        if (!playDate) {
+            return false;
+        }
+
+        // Check for matching scrobble (same track, artist, and similar timestamp)
+        return this.timeRangeScrobbles.some(scrobbled => {
+            const scrobbledDate = scrobbled.data.playDate;
+            if (!scrobbledDate) {
+                return false;
+            }
+
+            // Check if timestamps are within 30 seconds of each other
+            const timeDiffSeconds = Math.abs(playDate.diff(scrobbledDate, 'second'));
+            if (timeDiffSeconds > 30) {
+                return false;
+            }
+
+            // Check if track and artists match
+            const trackMatch = play.data.track?.toLowerCase() === scrobbled.data.track?.toLowerCase();
+            const artistsMatch = play.data.artists?.join(',').toLowerCase() === scrobbled.data.artists?.join(',').toLowerCase();
+
+            return trackMatch && artistsMatch;
+        });
+    }
+
     private async processPlay(play: PlayObject, client: AbstractScrobbleClient): Promise<void> {
         const transformedPlay = client.transformPlay(play, TRANSFORM_HOOK.preCompare);
 
@@ -366,9 +412,17 @@ export class TransferJob {
         // 2. timeFrameIsValid is designed to prevent re-scrobbling during normal operation
         // 3. We have our own duplicate detection via time-range fetching
 
+        // Check against our time-range-specific scrobbles (the actual duplicate detection)
+        if (this.isAlreadyScrobbledInTimeRange(transformedPlay)) {
+            this.logger.verbose(`DUPLICATE (time-range): ${buildTrackString(play)} - found in ${this.clientName}'s ${this.timeRangeScrobbles.length} scrobbles`);
+            this.progress.duplicates++;
+            return;
+        }
+
+        // Check using the client's built-in method (for recently added scrobbles from this transfer)
         const alreadyScrobbled = await client.alreadyScrobbled(transformedPlay);
         if (alreadyScrobbled) {
-            this.logger.debug(`Skipping ${buildTrackString(play)}: already scrobbled`);
+            this.logger.verbose(`DUPLICATE (recent): ${buildTrackString(play)} - found in ${this.clientName}'s recent scrobbles`);
             this.progress.duplicates++;
             return;
         }
diff --git a/src/backend/transfer/TransferManager.ts b/src/backend/transfer/TransferManager.ts
@@ -129,6 +129,39 @@ export class TransferManager {
             throw new Error('Play count must be greater than 0');
         }
 
+        // Validate date range if provided
+        if (fromDate || toDate) {
+            const now = new Date();
+
+            if (fromDate) {
+                const from = new Date(fromDate);
+                if (isNaN(from.getTime())) {
+                    throw new Error('Invalid fromDate format');
+                }
+                if (from > now) {
+                    throw new Error('fromDate cannot be in the future');
+                }
+            }
+
+            if (toDate) {
+                const to = new Date(toDate);
+                if (isNaN(to.getTime())) {
+                    throw new Error('Invalid toDate format');
+                }
+                if (to > now) {
+                    throw new Error('toDate cannot be in the future');
+                }
+            }
+
+            if (fromDate && toDate) {
+                const from = new Date(fromDate);
+                const to = new Date(toDate);
+                if (from > to) {
+                    throw new Error('fromDate must be before toDate');
+                }
+            }
+        }
+
         const source = this.scrobbleSources.getByName(sourceName);
         if (!source) {
             throw new Error(`Source '${sourceName}' not found`);
@@ -147,6 +180,11 @@ export class TransferManager {
             throw new Error(`Client '${clientName}' is not ready`);
         }
 
+        // Check if client supports time-range fetching (required for duplicate detection)
+        if (!('getScrobblesForTimeRange' in client) || typeof (client as any).getScrobblesForTimeRange !== 'function') {
+            throw new Error(`Client '${clientName}' does not support time-range fetching, which is required for accurate duplicate detection during transfers. Supported clients: Last.fm, Listenbrainz`);
+        }
+
         if (playCount !== undefined && playCount > 10000) {
             this.logger.warn(`Play count ${playCount} is very large. This may take a long time.`);
         }
diff --git a/src/backend/utils/StringUtils.ts b/src/backend/utils/StringUtils.ts
@@ -12,7 +12,7 @@ export const SYMBOLS_WHITESPACE_REGEX = new RegExp(/[`=(){}<>;'’,.~!@#$%^&*_+|
 export const SYMBOLS_REGEX = new RegExp(/[`=(){}<>;'’,.~!@#$%^&*_+|:"?\-\\[\]/]/g);
 
 export const MULTI_WHITESPACE_REGEX = new RegExp(/\s{2,}/g);
-export const uniqueNormalizedStrArr = (arr: string[]): string[] => arr.reduce((acc: string[], curr) => {
+export const uniqueNormalizedStrArr = (arr: string[]): string[] => arr.filter(x => x != null).reduce((acc: string[], curr) => {
         const normalizedCurr = normalizeStr(curr)
         if (!acc.some(x => normalizeStr(x) === normalizedCurr)) {
             return acc.concat(curr);
@@ -21,6 +21,9 @@ export const uniqueNormalizedStrArr = (arr: string[]): string[] => arr.reduce((a
     }, [])
 // https://stackoverflow.com/a/37511463/1469797
 export const normalizeStr = (str: string, options?: {keepSingleWhitespace?: boolean}): string => {
+    if (!str) {
+        return '';
+    }
     const {keepSingleWhitespace = false} = options || {};
     const normal = str.normalize('NFD').replace(/[\u0300-\u036f]/g, "");
     if(!keepSingleWhitespace) {
@@ -96,7 +99,7 @@ export const PRIMARY_SECONDARY_SECTIONS_REGEX = new RegExp(/^(?<primary>.+?)(?<s
  * */
 // export const SECONDARY_ARTISTS_REGEX = new RegExp(//ig);
 export const parseCredits = (str: string, delimiters?: boolean | string[]): PlayCredits => {
-    if (str.trim() === '') {
+    if (!str || str.trim() === '') {
         return undefined;
     }
 
@@ -136,7 +139,7 @@ export const parseCredits = (str: string, delimiters?: boolean | string[]): Play
     return undefined;
 }
 export const parseArtistCredits = (str: string, delimiters?: boolean | string[], ignoreGlobalAmpersand?: boolean): PlayCredits | undefined => {
-    if (str.trim() === '') {
+    if (!str || str.trim() === '') {
         return undefined;
     }
     let delims: string[] | undefined;
diff --git a/src/backend/utils/TimeUtils.ts b/src/backend/utils/TimeUtils.ts
@@ -34,10 +34,16 @@ dayjs.extend(isToday);
 export const temporalPlayComparisonSummary = (data: TemporalPlayComparison, existingPlay?: PlayObject, candidatePlay?: PlayObject) => {
     const parts: string[] = [];
     if (existingPlay !== undefined && candidatePlay !== undefined) {
-        if (existingPlay.data.playDate.isSame(candidatePlay.data.playDate, 'day')) {
-            parts.push(`Existing: ${existingPlay.data.playDate.format('HH:mm:ssZ')} - Candidate: ${candidatePlay.data.playDate.format('HH:mm:ssZ')}`);
+        const existingDate = existingPlay.data.playDate;
+        const candidateDate = candidatePlay.data.playDate;
+
+        // Check if dates are valid before comparing/formatting
+        if (!existingDate?.isValid() || !candidateDate?.isValid()) {
+            parts.push(`Existing: ${existingDate?.isValid() ? existingDate.toISOString() : 'Invalid Date'} - Candidate: ${candidateDate?.isValid() ? candidateDate.toISOString() : 'Invalid Date'}`);
+        } else if (existingDate.isSame(candidateDate, 'day')) {
+            parts.push(`Existing: ${existingDate.format('HH:mm:ssZ')} - Candidate: ${candidateDate.format('HH:mm:ssZ')}`);
         } else {
-            parts.push(`Existing: ${existingPlay.data.playDate.toISOString()} - Candidate: ${candidatePlay.data.playDate.toISOString()}`);
+            parts.push(`Existing: ${existingDate.toISOString()} - Candidate: ${candidateDate.toISOString()}`);
         }
     }
     parts.push(`Temporal Sameness: ${capitalize(temporalAccuracyToString(data.match))}`);
