feat: implement real-time event push with SSE and delegation management

Author: benjinus

docs/reply/20260309-p2p-event-subscription-proxy.md

@@ -5,16 +5,22 @@
 > **日期**: 2026-03-09
 > **关联文档**: `docs/issues/clawnet-p2p-event-subscription-proxy.md`
 > **实施文档**: `docs/implementation/subscription-delegation.md`
-> **涉及包**: `@claw-network/protocol`, `@claw-network/node`, `@claw-network/sdk`
+> **涉及包**: `@claw-network/protocol`, `@claw-network/node`, `@claw-network/sdk`（均已发布 **v0.6.1**）
 
 ---
 
-## 状态：方案 B 已实现 ✅
+## 状态：方案 B 已实现并发布 ✅
 
 采纳需求文档推荐的 **方案 B（ClawNet 协议层事件订阅代理）**，已完成全链路实现：协议类型 → SQLite 存储 → 服务层（自动转发 + 反压控制） → REST API → WebSocket 代理订阅端点 → SDK 方法。
 
 编译通过，lint 零错误，全部 304 个测试通过（含新增 20 个 delegation 测试）。
 
+**npm 已发布**：`@claw-network/core@0.6.1`、`@claw-network/protocol@0.6.1`、`@claw-network/sdk@0.6.1`、`@claw-network/node@0.6.1`。
+
+```bash
+npm install @claw-network/sdk@0.6.1
+```
+
 ---
 
 ## 实现概览
