collabcentral/index.js

#!/usr/bin/env node

import fs from 'fs';
import fetch from 'node-fetch';
import { Headers } from 'node-fetch';
import express from 'express';
import bodyParser from 'body-parser';
import multer from 'multer';
import sharp from 'sharp';
import { v4 as uuid } from 'uuid';
import cookieParser from 'cookie-parser';
import FormData from 'form-data';
import cron from "node-cron";
import PQueue from 'p-queue';
import * as helpers from './lib/helpers.js';
import { createWebexClient } from './lib/webex.js';
import { createTranslator } from './lib/translation.js';
import { createJobsPipeline } from './lib/jobs.js';
import { registerInfoRoutes } from './routes/info.js';
import { registerAuthRoutes } from './routes/auth.js';
import { registerUserRoutes } from './routes/user.js';
import { registerAdminRoutes } from './routes/admin.js';
import { registerJobsRoutes } from './routes/jobs.js';

const queue = new PQueue({ concurrency: 10 });

//Load the config files from storage.
// config.json  = committed, structural (server port, bot labels, integration ids).
// authorized.json = gitignored, mutable per-user data:
//     { admins: [personId...], bot: { <appName>: { <personId>: { ...profile, groups: [...] } } } }
// jobs.json    = gitignored, runtime queue state.
// userPrefs.json = gitignored, per-recipient preferences (language, ...).
var config = JSON.parse(fs.readFileSync('./config/config.json'));
var jobs = JSON.parse(fs.readFileSync('./config/jobs.json'));
var userPrefs = JSON.parse(fs.readFileSync('./config/userPrefs.json'));
var authorized = loadAuthorized();

function loadAuthorized() {
    try {
        var doc = JSON.parse(fs.readFileSync('./config/authorized.json'));
        doc.admins = Array.isArray(doc.admins) ? doc.admins : [];
        doc.bot = (doc.bot && typeof doc.bot === 'object') ? doc.bot : {};
        return doc;
    } catch (err) {
        // Missing file (fresh deploy) is fine — we start empty and the admin
        // page can populate. Any other parse error is fatal because otherwise
        // the whole authorization layer silently permits nothing.
        if (err && err.code === 'ENOENT') {
            console.log('loadAuthorized: authorized.json not found, starting empty.');
            return { admins: [], bot: {} };
        }
        throw err;
    }
}

// Returns the mutable per-user record for (appName, personId), creating the
// bot bucket if this is the first write. Never returns null when personId is
// truthy — callers use this to add favorites / users without null-checking.
// Read-only callers should prefer helpers.getAuthorizedEntry(authorized, ...).
function getOrCreateAuthorizedEntry(appName, personId) {
    if (!authorized.bot[appName]) authorized.bot[appName] = {};
    return authorized.bot[appName][personId] || null;
}

// Defensive init so code below can safely `.push`, `.filter`, and index into
// these regardless of what the on-disk jobs.json happens to contain.
jobs.building = jobs.building || {};
jobs.running = Array.isArray(jobs.running) ? jobs.running : [];
jobs.scheduled = Array.isArray(jobs.scheduled) ? jobs.scheduled : [];
jobs.completed = Array.isArray(jobs.completed) ? jobs.completed : [];
if (typeof jobs.lastJobNumber !== 'number') jobs.lastJobNumber = 0;

// Per-bot tokens live in their own gitignored file so adding a new bot is one
// JSON entry + an env-free restart. Shape:
//   { "<appName>": { "token": "...", "enabled": true } }
var botTokens = loadBotTokens();

function loadBotTokens() {
    try {
        return JSON.parse(fs.readFileSync('./config/botTokens.json'));
    } catch (err) {
        console.log('loadBotTokens failed: ' + err.message);
        return {};
    }
}

function getBotToken(appName) {
    return helpers.getBotToken(botTokens, appName);
}

function isBotEnabled(appName) {
    return helpers.isBotEnabled(botTokens, appName);
}

// Returns the bot's config block if the bot is both defined in config.json AND
// has an enabled token entry. Returns null otherwise. Use this everywhere
// instead of reaching into config.webex.bot[...] directly so unknown or
// disabled bots can't crash the request.
function getBotConfig(appName) {
    return helpers.getBotConfig(config, botTokens, appName);
}

// Cached bot profiles populated from Webex /people/me at startup so the
// front-end and the completion card can use the bot's actual displayName and
// avatar without hardcoding either in config.json.
//   botProfiles[appName] = { displayName, avatar }
var botProfiles = {};

