Skip to content

Commit 731d9a7

Browse files
sommiosommio
committed
feat(api): add wake-triggered feed auto-refresh
Add a same-process wake detector, durable refresh state, and bounded retry support so the API can refresh stale feeds after sleep without changing bootstrap ingestion semantics.
1 parent eff82af commit 731d9a7

25 files changed

Lines changed: 2280 additions & 121 deletions

‎apps/api/.env.example‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
DATABASE_URL=postgresql://rssift:rssift@127.0.0.1:5432/rssift
22
TEST_DATABASE_URL=postgresql://rssift:rssift@127.0.0.1:5432/rssift_test
3+
FEED_AUTO_REFRESH_INTERVAL_HOURS=6
34
FEED_MAX_ARTICLES_PER_FEED=10
45
FEED_OPML_PATH=./feeds.opml
56
INGEST_ON_BOOT=true

‎apps/api/README.md‎

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,9 @@ cp apps/api/feeds.opml.example apps/api/feeds.opml
1919
- `DATABASE_URL` is the development database used by the API runtime and by
2020
`db:deploy`, `db:reset`, and `db:seed`.
2121
- `TEST_DATABASE_URL` is reserved for automated tests.
22+
- `FEED_AUTO_REFRESH_INTERVAL_HOURS` defaults to `6`; after same-process
23+
wake/resume events, the API only runs a background refresh when the last
24+
successful wake-driven refresh is older than this interval.
2225
- `FEED_MAX_ARTICLES_PER_FEED` defaults to `10`; each feed ingestion run only
2326
persists the newest N entries from that feed.
2427
- `FEED_OPML_PATH` defaults to `./feeds.opml`.
@@ -62,9 +65,10 @@ The API runs on `http://127.0.0.1:3000` by default.
6265
## Runtime Ownership
6366

6467
- `apps/api/.env.local` owns `DATABASE_URL`, `TEST_DATABASE_URL`,
65-
`FEED_MAX_ARTICLES_PER_FEED`, `FEED_OPML_PATH`, `INGEST_ON_BOOT`,
66-
`LLM_BASE_URL`, `LLM_API_KEY`, `LLM_MODEL`, `LLM_SUMMARY_CONCURRENCY`,
67-
`LLM_SUMMARY_LANGUAGE`, optional `LLM_TIMEOUT_MS`, and `PORT`.
68+
`FEED_AUTO_REFRESH_INTERVAL_HOURS`, `FEED_MAX_ARTICLES_PER_FEED`,
69+
`FEED_OPML_PATH`, `INGEST_ON_BOOT`, `LLM_BASE_URL`, `LLM_API_KEY`,
70+
`LLM_MODEL`, `LLM_SUMMARY_CONCURRENCY`, `LLM_SUMMARY_LANGUAGE`, optional
71+
`LLM_TIMEOUT_MS`, and `PORT`.
6872
- `apps/api/feeds.opml` is the app-owned local subscription input.
6973
- Feed parsing uses `feedsmith`.
7074
- Article body extraction uses `@mozilla/readability`, `jsdom`, and `turndown`
@@ -73,6 +77,12 @@ The API runs on `http://127.0.0.1:3000` by default.
7377
OpenAI-compatible gateway, validate the structured result with `zod`, and
7478
persist canonical Markdown, `translatedTitle`, and any terminal failure reason
7579
in `summaryErrorReason`.
80+
- Same-process sleep/freeze recovery is allowed to trigger a background feed
81+
auto-refresh in the future, but `INGEST_ON_BOOT` remains startup-only; the
82+
wake interval is an elapsed-hours check, not a cron schedule.
83+
- Wake auto-refresh is explicitly single-process only. It dedupes overlapping
84+
resume events inside one Node process, but it is not a distributed lock for
85+
multi-replica deployments.
7686
- Historical rows with `contentMarkdown` and empty summary fields are picked up
7787
by an internal bootstrap backfill; there is no public regenerate endpoint.
7888
- Prisma schema, migrations, and generated client stay inside `apps/api`.
@@ -131,6 +141,15 @@ do not apply Prisma migrations for you. Run `pnpm --filter api
131141
db:deploy` yourself before booting Nest if the local database schema is
132142
behind.
133143

