godcrm/backend/services/content-aggregator/aggregator.js

// backend/services/content-aggregator/aggregator.js
// Main content aggregation orchestrator

import { log, getCRMAuthToken } from './helpers.js';
import { fetchHackerNews } from './hackernews.js';
import { fetchRSSFeeds } from './rss.js';
import { fetchReddit } from './reddit.js';
import { getExistingTitles, storeInCRM } from './crm-storage.js';

/**
 * Main content aggregation function
 * Fetches from all specified sources, deduplicates, stores in CRM
 *
 * @param {string[]} sources - Sources to fetch from: 'hackernews', 'rss', 'reddit'
 * @returns {Promise<{
 *   total_fetched: number,
 *   total_new: number,
 *   total_stored: number,
 *   total_duplicates: number,
 *   by_source: object,
 *   errors: string[],
 *   items: Array
 * }>}
 */
export async function aggregateContent(sources = ['hackernews', 'rss', 'reddit']) {
  log.info({ sources }, 'Starting content aggregation...');

  const startTime = Date.now();
  const errors = [];
  const allItems = [];

  // Fetch from all sources in parallel
  const fetchPromises = [];

  if (sources.includes('hackernews')) {
    fetchPromises.push(
      fetchHackerNews(10).then(items => {
        allItems.push(...items);
        return { source: 'hackernews', count: items.length };
      }).catch(err => {
        errors.push(`hackernews: ${err.message}`);
        return { source: 'hackernews', count: 0 };
      })
    );
  }

  if (sources.includes('rss')) {
    fetchPromises.push(
      fetchRSSFeeds().then(items => {
        allItems.push(...items);
        return { source: 'rss', count: items.length };
      }).catch(err => {
        errors.push(`rss: ${err.message}`);
        return { source: 'rss', count: 0 };
      })
    );
  }

  if (sources.includes('reddit')) {
    fetchPromises.push(
      fetchReddit().then(items => {
        allItems.push(...items);
        return { source: 'reddit', count: items.length };
      }).catch(err => {
        errors.push(`reddit: ${err.message}`);
        return { source: 'reddit', count: 0 };
      })
    );
  }

  const sourceCounts = await Promise.all(fetchPromises);
  const bySource = {};
  for (const sc of sourceCounts) {
    bySource[sc.source] = { fetched: sc.count, stored: 0 };
  }

  log.info({ totalFetched: allItems.length }, 'All sources fetched, starting dedup...');

  // Get auth token for CRM
  const token = await getCRMAuthToken();

  // Get existing titles for deduplication
  const existingTitles = await getExistingTitles(token);

  // Also deduplicate within the batch itself
  const seenTitles = new Set();
  const newItems = [];

  for (const item of allItems) {
    const titleKey = (item.title || '').toLowerCase().trim();

    if (!titleKey) continue;
    if (existingTitles.has(titleKey)) continue;
    if (seenTitles.has(titleKey)) continue;

    seenTitles.add(titleKey);
    newItems.push(item);
  }

  const totalDuplicates = allItems.length - newItems.length;
  log.info({ newItems: newItems.length, duplicates: totalDuplicates }, 'Dedup complete');

  // Store new items in CRM (sequentially to avoid overwhelming the API)
  let totalStored = 0;

  for (const item of newItems) {
    const result = await storeInCRM(item, token);
    if (result.success) {
      totalStored++;
      item.crm_id = result.id;
      if (bySource[item.source]) {
        bySource[item.source].stored++;
      }
    } else {
      errors.push(`store(${item.source}/${item.title?.slice(0, 40)}): ${result.error}`);
    }
  }

  const elapsed = Date.now() - startTime;

  const summary = {
    total_fetched: allItems.length,
    total_new: newItems.length,
    total_stored: totalStored,
    total_duplicates: totalDuplicates,
    elapsed_ms: elapsed,
    by_source: bySource,
    errors: errors.length > 0 ? errors : undefined,
    items: newItems.map(item => ({
      title: item.title,
      source: item.source,
      url: item.url,
      score: item.score || null,
      crm_id: item.crm_id || null,
    })),
  };

  log.info({
    totalFetched: allItems.length,
    totalNew: newItems.length,
    totalStored,
    totalDuplicates,
    elapsed,
  }, 'Content aggregation complete');

  return summary;
}