- commit
- 25b3059438616f7e3c02d1942b2a32d1d4482005
- parent
- 654cf1041c3896ce1e130a2652fa05733e6c9fed
- author
- Dami <dami.etoile@gmail.com>
| M | package/ | 1 | + |
| M | package/ | 1 | + |
| A | package/ | 273 | ++++++++++++++++++++++++++++++ |
| A | package/ | 242 | +++++++++++++++++++++++++++ |
| M | package/ | 6 | + |
| M | package/ | 9 | + |
| M | package/ | 11 | +- |
| M | package/ | 15 | ++ |
@@ -28,6 +28,7 @@ const runtimeFiles = [
2828 "foreground-recovery.ts",2929 "inspector.ts",3030 "lifecycle.ts",31+ "live-query-store.ts",3132 "mod.ts",3233 "passkey-recovery.ts",3334 "probe.ts",@@ -27,6 +27,7 @@ Deno.test("prepareLofiAstroConfig materializes package-owned vendor integration"
2727 await Deno.stat(`${root}/.lofi/package/preact/DeviceStatus.tsx`);2828 await Deno.stat(`${root}/.lofi/package/preact/RuntimeRecovery.tsx`);2929 await Deno.stat(`${root}/.lofi/package/runtime/passkey-recovery.ts`);30+ await Deno.stat(`${root}/.lofi/package/runtime/live-query-store.ts`);3031 await Deno.stat(`${root}/.lofi/package/runtime/startup-recovery.ts`);3132 await Deno.stat(`${root}/.lofi/package/access/mod.ts`);3233 } finally {@@ -0,0 +1,273 @@
1+import type { Db, QueryBuilder } from "jazz-tools";2+import type { RuntimeDiagnostics } from "./diagnostics.ts";3+import { getRuntime, runtimeRecreatedEvent, updateRuntimeDiagnostics } from "./runtime.ts";4+import type { TableRow } from "./table-store.ts";5+6+/** Honest read state for an arbitrary typed Jazz query. */7+export type LiveQuerySnapshot<T extends TableRow> = {8+ status: "loading" | "ready" | "error";9+ rows: T[];10+ error: string | null;11+};12+13+type Listener = () => void;14+type QuerySubscriptionDb = Pick<Db, "subscribeAll">;15+16+/** Runtime seams used by a live-query registry. Exported for deterministic framework tests. */17+export type LiveQueryEnvironment = {18+ getDb(): Promise<QuerySubscriptionDb>;19+ subscribeRuntimeRecreation(listener: () => void): () => void;20+ updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void;21+};22+23+function canonicalJson(value: unknown): unknown {24+ if (Array.isArray(value)) return value.map(canonicalJson);25+ if (!value || typeof value !== "object") return value;26+ return Object.fromEntries(27+ Object.entries(value as Record<string, unknown>)28+ .sort(([left], [right]) => left.localeCompare(right))29+ .map(([key, nested]) => [key, canonicalJson(nested)]),30+ );31+}32+33+/** Returns the stable serialized plan used to share equivalent query stores. */34+export function liveQueryKey<T extends TableRow>(query: QueryBuilder<T>): string {35+ return JSON.stringify(canonicalJson(JSON.parse(query._build())));36+}37+38+/** A shared framework-neutral reactive store for one typed Jazz query. */39+export class LiveQueryStore<T extends TableRow> {40+ readonly #query: QueryBuilder<T>;41+ readonly #environment: LiveQueryEnvironment;42+ readonly #onIdle: () => void;43+ readonly #listeners = new Set<Listener>();44+ #snapshot: LiveQuerySnapshot<T> = { status: "loading", rows: [], error: null };45+ #vendorUnsubscribe: (() => void) | null = null;46+ #stopRuntimeRecreation: (() => void) | null = null;47+ #generation = 0;48+ #disposed = false;49+50+ /** Creates one registry-owned store. Applications acquire stores through {@link acquireLiveQuery}. */51+ constructor(52+ query: QueryBuilder<T>,53+ environment: LiveQueryEnvironment,54+ onIdle: () => void,55+ ) {56+ this.#query = query;57+ this.#environment = environment;58+ this.#onIdle = onIdle;59+ }60+61+ /** Returns the current immutable query snapshot. */62+ getSnapshot = (): LiveQuerySnapshot<T> => this.#snapshot;63+64+ /** Subscribes one consumer and returns an idempotent cleanup function. */65+ subscribe = (listener: Listener): () => void => {66+ if (this.#disposed) throw new Error("live query store has been released");67+ const subscriber = () => listener();68+ this.#listeners.add(subscriber);69+ this.#updateDiagnostics((diagnostics) => diagnostics.activeConsumers += 1);70+ if (this.#listeners.size === 1) this.#start();71+ listener();72+ let active = true;73+ return () => {74+ if (!active) return;75+ active = false;76+ if (!this.#listeners.delete(subscriber)) return;77+ this.#updateDiagnostics((diagnostics) => diagnostics.activeConsumers -= 1);78+ if (this.#listeners.size === 0) {79+ this.#stop();80+ this.#onIdle();81+ }82+ };83+ };84+85+ /** Number of mounted framework-neutral consumers currently sharing this store. */86+ get consumerCount(): number {87+ return this.#listeners.size;88+ }89+90+ /** Releases every subscription owned by an evicted or hot-reloaded store. */91+ dispose(): void {92+ if (this.#disposed) return;93+ this.#disposed = true;94+ const consumers = this.#listeners.size;95+ this.#listeners.clear();96+ if (consumers > 0) {97+ this.#updateDiagnostics((diagnostics) => diagnostics.activeConsumers -= consumers);98+ }99+ this.#stop();100+ }101+102+ #start(): void {103+ try {104+ this.#stopRuntimeRecreation = this.#environment.subscribeRuntimeRecreation(() =>105+ this.#restart()106+ );107+ this.#connect();108+ } catch (error) {109+ this.#fail(error);110+ }111+ }112+113+ #restart(): void {114+ if (this.#disposed || this.#listeners.size === 0) return;115+ this.#generation += 1;116+ this.#closeVendorSubscription();117+ this.#snapshot = { status: "loading", rows: [], error: null };118+ this.#emit();119+ this.#connect();120+ }121+122+ #connect(): void {123+ const generation = ++this.#generation;124+ void this.#environment.getDb().then((db) => {125+ if (this.#disposed || generation !== this.#generation || this.#listeners.size === 0) return;126+ try {127+ const unsubscribe = db.subscribeAll(this.#query, (delta) => {128+ if (this.#disposed || generation !== this.#generation) return;129+ this.#snapshot = { status: "ready", rows: delta.all, error: null };130+ this.#emit();131+ });132+ if (this.#disposed || generation !== this.#generation || this.#listeners.size === 0) {133+ unsubscribe();134+ return;135+ }136+ this.#vendorUnsubscribe = unsubscribe;137+ this.#updateDiagnostics((diagnostics) => {138+ diagnostics.activeVendorSubscriptions += 1;139+ diagnostics.totalVendorSubscriptions += 1;140+ });141+ } catch (error) {142+ this.#fail(error, generation);143+ }144+ }, (error) => this.#fail(error, generation));145+ }146+147+ #stop(): void {148+ this.#generation += 1;149+ this.#closeVendorSubscription();150+ this.#stopRuntimeRecreation?.();151+ this.#stopRuntimeRecreation = null;152+ }153+154+ #closeVendorSubscription(): void {155+ if (!this.#vendorUnsubscribe) return;156+ this.#vendorUnsubscribe();157+ this.#vendorUnsubscribe = null;158+ this.#updateDiagnostics((diagnostics) => {159+ diagnostics.activeVendorSubscriptions -= 1;160+ diagnostics.unsubscribeCalls += 1;161+ });162+ }163+164+ #fail(error: unknown, generation = this.#generation): void {165+ if (this.#disposed || generation !== this.#generation) return;166+ this.#snapshot = {167+ status: "error",168+ rows: [],169+ error: error instanceof Error ? error.message : String(error),170+ };171+ this.#emit();172+ }173+174+ #updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void {175+ this.#environment.updateDiagnostics(update);176+ }177+178+ #emit(): void {179+ for (const listener of this.#listeners) listener();180+ }181+}182+183+/** A retained reference to a shared live-query store. Release it after its consumer unsubscribes. */184+export type LiveQueryLease<T extends TableRow> = {185+ store: LiveQueryStore<T>;186+ release(): void;187+};188+189+type RegistryEntry = {190+ store: LiveQueryStore<TableRow>;191+ leases: number;192+};193+194+/** Registry that keys stores by schema identity and stable serialized query plan. */195+export class LiveQueryRegistry {196+ readonly #environment: LiveQueryEnvironment;197+ readonly #schemas = new WeakMap<object, Map<string, RegistryEntry>>();198+ readonly #stores = new Set<LiveQueryStore<TableRow>>();199+200+ /** Creates an isolated registry over the supplied runtime environment. */201+ constructor(environment: LiveQueryEnvironment) {202+ this.#environment = environment;203+ }204+205+ /** Acquires a shared store and returns an idempotent ownership release. */206+ acquire<T extends TableRow>(query: QueryBuilder<T>): LiveQueryLease<T> {207+ const schema = query._schema as object;208+ const key = liveQueryKey(query);209+ let entries = this.#schemas.get(schema);210+ if (!entries) {211+ entries = new Map();212+ this.#schemas.set(schema, entries);213+ }214+ let entry = entries.get(key);215+ if (!entry) {216+ const store = new LiveQueryStore(217+ query,218+ this.#environment,219+ () => this.#evictIfUnused(entries!, key),220+ );221+ entry = { store: store as LiveQueryStore<TableRow>, leases: 0 };222+ entries.set(key, entry);223+ this.#stores.add(entry.store);224+ }225+ entry.leases += 1;226+ let retained = true;227+ return {228+ store: entry.store as LiveQueryStore<T>,229+ release: () => {230+ if (!retained) return;231+ retained = false;232+ entry!.leases -= 1;233+ this.#evictIfUnused(entries!, key);234+ },235+ };236+ }237+238+ /** Disposes every retained store, including active stores, during hot replacement. */239+ dispose(): void {240+ for (const store of this.#stores) store.dispose();241+ this.#stores.clear();242+ }243+244+ #evictIfUnused(entries: Map<string, RegistryEntry>, key: string): void {245+ const entry = entries.get(key);246+ if (!entry || entry.leases > 0 || entry.store.consumerCount > 0) return;247+ entries.delete(key);248+ this.#stores.delete(entry.store);249+ entry.store.dispose();250+ }251+}252+253+const defaultRegistry = new LiveQueryRegistry({254+ getDb: async () => (await getRuntime()).db,255+ subscribeRuntimeRecreation(listener) {256+ globalThis.addEventListener(runtimeRecreatedEvent, listener);257+ return () => globalThis.removeEventListener(runtimeRecreatedEvent, listener);258+ },259+ updateDiagnostics,260+});261+262+function updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void {263+ updateRuntimeDiagnostics(update);264+}265+266+/** Acquires the package-wide shared store for an arbitrary typed query. */267+export function acquireLiveQuery<T extends TableRow>(268+ query: QueryBuilder<T>,269+): LiveQueryLease<T> {270+ return defaultRegistry.acquire(query);271+}272+273+import.meta.hot?.dispose(() => defaultRegistry.dispose());@@ -0,0 +1,242 @@
1+import type { Db, QueryBuilder } from "jazz-tools";2+import { schema as s } from "jazz-tools";3+import { createDiagnostics } from "./diagnostics.ts";4+import {5+ type LiveQueryEnvironment,6+ LiveQueryRegistry,7+ type LiveQuerySnapshot,8+} from "./live-query-store.ts";9+import { assert, assertCount } from "./test-assert.ts";10+11+type Row = { id: string; title: string };12+13+function query(14+ schema: object,15+ plan: Record<string, unknown>,16+): QueryBuilder<Row> {17+ return {18+ _table: String(plan.table ?? "records"),19+ _schema: schema,20+ _build: () => JSON.stringify(plan),21+ } as QueryBuilder<Row>;22+}23+24+class FakeDb {25+ subscribeCalls = 0;26+ unsubscribeCalls = 0;27+ throwOnSubscribe: Error | null = null;28+ readonly callbacks = new Set<(rows: Row[]) => void>();29+30+ value(): Pick<Db, "subscribeAll"> {31+ return this as unknown as Pick<Db, "subscribeAll">;32+ }33+34+ subscribeAll(35+ _query: QueryBuilder<Row>,36+ callback: (delta: { all: Row[] }) => void,37+ ): () => void {38+ if (this.throwOnSubscribe) throw this.throwOnSubscribe;39+ this.subscribeCalls += 1;40+ const publish = (rows: Row[]) => callback({ all: rows });41+ this.callbacks.add(publish);42+ let active = true;43+ return () => {44+ if (!active) return;45+ active = false;46+ this.callbacks.delete(publish);47+ this.unsubscribeCalls += 1;48+ };49+ }50+51+ publish(rows: Row[]): void {52+ for (const callback of this.callbacks) callback(rows);53+ }54+}55+56+function testEnvironment(db: FakeDb, diagnostics = createDiagnostics()) {57+ const recreationListeners = new Set<() => void>();58+ const environment: LiveQueryEnvironment = {59+ getDb: () => Promise.resolve(db.value()),60+ subscribeRuntimeRecreation(listener) {61+ recreationListeners.add(listener);62+ return () => recreationListeners.delete(listener);63+ },64+ updateDiagnostics(update) {65+ update(diagnostics);66+ },67+ };68+ return {69+ environment,70+ diagnostics,71+ recreate() {72+ for (const listener of recreationListeners) listener();73+ },74+ recreationListeners,75+ };76+}77+78+async function flush(): Promise<void> {79+ await Promise.resolve();80+ await Promise.resolve();81+}82+83+Deno.test("equivalent queries share one subscription and final release evicts the store", async () => {84+ const db = new FakeDb();85+ const runtime = testEnvironment(db);86+ const registry = new LiveQueryRegistry(runtime.environment);87+ const schema = {};88+ const first = registry.acquire(query(schema, {89+ table: "records",90+ conditions: [{ column: "archived", op: "eq", value: false }],91+ includes: { owner: true, comments: true },92+ }));93+ const second = registry.acquire(query(schema, {94+ includes: { comments: true, owner: true },95+ conditions: [{ value: false, op: "eq", column: "archived" }],96+ table: "records",97+ }));98+ assert(first.store === second.store, "stable query serialization did not deduplicate stores");99+100+ let firstUpdates = 0;101+ let secondUpdates = 0;102+ const stopFirst = first.store.subscribe(() => firstUpdates += 1);103+ const stopSecond = second.store.subscribe(() => secondUpdates += 1);104+ await flush();105+ assertCount(db.subscribeCalls, 1, "equivalent consumers opened duplicate vendor subscriptions");106+ assertCount(runtime.diagnostics.activeConsumers, 2, "mounted consumers were not counted");107+ db.publish([]);108+ assert(first.store.getSnapshot().status === "ready", "an empty result remained loading");109+ db.publish([{ id: "row-1", title: "shared" }]);110+ assert(firstUpdates > 1 && secondUpdates > 1, "shared rows did not reach both consumers");111+112+ stopFirst();113+ first.release();114+ assertCount(db.unsubscribeCalls, 0, "one remaining consumer lost its shared subscription");115+ stopSecond();116+ second.release();117+ assertCount(db.unsubscribeCalls, 1, "the final consumer did not close the vendor subscription");118+ assertCount(runtime.diagnostics.activeConsumers, 0, "consumer diagnostics leaked after teardown");119+120+ const replacement = registry.acquire(query(schema, {121+ table: "records",122+ conditions: [{ column: "archived", op: "eq", value: false }],123+ includes: { owner: true, comments: true },124+ }));125+ assert(replacement.store !== first.store, "the final release retained an obsolete query store");126+ replacement.release();127+});128+129+Deno.test("identical query plans from different schemas never share a store", () => {130+ const registry = new LiveQueryRegistry(testEnvironment(new FakeDb()).environment);131+ const first = registry.acquire(query({}, { table: "records" }));132+ const second = registry.acquire(query({}, { table: "records" }));133+ assert(first.store !== second.store, "query identity crossed schema boundaries");134+ first.release();135+ second.release();136+});137+138+Deno.test("changing query dependencies closes the obsolete query before opening the next", async () => {139+ const db = new FakeDb();140+ const registry = new LiveQueryRegistry(testEnvironment(db).environment);141+ const schema = {};142+ const first = registry.acquire(query(schema, {143+ table: "records",144+ conditions: [{ column: "workspaceId", op: "eq", value: "one" }],145+ }));146+ const stopFirst = first.store.subscribe(() => undefined);147+ await flush();148+ stopFirst();149+ first.release();150+ assertCount(db.unsubscribeCalls, 1, "obsolete dependency retained its subscription");151+152+ const second = registry.acquire(query(schema, {153+ table: "records",154+ conditions: [{ column: "workspaceId", op: "eq", value: "two" }],155+ }));156+ const stopSecond = second.store.subscribe(() => undefined);157+ await flush();158+ assertCount(db.subscribeCalls, 2, "new dependency did not open its query");159+ stopSecond();160+ second.release();161+});162+163+Deno.test("synchronous vendor setup errors become stable query read state", async () => {164+ const db = new FakeDb();165+ db.throwOnSubscribe = new Error("query setup failed");166+ const registry = new LiveQueryRegistry(testEnvironment(db).environment);167+ const lease = registry.acquire(query({}, { table: "records" }));168+ const stop = lease.store.subscribe(() => undefined);169+ await flush();170+ const snapshot = lease.store.getSnapshot();171+ assert(snapshot.status === "error", "synchronous setup failure left the query loading");172+ assert(snapshot.rows.length === 0, "failed setup retained stale rows");173+ assert(snapshot.error === "query setup failed", "setup failure was hidden");174+ stop();175+ lease.release();176+});177+178+Deno.test("runtime recreation ignores stale async clients and reconnects mounted queries", async () => {179+ const diagnostics = createDiagnostics();180+ const recreationListeners = new Set<() => void>();181+ const pending: Array<(db: Pick<Db, "subscribeAll">) => void> = [];182+ const environment: LiveQueryEnvironment = {183+ getDb: () => new Promise((resolve) => pending.push(resolve)),184+ subscribeRuntimeRecreation(listener) {185+ recreationListeners.add(listener);186+ return () => recreationListeners.delete(listener);187+ },188+ updateDiagnostics(update) {189+ update(diagnostics);190+ },191+ };192+ const registry = new LiveQueryRegistry(environment);193+ const lease = registry.acquire(query({}, { table: "records" }));194+ let updates = 0;195+ const stop = lease.store.subscribe(() => updates += 1);196+ assertCount(pending.length, 1, "initial runtime acquisition did not start");197+ for (const listener of recreationListeners) listener();198+ assertCount(pending.length, 2, "recreation did not request the replacement runtime");199+200+ const obsolete = new FakeDb();201+ const replacement = new FakeDb();202+ pending[1]!(replacement.value());203+ await flush();204+ pending[0]!(obsolete.value());205+ await flush();206+ assertCount(obsolete.subscribeCalls, 0, "stale runtime won the recreation race");207+ assertCount(replacement.subscribeCalls, 1, "replacement runtime did not receive the query");208+ replacement.publish([{ id: "row-new", title: "replacement" }]);209+ const latest: LiveQuerySnapshot<Row> = lease.store.getSnapshot();210+ assert(latest.rows[0]?.id === "row-new", "replacement rows did not reach the mounted query");211+ assert(updates > 1, "replacement rows did not notify the mounted consumer");212+213+ stop();214+ lease.release();215+});216+217+Deno.test("registry disposal releases active HMR-owned resources", async () => {218+ const db = new FakeDb();219+ const runtime = testEnvironment(db);220+ const registry = new LiveQueryRegistry(runtime.environment);221+ const lease = registry.acquire(query({}, { table: "records" }));222+ const stop = lease.store.subscribe(() => undefined);223+ await flush();224+ registry.dispose();225+ assertCount(db.unsubscribeCalls, 1, "registry disposal retained the vendor subscription");226+ assertCount(runtime.diagnostics.activeConsumers, 0, "registry disposal retained consumers");227+ assertCount(runtime.recreationListeners.size, 0, "registry disposal retained event listeners");228+ stop();229+ lease.release();230+});231+232+Deno.test("query row projections remain inferred from the exact Jazz builder", () => {233+ const app = s.defineApp({ records: s.table({ title: s.string(), archived: s.boolean() }) });234+ const registry = new LiveQueryRegistry(testEnvironment(new FakeDb()).environment);235+ const lease = registry.acquire(app.records.select("title"));236+ const row = lease.store.getSnapshot().rows[0];237+ const title: string | undefined = row?.title;238+ assert(title === undefined, "an unsubscribed projection unexpectedly produced a row");239+ // @ts-expect-error select("title") must not retain unselected application columns.240+ row?.archived;241+ lease.release();242+});@@ -58,6 +58,12 @@ export {
5858 subscribePwaState,5959 } from "./pwa.ts";6060 export { type RuntimeDiagnostics } from "./diagnostics.ts";61+export {62+ acquireLiveQuery,63+ type LiveQueryLease,64+ type LiveQuerySnapshot,65+ type LiveQueryStore,66+} from "./live-query-store.ts";6167 export {6268 getRuntime,6369 getRuntimeDiagnostics,@@ -261,6 +261,15 @@ export function getRuntimeDiagnostics(): RuntimeDiagnostics {
261261 return { ...slot().diagnostics };262262 }263263 264+/** Applies one package-internal diagnostics update and notifies active observers. */265+export function updateRuntimeDiagnostics(266+ update: (diagnostics: RuntimeDiagnostics) => void,267+): void {268+ const state = slot();269+ update(state.diagnostics);270+ notifyDiagnostics(state);271+}272+264273 /** The stable Jazz principal currently opened by the package runtime. */265274 export function getRuntimePrincipal(): string | null {266275 return slot().principal;@@ -80,8 +80,9 @@ export class TableStore<T extends TableRow, Init> {
8080 8181 /** Subscribes to snapshot changes and opens the vendor subscription on first use. */8282 subscribe = (listener: Listener): () => void => {83- this.#listeners.add(listener);84- this.#diagnostics.activeConsumers = this.#listeners.size;83+ const subscriber = () => listener();84+ this.#listeners.add(subscriber);85+ this.#diagnostics.activeConsumers += 1;8586 this.#diagnosticsChanged();8687 if (this.#listeners.size === 1) this.#openVendorSubscription();8788 this.#emit();@@ -89,8 +90,8 @@ export class TableStore<T extends TableRow, Init> {
8990 return () => {9091 if (!active) return;9192 active = false;92- this.#listeners.delete(listener);93- this.#diagnostics.activeConsumers = this.#listeners.size;93+ this.#listeners.delete(subscriber);94+ this.#diagnostics.activeConsumers -= 1;9495 this.#diagnosticsChanged();9596 if (this.#listeners.size === 0) this.#closeVendorSubscription();9697 else this.#emit();@@ -121,8 +122,8 @@ export class TableStore<T extends TableRow, Init> {
121122 122123 /** Releases vendor subscriptions and mutation listeners owned by this store. */123124 close(): void {125+ this.#diagnostics.activeConsumers -= this.#listeners.size;124126 this.#listeners.clear();125- this.#diagnostics.activeConsumers = 0;126127 this.#closeVendorSubscription();127128 this.#stopMutationErrors();128129 this.#diagnostics.activeMutationListeners -= 1;@@ -142,6 +142,21 @@ test("two consumers share one table subscription and clean up idempotently", ()
142142 assertCount(counts.unsubscribeCalls, 1, "cleanup must remain externally diagnosable");143143 });144144 145+test("consumer diagnostics remain additive across distinct table stores", () => {146+ const counts = diagnostics();147+ const first = createTableStore<Row, RowInit>(fakeDb().value, table, counts);148+ const second = createTableStore<Row, RowInit>(fakeDb().value, table, counts);149+ const stopFirst = first.subscribe(() => undefined);150+ const stopSecond = second.subscribe(() => undefined);151+ assertCount(counts.activeConsumers, 2, "distinct stores overwrote the shared consumer count");152+ stopFirst();153+ assertCount(counts.activeConsumers, 1, "one store teardown erased another store's consumer");154+ stopSecond();155+ first.close();156+ second.close();157+ assertCount(counts.activeConsumers, 0, "distinct store cleanup leaked consumers");158+});159+145160 test("every mutation exposes pending work and waits for local durability", async () => {146161 const db = fakeDb();147162 const counts = diagnostics();