feat(supabase-db-collection): add experimental_queryOnce

.github/workflows/e2e-tests.yml

@@ -18,7 +18,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        group: [browser, desktop, cloud-mobile]
+        group: [browser, desktop, cloud-mobile, supabase]
     runs-on: ubuntu-latest
     timeout-minutes: 35
     steps:
@@ -76,6 +76,25 @@ jobs:
           --filter '@tanstack/react-native-db-sqlite-persistence...'
           build
 
+      - name: Setup Supabase CLI
+        if: matrix.group == 'supabase'
+        uses: supabase/setup-cli@689adcfb80d947100a1eab127289b97d1fb07b13 # v3.2.0
+        with:
+          version: 2.116.0
+
+      - name: Start Supabase
+        if: matrix.group == 'supabase'
+        # The collection needs only the database, PostgREST, Realtime and the
+        # API gateway (Kong). Skipping the other services saves image pulls and
+        # startup time. The e2e global setup reuses this running stack.
+        working-directory: packages/supabase-db-collection
+        run: supabase start -x studio,mailpit,storage-api,imgproxy,postgres-meta,edge-runtime,logflare,vector,supavisor
+
+      - name: Run Supabase E2E tests
+        if: matrix.group == 'supabase'
+        working-directory: packages/supabase-db-collection
+        run: pnpm test:e2e
+
       - name: Run Electric E2E tests
         if: matrix.group == 'browser'
         run: |
@@ -233,6 +252,11 @@ jobs:
           cd packages/db-collection-e2e/docker
           docker compose down -v
 
+      - name: Stop Supabase
+        if: always() && matrix.group == 'supabase'
+        working-directory: packages/supabase-db-collection
+        run: supabase stop --no-backup
+
       - name: Upload test results
         if: failure()
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2

packages/supabase-db-collection/e2e/e2e.utils.ts

@@ -0,0 +1,164 @@
+/* eslint-disable no-empty-pattern -- vitest fixtures without dependencies must destructure an empty context object */
+import { createClient } 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 { supabaseCollectionOptions } from '../src/index'
+import { realtimeChannelTopicPrefix } from '../src/realtime'
+import { usersSchema } from '../tests/test.utils'
+import { makeAdminClient, resetE2e } from './setup'
+import type { SupabaseClient } from '@supabase/supabase-js'
+
+// Shared polling window for awaiting async PostgREST/realtime propagation.
+export const WAIT = { timeout: 15_000, interval: 200 } as const
+
+// A fresh Supabase client using the anon key, exactly like a browser would.
+const makeSupabase = (): SupabaseClient =>
+  createClient(inject('supabaseUrl'), inject('supabaseAnonKey'))
+
+// Each collection gets its own QueryClient: the realtime channel registry in
+// src/db.ts is keyed by QueryClient, so a fresh one per collection prevents
+// channels leaking across tests.
+export const makeUsersCollection = ({
+  realtime = false,
+  realtimeUseFilter = false,
+} = {}) => {
+  const supabase = makeSupabase()
+  const collection = createCollection(
+    supabaseCollectionOptions({
+      tableName: 'users',
+      keys: ['id'],
+      schema: usersSchema,
+      supabase,
+      queryClient: new QueryClient(),
+      realtime,
+      realtimeUseFilter,
+    }),
+  )
+  return { collection, supabase }
+}
+
+// A realtime collection that pushes each live query's WHERE down to the
+// server-side postgres_changes filter (see realtimeUseFilter in src/db.ts).
+export const makeFilteredRealtimeUsers = () =>
+  makeUsersCollection({ realtime: true, realtimeUseFilter: true })
+
+type UsersContext = ReturnType<typeof makeUsersCollection>
+type UsersCollection = UsersContext['collection']
+
+// An on-demand collection only loads when a live query drives demand. This live
+// query also creates the active query that the realtime channel attaches to.
+const liveUsers = (base: UsersCollection) =>
+  createLiveQueryCollection((q) =>
+    q.from({ row: base }).select(({ row }) => ({
+      id: row.id,
+      name: row.name,
+      email: row.email,
+      active: row.active,
+    })),
+  )
+
+type LiveUsersCollection = ReturnType<typeof liveUsers>
+
+// 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.
+export const waitForChannel = (supabase: SupabaseClient, table: string) =>
+  vi.waitFor(() => {
+    const prefix = `realtime:${realtimeChannelTopicPrefix(table)}`
+    const joined = supabase
+      .getChannels()
+      .some((c) => c.topic.startsWith(prefix) && c.state === 'joined')
+    expect(joined).toBe(true)
+  }, WAIT)
+
+// A live query and its base collection both expose these; the helper below only
+// needs to drive their lifecycle, not their query shape.
+type Preloadable = {
+  preload: () => Promise<unknown>
+  cleanup: () => Promise<unknown>
+}
+
+// Preloads one or more filtered realtime live queries over a shared collection
+// and blocks until the channel has actually joined — which only happens once the
+// server accepted the filters these queries emit, so it doubles as proof the
+// emitted filter syntax is valid. Returns a cleanup to await in the test's
+// `finally`, tearing down the live queries, the base collection, and the
+// client's channels.
+export const startFilteredRealtime = async (
+  ctx: UsersContext,
+  ...lives: Array<Preloadable>
+) => {
+  await Promise.all(lives.map((live) => live.preload()))
+  await waitForChannel(ctx.supabase, 'users')
+  return async () => {
+    await Promise.all(lives.map((live) => live.cleanup()))
+    await ctx.collection.cleanup()
+    await ctx.supabase.removeAllChannels()
+  }
+}
+
+const preloadSeeded = async (live: LiveUsersCollection) => {
+  await live.preload()
+  // The reset_e2e() seed always yields exactly Alice and Bob.
+  await vi.waitFor(() => expect(live.size).toBe(2), WAIT)
+}
+
+// Fixtures are lazy: a test only pays for what it destructures. Teardown runs
+// in reverse initialization order and ALWAYS runs, even when the test fails —
+// unlike trailing cleanup calls in a test body.
+export const test = baseTest.extend<{
+  /** Base users collection + its anon client. */
+  users: UsersContext
+  /** Preloaded live query over `users`, seeded rows awaited. */
+  usersLive: LiveUsersCollection
+  /** Like `users` but with realtime enabled. */
+  rtUsers: UsersContext
+  /** Preloaded live query over `rtUsers`, realtime channel joined. */
+  rtUsersLive: LiveUsersCollection
+  /** A second anon client acting as "another user" for realtime writes. */
+  other: SupabaseClient
+  /** Resets the users tables to the Alice/Bob fixture before every test. */
+  resetDatabase: void
+}>({
+  resetDatabase: [
+    async ({}, use) => {
+      await resetE2e(makeAdminClient())
+      await use()
+    },
+    { auto: true },
+  ],
+  users: async ({}, use) => {
+    const ctx = makeUsersCollection()
+    await use(ctx)
+    await ctx.collection.cleanup()
+    await ctx.supabase.removeAllChannels()
+  },
+  usersLive: async ({ users }, use) => {
+    const live = liveUsers(users.collection)
+    await preloadSeeded(live)
+    await use(live)
+    await live.cleanup()
+  },
+  rtUsers: async ({}, use) => {
+    const ctx = makeUsersCollection({ realtime: true })
+    await use(ctx)
+    await ctx.collection.cleanup()
+    await ctx.supabase.removeAllChannels()
+  },
+  rtUsersLive: async ({ rtUsers }, use) => {
+    const live = liveUsers(rtUsers.collection)
+    await preloadSeeded(live)
+    // Wait until the realtime channel is joined before the test writes,
+    // otherwise a change could be published before the subscription exists.
+    await waitForChannel(rtUsers.supabase, 'users')
+    await use(live)
+    await live.cleanup()
+  },
+  other: async ({}, use) => {
+    const client = makeSupabase()
+    await use(client)
+    await client.removeAllChannels()
+  },
+})

