diff --git a/CLAUDE.md b/CLAUDE.md index 65ed338..09c42c8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -271,6 +271,11 @@ REV official feed (2026-10-26, REV gave permission; REV is no longer crawled): - **REV's display terms.** Every REV listing links to `record.url`, REV's product page. `priceDate` (`app/model.ts`) shows "REV feed updated {date, time}" for official-feed records beside every price and stock: Buy list rows and summary, the part drawer (status strip, an Updated row, product choices), Overview's money card and Buy by vendor, and the Control Point preview caption; crawled records keep "as of {date}". The workbook's Buy list and the CSV gain a "Price and Stock As Of" column after Stock (14 Buy list columns; Line Total formulas unchanged), and Exceptions stock rows carry the same time. The Links sheet shows no prices, so it is unchanged. - **Not done here.** Publishing the feed to production KV is a separate owner-approved step (`catalog-load.ts stage --vendor rev`, `validate`, `publish`). +REV feed schedule and catalog refresh (2026-10-28, owner request "vendor data just recrawl slowly"): +- **REV cron.** `crons = ["*/10 * * * *"]`, mapped to REV in `cronVendors`; the four crawls stay off and a test checks that only REV is scheduled. REV's feed was published by hand on 2026-10-27 and nothing refreshed it. +- **CPU.** Staging all 976 rows in one run took about 17 ms in a fresh Node process (parse 10 ms, stage 7 ms; JSON.parse of the 322 KB body alone is 0.5 ms), over the free plan's 10 ms. `runRevFeed` therefore takes one step per run: a download (at most every six hours, with the ETag) checks the body is a list of at least 500 rows and keeps it at `cat:rev:feed:body`; each later run stages `revRowsPerRun` (125) rows as one batch, appending to the index; the last writes the manifest and deletes the body; a later run validates and publishes as before. Measured per step on the real feed, one fresh process each: 3.4–7.4 ms, idle runs 0.5–1.1 ms. `FeedState.building` holds the build (gen, the download's validators, rows done, counts, first rejection, `over`); the top-level ETag is replaced only when staging completes. A SKU listed twice keeps its first row (the one-run `stageCatalog` path keeps the last). A build is dropped, with its batches and body, when it fails, when a catalog is published by hand meanwhile, or three hours after its download (`revMaxBuildHours`), so a step stopped at the CPU limit cannot retry forever. +- **KV writes.** An idle run writes nothing; an unchanged check writes the state; a changed feed costs about 35 writes and deletes from download to publish, about 100 minutes. + ## Tuning, tests, and release Match normalization, thresholds, and alternative ranking are in `packages/shared/src/matching.ts`; an exact SKU still forces Buy absent a manual override. Disposition rules (`decideDisposition`), word lists, and print folders are in `packages/shared/src/classifier.ts`; order quantities are in `packages/shared/src/purchasing.ts`. Vendor stale hours and cron schedule are in `wrangler.toml`; catalog crawl batch limits are in `apps/worker/src/vendor-ingest.ts`. Scan request flow, source cache, and backoff are in `apps/worker/src/scan.ts`; BOM column mapping is in `packages/shared/src/bom.ts`; STEP export job bounds (per-call request budget, 5-minute translation timeout, 20 MB file cap) are in `apps/worker/src/step-export.ts`. Change these with fixture tests and record any changed data shape here. diff --git a/README.md b/README.md index 477296f..b7a706d 100644 --- a/README.md +++ b/README.md @@ -95,11 +95,11 @@ Failed parts are also shown in the app. If every part fails, no ZIP is produced ## Vendor catalog refresh -Vendor crawls are **paused** since 2026-10-25 at the owner's request: they used most of the free plan's daily KV writes, and the published catalogs stay as they are (staged loads with `scripts/catalog-load.ts` still work). When enabled, five staggered Worker crons run every hour: goBILDA at :00, ServoCity at :12, AndyMark at :24, REV at :36 (its official feed, never a crawl), and Pitsco at :48; the schedule is kept as a comment in `wrangler.toml`. Each run respects `robots.txt` and stays within [Cloudflare's 50-subrequest free-plan limit](https://developers.cloudflare.com/changelog/post/2026-02-11-subrequests-limit/). +Vendor crawls are **paused** since 2026-10-25 at the owner's request: they used most of the free plan's daily KV writes. goBILDA, ServoCity, AndyMark, and Pitsco are refreshed instead by a slow local crawl loaded with `scripts/catalog-load.ts` (see one-time loads below). The only scheduled cron is REV's official feed, every ten minutes (`*/10 * * * *`). When crawls are enabled again, each vendor's hourly cron is listed in `wrangler.toml`: goBILDA at :00, ServoCity at :12, AndyMark at :24, and Pitsco at :48. Each run respects `robots.txt` and stays within [Cloudflare's 50-subrequest free-plan limit](https://developers.cloudflare.com/changelog/post/2026-02-11-subrequests-limit/). - **AndyMark and Pitsco** are Shopify stores. A run reads up to four pages of the store's public `/products.json` feed, 250 products each, with every variant's SKU, price, and availability. A full catalog takes two or three runs (AndyMark: 9 pages, 5,313 SKUs; Pitsco: 6 pages, 1,625 SKUs, as of 2026-10-25). Fetching their product pages one by one drew HTTP 429 responses. - **goBILDA and ServoCity** are BigCommerce stores with no public feed. A run fetches at most 40 product pages from the product sitemap, at no more than two requests per second. A first full crawl takes about 60 runs for goBILDA (2,387 pages) and 130 for ServoCity (5,152). -- **REV** is never crawled. Since 2026-10-26 REV's catalog comes only from its official feed, used with REV's permission: `https://dashboard.revrobotics.com/feeds/usa/products.json`, refreshed daily at midnight Central Time (`apps/worker/src/rev-feed.ts`). Its cron checks the feed at most every six hours with the last ETag, stages a changed feed, and publishes it on a later run after validating it again; an unreachable, malformed, or truncated feed (under 500 products, or over 10% unusable rows) leaves the live catalog as it was. REV's terms are shown in the app and workbook: every REV product links to its REV product page, and every REV price and stock shows the feed's update time. To load it by hand: `npx tsx scripts/catalog-load.ts stage --vendor rev`, then `validate` and `publish` as below (REV refuses `--crawl` and `--spot`, which would fetch its pages). +- **REV** is never crawled. Since 2026-10-26 REV's catalog comes only from its official feed, used with REV's permission: `https://dashboard.revrobotics.com/feeds/usa/products.json`, refreshed daily at midnight Central Time (`apps/worker/src/rev-feed.ts`; the copy read on 2026-10-28 carried a Last-Modified of 11:00 UTC, so the cron checks every six hours rather than at one fixed time). Its cron runs every ten minutes and takes one small step per run, so each stays within the free plan's 10 ms of CPU (staging all 976 rows in one run took about 17 ms): at most every six hours it downloads the feed with the last ETag; a changed feed is kept in KV and staged 125 rows per run, then published on a later run after validating it again. A changed feed goes live about 100 minutes after its download, and a run with nothing to do writes nothing (a refresh costs about 35 KV writes and deletes). An unreachable, malformed, or truncated feed (under 500 products, or over 10% unusable rows), or a build unfinished three hours after its download, leaves the live catalog as it was. REV's terms are shown in the app and workbook: every REV product links to its REV product page, and every REV price and stock shows the feed's update time. To load it by hand: `npx tsx scripts/catalog-load.ts stage --vendor rev`, then `validate` and `publish` as below (REV refuses `--crawl` and `--spot`, which would fetch its pages). A rate-limited request (HTTP 429) is retried twice after the vendor's `Retry-After`, at most 10 seconds, before the run stops and resumes next hour. @@ -152,7 +152,7 @@ Linking and sending follow the two-sided contract in [docs/control-point-link.md ## Deploy and verify -The production Cloudflare Pages project, Worker, and two KV namespaces already exist. The namespace IDs and the (paused) crawl schedule are in `wrangler.toml`. In a new account, create equivalent Pages, Worker, and KV resources first and replace the namespace IDs. Put `SESSION_ENCRYPTION_KEY`, `ONSHAPE_OAUTH_CLIENT_ID`, and `ONSHAPE_OAUTH_CLIENT_SECRET` in the Worker as secrets, not in `wrangler.toml`, Pages variables, or Git. The optional AI review needs `GEMINI_API_KEY` and/or `ANTHROPIC_API_KEY` as Worker secrets (`pnpm --filter @control-kit/worker exec wrangler secret put GEMINI_API_KEY --config ../../wrangler.toml`, typed at the prompt, never in chat or Git); without either the review stays off and nothing else changes. Control Point linking needs `CONTROL_KIT_CLIENT_SECRET`, the same value Control Point holds in its Render environment, set the same way; without it linking stays off. The owner created the OAuth app and installed all three encrypted Worker secrets on 2026-10-23; production now returns `{"connected":false}` to an unauthenticated `/api/session` request rather than `setupPending`. +The production Cloudflare Pages project, Worker, and two KV namespaces already exist. The namespace IDs and the cron schedule (REV's feed only; crawls paused) are in `wrangler.toml`. In a new account, create equivalent Pages, Worker, and KV resources first and replace the namespace IDs. Put `SESSION_ENCRYPTION_KEY`, `ONSHAPE_OAUTH_CLIENT_ID`, and `ONSHAPE_OAUTH_CLIENT_SECRET` in the Worker as secrets, not in `wrangler.toml`, Pages variables, or Git. The optional AI review needs `GEMINI_API_KEY` and/or `ANTHROPIC_API_KEY` as Worker secrets (`pnpm --filter @control-kit/worker exec wrangler secret put GEMINI_API_KEY --config ../../wrangler.toml`, typed at the prompt, never in chat or Git); without either the review stays off and nothing else changes. Control Point linking needs `CONTROL_KIT_CLIENT_SECRET`, the same value Control Point holds in its Render environment, set the same way; without it linking stays off. The owner created the OAuth app and installed all three encrypted Worker secrets on 2026-10-23; production now returns `{"connected":false}` to an unauthenticated `/api/session` request rather than `setupPending`. After `pnpm install --frozen-lockfile` and the checks above, deploy from the repository root: diff --git a/apps/worker/src/index.test.ts b/apps/worker/src/index.test.ts index 3980373..0d6fb0c 100644 --- a/apps/worker/src/index.test.ts +++ b/apps/worker/src/index.test.ts @@ -61,9 +61,10 @@ describe('match decisions', () => { describe('vendor crons', () => { it('runs every vendor from a cron that wrangler.toml schedules', () => { const crons = JSON.parse(readFileSync(new URL('../../../wrangler.toml', import.meta.url), 'utf8').match(/^crons = (\[.*\])$/m)![1]) as string[]; - // Paused at the owner's request: nothing is scheduled, and anything scheduled later must map to a vendor. - expect(crons).toEqual([]); + // Crawls are paused at the owner's request: only REV's official feed runs, and anything scheduled later must map to a vendor. + expect(crons).toEqual(['*/10 * * * *']); expect(crons.every(cron => cron in cronVendors)).toBe(true); + expect(crons.map(cron => cronVendors[cron])).toEqual(['rev']); expect(new Set(Object.values(cronVendors)).size).toBe(5); }); }); diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 141ee34..2c9238e 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -541,11 +541,12 @@ async function route(request: Request, env: Env): Promise { return reply({ error: 'Not found' }, 404); } /** - * Each vendor crawls hourly, 12 minutes apart, when its cron is scheduled in wrangler.toml; REV's cron reads REV's official - * feed instead and never crawls. The crons are paused (an empty list) at the owner's request; every cron that is - * scheduled must be one of these (a test checks). + * Each vendor crawls hourly, 12 minutes apart, when its cron is scheduled in wrangler.toml. The crawls are paused at the + * owner's request; only REV's cron is scheduled. It never crawls: every ten minutes it takes one small step of reading + * REV's official feed (a download at most every six hours, then a few runs to stage and publish it), and a run with + * nothing to do writes nothing. Every cron that is scheduled must be one of these (a test checks). */ -export const cronVendors: Record = { '0 * * * *': 'gobilda', '12 * * * *': 'servocity', '24 * * * *': 'andymark', '36 * * * *': 'rev', '48 * * * *': 'pitsco' }; +export const cronVendors: Record = { '0 * * * *': 'gobilda', '12 * * * *': 'servocity', '24 * * * *': 'andymark', '*/10 * * * *': 'rev', '48 * * * *': 'pitsco' }; export default { async fetch(request: Request, env: Env) { @@ -566,7 +567,7 @@ export default { }, scheduled(event: { cron: string }, env: Env, context: { waitUntil(promise: Promise): void }) { const vendor = cronVendors[event.cron]; - // REV is never crawled: its cron reads REV's official feed (at most every six hours) and publishes it once staged. + // REV is never crawled: its cron reads REV's official feed (at most every six hours) and stages and publishes it in steps. if (vendor === 'rev') context.waitUntil(runRevFeed(env.VENDOR_CATALOG, fetch)); else if (vendor) context.waitUntil(runVendor(vendor, env.VENDOR_CATALOG, fetch, { feedFreshHours: Number(env.VENDOR_STALE_HOURS ?? 336) })); }, diff --git a/apps/worker/src/rev-feed.test.ts b/apps/worker/src/rev-feed.test.ts index d53bf6c..36576f9 100644 --- a/apps/worker/src/rev-feed.test.ts +++ b/apps/worker/src/rev-feed.test.ts @@ -4,7 +4,7 @@ import { aggregateAssembly, classifyScan, type AssemblyDefinition, type VendorRe import { clearCatalogMemoryCache, enrichScan } from './catalog'; import worker, { type Env, type Kv } from './index'; import { propagationMs } from './catalog-publish'; -import { parseRevFeed, revFeedStateKey, revFeedUrl, runRevFeed } from './rev-feed'; +import { parseRevFeed, revFeedBodyKey, revFeedStateKey, revFeedUrl, revRowsPerRun, runRevFeed, type RevFeedRun } from './rev-feed'; import { batchKey, indexKey, pointerKey, runVendor, snapshotKey, type CatalogPointer, type VendorSnapshot } from './vendor-ingest'; class MemoryKv implements Kv { @@ -42,6 +42,12 @@ async function liveRecords(kv: MemoryKv) { const pointer = json(kv, pointerKey('rev')); return Array.from({ length: pointer.batches ?? 0 }, (_, batch) => json(kv, batchKey('rev', pointer.gen, batch))).flat(); } +/** Runs the cron until the downloaded feed is staged: one run downloads it, then each run stages `revRowsPerRun` rows. */ +async function stageFeed(kv: MemoryKv, fetcher: Parameters[1], now: () => Date) { + const runs: RevFeedRun[] = []; + do runs.push(await runRevFeed(kv, fetcher, { now })); while (runs.at(-1)!.action === 'building' && runs.length < 20); + return { ...runs.at(-1)!, runs: runs.length }; +} afterEach(() => { vi.restoreAllMocks(); vi.unstubAllGlobals(); }); describe('REV official feed mapping', () => { @@ -89,7 +95,7 @@ describe('REV catalog from the official feed', () => { vi.stubGlobal('fetch', fetcher); const kv = new MemoryKv(); const pending: Promise[] = []; - worker.scheduled({ cron: '36 * * * *' }, { VENDOR_CATALOG: kv } as unknown as Env, { waitUntil: promise => pending.push(promise) }); + worker.scheduled({ cron: '*/10 * * * *' }, { VENDOR_CATALOG: kv } as unknown as Env, { waitUntil: promise => pending.push(promise) }); await Promise.all(pending); expect(fetcher.mock.calls.map(([url]) => url)).toEqual([revFeedUrl]); expect(new URL(revFeedUrl).hostname).toBe('dashboard.revrobotics.com'); @@ -101,7 +107,9 @@ describe('REV catalog from the official feed', () => { const fetcher = feedServer(() => fullFeed()); let clock = Date.parse('2026-10-26T12:00:00Z'); const now = () => new Date(clock); - expect(await runRevFeed(kv, fetcher, { now })).toMatchObject({ action: 'staged', records: 607 }); + // One run downloads the feed; five more stage its 607 rows, 125 at a time. + expect(await stageFeed(kv, fetcher, now)).toMatchObject({ action: 'staged', records: 607, runs: 6 }); + expect(kv.values.has(revFeedBodyKey)).toBe(false); // Staged, not live: scans still read the crawled catalog until KV has propagated the new one. expect(json(kv, pointerKey('rev')).gen).toBe('crawl1'); expect(await runRevFeed(kv, fetcher, { now })).toMatchObject({ action: 'waiting' }); @@ -141,7 +149,7 @@ describe('REV catalog from the official feed', () => { crawledCatalog(kv); let clock = Date.parse('2026-10-26T12:00:00Z'); const now = () => new Date(clock); - await runRevFeed(kv, feedServer(() => fullFeed()), { now }); + await stageFeed(kv, feedServer(() => fullFeed()), now); const { staged } = json<{ staged: { gen: string } }>(kv, revFeedStateKey); // An operator publishes another generation (catalog-load) before the cron's next run. kv.values.set(pointerKey('rev'), JSON.stringify({ ...json(kv, pointerKey('rev')), gen: 'manual' })); @@ -157,7 +165,7 @@ describe('REV catalog from the official feed', () => { let clock = Date.parse('2026-10-26T12:00:00Z'); const now = () => new Date(clock); const fetcher = feedServer(() => fullFeed()); - await runRevFeed(kv, fetcher, { now }); + await stageFeed(kv, fetcher, now); const state = kv.values.get(revFeedStateKey)!; clock += propagationMs; await runRevFeed(kv, fetcher, { now }); @@ -179,7 +187,7 @@ describe('REV catalog from the official feed', () => { let clock = Date.parse('2026-10-26T12:00:00Z'); const now = () => new Date(clock); const fetcher = feedServer(() => fullFeed()); - await runRevFeed(kv, fetcher, { now }); + await stageFeed(kv, fetcher, now); clock += propagationMs; await runRevFeed(kv, fetcher, { now }); const pointer = json(kv, pointerKey('rev')); @@ -194,7 +202,7 @@ describe('REV catalog from the official feed', () => { crawledCatalog(kv); let clock = Date.parse('2026-10-26T12:00:00Z'); const now = () => new Date(clock); - await runRevFeed(kv, feedServer(() => fullFeed()), { now }); + await stageFeed(kv, feedServer(() => fullFeed()), now); const { staged } = json<{ staged: { gen: string } }>(kv, revFeedStateKey); kv.values.delete(batchKey('rev', staged.gen, 0)); clock += propagationMs; @@ -202,6 +210,120 @@ describe('REV catalog from the official feed', () => { expect(json(kv, pointerKey('rev')).gen).toBe('crawl1'); expect([...kv.values.keys()].some(key => key.includes(staged.gen))).toBe(false); }); + it('stages a downloaded feed a few rows per run, so each run stays within the free plan\'s CPU limit', async () => { + quiet(); + const kv = new MemoryKv(); + crawledCatalog(kv); + const now = () => new Date('2026-10-26T12:00:00Z'); + const fetcher = feedServer(() => fullFeed()); + // The download run keeps the body and maps no rows. + expect(await runRevFeed(kv, fetcher, { now })).toMatchObject({ action: 'building', records: 0 }); + expect(kv.values.has(revFeedBodyKey)).toBe(true); + const gen = json<{ building: { gen: string } }>(kv, revFeedStateKey).building.gen; + expect(await runRevFeed(kv, fetcher, { now })).toMatchObject({ action: 'building', records: revRowsPerRun }); + expect(json(kv, batchKey('rev', gen, 0))).toHaveLength(revRowsPerRun); + // Nothing being built is visible: the crawled catalog stays live and no manifest exists until the last rows. + expect(json(kv, pointerKey('rev')).gen).toBe('crawl1'); + expect([...kv.values.keys()].some(key => key.includes(`${gen}:manifest`))).toBe(false); + expect(await stageFeed(kv, fetcher, now)).toMatchObject({ action: 'staged', records: 607 }); + expect(json(kv, indexKey('rev', gen))).toHaveLength(607); + expect(fetcher).toHaveBeenCalledTimes(1); + }); + it('keeps the first row of a SKU the feed lists twice', async () => { + quiet(); + const kv = new MemoryKv(); + const rows = JSON.parse(fullFeed()) as Array>; + // The repeat is in a later run's rows, so it is caught against what earlier runs staged. + rows.splice(450, 0, { ...rows[0], price: 1 }); + await stageFeed(kv, feedServer(() => JSON.stringify(rows)), () => new Date('2026-10-26T12:00:00Z')); + const { staged } = json<{ staged: { gen: string; records: number } }>(kv, revFeedStateKey); + expect(staged.records).toBe(607); + const all = Array.from({ length: 5 }, (_, batch) => json(kv, batchKey('rev', staged.gen, batch)) ?? []).flat(); + expect(all.filter(record => record.sku === rows[0].sku)).toMatchObject([{ priceUsd: rows[0].price }]); + }); + it('discards a half-staged feed whose rows turn out mostly unusable, with its saved body', async () => { + quiet(); + const kv = new MemoryKv(); + crawledCatalog(kv); + const now = () => new Date('2026-10-26T12:00:00Z'); + const rows = JSON.parse(fullFeed()) as Array>; + for (const row of rows.slice(500)) row.url = 'https://example.com/'; + const fetcher = feedServer(() => JSON.stringify(rows)); + await expect(stageFeed(kv, fetcher, now)).rejects.toThrow(/rows rejected/); + expect(json(kv, pointerKey('rev')).gen).toBe('crawl1'); + expect([...kv.values.keys()].filter(key => key.startsWith('cat:rev:gen:') && !key.includes('crawl1'))).toEqual([]); + expect(kv.values.has(revFeedBodyKey)).toBe(false); + expect(json<{ building: unknown; etag: unknown }>(kv, revFeedStateKey)).toMatchObject({ building: null, etag: null }); + expect(json(kv, snapshotKey('rev'))).toMatchObject({ status: 'failed', version: 'crawl1' }); + // Within six hours of the download nothing is fetched again. + expect(await runRevFeed(kv, fetcher, { now })).toMatchObject({ action: 'skipped' }); + expect(fetcher).toHaveBeenCalledTimes(1); + }); + it('ends a build that has not finished three hours after its download instead of retrying it every run', async () => { + quiet(); + const kv = new MemoryKv(); + crawledCatalog(kv); + let clock = Date.parse('2026-10-26T12:00:00Z'); + const now = () => new Date(clock); + const fetcher = feedServer(() => fullFeed()); + await runRevFeed(kv, fetcher, { now }); + await runRevFeed(kv, fetcher, { now }); + const gen = json<{ building: { gen: string } }>(kv, revFeedStateKey).building.gen; + clock += 3 * 3600000 + 1; + await expect(runRevFeed(kv, fetcher, { now })).rejects.toThrow(/did not finish/); + expect([...kv.values.keys()].some(key => key.includes(gen))).toBe(false); + expect(kv.values.has(revFeedBodyKey)).toBe(false); + expect(json(kv, snapshotKey('rev'))).toMatchObject({ status: 'failed', version: 'crawl1' }); + // The next download waits for the six-hour check. + expect(await runRevFeed(kv, fetcher, { now })).toMatchObject({ action: 'skipped' }); + clock += 3 * 3600000; + expect(await runRevFeed(kv, fetcher, { now })).toMatchObject({ action: 'building' }); + }); + it('retries a step that stopped before saving its progress without counting its own rows as duplicates', async () => { + quiet(); + const kv = new MemoryKv(); + const now = () => new Date('2026-10-26T12:00:00Z'); + const fetcher = feedServer(() => fullFeed()); + await runRevFeed(kv, fetcher, { now }); + await runRevFeed(kv, fetcher, { now }); + const saved = kv.values.get(revFeedStateKey)!; + // The second batch and the index are written, then the run stops before its state is saved. + await runRevFeed(kv, fetcher, { now }); + kv.values.set(revFeedStateKey, saved); + expect(await stageFeed(kv, fetcher, now)).toMatchObject({ action: 'staged', records: 607 }); + const { staged } = json<{ staged: { gen: string } }>(kv, revFeedStateKey); + expect(json<{ rejected: number }>(kv, `cat:rev:gen:${staged.gen}:manifest`).rejected).toBe(0); + expect(json(kv, indexKey('rev', staged.gen))).toHaveLength(607); + }); + it('removes a batch written by a step whose index write failed', async () => { + quiet(); + const kv = new MemoryKv(); + crawledCatalog(kv); + const now = () => new Date('2026-10-26T12:00:00Z'); + const fetcher = feedServer(() => fullFeed()); + await runRevFeed(kv, fetcher, { now }); + const gen = json<{ building: { gen: string } }>(kv, revFeedStateKey).building.gen; + const put = kv.put.bind(kv); + vi.spyOn(kv, 'put').mockImplementation(async (key, value) => { if (key === indexKey('rev', gen)) throw new Error('KV write failed'); return put(key, value); }); + await expect(runRevFeed(kv, fetcher, { now })).rejects.toThrow('KV write failed'); + expect([...kv.values.keys()].some(key => key.includes(gen))).toBe(false); + expect(kv.values.has(revFeedBodyKey)).toBe(false); + }); + it('drops a half-staged feed when a catalog is published by hand meanwhile', async () => { + quiet(); + const kv = new MemoryKv(); + crawledCatalog(kv); + const now = () => new Date('2026-10-26T12:00:00Z'); + const fetcher = feedServer(() => fullFeed()); + await runRevFeed(kv, fetcher, { now }); + await runRevFeed(kv, fetcher, { now }); + const gen = json<{ building: { gen: string } }>(kv, revFeedStateKey).building.gen; + kv.values.set(pointerKey('rev'), JSON.stringify({ ...json(kv, pointerKey('rev')), gen: 'manual' })); + expect(await runRevFeed(kv, fetcher, { now })).toMatchObject({ action: 'abandoned' }); + expect([...kv.values.keys()].some(key => key.includes(gen))).toBe(false); + expect(kv.values.has(revFeedBodyKey)).toBe(false); + expect(json(kv, pointerKey('rev')).gen).toBe('manual'); + }); }); describe('a REV buy line from the feed', () => { @@ -212,7 +334,7 @@ describe('a REV buy line from the feed', () => { crawledCatalog(kv); let clock = Date.parse('2026-10-26T12:00:00Z'); const fetcher = feedServer(() => fullFeed()); - await runRevFeed(kv, fetcher, { now: () => new Date(clock) }); + await stageFeed(kv, fetcher, () => new Date(clock)); clock += propagationMs; await runRevFeed(kv, fetcher, { now: () => new Date(clock) }); const fixture = JSON.parse(readFileSync(new URL('../../../fixtures/assembly-simple.json', import.meta.url), 'utf8')) as AssemblyDefinition; diff --git a/apps/worker/src/rev-feed.ts b/apps/worker/src/rev-feed.ts index 5668f87..4be8b33 100644 --- a/apps/worker/src/rev-feed.ts +++ b/apps/worker/src/rev-feed.ts @@ -1,7 +1,7 @@ import { isBundle, normalizeName, normalizeSku, type VendorRecord } from '@control-kit/shared'; import type { Kv } from './index'; -import { manifestKey, propagationMs, publishStaged, stageCatalog, validateStaged } from './catalog-publish'; -import { allowedUrl, batchKey, indexKey, pointerKey, retireReplaced, snapshotKey, vendorSources, type CatalogPointer, type VendorSnapshot } from './vendor-ingest'; +import { manifestKey, propagationMs, publishStaged, validateStaged, type StagedCatalog } from './catalog-publish'; +import { allowedUrl, batchKey, indexKey, pointerKey, retireReplaced, snapshotKey, vendorSources, type CatalogPointer, type IndexEntry, type VendorSnapshot } from './vendor-ingest'; /* * REV Robotics' official product feed, used with REV's permission since 2026-10-26; REV is never crawled. The feed is @@ -17,19 +17,30 @@ import { allowedUrl, batchKey, indexKey, pointerKey, retireReplaced, snapshotKey */ export const revFeedUrl = vendorSources.rev.officialFeed!; /** - * Cron state: the last response's validators, when it was checked, and a staged generation waiting to be published with - * the live generation it was staged against (`over`), so a catalog published meanwhile by hand is never replaced by it. + * Cron state: the last staged response's validators, when the feed was last checked, a generation being built from a + * downloaded feed (`building`), and a staged generation waiting to be published. Both carry the live generation they + * were started against (`over`), so a catalog published meanwhile by hand is never replaced by them. */ export const revFeedStateKey = 'cat:rev:feed'; +/** The downloaded feed body while its rows are being staged, deleted once staging ends. */ +export const revFeedBodyKey = 'cat:rev:feed:body'; /** The feed listed 976 products on 2026-10-26; one with under half of that is truncated, not a smaller catalog. */ export const revMinRecords = 500; +/** + * Rows staged per cron run. The free plan allows 10 ms of CPU per run. Mapping and staging all 976 rows in one run took + * about 17 ms in a fresh Node process (2026-10-28); a run of 125 rows, including parsing the 322 KB body, took 3.4–7.4 ms. + */ +export const revRowsPerRun = 125; +/** A build still unfinished this long after its download (runs are ten minutes apart) is dropped rather than retried. */ +const revMaxBuildHours = 3; /** More rejected rows than this means the feed's format changed; staging what is left would drop most of the catalog. */ const maxRejectedShare = 0.1; const userAgent = 'Control Kit vendor catalog (+https://github.com/ZOGWORK/control-kit)'; export type RevFeedRejection = { row: number; sku: string | null; reason: string }; -type FeedState = { etag: string | null; lastModified: string | null; checkedAt: string; staged: { gen: string; stagedAt: string; updatedAt: string; records: number; over: string | null } | null }; -export type RevFeedRun = { vendor: 'rev'; source: 'official-feed'; action: 'skipped' | 'unchanged' | 'staged' | 'waiting' | 'published' | 'rejected' | 'abandoned'; records: number; problems: string[] }; +type Building = { gen: string; etag: string | null; lastModified: string | null; updatedAt: string; rows: number; next: number; batches: number; records: number; rejected: number; firstRejection: RevFeedRejection | null; over: string | null }; +type FeedState = { etag: string | null; lastModified: string | null; checkedAt: string; staged: { gen: string; stagedAt: string; updatedAt: string; records: number; over: string | null } | null; building?: Building | null }; +export type RevFeedRun = { vendor: 'rev'; source: 'official-feed'; action: 'skipped' | 'unchanged' | 'building' | 'staged' | 'waiting' | 'published' | 'rejected' | 'abandoned'; records: number; problems: string[] }; /** * Pack size from REV's own SKU: REV sells every multi-unit product under a "-PKn" SKU (387 of the 976 feed rows). A @@ -52,12 +63,24 @@ function revStock(status: unknown): VendorRecord['stock'] { /** Feed body → records, with each unusable row and why. Throws when the body is not a list of products. */ export function parseRevFeed(body: string, updatedAt: string): { records: VendorRecord[]; rejected: RevFeedRejection[]; rows: number } { + const rows = feedRows(body); + return { ...parseRevRows(rows, updatedAt), rows: rows.length }; +} +function feedRows(body: string): unknown[] { let rows: unknown; try { rows = JSON.parse(body); } catch { throw new Error('REV feed was not JSON'); } if (!Array.isArray(rows)) throw new Error('REV feed was not a list of products'); + return rows; +} +/** + * Rows → records; `first` is the position of `rows[0]` in the feed, so rejections name the feed's own row. With `seen` + * (SKUs already staged), a SKU listed again is rejected and its first row kept. + */ +function parseRevRows(rows: unknown[], updatedAt: string, first = 0, seen?: Set): { records: VendorRecord[]; rejected: RevFeedRejection[] } { const records: VendorRecord[] = []; const rejected: RevFeedRejection[] = []; - rows.forEach((raw: unknown, row) => { + rows.forEach((raw: unknown, at) => { + const row = first + at; const item = raw && typeof raw === 'object' ? raw as Record : {}; const sku = typeof item.sku === 'string' ? item.sku.trim() : ''; const name = typeof item.name === 'string' ? item.name.trim() : ''; @@ -65,23 +88,25 @@ export function parseRevFeed(body: string, updatedAt: string): { records: Vendor const normalizedSku = normalizeSku(sku, 'rev'); if (!sku || !/^[A-Z0-9]+$/.test(normalizedSku)) return rejected.push({ row, sku: sku || null, reason: 'no usable SKU' }); if (!name) return rejected.push({ row, sku, reason: 'no name' }); + if (seen?.has(normalizedSku)) return rejected.push({ row, sku, reason: 'SKU listed twice' }); // REV's first condition: the link is REV's own product page, so a row without one is not listed at all. if (!allowedUrl('rev', url) || new URL(url).pathname === '/') return rejected.push({ row, sku, reason: 'no product page on www.revrobotics.com' }); const price = typeof item.price === 'number' ? item.price : typeof item.price === 'string' && /^\d+(?:\.\d+)?$/.test(item.price.trim()) ? Number(item.price) : NaN; + seen?.add(normalizedSku); records.push({ vendorId: 'rev', sku, normalizedSku, name, normalizedName: normalizeName(name), url, priceUsd: Number.isFinite(price) && price > 0 ? Math.round(price * 100) / 100 : null, ...revPack(sku, name), stock: revStock(item.availability_status), stockQty: null, category: 'uncategorized', updatedAt, source: 'feed', parser: 'official-feed' }); }); - return { records, rejected, rows: rows.length }; + return { records, rejected }; } /** Records the cron or the load script may stage, or why the feed must not replace the live catalog. */ export function checkRevFeed(parsed: ReturnType): string[] { + return feedProblems(parsed.records.length, parsed.rejected.length, parsed.rows, parsed.rejected[0] ?? null); +} +function feedProblems(records: number, rejected: number, rows: number, first: RevFeedRejection | null) { const problems: string[] = []; - if (parsed.records.length < revMinRecords) problems.push(`Only ${parsed.records.length} usable products; expected at least ${revMinRecords}`); - if (parsed.rows && parsed.rejected.length / parsed.rows > maxRejectedShare) { - const first = parsed.rejected[0]; - problems.push(`${parsed.rejected.length} of ${parsed.rows} rows rejected (row ${first.row}${first.sku ? ` ${first.sku}` : ''}: ${first.reason})`); - } + if (records < revMinRecords) problems.push(`Only ${records} usable products; expected at least ${revMinRecords}`); + if (rows && rejected / rows > maxRejectedShare && first) problems.push(`${rejected} of ${rows} rows rejected (row ${first.row}${first.sku ? ` ${first.sku}` : ''}: ${first.reason})`); return problems; } @@ -108,20 +133,30 @@ export async function fetchRevFeed(fetcher: Fetcher = fetch, options: { etag?: s const json = async (kv: Kv, key: string): Promise => { const raw = await kv.get(key); if (!raw) return null; try { return JSON.parse(raw) as T; } catch { return null; } }; /** A staged generation no reader was ever pointed at, so its keys can go at once. One the pointer names is never touched. */ -async function discardStaged(kv: Kv, gen: string) { +async function discardStaged(kv: Kv, gen: string, batches = 0) { const live = await json(kv, pointerKey('rev')); if (live && [live.gen, live.previous?.gen, ...(live.retiring ?? []).map(entry => entry.gen)].includes(gen)) return; + // A generation still being built has no manifest yet; the cron state counts its batches instead. const manifest = await json<{ batches: number }>(kv, manifestKey('rev', gen)); - for (let batch = 0; batch < (manifest?.batches ?? 0); batch++) await kv.delete(batchKey('rev', gen, batch)); + for (let batch = 0; batch < Math.max(manifest?.batches ?? 0, batches); batch++) await kv.delete(batchKey('rev', gen, batch)); await kv.delete(indexKey('rev', gen)); await kv.delete(manifestKey('rev', gen)); } +/** Ends an unfinished build: its batches and the saved body go, and its validators are never kept. */ +async function dropBuilding(kv: Kv, state: FeedState) { + // One more than the saved count: a run may have written its batch and then failed before saving its progress. + if (state.building) await discardStaged(kv, state.building.gen, state.building.batches + 1); + await kv.delete(revFeedBodyKey); + await kv.put(revFeedStateKey, JSON.stringify({ ...state, etag: null, lastModified: null, building: null } satisfies FeedState)); +} /** * One scheduled step of the REV catalog, through the same staged publish as `scripts/catalog-load.ts`, so a bad feed never * replaces a good catalog. A run does one thing: publish a staged generation once KV has propagated it (after - * validating it again), or, at most every `checkEveryHours`, fetch the feed and stage it if it changed. A feed that is - * unreachable, malformed, truncated, or mostly unusable fails the run and leaves the live catalog as it was. Every run + * validating it again), stage the next `revRowsPerRun` rows of a downloaded feed, or, at most every `checkEveryHours`, + * fetch the feed and keep it for staging if it changed. A changed feed of 976 rows therefore goes live over ten runs (one + * download, eight of rows, one to publish). A feed that is unreachable, malformed, truncated, or mostly + * unusable fails the run, discards what was staged of it, and leaves the live catalog as it was. Every run * first deletes replaced generations whose day of rollback is over (at most 20 keys), as a crawl run would. */ export async function runRevFeed(kv: Kv, fetcher: Fetcher = fetch, options: { now?: () => Date; checkEveryHours?: number } = {}): Promise { @@ -169,6 +204,47 @@ export async function runRevFeed(kv: Kv, fetcher: Fetcher = fetch, options: { no await kv.put(revFeedStateKey, JSON.stringify({ ...state, staged: null } satisfies FeedState)); return summary('published', published.records); } + if (state?.building) { + const building = state.building; + if ((pointer?.gen ?? null) !== building.over) { + await dropBuilding(kv, state); + return summary('abandoned', building.records); + } + // A step that keeps failing (stopped at the CPU limit, say) would retry the same rows every run; the build ends instead. + if (now().getTime() - Date.parse(state.checkedAt) > revMaxBuildHours * 3600000) throw new Error(`REV feed staging did not finish within ${revMaxBuildHours} hours`); + // The next `revRowsPerRun` rows become one batch. A SKU listed twice keeps its first row; the later one is rejected. + const rows = feedRows(await kv.get(revFeedBodyKey) ?? ''); + if (rows.length !== building.rows) throw new Error(`REV feed body changed while staging (${rows.length} rows, expected ${building.rows})`); + // Entries for this batch can only be left by a run that stopped before saving its progress; this run rewrites them. + const index = building.batches ? (await json(kv, indexKey('rev', building.gen)) ?? []).filter(entry => entry[2] < building.batches) : []; + const part = parseRevRows(rows.slice(building.next, building.next + revRowsPerRun), building.updatedAt, building.next, new Set(index.map(entry => entry[0]))); + const records = part.records.map(record => ({ ...record, catalogVersion: building.gen })); + if (records.length) { + await kv.put(batchKey('rev', building.gen, building.batches), JSON.stringify(records)); + for (const record of records) index.push([record.normalizedSku, record.normalizedName, building.batches]); + await kv.put(indexKey('rev', building.gen), JSON.stringify(index)); + building.batches++; + } + building.records += records.length; + building.rejected += part.rejected.length; + building.firstRejection ??= part.rejected[0] ?? null; + building.next += revRowsPerRun; + if (building.next < building.rows) { + await kv.put(revFeedStateKey, JSON.stringify(state satisfies FeedState)); + return summary('building', building.records); + } + const problems = feedProblems(building.records, building.rejected, building.rows, building.firstRejection); + if (problems.length) throw new Error(`REV feed not staged: ${problems.join('; ')}`); + const stagedAt = now().toISOString(); + const manifest: StagedCatalog = { vendor: 'rev', gen: building.gen, source: 'official-feed', records: building.records, batches: building.batches, stagedAt, skipped: {}, rejected: building.rejected, + note: `REV official feed, updated ${building.updatedAt}` }; + await kv.put(manifestKey('rev', building.gen), JSON.stringify(manifest)); + // Only now do the validators describe the staged catalog, so a 304 can be trusted for it. + await kv.put(revFeedStateKey, JSON.stringify({ etag: building.etag, lastModified: building.lastModified, checkedAt: state.checkedAt, + staged: { gen: building.gen, stagedAt, updatedAt: building.updatedAt, records: building.records, over: building.over }, building: null } satisfies FeedState)); + await kv.delete(revFeedBodyKey); + return summary('staged', building.records); + } if (state && now().getTime() - Date.parse(state.checkedAt) < (options.checkEveryHours ?? 6) * 3600000) return summary('skipped'); // Validators only count while the live catalog came from this feed; otherwise the feed is read in full. const fromFeed = pointer?.source === 'official-feed'; @@ -180,15 +256,19 @@ export async function runRevFeed(kv: Kv, fetcher: Fetcher = fetch, options: { no if (snapshot?.status === 'failed' && pointer) await kv.put(snapshotKey('rev'), JSON.stringify({ ...snapshot, status: 'ok', lastRunAt: now().toISOString(), error: undefined } satisfies VendorSnapshot)); return summary('unchanged', pointer?.records ?? 0); } - const parsed = parseRevFeed(response.body, response.updatedAt); - const problems = checkRevFeed(parsed); - if (problems.length) throw new Error(`REV feed not staged: ${problems.join('; ')}`); - const manifest = await stageCatalog('rev', parsed.records, kv, { source: 'official-feed', rejected: parsed.rejected.length, note: `REV official feed, updated ${response.updatedAt}`, now: now() }); - await kv.put(revFeedStateKey, JSON.stringify({ etag: response.etag, lastModified: response.lastModified, checkedAt: now().toISOString(), - staged: { gen: manifest.gen, stagedAt: manifest.stagedAt, updatedAt: response.updatedAt, records: manifest.records, over: pointer?.gen ?? null } } satisfies FeedState)); - return summary('staged', manifest.records); + // Staging is spread over the next runs to stay within the free plan's CPU limit. This run only checks that the body + // is a list long enough to be the whole feed and keeps it; the rows are mapped and checked as they are staged. + const rows = feedRows(response.body); + if (rows.length < revMinRecords) throw new Error(`REV feed not staged: Only ${rows.length} rows; expected at least ${revMinRecords}`); + await kv.put(revFeedBodyKey, response.body); + await kv.put(revFeedStateKey, JSON.stringify({ etag: state?.etag ?? null, lastModified: state?.lastModified ?? null, checkedAt: now().toISOString(), staged: null, + building: { gen: crypto.randomUUID().slice(0, 8), etag: response.etag, lastModified: response.lastModified, updatedAt: response.updatedAt, rows: rows.length, next: 0, batches: 0, records: 0, + rejected: 0, firstRejection: null, over: pointer?.gen ?? null } } satisfies FeedState)); + return summary('building', 0); } catch (cause) { const message = cause instanceof Error ? cause.message : 'REV feed import failed'; + // A generation that cannot be finished is removed with its body, and the next check downloads the feed again. + if (state?.building) await dropBuilding(kv, state).catch(() => {}); await failed(message); console.error(JSON.stringify({ vendor: 'rev', source: 'official-feed', error: message.slice(0, 200) })); throw cause; diff --git a/wrangler.toml b/wrangler.toml index c4863d8..a1a47f0 100644 --- a/wrangler.toml +++ b/wrangler.toml @@ -36,7 +36,10 @@ new_sqlite_classes = ["AiAllowance"] [triggers] # Vendor crawls are paused at the owner's request (2026-10-25): they used most of the free plan's daily KV writes, -# and the published catalogs stay as they are. To resume, restore the hourly schedule that `cronVendors` maps: -# ["0 * * * *", "12 * * * *", "24 * * * *", "36 * * * *", "48 * * * *"] -# REV is never crawled: "36 * * * *" reads REV's official feed (apps/worker/src/rev-feed.ts) at most every six hours. -crons = [] +# and the published catalogs stay as they are; they are refreshed by hand with scripts/catalog-load.ts. To resume a +# crawl, add its hourly cron from `cronVendors`: "0 * * * *" goBILDA, "12 * * * *" ServoCity, "24 * * * *" AndyMark, +# "48 * * * *" Pitsco. +# REV is never crawled. "*/10 * * * *" reads REV's official feed (apps/worker/src/rev-feed.ts): a download at most every +# six hours, then one small step per run to stage and publish it within the free plan's 10 ms of CPU. A run with nothing +# to do writes nothing to KV. +crons = ["*/10 * * * *"]