feat(xtream-proxy): rewrite proxy logic and update configuration format

Refactor the xtream-proxy from a single-file Express-based server to a
modular Node.js application using native http/https modules. This
change introduces a major version bump to 2.0.0.

- Replace `hosts.txt` with `upstreams.json` for configuration
- Remove heavy dependencies (`express`, `ws`) in favor of native modules
- Implement new environment variables for proxy credentials and timeouts
- Add `src/` directory for modular code structure
- Add `docker-compose.example.yml` for easier deployment
- Update Dockerfile to support the new directory structure and environment
- Improve error handling and probe logic for upstream selection

Dockers/xtream-proxy/src/config.js

@@ -0,0 +1,92 @@
+'use strict';
+
+const fs = require('fs');
+const { settings } = require('./settings');
+const { Upstream } = require('./upstream');
+
+let upstreams = [];
+
+function readConfigEntries() {
+  const raw = fs.readFileSync(settings.configFile, 'utf8');
+  let data;
+  try {
+    data = JSON.parse(raw);
+  } catch (e) {
+    throw new Error(`invalid JSON: ${e.message}`);
+  }
+  if (!Array.isArray(data)) {
+    throw new Error('expected a JSON array');
+  }
+
+  const entries = [];
+  data.forEach((item, index) => {
+    if (!item || typeof item !== 'object' || Array.isArray(item)) {
+      console.warn(`Config entry #${index + 1}: not an object, skipping`);
+      return;
+    }
+    const { url, username, password } = item;
+    if (typeof url !== 'string' || !url.trim()) {
+      console.warn(`Config entry #${index + 1}: missing url, skipping`);
+      return;
+    }
+    if (typeof username !== 'string' || !username) {
+      console.warn(`Config entry #${index + 1}: missing username, skipping`);
+      return;
+    }
+    if (typeof password !== 'string' || !password) {
+      console.warn(`Config entry #${index + 1}: missing password, skipping`);
+      return;
+    }
+
+    let parsed;
+    try {
+      parsed = new URL(url);
+    } catch {
+      console.warn(`Config entry #${index + 1}: invalid url, skipping`);
+      return;
+    }
+    if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') {
+      console.warn(`Config entry #${index + 1}: url must be http(s), skipping`);
+      return;
+    }
+
+    entries.push({ url, username, password });
+  });
+  return entries;
+}
+
+function reloadConfig() {
+  let entries;
+  try {
+    entries = readConfigEntries();
+  } catch (e) {
+    console.error(`Config load failed (${settings.configFile}): ${e.message}`);
+    return false;
+  }
+
+  const previous = new Map(upstreams.map((u) => [u.key, u]));
+  const next = [];
+  const seen = new Set();
+
+  for (const entry of entries) {
+    const upstream = new Upstream(entry);
+    if (seen.has(upstream.key)) continue;
+    seen.add(upstream.key);
+    next.push(previous.get(upstream.key) || upstream);
+  }
+
+  const removed = upstreams.filter((u) => !seen.has(u.key)).length;
+  const added = next.filter((u) => !previous.has(u.key)).length;
+  if (added || removed) {
+    console.log(`Config reloaded: ${next.length} upstream(s), +${added} / -${removed}`);
+  }
+
+  upstreams = next;
+  return true;
+}
+
+function getUpstreams() {
+  return upstreams;
+}
+
+module.exports = { reloadConfig, getUpstreams };

Dockers/xtream-proxy/src/health.js

@@ -0,0 +1,26 @@
+'use strict';
+
+const { settings } = require('./settings');
+const { getUpstreams } = require('./config');
+const {
+  getLastProbeAt,
+  rankedUpstreams,
+  describeUpstream,
+} = require('./upstreams');
+
+function healthPayload() {
+  const upstreams = getUpstreams();
+  const ranked = rankedUpstreams();
+  const selected = ranked.find((entry) => !entry.busy) || ranked[0] || null;
+
+  return {
+    ok: upstreams.some((upstream) => upstream.state.healthy),
+    configFile: settings.configFile,
+    probeIntervalSec: settings.probeIntervalSec,
+    lastProbeAt: getLastProbeAt(),
+    selected: selected ? describeUpstream(selected.upstream) : null,
+    upstreams: upstreams.map(describeUpstream),
+  };
+}
+
+module.exports = { healthPayload };

