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// Functional Event Handling
2export function createUserEventHandlers(services: any) {
3  return {
4    async handleUserCreated(data: { user: User }) {
5      console.log('Handling user creation:', data.user.id);
6      
7      // Send welcome email
8      if (services.emailService) {
9        await services.emailService.sendWelcomeEmail(data.user);
10      }
11      
12      // Create user profile
13      if (services.profileService) {
14        await services.profileService.createUserProfile(data.user);
15      }
16      
17      // Track analytics
18      if (services.analyticsService) {
19        await services.analyticsService.trackSignup(data.user);
20      }
21    },
22    
23    async handleUserUpdated(data: { user: User, changes: Partial<User> }) {
24      console.log('User updated:', data.user.id, data.changes);
25      
26      // Invalidate cache
27      if (services.cacheService) {
28        await services.cacheService.invalidateUserCache(data.user.id);
29      }
30      
31      // Notify connected clients
32      if (services.notificationService) {
33        await services.notificationService.notifyUserUpdate(data.user);
34      }
35    },
36    
37    async handleUserDeleted(data: { userId: string }) {
38      console.log('User deleted:', data.userId);
39      
40      // Clean up user data
41      if (services.cleanupService) {
42        await services.cleanupService.cleanupUserData(data.userId);
43      }
44      
45      // Archive user information
46      if (services.archiveService) {
47        await services.archiveService.archiveUser(data.userId);
48      }
49    }
50  };
51}

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});