MangaCircus/peering/gateway.js

#!/usr/bin/env node
'use strict';

const https = require('node:https');
const fs = require('node:fs');
const fsp = fs.promises;
const path = require('node:path');
const crypto = require('node:crypto');
const { once } = require('node:events');
const { PeeringState, fp, hash, requirePeer, grantStatus } = require('./state');
const { parseKey, parseSpecialFolder } = require('../chapter-identity');
const { peerDates } = require('../chapter-dates');
const { resolveWorkIdentity } = require('../scraper/lib/work-identity');

const DATA = path.resolve(process.env.PEER_DATA || path.join(process.env.DATA || 'data', 'peering'));
const LIBRARY = path.resolve(process.env.LIBRARY || 'library');
const HOST = process.env.PEER_HOST || '127.0.0.1';
const PORT = Number(process.env.PEER_PORT || 8093);
const peering = new PeeringState(DATA);
const identity = peering.identity();
if (!identity) throw new Error('Generate peering identity in the admin UI before starting the gateway');
const MAX_PAGE = 30 * 1024 * 1024;
const MAX_COVER = 5 * 1024 * 1024;
const MIME = { '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.png': 'image/png', '.webp': 'image/webp', '.gif': 'image/gif', '.avif': 'image/avif', '.bmp': 'image/bmp' };
const usageFile = path.resolve(process.env.PEER_USAGE_FILE || path.join(DATA, 'gateway-usage.json'));
fs.mkdirSync(path.dirname(usageFile), { recursive: true, mode: 0o700 });
const usage = (() => { try { return JSON.parse(fs.readFileSync(usageFile, 'utf8')); } catch (e) { if (e.code === 'ENOENT') return { day: '', peers: {}, usedInvites: [] }; throw e; } })();
const active = new Map();
const uploadSlots = new Map();
const GLOBAL_STREAMS = Math.max(1, Math.min(32, Number(process.env.PEER_GLOBAL_STREAMS || 4)));
const indexCache = new Map();
const blobCache = new Map();
const sha = s => crypto.createHash('sha256').update(s).digest('hex');
const opaque = s => sha(s).slice(0, 32);
function saveUsage() { const tmp = `${usageFile}.tmp`; fs.writeFileSync(tmp, JSON.stringify(usage), { mode: 0o600 }); fs.renameSync(tmp, usageFile); }
function fail(code, message, retryAfter) { const e = new Error(message); e.code = code; e.retryAfter = retryAfter; throw e; }
function json(res, code, value, extra = {}) {
  res.writeHead(code, { 'content-type': 'application/json; charset=utf-8', 'cache-control': 'no-store', ...extra });
  res.end(JSON.stringify(value));
}
function peerOf(req) {
  const cert = req.socket.getPeerCertificate(true);
  if (!cert?.raw || Date.parse(cert.valid_to) <= Date.now() || Date.parse(cert.valid_from) > Date.now()) fail(401, 'Valid client certificate required');
  const fingerprint = fp(cert.raw);
  const state = peering.load();
  const peer = state.peers.find(p => p.fingerprint === fingerprint);
  if (!peer || fp(peer.cert) !== fingerprint) fail(401, 'Unrecognized client certificate');
  return { state, peer };
}
function grant(req, peer, scope = 'catalog') {
  const status = grantStatus(peer);
  if (!peer.paired || status.status !== 'active') fail(403, `Peering grant ${status.status}`);
  const match = /^Bearer ([A-Za-z0-9_-]{40,60})$/.exec(String(req.headers.authorization || ''));
  if (!match) fail(401, 'Grant required');
  const got = Buffer.from(hash(match[1]), 'hex'), want = Buffer.from(peer.grantHash, 'hex');
  if (got.length !== want.length || !crypto.timingSafeEqual(got, want)) fail(401, 'Invalid grant');
  if (!status.scopes.includes(scope)) fail(403, `Grant does not allow ${scope}`);
}
function quota(peer, bytes = 0, catalog = false) {
  const now = new Date(), day = now.toISOString().slice(0, 10), minute = Math.floor(Date.now() / 60000);
  if (usage.day !== day) { usage.day = day; usage.peers = {}; saveUsage(); }
  const entry = usage.peers[peer.id] ||= { bytes: 0, minute, catalogs: 0 };
  if (entry.minute !== minute) { entry.minute = minute; entry.catalogs = 0; }
  if (catalog && entry.catalogs >= peer.limits.catalogPerMinute) fail(429, 'Catalog request limit reached', 60);
  if (entry.bytes + bytes > peer.limits.bytesPerDay) fail(429, 'Daily upload limit reached', 3600);
  if (catalog) entry.catalogs++;
  if (bytes) entry.bytes += bytes; // Reserve before streaming; an interrupted stream never resets quota.
  saveUsage();
}
function safeDir(parent, name) {
  if (typeof name !== 'string' || !name || name.startsWith('.') || name.includes('/') || name.includes('\\') || name === '..') return null;
  const file = path.join(parent, name);
  if (!file.startsWith(parent + path.sep)) return null;
  try { if (!fs.lstatSync(file).isDirectory()) return null; } catch { return null; }
  return file;
}
async function dirs(root) {
  try { return (await fsp.readdir(root, { withFileTypes: true })).filter(e => e.isDirectory() && !e.isSymbolicLink() && !e.name.startsWith('.')).map(e => e.name).sort(); }
  catch { return []; }
}
async function jsonFile(file) { try { return JSON.parse(await fsp.readFile(file, 'utf8')); } catch { return {}; } }
function publicMetadata(meta = {}) {
  const tags = value => Array.isArray(value) ? value.map(item => String(typeof item === 'string' ? item : item?.name || '')).filter(Boolean).slice(0, 40) : [];
  return { year: Number(meta.year || meta.startDate?.year) || null, status: String(meta.status || 'unknown').slice(0, 40),
    description: String(meta.description || meta.synopsis || '').slice(0, 3000), tags: tags(meta.tags), genres: tags(meta.genres),
    direction: ['rtl', 'ltr', 'ttb'].includes(meta.direction) ? meta.direction : null };
}
function coverFile(mangaDir) {
  try {
    const entry = fs.readdirSync(mangaDir, { withFileTypes: true }).find(item => item.isFile() && !item.isSymbolicLink()
      && /^cover\.(jpe?g|png|webp|gif|avif|bmp)$/i.test(item.name));
    if (!entry) return null;
    const file = path.join(mangaDir, entry.name), stat = fs.lstatSync(file);
    return stat.isFile() && stat.size > 0 && stat.size <= MAX_COVER ? { file, size: stat.size, mime: MIME[path.extname(file).toLowerCase()], mtimeMs: stat.mtimeMs } : null;
  } catch { return null; }
}
async function mangaIndex(peer, share) {
  const key = `${peer.id}:${share.id}`;
  const old = indexCache.get(key);
  if (old && Date.now() - old.at < 30000) return old;
  const mangaDir = safeDir(LIBRARY, share.mangaId);
  if (!mangaDir) fail(404, 'Manga not found');
  const meta = await jsonFile(path.join(mangaDir, 'manga.json'));
  const chapters = [], byId = new Map();
  for (const bookName of await dirs(mangaDir)) {
    const book = parseKey(bookName);
    if (!book) continue;
    const bookDir = safeDir(mangaDir, bookName);
    for (const chapterName of await dirs(bookDir)) {
      const parsed = parseSpecialFolder(chapterName) || parseKey(chapterName);
      if (!parsed) continue;
      const chapterDir = safeDir(bookDir, chapterName);
      // A chapter without final metadata may still be a manually written partial.
      if (!fs.existsSync(path.join(chapterDir, 'chapter.json'))) continue;
      const entries = await fsp.readdir(chapterDir, { withFileTypes: true });
      const pages = entries.filter(e => e.isFile() && MIME[path.extname(e.name).toLowerCase()]).map(e => e.name).sort();
      if (!pages.length) continue;
      const id = `c${opaque(`${share.id}/${bookName}/${chapterName}`)}`;
      const chapterMeta = await jsonFile(path.join(chapterDir, 'chapter.json'));
      const dirStat = await fsp.stat(chapterDir);
      const item = { id, number: parsed.key, volume: book.key, title: String(chapterMeta.title || parsed.title || '').slice(0, 200),
        pageCount: pages.length, chapterDir, pages, dates: peerDates(chapterMeta, dirStat.mtimeMs) };
      chapters.push(item); byId.set(id, item);
      if (chapters.length > 10000) fail(413, 'Shared manga chapter limit exceeded');
    }
  }
  // Every version of a chapter number offers the earliest ecosystem date known for it.
  const earliest = new Map();
  for (const item of chapters) {
    const key = `${item.volume}/${item.number}`, at = item.dates.ecosystem;
    if (at && (!earliest.has(key) || Date.parse(at) < Date.parse(earliest.get(key)))) earliest.set(key, at);
  }
  for (const item of chapters) {
    const at = earliest.get(`${item.volume}/${item.number}`);
    if (at) item.dates = { ...item.dates, ecosystem: at };
  }
  const result = { at: Date.now(), mangaDir, meta, chapters, byId };
  indexCache.set(key, result);
  return result;
}
function permittedShare(peer, id) {
  const share = (peer.shares || []).find(s => s.id === id);
  if (!share) fail(404, 'Manga not found');
  return share;
}
async function chapterManifest(peer, share, chapterId) {
  const index = await mangaIndex(peer, share), chapter = index.byId.get(chapterId);
  if (!chapter) fail(404, 'Chapter not found');
  const pages = [];
  for (const name of chapter.pages) {
    const file = path.join(chapter.chapterDir, name), stat = await fsp.lstat(file);
    if (!stat.isFile() || stat.size < 1 || stat.size > MAX_PAGE) fail(413, 'Page size outside transfer limits');
    const digest = crypto.createHash('sha256');
    for await (const chunk of fs.createReadStream(file)) digest.update(chunk);
    const id = `b${opaque(`${share.id}/${chapter.id}/${name}`)}`;
    const etag = `"${id}-${stat.size}-${Math.round(stat.mtimeMs)}"`;
    pages.push({ id, bytes: stat.size, sha256: digest.digest('hex'), mime: MIME[path.extname(name).toLowerCase()], etag });
    blobCache.set(`${peer.id}:${id}`, { file, chapterDir: chapter.chapterDir, shareId: share.id, chapterId,
      size: stat.size, mtimeMs: stat.mtimeMs, at: Date.now() });
    if (pages.length > 500) fail(413, 'Chapter page limit exceeded');
  }
  return { id: chapter.id, number: chapter.number, volume: chapter.volume, title: chapter.title, dates: chapter.dates, pages };
}
async function streamBlob(req, res, peer, id) {
  const item = blobCache.get(`${peer.id}:${id}`);
  if (!item || Date.now() - item.at > 30 * 60 * 1000) fail(404, 'Refresh chapter manifest');
  permittedShare(peer, item.shareId);
  if (!fs.existsSync(path.join(item.chapterDir, 'chapter.json'))) fail(404, 'Chapter no longer complete');
  const stat = await fsp.lstat(item.file).catch(() => null);
  if (!stat?.isFile() || stat.size !== item.size || stat.mtimeMs !== item.mtimeMs) fail(412, 'Page changed; refresh manifest');
  const etag = `"${id}-${stat.size}-${Math.round(stat.mtimeMs)}"`;
  if (req.headers['if-match'] && req.headers['if-match'] !== etag) fail(412, 'Page changed; refresh manifest');
  const match = /^bytes=(\d+)-$/.exec(String(req.headers.range || ''));
  if (req.headers.range && !match) fail(416, 'Unsupported range');
  const start = match ? Number(match[1]) : 0;
  if (!Number.isSafeInteger(start) || start >= stat.size) fail(416, 'Range outside page');
  const count = stat.size - start;
  if ((active.get(peer.id) || 0) >= peer.limits.concurrent) fail(429, 'Too many transfers', 2);
  if ([...active.values()].reduce((a, b) => a + b, 0) >= GLOBAL_STREAMS) fail(429, 'Server transfer limit reached', 2);
  quota(peer, count);
  active.set(peer.id, (active.get(peer.id) || 0) + 1);
  try {
    res.writeHead(start ? 206 : 200, {
      'content-type': MIME[path.extname(item.file).toLowerCase()], 'content-length': count,
      'accept-ranges': 'bytes', etag,
      ...(start ? { 'content-range': `bytes ${start}-${stat.size - 1}/${stat.size}` } : {}),
      'cache-control': 'no-store',
    });
    const stream = fs.createReadStream(item.file, { start, highWaterMark: 64 * 1024 });
    for await (const chunk of stream) {
      if (req.destroyed || res.destroyed) break;
      // Reserve each chunk on a shared per-peer timeline, including simultaneous streams.
      const now = Date.now(), ready = Math.max(now, uploadSlots.get(peer.id) || now) + chunk.length * 1000 / peer.limits.bytesPerSecond;
      uploadSlots.set(peer.id, ready);
      if (ready > now) await new Promise(resolve => setTimeout(resolve, ready - now));
      if (!res.write(chunk)) await once(res, 'drain');
    }
    res.end();
  } finally { active.set(peer.id, Math.max(0, (active.get(peer.id) || 1) - 1)); }
}
async function streamCover(req, res, peer, shareId) {
  const share = permittedShare(peer, shareId), mangaDir = safeDir(LIBRARY, share.mangaId);
  if (!mangaDir) fail(404, 'Manga not found');
  const cover = coverFile(mangaDir);
  if (!cover) fail(404, 'Cover not available');
  const etag = `\"${opaque(`${share.id}/${cover.size}/${cover.mtimeMs}`)}\"`;
  if (req.headers['if-none-match'] === etag) { res.writeHead(304, { etag, 'cache-control': 'private, max-age=3600' }); return res.end(); }
  quota(peer, cover.size);
  res.writeHead(200, { 'content-type': cover.mime, 'content-length': cover.size, etag, 'cache-control': 'private, max-age=3600', 'x-content-type-options': 'nosniff' });
  if (req.method === 'HEAD') return res.end();
  await new Promise((resolve, reject) => fs.createReadStream(cover.file).on('error', reject).on('end', resolve).pipe(res));
}

