openvibers/openvibe.sources Commits
Sources: a retention rule per category, pruned daily (#204)
Changes
@@@ -247,6 +247,18 @@ consumer following `GET /api/v1/items?include_removed=1` learns that the job isthe endpoint's last good parse saw. At most 500 items are retired a run; the rest go on the next. Contract (released in openvibe-contracts v0.12.0):[`sources.item@1`](docs/contracts-proposal/).++**Retention.** Items do not grow forever: `server/retention.js` declares, in one place, a window per category (news 30++days, blog 90, reviews 180, deals 14, coupons 90, trade 365, hosting 30, crowdfunding 90, games 14, food 365, jobs 14)++and one `keepLatest` rule for housing, where HUD's yearly Fair Market Rents must keep the latest value per source key.++An item's age is `retrieved_at` while it is live and `removed_at` once removed, so a consumer following the change feed++still sees a removal for the full window; `first_seen_at` is never the age, since a source that keeps listing a job or++a price keeps it current. The daily job (`server/jobs/retention.js`, started with the worker) deletes at most 2,000 rows++a statement, 50 batches a category a run, and its log line carries counts only. The rules are in code because Sources++has no `openvibe-shared/config` namespace. At runtime a category the contract has but the rules do not is kept (never++pruned) and warned about once a run, and a rule for a category the contract lacks is ignored with a warning, so a++Contracts pin bump cannot stop Sources at boot; `test/retention.test.js` keeps the strict one-to-one check so the bump++PR fails CI and names the missing rule.++## API (service tokens, one capability per route)Callers use an OpenVibe.Network client-credentials token for audience `openvibe.sources`. The merchant routes alsonnew file mode 100644@@@ -0,0 +1,7 @@++-- phase: expand++-- Item retention (plan 2026-10-22, track F): the daily prune (server/retention.js, server/jobs/) deletes++-- items older than their category's window. Two partial indexes serve the two age expressions: a live++-- item ages from retrieved_at, a removed one from removed_at (its removal is the change consumers follow).++-- The keep-latest categories use the existing unique (source_key, identity) index.++CREATE INDEX IF NOT EXISTS idx_items_retention_live ON items (category, retrieved_at) WHERE removed_at IS NULL;++CREATE INDEX IF NOT EXISTS idx_items_retention_removed ON items (category, removed_at) WHERE removed_at IS NOT NULL;@@@ -17,6 +17,7 @@ const { createSpacer } = require('./spacer');const { createRobots } = require('./robots');const { createIngest } = require('./ingest');const { createScheduler } = require('./scheduler');++const { createJobs } = require('./jobs');const { createServiceOutbox } = require('openvibe-sdk/events');const { validate } = require('openvibe-contracts');const { createEnvelope } = require('./events/envelope');@@@ -56,6 +57,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch});const ingest = createIngest({ db, config, registry, items, robots, fetcher, spacer, emitEvent, now, log, outbox });const scheduler = createScheduler({ db, ingest, config, now, log });++ const jobs = createJobs({ db, now, log });const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });const auth = createAuth({ config, log, jwks: keys });const merchant = createMerchant({ db, config, registry, fetcher, resolver: merchantResolver, now, log });@@@ -68,7 +70,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchkeys.start();const keyLoaded = keys.keys().catch(() => null);outbox.start();-- if (config.worker.enabled) scheduler.start();++ if (config.worker.enabled) { scheduler.start(); jobs.start(); }// Pending merchant proofs are checked again on a slow cadence (the merchant can always press "Check now"). A claim// that never verifies fails after merchant.pendingDays. 0 turns the worker pass off (tests).let merchantTimer = null;@@@ -107,6 +109,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchif (merchantKick) clearTimeout(merchantKick);keys.stop();if (accountSubscription) accountSubscription.stop();++ await jobs.stop();await scheduler.stop();await outbox.stop();if (server) {@@@ -116,7 +119,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchif (!givenDb) await db.close();}-- return { config, db, registry, items, guard, fetcher, robots, ingest, scheduler, outbox, keys, keyLoaded, auth, merchant, merchantAuth, app, server, close };++ return { config, db, registry, items, guard, fetcher, robots, ingest, scheduler, jobs, outbox, keys, keyLoaded, auth, merchant, merchantAuth, app, server, close };}if (require.main === module) {nnew file mode 100644@@@ -0,0 +1,42 @@++'use strict';++/**++ * The repo's scheduled jobs, in one place: each is a small object with runOnce/start/stop, and this++ * helper arms them (the scheduler in server/scheduler.js is the ingestion worker, not a job). Today++ * there is one: item retention (server/jobs/retention.js). server/index.js starts it with the worker,++ * so tests and a worker-less process never run a prune behind a test's back.++ */++const { createRetentionJob } = require('./retention');++++function createJobs({ db, now = () => Date.now(), log = console, retention: retentionOptions = {} } = {}) {++ const retention = createRetentionJob({ db, now, log, ...retentionOptions });++ const all = { retention };++ const running = new Map();++ let started = false;++++ /** Run one job now; a second call while it runs shares it. → the job's counts. */++ function runOnce(name = 'retention') {++ const job = all[name];++ if (!job) return Promise.reject(new Error(`no job named ${name}`));++ if (running.has(name)) return running.get(name);++ const p = Promise.resolve().then(() => job.runOnce()).finally(() => running.delete(name));++ running.set(name, p);++ return p;++ }++++ function start() {++ if (started) return false;++ started = true;++ for (const job of Object.values(all)) job.start();++ return true;++ }++++ async function stop() {++ started = false;++ for (const job of Object.values(all)) await job.stop();++ await Promise.allSettled([...running.values()]);++ }++++ return { retention, all, runOnce, start, stop, started: () => started, running: () => running.size };++}++++module.exports = { createJobs };nnew file mode 100644@@@ -0,0 +1,45 @@++'use strict';++/**++ * The daily item-retention job (server/retention.js for the rules and the prune itself).++ *++ * The first run is five minutes after boot, not a day out, so a service redeployed more often than++ * once a day still prunes; later runs are every 24 hours. Timers are unref'd, runs never overlap, and++ * a failed run is logged and retried at the next tick. The job logs counts only.++ */++const { createRetention } = require('../retention');++++const FIRST_MS = 5 * 60_000;++const EVERY_MS = 24 * 3600 * 1000;++++function createRetentionJob({ db, now = () => Date.now(), log = console, firstMs = FIRST_MS, everyMs = EVERY_MS } = {}) {++ const retention = createRetention({ db, now, log });++ let first = null;++ let every = null;++ let running = null;++++ function runOnce() {++ if (!running) running = retention.run().finally(() => { running = null; });++ return running;++ }++++ function start() {++ if (first || every) return false;++ first = setTimeout(() => { runOnce().catch((err) => log.error(`[retention] ${err.stack || err}`)); }, firstMs);++ every = setInterval(() => { runOnce().catch((err) => log.error(`[retention] ${err.stack || err}`)); }, everyMs);++ first.unref?.();++ every.unref?.();++ return true;++ }++++ async function stop() {++ if (first) clearTimeout(first);++ if (every) clearInterval(every);++ first = null;++ every = null;++ if (running) await running.catch(() => {});++ }++++ return { name: 'retention', runOnce, run: runOnce, start, stop, running: () => Boolean(running), retention };++}++++module.exports = { createRetentionJob, FIRST_MS, EVERY_MS };nnew file mode 100644@@@ -0,0 +1,154 @@++'use strict';++/**++ * Item retention: the one place that says how long Sources keeps an item of each category, and the++ * batched prune the daily job runs (server/jobs/retention.js).++ *++ * The age of an item is the last time it was current: `retrieved_at` while it is live, `removed_at`++ * once it was removed (a removal is a change consumers following the change feed may still need;++ * keeping it for the category's window after the removal, not after the last fetch, is what lets a++ * consumer that was down catch up). `first_seen_at` is not the age: a source that keeps listing a++ * job or a price keeps that item current, however old the record is.++ *++ * Two rule shapes:++ * - `days`: prune items whose age is older than that many days.++ * - `keepLatest`: keep the newest item per (source_key, identity), the source's own key for the++ * record, and prune an older row for that key. Yearly values (HUD's Fair Market Rents, one++ * current value per area, fetched weekly and replaced each October) must never age out: the++ * latest value is the one Rent shows, so there is no day window for the category. The items++ * table already keeps one row per (source_key, identity); the rule is what keeps a future++ * identity change from leaving two rows for the same source key, and it documents that the++ * category is not age-pruned.++ *++ * Why each window (each consumer was read before choosing it; all of them copy what they show):++ * - news 30 days: News shows the last 30 days and follows the change feed into its own table.++ * - blog 90 days: the weekly desks read a week of items; a quarter covers a failed pull.++ * - reviews 180 days: reviews and official records are kept by Reviews; half a year covers an++ * import rebuild without holding the raw stream forever.++ * - deals 14 days: offers are short-lived (DealNews' daily feed, shops' hourly crawls) and Deals++ * keeps the offers it shows.++ * - coupons 90 days: a code's stated expiry can be months out; Coupons needs its evidence until++ * then.++ * - trade 365 days: filings and market observations are annual records Trade copies and cites.++ * - hosting 30 days: prices are current; Host copies the catalog and polls at least weekly, so a++ * month survives two missed polls.++ * - crowdfunding 90 days: campaigns run for weeks or months and Deals copies the ones it lists.++ * - games 14 days: top sellers, new releases and specials are current lists, polled every three++ * hours; Games copies them.++ * - housing keepLatest: HUD Fair Market Rents and Small Area FMRs are yearly; one current value++ * per area must stay. BLS' monthly rent series lives on its own identities, so a duplicate-key++ * prune never touches a month.++ * - food 365 days: Food keeps the FDA recall history it shows (and BLS prices); Sources keeps a++ * year for a rebuild and the Open Food Facts bulk snapshot, which is regenerated.++ * - jobs 14 days: Arbeitnow and We Work Remotely retire what the board stopped showing after 7++ * days; Work copies each listing, so a fortnight keeps every tombstone it may pull.++ *++ * Sources has no openvibe-shared/config namespace, so the rules stay in code (the brief's++ * alternative); a day count this consequential should change in review, not in a database row.++ *++ * A contract pin can add a category before this table learns its window. That must not crash the++ * service at boot: at runtime an unruled category is kept (never pruned), and a rule for a category++ * the contract no longer has is ignored, each with one warning line per run. The strict one-to-one++ * check lives in test/retention.test.js instead, so the pin bump PR fails CI and names the rule to++ * add while production keeps serving in the meantime.++ */++const { CATEGORIES } = require('./registry');++++const DAY = 86_400_000;++/** Rows per DELETE; a few thousand is fast on PostgreSQL and stays well under the parameter limit. */++const BATCH_SIZE = 2_000;++/** Batches per category per run: 100,000 rows. A larger backlog continues on the next daily run. */++const MAX_BATCHES = 50;++++const RULES = Object.freeze({++ news: { days: 30, reason: 'News shows the last 30 days and copies every item it follows' },++ blog: { days: 90, reason: 'weekly desks read a week; a quarter covers a failed pull' },++ reviews: { days: 180, reason: 'Reviews keeps what it shows; half a year covers an import rebuild' },++ deals: { days: 14, reason: 'offers are short-lived and Deals copies them' },++ coupons: { days: 90, reason: 'a code\'s stated expiry can be months out and Coupons needs its evidence' },++ trade: { days: 365, reason: 'SEC filings and observations are annual records Trade copies' },++ hosting: { days: 30, reason: 'prices are current, Host copies them, two weekly polls fit in a month' },++ crowdfunding: { days: 90, reason: 'campaigns run for months and Deals copies the ones it lists' },++ games: { days: 14, reason: 'featured lists are current and polled every three hours; Games copies them' },++ housing: { keepLatest: true, reason: 'HUD rents change yearly: the latest value per area is what Rent shows' },++ food: { days: 365, reason: 'Food keeps the recall history it shows; a year covers a rebuild' },++ jobs: { days: 14, reason: 'boards retire unseen jobs after 7 days and Work pulls hourly' },++});++++// The rules are the contract's categories in review; the module deliberately does not throw when++// they disagree (a pin bump must not stop Sources at boot; test/retention.test.js is strict).++++/** The age expression shared by the prune statements: last current, not first seen. */++const AGE_SQL = 'CASE WHEN removed_at IS NULL THEN retrieved_at ELSE removed_at END';++/** The same age as a predicate the two partial indexes (migration 0010) can serve: live rows by++ * retrieved_at, removed rows by removed_at. */++const OLDER_THAN_SQL = '((removed_at IS NULL AND retrieved_at < $2) OR (removed_at IS NOT NULL AND removed_at < $2))';++++function createRetention({ db, rules = RULES, now = () => Date.now(), log = console, batchSize = BATCH_SIZE, maxBatches = MAX_BATCHES } = {}) {++ if (!db || typeof db.prepare !== 'function' || typeof db.tx !== 'function') throw new TypeError('retention: an openvibe-sdk/db handle is required');++ if (!Number.isInteger(batchSize) || batchSize < 1 || batchSize > 10_000) throw new TypeError('retention: batchSize must be an integer between 1 and 10000');++ if (!Number.isInteger(maxBatches) || maxBatches < 1) throw new TypeError('retention: maxBatches must be a positive integer');++++ // One statement/batch: delete the rows, then the revisions that belonged to them (no foreign++ // key keeps them, and a revision of a pruned item would otherwise grow forever).++ async function pruneBatch(selectSql, params) {++ return await db.tx(async (t) => {++ const rows = await t.query(`DELETE FROM items WHERE rid IN (${selectSql}) RETURNING id`, params);++ const ids = rows.rows.map((r) => r.id);++ let revisions = 0;++ if (ids.length) revisions = (await t.query('DELETE FROM item_revisions WHERE item_id = ANY($1)', [ids])).rowCount;++ return { items: ids.length, revisions };++ });++ }++++ async function pruneCategory(category, rule, cutoff, counts) {++ const select = rule.keepLatest++ ? { sql: `SELECT rid FROM (++ SELECT rid, row_number() OVER (PARTITION BY source_key, identity ORDER BY retrieved_at DESC, rid DESC) AS rn++ FROM items WHERE category = $1++ ) ranked WHERE rn > 1 ORDER BY rid LIMIT $2`, params: (limit) => [category, limit] }++ : { sql: `SELECT rid FROM items WHERE category = $1 AND ${OLDER_THAN_SQL} ORDER BY rid LIMIT $3`, params: (limit) => [category, cutoff, limit] };++ for (let batch = 0; batch < maxBatches; batch++) {++ const out = await pruneBatch(select.sql, select.params(batchSize));++ counts.batches++;++ counts.pruned += out.items;++ counts.revisions += out.revisions;++ if (out.items) counts.categories[category] = (counts.categories[category] || 0) + out.items;++ if (out.items < batchSize) return false;++ }++ return true; // a full run of batches: more may be left for the next daily run++ }++++ async function runOnce() {++ const at = now();++ const counts = { pruned: 0, revisions: 0, batches: 0, categories: {}, capped: [] };++ // A pin bump may reach production before this table does: keep an unruled category, ignore a++ // rule the contract no longer has, and say so once per run. The strict check is in the test.++ const known = new Set(CATEGORIES);++ for (const category of CATEGORIES) {++ if (!rules[category]) log.warn(`[retention] no rule for category ${category}: items kept`);++ }++ for (const category of Object.keys(rules)) {++ if (!known.has(category)) log.warn(`[retention] rule for ${category} ignored: not a sources.source@1 category`);++ }++ for (const [category, rule] of Object.entries(rules)) {++ if (!known.has(category)) continue;++ const cutoff = rule.days ? at - rule.days * DAY : null;++ if (await pruneCategory(category, rule, cutoff, counts)) counts.capped.push(category);++ }++ const perCategory = Object.entries(counts.categories).map(([c, n]) => `${c} ${n}`).join(', ');++ log.log(`[retention] items: ${counts.pruned} pruned${perCategory ? ` (${perCategory})` : ''}, ${counts.revisions} revisions, ${counts.batches} batches`++ + `${counts.capped.length ? `, more: ${counts.capped.join(', ')}` : ''}`);++ return counts;++ }++++ let running = null;++ /** One prune at a time; concurrent callers share the run. */++ function run() {++ if (!running) running = runOnce().finally(() => { running = null; });++ return running;++ }++++ return { run, runOnce, rules };++}++++module.exports = { createRetention, RULES, AGE_SQL, OLDER_THAN_SQL, BATCH_SIZE, MAX_BATCHES, DAY };nnew file mode 100644@@@ -0,0 +1,161 @@++'use strict';++/**++ * Item retention (server/retention.js, server/jobs/): the category windows, the age of a removed item,++ * keep-latest per source key, batching, idempotence, and the runtime grace for a category a pin bump++ * added before the rule table did. A real migrated database (PGlite, or the containers with++ * SOURCES_TEST_STORE=pg), no network.++ *++ * node test/retention.test.js++ */++const assert = require('assert');++const { createDb } = require('openvibe-sdk/db');++const { testDb } = require('./db');++const { createRetention, RULES, DAY } = require('../server/retention');++const { createJobs } = require('../server/jobs');++const { CATEGORIES } = require('../server/registry');++++const NOW = Date.parse('2026-11-05T12:00:00Z');++const silent = { log() {}, warn() {}, error() {} };++let db;++let seq = 0;++++async function item({ id, category = 'news', source = 'src-a', identity = id, retrieved, removed = null, title = id }) {++ seq++;++ await db.query(`INSERT INTO items (id, source_key, category, identity, kind, title, content_hash, parser_version,++ first_seen_at, retrieved_at, removed_at, removed_reason, change_seq)++ VALUES ($1, $2, $3, $4, 'record', $5, 'hash', 'test@1', $6, $6, $7, $8, $9)`,++ [id, source, category, identity, title, retrieved, removed, removed ? 'test removal' : null, seq]);++ await db.query(`INSERT INTO item_revisions (item_id, revision, content_hash, parser_version, retrieved_at, snapshot)++ VALUES ($1, 1, 'hash', 'test@1', $2, '{}')`, [id, retrieved]);++}++++const rows = async (where = '') => await db.query(`SELECT id, title, identity FROM items ${where} ORDER BY id`);++const ids = async (where = '') => (await rows(where)).rows.map((r) => r.id);++const revisions = async (id) => Number(await db.value('SELECT count(*) FROM item_revisions WHERE item_id = $1', [id]));++++(async () => {++ const testdb = await testDb();++ db = testdb.db;++ // With SOURCES_TEST_STORE=pg the handle is the DML-only runtime role; DDL (the keep-latest test++ // drops a unique constraint) has to go through the owner connection, as migrations do.++ const owner = testdb.directUrl ? createDb({ url: testdb.directUrl, service: 'retention-test-owner', max: 1, log: silent }) : null;++ try {++ // STRICT BY DESIGN, TEST-ONLY: a Contracts pin that adds a category must fail here, on the++ // bump PR, naming the missing rule. server/retention.js itself keeps + warns at runtime.++ assert.deepStrictEqual([...CATEGORIES].sort(), Object.keys(RULES).sort(), 'one rule per source.v1 category');++ for (const [category, rule] of Object.entries(RULES)) {++ assert.strictEqual(typeof rule.reason, 'string');++ assert.ok(rule.reason.length > 10, `${category} says why`);++ assert.ok((Number.isInteger(rule.days) && rule.days >= 1) || rule.keepLatest === true, `${category} has a day window or keep-latest`);++ assert.ok(!(rule.days && rule.keepLatest), `${category} is one shape`);++ }++ assert.strictEqual(RULES.deals.days, 14);++ assert.strictEqual(RULES.news.days, 30);++ assert.strictEqual(RULES.housing.keepLatest, true);++++ const logs = [];++ const retention = createRetention({ db, now: () => NOW, log: { log: (m) => logs.push(m), warn() {}, error() {} } });++++ // A day window: older than the category's rule goes, newer stays; a removal is aged from++ // removed_at, not from retrieved_at, so a consumer gets the full window after the removal.++ await item({ id: 'news-old-live', retrieved: NOW - 31 * DAY });++ await item({ id: 'news-new-live', retrieved: NOW - 29 * DAY });++ await item({ id: 'news-old-removed-recently', retrieved: NOW - 100 * DAY, removed: NOW - 2 * DAY });++ await item({ id: 'news-removed-long-ago', retrieved: NOW - 3 * DAY, removed: NOW - 31 * DAY });++ await item({ id: 'deals-old', category: 'deals', retrieved: NOW - 15 * DAY });++ await item({ id: 'deals-new', category: 'deals', retrieved: NOW - 13 * DAY });++++ const first = await retention.runOnce();++ assert.deepStrictEqual(await ids(), ['deals-new', 'news-new-live', 'news-old-removed-recently']);++ assert.deepStrictEqual(first.categories, { news: 2, deals: 1 });++ assert.strictEqual(first.pruned, 3);++ assert.strictEqual(first.revisions, 3);++ // Counts only: no item id, title or URL reaches the log.++ assert.match(logs[0], /^\[retention\] items: 3 pruned \(news 2, deals 1\), 3 revisions, \d+ batches$/);++ assert.ok(!logs[0].includes('news-old-live') && !logs[0].includes('deals-old'));++ assert.strictEqual(await revisions('news-old-live'), 0, 'a pruned item takes its revisions');++ assert.strictEqual(await revisions('news-new-live'), 1, 'a kept item keeps its revisions');++ assert.strictEqual(await revisions('news-old-removed-recently'), 1);++++ // Idempotent: nothing left to prune, and the second run still checks every category.++ const second = await retention.runOnce();++ assert.strictEqual(second.pruned, 0);++ assert.strictEqual(second.revisions, 0);++ assert.deepStrictEqual(second.categories, {});++ assert.deepStrictEqual(await ids(), ['deals-new', 'news-new-live', 'news-old-removed-recently']);++++ // Batched: small statements delete the backlog over several batches.++ for (let i = 0; i < 5; i++) await item({ id: `games-old-${i}`, category: 'games', retrieved: NOW - 15 * DAY });++ const batched = createRetention({ db, now: () => NOW, log: silent, batchSize: 2, maxBatches: 10 });++ const out = await batched.runOnce();++ assert.deepStrictEqual(await ids("WHERE id LIKE 'games-old-%'"), []);++ assert.deepStrictEqual(out.categories, { games: 5 });++++ // A cap leaves the rest for the next daily run; the next runs finish it.++ for (let i = 0; i < 5; i++) await item({ id: `deals-capped-${i}`, category: 'deals', retrieved: NOW - 20 * DAY });++ const capped = createRetention({ db, now: () => NOW, log: silent, batchSize: 2, maxBatches: 1 });++ const c1 = await capped.runOnce();++ assert.deepStrictEqual(c1.categories, { deals: 2 });++ assert.deepStrictEqual(c1.capped, ['deals']);++ const c2 = await capped.runOnce();++ assert.deepStrictEqual(c2.categories, { deals: 2 });++ const c3 = await capped.runOnce();++ assert.deepStrictEqual(c3.categories, { deals: 1 });++ assert.deepStrictEqual(c3.capped, []);++ assert.deepStrictEqual(await ids("WHERE id LIKE 'deals-capped-%'"), []);++++ // Runtime grace: a category the contract has but the rules do not is kept (never pruned),++ // and an extra rule is ignored, each with one warning line per run. The strict check above is++ // what makes the pin bump fail in CI; this is what keeps production serving meanwhile.++ await item({ id: 'news-unruled', retrieved: NOW - 100 * DAY });++ await item({ id: 'deals-injected', category: 'deals', retrieved: NOW - 20 * DAY });++ const injectedRules = { ...RULES, weather: { days: 1, reason: 'test-only category the contract lacks' } };++ delete injectedRules.news;++ const injectedLogs = [];++ const injected = createRetention({++ db, rules: injectedRules, now: () => NOW,++ log: { log: (m) => injectedLogs.push(m), warn: (m) => injectedLogs.push(m), error() {} },++ });++ const injectedOut = await injected.runOnce();++ assert.deepStrictEqual(injectedOut.categories, { deals: 1 });++ assert.deepStrictEqual(await ids("WHERE id IN ('news-unruled', 'deals-injected')"), ['news-unruled'], 'unruled news kept');++ assert.strictEqual(injectedLogs.filter((m) => m === '[retention] no rule for category news: items kept').length, 1);++ assert.strictEqual(injectedLogs.filter((m) => m === '[retention] rule for weather ignored: not a sources.source@1 category').length, 1);++ await injected.runOnce();++ assert.strictEqual(injectedLogs.filter((m) => m === '[retention] no rule for category news: items kept').length, 2, 'one warning per run');++ assert.deepStrictEqual(await ids("WHERE id = 'news-unruled'"), ['news-unruled']);++ await db.query("DELETE FROM item_revisions WHERE item_id = 'news-unruled'");++ await db.query("DELETE FROM items WHERE id = 'news-unruled'");++++ // Keep-latest: one item per (source_key, identity), the newest. The items table already keeps++ // one row per key, so the test drops its unique constraint to prove what the rule does if a++ // source ever changes its identity format and leaves an older row for the same key.++ const unique = await db.value(`SELECT conname FROM pg_constraint++ WHERE conrelid = 'items'::regclass AND contype = 'u' AND pg_get_constraintdef(oid) LIKE 'UNIQUE (source_key, identity)'`);++ assert.ok(unique, 'the items unique key is there');++ await (owner || db).query(`ALTER TABLE items DROP CONSTRAINT "${unique}"`);++ await item({ id: 'rent-old', category: 'housing', source: 'hud-fair-market-rents', identity: 'METRO1', retrieved: NOW - 2 * DAY, title: 'old value' });++ await item({ id: 'rent-new', category: 'housing', source: 'hud-fair-market-rents', identity: 'METRO1', retrieved: NOW - 1 * DAY, title: 'new value' });++ await item({ id: 'rent-other', category: 'housing', source: 'hud-fair-market-rents', identity: 'METRO2', retrieved: NOW - 400 * DAY, title: 'other area' });++ const keep = await retention.runOnce();++ assert.deepStrictEqual(keep.categories, { housing: 1 });++ assert.deepStrictEqual(await ids("WHERE id LIKE 'rent-%'"), ['rent-new', 'rent-other']);++ assert.strictEqual((await rows("WHERE id = 'rent-new'")).rows[0].title, 'new value');++++ // The repo's job helper arms the job, runs one on demand and stops it. (The real first run is++ // five minutes out, so a test boot never prunes behind a test's back.)++ const jobs = createJobs({ db, now: () => NOW, log: silent, retention: { firstMs: 3_600_000 } });++ assert.strictEqual(jobs.start(), true);++ assert.strictEqual(jobs.start(), false, 'starting twice is a no-op');++ const viaHelper = await jobs.runOnce('retention');++ assert.strictEqual(viaHelper.pruned, 0, 'the helper reaches the same prune');++ assert.strictEqual(jobs.running(), 0);++ await jobs.stop();++ await assert.rejects(jobs.runOnce('nope'), /no job named nope/);++++ console.log('retention: all checks passed');++ } finally {++ if (owner) await owner.close().catch(() => {});++ await testdb.close();++ }++})().catch((err) => { console.error(err); process.exit(1); });
@@ -247,6 +247,18 @@ consumer following `GET /api/v1/items?include_removed=1` learns that the job is
@@ -247,6 +247,18 @@ consumer following `GET /api/v1/items?include_removed=1` learns that the job is
the endpoint's last good parse saw. At most 500 items are retired a run; the rest go on the next. Contract (released in openvibe-contracts v0.12.0):
the endpoint's last good parse saw. At most 500 items are retired a run; the rest go on the next. Contract (released in openvibe-contracts v0.12.0):
[`sources.item@1`](docs/contracts-proposal/).
[`sources.item@1`](docs/contracts-proposal/).
+**Retention.** Items do not grow forever: `server/retention.js` declares, in one place, a window per category (news 30
+days, blog 90, reviews 180, deals 14, coupons 90, trade 365, hosting 30, crowdfunding 90, games 14, food 365, jobs 14)
+and one `keepLatest` rule for housing, where HUD's yearly Fair Market Rents must keep the latest value per source key.
+An item's age is `retrieved_at` while it is live and `removed_at` once removed, so a consumer following the change feed
+still sees a removal for the full window; `first_seen_at` is never the age, since a source that keeps listing a job or
+a price keeps it current. The daily job (`server/jobs/retention.js`, started with the worker) deletes at most 2,000 rows
+a statement, 50 batches a category a run, and its log line carries counts only. The rules are in code because Sources
+has no `openvibe-shared/config` namespace. At runtime a category the contract has but the rules do not is kept (never
+pruned) and warned about once a run, and a rule for a category the contract lacks is ignored with a warning, so a
+Contracts pin bump cannot stop Sources at boot; `test/retention.test.js` keeps the strict one-to-one check so the bump
+PR fails CI and names the missing rule.
+
## API (service tokens, one capability per route)
## API (service tokens, one capability per route)
Callers use an OpenVibe.Network client-credentials token for audience `openvibe.sources`. The merchant routes also
Callers use an OpenVibe.Network client-credentials token for audience `openvibe.sources`. The merchant routes also
new file mode 100644
new file mode 100644
@@ -0,0 +1,7 @@
@@ -0,0 +1,7 @@
+-- phase: expand
+-- Item retention (plan 2026-10-22, track F): the daily prune (server/retention.js, server/jobs/) deletes
+-- items older than their category's window. Two partial indexes serve the two age expressions: a live
+-- item ages from retrieved_at, a removed one from removed_at (its removal is the change consumers follow).
+-- The keep-latest categories use the existing unique (source_key, identity) index.
+CREATE INDEX IF NOT EXISTS idx_items_retention_live ON items (category, retrieved_at) WHERE removed_at IS NULL;
+CREATE INDEX IF NOT EXISTS idx_items_retention_removed ON items (category, removed_at) WHERE removed_at IS NOT NULL;
@@ -17,6 +17,7 @@ const { createSpacer } = require('./spacer');
@@ -17,6 +17,7 @@ const { createSpacer } = require('./spacer');
const { createRobots } = require('./robots');
const { createRobots } = require('./robots');
const { createIngest } = require('./ingest');
const { createIngest } = require('./ingest');
const { createScheduler } = require('./scheduler');
const { createScheduler } = require('./scheduler');
+const { createJobs } = require('./jobs');
const { createServiceOutbox } = require('openvibe-sdk/events');
const { createServiceOutbox } = require('openvibe-sdk/events');
const { validate } = require('openvibe-contracts');
const { validate } = require('openvibe-contracts');
const { createEnvelope } = require('./events/envelope');
const { createEnvelope } = require('./events/envelope');
@@ -56,6 +57,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -56,6 +57,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
});
});
const ingest = createIngest({ db, config, registry, items, robots, fetcher, spacer, emitEvent, now, log, outbox });
const ingest = createIngest({ db, config, registry, items, robots, fetcher, spacer, emitEvent, now, log, outbox });
const scheduler = createScheduler({ db, ingest, config, now, log });
const scheduler = createScheduler({ db, ingest, config, now, log });
+ const jobs = createJobs({ db, now, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const auth = createAuth({ config, log, jwks: keys });
const auth = createAuth({ config, log, jwks: keys });
const merchant = createMerchant({ db, config, registry, fetcher, resolver: merchantResolver, now, log });
const merchant = createMerchant({ db, config, registry, fetcher, resolver: merchantResolver, now, log });
@@ -68,7 +70,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -68,7 +70,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
keys.start();
keys.start();
const keyLoaded = keys.keys().catch(() => null);
const keyLoaded = keys.keys().catch(() => null);
outbox.start();
outbox.start();
- if (config.worker.enabled) scheduler.start();
+ if (config.worker.enabled) { scheduler.start(); jobs.start(); }
// Pending merchant proofs are checked again on a slow cadence (the merchant can always press "Check now"). A claim
// Pending merchant proofs are checked again on a slow cadence (the merchant can always press "Check now"). A claim
// that never verifies fails after merchant.pendingDays. 0 turns the worker pass off (tests).
// that never verifies fails after merchant.pendingDays. 0 turns the worker pass off (tests).
let merchantTimer = null;
let merchantTimer = null;
@@ -107,6 +109,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -107,6 +109,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
if (merchantKick) clearTimeout(merchantKick);
if (merchantKick) clearTimeout(merchantKick);
keys.stop();
keys.stop();
if (accountSubscription) accountSubscription.stop();
if (accountSubscription) accountSubscription.stop();
+ await jobs.stop();
await scheduler.stop();
await scheduler.stop();
await outbox.stop();
await outbox.stop();
if (server) {
if (server) {
@@ -116,7 +119,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -116,7 +119,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
if (!givenDb) await db.close();
if (!givenDb) await db.close();
}
}
- return { config, db, registry, items, guard, fetcher, robots, ingest, scheduler, outbox, keys, keyLoaded, auth, merchant, merchantAuth, app, server, close };
+ return { config, db, registry, items, guard, fetcher, robots, ingest, scheduler, jobs, outbox, keys, keyLoaded, auth, merchant, merchantAuth, app, server, close };
}
}
if (require.main === module) {
if (require.main === module) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,42 @@
@@ -0,0 +1,42 @@
+'use strict';
+/**
+ * The repo's scheduled jobs, in one place: each is a small object with runOnce/start/stop, and this
+ * helper arms them (the scheduler in server/scheduler.js is the ingestion worker, not a job). Today
+ * there is one: item retention (server/jobs/retention.js). server/index.js starts it with the worker,
+ * so tests and a worker-less process never run a prune behind a test's back.
+ */
+const { createRetentionJob } = require('./retention');
+
+function createJobs({ db, now = () => Date.now(), log = console, retention: retentionOptions = {} } = {}) {
+ const retention = createRetentionJob({ db, now, log, ...retentionOptions });
+ const all = { retention };
+ const running = new Map();
+ let started = false;
+
+ /** Run one job now; a second call while it runs shares it. → the job's counts. */
+ function runOnce(name = 'retention') {
+ const job = all[name];
+ if (!job) return Promise.reject(new Error(`no job named ${name}`));
+ if (running.has(name)) return running.get(name);
+ const p = Promise.resolve().then(() => job.runOnce()).finally(() => running.delete(name));
+ running.set(name, p);
+ return p;
+ }
+
+ function start() {
+ if (started) return false;
+ started = true;
+ for (const job of Object.values(all)) job.start();
+ return true;
+ }
+
+ async function stop() {
+ started = false;
+ for (const job of Object.values(all)) await job.stop();
+ await Promise.allSettled([...running.values()]);
+ }
+
+ return { retention, all, runOnce, start, stop, started: () => started, running: () => running.size };
+}
+
+module.exports = { createJobs };
new file mode 100644
new file mode 100644
@@ -0,0 +1,45 @@
@@ -0,0 +1,45 @@
+'use strict';
+/**
+ * The daily item-retention job (server/retention.js for the rules and the prune itself).
+ *
+ * The first run is five minutes after boot, not a day out, so a service redeployed more often than
+ * once a day still prunes; later runs are every 24 hours. Timers are unref'd, runs never overlap, and
+ * a failed run is logged and retried at the next tick. The job logs counts only.
+ */
+const { createRetention } = require('../retention');
+
+const FIRST_MS = 5 * 60_000;
+const EVERY_MS = 24 * 3600 * 1000;
+
+function createRetentionJob({ db, now = () => Date.now(), log = console, firstMs = FIRST_MS, everyMs = EVERY_MS } = {}) {
+ const retention = createRetention({ db, now, log });
+ let first = null;
+ let every = null;
+ let running = null;
+
+ function runOnce() {
+ if (!running) running = retention.run().finally(() => { running = null; });
+ return running;
+ }
+
+ function start() {
+ if (first || every) return false;
+ first = setTimeout(() => { runOnce().catch((err) => log.error(`[retention] ${err.stack || err}`)); }, firstMs);
+ every = setInterval(() => { runOnce().catch((err) => log.error(`[retention] ${err.stack || err}`)); }, everyMs);
+ first.unref?.();
+ every.unref?.();
+ return true;
+ }
+
+ async function stop() {
+ if (first) clearTimeout(first);
+ if (every) clearInterval(every);
+ first = null;
+ every = null;
+ if (running) await running.catch(() => {});
+ }
+
+ return { name: 'retention', runOnce, run: runOnce, start, stop, running: () => Boolean(running), retention };
+}
+
+module.exports = { createRetentionJob, FIRST_MS, EVERY_MS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,154 @@
@@ -0,0 +1,154 @@
+'use strict';
+/**
+ * Item retention: the one place that says how long Sources keeps an item of each category, and the
+ * batched prune the daily job runs (server/jobs/retention.js).
+ *
+ * The age of an item is the last time it was current: `retrieved_at` while it is live, `removed_at`
+ * once it was removed (a removal is a change consumers following the change feed may still need;
+ * keeping it for the category's window after the removal, not after the last fetch, is what lets a
+ * consumer that was down catch up). `first_seen_at` is not the age: a source that keeps listing a
+ * job or a price keeps that item current, however old the record is.
+ *
+ * Two rule shapes:
+ * - `days`: prune items whose age is older than that many days.
+ * - `keepLatest`: keep the newest item per (source_key, identity), the source's own key for the
+ * record, and prune an older row for that key. Yearly values (HUD's Fair Market Rents, one
+ * current value per area, fetched weekly and replaced each October) must never age out: the
+ * latest value is the one Rent shows, so there is no day window for the category. The items
+ * table already keeps one row per (source_key, identity); the rule is what keeps a future
+ * identity change from leaving two rows for the same source key, and it documents that the
+ * category is not age-pruned.
+ *
+ * Why each window (each consumer was read before choosing it; all of them copy what they show):
+ * - news 30 days: News shows the last 30 days and follows the change feed into its own table.
+ * - blog 90 days: the weekly desks read a week of items; a quarter covers a failed pull.
+ * - reviews 180 days: reviews and official records are kept by Reviews; half a year covers an
+ * import rebuild without holding the raw stream forever.
+ * - deals 14 days: offers are short-lived (DealNews' daily feed, shops' hourly crawls) and Deals
+ * keeps the offers it shows.
+ * - coupons 90 days: a code's stated expiry can be months out; Coupons needs its evidence until
+ * then.
+ * - trade 365 days: filings and market observations are annual records Trade copies and cites.
+ * - hosting 30 days: prices are current; Host copies the catalog and polls at least weekly, so a
+ * month survives two missed polls.
+ * - crowdfunding 90 days: campaigns run for weeks or months and Deals copies the ones it lists.
+ * - games 14 days: top sellers, new releases and specials are current lists, polled every three
+ * hours; Games copies them.
+ * - housing keepLatest: HUD Fair Market Rents and Small Area FMRs are yearly; one current value
+ * per area must stay. BLS' monthly rent series lives on its own identities, so a duplicate-key
+ * prune never touches a month.
+ * - food 365 days: Food keeps the FDA recall history it shows (and BLS prices); Sources keeps a
+ * year for a rebuild and the Open Food Facts bulk snapshot, which is regenerated.
+ * - jobs 14 days: Arbeitnow and We Work Remotely retire what the board stopped showing after 7
+ * days; Work copies each listing, so a fortnight keeps every tombstone it may pull.
+ *
+ * Sources has no openvibe-shared/config namespace, so the rules stay in code (the brief's
+ * alternative); a day count this consequential should change in review, not in a database row.
+ *
+ * A contract pin can add a category before this table learns its window. That must not crash the
+ * service at boot: at runtime an unruled category is kept (never pruned), and a rule for a category
+ * the contract no longer has is ignored, each with one warning line per run. The strict one-to-one
+ * check lives in test/retention.test.js instead, so the pin bump PR fails CI and names the rule to
+ * add while production keeps serving in the meantime.
+ */
+const { CATEGORIES } = require('./registry');
+
+const DAY = 86_400_000;
+/** Rows per DELETE; a few thousand is fast on PostgreSQL and stays well under the parameter limit. */
+const BATCH_SIZE = 2_000;
+/** Batches per category per run: 100,000 rows. A larger backlog continues on the next daily run. */
+const MAX_BATCHES = 50;
+
+const RULES = Object.freeze({
+ news: { days: 30, reason: 'News shows the last 30 days and copies every item it follows' },
+ blog: { days: 90, reason: 'weekly desks read a week; a quarter covers a failed pull' },
+ reviews: { days: 180, reason: 'Reviews keeps what it shows; half a year covers an import rebuild' },
+ deals: { days: 14, reason: 'offers are short-lived and Deals copies them' },
+ coupons: { days: 90, reason: 'a code\'s stated expiry can be months out and Coupons needs its evidence' },
+ trade: { days: 365, reason: 'SEC filings and observations are annual records Trade copies' },
+ hosting: { days: 30, reason: 'prices are current, Host copies them, two weekly polls fit in a month' },
+ crowdfunding: { days: 90, reason: 'campaigns run for months and Deals copies the ones it lists' },
+ games: { days: 14, reason: 'featured lists are current and polled every three hours; Games copies them' },
+ housing: { keepLatest: true, reason: 'HUD rents change yearly: the latest value per area is what Rent shows' },
+ food: { days: 365, reason: 'Food keeps the recall history it shows; a year covers a rebuild' },
+ jobs: { days: 14, reason: 'boards retire unseen jobs after 7 days and Work pulls hourly' },
+});
+
+// The rules are the contract's categories in review; the module deliberately does not throw when
+// they disagree (a pin bump must not stop Sources at boot; test/retention.test.js is strict).
+
+/** The age expression shared by the prune statements: last current, not first seen. */
+const AGE_SQL = 'CASE WHEN removed_at IS NULL THEN retrieved_at ELSE removed_at END';
+/** The same age as a predicate the two partial indexes (migration 0010) can serve: live rows by
+ * retrieved_at, removed rows by removed_at. */
+const OLDER_THAN_SQL = '((removed_at IS NULL AND retrieved_at < $2) OR (removed_at IS NOT NULL AND removed_at < $2))';
+
+function createRetention({ db, rules = RULES, now = () => Date.now(), log = console, batchSize = BATCH_SIZE, maxBatches = MAX_BATCHES } = {}) {
+ if (!db || typeof db.prepare !== 'function' || typeof db.tx !== 'function') throw new TypeError('retention: an openvibe-sdk/db handle is required');
+ if (!Number.isInteger(batchSize) || batchSize < 1 || batchSize > 10_000) throw new TypeError('retention: batchSize must be an integer between 1 and 10000');
+ if (!Number.isInteger(maxBatches) || maxBatches < 1) throw new TypeError('retention: maxBatches must be a positive integer');
+
+ // One statement/batch: delete the rows, then the revisions that belonged to them (no foreign
+ // key keeps them, and a revision of a pruned item would otherwise grow forever).
+ async function pruneBatch(selectSql, params) {
+ return await db.tx(async (t) => {
+ const rows = await t.query(`DELETE FROM items WHERE rid IN (${selectSql}) RETURNING id`, params);
+ const ids = rows.rows.map((r) => r.id);
+ let revisions = 0;
+ if (ids.length) revisions = (await t.query('DELETE FROM item_revisions WHERE item_id = ANY($1)', [ids])).rowCount;
+ return { items: ids.length, revisions };
+ });
+ }
+
+ async function pruneCategory(category, rule, cutoff, counts) {
+ const select = rule.keepLatest
+ ? { sql: `SELECT rid FROM (
+ SELECT rid, row_number() OVER (PARTITION BY source_key, identity ORDER BY retrieved_at DESC, rid DESC) AS rn
+ FROM items WHERE category = $1
+ ) ranked WHERE rn > 1 ORDER BY rid LIMIT $2`, params: (limit) => [category, limit] }
+ : { sql: `SELECT rid FROM items WHERE category = $1 AND ${OLDER_THAN_SQL} ORDER BY rid LIMIT $3`, params: (limit) => [category, cutoff, limit] };
+ for (let batch = 0; batch < maxBatches; batch++) {
+ const out = await pruneBatch(select.sql, select.params(batchSize));
+ counts.batches++;
+ counts.pruned += out.items;
+ counts.revisions += out.revisions;
+ if (out.items) counts.categories[category] = (counts.categories[category] || 0) + out.items;
+ if (out.items < batchSize) return false;
+ }
+ return true; // a full run of batches: more may be left for the next daily run
+ }
+
+ async function runOnce() {
+ const at = now();
+ const counts = { pruned: 0, revisions: 0, batches: 0, categories: {}, capped: [] };
+ // A pin bump may reach production before this table does: keep an unruled category, ignore a
+ // rule the contract no longer has, and say so once per run. The strict check is in the test.
+ const known = new Set(CATEGORIES);
+ for (const category of CATEGORIES) {
+ if (!rules[category]) log.warn(`[retention] no rule for category ${category}: items kept`);
+ }
+ for (const category of Object.keys(rules)) {
+ if (!known.has(category)) log.warn(`[retention] rule for ${category} ignored: not a sources.source@1 category`);
+ }
+ for (const [category, rule] of Object.entries(rules)) {
+ if (!known.has(category)) continue;
+ const cutoff = rule.days ? at - rule.days * DAY : null;
+ if (await pruneCategory(category, rule, cutoff, counts)) counts.capped.push(category);
+ }
+ const perCategory = Object.entries(counts.categories).map(([c, n]) => `${c} ${n}`).join(', ');
+ log.log(`[retention] items: ${counts.pruned} pruned${perCategory ? ` (${perCategory})` : ''}, ${counts.revisions} revisions, ${counts.batches} batches`
+ + `${counts.capped.length ? `, more: ${counts.capped.join(', ')}` : ''}`);
+ return counts;
+ }
+
+ let running = null;
+ /** One prune at a time; concurrent callers share the run. */
+ function run() {
+ if (!running) running = runOnce().finally(() => { running = null; });
+ return running;
+ }
+
+ return { run, runOnce, rules };
+}
+
+module.exports = { createRetention, RULES, AGE_SQL, OLDER_THAN_SQL, BATCH_SIZE, MAX_BATCHES, DAY };
new file mode 100644
new file mode 100644
@@ -0,0 +1,161 @@
@@ -0,0 +1,161 @@
+'use strict';
+/**
+ * Item retention (server/retention.js, server/jobs/): the category windows, the age of a removed item,
+ * keep-latest per source key, batching, idempotence, and the runtime grace for a category a pin bump
+ * added before the rule table did. A real migrated database (PGlite, or the containers with
+ * SOURCES_TEST_STORE=pg), no network.
+ *
+ * node test/retention.test.js
+ */
+const assert = require('assert');
+const { createDb } = require('openvibe-sdk/db');
+const { testDb } = require('./db');
+const { createRetention, RULES, DAY } = require('../server/retention');
+const { createJobs } = require('../server/jobs');
+const { CATEGORIES } = require('../server/registry');
+
+const NOW = Date.parse('2026-11-05T12:00:00Z');
+const silent = { log() {}, warn() {}, error() {} };
+let db;
+let seq = 0;
+
+async function item({ id, category = 'news', source = 'src-a', identity = id, retrieved, removed = null, title = id }) {
+ seq++;
+ await db.query(`INSERT INTO items (id, source_key, category, identity, kind, title, content_hash, parser_version,
+ first_seen_at, retrieved_at, removed_at, removed_reason, change_seq)
+ VALUES ($1, $2, $3, $4, 'record', $5, 'hash', 'test@1', $6, $6, $7, $8, $9)`,
+ [id, source, category, identity, title, retrieved, removed, removed ? 'test removal' : null, seq]);
+ await db.query(`INSERT INTO item_revisions (item_id, revision, content_hash, parser_version, retrieved_at, snapshot)
+ VALUES ($1, 1, 'hash', 'test@1', $2, '{}')`, [id, retrieved]);
+}
+
+const rows = async (where = '') => await db.query(`SELECT id, title, identity FROM items ${where} ORDER BY id`);
+const ids = async (where = '') => (await rows(where)).rows.map((r) => r.id);
+const revisions = async (id) => Number(await db.value('SELECT count(*) FROM item_revisions WHERE item_id = $1', [id]));
+
+(async () => {
+ const testdb = await testDb();
+ db = testdb.db;
+ // With SOURCES_TEST_STORE=pg the handle is the DML-only runtime role; DDL (the keep-latest test
+ // drops a unique constraint) has to go through the owner connection, as migrations do.
+ const owner = testdb.directUrl ? createDb({ url: testdb.directUrl, service: 'retention-test-owner', max: 1, log: silent }) : null;
+ try {
+ // STRICT BY DESIGN, TEST-ONLY: a Contracts pin that adds a category must fail here, on the
+ // bump PR, naming the missing rule. server/retention.js itself keeps + warns at runtime.
+ assert.deepStrictEqual([...CATEGORIES].sort(), Object.keys(RULES).sort(), 'one rule per source.v1 category');
+ for (const [category, rule] of Object.entries(RULES)) {
+ assert.strictEqual(typeof rule.reason, 'string');
+ assert.ok(rule.reason.length > 10, `${category} says why`);
+ assert.ok((Number.isInteger(rule.days) && rule.days >= 1) || rule.keepLatest === true, `${category} has a day window or keep-latest`);
+ assert.ok(!(rule.days && rule.keepLatest), `${category} is one shape`);
+ }
+ assert.strictEqual(RULES.deals.days, 14);
+ assert.strictEqual(RULES.news.days, 30);
+ assert.strictEqual(RULES.housing.keepLatest, true);
+
+ const logs = [];
+ const retention = createRetention({ db, now: () => NOW, log: { log: (m) => logs.push(m), warn() {}, error() {} } });
+
+ // A day window: older than the category's rule goes, newer stays; a removal is aged from
+ // removed_at, not from retrieved_at, so a consumer gets the full window after the removal.
+ await item({ id: 'news-old-live', retrieved: NOW - 31 * DAY });
+ await item({ id: 'news-new-live', retrieved: NOW - 29 * DAY });
+ await item({ id: 'news-old-removed-recently', retrieved: NOW - 100 * DAY, removed: NOW - 2 * DAY });
+ await item({ id: 'news-removed-long-ago', retrieved: NOW - 3 * DAY, removed: NOW - 31 * DAY });
+ await item({ id: 'deals-old', category: 'deals', retrieved: NOW - 15 * DAY });
+ await item({ id: 'deals-new', category: 'deals', retrieved: NOW - 13 * DAY });
+
+ const first = await retention.runOnce();
+ assert.deepStrictEqual(await ids(), ['deals-new', 'news-new-live', 'news-old-removed-recently']);
+ assert.deepStrictEqual(first.categories, { news: 2, deals: 1 });
+ assert.strictEqual(first.pruned, 3);
+ assert.strictEqual(first.revisions, 3);
+ // Counts only: no item id, title or URL reaches the log.
+ assert.match(logs[0], /^\[retention\] items: 3 pruned \(news 2, deals 1\), 3 revisions, \d+ batches$/);
+ assert.ok(!logs[0].includes('news-old-live') && !logs[0].includes('deals-old'));
+ assert.strictEqual(await revisions('news-old-live'), 0, 'a pruned item takes its revisions');
+ assert.strictEqual(await revisions('news-new-live'), 1, 'a kept item keeps its revisions');
+ assert.strictEqual(await revisions('news-old-removed-recently'), 1);
+
+ // Idempotent: nothing left to prune, and the second run still checks every category.
+ const second = await retention.runOnce();
+ assert.strictEqual(second.pruned, 0);
+ assert.strictEqual(second.revisions, 0);
+ assert.deepStrictEqual(second.categories, {});
+ assert.deepStrictEqual(await ids(), ['deals-new', 'news-new-live', 'news-old-removed-recently']);
+
+ // Batched: small statements delete the backlog over several batches.
+ for (let i = 0; i < 5; i++) await item({ id: `games-old-${i}`, category: 'games', retrieved: NOW - 15 * DAY });
+ const batched = createRetention({ db, now: () => NOW, log: silent, batchSize: 2, maxBatches: 10 });
+ const out = await batched.runOnce();
+ assert.deepStrictEqual(await ids("WHERE id LIKE 'games-old-%'"), []);
+ assert.deepStrictEqual(out.categories, { games: 5 });
+
+ // A cap leaves the rest for the next daily run; the next runs finish it.
+ for (let i = 0; i < 5; i++) await item({ id: `deals-capped-${i}`, category: 'deals', retrieved: NOW - 20 * DAY });
+ const capped = createRetention({ db, now: () => NOW, log: silent, batchSize: 2, maxBatches: 1 });
+ const c1 = await capped.runOnce();
+ assert.deepStrictEqual(c1.categories, { deals: 2 });
+ assert.deepStrictEqual(c1.capped, ['deals']);
+ const c2 = await capped.runOnce();
+ assert.deepStrictEqual(c2.categories, { deals: 2 });
+ const c3 = await capped.runOnce();
+ assert.deepStrictEqual(c3.categories, { deals: 1 });
+ assert.deepStrictEqual(c3.capped, []);
+ assert.deepStrictEqual(await ids("WHERE id LIKE 'deals-capped-%'"), []);
+
+ // Runtime grace: a category the contract has but the rules do not is kept (never pruned),
+ // and an extra rule is ignored, each with one warning line per run. The strict check above is
+ // what makes the pin bump fail in CI; this is what keeps production serving meanwhile.
+ await item({ id: 'news-unruled', retrieved: NOW - 100 * DAY });
+ await item({ id: 'deals-injected', category: 'deals', retrieved: NOW - 20 * DAY });
+ const injectedRules = { ...RULES, weather: { days: 1, reason: 'test-only category the contract lacks' } };
+ delete injectedRules.news;
+ const injectedLogs = [];
+ const injected = createRetention({
+ db, rules: injectedRules, now: () => NOW,
+ log: { log: (m) => injectedLogs.push(m), warn: (m) => injectedLogs.push(m), error() {} },
+ });
+ const injectedOut = await injected.runOnce();
+ assert.deepStrictEqual(injectedOut.categories, { deals: 1 });
+ assert.deepStrictEqual(await ids("WHERE id IN ('news-unruled', 'deals-injected')"), ['news-unruled'], 'unruled news kept');
+ assert.strictEqual(injectedLogs.filter((m) => m === '[retention] no rule for category news: items kept').length, 1);
+ assert.strictEqual(injectedLogs.filter((m) => m === '[retention] rule for weather ignored: not a sources.source@1 category').length, 1);
+ await injected.runOnce();
+ assert.strictEqual(injectedLogs.filter((m) => m === '[retention] no rule for category news: items kept').length, 2, 'one warning per run');
+ assert.deepStrictEqual(await ids("WHERE id = 'news-unruled'"), ['news-unruled']);
+ await db.query("DELETE FROM item_revisions WHERE item_id = 'news-unruled'");
+ await db.query("DELETE FROM items WHERE id = 'news-unruled'");
+
+ // Keep-latest: one item per (source_key, identity), the newest. The items table already keeps
+ // one row per key, so the test drops its unique constraint to prove what the rule does if a
+ // source ever changes its identity format and leaves an older row for the same key.
+ const unique = await db.value(`SELECT conname FROM pg_constraint
+ WHERE conrelid = 'items'::regclass AND contype = 'u' AND pg_get_constraintdef(oid) LIKE 'UNIQUE (source_key, identity)'`);
+ assert.ok(unique, 'the items unique key is there');
+ await (owner || db).query(`ALTER TABLE items DROP CONSTRAINT "${unique}"`);
+ await item({ id: 'rent-old', category: 'housing', source: 'hud-fair-market-rents', identity: 'METRO1', retrieved: NOW - 2 * DAY, title: 'old value' });
+ await item({ id: 'rent-new', category: 'housing', source: 'hud-fair-market-rents', identity: 'METRO1', retrieved: NOW - 1 * DAY, title: 'new value' });
+ await item({ id: 'rent-other', category: 'housing', source: 'hud-fair-market-rents', identity: 'METRO2', retrieved: NOW - 400 * DAY, title: 'other area' });
+ const keep = await retention.runOnce();
+ assert.deepStrictEqual(keep.categories, { housing: 1 });
+ assert.deepStrictEqual(await ids("WHERE id LIKE 'rent-%'"), ['rent-new', 'rent-other']);
+ assert.strictEqual((await rows("WHERE id = 'rent-new'")).rows[0].title, 'new value');
+
+ // The repo's job helper arms the job, runs one on demand and stops it. (The real first run is
+ // five minutes out, so a test boot never prunes behind a test's back.)
+ const jobs = createJobs({ db, now: () => NOW, log: silent, retention: { firstMs: 3_600_000 } });
+ assert.strictEqual(jobs.start(), true);
+ assert.strictEqual(jobs.start(), false, 'starting twice is a no-op');
+ const viaHelper = await jobs.runOnce('retention');
+ assert.strictEqual(viaHelper.pruned, 0, 'the helper reaches the same prune');
+ assert.strictEqual(jobs.running(), 0);
+ await jobs.stop();
+ await assert.rejects(jobs.runOnce('nope'), /no job named nope/);
+
+ console.log('retention: all checks passed');
+ } finally {
+ if (owner) await owner.close().catch(() => {});
+ await testdb.close();
+ }
+})().catch((err) => { console.error(err); process.exit(1); });