@codehz/sync

@codehz/sync 是一个 Bun-only、TypeScript-first 的 WebSocket 同步库。

当前版本围绕 WebSocket 会话模型和一个对话预设组织：

• invoke：统一读写订阅协议（query / command / action / snapshot / stream）
• action：单次启动、可发事件、可取消、最终只返回一个终态结果的写操作
• conversation：AI 多轮对话预设（start / resume / submitInput / abort + output 流）

安装

bun add @codehz/sync

核心模型

服务端定义 sync 树，客户端通过 WebSocket 访问 scope 实例。

import {
  createBunSyncServer,
  createClient,
  createSync,
  defineAction,
  defineCommand,
  defineConversationScope,
  defineScope,
  invalidateTag,
  query,
  scope,
  stream,
} from "@codehz/sync";

服务端示例

const jobs = new Map<string, JobState>();

const sync = createSync({
  root: defineScope({
    snapshot: () => ({ online: true }),
    queries: {
      online: query<boolean>({ read: async () => true }),
    },
    commands: {
      identify: async () => ({ scope: "root" as const }),
    },
  }),
  scopes: {
    job: scope<{ id: string }>()(
      defineConversationScope({
        snapshot: ({ id }) => jobs.get(id)!,
        commands: {
          start: defineCommand(async ({ id }, input: { prompt: string }) => {
            const job = jobs.get(id)!;
            job.status = "running";
            await syncServer.publishScope({ name: "job", params: { id } });
            return { accepted: true };
          }),
          submitInput: defineCommand(
            async ({ id }, input: { approved: boolean }) => {
              const job = jobs.get(id)!;
              job.status = "completed";
              job.inputRequest = undefined;
              job.result = {
                output: input.approved ? "approved" : "rejected",
              };
              await syncServer.publishScope({ name: "job", params: { id } });
              return job.result;
            }
          ),
        },
        streams: {
          lifecycle: stream<{ type: string }, number, { id: string }>({
            replay: ({ id }, _ctx, from) => ({
              items: (jobEvents.get(id) ?? []).filter((item) =>
                from === null ? true : item.cursor > from
              ),
              nextCursor: null,
            }),
          }),
          output: stream<{ delta: string }, number, { id: string }>(),
        },
      })
    ),
  },
});

const syncServer = createBunSyncServer(sync);

Bun.serve({
  port: 3000,
  routes: {
    "/sync/v1": syncServer.fetch,
    "/*": new Response("Not Found", { status: 404 }),
  },
  websocket: syncServer.websocket,
});

Dependency Tags

读侧 resolver 现在接收 ctx 对象：

• ctx.context：应用自己的 request/context 数据
• ctx.watchTag([...])：声明本次读取依赖的语义 tag

const sync = createSync({
  root: defineScope({
    snapshot: (_params, ctx) => {
      ctx.watchTag(["catalog"]);
      return { count: items.size };
    },
    queries: {
      detail: query<{ id: string; label: string }, { id: string }>({
        read: (_params, ctx, input) => {
          ctx.watchTag(["entity", input.id]);
          return items.get(input.id)!;
        },
      }),
    },
    commands: {
      rename: defineCommand(
        async (_params, input: { id: string; label: string }) => {
          items.get(input.id)!.label = input.label;
          return { id: input.id };
        },
        {
          invalidate: [invalidateTag((_params, input) => ["entity", input.id])],
        }
      ),
    },
  }),
  scopes: {},
});

await syncServer.publishTag(["entity", "item_1"]);

Action 模型

defineAction(...) 适合“由客户端启动，但执行过程中需要回传进度/领域事件”的写操作。

可恢复 action 通过 recoverable: true 开启。客户端用 clientActionKey 避免重复启动，并可在重连后继续恢复同一个 action 实例。

const sync = createSync({
  root: defineScope({
    snapshot: () => ({ ready: true }),
    actions: {
      generate: defineAction({
        recoverable: true,
        run: async (_params, input: { prompt: string }, ctx) => {
          ctx.emit({ type: "started" });
          ctx.emit({ type: "progress", value: input.prompt });
          return {
            output: input.prompt.toUpperCase(),
          };
        },
        invalidateTags: [(_params, _input, result) => ["job", result.output]],
      }),
    },
  }),
  scopes: {},
});

const run = client
  .root()
  .actions.generate.start(
    { prompt: "hello" },
    { clientActionKey: "draft:42:submit:3" }
  );

for await (const event of run) {
  console.log(event.event);
}

console.log(run.getActionInstanceId());
console.log(await run.result);

createBunSyncServer(sync) 默认启用进程内 recoverable action store。需要自定义保留策略或持久化时，可传入：

const syncServer = createBunSyncServer(sync, {
  actionStore: createInMemoryActionStore({
    completedRetentionMs: 300_000,
  }),
});

客户端示例

const client = createClient<typeof sync>({
  url: "ws://127.0.0.1:3000/api/rpc",
  definition: sync,
});

const conv = client.conversation("job", { id: "job_123" });

await conv.start({ prompt: "hello" });

const statusSource = conv.queries.status.source();
await statusSource.ensure();

await conv.submitInput({ approved: true });
console.log(await conv.queries.result.source().ensure());

for await (const item of conv.streams.lifecycle.subscribe()) {
  console.log(item.cursor, item.event);
}

client.close();

Conversation 模型

defineConversationScope(...) 为 AI 对话提供标准约束：

• 生命周期：pending / running / awaiting_input / completed / failed / cancelled
• 控制：start / abort / resume / submitInput
• 标准查询：status / result / error / inputRequest / checkpoint / capabilities
• 事件流：streams.lifecycle（生命周期）+ streams.output（turn 内 token/delta）

capabilities 由服务端根据当前 snapshot 计算，React 与原生 client 行为一致。 标准布尔字段保持稳定，领域扩展数据放在 capabilities.domain。

defineConversationScope({
  snapshot: ({ id }) => jobs.get(id)!,
  conversation: {
    resolveCapabilities: (snapshot, { defaults, params }) => ({
      ...defaults,
      canAbort: false,
      domain: {
        scopeId: params.id,
        hasCheckpoint: snapshot.checkpoint !== undefined,
      },
    }),
  },
  commands: {
    start: defineCommand(async () => ({ accepted: true })),
  },
  streams: {
    lifecycle: stream<{ type: string }, number, { id: string }>({}),
  },
});

React 集成

import { createQueryClient } from "@codehz/sync";
import {
  useAction,
  useConversation,
  useManagedQuery,
  useMutation,
} from "@codehz/sync/react";

const controller = useConversation(conv, {
  channels: {
    lifecycle: {
      initialReduced: 0,
      reduce: (count: number) => count + 1,
    },
  },
  resume: {
    key: "job:job_123",
    storage: localStorage,
  },
});

const queryClient = createQueryClient();
const statusQuery = queryClient.query(client.root().queries.online.source());
const status = useManagedQuery(statusQuery, { client: queryClient });

const rename = useMutation(client.root().commands.identify, {
  onMutate(tx) {
    const previous = tx.get(statusQuery);
    tx.set(statusQuery, true);
    return { previous };
  },
  onError(_error, ctx) {
    console.log("rollback context", ctx.previous);
  },
});

const generate = useAction(client.root().actions.generate, {
  onMutate(tx, input) {
    tx.set(statusQuery, false);
    return { input };
  },
  onEvent(event) {
    console.log(event.event);
  },
});

await rename.execute();
await generate.startAsync({ prompt: "hello" });

参考文档

• 设计说明
• 建模模式