doudou/telegram-management-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Telegram Management System
Some checks failed
Deploy / deploy (push) Has been cancelled
Details

Browse Source 
Full-stack web application for Telegram management
- Frontend: Vue 3 + Vben Admin
- Backend: NestJS
- Features: User management, group broadcast, statistics

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
你的用户名
2025-11-04 15:37:50 +08:00
commit 237c7802e5
3674 changed files with 525172 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
marketing-agent/services/analytics/Dockerfile Normal file
View File

@@ -0,0 +1,34 @@
FROM node:20-alpine
# Install build dependencies
RUN apk add --no-cache python3 make g++
# Create app directory
WORKDIR /app
# Copy package files
COPY package*.json ./
# Install dependencies
RUN npm install --production
# Copy app source
COPY . .
# Create non-root user
RUN addgroup -g 1001 -S nodejs
RUN adduser -S nodejs -u 1001
# Create logs directory with proper permissions
RUN mkdir -p logs && chown -R nodejs:nodejs logs
USER nodejs
# Expose port
EXPOSE 3005
# Health check
HEALTHCHECK --interval=30s --timeout=3s --start-period=40s \
CMD node healthcheck.js || exit 1
# Start the service
CMD ["node", "src/app.js"]

331
marketing-agent/services/analytics/README.md Normal file
View File

@@ -0,0 +1,331 @@
# Analytics Service
Real-time analytics and reporting service for the Telegram Marketing Intelligence Agent system.
## Overview
The Analytics service provides comprehensive event tracking, metrics processing, real-time analytics, report generation, and alert management capabilities.
## Features
### Event Tracking
- Real-time event ingestion
- Batch event processing
- Event validation and enrichment
- Multi-dimensional event storage
### Metrics Processing
- Custom metric definitions
- Real-time metric calculation
- Aggregation and rollups
- Time-series data management
### Report Generation
- Scheduled report generation
- Custom report templates
- Multiple export formats (PDF, Excel, CSV)
- Report distribution
### Real-time Analytics
- WebSocket streaming
- Live dashboards
- Real-time alerts
- Performance monitoring
### Alert Management
- Threshold-based alerts
- Anomaly detection
- Multi-channel notifications
- Alert history tracking
## Architecture
```
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ API Gateway │────▶│ Analytics Service │────▶│ Elasticsearch │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│ │
├───────────────────────────┤
▼ ▼
┌─────────────┐ ┌──────────────┐
│ MongoDB │ │ ClickHouse │
└─────────────┘ └──────────────┘
```
## API Endpoints
### Events
- `POST /api/events` - Track single event
- `POST /api/events/batch` - Track multiple events
- `GET /api/events/search` - Search events
### Metrics
- `GET /api/metrics` - List all metrics
- `POST /api/metrics` - Create custom metric
- `GET /api/metrics/:id` - Get metric details
- `GET /api/metrics/:id/data` - Get metric data
### Reports
- `GET /api/reports` - List reports
- `POST /api/reports` - Generate report
- `GET /api/reports/:id` - Get report details
- `GET /api/reports/:id/download` - Download report
### Alerts
- `GET /api/alerts` - List alerts
- `POST /api/alerts` - Create alert
- `PUT /api/alerts/:id` - Update alert
- `DELETE /api/alerts/:id` - Delete alert
- `GET /api/alerts/:id/history` - Get alert history
### Real-time
- `WS /ws/analytics` - Real-time analytics stream
- `WS /ws/metrics/:id` - Metric-specific stream
## Data Models
### Event Schema
```javascript
{
eventId: String,
accountId: String,
sessionId: String,
eventType: String,
eventName: String,
timestamp: Date,
properties: Object,
context: {
ip: String,
userAgent: String,
locale: String
}
}
```
### Metric Schema
```javascript
{
metricId: String,
accountId: String,
name: String,
type: String, // 'counter', 'gauge', 'histogram'
unit: String,
formula: String,
aggregation: String,
filters: Array,
dimensions: Array
}
```
### Report Schema
```javascript
{
reportId: String,
accountId: String,
name: String,
template: String,
schedule: String,
parameters: Object,
recipients: Array,
format: String
}
```
### Alert Schema
```javascript
{
alertId: String,
accountId: String,
name: String,
metric: String,
condition: Object,
threshold: Number,
severity: String,
channels: Array,
cooldown: Number
}
```
## Configuration
### Environment Variables
- `PORT` - Service port (default: 3004)
- `MONGODB_URI` - MongoDB connection string
- `ELASTICSEARCH_NODE` - Elasticsearch URL
- `CLICKHOUSE_HOST` - ClickHouse host
- `REDIS_URL` - Redis connection URL
- `RABBITMQ_URL` - RabbitMQ connection URL
### Storage Configuration
- `REPORTS_DIR` - Report storage directory
- `EXPORTS_DIR` - Export storage directory
- `RETENTION_DAYS` - Data retention period
### Processing Configuration
- `BATCH_SIZE` - Event batch size
- `PROCESSING_INTERVAL` - Processing interval (ms)
- `STREAM_BUFFER_SIZE` - Real-time buffer size
## Deployment
### Docker
```bash
docker build -t analytics-service .
docker run -p 3004:3004 --env-file .env analytics-service
```
### Kubernetes
```bash
kubectl apply -f k8s/deployment.yaml
kubectl apply -f k8s/service.yaml
```
## Usage Examples
### Track Event
```javascript
const event = {
eventType: 'message',
eventName: 'message_sent',
properties: {
campaignId: '123',
groupId: '456',
messageType: 'text',
charactersCount: 150
}
};
await analyticsClient.trackEvent(event);
```
### Create Custom Metric
```javascript
const metric = {
name: 'Message Delivery Rate',
type: 'gauge',
formula: '(delivered / sent) * 100',
unit: 'percentage',
aggregation: 'avg',
dimensions: ['campaignId', 'groupId']
};
await analyticsClient.createMetric(metric);
```
### Generate Report
```javascript
const report = {
template: 'campaign_performance',
parameters: {
campaignId: '123',
dateRange: {
start: '2024-01-01',
end: '2024-01-31'
}
},
format: 'pdf'
};
const result = await analyticsClient.generateReport(report);
```
### Create Alert
```javascript
const alert = {
name: 'High Error Rate',
metric: 'error_rate',
condition: 'greater_than',
threshold: 5,
severity: 'critical',
channels: ['email', 'slack']
};
await analyticsClient.createAlert(alert);
```
## Monitoring
### Health Check
```bash
curl http://localhost:3004/health
```
### Metrics
- Prometheus metrics available at `/metrics`
- Grafana dashboards included in `/dashboards`
### Logging
- Structured JSON logging
- Log levels: error, warn, info, debug
- Logs shipped to Elasticsearch
## Development
### Setup
```bash
npm install
cp .env.example .env
npm run dev
```
### Testing
```bash
npm test
npm run test:integration
npm run test:e2e
```
### Code Quality
```bash
npm run lint
npm run format
npm run type-check
```
## Best Practices
1. **Event Tracking**
- Use consistent event naming
- Include relevant context
- Batch events when possible
- Validate event schema
2. **Metrics Design**
- Keep metrics simple and focused
- Use appropriate aggregations
- Consider cardinality
- Plan for scale
3. **Report Generation**
- Schedule during off-peak hours
- Use caching for common reports
- Optimize queries
- Monitor generation time
4. **Alert Configuration**
- Set appropriate thresholds
- Use severity levels wisely
- Configure cooldown periods
- Test alert channels
## Security
- JWT authentication for API access
- Field-level encryption for sensitive data
- Rate limiting per account
- Audit logging for all operations
- RBAC for multi-tenant access
## Performance
- Event ingestion: 10K events/second
- Query response: <100ms p99
- Report generation: <30s for standard reports
- Real-time latency: <50ms
## Support
For issues and questions:
- Check the documentation
- Review common issues in FAQ
- Contact the development team

42
marketing-agent/services/analytics/package.json Normal file
View File

@@ -0,0 +1,42 @@
{
"name": "@marketing-agent/analytics",
"version": "1.0.0",
"description": "Analytics service for marketing intelligence tracking and reporting",
"type": "module",
"main": "src/index.js",
"scripts": {
"start": "node src/index.js",
"dev": "nodemon src/index.js",
"test": "jest"
},
"dependencies": {
"@hapi/hapi": "^21.3.2",
"@hapi/joi": "^17.1.1",
"@hapi/boom": "^10.0.1",
"@elastic/elasticsearch": "^8.11.0",
"@clickhouse/client": "^0.2.6",
"mongoose": "^8.0.3",
"redis": "^4.6.12",
"ioredis": "^5.3.2",
"dotenv": "^16.3.1",
"winston": "^3.11.0",
"axios": "^1.6.5",
"node-schedule": "^2.1.1",
"mathjs": "^12.2.1",
"simple-statistics": "^7.8.3",
"date-fns": "^3.1.0",
"lodash": "^4.17.21",
"uuid": "^9.0.0",
"exceljs": "^4.4.0",
"pdfkit": "^0.14.0"
},
"devDependencies": {
"nodemon": "^3.0.2",
"jest": "^29.7.0",
"@babel/preset-env": "^7.23.7",
"eslint": "^8.56.0"
},
"engines": {
"node": ">=18.0.0"
}
}

1
marketing-agent/services/analytics/src/app.js Normal file
View File

@@ -0,0 +1 @@
import './index.js';

187
marketing-agent/services/analytics/src/config/clickhouse.js Normal file
View File

@@ -0,0 +1,187 @@
import { createClient } from '@clickhouse/client';
import { logger } from '../utils/logger.js';
export class ClickHouseClient {
constructor() {
this.client = null;
}
static getInstance() {
if (!ClickHouseClient.instance) {
ClickHouseClient.instance = new ClickHouseClient();
}
return ClickHouseClient.instance;
}
async connect() {
const config = {
host: process.env.CLICKHOUSE_HOST || 'http://localhost:8123',
username: process.env.CLICKHOUSE_USER || 'default',
password: process.env.CLICKHOUSE_PASSWORD || '',
database: process.env.CLICKHOUSE_DATABASE || 'analytics',
request_timeout: 30000,
compression: {
request: true,
response: true
}
};
try {
this.client = createClient(config);
// Test connection and create database if needed
await this.initializeDatabase();
logger.info('ClickHouse connected successfully');
return this.client;
} catch (error) {
logger.error('Failed to connect to ClickHouse:', error);
throw error;
}
}
async initializeDatabase() {
try {
// Create database if it doesn't exist
await this.client.exec({
query: `CREATE DATABASE IF NOT EXISTS analytics`
});
// Create tables
await this.createTables();
} catch (error) {
logger.error('Failed to initialize ClickHouse database:', error);
throw error;
}
}
async createTables() {
const tables = [
{
name: 'events',
query: `
CREATE TABLE IF NOT EXISTS analytics.events (
id String,
timestamp DateTime,
type String,
accountId String,
userId Nullable(String),
sessionId Nullable(String),
action String,
target Nullable(String),
value Nullable(Float64),
metadata String,
properties String,
date Date DEFAULT toDate(timestamp)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (accountId, type, timestamp)
TTL date + INTERVAL 1 YEAR DELETE
SETTINGS index_granularity = 8192
`
},
{
name: 'metrics',
query: `
CREATE TABLE IF NOT EXISTS analytics.metrics (
metricId String,
timestamp DateTime,
dimensions String,
value Float64,
aggregations String,
date Date DEFAULT toDate(timestamp)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (metricId, timestamp)
TTL date + INTERVAL 90 DAY DELETE
SETTINGS index_granularity = 8192
`
},
{
name: 'user_sessions',
query: `
CREATE TABLE IF NOT EXISTS analytics.user_sessions (
sessionId String,
userId String,
accountId String,
startTime DateTime,
endTime Nullable(DateTime),
duration Nullable(UInt32),
eventCount UInt32,
properties String,
date Date DEFAULT toDate(startTime)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (accountId, userId, startTime)
TTL date + INTERVAL 180 DAY DELETE
SETTINGS index_granularity = 8192
`
}
];
for (const table of tables) {
try {
await this.client.exec({ query: table.query });
logger.info(`Created ClickHouse table: ${table.name}`);
} catch (error) {
logger.error(`Failed to create table ${table.name}:`, error);
}
}
}
async checkHealth() {
try {
const result = await this.client.query({
query: 'SELECT 1',
format: 'JSONEachRow'
});
return true;
} catch (error) {
logger.error('ClickHouse health check failed:', error);
return false;
}
}
async disconnect() {
if (this.client) {
await this.client.close();
logger.info('ClickHouse connection closed');
}
}
// Helper methods
async query(params) {
try {
const result = await this.client.query({
...params,
format: params.format || 'JSONEachRow'
});
return result;
} catch (error) {
logger.error('ClickHouse query error:', error);
throw error;
}
}
async insert(params) {
try {
return await this.client.insert(params);
} catch (error) {
logger.error('ClickHouse insert error:', error);
throw error;
}
}
async exec(params) {
try {
return await this.client.exec(params);
} catch (error) {
logger.error('ClickHouse exec error:', error);
throw error;
}
}
}

49
marketing-agent/services/analytics/src/config/database.js Normal file
View File

@@ -0,0 +1,49 @@
import mongoose from 'mongoose';
import { logger } from '../utils/logger.js';
export const connectDatabase = async () => {
const mongoUri = process.env.MONGODB_URI || 'mongodb://localhost:27017/analytics';
const options = {
useNewUrlParser: true,
useUnifiedTopology: true,
autoIndex: true,
maxPoolSize: 10,
serverSelectionTimeoutMS: 5000,
socketTimeoutMS: 45000,
family: 4
};
try {
await mongoose.connect(mongoUri, options);
logger.info('MongoDB connected successfully');
// Handle connection events
mongoose.connection.on('error', (err) => {
logger.error('MongoDB connection error:', err);
});
mongoose.connection.on('disconnected', () => {
logger.warn('MongoDB disconnected');
});
mongoose.connection.on('reconnected', () => {
logger.info('MongoDB reconnected');
});
return mongoose.connection;
} catch (error) {
logger.error('Failed to connect to MongoDB:', error);
throw error;
}
};
export const disconnectDatabase = async () => {
try {
await mongoose.disconnect();
logger.info('MongoDB disconnected successfully');
} catch (error) {
logger.error('Error disconnecting from MongoDB:', error);
throw error;
}
};

162
marketing-agent/services/analytics/src/config/elasticsearch.js Normal file
View File

@@ -0,0 +1,162 @@
import { Client } from '@elastic/elasticsearch';
import { logger } from '../utils/logger.js';
export class ElasticsearchClient {
constructor() {
this.client = null;
}
static getInstance() {
if (!ElasticsearchClient.instance) {
ElasticsearchClient.instance = new ElasticsearchClient();
}
return ElasticsearchClient.instance;
}
async connect() {
const config = {
node: process.env.ELASTICSEARCH_NODE || 'http://localhost:9200',
auth: {
username: process.env.ELASTICSEARCH_USERNAME || 'elastic',
password: process.env.ELASTICSEARCH_PASSWORD || 'changeme'
},
maxRetries: 5,
requestTimeout: 60000,
sniffOnStart: true
};
try {
this.client = new Client(config);
// Test connection
const info = await this.client.info();
logger.info(`Elasticsearch connected: ${info.name} (${info.version.number})`);
// Create indexes if they don't exist
await this.createIndexes();
return this.client;
} catch (error) {
logger.error('Failed to connect to Elasticsearch:', error);
throw error;
}
}
async createIndexes() {
const indexes = [
{
name: 'events',
mappings: {
properties: {
id: { type: 'keyword' },
type: { type: 'keyword' },
accountId: { type: 'keyword' },
userId: { type: 'keyword' },
sessionId: { type: 'keyword' },
action: { type: 'keyword' },
target: { type: 'keyword' },
value: { type: 'double' },
timestamp: { type: 'date' },
metadata: { type: 'object', enabled: false },
properties: { type: 'object', enabled: false }
}
},
settings: {
number_of_shards: 3,
number_of_replicas: 1,
'index.lifecycle.name': 'events-policy',
'index.lifecycle.rollover_alias': 'events'
}
},
{
name: 'metrics',
mappings: {
properties: {
metricId: { type: 'keyword' },
timestamp: { type: 'date' },
value: { type: 'double' },
dimensions: { type: 'object' },
aggregations: { type: 'object' }
}
},
settings: {
number_of_shards: 2,
number_of_replicas: 1
}
}
];
for (const index of indexes) {
try {
const exists = await this.client.indices.exists({ index: index.name });
if (!exists) {
await this.client.indices.create({
index: index.name,
body: {
mappings: index.mappings,
settings: index.settings
}
});
logger.info(`Created Elasticsearch index: ${index.name}`);
}
} catch (error) {
logger.error(`Failed to create index ${index.name}:`, error);
}
}
}
async checkHealth() {
try {
const health = await this.client.cluster.health();
return health.status === 'green' || health.status === 'yellow';
} catch (error) {
logger.error('Elasticsearch health check failed:', error);
return false;
}
}
async disconnect() {
if (this.client) {
await this.client.close();
logger.info('Elasticsearch connection closed');
}
}
// Helper methods
async search(params) {
try {
return await this.client.search(params);
} catch (error) {
logger.error('Elasticsearch search error:', error);
throw error;
}
}
async bulk(params) {
try {
return await this.client.bulk(params);
} catch (error) {
logger.error('Elasticsearch bulk error:', error);
throw error;
}
}
async index(params) {
try {
return await this.client.index(params);
} catch (error) {
logger.error('Elasticsearch index error:', error);
throw error;
}
}
async deleteByQuery(params) {
try {
return await this.client.deleteByQuery(params);
} catch (error) {
logger.error('Elasticsearch delete by query error:', error);
throw error;
}
}
}

