1import type { Db, QueryBuilder } from "jazz-tools";2import { schema as s } from "jazz-tools";3import { createDiagnostics } from "./diagnostics.ts";4import {5 type LiveQueryEnvironment,6 LiveQueryRegistry,7 type LiveQuerySnapshot,8} from "./live-query-store.ts";9import { assert, assertCount } from "./test-assert.ts";1011type Row = { id: string; title: string };1213function 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}2324class FakeDb {25 subscribeCalls = 0;26 unsubscribeCalls = 0;27 throwOnSubscribe: Error | null = null;28 readonly callbacks = new Set<(rows: Row[]) => void>();2930 value(): Pick<Db, "subscribeAll"> {31 return this as unknown as Pick<Db, "subscribeAll">;32 }3334 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 }5051 publish(rows: Row[]): void {52 for (const callback of this.callbacks) callback(rows);53 }54}5556function 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}7778async function flush(): Promise<void> {79 await Promise.resolve();80 await Promise.resolve();81}8283Deno.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");99100 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");111112 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");119120 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});128129Deno.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});137138Deno.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");151152 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});162163Deno.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});177178Deno.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");199200 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");212213 stop();214 lease.release();215});216217Deno.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});231232Deno.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});