본문으로 건너뛰기

컬렉션 옵션 생성기 만들기

컬렉션 옵션 생성기는 TanStack DB 컬렉션의 구성 옵션을 생성하는 팩토리 함수입니다. 다양한 동기화 엔진과 데이터 소스를 TanStack DB의 반응형 sync-first 아키텍처와 통합하는 표준화된 방법을 제공합니다.

개요​

컬렉션 옵션 생성기는 일관된 패턴을 따릅니다:

  1. 동기화 엔진에 맞는 구성을 받습니다
  2. CollectionConfig 인터페이스를 충족하는 객체를 반환합니다
  3. 동기화 초기화, 데이터 구문 분석 및 트랜잭션 관리를 처리합니다
  4. 동기화 엔진에 특화된 유틸리티 함수를 선택적으로 제공합니다

사용자 지정 컬렉션을 만들어야 하는 경우​

다음과 같은 경우 사용자 지정 컬렉션을 만들어야 합니다:

  • 전용 동기화 엔진이 있는 경우(ElectricSQL, Trailbase, Firebase, RxDB 또는 사용자 지정 WebSocket 솔루션 등)
  • 쿼리 컬렉션에서 지원하지 않는 특정 동기화 동작이 필요한 경우
  • 자체 동기화 프로토콜을 사용하는 백엔드와 통합하려는 경우

참고: API를 호출하고 데이터를 반환하기만 한다면 쿼리 컬렉션을 사용합니다.

핵심 요구 사항​

모든 컬렉션 옵션 생성기는 다음과 같은 주요 책임을 구현해야 합니다:

1. 구성 인터페이스​

표준 컬렉션 속성을 확장하거나 포함하는 구성 인터페이스를 정의합니다:

// Pattern A: User provides handlers (Query / ElectricSQL style)
interface MyCollectionConfig<TItem extends object> {
  // Your sync engine specific options
  connectionUrl: string
  apiKey?: string
  
  // Standard collection properties
  id?: string
  schema?: StandardSchemaV1
  getKey: (item: TItem) => string | number
  sync?: SyncConfig<TItem>
  
  rowUpdateMode?: 'partial' | 'full'
  
  // User provides mutation handlers
  onInsert?: InsertMutationFn<TItem>
  onUpdate?: UpdateMutationFn<TItem>
  onDelete?: DeleteMutationFn<TItem>
}

// Pattern B: Built-in handlers (Trailbase style)
interface MyCollectionConfig<TItem extends object> 
  extends Omit<CollectionConfig<TItem>, 'onInsert' | 'onUpdate' | 'onDelete'> {
  // Your sync engine specific options
  recordApi: MyRecordApi<TItem>
  connectionUrl: string
  
  rowUpdateMode?: 'partial' | 'full'
  
  // Note: onInsert/onUpdate/onDelete are implemented by your collection creator
}

2. 동기화 구현​

sync 함수는 컬렉션의 핵심입니다. 다음을 수행해야 합니다:

sync 함수는 적절한 가비지 컬렉션을 위해 정리 함수를 반환해야 합니다:

