Merge pull request #174 from FoxxMD/issue-173-scrobbler-improvement

Enable more agressive scrobble checking behavior

src/backend/common/infrastructure/config/client/index.ts

@@ -37,6 +37,16 @@ export interface CommonClientOptions extends RequestRetryOptions {
      * @examples [true]
      * */
     refreshEnabled?: boolean
+    /**
+     * Force client to refresh scrobbled plays from upstream service if last refresh was at least X seconds ago
+     *
+     * **In most case this setting should NOT be used.** MS intelligently refreshes based on activity so using this setting may increase upstream service load and slow down scrobbles.
+     *
+     * This setting should only be used in specific scenarios where MS is handling multiple "relaying" client-services (IE lfm -> lz -> lfm) and there is the potential for a client to be out of sync after more than a few seconds.
+     *
+     * @examples [3]
+     * */
+    refreshStaleAfter?: number
 
     /**
      * The number of tracks to retrieve on initial refresh (related to scrobbleBacklogCount). If not specified this is the maximum supported for the client.

src/backend/common/schema/aio-client.json

@@ -57,6 +57,15 @@
                     "title": "refreshEnabled",
                     "type": "boolean"
                 },
+                "refreshForce": {
+                    "default": false,
+                    "description": "Force client to always refresh scrobbled plays from service before scrobbling new play\n\nWARNING: This will cause increased load on the scrobble service and potentially slow down scrobble speed as well. This should be used as a debugging tool and not be always-on.",
+                    "examples": [
+                        false
+                    ],
+                    "title": "refreshForce",
+                    "type": "boolean"
+                },
                 "refreshInitialCount": {
                     "description": "The number of tracks to retrieve on initial refresh (related to scrobbleBacklogCount). If not specified this is the maximum supported for the client.",
                     "title": "refreshInitialCount",

src/backend/common/schema/aio.json

@@ -346,6 +346,15 @@
                     "title": "refreshEnabled",
                     "type": "boolean"
                 },
+                "refreshForce": {
+                    "default": false,
+                    "description": "Force client to always refresh scrobbled plays from service before scrobbling new play\n\nWARNING: This will cause increased load on the scrobble service and potentially slow down scrobble speed as well. This should be used as a debugging tool and not be always-on.",
+                    "examples": [
+                        false
+                    ],
+                    "title": "refreshForce",
+                    "type": "boolean"
+                },
                 "refreshInitialCount": {
                     "description": "The number of tracks to retrieve on initial refresh (related to scrobbleBacklogCount). If not specified this is the maximum supported for the client.",
                     "title": "refreshInitialCount",
@@ -417,6 +426,15 @@
                     "title": "refreshEnabled",
                     "type": "boolean"
                 },
+                "refreshForce": {
+                    "default": false,
+                    "description": "Force client to always refresh scrobbled plays from service before scrobbling new play\n\nWARNING: This will cause increased load on the scrobble service and potentially slow down scrobble speed as well. This should be used as a debugging tool and not be always-on.",
+                    "examples": [
+                        false
+                    ],
+                    "title": "refreshForce",
+                    "type": "boolean"
+                },
                 "refreshInitialCount": {
                     "description": "The number of tracks to retrieve on initial refresh (related to scrobbleBacklogCount). If not specified this is the maximum supported for the client.",
                     "title": "refreshInitialCount",

src/backend/common/schema/client.json

@@ -54,6 +54,15 @@
                     "title": "refreshEnabled",
                     "type": "boolean"
                 },
+                "refreshForce": {
+                    "default": false,
+                    "description": "Force client to always refresh scrobbled plays from service before scrobbling new play\n\nWARNING: This will cause increased load on the scrobble service and potentially slow down scrobble speed as well. This should be used as a debugging tool and not be always-on.",
+                    "examples": [
+                        false
+                    ],
+                    "title": "refreshForce",
+                    "type": "boolean"
+                },
                 "refreshInitialCount": {
                     "description": "The number of tracks to retrieve on initial refresh (related to scrobbleBacklogCount). If not specified this is the maximum supported for the client.",
                     "title": "refreshInitialCount",

src/backend/common/vendor/ListenbrainzApiClient.ts

@@ -1,6 +1,6 @@
 import { stringSameness } from '@foxxmd/string-sameness';
 import dayjs from "dayjs";
-import request, { Request } from 'superagent';
+import request, { Request, Response } from 'superagent';
 import { PlayObject } from "../../../core/Atomic.js";
 import { slice } from "../../../core/StringUtils.js";
 import { combinePartsToString } from "../../utils.js";
@@ -127,7 +127,7 @@ export class ListenbrainzApiClient extends AbstractApiClient {
     }
 
 
-    callApi = async <T>(req: Request, retries = 0): Promise<T> => {
+    callApi = async <T = Response>(req: Request, retries = 0): Promise<T> => {
         const {
             maxRequestRetries = 2,
             retryMultiplier = DEFAULT_RETRY_MULTIPLIER
@@ -248,17 +248,27 @@ export class ListenbrainzApiClient extends AbstractApiClient {
         }
     }
 
-    submitListen = async (play: PlayObject) => {
+    submitListen = async (play: PlayObject, log: boolean = false) => {
         try {
             const listenPayload: SubmitPayload = {listen_type: 'single', payload: [ListenbrainzApiClient.playToListenPayload(play)]};
-            await this.callApi(request.post(`${this.url}1/submit-listens`).type('json').send(listenPayload));
+            if(log) {
+                this.logger.debug(`Submit Payload: ${JSON.stringify(listenPayload)}`);
+            }
+            // response consists of {"status": "ok"}
+            // so no useful information
+            // https://listenbrainz.readthedocs.io/en/latest/users/api-usage.html#submitting-listens
+            // TODO may we should make a call to recent-listens to get the parsed scrobble?
+            const resp = await this.callApi(request.post(`${this.url}1/submit-listens`).type('json').send(listenPayload));
+            if(log) {
+                this.logger.debug(`Submit Response: ${resp.text}`)
+            }
             return listenPayload;
         } catch (e) {
             throw e;
         }
     }
 
-    static playToListenPayload = (play: PlayObject): ListenPayload => {
+    static playToListenPayload(play: PlayObject): ListenPayload {
         const {
             data: {
                 playDate,
@@ -581,6 +591,21 @@ export class ListenbrainzApiClient extends AbstractApiClient {
         }
     }
 
+    static submitToPlayObj(submitObj: SubmitPayload, playObj: PlayObject): PlayObject {
+        if (submitObj.payload.length > 0) {
+            const respPlay = {
+                ...playObj,
+            };
+            respPlay.data = {
+                ...playObj.data,
+                album: submitObj.payload[0].track_metadata?.release_name ?? playObj.data.album,
+                track: submitObj.payload[0].track_metadata?.track_name ?? playObj.data.album,
+            };
+            return respPlay;
+        }
+        return playObj;
+    }
+
     static formatPlayObj(obj: any, options: FormatPlayObjectOptions): PlayObject {
         return ListenbrainzApiClient.listenResponseToPlay(obj);
     }

src/backend/scrobblers/AbstractScrobbleClient.ts

@@ -148,8 +148,60 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
         await this.refreshScrobbles(initialLimit);
     }
 
-    refreshScrobbles = async (limit?: number) => {
-        this.logger.debug('Scrobbler does not have refresh function implemented!');
+    refreshScrobbles = async (limit: number = this.MAX_STORED_SCROBBLES) => {
+        if (this.refreshEnabled) {
+            this.logger.debug('Refreshing recent scrobbles');
+            const recent = await this.getScrobblesForRefresh(limit);
+            this.logger.debug(`Found ${recent.length} recent scrobbles`);
+            if (this.recentScrobbles.length > 0) {
+                const [{data: {playDate: newestScrobbleTime = dayjs()} = {}} = {}] = this.recentScrobbles.slice(-1);
+                const [{data: {playDate: oldestScrobbleTime = dayjs()} = {}} = {}] = this.recentScrobbles.slice(0, 1);
+                this.newestScrobbleTime = newestScrobbleTime;
+                this.oldestScrobbleTime = oldestScrobbleTime;
+
+                this.filterScrobbledTracks();
+            }
+        }
+        this.lastScrobbleCheck = dayjs();
+    }
+
+    protected abstract getScrobblesForRefresh(limit: number): Promise<PlayObject[]>;
+
+    shouldRefreshScrobble = () => {
+        const {
+            refreshStaleAfter
+        } = this.config.options || {};
+
+        if (!this.refreshEnabled) {
+            this.logger.debug(`Should NOT refresh scrobbles => refreshEnabled is false`);
+            return false;
+        }
+
+        const queuedPlayedDate = this.getLatestQueuePlayDate();
+
+        // if next queued play was played more recently than the last time we refreshed upstream scrobbles
+        if (this.lastScrobbleCheck.unix() < queuedPlayedDate.unix()) {
+            this.logger.debug('Should refresh scrobbles => queued scrobble playDate is newer than last upstream scrobble refresh');
+            return true;
+        }
+
+        // if the last scrobbled play is at or is newer than the next scrobble then we are inserting (or potentially duping)
+        // in which case our data is probably stale
+        if(this.newestScrobbleTime !== undefined && this.newestScrobbleTime.unix() >= queuedPlayedDate.unix()) {
+            this.logger.debug('Should refresh scrobbles => queued scrobble playDate is equal to or older than the newest upstream scrobble');
+            return true;
+        }
+
+        if(refreshStaleAfter !== undefined) {
+            const diff = dayjs().diff(this.lastScrobbleCheck, 's');
+            if(diff > refreshStaleAfter) {
+                this.logger.debug(`Should refresh scrobbles => last refresh (${diff}s ago) was longer than refreshStaleAfter (${refreshStaleAfter}s)`);
+                return true;
+            }
+        }
+
+        this.logger.debug('Scrobble refresh not needed');
+        return false;
     }
 
     public abstract alreadyScrobbled(playObj: PlayObject, log?: boolean): Promise<boolean>;
@@ -525,7 +577,7 @@ ${closestMatch.breakdowns.join('\n')}`, {leaf: ['Dupe Check']});
             this.scrobbling = true;
             while (!this.shouldStopScrobbleProcessing()) {
                 while (this.queuedScrobbles.length > 0) {
-                    if (this.lastScrobbleCheck.unix() < this.getLatestQueuePlayDate().unix()) {
+                    if (this.shouldRefreshScrobble()) {
                         await this.refreshScrobbles();
                     }
                     const currQueuedPlay = this.queuedScrobbles.shift();

src/backend/scrobblers/LastfmScrobbler.ts

@@ -46,9 +46,7 @@ export default class LastfmScrobbler extends AbstractScrobbleClient {
         }
     }
 
-    refreshScrobbles = async (limit = this.MAX_STORED_SCROBBLES) => {
-        if (this.refreshEnabled) {
-            this.logger.debug('Refreshing recent scrobbles');
+    getScrobblesForRefresh = async (limit: number) => {
             const resp = await this.api.callApi<UserGetRecentTracksResponse>((client: any) => client.userGetRecentTracks({
                 user: this.api.user,
                 sk: this.api.client.sessionKey,
@@ -60,7 +58,7 @@ export default class LastfmScrobbler extends AbstractScrobbleClient {
                     track: list = [],
                 }
             } = resp;
-            this.recentScrobbles = list.reduce((acc: any, x: any) => {
+            return list.reduce((acc: any, x: any) => {
                 try {
                     const formatted = LastfmApiClient.formatPlayObj(x);
                     const {
@@ -91,17 +89,6 @@ export default class LastfmScrobbler extends AbstractScrobbleClient {
                     return acc;
                 }
             }, []);
-            this.logger.debug(`Found ${this.recentScrobbles.length} recent scrobbles`);
-            if (this.recentScrobbles.length > 0) {
-                const [{data: {playDate: newestScrobbleTime = dayjs()} = {}} = {}] = this.recentScrobbles.slice(-1);
-                const [{data: {playDate: oldestScrobbleTime = dayjs()} = {}} = {}] = this.recentScrobbles.slice(0, 1);
-                this.newestScrobbleTime = newestScrobbleTime;
-                this.oldestScrobbleTime = oldestScrobbleTime;
-
-                this.filterScrobbledTracks();
-            }
-        }
-        this.lastScrobbleCheck = dayjs();
     }
 
     cleanSourceSearchTitle = (playObj: PlayObject) => {

src/backend/scrobblers/ListenbrainzScrobbler.ts

@@ -59,22 +59,8 @@ export default class ListenbrainzScrobbler extends AbstractScrobbleClient {
         }
     }
 
-    refreshScrobbles = async (limit = this.MAX_STORED_SCROBBLES) => {
-        if (this.refreshEnabled) {
-            this.logger.debug('Refreshing recent scrobbles');
-            const resp = await this.api.getRecentlyPlayed(limit);
-            this.logger.debug(`Found ${resp.length} recent scrobbles`);
-            this.recentScrobbles = resp;
-            if (this.recentScrobbles.length > 0) {
-                const [{data: {playDate: newestScrobbleTime = dayjs()} = {}} = {}] = this.recentScrobbles.slice(-1);
-                const [{data: {playDate: oldestScrobbleTime = dayjs()} = {}} = {}] = this.recentScrobbles.slice(0, 1);
-                this.newestScrobbleTime = newestScrobbleTime;
-                this.oldestScrobbleTime = oldestScrobbleTime;
-
-                this.filterScrobbledTracks();
-            }
-        }
-        this.lastScrobbleCheck = dayjs();
+    getScrobblesForRefresh = async (limit: number) => {
+        return await this.api.getRecentlyPlayed(limit);
     }
 
     alreadyScrobbled = async (playObj: PlayObject, log = false) => (await this.existingScrobble(playObj)) !== undefined
@@ -91,30 +77,18 @@ export default class ListenbrainzScrobbler extends AbstractScrobbleClient {
             } = {}
         } = playObj;
 
-        let rawPayload = {listen_type: 'single', payload: [this.playToClientPayload(playObj)]};
-
         try {
-            const resp = await this.api.submitListen(playObj);
-            rawPayload = resp;
+            await this.api.submitListen(playObj, true);
 
             if (newFromSource) {
                 this.logger.info(`Scrobbled (New)     => (${source}) ${buildTrackString(playObj)}`);
             } else {
                 this.logger.info(`Scrobbled (Backlog) => (${source}) ${buildTrackString(playObj)}`);
             }
-            // last fm has rate limits but i can't find a specific example of what that limit is. going to default to 1 scrobble/sec to be safe
-            //await sleep(1000);
             return playObj;
         } catch (e) {
             await this.notifier.notify({title: `Client - ${capitalize(this.type)} - ${this.name} - Scrobble Error`, message: `Failed to scrobble => ${buildTrackString(playObj)} | Error: ${e.message}`, priority: 'error'});
-            this.logger.error(`Failed to scrobble => ${e.message}`, {payload: rawPayload});
-            if(e instanceof UpstreamError) {
-                throw e;
-            } else {
-                throw new UpstreamError(`Error occurred while making Listenbrainz API request: ${e.message}`, {cause: e, showStopper: true});
-            }
-        } finally {
-            this.logger.debug(`Raw Payload:`, {rawPayload});
+            throw new UpstreamError(`Error occurred while making Listenbrainz API scrobble request: ${e.message}`, {cause: e, showStopper: !(e instanceof UpstreamError)});
         }
     }
 }

src/backend/scrobblers/MalojaScrobbler.ts

@@ -299,28 +299,15 @@ export default class MalojaScrobbler extends AbstractScrobbleClient {
         }
     }
 
-    refreshScrobbles = async (limit = this.MAX_STORED_SCROBBLES) => {
-        if (this.refreshEnabled) {
-            this.logger.debug('Refreshing recent scrobbles');
-            const {url} = this.config.data;
-            const resp = await this.callApi(request.get(`${url}/apis/mlj_1/scrobbles?perpage=${limit}`));
-            const {
-                body: {
-                    list = [],
-                } = {},
-            } = resp;
-            this.logger.debug(`Found ${list.length} recent scrobbles`);
-            this.recentScrobbles = list.map((x: any) => this.formatPlayObj(x));
-            if (this.recentScrobbles.length > 0) {
-                const [{data: {playDate: newestScrobbleTime = dayjs()} = {}} = {}] = this.recentScrobbles.slice(-1);
-                const [{data: {playDate: oldestScrobbleTime = dayjs()} = {}} = {}] = this.recentScrobbles.slice(0, 1);
-                this.newestScrobbleTime = newestScrobbleTime;
-                this.oldestScrobbleTime = oldestScrobbleTime;
-
-                this.filterScrobbledTracks();
-            }
-        }
-        this.lastScrobbleCheck = dayjs();
+    getScrobblesForRefresh = async (limit: number) => {
+        const {url} = this.config.data;
+        const resp = await this.callApi(request.get(`${url}/apis/mlj_1/scrobbles?perpage=${limit}`));
+        const {
+            body: {
+                list = [],
+            } = {},
+        } = resp;
+        return list.map((x: any) => this.formatPlayObj(x));
     }
 
     cleanSourceSearchTitle = (playObj: PlayObject) => {

src/backend/tests/scrobbler/TestScrobbler.ts

@@ -6,6 +6,9 @@ import { Notifiers } from "../../notifier/Notifiers.js";
 import AbstractScrobbleClient from "../../scrobblers/AbstractScrobbleClient.js";
 
 export class TestScrobbler extends AbstractScrobbleClient {
+    protected async getScrobblesForRefresh(limit: number): Promise<PlayObject[]> {
+        return [];
+    }
 
     constructor() {
         const logger = loggerTest;

src/backend/tests/scrobbler/scrobblers.test.ts

@@ -425,14 +425,74 @@ describe('Detects duplicate and unique scrobbles using actively tracked scrobble
     });
 });
 
+describe('Detects when upstream scrobbles should be refreshed', function() {
+
+    const normalizedClose = normalizePlays(withDurPlays, {initialDate: dayjs().subtract(100, 'seconds')});
+
+    beforeEach(function () {
+        testScrobbler.recentScrobbles = normalizedWithMixedDur;
+        testScrobbler.newestScrobbleTime = normalizedWithMixedDur[0].data.playDate;
+        testScrobbler.lastScrobbleCheck = dayjs().subtract(60, 'seconds');
+        testScrobbler.queuedScrobbles = [];
+        testScrobbler.config.options = {};
+    });
+
+    it('Detects queued scrobble date is newer than last scrobble refresh', async function() {
+        const newScrobble = generatePlay({
+            playDate: dayjs()
+        });
+
+        testScrobbler.queueScrobble(newScrobble, 'test');
+        assert.isTrue(testScrobbler.shouldRefreshScrobble());
+    });
+
+    it('Detects queued scrobble date is older than newest scrobble', async function() {
+        testScrobbler.recentScrobbles = normalizedClose;
+        testScrobbler.newestScrobbleTime = normalizedClose[0].data.playDate;
+
+        const newScrobble = generatePlay({
+            playDate: dayjs().subtract(120, 'seconds')
+        });
+
+        testScrobbler.queueScrobble(newScrobble, 'test');
+        assert.isTrue(testScrobbler.shouldRefreshScrobble());
+    });
+
+    it('Forces refresh if refreshStaleAfter is set', async function() {
+        testScrobbler.recentScrobbles = normalizedClose;
+        testScrobbler.newestScrobbleTime = normalizedClose[0].data.playDate;
+        testScrobbler.config.options = { refreshStaleAfter: 10 };
+
+        const newScrobble = generatePlay({
+            playDate: dayjs().subtract(80, 'seconds')
+        });
+
+        testScrobbler.queueScrobble(newScrobble, 'test');
+        assert.isTrue(testScrobbler.shouldRefreshScrobble());
+    });
+
+    it('Does not refresh if scrobble is older than last check but newer than newest upstream scrobble', async function() {
+        testScrobbler.recentScrobbles = normalizedClose;
+        testScrobbler.newestScrobbleTime = normalizedClose[0].data.playDate;
+
+        const newScrobble = generatePlay({
+            playDate: dayjs().subtract(80, 'seconds')
+        });
+
+        testScrobbler.queueScrobble(newScrobble, 'test');
+        assert.isFalse(testScrobbler.shouldRefreshScrobble());
+    });
+});
+
 describe('Manages scrobble queue', function() {
 
-    before(function() {
+    before(async function() {
+        await testScrobbler.initialize();
         testScrobbler.recentScrobbles = normalizedWithMixedDur;
         testScrobbler.scrobbleSleep = 500;
         testScrobbler.scrobbleDelay = 0;
         testScrobbler.lastScrobbleCheck = dayjs().subtract(60, 'seconds');
-        testScrobbler.initScrobbleMonitoring();
+        testScrobbler.initScrobbleMonitoring().catch(console.error);
     });
 
     it('Scrobbles a uniquely queued play', async function() {