@@ -387,6 +393,11 @@ ws.on('close', () => {
 | `packages/node/src/api/ws-messaging.ts` | 新增 WS 端点 | `/subscribe-delegated` 含回放 + 心跳 |
 | `packages/sdk/src/messaging.ts` | 新增 SDK 方法 | 3 个类型安全的 REST 客户端方法 |
 | `packages/node/test/delegation.test.ts` | 新增测试 | 20 个测试用例覆盖存储、API、去重、回放 |
+| `CHANGELOG.md` | 更新 | v0.6.1 变更日志 |
+| `docs/api/openapi.yaml` | 更新 | 4 个 REST 端点 + `DelegationRecord` schema |
+| `docs/API_REFERENCE.md` | 更新 | Messaging 章节 + `DELEGATION_LIMIT` 错误码 |
+| `docs/API_ROUTE_CATALOG.md` | 更新 | 路由总表 #86–#89 |
+| `docs/SDK_GUIDE.md` | 更新 | Messaging API 章节（delegation CRUD + WS 说明） |
 
 ---
 

packages/node/src/api/routes/events.ts

@@ -0,0 +1,77 @@
+import type { RuntimeContext } from '../types.js';
+import { Router } from '../router.js';
+import { ok, created } from '../response.js';
+import { handleError } from '../route-utils.js';
+
+/**
+ * Event routes:
+ *   GET  /                — Local SSE event stream
+ *   POST /subscribe       — Target: create delegation (called by gateway via API Proxy)
+ *   POST /unsubscribe     — Target: revoke delegation
+ */
+export function eventRoutes(ctx: RuntimeContext): Router {
+  const router = new Router();
+
+  // Local SSE endpoint — Webapp directly connected to this node
+  router.get('/', ({ res }) => {
+    if (!ctx.eventPushService) {
+      res.writeHead(503, { 'Content-Type': 'application/json' });
+      res.end(JSON.stringify({ error: 'Event push service unavailable' }));
+      return;
+    }
+    // SSE connection: addLocalClient handles headers, heartbeat, and cleanup.
+    // Do NOT call res.end() — SSE is a long-lived connection.
+    ctx.eventPushService.addLocalClient(res);
+  });
+
+  // Target role: gateway asks us to create a delegation
+  router.post('/subscribe', async ({ res, body }) => {
+    if (!ctx.eventPushService) {
+      res.writeHead(503, { 'Content-Type': 'application/json' });
+      res.end(JSON.stringify({ error: 'Event push service unavailable' }));
+      return;
+    }
+
+    const payload = body as Record<string, unknown> | undefined;
+    const gatewayDid = payload?.gatewayDid as string;
+    if (!gatewayDid || typeof gatewayDid !== 'string') {
+      res.writeHead(400, { 'Content-Type': 'application/json' });
+      res.end(JSON.stringify({ error: 'Missing gatewayDid' }));
+      return;
+    }
+
+    try {
+      const result = await ctx.eventPushService.createDelegation(gatewayDid);
+      created(res, result, { self: '/api/v1/events/subscribe' });
+    } catch (err) {
+      handleError(res, err);
+    }
+  });
+
+  // Target role: gateway asks us to revoke a delegation
+  router.post('/unsubscribe', async ({ res, body }) => {
+    if (!ctx.eventPushService) {
+      res.writeHead(503, { 'Content-Type': 'application/json' });
+      res.end(JSON.stringify({ error: 'Event push service unavailable' }));
+      return;
+    }
+
+    const payload = body as Record<string, unknown> | undefined;
+    const delegationId = payload?.delegationId as string;
+    if (!delegationId || typeof delegationId !== 'string') {
+      res.writeHead(400, { 'Content-Type': 'application/json' });
+      res.end(JSON.stringify({ error: 'Missing delegationId' }));
+      return;
+    }
+
+    try {
+      await ctx.eventPushService.revokeDelegation(delegationId);
+      res.writeHead(204);
+      res.end();
+    } catch (err) {
+      handleError(res, err);
+    }
+  });
+
+  return router;
+}

packages/node/src/api/routes/relay.ts

@@ -138,6 +138,28 @@ export async function handleRelayRequest(
     return true;
   }
 
+  // ── SSE relay: GET /:targetDid/api/v1/events ──────────
+  // This is a long-lived SSE connection — NOT a normal API proxy call.
+  if (remaining === '/api/v1/events' || remaining === '/api/v1/events/') {
+    if (req.method !== 'GET') {
+      writeJson(res, 405, { error: 'Method not allowed' });
+      return true;
+    }
+    if (!ctx.eventPushService) {
+      writeJson(res, 503, { error: 'Event push service unavailable' });
+      return true;
+    }
+    try {
+      await ctx.eventPushService.addGatewayClient(res, decodedDid);
+      // Do NOT call res.end() — SSE is a long-lived connection
+    } catch (err) {
+      if (!res.headersSent) {
+        writeJson(res, 502, { error: `Event subscription failed: ${(err as Error).message}` });
+      }
+    }
+    return true;
+  }
+
   // Read request body
   let body: string | undefined;
   if (req.method !== 'GET' && req.method !== 'HEAD' && req.method !== 'DELETE') {

packages/node/src/api/server.ts

@@ -20,6 +20,7 @@ import { profileRoutes } from './routes/profile.js';
 import { sessionRoutes } from './routes/session.js';
 import { walletRoutes } from './routes/wallets.js';
 import { handleRelayRequest } from './routes/relay.js';
+import { eventRoutes } from './routes/events.js';
 import type { RuntimeContext } from './types.js';
 
 function buildRouter(ctx: RuntimeContext): Router {
@@ -38,6 +39,7 @@ function buildRouter(ctx: RuntimeContext): Router {
   router.mount('/api/v1/session', sessionRoutes(ctx));
   router.mount('/api/v1/clawnet', clawnetRoutes(ctx));
   router.mount('/api/v1/profile', profileRoutes(ctx));
+  router.mount('/api/v1/events', eventRoutes(ctx));
 
   return router;
 }
@@ -61,6 +63,8 @@ const AUTH_WHITELIST: Array<{ method?: string; path: string }> = [
   // (matched via startsWith in isAuthExempt since :did varies)
   // Relay routes — auth is enforced on the target node side
   { path: '/relay' },
+  // Local SSE events — auth enforced separately (session token in query)
+  { method: 'GET', path: '/api/v1/events' },
 ];
 
 function isAuthExempt(method: string, pathname: string): boolean {

packages/node/src/api/types.ts

@@ -13,6 +13,7 @@ import type { OwnerPermissionService } from '../services/owner-permission-servic
 import type { SelfProfileStore } from '../storage/profile-store.js';
 import type { PeerProfileRepository } from '../storage/peer-profile-repository.js';
 import type { ApiProxyService } from '../services/api-proxy-service.js';
+import type { EventPushService } from '../services/event-push-service.js';
 
 export interface ApiServerConfig {
   host: string;
@@ -38,4 +39,5 @@ export interface RuntimeContext {
   peerProfileRepository: PeerProfileRepository;
   configuredPassphrase?: string;
   apiProxyService?: ApiProxyService;
+  eventPushService?: EventPushService;
 }

packages/node/src/app.ts

@@ -11,6 +11,7 @@ import { GroupIndexer } from './indexer/group-indexer.js';
 import { AttachmentService } from './services/attachment-service.js';
 import { ClawNetTransportService } from './services/clawnet-transport-service.js';
 import { ApiProxyService } from './services/api-proxy-service.js';
+import { EventPushService } from './services/event-push-service.js';
 import { ContractProvider } from './services/contract-provider.js';
 import { GroupService } from './services/group-service.js';
 import { IdentityAdapterService } from './services/identity-adapter-service.js';
@@ -51,6 +52,7 @@ export class TelagentNode {
   private attachmentService: AttachmentService | null = null;
   private clawnetTransportService: ClawNetTransportService | null = null;
   private apiProxyService: ApiProxyService | null = null;
+  private eventPushService: EventPushService | null = null;
   private monitoringService: NodeMonitoringService | null = null;
   private ownerPermissionService: OwnerPermissionService | null = null;
   private indexer: GroupIndexer | null = null;
@@ -253,6 +255,12 @@ export class TelagentNode {
       );
     }
 
+    // Event Push Service (real-time SSE for Webapp)
+    this.eventPushService = new EventPushService(
+      this.clawnetGateway,
+      this.apiProxyService ?? undefined,
+    );
+
     this.monitoringService = new NodeMonitoringService({
       thresholds: {
         errorRateWarnRatio: this.config.monitoring.errorRateWarnRatio,
@@ -290,6 +298,7 @@ export class TelagentNode {
       peerProfileRepository: this.peerProfileRepository,
       configuredPassphrase: passphrase ?? undefined,
       apiProxyService: this.apiProxyService ?? undefined,
+      eventPushService: this.eventPushService ?? undefined,
     };
 
     this.apiServer = new ApiServer(runtime);
@@ -303,7 +312,17 @@ export class TelagentNode {
     logger.info('[telagent] [startup] starting API server...');
     await this.apiServer.start();
     this.clawnetTransportService.startListening({
-      onEnvelope: (raw, sourceDid) => this.messageService!.ingestFederatedEnvelope(raw, sourceDid),
+      onEnvelope: async (raw, sourceDid) => {
+        await this.messageService!.ingestFederatedEnvelope(raw, sourceDid);
+        // Push real-time event to local SSE clients
+        this.eventPushService?.emitLocal({
+          type: 'new-envelope',
+          sourceDid,
+          envelopeId: typeof raw.envelopeId === 'string' ? raw.envelopeId : undefined,
+          conversationId: typeof raw.conversationId === 'string' ? raw.conversationId : undefined,
+          atMs: Date.now(),
+        });
+      },
       onAttachment: async (info, _sourceDid) => {
         // A peer relayed an attachment to us via ClawNet P2P.
         // Download from our local ClawNet node and save under the same objectKey
@@ -398,6 +417,7 @@ export class TelagentNode {
 
     this.stopMailboxCleaner();
     this.apiProxyService?.dispose();
+    this.eventPushService?.dispose();
     this.clawnetTransportService?.stopListening();
     await this.apiServer?.stop();
     await this.indexer?.stop();