@可持久流/状态
传输结构状态超过可持久流的构件。 使用这些原始的任意实时协议: AI标志流,存在更新,合作编辑,或数据库同步.
安装
pnpm add @durable-streams/state
该套件暴露出两个切入点:
@durable-streams/state- 无 db 协议表面:createStateSchema以援助者发誓,MaterializedState,以及事件类型/守护符。 没有额外的依赖性.@durable-streams/state/db反应式的 TanStack DB 背层(createStreamDB,现场询问,乐观行动). 本条目要求@tanstack/db对等依赖 :
pnpm add @durable-streams/state @tanstack/db
说明:
@tanstack/db属于同位素依赖/db仅输入。 在使用 StreamDB 收藏时安装它确保类型与 TanStack DB 查询工具的兼容性useLiveQuery从@tanstack/react-db。 。 。 .
概览
这个软件包为流线结构状态提供了灵活的原始. 您选择您需要多少结构 :
- 简单状态更新:流出 JSON 有效载荷并跟踪当前值
- 类型收藏: 为结构化实体添加计划和主密钥
- 反应查询: 利用 TanStack DB 进行订阅和乐观更新
随时随地都行.
快速启动
简单状态
流结构 JSON 并查询当前值 :
import { MaterializedState } from "@durable-streams/state"
const state = new MaterializedState()
// Apply any structured change
state.apply({
type: "token",
key: "stream-1",
value: { content: "Hello", model: "claude-3" },
headers: { operation: "insert" },
})
// Query current state
const token = state.get("token", "stream-1")
const allTokens = state.getType("token")
类型收藏
为结构化实体添加计划和验证 :
import { createStateSchema } from "@durable-streams/state"
import { createStreamDB } from "@durable-streams/state/db"
// Define your schema
const schema = createStateSchema({
users: {
schema: userSchema, // Standard Schema validator
type: "user", // Event type field
primaryKey: "id", // Primary key field name
},
messages: {
schema: messageSchema,
type: "message",
primaryKey: "id",
},
})
// Create a stream-backed database
const db = createStreamDB({
streamOptions: {
url: "https://api.example.com/streams/my-stream",
contentType: "application/json",
},
live: "sse", // optional: true, "long-poll", "sse", or false
state: schema,
})
// Load initial data
await db.preload()
// Reactive queries with useLiveQuery
import { useLiveQuery } from "@tanstack/react-db" // or solid-db, vue-db
import { eq } from "@tanstack/db"
const userQuery = useLiveQuery((q) =>
q
.from({ users: db.collections.users })
.where(({ users }) => eq(users.id, "123"))
.findOne()
)
const allUsersQuery = useLiveQuery((q) =>
q.from({ users: db.collections.users })
)
核心概念
国家议定书
《持久流州议定书》界定了州变化事件的标准格式:
- 更改事件数字 :
insert, (中文(简体) ),update, (中文(简体) ),delete对实体的业务 - 控制事件数字 :
snapshot-start, (中文(简体) ),snapshot-end, (中文(简体) ),reset信号 - 实体类型: 引导事件到收藏的歧义字段
- 主密钥: 从实体值中提取的独特标识符
见 国家-波罗多克州 完整规格.
物质化状态
用于基本用途的简单模拟状态容器:
import { MaterializedState } from "@durable-streams/state"
const state = new MaterializedState()
// Apply change events
state.apply({
type: "user",
key: "1",
value: { name: "Kyle" },
headers: { operation: "insert" },
})
// Query state
const user = state.get("user", "1")
const allUsers = state.getType("user")
流出数据库
有 TanStack DB 收藏的流回数据库. 提供被动的询问,订阅,以及乐观的更新.
Schema 定义
创建状态Schema ()
定义您的应用程序状态结构 :
const schema = createStateSchema({
users: {
schema: userSchema, // Standard Schema validator
type: "user", // Event type for routing
primaryKey: "id", // Field to use as primary key
},
messages: {
schema: messageSchema,
type: "message",
primaryKey: "id",
},
})
标准Schema 支持
用途 标准Schema 用于验证,支持多个库:
// Zod
import { z } from "zod"
const userSchema = z.object({
id: z.string(),
name: z.string(),
email: z.string().email(),
})
// Valibot
import * as v from "valibot"
const userSchema = v.object({
id: v.string(),
name: v.string(),
email: v.pipe(v.string(), v.email()),
})
// Manual Standard Schema
const userSchema = {
"~standard": {
version: 1,
vendor: "my-app",
validate: (value) => {
// Your validation logic
if (isValid(value)) {
return { value }
}
return { issues: [{ message: "Invalid user" }] }
},
},
}
事件帮助者
Schema 提供类型事件创建帮助 :
// Insert
const insertEvent = schema.users.insert({
value: { id: "1", name: "Kyle", email: "[email protected]" },
key: "1", // Optional, defaults to value[primaryKey]
})
// Update
const updateEvent = schema.users.update({
value: { id: "1", name: "Kyle Mathews", email: "[email protected]" },
oldValue: { id: "1", name: "Kyle", email: "[email protected]" }, // Optional
})
// Delete
const deleteEvent = schema.users.delete({
key: "1",
oldValue: { id: "1", name: "Kyle", email: "[email protected]" }, // Optional
})
// Custom headers
const eventWithTxId = schema.users.insert({
value: { id: "1", name: "Kyle" },
headers: {
txid: crypto.randomUUID(),
timestamp: new Date().toISOString(),
},
})
流出数据库
创建数据库
const db = createStreamDB({
streamOptions: {
url: "https://api.example.com/streams/my-stream",
contentType: "application/json",
// All DurableStream options supported
headers: { Authorization: "Bearer token" },
batching: true,
},
state: schema,
})
// The stream is created lazily when preload() is called
await db.preload()
用 TanStack DB 进行反应查询
StreamDB收藏为TanStack DB收藏. 使用 TanStack DB 的查询构建器进行过滤、排序、聚合和加入 差异数据流量 - 大大快于JavaScript过滤:
import { useLiveQuery } from "@tanstack/[framework]-db" // react-db, solid-db, etc
import { eq, gt, and, count } from "@tanstack/db"
// Simple collection access
const query = useLiveQuery((q) => q.from({ users: db.collections.users }))
// Filtering with WHERE
const activeQuery = useLiveQuery((q) =>
q
.from({ users: db.collections.users })
.where(({ users }) => eq(users.active, true))
)
// Complex conditions
const query = useLiveQuery((q) =>
q
.from({ users: db.collections.users })
.where(({ users }) => and(eq(users.active, true), gt(users.age, 18)))
)
// Sorting and limiting
const topUsersQuery = useLiveQuery((q) =>
q
.from({ users: db.collections.users })
.orderBy(({ users }) => users.lastSeen, "desc")
.limit(10)
)
// Aggregation with GROUP BY and ordering
const langStatsQuery = useLiveQuery((q) => {
const languageCounts = q
.from({ events: db.collections.events })
.groupBy(({ events }) => events.language)
.select(({ events }) => ({
language: events.language,
total: count(events.id),
}))
return q
.from({ stats: languageCounts })
.orderBy(({ stats }) => stats.total, "desc")
})
// Joins across collections
const query = useLiveQuery((q) =>
q
.from({ messages: db.collections.messages })
.join({ users: db.collections.users }, ({ messages, users }) =>
eq(messages.userId, users.id)
)
.select(({ messages, users }) => ({
messageId: messages.id,
text: messages.text,
userName: users.name,
}))
)
为什么使用查询构建器?
- 差异数据流: 递增更新只重算受影响的结果
- 戏剧性快: 将过滤/排序推入 DB 引擎对 JavaScript
- 反应: 当数据变化时查询自动更新
- 类型安全: 自动完成的完整类型脚本支持
框架整合: 见 TanStack DB 文档 特定框架指南:
生命周期方法
// Load all data until up-to-date
await db.preload()
// Stop syncing and cleanup
db.close()
// Wait for a transaction to be confirmed
await db.utils.awaitTxId("txid-uuid", 5000) // 5 second timeout
优化动作
定义带有乐观更新和服务器确认的动作 :
const db = createStreamDB({
streamOptions: { url: streamUrl, contentType: "application/json" },
state: schema,
actions: ({ db, stream }) => ({
addUser: {
// Optimistic update (runs immediately)
onMutate: (user) => {
db.collections.users.insert(user)
},
// Server mutation (runs async)
mutationFn: async (user) => {
const txid = crypto.randomUUID()
await stream.append(
JSON.stringify(
schema.users.insert({
value: user,
headers: { txid },
})
)
)
// Wait for confirmation
await db.utils.awaitTxId(txid)
},
},
updateUser: {
onMutate: ({ id, updates }) => {
db.collections.users.update(id, (draft) => {
Object.assign(draft, updates)
})
},
mutationFn: async ({ id, updates }) => {
const txid = crypto.randomUUID()
const current = await db.collections.users.get(id)
await stream.append(
JSON.stringify(
schema.users.update({
value: { ...current, ...updates },
oldValue: current,
headers: { txid },
})
)
)
await db.utils.awaitTxId(txid)
},
},
}),
})
// Call actions
await db.actions.addUser({ id: "1", name: "Kyle", email: "[email protected]" })
await db.actions.updateUser({ id: "1", updates: { name: "Kyle Mathews" } })
框架整合
使用 TanStack DB 的框架适配器进行被动查询 :
反应
import { useLiveQuery } from '@tanstack/react-db'
import { eq } from '@tanstack/db'
function UserProfile({ userId }: { userId: string }) {
const userQuery = useLiveQuery((q) =>
q.from({ users: db.collections.users })
.where(({ users }) => eq(users.id, userId))
.findOne()
)
if (userQuery.isLoading()) return <div>Loading...</div>
if (!userQuery.data) return <div>Not found</div>
return (
<div>
<h1>{userQuery.data.name}</h1>
<p>{userQuery.data.email}</p>
</div>
)
}
见 @tastack/ react-db 文档 为了更多.
实相.js
import { useLiveQuery } from '@tanstack/solid-db'
import { eq } from '@tanstack/db'
function MessageList() {
const messagesQuery = useLiveQuery((q) =>
q.from({ messages: db.collections.messages })
.orderBy(({ messages }) => messages.timestamp, 'desc')
.limit(50)
)
return (
<For each={messagesQuery.data}>
{(message) => <MessageCard message={message} />}
</For>
)
}
见 @ tastack/ solid-db 文档 为了更多.
常见模式
密钥/ 价值存储
const schema = createStateSchema({
config: {
schema: configSchema,
type: "config",
primaryKey: "key",
},
})
// Set value
await stream.append(
JSON.stringify(
schema.config.insert({
value: { key: "theme", value: "dark" },
})
)
)
// Query value reactively
const themeQuery = useLiveQuery((q) =>
q
.from({ config: db.collections.config })
.where(({ config }) => eq(config.key, "theme"))
.findOne()
)
追踪
const schema = createStateSchema({
presence: {
schema: presenceSchema,
type: "presence",
primaryKey: "userId",
},
})
// Update presence
await stream.append(
JSON.stringify(
schema.presence.update({
value: {
userId: "kyle",
status: "online",
lastSeen: Date.now(),
},
})
)
)
// Query presence with TanStack DB
const presenceQuery = useLiveQuery((q) =>
q
.from({ presence: db.collections.presence })
.where(({ presence }) => eq(presence.status, "online"))
)
多类型聊天室
const schema = createStateSchema({
users: { schema: userSchema, type: "user", primaryKey: "id" },
messages: { schema: messageSchema, type: "message", primaryKey: "id" },
reactions: { schema: reactionSchema, type: "reaction", primaryKey: "id" },
typing: { schema: typingSchema, type: "typing", primaryKey: "userId" },
})
// Different types coexist in the same stream
await stream.append(JSON.stringify(schema.users.insert({ value: user })))
await stream.append(JSON.stringify(schema.messages.insert({ value: message })))
await stream.append(
JSON.stringify(schema.reactions.insert({ value: reaction }))
)
最佳做法
1. 使用对象值
StreamDB 对主密钥模式需要对象值(不是原始值):
// ❌ Won't work
{ type: 'count', key: 'views', value: 42 }
// ✅ Works
{ type: 'count', key: 'views', value: { count: 42 } }
2. 总是叫停()
useEffect(() => {
const db = createStreamDB({ streamOptions, state: schema })
return () => db.close() // Cleanup on unmount
}, [])
3. 边界验证
使用标准Schema验证系统边界的数据 :
const userSchema = z.object({
id: z.string().uuid(),
email: z.string().email(),
age: z.number().min(0).max(150),
})
4. 使用交易ID
对于关键操作,始终使用交易ID以确保确认:
const txid = crypto.randomUUID()
await stream.append(
JSON.stringify(schema.users.insert({ value: user, headers: { txid } }))
)
await db.utils.awaitTxId(txid, 10000) // Wait up to 10 seconds
5. 处理错误
try {
await db.actions.addUser(user)
} catch (error) {
if (error.message.includes("Timeout")) {
// Handle timeout
} else {
// Handle other errors
}
}
API 参考
类型
export type Operation = "insert" | "update" | "delete"
export interface ChangeEvent<T = unknown> {
type: string
key: string
value?: T
old_value?: T
headers: ChangeHeaders
}
export interface ChangeHeaders {
operation: Operation
txid?: string
timestamp?: string
}
export interface ControlEvent {
headers: {
control: "snapshot-start" | "snapshot-end" | "reset"
offset?: string
}
}
export type StateEvent<T = unknown> = ChangeEvent<T> | ControlEvent
职能
// Create a state schema with typed collections and event helpers
export function createStateSchema<
T extends Record<string, CollectionDefinition>,
>(collections: T): StateSchema<T>
// Create a stream-backed database
export async function createStreamDB<
TDef extends StreamStateDefinition,
TActions extends Record<string, ActionDefinition<any>>,
>(
options: CreateStreamDBOptions<TDef, TActions>
): Promise<StreamDB<TDef> | StreamDBWithActions<TDef, TActions>>
类
export class MaterializedState {
apply(event: ChangeEvent): void
applyBatch(events: ChangeEvent[]): void
get<T>(type: string, key: string): T | undefined
getType(type: string): Map<string, unknown>
clear(): void
readonly typeCount: number
readonly types: string[]
}
许可证
阿帕奇 - 2.0
学习更多
- 国家-波罗多克州 - 协议的完整规格
- 持久流议定书 - 基础协议
- 标准Schema - Schema验证
- TanStack 数据库 - 反应收集