1import type { Db, QueryBuilder } from "jazz-tools";2import type { RuntimeDiagnostics } from "./diagnostics.ts";3import { getRuntime, runtimeRecreatedEvent, updateRuntimeDiagnostics } from "./runtime.ts";4import { subscribeSharedKeyring } from "../schema/shared-keyring.ts";5import type { TableRow } from "./table-store.ts";67/** Honest read state for an arbitrary typed Jazz query. */8export type LiveQuerySnapshot<T extends TableRow> = {9 status: "loading" | "ready" | "error";10 rows: T[];11 error: string | null;12};1314/** Runtime seams used by a live-query registry. Exported for deterministic framework tests. */15export type LiveQueryEnvironment = {16 getDb(): Promise<Pick<Db, "subscribeAll">>;17 subscribeRuntimeRecreation(listener: () => void): () => void;18 /** Fires when a shared field key is installed; stores resubscribe so rows19 * previously surfaced as key-pending re-materialize into plaintext. Key20 * installs are rare (boot, membership changes), so restarting every active21 * query is simpler than tracking which queries touch shared columns and22 * costs nothing in steady state. */23 subscribeSharedKeyring?(listener: () => void): () => void;24 updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void;25};2627function canonicalJson(value: unknown): unknown {28 if (Array.isArray(value)) return value.map(canonicalJson);29 if (!value || typeof value !== "object") return value;30 return Object.fromEntries(31 Object.entries(value as Record<string, unknown>)32 .sort(([left], [right]) => left.localeCompare(right))33 .map(([key, nested]) => [key, canonicalJson(nested)]),34 );35}3637/** Returns the stable serialized plan used to share equivalent query stores. */38export function liveQueryKey<T extends TableRow>(query: QueryBuilder<T>): string {39 return JSON.stringify(canonicalJson(JSON.parse(query._build())));40}4142/** A shared framework-neutral reactive store for one typed Jazz query. */43export class LiveQueryStore<T extends TableRow> {44 readonly #query: QueryBuilder<T>;45 readonly #environment: LiveQueryEnvironment;46 readonly #onIdle: () => void;47 readonly #listeners = new Set<() => void>();48 #snapshot: LiveQuerySnapshot<T> = { status: "loading", rows: [], error: null };49 #vendorUnsubscribe: (() => void) | null = null;50 #stopRuntimeRecreation: (() => void) | null = null;51 #stopKeyring: (() => void) | null = null;52 #generation = 0;53 #disposed = false;5455 /** Creates one registry-owned store. Applications acquire stores through {@link acquireLiveQuery}. */56 constructor(57 query: QueryBuilder<T>,58 environment: LiveQueryEnvironment,59 onIdle: () => void,60 ) {61 this.#query = query;62 this.#environment = environment;63 this.#onIdle = onIdle;64 }6566 /** Returns the current immutable query snapshot. */67 getSnapshot = (): LiveQuerySnapshot<T> => this.#snapshot;6869 /** Subscribes one consumer and returns an idempotent cleanup function. */70 subscribe = (listener: () => void): () => void => {71 if (this.#disposed) throw new Error("live query store has been released");72 const subscriber = () => listener();73 this.#listeners.add(subscriber);74 this.#updateDiagnostics((diagnostics) => diagnostics.activeConsumers += 1);75 if (this.#listeners.size === 1) this.#start();76 listener();77 let active = true;78 return () => {79 if (!active) return;80 active = false;81 if (!this.#listeners.delete(subscriber)) return;82 this.#updateDiagnostics((diagnostics) => diagnostics.activeConsumers -= 1);83 if (this.#listeners.size === 0) {84 this.#stop();85 this.#onIdle();86 }87 };88 };8990 /** Number of mounted framework-neutral consumers currently sharing this store. */91 get consumerCount(): number {92 return this.#listeners.size;93 }9495 /** Releases every subscription owned by an evicted or hot-reloaded store. */96 dispose(): void {97 if (this.#disposed) return;98 this.#disposed = true;99 const consumers = this.#listeners.size;100 this.#listeners.clear();101 if (consumers > 0) {102 this.#updateDiagnostics((diagnostics) => diagnostics.activeConsumers -= consumers);103 }104 this.#stop();105 }106107 #start(): void {108 try {109 this.#stopRuntimeRecreation = this.#environment.subscribeRuntimeRecreation(() =>110 this.#restart()111 );112 this.#stopKeyring = this.#environment.subscribeSharedKeyring?.(() => this.#restart()) ??113 null;114 this.#connect();115 } catch (error) {116 this.#fail(error);117 }118 }119120 #restart(): void {121 if (this.#disposed || this.#listeners.size === 0) return;122 this.#generation += 1;123 this.#closeVendorSubscription();124 this.#snapshot = { status: "loading", rows: [], error: null };125 this.#emit();126 this.#connect();127 }128129 #connect(): void {130 const generation = ++this.#generation;131 void this.#environment.getDb().then((db) => {132 if (this.#disposed || generation !== this.#generation || this.#listeners.size === 0) return;133 try {134 const unsubscribe = db.subscribeAll(this.#query, (delta) => {135 if (this.#disposed || generation !== this.#generation) return;136 this.#snapshot = { status: "ready", rows: delta.all, error: null };137 this.#emit();138 });139 if (this.#disposed || generation !== this.#generation || this.#listeners.size === 0) {140 unsubscribe();141 return;142 }143 this.#vendorUnsubscribe = unsubscribe;144 this.#updateDiagnostics((diagnostics) => {145 diagnostics.activeVendorSubscriptions += 1;146 diagnostics.totalVendorSubscriptions += 1;147 });148 } catch (error) {149 this.#fail(error, generation);150 }151 }, (error) => this.#fail(error, generation));152 }153154 #stop(): void {155 this.#generation += 1;156 this.#closeVendorSubscription();157 this.#stopRuntimeRecreation?.();158 this.#stopRuntimeRecreation = null;159 this.#stopKeyring?.();160 this.#stopKeyring = null;161 }162163 #closeVendorSubscription(): void {164 if (!this.#vendorUnsubscribe) return;165 this.#vendorUnsubscribe();166 this.#vendorUnsubscribe = null;167 this.#updateDiagnostics((diagnostics) => {168 diagnostics.activeVendorSubscriptions -= 1;169 diagnostics.unsubscribeCalls += 1;170 });171 }172173 #fail(error: unknown, generation = this.#generation): void {174 if (this.#disposed || generation !== this.#generation) return;175 this.#snapshot = {176 status: "error",177 rows: [],178 error: error instanceof Error ? error.message : String(error),179 };180 this.#emit();181 }182183 #updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void {184 this.#environment.updateDiagnostics(update);185 }186187 #emit(): void {188 for (const listener of this.#listeners) listener();189 }190}191192/** A retained reference to a shared live-query store. Release it after its consumer unsubscribes. */193export type LiveQueryLease<T extends TableRow> = {194 store: LiveQueryStore<T>;195 release(): void;196};197198type RegistryEntry = {199 store: LiveQueryStore<TableRow>;200 leases: number;201};202203/** Registry that keys stores by schema identity and stable serialized query plan. */204export class LiveQueryRegistry {205 readonly #environment: LiveQueryEnvironment;206 readonly #schemas = new WeakMap<object, Map<string, RegistryEntry>>();207 readonly #stores = new Set<LiveQueryStore<TableRow>>();208209 /** Creates an isolated registry over the supplied runtime environment. */210 constructor(environment: LiveQueryEnvironment) {211 this.#environment = environment;212 }213214 /** Acquires a shared store and returns an idempotent ownership release. */215 acquire<T extends TableRow>(query: QueryBuilder<T>): LiveQueryLease<T> {216 const schema = query._schema as object;217 const key = liveQueryKey(query);218 let entries = this.#schemas.get(schema);219 if (!entries) {220 entries = new Map();221 this.#schemas.set(schema, entries);222 }223 let entry = entries.get(key);224 if (!entry) {225 const store = new LiveQueryStore(226 query,227 this.#environment,228 () => this.#evictIfUnused(entries!, key),229 );230 entry = { store: store as LiveQueryStore<TableRow>, leases: 0 };231 entries.set(key, entry);232 this.#stores.add(entry.store);233 }234 entry.leases += 1;235 let retained = true;236 return {237 store: entry.store as LiveQueryStore<T>,238 release: () => {239 if (!retained) return;240 retained = false;241 entry!.leases -= 1;242 this.#evictIfUnused(entries!, key);243 },244 };245 }246247 /** Disposes every retained store, including active stores, during hot replacement. */248 dispose(): void {249 for (const store of this.#stores) store.dispose();250 this.#stores.clear();251 }252253 #evictIfUnused(entries: Map<string, RegistryEntry>, key: string): void {254 const entry = entries.get(key);255 if (!entry || entry.leases > 0 || entry.store.consumerCount > 0) return;256 entries.delete(key);257 this.#stores.delete(entry.store);258 entry.store.dispose();259 }260}261262const defaultRegistry = new LiveQueryRegistry({263 getDb: async () => (await getRuntime()).db,264 subscribeRuntimeRecreation(listener) {265 globalThis.addEventListener(runtimeRecreatedEvent, listener);266 return () => globalThis.removeEventListener(runtimeRecreatedEvent, listener);267 },268 subscribeSharedKeyring,269 updateDiagnostics,270});271272function updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void {273 updateRuntimeDiagnostics(update);274}275276/** Acquires the package-wide shared store for an arbitrary typed query. */277export function acquireLiveQuery<T extends TableRow>(278 query: QueryBuilder<T>,279): LiveQueryLease<T> {280 return defaultRegistry.acquire(query);281}282283import.meta.hot?.dispose(() => defaultRegistry.dispose());