172
marketing-agent/services/analytics/src/config/redis.js Normal file
View File

@@ -0,0 +1,172 @@
import Redis from 'ioredis';
import { logger } from '../utils/logger.js';
export class RedisClient {
constructor() {
this.client = null;
}
static getInstance() {
if (!RedisClient.instance) {
RedisClient.instance = new RedisClient();
}
return RedisClient.instance;
}
async connect() {
const config = {
host: process.env.REDIS_HOST || 'localhost',
port: process.env.REDIS_PORT || 6379,
password: process.env.REDIS_PASSWORD || undefined,
db: parseInt(process.env.REDIS_DB) || 3, // Different DB for analytics
retryStrategy: (times) => {
const delay = Math.min(times * 50, 2000);
return delay;
},
enableOfflineQueue: true
};
try {
this.client = new Redis(config);
this.client.on('connect', () => {
logger.info('Redis connection established');
});
this.client.on('error', (err) => {
logger.error('Redis error:', err);
});
this.client.on('close', () => {
logger.warn('Redis connection closed');
});
// Wait for connection
await this.client.ping();
return this.client;
} catch (error) {
logger.error('Failed to connect to Redis:', error);
throw error;
}
}
async checkHealth() {
try {
const result = await this.client.ping();
return result === 'PONG';
} catch (error) {
logger.error('Redis health check failed:', error);
return false;
}
}
async disconnect() {
if (this.client) {
await this.client.quit();
logger.info('Redis connection closed');
}
}
// Cache methods with JSON serialization
async setWithExpiry(key, value, ttl) {
return await this.client.setex(key, ttl, JSON.stringify(value));
}
async get(key) {
const value = await this.client.get(key);
return value ? JSON.parse(value) : null;
}
async del(key) {
return await this.client.del(key);
}
async exists(key) {
return await this.client.exists(key);
}
// Hash operations
async hset(key, field, value) {
return await this.client.hset(key, field, JSON.stringify(value));
}
async hget(key, field) {
const value = await this.client.hget(key, field);
return value ? JSON.parse(value) : null;
}
async hdel(key, field) {
return await this.client.hdel(key, field);
}
async hgetall(key) {
const data = await this.client.hgetall(key);
const result = {};
for (const [field, value] of Object.entries(data)) {
try {
result[field] = JSON.parse(value);
} catch (e) {
result[field] = value;
}
}
return result;
}
// List operations
async lpush(key, value) {
return await this.client.lpush(key, JSON.stringify(value));
}
async rpush(key, value) {
return await this.client.rpush(key, JSON.stringify(value));
}
async lrange(key, start, stop) {
const items = await this.client.lrange(key, start, stop);
return items.map(item => {
try {
return JSON.parse(item);
} catch (e) {
return item;
}
});
}
async ltrim(key, start, stop) {
return await this.client.ltrim(key, start, stop);
}
// Set operations
async sadd(key, member) {
return await this.client.sadd(key, JSON.stringify(member));
}
async srem(key, member) {
return await this.client.srem(key, JSON.stringify(member));
}
async smembers(key) {
const members = await this.client.smembers(key);
return members.map(member => {
try {
return JSON.parse(member);
} catch (e) {
return member;
}
});
}
async sismember(key, member) {
return await this.client.sismember(key, JSON.stringify(member));
}
// Expiry operations
async expire(key, seconds) {
return await this.client.expire(key, seconds);
}
async ttl(key) {
return await this.client.ttl(key);
}
}

130
marketing-agent/services/analytics/src/index.js Normal file
View File

@@ -0,0 +1,130 @@
import 'dotenv/config';
import Hapi from '@hapi/hapi';
import { logger } from './utils/logger.js';
import { connectDatabase } from './config/database.js';
import { RedisClient } from './config/redis.js';
import { ElasticsearchClient } from './config/elasticsearch.js';
import { ClickHouseClient } from './config/clickhouse.js';
import routes from './routes/index.js';
import { EventCollector } from './services/EventCollector.js';
import { MetricsProcessor } from './services/MetricsProcessor.js';
import { RealtimeAnalytics } from './services/RealtimeAnalytics.js';
import { ReportGenerator } from './services/ReportGenerator.js';
import { AlertManager } from './services/AlertManager.js';
const init = async () => {
// Initialize database connections
await connectDatabase();
const redisClient = RedisClient.getInstance();
await redisClient.connect();
const elasticsearchClient = ElasticsearchClient.getInstance();
await elasticsearchClient.connect();
const clickhouseClient = ClickHouseClient.getInstance();
await clickhouseClient.connect();
// Initialize services
EventCollector.getInstance();
MetricsProcessor.getInstance();
RealtimeAnalytics.getInstance();
ReportGenerator.getInstance();
AlertManager.getInstance();
// Create Hapi server
const server = Hapi.server({
port: process.env.PORT || 3005,
host: process.env.HOST || 'localhost',
routes: {
cors: {
origin: ['*'],
headers: ['Accept', 'Content-Type', 'Authorization'],
credentials: true
},
payload: {
maxBytes: 10485760 // 10MB
}
}
});
// Register routes
server.route(routes);
// Health check endpoint
server.route({
method: 'GET',
path: '/health',
handler: async (request, h) => {
const dbHealth = await checkDatabaseHealth();
const redisHealth = await redisClient.checkHealth();
const esHealth = await elasticsearchClient.checkHealth();
const chHealth = await clickhouseClient.checkHealth();
const isHealthy = dbHealth && redisHealth && esHealth && chHealth;
return h.response({
status: isHealthy ? 'healthy' : 'unhealthy',
service: 'analytics',
timestamp: new Date().toISOString(),
checks: {
database: dbHealth,
redis: redisHealth,
elasticsearch: esHealth,
clickhouse: chHealth
}
}).code(isHealthy ? 200 : 503);
}
});
// Start server
await server.start();
logger.info(`Analytics service running on ${server.info.uri}`);
// Start background processors
const metricsProcessor = MetricsProcessor.getInstance();
metricsProcessor.startProcessing();
const alertManager = AlertManager.getInstance();
alertManager.startMonitoring();
// Graceful shutdown
process.on('SIGTERM', async () => {
logger.info('SIGTERM signal received');
await server.stop({ timeout: 10000 });
await redisClient.disconnect();
await elasticsearchClient.disconnect();
await clickhouseClient.disconnect();
process.exit(0);
});
process.on('SIGINT', async () => {
logger.info('SIGINT signal received');
await server.stop({ timeout: 10000 });
await redisClient.disconnect();
await elasticsearchClient.disconnect();
await clickhouseClient.disconnect();
process.exit(0);
});
};
const checkDatabaseHealth = async () => {
try {
const { connection } = await import('mongoose');
return connection.readyState === 1;
} catch (error) {
logger.error('Database health check failed:', error);
return false;
}
};
// Handle unhandled rejections
process.on('unhandledRejection', (err) => {
logger.error('Unhandled rejection:', err);
process.exit(1);
});
// Start the service
init().catch((err) => {
logger.error('Failed to start Analytics service:', err);
process.exit(1);
});

88
marketing-agent/services/analytics/src/models/Alert.js Normal file
View File

@@ -0,0 +1,88 @@
import mongoose from 'mongoose';
const alertSchema = new mongoose.Schema({
// Multi-tenant support
tenantId: {
type: mongoose.Schema.Types.ObjectId,
ref: 'Tenant',
required: true,
index: true
},
ruleId: {
type: String,
required: true,
index: true
},
ruleName: {
type: String,
required: true
},
severity: {
type: String,
enum: ['low', 'medium', 'high', 'critical'],
required: true,
index: true
},
metric: {
type: String,
required: true
},
value: {
type: Number,
required: true
},
threshold: {
type: Number,
required: true
},
operator: {
type: String,
required: true
},
status: {
type: String,
enum: ['triggered', 'acknowledged', 'resolved'],
default: 'triggered',
index: true
},
triggeredAt: {
type: Date,
required: true,
index: true
},
acknowledgedAt: Date,
resolvedAt: Date,
resolution: String,
notificationsSent: [{
channel: String,
sentAt: Date,
status: String
}],
metadata: {
type: Map,
of: mongoose.Schema.Types.Mixed
}
}, {
timestamps: true
});
// Indexes
alertSchema.index({ ruleId: 1, triggeredAt: -1 });
alertSchema.index({ severity: 1, status: 1, triggeredAt: -1 });
alertSchema.index({ status: 1, triggeredAt: -1 });
// Multi-tenant indexes
alertSchema.index({ tenantId: 1, ruleId: 1, triggeredAt: -1 });
alertSchema.index({ tenantId: 1, severity: 1, status: 1, triggeredAt: -1 });
alertSchema.index({ tenantId: 1, status: 1, triggeredAt: -1 });
// TTL index to auto-delete resolved alerts after 30 days
alertSchema.index(
{ resolvedAt: 1 },
{
expireAfterSeconds: 2592000,
partialFilterExpression: { status: 'resolved' }
}
);
export const Alert = mongoose.model('Alert', alertSchema);

88
marketing-agent/services/analytics/src/models/AlertRule.js Normal file
View File

@@ -0,0 +1,88 @@
import mongoose from 'mongoose';
const alertRuleSchema = new mongoose.Schema({
// Multi-tenant support
tenantId: {
type: mongoose.Schema.Types.ObjectId,
ref: 'Tenant',
required: true,
index: true
},
ruleId: {
type: String,
required: true,
unique: true,
index: true
},
name: {
type: String,
required: true
},
description: String,
metric: {
type: String,
required: true,
index: true
},
condition: {
operator: {
type: String,
enum: ['>', '>=', '<', '<=', '=', '==', '!='],
required: true
},
threshold: {
type: Number,
required: true
},
duration: {
type: Number,
default: 300 // 5 minutes in seconds
}
},
severity: {
type: String,
enum: ['low', 'medium', 'high', 'critical'],
required: true,
index: true
},
channels: [{
type: String,
enum: ['email', 'sms', 'webhook', 'slack']
}],
cooldown: {
type: Number,
default: 1800 // 30 minutes in seconds
},
accountId: {
type: String,
index: true
},
active: {
type: Boolean,
default: true,
index: true
},
lastTriggered: Date,
triggerCount: {
type: Number,
default: 0
},
metadata: {
type: Map,
of: mongoose.Schema.Types.Mixed
}
}, {
timestamps: true
});
// Indexes
alertRuleSchema.index({ active: 1, metric: 1 });
alertRuleSchema.index({ accountId: 1, active: 1 });
alertRuleSchema.index({ severity: 1, active: 1 });
// Multi-tenant indexes
alertRuleSchema.index({ tenantId: 1, active: 1, metric: 1 });
alertRuleSchema.index({ tenantId: 1, accountId: 1, active: 1 });
alertRuleSchema.index({ tenantId: 1, severity: 1, active: 1 });
export const AlertRule = mongoose.model('AlertRule', alertRuleSchema);

80
marketing-agent/services/analytics/src/models/Event.js Normal file
View File

@@ -0,0 +1,80 @@
import mongoose from 'mongoose';
const eventSchema = new mongoose.Schema({
// Multi-tenant support
tenantId: {
type: mongoose.Schema.Types.ObjectId,
ref: 'Tenant',
required: true,
index: true
},
id: {
type: String,
required: true,
unique: true,
index: true
},
type: {
type: String,
required: true,
index: true
},
accountId: {
type: String,
required: true,
index: true
},
userId: {
type: String,
index: true
},
sessionId: {
type: String,
index: true
},
action: {
type: String,
required: true,
index: true
},
target: String,
value: Number,
metadata: {
type: Map,
of: mongoose.Schema.Types.Mixed
},
properties: {
type: Map,
of: mongoose.Schema.Types.Mixed
},
timestamp: {
type: Date,
required: true,
index: true
}
}, {
timestamps: true,
timeseries: {
timeField: 'timestamp',
metaField: 'metadata',
granularity: 'seconds'
}
});
// Compound indexes for common queries
eventSchema.index({ accountId: 1, timestamp: -1 });
eventSchema.index({ accountId: 1, type: 1, timestamp: -1 });
eventSchema.index({ accountId: 1, userId: 1, timestamp: -1 });
eventSchema.index({ type: 1, action: 1, timestamp: -1 });
// TTL index to auto-delete old events after 1 year
eventSchema.index({ timestamp: 1 }, { expireAfterSeconds: 31536000 });
// Multi-tenant indexes
eventSchema.index({ tenantId: 1, accountId: 1, timestamp: -1 });
eventSchema.index({ tenantId: 1, accountId: 1, type: 1, timestamp: -1 });
eventSchema.index({ tenantId: 1, accountId: 1, userId: 1, timestamp: -1 });
eventSchema.index({ tenantId: 1, type: 1, action: 1, timestamp: -1 });
eventSchema.index({ tenantId: 1, timestamp: 1 }, { expireAfterSeconds: 31536000 });
export const Event = mongoose.model('Event', eventSchema);

67
marketing-agent/services/analytics/src/models/MetricDefinition.js Normal file
View File

@@ -0,0 +1,67 @@
import mongoose from 'mongoose';
const metricDefinitionSchema = new mongoose.Schema({
// Multi-tenant support
tenantId: {
type: mongoose.Schema.Types.ObjectId,
ref: 'Tenant',
required: true,
index: true
},
metricId: {
type: String,
required: true,
unique: true,
index: true
},
name: {
type: String,
required: true
},
description: String,
type: {
type: String,
enum: ['percentage', 'count', 'duration', 'currency', 'rate'],
required: true
},
formula: {
type: String,
required: true
},
dimensions: [{
type: String
}],
aggregations: [{
type: String,
enum: ['avg', 'sum', 'min', 'max', 'median', 'p95', 'stddev']
}],
refreshInterval: {
type: Number,
default: 300 // 5 minutes in seconds
},
retentionDays: {
type: Number,
default: 90
},
active: {
type: Boolean,
default: true
},
lastProcessed: Date,
metadata: {
type: Map,
of: mongoose.Schema.Types.Mixed
}
}, {
timestamps: true
});
// Indexes
metricDefinitionSchema.index({ active: 1, lastProcessed: 1 });
metricDefinitionSchema.index({ type: 1 });
// Multi-tenant indexes
metricDefinitionSchema.index({ tenantId: 1, active: 1, lastProcessed: 1 });
metricDefinitionSchema.index({ tenantId: 1, type: 1 });
export const MetricDefinition = mongoose.model('MetricDefinition', metricDefinitionSchema);

60
marketing-agent/services/analytics/src/models/ProcessedMetric.js Normal file
View File

@@ -0,0 +1,60 @@
import mongoose from 'mongoose';
const processedMetricSchema = new mongoose.Schema({
// Multi-tenant support
tenantId: {
type: mongoose.Schema.Types.ObjectId,
ref: 'Tenant',
required: true,
index: true
},
metricId: {
type: String,
required: true,
index: true
},
dimensions: {
type: Map,
of: String,
index: true
},
value: {
type: Number,
required: true
},
aggregations: {
type: Map,
of: Number
},
dataPoints: {
type: Number,
default: 0
},
timestamp: {
type: Date,
required: true,
index: true
}
}, {
timestamps: true,
timeseries: {
timeField: 'timestamp',
metaField: 'dimensions',
granularity: 'minutes'
}
});
// Compound indexes
processedMetricSchema.index({ metricId: 1, timestamp: -1 });
processedMetricSchema.index({ metricId: 1, 'dimensions.*': 1, timestamp: -1 });
// TTL index based on metric retention settings
// This should be dynamic based on metric definition, but we'll use 90 days as default
processedMetricSchema.index({ timestamp: 1 }, { expireAfterSeconds: 7776000 });
// Multi-tenant indexes
processedMetricSchema.index({ tenantId: 1, metricId: 1, timestamp: -1 });
processedMetricSchema.index({ tenantId: 1, metricId: 1, 'dimensions.*': 1, timestamp: -1 });
processedMetricSchema.index({ tenantId: 1, timestamp: 1 }, { expireAfterSeconds: 7776000 }); // 90 days
export const ProcessedMetric = mongoose.model('ProcessedMetric', processedMetricSchema);