const sync: SyncConfig<T>['sync'] = (params) => {
  const { begin, write, commit, markReady, markError, collection } = params
  
  // 1. Initialize connection to your sync engine
  const connection = initializeConnection(config)
  const initialSyncAbort = new AbortController()
  
  // 2. Set up real-time subscription FIRST (prevents race conditions)
  const eventBuffer: Array<any> = []
  let isInitialSyncComplete = false
  
  connection.subscribe((event) => {
    if (!isInitialSyncComplete) {
      // Buffer events during initial sync to prevent race conditions
      eventBuffer.push(event)
      return
    }
    
    // Process real-time events
    begin()
    
    switch (event.type) {
      case 'insert':
        write({ type: 'insert', value: event.data })
        break
      case 'update':
        write({ type: 'update', value: event.data })
        break
      case 'delete':
        write({ type: 'delete', value: event.data })
        break
    }
    
    commit()
  })
  
  // 3. Perform initial data fetch
  async function initialSync() {
    try {
      const data = await fetchInitialData({ signal: initialSyncAbort.signal })
      
      begin() // Start a transaction
      
      for (const item of data) {
        write({
          type: 'insert',
          value: item
        })
      }
      
      commit() // Commit the transaction
      
      // 4. Process buffered events
      isInitialSyncComplete = true
      if (eventBuffer.length > 0) {
        begin()
        for (const event of eventBuffer) {
          // Deduplicate if necessary based on your sync engine
          write({ type: event.type, value: event.data })
        }
        commit()
        eventBuffer.splice(0)
      }

      // A complete initial snapshot is now available.
      markReady()
    } catch (error) {
      if (initialSyncAbort.signal.aborted) return
      console.error('Initial sync failed:', error)
      // No usable initial snapshot exists.
      // Only initial startup owns collection readiness. A later refetch
      // failure must keep the last ready snapshot usable.
      if (collection.status === 'loading') markError(error)
    }
  }

  initialSync()
  
  // 4. Return cleanup function
  return () => {
    initialSyncAbort.abort()
    connection.close()
    // Clean up any timers, intervals, or other resources
  }
}

3. 트랜잭션 수명 주기​

올바르게 구현하려면 트랜잭션 수명 주기를 이해하는 것이 중요합니다.

동기화 프로세스는 다음 수명 주기를 따릅니다:

  1. begin() - 변경 사항 수집을 시작합니다
  2. write() - 보류 중인 트랜잭션에 변경 사항을 추가합니다(커밋될 때까지 버퍼링됨)
  3. commit() - 모든 변경 사항을 컬렉션 상태에 원자적으로 적용합니다
  4. markReady() - 사용할 수 있는 초기 또는 복구된 스냅샷이 있음을 알립니다
  5. markError(error?) - 사용할 수 있는 스냅샷을 생성하기 전에 초기 동기화가 실패했음을 알립니다. readiness 대기가 해당 오류와 함께 거부되도록 원인을 전달합니다

경쟁 조건 방지: 많은 동기화 엔진은 초기 동기화가 완료되기 전에 실시간 구독을 시작합니다. 초기 동기화와 동일한 데이터를 나타내는 구독 이벤트가 중복 처리되지 않도록 구현에서 반드시 중복 제거해야 합니다. 다음을 고려합니다:

  • 초기 가져오기 전에 리스너를 시작하고 이벤트를 버퍼링합니다
  • 타임스탬프, 시퀀스 번호 또는 문서 버전을 추적합니다
  • 읽기 타임스탬프 또는 기타 순서 지정 메커니즘을 사용합니다

4. 데이터 구문 분석 및 타입 변환​

동기화 엔진이 서로 다른 타입의 데이터를 반환하는 경우 특정 필드에 대한 변환 함수를 제공합니다:

interface MyCollectionConfig<TItem, TRecord> {
  // ... other config
  
  // Only specify conversions for fields that need type conversion
  parse: {
    created_at: (ts: number) => new Date(ts * 1000),  // timestamp -> Date
    updated_at: (ts: number) => new Date(ts * 1000),  // timestamp -> Date
    metadata?: (str: string) => JSON.parse(str)       // JSON string -> object
  }
  
  serialize: {
    created_at: (date: Date) => Math.floor(date.valueOf() / 1000),  // Date -> timestamp
    updated_at: (date: Date) => Math.floor(date.valueOf() / 1000),  // Date -> timestamp  
    metadata?: (obj: object) => JSON.stringify(obj)                 // object -> JSON string
  }
}

타입 변환 예시:

// Firebase Timestamp to Date
parse: {
  createdAt: (timestamp) => timestamp?.toDate?.() || new Date(timestamp),
  updatedAt: (timestamp) => timestamp?.toDate?.() || new Date(timestamp),
}

// PostGIS geometry to GeoJSON
parse: {
  location: (wkb: string) => parseWKBToGeoJSON(wkb)
}

// JSON string to object with error handling
parse: {
  metadata: (str: string) => {
    try {
      return JSON.parse(str)
    } catch {
      return {}
    }
  }
}

