feat: Initialize usage refactored everywhere · foxxmd.dev/multi-scrobbler@626bca9

+15 -12

src/backend/common/AbstractComponent.ts

··· 25 25 import play = Simulate.play; 26 26 import { WebhookPayload } from "./infrastructure/config/health/webhooks.js"; 27 27 import { AuthCheckError, BuildDataError, ConnectionCheckError, PostInitError, TransformRulesError } from "./errors/MSErrors.js"; 28 - import { messageWithCauses } from "../utils/ErrorUtils.js"; 28 + import { messageWithCauses, messageWithCausesTruncatedDefault } from "../utils/ErrorUtils.js"; 29 29 30 30 export default abstract class AbstractComponent { 31 31 requiresAuth: boolean = false; ··· 50 50 this.config = config; 51 51 } 52 52 53 - protected abstract notify(payload: WebhookPayload): Promise<void>; 53 + public abstract notify(payload: WebhookPayload): Promise<void>; 54 54 55 55 protected abstract getIdentifier(): string; 56 56 57 - // TODO refactor throw error 58 - initialize = async (options: {force?: boolean, notify?: boolean} = {}) => { 57 + initialize = async (options: {force?: boolean, notify?: boolean, notifyTitle?: string} = {}) => { 59 58 60 - const {force = false, notify = false} = options; 59 + const {force = false, notify = false, notifyTitle = 'Init Error'} = options; 61 60 62 61 this.logger.debug('Attempting to initialize...'); 63 62 try { ··· 77 76 } 78 77 return true; 79 78 } catch(e) { 80 - this.logger.error(new Error('Initialization failed', {cause: e})); 81 79 if(notify) { 82 - await this.notify({title: `${this.getIdentifier()} - Init Error`, message: truncateStringToLength(150)(messageWithCauses(e)), priority: 'error'}); 80 + await this.notify({title: `${this.getIdentifier()} - ${notifyTitle}`, message: truncateStringToLength(500)(messageWithCausesTruncatedDefault(e)), priority: 'error'}); 83 81 } 84 - return false; 82 + throw new Error('Initialization failed', {cause: e}); 85 83 } finally { 86 84 this.initializing = false; 87 85 } ··· 96 94 return; 97 95 } 98 96 99 - tryInitialize = async (options: {force?: boolean, notify?: boolean} = {}) => { 97 + tryInitialize = async (options: {force?: boolean, notify?: boolean, notifyTitle?: string} = {}) => { 100 98 if(this.initializing) { 101 - this.logger.warn(`Already trying to initialize, cannot attempt while an existing initialization attempt is running.`); 102 - return; 99 + throw new Error(`Already trying to initialize, cannot attempt while an existing initialization attempt is running.`) 100 + } 101 + try { 102 + return await this.initialize(options); 103 + } catch (e) { 104 + throw e; 103 105 } 104 - return await this.initialize(options); 105 106 } 106 107 107 108 public async buildInitData(force: boolean = false) { ··· 257 258 authGated = () => this.requiresAuth && !this.authed 258 259 259 260 canTryAuth = () => this.isUsable() && this.authGated() && this.authFailure !== true 261 + 262 + canAuthUnattended = () => !this.authGated || !this.requiresAuthInteraction || (this.requiresAuthInteraction && !this.authFailure); 260 263 261 264 protected doAuthentication = async (): Promise<boolean> => this.authed 262 265

+25 -33

src/backend/index.ts