Dockers/xtream-proxy/src/http-utils.js

@@ -0,0 +1,39 @@
+'use strict';
+
+function sendText(res, statusCode, text, extraHeaders = {}) {
+  res.writeHead(statusCode, {
+    'Content-Type': 'text/plain; charset=utf-8',
+    'Cache-Control': 'no-store',
+    ...extraHeaders,
+  });
+  res.end(text);
+}
+
+function sendJson(res, statusCode, body) {
+  const payload = JSON.stringify(body);
+  res.writeHead(statusCode, {
+    'Content-Type': 'application/json; charset=utf-8',
+    'Cache-Control': 'no-store',
+    'Content-Length': Buffer.byteLength(payload),
+  });
+  res.end(payload);
+}
+
+function firstHeader(value) {
+  if (Array.isArray(value)) value = value[0];
+  if (typeof value !== 'string') return '';
+  return value.split(',')[0].trim();
+}
+
+function getProxyOrigin(req) {
+  const proto =
+    firstHeader(req.headers['x-forwarded-proto']) ||
+    (req.socket.encrypted ? 'https' : 'http');
+  const host =
+    firstHeader(req.headers['x-forwarded-host']) ||
+    req.headers.host ||
+    'localhost';
+  return `${proto}://${host}`;
+}
+
+module.exports = { sendText, sendJson, firstHeader, getProxyOrigin };

Dockers/xtream-proxy/src/probe.js

@@ -0,0 +1,154 @@
+'use strict';
+
+const http = require('http');
+const https = require('https');
+const { settings } = require('./settings');
+const { baseState, nowIso, buildUpstreamUrl } = require('./upstream');
+
+function toInt(value) {
+  if (value === null || value === undefined || value === '') return null;
+  const n = Number(value);
+  return Number.isFinite(n) ? n : null;
+}
+
+function interpretProbe(statusCode, body, elapsedMs) {
+  const state = { ...baseState(), lastChecked: nowIso() };
+
+  if (statusCode < 200 || statusCode >= 300) {
+    state.error = `http_${statusCode}`;
+    return state;
+  }
+
+  let json;
+  try {
+    json = JSON.parse(body);
+  } catch {
+    state.error = 'invalid_json';
+    return state;
+  }
+
+  const info = json && (json.user_info || json.userInfo);
+  if (!info || typeof info !== 'object') {
+    state.error = 'no_user_info';
+    return state;
+  }
+
+  state.auth = info.auth === undefined ? null : info.auth;
+  state.status = info.status === undefined ? null : info.status;
+  state.activeCons = toInt(info.active_cons);
+  state.maxCons = toInt(info.max_connections);
+
+  if (state.auth !== null && Number(state.auth) !== 1) {
+    state.error = 'not_authorized';
+    return state;
+  }
+  if (
+    state.status !== null &&
+    String(state.status).toLowerCase() !== 'active'
+  ) {
+    state.error = `status_${state.status}`;
+    return state;
+  }
+
+  state.healthy = true;
+  state.latencyMs = Math.max(0, Math.round(elapsedMs));
+  state.busy =
+    state.maxCons !== null &&
+    state.maxCons > 0 &&
+    state.activeCons !== null &&
+    state.activeCons >= state.maxCons;
+
+  return state;
+}
+
+function probeUpstream(upstream) {
+  const search = `?${new URLSearchParams({
+    username: upstream.username,
+    password: upstream.password,
+  }).toString()}`;
+  const target = buildUpstreamUrl(upstream, '/player_api.php', search);
+  const lib = target.protocol === 'https:' ? https : http;
+  const started = Date.now();
+
+  return new Promise((resolve) => {
+    let settled = false;
+    let req = null;
+
+    const finish = (state) => {
+      if (settled) return;
+      settled = true;
+      clearTimeout(hardTimeout);
+      if (req) {
+        try {
+          req.destroy();
+        } catch {}
+      }
+      resolve(state);
+    };
+
+    const hardTimeout = setTimeout(
+      () => finish({ ...baseState(), lastChecked: nowIso(), error: 'timeout' }),
+      settings.probeTimeoutMs + 2000
+    );
+
+    try {
+      req = lib.request(
+        {
+          protocol: target.protocol,
+          hostname: target.hostname,
+          port: target.port || undefined,
+          path: `${target.pathname}${target.search}`,
+          method: 'GET',
+          headers: {
+            'User-Agent': settings.userAgent,
+            Accept: 'application/json',
+            'Accept-Encoding': 'identity',
+            Connection: 'close',
+          },
+          rejectUnauthorized: settings.rejectUnauthorized,
+        },
+        (res) => {
+          let body = '';
+          res.setEncoding('utf8');
+          res.on('data', (chunk) => {
+            if (body.length < 2_000_000) body += chunk;
+          });
+          res.on('end', () =>
+            finish(interpretProbe(res.statusCode, body, Date.now() - started))
+          );
+          res.on('error', () =>
+            finish({
+              ...baseState(),
+              lastChecked: nowIso(),
+              error: 'response_error',
+            })
+          );
+        }
+      );
+    } catch (e) {
+      finish({
+        ...baseState(),
+        lastChecked: nowIso(),
+        error: e.code || 'request_error',
+      });
+      return;
+    }
+
+    req.setTimeout(settings.probeTimeoutMs, () => {
+      try {
+        req.destroy();
+      } catch {}
+      finish({ ...baseState(), lastChecked: nowIso(), error: 'timeout' });
+    });
+    req.on('error', (e) =>
+      finish({
+        ...baseState(),
+        lastChecked: nowIso(),
+        error: e.code || 'request_error',
+      })
+    );
+    req.end();
+  });
+}
+
+module.exports = { interpretProbe, probeUpstream };

