feat(subscription): realtime mutation publishing + MCP resource notifications

[Azzerty23]

Add a single-source-of-truth realtime path so data changes reach both the app (oRPC SSE) and MCP clients, without a second event bus.

• MutationPublisher port + ModelMutationEvent, structurally identical to @viiite/server's Publisher so a Durable-Object-backed instance can be passed directly. Ships createInMemoryPublisher() default + defaultChannel().
• execute tool publishes a ModelMutationEvent after every successful write (channel = model.toLowerCase()); await-with-catch so a publish failure never fails the write and post-response async isn't cut on serverless.
• registerModelResources + bridgeModelMutations: expose models as subscribable MCP resources and push notifications/resources/updated while subscribed.
• server-adapters/workers: McpAgent (Durable Object) + OAuthProvider adapter wiring buildMcpServer, resources, and the publisher bridge; resolves the user from OAuth props. agents/@cloudflare/workers-oauth-provider are optional peers.
• docs/realtime-notifications.md covers piece 1, piece 2, and the Workers wiring.

Tests: in-memory publisher, execute-tool publishing, and the resource bridge.

Summary by CodeRabbit

Release Notes

• New Features

  • Added Cloudflare Workers server adapter for hosting ZenStack MCP servers
  • Introduced real-time mutation notifications system enabling database writes to trigger notifications for connected MCP/LLM clients
  • Added pub/sub mutation publishing capabilities for tracking and broadcasting database changes

• Documentation

  • New guide on configuring real-time notifications and integrating mutation publishers with MCP servers

• Tests

  • Added comprehensive test coverage for mutation publishing, real-time notifications, and in-memory publisher functionality

[coderabbitai]

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Free

Run ID: 1c9053c7-22f6-4f36-ab7a-f56a803187ef

📥 Commits

Reviewing files that changed from the base of the PR and between c16a282 and 69ed576.

⛔ Files ignored due to path filters (1)

