# Event Bus Examples

Complete, working examples of event bus attachments for WildDuck.

## Table of Contents

1. [Metrics Collector](#1-metrics-collector)
2. [Audit Logger](#2-audit-logger)
3. [External Webhook Forwarder](#3-external-webhook-forwarder)

---

## 1. Metrics Collector

Track event counts, timing, and statistics for monitoring.

**Use case:** Monitor system activity, generate reports, create dashboards

**File:** `lib/event-bus/attachments/metrics-collector.js`

```javascript
'use strict';

const EventAttachment = require('../attachment');

/**
 * Metrics Collector Attachment
 *
 * Collects statistics about events:
 * - Event counts by type, category, source
 * - User activity tracking
 * - Timing information
 * - Rate calculations
 */
class MetricsCollector extends EventAttachment {
    constructor() {
        super('metrics-collector');

        // Event counters
        this.counters = {
            total: 0,
            byType: {},
            byCategory: {},
            bySource: {},
            byUser: {}
        };

        // Timing tracking
        this.startTime = Date.now();
        this.lastEventTime = null;

        // Rate tracking (events per minute)
        this.rateWindow = 60 * 1000; // 1 minute
        this.recentEvents = [];
    }

    /**
     * Handle event
     */
    async onEvent(event) {
        const now = Date.now();

        // Update counters
        this.counters.total++;
        this.counters.byType[event.type] = (this.counters.byType[event.type] || 0) + 1;
        this.counters.byCategory[event.category] = (this.counters.byCategory[event.category] || 0) + 1;
        this.counters.bySource[event.source] = (this.counters.bySource[event.source] || 0) + 1;

        if (event.user) {
            this.counters.byUser[event.user] = (this.counters.byUser[event.user] || 0) + 1;
        }

        // Track timing
        this.lastEventTime = now;

        // Track rate (keep last 1 minute of events)
        this.recentEvents.push(now);
        this.recentEvents = this.recentEvents.filter(time => now - time < this.rateWindow);
    }

    /**
     * Get current metrics
     */
    getMetrics() {
        const now = Date.now();
        const uptimeSeconds = Math.floor((now - this.startTime) / 1000);
        const eventsPerSecond = uptimeSeconds > 0 ? (this.counters.total / uptimeSeconds).toFixed(2) : 0;
        const eventsPerMinute = this.recentEvents.length;

        return {
            uptime: uptimeSeconds,
            total: this.counters.total,
            eventsPerSecond: parseFloat(eventsPerSecond),
            eventsPerMinute,
            byType: this.counters.byType,
            byCategory: this.counters.byCategory,
            bySource: this.counters.bySource,
            topUsers: this.getTopUsers(10),
            lastEvent: this.lastEventTime ? new Date(this.lastEventTime).toISOString() : null
        };
    }

    /**
     * Get top N active users
     */
    getTopUsers(limit = 10) {
        return Object.entries(this.counters.byUser)
            .sort((a, b) => b[1] - a[1])
            .slice(0, limit)
            .map(([userId, count]) => ({ userId, count }));
    }

    /**
     * Get metrics for specific category
     */
    getCategoryMetrics(category) {
        const categoryEvents = Object.entries(this.counters.byType)
            .filter(([type]) => type.startsWith(category + '.'))
            .reduce((acc, [type, count]) => {
                acc[type] = count;
                return acc;
            }, {});

        return {
            category,
            total: this.counters.byCategory[category] || 0,
            events: categoryEvents
        };
    }

    /**
     * Reset metrics
     */
    reset() {
        this.counters = {
            total: 0,
            byType: {},
            byCategory: {},
            bySource: {},
            byUser: {}
        };
        this.startTime = Date.now();
        this.lastEventTime = null;
        this.recentEvents = [];
    }

    /**
     * Cleanup on shutdown
     */
    async onStop() {
        console.log('Metrics Collector final stats:');
        console.log(JSON.stringify(this.getMetrics(), null, 2));
    }
}

module.exports = MetricsCollector;
```

### Usage

```javascript
// In worker.js or server.js
const eventBus = require('./lib/event-bus');
const MetricsCollector = require('./lib/event-bus/attachments/metrics-collector');

const metrics = new MetricsCollector();
eventBus.registerAttachment(metrics);

// Expose metrics via API endpoint
server.get('/metrics/events', (req, res) => {
    res.json(metrics.getMetrics());
});

// Get category-specific metrics
server.get('/metrics/events/:category', (req, res) => {
    const categoryMetrics = metrics.getCategoryMetrics(req.params.category);
    res.json(categoryMetrics);
});
```

### Example Output

```json
{
  "uptime": 3600,
  "total": 1523,
  "eventsPerSecond": 0.42,
  "eventsPerMinute": 25,
  "byType": {
    "message.added": 850,
    "message.flags.changed": 320,
    "mailbox.created": 15,
    "address.user.created": 12
  },
  "byCategory": {
    "message": 1170,
    "mailbox": 45,
    "address": 24
  },
  "bySource": {
    "lmtp": 850,
    "imap": 450,
    "api": 223
  },
  "topUsers": [
    { "userId": "507f1f77bcf86cd799439011", "count": 342 },
    { "userId": "507f1f77bcf86cd799439012", "count": 198 }
  ],
  "lastEvent": "2025-11-13T22:00:00.000Z"
}
```

---

## 2. Audit Logger

Write events to a log file for audit trail and compliance.

**Use case:** Compliance, security auditing, forensics

**File:** `lib/event-bus/attachments/audit-logger.js`

```javascript
'use strict';

const EventAttachment = require('../attachment');
const fs = require('fs');
const path = require('path');
const { promisify } = require('util');

const appendFile = promisify(fs.appendFile);
const mkdir = promisify(fs.mkdir);

/**
 * Audit Logger Attachment
 *
 * Writes events to a rotating log file for audit purposes.
 * Logs include timestamp, user, event type, and relevant data.
 */
class AuditLogger extends EventAttachment {
    constructor(options = {}) {
        super('audit-logger');

        this.logDir = options.logDir || path.join(__dirname, '../../../logs/audit');
        this.logPrefix = options.logPrefix || 'audit';
        this.rotateDaily = options.rotateDaily !== false; // Default true
        this.includeData = options.includeData !== false; // Default true
        this.filterCategories = options.filterCategories || null; // null = all

        // Track current log file
        this.currentLogFile = null;
        this.currentDate = null;

        // Ensure log directory exists
        this.ensureLogDirectory();
    }

    /**
     * Ensure log directory exists
     */
    async ensureLogDirectory() {
        try {
            await mkdir(this.logDir, { recursive: true });
        } catch (err) {
            if (err.code !== 'EEXIST') {
                console.error('Failed to create audit log directory:', err.message);
            }
        }
    }

    /**
     * Filter events by category if configured
     */
    shouldHandle(event) {
        if (!this.filterCategories) {
            return true;
        }

        return this.filterCategories.includes(event.category);
    }

    /**
     * Handle event - write to log file
     */
    async onEvent(event) {
        try {
            const logFile = this.getLogFile();
            const logEntry = this.formatLogEntry(event);

            await appendFile(logFile, logEntry + '\n', 'utf8');
        } catch (err) {
            console.error('Audit logger error:', err.message);
        }
    }

    /**
     * Get current log file path (with daily rotation)
     */
    getLogFile() {
        const now = new Date();
        const dateStr = now.toISOString().split('T')[0]; // YYYY-MM-DD

        if (this.rotateDaily && dateStr !== this.currentDate) {
            this.currentDate = dateStr;
            this.currentLogFile = path.join(this.logDir, `${this.logPrefix}-${dateStr}.log`);
        } else if (!this.currentLogFile) {
            this.currentLogFile = path.join(this.logDir, `${this.logPrefix}.log`);
        }

        return this.currentLogFile;
    }

    /**
     * Format log entry as JSON line
     */
    formatLogEntry(event) {
        const entry = {
            timestamp: event.timestamp.toISOString(),
            type: event.type,
            category: event.category,
            action: event.action,
            source: event.source,
            user: event.user || null,
            mailbox: event.mailbox || null,
            message: event.message || null
        };

        // Include data if configured
        if (this.includeData && event.data && Object.keys(event.data).length > 0) {
            entry.data = event.data;
        }

        return JSON.stringify(entry);
    }

    /**
     * Cleanup on shutdown
     */
    async onStop() {
        console.log('Audit logger stopped. Logs written to:', this.logDir);
    }
}

module.exports = AuditLogger;
```

### Usage

```javascript
// In worker.js or server.js
const eventBus = require('./lib/event-bus');
const AuditLogger = require('./lib/event-bus/attachments/audit-logger');

// Log all events
const auditLogger = new AuditLogger({
    logDir: '/var/log/wildduck/audit',
    rotateDaily: true,
    includeData: true
});

eventBus.registerAttachment(auditLogger);

// Or log only specific categories (e.g., security-relevant events)
const securityAudit = new AuditLogger({
    logDir: '/var/log/wildduck/security',
    logPrefix: 'security',
    filterCategories: ['address', 'filter', 'settings']
});

eventBus.registerAttachment(securityAudit);
```

### Example Log Output

**File:** `logs/audit/audit-2025-11-13.log`

```json
{"timestamp":"2025-11-13T22:00:00.123Z","type":"message.added","category":"message","action":"added","source":"lmtp","user":"507f1f77bcf86cd799439011","mailbox":"507f1f77bcf86cd799439012","message":"507f1f77bcf86cd799439013","data":{"size":1024,"uid":42}}
{"timestamp":"2025-11-13T22:00:01.456Z","type":"mailbox.created","category":"mailbox","action":"created","source":"api","user":"507f1f77bcf86cd799439011","mailbox":"507f1f77bcf86cd799439014","message":null,"data":{"path":"Sent"}}
{"timestamp":"2025-11-13T22:00:02.789Z","type":"filter.created","category":"filter","action":"created","source":"api","user":"507f1f77bcf86cd799439011","mailbox":null,"message":null,"data":{"filterName":"Spam Filter"}}
```

---

## 3. External Webhook Forwarder

Forward events to external HTTP endpoints for integration with third-party systems.

**Use case:** Integrate with Slack, webhooks, monitoring systems, custom applications

**File:** `lib/event-bus/attachments/webhook-forwarder.js`

```javascript
'use strict';

const EventAttachment = require('../attachment');
const fetch = require('node-fetch');

/**
 * Webhook Forwarder Attachment
 *
 * Forwards events to external HTTP endpoints.
 * Supports retry logic, filtering, and batching.
 */
class WebhookForwarder extends EventAttachment {
    constructor(options = {}) {
        super('webhook-forwarder');

        this.webhookUrl = options.webhookUrl;
        this.secret = options.secret || null; // Optional HMAC secret
        this.filterTypes = options.filterTypes || null; // null = all
        this.batchSize = options.batchSize || 1; // 1 = send immediately
        this.batchTimeout = options.batchTimeout || 5000; // 5 seconds
        this.retries = options.retries || 3;
        this.timeout = options.timeout || 10000; // 10 seconds

        // Validation
        if (!this.webhookUrl) {
            throw new Error('webhookUrl is required');
        }

        // Batching state
        this.eventBatch = [];
        this.batchTimer = null;
    }

    /**
     * Filter events by type if configured
     */
    shouldHandle(event) {
        if (!this.filterTypes) {
            return true;
        }

        return this.filterTypes.includes(event.type);
    }

    /**
     * Handle event - add to batch or send immediately
     */
    async onEvent(event) {
        if (this.batchSize === 1) {
            // Send immediately
            await this.sendEvents([event]);
        } else {
            // Add to batch
            this.eventBatch.push(event);

            // Send if batch is full
            if (this.eventBatch.length >= this.batchSize) {
                await this.flushBatch();
            } else {
                // Set timer to flush batch
                this.resetBatchTimer();
            }
        }
    }

    /**
     * Reset batch timer
     */
    resetBatchTimer() {
        if (this.batchTimer) {
            clearTimeout(this.batchTimer);
        }

        this.batchTimer = setTimeout(() => {
            this.flushBatch().catch(err => {
                console.error('Webhook batch flush error:', err.message);
            });
        }, this.batchTimeout);
    }

    /**
     * Flush current batch
     */
    async flushBatch() {
        if (this.batchTimer) {
            clearTimeout(this.batchTimer);
            this.batchTimer = null;
        }

        if (this.eventBatch.length === 0) {
            return;
        }

        const events = this.eventBatch.splice(0);
        await this.sendEvents(events);
    }

    /**
     * Send events to webhook with retry
     */
    async sendEvents(events) {
        const payload = {
            timestamp: new Date().toISOString(),
            count: events.length,
            events: events.map(e => this.formatEvent(e))
        };

        // Add signature if secret is configured
        if (this.secret) {
            const crypto = require('crypto');
            const signature = crypto
                .createHmac('sha256', this.secret)
                .update(JSON.stringify(payload))
                .digest('hex');
            payload.signature = signature;
        }

        let lastError = null;

        // Retry loop
        for (let attempt = 0; attempt <= this.retries; attempt++) {
            try {
                const controller = new AbortController();
                const timeoutId = setTimeout(() => controller.abort(), this.timeout);

                const response = await fetch(this.webhookUrl, {
                    method: 'POST',
                    headers: {
                        'Content-Type': 'application/json',
                        'User-Agent': 'WildDuck-EventBus/1.0'
                    },
                    body: JSON.stringify(payload),
                    signal: controller.signal
                });

                clearTimeout(timeoutId);

                if (response.ok) {
                    return; // Success
                }

                lastError = new Error(`HTTP ${response.status}: ${response.statusText}`);
            } catch (err) {
                lastError = err;

                if (err.name === 'AbortError') {
                    lastError = new Error('Request timeout');
                }
            }

            // Wait before retry (exponential backoff)
            if (attempt < this.retries) {
                await new Promise(resolve => setTimeout(resolve, Math.pow(2, attempt) * 1000));
            }
        }

        console.error(`Webhook forward failed after ${this.retries + 1} attempts:`, lastError.message);
    }

    /**
     * Format event for webhook payload
     */
    formatEvent(event) {
        return {
            type: event.type,
            category: event.category,
            action: event.action,
            timestamp: event.timestamp.toISOString(),
            source: event.source,
            user: event.user || null,
            mailbox: event.mailbox || null,
            message: event.message || null,
            data: event.data || {}
        };
    }

    /**
     * Cleanup on shutdown - flush pending batch
     */
    async onStop() {
        console.log('Webhook forwarder stopping...');
        await this.flushBatch();
        console.log('Webhook forwarder stopped');
    }
}

module.exports = WebhookForwarder;
```

### Usage

```javascript
// In worker.js or server.js
const eventBus = require('./lib/event-bus');
const WebhookForwarder = require('./lib/event-bus/attachments/webhook-forwarder');

// Send all events to external webhook
const webhook = new WebhookForwarder({
    webhookUrl: 'https://example.com/webhooks/wildduck',
    secret: 'your-webhook-secret',
    batchSize: 10,        // Send in batches of 10
    batchTimeout: 5000,   // Or after 5 seconds
    retries: 3            // Retry 3 times on failure
});

eventBus.registerAttachment(webhook);

// Or only send specific event types
const messageWebhook = new WebhookForwarder({
    webhookUrl: 'https://example.com/webhooks/messages',
    filterTypes: ['message.added', 'message.deleted'],
    batchSize: 1 // Send immediately
});

eventBus.registerAttachment(messageWebhook);
```

### Example Webhook Payload

```json
{
  "timestamp": "2025-11-13T22:00:00.000Z",
  "count": 2,
  "signature": "a1b2c3d4e5f6...",
  "events": [
    {
      "type": "message.added",
      "category": "message",
      "action": "added",
      "timestamp": "2025-11-13T22:00:00.123Z",
      "source": "lmtp",
      "user": "507f1f77bcf86cd799439011",
      "mailbox": "507f1f77bcf86cd799439012",
      "message": "507f1f77bcf86cd799439013",
      "data": {
        "size": 1024,
        "uid": 42
      }
    },
    {
      "type": "mailbox.created",
      "category": "mailbox",
      "action": "created",
      "timestamp": "2025-11-13T22:00:01.456Z",
      "source": "api",
      "user": "507f1f77bcf86cd799439011",
      "mailbox": "507f1f77bcf86cd799439014",
      "message": null,
      "data": {
        "path": "Sent"
      }
    }
  ]
}
```

---

## Slack Integration Example

Combine webhook forwarder with Slack formatting:

```javascript
const EventAttachment = require('./lib/event-bus/attachment');
const fetch = require('node-fetch');

class SlackNotifier extends EventAttachment {
    constructor(slackWebhookUrl) {
        super('slack-notifier');
        this.slackWebhookUrl = slackWebhookUrl;
    }

    shouldHandle(event) {
        // Only notify for important events
        return [
            'marked.spam',
            'filter.created',
            'mailbox.deleted'
        ].includes(event.type);
    }

    async onEvent(event) {
        const message = this.formatSlackMessage(event);

        await fetch(this.slackWebhookUrl, {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify(message)
        });
    }

    formatSlackMessage(event) {
        let color = '#36a64f'; // green
        if (event.type === 'marked.spam') {
            color = '#ff0000'; // red
        }

        return {
            attachments: [{
                color,
                title: `WildDuck Event: ${event.type}`,
                fields: [
                    { title: 'Category', value: event.category, short: true },
                    { title: 'Source', value: event.source, short: true },
                    { title: 'User', value: event.user || 'N/A', short: true },
                    { title: 'Time', value: event.timestamp.toISOString(), short: true }
                ],
                footer: 'WildDuck Event Bus',
                ts: Math.floor(event.timestamp.getTime() / 1000)
            }]
        };
    }
}

// Usage
const slack = new SlackNotifier('https://hooks.slack.com/services/YOUR/WEBHOOK/URL');
eventBus.registerAttachment(slack);
```

---

## Testing Your Attachment

Example test for custom attachment:

```javascript
const { expect } = require('chai');
const eventBus = require('../../lib/event-bus');
const MyAttachment = require('../../lib/event-bus/attachments/my-attachment');

describe('MyAttachment', () => {
    let attachment;

    beforeEach(() => {
        attachment = new MyAttachment();
        eventBus.registerAttachment(attachment);
    });

    afterEach(() => {
        eventBus.unregisterAttachment('my-attachment');
    });

    it('should handle events', async () => {
        eventBus.emit({
            ev: 'message.added',
            user: '123',
            time: Date.now(),
            source: 'api'
        });

        // Wait for async event
        await new Promise(resolve => setTimeout(resolve, 50));

        // Verify attachment received event
        expect(attachment.eventCount).to.equal(1);
    });
});
```

---

## See Also

- [Event Bus User Guide](./event-bus.md)
- [Event Bus API Reference](./event-bus-api.md)
- [Testing Documentation](../TESTING.md)