Dockers/xtream-proxy/src/proxy.js

@@ -0,0 +1,245 @@
+'use strict';
+
+const http = require('http');
+const https = require('https');
+const { settings } = require('./settings');
+const { getUpstreams } = require('./config');
+const { buildUpstreamUrl } = require('./upstream');
+const {
+  selectUpstream,
+  requestProbeSoon,
+} = require('./upstreams');
+const {
+  ReplaceStream,
+  buildReplacements,
+  rewriteUrlValue,
+  shouldRewriteBody,
+  filterHopByHop,
+  filterResponseHeaders,
+  isStreamPath,
+  extractCreds,
+  isAuthenticated,
+  rewritePath,
+  rewriteSearch,
+} = require('./rewrite');
+const { sendText, getProxyOrigin } = require('./http-utils');
+
+function nowIso() {
+  return new Date().toISOString();
+}
+
+function canRetry(req) {
+  return req.method === 'GET' || req.method === 'HEAD';
+}
+
+function markUpstreamFailure(upstream, error) {
+  upstream.state.healthy = false;
+  upstream.state.error = error?.code || error?.message || 'request_error';
+  upstream.state.lastChecked = nowIso();
+  requestProbeSoon();
+}
+
+function handleProxy(req, res, reqUrl) {
+  const creds = extractCreds(reqUrl);
+
+  if (!isAuthenticated(creds)) {
+    sendText(res, 401, 'Unauthorized');
+    return;
+  }
+
+  const streamRequest = isStreamPath(reqUrl.pathname);
+  const attempted = new Set();
+  const upstreamCount = getUpstreams().length;
+  let currentProxyReq = null;
+  let currentRelease = null;
+  let closed = false;
+
+  const releaseCurrent = () => {
+    if (currentRelease) {
+      const release = currentRelease;
+      currentRelease = null;
+      release();
+    }
+  };
+
+  res.on('close', () => {
+    closed = true;
+    if (!res.writableEnded && currentProxyReq) {
+      try {
+        currentProxyReq.destroy();
+      } catch {}
+    }
+    releaseCurrent();
+  });
+
+  req.on('error', () => {
+    if (currentProxyReq) {
+      try {
+        currentProxyReq.destroy();
+      } catch {}
+    }
+    releaseCurrent();
+  });
+
+  const attempt = () => {
+    if (closed || res.headersSent) return;
+
+    const upstream = selectUpstream(attempted);
+    if (!upstream) {
+      sendText(res, 503, 'No upstream available', { 'Retry-After': '5' });
+      requestProbeSoon();
+      return;
+    }
+
+    attempted.add(upstream.key);
+
+    if (streamRequest) upstream.state.inFlight += 1;
+    let released = false;
+    const release = () => {
+      if (released) return;
+      released = true;
+      if (streamRequest) {
+        upstream.state.inFlight = Math.max(0, upstream.state.inFlight - 1);
+      }
+    };
+    currentRelease = release;
+
+    const pathname = rewritePath(creds, upstream);
+    const search = rewriteSearch(reqUrl, upstream);
+    const target = buildUpstreamUrl(upstream, pathname, search);
+
+    const headers = filterHopByHop(req.headers);
+    delete headers.host;
+    delete headers['accept-encoding'];
+    headers.host = target.host;
+    headers['accept-encoding'] = 'identity';
+
+    const clientIp = req.socket.remoteAddress || '';
+    const existingForwardedFor = req.headers['x-forwarded-for'];
+    headers['x-forwarded-for'] = existingForwardedFor
+      ? `${existingForwardedFor}, ${clientIp}`
+      : clientIp;
+    if (!headers['x-forwarded-proto']) {
+      headers['x-forwarded-proto'] = req.socket.encrypted ? 'https' : 'http';
+    }
+    if (!headers['x-forwarded-host'] && req.headers.host) {
+      headers['x-forwarded-host'] = req.headers.host;
+    }
+
+    const lib = target.protocol === 'https:' ? https : http;
+    let proxyReq;
+    let headersTimeout = null;
+    try {
+      proxyReq = lib.request(
+        {
+          protocol: target.protocol,
+          hostname: target.hostname,
+          port: target.port || undefined,
+          path: `${target.pathname}${target.search}`,
+          method: req.method,
+          headers,
+          rejectUnauthorized: settings.rejectUnauthorized,
+        },
+        (proxyRes) => {
+          clearTimeout(headersTimeout);
+
+          const proxyOrigin = getProxyOrigin(req);
+          const rewriting = shouldRewriteBody(req, proxyRes);
+          const responseHeaders = filterResponseHeaders(
+            proxyRes.headers,
+            rewriting
+          );
+
+          if (responseHeaders.location) {
+            responseHeaders.location = rewriteUrlValue(
+              responseHeaders.location,
+              upstream,
+              proxyOrigin
+            );
+          }
+
+          try {
+            res.writeHead(proxyRes.statusCode || 502, responseHeaders);
+          } catch {
+            release();
+            try {
+              proxyRes.destroy();
+            } catch {}
+            try {
+              res.destroy();
+            } catch {}
+            return;
+          }
+
+          proxyRes.on('close', release);
+          proxyRes.on('error', () => {
+            release();
+            if (!res.headersSent) {
+              sendText(res, 502, 'Upstream response error');
+            } else {
+              res.destroy();
+            }
+          });
+
+          if (rewriting) {
+            const replacer = new ReplaceStream(
+              buildReplacements(upstream, proxyOrigin)
+            );
+            replacer.on('error', () => {
+              try {
+                res.destroy();
+              } catch {}
+              release();
+            });
+            proxyRes.pipe(replacer).pipe(res);
+          } else {
+            proxyRes.pipe(res);
+          }
+        }
+      );
+    } catch (e) {
+      release();
+      markUpstreamFailure(upstream, e);
+      if (canRetry(req) && attempted.size < upstreamCount) {
+        attempt();
+      } else {
+        sendText(res, 502, 'Upstream request failed');
+      }
+      return;
+    }
+
+    headersTimeout = setTimeout(() => {
+      try {
+        proxyReq.destroy(new Error('upstream_timeout'));
+      } catch {}
+    }, settings.upstreamTimeoutMs);
+
+    currentProxyReq = proxyReq;
+
+    proxyReq.on('error', (err) => {
+      clearTimeout(headersTimeout);
+      if (currentProxyReq === proxyReq) currentProxyReq = null;
+      release();
+      if (closed) return;
+
+      markUpstreamFailure(upstream, err);
+
+      if (!res.headersSent && canRetry(req) && attempted.size < upstreamCount) {
+        attempt();
+        return;
+      }
+
+      if (!res.headersSent) {
+        sendText(res, 502, 'Upstream request failed');
+      } else {
+        res.destroy();
+      }
+    });
+
+    req.pipe(proxyReq);
+  };
+
+  attempt();
+}
+
+module.exports = { handleProxy };

