feat: consume is fully typed

Author: hmbanan666

README.md

@@ -30,29 +30,32 @@ npm install @nextorders/queue
 Create type definitions for your events:
 
 ```typescript
-import type { BaseEventMessage } from '@nextorders/queue'
+import type { BaseEventMap, BaseEventMessage, BaseEventMessageHandlerMap } from '@nextorders/queue'
 
 export enum Events {
   UserCreated = 'userCreated',
   EmailSent = 'emailSent',
 }
 
-export type EventMessage = UserCreated | EmailSent
+type EventMessage = UserCreated | EmailSent
+type EventMap = BaseEventMap<EventMessage>
 
-export interface UserCreated extends BaseEventMessage {
+export type EventHandlerMap = Partial<BaseEventMessageHandlerMap<EventMap>>
+
+type UserCreatedData = {
+  id: string
+  name: string
+  email: string
+}
+export interface UserCreated extends BaseEventMessage<UserCreatedData> {
   event: typeof Events.UserCreated
-  data: {
-    id: string
-    name: string
-    email: string
-  }
 }
 
-export interface EmailSent extends BaseEventMessage {
+type EmailSentData = {
+  email: string
+}
+export interface EmailSent extends BaseEventMessage<EmailSentData> {
   event: typeof Events.EmailSent
-  data: {
-    email: string
-  }
 }
 ```
 
@@ -97,14 +100,6 @@ import { Email, User } from './entities'
 class QueueRepository extends Repository {
   user: User = new User(this)
   email: Email = new Email(this)
-
-  // Override publish method with proper typing
-  publish<T extends EventMessage>(
-    event: T['event'],
-    data: T['data']
-  ): Promise<void> {
-    return super.publish(event, data)
-  }
 }
 
 export const repository = new QueueRepository()
@@ -125,7 +120,7 @@ await repository.connect('amqp://guest:guest@localhost:5672')
 Create and publish events from your services:
 
 ```typescript
-await repository.publish(Events.UserCreated, {
+await repository.publish<UserCreated>(Events.UserCreated, {
   id: newUser.id,
   name: newUser.name,
   email: newUser.email,
@@ -137,9 +132,14 @@ await repository.publish(Events.UserCreated, {
 Subscribe to events and handle them:
 
 ```typescript
