feat(transfer): Add transfer/backfill system for copying scrobble history

Author: zbrox

docsite/static/aio.json

@@ -21,6 +21,7 @@ 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);
@@ -105,7 +106,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');

docsite/static/source.json

@@ -792,7 +792,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(this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.preCompare)))) {
                         const transformedScrobble = this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.postCompare);
                         try {
@@ -877,7 +879,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(this.transformPlay(deadScrobble.play, TRANSFORM_HOOK.preCompare)))) {
             const transformedScrobble = this.transformPlay(deadScrobble.play, TRANSFORM_HOOK.postCompare);
             try {
@@ -938,6 +942,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') {

src/backend/common/schema/aio-source.json

@@ -33,6 +33,8 @@ import { setupTautulliRoutes } from "./tautulliRoutes.js";
 import { setupWebscrobblerRoutes } from "./webscrobblerRoutes.js";
 import ScrobbleSources from "../sources/ScrobbleSources.js";
 import ScrobbleClients from "../scrobblers/ScrobbleClients.js";
+import { TransferManager } from "../transfer/TransferManager.js";
+import { TransferOptions } from "../transfer/TransferJob.js";
 
 const maxBufferSize = 300;
 const output: Record<number, FixedSizeList<LogDataPretty>> =  {};
@@ -54,7 +56,7 @@ const getLogs = (minLevel: number, limit: number = maxBufferSize, sort: 'asc' |
     return allLogs.flat(1).sort((a, b) => a.time - b.time).slice(0, limit);
 }
 
-export const setupApi = (app: ExpressWithAsync, logger: Logger, appLoggerStream: PassThrough, initialLogOutput: LogDataPretty[] = [], scrobbleSources: ScrobbleSources, scrobbleClients: ScrobbleClients) => {
+export const setupApi = (app: ExpressWithAsync, logger: Logger, appLoggerStream: PassThrough, initialLogOutput: LogDataPretty[] = [], scrobbleSources: ScrobbleSources, scrobbleClients: ScrobbleClients, transferManager: TransferManager) => {
     for(const level of Object.keys(logger.levels.labels)) {
         output[level] = new FixedSizeList<LeveledLogData>(maxBufferSize);
     }
@@ -520,6 +522,49 @@ export const setupApi = (app: ExpressWithAsync, logger: Logger, appLoggerStream:
        return res.json({version: root.get('version')});
     });
 
+    app.getAsync('/api/transfer/sources-clients', async (req, res) => {
+        try {
+            const result = transferManager.getActiveSourcesAndClients();
+            return res.json(result);
+        } catch (e) {
+            logger.error(`Error getting sources and clients: ${e.message}`);
+            return res.status(500).json({ error: e.message });
+        }
+    });
+
+    app.postAsync('/api/transfer', bodyParser.json(), async (req, res) => {
+        try {
+            const options: TransferOptions = req.body;
+            const id = await transferManager.startTransfer(options);
+            return res.json({ id });
+        } catch (e) {
+            logger.error(`Error starting transfer: ${e.message}`);
+            return res.status(400).json({ error: e.message });
+        }
+    });
+
+    app.getAsync('/api/transfer/:id?', async (req, res) => {
+        try {
+            const { id } = req.params;
+            const result = transferManager.getTransferStatus(id);
+            return res.json(result);
+        } catch (e) {
+            logger.error(`Error getting transfer status: ${e.message}`);
+            return res.status(404).json({ error: e.message });
+        }
+    });
+
+    app.deleteAsync('/api/transfer/:id', async (req, res) => {
+        try {
+            const { id } = req.params;
+            transferManager.cancelTransfer(id);
+            return res.status(200).json({ message: 'Transfer cancelled' });
+        } catch (e) {
+            logger.error(`Error cancelling transfer: ${e.message}`);
+            return res.status(400).json({ error: e.message });
+        }
+    });
+
     app.useAsync('/api/*', async (req, res) => {
         const remote = req.connection.remoteAddress;
         const proxyRemote = req.headers["x-forwarded-for"];

src/backend/common/schema/aio.json

@@ -15,11 +15,12 @@ import { getAddress } from "../utils/NetworkUtils.js";
 import { setupApi } from "./api.js";
 import ScrobbleSources from '../sources/ScrobbleSources.js';
 import ScrobbleClients from '../scrobblers/ScrobbleClients.js';
+import { TransferManager } from '../transfer/TransferManager.js';
 
 const app = addAsync(express());
 const router = Router();
 
-export const initServer = async (parentLogger: Logger, appLoggerStream: PassThrough, initialOutput: LogDataPretty[] = [], sources: ScrobbleSources, clients: ScrobbleClients) => {
+export const initServer = async (parentLogger: Logger, appLoggerStream: PassThrough, initialOutput: LogDataPretty[] = [], sources: ScrobbleSources, clients: ScrobbleClients, transferManager: TransferManager) => {
 
     const logger = childLogger(parentLogger, 'API'); // parentLogger.child({labels: ['API']}, mergeArr);
 
@@ -50,7 +51,7 @@ export const initServer = async (parentLogger: Logger, appLoggerStream: PassThro
         const local = root.get('localUrl');
         const localDefined = root.get('hasDefinedBaseUrl');
 
-        setupApi(app, logger, appLoggerStream, initialOutput, sources, clients);
+        setupApi(app, logger, appLoggerStream, initialOutput, sources, clients, transferManager);
 
         const addy = getAddress();
         const addresses: string[] = [];