diff --git a/src/db.ts b/src/db.ts index 50d17ac..e3aefa6 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,9 +39,117 @@ interface SupabaseCollectionOptions { interface TableEntry { collectionRef: Collection | null 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 + /** + * 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 + * 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 + +/** + * 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 `${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 const queryClientRegistries = new Map>() @@ -62,16 +171,46 @@ 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) { - entry.supabase.removeChannel(entry.realtimeChannel) - entry.realtimeChannel = null + // 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 + entry.realtimeSubscribed = 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 + ) + 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) { + continue + } + + // Filters changed (or no channel yet): subscribe with the new filters. + subscribeToChanges( + entry, + tableName, + entry.collectionRef, + filters, + filtersKey + ) } }) } @@ -90,6 +229,9 @@ const registerTable = ( supabase, collectionRef: null, realtimeChannel: null, + realtimeFiltersKey: null, + realtimeSubscribed: null, + rejectedFilterKeys: new Set(), }) } @@ -140,7 +282,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), @@ -164,42 +315,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/functions.ts b/src/functions.ts index 80bc68f..85f1b38 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 @@ -182,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) }) ) @@ -248,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 new file mode 100644 index 0000000..18a4d48 --- /dev/null +++ b/src/realtime.ts @@ -0,0 +1,335 @@ +import { + REALTIME_SUBSCRIBE_STATES, + type RealtimePostgresChangesFilter, + type RealtimePostgresChangesPayload, + type SupabaseClient, +} from "@supabase/supabase-js" +import { + type Collection, + extractSimpleComparisons, + type LoadSubsetOptions, + type SimpleComparison, +} from "@tanstack/db" + +type WhereExpression = LoadSubsetOptions["where"] + +type ChangeEvent = "INSERT" | "UPDATE" | "DELETE" + +export type RealtimeSubscription = { + channel: ReturnType + /** + * 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. + */ + 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 +} + +/** + * Maps TanStack DB comparison operators to the operators supported by Supabase + * 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. + */ +const toRealtimeFilter = (comparison: SimpleComparison): string | null => { + const operator = REALTIME_OPERATORS[comparison.operator] + if (!operator) { + return null + } + + // 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 + } + + 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}` + } + + 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. + * + * 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 +): 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 === 0) { + return [null] + } + + const conditions: Array = [] + for (const comparison of comparisons) { + const condition = toRealtimeFilter(comparison) + if (condition === null) { + return [null] + } + conditions.push(condition) + } + // 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).sort() : [null] +} + +/** + * Subscribes to Supabase Realtime changes for a table and writes inserts, + * 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] +): RealtimeSubscription | null => { + if (!supabase.channel) { + return null + } + + const channel = supabase.channel(topic) + + const changesFilter = ( + event: TEvent, + filter: string | null + ): RealtimePostgresChangesFilter => { + const config: RealtimePostgresChangesFilter = { + event, + schema: "public", + table: tableName, + } + if (filter) { + config.filter = filter + } + 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. 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 + } + collection.utils.writeUpsert(payload.new as T) + } + + // 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 + } + const row = payload.new as T + if (isSynced(collection.getKeyFromItem(row))) { + collection.utils.writeUpdate(row) + } + } + + const handleDelete = (payload: RealtimePostgresChangesPayload) => { + if (payload.eventType !== "DELETE") { + return + } + const id = collection.getKeyFromItem(payload.old as T) + if (isSynced(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) + ) + } + + 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) + ) + + 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) + channel.subscribe((status) => { + clearTimeout(timeout) + resolve() + 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, 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/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-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..4e14135 --- /dev/null +++ b/tests/realtime.test.ts @@ -0,0 +1,658 @@ +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, + emit, + filtersFor, + listenersFor, + 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"] + +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( + buildQuery: (collection: RealtimeCollection) => QueryFn +) { + const mockFetch = createMockFetch() + const mockChannel = createMockChannel() + const { collection } = createRealtimeUsersCollection(mockFetch, mockChannel) + track(collection) + + const live = startLiveQuery(collection, buildQuery) + + await live.preload() + await live.toArrayWhenReady() + await vi.waitFor(() => expect(mockChannel.on).toHaveBeenCalled()) + + return { mockChannel, collection } +} + +// 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", () => { + 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(insertFilters(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(insertFilters(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(insertFilters(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(insertFilters(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(insertFilters(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(insertFilters(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(insertFilters(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(insertFilters(mockChannel)).toEqual(["active=neq.false"]) + }) + + test("isNull maps to is.null", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + q.from({ user: collection }).where(({ user }) => isNull(user.name)) + ) + expect(insertFilters(mockChannel)).toEqual(["name=is.null"]) + }) + + test("not(isNull) maps to not.is.null", async () => { + const { mockChannel } = await captureChannel( + (collection) => (q) => + 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(insertFilters(mockChannel)).toEqual(["id=not.gt.5"]) + }) + + 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(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]) + }) + }) + + 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 }) + ) + + emit(mockChannel, { eventType: "INSERT", new: row, old: {} }) + + 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 }) + ) + + emit(mockChannel, { eventType: "INSERT", new: row, old: {} }) + emit(mockChannel, { + eventType: "UPDATE", + new: { ...row, name: "Updated" }, + old: {}, + }) + + 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 () => { + 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 }) + ) + + emit(mockChannel, { eventType: "INSERT", new: row, old: {} }) + const key = collection.getKeyFromItem(row) + expect(collection.has(key)).toBe(true) + + 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(() => + emit(mockChannel, { + eventType: "DELETE", + new: {}, + old: { id: 12_345, name: "", email: "", active: false }, + }) + ).not.toThrow() + }) + }) + + 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]) + ) + }) + }) + + 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. + 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..010cfcf 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,109 @@ 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 + /** + * Reports a subscription status (SUBSCRIBED by default) when the channel was + * created with `autoSubscribe: false`. + */ + confirmSubscribed: (status?: string) => void +} + +export function createMockChannel({ + autoSubscribe = true, +}: { + autoSubscribe?: boolean +} = {}): MockChannel { + const onCalls: MockChannelOnCall[] = [] + 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 + }), + // 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: (status = "SUBSCRIBED") => { + pending?.(status) + 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) +) { + // 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(() => + typeof mockChannel === "function" ? mockChannel() : 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(