Electric Collection

Electric Collections bieten eine nahtlose Integration zwischen TanStack DB und ElectricSQL und ermöglichen die Echtzeit-Datensynchronisation mit Ihrer Postgres-Datenbank über den Sync-Engine von Electric.

Übersicht

Das Paket @tanstack/electric-db-collection ermöglicht Ihnen die Erstellung von Collections, die

Daten automatisch von Postgres über Electric Shapes synchronisieren
Optimistic Updates mit Transaktionsabgleich und automatischem Rollback bei Fehlern unterstützen
Persistence durch anpassbare Mutations-Handler verwalten

Installation

bash

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

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

Grundlegende Verwendung

typescript

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,
  })
)

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,
  })
)

Konfigurationsoptionen

Die Funktion electricCollectionOptions akzeptiert die folgenden Optionen

Erforderliche Optionen

shapeOptions: Konfiguration für den ElectricSQL ShapeStream
- url: Die URL Ihres Proxys zu Electric
getKey: Funktion zur Extraktion des eindeutigen Schlüssels aus einem Element

Optional

id: Eindeutiger Bezeichner für die Collection
schema: Schema zur Validierung von Elementen. Jedes mit Standard-Schema kompatible Schema
sync: Benutzerdefinierte Sync-Konfiguration

Persistence Handlers

onInsert: Handler, der vor Insert-Operationen aufgerufen wird
onUpdate: Handler, der vor Update-Operationen aufgerufen wird
onDelete: Handler, der vor Delete-Operationen aufgerufen wird

Persistence Handlers

Handler können definiert werden, um bei Mutationen ausgeführt zu werden. Sie sind nützlich, um Mutationen an das Backend zu senden und sie zu bestätigen, sobald Electric die entsprechenden Transaktionen liefert. Bis zur Bestätigung blockiert TanStack DB Sync-Daten für die Collection, um Race Conditions zu vermeiden. Um Verzögerungen zu vermeiden, ist es wichtig, eine passende Strategie zu verwenden.

Die zuverlässigste Strategie ist, dass das Backend die Transaktions-ID (txid) in seiner Antwort mitliefert, damit der Client jede Mutation mit den Transaktions-Identifikatoren von Electric für eine präzise Bestätigung abgleichen kann. Wenn keine Strategie angegeben wird, werden Client-Mutationen automatisch nach drei Sekunden bestätigt.

typescript

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: response.txid }
    },
    
    // you can also implement onUpdate and onDelete handlers
  })
)

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: response.txid }
    },
    
    // you can also implement onUpdate and onDelete handlers
  })
)

Auf dem Backend können Sie die txid für eine Transaktion extrahieren, indem Sie direkt Postgres abfragen.

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)
}

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 Proxy Beispiel

Electric wird typischerweise hinter einem Proxy-Server bereitgestellt, der die Shape-Konfiguration, Authentifizierung und Autorisierung handhabt. Dies bietet eine bessere Sicherheit und ermöglicht es Ihnen zu kontrollieren, auf welche Daten Benutzer zugreifen können, ohne Electric dem Client auszusetzen.

Hier ist ein Beispiel für eine Proxy-Implementierung mit TanStack Starter

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,
})

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,
})

Optimistic Updates mit expliziten Transaktionen

Für fortgeschrittenere Anwendungsfälle können Sie benutzerdefinierte Aktionen erstellen, die mehrere Mutationen über Collections hinweg transaktional ausführen können. In diesem Fall müssen Sie explizit auf die Transaktions-ID warten, indem Sie utils.awaitTxId() verwenden.

typescript

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 }
    })
    
    await todosCollection.utils.awaitTxId(response.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 }
    })
    
    await todosCollection.utils.awaitTxId(response.txid)
  }
})