Add #rss command family with LLM tags and summaries (v0.20.0)
CI / build + unit (pull_request) Successful in 1m15s

#rss subcommands on the command channel: help / add <url> <name> /
remove <name> / tags / sources / get <name|#tag>. Sources are stored in
a new rss_sources table in the existing MESH_DB_PATH file (second
node:sqlite connection, case-insensitive names); feeds parsed with
fast-xml-parser (RSS 2.0 + Atom); AI tags and headline summaries via
injected LlmService.chat() with bounded maxTokens (50/200).

CommandService gains prefix registration so one handler owns the whole
#rss family while #ping stays an exact match; wired in CommandModule's
factory — still no constructor-level cycle, no forwardRef. One dead feed
does not sink a tag summary.
This commit is contained in:
2026-09-25 18:44:06 +00:00
parent 023cd3eef3
commit eda1cbd125
15 changed files with 993 additions and 14 deletions
+2
View File
@@ -3,6 +3,7 @@ import { BatteryAlertModule } from './battery/battery.module.js';
import { CommandModule } from './commands/command.module.js';
import { LlmModule } from './llm/llm.module.js';
import { MeshModule } from './mesh/mesh.module.js';
import { RssModule } from './rss/rss.module.js';
import { WeatherModule } from './weather/weather.module.js';
@Module({
@@ -12,6 +13,7 @@ import { WeatherModule } from './weather/weather.module.js';
BatteryAlertModule,
CommandModule,
LlmModule,
RssModule,
],
})
export class AppModule {}
+6 -4
View File
@@ -1,20 +1,22 @@
import { Module } from '@nestjs/common';
import { MeshModule } from '../mesh/mesh.module.js';
import { MeshService } from '../mesh/mesh.service.js';
import { RssModule } from '../rss/rss.module.js';
import { RssService } from '../rss/rss.service.js';
import { CommandService } from './command.service.js';
@Module({
imports: [MeshModule], // one-way only — the reverse edge is wired in the factory below, not via DI
imports: [MeshModule, RssModule], // one-way only — the MeshService reverse edge is wired in the factory below, not via DI
providers: [
{
provide: CommandService,
useFactory: (mesh: MeshService) => {
const commands = new CommandService(mesh);
useFactory: (mesh: MeshService, rss: RssService) => {
const commands = new CommandService(mesh, rss);
// break the cycle: MeshService can't inject us (its own factory would deadlock on this edge), so attach ourselves once we exist — mesh is guaranteed created here because it's our injected dependency
mesh.attachCommands(commands);
return commands;
},
inject: [MeshService],
inject: [MeshService, RssService],
},
],
exports: [CommandService],
+73 -2
View File
@@ -1,5 +1,10 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { BROADCAST_ADDR } from '../mesh/codec.js';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { LlmService } from '../llm/llm.service.js';
import { RssService } from '../rss/rss.service.js';
import {
startBridge,
textPacketFrame,
@@ -7,6 +12,7 @@ import {
outboundPackets,
KAMIL_NUM,
} from '../mesh/fake-bridge.js';
import { BROADCAST_ADDR } from '../mesh/codec.js';
import type { MeshService } from '../mesh/mesh.service.js';
import { MeshService as RealMeshService } from '../mesh/mesh.service.js';
import { CommandService } from './command.service.js';
@@ -134,6 +140,71 @@ describe('CommandService.onRx', () => {
});
});
describe('#rss prefix dispatch', () => {
let dir: string | undefined;
const makeRss = (): RssService => {
dir = mkdtempSync(join(tmpdir(), 'mesh-bot-cmd-rss-'));
return new RssService(
{ chat: vi.fn() } as unknown as LlmService,
new DatabaseSync(join(dir, 'test.db')),
);
};
afterEach(() => {
if (dir) rmSync(dir, { recursive: true, force: true });
dir = undefined;
});
it('routes #rss subcommands through respond() like any command reply', async () => {
const mesh = {
sendText: vi.fn(),
respond: vi.fn().mockResolvedValue({ id: 1 }),
} as unknown as MeshService;
const svc = new CommandService(mesh, makeRss());
await svc.onRx(2, '#rss sources', 42, 9);
expect(mesh.respond).toHaveBeenCalledWith(9, 'no sources yet');
expect(mesh.sendText).not.toHaveBeenCalled();
});
it('routes bare #rss to the help text via respond()', async () => {
const mesh = {
sendText: vi.fn(),
respond: vi.fn().mockResolvedValue({ id: 1 }),
} as unknown as MeshService;
const svc = new CommandService(mesh, makeRss());
await svc.onRx(2, '#rss', 42, 9);
expect(mesh.respond).toHaveBeenCalledTimes(1);
expect(String(mesh.respond.mock.calls[0][1])).toContain(
'#rss add <url> <name>',
);
});
it('leaves exact commands untouched and does not partial-match prefix names', async () => {
const mesh = {
sendText: vi.fn(),
respond: vi.fn().mockResolvedValue({ id: 1 }),
} as unknown as MeshService;
const svc = new CommandService(mesh, makeRss());
await svc.onRx(2, '#ping', 42, 3);
expect(mesh.respond).toHaveBeenCalledWith(3, '#pong!');
await svc.onRx(2, '#rssfoo', 42, 4); // starts with 'rss' but is not the family — no partial prefix match
expect(mesh.respond).toHaveBeenCalledTimes(1);
});
it('does nothing for #rss when no RssService was wired (legacy construction)', async () => {
const [svc, mesh] = makeService(); // rss omitted
await svc.onRx(2, '#rss sources', 42, 9);
expect(mesh.respond).not.toHaveBeenCalled();
expect(mesh.sendText).not.toHaveBeenCalled();
});
});
describe('MeshService wiring (event-driven trigger)', () => {
it('answers a #ping DM on ch2 with a directed #pong! back to the sender, same channel, threaded quote', async () => {
const bridge = await startBridge();
+26 -4
View File
@@ -1,7 +1,8 @@
import { Injectable, Logger } from '@nestjs/common';
import { BROADCAST_ADDR } from '../mesh/codec.js';
// type-only: the instance comes from DI (command.module.ts factory) or direct construction in tests — no runtime import cycle with src/mesh
// type-only: the instances come from DI (command.module.ts factory) or direct construction in tests — no runtime import cycles with src/mesh / src/rss
import type { MeshService } from '../mesh/mesh.service.js';
import type { RssService } from '../rss/rss.service.js';
/** Context handed to a command handler: the trimmed text plus sender/packet id when present. */
export interface CommandContext {
@@ -28,10 +29,16 @@ export class CommandService {
);
// single registration point: the constructor registers every built-in command — add the next one there (or via register())
private readonly commands = new Map<string, CommandHandler>();
// prefix handlers own a whole family (`#rss`, `#rss add ...`) instead of one exact name
private readonly prefixes = new Map<string, CommandHandler>();
/** mesh is optional so tests can construct without a bridge; production DI always passes one. */
constructor(private readonly mesh?: MeshService) {
constructor(
private readonly mesh?: MeshService,
rss?: RssService,
) {
this.register('ping', () => '#pong!');
if (rss) this.registerPrefix('rss', (ctx) => rss.handle(ctx));
}
/** Register a command handler (names are case-sensitive; the leading `#` is stripped before lookup). */
@@ -39,9 +46,14 @@ export class CommandService {
this.commands.set(name, fn);
}
/** Register a prefix handler: owns bare `#<name>` and everything starting with `#<name> `. Exact matches win over prefixes. */
registerPrefix(name: string, fn: CommandHandler): void {
this.prefixes.set(name, fn);
}
/**
* Event-driven entry point called by MeshService for every decoded rx text.
* No-op unless on COMMAND_CHANNEL with an exact `#command` match (case-sensitive).
* No-op unless on COMMAND_CHANNEL with a case-sensitive `#command` match — exact name or registered prefix family (`#rss ...`).
* Replies go through MeshService.respond(id, reply): auto-routed to the original message's
* destination and channel as a threaded quote (replyId = id). Only when the trigger has no packet
* id does it fall back to legacy sendText routing on COMMAND_CHANNEL. Never throws — a failed
@@ -56,7 +68,17 @@ export class CommandService {
if (channel !== this.commandChannel) return;
const trimmed = text.trim();
if (!trimmed.startsWith('#')) return; // only `#command` texts trigger handlers
const handler = this.commands.get(trimmed.slice(1));
const name = trimmed.slice(1);
let handler = this.commands.get(name);
if (!handler) {
for (const [prefix, fn] of this.prefixes) {
// a prefix owns bare `#<prefix>` and everything starting with `#<prefix> ` — e.g. 'rss' → #rss / #rss add ...
if (name === prefix || name.startsWith(prefix + ' ')) {
handler = fn;
break;
}
}
}
if (!handler) return; // exact, case-sensitive match — no partials
try {
const reply = await handler({ text: trimmed, from, id });
+64
View File
@@ -0,0 +1,64 @@
import { describe, expect, it } from 'vitest';
import { parseFeed } from './rss.feed.js';
const RSS2 = `<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<title>Some Feed</title>
<item>
<title>First headline</title>
<link>https://example.com/first</link>
</item>
<item>
<title><![CDATA[Second: with CDATA]]></title>
<description>ignored</description>
</item>
<item>
<title></title>
<link>https://example.com/empty-title</link>
</item>
</channel>
</rss>`;
const ATOM = `<?xml version="1.0" encoding="utf-8"?>
<feed xmlns="http://www.w3.org/2005/Atom">
<title>An Atom feed</title>
<entry>
<title>Only entry</title>
<link rel="alternate" href="https://example.com/atom"/>
</entry>
</feed>`;
describe('parseFeed', () => {
it('parses RSS 2.0 items in document order, keeping links where present and skipping empty titles', () => {
expect(parseFeed(RSS2)).toEqual([
{ title: 'First headline', link: 'https://example.com/first' },
{ title: 'Second: with CDATA' },
]);
});
it('parses Atom entries (single entry object, link as href attribute)', () => {
expect(parseFeed(ATOM)).toEqual([
{ title: 'Only entry', link: 'https://example.com/atom' },
]);
});
it('returns [] for malformed XML', () => {
expect(parseFeed('this is not xml at all')).toEqual([]);
expect(parseFeed('<rss><channel><item><title>broken')).toEqual([]);
});
it('returns [] when there are no items or none with a usable title', () => {
expect(
parseFeed('<rss version="2.0"><channel><title>x</title></channel></rss>'),
).toEqual([]);
expect(
parseFeed('<feed xmlns="http://www.w3.org/2005/Atom"></feed>'),
).toEqual([]);
expect(
parseFeed(
'<rss version="2.0"><channel><item><title> </title></item></channel></rss>',
),
).toEqual([]);
});
});
+60
View File
@@ -0,0 +1,60 @@
import { XMLParser } from 'fast-xml-parser';
/** One feed item: title always present (non-empty), link where the feed provides one. */
export interface FeedItem {
title: string;
link?: string;
}
const parser = new XMLParser({ ignoreAttributes: false });
function asText(value: unknown): string | undefined {
if (typeof value === 'string') return value.trim() || undefined;
// element with attributes/CDATA parses to an object holding its text under #text
const t = (value as Record<string, unknown>)?.['#text'];
if (typeof t === 'string' && t.trim()) return t.trim();
return undefined;
}
function asLink(value: unknown): string | undefined {
const fromText = asText(value);
if (fromText) return fromText; // RSS 2.0 <link>url</link>
// Atom <link rel="alternate" href="..."/> — the URL is an attribute
const href = (value as Record<string, unknown>)?.['@_href'];
if (typeof href === 'string' && href.trim()) return href.trim();
return undefined;
}
function collect(value: unknown): Record<string, unknown>[] {
// a single item/entry parses to an object, several to an array
if (!value) return [];
const arr = Array.isArray(value) ? value : [value];
return arr.filter(
(v): v is Record<string, unknown> => !!v && typeof v === 'object',
);
}
/**
* Parse RSS 2.0 (`<item><title>/<link>`) or Atom (`<entry><title>`) XML into items with
* non-empty titles, in document order (feeds are newest-first). Returns [] for malformed
* input or feeds without usable items — that empty result IS the "not a valid feed" signal.
*/
export function parseFeed(xml: string): FeedItem[] {
let doc: unknown;
try {
doc = parser.parse(xml);
} catch {
return []; // fast-xml-parser is lenient, but a hard failure means no items either
}
const root = doc as Record<string, any>;
// coalesce the RAW values first — collect() always returns an array, so `collect(a) ?? collect(b)` could never fall through to b
const rawItems = collect(root?.rss?.channel?.item ?? root?.feed?.entry);
const items: FeedItem[] = [];
for (const item of rawItems) {
const title = asText(item?.title);
if (!title) continue; // skip entries without a usable title
const link = asLink(item.link);
items.push(link ? { title, link } : { title });
}
return items;
}
+28
View File
@@ -0,0 +1,28 @@
import { Module } from '@nestjs/common';
import { mkdirSync } from 'node:fs';
import { dirname } from 'node:path';
import { DatabaseSync } from 'node:sqlite';
import { LlmModule } from '../llm/llm.module.js';
import { LlmService } from '../llm/llm.service.js';
import { RssService } from './rss.service.js';
// The rss_sources table lives in the SAME file as the mesh message history (MESH_DB_PATH) — a second
// connection is fine, SQLite serializes writes. The service initializes its own table on construction.
function openRssDb(): DatabaseSync {
const path = process.env.MESH_DB_PATH ?? './data/mesh-bot.db';
mkdirSync(dirname(path), { recursive: true });
return new DatabaseSync(path);
}
@Module({
imports: [LlmModule], // one-way — LlmService has no dependencies of its own
providers: [
{
provide: RssService,
useFactory: (llm: LlmService) => new RssService(llm, openRssDb()),
inject: [LlmService],
},
],
exports: [RssService],
})
export class RssModule {}
+206
View File
@@ -0,0 +1,206 @@
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { LlmService } from '../llm/llm.service.js';
import { RssStore } from './rss.store.js';
import { RssService } from './rss.service.js';
const FEED_A = `<rss version="2.0"><channel>
<item><title>Alpha headline</title></item>
<item><title>Beta headline</title></item>
</channel></rss>`;
const FEED_B = `<rss version="2.0"><channel>
<item><title>Gamma headline</title></item>
</channel></rss>`;
function feedResponse(xml: string): Response {
return { ok: true, status: 200, text: async () => xml } as Response;
}
describe('RssService.handle', () => {
let dir: string;
let dbFile: string;
let llm: { chat: ReturnType<typeof vi.fn> };
let svc: RssService;
let fetchMock: ReturnType<typeof vi.fn>;
beforeEach(() => {
dir = mkdtempSync(join(tmpdir(), 'mesh-bot-rss-svc-'));
dbFile = join(dir, 'test.db');
llm = { chat: vi.fn() };
svc = new RssService(
llm as unknown as LlmService,
new DatabaseSync(dbFile),
);
fetchMock = vi.fn();
vi.stubGlobal('fetch', fetchMock);
});
afterEach(() => {
vi.unstubAllGlobals();
rmSync(dir, { recursive: true, force: true });
});
const ctx = (text: string) => ({ text });
// second connection — the same shape production uses for a separate table in one file
const stored = () => new RssStore(new DatabaseSync(dbFile));
it('replies with usage on bare #rss and on unknown subcommands', async () => {
const help = await svc.handle(ctx('#rss'));
expect(help).toContain('#rss add <url> <name>');
expect(help).toContain('#rss get <name|#tag>');
expect(await svc.handle(ctx('#rss frobnicate'))).toBe(help);
});
it('add: validates the feed, stores source with LLM tags, replies added (tags: ...)', async () => {
fetchMock.mockResolvedValueOnce(feedResponse(FEED_A));
llm.chat.mockResolvedValueOnce('#AI, News');
const reply = await svc.handle(
ctx('#rss add https://a.example/rss my feed'),
);
expect(reply).toBe('added my feed (tags: ai, news)');
expect(stored().get('my feed')).toMatchObject({
url: 'https://a.example/rss',
tags: ['ai', 'news'],
});
const [messages, opts] = llm.chat.mock.calls[0];
expect(opts).toEqual({ maxTokens: 50 });
expect(messages[0].content).toContain('Alpha headline');
});
it('add: LLM failure still stores the source with empty tags and says so', async () => {
fetchMock.mockResolvedValueOnce(feedResponse(FEED_A));
llm.chat.mockRejectedValueOnce(new Error('llm down'));
const reply = await svc.handle(
ctx('#rss add https://a.example/rss myfeed'),
);
expect(reply).toBe('added myfeed (tag generation failed - no tags)');
expect(stored().get('myfeed')?.tags).toEqual([]);
});
it('add: rejects invalid urls without fetching or storing', async () => {
expect(await svc.handle(ctx('#rss add not a url x'))).toBe('invalid url');
expect(await svc.handle(ctx('#rss add ftp://a.example/rss x'))).toBe(
'invalid url',
);
expect(fetchMock).not.toHaveBeenCalled();
expect(stored().list()).toEqual([]);
});
it('add: rejects duplicate names case-insensitively without fetching', async () => {
stored().add({ name: 'MyFeed', url: 'https://old.example/rss', tags: [] });
const reply = await svc.handle(
ctx('#rss add https://a.example/rss myfeed'),
);
expect(reply).toBe('myfeed already exists - remove it first');
expect(fetchMock).not.toHaveBeenCalled();
expect(stored().get('myfeed')?.url).toBe('https://old.example/rss'); // untouched
});
it('add: reports non-2xx and unparseable feeds, stores nothing', async () => {
fetchMock.mockResolvedValueOnce({ ok: false, status: 404 } as Response);
expect(await svc.handle(ctx('#rss add https://a.example/rss x'))).toBe(
'feed fetch failed (HTTP 404)',
);
fetchMock.mockResolvedValueOnce(feedResponse('this is not xml'));
expect(await svc.handle(ctx('#rss add https://b.example/rss y'))).toBe(
'not a valid feed (no items)',
);
expect(stored().list()).toEqual([]);
});
it('remove: deletes case-insensitively, reports missing names', async () => {
stored().add({ name: 'Gone', url: 'https://g.example/rss', tags: [] });
expect(await svc.handle(ctx('#rss remove gone'))).toBe('removed gone');
expect(stored().list()).toEqual([]);
expect(await svc.handle(ctx('#rss remove nope'))).toBe(
'no source named nope',
);
});
it('tags: no tags yet when empty, distinct sorted list otherwise', async () => {
expect(await svc.handle(ctx('#rss tags'))).toBe('no tags yet');
stored().add({
name: 'A',
url: 'https://a.example/rss',
tags: ['zeta', 'alpha'],
});
stored().add({
name: 'B',
url: 'https://b.example/rss',
tags: ['beta', 'alpha'],
});
expect(await svc.handle(ctx('#rss tags'))).toBe('alpha, beta, zeta');
});
it('sources: no sources yet when empty, names otherwise', async () => {
expect(await svc.handle(ctx('#rss sources'))).toBe('no sources yet');
stored().add({ name: 'Bee', url: 'https://b.example/rss', tags: [] });
stored().add({ name: 'Aye', url: 'https://a.example/rss', tags: [] });
expect(await svc.handle(ctx('#rss sources'))).toBe('Aye, Bee');
});
it('get <name>: summarizes the latest titles via chat with a bounded maxTokens', async () => {
stored().add({ name: 'myfeed', url: 'https://a.example/rss', tags: [] });
fetchMock.mockResolvedValueOnce(feedResponse(FEED_A));
llm.chat.mockResolvedValueOnce('Alpha and beta happened.');
const reply = await svc.handle(ctx('#rss get myfeed'));
expect(reply).toBe('Alpha and beta happened.');
const [messages, opts] = llm.chat.mock.calls[0];
expect(opts).toEqual({ maxTokens: 200 });
expect(messages[0].content).toContain('Alpha headline');
expect(messages[0].content).toContain('Beta headline');
});
it('get #tag: collects items across all sources with the tag; one dead feed does not sink it', async () => {
stored().add({ name: 'A', url: 'https://a.example/rss', tags: ['tech'] });
stored().add({ name: 'B', url: 'https://b.example/rss', tags: ['tech'] });
stored().add({ name: 'C', url: 'https://c.example/rss', tags: ['other'] }); // not matched by #tech
fetchMock
.mockResolvedValueOnce(feedResponse(FEED_A)) // A answers (Alpha, Beta)
.mockRejectedValueOnce(new Error('down')); // B is dead
llm.chat.mockResolvedValueOnce('Only alpha survived.');
const reply = await svc.handle(ctx('#rss get #tech'));
// B's fetch rejected, yet the summary still went out from A's titles alone
expect(reply).toBe('Only alpha survived.');
expect(llm.chat).toHaveBeenCalledTimes(1);
const content = llm.chat.mock.calls[0][0][0].content;
expect(content).toContain('Alpha headline');
});
it('get #tag: merges titles from every source carrying the tag', async () => {
stored().add({ name: 'A', url: 'https://a.example/rss', tags: ['tech'] });
stored().add({ name: 'B', url: 'https://b.example/rss', tags: ['tech'] });
fetchMock
.mockResolvedValueOnce(feedResponse(FEED_A)) // A → Alpha, Beta
.mockResolvedValueOnce(feedResponse(FEED_B)); // B → Gamma
llm.chat.mockResolvedValueOnce('all three.');
await svc.handle(ctx('#rss get #tech'));
const content = llm.chat.mock.calls[0][0][0].content;
expect(content).toContain('Alpha headline');
expect(content).toContain('Gamma headline'); // proves B's feed was collected too
});
it('get: unknown name and tag get distinct error replies', async () => {
expect(await svc.handle(ctx('#rss get nope'))).toBe('no source named nope');
stored().add({ name: 'A', url: 'https://a.example/rss', tags: ['tech'] });
expect(await svc.handle(ctx('#rss get #nothing'))).toBe(
'no sources with tag nothing',
);
});
});
+205
View File
@@ -0,0 +1,205 @@
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;
const SUMMARY_MAX_TOKENS = 200;
/**
* The `#rss` command family: stored RSS sources (SQLite, shared MESH_DB_PATH file) with
* LLM-generated tags and on-demand headline summaries. All replies are short strings —
* CommandService routes them through MeshService.respond() like any other command reply.
*/
@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). */
async handle(ctx: CommandContext): Promise<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>`. */
private async get(target: string): Promise<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 {
return await this.llm.chat(
[
{
role: 'user',
content: `Summarize these RSS headlines into a few short one-line highlights. Be concise:\n${titles.join('\n')}`,
},
],
{ maxTokens: SUMMARY_MAX_TOKENS },
);
} 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);
}
}
/** 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];
}
+80
View File
@@ -0,0 +1,80 @@
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { RssStore } from './rss.store.js';
describe('RssStore (real SQLite file)', () => {
let dir: string;
let store: RssStore;
beforeEach(() => {
dir = mkdtempSync(join(tmpdir(), 'mesh-bot-rss-'));
const db = new DatabaseSync(join(dir, 'test.db'));
store = new RssStore(db);
});
afterEach(() => {
rmSync(dir, { recursive: true, force: true });
});
it('adds and lists sources with tags round-tripped through the comma string', () => {
store.add({
name: 'Hacker News',
url: 'https://news.ycombinator.com/rss',
tags: ['tech', 'news'],
});
store.add({ name: 'LWN', url: 'https://lwn.net/headlines.rss', tags: [] });
expect(store.list()).toEqual([
{
name: 'Hacker News',
url: 'https://news.ycombinator.com/rss',
tags: ['tech', 'news'],
createdAt: expect.any(Number),
},
{
name: 'LWN',
url: 'https://lwn.net/headlines.rss',
tags: [],
createdAt: expect.any(Number),
},
]);
});
it('looks up and checks existence case-insensitively', () => {
store.add({ name: 'My Feed', url: 'https://example.com/rss', tags: ['a'] });
expect(store.get('my feed')?.url).toBe('https://example.com/rss');
expect(store.get('MY FEED')).toBeDefined();
expect(store.exists('mY fEeD')).toBe(true);
expect(store.exists('other')).toBe(false);
});
it('removes case-insensitively and reports whether a row went', () => {
store.add({ name: 'Gone', url: 'https://example.com/gone', tags: [] });
expect(store.remove('gone')).toBe(true);
expect(store.list()).toEqual([]);
expect(store.remove('gone')).toBe(false);
});
it('collects all distinct tags across sources, sorted', () => {
store.add({
name: 'A',
url: 'https://a.example/rss',
tags: ['zeta', 'alpha'],
});
store.add({
name: 'B',
url: 'https://b.example/rss',
tags: ['beta', 'alpha'],
});
expect(store.allTags()).toEqual(['alpha', 'beta', 'zeta']);
});
it('persists across a second connection to the same file (the production shape)', () => {
store.add({ name: 'Durable', url: 'https://d.example/rss', tags: ['x'] });
const db2 = new DatabaseSync(join(dir, 'test.db'));
expect(new RssStore(db2).get('durable')?.tags).toEqual(['x']);
});
});
+108
View File
@@ -0,0 +1,108 @@
import { DatabaseSync } from 'node:sqlite';
/** A stored RSS source; tags are a comma-separated string in the DB, split here. */
export interface RssSource {
name: string;
url: string;
tags: string[];
createdAt: number;
}
/** Idempotent — safe to call on every open (the table lives in the shared MESH_DB_PATH file). */
export function initRssTable(db: DatabaseSync): void {
db.exec(`CREATE TABLE IF NOT EXISTS rss_sources (
name TEXT PRIMARY KEY,
url TEXT NOT NULL,
tags TEXT NOT NULL DEFAULT '',
created_at INTEGER
)`);
}
/** CRUD for the rss_sources table. Names are matched case-insensitively everywhere. */
export class RssStore {
constructor(private readonly db: DatabaseSync) {
initRssTable(db);
}
list(): RssSource[] {
const rows = this.db
.prepare(
'SELECT name, url, tags, created_at FROM rss_sources ORDER BY name',
)
.all() as {
name: string;
url: string;
tags: string;
created_at: number;
}[];
return rows.map((r) => ({
name: r.name,
url: r.url,
tags: splitTags(r.tags),
createdAt: r.created_at,
}));
}
/** Case-insensitive lookup by name. */
get(name: string): RssSource | undefined {
const row = this.db
.prepare(
'SELECT name, url, tags, created_at FROM rss_sources WHERE name = ? COLLATE NOCASE',
)
.get(name) as
| { name: string; url: string; tags: string; created_at: number }
| undefined;
return (
row && {
name: row.name,
url: row.url,
tags: splitTags(row.tags),
createdAt: row.created_at,
}
);
}
/** Case-insensitive duplicate check. */
exists(name: string): boolean {
const row = this.db
.prepare('SELECT 1 FROM rss_sources WHERE name = ? COLLATE NOCASE')
.get(name);
return row !== undefined;
}
add(source: { name: string; url: string; tags: string[] }): void {
this.db
.prepare(
'INSERT INTO rss_sources (name, url, tags, created_at) VALUES (?, ?, ?, ?)',
)
.run(source.name, source.url, joinTags(source.tags), Date.now());
}
/** Case-insensitive delete; true when a row was removed. */
remove(name: string): boolean {
const result = this.db
.prepare('DELETE FROM rss_sources WHERE name = ? COLLATE NOCASE')
.run(name);
return Number(result.changes) > 0;
}
/** All distinct tags across sources, sorted. */
allTags(): string[] {
const set = new Set<string>();
for (const source of this.list()) {
for (const tag of source.tags) set.add(tag);
}
return [...set].sort();
}
}
export function splitTags(raw: string): string[] {
return raw
.split(',')
.map((t) => t.trim())
.filter(Boolean);
}
export function joinTags(tags: string[]): string {
return tags.join(',');
}