CRT/storebound
1
0
Fork 0
Code Issues Pull requests Releases 1 Packages 1 Activity Actions

desktop: one read registry for every read button: single-flight per canonical key, broad reads cover narrow keys, overlapping buttons all busy and disabled (useReadState), 20 s 'read just now' memo, quota pause on the buttons, every read through the scheduler (Check again, Import apps, account Test, Read the stores, Dependencies), writes invalidate only the reads they change; tests/read-registry.test.ts; fixtures ?blocked and window.__fakeCalls

Browse source
This commit is contained in:
CRTkafa 2026-10-23 11:09:51 +08:00
parent 7345de2f7c
commit c03938f9dc
14 changed files with 861 additions and 164 deletions
Download patch file Download diff file Expand all files Collapse all files

283
apps/desktop/src/lib/read-registry.ts Normal file
View file

@ -0,0 +1,283 @@
// The one read registry of the window: every read button, notice, card, the background reader, the tray and Ctrl+K
// ask it, never the host directly. Pure logic over lib/scheduler.ts (tests/read-registry.test.ts).
//
// - Canonical key: target + method + args (lib/freshness.ts readKey), or a named key for reads without a target
// (the portfolio, an account's app list, an account check).
// - Single-flight: a read whose key is in flight (queued or running) is joined: the caller gets the same promise, its
// result or its error. No second host call, ever.
// - Scope: a read carries tags ('all', 'app:<target>', 'import', 'account:<id>' ...). A query asks by keys and / or
// tags; it is busy when any in-flight read has one of its keys, one of its tags, or COVERS one of its keys.
// - Covering: a broad read (Refresh all, Import all) is a group: it lists the narrow keys it stands for. While it runs
// those keys show busy everywhere and a click on one of them waits for the group, then reads only if the group did
// not (it usually did, so the click costs nothing). A narrow read in flight makes every broad button whose tag it
// carries busy too (decided with the owner: overlapping buttons are all busy and disabled; one click never starts a
// duplicate, it joins).
// - Just now: a read that finished less than `justNowMs` ago is not run again by a second click: the caller gets the
// last result ("read just now"). A write's invalidation drops that memo for the keys it touched.
// - Rate limits stay the scheduler's (lanes per store, spacing and the App Store Connect quota per account, shared
// 429 backoff); the registry reports a paused account on every query that uses it.
import { type Clock, type Lane, type Priority, type ReadScheduler, realClock } from './scheduler';
export interface ReadSpec<R> {
key: string;
/** Scopes the read belongs to: a query by any of these tags sees it. */
tags?: string[];
lanes: Lane[];
/** Credentials it uses ("apple:<account>", "play:<account>"): their quota pause shows on its buttons. */
accounts: string[];
priority: Priority;
run: () => Promise<R>;
/** Stops a running read (the host job). */
cancel?: () => void;
}
export interface GroupSpec<R> {
key: string;
tags?: string[];
/** Narrow keys this read stands for until it releases them (release()) or ends. */
covers: string[];
run: (g: GroupHandle) => Promise<R>;
}
export interface GroupHandle {
/** This key is no longer the group's (read by its own entry now, or skipped as fresh). */
release(key: string): void;
}
export type ReadQuery = string | { keys?: string[]; tags?: string[] };
export interface ReadStatus {
/** Something this query overlaps is queued or running. */
busy: boolean;
/** How many in-flight reads overlap it. */
inFlight: number;
/** Every key of the query finished successfully less than justNowMs ago. */
justNow: boolean;
/** Oldest last-finished time over the query's keys (epoch ms), null when one never finished here. */
lastAt: number | null;
/** The last error of a single-key query (cleared by its next success). */
error: string | null;
/** An account the query uses waits on a quota pause or a 429 backoff until this time (epoch ms). */
pausedUntil: number | null;
}
export type ReadEvent = { kind: 'start' } | { kind: 'done'; result: unknown } | { kind: 'error'; error: unknown };
interface Entry {
key: string;
tags: Set<string>;
covers: Set<string>;
accounts: string[];
promise: Promise<unknown>;
cancel?: () => void;
group: boolean;
}
interface Done { at: number; ok: boolean; result?: unknown; error?: string }
export interface RegistryOptions {
justNowMs?: number;
clock?: Clock;
/** Per-key start / done / error, for the cards' read bus (only real reads: never a join or a just-now answer). */
onEvent?: (key: string, e: ReadEvent) => void;
}
const errText = (e: unknown): string => (e instanceof Error ? e.message : typeof e === 'string' ? e : (e as { message?: string })?.message ?? String(e));
export class ReadRegistry {
private entries = new Map<string, Entry>();
private done = new Map<string, Done>();
private accountsOf = new Map<string, string[]>();
private listeners = new Set<() => void>();
private ver = 0;
private justNowMs: number;
private clock: Clock;
private onEvent?: RegistryOptions['onEvent'];
private expiry: unknown = null;
constructor(private sched: ReadScheduler, o: RegistryOptions = {}) {
this.justNowMs = o.justNowMs ?? 20_000;
this.clock = o.clock ?? realClock;
this.onEvent = o.onEvent;
// a quota pause or backoff of the scheduler changes what the buttons say
sched.subscribe(() => this.emit());
}
subscribe = (fn: () => void): (() => void) => { this.listeners.add(fn); return () => { this.listeners.delete(fn); }; };
/** Changes on every start, finish, cover release, invalidation and scheduler change (useSyncExternalStore). */
version = (): number => this.ver;
/** True when this exact key is in flight (its own read, or a group that covers it). */
active(key: string): boolean {
if (this.entries.has(key)) return true;
for (const e of this.entries.values()) if (e.covers.has(key)) return true;
return false;
}
/** The last result of a key that finished less than justNowMs ago (undefined when there is none). */
recent(key: string): unknown {
const d = this.done.get(key);
return d?.ok && this.clock.now() - d.at < this.justNowMs ? d.result : undefined;
}
/**
* Reads one key: joins it when it is in flight, waits for a group that covers it, answers a just-finished read
* from its memo (unless `force`), else runs it through the scheduler.
*/
read<R>(spec: ReadSpec<R>, o: { force?: boolean } = {}): Promise<R> {
this.accountsOf.set(spec.key, spec.accounts);
const own = this.entries.get(spec.key);
if (own) {
// a higher priority asker raises the queued read
if (!own.group) this.sched.enqueue({ key: spec.key, lanes: spec.lanes, accounts: spec.accounts, priority: spec.priority, run: spec.run as () => Promise<unknown> }).catch(() => {});
return own.promise as Promise<R>;
}
for (const e of this.entries.values()) {
if (e.covers.has(spec.key)) {
// the broad read stands for this key: wait for it, then read only what it did not
return e.promise.then(() => undefined, () => undefined).then(() => this.read(spec, { force: false }));
}
}
const d = this.done.get(spec.key);
if (!o.force && d?.ok && this.clock.now() - d.at < this.justNowMs) return Promise.resolve(d.result as R);
const p = this.sched.enqueue<R>({ key: spec.key, lanes: spec.lanes, accounts: spec.accounts, priority: spec.priority, run: spec.run });
const entry: Entry = { key: spec.key, tags: new Set(spec.tags ?? []), covers: new Set(), accounts: spec.accounts, promise: p, cancel: spec.cancel, group: false };
this.entries.set(spec.key, entry);
this.onEvent?.(spec.key, { kind: 'start' });
this.emit();
p.then(r => this.settle(entry, true, r), e => this.settle(entry, false, undefined, e));
return p;
}
/** A broad read made of narrow ones (Refresh all, Import all): single-flight on its own key too. */
group<R>(spec: GroupSpec<R>): Promise<R> {
const own = this.entries.get(spec.key);
if (own) return own.promise as Promise<R>;
const covers = new Set(spec.covers);
let resolve!: (r: R) => void, reject!: (e: unknown) => void;
const promise = new Promise<R>((a, b) => { resolve = a; reject = b; });
const entry: Entry = { key: spec.key, tags: new Set(spec.tags ?? []), covers, accounts: [], promise, group: true };
this.entries.set(spec.key, entry);
this.emit();
const handle: GroupHandle = { release: k => { if (covers.delete(k)) this.emit(); } };
// run after the entry exists, so a read started inside joins nothing by mistake (release first, then read)
Promise.resolve().then(() => spec.run(handle)).then(r => { this.settle(entry, true, r); resolve(r); }, e => { this.settle(entry, false, undefined, e); reject(e); });
return promise;
}
/** Stops what overlaps the query: queued reads are dropped (their waiters get "cancelled"), running ones cancel their job. */
cancel(q: ReadQuery): void {
const { keys, tags } = norm(q);
for (const e of [...this.entries.values()]) {
if (!this.overlaps(e, keys, tags) || e.group) continue;
if (this.sched.isRunning(e.key)) e.cancel?.();
else this.sched.clear(k => k === e.key);
}
}
/** A write changed what these keys read: their "just now" memo goes, so the next read really reads. */
invalidate(match: (key: string) => boolean): string[] {
const hit: string[] = [];
for (const k of [...this.done.keys()]) if (match(k)) { this.done.delete(k); hit.push(k); }
if (hit.length) this.emit();
return hit;
}
state(q: ReadQuery): ReadStatus {
const { keys, tags } = norm(q);
const now = this.clock.now();
let inFlight = 0;
const accts = new Set<string>();
for (const e of this.entries.values()) {
if (!this.overlaps(e, keys, tags)) continue;
inFlight++;
for (const a of e.accounts) accts.add(a);
}
for (const k of keys) for (const a of this.accountsOf.get(k) ?? []) accts.add(a);
let lastAt: number | null = keys.length ? Infinity : null;
let justNow = keys.length > 0;
for (const k of keys) {
const d = this.done.get(k);
if (!d) { lastAt = null; justNow = false; break; }
lastAt = Math.min(lastAt as number, d.at);
if (!d.ok || now - d.at >= this.justNowMs) justNow = false;
}
const paused = this.sched.stats().paused;
let pausedUntil: number | null = null;
for (const a of accts) if (paused[a] && paused[a] > now) pausedUntil = Math.max(pausedUntil ?? 0, paused[a]);
const one = keys.length === 1 ? this.done.get(keys[0]!) : undefined;
return { busy: inFlight > 0, inFlight, justNow: justNow && inFlight === 0, lastAt: lastAt === Infinity ? null : lastAt, error: one && !one.ok ? one.error ?? null : null, pausedUntil };
}
private overlaps(e: Entry, keys: string[], tags: string[]): boolean {
for (const k of keys) if (e.key === k || e.covers.has(k)) return true;
for (const t of tags) if (e.tags.has(t) || e.key === t) return true;
return false;
}
private settle(e: Entry, ok: boolean, result: unknown, error?: unknown): void {
if (this.entries.get(e.key) === e) this.entries.delete(e.key);
const at = this.clock.now();
if (ok) this.done.set(e.key, { at, ok, result });
else if (!/cancel/i.test(errText(error))) this.done.set(e.key, { at, ok, error: errText(error) });
if (!e.group) this.onEvent?.(e.key, ok ? { kind: 'done', result } : { kind: 'error', error });
this.armExpiry();
this.emit();
}
/** "Read just now" ends by itself: one timer re-renders the buttons when the oldest memo runs out. */
private armExpiry(): void {
if (this.expiry !== null) return;
this.expiry = this.clock.setTimeout(() => { this.expiry = null; this.emit(); }, this.justNowMs + 50);
}
private emit(): void {
this.ver++;
for (const f of this.listeners) { try { f(); } catch { /* a listener never breaks the registry */ } }
}
}
function norm(q: ReadQuery): { keys: string[]; tags: string[] } {
return typeof q === 'string' ? { keys: [q], tags: [] } : { keys: q.keys ?? [], tags: q.tags ?? [] };
}
// ---- canonical keys of reads without an app target ----
export const PORTFOLIO_KEY = '*\u001fportfolio.status\u001f{"live":true}';
export const PORTFOLIO_CHECK_KEY = '*\u001fportfolio.status\u001f{}';
export const READ_ALL_KEY = '*\u001freadAll';
export const IMPORT_ALL_KEY = '*\u001fimport';
export const importKey = (account: string): string => `account:${account}\u001fstore_apps.list`;
export const accountCheckKey = (account: string): string => `account:${account}\u001faccount.check`;
/** The scope tags of a read of one app (Refresh all sees every app read; an app's own buttons see `app:<target>`). */
export const appTags = (target: string): string[] => ['all', `app:${target}`];
// ---- which reads a write changes ----
const ALL = null;
const LISTING = ['project.listings', 'project.editable', 'listing.live', 'validate', 'scan', 'aso.audit'];
const CATALOG = ['prices.current', 'prices.diff', 'catalog.diff', 'validate', 'scan'];
const ART = ['art.live', 'project.renders', 'project.icon', 'validate', 'scan'];
/**
* The read methods of the same app a write makes outdated (null: all of them). `portfolio` true when the write
* changes release state the portfolio read shows. Unknown writes invalidate everything of the app (safe default).
*/
export function affectedReads(writeMethod: string): { methods: string[] | null; portfolio: boolean } {
const m = writeMethod;
const r = (methods: string[] | null, portfolio = false) => ({ methods, portfolio });
if (m === 'owner.play_rollout' || m === 'owner.ios_phased') return r(['apple.status', 'play.status'], true);
if (m.startsWith('reviews.') || m === 'owner.reply_send') return r(['reviews.list', 'reviews.drafts']);
if (m.startsWith('prices.') || m === 'owner.prices_import' || m === 'apple.iap' || m === 'apple.subscriptions' || m === 'play.products' || m === 'play.subscriptions') return r(CATALOG);
if (m === 'apple.listing_pull' || m === 'play.listing_pull' || m === 'play.listing' || m === 'apple.appinfo' || m === 'apple.review_details') return r(LISTING);
if (m.startsWith('art.') || m.startsWith('shots.') || m === 'apple.screenshots' || m === 'apple.previews' || m === 'play.images' || m === 'icons') return r(ART);
if (m.startsWith('apple.testflight_') || m === 'owner.testflight_testers') return r(['apple.testflight_status']);
if (m.startsWith('owner.ppo_') || m.startsWith('apple.ppo_')) return r(['apple.ppo_status']);
if (m === 'apple.game_center' || m === 'play.games') return r(['games.status']);
if (m === 'apple.encryption' || m === 'play.data_safety') return r(['project.declarations', 'validate', 'scan']);
if (m.startsWith('analytics.')) return r(['analytics.compare']);
return r(ALL, true);
}
/** The method part of a canonical key (readKey: target \u001f method \u001f args). */
export const methodOfKey = (key: string): string => key.split('\u001f')[1] ?? '';
export const targetOfKey = (key: string): string => key.split('\u001f')[0] ?? '';

3
apps/desktop/src/lib/scheduler.ts
View file

@ -127,6 +127,9 @@ export class ReadScheduler {
subscribe(fn: () => void): () => void { this.listeners.add(fn); return () => { this.listeners.delete(fn); }; }
/** True while the read of this key is running (not just queued). */
isRunning(key: string): boolean { return this.running.has(key); }
has(key: string): boolean { return this.running.has(key) || this.queue.some(i => i.task.key === key); }
enqueue<R>(task: ReadTask<R>): Promise<R> {