5. 스키마 및 타입 변환​

사용자 지정 컬렉션을 만들 때는 백엔드의 저장 형식과 사용자가 컬렉션에서 사용하는 클라이언트 측 타입 간의 관계를 처리하는 방법을 결정해야 합니다.

두 가지 별도 관심사​

백엔드 형식 - 저장소 계층에서 사용하는 타입(SQLite, Postgres, Firebase 등)

  • 예시: Unix 타임스탬프, ISO 문자열, JSON 문자열, PostGIS 지오메트리

클라이언트 형식 - 사용자가 TanStack DB 컬렉션에서 사용하는 타입

  • 예시: Date 객체, 구문 분석된 JSON, GeoJSON

TanStack DB의 스키마는 클라이언트 형식(뮤테이션의 TInput/TOutput)을 정의합니다. 백엔드 형식과 클라이언트 형식을 연결하는 방법은 통합 설계에 따라 달라집니다.

접근 방식 1: 통합에서 구문 분석/직렬화 헬퍼 제공​

특정 저장 형식을 사용하는 백엔드의 경우 사용자가 구성하는 parse/serialize 옵션을 제공합니다:

// TrailBase example: User specifies field conversions
export function trailbaseCollectionOptions(config) {
  return {
    parse: config.parse,      // User provides field conversions
    serialize: config.serialize,

    onInsert: async ({ transaction }) => {
      const serialized = transaction.mutations.map(m =>
        serializeFields(m.modified, config.serialize)
      )
      await config.recordApi.createBulk(serialized)
    }
  }
}

// User explicitly configures conversions
const collection = createCollection(
  trailbaseCollectionOptions({
    schema: todoSchema,
    parse: {
      created_at: (ts: number) => new Date(ts * 1000)  // Unix → Date
    },
    serialize: {
      created_at: (date: Date) => Math.floor(date.valueOf() / 1000)  // Date → Unix
    }
  })
)

이점: 타입 변환을 명시적으로 제어할 수 있습니다. 통합이 해당 변환을 일관되게 적용합니다.

접근 방식 2: 사용자가 QueryFn/핸들러에서 모든 작업 처리​

간단한 API를 사용하거나 사용자가 완전한 제어를 원할 때는 구문 분석/직렬화를 직접 처리합니다:

// Query Collection: User handles all transformations
const collection = createCollection(
  queryCollectionOptions({
    schema: todoSchema,
    queryFn: async () => {
      const response = await fetch('/api/todos')
      const todos = await response.json()
      // User manually parses to match their schema's TOutput
      return todos.map(todo => ({
        ...todo,
        created_at: new Date(todo.created_at)  // ISO string → Date
      }))
    },
    onInsert: async ({ transaction }) => {
      // User manually serializes for their backend
      await fetch('/api/todos', {
        method: 'POST',
        body: JSON.stringify({
          ...transaction.mutations[0].modified,
          created_at: transaction.mutations[0].modified.created_at.toISOString()  // Date → ISO string
        })
      })
    }
  })
)

이점: 유연성이 가장 높고 추상화 오버헤드가 없습니다. 사용자는 정확히 무슨 일이 일어나는지 확인할 수 있습니다.

접근 방식 3: 핸들러에서 자동 직렬화​

백엔드에 잘 정의된 타입이 있다면 뮤테이션 핸들러에서 자동으로 직렬화할 수 있습니다:

export function myCollectionOptions(config) {
  return {
    onInsert: async ({ transaction }) => {
      // Automatically serialize known types for your backend
      const serialized = transaction.mutations.map(m => ({
        ...m.modified,
        // Date objects → Unix timestamps for your backend
        created_at: m.modified.created_at instanceof Date
          ? Math.floor(m.modified.created_at.valueOf() / 1000)
          : m.modified.created_at
      }))
      await backend.insert(serialized)
    }
  }
}

이점: 사용자에게 필요한 구성이 가장 적습니다. 통합이 백엔드 형식을 자동으로 처리합니다.

