MeshBot/src/rss/rss.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { DatabaseSync } from 'node:sqlite';
// type-only both ways — no runtime import cycle between src/rss and src/commands
import type { CommandContext } from '../commands/command.service.js';
import { LlmService } from '../llm/llm.service.js';
import { parseFeed } from './rss.feed.js';
import { RssStore } from './rss.store.js';

// ponytail: hardcoded tunables, promote to env if the owner wants them adjustable
const FETCH_TIMEOUT_MS = 15000; // feed fetches — same budget as the opencode notifier's POSTs
const MAX_TITLES = 8; // "up to 8 latest item titles" per the design
const TAG_MAX_TOKENS = 50;
// The LLM server (qwen3.8-27b) emits reasoning_content and those tokens COUNT against max_tokens
// (verified: cap 150 → completion_tokens 150 of which 105 were reasoning, finish_reason 'length').
// Thinking length for this task is unbounded and uncontrollable server-side (/no_think,
// chat_template_kwargs.enable_thinking and thinking_budget are all ignored) — observed 32..1776
// tokens across runs. So the cap must be headroom for reasoning + ~80 content tokens:
// 2500 covers every observed run (worst case ≈42 s at ~60 tok/s, under LLM_TIMEOUT_MS=60 s).
const SUMMARY_MAX_TOKENS = 2500;
// Mesh fragments replies longer than ~190 chars into fire-and-forget packets that recipients
// often never get — each summary line must fit ONE packet on its own. The prompt asks for short
// headered lines; this is the hard backstop if the model ignores it (cut at a word boundary).
const MAX_LINE_CHARS = 185; // per-line one-packet threshold
const LINE_CUT_AT = 180; // where over-long lines get cut, snapped back to a word boundary
const MAX_ITEMS = 5; // cap on summary items sent as separate messages

/**
 * The `#rss` command family: stored RSS sources (SQLite, shared MESH_DB_PATH file) with
 * LLM-generated tags and on-demand headline summaries. Replies are short strings — or an array
 * of lines for multi-message summaries, which CommandService sends one respond() per line.
 */
@Injectable()
export class RssService {
  private readonly logger = new Logger(RssService.name);
  private readonly store: RssStore;

  constructor(
    private readonly llm: LlmService,
    db: DatabaseSync,
  ) {
    this.store = new RssStore(db);
  }

  /** Entry point for the `#rss` prefix — always replies (errors included); `get` may reply with several lines. */
  async handle(ctx: CommandContext): Promise<string | string[]> {
    const words = ctx.text.trim().split(/\s+/);
    switch (words[1]) {
      case undefined: // bare `#rss`
      case 'help':
        return this.help();
      case 'add':
        return this.add(words.slice(2));
      case 'remove':
        return this.remove(words.slice(2).join(' '));
      case 'tags':
        return this.tags();
      case 'sources':
        return this.sources();
      case 'get':
        return this.get((words[2] ?? '').trim());
      default:
        return this.help(); // unknown subcommand → help
    }
  }

  private help(): string {
    return [
      '#rss add <url> <name> - add source with AI tags',
      '#rss remove <name> - delete a source',
      '#rss get <name|#tag> - summarize latest items',
      '#rss tags - all distinct tags',
      '#rss sources - all source names',
    ].join('\n');
  }

  /** `#rss add <url> <name>` — name may contain spaces (everything after the url). */
  private async add(args: string[]): Promise<string> {
    const url = args[0];
    const name = args.slice(1).join(' ').trim();
    if (!url || !name) return 'usage: #rss add <url> <name>';
    let parsedUrl: URL;
    try {
      parsedUrl = new URL(url);
    } catch {
      return 'invalid url';
    }
    if (parsedUrl.protocol !== 'http:' && parsedUrl.protocol !== 'https:')
      return 'invalid url';
    if (this.store.exists(name))
      return `${name} already exists - remove it first`;

    let titles: string[];
    try {
      titles = await this.fetchLatestTitles(url);
    } catch (err) {
      const msg = (err as Error).message;
      // fetchLatestTitles throws `HTTP <status>` on non-2xx, other errors carry their own message
      return msg.startsWith('HTTP')
        ? `feed fetch failed (${msg})`
        : 'not a valid feed';
    }
    if (!titles.length) return 'not a valid feed (no items)';

    // AI tags — best effort: a source is still added when the LLM fails
    let tags: string[] = [];
    try {
      const raw = await this.llm.chat(
        [
          {
            role: 'user',
            content: `Give 2-4 short lowercase tags for these RSS feed item titles. Reply with comma-separated tags only - no # symbols, no explanation:\n${titles.join('\n')}`,
          },
        ],
        { maxTokens: TAG_MAX_TOKENS },
      );
      tags = sanitizeTags(raw);
    } catch (err) {
      this.logger.warn(
        `[rss] tag generation failed for '${name}': ${(err as Error).message}`,
      );
    }

    this.store.add({ name, url, tags });
    return tags.length
      ? `added ${name} (tags: ${tags.join(', ')})`
      : 'added ' + name + ' (tag generation failed - no tags)';
  }

  private remove(name: string): string {
    if (!name) return 'usage: #rss remove <name>';
    return this.store.remove(name)
      ? `removed ${name}`
      : `no source named ${name}`;
  }

  private tags(): string {
    const all = this.store.allTags();
    return all.length ? all.join(', ') : 'no tags yet';
  }

  private sources(): string {
    const names = this.store.list().map((s) => s.name);
    return names.length ? names.join(', ') : 'no sources yet';
  }

  /** `#rss get <name>` or `#rss get #<tag>` — replies with 3-5 separate one-packet lines. */
  private async get(target: string): Promise<string | string[]> {
    if (!target) return 'usage: #rss get <name|#tag>';

    let urls: string[];
    if (target.startsWith('#')) {
      const tag = target.slice(1).toLowerCase();
      if (!tag) return 'usage: #rss get <name|#tag>';
      const matching = this.store.list().filter((s) => s.tags.includes(tag));
      if (!matching.length) return `no sources with tag ${tag}`;
      urls = matching.map((s) => s.url);
    } else {
      const source = this.store.get(target);
      if (!source) return `no source named ${target}`;
      urls = [source.url];
    }

    // one dead feed must not sink the whole summary — collect what answers
    const results = await Promise.allSettled(
      urls.map((u) => this.fetchLatestTitles(u)),
    );
    const titles: string[] = [];
    for (const r of results) {
      if (r.status === 'fulfilled') titles.push(...r.value);
      else
        this.logger.warn(
          `[rss] feed fetch failed: ${(r.reason as Error).message}`,
        );
    }
    if (!titles.length) return 'no items found';

    try {
      const reply = await this.llm.chat(
        [
          {
            role: 'user',
            content: `Summarize these RSS headlines into exactly 3-5 lines, one item per line, each in the format "<short header>: <one or two short sentences>" where the header is at most 20 characters (e.g. "Culture: Museum opens new wing"). No intro, no outro, no numbering - items only:\n${titles.join('\n')}`,
          },
        ],
        { maxTokens: SUMMARY_MAX_TOKENS },
      );
      // each item goes out as its own mesh message — CommandService sends the array line by line;
      // fewer than 3 valid lines is fine (the model is non-deterministic), we send what we got
      return parseSummaryItems(reply).map(fitOneLine);
    } catch (err) {
      this.logger.warn(`[rss] summary failed: ${(err as Error).message}`);
      return 'summary failed';
    }
  }

  /** Fetch a feed and return up to MAX_TITLES latest item titles; throws on non-2xx. */
  private async fetchLatestTitles(url: string): Promise<string[]> {
    const res = await fetch(url, {
      signal: AbortSignal.timeout(FETCH_TIMEOUT_MS),
    });
    if (!res.ok) throw new Error(`HTTP ${res.status}`);
    return parseFeed(await res.text())
      .slice(0, MAX_TITLES)
      .map((i) => i.title);
  }
}

/** Parse an LLM summary into per-line items: drop empty lines and lines without a `:` header separator, cap at MAX_ITEMS. */
export function parseSummaryItems(reply: string): string[] {
  return reply
    .split('\n')
    .map((l) => l.trim())
    .filter((l) => l && l.includes(':'))
    .slice(0, MAX_ITEMS);
}

/** Hard-cut a single summary line at a word boundary near LINE_CUT_AT when it exceeds MAX_LINE_CHARS. */
export function fitOneLine(line: string): string {
  if (line.length <= MAX_LINE_CHARS) return line;
  const slice = line.slice(0, LINE_CUT_AT);
  const at = slice.lastIndexOf(' ');
  return (at > 0 ? slice.slice(0, at) : slice).trimEnd();
}

/** Normalize LLM tag output: lowercase, no #, whitespace → '-', deduped. */
export function sanitizeTags(raw: string): string[] {
  const seen = new Set<string>();
  for (const part of raw.split(',')) {
    const tag = part
      .trim()
      .toLowerCase()
      .replace(/^#+/, '')
      .replace(/\s+/g, '-');
    if (tag) seen.add(tag);
  }
  return [...seen];
}