Skip to content

Commit a6069f4

Browse files
wukkowukko
committed
api & web: merge base queue ui & api updates
2 parents 50db4d3 + 45e7b69 commit a6069f4

28 files changed

+1047
-156
lines changed

‎api/src/core/api.js‎

Lines changed: 15 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -8,18 +8,19 @@ import jwt from "../security/jwt.js";
88
import stream from "../stream/stream.js";
99
import match from "../processing/match.js";
1010

11-
import { env, isCluster, setTunnelPort } from "../config.js";
11+
import { env } from "../config.js";
1212
import { extract } from "../processing/url.js";
13-
import { Green, Bright, Cyan } from "../misc/console-text.js";
13+
import { Bright, Cyan } from "../misc/console-text.js";
1414
import { hashHmac } from "../security/secrets.js";
1515
import { createStore } from "../store/redis-ratelimit.js";
1616
import { randomizeCiphers } from "../misc/randomize-ciphers.js";
1717
import { verifyTurnstileToken } from "../security/turnstile.js";
1818
import { friendlyServiceName } from "../processing/service-alias.js";
19-
import { verifyStream, getInternalStream } from "../stream/manage.js";
19+
import { verifyStream } from "../stream/manage.js";
2020
import { createResponse, normalizeRequest, getIP } from "../processing/request.js";
2121
import * as APIKeys from "../security/api-keys.js";
2222
import * as Cookies from "../processing/cookie/manager.js";
23+
import { setupTunnelHandler } from "./itunnel.js";
2324

2425
const git = {
2526
branch: await getBranch(),
@@ -263,6 +264,15 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
263264
}
264265
})
265266

267+
app.use('/tunnel', cors({
268+
methods: ['GET'],
269+
exposedHeaders: [
270+
'Estimated-Content-Length',
271+
'Content-Disposition'
272+
],
273+
...corsConfig,
274+
}));
275+
266276
app.get('/tunnel', apiTunnelLimiter, async (req, res) => {
267277
const id = String(req.query.id);
268278
const exp = String(req.query.exp);
@@ -292,31 +302,7 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
292302
}
293303

294304
return stream(res, streamInfo);
295-
})
296-
297-
const itunnelHandler = (req, res) => {
298-
if (!req.ip.endsWith('127.0.0.1')) {
299-
return res.sendStatus(403);
300-
}
301-
302-
if (String(req.query.id).length !== 21) {
303-
return res.sendStatus(400);
304-
}
305-
306-
const streamInfo = getInternalStream(req.query.id);
307-
if (!streamInfo) {
308-
return res.sendStatus(404);
309-
}
310-
311-
streamInfo.headers = new Map([
312-
...(streamInfo.headers || []),
313-
...Object.entries(req.headers)
314-
]);
315-
316-
return stream(res, { type: 'internal', ...streamInfo });
317-
};
318-
319-
app.get('/itunnel', itunnelHandler);
305+
});
320306

321307
app.get('/', (_, res) => {
322308
res.type('json');
@@ -378,17 +364,5 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
378364
}
379365
});
380366

381-
if (isCluster) {
382-
const istreamer = express();
383-
istreamer.get('/itunnel', itunnelHandler);
384-
const server = istreamer.listen({
385-
port: 0,
386-
host: '127.0.0.1',
387-
exclusive: true
388-
}, () => {
389-
const { port } = server.address();
390-
console.log(`${Green('[✓]')} cobalt sub-instance running on 127.0.0.1:${port}`);
391-
setTunnelPort(port);
392-
});
393-
}
367+
setupTunnelHandler();
394368
}

‎api/src/core/itunnel.js‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
import stream from "../stream/stream.js";
2+
import { getInternalTunnel } from "../stream/manage.js";
3+
import { setTunnelPort } from "../config.js";
4+
import { Green } from "../misc/console-text.js";
5+
import express from "express";
6+
7+
const validateTunnel = (req, res) => {
8+
if (!req.ip.endsWith('127.0.0.1')) {
9+
res.sendStatus(403);
10+
return;
11+
}
12+
13+
if (String(req.query.id).length !== 21) {
14+
res.sendStatus(400);
15+
return;
16+
}
17+
18+
const streamInfo = getInternalTunnel(req.query.id);
19+
if (!streamInfo) {
20+
res.sendStatus(404);
21+
return;
22+
}
23+
24+
return streamInfo;
25+
}
26+
27+
const streamTunnel = (req, res) => {
28+
const streamInfo = validateTunnel(req, res);
29+
if (!streamInfo) {
30+
return;
31+
}
32+
33+
streamInfo.headers = new Map([
34+
...(streamInfo.headers || []),
35+
...Object.entries(req.headers)
36+
]);
37+
38+
return stream(res, { type: 'internal', ...streamInfo });
39+
}
40+
41+
export const setupTunnelHandler = () => {
42+
const tunnelHandler = express();
43+
44+
tunnelHandler.get('/itunnel', streamTunnel);
45+
46+
// fallback
47+
tunnelHandler.use((_, res) => res.sendStatus(400));
48+
// error handler
49+
tunnelHandler.use((_, __, res, ____) => res.socket.end());
50+
51+
52+
const server = tunnelHandler.listen({
53+
port: 0,
54+
host: '127.0.0.1',
55+
exclusive: true
56+
}, () => {
57+
const { port } = server.address();
58+
console.log(`${Green('[✓]')} internal tunnel handler running on 127.0.0.1:${port}`);
59+
setTunnelPort(port);
60+
});
61+
}

