feat: implement NotificationsHub for real-time vault sync notifications

- Added NotificationsHub durable object to handle WebSocket connections for vault sync notifications.
- Integrated SignalR protocol for message framing and communication.
- Updated storage service methods to return revision date and user ID for vault sync notifications.
- Enhanced existing handlers (attachments, ciphers, folders, sends, and import) to notify users of vault sync events.
- Created new notifications handler for WebSocket negotiation and binding user IDs.
- Updated frontend to establish WebSocket connection for receiving vault sync notifications.
- Improved CORS headers to support new notification endpoints.
- Bumped wrangler version in package.json to 4.71.0.

src/handlers/attachments.ts

@@ -1,10 +1,12 @@
 import { Env, Attachment, DEFAULT_DEV_SECRET } from '../types';
+import { notifyUserVaultSync } from '../durable/notifications-hub';
 import { StorageService } from '../services/storage';
 import { jsonResponse, errorResponse } from '../utils/response';
 import { generateUUID } from '../utils/uuid';
 import { createFileDownloadToken, verifyFileDownloadToken } from '../utils/jwt';
 import { cipherToResponse, shouldOmitPasskeysForResponse } from './ciphers';
 import { LIMITS } from '../config/limits';
+import { readActingDeviceIdentifier } from '../utils/device';
 import {
   deleteBlobObject,
   getAttachmentObjectKey,
@@ -13,6 +15,15 @@ import {
   putBlobObject,
 } from '../services/blob-store';
 
+async function notifyVaultSyncForRequest(
+  request: Request,
+  env: Env,
+  userId: string,
+  revisionDate: string
+): Promise<void> {
+  await notifyUserVaultSync(env, userId, revisionDate, readActingDeviceIdentifier(request));
+}
+
 // Format file size to human readable
 function formatSize(bytes: number): string {
   if (bytes < 1024) return `${bytes} Bytes`;
@@ -73,7 +84,10 @@ export async function handleCreateAttachment(
   await storage.addAttachmentToCipher(cipherId, attachmentId);
 
   // Update cipher revision date
-  await storage.updateCipherRevisionDate(cipherId);
+  const revisionInfo = await storage.updateCipherRevisionDate(cipherId);
+  if (revisionInfo) {
+    await notifyVaultSyncForRequest(request, env, revisionInfo.userId, revisionInfo.revisionDate);
+  }
 
   // Get updated cipher for response
   const updatedCipher = await storage.getCipher(cipherId);
@@ -165,7 +179,10 @@ export async function handleUploadAttachment(
   }
 
   // Update cipher revision date
-  await storage.updateCipherRevisionDate(cipherId);
+  const revisionInfo = await storage.updateCipherRevisionDate(cipherId);
+  if (revisionInfo) {
+    await notifyVaultSyncForRequest(request, env, revisionInfo.userId, revisionInfo.revisionDate);
+  }
 
   return new Response(null, { status: 200 });
 }
@@ -304,7 +321,10 @@ export async function handleDeleteAttachment(
   await storage.removeAttachmentFromCipher(cipherId, attachmentId);
 
   // Update cipher revision date
-  await storage.updateCipherRevisionDate(cipherId);
+  const revisionInfo = await storage.updateCipherRevisionDate(cipherId);
+  if (revisionInfo) {
+    await notifyVaultSyncForRequest(request, env, revisionInfo.userId, revisionInfo.revisionDate);
+  }
 
   // Get updated cipher for response
   const updatedCipher = await storage.getCipher(cipherId);

src/handlers/ciphers.ts

@@ -1,9 +1,20 @@
 import { Env, Cipher, CipherResponse, Attachment } from '../types';
 import { StorageService } from '../services/storage';
+import { notifyUserVaultSync } from '../durable/notifications-hub';
 import { jsonResponse, errorResponse } from '../utils/response';
 import { generateUUID } from '../utils/uuid';
 import { deleteAllAttachmentsForCipher } from './attachments';
 import { parsePagination, encodeContinuationToken } from '../utils/pagination';
+import { readActingDeviceIdentifier } from '../utils/device';
+
+async function notifyVaultSyncForRequest(
+  request: Request,
+  env: Env,
+  userId: string,
+  revisionDate: string
+): Promise<void> {
+  await notifyUserVaultSync(env, userId, revisionDate, readActingDeviceIdentifier(request));
+}
 
 function getAliasedProp(source: any, aliases: string[]): { present: boolean; value: any } {
   if (!source || typeof source !== 'object') return { present: false, value: undefined };
@@ -276,7 +287,8 @@ export async function handleCreateCipher(request: Request, env: Env, userId: str
   }
 
   await storage.saveCipher(cipher);
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(
     cipherToResponse(cipher, [], {
@@ -342,7 +354,8 @@ export async function handleUpdateCipher(request: Request, env: Env, userId: str
   }
 
   await storage.saveCipher(cipher);
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(
     cipherToResponse(cipher, [], {
@@ -364,7 +377,8 @@ export async function handleDeleteCipher(request: Request, env: Env, userId: str
   cipher.deletedAt = new Date().toISOString();
   cipher.updatedAt = cipher.deletedAt;
   await storage.saveCipher(cipher);
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(
     cipherToResponse(cipher, [], {
@@ -389,7 +403,8 @@ export async function handleDeleteCipherCompat(request: Request, env: Env, userI
   if (cipher.deletedAt) {
     await deleteAllAttachmentsForCipher(env, id);
     await storage.deleteCipher(id, userId);
-    await storage.updateRevisionDate(userId);
+    const revisionDate = await storage.updateRevisionDate(userId);
+    await notifyVaultSyncForRequest(request, env, userId, revisionDate);
     return new Response(null, { status: 204 });
   }
 
@@ -409,7 +424,8 @@ export async function handlePermanentDeleteCipher(request: Request, env: Env, us
   await deleteAllAttachmentsForCipher(env, id);
 
   await storage.deleteCipher(id, userId);
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return new Response(null, { status: 204 });
 }
@@ -426,7 +442,8 @@ export async function handleRestoreCipher(request: Request, env: Env, userId: st
   cipher.deletedAt = null;
   cipher.updatedAt = new Date().toISOString();
   await storage.saveCipher(cipher);
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(
     cipherToResponse(cipher, [], {
@@ -464,7 +481,8 @@ export async function handlePartialUpdateCipher(request: Request, env: Env, user
   cipher.updatedAt = new Date().toISOString();
 
   await storage.saveCipher(cipher);
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(
     cipherToResponse(cipher, [], {
@@ -493,7 +511,10 @@ export async function handleBulkMoveCiphers(request: Request, env: Env, userId:
     if (!folderOk) return errorResponse('Folder not found', 404);
   }
 
-  await storage.bulkMoveCiphers(body.ids, body.folderId || null, userId);
+  const revisionDate = await storage.bulkMoveCiphers(body.ids, body.folderId || null, userId);
+  if (revisionDate) {
+    await notifyVaultSyncForRequest(request, env, userId, revisionDate);
+  }
 
   return new Response(null, { status: 204 });
 }

src/handlers/folders.ts

@@ -1,9 +1,20 @@
 import { Env, Folder, FolderResponse } from '../types';
+import { notifyUserVaultSync } from '../durable/notifications-hub';
 import { StorageService } from '../services/storage';
 import { jsonResponse, errorResponse } from '../utils/response';
+import { readActingDeviceIdentifier } from '../utils/device';
 import { generateUUID } from '../utils/uuid';
 import { parsePagination, encodeContinuationToken } from '../utils/pagination';
 
+async function notifyVaultSyncForRequest(
+  request: Request,
+  env: Env,
+  userId: string,
+  revisionDate: string
+): Promise<void> {
+  await notifyUserVaultSync(env, userId, revisionDate, readActingDeviceIdentifier(request));
+}
+
 // Convert internal folder to API response format
 function folderToResponse(folder: Folder): FolderResponse {
   return {
@@ -75,7 +86,8 @@ export async function handleCreateFolder(request: Request, env: Env, userId: str
   };
 
   await storage.saveFolder(folder);
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(folderToResponse(folder), 200);
 }
@@ -102,7 +114,8 @@ export async function handleUpdateFolder(request: Request, env: Env, userId: str
   folder.updatedAt = new Date().toISOString();
 
   await storage.saveFolder(folder);
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(folderToResponse(folder));
 }
@@ -118,7 +131,8 @@ export async function handleDeleteFolder(request: Request, env: Env, userId: str
 
   await storage.clearFolderFromCiphers(userId, id);
   await storage.deleteFolder(id, userId);
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return new Response(null, { status: 204 });
 }

src/handlers/import.ts

@@ -1,6 +1,8 @@
 import { Env, Cipher, Folder, CipherType } from '../types';
+import { notifyUserVaultSync } from '../durable/notifications-hub';
 import { StorageService } from '../services/storage';
 import { errorResponse, jsonResponse } from '../utils/response';
+import { readActingDeviceIdentifier } from '../utils/device';
 import { generateUUID } from '../utils/uuid';
 import { LIMITS } from '../config/limits';
 import { normalizeCipherLoginForStorage, normalizeCipherSshKeyForCompatibility } from './ciphers';
@@ -268,7 +270,8 @@ export async function handleCiphersImport(request: Request, env: Env, userId: st
   }
 
   // Update revision date
-  await storage.updateRevisionDate(userId);
+  const revisionDate = await storage.updateRevisionDate(userId);
+  await notifyUserVaultSync(env, userId, revisionDate, readActingDeviceIdentifier(request));
 
   if (returnCipherMap) {
     return jsonResponse({

src/handlers/notifications.ts

@@ -0,0 +1,61 @@
+import { AuthService } from '../services/auth';
+import type { Env } from '../types';
+import { errorResponse, jsonResponse } from '../utils/response';
+import { generateUUID } from '../utils/uuid';
+
+function extractAccessToken(request: Request): string | null {
+  const url = new URL(request.url);
+  const queryToken = String(url.searchParams.get('access_token') || '').trim();
+  if (queryToken) return queryToken;
+
+  const authHeader = String(request.headers.get('Authorization') || '').trim();
+  const match = authHeader.match(/^Bearer\s+(.+)$/i);
+  return match?.[1]?.trim() || null;
+}
+
+async function authenticateNotificationsRequest(request: Request, env: Env): Promise<string | null> {
+  const accessToken = extractAccessToken(request);
+  if (!accessToken) return null;
+
+  const auth = new AuthService(env);
+  const payload = await auth.verifyAccessToken(`Bearer ${accessToken}`);
+  return payload?.sub || null;
+}
+
+export async function handleNotificationsNegotiate(request: Request, env: Env): Promise<Response> {
+  const userId = await authenticateNotificationsRequest(request, env);
+  if (!userId) return errorResponse('Unauthorized', 401);
+
+  const connectionId = generateUUID();
+  return jsonResponse({
+    connectionId,
+    connectionToken: connectionId,
+    negotiateVersion: 1,
+    availableTransports: [
+      {
+        transport: 'WebSockets',
+        transferFormats: ['Text', 'Binary'],
+      },
+    ],
+  });
+}
+
+export async function handleNotificationsHub(request: Request, env: Env): Promise<Response> {
+  const userId = await authenticateNotificationsRequest(request, env);
+  if (!userId) return errorResponse('Unauthorized', 401);
+  if (request.headers.get('Upgrade')?.toLowerCase() !== 'websocket') {
+    return errorResponse('Expected websocket', 426);
+  }
+
+  const id = env.NOTIFICATIONS_HUB.idFromName(userId);
+  const stub = env.NOTIFICATIONS_HUB.get(id);
+  await stub.fetch('https://notifications/internal/bind-user', {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      'X-NodeWarden-UserId': userId,
+    },
+    body: JSON.stringify({ userId }),
+  });
+  return stub.fetch(request);
+}

src/handlers/sends.ts

@@ -1,7 +1,9 @@
 import { Env, Send, SendAuthType, SendResponse, SendType, DEFAULT_DEV_SECRET } from '../types';
+import { notifyUserVaultSync } from '../durable/notifications-hub';
 import { StorageService } from '../services/storage';
 import { RateLimitService, getClientIdentifier } from '../services/ratelimit';
 import { jsonResponse, errorResponse } from '../utils/response';
+import { readActingDeviceIdentifier } from '../utils/device';
 import { generateUUID } from '../utils/uuid';
 import { parsePagination, encodeContinuationToken } from '../utils/pagination';
 import { LIMITS } from '../config/limits';
@@ -23,6 +25,15 @@ const SEND_INACCESSIBLE_MSG = 'Send does not exist or is no longer available';
 const SEND_PASSWORD_ITERATIONS = 100_000;
 const SEND_PASSWORD_LIMIT_SCOPE = 'send-password';
 
+async function notifyVaultSyncForRequest(
+  request: Request,
+  env: Env,
+  userId: string,
+  revisionDate: string
+): Promise<void> {
+  await notifyUserVaultSync(env, userId, revisionDate, readActingDeviceIdentifier(request));
+}
+
 function getAliasedProp(source: unknown, aliases: string[]): { present: boolean; value: unknown } {
   if (!source || typeof source !== 'object') return { present: false, value: undefined };
   for (const key of aliases) {
@@ -604,7 +615,8 @@ export async function handleCreateSend(request: Request, env: Env, userId: strin
   }
 
   await storage.saveSend(send);
-  await storage.updateRevisionDate(userId);
+  let revisionDate = await storage.updateRevisionDate(userId);
+    await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(sendToResponse(send));
 }
@@ -727,7 +739,8 @@ export async function handleCreateFileSendV2(request: Request, env: Env, userId:
   }
 
   await storage.saveSend(send);
-  await storage.updateRevisionDate(userId);
+  let revisionDate = await storage.updateRevisionDate(userId);
+    await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse({
     fileUploadType: 0,
@@ -835,7 +848,8 @@ export async function handleUploadSendFile(
     return errorResponse('Attachment storage is not configured', 500);
   }
 
-  await storage.updateRevisionDate(userId);
+  let revisionDate = await storage.updateRevisionDate(userId);
+    await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return new Response(null, { status: 200 });
 }
@@ -981,7 +995,8 @@ export async function handleUpdateSend(request: Request, env: Env, userId: strin
 
   send.updatedAt = new Date().toISOString();
   await storage.saveSend(send);
-  await storage.updateRevisionDate(userId);
+  let revisionDate = await storage.updateRevisionDate(userId);
+    await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(sendToResponse(send));
 }
@@ -1004,7 +1019,8 @@ export async function handleDeleteSend(request: Request, env: Env, userId: strin
   }
 
   await storage.deleteSend(sendId, userId);
-  await storage.updateRevisionDate(userId);
+  let revisionDate = await storage.updateRevisionDate(userId);
+    await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return new Response(null, { status: 200 });
 }
@@ -1021,7 +1037,8 @@ export async function handleRemoveSendPassword(request: Request, env: Env, userI
   await setSendPassword(send, null);
   send.updatedAt = new Date().toISOString();
   await storage.saveSend(send);
-  await storage.updateRevisionDate(userId);
+  let revisionDate = await storage.updateRevisionDate(userId);
+    await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(sendToResponse(send));
 }
@@ -1039,7 +1056,8 @@ export async function handleRemoveSendAuth(request: Request, env: Env, userId: s
   send.emails = null;
   send.updatedAt = new Date().toISOString();
   await storage.saveSend(send);
-  await storage.updateRevisionDate(userId);
+  let revisionDate = await storage.updateRevisionDate(userId);
+    await notifyVaultSyncForRequest(request, env, userId, revisionDate);
 
   return jsonResponse(sendToResponse(send));
 }
@@ -1100,7 +1118,8 @@ export async function handleAccessSend(request: Request, env: Env, accessId: str
       return errorResponse(SEND_INACCESSIBLE_MSG, 404);
     }
     send.accessCount += 1;
-    await storage.updateRevisionDate(send.userId);
+    const revisionDate = await storage.updateRevisionDate(send.userId);
+      await notifyVaultSyncForRequest(request, env, send.userId, revisionDate);
   }
 
   const creatorIdentifier = await getCreatorIdentifier(storage, send);
@@ -1173,7 +1192,8 @@ export async function handleAccessSendFile(
     return errorResponse(SEND_INACCESSIBLE_MSG, 404);
   }
   send.accessCount += 1;
-  await storage.updateRevisionDate(send.userId);
+  const revisionDate = await storage.updateRevisionDate(send.userId);
+  await notifyVaultSyncForRequest(request, env, send.userId, revisionDate);
 
   const token = await createSendFileDownloadToken(send.id, fileId, secret);
   const url = new URL(request.url);
@@ -1213,7 +1233,8 @@ export async function handleAccessSendV2(request: Request, env: Env): Promise<Re
       return errorResponse(SEND_INACCESSIBLE_MSG, 404);
     }
     send.accessCount += 1;
-    await storage.updateRevisionDate(send.userId);
+    const revisionDate = await storage.updateRevisionDate(send.userId);
+      await notifyVaultSyncForRequest(request, env, send.userId, revisionDate);
   }
 
   const creatorIdentifier = await getCreatorIdentifier(storage, send);
@@ -1254,7 +1275,8 @@ export async function handleAccessSendFileV2(request: Request, env: Env, fileId:
     return errorResponse(SEND_INACCESSIBLE_MSG, 404);
   }
   send.accessCount += 1;
-  await storage.updateRevisionDate(send.userId);
+  const revisionDate = await storage.updateRevisionDate(send.userId);
+  await notifyVaultSyncForRequest(request, env, send.userId, revisionDate);
 
   const downloadToken = await createSendFileDownloadToken(send.id, fileId, secret);
   const url = new URL(request.url);