packages/supabase-db-collection/e2e/filters.e2e.test.ts

@@ -0,0 +1,171 @@
+import {
+  IR,
+  and,
+  createLiveQueryCollection,
+  eq,
+  ilike,
+  inArray,
+  like,
+  not,
+  or,
+  upper,
+} from '@tanstack/db'
+import { expect } from 'vitest'
+import { experimental_queryOnce } from '../src/index'
+import { executeQuery } from '../src/query-once'
+import { makeUsersCollection, test } from './e2e.utils'
+
+const name = new IR.PropRef<string>([`user`, `name`])
+const active = new IR.PropRef<boolean>([`user`, `active`])
+const id = new IR.PropRef<number>([`user`, `id`])
+
+// Each case reads the same filter on three paths, and every path asserts the
+// rows it returns, not the shape of a URL:
+//
+// - An on-demand live query sends its WHERE to PostgREST and becomes ready
+//   once that read has finished, so its rows are the server's answer filtered
+//   again on the client.
+// - `experimental_queryOnce` runs the same one-shot query through TanStack
+//   DB's `queryOnce`.
+// - `executeQuery` sends a serialized query IR straight to PostgREST, through
+//   the translation that server-side aggregates use. Nothing filters its rows
+//   again, so a filter the server applies too loosely shows up here.
+//
+// A filter the server applies too strictly loses rows, which the client
+// cannot add back. Each call reads through a fresh collection, so rows an
+// earlier query loaded cannot hide the loss.
+const withUsers = async <T>(
+  read: (users: ReturnType<typeof makeUsersCollection>) => Promise<T>,
+): Promise<T> => {
+  const users = makeUsersCollection()
+  try {
+    return await read(users)
+  } finally {
+    await users.collection.cleanup()
+    await users.supabase.removeAllChannels()
+  }
+}
+
+const sortedNames = (rows: Array<unknown>) =>
+  rows.map((row) => (row as { name: string }).name).sort()
+
+const liveNames = (where: IR.BasicExpression<boolean>) =>
+  withUsers(async ({ collection }) => {
+    const live = createLiveQueryCollection((q) =>
+      q.from({ user: collection }).where(() => where),
+    )
+    try {
+      await live.preload()
+      return sortedNames(live.toArray)
+    } finally {
+      await live.cleanup()
+    }
+  })
+
+const queryOnceNames = (where: IR.BasicExpression<boolean>) =>
+  withUsers(async ({ collection, supabase }) =>
+    sortedNames(
+      await experimental_queryOnce(
+        (q) => q.from({ user: collection }).where(() => where),
+        supabase,
+      ),
+    ),
+  )
+
+const serverNames = (where: IR.BasicExpression<boolean>) =>
+  withUsers(async ({ supabase }) =>
+    sortedNames(
+      await executeQuery(supabase, {
+        from: { type: `table`, name: `users`, alias: `user` },
+        where: [where],
+      }),
+    ),
+  )
+
+const expectNamesOnEveryPath = async (
+  where: IR.BasicExpression<boolean>,
+  expected: ReadonlyArray<string>,
+) => {
+  const label = JSON.stringify(where)
+  expect(await liveNames(where), `live query ${label}`).toEqual(expected)
+  expect(await queryOnceNames(where), `queryOnce ${label}`).toEqual(expected)
+  expect(await serverNames(where), `executeQuery ${label}`).toEqual(expected)
+}
+
+const cases = [
+  [`OR`, or(eq(name, `Alice`), eq(name, `Nobody`)), [`Alice`]],
+  [
+    `nested AND/OR`,
+    or(and(eq(active, true), eq(name, `Bob`)), eq(id, 2)),
+    [`Bob`],
+  ],
+  [`NOT AND`, not(and(eq(active, true), eq(name, `Alice`))), [`Bob`]],
+  [`NOT OR`, not(or(eq(active, true), eq(name, `Nobody`))), [`Bob`]],
+  [`double NOT`, not(not(or(eq(active, true), eq(name, `Nobody`)))), [`Alice`]],
+  [`LIKE`, like(name, `A_i%`), [`Alice`]],
+  [`ILIKE`, ilike(name, `%ALI%`), [`Alice`]],
+  [`NOT LIKE`, not(like(name, `A%`)), [`Bob`]],
+  [
+    `OR ILIKE`,
+    or(ilike(name, `%ali%`), ilike(name, `%bOB%`)),
+    [`Alice`, `Bob`],
+  ],
+  [`overlapping IN`, and(inArray(id, [1, 2]), inArray(id, [2, 3])), [`Bob`]],
+  [`empty IN`, inArray(id, []), []],
+  [`NOT empty IN`, not(inArray(id, [])), [`Alice`, `Bob`]],
+  [`NOT IN with null`, not(inArray(name, [`Alice`, null])), [`Bob`]],
+] as const
+
+for (const [label, where, expected] of cases) {
+  test(`${label} returns matching rows on every query path`, async () => {
+    await expectNamesOnEveryPath(where, expected)
+  })
+}
+
+for (const value of [``, ` a `, `a,b(c)`, `a"b\\c`, `a.b:c&d=1+%`, `{a,b}`]) {
+  test(`reserved value ${JSON.stringify(value)} round-trips through filters`, async ({
+    other,
+  }) => {
+    const { error } = await other
+      .from(`users`)
+      .insert({ name: value, email: `special@test.com` } as unknown as never)
+    expect(error).toBeNull()
+    for (const where of [
+      eq(name, value),
+      or(eq(name, value), eq(id, -1)),
+      inArray(name, [value]),
+    ]) {
+      await expectNamesOnEveryPath(where, [value])
+    }
+  })
+}
+
+test(`LIKE treats backslashes literally on every query path`, async ({
+  other,
+}) => {
+  const value = `a\\path`
+  const { error } = await other
+    .from(`users`)
+    .insert({ name: value, email: `slash@test.com` } as unknown as never)
+  expect(error).toBeNull()
+  for (const where of [
+    like(name, `a\\%`),
+    or(like(name, `a\\%`), eq(id, -1)),
+  ]) {
+    await expectNamesOnEveryPath(where, [value])
+  }
+})
+
+// executeQuery is left out here: it returns PostgREST's rows without
+// filtering them again, so a filter PostgREST cannot express has no exact
+// answer on that path.
+test(`unsupported OR and negated patterns retain rows for client filtering`, async () => {
+  for (const read of [liveNames, queryOnceNames]) {
+    expect(
+      await read(or(eq(name, `Nobody`), eq(upper(name), `ALICE`))),
+    ).toEqual([`Alice`])
+    // PostgREST would interpret * as a wildcard and remove Alice here, whereas
+    // TanStack treats it literally and should keep both seeded names.
+    expect(await read(not(like(name, `*Ali*`)))).toEqual([`Alice`, `Bob`])
+  }
+})