fix(sync): mark shows whose Podcast Index IDs differ from their served IDs for the overlay

scripts/README.md

@@ -30,7 +30,7 @@ A Git push does not deploy this pipeline. Changes to `scripts/sync/` or `scripts
 | [`scripts/sync/lib/catalog-rows.js`](sync/lib/catalog-rows.js) | Scrubbed values for stage 2's API import and stage 3's latest-episode update: an odd optional field from Podcast Index becomes empty instead of failing the write |
 | [`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-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`. Every book write also marks the show in `served_show` when one of its episodes is listed by Podcast Index under an ID other than its served ID (renumbered, duplicated or recipe-served), so the Worker's v1 overlay translates that ID back. `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 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/sync/lib/episode-book.js

@@ -43,7 +43,9 @@ const BOOK_TABLES = [
     `CREATE TABLE IF NOT EXISTS episode_pi_alias (
         pi_id INTEGER PRIMARY KEY, rid INTEGER NOT NULL, seen_at INTEGER NOT NULL
     )`,
-    // Shows where any episode was served with a recipe ID (drives the Worker's v1 overlay).
+    // Shows where some episode's served ID differs from a Podcast Index ID listed for it: served
+    // under a recipe ID, or PI renumbered or duplicated it. Drives the Worker's v1 overlay, which
+    // translates those PI IDs back to the served ID.
     `CREATE TABLE IF NOT EXISTS served_show (
         sid INTEGER PRIMARY KEY, first_served_at INTEGER NOT NULL
     )`,
@@ -142,10 +144,26 @@ async function bootstrapShow(db, sid, items, { now, source = 'bootstrap' }) {
             args: [it.servedId, sid, db.text(it.key)],
         });
     }
+    stmts.push(servedShowStatement(sid, now));
     const res = await db.atomic(stmts);
     return { ok: res.ok, error: res.error };
 }
 
+/**
+ * Marks the show in `served_show` when one of its episodes is listed by Podcast Index under an ID
+ * other than its served ID (a renumbered or duplicated episode, or a recipe-served one), so the
+ * Worker's v1 overlay translates that PI ID back to the served ID. Runs in the writer's transaction.
+ * COMPAT(own-ids): permanent.
+ */
+function servedShowStatement(sid, now) {
+    return {
+        sql: `INSERT OR IGNORE INTO served_show (sid, first_served_at)
+              SELECT ?, ? WHERE EXISTS (SELECT 1 FROM episode_pi_alias a JOIN episode_book b ON b.rid = a.rid
+                WHERE b.sid = ? AND b.served_id IS NOT NULL AND a.pi_id != b.served_id)`,
+        args: [sid, now, sid],
+    };
+}
+
 /**
  * Bootstrap: records the ID an existing point already serves (its Podcast Index ID). Set once;
  * a row that already has a served ID keeps it. Fails, changing nothing, if another row holds it.

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

@@ -100,6 +100,15 @@ describe('episode book', () => {
         assert.deepEqual(await db.query('SELECT COUNT(*) FROM served_show'), [[0]], 'a clash writes nothing');
     });
 
+    it('marks the show for the overlay when a PI ID differs from the served ID, and only then', async () => {
+        await book.bootstrapShow(db, 9, [ep('a', { piId: 61, servedId: 61 })], { now: 5 });
+        assert.deepEqual(await db.query('SELECT COUNT(*) FROM served_show'), [[0]], 'PI ID = served ID: nothing to translate');
+        await book.bootstrapShow(db, 9, [ep('a', { piId: 75 })], { now: 6, source: 'pi' });
+        assert.deepEqual(await db.query('SELECT sid, first_served_at FROM served_show'), [[9, 6]], 'PI renumbered the episode');
+        await book.bootstrapShow(db, 11, [ep('b', { piId: 80, servedId: 80, extraPiIds: [81] })], { now: 7 });
+        assert.deepEqual(await db.query('SELECT sid FROM served_show ORDER BY sid'), [[9], [11]], 'PI listed it twice');
+    });
+
     it('bootstraps a show in one transaction, or not at all', async () => {
         const items = [ep('a', { piId: 61, servedId: 61, extraPiIds: [70] }), ep('b', { piId: 62, servedId: 62 })];
         assert.deepEqual(await book.bootstrapShow(db, 9, items, { now: 5 }), { ok: true, error: null });