Dockers/xtream-proxy/src/rewrite.js

@@ -0,0 +1,273 @@
+'use strict';
+
+const { Transform } = require('stream');
+const { settings } = require('./settings');
+
+const HOP_BY_HOP = new Set([
+  'connection',
+  'keep-alive',
+  'proxy-authenticate',
+  'proxy-authorization',
+  'te',
+  'trailers',
+  'transfer-encoding',
+  'upgrade',
+]);
+
+const STREAM_PATH = /^\/(live|movie|series|timeshift|hls|stream|vod)\//i;
+const STREAM_EXT =
+  /\.(ts|m3u8|mp4|mkv|avi|mov|mpg|mpeg|flv|mp3|aac|m4a|m4v)$/i;
+
+class ReplaceStream extends Transform {
+  constructor(replacements) {
+    super();
+    this.replacements = replacements
+      .filter((r) => r.from && r.to !== undefined && r.from !== r.to)
+      .map((r) => ({ from: Buffer.from(r.from), to: Buffer.from(r.to) }));
+    this.maxLen = this.replacements.reduce(
+      (max, r) => Math.max(max, r.from.length),
+      0
+    );
+    this.carry = Buffer.alloc(0);
+  }
+
+  _transform(chunk, _encoding, callback) {
+    this._process(Buffer.concat([this.carry, chunk]), false, callback);
+  }
+
+  _flush(callback) {
+    this._process(this.carry, true, callback);
+  }
+
+  _process(buffer, isEnd, callback) {
+    const output = [];
+    let position = 0;
+
+    while (position < buffer.length) {
+      let best = null;
+      for (const replacement of this.replacements) {
+        const index = buffer.indexOf(replacement.from, position);
+        if (index === -1) continue;
+        if (!best || index < best.index) best = { index, replacement };
+      }
+
+      if (!best) break;
+      const end = best.index + best.replacement.from.length;
+      if (!isEnd && end > buffer.length) break;
+
+      if (best.index > position) {
+        output.push(buffer.subarray(position, best.index));
+      }
+      output.push(best.replacement.to);
+      position = end;
+    }
+
+    if (position < buffer.length) {
+      const keepFrom = isEnd
+        ? buffer.length
+        : Math.max(position, buffer.length - (this.maxLen - 1));
+      if (keepFrom > position) {
+        output.push(buffer.subarray(position, keepFrom));
+      }
+      this.carry = isEnd ? Buffer.alloc(0) : buffer.subarray(keepFrom);
+    } else {
+      this.carry = Buffer.alloc(0);
+    }
+
+    if (output.length) this.push(Buffer.concat(output));
+    callback();
+  }
+}
+
+function encodeQueryComponent(value) {
+  return encodeURIComponent(value).replace(/%20/g, '+');
+}
+
+function buildReplacements(upstream, proxyOrigin) {
+  const proxyHost = new URL(proxyOrigin).host;
+  const replacements = [];
+  const seen = new Set();
+
+  const add = (from, to) => {
+    if (!from || to === undefined || from === to || seen.has(from)) return;
+    seen.add(from);
+    replacements.push({ from, to });
+  };
+
+  add(upstream.baseNoSlash, proxyOrigin);
+  add(upstream.base.origin, proxyOrigin);
+
+  const bareHost = upstream.base.host;
+  add(`"url":"${bareHost}"`, `"url":"${proxyHost}"`);
+  add(`"url": "${bareHost}"`, `"url": "${proxyHost}"`);
+
+  add(upstream.username, settings.proxyUsername);
+  add(
+    encodeURIComponent(upstream.username),
+    encodeURIComponent(settings.proxyUsername)
+  );
+  add(
+    encodeQueryComponent(upstream.username),
+    encodeQueryComponent(settings.proxyUsername)
+  );
+
+  add(upstream.password, settings.proxyPassword);
+  add(
+    encodeURIComponent(upstream.password),
+    encodeURIComponent(settings.proxyPassword)
+  );
+  add(
+    encodeQueryComponent(upstream.password),
+    encodeQueryComponent(settings.proxyPassword)
+  );
+
+  return replacements;
+}
+
+function rewriteUrlValue(value, upstream, proxyOrigin) {
+  if (typeof value !== 'string') return value;
+  let result = value;
+  for (const { from, to } of buildReplacements(upstream, proxyOrigin)) {
+    result = result.split(from).join(to);
+  }
+  return result;
+}
+
+function isRewritableContentType(contentType) {
+  if (!contentType) return false;
+  const type = contentType.split(';')[0].trim().toLowerCase();
+  return (
+    type.startsWith('text/') ||
+    type === 'application/json' ||
+    type === 'application/xml' ||
+    type === 'application/javascript' ||
+    type === 'application/x-mpegurl' ||
+    type === 'application/vnd.apple.mpegurl' ||
+    type === 'audio/mpegurl' ||
+    type === 'audio/x-mpegurl' ||
+    type === 'application/octet-stream'
+  );
+}
+
+function shouldRewriteBody(req, upstreamResponse) {
+  if (req.method === 'HEAD') return false;
+  if (
+    upstreamResponse.statusCode === 204 ||
+    upstreamResponse.statusCode === 304
+  ) {
+    return false;
+  }
+  const encoding = upstreamResponse.headers['content-encoding'];
+  if (encoding && encoding.toLowerCase() !== 'identity') return false;
+  return isRewritableContentType(upstreamResponse.headers['content-type']);
+}
+
+function filterHopByHop(headers) {
+  const result = {};
+  if (!headers) return result;
+  for (const [key, value] of Object.entries(headers)) {
+    if (value === undefined) continue;
+    if (HOP_BY_HOP.has(key.toLowerCase())) continue;
+    result[key] = value;
+  }
+  return result;
+}
+
+function filterResponseHeaders(headers, rewriting) {
+  const result = {};
+  for (const [key, value] of Object.entries(headers)) {
+    if (value === undefined) continue;
+    const lower = key.toLowerCase();
+    if (HOP_BY_HOP.has(lower)) continue;
+    if (
+      rewriting &&
+      (lower === 'content-length' ||
+        lower === 'etag' ||
+        lower === 'content-md5')
+    ) {
+      continue;
+    }
+    result[key] = value;
+  }
+  return result;
+}
+
+function isStreamPath(pathname) {
+  return STREAM_PATH.test(pathname) || STREAM_EXT.test(pathname);
+}
+
+function extractCreds(reqUrl) {
+  const queryUsername = reqUrl.searchParams.get('username');
+  const queryPassword = reqUrl.searchParams.get('password');
+  const hasQueryCreds = queryUsername !== null || queryPassword !== null;
+
+  const rawSegments = reqUrl.pathname.split('/');
+  const decodedSegments = rawSegments.map((segment) => {
+    try {
+      return decodeURIComponent(segment);
+    } catch {
+      return segment;
+    }
+  });
+
+  const pathUsername = decodedSegments.includes(settings.proxyUsername);
+  const pathPassword = decodedSegments.includes(settings.proxyPassword);
+
+  return {
+    queryUsername,
+    queryPassword,
+    hasQueryCreds,
+    rawSegments,
+    hasPathCreds: pathUsername && pathPassword,
+  };
+}
+
+function isAuthenticated(creds) {
+  if (creds.hasQueryCreds) {
+    return (
+      creds.queryUsername === settings.proxyUsername &&
+      creds.queryPassword === settings.proxyPassword
+    );
+  }
+  return creds.hasPathCreds;
+}
+
+function rewritePath(creds, upstream) {
+  return creds.rawSegments
+    .map((segment) => {
+      let decoded = segment;
+      try {
+        decoded = decodeURIComponent(segment);
+      } catch {}
+      if (decoded === settings.proxyUsername) {
+        return encodeURIComponent(upstream.username);
+      }
+      if (decoded === settings.proxyPassword) {
+        return encodeURIComponent(upstream.password);
+      }
+      return segment;
+    })
+    .join('/');
+}
+
+function rewriteSearch(reqUrl, upstream) {
+  const params = new URLSearchParams(reqUrl.searchParams);
+  if (params.has('username')) params.set('username', upstream.username);
+  if (params.has('password')) params.set('password', upstream.password);
+  const search = params.toString();
+  return search ? `?${search}` : '';
+}
+
+module.exports = {
+  ReplaceStream,
+  buildReplacements,
+  rewriteUrlValue,
+  shouldRewriteBody,
+  filterHopByHop,
+  filterResponseHeaders,
+  isStreamPath,
+  extractCreds,
+  isAuthenticated,
+  rewritePath,
+  rewriteSearch,
+};

