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.
21 kB · 445 lines
TypeScript
at queueNext
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446import type {Logger} from "@foxxmd/logging";import dayjs, { type Dayjs } from "dayjs";import type { EventEmitter } from "events";import { AsyncTask, SimpleIntervalJob, ToadScheduler } from "toad-scheduler";import { PARSED_FROM, type PlayObject, SOURCE_SOT, type SOURCE_SOT_TYPES, type SourcePlayerJson, type SourcePlayerObj } from "../../core/Atomic.ts";import { buildTrackString } from "../../core/StringUtils.ts";import { asPlayerStateDataMaybePlay, type InternalConfig, type PlayerStateDataMaybePlay, type ProgressAwarePlayObject} from "../common/infrastructure/Atomic.ts";import { CALCULATED_PLAYER_STATUSES } from '../../core/Atomic.ts';import type {PlayPlatformId} from '../../core/Atomic.ts';import type {SourceConfig} from '../common/infrastructure/config/source/sources.ts';import type {SourceType} from "../../core/Atomic.ts";import type {PollingOptions} from "../common/infrastructure/config/common.ts";import { getPlatformIdFromData, isDebugMode, playObjDataMatch, thresholdResultSummary,} from "../utils.ts";import { genGroupIdStr } from '../../core/PlayUtils.ts';import { formatNumber } from '../../core/DataUtils.ts';import { timePassesScrobbleThreshold } from "../utils/TimeUtils.ts";import { timeToHumanTimestamp } from "../../core/TimeUtils.ts";import { PromisePool } from "@supercharge/promise-pool";import AbstractSource from "./AbstractSource.ts";import type { AbstractPlayerState} from "./PlayerState/AbstractPlayerState.ts";import { createPlayerOptions, type PlayerStateOptions } from "./PlayerState/AbstractPlayerState.ts";import { GenericPlayerState } from "./PlayerState/GenericPlayerState.ts";import { hashObject } from "../utils/StringUtils.ts";import type {ComponentSourceApiJson} from "../../core/Api.ts";
const EXPECTED_NON_DISCOVERED_REASON = 'not added because an identical play with the same timestamp was already discovered.';
export default class MemorySource extends AbstractSource {
playerSourceOfTruth: SOURCE_SOT_TYPES = SOURCE_SOT.PLAYER;
/* * MemorySource uses its own state to maintain a list of recently played tracks and determine if a track is valid. * This is necessary for any source that * * doesn't have its own source of truth for "recently played" or * * that does not return "started at" and "duration" timestamps for recent plays or * * where these timestamps don't have enough granularity (IE second accuracy) * such as subsonic and jellyfin */
players: Map<string, AbstractPlayerState> = new Map(); playerState: Map<string, string> = new Map(); playerCleanupDiscoveryAttempt: Map<string, boolean> = new Map();
deceasedPlayers: Map<string, Dayjs> = new Map();
scheduler: ToadScheduler = new ToadScheduler();
protected isPositional: boolean = false;
constructor(type: SourceType, name: string, config: SourceConfig, internal: InternalConfig, emitter: EventEmitter) { super(type, name, config, internal, emitter);
// player cleanup on *schedule* is needed when the Source is non-polling (ingress) // because if the source stops sending updates then processRecentPlays() was never called so we never remove old players this.scheduler.addSimpleIntervalJob(new SimpleIntervalJob({ seconds: 10 }, new AsyncTask('Player Cleanup', (): Promise<any> => { if (this.canPoll) { return Promise.resolve(); } return PromisePool .withConcurrency(1) .for(this.players.keys()) .process(async (key) => { await this.cleanupPlayer(key); }); }))); }
[Symbol.dispose]() { this.scheduler.stop(); for(const job of this.scheduler.getAllJobs()) { this.scheduler.removeById(job.id); } for(const p of this.players.keys()) { this.deletePlayer(p); } }
async [Symbol.asyncDispose]() { await super[Symbol.asyncDispose](); this.scheduler.stop(); for(const job of this.scheduler.getAllJobs()) { this.scheduler.removeById(job.id); } this[Symbol.dispose](); }
cleanupPlayer = async (key: string): Promise<PlayObject | undefined> => { const player = this.players.get(key); if(player === undefined) { this.logger.warn({labels: 'Player Cleanup'},`No Player with ID ${key} exists! Cannot cleanup.`); return; } let discoveredCleanupPlay: PlayObject | undefined; let label = 'Player Cleanup', deletePlayer = false; // no communication from the source was received for this player const isStale = player.checkStale(); if (isStale && player.checkOrphaned() && player.isDead()) { deletePlayer = true; label = 'Dead Player Cleanup'; } else if (isStale) { label = 'Stale Player Cleanup'; const state = player.getApiState(); const stateHash = hashObject(state); if(stateHash !== this.playerState.get(key)) { this.playerState.set(key, stateHash); this.emitEvent('playerUpdate', { ...state, options: { wasMonitored: this.isMonitoring(), scrobbleTo: this.clients } }); } if(this.config.options?.logPlayerState === true || isDebugMode()) { player.logSummary(); } } else { // player is not stale this.playerCleanupDiscoveryAttempt.delete(key); return; }
// player was stale or orphaned/dead // if we haven't already tried to discover any in-progress plays then do it now (and only once) if(!this.playerCleanupDiscoveryAttempt.has(key)) { this.playerCleanupDiscoveryAttempt.set(key, true); // get play as completed const cleanupPlay = player.getPlayedObject(true); let discoverablePlay: boolean; if(cleanupPlay !== undefined) { const [discoverable, discoverableReason] = await this.isListenedPlayDiscoverable(cleanupPlay); discoverablePlay = discoverable; if(this.playerSourceOfTruth === SOURCE_SOT.PLAYER) { player.logger.verbose({labels: label}, discoverableReason); } if(discoverable) { discoveredCleanupPlay = cleanupPlay; // we are discovering/scrobbling play // and since player is now stale we should treat this "session" as ended // -- so if user resumes a stale play later its a new session (new real time period of them listening) // basically this is the same as if the player was orphaned and removed // // so we remove listen ranges so the old accumulated listen time can't be used for the "new" listening session player.listenRanges = []; player.currentListenRange = undefined; } } } if(deletePlayer) { this.deletePlayer(player.platformIdStr, `Removed after being orphaned for ${timeToHumanTimestamp(dayjs.duration(player.stateIntervalOptions.orphanedInterval, 'seconds'))}`); }
return discoveredCleanupPlay; }
playersToObject = (): Record<string, SourcePlayerObj> => { if(this.players.size === 0) { return {}; } const record: Record<string, SourcePlayerObj> = {}; for(const [k,v] of this.players.entries()) { record[k] = v.getApiState(); } return record; }
public getApiData(): ComponentSourceApiJson { return { ...super.getApiData(), sot: this.playerSourceOfTruth, players: this.playersToObject() as unknown as Record<string, SourcePlayerJson> } }
getNewPlayer = (logger: Logger, id: PlayPlatformId, opts: PlayerStateOptions): AbstractPlayerState => new GenericPlayerState(logger, id, opts)
setNewPlayer = (idStr: string, logger: Logger, id: PlayPlatformId, opts: PlayerStateOptions = {}) => { this.players.set(idStr, this.getNewPlayer(this.logger, id, { ...createPlayerOptions(this.config.data as Partial<PollingOptions>, this.playerSourceOfTruth, this.logger), ...opts })); this.playerState.set(idStr, ''); }
hasPlayer = (data: string | PlayerStateDataMaybePlay): boolean => { let id: string; if(typeof data === 'string') { id = data; } else { id = genGroupIdStr(getPlatformIdFromData(data)); } return this.players.has(id); }
isZombiePlayer = (id: string, lastUpdated: Dayjs): boolean => { return this.deceasedPlayers.has(id) && this.deceasedPlayers.get(id).isSame(lastUpdated); }
genPlayerId = (data: PlayObject | PlayerStateDataMaybePlay): string => { return genGroupIdStr(getPlatformIdFromData(data)); }
deletePlayer = (id: string, reason?: string) => { if(!this.players.has(id)) { return; } if(reason !== undefined) { this.players.get(id)?.logger.debug(reason); } using player = this.players.get(id); this.deceasedPlayers.set(id, player.stateLastUpdatedAt); player[Symbol.dispose](); this.players.delete(id); this.playerState.delete(id); this.emitEvent('playerDelete', {platformId: id}); }
pickPlatformSession = (sessions: (PlayObject | PlayerStateDataMaybePlay)[], player: AbstractPlayerState): PlayObject | PlayerStateDataMaybePlay => { if(sessions.length > 1) { player.logger.debug(`More than one data/state found in incoming data, will only use first found.`); } return sessions[0]; } processRecentPlays = async (datas: (PlayObject | PlayerStateDataMaybePlay)[], reportedTS?: Dayjs) => {
const { options: { scrobbleThresholds = {} } } = this.config;
const newStatefulPlays: PlayObject[] = [];
// create any new players from incoming data for (const data of datas) { const id = getPlatformIdFromData(data); const idStr = this.genPlayerId(data); if (!this.players.has(idStr)) { if(asPlayerStateDataMaybePlay(data) && data.stateUpdatedAt !== undefined) { if(this.isZombiePlayer(idStr, data.stateUpdatedAt)) { this.logger.trace(`Not creating player for ${idStr} because last state update timestamp has not changed since it was deleted.`); continue; } else { // cleaning up in case it already existed but has new timestamp this.deceasedPlayers.delete(idStr); } } this.setNewPlayer(idStr, this.logger, id);
if(!this.multiPlatform && this.players.size > 1) { // new platform should have old platform data transferred const [id,firstPlayer] = Array.from(this.players.entries())[0]; const newPlayer = this.players.get(idStr); firstPlayer.transferToNewPlayer(newPlayer); this.deletePlayer(id, 'Removed due to player transfer'); } } }
for (const [key, player] of this.players.entries()) {
let incomingData: PlayObject | PlayerStateDataMaybePlay; // get all incoming datas relevant for each player (this should only be one) const relevantDatas = datas.filter(x => { const id = getPlatformIdFromData(x); return player.platformEquals(id); });
let hasFreshState = false; if (relevantDatas.length > 0) { hasFreshState = true; this.lastActivityAt = dayjs(); incomingData = this.pickPlatformSession(relevantDatas, player); if(asPlayerStateDataMaybePlay(incomingData) && incomingData.stateUpdatedAt !== undefined && player.stateLastUpdatedAt.isSame(incomingData.stateUpdatedAt)) { hasFreshState = false; player.logger.trace('Skipping update because it has the same timestamp as the previous update'); } }
// we've received some form of communication from the source for this player if (hasFreshState) { this.lastActivityAt = dayjs();
// reset any player cleanup state since we got fresh data this.playerCleanupDiscoveryAttempt.delete(key);
incomingData = this.pickPlatformSession(relevantDatas, player);
let playerState: PlayerStateDataMaybePlay; if(asPlayerStateDataMaybePlay(incomingData)) { playerState = incomingData; } else { playerState = {play: incomingData, platformId: getPlatformIdFromData(incomingData)}; } if(playerState.position === undefined && playerState.play !== undefined && playerState.play.meta.trackProgressPosition !== undefined) { playerState.position = playerState.play.meta?.trackProgressPosition; }
const [currPlay, prevPlay] = player.update(playerState, reportedTS); const candidate = prevPlay !== undefined ? prevPlay : currPlay; const playChanged = prevPlay !== undefined;
// wait to discover play until it is stale or current play has changed // so that our discovered track has an accurate "listenedFor" count if (candidate !== undefined && (playChanged || player.isUpdateStale())) { const [discoverable, discoverableReason] = await this.isListenedPlayDiscoverable(candidate); if(discoverable) { if(this.playerSourceOfTruth === SOURCE_SOT.PLAYER) { player.logger.verbose(discoverableReason); } newStatefulPlays.push(candidate) } else if(playChanged && this.playerSourceOfTruth === SOURCE_SOT.PLAYER) { player.logger.verbose(discoverableReason); } }
if(this.config.options?.logPlayerState === true || isDebugMode()) { player.logSummary(); } const apiState = player.getApiState(); this.playerState.set(key, hashObject(apiState)) this.emitEvent('playerUpdate', { ...apiState, options: { wasMonitored: this.isMonitoring(), scrobbleTo: this.clients } }); } else { const playFromCleanup = await this.cleanupPlayer(key); if(playFromCleanup !== undefined) { newStatefulPlays.push(playFromCleanup); } } }
if(this.playerSourceOfTruth === SOURCE_SOT.PLAYER) { if(this.players.size > 0) { this.setStatus(`Monitoring ${this.players.size} active Players`); } else { this.setStatus(`Monitoring for new Players`); } }
return newStatefulPlays.map((x) => ({...x, meta: {...x.meta, parsedFrom: PARSED_FROM.player}})); }
protected isListenedPlayDiscoverable = async (candidate: PlayObject): Promise<[boolean, string]> => {
const { options: { scrobbleThresholds = {} } } = this.config;
const stPrefix = `${buildTrackString(candidate, {include: ['trackId', 'artist', 'track']})}`; const thresholdResults = timePassesScrobbleThreshold(scrobbleThresholds, candidate.data.listenedFor, candidate.data.duration);
if (thresholdResults.passes) { const matchingRecent = await this.existingDiscovered(candidate); //sRecentlyPlayed.find(x => playObjDataMatch(x, candidate)); if (matchingRecent.match === false) { return [true,`${stPrefix} added after ${thresholdResultSummary(thresholdResults)} and not matching any prior plays`]; } else { const {data: {playDate, duration}} = candidate; const {closestMatchedPlay: {data: {playDate: rplayDate}} = {}} = matchingRecent; if (!playDate.isSame(rplayDate)) { if (duration !== undefined) { if (playDate.isAfter(rplayDate.add(duration, 's'))) { return [true,`${stPrefix} added after ${thresholdResultSummary(thresholdResults)} and having a different timestamp than a prior play`]; } return [false, `${stPrefix} ${EXPECTED_NON_DISCOVERED_REASON}`] } else { const discoveredPlays = await this.getRecentlyDiscoveredPlays(); if (discoveredPlays.length === 0 || !playObjDataMatch(discoveredPlays[0], candidate)) { // if most recent stateful play is not this track we'll add it return [true,`${stPrefix} added after ${thresholdResultSummary(thresholdResults)}. Matched other recent play but could not determine time frame due to missing duration. Allowed due to not being last played track.`]; } return [false, `${stPrefix} not added because it matched the last discovered play and could not determine time frame of play`]; } } else { return [false, `${stPrefix} ${EXPECTED_NON_DISCOVERED_REASON}`]; } } } return [false,`${stPrefix} not added because ${thresholdResultSummary(thresholdResults)}.`]; }
recentlyPlayedTrackIsValid = (playObj: any) => playObj.data.playDate.isBefore(dayjs().subtract(30, 's'))
protected getInterval(log?: boolean): number { /** * If any player is progressing, reports position, and play has duration * then we can modify polling interval so that we check source data just before track is supposed to end * which will give us more accurate data on when player moves to the next play = better duration reporting to scrobble clients * -- additionally, will have better confidence for fudging 100% duration played * */ let interval = super.getInterval(); if(this.players.size === 0) { return interval; } let logDecrease: undefined | string; for(const player of this.players.values()) { if(player.calculatedStatus === CALCULATED_PLAYER_STATUSES.playing) { const pos = player.getPosition(); if(pos !== undefined && this.isPositional && player.currentPlay !== undefined && player.currentPlay.data?.duration !== undefined) { const { data: { duration } = {} } = player.currentPlay; const remaining = duration - pos; if(remaining < interval + 2) { // interval should be at least 1 second so we don't spam sources when polling interval = Math.max(1, remaining - 2); logDecrease = `Temporarily decreasing polling interval to ${formatNumber(interval)}s due to Player ${player.platformIdStr} reporting track duration remaining (${formatNumber(remaining)}s) less than normal interval (${formatNumber(super.getInterval())}s)`; } } } } if(logDecrease !== undefined && log) { this.logger.debug(logDecrease); } return interval; }
public async destroy() { this.scheduler.stop(); await super.destroy(); }}
const sortByPlayDate = (a: ProgressAwarePlayObject, b: ProgressAwarePlayObject): number => { throw new Error("Function not implemented.");};