const server = https.createServer({ key: identity.key, cert: identity.cert, requestCert: true, rejectUnauthorized: false, minVersion: 'TLSv1.3' }, async (req, res) => {
  try {
    const url = new URL(req.url, 'https://peer');
    if (req.method !== 'GET') fail(405, 'GET only');
    if (url.pathname === '/health') return json(res, 200, { ok: true });
    const { state, peer } = peerOf(req);
    if (url.pathname === '/peer/v1/hello') {
      if (!peer.paired) {
        const invitation = state.invitations.find(i => i.id === req.headers['x-invitation-id']);
        const code = String(req.headers['x-invitation-code'] || '');
        if (!invitation || Date.parse(invitation.expiresAt) <= Date.now() || invitation.codeHash !== hash(code)
          || usage.usedInvites.includes(invitation.id)) fail(403, 'Valid unused invitation required');
        usage.usedInvites.push(invitation.id); usage.usedInvites = usage.usedInvites.slice(-1000); saveUsage();
      }
      return json(res, 200, { version: 1, id: identity.id, fingerprint: identity.fingerprint });
    }
    if (url.pathname === '/peer/v1/access') {
      return json(res, 200, { version: 1, paired: !!peer.paired, name: peering.load().name,
        certificateExpiresAt: new crypto.X509Certificate(identity.cert).validTo,
        grant: grantStatus(peer) });
    }
    const requiredScope = url.pathname.startsWith('/peer/v1/blobs/') ? 'download' : 'catalog';
    grant(req, peer, requiredScope);
    if (url.pathname === '/peer/v1/catalog') {
      quota(peer, 0, true);
      const cursor = Number(url.searchParams.get('cursor') || 0);
      if (!Number.isSafeInteger(cursor) || cursor < 0) fail(400, 'Invalid cursor');
      const shares = peer.shares || [], items = [];
      for (const share of shares.slice(cursor, cursor + 50)) {
        const mangaDir = safeDir(LIBRARY, share.mangaId);
        if (!mangaDir) continue;
        const meta = await jsonFile(path.join(mangaDir, 'manga.json'));
        items.push({ id: share.id, title: String(meta.title || share.mangaId).slice(0, 200),
          ...publicMetadata(meta), workIdentity: resolveWorkIdentity(meta, share.mangaId, identity.id),
          coverAvailable: !!coverFile(mangaDir) });
      }
      return json(res, 200, { version: 1, items, nextCursor: cursor + 50 < shares.length ? String(cursor + 50) : null });
    }
    const coverMatch = /^\/peer\/v1\/covers\/(m[0-9a-f]{24})$/.exec(url.pathname);
    if (coverMatch) return await streamCover(req, res, peer, coverMatch[1]);
    const mangaMatch = /^\/peer\/v1\/manga\/(m[0-9a-f]{24})$/.exec(url.pathname);
    if (mangaMatch) {
      quota(peer, 0, true);
      const share = permittedShare(peer, mangaMatch[1]), index = await mangaIndex(peer, share);
      const cursor = Number(url.searchParams.get('cursor') || 0);
      if (!Number.isSafeInteger(cursor) || cursor < 0) fail(400, 'Invalid cursor');
      return json(res, 200, { id: share.id, title: String(index.meta.title || share.mangaId).slice(0, 200),
        workIdentity: resolveWorkIdentity(index.meta, share.mangaId, identity.id),
        ...publicMetadata(index.meta), coverAvailable: !!coverFile(index.mangaDir),
        chapters: index.chapters.slice(cursor, cursor + 100).map(({ id, number, volume, title, pageCount, dates }) => ({ id, number, volume, title, pageCount, dates })),
        nextCursor: cursor + 100 < index.chapters.length ? String(cursor + 100) : null });
    }
    const chapterMatch = /^\/peer\/v1\/manga\/(m[0-9a-f]{24})\/chapters\/(c[0-9a-f]{32})$/.exec(url.pathname);
    if (chapterMatch) {
      quota(peer, 0, true);
      return json(res, 200, await chapterManifest(peer, permittedShare(peer, chapterMatch[1]), chapterMatch[2]));
    }
    const blobMatch = /^\/peer\/v1\/blobs\/(b[0-9a-f]{32})$/.exec(url.pathname);
    if (blobMatch) return await streamBlob(req, res, peer, blobMatch[1]);
    fail(404, 'Not found');
  } catch (error) {
    if (!res.headersSent) json(res, Number(error.code) || 500, { error: Number(error.code) ? error.message : 'Internal error' }, error.retryAfter ? { 'retry-after': String(error.retryAfter) } : {});
    else res.destroy();
    if (!Number(error.code)) console.error('[peer-gateway]', error.message);
  }
});
server.listen(PORT, HOST, () => console.log(`[peer-gateway] listening on ${HOST}:${PORT}`));