async function loadBotProfile(appName) {
    var token = getBotToken(appName);
    if (!token) return;
    try {
        var response = await webex.fetchWithRateLimit('https://webexapis.com/v1/people/me', {
            method: 'GET',
            headers: { 'Authorization': 'Bearer ' + token }
        });
        if (!response.ok) {
            logger('loadBotProfile', appName + ' /people/me failed: ' + response.status + ' ' + response.statusText);
            return;
        }
        var data = await response.json();
        botProfiles[appName] = {
            displayName: data.displayName,
            avatar: data.avatar,
            personId: data.id,
            emails: data.emails
        };
        logger('loadBotProfile', appName + ' = ' + data.displayName);
    } catch (err) {
        logger('loadBotProfile', appName + ' error: ' + (err && err.message || err));
    }
}

function loadBotProfiles() {
    var apps = Object.keys((config.webex && config.webex.bot) || {});
    return Promise.allSettled(apps.filter(isBotEnabled).map(loadBotProfile));
}

function getBotProfile(appName) {
    return botProfiles[appName] || null;
}

// Cached list of every Webex group in the org, used by the "Additional groups"
// dropdown on the compose page. findWebexGroup() paginates through ~10k groups
// 500 at a time, which used to run on every /user/groups/find request; we now
// refresh once at startup and once a day and serve everything else from
// memory. `refreshing` is a shared in-flight Promise so a concurrent refresh
// (e.g. the cron firing during a slow startup fetch) can't stampede.
var groupsCache = {
    data: null,
    lastRefreshed: null,
    refreshing: null,
};

function refreshGroupsCache() {
    if (groupsCache.refreshing) return groupsCache.refreshing;
    var startedAt = Date.now();
    logger('groupsCache', 'Refreshing group list from Webex...');
    groupsCache.refreshing = webex.findWebexGroup()
        .then(function (groups) {
            groupsCache.data = groups;
            groupsCache.lastRefreshed = new Date();
            logger('groupsCache',
                'Cached ' + groups.length + ' groups in ' +
                (Date.now() - startedAt) + 'ms.');
            return groups;
        })
        .catch(function (err) {
            logger('groupsCache', 'Refresh failed: ' + (err && err.message || err));
            throw err;
        })
        .finally(function () {
            groupsCache.refreshing = null;
        });
    return groupsCache.refreshing;
}

// Returns the cached groups if we have any; otherwise waits for whatever
// refresh is currently in flight (or kicks one off). Requests only ever block
// on the very first call after startup — every subsequent call resolves
// synchronously from memory.
function getGroupsFromCache() {
    if (groupsCache.data) return Promise.resolve(groupsCache.data);
    return refreshGroupsCache();
}

// Express middleware: 404s any /CollabCentral/:app/* request whose `:app` is
// not a known + enabled bot. Attaches the bot config to req.botConfig for
// downstream handlers.
function requireBot(req, res, next) {
    var botCfg = getBotConfig(req.params.app);
    if (!botCfg) {
        logger("requireBot", "Unknown or disabled bot: " + req.params.app);
        return res.status(404).send("Unknown bot: " + req.params.app);
    }
    req.botConfig = botCfg;
    next();
}

// Building (draft) jobs are keyed by cookieId AND appName so a user who is
// authorized for more than one bot can have one independent draft per bot.
function buildingKey(req) {
    return helpers.buildingKey(req.cookies && req.cookies.id, req.params.app);
}

function jobsForApp(arr, appName) {
    return helpers.jobsForApp(arr, appName);
}

// Service-account OAuth tokens are rewritten in place by refreshToken(), so
// they live in their own file rather than mixed into config.json.
var serviceAccountToken = loadServiceAccountToken();

function loadServiceAccountToken() {
    try {
        return JSON.parse(fs.readFileSync('./config/token.json'));
    } catch (err) {
        console.log('loadServiceAccountToken failed: ' + err.message);
        return {};
    }
}

function saveServiceAccountToken(tokenObj) {
    // Keep the in-memory copy in sync even if the disk write fails so the
    // refreshed token is still usable for the rest of the process lifetime.
    serviceAccountToken = tokenObj;
    try {
        fs.writeFileSync('./config/token.json', JSON.stringify(tokenObj, null, 4));
    } catch (err) {
        logger("saveServiceAccountToken", "Failed to persist token.json: " + err.message);
    }
}