‎api/src/processing/services/bilibili.js‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,8 @@ async function com_download(id) {
5858
return {
5959
urls: [video.baseUrl, audio.baseUrl],
6060
audioFilename: `bilibili_${id}_audio`,
61-
filename: `bilibili_${id}_${video.width}x${video.height}.mp4`
61+
filename: `bilibili_${id}_${video.width}x${video.height}.mp4`,
62+
isHLS: true
6263
};
6364
}
6465

‎api/src/stream/internal-hls.js‎

Lines changed: 62 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import HLS from "hls-parser";
22
import { createInternalStream } from "./manage.js";
3+
import { request } from "undici";
34

45
function getURL(url) {
56
try {
@@ -55,8 +56,11 @@ function transformMediaPlaylist(streamInfo, hlsPlaylist) {
5556

5657
const HLS_MIME_TYPES = ["application/vnd.apple.mpegurl", "audio/mpegurl", "application/x-mpegURL"];
5758

58-
export function isHlsResponse (req) {
59-
return HLS_MIME_TYPES.includes(req.headers['content-type']);
59+
export function isHlsResponse(req, streamInfo) {
60+
return HLS_MIME_TYPES.includes(req.headers['content-type'])
61+
// bluesky's cdn responds with wrong content-type for the hls playlist,
62+
// so we enforce it here until they fix it
63+
|| (streamInfo.service === 'bsky' && streamInfo.url.endsWith('.m3u8'));
6064
}
6165

6266
export async function handleHlsPlaylist(streamInfo, req, res) {
@@ -71,3 +75,59 @@ export async function handleHlsPlaylist(streamInfo, req, res) {
7175

7276
res.send(hlsPlaylist);
7377
}
78+
79+
async function getSegmentSize(url, config) {
80+
const segmentResponse = await request(url, {
81+
...config,
82+
throwOnError: true
83+
});
84+
85+
if (segmentResponse.headers['content-length']) {
86+
segmentResponse.body.dump();
87+
return +segmentResponse.headers['content-length'];
88+
}
89+
90+
// if the response does not have a content-length
91+
// header, we have to compute it ourselves
92+
let size = 0;
93+
94+
for await (const data of segmentResponse.body) {
95+
size += data.length;
96+
}
97+
98+
return size;
99+
}
100+
101+
export async function probeInternalHLSTunnel(streamInfo) {
102+
const { url, headers, dispatcher, signal } = streamInfo;
103+
104+
// remove all falsy headers
105+
Object.keys(headers).forEach(key => {
106+
if (!headers[key]) delete headers[key];
107+
});
108+
109+
const config = { headers, dispatcher, signal, maxRedirections: 16 };
110+
111+
const manifestResponse = await fetch(url, config);
112+
113+
const manifest = HLS.parse(await manifestResponse.text());
114+
if (manifest.segments.length === 0)
115+
return -1;
116+
117+
const segmentSamples = await Promise.all(
118+
Array(5).fill().map(async () => {
119+
const manifestIdx = Math.floor(Math.random() * manifest.segments.length);
120+
const randomSegment = manifest.segments[manifestIdx];
121+
if (!randomSegment.uri)
122+
throw "segment is missing URI";
123+
124+
const segmentSize = await getSegmentSize(randomSegment.uri, config) / randomSegment.duration;
125+
return segmentSize;
126+
})
127+
);
128+
129+
const averageBitrate = segmentSamples.reduce((a, b) => a + b) / segmentSamples.length;
130+
const totalDuration = manifest.segments.reduce((acc, segment) => acc + segment.duration, 0);
131+
132+
return averageBitrate * totalDuration;
133+
}

‎api/src/stream/internal.js‎

Lines changed: 39 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
import { request } from "undici";
22
import { Readable } from "node:stream";
33
import { closeRequest, getHeaders, pipe } from "./shared.js";
4-
import { handleHlsPlaylist, isHlsResponse } from "./internal-hls.js";
4+
import { handleHlsPlaylist, isHlsResponse, probeInternalHLSTunnel } from "./internal-hls.js";
55

66
const CHUNK_SIZE = BigInt(8e6); // 8 MB
77
const min = (a, b) => a < b ? a : b;
@@ -96,10 +96,7 @@ async function handleGenericStream(streamInfo, res) {
9696
res.status(fileResponse.statusCode);
9797
fileResponse.body.on('error', () => {});
9898

99-
// bluesky's cdn responds with wrong content-type for the hls playlist,
100-
// so we enforce it here until they fix it
101-
const isHls = isHlsResponse(fileResponse)
102-
|| (streamInfo.service === "bsky" && streamInfo.url.endsWith('.m3u8'));
99+
const isHls = isHlsResponse(fileResponse, streamInfo);
103100

104101
for (const [ name, value ] of Object.entries(fileResponse.headers)) {
105102
if (!isHls || name.toLowerCase() !== 'content-length') {
@@ -133,3 +130,40 @@ export function internalStream(streamInfo, res) {
133130

134131
return handleGenericStream(streamInfo, res);
135132
}
133+
134+
export async function probeInternalTunnel(streamInfo) {
135+
try {
136+
const signal = AbortSignal.timeout(3000);
137+
const headers = {
138+
...Object.fromEntries(streamInfo.headers || []),
139+
...getHeaders(streamInfo.service),
140+
host: undefined,
141+
range: undefined
142+
};
143+
144+
if (streamInfo.isHLS) {
145+
return probeInternalHLSTunnel({
146+
...streamInfo,
147+
signal,
148+
headers
149+
});
150+
}
151+
152+
const response = await request(streamInfo.url, {
153+
method: 'HEAD',
154+
headers,
155+
dispatcher: streamInfo.dispatcher,
156+
signal,
157+
maxRedirections: 16
158+
});
159+
160+
if (response.statusCode !== 200)
161+
throw "status is not 200 OK";
162+
163+
const size = +response.headers['content-length'];
164+
if (isNaN(size))
165+
throw "content-length is not a number";
166+
167+
return size;
168+
} catch {}
169+
}

‎api/src/stream/manage.js‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,10 +68,20 @@ export function createStream(obj) {
6868
return streamLink.toString();
6969
}
7070

71-
export function getInternalStream(id) {
71+
export function getInternalTunnel(id) {
7272
return internalStreamCache.get(id);
7373
}
7474

75+
export function getInternalTunnelFromURL(url) {
76+
url = new URL(url);
77+
if (url.hostname !== '127.0.0.1') {
78+
return;
79+
}
80+
81+
const id = url.searchParams.get('id');
82+
return getInternalTunnel(id);
83+
}
84+
7585
export function createInternalStream(url, obj = {}) {
7686
assert(typeof url === 'string');
7787

@@ -124,7 +134,7 @@ export function destroyInternalStream(url) {
124134
const id = url.searchParams.get('id');
125135

126136
if (internalStreamCache.has(id)) {
127-
closeRequest(getInternalStream(id)?.controller);
137+
closeRequest(getInternalTunnel(id)?.controller);
128138
internalStreamCache.delete(id);
129139
}
130140
}

‎api/src/stream/shared.js‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
import { genericUserAgent } from "../config.js";
22
import { vkClientAgent } from "../processing/services/vk.js";
3+
import { getInternalTunnelFromURL } from "./manage.js";
4+
import { probeInternalTunnel } from "./internal.js";
35

46
const defaultHeaders = {
57
'user-agent': genericUserAgent
@@ -47,3 +49,40 @@ export function pipe(from, to, done) {
4749

4850
from.pipe(to);
4951
}
52+
53+
export async function estimateTunnelLength(streamInfo, multiplier = 1.1) {
54+
let urls = streamInfo.urls;
55+
if (!Array.isArray(urls)) {
56+
urls = [ urls ];
57+
}
58+
59+
const internalTunnels = urls.map(getInternalTunnelFromURL);
60+
if (internalTunnels.some(t => !t))
61+
return -1;
62+
63+
const sizes = await Promise.all(internalTunnels.map(probeInternalTunnel));
64+
const estimatedSize = sizes.reduce(
65+
// if one of the sizes is missing, let's just make a very
66+
// bold guess that it's the same size as the existing one
67+
(acc, cur) => cur <= 0 ? acc * 2 : acc + cur,
68+
0
69+
);
70+
71+
if (isNaN(estimatedSize) || estimatedSize <= 0) {
72+
return -1;
73+
}
74+
75+
return Math.floor(estimatedSize * multiplier);
76+
}
77+
78+
export function estimateAudioMultiplier(streamInfo) {
79+
if (streamInfo.audioFormat === 'wav') {
80+
return 1411 / 128;
81+
}
82+
83+
if (streamInfo.audioCopy) {
84+
return 1;
85+
}
86+
87+
return streamInfo.audioBitrate / 128;
88+
}

0 commit comments

Comments
 (0)