주요 설계 원칙​

  1. 스키마는 클라이언트 뮤테이션만 검증합니다 - 동기화 중 백엔드 데이터의 구문 분석 방식에는 영향을 주지 않습니다
  2. TOutput은 애플리케이션에서 사용하는 타입입니다 - 사용자가 애플리케이션에서 다루는 타입입니다
  3. 백엔드 제약 조건에 따라 접근 방식을 선택합니다 - 고정된 타입 → 자동 직렬화, 가변적인 타입 → 사용자 구성
  4. 백엔드 형식을 명확하게 문서화합니다 - 저장소에서 사용하는 타입과 처리 방법을 설명합니다

사용자 관점의 스키마에 대한 자세한 내용은 스키마 가이드를 참조합니다.

6. 뮤테이션 핸들러 패턴​

컬렉션 옵션 생성기에서 뮤테이션을 처리하는 방식에는 서로 다른 두 가지 패턴이 있습니다:

패턴 A: 사용자 제공 핸들러(ElectricSQL, Query)​

사용자가 구성에 뮤테이션 핸들러를 제공합니다. 컬렉션 생성기는 이를 전달합니다:

interface MyCollectionConfig<TItem extends object> {
  // ... other config
  
  // User provides these handlers
  onInsert?: InsertMutationFn<TItem>
  onUpdate?: UpdateMutationFn<TItem>
  onDelete?: DeleteMutationFn<TItem>
}

export function myCollectionOptions<TItem extends object>(
  config: MyCollectionConfig<TItem>
) {
  return {
    // ... other options
    rowUpdateMode: config.rowUpdateMode || 'partial',
    
    // Pass through user-provided handlers (possibly with additional logic)
    onInsert: config.onInsert ? async (params) => {
      const result = await config.onInsert!(params)
      // Additional sync coordination logic
      return result
    } : undefined
  }
}

패턴 B: 기본 제공 핸들러(Trailbase, WebSocket, Firebase)​

컬렉션 생성기가 동기화 엔진의 API를 사용하여 핸들러를 직접 구현합니다:

interface MyCollectionConfig<TItem extends object> 
  extends Omit<CollectionConfig<TItem>, 'onInsert' | 'onUpdate' | 'onDelete'> {
  // ... sync engine specific config
  // Note: onInsert/onUpdate/onDelete are NOT in the config
}

export function myCollectionOptions<TItem extends object>(
  config: MyCollectionConfig<TItem>
) {
  return {
    // ... other options
    rowUpdateMode: config.rowUpdateMode || 'partial',
    
    // Implement handlers using sync engine APIs
    onInsert: async ({ transaction }) => {
      // Handle provider-specific batch limits (e.g., Firestore's 500 limit)
      const chunks = chunkArray(transaction.mutations, PROVIDER_BATCH_LIMIT)
      
      for (const chunk of chunks) {
        const ids = await config.recordApi.createBulk(
          chunk.map(m => serialize(m.modified))
        )
        await awaitIds(ids)
      }
      
      return transaction.mutations.map(m => m.key)
    },
    
    onUpdate: async ({ transaction }) => {
      const chunks = chunkArray(transaction.mutations, PROVIDER_BATCH_LIMIT)
      
      for (const chunk of chunks) {
        await Promise.all(
          chunk.map(m => 
            config.recordApi.update(m.key, serialize(m.changes))
          )
        )
      }
      
      await awaitIds(transaction.mutations.map(m => String(m.key)))
    }
  }
}

많은 공급자는 배치 크기 제한을 두므로(Firestore: 500, DynamoDB: 25 등) 큰 트랜잭션을 그에 맞게 청크로 나눕니다.

사용자가 자체 API를 제공해야 하는 경우 패턴 A를 선택하고, 동기화 엔진이 쓰기를 직접 처리하는 경우 패턴 B를 선택합니다.

행 업데이트 모드​

컬렉션은 두 가지 업데이트 모드를 지원합니다:

  • partial (기본값) - 업데이트가 기존 데이터와 병합됩니다
  • full - 업데이트가 전체 행을 대체합니다

동기화 구성에서 이를 설정합니다:

sync: {
  sync: syncFn,
  rowUpdateMode: 'full' // or 'partial'
}

프로덕션 예시​

완전하고 프로덕션에 바로 사용할 수 있는 예시는 TanStack DB 리포지토리의 컬렉션 패키지를 참조합니다:

프로덕션 컬렉션의 주요 교훈​

Query Collection에서:

  • 가장 간단한 접근 방식: 뮤테이션 후 전체 다시 가져오기
  • 적합한 경우: 실시간 동기화가 없는 API
  • 패턴: 사용자가 onInsert/onUpdate/onDelete 핸들러를 제공합니다

Trailbase Collection에서:

  • ID 기반 낙관적 상태 관리를 보여줍니다
  • 제공자의 배치 제한을 처리합니다(대규모 작업을 청크로 나눔)
  • 패턴: 컬렉션이 레코드 API를 사용하는 뮤테이션 핸들러를 제공합니다

Electric Collection에서:

  • 분산 동기화를 위한 복잡한 트랜잭션 ID 추적
  • 고급 중복 제거 기법을 보여줍니다
  • 동기화 조정을 위해 사용자 핸들러를 래핑하는 방법을 보여줍니다

RxDB Collection에서:

  • RxDB의 기본 제공 쿼리와 변경 스트림을 사용합니다
  • RxCollection.$을 사용하여 삽입/업데이트/삭제를 구독하고 begin-write-commit을 통해 TanStack DB로 전달합니다
  • RxDB API(bulkUpsert, incrementalPatch, bulkRemove)를 호출하는 기본 제공 뮤테이션 핸들러(onInsert, onUpdate, onDelete)를 구현합니다

전체 예제: WebSocket 컬렉션​

다음은 전체 왕복 흐름을 보여주는 WebSocket 기반 컬렉션 옵션 생성기의 전체 예제입니다:

  1. 클라이언트가 모든 뮤테이션을 함께 배치한 트랜잭션을 보냅니다
  2. 서버가 트랜잭션을 처리하고 데이터를 수정할 수 있습니다(검증, 타임스탬프 등)
  3. 서버가 승인과 실제 처리된 데이터를 돌려보냅니다
  4. 클라이언트가 낙관적 상태를 삭제하기 전에 이 왕복을 기다립니다
import type {
  CollectionConfig,
  SyncConfig,
  InsertMutationFnParams,
  UpdateMutationFnParams,
  DeleteMutationFnParams,
  UtilsRecord
} from '@tanstack/db'

interface WebSocketMessage<T> {
  type: 'insert' | 'update' | 'delete' | 'sync' | 'transaction' | 'ack'
  data?: T | T[]
  mutations?: Array<{
    type: 'insert' | 'update' | 'delete'
    data: T
    id?: string
  }>
  transactionId?: string
  id?: string
}

interface WebSocketCollectionConfig<TItem extends object>
  extends Omit<CollectionConfig<TItem>, 'onInsert' | 'onUpdate' | 'onDelete' | 'sync'> {
  url: string
  reconnectInterval?: number
  
  // Note: onInsert/onUpdate/onDelete are handled by the WebSocket connection
  // Users don't provide these handlers
}

interface WebSocketUtils extends UtilsRecord {
  reconnect: () => void
  getConnectionState: () => 'connected' | 'disconnected' | 'connecting'
}

export function webSocketCollectionOptions<TItem extends object>(
  config: WebSocketCollectionConfig<TItem>
): CollectionConfig<TItem> & { utils: WebSocketUtils } {
  let ws: WebSocket | null = null
  let reconnectTimer: NodeJS.Timeout | null = null
  let connectionState: 'connected' | 'disconnected' | 'connecting' = 'disconnected'
  
  // Track pending transactions awaiting acknowledgment
  const pendingTransactions = new Map<string, {
    resolve: () => void
    reject: (error: Error) => void
    timeout: NodeJS.Timeout
  }>()
  
  const sync: SyncConfig<TItem>['sync'] = (params) => {
    const { begin, write, commit, markReady } = params
    
    function connect() {
      connectionState = 'connecting'
      ws = new WebSocket(config.url)
      
      ws.onopen = () => {
        connectionState = 'connected'
        // Request initial sync
        ws.send(JSON.stringify({ type: 'sync' }))
      }
      
      ws.onmessage = (event) => {
        const message: WebSocketMessage<TItem> = JSON.parse(event.data)
        
        switch (message.type) {
          case 'sync':
            // Initial sync with array of items
            begin()
            if (Array.isArray(message.data)) {
              for (const item of message.data) {
                write({ type: 'insert', value: item })
              }
            }
            commit()
            markReady()
            break
            
          case 'insert':
          case 'update':
          case 'delete':
            // Real-time updates from other clients
            begin()
            write({ 
              type: message.type, 
              value: message.data as TItem 
            })
            commit()
            break
            
          case 'ack':
            // Server acknowledged our transaction
            if (message.transactionId) {
              const pending = pendingTransactions.get(message.transactionId)
              if (pending) {
                clearTimeout(pending.timeout)
                pendingTransactions.delete(message.transactionId)
                pending.resolve()
              }
            }
            break
            
          case 'transaction':
            // Server sending back the actual data after processing our transaction
            if (message.mutations) {
              begin()
              for (const mutation of message.mutations) {
                write({
                  type: mutation.type,
                  value: mutation.data
                })
              }
              commit()
            }
            break
        }
      }
      
      ws.onerror = (error) => {
        console.error('WebSocket error:', error)
        connectionState = 'disconnected'
      }
      
      ws.onclose = () => {
        connectionState = 'disconnected'
        // Auto-reconnect
        if (!reconnectTimer) {
          reconnectTimer = setTimeout(() => {
            reconnectTimer = null
            connect()
          }, config.reconnectInterval || 5000)
        }
      }
    }
    
    // Start connection
    connect()
    
    // Return cleanup function
    return () => {
      if (reconnectTimer) {
        clearTimeout(reconnectTimer)
        reconnectTimer = null
      }
      if (ws) {
        ws.close()
        ws = null
      }
    }
  }
  
  // Helper function to send transaction and wait for server acknowledgment
  const sendTransaction = async (
    params: InsertMutationFnParams<TItem> | UpdateMutationFnParams<TItem> | DeleteMutationFnParams<TItem>
  ): Promise<void> => {
    if (ws?.readyState !== WebSocket.OPEN) {
      throw new Error('WebSocket not connected')
    }
    
    const transactionId = crypto.randomUUID()
    
    // Convert all mutations in the transaction to the wire format
    const mutations = params.transaction.mutations.map(mutation => ({
      type: mutation.type,
      id: mutation.key,
      data: mutation.type === 'delete' ? undefined : 
           mutation.type === 'update' ? mutation.changes : 
           mutation.modified
    }))
    
    // Send the entire transaction at once
    ws.send(JSON.stringify({
      type: 'transaction',
      transactionId,
      mutations
    }))
    
    // Wait for server acknowledgment
    return new Promise<void>((resolve, reject) => {
      const timeout = setTimeout(() => {
        pendingTransactions.delete(transactionId)
        reject(new Error(`Transaction ${transactionId} timed out`))
      }, 10000) // 10 second timeout
      
      pendingTransactions.set(transactionId, {
        resolve,
        reject,
        timeout
      })
    })
  }
  
  // All mutation handlers use the same transaction sender
  const onInsert = async (params: InsertMutationFnParams<TItem>) => {
    await sendTransaction(params)
  }
  
  const onUpdate = async (params: UpdateMutationFnParams<TItem>) => {
    await sendTransaction(params)
  }
  
  const onDelete = async (params: DeleteMutationFnParams<TItem>) => {
    await sendTransaction(params)
  }
  
  return {
    id: config.id,
    schema: config.schema,
    getKey: config.getKey,
    sync: { sync },
    onInsert,
    onUpdate,
    onDelete,
    utils: {
      reconnect: () => {
        if (ws) ws.close()
        connect()
      },
      getConnectionState: () => connectionState
    }
  }
}

