feat: implement immediate song document updates for perfect sync resilience
- Remove end-of-job cleanup phases from S3 sync and song matching jobs - Update Song documents immediately after each successful match in both Phase 1 and Phase 2 - Ensure hasS3File flag is set to true immediately for each matched song - Enable play buttons to appear instantly as songs are processed - Make system perfectly resilient to interruptions - no orphaned files - Allow seamless resume capability for long-running sync jobs - Provide real-time availability of matched songs without waiting for job completion - Maintain system consistency regardless of when sync gets interrupted
This commit is contained in:
Geert Rademakes
parent
96c43dbcff
commit
b436d1dabc
@ -7,6 +7,7 @@ import { playlistsRouter } from './routes/playlists.js';
|
||||
import { musicRouter } from './routes/music.js';
|
||||
import { matchingRouter } from './routes/matching.js';
|
||||
import { configRouter } from './routes/config.js';
|
||||
import backgroundJobsRouter from './routes/backgroundJobs.js';
|
||||
import { Song } from './models/Song.js';
|
||||
import { Playlist } from './models/Playlist.js';
|
||||
import { MusicFile } from './models/MusicFile.js';
|
||||
@ -67,6 +68,7 @@ app.use('/api/playlists', playlistsRouter);
|
||||
app.use('/api/music', musicRouter);
|
||||
app.use('/api/matching', matchingRouter);
|
||||
app.use('/api/config', configRouter);
|
||||
app.use('/api/background-jobs', backgroundJobsRouter);
|
||||
|
||||
app.listen(port, () => {
|
||||
console.log(`Server is running on port ${port}`);
|
||||
|
||||
78
packages/backend/src/routes/backgroundJobs.ts
Normal file
78
packages/backend/src/routes/backgroundJobs.ts
Normal file
@ -0,0 +1,78 @@
|
||||
import express from 'express';
|
||||
import { backgroundJobService } from '../services/backgroundJobService.js';
|
||||
|
||||
const router = express.Router();
|
||||
|
||||
/**
|
||||
* Start a new background job
|
||||
*/
|
||||
router.post('/start', async (req, res) => {
|
||||
try {
|
||||
const { type, options } = req.body;
|
||||
|
||||
if (!type || !['s3-sync', 'song-matching'].includes(type)) {
|
||||
return res.status(400).json({ error: 'Invalid job type. Must be "s3-sync" or "song-matching"' });
|
||||
}
|
||||
|
||||
console.log(`🚀 Starting background job: ${type}`);
|
||||
const jobId = await backgroundJobService.startJob({ type, options });
|
||||
|
||||
res.json({
|
||||
message: 'Background job started',
|
||||
jobId,
|
||||
type
|
||||
});
|
||||
} catch (error) {
|
||||
console.error('Error starting background job:', error);
|
||||
res.status(500).json({ error: 'Failed to start background job' });
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* Get job progress
|
||||
*/
|
||||
router.get('/progress/:jobId', async (req, res) => {
|
||||
try {
|
||||
const { jobId } = req.params;
|
||||
const progress = backgroundJobService.getJobProgress(jobId);
|
||||
|
||||
if (!progress) {
|
||||
return res.status(404).json({ error: 'Job not found' });
|
||||
}
|
||||
|
||||
res.json(progress);
|
||||
} catch (error) {
|
||||
console.error('Error getting job progress:', error);
|
||||
res.status(500).json({ error: 'Failed to get job progress' });
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* Get all jobs
|
||||
*/
|
||||
router.get('/jobs', async (req, res) => {
|
||||
try {
|
||||
const jobs = backgroundJobService.getAllJobs();
|
||||
res.json({ jobs });
|
||||
} catch (error) {
|
||||
console.error('Error getting jobs:', error);
|
||||
res.status(500).json({ error: 'Failed to get jobs' });
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* Clean up old jobs
|
||||
*/
|
||||
router.post('/cleanup', async (req, res) => {
|
||||
try {
|
||||
backgroundJobService.cleanupOldJobs();
|
||||
res.json({ message: 'Old jobs cleaned up successfully' });
|
||||
} catch (error) {
|
||||
console.error('Error cleaning up jobs:', error);
|
||||
res.status(500).json({ error: 'Failed to clean up jobs' });
|
||||
}
|
||||
});
|
||||
|
||||
export default router;
|
||||
@ -74,12 +74,11 @@ router.get('/suggestions', async (req, res) => {
|
||||
});
|
||||
|
||||
/**
|
||||
* Auto-match and link music files to songs
|
||||
* Auto-match and link music files to songs - now uses background job system
|
||||
*/
|
||||
router.post('/auto-link', async (req, res) => {
|
||||
try {
|
||||
console.log('🚀 Starting auto-match and link request...');
|
||||
const startTime = Date.now();
|
||||
console.log('🚀 Starting auto-match and link background job...');
|
||||
|
||||
const options = {
|
||||
minConfidence: parseFloat(req.body.minConfidence as string) || 0.7,
|
||||
@ -89,23 +88,24 @@ router.post('/auto-link', async (req, res) => {
|
||||
|
||||
console.log('⚙️ Auto-linking options:', options);
|
||||
|
||||
const result = await matchingService.autoMatchAndLink(options);
|
||||
// Import background job service
|
||||
const { backgroundJobService } = await import('../services/backgroundJobService.js');
|
||||
|
||||
const endTime = Date.now();
|
||||
const duration = ((endTime - startTime) / 1000).toFixed(2);
|
||||
|
||||
console.log(`✅ Auto-linking completed in ${duration} seconds`);
|
||||
console.log(`📊 Results: ${result.linked} linked, ${result.unmatched} unmatched`);
|
||||
// Start the background job
|
||||
const jobId = await backgroundJobService.startJob({
|
||||
type: 'song-matching',
|
||||
options
|
||||
});
|
||||
|
||||
res.json({
|
||||
message: 'Auto-linking completed',
|
||||
result,
|
||||
message: 'Auto-linking started as background job',
|
||||
jobId,
|
||||
options,
|
||||
duration: `${duration}s`
|
||||
status: 'started'
|
||||
});
|
||||
} catch (error) {
|
||||
console.error('❌ Error during auto-linking:', error);
|
||||
res.status(500).json({ error: 'Failed to auto-link music files' });
|
||||
console.error('❌ Error starting auto-linking job:', error);
|
||||
res.status(500).json({ error: 'Failed to start auto-linking job' });
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@ -166,161 +166,30 @@ router.get('/files', async (req, res) => {
|
||||
});
|
||||
|
||||
/**
|
||||
* Sync S3 files with database - recursively list all files in S3 and sync
|
||||
* Sync S3 files with database - now uses background job system
|
||||
*/
|
||||
router.post('/sync-s3', async (req, res) => {
|
||||
try {
|
||||
console.log('🔄 Starting S3 sync process...');
|
||||
const startTime = Date.now();
|
||||
console.log('🔄 Starting S3 sync background job...');
|
||||
|
||||
// Get all files from S3 recursively
|
||||
console.log('📁 Fetching files from S3 bucket...');
|
||||
const s3Files = await s3Service.listAllFiles();
|
||||
console.log(`✅ Found ${s3Files.length} files in S3 bucket`);
|
||||
// Import background job service
|
||||
const { backgroundJobService } = await import('../services/backgroundJobService.js');
|
||||
|
||||
// Pre-filter audio files to reduce processing
|
||||
const audioFiles = s3Files.filter(s3File => {
|
||||
const filename = s3File.key.split('/').pop() || s3File.key;
|
||||
return audioMetadataService.isAudioFile(filename);
|
||||
// Start the background job
|
||||
const jobId = await backgroundJobService.startJob({
|
||||
type: 's3-sync',
|
||||
options: req.body
|
||||
});
|
||||
|
||||
console.log(`🎵 Found ${audioFiles.length} audio files out of ${s3Files.length} total files`);
|
||||
|
||||
// Get all existing S3 keys from database in one query for faster lookup
|
||||
console.log('📊 Fetching existing files from database...');
|
||||
const existingFiles = await MusicFile.find({}, { s3Key: 1 });
|
||||
const existingS3Keys = new Set(existingFiles.map(f => f.s3Key));
|
||||
console.log(`📊 Found ${existingFiles.length} existing files in database`);
|
||||
|
||||
// Filter out already processed files
|
||||
const newAudioFiles = audioFiles.filter(s3File => !existingS3Keys.has(s3File.key));
|
||||
console.log(`🆕 Found ${newAudioFiles.length} new audio files to process`);
|
||||
|
||||
const results = {
|
||||
total: s3Files.length,
|
||||
audioFiles: audioFiles.length,
|
||||
newAudioFiles: newAudioFiles.length,
|
||||
synced: 0,
|
||||
skipped: s3Files.length - newAudioFiles.length,
|
||||
errors: 0,
|
||||
newFiles: 0,
|
||||
nonAudioFiles: s3Files.length - audioFiles.length
|
||||
};
|
||||
|
||||
if (newAudioFiles.length === 0) {
|
||||
console.log('✅ No new files to sync');
|
||||
const endTime = Date.now();
|
||||
const duration = ((endTime - startTime) / 1000).toFixed(2);
|
||||
|
||||
res.json({
|
||||
message: 'S3 sync completed - no new files',
|
||||
results,
|
||||
duration: `${duration}s`
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
console.log('🔍 Processing new audio files...');
|
||||
let processedCount = 0;
|
||||
const batchSize = 10; // Process in batches for better performance
|
||||
const musicFilesToSave = [];
|
||||
|
||||
for (const s3File of newAudioFiles) {
|
||||
processedCount++;
|
||||
const progress = ((processedCount / newAudioFiles.length) * 100).toFixed(1);
|
||||
|
||||
try {
|
||||
console.log(`📄 [${progress}%] Processing: ${s3File.key} (${audioMetadataService.formatFileSize(s3File.size)})`);
|
||||
|
||||
// Extract filename from S3 key
|
||||
const filename = s3File.key.split('/').pop() || s3File.key;
|
||||
console.log(`🎵 Processing audio file: ${filename}`);
|
||||
|
||||
// Get file content to extract metadata
|
||||
try {
|
||||
console.log(`⬇️ Downloading file content: ${s3File.key}`);
|
||||
const fileBuffer = await s3Service.getFileContent(s3File.key);
|
||||
console.log(`📊 Extracting metadata: ${filename}`);
|
||||
const metadata = await audioMetadataService.extractMetadata(fileBuffer, filename);
|
||||
|
||||
// Create music file object (don't save yet, batch save later)
|
||||
const musicFile = new MusicFile({
|
||||
originalName: filename,
|
||||
s3Key: s3File.key,
|
||||
s3Url: `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
contentType: 'audio/mpeg', // Default, will be updated by metadata
|
||||
size: s3File.size,
|
||||
...metadata,
|
||||
});
|
||||
|
||||
musicFilesToSave.push(musicFile);
|
||||
results.synced++;
|
||||
results.newFiles++;
|
||||
console.log(`✅ Successfully processed: ${filename}`);
|
||||
|
||||
} catch (metadataError) {
|
||||
console.error(`❌ Error extracting metadata for ${s3File.key}:`, metadataError);
|
||||
console.log(`🔄 Saving file without metadata: ${filename}`);
|
||||
|
||||
// Still save the file without metadata
|
||||
const musicFile = new MusicFile({
|
||||
originalName: filename,
|
||||
s3Key: s3File.key,
|
||||
s3Url: `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
contentType: 'audio/mpeg',
|
||||
size: s3File.size,
|
||||
});
|
||||
|
||||
musicFilesToSave.push(musicFile);
|
||||
results.synced++;
|
||||
results.newFiles++;
|
||||
console.log(`✅ Processed without metadata: ${filename}`);
|
||||
}
|
||||
|
||||
// Batch save every batchSize files for better performance
|
||||
if (musicFilesToSave.length >= batchSize) {
|
||||
console.log(`💾 Batch saving ${musicFilesToSave.length} files to database...`);
|
||||
await MusicFile.insertMany(musicFilesToSave);
|
||||
musicFilesToSave.length = 0; // Clear the array
|
||||
}
|
||||
|
||||
} catch (error) {
|
||||
console.error(`❌ Error processing ${s3File.key}:`, error);
|
||||
results.errors++;
|
||||
}
|
||||
}
|
||||
|
||||
// Save any remaining files
|
||||
if (musicFilesToSave.length > 0) {
|
||||
console.log(`💾 Saving final ${musicFilesToSave.length} files to database...`);
|
||||
await MusicFile.insertMany(musicFilesToSave);
|
||||
}
|
||||
|
||||
const endTime = Date.now();
|
||||
const duration = ((endTime - startTime) / 1000).toFixed(2);
|
||||
|
||||
console.log('🎉 S3 sync completed!');
|
||||
console.log(`⏱️ Duration: ${duration} seconds`);
|
||||
console.log(`📊 Results:`, results);
|
||||
console.log(` Total files: ${results.total}`);
|
||||
console.log(` Audio files: ${results.audioFiles}`);
|
||||
console.log(` Non-audio files: ${results.nonAudioFiles}`);
|
||||
console.log(` New files to process: ${results.newAudioFiles}`);
|
||||
console.log(` New files synced: ${results.newFiles}`);
|
||||
console.log(` Files skipped: ${results.skipped}`);
|
||||
console.log(` Errors: ${results.errors}`);
|
||||
console.log(` Processing speed: ${(results.newFiles / parseFloat(duration)).toFixed(1)} files/second`);
|
||||
|
||||
res.json({
|
||||
message: 'S3 sync completed',
|
||||
results,
|
||||
duration: `${duration}s`,
|
||||
processingSpeed: `${(results.newFiles / parseFloat(duration)).toFixed(1)} files/second`
|
||||
message: 'S3 sync started as background job',
|
||||
jobId,
|
||||
status: 'started'
|
||||
});
|
||||
|
||||
} catch (error) {
|
||||
console.error('❌ S3 sync error:', error);
|
||||
res.status(500).json({ error: 'Failed to sync S3 files' });
|
||||
console.error('❌ Error starting S3 sync job:', error);
|
||||
res.status(500).json({ error: 'Failed to start S3 sync job' });
|
||||
}
|
||||
});
|
||||
|
||||
@ -496,4 +365,54 @@ router.post('/:id/link-song/:songId', async (req, res) => {
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* Fix orphaned music files (MusicFile exists but Song doesn't have hasS3File: true)
|
||||
*/
|
||||
router.post('/fix-orphaned', async (req, res) => {
|
||||
try {
|
||||
console.log('🔧 Starting orphaned music files fix...');
|
||||
|
||||
const orphanedMusicFiles = await MusicFile.find({
|
||||
songId: { $exists: true },
|
||||
s3Key: { $exists: true }
|
||||
});
|
||||
|
||||
let fixedCount = 0;
|
||||
const fixedFiles = [];
|
||||
|
||||
for (const musicFile of orphanedMusicFiles) {
|
||||
// Check if the corresponding Song document needs to be updated
|
||||
const song = await Song.findById(musicFile.songId);
|
||||
if (song && !song.s3File?.hasS3File) {
|
||||
await Song.updateOne(
|
||||
{ _id: musicFile.songId },
|
||||
{
|
||||
$set: {
|
||||
's3File.musicFileId': musicFile._id,
|
||||
's3File.s3Key': musicFile.s3Key,
|
||||
's3File.s3Url': musicFile.s3Url,
|
||||
's3File.streamingUrl': musicFile.s3Url,
|
||||
's3File.hasS3File': true
|
||||
}
|
||||
}
|
||||
);
|
||||
fixedCount++;
|
||||
fixedFiles.push(musicFile.originalName);
|
||||
console.log(`🔧 Fixed orphaned music file: ${musicFile.originalName}`);
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`✅ Orphaned music files fix completed: Fixed ${fixedCount} files`);
|
||||
|
||||
res.json({
|
||||
message: `Fixed ${fixedCount} orphaned music files`,
|
||||
fixedCount,
|
||||
fixedFiles
|
||||
});
|
||||
} catch (error) {
|
||||
console.error('Error fixing orphaned music files:', error);
|
||||
res.status(500).json({ message: 'Error fixing orphaned music files', error });
|
||||
}
|
||||
});
|
||||
|
||||
export { router as musicRouter };
|
||||
@ -20,18 +20,31 @@ router.get('/structure', async (req: Request, res: Response) => {
|
||||
try {
|
||||
const playlists = await Playlist.find({});
|
||||
|
||||
// Remove track data from playlists to reduce payload size
|
||||
// Keep track counts but remove full track data to reduce payload size
|
||||
const structureOnly = playlists.map(playlist => {
|
||||
const cleanPlaylist = playlist.toObject() as any;
|
||||
|
||||
// Remove tracks from the main playlist
|
||||
// Replace tracks array with track count for the main playlist
|
||||
if (cleanPlaylist.tracks) {
|
||||
cleanPlaylist.trackCount = cleanPlaylist.tracks.length;
|
||||
delete cleanPlaylist.tracks;
|
||||
} else {
|
||||
cleanPlaylist.trackCount = 0;
|
||||
}
|
||||
|
||||
// Recursively remove tracks from children
|
||||
// Recursively process children
|
||||
const cleanChildren = (children: any[]): any[] => {
|
||||
return children.map(child => {
|
||||
const cleanChild = { ...child } as any;
|
||||
|
||||
// Replace tracks array with track count
|
||||
if (cleanChild.tracks) {
|
||||
cleanChild.trackCount = cleanChild.tracks.length;
|
||||
delete cleanChild.tracks;
|
||||
} else {
|
||||
cleanChild.trackCount = 0;
|
||||
}
|
||||
|
||||
if (cleanChild.children && cleanChild.children.length > 0) {
|
||||
cleanChild.children = cleanChildren(cleanChild.children);
|
||||
}
|
||||
|
||||
@ -54,6 +54,51 @@ export class AudioMetadataService {
|
||||
return container.toUpperCase();
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract basic metadata from filename (fallback method)
|
||||
*/
|
||||
extractBasicMetadataFromFilename(fileName: string): AudioMetadata {
|
||||
console.log(`🔄 Using filename-based metadata extraction for: ${fileName}`);
|
||||
|
||||
// Remove file extension
|
||||
const nameWithoutExt = fileName.replace(/\.[^/.]+$/, '');
|
||||
|
||||
// Try to extract artist and title from common patterns
|
||||
let title = nameWithoutExt;
|
||||
let artist: string | undefined;
|
||||
|
||||
// Common patterns: "Artist - Title", "Artist feat. Title", etc.
|
||||
const patterns = [
|
||||
/^(.+?)\s*[-–—]\s*(.+)$/, // Artist - Title
|
||||
/^(.+?)\s+feat\.?\s+(.+)$/i, // Artist feat. Title
|
||||
/^(.+?)\s+ft\.?\s+(.+)$/i, // Artist ft. Title
|
||||
/^(.+?)\s+featuring\s+(.+)$/i, // Artist featuring Title
|
||||
/^(.+?)\s*&\s*(.+)$/, // Artist & Title
|
||||
/^(.+?)\s+vs\s+(.+)$/i, // Artist vs Title
|
||||
/^(.+?)\s+x\s+(.+)$/i, // Artist x Title
|
||||
];
|
||||
|
||||
for (const pattern of patterns) {
|
||||
const match = nameWithoutExt.match(pattern);
|
||||
if (match) {
|
||||
artist = match[1].trim();
|
||||
title = match[2].trim();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Determine format from extension
|
||||
const extension = fileName.split('.').pop()?.toLowerCase();
|
||||
const format = extension ? this.mapFormatToDisplayName(extension, fileName) : 'UNKNOWN';
|
||||
|
||||
return {
|
||||
title,
|
||||
artist,
|
||||
format,
|
||||
size: 0, // Will be set by caller
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate and sanitize numeric values
|
||||
*/
|
||||
@ -99,7 +144,7 @@ export class AudioMetadataService {
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract metadata from audio file buffer
|
||||
* Extract metadata from audio file buffer with improved error handling
|
||||
*/
|
||||
async extractMetadata(fileBuffer: Buffer, fileName: string): Promise<AudioMetadata> {
|
||||
console.log(`🎵 Extracting metadata for: ${fileName} (${this.formatFileSize(fileBuffer.length)})`);
|
||||
@ -141,18 +186,16 @@ export class AudioMetadataService {
|
||||
} catch (error) {
|
||||
console.error(`❌ Error extracting metadata for ${fileName}:`, error);
|
||||
|
||||
// Try to determine format from file extension as fallback
|
||||
const extension = fileName.split('.').pop()?.toLowerCase();
|
||||
const fallbackFormat = extension ? this.mapFormatToDisplayName(extension, fileName) : 'UNKNOWN';
|
||||
// Use filename-based extraction as fallback
|
||||
const fallbackMetadata = this.extractBasicMetadataFromFilename(fileName);
|
||||
fallbackMetadata.size = fileBuffer.length;
|
||||
|
||||
console.log(`🔄 Using fallback metadata for ${fileName}`);
|
||||
console.log(`✅ Fallback metadata created for ${fileName}:`);
|
||||
console.log(` Title: ${fallbackMetadata.title}`);
|
||||
console.log(` Artist: ${fallbackMetadata.artist || 'Unknown'}`);
|
||||
console.log(` Format: ${fallbackMetadata.format}`);
|
||||
|
||||
// Return basic metadata if extraction fails
|
||||
return {
|
||||
title: fileName.replace(/\.[^/.]+$/, ''), // Remove file extension
|
||||
format: fallbackFormat,
|
||||
size: fileBuffer.length,
|
||||
};
|
||||
return fallbackMetadata;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
593
packages/backend/src/services/backgroundJobService.ts
Normal file
593
packages/backend/src/services/backgroundJobService.ts
Normal file
@ -0,0 +1,593 @@
|
||||
export interface JobProgress {
|
||||
jobId: string;
|
||||
type: 's3-sync' | 'song-matching';
|
||||
status: 'running' | 'completed' | 'failed';
|
||||
progress: number; // 0-100
|
||||
current: number;
|
||||
total: number;
|
||||
message: string;
|
||||
startTime: Date;
|
||||
endTime?: Date;
|
||||
result?: any;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export interface JobOptions {
|
||||
type: 's3-sync' | 'song-matching';
|
||||
options?: any;
|
||||
}
|
||||
|
||||
class BackgroundJobService {
|
||||
private jobs: Map<string, JobProgress> = new Map();
|
||||
private jobIdCounter = 0;
|
||||
|
||||
/**
|
||||
* Start a new background job
|
||||
*/
|
||||
async startJob(jobOptions: JobOptions): Promise<string> {
|
||||
const jobId = `job_${++this.jobIdCounter}_${Date.now()}`;
|
||||
|
||||
const job: JobProgress = {
|
||||
jobId,
|
||||
type: jobOptions.type,
|
||||
status: 'running',
|
||||
progress: 0,
|
||||
current: 0,
|
||||
total: 0,
|
||||
message: `Starting ${jobOptions.type}...`,
|
||||
startTime: new Date(),
|
||||
};
|
||||
|
||||
this.jobs.set(jobId, job);
|
||||
|
||||
// Start the job in the background
|
||||
this.runJob(jobId, jobOptions).catch(error => {
|
||||
console.error(`Job ${jobId} failed:`, error);
|
||||
const job = this.jobs.get(jobId);
|
||||
if (job) {
|
||||
job.status = 'failed';
|
||||
job.error = error.message;
|
||||
job.endTime = new Date();
|
||||
}
|
||||
});
|
||||
|
||||
return jobId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get job progress
|
||||
*/
|
||||
getJobProgress(jobId: string): JobProgress | null {
|
||||
return this.jobs.get(jobId) || null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all jobs
|
||||
*/
|
||||
getAllJobs(): JobProgress[] {
|
||||
return Array.from(this.jobs.values());
|
||||
}
|
||||
|
||||
/**
|
||||
* Update job progress
|
||||
*/
|
||||
updateProgress(jobId: string, progress: Partial<JobProgress>): void {
|
||||
const job = this.jobs.get(jobId);
|
||||
if (job) {
|
||||
Object.assign(job, progress);
|
||||
|
||||
// Calculate percentage if current and total are provided
|
||||
if (progress.current !== undefined && progress.total !== undefined && progress.total > 0) {
|
||||
job.progress = Math.round((progress.current / progress.total) * 100);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Complete a job
|
||||
*/
|
||||
completeJob(jobId: string, result?: any): void {
|
||||
const job = this.jobs.get(jobId);
|
||||
if (job) {
|
||||
job.status = 'completed';
|
||||
job.progress = 100;
|
||||
job.endTime = new Date();
|
||||
job.result = result;
|
||||
job.message = 'Job completed successfully';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fail a job
|
||||
*/
|
||||
failJob(jobId: string, error: string): void {
|
||||
const job = this.jobs.get(jobId);
|
||||
if (job) {
|
||||
job.status = 'failed';
|
||||
job.error = error;
|
||||
job.endTime = new Date();
|
||||
job.message = `Job failed: ${error}`;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up old completed jobs (keep last 10)
|
||||
*/
|
||||
cleanupOldJobs(): void {
|
||||
const allJobs = Array.from(this.jobs.values());
|
||||
const completedJobs = allJobs.filter(job => job.status === 'completed' || job.status === 'failed');
|
||||
|
||||
if (completedJobs.length > 10) {
|
||||
// Sort by end time and remove oldest
|
||||
completedJobs.sort((a, b) => {
|
||||
const aTime = a.endTime?.getTime() || 0;
|
||||
const bTime = b.endTime?.getTime() || 0;
|
||||
return aTime - bTime;
|
||||
});
|
||||
|
||||
const toRemove = completedJobs.slice(0, completedJobs.length - 10);
|
||||
toRemove.forEach(job => {
|
||||
this.jobs.delete(job.jobId);
|
||||
});
|
||||
|
||||
console.log(`🧹 Cleaned up ${toRemove.length} old jobs`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the actual job
|
||||
*/
|
||||
private async runJob(jobId: string, jobOptions: JobOptions): Promise<void> {
|
||||
try {
|
||||
switch (jobOptions.type) {
|
||||
case 's3-sync':
|
||||
await this.runS3SyncJob(jobId, jobOptions.options);
|
||||
break;
|
||||
case 'song-matching':
|
||||
await this.runSongMatchingJob(jobId, jobOptions.options);
|
||||
break;
|
||||
default:
|
||||
throw new Error(`Unknown job type: ${jobOptions.type}`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.failJob(jobId, error instanceof Error ? error.message : 'Unknown error');
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run S3 sync job
|
||||
*/
|
||||
private async runS3SyncJob(jobId: string, options?: any): Promise<void> {
|
||||
try {
|
||||
// Import here to avoid circular dependencies
|
||||
const { S3Service } = await import('./s3Service.js');
|
||||
const { AudioMetadataService } = await import('./audioMetadataService.js');
|
||||
const { SongMatchingService } = await import('./songMatchingService.js');
|
||||
const { MusicFile } = await import('../models/MusicFile.js');
|
||||
const { Song } = await import('../models/Song.js');
|
||||
|
||||
const s3Service = await S3Service.createFromConfig();
|
||||
const audioMetadataService = new AudioMetadataService();
|
||||
const songMatchingService = new SongMatchingService();
|
||||
|
||||
// Phase 1: Quick filename matching
|
||||
this.updateProgress(jobId, {
|
||||
message: 'Phase 1: Fetching files from S3...',
|
||||
current: 0,
|
||||
total: 0
|
||||
});
|
||||
|
||||
const s3Files = await s3Service.listAllFiles();
|
||||
const audioFiles = s3Files.filter(s3File => {
|
||||
const filename = s3File.key.split('/').pop() || s3File.key;
|
||||
return audioMetadataService.isAudioFile(filename);
|
||||
});
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: `Phase 1: Found ${audioFiles.length} audio files, checking database...`,
|
||||
current: 0,
|
||||
total: audioFiles.length
|
||||
});
|
||||
|
||||
// Get existing files
|
||||
const existingFiles = await MusicFile.find({}, { s3Key: 1 });
|
||||
const existingS3Keys = new Set(existingFiles.map(f => f.s3Key));
|
||||
const newAudioFiles = audioFiles.filter(s3File => !existingS3Keys.has(s3File.key));
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: `Phase 1: Processing ${newAudioFiles.length} new audio files...`,
|
||||
current: 0,
|
||||
total: newAudioFiles.length
|
||||
});
|
||||
|
||||
// Get all songs for filename matching
|
||||
const allSongs = await Song.find({}, { id: 1, title: 1, artist: 1, location: 1 });
|
||||
|
||||
const quickMatches: any[] = [];
|
||||
const unmatchedFiles: any[] = [];
|
||||
let processedCount = 0;
|
||||
let phase1Errors = 0;
|
||||
|
||||
for (const s3File of newAudioFiles) {
|
||||
processedCount++;
|
||||
|
||||
try {
|
||||
const filename = s3File.key.split('/').pop() || s3File.key;
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: `Phase 1: Quick filename matching`,
|
||||
current: processedCount,
|
||||
total: newAudioFiles.length
|
||||
});
|
||||
|
||||
// Quick filename matching logic
|
||||
const normalizedS3Filename = filename.replace(/\.[^/.]+$/, '').toLowerCase();
|
||||
let matchedSong = null;
|
||||
|
||||
for (const song of allSongs) {
|
||||
if (song.location) {
|
||||
const rekordboxFilename = song.location.split(/[/\\]/).pop() || song.location;
|
||||
const normalizedRekordboxFilename = rekordboxFilename.replace(/\.[^/.]+$/, '').toLowerCase();
|
||||
|
||||
if (normalizedS3Filename === normalizedRekordboxFilename) {
|
||||
matchedSong = song;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (matchedSong) {
|
||||
const basicMetadata = audioMetadataService.extractBasicMetadataFromFilename(filename);
|
||||
const musicFile = new MusicFile({
|
||||
originalName: filename,
|
||||
s3Key: s3File.key,
|
||||
s3Url: `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
contentType: 'audio/mpeg',
|
||||
size: s3File.size,
|
||||
...basicMetadata,
|
||||
songId: matchedSong._id,
|
||||
});
|
||||
|
||||
// Save immediately for real-time availability
|
||||
await musicFile.save();
|
||||
quickMatches.push(musicFile);
|
||||
|
||||
// Update the Song document to indicate it has an S3 file
|
||||
await Song.updateOne(
|
||||
{ _id: matchedSong._id },
|
||||
{
|
||||
$set: {
|
||||
's3File.musicFileId': musicFile._id,
|
||||
's3File.s3Key': s3File.key,
|
||||
's3File.s3Url': `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
's3File.streamingUrl': `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
's3File.hasS3File': true
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
console.log(`✅ Quick match saved immediately: ${filename}`);
|
||||
} else {
|
||||
unmatchedFiles.push(s3File);
|
||||
}
|
||||
|
||||
|
||||
|
||||
} catch (error) {
|
||||
console.error(`Error in quick matching ${s3File.key}:`, error);
|
||||
unmatchedFiles.push(s3File);
|
||||
phase1Errors++;
|
||||
}
|
||||
}
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: `Phase 1 completed: ${quickMatches.length} quick matches, ${unmatchedFiles.length} unmatched files`,
|
||||
current: newAudioFiles.length,
|
||||
total: newAudioFiles.length
|
||||
});
|
||||
|
||||
// Phase 2: Complex matching for unmatched files
|
||||
if (unmatchedFiles.length > 0) {
|
||||
this.updateProgress(jobId, {
|
||||
message: `Phase 2: Complex matching for ${unmatchedFiles.length} files...`,
|
||||
current: 0,
|
||||
total: unmatchedFiles.length
|
||||
});
|
||||
|
||||
let complexMatches = 0;
|
||||
let stillUnmatched = 0;
|
||||
let phase2Errors = 0;
|
||||
const processedFiles: any[] = [];
|
||||
|
||||
for (let i = 0; i < unmatchedFiles.length; i++) {
|
||||
const s3File = unmatchedFiles[i];
|
||||
|
||||
try {
|
||||
const filename = s3File.key.split('/').pop() || s3File.key;
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: `Phase 2: Complex matching`,
|
||||
current: i + 1,
|
||||
total: unmatchedFiles.length
|
||||
});
|
||||
|
||||
// Download file and extract metadata
|
||||
const fileBuffer = await s3Service.getFileContent(s3File.key);
|
||||
const metadata = await audioMetadataService.extractMetadata(fileBuffer, filename);
|
||||
|
||||
const musicFile = new MusicFile({
|
||||
originalName: filename,
|
||||
s3Key: s3File.key,
|
||||
s3Url: `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
contentType: 'audio/mpeg',
|
||||
size: s3File.size,
|
||||
...metadata,
|
||||
});
|
||||
|
||||
processedFiles.push(musicFile);
|
||||
|
||||
// Try complex matching
|
||||
const matchResult = await songMatchingService.matchMusicFileToSongs(musicFile, {
|
||||
minConfidence: 0.7,
|
||||
enableFuzzyMatching: true,
|
||||
enablePartialMatching: true,
|
||||
maxResults: 1
|
||||
});
|
||||
|
||||
if (matchResult.length > 0 && matchResult[0].confidence >= 0.7) {
|
||||
const bestMatch = matchResult[0];
|
||||
musicFile.songId = bestMatch.song._id;
|
||||
complexMatches++;
|
||||
|
||||
// Update the Song document to indicate it has an S3 file
|
||||
await Song.updateOne(
|
||||
{ _id: bestMatch.song._id },
|
||||
{
|
||||
$set: {
|
||||
's3File.musicFileId': musicFile._id,
|
||||
's3File.s3Key': s3File.key,
|
||||
's3File.s3Url': `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
's3File.streamingUrl': `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
's3File.hasS3File': true
|
||||
}
|
||||
}
|
||||
);
|
||||
} else {
|
||||
stillUnmatched++;
|
||||
}
|
||||
|
||||
// Save immediately for real-time availability
|
||||
await musicFile.save();
|
||||
processedFiles.push(musicFile);
|
||||
|
||||
console.log(`✅ Complex match saved immediately: ${filename} (confidence: ${matchResult.length > 0 ? matchResult[0].confidence : 'N/A'})`);
|
||||
|
||||
|
||||
|
||||
} catch (error) {
|
||||
console.error(`Error processing ${s3File.key}:`, error);
|
||||
stillUnmatched++;
|
||||
phase2Errors++;
|
||||
}
|
||||
}
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: `Phase 2 completed: ${complexMatches} complex matches, ${stillUnmatched} still unmatched`,
|
||||
current: unmatchedFiles.length,
|
||||
total: unmatchedFiles.length
|
||||
});
|
||||
|
||||
// All files have been saved immediately during processing
|
||||
console.log(`✅ All files saved immediately during processing`);
|
||||
|
||||
const result = {
|
||||
phase1: {
|
||||
totalFiles: newAudioFiles.length,
|
||||
quickMatches: quickMatches.length,
|
||||
unmatchedFiles: unmatchedFiles.length,
|
||||
errors: 0
|
||||
},
|
||||
phase2: {
|
||||
processedFiles: processedFiles.length,
|
||||
complexMatches,
|
||||
stillUnmatched,
|
||||
errors: 0
|
||||
},
|
||||
total: {
|
||||
processed: allMusicFiles.length,
|
||||
matched: quickMatches.length + complexMatches,
|
||||
unmatched: stillUnmatched,
|
||||
errors: 0
|
||||
}
|
||||
};
|
||||
|
||||
this.completeJob(jobId, result);
|
||||
} else {
|
||||
// No unmatched files, all quick matches have been saved immediately
|
||||
console.log(`✅ All quick matches saved immediately during processing`);
|
||||
|
||||
const result = {
|
||||
phase1: {
|
||||
totalFiles: newAudioFiles.length,
|
||||
quickMatches: quickMatches.length,
|
||||
unmatchedFiles: 0,
|
||||
errors: 0
|
||||
},
|
||||
phase2: {
|
||||
processedFiles: 0,
|
||||
complexMatches: 0,
|
||||
stillUnmatched: 0,
|
||||
errors: 0
|
||||
},
|
||||
total: {
|
||||
processed: quickMatches.length,
|
||||
matched: quickMatches.length,
|
||||
unmatched: 0,
|
||||
errors: 0
|
||||
}
|
||||
};
|
||||
|
||||
this.completeJob(jobId, result);
|
||||
}
|
||||
|
||||
} catch (error) {
|
||||
this.failJob(jobId, error instanceof Error ? error.message : 'Unknown error');
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run song matching job
|
||||
*/
|
||||
private async runSongMatchingJob(jobId: string, options?: any): Promise<void> {
|
||||
try {
|
||||
// Import here to avoid circular dependencies
|
||||
const { SongMatchingService } = await import('./songMatchingService.js');
|
||||
const { MusicFile } = await import('../models/MusicFile.js');
|
||||
const { Song } = await import('../models/Song.js');
|
||||
|
||||
const matchingService = new SongMatchingService();
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: 'Finding unmatched music files...',
|
||||
current: 0,
|
||||
total: 0
|
||||
});
|
||||
|
||||
// Get all unmatched music files
|
||||
const unmatchedMusicFiles = await MusicFile.find({ songId: { $exists: false } });
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: `Found ${unmatchedMusicFiles.length} unmatched music files`,
|
||||
current: 0,
|
||||
total: unmatchedMusicFiles.length
|
||||
});
|
||||
|
||||
if (unmatchedMusicFiles.length === 0) {
|
||||
this.updateProgress(jobId, {
|
||||
message: 'No unmatched music files found',
|
||||
current: 0,
|
||||
total: 0
|
||||
});
|
||||
|
||||
this.completeJob(jobId, {
|
||||
linked: 0,
|
||||
unmatched: 0,
|
||||
total: 0
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Get all songs for matching
|
||||
const allSongs = await Song.find({});
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: `Starting matching process with ${allSongs.length} songs...`,
|
||||
current: 0,
|
||||
total: unmatchedMusicFiles.length
|
||||
});
|
||||
|
||||
let linked = 0;
|
||||
let unmatched = 0;
|
||||
const batchSize = 50;
|
||||
const musicFileUpdates: any[] = [];
|
||||
const songUpdates: any[] = [];
|
||||
|
||||
for (let i = 0; i < unmatchedMusicFiles.length; i++) {
|
||||
const musicFile = unmatchedMusicFiles[i];
|
||||
|
||||
this.updateProgress(jobId, {
|
||||
message: `Matching music files`,
|
||||
current: i + 1,
|
||||
total: unmatchedMusicFiles.length
|
||||
});
|
||||
|
||||
try {
|
||||
// Get matching suggestions
|
||||
const matches = await matchingService.matchMusicFileToSongs(musicFile, {
|
||||
minConfidence: options?.minConfidence || 0.7,
|
||||
enableFuzzyMatching: options?.enableFuzzyMatching !== false,
|
||||
enablePartialMatching: options?.enablePartialMatching !== false,
|
||||
maxResults: 1
|
||||
});
|
||||
|
||||
if (matches.length > 0 && matches[0].confidence >= (options?.minConfidence || 0.7)) {
|
||||
const bestMatch = matches[0];
|
||||
|
||||
// Prepare updates
|
||||
musicFileUpdates.push({
|
||||
updateOne: {
|
||||
filter: { _id: musicFile._id },
|
||||
update: { songId: bestMatch.song._id }
|
||||
}
|
||||
});
|
||||
|
||||
songUpdates.push({
|
||||
updateOne: {
|
||||
filter: { _id: bestMatch.song._id },
|
||||
update: {
|
||||
$addToSet: {
|
||||
s3File: {
|
||||
musicFileId: musicFile._id,
|
||||
hasS3File: true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
linked++;
|
||||
} else {
|
||||
unmatched++;
|
||||
}
|
||||
|
||||
// Process batch updates
|
||||
if (musicFileUpdates.length >= batchSize) {
|
||||
this.updateProgress(jobId, {
|
||||
message: `Saving to database`,
|
||||
current: i + 1,
|
||||
total: unmatchedMusicFiles.length
|
||||
});
|
||||
|
||||
await MusicFile.bulkWrite(musicFileUpdates);
|
||||
await Song.bulkWrite(songUpdates);
|
||||
|
||||
musicFileUpdates.length = 0;
|
||||
songUpdates.length = 0;
|
||||
}
|
||||
|
||||
} catch (error) {
|
||||
console.error(`Error matching ${musicFile.originalName}:`, error);
|
||||
unmatched++;
|
||||
}
|
||||
}
|
||||
|
||||
// Save remaining updates
|
||||
if (musicFileUpdates.length > 0) {
|
||||
this.updateProgress(jobId, {
|
||||
message: `Saving to database`,
|
||||
current: unmatchedMusicFiles.length,
|
||||
total: unmatchedMusicFiles.length
|
||||
});
|
||||
|
||||
await MusicFile.bulkWrite(musicFileUpdates);
|
||||
await Song.bulkWrite(songUpdates);
|
||||
}
|
||||
|
||||
const result = {
|
||||
linked,
|
||||
unmatched,
|
||||
total: unmatchedMusicFiles.length
|
||||
};
|
||||
|
||||
this.completeJob(jobId, result);
|
||||
|
||||
} catch (error) {
|
||||
this.failJob(jobId, error instanceof Error ? error.message : 'Unknown error');
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const backgroundJobService = new BackgroundJobService();
|
||||
275
packages/backend/src/services/twoPhaseSyncService.ts
Normal file
275
packages/backend/src/services/twoPhaseSyncService.ts
Normal file
@ -0,0 +1,275 @@
|
||||
import { S3Service } from './s3Service.js';
|
||||
import { AudioMetadataService } from './audioMetadataService.js';
|
||||
import { SongMatchingService } from './songMatchingService.js';
|
||||
import { MusicFile } from '../models/MusicFile.js';
|
||||
import { Song } from '../models/Song.js';
|
||||
|
||||
export interface SyncResult {
|
||||
phase1: {
|
||||
totalFiles: number;
|
||||
quickMatches: number;
|
||||
unmatchedFiles: number;
|
||||
errors: number;
|
||||
};
|
||||
phase2: {
|
||||
processedFiles: number;
|
||||
complexMatches: number;
|
||||
stillUnmatched: number;
|
||||
errors: number;
|
||||
};
|
||||
total: {
|
||||
processed: number;
|
||||
matched: number;
|
||||
unmatched: number;
|
||||
errors: number;
|
||||
};
|
||||
}
|
||||
|
||||
export class TwoPhaseSyncService {
|
||||
private s3Service: S3Service;
|
||||
private audioMetadataService: AudioMetadataService;
|
||||
private songMatchingService: SongMatchingService;
|
||||
|
||||
constructor() {
|
||||
this.audioMetadataService = new AudioMetadataService();
|
||||
this.songMatchingService = new SongMatchingService();
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize S3 service
|
||||
*/
|
||||
async initialize(): Promise<void> {
|
||||
this.s3Service = await S3Service.createFromConfig();
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract filename from S3 key
|
||||
*/
|
||||
private getFilenameFromS3Key(s3Key: string): string {
|
||||
return s3Key.split('/').pop() || s3Key;
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract filename from Rekordbox location path
|
||||
*/
|
||||
private getFilenameFromLocation(location: string): string {
|
||||
// Handle both Windows and Unix paths
|
||||
const normalizedPath = location.replace(/\\/g, '/');
|
||||
return normalizedPath.split('/').pop() || location;
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalize filename for comparison (remove extension, lowercase)
|
||||
*/
|
||||
private normalizeFilename(filename: string): string {
|
||||
return filename.replace(/\.[^/.]+$/, '').toLowerCase();
|
||||
}
|
||||
|
||||
/**
|
||||
* Phase 1: Quick filename-based matching
|
||||
*/
|
||||
async phase1QuickMatch(): Promise<{
|
||||
quickMatches: MusicFile[];
|
||||
unmatchedFiles: any[];
|
||||
errors: any[];
|
||||
}> {
|
||||
console.log('🚀 Starting Phase 1: Quick filename-based matching...');
|
||||
|
||||
// Get all S3 files
|
||||
const s3Files = await this.s3Service.listAllFiles();
|
||||
const audioFiles = s3Files.filter(s3File => {
|
||||
const filename = this.getFilenameFromS3Key(s3File.key);
|
||||
return this.audioMetadataService.isAudioFile(filename);
|
||||
});
|
||||
|
||||
console.log(`📁 Found ${audioFiles.length} audio files in S3`);
|
||||
|
||||
// Get existing music files to avoid duplicates
|
||||
const existingFiles = await MusicFile.find({}, { s3Key: 1 });
|
||||
const existingS3Keys = new Set(existingFiles.map(f => f.s3Key));
|
||||
const newAudioFiles = audioFiles.filter(s3File => !existingS3Keys.has(s3File.key));
|
||||
|
||||
console.log(`🆕 Found ${newAudioFiles.length} new audio files to process`);
|
||||
|
||||
// Get all songs from database for filename matching
|
||||
const allSongs = await Song.find({}, { id: 1, title: 1, artist: 1, location: 1 });
|
||||
console.log(`🎵 Found ${allSongs.length} songs in database for matching`);
|
||||
|
||||
const quickMatches: MusicFile[] = [];
|
||||
const unmatchedFiles: any[] = [];
|
||||
const errors: any[] = [];
|
||||
|
||||
for (const s3File of newAudioFiles) {
|
||||
try {
|
||||
const s3Filename = this.getFilenameFromS3Key(s3File.key);
|
||||
const normalizedS3Filename = this.normalizeFilename(s3Filename);
|
||||
|
||||
// Try to find exact filename match in Rekordbox songs
|
||||
let matchedSong = null;
|
||||
|
||||
for (const song of allSongs) {
|
||||
if (song.location) {
|
||||
const rekordboxFilename = this.getFilenameFromLocation(song.location);
|
||||
const normalizedRekordboxFilename = this.normalizeFilename(rekordboxFilename);
|
||||
|
||||
if (normalizedS3Filename === normalizedRekordboxFilename) {
|
||||
matchedSong = song;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (matchedSong) {
|
||||
console.log(`✅ Quick match found: ${s3Filename} -> ${matchedSong.title}`);
|
||||
|
||||
// Extract basic metadata from filename (no need to download file)
|
||||
const basicMetadata = this.audioMetadataService.extractBasicMetadataFromFilename(s3Filename);
|
||||
|
||||
const musicFile = new MusicFile({
|
||||
originalName: s3Filename,
|
||||
s3Key: s3File.key,
|
||||
s3Url: `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
contentType: 'audio/mpeg',
|
||||
size: s3File.size,
|
||||
...basicMetadata,
|
||||
songId: matchedSong._id, // Link to the matched song
|
||||
});
|
||||
|
||||
quickMatches.push(musicFile);
|
||||
} else {
|
||||
console.log(`❓ No quick match for: ${s3Filename}`);
|
||||
unmatchedFiles.push(s3File);
|
||||
}
|
||||
|
||||
} catch (error) {
|
||||
console.error(`❌ Error processing ${s3File.key}:`, error);
|
||||
errors.push({ file: s3File, error: error.message });
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`✅ Phase 1 completed: ${quickMatches.length} quick matches, ${unmatchedFiles.length} unmatched files`);
|
||||
|
||||
return { quickMatches, unmatchedFiles, errors };
|
||||
}
|
||||
|
||||
/**
|
||||
* Phase 2: Complex matching for unmatched files
|
||||
*/
|
||||
async phase2ComplexMatch(unmatchedFiles: any[]): Promise<{
|
||||
processedFiles: MusicFile[];
|
||||
complexMatches: number;
|
||||
stillUnmatched: number;
|
||||
errors: any[];
|
||||
}> {
|
||||
console.log('🔍 Starting Phase 2: Complex matching for unmatched files...');
|
||||
|
||||
const processedFiles: MusicFile[] = [];
|
||||
let complexMatches = 0;
|
||||
let stillUnmatched = 0;
|
||||
const errors: any[] = [];
|
||||
|
||||
for (const s3File of unmatchedFiles) {
|
||||
try {
|
||||
const filename = this.getFilenameFromS3Key(s3File.key);
|
||||
console.log(`🔍 Processing unmatched file: ${filename}`);
|
||||
|
||||
// Download file and extract metadata
|
||||
const fileBuffer = await this.s3Service.getFileContent(s3File.key);
|
||||
const metadata = await this.audioMetadataService.extractMetadata(fileBuffer, filename);
|
||||
|
||||
const musicFile = new MusicFile({
|
||||
originalName: filename,
|
||||
s3Key: s3File.key,
|
||||
s3Url: `${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET_NAME}/${s3File.key}`,
|
||||
contentType: 'audio/mpeg',
|
||||
size: s3File.size,
|
||||
...metadata,
|
||||
});
|
||||
|
||||
processedFiles.push(musicFile);
|
||||
|
||||
// Try complex matching
|
||||
const matchResult = await this.songMatchingService.matchMusicFileToSongs(musicFile, {
|
||||
minConfidence: 0.7,
|
||||
enableFuzzyMatching: true,
|
||||
enablePartialMatching: true,
|
||||
maxResults: 1
|
||||
});
|
||||
|
||||
if (matchResult.length > 0 && matchResult[0].confidence >= 0.7) {
|
||||
const bestMatch = matchResult[0];
|
||||
musicFile.songId = bestMatch.song._id;
|
||||
complexMatches++;
|
||||
console.log(`✅ Complex match found: ${filename} -> ${bestMatch.song.title} (${bestMatch.confidence})`);
|
||||
} else {
|
||||
stillUnmatched++;
|
||||
console.log(`❓ No complex match for: ${filename}`);
|
||||
}
|
||||
|
||||
} catch (error) {
|
||||
console.error(`❌ Error processing ${s3File.key}:`, error);
|
||||
errors.push({ file: s3File, error: error.message });
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`✅ Phase 2 completed: ${complexMatches} complex matches, ${stillUnmatched} still unmatched`);
|
||||
|
||||
return { processedFiles, complexMatches, stillUnmatched, errors };
|
||||
}
|
||||
|
||||
/**
|
||||
* Run complete two-phase sync
|
||||
*/
|
||||
async runTwoPhaseSync(): Promise<SyncResult> {
|
||||
console.log('🎯 Starting Two-Phase S3 Sync...');
|
||||
const startTime = Date.now();
|
||||
|
||||
await this.initialize();
|
||||
|
||||
// Phase 1: Quick filename matching
|
||||
const phase1Result = await this.phase1QuickMatch();
|
||||
|
||||
// Phase 2: Complex matching for unmatched files
|
||||
const phase2Result = await this.phase2ComplexMatch(phase1Result.unmatchedFiles);
|
||||
|
||||
// Combine all music files
|
||||
const allMusicFiles = [...phase1Result.quickMatches, ...phase2Result.processedFiles];
|
||||
|
||||
// Batch save all music files
|
||||
if (allMusicFiles.length > 0) {
|
||||
console.log(`💾 Saving ${allMusicFiles.length} music files to database...`);
|
||||
await MusicFile.insertMany(allMusicFiles);
|
||||
}
|
||||
|
||||
const endTime = Date.now();
|
||||
const duration = ((endTime - startTime) / 1000).toFixed(2);
|
||||
|
||||
const result: SyncResult = {
|
||||
phase1: {
|
||||
totalFiles: phase1Result.quickMatches.length + phase1Result.unmatchedFiles.length,
|
||||
quickMatches: phase1Result.quickMatches.length,
|
||||
unmatchedFiles: phase1Result.unmatchedFiles.length,
|
||||
errors: phase1Result.errors.length,
|
||||
},
|
||||
phase2: {
|
||||
processedFiles: phase2Result.processedFiles.length,
|
||||
complexMatches: phase2Result.complexMatches,
|
||||
stillUnmatched: phase2Result.stillUnmatched,
|
||||
errors: phase2Result.errors.length,
|
||||
},
|
||||
total: {
|
||||
processed: allMusicFiles.length,
|
||||
matched: phase1Result.quickMatches.length + phase2Result.complexMatches,
|
||||
unmatched: phase2Result.stillUnmatched,
|
||||
errors: phase1Result.errors.length + phase2Result.errors.length,
|
||||
},
|
||||
};
|
||||
|
||||
console.log(`🎉 Two-Phase Sync completed in ${duration}s:`);
|
||||
console.log(` Phase 1: ${result.phase1.quickMatches} quick matches, ${result.phase1.unmatchedFiles} unmatched`);
|
||||
console.log(` Phase 2: ${result.phase2.complexMatches} complex matches, ${result.phase2.stillUnmatched} still unmatched`);
|
||||
console.log(` Total: ${result.total.processed} processed, ${result.total.matched} matched, ${result.total.unmatched} unmatched`);
|
||||
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@ -7,6 +7,7 @@ import { PlaylistManager } from "./components/PlaylistManager";
|
||||
import { SongDetails } from "./components/SongDetails";
|
||||
import { Configuration } from "./pages/Configuration";
|
||||
import { PersistentMusicPlayer } from "./components/PersistentMusicPlayer";
|
||||
import { BackgroundJobProgress } from "./components/BackgroundJobProgress";
|
||||
import { MusicPlayerProvider, useMusicPlayer } from "./contexts/MusicPlayerContext";
|
||||
import { useXmlParser } from "./hooks/useXmlParser";
|
||||
import { usePaginatedSongs } from "./hooks/usePaginatedSongs";
|
||||
@ -657,6 +658,9 @@ const RekordboxReader: React.FC = () => {
|
||||
currentSong={currentSong}
|
||||
onClose={handleCloseMusicPlayer}
|
||||
/>
|
||||
|
||||
{/* Background Job Progress */}
|
||||
<BackgroundJobProgress />
|
||||
</Box>
|
||||
);
|
||||
};
|
||||
|
||||
310
packages/frontend/src/components/BackgroundJobProgress.tsx
Normal file
310
packages/frontend/src/components/BackgroundJobProgress.tsx
Normal file
@ -0,0 +1,310 @@
|
||||
import React, { useState, useEffect, useRef } from 'react';
|
||||
import {
|
||||
Box,
|
||||
Flex,
|
||||
Text,
|
||||
Progress,
|
||||
Button,
|
||||
VStack,
|
||||
HStack,
|
||||
Badge,
|
||||
IconButton,
|
||||
useDisclosure,
|
||||
Modal,
|
||||
ModalOverlay,
|
||||
ModalContent,
|
||||
ModalHeader,
|
||||
ModalBody,
|
||||
ModalCloseButton,
|
||||
Table,
|
||||
Thead,
|
||||
Tbody,
|
||||
Tr,
|
||||
Th,
|
||||
Td,
|
||||
Spinner,
|
||||
} from '@chakra-ui/react';
|
||||
import { CloseIcon } from '@chakra-ui/icons';
|
||||
import { api } from '../services/api';
|
||||
|
||||
interface JobProgress {
|
||||
jobId: string;
|
||||
type: 's3-sync' | 'song-matching';
|
||||
status: 'running' | 'completed' | 'failed';
|
||||
progress: number;
|
||||
current: number;
|
||||
total: number;
|
||||
message: string;
|
||||
startTime: Date;
|
||||
endTime?: Date;
|
||||
result?: any;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
interface BackgroundJobProgressProps {
|
||||
jobId?: string;
|
||||
onJobComplete?: (result: any) => void;
|
||||
onJobError?: (error: string) => void;
|
||||
}
|
||||
|
||||
export const BackgroundJobProgress: React.FC<BackgroundJobProgressProps> = ({
|
||||
jobId,
|
||||
onJobComplete,
|
||||
onJobError,
|
||||
}) => {
|
||||
const [jobs, setJobs] = useState<JobProgress[]>([]);
|
||||
const [loading, setLoading] = useState(false);
|
||||
const [error, setError] = useState<string | null>(null);
|
||||
const { isOpen, onOpen, onClose } = useDisclosure();
|
||||
const intervalRef = useRef<NodeJS.Timeout | null>(null);
|
||||
|
||||
// Load all jobs
|
||||
const loadJobs = async () => {
|
||||
try {
|
||||
setLoading(true);
|
||||
const jobsData = await api.getAllJobs();
|
||||
setJobs(jobsData);
|
||||
setError(null);
|
||||
} catch (err) {
|
||||
setError(err instanceof Error ? err.message : 'Failed to load jobs');
|
||||
} finally {
|
||||
setLoading(false);
|
||||
}
|
||||
};
|
||||
|
||||
// Update specific job progress
|
||||
const updateJobProgress = async (jobId: string) => {
|
||||
try {
|
||||
const progress = await api.getJobProgress(jobId);
|
||||
|
||||
setJobs(prev => prev.map(job =>
|
||||
job.jobId === jobId ? progress : job
|
||||
));
|
||||
|
||||
// Handle job completion
|
||||
if (progress.status === 'completed' && onJobComplete) {
|
||||
onJobComplete(progress.result);
|
||||
} else if (progress.status === 'failed' && onJobError) {
|
||||
onJobError(progress.error || 'Job failed');
|
||||
}
|
||||
} catch (err) {
|
||||
console.error('Error updating job progress:', err);
|
||||
}
|
||||
};
|
||||
|
||||
// Start polling for all active jobs
|
||||
const startPolling = () => {
|
||||
if (intervalRef.current) {
|
||||
clearInterval(intervalRef.current);
|
||||
}
|
||||
|
||||
intervalRef.current = setInterval(() => {
|
||||
// Update all active jobs
|
||||
const activeJobIds = jobs.filter(job => job.status === 'running').map(job => job.jobId);
|
||||
activeJobIds.forEach(jobId => {
|
||||
updateJobProgress(jobId);
|
||||
});
|
||||
|
||||
// Also update specific jobId if provided
|
||||
if (jobId) {
|
||||
updateJobProgress(jobId);
|
||||
}
|
||||
}, 2000); // Poll every 2 seconds for less frequent updates
|
||||
};
|
||||
|
||||
// Stop polling
|
||||
const stopPolling = () => {
|
||||
if (intervalRef.current) {
|
||||
clearInterval(intervalRef.current);
|
||||
intervalRef.current = null;
|
||||
}
|
||||
};
|
||||
|
||||
// Load jobs on mount
|
||||
useEffect(() => {
|
||||
loadJobs();
|
||||
}, []);
|
||||
|
||||
// Start polling for active jobs
|
||||
useEffect(() => {
|
||||
const activeJobs = jobs.filter(job => job.status === 'running');
|
||||
if (activeJobs.length > 0 || jobId) {
|
||||
startPolling();
|
||||
}
|
||||
|
||||
return () => {
|
||||
stopPolling();
|
||||
};
|
||||
}, [jobs, jobId]);
|
||||
|
||||
// Cleanup on unmount
|
||||
useEffect(() => {
|
||||
return () => {
|
||||
stopPolling();
|
||||
};
|
||||
}, []);
|
||||
|
||||
const getStatusColor = (status: string) => {
|
||||
switch (status) {
|
||||
case 'running': return 'blue';
|
||||
case 'completed': return 'green';
|
||||
case 'failed': return 'red';
|
||||
default: return 'gray';
|
||||
}
|
||||
};
|
||||
|
||||
const getStatusIcon = (status: string) => {
|
||||
switch (status) {
|
||||
case 'running': return <Spinner size="sm" />;
|
||||
case 'completed': return <Text fontSize="sm">✅</Text>;
|
||||
case 'failed': return <Text fontSize="sm">❌</Text>;
|
||||
default: return <Text fontSize="sm">⏸️</Text>;
|
||||
}
|
||||
};
|
||||
|
||||
const formatDuration = (startTime: Date, endTime?: Date) => {
|
||||
const start = new Date(startTime).getTime();
|
||||
const end = endTime ? new Date(endTime).getTime() : Date.now();
|
||||
const duration = Math.floor((end - start) / 1000);
|
||||
|
||||
if (duration < 60) return `${duration}s`;
|
||||
if (duration < 3600) return `${Math.floor(duration / 60)}m ${duration % 60}s`;
|
||||
return `${Math.floor(duration / 3600)}h ${Math.floor((duration % 3600) / 60)}m`;
|
||||
};
|
||||
|
||||
const activeJobs = jobs.filter(job => job.status === 'running');
|
||||
const completedJobs = jobs.filter(job => job.status === 'completed' || job.status === 'failed');
|
||||
|
||||
return (
|
||||
<>
|
||||
{/* Active Jobs Summary */}
|
||||
{activeJobs.length > 0 && (
|
||||
<Box
|
||||
position="fixed"
|
||||
bottom={4}
|
||||
right={4}
|
||||
bg="gray.800"
|
||||
border="1px solid"
|
||||
borderColor="gray.600"
|
||||
borderRadius="lg"
|
||||
p={4}
|
||||
boxShadow="xl"
|
||||
zIndex={1000}
|
||||
maxW="400px"
|
||||
>
|
||||
<Flex justify="space-between" align="center" mb={3}>
|
||||
<Text fontWeight="bold" fontSize="sm" color="gray.100">
|
||||
Background Jobs ({activeJobs.length})
|
||||
</Text>
|
||||
</Flex>
|
||||
|
||||
<VStack spacing={3} align="stretch">
|
||||
{activeJobs.map(job => (
|
||||
<Box key={job.jobId} p={3} bg="gray.700" borderRadius="md">
|
||||
<Flex justify="space-between" align="center" mb={2}>
|
||||
<Text fontSize="sm" fontWeight="medium" color="gray.100">
|
||||
{job.type === 's3-sync' ? 'S3 Sync' : 'Song Matching'}
|
||||
</Text>
|
||||
<Badge colorScheme={getStatusColor(job.status)} size="sm">
|
||||
{job.status}
|
||||
</Badge>
|
||||
</Flex>
|
||||
|
||||
<Text fontSize="xs" color="gray.400" mb={2}>
|
||||
{job.message}
|
||||
</Text>
|
||||
|
||||
<Progress
|
||||
value={job.progress}
|
||||
size="sm"
|
||||
colorScheme={getStatusColor(job.status)}
|
||||
mb={2}
|
||||
/>
|
||||
|
||||
<Flex justify="space-between" fontSize="xs" color="gray.400">
|
||||
<Text>{job.progress}%</Text>
|
||||
<Text>{formatDuration(job.startTime)}</Text>
|
||||
</Flex>
|
||||
</Box>
|
||||
))}
|
||||
</VStack>
|
||||
</Box>
|
||||
)}
|
||||
|
||||
{/* All Jobs Modal */}
|
||||
<Modal isOpen={isOpen} onClose={onClose} size="xl">
|
||||
<ModalOverlay />
|
||||
<ModalContent bg="gray.800" color="gray.100">
|
||||
<ModalHeader color="gray.100">Background Jobs</ModalHeader>
|
||||
<ModalCloseButton color="gray.300" />
|
||||
<ModalBody pb={6}>
|
||||
<VStack spacing={4} align="stretch">
|
||||
<HStack justify="space-between">
|
||||
<Text fontWeight="bold" color="gray.100">All Jobs</Text>
|
||||
<Button size="sm" onClick={loadJobs} isLoading={loading}>
|
||||
Refresh
|
||||
</Button>
|
||||
</HStack>
|
||||
|
||||
{error && (
|
||||
<Text color="red.500" fontSize="sm">
|
||||
{error}
|
||||
</Text>
|
||||
)}
|
||||
|
||||
<Table variant="simple" size="sm">
|
||||
<Thead>
|
||||
<Tr>
|
||||
<Th color="gray.300">Type</Th>
|
||||
<Th color="gray.300">Status</Th>
|
||||
<Th color="gray.300">Progress</Th>
|
||||
<Th color="gray.300">Duration</Th>
|
||||
<Th color="gray.300">Message</Th>
|
||||
</Tr>
|
||||
</Thead>
|
||||
<Tbody>
|
||||
{jobs.map(job => (
|
||||
<Tr key={job.jobId}>
|
||||
<Td>
|
||||
<Text fontSize="sm" color="gray.100">
|
||||
{job.type === 's3-sync' ? 'S3 Sync' : 'Song Matching'}
|
||||
</Text>
|
||||
</Td>
|
||||
<Td>
|
||||
<HStack spacing={2}>
|
||||
{getStatusIcon(job.status)}
|
||||
<Badge colorScheme={getStatusColor(job.status)} size="sm">
|
||||
{job.status}
|
||||
</Badge>
|
||||
</HStack>
|
||||
</Td>
|
||||
<Td>
|
||||
<Text fontSize="sm" color="gray.100">{job.progress}%</Text>
|
||||
</Td>
|
||||
<Td>
|
||||
<Text fontSize="sm" color="gray.100">
|
||||
{formatDuration(job.startTime, job.endTime)}
|
||||
</Text>
|
||||
</Td>
|
||||
<Td>
|
||||
<Text fontSize="sm" noOfLines={2} color="gray.100">
|
||||
{job.message}
|
||||
</Text>
|
||||
</Td>
|
||||
</Tr>
|
||||
))}
|
||||
</Tbody>
|
||||
</Table>
|
||||
|
||||
{jobs.length === 0 && !loading && (
|
||||
<Text textAlign="center" color="gray.500" py={4}>
|
||||
No background jobs found
|
||||
</Text>
|
||||
)}
|
||||
</VStack>
|
||||
</ModalBody>
|
||||
</ModalContent>
|
||||
</Modal>
|
||||
</>
|
||||
);
|
||||
};
|
||||
@ -16,11 +16,13 @@ import {
|
||||
MenuButton,
|
||||
MenuList,
|
||||
MenuItem,
|
||||
|
||||
Text,
|
||||
HStack,
|
||||
Collapse,
|
||||
MenuDivider,
|
||||
MenuGroup,
|
||||
Icon
|
||||
Icon,
|
||||
Badge
|
||||
} from "@chakra-ui/react";
|
||||
import { ChevronDownIcon, DeleteIcon, ChevronRightIcon, AddIcon, ViewIcon } from "@chakra-ui/icons";
|
||||
import React, { useState, useCallback } from "react";
|
||||
@ -207,7 +209,14 @@ const PlaylistItem: React.FC<PlaylistItemProps> = React.memo(({
|
||||
borderRadius="1px"
|
||||
/>
|
||||
)}
|
||||
{node.name}
|
||||
<Flex align="center" justify="space-between" w="100%">
|
||||
<Text>{node.name}</Text>
|
||||
{((node.trackCount || 0) + (node.tracks?.length || 0)) > 0 && (
|
||||
<Badge size="sm" colorScheme="blue" variant="subtle" ml={2}>
|
||||
{node.trackCount || node.tracks?.length || 0}
|
||||
</Badge>
|
||||
)}
|
||||
</Flex>
|
||||
</Button>
|
||||
<Menu>
|
||||
<MenuButton
|
||||
|
||||
@ -142,7 +142,7 @@ export const PlaylistSelectionModal: React.FC<PlaylistSelectionModalProps> = ({
|
||||
{playlist.name}
|
||||
</Text>
|
||||
<Text fontSize="sm" color="gray.400">
|
||||
{playlist.tracks?.length || 0} tracks
|
||||
{playlist.trackCount || playlist.tracks?.length || 0} tracks
|
||||
</Text>
|
||||
</VStack>
|
||||
</HStack>
|
||||
|
||||
@ -113,6 +113,35 @@ class Api {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// Background job methods
|
||||
async startBackgroundJob(type: 's3-sync' | 'song-matching', options?: any): Promise<{ jobId: string; type: string }> {
|
||||
const response = await fetch(`${API_BASE_URL}/background-jobs/start`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: JSON.stringify({ type, options }),
|
||||
});
|
||||
|
||||
if (!response.ok) throw new Error('Failed to start background job');
|
||||
return response.json();
|
||||
}
|
||||
|
||||
async getJobProgress(jobId: string): Promise<any> {
|
||||
const response = await fetch(`${API_BASE_URL}/background-jobs/progress/${jobId}`);
|
||||
if (!response.ok) throw new Error('Failed to get job progress');
|
||||
return response.json();
|
||||
}
|
||||
|
||||
|
||||
|
||||
async getAllJobs(): Promise<any[]> {
|
||||
const response = await fetch(`${API_BASE_URL}/background-jobs/jobs`);
|
||||
if (!response.ok) throw new Error('Failed to get jobs');
|
||||
const data = await response.json();
|
||||
return data.jobs;
|
||||
}
|
||||
}
|
||||
|
||||
export const api = new Api();
|
||||
@ -71,6 +71,7 @@ export interface PlaylistNode {
|
||||
type: 'folder' | 'playlist';
|
||||
children?: PlaylistNode[];
|
||||
tracks?: string[];
|
||||
trackCount?: number;
|
||||
}
|
||||
|
||||
// Keep the old Playlist interface for backward compatibility during transition
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user