Checkpoint peering identity and opt-in discovery

peering/adapter.js

@@ -11,6 +11,7 @@ const { scanLibrary, findManga, shouldDownload } = require('../scraper/lib/libra
 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 dataRoot = () => path.resolve(process.env.PEER_DATA || path.join(process.env.DATA || 'data', 'peering'));
 const store = () => new PeeringState(dataRoot());
@@ -28,9 +29,41 @@ 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, chapters: refsOf(result), availability: { selectedChapters: result.chapters.length, coverage: 'shared-selection' } };
+  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);
@@ -38,7 +71,10 @@ async function run(info, item, options) {
   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' } };
+  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}`);
@@ -50,14 +86,28 @@ async function run(info, item, options) {
   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 });
+  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 }, scraper: { ids: { [source]: info.id } } };
+    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];

peering/client.js

@@ -4,6 +4,7 @@ const https = require('node:https');
 const fs = require('node:fs');
 const fsp = fs.promises;
 const crypto = require('node:crypto');
+const { normalizeWorkIdentity } = require('../scraper/lib/work-identity');
 const { pipeline } = require('node:stream/promises');
 const { Transform } = require('node:stream');
 const { PeeringState, fp, requirePeer, grantStatus } = require('./state');
@@ -101,6 +102,7 @@ async function catalog(store, id) {
   for (let page = 0; page < 100; page++) {
     const result = await jsonRequest(store, id, `/peer/v1/catalog${cursor ? `?cursor=${encodeURIComponent(cursor)}` : ''}`);
     if (!Array.isArray(result.items) || result.items.length > 50) throw new Error('Invalid peer catalog');
+    for (const item of result.items) if (item.workIdentity != null && !normalizeWorkIdentity(item.workIdentity)) throw new Error('Invalid work identity in peer catalog');
     items.push(...result.items); cursor = result.nextCursor;
     if (!cursor) return items;
   }
@@ -112,6 +114,7 @@ async function manga(store, id, shareId) {
   for (let page = 0; page < 100; page++) {
     const result = await jsonRequest(store, id, `/peer/v1/manga/${shareId}${cursor ? `?cursor=${encodeURIComponent(cursor)}` : ''}`);
     if (result.id !== shareId || !Array.isArray(result.chapters) || result.chapters.length > 100) throw new Error('Invalid peer manga manifest');
+    if (result.workIdentity != null && !normalizeWorkIdentity(result.workIdentity)) throw new Error('Invalid work identity in peer manga manifest');
     first ||= result; chapters.push(...result.chapters); cursor = result.nextCursor;
     if (!cursor) return { id: shareId, title: first.title, ...first, chapters };
   }

peering/gateway.js

@@ -9,6 +9,7 @@ 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 { 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');
@@ -231,7 +232,8 @@ const server = https.createServer({ key: identity.key, cert: identity.cert, requ
         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), coverAvailable: !!coverFile(mangaDir) });
+          ...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 });
     }
@@ -244,6 +246,7 @@ const server = https.createServer({ key: identity.key, cert: identity.cert, requ
       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 }) => ({ id, number, volume, title, pageCount })),
         nextCursor: cursor + 100 < index.chapters.length ? String(cursor + 100) : null });

peering/test/peering.test.js

@@ -71,7 +71,7 @@ test('mTLS pairing, opt-in catalog, peer queue publication, and revocation', { t
   await fsp.writeFile(path.join(chapterDir, 'chapter.json'), JSON.stringify({ title: 'The first chapter', source: { name: 'Local' } }));
   await fsp.writeFile(path.join(chapterDir, '0001.png'), PNG);
   const ga = await gateway(dataA, libA, portA, path.join(root, 'usage-a.json'));
-  const gb = await gateway(dataB, libB, portB, path.join(root, 'usage-b.json'));
+  let gb = await gateway(dataB, libB, portB, path.join(root, 'usage-b.json'));
   t.after(async () => { await stop(ga); await stop(gb); });
   await client.probe(b, inB.id);
   await client.probe(a, inA.id);
@@ -90,6 +90,7 @@ test('mTLS pairing, opt-in catalog, peer queue publication, and revocation', { t
   assert.equal(listing.length, 1);
   assert.equal(listing[0].title, 'Sample Manga');
   assert.equal(listing[0].coverAvailable, true);
+  assert.deepEqual(listing[0].workIdentity, require('../../scraper/lib/work-identity').resolveWorkIdentity({}, 'Sample Manga', a.identity().id));
   assert.deepEqual(listing[0].tags, []);
   assert.equal(JSON.stringify(listing).includes('Sample Manga\\v01'), false);
   const remote = await client.manga(b, inB.id, listing[0].id);
@@ -131,10 +132,78 @@ test('mTLS pairing, opt-in catalog, peer queue publication, and revocation', { t
   const result = await run(info, { scope: { mode: 'missing' }, existingPolicy: 'skip-any' },
     { libraryRoot: libB, settings: { chapterMinMs: 500 }, shouldStop: () => false });
   assert.equal(result.publishedCount, 1);
+  assert.deepEqual(JSON.parse(await fsp.readFile(path.join(libB, 'Sample Manga', 'manga.json'), 'utf8'))['x-work-identity'], remote.workIdentity,
+    'The stable work identity is stored with the imported peer edition');
   assert.equal(fs.existsSync(path.join(libB, 'Sample Manga', 'v01', 'c0001 The first chapter [Peer ServerA ' + inB.id.slice(1, 7) + ']', '0001.png')), true);
   const second = await run(info, { scope: { mode: 'missing' }, existingPolicy: 'skip-any' },
     { libraryRoot: libB, settings: { chapterMinMs: 500 }, shouldStop: () => false });
   assert.equal(second.publishedCount, 0);
+
+  // Simulate B adding one chapter to its imported edition, then sharing it back.
+  // A must identify its original folder by the carried origin identity and fetch
+  // only the new chapter, rather than creating a title-conflict quarantine.
+  const chapterTwo = path.join(libB, 'Sample Manga', 'v01', 'c0002 Chapter from B');
+  await fsp.mkdir(chapterTwo, { recursive: true });
+  await fsp.writeFile(path.join(chapterTwo, 'chapter.json'), JSON.stringify({ title: 'Chapter from B', source: { name: 'Local' } }));
+  await fsp.writeFile(path.join(chapterTwo, '0001.png'), PNG);
+  b.setShare(inB.id, 'Sample Manga', true);
+  b.updatePeer(inB.id, { serve: true });
+  const reverseGrant = b.issueGrant(inB.id, ['download']);
+  a.acceptGrant(inA.id, reverseGrant.token);
+  a.updatePeer(inA.id, { receive: true });
+  const reverseCatalog = await client.catalog(a, inA.id);
+  assert.equal(reverseCatalog.length, 1);
+  const reverseManga = await client.manga(a, inA.id, reverseCatalog[0].id);
+  assert.deepEqual(reverseManga.workIdentity, remote.workIdentity, 'The original work identity survives a peer round trip');
+  process.env.PEER_DATA = dataA;
+  const reverseInfo = require('../adapter').infoFor(`peer://${inA.id}/${reverseCatalog[0].id}`);
+  const returned = await require('../adapter').run(reverseInfo, { scope: { mode: 'missing' }, existingPolicy: 'skip-any', workIdentity: reverseManga.workIdentity },
+    { libraryRoot: libA, settings: { chapterMinMs: 500 }, shouldStop: () => false });
+  assert.equal(returned.finalPath, path.join(libA, 'Sample Manga'), 'The round-trip publication targets the original library folder');
+  assert.equal(returned.selectedCount, 1, 'Only the chapter added on B is selected for sync-missing');
+  assert.equal(returned.publishedCount, 1);
+  assert.equal(fs.existsSync(path.join(libA, 'Sample Manga', 'v01', 'c0002 Chapter from B [Peer ServerB ' + inA.id.slice(1, 7) + ']', '0001.png')), true);
+  const noChanges = await require('../adapter').run(reverseInfo, { scope: { mode: 'missing' }, existingPolicy: 'skip-any', workIdentity: reverseManga.workIdentity },
+    { libraryRoot: libA, settings: { chapterMinMs: 500 }, shouldStop: () => false });
+  assert.equal(noChanges.selectedCount, 0, 'A repeated round trip queues no duplicate chapters');
+
+  // Simulate an older B copy which predates persisted identities. Exact shared
+  // page hashes may bridge that legacy gap, but title similarity alone may not.
+  const legacyMetadataPath = path.join(libB, 'Sample Manga', 'manga.json');
+  const legacyMetadata = JSON.parse(await fsp.readFile(legacyMetadataPath, 'utf8'));
+  delete legacyMetadata['x-work-identity'];
+  await fsp.writeFile(legacyMetadataPath, JSON.stringify(legacyMetadata));
+  const chapterThree = path.join(libB, 'Sample Manga', 'v01', 'c0003 Chapter from B');
+  await fsp.mkdir(chapterThree, { recursive: true });
+  await fsp.writeFile(path.join(chapterThree, 'chapter.json'), JSON.stringify({ title: 'Chapter from B', source: { name: 'Local' } }));
+  await fsp.writeFile(path.join(chapterThree, '0001.png'), PNG);
+  await stop(gb); // Clear the gateway's short-lived manifest cache after this legacy-data migration.
+  gb = await gateway(dataB, libB, portB, path.join(root, 'usage-b.json'));
+  const legacyManga = await client.manga(a, inA.id, reverseCatalog[0].id);
+  assert.notDeepEqual(legacyManga.workIdentity, remote.workIdentity, 'Legacy copy has no propagated identity and gets a peer-local fallback');
+  const legacySync = await require('../adapter').run(reverseInfo, { scope: { mode: 'missing' }, existingPolicy: 'skip-any', workIdentity: legacyManga.workIdentity },
+    { libraryRoot: libA, settings: { chapterMinMs: 500 }, shouldStop: () => false });
+  assert.equal(legacySync.finalPath, path.join(libA, 'Sample Manga'), 'Content evidence safely reuses the original folder for legacy peer copies');
+  assert.equal(legacySync.selectedCount, 1, 'Legacy matching still selects only the chapter absent on A');
+  assert.equal(legacySync.publishedCount, 1);
+  assert.equal(fs.existsSync(path.join(libA, 'Sample Manga', 'v01', 'c0003 Chapter from B [Peer ServerB ' + inA.id.slice(1, 7) + ']', '0001.png')), true);
+  for (const book of await fsp.readdir(path.join(libB, 'Sample Manga'))) {
+    const bookPath = path.join(libB, 'Sample Manga', book);
+    if (!(await fsp.stat(bookPath)).isDirectory()) continue;
+    for (const chapter of await fsp.readdir(bookPath)) {
+      const page = path.join(bookPath, chapter, '0001.png');
+      if (fs.existsSync(page)) await fsp.writeFile(page, Buffer.from(`different-page-${chapter}`));
+    }
+  }
+  await stop(gb);
+  gb = await gateway(dataB, libB, portB, path.join(root, 'usage-b.json'));
+  const unverifiedLegacy = await client.manga(a, inA.id, reverseCatalog[0].id);
+  await assert.rejects(() => require('../adapter').run(reverseInfo,
+    { scope: { mode: 'missing' }, existingPolicy: 'skip-any', workIdentity: unverifiedLegacy.workIdentity },
+    { libraryRoot: libA, settings: { chapterMinMs: 500 }, shouldStop: () => false }),
+  /neither its canonical identity nor shared chapter page hashes verify.*No files were downloaded or changed/);
+  assert.equal((await require('../../scraper/lib/library-inventory').scanLibrary(libA)).mangas[0].chapters.length, 3,
+    'A title collision with differing chapter bytes leaves the local library untouched');
   a.updatePeer(inA.id, { servingPaused: true });
   assert.equal((await client.access(b, inB.id)).grant.status, 'paused');
   await assert.rejects(() => client.catalog(b, inB.id), /Peer HTTP 403/);