Files
bitvid-seeder/server.js
T
2025-02-22 19:53:16 -05:00

487 lines
15 KiB
JavaScript

// server.js
import { EventEmitter } from "events";
EventEmitter.defaultMaxListeners = 50;
import express from "express";
import WebTorrent from "webtorrent";
import fs from "fs";
import { promises as fsPromises } from "fs";
import path from "path";
import { fileURLToPath } from "url";
import rateLimit from "express-rate-limit";
import winston from "winston";
import { exec } from "child_process";
// Set up Winston logger (debug level)
const logger = winston.createLogger({
level: "debug",
format: winston.format.combine(
winston.format.timestamp(),
winston.format.printf(
({ timestamp, level, message }) => `${timestamp} ${level}: ${message}`
)
),
transports: [new winston.transports.Console()],
});
const app = express();
const port = process.env.PORT || 3000;
const client = new WebTorrent();
client.setMaxListeners(50);
// ES module __dirname workaround
const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);
// Rate limiting middleware
const limiter = rateLimit({
windowMs: 60 * 1000,
max: 100,
});
app.use(limiter);
app.use(express.json());
// --- Global Variables and Helper Functions ---
// Configuration
const CONFIG_FILE = path.join(__dirname, "config.json");
let config = { storageDir: path.join(__dirname, "downloads") };
// Torrents data
const TORRENTS_FILE = path.join(__dirname, "torrents.json");
let savedTorrents = [];
// Helper: Check if a path exists
async function pathExists(p) {
try {
await fsPromises.access(p);
return true;
} catch {
return false;
}
}
// Load configuration from file
async function loadConfig() {
try {
if (await pathExists(CONFIG_FILE)) {
const data = await fsPromises.readFile(CONFIG_FILE, "utf-8");
config = JSON.parse(data);
config.storageDir = path.resolve(__dirname, config.storageDir);
} else {
await fsPromises.writeFile(
CONFIG_FILE,
JSON.stringify({ storageDir: "./downloads" }, null, 2)
);
}
} catch (err) {
logger.error("Error reading config file, using default config: " + err);
}
}
await loadConfig();
// Load saved torrents from file
async function loadSavedTorrentsFromFile() {
try {
if (fs.existsSync(TORRENTS_FILE)) {
const data = await fsPromises.readFile(TORRENTS_FILE, "utf-8");
savedTorrents = JSON.parse(data);
logger.info(`Loaded saved torrents: ${JSON.stringify(savedTorrents)}`);
} else {
logger.info("No torrents.json file found. Starting with an empty list.");
}
} catch (err) {
logger.error("Error loading torrents file: " + err);
}
}
await loadSavedTorrentsFromFile();
// Save torrents to file
async function saveTorrents() {
try {
await fsPromises.writeFile(
TORRENTS_FILE,
JSON.stringify(savedTorrents, null, 2)
);
} catch (err) {
logger.error("Error saving torrents file: " + err);
}
}
// Ensure storage directory exists
let STORAGE_DIR = config.storageDir;
if (!fs.existsSync(STORAGE_DIR)) {
fs.mkdirSync(STORAGE_DIR, { recursive: true });
}
// Attach torrent debug events
function attachTorrentDebugEvents(torrent) {
torrent.on("wire", (wire, addr) => {
logger.debug(`Torrent ${torrent.infoHash} connected to peer: ${addr}`);
});
torrent.on("upload", (bytes) => {
logger.debug(`Torrent ${torrent.infoHash} uploaded ${bytes} bytes`);
});
torrent.on("download", (bytes) => {
logger.debug(`Torrent ${torrent.infoHash} downloaded ${bytes} bytes`);
});
}
// --- Resume Active Torrents on Startup ---
// For each saved torrent that is not paused, add (or seed) it.
async function resumeActiveTorrents() {
logger.info("Resuming active torrents...");
for (const torrent of savedTorrents) {
if (!torrent.paused) {
try {
const torrentPath = path.join(STORAGE_DIR, torrent.name);
const exists = await pathExists(torrentPath);
if (exists) {
client.seed(
torrentPath,
{ name: torrent.name, keepSeeding: true },
(torrentInstance) => {
attachTorrentDebugEvents(torrentInstance);
logger.info(
`Resumed seeding torrent: ${torrentInstance.infoHash}`
);
}
);
} else {
client.add(
torrent.magnet,
{ path: STORAGE_DIR, keepSeeding: true },
(torrentInstance) => {
attachTorrentDebugEvents(torrentInstance);
logger.info(
`Resumed downloading torrent: ${torrentInstance.infoHash}`
);
}
);
}
} catch (err) {
logger.error(`Error resuming torrent ${torrent.name}: ${err}`);
}
}
}
}
await resumeActiveTorrents();
// --- API Routes ---
// GET /config - Return current configuration as JSON
app.get("/config", async (req, res) => {
try {
res.json({ storageDir: config.storageDir });
} catch (err) {
logger.error("Error getting config: " + err);
res.status(500).json({ error: "Error reading config" });
}
});
// GET /torrents - Return both active and paused torrents
app.get("/torrents", (req, res) => {
try {
const active = client.torrents.map((t) => ({
infoHash: t.infoHash,
name: t.name,
progress: t.progress,
numPeers: t.numPeers,
downloaded: t.downloaded,
uploaded: t.uploaded,
paused: false,
}));
const paused = savedTorrents
.filter((s) => s.paused)
.map((s) => ({
infoHash: s.infoHash,
name: s.name,
progress: 1,
numPeers: 0,
downloaded: 0,
uploaded: 0,
paused: true,
}));
const torrents = [...active, ...paused];
logger.info("GET /torrents returning: " + JSON.stringify(torrents));
res.json(torrents);
} catch (err) {
logger.error("Error listing torrents: " + err);
res.status(500).json({ error: "Error listing torrents" });
}
});
// POST /pause/:infoHash - Pause a torrent
app.post("/pause/:infoHash", async (req, res) => {
const infoHash = req.params.infoHash;
const torrent = client.get(infoHash);
if (!torrent) {
return res
.status(400)
.json({ error: "Torrent not active or already paused" });
}
client.remove(infoHash, { destroyStore: false }, async (err) => {
if (err) {
logger.error("Error pausing torrent: " + err);
return res.status(500).json({ error: "Error pausing torrent" });
}
const entry = savedTorrents.find((s) => s.infoHash === infoHash);
if (entry) entry.paused = true;
await saveTorrents();
logger.info(`Torrent paused: ${infoHash}`);
res.json({ message: "Torrent paused" });
});
});
// POST /resume/:infoHash - Resume a paused torrent
app.post("/resume/:infoHash", async (req, res) => {
const infoHash = req.params.infoHash;
const entry = savedTorrents.find((s) => s.infoHash === infoHash);
if (!entry) {
return res.status(400).json({ error: "Torrent not found in saved data" });
}
entry.paused = false;
await saveTorrents();
const torrentPath = path.join(STORAGE_DIR, entry.name);
const exists = await pathExists(torrentPath);
try {
if (exists) {
client.seed(
torrentPath,
{ name: entry.name, keepSeeding: true },
(torrent) => {
attachTorrentDebugEvents(torrent);
logger.info(
`Torrent resumed (seeding from file): ${torrent.infoHash}`
);
res.json({ message: "Torrent resumed" });
}
);
} else {
client.add(
entry.magnet,
{ path: STORAGE_DIR, keepSeeding: true },
(torrent) => {
attachTorrentDebugEvents(torrent);
logger.info(`Torrent resumed (downloading): ${torrent.infoHash}`);
res.json({ message: "Torrent resumed" });
}
);
}
} catch (err) {
logger.error("Error resuming torrent: " + err);
res.status(500).json({ error: "Error resuming torrent" });
}
});
// POST /config - Update storage directory
app.post("/config", async (req, res) => {
const { storageDir } = req.body;
if (!storageDir || typeof storageDir !== "string") {
return res.status(400).json({ error: "Invalid storageDir provided" });
}
try {
STORAGE_DIR = path.resolve(__dirname, storageDir);
config.storageDir = STORAGE_DIR;
if (!fs.existsSync(STORAGE_DIR)) {
fs.mkdirSync(STORAGE_DIR, { recursive: true });
}
await fsPromises.writeFile(
CONFIG_FILE,
JSON.stringify({ storageDir }, null, 2)
);
res.json({ message: "Config updated", storageDir: STORAGE_DIR });
} catch (err) {
logger.error("Error updating config: " + err);
res.status(500).json({ error: "Error updating config" });
}
});
// POST /add - Add a new torrent
app.post("/add", (req, res) => {
const { magnet } = req.body;
if (!magnet || typeof magnet !== "string") {
return res.status(400).json({ error: "No valid magnet link provided" });
}
if (!magnet.startsWith("magnet:?")) {
return res.status(400).json({ error: "Invalid magnet link format" });
}
if (savedTorrents.some((t) => t.magnet === magnet)) {
logger.info("Attempted to add duplicate torrent.");
return res.status(409).json({ error: "Torrent already added." });
}
try {
client.add(magnet, { path: STORAGE_DIR, keepSeeding: true }, (torrent) => {
attachTorrentDebugEvents(torrent);
logger.info(`Torrent added: ${torrent.infoHash}`);
const torrentData = {
magnet,
infoHash: torrent.infoHash,
name: torrent.name,
paused: false,
};
savedTorrents.push(torrentData);
saveTorrents();
res.json(torrentData);
});
} catch (err) {
logger.error("Error adding torrent: " + err);
res.status(500).json({ error: "Error adding torrent" });
}
});
// DELETE /remove/:infoHash - Remove a torrent completely.
app.delete("/remove/:infoHash", (req, res) => {
const infoHash = req.params.infoHash;
if (!infoHash || typeof infoHash !== "string") {
return res.status(400).json({ error: "Invalid infoHash" });
}
const torrent = client.get(infoHash);
const savedData = savedTorrents.find((t) => t.infoHash === infoHash);
const torrentName =
(torrent && torrent.name) || (savedData && savedData.name);
if (!torrentName) {
return res.status(400).json({ error: "Torrent name not found" });
}
try {
client.remove(infoHash, { destroyStore: true }, async (err) => {
if (err) {
logger.error("Error removing torrent: " + err);
return res.status(500).json({ error: "Error removing torrent" });
}
savedTorrents = savedTorrents.filter((t) => t.infoHash !== infoHash);
await saveTorrents();
const torrentPath = path.join(STORAGE_DIR, torrentName);
if (await pathExists(torrentPath)) {
try {
await fsPromises.rm(torrentPath, { recursive: true, force: true });
} catch (rmErr) {
logger.error("Error removing torrent files: " + rmErr);
}
}
logger.info(`Torrent removed: ${infoHash}`);
res.json({ message: "Torrent removed" });
});
} catch (err) {
logger.error("Error processing removal: " + err);
res.status(500).json({ error: "Error processing removal" });
}
});
// --- End API Routes ---
// Serve static files from "public"
app.use(express.static("public"));
// Log system limits periodically
function logSystemLimits() {
const platform = process.platform;
if (platform === "win32") {
logger.info("Running on Windows.");
const activeHandles = process._getActiveHandles
? process._getActiveHandles().length
: "N/A";
logger.info(`Active handles: ${activeHandles}`);
exec("netsh int tcp show global", (err, stdout, stderr) => {
if (err) {
logger.error(`Error running netsh: ${err.message}`);
} else {
logger.info(`TCP Global Settings:\n${stdout}`);
}
});
} else if (platform === "linux") {
exec('bash -c "ulimit -n"', (err, stdout, stderr) => {
if (err) {
logger.error(`Error fetching file descriptor limit: ${err.message}`);
} else {
logger.info(`File descriptor limit: ${stdout.trim()}`);
}
});
exec(`lsof -p ${process.pid} | wc -l`, (err, stdout, stderr) => {
if (err) {
logger.error(`Error counting open file descriptors: ${err.message}`);
} else {
logger.info(`Current open file descriptors: ${stdout.trim()}`);
}
});
exec("cat /proc/sys/net/core/rmem_max", (err, stdout, stderr) => {
if (err) {
logger.error(`Error fetching rmem_max: ${err.message}`);
} else {
logger.info(`Receive buffer max (rmem_max): ${stdout.trim()}`);
}
});
exec("cat /proc/sys/net/core/wmem_max", (err, stdout, stderr) => {
if (err) {
logger.error(`Error fetching wmem_max: ${err.message}`);
} else {
logger.info(`Send buffer max (wmem_max): ${stdout.trim()}`);
}
});
} else if (platform === "darwin") {
exec("ulimit -n", (err, stdout, stderr) => {
if (err) {
logger.error(`Error fetching file descriptor limit: ${err.message}`);
} else {
logger.info(`File descriptor limit: ${stdout.trim()}`);
}
});
exec(`lsof -p ${process.pid} | wc -l`, (err, stdout, stderr) => {
if (err) {
logger.error(`Error counting open file descriptors: ${err.message}`);
} else {
logger.info(`Current open file descriptors: ${stdout.trim()}`);
}
});
exec("sysctl kern.ipc.maxsockbuf", (err, stdout, stderr) => {
if (err) {
logger.error(`Error fetching kern.ipc.maxsockbuf: ${err.message}`);
} else {
logger.info(`kern.ipc.maxsockbuf: ${stdout.trim()}`);
}
});
exec("sysctl net.inet.tcp.recvspace", (err, stdout, stderr) => {
if (err) {
logger.error(`Error fetching net.inet.tcp.recvspace: ${err.message}`);
} else {
logger.info(`net.inet.tcp.recvspace: ${stdout.trim()}`);
}
});
exec("sysctl net.inet.tcp.sendspace", (err, stdout, stderr) => {
if (err) {
logger.error(`Error fetching net.inet.tcp.sendspace: ${err.message}`);
} else {
logger.info(`net.inet.tcp.sendspace: ${stdout.trim()}`);
}
});
} else {
logger.info(
`Platform ${platform} not specifically supported for system limits logging.`
);
}
}
logSystemLimits();
setInterval(logSystemLimits, 60000);
// Start the server
const server = app.listen(port, () => {
logger.info(`Server running on http://localhost:${port}`);
});
// Graceful shutdown
function shutdown() {
logger.info("Shutting down server...");
server.close(() => {
logger.info("HTTP server closed.");
client.destroy((err) => {
if (err) {
logger.error("Error destroying WebTorrent client: " + err);
} else {
logger.info("WebTorrent client destroyed.");
}
process.exit(0);
});
});
}
process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);