21689294588

Committed 04 Feb 2026 09:35PM UTC coverage: 93.613% (-0.09%) from 93.701%

Build # 21689294588

Build Type

push

github

Committed by

xxczaki

Commit Message

fix

Run Details

972 of 1091 branches covered (89.09%)

Branch coverage included in aggregate %.

2135 of 2228 relevant lines covered (95.83%)

11.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.2

/src/utils/initializePlayer.ts

import { createReadStream, createWriteStream } from 'node:fs';
import { stat } from 'node:fs/promises';
import { Readable } from 'node:stream';
import type { Client } from 'discord.js';
import {
        InterceptedStream,
        onBeforeCreateStream,
        onStreamExtracted,
        Player,
} from 'discord-player';
import { YoutubeSabrExtractor } from 'discord-player-googlevideo';
import { SpotifyExtractor } from 'discord-player-spotify';
import defineCustomFilters from './defineCustomFilters';
import logger from './logger';
import { OpusCacheManager } from './OpusCacheManager';
import { RedisQueryCache } from './RedisQueryCache';
import redis from './redis';

const CACHE_WRITE_BUFFER_MS = 5000;
const MIN_CACHE_FILE_SIZE_BYTES = 1024;

/*
        YouTube streams Opus audio at ~128-136 kbps (~17 KB/s).
        We use this to estimate expected file size from track duration
        and detect corrupted cache files (e.g., from interrupted downloads).
*/
const EXPECTED_BYTES_PER_SECOND = 17_000;
const MIN_CACHE_SIZE_RATIO = 0.8;

let initializedPlayer: Player;

const activeWrites = new Set<string>();
const opusCacheManager = OpusCacheManager.initialize();

defineCustomFilters();

export default async function getInitializedPlayer(client: Client<boolean>) {
        if (!initializedPlayer) {
                initializedPlayer = new Player(client, {
                        queryCache: new RedisQueryCache(redis),
                });

                await opusCacheManager.scan();

                /*
                        See:
                        - https://github.com/Androz2091/discord-player/discussions/1962
                        - https://github.com/Androz2091/discord-player/discussions/1985
                */
                onBeforeCreateStream(async (track) => {
                        const durationSeconds = Math.round(track.durationMS / 1000);
                        const matchedEntry = opusCacheManager.findMatch(
                                track.cleanTitle,
                                track.author,
                                durationSeconds,
                        );

                        if (!matchedEntry) {
                                return null;
                        }

                        const filePath = opusCacheManager.getFilePath(matchedEntry.filename);

                        if (activeWrites.has(filePath)) {
                                return null;
                        }

                        try {
                                const stats = await stat(filePath);

                                const now = Date.now();
                                const fileAge = now - stats.mtime.getTime();

                                if (fileAge < CACHE_WRITE_BUFFER_MS) {
                                        return null;
                                }

                                if (stats.size < MIN_CACHE_FILE_SIZE_BYTES) {
                                        logger.warn(
                                                {
                                                        filePath,
                                                        size: stats.size,
                                                },
                                                'Deleting undersized cache file',
                                        );

                                        void opusCacheManager.deleteEntry(matchedEntry.filename);

                                        return null;
                                }

                                if (track.durationMS > 0) {
                                        const expectedSize =
                                                (track.durationMS / 1000) * EXPECTED_BYTES_PER_SECOND;
                                        const minValidSize = expectedSize * MIN_CACHE_SIZE_RATIO;

                                        if (stats.size < minValidSize) {
                                                logger.warn(
                                                        {
                                                                filePath,
                                                                actualSize: stats.size,
                                                                expectedSize: Math.round(expectedSize),
                                                                durationMS: track.durationMS,
                                                        },
                                                        'Deleting corrupted cache file',
                                                );

                                                void opusCacheManager.deleteEntry(matchedEntry.filename);

                                                track.setMetadata({
                                                        ...(track.metadata ?? {}),
                                                        cacheInvalidated: true,
                                                });

                                                return null;
                                        }
                                }

                                track.setMetadata({
                                        ...(track.metadata ?? {}),
                                        isFromCache: true,
                                        cacheFilename: matchedEntry.filename,
                                });

                                return createReadStream(filePath);
                        } catch {
                                return null;
                        }
                });

                onStreamExtracted(async (stream, track) => {
                        if (typeof stream === 'string') {
                                return stream;
                        }

                        const isReadable = stream instanceof Readable;
                        const readable = isReadable ? stream : stream.stream;

                        if (
                                track.metadata &&
                                typeof track.metadata === 'object' &&
                                'isFromCache' in track.metadata &&
                                track.metadata.isFromCache
                        ) {
                                if (isReadable) {
                                        return readable;
                                }

                                return stream;
                        }

                        const interceptor = new InterceptedStream();

                        const trackMetadata = {
                                title: track.cleanTitle,
                                author: track.author,
                                durationMS: track.durationMS,
                        };
                        const filename = opusCacheManager.generateFilename(trackMetadata);
                        const filePath = opusCacheManager.getFilePath(filename);

                        try {
                                activeWrites.add(filePath);

                                const writeStream = createWriteStream(filePath);

                                let streamEndedNormally = false;

                                const cleanup = () => {
                                        interceptor.interceptors.delete(writeStream);

                                        if (!writeStream.destroyed) {
                                                writeStream.destroy();
                                        }

                                        activeWrites.delete(filePath);
                                        void opusCacheManager.deleteEntry(filename);
                                };

                                writeStream.on('error', (error) => {
                                        logger.error(error, 'Opus cache write stream error');
                                        cleanup();
                                });

                                writeStream.on('close', () => {
                                        activeWrites.delete(filePath);

                                        opusCacheManager.addEntry({
                                                filename,
                                                title: trackMetadata.title,
                                                author: trackMetadata.author,
                                                durationSeconds: Math.round(trackMetadata.durationMS / 1000),
                                        });
                                });

                                readable.on('error', () => {
                                        cleanup();
                                });

                                readable.on('end', () => {
                                        streamEndedNormally = true;
                                });

                                readable.on('close', () => {
                                        if (!streamEndedNormally && activeWrites.has(filePath)) {
                                                void cleanup();
                                        }
                                });

                                interceptor.interceptors.add(writeStream);

                                if (isReadable) {
                                        return readable.pipe(interceptor);
                                }

                                return {
                                        stream: readable.pipe(interceptor),
                                        $fmt: stream.$fmt,
                                };
                        } catch (error) {
                                activeWrites.delete(filePath);

                                logger.error(error, 'Failed to create opus cache write stream');

                                if (isReadable) {
                                        return readable;
                                }

                                return stream;
                        }
                });

                /* v8 ignore start */
                await initializedPlayer.extractors.register(YoutubeSabrExtractor, {});
                await initializedPlayer.extractors.register(SpotifyExtractor, {
                        market: 'PL',
                });

                await initializedPlayer.extractors.loadMulti([
                        YoutubeSabrExtractor,
                        SpotifyExtractor,
                ]);
                /* v8 ignore stop */
        }

        return initializedPlayer;
}

