Electric 컬렉션

Electric 컬렉션은 TanStack DB와 ElectricSQL을 원활하게 통합하여 Electric의 동기화 엔진을 통해 Postgres 데이터베이스와 실시간 데이터 동기화를 제공합니다.

개요

@tanstack/electric-db-collection 패키지를 사용하면 다음과 같은 컬렉션을 만들 수 있습니다:

• Electric 셰이프를 통해 Postgres의 데이터를 자동으로 동기화합니다
• 트랜잭션 매칭과 오류 발생 시 자동 롤백을 사용하여 낙관적 업데이트를 지원합니다
• 사용자 지정 뮤테이션 핸들러를 통해 영속성을 처리합니다

설치

npm install @tanstack/electric-db-collection @tanstack/react-db

기본 사용법

import { createCollection } from '@tanstack/react-db'
import { electricCollectionOptions } from '@tanstack/electric-db-collection'

const todosCollection = createCollection(
  electricCollectionOptions({
    shapeOptions: {
      url: '/api/todos',
    },
    getKey: (item) => item.id,
  })
)

구성 옵션

electricCollectionOptions 함수는 다음 옵션을 허용합니다:

필수 옵션

• shapeOptions: ElectricSQL ShapeStream 구성입니다

  • url: Electric에 연결하는 프록시의 URL입니다

• getKey: 항목에서 고유 키를 추출하는 함수입니다

선택 사항

• id: 컬렉션의 고유 식별자입니다
• schema: 항목을 검증하기 위한 스키마입니다. 모든 Standard Schema 호환 스키마를 사용할 수 있습니다
• sync: 사용자 지정 동기화 구성입니다

영속성 핸들러

핸들러는 변경 사항을 백엔드에 영속화하기 위해 뮤테이션 전에 호출됩니다:

• onInsert: 삽입 작업 전에 호출되는 핸들러입니다
• onUpdate: 업데이트 작업 전에 호출되는 핸들러입니다
• onDelete: 삭제 작업 전에 호출되는 핸들러입니다

각 핸들러는 동기화를 기다리기 위해 { txid }를 반환해야 합니다. API가 txid를 반환할 수 없는 경우에는 awaitMatch 유틸리티 함수를 사용합니다.

영속성 핸들러 및 동기화

핸들러는 뮤테이션을 백엔드에 영속화하고 Electric이 변경 사항을 다시 동기화할 때까지 기다립니다. 이를 통해 낙관적 업데이트가 제거되었다가 다시 추가되는 UI 깜박임을 방지합니다. TanStack DB는 뮤테이션이 확인될 때까지 동기화 데이터를 차단하여 원활한 사용자 경험을 보장합니다.

1. Txid 사용(권장)

권장 방식은 정확한 매칭을 위해 PostgreSQL 트랜잭션 ID(txid)를 사용합니다. 백엔드는 txid를 반환하고 클라이언트는 해당 txid가 Electric 스트림에 나타날 때까지 기다립니다.

const todosCollection = createCollection(
  electricCollectionOptions({
    id: 'todos',
    schema: todoSchema,
    getKey: (item) => item.id,
    shapeOptions: {
      url: '/api/todos',
      params: { table: 'todos' },
    },

    onInsert: async ({ transaction }) => {
      const newItem = transaction.mutations[0].modified
      const response = await api.todos.create(newItem)

      // Return txid to wait for sync
      return { txid: response.txid }
    },

    onUpdate: async ({ transaction }) => {
      const { original, changes } = transaction.mutations[0]
      const response = await api.todos.update({
        where: { id: original.id },
        data: changes
      })

      return { txid: response.txid }
    }
  })
)

2. 사용자 지정 매칭 함수 사용

txid를 사용할 수 없는 경우 awaitMatch 유틸리티 함수를 사용하여 사용자 지정 매칭 로직으로 동기화를 기다립니다:

import { isChangeMessage } from '@tanstack/electric-db-collection'

const todosCollection = createCollection(
  electricCollectionOptions({
    id: 'todos',
    getKey: (item) => item.id,
    shapeOptions: {
      url: '/api/todos',
      params: { table: 'todos' },
    },

    onInsert: async ({ transaction, collection }) => {
      const newItem = transaction.mutations[0].modified
      await api.todos.create(newItem)

      // Use awaitMatch utility for custom matching
      await collection.utils.awaitMatch(
        (message) => {
          return isChangeMessage(message) &&
                 message.headers.operation === 'insert' &&
                 message.value.text === newItem.text
        },
        5000 // timeout in ms (optional, defaults to 3000)
      )
    }
  })
)

3. 단순 타임아웃 사용

빠른 프로토타이핑을 하거나 타이밍을 확신할 수 있는 경우 단순한 타임아웃을 사용할 수 있습니다. 이 방법은 단순하지만 데이터가 거의 항상 2초 이내에 다시 동기화되므로 작동합니다:

