Build an RSS Feed Aggregation Service

// src/feeds/aggregator.ts — RSS/Atom feed aggregation with dedup and smart digest
import { pool } from "../db";
import { Redis } from "ioredis";
import { parseStringPromise } from "xml2js";
import { createHash, randomBytes } from "node:crypto";

const redis = new Redis(process.env.REDIS_URL!);

interface Feed {
  id: string;
  title: string;
  url: string;
  siteUrl: string;
  format: "rss" | "atom" | "json";
  category: string;
  tags: string[];
  refreshInterval: number;     // minutes
  lastFetchedAt: string | null;
  lastItemAt: string | null;
  errorCount: number;
  status: "active" | "error" | "paused";
}

interface FeedItem {
  id: string;
  feedId: string;
  title: string;
  url: string;
  content: string;
  summary: string;
  author: string;
  publishedAt: string;
  categories: string[];
  relevanceScore: number;
  contentHash: string;
  isDuplicate: boolean;
  isRead: boolean;
  isStarred: boolean;
}

// Import feeds from OPML file
export async function importOPML(opmlXml: string, userId: string): Promise<Feed[]> {
  const parsed = await parseStringPromise(opmlXml, { explicitArray: false });
  const outlines = extractOutlines(parsed.opml.body.outline);
  const feeds: Feed[] = [];

  for (const outline of outlines) {
    if (!outline.xmlUrl) continue;
    const id = `feed-${randomBytes(4).toString("hex")}`;
    const feed: Feed = {
      id, title: outline.title || outline.text || "Untitled",
      url: outline.xmlUrl, siteUrl: outline.htmlUrl || "",
      format: "rss", category: outline.category || "Uncategorized",
      tags: [], refreshInterval: 30,
      lastFetchedAt: null, lastItemAt: null,
      errorCount: 0, status: "active",
    };

    await pool.query(
      `INSERT INTO feeds (id, title, url, site_url, category, refresh_interval, status, user_id, created_at)
       VALUES ($1, $2, $3, $4, $5, $6, 'active', $7, NOW()) ON CONFLICT (url, user_id) DO NOTHING`,
      [id, feed.title, feed.url, feed.siteUrl, feed.category, feed.refreshInterval, userId]
    );
    feeds.push(feed);
  }

  return feeds;
}

// Fetch and process a single feed
export async function fetchFeed(feed: Feed): Promise<FeedItem[]> {
  try {
    const response = await fetch(feed.url, {
      headers: { "User-Agent": "FeedAggregator/1.0", "Accept": "application/rss+xml, application/atom+xml, application/xml" },
      signal: AbortSignal.timeout(10000),
    });

    const text = await response.text();
    const items = await parseFeed(text);
    const newItems: FeedItem[] = [];

    for (const item of items) {
      const contentHash = createHash("sha256")
        .update((item.title || "") + (item.link || ""))
        .digest("hex").slice(0, 16);

      // Skip if already processed
      const exists = await redis.get(`feed:item:${contentHash}`);
      if (exists) continue;
      await redis.setex(`feed:item:${contentHash}`, 86400 * 7, "1");

      // Deduplication across feeds
      const isDuplicate = await checkDuplicate(item.title || "");

      // Relevance scoring
      const relevanceScore = scoreRelevance(item, feed);

      const feedItem: FeedItem = {
        id: `fi-${randomBytes(6).toString("hex")}`,
        feedId: feed.id,
        title: item.title || "Untitled",
        url: item.link || "",
        content: item["content:encoded"] || item.description || "",
        summary: (item.description || "").replace(/<[^>]+>/g, "").slice(0, 300),
        author: item.author || item["dc:creator"] || "",
        publishedAt: item.pubDate ? new Date(item.pubDate).toISOString() : new Date().toISOString(),
        categories: extractCategories(item),
        relevanceScore,
        contentHash,
        isDuplicate,
        isRead: false,
        isStarred: false,
      };

      await pool.query(
        `INSERT INTO feed_items (id, feed_id, title, url, content, summary, author, published_at, categories, relevance_score, content_hash, is_duplicate, created_at)
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, NOW())`,
        [feedItem.id, feed.id, feedItem.title, feedItem.url, feedItem.content,
         feedItem.summary, feedItem.author, feedItem.publishedAt,
         JSON.stringify(feedItem.categories), feedItem.relevanceScore,
         feedItem.contentHash, feedItem.isDuplicate]
      );

      newItems.push(feedItem);
    }

    // Update feed status
    await pool.query(
      "UPDATE feeds SET last_fetched_at = NOW(), error_count = 0, status = 'active' WHERE id = $1",
      [feed.id]
    );

    return newItems;
  } catch (error: any) {
    await pool.query(
      "UPDATE feeds SET error_count = error_count + 1, status = CASE WHEN error_count >= 5 THEN 'error' ELSE status END WHERE id = $1",
      [feed.id]
    );
    return [];
  }
}

// Generate daily digest — top N most relevant items
export async function generateDigest(userId: string, maxItems: number = 20): Promise<{
  date: string;
  items: FeedItem[];
  byCategory: Record<string, FeedItem[]>;
}> {
  const { rows } = await pool.query(
    `SELECT fi.* FROM feed_items fi
     JOIN feeds f ON fi.feed_id = f.id
     WHERE f.user_id = $1
       AND fi.is_duplicate = false
       AND fi.published_at > NOW() - INTERVAL '24 hours'
     ORDER BY fi.relevance_score DESC
     LIMIT $2`,
    [userId, maxItems]
  );

  const byCategory: Record<string, FeedItem[]> = {};
  for (const item of rows) {
    const cats = JSON.parse(item.categories);
    const cat = cats[0] || "General";
    if (!byCategory[cat]) byCategory[cat] = [];
    byCategory[cat].push(item);
  }

  return { date: new Date().toISOString().slice(0, 10), items: rows, byCategory };
}

function scoreRelevance(item: any, feed: Feed): number {
  let score = 0.5;
  const title = (item.title || "").toLowerCase();
  // Boost for trending keywords
  const trending = ["ai", "llm", "agent", "startup", "funding", "launch"];
  for (const kw of trending) {
    if (title.includes(kw)) score += 0.1;
  }
  score = Math.min(1, score);
  return score;
}

async function checkDuplicate(title: string): Promise<boolean> {
  const { rows } = await pool.query(
    `SELECT id FROM feed_items WHERE similarity(title, $1) > 0.6 AND created_at > NOW() - INTERVAL '48 hours' LIMIT 1`,
    [title]
  );
  return rows.length > 0;
}

async function parseFeed(xml: string): Promise<any[]> {
  const result = await parseStringPromise(xml, { explicitArray: false });
  if (result.rss) return [].concat(result.rss.channel.item || []);
  if (result.feed) return [].concat(result.feed.entry || []).map((e: any) => ({
    title: e.title?.["_"] || e.title,
    link: e.link?.$?.href || e.link,
    description: e.summary?.["_"] || e.summary || "",
    pubDate: e.published || e.updated,
    author: e.author?.name || "",
  }));
  return [];
}

function extractOutlines(outline: any): any[] {
  if (!outline) return [];
  const items = Array.isArray(outline) ? outline : [outline];
  const results: any[] = [];
  for (const item of items) {
    if (item.xmlUrl) results.push(item);
    if (item.outline) results.push(...extractOutlines(item.outline));
  }
  return results;
}

function extractCategories(item: any): string[] {
  if (!item.category) return [];
  const cats = Array.isArray(item.category) ? item.category : [item.category];
  return cats.map((c: any) => (typeof c === "string" ? c : c._ || c.$ || "")).filter(Boolean);
}