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 { 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 - add source with AI tags', '#rss remove - delete a source', '#rss get - summarize latest items', '#rss tags - all distinct tags', '#rss sources - all source names', ].join('\n'); } /** `#rss add ` — name may contain spaces (everything after the url). */ private async add(args: string[]): Promise { const url = args[0]; const name = args.slice(1).join(' ').trim(); if (!url || !name) return 'usage: #rss add '; 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 ` 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 '; 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 ` or `#rss get #` — replies with 3-5 separate one-packet lines. */ private async get(target: string): Promise { if (!target) return 'usage: #rss get '; let urls: string[]; if (target.startsWith('#')) { const tag = target.slice(1).toLowerCase(); if (!tag) return 'usage: #rss get '; 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 ": " 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 { 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(); for (const part of raw.split(',')) { const tag = part .trim() .toLowerCase() .replace(/^#+/, '') .replace(/\s+/g, '-'); if (tag) seen.add(tag); } return [...seen]; }