142 lines
13 KiB
TypeScript
142 lines
13 KiB
TypeScript
import type { Id } from '../canvas-core/types.ts';
|
|
import { equal, fault, freeze } from '../canvas-core/values.ts';
|
|
|
|
export type LoadScope = {
|
|
readonly accountId: Id; readonly workspaceId: Id; readonly projectId: Id | null; readonly canvasId: Id;
|
|
readonly viewKey: string; readonly format: string; readonly queryFingerprint: string; readonly source: string;
|
|
readonly authEpoch: number; readonly navigationGeneration: number;
|
|
};
|
|
export type LoadChunk<T> = {
|
|
readonly queryKey: string; readonly projectId: Id; readonly snapshotToken: string;
|
|
readonly serverVersions: Readonly<Record<Id, number>>; readonly permissionEpoch: number;
|
|
readonly sequence: number; readonly coverage: readonly string[]; readonly complete: boolean;
|
|
readonly byteLength: number; readonly value: T;
|
|
};
|
|
export type LoadSnapshot<T> = {
|
|
readonly scope: LoadScope; readonly generation: number;
|
|
readonly cacheStatus: 'bypassed' | 'miss' | 'invalid' | 'hit';
|
|
readonly status: 'queued' | 'loading' | 'initial-ready' | 'partial' | 'complete' | 'failed' | 'failed-with-partial' | 'cancelled';
|
|
readonly chunks: readonly LoadChunk<T>[]; readonly coverage: readonly string[]; readonly byteLength: number;
|
|
readonly error?: { readonly code: string; readonly message: string };
|
|
};
|
|
export type LoadPort<T> = (scope: LoadScope, signal: AbortSignal) => AsyncIterable<LoadChunk<T>>;
|
|
export type LoadHandle<T> = { readonly ready: Promise<LoadSnapshot<T>>; getSnapshot(): LoadSnapshot<T>; release(): void };
|
|
|
|
/** Navigation belongs to a subscriber, not the shared query identity. */
|
|
export function loadQueryKey(scope: LoadScope): string {
|
|
if (!scope || ![scope.accountId, scope.workspaceId, scope.canvasId, scope.viewKey, scope.format, scope.queryFingerprint, scope.source].every(value => typeof value === 'string' && value.length > 0)
|
|
|| scope.projectId !== null && (typeof scope.projectId !== 'string' || !scope.projectId)
|
|
|| ![scope.authEpoch, scope.navigationGeneration].every(value => Number.isSafeInteger(value) && value >= 0)) throw fault('invalid_load_scope');
|
|
return JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId, scope.canvasId, scope.viewKey, scope.format, scope.queryFingerprint, scope.source, scope.authEpoch]);
|
|
}
|
|
|
|
/** Read copies only. This coordinator never publishes a document or a server ACK.
|
|
* Adapters validate payloads and declare their measured byte cost before yielding.
|
|
*/
|
|
export function createScopedLoader<T>(options: {
|
|
load: LoadPort<T>; current(scope: LoadScope): boolean; concurrency?: number; maxQueued?: number;
|
|
maxResponseBytes?: number; maxChunks?: number; maxCacheEntries?: number; maxCacheBytes?: number;
|
|
cacheMaxAgeMs?: number; now?: () => number;
|
|
acceptCached?(scope: LoadScope, snapshot: LoadSnapshot<T>): boolean;
|
|
onObserverError?(error: unknown): void;
|
|
}) {
|
|
type Subscriber = { readonly scope: LoadScope; active: boolean; cancelled?: LoadSnapshot<T>; lastState?: LoadSnapshot<T>; snapshot?: LoadSnapshot<T>;resolve(value: LoadSnapshot<T>): void; reject(error: unknown): void; cleanup(): void; notify?(snapshot: LoadSnapshot<T>): void };
|
|
type Job = { readonly key: string; readonly scope: LoadScope; readonly generation: number; readonly controller: AbortController; readonly subscribers: Set<Subscriber>; state: LoadSnapshot<T>; retired: boolean };
|
|
const concurrency = options.concurrency ?? 4, maxQueued = options.maxQueued ?? 128, maxBytes = options.maxResponseBytes ?? 64 * 1024 * 1024, maxChunks = options.maxChunks ?? 4096;
|
|
const maxCacheEntries = options.maxCacheEntries ?? 16, maxCacheBytes = options.maxCacheBytes ?? 16 * 1024 * 1024, cacheMaxAge = options.cacheMaxAgeMs ?? 0;
|
|
for (const [name, value] of Object.entries({ concurrency, maxQueued, maxBytes, maxChunks, maxCacheEntries, maxCacheBytes, cacheMaxAge })) if (!Number.isSafeInteger(value) || value < (['maxCacheEntries', 'maxCacheBytes', 'cacheMaxAge'].includes(name) ? 0 : 1)) throw fault('invalid_loader_budget');
|
|
const inflight = new Map<string, Job>(), queue = new Map<number, Job>(), cache = new Map<string, { state: LoadSnapshot<T>; at: number }>();
|
|
let active = 0, generation = 0, closed = false, cacheBytes = 0;
|
|
const now = () => { const value = (options.now ?? Date.now)(); if (!Number.isSafeInteger(value) || value < 0) throw fault('invalid_loader_clock'); return value; };
|
|
const current = (scope: LoadScope) => { if (closed) throw fault('loader_closed'); if (!options.current(scope)) throw fault('identity_changed'); };
|
|
const view = (job: Job, subscriber: Subscriber): LoadSnapshot<T> => {if(subscriber.lastState!==job.state){subscriber.lastState=job.state;subscriber.snapshot=Object.freeze({...job.state,scope:subscriber.scope});}return subscriber.snapshot!;};
|
|
const notifyOne = (job: Job, subscriber: Subscriber) => { if (subscriber.active) { try { subscriber.notify?.(view(job, subscriber)); } catch (error) { try { options.onObserverError?.(error); } catch { /* Observers cannot change scheduling. */ } } } };
|
|
const notify = (job: Job) => { for (const subscriber of [...job.subscribers]) notifyOne(job, subscriber); };
|
|
const removeCache = (key: string) => { const entry = cache.get(key); if (entry) { cacheBytes -= entry.state.byteLength; cache.delete(key); } };
|
|
function retire(job: Job) {
|
|
if (job.retired) return; job.retired = true; job.controller.abort(); queue.delete(job.generation);
|
|
if (inflight.get(job.key) === job) inflight.delete(job.key);
|
|
}
|
|
function pump() {
|
|
while (!closed && active < concurrency && queue.size) {
|
|
const job = queue.values().next().value!; queue.delete(job.generation);
|
|
if (job.retired || !job.subscribers.size) continue;
|
|
active++; job.state = Object.freeze({ ...job.state, status: 'loading' }); notify(job);
|
|
void (async () => {
|
|
try {
|
|
current(job.scope); let terminal = false, projectId: Id | undefined, token: string | undefined, versions: Readonly<Record<Id, number>> | undefined;
|
|
for await (const chunk of options.load(job.scope, job.controller.signal)) {
|
|
if (job.retired) return; current(job.scope);
|
|
if (!chunk || terminal || chunk.queryKey !== job.key || chunk.permissionEpoch !== job.scope.authEpoch || typeof chunk.projectId !== 'string' || !chunk.projectId
|
|
|| job.scope.projectId !== null && chunk.projectId !== job.scope.projectId || typeof chunk.snapshotToken !== 'string' || !chunk.snapshotToken
|
|
|| chunk.sequence !== job.state.chunks.length || typeof chunk.complete !== 'boolean' || !Number.isSafeInteger(chunk.byteLength) || chunk.byteLength < 0
|
|
|| !Array.isArray(chunk.coverage) || chunk.coverage.some(key => typeof key !== 'string' || !key) || new Set(chunk.coverage).size !== chunk.coverage.length
|
|
|| !chunk.serverVersions || typeof chunk.serverVersions !== 'object' || Array.isArray(chunk.serverVersions)
|
|
|| ![Object.prototype,null].includes(Object.getPrototypeOf(chunk.serverVersions))
|
|
|| Object.entries(chunk.serverVersions).some(([id, version]) => !id || !Number.isSafeInteger(version) || version < 1)) throw fault('invalid_load_chunk');
|
|
const orderedVersions = Object.fromEntries(Object.entries(chunk.serverVersions).sort(([a], [b]) => a.localeCompare(b)));
|
|
if (projectId !== undefined && (chunk.projectId !== projectId || chunk.snapshotToken !== token || !equal(orderedVersions, versions))) throw fault('load_snapshot_changed');
|
|
if (job.state.chunks.length >= maxChunks || chunk.byteLength > maxBytes - job.state.byteLength) throw fault('load_response_budget');
|
|
projectId = chunk.projectId; token = chunk.snapshotToken; versions = orderedVersions; terminal = chunk.complete;
|
|
// Own metadata; the adapter supplies an immutable/owned payload value.
|
|
const owned = Object.freeze({ ...chunk, serverVersions: freeze(orderedVersions), coverage: freeze([...chunk.coverage]) });
|
|
job.state = Object.freeze({ ...job.state, status: job.state.chunks.length ? 'partial' : 'initial-ready', chunks: Object.freeze([...job.state.chunks, owned]),
|
|
coverage: freeze([...new Set([...job.state.coverage, ...chunk.coverage])]), byteLength: job.state.byteLength + chunk.byteLength }); notify(job);
|
|
}
|
|
if (job.retired) return; current(job.scope); if (!terminal) throw fault('incomplete_load');
|
|
job.state = Object.freeze({ ...job.state, status: 'complete' });
|
|
if (maxCacheEntries && maxCacheBytes && cacheMaxAge && job.state.byteLength <= maxCacheBytes) {
|
|
removeCache(job.key); cache.set(job.key, { state: job.state, at: now() }); cacheBytes += job.state.byteLength;
|
|
while (cache.size > maxCacheEntries || cacheBytes > maxCacheBytes) removeCache(cache.keys().next().value!);
|
|
}
|
|
notify(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cleanup(); subscriber.resolve(view(job, subscriber)); }
|
|
} catch (error) {
|
|
if (job.retired) return;
|
|
const failure = error as { code?: string; message?: string };
|
|
job.state = Object.freeze({ ...job.state, status: job.state.chunks.length ? 'failed-with-partial' : 'failed', error: freeze({ code: typeof failure?.code === 'string' ? failure.code : 'load_failed', message: typeof failure?.message === 'string' ? failure.message : 'Load failed' }) });
|
|
notify(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cleanup(); subscriber.reject(error); }
|
|
} finally {
|
|
if (inflight.get(job.key) === job) inflight.delete(job.key);
|
|
active--; pump();
|
|
}
|
|
})();
|
|
}
|
|
}
|
|
return Object.freeze({
|
|
acquire(scopeInput: LoadScope, settings: { signal?: AbortSignal; onUpdate?: (snapshot: LoadSnapshot<T>) => void; cache?: 'fresh' | 'allow-validated' } = {}): LoadHandle<T> {
|
|
if(!scopeInput)throw fault('invalid_load_scope');
|
|
const scope:LoadScope=Object.freeze({accountId:scopeInput.accountId,workspaceId:scopeInput.workspaceId,projectId:scopeInput.projectId,canvasId:scopeInput.canvasId,viewKey:scopeInput.viewKey,format:scopeInput.format,queryFingerprint:scopeInput.queryFingerprint,source:scopeInput.source,authEpoch:scopeInput.authEpoch,navigationGeneration:scopeInput.navigationGeneration});
|
|
const key = loadQueryKey(scope); current(scope); if (settings.signal?.aborted) throw fault('load_aborted');
|
|
let cacheStatus: LoadSnapshot<T>['cacheStatus'] = settings.cache === 'allow-validated' ? 'miss' : 'bypassed';
|
|
let cached = cache.get(key);
|
|
if (cached) {
|
|
const age = now() - cached.at;
|
|
if (age < 0 || age > cacheMaxAge || settings.cache !== 'allow-validated' || !options.acceptCached || options.acceptCached(scope, cached.state) !== true) { if(settings.cache==='allow-validated')cacheStatus='invalid';removeCache(key); cached = undefined; }
|
|
else { cacheStatus='hit';cache.delete(key); cache.set(key, cached); }
|
|
}
|
|
current(scope);
|
|
if(settings.signal?.aborted)throw fault('load_aborted');
|
|
let job = inflight.get(key);
|
|
if (!job) {
|
|
if (!cached && queue.size >= maxQueued + Math.max(0, concurrency - active)) throw fault('load_capacity');
|
|
if (generation >= Number.MAX_SAFE_INTEGER) throw fault('load_generation_exhausted');
|
|
job = { key, scope, generation: ++generation, controller: new AbortController(), subscribers: new Set(), retired: false,
|
|
state: cached ? Object.freeze({ ...cached.state, scope, generation,cacheStatus }) : Object.freeze({ scope, generation,cacheStatus,status: 'queued', chunks: Object.freeze([]), coverage: Object.freeze([]), byteLength: 0 }) };
|
|
if (!cached) { inflight.set(key, job); queue.set(job.generation, job); }
|
|
}
|
|
const shared = job; let subscriber: Subscriber;
|
|
const ready = new Promise<LoadSnapshot<T>>((resolve, reject) => { subscriber = { scope, active: true, resolve, reject, notify: settings.onUpdate, cleanup: () => settings.signal?.removeEventListener('abort', release) }; });
|
|
const release = () => { if (!subscriber.active) return; subscriber.cancelled=Object.freeze({...view(shared,subscriber),status:'cancelled'});subscriber.active = false; subscriber.cleanup(); shared.subscribers.delete(subscriber); subscriber.reject(fault('load_aborted')); if (!shared.subscribers.size) retire(shared); };
|
|
shared.subscribers.add(subscriber!); settings.signal?.addEventListener('abort', release, { once: true });
|
|
if (cached) { subscriber!.resolve(view(shared, subscriber!)); notifyOne(shared, subscriber!); } else { notifyOne(shared, subscriber!); pump(); }
|
|
return Object.freeze({ ready, getSnapshot: () => subscriber.cancelled??view(shared,subscriber), release });
|
|
},
|
|
close() {
|
|
if (closed) return; closed = true;
|
|
for (const job of [...inflight.values()]) { retire(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cancelled=Object.freeze({...view(job,subscriber),status:'cancelled'});subscriber.active = false; subscriber.cleanup(); subscriber.reject(fault('loader_closed')); } job.subscribers.clear(); }
|
|
queue.clear(); cache.clear(); cacheBytes = 0;
|
|
},
|
|
});
|
|
}
|
|
export type ScopedLoader<T> = ReturnType<typeof createScopedLoader<T>>;
|