[READ-ONLY] Mirror of https://github.com/FoxxMD/multi-scrobbler. Scrobble plays from multiple sources to multiple clients
docs.multi-scrobbler.app
deezer
docker
jellyfin
koito
lastfm
listenbrainz
maloja
mopidy
mpris
music
music-assistant
plex
scrobble
self-hosted
spotify
subsonic
tautulli
youtube-music
somewhat working scrobble caching
+114
-36
+42
-1
src/backend/common/AbstractComponent.ts
+42
-1
src/backend/common/AbstractComponent.ts
···
24
24
import { CommonSourceConfig } from "./infrastructure/config/source/index.js";
25
25
import play = Simulate.play;
26
26
import { WebhookPayload } from "./infrastructure/config/health/webhooks.js";
27
-
import { AuthCheckError, BuildDataError, ConnectionCheckError, PostInitError, TransformRulesError } from "./errors/MSErrors.js";
27
+
import { AuthCheckError, BuildDataError, ConnectionCheckError, ParseCacheError, PostInitError, TransformRulesError } from "./errors/MSErrors.js";
28
28
import { messageWithCauses, messageWithCausesTruncatedDefault } from "../utils/ErrorUtils.js";
29
29
30
30
export default abstract class AbstractComponent {
···
35
35
36
36
buildOK?: boolean | null;
37
37
connectionOK?: boolean | null;
38
+
cacheOK?: boolean | null;
38
39
39
40
initializing: boolean = false;
40
41
···
65
66
await this.buildComponentLogger();
66
67
}
67
68
await this.buildInitData(force);
69
+
await this.parseCache(force);
68
70
this.buildTransformRules();
69
71
await this.checkConnection(force);
70
72
await this.testAuth(force);
···
104
106
throw e;
105
107
}
106
108
}
109
+
110
+
public async parseCache(force: boolean = false) {
111
+
if(this.cacheOK) {
112
+
if(!force) {
113
+
return;
114
+
}
115
+
this.logger.debug('Cache OK but step was forced');
116
+
}
117
+
try {
118
+
const res = await this.doParseCache();
119
+
if(res === undefined) {
120
+
this.cacheOK = null;
121
+
this.logger.debug('No cache to parse.');
122
+
return;
123
+
}
124
+
if (res === true) {
125
+
this.logger.verbose('Parsing caching succeeded');
126
+
} else if (typeof res === 'string') {
127
+
this.logger.verbose(`Parsing caching succeeded => ${res}`);
128
+
}
129
+
this.cacheOK = true;
130
+
} catch (e) {
131
+
this.cacheOK = false;
132
+
throw new ParseCacheError('Parsing cache for initialization failed', {cause: e});
133
+
}
134
+
}
135
+
136
+
/**
137
+
* Build or parse any cache required for this Component
138
+
*
139
+
* * Return undefined if not possible or not required
140
+
* * Return TRUE if build succeeded
141
+
* * Return string if build succeeded and should log result
142
+
* * Throw error on failure
143
+
* */
144
+
protected async doParseCache(): Promise<true | string | undefined> {
145
+
return;
146
+
}
147
+
107
148
108
149
public async buildInitData(force: boolean = false) {
109
150
if(this.buildOK) {
+7
-30
src/backend/common/Cache.ts
+7
-30
src/backend/common/Cache.ts
···
5
5
import { projectDir } from './index.js';
6
6
import path from 'path';
7
7
import { fileOrDirectoryIsWriteable } from '../utils.js';
8
-
9
-
export type CacheProvider = 'memory' | 'valkey' | 'file';
10
-
11
-
interface CacheConfig<T extends CacheProvider = CacheProvider> {
12
-
provider: T
13
-
connection?: string
14
-
}
15
-
16
-
export type CacheMetadaProvider = Exclude<CacheProvider, 'file'>;
17
-
export type CacheMetadataConfig = CacheConfig<CacheMetadaProvider>
18
-
19
-
const asCacheMetadataProvider = (val: string): val is CacheScrobbleProvider => {
20
-
return ['memory', 'valkey'].includes(val);
21
-
}
22
-
23
-
export type CacheScrobbleProvider = CacheProvider;
24
-
export type CacheScrobbleConfig = CacheConfig<CacheScrobbleProvider>;
25
-
26
-
const asCacheScrobbleProvider = (val: string): val is CacheScrobbleProvider => {
27
-
return ['memory', 'valkey', 'file'].includes(val);
28
-
}
29
-
30
-
export interface CacheConfigOptions {
31
-
metadata?: CacheMetadataConfig
32
-
scrobble?: CacheScrobbleConfig
33
-
}
8
+
import { asCacheMetadataProvider, asCacheScrobbleProvider, CacheConfig, CacheConfigOptions, CacheMetadaProvider, CacheProvider } from './infrastructure/Atomic.js';
34
9
35
10
const configDir = process.env.CONFIG_DIR || path.resolve(projectDir, `./config`);
36
11
···
43
18
44
19
logger: Logger;
45
20
46
-
constructor(logger: Logger, config: CacheConfigOptions) {
21
+
constructor(logger: Logger, config: CacheConfigOptions = {}) {
47
22
this.logger = childLogger(logger, 'Cache');
48
23
49
24
const {
50
25
metadata: {
51
-
provider: mProvider = 'memory',
26
+
provider: mProvider = (process.env.CACHE_METADATA as (CacheMetadaProvider | undefined) ?? 'memory'),
27
+
connection: mConn = process.env.CACHE_METADATA_CONN,
52
28
...restMetadata
53
29
} = {},
54
30
scrobble: {
55
-
provider: sProvider = 'memory',
56
-
connection = configDir,
31
+
provider: sProvider = (process.env.CACHE_SCROBBLE as (CacheProvider | undefined) ?? 'file'),
32
+
connection = (process.env.CACHE_SCROBBLE_CONN ?? configDir),
57
33
...restScrobble
58
34
} = {},
59
35
} = config;
···
61
37
this.config = {
62
38
metadata: {
63
39
provider: mProvider,
40
+
connection: mConn,
64
41
...restMetadata,
65
42
},
66
43
scrobble: {
+5
-1
src/backend/common/errors/MSErrors.ts
+5
-1
src/backend/common/errors/MSErrors.ts
···
8
8
name = 'Init Build Data';
9
9
}
10
10
11
+
export class ParseCacheError extends StageError {
12
+
name = 'Init Parse Cache';
13
+
}
14
+
11
15
export class TransformRulesError extends StageError {
12
16
name = 'Transform Rules';
13
17
}
14
18
15
19
export class ConnectionCheckError extends StageError {
16
-
name = 'Conenction Check';
20
+
name = 'Connection Check';
17
21
}
18
22
19
23
export class AuthCheckError extends StageError {
+21
-1
src/backend/common/infrastructure/Atomic.ts
+21
-1
src/backend/common/infrastructure/Atomic.ts
···
348
348
349
349
export type WithRequiredProperty<Type, Key extends keyof Type> = Type & {
350
350
[Property in Key]-?: Type[Property];
351
-
};
351
+
};
352
+
export type CacheProvider = 'memory' | 'valkey' | 'file';
353
+
export interface CacheConfig<T extends CacheProvider = CacheProvider> {
354
+
provider: T;
355
+
connection?: string;
356
+
}
357
+
export type CacheMetadaProvider = Exclude<CacheProvider, 'file'>;
358
+
export type CacheMetadataConfig = CacheConfig<CacheMetadaProvider>;
359
+
export const asCacheMetadataProvider = (val: string): val is CacheScrobbleProvider => {
360
+
return ['memory', 'valkey'].includes(val);
361
+
};
362
+
export type CacheScrobbleProvider = CacheProvider;
363
+
export type CacheScrobbleConfig = CacheConfig<CacheScrobbleProvider>;
364
+
export const asCacheScrobbleProvider = (val: string): val is CacheScrobbleProvider => {
365
+
return ['memory', 'valkey', 'file'].includes(val);
366
+
};
367
+
export interface CacheConfigOptions {
368
+
metadata?: CacheMetadataConfig;
369
+
scrobble?: CacheScrobbleConfig;
370
+
}
371
+
+3
-1
src/backend/common/infrastructure/config/aioConfig.ts
+3
-1
src/backend/common/infrastructure/config/aioConfig.ts
···
5
5
import { WebhookConfig } from "./health/webhooks.js";
6
6
import { CommonSourceOptions, SourceRetryOptions } from "./source/index.js";
7
7
import { SourceAIOConfig } from "./source/sources.js";
8
-
import { ClientType, SourceType } from "../Atomic.js";
8
+
import { CacheConfigOptions, ClientType, SourceType } from "../Atomic.js";
9
9
10
10
11
11
export interface SourceDefaults extends CommonSourceOptions {
···
64
64
* @examples [false]
65
65
* */
66
66
debugMode?: boolean
67
+
68
+
cache?: CacheConfigOptions
67
69
}
68
70
69
71
export interface AIOClientConfig {
+2
src/backend/index.ts
+2
src/backend/index.ts
+6
-2
src/backend/ioc.ts
+6
-2
src/backend/ioc.ts
···
8
8
9
9
import { generateBaseURL } from "./utils/NetworkUtils.js";
10
10
import { PassThrough } from "stream";
11
+
import { CacheConfigOptions } from "./common/infrastructure/Atomic.js";
12
+
import { MSCache } from "./common/Cache.js";
11
13
12
14
export let version: string = 'unknown';
13
15
···
24
26
disableWeb?: boolean
25
27
loggerStream?: PassThrough
26
28
loggingConfig?: LogOptions
29
+
cache?: CacheConfigOptions
27
30
}
28
31
29
32
const createRoot = (options?: RootOptions) => {
···
32
35
baseUrl = process.env.BASE_URL,
33
36
disableWeb: dw,
34
37
loggerStream,
35
-
loggingConfig
38
+
loggingConfig,
36
39
} = options || {};
37
40
const configDir = process.env.CONFIG_DIR || path.resolve(projectDir, `./config`);
38
41
let disableWeb = dw;
···
62
65
notifierEmitter: () => new EventEmitter(),
63
66
loggerStream,
64
67
loggingConfig,
65
-
logger: options.logger
68
+
logger: options.logger,
69
+
cache: new MSCache(options.logger, options.cache)
66
70
}).add((items) => {
67
71
const localUrl = generateBaseURL(baseUrl, items.port)
68
72
return {
+24
src/backend/scrobblers/AbstractScrobbleClient.ts
+24
src/backend/scrobblers/AbstractScrobbleClient.ts
···
52
52
} from "../utils/TimeUtils.js";
53
53
import { WebhookPayload } from "../common/infrastructure/config/health/webhooks.js";
54
54
import { AsyncTask, SimpleIntervalJob, Task, ToadScheduler } from "toad-scheduler";
55
+
import { MSCache } from "../common/Cache.js";
56
+
import { getRoot } from "../ioc.js";
55
57
56
58
type PlatformMappedPlays = Map<string, {play: PlayObject, source: SourceIdentifier}>;
57
59
type NowPlayingQueue = Map<string, PlatformMappedPlays>;
···
97
99
nowPlayingTaskInterval: number = 5000;
98
100
npLogger: Logger;
99
101
102
+
cache: MSCache;
103
+
100
104
declare config: CommonClientConfig;
101
105
102
106
notifier: Notifiers;
···
110
114
this.npLogger = childLogger(this.logger, 'Now Playing');
111
115
this.notifier = notifier;
112
116
this.emitter = emitter;
117
+
this.cache = getRoot().get('cache');
113
118
114
119
this.scrobbledPlayObjs = new FixedSizeList<ScrobbledPlayObject>(this.MAX_STORED_SCROBBLES);
115
120
···
169
174
protected getIdentifier() {
170
175
return `${capitalize(this.type)} - ${this.name}`
171
176
}
177
+
protected getMachineId() {
178
+
return `${this.type}-${this.name}`;
179
+
}
172
180
173
181
public notify = async (payload: WebhookPayload) => {
174
182
this.emitEvent('notify', payload);
···
293
301
return plays[0][1].play;
294
302
}
295
303
}
304
+
}
305
+
306
+
protected async doParseCache(): Promise<true | string | undefined> {
307
+
const cachedQueue = (await this.cache.cacheScrobble.get(`${this.getMachineId()}-queue`) as QueuedScrobble<PlayObject>[] ?? []);
308
+
const cachedQLength = cachedQueue.length;
309
+
this.queuedScrobbles = cachedQueue;
310
+
311
+
const cachedDead = (await this.cache.cacheScrobble.get(`${this.getMachineId()}-dead`) as DeadLetterScrobble<PlayObject>[] ?? []);
312
+
const cachedDLength = cachedDead.length;
313
+
this.deadLetterScrobbles = cachedDead;
314
+
315
+
return `Scrobbled from Cache: ${cachedQLength} Queue | ${cachedDLength} Dead Letter`;
296
316
}
297
317
298
318
protected async postInitialize(): Promise<void> {
···
887
907
}
888
908
this.logger.info(`Removed scrobble ${buildTrackString(this.deadLetterScrobbles[index].play)} from queue`, {leaf: 'Dead Letter'});
889
909
this.deadLetterScrobbles.splice(index, 1);
910
+
this.cache.cacheScrobble.set(`${this.getMachineId()}-dead`, this.deadLetterScrobbles);
890
911
}
891
912
892
913
removeDeadLetterScrobbles = () => {
893
914
this.deadLetterScrobbles = [];
915
+
this.cache.cacheScrobble.set(`${this.getMachineId()}-dead`, []);
894
916
this.logger.info('Removed all scrobbles from queue', {leaf: 'Dead Letter'});
895
917
}
896
918
···
910
932
this.queuedScrobbles.push(queuedPlay);
911
933
}
912
934
this.queuedScrobbles.sort((a, b) => sortByOldestPlayDate(a.play, b.play));
935
+
this.cache.cacheScrobble.set(`${this.getMachineId()}-queue`, this.queuedScrobbles);
913
936
}
914
937
915
938
protected addDeadLetterScrobble = (data: QueuedScrobble<PlayObject>, error: (Error | string) = 'Unspecified error') => {
···
923
946
this.deadLetterScrobbles.push(deadData);
924
947
this.deadLetterScrobbles.sort((a, b) => sortByOldestPlayDate(a.play, b.play));
925
948
this.emitEvent('deadLetter', {dead: deadData});
949
+
this.cache.cacheScrobble.set(`${this.getMachineId()}-dead`, this.deadLetterScrobbles);
926
950
}
927
951
928
952
queuePlayingNow = (data: PlayObject, source: SourceIdentifier) => {
+4
src/backend/tests/scrobbler/scrobblers.test.ts
+4
src/backend/tests/scrobbler/scrobblers.test.ts
···
16
16
17
17
import { NowPlayingScrobbler, TestAuthScrobbler, TestScrobbler } from "./TestScrobbler.js";
18
18
import { PlayPlatformId } from '../../common/infrastructure/Atomic.js';
19
+
import { getRoot } from '../../ioc.js';
20
+
import { loggerTest } from '@foxxmd/logging';
19
21
20
22
chai.use(asPromised);
21
23
···
28
30
const normalizedWithMixedDur = normalizePlays(mixedDurPlays, {initialDate: firstPlayDate});
29
31
30
32
const normalizedWithMixedDurOlder = normalizePlays(mixedDurPlays, {initialDate: olderFirstPlayDate});
33
+
34
+
getRoot({cache: {scrobble: {provider: 'memory'}}, logger: loggerTest});
31
35
32
36
const generateTestScrobbler = () => {
33
37
const testScrobbler = new TestScrobbler();