multi-scrobbler/multi-scrobbler
1
0
Fork 0
mirror of https://github.com/FoxxMD/multi-scrobbler.git synced 2026-09-03 05:10:00 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix: further queue state cleanup before converting to event

Browse Source
This commit is contained in:
FoxxMD
2026-08-27 19:11:16 +00:00
parent b494e8f93f
commit 935845a976
3 changed files with 32 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/common/database/drizzle/entityUtils.ts
+12
View File
@@ -128,6 +128,18 @@ export const queueStateToPlayEvent = (partial: MarkOptional<QueueStateSelect, 'c
createdAt: dayjs(),
data: partial
});
export const queueCompletionStateToPlayEvent = (partial: MarkOptional<QueueStateSelect, 'context' | 'retries'>): Omit<PlayEventQueueStateChange, 'playId'> => {
const {
context,
retries,
...rest
} = partial;
return {
eventName: PLAY_EVENT_TYPE.queueStateChange,
createdAt: dayjs(),
data: rest
}
}
export const dupeCheckToPlayEvent = (partial: MarkRequired<Partial<PlayEventDupeCheckData>, 'match'>): Omit<PlayEventDupeCheck, 'playId'> => {
const {match, ...rest} = partial;
src/backend/scrobblers/AbstractScrobbleClient.ts
+12 -11
View File
@@ -74,7 +74,7 @@ import { COMPONENT_STATE, type ComponentClientApiJson, type PlayApiCommonDetaile
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, entityIsPlayEntity, queueStateToPlayEvent, scrobbleToPlayEvent, stateChangeToPlayEvent, transformToPlayEvent } from "../common/database/drizzle/entityUtils.ts";
import { dupeCheckToPlayEvent, entityIsPlayEntity, queueCompletionStateToPlayEvent, 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";
@@ -1195,6 +1195,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
signal?.throwIfAborted();
const queueState = playEntity.queueStates.find(x => x.queueName === INGRESS_QUEUE);
queueState.error = undefined;
const {
context,
} = queueState
@@ -1217,7 +1218,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
if (!isRetry && !playEntity.play.meta.wasMonitored) {
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, context: undefined, queueStatus: 'completed'}));
events.push(queueCompletionStateToPlayEvent({...queueState, queueStatus: 'completed'}));
playEntity.state = 'discarded';
return {playEntity, queue: queueState, events};
}
@@ -1242,7 +1243,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
events.push(stateChangeToPlayEvent({state: 'failed'}));
queueState.queueStatus = QUEUE_STATUS_FAILED;
queueState.error = processError;
events.push(queueStateToPlayEvent({...queueState, context: undefined,}));
events.push(queueCompletionStateToPlayEvent({...queueState}));
throw new PlayProcessingError(processError, {playEntity, events, queue: queueState, showStopping: false});
//deadQueueEntity = await this.addDeadLetterScrobble(playEntity, e);
}
@@ -1273,7 +1274,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
//currQueuedPlay.play = scrobbledPlay;
await this.addScrobbledTrack(scrobbledPlay);
events.push(stateChangeToPlayEvent({state: 'scrobbled'}));
events.push(queueStateToPlayEvent({...queueState, context: undefined, queueStatus: QUEUE_STATUS_COMPLETED}));
events.push(queueCompletionStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED}));
this.scrobbleRetries = 0;
playEntity.state = 'scrobbled';
playEntity.error = undefined;
@@ -1299,7 +1300,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
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, context: undefined, queueStatus: QUEUE_STATUS_FAILED, error: nonShowStoppingError}));
events.push(queueCompletionStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: nonShowStoppingError}));
queueState.error = nonShowStoppingError;
logger.warn(nonShowStoppingError);
processError = nonShowStoppingError;
@@ -1308,7 +1309,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
//this.queuedScrobbles.unshift(currQueuedPlay);
//handledShiftedPlay = true;
const showStoppingError = new Error('Error occurred while trying to scrobble', { cause: e });
events.push(queueStateToPlayEvent({...queueState, context: undefined, queueStatus: QUEUE_STATUS_FAILED, error: showStoppingError}));
events.push(queueCompletionStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: showStoppingError}));
queueState.error = showStoppingError;
processError = showStoppingError;
throw new PlayProcessingError(showStoppingError, {playEntity, queue: queueState, events, showStopping: true});
@@ -1319,7 +1320,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
this.scrobbleRetries = 0;
playEntity.state = 'duped';
events.push(stateChangeToPlayEvent({state: 'duped'}));
events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED}));
events.push(queueCompletionStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED}));
return {playEntity, events, queue: queueState};
}
} catch (e) {
@@ -1328,7 +1329,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
}
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)}));
events.push(queueCompletionStateToPlayEvent({...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)) {
@@ -1336,7 +1337,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
playEntity.state = 'failed';
}
if(!events.some(x => x.eventName === PLAY_EVENT_TYPE.queueStateChange)) {
events.push(queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: e}));
events.push(queueCompletionStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: e}));
}
throw new PlayProcessingError(e, {playEntity, queue: queueState, events, showStopping: true});
}
@@ -1357,7 +1358,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
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'}}) }
{ playId: dead.id, ...queueCompletionStateToPlayEvent({...queueState, error: undefined, queueStatus: QUEUE_STATUS_COMPLETED, context: {reason: 'Dead Play marked as completed by user'}}) }
];
await this.queueRepo.deleteByIds([queueState.id]);
@@ -1442,7 +1443,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
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 });
this.queueRepo.updateById(queue.id, { queueStatus: 'queued', context, error: undefined });
}
const events = await this.playEventsRepo.createMany([
{ playId: playSelect.id, ...stateChangeToPlayEvent({ state: 'queued' }) },
src/backend/sources/AbstractSource.ts
+8 -7
View File
@@ -51,7 +51,7 @@ import type { PlayEventNew, PlayEventSelect, PlaySelect, PlaySelectWithQueueStat
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 { dupeCheckToPlayEvent, entityIsPlayEntity, queueCompletionStateToPlayEvent, queueStateToPlayEvent, stateChangeToPlayEvent, transformToPlayEvent } from '../common/database/drizzle/entityUtils.ts';
import type { PlayProcessingResult } from '../common/infrastructure/PlayProcessing.ts';
import { PlayProcessingError } from '../common/errors/PlayProcessingError.ts';
@@ -431,7 +431,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
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 });
this.queueRepo.updateById(queue.id, { queueStatus: 'queued', context, error: undefined });
}
const events = await this.playEventsRepo.createMany([
{ playId: playSelect.id, ...stateChangeToPlayEvent({ state: 'queued' }) },
@@ -1157,6 +1157,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
this.setStatus(`Processing Play ${playEntity.uid}`);
const queueState = playEntity.queueStates.find(x => x.queueName === INGRESS_QUEUE);
queueState.error = undefined;
const {
context,
} = queueState
@@ -1178,7 +1179,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
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, context: undefined, queueStatus: QUEUE_STATUS_COMPLETED}));
events.push(queueCompletionStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED}));
return {playEntity, queue: queueState, events};
}
let preCompared = playEntity.play;
@@ -1238,7 +1239,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
this.cache.cacheDb.set(this.recentDiscoveredCacheKey(), recentDiscoveredPlays, '2m');
}
}
events.push(queueStateToPlayEvent({...queueState, context: undefined, queueStatus: QUEUE_STATUS_COMPLETED}));
events.push(queueCompletionStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED}));
logger.info(`${capitalize(playEntity.state)} => ${buildTrackString(preCompared)}`);
return {playEntity, events, queue: queueState};
} catch (e) {
@@ -1247,7 +1248,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
}
if(isAbortError(e)) {
events.push(stateChangeToPlayEvent({state: 'failed'}));
events.push(queueStateToPlayEvent({...queueState, context: undefined, queueStatus: QUEUE_STATUS_FAILED, error: generateLoggableAbortReason('Interrupted by abort signal', this.discoverQueueAbortController.signal)}));
events.push(queueCompletionStateToPlayEvent({...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)) {
@@ -1255,7 +1256,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
playEntity.state = 'failed';
}
if(!events.some(x => x.eventName === PLAY_EVENT_TYPE.queueStateChange)) {
events.push(queueStateToPlayEvent({...queueState, context: undefined, queueStatus: QUEUE_STATUS_FAILED, error: e}));
events.push(queueCompletionStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_FAILED, error: e}));
}
throw new PlayProcessingError(e, {playEntity, queue: queueState, events, showStopping: true});
}
@@ -1276,7 +1277,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
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'}}) }
{ playId: dead.id, ...queueStateToPlayEvent({...queueState, error: undefined, queueStatus: QUEUE_STATUS_COMPLETED, context: {reason: 'Dead Play marked as completed by user'}}) }
];
await this.queueRepo.deleteByIds([queueState.id]);