pulse-signage/src/web/lib/background-tasks/tasks-scheduled/data-source-refresh.js

const TASK = {
  key: 'recurring-data-source-refreshes',
  title: 'Recurring data source refreshes',
  category: 'data-source',
  trigger: 'scheduled recurring task definitions from the database',
  purpose: 'keep API sources and RSS feeds refreshed on their configured intervals.',
  taskType: 'data-source-refresh',
  intervalMs: null
};

const { normalizeIntervalMs } = require('../queue');
const { refreshApiSource, refreshRssFeed } = require('../../data-source-refresh');

function registerRecurringDataSourceRefreshes(options) {
  const pool = options && options.pool;
  const common = options && options.common;
  const backgroundTaskQueue = options && options.backgroundTaskQueue;

  if (!pool || !common || !backgroundTaskQueue) {
    throw new Error('registerRecurringDataSourceRefreshes requires the recurring refresh dependencies.');
  }

  return (async function () {
    const apiSourcesData = await common.fetchApiSourcesData(pool);
    (apiSourcesData.apiSources || []).forEach(function (apiSource) {
      backgroundTaskQueue.registerRecurringTask({
        key: 'api-source-refresh:' + Number(apiSource.id),
        title: 'API source refresh',
        category: TASK.category,
        intervalMs: normalizeIntervalMs(apiSource.update_interval_value, apiSource.update_interval_unit),
        metadata: {
          sourceType: 'api-source',
          sourceId: Number(apiSource.id),
          sourceName: apiSource.name
        },
        run: function () {
          return refreshApiSource(pool, common, apiSource, null);
        }
      });
    });

    const rssFeedsData = await common.fetchRssFeedsData(pool);
    (rssFeedsData.rssFeeds || []).forEach(function (rssFeed) {
      backgroundTaskQueue.registerRecurringTask({
        key: 'rss-feed-refresh:' + Number(rssFeed.id),
        title: 'RSS feed refresh',
        category: TASK.category,
        intervalMs: normalizeIntervalMs(rssFeed.update_interval_value, rssFeed.update_interval_unit),
        metadata: {
          sourceType: 'rss-feed',
          sourceId: Number(rssFeed.id),
          sourceName: rssFeed.name
        },
        run: function () {
          return refreshRssFeed(pool, common, rssFeed.id, rssFeed.feed_url, rssFeed.item_limit, null);
        }
      });
    });
  })();
}

function createDataSourceTaskService(options) {
  const pool = options && options.pool;
  const common = options && options.common;
  const backgroundTaskQueue = options && options.backgroundTaskQueue;

  if (!pool || !common || !backgroundTaskQueue) {
    throw new Error('createDataSourceTaskService requires the data source task dependencies.');
  }

  function formatRecurringKey(sourceType, id) {
    return sourceType + '-refresh:' + Number(id);
  }

  function buildRecurringTitle(sourceType) {
    return sourceType === 'rss-feed' ? 'RSS feed refresh' : 'API source refresh';
  }

  function registerRecurringRefresh(sourceType, id, name, intervalValue, intervalUnit, run) {
    backgroundTaskQueue.registerRecurringTask({
      key: formatRecurringKey(sourceType, id),
      title: buildRecurringTitle(sourceType),
      category: 'data-source',
      intervalMs: normalizeIntervalMs(intervalValue, intervalUnit),
      metadata: {
        sourceType: sourceType,
        sourceId: Number(id),
        sourceName: name
      },
      run: run
    });
  }

  function removeRecurringRefresh(sourceType, id) {
    backgroundTaskQueue.removeRecurringTask(formatRecurringKey(sourceType, id));
  }

  async function getTaskStatusById(taskId) {
    if (!backgroundTaskQueue || typeof backgroundTaskQueue.getTaskById !== 'function') {
      return null;
    }

    const task = await backgroundTaskQueue.getTaskById(taskId);
    if (!task) {
      return null;
    }

    return {
      id: task.id,
      key: task.key,
      status: task.status,
      finishedAt: task.finishedAt || '',
      errorMessage: task.errorMessage || ''
    };
  }

  async function refreshApiSourceInBackground(apiSourceId, actorId) {
    return refreshApiSource(pool, common, apiSourceId, actorId);
  }

  async function refreshRssFeedInBackground(rssFeedId, feedUrl, itemLimit, actorId) {
    return refreshRssFeed(pool, common, rssFeedId, feedUrl, itemLimit, actorId);
  }

  return {
    formatRecurringKey: formatRecurringKey,
    buildRecurringTitle: buildRecurringTitle,
    registerRecurringRefresh: registerRecurringRefresh,
    removeRecurringRefresh: removeRecurringRefresh,
    getTaskStatusById: getTaskStatusById,
    refreshApiSourceInBackground: refreshApiSourceInBackground,
    refreshRssFeedInBackground: refreshRssFeedInBackground
  };
}

module.exports = {
  registerRecurringDataSourceRefreshes: registerRecurringDataSourceRefreshes,
  createDataSourceTaskService: createDataSourceTaskService
};