src/rss.rs
//! **RSS / Atom syndication feed** — the ubiquitous "publish a stream of dated
//! items" access method behind government + agency open feeds (Krisinformation.se
//! alerts, SMHI warnings, EU/EEA news, agency press wires). Two on-the-wire
//! dialects answer the same shape: **RSS 2.0** (`<rss><channel><item>…`) and
//! **Atom 1.0** (`<feed><entry>…`). This is a DISTINCT parser from the crate's
//! other collectors: the WFS collector parses **GML** feature geometry, ENTSO-E
//! parses market-document XML, and the JSON collectors (PxWeb / SDMX / USGS /
//! generic REST) never touch syndication XML at all — none flatten a feed's
//! dated items to rows.
//!
//! [`parse_feed`] streams either dialect with `quick-xml` (same engine as the WFS
//! and ENTSO-E parsers) and maps each `item`/`entry` → one row: `feed_title`
//! (the channel/feed title carried onto every row), `id` (`guid` / atom `id`),
//! `title`, `link` (RSS `<link>` text OR the Atom `<link href=…>` alternate),
//! `summary` (`description` / `summary` / `content`), `author` (RSS `<author>` /
//! `dc:creator` / the Atom nested `<author><name>`), `published` (`pubDate` /
//! `published` / `updated`, kept as the source string so downstream owns date
//! parsing), and `categories` (every `<category>` term, as a compact JSON array
//! string — the same "many-values → JSON" shape the WFS/Overpass collectors use).
//! Pure-Rust, offline-testable — no network, no external feed library.
//!
//! Offline contract: [`parse_feed`] turns a canned RSS/Atom document into
//! per-item rows → Arrow; an item with neither title nor link nor id still yields
//! a row only when it carried SOME content, and a feed with no items becomes a
//! RAGNARÖK `reason` (never a silent empty).
use anyhow::Result;
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Schema};
use quick_xml::events::{BytesStart, Event};
use super::arrow_util::{batch, schema, str_col};
use super::{
AuthKind, BatchSink, Bounds, Checkpoint, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
/// One flattened syndication entry — a single RSS `<item>` or Atom `<entry>`.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct FeedItem {
/// The channel/feed `<title>` (carried onto every item so a row self-describes
/// its source feed). `None` if the feed declared none.
pub feed_title: Option<String>,
/// **The feed's OWN address** — the RSS channel `<link>` or the Atom feed-level
/// `<link rel="self" href=…>` — carried onto every item.
///
/// This exists because `feed_title` is not always an identity. All five
/// regeringen.se feeds korp configures publish the same channel title,
/// `RSS - Regeringen.se` (measured 2026-09-05), so a row landed from the
/// propositions feed and a row landed from the press feed are indistinguishable by
/// title. The channel `<link>` echoes the request URL, `preFilteredCategories` and
/// all, and separates them.
pub feed_link: Option<String>,
/// Stable item id — RSS `<guid>` or Atom `<id>`. The CDC / join key.
pub id: Option<String>,
/// Item headline (`<title>`).
pub title: Option<String>,
/// Canonical item URL — RSS `<link>` text, or the Atom `<link href=…>` whose
/// `rel` is `alternate`/absent (a `rel="self"` link is skipped).
pub link: Option<String>,
/// Item body / abstract (`<description>` / `<summary>` / `<content>`).
pub summary: Option<String>,
/// Author (RSS `<author>` / `dc:creator` / the Atom nested `<author><name>`).
pub author: Option<String>,
/// Publication timestamp as the RAW source string (`<pubDate>` RFC-822 /
/// Atom `<published>`/`<updated>` RFC-3339) — downstream owns date parsing.
pub published: Option<String>,
/// Every `<category>` term (RSS text content or Atom `term` attribute),
/// de-duplicated in source order.
pub categories: Vec<String>,
/// **The taxonomy id each category carried**, INDEX-ALIGNED with
/// [`categories`](FeedItem::categories) — the RSS `domain` attribute or the Atom
/// `scheme`, and the empty string where the feed gave none.
///
/// regeringen.se puts a real node id in `domain`: `1334` Pressmeddelande, `1329`
/// Proposition, `1287` Justitiedepartementet (measured 2026-09-05, 800/800 items
/// over 8 pages). The label beside it is Swedish prose that can be re-worded; the
/// id is what a downstream query should read, and dropping it — which this parser
/// did — left the department reachable only by string-matching the prose.
///
/// Alignment is by construction: a term and its id are pushed together, and the
/// de-duplication is on the PAIR, so `categories[i]` and `category_ids[i]` always
/// describe the same `<category>` element.
pub category_ids: Vec<String>,
}
/// The item-boundary element names (RSS `item`, Atom `entry`).
fn is_item(local: &str) -> bool {
matches!(local, "item" | "entry")
}
fn local_name(name: quick_xml::name::QName) -> String {
String::from_utf8_lossy(name.local_name().as_ref()).into_owned()
}
/// Read an attribute value by its local name (namespace-prefix-insensitive).
fn attr_local(e: &BytesStart, want: &str) -> Option<String> {
for a in e.attributes().flatten() {
if a.key.local_name().as_ref() == want.as_bytes() {
return Some(String::from_utf8_lossy(a.value.as_ref()).into_owned());
}
}
None
}
/// Apply an Atom `<link href=… rel=…>` to the item: take the first non-`self`
/// href (an `alternate`/absent `rel` is the canonical item URL).
fn apply_link_attrs(e: &BytesStart, cur: &mut FeedItem) {
if attr_local(e, "rel").as_deref() == Some("self") {
return; // the feed's own address, not the item's
}
if cur.link.is_none() {
if let Some(href) = attr_local(e, "href") {
cur.link = Some(href);
}
}
}
/// Apply an Atom **feed-level** `<link rel=… href=…>` to the feed's own address.
///
/// The mirror image of [`apply_link_attrs`], and deliberately the opposite rule: inside
/// an entry the `rel="self"` link is the one to SKIP (it is the feed's address, not the
/// entry's), and at feed level it is the one to TAKE. `rel="self"` always wins; an
/// `alternate`/absent `rel` (the publisher's home page) only fills a gap.
fn apply_feed_link(e: &BytesStart, feed_link: &mut Option<String>) {
let Some(href) = attr_local(e, "href") else {
return;
};
if attr_local(e, "rel").as_deref() == Some("self") {
*feed_link = Some(href);
} else if feed_link.is_none() {
*feed_link = Some(href);
}
}
/// Push a category term with its taxonomy id, trimmed, ignoring blanks.
///
/// De-duplication is on the **pair**, not on the term: two `<category>` elements with
/// the same label under different taxonomies are two different nodes, and dropping one
/// of them would also break the index alignment
/// [`FeedItem::category_ids`] promises.
fn push_cat(cur: &mut FeedItem, id: Option<&str>, term: &str) {
let t = term.trim();
if t.is_empty() {
return;
}
let id = id.unwrap_or("").trim().to_string();
if cur
.categories
.iter()
.zip(&cur.category_ids)
.any(|(c, i)| c == t && *i == id)
{
return;
}
cur.categories.push(t.to_string());
cur.category_ids.push(id);
}
/// The taxonomy id a `<category>` element carries: the RSS 2.0 `domain` attribute, or
/// the Atom `scheme`. `None` when it carries neither.
fn category_id(e: &BytesStart) -> Option<String> {
attr_local(e, "domain").or_else(|| attr_local(e, "scheme"))
}
/// Fold a text leaf (closing element `local` with trimmed `txt`) into the current
/// item. First non-empty value wins per field (feeds vary the element order).
fn apply_leaf(cur: &mut FeedItem, local: &str, txt: &str, in_author: bool) {
if txt.is_empty() {
return;
}
let set = |slot: &mut Option<String>| {
if slot.is_none() {
*slot = Some(txt.to_string());
}
};
match local {
"title" => set(&mut cur.title),
// RSS `<link>text</link>` (the Atom href form is captured on Start/Empty).
"link" => set(&mut cur.link),
"description" | "summary" | "content" | "encoded" => set(&mut cur.summary),
"guid" | "id" => set(&mut cur.id),
"pubDate" | "published" | "updated" | "date" => set(&mut cur.published),
// Atom `<author><name>…` (only inside an author element) …
"name" if in_author => set(&mut cur.author),
// … RSS `<author>` / `dc:creator` text.
"creator" => set(&mut cur.author),
// `category` is deliberately NOT here: its text and its `domain`/`scheme`
// attribute arrive on two different events, so it is folded in `parse_feed`
// where the pending id from the Start event is still in scope.
_ => {}
}
}
/// Parse an RSS 2.0 or Atom 1.0 document → per-item rows. Dialect-agnostic: the
/// item boundary is `item`/`entry`, and each dialect's field spellings all fold
/// through [`apply_leaf`] / [`apply_link_attrs`]. No network, no feed library.
pub fn parse_feed(xml: &str) -> Result<Vec<FeedItem>> {
let mut reader = quick_xml::Reader::from_str(xml);
let mut items: Vec<FeedItem> = Vec::new();
let mut feed_title: Option<String> = None;
let mut feed_link: Option<String> = None;
let mut in_item = false;
let mut in_author = false;
let mut cur = FeedItem::default();
// Text buffer for the leaf currently open; cleared when a new element opens.
let mut buf = String::new();
// The `domain`/`scheme` of the `<category>` element currently open. RSS 2.0 puts
// the id on the Start event and the label in the text, so the two only meet here.
let mut pending_cat_id: Option<String> = None;
loop {
match reader.read_event()? {
Event::Start(e) => {
let local = local_name(e.name());
buf.clear();
if is_item(&local) {
in_item = true;
cur = FeedItem {
feed_title: feed_title.clone(),
feed_link: feed_link.clone(),
..Default::default()
};
} else if in_item {
match local.as_str() {
"author" => in_author = true,
"link" => apply_link_attrs(&e, &mut cur),
"category" => {
let id = category_id(&e);
match attr_local(&e, "term") {
// Atom: the label IS the `term` attribute, so the
// element is complete already.
Some(term) => push_cat(&mut cur, id.as_deref(), &term),
// RSS: hold the id until the closing tag brings the
// label.
None => pending_cat_id = id,
}
}
_ => {}
}
} else if local == "link" {
// An Atom feed-level `<link rel="self" href=…>` opened as a pair.
apply_feed_link(&e, &mut feed_link);
}
}
// Self-closing tags (Atom `<link .../>`, `<category .../>`) never fire
// an End — capture their attribute payload here.
Event::Empty(e) => {
let local = local_name(e.name());
if in_item {
match local.as_str() {
"link" => apply_link_attrs(&e, &mut cur),
"category" => {
if let Some(term) = attr_local(&e, "term") {
push_cat(&mut cur, category_id(&e).as_deref(), &term);
}
}
_ => {}
}
} else if local == "link" {
// The usual Atom spelling of the feed's own address.
apply_feed_link(&e, &mut feed_link);
}
}
Event::Text(t) => buf.push_str(&t.unescape()?),
// CDATA-wrapped bodies (`<description><![CDATA[…]]></description>`) are
// raw — appended verbatim, not entity-unescaped.
Event::CData(t) => buf.push_str(&String::from_utf8_lossy(&t.into_inner())),
Event::End(e) => {
let local = local_name(e.name());
let txt = buf.trim().to_string();
buf.clear();
if is_item(&local) {
items.push(std::mem::take(&mut cur));
in_item = false;
in_author = false;
pending_cat_id = None;
} else if local == "author" {
// A plain RSS `<author>email</author>` (no nested <name>).
apply_leaf(&mut cur, "creator", &txt, in_author);
in_author = false;
} else if in_item && local == "category" {
push_cat(&mut cur, pending_cat_id.take().as_deref(), &txt);
} else if in_item {
apply_leaf(&mut cur, &local, &txt, in_author);
} else if local == "title" && feed_title.is_none() && !txt.is_empty() {
// The channel/feed title (the first title outside any item).
feed_title = Some(txt);
} else if local == "link" && feed_link.is_none() && !txt.is_empty() {
// The RSS channel `<link>` — regeringen.se echoes the request URL
// here, which is the only thing in the document that tells its five
// identically-titled feeds apart.
feed_link = Some(txt);
}
}
Event::Eof => break,
_ => {}
}
}
Ok(items)
}
/// Feed items → one Arrow `RecordBatch`. `categories` is a compact JSON array
/// string (`["Alert","Weather"]`), mirroring the WFS `properties` column shape.
pub fn feed_batch(items: &[FeedItem]) -> Result<RecordBatch> {
let cats = |i: &FeedItem| {
Some(serde_json::to_string(&i.categories).unwrap_or_else(|_| "[]".to_string()))
};
let cat_ids = |i: &FeedItem| {
Some(serde_json::to_string(&i.category_ids).unwrap_or_else(|_| "[]".to_string()))
};
let cols = vec![
str_col(
&items
.iter()
.map(|i| i.feed_title.clone())
.collect::<Vec<_>>(),
),
str_col(
&items
.iter()
.map(|i| i.feed_link.clone())
.collect::<Vec<_>>(),
),
str_col(&items.iter().map(|i| i.id.clone()).collect::<Vec<_>>()),
str_col(&items.iter().map(|i| i.title.clone()).collect::<Vec<_>>()),
str_col(&items.iter().map(|i| i.link.clone()).collect::<Vec<_>>()),
str_col(&items.iter().map(|i| i.summary.clone()).collect::<Vec<_>>()),
str_col(&items.iter().map(|i| i.author.clone()).collect::<Vec<_>>()),
str_col(
&items
.iter()
.map(|i| i.published.clone())
.collect::<Vec<_>>(),
),
str_col(&items.iter().map(cats).collect::<Vec<_>>()),
str_col(&items.iter().map(cat_ids).collect::<Vec<_>>()),
];
batch(&feed_schema(), cols)
}
pub fn feed_schema() -> Schema {
schema(&[
("feed_title", DataType::Utf8, true),
("feed_link", DataType::Utf8, true),
("id", DataType::Utf8, true),
("title", DataType::Utf8, true),
("link", DataType::Utf8, true),
("summary", DataType::Utf8, true),
("author", DataType::Utf8, true),
("published", DataType::Utf8, true),
("categories", DataType::Utf8, false),
("category_ids", DataType::Utf8, false),
])
}
/// **regeringen.se's press-release feed** — the default this collector points at.
///
/// The Swedish government publishes its whole document archive through ONE RSS endpoint
/// filtered by taxonomy node, and `preFilteredCategories` is the node. Measured live
/// 2026-09-05:
///
/// | id | feed | page 1 |
/// |---|---|---|
/// | `1334` | Pressmeddelande | 100 items, 2026-08-28 → 2026-09-04 (~7 days, ~14/day) |
/// | `1329` | Proposition | 100 items, 2026-03-17 → 2026-09-01 |
/// | `1330` | Skrivelse | — |
/// | `1331` | Statens offentliga utredningar | — |
/// | `1341` | Ds | — |
///
/// All five are configured in korp's `known_sources()`; this constant is the one the
/// collector runs with when nothing configured it, and it is a REAL feed rather than the
/// RFC 2606 placeholder that used to sit here — a keyless government press wire is a
/// better default than a collector that cannot run.
///
/// The archive behind it is deep, not a rolling window: paging the unfiltered feed
/// reaches **page 511** (57 items on the last), i.e. **~51 057 documents**, and page
/// 2000 answers 200 with zero items. This collector reads page 1 only — see the card.
pub const REGERINGEN_PRESSMEDDELANDEN_FEED: &str =
"https://www.regeringen.se/Filter/RssFeed?filterType=Taxonomy\
&filterByType=FilterablePageBase&preFilteredCategories=1334&rootPageReference=0";
/// The RSS / Atom syndication-feed collector. `cfg.base_url` is the feed URL; the
/// RSS/Atom document is fetched then flattened to per-item rows → warehouse table.
///
/// The fetch is CONDITIONAL: a feed is one URL rewritten in place, so the run sends the
/// validators the last read stored and a `304` lands nothing and transfers nothing. See
/// [`Collector::freshness`] on the impl below for the measurement that establishes it.
#[derive(Default)]
pub struct RssFeedCollector {
/// What the last [`Collector::collect`] read — the validators the feed served and
/// the bytes it cost, reported back through [`Collector::checkpoint`]. The collector
/// REPORTS; the runner decides whether to persist.
last_read: std::sync::Mutex<Option<Checkpoint>>,
}
#[async_trait::async_trait]
impl Collector for RssFeedCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "rss-atom".into(),
source_kind: SourceKind::Rest,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["rss".into(), "atom".into(), "xml".into()],
license: "per-source (public government/agency feeds; typically CC-BY / open)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["id".into(), "link".into()],
description: "RSS 2.0 / Atom 1.0 syndication feed — each item/entry flattened to a \
row (feed_title, feed_link, id, title, link, summary, author, \
published, categories, category_ids) via a pure-Rust quick-xml parse \
→ warehouse table. Defaults to regeringen.se's press-release feed; \
korp configures five Regeringen taxonomy feeds against it."
.into(),
}
}
/// **A generic reader whose FORMAT publishes a rate limit even though its source
/// cannot.** RSS 2.0 is unusual that way: `<ttl>` (minutes a channel may be cached)
/// plus `<skipHours>`/`<skipDays>` are the publisher telling a poller how often to
/// come back, in the feed itself. This parser reads none of them — it lands the
/// items and nothing else — so the schedule is the operator's to set, and the card
/// says where the publisher's own answer is written.
///
/// Two dialects, two documents: RSS 2.0 from the RSS Advisory Board and Atom 1.0
/// from the IETF as RFC 4287. `docs_url` is the RSS specification because that is
/// the one that also carries `ttl`/`skipHours`; the Atom standard is named in the
/// quirks rather than lost.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
Some(
HumanDoc::new(
"whoever publishes the feed you configure — this is a generic reader. The \
FORMATS have owners (the RSS Advisory Board for RSS 2.0, the IETF for \
Atom 1.0/RFC 4287); the items do not.",
"the agency whose wire it is — the feed's own `<title>` is carried onto \
every row as `feed_title` so a landed row still names its source",
"https://www.rssboard.org/rss-specification",
"https://www.rfc-editor.org/rfc/rfc4287",
)
.with_contact(Contact::Unknown)
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Documented(
"the FEED states it, not the format: RSS 2.0 defines `<ttl>` (minutes the \
channel may be cached) and `<skipHours>`/`<skipDays>`. This parser reads \
none of the three, so read them off the feed yourself and honour them in \
the schedule."
.into(),
))
.with_quirk(
"`terms_url` is the RSS SPECIFICATION (itself CC Attribution/Share-Alike, \
authored by the RSS Advisory Board) — it is not anyone's data licence. \
A government feed's terms live on the publishing agency's own open-data \
page, and korp cannot know which agency that is until `base_url` is set.",
)
.with_quirk(
"The declared source is regeringen.se's press-release feed \
(`preFilteredCategories=1334`), keyless and with no agreement — it used \
to be the RFC 2606 placeholder `https://example.se/feed.xml`, which made \
the collector report `Missing::PlaceholderSource` and never run. Point \
`base_url` anywhere; korp already configures five Regeringen taxonomy \
feeds (1329 Proposition, 1330 Skrivelse, 1331 SOU, 1334 Pressmeddelande, \
1341 Ds) against this one preset.",
)
.with_quirk(
"`published` is kept as the SOURCE STRING, unparsed — and the two dialects \
disagree: RSS `<pubDate>` is RFC-822 (`Tue, 13 Jul 2026 10:15:00 +0200`) \
while Atom `<published>`/`<updated>` is RFC-3339. Downstream owns the \
date parsing and must handle both, because one feed URL can change \
dialect on you.",
)
.with_quirk(
"Item identity is best-effort: `id` is the RSS `<guid>` or the Atom \
`<id>`, and plenty of agency feeds ship neither. `join_keys` therefore \
lists `id` AND `link`, and a feed with neither cannot be de-duplicated \
across runs — check for a populated `id` column before relying on CDC.",
)
.with_quirk(
"The fetch is CONDITIONAL but the READ is not incremental: a `304` costs \
nothing, and a `200` re-reads the whole feed window the publisher happens \
to be serving. Items that fell off the end of that window are gone. \
Measured on regeringen.se 2026-09-05: the press feed's 100-item page 1 \
spans 2026-08-28 → 2026-09-04, about 14 items a day, so a 15-minute \
cadence has ~7 days of slack — but the propositions feed's 100 items span \
six months, and a feed with a busier day will not.",
)
.with_quirk(
"regeringen.se paginates with `&page=N` and this collector does NOT follow \
it — page 1 only. The archive behind it is deep: the unfiltered feed \
reaches page 511 (57 items on the last), ~51 057 documents, measured \
2026-09-05. A BACKFILL therefore needs the page parameter walked by hand; \
a running poll does not.",
)
.with_quirk(
"The channel `<title>` is not always an identity. All five Regeringen \
feeds publish `RSS - Regeringen.se`, so `feed_title` cannot tell a \
proposition row from a press-release row — read `feed_link`, which is the \
channel `<link>` and echoes the taxonomy id. Regeringen also publishes no \
`<author>` at all (0 of 100 items, 2026-09-05), so that column is NULL \
for every Regeringen row.",
),
)
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(
SourceEndpoint::Feed,
REGERINGEN_PRESSMEDDELANDEN_FEED,
"feed_items",
)
}
/// **A feed is a MUTABLE DOCUMENT** — one URL, rewritten in place, with validators.
///
/// Measured against regeringen.se on 2026-09-05:
///
/// * `last-modified: Fri, 04 Sep 2026 20:36:17 GMT` — and it is exact, equal to the
/// newest item's `<pubDate>` to the second. No `ETag`; `cache-control:
/// public,max-age=60`.
/// * sending that back as `If-Modified-Since` → **HTTP 304, 0 bytes**, against
/// **111 662 bytes** for the unconditional GET.
///
/// Before this the collector declared nothing, so
/// [`Freshness::derived`](crate::freshness::Freshness::derived) over an absent
/// incremental key made it `AlwaysFull` — a 24-hour cadence for a press wire, and
/// 111 KB transferred every time.
///
/// Both validators are declared, not just the one regeringen.se serves, because this
/// is a GENERIC reader: it stores and replays whatever the CONFIGURED feed answered
/// with, and a feed behind a CDN that serves an `ETag` gets the same saving through
/// the same code. A feed serving neither simply never sees a `304` — the run is
/// correct, only not cheap.
fn freshness(&self) -> crate::freshness::Freshness {
crate::freshness::Freshness::conditional()
}
/// **No graph — and that is the answer, not an omission.**
///
/// The rows really do carry a relationship: a Regeringen item names its department
/// and its document type as taxonomy nodes (`category_ids` `1287`
/// Justitiedepartementet, `1334` Pressmeddelande), and those are honest edges. But
/// THIS collector is a generic syndication reader — the same code answers for SMHI's
/// warnings and Krisinformation's alerts, whose `category_ids` mean something else
/// entirely — so a mapping declared here would assert Regeringen's taxonomy over
/// every feed anyone points it at. A per-feed graph belongs to a per-feed collector
/// or to korp's own config, never to the shared reader.
fn graph_mapping(&self) -> Option<crate::graph_mapping::GraphMapping> {
None
}
/// **One row per feed entry** — the entry's own `guid`/`id`.
///
/// The derived key also named `link`, and a link is not identity: a publisher that
/// moves a proposition from one URL to another (Regeringen does, when a document
/// gets its final beteckning) would land the same entry twice.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["id"],
"one row per entry — the feed's own `guid`/`id`; `link` is where the entry \
currently lives and can move",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(feed_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
use crate::freshness::Plan;
let http = cfg.http(10);
let auth = cfg.auth.resolve_async().await?;
// The validators the last read stored, when the runner planned a conditional
// fetch. Absent on the first run — there is nothing to send yet — and the
// request then goes out plain and comes back with the validators to keep.
let plan = bounds
.plan
.clone()
.unwrap_or_else(|| Plan::of(&self.freshness(), None));
let (if_none_match, if_modified_since) = match &plan {
Plan::Conditional {
if_none_match,
if_modified_since,
} => (if_none_match.clone(), if_modified_since.clone()),
_ => (None, None),
};
let fetched = http
.get_conditional(
&cfg.base_url,
auth.as_ref(),
if_none_match.as_deref(),
if_modified_since.as_deref(),
)
.await?;
let transferred = fetched.transferred();
let (etag, last_modified) = fetched.validators();
// ── 304: the publisher PROVED the feed did not change ────────────────
//
// Nothing is re-landed and nothing is red. This is the whole point of the
// conditional fetch, and on regeringen.se it is 0 bytes against 111 662.
let Some(xml) = fetched.text() else {
*self.last_read.lock().expect("checkpoint mutex") =
Some(Checkpoint::default().unchanged(transferred));
return Ok(CollectStats {
table: cfg.table.clone(),
reason: Some(crate::freshness::unchanged_reason(
"rss-atom",
"304 Not Modified",
)),
..Default::default()
});
};
let ck = Checkpoint::default()
.with_validators(etag, last_modified)
.with_bytes(transferred);
let items = parse_feed(&xml)?;
if items.is_empty() {
// The validators are kept even here: the feed WAS read, and the next run
// should still be able to ask conditionally rather than re-transfer a
// document that is empty for the publisher's own reasons.
*self.last_read.lock().expect("checkpoint mutex") = Some(ck);
return Ok(CollectStats {
table: cfg.table.clone(),
reason: Some("feed carried no item/entry elements (empty or non-feed URL?)".into()),
..Default::default()
});
}
*self.last_read.lock().expect("checkpoint mutex") = Some(ck);
let n = items.len() as u64;
sink.push(feed_batch(&items)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
/// What the last [`collect`](Collector::collect) read — the validators the feed
/// served, the bytes it cost, and whether it proved unchanged. `None` until a
/// collect has run.
///
/// This used to be an unconditional `None`, which is what made the conditional GET
/// impossible: with no validators reported there is nothing for the runner to store
/// and nothing for the next run to send back. Driven red 2026-09-05: with `Ok(None)`
/// here, run 2 plans `Full` rather than `Conditional` and `feed_items` holds **4**
/// rows after two runs of a two-item feed.
async fn checkpoint(&self) -> Result<Option<Checkpoint>> {
Ok(self.last_read.lock().expect("checkpoint mutex").clone())
}
}
#[cfg(test)]
mod tests {
use super::*;
/// An RSS 2.0 channel with two `<item>`s flattens to two rows: the channel
/// title rides every row, `guid`→id, CDATA `<description>`→summary, `<author>`
/// / `dc:creator`→author, multiple `<category>`→a JSON array. Offline DATA
/// contract (no network) + a testmatrix cell.
#[test]
fn rss2_items_flatten_to_feed_rows() {
let xml = r#"<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0" xmlns:dc="http://purl.org/dc/elements/1.1/">
<channel>
<title>Krisinformation</title>
<link>https://www.krisinformation.se</link>
<description>Official alerts</description>
<item>
<title>Flood warning, Dalälven</title>
<link>https://www.krisinformation.se/en/1</link>
<description><![CDATA[High water levels <b>expected</b>.]]></description>
<pubDate>Mon, 14 Jul 2026 08:00:00 GMT</pubDate>
<guid>https://www.krisinformation.se/en/1</guid>
<author>vmr@krisinformation.se</author>
<category>Weather</category>
<category>Flood</category>
</item>
<item>
<title>Power outage, Norrland</title>
<link>https://www.krisinformation.se/en/2</link>
<description>Restoration underway.</description>
<pubDate>Mon, 14 Jul 2026 09:30:00 GMT</pubDate>
<guid>tag:kris,2</guid>
<dc:creator>Elnät</dc:creator>
<category>Power</category>
</item>
</channel>
</rss>"#;
let items = parse_feed(xml).expect("parse rss");
assert_eq!(items.len(), 2, "two <item> → two rows");
let a = &items[0];
assert_eq!(
a.feed_title.as_deref(),
Some("Krisinformation"),
"channel title rides the row"
);
assert_eq!(a.title.as_deref(), Some("Flood warning, Dalälven"));
assert_eq!(
a.id.as_deref(),
Some("https://www.krisinformation.se/en/1"),
"guid → id"
);
assert_eq!(
a.link.as_deref(),
Some("https://www.krisinformation.se/en/1")
);
assert_eq!(
a.summary.as_deref(),
Some("High water levels <b>expected</b>."),
"CDATA body verbatim"
);
assert_eq!(
a.published.as_deref(),
Some("Mon, 14 Jul 2026 08:00:00 GMT"),
"raw pubDate"
);
assert_eq!(
a.author.as_deref(),
Some("vmr@krisinformation.se"),
"RSS <author>"
);
assert_eq!(
a.categories,
vec!["Weather".to_string(), "Flood".to_string()]
);
let b = &items[1];
assert_eq!(b.id.as_deref(), Some("tag:kris,2"));
assert_eq!(b.author.as_deref(), Some("Elnät"), "dc:creator → author");
assert_eq!(b.categories, vec!["Power".to_string()]);
let rb = feed_batch(&items).expect("feed batch");
assert_eq!(rb.num_rows(), 2);
let sch = rb.schema();
let cols: Vec<&str> = sch.fields().iter().map(|f| f.name().as_str()).collect();
for want in [
"feed_title",
"id",
"title",
"link",
"summary",
"author",
"published",
"categories",
] {
assert!(
cols.contains(&want),
"feed table carries `{want}`: {cols:?}"
);
}
// categories is a JSON array string.
let cats = rb
.column_by_name("categories")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("categories utf8");
assert_eq!(cats.value(0), r#"["Weather","Flood"]"#);
crate::functional_status(
"korp-collectors/rss-atom",
"rss2_flatten",
true,
&format!("{} RSS items → feed rows", rb.num_rows()),
);
}
/// An Atom 1.0 feed: the `<link href=…>` alternate is the item URL (the
/// `rel="self"` link is ignored), the nested `<author><name>` is the author,
/// `<id>`→id, `<summary>`/`<content>`→summary, `<category term=…>` self-closing,
/// and `<updated>`/`<published>` supply the timestamp.
#[test]
fn atom_entries_flatten_with_href_link_and_nested_author() {
let xml = r#"<?xml version="1.0" encoding="utf-8"?>
<feed xmlns="http://www.w3.org/2005/Atom">
<title>SMHI Warnings</title>
<link rel="self" href="https://opendata.smhi.se/atom"/>
<entry>
<title>Class 2 wind warning</title>
<link rel="self" href="https://opendata.smhi.se/atom/self/1"/>
<link rel="alternate" href="https://opendata.smhi.se/warning/1"/>
<id>urn:smhi:warning:1</id>
<summary>Gusts up to 25 m/s along the coast.</summary>
<published>2026-07-14T06:00:00Z</published>
<updated>2026-07-14T06:30:00Z</updated>
<author><name>SMHI</name><email>noreply@smhi.se</email></author>
<category term="Wind"/>
<category term="Coast"/>
</entry>
</feed>"#;
let items = parse_feed(xml).expect("parse atom");
assert_eq!(items.len(), 1, "one <entry> → one row");
let e = &items[0];
assert_eq!(e.feed_title.as_deref(), Some("SMHI Warnings"));
assert_eq!(e.title.as_deref(), Some("Class 2 wind warning"));
assert_eq!(
e.link.as_deref(),
Some("https://opendata.smhi.se/warning/1"),
"the alternate href wins; rel=self is skipped"
);
assert_eq!(e.id.as_deref(), Some("urn:smhi:warning:1"));
assert_eq!(
e.summary.as_deref(),
Some("Gusts up to 25 m/s along the coast.")
);
assert_eq!(e.author.as_deref(), Some("SMHI"), "nested <author><name>");
assert_eq!(
e.published.as_deref(),
Some("2026-07-14T06:00:00Z"),
"published wins over updated"
);
assert_eq!(
e.categories,
vec!["Wind".to_string(), "Coast".to_string()],
"term attrs"
);
let rb = feed_batch(&items).expect("batch");
assert_eq!(rb.num_rows(), 1);
let link = rb
.column_by_name("link")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("link utf8");
assert_eq!(link.value(0), "https://opendata.smhi.se/warning/1");
crate::functional_status(
"korp-collectors/rss-atom",
"atom_flatten",
true,
"Atom entry → feed row (alternate href, nested author, term categories)",
);
}
/// A feed with a channel but no items yields zero rows — the collector maps
/// that to a RAGNARÖK reason (the RED-when-broken empty guard). A non-feed
/// document is likewise empty (not a panic).
#[test]
fn empty_feed_yields_no_items() {
let empty = r#"<rss version="2.0"><channel><title>Empty</title></channel></rss>"#;
assert!(
parse_feed(empty).expect("parse").is_empty(),
"no <item> → no rows"
);
let atom_empty = r#"<feed xmlns="http://www.w3.org/2005/Atom"><title>Empty</title></feed>"#;
assert!(parse_feed(atom_empty).expect("parse").is_empty());
// A document that is not a feed at all parses to nothing, not an error.
assert!(
parse_feed("<html><body>not a feed</body></html>")
.expect("parse")
.is_empty()
);
}
/// **The LIVE regeringen.se press-release feed**, verbatim — BOM, `a10` namespace
/// declaration, `isPermaLink="false"` guid, `domain` category attributes and all.
/// Fetched 2026-09-05 from
/// `…/Filter/RssFeed?…&preFilteredCategories=1334&rootPageReference=0`, trimmed to
/// two items with the descriptions shortened; nothing about the SHAPE is altered,
/// because the shape is the whole subject of the tests below.
///
/// The leading `\u{feff}` is real: the live body begins `EF BB BF 3C 3F 78 6D 6C`
/// and [`crate::http::Fetched::text`] hands it through as a `U+FEFF` character. It
/// is kept here because it is what arrives, but it needs no handling and gets none
/// — measured 2026-09-05, `quick-xml` reports leading non-markup as a `Text` event
/// outside any element and parses the two items regardless. That was worth checking
/// and is not worth a guard: an assertion that stays green when the BOM is replaced
/// by any other stray character (verified) is not testing the BOM.
const REGERINGEN_PRESS_RSS: &str = "\u{feff}<?xml version=\"1.0\" encoding=\"utf-8\"?>
<rss xmlns:a10=\"http://www.w3.org/2005/Atom\" version=\"2.0\">
<channel>
<title>RSS - Regeringen.se</title>
<link>https://www.regeringen.se/Filter/RssFeed?filterType=Taxonomy&filterByType=FilterablePageBase&preFilteredCategories=1334&rootPageReference=0</link>
<description>RSS - Regeringen.se</description>
<item>
<guid isPermaLink=\"false\">792bec3d-f13b-412f-9da7-667f546a841f</guid>
<link>https://www.regeringen.se/pressmeddelanden/2026/09/utredning-ska-se-over-mojligheten/</link>
<category domain=\"1263\">Rättsväsendet</category>
<category domain=\"1287\">Justitiedepartementet</category>
<category domain=\"1334\">Pressmeddelande</category>
<category domain=\"2615\">Gunnar Strömmer</category>
<title>Utredning ska se över möjligheten att stoppa vinster</title>
<description>Regeringen har beslutat om tilläggsdirektiv till utredningen.</description>
<pubDate>Fri, 04 Sep 2026 20:36:17 +0200</pubDate>
<a10:updated>2026-09-04T20:36:17+02:00</a10:updated>
</item>
<item>
<guid isPermaLink=\"false\">1f9338d2-6d18-450e-be3b-1afead8b321c</guid>
<link>https://www.regeringen.se/pressmeddelanden/2026/09/obligatorisk-forskola/</link>
<category domain=\"1276\">Barn- och ungdomsutbildning</category>
<category domain=\"1294\">Utbildningsdepartementet</category>
<category domain=\"1334\">Pressmeddelande</category>
<title>Uppdrag att föreslå obligatorisk förskola</title>
<description>En utredare ges i uppdrag att lämna författningsförslag.</description>
<pubDate>Fri, 04 Sep 2026 10:33:12 +0200</pubDate>
<a10:updated>2026-09-04T10:33:12+02:00</a10:updated>
</item>
</channel>
</rss>";
/// **A Regeringen row must say WHICH of the five feeds it came from, and carry the
/// machine-readable taxonomy — not only the Swedish label.**
///
/// Two measurements from 2026-09-05 force this:
///
/// * All five configured Regeringen feeds (`preFilteredCategories` 1329, 1330,
/// 1331, 1334, 1341) publish the SAME channel `<title>`: `RSS - Regeringen.se`.
/// So `feed_title` — the column whose whole purpose is "a landed row still names
/// its source" — cannot distinguish them. The channel `<link>` can: it echoes the
/// request URL, taxonomy id included.
/// * Every `<category>` carries `domain="NNNN"`, the taxonomy node id (1334
/// Pressmeddelande, 1287 Justitiedepartementet, 1329 Proposition). Dropping it
/// leaves downstream string-matching Swedish prose to find a department.
///
/// **RED before the fix:** neither column exists — `feed_link` and `category_ids`
/// are `None` on the batch, and the `domain` attributes are discarded by the parser.
#[test]
fn a_regeringen_row_carries_its_feed_link_and_taxonomy_ids() {
let items = parse_feed(REGERINGEN_PRESS_RSS).expect("parse");
assert_eq!(items.len(), 2);
let a = &items[0];
assert_eq!(
a.feed_title.as_deref(),
Some("RSS - Regeringen.se"),
"the channel title is the SAME on all five feeds — which is the problem"
);
assert_eq!(
a.feed_link.as_deref(),
Some(
"https://www.regeringen.se/Filter/RssFeed?filterType=Taxonomy\
&filterByType=FilterablePageBase&preFilteredCategories=1334\
&rootPageReference=0"
),
"the channel <link> is the only thing in the document that separates the \
press feed from the propositions feed"
);
assert_eq!(
a.categories,
vec![
"Rättsväsendet".to_string(),
"Justitiedepartementet".to_string(),
"Pressmeddelande".to_string(),
"Gunnar Strömmer".to_string(),
],
);
assert_eq!(
a.category_ids,
vec![
"1263".to_string(),
"1287".to_string(),
"1334".to_string(),
"2615".to_string(),
],
"the `domain` attribute is the taxonomy node id, index-aligned with the \
labels — 1334 is Pressmeddelande, 1287 is Justitiedepartementet"
);
// The identity is the guid, and the guid is NOT the link: `isPermaLink=\"false\"`
// says so in the feed itself, and Regeringen moves a document's URL when it
// gets its final beteckning.
assert_eq!(a.id.as_deref(), Some("792bec3d-f13b-412f-9da7-667f546a841f"));
assert_ne!(a.id.as_deref(), a.link.as_deref(), "a guid is not a link");
assert_eq!(
a.published.as_deref(),
Some("Fri, 04 Sep 2026 20:36:17 +0200"),
"RFC-822 pubDate wins over the a10:updated RFC-3339 twin"
);
assert_eq!(a.author, None, "regeringen.se publishes no <author> at all");
// …and the columns really land.
let rb = feed_batch(&items).expect("batch");
assert_eq!(rb.num_rows(), 2);
let s = |name: &str| -> arrow_array::StringArray {
rb.column_by_name(name)
.unwrap_or_else(|| panic!("no `{name}` column"))
.as_any()
.downcast_ref::<arrow_array::StringArray>()
.unwrap_or_else(|| panic!("`{name}` is not Utf8"))
.clone()
};
assert!(
s("feed_link").value(0).contains("preFilteredCategories=1334"),
"the landed row names its feed"
);
assert_eq!(s("category_ids").value(1), r#"["1276","1294","1334"]"#);
assert_eq!(
s("categories").value(1),
r#"["Barn- och ungdomsutbildning","Utbildningsdepartementet","Pressmeddelande"]"#
);
crate::functional_status(
"korp-collectors/rss-atom",
"regeringen_taxonomy",
true,
"regeringen.se press feed → rows carrying feed_link + category_ids",
);
}
/// **The collector must declare the kind it actually is.**
///
/// A syndication feed is a MUTABLE DOCUMENT: one URL, rewritten in place, with
/// validators. Measured against regeringen.se on 2026-09-05:
/// `last-modified: Fri, 04 Sep 2026 20:36:17 GMT` (no `ETag`), and sending it back
/// as `If-Modified-Since` answers **HTTP 304, 0 bytes** where the unconditional GET
/// transfers **111 662**.
///
/// **RED before the fix:** `freshness()` fell through to the derived default, which
/// with no declared incremental key is `AlwaysFull` — `key()` is `"always_full"`,
/// `saves_bytes()` is `false`, and `default_refresh_secs()` is 86 400. A press wire
/// polled once a day, re-reading 111 KB each time it is polled.
#[test]
fn the_feed_declares_itself_a_mutable_document_and_not_a_full_re_read() {
let f = RssFeedCollector::default().freshness();
assert_eq!(f.key(), "mutable_document", "a feed is one URL rewritten in place");
assert!(
f.saves_bytes(),
"…and the conditional GET is a REAL saving: 0 B on a 304 against 111 662 B \
unconditional (regeringen.se, measured 2026-09-05)"
);
assert_eq!(
f.default_refresh_secs(),
900,
"15 minutes — the MutableDocument cadence; 86 400 would be the AlwaysFull \
default this collector used to fall through to"
);
assert_eq!(f.undeclared(), None, "the validators are named, not left blank");
}
}