vrr/multi-scrobbler
2
0
Fork 0
mirror of https://github.com/FoxxMD/multi-scrobbler.git synced 2026-07-09 17:28:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

feat: Implement error persistence and live updated for components

Browse source
This commit is contained in:
FoxxMD 2026-07-08 15:02:02 +00:00
parent 153f37e927
commit b57f325684
10 changed files with 103 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

25
src/backend/common/AbstractComponent.ts
View file

@ -37,9 +37,10 @@ import { ClientType } from "./infrastructure/config/client/clients.js";
import { SourceType } from "./infrastructure/config/source/sources.js";
import { DrizzleComponentRepository } from "./database/drizzle/repositories/ComponentRepository.js";
import dayjs from "dayjs";
import { COMPONENT_STATE, ComponentCommonApi, ComponentState, PlayApiCommonDetailed } from "../../core/Api.js";
import { COMPONENT_STATE, ComponentCommonApi, ComponentCommonApiJson, ComponentState, PlayApiCommonDetailed } from "../../core/Api.js";
import { WebhookPayload } from "./infrastructure/config/health/webhooks.js";
import { MarkRequired } from "ts-essentials";
import { serializeError } from "serialize-error";
export type AbstractComponentConfig = (CommonClientConfig | CommonSourceConfig) & { transformManager?: TransformerManager };
@ -223,7 +224,9 @@ export default abstract class AbstractComponent extends AbstractInitializable {
await repo.retentionCleanup(this.componentType, this.retentionOpts);
this.setStatus('Retention cleanup finished');
} catch (e) {
this.logger.warn(new Error('Failed to do retention cleanup', {cause: e}));
const retentionErr = new Error('Failed to do retention cleanup', {cause: e});
this.warning = retentionErr;
this.logger.warn(retentionErr);
this.setStatus('Retention cleanup failed');
}
}
@ -522,7 +525,7 @@ export default abstract class AbstractComponent extends AbstractInitializable {
public abstract getRunningState(): ComponentState
public getApiData(): Omit<ComponentMinimalSelect, 'type' | 'countLive'> & Pick<ComponentCommonApi, 'state'> {
public getApiData(): Omit<ComponentCommonApiJson, 'type' | 'countLive' | 'players'> & Pick<ComponentCommonApi, 'state' | 'error' | 'warning'> {
let state: ComponentState;
if(!this.initializedOnce || this.initializing) {
state = COMPONENT_STATE.INITIALIZING;
@ -538,9 +541,11 @@ export default abstract class AbstractComponent extends AbstractInitializable {
state,
mode: this.dbComponent.mode,
countNonLive: this.dbComponent.countNonLive,
createdAt: this.dbComponent.createdAt,
lastReadyAt: this.dbComponent.lastReadyAt,
lastActiveAt: this.dbComponent.lastActiveAt,
createdAt: this.dbComponent.createdAt?.toISOString(),
lastReadyAt: this.dbComponent.lastReadyAt?.toISOString(),
lastActiveAt: this.dbComponent.lastActiveAt?.toISOString(),
error: this.error !== undefined && this.error instanceof Error ? serializeError(this.error) : this.error,
warning: this.warning !== undefined && this.warning instanceof Error ? serializeError(this.warning) : this.warning,
...this.additionalApiData()
}
}
@ -555,7 +560,13 @@ export default abstract class AbstractComponent extends AbstractInitializable {
});
}
protected emitComponentUpdate = <T = Partial<typeof this.getApiData>>(payload: T) => {
protected emitComponentUpdate = <T extends Partial<ReturnType<typeof this.getApiData>>>(payload: T) => {
if('error' in payload && payload.error instanceof Error) {
payload.error = serializeError(payload.error);
}
if('warning' in payload && payload.warning instanceof Error) {
payload.warning = serializeError(payload.warning);
}
this.emitEvent('componentUpdate', payload);
}
protected emitPlayUpdate = (payload: MarkRequired<Partial<PlayApiCommonDetailed>, 'uid'>) => {

12
src/backend/common/AbstractInitializable.ts
View file

@ -1,9 +1,9 @@
import { childLogger, Logger } from "@foxxmd/logging";
import { Logger } from "@foxxmd/logging";
import {truncateStringToLength } from "../../core/StringUtils.js";
import { hasNodeNetworkException } from "./errors/NodeErrors.js";
import { hasUpstreamError } from "./errors/UpstreamError.js";
import { WebhookPayload } from "./infrastructure/config/health/webhooks.js";
import { AuthCheckError, BuildDataError, ConnectionCheckError, ParseCacheError, PostInitError, StageError, TransformRulesError } from "./errors/MSErrors.js";
import { AuthCheckError, BuildDataError, ConnectionCheckError, ParseCacheError, PostInitError, StageError } from "./errors/MSErrors.js";
import { messageWithCausesTruncatedDefault } from "../../core/ErrorUtils.js";
import { spawn } from 'abort-controller-x';
@ -19,6 +19,8 @@ export default abstract class AbstractInitializable {
databaseOK?: boolean | null;
connectionOK?: boolean | null;
cacheOK?: boolean | null;
error?: Error;
warning?: Error;
protected initializedOnce: boolean = false;
initializing: boolean = false;
@ -86,12 +88,16 @@ export default abstract class AbstractInitializable {
} catch (e) {
throw new PostInitError('Error occurred during post-initialization hook', {cause: e});
}
this.error = undefined;
this.warning = undefined;
return true;
} catch(e) {
if(notify) {
await this.notify({identifier: this.getIdentifier(), title: notifyTitle, message: truncateStringToLength(500)(messageWithCausesTruncatedDefault(e)), priority: 'error'});
}
throw new Error('Initialization failed', {cause: e});
const initError = new Error('Initialization failed', {cause: e});
this.error = initError;
throw initError;
} finally {
this.initializing = false;
}

6
src/backend/common/database/drizzle/repositories/PlayRepository.ts
View file

@ -254,7 +254,11 @@ export class DrizzlePlayRepository extends DrizzleBaseRepository<'plays'> {
const where = buildPlayWhere({componentId: args.componentId ?? this.componentId, ...rest});
// https://github.com/drizzle-team/drizzle-orm/issues/5218#issuecomment-3854900241
const filter = relationsFilterToSQL(plays, where, relations.plays.relations, relations);
const filter = relationsFilterToSQL(plays,
// @ts-expect-error don't have a good way to type this yet
where,
relations.plays.relations,
relations);
// https://github.com/drizzle-team/drizzle-orm/discussions/3119#discussioncomment-16379557
const count = await this.db.$count(plays, filter);

53
src/backend/scrobblers/AbstractScrobbleClient.ts
View file

@ -254,10 +254,12 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
'Heartbeat',
(): Promise<any> => {
return this.heartbeatTask().then(() => null).catch((err) => {
this.error = err;
this.logger.error(err);
});
},
(err: Error) => {
this.error = err;
this.logger.error(err);
}
), {id: 'heartbeat'}));
@ -281,12 +283,14 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
(): Promise<any> => {
if(this.isReady()) {
return this.processDeadLetterQueue().then(() => null).catch((e) => {
this.warning = e;
this.logger.error(e);
})
}
return new Promise((resolve, reject) => resolve);
},
(err: Error) => {
this.warning = err;
this.logger.error(err);
}
), {id: 'dead'}));
@ -426,7 +430,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
}
}
public getApiData(): ComponentClientApi {
public getApiData(): ComponentClientApiJson {
return {
...super.getApiData(),
...this.getComponentApiData(),
@ -501,7 +505,10 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
const t = new AsyncTask('Playing Now', (): Promise<any> => {
return this.processingPlayingNow();
}, (err: Error) => {
this.npLogger.error(new Error('Unexpected error while processing Now Playing queue', {cause: err}));
const npErr = new Error('Unexpected error while processing Now Playing queue', {cause: err});
this.npLogger.error(npErr);
this.warning = npErr;
this.emitComponentUpdate<Partial<ComponentClientApiJson>>({warning: npErr});
});
// even though we are processing every 5 seconds the interval that Now Playing is updated at, and that the queue is cleared on,
@ -806,7 +813,10 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
}
}
} catch (e) {
this.logger.warn(new SimpleError('Could not preload scrobbles', {cause: e, shortStack: true}));
const preloadErr = new SimpleError('Could not preload scrobbles', {cause: e, shortStack: true});
this.warning = preloadErr;
this.emitComponentUpdate<Partial<ComponentClientApiJson>>({warning: preloadErr});
this.logger.warn(preloadErr);
}
}
}
@ -973,17 +983,21 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
await this.startScrobbling(signal);
}).catch((e) => {
let status: string;
const componentUpdate: Partial<ComponentClientApiJson> = {
state: COMPONENT_STATE.IDLE
};
if (isAbortError(e)) {
const err = generateLoggableAbortReason('Scrobble processing stopped', this.scrobbleQueueAbortController.signal);
this.logger.info(err);
this.logger.trace(e);
status = 'Processing cancelled';
componentUpdate.status = 'Processing cancelled';
} else {
this.logger.warn(new Error('Scrobble processing stopped with error', { cause: e }));
status = 'Processing stopped with error';
const err = new Error('Scrobble processing stopped with error', { cause: e });
this.logger.warn(err);
componentUpdate.status = 'Processing stopped with error';
componentUpdate.warning = err;
}
this.emitComponentUpdate<Partial<ComponentClientApiJson>>({state: COMPONENT_STATE.IDLE, status});
this.emitComponentUpdate<Partial<ComponentClientApiJson>>(componentUpdate);
}).finally(() => {
this.scrobbleQueueAbortController = undefined;
this.scrobbleQueuePromise = undefined;
@ -1021,11 +1035,11 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
if(!this.isUsable()) {
this.logger.warn('Stopping scrobble processing due to client no longer usable.');
await this.notify({title: `Processing Error`, message: `Encountered error while scrobble processing and client is no longer usable, stopping processing!. | Error: ${e.message}`, priority: 'error'});
break;
throw e;
} else if (this.authGated()) {
this.logger.warn('Stopping scrobble processing due to client no longer being authenticated.');
await this.notify({title: ` Processing Error`, message: `Encountered error while scrobble processing and client is no longer authenticated, stopping processing!. | Error: ${e.message}`, priority: 'error'});
break;
throw e;
} else if (this.scrobbleRetries < maxRetries) {
const delayFor = pollingBackoff(this.scrobbleRetries + 1, retryMultiplier);
this.logger.info(`Scrobble processing retries (${this.scrobbleRetries}) less than max processing retries (${maxRetries}), restarting processing after ${delayFor} second delay...`);
@ -1034,6 +1048,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
} else {
this.logger.warn(`Scrobble processing retries (${this.scrobbleRetries}) equal to max processing retries (${maxRetries}), stopping processing!`);
await this.notify({title: `Processing Error`, message: `Encountered error while scrobble processing and retries (${this.scrobbleRetries}) are equal to max processing retries (${maxRetries}), stopping processing!. | Error: ${e.message}`, priority: 'error'});
throw e;
}
this.scrobbleRetries++;
}
@ -1067,6 +1082,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
signal.throwIfAborted();
this.logger.info('Scrobble processing started');
this.emitEvent('statusChange', {status: 'Running'});
this.emitComponentUpdate<Partial<ComponentClientApiJson>>({state: COMPONENT_STATE.RUNNING});
try {
this.setStatus('Waiting for Plays from Sources');
@ -1081,6 +1097,11 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
if(nextQueued !== undefined) {
while (nextQueued !== undefined) {
await this.processQueueCurrentScrobble(nextQueued, signal);
if(this.error !== undefined) {
// we made it through a scrobble without any issues so clear any issue we may have previously had
this.error = undefined;
this.emitComponentUpdate<Partial<ComponentClientApiJson>>({error: null});
}
nextQueued = await this.playRepo.getQueueNext(CLIENT_INGRESS_QUEUE)
}
this.emitEvent('queueEmptied', {});
@ -1095,6 +1116,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
this.logger.error(e);
}
this.emitEvent('statusChange', {status: 'Idle'});
this.emitComponentUpdate<Partial<ComponentClientApiJson>>({state: COMPONENT_STATE.IDLE});
this.scrobbling = false;
throw e;
}
@ -1825,16 +1847,9 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i
}
}
export const nowPlayingUpdateByPlayDuration: NowPlayingUpdateThreshold = (play?: PlayObject) => {
if(play === undefined) {
31;
}
return (play?.data?.duration ?? 30) + 1;
}
export const nowPlayingUpdateByPlayDuration: NowPlayingUpdateThreshold = (play?: PlayObject) => (play?.data?.duration ?? 30) + 1
export const shouldClearNPStatus = (data: SourcePlayerObj) => {
return [CALCULATED_PLAYER_STATUSES.stopped, CALCULATED_PLAYER_STATUSES.paused].includes(data.status.calculated as ReportedPlayerStatus);
}
export const shouldClearNPStatus = (data: SourcePlayerObj) => [CALCULATED_PLAYER_STATUSES.stopped, CALCULATED_PLAYER_STATUSES.paused].includes(data.status.calculated as ReportedPlayerStatus)
export const playerInNPPlayingOnlyState = (data: SourcePlayerObj): [boolean, string] => {
// for lower-interval update clients (like listenbrainz, lastfm) IE not real-time

6
src/backend/server/api.ts
View file

@ -41,7 +41,7 @@ import { DrizzlePlayRepository, QueryPlaysOpts, QueryPlaysOptsJson } from "../co
import { playSelectToDeadScrobble } from "../common/database/drizzle/entityUtils.js";
import AbstractHistoricalScrobbleClient from "../scrobblers/AbstractHistoricalScrobbleClient.js";
import { DrizzlePlayHistoricalRepository } from "../common/database/drizzle/repositories/PlayHistoricalRepository.js";
import { ComponentClientApi, ComponentSourceApi, ComponentSourceApiJson } from "../../core/Api.js";
import { ComponentClientApi, ComponentClientApiJson, ComponentSourceApi, ComponentSourceApiJson } from "../../core/Api.js";
import { asDayjsHydratedObject } from "../../core/DataUtils.js";
import { Dayjs } from "dayjs";
import { asSerializablePlaySelect } from "../../core/PlayMarshalUtils.js";
@ -206,7 +206,7 @@ export const setupApi = (app: Express, logger: Logger, appLoggerStream: PassThro
requiresAuthInteraction = false,
authed = false
} = x;
const base: ComponentSourceApi = x.getApiData();
const base: ComponentSourceApiJson = x.getApiData();
if(!x.isReady()) {
if(x.buildOK === false) {
base.status = 'Initializing Data Failed';
@ -235,7 +235,7 @@ export const setupApi = (app: Express, logger: Logger, appLoggerStream: PassThro
authed = false,
scrobbling = false,
} = x;
const base: ComponentClientApi = x.getApiData();
const base: ComponentClientApiJson = x.getApiData();
if (!x.isReady()) {
if(x.buildOK === false) {

32
src/backend/sources/AbstractSource.ts
View file

@ -159,11 +159,13 @@ export default abstract class AbstractSource extends AbstractComponent implement
'Heartbeat',
(): Promise<any> => {
return this.heartbeatTask().then(() => null).catch((err) => {
this.error = err;
this.logger.error(err);
});
},
(err: Error) => {
this.logger.error(err);
this.error = err;
}
), {id: 'heartbeat'}));
} else {
@ -182,7 +184,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
await this.initialize({force: false, notify: true, notifyTitle: 'Could not initialize automatically'});
} catch (e) {
this.logger.error(new Error('Could not initialize automatically', {cause: e}));
this.setStatus('Could not initialzie automatically');
this.setStatus('Could not initialize automatically');
return false;
}
@ -286,7 +288,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
}
}
public getApiData(): ComponentSourceApi {
public getApiData(): ComponentSourceApiJson {
return {
...super.getApiData(),
...this.getComponentApiData(),
@ -528,8 +530,11 @@ export default abstract class AbstractSource extends AbstractComponent implement
try {
await this.initialize(options);
} catch (e) {
this.logger.error(new Error('Cannot start polling because Source is not ready', {cause: e}));
const err = new Error('Cannot start polling because Source is not ready', {cause: e});
this.logger.error(err);
this.setStatus('Polling Error');
this.emitComponentUpdate<Partial<ComponentSourceApiJson>>({error: err});
this.error = err;
if(notify) {
await this.notify( {title: `Polling Error`, message: `Cannot start polling because Source is not ready: ${truncateStringToLength(500)(messageWithCausesTruncatedDefault(e))}`, priority: 'error'});
}
@ -569,17 +574,22 @@ export default abstract class AbstractSource extends AbstractComponent implement
});
await this.startPolling(signal);
}).catch((e) => {
let status: string;
const componentUpdate: Partial<ComponentSourceApiJson> = {
state: COMPONENT_STATE.IDLE
};
if (isAbortError(e)) {
const err = generateLoggableAbortReason('Polling stopped', this.abortController.signal);
this.logger.info(err);
this.logger.trace(e)
status = 'Polling cancelled';
componentUpdate.status = 'Polling cancelled';
} else {
this.logger.warn(new Error('Polling stopped with error', { cause: e }));
status = 'Polling stopped with error';
const err = new Error('Polling stopped with error', { cause: e });
this.logger.warn(err);
componentUpdate.status = 'Polling stopped with error';
componentUpdate.warning = err;
this.warning = err;
}
this.emitComponentUpdate<Partial<ComponentSourceApiJson>>({state: COMPONENT_STATE.IDLE, status});
this.emitComponentUpdate<Partial<ComponentSourceApiJson>>(componentUpdate);
}).finally(() => {
this.abortController = undefined;
this.pollingPromise = undefined;
@ -764,7 +774,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
this.logger[lastActivityLogLevel](activityMsgs.join(' | '));
this.setWakeAt(pollFrom.add(sleepTime, 'seconds'));
this.setIsSleeping(true);
this.emitComponentUpdate({sleeping: true, wakeAt: this.getWakeAt().toISOString()})
this.emitComponentUpdate<Partial<ComponentSourceApiJson>>({sleeping: true, wakeAt: this.getWakeAt().toISOString()})
// set last active before we sleep
this.componentRepo.updateById(this.dbComponent.id, {lastActiveAt: dayjs(), lastReadyAt: dayjs()});
while(dayjs().isBefore(this.getWakeAt())) {
@ -772,7 +782,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
await delay(signal, 500);
}
this.setIsSleeping(false);
this.emitComponentUpdate({sleeping: false});
this.emitComponentUpdate<Partial<ComponentSourceApiJson>>({sleeping: false});
// if we have made it this far in the loop we can reset poll retries
this.pollRetries = 0;
}
@ -787,7 +797,7 @@ export default abstract class AbstractSource extends AbstractComponent implement
throw e;
} finally {
this.setIsSleeping(false);
this.emitComponentUpdate({sleeping: false});
this.emitComponentUpdate<Partial<ComponentSourceApiJson>>({sleeping: false});
}
}

