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.
25 kB · 602 lines
TypeScript
at database
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603import { LogDataPretty, Logger, LogLevel } from "@foxxmd/logging";import { Express } from 'express';import bsseDef from 'better-sse';import bodyParser from "body-parser";import { FixedSizeList } from 'fixed-size-list';import { PassThrough } from "node:stream";import { Transform } from "stream";import { CLIENT_DEAD_QUEUE, ClientStatusData, DeadLetterScrobble, LeveledLogData, LogOutputConfig, PlayObject, SOURCE_SOT, SOURCE_SOT_TYPES, SourcePlayerJson, SourceStatusData,} from "../../core/Atomic.js";import { capitalize } from "../../core/StringUtils.js";import { ExpressHandler } from "../common/infrastructure/Atomic.js";import { getRoot } from "../ioc.js";import AbstractScrobbleClient from "../scrobblers/AbstractScrobbleClient.js";import AbstractSource from "../sources/AbstractSource.js";import MemorySource from "../sources/MemorySource.js";import { parseBool } from "../utils.js";import { sortByNewestPlayDate } from '../../core/PlayUtils.js';import { setupAuthRoutes } from "./auth.js";import { setupDeezerRoutes } from "./deezerRoutes.js";import {setupLZEndpointRoutes} from "./endpointListenbrainzRoutes.js";import {setupLastfmEndpointRoutes} from "./endpointLastfmRoutes.js";import { makeClientCheckMiddle, makeSourceCheckMiddle } from "./middleware.js";import { setupWebscrobblerRoutes } from "./webscrobblerRoutes.js";import ScrobbleSources from "../sources/ScrobbleSources.js";import ScrobbleClients from "../scrobblers/ScrobbleClients.js";import prom from 'prom-client';import { SimpleError } from "../common/errors/MSErrors.js";import { QueryPlaysOpts } from "../common/database/drizzle/repositories/PlayRepository.js";import { playSelectToDeadScrobble } from "../common/database/drizzle/entityUtils.js";
const maxBufferSize = 300;const output: Record<number, FixedSizeList<LogDataPretty>> = {};
const createAddToLogBuffer = (levelMap: {[p: number]: string}) => (log: LogDataPretty) => { output[log.level].add({...log, levelLabel: levelMap[log.level]});}
const getLogs = (minLevel: number, limit: number = maxBufferSize, sort: 'asc' | 'desc' = 'desc'): LeveledLogData[] => { const allLogs: LeveledLogData[][] = []; for(const level of Object.keys(output)) { if(Number.parseInt(level) >= minLevel) { allLogs.push(output[level].data); } } if(sort === 'desc') { return allLogs.flat(1).sort((a, b) => b.time - a.time).slice(0, limit); } return allLogs.flat(1).sort((a, b) => a.time - b.time).slice(0, limit);}
export const setupApi = (app: Express, logger: Logger, appLoggerStream: PassThrough, initialLogOutput: LogDataPretty[] = [], scrobbleSources: ScrobbleSources, scrobbleClients: ScrobbleClients) => { for(const level of Object.keys(logger.levels.labels)) { output[level] = new FixedSizeList<LeveledLogData>(maxBufferSize); }
const addToLogBuffer = createAddToLogBuffer(logger.levels.labels); for(const log of initialLogOutput) { addToLogBuffer(log); } const root = getRoot();
//let logWebLevel: LogLevel = logger.level as LogLevel || (process.env.LOG_LEVEL || 'info') as LogLevel;
const logConfig: LogOutputConfig = { level: logger.level as LogLevel || (process.env.LOG_LEVEL || 'trace') as LogLevel, sort: 'descending', limit: 50, }
let logObjectStream: Transform; try { logObjectStream = new Transform({ transform: (chunk, e, cb) => { cb(null, chunk) }, objectMode: true, allowHalfOpen: true }) } catch (e) { console.log(e); }
appLoggerStream.on('data', (log: LogDataPretty) => { addToLogBuffer(log); if(log.level >= logger.levels.values[logConfig.level]) { logObjectStream.write({message: log.line, level: log.level, levelLabel: logger.levels.labels[log.level]}); } });
const clientMiddleFunc = makeClientCheckMiddle(scrobbleClients); const sourceMiddleFunc = makeSourceCheckMiddle(scrobbleSources);
const clientRequiredMiddle = clientMiddleFunc(true); const sourceRequiredMiddle = sourceMiddleFunc(true);
const setLogWebSettings: ExpressHandler = async (req, res, next) => { // @ts-expect-error logLevel not part of session const sessionLevel: LogLevel | undefined = req.session.logLevel as LogLevel | undefined; if(sessionLevel !== undefined && logConfig.level !== sessionLevel) { logConfig.level = sessionLevel; } // @ts-expect-error limit not part of session const sessionLimit: number | undefined = req.session.limit as number | undefined; if(sessionLimit !== undefined && logConfig.limit !== sessionLimit) { logConfig.limit = sessionLimit; } next(); }
app.get('/api/logs/stream', setLogWebSettings, async (req, res) => { const session = await bsseDef.createSession(req, res); await session.stream(logObjectStream); });
app.get('/api/logs', setLogWebSettings, async (req, res) => { const slicedLog = getLogs(logger.levels.values[logConfig.level], logConfig.limit + 1, logConfig.sort === 'ascending' ? 'asc' : 'desc'); return res.json({data: slicedLog, settings: logConfig}); });
app.put('/api/logs', async (req, res) => { logConfig.level = req.body.level as LogLevel | undefined ?? logConfig.level; logConfig.limit = req.body.limit ?? logConfig.limit; const slicedLog = getLogs(logger.levels.values[logConfig.level], logConfig.limit + 1, logConfig.sort === 'ascending' ? 'asc' : 'desc'); // @ts-expect-error logLevel not part of session req.session.logLevel = logConfig.level; // @ts-expect-error limit not part of session req.session.limit = logConfig.limit; return res.json({data: slicedLog, settings: logConfig}); });
app.get('/api/events', async (req, res) => { const session = await bsseDef.createSession(req, res); scrobbleSources.emitter.on('*', (payload: any, eventName: string) => { if(payload.from !== undefined) { session.push({event: eventName, ...payload}, payload.from); } }); scrobbleClients.emitter.on('*', (payload: any, eventName: string) => { if(payload.from !== undefined) { session.push({event: eventName, ...payload}, payload.from); } }); });
setupDeezerRoutes(app, logger, scrobbleSources); setupWebscrobblerRoutes(app, logger, scrobbleSources); setupLZEndpointRoutes(app, logger, scrobbleSources, scrobbleClients); setupLastfmEndpointRoutes(app, logger, scrobbleSources); setupAuthRoutes(app, logger, sourceRequiredMiddle, clientRequiredMiddle, scrobbleSources, scrobbleClients);
app.put('/api/webscrobbler', bodyParser.json({type: ['text/*', 'application/json']}), async (req, res) => { logger.info(req.body); res.sendStatus(200); });
app.get('/api/webscrobbler', bodyParser.json({type: ['text/*', 'application/json']}), async (req, res) => { logger.info(req.body); res.sendStatus(200); });
app.get('/api/status', async (req, res, next) => {
const sourceData = scrobbleSources.sources.map((x) => { const { type, tracksDiscovered = 0, name, canPoll = false, polling = false, requiresAuth = false, requiresAuthInteraction = false, authed = false, } = x; const base: SourceStatusData = { status: '', type, display: capitalize(type), tracksDiscovered, name, canPoll, hasAuth: requiresAuth, hasAuthInteraction: requiresAuthInteraction, authed, players: 'players' in x ? (x as MemorySource).playersToObject() as unknown as Record<string,SourcePlayerJson> : {}, sot: ('playerSourceOfTruth' in x) ? x.playerSourceOfTruth as SOURCE_SOT_TYPES : SOURCE_SOT.HISTORY, supportsUpstreamRecentlyPlayed: x.supportsUpstreamRecentlyPlayed, supportsManualListening: x.supportsManualListening, manualListening: x.manualListening, systemListeningBehavior: x.getSystemListeningBehavior(), ...x.additionalApiData() }; if(!x.isReady()) { if(x.buildOK === false) { base.status = 'Initializing Data Failed'; } else if(x.connectionOK === false) { base.status = 'Communication Failed'; } else if (requiresAuth && !authed) { base.status = requiresAuthInteraction ? 'Auth Interaction Required' : 'Authentication Failed Or Not Attempted' } else { base.status = 'Not Ready'; } } else { if (canPoll) { base.status = polling ? 'Polling' : 'Idle'; } else { base.status = !x.instantiatedAt.isSame(x.lastActivityAt) ? 'Received Data' : 'Awaiting Data'; } } return base; });
const clientData = scrobbleClients.clients.map((x) => { const { type, tracksScrobbled = 0, name, requiresAuth = false, requiresAuthInteraction = false, authed = false, scrobbling = false, } = x; const base: ClientStatusData = { status: '', type, display: capitalize(type), scrobbled: tracksScrobbled, name, hasAuth: requiresAuth, hasAuthInteraction: requiresAuthInteraction, authed, initialized: x.isReady(), deadLetterScrobbles: x.deadLetterQueued, // x.deadLetterScrobbles.length, deadLetterScrobblesTotal: x.deadLetterLength, queued: x.queuedLength // x.queuedScrobbles.length }; if (!base.initialized) { if(x.buildOK === false) { base.status = 'Initializing Data Failed'; } else if(x.connectionOK === false) { base.status = 'Communication Failed'; } else if (requiresAuth && !authed) { base.status = requiresAuthInteraction ? 'Auth Interaction Required' : 'Authentication Failed Or Not Attempted' } else { base.status = 'Not Ready'; } } else { base.status = scrobbling ? 'Running' : 'Idle'; } return base; }); return res.json({sources: sourceData, clients: clientData}); });
app.get('/api/recent', sourceMiddleFunc(false), async (req, res, next) => { const { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message scrobbleSource: source, query: { upstream = 'false', next: queryNext = 'false', ...rest } } = req;
let result: PlayObject[] = []; if (source !== undefined) { if (upstream === 'true' || upstream === '1') { if (!(source as AbstractSource).supportsUpstreamRecentlyPlayed) { return res.status(409).json({message: 'Fetching upstream recently played is not supported for this source'}); } try { result = await (source as AbstractSource).getUpstreamRecentlyPlayed(); } catch (e) { return res.status(500).json({message: e.message}); } } else { if(queryNext === 'true') { return res.json(await (source as AbstractSource).getRecentPlaysApi(rest)); } result = await (source as AbstractSource).getFlatRecentlyDiscoveredPlays(); } }
return res.json(result); });
app.get('/api/source/art', sourceMiddleFunc(false), async (req, res, next) => { const { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message scrobbleSource, query: { data } } = req;
const source = scrobbleSource as AbstractSource; if(!(source instanceof MemorySource)) { return res.status(500).json({message: 'Source does not support players'}); }
if('getSourceArt' in source && typeof source.getSourceArt === 'function') { const [stream, contentType] = await source.getSourceArt(data); res.writeHead(200, {'Content-Type': contentType}); try { return stream.pipe(res); } catch (e) { logger.error(new Error(`Error occurred while trying to stream art for ${source.name} (${source.type}) | Data ${data}`, {cause: e})); return res.status(500).json({message: 'Error during art retrieval'}); } } else { return res.status(500).json({message: `Source ${source.name} (${source.type} does not support art retrieval`}); } });
app.get('/api/dead', clientMiddleFunc(true), async (req, res, next) => { const { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message scrobbleClient: client, query = {} } = req;
const deadQuery: QueryPlaysOpts = { ...query as Partial<QueryPlaysOpts>, queues: [ { queueName: CLIENT_DEAD_QUEUE, queueStatus: ['queued','failed'] } ] }
// @ts-ignore const result: DeadLetterScrobble<PlayObject>[] = (await (client as AbstractScrobbleClient).getPlaysPaginated(deadQuery)).data.map(x => playSelectToDeadScrobble(x, true));
return res.json(result); });
app.put('/api/dead', clientMiddleFunc(true), async (req, res, next) => { const { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message scrobbleClient: client, } = req;
(client as AbstractScrobbleClient).logger.verbose('User requested processing of all dead letter scrobbles via API');
res.status(200).send('OK');
await ((client as AbstractScrobbleClient).processDeadLetterQueue(1000)); });
app.put('/api/dead/:id', clientMiddleFunc(true), async (req, res, next) => { const { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message scrobbleClient: client, params: { id } = {} } = req;
const deadId = id as string;
(client as AbstractScrobbleClient).logger.verbose(`User requested processing of dead letter scrobble ${deadId} via API call`)
try { const [scrobbled, dead] = await (client as AbstractScrobbleClient).processDeadLetterScrobble(deadId); if(scrobbled) { return res.status(200).send(); } return res.json(playSelectToDeadScrobble(dead, true)); } catch (e) { if(e.message.includes(`Play ${deadId} does not exist`)) { logger.warn(e); return res.status(404).json({error: e}); } logger.error(e); return res.status(500).json({error: e}); } });
app.delete('/api/dead', clientMiddleFunc(true), async (req, res, next) => { const { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message scrobbleClient: client, } = req;
(client as AbstractScrobbleClient).logger.verbose('User requested deletion of all dead letter scrobbles via API');
(client as AbstractScrobbleClient).removeDeadLetterScrobbles(['queued', 'failed'], 'failed', false).then(() => null).catch((e) => logger.error(e));
return res.sendStatus(200); });
app.delete('/api/dead/:id', clientMiddleFunc(true), async (req, res, next) => { const { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message scrobbleClient: client, params: { id } = {} } = req;
const deadId = id as string;
(client as AbstractScrobbleClient).logger.verbose(`User requested removal of dead letter scrobble ${deadId} via API call`)
try { await (client as AbstractScrobbleClient).removeDeadLetterScrobble(deadId,'failed', false); return res.status(200).send(); } catch (e) { if(e.message.includes(`Play ${deadId} does not exist`)) { logger.warn(e); return res.status(404).json({error: e}); } logger.error(e); return res.status(500).json({error: e}); } });
app.get('/api/scrobbled', clientMiddleFunc(false), async (req, res, next) => { const { // @ts-expect-error scrobbleClient not part of req scrobbleClient: client, query } = req;
let result: PlayObject[] = []; if (client !== undefined) { const q: Partial<QueryPlaysOpts> = { ...query as Partial<QueryPlaysOpts>, state: ['scrobbled'] } result = [...(await (client as AbstractScrobbleClient).getPlaysPaginated(q)).data.map(x => x.play)].sort(sortByNewestPlayDate); }
return res.json(result); });
app.use('/api/source/init', sourceRequiredMiddle); app.post('/api/source/init', async (req, res) => { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message const source = req.scrobbleSource as AbstractSource;
const { query: { force: forceQ = false } } = req;
const force = parseBool(forceQ, false);
source.logger.verbose(`User requested${force ? ' a FORCED' :''} (re)init via API call`);
res.status(200).send('OK');
if(source.polling) { source.logger.info('Source is already polling! Restarting polling...'); const stopRes = await source.tryStopPolling(new SimpleError('user initiated', {simple: true, shortStack: true})); if(stopRes === true) { source.poll({force, notify: false}).catch(e => source.logger.error(e)); } } else { source.poll({force, notify: false}).catch(e => source.logger.error(e)); } });
app.use('/api/source/listen', sourceRequiredMiddle); app.post('/api/source/listen', async (req, res) => { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message const source = req.scrobbleSource as AbstractSource;
const { query: { listening: listeningQ } } = req;
if(!source.supportsManualListening) { source.logger.warn('This source does not support manual Should Scrobble state'); res.status(400).send(); return; } let listening: boolean | undefined; if(listeningQ !== undefined) { listening = parseBool(listeningQ) } source.logger.verbose(`User requested Should Scrobble status ${listening === undefined ? 'system' : listening}`);
source.manualListening = listening;
res.status(200).json({listening}); });
app.use('/api/client/init', clientRequiredMiddle); app.post('/api/client/init', async (req, res) => { // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message const client = req.scrobbleClient as AbstractScrobbleClient;
const { query: { force: forceQ = false } } = req;
const force = parseBool(forceQ, false);
client.logger.verbose(`User requested${force ? ' a FORCED' :''} (re)init via API call`);
client.logger.info('Checking (and trying) to stop scrobbler if already running...'); if(false === (await client.tryStopScrobbling(new SimpleError('user initiated', {simple: true, shortStack: true})))) { return res.status(500).send(); }
client.logger.info('Trying to start scrobbler...'); client.initScrobbleMonitoring({force, notify: false}).catch(e => client.logger.error(e)); res.status(200).send('OK'); });
app.get('/health', async (req, res) => res.redirect(307, `/api/${req.url.slice(1)}`)); app.get('/api/health', async (req, res) => { const { type, name } = req.query;
const [sourcesReady, sourceMessages] = await scrobbleSources.getStatusSummary(type as string|undefined, name as string|undefined); const [clientsReady, clientMessages] = await scrobbleClients.getStatusSummary(type as string|undefined, name as string|undefined);
return res.status((clientsReady && sourcesReady) ? 200 : 500).json({messages: sourceMessages.concat(clientMessages)}); });
const issuesClientGauge = new prom.Gauge({ name: 'multiscrobbler_client_issues', help: 'Number of errors/issues with Client', labelNames: ['name', 'type'], async collect() { for(const client of scrobbleClients.clients) { let issues = 0; if(!(await client.isReady())) { issues++; } this.labels({name: client.getSafeExternalName(), type: client.type}).set(issues); } } });
const sourceIssues = new prom.Gauge({ name: 'multiscrobbler_source_issues', help: 'Number of errors/issues with Source', labelNames: ['name', 'type'], async collect() { for(const source of scrobbleSources.sources) { let issues = 0; if(source.requiresAuth && !source.authed) { issues++; } if(source.canPoll && !source.polling) { issues++; } this.labels({name: source.getSafeExternalName(), type: source.type}).set(issues); } } });
if(process.env.PROMETHEUS_FULL === 'true') { prom.collectDefaultMetrics(); }
app.get('/api/metrics', async (req, res) => {
const metricsString = await prom.register.metrics(); return res .status(200) .set('Content-Type', 'text/plain') .send(metricsString);
});
app.get('/api/version', async (req, res) => { return res.json({version: root.get('version')}); });
app.use('/api/*path', async (req, res) => { const remote = req.connection.remoteAddress; const proxyRemote = req.headers["x-forwarded-for"]; const ua = req.headers["user-agent"]; logger.debug(`Server received ${req.method} request from ${remote}${proxyRemote !== undefined ? ` (${proxyRemote})` : ''}${ua !== undefined ? ` (UA: ${ua})` : ''} to unknown route: ${req.originalUrl}`); return res.sendStatus(404); });}