feat: Transfer all data between services

package.json

@@ -125,6 +125,7 @@
     "tsx": "^4.7.0",
     "typeson": "^9.0.4",
     "typeson-registry": "^11.1.1",
+    "undici": "^5.29.0",
     "vite-express": "^0.16.0",
     "vlc-client": "^1.1.1",
     "xml2js": "0.6.1",

src/backend/common/infrastructure/Atomic.ts

@@ -1,6 +1,6 @@
 import { Logger } from '@foxxmd/logging';
 import { SearchAndReplaceRegExp } from "@foxxmd/regex-buddy-core";
-import { Dayjs } from "dayjs";
+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';
@@ -385,3 +385,55 @@ export interface CacheConfigOptions {
     auth?: CacheAuthConfig;
 }
 
+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/infrastructure/typings/lastfm-node-client.d.ts

@@ -90,11 +90,21 @@ declare module 'lastfm-node-client' {
         user: string
         limit?: number
         extended?: boolean
+        page?: number
+        from?: number
+        to?: number
     }
 
     export interface UserGetRecentTracksResponse {
         recenttracks: {
             track: TrackObject[]
+            '@attr'?: {
+                user: string
+                totalPages: string
+                page: string
+                perPage: string
+                total: string
+            }
         }
     }
 

src/backend/common/vendor/LastfmApiClient.ts

@@ -3,7 +3,8 @@ import LastFm, {
     AuthGetSessionResponse,
     LastfmTrackUpdateRequest, TrackObject,
     TrackScrobblePayload,
-    UserGetInfoResponse
+    UserGetInfoResponse,
+    UserGetRecentTracksResponse
 } from "lastfm-node-client";
 import { PlayObject } from "../../../core/Atomic.js";
 import { nonEmptyStringOrDefault, splitByFirstFound } from "../../../core/StringUtils.js";
@@ -204,6 +205,25 @@ export default class LastfmApiClient extends AbstractApiClient {
             throw e;
         }
     }
+
+    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
+        }));
+    }
 }
 
 export const scrobblePayloadToPlay = (obj: LastfmTrackUpdateRequest): PlayObject => {

src/backend/common/vendor/ListenbrainzApiClient.ts

@@ -263,6 +263,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, 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";
@@ -18,7 +18,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;
@@ -144,6 +144,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, 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";
@@ -16,7 +16,7 @@ import { buildTrackString } from '../../../../core/StringUtils.js';
 
 
 
-export class MalojaApiClient extends AbstractApiClient {
+export class MalojaApiClient extends AbstractApiClient implements PaginatedTimeRangeListens {
 
     declare config: MalojaData;
     url: URLData;
@@ -187,6 +187,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<[(MalojaScrobbleData | undefined), MalojaScrobbleV3ResponseData, string?]> => {
 
         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";
@@ -21,13 +22,15 @@ import { createVegaGenerator } from './utils/SchemaUtils.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 () {
@@ -105,7 +108,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');