From f1b85fae10080b7dffe4ed71b77efd4dcd0c4709 Mon Sep 17 00:00:00 2001 From: Aswin C Date: Fri, 23 Oct 2026 02:11:30 +0530 Subject: [PATCH 1/2] feat(sync): record each vector's last chart day --- scripts/README.md | 1 + scripts/package.json | 2 +- scripts/sync/04-vectorize-episodes.js | 3 + scripts/sync/05-vectorize-shows.js | 3 + scripts/sync/06-cleanup-non-chart.js | 32 +++++++- scripts/sync/lib/charted-at.js | 109 ++++++++++++++++++++++++++ scripts/sync/lib/charted-at.test.js | 88 +++++++++++++++++++++ scripts/sync/lib/config.js | 3 + scripts/sync/lib/qdrant.js | 10 ++- scripts/sync/stamp-charted-at.js | 62 +++++++++++++++ 10 files changed, 308 insertions(+), 5 deletions(-) create mode 100644 scripts/sync/lib/charted-at.js create mode 100644 scripts/sync/lib/charted-at.test.js create mode 100644 scripts/sync/stamp-charted-at.js diff --git a/scripts/README.md b/scripts/README.md index 0d1e68684..be67c82bc 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -39,6 +39,7 @@ A Git push does not deploy this pipeline. Changes to `scripts/sync/` or `scripts | [`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. That check reaches a show's aliases through `episode_pi_alias_rid` (an index on `rid`); without it each check scanned every alias inside the write. `ensure` creates the tables, indexes 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`. `--retry-missing` (own-episode-IDs P3; monthly on the VPS) runs the pass again for shows with `eid_missing` points: a point mapped this time gets its stamp and loses the mark | +| [`scripts/sync/lib/charted-at.js`](sync/lib/charted-at.js) + [`stamp-charted-at.js`](sync/stamp-charted-at.js) | Last-charted dates on Qdrant points (own-episode-IDs P3b): every point in both collections carries `charted_at`, the UTC day (epoch seconds) its show was last seen on a chart, so a 90-day vector retention can outlive the show's catalog row and state record. Stage 6 stamps every show it keeps from its chart clock (`shows[id].s`) before deleting; a point is rewritten only when it has no date or one more than `CHARTED_AT_REFRESH_DAYS` (default 7) older than its show's, so dates only move forward and a charting show's points are rewritten about weekly. Stages 4 and 5 give new points the day they are embedded. Failures are logged and never block the cleanup. `stamp-charted-at.js` stamps every point once from the catalog and the chart clock without changing anything else (`--dry` counts only; run from the repository root under the shared job lock). Nothing deletes by this date yet | | [`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/package.json b/scripts/package.json index e95ff76fb..b9a3a4192 100644 --- a/scripts/package.json +++ b/scripts/package.json @@ -16,7 +16,7 @@ "test:sync-scalars": "node --test sync/lib/scalars.test.js", "test:sync-turso-page": "node --test sync/lib/turso-page.test.js", "test:sync-turso": "node --test sync/lib/turso.test.js", - "test:sync": "node --test sync/lib/csv.test.js sync/lib/pi-unavailable.test.js sync/lib/scalars.test.js sync/lib/turso-page.test.js sync/lib/episode-caps.test.js sync/lib/staleness.test.js sync/lib/podcast-index.test.js sync/lib/chart-countries.test.js sync/lib/log.test.js sync/lib/text.test.js sync/lib/vectorize-lanes.test.js sync/lib/tip-queue.test.js sync/lib/import-missing.test.js sync/lib/episode-recipe-id.test.js sync/lib/attribution.test.js sync/lib/vector-window.test.js sync/lib/trim-episode-caps.test.js sync/lib/remediate-qdrant.test.js sync/lib/turso.test.js sync/lib/catalog-rows.test.js sync/lib/episode-book.test.js sync/lib/book-bootstrap.test.js sync/lib/episode-identity.test.js sync/lib/stale-tips.test.js sync/lib/show-window.test.js sync/lib/cadence.test.js", + "test:sync": "node --test sync/lib/csv.test.js sync/lib/pi-unavailable.test.js sync/lib/scalars.test.js sync/lib/turso-page.test.js sync/lib/episode-caps.test.js sync/lib/staleness.test.js sync/lib/podcast-index.test.js sync/lib/chart-countries.test.js sync/lib/log.test.js sync/lib/text.test.js sync/lib/vectorize-lanes.test.js sync/lib/tip-queue.test.js sync/lib/import-missing.test.js sync/lib/episode-recipe-id.test.js sync/lib/attribution.test.js sync/lib/vector-window.test.js sync/lib/trim-episode-caps.test.js sync/lib/remediate-qdrant.test.js sync/lib/turso.test.js sync/lib/catalog-rows.test.js sync/lib/episode-book.test.js sync/lib/book-bootstrap.test.js sync/lib/episode-identity.test.js sync/lib/stale-tips.test.js sync/lib/show-window.test.js sync/lib/cadence.test.js sync/lib/charted-at.test.js", "test:sync-text": "node --test sync/lib/text.test.js", "test:sync-recipe": "node --test sync/lib/episode-recipe-id.test.js", "test:sync-attribution": "node --test sync/lib/attribution.test.js", diff --git a/scripts/sync/04-vectorize-episodes.js b/scripts/sync/04-vectorize-episodes.js index 47b43ead2..4d39a26d8 100644 --- a/scripts/sync/04-vectorize-episodes.js +++ b/scripts/sync/04-vectorize-episodes.js @@ -48,6 +48,7 @@ const { const scalars = require('./lib/scalars'); const showWindow = require('./lib/show-window'); const attribution = require('./lib/attribution'); +const chartedAt = require('./lib/charted-at'); const UPSERT_BATCH = 100; @@ -508,6 +509,8 @@ async function main() { continue; } if (item.stamps) Object.assign(prepared.payload, item.stamps); + // Embedded shows chart today (lib/charted-at.js); the cleanup moves the date on. + prepared.payload[chartedAt.FIELD] = chartedAt.dayStartSec(Date.now()); if (prepared.scrubbed.length) { log.warn( `Scrubbed optional ep fields for show ${pod.id}: ${prepared.scrubbed.join(',')}`, diff --git a/scripts/sync/05-vectorize-shows.js b/scripts/sync/05-vectorize-shows.js index adaf92f4c..84525522e 100644 --- a/scripts/sync/05-vectorize-shows.js +++ b/scripts/sync/05-vectorize-shows.js @@ -16,6 +16,7 @@ const text = require('./lib/text'); const cfg = require('./lib/config'); const scalars = require('./lib/scalars'); const attribution = require('./lib/attribution'); +const chartedAt = require('./lib/charted-at'); const { loadCountriesByItunesId, itunesInCountries, @@ -212,6 +213,8 @@ async function main() { `Scrubbed optional show fields for ${pod.id}: ${prepared.scrubbed.join(',')}`, ); } + // Embedded shows chart today (lib/charted-at.js); the cleanup moves the date on. + prepared.payload[chartedAt.FIELD] = chartedAt.dayStartSec(Date.now()); pointsQueue.push({ id: uuid, vector, diff --git a/scripts/sync/06-cleanup-non-chart.js b/scripts/sync/06-cleanup-non-chart.js index f1f6b1e00..213f7aecd 100644 --- a/scripts/sync/06-cleanup-non-chart.js +++ b/scripts/sync/06-cleanup-non-chart.js @@ -18,6 +18,9 @@ * - Episode-cap trim is best-effort: failure is logged but does not roll * back a successful non-chart cleanup. * + * Last-charted dates (lib/charted-at.js, P3b): before deleting, every kept show's Qdrant points + * get the day it was last seen on a chart. Best-effort: a failure never blocks the cleanup. + * * FTS safety: the podcasts_ad delete trigger does a full FTS-content scan * PER ROW, so we drop it, bulk-delete, clean FTS orphans in one scan, and * recreate it. @@ -33,6 +36,7 @@ const handoff = require('./lib/pi-handoff'); const { trimEpisodeCaps } = require('./lib/trim-episode-caps'); const { CHARTS_ITUNES_FIRST_CURSOR } = require('./lib/chart-countries'); const attribution = require('./lib/attribution'); +const chartedAt = require('./lib/charted-at'); const QDRANT_CHUNK = 500; const DELETE_CHUNK = 200; @@ -50,6 +54,24 @@ async function runCapTrim() { } } +/** Stamps kept shows' points with their last chart day. Never throws; returns the report line. */ +async function stampChartedAt(pods, shows) { + try { + for (const c of chartedAt.COLLECTIONS) await qdrant.ensurePayloadIndex(c.name, chartedAt.FIELD, 'integer'); + const r = await chartedAt.stamp(qdrant, chartedAt.planBatches(pods, shows), { + collections: chartedAt.COLLECTIONS, + refreshSec: cfg.CHARTED_AT_REFRESH_DAYS * chartedAt.DAY_S, + warn: (m) => log.warn(m), + }); + const line = chartedAt.summary(r); + log.info(`[CHARTED_AT] ${line}`); + return line; + } catch (e) { + log.warn(`[CHARTED_AT] skipped: ${e.message}`); + return 'charted_at skipped'; + } +} + async function main() { turso.assertEnv(); qdrant.assertEnv(); @@ -149,6 +171,10 @@ async function main() { // Persist grace clocks even when nothing is deleted yet. state.save(st); + // Last-charted dates for the shows that stay (points of shows deleted below go anyway). + const deleting = new Set(toDelete.map(p => p.id)); + const chartedLine = await stampChartedAt(allPods.filter(p => !deleting.has(p.id)), st.shows); + if (toDelete.length === 0) { log.info('Nothing past the grace period - done'); const trim = await runCapTrim(); @@ -156,7 +182,7 @@ async function main() { stage: 'cleanup-non-chart', reads: turso.getStats().reads, writes: turso.getStats().writes, - detail: `0 deleted, ${inGrace} in grace, cap-trim ${trim.deletedPoints} pts${trim.failed ? ' (trim failed)' : ''}`, + detail: `0 deleted, ${inGrace} in grace, cap-trim ${trim.deletedPoints} pts${trim.failed ? ' (trim failed)' : ''} · ${chartedLine}`, }]); if (trim.failed) process.exitCode = 1; return; @@ -206,7 +232,7 @@ async function main() { stage: 'cleanup-non-chart', reads: turso.getStats().reads, writes: turso.getStats().writes, - detail: `0 turso deleted (${toDelete.length} pending qdrant), cap-trim ${trim.deletedPoints} pts`, + detail: `0 turso deleted (${toDelete.length} pending qdrant), cap-trim ${trim.deletedPoints} pts · ${chartedLine}`, }]); process.exitCode = 1; return; @@ -276,7 +302,7 @@ async function main() { stage: 'cleanup-non-chart', reads: stats.reads, writes: stats.writes, - detail: `${qdrantOk.length}/${toDelete.length} shows deleted, ${inGrace} in grace, ${pruned} state pruned, cap-trim ${trim.deletedPoints} pts${trim.failed || qdrantFailedChunks ? ' (partial/errors)' : ''}`, + detail: `${qdrantOk.length}/${toDelete.length} shows deleted, ${inGrace} in grace, ${pruned} state pruned, cap-trim ${trim.deletedPoints} pts${trim.failed || qdrantFailedChunks ? ' (partial/errors)' : ''} · ${chartedLine}`, }]); if (trim.failed || qdrantFailedChunks > 0) process.exitCode = 1; diff --git a/scripts/sync/lib/charted-at.js b/scripts/sync/lib/charted-at.js new file mode 100644 index 000000000..c9bb4092e --- /dev/null +++ b/scripts/sync/lib/charted-at.js @@ -0,0 +1,109 @@ +'use strict'; + +/** + * Last-charted dates on Qdrant points (own-episode-IDs P3b, plan section 10b item 4). + * + * Every point in both collections carries `charted_at`: the UTC day (epoch seconds) its show was + * last seen on a chart. The show's catalog row and state record go 7 days after it leaves the + * charts, so the date has to live on the points for the 90-day vector retention to read it. + * Nothing here deletes: the cleanup keeps its 7-day rule until the 90-day rule is switched on. + * + * Stamps come from the cleanup's chart clock (state `shows[id].s`, set daily by stage 6): + * - a show's points get the day of its last chart sighting; + * - a point is rewritten only when it has no date or its date is more than REFRESH_DAYS older + * than the show's, so dates only move forward and a charting show's points are rewritten + * about once a week, not daily; + * - new points get the day they were embedded (stage 4 and 5), since those shows chart. + */ + +const cfg = require('./config'); + +const FIELD = 'charted_at'; +/** Both collections: episodes name their show in podcast_id, show points in id. */ +const COLLECTIONS = [ + { name: cfg.EPISODES_COLLECTION, showKey: 'podcast_id' }, + { name: cfg.PODCASTS_COLLECTION, showKey: 'id' }, +]; +const DAY_S = 86_400; +/** Shows per Qdrant filter call. */ +const CHUNK = 500; + +/** UTC day start in epoch seconds for a time in ms. */ +function dayStartSec(ms) { + return Math.floor(Number(ms) / 1000 / DAY_S) * DAY_S; +} + +/** + * Groups shows by stamp day into filter-sized batches. + * @param {{ id: string|number }[]} pods + * @param {object} shows state.shows (rec.s = last chart sighting, ms) + * @returns {{ value: number, ids: number[] }[]} newest day first; shows without a sighting are skipped + */ +function planBatches(pods, shows, chunk = CHUNK) { + const byDay = new Map(); + for (const pod of pods || []) { + const id = Number(pod.id); + const seen = shows?.[String(pod.id)]?.s; + if (!Number.isSafeInteger(id) || id <= 0 || !(seen > 0)) continue; + const day = dayStartSec(seen); + if (!byDay.has(day)) byDay.set(day, []); + byDay.get(day).push(id); + } + const out = []; + for (const day of [...byDay.keys()].sort((a, b) => b - a)) { + const ids = byDay.get(day).sort((a, b) => a - b); + for (let i = 0; i < ids.length; i += chunk) out.push({ value: day, ids: ids.slice(i, i + chunk) }); + } + return out; +} + +/** + * Points of these shows whose date is missing or more than refreshSec older than `value`. + * @param {'podcast_id'|'id'} showKey episodes use podcast_id, the podcasts collection id + */ +function staleFilter(showKey, ids, value, refreshSec) { + return { + must: [{ key: showKey, match: { any: ids } }], + should: [ + { is_empty: { key: FIELD } }, + { key: FIELD, range: { lt: value - refreshSec } }, + ], + }; +} + +/** + * Stamps both collections. Never throws: a failed batch is counted and the rest still run. + * @param {{ setPayloadByFilter: Function, countPoints: Function }} qdrant + * @param {{ value: number, ids: number[] }[]} batches + * @param {{ collections: { name: string, showKey: string }[], refreshSec: number, dry?: boolean, + * warn?: (msg: string) => void }} opts + * @returns {Promise<{ batches: number, failed: number, points: Record, dry: boolean }>} + * points: per collection, points that needed a date (counted before writing) + */ +async function stamp(qdrant, batches, { collections, refreshSec, dry = false, warn = () => {} }) { + const out = { batches: batches.length, failed: 0, points: {}, dry }; + for (const c of collections) out.points[c.name] = 0; + for (const b of batches) { + for (const c of collections) { + const filter = staleFilter(c.showKey, b.ids, b.value, refreshSec); + try { + const n = await qdrant.countPoints(c.name, filter); + out.points[c.name] += n; + if (!dry && n > 0) await qdrant.setPayloadByFilter(c.name, { [FIELD]: b.value }, filter, true); + } catch (e) { + out.failed++; + warn(`[CHARTED_AT] ${c.name}: ${b.ids.length} shows at ${new Date(b.value * 1000).toISOString().slice(0, 10)} not stamped (${e.message})`); + } + } + } + return out; +} + +/** One line for the run report. */ +function summary(r) { + const pts = Object.entries(r.points).map(([name, n]) => `${name} ${n}`).join(', '); + return `charted_at ${r.dry ? 'dry run' : 'stamped'}: ${pts} points in ${r.batches} batches` + + (r.failed ? `, ${r.failed} failed calls` : ''); +} + +module.exports = { FIELD, COLLECTIONS, CHUNK, DAY_S, dayStartSec, planBatches, staleFilter, stamp, summary }; diff --git a/scripts/sync/lib/charted-at.test.js b/scripts/sync/lib/charted-at.test.js new file mode 100644 index 000000000..b4dd66fb0 --- /dev/null +++ b/scripts/sync/lib/charted-at.test.js @@ -0,0 +1,88 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); +const { FIELD, COLLECTIONS, DAY_S, dayStartSec, planBatches, staleFilter, stamp, summary } = require('./charted-at'); + +const DAY = 1_792_627_200; // a UTC midnight, seconds +const ms = (s) => s * 1000; + +test('dayStartSec: the UTC day a time falls in', () => { + assert.equal(dayStartSec(ms(DAY)), DAY); + assert.equal(dayStartSec(ms(DAY + DAY_S - 1)), DAY); + assert.equal(dayStartSec(ms(DAY + DAY_S)), DAY + DAY_S); +}); + +test('planBatches: shows grouped by their last chart day, newest first, in bounded chunks', () => { + const shows = { + 1: { s: ms(DAY + 3600) }, + 2: { s: ms(DAY + 7200) }, + 3: { s: ms(DAY - 2 * DAY_S) }, + 4: {}, + 5: { s: ms(DAY + 10) }, + }; + const pods = [{ id: '5' }, { id: '1' }, { id: '3' }, { id: '4' }, { id: '2' }, { id: 'x' }, { id: '0' }]; + assert.deepEqual(planBatches(pods, shows, 2), [ + { value: DAY, ids: [1, 2] }, + { value: DAY, ids: [5] }, + { value: DAY - 2 * DAY_S, ids: [3] }, + ], 'shows without a chart day or a valid ID are skipped'); + assert.deepEqual(planBatches(null, shows), []); +}); + +test('staleFilter: only points of these shows with no date, or one older than the refresh window', () => { + assert.deepEqual(staleFilter('podcast_id', [7, 8], DAY, 7 * DAY_S), { + must: [{ key: 'podcast_id', match: { any: [7, 8] } }], + should: [{ is_empty: { key: FIELD } }, { key: FIELD, range: { lt: DAY - 7 * DAY_S } }], + }); +}); + +test('both collections are stamped, episodes by podcast_id and shows by id', () => { + assert.deepEqual(COLLECTIONS.map((c) => c.showKey), ['podcast_id', 'id']); +}); + +function fakeQdrant(counts, failOn = null) { + const calls = []; + return { + calls, + countPoints: async (name, filter) => { + calls.push(['count', name, filter]); + if (failOn && failOn(name, filter)) throw new Error('boom'); + return counts[name] ?? 0; + }, + setPayloadByFilter: async (name, payload, filter, wait) => { + calls.push(['set', name, payload, filter, wait]); + }, + }; +} + +const COLS = [{ name: 'episodes', showKey: 'podcast_id' }, { name: 'podcasts', showKey: 'id' }]; + +test('stamp: counts, then writes the day with wait; skips writes where nothing is stale', async () => { + const q = fakeQdrant({ episodes: 40, podcasts: 0 }); + const r = await stamp(q, [{ value: DAY, ids: [1, 2] }], { collections: COLS, refreshSec: 0 }); + assert.deepEqual(r, { batches: 1, failed: 0, points: { episodes: 40, podcasts: 0 }, dry: false }); + const sets = q.calls.filter((c) => c[0] === 'set'); + assert.equal(sets.length, 1, 'no write for a collection with nothing to stamp'); + assert.deepEqual(sets[0].slice(1, 3), ['episodes', { [FIELD]: DAY }]); + assert.deepEqual(sets[0][3], staleFilter('podcast_id', [1, 2], DAY, 0)); + assert.equal(sets[0][4], true); + assert.equal(summary(r), 'charted_at stamped: episodes 40, podcasts 0 points in 1 batches'); +}); + +test('stamp: a dry run only counts; a failed call is counted and the rest still run', async () => { + const dry = fakeQdrant({ episodes: 5, podcasts: 1 }); + const r = await stamp(dry, [{ value: DAY, ids: [1] }], { collections: COLS, refreshSec: 7 * DAY_S, dry: true }); + assert.equal(dry.calls.filter((c) => c[0] === 'set').length, 0); + assert.match(summary(r), /^charted_at dry run: episodes 5, podcasts 1 points/); + + const warned = []; + const q = fakeQdrant({ episodes: 3, podcasts: 1 }, (name, f) => name === 'episodes' && f.must[0].match.any[0] === 1); + const out = await stamp(q, [{ value: DAY, ids: [1] }, { value: DAY, ids: [2] }], { + collections: COLS, refreshSec: 0, warn: (m) => warned.push(m), + }); + assert.deepEqual([out.failed, out.points], [1, { episodes: 3, podcasts: 2 }]); + assert.equal(q.calls.filter((c) => c[0] === 'set').length, 3); + assert.match(warned[0], /episodes: 1 shows at \d{4}-\d{2}-\d{2} not stamped \(boom\)/); + assert.match(summary(out), /, 1 failed calls$/); +}); diff --git a/scripts/sync/lib/config.js b/scripts/sync/lib/config.js index cd5214209..f4698c8a8 100644 --- a/scripts/sync/lib/config.js +++ b/scripts/sync/lib/config.js @@ -164,6 +164,9 @@ module.exports = { // --- Cleanup --- CLEANUP_GRACE_DAYS: 7, + // Last-charted dates on Qdrant points (lib/charted-at.js, P3b): a point's date is rewritten + // only when it is more than this many days older than its show's last chart sighting. + CHARTED_AT_REFRESH_DAYS: Math.max(0, parseInt(process.env.CHARTED_AT_REFRESH_DAYS || '7', 10) || 0), CLEANUP_SAFETY_MIN_CHARTS: 500, // --- State / files --- diff --git a/scripts/sync/lib/qdrant.js b/scripts/sync/lib/qdrant.js index bae937374..0366ee825 100644 --- a/scripts/sync/lib/qdrant.js +++ b/scripts/sync/lib/qdrant.js @@ -231,6 +231,14 @@ async function deleteByFilter(collection, filter, wait = false) { }); } +/** Set payload fields on every point matching a filter (wait=true: durable before it returns). */ +async function setPayloadByFilter(collection, payload, filter, wait = true) { + await request(`/collections/${collection}/points/payload?wait=${wait}`, { + method: 'POST', + body: JSON.stringify({ payload, filter }), + }); +} + /** * Scroll all points matching a filter. Returns array of {id, payload}. * Payload selector defaults to false (ids only) to keep responses small. @@ -270,6 +278,6 @@ async function enableQuantization(collection) { module.exports = { assertEnv, stableUUID, request, collectionInfo, ensureCollection, dropCollection, payloadSchema, ensurePayloadIndex, ensureEpisodesCollection, ensurePodcastsCollection, - existingIds, payloadsByIds, facetCounts, countPoints, upsert, deleteByIds, deleteByFilter, scrollAll, + existingIds, payloadsByIds, facetCounts, countPoints, upsert, deleteByIds, deleteByFilter, setPayloadByFilter, scrollAll, enableQuantization, }; diff --git a/scripts/sync/stamp-charted-at.js b/scripts/sync/stamp-charted-at.js new file mode 100644 index 000000000..f206d21e5 --- /dev/null +++ b/scripts/sync/stamp-charted-at.js @@ -0,0 +1,62 @@ +#!/usr/bin/env node +'use strict'; + +/** + * One-off and catch-up: stamps every Qdrant point with its show's last chart day (`charted_at`, + * lib/charted-at.js, own-episode-IDs P3b). Stage 6 keeps the dates current afterwards. + * + * Reads the shows in the catalog and the cleanup's chart clock (state `shows[id].s`); never + * changes the state, the catalog or any vector. Points that already have a date of the show's day + * are skipped, so it can be stopped and run again. Run it under the shared job lock. + * + * node sync/stamp-charted-at.js --dry # count the points that need a date, write nothing + * node sync/stamp-charted-at.js # stamp them + */ + +const log = require('./lib/log'); +const turso = require('./lib/turso'); +const qdrant = require('./lib/qdrant'); +const state = require('./lib/state'); +const cfg = require('./lib/config'); +const chartedAt = require('./lib/charted-at'); + +async function main() { + const dry = process.argv.includes('--dry'); + turso.assertEnv(); + qdrant.assertEnv(); + turso.beginStep('stamp-charted-at'); + log.banner('Stamp last-charted dates', { Mode: dry ? 'dry run (counts only)' : 'stamp' }); + + const rows = await turso.fetchAllPaged({ + pageSize: cfg.TURSO_PAGE_SIZE, + rowId: (r) => Number(r[0]), + buildPage: (after, limit) => ({ + sql: 'SELECT id FROM podcasts WHERE id > ? ORDER BY id ASC LIMIT ?', + args: [after == null ? 0 : after, limit], + }), + }); + const st = state.load(); + const pods = rows.map((r) => ({ id: String(r[0]) })); + const batches = chartedAt.planBatches(pods, st.shows); + const withClock = pods.filter((p) => st.shows[p.id]?.s > 0).length; + log.info(`Shows: ${log.fmt(pods.length)} in the catalog, ${log.fmt(withClock)} with a chart day, ${batches.length} batches`); + + if (!dry) { + for (const c of chartedAt.COLLECTIONS) await qdrant.ensurePayloadIndex(c.name, chartedAt.FIELD, 'integer'); + } + const started = Date.now(); + const r = await chartedAt.stamp(qdrant, batches, { + collections: chartedAt.COLLECTIONS, + refreshSec: cfg.CHARTED_AT_REFRESH_DAYS * chartedAt.DAY_S, + dry, + warn: (m) => log.warn(m), + }); + log.info(`[CHARTED_AT] ${chartedAt.summary(r)} (${Math.round((Date.now() - started) / 1000)} s)`); + turso.flushStats(); + if (r.failed) process.exitCode = 1; +} + +main().catch((e) => { + log.error(`stamp-charted-at failed: ${e.message}`); + process.exit(1); +}); From 4269d07ba5bf955d300c4878ad1dfd3b257039fc Mon Sep 17 00:00:00 2001 From: Aswin C Date: Fri, 23 Oct 2026 02:50:52 +0530 Subject: [PATCH 2/2] fix(sync): say when a charted_at dry run counts without the index --- scripts/sync/stamp-charted-at.js | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/scripts/sync/stamp-charted-at.js b/scripts/sync/stamp-charted-at.js index f206d21e5..4181d99d7 100644 --- a/scripts/sync/stamp-charted-at.js +++ b/scripts/sync/stamp-charted-at.js @@ -43,6 +43,15 @@ async function main() { if (!dry) { for (const c of chartedAt.COLLECTIONS) await qdrant.ensurePayloadIndex(c.name, chartedAt.FIELD, 'integer'); + } else { + // A dry run creates nothing: without the index, counts filter unindexed, which a + // collection in strict mode refuses (each refused call is counted as failed). + for (const c of chartedAt.COLLECTIONS) { + const schema = await qdrant.payloadSchema(c.name); + if (schema && !schema[chartedAt.FIELD]) { + log.warn(`[CHARTED_AT] ${c.name} has no ${chartedAt.FIELD} index yet (the stamp run creates it); counting without it`); + } + } } const started = Date.now(); const r = await chartedAt.stamp(qdrant, batches, {