1	import { createReadStream, createWriteStream } from 'node:fs';
2	import { stat } from 'node:fs/promises';
3	import { Readable } from 'node:stream';
4	import type { Client } from 'discord.js';
5	import {
6	InterceptedStream,
7	onBeforeCreateStream,
8	onStreamExtracted,
9	Player,
10	} from 'discord-player';
11	import { YoutubeSabrExtractor } from 'discord-player-googlevideo';
12	import { SpotifyExtractor } from 'discord-player-spotify';
13	import defineCustomFilters from './defineCustomFilters';
14	import logger from './logger';
15	import { OpusCacheManager } from './OpusCacheManager';
16	import { RedisQueryCache } from './RedisQueryCache';
17	import redis from './redis';
18
19	const CACHE_WRITE_BUFFER_MS = 5000;	1✔
20	const MIN_CACHE_FILE_SIZE_BYTES = 1024;	1✔
21
22	/*
23	YouTube streams Opus audio at ~128-136 kbps (~17 KB/s).
24	We use this to estimate expected file size from track duration
25	and detect corrupted cache files (e.g., from interrupted downloads).
26	*/
27	const EXPECTED_BYTES_PER_SECOND = 17_000;	1✔
28	const MIN_CACHE_SIZE_RATIO = 0.8;	1✔
29
30	let initializedPlayer: Player;
31
32	const activeWrites = new Set<string>();	1✔
33	const opusCacheManager = OpusCacheManager.initialize();	1✔
34
35	defineCustomFilters();	1✔
36
37	export default async function getInitializedPlayer(client: Client<boolean>) {
38	if (!initializedPlayer) {	20✔
39	initializedPlayer = new Player(client, {	1✔
40	queryCache: new RedisQueryCache(redis),
41	});
42
43	await opusCacheManager.scan();	1✔
44
45	/*
46	See:
47	- https://github.com/Androz2091/discord-player/discussions/1962
48	- https://github.com/Androz2091/discord-player/discussions/1985
49	*/
50	onBeforeCreateStream(async (track) => {	1✔
51	const durationSeconds = Math.round(track.durationMS / 1000);	8✔
52	const matchedEntry = opusCacheManager.findMatch(	8✔
53	track.cleanTitle,
54	track.author,
55	durationSeconds,
56	);
57
58	if (!matchedEntry) {	8✔
59	return null;	1✔
60	}
61
62	const filePath = opusCacheManager.getFilePath(matchedEntry.filename);	7✔
63
64	if (activeWrites.has(filePath)) {	7!
65	return null;	×
66	}
67
68	try {	7✔
69	const stats = await stat(filePath);	7✔
70
71	const now = Date.now();	6✔
72	const fileAge = now - stats.mtime.getTime();	6✔
73
74	if (fileAge < CACHE_WRITE_BUFFER_MS) {	6✔
75	return null;	1✔
76	}
77
78	if (stats.size < MIN_CACHE_FILE_SIZE_BYTES) {	5✔
79	logger.warn(	1✔
80	{
81	filePath,
82	size: stats.size,
83	},
84	'Deleting undersized cache file',
85	);
86
87	void opusCacheManager.deleteEntry(matchedEntry.filename);	1✔
88
89	return null;	1✔
90	}
91
92	if (track.durationMS > 0) {	4✔
93	const expectedSize =
94	(track.durationMS / 1000) * EXPECTED_BYTES_PER_SECOND;	3✔
95	const minValidSize = expectedSize * MIN_CACHE_SIZE_RATIO;	3✔
96
97	if (stats.size < minValidSize) {	3✔
98	logger.warn(	1✔
99	{
100	filePath,
101	actualSize: stats.size,
102	expectedSize: Math.round(expectedSize),
103	durationMS: track.durationMS,
104	},
105	'Deleting corrupted cache file',
106	);
107
108	void opusCacheManager.deleteEntry(matchedEntry.filename);	1✔
109
110	track.setMetadata({	1✔
111	...(track.metadata ?? {}),	1!
112	cacheInvalidated: true,
113	});
114
115	return null;	1✔
116	}
117	}
118
119	track.setMetadata({	3✔
120	...(track.metadata ?? {}),	3!
121	isFromCache: true,
122	cacheFilename: matchedEntry.filename,
123	});
124
125	return createReadStream(filePath);	8✔
126	} catch {
127	return null;	1✔
128	}
129	});
130
131	onStreamExtracted(async (stream, track) => {	1✔
132	if (typeof stream === 'string') {	9✔
133	return stream;	1✔
134	}
135
136	const isReadable = stream instanceof Readable;	8✔
137	const readable = isReadable ? stream : stream.stream;	8✔
138
139	if (	9!
140	track.metadata &&	9!
141	typeof track.metadata === 'object' &&
142	'isFromCache' in track.metadata &&
143	track.metadata.isFromCache
144	) {
145	if (isReadable) {	×
146	return readable;	×
147	}
148
149	return stream;	×
150	}
151
152	const interceptor = new InterceptedStream();	8✔
153
154	const trackMetadata = {	8✔
155	title: track.cleanTitle,
156	author: track.author,
157	durationMS: track.durationMS,
158	};
159	const filename = opusCacheManager.generateFilename(trackMetadata);	8✔
160	const filePath = opusCacheManager.getFilePath(filename);	8✔
161
162	try {	8✔
163	activeWrites.add(filePath);	8✔
164
165	const writeStream = createWriteStream(filePath);	8✔
166
167	let streamEndedNormally = false;	8✔
168
169	const cleanup = () => {	8✔
170	interceptor.interceptors.delete(writeStream);	3✔
171
172	if (!writeStream.destroyed) {	3!
173	writeStream.destroy();	3✔
174	}
175
176	activeWrites.delete(filePath);	3✔
177	void opusCacheManager.deleteEntry(filename);	3✔
178	};
179
180	writeStream.on('error', (error) => {	8✔
181	logger.error(error, 'Opus cache write stream error');	1✔
182	cleanup();	1✔
183	});
184
185	writeStream.on('close', () => {	8✔
186	activeWrites.delete(filePath);	1✔
187
188	opusCacheManager.addEntry({	1✔
189	filename,
190	title: trackMetadata.title,
191	author: trackMetadata.author,
192	durationSeconds: Math.round(trackMetadata.durationMS / 1000),
193	});
194	});
195
196	readable.on('error', () => {	8✔
197	cleanup();	1✔
198	});
199
200	readable.on('end', () => {	8✔
201	streamEndedNormally = true;	×
202	});
203
204	readable.on('close', () => {	8✔
205	if (!streamEndedNormally && activeWrites.has(filePath)) {	1!
206	void cleanup();	1✔
207	}
208	});
209
210	interceptor.interceptors.add(writeStream);	8✔
211
212	if (isReadable) {	8✔
213	return readable.pipe(interceptor);	5✔
214	}
215
216	return {	1✔
217	stream: readable.pipe(interceptor),
218	$fmt: stream.$fmt,
219	};
220	} catch (error) {
221	activeWrites.delete(filePath);	2✔
222
223	logger.error(error, 'Failed to create opus cache write stream');	2✔
224
225	if (isReadable) {	2✔
226	return readable;	1✔
227	}
228
229	return stream;	1✔
230	}
231	});
232
233	/* v8 ignore start */
234	await initializedPlayer.extractors.register(YoutubeSabrExtractor, {});
235	await initializedPlayer.extractors.register(SpotifyExtractor, {
236	market: 'PL',
237	});
238
239	await initializedPlayer.extractors.loadMulti([
240	YoutubeSabrExtractor,
241	SpotifyExtractor,
242	]);
243	/* v8 ignore stop */
244	}
245
246	return initializedPlayer;	20✔
247	}

xxczaki / discord-bot / 21689294588

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous