1import type { Db, QueryBuilder } from "jazz-tools";2import type { RuntimeDiagnostics } from "./diagnostics.ts";3import { getRuntime, runtimeRecreatedEvent, updateRuntimeDiagnostics } from "./runtime.ts";4import type { TableRow } from "./table-store.ts";56/** Honest read state for an arbitrary typed Jazz query. */7export type LiveQuerySnapshot<T extends TableRow> = {8 status: "loading" | "ready" | "error";9 rows: T[];10 error: string | null;11};1213type Listener = () => void;14type QuerySubscriptionDb = Pick<Db, "subscribeAll">;1516/** Runtime seams used by a live-query registry. Exported for deterministic framework tests. */17export type LiveQueryEnvironment = {18 getDb(): Promise<QuerySubscriptionDb>;19 subscribeRuntimeRecreation(listener: () => void): () => void;20 updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void;21};2223function 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}3233/** Returns the stable serialized plan used to share equivalent query stores. */34export function liveQueryKey<T extends TableRow>(query: QueryBuilder<T>): string {35 return JSON.stringify(canonicalJson(JSON.parse(query._build())));36}3738/** A shared framework-neutral reactive store for one typed Jazz query. */39export 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;4950 /** 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 }6061 /** Returns the current immutable query snapshot. */62 getSnapshot = (): LiveQuerySnapshot<T> => this.#snapshot;6364 /** 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 };8485 /** Number of mounted framework-neutral consumers currently sharing this store. */86 get consumerCount(): number {87 return this.#listeners.size;88 }8990 /** 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 }101102 #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 }112113 #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 }121122 #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 }146147 #stop(): void {148 this.#generation += 1;149 this.#closeVendorSubscription();150 this.#stopRuntimeRecreation?.();151 this.#stopRuntimeRecreation = null;152 }153154 #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 }163164 #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 }173174 #updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void {175 this.#environment.updateDiagnostics(update);176 }177178 #emit(): void {179 for (const listener of this.#listeners) listener();180 }181}182183/** A retained reference to a shared live-query store. Release it after its consumer unsubscribes. */184export type LiveQueryLease<T extends TableRow> = {185 store: LiveQueryStore<T>;186 release(): void;187};188189type RegistryEntry = {190 store: LiveQueryStore<TableRow>;191 leases: number;192};193194/** Registry that keys stores by schema identity and stable serialized query plan. */195export class LiveQueryRegistry {196 readonly #environment: LiveQueryEnvironment;197 readonly #schemas = new WeakMap<object, Map<string, RegistryEntry>>();198 readonly #stores = new Set<LiveQueryStore<TableRow>>();199200 /** Creates an isolated registry over the supplied runtime environment. */201 constructor(environment: LiveQueryEnvironment) {202 this.#environment = environment;203 }204205 /** 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 }237238 /** 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 }243244 #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}252253const 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});261262function updateDiagnostics(update: (diagnostics: RuntimeDiagnostics) => void): void {263 updateRuntimeDiagnostics(update);264}265266/** Acquires the package-wide shared store for an arbitrary typed query. */267export function acquireLiveQuery<T extends TableRow>(268 query: QueryBuilder<T>,269): LiveQueryLease<T> {270 return defaultRegistry.acquire(query);271}272273import.meta.hot?.dispose(() => defaultRegistry.dispose());