Dockers/xtream-proxy/src/settings.js

@@ -0,0 +1,35 @@
+'use strict';
+
+const settings = {
+  listenPort: Number(process.env.LISTEN_PORT || 8080),
+  configFile: process.env.CONFIG_FILE || '/config/upstreams.json',
+  proxyUsername:
+    process.env.PROXY_USERNAME || process.env.XTREAM_USERNAME || '',
+  proxyPassword:
+    process.env.PROXY_PASSWORD || process.env.XTREAM_PASSWORD || '',
+  probeIntervalSec: Math.max(
+    5,
+    Number(process.env.PROBE_INTERVAL_SEC || 30) || 30
+  ),
+  probeTimeoutMs: Math.max(
+    1000,
+    Number(process.env.PROBE_TIMEOUT_MS || 8000) || 8000
+  ),
+  upstreamTimeoutMs: Math.max(
+    1000,
+    Number(process.env.UPSTREAM_TIMEOUT_MS || 15000) || 15000
+  ),
+  rejectUnauthorized: process.env.PROXY_INSECURE_TLS !== '1',
+  allowBusyFallback: process.env.ALLOW_BUSY_FALLBACK !== '0',
+  userAgent: process.env.USER_AGENT || 'xtream-proxy/2.0',
+};
+
+function validate() {
+  if (!settings.proxyUsername || !settings.proxyPassword) {
+    throw new Error(
+      'PROXY_USERNAME and PROXY_PASSWORD (or XTREAM_USERNAME / XTREAM_PASSWORD) are required'
+    );
+  }
+}
+
+module.exports = { settings, validate };

