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.
55 kB · 1154 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155import { childLogger, type Logger } from "@foxxmd/logging";import { cacheFunctions,} from "@foxxmd/regex-buddy-core";import type EventEmitter from "events";import {COMPONENT_TYPE_CLIENT, DEAD_LETTER_RETRIES_DEFAULT, DEAD_QUEUE, INGRESS_QUEUE, isPlayObject, MONITORING_ORIGIN_SYSTEM, MONITORING_ORIGIN_USER, QUEUE_STATUS_COMPLETED, type ComponentType, type LifecycleInput, type LifecycleStep, type PlayData, type PlayObject} from "../../core/Atomic.ts";import { capitalize } from "../../core/StringUtils.ts";import { buildTrackString } from "../../core/MusicMetadata.ts";import type {CommonClientConfig} from "./infrastructure/config/client/index.ts";import type {CommonSourceConfig} from "./infrastructure/config/source/index.ts";import { generateLoggableAbortReason, mergeSimpleError, SimpleError, SkipTransformStageError, StageChangeError, StagePrerequisiteError, StageTransformError, TransformRulesError } from "./errors/MSErrors.ts";import { FLOW_CONTROL_TERM, type PlayTransformRules, type StageConfig, TRANSFORM_HOOK, type TransformHook} from "../../core/Transform.ts";import AbstractInitializable from "./AbstractInitializable.ts";import type TransformerManager from "./transforms/TransformerManager.ts";import { getRoot } from "../ioc.ts";import { nanoid } from "nanoid";import { isDebugMode, sleep } from "../utils.ts";import { findCauseByFunc, findCauseByReference } from "../utils/ErrorUtils.ts";import { hashObject, parseArrayFromMaybeString } from "../utils/StringUtils.ts";import { normalizeStr } from "../../core/StringUtils.ts";import { playContentCacheHash } from "../utils/PlayComparisonUtils.ts";import type { MSCache } from "./Cache.ts";import { diffObjects, diffObjectsConsoleOutput, patchObject } from "../../core/DataUtils.ts";import clone from "clone";import { loggerNoop } from "./MaybeLogger.ts";import { objectsEqual } from "../utils/DataUtils.ts";import type {RetentionOptions} from "./infrastructure/config/database.ts";import { getRetentionCompactAfterFromEnv, getRetentionDeleteAfterFromEnv, isCompactableProperty, parseRetentionOptions, parseRetentionOptionsDurations } from "./database/Database.ts";import type {DbConcrete} from "./database/drizzle/drizzleUtils.ts";import type {ComponentSelect, PlayEventNew, PlayEventSelect, PlaySelect, PlaySelectWithQueueStates, PlayWith, QueueStateSelect} from "./database/drizzle/drizzleTypes.ts";import { DrizzlePlayRepository, playToRepositoryCreatePlayOpts, type WithPlayRelation } from "./database/drizzle/repositories/PlayRepository.ts";import type {ClientType, JsonPlayObject, MonitoringStatus, OptionalCacheUsage, PlayMatchResult, QueueContext, TrackData} from "../../core/Atomic.ts";import type {SourceType} from "../../core/Atomic.ts";import { DrizzleComponentRepository } from "./database/drizzle/repositories/ComponentRepository.ts";import dayjs, { type Dayjs } from "dayjs";import { COMPONENT_STATE, type ComponentCommonApi, type ComponentCommonApiJson, type ComponentState, type PlayApiCommonDetailed, type QueueStateApi } from "../../core/Api.ts";import type {WebhookPayload} from "./infrastructure/config/health/webhooks.ts";import type { ElementOf, MarkRequired } from "ts-essentials";import { serializeError } from "serialize-error";import { DrizzleQueueRepository } from "./database/drizzle/repositories/QueueRepository.ts";import { DrizzlePlayEventsRepository } from "./database/drizzle/repositories/PlayEventsRepository.ts";import { entityIsPlayEntity, playUpdateToEvent, queueCompletionStateToPlayEvent, queueStateToPlayEvent, stateChangeToPlayEvent } from "./database/drizzle/entityUtils.ts";import pMap from "p-map";import type { Gauge } from 'prom-client';import type { PlayProcessingResult } from "./infrastructure/PlayProcessing.ts";import { isAbortError } from "abort-controller-x";import { PlayProcessingError } from "./errors/PlayProcessingError.ts";import { type PlayEvent, type PlayEventPlayStateChange } from "../../core/PlayEvent.ts";
export type AbstractComponentConfig = (CommonClientConfig | CommonSourceConfig) & { transformManager?: TransformerManager };
const noopTransform = async (x: any) => x;
export default abstract class AbstractComponent extends AbstractInitializable {
declare config: CommonClientConfig | CommonSourceConfig;
transformRules: PlayTransformRules = {}; regexCache!: ReturnType<typeof cacheFunctions>; protected transformManager: TransformerManager; protected cache: MSCache; protected db!: DbConcrete; protected componentRepo!: DrizzleComponentRepository; protected dbComponent!: ComponentSelect; public playRepo!: DrizzlePlayRepository; protected queueRepo!: DrizzleQueueRepository; protected playEventsRepo!: DrizzlePlayEventsRepository; componentId!: number; protected retentionOpts: RetentionOptions; status: string = 'Waiting to initialize...'; emitter!: EventEmitter;
protected ingressQueueAbortController: AbortController | undefined; protected ingressQueuePromise: Promise<void> | undefined;
monitoringActivity?: boolean | undefined; monitoringActivityDefault: boolean = true;
protected componentType!: ComponentType; type!: ClientType | SourceType; name!: string;
lastActiveAt?: Dayjs; lastReadyAt?: Dayjs; protected lastUpdatedComponentDatesAt?: Dayjs
queuedLength: number = 0; deadLetterLength: number = 0; deadLetterQueued: number = 0;
dupeLogger!: Logger; deadLogger!: Logger;
protected queuedGauge!: Gauge; protected deadLetterGauge!: Gauge;
abstract existingPlay(playObjPre: PlayObject, existingScrobbles: PlayObject[], log?: boolean): Promise<PlayMatchResult>
protected constructor(config: AbstractComponentConfig) { super(config); this.transformManager = config.transformManager ?? getRoot().items.transformerManager; this.cache = getRoot().items.cache(); const cProps = config.options?.retention?.compact ?? parseArrayFromMaybeString(process.env.COMPACT_PROPERTIES ?? 'input,transform', {lower: true}); if(!cProps.every(isCompactableProperty)) { throw new SimpleError(`Compactable properties must be one of 'transform' or 'input'. Given: ${cProps.join(',')}`); } this.retentionOpts = { deleteAfter: parseRetentionOptionsDurations(config.options?.retention?.deleteAfter, getRetentionDeleteAfterFromEnv()), compactAfter: parseRetentionOptions(config.options?.retention?.compactAfter, getRetentionCompactAfterFromEnv()), compact: cProps }; }
public getUid() { return this.config?.id ?? this.config?.name ?? this.name; }
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}; }
protected async postCache(): Promise<void> { try { await this.buildTransformRules(); return; } catch (e) { throw e; } }
protected async postInitialize(): Promise<void> { await super.postInitialize(); this.componentRepo.updateById(this.dbComponent.id, {lastReadyAt: dayjs()}); }
protected async getDatabase(): Promise<DbConcrete> { return getRoot().items.db(); }
/** * Add to this component's lifetime Discovered/Scrobbled total, in the database and on * the in-memory row, and return the new total. */ protected async incrementCountLive(by: number = 1): Promise<number> { this.dbComponent.countLive = await this.componentRepo.incrementCountLive(this.dbComponent.id, by); return this.dbComponent.countLive; }
protected async doBuildDatabase(): Promise<true | string | undefined> { await super.doBuildDatabase();
this.db = await this.getDatabase(); this.componentRepo = new DrizzleComponentRepository(this.db, {logger: this.logger}); this.dbComponent = await this.componentRepo.findOrInsert({ mode: this.componentType, type: this.type, uid: this.getUid(), name: this.config?.name ?? this.name }); this.componentId = this.dbComponent.id; // update name on startup in case user has changed it if(this.dbComponent.name !== (this.config?.name ?? this.name)) { await this.componentRepo.updateById(this.componentId, {name: this.config?.name ?? this.name}); this.dbComponent.name = (this.config?.name ?? this.name); } this.playRepo = new DrizzlePlayRepository(this.db, {logger: this.logger}); this.queueRepo = new DrizzleQueueRepository(this.db, {logger: this.logger}); this.playEventsRepo = new DrizzlePlayEventsRepository(this.db, {logger: this.logger}); this.playRepo.componentId = this.dbComponent.id; this.queueRepo.componentId = this.dbComponent.id; this.lastActiveAt = this.dbComponent.lastActiveAt ?? undefined; this.lastReadyAt = this.dbComponent.lastReadyAt ?? undefined; return true; }
public async buildTransformRules() { this.logger.debug('Building transformer rules...'); try { await this.doBuildTransformRules(); } catch (e) { this.buildOK = false; throw new TransformRulesError('Could not build playTransform rules. Check your configuration is valid.', {cause: e}); } try { //const ruleCount = countRegexes(this.transformRules); this.regexCache = cacheFunctions(200); } catch (e) { this.logger.warn(new TransformRulesError('Failed to count number of rule regexes for caching but will continue will fallback to 100', {cause: e})); } }
protected async doBuildTransformRules() { const { options: { playTransform } = {} } = this.config;
if (playTransform === undefined) { this.logger.debug(`No rules found under property 'playTransform'`); return; }
const { preCompare: preConfig, compare: { candidate: candidateConfig, existing: existingConfig, } = {}, postCompare: postConfig } = playTransform;
let preCompare: StageConfig[] | undefined, candidate: StageConfig[] | undefined, existing: StageConfig[] | undefined, postCompare: StageConfig[] | undefined;
const builtHooks: string[] = []; const emptyHooks: string[] = []; try { preCompare = await this.transformPartToStrong(preConfig); if(preCompare === undefined) { emptyHooks.push('preCompare') } else { builtHooks.push(`preCompare => ${preCompare.map(x => `${x.type}${x.name !== undefined ? `-${x.name}` : ''}`)}`); } } catch (e) { throw new Error('preCompare was not valid', {cause: e}); }
try { candidate = await this.transformPartToStrong(candidateConfig); if(candidate === undefined) { emptyHooks.push('candidate') } else { builtHooks.push(`canidate => ${candidate.map(x => `${x.type}${x.name !== undefined ? `-${x.name}` : ''}`)}`); } } catch (e) { throw new Error('candidate was not valid', {cause: e}); }
try { existing = await this.transformPartToStrong(existingConfig); if(existing === undefined) { emptyHooks.push('existing') } else { builtHooks.push(`existing => ${existing.map(x => `${x.type}${x.name !== undefined ? `-${x.name}` : ''}`)}`); } } catch (e) { throw new Error('existing was not valid', {cause: e}); }
try { postCompare = await this.transformPartToStrong(postConfig); if(postCompare === undefined) { emptyHooks.push('postCompare') } else { builtHooks.push(`postCompare => ${postCompare.map(x => `${x.type}${x.name !== undefined ? `-${x.name}` : ''}`)}`); } } catch (e) { throw new Error('postCompare was not valid', {cause: e}); }
this.logger.debug(`Hooks built! Empty: ${emptyHooks.join(', ')} | Configured: ${builtHooks.length === 0 ? 'None' : `\n${builtHooks.join('\n')}`}`);
this.transformRules = { preCompare, compare: { candidate, existing, }, postCompare, } }
protected async updateQueueStats(queueNames: string[]) { if(queueNames.includes(INGRESS_QUEUE)) { this.queuedLength = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE], {queueStatus: ['queued'], retries: 0}); this.queuedGauge.labels(this.getPrometheusLabels()).set(this.queuedLength); } if(queueNames.includes(DEAD_QUEUE)) { this.deadLetterQueued = await this.queueRepo.getQueueCount(this.dbComponent.id, [INGRESS_QUEUE], {queueStatus: ['failed','queued'], retries: 1, retryEq: 'gte'}); this.deadLetterLength = await this.playRepo.getComponentFailedNoRetryCount() + this.deadLetterQueued; // TODO this.deadLetterGauge.labels(this.getPrometheusLabels()).set(this.deadLetterLength); } }
public abstract stop(opts?: {reason?: string | Error}): Promise<void> public abstract start(opts?: {forceInit?: boolean}): Promise<boolean>
public async restart(opts: { reason?: string | Error, forceInit?: boolean } = {}): Promise<void> { try { await this.stop(opts); await this.start(opts); } catch (e) { const err = new StageChangeError('Failed to restart', { cause: e }); this.replaceErrors(err); throw err; } }
public retentionCleanup = async () => { if(this.databaseOK !== true) { this.logger.warn(`Cannot run retention cleanup because ${this.componentType} database state is not OK`); return; } this.setStatus('Running retention cleanup'); try { const repo = new DrizzlePlayRepository(this.db, {logger: this.logger}); await repo.retentionCleanup(this.componentType, this.retentionOpts); this.setStatus('Retention cleanup finished'); } catch (e) { const retentionErr = new Error('Failed to do retention cleanup', {cause: e}); this.warnings.push(retentionErr); this.logger.warn(retentionErr); this.setStatus('Retention cleanup failed'); } }
protected async transformPartToStrong(data: any) { if(data === undefined) { return undefined; } const partArr = (Array.isArray(data) ? data : [data]);
return await pMap(partArr, async (x) => this.transformManager.parseTransformerConfig(x)); }
public transformPlay = async (play: PlayObject, hookType: TransformHook, transformOpts: {log?: boolean | 'all'} & OptionalCacheUsage = {}) => {
const { log, useCachedResult = true } = transformOpts;
let logger!: Logger;
try { let hook: StageConfig[] | undefined;
switch (hookType) { case TRANSFORM_HOOK.preCompare: hook = this.transformRules.preCompare; break; case TRANSFORM_HOOK.candidate: hook = this.transformRules.compare?.candidate; break; case TRANSFORM_HOOK.existing: hook = this.transformRules.compare?.existing; break; case TRANSFORM_HOOK.postCompare: hook = this.transformRules.postCompare; break; }
if (hook === undefined) { return play; }
const asyncId = nanoid(6); logger = childLogger(this.logger, ['Play Transform', hookType, asyncId]);
const shouldLog = log ?? this.config.options?.playTransform?.log ?? isDebugMode();
const transformHash = `playTransform-${hashObject(hook)}-${playContentCacheHash(play)}`; const cachedSteps = useCachedResult ? await this.cache.cacheTransform.get<LifecycleStep[]>(transformHash) : undefined; if(cachedSteps !== undefined) { logger.trace(`Cache hit for Steps => ${transformHash}`); //return cachedTransformPlay; }
logger.debug(`Transform start for => ${buildTrackString(play)}`); let transformedPlay: PlayObject = clone(play); let cacheOk: boolean = true, generateSteps = true; const steps: LifecycleStep[] = [];
const opts = { logger, asyncId, useCachedResult }
if(cachedSteps !== undefined) { generateSteps = false; // don't re-cache cached steps cacheOk = false; try { // only patch play if steps didn't end in a failure that we are okay with returning partial from let shouldTransform = true; const lastCachedStep = cachedSteps[cachedSteps.length - 1]; if(lastCachedStep.error !== undefined && lastCachedStep.flowKnownState !== 'skip' && !lastCachedStep.returnPartial) { shouldTransform = false; }
for(const s of cachedSteps) { steps.push({...s, cached: true, createdAt: dayjs().toISOString()}); if(shouldTransform && s.patch !== undefined) { transformedPlay.data = patchObject(transformedPlay.data, s.patch); // jdiff.patch(clone(transformedPlay.data),s.patch); } } } catch (e) { logger.warn(new Error('Error occurred while trying to use cached steps. Falling back to generating fresh transform steps', {cause: e})); generateSteps = true; } } if(generateSteps) { for(const hookItem of hook) { const [step, stepPlay] = await this.generateStepFromStage(transformedPlay, hookItem, hookType, opts); steps.push(step); cacheOk = step.error === undefined || (step.error !== undefined && step.flowKnownState !== undefined); if(step.flowResult === FLOW_CONTROL_TERM.stop) { if(step.error !== undefined && step.flowKnownState !== 'skip' && !step.returnPartial) { // revert to original play but keep steps for paper trail transformedPlay = play; transformedPlay.lifecycle = steps; } else { transformedPlay = stepPlay; } break; } else { transformedPlay = stepPlay; } } }
const historyToDiff: {name: string, data?: PlayData}[] = [ {name: 'Pre Transform', data: play.data} ]; let isNew = false, diffFailure = false; if(steps.length > 0 && transformedPlay.lifecycle === undefined) { transformedPlay.lifecycle = steps; isNew = true; } const lifecycle = transformedPlay.lifecycle ?? []; transformedPlay.lifecycle = lifecycle; steps.forEach((s, index) => { if(diffFailure) { return; } const stepIdHint = `${s.source}-${s.hook}-${s.stageType}-${s.stageName}`; try { if(!isNew) { const existingStepIndex = lifecycle.findIndex(x => x.stageName === s.stageName && x.stageType === s.stageType && x.hook === s.hook && x.source === this.getIdentifier()); if(existingStepIndex !== -1) { lifecycle[existingStepIndex] = s; } else { lifecycle.push(s); } }
if(shouldLog === 'all') { if(s.patch === undefined) { historyToDiff.push({name: `${stepIdHint}`}); } else { const lastTransformedData = historyToDiff[historyToDiff.findLastIndex(x => x.data !== undefined)].data; const patched = patchObject(lastTransformedData, s.patch); historyToDiff.push({name: `${stepIdHint} ${s.cached ? ' (Cached)' : ''}`, data: patched}); } } } catch (e) { if(`patch` in s) { this.logger.debug({ patch: s.patch, playData: historyToDiff[historyToDiff.findLastIndex(x => x.data !== undefined)].data, cached: s.cached }, 'Patch and Play data context'); } this.logger.warn(new SimpleError(`Error occurred while trying to generate diffed Plays for console logging on step ${stepIdHint} but will continue`, {cause: e})); diffFailure = true; } }); if(shouldLog !== false) { if(steps.filter(x => x.patch !== undefined).length === 0) { logger.debug('Transform Diff: No Change'); } else { const diffs: string[] = [];
try { let lastTransformed: ElementOf<typeof historyToDiff>; historyToDiff.forEach((curr, index) => { if(index === 0) { lastTransformed = curr; return; } const last = historyToDiff[index - 1]; if(curr.data === undefined) { diffs.push(`${last.name} => ${curr.name} -- No Change`); } else { try { const formattedDiff = diffObjectsConsoleOutput(lastTransformed.data ?? {}, curr.data); diffs.push(`${last.name} => ${curr.name}\n${formattedDiff}`); lastTransformed = curr; } catch(e) { this.logger.debug({pre: lastTransformed.data, post: curr.data}, 'Compared Pre and Post play data'); throw e; } } }); } catch (e) { this.logger.warn(new SimpleError('Error occurred while trying to generate Play diffs for console but will continue', {cause: e})); diffFailure = true; }
if(shouldLog === true || steps.filter(x => x.patch !== undefined).length > 2) { const finalData = transformedPlay.data; try { const formattedDiff = diffObjectsConsoleOutput(play.data, finalData, true); diffs.push(`Original => Final\n${formattedDiff}`); } catch (e) { this.logger.warn(new SimpleError('Error occurred while trying to generate Play diffs for console but will continue', {cause: e})); diffFailure = true; } } if(diffFailure) { diffs.push('A failure during diff generation occurred but the transform was successful.'); } logger.debug(`Transform Diff\n${diffs.join('\n')}`); } }
if(cacheOk) { await this.cache.cacheTransform.set<LifecycleStep[]>(transformHash, steps, '2m'); }
return transformedPlay; } catch (e) { const err = new Error(`Unexpected error occurred, returning original play.`, {cause: e}); if(logger === undefined) { this.logger.warn(err); } else { logger.warn(err); } return play; } }
protected generateStepFromStage = async (playTruth: PlayObject, hookItem: StageConfig, hookType: TransformHook, opts: { logger?: Logger, asyncId?: string } & OptionalCacheUsage = {}): Promise<[LifecycleStep, PlayObject]> => { const { onSuccess = 'continue', onFailure = 'stop', onSkip = 'continue', failureReturnPartial = false, } = hookItem;
const { logger = loggerNoop, asyncId = nanoid(6), useCachedResult } = opts;
const { lifecycle = [] } = playTruth;
//const stepName = `${hookType} - ${hookItem.type} - ${hookItem.name}` const existingStepIndex = lifecycle.findIndex(x => x.hook === hookType && hookItem.name === x.stageName && x.stageType === hookItem.type && x.source === this.getIdentifier()); const step: LifecycleStep = existingStepIndex !== -1 && lifecycle[existingStepIndex] !== undefined ? lifecycle[existingStepIndex] : { // overwritten below with resolved stage name stageName: hookItem.name ?? 'Unnamed', hook: hookType, stageType: hookItem.type, source: this.getIdentifier(), createdAt: dayjs().toString() }
let newTransformedPlay!: PlayObject, stageName: string = 'Unnamed', err: Error | undefined; try { [newTransformedPlay, stageName] = await this.transformManager.handleStage(hookItem, playTruth, {asyncId, useCachedResult}); newTransformedPlay = clone(newTransformedPlay); } catch (e) { err = e as Error; if (e instanceof StageTransformError) { stageName = e.stageName; } }
step.stageName = stageName;
if (err !== undefined) { const lifecycleErrorContext = findCauseByFunc(err, (e) => 'lifecycleInputs' in e && e.lifecycleInputs !== undefined); if (lifecycleErrorContext !== undefined) { // @ts-expect-error it does exist step.inputs = clone(lifecycleErrorContext.lifecycleInputs) as LifecycleInput[]; // @ts-expect-error it does exist delete lifecycleErrorContext.lifecycleInputs; } const merged = mergeSimpleError(err); step.error = merged;
const skipError = findCauseByReference(err, SkipTransformStageError); if (skipError !== undefined) { let skipMsg = `Stage '${stageName}' was skipped`; step.flowResult = onSkip; step.flowKnownState = 'skip';
if (onSkip === 'stop') { skipMsg += ' and will stop transform due to onSkip: stop'; } step.flowReason = skipMsg;
logger.debug(merged, skipMsg); return [step, playTruth]; } else { step.flowResult = onFailure; let reason: string;
const reqError = findCauseByReference(err, StagePrerequisiteError); if (reqError !== undefined) { reason = 'Transform could not be completed due to prerequisite failure'; step.flowKnownState = 'prereq'; } else { reason = 'Transform encountered an error'; }
if (onFailure === 'continue') { reason += ' but will continue due to onFailure: continue'; logger.warn(merged, reason); } else { if (failureReturnPartial) { reason += ' | Preserving play transformations up to this point due to failureReturnPartial=true'; step.returnPartial = true; }
logger[reqError !== undefined ? 'warn' : 'error'](merged, reason); } step.flowReason = reason;
return [step, playTruth]; } } else { step.flowResult = onSuccess;
if (!objectsEqual(playTruth.data, newTransformedPlay.data)) { const o = JSON.parse(JSON.stringify(playTruth.data)); const t = JSON.parse(JSON.stringify(newTransformedPlay.data)); const patch = diffObjects(o, t); // jdiff.diff(o, t); step.patch = patch; }
if ((newTransformedPlay.meta.lifecycleInputs?.length ?? 0) > 0) { step.inputs = clone(newTransformedPlay.meta.lifecycleInputs) } else if ((playTruth.meta.lifecycleInputs?.length ?? 0) > 0) { logger.warn({ label: `${hookItem.type} - ${hookItem.name}` }, `Should only be adding inputs to transformed play!`); step.inputs = clone(playTruth.meta.lifecycleInputs) }
delete newTransformedPlay.meta.lifecycleInputs; }
return [step, newTransformedPlay ?? playTruth]; }
public abstract getRunningState(): ComponentState
public getApiData(): Omit<ComponentCommonApiJson, 'type' | 'countLive' | 'players' | 'deadLetterPlays' | 'deadLetterPlaysTotal' | 'queued' | 'historicalPlays'> & Pick<ComponentCommonApi, 'state' | 'errors' | 'warnings'> { let state: ComponentState; if(!this.initializedOnce || this.initializing) { state = COMPONENT_STATE.INITIALIZING; } else if(!this.isReady()) { state = COMPONENT_STATE.NOT_READY; } else { state = this.getRunningState() } return { id: this.dbComponent.id, uid: this.dbComponent.uid, name: this.dbComponent.name, state, mode: this.dbComponent.mode, monitoringStatus: this.getMonitoringStatus(), countNonLive: this.dbComponent.countNonLive, createdAt: this.dbComponent.createdAt?.toISOString(), lastReadyAt: this.lastReadyAt?.toISOString(), lastActiveAt: this.lastActiveAt?.toISOString(), errors: this.errors.map(x => x instanceof Error ? serializeError(x) : x), warnings: this.warnings.map(x => x instanceof Error ? serializeError(x) : x), ...this.additionalApiData() } }
public emitEvent = (eventName: string, payload: Record<string, any> = {}) => { this.emitter.emit(eventName, { type: this.type, name: this.name, componentId: this.dbComponent?.id, from: this.componentType, data: payload, }); }
public emitComponentUpdate = <T extends Partial<ReturnType<typeof this.getApiData>>>(payload: T) => { if('errors' in payload) { if(payload.errors !== undefined && payload.errors.length > 0) { payload.errors = payload.errors.map(x => x instanceof Error ? serializeError(x) : x); } else { payload.errors = []; } } if('warnings' in payload) { if(payload.warnings !== undefined && payload.warnings.length > 0) { payload.warnings = payload.warnings.map(x => x instanceof Error ? serializeError(x) : x); } else { payload.warnings = []; } } this.emitEvent('componentUpdate', payload); } protected emitPlayUpdate = (payload: MarkRequired<Partial<PlayApiCommonDetailed>, 'uid'>) => { this.emitEvent('playUpdate', payload); } protected emitPlayInsert = (payload: PlayApiCommonDetailed) => { this.emitEvent('playInsert', payload); }
async notify(payload: Omit<WebhookPayload, 'identifier'>) { this.emitEvent('notify', {...payload, identifier: this.getIdentifier()}); this.setStatus(payload.title ?? payload.message); }
public setStatus = (status: string) => { this.status = status; this.emitComponentUpdate({status}); }
public getSystemMonitoring = (): boolean => this.config.options?.autoMonitor ?? this.getSystemDefaultMonitoring();
protected getSystemDefaultMonitoring = (): boolean => this.monitoringActivityDefault;
public isMonitoring = (): boolean => this.monitoringActivity ?? this.getSystemMonitoring();
public getMonitoringStatus = (): MonitoringStatus => ({ monitoring: this.isMonitoring(), origin: this.monitoringActivity !== undefined ? MONITORING_ORIGIN_USER : MONITORING_ORIGIN_SYSTEM });
protected async updateDates(data: {lastActiveAt?: Dayjs, lastReadyAt?: Dayjs, force?: boolean}) { const { lastActiveAt, lastReadyAt, force = false } = data; let updated = false; if(lastActiveAt !== undefined && lastActiveAt !== this.lastActiveAt) { this.lastActiveAt = lastActiveAt; updated = true; } if(lastReadyAt !== undefined && lastReadyAt !== this.lastReadyAt) { this.lastReadyAt = lastReadyAt; updated = true; } if(updated && ( force || this.lastUpdatedComponentDatesAt === undefined || Math.abs(this.lastUpdatedComponentDatesAt.diff(dayjs(), 's')) >= 60 ) ) { await this.componentRepo.updateById(this.dbComponent.id, { lastActiveAt: this.lastActiveAt, lastReadyAt: this.lastReadyAt }); this.lastUpdatedComponentDatesAt = dayjs(); } }
protected getDefaultDeadLetterRetries() { return this.config.options?.deadLetterRetries ?? DEAD_LETTER_RETRIES_DEFAULT; }
queuePlay = async (data: (PlayObject | PlayObject[]) | (PlaySelectWithQueueStates | PlaySelectWithQueueStates[]), context?: QueueContext) => { const createdQueuedPlays: PlaySelect[] = [];
const dataArray = Array.isArray(data) ? data : [data]; if (dataArray.every(x => entityIsPlayEntity(x))) { /** * If the incoming objects are already all play entities (from db) * then it is being requeued by the user or re-run by dead letter functionality * and it has already passed checkExisting so just queue it up */ for (const playSelect of dataArray) { let queue = playSelect.queueStates.find(x => x.queueName === INGRESS_QUEUE); 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, error: undefined }); } const events = await this.playEventsRepo.createMany([ { playId: playSelect.id, ...stateChangeToPlayEvent({ state: 'queued' }) }, { playId: playSelect.id, ...queueStateToPlayEvent({ ...queue, queueStatus: 'queued', error: undefined, context: context ?? queue.context }) } ]) as PlayEventSelect[]; playSelect.state = 'queued'; await this.playRepo.updateById(playSelect.id, {state: 'queued'}); if (`events` in playSelect) { (playSelect as PlayWith<'events'>).events = (playSelect as PlayWith<'events'>).events.concat(events); } else { (playSelect as unknown as PlayWith<'events'>).events = events; } this.emitPlayUpdate({ ...playSelect } as unknown as PlayApiCommonDetailed); this.emitEvent('playQueued', {queuedPlay: playSelect}); createdQueuedPlays.push(playSelect); this.queuedLength += 1; this.queuedGauge.labels(this.getPrometheusLabels()).inc(); if(queue.retries > 0) { // was previously in 'failed' with retries // dequeue from failed-queue this.deadLetterQueued -= 1; this.emitEvent('deadLetterDequeued', playSelect); } } } else if (dataArray.every(x => isPlayObject(x))) { /** * If all incoming objects are only PlayObjects they are new to the component: * for source => found from polling or through ingress request * for client => passed by Source discovery mechanism or play migration (future functionality) * * so we need to do first-pass of dupe checking to identify obvious copies and reduce noise in the queue */
// instead of dropping plays that were queued when monitoring was disabled // we add this state signal to the play meta data and queue it // this way we persist the play and dupe it in queue processing so the user has an audit trail const monitoring = this.getMonitoringStatus(); const playDatas = dataArray.map(x => ({ ...x, meta: { ...x.meta, wasMonitored: monitoring.monitoring, seenAt: dayjs() } }));
await pMap(playDatas, async (queueablePlay) => { try { const existing = await this.findPreQueueExistingPlay(queueablePlay, context); if(existing !== undefined) { return; } } catch (e) { // if something went wrong we don't want to lose the input so fallback to persisting regardless of outcome this.logger.warn(new SimpleError('Failed to check queued scrobble for existing before adding, will continue with adding anyway', { cause: e })); }
// not in queue, doesn't exist elsewhere, or existing queued check failed for some reason and we don't want to lose Play const { data, meta, original } = queueablePlay const createPlayData = playToRepositoryCreatePlayOpts({ play: { data, meta, // only exists on plays from Source original }, componentId: this.dbComponent.id, state: 'queued', // only exists on plays from Source parentId: queueablePlay.id });
const playRow = await this.playRepo.createPlays([createPlayData]); const queueState = await this.queueRepo.create({ componentId: this.dbComponent.id, playId: playRow[0].id, queueName: INGRESS_QUEUE, context }) as QueueStateSelect; const createdEvents = await this.playEventsRepo.createMany([ { playId: playRow[0].id, ...stateChangeToPlayEvent({ state: 'queued' }), createdAt: (playRow[0].seenAt ?? dayjs()).add(1, 'ms') }, { playId: playRow[0].id, ...queueStateToPlayEvent(queueState), createdAt: queueState.createdAt } ]); createdQueuedPlays.push(playRow[0]); this.logger.debug(`Added ${buildTrackString(queueablePlay)} to the queue`); if(this.componentType === COMPONENT_TYPE_CLIENT) { this.setStatus(`Added Play from parent ${queueablePlay.uid} to queue`); } this.emitEvent('playQueued', {queuedPlay: queueablePlay}); this.emitPlayInsert({ ...playRow[0], queueStates: [queueState], events: createdEvents } as unknown as PlayApiCommonDetailed); this.queuedLength += 1; this.queuedGauge.labels(this.getPrometheusLabels()).inc(); // nothing to do with failed/deadletter since these are brand-new plays }); } else { throw new Error('Data passed to queuePlay must be either all be PlayObject or all PlaySelect objects'); } return createdQueuedPlays; }
abstract findPreQueueExistingPlay(queueablePlay: PlayObject, context?: QueueContext & {isRetry?: boolean}): Promise<PlayWith<'parent' | 'queueStates'> | undefined>
protected handlePlayProcessing = async (playEntity: PlayWith<'queueStates'|'events'>, signal?: AbortSignal) => { let res!: PlayProcessingResult, err: Error | undefined; try { res = await this.processPlay(playEntity, signal); } catch (e: unknown | Error | PlayProcessingError) { if(isAbortError(e)) { const abortSignal = signal ?? this.ingressQueueAbortController?.signal; err = abortSignal !== undefined ? generateLoggableAbortReason('Interrupted by abort signal', abortSignal) : e as Error; 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 { // every play handled by this function was previously added to queuedLength and emitted a 'playQueued' event // it also already emitted 'deadLetterDequeued' when it was queued this.queuedGauge.labels(this.getPrometheusLabels()).dec(); this.queuedLength -= 1; this.emitEvent('playDequeued', { queuedScrobble: playEntity });
let queueStates: QueueStateSelect[]; const initialRetries = res.queue.retries ?? 0; const last = res.playEntity.events.findLast(x => x.eventName === 'playStateChange' && (x as PlayEventPlayStateChange).data.state !== 'queued'); const lastWasNonFailureState = last === undefined || (last as PlayEventPlayStateChange).data.state !== 'failed'; if(err !== undefined) { res.queue.retries = (initialRetries + 1); res.queue.updatedAt = dayjs(); if(lastWasNonFailureState) { // play moved from non-failure state to failure // so we should increment total failed this.emitEvent('deadLetter', res.playEntity); this.deadLetterGauge.labels(this.getPrometheusLabels()).inc(); this.deadLetterLength += 1; } let shouldDeleteQueue = false; if(res.queue.retries > this.getDefaultDeadLetterRetries()) { shouldDeleteQueue = true; } else if((res.queue.context ?? {}).isRetry === true) { // isRetry is initiated by the user // so we only want this play to be processed once shouldDeleteQueue = true; } if(shouldDeleteQueue) { queueStates = res.playEntity.queueStates.filter(x => x.queueName !== res.queue.queueName) await this.queueRepo.deleteByIds([res.queue.id]); } else { await this.queueRepo.updateById(res.queue.id, { ...res.queue, }); // play may have already have been in failure state // but failed this queue run so is back to fail queued this.deadLetterQueued += 1; this.emitEvent('deadQueued', 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 || !lastWasNonFailureState) { // play had retries or last non-queue state was failure, so it was counted in failure count // now we remove it this.deadLetterGauge.labels(this.getPrometheusLabels()).dec(); this.deadLetterLength -= 1; this.emitEvent('deadLetterRemoved', { dead: { id: res.playEntity.uid } }); } queueStates = res.playEntity.queueStates.filter(x => x.queueName !== res.queue.queueName) } 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); } }
markPlayDiscarded = async (dead: PlayWith<'events' | 'queueStates'>) => {
const queueState = dead.queueStates.find(x => x.queueName === INGRESS_QUEUE); const last = dead.events.findLast(x => x.eventName === 'playStateChange' && (x as PlayEventPlayStateChange).data.state !== 'queued'); const lastWasNonFailureState = last === undefined || (last as PlayEventPlayStateChange).data.state !== 'failed';
this.setStatus(`Marking Play ${dead.uid} as discarded`);
const lastState = dead.state; const events: PlayEventNew[] = [];
if(dead.state !== 'discarded') { dead.state = 'discarded'; this.playRepo.updateById(dead.id, {state: 'discarded'}); events.push( { playId: dead.id, ...stateChangeToPlayEvent({ state: 'discarded', reason: 'Play marked as discarded by user' }) } ) } if(queueState !== undefined) { events.push( { playId: dead.id, ...queueCompletionStateToPlayEvent({...queueState, error: undefined, queueStatus: QUEUE_STATUS_COMPLETED, context: {reason: 'Play marked as discarded by user'}}) } ); } if(queueState !== undefined) { await this.queueRepo.deleteByIds([queueState.id]); } let createdEvents: PlayEventSelect[] = []; if(events.length > 0) { 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<string>[] });
if(!lastWasNonFailureState) { // play was in failure state this.deadLetterLength -= 1; this.deadLetterGauge.labels(this.getPrometheusLabels()).dec(); this.emitEvent('deadLetterRemoved', dead); if(queueState !== undefined && queueState.retries > 0) { // and was queued this.deadLetterQueued -= 1; // give sse time to catch up await sleep(20); this.emitEvent('deadLetterDequeued', dead); } } else if(lastState === 'queued') { this.queuedLength -= 1; this.queuedGauge.labels(this.getPrometheusLabels()).dec(); this.emitEvent('playDequeued', dead); } }
markPlaysDiscarded = async () => { const ids = await this.playRepo.findPlayIdentifiers({ queues: [ { queueName: INGRESS_QUEUE, queueStatus: 'failed' } ] }, 'id'); this.deadLogger.info(`Marking ${ids.length} as discarded...`); await pMap(ids, async (id) => { const entity = await this.playRepo.findByIdWith<'queueStates' | 'events'>(id, ['queues','events']); if(entity !== undefined) { await this.markPlayDiscarded(entity); } }, {concurrency: 10}); this.deadLogger.info('Finished processing failed scrobbles.'); await this.updateQueueStats([INGRESS_QUEUE]); }
public cancelQueuedPlay = async (playEntity: PlayWith<'queueStates' | 'events'>) => { const queueState = playEntity.queueStates.find(x => x.queueName === INGRESS_QUEUE); if(queueState === undefined) { throw new SimpleError('Play does not have an associated queued'); } const last = playEntity.events.findLast(x => x.eventName === 'playStateChange' && (x as PlayEventPlayStateChange).data.state !== 'queued'); const lastWasNonFailureState = last === undefined || (last as PlayEventPlayStateChange).data.state !== 'failed';
const playState = playEntity.state; if(last === undefined) { // if play has only ever been queued then it has failed the initial lifecycle playEntity.state = 'failed'; if(lastWasNonFailureState) { this.deadLetterLength += 1; this.deadLetterGauge.labels(this.getPrometheusLabels()).inc(); this.emitEvent('deadLetter', playEntity); } } else { //if it had a previous, completed state then use that instead playEntity.state = (last as PlayEventPlayStateChange).data.state; } const eventsToCreate: PlayEventNew[] = []; if(playEntity.state !== playState) { eventsToCreate.push({playId: playEntity.id, ...stateChangeToPlayEvent({state: playEntity.state})}); } eventsToCreate.push({playId: playEntity.id, ...queueStateToPlayEvent({...queueState, queueStatus: QUEUE_STATUS_COMPLETED, context: {reason: 'Cancelled by user'}})}); const createdEvents = await this.playEventsRepo.createMany(eventsToCreate) as PlayEventSelect[]; await this.queueRepo.deleteByIds([queueState.id]); await this.playRepo.updateById(playEntity.id, {state: playEntity.state}); this.emitPlayUpdate({ ...playEntity, queueStates: playEntity.queueStates.filter(x => x.queueName !== INGRESS_QUEUE) as unknown as QueueStateApi[], events: ((playEntity as unknown as PlayWith<'events'>).events ?? []).concat(createdEvents), } as unknown as PlayApiCommonDetailed); if(queueState.queueStatus !== 'queued') { // in failure queue state with retries this.emitEvent('deadLetterDequeued', { queuedScrobble: playEntity }); this.deadLetterQueued -= 1; } else { // in actual queue state (may still be failure) this.emitEvent('playDequeued', { queuedScrobble: playEntity }); this.queuedLength -= 1; this.queuedGauge.labels(this.getPrometheusLabels()).dec(); } }
abstract processPlay(playEntity: PlaySelectWithQueueStates, signal?: AbortSignal): Promise<PlayProcessingResult>
public async updatePlay(uid: string, data: Partial<PlayData>): Promise<PlayApiCommonDetailed> { const playRow = await this.playRepo.findByUidWith<'events'>(uid, ['events']); if(playRow === undefined) { throw new SimpleError(`No play exists with uid ${uid}`); } const original = JSON.parse(JSON.stringify(playRow.play)); const merged: PlayData = { ...playRow.play.data, ...data, repeat: playRow.play.data.repeat }; const samePlayDate = data.playDate?.isSame(playRow.play.data.playDate);
if(playRow.play.data.listenRanges !== undefined && samePlayDate) { merged.listenRanges = playRow.play.data.listenRanges merged.playDateCompleted = playRow.play.data.playDateCompleted; } else if(data.listenedFor !== undefined) { merged.playDateCompleted = data.playDate?.add(data.listenedFor, 's') }
playRow.play.data = merged; const modified = JSON.parse(JSON.stringify(playRow.play)); const patch = diffObjects(original, modified); const event = playUpdateToEvent({input: {data}, patch}); await this.playRepo.updateById(playRow.id, {play: playRow.play}); const createdEvent = await this.playEventsRepo.create({...event, playId: playRow.id}); // @ts-expect-error this should be fine playRow.events.push(createdEvent); this.emitPlayUpdate({ uid, events: playRow.events as unknown as PlayEvent<string>[], play: playRow.play as unknown as JsonPlayObject }) return playRow as unknown as PlayApiCommonDetailed; }}