• bun.lock is excluded by !**/*.lock

📒 Files selected for processing (13)

• packages/zenstack-mcp/docs/realtime-notifications.md
• packages/zenstack-mcp/index.ts
• packages/zenstack-mcp/package.json
• packages/zenstack-mcp/src/__tests__/execute-tool.test.ts
• packages/zenstack-mcp/src/__tests__/notifications.test.ts
• packages/zenstack-mcp/src/__tests__/publisher.test.ts
• packages/zenstack-mcp/src/events/notifications.ts
• packages/zenstack-mcp/src/events/publisher.ts
• packages/zenstack-mcp/src/server-adapters/workers.ts
• packages/zenstack-mcp/src/server.ts
• packages/zenstack-mcp/src/tools/execute-tool.ts
• packages/zenstack-mcp/src/types.ts
• packages/zenstack-mcp/tsdown.config.ts

---

Walkthrough

Adds a mutation pub/sub system to zenstack-mcp: new ModelMutationEvent/MutationPublisher types, an in-memory publisher, optional post-write publishing in the execute tool, an MCP resource notification bridge (registerModelResources, bridgeModelMutations), and a new Cloudflare Workers/Durable Object server adapter (createZenStackMcpAgent, createWorkersMcpHandler) with matching package exports, build config, and documentation.

Changes

Mutation pub/sub and MCP resource notifications

Layer / File(s)	Summary
Pub/sub type contracts and McpServerConfig extension packages/zenstack-mcp/src/types.ts	Introduces ModelMutationEvent (operation, modelName, data, ids, timestamp) and MutationPublisher (channel-based publish/subscribe async iterable). Extends McpServerConfig with optional publisher and channelFormatter.
In-memory publisher and unit tests packages/zenstack-mcp/src/events/publisher.ts, packages/zenstack-mcp/src/__tests__/publisher.test.ts	Implements defaultChannel (lowercases model name) and createInMemoryPublisher with buffered async-iterable subscriptions, best-effort delivery, and abort-signal teardown. Tests cover delivery, channel isolation, queueing, abort handling, and subscriber failure isolation.
Execute tool mutation publishing, server wiring, and tests packages/zenstack-mcp/src/tools/execute-tool.ts, packages/zenstack-mcp/src/server.ts, packages/zenstack-mcp/src/__tests__/execute-tool.test.ts	Extends registerExecuteTool to build and publish a ModelMutationEvent after each successful write, with ID extraction from args/results. buildMcpServer threads config.publisher and config.channelFormatter through. Tests assert correct payloads for create/delete, no-publish for reads, custom channel formatters, and publish-failure resilience.
MCP resource notification bridge and tests packages/zenstack-mcp/src/events/notifications.ts, packages/zenstack-mcp/src/__tests__/notifications.test.ts	Adds modelResourceUri, registerModelResources (auth-enforced findMany as readable MCP resources), and bridgeModelMutations (installs subscribe/unsubscribe handlers, runs per-URI async publisher loops emitting notifications/resources/updated, tears down on server close). Tests cover all subscribe, unsubscribe, unknown-URI, and teardown paths.
Cloudflare Workers/Durable Object adapter and package wiring packages/zenstack-mcp/src/server-adapters/workers.ts, packages/zenstack-mcp/package.json, packages/zenstack-mcp/tsdown.config.ts	Adds createZenStackMcpAgent (a McpAgent subclass that wires model extraction, MCP server construction, and optional notification bridging on init) and createWorkersMcpHandler (wraps in OAuthProvider, returns { handler, Agent }). Registers the ./server-adapters/workers export and optional peer deps for agents and @cloudflare/workers-oauth-provider.
Public exports and realtime-notifications docs packages/zenstack-mcp/index.ts, packages/zenstack-mcp/docs/realtime-notifications.md	Expands index.ts to re-export all new pub/sub and notification APIs. Adds documentation covering the single-publisher pattern, Cloudflare Workers setup with Durable Objects, runtime-agnostic advanced usage, operational flow, and ModelMutationEvent payload reference.

Sequence Diagram(s)

sequenceDiagram
  participant LLMClient as LLM Client
  rect rgba(100, 149, 237, 0.5)
    Note over LLMClient,MutationPublisher: Setup
    LLMClient->>McpServer: resources/subscribe (zenstack://Post)
    McpServer->>bridgeModelMutations: subscribe handler
    bridgeModelMutations->>MutationPublisher: subscribe("post", {signal})
  end
  rect rgba(144, 238, 144, 0.5)
    Note over LLMClient,MutationPublisher: Write path
    LLMClient->>McpServer: tool call: execute(create Post)
    McpServer->>ZenStackClient: db.post.create(...)
    ZenStackClient-->>McpServer: created record
    McpServer->>MutationPublisher: publish("post", ModelMutationEvent{create})
    MutationPublisher->>bridgeModelMutations: event delivered
    bridgeModelMutations->>McpServer: sendResourceUpdated(zenstack://Post)
    McpServer->>LLMClient: notifications/resources/updated
  end
  rect rgba(255, 165, 0, 0.5)
    Note over LLMClient,MutationPublisher: Teardown
    LLMClient->>McpServer: resources/unsubscribe
    McpServer->>bridgeModelMutations: unsubscribe handler → abort loop
  end

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Poem

🐇 Hoppity-hop, a publisher hops!
Each write pings the channel and never stops.
LLM clients now watch models update,
SSE and MCP both celebrate.
Workers stay stateful in Durable land —
Realtime notifications, perfectly planned! 🎉

---

Note

🎁 Summarized by CodeRabbit Free

Your organization is on the Free plan. CodeRabbit will generate a high-level summary and a walkthrough for each pull request. For a comprehensive line-by-line review, please upgrade your subscription to CodeRabbit Pro by visiting https://app.coderabbit.ai/login.

_{Comment @coderabbitai help to get the list of available commands and usage tips.}