function getServiceAccountAccessToken() {
    return serviceAccountToken && serviceAccountToken.access_token;
}

// Builds the Webex OAuth authorize URL for a given bot. The redirect_uri must
// match one of the URIs registered with the Webex integration in the Webex
// developer portal (one URI per bot). Returns null if required env vars are
// missing so callers can return an actionable error instead of a malformed URL.
function buildAuthUrl(appName) {
    var url = helpers.buildAuthUrl({
        clientId: process.env.WEBEX_INTEGRATION_CLIENT_ID,
        template: process.env.OAUTH_CALLBACK_URL_TEMPLATE,
        appName: appName,
    });
    if (!url) logger('buildAuthUrl', 'Missing WEBEX_INTEGRATION_CLIENT_ID or OAUTH_CALLBACK_URL_TEMPLATE env var.');
    return url;
}

function getOAuthRedirectUri(appName) {
    return helpers.getOAuthRedirectUri(process.env.OAUTH_CALLBACK_URL_TEMPLATE, appName);
}

// ---------------------------------------------------------------------------
// Wire the lib/* modules. Each takes a dependency bag so the modules
// themselves stay free of module-level mutable state — everything mutable
// (tokens, jobs, queue) still lives here in index.js.
// ---------------------------------------------------------------------------
var webex = createWebexClient({
    getBotToken: getBotToken,
    getServiceAccountAccessToken: getServiceAccountAccessToken,
    getServiceAccountRefreshToken: function () { return serviceAccountToken && serviceAccountToken.refresh_token; },
    saveServiceAccountToken: saveServiceAccountToken,
    env: process.env,
    logger: logger,
});

var translator = createTranslator({
    languages: config.languages || [],
    apiKey: process.env.GOOGLE_TRANSLATE_API_KEY || '',
    fetchWithRateLimit: webex.fetchWithRateLimit,
    logger: logger,
});

var jobsPipeline = createJobsPipeline({
    jobs: jobs,
    queue: queue,
    userPrefs: userPrefs,
    webex: webex,
    getBotConfig: function (appName) { return helpers.getBotConfig(config, botTokens, appName); },
    getBotProfile: getBotProfile,
    saveJobs: function () { saveConfig(jobs, './config/jobs.json'); },
    msToTime: helpers.msToTime,
    logger: logger,
});

//Load Express Server
var app = express();
app.use(bodyParser.json());
app.use(cookieParser());

var serverPort = process.env.SERVER_PORT || config.server.port;
var server = app.listen(serverPort, function () {
    logger("startup", config.server.name + " running on port " + serverPort + ".");
    loadBotProfiles()
        .then(() => logger("startup", "Loaded " + Object.keys(botProfiles).length + " bot profile(s)."))
        .catch(err => logger("startup", "loadBotProfiles error: " + (err && err.message || err)));
    // Warm the groups cache in the background so the very first user hitting
    // /user/groups/find gets an instant response. Failures are non-fatal:
    // getGroupsFromCache() will simply try again on the first request.
    refreshGroupsCache().catch(function () { /* logged inside */ });
});
server.setTimeout(3000000);
const upload = multer({ limits: { fileSize: 4000000 } }).fields(
    [
        { name: 'uploadImage' },
        { name: 'uploadCSV' }
    ]
);

// Anchor both crons to a specific timezone so the schedule is deterministic
// regardless of the host/container clock. Override with CRON_TIMEZONE if the
// deployment moves to a different region.
var cronTimezone = process.env.CRON_TIMEZONE || 'America/New_York';

// Daily at 01:10 local time: prune completed jobs older than the retention
// window and rewrite jobs.json.
cron.schedule('0 10 1 * * *', async function () {
    var jobFile = JSON.parse(fs.readFileSync('./config/jobs.json'));
    var cleanJobs = await cleanCompletedJobs(jobFile);
    saveConfig(cleanJobs, './config/jobs.json');
    jobs = JSON.parse(fs.readFileSync('./config/jobs.json'));

}, { timezone: cronTimezone })