const todosCollection = createCollection(
  electricCollectionOptions({
    id: 'todos',
    getKey: (item) => item.id,
    shapeOptions: {
      url: '/api/todos',
      params: { table: 'todos' },
    },

    onInsert: async ({ transaction }) => {
      const newItem = transaction.mutations[0].modified
      await api.todos.create(newItem)

      // Simple timeout approach
      await new Promise(resolve => setTimeout(resolve, 2000))
    }
  })
)

백엔드에서는 Postgres를 직접 쿼리하여 트랜잭션의 txid를 추출할 수 있습니다.

async function generateTxId(tx) {
  // The ::xid cast strips off the epoch, giving you the raw 32-bit value
  // that matches what PostgreSQL sends in logical replication streams
  // (and then exposed through Electric which we'll match against
  // in the client).
  const result = await tx.execute(
    sql`SELECT pg_current_xact_id()::xid::text as txid`
  )
  const txid = result.rows[0]?.txid

  if (txid === undefined) {
    throw new Error(`Failed to get transaction ID`)
  }

  return parseInt(txid as string, 10)
}

Electric 프록시 예제

Electric은 일반적으로 셰이프 구성, 인증 및 권한 부여를 처리하는 프록시 서버 뒤에 배포됩니다. 이를 통해 보안이 강화되고 Electric을 클라이언트에 노출하지 않으면서 사용자가 액세스할 수 있는 데이터를 제어할 수 있습니다.

다음은 TanStack Start를 사용한 프록시 구현 예제입니다:

import { createServerFileRoute } from "@tanstack/react-start/server"
import { ELECTRIC_PROTOCOL_QUERY_PARAMS } from "@electric-sql/client"

// Electric URL
const baseUrl = 'http://.../v1/shape'

const serve = async ({ request }: { request: Request }) => {
  // ...check user authorization  
  const url = new URL(request.url)
  const originUrl = new URL(baseUrl)

  // passthrough parameters from electric client
  url.searchParams.forEach((value, key) => {
    if (ELECTRIC_PROTOCOL_QUERY_PARAMS.includes(key)) {
      originUrl.searchParams.set(key, value)
    }
  })

  // set shape parameters 
  // full spec: https://github.com/electric-sql/electric/blob/main/website/electric-api.yaml
  originUrl.searchParams.set("table", "todos")
  // Where clause to filter rows in the table (optional).
  // originUrl.searchParams.set("where", "completed = true")
  
  // Select the columns to sync (optional)
  // originUrl.searchParams.set("columns", "id,text,completed")

  const response = await fetch(originUrl)
  const headers = new Headers(response.headers)
  headers.delete("content-encoding")
  headers.delete("content-length")

  return new Response(response.body, {
    status: response.status,
    statusText: response.statusText,
    headers,
  })
}

export const ServerRoute = createServerFileRoute("/api/todos").methods({
  GET: serve,
})

명시적 트랜잭션을 사용한 낙관적 업데이트

고급 사용 사례에서는 여러 컬렉션에 걸쳐 여러 뮤테이션을 트랜잭션 방식으로 수행하는 사용자 지정 작업을 만들 수 있습니다. 유틸리티 메서드를 사용하면 다양한 전략으로 동기화를 기다릴 수 있습니다:

Txid 전략 사용

const addTodoAction = createOptimisticAction({
  onMutate: ({ text }) => {
    // optimistically insert with a temporary ID
    const tempId = crypto.randomUUID()
    todosCollection.insert({
      id: tempId,
      text,
      completed: false,
      created_at: new Date(),
    })
    
    // ... mutate other collections
  },
  
  mutationFn: async ({ text }) => {
    const response = await api.todos.create({
      data: { text, completed: false }
    })
    
    // Wait for the specific txid
    await todosCollection.utils.awaitTxId(response.txid)
  }
})

사용자 지정 매칭 함수 사용

import { isChangeMessage } from '@tanstack/electric-db-collection'

const addTodoAction = createOptimisticAction({
  onMutate: ({ text }) => {
    const tempId = crypto.randomUUID()
    todosCollection.insert({
      id: tempId,
      text,
      completed: false,
      created_at: new Date(),
    })
  },
  
  mutationFn: async ({ text }) => {
    await api.todos.create({
      data: { text, completed: false }
    })
    
    // Wait for matching message
    await todosCollection.utils.awaitMatch(
      (message) => {
        return isChangeMessage(message) && 
               message.headers.operation === 'insert' &&
               message.value.text === text
      }
    )
  }
})

유틸리티 메서드

컬렉션은 collection.utils를 통해 다음 유틸리티 메서드를 제공합니다:

awaitTxId(txid, timeout?)

특정 트랜잭션 ID가 동기화될 때까지 수동으로 기다립니다:

