diff --git a/docs/api.md b/docs/api.md index 6f1692c..e8e08ad 100644 --- a/docs/api.md +++ b/docs/api.md @@ -687,6 +687,12 @@ leaves it free to keep a group however it likes, an append-log replayed on read included, so long as `readState` answers with what the last `writeState` was given. +The package keeps four for a site: `channels`, a list it fetched; `validators`, +what each url last answered with; `state`, the site's own bag; and +`remembered`, what a context's +[`remember`](./site-config.md#a-programmes-details-asked-once) kept, each +value with when it is forgotten. `StateKey` names them. + `FsCacheDriver` puts each group in `//.json`, beside the channel directories of the same site; a prune only ever looks at the day files, so state outlives every listing a site has. `SqliteCacheDriver` keeps a row per diff --git a/docs/site-config.md b/docs/site-config.md index 33f1b3a..15db351 100644 --- a/docs/site-config.md +++ b/docs/site-config.md @@ -305,6 +305,7 @@ What a `parseDay` call is given: | `programme` | `(start, title, options?) => ProgrammeBuilder` | A [builder bound to this channel-day](#building-programmes). | | `http` | `KyInstance` | The site's client — the very instance `request` was handed. See [a parse that needs another request](#a-parse-that-needs-another-request). | | `paced` | `(task) => Promise` | Run a request through the site's queue, so its `concurrency`, `rateLimit` and backoff apply to it. | +| `remember` | `(key, { until, miss? }, task) => Promise` | What an earlier run kept under `key`, or `task`'s answer, kept until `until` — see [a programme's details, asked once](#a-programmes-details-asked-once). | | `state` | `Map` | What this site [remembers between runs](#remembering-something-between-runs). | | `signal` | `AbortSignal \| undefined` | Already applied to `http`; here for work that does not go through it. | | `log`, `warn` | `(message, fields?) => void` | Somewhere to [say something](#saying-something), under the site's name. | @@ -424,6 +425,68 @@ Nothing else in the package reads it, and it is stored [beside the listings](./api.md#what-a-site-remembers-between-runs) — `state.json` in the site's cache directory. +### A programme's details, asked once + +A source that wants a request per programme for its details makes a nightly +refetch expensive: today is fetched again every run, and with it every detail +call for programmes already asked about. `remember` asks once and keeps the +answer for the runs after: + +```ts +async parseDay({ payload, http, paced, remember, programme }) { + return Promise.all(payload.items.map(async (item) => { + const details = await remember( + `detail:${item.id}|${item.start}`, + { + // Asked about while its day can be fetched again, and no longer. + until: Date.parse(item.end) + 86_400_000, + // "This one has no details" is an answer, and worth keeping. + miss: (error) => error instanceof HTTPError && error.response.status === 404, + }, + async () => { + const full = await paced(({ signal }) => + http.get(`detail/${item.id}`, { signal }).json()); + + return { desc: full.description, cast: full.cast }; // only what the parse uses + }, + ); + + const p = programme(new Date(item.start), item.title); + return details ? p.desc(details.desc) : p; + })); +} +``` + +It hands back what an earlier run kept under the key, or the task's answer, +kept until `until` — a `Date` or epoch milliseconds, always given, since how +long something is worth keeping is the site's to say. Only an answer is kept: a +failure `miss` matches is kept as `null`, and any other is thrown and kept by +nobody, so the next run asks again. A kept "none" is an answer only to an ask +with a `miss` of its own; one without asks again, and its type has no `null` +in it. + +Two asks of one key while the first is still on its way are one task — a +broadcast past midnight is in two days' listings — and each reads what that +task came to with its own `miss`. A parse waiting on another's lets its slot go +meanwhile, as it does for `paced`. A request cannot let its queue slot go, so +one asks for itself instead. + +Each caller is handed a copy, so changing what it was given changes nothing +kept. Like `state`, a value must be plain; a `Date` or a `Map` is refused +rather than coming back as a string or `{}` on the next run. An `until` already +past keeps the value for this run only, and says so once — usually an `until` +in seconds where milliseconds were meant. + +It is kept in a group of its own beside `state`, written whole when anything in +it changes, and what has passed its `until` is dropped at the end of every +run. So keep only what the parse uses, and pick keys that change when what they +are about does — a programme moved to another start is worth asking about +again. `--refresh` reads nothing an earlier run kept, as it reads no cached +listings, and still keeps what it is answered. Every context has it: a +request, a parse, and a pass over a whole document. A mark that lasts — "this +channel has no details, ask again in a week" — is a value remembered for a +week. + ### A parse that needs another request Plenty of sources put the schedule in one response and the description of each @@ -772,8 +835,8 @@ channel-day, its `grabbedAt` and `programmeCount`, and absent where it holds nothing — since a stale channel-day is usually there and merely old enough to ask about again. -And every mode gets `timeZone`, `dayZone` and [`at`](#times-of-day), as a parse -does. Under `'days'` and `'both'` there is no one day to assume, so `at` takes +And every mode gets `timeZone`, `dayZone`, [`at`](#times-of-day) and +[`remember`](#a-programmes-details-asked-once), as a parse does. Under `'days'` and `'both'` there is no one day to assume, so `at` takes the day as well — from the first day's midnight to the end of the last is the stretch a range API wants, on the clock of the site's days, where `from` and `to` are only UTC midnights: diff --git a/src/cli/try.ts b/src/cli/try.ts index bf348de..6659e57 100644 --- a/src/cli/try.ts +++ b/src/cli/try.ts @@ -36,6 +36,7 @@ import { kyWith, resolveChannels, siteHttp } from '../grabber/channels.js'; import { parseContext, requestContext, streamContext } from '../grabber/context.js'; import { planRequests } from '../grabber/planner.js'; import { resolveSite } from '../grabber/site.js'; +import { Memory } from '../grabber/remember.js'; import { SiteStateHandle } from '../grabber/state.js'; import type { AnySiteConfig, GrabberChannel, ParsedProgramme } from '../grabber/types.js'; import { ProgrammeBuilder } from '../xmltv/builder.js'; @@ -405,6 +406,10 @@ export async function tryChannelDay( task({ ...(options.signal ? { signal: options.signal } : {}) }), timeZone: config.timeZone, dayZone: resolved.dayZone, + // Over a store that keeps nothing, as `state` is: remembered for as long + // as this command, so a parse asking twice sees the first answer, and by + // no run after it. + memory: new Memory(state, site.site, now), }; const began = Date.now(); diff --git a/src/core/plain.ts b/src/core/plain.ts new file mode 100644 index 0000000..af569a1 --- /dev/null +++ b/src/core/plain.ts @@ -0,0 +1,71 @@ +/** + * Whether a value comes back from JSON as it went in. + * + * What the cache keeps is JSON, so anything handed to it to keep — a channel's + * `data`, a remembered value — has to be a value JSON gives back unchanged, or a + * run that wrote it and the run that reads it would see different things, and a + * site would find out only on the second. + */ + +/** + * The first thing in a value that is not plain, named with where it is — or + * nothing, when all of it is. + * + * Plain is what JSON gives back as it was given: strings, finite numbers, + * booleans, `null`, arrays and plain objects. A property holding `undefined` is + * fine, since leaving it out is all JSON does with it and reading it gives the + * same. Anything else comes back as something else — a `Date` a string, a `Map` + * an empty object, an instance of a class without its methods, `NaN` a `null` — + * or not at all, as a function does. + */ +export function unplain( + value: unknown, + at: string, + ancestors = new Set(), +): string | undefined { + if (value === null || typeof value === 'string' || typeof value === 'boolean') { + return undefined; + } + + if (typeof value === 'number') { + return Number.isFinite(value) ? undefined : `${String(value)} at ${at}`; + } + + if (typeof value !== 'object') { + return `${typeof value === 'undefined' ? 'undefined' : `a ${typeof value}`} at ${at}`; + } + + // The objects above this one, not every one seen: the same object twice in + // a list is fine — JSON writes it twice — and only one inside itself is not. + if (ancestors.has(value)) { + return `a circular reference at ${at}`; + } + + const prototype: unknown = Object.getPrototypeOf(value); + + if (!Array.isArray(value) && prototype !== Object.prototype && prototype !== null) { + const name = (value as { constructor?: { name?: string } }).constructor?.name; + + return `${name ? `an instance of ${name}` : 'an object of a class'} at ${at}`; + } + + ancestors.add(value); + + try { + for (const [key, item] of Object.entries(value)) { + const where = Array.isArray(value) ? `${at}[${key}]` : `${at}.${key}`; + // `undefined` as a property is left out, which reads the same; in an + // array it becomes `null`, which does not. + const found = + item === undefined && !Array.isArray(value) ? undefined : unplain(item, where, ancestors); + + if (found !== undefined) { + return found; + } + } + + return undefined; + } finally { + ancestors.delete(value); + } +} diff --git a/src/grabber/channels.ts b/src/grabber/channels.ts index c4f39dd..54e64f6 100644 --- a/src/grabber/channels.ts +++ b/src/grabber/channels.ts @@ -8,6 +8,7 @@ import type { ChannelCoverage } from '../merge/types.js'; import type { CacheStore } from '../cache/types.js'; import { localTimeZone } from '../core/days.js'; import { errorMessage, unanswered } from '../core/error.js'; +import { unplain } from '../core/plain.js'; import type { Emit, Says } from '../core/events.js'; import type { SiteSession } from './session.js'; import type { @@ -186,65 +187,6 @@ export interface ResolveChannelsOptions { paced?: (fetch: () => Promise) => Promise; } -/** - * The first thing in a channel's `data` that is not a plain value, named with - * where it is — or nothing, when all of it is. - * - * Plain is what JSON gives back as it was given: strings, finite numbers, - * booleans, `null`, arrays and plain objects. A property holding `undefined` is - * fine, since leaving it out is all JSON does with it and reading it gives the - * same. Anything else comes back as something else — a `Date` a string, a `Map` - * an empty object, an instance of a class without its methods, `NaN` a `null` — - * or not at all, as a function does. - */ -function unplain(value: unknown, at: string, ancestors = new Set()): string | undefined { - if (value === null || typeof value === 'string' || typeof value === 'boolean') { - return undefined; - } - - if (typeof value === 'number') { - return Number.isFinite(value) ? undefined : `${String(value)} at ${at}`; - } - - if (typeof value !== 'object') { - return `${typeof value === 'undefined' ? 'undefined' : `a ${typeof value}`} at ${at}`; - } - - // The objects above this one, not every one seen: the same object twice in - // a list is fine — JSON writes it twice — and only one inside itself is not. - if (ancestors.has(value)) { - return `a circular reference at ${at}`; - } - - const prototype: unknown = Object.getPrototypeOf(value); - - if (!Array.isArray(value) && prototype !== Object.prototype && prototype !== null) { - const name = (value as { constructor?: { name?: string } }).constructor?.name; - - return `${name ? `an instance of ${name}` : 'an object of a class'} at ${at}`; - } - - ancestors.add(value); - - try { - for (const [key, item] of Object.entries(value)) { - const where = Array.isArray(value) ? `${at}[${key}]` : `${at}.${key}`; - // `undefined` as a property is left out, which reads the same; in an - // array it becomes `null`, which does not. - const found = - item === undefined && !Array.isArray(value) ? undefined : unplain(item, where, ancestors); - - if (found !== undefined) { - return found; - } - } - - return undefined; - } finally { - ancestors.delete(value); - } -} - /** * Refuse a channel list whose `data` holds anything but plain values. * diff --git a/src/grabber/context.ts b/src/grabber/context.ts index 27a45b7..13e44c3 100644 --- a/src/grabber/context.ts +++ b/src/grabber/context.ts @@ -15,6 +15,7 @@ import { dayToDate, wallClock } from '../core/days.js'; import type { Says } from '../core/events.js'; import { ProgrammeBuilder } from '../xmltv/builder.js'; +import type { Away, Memory } from './remember.js'; import type { Request } from './planner.js'; import type { ResolvedBatching } from './planner.js'; import type { @@ -24,6 +25,8 @@ import type { GrabberChannel, PacedRequest, ParseContext, + Remember, + RememberOptions, RequestContextFor, SiteState, StreamContext, @@ -42,6 +45,16 @@ export interface ContextDeps { timeZone: string; /** The zone the site's days are on: the run's, or the one its `dayZone` names. */ dayZone: string; + /** What the site remembers between runs, for a context's `remember`. */ + memory: Memory; + /** + * How this context waits on something another one started — an ask of + * `remember`'s — without holding a slot meanwhile: a parse lends its + * `localWork` slot, as `paced` does, and a stream holds none. Absent where + * the context holds one it cannot lend, a request its site's queue slot, so + * it asks for itself rather than wait. + */ + away?: Away; } /** What every request context carries, whichever shape the rest of it takes. */ @@ -59,6 +72,7 @@ export function baseContext(request: Request, deps: ContextDeps): BaseRequestCon ...(deps.signal ? { signal: deps.signal } : {}), timeZone: deps.timeZone, dayZone: deps.dayZone, + remember: remembering(deps), }; } @@ -76,6 +90,20 @@ function clock(deps: ContextDeps, assumed?: string): (time: string, day?: string }; } +/** + * A context's `remember`: the site's memory, and how this context waits on an + * ask another one started. + * + * The same on every context, whatever days it covers, since how long a value + * is worth keeping is the site's to say — a programme's details for a while + * after it ends, a mark that a channel has none for a week. The options are + * the memory's to check, which knows whose site it is and says so. + */ +function remembering(deps: ContextDeps): Remember { + return ((key: string, options: RememberOptions, task: () => unknown) => + deps.memory.remember(key, options, task, deps.away)) as Remember; +} + /** * The days a request covers, as a context that batches them says them. * @@ -179,6 +207,7 @@ export function parseContext( timeZone: deps.timeZone, dayZone: deps.dayZone, at: clock(deps, day), + remember: remembering(deps), programme: (start, title, options) => new ProgrammeBuilder({ channel: channel.xmltvId, diff --git a/src/grabber/remember.ts b/src/grabber/remember.ts new file mode 100644 index 0000000..f6ea35c --- /dev/null +++ b/src/grabber/remember.ts @@ -0,0 +1,289 @@ +/** + * What a site remembers between runs about one thing at a time — a + * programme's details, say — and for how long. + * + * Kept in a state group of its own rather than in the site's bag. A group is + * one blob, rewritten whole when anything in it changes: a few thousand + * programmes' details beside a token would rewrite the token's file every run + * for the details' sake. Each value carries when it is forgotten, and what has + * passed that is dropped at the end of every run, so the group holds about a + * window's worth however long a site has been running. + */ + +import type { Says } from '../core/events.js'; +import { unplain } from '../core/plain.js'; +import { StateKey, type SiteStateHandle, type TrackedMap } from './state.js'; + +/** + * One remembered value, and the instant it is forgotten. + * + * `missed` marks a `null` that a failure `miss` matched stands for — "this has + * no details" — which is an answer only to an ask that says such failures are + * answers. One that does not is asked again. + */ +interface Kept { + until: number; + value: unknown; + missed?: true; +} + +/** + * What one ask came to: an answer, or the failure it ended in. Shared by every + * ask of the key while it was on its way, each of which reads it with its own + * `miss`. + */ +type Outcome = { value: unknown } | { error: unknown; refused?: true }; + +/** + * How a caller waits on an ask somebody else started without holding a slot + * meanwhile — a parse's `localWork` slot, which the ask it is waiting on may + * need back before it can finish. + */ +export type Away = (wait: () => Promise) => Promise; + +/** One site's remembered values for the length of a run. */ +export class Memory { + readonly #handle: SiteStateHandle; + readonly #site: string; + readonly #now: number; + readonly #refresh: boolean; + readonly #warn: Says['warn'] | undefined; + #kept: Promise> | undefined; + /** Asks under way, so two of one key at once are one task. */ + readonly #asking = new Map>(); + /** Answered by this run — what `--refresh` still lets an ask read. */ + readonly #fresh = new Set(); + /** + * Answers given an `until` this run has already passed: good for the run, + * and kept by nobody after it. + */ + readonly #passing = new Map(); + #warnedPast = false; + + /** + * This site's memory, in the state its run holds, as of the run's `now`. + * + * Made up front and read the first time it is asked, so a site that never + * remembers anything reads nothing more than it did until the run ends — see + * {@link prune}. Under `refresh` nothing kept by an earlier run is read: as + * everywhere else `--refresh` reaches, the cache's reading is turned off and + * its writing is not. + */ + constructor( + handle: SiteStateHandle, + site: string, + now: Date, + options: { refresh?: boolean; warn?: Says['warn'] } = {}, + ) { + this.#handle = handle; + this.#site = site; + this.#now = now.getTime(); + this.#refresh = options.refresh === true; + this.#warn = options.warn; + } + + /** + * What the site kept under `key`, or the task's answer, kept until the + * options' `until` — a context's `remember`. + * + * `away` is how this caller waits on an ask another one started; a caller + * without one holds something it cannot let go of meanwhile — a request, its + * site's queue slot — and runs a task of its own instead of waiting. + */ + async remember( + key: string, + options: unknown, + task: () => unknown, + away?: Away, + ): Promise { + const { until, miss } = this.#read(key, options); + const kept = await this.#open(); + const held = + this.#passing.get(key) ?? + (this.#refresh && !this.#fresh.has(key) ? undefined : kept.get(key)); + + // A "none" is an answer only to an ask that takes such failures for one. + if (held !== undefined && (held.missed !== true || miss !== undefined)) { + return structuredClone(held.value); + } + + const asking = this.#asking.get(key); + let outcome: Outcome; + + if (asking !== undefined && away !== undefined) { + outcome = await away(() => asking); + } else { + const own = this.#ask(kept, key, until, miss, task); + + if (asking === undefined) { + this.#asking.set(key, own); + void own.then(() => { + if (this.#asking.get(key) === own) { + this.#asking.delete(key); + } + }); + } + + outcome = await own; + } + + if ('value' in outcome) { + return structuredClone(outcome.value); + } + + // Each caller's own `miss` on what the ask came to, so the answer is the + // one its types promise: a caller without one gets the failure, whoever + // asked first and however they took it. + if (outcome.refused !== true && miss?.(outcome.error) === true) { + return null; + } + + throw outcome.error; + } + + /** + * Drop what has passed its `until` — at the end of a run, whether or not + * anything asked, so a site that stopped remembering stops keeping too. + * + * A group that cannot be read has nothing to drop: what failed is a read, + * and the run it ends is not the place to say so. + */ + async prune(): Promise { + await this.#open().catch(() => undefined); + } + + /** The options, checked: an `until` that is a time, and a `miss` that is a function. */ + #read( + key: string, + options: unknown, + ): { until: number; miss: ((error: unknown) => boolean) | undefined } { + const { until, miss } = (typeof options === 'object' && options !== null ? options : {}) as { + until?: unknown; + miss?: unknown; + }; + const at = + typeof until === 'number' ? until : until instanceof Date ? until.getTime() : Number.NaN; + + if (!Number.isFinite(at)) { + throw new TypeError( + `Site "${this.#site}": remember(${JSON.stringify(key)}) needs an until, a Date or epoch ` + + `milliseconds (got ${until === undefined ? 'none' : JSON.stringify(until)})`, + ); + } + + if (miss !== undefined && typeof miss !== 'function') { + throw new TypeError( + `Site "${this.#site}": remember(${JSON.stringify(key)})'s miss must be a function`, + ); + } + + return { until: at, miss: miss as ((error: unknown) => boolean) | undefined }; + } + + /** + * The group, without what was to be forgotten by the run's `now`. + * + * Dropped as it is read, which does two things: the group is saved without + * them, and within a run everything held is good. A read that fails is not + * remembered, so the next ask tries again. + */ + #open(): Promise> { + this.#kept ??= this.#handle.bag(StateKey.REMEMBERED).then( + (kept) => { + for (const [key, entry] of kept) { + // A pair that is not one is forgotten with the rest: whatever wrote + // it, it is no answer to anything. + if (typeof entry?.until !== 'number' || !(entry.until > this.#now)) { + kept.delete(key); + } + } + + return kept; + }, + (error: unknown) => { + this.#kept = undefined; + throw error; + }, + ); + + return this.#kept; + } + + /** + * Ask, and keep the answer by the asker's own options — a failure its + * `miss` matches as `null`, marked as standing for one. What it hands back + * is the outcome itself, the failure included, for every caller to read with + * its own `miss`. + */ + async #ask( + kept: TrackedMap, + key: string, + until: number, + miss: ((error: unknown) => boolean) | undefined, + task: () => unknown, + ): Promise { + let value: unknown; + + try { + value = await task(); + } catch (error) { + // An answer that says "none" — a 404 for a programme with no details — + // is as much worth keeping as one with something in it. Anything else + // is a failure, kept by nobody, so the next run asks again. + if (miss?.(error) === true) { + this.#keep(kept, key, { until, value: null, missed: true }); + } + + return { error }; + } + + // Refused rather than kept as whatever JSON makes of it: this run would be + // handed a `Date` and the next one a string, and a site would find out + // only on the run after the one it was tried on. + const found = value === undefined ? undefined : unplain(value, 'value'); + + if (found !== undefined) { + return { + error: new TypeError( + `Site "${this.#site}": remember(${JSON.stringify(key)}) was answered with ${found}. ` + + `A remembered value must be plain — strings, finite numbers, booleans, null, ` + + `arrays and plain objects — since it is stored as JSON and read back by the next run`, + ), + refused: true, + }; + } + + // A copy, so what the task handed back can be changed by whoever was + // handed it without changing what is kept. + const entry: Kept = { until, value: structuredClone(value) }; + + this.#keep(kept, key, entry); + + return { value: entry.value }; + } + + #keep(kept: TrackedMap, key: string, entry: Kept): void { + if (entry.until > this.#now) { + kept.set(key, entry); + this.#fresh.add(key); + + return; + } + + // Already past: kept for this run, which is all it would have been good + // for, rather than written to be dropped by the next one. Said once, + // since an `until` in seconds where milliseconds were meant puts every + // value here, and a run asking everything again with nothing said would + // be a puzzle. + this.#passing.set(key, entry); + + if (!this.#warnedPast) { + this.#warnedPast = true; + this.#warn?.( + `remember(${JSON.stringify(key)}) was given an until already past ` + + `(${new Date(entry.until).toISOString()}), so it is kept for this run only — ` + + `an until in seconds rather than milliseconds?`, + ); + } + } +} diff --git a/src/grabber/site-run.ts b/src/grabber/site-run.ts index 47838b6..1e63ca2 100644 --- a/src/grabber/site-run.ts +++ b/src/grabber/site-run.ts @@ -29,6 +29,7 @@ import { resolveChannels } from './channels.js'; import { parseContext, requestContext, streamContext, type ContextDeps } from './context.js'; import { LocalSlot } from './local-slot.js'; import { planRequests, type Pair, type Request } from './planner.js'; +import { Memory } from './remember.js'; import { forget, isUnchanged, @@ -186,6 +187,8 @@ export class SiteRun { /** Filled in by {@link run}, since each of them has to be waited for. */ #siteState: SiteState = new Map(); + /** What the site remembers between runs — opened the first time it asks. */ + readonly #memory: Memory; #validators: TrackedMap = new TrackedMap(); #channels: GrabberChannel[] = []; @@ -222,6 +225,10 @@ export class SiteRun { }); this.#http = this.#session.pacing.http; this.#state = this.#session.state; + this.#memory = new Memory(this.#state, this.#site, run.now, { + refresh: this.#resolved.staleness.refetchAll, + warn: this.#says.warn, + }); /** * What this bounds is how many responses the site has in hand at once: @@ -301,7 +308,9 @@ export class SiteRun { // Beside `dispose`, and for the same reason: a site that threw, or was // cancelled part way, has as much to hand back as one that finished — the // channel list it fetched, and whatever its own code remembered. Only the - // groups that changed are written. + // groups that changed are written. What the site remembered is pruned + // first, so a site that no longer asks stops keeping what it kept. + await this.#memory.prune(); await this.#state.save(); // For a failed site too, where every count is nought — which is the truth, // and reads correctly beside the `site:failed` that says why. @@ -508,6 +517,7 @@ export class SiteRun { ...(signal ? { signal } : {}), timeZone: this.#run.timeZone, dayZone: this.#resolved.dayZone, + memory: this.#memory, }; } @@ -640,6 +650,7 @@ export class SiteRun { // the request being parsed did — ahead of the planned ones, so a // channel-day in hand is finished rather than joined by another. paced: (task) => slot.away(() => this.#sending(task, { priority: 1 })), + away: (wait) => slot.away(wait), }), ); @@ -901,6 +912,8 @@ export class SiteRun { // As a parse's is, and ahead of the planned ones for the same // reason: a pass already under way is finished rather than joined. paced: (task) => this.#sending(task, { priority: 1 }), + // A pass holds no slot while it runs, so it waits as it is. + away: (wait) => wait(), }), )) { // Between emissions, which is as often as this has anything to say: a diff --git a/src/grabber/state.ts b/src/grabber/state.ts index d14fb81..beb9690 100644 --- a/src/grabber/state.ts +++ b/src/grabber/state.ts @@ -31,6 +31,8 @@ export const StateKey = { VALIDATORS: 'validators', /** Whatever the site itself put there. */ SITE: 'state', + /** What a site's `remember` kept, each value with when it is forgotten. */ + REMEMBERED: 'remembered', } as const; /** @@ -448,7 +450,16 @@ export class SiteStateHandle { let pending = this.#groups.get(key); if (pending === undefined) { - pending = this.#store.getState(this.#site, key).then(make); + const reading = this.#store.getState(this.#site, key).then(make); + + // A read that failed is not remembered as the group: a store that was + // briefly out of descriptors is asked again by whoever asks next. + reading.catch(() => { + if (this.#groups.get(key) === reading) { + this.#groups.delete(key); + } + }); + pending = reading; this.#groups.set(key, pending); } diff --git a/src/grabber/types.ts b/src/grabber/types.ts index ec266ce..40a1a9e 100644 --- a/src/grabber/types.ts +++ b/src/grabber/types.ts @@ -169,6 +169,39 @@ interface SiteZones { dayZone: string; } +/** How long a remembered value is kept, and which failures are an answer. */ +export interface RememberOptions { + /** + * When it is forgotten: a `Date`, or epoch milliseconds. For a programme's + * details, a while after it ends — once its day has left the window nothing + * asks for them again. + */ + until: Date | number; + /** + * A failure that is an answer — a 404 for a programme with no details — kept + * as `null`, so the next run does not ask again. Any other is thrown and kept + * by nobody, and the next run does ask. + */ + miss?: (error: unknown) => boolean; +} + +/** + * What a site kept under `key` on an earlier run, or `task`'s answer, kept for + * the next — see {@link ParseContext.remember}. + * + * `null` is in the answer exactly when the ask says some failure is one — has + * a `miss`, or options typed as possibly having one. A "none" kept for an ask + * with a `miss` is never handed to one without: it asks for itself. + */ +export interface Remember { + ( + key: string, + options: RememberOptions & { miss?: undefined }, + task: () => T | Promise, + ): Promise; + (key: string, options: RememberOptions, task: () => T | Promise): Promise; +} + /** * What every context carries, whichever shape the rest of it takes: the * channel-days it is for, the site's client, its state, and the run's signal. @@ -196,6 +229,8 @@ export interface BaseRequestContext extends Says, SiteZones { signal?: AbortSignal; /** What this site remembers between runs — see {@link SiteState}. */ state: SiteState; + /** Something kept between runs, with a time to forget it — see {@link ParseContext.remember}. */ + remember: Remember; } /** @@ -510,6 +545,30 @@ export interface ParseContext extends Says, SiteZones { * morning. Anything else throws `XmltvDateError`. */ at(time: string, day?: string): Date; + /** + * What the site kept under `key` on an earlier run, or `task`'s answer, kept + * for the next — a programme's details, fetched once rather than every time + * its day is fetched again. + * + * ```ts + * const details = await remember( + * `detail:${item.id}|${item.start}`, + * { + * until: new Date(item.stop).getTime() + 86_400_000, + * miss: (error) => error instanceof HTTPError && error.response.status === 404, + * }, + * async () => trim(await paced(({ signal }) => http.get(`detail/${item.id}`, { signal }).json())), + * ); + * ``` + * + * Kept until `until`, and only an answer is kept: a failure `miss` does not + * match is thrown, and the next run asks again. Two asks of one key at once + * are one task, each reading what it came to with its own `miss`, and each + * handed a copy. What is kept must be plain — it is stored as JSON — so keep only what + * the parse uses: the site's memory is one group, written whole when it + * changes. + */ + remember: Remember; /** * What {@link SiteConfig.request} handed back, whole and unchanged. A request * covering several channel-days is parsed once for each of them, and every diff --git a/test/grabber.test.ts b/test/grabber.test.ts index e0b4f0d..8e9ec8a 100644 --- a/test/grabber.test.ts +++ b/test/grabber.test.ts @@ -2,7 +2,7 @@ import { createServer } from 'node:http'; import type { IncomingMessage, ServerResponse } from 'node:http'; import type { AddressInfo } from 'node:net'; import { HTTPError, NetworkError } from 'ky'; -import { describe, expect, it, vi } from 'vitest'; +import { describe, expect, expectTypeOf, it, vi } from 'vitest'; import type { CacheEntryMeta, CacheStore, @@ -10,6 +10,7 @@ import type { StateEntry, StoredStateMeta, } from '../src/cache/types.js'; +import { CacheManager, MemoryCacheDriver } from '../src/cache/main.js'; import type { XmltvProgramme } from '../src/xmltv/types.js'; import { fellShort, @@ -30,6 +31,8 @@ import type { ChannelsContext, GrabberChannel, GrabSummary, + Remember, + RememberOptions, SiteConfig, StreamedChannelDay, StreamSiteConfig, @@ -1382,9 +1385,10 @@ describe('grab', () => { await grab([config], { cache, now: NOW }); // A site that never touches its bag costs no write — the group is read - // once per site run and left as it was. + // once per site run and left as it was. So is what it remembers, read at + // the end to drop what has passed, and finding nothing to drop. expect(cache.stateWrites).toEqual([]); - expect(cache.stateReads).toEqual(['example.com|state']); + expect(cache.stateReads).toEqual(['example.com|state', 'example.com|remembered']); }); it('is empty for a store that remembers nothing, and still works', async () => { @@ -3694,6 +3698,424 @@ describe('a stream that says nothing has changed', () => { }); }); +describe('what a site remembers between runs', () => { + const LATER = new Date(NOW.getTime() + 3_600_000); + const UNTIL = NOW.getTime() + 86_400_000; + + /** A site asking for one programme's details through `remember`, saying how often it asked. */ + function detailed( + asked: string[], + answer: () => unknown = () => ({ desc: 'Details' }), + options: Partial = {}, + ): SiteConfig { + return makeConfig({ + async parseDay({ day, remember }) { + const details = await remember(`detail:${day}`, { until: UNTIL, ...options }, () => { + asked.push(day); + + return answer(); + }); + + return [ + { + ...programme(`${day}T06:00:00.000Z`), + ...(details === null ? {} : { desc: [{ value: JSON.stringify(details) }] }), + }, + ]; + }, + }); + } + + it('asks once, and the next run is handed what it kept', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + + await grab([detailed(asked)], { cache, now: NOW }); + // Today is fetched again every run, and parsed again with it. + await grab([detailed(asked)], { cache, now: LATER }); + + expect(asked).toEqual([TODAY]); + expect( + cache.get({ site: 'example.com', channelId: 'one.example', day: TODAY })?.programmes, + ).toMatchObject([{ desc: [{ value: '{"desc":"Details"}' }] }]); + }); + + it('keeps it in a group of its own, leaving the site’s state alone', async () => { + const cache = new MemoryCache(); + + await grab([detailed([])], { cache, now: NOW }); + + expect(cache.stateWrites).toEqual(['example.com|remembered']); + expect(cache.state.get('example.com|remembered')?.data).toEqual([ + [`detail:${TODAY}`, { until: UNTIL, value: { desc: 'Details' } }], + ]); + }); + + it('keeps a failure `miss` matches as null, and asks again after one it does not', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + const missing = Object.assign(new Error('404'), { status: 404 }); + const miss = (error: unknown) => (error as { status?: number }).status === 404; + + await grab([detailed(asked, () => Promise.reject(missing), { miss })], { cache, now: NOW }); + await grab([detailed(asked, () => Promise.reject(missing), { miss })], { cache, now: LATER }); + + // Answered "none" once, and remembered as that. + expect(asked).toEqual([TODAY]); + + const failing = new MemoryCache(); + const tried: string[] = []; + const down = () => Promise.reject(new Error('the source is down')); + + const first = await grab([detailed(tried, down, { miss })], { cache: failing, now: NOW }); + await grab([detailed(tried, down, { miss })], { cache: failing, now: LATER }); + + expect(first.failed).toBe(1); + expect(tried).toEqual([TODAY, TODAY]); + }); + + it('forgets a value once its until has passed', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + const until = NOW.getTime() + 60_000; + + await grab([detailed(asked, undefined, { until })], { cache, now: NOW }); + await grab([detailed(asked, undefined, { until })], { cache, now: LATER }); + + expect(asked).toEqual([TODAY, TODAY]); + }); + + it('drops what has passed from what it stores, though nothing is asked', async () => { + const cache = new MemoryCache(); + + cache.seedState('example.com', 'remembered', [ + ['old', { until: NOW.getTime() - 1, value: 'gone' }], + ['current', { until: UNTIL, value: 'kept' }], + ]); + + // A site that no longer remembers anything. + await grab([makeConfig()], { cache, now: NOW }); + + expect(cache.state.get('example.com|remembered')?.data).toEqual([ + ['current', { until: UNTIL, value: 'kept' }], + ]); + }); + + it('asks once for two asks of one key at the same time', async () => { + const cache = new MemoryCache(); + let asked = 0; + const answers: unknown[] = []; + + await grab( + [ + makeConfig({ + async parseDay({ day, remember }) { + // A broadcast past midnight is in both days' listings, and both + // ask about it while the first answer is still on its way. + const ask = () => + remember('shared', { until: UNTIL }, async () => { + asked++; + await new Promise((resolve) => setTimeout(resolve, 20)); + + return 'once'; + }); + + answers.push(...(await Promise.all([ask(), ask()]))); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW }, + ); + + expect(asked).toBe(1); + expect(answers).toEqual(['once', 'once']); + }); + + it('refuses a value JSON would bring back as something else', async () => { + const cache = new MemoryCache(); + const report = collect(); + + await grab([detailed([], () => new Date(NOW))], { cache, now: NOW, reporter: report.reporter }); + + expect(errorMessage(report.of('entry:failed')[0]?.error)).toContain( + 'remember("detail:2026-07-17") was answered with an instance of Date at value', + ); + }); + + it('refuses to keep anything without an until', async () => { + const cache = new MemoryCache(); + const report = collect(); + + await grab([detailed([], undefined, { until: undefined as unknown as number })], { + cache, + now: NOW, + reporter: report.reporter, + }); + + expect(errorMessage(report.of('entry:failed')[0]?.error)).toContain( + 'remember("detail:2026-07-17") needs an until', + ); + }); + + it('keeps what it was answered, not what a caller made of it afterwards', async () => { + // A store that keeps state as JSON, as every real one does. + const cache = new CacheManager({ driver: new MemoryCacheDriver() }); + const seen: unknown[] = []; + const site = () => + makeConfig({ + async parseDay({ day, remember }) { + // An answer the site holds on to, and a copy it was handed: changing + // either afterwards changes neither what is kept nor the next answer. + const answer = { cast: ['Ann'], year: 2001 }; + const ask = () => remember('details', { until: UNTIL }, () => answer); + const first = await ask(); + + answer.year = 1999; + first.cast.push('changed by the parse'); + seen.push(await ask()); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }); + + await grab([site()], { cache, now: NOW }); + await grab([site()], { cache, now: LATER }); + + expect(seen).toEqual([ + { cast: ['Ann'], year: 2001 }, + { cast: ['Ann'], year: 2001 }, + ]); + }); + + it('reads nothing kept by an earlier run under --refresh, and still keeps', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + const site = () => + makeConfig({ + async parseDay({ day, remember }) { + for (const _ of [1, 2]) { + await remember('details', { until: UNTIL }, () => { + asked.push(day); + + return 'fresh'; + }); + } + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }); + + await grab([site()], { cache, now: NOW }); + await grab([site()], { cache, now: LATER, staleness: { refetchAll: true } }); + + // Asked again under --refresh, once: what this run asked it may read. + expect(asked).toEqual([TODAY, TODAY]); + }); + + it('hands a kept "none" only to an ask that takes a failure for one', async () => { + const cache = new MemoryCache(); + const missing = Object.assign(new Error('404'), { status: 404 }); + const miss = (error: unknown) => (error as { status?: number }).status === 404; + let asked = 0; + const answers: unknown[] = []; + + await grab( + [ + makeConfig({ + async parseDay({ day, remember }) { + const task = () => { + asked++; + + return Promise.reject(missing); + }; + + // Both at once: the second joins the first, and reads its failure + // with its own options. + const both = await Promise.allSettled([ + remember('details', { until: UNTIL, miss }, task), + remember('details', { until: UNTIL }, task), + ]); + + answers.push(...both.map((one) => (one.status === 'fulfilled' ? one.value : 'thrown'))); + // And after: kept as "none", which this ask does not take for an answer. + answers.push(await remember('details', { until: UNTIL }, () => 'its own')); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW }, + ); + + expect(answers).toEqual([null, 'thrown', 'its own']); + expect(asked).toBe(1); + }); + + it('keeps a value whose until has already passed for this run only, and says so', async () => { + const cache = new MemoryCache(); + const report = collect(); + let asked = 0; + + await grab( + [ + makeConfig({ + async parseDay({ day, remember }) { + for (const key of ['one', 'one', 'two']) { + // In seconds, where milliseconds were meant. + await remember(key, { until: UNTIL / 1000 }, () => ++asked); + } + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW, reporter: report.reporter }, + ); + + expect(asked).toBe(2); + expect(cache.stateWrites).toEqual([]); + expect( + report.of('site:warning').filter(({ message }) => message.includes('already past')), + ).toHaveLength(1); + }); + + it('names the site and the key when the options are not ones', async () => { + const cache = new MemoryCache(); + const report = collect(); + + await grab( + [ + makeConfig({ + channels: [channel('one.example'), channel('two.example')], + async parseDay({ channel: { xmltvId }, day, remember }) { + await (xmltvId === 'one.example' + ? remember('a', { until: '2026-07-18' as unknown as number }, () => 1) + : ( + remember as (key: string, options: unknown, task: () => number) => Promise + )('b', undefined, () => 1)); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW, reporter: report.reporter }, + ); + + expect(report.of('entry:failed').map(({ error }) => errorMessage(error))).toEqual([ + 'Site "example.com": remember("a") needs an until, a Date or epoch milliseconds (got "2026-07-18")', + 'Site "example.com": remember("b") needs an until, a Date or epoch milliseconds (got none)', + ]); + }); + + it('asks again after a read of what it kept failed, and finishes the run', async () => { + const cache = new MemoryCache(); + const report = collect(); + const read = cache.getState.bind(cache); + let failures = 1; + + cache.getState = async (site, key) => { + if (key === 'remembered' && failures-- > 0) { + throw Object.assign(new Error('EMFILE: too many open files'), { code: 'EMFILE' }); + } + + return read(site, key); + }; + + const summary = await grab( + [ + makeConfig({ + channels: [channel('one.example'), channel('two.example')], + async parseDay({ day, remember }) { + await remember('details', { until: UNTIL }, () => 'answer'); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW, reporter: report.reporter, localConcurrency: 1 }, + ); + + // The first channel-day met the failed read; the second asked again. + expect(summary).toMatchObject({ fetched: 1, failed: 1, sitesFailed: 0 }); + expect(report.of('site:done')).toHaveLength(1); + expect(report.of('site:failed')).toHaveLength(0); + }); + + it('lets a parse waiting on another parse’s ask give up its slot meanwhile', async () => { + // One local slot, and one request covering two days whose listings both + // name a broadcast past midnight: the second day's parse waits on the + // first's ask, whose detail request needs the slot back to finish. + const cache = new MemoryCache(); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), 5000); + let asked = 0; + + try { + const summary = await grab( + [ + makeDaysConfig({ + days: 2, + async parseDay({ day, remember, paced }) { + await remember('past midnight', { until: UNTIL }, () => + paced(async () => { + asked++; + await new Promise((resolve) => setTimeout(resolve, 20)); + + return 'details'; + }), + ); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW, localConcurrency: 1, signal: controller.signal }, + ); + + expect(controller.signal.aborted).toBe(false); + expect(summary.fetched).toBe(2); + expect(asked).toBe(1); + } finally { + clearTimeout(timer); + } + }); + + it('types an answer as possibly null only where a failure may be one', () => { + // Typechecked, never run: what each way of asking is promised. + const typed = (remember: Remember, held: RememberOptions) => { + expectTypeOf(remember('k', { until: 1 }, () => 1)).toEqualTypeOf>(); + expectTypeOf(remember('k', { until: 1, miss: () => true }, () => 1)).toEqualTypeOf< + Promise + >(); + expectTypeOf(remember('k', held, () => 1)).toEqualTypeOf>(); + }; + + expect(typed).toBeTypeOf('function'); + }); + + it('is there for a request too', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + const site = () => + makeConfig({ + async request({ day, remember }) { + return remember(`token:${day}`, { until: UNTIL }, () => { + asked.push(day); + + return 'token'; + }); + }, + }); + + await grab([site()], { cache, now: NOW }); + await grab([site()], { cache, now: LATER }); + + expect(asked).toEqual([TODAY]); + }); +}); + describe('what a run reports', () => { const types = (events: EpgEvent[]): string[] => events.map((event) => event.type); diff --git a/test/readme-examples.ts b/test/readme-examples.ts index d98c03d..c0a0282 100644 --- a/test/readme-examples.ts +++ b/test/readme-examples.ts @@ -508,6 +508,47 @@ export const remembering = defineSiteConfig({ parseDay: () => [], }); +// --- docs/site-config.md: A programme's details, asked once ----------------- +interface Detail { + description: string; + cast: string[]; +} + +export const askedOnce = defineSiteConfig<{ + items: { id: string; start: string; end: string; title: string }[]; +}>({ + site: 'example.tv', + channels: [], + async request({ channel, day, http }) { + return http.get(`epg/${channel.siteId}/${day}`).json(); + }, + async parseDay({ payload, http, paced, remember, programme }) { + return Promise.all( + payload.items.map(async (item) => { + const details = await remember( + `detail:${item.id}|${item.start}`, + { + // Asked about while its day can be fetched again, and no longer. + until: Date.parse(item.end) + 86_400_000, + // "This one has no details" is an answer, and worth keeping. + miss: (error) => error instanceof HTTPError && error.response.status === 404, + }, + async () => { + const full = await paced(({ signal }) => + http.get(`detail/${item.id}`, { signal }).json(), + ); + + return { desc: full.description, cast: full.cast }; // only what the parse uses + }, + ); + + const p = programme(new Date(item.start), item.title); + return details ? p.desc(details.desc) : p; + }), + ); + }, +}); + // The mode decides which caps are accepted, and the shape of the context. export const wrongCap = defineSiteConfig({ site: 'example.tv',