feat: Transfer all data between services

src/backend/common/infrastructure/Atomic.ts

@@ -1,5 +1,6 @@
 import { Logger } from '@foxxmd/logging';
-import { Dayjs } from "dayjs";
+import { SearchAndReplaceRegExp } from "@foxxmd/regex-buddy-core";
+import { Dayjs, ManipulateType } from "dayjs";
 import { Request, Response } from "express";
 import { NextFunction, ParamsDictionary, Query } from "express-serve-static-core";
 import { FixedSizeList } from 'fixed-size-list';
@@ -399,4 +400,55 @@ export interface MusicbrainzApiConfigData {
 export const MUSICBRAINZ_URL = 'https://musicbrainz.org';
 export const MBID_VARIOUS_ARTISTS = "89ad4ac3-39f7-470e-963a-56509c546377";
 
-export type MusicBrainzSingletonMap = Map<string,MusicBrainzApi>;
+export type MusicBrainzSingletonMap = Map<string,MusicBrainzApi>;export interface PaginatedLimit {
+    /** per page max number of results to return */
+    limit?: number
+}
+
+export interface PaginatedTimeRangeOptions {
+    /** Unix timestamp */
+    from: number
+    /** Unix timestamp */
+    to: number
+}
+
+export interface PaginatedListensOptions extends PaginatedLimit {
+    page: number
+}
+
+export interface PagelessListensTimeRangeOptions extends Partial<PaginatedTimeRangeOptions>, PaginatedLimit {
+}
+
+export interface PaginatedListensTimeRangeOptions extends Partial<PaginatedTimeRangeOptions>, PaginatedListensOptions {
+}
+
+export interface PaginatedResults {
+    total?: number
+}
+
+export interface PaginatedListens {
+    getPaginatedListens(params: PaginatedListensOptions): Promise<{data: PlayObject[], meta: PaginatedListensOptions & PaginatedResults}>
+}
+
+export const hasPaginagedListens = (obj: Object): obj is PaginatedListens => {
+    return 'getPaginatedListens' in obj;
+}
+
+export interface PaginatedTimeRangeListens {
+    getPaginatedTimeRangeListens(params: PaginatedListensTimeRangeOptions): Promise<{data: PlayObject[], meta: PaginatedListensTimeRangeOptions & PaginatedResults}>
+    getPaginatedUnitOfTime(): ManipulateType;
+}
+
+export const hasPaginatedTimeRangeListens = (obj: Object): obj is PaginatedTimeRangeListens => {
+    return 'getPaginatedTimeRangeListens' in obj;
+}
+
+export interface PagelessTimeRangeListens {
+    getPagelessTimeRangeListens(params: PagelessListensTimeRangeOptions): Promise<{data: PlayObject[], meta: PagelessListensTimeRangeOptions & PaginatedResults}>
+}
+
+export const hasPagelessTimeRangeListens = (obj: Object): obj is PagelessTimeRangeListens => {
+    return 'getPaginatedTimeRangeListens' in obj;
+}
+
+export type PaginatedSource = PaginatedListens | PaginatedTimeRangeListens | PagelessTimeRangeListens;

src/backend/common/vendor/LastfmApiClient.ts

@@ -243,6 +243,25 @@ export default class LastfmApiClient extends AbstractApiClient {
         }
     }
 
