refactor: Move stagger config responsibility to transformer · foxxmd.dev/multi-scrobbler@358355c

+1 -4

src/backend/ioc.ts

··· 27 27 cache?: CacheConfigOptions | MSCache | (() => MSCache) 28 28 mbMap?: MusicBrainzSingletonMap | (() => MusicBrainzSingletonMap) 29 29 transformers?: TransformerCommonConfig[] 30 - staggerOptions?: Partial<StaggerOptions> 31 30 } 32 31 33 32 const discovered = new prom.Counter({ ··· 61 60 logger, 62 61 cache, 63 62 mbMap, 64 - transformers = [], 65 - staggerOptions, 63 + transformers = [] 66 64 } = options || {}; 67 65 const configDir = process.env.CONFIG_DIR || path.resolve(projectDir, `./config`); 68 66 let disableWeb = dw; ··· 150 148 cache: () => maybeSingletonCache !== undefined ? () => maybeSingletonCache : cacheFunc, 151 149 mbMap: () => maybeSingletonMb !== undefined ? () => maybeSingletonMb : mbFunc, 152 150 coverArtApi, 153 - staggerOptions: staggerOptions ?? {}, 154 151 }).add((items) => { 155 152 const localUrl = generateBaseURL(baseUrl, items.port) 156 153 return {

+42 -5

src/backend/scrobblers/AbstractScrobbleClient.ts

··· 125 125 protected problemGauge: Gauge; 126 126 127 127 protected staggerOpts: Partial<StaggerOptions>; 128 + protected staggerMappers = { 129 + preCompare: staggerMapper<PlayObject, PlayObject>({concurrency: 2}), 130 + existing: staggerMapper<ScrobbledPlayObject, ScrobbledPlayObject>({concurrency: 2}) 131 + } 128 132 129 133 constructor(type: any, name: any, config: CommonClientConfig, notifier: Notifiers, emitter: EventEmitter, logger: Logger) { 130 134 super(config); ··· 190 194 existingSubmitted: this.findExistingSubmittedPlayObj 191 195 } 192 196 this.existingScrobble = (playObjPre: PlayObject, existingScrobbles: PlayObject[], log?: boolean) => existingScrobble(playObjPre, existingScrobbles, existingScrobbleOpts, log); 193 - this.staggerOpts = getRoot().items.staggerOptions; 197 + } 198 + 199 + protected async postCache(): Promise<void> { 200 + await super.postCache(); 201 + this.generateStaggerMappers(); 202 + } 203 + 204 + protected generateStaggerMappers() { 205 + const { 206 + preCompare = [], 207 + compare: { 208 + existing = [] 209 + } = {} 210 + } = this.transformRules; 211 + 212 + if(preCompare.length > 0) { 213 + let pcInits: number[] = [0], 214 + pcMaxStagger: number[] = []; 215 + for(const hook of this.transformRules.preCompare) { 216 + const t = this.transformManager.getTransformerByStage({type: hook.type, name: hook.name}); 217 + pcInits.push(t.staggerOpts?.initialInterval ?? 0); 218 + pcMaxStagger.push(t.staggerOpts?.maxRandomStagger ?? 0) 219 + } 220 + this.staggerMappers.preCompare = staggerMapper<PlayObject, PlayObject>({initialInterval: Math.max(...pcInits), maxRandomStagger: Math.max(...pcMaxStagger), concurrency: 2}); 221 + } 222 + 223 + if(existing.length > 0) { 224 + let eInits: number[] = [0], 225 + eMaxStagger: number[] = []; 226 + for(const hook of this.transformRules.postCompare) { 227 + const t = this.transformManager.getTransformerByStage({type: hook.type, name: hook.name}); 228 + eInits.push(t.staggerOpts?.initialInterval ?? 0); 229 + eMaxStagger.push(t.staggerOpts?.maxRandomStagger ?? 0) 230 + } 231 + this.staggerMappers.existing = staggerMapper<ScrobbledPlayObject, ScrobbledPlayObject>({initialInterval: Math.max(...eInits), maxRandomStagger: Math.max(...eMaxStagger), concurrency: 2}); 232 + } 194 233 } 195 234 196 235 protected getIdentifier() { ··· 449 488 450 489 const playObj = await this.transformPlay(playObjPre, TRANSFORM_HOOK.candidate); 451 490 452 - const sm = staggerMapper<ScrobbledPlayObject, ScrobbledPlayObject>({...this.staggerOpts, concurrency: 2}); 453 - const dtInvariantMatches = (await pMap(this.scrobbledPlayObjs.data, sm(async x => ({...x, play: await this.transformPlay(x.play, TRANSFORM_HOOK.existing)})), {concurrency: 2})) 491 + const dtInvariantMatches = (await pMap(this.scrobbledPlayObjs.data, this.staggerMappers.existing(async x => ({...x, play: await this.transformPlay(x.play, TRANSFORM_HOOK.existing)})), {concurrency: 2})) 454 492 .filter(x => playObjDataMatch(playObj, x.play)); 455 493 456 494 if (dtInvariantMatches.length === 0) { ··· 851 889 852 890 queueScrobble = async (data: PlayObject | PlayObject[], source: string) => { 853 891 const plays = (Array.isArray(data) ? data : [data]).map(x => ({...x, meta: {...x.meta, seenAt: dayjs()}})); 854 - const sm = staggerMapper<PlayObject, PlayObject>({...this.staggerOpts, concurrency: 2}); 855 - for await(const play of pMapIterable(plays, sm(async x => await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 2})) { 892 + for await(const play of pMapIterable(plays, this.staggerMappers.preCompare(async x => await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 2})) { 856 893 try { 857 894 const existingQueued = await this.existingScrobble(play, this.queuedScrobbles.map(x => x.play), false); 858 895 // want to be very confident of this

+40 -6

src/backend/sources/AbstractSource.ts

··· 94 94 95 95 protected discoveredCounter: Counter; 96 96 97 - protected staggerOpts: Partial<StaggerOptions>; 97 + protected staggerMappers = { 98 + preCompare: staggerMapper<PlayObject, PlayObject>({concurrency: 2}), 99 + postCompare: staggerMapper<PlayObject, PlayObject>({concurrency: 2}) 100 + } 98 101 99 102 constructor(type: SourceType, name: string, config: SourceConfig, internal: InternalConfig, emitter: EventEmitter) { 100 103 super(config); ··· 112 115 this.emitter = emitter; 113 116 114 117 this.discoveredCounter = getRoot().items.sourceMetics.discovered; 115 - this.staggerOpts = getRoot().items.staggerOptions; 118 + } 119 + 120 + protected async postCache(): Promise<void> { 121 + await super.postCache(); 122 + this.generateStaggerMappers(); 123 + } 124 + 125 + protected generateStaggerMappers() { 126 + const { 127 + preCompare = [], 128 + postCompare = [], 129 + } = this.transformRules; 130 + 131 + if (preCompare.length > 0) { 132 + let pcInits: number[] = [0], 133 + pcMaxStagger: number[] = [0]; 134 + for (const hook of this.transformRules.preCompare) { 135 + const t = this.transformManager.getTransformerByStage({ type: hook.type, name: hook.name }); 136 + pcInits.push(t.staggerOpts?.initialInterval ?? 0); 137 + pcMaxStagger.push(t.staggerOpts?.maxRandomStagger ?? 0) 138 + } 139 + this.staggerMappers.preCompare = staggerMapper<PlayObject, PlayObject>({ initialInterval: Math.max(...pcInits), maxRandomStagger: Math.max(...pcMaxStagger), concurrency: 2 }); 140 + } 141 + 142 + if (postCompare.length > 0) { 143 + let postInits: number[] = [0], 144 + postMaxStagger: number[] = [0]; 145 + for (const hook of this.transformRules.postCompare) { 146 + const t = this.transformManager.getTransformerByStage({ type: hook.type, name: hook.name }); 147 + postInits.push(t.staggerOpts?.initialInterval ?? 0); 148 + postMaxStagger.push(t.staggerOpts?.maxRandomStagger ?? 0) 149 + } 150 + this.staggerMappers.postCompare = staggerMapper<PlayObject, PlayObject>({ initialInterval: Math.max(...postInits), maxRandomStagger: Math.max(...postMaxStagger), concurrency: 2 }); 151 + } 116 152 } 117 153 118 154 protected getIdentifier() { ··· 222 258 discover = async (plays: PlayObject[], options: { checkAll?: boolean, [key: string]: any } = {}): Promise<PlayObject[]> => { 223 259 const newDiscoveredPlays: PlayObject[] = []; 224 260 225 - const sm = staggerMapper<PlayObject, PlayObject>({...this.staggerOpts, concurrency: 2}); 226 - for await(const play of pMapIterable(plays, sm(async x => await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 2})) { 261 + for await(const play of pMapIterable(plays, this.staggerMappers.preCompare(async x => await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 2})) { 227 262 if(!(await this.alreadyDiscovered(play, options))) { 228 263 this.addPlayToDiscovered(play); 229 264 newDiscoveredPlays.push(play); ··· 254 289 return; 255 290 } 256 291 newDiscoveredPlays.sort(sortByOldestPlayDate); 257 - const sm = staggerMapper<PlayObject, PlayObject>({...this.staggerOpts, concurrency: 2}); 258 292 this.emitter.emit('discoveredToScrobble', { 259 - data: await pMap(newDiscoveredPlays, sm(async (x) => await this.transformPlay(x, TRANSFORM_HOOK.postCompare)), {concurrency: 2}), 293 + data: await pMap(newDiscoveredPlays, this.staggerMappers.postCompare(async (x) => await this.transformPlay(x, TRANSFORM_HOOK.postCompare)), {concurrency: 2}), 260 294 options: { 261 295 ...options, 262 296 checkTime: newDiscoveredPlays[newDiscoveredPlays.length-1].data.playDate.add(2, 'second'),

+1 -1

src/backend/tests/setup.ts

··· 2 2 import { getRoot } from "../ioc.js"; 3 3 import { transientCache } from './utils/CacheTestUtils.js'; 4 4 5 - const root = getRoot({cache: transientCache, logger: loggerTest, staggerOptions: {initialInterval: 1, maxRandomStagger: 1}}); 5 + const root = getRoot({cache: transientCache, logger: loggerTest}); 6 6 root.items.cache().init();

+2 -2

src/backend/utils/AsyncUtils.ts

··· 54 54 } 55 55 export function staggerMapper<Element, NewElement>(options: StaggerOptions) { 56 56 const { 57 - initialInterval = 300, 58 - maxRandomStagger = 300, 57 + initialInterval = 0, 58 + maxRandomStagger = 0, 59 59 concurrency 60 60 } = options; 61 61 let initialStagger = 0;

+3

src/backend/common/transforms/AbstractTransformer.ts

··· 9 9 import { playContentInvariantTransform } from "../../utils/PlayComparisonUtils.js"; 10 10 import { isSimpleError, SkipTransformStageError, StagePrerequisiteError } from "../errors/MSErrors.js"; 11 11 import { capitalize } from "../../../core/StringUtils.js"; 12 + import { StaggerOptions } from "../../utils/AsyncUtils.js"; 12 13 13 14 export interface TransformerOptions { 14 15 logger: Logger ··· 33 34 cache: Cacheable; 34 35 35 36 name: string; 37 + 38 + public staggerOpts: Partial<StaggerOptions> = { initialInterval: 0, maxRandomStagger: 0}; 36 39 37 40 public constructor(config: TransformerCommon, options: TransformerOptions) { 38 41 super(config);

+4

src/backend/common/transforms/MusicbrainzTransformer.ts

··· 355 355 public constructor(config: MusicbrainzTransformerConfig, options: TransformerOptions & {clientCache?: Cacheable}) { 356 356 super(config, options); 357 357 this.clientCache = options.clientCache; 358 + this.staggerOpts = { 359 + initialInterval: 300, 360 + maxRandomStagger: 300 361 + } 358 362 } 359 363 360 364 protected async doBuildInitData(): Promise<true | string | undefined> {

+6 -22

src/backend/common/transforms/TransformerManager.ts

··· 110 110 return this.transformers.has(type); 111 111 } 112 112 113 - protected getTransformerByStage(data: StageConfig): AbstractTransformer { 113 + public getTransformerType(type: string): AbstractTransformer[] | undefined { 114 + return this.transformers.get(type); 115 + } 116 + 117 + public getTransformerByStage(data: StageConfig): AbstractTransformer { 114 118 const list = this.transformers.get(data.type); 115 119 if (list === undefined || list.length === 0) { 116 120 throw new Error(`No transformer of type '${data.type}' is registered.`); ··· 143 147 } 144 148 145 149 public async handleStage(data: StageConfig, play: PlayObject, asyncId: string = nanoid(6)): Promise<[PlayObject, string]> { 146 - const list = this.transformers.get(data.type); 147 - if (list === undefined || list.length === 0) { 148 - throw new Error(`No transformer of type '${data.type}' is registered.`); 149 - } 150 - 151 - let t: AbstractTransformer; 152 - if (list.length > 1) { 153 - if(data.name === undefined) { 154 - this.logger.warn(`More than one '${data.type}' transformer but name was not specified, using first registered`); 155 - t = list[0]; 156 - } else { 157 - const named = list.find(x => x.name === data.name); 158 - if(named === undefined) { 159 - throw new Error(`No ${data.type} transformer with name '${data.name}'`) 160 - } 161 - t = named; 162 - } 163 - } else { 164 - t = list[0]; 165 - } 166 - 150 + const t: AbstractTransformer = this.getTransformerByStage(data); 167 151 try { 168 152 const transformedPlay = await this.asyncStore.run(asyncId, async () => { 169 153 return await t.handle(data, play);

Configure Feed

Configure Feed