collabSupport/services/cdrFeedQueue.js

// services/cdrFeedQueue.js
// Serialize cdr_feed HTTP calls — Webex allows ~1 request/minute per token.

import { logger } from '../utils/logger.js';

const LOG_SCOPE = 'cdr:queue';

let testCooldownOverride = null;

function cooldownMs() {
  if (testCooldownOverride != null) return testCooldownOverride;
  const v = Number(process.env.CDR_FEED_COOLDOWN_MS);
  return Number.isFinite(v) && v > 0 ? v : 65_000;
}

export const CDR_FEED_COOLDOWN_MS = 65_000;

let chain = Promise.resolve();
let lastStartMs = 0;
let pendingCount = 0;

function sleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

function estimateRunAtMs(position) {
  const now = Date.now();
  const firstSlot = lastStartMs > 0
    ? Math.max(now, lastStartMs + cooldownMs())
    : now;
  if (position <= 0) return firstSlot;
  return firstSlot + position * cooldownMs();
}

/**
 * Run a cdr_feed fetch when the global cooldown allows. Queues concurrent callers.
 *
 * @template T
 * @param {() => Promise<T>} fn
 * @param {{ onQueued?: (info: { runAt: Date, waitMs: number, position: number }) => void|Promise<void>, label?: string }} [opts]
 * @returns {Promise<T>}
 */
export function enqueueCdrFeedFetch(fn, opts = {}) {
  const { onQueued, label } = opts;
  const now = Date.now();
  const position = pendingCount;
  pendingCount += 1;

  const inCooldown = lastStartMs > 0 && now < lastStartMs + cooldownMs();
  if ((position > 0 || inCooldown) && onQueued) {
    const runAtMs = estimateRunAtMs(position);
    Promise.resolve(onQueued({
      runAt: new Date(runAtMs),
      waitMs: Math.max(0, runAtMs - now),
      position: position + 1,
    })).catch((err) => {
      logger(LOG_SCOPE, `onQueued callback failed${label ? ` (${label})` : ''}: ${err.message}`, 'warn');
    });
  }

  const job = chain.then(async () => {
    const waitMs = Math.max(0, (lastStartMs > 0 ? lastStartMs + cooldownMs() : 0) - Date.now());
    if (waitMs > 0) {
      logger(
        LOG_SCOPE,
        `Waiting ${Math.ceil(waitMs / 1000)}s for cdr_feed cooldown${label ? ` (${label})` : ''}`,
        'info',
      );
      await sleep(waitMs);
    }
    lastStartMs = Date.now();
    return fn();
  });

  chain = job.catch(() => {});
  return job.finally(() => {
    pendingCount = Math.max(0, pendingCount - 1);
  });
}

/** @internal test helper */
export function _resetCdrFeedQueueForTests() {
  chain = Promise.resolve();
  lastStartMs = 0;
  pendingCount = 0;
  testCooldownOverride = null;
}

/** @internal test helper */
export function _setCdrFeedCooldownForTests(ms) {
  testCooldownOverride = ms;
}