Event System

1import { createApp, getEventBus } from '@morojs/moro';
2
3const app = createApp();
4const events = getEventBus();
5
6// Emit events from route handlers
7app.post('/users', {
8  body: z.object({
9    email: z.string().email(),
10    name: z.string()
11  }),
12  handler: async ({ body }) => {
13    const user = await createUser(body);
14    
15    // Emit user creation event
16    await events.emit('user.created', {
17      user,
18      timestamp: new Date().toISOString()
19    });
20    
21    return user;
22  }
23});
24
25// Listen for events
26events.on('user.created', async (data) => {
27  console.log('New user created:', data.user.email);
28  
29  // Send welcome email
30  await sendWelcomeEmail(data.user.email);
31  
32  // Log to analytics
33  await trackUserSignup(data.user);
34});
35
36// One-time event listeners
37events.once('app.started', () => {
38  console.log('Application has started successfully');
39});
40
41// Remove event listeners
42const handler = (data) => console.log('User updated:', data);
43events.on('user.updated', handler);
44events.off('user.updated', handler);

1// decorators/events.ts
2import { EventBus } from '@morojs/moro/events';
3
4export function EventHandler() {
5  return function (target: any) {
6    // Register class as event handler
7    const eventBus = getEventBus();
8    const instance = new target();
9    
10    // Register event handlers functionally
11    const methods = Object.getOwnPropertyNames(target.prototype);
12    methods.forEach(methodName => {
13      const method = target.prototype[methodName];
14      if (method.eventName) {
15        eventBus.on(method.eventName, method.bind(instance));
16      }
17    });
18    
19    return target;
20  };
21}
22
23export function OnEvent(eventName: string) {
24  return function (target: any, propertyKey: string, descriptor: PropertyDescriptor) {
25    descriptor.value.eventName = eventName;
26    return descriptor;
27  };
28}
29
30// Functional Event Handling
31export function createUserEventHandlers(services: any) {
32  return {
33    async handleUserCreated(data: { user: User }) {
34      console.log('Handling user creation:', data.user.id);
35      
36      // Send welcome email
37      if (services.emailService) {
38        await services.emailService.sendWelcomeEmail(data.user);
39      }
40      
41      // Create user profile
42      if (services.profileService) {
43        await services.profileService.createUserProfile(data.user);
44      }
45      
46      // Track analytics
47      if (services.analyticsService) {
48        await services.analyticsService.trackSignup(data.user);
49      }
50    },
51    
52    async handleUserUpdated(data: { user: User, changes: Partial<User> }) {
53      console.log('User updated:', data.user.id, data.changes);
54      
55      // Invalidate cache
56      if (services.cacheService) {
57        await services.cacheService.invalidateUserCache(data.user.id);
58      }
59      
60      // Notify connected clients
61      if (services.notificationService) {
62        await services.notificationService.notifyUserUpdate(data.user);
63      }
64    },
65    
66    async handleUserDeleted(data: { userId: string }) {
67      console.log('User deleted:', data.userId);
68      
69      // Clean up user data
70      if (services.cleanupService) {
71        await services.cleanupService.cleanupUserData(data.userId);
72      }
73      
74      // Archive user information
75      if (services.archiveService) {
76        await services.archiveService.archiveUser(data.userId);
77      }
78    }
79  };
80}
81  
82  private async sendWelcomeEmail(user: User) {
83    // Email service integration
84  }
85  
86  private async createUserProfile(user: User) {
87    // Profile creation logic
88  }
89  
90  private async trackSignup(user: User) {
91    // Analytics tracking
92  }
93}

1// Event middleware for processing
2const eventMiddleware = {
3  // Logging middleware
4  logging: (eventName: string, data: any, next: Function) => {
5    console.log(`Event emitted: ${eventName}`, { data, timestamp: new Date() });
6    next();
7  },
8  
9  // Validation middleware
10  validation: (eventName: string, data: any, next: Function) => {
11    // Validate event data structure
12    if (!data || typeof data !== 'object') {
13      throw new Error(`Invalid event data for ${eventName}`);
14    }
15    next();
16  },
17  
18  // Rate limiting middleware
19  rateLimiting: (eventName: string, data: any, next: Function) => {
20    const key = `event_rate:${eventName}`;
21    const count = getEventCount(key);
22    
23    if (count > 100) { // Max 100 events per minute
24      throw new Error(`Rate limit exceeded for event ${eventName}`);
25    }
26    
27    incrementEventCount(key);
28    next();
29  },
30  
31  // Error handling middleware
32  errorHandling: (eventName: string, data: any, next: Function) => {
33    try {
34      next();
35    } catch (error) {
36      console.error(`Event processing error for ${eventName}:`, error);
37      
38      // Emit error event
39      events.emit('event.error', {
40        originalEvent: eventName,
41        error: error.message,
42        data
43      });
44    }
45  }
46};
47
48// Apply middleware to event bus
49const events = getEventBus();
50events.use(eventMiddleware.logging);
51events.use(eventMiddleware.validation);
52events.use(eventMiddleware.rateLimiting);
53events.use(eventMiddleware.errorHandling);
54
55// Queue-based event processing
56import { Queue } from '@morojs/moro/queue';
57
58const eventQueue = new Queue({
59  name: 'events',
60  concurrency: 5,
61  retries: 3,
62  backoff: 'exponential'
63});
64
65// Process events asynchronously
66events.on('user.created', async (data) => {
67  // Add to queue for async processing
68  await eventQueue.add('process-user-creation', data, {
69    delay: 0,
70    priority: 10
71  });
72});
73
74eventQueue.process('process-user-creation', async (job) => {
75  const { user } = job.data;
76  
77  // Heavy processing that shouldn't block request
78  await generateUserRecommendations(user);
79  await sendWelcomeEmailSequence(user);
80  await createUserAnalyticsProfile(user);
81});