Add scoped peer access and reversible serving controls

peering/adapter.js

@@ -0,0 +1,110 @@
+'use strict';
+
+const fs = require('node:fs').promises;
+const path = require('node:path');
+const crypto = require('node:crypto');
+const { PeeringState, requirePeer } = require('./state');
+const client = require('./client');
+const { ArchiveControlError } = require('../scraper/lib/control');
+const { selectByScope } = require('../scraper/lib/scope');
+const { scanLibrary, findManga, shouldDownload } = require('../scraper/lib/library-inventory');
+const { publishManifest } = require('../scraper/lib/publisher');
+const { writeJsonAtomic } = require('../scraper/lib/json-file');
+const { sleep } = require('../scraper/lib/delay');
+
+const dataRoot = () => path.resolve(process.env.PEER_DATA || path.join(process.env.DATA || 'data', 'peering'));
+const store = () => new PeeringState(dataRoot());
+const infoFor = value => {
+  try {
+    const u = new URL(String(value));
+    if (u.protocol !== 'peer:' || u.username || u.password || u.search || u.hash || !/^p[0-9a-f]{24}$/.test(u.hostname)
+      || !/^\/m[0-9a-f]{24}$/.test(u.pathname)) return null;
+    const peer = requirePeer(store().load(), u.hostname);
+    return { source: `peer-${peer.id}`, id: `${peer.id}-${u.pathname.slice(1)}`, peerId: peer.id,
+      shareId: u.pathname.slice(1), title: peer.alias, url: `peer://${peer.id}${u.pathname}` };
+  } catch { return null; }
+};
+function refsOf(result) {
+  return result.chapters.map(ref => ({ id: ref.id, number: ref.number, volume: ref.volume,
+    book: { number: Number(ref.volume) || 1 }, title: ref.title, pages: ref.pageCount }));
+}
+async function discover(info) {
+  const result = await client.manga(store(), info.peerId, info.shareId);
+  return { title: result.title, chapters: refsOf(result), availability: { selectedChapters: result.chapters.length, coverage: 'shared-selection' } };
+}
+async function run(info, item, options) {
+  const stateStore = store(), peer = requirePeer(stateStore.load(), info.peerId);
+  if (!peer.paired || !peer.receive) throw new Error('Downloads from this peer are disabled');
+  const stop = () => { const reason = options.shouldStop?.(); if (reason) throw new ArchiveControlError(reason === true ? 'paused' : String(reason)); };
+  stop();
+  const remote = await client.manga(stateStore, info.peerId, info.shareId);
+  const catalog = { title: remote.title, chapters: refsOf(remote), availability: { selectedChapters: remote.chapters.length, coverage: 'shared-selection' } };
+  if (options.onCatalog) await options.onCatalog(catalog);
+  const sourceUrl = info.url, source = info.source;
+  const workRoot = path.join(options.libraryRoot, `.mcscraper-work-${info.id}`);
+  const stateFile = path.join(workRoot, 'job.json');
+  await fs.mkdir(workRoot, { recursive: true });
+  let job = { version: 1, publicationMode: 'incremental', source, seriesUrl: sourceUrl, completed: [], jobCompleted: 0,
+    status: 'running', started: new Date().toISOString(), updated: new Date().toISOString(), total: 0, current: null };
+  try { const prior = JSON.parse(await fs.readFile(stateFile, 'utf8')); if (prior.seriesUrl === sourceUrl) job = { ...prior, status: 'running' }; }
+  catch (e) { if (e.code !== 'ENOENT') throw e; }
+  const record = async () => { job.updated = new Date().toISOString(); await writeJsonAtomic(stateFile, job); };
+  const inventory = await scanLibrary(options.libraryRoot);
+  let manga = findManga(inventory, { source, seriesId: info.id, sourceUrl, libraryMangaId: item.libraryMangaId });
+  if (item.libraryMangaId && !manga) throw new Error(`Library manga target not found: ${item.libraryMangaId}`);
+  const selected = selectByScope(catalog.chapters.filter(ref => shouldDownload(manga, ref, source, item.existingPolicy)), item.scope);
+  job.total = selected.length; job.jobCompleted = 0; job.current = null; await record();
+  let publishedCount = 0;
+  const series = { title: remote.title, direction: ['rtl', 'ltr', 'ttb'].includes(remote.direction) ? remote.direction : 'rtl',
+    sources: [{ name: `Peer ${peer.alias}`, url: sourceUrl }],
+    metadata: { title: remote.title }, scraper: { ids: { [source]: info.id } } };
+  for (let index = 0; index < selected.length; index++) {
+    stop();
+    const ref = selected[index];
+    if (!shouldDownload(manga, ref, source, item.existingPolicy)) { job.jobCompleted++; await record(); continue; }
+    const chapter = await client.chapter(stateStore, info.peerId, info.shareId, ref.id);
+    if (String(chapter.number) !== String(ref.number) || String(chapter.volume) !== String(ref.volume)) throw new Error('Peer chapter identity changed');
+    const digest = crypto.createHash('sha256').update(JSON.stringify(chapter.pages.map(p => [p.id, p.sha256]))).digest('hex').slice(0, 20);
+    const stage = path.join(workRoot, 'pages', `${ref.id}-${digest}`);
+    await fs.mkdir(stage, { recursive: true });
+    job.current = { number: ref.number, sourceId: ref.id, position: index + 1, total: selected.length,
+      downloadedPages: 0, expectedPages: chapter.pages.length };
+    await record();
+    const pages = chapter.pages.map((page, i) => ({ ...page, path: path.join(stage, `${String(i + 1).padStart(4, '0')}.${page.mime.split('/')[1] === 'jpeg' ? 'jpg' : page.mime.split('/')[1]}`) }));
+    const concurrency = Math.min(4, Math.max(1, Number(requirePeer(stateStore.load(), info.peerId).download?.concurrent || 2)));
+    for (let offset = 0; offset < pages.length; offset += concurrency) {
+      stop();
+      const batch = pages.slice(offset, offset + concurrency);
+      await Promise.all(batch.map(page => client.downloadPage(stateStore, info.peerId, page, page.path)));
+      job.current.downloadedPages = Math.min(pages.length, offset + batch.length);
+      await record();
+      stop();
+    }
+    const publication = await publishManifest({ version: 1, series, chapters: [{
+      number: ref.number, title: ref.title, book: { number: Number(ref.volume) || 1 },
+      source: { name: `Peer ${peer.alias} ${peer.id.slice(1, 7)}`, url: sourceUrl },
+      pages: pages.map(page => ({ path: page.path })),
+    }] }, { libraryRoot: options.libraryRoot, ...(manga ? { mangaDir: manga.path } : {}) });
+    if (publication.results[0].status === 'published') {
+      publishedCount++;
+      if (!manga) manga = { id: path.basename(publication.mangaDir), path: publication.mangaDir, chapters: [], meta: {} };
+      manga.chapters.push({ book: String(Number(ref.volume) || 1), number: String(ref.number), sourceKey: source });
+      if (options.onPublished) await options.onPublished({ mangaDir: publication.mangaDir, chapter: ref.number });
+    }
+    job.jobCompleted++; job.lastChapter = ref.number; job.current = null; await record();
+    // Only a fully verified and published chapter loses its resumable page cache.
+    await fs.rm(stage, { recursive: true, force: true });
+    if (index < selected.length - 1) { stop(); await sleep(Math.max(500, Number(options.settings.chapterMinMs || 500))); }
+  }
+  job.status = 'complete'; job.current = null; await record();
+  return { finalPath: manga?.path || null, publishedCount, selectedCount: selected.length };
+}
+function adapterForPeer(value) {
+  const info = infoFor(value);
+  if (!info) return null;
+  const adapter = { id: info.source, label: `Peer ${info.peerId.slice(1, 7)}`,
+    async discover() { return discover(info); },
+    async run(item, options) { return run(info, item, options); } };
+  return { adapter, info };
+}
+module.exports = { adapterForPeer, infoFor, discover, run };