fix(transport): target resource update notifications

.changeset/resource-subscription-lifecycle.md

@@ -0,0 +1,5 @@
+---
+'@contextvm/sdk': patch
+---
+
+Tighten the resource-subscription lifecycle on `NostrServerTransport`. `resources/subscribe`/`resources/unsubscribe` are now recorded only after the request is actually forwarded to the server, so middleware-dropped requests can no longer leave phantom subscription state behind (and a dropped unsubscribe can never re-add a subscription that never existed). A client's subscriptions are also cleared as soon as the authorization policy rejects it, so a revoked client stops receiving resource updates immediately instead of lingering until session eviction.

.changeset/resource-update-subscriptions.md

@@ -0,0 +1,5 @@
+---
+'@contextvm/sdk': patch
+---
+
+Route `notifications/resources/updated` only to clients subscribed to the matching resource URI. Servers can configure matching for parent/sub-resource relationships.

src/transport/nostr-server-transport.resource-updates.test.ts

@@ -0,0 +1,126 @@
+import { expect, test } from 'bun:test';
+import { Client } from '@contextvm/mcp-sdk/client';
+import { McpServer } from '@contextvm/mcp-sdk/server/mcp';
+import {
+  ResourceUpdatedNotificationSchema,
+  SubscribeRequestSchema,
+  UnsubscribeRequestSchema,
+} from '@contextvm/mcp-sdk/types.js';
+import { generateSecretKey, getPublicKey } from 'nostr-tools/pure';
+import { bytesToHex, hexToBytes } from 'nostr-tools/utils';
+import { sleep } from '../core/utils/utils.js';
+import { PrivateKeySigner } from '../signer/private-key-signer.js';
+import { ApplesauceRelayPool } from '../relay/applesauce-relay-pool.js';
+import { spawnMockRelay } from '../__mocks__/test-relay-helpers.js';
+import { NostrClientTransport } from './nostr-client-transport.js';
+import { NostrServerTransport } from './nostr-server-transport.js';
+
+test('routes resources/updated only to clients subscribed to that resource', async () => {
+  const relay = await spawnMockRelay();
+  const serverPrivateKey = bytesToHex(generateSecretKey());
+  const serverPublicKey = getPublicKey(hexToBytes(serverPrivateKey));
+  const alphaPrivateKey = bytesToHex(generateSecretKey());
+  const betaPrivateKey = bytesToHex(generateSecretKey());
+
+  const server = new McpServer(
+    { name: 'Resource update server', version: '1.0.0' },
+    { capabilities: { resources: { subscribe: true } } },
+  );
+  server.server.setRequestHandler(SubscribeRequestSchema, async () => ({}));
+  server.server.setRequestHandler(UnsubscribeRequestSchema, async () => ({}));
+
+  const serverTransport = new NostrServerTransport({
+    signer: new PrivateKeySigner(serverPrivateKey),
+    relayHandler: new ApplesauceRelayPool([relay.relayUrl]),
+    matchesSubResource: (subscribedUri, updatedUri) =>
+      updatedUri.startsWith(`${subscribedUri}/`),
+  });
+  const alphaClient = new Client({ name: 'Alpha client', version: '1.0.0' });
+  const betaClient = new Client({ name: 'Beta client', version: '1.0.0' });
+  const alphaUpdates: string[] = [];
+  const betaUpdates: string[] = [];
+
+  alphaClient.setNotificationHandler(
+    ResourceUpdatedNotificationSchema,
+    (notification) => {
+      alphaUpdates.push(notification.params.uri);
+    },
+  );
+  betaClient.setNotificationHandler(
+    ResourceUpdatedNotificationSchema,
+    (notification) => {
+      betaUpdates.push(notification.params.uri);
+    },
+  );
+
+  try {
+    await server.connect(serverTransport);
+    await alphaClient.connect(
+      new NostrClientTransport({
+        signer: new PrivateKeySigner(alphaPrivateKey),
+        relayHandler: new ApplesauceRelayPool([relay.relayUrl]),
+        serverPubkey: serverPublicKey,
+      }),
+    );
+    await betaClient.connect(
+      new NostrClientTransport({
+        signer: new PrivateKeySigner(betaPrivateKey),
+        relayHandler: new ApplesauceRelayPool([relay.relayUrl]),
+        serverPubkey: serverPublicKey,
+      }),
+    );
+
+    await alphaClient.subscribeResource({ uri: 'resource://alpha' });
+    await betaClient.subscribeResource({ uri: 'resource://beta' });
+
+    await server.server.sendResourceUpdated({ uri: 'resource://alpha' });
+    await sleep(150);
+
+    expect(alphaUpdates).toEqual(['resource://alpha']);
+    expect(betaUpdates).toEqual([]);
+
+    await server.server.sendResourceUpdated({ uri: 'resource://alpha/child' });
+    await sleep(150);
+
+    expect(alphaUpdates).toEqual([
+      'resource://alpha',
+      'resource://alpha/child',
+    ]);
+    expect(betaUpdates).toEqual([]);
+
+    await server.server.sendResourceUpdated({
+      uri: 'resource://alphabet/child',
+    });
+    await sleep(150);
+
+    expect(alphaUpdates).toEqual([
+      'resource://alpha',
+      'resource://alpha/child',
+    ]);
+    expect(betaUpdates).toEqual([]);
+
+    await alphaClient.unsubscribeResource({ uri: 'resource://alpha' });
+    await server.server.sendResourceUpdated({ uri: 'resource://alpha' });
+    await sleep(150);
+
+    expect(alphaUpdates).toEqual([
+      'resource://alpha',
+      'resource://alpha/child',
+    ]);
+    expect(betaUpdates).toEqual([]);
+
+    await server.server.sendResourceUpdated({ uri: 'resource://alpha/child' });
+    await sleep(150);
+
+    expect(alphaUpdates).toEqual([
+      'resource://alpha',
+      'resource://alpha/child',
+    ]);
+    expect(betaUpdates).toEqual([]);
+  } finally {
+    await alphaClient.close().catch(() => undefined);
+    await betaClient.close().catch(() => undefined);
+    await server.close().catch(() => undefined);
+    relay.stop();
+  }
+});

