diff --git a/.env.example b/.env.example new file mode 100644 index 00000000..cdbf6f1b --- /dev/null +++ b/.env.example @@ -0,0 +1,17 @@ +# Environment Configuration +NODE_ENV=development +PORT=3000 + +# Contract Configuration +CONTRACT_ID=CB7PSJZALNWNX7NLOAM6LOEL4OJZMFPQZJMIYO522ZSACYWXTZIDEDSS + +# Logging Configuration +LOG_LEVEL=info + +# Security Configuration +# Maximum request body size (e.g., '1mb', '10mb', '100kb') +# Default: '1mb' +MAX_BODY_SIZE=1mb + +# OpenTelemetry Configuration +OTEL_SERVICE_NAME=equipchain-api diff --git a/.eslintrc.json b/.eslintrc.json new file mode 100644 index 00000000..4fdd15ab --- /dev/null +++ b/.eslintrc.json @@ -0,0 +1,29 @@ +{ + "env": { + "node": true, + "es2021": true + }, + "extends": "eslint:recommended", + "parserOptions": { + "ecmaVersion": "latest", + "sourceType": "module" + }, + "plugins": [ + "security" + ], + "rules": { + "security/detect-object-injection": "warn", + "security/detect-non-literal-fs-filename": "warn", + "security/detect-non-literal-regexp": "warn", + "security/detect-non-literal-require": "warn", + "security/detect-unsafe-regex": "error", + "security/detect-buffer-noassert": "error", + "security/detect-child-process": "warn", + "security/detect-disable-mustache-escape": "error", + "security/detect-eval-with-expression": "error", + "security/detect-no-csrf-before-method-override": "warn", + "security/detect-possible-timing-attacks": "warn", + "security/detect-pseudoRandomBytes": "warn", + "no-console": "off" + } +} diff --git a/README.md b/README.md index 97e6c36f..c5cf4d95 100644 --- a/README.md +++ b/README.md @@ -229,8 +229,60 @@ Same parameters as daily-summary, returns monthly rollups. | Method | Path | Description | Auth | |--------|------|-------------|------| -| `GET` | `/exports/meters` | Export meter data (CSV/JSON) | Yes | -| `GET` | `/exports/readings` | Export readings report | Yes | +| `GET` | `/api/exports/readings` | Export meter readings (CSV/JSON/NDJSON) | Yes | +| `GET` | `/api/exports/analytics/:summaryType` | Export analytics summaries (daily/weekly/monthly) | Yes | +| `GET` | `/api/exports/system-report` | Export system-wide report (meters, readings, alerts) | Admin | +| `GET` | `/api/exports/meters` | Export meter registry | Yes | + +#### Export Query Parameters + +All export endpoints support the following query parameters: + +- `format` - Output format: `csv`, `json`, or `ndjson` (default: `csv`) +- `fields` - Comma-separated list of fields to include (e.g., `id,timestamp,value`) +- `startDate` - Filter by start date (ISO format: `2026-01-01` or `2026-01-01T00:00:00Z`) +- `endDate` - Filter by end date (ISO format) +- `pretty` - Set to `true` for pretty-printed JSON (default: `false`) + +Additional parameters specific to endpoints: + +- `/api/exports/readings`: `meterIds` (comma-separated), `status` +- `/api/exports/analytics/:summaryType`: Date range filtering for daily summaries +- `/api/exports/system-report`: `sections` (comma-separated: `meters,readings,alerts,summary`) +- `/api/exports/meters`: `status`, `location` + +#### Export Examples + +```bash +# Export all readings as CSV +curl -H "Authorization: Bearer YOUR_TOKEN" \ + "http://localhost:3000/api/exports/readings?format=csv" + +# Export specific fields as JSON +curl -H "Authorization: Bearer YOUR_TOKEN" \ + "http://localhost:3000/api/exports/readings?format=json&fields=id,timestamp,value" + +# Export readings for specific meters and date range +curl -H "Authorization: Bearer YOUR_TOKEN" \ + "http://localhost:3000/api/exports/readings?format=csv&meterIds=meter-001,meter-002&startDate=2026-01-01&endDate=2026-06-01" + +# Export daily analytics as NDJSON (streaming) +curl -H "Authorization: Bearer YOUR_TOKEN" \ + "http://localhost:3000/api/exports/analytics/daily?format=ndjson" + +# Export system report (admin only) +curl -H "Authorization: Bearer YOUR_TOKEN" \ + -H "x-role: admin" \ + "http://localhost:3000/api/exports/system-report?format=json§ions=meters,summary" +``` + +#### Streaming Support + +All export endpoints use streaming to handle large datasets efficiently: +- Responses use `Transfer-Encoding: chunked` +- Data is streamed row-by-row for CSV and line-by-line for NDJSON +- Memory usage remains constant regardless of dataset size +- Suitable for exporting 10,000+ records ### Webhooks (Planned) diff --git a/package.json b/package.json index b5e23392..f75b5316 100644 --- a/package.json +++ b/package.json @@ -19,13 +19,15 @@ "@opentelemetry/exporter-trace-otlp-http": "^0.221.0", "@opentelemetry/sdk-node": "^0.221.0", "cors": "^2.8.6", + "csv-stringify": "^6.5.1", "dotenv": "^17.3.1", + "escape-html": "^1.0.3", "express": "^5.2.1", - "jsonwebtoken": "^9.0.3", - "pino": "^10.3.1", - "zod": "^4.4.3" + "helmet": "^8.0.0", + "pino": "^10.3.1" }, "devDependencies": { + "eslint-plugin-security": "^3.0.1", "pino-pretty": "^13.1.3" } } \ No newline at end of file diff --git a/src/app.js b/src/app.js new file mode 100644 index 00000000..d1bf8182 --- /dev/null +++ b/src/app.js @@ -0,0 +1,101 @@ +const crypto = require('crypto'); +const express = require('express'); +const cors = require('cors'); +const helmet = require('helmet'); +const { trace } = require('@opentelemetry/api'); +const { childLogger } = require('./config/logger'); +const config = require('./config'); +const routes = require('./routes'); +const { sanitizeForLogging, sanitize } = require('./utils/sanitize'); + +const app = express(); +const log = childLogger('http'); + +// Security middleware +app.use(helmet()); +app.use(cors()); + +// Ensure Content-Type is application/json for all API responses +app.use((req, res, next) => { + const originalJson = res.json; + res.json = function (data) { + if (!res.headersSent) { + res.setHeader('Content-Type', 'application/json'); + } + return originalJson.call(this, data); + }; + next(); +}); + +// Body parsing middleware with size limits +app.use(express.json({ limit: config.maxBodySize })); +app.use(express.urlencoded({ extended: true, limit: config.maxBodySize })); + +// Correlation ID and request logging middleware +app.use((req, res, next) => { + const correlationId = req.headers['x-correlation-id'] || crypto.randomUUID(); + req.correlationId = correlationId; + res.setHeader('x-correlation-id', correlationId); + + const activeSpan = trace.getActiveSpan(); + if (activeSpan) { + activeSpan.setAttribute('correlation.id', correlationId); + } + + const start = process.hrtime.bigint(); + + res.on('finish', () => { + const durationMs = Number(process.hrtime.bigint() - start) / 1e6; + log.info( + { + correlationId, + method: req.method, + url: sanitizeForLogging(req.originalUrl), + status: res.statusCode, + durationMs, + }, + 'request completed' + ); + }); + + next(); +}); + +// Mount routes +app.use('/', routes); + +// Root route with project info +app.get('/', (req, res) => { + res.json({ + project: 'Equipchain', + status: 'Monitoring Meters', + contract: config.contractId, + }); +}); + +// 404 handler +app.use((req, res) => { + res.status(404).json({ + error: 'Not Found', + message: sanitize(`Cannot ${req.method} ${req.originalUrl}`), + }); +}); + +// Error handling middleware +app.use((err, req, res, next) => { + log.error( + { + correlationId: req.correlationId, + error: sanitizeForLogging(err.message), + stack: err.stack, + }, + 'request error' + ); + + res.status(err.status || 500).json({ + error: err.name || 'Internal Server Error', + message: config.isProduction ? 'An error occurred' : sanitize(err.message), + }); +}); + +module.exports = app; diff --git a/src/config/index.js b/src/config/index.js new file mode 100644 index 00000000..e4773d6f --- /dev/null +++ b/src/config/index.js @@ -0,0 +1,35 @@ +require('dotenv').config(); + +const { + NODE_ENV = 'development', + PORT = '3000', + CONTRACT_ID = 'CB7PSJZALNWNX7NLOAM6LOEL4OJZMFPQZJMIYO522ZSACYWXTZIDEDSS', + LOG_LEVEL = 'info', + OTEL_SERVICE_NAME = 'equipchain-api', + MAX_BODY_SIZE = '1mb', +} = process.env; + +// Validate required environment variables for production +const isProduction = NODE_ENV === 'production'; +const isTest = NODE_ENV === 'test'; + +if (isProduction) { + // Add any production-specific required variables here + // For example: JWT_SECRET, DATABASE_URL, etc. + // Currently no additional required variables for this project +} + +const config = Object.freeze({ + env: NODE_ENV, + port: parseInt(PORT, 10), + contractId: CONTRACT_ID, + logLevel: LOG_LEVEL, + maxBodySize: MAX_BODY_SIZE, + otel: { + serviceName: OTEL_SERVICE_NAME, + }, + isProduction, + isTest, +}); + +module.exports = config; diff --git a/src/config/logger.js b/src/config/logger.js index 26c83a60..29c2e66b 100644 --- a/src/config/logger.js +++ b/src/config/logger.js @@ -5,6 +5,20 @@ const isTest = process.env.NODE_ENV === 'test'; const logger = pino({ level: process.env.LOG_LEVEL || 'info', + // Redact sensitive fields to prevent log injection and protect sensitive data + redact: { + paths: [ + 'password', + 'token', + 'apiKey', + 'secret', + 'authorization', + 'cookie', + 'req.headers.authorization', + 'req.headers.cookie', + ], + remove: true, + }, transport: isProduction || isTest ? undefined diff --git a/src/jobs/billing.job.js b/src/jobs/billing.job.js new file mode 100644 index 00000000..6a1c446d --- /dev/null +++ b/src/jobs/billing.job.js @@ -0,0 +1,42 @@ +const { childLogger } = require('../config/logger'); + +const log = childLogger('job:billing'); + +/** + * Billing job handler + * Aggregates meter readings and computes charges + * + * @param {Object} data - Job data + * @param {string} data.period - Billing period (e.g., "2024-01") + * @param {string} data.accountId - Account ID to bill (optional, null for all accounts) + * @returns {Object} Billing results + */ +async function billingHandler(data) { + const { period, accountId } = data; + + log.info({ period, accountId }, 'Starting billing job'); + + // TODO: Implement actual billing logic + // 1. Fetch meter readings for the period + // 2. Aggregate readings by account + // 3. Compute charges based on rates + // 4. Generate invoices + // 5. Store results in database + + // Placeholder implementation + await new Promise(resolve => setTimeout(resolve, 1000)); + + const result = { + period, + accountId: accountId || 'all', + invoicesGenerated: Math.floor(Math.random() * 100), + totalAmount: (Math.random() * 10000).toFixed(2), + processedAt: new Date().toISOString(), + }; + + log.info({ result }, 'Billing job completed'); + + return result; +} + +module.exports = billingHandler; diff --git a/src/jobs/cacheWarm.job.js b/src/jobs/cacheWarm.job.js new file mode 100644 index 00000000..868f4915 --- /dev/null +++ b/src/jobs/cacheWarm.job.js @@ -0,0 +1,41 @@ +const { childLogger } = require('../config/logger'); + +const log = childLogger('job:cacheWarm'); + +/** + * Cache warm job handler + * Warms cache for frequently accessed data + * + * @param {Object} data - Job data + * @param {string} data.cacheType - Type of cache to warm ('meter_data', 'user_profiles', 'contract_state') + * @param {Array} data.keys - Specific cache keys to warm (optional) + * @returns {Object} Cache warming results + */ +async function cacheWarmHandler(data) { + const { cacheType, keys } = data; + + log.info({ cacheType, keys }, 'Starting cache warm job'); + + // TODO: Implement actual cache warming logic + // 1. Identify frequently accessed data patterns + // 2. Fetch data from database or blockchain + // 3. Populate cache with pre-fetched data + // 4. Set appropriate TTL values + // 5. Monitor cache hit rates + + // Placeholder implementation + await new Promise(resolve => setTimeout(resolve, 300)); + + const result = { + cacheType, + keysWarmed: keys?.length || Math.floor(Math.random() * 50), + cacheSize: (Math.random() * 10).toFixed(2) + 'MB', + warmedAt: new Date().toISOString(), + }; + + log.info({ result }, 'Cache warm job completed'); + + return result; +} + +module.exports = cacheWarmHandler; diff --git a/src/jobs/reports.job.js b/src/jobs/reports.job.js new file mode 100644 index 00000000..873ccb60 --- /dev/null +++ b/src/jobs/reports.job.js @@ -0,0 +1,43 @@ +const { childLogger } = require('../config/logger'); + +const log = childLogger('job:reports'); + +/** + * Reports job handler + * Generates daily/monthly usage reports + * + * @param {Object} data - Job data + * @param {string} data.type - Report type ('daily' or 'monthly') + * @param {string} data.date - Date for the report (YYYY-MM-DD) + * @param {string} data.format - Output format ('json', 'csv', 'pdf') + * @returns {Object} Report generation results + */ +async function reportsHandler(data) { + const { type, date, format = 'json' } = data; + + log.info({ type, date, format }, 'Starting reports job'); + + // TODO: Implement actual report generation logic + // 1. Fetch usage data for the specified period + // 2. Aggregate and analyze data + // 3. Generate report in requested format + // 4. Store or deliver the report + + // Placeholder implementation + await new Promise(resolve => setTimeout(resolve, 1500)); + + const result = { + type, + date, + format, + reportId: `report_${Date.now()}`, + recordsProcessed: Math.floor(Math.random() * 1000), + generatedAt: new Date().toISOString(), + }; + + log.info({ result }, 'Reports job completed'); + + return result; +} + +module.exports = reportsHandler; diff --git a/src/jobs/sync.job.js b/src/jobs/sync.job.js new file mode 100644 index 00000000..2c69c17a --- /dev/null +++ b/src/jobs/sync.job.js @@ -0,0 +1,45 @@ +const { childLogger } = require('../config/logger'); + +const log = childLogger('job:sync'); + +/** + * Sync job handler + * Syncs on-chain data with local state + * + * @param {Object} data - Job data + * @param {string} data.syncType - Type of sync ('transactions', 'contract_state', 'full') + * @param {number} data.fromBlock - Starting block number (optional) + * @param {number} data.toBlock - Ending block number (optional) + * @returns {Object} Sync results + */ +async function syncHandler(data) { + const { syncType, fromBlock, toBlock } = data; + + log.info({ syncType, fromBlock, toBlock }, 'Starting sync job'); + + // TODO: Implement actual sync logic + // 1. Connect to Soroban blockchain + // 2. Fetch transactions or contract state + // 3. Verify transaction statuses + // 4. Update local database with on-chain data + // 5. Handle conflicts and reconciliation + + // Placeholder implementation + await new Promise(resolve => setTimeout(resolve, 2000)); + + const result = { + syncType, + fromBlock: fromBlock || 'latest', + toBlock: toBlock || 'latest', + blocksSynced: Math.floor(Math.random() * 100), + transactionsProcessed: Math.floor(Math.random() * 500), + stateUpdated: true, + syncedAt: new Date().toISOString(), + }; + + log.info({ result }, 'Sync job completed'); + + return result; +} + +module.exports = syncHandler; diff --git a/src/jobs/webhookRetry.job.js b/src/jobs/webhookRetry.job.js new file mode 100644 index 00000000..35c51226 --- /dev/null +++ b/src/jobs/webhookRetry.job.js @@ -0,0 +1,52 @@ +const { childLogger } = require('../config/logger'); + +const log = childLogger('job:webhookRetry'); + +/** + * Webhook retry job handler + * Retries failed webhook deliveries + * + * @param {Object} data - Job data + * @param {string} data.webhookId - Webhook delivery ID + * @param {string} data.url - Target URL + * @param {Object} data.payload - Webhook payload + * @param {number} data.attempt - Current attempt number + * @returns {Object} Delivery result + */ +async function webhookRetryHandler(data) { + const { webhookId, url, payload, attempt } = data; + + log.info({ webhookId, url, attempt }, 'Starting webhook retry job'); + + // TODO: Implement actual webhook delivery logic + // 1. Prepare HTTP request with payload + // 2. Send POST request to webhook URL + // 3. Handle response and retry logic + // 4. Update delivery status in database + // 5. Apply exponential backoff on failure + + // Placeholder implementation + await new Promise(resolve => setTimeout(resolve, 500)); + + // Simulate occasional failure for testing + const shouldFail = Math.random() < 0.3; + + if (shouldFail) { + throw new Error(`Webhook delivery failed: ${url} returned 500`); + } + + const result = { + webhookId, + url, + attempt, + success: true, + statusCode: 200, + deliveredAt: new Date().toISOString(), + }; + + log.info({ result }, 'Webhook retry job completed'); + + return result; +} + +module.exports = webhookRetryHandler; diff --git a/src/routes/exports.js b/src/routes/exports.js new file mode 100644 index 00000000..5c3e967e --- /dev/null +++ b/src/routes/exports.js @@ -0,0 +1,377 @@ +const express = require('express'); +const router = express.Router(); +const { handleExport } = require('../services/exporter'); +const { childLogger } = require('../config/logger'); + +const log = childLogger('routes:exports'); + +/** + * Mock data for meter readings + * In production, this would come from a repository/service layer + */ +const mockMeterReadings = [ + { + id: 'reading-001', + meterId: 'meter-001', + timestamp: '2026-01-15T08:00:00Z', + value: 1234.56, + unit: 'kWh', + status: 'verified', + }, + { + id: 'reading-002', + meterId: 'meter-001', + timestamp: '2026-01-15T09:00:00Z', + value: 1245.78, + unit: 'kWh', + status: 'verified', + }, + { + id: 'reading-003', + meterId: 'meter-002', + timestamp: '2026-01-15T08:00:00Z', + value: 987.65, + unit: 'kWh', + status: 'pending', + }, +]; + +/** + * Mock data for analytics summaries + */ +const mockAnalyticsData = { + daily: [ + { + date: '2026-01-15', + totalConsumption: 3456.78, + averageConsumption: 1152.26, + peakConsumption: 1245.78, + meterCount: 3, + activeAlerts: 0, + }, + { + date: '2026-01-16', + totalConsumption: 3678.90, + averageConsumption: 1226.30, + peakConsumption: 1345.67, + meterCount: 3, + activeAlerts: 1, + }, + ], + weekly: [ + { + weekStart: '2026-01-13', + weekEnd: '2026-01-19', + totalConsumption: 24567.89, + averageDailyConsumption: 3509.70, + peakDay: '2026-01-16', + meterCount: 3, + activeAlerts: 3, + }, + ], + monthly: [ + { + month: '2026-01', + totalConsumption: 98765.43, + averageDailyConsumption: 3185.98, + peakDay: '2026-01-25', + meterCount: 3, + activeAlerts: 8, + }, + ], +}; + +/** + * Mock data for system report + */ +const mockSystemReport = { + meters: [ + { + id: 'meter-001', + name: 'Main Building Meter', + location: 'Building A', + status: 'online', + lastReading: '2026-01-15T09:00:00Z', + totalReadings: 1523, + }, + { + id: 'meter-002', + name: 'Auxiliary Meter', + location: 'Building B', + status: 'online', + lastReading: '2026-01-15T08:00:00Z', + totalReadings: 987, + }, + ], + readings: mockMeterReadings, + alerts: [ + { + id: 'alert-001', + type: 'anomaly', + severity: 'warning', + message: 'Unusual consumption pattern detected', + meterId: 'meter-001', + timestamp: '2026-01-15T10:30:00Z', + resolved: false, + }, + ], + summary: { + totalMeters: 2, + onlineMeters: 2, + offlineMeters: 0, + totalReadings: 2510, + activeAlerts: 1, + reportGenerated: '2026-01-15T12:00:00Z', + }, +}; + +/** + * Available fields for each export type + */ +const AVAILABLE_FIELDS = { + readings: ['id', 'meterId', 'timestamp', 'value', 'unit', 'status'], + analytics: ['date', 'weekStart', 'weekEnd', 'month', 'totalConsumption', 'averageConsumption', 'averageDailyConsumption', 'peakConsumption', 'peakDay', 'meterCount', 'activeAlerts'], + meters: ['id', 'name', 'location', 'status', 'lastReading', 'totalReadings'], + alerts: ['id', 'type', 'severity', 'message', 'meterId', 'timestamp', 'resolved'], + summary: ['totalMeters', 'onlineMeters', 'offlineMeters', 'totalReadings', 'activeAlerts', 'reportGenerated'], +}; + +/** + * Authentication middleware placeholder + * In production, this would verify JWT tokens or API keys + */ +function authenticate(req, res, next) { + const authHeader = req.headers.authorization; + + if (!authHeader) { + return res.status(401).json({ error: 'Unauthorized', message: 'Authentication required' }); + } + + // TODO: Implement actual JWT/API key verification + // For now, we'll accept any Bearer token + if (authHeader.startsWith('Bearer ')) { + next(); + } else { + res.status(401).json({ error: 'Unauthorized', message: 'Invalid authentication format' }); + } +} + +/** + * Admin authorization middleware placeholder + */ +function requireAdmin(req, res, next) { + // TODO: Implement actual admin role verification + // For now, we'll check for a custom header + if (req.headers['x-role'] === 'admin') { + next(); + } else { + res.status(403).json({ error: 'Forbidden', message: 'Admin access required' }); + } +} + +/** + * GET /api/exports/readings + * Export meter readings with filtering options + */ +router.get('/readings', authenticate, async (req, res) => { + try { + log.info({ query: req.query }, 'Export readings request'); + + // Apply filters (mock implementation) + let filteredData = [...mockMeterReadings]; + + // Filter by meter IDs + if (req.query.meterIds) { + const meterIds = req.query.meterIds.split(',').map(id => id.trim()); + filteredData = filteredData.filter(reading => meterIds.includes(reading.meterId)); + } + + // Filter by date range + if (req.query.startDate) { + const startDate = new Date(req.query.startDate); + filteredData = filteredData.filter(reading => new Date(reading.timestamp) >= startDate); + } + + if (req.query.endDate) { + const endDate = new Date(req.query.endDate); + filteredData = filteredData.filter(reading => new Date(reading.timestamp) <= endDate); + } + + // Filter by status + if (req.query.status) { + filteredData = filteredData.filter(reading => reading.status === req.query.status); + } + + log.info({ recordCount: filteredData.length }, 'Exporting readings'); + + await handleExport(req, res, filteredData, AVAILABLE_FIELDS.readings, 'meter-readings'); + } catch (error) { + log.error({ error }, 'Export readings error'); + if (!res.headersSent) { + res.status(500).json({ error: 'Export failed', message: error.message }); + } + } +}); + +/** + * GET /api/exports/analytics/:summaryType + * Export analytics summaries by type (daily, weekly, monthly) + */ +router.get('/analytics/:summaryType', authenticate, async (req, res) => { + try { + const { summaryType } = req.params; + + log.info({ summaryType, query: req.query }, 'Export analytics request'); + + // Validate summary type + const validTypes = ['daily', 'weekly', 'monthly']; + if (!validTypes.includes(summaryType)) { + return res.status(400).json({ + error: 'Invalid summary type', + message: `Valid types: ${validTypes.join(', ')}`, + }); + } + + const data = mockAnalyticsData[summaryType] || []; + + // Apply date range filters if applicable + let filteredData = [...data]; + if (req.query.startDate && summaryType === 'daily') { + const startDate = req.query.startDate; + filteredData = filteredData.filter(item => item.date >= startDate); + } + if (req.query.endDate && summaryType === 'daily') { + const endDate = req.query.endDate; + filteredData = filteredData.filter(item => item.date <= endDate); + } + + log.info({ summaryType, recordCount: filteredData.length }, 'Exporting analytics'); + + await handleExport(req, res, filteredData, AVAILABLE_FIELDS.analytics, `analytics-${summaryType}`); + } catch (error) { + log.error({ error }, 'Export analytics error'); + if (!res.headersSent) { + res.status(500).json({ error: 'Export failed', message: error.message }); + } + } +}); + +/** + * GET /api/exports/system-report + * Export system-wide report combining meters, readings, and alerts + */ +router.get('/system-report', authenticate, requireAdmin, async (req, res) => { + try { + log.info({ query: req.query }, 'Export system report request'); + + // Determine which sections to include + const sections = req.query.sections ? req.query.sections.split(',').map(s => s.trim()) : ['meters', 'readings', 'alerts', 'summary']; + + // Build report data based on requested sections + const reportData = {}; + let allFields = []; + + if (sections.includes('meters')) { + reportData.meters = mockSystemReport.meters; + allFields = allFields.concat(AVAILABLE_FIELDS.meters); + } + + if (sections.includes('readings')) { + reportData.readings = mockSystemReport.readings; + allFields = allFields.concat(AVAILABLE_FIELDS.readings); + } + + if (sections.includes('alerts')) { + reportData.alerts = mockSystemReport.alerts; + allFields = allFields.concat(AVAILABLE_FIELDS.alerts); + } + + if (sections.includes('summary')) { + reportData.summary = mockSystemReport.summary; + allFields = allFields.concat(AVAILABLE_FIELDS.summary); + } + + // Flatten the report for CSV export + let exportData; + if (req.query.format === 'csv') { + // For CSV, we need to flatten the structure + // This is a simplified approach - in production, you might want separate CSV files per section + exportData = []; + + if (reportData.meters) { + reportData.meters.forEach(item => { + exportData.push({ ...item, _section: 'meters' }); + }); + } + + if (reportData.readings) { + reportData.readings.forEach(item => { + exportData.push({ ...item, _section: 'readings' }); + }); + } + + if (reportData.alerts) { + reportData.alerts.forEach(item => { + exportData.push({ ...item, _section: 'alerts' }); + }); + } + + if (reportData.summary) { + exportData.push({ ...reportData.summary, _section: 'summary' }); + } + + allFields.push('_section'); + } else { + // For JSON, keep the nested structure + exportData = reportData; + allFields = AVAILABLE_FIELDS.meters.concat( + AVAILABLE_FIELDS.readings, + AVAILABLE_FIELDS.alerts, + AVAILABLE_FIELDS.summary + ); + } + + log.info({ sections, recordCount: Array.isArray(exportData) ? exportData.length : 1 }, 'Exporting system report'); + + await handleExport(req, res, exportData, allFields, 'system-report'); + } catch (error) { + log.error({ error }, 'Export system report error'); + if (!res.headersSent) { + res.status(500).json({ error: 'Export failed', message: error.message }); + } + } +}); + +/** + * GET /api/exports/meters + * Export meter registry + */ +router.get('/meters', authenticate, async (req, res) => { + try { + log.info({ query: req.query }, 'Export meters request'); + + const data = mockSystemReport.meters; + + // Apply filters + let filteredData = [...data]; + + if (req.query.status) { + filteredData = filteredData.filter(meter => meter.status === req.query.status); + } + + if (req.query.location) { + filteredData = filteredData.filter(meter => meter.location === req.query.location); + } + + log.info({ recordCount: filteredData.length }, 'Exporting meters'); + + await handleExport(req, res, filteredData, AVAILABLE_FIELDS.meters, 'meters'); + } catch (error) { + log.error({ error }, 'Export meters error'); + if (!res.headersSent) { + res.status(500).json({ error: 'Export failed', message: error.message }); + } + } +}); + +module.exports = router; diff --git a/src/routes/index.js b/src/routes/index.js new file mode 100644 index 00000000..20ac8c46 --- /dev/null +++ b/src/routes/index.js @@ -0,0 +1,40 @@ +const express = require('express'); +const router = express.Router(); +const { services } = require('../services'); + +// Import route modules here as they are created +// const authRoutes = require('./auth'); +// const adminRoutes = require('./admin'); +// const analyticsRoutes = require('./analytics'); +const exportRoutes = require('./exports'); + +// Mount routes under their respective prefixes +// router.use('/api/auth', authRoutes); +// router.use('/api/admin', adminRoutes); +// router.use('/api/analytics', analyticsRoutes); +router.use('/api/exports', exportRoutes); + +// Health check route +router.get('/health', (req, res) => { + const healthData = { + status: 'ok', + timestamp: new Date().toISOString(), + }; + + // Add queue stats if queue service is available + if (services.queue) { + healthData.queue = services.queue.getStats(); + } + + // Add scheduler stats if scheduler service is available + if (services.scheduler) { + healthData.scheduler = { + schedules: services.scheduler.getAllSchedules().length, + isRunning: services.scheduler.isRunning, + }; + } + + res.json(healthData); +}); + +module.exports = router; diff --git a/src/server.js b/src/server.js new file mode 100644 index 00000000..299425e0 --- /dev/null +++ b/src/server.js @@ -0,0 +1,115 @@ +require('./config/tracing'); +const http = require('http'); +const { childLogger } = require('./config/logger'); +const config = require('./config'); +const app = require('./app'); +const { initServices, shutdownServices } = require('./services'); + +const log = childLogger('server'); + +let server; +let isShuttingDown = false; + +/** + * Start the HTTP server + */ +async function startServer() { + try { + // Initialize services + await initServices(app); + + // Create HTTP server + server = http.createServer(app); + + // Start listening + server.listen(config.port, () => { + log.info( + { + port: config.port, + env: config.env, + contractId: config.contractId, + }, + 'Equipchain API server started' + ); + }); + + // Handle server errors + server.on('error', (error) => { + if (error.code === 'EADDRINUSE') { + log.error({ port: config.port }, 'Port already in use'); + } else { + log.error({ error }, 'Server error'); + } + process.exit(1); + }); + } catch (error) { + log.error({ error }, 'Failed to start server'); + process.exit(1); + } +} + +/** + * Gracefully shutdown the server + */ +async function gracefulShutdown(signal) { + if (isShuttingDown) { + log.warn('Shutdown already in progress, ignoring signal'); + return; + } + + isShuttingDown = true; + log.info({ signal }, 'Received shutdown signal, starting graceful shutdown'); + + // Stop accepting new connections + if (server) { + server.close(async (err) => { + if (err) { + log.error({ error: err }, 'Error closing server'); + process.exit(1); + } + + log.info('HTTP server closed'); + + try { + // Shutdown services + await shutdownServices(); + log.info('Graceful shutdown complete'); + process.exit(0); + } catch (error) { + log.error({ error }, 'Error during service shutdown'); + process.exit(1); + } + }); + + // Force shutdown after timeout + setTimeout(() => { + log.error('Forced shutdown after timeout'); + process.exit(1); + }, 10000); + } else { + process.exit(0); + } +} + +// Register signal handlers +process.on('SIGTERM', () => gracefulShutdown('SIGTERM')); +process.on('SIGINT', () => gracefulShutdown('SIGINT')); + +// Handle uncaught exceptions +process.on('uncaughtException', (error) => { + log.error({ error }, 'Uncaught exception'); + gracefulShutdown('uncaughtException'); +}); + +// Handle unhandled promise rejections +process.on('unhandledRejection', (reason, promise) => { + log.error({ reason, promise }, 'Unhandled promise rejection'); + gracefulShutdown('unhandledRejection'); +}); + +// Start server if this file is run directly +if (require.main === module) { + startServer(); +} + +module.exports = { startServer, gracefulShutdown }; diff --git a/src/services/exporter.js b/src/services/exporter.js new file mode 100644 index 00000000..150faec5 --- /dev/null +++ b/src/services/exporter.js @@ -0,0 +1,286 @@ +const { Readable } = require('stream'); +const { stringify } = require('csv-stringify'); +const { childLogger } = require('../config/logger'); + +const log = childLogger('exporter'); + +/** + * Supported export formats + */ +const SUPPORTED_FORMATS = ['csv', 'json', 'ndjson']; + +/** + * Default CSV options + */ +const DEFAULT_CSV_OPTIONS = { + header: true, + delimiter: ',', + quoted: false, + quotedEmpty: true, + quotedString: false, + escape: '\\', +}; + +/** + * Generate a descriptive filename based on export type and date range + * @param {string} type - Export type (e.g., 'readings', 'analytics', 'system-report') + * @param {string} startDate - Start date in ISO format + * @param {string} endDate - End date in ISO format + * @param {string} format - File format ('csv' or 'json') + * @returns {string} Generated filename + */ +function generateFilename(type, startDate, endDate, format) { + const formatDate = (dateStr) => { + if (!dateStr) return 'all'; + return dateStr.split('T')[0]; + }; + + const start = formatDate(startDate); + const end = formatDate(endDate); + const ext = format === 'ndjson' ? 'json' : format; + + if (start === 'all' && end === 'all') { + return `${type}.${ext}`; + } + + return `${type}-${start}-to-${end}.${ext}`; +} + +/** + * Validate and normalize export format + * @param {string} format - Requested format + * @returns {string} Normalized format + * @throws {Error} If format is invalid + */ +function validateFormat(format) { + const normalized = format?.toLowerCase(); + if (!SUPPORTED_FORMATS.includes(normalized)) { + throw new Error(`Invalid format '${format}'. Supported formats: ${SUPPORTED_FORMATS.join(', ')}`); + } + return normalized; +} + +/** + * Validate requested fields against available columns + * @param {Array} requestedFields - Fields requested by user + * @param {Array} availableFields - Available fields in data + * @returns {Array} Validated fields + * @throws {Error} If invalid fields are requested + */ +function validateFields(requestedFields, availableFields) { + if (!requestedFields || requestedFields.length === 0) { + return availableFields; + } + + const invalidFields = requestedFields.filter(field => !availableFields.includes(field)); + if (invalidFields.length > 0) { + throw new Error(`Invalid fields requested: ${invalidFields.join(', ')}`); + } + + return requestedFields; +} + +/** + * Filter object to include only specified fields + * @param {Object} obj - Object to filter + * @param {Array} fields - Fields to include + * @returns {Object} Filtered object + */ +function filterFields(obj, fields) { + const filtered = {}; + fields.forEach(field => { + if (obj.hasOwnProperty(field)) { + filtered[field] = obj[field]; + } + }); + return filtered; +} + +/** + * Create a Readable stream from an array of objects + * @param {Array} data - Data to stream + * @returns {Readable} Readable stream + */ +function createReadableStream(data) { + return Readable.from(data); +} + +/** + * Export data to CSV format with streaming + * @param {Readable} dataStream - Readable stream of data objects + * @param {Array} columns - Columns to include + * @param {Object} options - CSV options + * @returns {Readable} CSV stream + */ +function exportToCSV(dataStream, columns, options = {}) { + const csvOptions = { + ...DEFAULT_CSV_OPTIONS, + columns, + ...options, + }; + + log.debug({ columns, options: csvOptions }, 'Creating CSV export stream'); + + const csvStream = stringify(csvOptions); + + // Transform data to include only specified columns + const transformStream = new Readable({ + objectMode: true, + read() {}, + }); + + dataStream.on('data', (chunk) => { + const filtered = filterFields(chunk, columns); + transformStream.push(filtered); + }); + + dataStream.on('end', () => { + transformStream.push(null); + }); + + dataStream.on('error', (error) => { + transformStream.emit('error', error); + }); + + transformStream.pipe(csvStream); + return csvStream; +} + +/** + * Export data to JSON format with streaming + * @param {Readable} dataStream - Readable stream of data objects + * @param {Array} columns - Columns to include + * @param {Object} options - Export options + * @returns {Readable} JSON stream + */ +function exportToJSON(dataStream, columns, options = {}) { + const { pretty = false, ndjson = false } = options; + + log.debug({ columns, pretty, ndjson }, 'Creating JSON export stream'); + + const jsonStream = new Readable({ + read() {}, + }); + + let first = true; + let itemCount = 0; + + if (!ndjson) { + // JSON array format + jsonStream.push('['); + } + + dataStream.on('data', (chunk) => { + const filtered = filterFields(chunk, columns); + const jsonStr = JSON.stringify(filtered, null, pretty ? 2 : 0); + + if (ndjson) { + // Newline-delimited JSON + jsonStream.push(jsonStr + '\n'); + } else { + // JSON array + if (!first) { + jsonStream.push(','); + } + jsonStream.push(jsonStr); + first = false; + } + itemCount++; + }); + + dataStream.on('end', () => { + if (!ndjson) { + jsonStream.push(']'); + } + jsonStream.push(null); + log.debug({ itemCount }, 'JSON export stream completed'); + }); + + dataStream.on('error', (error) => { + jsonStream.emit('error', error); + }); + + return jsonStream; +} + +/** + * Set appropriate response headers for export + * @param {Object} res - Express response object + * @param {string} format - Export format + * @param {string} filename - Filename for Content-Disposition + */ +function setExportHeaders(res, format, filename) { + const contentTypes = { + csv: 'text/csv; charset=utf-8', + json: 'application/json; charset=utf-8', + ndjson: 'application/x-ndjson; charset=utf-8', + }; + + res.setHeader('Content-Type', contentTypes[format] || 'application/octet-stream'); + res.setHeader('Content-Disposition', `attachment; filename="${filename}"`); + res.setHeader('Transfer-Encoding', 'chunked'); +} + +/** + * Handle export request with streaming response + * @param {Object} req - Express request object + * @param {Object} res - Express response object + * @param {Object} data - Data to export (array or stream) + * @param {Array} availableFields - Available fields in data + * @param {string} exportType - Type of export for filename + * @returns {Promise} + */ +async function handleExport(req, res, data, availableFields, exportType) { + try { + const format = validateFormat(req.query.format || 'csv'); + const requestedFields = req.query.fields ? req.query.fields.split(',').map(f => f.trim()) : null; + const startDate = req.query.startDate; + const endDate = req.query.endDate; + + const columns = validateFields(requestedFields, availableFields); + const filename = generateFilename(exportType, startDate, endDate, format); + + setExportHeaders(res, format, filename); + + // Create data stream + const dataStream = Array.isArray(data) ? createReadableStream(data) : data; + + // Pipe appropriate export stream to response + let exportStream; + if (format === 'csv') { + exportStream = exportToCSV(dataStream, columns, req.query); + } else { + exportStream = exportToJSON(dataStream, columns, { + pretty: req.query.pretty === 'true', + ndjson: format === 'ndjson', + }); + } + + exportStream.pipe(res); + + // Handle stream errors + exportStream.on('error', (error) => { + log.error({ error }, 'Export stream error'); + if (!res.headersSent) { + res.status(500).json({ error: 'Export failed', message: error.message }); + } + }); + + } catch (error) { + log.error({ error }, 'Export request error'); + if (!res.headersSent) { + res.status(400).json({ error: error.message }); + } + } +} + +module.exports = { + generateFilename, + validateFormat, + validateFields, + exportToCSV, + exportToJSON, + setExportHeaders, + handleExport, + SUPPORTED_FORMATS, +}; diff --git a/src/services/index.js b/src/services/index.js new file mode 100644 index 00000000..75276bda --- /dev/null +++ b/src/services/index.js @@ -0,0 +1,157 @@ +const { childLogger } = require('../config/logger'); +const { queue, Priority } = require('./queue'); +const { scheduler } = require('./scheduler'); +const billingHandler = require('../jobs/billing.job'); +const reportsHandler = require('../jobs/reports.job'); +const syncHandler = require('../jobs/sync.job'); +const webhookRetryHandler = require('../jobs/webhookRetry.job'); +const cacheWarmHandler = require('../jobs/cacheWarm.job'); + +const log = childLogger('services'); + +// Service registry to track initialized services +const services = { + cache: null, + queue: null, + scheduler: null, + eventListener: null, + websocket: null, +}; + +/** + * Initialize all services in the correct dependency order + * @param {Object} app - Express application instance + */ +async function initServices(app) { + try { + log.info('Initializing services...'); + + // Initialize cache service (if implemented) + // services.cache = await initCache(); + // log.info('Cache service initialized'); + + // Initialize queue service + await initQueue(); + log.info('Queue service initialized'); + + // Initialize scheduler service + await initScheduler(); + log.info('Scheduler service initialized'); + + // Initialize event listener (if implemented) + // services.eventListener = await initEventListener(); + // log.info('Event listener initialized'); + + // Initialize WebSocket (if implemented) + // services.websocket = await initWebSocket(app); + // log.info('WebSocket initialized'); + + log.info('All services initialized successfully'); + } catch (error) { + log.error({ error }, 'Failed to initialize services'); + throw error; + } +} + +/** + * Initialize queue service and register job handlers + */ +async function initQueue() { + // Register job handlers + queue.registerHandler('billing', billingHandler); + queue.registerHandler('reports', reportsHandler); + queue.registerHandler('sync', syncHandler); + queue.registerHandler('webhookRetry', webhookRetryHandler); + queue.registerHandler('cacheWarm', cacheWarmHandler); + + // Start processing jobs + queue.start(); + + services.queue = queue; +} + +/** + * Initialize scheduler service and register recurring jobs + */ +async function initScheduler() { + // Schedule recurring jobs (examples - adjust intervals as needed) + + // Billing job - runs every hour + scheduler.schedule('billing-hourly', '0 * * * *', async () => { + const period = new Date().toISOString().slice(0, 7); // YYYY-MM + queue.add('billing', { period }); + }); + + // Daily reports - runs every day at midnight + scheduler.schedule('reports-daily', '0 0 * * *', async () => { + const date = new Date().toISOString().slice(0, 10); // YYYY-MM-DD + queue.add('reports', { type: 'daily', date }); + }); + + // Monthly reports - runs on 1st of each month at midnight + scheduler.schedule('reports-monthly', '0 0 1 * *', async () => { + const date = new Date().toISOString().slice(0, 10); // YYYY-MM-DD + queue.add('reports', { type: 'monthly', date }); + }); + + // Sync job - runs every 30 minutes + scheduler.schedule('sync-periodic', '1800000', async () => { + queue.add('sync', { syncType: 'contract_state' }); + }); + + // Cache warm - runs every 15 minutes + scheduler.schedule('cache-warm', '900000', async () => { + queue.add('cacheWarm', { cacheType: 'meter_data' }); + }); + + // Start the scheduler + scheduler.start(); + + services.scheduler = scheduler; +} + +/** + * Gracefully shutdown all services + */ +async function shutdownServices() { + try { + log.info('Shutting down services...'); + + // Shutdown services in reverse dependency order + if (services.websocket) { + await services.websocket.close(); + log.info('WebSocket shutdown complete'); + } + + if (services.eventListener) { + await services.eventListener.stop(); + log.info('Event listener shutdown complete'); + } + + if (services.scheduler) { + await services.scheduler.close(); + log.info('Scheduler shutdown complete'); + } + + if (services.queue) { + await services.queue.close(); + log.info('Queue shutdown complete'); + } + + if (services.cache) { + await services.cache.quit(); + log.info('Cache shutdown complete'); + } + + log.info('All services shutdown complete'); + } catch (error) { + log.error({ error }, 'Error during service shutdown'); + throw error; + } +} + +module.exports = { + initServices, + shutdownServices, + services, +}; diff --git a/src/services/queue.js b/src/services/queue.js new file mode 100644 index 00000000..570c9fb7 --- /dev/null +++ b/src/services/queue.js @@ -0,0 +1,395 @@ +const EventEmitter = require('events'); +const { childLogger } = require('../config/logger'); + +const log = childLogger('queue'); + +// Environment configuration +const JOB_CONCURRENCY = parseInt(process.env.JOB_CONCURRENCY || '5', 10); +const JOB_RETRY_ATTEMPTS = parseInt(process.env.JOB_RETRY_ATTEMPTS || '3', 10); + +// Job status constants +const JobStatus = { + QUEUED: 'queued', + RUNNING: 'running', + COMPLETED: 'completed', + FAILED: 'failed', + CANCELLED: 'cancelled', +}; + +// Priority levels +const Priority = { + HIGH: 3, + NORMAL: 2, + LOW: 1, +}; + +class JobQueue extends EventEmitter { + constructor() { + super(); + this.jobs = new Map(); // jobId -> job object + this.queuedJobs = []; // array of jobIds sorted by priority + this.runningJobs = new Set(); // set of jobIds currently running + this.handlers = new Map(); // job type -> handler function + this.activeCount = 0; + this.isProcessing = false; + } + + /** + * Register a job handler + * @param {string} type - Job type identifier + * @param {Function} handler - Handler function + */ + registerHandler(type, handler) { + if (typeof handler !== 'function') { + throw new Error(`Handler for job type "${type}" must be a function`); + } + this.handlers.set(type, handler); + log.info({ type }, 'Job handler registered'); + } + + /** + * Add a job to the queue + * @param {string} type - Job type + * @param {Object} data - Job data + * @param {Object} options - Job options + * @returns {string} Job ID + */ + add(type, data = {}, options = {}) { + const jobId = `job_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`; + + const job = { + id: jobId, + type, + data, + status: JobStatus.QUEUED, + priority: options.priority || Priority.NORMAL, + attempts: 0, + maxAttempts: options.maxAttempts || JOB_RETRY_ATTEMPTS, + delay: options.delay || 0, + createdAt: new Date(), + startedAt: null, + completedAt: null, + failedAt: null, + result: null, + error: null, + }; + + this.jobs.set(jobId, job); + + if (job.delay > 0) { + // Schedule for delayed execution + setTimeout(() => { + if (job.status === JobStatus.QUEUED) { + this._enqueue(jobId); + } + }, job.delay); + log.info({ jobId, type, delay: job.delay }, 'Job scheduled with delay'); + } else { + this._enqueue(jobId); + } + + this.emit('added', job); + log.info({ jobId, type, priority: job.priority }, 'Job added to queue'); + + return jobId; + } + + /** + * Schedule a recurring job + * @param {string} type - Job type + * @param {Object} data - Job data + * @param {string} cronExpression - Cron expression (simplified for MVP: interval in ms) + * @returns {string} Schedule ID + */ + schedule(type, data, cronExpression) { + const scheduleId = `schedule_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`; + + // For MVP, treat cronExpression as interval in milliseconds + const interval = parseInt(cronExpression, 10); + + if (isNaN(interval) || interval <= 0) { + throw new Error('Invalid cron expression. For MVP, provide interval in milliseconds'); + } + + const intervalId = setInterval(() => { + this.add(type, data); + }, interval); + + this.emit('scheduled', { scheduleId, type, interval }); + log.info({ scheduleId, type, interval }, 'Recurring job scheduled'); + + return scheduleId; + } + + /** + * Get job status + * @param {string} jobId - Job ID + * @returns {Object|null} Job object or null if not found + */ + getStatus(jobId) { + const job = this.jobs.get(jobId); + if (!job) { + return null; + } + return { + id: job.id, + type: job.type, + status: job.status, + attempts: job.attempts, + createdAt: job.createdAt, + startedAt: job.startedAt, + completedAt: job.completedAt, + failedAt: job.failedAt, + result: job.result, + error: job.error, + }; + } + + /** + * Cancel a job + * @param {string} jobId - Job ID + * @returns {boolean} Success status + */ + cancel(jobId) { + const job = this.jobs.get(jobId); + if (!job) { + return false; + } + + if (job.status === JobStatus.RUNNING) { + // Cannot cancel running jobs in this implementation + return false; + } + + job.status = JobStatus.CANCELLED; + this._removeFromQueue(jobId); + this.emit('cancelled', job); + log.info({ jobId }, 'Job cancelled'); + + return true; + } + + /** + * Start processing jobs + */ + start() { + if (this.isProcessing) { + log.warn('Queue is already processing'); + return; + } + + this.isProcessing = true; + log.info('Queue processing started'); + this._process(); + } + + /** + * Stop processing jobs + */ + async stop() { + this.isProcessing = false; + log.info('Queue processing stopped'); + } + + /** + * Get queue statistics + * @returns {Object} Queue stats + */ + getStats() { + let queued = 0; + let running = 0; + let completed = 0; + let failed = 0; + let cancelled = 0; + + for (const job of this.jobs.values()) { + switch (job.status) { + case JobStatus.QUEUED: + queued++; + break; + case JobStatus.RUNNING: + running++; + break; + case JobStatus.COMPLETED: + completed++; + break; + case JobStatus.FAILED: + failed++; + break; + case JobStatus.CANCELLED: + cancelled++; + break; + } + } + + return { + queued, + running, + completed, + failed, + cancelled, + total: this.jobs.size, + activeCount: this.activeCount, + maxConcurrency: JOB_CONCURRENCY, + }; + } + + /** + * Enqueue a job (internal method) + * @param {string} jobId - Job ID + */ + _enqueue(jobId) { + const job = this.jobs.get(jobId); + if (!job || job.status !== JobStatus.QUEUED) { + return; + } + + // Insert based on priority (higher priority first) + let inserted = false; + for (let i = 0; i < this.queuedJobs.length; i++) { + const queuedJob = this.jobs.get(this.queuedJobs[i]); + if (queuedJob && job.priority > queuedJob.priority) { + this.queuedJobs.splice(i, 0, jobId); + inserted = true; + break; + } + } + + if (!inserted) { + this.queuedJobs.push(jobId); + } + + this._process(); + } + + /** + * Remove job from queue (internal method) + * @param {string} jobId - Job ID + */ + _removeFromQueue(jobId) { + const index = this.queuedJobs.indexOf(jobId); + if (index > -1) { + this.queuedJobs.splice(index, 1); + } + } + + /** + * Process jobs from the queue + */ + async _process() { + if (!this.isProcessing) { + return; + } + + // Process jobs while we have capacity + while (this.activeCount < JOB_CONCURRENCY && this.queuedJobs.length > 0) { + const jobId = this.queuedJobs.shift(); + const job = this.jobs.get(jobId); + + if (!job || job.status !== JobStatus.QUEUED) { + continue; + } + + this._executeJob(job); + } + } + + /** + * Execute a single job + * @param {Object} job - Job object + */ + async _executeJob(job) { + const handler = this.handlers.get(job.type); + + if (!handler) { + job.status = JobStatus.FAILED; + job.error = `No handler registered for job type "${job.type}"`; + job.failedAt = new Date(); + this.emit('failed', job); + log.error({ jobId: job.id, type: job.type }, job.error); + this._process(); + return; + } + + job.status = JobStatus.RUNNING; + job.startedAt = new Date(); + this.activeCount++; + this.runningJobs.add(job.id); + this.emit('started', job); + log.info({ jobId: job.id, type: job.type }, 'Job started'); + + try { + const result = await handler(job.data); + + job.status = JobStatus.COMPLETED; + job.result = result; + job.completedAt = new Date(); + this.activeCount--; + this.runningJobs.delete(job.id); + this.emit('completed', job); + log.info({ jobId: job.id, type: job.type }, 'Job completed'); + } catch (error) { + job.attempts++; + job.error = error.message; + + if (job.attempts < job.maxAttempts) { + // Retry with exponential backoff + const backoffDelay = Math.pow(2, job.attempts) * 1000; + job.status = JobStatus.QUEUED; + job.startedAt = null; + this.activeCount--; + this.runningJobs.delete(job.id); + + setTimeout(() => { + this._enqueue(job.id); + }, backoffDelay); + + this.emit('retry', job); + log.warn({ + jobId: job.id, + type: job.type, + attempt: job.attempts, + maxAttempts: job.maxAttempts, + backoffDelay + }, 'Job retry scheduled'); + } else { + // Max attempts reached + job.status = JobStatus.FAILED; + job.failedAt = new Date(); + this.activeCount--; + this.runningJobs.delete(job.id); + this.emit('failed', job); + log.error({ + jobId: job.id, + type: job.type, + attempts: job.attempts, + error: error.message + }, 'Job failed after max attempts'); + } + } + + // Process next jobs + this._process(); + } + + /** + * Close the queue and cleanup + */ + async close() { + await this.stop(); + this.jobs.clear(); + this.queuedJobs = []; + this.runningJobs.clear(); + this.handlers.clear(); + log.info('Queue closed'); + } +} + +// Create singleton instance +const queue = new JobQueue(); + +module.exports = { + queue, + JobStatus, + Priority, +}; diff --git a/src/services/scheduler.js b/src/services/scheduler.js new file mode 100644 index 00000000..1eaeeb9d --- /dev/null +++ b/src/services/scheduler.js @@ -0,0 +1,283 @@ +const { childLogger } = require('../config/logger'); + +const log = childLogger('scheduler'); + +class Scheduler { + constructor() { + this.schedules = new Map(); // scheduleId -> schedule object + this.isRunning = false; + } + + /** + * Schedule a recurring job + * @param {string} name - Schedule name + * @param {string} cronExpression - Cron expression (simplified for MVP: interval in ms) + * @param {Function} handler - Handler function to execute + * @returns {string} Schedule ID + */ + schedule(name, cronExpression, handler) { + if (this.schedules.has(name)) { + throw new Error(`Schedule with name "${name}" already exists`); + } + + if (typeof handler !== 'function') { + throw new Error('Handler must be a function'); + } + + // For MVP, treat cronExpression as interval in milliseconds + // In production, this would use a proper cron parser + const interval = this._parseInterval(cronExpression); + + if (interval <= 0) { + throw new Error('Invalid cron expression. For MVP, provide interval in milliseconds or a simple cron format'); + } + + const scheduleId = `schedule_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`; + + const schedule = { + id: scheduleId, + name, + cronExpression, + interval, + handler, + intervalId: null, + lastRun: null, + nextRun: null, + runCount: 0, + createdAt: new Date(), + }; + + this.schedules.set(name, schedule); + + if (this.isRunning) { + this._startSchedule(schedule); + } + + log.info({ name, cronExpression, interval }, 'Schedule created'); + + return scheduleId; + } + + /** + * Cancel a schedule by name + * @param {string} name - Schedule name + * @returns {boolean} Success status + */ + cancelSchedule(name) { + const schedule = this.schedules.get(name); + + if (!schedule) { + return false; + } + + if (schedule.intervalId) { + clearInterval(schedule.intervalId); + schedule.intervalId = null; + } + + this.schedules.delete(name); + log.info({ name }, 'Schedule cancelled'); + + return true; + } + + /** + * Get schedule information + * @param {string} name - Schedule name + * @returns {Object|null} Schedule info or null + */ + getSchedule(name) { + const schedule = this.schedules.get(name); + + if (!schedule) { + return null; + } + + return { + id: schedule.id, + name: schedule.name, + cronExpression: schedule.cronExpression, + interval: schedule.interval, + lastRun: schedule.lastRun, + nextRun: schedule.nextRun, + runCount: schedule.runCount, + createdAt: schedule.createdAt, + isActive: schedule.intervalId !== null, + }; + } + + /** + * Get all schedules + * @returns {Array} Array of schedule info objects + */ + getAllSchedules() { + const schedules = []; + + for (const schedule of this.schedules.values()) { + schedules.push({ + id: schedule.id, + name: schedule.name, + cronExpression: schedule.cronExpression, + interval: schedule.interval, + lastRun: schedule.lastRun, + nextRun: schedule.nextRun, + runCount: schedule.runCount, + createdAt: schedule.createdAt, + isActive: schedule.intervalId !== null, + }); + } + + return schedules; + } + + /** + * Start the scheduler + */ + start() { + if (this.isRunning) { + log.warn('Scheduler is already running'); + return; + } + + this.isRunning = true; + log.info('Scheduler started'); + + // Start all schedules + for (const schedule of this.schedules.values()) { + this._startSchedule(schedule); + } + } + + /** + * Stop the scheduler + */ + stop() { + if (!this.isRunning) { + return; + } + + this.isRunning = false; + log.info('Scheduler stopping'); + + // Stop all schedules + for (const schedule of this.schedules.values()) { + if (schedule.intervalId) { + clearInterval(schedule.intervalId); + schedule.intervalId = null; + } + } + + log.info('Scheduler stopped'); + } + + /** + * Parse interval from cron expression + * For MVP, supports: + * - Numeric milliseconds (e.g., "60000" for 1 minute) + * - Simple cron format: "* * * * *" (min hour day month weekday) + * Currently only supports interval-based scheduling + * @param {string} cronExpression - Cron expression or interval + * @returns {number} Interval in milliseconds + */ + _parseInterval(cronExpression) { + // If it's a number, treat as milliseconds + const numericValue = parseInt(cronExpression, 10); + if (!isNaN(numericValue) && numericValue > 0) { + return numericValue; + } + + // Simple cron parsing for common intervals + // Format: minute hour day month weekday + const parts = cronExpression.split(' ').map(p => p.trim()); + + if (parts.length !== 5) { + return -1; + } + + const [minute, hour, day, month, weekday] = parts; + + // Every minute + if (minute === '*' && hour === '*' && day === '*' && month === '*' && weekday === '*') { + return 60 * 1000; + } + + // Every hour at minute 0 + if (minute === '0' && hour === '*' && day === '*' && month === '*' && weekday === '*') { + return 60 * 60 * 1000; + } + + // Every day at midnight + if (minute === '0' && hour === '0' && day === '*' && month === '*' && weekday === '*') { + return 24 * 60 * 60 * 1000; + } + + // Every Monday at midnight + if (minute === '0' && hour === '0' && day === '*' && month === '*' && weekday === '1') { + return 7 * 24 * 60 * 60 * 1000; + } + + // First day of every month at midnight + if (minute === '0' && hour === '0' && day === '1' && month === '*' && weekday === '*') { + return 30 * 24 * 60 * 60 * 1000; // Approximate + } + + // If we can't parse it, return -1 + return -1; + } + + /** + * Start a single schedule + * @param {Object} schedule - Schedule object + */ + _startSchedule(schedule) { + if (schedule.intervalId) { + return; + } + + schedule.intervalId = setInterval(async () => { + schedule.lastRun = new Date(); + schedule.runCount++; + schedule.nextRun = new Date(Date.now() + schedule.interval); + + log.info({ + name: schedule.name, + runCount: schedule.runCount, + lastRun: schedule.lastRun, + }, 'Executing scheduled job'); + + try { + await schedule.handler(); + log.info({ name: schedule.name }, 'Scheduled job completed successfully'); + } catch (error) { + log.error({ + name: schedule.name, + error: error.message + }, 'Scheduled job failed'); + } + }, schedule.interval); + + schedule.nextRun = new Date(Date.now() + schedule.interval); + + log.info({ + name: schedule.name, + interval: schedule.interval, + nextRun: schedule.nextRun, + }, 'Schedule started'); + } + + /** + * Cleanup and close scheduler + */ + async close() { + this.stop(); + this.schedules.clear(); + log.info('Scheduler closed'); + } +} + +// Create singleton instance +const scheduler = new Scheduler(); + +module.exports = { + scheduler, +}; diff --git a/src/services/webhook.js b/src/services/webhook.js new file mode 100644 index 00000000..70ee76ce --- /dev/null +++ b/src/services/webhook.js @@ -0,0 +1,58 @@ +const { childLogger } = require('../config/logger'); +const { services } = require('./index'); + +const log = childLogger('webhook'); + +/** + * Webhook service for delivering webhooks with queue-based retries + */ +class WebhookService { + /** + * Deliver a webhook with automatic retry via queue + * @param {string} url - Target webhook URL + * @param {Object} payload - Webhook payload + * @param {Object} options - Delivery options + * @returns {string} Job ID + */ + async deliver(url, payload, options = {}) { + const webhookId = `webhook_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`; + + log.info({ webhookId, url }, 'Queuing webhook for delivery'); + + // Add webhook retry job to queue + const jobId = services.queue.add('webhookRetry', { + webhookId, + url, + payload, + attempt: 1, + }, { + priority: options.priority || 2, // Normal priority + maxAttempts: options.maxAttempts || 3, + }); + + return jobId; + } + + /** + * Get webhook delivery status + * @param {string} jobId - Job ID from deliver() + * @returns {Object|null} Job status + */ + getStatus(jobId) { + return services.queue.getStatus(jobId); + } + + /** + * Cancel a pending webhook delivery + * @param {string} jobId - Job ID from deliver() + * @returns {boolean} Success status + */ + cancel(jobId) { + return services.queue.cancel(jobId); + } +} + +// Create singleton instance +const webhookService = new WebhookService(); + +module.exports = webhookService; diff --git a/src/utils/sanitize.js b/src/utils/sanitize.js new file mode 100644 index 00000000..2a4c3a7b --- /dev/null +++ b/src/utils/sanitize.js @@ -0,0 +1,111 @@ +const escape = require('escape-html'); + +/** + * Sanitize a string by escaping HTML entities to prevent XSS attacks + * @param {string} str - The string to sanitize + * @returns {string} The sanitized string with HTML entities encoded + */ +function sanitize(str) { + if (typeof str !== 'string') { + return str; + } + return escape(str); +} + +/** + * Sanitize specific string fields in an object + * @param {Object} obj - The object to sanitize + * @param {string[]} fields - Array of field names to sanitize + * @returns {Object} The object with specified fields sanitized + */ +function sanitizeObject(obj, fields) { + if (!obj || typeof obj !== 'object') { + return obj; + } + + const sanitized = { ...obj }; + + for (const field of fields) { + if (sanitized[field] !== undefined && typeof sanitized[field] === 'string') { + sanitized[field] = sanitize(sanitized[field]); + } + } + + return sanitized; +} + +/** + * Sanitize all string values in an object recursively + * @param {*} value - The value to sanitize + * @returns {*} The sanitized value + */ +function sanitizeDeep(value) { + if (typeof value === 'string') { + return sanitize(value); + } + + if (Array.isArray(value)) { + return value.map(item => sanitizeDeep(item)); + } + + if (value !== null && typeof value === 'object') { + const sanitized = {}; + for (const [key, val] of Object.entries(value)) { + sanitized[key] = sanitizeDeep(val); + } + return sanitized; + } + + return value; +} + +/** + * Remove control characters from a string to prevent log injection + * @param {string} str - The string to clean + * @returns {string} The string with control characters removed + */ +function removeControlChars(str) { + if (typeof str !== 'string') { + return str; + } + // Remove control characters except newline, tab, and carriage return + return str.replace(/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/g, ''); +} + +/** + * Sanitize data for logging (remove control characters and limit length) + * @param {*} data - The data to sanitize for logging + * @param {number} maxLength - Maximum length for strings (default: 1000) + * @returns {*} The sanitized data + */ +function sanitizeForLogging(data, maxLength = 1000) { + if (typeof data === 'string') { + let sanitized = removeControlChars(data); + if (sanitized.length > maxLength) { + sanitized = sanitized.substring(0, maxLength) + '...[truncated]'; + } + return sanitized; + } + + if (Array.isArray(data)) { + return data.map(item => sanitizeForLogging(item, maxLength)); + } + + if (data !== null && typeof data === 'object') { + const sanitized = {}; + for (const [key, val] of Object.entries(data)) { + sanitized[key] = sanitizeForLogging(val, maxLength); + } + return sanitized; + } + + return data; +} + +module.exports = { + sanitize, + sanitizeObject, + sanitizeDeep, + removeControlChars, + sanitizeForLogging, +}; diff --git a/test/app.test.js b/test/app.test.js new file mode 100644 index 00000000..972bc555 --- /dev/null +++ b/test/app.test.js @@ -0,0 +1,55 @@ +const { describe, it, after } = require('node:test'); +const assert = require('node:assert'); + +const app = require('../src/app'); +const server = app.listen(0); + +after(() => server.close()); + +describe('app', () => { + it('creates an Express application instance', () => { + assert.strictEqual(typeof app, 'function'); + assert.strictEqual(app.name, 'express'); + }); + + it('responds to root route with project info', async () => { + const res = await fetch(`http://localhost:${server.address().port}/`); + assert.strictEqual(res.status, 200); + + const data = await res.json(); + assert.strictEqual(data.project, 'Equipchain'); + assert.strictEqual(data.status, 'Monitoring Meters'); + assert.ok(data.contract); + }); + + it('responds to health check route', async () => { + const res = await fetch(`http://localhost:${server.address().port}/health`); + assert.strictEqual(res.status, 200); + + const data = await res.json(); + assert.strictEqual(data.status, 'ok'); + assert.ok(data.timestamp); + }); + + it('returns 404 for non-existent routes', async () => { + const res = await fetch(`http://localhost:${server.address().port}/non-existent-route`); + assert.strictEqual(res.status, 404); + + const data = await res.json(); + assert.strictEqual(data.error, 'Not Found'); + }); + + it('includes correlation ID header in response', async () => { + const res = await fetch(`http://localhost:${server.address().port}/`); + assert.ok(res.headers.get('x-correlation-id')); + }); + + it('uses provided correlation ID from header', async () => { + const testCorrelationId = 'test-correlation-id-123'; + const res = await fetch(`http://localhost:${server.address().port}/`, { + headers: { 'x-correlation-id': testCorrelationId }, + }); + + assert.strictEqual(res.headers.get('x-correlation-id'), testCorrelationId); + }); +}); diff --git a/test/exporter.test.js b/test/exporter.test.js new file mode 100644 index 00000000..57da594c --- /dev/null +++ b/test/exporter.test.js @@ -0,0 +1,342 @@ +const { test, describe, beforeEach, afterEach } = require('node:test'); +const assert = require('node:assert'); +const { Readable } = require('stream'); +const { + generateFilename, + validateFormat, + validateFields, + exportToCSV, + exportToJSON, + setExportHeaders, + SUPPORTED_FORMATS, +} = require('../src/services/exporter'); + +describe('Exporter Service', () => { + + describe('generateFilename', () => { + test('should generate filename with date range', () => { + const filename = generateFilename('readings', '2026-01-01T00:00:00Z', '2026-06-01T00:00:00Z', 'csv'); + assert.strictEqual(filename, 'readings-2026-01-01-to-2026-06-01.csv'); + }); + + test('should generate filename without date range', () => { + const filename = generateFilename('readings', null, null, 'csv'); + assert.strictEqual(filename, 'readings.csv'); + }); + + test('should use json extension for ndjson format', () => { + const filename = generateFilename('analytics', '2026-01-01', '2026-01-31', 'ndjson'); + assert.strictEqual(filename, 'analytics-2026-01-01-to-2026-01-31.json'); + }); + + test('should handle single date', () => { + const filename = generateFilename('meters', '2026-01-01', null, 'json'); + assert.strictEqual(filename, 'meters-2026-01-01-to-all.json'); + }); + }); + + describe('validateFormat', () => { + test('should accept valid formats', () => { + assert.strictEqual(validateFormat('csv'), 'csv'); + assert.strictEqual(validateFormat('CSV'), 'csv'); + assert.strictEqual(validateFormat('json'), 'json'); + assert.strictEqual(validateFormat('ndjson'), 'ndjson'); + }); + + test('should throw error for invalid format', () => { + assert.throws(() => validateFormat('xml'), /Invalid format 'xml'/); + assert.throws(() => validateFormat('pdf'), /Invalid format 'pdf'/); + assert.throws(() => validateFormat(''), /Invalid format/); + }); + + test('should list supported formats in error message', () => { + try { + validateFormat('invalid'); + assert.fail('Should have thrown error'); + } catch (error) { + assert.ok(error.message.includes('csv')); + assert.ok(error.message.includes('json')); + assert.ok(error.message.includes('ndjson')); + } + }); + }); + + describe('validateFields', () => { + const availableFields = ['id', 'name', 'value', 'timestamp']; + + test('should return all fields when none requested', () => { + const result = validateFields(null, availableFields); + assert.deepStrictEqual(result, availableFields); + }); + + test('should return all fields when empty array requested', () => { + const result = validateFields([], availableFields); + assert.deepStrictEqual(result, availableFields); + }); + + test('should return only requested valid fields', () => { + const result = validateFields(['id', 'name'], availableFields); + assert.deepStrictEqual(result, ['id', 'name']); + }); + + test('should throw error for invalid fields', () => { + assert.throws(() => validateFields(['id', 'invalid'], availableFields), /Invalid fields requested/); + }); + + test('should list invalid fields in error message', () => { + try { + validateFields(['id', 'nonexistent', 'invalid'], availableFields); + assert.fail('Should have thrown error'); + } catch (error) { + assert.ok(error.message.includes('nonexistent')); + assert.ok(error.message.includes('invalid')); + } + }); + }); + + describe('exportToCSV', () => { + test('should create CSV stream with headers', (t, done) => { + const data = [ + { id: '1', name: 'Test', value: 100 }, + { id: '2', name: 'Test2', value: 200 }, + ]; + const dataStream = Readable.from(data); + const columns = ['id', 'name', 'value']; + + const csvStream = exportToCSV(dataStream, columns); + + let chunks = ''; + csvStream.on('data', (chunk) => { + chunks += chunk.toString(); + }); + + csvStream.on('end', () => { + assert.ok(chunks.includes('id,name,value')); + assert.ok(chunks.includes('1,Test,100')); + assert.ok(chunks.includes('2,Test2,200')); + done(); + }); + }); + + test('should filter to specified columns', (t, done) => { + const data = [ + { id: '1', name: 'Test', value: 100, extra: 'ignored' }, + { id: '2', name: 'Test2', value: 200, extra: 'ignored2' }, + ]; + const dataStream = Readable.from(data); + const columns = ['id', 'value']; + + const csvStream = exportToCSV(dataStream, columns); + + let chunks = ''; + csvStream.on('data', (chunk) => { + chunks += chunk.toString(); + }); + + csvStream.on('end', () => { + assert.ok(chunks.includes('id,value')); + assert.ok(!chunks.includes('extra')); + done(); + }); + }); + + test('should handle empty data', (t, done) => { + const data = []; + const dataStream = Readable.from(data); + const columns = ['id', 'name']; + + const csvStream = exportToCSV(dataStream, columns); + + let chunks = ''; + csvStream.on('data', (chunk) => { + chunks += chunk.toString(); + }); + + csvStream.on('end', () => { + assert.ok(chunks.includes('id,name')); + done(); + }); + }); + + test('should handle stream errors', (t, done) => { + const dataStream = new Readable({ + read() { + this.emit('error', new Error('Stream error')); + }, + }); + const columns = ['id', 'name']; + + const csvStream = exportToCSV(dataStream, columns); + + csvStream.on('error', (error) => { + assert.strictEqual(error.message, 'Stream error'); + done(); + }); + }); + }); + + describe('exportToJSON', () => { + test('should create JSON array stream', (t, done) => { + const data = [ + { id: '1', name: 'Test', value: 100 }, + { id: '2', name: 'Test2', value: 200 }, + ]; + const dataStream = Readable.from(data); + const columns = ['id', 'name', 'value']; + + const jsonStream = exportToJSON(dataStream, columns, { pretty: false }); + + let chunks = ''; + jsonStream.on('data', (chunk) => { + chunks += chunk.toString(); + }); + + jsonStream.on('end', () => { + const parsed = JSON.parse(chunks); + assert.strictEqual(parsed.length, 2); + assert.strictEqual(parsed[0].id, '1'); + assert.strictEqual(parsed[1].id, '2'); + done(); + }); + }); + + test('should create pretty-printed JSON', (t, done) => { + const data = [{ id: '1', name: 'Test' }]; + const dataStream = Readable.from(data); + const columns = ['id', 'name']; + + const jsonStream = exportToJSON(dataStream, columns, { pretty: true }); + + let chunks = ''; + jsonStream.on('data', (chunk) => { + chunks += chunk.toString(); + }); + + jsonStream.on('end', () => { + assert.ok(chunks.includes('\n')); + assert.ok(chunks.includes(' ')); + done(); + }); + }); + + test('should create NDJSON stream', (t, done) => { + const data = [ + { id: '1', name: 'Test' }, + { id: '2', name: 'Test2' }, + ]; + const dataStream = Readable.from(data); + const columns = ['id', 'name']; + + const jsonStream = exportToJSON(dataStream, columns, { ndjson: true }); + + let chunks = ''; + jsonStream.on('data', (chunk) => { + chunks += chunk.toString(); + }); + + jsonStream.on('end', => { + const lines = chunks.trim().split('\n'); + assert.strictEqual(lines.length, 2); + assert.strictEqual(JSON.parse(lines[0]).id, '1'); + assert.strictEqual(JSON.parse(lines[1]).id, '2'); + done(); + }); + }); + + test('should filter to specified columns', (t, done) => { + const data = [ + { id: '1', name: 'Test', value: 100, extra: 'ignored' }, + ]; + const dataStream = Readable.from(data); + const columns = ['id', 'name']; + + const jsonStream = exportToJSON(dataStream, columns); + + let chunks = ''; + jsonStream.on('data', (chunk) => { + chunks += chunk.toString(); + }); + + jsonStream.on('end', () => { + const parsed = JSON.parse(chunks); + assert.strictEqual(parsed[0].id, '1'); + assert.strictEqual(parsed[0].name, 'Test'); + assert.strictEqual(parsed[0].value, undefined); + assert.strictEqual(parsed[0].extra, undefined); + done(); + }); + }); + + test('should handle empty data', (t, done) => { + const data = []; + const dataStream = Readable.from(data); + const columns = ['id', 'name']; + + const jsonStream = exportToJSON(dataStream, columns); + + let chunks = ''; + jsonStream.on('data', (chunk) => { + chunks += chunk.toString(); + }); + + jsonStream.on('end', () => { + const parsed = JSON.parse(chunks); + assert.deepStrictEqual(parsed, []); + done(); + }); + }); + }); + + describe('setExportHeaders', () => { + test('should set CSV headers', () => { + const res = { + setHeader: function(name, value) { + this.headers[name] = value; + }, + headers: {}, + }; + + setExportHeaders(res, 'csv', 'test.csv'); + + assert.strictEqual(res.headers['Content-Type'], 'text/csv; charset=utf-8'); + assert.strictEqual(res.headers['Content-Disposition'], 'attachment; filename="test.csv"'); + assert.strictEqual(res.headers['Transfer-Encoding'], 'chunked'); + }); + + test('should set JSON headers', () => { + const res = { + setHeader: function(name, value) { + this.headers[name] = value; + }, + headers: {}, + }; + + setExportHeaders(res, 'json', 'test.json'); + + assert.strictEqual(res.headers['Content-Type'], 'application/json; charset=utf-8'); + assert.strictEqual(res.headers['Content-Disposition'], 'attachment; filename="test.json"'); + }); + + test('should set NDJSON headers', () => { + const res = { + setHeader: function(name, value) { + this.headers[name] = value; + }, + headers: {}, + }; + + setExportHeaders(res, 'ndjson', 'test.json'); + + assert.strictEqual(res.headers['Content-Type'], 'application/x-ndjson; charset=utf-8'); + }); + }); + + describe('SUPPORTED_FORMATS', () => { + test('should contain expected formats', () => { + assert.ok(Array.isArray(SUPPORTED_FORMATS)); + assert.ok(SUPPORTED_FORMATS.includes('csv')); + assert.ok(SUPPORTED_FORMATS.includes('json')); + assert.ok(SUPPORTED_FORMATS.includes('ndjson')); + }); + }); +}); diff --git a/test/exports-integration.test.js b/test/exports-integration.test.js new file mode 100644 index 00000000..9fa38929 --- /dev/null +++ b/test/exports-integration.test.js @@ -0,0 +1,387 @@ +const { test, describe, before, after } = require('node:test'); +const assert = require('node:assert'); +const http = require('http'); + +describe('Export Endpoints Integration Tests', () => { + let server; + const PORT = 3456; // Use different port for tests + + before(async () => { + // Start the server for testing + process.env.PORT = PORT; + process.env.NODE_ENV = 'test'; + + const app = require('../src/app'); + server = app.listen(PORT); + + // Wait for server to be ready + await new Promise(resolve => setTimeout(resolve, 100)); + }); + + after(() => { + if (server) { + server.close(); + } + }); + + function makeRequest(path, options = {}) { + return new Promise((resolve, reject) => { + const url = new URL(path, `http://localhost:${PORT}`); + + const requestOptions = { + hostname: 'localhost', + port: PORT, + path: url.pathname + url.search, + method: options.method || 'GET', + headers: options.headers || {}, + }; + + const req = http.request(requestOptions, (res) => { + let data = ''; + res.on('data', chunk => data += chunk); + res.on('end', () => { + resolve({ + statusCode: res.statusCode, + headers: res.headers, + body: data, + }); + }); + }); + + req.on('error', reject); + + if (options.body) { + req.write(options.body); + } + + req.end(); + }); + } + + describe('GET /api/exports/readings', () => { + test('should return 401 without authentication', async () => { + const response = await makeRequest('/api/exports/readings?format=csv'); + assert.strictEqual(response.statusCode, 401); + }); + + test('should return CSV with valid authentication', async () => { + const response = await makeRequest('/api/exports/readings?format=csv', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.strictEqual(response.headers['content-type'], 'text/csv; charset=utf-8'); + assert.ok(response.headers['content-disposition'].includes('attachment')); + assert.ok(response.headers['content-disposition'].includes('meter-readings')); + assert.ok(response.body.includes('id,meterId,timestamp,value,unit,status')); + }); + + test('should return JSON when format=json', async () => { + const response = await makeRequest('/api/exports/readings?format=json', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.strictEqual(response.headers['content-type'], 'application/json; charset=utf-8'); + + const data = JSON.parse(response.body); + assert.ok(Array.isArray(data)); + assert.ok(data.length > 0); + }); + + test('should return NDJSON when format=ndjson', async () => { + const response = await makeRequest('/api/exports/readings?format=ndjson', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.strictEqual(response.headers['content-type'], 'application/x-ndjson; charset=utf-8'); + + const lines = response.body.trim().split('\n'); + assert.ok(lines.length > 0); + lines.forEach(line => { + JSON.parse(line); // Should not throw + }); + }); + + test('should filter by fields parameter', async () => { + const response = await makeRequest('/api/exports/readings?format=csv&fields=id,meterId,value', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.ok(response.body.includes('id,meterId,value')); + assert.ok(!response.body.includes('timestamp')); + assert.ok(!response.body.includes('unit')); + }); + + test('should return 400 for invalid fields', async () => { + const response = await makeRequest('/api/exports/readings?format=csv&fields=invalid,nonexistent', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 400); + const data = JSON.parse(response.body); + assert.ok(data.error.includes('Invalid fields')); + }); + + test('should filter by meterIds', async () => { + const response = await makeRequest('/api/exports/readings?format=json&meterIds=meter-001', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + const data = JSON.parse(response.body); + data.forEach(reading => { + assert.strictEqual(reading.meterId, 'meter-001'); + }); + }); + + test('should filter by date range', async () => { + const response = await makeRequest('/api/exports/readings?format=json&startDate=2026-01-15&endDate=2026-01-15', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + const data = JSON.parse(response.body); + data.forEach(reading => { + const date = reading.timestamp.split('T')[0]; + assert.strictEqual(date, '2026-01-15'); + }); + }); + + test('should include date range in filename', async () => { + const response = await makeRequest('/api/exports/readings?format=csv&startDate=2026-01-01&endDate=2026-06-01', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.ok(response.headers['content-disposition'].includes('2026-01-01-to-2026-06-01')); + }); + + test('should return 400 for invalid format', async () => { + const response = await makeRequest('/api/exports/readings?format=xml', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 400); + }); + }); + + describe('GET /api/exports/analytics/:summaryType', () => { + test('should return 401 without authentication', async () => { + const response = await makeRequest('/api/exports/analytics/daily'); + assert.strictEqual(response.statusCode, 401); + }); + + test('should return daily analytics', async () => { + const response = await makeRequest('/api/exports/analytics/daily?format=csv', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.ok(response.body.includes('date,totalConsumption')); + }); + + test('should return weekly analytics', async () => { + const response = await makeRequest('/api/exports/analytics/weekly?format=json', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + const data = JSON.parse(response.body); + assert.ok(Array.isArray(data)); + }); + + test('should return monthly analytics', async () => { + const response = await makeRequest('/api/exports/analytics/monthly?format=json', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + const data = JSON.parse(response.body); + assert.ok(Array.isArray(data)); + }); + + test('should return 400 for invalid summary type', async () => { + const response = await makeRequest('/api/exports/analytics/invalid', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 400); + const data = JSON.parse(response.body); + assert.ok(data.error.includes('Invalid summary type')); + }); + + test('should filter daily analytics by date range', async () => { + const response = await makeRequest('/api/exports/analytics/daily?format=json&startDate=2026-01-15&endDate=2026-01-16', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + const data = JSON.parse(response.body); + data.forEach(item => { + assert.ok(item.date >= '2026-01-15'); + assert.ok(item.date <= '2026-01-16'); + }); + }); + }); + + describe('GET /api/exports/system-report', () => { + test('should return 401 without authentication', async () => { + const response = await makeRequest('/api/exports/system-report'); + assert.strictEqual(response.statusCode, 401); + }); + + test('should return 403 without admin role', async () => { + const response = await makeRequest('/api/exports/system-report', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 403); + }); + + test('should return system report with admin role', async () => { + const response = await makeRequest('/api/exports/system-report?format=json', { + headers: { + Authorization: 'Bearer test-token', + 'x-role': 'admin', + }, + }); + + assert.strictEqual(response.statusCode, 200); + const data = JSON.parse(response.body); + assert.ok(data.meters); + assert.ok(data.readings); + assert.ok(data.alerts); + assert.ok(data.summary); + }); + + test('should filter sections', async () => { + const response = await makeRequest('/api/exports/system-report?format=json§ions=meters,summary', { + headers: { + Authorization: 'Bearer test-token', + 'x-role': 'admin', + }, + }); + + assert.strictEqual(response.statusCode, 200); + const data = JSON.parse(response.body); + assert.ok(data.meters); + assert.ok(data.summary); + assert.ok(!data.readings); + assert.ok(!data.alerts); + }); + + test('should handle CSV format for system report', async () => { + const response = await makeRequest('/api/exports/system-report?format=csv', { + headers: { + Authorization: 'Bearer test-token', + 'x-role': 'admin', + }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.strictEqual(response.headers['content-type'], 'text/csv; charset=utf-8'); + assert.ok(response.body.includes('_section')); + }); + }); + + describe('GET /api/exports/meters', () => { + test('should return 401 without authentication', async () => { + const response = await makeRequest('/api/exports/meters'); + assert.strictEqual(response.statusCode, 401); + }); + + test('should return meters with valid authentication', async () => { + const response = await makeRequest('/api/exports/meters?format=csv', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.ok(response.body.includes('id,name,location,status')); + }); + + test('should filter by status', async () => { + const response = await makeRequest('/api/exports/meters?format=json&status=online', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + const data = JSON.parse(response.body); + data.forEach(meter => { + assert.strictEqual(meter.status, 'online'); + }); + }); + + test('should filter by location', async () => { + const response = await makeRequest('/api/exports/meters?format=json&location=Building A', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + const data = JSON.parse(response.body); + data.forEach(meter => { + assert.strictEqual(meter.location, 'Building A'); + }); + }); + }); + + describe('Streaming and Performance', () => { + test('should set Transfer-Encoding: chunked', async () => { + const response = await makeRequest('/api/exports/readings?format=csv', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.strictEqual(response.headers['transfer-encoding'], 'chunked'); + }); + + test('should handle pretty-printed JSON', async () => { + const response = await makeRequest('/api/exports/readings?format=json&pretty=true', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.ok(response.body.includes('\n')); + }); + + test('should return empty file for no data', async () => { + // This test would require modifying the mock to return empty data + // For now, we just verify the endpoint handles the request + const response = await makeRequest('/api/exports/readings?format=csv&meterIds=nonexistent', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 200); + assert.ok(response.headers['content-disposition'].includes('attachment')); + }); + }); + + describe('Error Handling', () => { + test('should return 400 for malformed fields parameter', async () => { + const response = await makeRequest('/api/exports/readings?format=csv&fields=invalid', { + headers: { Authorization: 'Bearer test-token' }, + }); + + assert.strictEqual(response.statusCode, 400); + }); + + test('should handle invalid date format gracefully', async () => { + const response = await makeRequest('/api/exports/readings?format=csv&startDate=invalid-date', { + headers: { Authorization: 'Bearer test-token' }, + }); + + // Should not crash - may return empty results or error + assert.ok([200, 400].includes(response.statusCode)); + }); + + test('should reject invalid authentication format', async () => { + const response = await makeRequest('/api/exports/readings?format=csv', { + headers: { Authorization: 'InvalidFormat token' }, + }); + + assert.strictEqual(response.statusCode, 401); + }); + }); +}); diff --git a/test/queue.test.js b/test/queue.test.js new file mode 100644 index 00000000..668b2c18 --- /dev/null +++ b/test/queue.test.js @@ -0,0 +1,210 @@ +const { test, describe, before, after } = require('node:test'); +const assert = require('node:assert'); +const { queue, JobStatus, Priority } = require('../src/services/queue'); + +describe('JobQueue', () => { + before(() => { + // Reset queue state before tests + queue.jobs.clear(); + queue.queuedJobs = []; + queue.runningJobs.clear(); + queue.handlers.clear(); + queue.activeCount = 0; + }); + + after(async () => { + // Cleanup after tests + await queue.stop(); + queue.jobs.clear(); + queue.queuedJobs = []; + queue.runningJobs.clear(); + queue.handlers.clear(); + }); + + test('should register a handler', () => { + const handler = async (data) => ({ result: 'ok', data }); + queue.registerHandler('test', handler); + + assert.strictEqual(queue.handlers.has('test'), true); + }); + + test('should throw error when registering non-function handler', () => { + assert.throws(() => { + queue.registerHandler('invalid', 'not a function'); + }, /Handler for job type "invalid" must be a function/); + }); + + test('should add a job to the queue', () => { + const handler = async (data) => ({ result: 'ok', data }); + queue.registerHandler('addTest', handler); + + const jobId = queue.add('addTest', { test: 'data' }); + + assert.strictEqual(typeof jobId, 'string'); + assert.strictEqual(queue.jobs.has(jobId), true); + + const job = queue.jobs.get(jobId); + assert.strictEqual(job.type, 'addTest'); + assert.strictEqual(job.status, JobStatus.QUEUED); + assert.strictEqual(job.data.test, 'data'); + }); + + test('should get job status', () => { + const handler = async (data) => ({ result: 'ok', data }); + queue.registerHandler('statusTest', handler); + + const jobId = queue.add('statusTest', { test: 'data' }); + const status = queue.getStatus(jobId); + + assert.strictEqual(status.id, jobId); + assert.strictEqual(status.type, 'statusTest'); + assert.strictEqual(status.status, JobStatus.QUEUED); + }); + + test('should return null for non-existent job', () => { + const status = queue.getStatus('non-existent-job-id'); + assert.strictEqual(status, null); + }); + + test('should cancel a queued job', () => { + const handler = async (data) => ({ result: 'ok', data }); + queue.registerHandler('cancelTest', handler); + + const jobId = queue.add('cancelTest', { test: 'data' }); + const cancelled = queue.cancel(jobId); + + assert.strictEqual(cancelled, true); + + const job = queue.jobs.get(jobId); + assert.strictEqual(job.status, JobStatus.CANCELLED); + }); + + test('should not cancel a non-existent job', () => { + const cancelled = queue.cancel('non-existent-job-id'); + assert.strictEqual(cancelled, false); + }); + + test('should get queue statistics', () => { + const stats = queue.getStats(); + + assert.strictEqual(typeof stats.queued, 'number'); + assert.strictEqual(typeof stats.running, 'number'); + assert.strictEqual(typeof stats.completed, 'number'); + assert.strictEqual(typeof stats.failed, 'number'); + assert.strictEqual(typeof stats.cancelled, 'number'); + assert.strictEqual(typeof stats.total, 'number'); + assert.strictEqual(typeof stats.activeCount, 'number'); + assert.strictEqual(typeof stats.maxConcurrency, 'number'); + }); + + test('should process a job successfully', async () => { + const handler = async (data) => ({ result: 'success', data }); + queue.registerHandler('processTest', handler); + + queue.start(); + + const jobId = queue.add('processTest', { test: 'data' }); + + // Wait for job to complete + await new Promise(resolve => setTimeout(resolve, 100)); + + const job = queue.jobs.get(jobId); + assert.strictEqual(job.status, JobStatus.COMPLETED); + assert.strictEqual(job.result.result, 'success'); + }); + + test('should retry failed job with backoff', async () => { + let attemptCount = 0; + const handler = async (data) => { + attemptCount++; + if (attemptCount < 3) { + throw new Error('Simulated failure'); + } + return { result: 'success after retries' }; + }; + + queue.registerHandler('retryTest', handler); + + const jobId = queue.add('retryTest', { test: 'data' }, { maxAttempts: 3 }); + + // Wait for retries to complete + await new Promise(resolve => setTimeout(resolve, 5000)); + + const job = queue.jobs.get(jobId); + assert.strictEqual(job.status, JobStatus.COMPLETED); + assert.strictEqual(job.attempts, 3); + }); + + test('should fail job after max attempts', async () => { + const handler = async (data) => { + throw new Error('Always fails'); + }; + + queue.registerHandler('failTest', handler); + + const jobId = queue.add('failTest', { test: 'data' }, { maxAttempts: 2 }); + + // Wait for retries to complete + await new Promise(resolve => setTimeout(resolve, 3000)); + + const job = queue.jobs.get(jobId); + assert.strictEqual(job.status, JobStatus.FAILED); + assert.strictEqual(job.attempts, 2); + assert.strictEqual(job.error, 'Always fails'); + }); + + test('should respect priority ordering', () => { + queue.registerHandler('priorityTest', async (data) => data); + + const lowJob = queue.add('priorityTest', {}, { priority: Priority.LOW }); + const normalJob = queue.add('priorityTest', {}, { priority: Priority.NORMAL }); + const highJob = queue.add('priorityTest', {}, { priority: Priority.HIGH }); + + // High priority should be first in queue + assert.strictEqual(queue.queuedJobs[0], highJob); + assert.strictEqual(queue.queuedJobs[1], normalJob); + assert.strictEqual(queue.queuedJobs[2], lowJob); + }); + + test('should handle delayed job execution', async () => { + let executed = false; + const handler = async (data) => { + executed = true; + return { result: 'ok' }; + }; + + queue.registerHandler('delayTest', handler); + + const jobId = queue.add('delayTest', {}, { delay: 100 }); + + const job = queue.jobs.get(jobId); + assert.strictEqual(job.status, JobStatus.QUEUED); + assert.strictEqual(executed, false); + + // Wait for delay + await new Promise(resolve => setTimeout(resolve, 200)); + + assert.strictEqual(executed, true); + }); + + test('should emit events', async () => { + const events = []; + + queue.on('added', (job) => events.push({ type: 'added', jobId: job.id })); + queue.on('started', (job) => events.push({ type: 'started', jobId: job.id })); + queue.on('completed', (job) => events.push({ type: 'completed', jobId: job.id })); + + const handler = async (data) => ({ result: 'ok' }); + queue.registerHandler('eventTest', handler); + + const jobId = queue.add('eventTest', {}); + + // Wait for processing + await new Promise(resolve => setTimeout(resolve, 100)); + + assert.strictEqual(events.length, 3); + assert.strictEqual(events[0].type, 'added'); + assert.strictEqual(events[1].type, 'started'); + assert.strictEqual(events[2].type, 'completed'); + }); +}); diff --git a/test/scheduler.test.js b/test/scheduler.test.js new file mode 100644 index 00000000..d36aedf9 --- /dev/null +++ b/test/scheduler.test.js @@ -0,0 +1,201 @@ +const { test, describe, before, after } = require('node:test'); +const assert = require('node:assert'); +const { scheduler } = require('../src/services/scheduler'); + +describe('Scheduler', () => { + before(() => { + // Reset scheduler state before tests + scheduler.schedules.clear(); + scheduler.isRunning = false; + }); + + after(async () => { + // Cleanup after tests + await scheduler.stop(); + scheduler.schedules.clear(); + }); + + test('should create a schedule', () => { + const handler = async () => ({ result: 'ok' }); + const scheduleId = scheduler.schedule('test-schedule', '1000', handler); + + assert.strictEqual(typeof scheduleId, 'string'); + assert.strictEqual(scheduler.schedules.has('test-schedule'), true); + + const schedule = scheduler.schedules.get('test-schedule'); + assert.strictEqual(schedule.name, 'test-schedule'); + assert.strictEqual(schedule.interval, 1000); + }); + + test('should throw error when creating duplicate schedule', () => { + const handler = async () => ({ result: 'ok' }); + scheduler.schedule('duplicate-test', '1000', handler); + + assert.throws(() => { + scheduler.schedule('duplicate-test', '1000', handler); + }, /Schedule with name "duplicate-test" already exists/); + }); + + test('should throw error when handler is not a function', () => { + assert.throws(() => { + scheduler.schedule('invalid-handler', '1000', 'not a function'); + }, /Handler must be a function/); + }); + + test('should get schedule information', () => { + const handler = async () => ({ result: 'ok' }); + scheduler.schedule('get-test', '2000', handler); + + const scheduleInfo = scheduler.getSchedule('get-test'); + + assert.strictEqual(scheduleInfo.name, 'get-test'); + assert.strictEqual(scheduleInfo.interval, 2000); + assert.strictEqual(scheduleInfo.isActive, false); + assert.strictEqual(typeof scheduleInfo.createdAt, 'object'); + }); + + test('should return null for non-existent schedule', () => { + const scheduleInfo = scheduler.getSchedule('non-existent'); + assert.strictEqual(scheduleInfo, null); + }); + + test('should get all schedules', () => { + const handler = async () => ({ result: 'ok' }); + scheduler.schedule('all-test-1', '3000', handler); + scheduler.schedule('all-test-2', '4000', handler); + + const allSchedules = scheduler.getAllSchedules(); + + assert.strictEqual(allSchedules.length >= 2, true); + assert.strictEqual(allSchedules.some(s => s.name === 'all-test-1'), true); + assert.strictEqual(allSchedules.some(s => s.name === 'all-test-2'), true); + }); + + test('should cancel a schedule', () => { + const handler = async () => ({ result: 'ok' }); + scheduler.schedule('cancel-test', '5000', handler); + + const cancelled = scheduler.cancelSchedule('cancel-test'); + + assert.strictEqual(cancelled, true); + assert.strictEqual(scheduler.schedules.has('cancel-test'), false); + }); + + test('should not cancel non-existent schedule', () => { + const cancelled = scheduler.cancelSchedule('non-existent'); + assert.strictEqual(cancelled, false); + }); + + test('should start the scheduler', () => { + const handler = async () => ({ result: 'ok' }); + scheduler.schedule('start-test', '10000', handler); + + scheduler.start(); + + assert.strictEqual(scheduler.isRunning, true); + + const schedule = scheduler.schedules.get('start-test'); + assert.strictEqual(schedule.intervalId !== null, true); + }); + + test('should stop the scheduler', () => { + scheduler.start(); + + scheduler.stop(); + + assert.strictEqual(scheduler.isRunning, false); + + for (const schedule of scheduler.schedules.values()) { + assert.strictEqual(schedule.intervalId, null); + } + }); + + test('should parse numeric interval', () => { + const interval = scheduler._parseInterval('5000'); + assert.strictEqual(interval, 5000); + }); + + test('should parse cron expression for every minute', () => { + const interval = scheduler._parseInterval('* * * * *'); + assert.strictEqual(interval, 60 * 1000); + }); + + test('should parse cron expression for every hour', () => { + const interval = scheduler._parseInterval('0 * * * *'); + assert.strictEqual(interval, 60 * 60 * 1000); + }); + + test('should parse cron expression for every day', () => { + const interval = scheduler._parseInterval('0 0 * * *'); + assert.strictEqual(interval, 24 * 60 * 60 * 1000); + }); + + test('should return -1 for invalid cron expression', () => { + const interval = scheduler._parseInterval('invalid'); + assert.strictEqual(interval, -1); + }); + + test('should execute scheduled job', async () => { + let executed = false; + let executionCount = 0; + + const handler = async () => { + executed = true; + executionCount++; + return { result: 'ok' }; + }; + + scheduler.schedule('exec-test', '100', handler); + scheduler.start(); + + // Wait for first execution + await new Promise(resolve => setTimeout(resolve, 150)); + + assert.strictEqual(executed, true); + assert.strictEqual(executionCount, 1); + + // Wait for second execution + await new Promise(resolve => setTimeout(resolve, 100)); + + assert.strictEqual(executionCount, 2); + + scheduler.stop(); + }); + + test('should handle handler errors gracefully', async () => { + let errorLogged = false; + + const handler = async () => { + throw new Error('Handler error'); + }; + + scheduler.schedule('error-test', '100', handler); + scheduler.start(); + + // Wait for execution + await new Promise(resolve => setTimeout(resolve, 150)); + + // Should not crash, just log error + assert.strictEqual(scheduler.isRunning, true); + + scheduler.stop(); + }); + + test('should update run count and timestamps', async () => { + const handler = async () => ({ result: 'ok' }); + + scheduler.schedule('timestamp-test', '100', handler); + scheduler.start(); + + const schedule = scheduler.schedules.get('timestamp-test'); + + // Wait for execution + await new Promise(resolve => setTimeout(resolve, 150)); + + assert.strictEqual(schedule.runCount > 0, true); + assert.strictEqual(schedule.lastRun !== null, true); + assert.strictEqual(schedule.nextRun !== null, true); + + scheduler.stop(); + }); +}); diff --git a/test/security.test.js b/test/security.test.js new file mode 100644 index 00000000..0e154079 --- /dev/null +++ b/test/security.test.js @@ -0,0 +1,188 @@ +const { describe, it, after } = require('node:test'); +const assert = require('node:assert'); + +const app = require('../src/app'); +const server = app.listen(0); + +after(() => server.close()); + +describe('Security Tests', () => { + describe('Request Body Size Limits', () => { + it('rejects oversized JSON payload with 413', async () => { + const port = server.address().port; + // Create a payload larger than 1MB (default limit) + const largePayload = { + data: 'x'.repeat(2 * 1024 * 1024), // 2MB of data + }; + + try { + const res = await fetch(`http://localhost:${port}/test`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(largePayload), + }); + assert.strictEqual(res.status, 413); + + const data = await res.json(); + assert.ok(data.error || data.message); + } catch (error) { + // Express will reject the payload before it reaches our route + // The error might be a network error due to payload size + assert.ok(error.message.includes('payload') || error.message.includes('body')); + } + }); + + it('accepts payload within size limit', async () => { + const port = server.address().port; + const validPayload = { + data: 'x'.repeat(500 * 1024), // 500KB - within 1MB limit + }; + + try { + const res = await fetch(`http://localhost:${port}/test`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(validPayload), + }); + // Should return 404 (route doesn't exist) but not 413 + assert.notStrictEqual(res.status, 413); + } catch (error) { + // Network errors are acceptable for non-existent routes + assert.ok(true); + } + }); + }); + + describe('XSS Protection', () => { + it('sanitizes XSS payload in error responses', async () => { + const port = server.address().port; + const xssPayload = ''; + + const res = await fetch(`http://localhost:${port}/${xssPayload}`); + assert.strictEqual(res.status, 404); + + const data = await res.json(); + // The message should be sanitized (HTML entities encoded) + assert.ok(!data.message.includes(''; + + const res = await fetch(`http://localhost:${port}/`, { + headers: { 'x-correlation-id': maliciousCorrelationId }, + }); + + assert.strictEqual(res.status, 200); + const returnedCorrelationId = res.headers.get('x-correlation-id'); + + // The returned correlation ID should be the same but not cause issues + assert.ok(returnedCorrelationId); + }); + }); + + describe('Error Message Security', () => { + it('does not expose sensitive information in production mode', async () => { + const port = server.address().port; + + // In non-production mode, error messages are shown + // In production, they should be generic + const res = await fetch(`http://localhost:${port}/non-existent-route`); + assert.strictEqual(res.status, 404); + + const data = await res.json(); + assert.ok(data.error); + assert.ok(data.message); + }); + }); +}); diff --git a/test/server-integration.test.js b/test/server-integration.test.js new file mode 100644 index 00000000..9a7c1bba --- /dev/null +++ b/test/server-integration.test.js @@ -0,0 +1,78 @@ +const { describe, it, before, after } = require('node:test'); +const assert = require('node:assert'); +const { spawn } = require('node:child_process'); +const http = require('node:http'); + +describe('server integration', () => { + let serverProcess; + let serverPort = 3001; // Use different port to avoid conflicts + + before(async () => { + // Start server in a separate process + serverProcess = spawn('node', ['src/server.js'], { + env: { ...process.env, PORT: serverPort.toString() }, + stdio: 'pipe', + }); + + // Wait for server to start + await new Promise((resolve, reject) => { + let attempts = 0; + const maxAttempts = 10; + + const checkServer = () => { + attempts++; + const req = http.get(`http://localhost:${serverPort}/`, (res) => { + if (res.statusCode === 200) { + resolve(); + } else { + reject(new Error(`Server responded with status ${res.statusCode}`)); + } + }); + + req.on('error', (err) => { + if (attempts < maxAttempts) { + setTimeout(checkServer, 500); + } else { + reject(new Error('Server failed to start after multiple attempts')); + } + }); + }; + + checkServer(); + }); + }); + + after(() => { + if (serverProcess) { + serverProcess.kill('SIGTERM'); + } + }); + + it('server starts and responds to HTTP requests', async () => { + const response = await fetch(`http://localhost:${serverPort}/`); + assert.strictEqual(response.status, 200); + + const data = await response.json(); + assert.strictEqual(data.project, 'Equipchain'); + assert.strictEqual(data.status, 'Monitoring Meters'); + }); + + it('server handles health check endpoint', async () => { + const response = await fetch(`http://localhost:${serverPort}/health`); + assert.strictEqual(response.status, 200); + + const data = await response.json(); + assert.strictEqual(data.status, 'ok'); + assert.ok(data.timestamp); + }); + + it('server includes correlation ID in responses', async () => { + const response = await fetch(`http://localhost:${serverPort}/`); + assert.ok(response.headers.get('x-correlation-id')); + }); + + it('server returns 404 for non-existent routes', async () => { + const response = await fetch(`http://localhost:${serverPort}/non-existent`); + assert.strictEqual(response.status, 404); + }); +}); diff --git a/test/server.test.js b/test/server.test.js index 61ba433f..54103e5e 100644 --- a/test/server.test.js +++ b/test/server.test.js @@ -1,7 +1,7 @@ const { describe, it, after } = require('node:test'); const assert = require('node:assert'); -const app = require('../index'); +const app = require('../src/app'); const server = app.listen(0); after(() => server.close());