From 09dcb22cb335f0b3384208f2a55128e3f0b40776 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Sat, 15 Aug 2026 00:40:20 +0000 Subject: [PATCH] finish source queue discovery migration --- .../database/appMigrations/002_inputHash.ts | 4 +- src/backend/sources/AbstractSource.ts | 127 +----------------- src/backend/sources/AppleMusicSource.ts | 8 +- src/backend/sources/YTMusicSource.ts | 3 +- src/backend/tests/source/TestSource.ts | 2 +- src/backend/tests/source/source.test.ts | 64 +++++---- src/backend/tests/ytm/ytm.test.ts | 15 ++- 7 files changed, 59 insertions(+), 164 deletions(-) diff --git a/src/backend/common/database/appMigrations/002_inputHash.ts b/src/backend/common/database/appMigrations/002_inputHash.ts index a4afe715..78ff54ff 100644 --- a/src/backend/common/database/appMigrations/002_inputHash.ts +++ b/src/backend/common/database/appMigrations/002_inputHash.ts @@ -50,9 +50,9 @@ export const up: Migration['up'] = async (db: SqliteDatabase const playsRows = await ctx.db.select().from(drizzlePlays).limit(100).offset(offset); for (const row of playsRows) { try { - await ctx.db.update(playInputs).set({ + await ctx.db.update(drizzlePlays).set({ playHash: hashObject(playContentBasicInvariantTransform(row.play).data) - }).where(eq(playInputs.id, row.id)); + }).where(eq(drizzlePlays.id, row.id)); updated++; processed++; } catch (e) { diff --git a/src/backend/sources/AbstractSource.ts b/src/backend/sources/AbstractSource.ts index 4d626bc0..2e660868 100644 --- a/src/backend/sources/AbstractSource.ts +++ b/src/backend/sources/AbstractSource.ts @@ -36,8 +36,8 @@ import { componentFileLogger } from '../common/logging.ts'; ; import { messageWithCausesTruncatedDefault } from "../../core/ErrorUtils.ts"; import { existingScrobble, type ExistingScrobbleOpts } from '../utils/PlayComparisonUtils.ts'; -import { consumeQueue, staggerMapper } from '../utils/AsyncUtils.ts'; -import pMap, {pMapIterable} from 'p-map'; +import { consumeQueue } from '../utils/AsyncUtils.ts'; +import pMap from 'p-map'; import type { Counter, Gauge } from 'prom-client'; import { normalizeStr } from '../utils/StringUtils.ts'; import { spawn, isAbortError, delay, throwIfAborted } from 'abort-controller-x'; @@ -47,9 +47,8 @@ import { asPlay } from '../../core/PlayMarshalUtils.ts'; import { AsyncTask, SimpleIntervalJob, ToadScheduler } from 'toad-scheduler'; import { COMPONENT_STATE, type ComponentSourceApiJson, type ComponentState, type PlayApiCommonDetailed } from '../../core/Api.ts'; import type {PaginatedResponse} from "../../core/Api.ts"; -import type { PlaySelect, PlaySelectWithQueueStates, PlayWith, QueueStateNew } from '../common/database/drizzle/drizzleTypes.ts'; +import type { PlaySelect, PlaySelectWithQueueStates, QueueStateNew } from '../common/database/drizzle/drizzleTypes.ts'; import { DrizzleQueueRepository } from '../common/database/drizzle/repositories/QueueRepository.ts'; -import { nanoid } from 'nanoid'; export interface RecentlyPlayedOptions { limit?: number @@ -106,11 +105,6 @@ export default abstract class AbstractSource extends AbstractComponent implement protected discoveredCounter: Counter; - protected staggerMappers = { - preCompare: staggerMapper({concurrency: 2}), - postCompare: staggerMapper({concurrency: 2}) - } - declare protected componentType: 'source'; protected playRepo!: DrizzlePlayRepository; @@ -284,7 +278,6 @@ export default abstract class AbstractSource extends AbstractComponent implement protected async postCache(): Promise { await super.postCache(); - this.generateStaggerMappers(); } protected async postDatabase(): Promise { @@ -306,35 +299,6 @@ export default abstract class AbstractSource extends AbstractComponent implement } } - protected generateStaggerMappers() { - const { - preCompare = [], - postCompare = [], - } = this.transformRules; - - if (preCompare.length > 0) { - const pcInits: number[] = [0], - pcMaxStagger: number[] = [0]; - for (const hook of this.transformRules.preCompare) { - const t = this.transformManager.getTransformerByStage({ type: hook.type, name: hook.name }); - pcInits.push(t.staggerOpts?.initialInterval ?? 0); - pcMaxStagger.push(t.staggerOpts?.maxRandomStagger ?? 0) - } - this.staggerMappers.preCompare = staggerMapper({ initialInterval: Math.max(...pcInits), maxRandomStagger: Math.max(...pcMaxStagger), concurrency: 2 }); - } - - if (postCompare.length > 0) { - const postInits: number[] = [0], - postMaxStagger: number[] = [0]; - for (const hook of this.transformRules.postCompare) { - const t = this.transformManager.getTransformerByStage({ type: hook.type, name: hook.name }); - postInits.push(t.staggerOpts?.initialInterval ?? 0); - postMaxStagger.push(t.staggerOpts?.maxRandomStagger ?? 0) - } - this.staggerMappers.postCompare = staggerMapper({ initialInterval: Math.max(...postInits), maxRandomStagger: Math.max(...postMaxStagger), concurrency: 2 }); - } - } - protected getIdentifier() { return `${capitalize(this.type)} - ${this.name}` } @@ -462,35 +426,6 @@ export default abstract class AbstractSource extends AbstractComponent implement return createdQueuedPlays; } - protected addPlayToDB = async (play: PlayObject): Promise> => { - const monitorStatus = this.getMonitoringStatus(); - let state: PlayState = 'discovered'; - if(!monitorStatus.monitoring) { - this.logger.debug(`Not adding ${buildTrackString(play)} as discovered because monitoring is disabled by ${capitalize(monitorStatus.origin)}`); - state = 'discarded'; - } - const playRow = await this.playRepo.createPlays([(playToRepositoryCreatePlayOpts({play, state}))]); - const recentPlays = await this.getRecentlyDiscoveredPlays(false); - // only need to update if its already in memory, - // and better to update in-memory than clear cache so we aren't refetching from db on every discover - if(recentPlays !== undefined) { - recentPlays.push({...play, id: playRow[0].id, uid: playRow[0].uid}); - recentPlays.sort(sortByOldestPlayDate); - this.cache.cacheDb.set(this.recentDiscoveredCacheKey(), recentPlays, '2m'); - } - if(state === 'discovered') { - this.tracksDiscovered++; - this.tracksDiscoveredTotal++ - this.discoveredCounter.labels(this.getPrometheusLabels()).inc(); - } - this.logger.info(`${capitalize(state)} => ${buildTrackString(play)}`); - this.emitEvent('discovered', {play}); - this.emitPlayInsert({...playRow[0], queueStates: []} as unknown as PlayApiCommonDetailed); - playRow[0].play.id = playRow[0].id; - playRow[0].play.uid = playRow[0].uid; - return playRow[0]; - } - getFlatRecentlyDiscoveredPlays = async (): Promise => { const list: PlayObject[] = await this.getRecentlyDiscoveredPlays(); return list.sort(sortByNewestPlayDate); @@ -535,45 +470,12 @@ export default abstract class AbstractSource extends AbstractComponent implement return undefined; } - discover = async (plays: PlayObject[], options: { checkAll?: boolean, signal?: AbortSignal, discoverLocation?: string, [key: string]: any } = {}): Promise => { - const newDiscoveredPlays: PlayObject[] = []; - - this.setStatus(`Discovering new Plays${options.discoverLocation !== undefined ? ` from ${options.discoverLocation} ` : ''}`); - for await(const play of pMapIterable(plays, this.staggerMappers.preCompare(async x => await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 3})) { - options.signal?.throwIfAborted(); - const existing = await this.existingDiscovered(play); - if(existing === undefined) { - options.signal?.throwIfAborted() - const hydratedPlay = await this.addPlayToDB(play); - if(hydratedPlay.state === 'discovered') { - newDiscoveredPlays.push(hydratedPlay.play); - } - } else { - this.playRepo.updateById(existing.id, {updatedAt: dayjs()}); - } - } - if(newDiscoveredPlays.length > 0) { - this.setStatus(`Discovered ${newDiscoveredPlays.length} new Plays${options.discoverLocation !== undefined ? ` from ${options.discoverLocation} ` : ''}`); - try { - await this.componentRepo.updateById(this.dbComponent.id, {countLive: this.dbComponent.countLive + newDiscoveredPlays.length}); - } catch (e) { - this.logger.warn(new Error('Unable to update discovered count', {cause: e})); - } - } else { - this.setStatus(`No new Plays discovered${options.discoverLocation !== undefined ? ` from ${options.discoverLocation} ` : ''}`); - } - newDiscoveredPlays.sort(sortByOldestPlayDate); - - return newDiscoveredPlays; - } - - protected scrobble = async (newDiscoveredPlays: PlayObject[], options: { forceRefresh?: boolean, [key: string]: any, discoverLocation?: 'backlog' | [key: string] } = {}) => { if(newDiscoveredPlays.length > 0) { newDiscoveredPlays.sort(sortByOldestPlayDate); this.emitEvent('discoveredToScrobble', { - data: await pMap(newDiscoveredPlays, this.staggerMappers.postCompare(async (x) => await this.transformPlay(x, TRANSFORM_HOOK.postCompare)), {concurrency: 3}), + data: await pMap(newDiscoveredPlays, async (x) => await this.transformPlay(x, TRANSFORM_HOOK.postCompare), {concurrency: 3}), options: { ...options, checkTime: newDiscoveredPlays[newDiscoveredPlays.length-1].data.playDate.add(2, 'second'), @@ -620,20 +522,6 @@ export default abstract class AbstractSource extends AbstractComponent implement } await this.queuePlay(backlogPlays); this.logger.info('Backlog Plays added to discovery queue.'); - //const discovered = await this.discover(backlogPlays, {discoverLocation: 'backlog', signal}); - - // if (scrobbleBacklog) { - // if (discovered.length > 0) { - // this.logger.info('Scrobbling backlogged tracks...'); - // signal.throwIfAborted(); - // await this.scrobble(discovered); - // this.logger.info('Backlog scrobbling complete.'); - // } else { - // this.logger.info('All tracks already discovered!'); - // } - // } else { - // this.logger.info('Backlog scrobbling is disabled by config, skipping...'); - // } } return; } @@ -838,8 +726,6 @@ export default abstract class AbstractSource extends AbstractComponent implement const maxBackoff = this.getMaxBackoff(); let sleepTime = interval; - let newDiscovered: PlayObject[] = []; - if(playObjs.length > 0) { const now = dayjs().unix(); const closeToInterval = playObjs.some(x => now - x.data.playDate.unix() < 5); @@ -867,8 +753,9 @@ export default abstract class AbstractSource extends AbstractComponent implement if(playObjs.length > 0) { playObjs.sort(sortByNewestPlayDate); // only update date if the play date is after the current activity date (in the case of backlogged plays) - this.lastActivityAt = playObjs[0].data.playDate.isAfter(this.lastActivityAt) ? newDiscovered[0].data.playDate : this.lastActivityAt; - checkCount = 0; + if(playObjs[0].data.playDate.isAfter(this.lastActivityAt)) { + this.lastActivityAt = playObjs[0].data.playDate; + } checksOverThreshold = 0; } diff --git a/src/backend/sources/AppleMusicSource.ts b/src/backend/sources/AppleMusicSource.ts index e66f4218..d01b5185 100644 --- a/src/backend/sources/AppleMusicSource.ts +++ b/src/backend/sources/AppleMusicSource.ts @@ -356,13 +356,7 @@ export default class AppleMusicSource extends AbstractSource { onPollPostAuthCheck = async () => { if(!this.polling) { this.logger.verbose('Hydrating initial recently played tracks for reference.'); - const referencePlays = await this.getRecentlyPlayed(); - const reversedPlays = [...referencePlays]; - reversedPlays.reverse(); - - for(const refPlay of reversedPlays) { - await this.addPlayToDB(refPlay); - } + await this.getRecentlyPlayed(); } return true; } diff --git a/src/backend/sources/YTMusicSource.ts b/src/backend/sources/YTMusicSource.ts index e21e9402..0216d549 100644 --- a/src/backend/sources/YTMusicSource.ts +++ b/src/backend/sources/YTMusicSource.ts @@ -695,8 +695,7 @@ ${humanDiff}`; if(this.transientDiscovered.data.length === 0) { // and add to discovered since its empty for(const refPlay of reversedPlays) { - //this.transientDiscovered.add(refPlay); - await this.addPlayToDB(refPlay); + this.transientDiscovered.add(refPlay); } } } diff --git a/src/backend/tests/source/TestSource.ts b/src/backend/tests/source/TestSource.ts index 8be9c4d7..ae2f2f11 100644 --- a/src/backend/tests/source/TestSource.ts +++ b/src/backend/tests/source/TestSource.ts @@ -4,7 +4,7 @@ import { MemoryPositionalSource } from "../../sources/MemoryPositionalSource.ts" import MemorySource from "../../sources/MemorySource.ts"; export class TestSource extends AbstractSource { - override queueIdleMs: number = 2; + override queueIdleMs: number = 1; override queueConcurrency: number = 1; handle(plays: PlayObject[]) { this.scrobble(plays); diff --git a/src/backend/tests/source/source.test.ts b/src/backend/tests/source/source.test.ts index 24290151..1a543dcf 100644 --- a/src/backend/tests/source/source.test.ts +++ b/src/backend/tests/source/source.test.ts @@ -482,7 +482,7 @@ describe('Deezer Internal Source', function() { const fuzzyPlay = clone(targetPlay); fuzzyPlay.data.playDate = targetPlay.data.playDate.add(targetPlay.data.duration, 's'); - const source = await generateDeezerSource(); + await using source = await generateDeezerSource(); const queued = await source.queuePlay([...normalizedPlays, interimPlay]); expect(queued).length(normalizedPlays.length + 1); await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); @@ -516,11 +516,12 @@ describe('Deezer Internal Source', function() { fuzzyPlay.data.playDate = targetPlay.data.playDate.add(targetPlay.data.duration, 's'); await using source = await generateDeezerSource({fuzzyDiscoveryIgnore: true}); - await source.discover(normalizedPlays); + const queued = await source.queuePlay(normalizedPlays); + expect(queued).length(normalizedPlays.length); - const discovered = await source.discover([fuzzyPlay]); - - expect(discovered.length).to.eq(1); + await source.queuePlay([fuzzyPlay]); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); + expect(await source.getRecentlyDiscoveredPlays()).length(normalizedPlays.length + 1); }); it('discovers fuzzy play when it is played consecutively', async function() { @@ -529,10 +530,11 @@ describe('Deezer Internal Source', function() { fuzzyPlay.data.playDate = targetPlay.data.playDate.add(targetPlay.data.duration, 's'); const morePlays = normalizePlays([...normalizedPlays, fuzzyPlay, ...generatePlays(2)], {initialDate: firstPlayDate}); - await using source = await generateDeezerSource({fuzzyDiscoveryIgnore: true}); - const discovered = await source.discover(morePlays); - - expect(discovered.length).to.eq(morePlays.length); + await using source = await generateDeezerSource({fuzzyDiscoveryIgnore: false}); + const queued = await source.queuePlay(morePlays); + expect(queued).length(morePlays.length); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); + expect(await source.getRecentlyDiscoveredPlays()).length(morePlays.length); }); }); @@ -545,11 +547,14 @@ describe('Deezer Internal Source', function() { fuzzyPlay.data.playDate = targetPlay.data.playDate.add(targetPlay.data.duration, 's'); await using source = await generateDeezerSource({fuzzyDiscoveryIgnore: 'aggressive'}); - await source.discover([...normalizedPlays, interimPlay]); + const queued = await source.queuePlay([...normalizedPlays, interimPlay]); + expect(queued).length(normalizedPlays.length + 1); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); - const discovered = await source.discover([fuzzyPlay]); + await source.queuePlay([fuzzyPlay]); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); - expect(discovered.length).to.eq(0); + expect(await source.getRecentlyDiscoveredPlays()).length(normalizedPlays.length + 1); }); it('does not discover play found during duration of previous', async function() { @@ -559,11 +564,13 @@ describe('Deezer Internal Source', function() { duringPlay.data.playDate = targetPlay.data.playDate.add(targetPlay.data.duration * 0.5, 's'); await using source = await generateDeezerSource({fuzzyDiscoveryIgnore: 'aggressive'}); - await source.discover([...normalizedPlays, interimPlay]); - - const discovered = await source.discover([duringPlay]); + //await source.discover([...normalizedPlays, interimPlay]); + await source.queuePlay([...normalizedPlays, interimPlay]); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); + await source.queuePlay([duringPlay]); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); - expect(discovered.length).to.eq(0); + expect(await source.getRecentlyDiscoveredPlays()).length(normalizedPlays.length + 1) }); it('does not discover fuzzy play with delay of up to 40 seconds', async function() { @@ -573,11 +580,12 @@ describe('Deezer Internal Source', function() { fuzzyPlay.data.playDate = targetPlay.data.playDate.add(targetPlay.data.duration + 39, 's'); await using source = await generateDeezerSource({fuzzyDiscoveryIgnore: 'aggressive'}); - await source.discover([...normalizedPlays, interimPlay]); - - const discovered = await source.discover([fuzzyPlay]); + await source.queuePlay([...normalizedPlays, interimPlay]); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); + await source.queuePlay([fuzzyPlay]); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); - expect(discovered.length).to.eq(0); + expect(await source.getRecentlyDiscoveredPlays()).length(normalizedPlays.length + 1) }); it('it does not discover fuzzy play when it is the last play ', async function() { @@ -586,11 +594,11 @@ describe('Deezer Internal Source', function() { fuzzyPlay.data.playDate = targetPlay.data.playDate.add(targetPlay.data.duration, 's'); await using source = await generateDeezerSource({fuzzyDiscoveryIgnore: 'aggressive'}); - await source.discover(normalizedPlays); - - const discovered = await source.discover([fuzzyPlay]); - - expect(discovered.length).to.eq(0); + await source.queuePlay(normalizedPlays); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); + await source.queuePlay([fuzzyPlay]); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); + expect(await source.getRecentlyDiscoveredPlays()).length(normalizedPlays.length); }); it('does not discover fuzzy play when it is played consecutively', async function() { @@ -600,9 +608,9 @@ describe('Deezer Internal Source', function() { const morePlays = normalizePlays([...normalizedPlays, fuzzyPlay, ...generatePlays(2)], {initialDate: firstPlayDate}); await using source = await generateDeezerSource({fuzzyDiscoveryIgnore: 'aggressive'}); - const discovered = await source.discover(morePlays); - - expect(discovered.length).to.eq(morePlays.length - 1); + await source.queuePlay(morePlays); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); + expect(await source.getRecentlyDiscoveredPlays()).length(morePlays.length - 1); }); }); diff --git a/src/backend/tests/ytm/ytm.test.ts b/src/backend/tests/ytm/ytm.test.ts index 4fa61e3f..9032e5c9 100644 --- a/src/backend/tests/ytm/ytm.test.ts +++ b/src/backend/tests/ytm/ytm.test.ts @@ -12,6 +12,7 @@ import { sleep } from '../../utils.ts'; import dayjs from 'dayjs'; import type {ApiResponse} from 'youtubei.js'; import type { MarkOptional } from 'ts-essentials'; +import pEvent from 'p-event'; chai.use(asPromised); @@ -168,7 +169,8 @@ describe('Handles interim tracks', function () { it(`Does not add skipped plays`, async function () { - const source = await createYtSource(); + await using source = await createYtSource(); + await source.startDiscoveryQueue(); const plays = [...generatePlays(10, {playDate: dayjs().subtract(20, 'seconds')}, { comment: 'Today' }), ...generatePlays(10, {playDate: dayjs().subtract(20, 'seconds')}, { comment: 'Yesterday' })]; @@ -176,7 +178,9 @@ describe('Handles interim tracks', function () { expect(source.parseRecentAgainstResponse(plays).plays).length(20); source.polling = true; - await source.discover(plays); + await source.queuePlay(plays); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); + //await source.discover(plays); // first true poll emulating no new tracks played (should not add new tracks from base truth) expect(source.parseRecentAgainstResponse(plays).plays).length(0); @@ -195,7 +199,8 @@ describe('Handles interim tracks', function () { it(`Adds interim plays when discover time is plausible`, async function () { - const source = await createYtSource(); + await using source = await createYtSource(); + await source.startDiscoveryQueue(); const plays = [...generatePlays(10, {playDate: dayjs().subtract(2, 'minutes')}, { comment: 'Today' }), ...generatePlays(10, {playDate: dayjs().subtract(2, 'minutes')}, { comment: 'Yesterday' })]; @@ -203,7 +208,9 @@ describe('Handles interim tracks', function () { expect(source.parseRecentAgainstResponse(plays).plays).length(20); source.polling = true; - await source.discover(plays); + await source.queuePlay(plays); + await Promise.race([pEvent(source.emitter, 'queueEmptied'), pEvent(source.emitter, 'discoveryQueueError')]); + //await source.discover(plays); // first true poll emulating no new tracks played (should not add new tracks from base truth) expect(source.parseRecentAgainstResponse(plays).plays).length(0); -- 2.51.2