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.
20 kB · 544 lines
TypeScript
at queueNext
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544import { Cacheable, createKeyv, Keyv, type KeyvStoreAdapter, type KeyvOptions, type CacheableOptions, KeyvCacheableMemory } from 'cacheable';import { FlatCache, type FlatCacheOptions } from 'flat-cache';import KeyvValkey, { type KeyvValkeyOptions } from '@keyv/valkey';import dayjs, { type Dayjs } from 'dayjs';import duration from 'dayjs/plugin/duration.js';import isBetween from 'dayjs/plugin/isBetween.js';import relativeTime from 'dayjs/plugin/relativeTime.js';import isToday from 'dayjs/plugin/isToday.js';import timezone from 'dayjs/plugin/timezone.js';import utc from 'dayjs/plugin/utc.js';import clone from 'clone';import { childLogger, type Logger } from '@foxxmd/logging';import { getConfigDir } from './index.ts';import path from 'path';import { cacheFunctions } from "@foxxmd/regex-buddy-core";import { fileOrDirectoryIsWriteable } from '../utils/FSUtils.ts';import { asCacheConfig, type CacheAuthProvider, type CacheConfig, type CacheConfigOptions, type CacheConfigUser } from './infrastructure/Atomic.ts';import { Typeson } from 'typeson';import { builtin } from 'typeson-registry';import { loggerNoop } from './MaybeLogger.ts';import type { Gauge } from 'prom-client';import prom from 'prom-client';import { nonEmptyStringOrDefault } from '../../core/StringUtils.ts';
dayjs.extend(utc)dayjs.extend(isBetween);dayjs.extend(relativeTime);dayjs.extend(duration);dayjs.extend(timezone);dayjs.extend(isToday);
const typeson = new Typeson().register([ builtin,]);typeson.register({ Dayjs: [ (x) => dayjs.isDayjs(x), (d: Dayjs) => d.toJSON(), (date) => dayjs(date) ]});
export class MSCache {
config: Required<CacheConfigOptions>
cacheMetadata: Cacheable; cacheScrobble: Cacheable; cacheDb: Cacheable; cacheAuth: Cacheable; regexCache: ReturnType<typeof cacheFunctions>; cacheTransform: Cacheable; cacheClientScrobbles: Cacheable; cacheApi: Cacheable; hasInit: boolean = false;
logger: Logger;
cacheHits: Gauge; cacheMisses: Gauge; cacheSets: Gauge; cacheCount: Gauge; //cacheVSize: Gauge;
constructor(logger: Logger, config: CacheConfigOptions = {}) { this.logger = childLogger(logger, 'Cache');
const { metadata: { provider: mProvider = false, connection: mConn, //...restMetadata } = {}, scrobble: { provider: sProvider = false, connection: sConnection, //...restScrobble } = {}, auth: { provider: aProvider = false, connection: aConn, //...restAuth } = {}, } = config;
this.config = { metadata: { provider: mProvider, connection: mConn, }, scrobble: { provider: sProvider, connection: sConnection, }, auth: { provider: aProvider, connection: aConn, } };
this.regexCache = cacheFunctions(200);
// for testing we default to in memory const inMemory = new Cacheable({primary: initMemoryCache({lruSize: 500, ttl: '1m'})}); this.cacheTransform = inMemory; this.cacheClientScrobbles = inMemory; this.cacheMetadata = inMemory; this.cacheAuth = inMemory; this.cacheScrobble = inMemory; this.cacheApi = inMemory; this.cacheDb = new Cacheable({primary: initMemoryCache({lruSize: 500, ttl: '1m'})}); }
init = async (enableCollectors: boolean = false) => { await this.initMetadataCache(); await this.initScrobbleCache(); await this.initAuthCache();
if(enableCollectors) { this.enableCollectors(); } }
protected enableCollectors = () => {
const collectors: {cache: Cacheable, name: string}[] = [ { cache: this.cacheMetadata, name: 'metadata' }, { cache: this.cacheScrobble, name: 'queued_scrobbles' }, { cache: this.cacheTransform, name: 'transformer' }, { cache: this.cacheClientScrobbles, name: 'historical_scrobbles' }, { cache: this.cacheApi, name: 'external_apis' }, { cache: this.cacheDb, name: 'database' } ];
this.cacheHits = new prom.Gauge({ name: 'multiscrobbler_cache_hits', help: 'cache hits', labelNames: ['cacheType', 'tier'], collect() { for(const set of collectors) { const [primary, secondary] = getStat(set.cache, 'hits'); this.labels({cacheType: set.name, tier: 'primary'}).set(primary); if(secondary !== undefined) { this.labels({cacheType: set.name, tier: 'secondary'}).set(secondary); } }
} }); this.cacheMisses = new prom.Gauge({ name: 'multiscrobbler_cache_misses', help: 'cache misses', labelNames: ['cacheType', 'tier'], collect() { for(const set of collectors) { const [primary, secondary] = getStat(set.cache, 'misses'); this.labels({cacheType: set.name, tier: 'primary'}).set(primary); if(secondary !== undefined) { this.labels({cacheType: set.name, tier: 'secondary'}).set(secondary); } }
} });
this.cacheMisses = new prom.Gauge({ name: 'multiscrobbler_cache_sets', help: 'cache sets', labelNames: ['cacheType', 'tier'], collect() { for(const set of collectors) { const [primary, secondary] = getStat(set.cache, 'sets'); this.labels({cacheType: set.name, tier: 'primary'}).set(primary); if(secondary !== undefined) { this.labels({cacheType: set.name, tier: 'secondary'}).set(secondary); } }
} });
this.cacheCount = new prom.Gauge({ name: 'multiscrobbler_cache_count', help: 'number of keys in cache', labelNames: ['cacheType', 'tier'], collect() { for(const set of collectors) { const [primary] = getStat(set.cache, 'count', false); this.labels({cacheType: set.name, tier: 'primary'}).set(primary); } } });
// this.cacheVSize = new prom.Gauge({ // name: 'multiscrobbler_cache_vsize', // help: 'estimated byte size of values in cache', // labelNames: ['cacheType', 'tier'], // collect() { // for(const set of collectors) { // const [primary] = getStat(set.cache, 'vsize', false); // this.labels({cacheType: set.name, tier: 'primary'}).set(primary); // } // } // }); }
protected initCacheable = async (cacheFor: string, primaryConfig: CacheConfig, secondaryConfig?: CacheConfig) => {
const logger = childLogger(this.logger, cacheFor); const providerHints = []; if(primaryConfig.provider === false) { const cache = new Cacheable({primary: noopKeyv}); cache.stats.enabled = true; logger.verbose(`Cache Providers: Disabled`); return cache; }
providerHints.push(`${primaryConfig.provider} (Primary)`)
if(secondaryConfig === undefined || secondaryConfig.provider !== false) { providerHints.push(`Disabled (Secondary)`) } else { providerHints.push(`${secondaryConfig.provider} (Secondary)`); } logger.verbose(`Cache Providers: ${providerHints.join(' | ')}`);
const ns = `ms-${cacheFor.toLocaleLowerCase()}`;
const cacheOpts: CacheableOptions = {
}
try { cacheOpts.primary = await this.initCachableType(ns, primaryConfig, logger); } catch (e) { throw new Error('Could not init primary cache', {cause: e}); }
if(secondaryConfig !== undefined && secondaryConfig.provider !== false) { try { cacheOpts.secondary = await this.initCachableType(ns, secondaryConfig, logger); } catch (e) { this.logger.warn(e); } }
const cache = new Cacheable(cacheOpts); cache.stats.enabled = true; return cache;
}
protected initCachableType = async (namespace: string, config: CacheConfig, logger: Logger): Promise<Keyv<any> | KeyvStoreAdapter> => {
if (config.provider === 'memory') { return initMemoryCache({ namespace, lruSize: config.lruSize, ttl: config.ttl }); }
if (config.provider === 'valkey') { logger.debug(`Building valkey cache from ${config.connection}`); try { const cache = await initValkeyCache(namespace, config.connection, undefined, {ttl: config.ttl}); logger.debug('valkey cache connected'); return cache; } catch (e) { throw e; } } const confDir = getConfigDir(); if (config.provider === 'file') { logger.debug(`Building file cache from ${path.join(config.connection ?? confDir, `${namespace}.cache`)}`);
try { const [keyvFile] = await initFileCache({ ...config, cacheDir: config.connection ?? confDir, cacheId: `${namespace}.cache` }, {ttl: config.ttl}, logger); return keyvFile; } catch (e) { throw e; } } }
initScrobbleCache = async () => { if (!this.hasInit) { let scrobbleConfig: CacheConfig | undefined; try { if(asCacheConfig(this.config.scrobble)) { scrobbleConfig = this.config.scrobble; this.cacheScrobble = await this.initCacheable('Scrobble', this.config.scrobble); } } catch (e) { this.logger.warn(new Error('Could not validate scrobble config! No fallback is possible', {cause: e})); this.cacheScrobble = await this.initCacheable('Scrobble', {provider: false}); } } }
initMetadataCache = async () => { if (!this.hasInit) { let metadataConfig: CacheConfig | undefined; try { if(asCacheConfig(this.config.metadata)) { metadataConfig = this.config.metadata; } } catch (e) { this.logger.warn(new Error('Could not validate metadata config, will fallback to memory cache only', {cause: e})); } this.cacheMetadata = await this.initCacheable('Metadata', {provider: 'memory', ttl: '3m', lruSize: 100}, metadataConfig === undefined ? undefined : {...this.config.metadata, ttl: '15m'}); this.cacheMetadata.stats.enabled = true; this.cacheClientScrobbles = await this.initCacheable('Historical Scrobbles', {provider: 'memory', ttl: '2m', lruSize: 50}, metadataConfig === undefined ? undefined : {...this.config.metadata, ttl: '10m'}); this.cacheClientScrobbles.stats.enabled = true; this.cacheTransform = await this.initCacheable('Transform Data', {provider: 'memory', ttl: '2m', lruSize: 100}, metadataConfig === undefined ? undefined : {...this.config.metadata, ttl: '5m'}); this.cacheTransform.stats.enabled = true; this.cacheApi = await this.initCacheable('External API Responses', {provider: 'memory', ttl: '30s', lruSize: 100}, metadataConfig === undefined ? undefined : {...this.config.metadata, ttl: '20m'}); this.cacheApi.stats.enabled = true; } }
initAuthCache = async () => { if (!this.hasInit) { let authConfig: CacheConfig | undefined; try { if(asCacheConfig(this.config.auth)) { authConfig = this.config.auth; } } catch (e) { this.logger.warn(new Error('Could not validate auth config! will fallback to memory cache only', {cause: e})); this.cacheAuth = await this.initCacheable('Auth', {provider: false}); }
this.cacheAuth = await this.initCacheable('Auth', {provider: 'memory', ttl: '3m'}, authConfig); } }
}
export const initMemoryCache = <T = any>(opts: Parameters<typeof createKeyv>[0] = {}): Keyv<T> | KeyvStoreAdapter => { const { ttl = '60s', lruSize = 200, ...restOpts } = opts; const memory = createKeyv({ ttl, lruSize, // millisecond interval before checking for expired keys and deleting checkInterval: 10000, ...restOpts, useClone: false, }); // structuredClone does not work well with dayjs https://github.com/iamkun/dayjs/issues/2236 // but deep cloning is fine so disable useClone and provide our own cloning function memory.serialize = (data) => { return clone(data) as string; } memory.stats.enabled = true; return memory;}
export const flatCacheCreate = (opts: FlatCacheOptions) => { return new FlatCache({ ttl: 0, lruSize: 2000, cacheDir: opts.cacheDir, cacheId: opts.cacheId ?? 'ms.cache', persistInterval: 1 * 1000 * 10, expirationInterval: 1 * 1000 * 10, // 10 seconds ...opts });}
export const flatCacheLoad = async (flatCache: FlatCache, logger: Logger = loggerNoop): Promise<void> => {
const cachePath = path.join(flatCache.cacheDir, flatCache.cacheId); try { fileOrDirectoryIsWriteable(cachePath); } catch (e) { throw new Error(`Unable to use path for file cache at ${cachePath}`, { cause: e }) }
// if(fileExists(cachePath) && !fileExists(`${cachePath}.bak`)) { // logger.info(`Backing up ${cachePath} in preparation for migration to database...`); // await copyFile(cachePath, `${cachePath}.bak`); // logger.info(`Done! Backed up to ${cachePath}.bak\nAll data has been loaded into cache. It will be deleted (from cache) after migrating to database.\nIf there are migration issues or you wish to downgrade then overwrite ${cachePath} with the .bak backup copy`); // }
const streamPromise = new Promise((resolve, reject) => { flatCache.loadFileStream(cachePath, (progress: number, total: number) => { logger.trace(`Loading ${progress}/${total} chunks...`); }, () => { resolve(true); }, (err: Error) => { reject(err); }); });
try { await streamPromise; logger.debug(`File cache loaded`); return; } catch (e) { if (null !== e.message.match(/Cache file .+ does not exist/)) { let loadError: Error; try { const onlySaveError = (e: Error) => { loadError = e; }; flatCache.on('error', onlySaveError); flatCache.load(); if (loadError !== undefined) { throw loadError; } flatCache.off('error', onlySaveError); logger.debug(`File cache loaded`); return; } catch (e) { throw new Error(`Unable to use file cache at ${cachePath}`, { cause: e }); } } else { throw new Error(`Unable to use file cache at ${cachePath}`, { cause: e }); } }}
export const initFileCache = async (opts: FlatCacheOptions = {}, keyvOpts: KeyvOptions = {}, logger: Logger = loggerNoop): Promise<[Keyv | KeyvStoreAdapter | undefined, FlatCache | undefined]> => { const flatCache = flatCacheCreate(opts); try { await flatCacheLoad(flatCache, logger); flatCache.on('error', (e) => { logger.warn(e); }); flatCache.on('save', () => { logger.debug('Saved cache to file'); });
const cache = new Keyv({ store: flatCache, throwOnErrors: true, ...typesonMarshalling, ...keyvOpts }); cache.stats.enabled = true; return [cache, flatCache]; } catch (e) { throw e; }}
export const valkeyCacheCreate = (ns: string, connection: string, valkeyOpts: KeyvValkeyOptions = {}, keyvOpts: KeyvOptions = {}): Keyv => { const valkey = new KeyvValkey(connection, { maxRetriesPerRequest: 5, connectTimeout: 1100, ...valkeyOpts }); const kv = new Keyv({ store: valkey, throwOnErrors: true, namespace: ns, ...typesonMarshalling, ...keyvOpts }); kv.stats.enabled = true; return kv;}
export const initValkeyCache = async (ns: string, connection: string, valkeyOpts: KeyvValkeyOptions = {}, keyvOpts: KeyvOptions = {}): Promise<Keyv> => { const kv = valkeyCacheCreate(ns, connection, valkeyOpts, keyvOpts); try { await kv.get('test'); return kv; } catch (e) { throw new Error(`Unable to connect to cache ${connection}`, { cause: e }) }}
const typesonMarshalling: Pick<KeyvOptions, 'serialize' | 'deserialize'> = { serialize: (data) => { const str = typeson.stringifySync(data); return str; }, deserialize: (str) => { const data = typeson.parseSync(str); return data; }}
const getStat = (cache: Cacheable, statName: string, getSecondary: boolean = true): [number, number?] => { let primary = cache.stats[statName]; if(statName === 'count' && cache.primary.store instanceof KeyvCacheableMemory) { primary = cache.primary.store.store.size; } let secondary: number; if(getSecondary && cache.secondary !== undefined) { secondary = cache.secondary.stats[statName]; } return [primary, secondary];}
const noopKeyv: KeyvStoreAdapter = { opts: {}, namespace: 'noop', get: (_) => undefined, set: (_, __, ___) => undefined, delete: (_) => undefined, clear: () => Promise.resolve(), on: (_, __) => undefined}
export const parseUserConfig = (config: CacheConfigUser = {}, parentLogger: Logger = loggerNoop): CacheConfigOptions => { const logger = childLogger(parentLogger, 'Cache');
const valkeyEnvVal: string | undefined = nonEmptyStringOrDefault(process.env.CACHE_VALKEY); const { valkey = valkeyEnvVal, auth: { provider: aProvider = (process.env.CACHE_AUTH as (CacheAuthProvider | undefined) ?? 'file'), } = {} } = config; let authConn: string, authProvider = aProvider; if(authProvider === 'valkey') { if(valkey === undefined) { logger.warn(`Auth Provider set to 'valkey' but not valkey connection string was not provided, falling back to file.`); authConn = getConfigDir(); authProvider = 'file'; } else { authConn = valkey; } } else { if(authProvider !== 'file') { logger.warn(`Unsupported provider given for auth: ${authProvider}`); } authConn = getConfigDir(); authProvider = 'file'; }
return { metadata: { provider: valkey !== undefined ? 'valkey' : false, connection: valkey, }, auth: { provider: authProvider, connection: authConn, } };}