diff --git a/scripts/README.md b/scripts/README.md index 7a27ac4de..6a40d2a25 100644 --- a/scripts/README.md +++ b/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` | diff --git a/scripts/sync/bootstrap-book.js b/scripts/sync/bootstrap-book.js index 96c25f01b..6f74a84e7 100644 --- a/scripts/sync/bootstrap-book.js +++ b/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'); } diff --git a/scripts/sync/lib/book-bootstrap.js b/scripts/sync/lib/book-bootstrap.js index 0acf9b3b4..86629a4e4 100644 --- a/scripts/sync/lib/book-bootstrap.js +++ b/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} [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; } diff --git a/scripts/sync/lib/book-bootstrap.test.js b/scripts/sync/lib/book-bootstrap.test.js index 86b49a848..238133e22 100644 --- a/scripts/sync/lib/book-bootstrap.test.js +++ b/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'] } }); + }); }); diff --git a/scripts/sync/lib/episode-identity.js b/scripts/sync/lib/episode-identity.js index 78d960e19..fd9cb7116 100644 --- a/scripts/sync/lib/episode-identity.js +++ b/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 }; diff --git a/scripts/sync/lib/episode-identity.test.js b/scripts/sync/lib/episode-identity.test.js index 4a36dfe37..193668942 100644 --- a/scripts/sync/lib/episode-identity.test.js +++ b/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);