diff --git a/src/backend/common/database/drizzle/repositories/PlayRepository.ts b/src/backend/common/database/drizzle/repositories/PlayRepository.ts index 207ef550..5229d4c9 100644 --- a/src/backend/common/database/drizzle/repositories/PlayRepository.ts +++ b/src/backend/common/database/drizzle/repositories/PlayRepository.ts @@ -1,9 +1,9 @@ import { childLogger } from "@foxxmd/logging"; import dayjs, { type Dayjs } from "dayjs"; -import { eq, inArray, relationsFilterToSQL, sql } from "drizzle-orm"; +import { eq, inArray, relationsFilterToSQL, sql, lte } from "drizzle-orm"; import assert from "node:assert"; import type { MarkOptional } from "ts-essentials"; -import { type DateLike, type DeepReplaceValue, type PlayObject, type PlayState, type QueueName, SCROBBLE_TS_SOC_END, TA_DEFAULT_ACCURACY, type TemporalAccuracy } from "../../../../../core/Atomic.ts"; +import { type DateLike, type DeepReplaceValue, type PlayObject, type PlayState, QUEUE_STATUS_QUEUED, type QueueName, SCROBBLE_TS_SOC_END, TA_DEFAULT_ACCURACY, type TemporalAccuracy } from "../../../../../core/Atomic.ts"; import { removeUndefinedKeys } from '../../../../../core/DataUtils.ts'; import { shortTodayAwareFormat } from "../../../../../core/TimeUtils.ts"; import { playContentBasicInvariantTransform, playMbidIdentifier } from "../../../../utils/PlayComparisonUtils.ts"; @@ -14,7 +14,7 @@ import type {ErrorLike, SourceType} from "../../../../../core/Atomic.ts"; import type {FindMany, FindWhere, FindWith, PlayInputNew, PlayNew, PlaySelect, PlaySelectWithQueueStates, PlayWith, QueueStateSelect, WhereClause} from "../drizzleTypes.ts"; import { type DbConcrete, runTransaction } from "../drizzleUtils.ts"; import { generateInputEntity, generatePlayEntity, hydratePlaySelect, stateChangeToPlayEvent, transformToPlayEvent, type PlayEntityOpts, type PlayHydateOptions } from "../entityUtils.ts"; -import { playEvents, playInputs, plays, relations } from "../schema/schema.ts"; +import { playEvents, playInputs, plays, relations, type TSchema } from "../schema/schema.ts"; import { buildDateCompare, type CompareDateOp, type ComponentConstrainedRepoOpts, DrizzleBaseRepository, type DrizzleRepositoryOpts } from "./BaseRepository.ts"; import type {PaginatedResponse} from "../../../../../core/Api.ts"; import type {PaginatedQueryResponse} from "../../../../../core/Api.ts"; @@ -25,6 +25,7 @@ import { type PlayEventTransform } from "../../../../../core/PlayEvent.ts"; export interface QueueCriteria { queueName: QueueName queueStatus: QueueStateSelect['queueStatus'][] | QueueStateSelect['queueStatus'] + retries?: number } export interface PlayWhereOpts { @@ -92,6 +93,16 @@ export class DrizzlePlayRepository extends DrizzleBaseRepository<'plays'> { return res as PlaySelectWithQueueStates; } + findByIdWith = async (id: number, args: WithPlayRelation[]): Promise | undefined> => { + const res = await this.db.query.plays.findFirst({ + where: { + id + }, + with: buildPlayWith(args) + }); + return res as unknown as PlayWith | undefined; + } + createPlays = async (entitiesOpts: RepositoryCreatePlayOpts[], opts: HydrateOpts = {}): Promise[]> => { const { @@ -480,25 +491,30 @@ export class DrizzlePlayRepository extends DrizzleBaseRepository<'plays'> { componentId: sql.placeholder('componentId'), queueStates: { queueName: sql.placeholder('queueName'), - queueStatus: 'queued', + queueStatus: sql.placeholder('status'), retries: { lte: sql.placeholder('retries') } }, }, with: { - queueStates: true - }, - orderBy: { - seenAt: 'asc' + queueStates: { + orderBy: { + updatedAt: 'asc' + } + } }, + // orderBy: { + // seenAt: 'asc' + // }, }).prepare() - public getQueueNext = async (queueName: string, opts: {order?: 'asc' | 'desc', retries?: number, notIds?: number[]} & ComponentConstrainedRepoOpts = {}): Promise => { + public getQueueNext = async (queueName: string, opts: {order?: 'asc' | 'desc', retries?: number, notIds?: number[], status?: QueueStateSelect['queueStatus']} & ComponentConstrainedRepoOpts = {}): Promise => { const { - retries = 0, + retries = 1000, notIds, order = 'asc', + status = QUEUE_STATUS_QUEUED, componentId = this.componentId } = opts; @@ -509,7 +525,7 @@ export class DrizzlePlayRepository extends DrizzleBaseRepository<'plays'> { this.getQueueNextPrepared = this.prepareGetQueueNext(); } - res = await this.getQueueNextPrepared.execute({ queueName, retries, componentId }); + res = await this.getQueueNextPrepared.execute({ queueName, retries, componentId, status }); } else { // cannot bind arrays in sqlite so this has to be non-prepared res = await this.db.query.plays.findFirst({ @@ -520,18 +536,22 @@ export class DrizzlePlayRepository extends DrizzleBaseRepository<'plays'> { componentId: componentId, queueStates: { queueName: queueName, - queueStatus: 'queued', + queueStatus: status, retries: { lte: retries } }, }, with: { - queueStates: true - }, - orderBy: { - seenAt: 'asc' + queueStates: { + orderBy: { + updatedAt: 'asc' + } + } }, + // orderBy: { + // seenAt: 'asc' + // }, }); } @@ -899,14 +919,26 @@ export const buildPlayWhere = (args: PlayWhereOpts): WhereClause<'plays'> => { // or else assigning an array to OR using only `typeof where.queueStates` causes a type error const queueWhere: typeof where.queueStates.OR[0][] = []; for(const q of queues) { - queueWhere.push( - { - queueName: q.queueName, - queueStatus: typeof q.queueStatus === 'string' ? q.queueStatus : { - in: q.queueStatus - } + const qWhereCriteria: typeof where.queueStates.OR[0] = { + queueName: q.queueName, + queueStatus: typeof q.queueStatus === 'string' ? q.queueStatus : { + in: q.queueStatus } - ) + }; + if(q.retries !== undefined) { + qWhereCriteria.retries = { + lte: q.retries + } + } + queueWhere.push(qWhereCriteria); + // queueWhere.push( + // { + // queueName: q.queueName, + // queueStatus: typeof q.queueStatus === 'string' ? q.queueStatus : { + // in: q.queueStatus + // } + // } + // ) } if(queueWhere.length === 1) { where.queueStates = queueWhere[0]; diff --git a/src/backend/common/database/drizzle/repositories/QueueRepository.ts b/src/backend/common/database/drizzle/repositories/QueueRepository.ts index a6296e52..b47b4191 100644 --- a/src/backend/common/database/drizzle/repositories/QueueRepository.ts +++ b/src/backend/common/database/drizzle/repositories/QueueRepository.ts @@ -1,9 +1,9 @@ -import { eq, and, lte, inArray } from "drizzle-orm"; +import { eq, and, gte, lte, inArray } from "drizzle-orm"; import { DrizzleBaseRepository, type DrizzleRepositoryOpts } from "./BaseRepository.ts"; import type {DbConcrete} from "../drizzleUtils.ts"; import type {PlaySelect, QueueStateSelect} from "../drizzleTypes.ts"; import { playEvents, queueStates } from "../schema/schema.ts"; -import { DEAD_QUEUE } from "../../../../../core/Atomic.ts"; +import { DEAD_QUEUE, INGRESS_QUEUE } from "../../../../../core/Atomic.ts"; import { queueStateToPlayEvent } from "../entityUtils.ts"; export class DrizzleQueueRepository extends DrizzleBaseRepository<'queueStates'> { @@ -18,7 +18,7 @@ export class DrizzleQueueRepository extends DrizzleBaseRepository<'queueStates'> eq(queueStates.componentId, componentId), lte(queueStates.retries, retries), eq(queueStates.queueStatus, 'failed'), - eq(queueStates.queueName, DEAD_QUEUE) + eq(queueStates.queueName, INGRESS_QUEUE) )); } @@ -32,12 +32,29 @@ export class DrizzleQueueRepository extends DrizzleBaseRepository<'queueStates'> )); } - public getQueueCount = async (componentId: number, queueNames: string[], queueStatus: QueueStateSelect['queueStatus'][] = ['queued']): Promise => { + public getQueueCount = async (componentId: number, queueNames: string[], opts: { + queueStatus?: QueueStateSelect['queueStatus'][], + retries?: number + retryEq?: 'lte' | 'gte' + } = {}): Promise => { + const { + queueStatus = ['queued'], + retries, + retryEq = 'lte' + } = opts + if(retries === undefined) { + return await this.db.$count(queueStates, and( + eq(queueStates.componentId, componentId), + inArray(queueStates.queueName, queueNames), + inArray(queueStates.queueStatus, queueStatus) + )); + } return await this.db.$count(queueStates, and( eq(queueStates.componentId, componentId), inArray(queueStates.queueName, queueNames), - inArray(queueStates.queueStatus, queueStatus) - )); + inArray(queueStates.queueStatus, queueStatus), + retryEq === 'lte' ? lte(queueStates.retries, retries) : gte(queueStates.retries, retries) + )); } async create(data: typeof this.table.$inferInsert & {playId?: PlaySelect['id'], event?: boolean}): Promise { diff --git a/src/backend/common/errors/PlayProcessingError.ts b/src/backend/common/errors/PlayProcessingError.ts new file mode 100644 index 00000000..55afed13 --- /dev/null +++ b/src/backend/common/errors/PlayProcessingError.ts @@ -0,0 +1,18 @@ +import { NamedError } from "./MSErrors.ts"; +import type { PlayProcessingResult } from "../infrastructure/PlayProcessing.ts"; + +export interface PlayProcessingErrorData extends PlayProcessingResult { + showStopping: boolean +} + +export class PlayProcessingError extends NamedError { + override name = 'Play Processing Error'; + result: PlayProcessingResult + showStopping: boolean + + constructor(error: Error, data: PlayProcessingErrorData, opts: ErrorOptions = {}) { + super('Error occured while processing Play', {...opts, cause: error}); + this.result = data; + this.showStopping = data.showStopping; + } +} \ No newline at end of file diff --git a/src/backend/common/infrastructure/PlayProcessing.ts b/src/backend/common/infrastructure/PlayProcessing.ts new file mode 100644 index 00000000..e402dcdf --- /dev/null +++ b/src/backend/common/infrastructure/PlayProcessing.ts @@ -0,0 +1,8 @@ +import type { PlayEvent } from "../../../core/PlayEvent.ts" +import type { PlaySelectWithQueueStates, QueueStateSelect } from "../database/drizzle/drizzleTypes.ts" + +export interface PlayProcessingResult { + playEntity: PlaySelectWithQueueStates, + events: Omit[], + queue: QueueStateSelect +} \ No newline at end of file diff --git a/src/backend/scrobblers/AbstractScrobbleClient.ts b/src/backend/scrobblers/AbstractScrobbleClient.ts index 21786b54..390cc559 100644 --- a/src/backend/scrobblers/AbstractScrobbleClient.ts +++ b/src/backend/scrobblers/AbstractScrobbleClient.ts @@ -5,20 +5,20 @@ import type EventEmitter from "events"; import { nanoid } from "nanoid"; import type { MarkOptional } from "ts-essentials"; import { - type DeadLetterScrobble, type NowPlayingUpdateThreshold, type PlayObject, - type QueuedScrobble, type ScrobbleActionResult, type PlayMatchResult, type SourcePlayerObj, + type ScrobbleActionResult, type PlayMatchResult, type SourcePlayerObj, type ErrorLike, INGRESS_QUEUE, DEAD_QUEUE, - type PlayOriginal, - type PlayLifecycle, type SourcePlayerJson, QUEUE_STATUS_COMPLETED, - SOURCE_SOT + SOURCE_SOT, + QUEUE_STATUS_FAILED, + isPlayObject, + DEAD_LETTER_RETRIES_DEFAULT } from "../../core/Atomic.ts"; -import { artistNamesToCredits, buildTrackString, capitalize, truncateStringToLength } from "../../core/StringUtils.ts"; +import { buildTrackString, capitalize, truncateStringToLength } from "../../core/StringUtils.ts"; import AbstractComponent from "../common/AbstractComponent.ts"; import { hasUpstreamError } from "../common/errors/UpstreamError.ts"; import { @@ -45,7 +45,6 @@ import { sleep, sortByOldestPlayDate, } from "../utils.ts"; -import { removeUndefinedKeys } from '../../core/DataUtils.ts'; import { findCauseByReference } from "../utils/ErrorUtils.ts"; import { messageWithCausesTruncatedDefault } from "../../core/ErrorUtils.ts"; import { @@ -64,18 +63,20 @@ import type { Counter, Gauge } from 'prom-client'; import { generateLoggableAbortReason, ScrobbleSubmitError, SimpleError, StageChangeError } from "../common/errors/MSErrors.ts"; import {isErrorLike, serializeError} from 'serialize-error'; import { DEFAULT_NEW_PADDING, groupPlaysToTimeRanges } from "../utils/ListenFetchUtils.ts"; -import { spawn, isAbortError, delay } from 'abort-controller-x'; +import { spawn, isAbortError, delay, waitForEvent, type AbortError } from 'abort-controller-x'; import { DrizzlePlayRepository, playToRepositoryCreatePlayOpts, type QueryPlaysOpts, type WithPlayRelation } from "../common/database/drizzle/repositories/PlayRepository.ts"; -import type {ComponentMigrationNew, PlaySelect, PlaySelectWithQueueStates, QueueStateNew, QueueStateSelect} from "../common/database/drizzle/drizzleTypes.ts"; +import type {PlayEventNew, PlayEventSelect, PlaySelect, PlaySelectWithQueueStates, PlayWith, QueueStateSelect} from "../common/database/drizzle/drizzleTypes.ts"; import { asPlay } from "../../core/PlayMarshalUtils.ts"; import { DrizzleQueueRepository } from "../common/database/drizzle/repositories/QueueRepository.ts"; import { GenericRepository } from "../common/database/drizzle/repositories/BaseRepository.ts"; import assert from "node:assert"; -import { COMPONENT_STATE, type ComponentClientApiJson, type PlayApiCommonDetailed } from "../../core/Api.ts"; +import { COMPONENT_STATE, type ComponentClientApiJson, type PlayApiCommonDetailed, type QueueStateApi } from "../../core/Api.ts"; import type {ComponentState} from "react"; import { DrizzlePlayEventsRepository } from "../common/database/drizzle/repositories/PlayEventsRepository.ts"; -import { PLAY_EVENT_TYPE, type PlayEvent } from "../../core/PlayEvent.ts"; -import { dupeCheckToPlayEvent, queueStateToPlayEvent, scrobbleToPlayEvent, stateChangeToPlayEvent, transformToPlayEvent } from "../common/database/drizzle/entityUtils.ts"; +import { type PlayEvent } from "../../core/PlayEvent.ts"; +import { dupeCheckToPlayEvent, entityIsPlayEntity, queueStateToPlayEvent, scrobbleToPlayEvent, stateChangeToPlayEvent, transformToPlayEvent } from "../common/database/drizzle/entityUtils.ts"; +import type { PlayProcessingResult } from "../common/infrastructure/PlayProcessing.ts"; +import { PlayProcessingError } from "../common/errors/PlayProcessingError.ts"; type SourceMappedPlayer = {player: SourcePlayerObj, source: SourceIdentifier}; type PlatformMappedPlays = Map; @@ -160,7 +161,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i declare protected componentType: 'client'; - protected playRepo!: DrizzlePlayRepository; + public playRepo!: DrizzlePlayRepository; protected queueRepo!: DrizzleQueueRepository; protected playEventsRepo!: DrizzlePlayEventsRepository; protected migrationRepo!: GenericRepository<'componentMigrations'>; @@ -384,11 +385,6 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i protected async postCache(): Promise { await super.postCache(); - try { - await this.migrateCachedScrobbles(); - } catch (e) { - this.logger.warn(new Error('Unable to migrate cached scrobbles (if any). Will continue init and ignore this error.', {cause: e})); - } this.generateStaggerMappers(); } @@ -409,12 +405,12 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i protected async updateQueueStats(queueNames: string[]) { if(queueNames.includes(INGRESS_QUEUE)) { - this.queuedLength = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE]); + this.queuedLength = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE], {retries: 0}); this.queuedGauge.labels(this.getPrometheusLabels()).set(this.queuedLength); } if(queueNames.includes(DEAD_QUEUE)) { - this.deadLetterLength = await this.queueRepo.getQueueCount(this.dbComponent.id, [DEAD_QUEUE], ['queued', 'failed']); - this.deadLetterQueued = await this.queueRepo.getQueueCount(this.dbComponent.id, [DEAD_QUEUE], ['queued']); + this.deadLetterLength = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE], {queueStatus: ['failed','queued'], retries: 1, retryEq: 'gte'}); + this.deadLetterQueued = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE], {queueStatus: ['queued'], retries: 1, retryEq: 'gte'}); // TODO this.deadLetterGauge.labels(this.getPrometheusLabels()).set(this.deadLetterLength); } @@ -662,175 +658,6 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i } } - protected async migrateCachedScrobbles(): Promise { - const logger = childLogger(this.logger, ['Cached Scrobble Migration']); - let shouldMigrate: boolean = false; - const migration = this.dbComponent.migrations.find(x => x.name === 'cachedScrobbles'); - if (migration === undefined) { - logger.verbose('No migration has run yet, running now...'); - shouldMigrate = true; - } else if (migration.success === false) { - logger.verbose('Re-running previously failed migration now...'); - shouldMigrate = true; - } - if (shouldMigrate) { - const migrationEntry: ComponentMigrationNew = migration !== undefined ? migration : {componentId: this.dbComponent.id, name: 'cachedScrobbles'}; - try { - const cachedQueue = (await this.cache.cacheScrobble.get(`${this.getMachineId()}-queue`) as QueuedScrobble>[] ?? []); - const migratedQueue: QueuedScrobble[] = []; - let allGood = true; - if (cachedQueue.length > 0) { - logger.info('Migrating cached scrobbles to database...'); - for (const cachedQueuedScrobble of cachedQueue) { - if(cachedQueuedScrobble.play.meta?.migrated === true) { - logger.debug(`Skipping already migrated play => ${buildTrackString(cachedQueuedScrobble.play)}`); - continue; - } - const play = asPlay(cachedQueuedScrobble.play) as PlayObject<{migrated?: boolean, lifecycle?: PlayLifecycle}>; - const { - meta: { - lifecycle, - ...metaRest - }, - data: { - listenRanges, - artists, - albumArtists, - ...dataRest - } = {}, - } = play; - try { - const updatedPlay: PlayObject = { - ...play, - data: { - artists: artists === undefined ? undefined : artistNamesToCredits(artists as unknown as string[]), - albumArtists: albumArtists === undefined ? undefined : artistNamesToCredits(albumArtists as unknown as string[]), - ...dataRest - }, - meta: metaRest, - - lifecycle: lifecycle?.steps - } - if(lifecycle !== undefined) { - if('scrobble' in lifecycle) { - updatedPlay.scrobble = lifecycle.scrobble; - } - if('input' in lifecycle || 'original' in lifecycle) { - updatedPlay.original = removeUndefinedKeys({ - - data: lifecycle.input, - play: lifecycle.original - }) - } - } - // return play object without going through transform since it was (presumably) already transformed before being cached - const res = await this.queueScrobble(updatedPlay, {transform: false}); - if(res.length === 1) { - logger.verbose(`Migrated Play ${res[0].uid} => ${buildTrackString(play)}`); - } - cachedQueuedScrobble.play.meta.migrated = true; - migratedQueue.push(cachedQueuedScrobble) - } catch (e) { - migratedQueue.push(cachedQueuedScrobble); - allGood = false; - logger.warn(new Error(`Failed to migrate Play ${buildTrackString(play)}`, { cause: e })); - } - } - await this.cache.cacheScrobble.set(`${this.getMachineId()}-queue`, migratedQueue); - logger[allGood ? 'info' : 'warn'](allGood ? 'Finished migrating all queued scrobbles.' : 'Migrated queued scrobbles with errors'); - } else { - logger.info('No scrobbles to migrate'); - } - - const cachedDead = (await this.cache.cacheScrobble.get(`${this.getMachineId()}-dead`) as DeadLetterScrobble>[] ?? []); - const migratedDead: DeadLetterScrobble[] = []; - if (cachedDead.length > 0) { - logger.info('Migrating failed scrobbles to database...'); - let allGood = true; - for (const cDeadScrobble of cachedDead) { - if(cDeadScrobble.play.meta?.migrated === true) { - logger.debug(`Skipping already migrated play => ${buildTrackString(cDeadScrobble.play)}`) - continue; - } - const play = asPlay(cDeadScrobble.play) as PlayObject<{migrated?: boolean, lifecycle?: PlayLifecycle}>; - const { - meta: { - lifecycle, - ...metaRest - }, - data: { - listenRanges, - artists, - albumArtists, - ...dataRest - } = {}, - } = play; - const updatedDeadPlay: PlayObject = { - ...play, - data: { - artists: artists === undefined ? undefined : artistNamesToCredits(artists as unknown as string[]), - albumArtists: albumArtists === undefined ? undefined : artistNamesToCredits(albumArtists as unknown as string[]), - ...dataRest - }, - meta: metaRest, - lifecycle: lifecycle?.steps - } - if(lifecycle !== undefined) { - if('scrobble' in lifecycle) { - updatedDeadPlay.scrobble = lifecycle.scrobble; - } - if('input' in lifecycle || 'original' in lifecycle) { - updatedDeadPlay.original = removeUndefinedKeys({ - data: lifecycle.input, - play: lifecycle.original - }) - } - } - try { - const res = await this.playRepo.createPlays([ - playToRepositoryCreatePlayOpts({ - play: updatedDeadPlay, - componentId: this.dbComponent.id, - state: 'failed', - parentId: play.id - }) - ]); - logger.verbose(`Added Play ${res[0].uid} to database => ${buildTrackString(play)}`); - await this.addDeadLetterScrobble(res[0], cDeadScrobble.error); - logger.verbose(`Added Play ${res[0].uid} to Failed Queue`); - cDeadScrobble.play.meta.migrated = true; - migratedDead.push(cDeadScrobble); - } catch (e) { - migratedDead.push(cDeadScrobble); - allGood = false; - logger.warn(new Error(`Failed to migrate Play to failed queued ${buildTrackString(play)}`, { cause: e })); - } - } - logger[allGood ? 'info' : 'warn'](allGood ? 'Finished migrating all failed scrobbles.' : 'Migrated failed scrobbles with errors'); - await this.cache.cacheScrobble.set(`${this.getMachineId()}-dead`, migratedDead); - } else { - logger.info('No dead scrobbles to migrate'); - } - - if(migration === undefined) { - await this.migrationRepo.create({...migrationEntry, success: allGood}); - } else { - await this.migrationRepo.updateById(migration.id, {success: allGood}); - } - logger[allGood ? 'info' : 'warn'](`Migration done${allGood === false ? ' with errors' : ''}`); - } catch (e) { - if(migration === undefined) { - this.migrationRepo.create({...migrationEntry, success: false, error: e}); - } else { - this.migrationRepo.updateById(migration.id, {success: false, error: e}); - } - throw new Error('Cached Scrobble Migration failed with unexpected error', {cause: e}); - } - } else { - logger.debug('Cached Scrobbles Migration already run successfully!'); - } - } - protected async postInitialize(): Promise { super.postInitialize(); const { @@ -1189,7 +1016,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i let nextQueued = await this.playRepo.getQueueNext(INGRESS_QUEUE); if(nextQueued !== undefined) { while (nextQueued !== undefined) { - await this.processQueueCurrentScrobble(nextQueued, signal); + await this.handlePlayProcessing(nextQueued, signal); if(this.errors.length > 0) { // we made it through a scrobble without any issues so clear any issue we may have previously had this.errors = []; @@ -1215,170 +1042,77 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i } } - protected processQueueCurrentScrobble = async (currQueuedPlay: PlaySelectWithQueueStates, signal: AbortSignal) => { - signal.throwIfAborted(); - //const currQueuedPlay = await this.playRepo.getQueueNext(CLIENT_INGRESS_QUEUE); - // if (currQueuedPlay === undefined) { - // this.logger.trace('Nothing queued'); - // return; - // } - this.setStatus(`Processing Play ${currQueuedPlay.id}`); - await this.handleQueuedScrobbleRanges(); - if (!this.upstreamRefresh.refreshEnabled) { - // TODO add signal for this to scrobble match - this.logger.trace('Scrobble refresh is DISABLED.'); - } - - //let handledShiftedPlay = false; - //const currQueuedPlay = await this.playRepo.getQueueNext(CLIENT_INGRESS_QUEUE); - //const currQueuedPlay = this.queuedScrobbles.shift(); - - let historicalPlays: PlayObject[] = []; - let historicalError: Error | undefined; - let queueError: Error | undefined; - let successState: PlaySelect['state']; - let deadQueueEntity: QueueStateSelect; - - const events: Omit[] = []; - + protected handlePlayProcessing = async (playEntity: PlaySelectWithQueueStates, signal?: AbortSignal) => { + let res: PlayProcessingResult, + err: Error; try { - - if(!currQueuedPlay.play.meta.wasMonitored) { - successState = 'discarded'; - this.logger.debug(`Not processing ${buildTrackString(currQueuedPlay.play)} because monitoring was disabled when Play was queued.`); - events.push(stateChangeToPlayEvent({state: 'discarded', reason: 'Monitoring was disabled when Play was queued'})); - return; + res = await this.processPlay(playEntity, signal); + } catch (e: unknown | Error | AbortError | PlayProcessingError) { + if(isAbortError(e)) { + err = generateLoggableAbortReason('Interrupted by abort signal', this.scrobbleQueueAbortController.signal); + throw e; } - - if (this.upstreamRefresh.refreshEnabled) { - try { - historicalPlays = await this.getSOTScrobblesForPlay(currQueuedPlay.play); - } catch (e) { - historicalError = e; - if (e.message === 'Cannot get historical plays due to cached error') { - this.logger.warn(`${buildTrackString(currQueuedPlay.play)} from Source '${currQueuedPlay.play.meta.source}' => Previous error while getting historical scrobbles means this scrobble cannot be compared, will queue as dead for now.`); - this.logger.trace(e); - queueError = e; - } else { - queueError = new SimpleError(`${buildTrackString(currQueuedPlay.play)} from Source '${currQueuedPlay.play.meta.source}' => cannot get historical scrobbles, will queue as dead for now.`, { cause: e, shortStack: true }); - this.logger.warn(queueError); - } - deadQueueEntity = await this.addDeadLetterScrobble(currQueuedPlay, e); - //handledShiftedPlay = true; + if(e instanceof PlayProcessingError) { + err = e.cause as Error; + res = e.result; + if(e.showStopping) { + throw e.cause; } - signal.throwIfAborted(); - } - if (historicalError === undefined) { - const { summary, ...matchResult } = await this.existingScrobble(currQueuedPlay.play, historicalPlays); - events.push(dupeCheckToPlayEvent({summary, ...matchResult})); - // currQueuedPlay.play.scrobble = { - // ...(currQueuedPlay.play.scrobble ?? {}), - // match: matchResult, - // createdAt: dayjs() - // } - signal.throwIfAborted(); - if (!matchResult.match) { - const transformedScrobble = await this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.postCompare); - const { lifecycle = [], ...restPlay } = transformedScrobble; - currQueuedPlay.play = restPlay; - const psLifecycle = lifecycle.filter(x => x.hook === TRANSFORM_HOOK.postCompare); - if(psLifecycle.length > 0) { - events.push({...transformToPlayEvent(psLifecycle), createdAt: dayjs()}); - } - signal.throwIfAborted(); - try { - const scrobbledPlay = await this.scrobble(transformedScrobble, {signal}); - const {scrobble} = scrobbledPlay; - events.push(scrobbleToPlayEvent(scrobble)); - //currQueuedPlay.play = scrobbledPlay; - await this.addScrobbledTrack(scrobbledPlay); - //handledShiftedPlay = true; - } catch (e) { - const scrobbleRes: ScrobbleResult = { - createdAt: dayjs() - } - - const submitError = findCauseByReference(e, ScrobbleSubmitError); - if (submitError !== undefined) { - scrobbleRes.payload = submitError.payload; - scrobbleRes.response = submitError.responseBody; - scrobbleRes.error = serializeError(submitError); - } else { - scrobbleRes.payload = this.playToClientPayload(transformedScrobble); - scrobbleRes.error = serializeError(e); - } - events.push(scrobbleToPlayEvent(scrobbleRes)); - queueError = e; - deadQueueEntity = await this.addDeadLetterScrobble(currQueuedPlay, e); - //handledShiftedPlay = true; - if (hasUpstreamError(e, false)) { - //handledShiftedPlay = true; - const nonShowStoppingError = new Error(`Could not scrobble ${buildTrackString(transformedScrobble)} from Source '${currQueuedPlay.play.meta.source}' but error was not show stopping. Adding scrobble to Dead Letter Queue and will retry on next heartbeat.`, { cause: e }); - this.logger.warn(nonShowStoppingError); - queueError = nonShowStoppingError; - } else { - //this.queuedScrobbles.unshift(currQueuedPlay); - //handledShiftedPlay = true; - const showStoppingError = new Error('Error occurred while trying to scrobble', { cause: e }); - queueError = showStoppingError; - throw showStoppingError; - } - } + } else { + const unhandledError = new Error('Unhandled error type while processing Play', {cause: e}); + if(e instanceof Error) { + err = e; } else { - successState = 'duped'; - this.setStatus(`Play ${currQueuedPlay.id} detected as dupe`); + err = unhandledError; } + throw e; } - signal.throwIfAborted(); - // reset retries if we've made this far - this.scrobbleRetries = 0; - } catch (e) { - if(queueError === undefined) { - queueError = e; - } - // if(!handledShiftedPlay) { - // this.queuedScrobbles.unshift(currQueuedPlay); - // } - throw e; } finally { - const queueState = currQueuedPlay.queueStates.find(x => x.queueName === INGRESS_QUEUE); - if(queueError !== undefined) { - await this.queueRepo.updateById(queueState.id, { - queueStatus: 'failed', - error: queueError, - // ensure that ingress queue updatedAt is always older than dead queue creation so timeline is ordered correctly - updatedAt: deadQueueEntity !== undefined ? deadQueueEntity.createdAt.subtract(1, 'ms') : dayjs() + let queueStates: QueueStateSelect[]; + const initialRetries = res.queue.retries ?? 0; + if(err !== undefined) { + res.queue.retries = (initialRetries + 1); + res.queue.updatedAt = dayjs(); + await this.queueRepo.updateById(res.queue.id, { + ...res.queue, }); - queueState.queueStatus = 'failed'; - queueState.error = queueError; - await this.playRepo.updateById(currQueuedPlay.id, {state: 'failed', error: queueError, play: currQueuedPlay.play}); - events.push(stateChangeToPlayEvent({state: 'failed'})); - events.push(queueStateToPlayEvent(queueState)); - currQueuedPlay.state = 'failed'; - //currQueuedPlay.error = queueError; + if(initialRetries === 0) { + this.deadLetterGauge.labels(this.getPrometheusLabels()).inc(); + this.deadLetterLength += 1; + this.deadLetterQueued += 1; + } else if(res.queue.retries > this.getDefaultDeadLetterRetries()) { + this.deadLetterQueued -= 1; + } + queueStates = res.playEntity.queueStates.filter(x => x.queueName !== res.queue.queueName).concat([res.queue]); } else { - await this.queueRepo.updateById(queueState.id, {queueStatus: 'completed'}); - await this.playRepo.updateById(currQueuedPlay.id, {state: successState ?? 'scrobbled', play: currQueuedPlay.play}); - if(!events.some(x => x.eventName === PLAY_EVENT_TYPE.playStateChange)) { - events.push(stateChangeToPlayEvent({state: successState ?? 'scrobbled'})); + await this.queueRepo.deleteByIds([res.queue.id]); + if(res.queue.retries > 0) { + this.deadLetterGauge.labels(this.getPrometheusLabels()).dec(); + this.deadLetterLength -= 1; + this.deadLetterQueued -= 1; } - currQueuedPlay.state = successState ?? 'scrobbled'; - queueState.queueStatus = 'completed'; - events.push(queueStateToPlayEvent(queueState)); + queueStates = res.playEntity.queueStates.filter(x => x.queueName !== res.queue.queueName) } - const createdEvents = await this.playEventsRepo.createMany(events.map(x => ({...x, playId: currQueuedPlay.id}))); - this.emitPlayUpdate({...currQueuedPlay, events: createdEvents, queueStates: [queueState]} as unknown as PlayApiCommonDetailed); - this.emitEvent('scrobbleDequeued', { queuedScrobble: currQueuedPlay }) - this.queuedGauge.labels(this.getPrometheusLabels()).dec(); - this.queuedLength -= 1; + if(initialRetries === 0) { + this.queuedGauge.labels(this.getPrometheusLabels()).dec(); + this.queuedLength -= 1; + } + this.playRepo.updateById(playEntity.id, {play: res.playEntity.play, state: res.playEntity.state, error: res.playEntity.error}); + const createdEvents = await this.playEventsRepo.createMany(res.events.map(x => ({...x, playId: playEntity.id}))) as PlayEventSelect[]; + this.emitPlayUpdate({ + ...res.playEntity, + events: ((res.playEntity as unknown as PlayWith<'events'>).events ?? []).concat(createdEvents), + queueStates + } as unknown as PlayApiCommonDetailed); + this.emitEvent('scrobbleDequeued', { queuedScrobble: playEntity }); } } - processDeadLetterQueue = async (attemptWithRetries?: number) => { + protected getDefaultDeadLetterRetries() { + return this.config.options?.deadLetterRetries ?? DEAD_LETTER_RETRIES_DEFAULT; + } - // if (this.deadLetterScrobbles.length === 0) { - // return; - // } + processDeadLetterQueue = async (attemptWithRetries?: number, reason?: string, sync?: boolean) => { if (!(await this.isReady())) { this.deadLogger.warn('Cannot process dead letter scrobbles because client is not ready.'); @@ -1396,7 +1130,6 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i } = this.config; const retries = attemptWithRetries ?? deadLetterRetries; - const removedIds = []; this.deadQueueAbortController = new AbortController(); this.deadQueuePromise = spawn(this.deadQueueAbortController.signal, async (signal, { defer, fork }) => { @@ -1407,37 +1140,39 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i }); this.emitEvent('queueState', {queueName: 'dead', status: 'Running'}); - await this.queueRepo.deadFailedToQueue(this.dbComponent.id, retries); - const processable = await this.queueRepo.getQueueCount(this.dbComponent.id, [DEAD_QUEUE]); //this.deadLetterScrobbles.filter(x => x.retries < retries); - this.deadLetterQueued = processable; + //const processable = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE], [QUEUE_STATUS_FAILED], retries); + const processableArgs: QueryPlaysOpts = {queues: [{queueName: INGRESS_QUEUE, queueStatus: QUEUE_STATUS_FAILED, retries}], with: ['queues']}; + let processable = await this.playRepo.findPlaysPaginated(processableArgs); + this.deadLetterQueued = processable.meta.total; - const total = await this.queueRepo.getQueueCount(this.dbComponent.id, [DEAD_QUEUE], ['queued','failed']); + const total = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE], {queueStatus: [QUEUE_STATUS_FAILED], retries: 10000}); this.deadLetterLength = total; - const queueStatus = `${processable} of ${total} dead scrobbles have less than ${retries} retries, ${processable === 0 ? 'will skip processing.': 'processing now...'}`; - if (processable === 0) { + const queueStatus = `${processable.meta.total} of ${total} dead scrobbles have less than ${retries} retries, ${processable.meta.total === 0 ? 'will skip processing.': 'processing now...'}`; + if (processable.meta.total === 0) { this.deadLogger.verbose(queueStatus); return; } - this.setStatus(`Processing ${processable} Dead Plays`); + this.setStatus(`Queuing ${processable} Dead Plays...`); this.logger.info(queueStatus); - if(!this.upstreamRefresh.refreshEnabled) { - this.deadLogger.verbose('Scrobble refresh is DISABLED. All dead scrobbles will likely always be scrobbled (nothing to check duplicates against).'); - } - // await this.handleQueuedScrobbleRanges(); - - let nextQueued: PlaySelectWithQueueStates = await this.playRepo.getQueueNext(DEAD_QUEUE, {retries}); - if(nextQueued !== undefined) { - while(nextQueued !== undefined) { - const [scrobbled, dead] = await this.processDeadLetterScrobble(nextQueued.uid, signal); - await sleep(this.scrobbleSleep); - if(scrobbled) { - removedIds.push(dead.id); - } - nextQueued = await this.playRepo.getQueueNext(DEAD_QUEUE, {retries}); + let more = true; + let offset = 0; + while(more) { + await this.queueScrobble(processable.data, {reason}); + more = processable.data.length === processable.meta.limit; + if(more) { + offset += processable.meta.limit; + processable = await this.playRepo.findPlaysPaginated({...processableArgs, offset}); } } + this.setStatus(`All processable Dead Plays have been queued`); + this.logger.info(`All processable Dead Plays have been queued`); + if(sync) { + await waitForEvent(signal,this.emitter,'queueEmptied'); + this.setStatus(`Finished processing Dead Plays`); + this.logger.info('Finished processing Dead Plays'); + } }).catch((e) => { if (isAbortError(e)) { const err = generateLoggableAbortReason('Dead scrobble processing stopped', this.deadQueueAbortController.signal); @@ -1447,107 +1182,102 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i this.logger.warn(new Error('Dead scrobble processing stopped with error', { cause: e })); } }).finally(() => { - if (removedIds.length > 0) { - this.deadLogger.info(`Removed ${removedIds.length} scrobbles from dead letter queue`); - } this.deadQueueAbortController = undefined; this.deadQueuePromise = undefined; }); } - processDeadLetterScrobble = async (uid: string, signal?: AbortSignal): Promise<[boolean, PlaySelectWithQueueStates?]> => { + processPlay = async (playEntity: PlaySelectWithQueueStates, signal?: AbortSignal): Promise => { signal?.throwIfAborted(); - // const deadScrobbleIndex = this.deadLetterScrobbles.findIndex(x => x.id === id); - // if(deadScrobbleIndex === -1) { - // this.deadLogger.warn(`Could not find a dead scrobble with id ${id}`); - // return [false]; - // } - const deadScrobble: PlaySelectWithQueueStates = await this.playRepo.findByUid(uid, {hydrate: ['asPlay']}); - if(deadScrobble === undefined) { - throw new Error(`Play ${uid} does not exist for ${this.name}`); - } + const queueState = playEntity.queueStates.find(x => x.queueName === INGRESS_QUEUE); + + const { + useCache = true, + isRetry = false, + transform = true, + dupeCheck = true, + } = queueState.context || {}; + + const isDead = queueState.retries > 0 || isRetry; + + const logger = isDead ? childLogger(this.logger, ['Dead', `Play ${playEntity.uid}`]) : childLogger(this.logger, [`Play ${playEntity.uid}`]); + this.setStatus(`Processing ${isDead ? 'Dead ' : ''}Play ${playEntity.uid}`); + + let processError: Error | undefined; + let successState: PlaySelect['state']; const events: Omit[] = []; - let deadQueueState: QueueStateSelect; + try { - if (deadScrobble.state === 'scrobbled') { - throw new Error(`Play ${uid} is already scrobbled.`); - } - deadQueueState = deadScrobble.queueStates.find(x => x.queueName === DEAD_QUEUE); - if (deadQueueState === undefined) { - throw new Error(`Play ${uid} is not currently queued in dead letter.`); - } - this.setStatus(`Processing Dead Play ${uid}`); - //const deadScrobble = await this.playRepo.getQueueNext(this.dbComponent.id, CLIENT_INGRESS_QUEUE); - const deadLabel = { labels: deadScrobble.uid }; - //const deadScrobble = this.deadLetterScrobbles[deadScrobbleIndex]; - this.deadLogger.trace(deadLabel, `Processing dead scrobble => ${buildTrackString(deadScrobble.play)}`); - - await this.handleQueuedScrobbleRanges(); - signal?.throwIfAborted(); - - if (!(await this.isReady())) { - this.deadLogger.warn(deadLabel, 'Cannot process dead letter scrobble because client is not ready.'); - return [false, deadScrobble]; + if (!isRetry && !playEntity.play.meta.wasMonitored) { + successState = 'discarded'; + this.logger.debug(`Not processing ${buildTrackString(playEntity.play)} because monitoring was disabled when Play was queued.`); + events.push(stateChangeToPlayEvent({ state: 'discarded', reason: 'Monitoring was disabled when Play was queued' })); + playEntity.state = 'discarded'; + return {playEntity, queue: queueState, events}; } + let historicalPlays: PlayObject[] = []; - if (this.upstreamRefresh.refreshEnabled) { + + if (dupeCheck && this.upstreamRefresh.refreshEnabled) { + await this.handleQueuedScrobbleRanges(); try { - historicalPlays = await this.getSOTScrobblesForPlay(deadScrobble.play); + historicalPlays = await this.getSOTScrobblesForPlay(playEntity.play); } catch (e) { + if (e.message === 'Cannot get historical plays due to cached error') { - this.deadLogger.warn(deadLabel, `Previous error while getting historical scrobbles means this scrobble cannot be compared`); - this.deadLogger.trace(e); + logger.warn(`${buildTrackString(playEntity.play)} from Source '${playEntity.play.meta.source}' => Previous error while getting historical scrobbles means this scrobble cannot be compared, will queue as dead for now.`); + logger.trace(e); + processError = e; } else { - this.deadLogger.warn(new SimpleError(`${deadScrobble.uid} - ${buildTrackString(deadScrobble.play)} from Source '${deadScrobble.play.meta.source}' => cannot get historical scrobbles`, { cause: e, shortStack: true })); + processError = new SimpleError(`${buildTrackString(playEntity.play)} from Source '${playEntity.play.meta.source}' => cannot get historical scrobbles, will queue as dead for now.`, { cause: e, shortStack: true }); + logger.warn(processError); } - - events.push(queueStateToPlayEvent({...deadQueueState, queueStatus: 'failed', error: e})); - this.queueRepo.updateById(deadQueueState.id, { retries: deadQueueState.retries + 1, error: e, updatedAt: dayjs(), queueStatus: 'failed' }); - //this.playRepo.updateById(deadScrobble.id, {error: e}); - // deadScrobble.retries++; - // deadScrobble.error = messageWithCauses(e); - // deadScrobble.lastRetry = dayjs(); - // this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble; - this.emitEvent('updateDeadLetter', { dead: deadScrobble }); - return [false, deadScrobble]; + playEntity.state = 'failed'; + events.push(stateChangeToPlayEvent({state: 'failed'})); + queueState.queueStatus = QUEUE_STATUS_FAILED; + queueState.error = processError; + events.push(queueStateToPlayEvent(queueState)); + throw new PlayProcessingError(processError, {playEntity, events, queue: queueState, showStopping: false}); + //deadQueueEntity = await this.addDeadLetterScrobble(playEntity, e); } + signal.throwIfAborted(); } - signal?.throwIfAborted(); - const { summary, ...matchResult } = await this.existingScrobble(deadScrobble.play, historicalPlays); - events.push(dupeCheckToPlayEvent({summary, ...matchResult})) - // deadScrobble.play.scrobble = { - // ...(deadScrobble.play.scrobble ?? {}), - // match: matchResult, - // createdAt: dayjs() - // } - if (!matchResult.match) { - const transformedScrobble = await this.transformPlay(deadScrobble.play, TRANSFORM_HOOK.postCompare); - const { lifecycle = [] } = transformedScrobble; + + let isDupe = false; + if(dupeCheck) { + const { summary, ...matchResult } = await this.existingScrobble(playEntity.play, historicalPlays); + events.push(dupeCheckToPlayEvent({ summary, ...matchResult })); + isDupe = matchResult.match; + } + + signal.throwIfAborted(); + if (!isDupe) { + const transformedScrobble = transform ? await this.transformPlay(playEntity.play, TRANSFORM_HOOK.postCompare, {useCachedResult: useCache}) : playEntity.play; + const { lifecycle = [], ...restPlay } = transformedScrobble; + playEntity.play = restPlay; const psLifecycle = lifecycle.filter(x => x.hook === TRANSFORM_HOOK.postCompare); - if(psLifecycle.length > 0) { - events.push({...transformToPlayEvent(psLifecycle), createdAt: dayjs()}); + if (psLifecycle.length > 0) { + events.push({ ...transformToPlayEvent(psLifecycle), createdAt: dayjs() }); } - signal?.throwIfAborted(); + signal.throwIfAborted(); try { - const scrobbledPlay = await this.scrobble(transformedScrobble); - const {scrobble} = scrobbledPlay; + const scrobbledPlay = await this.scrobble(transformedScrobble, { signal }); + const { scrobble } = scrobbledPlay; events.push(scrobbleToPlayEvent(scrobble)); - deadScrobble.play = scrobbledPlay; + //currQueuedPlay.play = scrobbledPlay; await this.addScrobbledTrack(scrobbledPlay); - this.playRepo.updateById(deadScrobble.id, { play: deadScrobble.play, state: 'scrobbled' }); - this.queueRepo.updateById(deadQueueState.id, { error: null, updatedAt: dayjs(), queueStatus: QUEUE_STATUS_COMPLETED }); - events.push(queueStateToPlayEvent({...deadQueueState, queueStatus: QUEUE_STATUS_COMPLETED})); events.push(stateChangeToPlayEvent({state: 'scrobbled'})); - this.removeDeadLetterScrobble(deadScrobble, 'scrobbled', true); + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED})); + this.scrobbleRetries = 0; + playEntity.state = 'scrobbled'; + playEntity.error = undefined; + return {playEntity, events, queue: queueState}; } catch (e) { const scrobbleRes: ScrobbleResult = { createdAt: dayjs() } - // deadScrobble.play.scrobble = { - // ...(deadScrobble.play.scrobble ?? {}), - // //createdAt: dayjs() - // } + const submitError = findCauseByReference(e, ScrobbleSubmitError); if (submitError !== undefined) { scrobbleRes.payload = submitError.payload; @@ -1558,173 +1288,208 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i scrobbleRes.error = serializeError(e); } events.push(scrobbleToPlayEvent(scrobbleRes)); - this.queueRepo.updateById(deadQueueState.id, { retries: deadQueueState.retries + 1, error: e, updatedAt: dayjs(), queueStatus: 'failed' }); - events.push(queueStateToPlayEvent({...deadQueueState, queueStatus: 'failed', error: e})); - //this.playRepo.updateById(deadScrobble.id, { play: deadScrobble.play }); - // deadScrobble.retries++; - // deadScrobble.error = messageWithCauses(e); - // deadScrobble.lastRetry = dayjs(); - this.deadLogger.error(new Error(`${deadScrobble.uid} - Could not scrobble ${buildTrackString(transformedScrobble)} from Source '${deadScrobble.play.meta.source}' due to error`, { cause: e })); - //this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble; - this.emitEvent('updateDeadLetter', { dead: deadScrobble }); - return [false, deadScrobble]; + playEntity.state = 'failed'; + events.push(stateChangeToPlayEvent({state: 'failed'})); + queueState.queueStatus = QUEUE_STATUS_FAILED; + if (hasUpstreamError(e, false)) { + //handledShiftedPlay = true; + const nonShowStoppingError = new Error(`Could not scrobble but error was not show stopping. May be retried automatically in Dead Queue`, { cause: e }); + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: nonShowStoppingError})); + queueState.error = nonShowStoppingError; + logger.warn(nonShowStoppingError); + processError = nonShowStoppingError; + throw new PlayProcessingError(nonShowStoppingError, {playEntity, queue: queueState, events, showStopping: false}); + } else { + //this.queuedScrobbles.unshift(currQueuedPlay); + //handledShiftedPlay = true; + const showStoppingError = new Error('Error occurred while trying to scrobble', { cause: e }); + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: showStoppingError})); + queueState.error = showStoppingError; + processError = showStoppingError; + throw new PlayProcessingError(showStoppingError, {playEntity, queue: queueState, events, showStopping: true}); + } } } else { - this.playRepo.updateById(deadScrobble.id, { play: deadScrobble.play }); - this.deadLogger.verbose(`Looks like ${buildTrackString(deadScrobble.play)} was already scrobbled!\n${summary}`); - this.removeDeadLetterScrobble(deadScrobble, 'duped', true); - events.push(queueStateToPlayEvent({...deadQueueState, queueStatus: QUEUE_STATUS_COMPLETED})); - events.push(stateChangeToPlayEvent({state: 'duped', reason: 'Looks like it was already scrobbled downstream'})); + successState = 'duped'; + this.setStatus(`Play ${playEntity.id} detected as dupe`); + this.scrobbleRetries = 0; + playEntity.state = 'duped'; + return {playEntity, events, queue: queueState}; } - - return [true, deadScrobble]; } catch (e) { - if(deadQueueState !== undefined) { - events.push(queueStateToPlayEvent({...deadQueueState, queueStatus: 'failed', error: e})); + if(e instanceof PlayProcessingError) { + throw e; } - } finally { - await this.playEventsRepo.createMany(events.map(x => ({...x, playId: deadScrobble.id}))); + if(isAbortError(e)) { + events.push(stateChangeToPlayEvent({state: 'failed'})); + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: generateLoggableAbortReason('Interrupted by abort signal', this.scrobbleQueueAbortController.signal)})); + throw e; + } + throw new PlayProcessingError(e, {playEntity, queue: queueState, events, showStopping: true}); } } - removeDeadLetterScrobble = async (dead: (PlaySelect & {queueStates: QueueStateSelect[]}) | string, state: PlaySelect['state'], success: boolean) => { + removeDeadLetterScrobble = async (dead: PlaySelectWithQueueStates) => { - let deadScrobble: PlaySelect & {queueStates: QueueStateSelect[]}; - - if(typeof dead === 'string'){ - deadScrobble = await this.playRepo.findByUid(dead, {hydrate: ['asPlay']}); - if(deadScrobble === undefined) { - throw new Error(`Play ${dead} does not exist for ${this.name}`); - } - } else { - deadScrobble = dead; - } - this.setStatus(`Removing Dead Play ${dead} from queue`); - // const index = this.deadLetterScrobbles.findIndex(x => x.id === id); - // if (index === -1) { - // this.deadLogger.warn(`No scrobble found with ID ${id}`); - // return; - // } - const deadQueueState = deadScrobble.queueStates.find(x => x.queueName === DEAD_QUEUE && x.queueStatus !== 'completed'); - if(deadQueueState === undefined) { - throw new Error(`Play ${deadScrobble.uid} is not currently queued in dead letter.`); - } - const isQueued = deadQueueState.queueStatus === 'queued'; - //this.deadLetterScrobbles.splice(index, 1); - this.deadLetterGauge.labels(this.getPrometheusLabels()).dec(); - const queueUpdate: Partial = { - updatedAt: dayjs(), - queueStatus: 'completed' + const queueState = dead.queueStates.find(x => x.queueName === INGRESS_QUEUE); + if(queueState === undefined) { + this.logger.warn(`Play ${dead.uid} does not have a dead state, nothing to remove.`); + return; } - if(success) { - queueUpdate.error = null; + if(queueState.retries === 0) { + this.logger.warn(`Play ${dead.uid} has not failed yet, not removing.`); + return; } - await this.queueRepo.updateById(deadQueueState.id, queueUpdate); - await this.playRepo.updateById(deadScrobble.id, removeUndefinedKeys({state, error: success ? null : undefined})); - this.deadLogger.info({labels: deadScrobble.uid}, `Scrobble ${buildTrackString(deadScrobble.play)} marked as completed`); + this.setStatus(`Marking Dead Play ${dead.uid} as completed`); + + const events: PlayEventNew[] = [ + { playId: dead.id, ...queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED, context: {reason: 'Dead Play marked as completed by user'}}) } + ]; + await this.queueRepo.deleteByIds([queueState.id]); + + if(dead.state === 'queued') { + dead.state = 'failed'; + this.playRepo.updateById(dead.id, {state: 'failed'}); + events.push( + { playId: dead.id, ...stateChangeToPlayEvent({ state: 'failed' }) } + ) + } + const createdEvents = await this.playEventsRepo.createMany(events) as PlayEventSelect[]; + this.emitPlayUpdate({uid: dead.uid, + state: dead.state, + queueStates: dead.queueStates.filter(x => x.queueName !== INGRESS_QUEUE) as unknown as QueueStateApi[], + events: createdEvents as unknown as PlayEvent[] + }); + this.deadLetterLength -= 1; - if(isQueued) { + if(queueState.queueStatus === 'queued') { this.deadLetterQueued -= 1; } - if(state === 'scrobbled') { - this.componentRepo.updateById(this.dbComponent.id, {countLive: this.dbComponent.countLive + 1}); - } - this.emitEvent('removeDeadLetter', { dead: { id: deadScrobble.uid } }); + this.deadLetterGauge.labels(this.getPrometheusLabels()).dec(); + this.emitEvent('removeDeadLetter', { dead: { id: dead.uid } }); } - removeDeadLetterScrobbles = async (types: QueueStateSelect['queueStatus'][] = ['queued'], state: PlaySelect['state'], success: boolean) => { + removeDeadLetterScrobbles = async () => { const ids = await this.playRepo.findPlayIdentifiers({ queues: [ { - queueName: DEAD_QUEUE, - queueStatus: types + queueName: INGRESS_QUEUE, + queueStatus: 'failed' } ] - }, 'uid'); - this.deadLogger.info(`Marking ${ids} as completed but unsuccessful...`); - await Promise.all(ids.map((x) => this.removeDeadLetterScrobble(x, state, success))); + }, 'id'); + this.deadLogger.info(`Marking ${ids.length} as completed...`); + await pMap(ids, async (id) => { + const entity = await this.playRepo.findByIdWith<'queueStates'>(id, ['queues']); + if(entity !== undefined) { + await this.removeDeadLetterScrobble(entity); + } + }, {concurrency: 10}); this.deadLogger.info('Finished processing dead scrobbles.'); await this.updateQueueStats([DEAD_QUEUE]); } - queueScrobble = async (data: PlayObject | PlayObject[], context?: QueueContext) => { - const monitoring = this.getMonitoringStatus(); - const { - transform = true, - } = context || {}; - const playDatas = (Array.isArray(data) ? data : [data]).map(x => ({...x, meta: {...x.meta, wasMonitored: monitoring.monitoring, seenAt: dayjs()}})); - + queueScrobble = async (data: (PlayObject | PlayObject[]) | (PlaySelectWithQueueStates | PlaySelectWithQueueStates[]), context?: QueueContext) => { const createdQueuedPlays: PlaySelect[] = []; - for await(const play of pMapIterable(playDatas, this.staggerMappers.preCompare(async x => transform === false ? await noopTransform(x) : await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 3})) { - const events: Omit[] = []; - try { - // cheap check, looks for play data (non-meta) hash, playdate, and optionally mbid recording - const cheapExisting = await this.playRepo.checkExisting(play, { queueName: INGRESS_QUEUE }); - if (cheapExisting !== undefined) { - const qs = cheapExisting.queueStates.find(x => x.queueName === INGRESS_QUEUE); - this.logger.trace(`Not adding to queue because it is already in the queue, discovered via hash/mbid, last queued at ${todayAwareFormat(qs.createdAt)}`); - continue; + const dataArray = Array.isArray(data) ? data : [data]; + + if (dataArray.every(x => entityIsPlayEntity(x))) { + for (const playSelect of dataArray) { + let queue = playSelect.queueStates.find(x => x.queueName === INGRESS_QUEUE); + if (queue === undefined) { + queue = await this.queueRepo.create({ componentId: this.dbComponent.id, playId: playSelect.id, queueName: INGRESS_QUEUE, context }) as QueueStateSelect; + } else { + this.queueRepo.updateById(queue.id, { queueStatus: 'queued', context }); + } + const events = await this.playEventsRepo.createMany([ + { playId: playSelect.id, ...stateChangeToPlayEvent({ state: 'queued' }) }, + { playId: playSelect.id, ...queueStateToPlayEvent({ ...queue, context: context ?? queue.context }) } + ]) as PlayEventSelect[]; + playSelect.state = 'queued'; + await this.playRepo.updateById(playSelect.id, {state: 'queued'}); + if (`events` in playSelect) { + (playSelect as PlayWith<'events'>).events = (playSelect as PlayWith<'events'>).events.concat(events); + } else { + (playSelect as unknown as PlayWith<'events'>).events = events; } - // then chunked queued plays - let offset = 0; - let inQueue = false; - while (true) { - const { data, meta } = await this.playRepo.getQueued(INGRESS_QUEUE, { offset }); - const existingQueued = await this.existingScrobble(play, data.map(x => asPlay(x.play)), false); - // want to be very confident of this - if (existingQueued.match && existingQueued.score > 0.99) { - this.logger.trace(`Not adding to queue because it is already in the queue\n${existingQueued.summary}`); - inQueue = true; - break; + this.emitPlayUpdate({ ...playSelect } as unknown as PlayApiCommonDetailed); + createdQueuedPlays.push(playSelect); + } + } else if (dataArray.every(x => isPlayObject(x))) { + const monitoring = this.getMonitoringStatus(); + const { + transform = true, + } = context || {}; + const playDatas = dataArray.map(x => ({...x, meta: {...x.meta, wasMonitored: monitoring.monitoring, seenAt: dayjs()}})); + + for await(const play of pMapIterable(playDatas, this.staggerMappers.preCompare(async x => transform === false ? await noopTransform(x) : await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 3})) { + try { + // cheap check, looks for play data (non-meta) hash, playdate, and optionally mbid recording + const cheapExisting = await this.playRepo.checkExisting(play, { queueName: INGRESS_QUEUE }); + if (cheapExisting !== undefined) { + const qs = cheapExisting.queueStates.find(x => x.queueName === INGRESS_QUEUE); + this.logger.trace(`Not adding to queue because it is already in the queue, discovered via hash/mbid, last queued at ${todayAwareFormat(qs.createdAt)}`); + continue; } - if (data.length < meta.limit) { - break; + // then chunked queued plays + let offset = 0; + let inQueue = false; + while (true) { + const { data, meta } = await this.playRepo.getQueued(INGRESS_QUEUE, { offset, retries: 0 }); + const existingQueued = await this.existingScrobble(play, data.map(x => asPlay(x.play)), false); + // want to be very confident of this + if (existingQueued.match && existingQueued.score > 0.99) { + this.logger.trace(`Not adding to queue because it is already in the queue\n${existingQueued.summary}`); + inQueue = true; + break; + } + if (data.length < meta.limit) { + break; + } + offset += meta.limit; } - offset += meta.limit; - } - if (inQueue) { - continue; + if (inQueue) { + continue; + } + } catch (e) { + this.logger.warn(new SimpleError('Failed to check queued scrobble for existing before adding, will continue with adding anyway', { cause: e })); } - } catch (e) { - this.logger.warn(new SimpleError('Failed to check queued scrobble for existing before adding, will continue with adding anyway', { cause: e })); - } - // not in queue or existing queued check failed for some reason and we don't want to lose scrobble - const { - data, - meta - } = play - const createPlayData = playToRepositoryCreatePlayOpts({ - play: { + // not in queue or existing queued check failed for some reason and we don't want to lose scrobble + const { data, meta - }, - componentId: this.dbComponent.id, - state: 'queued', - parentId: play.id - }); + } = play + const createPlayData = playToRepositoryCreatePlayOpts({ + play: { + data, + meta + }, + componentId: this.dbComponent.id, + state: 'queued', + parentId: play.id + }); - const playRow = await this.playRepo.createPlays([createPlayData]); - const queueState = await this.queueRepo.create({componentId: this.dbComponent.id, playId: playRow[0].id, queueName: INGRESS_QUEUE}) as QueueStateSelect; - await this.playEventsRepo.createMany([ - {playId: playRow[0].id, ...stateChangeToPlayEvent({state: 'queued'}), createdAt: playRow[0].seenAt.add(1,'ms')}, - {playId: playRow[0].id, ...queueStateToPlayEvent(queueState), createdAt: queueState.createdAt} - ]); - createdQueuedPlays.push(playRow[0]); - this.logger.debug(`Added ${buildTrackString(play)} to the queue`); - this.setStatus(`Added Play from parent ${play.uid} to queue`); - - const queuedPlay = {id: nanoid(), source: play.meta.source, play: play} - //await this.playRepo.updateById(play.meta.dbId, {play}); - this.emitEvent('scrobbleQueued', {queuedPlay: queuedPlay}); - this.emitPlayInsert({...playRow[0], queueStates: [queueState]} as unknown as PlayApiCommonDetailed); - this.queuedLength += 1; - //this.queuedScrobbles.push(queuedPlay); - this.queuedGauge.labels(this.getPrometheusLabels()).inc(); - // this is wasteful but we don't want the processing loop popping out-of-order (by date) scrobbles - //this.queuedScrobbles.sort((a, b) => sortByOldestPlayDate(a.play, b.play)); + const playRow = await this.playRepo.createPlays([createPlayData]); + const queueState = await this.queueRepo.create({componentId: this.dbComponent.id, playId: playRow[0].id, queueName: INGRESS_QUEUE, context}) as QueueStateSelect; + const createdEvents = await this.playEventsRepo.createMany([ + {playId: playRow[0].id, ...stateChangeToPlayEvent({state: 'queued'}), createdAt: playRow[0].seenAt.add(1,'ms')}, + {playId: playRow[0].id, ...queueStateToPlayEvent(queueState), createdAt: queueState.createdAt} + ]); + createdQueuedPlays.push(playRow[0]); + this.logger.debug(`Added ${buildTrackString(play)} to the queue`); + this.setStatus(`Added Play from parent ${play.uid} to queue`); + + const queuedPlay = {id: nanoid(), source: meta.source, play: play} + this.emitEvent('scrobbleQueued', {queuedPlay: queuedPlay}); + this.emitPlayInsert({...playRow[0], queueStates: [queueState], events: createdEvents} as unknown as PlayApiCommonDetailed); + this.queuedLength += 1; + this.queuedGauge.labels(this.getPrometheusLabels()).inc(); + } + } else { + throw new Error('Data passed to queuePlay must be either all be PlayObject or all PlaySelect objects'); } return createdQueuedPlays; } diff --git a/src/backend/server/api.ts b/src/backend/server/api.ts index a30b715f..484e9a4f 100644 --- a/src/backend/server/api.ts +++ b/src/backend/server/api.ts @@ -40,13 +40,14 @@ import { DrizzlePlayRepository, type QueryPlaysOpts, type QueryPlaysOptsJson } f import { playSelectToDeadScrobble } from "../common/database/drizzle/entityUtils.ts"; import AbstractHistoricalScrobbleClient from "../scrobblers/AbstractHistoricalScrobbleClient.ts"; import { DrizzlePlayHistoricalRepository } from "../common/database/drizzle/repositories/PlayHistoricalRepository.ts"; -import {componentStateBodySchema, type ComponentClientApiJson, type ComponentSourceApiJson} from "../../core/Api.ts"; +import {componentStateBodySchema, type ComponentClientApiJson, type ComponentSourceApiJson, type PlayApiCommonDetailed} from "../../core/Api.ts"; import { asDayjsHydratedObject } from "../../core/DataUtils.ts"; import type {Dayjs} from "dayjs"; import { asSerializablePlaySelect } from "../../core/PlayMarshalUtils.ts"; import { serializeError } from "serialize-error"; import { z } from 'zod'; import type { createTypedRouter } from "@minisylar/express-typed-router"; +import pEvent from "p-event"; const maxBufferSize = 300; const output: Record> = {}; @@ -645,14 +646,24 @@ export const setupApi = (app: Express, router: ReturnType null); + const event = await pEvent(client.emitter, 'playUpdate', { + timeout: 10000, + filter: (val: PlayApiCommonDetailed) => val.uid === id + }) as PlayApiCommonDetailed; + if(event.state === 'scrobbled') { return res.status(200).send(); + } else { + // @ts-expect-error should be fine + return res.json(playSelectToDeadScrobble(event, true)); } - return res.json(playSelectToDeadScrobble(dead, true)); } catch (e) { if(e.message.includes(`Play ${deadId} does not exist`)) { logger.warn(e); @@ -670,7 +681,7 @@ export const setupApi = (app: Express, router: ReturnType null).catch((e) => logger.error(e)); + (client as AbstractScrobbleClient).removeDeadLetterScrobbles().then(() => null).catch((e) => logger.error(e)); return res.sendStatus(200); }); @@ -687,14 +698,15 @@ export const setupApi = (app: Express, router: ReturnType(3, () => ({ ...fixtureCreatePlay(), state: 'queued', input: {} }))) + const queuedPlays = await testScrobbler.playRepoTest.createPlays(generateArray(3, () => ({ ...fixtureCreatePlay(), state: 'failed', input: {} }))); - for(const dead of queuedPlayed) { - await testScrobbler.addDeadLetterScrobble(dead); - } + await testScrobbler.queueRepoTest.createMany(queuedPlays.map(x => ({queueName: INGRESS_QUEUE, queueStatus: QUEUE_STATUS_FAILED, retries: 1, playId: x.id, componentId: testScrobbler.componentId}))); - await testScrobbler.processDeadLetterQueue(); - await pEvent(testScrobbler.emitter, 'queueState'); + await testScrobbler.processDeadLetterQueue(undefined, undefined, true); - expect(testScrobbler.deadLetterQueued).eq(0); + const updatedPlays = await testScrobbler.playRepoTest.findPlays({uid: queuedPlays.map(x => x.uid)}); + for(const p of updatedPlays) { + expect(p.state === 'scrobbled'); + } }); }); diff --git a/src/core/Atomic.ts b/src/core/Atomic.ts index e9bf744c..97672e2d 100644 --- a/src/core/Atomic.ts +++ b/src/core/Atomic.ts @@ -655,11 +655,14 @@ export const QUEUE_STATUS_COMPLETED: QueueStatus = 'completed'; export const QUEUE_STATUS_FAILED: QueueStatus = 'failed'; export const QUEUE_STATUSES: QueueStatus[] = [QUEUE_STATUS_COMPLETED, QUEUE_STATUS_FAILED, QUEUE_STATUS_QUEUED]; +export const DEAD_LETTER_RETRIES_DEFAULT = 3; + export interface QueueContext { transform?: boolean dupeCheck?: boolean useCache?: boolean isRetry?: boolean + reason?: string } /** -- 2.51.2 From 37ba860203a7b965344f037e14697ad371aea928 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Fri, 21 Aug 2026 18:45:46 +0000 Subject: [PATCH 2/6] fix(deezer): Fix missing closest play for some fuzzy scenarios --- src/backend/sources/DeezerInternalSource.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/backend/sources/DeezerInternalSource.ts b/src/backend/sources/DeezerInternalSource.ts index 4edd3ee9..8f09aad8 100644 --- a/src/backend/sources/DeezerInternalSource.ts +++ b/src/backend/sources/DeezerInternalSource.ts @@ -334,7 +334,7 @@ export default class DeezerInternalSource extends MemorySource { async existingDiscovered(play: PlayObject): Promise { - const list: PlayObject[] = await this.getRecentlyDiscoveredPlays(); + const list: PlayObject[] = await this.getRecentPlays(); const candidate = await this.transformPlay(play, TRANSFORM_HOOK.candidate); const existing = await findAsync(list, async x => { const e = await this.transformPlay(x, TRANSFORM_HOOK.existing); @@ -386,7 +386,7 @@ export default class DeezerInternalSource extends MemorySource { score: 0.5, breakdowns: [], reason: 'Has matching data for previous play but assuming its on repeat', - closestMatchedPlay: existing, + closestMatchedPlay: list[fuzzyIndex], createdAt: dayjs().toISOString() } } @@ -397,7 +397,7 @@ export default class DeezerInternalSource extends MemorySource { score: 1, breakdowns: [], reason: 'Has matching data and looks like a misreported play', - closestMatchedPlay: existing, + closestMatchedPlay: list[fuzzyIndex], createdAt: dayjs().toISOString() } } -- 2.51.2 From 91c2cc6ef5490b18d847faea3ee1f14993888c3b Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Fri, 21 Aug 2026 19:20:14 +0000 Subject: [PATCH 3/6] Finish queue implementation for source and various fixes --- src/backend/common/AbstractComponent.ts | 2 +- .../drizzle/repositories/PlayRepository.ts | 2 +- .../infrastructure/config/client/index.ts | 14 +- .../common/infrastructure/config/common.ts | 15 + .../infrastructure/config/source/index.ts | 3 +- src/backend/index.ts | 6 +- src/backend/ioc.ts | 6 + .../scrobblers/AbstractScrobbleClient.ts | 66 ++-- src/backend/sources/AbstractSource.ts | 344 +++++++++++++++--- src/backend/utils/AsyncUtils.ts | 2 +- .../msComponent/MSComponentDetailed.tsx | 5 +- .../msComponent/MSComponentSummary.tsx | 27 +- src/client/components/msComponent/Stats.tsx | 45 ++- src/core/Api.ts | 6 +- src/core/tests/utils/apiFixtures.ts | 12 +- 15 files changed, 391 insertions(+), 164 deletions(-) diff --git a/src/backend/common/AbstractComponent.ts b/src/backend/common/AbstractComponent.ts index 7ebd9965..ec55ee01 100644 --- a/src/backend/common/AbstractComponent.ts +++ b/src/backend/common/AbstractComponent.ts @@ -554,7 +554,7 @@ export default abstract class AbstractComponent extends AbstractInitializable { public abstract getRunningState(): ComponentState - public getApiData(): Omit & Pick { + public getApiData(): Omit & Pick { let state: ComponentState; if(!this.initializedOnce || this.initializing) { state = COMPONENT_STATE.INITIALIZING; diff --git a/src/backend/common/database/drizzle/repositories/PlayRepository.ts b/src/backend/common/database/drizzle/repositories/PlayRepository.ts index 5229d4c9..ace3dd4c 100644 --- a/src/backend/common/database/drizzle/repositories/PlayRepository.ts +++ b/src/backend/common/database/drizzle/repositories/PlayRepository.ts @@ -1,6 +1,6 @@ import { childLogger } from "@foxxmd/logging"; import dayjs, { type Dayjs } from "dayjs"; -import { eq, inArray, relationsFilterToSQL, sql, lte } from "drizzle-orm"; +import { eq, inArray, relationsFilterToSQL, sql } from "drizzle-orm"; import assert from "node:assert"; import type { MarkOptional } from "ts-essentials"; import { type DateLike, type DeepReplaceValue, type PlayObject, type PlayState, QUEUE_STATUS_QUEUED, type QueueName, SCROBBLE_TS_SOC_END, TA_DEFAULT_ACCURACY, type TemporalAccuracy } from "../../../../../core/Atomic.ts"; diff --git a/src/backend/common/infrastructure/config/client/index.ts b/src/backend/common/infrastructure/config/client/index.ts index 0d6d704b..0261bcb7 100644 --- a/src/backend/common/infrastructure/config/client/index.ts +++ b/src/backend/common/infrastructure/config/client/index.ts @@ -1,6 +1,6 @@ import * as z from "zod"; import {playTransformOptionsSchema} from "../../../../../core/Transform.ts"; -import {commonConfigSchema, requestRetryOptionsSchema, monitorOptionsSchema, type CommonComponentEnvShape} from "../common.ts"; +import {commonConfigSchema, requestRetryOptionsSchema, monitorOptionsSchema, type CommonComponentEnvShape, deadLetterOptionsSchema} from "../common.ts"; import {retentionConfigDurationValueSchema} from "../database.ts"; import type { PipeUnwrapDirection } from "../../../../utils/ZodUtils.ts"; @@ -124,6 +124,7 @@ export const commonClientOptionsSchema = z.object({ ...monitorOptionsSchema.shape, ...requestRetryOptionsSchema.shape, ...upstreamRefreshOptionsSchema.shape, + ...deadLetterOptionsSchema.shape, /** * Check client for an existing scrobble at the same recorded time as the "new" track to be scrobbled. If an existing scrobble is found this track is not track scrobbled. @@ -143,17 +144,6 @@ export const commonClientOptionsSchema = z.object({ }).optional().meta({ description: "Options used for increasing verbosity of logging in MS (used for debugging)" }), - /** - * Number of times MS should automatically retry scrobbles in dead letter queue - * - * @default 3 - * @examples [3] - * */ - deadLetterRetries: z.number().optional().meta({ - description: "Number of times MS should automatically retry scrobbles in dead letter queue", - default: 3, - examples: [3] - }), /** Enhance/correct Play data by applying a transform pipeline */ playTransform: playTransformOptionsSchema.optional().meta({description: 'Enhance/correct Play data by applying a transform pipeline'}), diff --git a/src/backend/common/infrastructure/config/common.ts b/src/backend/common/infrastructure/config/common.ts index 4f314cdf..3734d7b4 100644 --- a/src/backend/common/infrastructure/config/common.ts +++ b/src/backend/common/infrastructure/config/common.ts @@ -155,6 +155,21 @@ export const monitorOptionsSchema = z.object({ }) export type MonitorOptions = z.infer; +export const deadLetterOptionsSchema = z.object({ + /** + * Number of times MS should automatically retry Plays in dead letter queue + * + * @default 3 + * @examples [3] + * */ + deadLetterRetries: z.number().optional().meta({ + description: "Number of times MS should automatically retry Plays in dead letter queue", + default: 3, + examples: [3] + }) +}) +export type DeadLetterOptions = z.infer; + export type UnparsedConfig = {config: object, type: T, source?: 'file' | 'aio' | 'env', pos: string}; export const generateConfigLocation = (configType: string, config: UnparsedConfig): string => { diff --git a/src/backend/common/infrastructure/config/source/index.ts b/src/backend/common/infrastructure/config/source/index.ts index 641daf60..6b47ae24 100644 --- a/src/backend/common/infrastructure/config/source/index.ts +++ b/src/backend/common/infrastructure/config/source/index.ts @@ -1,5 +1,5 @@ import * as z from "zod"; -import {requestRetryOptionsSchema, commonConfigSchema, monitorOptionsSchema} from "../common.ts"; +import {requestRetryOptionsSchema, commonConfigSchema, monitorOptionsSchema, deadLetterOptionsSchema} from "../common.ts"; import {retentionConfigDurationValueSchema} from "../database.ts"; import {playTransformOptionsSchema} from "../../../../../core/Transform.ts"; import type { PipeUnwrapDirection } from "../../../../utils/ZodUtils.ts"; @@ -73,6 +73,7 @@ export const fileLogOptionsSchema = z.object({ export const commonSourceOptionsSchema = z.object({ ...monitorOptionsSchema.shape, ...sourceRetryOptionsSchema.shape, + ...deadLetterOptionsSchema.shape, /** * * If this source has INGRESS to MS (sends a payload, rather than MS GETTING requesting a payload) then setting this option to true will make MS log the payload JSON to DEBUG output * * If this source is POLLING then it will log the raw data for each unique track/response the first time it is seen diff --git a/src/backend/index.ts b/src/backend/index.ts index e4a207e2..88fb1936 100644 --- a/src/backend/index.ts +++ b/src/backend/index.ts @@ -209,9 +209,9 @@ const dataDir = getDataDir(); if(nameColl.length > 0) { logger.warn(`Last.FM source and clients have same names [${nameColl.map(x => x.name).join(',')}] -- this may cause issues`); } - const clientInitOptions = {deadDelay: nonEmptyStringOrDefault(process.env.DEBUG_DEAD_DELAY, undefined) !== undefined ? Number.parseInt(process.env.DEBUG_DEAD_DELAY) : undefined}; + const initOptions = {deadDelay: nonEmptyStringOrDefault(process.env.DEBUG_DEAD_DELAY, undefined) !== undefined ? Number.parseInt(process.env.DEBUG_DEAD_DELAY) : undefined}; for(const c of scrobbleClients.clients) { - c.initTasks(clientInitOptions); + c.initTasks(initOptions); const res = await Promise.race([ sleep(2200), (async () => { @@ -227,7 +227,7 @@ const dataDir = getDataDir(); } for(const c of scrobbleSources.sources) { - c.initTasks(); + c.initTasks(initOptions); const res = await Promise.race([ sleep(2200), (async () => { diff --git a/src/backend/ioc.ts b/src/backend/ioc.ts index 79866d9c..e258af20 100644 --- a/src/backend/ioc.ts +++ b/src/backend/ioc.ts @@ -35,6 +35,11 @@ const discovered = new prom.Counter({ help: 'Number of discovered plays for a Source', labelNames: ['name', 'type'] }); +const sourceDead = new prom.Gauge({ + name: 'multiscrobbler_source_dead', + help: 'Number of dead letter plays for a Source', + labelNames: ['name', 'type'] +}); const queuedGauge = new prom.Gauge({ name: 'multiscrobbler_client_queued', help: 'Number of queued plays for a Client', @@ -150,6 +155,7 @@ const createRoot = (options: RootOptions = {logger: loggerDebug}) => { sourceMetics: { discovered: discovered, queued: queuedSourceGauge, + deadLetter: sourceDead //issues: sourceIssues }, clientMetrics: { diff --git a/src/backend/scrobblers/AbstractScrobbleClient.ts b/src/backend/scrobblers/AbstractScrobbleClient.ts index 390cc559..8f907f69 100644 --- a/src/backend/scrobblers/AbstractScrobbleClient.ts +++ b/src/backend/scrobblers/AbstractScrobbleClient.ts @@ -8,7 +8,6 @@ import { type NowPlayingUpdateThreshold, type PlayObject, type ScrobbleActionResult, type PlayMatchResult, type SourcePlayerObj, - type ErrorLike, INGRESS_QUEUE, DEAD_QUEUE, type SourcePlayerJson, @@ -61,9 +60,9 @@ import { statefulInvariantTransform } from "../../core/PlayUtils.ts"; import { normalizeStr } from "../utils/StringUtils.ts"; import type { Counter, Gauge } from 'prom-client'; import { generateLoggableAbortReason, ScrobbleSubmitError, SimpleError, StageChangeError } from "../common/errors/MSErrors.ts"; -import {isErrorLike, serializeError} from 'serialize-error'; +import { serializeError} from 'serialize-error'; import { DEFAULT_NEW_PADDING, groupPlaysToTimeRanges } from "../utils/ListenFetchUtils.ts"; -import { spawn, isAbortError, delay, waitForEvent, type AbortError } from 'abort-controller-x'; +import { spawn, isAbortError, delay, waitForEvent } from 'abort-controller-x'; import { DrizzlePlayRepository, playToRepositoryCreatePlayOpts, type QueryPlaysOpts, type WithPlayRelation } from "../common/database/drizzle/repositories/PlayRepository.ts"; import type {PlayEventNew, PlayEventSelect, PlaySelect, PlaySelectWithQueueStates, PlayWith, QueueStateSelect} from "../common/database/drizzle/drizzleTypes.ts"; import { asPlay } from "../../core/PlayMarshalUtils.ts"; @@ -73,7 +72,7 @@ import assert from "node:assert"; import { COMPONENT_STATE, type ComponentClientApiJson, type PlayApiCommonDetailed, type QueueStateApi } from "../../core/Api.ts"; import type {ComponentState} from "react"; import { DrizzlePlayEventsRepository } from "../common/database/drizzle/repositories/PlayEventsRepository.ts"; -import { type PlayEvent } from "../../core/PlayEvent.ts"; +import { PLAY_EVENT_TYPE, type PlayEvent } from "../../core/PlayEvent.ts"; import { dupeCheckToPlayEvent, entityIsPlayEntity, queueStateToPlayEvent, scrobbleToPlayEvent, stateChangeToPlayEvent, transformToPlayEvent } from "../common/database/drizzle/entityUtils.ts"; import type { PlayProcessingResult } from "../common/infrastructure/PlayProcessing.ts"; import { PlayProcessingError } from "../common/errors/PlayProcessingError.ts"; @@ -493,8 +492,8 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i queued: this.queuedLength, tracksScrobbled: this.tracksScrobbled, countLive: this.tracksScrobbledTotal, - deadLetterScrobbles: this.deadLetterQueued, - deadLetterScrobblesTotal: this.deadLetterLength, + deadLetterPlays: this.deadLetterQueued, + deadLetterPlaysTotal: this.deadLetterLength, supportsNowPlaying: this.supportsNowPlaying, players: {...this.getNowPlayingPlayers()} } @@ -1047,7 +1046,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i err: Error; try { res = await this.processPlay(playEntity, signal); - } catch (e: unknown | Error | AbortError | PlayProcessingError) { + } catch (e: unknown | Error | PlayProcessingError) { if(isAbortError(e)) { err = generateLoggableAbortReason('Interrupted by abort signal', this.scrobbleQueueAbortController.signal); throw e; @@ -1080,8 +1079,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i this.deadLetterGauge.labels(this.getPrometheusLabels()).inc(); this.deadLetterLength += 1; this.deadLetterQueued += 1; - } else if(res.queue.retries > this.getDefaultDeadLetterRetries()) { - this.deadLetterQueued -= 1; + this.emitEvent('deadLetter', res.playEntity); } queueStates = res.playEntity.queueStates.filter(x => x.queueName !== res.queue.queueName).concat([res.queue]); } else { @@ -1090,12 +1088,17 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i this.deadLetterGauge.labels(this.getPrometheusLabels()).dec(); this.deadLetterLength -= 1; this.deadLetterQueued -= 1; + this.emitEvent('deadLetterRemoved', res.playEntity); } queueStates = res.playEntity.queueStates.filter(x => x.queueName !== res.queue.queueName) } if(initialRetries === 0) { this.queuedGauge.labels(this.getPrometheusLabels()).dec(); this.queuedLength -= 1; + this.emitEvent('playDequeued', { queuedScrobble: playEntity }); + } else { + this.emitEvent('deadLetterDequeued', res.playEntity); + this.deadLetterQueued -= 1; } this.playRepo.updateById(playEntity.id, {play: res.playEntity.play, state: res.playEntity.state, error: res.playEntity.error}); const createdEvents = await this.playEventsRepo.createMany(res.events.map(x => ({...x, playId: playEntity.id}))) as PlayEventSelect[]; @@ -1104,7 +1107,6 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i events: ((res.playEntity as unknown as PlayWith<'events'>).events ?? []).concat(createdEvents), queueStates } as unknown as PlayApiCommonDetailed); - this.emitEvent('scrobbleDequeued', { queuedScrobble: playEntity }); } } @@ -1205,14 +1207,13 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i this.setStatus(`Processing ${isDead ? 'Dead ' : ''}Play ${playEntity.uid}`); let processError: Error | undefined; - let successState: PlaySelect['state']; const events: Omit[] = []; try { if (!isRetry && !playEntity.play.meta.wasMonitored) { - successState = 'discarded'; this.logger.debug(`Not processing ${buildTrackString(playEntity.play)} because monitoring was disabled when Play was queued.`); events.push(stateChangeToPlayEvent({ state: 'discarded', reason: 'Monitoring was disabled when Play was queued' })); + events.push(queueStateToPlayEvent({...queueState, queueStatus: 'completed'})); playEntity.state = 'discarded'; return {playEntity, queue: queueState, events}; } @@ -1310,10 +1311,11 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i } } } else { - successState = 'duped'; this.setStatus(`Play ${playEntity.id} detected as dupe`); this.scrobbleRetries = 0; playEntity.state = 'duped'; + events.push(stateChangeToPlayEvent({state: 'duped'})); + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED})); return {playEntity, events, queue: queueState}; } } catch (e) { @@ -1325,6 +1327,13 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: generateLoggableAbortReason('Interrupted by abort signal', this.scrobbleQueueAbortController.signal)})); throw e; } + if(!events.some(x => x.eventName === PLAY_EVENT_TYPE.playStateChange)) { + events.push(stateChangeToPlayEvent({state: 'failed'})); + playEntity.state = 'failed'; + } + if(!events.some(x => x.eventName === PLAY_EVENT_TYPE.queueStateChange)) { + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: e})); + } throw new PlayProcessingError(e, {playEntity, queue: queueState, events, showStopping: true}); } } @@ -1415,6 +1424,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i (playSelect as unknown as PlayWith<'events'>).events = events; } this.emitPlayUpdate({ ...playSelect } as unknown as PlayApiCommonDetailed); + this.emitEvent(queue.retries > 0 ? 'deadQueued' : 'playQueued', {queuedPlay: playSelect}); createdQueuedPlays.push(playSelect); } } else if (dataArray.every(x => isPlayObject(x))) { @@ -1483,7 +1493,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i this.setStatus(`Added Play from parent ${play.uid} to queue`); const queuedPlay = {id: nanoid(), source: meta.source, play: play} - this.emitEvent('scrobbleQueued', {queuedPlay: queuedPlay}); + this.emitEvent('playQueued', {queuedPlay: queuedPlay}); this.emitPlayInsert({...playRow[0], queueStates: [queueState], events: createdEvents} as unknown as PlayApiCommonDetailed); this.queuedLength += 1; this.queuedGauge.labels(this.getPrometheusLabels()).inc(); @@ -1494,34 +1504,6 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i return createdQueuedPlays; } - addDeadLetterScrobble = async (data: PlaySelect, error: (Error | string) = 'Unspecified error'): Promise => { - let e: ErrorLike; - if(isErrorLike(error)) - { - e = error; - } else if(typeof error === 'string') { - e = new Error(error); - } - this.deadLetterLength += 1; - this.deadLetterQueued += 1; - //this.playRepo.updateById(data.id, {state: 'failed', error: e}); - const newQueue = await this.queueRepo.create({ - componentId: this.dbComponent.id, - playId: data.id, - queueName: DEAD_QUEUE - }) as QueueStateSelect; - await this.playEventsRepo.createMany([ - {playId: data.id, ...queueStateToPlayEvent(newQueue), createdAt: newQueue.createdAt} - ]); - const deadData = {id: nanoid(), retries: 0, error: e, play: data.play}; - //this.deadLetterScrobbles.push(deadData); - //this.deadLetterScrobbles.sort((a, b) => sortByOldestPlayDate(a.play, b.play)); - this.emitEvent('deadLetter', {dead: deadData}); - this.setStatus(`Moved ${data.uid} to Dead Play queue`); - this.deadLetterGauge.labels(this.getPrometheusLabels()).inc(); - return newQueue; - } - queuePlayingNow = async (data: SourcePlayerObj, source: SourceIdentifier) => { if(!this.isReady()) { this.logger.debug('Not queueing now playing because scrobbler is not ready'); diff --git a/src/backend/sources/AbstractSource.ts b/src/backend/sources/AbstractSource.ts index 20bc26c8..dc3b779b 100644 --- a/src/backend/sources/AbstractSource.ts +++ b/src/backend/sources/AbstractSource.ts @@ -2,7 +2,7 @@ import { childLogger, type LogDataPretty, type LogLevel } from '@foxxmd/logging' import dayjs, { type Dayjs } from "dayjs"; import type { EventEmitter } from "events"; import type { FixedSizeList } from "fixed-size-list"; -import { INGRESS_QUEUE, isPlayObject, PARSED_FROM, PLAY_STATES, type PlayMatchResult, type PlayObject, QUEUE_STATUS_COMPLETED, SOURCE_SOT } from "../../core/Atomic.ts"; +import { DEAD_LETTER_RETRIES_DEFAULT, INGRESS_QUEUE, isPlayObject, PARSED_FROM, type PlayMatchResult, type PlayObject, QUEUE_STATUS_COMPLETED, QUEUE_STATUS_FAILED, SOURCE_SOT } from "../../core/Atomic.ts"; import { buildTrackString, capitalize, truncateStringToLength } from "../../core/StringUtils.ts"; import AbstractComponent from "../common/AbstractComponent.ts"; import { @@ -14,7 +14,7 @@ import { type InternalConfig, type ProgressAwarePlayObject, } from "../common/infrastructure/Atomic.ts"; -import type {PARSED_FROM_TYPE, PlayState, PlayUserId, QueueContext} from '../../core/Atomic.ts'; +import type {PARSED_FROM_TYPE, PlayUserId, QueueContext} from '../../core/Atomic.ts'; import type {DeviceId} from '../../core/Atomic.ts'; import type {SourceConfig} from '../common/infrastructure/config/source/sources.ts'; import type {SourceType} from "../../core/Atomic.ts"; @@ -40,18 +40,20 @@ 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'; +import { spawn, isAbortError, delay, throwIfAborted, waitForEvent } from 'abort-controller-x'; import { generateLoggableAbortReason, SimpleError, StageChangeError } from '../common/errors/MSErrors.ts'; import { DrizzlePlayRepository, playToRepositoryCreatePlayOpts, type QueryPlaysOpts, type RequestPlayQuery, type WithPlayRelation } from '../common/database/drizzle/repositories/PlayRepository.ts'; 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 { PlayEventSelect, PlaySelect, PlaySelectWithQueueStates, PlayWith, QueueStateNew, QueueStateSelect } from '../common/database/drizzle/drizzleTypes.ts'; +import type {PaginatedResponse, QueueStateApi} from "../../core/Api.ts"; +import type { PlayEventNew, PlayEventSelect, PlaySelect, PlaySelectWithQueueStates, PlayWith, QueueStateSelect } from '../common/database/drizzle/drizzleTypes.ts'; import { DrizzleQueueRepository } from '../common/database/drizzle/repositories/QueueRepository.ts'; import { DrizzlePlayEventsRepository } from '../common/database/drizzle/repositories/PlayEventsRepository.ts'; import { PLAY_EVENT_TYPE, type PlayEvent } from '../../core/PlayEvent.ts'; import { dupeCheckToPlayEvent, entityIsPlayEntity, queueStateToPlayEvent, stateChangeToPlayEvent, transformToPlayEvent } from '../common/database/drizzle/entityUtils.ts'; +import type { PlayProcessingResult } from '../common/infrastructure/PlayProcessing.ts'; +import { PlayProcessingError } from '../common/errors/PlayProcessingError.ts'; export interface RecentlyPlayedOptions { limit?: number @@ -80,6 +82,8 @@ export default abstract class AbstractSource extends AbstractComponent implement canBacklog: boolean = false; protected discoverQueueAbortController: AbortController | undefined; protected discoverQueuePromise: Promise | undefined; + protected deadQueueAbortController: AbortController | undefined; + protected deadQueuePromise: Promise | undefined; protected abortController: AbortController | undefined; protected pollingPromise: Promise | undefined; stopPollingWaitInterval: number = 200; @@ -87,6 +91,8 @@ export default abstract class AbstractSource extends AbstractComponent implement tracksDiscovered: number = 0; tracksDiscoveredTotal: number = 0; queuedLength: number = 0; + deadLetterLength: number = 0; + deadLetterQueued: number = 0; queueIdleMs: number = 1000; queueConcurrency: number = 3; @@ -99,6 +105,7 @@ export default abstract class AbstractSource extends AbstractComponent implement supportsManualListening: boolean = false; scheduler: ToadScheduler = new ToadScheduler(); + protected initDeadTimeout: NodeJS.Timeout | undefined; protected SCROBBLE_BACKLOG_COUNT: number = 30; @@ -107,6 +114,8 @@ export default abstract class AbstractSource extends AbstractComponent implement protected loggerLabel: string; protected discoveredCounter: Counter; + protected queuedGauge: Gauge; + protected deadLetterGauge: Gauge; declare protected componentType: 'source'; @@ -114,8 +123,6 @@ export default abstract class AbstractSource extends AbstractComponent implement protected queueRepo!: DrizzleQueueRepository; protected playEventsRepo!: DrizzlePlayEventsRepository; - protected queuedGauge: Gauge; - existingDiscoveredPlay: (playObjPre: PlayObject, existingScrobbles: PlayObject[], log?: boolean) => Promise constructor(type: SourceType, name: string, config: SourceConfig, internal: InternalConfig, emitter: EventEmitter) { @@ -138,6 +145,7 @@ export default abstract class AbstractSource extends AbstractComponent implement const metrics = getRoot().items.sourceMetics; this.discoveredCounter = metrics.discovered; this.queuedGauge = metrics.queued; + this.deadLetterGauge = metrics.deadLetter; const existingScrobbleOpts: ExistingScrobbleOpts = { logger: this.logger, @@ -164,7 +172,7 @@ export default abstract class AbstractSource extends AbstractComponent implement } } - public initTasks() { + public initTasks(opts: {deadDelay?: number} = {}) { if(this.scheduler.existsById('heartbeat') === false) { this.logger.info('Adding Heartbeat Task and running immediately'); this.scheduler.addSimpleIntervalJob(new SimpleIntervalJob({ @@ -188,6 +196,41 @@ export default abstract class AbstractSource extends AbstractComponent implement const j = this.scheduler.getById('heartbeat') as SimpleIntervalJob; j.start(); } + + if(this.scheduler.existsById('dead') === false && this.initDeadTimeout === undefined) { + const deadDelay = opts.deadDelay ?? 120; + this.logger.verbose(`Delaying Dead Scrobbler Processing Task by ${deadDelay} seconds`); + this.initDeadTimeout = setTimeout(() => { + this.logger.info('Adding Dead Scrobbler Processing Task and running immediately'); + this.initDeadTimeout = undefined; + this.scheduler.addSimpleIntervalJob(new SimpleIntervalJob({ + minutes: 20, + runImmediately: true + }, new AsyncTask( + 'Dead', + (): Promise => { + if(this.isReady()) { + return this.processDeadLetterQueue().then(() => null).catch((e) => { + this.warnings = e; + this.logger.error(e); + }) + } + return new Promise((resolve, reject) => resolve); + }, + (err: Error) => { + this.warnings.push(err); + this.logger.error(err); + } + ), {id: 'dead'})); + }, deadDelay * 1000); + + } else { + if(this.initDeadTimeout !== undefined) { + this.logger.verbose('Dead scrobble task timeout is already set'); + } else { + this.logger.verbose('Dead scrobble task is already added to the scheduler'); + } + } } protected async heartbeatTask(): Promise { @@ -352,6 +395,9 @@ export default abstract class AbstractSource extends AbstractComponent implement status: this.status, players: {}, tracksDiscovered: this.tracksDiscovered, + deadLetterPlays: this.deadLetterQueued, + queued: this.queuedLength, + deadLetterPlaysTotal: this.deadLetterLength, sot: SOURCE_SOT.HISTORY, supportsUpstreamRecentlyPlayed: this.supportsUpstreamRecentlyPlayed, sleeping: this.getIsSleeping(), @@ -399,6 +445,7 @@ export default abstract class AbstractSource extends AbstractComponent implement (playSelect as unknown as PlayWith<'events'>).events = events; } this.emitPlayUpdate({ ...playSelect } as unknown as PlayApiCommonDetailed); + this.emitEvent(queue.retries > 0 ? 'deadQueued' : 'playQueued', {queuedPlay: playSelect}); createdQueuedPlays.push(playSelect); } } else if (dataArray.every(x => isPlayObject(x))) { @@ -458,6 +505,7 @@ export default abstract class AbstractSource extends AbstractComponent implement ]); createdQueuedPlays.push(playRow[0]); this.logger.debug(`Added ${buildTrackString(queueablePlay)} to the queue`); + this.emitEvent('playQueued', {queuedPlay: queueablePlay}); this.emitPlayInsert({ ...playRow[0], queueStates: [queueState] } as unknown as PlayApiCommonDetailed); this.queuedLength += 1; this.queuedGauge.labels(this.getPrometheusLabels()).inc(); @@ -965,7 +1013,7 @@ export default abstract class AbstractSource extends AbstractComponent implement this.logger.debug(`Delaying discovery of Play ${item.uid} task for ${delayFor}ms due to non-zero prior failures (${taskFailures})`); await sleep(delayFor, { signal }); } - return this.processQueueCurrentPlay(item, signal) + return this.handlePlayProcessing(item, signal) }, { concurrency: this.queueConcurrency, @@ -999,34 +1047,108 @@ export default abstract class AbstractSource extends AbstractComponent implement } - protected processQueueCurrentPlay = async (currQueuedPlay: PlaySelectWithQueueStates, signal?: AbortSignal) => { + protected handlePlayProcessing = async (playEntity: PlaySelectWithQueueStates, signal?: AbortSignal) => { + let res: PlayProcessingResult, + err: Error; + try { + res = await this.processQueueCurrentPlay(playEntity, signal); + } catch (e: unknown | Error | PlayProcessingError) { + if(isAbortError(e)) { + err = generateLoggableAbortReason('Interrupted by abort signal', this.discoverQueueAbortController.signal); + throw e; + } + if(e instanceof PlayProcessingError) { + err = e.cause as Error; + res = e.result; + if(e.showStopping) { + throw e.cause; + } + } else { + const unhandledError = new Error('Unhandled error type while processing Play', {cause: e}); + if(e instanceof Error) { + err = e; + } else { + err = unhandledError; + } + throw e; + } + } finally { + let queueStates: QueueStateSelect[]; + const initialRetries = res.queue.retries ?? 0; + if(err !== undefined) { + res.queue.retries = (initialRetries + 1); + res.queue.updatedAt = dayjs(); + await this.queueRepo.updateById(res.queue.id, { + ...res.queue, + }); + if(initialRetries === 0) { + this.deadLetterGauge.labels(this.getPrometheusLabels()).inc(); + this.deadLetterLength += 1; + this.deadLetterQueued += 1; + this.emitEvent('deadLetter', res.playEntity); + } + queueStates = res.playEntity.queueStates.filter(x => x.queueName !== res.queue.queueName).concat([res.queue]); + } else { + await this.queueRepo.deleteByIds([res.queue.id]); + if(res.queue.retries > 0) { + this.deadLetterGauge.labels(this.getPrometheusLabels()).dec(); + this.deadLetterLength -= 1; + this.deadLetterQueued -= 1; + } + queueStates = res.playEntity.queueStates.filter(x => x.queueName !== res.queue.queueName); + } + if(initialRetries === 0) { + this.queuedGauge.labels(this.getPrometheusLabels()).dec(); + this.queuedLength -= 1; + this.emitEvent('playDequeued', { queuedScrobble: playEntity }); + } else { + this.emitEvent('deadLetterDequeued', res.playEntity); + this.deadLetterQueued -= 1; + } + this.playRepo.updateById(playEntity.id, {play: res.playEntity.play, state: res.playEntity.state, error: res.playEntity.error}); + const createdEvents = await this.playEventsRepo.createMany(res.events.map(x => ({...x, playId: playEntity.id}))) as PlayEventSelect[]; + this.emitPlayUpdate({ + ...res.playEntity, + events: ((res.playEntity as unknown as PlayWith<'events'>).events ?? []).concat(createdEvents), + queueStates + } as unknown as PlayApiCommonDetailed); + this.emitEvent('playDequeued', { queuedScrobble: playEntity }); + } + } + + protected getDefaultDeadLetterRetries() { + return this.config.options?.deadLetterRetries ?? DEAD_LETTER_RETRIES_DEFAULT; + } + + protected processQueueCurrentPlay = async (playEntity: PlaySelectWithQueueStates, signal?: AbortSignal): Promise => { signal?.throwIfAborted(); - this.setStatus(`Processing Play ${currQueuedPlay.uid}`); + this.setStatus(`Processing Play ${playEntity.uid}`); - const queueState = currQueuedPlay.queueStates.find(x => x.queueName === INGRESS_QUEUE); + const queueState = playEntity.queueStates.find(x => x.queueName === INGRESS_QUEUE); const { useCache = true, isRetry = false, transform = true, dupeCheck = true, } = queueState.context || {}; - const updatedQueueState: Partial = {}; - let state: PlayState; - let events: Omit[] = []; + + const isDead = queueState.retries > 0 || isRetry; + const logger = isDead ? childLogger(this.logger, ['Dead', `Play ${playEntity.uid}`]) : childLogger(this.logger, [`Play ${playEntity.uid}`]); + this.setStatus(`Processing ${isDead ? 'Dead ' : ''}Play ${playEntity.uid}`); + + const events: Omit[] = []; try { - if(isRetry !== true && !currQueuedPlay.play.meta.wasMonitored) { - this.logger.debug(`Not processing ${buildTrackString(currQueuedPlay.play)} because monitoring was disabled when Play was queued.`); - state = 'discarded'; - events.push(stateChangeToPlayEvent({state, reason: 'Not processing because monitoring was disabled when Play was queued'})); - updatedQueueState.queueStatus = QUEUE_STATUS_COMPLETED; - events.push(queueStateToPlayEvent({...queueState, ...updatedQueueState})); - this.playRepo.updateById(currQueuedPlay.id, {state}); - return; + if(isRetry !== true && !playEntity.play.meta.wasMonitored) { + logger.debug(`Not processing ${buildTrackString(playEntity.play)} because monitoring was disabled when Play was queued.`); + playEntity.state = 'discarded'; + events.push(stateChangeToPlayEvent({state: playEntity.state, reason: 'Not processing because monitoring was disabled when Play was queued'})); + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED})); + return {playEntity, queue: queueState, events}; } - let preCompared = currQueuedPlay.play; + let preCompared = playEntity.play; if(transform) { - const {lifecycle = [], ...rest} = await this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.preCompare, {useCachedResult: useCache}); + const {lifecycle = [], ...rest} = await this.transformPlay(playEntity.play, TRANSFORM_HOOK.preCompare, {useCachedResult: useCache}); preCompared = rest; if(lifecycle.length > 0) { events.push({...transformToPlayEvent(lifecycle), createdAt: dayjs()}); @@ -1035,75 +1157,183 @@ export default abstract class AbstractSource extends AbstractComponent implement let existing: PlayObject; if (dupeCheck) { // cheap check for existing - const cheapExisting = await this.playRepo.checkExisting(preCompared, { notId: currQueuedPlay.id }); + const cheapExisting = await this.playRepo.checkExisting(preCompared, { notId: playEntity.id }); if (cheapExisting !== undefined) { events.push(dupeCheckToPlayEvent({ match: true, reason: `Matched hash on existing Play ${cheapExisting.uid} with close temporality` })); - updatedQueueState.error = { message: `Matched hash on existing Play ${cheapExisting.uid} with close temporality` }; existing = { ...cheapExisting.play, id: cheapExisting.id, uid: cheapExisting.uid }; } else { const matchRes = await this.existingDiscovered(preCompared); events.push(dupeCheckToPlayEvent(matchRes)); if (matchRes.match) { existing = matchRes.closestMatchedPlay; - updatedQueueState.error = { message: `Matched with Play ${existing.uid ?? existing.id}` }; } } } - currQueuedPlay.play = preCompared; + playEntity.play = preCompared; signal?.throwIfAborted(); if(existing === undefined) { - state = 'discovered'; - events.push(stateChangeToPlayEvent({state})); + playEntity.state = 'discovered'; + //state = 'discovered'; + events.push(stateChangeToPlayEvent({state: 'discovered'})); this.tracksDiscovered++; this.tracksDiscoveredTotal++ this.discoveredCounter.labels(this.getPrometheusLabels()).inc(); this.emitEvent('discovered', {play: preCompared}); + await this.scrobble([{...playEntity.play, id: playEntity.id, uid: playEntity.uid}]); } else { - this.playRepo.updateById(existing.id, {updatedAt: dayjs()}); - state = 'duped'; - events.push(stateChangeToPlayEvent({state})); - currQueuedPlay.parentId = existing.id; + await this.playRepo.updateById(existing.id, {updatedAt: dayjs()}); + playEntity.state = 'duped'; + events.push(stateChangeToPlayEvent({state: 'duped'})); + playEntity.parentId = existing.id; } const recentPlays = await this.getRecentPlays(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({...preCompared, id: currQueuedPlay.id, uid: currQueuedPlay.uid}); + recentPlays.push({...preCompared, id: playEntity.id, uid: playEntity.uid}); recentPlays.sort(sortByOldestPlayDate); this.cache.cacheDb.set(this.recentCacheKey(), recentPlays, '2m'); } - if(state === 'discovered') { + if(playEntity.state === 'discovered') { const recentDiscoveredPlays = await this.getRecentlyDiscoveredPlays(false); if(recentDiscoveredPlays !== undefined) { - recentDiscoveredPlays.push({...preCompared, id: currQueuedPlay.id, uid: currQueuedPlay.uid}); + recentDiscoveredPlays.push({...preCompared, id: playEntity.id, uid: playEntity.uid}); recentDiscoveredPlays.sort(sortByOldestPlayDate); this.cache.cacheDb.set(this.recentDiscoveredCacheKey(), recentDiscoveredPlays, '2m'); } } - updatedQueueState.queueStatus = 'completed'; - events.push(queueStateToPlayEvent({...queueState, ...updatedQueueState})); - - this.playRepo.updateById(currQueuedPlay.id, {play: preCompared, state}); - this.logger.info(`${capitalize(state)} => ${buildTrackString(preCompared)}`); + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED})); + logger.info(`${capitalize(playEntity.state)} => ${buildTrackString(preCompared)}`); + return {playEntity, events, queue: queueState}; } catch (e) { - const err = new Error(`Error ocurred while trying to discover Play ${currQueuedPlay.uid}`, {cause: e}); - updatedQueueState.error = err; - updatedQueueState.queueStatus = 'failed'; - events = events.filter(x => x.eventName !== PLAY_EVENT_TYPE.playStateChange); - events.push(stateChangeToPlayEvent({state: 'failed'})); - events.push(queueStateToPlayEvent({...queueState, ...updatedQueueState})); - this.playRepo.updateById(currQueuedPlay.id, {state: 'failed', error: err}); - } finally { - await this.queueRepo.updateById(queueState.id, updatedQueueState); - const createdEvents = await this.playEventsRepo.createMany(events.map(x => ({...x, playId: currQueuedPlay.id}))); - this.emitPlayUpdate({...currQueuedPlay, events: createdEvents, queueStates: [{...queueState, ...updatedQueueState}]} as unknown as PlayApiCommonDetailed); + if(e instanceof PlayProcessingError) { + throw e; + } + if(isAbortError(e)) { + events.push(stateChangeToPlayEvent({state: 'failed'})); + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: generateLoggableAbortReason('Interrupted by abort signal', this.discoverQueueAbortController.signal)})); + throw e; + } + if(!events.some(x => x.eventName === PLAY_EVENT_TYPE.playStateChange)) { + events.push(stateChangeToPlayEvent({state: 'failed'})); + playEntity.state = 'failed'; + } + if(!events.some(x => x.eventName === PLAY_EVENT_TYPE.queueStateChange)) { + events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: e})); + } + throw new PlayProcessingError(e, {playEntity, queue: queueState, events, showStopping: true}); + } + } + + removeDeadLetterScrobble = async (dead: PlaySelectWithQueueStates) => { + + const queueState = dead.queueStates.find(x => x.queueName === INGRESS_QUEUE); + if(queueState === undefined) { + this.logger.warn(`Play ${dead.uid} does not have a dead state, nothing to remove.`); + return; + } + if(queueState.retries === 0) { + this.logger.warn(`Play ${dead.uid} has not failed yet, not removing.`); + return; + } + + this.setStatus(`Marking Dead Play ${dead.uid} as completed`); + + const events: PlayEventNew[] = [ + { playId: dead.id, ...queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED, context: {reason: 'Dead Play marked as completed by user'}}) } + ]; + await this.queueRepo.deleteByIds([queueState.id]); + + if(dead.state === 'queued') { + dead.state = 'failed'; + this.playRepo.updateById(dead.id, {state: 'failed'}); + events.push( + { playId: dead.id, ...stateChangeToPlayEvent({ state: 'failed' }) } + ) } + const createdEvents = await this.playEventsRepo.createMany(events) as PlayEventSelect[]; + this.emitPlayUpdate({uid: dead.uid, + state: dead.state, + queueStates: dead.queueStates.filter(x => x.queueName !== INGRESS_QUEUE) as unknown as QueueStateApi[], + events: createdEvents as unknown as PlayEvent[] + }); + + this.deadLetterLength -= 1; + if(queueState.queueStatus === 'queued') { + this.deadLetterQueued -= 1; + } + this.deadLetterGauge.labels(this.getPrometheusLabels()).dec(); + this.emitEvent('removeDeadLetter', { dead: { id: dead.uid } }); + } - if(state === 'discovered') { - await this.scrobble([{...currQueuedPlay.play, id: currQueuedPlay.id, uid: currQueuedPlay.uid}]); + processDeadLetterQueue = async (attemptWithRetries?: number, reason?: string, sync?: boolean) => { + + const logger = childLogger(this.logger, ['Dead']); + if (!(await this.isReady())) { + logger.warn('Cannot process dead letter scrobbles because client is not ready.'); + return; } - return currQueuedPlay; + if(this.deadQueueAbortController !== undefined) { + logger.warn('Dead scrobbles are currently being processed, cannot restart right now.'); + return; + } + + const { + options: { + deadLetterRetries = 3 + } = {} + } = this.config; + + const retries = attemptWithRetries ?? deadLetterRetries; + + this.deadQueueAbortController = new AbortController(); + this.deadQueuePromise = spawn(this.deadQueueAbortController.signal, async (signal, { defer, fork }) => { + + //const processable = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE], [QUEUE_STATUS_FAILED], retries); + const processableArgs: QueryPlaysOpts = {queues: [{queueName: INGRESS_QUEUE, queueStatus: QUEUE_STATUS_FAILED, retries}], with: ['queues']}; + let processable = await this.playRepo.findPlaysPaginated(processableArgs); + this.deadLetterQueued = processable.meta.total; + + const total = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE], {queueStatus: [QUEUE_STATUS_FAILED], retries: 10000}); + this.deadLetterLength = total; + const queueStatus = `${processable.meta.total} of ${total} dead Plays have less than ${retries} retries, ${processable.meta.total === 0 ? 'will skip processing.': 'processing now...'}`; + if (processable.meta.total === 0) { + logger.verbose(queueStatus); + return; + } + this.setStatus(`Queuing ${processable} Dead Plays...`); + logger.info(queueStatus); + let more = true; + let offset = 0; + while(more) { + await this.queuePlay(processable.data, {reason}); + more = processable.data.length === processable.meta.limit; + if(more) { + offset += processable.meta.limit; + processable = await this.playRepo.findPlaysPaginated({...processableArgs, offset}); + } + } + this.setStatus(`All processable Dead Plays have been queued`); + logger.info(`All processable Dead Plays have been queued`); + + if(sync) { + await waitForEvent(signal,this.emitter,'queueEmptied'); + this.setStatus(`Finished processing Dead Plays`); + logger.info('Finished processing Dead Plays'); + } + }).catch((e) => { + if (isAbortError(e)) { + const err = generateLoggableAbortReason('Dead scrobble processing stopped', this.deadQueueAbortController.signal); + this.logger.info(err); + logger.trace(e) + } else { + logger.warn(new Error('Dead scrobble processing stopped with error', { cause: e })); + } + }).finally(() => { + this.deadQueueAbortController = undefined; + this.deadQueuePromise = undefined; + }); } protected setIsSleeping(sleeping: boolean) { diff --git a/src/backend/utils/AsyncUtils.ts b/src/backend/utils/AsyncUtils.ts index 17e673b0..0d5cbb6a 100644 --- a/src/backend/utils/AsyncUtils.ts +++ b/src/backend/utils/AsyncUtils.ts @@ -107,7 +107,7 @@ export const consumeQueueOnce = async (next: () => Promise, pr export const consumeQueue = async ( next: (queueId: string) => Promise, - process: (item: T, queueId: string) => Promise, + process: (item: T, queueId: string) => Promise, opts: { concurrency: number; idleMs: number; diff --git a/src/client/components/msComponent/MSComponentDetailed.tsx b/src/client/components/msComponent/MSComponentDetailed.tsx index 9d2abf1f..5293b692 100644 --- a/src/client/components/msComponent/MSComponentDetailed.tsx +++ b/src/client/components/msComponent/MSComponentDetailed.tsx @@ -62,12 +62,11 @@ export const MSComponentStats = (props: { data?: ComponentCommonApiJson, live?: ) } - const isClient = isComponentClientApiJson(props.data); return ( - {isClient ? : null} - {isClient ? : null} + + ) diff --git a/src/client/components/msComponent/MSComponentSummary.tsx b/src/client/components/msComponent/MSComponentSummary.tsx index 64189933..95c9137f 100644 --- a/src/client/components/msComponent/MSComponentSummary.tsx +++ b/src/client/components/msComponent/MSComponentSummary.tsx @@ -28,7 +28,6 @@ export const MSComponentSummary = (props: { data: ComponentCommonApiJson, fetcha } = props; let sleepingRender: React.JSX.Element = null; - let body = ; const cardHeaderProps: Card.HeaderProps = {}; const isClient = isComponentClientApiJson(data); if(isComponentSourceApiJson(data)) { @@ -39,7 +38,7 @@ export const MSComponentSummary = (props: { data: ComponentCommonApiJson, fetcha sleepingRender = ; } } - body = ( + const body = ( {fetchable ? : } ); @@ -77,28 +76,7 @@ export const MSComponentSummary = (props: { data: ComponentCommonApiJson, fetcha // color={data.mode === 'client' ? 'purple' : 'pink'} const QuickStatsSource = (props: { data: ComponentCommonApiJson, streamable?: boolean }) => { - if (isComponentSourceApiJson(props.data)) { - const { - tracksDiscovered, - countLive - } = props.data; - return ( - - - {/* {tracksDiscovered} Discovered */} - - - - ) - } else if (isComponentClientApiJson(props.data)) { - const { - queued, - deadLetterScrobbles, - deadLetterScrobblesTotal, - countLive, - } = props.data; - - return ( + return ( @@ -111,7 +89,6 @@ const QuickStatsSource = (props: { data: ComponentCommonApiJson, streamable?: bo ) - } } export const MSComponentSummaryFetchable = (props: {componentId: number, data: ComponentCommonApiJson}) => { diff --git a/src/client/components/msComponent/Stats.tsx b/src/client/components/msComponent/Stats.tsx index 85b35585..5d6b03e3 100644 --- a/src/client/components/msComponent/Stats.tsx +++ b/src/client/components/msComponent/Stats.tsx @@ -112,9 +112,9 @@ export const QueuedIndicator = (props: { const client = useSSEContext(); useSSEAnyEvent(client, (payload) => { if ('componentId' in (payload.data as object) && (payload.data as Record).componentId === props.data.id) { + recentTimeout.stop(); switch (payload.type) { - case 'scrobbleQueued': - recentTimeout.stop(); + case 'playQueued': setCurrent(current + 1); if (recentDirection === 'down') { setRecent(1); @@ -122,10 +122,8 @@ export const QueuedIndicator = (props: { setRecent(recent + 1); } setRecentDirection('up'); - recentTimeout.start(); break; - case 'scrobbleDequeued': - recentTimeout.stop(); + case 'playDequeued': setCurrent(current - 1); if (recentDirection === 'up') { setRecent(1); @@ -133,9 +131,9 @@ export const QueuedIndicator = (props: { setRecent(recent + 1); } setRecentDirection('down'); - recentTimeout.start(); break; } + recentTimeout.start(); } }); } @@ -180,8 +178,8 @@ export const DeadLetterIndicator = (props: { ...rest } = props; - const [current, setCurrent] = useState(props.data.deadLetterScrobbles); - const [total, setTotal] = useState(props.data.deadLetterScrobblesTotal); + const [current, setCurrent] = useState(props.data.deadLetterPlays); + const [total, setTotal] = useState(props.data.deadLetterPlaysTotal); const [recent, setRecent] = useState(recentProp); const [recentDirection, setRecentDirection] = useState<'up' | 'down'>('up'); const resetRecent = useCallback(() => { @@ -194,6 +192,7 @@ export const DeadLetterIndicator = (props: { const client = useSSEContext(); useSSEAnyEvent(client, (payload) => { if ('componentId' in (payload.data as object) && (payload.data as Record).componentId === props.data.id) { + recentTimeout.stop(); switch (payload.type) { case 'deadLetter': recentTimeout.stop(); @@ -205,9 +204,37 @@ export const DeadLetterIndicator = (props: { setRecent(recent + 1); } setRecentDirection('up'); - recentTimeout.start(); break; + case 'deadLetterRemoved': + setCurrent(current - 1); + setTotal(total - 1); + if (recentDirection === 'down') { + setRecent(recent + 1); + } else { + setRecent(1); + } + setRecentDirection('down'); + break; + case 'deadLetterDequeued': + setCurrent(current - 1); + if (recentDirection === 'down') { + setRecent(recent + 1); + } else { + setRecent(1); + } + setRecentDirection('down'); + break; + case 'deadQueued': + setCurrent(current + 1); + if (recentDirection === 'down') { + setRecent(1); + } else { + setRecent(recent + 1); + } + setRecentDirection('up'); + break; } + recentTimeout.start(); } }); } diff --git a/src/core/Api.ts b/src/core/Api.ts index f9df5fc9..d133ad5c 100644 --- a/src/core/Api.ts +++ b/src/core/Api.ts @@ -86,6 +86,9 @@ export type ComponentCommonApi = { errors?: ErrorIsh[] warnings?: ErrorIsh[] monitoringStatus?: MonitoringStatus + deadLetterPlays: number + deadLetterPlaysTotal: number + queued: number } & Omit export type ComponentCommonApiJson = Replace, string>; @@ -99,10 +102,7 @@ export type ComponentDetailedApi = ComponentCommonApi & { } export type ComponentCientApiBase = { - queued: number tracksScrobbled: number - deadLetterScrobbles: number - deadLetterScrobblesTotal: number supportsNowPlaying: boolean players: Record } diff --git a/src/core/tests/utils/apiFixtures.ts b/src/core/tests/utils/apiFixtures.ts index 9f26a1ad..d038cf5e 100644 --- a/src/core/tests/utils/apiFixtures.ts +++ b/src/core/tests/utils/apiFixtures.ts @@ -117,6 +117,9 @@ export const generateComponentCommonApiJson = (data: Partial state = faker.number.int({min: 1, max: 7}) as ComponentState, monitoringStatus = { monitoring: faker.datatype.boolean({probability: 0.1}), origin: 'system' }, players = {}, + queued = faker.number.int({min: 1, max: 2000}), + deadLetterPlays = faker.number.int({min: 1, max: 2000}), + deadLetterPlaysTotal = faker.number.int({min: deadLetterPlays, max: 2000}), ...rest } = data; @@ -145,6 +148,9 @@ export const generateComponentCommonApiJson = (data: Partial players, status: faker.helpers.arrayElement(statusSamples), monitoringStatus, + queued, + deadLetterPlays, + deadLetterPlaysTotal, ...rest } } @@ -194,17 +200,11 @@ export const generateClientApiJson = (data: Partial = {}): C ...rest }); const { - queued = faker.number.int({min: 1, max: 2000}), - deadLetterScrobbles = faker.number.int({min: 1, max: 2000}), - deadLetterScrobblesTotal = faker.number.int({min: deadLetterScrobbles, max: 2000}), players = (data.players ?? {}), } = data; return { ...common, - queued, tracksScrobbled: common.countLive, - deadLetterScrobbles, - deadLetterScrobblesTotal, players, supportsNowPlaying: Object.keys(players).length > 0, initialized: true, -- 2.51.2 From 18be08279669d2eb49096a7b9905f0f84f06c830 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Fri, 21 Aug 2026 19:55:55 +0000 Subject: [PATCH 4/6] feat: dead queue to ingress migration --- .../appMigrations/003_deadConsolidation.ts | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) create mode 100644 src/backend/common/database/appMigrations/003_deadConsolidation.ts diff --git a/src/backend/common/database/appMigrations/003_deadConsolidation.ts b/src/backend/common/database/appMigrations/003_deadConsolidation.ts new file mode 100644 index 00000000..42760f21 --- /dev/null +++ b/src/backend/common/database/appMigrations/003_deadConsolidation.ts @@ -0,0 +1,76 @@ +import type { SqliteDatabase, Migration } from 'sqlite-up'; +import type { MigrateBaseContext } from '../appMigrator.ts'; +import { queueStates } from '../drizzle/schema/schema.ts'; +import { eq } from 'drizzle-orm'; + + +export const up: Migration['up'] = async (db: SqliteDatabase, ctx: MigrateBaseContext): Promise => { + + ctx.logger.info('Beginning queue entities consolidation.'); + + ctx.logger.verbose('Deleting (now) unused completed queue states...'); + + ctx.db.delete(queueStates).where(eq(queueStates.queueStatus,'completed')); + + ctx.logger.verbose('Done with completed queue state deletions'); + + ctx.logger.info('Converting dead queue states to ingress states with failure + retries...'); + + let more = true; + let offset = 0, + updated = 0; + + while (more) { + const plays = await ctx.db.query.plays.findMany({ + where: { + queueStates: { + queueName: 'dead' + } + }, + with: { + queueStates: true + }, + limit: 100, + offset + }); + + for(const p of plays) { + const ingress = p.queueStates.find(x => x.queueName === 'ingress'); + const dead = p.queueStates.find(x => x.queueName === 'dead'); + + if(ingress === undefined) { + await ctx.db.insert(queueStates).values([{ + componentId: p.componentId, + playId: p.id, + queueName: 'ingress', + queueStatus: 'failed', + retries: dead.retries, + error: dead.error, + createdAt: dead.updatedAt, + updatedAt: dead.updatedAt + }]); + } else { + await ctx.db.update(queueStates).set({ + queueStatus: 'failed', + retries: dead.retries, + error: dead.error, + updatedAt: dead.updatedAt}).where(eq(queueStates.id, ingress.id)); + } + updated++; + await ctx.db.delete(queueStates).where(eq(queueStates.id, dead.id)); + } + + offset += 100; + ctx.logger.verbose(`Conversion Progress: Updated ${updated}`); + if (plays.length < 100) { + more = false; + } + } + + ctx.logger.info('Done.'); +}; + +export const down: Migration['down'] = async (db: SqliteDatabase, ctx: MigrateBaseContext): Promise => { + // Rollback code here + // context is passed as ctx +}; \ No newline at end of file -- 2.51.2 From 7847881897c68a24b83c734776a55ffba4491f48 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Sat, 22 Aug 2026 00:28:36 +0000 Subject: [PATCH 5/6] feat: Implement api endpoitns for requeuing, cancelling, and deleting dead states --- .../common/database/drizzle/schema/schema.ts | 2 +- .../scrobblers/AbstractScrobbleClient.ts | 30 +++++++++- src/backend/server/api.ts | 60 ++++++++++++++++++- src/backend/sources/AbstractSource.ts | 32 +++++++++- src/core/Atomic.ts | 15 ++--- 5 files changed, 126 insertions(+), 13 deletions(-) diff --git a/src/backend/common/database/drizzle/schema/schema.ts b/src/backend/common/database/drizzle/schema/schema.ts index 19f507fa..3d3f8f67 100644 --- a/src/backend/common/database/drizzle/schema/schema.ts +++ b/src/backend/common/database/drizzle/schema/schema.ts @@ -144,7 +144,7 @@ export const queueStates = sqliteTable("play_queue_states", { queueStatus: text({enum: ['queued','completed','failed']}).notNull().default('queued'), retries: integer().notNull().default(0), error: ErrorLikeJson('error'), - context: text({mode: 'json'}).$type(), + context: text({mode: 'json'}).$type(), createdAt: DayjsTimestamp('createdAt').notNull().$defaultFn(() => dayjs()), updatedAt: DayjsTimestamp('updatedAt').notNull().$defaultFn(() => dayjs()).$onUpdate(() => dayjs()) }, (table) => [ diff --git a/src/backend/scrobblers/AbstractScrobbleClient.ts b/src/backend/scrobblers/AbstractScrobbleClient.ts index 8f907f69..1eec892e 100644 --- a/src/backend/scrobblers/AbstractScrobbleClient.ts +++ b/src/backend/scrobblers/AbstractScrobbleClient.ts @@ -1399,7 +1399,35 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i await this.updateQueueStats([DEAD_QUEUE]); } - queueScrobble = async (data: (PlayObject | PlayObject[]) | (PlaySelectWithQueueStates | PlaySelectWithQueueStates[]), context?: QueueContext) => { + public cancelQueuedPlay = async (playEntity: PlaySelectWithQueueStates) => { + const queueState = playEntity.queueStates.find(x => x.queueName === INGRESS_QUEUE); + if(queueState === undefined) { + throw new SimpleError('Play does not have an associated queued'); + } + if(queueState.queueStatus !== 'queued') { + throw new SimpleError('Play is not queued'); + } + + queueState.queueStatus = QUEUE_STATUS_FAILED; + playEntity.state = 'failed'; + const createdEvents = await this.playEventsRepo.createMany([ + {playId: playEntity.id, ...stateChangeToPlayEvent({state: playEntity.state})}, + {playId: playEntity.id, ...queueStateToPlayEvent({...queueState, context: {reason: 'Cancelled by user'}})} + ]) as PlayEventSelect[]; + await this.queueRepo.updateById(queueState.id, {queueStatus: QUEUE_STATUS_FAILED}); + await this.playRepo.updateById(playEntity.id, {state: 'failed'}); + this.emitPlayUpdate({ + ...playEntity, + events: ((playEntity as unknown as PlayWith<'events'>).events ?? []).concat(createdEvents), + } as unknown as PlayApiCommonDetailed); + if(queueState.retries === 0) { + this.emitEvent('playDequeued', { queuedScrobble: playEntity }); + } else { + this.emitEvent('deadLetterDequeued', { queuedScrobble: playEntity }); + } + } + + queueScrobble = async (data: (PlayObject | PlayObject[]) | (PlaySelectWithQueueStates | PlaySelectWithQueueStates[]), context?: QueueContext & {isRetry?: boolean}) => { const createdQueuedPlays: PlaySelect[] = []; const dataArray = Array.isArray(data) ? data : [data]; diff --git a/src/backend/server/api.ts b/src/backend/server/api.ts index 484e9a4f..ab1968b8 100644 --- a/src/backend/server/api.ts +++ b/src/backend/server/api.ts @@ -17,12 +17,13 @@ import { type SOURCE_SOT_TYPES, type SourcePlayerJson, type SourceStatusData, + queueContextSchema, } from "../../core/Atomic.ts"; import { capitalize } from "../../core/StringUtils.ts"; import type {ExpressHandler, LeveledLogData} from "../common/infrastructure/Atomic.ts"; import { getRoot } from "../ioc.ts"; import AbstractScrobbleClient from "../scrobblers/AbstractScrobbleClient.ts"; -import type AbstractSource from "../sources/AbstractSource.ts"; +import AbstractSource from "../sources/AbstractSource.ts"; import MemorySource from "../sources/MemorySource.ts"; import { parseBool } from "../utils.ts"; import { sortByNewestPlayDate } from '../../core/PlayUtils.ts'; @@ -412,7 +413,6 @@ export const setupApi = (app: Express, router: ReturnType { const { component, - query, params: { playUid } @@ -424,6 +424,62 @@ export const setupApi = (app: Express, router: ReturnType { + const { + component, + params: { + playUid + }, + body = {} + } = req; + + const play = await component.playRepo.findByUid(playUid); + if(play === undefined) { + return res.sendStatus(404); + } + + if(component instanceof AbstractSource) { + await component.queuePlay([play], {...body, isRetry: true}); + } else { + await component.queueScrobble([play], {...body, isRetry: true}); + } + return res.sendStatus(200); + }); + + router.delete('/components/:componentVal/plays/:playUid/queue', {middleware: [componentAwareMiddle]}, async (req, res, next) => { + const { + component, + params: { + playUid + } + } = req; + + const play = await component.playRepo.findByUid(playUid); + if(play === undefined) { + return res.sendStatus(404); + } + + await component.cancelQueuedPlay(play); + return res.sendStatus(200); + }); + + router.delete('/components/:componentVal/plays/:playUid/dead', {middleware: [componentAwareMiddle]}, async (req, res, next) => { + const { + component, + params: { + playUid + } + } = req; + + const play = await component.playRepo.findByUid(playUid); + if(play === undefined) { + return res.sendStatus(404); + } + + await component.removeDeadLetterScrobble(play); + return res.sendStatus(200); + }); + router.delete('/cache/:cacheType', async (req, res) => { const cache = await getRoot().items.cache(); logger.verbose(`User request cache deletion for ${req.params.cacheType}`); diff --git a/src/backend/sources/AbstractSource.ts b/src/backend/sources/AbstractSource.ts index dc3b779b..89abc1f5 100644 --- a/src/backend/sources/AbstractSource.ts +++ b/src/backend/sources/AbstractSource.ts @@ -119,7 +119,7 @@ export default abstract class AbstractSource extends AbstractComponent implement declare protected componentType: 'source'; - protected playRepo!: DrizzlePlayRepository; + public playRepo!: DrizzlePlayRepository; protected queueRepo!: DrizzleQueueRepository; protected playEventsRepo!: DrizzlePlayEventsRepository; @@ -421,7 +421,7 @@ export default abstract class AbstractSource extends AbstractComponent implement // TODO make this more descriptive? or move it elsewhere recentlyPlayedTrackIsValid = (playObj: PlayObject) => true - queuePlay = async (data: (PlayObject | PlayObject[]) | (PlaySelectWithQueueStates | PlaySelectWithQueueStates[]), context?: QueueContext) => { + queuePlay = async (data: (PlayObject | PlayObject[]) | (PlaySelectWithQueueStates | PlaySelectWithQueueStates[]), context?: QueueContext & {isRetry?: boolean}) => { const createdQueuedPlays: PlaySelect[] = []; const dataArray = Array.isArray(data) ? data : [data]; @@ -1047,6 +1047,34 @@ export default abstract class AbstractSource extends AbstractComponent implement } + public cancelQueuedPlay = async (playEntity: PlaySelectWithQueueStates) => { + const queueState = playEntity.queueStates.find(x => x.queueName === INGRESS_QUEUE); + if(queueState === undefined) { + throw new SimpleError('Play does not have an associated queued'); + } + if(queueState.queueStatus !== 'queued') { + throw new SimpleError('Play is not queued'); + } + + queueState.queueStatus = QUEUE_STATUS_FAILED; + playEntity.state = 'failed'; + const createdEvents = await this.playEventsRepo.createMany([ + {playId: playEntity.id, ...stateChangeToPlayEvent({state: playEntity.state})}, + {playId: playEntity.id, ...queueStateToPlayEvent({...queueState, context: {reason: 'Cancelled by user'}})} + ]) as PlayEventSelect[]; + await this.queueRepo.updateById(queueState.id, {queueStatus: QUEUE_STATUS_FAILED}); + await this.playRepo.updateById(playEntity.id, {state: 'failed'}); + this.emitPlayUpdate({ + ...playEntity, + events: ((playEntity as unknown as PlayWith<'events'>).events ?? []).concat(createdEvents), + } as unknown as PlayApiCommonDetailed); + if(queueState.retries === 0) { + this.emitEvent('playDequeued', { queuedScrobble: playEntity }); + } else { + this.emitEvent('deadLetterDequeued', { queuedScrobble: playEntity }); + } + } + protected handlePlayProcessing = async (playEntity: PlaySelectWithQueueStates, signal?: AbortSignal) => { let res: PlayProcessingResult, err: Error; diff --git a/src/core/Atomic.ts b/src/core/Atomic.ts index 97672e2d..d0beeedf 100644 --- a/src/core/Atomic.ts +++ b/src/core/Atomic.ts @@ -657,13 +657,14 @@ export const QUEUE_STATUSES: QueueStatus[] = [QUEUE_STATUS_COMPLETED, QUEUE_STAT export const DEAD_LETTER_RETRIES_DEFAULT = 3; -export interface QueueContext { - transform?: boolean - dupeCheck?: boolean - useCache?: boolean - isRetry?: boolean - reason?: string -} +export const queueContextSchema = z.object({ + transform: z.boolean().optional(), + dupeCheck: z.boolean().optional(), + useCache: z.boolean().optional(), + reason: z.string().optional() +}); + +export type QueueContext = z.infer; /** * @see https://github.com/ts-essentials/ts-essentials/issues/339#issuecomment-4681920369 */ -- 2.51.2 From b1f51a99ecda4ef2a0a44e4cbb13bba73beba654 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Sat, 22 Aug 2026 02:03:55 +0000 Subject: [PATCH 6/6] feat(ui): Implement play actions --- src/client/AppNext.tsx | 2 + src/client/components/ActivityDetail.tsx | 109 +++++++++++++++++--- src/client/components/Toaster.tsx | 41 ++++++++ src/client/components/icons/ChakraIcons.tsx | 16 ++- src/core/tests/utils/apiFixtures.ts | 1 + 5 files changed, 154 insertions(+), 15 deletions(-) create mode 100644 src/client/components/Toaster.tsx diff --git a/src/client/AppNext.tsx b/src/client/AppNext.tsx index e8848843..a75f5ae1 100644 --- a/src/client/AppNext.tsx +++ b/src/client/AppNext.tsx @@ -16,6 +16,7 @@ import { ComponentDetailedRoutable } from './components/msComponent/MSComponentD import { MSComponentListFetchable } from './components/msComponent/MSComponentList'; import { Provider } from './components/Provider'; import { SettingsContainer } from './components/settings/settings'; +import { Toaster } from './components/Toaster'; function NoMatch() { const location = useLocation(); @@ -115,6 +116,7 @@ export const sseProviderOptions = { function App() { return ( + {/* */} options={sseProviderOptions}> diff --git a/src/client/components/ActivityDetail.tsx b/src/client/components/ActivityDetail.tsx index 44c9f325..2dd0466f 100644 --- a/src/client/components/ActivityDetail.tsx +++ b/src/client/components/ActivityDetail.tsx @@ -1,11 +1,11 @@ -import { Accordion, Alert, Box, Code, Collapsible, Flex, HStack, Separator, Skeleton, SkeletonText, Span, Stack, useAccordionItemContext, type BadgeProps } from '@chakra-ui/react'; +import { Accordion, Alert, Box, Group, Portal, Code, Collapsible, Flex, HStack, Separator, Menu, Skeleton, SkeletonText, Span, Stack, useAccordionItemContext, type BadgeProps, type MenuItemProps, type MenuSelectionDetails, useClipboard } from '@chakra-ui/react'; import { useSSEContext, useSSEEvent } from "@flamefrontend/sse-runtime-react"; -import { useQuery, useQueryClient, type InfiniteData } from '@tanstack/react-query'; -import React, { Fragment, useEffect, useState } from "react"; +import { useMutation, useQuery, useQueryClient, type InfiniteData } from '@tanstack/react-query'; +import React, { Fragment, useCallback, useEffect, useState, type ComponentProps } from "react"; import { LuChevronRight } from "react-icons/lu"; import type { MarkOptional } from "ts-essentials"; import type { ComponentsApiJson, MsSseEvent, PaginatedResponse, PlayApiCommonDetailed, QueryPlaysOptsJson, SortPlaysByProps } from "../../core/Api"; -import { DEAD_QUEUE, type ComponentType, type Second } from "../../core/Atomic"; +import { INGRESS_QUEUE, type ComponentType, type Second } from "../../core/Atomic"; import { tanQueries, useQueryWatcher } from "../queries"; import { activityTimelineHasIssue } from "../utils/ComponentUtils"; import { ActivityTimeline } from "./ActivityTimeline"; @@ -13,10 +13,13 @@ import { EphemeralElement, PlayStateBadge } from "./Badges"; import { ShortDateDisplay } from "./DateDisplay"; import { ErrorAlert } from "./ErrorAlert"; import { ExpandCollapse } from "./ExpandCollapse"; -import { DebugCopy, ExclamationCircleIcon, ExclamationTriangleIcon, InsertedIcon, RetryButton, UpdatedIcon } from "./icons/ChakraIcons"; +import { DebugIcon, EllipsisButton, ExclamationCircleIcon, ExclamationTriangleIcon, FinishIconRaw, InsertedIcon, type PowerOffButton, RetryButton, RetryIcon, StopButton, StopIconRaw, TrashIconRaw, UpdatedIcon } from "./icons/ChakraIcons"; import { PlayData } from "./PlayData"; import { TextMuted } from "./TextMuted"; import { capitalize } from '../../core/StringUtils'; +import ky from 'ky'; +import type { IconType } from 'react-icons/lib'; +import { toaster } from "./Toaster" type UseActivityQueryOptions = { msQuery?: QueryPlaysOptsJson @@ -306,26 +309,104 @@ export const ActivityDetailFetchable = (props: ActivityDetailFetchableProps) => return } +const playStateMenuItem = (Icon: IconType, value: string, name?: string) => (props: Pick = {}) => { + return ({name ?? capitalize(value)}); +} + +const MenuItemRetry = playStateMenuItem(RetryIcon, 'retry'); +const MenuItemRetryWith = playStateMenuItem(RetryIcon, 'retry', 'Retry With...'); +const MenuItemCancel = playStateMenuItem(StopIconRaw, 'cancel'); +const MenuItemTrash = playStateMenuItem(TrashIconRaw, 'delete'); +const MenuItemDebug = playStateMenuItem(DebugIcon, 'debug'); +const MenuItemFinish = playStateMenuItem(FinishIconRaw, 'finish', 'Mark Completed'); + +const primaryActionProps: ComponentProps = { + margin: "1px", + variant: "subtle", + size: 'xs' +} + export const ActivityStateActions = (props: {activity: PlayApiCommonDetailed}) => { - let suffix: React.JSX.Element | null; + let suffix: React.JSX.Element | undefined; + let primaryAction: React.JSX.Element | undefined; + let menuElm: React.JSX.Element | undefined; + let menuItems: React.JSX.Element[] = []; const badgeProps: BadgeProps = {}; + + const clipboard = useClipboard({value: JSON.stringify(props.activity)}); + + const {mutate, isPending, variables, isSuccess} = useMutation({ + mutationKey: ['playAction', props.activity.uid], + mutationFn: () => ky.post(`/api/components/${props.activity.componentId}/state`,{ + json: {reason: 'User initiated from UI'} + }) + }); + + const menuCb = useCallback((select: MenuSelectionDetails) => { + if(select.value === 'debug') { + clipboard.copy(); + toaster.create({ + title: 'Copied debug data to clipboard', + type: 'success' + }); + } + //mutate(); + },[mutate, props.activity]); + const { activity: { queueStates = [] } = {} } = props; - if(props.activity.state === 'failed') { - suffix = ; - badgeProps.paddingRight = 0; + const hasDeadQueue = queueStates.some(x => x.queueName === INGRESS_QUEUE && x.retries > 0); + //badgeProps.paddingRight = 0; + switch(props.activity.state) { + case 'queued': + primaryAction = ; + menuItems = [,]; + break; + case 'failed': + primaryAction = ; + menuItems = [, ,]; + if(!hasDeadQueue) { + menuItems.unshift() + } + break + default: + primaryAction = ; + menuItems = [, ,]; + break + } - const hasDeadQueue = queueStates.some(x => x.queueName === DEAD_QUEUE && x.queueStatus === 'queued'); + if(menuItems.length > 0) { + menuElm = ( + + + {primaryAction} + + + + + + + + {menuItems} + + + + + ); + suffix = menuElm; + } else if(primaryAction !== undefined) { + suffix = primaryAction; + } + if(suffix !== undefined || primaryAction !== undefined) { + badgeProps.paddingRight = 0; + } return ( - - - - + ) diff --git a/src/client/components/Toaster.tsx b/src/client/components/Toaster.tsx new file mode 100644 index 00000000..ae5ca908 --- /dev/null +++ b/src/client/components/Toaster.tsx @@ -0,0 +1,41 @@ +"use client" + +import { + Toaster as ChakraToaster, + Portal, + Spinner, + Stack, + Toast, + createToaster, +} from "@chakra-ui/react" + +export const toaster = createToaster({ + placement: "bottom-end", + pauseOnPageIdle: true, +}) + +export const Toaster = () => ( + + + {(toast) => ( + + {toast.type === "loading" ? ( + + ) : ( + + )} + + {toast.title && {toast.title}} + {toast.description && ( + {toast.description} + )} + + {toast.action && ( + {toast.action.label} + )} + {toast.closable && } + + )} + + + ) diff --git a/src/client/components/icons/ChakraIcons.tsx b/src/client/components/icons/ChakraIcons.tsx index 2a0438ab..6d548d96 100644 --- a/src/client/components/icons/ChakraIcons.tsx +++ b/src/client/components/icons/ChakraIcons.tsx @@ -29,6 +29,8 @@ import { LuLockOpen } from "react-icons/lu" import { VscDebugRestart } from 'react-icons/vsc'; +import { HiMiniStop } from "react-icons/hi2"; +import { FaTrashCan, FaFlagCheckered } from "react-icons/fa6"; import { MdOutlineFiberNew } from "react-icons/md"; import { RiZzzFill } from "react-icons/ri"; import { SiGoogledocs } from "react-icons/si"; @@ -243,4 +245,16 @@ export const getMusicServiceIconElement = (service: string): ReactNode => { export const getMusicServiceChakraIcon = (service: string) => { const ServiceIcon = getMusicServiceIcon(service); return (props: IconProps = {}) => ; -} \ No newline at end of file +} + +export const StopIconRaw = HiMiniStop; +export const StopIcon = makeChakraIcon(StopIconRaw); +export const StopButton = makeIconButton(StopIconRaw); + +export const TrashIconRaw = FaTrashCan; +export const TrashIcon = makeChakraIcon(TrashIconRaw); +export const TrashIconButton = makeIconButton(TrashIconRaw); + +export const FinishIconRaw = FaFlagCheckered; +export const FinishIcon = makeChakraIcon(FinishIconRaw); +export const FinishButton = makeIconButton(FinishIconRaw); \ No newline at end of file diff --git a/src/core/tests/utils/apiFixtures.ts b/src/core/tests/utils/apiFixtures.ts index d038cf5e..cd6c8e66 100644 --- a/src/core/tests/utils/apiFixtures.ts +++ b/src/core/tests/utils/apiFixtures.ts @@ -146,6 +146,7 @@ export const generateComponentCommonApiJson = (data: Partial countNonLive: 0, state, players, + errors: [], status: faker.helpers.arrayElement(statusSamples), monitoringStatus, queued,