feat: Add a header to all PostgREST API calls

package.json

@@ -31,14 +31,14 @@
     "test:e2e": "vitest run --project e2e",
     "test:all": "vitest run",
     "typecheck": "tsc --noEmit",
-    "prepublishOnly": "pnpm run build",
+    "prepublishOnly": "pnpm run test && pnpm run build",
     "check": "ultracite check",
     "fix": "ultracite fix"
   },
   "dependencies": {
     "@standard-schema/spec": "^1.1.0",
     "@supabase/postgrest-js": "^2.0.0",
-    "@supabase/supabase-js": "^2.0.0",
+    "@supabase/supabase-js": "^2.74.0",
     "@tanstack/db": "^0.6.0",
     "@tanstack/query-core": "^5.0.0",
     "@tanstack/query-db-collection": "^1.0.0"

src/db.ts

@@ -184,10 +184,13 @@ export const attachSupabaseListeners = <
       "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.writeInsert(payload.new)
+          collection.utils.writeUpsert(payload.new)
         } else if (payload.eventType === "UPDATE") {
-          collection.utils.writeUpdate(payload.new)
+          collection.utils.writeUpsert(payload.new)
         } else if (payload.eventType === "DELETE") {
           const id = collection.getKeyFromItem(payload.old as T)
           if (collection.has(id)) {

src/functions.ts

@@ -12,6 +12,7 @@ import {
   type UpdateMutationFnParams,
 } from "@tanstack/db"
 import type { QueryClient, QueryMeta } from "@tanstack/query-core"
+import { CLIENT_INFO, CLIENT_INFO_HEADER } from "./request-headers"
 
 const buildQuery = (
   baseQuery: PostgrestFilterBuilder<any, any, any, any>,
@@ -127,7 +128,10 @@ export const supabaseQueryFn = async (
   // console.log(tableName, parsed);
   // console.log(tableName, cursorFilters);
 
-  let baseQuery = supabase.from(tableName).select("*")
+  let baseQuery = supabase
+    .from(tableName)
+    .select("*")
+    .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO)
 
   if (parsed.limit) {
     baseQuery = baseQuery.limit(parsed.limit)
@@ -170,6 +174,7 @@ export const supabaseOnInsert = async (
         .insert({
           ...mutation.modified,
         })
+        .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO)
         .select()
         .single()
 
@@ -198,10 +203,13 @@ export const supabaseOnUpdate = async (
     transaction.mutations.map(async (mutation) => {
       const { original, changes } = mutation
       const { error, data } = await filter(
-        supabase.from(tableName).update({
-          ...original,
-          ...changes,
-        }),
+        supabase
+          .from(tableName)
+          .update({
+            ...original,
+            ...changes,
+          })
+          .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO),
         mutation.original
       )
         .select()
@@ -230,7 +238,10 @@ export const supabaseOnDelete = async (
   await Promise.all(
     transaction.mutations.map(async (mutation) => {
       const { error } = await filter(
-        supabase.from(tableName).delete(),
+        supabase
+          .from(tableName)
+          .delete()
+          .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO),
         mutation.original
       )
 

src/query-once.ts

@@ -9,6 +9,7 @@ import {
   type QueryBuilder,
   queryOnce as queryOnceBase,
 } from "@tanstack/db"
+import { CLIENT_INFO, CLIENT_INFO_HEADER } from "./request-headers"
 import {
   type SerializedExpression,
   type SerializedFrom,
@@ -515,7 +516,10 @@ export function buildSupabaseQuery(
   const allWheres = [...subqueryWheres, ...(ir.where ?? [])]
 
   const selectString = buildSelectString(ir)
-  let query: SupabaseQuery = supabase.from(tableName).select(selectString)
+  let query: SupabaseQuery = supabase
+    .from(tableName)
+    .select(selectString)
+    .setHeader(CLIENT_INFO_HEADER, CLIENT_INFO)
 
   // Apply pushable where filters (skip residual / client-side filters)
   for (const w of allWheres) {

src/request-headers.ts

@@ -0,0 +1,13 @@
+import { VERSION } from "./version"
+
+/**
+ * Header name and value used to identify requests made by this library.
+ *
+ * Chain `.setHeader(CLIENT_INFO_HEADER, CLIENT_INFO)` onto a
+ * `PostgrestQueryBuilder`/`PostgrestFilterBuilder` call to stamp the
+ * request. This intentionally REPLACES supabase-js's own default
+ * `X-Client-Info` header (matching the `@supabase/ssr` convention) rather
+ * than appending to or preserving it.
+ */
+export const CLIENT_INFO_HEADER = "X-Client-Info"
+export const CLIENT_INFO = `@supabase-labs/tanstack-db/${VERSION}`

src/version.ts

@@ -0,0 +1 @@
+export const VERSION = "0.0.1"

tests/query-once.test.ts

@@ -22,14 +22,16 @@ import {
   sum,
   upper,
 } from "@tanstack/db"
-import { afterEach, beforeEach, describe, test } from "vitest"
+import { afterEach, beforeEach, describe, expect, test } from "vitest"
 import { queryOnce } from "../src/index"
+import { VERSION } from "../src/version"
 import {
   createMockedTodosCollection,
   createMockedUsersCollection,
   createMockedUsersTodosCollection,
   createMockFetch,
   expectFetchUrls,
+  getRequestHeaders,
   SUPABASE_KEY,
   SUPABASE_URL,
 } from "./test.utils"
@@ -64,6 +66,28 @@ describe("queryOnce PostgREST query generation", () => {
     })
   })
 
+  describe("automatic X-Client-Info header", () => {
+    test("is present on the request", async () => {
+      await queryOnce((q) => q.from({ user: usersCollection }), supabase)
+      expect(getRequestHeaders(mockFetch, 0).get("x-client-info")).toBe(
+        `@supabase-labs/tanstack-db/${VERSION}`
+      )
+    })
+
+    test("is present on the aggregate/groupBy PostgREST push-down path", async () => {
+      await queryOnce(
+        (q) =>
+          q
+            .from({ user: usersCollection })
+            .select(({ user }) => ({ totalUsers: count(user.id) })),
+        supabase
+      )
+      expect(getRequestHeaders(mockFetch, 0).get("x-client-info")).toBe(
+        `@supabase-labs/tanstack-db/${VERSION}`
+      )
+    })
+  })
+
   describe("WHERE", () => {
     test("WHERE active = true", async () => {
       await queryOnce(

tests/realtime-listeners.test.ts

@@ -0,0 +1,104 @@
+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<typeof createMockedUsersCollection>
+
+  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)
+  })
+})