MangaCircus/peering/adapter.js

'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 { normalizeWorkIdentity, sameWorkIdentity, resolveWorkIdentity } = require('../scraper/lib/work-identity');
const { fromPeer } = require('../chapter-dates');

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 }));
}
function sameTitle(left, right) {
  return String(left || '').normalize('NFKC').trim().replace(/\s+/g, ' ').toLowerCase()
    === String(right || '').normalize('NFKC').trim().replace(/\s+/g, ' ').toLowerCase();
}
async function pageHashes(folder) {
  const names = (await fs.readdir(folder, { withFileTypes: true }))
    .filter(entry => entry.isFile() && /\.(?:jpe?g|png|webp|gif|avif|bmp)$/i.test(entry.name))
    .map(entry => entry.name).sort();
  if (!names.length) return [];
  return Promise.all(names.map(async name => crypto.createHash('sha256')
    .update(await fs.readFile(path.join(folder, name))).digest('hex')));
}
async function matchLegacyPeerCopy(inventory, remote, peerStore, info) {
  // Pre-work-ID peer imports can be adopted only with byte-for-byte evidence;
  // a matching title alone is never sufficient to merge library works.
  const candidates = inventory.mangas.filter(manga => sameTitle(manga.title, remote.title));
  if (candidates.length !== 1) return null;
  const manga = candidates[0];
  for (const ref of remote.chapters) {
    const book = String(Number(ref.volume) || Number(ref.book?.number) || 1);
    const local = manga.chapters.filter(chapter => Number(chapter.book) === Number(book)
      && String(chapter.number) === String(ref.number));
    if (!local.length) continue;
    const manifest = await client.chapter(peerStore, info.peerId, info.shareId, ref.id);
    const expected = manifest.pages.map(page => page.sha256);
    for (const chapter of local) {
      const actual = await pageHashes(path.join(manga.path, chapter.bookFolder, chapter.version));
      if (actual.length === expected.length && actual.every((hash, index) => hash === expected[index])) return manga;
    }
  }
  return null;
}
async function discover(info) {
  const result = await client.manga(store(), info.peerId, info.shareId);
  return { title: result.title, workIdentity: normalizeWorkIdentity(result.workIdentity), 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 workIdentity = normalizeWorkIdentity(remote.workIdentity);
  const queuedIdentity = normalizeWorkIdentity(item.workIdentity);
  if (queuedIdentity && !sameWorkIdentity(queuedIdentity, workIdentity)) throw new Error('Shared manga identity changed after it was queued; review the title before syncing');
  const catalog = { title: remote.title, workIdentity, 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, workIdentity,
    localServerId: stateStore.identity()?.id, libraryMangaId: item.libraryMangaId });
  if (!manga && (!workIdentity || workIdentity.kind === 'origin')) {
    manga = await matchLegacyPeerCopy(inventory, remote, stateStore, info);
  }
  if (!manga && inventory.mangas.some(candidate => sameTitle(candidate.title, remote.title))) {
    throw new Error(`A manga titled “${remote.title}” already exists, but neither its canonical identity nor shared chapter page hashes verify that it is the same work. No files were downloaded or changed; review the identity before syncing.`);
  }
  if (item.libraryMangaId && !manga) throw new Error(`Library manga target not found: ${item.libraryMangaId}`);
  const localIdentity = manga ? resolveWorkIdentity(manga.meta, manga.id, stateStore.identity()?.id) : null;
  if (manga && localIdentity && !normalizeWorkIdentity(manga.meta['x-work-identity'])) {
    manga.meta = { ...manga.meta, 'x-work-identity': localIdentity };
    await writeJsonAtomic(path.join(manga.path, 'manga.json'), manga.meta);
  }
  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 identityToPersist = localIdentity || workIdentity;
  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, ...(identityToPersist ? { 'x-work-identity': identityToPersist } : {}) },
    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 },
      // The peer's ecosystem date travels along: the chapter entered Circus there, not here.
      dates: fromPeer(chapter.dates || ref.dates, peer.alias),
      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 };