··· 15 15 import { initServer } from "./server/index.js"; 16 16 import { createHeartbeatClientsTask } from "./tasks/heartbeatClients.js"; 17 17 import { createHeartbeatSourcesTask } from "./tasks/heartbeatSources.js"; 18 - import { parseBool, readJson, sleep } from "./utils.js"; 18 + import { parseBool, readJson, retry, sleep } from "./utils.js"; 19 19 import { createVegaGenerator } from './utils/SchemaUtils.js'; 20 + import ScrobbleClients from './scrobblers/ScrobbleClients.js'; 21 + import ScrobbleSources from './sources/ScrobbleSources.js'; 20 22 21 23 dayjs.extend(utc) 22 24 dayjs.extend(isBetween); ··· 106 108 /* 107 109 * setup clients 108 110 * */ 109 - const scrobbleClients = root.get('clients'); 111 + const scrobbleClients = root.get('clients') as ScrobbleClients; 110 112 await scrobbleClients.buildClientsFromConfig(notifiers); 111 - if (scrobbleClients.clients.length === 0) { 112 - logger.warn('No scrobble clients were configured!') 113 - } else { 114 - logger.info('Starting scrobble clients...'); 115 - } 116 - for(const client of scrobbleClients.clients) { 117 - await client.initScrobbleMonitoring(); 118 - } 119 - 120 - const scrobbleSources = root.get('sources'); 113 + /* 114 + * setup sources 115 + * */ 116 + const scrobbleSources = root.get('sources') as ScrobbleSources; 121 117 await scrobbleSources.buildSourcesFromConfig([]); 122 - for(const source of scrobbleSources.sources) { 123 - if(!source.isReady()) { 124 - await source.initialize(); 125 - } 126 - } 127 - scrobbleSources.logger.info('Finished initializing sources'); 128 118 129 119 // check ambiguous client/source types like this for now 130 120 const lastfmSources = scrobbleSources.getByType('lastfm'); ··· 136 126 logger.warn(`Last.FM source and clients have same names [${nameColl.map(x => x.name).join(',')}] -- this may cause issues`); 137 127 } 138 128 139 - let anyNotReady = false; 140 - for (const source of scrobbleSources.sources.filter(x => x.canPoll === true)) { 141 - await sleep(1500); // stagger polling by 1.5 seconds so that log messages for each source don't get mixed up 142 - if(source.isReady()) { 143 - source.poll(); 144 - } else { 145 - anyNotReady = true; 146 - } 129 + const clientTask = createHeartbeatClientsTask(scrobbleClients, logger); 130 + clientTask.execute(); 131 + try { 132 + await retry(() => { 133 + if(clientTask.isExecuting) { 134 + throw new Error('Waiting') 135 + } 136 + return true; 137 + },{retries: scrobbleClients.clients.length + 1, retryIntervalMs: 2000}); 138 + } catch (e) { 139 + logger.warn('Waited too long for clients to start! Moving ahead with sources init...'); 147 140 } 148 - if (anyNotReady) { 149 - logger.info(`Some sources are not ready, open the dashboard to continue`); 150 - } 151 - 152 141 scheduler.addSimpleIntervalJob(new SimpleIntervalJob({ 153 142 minutes: 20, 154 143 runImmediately: false 155 - }, createHeartbeatSourcesTask(scrobbleSources, logger))); 144 + }, clientTask, {id: 'clients_heart'})); 145 + 146 + const sourceTask = createHeartbeatSourcesTask(scrobbleSources, logger); 156 147 scheduler.addSimpleIntervalJob(new SimpleIntervalJob({ 157 148 minutes: 20, 158 - runImmediately: false 159 - }, createHeartbeatClientsTask(scrobbleClients, logger))); 149 + runImmediately: true 150 + }, sourceTask, {id: 'sources_heart'})); 151 + 160 152 logger.info('Scheduler started.'); 161 153 162 154 } catch (e) {

+11 -30

src/backend/scrobblers/AbstractScrobbleClient.ts

