Accept query options in live queries and React hooks

.changeset/live-query-options.md

@@ -0,0 +1,6 @@
+---
+'@electric-sql/pglite': patch
+'@electric-sql/pglite-react': patch
+---
+
+Allow `live.query` and React's `useLiveQuery` hook to accept query options such as `rowMode: 'array'`.

packages/pglite-react/src/hooks.ts

@@ -1,4 +1,5 @@
 import type { LiveQuery, LiveQueryResults } from '@electric-sql/pglite/live'
+import type { QueryOptions } from '@electric-sql/pglite'
 import { query as buildQuery } from '@electric-sql/pglite/template'
 import { useEffect, useRef, useState } from 'react'
 import { usePGlite } from './provider'
@@ -17,13 +18,44 @@ function paramsEqual(
   return true
 }
 
+function shallowRecordsEqual(a: object | undefined, b: object | undefined) {
+  if (!a && !b) return true
+  if (!a || !b) return false
+  const aRecord = a as Record<PropertyKey, unknown>
+  const bRecord = b as Record<PropertyKey, unknown>
+  const aKeys = Reflect.ownKeys(a)
+  const bKeys = Reflect.ownKeys(b)
+  return (
+    aKeys.length === bKeys.length &&
+    aKeys.every((key) => Object.is(aRecord[key], bRecord[key]))
+  )
+}
+
+function queryOptionsEqual(
+  a: QueryOptions | undefined,
+  b: QueryOptions | undefined,
+) {
+  if (!a && !b) return true
+  if (!a || !b) return false
+  return (
+    a.rowMode === b.rowMode &&
+    shallowRecordsEqual(a.parsers, b.parsers) &&
+    shallowRecordsEqual(a.serializers, b.serializers) &&
+    Object.is(a.blob, b.blob) &&
+    Object.is(a.onNotice, b.onNotice) &&
+    paramsEqual(a.paramTypes, b.paramTypes)
+  )
+}
+
 function useLiveQueryImpl<T = { [key: string]: unknown }>(
   query: string | LiveQuery<T> | Promise<LiveQuery<T>>,
   params: unknown[] | undefined | null,
   key?: string,
+  options?: QueryOptions,
 ): Omit<LiveQueryResults<T>, 'affectedRows'> | undefined {
   const db = usePGlite()
   const paramsRef = useRef(params)
+  const optionsRef = useRef(options)
   const liveQueryRef = useRef<LiveQuery<T> | undefined>(undefined)
   let liveQuery: LiveQuery<T> | undefined
   let liveQueryChanged = false
@@ -42,6 +74,12 @@ function useLiveQueryImpl<T = { [key: string]: unknown }>(
     currentParams = params
   }
 
+  let currentOptions = optionsRef.current
+  if (!queryOptionsEqual(optionsRef.current, options)) {
+    optionsRef.current = options
+    currentOptions = options
+  }
+
   /* eslint-disable @eslint-react/hooks-extra/no-direct-set-state-in-use-effect */
   useEffect(() => {
     let cancelled = false
@@ -53,7 +91,9 @@ function useLiveQueryImpl<T = { [key: string]: unknown }>(
       const ret =
         key !== undefined
           ? db.live.incrementalQuery<T>(query, currentParams, key, cb)
-          : db.live.query<T>(query, currentParams, cb)
+          : currentOptions
+            ? db.live.query<T>(query, currentParams, currentOptions, cb)
+            : db.live.query<T>(query, currentParams, cb)
 
       return () => {
         cancelled = true
@@ -80,7 +120,7 @@ function useLiveQueryImpl<T = { [key: string]: unknown }>(
     } else {
       throw new Error('Should never happen')
     }
-  }, [db, key, query, currentParams, liveQuery])
+  }, [db, key, query, currentParams, currentOptions, liveQuery])
   /* eslint-enable @eslint-react/hooks-extra/no-direct-set-state-in-use-effect */
 
   if (liveQueryChanged && liveQuery) {
@@ -101,6 +141,7 @@ function useLiveQueryImpl<T = { [key: string]: unknown }>(
 export function useLiveQuery<T = { [key: string]: unknown }>(
   query: string,
   params?: unknown[] | null,
+  options?: QueryOptions,
 ): LiveQueryResults<T> | undefined
 
 export function useLiveQuery<T = { [key: string]: unknown }>(
@@ -114,8 +155,9 @@ export function useLiveQuery<T = { [key: string]: unknown }>(
 export function useLiveQuery<T = { [key: string]: unknown }>(
   query: string | LiveQuery<T> | Promise<LiveQuery<T>>,
   params?: unknown[] | null,
+  options?: QueryOptions,
 ): LiveQueryResults<T> | undefined {
-  return useLiveQueryImpl<T>(query, params)
+  return useLiveQueryImpl<T>(query, params, undefined, options)
 }
 
 useLiveQuery.sql = function <T = { [key: string]: unknown }>(

packages/pglite-react/test/hooks-options.test.tsx

@@ -0,0 +1,71 @@
+import { act, renderHook } from '@testing-library/react'
+import { waitFor } from '@testing-library/dom'
+import { describe, expect, it, vi } from 'vitest'
+import type { LiveQueryResults } from '@electric-sql/pglite/live'
+import { useLiveQuery } from '../src/hooks'
+
+const { usePGliteMock } = vi.hoisted(() => ({
+  usePGliteMock: vi.fn(),
+}))
+
+vi.mock('../src/provider', () => ({
+  usePGlite: usePGliteMock,
+}))
+
+describe('useLiveQuery query options', () => {
+  it('passes options to live.query for initial and updated results', async () => {
+    type Row = [number, string]
+    let callback: ((results: LiveQueryResults<Row>) => void) | undefined
+    const initialResults: LiveQueryResults<Row> = {
+      rows: [[1, 'initial']],
+      fields: [
+        { name: 'id', dataTypeID: 23 },
+        { name: 'name', dataTypeID: 25 },
+      ],
+    }
+    const query = vi.fn(async (...args: unknown[]) => {
+      callback = args.find(
+        (arg): arg is (results: LiveQueryResults<Row>) => void =>
+          typeof arg === 'function',
+      )
+      callback?.(initialResults)
+      return {
+        initialResults,
+        subscribe: vi.fn(),
+        unsubscribe: vi.fn(),
+        refresh: vi.fn(),
+      }
+    })
+    usePGliteMock.mockReturnValue({ live: { query } })
+
+    const { result } = renderHook(() =>
+      useLiveQuery<Row>('SELECT id, name FROM test', [], {
+        rowMode: 'array',
+      }),
+    )
+
+    await waitFor(() => expect(result.current).toEqual(initialResults))
+    expect(query).toHaveBeenCalledTimes(1)
+    expect(query).toHaveBeenCalledWith(
+      'SELECT id, name FROM test',
+      [],
+      { rowMode: 'array' },
+      expect.any(Function),
+    )
+
+    act(() => {
+      callback?.({
+        ...initialResults,
+        rows: [
+          [1, 'initial'],
+          [2, 'updated'],
+        ],
+      })
+    })
+
+    expect(result.current?.rows).toEqual([
+      [1, 'initial'],
+      [2, 'updated'],
+    ])
+  })
+})

packages/pglite-react/test/hooks.test-d.tsx

@@ -0,0 +1,22 @@
+import { describe, it, expectTypeOf } from 'vitest'
+import type { QueryOptions } from '@electric-sql/pglite'
+import type { LiveQueryOptions } from '@electric-sql/pglite/live'
+import { useLiveQuery } from '../src'
+
+describe('useLiveQuery types', () => {
+  it('accepts exported query options in object and positional APIs', () => {
+    const queryOptions: QueryOptions = { rowMode: 'array' }
+    const liveOptions: LiveQueryOptions<[number, string]> = {
+      query: 'SELECT id, name FROM test',
+      ...queryOptions,
+    }
+
+    expectTypeOf(liveOptions.rowMode).toEqualTypeOf<QueryOptions['rowMode']>()
+    ;() =>
+      useLiveQuery<[number, string]>(
+        'SELECT id, name FROM test',
+        [],
+        queryOptions,
+      )
+  })
+})

packages/pglite/src/live/index.ts

@@ -3,6 +3,7 @@ import type {
   PGliteInterface,
   Results,
   Transaction,
+  QueryOptions,
 } from '../interface'
 import type {
   LiveQueryOptions,
@@ -23,6 +24,7 @@ export type {
   LiveChanges,
   Change,
   LiveQueryResults,
+  LiveQueryOptions,
 } from './interface.js'
 
 const MAX_RETRIES = 5
@@ -36,18 +38,44 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
     async query<T>(
       query: string | LiveQueryOptions<T>,
       params?: any[] | null,
+      optionsOrCallback?: QueryOptions | ((results: Results<T>) => void),
       callback?: (results: Results<T>) => void,
     ) {
       let signal: AbortSignal | undefined
       let offset: number | undefined
       let limit: number | undefined
+      let options: QueryOptions | undefined
       if (typeof query !== 'string') {
-        signal = query.signal
-        params = query.params
-        callback = query.callback
-        offset = query.offset
-        limit = query.limit
-        query = query.query
+        const {
+          signal: querySignal,
+          params: queryParams,
+          callback: queryCallback,
+          offset: queryOffset,
+          limit: queryLimit,
+          query: queryString,
+          ...queryOptions
+        } = query
+        signal = querySignal
+        params = queryParams
+        callback = queryCallback
+        offset = queryOffset
+        limit = queryLimit
+        options = queryOptions
+        query = queryString
+      } else if (typeof optionsOrCallback === 'function') {
+        callback = optionsOrCallback
+      } else {
+        options = optionsOrCallback
+      }
+
+      // The prepared EXECUTE has no bind parameters, and live-query metadata
+      // queries rely on object rows. Only pass options that apply to the user's
+      // result rows or notices.
+      const resultOptions: QueryOptions | undefined = options && {
+        rowMode: options.rowMode,
+        parsers: options.parsers,
+        blob: options.blob,
+        onNotice: options.onNotice,
       }
 
       // Offset and limit must be provided together
@@ -82,7 +110,7 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
           // Create a temporary view with the query
           const formattedQuery =
             params && params.length > 0
-              ? await formatQuery(pg, query, params, tx)
+              ? await formatQuery(pg, query, params, tx, options)
               : query
           await tx.exec(
             `CREATE OR REPLACE TEMP VIEW live_query_${id}_view AS ${formattedQuery}`,
@@ -110,6 +138,8 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
             results = {
               ...(await tx.query<T>(
                 `EXECUTE live_query_${id}_get(${limit}, ${offset});`,
+                undefined,
+                resultOptions,
               )),
               offset,
               limit,
@@ -120,7 +150,11 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
               PREPARE live_query_${id}_get AS
               SELECT * FROM live_query_${id}_view;
             `)
-            results = await tx.query<T>(`EXECUTE live_query_${id}_get;`)
+            results = await tx.query<T>(
+              `EXECUTE live_query_${id}_get;`,
+              undefined,
+              resultOptions,
+            )
           }
           // Setup the listeners
           unsubList = await Promise.all(
@@ -178,13 +212,19 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
                 results = {
                   ...(await pg.query<T>(
                     `EXECUTE live_query_${id}_get(${limit}, ${offset});`,
+                    undefined,
+                    resultOptions,
                   )),
                   offset,
                   limit,
                   totalCount, // This is the old total count
                 }
               } else {
-                results = await pg.query<T>(`EXECUTE live_query_${id}_get;`)
+                results = await pg.query<T>(
+                  `EXECUTE live_query_${id}_get;`,
+                  undefined,
+                  resultOptions,
+                )
               }
             } catch (e) {
               const msg = (e as Error).message

packages/pglite/src/live/interface.ts

@@ -1,6 +1,7 @@
-import type { Results } from '../interface'
+import type { QueryOptions, Results } from '../interface'
 
-export interface LiveQueryOptions<T = { [key: string]: any }> {
+export interface LiveQueryOptions<T = { [key: string]: any }>
+  extends QueryOptions {
   query: string
   params?: any[] | null
   offset?: number
@@ -40,6 +41,22 @@ export interface LiveNamespace {
     callback?: (results: Results<T>) => void,
   ): Promise<LiveQuery<T>>
 
+  /**
+   * Create a live query with query options
+   * @param query - The query to run
+   * @param params - The parameters to pass to the query
+   * @param options - The options to apply to the query results
+   * @param callback - A callback to run when the query is updated
+   * @returns A promise that resolves to an object with the initial results,
+   * an unsubscribe function, and a refresh function
+   */
+  query<T = { [key: string]: any }>(
+    query: string,
+    params: any[] | undefined | null,
+    options: QueryOptions,
+    callback?: (results: Results<T>) => void,
+  ): Promise<LiveQuery<T>>
+
   /**
    * Create a live query
    * @param options - The options to pass to the query