import { FastifyInstance } from 'fastify'; import { prisma } from '../services/prisma'; import { whatsappQueue } from '../services/queue'; import { logger } from '../logger'; import { z } from 'zod'; import { calculateWER, formatError } from '../utils/metrics'; import fs from 'fs'; import path from 'path'; import * as diff from 'diff'; import { normalizeWolof } from '../scripts/normalizeWolof'; import { redis } from '../lib/redis'; // Services import { AnalyticsService } from '../services/analytics-service'; import { UserAdminService } from '../services/user-admin-service'; import { TrackService } from '../services/track-service'; import { TrainingService } from '../services/training-service'; // ─── Zod Schemas ─────────────────────────────────────────────────────────────── const TrackSchema = z.object({ title: z.string().min(1), description: z.string().optional(), duration: z.number().int().positive(), language: z.enum(['FR', 'WOLOF']).default('FR'), isPremium: z.boolean().default(false), priceAmount: z.number().int().optional(), }); const TrackDaySchema = z.object({ dayNumber: z.number().int().positive(), title: z.string().optional(), lessonText: z.string().optional(), audioUrl: z.string().url().optional().or(z.literal('')), exerciseType: z.enum(['TEXT', 'AUDIO', 'BUTTON']).default('TEXT'), exercisePrompt: z.string().optional(), validationKeyword: z.string().optional(), buttonsJson: z.array(z.object({ id: z.string(), title: z.string() })).optional(), unlockCondition: z.string().optional(), exerciseCriteria: z.record(z.unknown()).optional(), badges: z.array(z.string()).optional(), }); const OverrideFeedbackSchema = z.object({ userId: z.string(), trackId: z.string(), transcription: z.string().min(1), overrideAudioUrl: z.string().url(), adminId: z.string() }); const PaginationSchema = z.object({ page: z.coerce.number().int().positive().default(1), limit: z.coerce.number().int().positive().max(100).default(50) }); export async function adminRoutes(fastify: FastifyInstance) { fastify.addHook('preHandler', async (request, reply) => { const role = (request as any).user?.role; if (!role || !['ORG_ADMIN', 'ADMIN', 'SUPER_ADMIN'].includes(role)) { return reply.code(403).send({ error: 'Admin access required' }); } }); // ── Dashboard Stats ──────────────────────────────────────────────────────── fastify.get('/stats', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const stats = await AnalyticsService.getDashboardStats(organizationId); return stats; }); // ── Users ────────────────────────────────────────────────────────────────── fastify.get('/users', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const parsed = PaginationSchema.safeParse(req.query); if (!parsed.success) return reply.code(400).send({ error: 'Invalid query parameters', details: parsed.error.flatten() }); const { page, limit } = parsed.data; const result = await UserAdminService.listUsers(organizationId, page, limit); return result; }); fastify.get('/users/:userId/messages', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const { userId } = req.params as { userId: string }; const result = await UserAdminService.getUserMessages(organizationId, userId); if (!result) return reply.status(404).send({ error: 'User not found or access denied' }); return result; }); // Soft-delete a user (sets deletedAt instead of hard delete) fastify.delete<{ Params: { userId: string } }>('/users/:userId', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const ok = await UserAdminService.softDeleteUser(organizationId, req.params.userId); return ok ? { ok: true } : reply.code(404).send({ error: 'User not found' }); }); // Release a user from AI handoff mode (clears the 24h Redis pause flag) fastify.delete<{ Params: { userId: string } }>('/users/:userId/handoff', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const key = `handoff:${req.params.userId}:${organizationId}`; const deleted = await redis.del(key); return { ok: true, wasActive: deleted > 0 }; }); // Check handoff status for a user fastify.get<{ Params: { userId: string } }>('/users/:userId/handoff', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const key = `handoff:${req.params.userId}:${organizationId}`; const active = await redis.exists(key); return { handoffActive: active > 0 }; }); // ── Enrollments ──────────────────────────────────────────────────────────── fastify.get('/enrollments', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const parsed = PaginationSchema.safeParse(req.query); if (!parsed.success) return reply.code(400).send({ error: 'Invalid query parameters', details: parsed.error.flatten() }); const { page, limit } = parsed.data; const [enrollments, total] = await Promise.all([ prisma.enrollment.findMany({ where: { organizationId, deletedAt: null }, include: { user: { select: { id: true, phone: true, name: true, language: true, activity: true, role: true, createdAt: true } }, track: true }, orderBy: { startedAt: 'desc' }, skip: (page - 1) * limit, take: limit, }), prisma.enrollment.count({ where: { organizationId, deletedAt: null } }), ]); return { data: enrollments, total, page, limit, totalPages: Math.ceil(total / limit) }; }); // Soft-delete an enrollment fastify.delete<{ Params: { enrollmentId: string } }>('/enrollments/:enrollmentId', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const ok = await UserAdminService.softDeleteEnrollment(organizationId, req.params.enrollmentId); return ok ? { ok: true } : reply.code(404).send({ error: 'Enrollment not found' }); }); // ── Human-in-the-Loop / Audio Overdrive ──────────────────────────────────── fastify.get('/live-feed', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const pendingReviews = await prisma.userProgress.findMany({ where: { organizationId, exerciseStatus: 'PENDING_REVIEW', user: { language: 'WOLOF' } }, include: { user: { select: { id: true, name: true, phone: true, activity: true, language: true } }, track: { select: { id: true, title: true } } }, orderBy: { updatedAt: 'asc' } }); if (pendingReviews.length === 0) return []; const userIds = pendingReviews.map(p => p.userId); const trackIds = [...new Set(pendingReviews.map(p => p.trackId))]; // Batch fetch enrollments and latest inbound media messages — no N+1 const [enrollments, messages] = await Promise.all([ prisma.enrollment.findMany({ where: { userId: { in: userIds }, trackId: { in: trackIds }, status: 'ACTIVE', organizationId }, }), prisma.message.findMany({ where: { userId: { in: userIds }, direction: 'INBOUND', mediaUrl: { not: null }, organizationId }, orderBy: { createdAt: 'desc' }, take: 200, }), ]); const enrollmentMap = new Map(enrollments.map(e => [`${e.userId}:${e.trackId}`, e])); const latestMessageMap = new Map(); for (const msg of messages) { if (msg.userId && !latestMessageMap.has(msg.userId)) latestMessageMap.set(msg.userId, msg); } const liveFeed = pendingReviews.map(progress => { const enrollment = enrollmentMap.get(`${progress.userId}:${progress.trackId}`); const lastMessage = latestMessageMap.get(progress.userId); return { ...progress, audioUrl: lastMessage?.mediaUrl || null, content: progress.adminTranscription || lastMessage?.content || null, dayNumber: enrollment ? enrollment.currentDay : null, }; }); return liveFeed; }); fastify.post('/override-feedback', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization context missing' }); const body = OverrideFeedbackSchema.safeParse(req.body); if (!body.success) return reply.code(400).send({ error: body.error.flatten() }); const { userId, trackId, transcription, overrideAudioUrl, adminId } = body.data; try { const progress = await prisma.userProgress.update({ where: { userId_trackId: { userId, trackId } }, data: { exerciseStatus: 'COMPLETED', adminTranscription: transcription, overrideAudioUrl: overrideAudioUrl, reviewedBy: adminId } }); const enrollment = await prisma.enrollment.findFirst({ where: { userId, trackId, organizationId, status: 'ACTIVE' } }); const currentDay = enrollment ? Math.floor(enrollment.currentDay) : 0; await prisma.businessProfile.upsert({ where: { userId }, update: { lastUpdatedFromDay: currentDay, organizationId }, create: { userId, lastUpdatedFromDay: currentDay, organizationId } }); try { await whatsappQueue.add('send-admin-audio-override', { userId, trackId, overrideAudioUrl, adminId, organizationId }); logger.info({ userId, trackId }, '[ADMIN] send-admin-audio-override enqueued'); } catch (qErr) { logger.error({ qErr, userId }, '[ADMIN] Failed to enqueue send-admin-audio-override'); } return reply.code(200).send({ ok: true, progress }); } catch (err: any) { if (err.code === 'P2025') { return reply.code(404).send({ error: 'UserProgress not found' }); } logger.error({ err, userId, trackId }, '[ADMIN] override-feedback failed'); return reply.code(500).send({ error: 'Failed to update progress' }); } }); fastify.post('/messages/send', async (req, reply) => { const schema = z.object({ userId: z.string(), text: z.string().min(1) }); const parsed = schema.safeParse(req.body); if (!parsed.success) return reply.code(400).send({ error: parsed.error.flatten() }); const body = parsed.data; const user = await prisma.user.findUnique({ where: { id: body.userId, organizationId: req.organizationId ?? undefined }, select: { phone: true, organizationId: true } }); if (!user || !user.phone) { return reply.code(404).send({ error: "User or phone number not found." }); } try { await whatsappQueue.add('send-message-direct', { phone: user.phone, text: body.text, organizationId: user.organizationId }); logger.info({ phone: user.phone }, '[ADMIN] send-message-direct enqueued'); } catch (qErr) { logger.error({ qErr, userId: body.userId }, '[ADMIN] Failed to enqueue send-message-direct'); return reply.code(500).send({ error: 'Failed to queue message' }); } return reply.code(200).send({ ok: true, message: "Custom message queued for sending." }); }); // ══════════════════════════════════════════════════════════════════════════ // TRACKS CRUD // ══════════════════════════════════════════════════════════════════════════ fastify.get('/tracks', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); return TrackService.listTracks(organizationId); }); fastify.get<{ Params: { id: string } }>('/tracks/:id', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const track = await TrackService.getTrack(organizationId, req.params.id); if (!track) return reply.code(404).send({ error: 'Track not found or access denied' }); return track; }); fastify.post('/tracks', async (req, reply) => { const body = TrackSchema.safeParse(req.body); if (!body.success) return reply.code(400).send({ error: body.error.flatten() }); const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization context missing' }); const track = await TrackService.createTrack(organizationId, body.data); return reply.code(201).send(track); }); fastify.put<{ Params: { id: string } }>('/tracks/:id', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const body = TrackSchema.partial().safeParse(req.body); if (!body.success) return reply.code(400).send({ error: body.error.flatten() }); try { const track = await TrackService.updateTrack(organizationId, req.params.id, body.data); return track; } catch (err) { logger.warn({ err, trackId: req.params.id }, '[ADMIN] updateTrack failed'); return reply.code(404).send({ error: 'Track not found or access denied' }); } }); fastify.delete<{ Params: { id: string } }>('/tracks/:id', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const ok = await TrackService.deleteTrack(organizationId, req.params.id); if (!ok) return reply.code(404).send({ error: 'Track not found or access denied' }); return { ok: true }; }); // ── STT Quality Calibration Endpoint ─────────────────────────────────────── fastify.get('/stats/confidence-distribution', async (_req, reply) => { const statsPath = path.join(__dirname, '../../data/calibration_stats.json'); try { if (fs.existsSync(statsPath)) { return JSON.parse(fs.readFileSync(statsPath, 'utf8')); } return reply.code(404).send({ error: "Calibration not run yet" }); } catch (err: unknown) { return reply.code(500).send({ error: formatError(err) }); } }); // ══════════════════════════════════════════════════════════════════════════ // TRACK DAYS CRUD // ══════════════════════════════════════════════════════════════════════════ fastify.get<{ Params: { trackId: string } }>('/tracks/:trackId/days', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); return TrackService.listDays(organizationId, req.params.trackId); }); fastify.post<{ Params: { trackId: string } }>('/tracks/:trackId/days', async (req, reply) => { const body = TrackDaySchema.safeParse(req.body); if (!body.success) return reply.code(400).send({ error: body.error.flatten() }); const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization context missing' }); const day = await TrackService.createDay(organizationId, req.params.trackId, body.data); return reply.code(201).send(day); }); fastify.put<{ Params: { trackId: string; dayId: string } }>('/tracks/:trackId/days/:dayId', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const body = TrackDaySchema.partial().safeParse(req.body); if (!body.success) return reply.code(400).send({ error: body.error.flatten() }); try { return await TrackService.updateDay(organizationId, req.params.dayId, body.data); } catch (err) { logger.warn({ err, dayId: req.params.dayId }, '[ADMIN] updateDay failed'); return reply.code(404).send({ error: 'Day not found or access denied' }); } }); fastify.delete<{ Params: { trackId: string; dayId: string } }>('/tracks/:trackId/days/:dayId', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); try { await TrackService.deleteDay(organizationId, req.params.dayId); return { ok: true }; } catch (err) { logger.warn({ err, dayId: req.params.dayId }, '[ADMIN] deleteDay failed'); return reply.code(404).send({ error: 'Day not found or access denied' }); } }); // ── Training Lab Endpoints ─────────────────────────────────────────────── fastify.get('/training/audios', async (req, reply) => { const parsed = PaginationSchema.safeParse(req.query); if (!parsed.success) return reply.code(400).send({ error: 'Invalid query', details: parsed.error.flatten() }); const { page, limit } = parsed.data; return TrainingService.getPendingAudios(page, limit); }); fastify.post('/training/submit', async (req, reply) => { const schema = z.object({ id: z.string().uuid().optional(), audioUrl: z.string(), transcription: z.string(), manualCorrection: z.string() }); const body = schema.safeParse(req.body); if (!body.success) return reply.code(400).send({ error: body.error.flatten() }); const result = await TrainingService.submitCorrection(body.data.id, body.data.audioUrl, body.data.transcription, body.data.manualCorrection); return reply.send(result); }); fastify.get('/training/suggestions', async (_req, reply) => { const trainingData = await prisma.trainingData.findMany({ where: { status: 'REVIEWED' }, take: 500 }); const substitutionCounts: Record = {}; trainingData.forEach(item => { if (!item.manualCorrection) return; const changes = diff.diffWords(item.transcription, item.manualCorrection); for (let i = 0; i < changes.length - 1; i++) { if (changes[i].removed && changes[i + 1].added) { const original = changes[i].value.trim().toLowerCase().replace(/[.,/#!$%^&*;:{}=\-_`~()]/g, ""); const replacement = changes[i + 1].value.trim().toLowerCase().replace(/[.,/#!$%^&*;:{}=\-_`~()]/g, ""); if (original && replacement && original !== replacement && original.split(' ').length === 1 && replacement.split(' ').length === 1) { const key = `${original}->${replacement}`; if (!substitutionCounts[key]) substitutionCounts[key] = { original, replacement, count: 0 }; substitutionCounts[key].count++; } } } }); return reply.send(Object.values(substitutionCounts).sort((a, b) => b.count - a.count).slice(0, 20)); }); fastify.get('/training/rules', async () => { return prisma.normalizationRule.findMany({ orderBy: { original: 'asc' } }); }); fastify.post('/training/apply-suggestions', async (req, reply) => { const schema = z.object({ suggestions: z.array(z.object({ original: z.string(), replacement: z.string() })) }); const body = schema.safeParse(req.body); if (!body.success) return reply.code(400).send({ error: body.error.flatten() }); if (body.data.suggestions.length === 0) return reply.send({ ok: true, message: "No suggestions provided" }); const { normalizationService } = await import('../services/normalization'); await normalizationService.saveRules(body.data.suggestions); return reply.send({ ok: true, injectedCount: body.data.suggestions.length }); }); fastify.post('/training/recalculate-wer', async (_req, reply) => { const trainingData = await prisma.trainingData.findMany({ where: { status: 'REVIEWED' }, take: 500 }); const { normalizationService } = await import('../services/normalization'); const customRules = await normalizationService.getRules('WOLOF'); let totalRawWER = 0, totalNormalizedWER = 0, count = 0; for (const item of trainingData) { if (!item.manualCorrection) continue; const rawWER = calculateWER(item.manualCorrection, item.transcription); const normResult = normalizeWolof(item.transcription, customRules); const normalizedWER = calculateWER(item.manualCorrection, normResult.normalizedText); await prisma.trainingData.update({ where: { id: item.id }, data: { rawWER, normalizedWER } }); totalRawWER += rawWER; totalNormalizedWER += normalizedWER; count++; } const avgRaw = count > 0 ? totalRawWER / count : 0; const avgNorm = count > 0 ? totalNormalizedWER / count : 0; return reply.send({ processed: count, avgRawWER: avgRaw, avgNormalizedWER: avgNorm, improvementPercent: count > 0 && avgRaw > 0 ? ((avgRaw - avgNorm) / avgRaw) * 100 : 0 }); }); fastify.post('/training/upload', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const parts = req.parts(); let fileBuffer: Buffer | null = null, filename = 'audio.ogg', mimeType = 'audio/ogg'; for await (const part of parts) { if (part.type === 'file') { fileBuffer = await part.toBuffer(); filename = part.filename || filename; mimeType = part.mimetype || mimeType; } } if (!fileBuffer || fileBuffer.length === 0) return reply.code(400).send({ error: 'No audio file provided' }); try { const { uploadFile } = await import('../services/storage'); const { aiService } = await import('../services/ai'); const { convertToMp3IfNeeded } = await import('@repo/ai-sdk'); const audioUrl = await uploadFile(fileBuffer, filename, mimeType, organizationId); const { buffer: mp3Buffer, format } = await convertToMp3IfNeeded(fileBuffer, filename); const { text: transcription } = await aiService.transcribeAudio(mp3Buffer, `upload.${format}`); const record = await TrainingService.createTrainingRecord(audioUrl, transcription); return reply.send({ ok: true, id: record.id, audioUrl, transcription }); } catch (err: any) { logger.error({ err, organizationId }, '[TRAINING_UPLOAD] Failed'); return reply.code(500).send({ error: 'Upload failed', detail: err.message }); } }); // Simple file upload for admin override audio (no training record created) fastify.post('/upload', async (req, reply) => { const organizationId = req.organizationId; if (!organizationId) return reply.code(400).send({ error: 'Organization ID required' }); const parts = req.parts(); let fileBuffer: Buffer | null = null, filename = 'audio.webm', mimeType = 'audio/webm'; for await (const part of parts) { if (part.type === 'file') { fileBuffer = await part.toBuffer(); filename = part.filename || filename; mimeType = part.mimetype || mimeType; } } if (!fileBuffer || fileBuffer.length === 0) return reply.code(400).send({ error: 'No file provided' }); try { const { uploadFile } = await import('../services/storage'); const url = await uploadFile(fileBuffer, filename, mimeType, organizationId); return reply.send({ ok: true, url }); } catch (err: any) { logger.error({ err, organizationId }, '[ADMIN_UPLOAD] Failed'); return reply.code(500).send({ error: 'Upload failed', detail: err.message }); } }); }