From 9a05ddb3e92fe021d7b1b1347e3bf45bf3752340 Mon Sep 17 00:00:00 2001 From: kq Date: Fri, 14 Aug 2026 21:00:34 +0200 Subject: [PATCH] Update index.js --- index.js | 200 +++++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 172 insertions(+), 28 deletions(-) diff --git a/index.js b/index.js index 1ffec6e..ba725d9 100644 --- a/index.js +++ b/index.js @@ -30,12 +30,31 @@ const VOLUME = 0.3; // 0.0 - 1.0 const CACHE_DIR = path.join(__dirname, 'cache'); if (!fs.existsSync(CACHE_DIR)) fs.mkdirSync(CACHE_DIR); -// On startup, remove any leftover .tmp files from downloads that never -// finished (e.g. the process was killed mid-write). These are never valid -// cache entries, so there's no reason to keep them around. +// A real encoded opus file is never this small. Anything under this size is +// treated as a corrupt/truncated download rather than a usable cache entry — +// used both when finalizing a fresh download and when reading an existing +// cache file back off disk. +const MIN_VALID_CACHE_BYTES = 4096; + +// Startup cleanup pass: +// - any leftover .tmp files are from downloads that never finished (e.g. +// the process was killed mid-write) — never valid, always safe to remove. +// - any .opus file smaller than MIN_VALID_CACHE_BYTES is a corrupt/truncated +// cache entry that could have been left behind by an older version of +// this bot (or a crash right at the rename boundary) — remove it so it +// gets cleanly redownloaded next time it's requested instead of being +// served as broken audio. for (const f of fs.readdirSync(CACHE_DIR)) { + const full = path.join(CACHE_DIR, f); if (f.endsWith('.tmp')) { - try { fs.unlinkSync(path.join(CACHE_DIR, f)); } catch {} + try { fs.unlinkSync(full); } catch {} + } else if (f.endsWith('.opus')) { + try { + if (fs.statSync(full).size < MIN_VALID_CACHE_BYTES) { + fs.unlinkSync(full); + console.log(`[Cache] Removed corrupt cache file on startup: ${f}`); + } + } catch {} } } @@ -48,7 +67,16 @@ if (fs.existsSync(META_FILE)) { } function saveMetaCache() { - fs.writeFileSync(META_FILE, JSON.stringify(metaCache, null, 2)); + // Write via tmp+rename (same pattern as the audio cache) so a process + // kill mid-write can never leave metadata.json half-written/corrupt + // JSON that fails to parse on next boot. + const tmp = META_FILE + '.tmp'; + try { + fs.writeFileSync(tmp, JSON.stringify(metaCache, null, 2)); + fs.renameSync(tmp, META_FILE); + } catch (err) { + console.error('[Meta] Failed to save metadata cache:', err.message); + } } function getAudioCachePath(url) { @@ -56,6 +84,24 @@ function getAudioCachePath(url) { return path.join(CACHE_DIR, `${hash}.opus`); } +// A cache file only counts as usable if it exists AND is large enough to be +// real audio. Anything smaller gets deleted on sight so a corrupt entry is +// never served twice. +function isCacheFileValid(cachePath) { + try { + return fs.statSync(cachePath).size >= MIN_VALID_CACHE_BYTES; + } catch { + return false; + } +} + +function evictIfInvalid(cachePath) { + if (fs.existsSync(cachePath) && !isCacheFileValid(cachePath)) { + console.warn(`[Cache] Discarding corrupt/empty cache file: ${path.basename(cachePath)}`); + try { fs.unlinkSync(cachePath); } catch {} + } +} + // ─── State (per guild) ──────────────────────────────────────────────────────── const guilds = new Map(); @@ -81,7 +127,23 @@ function getState(guildId) { // cache is populated correctly for next time instead of ending up with // a truncated file. // - The cache file only ever appears at its final path once fully written -// (write to .tmp, then atomic rename on success). +// AND large enough to be real audio (write to .tmp, size-check, then +// atomic rename on success). +// +// IMPORTANT — multi-guild fan-out: +// The shared ffmpeg output is forwarded to the cache file and to every +// subscribed guild MANUALLY (not via stream.pipe()), and a subscriber's +// write() return value is ignored. This is deliberate: with .pipe(), Node +// ties the production rate of the single shared stream to whichever +// destination is slowest to drain. One guild with a stalled voice +// connection (or a paused player) would backpressure the shared download +// and stall the cache write *and every other guild listening to the same +// song*. Writing manually means a stuck subscriber can only ever hurt +// itself. Subscribers also swallow their own 'error'/'close' quietly — +// previously a subscriber being destroyed (e.g. on skip) while still +// attached to the shared stream could throw an unhandled 'error', which +// crashes the entire Node process and kills playback in every guild, not +// just the one that skipped. const activeDownloads = new Map(); // url -> DownloadJob class DownloadJob { @@ -90,7 +152,6 @@ class DownloadJob { this.cachePath = getAudioCachePath(url); this.tmpPath = this.cachePath + '.tmp'; this.subscribers = new Set(); - this.source = new PassThrough(); // fan-out: ffmpeg output -> all subscribers + cache file this.failed = false; this._start(); } @@ -119,6 +180,15 @@ class DownloadJob { ytdlp.stderr.on('data', (d) => console.error('[yt-dlp]', d.toString().trim())); ffmpeg.stderr.on('data', (d) => console.error('[ffmpeg]', d.toString().trim())); + // Child-process pipe streams can each emit their own 'error' + // (e.g. EPIPE when one side of the pipe dies before the other). + // Without listeners here that's an unhandled exception that + // crashes the whole bot — this is the fix for downloads breaking + // whenever multiple guilds/voice channels were involved. + ytdlp.stdout.on('error', () => {}); + ffmpeg.stdin.on('error', () => {}); + ffmpeg.stdout.on('error', () => {}); + ytdlp.stdout.pipe(ffmpeg.stdin); ytdlp.on('error', (err) => this._fail(err)); @@ -133,9 +203,25 @@ class DownloadJob { const cacheStream = fs.createWriteStream(this.tmpPath); this.cacheStream = cacheStream; + cacheStream.on('error', (err) => this._fail(err)); - ffmpeg.stdout.pipe(this.source); - this.source.pipe(cacheStream); + // Manual fan-out — see class-level comment for why this replaces + // ffmpeg.stdout.pipe(...) to the cache file and every subscriber. + ffmpeg.stdout.on('data', (chunk) => { + if (this.failed) return; + if (!cacheStream.destroyed) cacheStream.write(chunk); + for (const sub of this.subscribers) { + if (!sub.destroyed) sub.write(chunk); + } + }); + + ffmpeg.stdout.on('end', () => { + if (this.failed) return; + if (!cacheStream.destroyed) cacheStream.end(); + for (const sub of this.subscribers) { + if (!sub.destroyed) sub.end(); + } + }); ffmpeg.on('close', (code) => { if (code !== 0 && code !== null) { @@ -143,17 +229,34 @@ class DownloadJob { } }); - cacheStream.on('error', (err) => this._fail(err)); - cacheStream.on('finish', () => { - if (this.failed) return; // already handled via _fail cleanup + if (this.failed) return; + this._finalizeCache(); + }); + } + + _finalizeCache() { + fs.stat(this.tmpPath, (err, stats) => { + activeDownloads.delete(this.url); + if (err) { + console.error('[Cache] Stat failed:', err.message); + fs.unlink(this.tmpPath, () => {}); + return; + } + // Guard against corrupt/empty cache files — e.g. yt-dlp or + // ffmpeg exited "cleanly" but produced little/no real audio. + // Never let a file this small become a permanent cache entry. + if (stats.size < MIN_VALID_CACHE_BYTES) { + console.error(`[Cache] Discarding suspiciously small file (${stats.size}B): ${this.url}`); + fs.unlink(this.tmpPath, () => {}); + return; + } fs.rename(this.tmpPath, this.cachePath, (err) => { if (err) { console.error('[Cache] Rename failed:', err.message); } else { console.log(`[Cache] Saved: ${path.basename(this.cachePath)}`); } - activeDownloads.delete(this.url); }); }); } @@ -166,20 +269,27 @@ class DownloadJob { try { this.ffmpeg.kill(); } catch {} try { this.cacheStream.destroy(); } catch {} fs.unlink(this.tmpPath, () => {}); // never leave a partial file behind - for (const sub of this.subscribers) sub.destroy(err); + for (const sub of this.subscribers) { + if (!sub.destroyed) sub.destroy(err); + } + this.subscribers.clear(); activeDownloads.delete(this.url); } - // Each consumer (a guild's playback) gets its own tap on the shared - // source. Closing this tap (e.g. because the guild skipped the track) - // only removes that listener — it does not affect the download itself. + // Each consumer (a guild's playback) gets its own independent tap on + // the shared download. Destroying this tap (e.g. because that guild + // skipped/stopped/left) must ONLY remove that one listener — it must + // never affect the shared child processes, the cache write, or any + // other guild's tap. subscribe() { const sub = new PassThrough(); this.subscribers.add(sub); - this.source.pipe(sub); const cleanup = () => this.subscribers.delete(sub); sub.on('close', cleanup); sub.on('end', cleanup); + // Swallow subscriber-side errors so a torn-down voice stream can + // never surface as an unhandled 'error' and crash the process. + sub.on('error', cleanup); return sub; } } @@ -223,6 +333,11 @@ client.on('messageCreate', async (message) => { if (!query) return message.channel.send('❌ Podaj link lub frazę do wyszukania!'); await cmdPlay(message, query); + } else if (command === 'fplay') { + const query = args.join(' '); + if (!query) return message.channel.send('❌ Podaj link lub frazę do wyszukania!'); + await cmdPlay(message, query, { force: true }); + } else if (command === 'skip') { await cmdSkip(message); @@ -239,6 +354,7 @@ client.on('messageCreate', async (message) => { await message.channel.send( '🎵 **Komendy:**\n' + '`!play ` – dodaj utwór do kolejki\n' + + '`!fplay ` – wymuś świeże pobranie (ignoruje cache)\n' + '`!skip` – pomiń aktualny utwór\n' + '`!queue` – pokaż kolejkę\n' + '`!stop` – zatrzymaj i wyczyść kolejkę\n' + @@ -253,7 +369,7 @@ client.on('messageCreate', async (message) => { // ─── Commands ───────────────────────────────────────────────────────────────── -async function cmdPlay(message, query) { +async function cmdPlay(message, query, { force = false } = {}) { const voiceChannel = message.member?.voice?.channel; if (!voiceChannel) return message.channel.send('❌ Dołącz do kanału głosowego!'); @@ -305,9 +421,14 @@ async function cmdPlay(message, query) { } const cacheKey = query.toLowerCase().trim(); - if (metaCache[cacheKey]) { + + // Force play skips the metadata-cache shortcut entirely: always do a + // fresh yt-dlp lookup, and below, always wipe any existing audio cache + // for the resolved URL so playback can't come from a stale/corrupt file. + if (!force && metaCache[cacheKey]) { const song = metaCache[cacheKey]; - const audioCached = fs.existsSync(getAudioCachePath(song.url)); + evictIfInvalid(getAudioCachePath(song.url)); + const audioCached = isCacheFileValid(getAudioCachePath(song.url)); console.log(`[Meta] HIT: "${query}" → ${song.url} (audio cached: ${audioCached})`); state.queue.push(song); await message.channel.send( @@ -317,7 +438,7 @@ async function cmdPlay(message, query) { return; } - await message.channel.send(`🔍 Szukam: **${query}**…`); + await message.channel.send(force ? `🔁 Wymuszam ponowne pobranie: **${query}**…` : `🔍 Szukam: **${query}**…`); const song = await resolveSong(query); metaCache[cacheKey] = song; @@ -325,6 +446,22 @@ async function cmdPlay(message, query) { saveMetaCache(); console.log(`[Meta] Cached: "${query}" → ${song.url}`); + if (force) { + const cachePath = getAudioCachePath(song.url); + if (fs.existsSync(cachePath)) { + try { + fs.unlinkSync(cachePath); + console.log(`[Cache] Force-cleared: ${path.basename(cachePath)}`); + } catch (err) { + console.error('[Cache] Force-clear failed:', err.message); + } + } + // If a download for this URL happens to already be in flight + // (e.g. another guild requested it a moment ago), that download is + // already fresh, not served from cache — buildAudioResource will + // simply join it like any other cache miss, so nothing more to do. + } + state.queue.push(song); await message.channel.send(`✅ Dodano do kolejki: **${song.title}** [${song.duration}]`); @@ -400,13 +537,20 @@ function playNext(guildId) { function buildAudioResource(url) { const cachePath = getAudioCachePath(url); - // Cache HIT — instant play from disk. Because writes are atomic - // (tmp file + rename), a file only exists here if it's complete. + // Cache HIT — instant play from disk, but only if the file is actually + // large enough to be real audio. Because writes are atomic (tmp file + + // size-check + rename), a file should only ever exist here if it's + // complete and valid — this check is a last-line-of-defense safety net + // in case an older/corrupt file is still sitting on disk. if (fs.existsSync(cachePath)) { - console.log(`[Cache] HIT: ${path.basename(cachePath)}`); - return createAudioResource(fs.createReadStream(cachePath), { - inputType: StreamType.OggOpus, - }); + if (isCacheFileValid(cachePath)) { + console.log(`[Cache] HIT: ${path.basename(cachePath)}`); + return createAudioResource(fs.createReadStream(cachePath), { + inputType: StreamType.OggOpus, + }); + } + console.warn(`[Cache] Corrupt/empty cache file, discarding and redownloading: ${path.basename(cachePath)}`); + try { fs.unlinkSync(cachePath); } catch {} } // Cache MISS — join (or start) a shared download for this URL.