feat: add direct event handling for home server (#608)

* feat(domain-events-infra): add direct call event message handler

* feat: add direct publishing of events into handlers

* fix: validating sessions with direct calls

.pnp.cjs

@@ -4625,6 +4625,7 @@ const RAW_RUNTIME_STATE =
           ["@standardnotes/api-gateway", "workspace:packages/api-gateway"],\
           ["@standardnotes/auth-server", "workspace:packages/auth"],\
           ["@standardnotes/domain-core", "workspace:packages/domain-core"],\
+          ["@standardnotes/domain-events-infra", "workspace:packages/domain-events-infra"],\
           ["@types/cors", "npm:2.8.13"],\
           ["@types/express", "npm:4.17.17"],\
           ["@types/prettyjson", "npm:0.0.30"],\

packages/api-gateway/src/Controller/AuthMiddleware.ts

@@ -5,17 +5,17 @@ import { NextFunction, Request, Response } from 'express'
 import { inject, injectable } from 'inversify'
 import { BaseMiddleware } from 'inversify-express-utils'
 import { verify } from 'jsonwebtoken'
-import { AxiosError, AxiosInstance } from 'axios'
+import { AxiosError } from 'axios'
 import { Logger } from 'winston'
 
 import { TYPES } from '../Bootstrap/Types'
 import { CrossServiceTokenCacheInterface } from '../Service/Cache/CrossServiceTokenCacheInterface'
+import { ServiceProxyInterface } from '../Service/Http/ServiceProxyInterface'
 
 @injectable()
 export class AuthMiddleware extends BaseMiddleware {
   constructor(
-    @inject(TYPES.HTTPClient) private httpClient: AxiosInstance,
-    @inject(TYPES.AUTH_SERVER_URL) private authServerUrl: string,
+    @inject(TYPES.ServiceProxy) private serviceProxy: ServiceProxyInterface,
     @inject(TYPES.AUTH_JWT_SECRET) private jwtSecret: string,
     @inject(TYPES.CROSS_SERVICE_TOKEN_CACHE_TTL) private crossServiceTokenCacheTTL: number,
     @inject(TYPES.CrossServiceTokenCache) private crossServiceTokenCache: CrossServiceTokenCacheInterface,
@@ -47,26 +47,16 @@ export class AuthMiddleware extends BaseMiddleware {
       }
 
       if (crossServiceToken === null) {
-        const authResponse = await this.httpClient.request({
-          method: 'POST',
-          headers: {
-            Authorization: authHeaderValue,
-            Accept: 'application/json',
-          },
-          validateStatus: (status: number) => {
-            return status >= 200 && status < 500
-          },
-          url: `${this.authServerUrl}/sessions/validate`,
-        })
+        const authResponse = await this.serviceProxy.validateSession(authHeaderValue)
 
         if (authResponse.status > 200) {
-          response.setHeader('content-type', authResponse.headers['content-type'] as string)
+          response.setHeader('content-type', authResponse.headers.contentType)
           response.status(authResponse.status).send(authResponse.data)
 
           return
         }
 
-        crossServiceToken = authResponse.data.authToken
+        crossServiceToken = (authResponse.data as { authToken: string }).authToken
         crossServiceTokenFetchedFromCache = false
       }
 
@@ -94,9 +84,7 @@ export class AuthMiddleware extends BaseMiddleware {
         ? JSON.stringify((error as AxiosError).response?.data)
         : (error as Error).message
 
-      this.logger.error(
-        `Could not pass the request to ${this.authServerUrl}/sessions/validate on underlying service: ${errorMessage}`,
-      )
+      this.logger.error(`Could not pass the request to sessions/validate on underlying service: ${errorMessage}`)
 
       this.logger.debug('Response error: %O', (error as AxiosError).response ?? error)
 

packages/api-gateway/src/Service/Http/HttpServiceProxy.ts

@@ -24,6 +24,30 @@ export class HttpServiceProxy implements ServiceProxyInterface {
     @inject(TYPES.Logger) private logger: Logger,
   ) {}
 
+  async validateSession(
+    authorizationHeaderValue: string,
+  ): Promise<{ status: number; data: unknown; headers: { contentType: string } }> {
+    const authResponse = await this.httpClient.request({
+      method: 'POST',
+      headers: {
+        Authorization: authorizationHeaderValue,
+        Accept: 'application/json',
+      },
+      validateStatus: (status: number) => {
+        return status >= 200 && status < 500
+      },
+      url: `${this.authServerUrl}/sessions/validate`,
+    })
+
+    return {
+      status: authResponse.status,
+      data: authResponse.data,
+      headers: {
+        contentType: authResponse.headers['content-type'] as string,
+      },
+    }
+  }
+
   async callSyncingServer(
     request: Request,
     response: Response,

packages/api-gateway/src/Service/Http/ServiceProxyInterface.ts

@@ -49,4 +49,11 @@ export interface ServiceProxyInterface {
     endpointOrMethodIdentifier: string,
     payload?: Record<string, unknown> | string,
   ): Promise<void>
+  validateSession(authorizationHeaderValue: string): Promise<{
+    status: number
+    data: unknown
+    headers: {
+      contentType: string
+    }
+  }>
 }

packages/api-gateway/src/Service/Proxy/DirectCallServiceProxy.ts

@@ -6,6 +6,34 @@ import { ServiceContainerInterface, ServiceIdentifier } from '@standardnotes/dom
 export class DirectCallServiceProxy implements ServiceProxyInterface {
   constructor(private serviceContainer: ServiceContainerInterface) {}
 
+  async validateSession(
+    authorizationHeaderValue: string,
+  ): Promise<{ status: number; data: unknown; headers: { contentType: string } }> {
+    const authService = this.serviceContainer.get(ServiceIdentifier.create(ServiceIdentifier.NAMES.Auth).getValue())
+    if (!authService) {
+      throw new Error('Auth service not found')
+    }
+
+    const serviceResponse = (await authService.handleRequest(
+      {
+        headers: {
+          authorization: authorizationHeaderValue,
+        },
+      } as never,
+      {} as never,
+      'auth.sessions.validate',
+    )) as {
+      statusCode: number
+      json: Record<string, unknown>
+    }
+
+    return {
+      status: serviceResponse.statusCode,
+      data: serviceResponse.json,
+      headers: { contentType: 'application/json' },
+    }
+  }
+
   async callEmailServer(_request: Request, _response: Response, _endpointOrMethodIdentifier: string): Promise<void> {
     throw new Error('Email server is not available.')
   }

packages/api-gateway/src/Service/Resolver/EndpointResolver.ts

@@ -4,6 +4,8 @@ export class EndpointResolver implements EndpointResolverInterface {
   constructor(private isConfiguredForHomeServer: boolean) {}
 
   private readonly endpointToIdentifierMap: Map<string, string> = new Map([
+    // Auth Middleware
+    ['[POST]:sessions/validate', 'auth.sessions.validate'],
     // Actions Controller
     ['[POST]:auth/sign_in', 'auth.signIn'],
     ['[GET]:auth/params', 'auth.params'],

packages/auth/bin/server.ts

@@ -4,7 +4,6 @@ import 'newrelic'
 
 import '../src/Controller/HealthCheckController'
 import '../src/Controller/SessionController'
-import '../src/Controller/SessionsController'
 import '../src/Controller/UsersController'
 import '../src/Controller/SettingsController'
 import '../src/Controller/FeaturesController'
@@ -18,6 +17,7 @@ import '../src/Controller/SubscriptionSettingsController'
 
 import '../src/Infra/InversifyExpressUtils/InversifyExpressAuthController'
 import '../src/Infra/InversifyExpressUtils/InversifyExpressAuthenticatorsController'
+import '../src/Infra/InversifyExpressUtils/InversifyExpressSessionsController'
 import '../src/Infra/InversifyExpressUtils/InversifyExpressSubscriptionInvitesController'
 import '../src/Infra/InversifyExpressUtils/InversifyExpressUserRequestsController'
 import '../src/Infra/InversifyExpressUtils/InversifyExpressWebSocketsController'

packages/auth/src/Bootstrap/Container.ts

@@ -6,6 +6,7 @@ import { Container } from 'inversify'
 import {
   DomainEventHandlerInterface,
   DomainEventMessageHandlerInterface,
+  DomainEventPublisherInterface,
   DomainEventSubscriberFactoryInterface,
 } from '@standardnotes/domain-events'
 import { TimerInterface, Timer } from '@standardnotes/time'
@@ -90,6 +91,8 @@ import { FeatureService } from '../Domain/Feature/FeatureService'
 import { SettingServiceInterface } from '../Domain/Setting/SettingServiceInterface'
 import { ExtensionKeyGrantedEventHandler } from '../Domain/Handler/ExtensionKeyGrantedEventHandler'
 import {
+  DirectCallDomainEventPublisher,
+  DirectCallEventMessageHandler,
   SNSDomainEventPublisher,
   SQSDomainEventSubscriberFactory,
   SQSEventMessageHandler,
@@ -237,12 +240,19 @@ import { InversifyExpressAuthenticatorsController } from '../Infra/InversifyExpr
 import { InversifyExpressSubscriptionInvitesController } from '../Infra/InversifyExpressUtils/InversifyExpressSubscriptionInvitesController'
 import { InversifyExpressUserRequestsController } from '../Infra/InversifyExpressUtils/InversifyExpressUserRequestsController'
 import { InversifyExpressWebSocketsController } from '../Infra/InversifyExpressUtils/InversifyExpressWebSocketsController'
+import { InversifyExpressSessionsController } from '../Infra/InversifyExpressUtils/InversifyExpressSessionsController'
 
 // eslint-disable-next-line @typescript-eslint/no-var-requires
 const newrelicFormatter = require('@newrelic/winston-enricher')
 
 export class ContainerConfigLoader {
-  async load(controllerConatiner?: ControllerContainerInterface): Promise<Container> {
+  async load(configuration?: {
+    controllerConatiner?: ControllerContainerInterface
+    directCallDomainEventPublisher?: DirectCallDomainEventPublisher
+  }): Promise<Container> {
+    const directCallDomainEventPublisher =
+      configuration?.directCallDomainEventPublisher ?? new DirectCallDomainEventPublisher()
+
     const env: Env = new Env()
     env.load()
 
@@ -280,33 +290,35 @@ export class ContainerConfigLoader {
 
     container.bind<TimerInterface>(TYPES.Auth_Timer).toConstantValue(new Timer())
 
-    const snsConfig: SNSClientConfig = {
-      region: env.get('SNS_AWS_REGION', true),
-    }
-    if (env.get('SNS_ENDPOINT', true)) {
-      snsConfig.endpoint = env.get('SNS_ENDPOINT', true)
-    }
-    if (env.get('SNS_ACCESS_KEY_ID', true) && env.get('SNS_SECRET_ACCESS_KEY', true)) {
-      snsConfig.credentials = {
-        accessKeyId: env.get('SNS_ACCESS_KEY_ID', true),
-        secretAccessKey: env.get('SNS_SECRET_ACCESS_KEY', true),
+    if (!isConfiguredForHomeServer) {
+      const snsConfig: SNSClientConfig = {
+        region: env.get('SNS_AWS_REGION', true),
       }
-    }
-    container.bind<SNSClient>(TYPES.Auth_SNS).toConstantValue(new SNSClient(snsConfig))
+      if (env.get('SNS_ENDPOINT', true)) {
+        snsConfig.endpoint = env.get('SNS_ENDPOINT', true)
+      }
+      if (env.get('SNS_ACCESS_KEY_ID', true) && env.get('SNS_SECRET_ACCESS_KEY', true)) {
+        snsConfig.credentials = {
+          accessKeyId: env.get('SNS_ACCESS_KEY_ID', true),
+          secretAccessKey: env.get('SNS_SECRET_ACCESS_KEY', true),
+        }
+      }
+      container.bind<SNSClient>(TYPES.Auth_SNS).toConstantValue(new SNSClient(snsConfig))
 
-    const sqsConfig: SQSClientConfig = {
-      region: env.get('SQS_AWS_REGION', true),
-    }
-    if (env.get('SQS_ENDPOINT', true)) {
-      sqsConfig.endpoint = env.get('SQS_ENDPOINT', true)
-    }
-    if (env.get('SQS_ACCESS_KEY_ID', true) && env.get('SQS_SECRET_ACCESS_KEY', true)) {
-      sqsConfig.credentials = {
-        accessKeyId: env.get('SQS_ACCESS_KEY_ID', true),
-        secretAccessKey: env.get('SQS_SECRET_ACCESS_KEY', true),
+      const sqsConfig: SQSClientConfig = {
+        region: env.get('SQS_AWS_REGION', true),
       }
+      if (env.get('SQS_ENDPOINT', true)) {
+        sqsConfig.endpoint = env.get('SQS_ENDPOINT', true)
+      }
+      if (env.get('SQS_ACCESS_KEY_ID', true) && env.get('SQS_SECRET_ACCESS_KEY', true)) {
+        sqsConfig.credentials = {
+          accessKeyId: env.get('SQS_ACCESS_KEY_ID', true),
+          secretAccessKey: env.get('SQS_SECRET_ACCESS_KEY', true),
+        }
+      }
+      container.bind<SQSClient>(TYPES.Auth_SQS).toConstantValue(new SQSClient(sqsConfig))
     }
-    container.bind<SQSClient>(TYPES.Auth_SQS).toConstantValue(new SQSClient(sqsConfig))
 
     // Mapping
     container
@@ -649,9 +661,11 @@ export class ContainerConfigLoader {
     container.bind<UserSubscriptionServiceInterface>(TYPES.Auth_UserSubscriptionService).to(UserSubscriptionService)
 
     container
-      .bind<SNSDomainEventPublisher>(TYPES.Auth_DomainEventPublisher)
+      .bind<DomainEventPublisherInterface>(TYPES.Auth_DomainEventPublisher)
       .toConstantValue(
-        new SNSDomainEventPublisher(container.get(TYPES.Auth_SNS), container.get(TYPES.Auth_SNS_TOPIC_ARN)),
+        isConfiguredForHomeServer
+          ? directCallDomainEventPublisher
+          : new SNSDomainEventPublisher(container.get(TYPES.Auth_SNS), container.get(TYPES.Auth_SNS_TOPIC_ARN)),
       )
 
     // use cases
@@ -836,7 +850,7 @@ export class ContainerConfigLoader {
     // Controller
     container
       .bind<ControllerContainerInterface>(TYPES.Auth_ControllerContainer)
-      .toConstantValue(controllerConatiner ?? new ControllerContainer())
+      .toConstantValue(configuration?.controllerConatiner ?? new ControllerContainer())
     container
       .bind<AuthController>(TYPES.Auth_AuthController)
       .toConstantValue(
@@ -956,22 +970,34 @@ export class ContainerConfigLoader {
       ['EMAIL_SUBSCRIPTION_UNSUBSCRIBED', container.get(TYPES.Auth_EmailSubscriptionUnsubscribedEventHandler)],
     ])
 
-    container
-      .bind<DomainEventMessageHandlerInterface>(TYPES.Auth_DomainEventMessageHandler)
-      .toConstantValue(
-        env.get('NEW_RELIC_ENABLED', true) === 'true'
-          ? new SQSNewRelicEventMessageHandler(eventHandlers, container.get(TYPES.Auth_Logger))
-          : new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Auth_Logger)),
-      )
-    container
-      .bind<DomainEventSubscriberFactoryInterface>(TYPES.Auth_DomainEventSubscriberFactory)
-      .toConstantValue(
-        new SQSDomainEventSubscriberFactory(
-          container.get(TYPES.Auth_SQS),
-          container.get(TYPES.Auth_SQS_QUEUE_URL),
-          container.get(TYPES.Auth_DomainEventMessageHandler),
-        ),
+    if (isConfiguredForHomeServer) {
+      const directCallEventMessageHandler = new DirectCallEventMessageHandler(
+        eventHandlers,
+        container.get(TYPES.Auth_Logger),
       )
+      directCallDomainEventPublisher.register(directCallEventMessageHandler)
+      container
+        .bind<DomainEventMessageHandlerInterface>(TYPES.Auth_DomainEventMessageHandler)
+        .toConstantValue(directCallEventMessageHandler)
+    } else {
+      container
+        .bind<DomainEventMessageHandlerInterface>(TYPES.Auth_DomainEventMessageHandler)
+        .toConstantValue(
+          env.get('NEW_RELIC_ENABLED', true) === 'true'
+            ? new SQSNewRelicEventMessageHandler(eventHandlers, container.get(TYPES.Auth_Logger))
+            : new SQSEventMessageHandler(eventHandlers, container.get(TYPES.Auth_Logger)),
+        )
+
+      container
+        .bind<DomainEventSubscriberFactoryInterface>(TYPES.Auth_DomainEventSubscriberFactory)
+        .toConstantValue(
+          new SQSDomainEventSubscriberFactory(
+            container.get(TYPES.Auth_SQS),
+            container.get(TYPES.Auth_SQS_QUEUE_URL),
+            container.get(TYPES.Auth_DomainEventMessageHandler),
+          ),
+        )
+    }
 
     container
       .bind<InversifyExpressAuthController>(TYPES.Auth_InversifyExpressAuthController)
@@ -987,6 +1013,8 @@ export class ContainerConfigLoader {
           container.get(TYPES.Auth_ControllerContainer),
         ),
       )
+
+    // Inversify Controllers
     container
       .bind<InversifyExpressAuthenticatorsController>(TYPES.Auth_InversifyExpressAuthenticatorsController)
       .toConstantValue(
@@ -1020,6 +1048,17 @@ export class ContainerConfigLoader {
           container.get(TYPES.Auth_ControllerContainer),
         ),
       )
+    container
+      .bind<InversifyExpressSessionsController>(TYPES.Auth_SessionsController)
+      .toConstantValue(
+        new InversifyExpressSessionsController(
+          container.get(TYPES.Auth_GetActiveSessionsForUser),
+          container.get(TYPES.Auth_AuthenticateRequest),
+          container.get(TYPES.Auth_SessionProjector),
+          container.get(TYPES.Auth_CreateCrossServiceToken),
+          container.get(TYPES.Auth_ControllerContainer),
+        ),
+      )
 
     return container
   }

packages/auth/src/Bootstrap/Service.ts

@@ -6,11 +6,13 @@ import {
 } from '@standardnotes/domain-core'
 
 import { ContainerConfigLoader } from './Container'
+import { DirectCallDomainEventPublisher } from '@standardnotes/domain-events-infra'
 
 export class Service implements ServiceInterface {
   constructor(
     private serviceContainer: ServiceContainerInterface,
     private controllerContainer: ControllerContainerInterface,
+    private directCallDomainEventPublisher: DirectCallDomainEventPublisher,
   ) {
     this.serviceContainer.register(ServiceIdentifier.create(ServiceIdentifier.NAMES.Auth).getValue(), this)
   }
@@ -28,7 +30,10 @@ export class Service implements ServiceInterface {
   async getContainer(): Promise<unknown> {
     const config = new ContainerConfigLoader()
 
-    return config.load(this.controllerContainer)
+    return config.load({
+      controllerConatiner: this.controllerContainer,
+      directCallDomainEventPublisher: this.directCallDomainEventPublisher,
+    })
   }
 
   getId(): ServiceIdentifier {

packages/auth/src/Bootstrap/Types.ts

@@ -222,6 +222,7 @@ const TYPES = {
   Auth_InversifyExpressSubscriptionInvitesController: Symbol.for('Auth_InversifyExpressSubscriptionInvitesController'),
   Auth_InversifyExpressUserRequestsController: Symbol.for('Auth_InversifyExpressUserRequestsController'),
   Auth_InversifyExpressWebSocketsController: Symbol.for('Auth_InversifyExpressWebSocketsController'),
+  Auth_SessionsController: Symbol.for('Auth_SessionsController'),
 }
 
 export default TYPES

packages/auth/src/Infra/InversifyExpressUtils/InversifyExpressSessionsController.spec.ts

@@ -2,17 +2,18 @@ import 'reflect-metadata'
 
 import * as express from 'express'
 
-import { SessionsController } from './SessionsController'
+import { InversifyExpressSessionsController } from './InversifyExpressSessionsController'
 import { results } from 'inversify-express-utils'
-import { Session } from '../Domain/Session/Session'
-import { ProjectorInterface } from '../Projection/ProjectorInterface'
-import { GetActiveSessionsForUser } from '../Domain/UseCase/GetActiveSessionsForUser'
-import { AuthenticateRequest } from '../Domain/UseCase/AuthenticateRequest'
-import { User } from '../Domain/User/User'
-import { CreateCrossServiceToken } from '../Domain/UseCase/CreateCrossServiceToken/CreateCrossServiceToken'
 import { ControllerContainerInterface } from '@standardnotes/domain-core'
+import { User } from '@standardnotes/responses'
 
-describe('SessionsController', () => {
+import { AuthenticateRequest } from '../../Domain/UseCase/AuthenticateRequest'
+import { CreateCrossServiceToken } from '../../Domain/UseCase/CreateCrossServiceToken/CreateCrossServiceToken'
+import { GetActiveSessionsForUser } from '../../Domain/UseCase/GetActiveSessionsForUser'
+import { ProjectorInterface } from '../../Projection/ProjectorInterface'
+import { Session } from '../../Domain/Session/Session'
+
+describe('InversifyExpressSessionsController', () => {
   let getActiveSessionsForUser: GetActiveSessionsForUser
   let authenticateRequest: AuthenticateRequest
   let sessionProjector: ProjectorInterface<Session>
@@ -24,7 +25,7 @@ describe('SessionsController', () => {
   let controllerContainer: ControllerContainerInterface
 
   const createController = () =>
-    new SessionsController(
+    new InversifyExpressSessionsController(
       getActiveSessionsForUser,
       authenticateRequest,
       sessionProjector,

packages/auth/src/Infra/InversifyExpressUtils/InversifyExpressSessionsController.ts

@@ -8,18 +8,19 @@ import {
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
   results,
 } from 'inversify-express-utils'
-import TYPES from '../Bootstrap/Types'
-import { Session } from '../Domain/Session/Session'
-import { AuthenticateRequest } from '../Domain/UseCase/AuthenticateRequest'
-import { GetActiveSessionsForUser } from '../Domain/UseCase/GetActiveSessionsForUser'
-import { User } from '../Domain/User/User'
-import { ProjectorInterface } from '../Projection/ProjectorInterface'
-import { SessionProjector } from '../Projection/SessionProjector'
-import { CreateCrossServiceToken } from '../Domain/UseCase/CreateCrossServiceToken/CreateCrossServiceToken'
 import { ControllerContainerInterface } from '@standardnotes/domain-core'
 
+import TYPES from '../../Bootstrap/Types'
+import { AuthenticateRequest } from '../../Domain/UseCase/AuthenticateRequest'
+import { CreateCrossServiceToken } from '../../Domain/UseCase/CreateCrossServiceToken/CreateCrossServiceToken'
+import { GetActiveSessionsForUser } from '../../Domain/UseCase/GetActiveSessionsForUser'
+import { ProjectorInterface } from '../../Projection/ProjectorInterface'
+import { SessionProjector } from '../../Projection/SessionProjector'
+import { User } from '../../Domain/User/User'
+import { Session } from '../../Domain/Session/Session'
+
 @controller('/sessions')
-export class SessionsController extends BaseHttpController {
+export class InversifyExpressSessionsController extends BaseHttpController {
   constructor(
     @inject(TYPES.Auth_GetActiveSessionsForUser) private getActiveSessionsForUser: GetActiveSessionsForUser,
     @inject(TYPES.Auth_AuthenticateRequest) private authenticateRequest: AuthenticateRequest,
@@ -30,6 +31,7 @@ export class SessionsController extends BaseHttpController {
     super()
 
     this.controllerContainer.register('auth.sessions.list', this.getSessions.bind(this))
+    this.controllerContainer.register('auth.sessions.validate', this.validate.bind(this))
   }
 
   @httpPost('/validate')

packages/domain-events-infra/src/Infra/DirectCall/DirectCallDomainEventPublisher.ts

@@ -0,0 +1,17 @@
+import {
+  DomainEventInterface,
+  DomainEventMessageHandlerInterface,
+  DomainEventPublisherInterface,
+} from '@standardnotes/domain-events'
+
+export class DirectCallDomainEventPublisher implements DomainEventPublisherInterface {
+  private handlers: Array<DomainEventMessageHandlerInterface> = []
+
+  register(domainEventMessageHandler: DomainEventMessageHandlerInterface): void {
+    this.handlers.push(domainEventMessageHandler)
+  }
+
+  async publish(event: DomainEventInterface): Promise<void> {
+    await Promise.all(this.handlers.map((handler) => handler.handleMessage(event)))
+  }
+}

packages/domain-events-infra/src/Infra/DirectCall/DirectCallEventMessageHandler.ts

@@ -0,0 +1,32 @@
+import { Logger } from 'winston'
+
+import {
+  DomainEventHandlerInterface,
+  DomainEventInterface,
+  DomainEventMessageHandlerInterface,
+} from '@standardnotes/domain-events'
+
+export class DirectCallEventMessageHandler implements DomainEventMessageHandlerInterface {
+  constructor(private handlers: Map<string, DomainEventHandlerInterface>, private logger: Logger) {}
+
+  async handleMessage(messageOrEvent: string | DomainEventInterface): Promise<void> {
+    if (typeof messageOrEvent === 'string') {
+      throw new Error('DirectCallEventMessageHandler does not support string messages')
+    }
+
+    const handler = this.handlers.get(messageOrEvent.type)
+    if (!handler) {
+      this.logger.debug(`Event handler for event type ${messageOrEvent.type} does not exist`)
+
+      return
+    }
+
+    this.logger.debug(`Received event: ${messageOrEvent.type}`)
+
+    await handler.handle(messageOrEvent)
+  }
+
+  async handleError(error: Error): Promise<void> {
+    this.logger.error('Error occured while handling SQS message: %O', error)
+  }
+}

packages/domain-events-infra/src/Infra/index.ts

@@ -1,3 +1,6 @@
+export * from './DirectCall/DirectCallDomainEventPublisher'
+export * from './DirectCall/DirectCallEventMessageHandler'
+
 export * from './Redis/RedisDomainEventPublisher'
 export * from './Redis/RedisDomainEventSubscriber'
 export * from './Redis/RedisDomainEventSubscriberFactory'

packages/domain-events/src/Domain/Handler/DomainEventMessageHandlerInterface.ts

@@ -1,4 +1,6 @@
+import { DomainEventInterface } from '../Event/DomainEventInterface'
+
 export interface DomainEventMessageHandlerInterface {
-  handleMessage(message: string): Promise<void>
+  handleMessage(messageOrEvent: string | DomainEventInterface): Promise<void>
   handleError(error: Error): Promise<void>
 }

packages/home-server/bin/server.ts

@@ -2,6 +2,7 @@ import 'reflect-metadata'
 
 import { ControllerContainer, ServiceContainer } from '@standardnotes/domain-core'
 import { Service as ApiGatewayService, TYPES as ApiGatewayTYPES } from '@standardnotes/api-gateway'
+import { DirectCallDomainEventPublisher } from '@standardnotes/domain-events-infra'
 import { Service as AuthService } from '@standardnotes/auth-server'
 import { Container } from 'inversify'
 import { InversifyExpressServer } from 'inversify-express-utils'
@@ -17,9 +18,10 @@ import { Env } from '../src/Bootstrap/Env'
 const startServer = async (): Promise<void> => {
   const controllerContainer = new ControllerContainer()
   const serviceContainer = new ServiceContainer()
+  const directCallDomainEventPublisher = new DirectCallDomainEventPublisher()
 
   const apiGatewayService = new ApiGatewayService(serviceContainer, controllerContainer)
-  const authService = new AuthService(serviceContainer, controllerContainer)
+  const authService = new AuthService(serviceContainer, controllerContainer, directCallDomainEventPublisher)
 
   const container = Container.merge(
     (await apiGatewayService.getContainer()) as Container,

packages/home-server/package.json

@@ -21,6 +21,7 @@
     "@standardnotes/api-gateway": "workspace:^",
     "@standardnotes/auth-server": "workspace:^",
     "@standardnotes/domain-core": "workspace:^",
+    "@standardnotes/domain-events-infra": "workspace:^",
     "cors": "2.8.5",
     "dotenv": "^16.0.1",
     "express": "^4.18.2",

yarn.lock

@@ -4248,6 +4248,7 @@ __metadata:
     "@standardnotes/api-gateway": "workspace:^"
     "@standardnotes/auth-server": "workspace:^"
     "@standardnotes/domain-core": "workspace:^"
+    "@standardnotes/domain-events-infra": "workspace:^"
     "@types/cors": "npm:^2.8.9"
     "@types/express": "npm:^4.17.14"
     "@types/prettyjson": "npm:^0.0.30"