1
src/backend/sources/DeezerSource.ts
View file

@ -15,7 +15,6 @@ import { baseFormatPlayObj } from "../utils/PlayTransformUtils.js";
export default class DeezerSource extends AbstractSource {
workingCredsPath;
error: any;
requiresAuth = true;
requiresAuthInteraction = true;

4
src/backend/sources/MemorySource.ts
View file

@ -29,7 +29,7 @@ import { AbstractPlayerState, createPlayerOptions, PlayerStateOptions } from "./
import { GenericPlayerState } from "./PlayerState/GenericPlayerState.js";
import { hashObject } from "../utils/StringUtils.js";
import { useDebugValue } from "react";
import { ComponentSourceApi } from "../../core/Api.js";
import { ComponentSourceApi, ComponentSourceApiJson } from "../../core/Api.js";
const EXPECTED_NON_DISCOVERED_REASON = 'not added because an identical play with the same timestamp was already discovered.';
@ -179,7 +179,7 @@ export default class MemorySource extends AbstractSource {
return record;
}
public getApiData(): ComponentSourceApi {
public getApiData(): ComponentSourceApiJson {
return {
...super.getApiData(),
sot: this.playerSourceOfTruth,

10
src/client/components/msComponent/MSComponentDetailed.tsx
View file

@ -147,7 +147,13 @@ export const ComponentStateBadgeActionable = (props: Omit<ComponentProps<typeof
export const ComponentDetailedDesktop = (props: {data?: ComponentCommonApiJson, live?: boolean}) => {
let sleepingRender: React.JSX.Element = null;
const {data} = props;
const {
data,
data: {
warning,
error
} = {}
} = props;
const isSource = isComponentSourceApiJson(data)
if(isSource) {
const {
@ -186,6 +192,8 @@ export const ComponentDetailedDesktop = (props: {data?: ComponentCommonApiJson,
<Flex justifyContent="flex-end" rowGap="6" flexDirection="row-reverse" wrap="wrap">
<Box marginEnd="auto"><MSComponentStats {...props}/></Box>
</Flex>
{error !== undefined && error !== null ? <ErrorAlert error={error}/> : undefined}
{warning !== undefined && warning !== null ? <ErrorAlert error={warning} status="warning"/> : undefined}
<MSErrorBoundary>{props.live ? <PlayersContainerFetchable nowPlaying={isSource ? undefined : true} data={props.data}/> : <PlayersContainer nowPlaying={isSource ? undefined : true} data={props.data} live={props.live}/>}</MSErrorBoundary>
<Heading size="3xl" width="100%">{isComponentTypeSource(props.data.mode) ? 'Plays' : 'Scrobbles'}</Heading>
<MSErrorBoundary><ListContainerFilterable render="virtDynamic" componentType={props.data.mode} componentId={props.data.id}/></MSErrorBoundary>

2
src/core/Api.ts
View file

@ -81,6 +81,8 @@ export type ComponentCommonApi = {
/** More specific, live activity state like "sleeping", "hydrating historical scrobbles", "processing dead scrobbles", etc... */
status?: string
players: Record<string, SourcePlayerJson>
error?: ErrorIsh
warning?: ErrorIsh
} & Omit<ComponentMinimalSelect, 'type'>
export type ComponentCommonApiJson = Replace<ComponentCommonApi, PickKeys<ComponentCommonApi, Dayjs>, string>;