81
marketing-agent/services/analytics/src/models/Report.js Normal file
View File

@@ -0,0 +1,81 @@
import mongoose from 'mongoose';
const reportSchema = new mongoose.Schema({
// Multi-tenant support
tenantId: {
type: mongoose.Schema.Types.ObjectId,
ref: 'Tenant',
required: true,
index: true
},
reportId: {
type: String,
required: true,
unique: true,
index: true
},
type: {
type: String,
required: true,
index: true
},
accountId: {
type: String,
required: true,
index: true
},
dateRange: {
start: {
type: Date,
required: true
},
end: {
type: Date,
required: true
}
},
data: {
type: mongoose.Schema.Types.Mixed,
required: true
},
format: {
type: String,
enum: ['json', 'pdf', 'excel', 'csv'],
default: 'json'
},
status: {
type: String,
enum: ['pending', 'processing', 'completed', 'failed'],
default: 'pending',
index: true
},
url: String,
error: String,
generatedAt: {
type: Date,
default: Date.now,
index: true
},
metadata: {
type: Map,
of: mongoose.Schema.Types.Mixed
}
}, {
timestamps: true
});
// Indexes
reportSchema.index({ accountId: 1, generatedAt: -1 });
reportSchema.index({ accountId: 1, type: 1, generatedAt: -1 });
reportSchema.index({ status: 1, generatedAt: -1 });
// TTL index to auto-delete old reports after 90 days
reportSchema.index({ generatedAt: 1 }, { expireAfterSeconds: 7776000 });
// Multi-tenant indexes
reportSchema.index({ tenantId: 1, accountId: 1, generatedAt: -1 });
reportSchema.index({ tenantId: 1, accountId: 1, type: 1, generatedAt: -1 });
reportSchema.index({ tenantId: 1, status: 1, generatedAt: -1 });
reportSchema.index({ tenantId: 1, generatedAt: 1 }, { expireAfterSeconds: 7776000 });
export const Report = mongoose.model('Report', reportSchema);

100
marketing-agent/services/analytics/src/routes/dashboard.js Normal file
View File

@@ -0,0 +1,100 @@
export default [
{
method: 'GET',
path: '/dashboard',
options: {
tags: ['api', 'analytics'],
description: 'Get dashboard overview data'
},
handler: async (request, h) => {
try {
// 返回模拟的仪表板数据
const dashboardData = {
overview: {
totalCampaigns: 12,
activeCampaigns: 5,
totalMessages: 45678,
deliveryRate: 98.5,
clickRate: 12.3,
conversionRate: 3.2
},
recentActivity: [
{
id: '1',
type: 'campaign_started',
campaign: 'Summer Sale 2025',
timestamp: new Date(Date.now() - 3600000).toISOString()
},
{
id: '2',
type: 'message_sent',
count: 1500,
campaign: 'Welcome Series',
timestamp: new Date(Date.now() - 7200000).toISOString()
},
{
id: '3',
type: 'campaign_completed',
campaign: 'Flash Promo',
results: {
sent: 3000,
delivered: 2950,
clicked: 450
},
timestamp: new Date(Date.now() - 10800000).toISOString()
}
],
performance: {
daily: [
{ date: '2025-07-20', sent: 5000, delivered: 4900, clicked: 600 },
{ date: '2025-07-21', sent: 5500, delivered: 5400, clicked: 720 },
{ date: '2025-07-22', sent: 4800, delivered: 4700, clicked: 580 },
{ date: '2025-07-23', sent: 6200, delivered: 6100, clicked: 850 },
{ date: '2025-07-24', sent: 5800, delivered: 5700, clicked: 690 },
{ date: '2025-07-25', sent: 6500, delivered: 6400, clicked: 820 },
{ date: '2025-07-26', sent: 3200, delivered: 3150, clicked: 390 }
]
},
topCampaigns: [
{
id: 'c1',
name: 'Summer Sale 2025',
status: 'active',
messages: 12500,
deliveryRate: 99.2,
clickRate: 15.8
},
{
id: 'c2',
name: 'Welcome Series',
status: 'active',
messages: 8900,
deliveryRate: 98.5,
clickRate: 22.1
},
{
id: 'c3',
name: 'Product Launch',
status: 'scheduled',
messages: 0,
deliveryRate: 0,
clickRate: 0
}
]
};
return h.response({
success: true,
data: dashboardData
}).code(200);
} catch (error) {
request.server.app.logger.error('Dashboard error:', error);
return h.response({
success: false,
error: 'Failed to fetch dashboard data'
}).code(500);
}
}
}
];

17
marketing-agent/services/analytics/src/routes/health.js Normal file
View File

@@ -0,0 +1,17 @@
import express from 'express';
const router = express.Router();
/**
* Health check endpoint
*/
router.get('/', (req, res) => {
res.json({
status: 'healthy',
service: 'analytics',
timestamp: new Date().toISOString(),
uptime: process.uptime()
});
});
export default router;

660
marketing-agent/services/analytics/src/routes/index.js Normal file
View File

