Spaces:

safetrack
/

edtech

Paused

edtech / apps /api /src /routes /admin.ts
CognxSafeTrack
fix(audit): resolve all remaining technical debt issues
a888244 6 months ago
27.1 kB
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<string, typeof messages[0]>();
        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<string, { original: string, replacement: string, count: number }> = {};

        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 });
        }
    });
}