Dockers/xtream-proxy/src/upstream.js

@@ -0,0 +1,51 @@
+'use strict';
+
+function nowIso() {
+  return new Date().toISOString();
+}
+
+function baseState() {
+  return {
+    healthy: false,
+    busy: false,
+    latencyMs: null,
+    activeCons: null,
+    maxCons: null,
+    status: null,
+    auth: null,
+    error: null,
+    lastChecked: nowIso(),
+  };
+}
+
+class Upstream {
+  constructor({ url, username, password }) {
+    const parsed = new URL(url);
+    parsed.hash = '';
+    parsed.search = '';
+    this.base = parsed;
+    this.basePath = parsed.pathname.replace(/\/+$/, '');
+    this.username = username;
+    this.password = password;
+    this.key = `${parsed.origin}${this.basePath}|${username}|${password}`;
+    this.state = { ...baseState(), inFlight: 0 };
+  }
+
+  get label() {
+    return `${this.base.origin}${this.basePath}`;
+  }
+
+  get baseNoSlash() {
+    return `${this.base.origin}${this.basePath}`;
+  }
+}
+
+function buildUpstreamUrl(upstream, pathname, search) {
+  const p = pathname.startsWith('/') ? pathname : `/${pathname}`;
+  const target = new URL(upstream.base.origin);
+  target.pathname = `${upstream.basePath}${p}` || '/';
+  target.search = search || '';
+  return target;
+}
+
+module.exports = { Upstream, baseState, nowIso, buildUpstreamUrl };

