feat: Add filters on the realtime subscriptions

src/db.ts

@@ -13,6 +13,7 @@ import {
   supabaseQueryFn,
 } from "./functions"
 import { getQueryClient } from "./query-client"
+import { attachSupabaseListeners, buildRealtimeFilters } from "./realtime"
 
 type GenericPostgrestFilterBuilder = PostgrestFilterBuilder<any, any, any, any>
 
@@ -38,9 +39,27 @@ interface SupabaseCollectionOptions<TSchema extends StandardSchemaV1> {
 interface TableEntry {
   collectionRef: Collection<any, any> | null
   realtimeChannel: ReturnType<SupabaseClient["channel"]> | 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<void> | 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<QueryClient, Map<string, TableEntry>>()
 
@@ -62,15 +81,58 @@ 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
+      )
+      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): subscribe with the new filters.
+      const previousChannel = entry.realtimeChannel
+      const subscription = attachSupabaseListeners(
+        entry.supabase,
+        nextChannelTopic(tableName),
+        tableName,
+        entry.collectionRef,
+        filters
+      )
+      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()
+        }
       }
     }
   })
@@ -90,6 +152,8 @@ const registerTable = (
       supabase,
       collectionRef: null,
       realtimeChannel: null,
+      realtimeFiltersKey: null,
+      realtimeSubscribed: null,
     })
   }
 
@@ -140,7 +204,16 @@ export const supabaseCollectionOptions = <TSchema extends StandardSchemaV1>({
     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,39 +237,3 @@ export const supabaseCollectionOptions = <TSchema extends StandardSchemaV1>({
     },
   }
 }
-
-export const attachSupabaseListeners = <
-  T extends object,
-  TKey extends string | number,
->(
-  supabase: SupabaseClient,
-  tableName: string,
-  collection: Collection<T, TKey>
-): ReturnType<SupabaseClient["channel"]> | null => {
-  if (!supabase.channel) {
-    console.log("Server supabase doesn't have a channel")
-    return null
-  }
-
-  const channel = supabase.channel(tableName)
-  channel
-    .on<T>(
-      "postgres_changes",
-      { event: "*", schema: "public", table: tableName },
-      (payload) => {
-        if (payload.eventType === "INSERT") {
-          collection.utils.writeInsert(payload.new)
-        } else if (payload.eventType === "UPDATE") {
-          collection.utils.writeUpdate(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
-}

src/functions.ts

@@ -13,29 +13,80 @@ import {
 } from "@tanstack/db"
 import type { QueryClient, QueryMeta } from "@tanstack/query-core"
 
+/** `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<any, any, any, any>,
+  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<any, any, any, any>,
   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 = (
@@ -70,9 +121,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}`)
@@ -145,9 +197,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