feat: send event to client upon items change on server (#941)

* feat(websockets): persist connections in mysql

* fix: add sending event to client upon items changed on server

* fix payload

* fix: add cathcing errors

* fix: send changed items event only on a 10% dice roll

packages/syncing-server/src/Domain/Event/DomainEventFactory.ts

@@ -7,6 +7,7 @@ import {
   ItemDumpedEvent,
   ItemRemovedFromSharedVaultEvent,
   ItemRevisionCreationRequestedEvent,
+  ItemsChangedOnServerEvent,
   MessageSentToUserEvent,
   NotificationAddedForUserEvent,
   RevisionsCopyRequestedEvent,
@@ -23,6 +24,25 @@ import { DomainEventFactoryInterface } from './DomainEventFactoryInterface'
 export class DomainEventFactory implements DomainEventFactoryInterface {
   constructor(private timer: TimerInterface) {}
 
+  createItemsChangedOnServerEvent(dto: {
+    userUuid: string
+    sessionUuid: string
+    timestamp: number
+  }): ItemsChangedOnServerEvent {
+    return {
+      type: 'ITEMS_CHANGED_ON_SERVER',
+      createdAt: this.timer.getUTCDate(),
+      meta: {
+        correlation: {
+          userIdentifier: dto.userUuid,
+          userIdentifierType: 'uuid',
+        },
+        origin: DomainEventService.SyncingServer,
+      },
+      payload: dto,
+    }
+  }
+
   createAccountDeletionVerificationPassedEvent(dto: {
     userUuid: string
     email: string
@@ -207,7 +227,11 @@ export class DomainEventFactory implements DomainEventFactoryInterface {
     }
   }
 
-  createWebSocketMessageRequestedEvent(dto: { userUuid: string; message: string }): WebSocketMessageRequestedEvent {
+  createWebSocketMessageRequestedEvent(dto: {
+    userUuid: string
+    message: string
+    originatingSessionUuid?: string
+  }): WebSocketMessageRequestedEvent {
     return {
       type: 'WEB_SOCKET_MESSAGE_REQUESTED',
       createdAt: this.timer.getUTCDate(),

packages/syncing-server/src/Domain/Event/DomainEventFactoryInterface.ts

@@ -5,6 +5,7 @@ import {
   ItemDumpedEvent,
   ItemRemovedFromSharedVaultEvent,
   ItemRevisionCreationRequestedEvent,
+  ItemsChangedOnServerEvent,
   MessageSentToUserEvent,
   NotificationAddedForUserEvent,
   RevisionsCopyRequestedEvent,
@@ -17,7 +18,16 @@ import {
 } from '@standardnotes/domain-events'
 
 export interface DomainEventFactoryInterface {
-  createWebSocketMessageRequestedEvent(dto: { userUuid: string; message: string }): WebSocketMessageRequestedEvent
+  createWebSocketMessageRequestedEvent(dto: {
+    userUuid: string
+    message: string
+    originatingSessionUuid?: string
+  }): WebSocketMessageRequestedEvent
+  createItemsChangedOnServerEvent(dto: {
+    userUuid: string
+    sessionUuid: string
+    timestamp: number
+  }): ItemsChangedOnServerEvent
   createUserInvitedToSharedVaultEvent(dto: {
     invite: {
       uuid: string

packages/syncing-server/src/Domain/UseCase/Syncing/SaveItems/SaveItems.spec.ts

@@ -8,6 +8,10 @@ import { Logger } from 'winston'
 import { ContentType, Dates, Result, Timestamps, Uuid } from '@standardnotes/domain-core'
 import { ItemHash } from '../../../Item/ItemHash'
 import { Item } from '../../../Item/Item'
+import { SendEventToClient } from '../SendEventToClient/SendEventToClient'
+import { DomainEventFactoryInterface } from '../../../Event/DomainEventFactoryInterface'
+import { ItemsChangedOnServerEvent } from '@standardnotes/domain-events'
+import { SelectorInterface } from '@standardnotes/security'
 
 describe('SaveItems', () => {
   let itemSaveValidator: ItemSaveValidatorInterface
@@ -18,11 +22,35 @@ describe('SaveItems', () => {
   let logger: Logger
   let itemHash1: ItemHash
   let savedItem: Item
+  let sendEventToClient: SendEventToClient
+  let domainEventFactory: DomainEventFactoryInterface
+  let deterministicSelector: SelectorInterface<number>
 
   const createUseCase = () =>
-    new SaveItems(itemSaveValidator, itemRepository, timer, saveNewItem, updateExistingItem, logger)
+    new SaveItems(
+      itemSaveValidator,
+      itemRepository,
+      timer,
+      saveNewItem,
+      updateExistingItem,
+      sendEventToClient,
+      domainEventFactory,
+      deterministicSelector,
+      logger,
+    )
 
   beforeEach(() => {
+    deterministicSelector = {} as jest.Mocked<SelectorInterface<number>>
+    deterministicSelector.select = jest.fn().mockReturnValue(1)
+
+    sendEventToClient = {} as jest.Mocked<SendEventToClient>
+    sendEventToClient.execute = jest.fn().mockReturnValue(Result.ok())
+
+    domainEventFactory = {} as jest.Mocked<DomainEventFactoryInterface>
+    domainEventFactory.createItemsChangedOnServerEvent = jest
+      .fn()
+      .mockReturnValue({} as jest.Mocked<ItemsChangedOnServerEvent>)
+
     itemSaveValidator = {} as jest.Mocked<ItemSaveValidatorInterface>
     itemSaveValidator.validate = jest.fn().mockResolvedValue({ passed: true })
 
@@ -92,6 +120,7 @@ describe('SaveItems', () => {
       userUuid: 'user-uuid',
       sessionUuid: 'session-uuid',
     })
+    expect(sendEventToClient.execute).toHaveBeenCalled()
   })
 
   it('should mark items as conflicts if saving new item fails', async () => {
@@ -115,6 +144,7 @@ describe('SaveItems', () => {
         type: 'uuid_conflict',
       },
     ])
+    expect(sendEventToClient.execute).not.toHaveBeenCalled()
   })
 
   it('should mark items as conflicts if saving new item throws an error', async () => {
@@ -197,6 +227,8 @@ describe('SaveItems', () => {
   })
 
   it('should update existing items', async () => {
+    deterministicSelector.select = jest.fn().mockReturnValue(0)
+
     const useCase = createUseCase()
 
     itemRepository.findByUuid = jest.fn().mockResolvedValue(savedItem)
@@ -217,6 +249,7 @@ describe('SaveItems', () => {
       sessionUuid: 'session-uuid',
       performingUserUuid: '00000000-0000-0000-0000-000000000000',
     })
+    expect(sendEventToClient.execute).not.toHaveBeenCalled()
   })
 
   it('should mark items as conflicts if updating existing item fails', async () => {

packages/syncing-server/src/Domain/UseCase/Syncing/SaveItems/SaveItems.ts

@@ -11,6 +11,9 @@ import { ItemSaveValidatorInterface } from '../../../Item/SaveValidator/ItemSave
 import { SaveNewItem } from '../SaveNewItem/SaveNewItem'
 import { UpdateExistingItem } from '../UpdateExistingItem/UpdateExistingItem'
 import { ItemRepositoryInterface } from '../../../Item/ItemRepositoryInterface'
+import { SendEventToClient } from '../SendEventToClient/SendEventToClient'
+import { DomainEventFactoryInterface } from '../../../Event/DomainEventFactoryInterface'
+import { SelectorInterface } from '@standardnotes/security'
 
 export class SaveItems implements UseCaseInterface<SaveItemsResult> {
   private readonly SYNC_TOKEN_VERSION = 2
@@ -21,6 +24,9 @@ export class SaveItems implements UseCaseInterface<SaveItemsResult> {
     private timer: TimerInterface,
     private saveNewItem: SaveNewItem,
     private updateExistingItem: UpdateExistingItem,
+    private sendEventToClient: SendEventToClient,
+    private domainEventFactory: DomainEventFactoryInterface,
+    private deterministicSelector: SelectorInterface<number>,
     private logger: Logger,
   ) {}
 
@@ -133,6 +139,8 @@ export class SaveItems implements UseCaseInterface<SaveItemsResult> {
 
     const syncToken = this.calculateSyncToken(lastUpdatedTimestamp, savedItems)
 
+    await this.notifyOtherClientsOfTheUserThatItemsChanged(dto, savedItems, lastUpdatedTimestamp)
+
     return Result.ok({
       savedItems,
       conflicts,
@@ -140,6 +148,37 @@ export class SaveItems implements UseCaseInterface<SaveItemsResult> {
     })
   }
 
+  private async notifyOtherClientsOfTheUserThatItemsChanged(
+    dto: SaveItemsDTO,
+    savedItems: Item[],
+    lastUpdatedTimestamp: number,
+  ): Promise<void> {
+    if (savedItems.length === 0 || !dto.sessionUuid) {
+      return
+    }
+
+    const tenPercentSpreadArray = Array.from(Array(10).keys())
+    const diceRoll = this.deterministicSelector.select(dto.userUuid, tenPercentSpreadArray)
+    if (diceRoll !== 1) {
+      return
+    }
+
+    const itemsChangedEvent = this.domainEventFactory.createItemsChangedOnServerEvent({
+      userUuid: dto.userUuid,
+      sessionUuid: dto.sessionUuid,
+      timestamp: lastUpdatedTimestamp,
+    })
+    const result = await this.sendEventToClient.execute({
+      userUuid: dto.userUuid,
+      originatingSessionUuid: dto.sessionUuid,
+      event: itemsChangedEvent,
+    })
+    /* istanbul ignore next */
+    if (result.isFailed()) {
+      this.logger.error(`[${dto.userUuid}] Sending items changed event to client failed. Error: ${result.getError()}`)
+    }
+  }
+
   private calculateSyncToken(lastUpdatedTimestamp: number, savedItems: Array<Item>): string {
     if (savedItems.length) {
       const sortedItems = savedItems.sort((itemA: Item, itemB: Item) => {

packages/syncing-server/src/Domain/UseCase/Syncing/SendEventToClient/SendEventToClient.spec.ts

@@ -17,6 +17,8 @@ describe('SendEventToClient', () => {
   beforeEach(() => {
     logger = {} as jest.Mocked<Logger>
     logger.info = jest.fn()
+    logger.debug = jest.fn()
+    logger.error = jest.fn()
 
     domainEventFactory = {} as jest.Mocked<DomainEventFactoryInterface>
     domainEventFactory.createWebSocketMessageRequestedEvent = jest
@@ -58,4 +60,21 @@ describe('SendEventToClient', () => {
 
     expect(result.isFailed()).toBe(true)
   })
+
+  it('should return a failed result if error is thrown', async () => {
+    const useCase = createUseCase()
+
+    domainEventFactory.createWebSocketMessageRequestedEvent = jest.fn().mockImplementation(() => {
+      throw new Error('test')
+    })
+
+    const result = await useCase.execute({
+      userUuid: '00000000-0000-0000-0000-000000000000',
+      event: {
+        type: 'test',
+      } as jest.Mocked<DomainEventInterface>,
+    })
+
+    expect(result.isFailed()).toBe(true)
+  })
 })

packages/syncing-server/src/Domain/UseCase/Syncing/SendEventToClient/SendEventToClient.ts

@@ -13,21 +13,26 @@ export class SendEventToClient implements UseCaseInterface<void> {
   ) {}
 
   async execute(dto: SendEventToClientDTO): Promise<Result<void>> {
-    const userUuidOrError = Uuid.create(dto.userUuid)
-    if (userUuidOrError.isFailed()) {
-      return Result.fail(userUuidOrError.getError())
+    try {
+      const userUuidOrError = Uuid.create(dto.userUuid)
+      if (userUuidOrError.isFailed()) {
+        return Result.fail(userUuidOrError.getError())
+      }
+      const userUuid = userUuidOrError.getValue()
+
+      this.logger.debug(`[WebSockets] Requesting message ${dto.event.type} to user ${dto.userUuid}`)
+
+      const event = this.domainEventFactory.createWebSocketMessageRequestedEvent({
+        userUuid: userUuid.value,
+        message: JSON.stringify(dto.event),
+        originatingSessionUuid: dto.originatingSessionUuid,
+      })
+
+      await this.domainEventPublisher.publish(event)
+
+      return Result.ok()
+    } catch (error) {
+      return Result.fail(`Failed to send event to client: ${(error as Error).message}`)
     }
-    const userUuid = userUuidOrError.getValue()
-
-    this.logger.info(`[WebSockets] Requesting message ${dto.event.type} to user ${dto.userUuid}`)
-
-    const event = this.domainEventFactory.createWebSocketMessageRequestedEvent({
-      userUuid: userUuid.value,
-      message: JSON.stringify(dto.event),
-    })
-
-    await this.domainEventPublisher.publish(event)
-
-    return Result.ok()
   }
 }

packages/syncing-server/src/Domain/UseCase/Syncing/SendEventToClient/SendEventToClientDTO.ts

@@ -3,4 +3,5 @@ import { DomainEventInterface } from '@standardnotes/domain-events'
 export interface SendEventToClientDTO {
   userUuid: string
   event: DomainEventInterface
+  originatingSessionUuid?: string
 }