src/transport/nostr-server-transport.ts

@@ -22,6 +22,10 @@ import { NostrEvent } from 'nostr-tools';
 import { LogLevel } from '../core/utils/logger.js';
 import { withTimeout } from '../core/utils/utils.js';
 import { CorrelationStore } from './nostr-server/correlation-store.js';
+import {
+  SubscriptionStore,
+  type ResourceSubscriptionMatcher,
+} from './nostr-server/subscription-store.js';
 import { ClientSession, SessionStore } from './nostr-server/session-store.js';
 import { LruCache } from '../core/utils/lru-cache.js';
 import { ApplesauceRelayPool } from '../relay/applesauce-relay-pool.js';
@@ -58,6 +62,7 @@ import type { InboundMiddlewareFn } from './middleware.js';
 import type { PaymentInteractionPolicy } from '../payments/types.js';
 
 export type { InboundMiddlewareFn } from './middleware.js';
+export type { ResourceSubscriptionMatcher } from './nostr-server/subscription-store.js';
 /**
  * Options for configuring the NostrServerTransport.
  */
@@ -95,6 +100,12 @@ export interface NostrServerTransportOptions extends BaseNostrTransportOptions {
   logLevel?: LogLevel;
   /** Maximum number of client sessions to keep in memory. @default 1000 */
   maxSessions?: number;
+  /**
+   * Resolve server-defined parent/sub-resource relationships for resource updates.
+   * Exact URI subscriptions always match, even when this callback is omitted.
+   * The callback is only called for different subscribed and updated URIs.
+   */
+  matchesSubResource?: ResourceSubscriptionMatcher;
   /**
    * Whether to inject the client's public key into the _meta field of incoming messages.
    * @default false
@@ -185,6 +196,7 @@ export class NostrServerTransport
 
   private readonly sessionStore: SessionStore;
   private readonly correlationStore: CorrelationStore;
+  private readonly subscriptionStore: SubscriptionStore;
   private readonly authorizationPolicy: AuthorizationPolicy;
   private readonly announcementManager: AnnouncementManager;
   private readonly injectClientPubkey: boolean;
@@ -246,6 +258,8 @@ export class NostrServerTransport
       isAnnouncedServer: options.isAnnouncedServer ?? options.isPublicServer,
     });
 
+    this.subscriptionStore = new SubscriptionStore();
+
     // Initialize session store with eviction callback for correlation cleanup
     this.sessionStore = new SessionStore({
       maxSessions: options.maxSessions ?? 1000,
@@ -257,6 +271,7 @@ export class NostrServerTransport
         // callback would also corrupt the cache's capacity accounting.)
         const removedCount =
           this.correlationStore.removeRoutesForClient(clientPubkey);
+        this.subscriptionStore.removeForClient(clientPubkey);
         this.logger.info(
           `Evicted session for ${clientPubkey} (removed ${removedCount} routes)`,
         );
@@ -350,6 +365,7 @@ export class NostrServerTransport
         await this.outboundResponseRouter.route(response);
       },
       sessionStore: this.sessionStore,
+      subscriptionStore: this.subscriptionStore,
       onClientSessionEvicted: this.onClientSessionEvicted,
       correlationStore: this.correlationStore,
       policy: options.openStream?.policy,
@@ -359,6 +375,7 @@ export class NostrServerTransport
     this.inboundCoordinator = new ServerInboundCoordinator({
       sessionStore: this.sessionStore,
       correlationStore: this.correlationStore,
+      subscriptionStore: this.subscriptionStore,
       authorizationPolicy: this.authorizationPolicy,
       openStreamFactory: this.openStreamFactory,
       inboundMiddlewares: this.inboundMiddlewares,
@@ -442,6 +459,8 @@ export class NostrServerTransport
     this.outboundNotificationBroadcaster = new OutboundNotificationBroadcaster({
       correlationStore: this.correlationStore,
       sessionStore: this.sessionStore,
+      subscriptionStore: this.subscriptionStore,
+      matchesSubResource: options.matchesSubResource,
       sendNotification: this.sendNotification.bind(this),
       enqueueTask: this.taskQueue.add.bind(this.taskQueue),
       logger: this.logger,
@@ -586,6 +605,7 @@ export class NostrServerTransport
       await this.disconnect();
       this.sessionStore.clear();
       this.correlationStore.clear();
+      this.subscriptionStore.clear();
       this.seenEventIds.clear();
       this.oversizedReceiver.clear();
       this.openStreamFactory.getReceiver().clear();