feat(sync): latest-episode tips carry the episode's served ID

scripts/README.md

@@ -27,7 +27,7 @@ A Git push does not deploy this pipeline. Changes to `scripts/sync/` or `scripts
 | [`scripts/sync/lib/podcast-index.js`](sync/lib/podcast-index.js) | PI API client — Retry-After / global cooldown failsafes |
 | [`scripts/sync/lib/text.js`](sync/lib/text.js) | Description cleaning + embed text (**uncapped**; `PAYLOAD_DESCRIPTION_MAX` null) |
 | [`scripts/sync/lib/scalars.js`](sync/lib/scalars.js) | Scrub non-scalars before Qdrant/Turso writes |
-| [`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/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. The latest-episode update stores the episode's served ID in `latest_ep_id` (what Home, trending and bootstrap send), plus `latest_ep_key` and `latest_ep_eid`, from `tipIdentity` in `lib/episode-identity.js` (by catalog key, else Podcast Index's ID as an alias in the same show, else that ID). Change detection and the tip queue keep Podcast Index's own ID |
 | [`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`. 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/03-sync-episodes.js

@@ -18,6 +18,8 @@ const turso = require('./lib/turso');
 const pi = require('./lib/podcast-index');
 const state = require('./lib/state');
 const { tipUpdateArgs } = require('./lib/catalog-rows');
+const { bookDb } = require('./lib/episode-book');
+const { tipIdentity } = require('./lib/episode-identity');
 const cfg = require('./lib/config');
 const { handoffList } = require('./lib/vector-window');
 const handoff = require('./lib/pi-handoff');
@@ -112,6 +114,7 @@ async function countDueByCountry(ids) {
 
 async function main() {
     turso.assertEnv();
+    const book = bookDb(turso);
     pi.assertEnv();
     turso.beginStep('sync-episodes');
     await turso.healthCheck();
@@ -233,16 +236,25 @@ async function main() {
                 // Prefer cached medium from import/prior sync; never spend a PI call on it.
                 const medium = rec.m || 'podcast';
 
+                // The tip carries the episode's served ID (own-episode-IDs P2); change detection and the
+                // tip queue keep PI's own ID.
+                let identity = null;
+                try {
+                    identity = await tipIdentity(book, Number(podId), latest);
+                } catch (e) {
+                    log.warn(`Show ${podId}: tip identity lookup failed (${e.message}); using its PI ID`);
+                }
                 await turso.execute(`
                     UPDATE podcasts SET
                         latest_ep_id = ?, latest_ep_title = ?, latest_ep_date = ?,
                         latest_ep_duration = ?, latest_ep_url = ?, latest_ep_image = ?,
                         latest_ep_type = ?, latest_ep_description = ?,
                         latest_ep_chapters_url = ?, latest_ep_transcript_url = ?,
                         latest_ep_persons = ?, latest_ep_transcripts = ?,
-                        medium = ?, last_ep_sync = ?, qdrant_vectorized = 0
+                        medium = ?, last_ep_sync = ?, latest_ep_key = ?, latest_ep_eid = ?,
+                        qdrant_vectorized = 0
                     WHERE id = ?
-                `, tipUpdateArgs(latest, medium, Date.now(), podId));
+                `, tipUpdateArgs(latest, medium, Date.now(), podId, identity));
                 await tipQueue.upsert(turso, podId, latest.id);
                 updated++;
                 tipReflags++;

scripts/sync/lib/catalog-rows.js

@@ -1,5 +1,7 @@
 'use strict';
 
+const turso = require('./turso');
+
 /**
  * Catalog row values built from Podcast Index data, scrubbed before they are bound.
  *
@@ -60,11 +62,13 @@ function apiImportArgs(feed, itunesId, categories) {
 
 /**
  * Stage 3: values for the latest-episode UPDATE, in the statement's SET order
- * (latest_ep_id … medium, last_ep_sync), followed by the show ID.
+ * (latest_ep_id … medium, last_ep_sync, latest_ep_key, latest_ep_eid), followed by the show ID.
+ * `identity` (lib/episode-identity.js tipIdentity) puts the episode's served ID in latest_ep_id, the
+ * ID every app build sees on Home, trending and bootstrap; without it, PI's ID is used as before.
  */
-function tipUpdateArgs(latest, medium, nowMs, podId) {
+function tipUpdateArgs(latest, medium, nowMs, podId, identity = null) {
     return [
-        episodeIdText(latest.id),
+        episodeIdText(identity?.servedId ?? latest.id),
         str(latest.title),
         asNonNegInt(latest.datePublished, 0),
         asNonNegInt(latest.duration, 0),
@@ -78,6 +82,8 @@ function tipUpdateArgs(latest, medium, nowMs, podId) {
         jsonOrNull(latest.transcripts),
         str(medium) || 'podcast',
         nowMs,
+        turso.text(identity?.key ?? null),
+        identity?.eid ?? null,
         podId,
     ];
 }

scripts/sync/lib/catalog-rows.test.js

@@ -73,7 +73,14 @@ describe('tipUpdateArgs', () => {
         const args = tipUpdateArgs(latest, 'podcast', 1234, '42');
         assert.deepEqual(args, ['9001', 'Ep 1', 1790000000, 1800, 'https://cdn.example/1.mp3',
             'https://img.example/f.jpg', 'audio/mp4', 'Hi & bye', 'https://cdn.example/ch.json', null,
-            JSON.stringify([{ name: 'Host', role: 'host' }]), null, 'podcast', 1234, '42']);
+            JSON.stringify([{ name: 'Host', role: 'host' }]), null, 'podcast', 1234, turso.text(null), null, '42']);
+    });
+
+    it('puts the served ID in the tip and keeps the catalog key as text', () => {
+        const args = tipUpdateArgs(latest, 'podcast', 1234, '42', { servedId: 4503599627370777, key: '00123', eid: 4503599627370777 });
+        assert.equal(args[0], '4503599627370777');
+        assert.deepEqual(bindAll(args.slice(14, 16)), [{ type: 'text', value: '00123' }, { type: 'integer', value: '4503599627370777' }]);
+        assert.equal(args.at(-1), '42');
     });
 
     it('fails on an invalid episode id instead of storing it as text', () => {

scripts/sync/lib/episode-identity.js

@@ -95,6 +95,26 @@ function ownerStamps(sid, owner) {
     return ownerSid === sid ? { eid: recipeId(sid, ownerKey), catalog_key: ownerKey } : { eid_shared: ownerSid };
 }
 
+/**
+ * Stage 3 tip identity: the show's latest PI item under its served ID (by catalog key, else through
+ * its PI ID as an alias in this show), with its key and recipe ID; PI's own ID when the book doesn't
+ * know the episode yet (stage 4 records it later in the same run, with that ID).
+ * @returns {Promise<{ servedId: number|null, key: string|null, eid: number|null }>}
+ */
+async function tipIdentity(db, sid, item) {
+    const piId = num(item?.id);
+    const key = catalogKey(item?.guid, item?.enclosureUrl);
+    const eid = key ? recipeId(sid, key) : null;
+    let servedId = key ? (await loadServedIds(db, sid, [key])).get(key) ?? null : null;
+    if (servedId == null && piId != null) {
+        const rows = await db.query(
+            `SELECT b.served_id FROM episode_pi_alias a JOIN episode_book b ON b.rid = a.rid
+             WHERE a.pi_id = ? AND b.sid = ? AND b.served_id IS NOT NULL`, [piId, sid]);
+        if (rows.length) servedId = Number(rows[0][0]);
+    }
+    return { servedId: servedId ?? piId, key, eid };
+}
+
 /** The book's served IDs for these keys of one show. */
 async function loadServedIds(db, sid, keys) {
     const out = new Map();
@@ -125,4 +145,4 @@ async function loadServedOwners(db, ids) {
     return out;
 }
 
-module.exports = { resolveIdentities, loadServedIds, loadServedOwners, ownerStamps };
+module.exports = { resolveIdentities, loadServedIds, loadServedOwners, ownerStamps, tipIdentity };

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

@@ -93,3 +93,31 @@ describe('loadServedIds', () => {
         assert.deepEqual([...owners.entries()].sort(), [[61, `${SID}\u000000123`], [62, `${SID}\u0000b`]]);
     });
 });
+
+describe('stage 3 tip identity', () => {
+    const { DatabaseSync } = require('node:sqlite');
+    const { tipIdentity } = require('./episode-identity');
+    const { BOOK_TABLES } = require('./episode-book');
+    function book() {
+        const raw = new DatabaseSync(':memory:');
+        for (const sql of BOOK_TABLES) raw.exec(sql);
+        const toArgs = (args = []) => args.map((a) => (a && typeof a === 'object' && 'value' in a ? a.value : a));
+        return {
+            raw,
+            text: (v) => ({ value: v === null || v === undefined ? null : String(v) }),
+            query: async (sql, args) => raw.prepare(sql).all(...toArgs(args)).map((r) => Object.values(r)),
+        };
+    }
+
+    it('uses the served ID by key, then by PI alias, else the PI ID', async () => {
+        const db = book();
+        db.raw.exec(`INSERT INTO episode_book (rid, eid, sid, catalog_key, served_id, source, first_seen_at, updated_at)
+            VALUES (1, 1, ${SID}, 'feed-only', 4503599627370777, 'feed', 1, 1), (2, 2, ${SID}, 'renamed', 61, 'bootstrap', 1, 1)`);
+        db.raw.exec('INSERT INTO episode_pi_alias (pi_id, rid, seen_at) VALUES (61, 2, 1), (777, 2, 1)');
+        assert.deepEqual(await tipIdentity(db, SID, { id: 999, guid: 'feed-only' }),
+            { servedId: 4503599627370777, key: 'feed-only', eid: recipeId(SID, 'feed-only') });
+        assert.equal((await tipIdentity(db, SID, { id: 777, guid: 'changed-guid' })).servedId, 61, 'PI renumbered it');
+        assert.equal((await tipIdentity(db, SID, { id: 555, guid: 'new' })).servedId, 555, 'not in the book yet');
+        assert.equal((await tipIdentity(db, 1, { id: 777, guid: 'x' })).servedId, 777, 'an alias in another show never counts');
+    });
+});