import type { Result } from "../../application/result.ts"; import type { CursorPage, CursorPaginationProfile, CursorPaginationRuntime, } from "../../contracts/cursor-pagination.ts"; import { createFailure } from "../../contracts/errors.ts"; const ABORTED = Symbol("PAGINATION_ABORTED"); /** * Resolves as soon as the operation settles or the signal aborts, whichever * comes first. A late operation result is observed and discarded, never thrown * as an unhandled rejection. * * OPT-NET-01. A loader rejection is *not* an abort. The presence of a signal * says nothing about why the loader failed, so a rejection is re-thrown exactly * as it would be with no signal at all; only a signal that has actually * aborted classifies the outcome as cancellation. */ async function raceAbort( operation: Promise, signal: AbortSignal | undefined, ): Promise { operation.catch(() => {}); if (!signal) return await operation; if (signal.aborted) return ABORTED; return await new Promise((resolve, reject) => { const onAbort = () => resolve(ABORTED); signal.addEventListener("abort", onAbort, { once: true }); operation.then( (value) => { signal.removeEventListener("abort", onAbort); resolve(value); }, (reason: unknown) => { signal.removeEventListener("abort", onAbort); if (signal.aborted) { resolve(ABORTED); return; } reject(reason); }, ); }); } export function createCursorPaginationRuntime(dependencies: Readonly<{ definitionId: string; profile: CursorPaginationProfile; loadPage( cursor: string | null, context: Readonly<{ signal?: AbortSignal }>, ): Promise>>; }>): CursorPaginationRuntime { validateProfile(dependencies.profile); return Object.freeze({ async loadAll(context) { const items: Value[] = []; const cursors = new Set(); let cursor: string | null = null; let snapshot: string | null | undefined; for ( let pageIndex = 0; pageIndex < dependencies.profile.maxPages; pageIndex += 1 ) { if (context.signal?.aborted) { return failure("REQUEST_ABORTED", "PAGINATION_ABORTED"); } // N-10. A non-cooperative loader may never settle, or may settle after // abort. Race the signal so `loadAll` is bounded, and re-check before // observing the page so a late completion is ignored rather than // accumulated into a successful result. const raced: Result> | typeof ABORTED = await raceAbort>>( dependencies.loadPage(cursor, context), context.signal, ); if (raced === ABORTED || context.signal?.aborted) { return failure("REQUEST_ABORTED", "PAGINATION_ABORTED"); } const result: Result> = raced; if (!result.ok) return result; const page: CursorPage = result.value; if (!isValidPage(page, dependencies.profile)) { return failure( "PAGINATION_CONTRACT_VIOLATION", "PAGINATION_PAGE_INVALID", ); } if (snapshot === undefined) { snapshot = page.snapshotToken; } else if (snapshot !== page.snapshotToken) { return failure( "PAGINATION_CONTRACT_VIOLATION", "PAGINATION_SNAPSHOT_CHANGED", ); } items.push(...page.items); if ( items.length > dependencies.profile.maxTotalItems || estimatedBytes(items) > dependencies.profile.maxEstimatedBytes ) { return failure( "RESULT_LIMIT_EXCEEDED", "PAGINATION_RESULT_LIMIT", ); } if (!page.hasMore) return { ok: true, value: Object.freeze(items) }; const nextCursor: string | null = page.nextCursor; if (!nextCursor || cursors.has(nextCursor)) { return failure( "PAGINATION_CONTRACT_VIOLATION", "PAGINATION_CURSOR_LOOP", ); } cursors.add(nextCursor); cursor = nextCursor; } return failure( "RESULT_LIMIT_EXCEEDED", "PAGINATION_PAGE_LIMIT", ); }, }); function failure( kind: | "PAGINATION_CONTRACT_VIOLATION" | "RESULT_LIMIT_EXCEEDED" | "REQUEST_ABORTED", code: string, ) { return { ok: false as const, error: createFailure(kind, dependencies.definitionId, 0, { code }), }; } } function validateProfile(profile: CursorPaginationProfile): void { if ( !profile.profileId || !Number.isSafeInteger(profile.maxPages) || profile.maxPages < 1 || profile.maxPages > 100 || !Number.isSafeInteger(profile.maxTotalItems) || profile.maxTotalItems < 1 || !Number.isSafeInteger(profile.maxEstimatedBytes) || profile.maxEstimatedBytes < 1 || !Number.isSafeInteger(profile.maxCursorBytes) || profile.maxCursorBytes < 1 || profile.maxCursorBytes > 4_096 ) { throw new TypeError("Invalid cursor pagination profile."); } } function isValidPage( page: CursorPage, profile: CursorPaginationProfile, ): boolean { const encoder = new TextEncoder(); return ( Boolean(page) && Array.isArray(page.items) && typeof page.hasMore === "boolean" && page.hasMore === (page.nextCursor !== null) && (page.nextCursor === null || (typeof page.nextCursor === "string" && page.nextCursor.length > 0 && encoder.encode(page.nextCursor).byteLength <= profile.maxCursorBytes)) && (page.snapshotToken === null || (typeof page.snapshotToken === "string" && page.snapshotToken.length > 0 && encoder.encode(page.snapshotToken).byteLength <= profile.maxCursorBytes)) && (profile.allowSparsePage || !page.hasMore || page.items.length > 0) ); } function estimatedBytes(value: unknown): number { try { return new TextEncoder().encode(JSON.stringify(value)).byteLength; } catch { return Number.POSITIVE_INFINITY; } }