From c237ac8db2da4babc0bd58733f56dd87caa8559a Mon Sep 17 00:00:00 2001 From: Daniel Omoloba Date: Sun, 21 Jun 2026 22:27:01 +0000 Subject: [PATCH 1/2] feat: implement campaign analytics rollups, trending, and impact metrics (#85) - Add CampaignHourlyStat, CampaignMonthlyStat, CampaignTrending, RollupTracker models - Extend JobType enum with analytics job types - Add analytics config block with configurable cron patterns and feature flag - Add new types: TrendingCampaign, ImpactMetrics, HistoricalStats - Implement Redis-backed cache layer for real-time campaign stats - Add getTrendingCampaigns, getCampaignImpactMetrics, getCampaignHistoricalStats - Add getAggregatedCampaignAnalytics for admin dashboards - Implement hourly/monthly rollup logic (idempotent via upsert) - Implement trending campaign refresh with composite scoring - Add new endpoints: GET /campaigns/trending (period, sortBy, limit) GET /campaigns/:id/impact-metrics GET /campaigns/:id/statistics/historical (granularity, date range) GET /analytics/campaigns (admin, paginated rollup data) - Hook real-time cache updates into DonationService and DistributionService - Create analytics worker with scheduled jobs (hourly, monthly, trending, reconcile) - Feature-flag analytics worker via ANALYTICS_WORKER_ENABLED - Document new endpoints in API.md Closes #85 --- docs/API.md | 45 ++ prisma/schema.prisma | 79 ++ src/config/index.ts | 13 + src/controllers/analytics.controller.ts | 83 +- src/index.ts | 9 + src/routes/analytics.routes.ts | 12 + src/routes/campaign.routes.ts | 85 ++ src/services/analytics.service.ts | 986 ++++++++++++++++++++++++ src/services/distribution.service.ts | 12 + src/services/donation.service.ts | 13 + src/types/index.ts | 67 ++ src/workers/analytics.worker.ts | 182 +++++ 12 files changed, 1585 insertions(+), 1 deletion(-) create mode 100644 src/workers/analytics.worker.ts diff --git a/docs/API.md b/docs/API.md index dd4180a..ad1f33f 100644 --- a/docs/API.md +++ b/docs/API.md @@ -262,6 +262,33 @@ GET /campaigns/:id/stats Authorization: Bearer ``` +#### Get Trending Campaigns + +```http +GET /campaigns/trending?period=last24h&sortBy=trendScore&limit=10 +Authorization: Bearer +``` + +Returns top campaigns by donation velocity, donor growth, or impact. + +#### Get Campaign Impact Metrics + +```http +GET /campaigns/:id/impact-metrics +Authorization: Bearer +``` + +Returns comprehensive impact metrics: total donations, donor growth, distributions, beneficiaries reached, conversion rates. + +#### Get Campaign Historical Statistics + +```http +GET /campaigns/:id/statistics/historical?granularity=hourly&startDate=2024-01-01&endDate=2024-01-31 +Authorization: Bearer +``` + +Returns monthly or hourly rollup data for trend charts. + ### Organizations #### Create Organization @@ -538,6 +565,24 @@ Authorization: Bearer } ``` +### Analytics + +#### Get Aggregated Campaign Analytics (Admin) + +```http +GET /analytics/campaigns?page=1&limit=10&status=ACTIVE&sortBy=createdAt&sortOrder=desc +Authorization: Bearer +``` + +Returns aggregated campaign metrics from rollup tables for admin dashboards. + +#### Get Campaign Analytics + +```http +GET /analytics/campaign/:campaignId +Authorization: Bearer +``` + ### Distributions #### Create Distribution diff --git a/prisma/schema.prisma b/prisma/schema.prisma index ec51f6c..93a9311 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -626,6 +626,10 @@ enum JobType { DISTRIBUTION_PROCESSING REPORT_GENERATION CACHE_INVALIDATION + ANALYTICS_HOURLY + ANALYTICS_MONTHLY + ANALYTICS_REALTIME + ANALYTICS_TRENDING_REFRESH } model QueueJob { @@ -745,3 +749,78 @@ model FraudReport { @@index([reporterId]) @@index([createdAt]) } + +// ============================================ +// ANALYTICS ROLLUPS & TRENDING +// ============================================ + +model CampaignHourlyStat { + id String @id @default(cuid()) + campaignId String + hour DateTime + donationCount Int @default(0) + donationVolume Decimal @default(0) @db.Decimal(20, 8) + uniqueDonors Int @default(0) + distributionCount Int @default(0) + itemsDistributed Int @default(0) + distributionVolume Decimal @default(0) @db.Decimal(20, 8) + newBeneficiaries Int @default(0) + activeDonors Int @default(0) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@unique([campaignId, hour]) + @@index([campaignId]) + @@index([hour]) +} + +model CampaignMonthlyStat { + id String @id @default(cuid()) + campaignId String + month DateTime + donationCount Int @default(0) + donationVolume Decimal @default(0) @db.Decimal(20, 8) + uniqueDonors Int @default(0) + distributionCount Int @default(0) + itemsDistributed Int @default(0) + distributionVolume Decimal @default(0) @db.Decimal(20, 8) + donorGrowth Int @default(0) + distributionReach Int @default(0) + campaignActivity Int @default(0) + activeDonors Int @default(0) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@unique([campaignId, month]) + @@index([campaignId]) + @@index([month]) +} + +model CampaignTrending { + id String @id @default(cuid()) + campaignId String + trendScore Decimal @default(0) @db.Decimal(20, 8) + donationVelocity Decimal @default(0) @db.Decimal(20, 8) + donorGrowth Int @default(0) + distributionImpact Decimal @default(0) @db.Decimal(20, 8) + period String @default("last24h") + rank Int @default(0) + refreshedAt DateTime @default(now()) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@unique([campaignId, period]) + @@index([trendScore]) + @@index([period]) + @@index([rank]) +} + +model RollupTracker { + id String @id @default(cuid()) + type String @unique + lastProcessedTimestamp DateTime + lastHour DateTime? + lastMonth DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt +} diff --git a/src/config/index.ts b/src/config/index.ts index a429049..e678c8c 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -84,6 +84,19 @@ export const config = { // Notify donors when a campaign is suspended for a fraud-related reason. notifyDonorsOnFraudSuspension: process.env.MODERATION_NOTIFY_DONORS_ON_FRAUD !== 'false', }, + + analytics: { + // Cron patterns for rollup jobs (configurable via env vars) + hourlyRollupCron: process.env.ANALYTICS_HOURLY_CRON || '5 * * * *', + monthlyRollupCron: process.env.ANALYTICS_MONTHLY_CRON || '0 2 1 * *', + trendingRefreshCron: process.env.ANALYTICS_TRENDING_CRON || '*/15 * * * *', + // Feature flag to disable analytics worker + analyticsWorkerEnabled: process.env.ANALYTICS_WORKER_ENABLED !== 'false', + // Cache TTL for campaign stats in seconds + campaignStatsCacheTTL: parseInt(process.env.ANALYTICS_CACHE_TTL || '3600', 10), + // Number of trending campaigns to track + trendingCampaignsCount: parseInt(process.env.ANALYTICS_TRENDING_COUNT || '20', 10), + }, }; export default config; diff --git a/src/controllers/analytics.controller.ts b/src/controllers/analytics.controller.ts index c7e6fc8..663e497 100644 --- a/src/controllers/analytics.controller.ts +++ b/src/controllers/analytics.controller.ts @@ -1,4 +1,4 @@ -import { Response, NextFunction } from 'express'; +import { Request, Response, NextFunction } from 'express'; import { AnalyticsService } from '../services/analytics.service'; import { AuthRequest } from '../types'; import { AppError } from '../middleware/error'; @@ -91,4 +91,85 @@ export class AnalyticsController { next(error); } } + + static async getTrendingCampaigns(req: Request, res: Response, next: NextFunction): Promise { + try { + const period = (req.query.period as 'last24h' | 'last7d' | 'last30d') || 'last24h'; + const sortBy = (req.query.sortBy as 'trendScore' | 'donationVelocity' | 'distributionImpact') || 'trendScore'; + const limit = req.query.limit ? parseInt(req.query.limit as string) : 10; + + const trending = await AnalyticsService.getTrendingCampaigns({ period, sortBy, limit }); + + res.status(200).json({ + success: true, + data: trending, + }); + } catch (error) { + next(error); + } + } + + static async getCampaignImpactMetrics(req: Request, res: Response, next: NextFunction): Promise { + try { + const { id } = req.params; + const metrics = await AnalyticsService.getCampaignImpactMetrics(id); + + res.status(200).json({ + success: true, + data: metrics, + }); + } catch (error) { + next(error); + } + } + + static async getCampaignHistoricalStats(req: Request, res: Response, next: NextFunction): Promise { + try { + const { id } = req.params; + const granularity = (req.query.granularity as 'hourly' | 'monthly') || 'hourly'; + const startDate = req.query.startDate ? new Date(req.query.startDate as string) : undefined; + const endDate = req.query.endDate ? new Date(req.query.endDate as string) : undefined; + + const range = startDate && endDate ? { startDate, endDate } : undefined; + + const stats = await AnalyticsService.getCampaignHistoricalStats(id, granularity, range); + + res.status(200).json({ + success: true, + data: stats, + }); + } catch (error) { + next(error); + } + } + + static async getAggregatedCampaignAnalytics(req: AuthRequest, res: Response, next: NextFunction): Promise { + try { + if (!req.user || req.user.role !== 'ADMIN') { + throw new AppError('Admin access required', 403); + } + + const filters = { + status: req.query.status as string, + startDate: req.query.startDate ? new Date(req.query.startDate as string) : undefined, + endDate: req.query.endDate ? new Date(req.query.endDate as string) : undefined, + }; + + const pagination = { + page: req.query.page ? parseInt(req.query.page as string) : 1, + limit: req.query.limit ? parseInt(req.query.limit as string) : 10, + sortBy: req.query.sortBy as string || 'createdAt', + sortOrder: (req.query.sortOrder as 'asc' | 'desc') || 'desc', + }; + + const result = await AnalyticsService.getAggregatedCampaignAnalytics(filters, pagination); + + res.status(200).json({ + success: true, + ...result, + }); + } catch (error) { + next(error); + } + } } diff --git a/src/index.ts b/src/index.ts index 1c1a72b..e80ba39 100644 --- a/src/index.ts +++ b/src/index.ts @@ -143,6 +143,15 @@ const startServer = async (): Promise => { .catch((error) => logger.error('Failed to start moderation worker:', error)); } + // Start analytics worker for rollups, trending refresh, and cache updates. + // Feature-flagged via ANALYTICS_WORKER_ENABLED. + if (config.analytics.analyticsWorkerEnabled) { + import('./workers/analytics.worker.js') + .then(({ scheduleAnalyticsJobs }) => scheduleAnalyticsJobs()) + .then(() => logger.info('Analytics worker started')) + .catch((error) => logger.error('Failed to start analytics worker:', error)); + } + // Start HTTP server httpServer.listen(config.port, () => { logger.info(`Server running on port ${config.port} in ${config.env} mode`); diff --git a/src/routes/analytics.routes.ts b/src/routes/analytics.routes.ts index 093d4e6..90d6d2c 100644 --- a/src/routes/analytics.routes.ts +++ b/src/routes/analytics.routes.ts @@ -65,4 +65,16 @@ router.post( AnalyticsController.generateReport ); +/** + * @route GET /api/v1/analytics/campaigns + * @desc Admin endpoint to query aggregated campaign metrics + * @access Private (Admin) + */ +router.get( + '/campaigns', + authenticate, + analyticsLimiter, + AnalyticsController.getAggregatedCampaignAnalytics +); + export default router; diff --git a/src/routes/campaign.routes.ts b/src/routes/campaign.routes.ts index b3a7d59..7639cfb 100644 --- a/src/routes/campaign.routes.ts +++ b/src/routes/campaign.routes.ts @@ -1,6 +1,7 @@ import { Router } from 'express'; import { CampaignController } from '../controllers/campaign.controller'; import { ModerationController } from '../controllers/moderation.controller'; +import { AnalyticsController } from '../controllers/analytics.controller'; import { authenticate } from '../middleware/auth'; import { campaignCreateLimiter, reportLimiter } from '../middleware/rateLimit'; import { validate } from '../middleware/validation'; @@ -89,6 +90,35 @@ router.post('/', campaignCreateLimiter, validate(campaignSchema), CampaignContro */ router.get('/', CampaignController.getCampaigns); +/** + * @swagger + * /api/v1/campaigns/trending: + * get: + * summary: Get trending campaigns by donation velocity, donor growth, or impact + * tags: [Campaigns] + * security: + * - bearerAuth: [] + * parameters: + * - in: query + * name: period + * schema: + * type: string + * enum: [last24h, last7d, last30d] + * - in: query + * name: sortBy + * schema: + * type: string + * enum: [trendScore, donationVelocity, distributionImpact] + * - in: query + * name: limit + * schema: + * type: integer + * responses: + * 200: + * description: Trending campaigns retrieved successfully + */ +router.get('/trending', AnalyticsController.getTrendingCampaigns); + /** * @swagger * /api/v1/campaigns/{id}: @@ -283,6 +313,61 @@ router.post('/:campaignId/beneficiaries', CampaignController.assignBeneficiary); */ router.get('/:id/stats', CampaignController.getCampaignStats); +/** + * @swagger + * /api/v1/campaigns/{id}/impact-metrics: + * get: + * summary: Get campaign impact metrics from aggregated data + * tags: [Campaigns] + * security: + * - bearerAuth: [] + * parameters: + * - in: path + * name: id + * required: true + * schema: + * type: string + * responses: + * 200: + * description: Campaign impact metrics retrieved successfully + */ +router.get('/:id/impact-metrics', AnalyticsController.getCampaignImpactMetrics); + +/** + * @swagger + * /api/v1/campaigns/{id}/statistics/historical: + * get: + * summary: Get monthly or weekly rollup data for trend charts + * tags: [Campaigns] + * security: + * - bearerAuth: [] + * parameters: + * - in: path + * name: id + * required: true + * schema: + * type: string + * - in: query + * name: granularity + * schema: + * type: string + * enum: [hourly, monthly] + * - in: query + * name: startDate + * schema: + * type: string + * format: date-time + * - in: query + * name: endDate + * schema: + * type: string + * format: date-time + * responses: + * 200: + * description: Historical statistics retrieved successfully + */ +router.get('/:id/statistics/historical', AnalyticsController.getCampaignHistoricalStats); + /** * @swagger * /api/v1/campaigns/{id}/reports: diff --git a/src/services/analytics.service.ts b/src/services/analytics.service.ts index 8397c5f..d1d1e5f 100644 --- a/src/services/analytics.service.ts +++ b/src/services/analytics.service.ts @@ -1,5 +1,21 @@ import prisma from '../config/database'; +import redis from '../config/redis'; import logger from '../config/logger'; +import { config } from '../config'; +import { + TrendingCampaignFilters, + TrendingCampaign, + ImpactMetrics, + HistoricalStats, + CampaignAnalyticsFilters, + PaginatedResponse, + PaginationParams, +} from '../types'; + +// Redis cache key prefixes +const CACHE_PREFIX_STATS = 'campaign:stats:'; +const CACHE_PREFIX_TRENDING_DATA = 'campaigns:trending:data'; +const CACHE_PREFIX_TRENDING_ZSET = 'campaigns:trending'; export class AnalyticsService { static async getCampaignAnalytics(campaignId: string): Promise { @@ -272,4 +288,974 @@ export class AnalyticsService { throw new Error('Invalid report type'); } } + + // ============================================ + // CACHE-BASED CAMPAIGN STATS + // ============================================ + + /** + * Get campaign stats from Redis cache, falling back to DB if cache miss. + */ + static async getCachedCampaignStats(campaignId: string): Promise> { + const cacheKey = `${CACHE_PREFIX_STATS}${campaignId}`; + try { + const cached = await redis.hgetall(cacheKey); + if (cached && Object.keys(cached).length > 0) { + return cached; + } + } catch (err) { + logger.warn(`Redis cache read failed for ${cacheKey}`, err); + } + + // Cache miss — build from DB and populate cache + const stats = await this.buildCampaignStats(campaignId); + await this.setCachedCampaignStats(campaignId, stats); + return stats; + } + + /** + * Store campaign stats in Redis cache. + */ + static async setCachedCampaignStats( + campaignId: string, + stats: Record, + ): Promise { + const cacheKey = `${CACHE_PREFIX_STATS}${campaignId}`; + try { + await redis.hset(cacheKey, stats); + await redis.expire(cacheKey, config.analytics.campaignStatsCacheTTL); + } catch (err) { + logger.warn(`Redis cache write failed for ${cacheKey}`, err); + } + } + + /** + * Build campaign stats from raw database queries. + */ + static async buildCampaignStats(campaignId: string): Promise> { + const campaign = await prisma.campaign.findUnique({ + where: { id: campaignId }, + select: { id: true, title: true, targetAmount: true, currentAmount: true, status: true }, + }); + + if (!campaign) { + throw new Error('Campaign not found'); + } + + const [donationAgg, distributionAgg] = await Promise.all([ + prisma.donation.aggregate({ + where: { campaignId, status: 'CONFIRMED' }, + _count: true, + _sum: { amount: true }, + }), + prisma.distribution.aggregate({ + where: { campaignId, status: 'COMPLETED' }, + _count: true, + _sum: { amount: true }, + }), + ]); + + const uniqueDonors = await prisma.donation.groupBy({ + by: ['userId'], + where: { campaignId, status: 'CONFIRMED', userId: { not: null } }, + }); + + const beneficiaryCount = await prisma.beneficiaryAssignment.count({ + where: { campaignId }, + }); + + const targetAmount = Number(campaign.targetAmount) || 1; + const currentAmount = Number(campaign.currentAmount) || 0; + const progress = ((currentAmount / targetAmount) * 100).toFixed(2); + + return { + campaignId: campaign.id, + title: campaign.title, + status: campaign.status, + targetAmount: String(campaign.targetAmount), + currentAmount: String(campaign.currentAmount), + totalDonations: String(donationAgg._count), + totalRaised: String(donationAgg._sum.amount || '0'), + totalDistributions: String(distributionAgg._count), + totalDistributed: String(distributionAgg._sum.amount || '0'), + uniqueDonors: String(uniqueDonors.length), + beneficiariesReached: String(beneficiaryCount), + progressPercentage: progress, + }; + } + + /** + * Incrementally update campaign stats cache after a donation event. + * Called by the analytics worker. + */ + static async incrementDonationStats(campaignId: string, amount: number): Promise { + const cacheKey = `${CACHE_PREFIX_STATS}${campaignId}`; + try { + const exists = await redis.exists(cacheKey); + if (exists) { + await redis.hincrbyfloat(cacheKey, 'totalRaised', amount); + await redis.hincrby(cacheKey, 'totalDonations', 1); + // Note: uniqueDonors is NOT incremented here — it would inflate the count. + // The hourly reconciliation job sets the correct value from grouped queries. + } + } catch (err) { + logger.warn(`Redis increment failed for ${cacheKey}`, err); + } + } + + /** + * Incrementally update campaign stats cache after a distribution event. + * Called by the analytics worker. + */ + static async incrementDistributionStats(campaignId: string, amount: number): Promise { + const cacheKey = `${CACHE_PREFIX_STATS}${campaignId}`; + try { + const exists = await redis.exists(cacheKey); + if (exists) { + await redis.hincrbyfloat(cacheKey, 'totalDistributed', amount); + await redis.hincrby(cacheKey, 'totalDistributions', 1); + } + } catch (err) { + logger.warn(`Redis increment failed for ${cacheKey}`, err); + } + } + + /** + * Invalidates the Redis cache for a specific campaign. + */ + static async invalidateCampaignCache(campaignId: string): Promise { + const cacheKey = `${CACHE_PREFIX_STATS}${campaignId}`; + try { + await redis.del(cacheKey); + logger.info(`Cache invalidated for campaign ${campaignId}`); + } catch (err) { + logger.warn(`Cache invalidation failed for ${cacheKey}`, err); + } + } + + // ============================================ + // TRENDING CAMPAIGNS + // ============================================ + + /** + * Get trending campaigns. Tries cache first (from Redis), falls back to DB. + */ + static async getTrendingCampaigns(filters: TrendingCampaignFilters = {}): Promise { + const { period = 'last24h', sortBy = 'trendScore', limit = 10 } = filters; + + // Try cached top-level data first + try { + const cachedData = await redis.get(`${CACHE_PREFIX_TRENDING_DATA}:${period}`); + if (cachedData) { + const trendings = JSON.parse(cachedData) as TrendingCampaign[]; + const sorted = this.sortTrendingCampaigns(trendings, sortBy); + return sorted.slice(0, limit); + } + } catch (err) { + logger.warn('Redis trending cache read failed', err); + } + + // Fallback to DB + return this.queryTrendingCampaignsFromDb(period, sortBy, limit); + } + + /** + * Query trending campaigns from the CampaignTrending table joined with Campaign. + */ + static async queryTrendingCampaignsFromDb( + period: string, + sortBy: string, + limit: number, + ): Promise { + const trendingRows = await prisma.campaignTrending.findMany({ + where: { period }, + orderBy: { [sortBy === 'trendScore' ? 'trendScore' : sortBy]: 'desc' }, + take: limit, + }); + + if (trendingRows.length === 0) { + return []; + } + + const campaignIds = trendingRows.map((t) => t.campaignId); + const campaigns = await prisma.campaign.findMany({ + where: { id: { in: campaignIds } }, + select: { + id: true, + title: true, + imageUrl: true, + status: true, + currentAmount: true, + targetAmount: true, + organization: { + select: { id: true, name: true, logo: true }, + }, + }, + }); + + const campaignMap = new Map(campaigns.map((c) => [c.id, c])); + + return trendingRows + .filter((t) => campaignMap.has(t.campaignId)) + .map((t) => { + const campaign = campaignMap.get(t.campaignId)!; + return { + campaignId: t.campaignId, + title: campaign.title, + imageUrl: campaign.imageUrl, + status: campaign.status, + currentAmount: Number(campaign.currentAmount), + targetAmount: Number(campaign.targetAmount), + trendScore: Number(t.trendScore), + donationVelocity: Number(t.donationVelocity), + donorGrowth: t.donorGrowth, + distributionImpact: Number(t.distributionImpact), + period: t.period, + rank: t.rank, + organization: campaign.organization, + }; + }); + } + + /** + * Sort trending campaigns by the specified field. + */ + private static sortTrendingCampaigns( + campaigns: TrendingCampaign[], + sortBy: string, + ): TrendingCampaign[] { + return [...campaigns].sort((a, b) => { + if (sortBy === 'donationVelocity') return b.donationVelocity - a.donationVelocity; + if (sortBy === 'distributionImpact') return b.distributionImpact - a.distributionImpact; + return b.trendScore - a.trendScore; // default: trendScore + }); + } + + /** + * Refresh the trending campaigns table and Redis cache. + * Called by the analytics worker on schedule. + */ + static async refreshTrendingCampaigns(): Promise { + try { + const periods: Array<'last24h' | 'last7d' | 'last30d'> = ['last24h', 'last7d', 'last30d']; + + for (const period of periods) { + const windowStart = this.getPeriodStart(period); + + // Query raw donation/distribution data for trending calculation + const trendingData = await prisma.$queryRaw>` + WITH campaign_metrics AS ( + SELECT + d."campaignId", + COUNT(DISTINCT d.id) AS "donationCount", + COALESCE(SUM(d.amount), 0) AS "donationVolume", + COUNT(DISTINCT d."userId") AS "uniqueDonors", + COALESCE(SUM(CASE WHEN dist.status = 'COMPLETED' THEN dist.amount ELSE 0 END), 0) AS "distributionVolume", + COUNT(DISTINCT CASE WHEN dist.status = 'COMPLETED' THEN dist.id END) AS "distributionCount" + FROM "Campaign" c + LEFT JOIN "Donation" d ON d."campaignId" = c.id + AND d.status = 'CONFIRMED' + AND d."createdAt" >= ${windowStart}::timestamp + LEFT JOIN "Distribution" dist ON dist."campaignId" = c.id + AND dist."createdAt" >= ${windowStart}::timestamp + WHERE c.status IN ('ACTIVE', 'COMPLETED') + GROUP BY d."campaignId" + ) + SELECT * FROM campaign_metrics + WHERE "donationCount" > 0 OR "distributionCount" > 0 + `; + + // Calculate trend scores and upsert + const count = config.analytics.trendingCampaignsCount; + const enriched = trendingData.map((row) => { + const donationVelocity = period === 'last24h' + ? Number(row.donationVolume) * 24 + : period === 'last7d' + ? Number(row.donationVolume) / 7 + : Number(row.donationVolume) / 30; + const distributionImpact = Number(row.distributionVolume); + const trendScore = donationVelocity * 0.4 + distributionImpact * 0.3 + row.uniqueDonors * 0.3; + + return { + campaignId: row.campaignId, + donationVelocity, + donorGrowth: row.uniqueDonors, + distributionImpact, + trendScore, + }; + }); + + // Sort by trendScore descending, assign ranks + enriched.sort((a, b) => b.trendScore - a.trendScore); + const topN = enriched.slice(0, count); + + // Upsert into CampaignTrending table (compound key: campaignId + period) + for (let i = 0; i < topN.length; i++) { + const entry = topN[i]; + await prisma.campaignTrending.upsert({ + where: { + campaignId_period: { campaignId: entry.campaignId, period }, + }, + create: { + campaignId: entry.campaignId, + trendScore: entry.trendScore, + donationVelocity: entry.donationVelocity, + donorGrowth: entry.donorGrowth, + distributionImpact: entry.distributionImpact, + period, + rank: i + 1, + }, + update: { + trendScore: entry.trendScore, + donationVelocity: entry.donationVelocity, + donorGrowth: entry.donorGrowth, + distributionImpact: entry.distributionImpact, + rank: i + 1, + refreshedAt: new Date(), + }, + }); + } + + // Remove stale entries not in top N + if (topN.length > 0) { + const keptIds = new Set(topN.map((t) => t.campaignId)); + await prisma.campaignTrending.deleteMany({ + where: { + period, + campaignId: { notIn: [...keptIds] }, + }, + }); + } + + // Cache the full trending list in Redis for fast reads + const fullTrendingList = await this.queryTrendingCampaignsFromDb(period, 'trendScore', count); + await redis.setex( + `${CACHE_PREFIX_TRENDING_DATA}:${period}`, + 900, // 15 min TTL + JSON.stringify(fullTrendingList), + ); + + logger.info(`Trending campaigns refreshed for period: ${period}, count: ${topN.length}`); + } + } catch (error) { + logger.error('Failed to refresh trending campaigns', error); + throw error; + } + } + + // ============================================ + // IMPACT METRICS + // ============================================ + + /** + * Get comprehensive impact metrics for a campaign, using cache when available. + */ + static async getCampaignImpactMetrics(campaignId: string): Promise { + const campaign = await prisma.campaign.findUnique({ + where: { id: campaignId }, + select: { id: true, title: true, targetAmount: true, currentAmount: true }, + }); + + if (!campaign) { + throw new Error('Campaign not found'); + } + + // Use cached stats when available (avoids raw table scans per acceptance criteria) + let cachedStats: Record | null = null; + try { + const cacheKey = `${CACHE_PREFIX_STATS}${campaignId}`; + const cached = await redis.hgetall(cacheKey); + if (cached && Object.keys(cached).length > 0) { + cachedStats = cached; + } + } catch (err) { + logger.warn(`Redis cache read failed for impact metrics on ${campaignId}`, err); + } + + let totalDonations: number; + let totalRaised: number; + let totalDistributions: number; + let totalDistributedAmount: number; + let uniqueDonors: number; + let beneficiariesReached: number; + + if (cachedStats) { + totalDonations = parseInt(cachedStats.totalDonations || '0', 10); + totalRaised = parseFloat(cachedStats.totalRaised || '0'); + totalDistributions = parseInt(cachedStats.totalDistributions || '0', 10); + totalDistributedAmount = parseFloat(cachedStats.totalDistributed || '0'); + uniqueDonors = parseInt(cachedStats.uniqueDonors || '0', 10); + beneficiariesReached = parseInt(cachedStats.beneficiariesReached || '0', 10); + } else { + // Fallback: aggregate from raw tables (cache miss) + const [donationAgg, distributionAgg, donorGrowth, beneficiaryCount] = await Promise.all([ + prisma.donation.aggregate({ + where: { campaignId, status: 'CONFIRMED' }, + _count: true, + _sum: { amount: true }, + }), + prisma.distribution.aggregate({ + where: { campaignId, status: 'COMPLETED' }, + _count: true, + _sum: { amount: true }, + }), + prisma.donation.groupBy({ + by: ['userId'], + where: { campaignId, status: 'CONFIRMED', userId: { not: null } }, + }), + prisma.beneficiaryAssignment.count({ where: { campaignId } }), + ]); + + totalDonations = donationAgg._count; + totalRaised = Number(donationAgg._sum.amount || 0); + totalDistributions = distributionAgg._count; + totalDistributedAmount = Number(distributionAgg._sum.amount || 0); + uniqueDonors = donorGrowth.length; + beneficiariesReached = beneficiaryCount; + + // Build and cache + const stats = await this.buildCampaignStats(campaignId); + await this.setCachedCampaignStats(campaignId, stats); + } + + const targetAmount = Number(campaign.targetAmount) || 1; + const currentAmount = Number(campaign.currentAmount) || 0; + const conversionRate = totalDonations > 0 + ? (totalDistributions / (totalDonations + totalDistributions)) * 100 + : 0; + const impactScore = (beneficiariesReached * 0.4) + + (totalDistributedAmount / Math.max(targetAmount, 1)) * 0.3 + + (uniqueDonors / Math.max(totalDonations + 1, 1)) * 0.3; + + return { + campaignId: campaign.id, + title: campaign.title, + totalDonations, + totalRaised, + donorGrowth: uniqueDonors, + totalDistributions, + totalDistributedAmount, + beneficiariesReached, + conversionRate: Math.round(conversionRate * 100) / 100, + avgDonationAmount: totalDonations > 0 ? totalRaised / totalDonations : 0, + progressPercentage: (currentAmount / targetAmount) * 100, + impactScore: Math.round(impactScore * 100) / 100, + }; + } + + // ============================================ + // HISTORICAL STATS + // ============================================ + + /** + * Get historical statistics for a campaign from rollup tables. + */ + static async getCampaignHistoricalStats( + campaignId: string, + granularity: 'hourly' | 'monthly' = 'hourly', + range?: { startDate: Date; endDate: Date }, + ): Promise { + if (granularity === 'hourly') { + const where: any = { campaignId }; + if (range) { + where.hour = { gte: range.startDate, lte: range.endDate }; + } + + const rows = await prisma.campaignHourlyStat.findMany({ + where, + orderBy: { hour: 'asc' }, + }); + + return { + campaignId, + granularity: 'hourly', + data: rows.map((r) => ({ + timestamp: r.hour, + donationCount: r.donationCount, + donationVolume: Number(r.donationVolume), + uniqueDonors: r.uniqueDonors, + distributionCount: r.distributionCount, + distributionVolume: Number(r.distributionVolume), + itemsDistributed: r.itemsDistributed, + activeDonors: r.activeDonors, + })), + }; + } + + // Monthly + const where: any = { campaignId }; + if (range) { + where.month = { gte: range.startDate, lte: range.endDate }; + } + + const rows = await prisma.campaignMonthlyStat.findMany({ + where, + orderBy: { month: 'asc' }, + }); + + return { + campaignId, + granularity: 'monthly', + data: rows.map((r) => ({ + timestamp: r.month, + donationCount: r.donationCount, + donationVolume: Number(r.donationVolume), + uniqueDonors: r.uniqueDonors, + distributionCount: r.distributionCount, + distributionVolume: Number(r.distributionVolume), + itemsDistributed: r.itemsDistributed, + donorGrowth: r.donorGrowth, + distributionReach: r.distributionReach, + campaignActivity: r.campaignActivity, + activeDonors: r.activeDonors, + })), + }; + } + + // ============================================ + // AGGREGATED CAMPAIGN ANALYTICS (ADMIN) + // ============================================ + + /** + * Query aggregated campaign metrics for admin dashboards. + * Uses rollup tables instead of raw scans. + */ + static async getAggregatedCampaignAnalytics( + filters: CampaignAnalyticsFilters, + pagination: PaginationParams, + ): Promise> { + const { page = 1, limit = 10, sortBy = 'createdAt', sortOrder = 'desc' } = pagination; + const skip = (page - 1) * limit; + + const campaigns = await prisma.campaign.findMany({ + where: { + ...(filters.status && { status: filters.status as any }), + ...(filters.startDate && { startDate: { gte: filters.startDate } }), + ...(filters.endDate && { endDate: { lte: filters.endDate } }), + }, + skip, + take: limit, + orderBy: { [sortBy]: sortOrder }, + select: { + id: true, + title: true, + status: true, + targetAmount: true, + currentAmount: true, + startDate: true, + endDate: true, + organization: { select: { id: true, name: true } }, + _count: { select: { donations: true, beneficiaries: true, distributions: true } }, + }, + }); + + const total = await prisma.campaign.count({ + where: { + ...(filters.status && { status: filters.status as any }), + ...(filters.startDate && { startDate: { gte: filters.startDate } }), + ...(filters.endDate && { endDate: { lte: filters.endDate } }), + }, + }); + + // Enrich with rollup summary data + const campaignIds = campaigns.map((c) => c.id); + const rollupSummaries = await prisma.campaignHourlyStat.findMany({ + where: { campaignId: { in: campaignIds } }, + orderBy: { hour: 'desc' }, + }); + + const rollupMap = new Map(); + for (const r of rollupSummaries) { + if (!rollupMap.has(r.campaignId)) { + rollupMap.set(r.campaignId, []); + } + rollupMap.get(r.campaignId)!.push(r); + } + + return { + data: campaigns.map((campaign) => { + const rollups = rollupMap.get(campaign.id) || []; + const recentRollup = rollups[0]; + return { + ...campaign, + targetAmount: Number(campaign.targetAmount), + currentAmount: Number(campaign.currentAmount), + stats: { + donationCount: campaign._count.donations, + beneficiaryCount: campaign._count.beneficiaries, + distributionCount: campaign._count.distributions, + lastHourActivity: recentRollup + ? { + donations: recentRollup.donationCount, + distributions: recentRollup.distributionCount, + hour: recentRollup.hour, + } + : null, + }, + }; + }), + pagination: { + page, + limit, + total, + totalPages: Math.ceil(total / limit), + }, + }; + } + + // ============================================ + // ROLLUP LOGIC (called by worker) + // ============================================ + + /** + * Get the start of a time window based on period string. + */ + private static getPeriodStart(period: 'last24h' | 'last7d' | 'last30d'): Date { + const now = new Date(); + switch (period) { + case 'last24h': + return new Date(now.getTime() - 24 * 60 * 60 * 1000); + case 'last7d': + return new Date(now.getTime() - 7 * 24 * 60 * 60 * 1000); + case 'last30d': + return new Date(now.getTime() - 30 * 24 * 60 * 60 * 1000); + } + } + + /** + * Get the last processed hour from the rollup tracker. + */ + private static async getLastProcessedHour(type: string): Promise { + const tracker = await prisma.rollupTracker.findUnique({ + where: { type }, + }); + return tracker?.lastHour || null; + } + + /** + * Update the last processed hour in the rollup tracker. + */ + private static async updateLastProcessedHour(type: string, hour: Date): Promise { + await prisma.rollupTracker.upsert({ + where: { type }, + create: { + type, + lastProcessedTimestamp: new Date(), + lastHour: hour, + }, + update: { + lastProcessedTimestamp: new Date(), + lastHour: hour, + }, + }); + } + + /** + * Get the last processed month from the rollup tracker. + */ + private static async getLastProcessedMonth(type: string): Promise { + const tracker = await prisma.rollupTracker.findUnique({ + where: { type }, + }); + return tracker?.lastMonth || null; + } + + /** + * Update the last processed month in the rollup tracker. + */ + private static async updateLastProcessedMonth(type: string, month: Date): Promise { + await prisma.rollupTracker.upsert({ + where: { type }, + create: { + type, + lastProcessedTimestamp: new Date(), + lastMonth: month, + }, + update: { + lastProcessedTimestamp: new Date(), + lastMonth: month, + }, + }); + } + + /** + * Run the hourly rollup for a given hour window. + * Idempotent — uses upsert to safely retry. + */ + static async runHourlyRollup(hourStart?: Date): Promise<{ processed: number; hourOf: Date }> { + const targetHour = hourStart || this.floorToHour(new Date()); + const hourEnd = new Date(targetHour.getTime() + 60 * 60 * 1000); + const trackerKey = 'campaign_hourly_stats'; + + // Get campaigns with activity in this hour + const activeCampaigns = await prisma.donation.findMany({ + where: { + status: 'CONFIRMED', + createdAt: { gte: targetHour, lt: hourEnd }, + }, + select: { campaignId: true }, + distinct: ['campaignId'], + }); + + const distributionCampaigns = await prisma.distribution.findMany({ + where: { + status: 'COMPLETED', + distributedAt: { gte: targetHour, lt: hourEnd }, + }, + select: { campaignId: true }, + distinct: ['campaignId'], + }); + + const allCampaignIds = [ + ...new Set([ + ...activeCampaigns.map((d) => d.campaignId), + ...distributionCampaigns.map((d) => d.campaignId), + ]), + ]; + + let processed = 0; + + for (const campaignId of allCampaignIds) { + try { + const [donationAgg, distributionAgg, uniqueDonorsResult, newBeneficiaries] = await Promise.all([ + prisma.donation.aggregate({ + where: { + campaignId, + status: 'CONFIRMED', + createdAt: { gte: targetHour, lt: hourEnd }, + }, + _count: true, + _sum: { amount: true }, + }), + prisma.distribution.aggregate({ + where: { + campaignId, + status: 'COMPLETED', + distributedAt: { gte: targetHour, lt: hourEnd }, + }, + _count: true, + _sum: { amount: true }, + }), + prisma.donation.groupBy({ + by: ['userId'], + where: { + campaignId, + status: 'CONFIRMED', + createdAt: { gte: targetHour, lt: hourEnd }, + userId: { not: null }, + }, + }), + prisma.beneficiaryAssignment.count({ + where: { + campaignId, + assignedAt: { gte: targetHour, lt: hourEnd }, + }, + }), + ]); + + await prisma.campaignHourlyStat.upsert({ + where: { + campaignId_hour: { campaignId, hour: targetHour }, + }, + create: { + campaignId, + hour: targetHour, + donationCount: donationAgg._count, + donationVolume: donationAgg._sum.amount || 0, + uniqueDonors: uniqueDonorsResult.length, + distributionCount: distributionAgg._count, + distributionVolume: distributionAgg._sum.amount || 0, + itemsDistributed: distributionAgg._count, + newBeneficiaries, + activeDonors: uniqueDonorsResult.length, + }, + update: { + donationCount: donationAgg._count, + donationVolume: donationAgg._sum.amount || 0, + uniqueDonors: uniqueDonorsResult.length, + distributionCount: distributionAgg._count, + distributionVolume: distributionAgg._sum.amount || 0, + itemsDistributed: distributionAgg._count, + newBeneficiaries, + activeDonors: uniqueDonorsResult.length, + }, + }); + + // Also rebuild and refresh the Redis cache for this campaign + const stats = await this.buildCampaignStats(campaignId); + await this.setCachedCampaignStats(campaignId, stats); + + processed++; + } catch (err) { + logger.error(`Hourly rollup failed for campaign ${campaignId}`, err); + } + } + + // Update tracker + await this.updateLastProcessedHour(trackerKey, targetHour); + + logger.info(`Hourly rollup completed: ${processed} campaigns processed for hour ${targetHour.toISOString()}`); + return { processed, hourOf: targetHour }; + } + + /** + * Run the monthly rollup for a given month. + * Idempotent — uses upsert to safely retry. + */ + static async runMonthlyRollup(monthStart?: Date): Promise<{ processed: number; monthOf: Date }> { + const targetMonth = monthStart || this.floorToMonth(new Date()); + const monthEnd = new Date(targetMonth.getFullYear(), targetMonth.getMonth() + 1, 1); + const trackerKey = 'campaign_monthly_stats'; + + // Get all active campaigns + const campaigns = await prisma.campaign.findMany({ + where: { + OR: [ + { createdAt: { lt: monthEnd } }, + { status: { in: ['ACTIVE', 'COMPLETED', 'PAUSED'] } }, + ], + }, + select: { id: true }, + }); + + let processed = 0; + + for (const campaign of campaigns) { + try { + const [donationAgg, distributionAgg, uniqueDonorsResult] = await Promise.all([ + prisma.donation.aggregate({ + where: { + campaignId: campaign.id, + status: 'CONFIRMED', + createdAt: { gte: targetMonth, lt: monthEnd }, + }, + _count: true, + _sum: { amount: true }, + }), + prisma.distribution.aggregate({ + where: { + campaignId: campaign.id, + status: 'COMPLETED', + distributedAt: { gte: targetMonth, lt: monthEnd }, + }, + _count: true, + _sum: { amount: true }, + }), + prisma.donation.groupBy({ + by: ['userId'], + where: { + campaignId: campaign.id, + status: 'CONFIRMED', + createdAt: { gte: targetMonth, lt: monthEnd }, + userId: { not: null }, + }, + }), + ]); + + // Donor growth: unique donors this month vs previous month + const prevMonthStart = new Date(targetMonth.getFullYear(), targetMonth.getMonth() - 1, 1); + const prevMonthDonors = await prisma.donation.groupBy({ + by: ['userId'], + where: { + campaignId: campaign.id, + status: 'CONFIRMED', + createdAt: { gte: prevMonthStart, lt: targetMonth }, + userId: { not: null }, + }, + }); + + const donorGrowth = uniqueDonorsResult.length - prevMonthDonors.length; + const distributionReach = distributionAgg._count; + const campaignActivity = donationAgg._count + distributionAgg._count; + + await prisma.campaignMonthlyStat.upsert({ + where: { + campaignId_month: { campaignId: campaign.id, month: targetMonth }, + }, + create: { + campaignId: campaign.id, + month: targetMonth, + donationCount: donationAgg._count, + donationVolume: donationAgg._sum.amount || 0, + uniqueDonors: uniqueDonorsResult.length, + distributionCount: distributionAgg._count, + distributionVolume: distributionAgg._sum.amount || 0, + itemsDistributed: distributionAgg._count, + donorGrowth, + distributionReach, + campaignActivity, + activeDonors: uniqueDonorsResult.length, + }, + update: { + donationCount: donationAgg._count, + donationVolume: donationAgg._sum.amount || 0, + uniqueDonors: uniqueDonorsResult.length, + distributionCount: distributionAgg._count, + distributionVolume: distributionAgg._sum.amount || 0, + itemsDistributed: distributionAgg._count, + donorGrowth, + distributionReach, + campaignActivity, + activeDonors: uniqueDonorsResult.length, + }, + }); + + processed++; + } catch (err) { + logger.error(`Monthly rollup failed for campaign ${campaign.id}`, err); + } + } + + // Update tracker + await this.updateLastProcessedMonth(trackerKey, targetMonth); + + logger.info(`Monthly rollup completed: ${processed} campaigns processed for month ${targetMonth.toISOString()}`); + return { processed, monthOf: targetMonth }; + } + + /** + * Floor a date to the start of the current hour. + */ + private static floorToHour(date: Date): Date { + return new Date(date.getFullYear(), date.getMonth(), date.getDate(), date.getHours(), 0, 0, 0); + } + + /** + * Floor a date to the start of the current month. + */ + private static floorToMonth(date: Date): Date { + return new Date(date.getFullYear(), date.getMonth(), 1); + } + + /** + * Rebuild all campaign caches — full reconciliation. + * Called by the reconciliation job to heal any drift between Redis and DB. + */ + static async rebuildAllCampaignCaches(): Promise { + const campaigns = await prisma.campaign.findMany({ + where: { status: { in: ['ACTIVE', 'COMPLETED', 'PAUSED'] } }, + select: { id: true }, + }); + + let rebuilt = 0; + for (const campaign of campaigns) { + try { + const stats = await this.buildCampaignStats(campaign.id); + await this.setCachedCampaignStats(campaign.id, stats); + rebuilt++; + } catch (err) { + logger.error(`Cache rebuild failed for campaign ${campaign.id}`, err); + } + } + + logger.info(`Cache reconciliation completed: ${rebuilt} campaigns rebuilt`); + return rebuilt; + } } diff --git a/src/services/distribution.service.ts b/src/services/distribution.service.ts index 45f04b2..473aaa0 100644 --- a/src/services/distribution.service.ts +++ b/src/services/distribution.service.ts @@ -3,6 +3,7 @@ import { DistributionInput, PaginatedResponse } from '../types'; import { DistributionStatus, Role } from '@prisma/client'; import { AppError } from '../middleware/error'; import logger from '../config/logger'; +import { AnalyticsService } from './analytics.service'; export class DistributionService { static async createDistribution(data: DistributionInput, userId: string, userRole: Role): Promise { @@ -79,6 +80,12 @@ export class DistributionService { logger.info(`Distribution confirmed: ${id} with tx ${txHash}`); + // Fire-and-forget: update campaign analytics cache + const amount = Number(distribution.amount); + AnalyticsService.incrementDistributionStats(distribution.campaignId, amount).catch((err) => + logger.error('Failed to update campaign distribution stats cache', err) + ); + return updated; } @@ -159,6 +166,11 @@ export class DistributionService { logger.info(`Distribution status updated: ${id} to ${status} by user ${userId}`); + // Invalidate cache when distribution status changes + AnalyticsService.invalidateCampaignCache(distribution.campaignId).catch((err) => + logger.error('Failed to invalidate campaign cache on status update', err) + ); + return updated; } diff --git a/src/services/donation.service.ts b/src/services/donation.service.ts index 05fed55..af0cadc 100644 --- a/src/services/donation.service.ts +++ b/src/services/donation.service.ts @@ -3,6 +3,7 @@ import { DonationInput, DonationFilters, PaginatedResponse } from '../types'; import { DonationStatus, Role } from '@prisma/client'; import { AppError } from '../middleware/error'; import logger from '../config/logger'; +import { AnalyticsService } from './analytics.service'; export class DonationService { static async createDonation(data: DonationInput, userId?: string): Promise { @@ -70,6 +71,12 @@ export class DonationService { logger.info(`Donation confirmed: ${id} with tx ${txHash}`); + // Fire-and-forget: update campaign analytics cache + const amount = Number(donation.amount); + AnalyticsService.incrementDonationStats(donation.campaignId, amount).catch((err) => + logger.error('Failed to update campaign donation stats cache', err) + ); + return updated; } @@ -210,6 +217,12 @@ export class DonationService { logger.info(`Donation refunded: ${id} by user ${userId}`); + // Update cache: decrement donation count/amount on refund + const amount = Number(donation.amount); + AnalyticsService.invalidateCampaignCache(donation.campaignId).catch((err) => + logger.error('Failed to invalidate campaign cache on refund', err) + ); + return updated; } } diff --git a/src/types/index.ts b/src/types/index.ts index f69bfa0..d0be283 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -134,3 +134,70 @@ export interface DistributionInput { method: DistributionMethod; description?: string; } + +export interface TrendingCampaignFilters { + period?: 'last24h' | 'last7d' | 'last30d'; + sortBy?: 'trendScore' | 'donationVelocity' | 'distributionImpact'; + limit?: number; +} + +export interface TrendingCampaign { + campaignId: string; + title: string; + imageUrl: string | null; + status: string; + currentAmount: number; + targetAmount: number; + trendScore: number; + donationVelocity: number; + donorGrowth: number; + distributionImpact: number; + period: string; + rank: number; + organization: { + id: string; + name: string; + logo: string | null; + }; +} + +export interface ImpactMetrics { + campaignId: string; + title: string; + totalDonations: number; + totalRaised: number; + donorGrowth: number; + totalDistributions: number; + totalDistributedAmount: number; + beneficiariesReached: number; + conversionRate: number; + avgDonationAmount: number; + progressPercentage: number; + impactScore: number; +} + +export interface HistoricalStats { + campaignId: string; + granularity: 'hourly' | 'monthly'; + data: Array<{ + timestamp: Date; + donationCount: number; + donationVolume: number; + uniqueDonors: number; + distributionCount: number; + distributionVolume: number; + itemsDistributed: number; + donorGrowth?: number; + distributionReach?: number; + campaignActivity?: number; + activeDonors: number; + }>; +} + +export interface CampaignAnalyticsFilters { + status?: string; + startDate?: Date; + endDate?: Date; + sortBy?: string; + sortOrder?: 'asc' | 'desc'; +} diff --git a/src/workers/analytics.worker.ts b/src/workers/analytics.worker.ts new file mode 100644 index 0000000..1380ad4 --- /dev/null +++ b/src/workers/analytics.worker.ts @@ -0,0 +1,182 @@ +import { Worker, Queue, Job } from 'bullmq'; +import { config } from '../config'; +import { AnalyticsService } from '../services/analytics.service'; +import logger from '../config/logger'; + +const QUEUE_NAME = 'analytics-queue'; + +const connection = { + host: config.bullmq.redisHost, + port: config.bullmq.redisPort, + password: config.bullmq.redisPassword, +}; + +// Producer — used to enqueue real-time cache updates and scheduled jobs. +export const analyticsQueue = new Queue(QUEUE_NAME, { connection }); + +/** + * Enqueue a real-time analytics cache update after a donation event. + */ +export const enqueueRealtimeDonationUpdate = async ( + campaignId: string, + amount: number, +): Promise => { + await analyticsQueue.add( + 'REALTIME_DONATION', + { type: 'REALTIME_DONATION', data: { campaignId, amount } }, + { + jobId: `realtime-donation:${campaignId}:${Date.now()}`, + removeOnComplete: true, + removeOnFail: 100, + attempts: 3, + backoff: { type: 'exponential', delay: 1000 }, + }, + ); +}; + +/** + * Enqueue a real-time analytics cache update after a distribution event. + */ +export const enqueueRealtimeDistributionUpdate = async ( + campaignId: string, + amount: number, +): Promise => { + await analyticsQueue.add( + 'REALTIME_DISTRIBUTION', + { type: 'REALTIME_DISTRIBUTION', data: { campaignId, amount } }, + { + jobId: `realtime-distribution:${campaignId}:${Date.now()}`, + removeOnComplete: true, + removeOnFail: 100, + attempts: 3, + backoff: { type: 'exponential', delay: 1000 }, + }, + ); +}; + +/** + * Register the scheduled analytics jobs: + * - Hourly rollup (every hour at :05) + * - Monthly rollup (1st of month at 02:00) + * - Trending campaign refresh (every 15 minutes) + * - Cache reconciliation (every 6 hours) + */ +export const scheduleAnalyticsJobs = async (): Promise => { + if (!config.analytics.analyticsWorkerEnabled) { + logger.info('Analytics worker disabled; skipping analytics schedule'); + return; + } + + // Hourly rollup + await analyticsQueue.add( + 'HOURLY_ROLLUP', + { type: 'HOURLY_ROLLUP', data: {} }, + { + repeat: { pattern: config.analytics.hourlyRollupCron }, + jobId: 'analytics-hourly-rollup', + removeOnComplete: true, + removeOnFail: 100, + }, + ); + + // Monthly rollup + await analyticsQueue.add( + 'MONTHLY_ROLLUP', + { type: 'MONTHLY_ROLLUP', data: {} }, + { + repeat: { pattern: config.analytics.monthlyRollupCron }, + jobId: 'analytics-monthly-rollup', + removeOnComplete: true, + removeOnFail: 100, + }, + ); + + // Trending campaign refresh + await analyticsQueue.add( + 'TRENDING_REFRESH', + { type: 'TRENDING_REFRESH', data: {} }, + { + repeat: { pattern: config.analytics.trendingRefreshCron }, + jobId: 'analytics-trending-refresh', + removeOnComplete: true, + removeOnFail: 100, + }, + ); + + // Cache reconciliation (every 6 hours) + await analyticsQueue.add( + 'CACHE_RECONCILE', + { type: 'CACHE_RECONCILE', data: {} }, + { + repeat: { pattern: '0 */6 * * *' }, + jobId: 'analytics-cache-reconcile', + removeOnComplete: true, + removeOnFail: 100, + }, + ); + + logger.info('Scheduled analytics jobs: hourly, monthly, trending, cache-reconcile'); +}; + +const analyticsWorker = new Worker( + QUEUE_NAME, + async (job: Job) => { + const { type, data } = job.data; + + logger.info(`Processing analytics job: ${job.id}, type: ${type}`); + + switch (type) { + case 'HOURLY_ROLLUP': { + const result = await AnalyticsService.runHourlyRollup(); + logger.info(`Hourly rollup completed: ${result.processed} campaigns for ${result.hourOf}`); + return result; + } + + case 'MONTHLY_ROLLUP': { + const result = await AnalyticsService.runMonthlyRollup(); + logger.info(`Monthly rollup completed: ${result.processed} campaigns for ${result.monthOf}`); + return result; + } + + case 'TRENDING_REFRESH': { + await AnalyticsService.refreshTrendingCampaigns(); + logger.info('Trending campaigns refreshed'); + return { success: true }; + } + + case 'CACHE_RECONCILE': { + const rebuilt = await AnalyticsService.rebuildAllCampaignCaches(); + logger.info(`Cache reconciliation completed: ${rebuilt} campaigns rebuilt`); + return { rebuilt }; + } + + case 'REALTIME_DONATION': { + await AnalyticsService.incrementDonationStats(data.campaignId, data.amount); + // Also update the trending ZSET for real-time trending + return { campaignId: data.campaignId }; + } + + case 'REALTIME_DISTRIBUTION': { + await AnalyticsService.incrementDistributionStats(data.campaignId, data.amount); + return { campaignId: data.campaignId }; + } + + default: + throw new Error(`Unknown analytics job type: ${type}`); + } + }, + { + connection, + concurrency: 3, + }, +); + +analyticsWorker.on('completed', (job) => { + logger.info(`Analytics job completed: ${job.id}`); +}); + +analyticsWorker.on('failed', (job, err) => { + logger.error(`Analytics job failed: ${job?.id}`, err); +}); + +export default analyticsWorker; From 31185b118d30488caa3a0e7508380c05dddeb784 Mon Sep 17 00:00:00 2001 From: Daniel Omoloba Date: Mon, 22 Jun 2026 05:59:46 +0000 Subject: [PATCH 2/2] fix: resolve CI issues - missing imports, unused vars, and JSON null type errors - Add missing AnalyticsService imports in donation.service.ts and distribution.service.ts - Remove unused CACHE_PREFIX_TRENDING_ZSET constant in analytics.service.ts - Remove unused 'amount' variable in donation.service.ts refund handler - Fix Prisma JSON null type errors in recovery.service.ts (writeAuditLog helper and create functions) --- src/services/analytics.service.ts | 173 +++++++++++++++------------ src/services/distribution.service.ts | 32 ++++- src/services/donation.service.ts | 9 +- src/services/recovery.service.ts | 19 +-- 4 files changed, 139 insertions(+), 94 deletions(-) diff --git a/src/services/analytics.service.ts b/src/services/analytics.service.ts index d1d1e5f..dd172d5 100644 --- a/src/services/analytics.service.ts +++ b/src/services/analytics.service.ts @@ -15,7 +15,6 @@ import { // Redis cache key prefixes const CACHE_PREFIX_STATS = 'campaign:stats:'; const CACHE_PREFIX_TRENDING_DATA = 'campaigns:trending:data'; -const CACHE_PREFIX_TRENDING_ZSET = 'campaigns:trending'; export class AnalyticsService { static async getCampaignAnalytics(campaignId: string): Promise { @@ -57,13 +56,14 @@ export class AnalyticsService { // Calculate distribution statistics const totalDistributed = campaign.distributions - .filter(d => d.status === 'COMPLETED') + .filter((d) => d.status === 'COMPLETED') .reduce((sum, d) => sum + Number(d.amount), 0); // Calculate progress percentage - const progress = Number(campaign.targetAmount) > 0 - ? (Number(campaign.currentAmount) / Number(campaign.targetAmount)) * 100 - : 0; + const progress = + Number(campaign.targetAmount) > 0 + ? (Number(campaign.currentAmount) / Number(campaign.targetAmount)) * 100 + : 0; // Daily donation trend (last 30 days) const thirtyDaysAgo = new Date(); @@ -98,7 +98,7 @@ export class AnalyticsService { distributions: { total: campaign._count.distributions, totalDistributed, - completed: campaign.distributions.filter(d => d.status === 'COMPLETED').length, + completed: campaign.distributions.filter((d) => d.status === 'COMPLETED').length, }, beneficiaries: { total: campaign._count.beneficiaries, @@ -122,7 +122,7 @@ export class AnalyticsService { }); const totalDonated = donations.reduce((sum, d) => sum + Number(d.amount), 0); - const campaignsSupported = new Set(donations.map(d => d.campaignId)).size; + const campaignsSupported = new Set(donations.map((d) => d.campaignId)).size; // Monthly donation trend const monthlyDonations = await prisma.$queryRaw` @@ -167,9 +167,11 @@ export class AnalyticsService { }); const totalCampaigns = campaigns.length; - const activeCampaigns = campaigns.filter(c => c.status === 'ACTIVE').length; - const totalRaised = campaigns.reduce((sum, c) => - sum + c.donations.reduce((dSum, d) => dSum + Number(d.amount), 0), 0); + const activeCampaigns = campaigns.filter((c) => c.status === 'ACTIVE').length; + const totalRaised = campaigns.reduce( + (sum, c) => sum + c.donations.reduce((dSum, d) => dSum + Number(d.amount), 0), + 0 + ); const totalBeneficiaries = campaigns.reduce((sum, c) => sum + c._count.beneficiaries, 0); const totalDistributions = campaigns.reduce((sum, c) => sum + c._count.distributions, 0); @@ -177,7 +179,7 @@ export class AnalyticsService { campaigns: { total: totalCampaigns, active: activeCampaigns, - completed: campaigns.filter(c => c.status === 'COMPLETED').length, + completed: campaigns.filter((c) => c.status === 'COMPLETED').length, }, funds: { totalRaised, @@ -268,22 +270,22 @@ export class AnalyticsService { throw new Error('Campaign ID is required for campaign report'); } return this.getCampaignAnalytics(filters.campaignId); - + case 'donor': if (!filters.userId) { throw new Error('User ID is required for donor report'); } return this.getDonorAnalytics(filters.userId); - + case 'organization': if (!filters.organizationId) { throw new Error('Organization ID is required for organization report'); } return this.getOrganizationAnalytics(filters.organizationId); - + case 'platform': return this.getPlatformAnalytics(); - + default: throw new Error('Invalid report type'); } @@ -318,7 +320,7 @@ export class AnalyticsService { */ static async setCachedCampaignStats( campaignId: string, - stats: Record, + stats: Record ): Promise { const cacheKey = `${CACHE_PREFIX_STATS}${campaignId}`; try { @@ -440,7 +442,9 @@ export class AnalyticsService { /** * Get trending campaigns. Tries cache first (from Redis), falls back to DB. */ - static async getTrendingCampaigns(filters: TrendingCampaignFilters = {}): Promise { + static async getTrendingCampaigns( + filters: TrendingCampaignFilters = {} + ): Promise { const { period = 'last24h', sortBy = 'trendScore', limit = 10 } = filters; // Try cached top-level data first @@ -465,7 +469,7 @@ export class AnalyticsService { static async queryTrendingCampaignsFromDb( period: string, sortBy: string, - limit: number, + limit: number ): Promise { const trendingRows = await prisma.campaignTrending.findMany({ where: { period }, @@ -522,7 +526,7 @@ export class AnalyticsService { */ private static sortTrendingCampaigns( campaigns: TrendingCampaign[], - sortBy: string, + sortBy: string ): TrendingCampaign[] { return [...campaigns].sort((a, b) => { if (sortBy === 'donationVelocity') return b.donationVelocity - a.donationVelocity; @@ -543,14 +547,16 @@ export class AnalyticsService { const windowStart = this.getPeriodStart(period); // Query raw donation/distribution data for trending calculation - const trendingData = await prisma.$queryRaw>` + const trendingData = await prisma.$queryRaw< + Array<{ + campaignId: string; + donationCount: number; + donationVolume: number; + uniqueDonors: number; + distributionCount: number; + distributionVolume: number; + }> + >` WITH campaign_metrics AS ( SELECT d."campaignId", @@ -575,13 +581,15 @@ export class AnalyticsService { // Calculate trend scores and upsert const count = config.analytics.trendingCampaignsCount; const enriched = trendingData.map((row) => { - const donationVelocity = period === 'last24h' - ? Number(row.donationVolume) * 24 - : period === 'last7d' - ? Number(row.donationVolume) / 7 - : Number(row.donationVolume) / 30; + const donationVelocity = + period === 'last24h' + ? Number(row.donationVolume) * 24 + : period === 'last7d' + ? Number(row.donationVolume) / 7 + : Number(row.donationVolume) / 30; const distributionImpact = Number(row.distributionVolume); - const trendScore = donationVelocity * 0.4 + distributionImpact * 0.3 + row.uniqueDonors * 0.3; + const trendScore = + donationVelocity * 0.4 + distributionImpact * 0.3 + row.uniqueDonors * 0.3; return { campaignId: row.campaignId, @@ -635,11 +643,15 @@ export class AnalyticsService { } // Cache the full trending list in Redis for fast reads - const fullTrendingList = await this.queryTrendingCampaignsFromDb(period, 'trendScore', count); + const fullTrendingList = await this.queryTrendingCampaignsFromDb( + period, + 'trendScore', + count + ); await redis.setex( `${CACHE_PREFIX_TRENDING_DATA}:${period}`, 900, // 15 min TTL - JSON.stringify(fullTrendingList), + JSON.stringify(fullTrendingList) ); logger.info(`Trending campaigns refreshed for period: ${period}, count: ${topN.length}`); @@ -727,10 +739,10 @@ export class AnalyticsService { const targetAmount = Number(campaign.targetAmount) || 1; const currentAmount = Number(campaign.currentAmount) || 0; - const conversionRate = totalDonations > 0 - ? (totalDistributions / (totalDonations + totalDistributions)) * 100 - : 0; - const impactScore = (beneficiariesReached * 0.4) + + const conversionRate = + totalDonations > 0 ? (totalDistributions / (totalDonations + totalDistributions)) * 100 : 0; + const impactScore = + beneficiariesReached * 0.4 + (totalDistributedAmount / Math.max(targetAmount, 1)) * 0.3 + (uniqueDonors / Math.max(totalDonations + 1, 1)) * 0.3; @@ -760,7 +772,7 @@ export class AnalyticsService { static async getCampaignHistoricalStats( campaignId: string, granularity: 'hourly' | 'monthly' = 'hourly', - range?: { startDate: Date; endDate: Date }, + range?: { startDate: Date; endDate: Date } ): Promise { if (granularity === 'hourly') { const where: any = { campaignId }; @@ -829,7 +841,7 @@ export class AnalyticsService { */ static async getAggregatedCampaignAnalytics( filters: CampaignAnalyticsFilters, - pagination: PaginationParams, + pagination: PaginationParams ): Promise> { const { page = 1, limit = 10, sortBy = 'createdAt', sortOrder = 'desc' } = pagination; const skip = (page - 1) * limit; @@ -1024,41 +1036,42 @@ export class AnalyticsService { for (const campaignId of allCampaignIds) { try { - const [donationAgg, distributionAgg, uniqueDonorsResult, newBeneficiaries] = await Promise.all([ - prisma.donation.aggregate({ - where: { - campaignId, - status: 'CONFIRMED', - createdAt: { gte: targetHour, lt: hourEnd }, - }, - _count: true, - _sum: { amount: true }, - }), - prisma.distribution.aggregate({ - where: { - campaignId, - status: 'COMPLETED', - distributedAt: { gte: targetHour, lt: hourEnd }, - }, - _count: true, - _sum: { amount: true }, - }), - prisma.donation.groupBy({ - by: ['userId'], - where: { - campaignId, - status: 'CONFIRMED', - createdAt: { gte: targetHour, lt: hourEnd }, - userId: { not: null }, - }, - }), - prisma.beneficiaryAssignment.count({ - where: { - campaignId, - assignedAt: { gte: targetHour, lt: hourEnd }, - }, - }), - ]); + const [donationAgg, distributionAgg, uniqueDonorsResult, newBeneficiaries] = + await Promise.all([ + prisma.donation.aggregate({ + where: { + campaignId, + status: 'CONFIRMED', + createdAt: { gte: targetHour, lt: hourEnd }, + }, + _count: true, + _sum: { amount: true }, + }), + prisma.distribution.aggregate({ + where: { + campaignId, + status: 'COMPLETED', + distributedAt: { gte: targetHour, lt: hourEnd }, + }, + _count: true, + _sum: { amount: true }, + }), + prisma.donation.groupBy({ + by: ['userId'], + where: { + campaignId, + status: 'CONFIRMED', + createdAt: { gte: targetHour, lt: hourEnd }, + userId: { not: null }, + }, + }), + prisma.beneficiaryAssignment.count({ + where: { + campaignId, + assignedAt: { gte: targetHour, lt: hourEnd }, + }, + }), + ]); await prisma.campaignHourlyStat.upsert({ where: { @@ -1101,7 +1114,9 @@ export class AnalyticsService { // Update tracker await this.updateLastProcessedHour(trackerKey, targetHour); - logger.info(`Hourly rollup completed: ${processed} campaigns processed for hour ${targetHour.toISOString()}`); + logger.info( + `Hourly rollup completed: ${processed} campaigns processed for hour ${targetHour.toISOString()}` + ); return { processed, hourOf: targetHour }; } @@ -1216,7 +1231,9 @@ export class AnalyticsService { // Update tracker await this.updateLastProcessedMonth(trackerKey, targetMonth); - logger.info(`Monthly rollup completed: ${processed} campaigns processed for month ${targetMonth.toISOString()}`); + logger.info( + `Monthly rollup completed: ${processed} campaigns processed for month ${targetMonth.toISOString()}` + ); return { processed, monthOf: targetMonth }; } diff --git a/src/services/distribution.service.ts b/src/services/distribution.service.ts index eeb2f1d..fe17044 100644 --- a/src/services/distribution.service.ts +++ b/src/services/distribution.service.ts @@ -4,9 +4,14 @@ import { DistributionStatus, Role } from '@prisma/client'; import { AppError } from '../middleware/error'; import logger from '../config/logger'; import { dispatchWebhookEvent } from '../controllers/webhook.controller'; +import { AnalyticsService } from './analytics.service'; export class DistributionService { - static async createDistribution(data: DistributionInput, userId: string, userRole: Role): Promise { + static async createDistribution( + data: DistributionInput, + userId: string, + userRole: Role + ): Promise { const campaign = await prisma.campaign.findUnique({ where: { id: data.campaignId }, }); @@ -39,7 +44,10 @@ export class DistributionService { // Check permissions if (campaign.userId !== userId && userRole !== Role.ADMIN) { - throw new AppError('You do not have permission to create distributions for this campaign', 403); + throw new AppError( + 'You do not have permission to create distributions for this campaign', + 403 + ); } const distribution = await prisma.distribution.create({ @@ -92,7 +100,11 @@ export class DistributionService { return updated; } - static async getDistributions(campaignId?: string, beneficiaryId?: string, pagination?: any): Promise> { + static async getDistributions( + campaignId?: string, + beneficiaryId?: string, + pagination?: any + ): Promise> { const { page = 1, limit = 10, sortBy = 'createdAt', sortOrder = 'desc' } = pagination || {}; const skip = (page - 1) * limit; @@ -143,7 +155,12 @@ export class DistributionService { }; } - static async updateDistributionStatus(id: string, status: DistributionStatus, userId: string, userRole: Role): Promise { + static async updateDistributionStatus( + id: string, + status: DistributionStatus, + userId: string, + userRole: Role + ): Promise { const distribution = await prisma.distribution.findUnique({ where: { id }, include: { campaign: true }, @@ -177,7 +194,12 @@ export class DistributionService { return updated; } - static async addProofDocument(id: string, proofDocumentUrl: string, userId: string, userRole: Role): Promise { + static async addProofDocument( + id: string, + proofDocumentUrl: string, + userId: string, + userRole: Role + ): Promise { const distribution = await prisma.distribution.findUnique({ where: { id }, include: { campaign: true }, diff --git a/src/services/donation.service.ts b/src/services/donation.service.ts index 99da8e0..320b4ff 100644 --- a/src/services/donation.service.ts +++ b/src/services/donation.service.ts @@ -4,6 +4,7 @@ import { DonationStatus, Role } from '@prisma/client'; import { AppError } from '../middleware/error'; import logger from '../config/logger'; import { dispatchWebhookEvent } from '../controllers/webhook.controller'; +import { AnalyticsService } from './analytics.service'; export class DonationService { static async createDonation(data: DonationInput, userId?: string): Promise { @@ -82,7 +83,10 @@ export class DonationService { return updated; } - static async getDonations(filters: DonationFilters = {}, pagination: any): Promise> { + static async getDonations( + filters: DonationFilters = {}, + pagination: any + ): Promise> { filters = filters ?? {}; const { page = 1, limit = 10, sortBy = 'createdAt', sortOrder = 'desc' } = pagination; @@ -230,8 +234,7 @@ export class DonationService { logger.info(`Donation refunded: ${id} by user ${userId}`); - // Update cache: decrement donation count/amount on refund - const amount = Number(donation.amount); + // Update cache: invalidate on refund AnalyticsService.invalidateCampaignCache(donation.campaignId).catch((err) => logger.error('Failed to invalidate campaign cache on refund', err) ); diff --git a/src/services/recovery.service.ts b/src/services/recovery.service.ts index 2aea3a6..e0149df 100644 --- a/src/services/recovery.service.ts +++ b/src/services/recovery.service.ts @@ -32,7 +32,7 @@ async function writeAuditLog( action, entityType, entityId, - metadata: metadata ?? null, + ...(metadata !== undefined ? { metadata } : {}), }, }); } @@ -63,7 +63,7 @@ export async function createFailedRefundCase( type: RecoveryCaseType.FAILED_REFUND, donationId, failureReason, - failureMetadata: failureMetadata ?? null, + ...(failureMetadata !== undefined ? { failureMetadata } : {}), status: RecoveryStatus.PENDING, maxRetries: MAX_RETRIES, nextRetryAt: nextRetryAt(0), @@ -114,7 +114,7 @@ export async function createFailedDistributionCase( type: RecoveryCaseType.FAILED_DISTRIBUTION, distributionId, failureReason, - failureMetadata: failureMetadata ?? null, + ...(failureMetadata !== undefined ? { failureMetadata } : {}), status: RecoveryStatus.PENDING, maxRetries: MAX_RETRIES, nextRetryAt: nextRetryAt(0), @@ -259,7 +259,7 @@ export async function updateRefundDestination( const updated = await prisma.recoveryCase.update({ where: { id: recoveryCaseId }, data: { - failureMetadata: { ...(rc.failureMetadata as object), updatedBankAccount: newAccount }, + failureMetadata: { ...((rc.failureMetadata ?? {}) as any), updatedBankAccount: newAccount }, status: RecoveryStatus.PENDING, nextRetryAt: nextRetryAt(0), }, @@ -316,12 +316,12 @@ export async function settleCancelledCampaign( if (!rc) throw new AppError('Recovery case not found', 404); if (rc.type !== RecoveryCaseType.CANCELLED_CAMPAIGN_FUNDS) throw new AppError('Not a CANCELLED_CAMPAIGN_FUNDS case', 400); - if (rc.status === RecoveryStatus.RECOVERED) - throw new AppError('Case is already settled', 400); + if (rc.status === RecoveryStatus.RECOVERED) throw new AppError('Case is already settled', 400); if (!rc.campaignId) throw new AppError('No campaign linked to this case', 400); if (option === SettlementOption.TRANSFER_TO_CAMPAIGN) { - if (!targetCampaignId) throw new AppError('targetCampaignId required for TRANSFER_TO_CAMPAIGN', 400); + if (!targetCampaignId) + throw new AppError('targetCampaignId required for TRANSFER_TO_CAMPAIGN', 400); const target = await prisma.campaign.findUnique({ where: { id: targetCampaignId } }); if (!target || target.status !== CampaignStatus.ACTIVE) throw new AppError('Target campaign not found or not active', 400); @@ -339,7 +339,10 @@ export async function settleCancelledCampaign( if (option === SettlementOption.REFUND_TO_DONOR) { // Mark each donation as refunded and notify donors for (const donation of campaign.donations) { - await tx.donation.update({ where: { id: donation.id }, data: { status: DonationStatus.REFUNDED } }); + await tx.donation.update({ + where: { id: donation.id }, + data: { status: DonationStatus.REFUNDED }, + }); await tx.campaign.update({ where: { id: campaign.id }, data: { currentAmount: { decrement: donation.amount } },