사용 예제​

import { DbClient, collectionOptions } from '@tanstack/react-db'
import { webSocketCollectionOptions } from './websocket-collection'

const db = new DbClient()

const todosCollection = collectionOptions('todos', () =>
  webSocketCollectionOptions({
    id: 'todos',
    url: 'ws://localhost:8080/todos',
    getKey: (todo) => todo.id,
    schema: todoSchema,
    // Note: No onInsert/onUpdate/onDelete - handled by WebSocket automatically
  })
)

const todos = db.collection(todosCollection)

// Use the collection
todos.insert({ id: '1', text: 'Buy milk', completed: false })

// Access utilities
todos.utils.getConnectionState() // 'connected'
todos.utils.reconnect() // Force reconnect

고급: 낙관적 상태 관리​

동기화 우선 앱의 핵심 과제는 낙관적 상태를 언제 삭제할지 아는 것입니다. 사용자가 변경하면 다음과 같이 진행됩니다:

  1. UI가 즉시 업데이트됩니다(낙관적 업데이트)
  2. 뮤테이션이 백엔드로 전송됩니다
  3. 백엔드가 변경 사항을 처리하고 영속화합니다
  4. 변경 사항이 클라이언트로 다시 동기화됩니다
  5. 동기화된 데이터를 사용하기 위해 낙관적 상태를 삭제해야 합니다

핵심 질문은 다음과 같습니다: 4단계가 완료되었는지 어떻게 알 수 있습니까?

많은 프로바이더는 동기화 완료를 기다리는 기본 제공 메서드를 제공합니다:

// Firebase
await waitForPendingWrites(firestore)

// Custom WebSocket
await websocket.waitForAck(transactionId)

전략 2: 트랜잭션 ID 추적(ElectricSQL)​

ElectricSQL은 추적할 수 있는 트랜잭션 ID를 반환합니다:

// Track seen transaction IDs
const seenTxids = new Store<Set<number>>(new Set())

// In sync, track txids from incoming messages
if (message.headers.txids) {
  message.headers.txids.forEach(txid => {
    seenTxids.setState(prev => new Set([...prev, txid]))
  })
}

// Mutation handlers return txids and wait for them
const wrappedOnInsert = async (params) => {
  const result = await config.onInsert!(params)
  
  // Wait for the txid to appear in synced data
  if (result.txid) {
    await awaitTxId(result.txid)
  }
  
  return result
}

// Utility function to wait for a txid
const awaitTxId = (txId: number): Promise<boolean> => {
  if (seenTxids.state.has(txId)) return Promise.resolve(true)
  
  return new Promise((resolve) => {
    const unsubscribe = seenTxids.subscribe(() => {
      if (seenTxids.state.has(txId)) {
        unsubscribe()
        resolve(true)
      }
    })
  })
}

전략 3: ID 기반 추적(Trailbase)​

Trailbase는 특정 레코드 ID가 동기화된 시점을 추적합니다:

// Track synced IDs with timestamps
const seenIds = new Store(new Map<string, number>())

// In sync, mark IDs as seen
write({ type: 'insert', value: item })
seenIds.setState(prev => new Map(prev).set(item.id, Date.now()))

// Wait for specific IDs after mutations
const wrappedOnInsert = async (params) => {
  const ids = await config.recordApi.createBulk(items)
  
  // Wait for all IDs to be synced back
  await awaitIds(ids)
}

const awaitIds = (ids: string[]): Promise<void> => {
  const allSynced = ids.every(id => seenIds.state.has(id))
  if (allSynced) return Promise.resolve()
  
  return new Promise((resolve) => {
    const unsubscribe = seenIds.subscribe((state) => {
      if (ids.every(id => state.has(id))) {
        unsubscribe()
        resolve()
      }
    })
  })
}

전략 4: 버전/타임스탬프 추적​

데이터가 최신인지 감지하려면 버전 번호 또는 타임스탬프를 추적합니다:

// Track latest sync timestamp
let lastSyncTime = 0