// Wait for specific txid
await todosCollection.utils.awaitTxId(12345)

// With custom timeout (default is 5 seconds)
await todosCollection.utils.awaitTxId(12345, 10000)

다른 작업을 진행하기 전에 뮤테이션이 동기화되었는지 확인해야 할 때 유용합니다.

awaitMatch(matchFn, timeout?)

사용자 지정 매칭 함수가 일치하는 메시지를 찾을 때까지 수동으로 기다립니다:

import { isChangeMessage } from '@tanstack/electric-db-collection'

// Wait for a specific message pattern
await todosCollection.utils.awaitMatch(
  (message) => {
    return isChangeMessage(message) &&
           message.headers.operation === 'insert' &&
           message.value.text === 'New Todo'
  },
  5000 // timeout in ms
)

헬퍼 함수

패키지는 사용자 지정 매칭 함수에서 사용할 헬퍼 함수를 내보냅니다:

• isChangeMessage(message): 메시지가 데이터 변경(삽입/업데이트/삭제)인지 확인합니다
• isControlMessage(message): 메시지가 제어 메시지(최신 상태, 다시 가져오기 필요)인지 확인합니다

import { isChangeMessage, isControlMessage } from '@tanstack/electric-db-collection'

// Use in custom match functions
const matchFn = (message) => {
  if (isChangeMessage(message)) {
    return message.headers.operation === 'insert'
  }
  return false
}

디버깅

일반적인 문제: awaitTxId가 중단되거나 시간 초과됨

개발자가 자주 겪는 문제는 awaitTxId(또는 트랜잭션의 isPersisted.promise)가 무기한 중단되고 결국 오류 메시지 없이 시간 초과되는 현상입니다. 데이터는 데이터베이스에 정상적으로 영속화되지만 낙관적 뮤테이션은 절대 해결되지 않습니다.

근본 원인: API에서 반환된 트랜잭션 ID(txid)가 Postgres에서 뮤테이션에 사용된 실제 트랜잭션 ID와 일치하지 않을 때 발생합니다. 뮤테이션을 수행하는 동일한 트랜잭션 외부에서 pg_current_xact_id()를 쿼리하면 이러한 불일치가 발생합니다.

디버그 로깅 활성화

txid 문제를 진단하려면 브라우저 콘솔에서 디버그 로깅을 활성화합니다:

localStorage.debug = 'ts/db:electric'

이를 통해 뮤테이션이 txid를 기다리기 시작하는 시점과 Electric 동기화 스트림에서 txid가 도착하는 시점을 확인할 수 있습니다.

이는 debug 패키지로 구동됩니다.

txid가 일치하지 않는 경우(일반적인 버그):

ts/db:electric awaitTxId called with txid 124
ts/db:electric new txids synced from pg [123]
// Stalls forever - 124 never arrives!

이 예제에서는 뮤테이션이 트랜잭션 123에서 발생했지만, 뮤테이션 이후 실행된 별도의 트랜잭션(124)에서 pg_current_xact_id()를 쿼리했습니다. 클라이언트는 도착하지 않을 124를 기다립니다.

txid가 일치하는 경우(올바른 방식):

ts/db:electric awaitTxId called with txid 123
ts/db:electric new txids synced from pg [123]
ts/db:electric awaitTxId found match for txid 123
// Resolves immediately!

해결 방법: 트랜잭션 내부에서 txid 쿼리

동일한 트랜잭션에서 뮤테이션을 수행하는 동안 pg_current_xact_id()를 호출해야 합니다:

❌ 잘못된 방식 - 트랜잭션 외부에서 txid 쿼리:

// DON'T DO THIS
async function createTodo(data) {
  const txid = await generateTxId(sql) // Wrong: separate transaction

  await sql.begin(async (tx) => {
    await tx`INSERT INTO todos ${tx(data)}`
  })

  return { txid } // This txid won't match!
}

✅ 올바른 방식 - 트랜잭션 내부에서 txid 쿼리:

// DO THIS
async function createTodo(data) {
  let txid!: Txid

  const result = await sql.begin(async (tx) => {
    // Call generateTxId INSIDE the transaction
    txid = await generateTxId(tx)

    const [todo] = await tx`
      INSERT INTO todos ${tx(data)}
      RETURNING *
    `
    return todo
  })

  return { todo: result, txid } // txid matches the mutation
}

async function generateTxId(tx: any): Promise<Txid> {
  const result = await tx`SELECT pg_current_xact_id()::xid::text as txid`
  const txid = result[0]?.txid

  if (txid === undefined) {
    throw new Error(`Failed to get transaction ID`)
  }

  return parseInt(txid, 10)
}

작동하는 예제는 다음에서 확인할 수 있습니다:

• examples/react/todo/src/routes/api/todos.ts
• examples/react/todo/src/api/server.ts