From 258cd61b4b12ffd80deb9a0521397ba2a2bff269 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Mon, 17 Aug 2026 14:32:20 +0000 Subject: [PATCH] feat(source): Improved queue prunning for backlog and history sources --- src/backend/sources/AbstractSource.ts | 35 +++++++++++-------- src/backend/sources/EndpointLastfmSource.ts | 4 +-- .../sources/EndpointListenbrainzSource.ts | 4 +-- src/backend/sources/WebScrobblerSource.ts | 2 +- src/core/Atomic.ts | 3 +- 5 files changed, 27 insertions(+), 21 deletions(-) diff --git a/src/backend/sources/AbstractSource.ts b/src/backend/sources/AbstractSource.ts index a7f688c0..80fcbf0c 100644 --- a/src/backend/sources/AbstractSource.ts +++ b/src/backend/sources/AbstractSource.ts @@ -378,23 +378,28 @@ export default abstract class AbstractSource extends AbstractComponent implement await pMap(playDatas, async (queueablePlay) => { try { - // we should be adding Plays to the queue without any transforms - // so run on "raw" play input - const cheapInputExisting = await this.playRepo.checkExisting(queueablePlay, { inputHash: queueablePlay }); - if (cheapInputExisting !== undefined) { - if(isDebugMode()) { - this.logger.trace(`Not adding ${buildTrackString(queueablePlay)} to queue because it already exists in db as Play ${cheapInputExisting.uid}`); - } + // for backlog and history plays we intentionally queue up plays that may have been processed before... + // ...for backlog we do this to catch any plays that may have been missed during network outage or MS offline or just the source reporting new things + // ...for history this is entirely how we "discover" new plays: we use MS's existing check logic to see find "new" plays on the same list (of history) that evolves over time + // + // for these two cases, for the majority of scenarios, we want to prune already processed plays from hitting the database + // otherwise we are causing a lot of noise for duped plays IE history polls every minute and we don't want all 100+ already seen plays being persisted as duped every minute. + if (([PARSED_FROM.history, PARSED_FROM.backlog] as PARSED_FROM_TYPE[]).includes(queueablePlay.meta.parsedFrom)) { + // we should be adding Plays to the queue without any transforms + // so run on "raw" play input // if we have seen a play with close temporality with the exact input hash then skip it entirely - // - // this is usually the case for 'history' based plays where we are brute-force adding all plays from an api call - // and we need to prune all these duplicates - // - // but lets log if this happens and *not* history or backlog... - if(!([PARSED_FROM.history, PARSED_FROM.backlog] as PARSED_FROM_TYPE[]).includes(queueablePlay.meta.parsedFrom)) { - this.logger.warn(`Play (${buildTrackString(queueablePlay)}) dropped pre-queue due to existing (${cheapInputExisting.uid}) was not from history/backlog...`); + const cheapInputExisting = await this.playRepo.checkExisting(queueablePlay, { inputHash: queueablePlay }); + if (cheapInputExisting !== undefined) { + if (isDebugMode() || PARSED_FROM.backlog === queueablePlay.meta.parsedFrom) { + // log to trace for backlog for some visibility into what was pruned + // this is fine noise-wise since this only happens when a component it (re)started + // + // for history we only want to do this if debugmode is enabled + // TODO implement debugmode per component so global debug doesn't cause noise if this isn't the component that is being debugged + this.logger.trace(`Not adding ${buildTrackString(queueablePlay)} to queue because it already exists in db as Play ${cheapInputExisting.uid}`); + } + return; } - return; } } catch (e) { this.logger.warn(new SimpleError('Failed to check queued scrobble for existing before adding, will continue with adding anyway', { cause: e })); diff --git a/src/backend/sources/EndpointLastfmSource.ts b/src/backend/sources/EndpointLastfmSource.ts index 6d775123..e21132c3 100644 --- a/src/backend/sources/EndpointLastfmSource.ts +++ b/src/backend/sources/EndpointLastfmSource.ts @@ -1,6 +1,6 @@ import dayjs from "dayjs"; import type EventEmitter from "events"; -import { type PlayObject, SOURCE_SOT } from "../../core/Atomic.ts"; +import { PARSED_FROM, type PlayObject, SOURCE_SOT } from "../../core/Atomic.ts"; import { type ExpressRequest, type FormatPlayObjectOptions, @@ -78,7 +78,7 @@ export class EndpointLastfmSource extends MemorySource { } const discoverable = stateData.filter(x => x.play.meta.nowPlaying === false); - await this.queuePlay(discoverable.map(x => x.play)); + await this.queuePlay(discoverable.map(x => ({...x.play, meta: {...x.play.meta, parsedFrom: PARSED_FROM.ingress}}))); // const discovered = await this.discover(discoverable.map(x => x.play)); // if (discovered.length > 0) { // await this.scrobble(discovered); diff --git a/src/backend/sources/EndpointListenbrainzSource.ts b/src/backend/sources/EndpointListenbrainzSource.ts index 37a55688..8d709eb0 100644 --- a/src/backend/sources/EndpointListenbrainzSource.ts +++ b/src/backend/sources/EndpointListenbrainzSource.ts @@ -1,6 +1,6 @@ import dayjs from "dayjs"; import type EventEmitter from "events"; -import { type PlayObject, SOURCE_SOT } from "../../core/Atomic.ts"; +import { PARSED_FROM, type PlayObject, SOURCE_SOT } from "../../core/Atomic.ts"; import { type ExpressRequest, type FormatPlayObjectOptions, @@ -104,7 +104,7 @@ export class EndpointListenbrainzSource extends MemorySource { } const discoverable = stateData.filter(x => x.play.meta.nowPlaying === false && this.isValidScrobble(x.play)); - await this.queuePlay(discoverable.map(x => x.play)) + await this.queuePlay(discoverable.map(x => ({...x.play, meta: {...x.play.meta, parsedFrom: PARSED_FROM.ingress}}))) // const discovered = await this.discover(discoverable.map(x => x.play)); // if (discovered.length > 0) { // await this.scrobble(discovered); diff --git a/src/backend/sources/WebScrobblerSource.ts b/src/backend/sources/WebScrobblerSource.ts index c9450a20..78d46811 100644 --- a/src/backend/sources/WebScrobblerSource.ts +++ b/src/backend/sources/WebScrobblerSource.ts @@ -143,7 +143,7 @@ export class WebScrobblerSource extends MemorySource { }, meta: { trackId: uniqueID, - parsedFrom: PARSED_FROM.nowPlaying, + parsedFrom: PARSED_FROM.ingress, url: { web: trackUrl, origin: originUrl diff --git a/src/core/Atomic.ts b/src/core/Atomic.ts index 939f058b..f12d50e5 100644 --- a/src/core/Atomic.ts +++ b/src/core/Atomic.ts @@ -480,10 +480,11 @@ export const SOURCE_SOT = { } as const satisfies Record export const sourceSotTypes: SOURCE_SOT_TYPES[] = ['player','history','ingress']; -export type PARSED_FROM_TYPE = 'backlog' | 'now playing' | 'player' | 'history'; +export type PARSED_FROM_TYPE = 'backlog' | 'now playing' | 'player' | 'history' | 'ingress'; export const PARSED_FROM = { backlog : 'backlog', nowPlaying: 'now playing', + ingress: 'ingress', player: 'player', history: 'history' } as const satisfies Record -- 2.51.2