-import type { UserCreated } from './types'
-import { repository } from './repository'
-import { Events } from './types'
+import type { EmailSent, EventHandlerMap, UserCreated } from '../repository/types'
+import { repository } from '../repository'
+import { Events } from '../repository/types'
+
+// Subscribe to Events and handle them
+repository.consume<EventHandlerMap>(repository.email.name, {
+  userCreated: handleUserCreated,
+})
 
 // Define event handlers
 async function handleUserCreated(data: UserCreated['data']): Promise<boolean> {
@@ -155,14 +155,11 @@ async function handleUserCreated(data: UserCreated['data']): Promise<boolean> {
 async function sendEmail(email: string): Promise<void> {
   console.warn('Sending email to:', email)
 
-  // Publish EmailSent event
-  await repository.publish(Events.EmailSent, { email })
+  // Publish Event for other services
+  await repository.publish<EmailSent>(Events.EmailSent, {
+    email,
+  })
 }
-
-// Subscribe to events
-await repository.consume(repository.email.name, {
-  [Events.UserCreated]: handleUserCreated,
-})
 ```
 
 ## 💁‍♂️ Example: Microservices Architecture

examples/microservices/repository/index.ts

@@ -1,15 +1,10 @@
-import type { EventMessage } from './types'
 import { Repository } from '@nextorders/queue'
 import { Email } from './entities/email'
 import { User } from './entities/user'
 
 class QueueRepository extends Repository {
   user = new User(this)
   email = new Email(this)
-
-  async publish<T extends EventMessage>(event: T['event'], data: T['data']) {
-    return super.publish(event, data)
-  }
 }
 
 export const repository = new QueueRepository()

examples/microservices/repository/types.ts

@@ -1,30 +1,28 @@
-import type { BaseEventMessage, Status } from '@nextorders/queue'
+import type { BaseEventMap, BaseEventMessage, BaseEventMessageHandlerMap } from '@nextorders/queue'
 
 // All possible events
 export enum Events {
   UserCreated = 'userCreated',
   EmailSent = 'emailSent',
 }
 
-export type EventMessage = UserCreated | EmailSent
+type EventMessage = UserCreated | EmailSent
+type EventMap = BaseEventMap<EventMessage>
 
-export type EventHandler = (msg: EventMessage) => Promise<Status>
-export type EventMessageHandler<T = EventMessage['data']> = (data: T) => Promise<boolean>
+export type EventHandlerMap = Partial<BaseEventMessageHandlerMap<EventMap>>
 
-export type EventHandlerMap = Record<EventMessage['event'], EventMessageHandler>
-
-export interface UserCreated extends BaseEventMessage {
+type UserCreatedData = {
+  id: string
+  name: string
+  email: string
+}
+export interface UserCreated extends BaseEventMessage<UserCreatedData> {
   event: typeof Events.UserCreated
-  data: {
-    id: string
-    name: string
-    email: string
-  }
 }
 
-export interface EmailSent extends BaseEventMessage {
+type EmailSentData = {
+  email: string
+}
+export interface EmailSent extends BaseEventMessage<EmailSentData> {
   event: typeof Events.EmailSent
-  data: {
-    email: string
-  }
 }

examples/microservices/service1/index.ts

@@ -1,3 +1,4 @@
+import type { UserCreated } from '../repository/types'
 import { repository } from '../repository'
 import { Events } from '../repository/types'
 
@@ -14,7 +15,7 @@ const newUser = {
 }
 
 // Publish Event for other services
-repository.publish(Events.UserCreated, {
+repository.publish<UserCreated>(Events.UserCreated, {
   id: newUser.id,
   name: newUser.name,
   email: newUser.email,

examples/microservices/service2/index.ts

@@ -1,18 +1,17 @@
-import type { EventHandlerMap, UserCreated } from '../repository/types'
+import type { EmailSent, EventHandlerMap, UserCreated } from '../repository/types'
 import { repository } from '../repository'
 import { Events } from '../repository/types'
 
 // Service 2: Email Service
 
-// Consume to Events
-repository.consume(repository.email.name, {
-  [Events.UserCreated]: handleUserCreated,
-} as EventHandlerMap)
+// Subscribe to Events and handle them
+repository.consume<EventHandlerMap>(repository.email.name, {
+  userCreated: handleUserCreated,
+})
 
 // Business logic
 async function handleUserCreated(data: UserCreated['data']): Promise<boolean> {
   try {
-    // Service logic: Send email
     await sendEmail(data.email)
     return true
   } catch (error) {
@@ -25,7 +24,7 @@ async function sendEmail(email: string) {
   console.warn('Sending email to', email)
 
   // Publish Event for other services
-  await repository.publish(Events.EmailSent, {
+  await repository.publish<EmailSent>(Events.EmailSent, {
     email,
   })
 }

packages/queue/src/repository.ts

@@ -1,5 +1,5 @@
 import type { Consumer, Publisher } from 'rabbitmq-client'
-import type { BaseEventHandlerMap, BaseEventMessage, Binding, ExchangesList, Queue, QueueRepository, Status } from './types'
+import type { BaseEventMap, BaseEventMessage, BaseEventMessageHandlerMap, Binding, ExchangesList, Queue, QueueRepository, Status } from './types'
 import { Connection, ConsumerStatus } from 'rabbitmq-client'
 import { COMMON_EXCHANGES } from './exchanges'
 
@@ -48,7 +48,7 @@ export class Repository implements QueueRepository {
     return this.#connection
   }
 
-  async publish<T extends BaseEventMessage>(event: T['event'], data: T['data']) {
+  async publish<T extends BaseEventMessage<any>>(event: T['event'], data: T['data']) {
     return this.publisher.send({
       exchange: this.exchanges.events.exchange,
       routingKey: event,
@@ -58,7 +58,7 @@ export class Repository implements QueueRepository {
     })
   }
 
-  async consume<T extends BaseEventMessage['data']>(queue: string, eventHandlers: BaseEventHandlerMap<T>): Promise<Consumer> {
+  async consume<T extends BaseEventMessageHandlerMap<BaseEventMap<any>>>(queue: string, eventHandlers: T): Promise<Consumer> {
     if (!queue) {
       throw new Error(`Queue "${queue}" not found`)
     }
@@ -72,7 +72,7 @@ export class Repository implements QueueRepository {
       qos: {
         prefetchCount: 1,
       },
-    }, async (msg) => this.handleEvent(eventHandlers, msg.body))
+    }, async (msg) => this.#handleEvent(eventHandlers, msg.body))
 
     consumer.on('error', (err) => {
       // Maybe the consumer was cancelled, or the connection was reset before a
@@ -83,7 +83,7 @@ export class Repository implements QueueRepository {
     return consumer
   }
 
-  async handleEvent<T extends BaseEventMessage['data']>(eventHandlers: BaseEventHandlerMap<T>, msg: BaseEventMessage<T>): Promise<Status> {
+  async #handleEvent(eventHandlers: BaseEventMessageHandlerMap<BaseEventMap<any>>, msg: BaseEventMessage<any>): Promise<Status> {
     const handler = eventHandlers[msg.event]
     if (!handler) {
       return this.ignore()

packages/queue/src/types.ts

@@ -15,9 +15,8 @@ export interface QueueRepository {
   exchanges: ExchangesList
   bindings: Binding[]
   queues: Queue[]
-  publish: <T extends BaseEventMessage>(event: T['event'], data: T['data']) => Promise<void>
-  consume: <T extends BaseEventMessage['data']>(queue: string, eventHandlers: BaseEventHandlerMap<T>) => Promise<Consumer>
-  handleEvent: <T extends BaseEventMessage['data']>(eventHandlers: BaseEventHandlerMap<T>, msg: BaseEventMessage<T>) => Promise<Status>
+  publish: <T extends BaseEventMessage<any>>(event: T['event'], data: T['data']) => Promise<void>
+  consume: <T extends BaseEventMessageHandlerMap<BaseEventMap<any>>>(queue: string, eventHandlers: T) => Promise<Consumer>
   connect: (connectionString: string) => Promise<void>
   checkHealth: () => boolean
   success: () => Status
@@ -32,25 +31,30 @@ export interface QueueEntity {
 }
 
 /**
- * @example export interface UserCreated extends BaseEventMessage {
- * event: 'userCreated'
- *   data: {
+ * @example export interface UserCreated extends BaseEventMessage<{
  *     id: string
  *     name: string
  *     email: string
- *   }
+ *   }> {
+ *   event: 'userCreated'
  * }
  */
-export interface BaseEventMessage<T = Record<string, unknown>> {
+export interface BaseEventMessage<T> {
   event: string
   data: T
 }
 
-export type BaseEventHandler<T = Record<string, unknown>> = (message: BaseEventMessage<T>) => Promise<ConsumerStatus>
+export type BaseEventMessageHandler<T> = (data: T) => Promise<boolean>
 
-export type BaseEventMessageHandler<T = Record<string, unknown>> = (data: T) => Promise<boolean>
+export type BaseEventMap<T extends Record<string, any>> = {
+  [K in T as K['event']]: K
+}
+
+export type BaseEventMessageHandlerMap<T extends BaseEventMap<any>> = {
+  [K in keyof T]: BaseEventMessageHandler<T[K]['data']>
+}
 
-export type BaseEventHandlerMap<T = Record<string, unknown>> = Record<string, BaseEventMessageHandler<T>>
+export type DistributePick<T, K extends keyof any> = T extends any ? Pick<T, Extract<keyof T, K>> : never
 
 export type Queue = MethodParams[Cmd.QueueDeclare]
 export type QueuesList = Record<string, Queue>