144+
## Auto-Refresh Logging
145+
146+
- `feed_auto_refresh` emits `skipped` with reasons such as
147+
`interval_not_elapsed` and `already_running`, plus `triggered`,
148+
`partial_success`, `all_success`, and `full_failure`.
149+
- `feed_ingestion_summary` also carries a `trigger` field so operators can
150+
distinguish `bootstrap` runs from `auto_refresh_resume` runs in downstream
151+
logs.
152+
134153
## Test Surfaces
135154

136155
- Colocated specs stay next to the feature code in `apps/api/src/**/*.spec.ts`.

‎apps/api/e2e/feed-auto-refresh.e2e-spec.ts‎

Lines changed: 332 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,332 @@
1+
import {
2+
afterAll,
3+
beforeAll,
4+
beforeEach,
5+
describe,
6+
expect,
7+
it,
8+
jest,
9+
} from "@jest/globals";
10+
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
11+
import { tmpdir } from "node:os";
12+
import { join } from "node:path";
13+
14+
import { ArticleContentExtractionService } from "../src/article-content/article-content-extraction.service";
15+
import { ArticleContentRepository } from "../src/article-content/article-content.repository";
16+
import { ArticleContentService } from "../src/article-content/article-content.service";
17+
import type { ArticleSummaryService } from "../src/article-summary/article-summary.service";
18+
import { FeedAutoRefreshRepository } from "../src/feeds/feed-auto-refresh.repository";
19+
import { FeedAutoRefreshService } from "../src/feeds/feed-auto-refresh.service";
20+
import { ArticleIdentityService } from "../src/feeds/article-identity.service";
21+
import { FeedIngestionService } from "../src/feeds/feed-ingestion.service";
22+
import type { PrismaService } from "../src/prisma/prisma.service";
23+
import {
24+
createTestPrismaClient,
25+
prepareTestDatabase,
26+
} from "../test-support/database";
27+
28+
function createFeedXml() {
29+
return `<?xml version="1.0"?>
30+
<rss version="2.0">
31+
<channel>
32+
<title>Wake Feed</title>
33+
<item>
34+
<title>Wake Article</title>
35+
<link>https://example.com/articles/wake-article</link>
36+
<description>Wake summary</description>
37+
<guid isPermaLink="false">wake-guid</guid>
38+
<pubDate>Tue, 20 Apr 2026 00:00:00 GMT</pubDate>
39+
</item>
40+
</channel>
41+
</rss>`;
42+
}
43+
44+
function createArticleHtml() {
45+
return `<!doctype html>
46+
<html>
47+
<body>
48+
<article>
49+
<h1>Wake Article</h1>
50+
<p>Recovered through the canonical ingestion pipeline.</p>
51+
</article>
52+
</body>
53+
</html>`;
54+
}
55+
56+
function createDeferred<T>() {
57+
let resolve!: (value: T) => void;
58+
let reject!: (reason?: unknown) => void;
59+
const promise = new Promise<T>((innerResolve, innerReject) => {
60+
resolve = innerResolve;
61+
reject = innerReject;
62+
});
63+
64+
return {
65+
promise,
66+
reject,
67+
resolve,
68+
};
69+
}
70+
71+
type FeedAutoRefreshSuiteState = {
72+
opmlPath: string;
73+
prisma: ReturnType<typeof createTestPrismaClient>;
74+
repository: FeedAutoRefreshRepository;
75+
service: FeedAutoRefreshService;
76+
ingestionService: FeedIngestionService;
77+
tempDir: string;
78+
};
79+
80+
function createServices(
81+
prisma: ReturnType<typeof createTestPrismaClient>,
82+
articleSummaryService: Pick<ArticleSummaryService, "schedule"> = {
83+
schedule: () => ({ status: "scheduled" }),
84+
},
85+
) {
86+
const articleContentRepository = new ArticleContentRepository(
87+
prisma as unknown as PrismaService,
88+
);
89+
const articleContentService = new ArticleContentService(
90+
articleContentRepository,
91+
new ArticleContentExtractionService(),
92+
articleSummaryService as ArticleSummaryService,
93+
);
94+
const ingestionService = new FeedIngestionService(
95+
prisma as unknown as PrismaService,
96+
new ArticleIdentityService(),
97+
articleContentService,
98+
articleSummaryService as ArticleSummaryService,
99+
);
100+
const repository = new FeedAutoRefreshRepository(
101+
prisma as unknown as PrismaService,
102+
);
103+
104+
return {
105+
ingestionService,
106+
repository,
107+
service: new FeedAutoRefreshService(repository, ingestionService),
108+
};
109+
}
110+
111+
async function setupSuite(state: FeedAutoRefreshSuiteState) {
112+
process.env["DATABASE_URL"] ??=
113+
"postgresql://rssift:rssift@127.0.0.1:5432/rssift_test";
114+
process.env["TEST_DATABASE_URL"] ??=
115+
"postgresql://rssift:rssift@127.0.0.1:5432/rssift_test";
116+
process.env["INGEST_ON_BOOT"] = "false";
117+
process.env["FEED_AUTO_REFRESH_INTERVAL_HOURS"] = "6";
118+
119+
await prepareTestDatabase();
120+
state.prisma = createTestPrismaClient();
121+
state.tempDir = mkdtempSync(join(tmpdir(), "rssift-feed-auto-refresh-e2e-"));
122+
state.opmlPath = join(state.tempDir, "feeds.opml");
123+
writeFileSync(
124+
state.opmlPath,
125+
`<?xml version="1.0" encoding="UTF-8"?>
126+
<opml version="2.0">
127+
<body>
128+
<outline text="Wake Feed" xmlUrl="https://example.com/feed-auto-refresh.xml" />
129+
</body>
130+
</opml>`,
131+
);
132+
process.env["FEED_OPML_PATH"] = state.opmlPath;
133+
134+
const services = createServices(state.prisma);
135+
state.ingestionService = services.ingestionService;
136+
state.repository = services.repository;
137+
state.service = services.service;
138+
}
139+
140+
async function resetSuite(state: FeedAutoRefreshSuiteState) {
141+
await state.prisma.article.deleteMany();
142+
await state.prisma.feed.deleteMany();
143+
await state.prisma.feedAutoRefreshState.deleteMany();
144+
state.service.onApplicationShutdown();
145+
jest.restoreAllMocks();
146+
jest.useRealTimers();
147+
}
148+
149+
async function teardownSuite(state: FeedAutoRefreshSuiteState) {
150+
await state.prisma.$disconnect();
151+
rmSync(state.tempDir, { force: true, recursive: true });
152+
}
153+
154+
async function triggerWakeCheck(state: FeedAutoRefreshSuiteState) {
155+
const baseMs = Date.parse("2026-04-20T00:00:00.000Z");
156+
const dateNowSpy = jest.spyOn(Date, "now").mockReturnValue(baseMs);
157+
state.service.onApplicationBootstrap();
158+
dateNowSpy.mockRestore();
159+
state.service.onApplicationShutdown();
160+
await state.service.checkHeartbeat(baseMs + 30_000);
161+
await state.service.checkHeartbeat(baseMs + 600_000);
162+
}
163+
164+
describe("Feed auto-refresh orchestration", () => {
165+
const state = {} as FeedAutoRefreshSuiteState;
166+
167+
beforeAll(async () => {
168+
await setupSuite(state);
169+
});
170+
171+
beforeEach(async () => {
172+
await resetSuite(state);
173+
});
174+
175+
afterAll(async () => {
176+
await teardownSuite(state);
177+
});
178+
179+
it("triggers the first wake-driven refresh when no state row exists yet", async () => {
180+
jest
181+
.spyOn(global, "fetch")
182+
.mockImplementation((input: string | URL | Request) => {
183+
const url =
184+
typeof input === "string"
185+
? input
186+
: input instanceof URL
187+
? input.toString()
188+
: input.url;
189+
190+
if (url === "https://example.com/feed-auto-refresh.xml") {
191+
return Promise.resolve(
192+
new Response(createFeedXml(), { status: 200 }),
193+
);
194+
}
195+
196+
if (url === "https://example.com/articles/wake-article") {
197+
return Promise.resolve(
198+
new Response(createArticleHtml(), { status: 200 }),
199+
);
200+
}
201+
202+
return Promise.resolve(new Response("missing", { status: 404 }));
203+
});
204+
205+
await triggerWakeCheck(state);
206+
207+
const articles = await state.prisma.article.findMany();
208+
const refreshState = await state.prisma.feedAutoRefreshState.findUnique({
209+
where: {
210+
id: "global",
211+
},
212+
});
213+
214+
expect(articles).toHaveLength(1);
215+
expect(articles[0]?.contentMarkdown).toBe(
216+
"# Wake Article\n\nRecovered through the canonical ingestion pipeline.",
217+
);
218+
expect(refreshState?.lastSuccessfulAutoRefreshAt?.toISOString()).toBe(
219+
"2026-04-20T00:10:00.000Z",
220+
);
221+
});
222+
223+
it("re-runs wake refresh when the stored success timestamp is stale", async () => {
224+
await state.prisma.feedAutoRefreshState.create({
225+
data: {
226+
id: "global",
227+
lastSuccessfulAutoRefreshAt: new Date("2026-04-19T16:00:00.000Z"),
228+
},
229+
});
230+
jest
231+
.spyOn(global, "fetch")
232+
.mockImplementation((input: string | URL | Request) => {
233+
const url =
234+
typeof input === "string"
235+
? input
236+
: input instanceof URL
237+
? input.toString()
238+
: input.url;
239+
240+
if (url === "https://example.com/feed-auto-refresh.xml") {
241+
return Promise.resolve(
242+
new Response(createFeedXml(), { status: 200 }),
243+
);
244+
}
245+
246+
if (url === "https://example.com/articles/wake-article") {
247+
return Promise.resolve(
248+
new Response(createArticleHtml(), { status: 200 }),
249+
);
250+
}
251+
252+
return Promise.resolve(new Response("missing", { status: 404 }));
253+
});
254+
255+
await triggerWakeCheck(state);
256+
257+
const refreshState = await state.prisma.feedAutoRefreshState.findUnique({
258+
where: {
259+
id: "global",
260+
},
261+
});
262+
263+
expect(await state.prisma.article.count()).toBe(1);
264+
expect(refreshState?.lastSuccessfulAutoRefreshAt?.toISOString()).toBe(
265+
"2026-04-20T00:10:00.000Z",
266+
);
267+
});
268+
269+
it("skips wake refresh when the last success timestamp is still fresh", async () => {
270+
await state.prisma.feedAutoRefreshState.create({
271+
data: {
272+
id: "global",
273+
lastSuccessfulAutoRefreshAt: new Date("2026-04-19T22:30:00.000Z"),
274+
},
275+
});
276+
const fetchSpy = jest.spyOn(global, "fetch");
277+
278+
await triggerWakeCheck(state);
279+
280+
const refreshState = await state.prisma.feedAutoRefreshState.findUnique({
281+
where: {
282+
id: "global",
283+
},
284+
});
285+
286+
expect(fetchSpy).not.toHaveBeenCalled();
287+
expect(await state.prisma.article.count()).toBe(0);
288+
expect(refreshState?.lastSuccessfulAutoRefreshAt?.toISOString()).toBe(
289+
"2026-04-19T22:30:00.000Z",
290+
);
291+
});
292+
293+
it("does not start a duplicate ingestion wave while the first wake-driven run is still active", async () => {
294+
const deferred = createDeferred<{
295+
failedCount: number;
296+
status: "all_success";
297+
successCount: number;
298+
totalFeeds: number;
299+
trigger: "auto_refresh_resume";
300+
}>();
301+
const started = createDeferred<undefined>();
302+
303+
const baseMs = Date.parse("2026-04-20T00:00:00.000Z");
304+
const dateNowSpy = jest.spyOn(Date, "now").mockReturnValue(baseMs);
305+
state.service.onApplicationBootstrap();
306+
dateNowSpy.mockRestore();
307+
state.service.onApplicationShutdown();
308+
await state.service.checkHeartbeat(baseMs + 30_000);
309+
310+
const ingestSpy = jest
311+
.spyOn(state.ingestionService, "ingestFromOpml")
312+
.mockImplementation(() => {
313+
started.resolve(undefined);
314+
return deferred.promise;
315+
});
316+
317+
void state.service.checkHeartbeat(baseMs + 600_000);
318+
await started.promise;
319+
await state.service.checkHeartbeat(baseMs + 1_200_000);
320+
321+
expect(ingestSpy).toHaveBeenCalledTimes(1);
322+
323+
deferred.resolve({
324+
failedCount: 0,
325+
status: "all_success",
326+
successCount: 1,
327+
totalFeeds: 1,
328+
trigger: "auto_refresh_resume",
329+
});
330+
await Promise.resolve();
331+
});
332+
});

0 commit comments

Comments
 (0)