+    getRecentTracksWithPagination = async (options: {
+        page?: number;
+        limit?: number;
+        from?: number;
+        to?: number;
+    } = {}) => {
+        const { page = 1, limit = 200, from, to } = options;
+
+        return await this.callApi<UserGetRecentTracksResponse>((client: any) => client.userGetRecentTracks({
+            user: this.user,
+            sk: this.client.sessionKey,
+            limit,
+            page,
+            from,
+            to,
+            extended: true
+        }));
+    }
+
     getRecentTracks = async (options: TracksFetchOptions = {}): Promise<PlayObject[]> => {
 
         let resp: LastFMUserGetRecentTracksResponse;

src/backend/common/vendor/ListenbrainzApiClient.ts

@@ -262,6 +262,39 @@ export class ListenbrainzApiClient extends AbstractApiClient {
         return playObj;
     }
 
+    getUserListensWithPagination = async (options: {
+        count?: number;
+        minTs?: number;
+        maxTs?: number;
+        user?: string;
+    } = {}): Promise<ListensResponse> => {
+        const { count = 100, minTs, maxTs, user } = options;
+
+        try {
+            const query: any = { count };
+            if (minTs !== undefined) {
+                query.min_ts = minTs;
+            }
+            if (maxTs !== undefined) {
+                query.max_ts = maxTs;
+            }
+
+            const resp = await this.callApi(request
+                .get(`${joinedUrl(this.url.url,'1/user', user ?? this.config.username, 'listens')}`)
+                .timeout({
+                    response: 15000,
+                    deadline: 30000
+                })
+                .query(query));
+
+            const {body: {payload}} = resp as any;
+            return payload as ListensResponse;
+        } catch (e) {
+            throw e;
+        }
+    }
+
+
     static formatPlayObj(obj: any, options: FormatPlayObjectOptions): PlayObject {
         return listenResponseToPlay(obj);
     }

src/backend/common/vendor/koito/KoitoApiClient.ts

@@ -1,6 +1,6 @@
 import dayjs from "dayjs";
 import { PlayObject, PlayObjectLifecycleless, ScrobbleActionResult, URLData } from "../../../../core/Atomic.js";
-import { AbstractApiOptions, DEFAULT_RETRY_MULTIPLIER } from "../../infrastructure/Atomic.js";
+import { AbstractApiOptions, DEFAULT_RETRY_MULTIPLIER, PaginatedListensTimeRangeOptions, PaginatedTimeRangeListens } from "../../infrastructure/Atomic.js";
 import { KoitoData, ListenObjectResponse, ListensResponse } from "../../infrastructure/config/client/koito.js";
 import AbstractApiClient from "../AbstractApiClient.js";
 import { getBaseFromUrl, isPortReachableConnect, joinedUrl, normalizeWebAddress } from "../../../utils/NetworkUtils.js";
@@ -20,7 +20,7 @@ interface SubmitOptions {
 
 const KOITO_LZ_PATH: RegExp = new RegExp(/^\/apis\/listenbrainz(\/?1?\/?)?$/);
 
-export class KoitoApiClient extends AbstractApiClient {
+export class KoitoApiClient extends AbstractApiClient implements PaginatedTimeRangeListens {
 
     declare config: KoitoData;
     url: URLData;
@@ -146,6 +146,83 @@ export class KoitoApiClient extends AbstractApiClient {
         }
     }
 
+    getUserListensWithPagination = async (options: {
+        count?: number;
+        minTs?: number;
+        maxTs?: number;
+    } = {}): Promise<ListensResponse> => {
+        const { count = 100, minTs, maxTs } = options;
+
+        try {
+            const query: any = { count };
+            if (minTs !== undefined) {
+                query.min_ts = minTs;
+            }
+            if (maxTs !== undefined) {
+                query.max_ts = maxTs;
+            }
+
+            const resp = await this.callApi(request
+                .get(`${joinedUrl(this.url.url, '/apis/listenbrainz/1/user', this.config.username, 'listens')}`)
+                .timeout({
+                    response: 15000,
+                    deadline: 30000
+                })
+                .query(query));
+
+            const { body: { payload } } = resp as any;
+            return payload as ListensResponse;
+        } catch (e) {
+            throw e;
+        }
+    }
+
+    getPaginatedTimeRangeListens = async (params: PaginatedListensTimeRangeOptions) => {
+        let dateData: {week?: number, month?: number, year?: number} = {};
+        if(params.from !== undefined && params.to !== undefined) {
+            const from = dayjs.unix(params.from);
+            const to = dayjs.unix(params.to);
+            if(from.week() === to.week()) {
+                from.subtract
+                dateData = {
+                    year: from.year(),
+                    month: from.month(),
+                    week: from.week()
+                }
+            } else if(from.month() === to.month()) {
+                dateData = {
+                    year: from.year(),
+                    month: from.month(),
+                }
+            } else {
+                dateData = {
+                    year: from.year()
+                } 
+            }
+        }
+        const resp = await this.callApi(request
+        .get(`${joinedUrl(this.url.url, '/apis/web/v1/listens')}`)
+        .query({
+            page: params.page,
+            limit: params.limit,
+            ...dateData
+        }));
+
+        const r = resp.body as ListensResponse;
+
+        return {
+            data: r.items.map((x => listenObjectResponseToPlay(x))),
+            meta: {
+                ...params,
+                total: r.total_record_count
+            }
+        }
+    }
+
+    getPaginatedUnitOfTime(): dayjs.ManipulateType {
+        return 'week';
+    }
+
     getRecentlyPlayed = async (maxTracks: number): Promise<PlayObject[]> => {
         try {
             const resp = await this.getUserListens(maxTracks);

src/backend/common/vendor/maloja/MalojaApiClient.ts

@@ -1,11 +1,11 @@
-import dayjs from 'dayjs';
+import dayjs, { ManipulateType } from 'dayjs';
 import request, { SuperAgentRequest, Response } from 'superagent';
 import compareVersions from "compare-versions";
 import AbstractApiClient from "../AbstractApiClient.js";
 import { getBaseFromUrl, isPortReachableConnect, joinedUrl, normalizeWebAddress } from "../../../utils/NetworkUtils.js";
 import { MalojaData } from "../../infrastructure/config/client/maloja.js";
 import { PlayObject, PlayObjectLifecycleless, ScrobbleActionResult, URLData } from "../../../../core/Atomic.js";
-import { AbstractApiOptions, DEFAULT_RETRY_MULTIPLIER, FormatPlayObjectOptions } from "../../infrastructure/Atomic.js";
+import { AbstractApiOptions, DEFAULT_RETRY_MULTIPLIER, FormatPlayObjectOptions, PaginatedListensTimeRangeOptions, PaginatedTimeRangeListens } from "../../infrastructure/Atomic.js";
 import { isNodeNetworkException } from "../../errors/NodeErrors.js";
 import { isSuperAgentResponseError } from "../../errors/ErrorUtils.js";
 import { getNonEmptyVal, parseRetryAfterSecsFromObj, removeUndefinedKeys, sleep } from "../../../utils.js";
@@ -18,7 +18,7 @@ import { ScrobbleSubmitError } from '../../errors/MSErrors.js';
 
 
 
-export class MalojaApiClient extends AbstractApiClient {
+export class MalojaApiClient extends AbstractApiClient implements PaginatedTimeRangeListens {
 
     declare config: MalojaData;
     url: URLData;
@@ -189,6 +189,24 @@ export class MalojaApiClient extends AbstractApiClient {
         return list.map(formatPlayObj);
     }
 
+    getPaginatedTimeRangeListens = async (params: PaginatedListensTimeRangeOptions) => {
+        const resp = await this.callApi(request.get(`${this.url.url}/apis/mlj_1/scrobbles`).query({
+            perpage: params.limit,
+            page: params.page,
+            from: params.from !== undefined ? dayjs.unix(params.from).format('YYYY/MM/DD') : undefined,
+            to: params.to !== undefined ? dayjs.unix(params.from).format('YYYY/MM/DD') : undefined
+        }));
+
+        return {
+            data: resp.body.list.map(formatPlayObj),
+            meta: params
+        }
+    }
+
+    getPaginatedUnitOfTime(): ManipulateType {
+        return 'day';
+    }
+
     scrobble = async (playObj: PlayObject): Promise<ScrobbleActionResult> => {
 
         const {

src/backend/index.ts

@@ -7,6 +7,7 @@ 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 week from 'dayjs/plugin/weekOfYear.js';
 import * as path from "path";
 import { SimpleIntervalJob, ToadScheduler } from "toad-scheduler";
 import { projectDir } from "./common/index.js";
@@ -22,13 +23,15 @@ import { readJson } from './utils/DataUtils.js';
 import ScrobbleClients from './scrobblers/ScrobbleClients.js';
 import ScrobbleSources from './sources/ScrobbleSources.js';
 import { Notifiers } from './notifier/Notifiers.js';
+import { TransferManager } from './transfer/TransferManager.js';
 
 dayjs.extend(utc)
 dayjs.extend(isBetween);
 dayjs.extend(relativeTime);
 dayjs.extend(duration);
 dayjs.extend(timezone);
 dayjs.extend(isToday);
+dayjs.extend(week);
 
 // eslint-disable-next-line prefer-arrow-functions/prefer-arrow-functions
 (async function () {
@@ -108,7 +111,9 @@ const configDir = process.env.CONFIG_DIR || path.resolve(projectDir, `./config`)
 
         await root.items.cache().init();
 
-        initServer(logger, appLoggerStream, output, scrobbleSources, scrobbleClients);
+        const transferManager = new TransferManager(scrobbleSources, scrobbleClients, logger);
+
+        initServer(logger, appLoggerStream, output, scrobbleSources, scrobbleClients, transferManager);
 
         if(process.env.IS_LOCAL === 'true') {
             logger.info('multi-scrobbler can be run as a background service! See: https://foxxmd.github.io/multi-scrobbler/docs/installation/service');

src/backend/scrobblers/AbstractScrobbleClient.ts

@@ -803,7 +803,9 @@ ${closestMatch.breakdowns.join('\n')}`, {leaf: ['Dupe Check']});
                     }
                     const currQueuedPlay = this.queuedScrobbles.shift();
 
-                    const [timeFrameValid, timeFrameValidLog] = this.timeFrameIsValid(currQueuedPlay.play);
+                    // Skip timeframe validation for transfers (historical data)
+                    const isTransfer = currQueuedPlay.source?.startsWith('transfer-');
+                    const [timeFrameValid, timeFrameValidLog] = isTransfer ? [true, ''] : this.timeFrameIsValid(currQueuedPlay.play);
                     if (timeFrameValid && !(await this.alreadyScrobbled(currQueuedPlay.play))) {
                         const transformedScrobble = await this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.postCompare);
                         if(transformedScrobble.meta.lifecycle === undefined) {
@@ -909,7 +911,9 @@ ${closestMatch.breakdowns.join('\n')}`, {leaf: ['Dupe Check']});
         if (this.getLatestQueuePlayDate() !== undefined && this.scrobblesLastCheckedAt().unix() < this.getLatestQueuePlayDate().unix()) {
             await this.refreshScrobbles();
         }
-        const [timeFrameValid, timeFrameValidLog] = this.timeFrameIsValid(deadScrobble.play);
+        // Skip timeframe validation for transfers (historical data)
+        const isTransfer = deadScrobble.source?.startsWith('transfer-');
+        const [timeFrameValid, timeFrameValidLog] = isTransfer ? [true, ''] : this.timeFrameIsValid(deadScrobble.play);
         if (timeFrameValid && !(await this.alreadyScrobbled(deadScrobble.play))) {
             const transformedScrobble = await this.transformPlay(deadScrobble.play, TRANSFORM_HOOK.postCompare);
             try {
@@ -986,6 +990,19 @@ ${closestMatch.breakdowns.join('\n')}`, {leaf: ['Dupe Check']});
         this.updateQueuedScrobblesCache();
     }
 
+    cancelQueuedItemsBySource = (source: string): number => {
+        const beforeMain = this.queuedScrobbles.length;
+        const beforeDead = this.deadLetterScrobbles.length;
+
+        this.queuedScrobbles = this.queuedScrobbles.filter(item => item.source !== source);
+        this.deadLetterScrobbles = this.deadLetterScrobbles.filter(item => item.source !== source);
+
+        this.updateQueuedScrobblesCache();
+        this.updateDeadLetterCache();
+
+        return (beforeMain + beforeDead) - (this.queuedScrobbles.length + this.deadLetterScrobbles.length);
+    }
+
     protected addDeadLetterScrobble = (data: QueuedScrobble<PlayObject>, error: (Error | string) = 'Unspecified error') => {
         let eString = '';
         if(typeof error === 'string') {