// Every minute: refresh the service-account token if it's inside the renewal
// window, and dispatch any scheduled jobs whose time has arrived.
cron.schedule('0 * * * * *', () => {
    var renewBy = serviceAccountToken && serviceAccountToken.renewBy;
    if (renewBy && new Date(new Date(renewBy) - 7200000) < new Date()) {
        logger("refreshToken", "Renew by: " + new Date(renewBy).toLocaleString())
        logger("refreshToken", "Renew after: " + new Date(new Date(renewBy) - 7200000).toLocaleString());
        logger("refreshToken", "Refreshing token.")
        webex.refreshToken()
            .then(response => {
                logger("tokenRefresh", response);
            })
            .catch(error => logger("refreshToken", error));
    }
    jobsPipeline.checkScheduledJobs()
        .catch(error => logger("checkScheduledJobs", error));
}, { timezone: cronTimezone })

// Daily at 03:00 local time: rebuild the org-wide groups cache. Offset from
// the 01:10 jobs cleanup so the two crons don't fight over the event loop.
cron.schedule('0 0 3 * * *', function () {
    refreshGroupsCache().catch(function () { /* logged inside */ });
}, { timezone: cronTimezone })

//Routes to be used
// Validate the :app segment for every /CollabCentral/:app/* request. This
// runs before the static mount and all the per-route handlers below, so any
// unknown or disabled bot gets a clean 404 instead of crashing on undefined.
app.use('/CollabCentral/:app', requireBot)
app.use('/CollabCentral/:app', express.static('html'))

// Route registrations. Each module receives a dep-bag so the routes
// themselves stay decoupled from the concrete state/utilities defined
// here. Order matters only for /oauth (must come after requireBot).
registerInfoRoutes(app, {
    getBotProfile: getBotProfile,
    isAuthorized: isAuthorized,
    isAdmin: isAdmin,
});

registerAuthRoutes(app, {
    buildAuthUrl: buildAuthUrl,
    getOAuthRedirectUri: getOAuthRedirectUri,
    webex: webex,
    env: process.env,
    logger: logger,
});

registerUserRoutes(app, {
    isAuthorized: isAuthorized,
    authorized: authorized,
    helpers: helpers,
    getGroupsFromCache: getGroupsFromCache,
    saveConfig: saveConfig,
    logger: logger,
});

registerAdminRoutes(app, {
    isAdmin: isAdmin,
    authorized: authorized,
    webex: webex,
    saveConfig: saveConfig,
    logger: logger,
});

registerJobsRoutes(app, {
    jobs: jobs,
    isAuthorized: isAuthorized,
    helpers: helpers,
    webex: webex,
    translator: translator,
    jobsPipeline: jobsPipeline,
    saveConfig: saveConfig,
    upload: upload,
    sharp: sharp,
    uuid: uuid,
    fs: fs,
    logger: logger,
});

function saveConfig(jsonObject, configFile) {
    const jsonData = JSON.stringify(jsonObject, null, 4);
    fs.writeFileSync(configFile, jsonData, (err) => {
        if (err) {
            throw err;
        } else {
            logger("saveConfig", "Wrote " + configFile);
        }
    })
}

// Retention window for completed jobs (re-exported from lib/helpers.js so
// index.js has a single symbol name callers can reason about). Bump this by
// editing helpers.js if you want a longer / shorter review window.
const COMPLETED_RETENTION_DAYS = helpers.COMPLETED_RETENTION_DAYS;

async function cleanCompletedJobs(jobs) {
    var result = helpers.cleanCompletedJobs(jobs, COMPLETED_RETENTION_DAYS);
    if (!Array.isArray(result.jobs.completed)) {
        logger('cleanCompletedJobs', 'No completed array found – nothing to do.');
    } else {
        logger('cleanCompletedJobs',
            'Removed ' + result.removed + ' job(s) older than ' + COMPLETED_RETENTION_DAYS +
            ' days (cutoff ' + result.cutoff.toISOString().split('T')[0] + '). Remaining: ' +
            result.jobs.completed.length);
    }
    return result.jobs;
}

function isAuthorized(appName, personId) {
    logger("isAuthorized", appName + " " + personId)
    return helpers.isAuthorized(config, botTokens, authorized, appName, personId);
}

function isAdmin(personId) {
    return helpers.isAdmin(authorized, personId);
}

function logger(activeFunction, logLine) {
    var d = new Date();
    console.log(d.toLocaleString() + " " + activeFunction + ": " + logLine);
}

// gracefully shutdown (ctrl-c)
process.on('SIGINT', function () {
    server.close(() => {
        logger("shutdown", config.server.name + ' stopped!')
        process.exit();
    });
})