| 1 | import type { InstanceWrapper } from '@nestjs/core/injector/instance-wrapper' |
| 2 | import type { EventStream } from './enum/event-stream.enum' |
| 3 | import type { EventTopic } from './enum/event-topic.enum' |
| 4 | import type { EventStreamHandlerMetadata } from './event-stream.decorator' |
| 5 | import { randomUUID } from 'node:crypto' |
| 6 | import { Injectable, Logger, OnModuleInit } from '@nestjs/common' |
| 7 | import { MetadataScanner, ModulesContainer } from '@nestjs/core' |
| 8 | import { EventStream as EventStreamEnum } from './enum/event-stream.enum' |
| 9 | import { ON_EVENT_STREAM_METADATA } from './event-stream.decorator' |
| 10 | import { EventStreamService } from './event-stream.service' |
| 11 | |
| 12 | @Injectable() |
| 13 | export class EventStreamExplorer implements OnModuleInit { |
| 14 | private static readonly registeredGroups = new Set<string>() |
| 15 | private readonly logger = new Logger(EventStreamExplorer.name) |
| 16 | private readonly metadataScanner = new MetadataScanner() |
| 17 | |
| 18 | constructor( |
| 19 | private readonly modulesContainer: ModulesContainer, |
| 20 | private readonly eventStreamService: EventStreamService, |
| 21 | ) {} |
| 22 | |
| 23 | onModuleInit() { |
| 24 | for (const moduleRef of this.modulesContainer.values()) { |
| 25 | this.scanWrappers(moduleRef.providers) |
| 26 | this.scanWrappers(moduleRef.controllers) |
| 27 | } |
| 28 | } |
| 29 | |
| 30 | private scanWrappers(wrappers: Map<unknown, InstanceWrapper>) { |
| 31 | for (const wrapper of wrappers.values()) { |
| 32 | if (!wrapper.instance || typeof wrapper.instance !== 'object') |
| 33 | continue |
| 34 | this.scanInstance(wrapper.instance) |
| 35 | } |
| 36 | } |
| 37 | |
| 38 | private scanInstance(instance: object) { |
| 39 | const prototype = Object.getPrototypeOf(instance) |
| 40 | if (!prototype) |
| 41 | return |
| 42 | |
| 43 | for (const methodName of this.metadataScanner.getAllMethodNames(prototype)) { |
| 44 | const method = prototype[methodName] |
| 45 | const metadata = Reflect.getMetadata(ON_EVENT_STREAM_METADATA, method) as EventStreamHandlerMetadata | undefined |
| 46 | if (!metadata) |
| 47 | continue |
| 48 | |
| 49 | const group = metadata.options.group ?? this.buildGroup(instance.constructor.name, methodName, metadata.topics) |
| 50 | if (EventStreamExplorer.registeredGroups.has(group)) |
| 51 | continue |
| 52 | |
| 53 | const streams = metadata.options.streams ?? this.inferStreams(metadata.topics) |
| 54 | if (!streams.length) { |
| 55 | throw new Error(`No event stream configured for ${instance.constructor.name}.${methodName}`) |
| 56 | } |
| 57 | this.eventStreamService.subscribe({ |
| 58 | group, |
| 59 | consumer: metadata.options.consumer ?? `${group}:${process.pid}:${randomUUID()}`, |
| 60 | streams, |
| 61 | topics: metadata.topics, |
| 62 | maxRetries: metadata.options.maxRetries, |
| 63 | pollInterval: metadata.options.pollInterval, |
| 64 | handler: async (envelope) => { |
| 65 | await (instance as Record<string, (...args: unknown[]) => unknown>)[methodName](envelope) |
| 66 | }, |
| 67 | }) |
| 68 | EventStreamExplorer.registeredGroups.add(group) |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | private buildGroup(className: string, methodName: string, topics: EventTopic[]) { |
| 73 | return `event-stream:${className}.${methodName}:${topics.join(',')}` |
| 74 | } |
| 75 | |
| 76 | private inferStreams(topics: EventTopic[]): EventStream[] { |
| 77 | const streams = new Set<EventStream>() |
| 78 | for (const topic of topics) { |
| 79 | if (topic.startsWith('channels.')) { |
| 80 | streams.add(EventStreamEnum.Channels) |
| 81 | continue |
| 82 | } |
| 83 | if (topic.startsWith('user.')) { |
| 84 | streams.add(EventStreamEnum.User) |
| 85 | continue |
| 86 | } |
| 87 | this.logger.error(`Unable to infer event stream for topic ${topic}`) |
| 88 | } |
| 89 | return Array.from(streams) |
| 90 | } |
| 91 | } |
| 92 |