fix(sync): leave duplicate-feed episodes to the show that holds their IDs

scripts/README.md

@@ -31,8 +31,8 @@ A Git push does not deploy this pipeline. Changes to `scripts/sync/` or `scripts
 | [`scripts/sync/lib/attribution.js`](sync/lib/attribution.js) | Vector attribution log: one JSON line per episode point stored or deleted (prune, cap trim), show vector stored, and show imported or deleted. Written to `SYNC_ATTRIBUTION_DIR` (default `scripts/data/attribution/`, gitignored) as daily files, gzipped after the day ends and kept 60 days. Best effort: a write failure never fails a stage. Stage 4 adds `stored`, `pruned` and `reasons …` to its run line |
 | [`scripts/sync/attribution-report.js`](sync/attribution-report.js) | Read-only report over that log: embeddings by reason (new episode, first coverage, returning show, re-embed after cap trim or prune, Podcast Index ID change, backfill), deletions, and how long chart entrants stay. `--days N`, `--json` |
 | [`scripts/sync/lib/episode-book.js`](sync/lib/episode-book.js) + [`episode-book-cli.js`](sync/episode-book-cli.js) | The episode book in sqld: one permanent row per (show, catalog key) with its recipe ID and served ID, plus Podcast Index aliases, `served_show` and `show_window`. `ensure` creates the tables and columns; `stats` counts. Writes go through `turso.atomic()` (one Hrana batch: every step runs only after the previous one succeeded, then COMMIT, else ROLLBACK). Rules: [`docs/EPISODE_IDS.md`](../docs/EPISODE_IDS.md) |
-| [`scripts/sync/lib/episode-identity.js`](sync/lib/episode-identity.js) | Stage 4's episode identity: each window episode gets its catalog key and recipe ID, and its point is named by the book's served ID. A new episode's book row (served ID = its current Podcast Index ID) is written before its vector; a Podcast Index ID change keeps the point and adds an alias instead of re-embedding. Points are stamped `eid` and `catalog_key` (`eid_missing` without a key). A new key whose Podcast Index ID is already served for another key (a changed GUID) is deferred: today's naming, no row, no stamps. If the book write fails, the show keeps today's naming for that run and the bootstrap's `--catch-up` records it later; a finished catch-up pass starts over when run again |
-| [`scripts/sync/bootstrap-book.js`](sync/bootstrap-book.js) + [`lib/book-bootstrap.js`](sync/lib/book-bootstrap.js) | One resumable pass over every show with episode points: maps each point's payload ID (its Podcast Index ID) to a catalog key through PI's episode list, writes the show's book rows in one transaction (served ID = that PI ID), then stamps `eid` and `catalog_key` on the points (`eid_missing` when unmapped). Point IDs and vectors never change. `--dry`, `--limit-shows N`, `--minutes N`; progress in `SYNC_BOOTSTRAP_STATE`. `--catch-up` takes only shows that still have unstamped points |
+| [`scripts/sync/lib/episode-identity.js`](sync/lib/episode-identity.js) | Stage 4's episode identity: each window episode gets its catalog key and recipe ID, and its point is named by the book's served ID. A new episode's book row (served ID = its current Podcast Index ID) is written before its vector; a Podcast Index ID change keeps the point and adds an alias instead of re-embedding. Points are stamped `eid` and `catalog_key` (`eid_missing` without a key). A new key whose Podcast Index ID is already served for another key is deferred: today's naming, no row. Its point belongs to the owner: stamped with the owner's `eid` and `catalog_key` when the owner is in the same show (a changed GUID), or `eid_shared` with the owning show when it is another show (a duplicate feed: Podcast Index lists one podcast under two show IDs that share episode IDs). If the book write fails, the show keeps today's naming for that run and the bootstrap's `--catch-up` records it later; a finished catch-up pass starts over when run again |
+| [`scripts/sync/bootstrap-book.js`](sync/bootstrap-book.js) + [`lib/book-bootstrap.js`](sync/lib/book-bootstrap.js) | One resumable pass over every show with episode points: maps each point's payload ID (its Podcast Index ID) to a catalog key through PI's episode list, writes the show's book rows in one transaction (served ID = that PI ID), then stamps `eid` and `catalog_key` on the points (`eid_missing` when unmapped). A point whose Podcast Index ID another key already serves gets no row, so the show's transaction can't fail on it: it is stamped with that key's identity when it is in the same show, or `eid_shared` with the owning show for a duplicate feed. Point IDs and vectors never change. `--dry`, `--limit-shows N`, `--minutes N`; progress in `SYNC_BOOTSTRAP_STATE`. `--catch-up` takes only shows that still have points with none of `eid`, `eid_missing` and `eid_shared` |
 | [`scripts/sync/lib/episode-recipe-id.js`](sync/lib/episode-recipe-id.js) | boxlore episode IDs from (show ID, catalog key); must match the app's `EpisodeRecipeId` and the shared fixture in `core/rss/src/test/resources/fixtures/` |
 | [`scripts/package.json`](package.json) | Sync Node deps + `npm run test:sync` |
 

scripts/sync/bootstrap-book.js

@@ -5,11 +5,12 @@
  * Book bootstrap (own-episode-IDs plan P1, section 8): one resumable pass over every show with
  * episode points. Per show: read its points' payload IDs, fetch PI's episode list once, write its
  * book rows in one transaction (served ID = the PI ID the point already has), then stamp `eid` and
- * `catalog_key` on each point (`eid_missing` when unmapped). Point IDs and vectors never change.
+ * `catalog_key` on each point (`eid_missing` when unmapped, `eid_shared` when a duplicate feed
+ * already holds its PI ID as a served ID). Point IDs and vectors never change.
  *
  *   node sync/bootstrap-book.js [--minutes N] [--dry] [--limit-shows N] [--catch-up]
  *
- * --catch-up only takes shows that still have points with neither `eid` nor `eid_missing` (added
+ * --catch-up only takes shows that still have points with none of `eid`, `eid_missing` and `eid_shared` (added
  * between the main pass and stage 4 writing identities itself, or by a run whose book write failed).
  * It keeps its own state file, and can be run again any time.
  *
@@ -27,6 +28,7 @@ const log = require('./lib/log');
 const cfg = require('./lib/config');
 const { ensureBookSchema, bootstrapShow, bookDb } = require('./lib/episode-book');
 const { planShow, stampOperations } = require('./lib/book-bootstrap');
+const { loadServedOwners } = require('./lib/episode-identity');
 
 const args = process.argv.slice(2);
 const opt = (name, def) => {
@@ -57,7 +59,7 @@ function saveState(st) {
     fs.renameSync(tmp, STATE_FILE);
 }
 
-const UNSTAMPED = { must: [{ is_empty: { key: 'eid' } }, { is_empty: { key: 'eid_missing' } }] };
+const UNSTAMPED = { must: [{ is_empty: { key: 'eid' } }, { is_empty: { key: 'eid_missing' } }, { is_empty: { key: 'eid_shared' } }] };
 
 async function listShows() {
     const points = await qdrant.scrollAll(cfg.EPISODES_COLLECTION, CATCH_UP ? UNSTAMPED : null, ['podcast_id'], 5000);
@@ -82,7 +84,7 @@ async function main() {
         st = {
             startedAt: new Date().toISOString(), shows, idx: 0,
             showsDone: 0, points: 0, stamped: 0, missing: 0, duplicates: 0, rows: 0,
-            piErrors: 0, bookErrors: 0, failedShows: [],
+            shared: 0, piErrors: 0, bookErrors: 0, failedShows: [],
         };
         saveState(st);
         log.info(`Bootstrap${CATCH_UP ? ' catch-up' : ''}: ${shows.length} shows ${CATCH_UP ? 'with unstamped points' : 'with episode points'}`);
@@ -105,7 +107,15 @@ async function main() {
             saveState(st);
             continue;
         }
-        const plan = planShow(sid, points, items);
+        // PI IDs another show already serves (duplicate feeds) are left to that show. A dry run on a
+        // database without the book yet has no owners.
+        let owners = new Map();
+        try {
+            owners = await loadServedOwners(db, points.map((p) => Number(p.payload?.id)).filter((n) => Number.isSafeInteger(n) && n > 0));
+        } catch (e) {
+            if (!(DRY && /no such table: episode_book/i.test(String(e.message)))) throw e;
+        }
+        const plan = planShow(sid, points, items, owners);
         if (!DRY && points.length) {
             const res = await bootstrapShow(db, sid, plan.items, { now: Math.floor(Date.now() / 1000) });
             if (!res.ok) {
@@ -128,6 +138,7 @@ async function main() {
         st.stamped += plan.stamps.length;
         st.missing += plan.missing.length;
         st.duplicates += plan.duplicates;
+        st.shared = (st.shared || 0) + plan.shared.length;
         st.rows += plan.items.length;
         st.idx++;
         saveState(st);
@@ -136,6 +147,7 @@ async function main() {
     const pct = st.points ? ((100 * st.stamped) / st.points).toFixed(2) : '0';
     log.info(`${DRY ? 'DRY ' : ''}Bootstrap shows ${from}..${st.idx} of ${st.shows.length}: ` +
         `${st.points} points, ${st.stamped} mapped (${pct}%), ${st.missing} missing, ${st.duplicates} duplicates, ` +
+        `${st.shared || 0} shared with a duplicate feed, ` +
         `${st.rows} rows, PI errors ${st.piErrors}, book errors ${st.bookErrors}, PI calls ${pi.getApiCallCount()}`);
     if (st.idx >= st.shows.length) log.info(CATCH_UP ? 'Bootstrap catch-up complete' : 'Bootstrap complete');
 }

scripts/sync/lib/book-bootstrap.js

@@ -10,6 +10,12 @@
  *   `eid_missing`: it keeps its identity and is retried later. A key is never guessed.
  * - Two points with the same key (PI re-listed the episode under a new ID) share one book row;
  *   the extra PI IDs become aliases. Stage 4's served-ID naming removes the duplicate later.
+ * - A point whose PI ID is already another key's served ID gets no row, so the show's transaction
+ *   never fails on it. The point belongs to the owner (it is named by that served ID):
+ *   - owner in this show (a changed GUID under the same PI ID): stamped with the owner's `eid` and
+ *     `catalog_key`, like any recorded point;
+ *   - owner in another show (a duplicate feed: PI lists one podcast under two show IDs that share
+ *     episode IDs): stamped `eid_shared` with the owning show.
  */
 
 const { catalogKey, recipeId } = require('./episode-recipe-id');
@@ -23,10 +29,11 @@ const num = (v) => {
  * @param {number} sid show ID
  * @param {{ id: string, payload: object }[]} points the show's Qdrant points
  * @param {object[]} piItems PI `episodes/byfeedid` items
+ * @param {Map<number, string>} [ownerByServed] who already holds these PI IDs as served IDs ("sid\u0000key")
  * @returns {{ items: object[], stamps: { pointId: string, eid: number, key: string }[],
- *   missing: string[], duplicates: number }}
+ *   missing: string[], duplicates: number, shared: { pointId: string, ownerSid: number }[] }}
  */
-function planShow(sid, points, piItems) {
+function planShow(sid, points, piItems, ownerByServed = new Map()) {
     const byPiId = new Map();
     for (const e of piItems || []) {
         const id = num(e?.id);
@@ -35,6 +42,7 @@ function planShow(sid, points, piItems) {
     const items = new Map();
     const stamps = [];
     const missing = [];
+    const shared = [];
     let duplicates = 0;
     for (const p of points) {
         const piId = num(p.payload?.id);
@@ -44,6 +52,15 @@ function planShow(sid, points, piItems) {
             missing.push(p.id);
             continue;
         }
+        const owner = ownerByServed.get(piId);
+        if (owner != null && owner !== `${sid}\u0000${key}`) {
+            const cut = owner.indexOf('\u0000');
+            const ownerSid = Number(owner.slice(0, cut));
+            const ownerKey = owner.slice(cut + 1);
+            if (ownerSid === sid) stamps.push({ pointId: p.id, eid: recipeId(sid, ownerKey), key: ownerKey });
+            else shared.push({ pointId: p.id, ownerSid });
+            continue;
+        }
         const eid = recipeId(sid, key);
         const existing = items.get(key);
         if (existing) {
@@ -67,13 +84,16 @@ function planShow(sid, points, piItems) {
         }
         stamps.push({ pointId: p.id, eid, key });
     }
-    return { items: [...items.values()], stamps, missing, duplicates };
+    return { items: [...items.values()], stamps, missing, duplicates, shared };
 }
 
 /** Qdrant batch operations that stamp a show's points (one value set per point). */
 function stampOperations(plan) {
     const ops = plan.stamps.map((s) => ({ set_payload: { payload: { eid: s.eid, catalog_key: s.key }, points: [s.pointId] } }));
     if (plan.missing.length) ops.push({ set_payload: { payload: { eid_missing: true }, points: plan.missing } });
+    const byOwner = new Map();
+    for (const { pointId, ownerSid } of plan.shared || []) byOwner.set(ownerSid, [...(byOwner.get(ownerSid) || []), pointId]);
+    for (const [ownerSid, ids] of byOwner) ops.push({ set_payload: { payload: { eid_shared: ownerSid }, points: ids } });
     return ops;
 }
 

scripts/sync/lib/book-bootstrap.test.js

@@ -46,4 +46,20 @@ describe('planShow', () => {
             { set_payload: { payload: { eid_missing: true }, points: ['u3'] } },
         ]);
     });
+
+    it('stamps a point whose ID another key of the same show serves with that key, without a row', () => {
+        const owners = new Map([[61, `${SID}\u0000old-guid`]]);
+        const plan = planShow(SID, [point('u1', 61), point('u2', 62)], [item(61, 'new-guid'), item(62, 'g-62')], owners);
+        assert.deepEqual(plan.shared, [], 'same show: never eid_shared');
+        assert.deepEqual(plan.items.map((x) => x.key), ['g-62'], 'no second row for the served ID');
+        assert.deepEqual(plan.stamps.find((x) => x.pointId === 'u1'), { pointId: 'u1', eid: recipeId(SID, 'old-guid'), key: 'old-guid' });
+    });
+
+    it('leaves a duplicate feed\'s episodes to the show that holds their IDs, and marks them', () => {
+        const owners = new Map([[61, '7650363\u0000g'], [62, `${SID}\u0000g-62`]]);
+        const plan = planShow(SID, [point('u1', 61), point('u2', 62), point('u4', 63)], [item(61, 'g'), item(62, 'g-62'), item(63, 'g-63')], owners);
+        assert.deepEqual(plan.shared, [{ pointId: 'u1', ownerSid: 7650363 }]);
+        assert.deepEqual(plan.items.map((x) => x.key), ['g-62', 'g-63'], 'its own episodes are still recorded');
+        assert.deepEqual(stampOperations(plan).at(-1), { set_payload: { payload: { eid_shared: 7650363 }, points: ['u1'] } });
+    });
 });

scripts/sync/lib/episode-identity.js

@@ -10,9 +10,14 @@
  *   written before the vector, so identity is committed first.
  * - An item with no GUID and no enclosure keeps today's naming and is stamped `eid_missing`.
  * - Two items with one key in a window: the first wins (PI's order; feeds use document order).
- * - A new key whose PI ID is already served for another key (a publisher changed the GUID and PI
- *   kept its ID) is deferred: no book row, today's naming, no stamps, so the rest of the show is still
- *   recorded and no served ID is ever taken from another episode (docs/EPISODE_IDS.md §4).
+ * - A new key whose PI ID is already served for another key is deferred: no book row, today's
+ *   naming, so the rest of the show is still recorded and no served ID is ever taken from another
+ *   episode (docs/EPISODE_IDS.md §4). Its point is the owner's (it is named by that served ID), so:
+ *   - owner in this show (the publisher changed the GUID, PI kept the ID): stamped with the owner's
+ *     `eid` and `catalog_key`;
+ *   - owner in another show (a duplicate feed: Podcast Index lists one podcast under two show IDs
+ *     that share episode IDs): stamped `eid_shared` with the owning show.
+ *   Either way the bootstrap's catch-up doesn't pick it up again.
  *
  * The bootstrap and stage 4 never overlap (both hold the shared job lock), so no per-show fence
  * is needed. COMPAT(own-ids): permanent.
@@ -58,7 +63,7 @@ function resolveIdentities(sid, episodes, servedByKey, ownerByServed = new Map()
         const known = servedByKey.get(key);
         if (known == null && piId != null && ownerByServed.has(piId) && ownerByServed.get(piId) !== `${sid}\u0000${key}`) {
             deferred++;
-            items.push({ raw, piId, servedId: piId, uuid: stableUUID(piId), stamps: null, deferred: true });
+            items.push({ raw, piId, servedId: piId, uuid: stableUUID(piId), stamps: ownerStamps(sid, ownerByServed.get(piId)), deferred: true });
             continue;
         }
         const servedId = known ?? piId;
@@ -82,6 +87,14 @@ function resolveIdentities(sid, episodes, servedByKey, ownerByServed = new Map()
     return { items, newRows, aliases, duplicates, renamed, deferred };
 }
 
+/** Stamps for a point whose served ID another (show, key) owns: that key's identity in this show, else `eid_shared`. */
+function ownerStamps(sid, owner) {
+    const cut = owner.indexOf('\u0000');
+    const ownerSid = Number(owner.slice(0, cut));
+    const ownerKey = owner.slice(cut + 1);
+    return ownerSid === sid ? { eid: recipeId(sid, ownerKey), catalog_key: ownerKey } : { eid_shared: ownerSid };
+}
+
 /** The book's served IDs for these keys of one show. */
 async function loadServedIds(db, sid, keys) {
     const out = new Map();
@@ -112,4 +125,4 @@ async function loadServedOwners(db, ids) {
     return out;
 }
 
-module.exports = { resolveIdentities, loadServedIds, loadServedOwners };
+module.exports = { resolveIdentities, loadServedIds, loadServedOwners, ownerStamps };

scripts/sync/lib/episode-identity.test.js

@@ -39,14 +39,22 @@ describe('resolveIdentities', () => {
 });
 
 describe('a changed GUID under the same Podcast Index ID', () => {
-    it('defers the new key instead of taking a served ID from another episode', () => {
+    it('defers the new key and stamps the point with the key that owns its served ID', () => {
         const owners = new Map([[61, `${SID}\u0000old-guid`]]);
         const r = resolveIdentities(SID, [ep(61, 'new-guid'), ep(62, 'other')], new Map(), owners);
         assert.equal(r.deferred, 1);
-        assert.deepEqual(r.items[0], { raw: r.items[0].raw, piId: 61, servedId: 61, uuid: stableUUID(61), stamps: null, deferred: true });
+        assert.deepEqual(r.items[0], { raw: r.items[0].raw, piId: 61, servedId: 61, uuid: stableUUID(61), stamps: { eid: recipeId(SID, 'old-guid'), catalog_key: 'old-guid' }, deferred: true });
         assert.deepEqual(r.newRows.map((x) => x.key), ['other'], 'the rest of the show is still recorded');
     });
 
+    it('marks an episode a duplicate feed already holds with the owning show', () => {
+        const owners = new Map([[61, `7650363\u0000g`]]);
+        const r = resolveIdentities(5893516, [ep(61, 'g')], new Map(), owners);
+        assert.equal(r.deferred, 1);
+        assert.deepEqual(r.items[0].stamps, { eid_shared: 7650363 });
+        assert.deepEqual(r.newRows, []);
+    });
+
     it('is not a clash when the ID is held by the same key', () => {
         const owners = new Map([[61, `${SID}\u0000g`]]);
         const r = resolveIdentities(SID, [ep(61, 'g')], new Map([['g', 61]]), owners);