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,