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.
28 kB · 666 lines
TypeScript
at queueNext
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667import { childLogger, type Logger } from "@foxxmd/logging";import { cacheFunctions,} from "@foxxmd/regex-buddy-core";import type EventEmitter from "events";import {MONITORING_ORIGIN_SYSTEM, MONITORING_ORIGIN_USER, type ComponentType, type LifecycleInput, type LifecycleStep, type PlayData, type PlayObject} from "../../core/Atomic.ts";import { buildTrackString } from "../../core/StringUtils.ts";import type {CommonClientConfig} from "./infrastructure/config/client/index.ts";import type {CommonSourceConfig} from "./infrastructure/config/source/index.ts";import { 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 } from "../utils.ts";import { findCauseByFunc, findCauseByReference } from "../utils/ErrorUtils.ts";import { hashObject, parseArrayFromMaybeString } from "../utils/StringUtils.ts";import { playContentInvariantTransform } 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} from "./database/drizzle/drizzleTypes.ts";import { DrizzlePlayRepository } from "./database/drizzle/repositories/PlayRepository.ts";import type {ClientType, MonitoringStatus, OptionalCacheUsage} 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 } from "../../core/Api.ts";import type {WebhookPayload} from "./infrastructure/config/health/webhooks.ts";import type { MarkRequired } from "ts-essentials";import { serializeError } from "serialize-error";
export type AbstractComponentConfig = (CommonClientConfig | CommonSourceConfig) & { transformManager?: TransformerManager };
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; componentId!: number; protected retentionOpts: RetentionOptions; status: string = 'Waiting to initialize...'; emitter: EventEmitter;
monitoringActivity?: boolean | undefined; monitoringActivityDefault: boolean = true;
protected componentType: ComponentType; type: ClientType | SourceType; name: string;
lastActiveAt?: Dayjs lastReadyAt?: Dayjs; protected lastUpdatedComponentDatesAt?: Dayjs
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 postCache(): Promise<void> { try { 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 doBuildDatabase(): Promise<true | string | undefined> { await super.doBuildDatabase();
this.db = await getRoot().items.db(); 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; this.lastActiveAt = this.dbComponent.lastActiveAt; this.lastReadyAt = this.dbComponent.lastReadyAt; return true; }
public buildTransformRules() { this.logger.debug('Building transformer rules...'); try { 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 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[], candidate: StageConfig[], existing: StageConfig[], postCompare: StageConfig[];
const builtHooks: string[] = []; const emptyHooks: string[] = []; try { preCompare = 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 = 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 = 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 = 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, } }
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 transformPartToStrong(data: any) { if(data === undefined) { return undefined; } const partArr = (Array.isArray(data) ? data : [data]);
return partArr.map(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[];
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)}-${hashObject(playContentInvariantTransform(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; if(steps.length > 0 && transformedPlay.lifecycle === undefined) { transformedPlay.lifecycle = steps; isNew = true; } const { lifecycle = [] } = transformedPlay; steps.forEach((s, index) => { 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) { transformedPlay.lifecycle[existingStepIndex] = s; } else { transformedPlay.lifecycle.push(s); } }
if(shouldLog === 'all') { if(s.patch === undefined) { historyToDiff.push({name: `${s.source}-${s.hook}-${s.stageType}-${s.stageName}`}); } else { const patched = patchObject(historyToDiff[historyToDiff.length - 1].data, s.patch); historyToDiff.push({name: `${s.source}-${s.hook}-${s.stageType}-${s.stageName} ${s.cached ? ' (Cached)' : ''}`, data: patched}); } } }); if(shouldLog !== false) { if(steps.filter(x => x.patch !== undefined).length === 0) { logger.debug('Transform Diff: No Change'); } else { const diffs: string[] = [];
historyToDiff.forEach((curr, index) => { if(index === 0) { return; } const last = historyToDiff[index - 1]; if(curr.data === undefined) { diffs.push(`${last.name} => ${curr.name} -- No Change`); } else { const formattedDiff = diffObjectsConsoleOutput(last.data, curr.data); diffs.push(`${last.name} => ${curr.name}\n${formattedDiff}`); } });
if(shouldLog === true || steps.filter(x => x.patch !== undefined).length > 2) { const finalData = transformedPlay.data; const formattedDiff = diffObjectsConsoleOutput(play.data, finalData, true); diffs.push(`Original => Final\n${formattedDiff}`); }
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] : { stageName: hookItem.name, hook: hookType, stageType: hookItem.type, source: this.getIdentifier(), createdAt: dayjs().toString() }
let newTransformedPlay: PlayObject, stageName: string = 'Unnamed', err: Error; try { [newTransformedPlay, stageName] = await this.transformManager.handleStage(hookItem, playTruth, {asyncId, useCachedResult}); newTransformedPlay = clone(newTransformedPlay); } catch (e) { err = e; 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) { step.inputs = clone(newTransformedPlay.meta.lifecycleInputs) } else if (playTruth.meta.lifecycleInputs?.length > 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'> & 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.length > 0) { payload.errors = payload.errors.map(x => x instanceof Error ? serializeError(x) : x); } else { payload.errors = []; } } if('warnings' in payload) { if(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); }
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(); } }}