feat: make main live captions stream public

Reviewed-on: http://192.168.68.119:3000/adolforeyna/EMI-Backend/pulls/5

index.js

@@ -29,6 +29,7 @@ const limiter = rateLimit({
     limit: 500, // Limit each IP to 100 requests per `window` (here, per 15 minutes).
     standardHeaders: 'draft-8', // draft-6: `RateLimit-*` headers; draft-7 & draft-8: combined `RateLimit` header
     legacyHeaders: false, // Disable the `X-RateLimit-*` headers.
+    skip: (req) => req.path.startsWith("/live-captions"),
     keyGenerator: (req) => {
         const forwarded = req.headers["x-forwarded-for"]?.split(",")[0]; // Take the first IP in the list
         const ip = forwarded || req.ip; // Fallback to req.ip
@@ -184,6 +185,7 @@ const songsRoute = require('./routes/songs.js');
 const paymentsRoute = require('./routes/payments.js');
 const bibleRoute = require('./routes/bible.js');
 const chatRoute = require('./routes/chat.js');
+const liveCaptionsRoute = require('./routes/liveCaptions.js');
 const sessionChecker = require('./middleware/sessionChecker');
 // -- Private Routes
 app.use('/user', sessionChecker, profileRoute);
@@ -192,6 +194,7 @@ app.use('/payments', paymentsRoute);
 app.use('/bible', sessionChecker, bibleRoute);
 app.use('/songs', sessionChecker, songsRoute);
 app.use('/chat', sessionChecker, chatRoute);
+app.use('/live-captions', liveCaptionsRoute);
 // -- Public Routes
 const subsplashRoute = require('./routes/subsplash.js');
 app.use('/subsplash', subsplashRoute);

package.json

@@ -6,6 +6,8 @@
   "scripts": {
     "test": "npx mocha test/auth.test.js",
     "start": "node index.js",
+    "dev": "node --watch index.js",
+    "live-captions:test-sender": "node scripts/liveCaptionsTestSender.js",
     "docker": "docker compose up -d",
     "docker_restore": "docker-compose exec mongo mongorestore --db EMI_SOCIAL /dump/EMI_SOCIAL/",
     "docker_dump": "docker-compose exec mongo mongodump --uri ${MONGO_URL} --out /dump"

routes/liveCaptions.js

@@ -0,0 +1,247 @@
+var express = require('express');
+var router = express.Router();
+const { rateLimit } = require("express-rate-limit");
+
+const sessionChecker = require("../middleware/sessionChecker.js");
+
+const MAX_BUFFER_SIZE = 300;
+const DEFAULT_INITIAL_LIMIT = 40;
+const MAX_INITIAL_LIMIT = 120;
+const INACTIVITY_RESET_MS = 10 * 60 * 1000;
+const CAPTION_META_KEYS = new Set(["sequence", "createdAt", "original", "draft", "sourceLang", "lang", "isDraft", "status", "translations"]);
+
+const liveCaptionState = {
+    startedAt: Date.now(),
+    lastIngestAt: 0,
+    latestSequence: 0,
+    captions: [],
+};
+
+const liveCaptionsLimiter = rateLimit({
+    windowMs: 10 * 60 * 1000,
+    limit: 6000,
+    standardHeaders: "draft-8",
+    legacyHeaders: false,
+    keyGenerator: (req) => {
+        const forwarded = req.headers["x-forwarded-for"]?.split(",")[0];
+        const ip = forwarded || req.ip || "";
+        return ip.includes(":") ? ip.split(":")[0] : ip;
+    },
+});
+
+router.use(liveCaptionsLimiter);
+
+const normalizeLang = (lang = "") => {
+    const value = String(lang || "").trim().toLowerCase();
+    if (!value) return "";
+    const base = value.split(",")[0].split("-")[0].trim();
+    return base || value;
+};
+
+const normalizeTranslations = (translations) => {
+    if (!translations || typeof translations !== "object" || Array.isArray(translations)) return {};
+    const normalized = {};
+    for (const [langKey, translatedText] of Object.entries(translations)) {
+        const lang = normalizeLang(langKey);
+        const text = typeof translatedText === "string" ? translatedText.trim() : "";
+        if (!lang || !text) continue;
+        normalized[lang] = text;
+    }
+    return normalized;
+};
+
+const readText = (value) => {
+    if (typeof value === "string") return value.trim();
+    return "";
+};
+
+const extractDraftText = (body = {}) => {
+    const directDraft = readText(body?.draft);
+    if (directDraft) return directDraft;
+    const nestedDraft = readText(body?.draft?.text);
+    if (nestedDraft) return nestedDraft;
+    const fallbackText = readText(body?.text);
+    if (fallbackText) return fallbackText;
+    return "";
+};
+
+const buildTranslationsFromFlatPayload = (payload) => {
+    const ignoredKeys = new Set(["original", "draft", "sourceLang", "lang", "isDraft", "status", "translations"]);
+    const normalized = {};
+    for (const [key, value] of Object.entries(payload || {})) {
+        if (ignoredKeys.has(key)) continue;
+        const lang = normalizeLang(key);
+        const text = typeof value === "string" ? value.trim() : "";
+        if (!lang || !text) continue;
+        normalized[lang] = text;
+    }
+    return normalized;
+};
+
+const inferSourceLangFromTranslations = (original, translations) => {
+    const normalizedOriginal = String(original || "").trim();
+    if (!normalizedOriginal) return "original";
+    for (const [lang, text] of Object.entries(translations || {})) {
+        if (String(text || "").trim() === normalizedOriginal) return normalizeLang(lang);
+    }
+    return "original";
+};
+
+const getAvailableLanguages = () => {
+    const langs = new Set();
+    for (const caption of liveCaptionState.captions) {
+        Object.keys(caption || {}).forEach((key) => {
+            if (CAPTION_META_KEYS.has(key)) return;
+            const lang = normalizeLang(key);
+            if (lang) langs.add(lang);
+        });
+    }
+    return Array.from(langs).filter(Boolean).sort();
+};
+
+const resetLiveCaptionState = () => {
+    liveCaptionState.startedAt = Date.now();
+    liveCaptionState.lastIngestAt = 0;
+    liveCaptionState.latestSequence = 0;
+    liveCaptionState.captions = [];
+};
+
+const maybeResetForInactivity = () => {
+    if (!liveCaptionState.lastIngestAt) return;
+    if ((Date.now() - liveCaptionState.lastIngestAt) < INACTIVITY_RESET_MS) return;
+    resetLiveCaptionState();
+};
+
+router.get("/stream", async (req, res) => {
+    try {
+        maybeResetForInactivity();
+        const sinceSequence = Number.parseInt(req.query?.sinceSequence, 10);
+        const requestedLimit = Number.parseInt(req.query?.limit, 10);
+        const initialLimit = Number.isFinite(requestedLimit)
+            ? Math.max(1, Math.min(requestedLimit, MAX_INITIAL_LIMIT))
+            : DEFAULT_INITIAL_LIMIT;
+
+        let captions = [];
+        if (Number.isFinite(sinceSequence) && sinceSequence >= 0) {
+            captions = liveCaptionState.captions.filter((item) => item.sequence > sinceSequence);
+        } else {
+            captions = liveCaptionState.captions.slice(-initialLimit);
+        }
+
+        return res.json({
+            status: "ok",
+            latestSequence: liveCaptionState.latestSequence,
+            startedAt: new Date(liveCaptionState.startedAt).toISOString(),
+            availableLanguages: getAvailableLanguages(),
+            captions,
+        });
+    } catch (error) {
+        console.error("Error getting live captions stream", error);
+        return res.status(500).json({
+            status: "Internal server error",
+            latestSequence: liveCaptionState.latestSequence,
+            captions: [],
+            availableLanguages: [],
+        });
+    }
+});
+
+router.post("/ingest", async (req, res) => {
+    try {
+        // TODO: Add basic auth/API key validation before production roll-out.
+        const draft = extractDraftText(req.body || {});
+        const originalFromPayload = readText(req.body?.original);
+        const original = originalFromPayload || draft;
+        const requestedLang = normalizeLang(req.body?.lang);
+        const sourceLangFromRequest = normalizeLang(req.body?.sourceLang || (requestedLang && requestedLang !== "draft" ? requestedLang : ""));
+        const isDraft = !!draft || requestedLang === "draft" || sourceLangFromRequest === "draft" || req.body?.isDraft === true || req.body?.status === "draft";
+        const mapFromNested = normalizeTranslations(req.body?.translations);
+        const mapFromFlat = buildTranslationsFromFlatPayload(req.body);
+        const translations = isDraft ? {} : { ...mapFromNested, ...mapFromFlat };
+        const inferredSource = inferSourceLangFromTranslations(original, translations);
+        const sourceLang = isDraft ? "" : (sourceLangFromRequest || inferredSource);
+
+        if (!original) {
+            return res.status(400).json({ status: "Original text is required" });
+        }
+        if (sourceLang && sourceLang !== "original" && sourceLang !== "draft" && !translations[sourceLang]) {
+            translations[sourceLang] = original;
+        }
+
+        const sequence = liveCaptionState.latestSequence + 1;
+        const caption = {
+            sequence,
+            createdAt: new Date().toISOString(),
+            original,
+            sourceLang: sourceLang || undefined,
+            lang: isDraft ? "draft" : (sourceLang || undefined),
+            isDraft,
+            status: isDraft ? "draft" : "final",
+            ...translations,
+        };
+
+        liveCaptionState.latestSequence = sequence;
+        liveCaptionState.lastIngestAt = Date.now();
+        liveCaptionState.captions.push(caption);
+        if (liveCaptionState.captions.length > MAX_BUFFER_SIZE) {
+            liveCaptionState.captions.splice(0, liveCaptionState.captions.length - MAX_BUFFER_SIZE);
+        }
+
+        return res.json({
+            status: "ok",
+            caption,
+            latestSequence: liveCaptionState.latestSequence,
+            availableLanguages: getAvailableLanguages(),
+        });
+    } catch (error) {
+        console.error("Error ingesting live captions", error);
+        return res.status(500).json({ status: "Internal server error" });
+    }
+});
+
+router.post("/reset", async (_, res) => {
+    try {
+        // TODO: Add admin authorization before exposing this endpoint.
+        resetLiveCaptionState();
+        return res.json({ status: "ok" });
+    } catch (error) {
+        console.error("Error resetting live captions state", error);
+        return res.status(500).json({ status: "Internal server error" });
+    }
+});
+
+router.get("/public/stream", async (req, res) => {
+    try {
+        maybeResetForInactivity();
+        const sinceSequence = Number.parseInt(req.query?.sinceSequence, 10);
+        const requestedLimit = Number.parseInt(req.query?.limit, 10);
+        const initialLimit = Number.isFinite(requestedLimit)
+            ? Math.max(1, Math.min(requestedLimit, MAX_INITIAL_LIMIT))
+            : DEFAULT_INITIAL_LIMIT;
+
+        let captions = [];
+        if (Number.isFinite(sinceSequence) && sinceSequence >= 0) {
+            captions = liveCaptionState.captions.filter((item) => item.sequence > sinceSequence);
+        } else {
+            captions = liveCaptionState.captions.slice(-initialLimit);
+        }
+
+        return res.json({
+            status: "ok",
+            latestSequence: liveCaptionState.latestSequence,
+            startedAt: new Date(liveCaptionState.startedAt).toISOString(),
+            availableLanguages: getAvailableLanguages(),
+            captions,
+        });
+    } catch (error) {
+        console.error("Error getting public live captions stream", error);
+        return res.status(500).json({
+            status: "Internal server error",
+            latestSequence: liveCaptionState.latestSequence,
+            captions: [],
+            availableLanguages: [],
+        });
+    }
+});
+
+module.exports = router;

scripts/liveCaptionsTestSender.js

@@ -0,0 +1,89 @@
+#!/usr/bin/env node
+
+require("dotenv").config();
+const axios = require("axios");
+
+const baseUrl = (process.env.CAPTION_TEST_BASE_URL || process.env.BASE_URL || "http://localhost:3000").replace(/\/+$/, "");
+const ingestUrl = `${baseUrl}/live-captions/ingest`;
+const intervalMs = 6000;
+
+const samples = [
+    {
+        original: "Bienvenidos a nuestro servicio de adoracion.",
+        es: "Bienvenidos a nuestro servicio de adoracion.",
+        en: "Welcome to our worship service.",
+        fr: "Bienvenue a notre service de louange.",
+    },
+    {
+        original: "Leamos juntos en el Salmo 23.",
+        es: "Leamos juntos en el Salmo 23.",
+        en: "Let us read together in Psalm 23.",
+        fr: "Lisons ensemble le Psaume 23.",
+    },
+    {
+        original: "Dios es fiel en todo tiempo.",
+        es: "Dios es fiel en todo tiempo.",
+        en: "God is faithful at all times.",
+        fr: "Dieu est fidele en tout temps.",
+    },
+    {
+        original: "Tomemos un momento para orar.",
+        es: "Tomemos un momento para orar.",
+        en: "Let us take a moment to pray.",
+        fr: "Prenons un moment pour prier.",
+    },
+];
+
+let sampleIndex = 0;
+let timer = null;
+
+const postPayload = async (payload) => {
+    const kind = payload?.draft ? "draft" : "final";
+
+    try {
+        const response = await axios.post(ingestUrl, payload, {
+            headers: { "Content-Type": "application/json" },
+            timeout: 10000,
+        });
+        const seq = response?.data?.caption?.sequence || response?.data?.latestSequence || "?";
+        const text = payload?.draft || payload?.original || "";
+        console.log(`[live-captions:test-sender] sent ${kind} sequence=${seq} text="${text}"`);
+    } catch (error) {
+        const status = error?.response?.status;
+        const body = error?.response?.data;
+        const message = error?.message || "request failed";
+        console.error("[live-captions:test-sender] send failed", { status, body, message });
+    }
+};
+
+const sendNextSample = async () => {
+    const payload = samples[sampleIndex];
+
+    const draftWords = String(payload?.original || "").split(" ").filter(Boolean);
+    if (draftWords.length > 2) {
+        await postPayload({ draft: draftWords.slice(0, 2).join(" ") });
+        await new Promise((resolve) => setTimeout(resolve, 550));
+        await postPayload({ draft: draftWords.slice(0, 4).join(" ") });
+        await new Promise((resolve) => setTimeout(resolve, 550));
+    }
+
+    await postPayload(payload);
+    sampleIndex = (sampleIndex + 1) % samples.length;
+};
+
+const start = async () => {
+    console.log(`[live-captions:test-sender] posting to ${ingestUrl} every ${intervalMs / 1000}s`);
+    await sendNextSample();
+    timer = setInterval(sendNextSample, intervalMs);
+};
+
+const shutdown = () => {
+    if (timer) clearInterval(timer);
+    console.log("[live-captions:test-sender] stopped");
+    process.exit(0);
+};
+
+process.on("SIGINT", shutdown);
+process.on("SIGTERM", shutdown);
+
+start();