Dockers/xtream-proxy/src/upstreams.js

@@ -0,0 +1,163 @@
+'use strict';
+
+const { settings } = require('./settings');
+const { reloadConfig, getUpstreams } = require('./config');
+const { probeUpstream } = require('./probe');
+
+let lastProbeAt = null;
+let probing = false;
+let probeTimer = null;
+
+function nowIso() {
+  return new Date().toISOString();
+}
+
+function effectiveLoad(upstream) {
+  return Math.max(
+    upstream.state.activeCons || 0,
+    upstream.state.inFlight || 0
+  );
+}
+
+function isBusy(upstream) {
+  const max = upstream.state.maxCons;
+  if (max === null || max <= 0) return false;
+  return effectiveLoad(upstream) >= max;
+}
+
+function rankedUpstreams() {
+  return getUpstreams()
+    .filter((u) => u.state.healthy)
+    .map((u) => ({ upstream: u, busy: isBusy(u) }))
+    .sort((a, b) => {
+      if (a.busy !== b.busy) return a.busy ? 1 : -1;
+      const la = a.upstream.state.latencyMs ?? Infinity;
+      const lb = b.upstream.state.latencyMs ?? Infinity;
+      return la - lb;
+    });
+}
+
+function selectUpstream(exclude = new Set()) {
+  const candidates = rankedUpstreams().filter(
+    ({ upstream }) => !exclude.has(upstream.key)
+  );
+
+  const idle = candidates.find((c) => !c.busy);
+  if (idle) return idle.upstream;
+
+  if (settings.allowBusyFallback) {
+    const fallback = candidates.find((c) => c.busy);
+    if (fallback) return fallback.upstream;
+  }
+
+  return null;
+}
+
+function describeUpstream(upstream) {
+  return {
+    url: upstream.label,
+    healthy: upstream.state.healthy,
+    busy: isBusy(upstream),
+    latencyMs: upstream.state.latencyMs,
+    activeCons: upstream.state.activeCons,
+    maxCons: upstream.state.maxCons,
+    inFlight: upstream.state.inFlight,
+    status: upstream.state.status,
+    error: upstream.state.error,
+    lastChecked: upstream.state.lastChecked,
+  };
+}
+
+function logProbeResults(snapshot) {
+  if (snapshot.length === 0) {
+    console.warn(`No upstreams configured in ${settings.configFile}`);
+    return;
+  }
+
+  const ranked = rankedUpstreams();
+  const lines = ranked.map(({ upstream, busy }, index) => {
+    const load = `${upstream.state.activeCons ?? '?'}/${upstream.state.maxCons ?? '?'}`;
+    return `${index + 1}. ${upstream.label} ${upstream.state.latencyMs ?? '?'}ms ${
+      busy ? `busy(${load})` : 'idle'
+    }${upstream.state.inFlight ? ` inflight=${upstream.state.inFlight}` : ''}`;
+  });
+
+  const failed = snapshot.filter((u) => !u.state.healthy);
+  if (failed.length) {
+    lines.push(
+      ...failed.map((u) => `x. ${u.label} ${u.state.error || 'unhealthy'}`)
+    );
+  }
+  console.log(`Probe results:\n  ${lines.join('\n  ')}`);
+}
+
+async function probeCycle() {
+  if (probing) return;
+  probing = true;
+
+  try {
+    reloadConfig();
+    const snapshot = getUpstreams().slice();
+    const results = await Promise.all(
+      snapshot.map(async (upstream) => ({
+        upstream,
+        state: await probeUpstream(upstream),
+      }))
+    );
+
+    for (const { upstream, state } of results) {
+      upstream.state = { ...upstream.state, ...state };
+    }
+
+    lastProbeAt = nowIso();
+    logProbeResults(snapshot);
+  } catch (e) {
+    console.error('Probe cycle failed:', e);
+  } finally {
+    probing = false;
+  }
+}
+
+function scheduleProbe(delayMs) {
+  if (probeTimer) {
+    clearTimeout(probeTimer);
+    probeTimer = null;
+  }
+  probeTimer = setTimeout(async () => {
+    probeTimer = null;
+    await probeCycle();
+    scheduleProbe(settings.probeIntervalSec * 1000);
+  }, delayMs);
+}
+
+function startProbing() {
+  scheduleProbe(settings.probeIntervalSec * 1000);
+}
+
+function stopProbing() {
+  if (probeTimer) {
+    clearTimeout(probeTimer);
+    probeTimer = null;
+  }
+}
+
+function requestProbeSoon() {
+  if (probing) return;
+  scheduleProbe(1000);
+}
+
+function getLastProbeAt() {
+  return lastProbeAt;
+}
+
+module.exports = {
+  probeCycle,
+  startProbing,
+  stopProbing,
+  requestProbeSoon,
+  selectUpstream,
+  rankedUpstreams,
+  isBusy,
+  describeUpstream,
+  getLastProbeAt,
+};