From f157c599a15846e49cc97f6ad4949325a23a661b Mon Sep 17 00:00:00 2001 From: Ivan Vasilov Date: Wed, 17 Jun 2026 12:40:32 +0200 Subject: [PATCH 1/3] Initial work on adding filters to realtime subscriptions. --- src/db.ts | 87 +++++----- src/realtime.ts | 150 ++++++++++++++++ tests/realtime-listeners.test.ts | 104 ------------ tests/realtime.test.ts | 283 +++++++++++++++++++++++++++++++ tests/test.utils.ts | 58 +++++++ 5 files changed, 531 insertions(+), 151 deletions(-) create mode 100644 src/realtime.ts delete mode 100644 tests/realtime-listeners.test.ts create mode 100644 tests/realtime.test.ts diff --git a/src/db.ts b/src/db.ts index 50d17ac..8440931 100644 --- a/src/db.ts +++ b/src/db.ts @@ -13,6 +13,7 @@ import { supabaseQueryFn, } from "./functions" import { getQueryClient } from "./query-client" +import { attachSupabaseListeners, buildRealtimeFilters } from "./realtime" type GenericPostgrestFilterBuilder = PostgrestFilterBuilder @@ -38,6 +39,8 @@ interface SupabaseCollectionOptions { interface TableEntry { collectionRef: Collection | null realtimeChannel: ReturnType | null + /** Serialized set of Realtime filters the current channel was subscribed with */ + realtimeFiltersKey: string | null supabase: SupabaseClient } @@ -62,16 +65,44 @@ const ensureQueryCacheSubscription = (queryClient: QueryClient) => { type: "active", }) - if (queries.length > 0 && !entry.realtimeChannel && entry.collectionRef) { - entry.realtimeChannel = attachSupabaseListeners( - entry.supabase, - tableName, - entry.collectionRef - ) - } else if (queries.length === 0 && entry.realtimeChannel) { + // No active queries: tear down any existing subscription. + if (queries.length === 0) { + if (entry.realtimeChannel) { + entry.supabase.removeChannel(entry.realtimeChannel) + entry.realtimeChannel = null + entry.realtimeFiltersKey = null + } + continue + } + + if (!entry.collectionRef) { + continue + } + + // Derive the Realtime filters from the WHERE clause of every active query + // so the subscription only receives changes that those queries care about. + const whereExpressions = queries.map( + (query) => query.meta?.loadSubsetOptions?.where + ) + const filters = buildRealtimeFilters(whereExpressions) + const filtersKey = JSON.stringify(filters) + + // Reuse the existing channel when the set of filters hasn't changed. + if (entry.realtimeChannel && entry.realtimeFiltersKey === filtersKey) { + continue + } + + // Filters changed (or no channel yet): (re)subscribe with the new filters. + if (entry.realtimeChannel) { entry.supabase.removeChannel(entry.realtimeChannel) - entry.realtimeChannel = null } + entry.realtimeChannel = attachSupabaseListeners( + entry.supabase, + tableName, + entry.collectionRef, + filters + ) + entry.realtimeFiltersKey = filtersKey } }) } @@ -90,6 +121,7 @@ const registerTable = ( supabase, collectionRef: null, realtimeChannel: null, + realtimeFiltersKey: null, }) } @@ -164,42 +196,3 @@ export const supabaseCollectionOptions = ({ }, } } - -export const attachSupabaseListeners = < - T extends object, - TKey extends string | number, ->( - supabase: SupabaseClient, - tableName: string, - collection: Collection -): ReturnType | null => { - if (!supabase.channel) { - console.log("Server supabase doesn't have a channel") - return null - } - - const channel = supabase.channel(tableName) - channel - .on( - "postgres_changes", - { event: "*", schema: "public", table: tableName }, - (payload) => { - // Realtime events can replay or race the initial PostgREST fetch, so - // an "INSERT" may already be present and an "UPDATE" may not be yet. - // Upsert handles both directions without throwing. - if (payload.eventType === "INSERT") { - collection.utils.writeUpsert(payload.new) - } else if (payload.eventType === "UPDATE") { - collection.utils.writeUpsert(payload.new) - } else if (payload.eventType === "DELETE") { - const id = collection.getKeyFromItem(payload.old as T) - if (collection.has(id)) { - collection.utils.writeDelete(id) - } - } - } - ) - .subscribe() - - return channel -} diff --git a/src/realtime.ts b/src/realtime.ts new file mode 100644 index 0000000..972e206 --- /dev/null +++ b/src/realtime.ts @@ -0,0 +1,150 @@ +import type { + RealtimePostgresChangesFilter, + RealtimePostgresChangesPayload, + SupabaseClient, +} from "@supabase/supabase-js" +import { + type Collection, + extractSimpleComparisons, + type LoadSubsetOptions, + type SimpleComparison, +} from "@tanstack/db" + +type WhereExpression = LoadSubsetOptions["where"] + +/** + * Maps TanStack DB comparison operators to the operators supported by Supabase + * Realtime postgres_changes filters. Realtime only supports a single filter on + * a single column per subscription, using one of these operators. + * @see https://supabase.com/docs/guides/realtime/postgres-changes#available-filters + */ +const REALTIME_OPERATORS: Record = { + eq: "eq", + not_eq: "neq", + gt: "gt", + gte: "gte", + lt: "lt", + lte: "lte", + in: "in", +} + +/** + * Converts a single TanStack DB comparison into a Supabase Realtime filter + * string (`column=operator.value`). Returns `null` when the comparison cannot + * be expressed as a Realtime filter (unsupported operator or missing column). + */ +const toRealtimeFilter = (comparison: SimpleComparison): string | null => { + const operator = REALTIME_OPERATORS[comparison.operator] + if (!operator) { + return null + } + + const column = comparison.field?.join(".") + if (!column) { + return null + } + + if (operator === "in") { + const values = Array.isArray(comparison.value) + ? comparison.value + : [comparison.value] + return `${column}=in.(${values.join(",")})` + } + + return `${column}=${operator}.${comparison.value}` +} + +/** + * Builds the set of Realtime filter strings for a table from the WHERE + * expressions of its active queries. + * + * Realtime only supports a single comparison per subscription, so a query is + * only translated into a filter when its WHERE clause is exactly one supported + * comparison. Any query that cannot be represented (no filter, multiple + * conditions, or an unsupported operator) forces a catch-all subscription, + * represented by a `null` entry, that receives every change for the table. + */ +export const buildRealtimeFilters = ( + whereExpressions: Array +): Array => { + const filters = new Set() + + for (const where of whereExpressions) { + let comparisons: Array + try { + comparisons = extractSimpleComparisons(where) + } catch { + // extractSimpleComparisons throws on expressions it can't represent as + // simple comparisons (e.g. or(), like). Fall back to receiving every + // change for the table. + return [null] + } + + if (comparisons.length !== 1) { + // No filter or a composite filter cannot be expressed in Realtime, so we + // must receive every change for the table. + return [null] + } + + const filter = toRealtimeFilter(comparisons[0]) + if (filter === null) { + return [null] + } + filters.add(filter) + } + + return filters.size > 0 ? Array.from(filters) : [null] +} + +/** + * Subscribes to Supabase Realtime changes for a table and writes inserts, + * updates, and deletes into the collection. One listener is registered per + * provided filter so the union of the active queries' filters is covered. + */ +export const attachSupabaseListeners = < + T extends Record, + TKey extends string | number, +>( + supabase: SupabaseClient, + tableName: string, + collection: Collection, + filters: Array = [null] +): ReturnType | null => { + if (!supabase.channel) { + return null + } + + const channel = supabase.channel(tableName) + + const handlePayload = (payload: RealtimePostgresChangesPayload) => { + if (payload.eventType === "INSERT") { + // Realtime events can replay or race the initial PostgREST fetch, so + // an "INSERT" may already be present and an "UPDATE" may not be yet. + // Upsert handles both directions without throwing. + collection.utils.writeUpsert(payload.new) + } else if (payload.eventType === "UPDATE") { + collection.utils.writeUpsert(payload.new) + } else if (payload.eventType === "DELETE") { + const id = collection.getKeyFromItem(payload.old as T) + if (collection.has(id)) { + collection.utils.writeDelete(id) + } + } + } + + for (const filter of filters) { + const changesFilter: RealtimePostgresChangesFilter<"*"> = { + event: "*", + schema: "public", + table: tableName, + } + if (filter) { + changesFilter.filter = filter + } + channel.on("postgres_changes", changesFilter, handlePayload) + } + + channel.subscribe() + + return channel +} diff --git a/tests/realtime-listeners.test.ts b/tests/realtime-listeners.test.ts deleted file mode 100644 index 7d1908c..0000000 --- a/tests/realtime-listeners.test.ts +++ /dev/null @@ -1,104 +0,0 @@ -import type { SupabaseClient } from "@supabase/supabase-js" -import { afterEach, describe, expect, test } from "vitest" -import { attachSupabaseListeners } from "../src/db" -import { - createMockedUsersCollection, - createMockFetch, - queryResult, -} from "./test.utils" - -type RealtimePayload = any - -describe("attachSupabaseListeners", () => { - let collection: ReturnType - - afterEach(() => { - collection?.cleanup() - }) - - async function setup() { - const captured: Array<(payload: RealtimePayload) => void> = [] - const fakeSupabase = { - channel: () => { - const stub = { - on: ( - _event: string, - _filter: unknown, - cb: (payload: RealtimePayload) => void - ) => { - captured.push(cb) - return stub - }, - subscribe: () => stub, - } - return stub - }, - } as unknown as SupabaseClient - - const mockFetch = createMockFetch() - collection = createMockedUsersCollection(mockFetch) - // Preload from the initial PostgREST fetch, mirroring how the collection - // is populated before any realtime event can race or replay against it. - await queryResult((q) => q.from({ user: collection })) - - attachSupabaseListeners(fakeSupabase, "users", collection) - - const [handler] = captured - if (!handler) { - throw new Error("expected attachSupabaseListeners to register a callback") - } - return handler - } - - test("INSERT for an already-present key with changed data does not throw", async () => { - const handler = await setup() - // Establish a row as already-present via a realtime INSERT, matching how - // a real row (numeric id, schema-valid) would already be synced. - const row = { id: 501, name: "Dana", email: "dana@test.com", active: true } - handler({ eventType: "INSERT", new: row }) - expect(collection.get(String(row.id))?.name).toBe("Dana") - - // Realtime replayed/duplicated the same INSERT with changed data: this is - // the exact shape of the unhandled `CollectionOperationError` from CI. - expect(() => - handler({ - eventType: "INSERT", - new: { ...row, name: "Dana Prime" }, - }) - ).not.toThrow() - - expect(collection.get(String(row.id))?.name).toBe("Dana Prime") - }) - - test("UPDATE for a key not present in the collection does not throw", async () => { - const handler = await setup() - const newRow = { - id: 777, - name: "Charlie", - email: "charlie@test.com", - active: true, - } - - expect(() => - handler({ - eventType: "UPDATE", - new: newRow, - }) - ).not.toThrow() - - expect(collection.get(String(newRow.id))?.name).toBe("Charlie") - }) - - test("DELETE for a key not present does not throw", async () => { - const handler = await setup() - - expect(() => - handler({ - eventType: "DELETE", - old: { id: "user_missing" }, - }) - ).not.toThrow() - - expect(collection.has("user_missing")).toBe(false) - }) -}) diff --git a/tests/realtime.test.ts b/tests/realtime.test.ts new file mode 100644 index 0000000..e024ebb --- /dev/null +++ b/tests/realtime.test.ts @@ -0,0 +1,283 @@ +import { + and, + type Collection, + createCollection, + eq, + gt, + gte, + inArray, + isNull, + liveQueryCollectionOptions, + lt, + lte, + not, +} from "@tanstack/db" +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest" +import { buildRealtimeFilters } from "../src/realtime" +import { + createMockChannel, + createMockFetch, + createRealtimeUsersCollection, + type MockChannel, +} from "./test.utils" + +// Collections created during a test, torn down afterwards so their query +// observers are removed (and the realtime channel detached). +let createdCollections: Array> = [] + +function track>(collection: T): T { + createdCollections.push(collection) + return collection +} + +beforeEach(() => { + createdCollections = [] +}) + +afterEach(() => { + // Tear down in reverse creation order so dependent live queries are cleaned + // up before the source collections they depend on. + for (const collection of [...createdCollections].reverse()) { + collection.cleanup() + } +}) + +type QueryFn = Parameters[0]["query"] +type RealtimeCollection = ReturnType< + typeof createRealtimeUsersCollection +>["collection"] + +// Runs a live query against a fresh realtime users collection and waits for the +// realtime channel to be attached, then returns the recording mock channel. +async function captureChannel( + buildQuery: (collection: RealtimeCollection) => QueryFn +) { + const mockFetch = createMockFetch() + const mockChannel = createMockChannel() + const { collection } = createRealtimeUsersCollection(mockFetch, mockChannel) + track(collection) + + const opts = liveQueryCollectionOptions({ query: buildQuery(collection) }) + const live = track( + createCollection(opts as Extract) + ) + + await live.preload() + await live.toArrayWhenReady() + await vi.waitFor(() => expect(mockChannel.on).toHaveBeenCalled()) + + return { mockChannel, collection } +} + +// Maps the recorded postgres_changes listeners to their filter strings (null +// when the listener is a catch-all with no `filter`). +function realtimeFilters(mockChannel: MockChannel): Array { + return mockChannel.onCalls.map((call) => call.config.filter ?? null) +} + +describe("realtime filter propagation", () => { + describe("supported single-comparison filters", () => { + test("eq on a number column", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => eq(user.id, 1)) + ) + expect(realtimeFilters(mockChannel)).toEqual(["id=eq.1"]) + }) + + test("eq on a boolean column", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q + .from({ user: collection }) + .where(({ user }) => eq(user.active, true)) + ) + expect(realtimeFilters(mockChannel)).toEqual(["active=eq.true"]) + }) + + test("gt", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => gt(user.id, 5)) + ) + expect(realtimeFilters(mockChannel)).toEqual(["id=gt.5"]) + }) + + test("gte", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => gte(user.id, 5)) + ) + expect(realtimeFilters(mockChannel)).toEqual(["id=gte.5"]) + }) + + test("lt", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => lt(user.id, 10)) + ) + expect(realtimeFilters(mockChannel)).toEqual(["id=lt.10"]) + }) + + test("lte", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => lte(user.id, 10)) + ) + expect(realtimeFilters(mockChannel)).toEqual(["id=lte.10"]) + }) + + test("inArray maps to in.(...)", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q + .from({ user: collection }) + .where(({ user }) => inArray(user.id, [1, 2, 3])) + ) + expect(realtimeFilters(mockChannel)).toEqual(["id=in.(1,2,3)"]) + }) + + test("not(eq) maps to neq", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q + .from({ user: collection }) + .where(({ user }) => not(eq(user.active, false))) + ) + expect(realtimeFilters(mockChannel)).toEqual(["active=neq.false"]) + }) + }) + + describe("catch-all subscriptions (no filter)", () => { + test("no WHERE clause", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => q.from({ user: collection }) + ) + expect(realtimeFilters(mockChannel)).toEqual([null]) + }) + + test("isNull has no Realtime operator", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => isNull(user.name)) + ) + expect(realtimeFilters(mockChannel)).toEqual([null]) + }) + + test("composite AND cannot be a single Realtime filter", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q + .from({ user: collection }) + .where(({ user }) => and(eq(user.active, true), gt(user.id, 5))) + ) + expect(realtimeFilters(mockChannel)).toEqual([null]) + }) + }) + + describe("payload routing into the collection", () => { + const row = { + id: 99, + name: "Zed", + email: "zed@test.com", + active: true, + } + + test("INSERT writes the new row", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }) + ) + const { handler } = mockChannel.onCalls[0] + + handler({ eventType: "INSERT", new: row, old: {} }) + + const key = collection.getKeyFromItem(row) + expect(collection.has(key)).toBe(true) + }) + + test("UPDATE writes the updated row", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }) + ) + const { handler } = mockChannel.onCalls[0] + + handler({ eventType: "INSERT", new: row, old: {} }) + handler({ + eventType: "UPDATE", + new: { ...row, name: "Updated" }, + old: {}, + }) + + const key = collection.getKeyFromItem(row) + expect(collection.get(key)?.name).toBe("Updated") + }) + + test("INSERT for an already-present key with changed data does not throw", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }) + ) + const { handler } = mockChannel.onCalls[0] + + // Realtime replayed/duplicated the same INSERT with changed data: this + // is the exact shape of an unhandled write-conflict error. + handler({ eventType: "INSERT", new: row, old: {} }) + expect(() => + handler({ + eventType: "INSERT", + new: { ...row, name: "Zed Prime" }, + old: {}, + }) + ).not.toThrow() + + const key = collection.getKeyFromItem(row) + expect(collection.get(key)?.name).toBe("Zed Prime") + }) + + test("UPDATE for a key not present in the collection does not throw", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }) + ) + const { handler } = mockChannel.onCalls[0] + + expect(() => + handler({ eventType: "UPDATE", new: row, old: {} }) + ).not.toThrow() + + const key = collection.getKeyFromItem(row) + expect(collection.get(key)?.name).toBe("Zed") + }) + + test("DELETE removes the row, and is a no-op for an unknown key", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }) + ) + const { handler } = mockChannel.onCalls[0] + + handler({ eventType: "INSERT", new: row, old: {} }) + const key = collection.getKeyFromItem(row) + expect(collection.has(key)).toBe(true) + + handler({ eventType: "DELETE", new: {}, old: row }) + expect(collection.has(key)).toBe(false) + + // A delete for a key the collection never had must not throw. + expect(() => + handler({ + eventType: "DELETE", + new: {}, + old: { id: 12_345, name: "", email: "", active: false }, + }) + ).not.toThrow() + }) + }) + + // `or(...)` (and other unsupported expressions) cannot be driven through the + // live-query path because the query's own supabaseQueryFn calls the same + // throwing extractSimpleComparisons. Cover the defensive fallback directly. + describe("buildRealtimeFilters falls back instead of throwing", () => { + test("an unsupported expression yields a catch-all", () => { + const orExpression = { type: "func", name: "or", args: [] } as any + expect(buildRealtimeFilters([orExpression])).toEqual([null]) + }) + }) +}) diff --git a/tests/test.utils.ts b/tests/test.utils.ts index 51d43ed..258734a 100644 --- a/tests/test.utils.ts +++ b/tests/test.utils.ts @@ -1,5 +1,6 @@ import { createClient } from "@supabase/supabase-js" import { createCollection, liveQueryCollectionOptions } from "@tanstack/db" +import { QueryClient } from "@tanstack/query-core" import { expect, vi } from "vitest" import { z } from "zod" import { supabaseCollectionOptions } from "../src/index" @@ -122,6 +123,63 @@ export function createMockedTodosCollection(mockFetch: typeof fetch) { ) } +// --- Realtime mock infrastructure --- + +type MockChannelOnCall = { + type: string + config: { event: string; schema: string; table: string; filter?: string } + handler: (payload: any) => void +} + +export type MockChannel = { + onCalls: MockChannelOnCall[] + on: ReturnType + subscribe: ReturnType +} + +export function createMockChannel(): MockChannel { + const onCalls: MockChannelOnCall[] = [] + const channel = { + onCalls, + on: vi.fn((type: string, config: any, handler: (payload: any) => void) => { + onCalls.push({ type, config, handler }) + return channel + }), + subscribe: vi.fn(() => channel), + } + return channel +} + +export function createRealtimeUsersCollection( + mockFetch: typeof fetch, + mockChannel: MockChannel +) { + // A fresh QueryClient keeps the module-level realtime registry in db.ts + // isolated per test. + const queryClient = new QueryClient() + const supabase = createClient(SUPABASE_URL, SUPABASE_KEY, { + global: { fetch: mockFetch }, + }) + // The real client would open a live WebSocket, so stub the realtime surface. + supabase.channel = vi.fn( + () => mockChannel + ) as unknown as typeof supabase.channel + supabase.removeChannel = vi.fn() as unknown as typeof supabase.removeChannel + + const collection = createCollection( + supabaseCollectionOptions({ + tableName: "users", + keys: ["id"], + schema: usersSchema, + supabase, + realtime: true, + queryClient, + }) + ) + + return { collection, supabase, queryClient } +} + // --- Query helpers --- export async function queryResult( From 2e17006034ba2561e0f0d999ec06fd9838b7c3a4 Mon Sep 17 00:00:00 2001 From: Ivan Vasilov Date: Fri, 28 Aug 2026 08:49:20 +0200 Subject: [PATCH 2/3] Add more filters to realtime. Fix various bugs. --- src/db.ts | 55 ++++++- src/functions.ts | 98 ++++++++++--- src/realtime.ts | 268 +++++++++++++++++++++++++++------- tests/index.test.ts | 58 ++++++++ tests/realtime.test.ts | 316 ++++++++++++++++++++++++++++++++++++----- tests/test.utils.ts | 55 ++++++- 6 files changed, 724 insertions(+), 126 deletions(-) diff --git a/src/db.ts b/src/db.ts index 8440931..a31c204 100644 --- a/src/db.ts +++ b/src/db.ts @@ -41,9 +41,25 @@ interface TableEntry { realtimeChannel: ReturnType | null /** Serialized set of Realtime filters the current channel was subscribed with */ realtimeFiltersKey: string | null + /** Resolves once the current channel finished subscribing (or gave up) */ + realtimeSubscribed: Promise | null supabase: SupabaseClient } +/** + * Channel topics are namespaced and numbered because `supabase.channel()` + * returns the *existing* channel for a topic that is already registered, and + * subscribing to an already-joined channel throws. Reusing the table name would + * hand back the channel currently being torn down, and could collide with a + * channel the application opened itself or with another QueryClient sharing the + * same Supabase client — so the counter is module-level, not per table. + */ +let channelCount = 0 +const nextChannelTopic = (tableName: string) => { + channelCount += 1 + return `supabase-tanstack-db:${tableName}:${channelCount}` +} + // Per-QueryClient registry of table entries, with a single cache subscription per client const queryClientRegistries = new Map>() @@ -71,6 +87,7 @@ const ensureQueryCacheSubscription = (queryClient: QueryClient) => { entry.supabase.removeChannel(entry.realtimeChannel) entry.realtimeChannel = null entry.realtimeFiltersKey = null + entry.realtimeSubscribed = null } continue } @@ -92,17 +109,31 @@ const ensureQueryCacheSubscription = (queryClient: QueryClient) => { continue } - // Filters changed (or no channel yet): (re)subscribe with the new filters. - if (entry.realtimeChannel) { - entry.supabase.removeChannel(entry.realtimeChannel) - } - entry.realtimeChannel = attachSupabaseListeners( + // Filters changed (or no channel yet): subscribe with the new filters. + const previousChannel = entry.realtimeChannel + const subscription = attachSupabaseListeners( entry.supabase, + nextChannelTopic(tableName), tableName, entry.collectionRef, filters ) - entry.realtimeFiltersKey = filtersKey + entry.realtimeChannel = subscription?.channel ?? null + entry.realtimeFiltersKey = subscription ? filtersKey : null + entry.realtimeSubscribed = subscription?.subscribed ?? null + + // Keep the previous channel listening until its replacement is + // subscribed, so no change slips through while the swap is in flight. + if (previousChannel) { + const removePrevious = () => { + entry.supabase.removeChannel(previousChannel) + } + if (subscription) { + subscription.subscribed.then(removePrevious, removePrevious) + } else { + removePrevious() + } + } } }) } @@ -122,6 +153,7 @@ const registerTable = ( collectionRef: null, realtimeChannel: null, realtimeFiltersKey: null, + realtimeSubscribed: null, }) } @@ -172,7 +204,16 @@ export const supabaseCollectionOptions = ({ schema, queryKey: (ctx) => subsetOptionsToQueryKey(tableName, ctx), syncMode: "on-demand", - queryFn: (ctx) => supabaseQueryFn(supabase, tableName, ctx), + queryFn: async (ctx) => { + // The channel is attached when the query's observer is added, which + // happens before this runs. Waiting for it means a row written between + // the fetch and the subscription arrives over Realtime instead of being + // missed by both. + if (entry?.realtimeSubscribed) { + await entry.realtimeSubscribed + } + return await supabaseQueryFn(supabase, tableName, ctx) + }, onInsert: (ctx) => supabaseOnInsert(supabase, tableName, ctx), onUpdate: (ctx) => supabaseOnUpdate(supabase, tableName, where, ctx), onDelete: (ctx) => supabaseOnDelete(supabase, tableName, where, ctx), diff --git a/src/functions.ts b/src/functions.ts index 80bc68f..fb6d752 100644 --- a/src/functions.ts +++ b/src/functions.ts @@ -14,29 +14,80 @@ import { import type { QueryClient, QueryMeta } from "@tanstack/query-core" import { CLIENT_INFO, CLIENT_INFO_HEADER } from "./request-headers" +/** `not(...)` comparisons reach us as the operator with this prefix. */ +const NEGATION_PREFIX = "not_" + +/** Comparison operators that map onto a PostgREST filter of the same name. */ +const COMPARISON_OPERATORS = new Set(["eq", "gt", "gte", "lt", "lte"]) + +/** + * postgrest-js interpolates filter values into the URL as-is, and + * `Date.prototype.toString()` produces something Postgres cannot cast to a + * timestamp. Rendering it the same way the Realtime filters do keeps the + * server query and the subscription in agreement. + */ +const toFilterValue = (value: unknown) => + value instanceof Date ? value.toISOString() : value + +const applyComparison = ( + baseQuery: PostgrestFilterBuilder, + column: string, + operator: string, + value: unknown +) => { + if (operator === "gt") { + return baseQuery.gt(column, value) + } + if (operator === "gte") { + return baseQuery.gte(column, value) + } + if (operator === "lt") { + return baseQuery.lt(column, value) + } + if (operator === "lte") { + return baseQuery.lte(column, value) + } + return baseQuery.eq(column, value) +} + const buildQuery = ( baseQuery: PostgrestFilterBuilder, filter: SimpleComparison ) => { - if (filter.operator === "eq") { - baseQuery = baseQuery.eq(filter.field?.join("."), filter.value) - } else if (filter.operator === "gt") { - baseQuery = baseQuery.gt(filter.field?.join("."), filter.value) - } else if (filter.operator === "gte") { - baseQuery = baseQuery.gte(filter.field?.join("."), filter.value) - } else if (filter.operator === "lt") { - baseQuery = baseQuery.lt(filter.field?.join("."), filter.value) - } else if (filter.operator === "lte") { - baseQuery = baseQuery.lte(filter.field?.join("."), filter.value) - } else if (filter.operator === "in") { - baseQuery = baseQuery.in(filter.field?.join("."), filter.value) - } else if (filter.operator === "isNull") { - baseQuery = baseQuery.is(filter.field?.join("."), null) - } else if (filter.operator === "not_eq") { - baseQuery = baseQuery.not(filter.field?.join("."), "eq", filter.value) - } else { + const column = filter.field?.join(".") + if (!column) { + return baseQuery + } + + const negated = filter.operator.startsWith(NEGATION_PREFIX) + const operator = negated + ? filter.operator.slice(NEGATION_PREFIX.length) + : filter.operator + + if (operator === "isNull") { + return negated + ? baseQuery.not(column, "is", null) + : baseQuery.is(column, null) + } + + if (operator === "in") { + const values = Array.isArray(filter.value) + ? filter.value.map(toFilterValue) + : filter.value + return negated + ? baseQuery.notIn(column, values) + : baseQuery.in(column, values) + } + + if (!COMPARISON_OPERATORS.has(operator)) { console.warn(`buildQuery: unsupported operator: ${filter.operator}`) + return baseQuery } + + const value = toFilterValue(filter.value) + return negated + ? baseQuery.not(column, operator, value) + : applyComparison(baseQuery, column, operator, value) } export const subsetOptionsToQueryKey = ( @@ -71,9 +122,10 @@ export const subsetOptionsToQueryKey = ( lte: (field, value) => { return `${field.join(".")}=lte.${value}` }, - not: (field, operator, value) => { - return field - }, + // The single argument is the already-parsed inner condition. Wrapping it + // is what keeps `not(gt(id, 5))` from sharing a cache entry with + // `gt(id, 5)`. + not: (inner) => (inner === null ? null : `not(${inner})`), }, onUnknownOperator: (operator, args) => { console.warn(`Unsupported operator: ${operator}`) @@ -149,9 +201,9 @@ export const supabaseQueryFn = async ( } if (parsed.filters) { - ;[...parsed.filters, ...cursorFilters].forEach((filter) => { - buildQuery(baseQuery, filter) - }) + for (const filter of [...parsed.filters, ...cursorFilters]) { + baseQuery = buildQuery(baseQuery, filter) + } } const { data, error } = await baseQuery diff --git a/src/realtime.ts b/src/realtime.ts index 972e206..28f9bce 100644 --- a/src/realtime.ts +++ b/src/realtime.ts @@ -12,26 +12,109 @@ import { type WhereExpression = LoadSubsetOptions["where"] +type ChangeEvent = "INSERT" | "UPDATE" | "DELETE" + +export type RealtimeSubscription = { + channel: ReturnType + /** + * Resolves once the channel reached a terminal subscription state, or after + * {@link SUBSCRIBE_TIMEOUT_MS} if the server never answers. Never rejects, so + * callers can safely gate work on it without an unreachable Realtime server + * blocking them forever. + */ + subscribed: Promise +} + /** * Maps TanStack DB comparison operators to the operators supported by Supabase - * Realtime postgres_changes filters. Realtime only supports a single filter on - * a single column per subscription, using one of these operators. + * Realtime postgres_changes filters. * @see https://supabase.com/docs/guides/realtime/postgres-changes#available-filters */ const REALTIME_OPERATORS: Record = { eq: "eq", not_eq: "neq", gt: "gt", + not_gt: "not.gt", gte: "gte", + not_gte: "not.gte", lt: "lt", + not_lt: "not.lt", lte: "lte", + not_lte: "not.lte", in: "in", + not_in: "not.in", + isNull: "is", + not_isNull: "not.is", +} + +/** Realtime rejects `in` filters with more than this many values. */ +const MAX_IN_VALUES = 100 + +/** Never let a fetch wait longer than this for the channel to subscribe. */ +const SUBSCRIBE_TIMEOUT_MS = 2000 + +/** + * Characters that carry meaning in a filter string: `,` separates ANDed + * conditions and the `in` list, `(`/`)` delimit that list, and `"`/`\` are the + * quoting characters themselves. + */ +const RESERVED_CHARACTERS = /[,()"\\]/ +const QUOTED_CHARACTERS = /["\\]/g + +const NOT_PREFIX = "not_" + +/** + * Renders a value the way PostgREST expects it inside a filter string, quoting + * strings that contain reserved characters. Returns `null` for values Realtime + * cannot compare against, which forces the caller to fall back to an unfiltered + * subscription rather than send a filter that means something else. + */ +const serializeValue = (value: unknown): string | null => { + if (typeof value === "boolean") { + return String(value) + } + if (typeof value === "number") { + return Number.isFinite(value) ? String(value) : null + } + if (typeof value === "bigint") { + return String(value) + } + if (value instanceof Date) { + return value.toISOString() + } + if (typeof value === "string") { + if (value.length === 0 || RESERVED_CHARACTERS.test(value)) { + return `"${value.replace(QUOTED_CHARACTERS, "\\$&")}"` + } + return value + } + return null +} + +const serializeInValues = (value: unknown): string | null => { + if (!Array.isArray(value)) { + return null + } + // An empty list matches nothing and `in.()` is not valid filter syntax. + if (value.length === 0 || value.length > MAX_IN_VALUES) { + return null + } + + const serialized: Array = [] + for (const entry of value) { + const rendered = serializeValue(entry) + if (rendered === null) { + return null + } + serialized.push(rendered) + } + return `(${serialized.join(",")})` } /** * Converts a single TanStack DB comparison into a Supabase Realtime filter * string (`column=operator.value`). Returns `null` when the comparison cannot - * be expressed as a Realtime filter (unsupported operator or missing column). + * be expressed as a Realtime filter. */ const toRealtimeFilter = (comparison: SimpleComparison): string | null => { const operator = REALTIME_OPERATORS[comparison.operator] @@ -39,30 +122,40 @@ const toRealtimeFilter = (comparison: SimpleComparison): string | null => { return null } - const column = comparison.field?.join(".") - if (!column) { + // Realtime evaluates filters against a single top-level column, so a nested + // field path has no equivalent. + const column = + comparison.field?.length === 1 ? comparison.field[0] : undefined + if (typeof column !== "string" || column.length === 0) { return null } - if (operator === "in") { - const values = Array.isArray(comparison.value) - ? comparison.value - : [comparison.value] - return `${column}=in.(${values.join(",")})` + const baseOperator = comparison.operator.startsWith(NOT_PREFIX) + ? comparison.operator.slice(NOT_PREFIX.length) + : comparison.operator + + if (baseOperator === "isNull") { + return `${column}=${operator}.null` + } + + if (baseOperator === "in") { + const values = serializeInValues(comparison.value) + return values === null ? null : `${column}=${operator}.${values}` } - return `${column}=${operator}.${comparison.value}` + const value = serializeValue(comparison.value) + return value === null ? null : `${column}=${operator}.${value}` } /** * Builds the set of Realtime filter strings for a table from the WHERE * expressions of its active queries. * - * Realtime only supports a single comparison per subscription, so a query is - * only translated into a filter when its WHERE clause is exactly one supported - * comparison. Any query that cannot be represented (no filter, multiple - * conditions, or an unsupported operator) forces a catch-all subscription, - * represented by a `null` entry, that receives every change for the table. + * A query's conditions are ANDed into a single filter string using the + * comma syntax Realtime supports. Any query that cannot be represented (no + * WHERE clause, an unsupported operator, or a value that cannot be rendered + * safely) forces a catch-all subscription, represented by a single `null` + * entry, that receives every change for the table. */ export const buildRealtimeFilters = ( whereExpressions: Array @@ -80,71 +173,144 @@ export const buildRealtimeFilters = ( return [null] } - if (comparisons.length !== 1) { - // No filter or a composite filter cannot be expressed in Realtime, so we - // must receive every change for the table. + if (comparisons.length === 0) { return [null] } - const filter = toRealtimeFilter(comparisons[0]) - if (filter === null) { - return [null] + const conditions: Array = [] + for (const comparison of comparisons) { + const condition = toRealtimeFilter(comparison) + if (condition === null) { + return [null] + } + conditions.push(condition) } - filters.add(filter) + // Sorted so that two queries expressing the same conditions in a different + // order share one subscription. + filters.add(conditions.sort().join(",")) } - return filters.size > 0 ? Array.from(filters) : [null] + return filters.size > 0 ? Array.from(filters).sort() : [null] } /** * Subscribes to Supabase Realtime changes for a table and writes inserts, - * updates, and deletes into the collection. One listener is registered per - * provided filter so the union of the active queries' filters is covered. + * updates, and deletes into the collection. + * + * Filters are applied to INSERT and UPDATE listeners only: a filtered event is + * Realtime telling us the row matches an active query, which is what makes it + * belong in the collection — including a row that an UPDATE moved into that + * query's window. Two extra unfiltered listeners cover what a filtered + * subscription cannot see: + * + * - UPDATE, so a row already in the collection still receives its changes after + * it stops matching the filter instead of going stale. + * - DELETE, because Realtime only delivers filtered delete events for tables + * with `replica identity full`, and delete payloads are just the key anyway. */ export const attachSupabaseListeners = < T extends Record, TKey extends string | number, >( supabase: SupabaseClient, + topic: string, tableName: string, collection: Collection, filters: Array = [null] -): ReturnType | null => { +): RealtimeSubscription | null => { if (!supabase.channel) { return null } - const channel = supabase.channel(tableName) - - const handlePayload = (payload: RealtimePostgresChangesPayload) => { - if (payload.eventType === "INSERT") { - // Realtime events can replay or race the initial PostgREST fetch, so - // an "INSERT" may already be present and an "UPDATE" may not be yet. - // Upsert handles both directions without throwing. - collection.utils.writeUpsert(payload.new) - } else if (payload.eventType === "UPDATE") { - collection.utils.writeUpsert(payload.new) - } else if (payload.eventType === "DELETE") { - const id = collection.getKeyFromItem(payload.old as T) - if (collection.has(id)) { - collection.utils.writeDelete(id) - } - } - } + const channel = supabase.channel(topic) - for (const filter of filters) { - const changesFilter: RealtimePostgresChangesFilter<"*"> = { - event: "*", + const changesFilter = ( + event: TEvent, + filter: string | null + ): RealtimePostgresChangesFilter => { + const config: RealtimePostgresChangesFilter = { + event, schema: "public", table: tableName, } if (filter) { - changesFilter.filter = filter + config.filter = filter + } + return config + } + + // The row matches an active query's filter, so it belongs in the collection + // whether or not we have seen it before. + const handleUpsert = (payload: RealtimePostgresChangesPayload) => { + if (payload.eventType !== "INSERT" && payload.eventType !== "UPDATE") { + return + } + const row = payload.new as T + const id = collection.getKeyFromItem(row) + if (collection.has(id)) { + collection.utils.writeUpdate(row) + } else { + collection.utils.writeInsert(row) + } + } + + // Unfiltered updates arrive for every row in the table, so only rows the + // collection already holds are written — otherwise it would mirror the whole + // table locally. + const handleKnownUpdate = (payload: RealtimePostgresChangesPayload) => { + if (payload.eventType !== "UPDATE") { + return } - channel.on("postgres_changes", changesFilter, handlePayload) + const row = payload.new as T + const id = collection.getKeyFromItem(row) + if (collection.has(id)) { + collection.utils.writeUpdate(row) + } + } + + const handleDelete = (payload: RealtimePostgresChangesPayload) => { + if (payload.eventType !== "DELETE") { + return + } + const id = collection.getKeyFromItem(payload.old as T) + if (collection.has(id)) { + collection.utils.writeDelete(id) + } + } + + for (const filter of filters) { + channel.on("postgres_changes", changesFilter("INSERT", filter), (p) => + handleUpsert(p as RealtimePostgresChangesPayload) + ) + channel.on("postgres_changes", changesFilter("UPDATE", filter), (p) => + handleUpsert(p as RealtimePostgresChangesPayload) + ) } - channel.subscribe() + if (filters.some((filter) => filter !== null)) { + channel.on("postgres_changes", changesFilter("UPDATE", null), (p) => + handleKnownUpdate(p as RealtimePostgresChangesPayload) + ) + } + + channel.on("postgres_changes", changesFilter("DELETE", null), (p) => + handleDelete(p as RealtimePostgresChangesPayload) + ) + + const subscribed = new Promise((resolve) => { + const timeout = setTimeout(resolve, SUBSCRIBE_TIMEOUT_MS) + const settle = () => { + clearTimeout(timeout) + resolve() + } + try { + channel.subscribe(settle) + } catch { + // subscribe() throws synchronously on an already-joined channel. Data + // loading must not be held up by a channel that will never connect. + settle() + } + }) - return channel + return { channel, subscribed } } diff --git a/tests/index.test.ts b/tests/index.test.ts index 633f429..70a3616 100644 --- a/tests/index.test.ts +++ b/tests/index.test.ts @@ -132,6 +132,38 @@ describe("PostgREST query generation", () => { ]) }) + test("WHERE NOT(id > 5)", async () => { + await queryResult((q) => + q + .from({ user: usersCollection }) + .where(({ user }) => not(gt(user.id, 5))) + ) + expectFetchUrls(mockFetch, ["/rest/v1/users?select=*&id=not.gt.5"]) + }) + + test("WHERE NOT(id IN (1, 2))", async () => { + await queryResult((q) => + q + .from({ user: usersCollection }) + .where(({ user }) => not(inArray(user.id, [1, 2]))) + ) + expectFetchUrls(mockFetch, ["/rest/v1/users?select=*&id=not.in.(1,2)"]) + }) + + test("a Date value is sent as ISO 8601", async () => { + await queryResult((q) => + q + .from({ user: usersCollection }) + .where(({ user }) => + gt(user.name, new Date("2026-01-02T03:04:05.000Z") as never) + ) + ) + // `Date.toString()` would produce something Postgres cannot cast. + expectFetchUrls(mockFetch, [ + "/rest/v1/users?select=*&name=gt.2026-01-02T03:04:05.000Z", + ]) + }) + test("WHERE name IS NULL", async () => { await queryResult((q) => q.from({ user: usersCollection }).where(({ user }) => isNull(user.name)) @@ -139,6 +171,32 @@ describe("PostgREST query generation", () => { expectFetchUrls(mockFetch, ["/rest/v1/users?select=*&name=is.null"]) }) + test("WHERE NOT(name IS NULL)", async () => { + await queryResult((q) => + q + .from({ user: usersCollection }) + .where(({ user }) => not(isNull(user.name))) + ) + expectFetchUrls(mockFetch, ["/rest/v1/users?select=*&name=not.is.null"]) + }) + + test("a negated condition does not reuse the cache entry of the plain one", async () => { + await queryResult((q) => + q.from({ user: usersCollection }).where(({ user }) => gt(user.id, 5)) + ) + await queryResult((q) => + q + .from({ user: usersCollection }) + .where(({ user }) => not(gt(user.id, 5))) + ) + // Two fetches, not one: a shared query key would have served the second + // query the first query's rows. + expectFetchUrls(mockFetch, [ + "/rest/v1/users?select=*&id=gt.5", + "/rest/v1/users?select=*&id=not.gt.5", + ]) + }) + test("AND: active = true AND id > 5", async () => { await queryResult((q) => q diff --git a/tests/realtime.test.ts b/tests/realtime.test.ts index e024ebb..2d4beab 100644 --- a/tests/realtime.test.ts +++ b/tests/realtime.test.ts @@ -18,6 +18,9 @@ import { createMockChannel, createMockFetch, createRealtimeUsersCollection, + emit, + filtersFor, + listenersFor, type MockChannel, } from "./test.utils" @@ -47,6 +50,16 @@ type RealtimeCollection = ReturnType< typeof createRealtimeUsersCollection >["collection"] +function startLiveQuery( + collection: RealtimeCollection, + buildQuery: (collection: RealtimeCollection) => QueryFn +) { + const opts = liveQueryCollectionOptions({ query: buildQuery(collection) }) + return track( + createCollection(opts as Extract) + ) +} + // Runs a live query against a fresh realtime users collection and waits for the // realtime channel to be attached, then returns the recording mock channel. async function captureChannel( @@ -57,10 +70,7 @@ async function captureChannel( const { collection } = createRealtimeUsersCollection(mockFetch, mockChannel) track(collection) - const opts = liveQueryCollectionOptions({ query: buildQuery(collection) }) - const live = track( - createCollection(opts as Extract) - ) + const live = startLiveQuery(collection, buildQuery) await live.preload() await live.toArrayWhenReady() @@ -69,10 +79,10 @@ async function captureChannel( return { mockChannel, collection } } -// Maps the recorded postgres_changes listeners to their filter strings (null -// when the listener is a catch-all with no `filter`). -function realtimeFilters(mockChannel: MockChannel): Array { - return mockChannel.onCalls.map((call) => call.config.filter ?? null) +// The filters applied to INSERT listeners — the ones that decide which rows +// belong in the collection. +function insertFilters(mockChannel: MockChannel): Array { + return filtersFor(mockChannel, "INSERT") } describe("realtime filter propagation", () => { @@ -82,7 +92,7 @@ describe("realtime filter propagation", () => { (collection) => (q) => q.from({ user: collection }).where(({ user }) => eq(user.id, 1)) ) - expect(realtimeFilters(mockChannel)).toEqual(["id=eq.1"]) + expect(insertFilters(mockChannel)).toEqual(["id=eq.1"]) }) test("eq on a boolean column", async () => { @@ -92,7 +102,7 @@ describe("realtime filter propagation", () => { .from({ user: collection }) .where(({ user }) => eq(user.active, true)) ) - expect(realtimeFilters(mockChannel)).toEqual(["active=eq.true"]) + expect(insertFilters(mockChannel)).toEqual(["active=eq.true"]) }) test("gt", async () => { @@ -100,7 +110,7 @@ describe("realtime filter propagation", () => { (collection) => (q) => q.from({ user: collection }).where(({ user }) => gt(user.id, 5)) ) - expect(realtimeFilters(mockChannel)).toEqual(["id=gt.5"]) + expect(insertFilters(mockChannel)).toEqual(["id=gt.5"]) }) test("gte", async () => { @@ -108,7 +118,7 @@ describe("realtime filter propagation", () => { (collection) => (q) => q.from({ user: collection }).where(({ user }) => gte(user.id, 5)) ) - expect(realtimeFilters(mockChannel)).toEqual(["id=gte.5"]) + expect(insertFilters(mockChannel)).toEqual(["id=gte.5"]) }) test("lt", async () => { @@ -116,7 +126,7 @@ describe("realtime filter propagation", () => { (collection) => (q) => q.from({ user: collection }).where(({ user }) => lt(user.id, 10)) ) - expect(realtimeFilters(mockChannel)).toEqual(["id=lt.10"]) + expect(insertFilters(mockChannel)).toEqual(["id=lt.10"]) }) test("lte", async () => { @@ -124,7 +134,7 @@ describe("realtime filter propagation", () => { (collection) => (q) => q.from({ user: collection }).where(({ user }) => lte(user.id, 10)) ) - expect(realtimeFilters(mockChannel)).toEqual(["id=lte.10"]) + expect(insertFilters(mockChannel)).toEqual(["id=lte.10"]) }) test("inArray maps to in.(...)", async () => { @@ -134,7 +144,7 @@ describe("realtime filter propagation", () => { .from({ user: collection }) .where(({ user }) => inArray(user.id, [1, 2, 3])) ) - expect(realtimeFilters(mockChannel)).toEqual(["id=in.(1,2,3)"]) + expect(insertFilters(mockChannel)).toEqual(["id=in.(1,2,3)"]) }) test("not(eq) maps to neq", async () => { @@ -144,34 +154,142 @@ describe("realtime filter propagation", () => { .from({ user: collection }) .where(({ user }) => not(eq(user.active, false))) ) - expect(realtimeFilters(mockChannel)).toEqual(["active=neq.false"]) + expect(insertFilters(mockChannel)).toEqual(["active=neq.false"]) }) - }) - describe("catch-all subscriptions (no filter)", () => { - test("no WHERE clause", async () => { + test("isNull maps to is.null", async () => { const { mockChannel } = await captureChannel( - (collection) => (q) => q.from({ user: collection }) + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => isNull(user.name)) ) - expect(realtimeFilters(mockChannel)).toEqual([null]) + expect(insertFilters(mockChannel)).toEqual(["name=is.null"]) }) - test("isNull has no Realtime operator", async () => { + test("not(isNull) maps to not.is.null", async () => { const { mockChannel } = await captureChannel( (collection) => (q) => - q.from({ user: collection }).where(({ user }) => isNull(user.name)) + q + .from({ user: collection }) + .where(({ user }) => not(isNull(user.name))) + ) + expect(insertFilters(mockChannel)).toEqual(["name=not.is.null"]) + }) + + test("not(inArray) maps to not.in", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q + .from({ user: collection }) + .where(({ user }) => not(inArray(user.id, [1, 2]))) + ) + expect(insertFilters(mockChannel)).toEqual(["id=not.in.(1,2)"]) + }) + + test("not(gt) maps to not.gt", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => not(gt(user.id, 5))) ) - expect(realtimeFilters(mockChannel)).toEqual([null]) + expect(insertFilters(mockChannel)).toEqual(["id=not.gt.5"]) }) - test("composite AND cannot be a single Realtime filter", async () => { + test("composite AND becomes a comma-separated filter", async () => { const { mockChannel } = await captureChannel( (collection) => (q) => q .from({ user: collection }) .where(({ user }) => and(eq(user.active, true), gt(user.id, 5))) ) - expect(realtimeFilters(mockChannel)).toEqual([null]) + expect(insertFilters(mockChannel)).toEqual(["active=eq.true,id=gt.5"]) + }) + }) + + describe("value serialization", () => { + test("a string containing a comma is PostgREST-quoted", async () => { + // Unquoted, the comma would read as a second ANDed condition. + const { mockChannel } = await captureChannel( + (collection) => (q) => + q + .from({ user: collection }) + .where(({ user }) => eq(user.name, "Doe, Jane")) + ) + expect(insertFilters(mockChannel)).toEqual(['name=eq."Doe, Jane"']) + }) + + test("a quote inside a value is escaped", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q + .from({ user: collection }) + .where(({ user }) => eq(user.name, 'He said "hi", ok')) + ) + expect(insertFilters(mockChannel)).toEqual([ + 'name=eq."He said \\"hi\\", ok"', + ]) + }) + + test("a plain string is left unquoted", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => eq(user.name, "Zed")) + ) + expect(insertFilters(mockChannel)).toEqual(["name=eq.Zed"]) + }) + + test("an in list of exactly 100 values is still filtered", async () => { + const values = Array.from({ length: 100 }, (_, index) => index) + const { mockChannel } = await captureChannel( + (collection) => (q) => + q + .from({ user: collection }) + .where(({ user }) => inArray(user.id, values)) + ) + expect(insertFilters(mockChannel)).toEqual([ + `id=in.(${values.join(",")})`, + ]) + }) + + test("an in list longer than 100 values falls back to a catch-all", async () => { + const values = Array.from({ length: 101 }, (_, index) => index) + const { mockChannel } = await captureChannel( + (collection) => (q) => + q + .from({ user: collection }) + .where(({ user }) => inArray(user.id, values)) + ) + expect(insertFilters(mockChannel)).toEqual([null]) + }) + }) + + describe("catch-all subscriptions (no filter)", () => { + test("no WHERE clause", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => q.from({ user: collection }) + ) + expect(insertFilters(mockChannel)).toEqual([null]) + }) + + test("no redundant unfiltered UPDATE listener is added", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => q.from({ user: collection }) + ) + expect(filtersFor(mockChannel, "UPDATE")).toEqual([null]) + }) + }) + + describe("listener layout", () => { + test("filters apply to INSERT and UPDATE, never to DELETE", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => eq(user.id, 1)) + ) + + expect(filtersFor(mockChannel, "INSERT")).toEqual(["id=eq.1"]) + // The filtered listener catches rows entering the window; the unfiltered + // one keeps rows already held up to date after they leave it. + expect(filtersFor(mockChannel, "UPDATE")).toEqual(["id=eq.1", null]) + // Realtime only delivers filtered deletes with `replica identity full`. + expect(filtersFor(mockChannel, "DELETE")).toEqual([null]) }) }) @@ -187,29 +305,75 @@ describe("realtime filter propagation", () => { const { mockChannel, collection } = await captureChannel( (c) => (q) => q.from({ user: c }) ) - const { handler } = mockChannel.onCalls[0] - handler({ eventType: "INSERT", new: row, old: {} }) + emit(mockChannel, { eventType: "INSERT", new: row, old: {} }) - const key = collection.getKeyFromItem(row) - expect(collection.has(key)).toBe(true) + expect(collection.has(collection.getKeyFromItem(row))).toBe(true) + }) + + test("INSERT for a row already held overwrites instead of throwing", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }) + ) + + emit(mockChannel, { eventType: "INSERT", new: row, old: {} }) + expect(() => + emit(mockChannel, { + eventType: "INSERT", + new: { ...row, name: "Twice" }, + old: {}, + }) + ).not.toThrow() + + expect(collection.get(collection.getKeyFromItem(row))?.name).toBe("Twice") }) test("UPDATE writes the updated row", async () => { const { mockChannel, collection } = await captureChannel( (c) => (q) => q.from({ user: c }) ) - const { handler } = mockChannel.onCalls[0] - handler({ eventType: "INSERT", new: row, old: {} }) - handler({ + emit(mockChannel, { eventType: "INSERT", new: row, old: {} }) + emit(mockChannel, { eventType: "UPDATE", new: { ...row, name: "Updated" }, old: {}, }) - const key = collection.getKeyFromItem(row) - expect(collection.get(key)?.name).toBe("Updated") + expect(collection.get(collection.getKeyFromItem(row))?.name).toBe( + "Updated" + ) + }) + + test("a matching UPDATE for an unseen row inserts it", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }).where(({ user }) => eq(user.id, 99)) + ) + const [filtered] = listenersFor(mockChannel, "UPDATE") + + // Realtime only delivers this to the filtered listener when the row + // matches, which means the row has moved into the query's window. + expect(() => + filtered.handler({ eventType: "UPDATE", new: row, old: {} }) + ).not.toThrow() + + expect(collection.has(collection.getKeyFromItem(row))).toBe(true) + }) + + test("an unfiltered UPDATE for an unseen row is ignored", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }).where(({ user }) => eq(user.id, 1)) + ) + const catchAll = listenersFor(mockChannel, "UPDATE").find( + (call) => !call.config.filter + ) + + // Would otherwise mirror every row in the table into the collection. + expect(() => + catchAll?.handler({ eventType: "UPDATE", new: row, old: {} }) + ).not.toThrow() + + expect(collection.has(collection.getKeyFromItem(row))).toBe(false) }) test("INSERT for an already-present key with changed data does not throw", async () => { @@ -251,18 +415,17 @@ describe("realtime filter propagation", () => { const { mockChannel, collection } = await captureChannel( (c) => (q) => q.from({ user: c }) ) - const { handler } = mockChannel.onCalls[0] - handler({ eventType: "INSERT", new: row, old: {} }) + emit(mockChannel, { eventType: "INSERT", new: row, old: {} }) const key = collection.getKeyFromItem(row) expect(collection.has(key)).toBe(true) - handler({ eventType: "DELETE", new: {}, old: row }) + emit(mockChannel, { eventType: "DELETE", new: {}, old: row }) expect(collection.has(key)).toBe(false) // A delete for a key the collection never had must not throw. expect(() => - handler({ + emit(mockChannel, { eventType: "DELETE", new: {}, old: { id: 12_345, name: "", email: "", active: false }, @@ -271,6 +434,81 @@ describe("realtime filter propagation", () => { }) }) + describe("subscribing before fetching", () => { + test("the first fetch waits for the channel to be subscribed", async () => { + const mockFetch = createMockFetch() + const mockChannel = createMockChannel({ autoSubscribe: false }) + const { collection } = createRealtimeUsersCollection( + mockFetch, + mockChannel + ) + track(collection) + + const live = startLiveQuery( + collection, + (c) => (q) => q.from({ user: c }).where(({ user }) => eq(user.id, 1)) + ) + const preloaded = live.preload() + + await vi.waitFor(() => expect(mockChannel.on).toHaveBeenCalled()) + // A row written now would be missed by the fetch, so it must not have + // started before the subscription is live. + expect(mockFetch).not.toHaveBeenCalled() + + mockChannel.confirmSubscribed() + await preloaded + await live.toArrayWhenReady() + + expect(mockFetch).toHaveBeenCalled() + }) + }) + + describe("resubscribing when the filter set changes", () => { + test("uses a fresh channel topic and drops the previous channel", async () => { + const mockFetch = createMockFetch() + const channels: Array = [] + const { collection, supabase } = createRealtimeUsersCollection( + mockFetch, + () => { + const channel = createMockChannel() + channels.push(channel) + return channel + } + ) + track(collection) + + const first = startLiveQuery( + collection, + (c) => (q) => q.from({ user: c }).where(({ user }) => eq(user.id, 1)) + ) + await first.preload() + await first.toArrayWhenReady() + await vi.waitFor(() => expect(channels.length).toBe(1)) + + const second = startLiveQuery( + collection, + (c) => (q) => q.from({ user: c }).where(({ user }) => gt(user.id, 5)) + ) + await second.preload() + await second.toArrayWhenReady() + await vi.waitFor(() => expect(channels.length).toBe(2)) + + const topics = ( + supabase.channel as unknown as ReturnType + ).mock.calls.map(([topic]) => topic) + // A repeated topic would hand back the channel being torn down. + expect(new Set(topics).size).toBe(topics.length) + + // The union of both queries' filters is now subscribed. + expect(insertFilters(channels[1])).toEqual(["id=eq.1", "id=gt.5"]) + + // The old channel is only dropped once the replacement is subscribed. + await vi.waitFor(() => + expect(supabase.removeChannel).toHaveBeenCalledWith(channels[0]) + ) + }) + }) + // `or(...)` (and other unsupported expressions) cannot be driven through the // live-query path because the query's own supabaseQueryFn calls the same // throwing extractSimpleComparisons. Cover the defensive fallback directly. diff --git a/tests/test.utils.ts b/tests/test.utils.ts index 258734a..acb4508 100644 --- a/tests/test.utils.ts +++ b/tests/test.utils.ts @@ -135,24 +135,67 @@ export type MockChannel = { onCalls: MockChannelOnCall[] on: ReturnType subscribe: ReturnType + /** Reports SUBSCRIBED when the channel was created with `autoSubscribe: false`. */ + confirmSubscribed: () => void } -export function createMockChannel(): MockChannel { +export function createMockChannel({ + autoSubscribe = true, +}: { + autoSubscribe?: boolean +} = {}): MockChannel { const onCalls: MockChannelOnCall[] = [] - const channel = { + let pending: ((status: string) => void) | undefined + + const channel: MockChannel = { onCalls, on: vi.fn((type: string, config: any, handler: (payload: any) => void) => { onCalls.push({ type, config, handler }) return channel }), - subscribe: vi.fn(() => channel), + // The collection's first fetch waits for the subscription to settle, so the + // mock has to report a status like the real channel does. + subscribe: vi.fn((callback?: (status: string) => void) => { + if (autoSubscribe) { + callback?.("SUBSCRIBED") + } else { + pending = callback + } + return channel + }), + confirmSubscribed: () => { + pending?.("SUBSCRIBED") + pending = undefined + }, } return channel } +/** Returns the postgres_changes listeners registered for a given event. */ +export function listenersFor(mockChannel: MockChannel, event: string) { + return mockChannel.onCalls.filter((call) => call.config.event === event) +} + +/** The filter strings of a given event's listeners (null when unfiltered). */ +export function filtersFor( + mockChannel: MockChannel, + event: string +): Array { + return listenersFor(mockChannel, event).map( + (call) => call.config.filter ?? null + ) +} + +/** Dispatches a payload to every listener registered for its event type. */ +export function emit(mockChannel: MockChannel, payload: any) { + for (const call of listenersFor(mockChannel, payload.eventType)) { + call.handler(payload) + } +} + export function createRealtimeUsersCollection( mockFetch: typeof fetch, - mockChannel: MockChannel + mockChannel: MockChannel | (() => MockChannel) ) { // A fresh QueryClient keeps the module-level realtime registry in db.ts // isolated per test. @@ -161,8 +204,8 @@ export function createRealtimeUsersCollection( global: { fetch: mockFetch }, }) // The real client would open a live WebSocket, so stub the realtime surface. - supabase.channel = vi.fn( - () => mockChannel + supabase.channel = vi.fn(() => + typeof mockChannel === "function" ? mockChannel() : mockChannel ) as unknown as typeof supabase.channel supabase.removeChannel = vi.fn() as unknown as typeof supabase.removeChannel From 1bbf26970fe0fe2658bfe1ede10d39faca9ad18e Mon Sep 17 00:00:00 2001 From: Ivan Vasilov Date: Mon, 7 Sep 2026 15:09:58 +0300 Subject: [PATCH 3/3] fix: harden realtime channel swaps and racing writes - Match e2e channel lookup to the numbered topic names by exporting the topic prefix from db.ts. - Route realtime INSERT/UPDATE through writeUpsert and gate update/delete on the synced store, so an echo of an in-flight local mutation no longer throws. Apply the same to the mutation write-backs in functions.ts. - Only drop the previous channel once the replacement reports SUBSCRIBED. On CHANNEL_ERROR keep the working channel, remember the rejected filter set, and fall back to a catch-all subscription. Co-Authored-By: Claude Fable 5.1 --- src/db.ts | 126 +++++++++++++++++++++++++++++-------- src/functions.ts | 15 +++-- src/realtime.ts | 75 +++++++++++++--------- tests/e2e/e2e.utils.ts | 16 ++--- tests/realtime.test.ts | 137 +++++++++++++++++++++++++++++++++++++++++ tests/test.utils.ts | 11 ++-- 6 files changed, 313 insertions(+), 67 deletions(-) diff --git a/src/db.ts b/src/db.ts index a31c204..e3aefa6 100644 --- a/src/db.ts +++ b/src/db.ts @@ -43,9 +43,18 @@ interface TableEntry { realtimeFiltersKey: string | null /** Resolves once the current channel finished subscribing (or gave up) */ realtimeSubscribed: Promise | null + /** + * Filter sets the server has rejected. They are replaced by a catch-all + * subscription instead of being retried on every observer change. + */ + rejectedFilterKeys: Set supabase: SupabaseClient } +/** A subscription that receives every change for the table. */ +const CATCH_ALL_FILTERS: Array = [null] +const CATCH_ALL_FILTERS_KEY = JSON.stringify(CATCH_ALL_FILTERS) + /** * Channel topics are namespaced and numbered because `supabase.channel()` * returns the *existing* channel for a topic that is already registered, and @@ -55,9 +64,90 @@ interface TableEntry { * same Supabase client — so the counter is module-level, not per table. */ let channelCount = 0 + +/** + * Every channel this adapter opens for `tableName` has a topic starting with + * this prefix (supabase-js exposes it under its own `realtime:` prefix). + */ +export const realtimeChannelTopicPrefix = (tableName: string) => + `supabase-tanstack-db:${tableName}:` + const nextChannelTopic = (tableName: string) => { channelCount += 1 - return `supabase-tanstack-db:${tableName}:${channelCount}` + return `${realtimeChannelTopicPrefix(tableName)}${channelCount}` +} + +/** + * Opens a channel for `filters` and makes it the table's current channel. The + * previous channel keeps listening until the replacement is actually joined, + * so no change slips through while the swap is in flight. If the server + * rejects the new subscription, the previous channel stays in place and a + * catch-all subscription is attempted instead of losing Realtime entirely. + */ +const subscribeToChanges = ( + entry: TableEntry, + tableName: string, + collection: Collection, + filters: Array, + filtersKey: string +) => { + const previousChannel = entry.realtimeChannel + const previousFiltersKey = entry.realtimeFiltersKey + + const subscription = attachSupabaseListeners( + entry.supabase, + nextChannelTopic(tableName), + tableName, + collection, + filters + ) + if (!subscription) { + if (previousChannel) { + entry.supabase.removeChannel(previousChannel) + } + entry.realtimeChannel = null + entry.realtimeFiltersKey = null + entry.realtimeSubscribed = null + return + } + + entry.realtimeChannel = subscription.channel + entry.realtimeFiltersKey = filtersKey + entry.realtimeSubscribed = subscription.ready + + const onJoined = () => { + if (previousChannel) { + entry.supabase.removeChannel(previousChannel) + } + } + + const onRejected = () => { + entry.supabase.removeChannel(subscription.channel) + if (entry.realtimeChannel !== subscription.channel) { + // A newer subscription has already taken over the swap. + return + } + // Hand the swap back to the still-open previous channel, so whatever is + // tried next treats it as its predecessor. + entry.realtimeChannel = previousChannel + entry.realtimeFiltersKey = previousFiltersKey + entry.realtimeSubscribed = null + if (filtersKey === CATCH_ALL_FILTERS_KEY) { + // Even the unfiltered subscription was refused: Realtime is unusable for + // this table right now. The next observer change will try again. + return + } + entry.rejectedFilterKeys.add(filtersKey) + subscribeToChanges( + entry, + tableName, + collection, + CATCH_ALL_FILTERS, + CATCH_ALL_FILTERS_KEY + ) + } + + subscription.joined.then((joined) => (joined ? onJoined() : onRejected())) } // Per-QueryClient registry of table entries, with a single cache subscription per client @@ -101,8 +191,12 @@ const ensureQueryCacheSubscription = (queryClient: QueryClient) => { const whereExpressions = queries.map( (query) => query.meta?.loadSubsetOptions?.where ) - const filters = buildRealtimeFilters(whereExpressions) - const filtersKey = JSON.stringify(filters) + let filters = buildRealtimeFilters(whereExpressions) + let filtersKey = JSON.stringify(filters) + if (entry.rejectedFilterKeys.has(filtersKey)) { + filters = CATCH_ALL_FILTERS + filtersKey = CATCH_ALL_FILTERS_KEY + } // Reuse the existing channel when the set of filters hasn't changed. if (entry.realtimeChannel && entry.realtimeFiltersKey === filtersKey) { @@ -110,30 +204,13 @@ const ensureQueryCacheSubscription = (queryClient: QueryClient) => { } // Filters changed (or no channel yet): subscribe with the new filters. - const previousChannel = entry.realtimeChannel - const subscription = attachSupabaseListeners( - entry.supabase, - nextChannelTopic(tableName), + subscribeToChanges( + entry, tableName, entry.collectionRef, - filters + filters, + filtersKey ) - entry.realtimeChannel = subscription?.channel ?? null - entry.realtimeFiltersKey = subscription ? filtersKey : null - entry.realtimeSubscribed = subscription?.subscribed ?? null - - // Keep the previous channel listening until its replacement is - // subscribed, so no change slips through while the swap is in flight. - if (previousChannel) { - const removePrevious = () => { - entry.supabase.removeChannel(previousChannel) - } - if (subscription) { - subscription.subscribed.then(removePrevious, removePrevious) - } else { - removePrevious() - } - } } }) } @@ -154,6 +231,7 @@ const registerTable = ( realtimeChannel: null, realtimeFiltersKey: null, realtimeSubscribed: null, + rejectedFilterKeys: new Set(), }) } diff --git a/src/functions.ts b/src/functions.ts index fb6d752..85f1b38 100644 --- a/src/functions.ts +++ b/src/functions.ts @@ -234,8 +234,10 @@ export const supabaseOnInsert = async ( throw error } mutation.modified = data - // The data has been inserted and confirmed by the server, so we can write it to the collection - collection.utils.writeInsert(data) + // The data has been inserted and confirmed by the server, so we can write + // it to the collection. Realtime may already have echoed the insert, in + // which case this is an update of the synced row rather than an insert. + collection.utils.writeUpsert(data) }) ) @@ -300,8 +302,13 @@ export const supabaseOnDelete = async ( if (error) { throw error } - // The data has been deleted and confirmed by the server, so we can write it to the collection - collection.utils.writeDelete(collection.getKeyFromItem(mutation.original)) + // The data has been deleted and confirmed by the server, so we can write + // it to the collection — unless Realtime already echoed the delete, in + // which case the synced row is gone and writing again would throw. + const key = collection.getKeyFromItem(mutation.original) + if (collection._state.syncedData.has(key)) { + collection.utils.writeDelete(key) + } }) ) diff --git a/src/realtime.ts b/src/realtime.ts index 28f9bce..18a4d48 100644 --- a/src/realtime.ts +++ b/src/realtime.ts @@ -1,7 +1,8 @@ -import type { - RealtimePostgresChangesFilter, - RealtimePostgresChangesPayload, - SupabaseClient, +import { + REALTIME_SUBSCRIBE_STATES, + type RealtimePostgresChangesFilter, + type RealtimePostgresChangesPayload, + type SupabaseClient, } from "@supabase/supabase-js" import { type Collection, @@ -17,12 +18,19 @@ type ChangeEvent = "INSERT" | "UPDATE" | "DELETE" export type RealtimeSubscription = { channel: ReturnType /** - * Resolves once the channel reached a terminal subscription state, or after + * Resolves once the channel reported any subscription status, or after * {@link SUBSCRIBE_TIMEOUT_MS} if the server never answers. Never rejects, so * callers can safely gate work on it without an unreachable Realtime server * blocking them forever. */ - subscribed: Promise + ready: Promise + /** + * Resolves `true` once the channel is joined and `false` once the server + * rejected the subscription (or the channel closed before joining). Unlike + * {@link ready} it is not time-bounded: while realtime-js is still retrying + * the join it stays pending. + */ + joined: Promise } /** @@ -239,19 +247,21 @@ export const attachSupabaseListeners = < return config } + // `collection.has()` reflects the optimistic view, which includes local + // mutations that are still in flight. The manual sync writes below validate + // against the synced store only, so that is the store to check: a Realtime + // echo of a pending local insert must not be treated as an update, and a + // row with a pending local delete is still there to be updated or deleted. + const isSynced = (id: TKey) => collection._state.syncedData.has(id) + // The row matches an active query's filter, so it belongs in the collection - // whether or not we have seen it before. + // whether or not we have seen it before. Upsert decides insert-vs-update on + // the synced store, which also makes replayed or racing events harmless. const handleUpsert = (payload: RealtimePostgresChangesPayload) => { if (payload.eventType !== "INSERT" && payload.eventType !== "UPDATE") { return } - const row = payload.new as T - const id = collection.getKeyFromItem(row) - if (collection.has(id)) { - collection.utils.writeUpdate(row) - } else { - collection.utils.writeInsert(row) - } + collection.utils.writeUpsert(payload.new as T) } // Unfiltered updates arrive for every row in the table, so only rows the @@ -262,8 +272,7 @@ export const attachSupabaseListeners = < return } const row = payload.new as T - const id = collection.getKeyFromItem(row) - if (collection.has(id)) { + if (isSynced(collection.getKeyFromItem(row))) { collection.utils.writeUpdate(row) } } @@ -273,7 +282,7 @@ export const attachSupabaseListeners = < return } const id = collection.getKeyFromItem(payload.old as T) - if (collection.has(id)) { + if (isSynced(id)) { collection.utils.writeDelete(id) } } @@ -297,20 +306,30 @@ export const attachSupabaseListeners = < handleDelete(p as RealtimePostgresChangesPayload) ) - const subscribed = new Promise((resolve) => { + let resolveJoined: (joined: boolean) => void = () => undefined + const joined = new Promise((resolve) => { + resolveJoined = resolve + }) + + const ready = new Promise((resolve) => { const timeout = setTimeout(resolve, SUBSCRIBE_TIMEOUT_MS) - const settle = () => { + channel.subscribe((status) => { clearTimeout(timeout) resolve() - } - try { - channel.subscribe(settle) - } catch { - // subscribe() throws synchronously on an already-joined channel. Data - // loading must not be held up by a channel that will never connect. - settle() - } + if (status === REALTIME_SUBSCRIBE_STATES.SUBSCRIBED) { + resolveJoined(true) + } else if ( + status === REALTIME_SUBSCRIBE_STATES.CHANNEL_ERROR || + status === REALTIME_SUBSCRIBE_STATES.CLOSED + ) { + // CHANNEL_ERROR is how the server rejects a subscription — typically a + // filter it cannot evaluate — and it errors the whole channel. + // TIMED_OUT is deliberately not terminal: realtime-js keeps retrying + // the join and reports the outcome through this same callback. + resolveJoined(false) + } + }) }) - return { channel, subscribed } + return { channel, ready, joined } } diff --git a/tests/e2e/e2e.utils.ts b/tests/e2e/e2e.utils.ts index 89e7b79..f3b0743 100644 --- a/tests/e2e/e2e.utils.ts +++ b/tests/e2e/e2e.utils.ts @@ -3,6 +3,7 @@ import { createClient, type SupabaseClient } from "@supabase/supabase-js" import { createCollection, createLiveQueryCollection } from "@tanstack/db" import { QueryClient } from "@tanstack/query-core" import { test as baseTest, expect, inject, vi } from "vitest" +import { realtimeChannelTopicPrefix } from "../../src/db" import { supabaseCollectionOptions } from "../../src/index" import { usersSchema } from "../test.utils" @@ -48,16 +49,17 @@ const liveUsers = (base: UsersCollection) => type LiveUsersCollection = ReturnType -// Waits until the adapter's realtime channel for the table is actually joined, -// so changes written afterwards are guaranteed to be captured. Coupled to the -// adapter naming its channel after the table (supabase.channel(tableName) in -// src/db.ts), which supabase-js exposes under the "realtime:" topic prefix. +// Waits until one of the adapter's realtime channels for the table is actually +// joined, so changes written afterwards are guaranteed to be captured. The +// adapter numbers its channel topics (see realtimeChannelTopicPrefix in +// src/db.ts), and supabase-js exposes them under its own "realtime:" prefix. const waitForChannel = (supabase: SupabaseClient, table: string) => vi.waitFor(() => { - const channel = supabase + const prefix = `realtime:${realtimeChannelTopicPrefix(table)}` + const joined = supabase .getChannels() - .find((c) => c.topic === `realtime:${table}`) - expect(channel?.state).toBe("joined") + .some((c) => c.topic.startsWith(prefix) && c.state === "joined") + expect(joined).toBe(true) }, WAIT) const preloadSeeded = async (live: LiveUsersCollection) => { diff --git a/tests/realtime.test.ts b/tests/realtime.test.ts index 2d4beab..4e14135 100644 --- a/tests/realtime.test.ts +++ b/tests/realtime.test.ts @@ -509,6 +509,143 @@ describe("realtime filter propagation", () => { }) }) + describe("racing local mutations", () => { + const row = { id: 77, name: "Race", email: "race@test.com", active: true } + + test("a Realtime echo of a pending local insert does not throw", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }) + ) + const key = collection.getKeyFromItem(row) + + const tx = collection.insert(row) + // The optimistic row is visible right away, but the synced store only + // learns about it once PostgREST answers — Realtime is usually faster. + expect(collection.has(key)).toBe(true) + expect(() => + emit(mockChannel, { eventType: "INSERT", new: row, old: {} }) + ).not.toThrow() + + await expect(tx.isPersisted.promise).resolves.toBeDefined() + expect(collection.get(key)?.name).toBe("Race") + }) + + test("a Realtime echo of a pending local delete does not throw", async () => { + const { mockChannel, collection } = await captureChannel( + (c) => (q) => q.from({ user: c }) + ) + emit(mockChannel, { eventType: "INSERT", new: row, old: {} }) + const key = collection.getKeyFromItem(row) + + const tx = collection.delete(key) + expect(collection.has(key)).toBe(false) + expect(() => + emit(mockChannel, { eventType: "DELETE", new: {}, old: row }) + ).not.toThrow() + + await expect(tx.isPersisted.promise).resolves.toBeDefined() + expect(collection.has(key)).toBe(false) + }) + }) + + describe("swapping channels safely", () => { + // The first channel joins immediately; every replacement waits for the test + // to report its subscription status. + async function startSwap() { + const mockFetch = createMockFetch() + const channels: Array = [] + const { collection, supabase } = createRealtimeUsersCollection( + mockFetch, + () => { + const channel = createMockChannel({ + autoSubscribe: channels.length === 0, + }) + channels.push(channel) + return channel + } + ) + track(collection) + + const first = startLiveQuery( + collection, + (c) => (q) => q.from({ user: c }).where(({ user }) => eq(user.id, 1)) + ) + await first.preload() + await first.toArrayWhenReady() + await vi.waitFor(() => expect(channels.length).toBe(1)) + + const second = startLiveQuery( + collection, + (c) => (q) => q.from({ user: c }).where(({ user }) => gt(user.id, 5)) + ) + const secondPreloaded = second.preload() + await vi.waitFor(() => expect(channels.length).toBe(2)) + await vi.waitFor(() => expect(channels[1].subscribe).toHaveBeenCalled()) + + return { channels, collection, supabase, secondPreloaded } + } + + test("keeps the previous channel until the replacement reports SUBSCRIBED", async () => { + const { channels, supabase, secondPreloaded } = await startSwap() + + // Dropping it now would leave a window with no subscription at all. + expect(supabase.removeChannel).not.toHaveBeenCalled() + + channels[1].confirmSubscribed() + await secondPreloaded + await vi.waitFor(() => + expect(supabase.removeChannel).toHaveBeenCalledWith(channels[0]) + ) + }) + + test("falls back to a catch-all when the server rejects the filtered subscription", async () => { + const { channels, collection, supabase, secondPreloaded } = + await startSwap() + + channels[1].confirmSubscribed("CHANNEL_ERROR") + + await vi.waitFor(() => expect(channels.length).toBe(3)) + expect(insertFilters(channels[2])).toEqual([null]) + expect(supabase.removeChannel).toHaveBeenCalledWith(channels[1]) + // The working channel is only replaced once the fallback has joined. + expect(supabase.removeChannel).not.toHaveBeenCalledWith(channels[0]) + + channels[2].confirmSubscribed() + await vi.waitFor(() => + expect(supabase.removeChannel).toHaveBeenCalledWith(channels[0]) + ) + await secondPreloaded + + // The rejected filter set is remembered: a query that would produce it + // again reuses the catch-all instead of failing the same way twice. + const third = startLiveQuery( + collection, + (c) => (q) => q.from({ user: c }).where(({ user }) => gt(user.id, 5)) + ) + await third.preload() + await third.toArrayWhenReady() + expect(channels.length).toBe(3) + }) + + test("keeps the previous channel when even the catch-all is rejected", async () => { + const { channels, supabase, secondPreloaded } = await startSwap() + + channels[1].confirmSubscribed("CHANNEL_ERROR") + await vi.waitFor(() => expect(channels.length).toBe(3)) + channels[2].confirmSubscribed("CHANNEL_ERROR") + + await vi.waitFor(() => + expect(supabase.removeChannel).toHaveBeenCalledWith(channels[2]) + ) + expect(supabase.removeChannel).toHaveBeenCalledWith(channels[1]) + expect(supabase.removeChannel).not.toHaveBeenCalledWith(channels[0]) + expect(channels.length).toBe(3) + + // Data loading is never held hostage by a failed subscription. + await secondPreloaded + }) + }) + // `or(...)` (and other unsupported expressions) cannot be driven through the // live-query path because the query's own supabaseQueryFn calls the same // throwing extractSimpleComparisons. Cover the defensive fallback directly. diff --git a/tests/test.utils.ts b/tests/test.utils.ts index acb4508..010cfcf 100644 --- a/tests/test.utils.ts +++ b/tests/test.utils.ts @@ -135,8 +135,11 @@ export type MockChannel = { onCalls: MockChannelOnCall[] on: ReturnType subscribe: ReturnType - /** Reports SUBSCRIBED when the channel was created with `autoSubscribe: false`. */ - confirmSubscribed: () => void + /** + * Reports a subscription status (SUBSCRIBED by default) when the channel was + * created with `autoSubscribe: false`. + */ + confirmSubscribed: (status?: string) => void } export function createMockChannel({ @@ -163,8 +166,8 @@ export function createMockChannel({ } return channel }), - confirmSubscribed: () => { - pending?.("SUBSCRIBED") + confirmSubscribed: (status = "SUBSCRIBED") => { + pending?.(status) pending = undefined }, }