@@ -0,0 +1,660 @@
import Joi from '@hapi/joi';
import { EventCollector } from '../services/EventCollector.js';
import { MetricsProcessor } from '../services/MetricsProcessor.js';
import { RealtimeAnalytics } from '../services/RealtimeAnalytics.js';
import { ReportGenerator } from '../services/ReportGenerator.js';
import { AlertManager } from '../services/AlertManager.js';
import { logger } from '../utils/logger.js';
import dashboardRoutes from './dashboard.js';
const eventCollector = EventCollector.getInstance();
const metricsProcessor = MetricsProcessor.getInstance();
const realtimeAnalytics = RealtimeAnalytics.getInstance();
const reportGenerator = ReportGenerator.getInstance();
const alertManager = AlertManager.getInstance();
export default [
...dashboardRoutes,
// Event Collection Routes
{
method: 'POST',
path: '/api/v1/events',
options: {
validate: {
payload: Joi.object({
type: Joi.string().required(),
accountId: Joi.string().required(),
userId: Joi.string().optional(),
sessionId: Joi.string().optional(),
action: Joi.string().required(),
target: Joi.string().optional(),
value: Joi.number().optional(),
metadata: Joi.object().optional(),
properties: Joi.object().optional()
})
},
handler: async (request, h) => {
try {
const result = await eventCollector.collectEvent(request.payload);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Event collection error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'POST',
path: '/api/v1/events/bulk',
options: {
validate: {
payload: Joi.object({
events: Joi.array().items(
Joi.object({
type: Joi.string().required(),
accountId: Joi.string().required(),
userId: Joi.string().optional(),
sessionId: Joi.string().optional(),
action: Joi.string().required(),
target: Joi.string().optional(),
value: Joi.number().optional(),
metadata: Joi.object().optional(),
properties: Joi.object().optional()
})
).required()
})
},
handler: async (request, h) => {
try {
const { events } = request.payload;
const result = await eventCollector.collectBulkEvents(events);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Bulk event collection error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'GET',
path: '/api/v1/events',
options: {
validate: {
query: Joi.object({
type: Joi.string().optional(),
accountId: Joi.string().optional(),
userId: Joi.string().optional(),
startTime: Joi.date().iso().optional(),
endTime: Joi.date().iso().optional(),
limit: Joi.number().min(1).max(1000).default(100),
offset: Joi.number().min(0).default(0),
aggregation: Joi.string().valid('hourly', 'daily', 'by_type').optional()
})
},
handler: async (request, h) => {
try {
const result = await eventCollector.queryEvents(request.query);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Event query error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
// Metrics Routes
{
method: 'GET',
path: '/api/v1/metrics/{metricId}',
options: {
validate: {
params: Joi.object({
metricId: Joi.string().required()
}),
query: Joi.object({
startTime: Joi.date().iso().optional(),
endTime: Joi.date().iso().optional(),
dimensions: Joi.object().optional(),
limit: Joi.number().min(1).max(10000).default(1000)
})
},
handler: async (request, h) => {
try {
const { metricId } = request.params;
const result = await metricsProcessor.getMetric(metricId, request.query);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Metric query error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'GET',
path: '/api/v1/metrics/{metricId}/summary',
options: {
validate: {
params: Joi.object({
metricId: Joi.string().required()
})
},
handler: async (request, h) => {
try {
const { metricId } = request.params;
const result = await metricsProcessor.getMetricSummary(metricId);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Metric summary error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'PUT',
path: '/api/v1/metrics/{metricId}',
options: {
validate: {
params: Joi.object({
metricId: Joi.string().required()
}),
payload: Joi.object({
name: Joi.string(),
description: Joi.string(),
formula: Joi.string(),
dimensions: Joi.array().items(Joi.string()),
aggregations: Joi.array().items(Joi.string()),
refreshInterval: Joi.number(),
retentionDays: Joi.number(),
active: Joi.boolean()
})
},
handler: async (request, h) => {
try {
const { metricId } = request.params;
const result = await metricsProcessor.updateMetricDefinition(metricId, request.payload);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Metric update error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
// Real-time Analytics Routes
{
method: 'GET',
path: '/api/v1/realtime/dashboard/{accountId}',
options: {
validate: {
params: Joi.object({
accountId: Joi.string().required()
})
},
handler: async (request, h) => {
try {
const { accountId } = request.params;
const result = await realtimeAnalytics.getDashboard(accountId);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Dashboard error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'POST',
path: '/api/v1/realtime/subscribe',
options: {
validate: {
payload: Joi.object({
accountId: Joi.string().required(),
metrics: Joi.array().items(Joi.string()).required(),
filters: Joi.object().optional()
})
},
handler: async (request, h) => {
try {
const { accountId, metrics, filters } = request.payload;
const subscriptionId = await realtimeAnalytics.subscribe(accountId, metrics, filters);
return h.response({
success: true,
data: { subscriptionId }
});
} catch (error) {
logger.error('Subscription error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'DELETE',
path: '/api/v1/realtime/subscribe/{subscriptionId}',
options: {
validate: {
params: Joi.object({
subscriptionId: Joi.string().required()
})
},
handler: async (request, h) => {
try {
const { subscriptionId } = request.params;
const result = realtimeAnalytics.unsubscribe(subscriptionId);
return h.response({
success: true,
data: { unsubscribed: result }
});
} catch (error) {
logger.error('Unsubscribe error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
// Report Generation Routes
{
method: 'POST',
path: '/api/v1/reports/generate',
options: {
validate: {
payload: Joi.object({
accountId: Joi.string().required(),
type: Joi.string().valid(
'campaign_performance',
'user_analytics',
'ab_test'
).required(),
period: Joi.string().valid(
'today', 'yesterday', 'this_week', 'last_week',
'this_month', 'last_month', 'last_30_days', 'last_90_days'
).optional(),
startDate: Joi.date().iso().optional(),
endDate: Joi.date().iso().optional(),
filters: Joi.object().optional(),
format: Joi.string().valid('json', 'pdf', 'excel', 'csv').default('json')
})
},
handler: async (request, h) => {
try {
const result = await reportGenerator.generateReport(request.payload);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Report generation error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'POST',
path: '/api/v1/reports/schedule',
options: {
validate: {
payload: Joi.object({
accountId: Joi.string().required(),
type: Joi.string().required(),
schedule: Joi.string().required(), // cron expression
recipients: Joi.array().items(Joi.string()).required(),
format: Joi.string().valid('pdf', 'excel', 'csv').default('pdf')
})
},
handler: async (request, h) => {
try {
const result = await reportGenerator.scheduleReport(request.payload);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Report scheduling error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'GET',
path: '/api/v1/reports/history',
options: {
validate: {
query: Joi.object({
accountId: Joi.string().required(),
limit: Joi.number().min(1).max(100).default(20)
})
},
handler: async (request, h) => {
try {
const { accountId, limit } = request.query;
const result = await reportGenerator.getReportHistory(accountId, limit);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Report history error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
// Alert Management Routes
{
method: 'POST',
path: '/api/v1/alerts/rules',
options: {
validate: {
payload: Joi.object({
ruleId: Joi.string().required(),
name: Joi.string().required(),
description: Joi.string().optional(),
metric: Joi.string().required(),
condition: Joi.object({
operator: Joi.string().valid('>', '>=', '<', '<=', '=', '==', '!=').required(),
threshold: Joi.number().required(),
duration: Joi.number().min(0).default(300)
}).required(),
severity: Joi.string().valid('low', 'medium', 'high', 'critical').required(),
channels: Joi.array().items(Joi.string()).required(),
cooldown: Joi.number().min(0).default(1800),
accountId: Joi.string().optional(),
active: Joi.boolean().default(true)
})
},
handler: async (request, h) => {
try {
const result = await alertManager.createAlertRule(request.payload);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Alert rule creation error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'PUT',
path: '/api/v1/alerts/rules/{ruleId}',
options: {
validate: {
params: Joi.object({
ruleId: Joi.string().required()
}),
payload: Joi.object({
name: Joi.string(),
description: Joi.string(),
condition: Joi.object(),
severity: Joi.string().valid('low', 'medium', 'high', 'critical'),
channels: Joi.array().items(Joi.string()),
cooldown: Joi.number(),
active: Joi.boolean()
})
},
handler: async (request, h) => {
try {
const { ruleId } = request.params;
const result = await alertManager.updateAlertRule(ruleId, request.payload);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Alert rule update error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'DELETE',
path: '/api/v1/alerts/rules/{ruleId}',
options: {
validate: {
params: Joi.object({
ruleId: Joi.string().required()
})
},
handler: async (request, h) => {
try {
const { ruleId } = request.params;
const result = await alertManager.deleteAlertRule(ruleId);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Alert rule deletion error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'GET',
path: '/api/v1/alerts/history',
options: {
validate: {
query: Joi.object({
ruleId: Joi.string().optional(),
severity: Joi.string().optional(),
startTime: Joi.date().iso().optional(),
endTime: Joi.date().iso().optional(),
limit: Joi.number().min(1).max(1000).default(100)
})
},
handler: async (request, h) => {
try {
const result = await alertManager.getAlertHistory(request.query);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Alert history error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'POST',
path: '/api/v1/alerts/{alertId}/acknowledge',
options: {
validate: {
params: Joi.object({
alertId: Joi.string().required()
})
},
handler: async (request, h) => {
try {
const { alertId } = request.params;
const result = await alertManager.acknowledgeAlert(alertId);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Alert acknowledge error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'POST',
path: '/api/v1/alerts/{alertId}/resolve',
options: {
validate: {
params: Joi.object({
alertId: Joi.string().required()
}),
payload: Joi.object({
resolution: Joi.string().required()
})
},
handler: async (request, h) => {
try {
const { alertId } = request.params;
const { resolution } = request.payload;
const result = await alertManager.resolveAlert(alertId, resolution);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Alert resolve error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
},
{
method: 'GET',
path: '/api/v1/alerts/stats',
options: {
validate: {
query: Joi.object({
period: Joi.string().valid('1h', '24h', '7d').default('24h')
})
},
handler: async (request, h) => {
try {
const { period } = request.query;
const result = await alertManager.getAlertStats(period);
return h.response({
success: true,
data: result
});
} catch (error) {
logger.error('Alert stats error:', error);
return h.response({
success: false,
error: error.message
}).code(500);
}
}
}
}
];

516
marketing-agent/services/analytics/src/services/AlertManager.js Normal file
View File

@@ -0,0 +1,516 @@
import { logger } from '../utils/logger.js';
import { Alert } from '../models/Alert.js';
import { AlertRule } from '../models/AlertRule.js';
import { RedisClient } from '../config/redis.js';
import { MetricsProcessor } from './MetricsProcessor.js';
import { RealtimeAnalytics } from './RealtimeAnalytics.js';
import { sendNotification } from '../utils/notifications.js';
import * as math from 'mathjs';
export class AlertManager {
constructor() {
this.redis = null;
this.metricsProcessor = null;
this.realtimeAnalytics = null;
this.rules = new Map();
this.checkInterval = 30000; // 30 seconds
this.isMonitoring = false;
}
static getInstance() {
if (!AlertManager.instance) {
AlertManager.instance = new AlertManager();
AlertManager.instance.initialize();
}
return AlertManager.instance;
}
async initialize() {
this.redis = RedisClient.getInstance();
this.metricsProcessor = MetricsProcessor.getInstance();
this.realtimeAnalytics = RealtimeAnalytics.getInstance();
// Load alert rules
await this.loadAlertRules();
logger.info('Alert manager initialized');
}
async loadAlertRules() {
try {
const rules = await AlertRule.find({ active: true });
for (const rule of rules) {
this.rules.set(rule.ruleId, rule);
}
// Create default rules if none exist
if (this.rules.size === 0) {
await this.createDefaultRules();
}
logger.info(`Loaded ${this.rules.size} alert rules`);
} catch (error) {
logger.error('Failed to load alert rules:', error);
}
}
async createDefaultRules() {
const defaultRules = [
{
ruleId: 'high_error_rate',
name: 'High Error Rate',
description: 'Alert when error rate exceeds threshold',
metric: 'error_rate',
condition: {
operator: '>',
threshold: 5,
duration: 300 // 5 minutes
},
severity: 'critical',
channels: ['email', 'webhook'],
cooldown: 1800 // 30 minutes
},
{
ruleId: 'low_engagement',
name: 'Low Engagement Rate',
description: 'Alert when engagement rate drops below threshold',
metric: 'engagement_rate',
condition: {
operator: '<',
threshold: 10,
duration: 1800 // 30 minutes
},
severity: 'warning',
channels: ['email'],
cooldown: 3600 // 1 hour
},
{
ruleId: 'message_delivery_failure',
name: 'Message Delivery Failure',
description: 'Alert when message delivery rate is low',
metric: 'message_delivery_rate',
condition: {
operator: '<',
threshold: 90,
duration: 600 // 10 minutes
},
severity: 'high',
channels: ['email', 'sms'],
cooldown: 1800
},
{
ruleId: 'high_response_time',
name: 'High Response Time',
description: 'Alert when response time exceeds threshold',
metric: 'response_time_p95',
condition: {
operator: '>',
threshold: 1000, // 1 second
duration: 300
},
severity: 'warning',
channels: ['email'],
cooldown: 1800
},
{
ruleId: 'rate_limit_violations',
name: 'Rate Limit Violations',
description: 'Alert when rate limit violations spike',
metric: 'rate_limit_violations',
condition: {
operator: '>',
threshold: 100,
duration: 300
},
severity: 'high',
channels: ['email', 'webhook'],
cooldown: 900 // 15 minutes
}
];
for (const ruleData of defaultRules) {
const rule = await AlertRule.create(ruleData);
this.rules.set(rule.ruleId, rule);
}
}
startMonitoring() {
setInterval(async () => {
if (!this.isMonitoring) {
await this.checkAlerts();
}
}, this.checkInterval);
// Check immediately
this.checkAlerts();
logger.info('Alert monitoring started');
}
async checkAlerts() {
this.isMonitoring = true;
try {
const activeRules = Array.from(this.rules.values())
.filter(rule => rule.active);
logger.debug(`Checking ${activeRules.length} alert rules`);
for (const rule of activeRules) {
try {
await this.checkRule(rule);
} catch (error) {
logger.error(`Failed to check rule ${rule.ruleId}:`, error);
}
}
} catch (error) {
logger.error('Alert checking failed:', error);
} finally {
this.isMonitoring = false;
}
}
async checkRule(rule) {
const { ruleId, metric, condition, severity, channels, cooldown } = rule;
// Check if rule is in cooldown
if (await this.isInCooldown(ruleId)) {
return;
}
// Get metric value
const metricValue = await this.getMetricValue(metric, rule.accountId);
if (metricValue === null) {
logger.warn(`No data for metric ${metric}`);
return;
}
// Check condition
const isTriggered = this.evaluateCondition(metricValue, condition);
if (isTriggered) {
// Check if condition has been met for required duration
const conditionMet = await this.checkConditionDuration(ruleId, condition.duration);
if (conditionMet) {
await this.triggerAlert(rule, metricValue);
}
} else {
// Reset condition tracking
await this.resetConditionTracking(ruleId);
}
}
async getMetricValue(metric, accountId) {
try {
// Try real-time analytics first
const realtimeData = await this.realtimeAnalytics.getRealtimeMetric(
accountId || 'global',
metric.replace('_p95', '').replace('_violations', '')
);
if (realtimeData) {
// Extract specific value based on metric type
if (metric.includes('p95')) {
return realtimeData.p95 || null;
} else if (metric.includes('rate')) {
return realtimeData.current || null;
} else {
return realtimeData.current || realtimeData.value || null;
}
}
// Fallback to processed metrics
const processedMetric = await this.metricsProcessor.getMetricSummary(metric);
return processedMetric?.summary?.averageValue || null;
} catch (error) {
logger.error(`Failed to get metric value for ${metric}:`, error);
return null;
}
}
evaluateCondition(value, condition) {
const { operator, threshold } = condition;
switch (operator) {
case '>':
return value > threshold;
case '>=':
return value >= threshold;
case '<':
return value < threshold;
case '<=':
return value <= threshold;
case '=':
case '==':
return value === threshold;
case '!=':
return value !== threshold;
default:
logger.warn(`Unknown operator: ${operator}`);
return false;
}
}
async checkConditionDuration(ruleId, duration) {
const key = `alert:condition:${ruleId}`;
const firstTrigger = await this.redis.get(key);
if (!firstTrigger) {
// First time condition is met
await this.redis.setWithExpiry(key, Date.now(), duration + 60);
return false;
}
// Check if duration has passed
const elapsed = Date.now() - parseInt(firstTrigger);
return elapsed >= duration * 1000;
}
async resetConditionTracking(ruleId) {
const key = `alert:condition:${ruleId}`;
await this.redis.del(key);
}
async isInCooldown(ruleId) {
const key = `alert:cooldown:${ruleId}`;
const exists = await this.redis.exists(key);
return exists;
}
async setCooldown(ruleId, duration) {
const key = `alert:cooldown:${ruleId}`;
await this.redis.setWithExpiry(key, Date.now(), duration);
}
async triggerAlert(rule, metricValue) {
const { ruleId, name, severity, channels, cooldown } = rule;
try {
// Create alert record
const alert = await Alert.create({
ruleId,
ruleName: name,
severity,
metric: rule.metric,
value: metricValue,
threshold: rule.condition.threshold,
operator: rule.condition.operator,
status: 'triggered',
triggeredAt: new Date()
});
// Send notifications
await this.sendAlertNotifications(alert, rule, channels);
// Set cooldown
await this.setCooldown(ruleId, cooldown);
// Reset condition tracking
await this.resetConditionTracking(ruleId);
logger.info(`Alert triggered: ${name} (${severity}) - Value: ${metricValue}`);
} catch (error) {
logger.error('Failed to trigger alert:', error);
}
}
async sendAlertNotifications(alert, rule, channels) {
const message = this.formatAlertMessage(alert, rule);
for (const channel of channels) {
try {
await sendNotification(channel, {
subject: `[${alert.severity.toUpperCase()}] ${alert.ruleName}`,
message,
alertId: alert._id,
metadata: {
ruleId: rule.ruleId,
metric: rule.metric,
value: alert.value,
threshold: alert.threshold
}
});
} catch (error) {
logger.error(`Failed to send ${channel} notification:`, error);
}
}
}
formatAlertMessage(alert, rule) {
const { ruleName, severity, metric, value, threshold, operator, triggeredAt } = alert;
return `
Alert: ${ruleName}
Severity: ${severity.toUpperCase()}
Time: ${triggeredAt.toISOString()}
Condition: ${metric} ${operator} ${threshold}
Current Value: ${value}
Description: ${rule.description}
Please investigate and take appropriate action.
`.trim();
}
async createAlertRule(ruleData) {
try {
const rule = await AlertRule.create(ruleData);
this.rules.set(rule.ruleId, rule);
logger.info(`Created alert rule: ${rule.ruleId}`);
return rule;
} catch (error) {
logger.error('Failed to create alert rule:', error);
throw error;
}
}
async updateAlertRule(ruleId, updates) {
try {
const rule = await AlertRule.findOneAndUpdate(
{ ruleId },
updates,
{ new: true }
);
if (rule) {
this.rules.set(ruleId, rule);
logger.info(`Updated alert rule: ${ruleId}`);
}
return rule;
} catch (error) {
logger.error('Failed to update alert rule:', error);
throw error;
}
}
async deleteAlertRule(ruleId) {
try {
await AlertRule.deleteOne({ ruleId });
this.rules.delete(ruleId);
logger.info(`Deleted alert rule: ${ruleId}`);
return { success: true };
} catch (error) {
logger.error('Failed to delete alert rule:', error);
throw error;
}
}
async getAlertHistory(filters = {}) {
const query = {};
if (filters.ruleId) {
query.ruleId = filters.ruleId;
}
if (filters.severity) {
query.severity = filters.severity;
}
if (filters.startTime || filters.endTime) {
query.triggeredAt = {};
if (filters.startTime) {
query.triggeredAt.$gte = new Date(filters.startTime);
}
if (filters.endTime) {
query.triggeredAt.$lte = new Date(filters.endTime);
}
}
const alerts = await Alert.find(query)
.sort({ triggeredAt: -1 })
.limit(filters.limit || 100);
return alerts;
}
async acknowledgeAlert(alertId) {
try {
const alert = await Alert.findByIdAndUpdate(
alertId,
{
status: 'acknowledged',
acknowledgedAt: new Date()
},
{ new: true }
);
logger.info(`Alert acknowledged: ${alertId}`);
return alert;
} catch (error) {
logger.error('Failed to acknowledge alert:', error);
throw error;
}
}
async resolveAlert(alertId, resolution) {
try {
const alert = await Alert.findByIdAndUpdate(
alertId,
{
status: 'resolved',
resolvedAt: new Date(),
resolution
},
{ new: true }
);
logger.info(`Alert resolved: ${alertId}`);
return alert;
} catch (error) {
logger.error('Failed to resolve alert:', error);
throw error;
}
}
async getAlertStats(period = '24h') {
const since = new Date();
switch (period) {
case '1h':
since.setHours(since.getHours() - 1);
break;
case '24h':
since.setDate(since.getDate() - 1);
break;
case '7d':
since.setDate(since.getDate() - 7);
break;
}
const stats = await Alert.aggregate([
{ $match: { triggeredAt: { $gte: since } } },
{
$group: {
_id: {
severity: '$severity',
status: '$status'
},
count: { $sum: 1 }
}
},
{
$group: {
_id: '$_id.severity',
statuses: {
$push: {
status: '$_id.status',
count: '$count'
}
},
total: { $sum: '$count' }
}
}
]);
return {
period,
since,
stats
};
}
}

363
marketing-agent/services/analytics/src/services/EventCollector.js Normal file
View File

@@ -0,0 +1,363 @@
import { v4 as uuidv4 } from 'uuid';
import { logger } from '../utils/logger.js';
import { RedisClient } from '../config/redis.js';
import { ElasticsearchClient } from '../config/elasticsearch.js';
import { ClickHouseClient } from '../config/clickhouse.js';
import { Event } from '../models/Event.js';
import { validateEvent } from '../utils/validators.js';
export class EventCollector {
constructor() {
this.redis = null;
this.elasticsearch = null;
this.clickhouse = null;
this.batchSize = 100;
this.batchInterval = 5000; // 5 seconds
this.eventQueue = [];
this.processing = false;
}
static getInstance() {
if (!EventCollector.instance) {
EventCollector.instance = new EventCollector();
EventCollector.instance.initialize();
}
return EventCollector.instance;
}
initialize() {
this.redis = RedisClient.getInstance();
this.elasticsearch = ElasticsearchClient.getInstance();
this.clickhouse = ClickHouseClient.getInstance();
// Start batch processing
this.startBatchProcessing();
logger.info('Event collector initialized');
}
async collectEvent(eventData) {
try {
// Validate event
const validation = validateEvent(eventData);
if (!validation.valid) {
throw new Error(`Invalid event: ${validation.errors.join(', ')}`);
}
// Enrich event
const event = {
id: uuidv4(),
timestamp: new Date(),
...eventData,
metadata: {
...eventData.metadata,
collectedAt: new Date().toISOString(),
version: '1.0'
}
};
// Add to queue for batch processing
this.eventQueue.push(event);
// Store in Redis for real-time access
await this.storeRealtimeEvent(event);
// Process immediately if batch is full
if (this.eventQueue.length >= this.batchSize) {
await this.processBatch();
}
return {
success: true,
eventId: event.id,
timestamp: event.timestamp
};
} catch (error) {
logger.error('Failed to collect event:', error);
throw error;
}
}
async collectBulkEvents(events) {
const results = [];
const errors = [];
for (const eventData of events) {
try {
const result = await this.collectEvent(eventData);
results.push(result);
} catch (error) {
errors.push({
event: eventData,
error: error.message
});
}
}
return {
success: errors.length === 0,
collected: results.length,
failed: errors.length,
results,
errors
};
}
async storeRealtimeEvent(event) {
try {
// Store in Redis for real-time access
const key = `event:realtime:${event.type}:${event.accountId}`;
await this.redis.lpush(key, event);
await this.redis.ltrim(key, 0, 999); // Keep last 1000 events
await this.redis.expire(key, 3600); // Expire after 1 hour
// Update real-time counters
await this.updateRealtimeCounters(event);
} catch (error) {
logger.error('Failed to store realtime event:', error);
}
}
async updateRealtimeCounters(event) {
const now = new Date();
const minuteKey = `counter:${event.type}:${now.getMinutes()}`;
const hourKey = `counter:${event.type}:${now.getHours()}`;
const dayKey = `counter:${event.type}:${now.getDate()}`;
await Promise.all([
this.redis.client.incr(minuteKey),
this.redis.client.incr(hourKey),
this.redis.client.incr(dayKey),
this.redis.client.expire(minuteKey, 300), // 5 minutes
this.redis.client.expire(hourKey, 7200), // 2 hours
this.redis.client.expire(dayKey, 172800) // 2 days
]);
}
startBatchProcessing() {
setInterval(async () => {
if (this.eventQueue.length > 0 && !this.processing) {
await this.processBatch();
}
}, this.batchInterval);
}
async processBatch() {
if (this.processing || this.eventQueue.length === 0) {
return;
}
this.processing = true;
const batch = this.eventQueue.splice(0, this.batchSize);
try {
// Store in different backends
await Promise.all([
this.storeInMongoDB(batch),
this.storeInElasticsearch(batch),
this.storeInClickHouse(batch)
]);
logger.info(`Processed batch of ${batch.length} events`);
} catch (error) {
logger.error('Failed to process batch:', error);
// Return events to queue for retry
this.eventQueue.unshift(...batch);
} finally {
this.processing = false;
}
}
async storeInMongoDB(events) {
try {
await Event.insertMany(events, { ordered: false });
} catch (error) {
logger.error('Failed to store events in MongoDB:', error);
throw error;
}
}
async storeInElasticsearch(events) {
try {
const body = events.flatMap(event => [
{ index: { _index: 'events', _id: event.id } },
event
]);
const response = await this.elasticsearch.client.bulk({ body });
if (response.errors) {
logger.error('Elasticsearch bulk insert had errors:', response.errors);
}
} catch (error) {
logger.error('Failed to store events in Elasticsearch:', error);
throw error;
}
}
async storeInClickHouse(events) {
try {
const values = events.map(event => ({
id: event.id,
timestamp: event.timestamp,
type: event.type,
accountId: event.accountId,
userId: event.userId || null,
sessionId: event.sessionId || null,
action: event.action,
target: event.target || null,
value: event.value || null,
metadata: JSON.stringify(event.metadata),
properties: JSON.stringify(event.properties || {})
}));
await this.clickhouse.insert({
table: 'events',
values,
format: 'JSONEachRow'
});
} catch (error) {
logger.error('Failed to store events in ClickHouse:', error);
throw error;
}
}
async queryEvents(params) {
const {
type,
accountId,
userId,
startTime,
endTime,
limit = 100,
offset = 0,
aggregation = null
} = params;
try {
// Build query
const query = {
bool: {
must: []
}
};
if (type) {
query.bool.must.push({ term: { type } });
}
if (accountId) {
query.bool.must.push({ term: { accountId } });
}
if (userId) {
query.bool.must.push({ term: { userId } });
}
if (startTime || endTime) {
const range = { timestamp: {} };
if (startTime) range.timestamp.gte = startTime;
if (endTime) range.timestamp.lte = endTime;
query.bool.must.push({ range });
}
// Add aggregations if requested
const aggs = {};
if (aggregation) {
switch (aggregation) {
case 'hourly':
aggs.events_over_time = {
date_histogram: {
field: 'timestamp',
calendar_interval: '1h'
}
};
break;
case 'daily':
aggs.events_over_time = {
date_histogram: {
field: 'timestamp',
calendar_interval: '1d'
}
};
break;
case 'by_type':
aggs.events_by_type = {
terms: {
field: 'type',
size: 50
}
};
break;
}
}
// Execute query
const response = await this.elasticsearch.client.search({
index: 'events',
body: {
query,
aggs,
size: limit,
from: offset,
sort: [{ timestamp: 'desc' }]
}
});
return {
total: response.hits.total.value,
events: response.hits.hits.map(hit => hit._source),
aggregations: response.aggregations || null
};
} catch (error) {
logger.error('Failed to query events:', error);
throw error;
}
}
async getEventTypes() {
try {
const response = await this.elasticsearch.client.search({
index: 'events',
body: {
size: 0,
aggs: {
event_types: {
terms: {
field: 'type',
size: 100
}
}
}
}
});
return response.aggregations.event_types.buckets.map(bucket => ({
type: bucket.key,
count: bucket.doc_count
}));
} catch (error) {
logger.error('Failed to get event types:', error);
throw error;
}
}
async getEventStream(accountId, types = []) {
try {
const keys = types.length > 0
? types.map(type => `event:realtime:${type}:${accountId}`)
: [`event:realtime:*:${accountId}`];
const events = [];
for (const key of keys) {
const keyEvents = await this.redis.lrange(key, 0, 49);
events.push(...keyEvents);
}
// Sort by timestamp and return latest 50
return events
.sort((a, b) => new Date(b.timestamp) - new Date(a.timestamp))
.slice(0, 50);
} catch (error) {
logger.error('Failed to get event stream:', error);
throw error;
}
}
}

468
marketing-agent/services/analytics/src/services/MetricsProcessor.js Normal file
View File

@@ -0,0 +1,468 @@
import { logger } from '../utils/logger.js';
import { RedisClient } from '../config/redis.js';
import { ClickHouseClient } from '../config/clickhouse.js';
import { MetricDefinition } from '../models/MetricDefinition.js';
import { ProcessedMetric } from '../models/ProcessedMetric.js';
import * as math from 'mathjs';
import * as stats from 'simple-statistics';
import { parseExpression } from '../utils/metricParser.js';
export class MetricsProcessor {
constructor() {
this.redis = null;
this.clickhouse = null;
this.definitions = new Map();
this.processingInterval = 60000; // 1 minute
this.isProcessing = false;
}
static getInstance() {
if (!MetricsProcessor.instance) {
MetricsProcessor.instance = new MetricsProcessor();
MetricsProcessor.instance.initialize();
}
return MetricsProcessor.instance;
}
async initialize() {
this.redis = RedisClient.getInstance();
this.clickhouse = ClickHouseClient.getInstance();
// Load metric definitions
await this.loadMetricDefinitions();
logger.info('Metrics processor initialized');
}
async loadMetricDefinitions() {
try {
const definitions = await MetricDefinition.find({ active: true });
for (const def of definitions) {
this.definitions.set(def.metricId, def);
}
// Create default metrics if none exist
if (this.definitions.size === 0) {
await this.createDefaultMetrics();
}
logger.info(`Loaded ${this.definitions.size} metric definitions`);
} catch (error) {
logger.error('Failed to load metric definitions:', error);
}
}
async createDefaultMetrics() {
const defaultMetrics = [
{
metricId: 'engagement_rate',
name: 'Engagement Rate',
description: 'Percentage of users who engaged with messages',
type: 'percentage',
formula: '(unique_engaged_users / unique_recipients) * 100',
dimensions: ['campaign', 'channel', 'time_period'],
aggregations: ['avg', 'min', 'max'],
refreshInterval: 300, // 5 minutes
retentionDays: 90
},
{
metricId: 'conversion_rate',
name: 'Conversion Rate',
description: 'Percentage of users who completed the desired action',
type: 'percentage',
formula: '(conversions / unique_recipients) * 100',
dimensions: ['campaign', 'channel', 'action_type'],
aggregations: ['avg', 'sum'],
refreshInterval: 300,
retentionDays: 90
},
{
metricId: 'message_delivery_rate',
name: 'Message Delivery Rate',
description: 'Percentage of messages successfully delivered',
type: 'percentage',
formula: '(delivered_messages / sent_messages) * 100',
dimensions: ['campaign', 'account', 'message_type'],
aggregations: ['avg'],
refreshInterval: 60,
retentionDays: 30
},
{
metricId: 'response_time',
name: 'Average Response Time',
description: 'Average time between message sent and user response',
type: 'duration',
formula: 'avg(response_timestamp - sent_timestamp)',
dimensions: ['campaign', 'user_segment'],
aggregations: ['avg', 'median', 'p95'],
refreshInterval: 600,
retentionDays: 30
},
{
metricId: 'user_retention',
name: 'User Retention Rate',
description: 'Percentage of users who remain active over time',
type: 'percentage',
formula: '(active_users_end / active_users_start) * 100',
dimensions: ['cohort', 'time_period'],
aggregations: ['avg'],
refreshInterval: 3600, // 1 hour
retentionDays: 365
}
];
for (const metricData of defaultMetrics) {
const metric = await MetricDefinition.create(metricData);
this.definitions.set(metric.metricId, metric);
}
}
startProcessing() {
setInterval(async () => {
if (!this.isProcessing) {
await this.processMetrics();
}
}, this.processingInterval);
// Process immediately
this.processMetrics();
}
async processMetrics() {
this.isProcessing = true;
const startTime = Date.now();
try {
const metricsToProcess = Array.from(this.definitions.values())
.filter(def => this.shouldProcess(def));
logger.info(`Processing ${metricsToProcess.length} metrics`);
for (const definition of metricsToProcess) {
try {
await this.processMetric(definition);
} catch (error) {
logger.error(`Failed to process metric ${definition.metricId}:`, error);
}
}
const duration = Date.now() - startTime;
logger.info(`Metric processing completed in ${duration}ms`);
} catch (error) {
logger.error('Metric processing failed:', error);
} finally {
this.isProcessing = false;
}
}
shouldProcess(definition) {
const lastProcessed = definition.lastProcessed;
if (!lastProcessed) return true;
const timeSinceLastProcess = Date.now() - lastProcessed.getTime();
return timeSinceLastProcess >= definition.refreshInterval * 1000;
}
async processMetric(definition) {
const { metricId, formula, dimensions, aggregations } = definition;
try {
// Parse formula to extract required data points
const dataPoints = parseExpression(formula);
// Fetch raw data from ClickHouse
const rawData = await this.fetchRawData(dataPoints, dimensions);
// Calculate metric values
const results = this.calculateMetric(rawData, formula, dimensions, aggregations);
// Store processed metrics
await this.storeProcessedMetrics(metricId, results);
// Update last processed time
definition.lastProcessed = new Date();
await definition.save();
logger.debug(`Processed metric ${metricId}: ${results.length} data points`);
} catch (error) {
logger.error(`Failed to process metric ${metricId}:`, error);
throw error;
}
}
async fetchRawData(dataPoints, dimensions) {
const queries = [];
for (const dataPoint of dataPoints) {
let query = `
SELECT
${dimensions.join(', ')},
${this.buildDataPointQuery(dataPoint)} as value,
toStartOfMinute(timestamp) as time_bucket
FROM events
WHERE timestamp >= now() - INTERVAL 1 DAY
`;
if (dimensions.length > 0) {
query += ` GROUP BY ${dimensions.join(', ')}, time_bucket`;
}
queries.push({ dataPoint, query });
}
const results = {};
for (const { dataPoint, query } of queries) {
const response = await this.clickhouse.query({ query });
results[dataPoint] = response.data;
}
return results;
}
buildDataPointQuery(dataPoint) {
// Convert data point names to ClickHouse queries
const queryMap = {
'unique_engaged_users': 'uniqExact(userId) WHERE action = \'engage\'',
'unique_recipients': 'uniqExact(userId) WHERE action = \'receive\'',
'conversions': 'count() WHERE action = \'convert\'',
'delivered_messages': 'count() WHERE type = \'message_delivered\'',
'sent_messages': 'count() WHERE type = \'message_sent\'',
'active_users_start': 'uniqExact(userId) WHERE date = today() - 30',
'active_users_end': 'uniqExact(userId) WHERE date = today()'
};
return queryMap[dataPoint] || `count() WHERE type = '${dataPoint}'`;
}
calculateMetric(rawData, formula, dimensions, aggregations) {
const results = [];
// Group data by dimensions
const groupedData = this.groupByDimensions(rawData, dimensions);
for (const [key, data] of groupedData.entries()) {
try {
// Calculate base value
const value = this.evaluateFormula(formula, data);
// Calculate aggregations
const aggregatedValues = {};
for (const agg of aggregations) {
aggregatedValues[agg] = this.calculateAggregation(agg, data);
}
results.push({
dimensions: this.parseDimensionKey(key, dimensions),
value,
aggregations: aggregatedValues,
timestamp: new Date(),
dataPoints: data.length
});
} catch (error) {
logger.error(`Failed to calculate metric for ${key}:`, error);
}
}
return results;
}
groupByDimensions(rawData, dimensions) {
const grouped = new Map();
// Flatten all data points
const allData = [];
for (const [dataPoint, values] of Object.entries(rawData)) {
for (const value of values) {
allData.push({ ...value, dataPoint });
}
}
// Group by dimension values
for (const item of allData) {
const key = dimensions.map(dim => item[dim] || 'unknown').join(':');
if (!grouped.has(key)) {
grouped.set(key, []);
}
grouped.get(key).push(item);
}
return grouped;
}
evaluateFormula(formula, data) {
// Create a context with data point values
const context = {};
// Aggregate data points by type
const dataByType = {};
for (const item of data) {
if (!dataByType[item.dataPoint]) {
dataByType[item.dataPoint] = [];
}
dataByType[item.dataPoint].push(item.value);
}
// Calculate sum for each data point
for (const [dataPoint, values] of Object.entries(dataByType)) {
context[dataPoint] = stats.sum(values);
}
// Evaluate formula
try {
return math.evaluate(formula, context);
} catch (error) {
logger.error('Formula evaluation error:', error);
return null;
}
}
calculateAggregation(aggregationType, data) {
const values = data.map(d => d.value).filter(v => v !== null);
if (values.length === 0) return null;
switch (aggregationType) {
case 'avg':
return stats.mean(values);
case 'sum':
return stats.sum(values);
case 'min':
return stats.min(values);
case 'max':
return stats.max(values);
case 'median':
return stats.median(values);
case 'p95':
return stats.quantile(values, 0.95);
case 'stddev':
return stats.standardDeviation(values);
default:
return null;
}
}
parseDimensionKey(key, dimensions) {
const values = key.split(':');
const result = {};
for (let i = 0; i < dimensions.length; i++) {
result[dimensions[i]] = values[i] || 'unknown';
}
return result;
}
async storeProcessedMetrics(metricId, results) {
try {
// Store in MongoDB for historical tracking
const documents = results.map(result => ({
metricId,
...result
}));
await ProcessedMetric.insertMany(documents);
// Store in Redis for real-time access
for (const result of results) {
const key = `metric:${metricId}:${Object.values(result.dimensions).join(':')}`;
await this.redis.setWithExpiry(key, result, 3600); // 1 hour cache
}
// Update metric summary
await this.updateMetricSummary(metricId, results);
} catch (error) {
logger.error('Failed to store processed metrics:', error);
throw error;
}
}
async updateMetricSummary(metricId, results) {
const summary = {
lastUpdate: new Date(),
dataPoints: results.length,
averageValue: stats.mean(results.map(r => r.value).filter(v => v !== null)),
minValue: stats.min(results.map(r => r.value).filter(v => v !== null)),
maxValue: stats.max(results.map(r => r.value).filter(v => v !== null))
};
await this.redis.hset('metric:summaries', metricId, summary);
}
async getMetric(metricId, filters = {}) {
try {
const definition = this.definitions.get(metricId);
if (!definition) {
throw new Error(`Metric ${metricId} not found`);
}
// Build query
const query = { metricId };
if (filters.startTime || filters.endTime) {
query.timestamp = {};
if (filters.startTime) query.timestamp.$gte = new Date(filters.startTime);
if (filters.endTime) query.timestamp.$lte = new Date(filters.endTime);
}
if (filters.dimensions) {
for (const [key, value] of Object.entries(filters.dimensions)) {
query[`dimensions.${key}`] = value;
}
}
// Fetch data
const data = await ProcessedMetric.find(query)
.sort({ timestamp: -1 })
.limit(filters.limit || 1000);
return {
metricId,
name: definition.name,
description: definition.description,
data
};
} catch (error) {
logger.error('Failed to get metric:', error);
throw error;
}
}
async getMetricSummary(metricId) {
try {
const summary = await this.redis.hget('metric:summaries', metricId);
const definition = this.definitions.get(metricId);
return {
metricId,
name: definition?.name,
description: definition?.description,
summary
};
} catch (error) {
logger.error('Failed to get metric summary:', error);
throw error;
}
}
async updateMetricDefinition(metricId, updates) {
try {
const metric = await MetricDefinition.findOneAndUpdate(
{ metricId },
updates,
{ new: true }
);
if (metric) {
this.definitions.set(metricId, metric);
logger.info(`Updated metric definition: ${metricId}`);
}
return metric;
} catch (error) {
logger.error('Failed to update metric definition:', error);
throw error;
}
}
}

394
marketing-agent/services/analytics/src/services/RealtimeAnalytics.js Normal file
View File

@@ -0,0 +1,394 @@
import { logger } from '../utils/logger.js';
import { RedisClient } from '../config/redis.js';
import { EventCollector } from './EventCollector.js';
import { WebSocketManager } from '../utils/websocket.js';
import { calculateTrend, detectAnomaly } from '../utils/analytics.js';
export class RealtimeAnalytics {
constructor() {
this.redis = null;
this.eventCollector = null;
this.wsManager = null;
this.subscribers = new Map();
this.updateInterval = 1000; // 1 second
this.trendWindow = 300; // 5 minutes
}
static getInstance() {
if (!RealtimeAnalytics.instance) {
RealtimeAnalytics.instance = new RealtimeAnalytics();
RealtimeAnalytics.instance.initialize();
}
return RealtimeAnalytics.instance;
}
initialize() {
this.redis = RedisClient.getInstance();
this.eventCollector = EventCollector.getInstance();
this.wsManager = WebSocketManager.getInstance();
// Start real-time processing
this.startRealtimeProcessing();
logger.info('Realtime analytics initialized');
}
startRealtimeProcessing() {
setInterval(async () => {
await this.processRealtimeData();
}, this.updateInterval);
}
async processRealtimeData() {
try {
// Get all active subscriptions
const subscriptions = Array.from(this.subscribers.values());
for (const subscription of subscriptions) {
await this.processSubscription(subscription);
}
} catch (error) {
logger.error('Realtime processing error:', error);
}
}
async processSubscription(subscription) {
const { id, accountId, metrics, filters } = subscription;
try {
const data = {};
// Fetch real-time data for each metric
for (const metric of metrics) {
data[metric] = await this.getRealtimeMetric(accountId, metric, filters);
}
// Send update to subscriber
this.wsManager.sendToClient(id, {
type: 'realtime_update',
data,
timestamp: new Date()
});
} catch (error) {
logger.error(`Failed to process subscription ${id}:`, error);
}
}
async getRealtimeMetric(accountId, metric, filters = {}) {
const now = new Date();
switch (metric) {
case 'active_users':
return await this.getActiveUsers(accountId, filters);
case 'message_rate':
return await this.getMessageRate(accountId, filters);
case 'engagement_rate':
return await this.getEngagementRate(accountId, filters);
case 'conversion_funnel':
return await this.getConversionFunnel(accountId, filters);
case 'error_rate':
return await this.getErrorRate(accountId, filters);
case 'response_time':
return await this.getResponseTime(accountId, filters);
default:
return null;
}
}
async getActiveUsers(accountId, filters) {
const windows = [
{ label: '1m', seconds: 60 },
{ label: '5m', seconds: 300 },
{ label: '15m', seconds: 900 },
{ label: '1h', seconds: 3600 }
];
const results = {};
for (const window of windows) {
const key = `active:${accountId}:${window.label}`;
const users = await this.redis.smembers(key);
results[window.label] = users.length;
}
// Calculate trend
const trend = await this.calculateMetricTrend('active_users', accountId);
return {
current: results['1m'],
windows: results,
trend,
anomaly: detectAnomaly(Object.values(results))
};
}
async getMessageRate(accountId, filters) {
const counters = await this.getTimeSeriesCounters('message', accountId, 60);
const rate = counters.reduce((sum, val) => sum + val, 0) / counters.length;
return {
current: rate,
timeSeries: counters,
trend: calculateTrend(counters),
peak: Math.max(...counters),
average: rate
};
}
async getEngagementRate(accountId, filters) {
const sent = await this.getCounter('sent', accountId);
const engaged = await this.getCounter('engaged', accountId);
const rate = sent > 0 ? (engaged / sent) * 100 : 0;
// Get historical rates for comparison
const historical = await this.getHistoricalRates('engagement', accountId, 24);
return {
current: rate,
sent,
engaged,
historical,
trend: calculateTrend(historical),
benchmark: 25 // Industry average
};
}
async getConversionFunnel(accountId, filters) {
const stages = filters.stages || [
'impression',
'click',
'engagement',
'conversion'
];
const funnel = [];
let previousCount = null;
for (const stage of stages) {
const count = await this.getCounter(stage, accountId);
const rate = previousCount ? (count / previousCount) * 100 : 100;
funnel.push({
stage,
count,
rate,
dropoff: previousCount ? previousCount - count : 0
});
previousCount = count;
}
return {
stages: funnel,
overallConversion: funnel.length > 0 ?
(funnel[funnel.length - 1].count / funnel[0].count) * 100 : 0,
timestamp: new Date()
};
}
async getErrorRate(accountId, filters) {
const errors = await this.getTimeSeriesCounters('error', accountId, 60);
const total = await this.getTimeSeriesCounters('request', accountId, 60);
const rates = errors.map((err, i) =>
total[i] > 0 ? (err / total[i]) * 100 : 0
);
const currentRate = rates[rates.length - 1] || 0;
return {
current: currentRate,
timeSeries: rates,
errors: errors.reduce((sum, val) => sum + val, 0),
requests: total.reduce((sum, val) => sum + val, 0),
trend: calculateTrend(rates),
alert: currentRate > 5 // Alert if error rate > 5%
};
}
async getResponseTime(accountId, filters) {
const key = `response_times:${accountId}`;
const times = await this.redis.lrange(key, -100, -1);
if (times.length === 0) {
return {
current: 0,
average: 0,
median: 0,
p95: 0,
p99: 0
};
}
const values = times.map(t => parseFloat(t)).sort((a, b) => a - b);
return {
current: values[values.length - 1],
average: values.reduce((sum, val) => sum + val, 0) / values.length,
median: values[Math.floor(values.length / 2)],
p95: values[Math.floor(values.length * 0.95)],
p99: values[Math.floor(values.length * 0.99)],
samples: values.length
};
}
async getTimeSeriesCounters(type, accountId, points) {
const counters = [];
const now = new Date();
for (let i = points - 1; i >= 0; i--) {
const minute = new Date(now - i * 60000).getMinutes();
const key = `counter:${type}:${accountId}:${minute}`;
const value = await this.redis.client.get(key);
counters.push(parseInt(value) || 0);
}
return counters;
}
async getCounter(type, accountId) {
const minute = new Date().getMinutes();
const key = `counter:${type}:${accountId}:${minute}`;
const value = await this.redis.client.get(key);
return parseInt(value) || 0;
}
async getHistoricalRates(type, accountId, hours) {
const rates = [];
const now = new Date();
for (let i = hours - 1; i >= 0; i--) {
const hour = new Date(now - i * 3600000).getHours();
const key = `rate:${type}:${accountId}:${hour}`;
const value = await this.redis.client.get(key);
rates.push(parseFloat(value) || 0);
}
return rates;
}
async calculateMetricTrend(metric, accountId) {
const historical = await this.getTimeSeriesCounters(metric, accountId, 10);
const trend = calculateTrend(historical);
return {
direction: trend > 0 ? 'up' : trend < 0 ? 'down' : 'stable',
percentage: Math.abs(trend),
confidence: this.calculateConfidence(historical)
};
}
calculateConfidence(data) {
// Simple confidence based on data variance
if (data.length < 3) return 0;
const mean = data.reduce((sum, val) => sum + val, 0) / data.length;
const variance = data.reduce((sum, val) => sum + Math.pow(val - mean, 2), 0) / data.length;
const cv = Math.sqrt(variance) / mean; // Coefficient of variation
// Lower CV means higher confidence
return Math.max(0, Math.min(100, (1 - cv) * 100));
}
async subscribe(accountId, metrics, filters = {}) {
const subscriptionId = `sub_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
const subscription = {
id: subscriptionId,
accountId,
metrics,
filters,
createdAt: new Date()
};
this.subscribers.set(subscriptionId, subscription);
// Send initial data
await this.processSubscription(subscription);
return subscriptionId;
}
unsubscribe(subscriptionId) {
const removed = this.subscribers.delete(subscriptionId);
if (removed) {
logger.info(`Unsubscribed: ${subscriptionId}`);
}
return removed;
}
async getDashboard(accountId) {
const metrics = [
'active_users',
'message_rate',
'engagement_rate',
'conversion_funnel',
'error_rate',
'response_time'
];
const dashboard = {};
for (const metric of metrics) {
dashboard[metric] = await this.getRealtimeMetric(accountId, metric);
}
// Add summary statistics
dashboard.summary = {
health: this.calculateHealthScore(dashboard),
alerts: this.checkAlerts(dashboard),
timestamp: new Date()
};
return dashboard;
}
calculateHealthScore(dashboard) {
let score = 100;
// Deduct points for issues
if (dashboard.error_rate?.current > 5) score -= 20;
if (dashboard.error_rate?.current > 10) score -= 30;
if (dashboard.engagement_rate?.current < 10) score -= 15;
if (dashboard.response_time?.p95 > 1000) score -= 10;
return Math.max(0, score);
}
checkAlerts(dashboard) {
const alerts = [];
if (dashboard.error_rate?.alert) {
alerts.push({
type: 'error_rate',
severity: 'high',
message: `Error rate is ${dashboard.error_rate.current.toFixed(2)}%`
});
}
if (dashboard.engagement_rate?.current < 10) {
alerts.push({
type: 'low_engagement',
severity: 'medium',
message: `Engagement rate is only ${dashboard.engagement_rate.current.toFixed(2)}%`
});
}
if (dashboard.response_time?.p95 > 1000) {
alerts.push({
type: 'slow_response',
severity: 'medium',
message: `95th percentile response time is ${dashboard.response_time.p95}ms`
});
}
return alerts;
}
}

583
marketing-agent/services/analytics/src/services/ReportGenerator.js Normal file
View File

@@ -0,0 +1,583 @@
import { logger } from '../utils/logger.js';
import { Report } from '../models/Report.js';
import { ProcessedMetric } from '../models/ProcessedMetric.js';
import { MetricsProcessor } from './MetricsProcessor.js';
import { EventCollector } from './EventCollector.js';
import { format, startOfDay, endOfDay, startOfWeek, endOfWeek, startOfMonth, endOfMonth } from 'date-fns';
import * as stats from 'simple-statistics';
import { generatePDF, generateExcel, generateCSV } from '../utils/exporters.js';
export class ReportGenerator {
constructor() {
this.metricsProcessor = null;
this.eventCollector = null;
this.templates = new Map();
}
static getInstance() {
if (!ReportGenerator.instance) {
ReportGenerator.instance = new ReportGenerator();
ReportGenerator.instance.initialize();
}
return ReportGenerator.instance;
}
initialize() {
this.metricsProcessor = MetricsProcessor.getInstance();
this.eventCollector = EventCollector.getInstance();
// Load report templates
this.loadReportTemplates();
logger.info('Report generator initialized');
}
loadReportTemplates() {
// Campaign Performance Report
this.templates.set('campaign_performance', {
name: 'Campaign Performance Report',
sections: [
{ type: 'summary', title: 'Executive Summary' },
{ type: 'metrics', title: 'Key Metrics', metrics: ['engagement_rate', 'conversion_rate', 'message_delivery_rate'] },
{ type: 'trends', title: 'Performance Trends' },
{ type: 'segments', title: 'Audience Segments' },
{ type: 'recommendations', title: 'Recommendations' }
]
});
// User Analytics Report
this.templates.set('user_analytics', {
name: 'User Analytics Report',
sections: [
{ type: 'overview', title: 'User Overview' },
{ type: 'demographics', title: 'Demographics' },
{ type: 'behavior', title: 'User Behavior' },
{ type: 'retention', title: 'Retention Analysis' },
{ type: 'segments', title: 'User Segments' }
]
});
// A/B Test Report
this.templates.set('ab_test', {
name: 'A/B Test Report',
sections: [
{ type: 'summary', title: 'Test Summary' },
{ type: 'results', title: 'Results Analysis' },
{ type: 'statistical_significance', title: 'Statistical Significance' },
{ type: 'recommendations', title: 'Recommendations' }
]
});
}
async generateReport(params) {
const {
accountId,
type,
period,
startDate,
endDate,
filters = {},
format = 'json'
} = params;
try {
// Determine date range
const dateRange = this.getDateRange(period, startDate, endDate);
// Get template
const template = this.templates.get(type);
if (!template) {
throw new Error(`Unknown report type: ${type}`);
}
// Generate report data
const reportData = {
metadata: {
reportId: `report_${Date.now()}`,
type,
accountId,
dateRange,
generatedAt: new Date(),
template: template.name
},
sections: {}
};
// Generate each section
for (const section of template.sections) {
reportData.sections[section.type] = await this.generateSection(
section,
accountId,
dateRange,
filters
);
}
// Save report
const report = await this.saveReport(reportData);
// Export in requested format
const exported = await this.exportReport(report, format);
return {
reportId: report._id,
type,
format,
data: exported,
url: report.url
};
} catch (error) {
logger.error('Failed to generate report:', error);
throw error;
}
}
getDateRange(period, startDate, endDate) {
const now = new Date();
if (startDate && endDate) {
return {
start: new Date(startDate),
end: new Date(endDate)
};
}
switch (period) {
case 'today':
return {
start: startOfDay(now),
end: endOfDay(now)
};
case 'yesterday':
const yesterday = new Date(now);
yesterday.setDate(yesterday.getDate() - 1);
return {
start: startOfDay(yesterday),
end: endOfDay(yesterday)
};
case 'this_week':
return {
start: startOfWeek(now),
end: endOfWeek(now)
};
case 'last_week':
const lastWeek = new Date(now);
lastWeek.setDate(lastWeek.getDate() - 7);
return {
start: startOfWeek(lastWeek),
end: endOfWeek(lastWeek)
};
case 'this_month':
return {
start: startOfMonth(now),
end: endOfMonth(now)
};
case 'last_month':
const lastMonth = new Date(now);
lastMonth.setMonth(lastMonth.getMonth() - 1);
return {
start: startOfMonth(lastMonth),
end: endOfMonth(lastMonth)
};
case 'last_30_days':
const thirtyDaysAgo = new Date(now);
thirtyDaysAgo.setDate(thirtyDaysAgo.getDate() - 30);
return {
start: thirtyDaysAgo,
end: now
};
case 'last_90_days':
const ninetyDaysAgo = new Date(now);
ninetyDaysAgo.setDate(ninetyDaysAgo.getDate() - 90);
return {
start: ninetyDaysAgo,
end: now
};
default:
// Default to last 7 days
const sevenDaysAgo = new Date(now);
sevenDaysAgo.setDate(sevenDaysAgo.getDate() - 7);
return {
start: sevenDaysAgo,
end: now
};
}
}
async generateSection(section, accountId, dateRange, filters) {
switch (section.type) {
case 'summary':
return await this.generateSummarySection(accountId, dateRange, filters);
case 'metrics':
return await this.generateMetricsSection(section.metrics, accountId, dateRange, filters);
case 'trends':
return await this.generateTrendsSection(accountId, dateRange, filters);
case 'segments':
return await this.generateSegmentsSection(accountId, dateRange, filters);
case 'demographics':
return await this.generateDemographicsSection(accountId, dateRange, filters);
case 'behavior':
return await this.generateBehaviorSection(accountId, dateRange, filters);
case 'retention':
return await this.generateRetentionSection(accountId, dateRange, filters);
case 'recommendations':
return await this.generateRecommendationsSection(accountId, dateRange, filters);
case 'results':
return await this.generateResultsSection(accountId, dateRange, filters);
case 'statistical_significance':
return await this.generateStatisticalSection(accountId, dateRange, filters);
default:
return { error: `Unknown section type: ${section.type}` };
}
}
async generateSummarySection(accountId, dateRange, filters) {
// Fetch key metrics for the period
const metrics = await this.fetchKeyMetrics(accountId, dateRange);
// Calculate comparisons with previous period
const previousRange = this.getPreviousPeriod(dateRange);
const previousMetrics = await this.fetchKeyMetrics(accountId, previousRange);
const comparisons = {};
for (const [key, value] of Object.entries(metrics)) {
const previous = previousMetrics[key] || 0;
const change = previous > 0 ? ((value - previous) / previous) * 100 : 0;
comparisons[key] = {
current: value,
previous,
change,
trend: change > 0 ? 'up' : change < 0 ? 'down' : 'stable'
};
}
return {
period: {
start: dateRange.start,
end: dateRange.end,
days: Math.ceil((dateRange.end - dateRange.start) / (1000 * 60 * 60 * 24))
},
highlights: this.generateHighlights(comparisons),
metrics: comparisons
};
}
async generateMetricsSection(metricIds, accountId, dateRange, filters) {
const metricsData = {};
for (const metricId of metricIds) {
const data = await this.metricsProcessor.getMetric(metricId, {
startTime: dateRange.start,
endTime: dateRange.end,
dimensions: filters
});
if (data && data.data.length > 0) {
const values = data.data.map(d => d.value).filter(v => v !== null);
metricsData[metricId] = {
name: data.name,
description: data.description,
summary: {
average: stats.mean(values),
median: stats.median(values),
min: stats.min(values),
max: stats.max(values),
total: stats.sum(values),
count: values.length
},
timeSeries: data.data.map(d => ({
timestamp: d.timestamp,
value: d.value,
dimensions: d.dimensions
}))
};
}
}
return metricsData;
}
async generateTrendsSection(accountId, dateRange, filters) {
const metrics = ['engagement_rate', 'conversion_rate', 'message_delivery_rate'];
const trends = {};
for (const metricId of metrics) {
const data = await this.metricsProcessor.getMetric(metricId, {
startTime: dateRange.start,
endTime: dateRange.end
});
if (data && data.data.length > 1) {
const timeSeries = data.data
.sort((a, b) => new Date(a.timestamp) - new Date(b.timestamp))
.map(d => ({ x: d.timestamp, y: d.value }));
// Calculate trend line
const xValues = timeSeries.map((_, i) => i);
const yValues = timeSeries.map(d => d.y);
const regression = stats.linearRegression([xValues, yValues]);
const slope = regression.m;
trends[metricId] = {
name: data.name,
timeSeries,
trend: {
direction: slope > 0 ? 'increasing' : slope < 0 ? 'decreasing' : 'stable',
slope,
r2: stats.rSquared([xValues, yValues], regression)
},
forecast: this.generateForecast(timeSeries, 7) // 7 day forecast
};
}
}
return trends;
}
async generateSegmentsSection(accountId, dateRange, filters) {
// Analyze performance by different segments
const segmentTypes = ['channel', 'campaign', 'user_segment', 'message_type'];
const segments = {};
for (const segmentType of segmentTypes) {
const segmentData = await this.analyzeSegment(segmentType, accountId, dateRange);
if (segmentData.length > 0) {
segments[segmentType] = {
topPerformers: segmentData.slice(0, 5),
bottomPerformers: segmentData.slice(-5).reverse(),
distribution: this.calculateDistribution(segmentData)
};
}
}
return segments;
}
async generateRecommendationsSection(accountId, dateRange, filters) {
const recommendations = [];
// Analyze current performance
const metrics = await this.fetchKeyMetrics(accountId, dateRange);
// Low engagement recommendation
if (metrics.engagement_rate < 15) {
recommendations.push({
priority: 'high',
category: 'engagement',
title: 'Low Engagement Rate',
description: `Your engagement rate is ${metrics.engagement_rate.toFixed(2)}%, which is below the industry average of 20-25%.`,
actions: [
'Personalize message content based on user preferences',
'Test different message timing and frequency',
'Use more compelling CTAs and visual content'
]
});
}
// High error rate recommendation
if (metrics.error_rate > 5) {
recommendations.push({
priority: 'critical',
category: 'technical',
title: 'High Error Rate',
description: `Your error rate is ${metrics.error_rate.toFixed(2)}%, indicating technical issues.`,
actions: [
'Review recent error logs and identify common issues',
'Implement better error handling and retry mechanisms',
'Monitor API rate limits and adjust sending patterns'
]
});
}
// Conversion optimization
if (metrics.conversion_rate < 2) {
recommendations.push({
priority: 'medium',
category: 'conversion',
title: 'Conversion Rate Optimization',
description: `Your conversion rate is ${metrics.conversion_rate.toFixed(2)}%, with room for improvement.`,
actions: [
'Implement A/B testing for different message variations',
'Optimize landing pages for mobile devices',
'Create more targeted campaigns for high-intent users'
]
});
}
return recommendations;
}
async fetchKeyMetrics(accountId, dateRange) {
// This would fetch actual metrics from the database
// For now, return mock data
const events = await this.eventCollector.queryEvents({
accountId,
startTime: dateRange.start,
endTime: dateRange.end
});
const metrics = {
total_messages: events.total,
unique_users: 0, // Calculate from events
engagement_rate: 22.5,
conversion_rate: 3.2,
message_delivery_rate: 98.5,
error_rate: 1.2,
average_response_time: 245 // ms
};
return metrics;
}
getPreviousPeriod(dateRange) {
const duration = dateRange.end - dateRange.start;
return {
start: new Date(dateRange.start - duration),
end: new Date(dateRange.start)
};
}
generateHighlights(comparisons) {
const highlights = [];
// Find biggest improvements
for (const [metric, data] of Object.entries(comparisons)) {
if (data.change > 20) {
highlights.push({
type: 'improvement',
metric,
message: `${metric} improved by ${data.change.toFixed(1)}%`
});
} else if (data.change < -20) {
highlights.push({
type: 'decline',
metric,
message: `${metric} declined by ${Math.abs(data.change).toFixed(1)}%`
});
}
}
return highlights.slice(0, 5); // Top 5 highlights
}
generateForecast(timeSeries, days) {
// Simple linear forecast
const xValues = timeSeries.map((_, i) => i);
const yValues = timeSeries.map(d => d.y);
const regression = stats.linearRegression([xValues, yValues]);
const forecast = [];
const lastIndex = timeSeries.length - 1;
const lastDate = new Date(timeSeries[lastIndex].x);
for (let i = 1; i <= days; i++) {
const forecastDate = new Date(lastDate);
forecastDate.setDate(forecastDate.getDate() + i);
const forecastValue = regression.m * (lastIndex + i) + regression.b;
forecast.push({
x: forecastDate,
y: Math.max(0, forecastValue), // Ensure non-negative
type: 'forecast'
});
}
return forecast;
}
async analyzeSegment(segmentType, accountId, dateRange) {
// This would perform actual segment analysis
// For now, return mock data
return [
{ name: 'Segment A', value: 85, count: 1000 },
{ name: 'Segment B', value: 72, count: 800 },
{ name: 'Segment C', value: 68, count: 600 },
{ name: 'Segment D', value: 45, count: 400 },
{ name: 'Segment E', value: 32, count: 200 }
];
}
calculateDistribution(data) {
const values = data.map(d => d.value);
return {
mean: stats.mean(values),
median: stats.median(values),
stddev: stats.standardDeviation(values),
quartiles: [
stats.quantile(values, 0.25),
stats.quantile(values, 0.5),
stats.quantile(values, 0.75)
]
};
}
async saveReport(reportData) {
const report = new Report({
...reportData.metadata,
data: reportData,
status: 'completed'
});
await report.save();
return report;
}
async exportReport(report, format) {
switch (format) {
case 'pdf':
return await generatePDF(report);
case 'excel':
return await generateExcel(report);
case 'csv':
return await generateCSV(report);
case 'json':
default:
return report.data;
}
}
async scheduleReport(params) {
const {
accountId,
type,
schedule, // cron expression
recipients,
format = 'pdf'
} = params;
// This would create a scheduled job
logger.info(`Scheduled report: ${type} for ${accountId} with schedule ${schedule}`);
return {
scheduleId: `schedule_${Date.now()}`,
status: 'scheduled'
};
}
async getReportHistory(accountId, limit = 20) {
const reports = await Report.find({ accountId })
.sort({ generatedAt: -1 })
.limit(limit);
return reports.map(report => ({
reportId: report._id,
type: report.type,
generatedAt: report.generatedAt,
status: report.status,
format: report.format
}));
}
}

187
marketing-agent/services/analytics/src/utils/analytics.js Normal file
View File

@@ -0,0 +1,187 @@
import * as stats from 'simple-statistics';
export const calculateTrend = (data) => {
if (!data || data.length < 2) return 0;
// Create index array for x values
const xValues = data.map((_, i) => i);
const yValues = data;
// Calculate linear regression
const regression = stats.linearRegression([xValues, yValues]);
// Return slope as trend indicator
// Positive slope = upward trend, negative = downward trend
return regression.m;
};
export const detectAnomaly = (data, threshold = 2) => {
if (!data || data.length < 3) return false;
// Calculate mean and standard deviation
const mean = stats.mean(data);
const stdDev = stats.standardDeviation(data);
// Check if latest value is an anomaly
const latestValue = data[data.length - 1];
const zScore = Math.abs((latestValue - mean) / stdDev);
return {
isAnomaly: zScore > threshold,
zScore,
mean,
stdDev,
value: latestValue
};
};
export const calculateGrowthRate = (current, previous) => {
if (previous === 0) return current > 0 ? 100 : 0;
return ((current - previous) / previous) * 100;
};
export const calculateMovingAverage = (data, window = 7) => {
if (!data || data.length < window) return [];
const movingAverages = [];
for (let i = window - 1; i < data.length; i++) {
const windowData = data.slice(i - window + 1, i + 1);
movingAverages.push(stats.mean(windowData));
}
return movingAverages;
};
export const calculatePercentiles = (data) => {
if (!data || data.length === 0) return {};
const sorted = [...data].sort((a, b) => a - b);
return {
p50: stats.quantile(sorted, 0.5),
p75: stats.quantile(sorted, 0.75),
p90: stats.quantile(sorted, 0.9),
p95: stats.quantile(sorted, 0.95),
p99: stats.quantile(sorted, 0.99)
};
};
export const segmentData = (data, segmentSize) => {
const segments = [];
for (let i = 0; i < data.length; i += segmentSize) {
segments.push(data.slice(i, i + segmentSize));
}
return segments;
};
export const calculateSeasonality = (data, period = 7) => {
if (!data || data.length < period * 2) return null;
// Calculate average for each position in the period
const seasonalPattern = [];
for (let i = 0; i < period; i++) {
const values = [];
for (let j = i; j < data.length; j += period) {
values.push(data[j]);
}
seasonalPattern.push(stats.mean(values));
}
// Calculate seasonality strength
const overallMean = stats.mean(data);
const seasonalVariance = stats.variance(seasonalPattern);
const totalVariance = stats.variance(data);
return {
pattern: seasonalPattern,
strength: totalVariance > 0 ? seasonalVariance / totalVariance : 0,
period
};
};
export const forecastTimeSeries = (data, steps = 7) => {
if (!data || data.length < 3) return [];
// Simple linear forecast
const xValues = data.map((_, i) => i);
const yValues = data;
const regression = stats.linearRegression([xValues, yValues]);
const forecast = [];
for (let i = 0; i < steps; i++) {
const x = data.length + i;
const y = regression.m * x + regression.b;
forecast.push(Math.max(0, y)); // Ensure non-negative
}
return forecast;
};
export const calculateCorrelation = (data1, data2) => {
if (!data1 || !data2 || data1.length !== data2.length || data1.length < 2) {
return null;
}
return stats.sampleCorrelation(data1, data2);
};
export const detectOutliers = (data, method = 'iqr') => {
if (!data || data.length < 4) return [];
const sorted = [...data].sort((a, b) => a - b);
const outliers = [];
if (method === 'iqr') {
// Interquartile range method
const q1 = stats.quantile(sorted, 0.25);
const q3 = stats.quantile(sorted, 0.75);
const iqr = q3 - q1;
const lowerBound = q1 - 1.5 * iqr;
const upperBound = q3 + 1.5 * iqr;
data.forEach((value, index) => {
if (value < lowerBound || value > upperBound) {
outliers.push({ index, value, type: value < lowerBound ? 'low' : 'high' });
}
});
} else if (method === 'zscore') {
// Z-score method
const mean = stats.mean(data);
const stdDev = stats.standardDeviation(data);
data.forEach((value, index) => {
const zScore = Math.abs((value - mean) / stdDev);
if (zScore > 3) {
outliers.push({ index, value, zScore, type: value < mean ? 'low' : 'high' });
}
});
}
return outliers;
};
export const calculateCohortRetention = (cohortData) => {
// cohortData: { cohortId: { period0: count, period1: count, ... } }
const retention = {};
for (const [cohortId, periods] of Object.entries(cohortData)) {
const initialCount = periods.period0 || 0;
if (initialCount === 0) continue;
retention[cohortId] = {};
for (const [period, count] of Object.entries(periods)) {
retention[cohortId][period] = (count / initialCount) * 100;
}
}
return retention;
};

309
marketing-agent/services/analytics/src/utils/exporters.js Normal file
View File

@@ -0,0 +1,309 @@
import ExcelJS from 'exceljs';
import PDFDocument from 'pdfkit';
import { createWriteStream } from 'fs';
import { promisify } from 'util';
import path from 'path';
import { logger } from './logger.js';
export const generatePDF = async (report) => {
try {
const doc = new PDFDocument();
const filename = `report_${report.reportId}_${Date.now()}.pdf`;
const filepath = path.join(process.env.REPORTS_DIR || './reports', filename);
// Pipe to file
doc.pipe(createWriteStream(filepath));
// Add content
doc.fontSize(20).text(report.data.metadata.template, 50, 50);
doc.fontSize(12).text(`Generated: ${new Date(report.data.metadata.generatedAt).toLocaleString()}`, 50, 80);
// Add sections
let yPosition = 120;
for (const [sectionType, sectionData] of Object.entries(report.data.sections)) {
yPosition = addPDFSection(doc, sectionType, sectionData, yPosition);
// Add new page if needed
if (yPosition > 700) {
doc.addPage();
yPosition = 50;
}
}
doc.end();
return {
filename,
filepath,
url: `/reports/${filename}`
};
} catch (error) {
logger.error('PDF generation failed:', error);
throw error;
}
};
function addPDFSection(doc, sectionType, sectionData, startY) {
let y = startY;
// Section title
doc.fontSize(16).text(sectionType.replace(/_/g, ' ').toUpperCase(), 50, y);
y += 30;
// Section content based on type
switch (sectionType) {
case 'summary':
doc.fontSize(10);
doc.text(`Period: ${new Date(sectionData.period.start).toLocaleDateString()} - ${new Date(sectionData.period.end).toLocaleDateString()}`, 50, y);
y += 20;
if (sectionData.highlights) {
for (const highlight of sectionData.highlights) {
doc.text(`• ${highlight.message}`, 70, y);
y += 15;
}
}
break;
case 'metrics':
doc.fontSize(10);
for (const [metricId, metricData] of Object.entries(sectionData)) {
doc.text(`${metricData.name}: ${metricData.summary.average.toFixed(2)}`, 50, y);
y += 15;
}
break;
case 'recommendations':
doc.fontSize(10);
for (const rec of sectionData) {
doc.text(`[${rec.priority.toUpperCase()}] ${rec.title}`, 50, y);
y += 15;
doc.fontSize(9).text(rec.description, 70, y, { width: 400 });
y += 30;
}
break;
default:
// Generic content
doc.fontSize(10).text(JSON.stringify(sectionData, null, 2), 50, y, { width: 500 });
y += 100;
}
return y + 20;
}
export const generateExcel = async (report) => {
try {
const workbook = new ExcelJS.Workbook();
workbook.creator = 'Analytics Service';
workbook.created = new Date();
// Add metadata sheet
const metaSheet = workbook.addWorksheet('Report Info');
metaSheet.columns = [
{ header: 'Property', key: 'property', width: 30 },
{ header: 'Value', key: 'value', width: 50 }
];
metaSheet.addRows([
{ property: 'Report Type', value: report.data.metadata.type },
{ property: 'Generated At', value: new Date(report.data.metadata.generatedAt).toLocaleString() },
{ property: 'Account ID', value: report.data.metadata.accountId },
{ property: 'Period Start', value: new Date(report.data.metadata.dateRange.start).toLocaleDateString() },
{ property: 'Period End', value: new Date(report.data.metadata.dateRange.end).toLocaleDateString() }
]);
// Add section sheets
for (const [sectionType, sectionData] of Object.entries(report.data.sections)) {
const sheet = workbook.addWorksheet(sectionType.replace(/_/g, ' '));
addExcelSection(sheet, sectionType, sectionData);
}
// Save file
const filename = `report_${report.reportId}_${Date.now()}.xlsx`;
const filepath = path.join(process.env.REPORTS_DIR || './reports', filename);
await workbook.xlsx.writeFile(filepath);
return {
filename,
filepath,
url: `/reports/${filename}`
};
} catch (error) {
logger.error('Excel generation failed:', error);
throw error;
}
};
function addExcelSection(sheet, sectionType, sectionData) {
switch (sectionType) {
case 'metrics':
sheet.columns = [
{ header: 'Metric', key: 'metric', width: 30 },
{ header: 'Average', key: 'average', width: 15 },
{ header: 'Min', key: 'min', width: 15 },
{ header: 'Max', key: 'max', width: 15 },
{ header: 'Total', key: 'total', width: 15 }
];
for (const [metricId, metricData] of Object.entries(sectionData)) {
sheet.addRow({
metric: metricData.name,
average: metricData.summary.average,
min: metricData.summary.min,
max: metricData.summary.max,
total: metricData.summary.total
});
}
break;
case 'trends':
sheet.columns = [
{ header: 'Metric', key: 'metric', width: 30 },
{ header: 'Trend', key: 'trend', width: 15 },
{ header: 'Slope', key: 'slope', width: 15 },
{ header: 'R²', key: 'r2', width: 15 }
];
for (const [metricId, trendData] of Object.entries(sectionData)) {
sheet.addRow({
metric: trendData.name,
trend: trendData.trend.direction,
slope: trendData.trend.slope,
r2: trendData.trend.r2
});
}
break;
case 'recommendations':
sheet.columns = [
{ header: 'Priority', key: 'priority', width: 15 },
{ header: 'Category', key: 'category', width: 20 },
{ header: 'Title', key: 'title', width: 40 },
{ header: 'Description', key: 'description', width: 80 }
];
for (const rec of sectionData) {
sheet.addRow(rec);
}
break;
default:
// Generic handling
if (Array.isArray(sectionData)) {
if (sectionData.length > 0) {
const columns = Object.keys(sectionData[0]).map(key => ({
header: key,
key: key,
width: 20
}));
sheet.columns = columns;
sheet.addRows(sectionData);
}
} else {
sheet.columns = [
{ header: 'Key', key: 'key', width: 30 },
{ header: 'Value', key: 'value', width: 50 }
];
for (const [key, value] of Object.entries(sectionData)) {
sheet.addRow({ key, value: JSON.stringify(value) });
}
}
}
// Apply styling
sheet.getRow(1).font = { bold: true };
sheet.getRow(1).fill = {
type: 'pattern',
pattern: 'solid',
fgColor: { argb: 'FFE0E0E0' }
};
}
export const generateCSV = async (report) => {
try {
const lines = [];
// Add metadata
lines.push('Report Information');
lines.push(`Type,${report.data.metadata.type}`);
lines.push(`Generated,${new Date(report.data.metadata.generatedAt).toLocaleString()}`);
lines.push(`Account ID,${report.data.metadata.accountId}`);
lines.push('');
// Add sections
for (const [sectionType, sectionData] of Object.entries(report.data.sections)) {
lines.push(`\n${sectionType.replace(/_/g, ' ').toUpperCase()}`);
lines.push(...generateCSVSection(sectionType, sectionData));
lines.push('');
}
const content = lines.join('\n');
const filename = `report_${report.reportId}_${Date.now()}.csv`;
const filepath = path.join(process.env.REPORTS_DIR || './reports', filename);
await promisify(require('fs').writeFile)(filepath, content);
return {
filename,
filepath,
url: `/reports/${filename}`
};
} catch (error) {
logger.error('CSV generation failed:', error);
throw error;
}
};
function generateCSVSection(sectionType, sectionData) {
const lines = [];
switch (sectionType) {
case 'metrics':
lines.push('Metric,Average,Min,Max,Total');
for (const [metricId, metricData] of Object.entries(sectionData)) {
lines.push([
metricData.name,
metricData.summary.average,
metricData.summary.min,
metricData.summary.max,
metricData.summary.total
].join(','));
}
break;
case 'recommendations':
lines.push('Priority,Category,Title,Description');
for (const rec of sectionData) {
lines.push([
rec.priority,
rec.category,
`"${rec.title}"`,
`"${rec.description.replace(/"/g, '""')}"`
].join(','));
}
break;
default:
// Generic CSV generation
if (Array.isArray(sectionData) && sectionData.length > 0) {
const headers = Object.keys(sectionData[0]);
lines.push(headers.join(','));
for (const row of sectionData) {
const values = headers.map(h => {
const value = row[h];
return typeof value === 'string' && value.includes(',')
? `"${value}"`
: value;
});
lines.push(values.join(','));
}
}
}
return lines;
}

91
marketing-agent/services/analytics/src/utils/logger.js Normal file
View File

@@ -0,0 +1,91 @@
import winston from 'winston';
import path from 'path';
import { fileURLToPath } from 'url';
const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);
const { combine, timestamp, printf, colorize, errors } = winston.format;
// Custom log format
const logFormat = printf(({ level, message, timestamp, stack, ...metadata }) => {
let msg = `${timestamp} [${level}] ${message}`;
if (stack) {
msg += `\n${stack}`;
}
if (Object.keys(metadata).length > 0) {
msg += ` ${JSON.stringify(metadata)}`;
}
return msg;
});
// Create logger instance
export const logger = winston.createLogger({
level: process.env.LOG_LEVEL || 'info',
format: combine(
errors({ stack: true }),
timestamp({ format: 'YYYY-MM-DD HH:mm:ss' }),
logFormat
),
transports: [
// Console transport
new winston.transports.Console({
format: combine(
colorize(),
logFormat
)
}),
// File transport for errors
new winston.transports.File({
filename: path.join(__dirname, '../../logs/error.log'),
level: 'error',
maxsize: 10485760, // 10MB
maxFiles: 5
}),
// File transport for all logs
new winston.transports.File({
filename: path.join(__dirname, '../../logs/combined.log'),
maxsize: 10485760, // 10MB
maxFiles: 10
}),
// File transport for analytics specific logs
new winston.transports.File({
filename: path.join(__dirname, '../../logs/analytics.log'),
level: 'debug',
maxsize: 10485760, // 10MB
maxFiles: 5
})
],
exceptionHandlers: [
new winston.transports.File({
filename: path.join(__dirname, '../../logs/exceptions.log')
})
],
rejectionHandlers: [
new winston.transports.File({
filename: path.join(__dirname, '../../logs/rejections.log')
})
]
});
// Add metrics logging helper
export const logMetric = (metric, value, dimensions = {}) => {
logger.debug('Metric', {
metric,
value,
dimensions,
timestamp: new Date().toISOString()
});
};
// Add event logging helper
export const logEvent = (event, properties = {}) => {
logger.info('Event', {
event,
properties,
timestamp: new Date().toISOString()
});
};

113
marketing-agent/services/analytics/src/utils/metricParser.js Normal file
View File

@@ -0,0 +1,113 @@
export const parseExpression = (formula) => {
const dataPoints = new Set();
// Extract all variable names from the formula
const variablePattern = /\b[a-zA-Z_][a-zA-Z0-9_]*\b/g;
const matches = formula.match(variablePattern) || [];
// Filter out mathematical functions
const mathFunctions = ['sum', 'avg', 'min', 'max', 'count', 'sqrt', 'abs', 'round', 'floor', 'ceil'];
for (const match of matches) {
if (!mathFunctions.includes(match) && isNaN(match)) {
dataPoints.add(match);
}
}
return Array.from(dataPoints);
};
export const evaluateFormula = (formula, context) => {
// Replace variables in formula with their values from context
let evaluableFormula = formula;
for (const [variable, value] of Object.entries(context)) {
// Use word boundaries to avoid partial replacements
const regex = new RegExp(`\\b${variable}\\b`, 'g');
evaluableFormula = evaluableFormula.replace(regex, value);
}
// Check if all variables have been replaced
const remainingVariables = parseExpression(evaluableFormula);
if (remainingVariables.length > 0) {
throw new Error(`Missing values for variables: ${remainingVariables.join(', ')}`);
}
return evaluableFormula;
};
export const validateFormulaSyntax = (formula) => {
try {
// Check for balanced parentheses
let depth = 0;
for (const char of formula) {
if (char === '(') depth++;
if (char === ')') depth--;
if (depth < 0) return false;
}
if (depth !== 0) return false;
// Check for valid characters
const validPattern = /^[a-zA-Z0-9_+\-*/().\s]+$/;
if (!validPattern.test(formula)) return false;
// Check for consecutive operators
const consecutiveOps = /[+\-*/]{2,}/;
if (consecutiveOps.test(formula)) return false;
return true;
} catch (error) {
return false;
}
};
export const extractAggregations = (formula) => {
const aggregations = new Set();
// Common aggregation functions
const aggPattern = /\b(sum|avg|count|min|max|median|stddev|variance|p\d{1,2})\s*\(/g;
const matches = formula.match(aggPattern) || [];
for (const match of matches) {
const funcName = match.replace(/\s*\($/, '');
aggregations.add(funcName);
}
return Array.from(aggregations);
};
export const getDependencies = (formula) => {
const dependencies = {
dataPoints: parseExpression(formula),
aggregations: extractAggregations(formula),
timeWindow: null
};
// Check for time window references
const timePattern = /\b(last_\d+[dhm]|today|yesterday|this_week|last_week)\b/g;
const timeMatches = formula.match(timePattern);
if (timeMatches && timeMatches.length > 0) {
dependencies.timeWindow = timeMatches[0];
}
return dependencies;
};
export const optimizeFormula = (formula) => {
let optimized = formula;
// Remove unnecessary whitespace
optimized = optimized.replace(/\s+/g, ' ').trim();
// Simplify redundant parentheses
optimized = optimized.replace(/\(\s*([a-zA-Z0-9_]+)\s*\)/g, '$1');
// Convert division by constants to multiplication
optimized = optimized.replace(/\/\s*(\d+(\.\d+)?)/g, (match, num) => {
const inverse = 1 / parseFloat(num);
return ` * ${inverse}`;
});
return optimized;
};

204
marketing-agent/services/analytics/src/utils/notifications.js Normal file
View File

@@ -0,0 +1,204 @@
import axios from 'axios';
import { logger } from './logger.js';
export const sendNotification = async (channel, notification) => {
const { subject, message, alertId, metadata } = notification;
try {
switch (channel) {
case 'email':
await sendEmailNotification({ subject, message, alertId, metadata });
break;
case 'sms':
await sendSMSNotification({ message, alertId, metadata });
break;
case 'webhook':
await sendWebhookNotification({ subject, message, alertId, metadata });
break;
case 'slack':
await sendSlackNotification({ subject, message, alertId, metadata });
break;
default:
logger.warn(`Unknown notification channel: ${channel}`);
}
} catch (error) {
logger.error(`Failed to send ${channel} notification:`, error);
throw error;
}
};
async function sendEmailNotification({ subject, message, alertId, metadata }) {
// This would integrate with an email service like SendGrid, AWS SES, etc.
const emailConfig = {
to: process.env.ALERT_EMAIL_TO || 'alerts@example.com',
from: process.env.ALERT_EMAIL_FROM || 'noreply@analytics.com',
subject: subject,
text: message,
html: formatEmailHTML(message, alertId, metadata)
};
// Mock implementation
logger.info('Email notification sent:', { to: emailConfig.to, subject });
// In production, use actual email service
// await sendgrid.send(emailConfig);
}
async function sendSMSNotification({ message, alertId, metadata }) {
// This would integrate with an SMS service like Twilio
const smsConfig = {
to: process.env.ALERT_SMS_TO || '+1234567890',
from: process.env.ALERT_SMS_FROM || '+0987654321',
body: `${message.substring(0, 140)}... Alert ID: ${alertId}`
};
// Mock implementation
logger.info('SMS notification sent:', { to: smsConfig.to });
// In production, use actual SMS service
// await twilio.messages.create(smsConfig);
}
async function sendWebhookNotification({ subject, message, alertId, metadata }) {
const webhookUrl = process.env.ALERT_WEBHOOK_URL;
if (!webhookUrl) {
logger.warn('No webhook URL configured');
return;
}
const payload = {
alertId,
subject,
message,
metadata,
timestamp: new Date().toISOString()
};
try {
const response = await axios.post(webhookUrl, payload, {
headers: {
'Content-Type': 'application/json',
'X-Alert-ID': alertId
},
timeout: 10000
});
logger.info('Webhook notification sent:', { url: webhookUrl, status: response.status });
} catch (error) {
logger.error('Webhook notification failed:', error.message);
throw error;
}
}
async function sendSlackNotification({ subject, message, alertId, metadata }) {
const slackWebhookUrl = process.env.SLACK_WEBHOOK_URL;
if (!slackWebhookUrl) {
logger.warn('No Slack webhook URL configured');
return;
}
const payload = {
text: subject,
attachments: [
{
color: getSlackColor(metadata.severity),
fields: [
{
title: 'Alert ID',
value: alertId,
short: true
},
{
title: 'Metric',
value: metadata.metric,
short: true
},
{
title: 'Current Value',
value: metadata.value,
short: true
},
{
title: 'Threshold',
value: metadata.threshold,
short: true
}
],
text: message,
ts: Math.floor(Date.now() / 1000)
}
]
};
try {
const response = await axios.post(slackWebhookUrl, payload);
logger.info('Slack notification sent');
} catch (error) {
logger.error('Slack notification failed:', error.message);
throw error;
}
}
function formatEmailHTML(message, alertId, metadata) {
return `
<!DOCTYPE html>
<html>
<head>
<style>
body { font-family: Arial, sans-serif; }
.alert-box { border: 1px solid #ddd; padding: 20px; margin: 20px 0; }
.severity-critical { border-color: #d32f2f; }
.severity-high { border-color: #f57c00; }
.severity-medium { border-color: #fbc02d; }
.severity-low { border-color: #388e3c; }
.metadata { background: #f5f5f5; padding: 10px; margin-top: 20px; }
.metadata dt { font-weight: bold; }
.metadata dd { margin-left: 20px; margin-bottom: 10px; }
</style>
</head>
<body>
<div class="alert-box severity-${metadata.severity || 'medium'}">
<h2>Analytics Alert</h2>
<p>${message.replace(/\n/g, '<br>')}</p>
<div class="metadata">
<h3>Alert Details</h3>
<dl>
<dt>Alert ID:</dt>
<dd>${alertId}</dd>
<dt>Metric:</dt>
<dd>${metadata.metric || 'N/A'}</dd>
<dt>Current Value:</dt>
<dd>${metadata.value || 'N/A'}</dd>
<dt>Threshold:</dt>
<dd>${metadata.threshold || 'N/A'}</dd>
<dt>Severity:</dt>
<dd>${metadata.severity || 'N/A'}</dd>
</dl>
</div>
</div>
</body>
</html>
`.trim();
}
function getSlackColor(severity) {
const colors = {
critical: '#d32f2f',
high: '#f57c00',
medium: '#fbc02d',
low: '#388e3c'
};
return colors[severity] || '#757575';
}

152
marketing-agent/services/analytics/src/utils/validators.js Normal file
View File

@@ -0,0 +1,152 @@
export const validateEvent = (event) => {
const errors = [];
// Required fields
if (!event.type) {
errors.push('Event type is required');
}
if (!event.accountId) {
errors.push('Account ID is required');
}
if (!event.action) {
errors.push('Action is required');
}
// Type validation
if (event.value !== undefined && typeof event.value !== 'number') {
errors.push('Value must be a number');
}
// Validate event type format
if (event.type && !/^[a-z_]+$/.test(event.type)) {
errors.push('Event type must be lowercase with underscores only');
}
// Validate action format
if (event.action && !/^[a-z_]+$/.test(event.action)) {
errors.push('Action must be lowercase with underscores only');
}
return {
valid: errors.length === 0,
errors
};
};
export const validateMetricFormula = (formula) => {
const errors = [];
// Check for basic syntax
try {
// Simple validation - check for balanced parentheses
let depth = 0;
for (const char of formula) {
if (char === '(') depth++;
if (char === ')') depth--;
if (depth < 0) {
errors.push('Unbalanced parentheses');
break;
}
}
if (depth !== 0) {
errors.push('Unbalanced parentheses');
}
// Check for valid operators
const validOperators = ['+', '-', '*', '/', '(', ')', ' '];
const invalidChars = formula.split('').filter(char => {
return !validOperators.includes(char) &&
!/[a-zA-Z0-9_.]/.test(char);
});
if (invalidChars.length > 0) {
errors.push(`Invalid characters: ${invalidChars.join(', ')}`);
}
} catch (error) {
errors.push(`Formula validation error: ${error.message}`);
}
return {
valid: errors.length === 0,
errors
};
};
export const validateDateRange = (start, end) => {
const errors = [];
const startDate = new Date(start);
const endDate = new Date(end);
if (isNaN(startDate.getTime())) {
errors.push('Invalid start date');
}
if (isNaN(endDate.getTime())) {
errors.push('Invalid end date');
}
if (startDate >= endDate) {
errors.push('Start date must be before end date');
}
// Check for reasonable date range (max 1 year)
const maxRange = 365 * 24 * 60 * 60 * 1000; // 1 year in milliseconds
if (endDate - startDate > maxRange) {
errors.push('Date range cannot exceed 1 year');
}
return {
valid: errors.length === 0,
errors
};
};
export const validateReportType = (type) => {
const validTypes = [
'campaign_performance',
'user_analytics',
'ab_test',
'engagement_analysis',
'conversion_funnel',
'retention_cohort'
];
return {
valid: validTypes.includes(type),
errors: validTypes.includes(type) ? [] : [`Invalid report type: ${type}`]
};
};
export const validateAlertCondition = (condition) => {
const errors = [];
const validOperators = ['>', '>=', '<', '<=', '=', '==', '!='];
if (!condition.operator) {
errors.push('Operator is required');
} else if (!validOperators.includes(condition.operator)) {
errors.push(`Invalid operator: ${condition.operator}`);
}
if (condition.threshold === undefined || condition.threshold === null) {
errors.push('Threshold is required');
} else if (typeof condition.threshold !== 'number') {
errors.push('Threshold must be a number');
}
if (condition.duration !== undefined) {
if (typeof condition.duration !== 'number') {
errors.push('Duration must be a number');
} else if (condition.duration < 0) {
errors.push('Duration must be non-negative');
}
}
return {
valid: errors.length === 0,
errors
};
};

102
marketing-agent/services/analytics/src/utils/websocket.js Normal file
View File

@@ -0,0 +1,102 @@
import { logger } from './logger.js';
export class WebSocketManager {
constructor() {
this.clients = new Map();
this.rooms = new Map();
}
static getInstance() {
if (!WebSocketManager.instance) {
WebSocketManager.instance = new WebSocketManager();
}
return WebSocketManager.instance;
}
addClient(clientId, ws) {
this.clients.set(clientId, ws);
logger.info(`WebSocket client connected: ${clientId}`);
ws.on('close', () => {
this.removeClient(clientId);
});
}
removeClient(clientId) {
this.clients.delete(clientId);
// Remove from all rooms
for (const [roomId, members] of this.rooms.entries()) {
members.delete(clientId);
if (members.size === 0) {
this.rooms.delete(roomId);
}
}
logger.info(`WebSocket client disconnected: ${clientId}`);
}
joinRoom(clientId, roomId) {
if (!this.rooms.has(roomId)) {
this.rooms.set(roomId, new Set());
}
this.rooms.get(roomId).add(clientId);
logger.debug(`Client ${clientId} joined room ${roomId}`);
}
leaveRoom(clientId, roomId) {
const room = this.rooms.get(roomId);
if (room) {
room.delete(clientId);
if (room.size === 0) {
this.rooms.delete(roomId);
}
}
logger.debug(`Client ${clientId} left room ${roomId}`);
}
sendToClient(clientId, data) {
const client = this.clients.get(clientId);
if (client && client.readyState === 1) { // WebSocket.OPEN
try {
client.send(JSON.stringify(data));
} catch (error) {
logger.error(`Failed to send to client ${clientId}:`, error);
}
}
}
sendToRoom(roomId, data, excludeClientId = null) {
const room = this.rooms.get(roomId);
if (!room) return;
for (const clientId of room) {
if (clientId !== excludeClientId) {
this.sendToClient(clientId, data);
}
}
}
broadcast(data, excludeClientId = null) {
for (const [clientId, client] of this.clients.entries()) {
if (clientId !== excludeClientId) {
this.sendToClient(clientId, data);
}
}
}
getClientCount() {
return this.clients.size;
}
getRoomCount() {
return this.rooms.size;
}
getRoomMembers(roomId) {
const room = this.rooms.get(roomId);
return room ? Array.from(room) : [];
}
}