··· 37 37 sleep, 38 38 sortByOldestPlayDate, 39 39 } from "../utils.js"; 40 - import { messageWithCauses } from "../utils/ErrorUtils.js"; 40 + import { messageWithCauses, messageWithCausesTruncatedDefault } from "../utils/ErrorUtils.js"; 41 41 import { compareScrobbleArtists, compareScrobbleTracks, normalizeStr } from "../utils/StringUtils.js"; 42 42 import { 43 43 comparePlayTemporally, ··· 148 148 return `${capitalize(this.type)} - ${this.name}` 149 149 } 150 150 151 - protected notify = async (payload: WebhookPayload) => { 151 + public notify = async (payload: WebhookPayload) => { 152 152 this.emitEvent('notify', payload); 153 153 } 154 154 ··· 506 506 507 507 public abstract playToClientPayload(playObject: PlayObject): object 508 508 509 - initScrobbleMonitoring = async () => { 510 - // TODO refactor to only use tryInitialize 511 - if(!this.isUsable()) { 512 - if(this.initializing) { 513 - this.logger.warn(`Cannot start scrobble processing because client is still initializing`); 514 - return; 515 - } 516 - if(!(await this.initialize())) { 517 - this.logger.warn(`Cannot start scrobble processing because client could not be initialized`); 518 - return; 519 - } 520 - } 509 + initScrobbleMonitoring = async (options: {force?: boolean, notify?: boolean} = {}) => { 510 + const {force = false, notify = false} = options; 521 511 522 - if (this.authGated()) { 523 - if (this.canTryAuth()) { 524 - try { 525 - await this.testAuth(); 526 - } catch (e) { 527 - this.logger.warn(new Error('Cannot start scrobbling process due to auth issue', { cause: e })); 512 + if(!this.isReady() || force) { 513 + try { 514 + await this.tryInitialize(options); 515 + } catch (e) { 516 + this.logger.error(new Error('Cannot start monitoring because Client is not ready', {cause: e})); 517 + if(notify) { 518 + await this.notify( {title: `${this.getIdentifier()} - Processing Error`, message: `Cannot start monitoring because Client is not ready: ${truncateStringToLength(500)(messageWithCausesTruncatedDefault(e))}`, priority: 'error'}); 528 519 } 529 - } else if (this.requiresAuthInteraction) { 530 - this.logger.warn(`Cannot start scrobble processing because user interaction is required for authentication`); 531 - return; 532 - } else { 533 - this.logger.warn(`Cannot start scrobble processing because client needs to be reauthenticated.`); 534 520 return; 535 521 } 536 - } 537 - 538 - if(!this.isReady()) { 539 - this.logger.warn(`Cannot start scrobble processing because client is not ready`); 540 - return; 541 522 } 542 523 543 524 this.startScrobbling().catch((e) => {

+10 -5

src/backend/server/auth.ts

··· 83 83 } 84 84 try { 85 85 await entity.api.authenticate(token); 86 - await entity.initialize(); 87 - await entity.doAuthentication(); 86 + entity.authFailure = false; 87 + if(entity instanceof LastfmSource) { 88 + entity.poll().catch((e) => logger.error(e)); 89 + } else { 90 + entity.tryInitialize().catch((e) => logger.error(e)); 91 + } 88 92 return res.send('OK'); 89 93 } catch (e) { 90 94 return res.send(e.message); ··· 97 101 const result = await entity.handleAuthCodeCallback(req.query); 98 102 let responseContent = 'OK'; 99 103 if(result === true) { 100 - entity.poll(); 104 + entity.authFailure = false; 105 + entity.poll().catch((e) => logger.error(e)); 101 106 } else { 102 107 responseContent = result; 103 108 } ··· 111 116 const tokenResult = await source.handleAuthCodeCallback(req.query); 112 117 let responseContent = 'OK'; 113 118 if (tokenResult === true) { 114 - await source.testAuth(); 115 - source.poll(); 119 + source.authFailure = false; 120 + source.poll().catch((e) => logger.error(e)); 116 121 } else { 117 122 responseContent = tokenResult; 118 123 }

+11 -21

src/backend/sources/AbstractSource.ts

··· 3 3 import { EventEmitter } from "events"; 4 4 import { FixedSizeList } from "fixed-size-list"; 5 5 import { PlayObject, TA_CLOSE } from "../../core/Atomic.js"; 6 - import { buildTrackString, capitalize } from "../../core/StringUtils.js"; 6 + import { buildTrackString, capitalize, truncateStringToLength } from "../../core/StringUtils.js"; 7 7 import AbstractComponent from "../common/AbstractComponent.js"; 8 8 import { 9 9 Authenticatable, ··· 36 36 import { getRoot } from '../ioc.js'; 37 37 import { componentFileLogger } from '../common/logging.js'; 38 38 import { WebhookPayload } from '../common/infrastructure/config/health/webhooks.js'; 39 + import { messageWithCauses, messageWithCausesTruncatedDefault } from '../utils/ErrorUtils.js'; 39 40 40 41 export interface RecentlyPlayedOptions { 41 42 limit?: number ··· 255 256 return []; 256 257 } 257 258 258 - protected notify = async (payload: WebhookPayload) => { 259 + public notify = async (payload: WebhookPayload) => { 259 260 this.emitter.emit('notify', payload); 260 261 } 261 262 ··· 268 269 269 270 // TODO refactor to only use tryInitialize 270 271 if(!this.isReady() || force) { 271 - const ready = await this.tryInitialize(options); 272 - if(!ready) { 273 - // TODO 272 + try { 273 + await this.tryInitialize(options); 274 + } catch (e) { 275 + this.logger.error(new Error('Cannot start polling because Source is not ready', {cause: e})); 276 + if(notify) { 277 + await this.notify( {title: `${this.getIdentifier()} - Polling Error`, message: `Cannot start polling because Source is not ready: ${truncateStringToLength(500)(messageWithCausesTruncatedDefault(e))}`, priority: 'error'}); 278 + } 279 + return; 274 280 } 275 281 } 276 282 if(!(await this.onPollPreAuthCheck())) { 277 - return; 278 - } 279 - if(this.authGated()) { 280 - if(this.canTryAuth()) { 281 - await this.testAuth(); 282 - if(!this.authed) { 283 - await this.notify( {title: `${this.getIdentifier()} - Polling Error`, message: 'Cannot start polling because source does not have authentication.', priority: 'error'}); 284 - this.logger.error('Cannot start polling because source is not authenticated correctly.'); 285 - } 286 - } else if(this.requiresAuthInteraction) { 287 - await this.notify({title: `${this.getIdentifier()} - Polling Error`, message: 'Cannot start polling because user interaction is required for authentication', priority: 'error'}); 288 - this.logger.error('Cannot start polling because user interaction is required for authentication'); 289 - } else { 290 - await this.notify( {title: `${this.getIdentifier()} - Polling Error`, message: 'Cannot start polling because source authentication previously failed and must be reauthenticated.', priority: 'error'}); 291 - this.logger.error('Cannot start polling because source authentication previously failed and must be reauthenticated.'); 292 - } 293 283 return; 294 284 } 295 285 if(!(await this.onPollPostAuthCheck())) {

+20 -16

src/backend/tasks/heartbeatClients.ts

··· 14 14 .withConcurrency(1) 15 15 .for(clients.clients) 16 16 .process(async (client) => { 17 - const ready = client.isReady(); 18 - if(ready || client.canTryAuth()) { 19 - if(!ready) { 20 - client.logger.info('Trying client auth...'); 21 - await client.testAuth(); 22 - if(!client.authed) { 23 - return 0; 24 - } 25 - if(!client.isReady()) { 26 - return 0; 27 - } 17 + if(!client.isReady()) { 18 + if(!client.canAuthUnattended()) { 19 + client.logger.warn({label: 'Heartbeat'}, 'Client is not ready but will not try to initialize because auth state is not good and cannot be correct unattended.') 20 + return 0; 28 21 } 29 - await client.processDeadLetterQueue(); 30 - if(!client.scrobbling) { 31 - client.logger.info('Should be processing scrobbles! Attempting to restart scrobbling...', {leaf: 'Heartbeat'}); 32 - client.initScrobbleMonitoring(); 22 + try { 23 + await client.tryInitialize({force: false, notify: true, notifyTitle: 'Could not initialize automatically'}); 24 + } catch (e) { 25 + client.logger.error(new Error('Could not initialize automatically', {cause: e})); 33 26 return 1; 34 27 } 28 + } 29 + 30 + if(!client.canAuthUnattended()) { 31 + client.logger.warn({label: 'Heartbeat'}, 'Should be monitoring scrobbles but will not attempt to start because auth state is not good and cannot be correct unattended.'); 35 32 return 0; 36 33 } 34 + 35 + await client.processDeadLetterQueue(); 36 + if(!client.scrobbling) { 37 + client.logger.info({label: 'Heartbeat'}, 'Should be processing scrobbles! Attempting to restart scrobbling...'); 38 + client.initScrobbleMonitoring(); 39 + return 1; 40 + } 37 41 }).then(({results, errors}) => { 38 42 logger.verbose(`Checked Dead letter queue for ${clients.clients.length} clients.`); 39 43 const restarted = results.reduce((acc, curr) => acc += curr, 0); 40 44 if (restarted > 0) { 41 - logger.info(`Attempted to restart ${restarted} clients that were not processing scrobbles.`); 45 + logger.info(`Attempted to start ${restarted} clients that were not processing scrobbles.`); 42 46 } 43 47 if (errors.length > 0) { 44 48 logger.error(`Encountered errors!`);

+26 -8

src/backend/tasks/heartbeatSources.ts

··· 15 15 .withConcurrency(1) 16 16 .for(sources.sources) 17 17 .process(async (source) => { 18 - if(source.isReady()) { 19 - if(source.type === 'chromecast') { 20 - (source as ChromecastSource).discoverDevices(); 18 + if(!source.isReady()) { 19 + if(!source.canAuthUnattended()) { 20 + source.logger.warn({label: 'Heartbeat'}, 'Source is not ready but will not try to initialize because auth state is not good and cannot be correct unattended.'); 21 + return 0; 21 22 } 22 - if (source.canPoll && !source.polling && (!source.authGated() || source.canTryAuth())) { 23 - source.logger.info('Should be polling! Attempting to restart polling...', {leaf: 'Heartbeat'}); 24 - source.poll().catch(e => logger.error(e)); 23 + try { 24 + await source.tryInitialize({force: false, notify: true, notifyTitle: 'Could not initialize automatically'}); 25 + } catch (e) { 26 + source.logger.error(new Error('Could not initialize source automatically', {cause: e})); 25 27 return 1; 26 28 } 27 29 } 30 + 31 + if(source.type === 'chromecast') { 32 + (source as ChromecastSource).discoverDevices(); 33 + } 34 + 35 + if (source.canPoll && !source.polling) { 36 + if(!source.canAuthUnattended()) { 37 + source.logger.warn({label: 'Heartbeat'}, 'Should be polling but will not attempt to start because auth state is not good and cannot be correct unattended.'); 38 + return 0; 39 + } else { 40 + source.logger.info({label: 'Heartbeat'}, 'Should be polling, attempting to start polling...'); 41 + source.poll({force: false, notify: true}).catch(e => source.logger.error(e)); 42 + } 43 + return 1; 44 + } 45 + 28 46 return 0; 29 47 }).then(({results, errors}) => { 30 - logger.verbose(`Checked ${sources.sources.length} sources for restart signals.`); 48 + logger.verbose(`Checked ${sources.sources.length} sources for start signals.`); 31 49 const restarted = results.reduce((acc, curr) => acc += curr, 0); 32 50 if (restarted > 0) { 33 - logger.info(`Attempted to restart ${restarted} sources that were not polling.`); 51 + logger.info(`Attempted to start ${restarted} sources.`); 34 52 } 35 53 if (errors.length > 0) { 36 54 logger.error(`Encountered errors!`);

+24

src/backend/utils.ts

··· 667 667 } 668 668 669 669 export const getFirstNonEmptyString = (values: unknown[]) => getFirstNonEmptyVal<string>(values, {ofType: 'string', test: (v) => v.trim() !== ''}); 670 + 671 + /** 672 + * Runs the function `fn` 673 + * and retries automatically if it fails. 674 + * 675 + * Tries max `1 + retries` times 676 + * with `retryIntervalMs` milliseconds between retries. 677 + * 678 + * From https://mtsknn.fi/blog/js-retry-on-fail/ 679 + */ 680 + export const retry = async <T>( 681 + fn: () => Promise<T> | T, 682 + { retries, retryIntervalMs }: { retries: number; retryIntervalMs: number } 683 + ): Promise<T> => { 684 + try { 685 + return await fn() 686 + } catch (error) { 687 + if (retries <= 0) { 688 + throw error 689 + } 690 + await sleep(retryIntervalMs) 691 + return retry(fn, { retries: retries - 1, retryIntervalMs }) 692 + } 693 + }

+17 -5

src/backend/utils/ErrorUtils.ts

··· 1 + import { truncateStringToLength } from "../../core/StringUtils.js"; 2 + 1 3 /** 2 4 * Adapted from https://github.com/voxpelli/pony-cause/blob/main/lib/helpers.js to find cause by truthy function 3 5 * */ ··· 77 79 currentErr = getErrorCause(currentErr); 78 80 } 79 81 }; 82 + 83 + export type MessageTransformer = (val: string) => string; 84 + export const MessageTransformerDefault = (val: string) => val; 80 85 /** 81 86 * Adapted from https://github.com/voxpelli/pony-cause 82 87 * */ 83 - const _messageWithCauses = (err: Error, seen = new Set<Error>()) => { 88 + const _messageWithCauses = (err: Error, seen = new Set<Error>(), msgTransform: MessageTransformer = MessageTransformerDefault) => { 84 89 if (!(err instanceof Error)) return ''; 85 90 86 91 const message = err.message; 87 92 88 93 // Ensure we don't go circular or crazily deep 89 94 if (seen.has(err)) { 90 - return message + ' => ...'; 95 + return msgTransform(message) + ' => ...'; 91 96 } 92 97 93 98 const cause = getErrorCause(err); ··· 95 100 if (cause) { 96 101 seen.add(err); 97 102 98 - return (message + ' => ' + 103 + return (msgTransform(message) + ' => ' + 99 104 _messageWithCauses(cause, seen)); 100 105 } else { 101 - return message; 106 + return msgTransform(message); 102 107 } 103 108 }; 104 109 /** 105 110 * Adapted from https://github.com/voxpelli/pony-cause 106 111 * */ 107 - export const messageWithCauses = (err: Error) => _messageWithCauses(err); 112 + export const messageWithCauses = (err: Error, msgTransformer?: MessageTransformer) => _messageWithCauses(err, new Set<Error>(), msgTransformer); 113 + 114 + export const messageWithCausesTruncated = (length: number) => { 115 + const t = truncateStringToLength(length); 116 + return (err: Error) => messageWithCauses(err, t); 117 + } 118 + 119 + export const messageWithCausesTruncatedDefault = messageWithCausesTruncated(100);

Configure Feed

Configure Feed