// In mutations, record when the operation was sent
const wrappedOnUpdate = async (params) => {
  const mutationTime = Date.now()
  await config.onUpdate(params)
  
  // Wait for sync to catch up
  await waitForSync(mutationTime)
}

const waitForSync = (afterTime: number): Promise<void> => {
  if (lastSyncTime > afterTime) return Promise.resolve()
  
  return new Promise((resolve) => {
    const check = setInterval(() => {
      if (lastSyncTime > afterTime) {
        clearInterval(check)
        resolve()
      }
    }, 100)
  })
}

전략 5: 전체 다시 가져오기(Query Collection)​

Query Collection은 뮤테이션 후 모든 데이터를 간단히 다시 가져옵니다:

const wrappedOnInsert = async (params) => {
  // Perform the mutation
  await config.onInsert(params)
  
  // Refetch the entire collection
  await refetch()
  
  // The refetch will trigger sync with fresh data,
  // automatically dropping optimistic state
}

전략 선택​

  • 기본 제공 메서드: 프로바이더가 동기화 완료 API를 제공할 때 가장 적합합니다
  • 트랜잭션 ID: 백엔드가 신뢰할 수 있는 트랜잭션 추적을 제공할 때 가장 적합합니다
  • ID 기반: 각 뮤테이션이 영향을 받은 ID를 반환하는 시스템에 적합합니다
  • 전체 다시 가져오기: 가장 간단하지만 효율성이 가장 낮으며, 작은 데이터 세트에 적합합니다
  • 버전/타임스탬프: 동기화에 신뢰할 수 있는 순서 정보가 포함될 때 작동합니다

구현 팁​

  1. 낙관적 상태가 올바르게 관리되도록 뮤테이션 핸들러에서 항상 동기화를 기다립니다
  2. 시간 초과를 처리합니다 - 동기화 확인을 무한정 기다리지 않습니다
  3. 추적 데이터를 정리합니다 - 메모리 누수를 방지하기 위해 오래된 txid/ID를 제거합니다
  4. 유틸리티를 제공합니다 - 고급 사용 사례를 위해 awaitTxId 또는 awaitSync와 같은 함수를 내보냅니다

모범 사례​

  1. 초기 동기화 상태를 보고합니다 - 사용 가능한 스냅샷 이후 markReady()을 호출하거나, 초기 동기화가 실패하면 markError(error)을 호출합니다
  2. 명시적으로 복구합니다 - 오류 후 이후 동기화에서 사용 가능한 스냅샷이 생성된 경우에만 markReady()를 호출합니다
  3. 리소스를 정리합니다 - 메모리 누수를 방지하기 위해 동기화에서 정리 함수를 반환합니다
  4. 작업을 배치합니다 - 더 나은 성능을 위해 begin/commit을 사용하여 여러 변경 사항을 배치합니다
  5. 경쟁 조건 - 초기 가져오기 전에 리스너를 시작하고 이벤트를 버퍼링합니다
  6. 타입 안전성 - 전체 과정에서 타입 안전성을 유지하도록 TypeScript 제네릭을 사용합니다
  7. 유틸리티를 제공합니다 - 고급 사용 사례를 위해 동기화 엔진별 유틸리티를 내보냅니다

컬렉션 테스트​

다음 항목으로 컬렉션 옵션 생성기를 테스트합니다:

  1. 단위 테스트 - 동기화 로직과 데이터 변환을 테스트합니다
  2. 통합 테스트 - 실제 동기화 엔진으로 테스트합니다
  3. 오류 시나리오 - 연결 실패와 잘못된 데이터를 테스트합니다
  4. 성능 - 대규모 데이터 세트와 빈번한 업데이트를 테스트합니다

결론​

컬렉션 옵션 생성기를 만들면 모든 동기화 엔진을 TanStack DB의 강력한 동기화 우선 아키텍처와 통합할 수 있습니다. 여기에서 보여준 패턴을 따르면 뛰어난 개발자 경험을 제공하는 견고하고 타입 안전한 통합을 구현할 수 있습니다.