File size: 56,823 Bytes
4e3b382 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 | // βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// CONSTITUENT SERVER
// Runs inside a HuggingFace Docker Space.
// Handles all HLS/FFmpeg streaming logic + constituent-specific APIs.
// Main web server communicates with this via HTTP only.
// config.json is auto-created on first boot storing the constituent owner id.
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const https = require('https');
const ffmpeg = require('fluent-ffmpeg');
const axios = require('axios');
const express = require('express');
const http = require('http');
const { Server: SocketIOServer } = require('socket.io');
const os = require('os');
// ββ Config ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// CONSTITUENT_OWNER_ID must be set as a HuggingFace Space secret/env var.
// It is the userId from your main database that "owns" this constituent.
const CONSTITUENT_OWNER_ID = process.env.CONSTITUENT_OWNER_ID;
const MAIN_SERVER_SECRET = process.env.MAIN_SERVER_SECRET || 'mysecretkeyforogudupaogeuwuwuhdg'; // shared secret to authenticate main server calls
const PORT = parseInt(process.env.PORT || '7860', 10);
const TMDB_KEY = process.env.TMDB_KEY || null;
const TMDB_BASE = 'https://api.themoviedb.org/3';
const TMDB_IMG = 'https://image.tmdb.org/t/p/w500';
if (!CONSTITUENT_OWNER_ID) {
console.error('CONSTITUENT_OWNER_ID env var is required. Set it as a HuggingFace Space secret.');
process.exit(1);
}
console.log(`π MAIN_SERVER_SECRET: ${process.env.MAIN_SERVER_SECRET ? 'loaded from env' : 'using built-in default'}`);
// ββ Auto-create config.json βββββββββββββββββββββββββββββββββββββββββββββββββββ
const CONFIG_PATH = path.join(__dirname, 'config.json');
let constituentConfig = {};
if (fs.existsSync(CONFIG_PATH)) {
try { constituentConfig = JSON.parse(fs.readFileSync(CONFIG_PATH, 'utf8')); }
catch { constituentConfig = {}; }
}
if (!constituentConfig.ownerId) {
constituentConfig.ownerId = CONSTITUENT_OWNER_ID;
constituentConfig.createdAt = new Date().toISOString();
fs.writeFileSync(CONFIG_PATH, JSON.stringify(constituentConfig, null, 2));
console.log(`β
config.json created for owner: ${CONSTITUENT_OWNER_ID}`);
}
// ββ Dirs & constants ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
const TEMP_DIR = path.join(__dirname, 'others', 'temp');
const SONGS_DIR = path.join(__dirname, 'others', 'songs');
const HLS_DIR = path.join(__dirname, 'others', 'hls');
const DATA_DIR = path.join(__dirname, 'others', 'data');
fs.mkdirSync(TEMP_DIR, { recursive: true });
fs.mkdirSync(SONGS_DIR, { recursive: true });
fs.mkdirSync(HLS_DIR, { recursive: true });
fs.mkdirSync(DATA_DIR, { recursive: true });
const SHOWPLAY_MAX_FILE_SIZE = 2 * 1024 * 1024 * 1024; // 2 GB
const SHOWPLAY_MAX_DURATION = 6 * 60 * 60; // 6 hrs
const MAX_FILE_SIZE = 50 * 1024 * 1024; // 50 MB (audio)
const MAX_DURATION = 15 * 60; // 15 min (audio)
const STREAM_CLEANUP_INTERVAL = 30 * 60 * 1000;
const DEFAULT_ARTWORK = 'https://touchio.vercel.app/tf14k0.jpeg';
const HLS_PLAYLIST_WINDOW = 6;
const HLS_MAX_SEGMENTS = 800;
// ββ SSL agent βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
const httpsAgentNoVerify = new https.Agent({ rejectUnauthorized: false });
axios.defaults.httpsAgent = httpsAgentNoVerify;
// ββ Express + Socket.IO βββββββββββββββββββββββββββββββββββββββββββββββββββββββ
const app = express();
const server = http.createServer(app);
const io = new SocketIOServer(server, {
cors: { origin: true, credentials: true, methods: ['GET', 'POST'] },
transports: ['websocket', 'polling']
});
app.use(express.json());
app.use('/hls', express.static(HLS_DIR, {
setHeaders: (res, filePath) => {
if (filePath.endsWith('.m3u8')) {
res.setHeader('Content-Type', 'application/vnd.apple.mpegurl');
res.setHeader('Cache-Control', 'no-cache, no-store');
res.setHeader('Access-Control-Allow-Origin', '*');
}
if (filePath.endsWith('.ts')) {
res.setHeader('Content-Type', 'video/MP2T');
res.setHeader('Cache-Control', 'public, max-age=3600');
res.setHeader('Access-Control-Allow-Origin', '*');
}
}
}));
app.use('/songs', express.static(SONGS_DIR));
// ββ In-memory streaming state βββββββββββββββββββββββββββββββββββββββββββββββββ
const streams = {};
const hlsState = {};
const hlsMutex = {};
const hlsGeneration = {};
const activeFFmpeg = {};
// ββ Auth middleware for main-server calls βββββββββββββββββββββββββββββββββββββ
function requireMainServer(req, res, next) {
const secret = req.headers['x-constituent-secret'];
if (!secret || secret !== MAIN_SERVER_SECRET) {
return res.status(403).json({ success: false, error: 'Forbidden: invalid or missing secret' });
}
next();
}
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// HEALTH / STATUS API
// Called by main server to check if this constituent is alive and ready.
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
app.get('/constituent/health', (req, res) => {
const totalMem = os.totalmem();
const freeMem = os.freemem();
const usedMem = totalMem - freeMem;
const cpuLoad = os.loadavg()[0]; // 1-min average
// Disk usage via df (Linux only β fine for HF Docker)
let diskTotal = null, diskUsed = null, diskFree = null;
try {
const { execSync } = require('child_process');
const dfOut = execSync("df -k / | tail -1").toString().trim().split(/\s+/);
diskTotal = parseInt(dfOut[1]) * 1024;
diskUsed = parseInt(dfOut[2]) * 1024;
diskFree = parseInt(dfOut[3]) * 1024;
} catch {}
const activeStreamCount = Object.keys(streams).filter(id => streams[id]?.isActive).length;
res.json({
success: true,
status: 'running',
ownerId: constituentConfig.ownerId,
createdAt: constituentConfig.createdAt,
uptime: process.uptime(),
memory: {
totalMB: Math.round(totalMem / 1024 / 1024),
usedMB: Math.round(usedMem / 1024 / 1024),
freeMB: Math.round(freeMem / 1024 / 1024),
usedPct: Math.round((usedMem / totalMem) * 100),
},
cpu: { loadAvg1min: cpuLoad.toFixed(2) },
disk: diskTotal ? {
totalGB: (diskTotal / 1024 ** 3).toFixed(1),
usedGB: (diskUsed / 1024 ** 3).toFixed(1),
freeGB: (diskFree / 1024 ** 3).toFixed(1),
usedPct: Math.round((diskUsed / diskTotal) * 100),
} : null,
streams: {
active: activeStreamCount,
total: Object.keys(streams).length,
},
});
});
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// SHOWPLAY API β search by title name (no raw link needed)
// Called by main server when a user (who owns this constituent) adds a movie or episode.
// Only the constituent's owner can trigger this.
//
// POST /constituent/add-movie β body: { streamId, title }
// Searches iktracks for the title, picks the first movie result, downloads it.
//
// POST /constituent/add-episode β body: { streamId, title, season, episode }
// Searches iktracks for the series, finds the matching S/E, downloads it.
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
const IKTRACKS_BASE = 'https://iktracks.vercel.app';
function spSeriesName(title) {
return (title || '').replace(/\s*\(?\d{4}\)?\s*$/, '').trim() || title;
}
function extractAllEpisodes(details) {
const allEps = [];
for (const season of (details.seasons || [])) {
for (const ep of (season.episodes || [])) {
if (ep && ep.downloadLink) {
allEps.push({ season: season.season, episode: ep.episode, downloadLink: ep.downloadLink });
}
}
}
return allEps;
}
app.post('/constituent/add-movie', requireMainServer, async (req, res) => {
// Supports two modes:
// 1. { streamId, movieLink, movieTitle, thumbnail?, tmdbInfo? } β direct link from server.js
// 2. { streamId, title } β search by name (legacy / direct constituent use)
const { streamId, movieLink, movieTitle, title: titleOnly, thumbnail, tmdbInfo } = req.body;
if (!streamId) {
return res.status(400).json({ success: false, error: 'streamId is required' });
}
if (streamId !== constituentConfig.ownerId) {
return res.status(403).json({ success: false, error: 'Only the constituent owner can add movies to this server' });
}
// ββ Mode 1: direct link provided βββββββββββββββββββββββββββββββββββββββββββ
if (movieLink && movieTitle) {
res.json({ success: true, message: 'Movie queued for download and encoding', streamId, title: movieTitle });
setImmediate(async () => {
try {
const result = await showplayEnqueueLink(streamId, movieLink, movieTitle, thumbnail || DEFAULT_ARTWORK, tmdbInfo || null);
console.log(`β
Movie added to stream ${streamId}: ${result.title}`);
} catch (err) {
console.error(`β Failed to add movie to stream ${streamId}:`, err.message);
}
});
return;
}
// ββ Mode 2: search by title ββββββββββββββββββββββββββββββββββββββββββββββββ
const title = titleOnly || movieTitle;
if (!title) {
return res.status(400).json({ success: false, error: 'Either (movieLink + movieTitle) or title is required' });
}
// Search for the title
let searchResults;
try {
const r = await axios.get(`${IKTRACKS_BASE}/search?query=${encodeURIComponent(title)}`, { timeout: 15000 });
searchResults = (r.data?.results || []).filter(r => r && r.link);
} catch (err) {
return res.status(500).json({ success: false, error: `Search failed: ${err.message}` });
}
if (!searchResults.length) {
return res.status(404).json({ success: false, error: `No results found for "${title}"` });
}
// Pick the first movie result (prefer type==='movie', fall back to first result)
const movieResult = searchResults.find(r => r.type === 'movie') || searchResults[0];
// Fetch details to get the download link
let details;
try {
const r = await axios.get(`${IKTRACKS_BASE}/details?url=${encodeURIComponent(movieResult.link)}`, { timeout: 15000 });
details = r.data;
if (!details) throw new Error('Empty details response');
} catch (err) {
return res.status(500).json({ success: false, error: `Details fetch failed: ${err.message}` });
}
if (details.type === 'series') {
return res.status(400).json({ success: false, error: 'This title is a series. Use /constituent/add-episode instead.' });
}
const link = details.downloadLinks?.[0]?.downloadLink;
if (!link) {
return res.status(404).json({ success: false, error: 'No download link found for this title' });
}
const pendingTitle = details.title || movieResult.title || title;
const pendingThumb = details.thumbnail || movieResult.thumbnail || DEFAULT_ARTWORK;
res.json({ success: true, message: 'Movie queued for download and encoding', streamId, title: pendingTitle });
setImmediate(async () => {
try {
const result = await showplayEnqueueLink(streamId, link, pendingTitle, pendingThumb, tmdbInfo || null);
console.log(`β
Movie added to stream ${streamId}: ${result.title}`);
} catch (err) {
console.error(`β Failed to add movie to stream ${streamId}:`, err.message);
}
});
});
// POST /constituent/add-episode β body: { streamId, title, season, episode }
app.post('/constituent/add-episode', requireMainServer, async (req, res) => {
const { streamId, title, season, episode } = req.body;
if (!streamId || !title) {
return res.status(400).json({ success: false, error: 'streamId and title are required' });
}
if (season == null || episode == null) {
return res.status(400).json({ success: false, error: 'season and episode are required' });
}
if (streamId !== constituentConfig.ownerId) {
return res.status(403).json({ success: false, error: 'Only the constituent owner can add episodes to this server' });
}
// Search for the series
let searchResults;
try {
const r = await axios.get(`${IKTRACKS_BASE}/search?query=${encodeURIComponent(title)}`, { timeout: 15000 });
searchResults = (r.data?.results || []).filter(r => r && r.link);
} catch (err) {
return res.status(500).json({ success: false, error: `Search failed: ${err.message}` });
}
if (!searchResults.length) {
return res.status(404).json({ success: false, error: `No results found for "${title}"` });
}
// Pick best series result
const seriesResult = searchResults.find(r => r.type === 'series') || searchResults[0];
// Fetch details
let details;
try {
const r = await axios.get(`${IKTRACKS_BASE}/details?url=${encodeURIComponent(seriesResult.link)}`, { timeout: 15000 });
details = r.data;
if (!details) throw new Error('Empty details response');
} catch (err) {
return res.status(500).json({ success: false, error: `Details fetch failed: ${err.message}` });
}
const allEps = extractAllEpisodes(details);
if (!allEps.length) {
return res.status(404).json({ success: false, error: 'No downloadable episodes found for this title' });
}
const ep = allEps.find(e => String(e.season) === String(season) && String(e.episode) === String(episode));
if (!ep) {
return res.status(404).json({ success: false, error: `Episode S${season}E${episode} not found` });
}
const seriesName = spSeriesName(details.title || seriesResult.title || title);
const epLabel = `S${String(ep.season).padStart(2,'0')} E${String(ep.episode).padStart(2,'0')}`;
const pendingTitle = `${seriesName} β’ ${epLabel}`;
const thumbnail = details.thumbnail || seriesResult.thumbnail || DEFAULT_ARTWORK;
res.json({ success: true, message: 'Episode queued for download and encoding', streamId, title: pendingTitle });
setImmediate(async () => {
try {
const result = await showplayEnqueueLink(streamId, ep.downloadLink, pendingTitle, thumbnail, null);
console.log(`β
Episode added to stream ${streamId}: ${result.title}`);
} catch (err) {
console.error(`β Failed to add episode to stream ${streamId}:`, err.message);
}
});
});
// POST /constituent/add-song β body: { streamId, songUrl, title, thumbnail? }
// Accepts a direct audio URL + title, downloads and enqueues without searching.
app.post('/constituent/add-song', requireMainServer, async (req, res) => {
const { streamId, songUrl, title, thumbnail } = req.body;
if (!streamId || !songUrl || !title) {
return res.status(400).json({ success: false, error: 'streamId, songUrl, and title are required' });
}
if (streamId !== constituentConfig.ownerId) {
return res.status(403).json({ success: false, error: 'Only the constituent owner can add songs to this server' });
}
res.json({ success: true, message: 'Song queued for download and encoding', streamId, title });
setImmediate(async () => {
try {
if (!streams[streamId]) {
streams[streamId] = { queue: [], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId: streamId, lastActivity: Date.now(), isActive: false };
}
// Download the audio
const fileName = crypto.randomUUID() + '.mp3';
const filePath = require('path').join(SONGS_DIR, fileName);
const writer = require('fs').createWriteStream(filePath);
const response = await axios({ url: songUrl, method: 'GET', responseType: 'stream', httpsAgent: httpsAgentNoVerify });
response.data.pipe(writer);
await new Promise((resolve, reject) => {
writer.on('finish', resolve);
writer.on('error', (e) => { writer.destroy(); reject(e); });
response.data.on('error', reject);
});
const mediaMeta = await getAudioMeta(filePath);
const songInfo = {
fileName,
meta: {
title,
thumbnail: thumbnail || DEFAULT_ARTWORK,
duration: mediaMeta.duration || 0,
views: 'N/A',
published: 'N/A',
source: songUrl,
videoUrl: null,
},
};
enqueueToStream(streamId, songInfo);
console.log(`β
Song added to stream ${streamId}: ${title}`);
} catch (err) {
console.error(`β Failed to add song to stream ${streamId}:`, err.message);
}
});
});
// βββ Queue status for a stream ββββββββββββββββββββββββββββββββββββββββββββββββ
app.get('/constituent/queue/:streamId', requireMainServer, (req, res) => {
const { streamId } = req.params;
const stream = streams[streamId];
if (!stream) return res.json({ success: true, streamId, queue: [], isActive: false });
const queue = stream.queue.map(s => ({
_sid: s._sid,
title: s.meta.title,
thumbnail: s.meta.thumbnail,
duration: s.meta.duration,
isVideo: !!s.meta.videoUrl,
hlsReady: !!(s._hlsPregened && typeof s._hlsStart === 'number'),
}));
res.json({
success: true,
streamId,
isActive: stream.isActive,
queue,
hlsUrl: stream.isActive ? `/stream-hls/${streamId}/live.m3u8` : null,
showplayInProgress: stream._showplayInProgress || 0,
});
});
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// HLS PLAYLIST ENDPOINT
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
app.get('/stream-hls/:streamId/live.m3u8', async (req, res) => {
const streamId = req.params.streamId;
const POLL_MS = 300;
const TIMEOUT_MS = 30000;
let waited = 0;
while (waited < TIMEOUT_MS) {
const state = hlsState[streamId];
if (state && state.segments.length > 0) break;
if (!streams[streamId]) return res.status(404).send('Stream not found');
if (state && !state.generating) {
return res.status(500).send('HLS generation failed');
}
await new Promise(r => setTimeout(r, POLL_MS));
waited += POLL_MS;
}
const state = hlsState[streamId];
if (!state || state.segments.length === 0) {
return res.status(503).set('Retry-After', '3').send('HLS generation timed out, retry shortly');
}
const stream = streams[streamId];
let elapsed = 0;
if (stream && stream.songStartTime) {
const current = stream.queue[0];
const withinSong = (Date.now() - stream.songStartTime) / 1000;
const hlsStart = (current && current._hlsStart !== undefined) ? current._hlsStart : (stream.streamTimeOffset || 0);
elapsed = hlsStart + withinSong;
}
pruneOldSegments(streamId, elapsed);
const playlist = buildLivePlaylistAt(streamId, elapsed);
if (!playlist) return res.status(503).set('Retry-After', '2').send('Segments not ready yet');
res.setHeader('Content-Type', 'application/vnd.apple.mpegurl');
res.setHeader('Cache-Control', 'no-cache, no-store');
res.setHeader('Access-Control-Allow-Origin', '*');
res.send(playlist);
});
// βββ Current track ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
app.get('/stream/:streamId/currentTrack', (req, res) => {
const streamId = req.params.streamId;
const stream = streams[streamId];
if (!stream) return res.status(404).json({ error: 'Stream not found' });
const current = stream.queue[0];
if (!current) return res.json({ queue: [], currentIndex: 0, elapsed: 0, withinSong: 0, hlsUrl: null });
const songDuration = (current.meta.duration > 0) ? current.meta.duration : (typeof current._hlsEnd === 'number' && typeof current._hlsStart === 'number') ? (current._hlsEnd - current._hlsStart) : 0;
const hlsStartOfSong = current._hlsStart !== undefined ? current._hlsStart : (stream.streamTimeOffset || 0);
const rawWithin = stream.songStartTime ? (Date.now() - stream.songStartTime) / 1000 : 0;
const withinSong = Math.max(0, Math.min(rawWithin, songDuration));
const elapsed = hlsStartOfSong + withinSong;
const hlsStateNow = hlsState[streamId];
const hlsReady = !!(hlsStateNow && hlsStateNow.segments.length > 0 && !hlsStateNow.generating);
res.json({ queue: stream.queue.map(s => ({ _sid: s._sid, meta: s.meta, tmdb: s.tmdb || null })), currentIndex: 0, elapsed, withinSong, streamTimeOffset: hlsStartOfSong, hlsUrl: `/stream-hls/${streamId}/live.m3u8`, isVideo: !!current.meta.videoUrl, hlsReady, songId: current._sid || null, tmdb: current.tmdb || null });
});
// βββ HLS status βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
app.get('/stream/:streamId/hlsStatus', (req, res) => {
const streamId = req.params.streamId;
const stream = streams[streamId];
if (!stream) return res.status(404).json({ error: 'Stream not found' });
const state = hlsState[streamId];
const current = stream.queue[0];
const generating = !!(state && state.generating);
const ready = !!(state && state.segments.length > 0);
res.json({ ready, generating, segmentsReady: ready, totalSegments: state ? state.segments.length : 0, currentSong: current ? current.meta.title : null, hlsUrl: ready ? `/stream-hls/${streamId}/live.m3u8` : null });
});
// POST /constituent/stop/:streamId β stop the stream entirely (owner only via main server)
app.post('/constituent/stop/:streamId', requireMainServer, (req, res) => {
const { streamId } = req.params;
const stream = streams[streamId];
if (!stream) return res.json({ success: true, message: 'No active stream' });
killActiveFFmpeg(streamId);
for (const song of stream.queue) {
const fp = path.join(SONGS_DIR, song.fileName);
if (fs.existsSync(fp)) { try { fs.unlinkSync(fp); } catch {} }
}
stream.queue = [];
stream.isActive = false;
stream.songStartTime = null;
stream.streamTimeOffset = 0;
if (hlsState[streamId]) {
const hlsStreamDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsStreamDir)) {
try { const files = fs.readdirSync(hlsStreamDir); for (const f of files) { try { fs.unlinkSync(path.join(hlsStreamDir, f)); } catch {} } } catch {}
}
delete hlsState[streamId]; delete hlsMutex[streamId];
}
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
io.to(`stream:${streamId}`).emit('message', { type: 'stream_ended', message: 'Stream stopped by owner.' });
res.json({ success: true, message: 'Stream stopped.' });
});
// POST /constituent/skip/:streamId
app.post('/constituent/skip/:streamId', requireMainServer, (req, res) => {
const { streamId } = req.params;
const stream = streams[streamId];
if (!stream) return res.status(404).json({ success: false, error: 'Stream not found' });
advanceToNextSong(streamId, false);
res.json({ success: true });
});
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// SOCKET.IO β real-time updates for stream viewers
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
io.on('connection', (socket) => {
const streamId = socket.handshake.query.streamId;
if (!streamId) { socket.emit('message', { type: 'error', message: 'streamId required' }); socket.disconnect(); return; }
socket.join(`stream:${streamId}`);
// Initialize stream entry if missing (e.g. adding media not started yet)
if (!streams[streamId]) {
streams[streamId] = { queue: [], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId: streamId, lastActivity: Date.now(), isActive: false };
}
const stream = streams[streamId];
stream.lastActivity = Date.now();
if (stream.queue.length > 0) sendStreamUpdate(streamId, socket);
// If media is being added, immediately notify this socket
if (stream._showplayInProgress) {
socket.emit('message', { type: 'showplay_progress', stage: 'processing', title: stream.queue[0]?.meta?.title || 'media' });
}
socket.on('join-stream', (data) => {
const sid = data?.streamId || streamId;
socket.join(`stream:${sid}`);
if (streams[sid]) {
streams[sid].lastActivity = Date.now();
sendStreamUpdate(sid, socket);
}
});
socket.on('heartbeat', (data) => {
const sid = data?.streamId || streamId;
if (streams[sid]) streams[sid].lastActivity = Date.now();
});
socket.on('disconnect', () => { console.log(`Socket disconnected from stream ${streamId}`); });
});
function sendStreamUpdate(streamId, specificSocket = null) {
const stream = streams[streamId];
if (!stream || (!stream.isActive && stream.queue.length === 0)) return;
const current = stream.queue[0];
if (!current) return;
const hlsStateNow = hlsState[streamId];
const hlsReady = !!(hlsStateNow && hlsStateNow.segments.length > 0 && !hlsStateNow.generating);
const songDuration = (current.meta.duration > 0) ? current.meta.duration : (typeof current._hlsEnd === 'number' && typeof current._hlsStart === 'number') ? (current._hlsEnd - current._hlsStart) : 0;
const hlsStartOfSong = current._hlsStart !== undefined ? current._hlsStart : (stream.streamTimeOffset || 0);
const rawWithin = stream.songStartTime ? (Date.now() - stream.songStartTime) / 1000 : 0;
const withinSong = Math.max(0, Math.min(rawWithin, songDuration));
const absoluteElapsed = hlsStartOfSong + withinSong;
const nextSong = stream.queue.length > 1 ? stream.queue[1] : null;
const payload = {
type: 'update',
elapsed: absoluteElapsed,
withinSong,
streamTimeOffset: hlsStartOfSong,
currentIndex: 0,
hlsReady,
current: { file: `/songs/${current.fileName}`, meta: current.meta, isVideo: !!current.meta.videoUrl, _sid: current._sid, tmdb: current.tmdb || null },
songId: current._sid,
next: nextSong ? { file: `/songs/${nextSong.fileName}`, meta: nextSong.meta, isVideo: !!nextSong.meta.videoUrl, tmdb: nextSong.tmdb || null } : null,
queue: stream.queue,
queueLength: stream.queue.length,
hlsUrl: `/stream-hls/${streamId}/live.m3u8`
};
if (specificSocket) specificSocket.emit('message', payload);
else io.to(`stream:${streamId}`).emit('message', payload);
}
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// HLS ENGINE (exact logic from main server)
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
function ensureHlsDir(streamId) {
const dir = path.join(HLS_DIR, streamId);
if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
return dir;
}
function killActiveFFmpeg(streamId) {
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
const cmd = activeFFmpeg[streamId];
if (cmd) {
try { cmd.kill('SIGKILL'); } catch {}
delete activeFFmpeg[streamId];
console.log(`πͺ FFmpeg killed for stream ${streamId}`);
}
hlsMutex[streamId] = Promise.resolve();
if (hlsState[streamId]) hlsState[streamId].generating = false;
}
function buildLivePlaylistAt(streamId, elapsed) {
const state = hlsState[streamId];
if (!state || !state.segments.length) return null;
const segs = state.segments;
let startIdx = -1;
for (let i = 0; i < segs.length; i++) {
if (segs[i].streamEnd > elapsed) { startIdx = i; break; }
}
if (startIdx === -1) return null;
const window = segs.slice(startIdx, startIdx + HLS_PLAYLIST_WINDOW);
const mediaSeq = state.mediaSeq + startIdx;
const lines = ['#EXTM3U','#EXT-X-VERSION:3','#EXT-X-TARGETDURATION:10',`#EXT-X-MEDIA-SEQUENCE:${mediaSeq}`];
let prevSid = null;
for (const seg of window) {
if (prevSid !== null && seg.ownerSid && seg.ownerSid !== prevSid) {
lines.push('#EXT-X-DISCONTINUITY');
}
prevSid = seg.ownerSid || prevSid;
lines.push(`#EXTINF:${seg.duration.toFixed(6)},`);
lines.push(seg.uri);
}
return lines.join('\n') + '\n';
}
function pruneOldSegments(streamId, elapsed) {
const state = hlsState[streamId];
if (!state) return;
const dropBefore = elapsed - HLS_PLAYLIST_WINDOW * 10 * 3;
let dropped = 0;
while (state.segments.length > HLS_MAX_SEGMENTS && state.segments[0].streamEnd < dropBefore) {
const seg = state.segments.shift();
dropped++;
const dir = ensureHlsDir(streamId);
const file = path.join(dir, path.basename(seg.uri));
try { if (fs.existsSync(file)) fs.unlinkSync(file); } catch {}
}
if (dropped > 0) console.log(`ποΈ Pruned ${dropped} segments for stream ${streamId}`);
}
function parseM3u8Durations(playlistPath) {
if (!fs.existsSync(playlistPath)) return [];
const lines = fs.readFileSync(playlistPath, 'utf8').split('\n');
const entries = [];
for (let i = 0; i < lines.length; i++) {
if (lines[i].startsWith('#EXTINF:')) {
const dur = parseFloat(lines[i].replace('#EXTINF:', ''));
const file = (lines[i + 1] || '').trim();
if (file && !file.startsWith('#')) entries.push({ file, dur });
}
}
return entries;
}
function watchForSegments(streamId, dir, segPrefix, songHlsStart, onFirstSeg, ownerSid, state) {
let cursor = songHlsStart, firstFlushed = false;
const stitched = new Set();
const playlistPath = path.join(dir, segPrefix + '.m3u8');
let pollCount = 0;
console.log(`π watchForSegments created: ownerSid=${ownerSid?.slice(0,8)} songHlsStart=${songHlsStart} playlistPath=${playlistPath}`);
const flush = () => {
pollCount++;
const entries = parseM3u8Durations(playlistPath);
if (pollCount <= 3 || entries.length > 0) {
console.log(`π watch poll #${pollCount} [${ownerSid?.slice(0,8)}]: playlist=${fs.existsSync(playlistPath)} entries=${entries.length} stitched=${stitched.size} firstFlushed=${firstFlushed}`);
}
for (const { file, dur } of entries) {
if (stitched.has(file)) continue;
const segPath = path.join(dir, file);
try { if (fs.statSync(segPath).size < 188) continue; } catch { continue; }
stitched.add(file);
const seg = { uri: `/hls/${streamId}/${file}`, _path: segPath, streamStart: cursor, streamEnd: cursor + dur, duration: dur, ownerSid };
cursor += dur;
state.segments.push(seg);
state.totalDuration = cursor;
if (!firstFlushed) {
firstFlushed = true;
state.generating = false;
if (streams[streamId]?.queue.length > 0) {
const q0 = streams[streamId].queue[0];
const sidMatch = ownerSid ? q0._sid === ownerSid : true;
const startMatch = typeof q0._hlsStart === 'number' && songHlsStart === q0._hlsStart;
console.log(`π Ownership check: sid=${q0._sid?.slice(0,8)}==${ownerSid?.slice(0,8)}:${sidMatch} hlsStart=${q0._hlsStart}==${songHlsStart}:${startMatch}`);
if (sidMatch && startMatch) {
streams[streamId].songStartTime = Date.now();
console.log(`β±οΈ songStartTime reset for "${q0.meta.title}" [${q0._sid}] (first segment ready)`);
} else if (sidMatch && q0._hlsStart === undefined) {
// brief window before _hlsStart is set β harmless
} else {
console.log(`β οΈ watchForSegments ownership mismatch β skipping songStartTime reset. watcher=[${ownerSid}@${songHlsStart}] queue[0]=[${q0._sid}@${q0._hlsStart}]`);
}
}
if (onFirstSeg) onFirstSeg();
}
}
};
let lastEntryCount = -1;
let stablePolls = 0;
const STABLE_NEEDED = 3;
const iv = setInterval(() => {
flush();
const entries = parseM3u8Durations(playlistPath);
if (entries.length === lastEntryCount && !activeFFmpeg[streamId]) {
stablePolls++;
if (stablePolls >= STABLE_NEEDED) {
console.log(`π watchForSegments auto-stop [${ownerSid?.slice(0,8)}]: stable for ${STABLE_NEEDED} polls, FFmpeg done`);
clearInterval(iv);
}
} else {
stablePolls = 0;
lastEntryCount = entries.length;
}
}, 800);
const markDone = () => { flush(); clearInterval(iv); return cursor; };
return { stop: () => clearInterval(iv), markDone };
}
async function generateSegmentsForSong(streamId, songInfo, isVideo, state) {
const dir = ensureHlsDir(streamId);
const songPath = path.join(SONGS_DIR, songInfo.fileName);
const segPrefix = `seg_${streamId}_${Date.now()}`;
console.log(`π¬ FFmpeg starting: ${songPath} isVideo=${isVideo}`);
if (!fs.existsSync(songPath)) throw new Error(`Source file missing: ${songPath}`);
const fileStat = fs.statSync(songPath);
if (fileStat.size === 0) throw new Error('Source file is empty');
console.log(`π Source file: ${(fileStat.size / 1024 / 1024).toFixed(1)}MB`);
const segPattern = path.join(dir, segPrefix + '_%03d.ts');
const playlistPath = path.join(dir, segPrefix + '.m3u8');
const songHlsStart = state.totalDuration;
console.log(`π― segPrefix=${segPrefix} songHlsStart=${songHlsStart} ownerSid=${songInfo._sid?.slice(0,8)}`);
return new Promise((resolve, reject) => {
const cmd = ffmpeg(songPath);
if (isVideo) {
cmd.outputOptions([
'-map','0:v:0','-map','0:a:0',
'-c:v','libx264','-preset','ultrafast','-crf','28',
'-profile:v','main','-level','3.1','-pix_fmt','yuv420p',
'-vf','scale=854:480',
'-c:a','aac','-b:a','128k',
'-f','segment','-segment_time','8',
'-segment_list',playlistPath,'-segment_list_flags','+live','-segment_format','mpegts',
]);
} else {
cmd.outputOptions(['-vn','-c:a','aac','-b:a','128k','-f','segment','-segment_time','8','-segment_list',playlistPath,'-segment_list_flags','+live','-segment_format','mpegts']);
}
let watcher = null;
cmd.output(segPattern)
.on('start', () => {
activeFFmpeg[streamId] = cmd;
console.log(`π¬ FFmpeg process started [${streamId}] gen=${hlsGeneration[streamId]}`);
watcher = watchForSegments(streamId, dir, segPrefix, songHlsStart, () => {
console.log(`β‘ First segment ready for stream ${streamId}`);
sendStreamUpdate(streamId);
}, songInfo._sid, state);
})
.on('stderr', line => {
if (line.includes('Error') || line.includes('error') || line.includes('Invalid')) {
console.error(`FFmpeg stderr: ${line}`);
}
})
.on('end', () => {
console.log(`β
FFmpeg done for ${streamId}`);
delete activeFFmpeg[streamId];
if (!watcher) { resolve(0); return; }
const finalCursor = watcher.markDone();
console.log(`π Final cursor from playlist: ${finalCursor.toFixed(3)}s`);
state.totalDuration = finalCursor;
try { fs.unlinkSync(playlistPath); } catch {}
resolve(finalCursor);
})
.on('error', (err) => {
console.log(`π₯ FFmpeg error for ${streamId}: ${err.message}`);
delete activeFFmpeg[streamId];
if (err.message && (err.message.includes('SIGKILL') || err.message.includes('killed'))) {
console.log(`β‘ FFmpeg killed cleanly for ${streamId} (skip)`);
if (watcher) watcher.stop();
resolve(0);
return;
}
console.error(`β FFmpeg error for ${streamId}:`, err.message);
if (watcher) watcher.stop();
reject(err);
})
.run();
});
}
async function appendSongToHls(streamId, songInfo) {
if (!hlsState[streamId]) {
hlsState[streamId] = { mediaSeq: 0, segments: [], totalDuration: 0, generating: true };
console.log(`π¦ appendSongToHls: created fresh hlsState for ${streamId}`);
}
const myGeneration = hlsGeneration[streamId] || 0;
const prev = hlsMutex[streamId] || Promise.resolve();
console.log(`π appendSongToHls queued: "${songInfo.meta.title}" [${songInfo._sid?.slice(0,8)}] gen=${myGeneration}`);
const next = prev.then(async () => {
const currentGen = hlsGeneration[streamId] || 0;
if (currentGen !== myGeneration) {
console.log(`β© Skipping stale appendSongToHls for "${songInfo.meta.title}" (gen ${myGeneration} vs ${currentGen})`);
return;
}
const isVideo = !!(songInfo.meta && songInfo.meta.videoUrl);
const state = hlsState[streamId];
if (!state) {
console.log(`β© Skipping appendSongToHls for "${songInfo.meta.title}" β hlsState gone`);
return;
}
state.generating = true;
songInfo._hlsStart = state.totalDuration;
console.log(`π _hlsStart set to ${songInfo._hlsStart.toFixed(2)}s for "${songInfo.meta.title}"`);
try {
const finalCursor = await generateSegmentsForSong(streamId, songInfo, isVideo, state);
if (typeof finalCursor === 'number' && finalCursor > 0) {
songInfo._hlsEnd = finalCursor;
const actualDuration = finalCursor - songInfo._hlsStart;
if (actualDuration > 0 && Math.abs(actualDuration - (songInfo.meta.duration || 0)) > 30) {
console.log(`π Correcting meta.duration for "${songInfo.meta.title}": ${(songInfo.meta.duration || 0).toFixed(1)}s β ${actualDuration.toFixed(1)}s`);
songInfo.meta.duration = actualDuration;
}
songInfo._hlsDurationTrusted = true;
} else {
songInfo._hlsEnd = state.totalDuration;
console.log(`β‘ Encode killed for "${songInfo.meta.title}" β hlsEnd set to ${songInfo._hlsEnd?.toFixed(2)}s`);
}
state.generating = false;
console.log(`πΊ HLS done for "${songInfo.meta.title}": hlsStart=${songInfo._hlsStart?.toFixed(2)}s hlsEnd=${songInfo._hlsEnd?.toFixed(2)}s segs=${state.segments.length}`);
const finalGen = hlsGeneration[streamId] || 0;
const liveStream = streams[streamId];
if (finalGen === myGeneration && liveStream && liveStream.queue[0]?._sid === songInfo._sid) {
if (!liveStream.songStartTime) {
liveStream.songStartTime = Date.now();
console.log(`β±οΈ songStartTime set post-encode for "${songInfo.meta.title}" [${songInfo._sid}]`);
sendStreamUpdate(streamId);
}
preGenerateNextSong(streamId).catch(console.error);
}
} catch (err) {
console.error(`HLS generation failed for stream ${streamId}:`, err);
if (hlsState[streamId]) hlsState[streamId].generating = false;
}
});
hlsMutex[streamId] = next;
return next;
}
async function preGenerateNextSong(streamId) {
const stream = streams[streamId];
if (!stream || stream.queue.length < 2) return;
const nextSong = stream.queue[1];
if (!nextSong || nextSong._hlsPregened || nextSong._hlsPregenInProgress) return;
nextSong._hlsPregenInProgress = true;
const sid = nextSong._sid;
console.log(`π Pre-generating HLS for next: ${nextSong.meta.title} [${sid}]`);
try {
await appendSongToHls(streamId, nextSong);
} catch (err) {
nextSong._hlsPregenInProgress = false;
console.error(`Pre-gen failed for "${nextSong.meta.title}":`, err.message);
return;
}
const streamNow = streams[streamId];
const stillQueued = streamNow && streamNow.queue.some(s => s._sid === sid);
const encodingFinished = typeof nextSong._hlsEnd === 'number' && typeof nextSong._hlsStart === 'number' && nextSong._hlsEnd > nextSong._hlsStart;
if (stillQueued && encodingFinished) {
nextSong._hlsPregened = true;
console.log(`β
Pre-gen confirmed for "${nextSong.meta.title}" [${sid}]: hlsStart=${nextSong._hlsStart.toFixed(2)}s hlsEnd=${nextSong._hlsEnd.toFixed(2)}s`);
} else {
nextSong._hlsPregened = false;
nextSong._hlsPregenInProgress = false;
delete nextSong._hlsStart;
delete nextSong._hlsEnd;
console.log(`β οΈ Pre-gen invalidated for "${nextSong.meta.title}" [${sid}]`);
}
}
function advanceToNextSong(streamId, autoAdvance = false) {
const stream = streams[streamId];
if (!stream) return false;
if (autoAdvance) stream._notifyOnStart = true;
else delete stream._notifyOnStart;
killActiveFFmpeg(streamId);
const finishedSong = stream.queue.shift();
const filePath = path.join(SONGS_DIR, finishedSong.fileName);
if (fs.existsSync(filePath)) { try { fs.unlinkSync(filePath); } catch {} }
if (stream.queue.length === 0) {
stream.isActive = false;
stream.streamTimeOffset = 0;
stream.songStartTime = null;
if (hlsState[streamId]) {
const hlsDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsDir)) {
try { const files = fs.readdirSync(hlsDir); for (const f of files) { try { fs.unlinkSync(path.join(hlsDir, f)); } catch {} } } catch {}
}
delete hlsState[streamId]; delete hlsMutex[streamId];
}
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
console.log(`π Stream ${streamId} queue empty β HLS state reset for fresh start`);
io.to(`stream:${streamId}`).emit('message', { type: 'stream_ended', message: 'Queue is empty.' });
return false;
}
const nextSong = stream.queue[0];
const pregenIsValid = nextSong._hlsPregened && typeof nextSong._hlsStart === 'number' && typeof nextSong._hlsEnd === 'number' && nextSong._hlsEnd > nextSong._hlsStart;
if (pregenIsValid) {
stream.streamTimeOffset = nextSong._hlsStart;
stream.songStartTime = Date.now();
stream.lastActivity = Date.now();
stream.isActive = true;
delete stream._notifyOnStart;
sendStreamUpdate(streamId);
preGenerateNextSong(streamId).catch(console.error);
} else {
nextSong._hlsPregened = nextSong._hlsPregenInProgress = false;
delete nextSong._hlsStart; delete nextSong._hlsEnd;
if (hlsState[streamId]) {
const hlsDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsDir)) {
try { const files = fs.readdirSync(hlsDir); for (const f of files) { try { fs.unlinkSync(path.join(hlsDir, f)); } catch {} } } catch {}
}
delete hlsState[streamId]; delete hlsMutex[streamId];
}
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
stream.songStartTime = null;
stream.lastActivity = Date.now();
stream.isActive = true;
appendSongToHls(streamId, nextSong).then(() => {
sendStreamUpdate(streamId);
preGenerateNextSong(streamId).catch(console.error);
}).catch(console.error);
}
return true;
}
function enqueueToStream(streamId, songInfo) {
if (!streams[streamId]) {
songInfo._sid = crypto.randomUUID();
streams[streamId] = { queue: [songInfo], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId: streamId, lastActivity: Date.now(), isActive: true };
appendSongToHls(streamId, songInfo).then(() => sendStreamUpdate(streamId)).catch(console.error);
return { songInfo, position: 1, started: true };
}
const stream = streams[streamId];
songInfo._sid = crypto.randomUUID();
stream.queue.push(songInfo);
stream.lastActivity = Date.now();
const position = stream.queue.length;
if (!stream.isActive && position === 1 && !stream._showplayInProgress) {
// Stream was idle/ended β ensure HLS state is fresh so this song starts at t=0.
if (!hlsState[streamId] || hlsState[streamId].totalDuration > 0) {
if (hlsState[streamId]) {
const hlsDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsDir)) {
try { const files = fs.readdirSync(hlsDir); for (const f of files) { try { fs.unlinkSync(path.join(hlsDir, f)); } catch {} } } catch {}
}
delete hlsState[streamId]; delete hlsMutex[streamId];
}
hlsGeneration[streamId] = (hlsGeneration[streamId] || 0) + 1;
}
stream.streamTimeOffset = 0;
stream.songStartTime = null;
stream.isActive = true;
appendSongToHls(streamId, songInfo).then(() => sendStreamUpdate(streamId)).catch(console.error);
return { songInfo, position, started: true };
}
if (stream.isActive && position >= 2) preGenerateNextSong(streamId).catch(console.error);
sendStreamUpdate(streamId);
return { songInfo, position, started: false };
}
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// DOWNLOAD HELPERS
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
const DIRECT_VIDEO_EXTS = /\.(mkv|mp4|mov|avi|webm|m4v|flv|wmv|ts)(\?.*)?$/i;
function isDirectVideoUrl(url) {
if (!url) return false;
try { return DIRECT_VIDEO_EXTS.test(new URL(url).pathname); } catch { return DIRECT_VIDEO_EXTS.test(url); }
}
function getAudioMeta(filePath) {
return new Promise((resolve, reject) => {
ffmpeg.ffprobe(filePath, (err, metadata) => {
if (err) return reject(err);
function parseDurationTag(tag) {
if (!tag || typeof tag !== 'string') return 0;
const m = tag.match(/^(\d+):(\d+):(\d+(?:\.\d+)?)$/);
if (!m) return 0;
return parseInt(m[1], 10) * 3600 + parseInt(m[2], 10) * 60 + parseFloat(m[3]);
}
const candidates = [
parseFloat(metadata.format?.duration) || 0,
...(metadata.streams || []).flatMap(s => [
parseFloat(s.duration) || 0,
parseDurationTag(s.tags?.DURATION),
parseDurationTag(s.tags?.duration),
]),
];
const duration = Math.max(...candidates.filter(n => isFinite(n) && n > 0), 0);
resolve({ duration, size: metadata.format.size, bit_rate: metadata.format.bit_rate });
});
});
}
async function downloadVideoFile(downloadUrl) {
const fileName = crypto.randomUUID() + '.mp4';
const filePath = path.join(SONGS_DIR, fileName);
const writer = fs.createWriteStream(filePath);
try {
const response = await axios({ url: downloadUrl, method: 'GET', responseType: 'stream', httpsAgent: httpsAgentNoVerify });
const contentLength = parseInt(response.headers['content-length'] || '0', 10);
let bytesWritten = 0;
response.data.on('data', chunk => { bytesWritten += chunk.length; });
response.data.pipe(writer);
await new Promise((resolve, reject) => {
writer.on('finish', resolve);
writer.on('error', reject);
response.data.on('error', reject);
});
if (contentLength > 0 && bytesWritten < contentLength * 0.95) {
throw new Error(`Download truncated: got ${bytesWritten} of ${contentLength} bytes`);
}
} catch (err) {
writer.destroy();
if (fs.existsSync(filePath)) { try { fs.unlinkSync(filePath); } catch {} }
throw err;
}
return { fileName, filePath };
}
async function showplayEnqueueLink(streamId, pendingLink, pendingTitle, thumbnail, tmdbInfo = null) {
if (!streams[streamId]) {
streams[streamId] = { queue: [], songStartTime: null, streamTimeOffset: 0, users: new Map(), ownerId: streamId, lastActivity: Date.now(), isActive: false };
}
streams[streamId]._showplayInProgress = (streams[streamId]._showplayInProgress || 0) + 1;
// Notify listeners that download is starting
io.to(`stream:${streamId}`).emit('message', { type: 'showplay_progress', stage: 'downloading', title: pendingTitle });
console.log(`π₯ Downloading: ${pendingTitle}`);
let directUrl;
if (isDirectVideoUrl(pendingLink)) {
directUrl = pendingLink;
} else {
let extractRes;
try {
extractRes = await axios.get(`https://downw.vercel.app/extract?url=${encodeURIComponent(pendingLink)}`, { timeout: 60000, httpsAgent: httpsAgentNoVerify });
} catch (err) { throw new Error(`Extract API failed: ${err.message}`); }
directUrl = extractRes.data?.downloadUrl;
if (!directUrl) throw new Error('No download URL returned by extractor');
}
let fileName, filePath;
try { ({ fileName, filePath } = await downloadVideoFile(directUrl)); }
catch (err) { throw new Error(`Download failed: ${err.message}`); }
// Notify listeners that encoding is starting
io.to(`stream:${streamId}`).emit('message', { type: 'showplay_progress', stage: 'encoding', title: pendingTitle });
console.log(`βοΈ Encoding: ${pendingTitle}`);
let mediaMeta;
try { mediaMeta = await getAudioMeta(filePath); }
catch (e) { try { fs.unlinkSync(filePath); } catch {} throw new Error('ffprobe could not read the video file'); }
if (mediaMeta.size > SHOWPLAY_MAX_FILE_SIZE) {
try { fs.unlinkSync(filePath); } catch {}
throw new Error(`File too large (${(mediaMeta.size / (1024 ** 3)).toFixed(2)} GB). Max 2 GB.`);
}
if (mediaMeta.duration > SHOWPLAY_MAX_DURATION) {
try { fs.unlinkSync(filePath); } catch {}
throw new Error(`Video too long. Max 6 hours.`);
}
if (streams[streamId]) streams[streamId]._showplayInProgress = Math.max(0, (streams[streamId]._showplayInProgress || 1) - 1);
const effectivePoster = tmdbInfo?.poster || thumbnail || DEFAULT_ARTWORK;
const songInfo = {
fileName,
meta: {
title: tmdbInfo?.title || pendingTitle || 'Unknown',
thumbnail: effectivePoster,
duration: mediaMeta.duration || 0,
views: 'N/A',
published: tmdbInfo?.releaseDate || 'N/A',
source: pendingLink,
videoUrl: pendingLink || 'showplay',
},
tmdb: tmdbInfo || null,
isShowplay: true,
};
enqueueToStream(streamId, songInfo);
return { title: songInfo.meta.title, duration: mediaMeta.duration, thumbnail: effectivePoster };
}
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// BACKGROUND TIMERS
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// Auto-advance
setInterval(async () => {
try {
for (const streamId in streams) {
const stream = streams[streamId];
if (!stream.isActive) continue;
const current = stream.queue[0];
if (!current) continue;
let songDuration;
if (current._hlsDurationTrusted && typeof current._hlsEnd === 'number' && typeof current._hlsStart === 'number') {
songDuration = current._hlsEnd - current._hlsStart;
} else if (current.meta.duration > 0) {
songDuration = current.meta.duration;
} else if (typeof current._hlsEnd === 'number' && typeof current._hlsStart === 'number') {
songDuration = current._hlsEnd - current._hlsStart;
} else continue;
if (songDuration < 5 || !stream.songStartTime) continue;
const elapsed = (Date.now() - stream.songStartTime) / 1000;
if (elapsed >= songDuration + 3) {
if (stream._advancingFromSid === current._sid) continue;
stream._advancingFromSid = current._sid;
console.log(`βοΈ Auto-advance "${current.meta.title}": elapsed=${elapsed.toFixed(1)}s duration=${songDuration.toFixed(1)}s`);
advanceToNextSong(streamId, true);
if (stream._advancingFromSid === current._sid) delete stream._advancingFromSid;
}
}
} catch (err) { console.error('Auto-advance error:', err); }
}, 1000);
// Segment pruning
setInterval(() => {
for (const streamId in streams) {
const stream = streams[streamId];
if (!stream.isActive || !stream.songStartTime) continue;
const current = stream.queue[0];
if (!current) continue;
const hlsStart = current._hlsStart !== undefined ? current._hlsStart : (stream.streamTimeOffset || 0);
const withinSong = (Date.now() - stream.songStartTime) / 1000;
pruneOldSegments(streamId, hlsStart + withinSong);
}
}, 30 * 1000);
// Inactivity cleanup
setInterval(() => {
const now = Date.now(), toDelete = [];
for (const streamId in streams) {
const stream = streams[streamId];
if ((!stream.users || stream.users.size === 0) && (now - (stream.lastActivity || 0)) > STREAM_CLEANUP_INTERVAL) {
toDelete.push(streamId);
}
}
for (const streamId of toDelete) {
const stream = streams[streamId];
killActiveFFmpeg(streamId);
for (const song of stream.queue) { const fp = path.join(SONGS_DIR, song.fileName); if (fs.existsSync(fp)) { try { fs.unlinkSync(fp); } catch {} } }
const hlsStreamDir = path.join(HLS_DIR, streamId);
if (fs.existsSync(hlsStreamDir)) { try { fs.rmSync(hlsStreamDir, { recursive: true }); } catch {} }
delete streams[streamId]; delete hlsState[streamId]; delete hlsMutex[streamId]; delete hlsGeneration[streamId];
console.log(`π§Ή Cleaned up stream: ${streamId}`);
}
}, 10 * 60 * 1000);
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// LAUNCH
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
server.listen(PORT, async () => {
console.log(`π Constituent server running on port ${PORT}`);
console.log(`π€ Owner ID: ${CONSTITUENT_OWNER_ID}`);
try {
const ipRes = await axios.get('https://api.ipify.org?format=json', { timeout: 5000 });
console.log(`π Public IP: ${ipRes.data.ip}`);
} catch {}
});
process.once('SIGINT', () => { server.close(); process.exit(0); });
process.once('SIGTERM', () => { server.close(); process.exit(0); }); |