[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
more implementation for clients
+180
-71
+15
-17
src/backend/common/database/drizzle/repositories/PlayRepository.ts
+15
-17
src/backend/common/database/drizzle/repositories/PlayRepository.ts
···
52
52
super(db, 'plays', 'Plays', opts);
53
53
}
54
54
55
+
findByUid = async (componentId: number, uid: string, opts: {hydrate?: PlayHydateOptions[]} = {}): Promise<PlaySelect & {queueStates: QueueStateSelect[]}> => {
56
+
const res = await this.db.query.plays.findFirst({
57
+
where: {
58
+
uid,
59
+
componentId
60
+
},
61
+
with: {
62
+
queueStates: true
63
+
}
64
+
});
65
+
res.play = hydratePlaySelect(res, opts.hydrate);
66
+
return res;
67
+
}
68
+
55
69
createPlays = async (entitiesOpts: RepositoryCreatePlayOpts[], opts: {hydrate?: PlayHydateOptions[]} = {}) => {
56
70
57
71
const {
···
349
363
return recentPlatformIds.map(x => x.platformId);
350
364
}
351
365
352
-
public getQueueCount = async (componentId: number, queueNames: string[]): Promise<number> => {
353
-
// this.db.select({id: queueStates.id}).from(queueStates).where(and(
354
-
// eq(queueStates.componentId, componentId),
355
-
// inArray(queueStates.queueName, states)
356
-
// ))
357
-
return await this.db.$count(queueStates, and(
358
-
eq(queueStates.componentId, componentId),
359
-
inArray(queueStates.queueName, queueNames),
360
-
eq(queueStates.queueStatus, 'queued')
361
-
));
362
-
363
-
// return await this.db.$count(plays, and(
364
-
// eq(plays.componentId, componentId),
365
-
// inArray(plays.state, states)
366
-
// ));
367
-
}
368
-
369
366
public getQueueNext = async (componentId: number, queueName: string, opts: {order?: 'asc' | 'desc', retries?: number} = {}): Promise<PlaySelect & {queueStates: QueueStateSelect[]} | undefined> => {
370
367
const {
371
368
retries,
···
400
397
queueStates: true
401
398
}
402
399
});
400
+
res.play = asPlay(res.play);
403
401
return res;
404
402
}
405
403
+42
src/backend/common/database/drizzle/repositories/QueueRepository.ts
+42
src/backend/common/database/drizzle/repositories/QueueRepository.ts
···
1
+
import { Logger, eq, and, lte, inArray } from "drizzle-orm";
2
+
import { DrizzleBaseRepository, DrizzleRepositoryOpts } from "./BaseRepository.js";
3
+
import { getDb } from "../drizzleUtils.js";
4
+
import { ComponentNew, ComponentSelect, FindWhere, QueueStateSelect } from "../drizzleTypes.js";
5
+
import { components, queueStates } from "../schema/schema.js";
6
+
import { generateComponentEntity } from "../entityUtils.js";
7
+
import { CLIENT_DEAD_QUEUE } from "../../../../../core/Atomic.js";
8
+
9
+
export class DrizzleQueueRepository extends DrizzleBaseRepository<'queueStates'> {
10
+
11
+
constructor(db: ReturnType<typeof getDb>, opts: DrizzleRepositoryOpts = {}) {
12
+
super(db, 'queueStates', 'Queue', opts);
13
+
}
14
+
15
+
public deadFailedToQueue = async (componentId: number, retries: number): Promise<void> => {
16
+
await this.db.update(queueStates).set({
17
+
queueStatus: 'queued',
18
+
queueName: CLIENT_DEAD_QUEUE
19
+
}).where(and(
20
+
eq(queueStates.componentId, componentId),
21
+
lte(queueStates.retries, retries)
22
+
));
23
+
}
24
+
25
+
public failedQueueToCompleted = async (componentId: number): Promise<void> => {
26
+
await this.db.update(queueStates).set({
27
+
queueStatus: 'completed',
28
+
queueName: CLIENT_DEAD_QUEUE
29
+
}).where(and(
30
+
eq(queueStates.componentId, componentId),
31
+
eq(queueStates.queueStatus, 'queued')
32
+
));
33
+
}
34
+
35
+
public getQueueCount = async (componentId: number, queueNames: string[], queueStatus: QueueStateSelect['queueStatus'][] = ['queued']): Promise<number> => {
36
+
return await this.db.$count(queueStates, and(
37
+
eq(queueStates.componentId, componentId),
38
+
inArray(queueStates.queueName, queueNames),
39
+
inArray(queueStates.queueStatus, queueStatus)
40
+
));
41
+
}
42
+
}
-1
src/backend/common/database/drizzle/schema/schema.ts
-1
src/backend/common/database/drizzle/schema/schema.ts
+123
-53
src/backend/scrobblers/AbstractScrobbleClient.ts
+123
-53
src/backend/scrobblers/AbstractScrobbleClient.ts
···
73
73
import { spawn, catchAbortError, isAbortError, rethrowAbortError, delay, forever, AbortError, throwIfAborted } from 'abort-controller-x';
74
74
import { Queue, MemoryStorage } from '@platformatic/job-queue'
75
75
import { DrizzlePlayRepository, playToRepositoryCreatePlayOpts } from "../common/database/drizzle/repositories/PlayRepository.js";
76
-
import { PlaySelect, QueueStateSelect } from "../common/database/drizzle/drizzleTypes.js";
77
-
import { GenericRepository } from "../common/database/drizzle/repositories/BaseRepository.js";
76
+
import { PlaySelect, QueueStateNew, QueueStateSelect } from "../common/database/drizzle/drizzleTypes.js";
78
77
import { asPlay } from "../../core/PlayMarshalUtils.js";
78
+
import { DrizzleQueueRepository } from "../common/database/drizzle/repositories/QueueRepository.js";
79
79
80
80
type PlatformMappedPlays = Map<string, {player: SourcePlayerObj, source: SourceIdentifier}>;
81
81
type NowPlayingQueue = Map<string, PlatformMappedPlays>;
···
113
113
queuedLength: number = 0;
114
114
deadLetterScrobbles: DeadLetterScrobble<PlayObject>[] = [];
115
115
deadLetterLength: number = 0;
116
+
deadLetterQueued: number = 0;
116
117
117
118
queuedConsumedSinceLastRangeRefresh: number = 0;
118
119
···
152
153
declare protected componentType: 'client';
153
154
154
155
protected playRepo: DrizzlePlayRepository;
155
-
protected queueRepo: GenericRepository<'queueStates'>;
156
+
protected queueRepo: DrizzleQueueRepository;
156
157
157
158
constructor(type: any, name: any, config: CommonClientConfig, notifier: Notifiers, emitter: EventEmitter, logger: Logger) {
158
159
super(config);
···
171
172
concurrency: 1
172
173
});
173
174
this.playRepo = new DrizzlePlayRepository(this.db, {logger: this.logger});
174
-
this.queueRepo = new GenericRepository(this.db, 'queueStates', 'Queue', {logger: this.logger});
175
+
this.queueRepo = new DrizzleQueueRepository(this.db, {logger: this.logger});
175
176
176
177
const {
177
178
options: {
···
247
248
248
249
protected async postDatabase(): Promise<void> {
249
250
this.tracksScrobbled = this.dbComponent.countLive;
250
-
this.queuedLength = await this.playRepo.getQueueCount(this.dbComponent.id, ['queued']);
251
+
this.queuedLength = await this.queueRepo.getQueueCount(this.dbComponent.id, [CLIENT_INGRESS_QUEUE]);
251
252
this.queuedGauge.labels(this.getPrometheusLabels()).set(this.queuedLength);
252
-
this.deadLetterLength = await this.playRepo.getQueueCount(this.dbComponent.id, [CLIENT_DEAD_QUEUE]);
253
+
this.deadLetterLength = await this.queueRepo.getQueueCount(this.dbComponent.id, [CLIENT_DEAD_QUEUE], ['queued', 'failed']);
254
+
this.deadLetterQueued = await this.queueRepo.getQueueCount(this.dbComponent.id, [CLIENT_DEAD_QUEUE], ['queued']);
255
+
// TODO
253
256
this.deadLetterGauge.labels(this.getPrometheusLabels()).set(this.deadLetterLength);
254
257
}
255
258
···
750
753
}
751
754
while (true) {
752
755
signal.throwIfAborted();
753
-
let queueEmpty = this.queuedScrobbles.length === 0;
754
-
while (this.queuedScrobbles.length > 0) {
756
+
let queueEmpty = this.queuedLength; // this.queuedScrobbles.length === 0;
757
+
while (this.queuedLength > 0) {
755
758
await this.processQueueCurrentScrobble(signal);
756
759
}
757
760
if(!queueEmpty) {
···
789
792
let historicalPlays: PlayObject[] = [];
790
793
let historicalError: Error | undefined;
791
794
let queueError: Error | undefined;
795
+
let successState: PlaySelect['state'];
792
796
793
797
try {
794
798
···
869
873
throw showStoppingError;
870
874
}
871
875
}
876
+
} else {
877
+
successState = 'duped';
872
878
}
873
879
}
874
880
//this.updateQueuedScrobblesCache();
···
890
896
await this.playRepo.updateById(currQueuedPlay.id, {state: 'failed', error: queueError});
891
897
} else {
892
898
await this.queueRepo.updateById(queueState.id, {queueStatus: 'completed'});
893
-
await this.playRepo.updateById(currQueuedPlay.id, {state: 'scrobbled'});
899
+
await this.playRepo.updateById(currQueuedPlay.id, {state: successState ?? 'scrobbled'});
894
900
}
895
901
this.emitEvent('scrobbleDequeued', { queuedScrobble: currQueuedPlay })
896
902
this.queuedGauge.labels(this.getPrometheusLabels()).dec();
···
900
906
901
907
processDeadLetterQueue = async (attemptWithRetries?: number) => {
902
908
903
-
if (this.deadLetterScrobbles.length === 0) {
909
+
// if (this.deadLetterScrobbles.length === 0) {
910
+
// return;
911
+
// }
912
+
913
+
if (!(await this.isReady())) {
914
+
this.deadLogger.warn('Cannot process dead letter scrobbles because client is not ready.');
904
915
return;
905
916
}
906
917
···
912
923
913
924
const retries = attemptWithRetries ?? deadLetterRetries;
914
925
915
-
const processable = this.deadLetterScrobbles.filter(x => x.retries < retries);
916
-
const queueStatus = `${processable.length} of ${this.deadLetterScrobbles.length} dead scrobbles have less than ${retries} retries, ${processable.length === 0 ? 'will skip processing.': 'processing now...'}`;
917
-
if (processable.length === 0) {
926
+
await this.queueRepo.deadFailedToQueue(this.dbComponent.id, retries);
927
+
928
+
const processable = await this.queueRepo.getQueueCount(this.dbComponent.id, [CLIENT_DEAD_QUEUE]); //this.deadLetterScrobbles.filter(x => x.retries < retries);
929
+
this.deadLetterQueued = processable;
930
+
931
+
const total = await this.queueRepo.getQueueCount(this.dbComponent.id, [CLIENT_DEAD_QUEUE], ['queued','failed']);
932
+
this.deadLetterLength = total;
933
+
const queueStatus = `${processable} of ${total} dead scrobbles have less than ${retries} retries, ${processable === 0 ? 'will skip processing.': 'processing now...'}`;
934
+
if (processable === 0) {
918
935
this.deadLogger.verbose(queueStatus);
919
936
return;
920
937
}
···
922
939
if(!this.upstreamRefresh.refreshEnabled) {
923
940
this.deadLogger.verbose('Scrobble refresh is DISABLED. All dead scrobbles will likely always be scrobbled (nothing to check duplicates against).');
924
941
}
925
-
await this.handleQueuedScrobbleRanges();
942
+
// await this.handleQueuedScrobbleRanges();
926
943
927
944
const removedIds = [];
928
-
for (const deadScrobble of processable) {
929
-
const [scrobbled, dead] = await this.processDeadLetterScrobble(deadScrobble.id);
930
-
if (scrobbled) {
931
-
removedIds.push(deadScrobble.id);
932
-
}
945
+
while(this.deadLetterQueued > 0) {
946
+
const [scrobbled, dead] = await this.processDeadLetterScrobble();
933
947
await sleep(this.scrobbleSleep);
948
+
//removedIds.push(deadScrobble.id);
934
949
}
935
950
if (removedIds.length > 0) {
936
951
this.deadLogger.info(`Removed ${removedIds.length} scrobbles from dead letter queue`);
937
952
}
938
953
}
939
954
940
-
processDeadLetterScrobble = async (id: string): Promise<[boolean, DeadLetterScrobble<PlayObject>?]> => {
941
-
const deadScrobbleIndex = this.deadLetterScrobbles.findIndex(x => x.id === id);
942
-
if(deadScrobbleIndex === -1) {
943
-
this.deadLogger.warn(`Could not find a dead scrobble with id ${id}`);
944
-
return [false];
955
+
processDeadLetterScrobble = async (uid?: string): Promise<[boolean, (PlaySelect & {queueStates: QueueStateSelect[]})?]> => {
956
+
// const deadScrobbleIndex = this.deadLetterScrobbles.findIndex(x => x.id === id);
957
+
// if(deadScrobbleIndex === -1) {
958
+
// this.deadLogger.warn(`Could not find a dead scrobble with id ${id}`);
959
+
// return [false];
960
+
// }
961
+
962
+
let deadScrobble: PlaySelect & {queueStates: QueueStateSelect[]};
963
+
let deadQueueState: QueueStateSelect;
964
+
if(uid !== undefined) {
965
+
deadScrobble = await this.playRepo.findByUid(this.dbComponent.id, uid, {hydrate: ['asPlay']});
966
+
if(deadScrobble === undefined) {
967
+
throw new Error(`Play ${uid} does not exist for ${this.name}`);
968
+
}
969
+
} else {
970
+
deadScrobble = await this.playRepo.getQueueNext(this.dbComponent.id, CLIENT_DEAD_QUEUE)
945
971
}
946
-
const deadLabel = {labels: id};
947
-
const deadScrobble = this.deadLetterScrobbles[deadScrobbleIndex];
972
+
deadQueueState = deadScrobble.queueStates.find(x => x.queueName === CLIENT_DEAD_QUEUE && x.queueStatus === 'queued');
973
+
if(deadQueueState === undefined) {
974
+
throw new Error(`Play ${uid} is not currently queued in dead letter.`);
975
+
}
976
+
//const deadScrobble = await this.playRepo.getQueueNext(this.dbComponent.id, CLIENT_INGRESS_QUEUE);
977
+
const deadLabel = {labels: deadScrobble.uid};
978
+
//const deadScrobble = this.deadLetterScrobbles[deadScrobbleIndex];
948
979
this.deadLogger.trace(deadLabel, `Processing dead scrobble => ${buildTrackString(deadScrobble.play)}`);
980
+
981
+
await this.handleQueuedScrobbleRanges();
949
982
950
983
if (!(await this.isReady())) {
951
984
this.deadLogger.warn(deadLabel, 'Cannot process dead letter scrobble because client is not ready.');
···
960
993
this.deadLogger.warn(deadLabel, `Previous error while getting historical scrobbles means this scrobble cannot be compared`);
961
994
this.deadLogger.trace(e);
962
995
} else {
963
-
this.deadLogger.warn(new SimpleError(`${id} - ${buildTrackString(deadScrobble.play)} from Source '${deadScrobble.source}' => cannot get historical scrobbles`, {cause: e, shortStack: true}));
996
+
this.deadLogger.warn(new SimpleError(`${deadScrobble.uid} - ${buildTrackString(deadScrobble.play)} from Source '${deadScrobble.play.meta.source}' => cannot get historical scrobbles`, {cause: e, shortStack: true}));
964
997
}
965
-
deadScrobble.retries++;
966
-
deadScrobble.error = messageWithCauses(e);
967
-
deadScrobble.lastRetry = dayjs();
968
-
this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble;
969
-
this.updateDeadLetterCache();
998
+
999
+
this.queueRepo.updateById(deadQueueState.id, {retries: deadQueueState.retries + 1, error: e, updatedAt: dayjs(), queueStatus: 'failed'});
1000
+
this.playRepo.updateById(deadScrobble.id, {error: e});
1001
+
// deadScrobble.retries++;
1002
+
// deadScrobble.error = messageWithCauses(e);
1003
+
// deadScrobble.lastRetry = dayjs();
1004
+
// this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble;
1005
+
// this.updateDeadLetterCache();
970
1006
this.emitEvent('updateDeadLetter', {dead: deadScrobble});
971
1007
return [false, deadScrobble];
972
1008
}
···
988
1024
try {
989
1025
const scrobbledPlay = await this.scrobble(transformedScrobble);
990
1026
await this.addScrobbledTrack(transformedScrobble, scrobbledPlay);
991
-
this.removeDeadLetterScrobble(deadScrobble.id)
1027
+
this.removeDeadLetterScrobble(deadScrobble, 'scrobbled', true);
992
1028
} catch (e) {
993
1029
994
1030
const submitError = findCauseByReference(e, ScrobbleSubmitError);
···
1001
1037
deadScrobble.play.meta.lifecycle.scrobble.error = serializeError(e);
1002
1038
}
1003
1039
1004
-
deadScrobble.retries++;
1005
-
deadScrobble.error = messageWithCauses(e);
1006
-
deadScrobble.lastRetry = dayjs();
1007
-
this.deadLogger.error(new Error(`${id} - Could not scrobble ${buildTrackString(transformedScrobble)} from Source '${deadScrobble.source}' due to error`, {cause: e}));
1008
-
this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble;
1040
+
this.queueRepo.updateById(deadQueueState.id, {retries: deadQueueState.retries + 1, error: e, updatedAt: dayjs(), queueStatus: 'failed'});
1041
+
this.playRepo.updateById(deadScrobble.id, {error: e});
1042
+
// deadScrobble.retries++;
1043
+
// deadScrobble.error = messageWithCauses(e);
1044
+
// deadScrobble.lastRetry = dayjs();
1045
+
this.deadLogger.error(new Error(`${deadScrobble.uid} - Could not scrobble ${buildTrackString(transformedScrobble)} from Source '${deadScrobble.play.meta.source}' due to error`, {cause: e}));
1046
+
//this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble;
1009
1047
this.updateDeadLetterCache();
1010
1048
this.emitEvent('updateDeadLetter', {dead: deadScrobble});
1011
1049
return [false, deadScrobble];
1012
1050
}
1013
1051
} else {
1014
1052
this.deadLogger.verbose(`Looks like ${buildTrackString(deadScrobble.play)} was already scrobbled!\n${summary}`);
1015
-
this.removeDeadLetterScrobble(deadScrobble.id)
1053
+
this.removeDeadLetterScrobble(deadScrobble, 'duped', true);
1016
1054
}
1017
1055
1018
1056
return [true, deadScrobble];
1019
1057
}
1020
1058
1021
-
removeDeadLetterScrobble = (id: string) => {
1022
-
const index = this.deadLetterScrobbles.findIndex(x => x.id === id);
1023
-
if (index === -1) {
1024
-
this.deadLogger.warn(`No scrobble found with ID ${id}`);
1025
-
return;
1059
+
removeDeadLetterScrobble = async (dead: (PlaySelect & {queueStates: QueueStateSelect[]}) | string, state: PlaySelect['state'], success: boolean) => {
1060
+
1061
+
let deadScrobble: PlaySelect & {queueStates: QueueStateSelect[]};
1062
+
1063
+
if(typeof dead === 'string'){
1064
+
deadScrobble = await this.playRepo.findByUid(this.dbComponent.id, dead, {hydrate: ['asPlay']});
1065
+
if(deadScrobble === undefined) {
1066
+
throw new Error(`Play ${dead} does not exist for ${this.name}`);
1026
1067
}
1027
-
this.deadLogger.info({labels: id}, `Removed scrobble ${buildTrackString(this.deadLetterScrobbles[index].play)} from queue`);
1028
-
this.deadLetterScrobbles.splice(index, 1);
1029
-
this.deadLetterGauge.labels(this.getPrometheusLabels()).set(this.deadLetterScrobbles.length);
1030
-
this.updateDeadLetterCache();
1031
-
this.emitEvent('removeDeadLetter', { dead: { id } });
1068
+
} else {
1069
+
deadScrobble = dead;
1070
+
}
1071
+
// const index = this.deadLetterScrobbles.findIndex(x => x.id === id);
1072
+
// if (index === -1) {
1073
+
// this.deadLogger.warn(`No scrobble found with ID ${id}`);
1074
+
// return;
1075
+
// }
1076
+
const deadQueueState = deadScrobble.queueStates.find(x => x.queueName === CLIENT_DEAD_QUEUE && x.queueStatus === 'queued');
1077
+
if(deadQueueState === undefined) {
1078
+
throw new Error(`Play ${deadScrobble.uid} is not currently queued in dead letter.`);
1079
+
}
1080
+
//this.deadLetterScrobbles.splice(index, 1);
1081
+
this.deadLetterGauge.labels(this.getPrometheusLabels()).dec();
1082
+
let queueUpdate: Partial<QueueStateNew> = {
1083
+
updatedAt: dayjs(),
1084
+
queueStatus: 'completed'
1085
+
}
1086
+
if(success) {
1087
+
queueUpdate.error = null;
1088
+
}
1089
+
await this.queueRepo.updateById(deadQueueState.id, queueUpdate);
1090
+
await this.playRepo.updateById(deadScrobble.id, {state, error: success ? null : undefined});
1091
+
this.deadLogger.info({labels: deadScrobble.uid}, `Scrobble ${buildTrackString(deadScrobble.play)} marked as completed`);
1092
+
this.deadLetterLength -= 1;
1093
+
this.deadLetterQueued -= 1;
1094
+
if(state === 'scrobbled') {
1095
+
this.componentRepo.updateById(this.dbComponent.id, {countLive: this.dbComponent.countLive + 1});
1096
+
}
1097
+
//this.updateDeadLetterCache();
1098
+
this.emitEvent('removeDeadLetter', { dead: { id: deadScrobble.uid } });
1032
1099
}
1033
1100
1034
1101
removeDeadLetterScrobbles = () => {
1035
-
this.deadLetterScrobbles = [];
1036
-
this.updateDeadLetterCache();
1037
-
this.deadLetterGauge.labels(this.getPrometheusLabels()).set(this.deadLetterScrobbles.length);
1038
-
this.logger.info('Removed all scrobbles from queue', {leaf: 'Dead Letter'});
1102
+
this.queueRepo.failedQueueToCompleted(this.dbComponent.id);
1103
+
//this.deadLetterScrobbles = [];
1104
+
this.deadLetterQueued = 0;
1105
+
//this.updateDeadLetterCache();
1106
+
this.deadLetterGauge.labels(this.getPrometheusLabels()).set(this.deadLetterQueued);
1107
+
this.deadLogger.info('Removed all scrobbles from queue');
1039
1108
}
1040
1109
1041
1110
queueScrobble = async (data: PlayObject | PlayObject[], source: string) => {
···
1136
1205
e = new Error(error);
1137
1206
}
1138
1207
this.deadLetterLength += 1;
1208
+
this.deadLetterQueued += 1;
1139
1209
//this.playRepo.updateById(data.id, {state: 'failed', error: e});
1140
1210
await this.queueRepo.create({
1141
1211
componentId: this.dbComponent.id,