Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/FoxxMD/multi-scrobbler. Scrobble plays from multiple sources to multiple clients docs.multi-scrobbler.app
Something went wrong. Try again.
38 kB · 862 lines
TypeScript
at debugStartup
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863import { 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 { type PlayMatchResult, type PlayObject, SOURCE_SOT } from "../../core/Atomic.ts";import { buildTrackString, capitalize, truncateStringToLength } from "../../core/StringUtils.ts";import AbstractComponent from "../common/AbstractComponent.ts";import { type Authenticatable, DEFAULT_POLLING_INTERVAL, DEFAULT_POLLING_MAX_INTERVAL, DEFAULT_RETRY_MULTIPLIER, type GroupedFixedPlays, type InternalConfig, type ProgressAwarePlayObject,} from "../common/infrastructure/Atomic.ts";import type {PlayState, PlayUserId} 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";import { TRANSFORM_HOOK } from "../../core/Transform.ts";import TupleMap from "../common/TupleMap.ts";import { difference, pollingBackoff, sleep, sortByOldestPlayDate,} from "../utils.ts";import { sortByNewestPlayDate } from '../../core/PlayUtils.ts';import { formatNumber } from '../../core/DataUtils.ts';import { timeToHumanTimestamp } from "../../core/TimeUtils.ts";import { todayAwareFormat } from "../../core/TimeUtils.ts";import { getRoot } from '../ioc.ts';import { componentFileLogger } from '../common/logging.ts';;import { messageWithCausesTruncatedDefault } from "../../core/ErrorUtils.ts";import { existingScrobble, type ExistingScrobbleOpts } from '../utils/PlayComparisonUtils.ts';import { staggerMapper } from '../utils/AsyncUtils.ts';import pMap, {pMapIterable} from 'p-map';import type { Counter } from 'prom-client';import { normalizeStr } from '../utils/StringUtils.ts';import { spawn, isAbortError, delay, throwIfAborted } from 'abort-controller-x';import { generateLoggableAbortReason } 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 { PlayWith } from '../common/database/drizzle/drizzleTypes.ts';
export interface RecentlyPlayedOptions { limit?: number formatted?: boolean
display?: boolean}
export default abstract class AbstractSource extends AbstractComponent implements Authenticatable {
declare type: SourceType;
declare config: SourceConfig; clients: string[]; instantiatedAt: Dayjs; lastActivityAt: Dayjs;
multiPlatform: boolean = false;
localUrl: URL;
configDir: string;
canPoll: boolean = false; polling: boolean = false; canBacklog: boolean = false; protected abortController: AbortController | undefined; protected pollingPromise: Promise<void> | undefined; stopPollingWaitInterval: number = 200; pollRetries: number = 0; tracksDiscovered: number = 0; tracksDiscoveredTotal: number = 0;
protected isSleeping: boolean = false; protected wakeAt: Dayjs = dayjs();
supportsUpstreamRecentlyPlayed: boolean = false; supportsUpstreamNowPlaying: boolean = false; supportsManualListening: boolean = false;
scheduler: ToadScheduler = new ToadScheduler();
protected SCROBBLE_BACKLOG_COUNT: number = 30;
protected recentDiscoveredPlays: GroupedFixedPlays = new TupleMap<DeviceId, PlayUserId, FixedSizeList<ProgressAwarePlayObject>>();
protected loggerLabel: string;
protected discoveredCounter: Counter;
protected staggerMappers = { preCompare: staggerMapper<PlayObject, PlayObject>({concurrency: 2}), postCompare: staggerMapper<PlayObject, PlayObject>({concurrency: 2}) }
declare protected componentType: 'source';
protected playRepo!: DrizzlePlayRepository;
existingDiscoveredPlay: (playObjPre: PlayObject, existingScrobbles: PlayObject[], log?: boolean) => Promise<PlayMatchResult>
constructor(type: SourceType, name: string, config: SourceConfig, internal: InternalConfig, emitter: EventEmitter) { super(config); this.componentType = 'source'; const {clients = [] } = config; this.type = type; this.name = name; this.logger = childLogger(internal.logger, this.getIdentifier()); this.loggerLabel = this.getIdentifier(); this.config = config; this.clients = clients; this.logger.debug(`Scrobble To: ${this.clients.length === 0 ? 'All' : this.clients.join(' | ')}`); this.instantiatedAt = dayjs(); this.lastActivityAt = this.instantiatedAt; this.localUrl = internal.localUrl; this.configDir = internal.configDir; this.emitter = emitter; this.discoveredCounter = getRoot().items.sourceMetics.discovered;
const existingScrobbleOpts: ExistingScrobbleOpts = { logger: this.logger, transformRules: this.transformRules, transformPlay: this.transformPlay, existingSubmitted: async (_) => [undefined, undefined] } this.existingDiscoveredPlay = (playObjPre: PlayObject, existingScrobbles: PlayObject[], log?: boolean) => existingScrobble(playObjPre, existingScrobbles, existingScrobbleOpts, log); }
async [Symbol.asyncDispose]() { if(this.canPoll) { await this.tryStopPolling('Source is being disposed'); } }
public initTasks() { if(this.scheduler.existsById('heartbeat') === false) { this.logger.info('Adding Heartbeat Task and running immediately'); this.scheduler.addSimpleIntervalJob(new SimpleIntervalJob({ minutes: 20, runImmediately: true }, new AsyncTask( '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 { this.logger.warn('Heartbeat task is already added to scheduler.'); } }
protected async heartbeatTask(): Promise<boolean> { if(!this.isReady()) { if(!this.canAuthUnattended()) { this.logger.warn({labels: 'Heartbeat'}, 'Source is not ready but will not try to initialize because auth state is not good and cannot be corrected unattended.') return false; } try { this.setStatus('Attempting to initialize...'); 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 initialize automatically'); return false; }
if('discoverDevices' in this && typeof this.discoverDevices === 'function') { this.discoverDevices(); } } if(this.isReady()) { if (this.canPoll && !this.polling) { if(!this.canAuthUnattended()) { this.logger.warn({labels: 'Heartbeat'}, 'Should be polling but will not attempt to start because auth state is not good and cannot be correct unattended.'); return false; } else { this.logger.info({labels: 'Heartbeat'}, 'Should be polling, attempting to start polling...'); this.setStatus('Attempting to start polling...'); this.poll({force: false, notify: true}).catch(e => this.logger.error(e)); } return true; } else if(!this.canPoll) { this.updateDates({lastReadyAt: dayjs(), force: true}); } } return true; }
protected async postCache(): Promise<void> { await super.postCache(); this.generateStaggerMappers(); }
protected async postDatabase(): Promise<void> { this.playRepo = new DrizzlePlayRepository(this.db, {logger: this.logger}); this.playRepo.componentId = this.dbComponent.id; const counts = await this.playRepo.getComponentPlayCountByState(); const discoveredCount = counts.find(x => x.state === 'discovered'); if(discoveredCount !== undefined) { this.tracksDiscoveredTotal = discoveredCount['count(*)']; } }
protected generateStaggerMappers() { const { preCompare = [], postCompare = [], } = this.transformRules;
if (preCompare.length > 0) { const pcInits: number[] = [0], pcMaxStagger: number[] = [0]; for (const hook of this.transformRules.preCompare) { const t = this.transformManager.getTransformerByStage({ type: hook.type, name: hook.name }); pcInits.push(t.staggerOpts?.initialInterval ?? 0); pcMaxStagger.push(t.staggerOpts?.maxRandomStagger ?? 0) } this.staggerMappers.preCompare = staggerMapper<PlayObject, PlayObject>({ initialInterval: Math.max(...pcInits), maxRandomStagger: Math.max(...pcMaxStagger), concurrency: 2 }); }
if (postCompare.length > 0) { const postInits: number[] = [0], postMaxStagger: number[] = [0]; for (const hook of this.transformRules.postCompare) { const t = this.transformManager.getTransformerByStage({ type: hook.type, name: hook.name }); postInits.push(t.staggerOpts?.initialInterval ?? 0); postMaxStagger.push(t.staggerOpts?.maxRandomStagger ?? 0) } this.staggerMappers.postCompare = staggerMapper<PlayObject, PlayObject>({ initialInterval: Math.max(...postInits), maxRandomStagger: Math.max(...postMaxStagger), concurrency: 2 }); } }
protected getIdentifier() { return `${capitalize(this.type)} - ${this.name}` } protected getMachineId() { return `${this.type}-${this.name}`; } public getSafeExternalName() { return normalizeStr(this.name, {keepSingleWhitespace: false}); } public getSafeExternalId() { return `${this.type}-${normalizeStr(this.name, {keepSingleWhitespace: false})}`; }
protected getPrometheusLabels() { return {name: this.getSafeExternalName(), type: this.type}; }
public getRunningState(): ComponentState { const running = (this.canPoll && this.polling) || !this.canPoll;
if(running && !this.isMonitoring()) { return COMPONENT_STATE.MUTED; } return running ? COMPONENT_STATE.RUNNING : COMPONENT_STATE.IDLE; }
protected getComponentApiData() { return { hasAuth: this.requiresAuth, hasAuthInteraction: this.requiresAuthInteraction, authed: this.authed, } }
public getApiData(): ComponentSourceApiJson { return { ...super.getApiData(), ...this.getComponentApiData(), type: this.type, status: this.status, players: {}, tracksDiscovered: this.tracksDiscovered, sot: SOURCE_SOT.HISTORY, supportsUpstreamRecentlyPlayed: this.supportsUpstreamRecentlyPlayed, sleeping: this.getIsSleeping(), wakeAt: this.wakeAt !== undefined ? this.wakeAt.toISOString() : undefined, countLive: this.tracksDiscoveredTotal } }
getRecentlyPlayed = async (options: RecentlyPlayedOptions = {}): Promise<PlayObject[]> => []
getUpstreamRecentlyPlayed = async (options: RecentlyPlayedOptions = {}): Promise<PlayObject[]> => { throw new Error('Not implemented'); }
getUpstreamNowPlaying = async(): Promise<PlayObject[]> => { throw new Error('Not implemented'); }
// by default if the track was recently played it is valid // this is useful for sources where the track doesn't have complete information like Subsonic // TODO make this more descriptive? or move it elsewhere recentlyPlayedTrackIsValid = (playObj: PlayObject) => true
protected addPlayToDB = async (play: PlayObject): Promise<PlayWith<'input'>> => { const monitorStatus = this.getMonitoringStatus(); let state: PlayState = 'discovered'; if(!monitorStatus.monitoring) { this.logger.debug(`Not adding ${buildTrackString(play)} as discovered because monitoring is disabled by ${capitalize(monitorStatus.origin)}`); state = 'discarded'; } const playRow = await this.playRepo.createPlays([(playToRepositoryCreatePlayOpts({play, state}))]); const recentPlays = await this.getRecentlyDiscoveredPlays(false); // only need to update if its already in memory, // and better to update in-memory than clear cache so we aren't refetching from db on every discover if(recentPlays !== undefined) { recentPlays.push({...play, id: playRow[0].id, uid: playRow[0].uid}); recentPlays.sort(sortByOldestPlayDate); this.cache.cacheDb.set(this.recentDiscoveredCacheKey(), recentPlays, '2m'); } if(state === 'discovered') { this.tracksDiscovered++; this.tracksDiscoveredTotal++ this.discoveredCounter.labels(this.getPrometheusLabels()).inc(); } this.logger.info(`${capitalize(state)} => ${buildTrackString(play)}`); this.emitEvent('discovered', {play}); this.emitPlayInsert({...playRow[0], queueStates: []} as unknown as PlayApiCommonDetailed); playRow[0].play.id = playRow[0].id; playRow[0].play.uid = playRow[0].uid; return playRow[0]; }
getFlatRecentlyDiscoveredPlays = async (): Promise<PlayObject[]> => { const list: PlayObject[] = await this.getRecentlyDiscoveredPlays(); return list.sort(sortByNewestPlayDate); }
getRecentPlaysApi = async (query: RequestPlayQuery) => { const res = await this.playRepo.findPlays({ limit: 100, }); return res.map((x) => { const {id, ...rest} = x; return rest; }) }
protected recentDiscoveredCacheKey = () => { return `recent-${this.dbComponent.id}`; }
getRecentlyDiscoveredPlays = async (hydrate: boolean = true): Promise<PlayObject[]> => { const cacheKey = this.recentDiscoveredCacheKey(); let list = await this.cache.cacheDb.get<PlayObject[]>(cacheKey); if(list === undefined && hydrate) { list = (await this.playRepo.findPlays({ stateNot: ['queued'], order: 'desc', sort: 'playedAt', limit: 200 })).map(x => ({...asPlay(x.play), id: x.id, uid: x.uid})) list.sort(sortByOldestPlayDate); await this.cache.cacheDb.set<PlayObject[]>(cacheKey, list, '2m'); } return list; }
existingDiscovered = async (play: PlayObject): Promise<PlayObject | undefined> => { const list: PlayObject[] = await this.getRecentlyDiscoveredPlays(true); const matchResults = await this.existingDiscoveredPlay(play, list); if(matchResults.match) { return matchResults.closestMatchedPlay; } return undefined; }
discover = async (plays: PlayObject[], options: { checkAll?: boolean, signal?: AbortSignal, discoverLocation?: string, [key: string]: any } = {}): Promise<PlayObject[]> => { const newDiscoveredPlays: PlayObject[] = [];
this.setStatus(`Discovering new Plays${options.discoverLocation !== undefined ? ` from ${options.discoverLocation} ` : ''}`); for await(const play of pMapIterable(plays, this.staggerMappers.preCompare(async x => await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 3})) { options.signal?.throwIfAborted(); const existing = await this.existingDiscovered(play); if(existing === undefined) { options.signal?.throwIfAborted() const hydratedPlay = await this.addPlayToDB(play); if(hydratedPlay.state === 'discovered') { newDiscoveredPlays.push(hydratedPlay.play); } } else { this.playRepo.updateById(existing.id, {updatedAt: dayjs()}); } } if(newDiscoveredPlays.length > 0) { this.setStatus(`Discovered ${newDiscoveredPlays.length} new Plays${options.discoverLocation !== undefined ? ` from ${options.discoverLocation} ` : ''}`); try { await this.componentRepo.updateById(this.dbComponent.id, {countLive: this.dbComponent.countLive + newDiscoveredPlays.length}); } catch (e) { this.logger.warn(new Error('Unable to update discovered count', {cause: e})); } } else { this.setStatus(`No new Plays discovered${options.discoverLocation !== undefined ? ` from ${options.discoverLocation} ` : ''}`); } newDiscoveredPlays.sort(sortByOldestPlayDate);
return newDiscoveredPlays; }
protected scrobble = async (newDiscoveredPlays: PlayObject[], options: { forceRefresh?: boolean, [key: string]: any, discoverLocation?: 'backlog' | [key: string] } = {}) => {
if(newDiscoveredPlays.length > 0) { newDiscoveredPlays.sort(sortByOldestPlayDate); this.emitEvent('discoveredToScrobble', { data: await pMap(newDiscoveredPlays, this.staggerMappers.postCompare(async (x) => await this.transformPlay(x, TRANSFORM_HOOK.postCompare)), {concurrency: 3}), options: { ...options, checkTime: newDiscoveredPlays[newDiscoveredPlays.length-1].data.playDate.add(2, 'second'), scrobbleFrom: this.getIdentifier(), scrobbleTo: this.clients } }); this.setStatus(`Forwarded ${newDiscoveredPlays.length} new Plays to Clients${options.discoverLocation !== undefined ? ` from ${options.discoverLocation} ` : ''}`); } }
protected processBacklog = async (signal: AbortSignal) => { if (this.canBacklog) {
const { options: { scrobbleBacklog = true } = {} } = this.config;
if(scrobbleBacklog === false) { this.logger.info('Source is able to scrobble backlog but was it disabled by user.'); this.setStatus('Not scrobbling backlog because it was disabled by user'); return; }
this.logger.info('Discovering backlogged tracks from recently played API...'); this.setStatus('Discovering backlogged tracks from recently played API...'); let backlogPlays: PlayObject[]; const { scrobbleBacklogCount = this.SCROBBLE_BACKLOG_COUNT } = this.config.options || {}; let backlogLimit = scrobbleBacklogCount; if(backlogLimit > this.SCROBBLE_BACKLOG_COUNT) { this.logger.warn(`scrobbleBacklogCount (${scrobbleBacklogCount}) cannot be greater than max API limit (${this.SCROBBLE_BACKLOG_COUNT}), reverting to max...`); backlogLimit = this.SCROBBLE_BACKLOG_COUNT; } try { this.logger.verbose(`Fetching the last ${backlogLimit}${backlogLimit === this.SCROBBLE_BACKLOG_COUNT ? ' (max) ' : ''} listens to check for backlogging...`); backlogPlays = await this.getBackloggedPlays({limit: backlogLimit}); signal.throwIfAborted(); } catch (e) { throw new Error('Error occurred while fetching backlogged plays', {cause: e}); } const discovered = await this.discover(backlogPlays, {discoverLocation: 'backlog', signal});
if (scrobbleBacklog) { if (discovered.length > 0) { this.logger.info('Scrobbling backlogged tracks...'); signal.throwIfAborted(); await this.scrobble(discovered); this.logger.info('Backlog scrobbling complete.'); } else { this.logger.info('All tracks already discovered!'); } } else { this.logger.info('Backlog scrobbling is disabled by config, skipping...'); } } return; }
protected getBackloggedPlays = async (options: RecentlyPlayedOptions): Promise<PlayObject[]> => { this.logger.debug('Backlogging not implemented'); return []; }
onPollPreAuthCheck = async (): Promise<boolean> => true
onPollPostAuthCheck = async (): Promise<boolean> => true
poll = async (options: {force?: boolean, notify?: boolean} = {}) => { const {force = false, notify = false} = options;
if(this.polling) { this.logger.error('Already polling!'); return; }
if(!this.isReady() || force) { try { await this.initialize(options); } catch (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'}); } return; } } if(!(await this.onPollPreAuthCheck())) { return; } if(!(await this.onPollPostAuthCheck())) { return; }
this.setStatus('Starting polling...');
this.abortController = new AbortController(); this.pollingPromise = spawn(this.abortController.signal, async (signal, { defer, fork }) => { defer(async () => { this.polling = false; this.isSleeping = false; this.emitEvent('statusChange', {status: 'Idle'}); this.emitComponentUpdate<Partial<ComponentSourceApiJson>>({state: COMPONENT_STATE.IDLE}); });
fork(async (fSignal) => { try { await this.processBacklog(fSignal); } catch (e) { throwIfAborted(fSignal); await this.notify({ title: `Polling Error`, message: 'Polling interrupted because error occurred while processing backlog.', priority: 'error' }); throw new Error('Polling interrupted because error occurred while processing backlog', { cause: e }); } }); await this.startPolling(signal); }).catch((e) => { 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) componentUpdate.status = 'Polling cancelled'; } else { 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>>(componentUpdate); }).finally(() => { this.abortController = undefined; this.pollingPromise = undefined; }); }
startPolling = async (signal: AbortSignal) => { signal.throwIfAborted(); // reset poll attempts if already previously run this.pollRetries = 0;
const { options: { maxPollRetries = 5, retryMultiplier = DEFAULT_RETRY_MULTIPLIER, } } = this.config;
// can't have negative retries! const maxRetries = Math.max(0, maxPollRetries);
if(this.polling === true) { this.logger.warn(`Already polling! Polling needs to be stopped before it can be started`); return; }
while (this.pollRetries <= maxRetries) { try { if(!this.isReady() && this.buildOK) { this.logger.verbose(`Source is no longer ready! Will attempt to reinitialize => Connection OK: ${this.connectionOK} | Auth OK: ${this.authed}`); const init = await this.initialize(); if(init === false) { throw new Error('Source failed reinitialization'); } signal.throwIfAborted(); } await this.doPolling(signal); } catch (e) { if(isAbortError(e)) { throw e; } if (this.pollRetries < maxRetries) { const delayFor = pollingBackoff(this.pollRetries + 1, retryMultiplier); this.logger.info(`Poll retries (${this.pollRetries}) less than max poll retries (${maxRetries}), restarting polling after ${delayFor} second delay...`); await this.notify({title: `Polling Retry`, message: `Encountered error while polling but retries (${this.pollRetries}) are less than max poll retries (${maxRetries}), restarting polling after ${delayFor} second delay. | Error: ${e.message}`, priority: 'warn'}); await sleep((delayFor) * 1000); this.pollRetries++; } else { this.logger.warn(`Poll retries (${this.pollRetries}) equal to max poll retries (${maxRetries}), stopping polling!`); await this.notify({title: `Polling Error`, message: `Encountered error while polling and retries (${this.pollRetries}) are equal to max poll retries (${maxRetries}), stopping polling!. | Error: ${e.message}`, priority: 'error'}); throw e; } } } }
tryStopPolling = async (reason?: string | Error) => { if(this.polling === false) { this.logger.warn(`Polling is already stopped!`); return true; } if(this.abortController === undefined) { this.logger.error('No abort controller found! Nothing to stop.'); return false; } this.abortController.abort(reason); let elapsed = 0; while(this.polling && elapsed < (10 * this.stopPollingWaitInterval)) { this.logger.verbose(`Waiting for polling stop signal to be acknowledged (waited ${formatNumber(elapsed/1000)}s)`); await sleep(this.stopPollingWaitInterval); elapsed += this.stopPollingWaitInterval; } if(this.polling) { this.logger.warn('Could not stop polling! Or polling signal was lost :('); return false; } return true; }
protected doPolling = async (signal: AbortSignal): Promise<true | undefined> => { signal.throwIfAborted();
this.logger.info('Polling started'); this.emitEvent('statusChange', {status: 'Running'}); this.emitComponentUpdate<Partial<ComponentSourceApiJson>>({state: COMPONENT_STATE.RUNNING}); await this.notify({title: `Polling Started`, message: 'Polling Started', priority: 'info'}); this.setStatus('Polling Started'); this.lastActivityAt = dayjs(); let checkCount = 0; let checksOverThreshold = 0; const checkActiveFor = 120; let maxInterval = DEFAULT_POLLING_MAX_INTERVAL;
if('maxInterval' in this.config.data) { maxInterval = this.config.data.maxInterval; } let isInactive = false;
try { this.polling = true; while (true) { signal.throwIfAborted(); const pollFrom = dayjs(); let lastActivityLogLevel: LogLevel = 'trace';
let playObjs: PlayObject[]; try { playObjs = await this.getRecentlyPlayed({formatted: true}); } catch (e) { throw new Error('Error occurred while refreshing recently played', {cause: e}); } finally { signal.throwIfAborted(); }
const interval = this.getInterval(true); const maxBackoff = this.getMaxBackoff(); let sleepTime = interval;
let newDiscovered: PlayObject[] = [];
if(playObjs.length > 0) { const now = dayjs().unix(); const closeToInterval = playObjs.some(x => now - x.data.playDate.unix() < 5); if (playObjs.length > 0 && closeToInterval) { // because the interval check was so close to the play date we are going to delay client calls for a few secs // this way we don't accidentally scrobble ahead of any other clients (we always want to be behind so we can check for dups) // additionally -- it should be ok to have this in the for loop because played_at will only decrease (be further in the past) so we should only hit this once, hopefully
// make sure delay is less than possible polling interval const maxDelay = Math.min(10, interval * 0.75); this.logger.info(`Potential plays were discovered close to polling interval! Delaying scrobble clients refresh by ${maxDelay} seconds so other clients have time to scrobble first`); await sleep(maxDelay * 1000); } newDiscovered = await this.discover(playObjs, {signal}); signal.throwIfAborted(); this.scrobble(newDiscovered, { forceRefresh: closeToInterval }); }
const activityMsgs: string[] = [];
if(newDiscovered.length > 0) { // only update date if the play date is after the current activity date (in the case of backlogged plays) this.lastActivityAt = newDiscovered[0].data.playDate.isAfter(this.lastActivityAt) ? newDiscovered[0].data.playDate : this.lastActivityAt; checkCount = 0; checksOverThreshold = 0; }
const activeThreshold = this.lastActivityAt.add(checkActiveFor, 's'); const inactiveFor = dayjs.duration(Math.abs(activeThreshold.diff(dayjs(), 'millisecond'))).humanize(false); const relativeActivity = dayjs.duration(this.lastActivityAt.diff(dayjs(), 'ms')); const humanRelativeActivity = relativeActivity.asSeconds() > -3 ? '' : ` (${timeToHumanTimestamp(relativeActivity)} ago)`; let friendlyInterval = `${formatNumber(sleepTime)}`; const friendlyLastFormat = todayAwareFormat(this.lastActivityAt); activityMsgs.push(`Last activity at ${friendlyLastFormat}${humanRelativeActivity}`); if (activeThreshold.isBefore(dayjs())) { friendlyInterval = formatNumber(maxInterval); checksOverThreshold++; if(sleepTime < maxInterval) { const checkVal = Math.min(checksOverThreshold, 1000); const backoff = Math.round(Math.max(Math.min(Math.min(checkVal, 1000) * 2 * (1.1 * checkVal), maxBackoff), 5)); friendlyInterval = `(${interval} + ${backoff})`; sleepTime = interval + backoff; } if(!isInactive) { lastActivityLogLevel = 'debug'; isInactive = true; } activityMsgs.push(`Inactive for ${inactiveFor} (last + ${checkActiveFor}s)`); } else if(isInactive) { activityMsgs.push('New Activity after inactive period'); lastActivityLogLevel = 'debug'; isInactive = false; } activityMsgs.push(`Next check in ${friendlyInterval}s`); this.logger[lastActivityLogLevel](activityMsgs.join(' | ')); this.setWakeAt(pollFrom.add(sleepTime, 'seconds')); this.setIsSleeping(true); this.emitComponentUpdate<Partial<ComponentSourceApiJson>>({sleeping: true, wakeAt: this.getWakeAt().toISOString()}) // set last active before we sleep this.updateDates({lastActiveAt: dayjs(), lastReadyAt: dayjs()}); while(dayjs().isBefore(this.getWakeAt())) { // check for polling status every half second and wait till wake up time await delay(signal, 500); } this.setIsSleeping(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; } } catch (e) { if(!isAbortError(e)) { this.logger.error(new Error('Error occurred while polling', {cause: e})); } if(e.message.includes('Status code: 401')) { this.authed = false; this.authFailure = true; } throw e; } finally { this.setIsSleeping(false); this.emitComponentUpdate<Partial<ComponentSourceApiJson>>({sleeping: false}); } }
protected setIsSleeping(sleeping: boolean) { this.isSleeping = sleeping; }
protected getIsSleeping() { return this.isSleeping; }
protected setWakeAt(dt: Dayjs) { this.wakeAt = dt; }
protected getWakeAt() { return this.wakeAt; }
protected getInterval(log?: boolean) { let interval = DEFAULT_POLLING_INTERVAL;
if('interval' in this.config.data) { interval = this.config.data.interval; } return interval; }
protected getMaxBackoff() { let maxInterval = DEFAULT_POLLING_MAX_INTERVAL;
if('maxInterval' in this.config.data) { maxInterval = this.config.data.maxInterval; } return maxInterval - this.getInterval(); }
public async getPlaysPaginated(args: QueryPlaysOpts): Promise<PaginatedResponse<PlayApiCommonDetailed>> { const { limit, offset, with: withQuery = ['input','parent-input','queues'], ...rest } = args; const parsedLimit = limit !== undefined ? Number.parseInt(limit as unknown as string) : undefined; const parsedOffset = offset !== undefined ? Number.parseInt(offset as unknown as string) : undefined; return this.playRepo.findPlaysPaginated({limit: parsedLimit, offset: parsedOffset, with: withQuery, ...rest}); }
public async getPlayApiResponse(uid: string, opts: {with?: WithPlayRelation[]} = {}): Promise<PlayApiCommonDetailed> { const { with: withQuery = ['input','parent-input','queues'], } = opts; return await this.playRepo.findByUid(uid, { with: withQuery as WithPlayRelation[] }) as unknown as PlayApiCommonDetailed; }
public emitEvent = (eventName: string, payload: object = {}) => { this.emitter.emit(eventName, { type: this.type, name: this.name, componentId: this.dbComponent?.id, from: 'source', data: payload, }); }
public async destroy() { this.emitter.removeAllListeners(); }
protected async doBuildComponentLogger(): Promise<void> { if(this.config?.options?.logToFile) { this.logger.debug('Enabling component logger...'); const root = getRoot(); const stream = root.get('loggerStream'); const logConfig = root.get('loggingConfig'); const cLogger = await componentFileLogger(this.type, this.name, true, logConfig); this.componentLogger = childLogger(cLogger, this.logger.labels); stream.on('data', (d: LogDataPretty) => { const {level, msg, line, labels, ...rest} = d; if(d.labels.includes(this.loggerLabel)) { this.componentLogger[this.componentLogger.levels.labels[d.level]]({...rest, labels: difference(labels, this.logger.labels)}, msg); } }); } }}