michael/ripster
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.11.0 Multi LW

Browse Source
This commit is contained in:
mboehmlaender
2026-03-17 19:58:32 +00:00
parent 16443c8a45
commit 16c783e1aa
25 changed files with 811 additions and 960 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/services/pipelineService.js
+282 -94
View File
@@ -3847,6 +3847,7 @@ class PipelineService extends EventEmitter {
context: {}
};
this.detectedDisc = null;
this.cdDrives = new Map(); // devicePath → per-drive CD state object
this.activeProcess = null;
this.activeProcesses = new Map();
this.cancelRequestedByJob = new Set();
@@ -4299,13 +4300,108 @@ class PipelineService extends EventEmitter {
for (const [id, data] of this.jobProgress) {
jobProgress[id] = data;
}
const cdDrives = {};
for (const [devicePath, driveState] of this.cdDrives) {
cdDrives[devicePath] = driveState;
}
return {
...this.snapshot,
jobProgress,
queue: this.lastQueueSnapshot
queue: this.lastQueueSnapshot,
cdDrives
};
}
_setCdDriveState(devicePath, updates) {
const existing = this.cdDrives.get(devicePath) || {
devicePath,
state: 'DISC_DETECTED',
jobId: null,
device: null,
progress: 0,
eta: null,
statusText: null,
context: {}
};
const next = {
...existing,
...updates,
devicePath
};
this.cdDrives.set(devicePath, next);
// Sync jobProgress so per-job queries work
const resolvedJobId = Number(next.jobId || 0);
if (Number.isFinite(resolvedJobId) && resolvedJobId > 0) {
const prev = this.jobProgress.get(resolvedJobId) || {};
const contextPatch = next.context && typeof next.context === 'object' ? next.context : null;
const mergedContext = contextPatch
? { ...(prev.context || {}), ...contextPatch }
: (prev.context || null);
const nextJobProgress = {
...prev,
state: next.state,
progress: next.progress ?? prev.progress ?? 0,
eta: next.eta ?? null,
statusText: next.statusText ?? null
};
if (mergedContext && Object.keys(mergedContext).length > 0) {
nextJobProgress.context = mergedContext;
}
this.jobProgress.set(resolvedJobId, nextJobProgress);
}
// Manage disc-polling suspension while any CD drive is actively ripping
const CD_ACTIVE_STATES = new Set(['CD_ANALYZING', 'CD_RIPPING']);
const anyActive = Array.from(this.cdDrives.values()).some((d) => CD_ACTIVE_STATES.has(d.state));
if (anyActive) {
diskDetectionService.suspendPolling();
} else {
diskDetectionService.resumePolling();
}
logger.info('cd:drive:state', { devicePath, state: next.state, jobId: next.jobId });
const snapshotPayload = this.getSnapshot();
wsService.broadcast('PIPELINE_STATE_CHANGED', snapshotPayload);
this.emit('stateChanged', snapshotPayload);
void this.emitQueueChanged();
}
_removeCdDrive(devicePath) {
const drive = this.cdDrives.get(devicePath);
if (!drive) {
return;
}
const jobId = Number(drive.jobId || 0);
this.cdDrives.delete(devicePath);
if (Number.isFinite(jobId) && jobId > 0) {
this.jobProgress.delete(jobId);
}
// Resume polling if no other active drives
const CD_ACTIVE_STATES = new Set(['CD_ANALYZING', 'CD_RIPPING']);
const anyActive = Array.from(this.cdDrives.values()).some((d) => CD_ACTIVE_STATES.has(d.state));
if (!anyActive) {
diskDetectionService.resumePolling();
}
logger.info('cd:drive:removed', { devicePath });
const snapshotPayload = this.getSnapshot();
wsService.broadcast('PIPELINE_STATE_CHANGED', snapshotPayload);
this.emit('stateChanged', snapshotPayload);
}
_getCdDriveByJobId(jobId) {
const normalizedId = Number(jobId);
if (!Number.isFinite(normalizedId) || normalizedId <= 0) {
return null;
}
for (const [devicePath, drive] of this.cdDrives) {
if (Number(drive.jobId) === normalizedId) {
return { ...drive, devicePath };
}
}
return null;
}
normalizeParallelJobsLimit(rawValue) {
const value = Number(rawValue);
if (!Number.isFinite(value) || value < 1) {
@@ -5207,21 +5303,39 @@ class PipelineService extends EventEmitter {
if (enabled !== 'true' && enabled !== '1') {
return;
}
const device = devicePath
|| (settingsMap?.drive_mode === 'explicit' ? String(settingsMap?.drive_device || '').trim() : null)
|| String(settingsMap?.drive_device || '').trim()
|| '/dev/sr0';
logger.info('eject:drive', { device });
await new Promise((resolve) => {
execFile('eject', [device], { timeout: 10000 }, (error) => {
if (error) {
logger.warn('eject:drive:failed', { device, error: errorToMeta(error) });
} else {
logger.info('eject:drive:ok', { device });
// Collect the list of drives to eject
const devicesToEject = [];
if (devicePath) {
devicesToEject.push(devicePath);
} else if (settingsMap?.drive_mode === 'explicit') {
try {
const parsed = JSON.parse(settingsMap?.drive_devices || '[]');
if (Array.isArray(parsed)) {
parsed.map((p) => String(p || '').trim()).filter(Boolean).forEach((p) => devicesToEject.push(p));
}
resolve();
} catch (_error) {
// ignore
}
if (devicesToEject.length === 0) {
const legacy = String(settingsMap?.drive_device || '').trim();
if (legacy) devicesToEject.push(legacy);
}
} else {
devicesToEject.push(String(settingsMap?.drive_device || '').trim() || '/dev/sr0');
}
for (const device of devicesToEject) {
logger.info('eject:drive', { device });
await new Promise((resolve) => {
execFile('eject', [device], { timeout: 10000 }, (error) => {
if (error) {
logger.warn('eject:drive:failed', { device, error: errorToMeta(error) });
} else {
logger.info('eject:drive:ok', { device });
}
resolve();
});
});
});
}
} catch (error) {
logger.warn('eject:drive:error', { error: errorToMeta(error) });
}
@@ -5459,17 +5573,28 @@ class PipelineService extends EventEmitter {
mediaProfile: resolvedMediaProfile
};
logger.info('disc:inserted', { deviceInfo: resolvedDevice, mediaProfile: resolvedMediaProfile });
wsService.broadcast('DISC_DETECTED', { device: resolvedDevice });
// CD discs are tracked per-drive in cdDrives, not in the global state machine
if (resolvedMediaProfile === 'cd') {
const cdDevicePath = String(resolvedDevice.path || '').trim();
if (cdDevicePath) {
const existingDrive = this.cdDrives.get(cdDevicePath);
const ACTIVE_CD_STATES = new Set(['CD_ANALYZING', 'CD_METADATA_SELECTION', 'CD_READY_TO_RIP', 'CD_RIPPING', 'CD_ENCODING']);
if (!ACTIVE_CD_STATES.has(existingDrive?.state)) {
this._setCdDriveState(cdDevicePath, { state: 'DISC_DETECTED', device: resolvedDevice });
}
}
return;
}
const previousDevice = this.snapshot.context?.device || this.detectedDisc;
const previousState = this.snapshot.state;
const previousJobId = this.snapshot.context?.jobId || this.snapshot.activeJobId || null;
const discChanged = previousDevice ? !this.isSameDisc(previousDevice, resolvedDevice) : false;
this.detectedDisc = resolvedDevice;
logger.info('disc:inserted', { deviceInfo: resolvedDevice, mediaProfile: resolvedMediaProfile });
wsService.broadcast('DISC_DETECTED', {
device: resolvedDevice
});
if (discChanged && !RUNNING_STATES.has(previousState) && previousState !== 'DISC_DETECTED' && previousState !== 'READY_TO_ENCODE') {
const message = `Disk gewechselt (${resolvedDevice.discLabel || resolvedDevice.path || 'unbekannt'}). Bitte neu analysieren.`;
@@ -5517,9 +5642,20 @@ class PipelineService extends EventEmitter {
async onDiscRemoved(deviceInfo) {
logger.info('disc:removed', { deviceInfo });
wsService.broadcast('DISC_REMOVED', {
device: deviceInfo
});
wsService.broadcast('DISC_REMOVED', { device: deviceInfo });
const removedPath = String(deviceInfo?.path || '').trim();
// If it's a tracked CD drive, remove or leave it depending on active state
if (removedPath && this.cdDrives.has(removedPath)) {
const driveState = this.cdDrives.get(removedPath);
const ACTIVE_CD_STATES = new Set(['CD_RIPPING', 'CD_ENCODING', 'CD_ANALYZING']);
if (!ACTIVE_CD_STATES.has(driveState?.state)) {
this._removeCdDrive(removedPath);
}
// If actively ripping, leave the entry – rip completion will clean it up
return;
}
this.detectedDisc = null;
if (this.snapshot.state === 'DISC_DETECTED') {
@@ -5865,11 +6001,25 @@ class PipelineService extends EventEmitter {
};
}
async analyzeDisc() {
async analyzeDisc(devicePath = null) {
this.ensureNotBusy('analyze');
logger.info('analyze:start');
logger.info('analyze:start', { devicePath });
let device;
if (devicePath) {
const driveEntry = this.cdDrives.get(devicePath);
device = driveEntry?.device || null;
if (!device && this.detectedDisc?.path === devicePath) {
device = this.detectedDisc;
}
if (!device) {
// Minimal device object — cdRipService only needs path
device = { path: devicePath, mediaProfile: 'cd' };
}
} else {
device = this.detectedDisc || this.snapshot.context?.device;
}
const device = this.detectedDisc || this.snapshot.context?.device;
if (!device) {
const error = new Error('Keine Disk erkannt.');
error.statusCode = 400;
@@ -11381,32 +11531,60 @@ class PipelineService extends EventEmitter {
)
: null;
await this.setState(finalState, {
activeJobId: jobId,
progress: this.snapshot.progress,
eta: null,
statusText: message,
context: {
const failContext = {
...(jobProgressContext && typeof jobProgressContext === 'object' ? jobProgressContext : {}),
jobId,
stage,
error: message,
rawPath: job?.raw_path || null,
outputPath: job?.output_path || null,
mediaProfile: isCdFailure ? 'cd' : resolvedMediaProfile,
inputPath: job?.encode_input_path || encodePlan?.encodeInputPath || null,
selectedMetadata: resolvedSelectedMetadata,
...(isCdFailure ? {
tracks: resolvedTracks,
cdRipConfig: resolvedCdRipConfig,
cdLive: jobProgressContext?.cdLive || null,
devicePath: String(job?.disc_device || jobProgressContext?.devicePath || '').trim() || null,
cdparanoiaCmd: String(makemkvInfo?.cdparanoiaCmd || jobProgressContext?.cdparanoiaCmd || '').trim() || null
} : {}),
canRestartEncodeFromLastSettings: hasConfirmedPlan,
canRestartReviewFromRaw: resolvedMediaProfile !== 'audiobook' && hasRawPath
};
if (isCdFailure) {
const cdDevicePath = String(job?.disc_device || jobProgressContext?.devicePath || '').trim() || null;
if (cdDevicePath) {
this._setCdDriveState(cdDevicePath, {
state: finalState,
jobId,
progress: this.cdDrives.get(cdDevicePath)?.progress ?? 0,
eta: null,
statusText: message,
context: failContext
});
}
} else {
await this.setState(finalState, {
activeJobId: jobId,
progress: this.snapshot.progress,
eta: null,
statusText: message,
context: {
...(jobProgressContext && typeof jobProgressContext === 'object' ? jobProgressContext : {}),
jobId,
stage,
error: message,
rawPath: job?.raw_path || null,
outputPath: job?.output_path || null,
mediaProfile: isCdFailure ? 'cd' : resolvedMediaProfile,
mediaProfile: resolvedMediaProfile,
inputPath: job?.encode_input_path || encodePlan?.encodeInputPath || null,
selectedMetadata: resolvedSelectedMetadata,
...(isCdFailure ? {
tracks: resolvedTracks,
cdRipConfig: resolvedCdRipConfig,
cdLive: jobProgressContext?.cdLive || null,
devicePath: String(job?.disc_device || jobProgressContext?.devicePath || '').trim() || null,
cdparanoiaCmd: String(makemkvInfo?.cdparanoiaCmd || jobProgressContext?.cdparanoiaCmd || '').trim() || null
} : {}),
canRestartEncodeFromLastSettings: hasConfirmedPlan,
canRestartReviewFromRaw: resolvedMediaProfile !== 'audiobook' && hasRawPath
}
});
}});
}
this.cancelRequestedByJob.delete(Number(jobId));
void this.notifyPushover(isCancelled ? 'job_cancelled' : 'job_error', {
@@ -12317,8 +12495,10 @@ class PipelineService extends EventEmitter {
const cdparanoiaCmd = String(settings.cdparanoia_command || 'cdparanoia').trim() || 'cdparanoia';
// Read TOC
await this.setState('CD_ANALYZING', {
activeJobId: job.id,
this._setCdDriveState(devicePath, {
state: 'CD_ANALYZING',
jobId: job.id,
device,
progress: 0,
eta: null,
statusText: 'CD wird analysiert …',
@@ -12354,28 +12534,26 @@ class PipelineService extends EventEmitter {
`CD analysiert: ${tracks.length} Track(s) gefunden.`
);
const runningJobs = await historyService.getRunningJobs();
const foreignRunningJobs = runningJobs.filter((item) => Number(item?.id) !== Number(job.id));
if (!foreignRunningJobs.length) {
const previewTrackPos = tracks[0]?.position ? Number(tracks[0].position) : null;
const cdparanoiaCommandPreview = `${cdparanoiaCmd} -d ${devicePath || '<device>'} ${previewTrackPos || '<trackNr>'} <temp>/trackNN.cdda.wav`;
await this.setState('CD_METADATA_SELECTION', {
activeJobId: job.id,
progress: 0,
eta: null,
statusText: 'CD-Metadaten auswählen',
context: {
jobId: job.id,
device,
mediaProfile: 'cd',
devicePath,
cdparanoiaCmd,
cdparanoiaCommandPreview,
detectedTitle,
tracks
}
});
}
const previewTrackPos = tracks[0]?.position ? Number(tracks[0].position) : null;
const cdparanoiaCommandPreview = `${cdparanoiaCmd} -d ${devicePath || '<device>'} ${previewTrackPos || '<trackNr>'} <temp>/trackNN.cdda.wav`;
this._setCdDriveState(devicePath, {
state: 'CD_METADATA_SELECTION',
jobId: job.id,
device,
progress: 0,
eta: null,
statusText: 'CD-Metadaten auswählen',
context: {
jobId: job.id,
device,
mediaProfile: 'cd',
devicePath,
cdparanoiaCmd,
cdparanoiaCommandPreview,
detectedTitle,
tracks
}
});
return { jobId: job.id, detectedTitle, tracks };
} catch (error) {
@@ -12480,18 +12658,20 @@ class PipelineService extends EventEmitter {
`Metadaten gesetzt: "${title}" (${artist || '-'}, ${year || '-'}).`
);
if (this.isPrimaryJob(jobId)) {
const resolvedDevicePath = String(job?.disc_device || this.snapshot?.context?.device?.path || '').trim() || null;
const resolvedCdparanoiaCmd = String(cdInfo?.cdparanoiaCmd || 'cdparanoia').trim() || 'cdparanoia';
const previewTrackPos = mergedTracks[0]?.position ? Number(mergedTracks[0].position) : null;
const cdparanoiaCommandPreview = `${resolvedCdparanoiaCmd} -d ${resolvedDevicePath || '<device>'} ${previewTrackPos || '<trackNr>'} <temp>/trackNN.cdda.wav`;
await this.setState('CD_READY_TO_RIP', {
activeJobId: jobId,
const resolvedDevicePath = String(job?.disc_device || '').trim() || null;
const resolvedCdparanoiaCmd = String(cdInfo?.cdparanoiaCmd || 'cdparanoia').trim() || 'cdparanoia';
const previewTrackPos = mergedTracks[0]?.position ? Number(mergedTracks[0].position) : null;
const cdparanoiaCommandPreview = `${resolvedCdparanoiaCmd} -d ${resolvedDevicePath || '<device>'} ${previewTrackPos || '<trackNr>'} <temp>/trackNN.cdda.wav`;
const existingDrive = resolvedDevicePath ? this.cdDrives.get(resolvedDevicePath) : null;
if (resolvedDevicePath) {
this._setCdDriveState(resolvedDevicePath, {
state: 'CD_READY_TO_RIP',
jobId,
progress: 0,
eta: null,
statusText: 'CD bereit zum Rippen',
context: {
...(this.snapshot.context || {}),
...(existingDrive?.context || {}),
jobId,
mediaProfile: 'cd',
tracks: mergedTracks,
@@ -12702,8 +12882,7 @@ class PipelineService extends EventEmitter {
}
const cdInfo = this.safeParseJson(activeJob.makemkv_info_json) || {};
const device = this.detectedDisc || this.snapshot.context?.device;
const devicePath = String(device?.path || activeJob.disc_device || '').trim();
const devicePath = String(activeJob.disc_device || '').trim();
if (!devicePath) {
const error = new Error('Kein CD-Laufwerk bekannt.');
@@ -12921,13 +13100,15 @@ class PipelineService extends EventEmitter {
encode_plan_json: JSON.stringify(cdEncodePlan)
});
await this.setState('CD_RIPPING', {
activeJobId,
const existingCdDrive = this.cdDrives.get(devicePath);
this._setCdDriveState(devicePath, {
state: 'CD_RIPPING',
jobId: activeJobId,
progress: 0,
eta: null,
statusText: 'CD wird gerippt …',
context: {
...(this.snapshot.context || {}),
...(existingCdDrive?.context || {}),
jobId: activeJobId,
mediaProfile: 'cd',
tracks: mergedTracks,
@@ -13107,7 +13288,7 @@ class PipelineService extends EventEmitter {
}
}
};
await cdRipService.ripAndEncode({
const { encodeResults: cdEncodeResults = [] } = await cdRipService.ripAndEncode({
jobId,
devicePath,
cdparanoiaCmd,
@@ -13189,6 +13370,17 @@ class PipelineService extends EventEmitter {
cdLive: buildLiveContext(null)
}
});
// Keep cdDrives entry in sync (no broadcast — PIPELINE_PROGRESS covers real-time updates)
const currentCdDriveEntry = this.cdDrives.get(devicePath);
if (currentCdDriveEntry) {
this.cdDrives.set(devicePath, {
...currentCdDriveEntry,
state: stage,
progress: clampedPercent,
statusText,
context: { ...(currentCdDriveEntry.context || {}), cdLive: buildLiveContext(null) }
});
}
},
onLog: async (level, msg) => {
await historyService.appendLog(jobId, 'SYSTEM', msg).catch(() => {});
@@ -13249,6 +13441,7 @@ class PipelineService extends EventEmitter {
output_path: outputDir,
handbrake_info_json: JSON.stringify({
mode: 'cd_rip',
tracks: cdEncodeResults,
preEncodeScripts: preEncodeScriptsSummary,
postEncodeScripts: postEncodeScriptsSummary
})
@@ -13273,8 +13466,9 @@ class PipelineService extends EventEmitter {
currentTrackPosition = null;
const finishedCdLive = buildLiveContext(null);
await this.setState('FINISHED', {
activeJobId: jobId,
this._setCdDriveState(devicePath, {
state: 'FINISHED',
jobId,
progress: 100,
eta: null,
statusText: finishedStatusText,
@@ -13298,20 +13492,14 @@ class PipelineService extends EventEmitter {
} catch (error) {
settleLifecycle();
const failedCdLive = buildLiveContext(currentTrackPosition || null);
await this.updateProgress(
this.snapshot.state === 'CD_ENCODING' ? 'CD_ENCODING' : 'CD_RIPPING',
this.snapshot.progress,
null,
this.snapshot.statusText,
processKey,
{
contextPatch: {
cdLive: failedCdLive
}
}
);
const currentDriveEntry = this.cdDrives.get(devicePath);
const currentDriveState = currentDriveEntry?.state || 'CD_RIPPING';
const failStage = currentDriveState === 'CD_ENCODING' ? 'CD_ENCODING' : 'CD_RIPPING';
await this.updateProgress(failStage, currentDriveEntry?.progress ?? 0, null, currentDriveEntry?.statusText ?? null, processKey, {
contextPatch: { cdLive: failedCdLive }
});
logger.error('cd:rip:failed', { jobId, error: errorToMeta(error) });
await this.failJob(jobId, this.snapshot.state === 'CD_ENCODING' ? 'CD_ENCODING' : 'CD_RIPPING', error);
await this.failJob(jobId, failStage, error);
} finally {
this.activeProcesses.delete(processKey);
this.syncPrimaryActiveProcess();