rama_http/protocols/rss/atom/
read.rs

1//! Async streaming Atom 1.0 reader.
2//!
3//! Same shape as the RSS 2.0 reader: header is read at construction time,
4//! [`AtomEntry`]s stream after.
5
6use std::pin::Pin;
7use std::task::{Context, Poll};
8
9use jiff::Timestamp;
10use quick_xml::NsReader;
11use quick_xml::events::Event;
12use rama_core::futures::Stream;
13use rama_core::futures::StreamExt as _;
14use rama_core::futures::async_stream::stream_fn;
15use rama_core::futures::stream::BoxStream;
16use rama_core::telemetry::tracing;
17use rama_net::uri::Uri;
18use tokio::io::AsyncBufRead;
19
20use super::names::elem;
21use crate::protocols::rss::atom::{
22    AtomCategory, AtomContent, AtomEntry, AtomFeed, AtomGenerator, AtomLink, AtomPerson,
23    AtomSource, AtomText,
24};
25use crate::protocols::rss::error::{AtomCollectError, CollectError, FeedParseError};
26use crate::protocols::rss::feed_ext::FeedExtensions;
27use crate::protocols::rss::feed_ext::names::attr;
28use crate::protocols::rss::feed_ext::parse::{FeedExtAcc, ItemExtAcc, Ns, classify_ns};
29use crate::protocols::rss::parse_util::{
30    atom_category_from_attrs, atom_link_from_attrs, attr_uri_reference, attr_value,
31    end_event_parts, make_atom_text, parse_rfc3339_lax, parse_uri, parse_uri_reference,
32    push_general_ref, push_text,
33};
34
35/// Feed-level metadata of an Atom 1.0 document — everything an [`AtomFeed`]
36/// carries except its `entries`.
37#[derive(Debug, Clone, PartialEq)]
38pub struct AtomHeader {
39    pub id: Uri,
40    pub title: AtomText,
41    pub updated: Timestamp,
42    pub authors: Vec<AtomPerson>,
43    pub links: Vec<AtomLink>,
44    pub categories: Vec<AtomCategory>,
45    pub contributors: Vec<AtomPerson>,
46    pub generator: Option<AtomGenerator>,
47    pub icon: Option<Uri>,
48    pub logo: Option<Uri>,
49    pub rights: Option<AtomText>,
50    pub subtitle: Option<AtomText>,
51    pub extensions: FeedExtensions,
52}
53
54impl Default for AtomHeader {
55    fn default() -> Self {
56        Self {
57            id: Uri::from_static("urn:rama:missing"),
58            title: AtomText::text(""),
59            updated: Timestamp::UNIX_EPOCH,
60            authors: Vec::new(),
61            links: Vec::new(),
62            categories: Vec::new(),
63            contributors: Vec::new(),
64            generator: None,
65            icon: None,
66            logo: None,
67            rights: None,
68            subtitle: None,
69            extensions: FeedExtensions::default(),
70        }
71    }
72}
73
74impl AtomHeader {
75    /// Combine this feed header with an iterator of entries into a full
76    /// [`AtomFeed`].
77    #[must_use]
78    pub fn into_feed_with_entries<I>(self, entries: I) -> AtomFeed
79    where
80        I: IntoIterator<Item = AtomEntry>,
81    {
82        AtomFeed {
83            id: self.id,
84            title: self.title,
85            updated: self.updated,
86            authors: self.authors,
87            links: self.links,
88            categories: self.categories,
89            contributors: self.contributors,
90            generator: self.generator,
91            icon: self.icon,
92            logo: self.logo,
93            rights: self.rights,
94            subtitle: self.subtitle,
95            entries: entries.into_iter().collect(),
96            extensions: self.extensions,
97        }
98    }
99}
100
101/// Async streaming reader for an Atom 1.0 feed.
102pub struct AtomFeedStream {
103    header: AtomHeader,
104    entries: BoxStream<'static, Result<AtomEntry, FeedParseError>>,
105}
106
107impl AtomFeedStream {
108    pub async fn new<R>(reader: R) -> Result<Self, FeedParseError>
109    where
110        R: AsyncBufRead + Unpin + Send + 'static,
111    {
112        Self::new_with_mode(reader, false).await
113    }
114
115    pub async fn new_strict<R>(reader: R) -> Result<Self, FeedParseError>
116    where
117        R: AsyncBufRead + Unpin + Send + 'static,
118    {
119        Self::new_with_mode(reader, true).await
120    }
121
122    pub(in crate::protocols::rss) async fn new_with_mode<R>(
123        reader: R,
124        strict: bool,
125    ) -> Result<Self, FeedParseError>
126    where
127        R: AsyncBufRead + Unpin + Send + 'static,
128    {
129        let mut state = AtomReader::new(reader, strict);
130        let header = state.read_header().await?;
131        let entries: BoxStream<'static, Result<AtomEntry, FeedParseError>> =
132            Box::pin(stream_fn(move |mut yielder| async move {
133                let mut state = state;
134                loop {
135                    match state.read_next_entry().await {
136                        Ok(Some(entry)) => yielder.yield_item(Ok(entry)).await,
137                        Ok(None) => return,
138                        Err(e) => {
139                            yielder.yield_item(Err(e)).await;
140                            return;
141                        }
142                    }
143                }
144            }));
145        Ok(Self { header, entries })
146    }
147
148    /// Borrow the feed-level metadata parsed at construction time.
149    #[must_use]
150    pub fn header(&self) -> &AtomHeader {
151        &self.header
152    }
153
154    /// Split into `(header, entries)`.
155    #[must_use]
156    pub fn drain(
157        self,
158    ) -> (
159        AtomHeader,
160        BoxStream<'static, Result<AtomEntry, FeedParseError>>,
161    ) {
162        (self.header, self.entries)
163    }
164
165    /// Drain into a full [`AtomFeed`]; on per-entry error returns a partial
166    /// feed with everything parsed so far.
167    pub async fn collect(mut self) -> Result<AtomFeed, AtomCollectError> {
168        let mut entries = Vec::new();
169        while let Some(entry) = self.entries.next().await {
170            match entry {
171                Ok(e) => entries.push(e),
172                Err(error) => {
173                    return Err(CollectError {
174                        error,
175                        partial: self.header.into_feed_with_entries(entries),
176                    });
177                }
178            }
179        }
180        Ok(self.header.into_feed_with_entries(entries))
181    }
182
183    /// Drain, silently dropping (and `tracing::debug!`-logging) entries that
184    /// fail to parse.
185    pub async fn collect_lossy(mut self) -> AtomFeed {
186        let mut entries = Vec::new();
187        while let Some(entry) = self.entries.next().await {
188            match entry {
189                Ok(e) => entries.push(e),
190                Err(err) => tracing::debug!(error = %err, "atom entry dropped by collect_lossy"),
191            }
192        }
193        self.header.into_feed_with_entries(entries)
194    }
195
196    /// Drain into a feed retaining only entries the predicate accepts.
197    pub async fn collect_filtered<F>(
198        mut self,
199        mut predicate: F,
200    ) -> Result<AtomFeed, AtomCollectError>
201    where
202        F: FnMut(&AtomEntry) -> bool + Send,
203    {
204        let mut entries = Vec::new();
205        while let Some(entry) = self.entries.next().await {
206            match entry {
207                Ok(e) => {
208                    if predicate(&e) {
209                        entries.push(e);
210                    }
211                }
212                Err(error) => {
213                    return Err(CollectError {
214                        error,
215                        partial: self.header.into_feed_with_entries(entries),
216                    });
217                }
218            }
219        }
220        Ok(self.header.into_feed_with_entries(entries))
221    }
222}
223
224impl Stream for AtomFeedStream {
225    type Item = Result<AtomEntry, FeedParseError>;
226    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
227        let this = Pin::into_inner(self);
228        this.entries.poll_next_unpin(cx)
229    }
230}
231
232// ---------------------------------------------------------------------------
233// AtomReader: state machine. Private.
234// ---------------------------------------------------------------------------
235
236enum Action {
237    Continue,
238    FirstEntryStarted,
239    /// Boxed so the enum stays small (a finished `AtomEntry` is ~1.3 KB).
240    EntryFinished(Box<AtomEntry>),
241    Eof,
242}
243
244struct AtomReader<R: AsyncBufRead + Unpin + Send> {
245    nsr: NsReader<R>,
246    buf: Vec<u8>,
247    strict: bool,
248
249    text_buf: String,
250    depth: i32,
251    saw_root: bool,
252    feed_id_set: bool,
253    feed_updated_parsed: bool,
254
255    // Feed-level header accumulator.
256    header: AtomHeader,
257    feed_acc: FeedExtAcc,
258    pending_generator: Option<AtomGenerator>,
259
260    // Entry / sub-element state.
261    in_entry: bool,
262    current_entry: AtomEntry,
263    current_entry_id_set: bool,
264    current_entry_title_set: bool,
265    current_entry_updated_parsed: bool,
266    entry_acc: ItemExtAcc,
267    in_author: bool,
268    in_feed_author: bool,
269    in_contributor: bool,
270    in_feed_contributor: bool,
271    current_author: AtomPerson,
272    current_contributor: AtomPerson,
273    /// Nesting depth of open `<atom:source>` elements. Zero means we are
274    /// not currently inside a source. The outermost source is depth 1; a
275    /// (malformed) nested `<source>` inside another bumps to 2, 3, … so the
276    /// inner `</source>` does not prematurely finalise the outer one.
277    /// Only depth-1 children mutate [`Self::current_source`]; deeper ones
278    /// are silently dropped (or in strict mode rejected at the Start).
279    source_depth: u32,
280    current_source: AtomSource,
281    /// Type attribute of the source's `<title>` — kept separate from the
282    /// outer entry's `current_title_type` so a `<source><title type="html">`
283    /// can't leak its type back to a still-open outer `<title>`.
284    current_source_title_type: String,
285
286    current_title_type: String,
287    current_summary_type: String,
288    current_content_type: String,
289    current_rights_type: String,
290    current_subtitle_type: String,
291}
292
293impl<R: AsyncBufRead + Unpin + Send> AtomReader<R> {
294    fn new(reader: R, strict: bool) -> Self {
295        let mut nsr = NsReader::from_reader(reader);
296        // Do NOT use `trim_text(true)`: quick-xml 0.40 splits a text run around
297        // every entity / character reference into separate `Text` and
298        // `GeneralRef` events, so per-event trimming strips whitespace that is
299        // *interior* to a field — e.g. `Hello &lt;b&gt;` (a `type="html"` body)
300        // would lose the space before `<b>`. We instead leave text verbatim and
301        // trim the fully reassembled field value once, in the `End` handler.
302        nsr.config_mut().trim_text(false);
303        Self {
304            nsr,
305            buf: Vec::with_capacity(4096),
306            strict,
307            text_buf: String::new(),
308            depth: 0,
309            saw_root: false,
310            feed_id_set: false,
311            feed_updated_parsed: false,
312            header: AtomHeader::default(),
313            feed_acc: FeedExtAcc::default(),
314            pending_generator: None,
315            in_entry: false,
316            current_entry: AtomEntry::new(
317                Uri::from_static("urn:rama:missing"),
318                AtomText::text(""),
319                Timestamp::UNIX_EPOCH,
320            ),
321            current_entry_id_set: false,
322            current_entry_title_set: false,
323            current_entry_updated_parsed: false,
324            entry_acc: ItemExtAcc::default(),
325            in_author: false,
326            in_feed_author: false,
327            in_contributor: false,
328            in_feed_contributor: false,
329            current_author: AtomPerson::new(""),
330            current_contributor: AtomPerson::new(""),
331            source_depth: 0,
332            current_source: AtomSource {
333                id: None,
334                title: None,
335                updated: None,
336            },
337            current_source_title_type: String::from("text"),
338            current_title_type: String::from("text"),
339            current_summary_type: String::from("text"),
340            current_content_type: String::from("text"),
341            current_rights_type: String::from("text"),
342            current_subtitle_type: String::from("text"),
343        }
344    }
345
346    async fn read_header(&mut self) -> Result<AtomHeader, FeedParseError> {
347        loop {
348            match self.step().await? {
349                Action::Continue => {}
350                Action::FirstEntryStarted | Action::Eof => return self.take_header(),
351                Action::EntryFinished(_) => {
352                    return Err(FeedParseError::new(
353                        "internal: entry finished during header phase",
354                    ));
355                }
356            }
357        }
358    }
359
360    async fn read_next_entry(&mut self) -> Result<Option<AtomEntry>, FeedParseError> {
361        loop {
362            match self.step().await? {
363                Action::Continue | Action::FirstEntryStarted => {}
364                Action::EntryFinished(entry) => return Ok(Some(*entry)),
365                Action::Eof => return Ok(None),
366            }
367        }
368    }
369
370    fn take_header(&mut self) -> Result<AtomHeader, FeedParseError> {
371        if !self.saw_root {
372            return Err(FeedParseError::new("no <feed> root encountered"));
373        }
374        let mut header = std::mem::take(&mut self.header);
375        header.extensions = std::mem::take(&mut self.feed_acc).finish();
376        if self.strict {
377            if !self.feed_id_set {
378                return Err(FeedParseError::new(
379                    "Atom feed missing required or valid URI <id>",
380                ));
381            }
382            if header.title.value.is_empty() {
383                return Err(FeedParseError::new("Atom feed missing required <title>"));
384            }
385            if !self.feed_updated_parsed {
386                return Err(FeedParseError::new("Atom feed missing required <updated>"));
387            }
388        }
389        Ok(header)
390    }
391
392    /// RFC 4287 §3.2: an Atom Person construct contains exactly `<name>`,
393    /// optionally `<uri>` and `<email>`. Anything else is malformed and must
394    /// NOT leak side effects into the enclosing entry/feed. Used by both the
395    /// Start and Empty arms of [`Self::step`]; the `in_person` boolean is
396    /// inlined at the call sites (a method call would borrow `self` as a
397    /// unit and clash with the `self.buf` borrow held by the current event).
398    fn person_child_is_valid(local: &str) -> bool {
399        matches!(local, elem::NAME | elem::URI | elem::EMAIL)
400    }
401
402    async fn step(&mut self) -> Result<Action, FeedParseError> {
403        self.buf.clear();
404        let (rr, ev) = match self.nsr.read_resolved_event_into_async(&mut self.buf).await {
405            Ok(p) => p,
406            Err(e) => {
407                if self.strict {
408                    return Err(FeedParseError::new(format!("xml error: {e}")));
409                }
410                tracing::debug!("atom stream xml error (lenient): {e}");
411                return Ok(Action::Eof);
412            }
413        };
414
415        match ev {
416            Event::Start(e) => {
417                self.depth += 1;
418                let ns = classify_ns(&rr);
419                let local_name = e.local_name();
420                let local = std::str::from_utf8(local_name.as_ref()).unwrap_or("");
421                self.text_buf.clear();
422
423                // Person-construct containment: any non-{name,uri,email}
424                // child of an <author>/<contributor> must not mutate the
425                // enclosing entry/feed state (link/category/content/typed
426                // text/etc.) Strict mode rejects; lenient mode swallows.
427                // For type="xhtml" typed-text we still consume the inner
428                // subtree so the parser depth stays in sync.
429                // (Inline the four-bool check rather than calling
430                // `self.in_person()` — the latter would borrow `self` as
431                // a unit and clash with the `self.buf` borrow held by `ev`.
432                // Reading individual bool fields is a split borrow.)
433                let in_person = self.in_author
434                    || self.in_feed_author
435                    || self.in_contributor
436                    || self.in_feed_contributor;
437                if in_person && !Self::person_child_is_valid(local) {
438                    if self.strict {
439                        return Err(FeedParseError::new(format!(
440                            "Atom person element may only contain <name>/<uri>/<email>, \
441                             found <{local}>"
442                        )));
443                    }
444                    if matches!(
445                        local,
446                        elem::TITLE | elem::SUMMARY | elem::CONTENT | elem::RIGHTS | elem::SUBTITLE
447                    ) && attr_value(&e, attr::TYPE).as_deref() == Some("xhtml")
448                    {
449                        drop(e);
450                        let _ =
451                            capture_xhtml_subtree_async(&mut self.nsr, &mut self.buf, self.strict)
452                                .await?;
453                        self.depth -= 1;
454                    }
455                    return Ok(Action::Continue);
456                }
457
458                let consumed = if self.in_entry {
459                    self.entry_acc.on_start(ns, local, &e)
460                } else {
461                    self.feed_acc.on_start(ns, local, &e)
462                };
463                if consumed {
464                    return Ok(Action::Continue);
465                }
466                if ns != Ns::Atom {
467                    return Ok(Action::Continue);
468                }
469
470                match local {
471                    elem::FEED => {
472                        self.saw_root = true;
473                        Ok(Action::Continue)
474                    }
475                    elem::ENTRY => {
476                        let first_entry = !self.in_entry;
477                        if !first_entry {
478                            // Nested / re-opened <entry> in malformed input.
479                            // Strict rejects (matches RSS behaviour). Lenient
480                            // resets and keeps going; the outer partial entry
481                            // is discarded — trace so operators can spot it.
482                            if self.strict {
483                                return Err(FeedParseError::new(format!(
484                                    "Atom: nested or re-opened <entry> at depth {}",
485                                    self.depth,
486                                )));
487                            }
488                            tracing::debug!(
489                                "atom: nested or re-opened <entry> at depth {} — \
490                                 partial outer entry discarded",
491                                self.depth,
492                            );
493                        }
494                        self.in_entry = true;
495                        self.current_entry = AtomEntry::new(
496                            Uri::from_static("urn:rama:missing"),
497                            AtomText::text(""),
498                            Timestamp::UNIX_EPOCH,
499                        );
500                        self.current_entry_id_set = false;
501                        self.current_entry_title_set = false;
502                        self.current_entry_updated_parsed = false;
503                        self.entry_acc = ItemExtAcc::default();
504                        if first_entry {
505                            Ok(Action::FirstEntryStarted)
506                        } else {
507                            Ok(Action::Continue)
508                        }
509                    }
510                    elem::AUTHOR if self.source_depth == 0 => {
511                        self.current_author = AtomPerson::new("");
512                        if self.in_entry {
513                            self.in_author = true;
514                        } else {
515                            self.in_feed_author = true;
516                        }
517                        Ok(Action::Continue)
518                    }
519                    elem::CONTRIBUTOR if self.source_depth == 0 => {
520                        self.current_contributor = AtomPerson::new("");
521                        if self.in_entry {
522                            self.in_contributor = true;
523                        } else {
524                            self.in_feed_contributor = true;
525                        }
526                        Ok(Action::Continue)
527                    }
528                    elem::SOURCE if self.in_entry => {
529                        self.source_depth += 1;
530                        if self.source_depth == 1 {
531                            self.current_source = AtomSource {
532                                id: None,
533                                title: None,
534                                updated: None,
535                            };
536                            self.current_source_title_type = String::from("text");
537                        } else if self.strict {
538                            return Err(FeedParseError::new(
539                                "Atom <source> may not be nested inside another <source>",
540                            ));
541                        }
542                        Ok(Action::Continue)
543                    }
544                    elem::LINK if self.source_depth == 0 => {
545                        if let Some(link) = atom_link_from_attrs(&e) {
546                            if self.in_entry {
547                                self.current_entry.links.push(link);
548                            } else {
549                                self.header.links.push(link);
550                            }
551                        }
552                        Ok(Action::Continue)
553                    }
554                    elem::CATEGORY if self.source_depth == 0 => {
555                        let cat = atom_category_from_attrs(&e);
556                        if self.in_entry {
557                            self.current_entry.categories.push(cat);
558                        } else {
559                            self.header.categories.push(cat);
560                        }
561                        Ok(Action::Continue)
562                    }
563                    elem::TITLE => {
564                        let t = attr_value(&e, attr::TYPE).unwrap_or_else(|| "text".into());
565                        drop(e);
566                        self.start_typed_text(elem::TITLE, t).await
567                    }
568                    elem::SUMMARY if self.in_entry => {
569                        let t = attr_value(&e, attr::TYPE).unwrap_or_else(|| "text".into());
570                        drop(e);
571                        self.start_typed_text(elem::SUMMARY, t).await
572                    }
573                    elem::CONTENT if self.in_entry && self.source_depth == 0 => {
574                        let t = attr_value(&e, attr::TYPE).unwrap_or_else(|| "text".into());
575                        drop(e);
576                        self.start_typed_text(elem::CONTENT, t).await
577                    }
578                    elem::RIGHTS => {
579                        let t = attr_value(&e, attr::TYPE).unwrap_or_else(|| "text".into());
580                        drop(e);
581                        self.start_typed_text(elem::RIGHTS, t).await
582                    }
583                    elem::SUBTITLE if !self.in_entry => {
584                        let t = attr_value(&e, attr::TYPE).unwrap_or_else(|| "text".into());
585                        drop(e);
586                        self.start_typed_text(elem::SUBTITLE, t).await
587                    }
588                    elem::GENERATOR if self.source_depth == 0 => {
589                        self.pending_generator = Some(AtomGenerator {
590                            value: String::new(),
591                            uri: attr_uri_reference(&e, attr::URI),
592                            version: attr_value(&e, attr::VERSION),
593                        });
594                        Ok(Action::Continue)
595                    }
596                    _ => Ok(Action::Continue),
597                }
598            }
599            Event::Empty(e) => {
600                let ns = classify_ns(&rr);
601                let local_name = e.local_name();
602                let local = std::str::from_utf8(local_name.as_ref()).unwrap_or("");
603
604                // Same person-construct containment as for Start events.
605                let in_person = self.in_author
606                    || self.in_feed_author
607                    || self.in_contributor
608                    || self.in_feed_contributor;
609                if in_person && !Self::person_child_is_valid(local) {
610                    if self.strict {
611                        return Err(FeedParseError::new(format!(
612                            "Atom person element may only contain <name>/<uri>/<email>, \
613                             found <{local}/>"
614                        )));
615                    }
616                    return Ok(Action::Continue);
617                }
618
619                let consumed = if self.in_entry {
620                    self.entry_acc.on_empty(ns, local, &e)
621                } else {
622                    self.feed_acc.on_empty(ns, local, &e)
623                };
624                if consumed || ns != Ns::Atom {
625                    return Ok(Action::Continue);
626                }
627                match local {
628                    elem::LINK if self.source_depth == 0 => {
629                        if let Some(link) = atom_link_from_attrs(&e) {
630                            if self.in_entry {
631                                self.current_entry.links.push(link);
632                            } else {
633                                self.header.links.push(link);
634                            }
635                        }
636                    }
637                    elem::CATEGORY if self.source_depth == 0 => {
638                        let cat = atom_category_from_attrs(&e);
639                        if self.in_entry {
640                            self.current_entry.categories.push(cat);
641                        } else {
642                            self.header.categories.push(cat);
643                        }
644                    }
645                    elem::CONTENT if self.in_entry && self.source_depth == 0 => {
646                        // Out-of-line <content src=".." type=".."/>.
647                        // A self-closing <content type="html"/> with no src
648                        // is degenerate but valid; we drop it (no body, no
649                        // link — nothing useful to store).
650                        if let Some(src) = attr_uri_reference(&e, attr::SRC) {
651                            let type_ = attr_value(&e, attr::TYPE).unwrap_or_else(|| "text".into());
652                            self.current_entry.content = Some(AtomContent::out_of_line(src, type_));
653                        }
654                    }
655                    _ => {}
656                }
657                Ok(Action::Continue)
658            }
659            Event::Text(e) => {
660                push_text(&mut self.text_buf, &e, self.strict)?;
661                Ok(Action::Continue)
662            }
663            // quick-xml 0.40 surfaces entity references as standalone events
664            // rather than expanding them inside `Text`; resolve and accumulate
665            // them so `&amp;` etc. survive into the captured text.
666            Event::GeneralRef(e) => {
667                push_general_ref(&mut self.text_buf, &e, self.strict)?;
668                Ok(Action::Continue)
669            }
670            Event::CData(e) => {
671                match std::str::from_utf8(e.as_ref()) {
672                    Ok(t) => self.text_buf.push_str(t),
673                    Err(err) => {
674                        if self.strict {
675                            return Err(FeedParseError::new(format!("invalid CDATA: {err}")));
676                        }
677                        tracing::debug!("atom stream CDATA utf8 error (lenient): {err}");
678                        self.text_buf.push_str(&String::from_utf8_lossy(e.as_ref()));
679                    }
680                }
681                Ok(Action::Continue)
682            }
683            Event::End(e) => {
684                self.depth -= 1;
685                let ns = classify_ns(&rr);
686                let mut name = [0u8; 64];
687                let (local, text) = end_event_parts(e, &mut name, &mut self.text_buf);
688                self.handle_end(ns, local, text)
689            }
690            Event::Eof => {
691                if self.strict && self.depth > 0 {
692                    return Err(FeedParseError::new(format!(
693                        "truncated Atom document ({} unclosed elements at EOF)",
694                        self.depth
695                    )));
696                }
697                Ok(Action::Eof)
698            }
699            _ => Ok(Action::Continue),
700        }
701    }
702
703    /// Common entry point for the typed text constructs (title/summary/
704    /// content/rights/subtitle). If type is `xhtml`, captures the inner
705    /// subtree directly; otherwise records the type and lets the End handler
706    /// finalise it from `text_buf`.
707    async fn start_typed_text(
708        &mut self,
709        which: &'static str,
710        t: String,
711    ) -> Result<Action, FeedParseError> {
712        if t == "xhtml" {
713            let xml =
714                capture_xhtml_subtree_async(&mut self.nsr, &mut self.buf, self.strict).await?;
715            self.depth -= 1;
716            // Children of `<atom:source>` belong to the source, not the
717            // enclosing entry. The text/html path is intercepted by the
718            // source branch in `handle_end`, but the xhtml path bypasses
719            // that (it consumes events inline and never returns an
720            // `Event::End` to `handle_end`) so we route here explicitly.
721            // AtomSource only carries id/title/updated, so any xhtml-typed
722            // source child other than `<title>` has nowhere to land and
723            // is intentionally dropped. Only the OUTERMOST source's title
724            // is captured — inner (malformed nested) sources' xhtml is
725            // discarded entirely.
726            if self.source_depth > 0 {
727                if self.source_depth == 1 && which == elem::TITLE {
728                    self.current_source.title = Some(AtomText::xhtml(xml));
729                }
730                return Ok(Action::Continue);
731            }
732            match which {
733                elem::TITLE => {
734                    if self.in_entry {
735                        self.current_entry.title = AtomText::xhtml(xml);
736                        self.current_entry_title_set = true;
737                    } else {
738                        self.header.title = AtomText::xhtml(xml);
739                    }
740                }
741                elem::SUMMARY => {
742                    self.current_entry.summary = Some(AtomText::xhtml(xml));
743                }
744                elem::CONTENT => {
745                    self.current_entry.content = Some(AtomContent {
746                        value: AtomText::xhtml(xml),
747                        src: None,
748                        out_of_line_type: None,
749                    });
750                }
751                elem::RIGHTS => {
752                    if self.in_entry {
753                        self.current_entry.rights = Some(AtomText::xhtml(xml));
754                    } else {
755                        self.header.rights = Some(AtomText::xhtml(xml));
756                    }
757                }
758                elem::SUBTITLE => {
759                    self.header.subtitle = Some(AtomText::xhtml(xml));
760                }
761                _ => {}
762            }
763            return Ok(Action::Continue);
764        }
765        // Inside a `<source>` only `<title>` is meaningful (id and updated
766        // carry no type); writing into the outer entry's `current_*_type`
767        // would leak the source's typing back to the entry. Keep the
768        // source's title type isolated and ignore the rest. Inside a
769        // nested (malformed) source the typing is discarded entirely.
770        if self.source_depth > 0 {
771            if self.source_depth == 1 && which == elem::TITLE {
772                self.current_source_title_type = t;
773            }
774            return Ok(Action::Continue);
775        }
776        match which {
777            elem::TITLE => self.current_title_type = t,
778            elem::SUMMARY => self.current_summary_type = t,
779            elem::CONTENT => self.current_content_type = t,
780            elem::RIGHTS => self.current_rights_type = t,
781            elem::SUBTITLE => self.current_subtitle_type = t,
782            _ => {}
783        }
784        Ok(Action::Continue)
785    }
786
787    fn handle_end(&mut self, ns: Ns, local: &str, text: String) -> Result<Action, FeedParseError> {
788        // Author / contributor sub-elements: shallow finalisation.
789        if self.in_author && ns == Ns::Atom {
790            return Ok(self.handle_person_end(local, text, &PersonKind::EntryAuthor));
791        }
792        if self.in_feed_author && ns == Ns::Atom {
793            return Ok(self.handle_person_end(local, text, &PersonKind::FeedAuthor));
794        }
795        if self.in_contributor && ns == Ns::Atom {
796            return Ok(self.handle_person_end(local, text, &PersonKind::EntryContributor));
797        }
798        if self.in_feed_contributor && ns == Ns::Atom {
799            return Ok(self.handle_person_end(local, text, &PersonKind::FeedContributor));
800        }
801        // Source sub-elements: route into current_source, then close on the
802        // outermost </source>. Inner sources (malformed nesting) just
803        // decrement the depth — neither their children nor their close
804        // touch the outer source's accumulated state.
805        if self.source_depth > 0 && ns == Ns::Atom {
806            if local == elem::SOURCE {
807                self.source_depth -= 1;
808                if self.source_depth == 0 {
809                    let source = std::mem::replace(
810                        &mut self.current_source,
811                        AtomSource {
812                            id: None,
813                            title: None,
814                            updated: None,
815                        },
816                    );
817                    self.current_entry.source = Some(source);
818                }
819                return Ok(Action::Continue);
820            }
821            // Only the outermost source's children mutate current_source.
822            if self.source_depth == 1 {
823                match local {
824                    elem::ID => self.current_source.id = parse_uri(&text),
825                    elem::TITLE => {
826                        self.current_source.title =
827                            Some(make_atom_text(&self.current_source_title_type, text));
828                    }
829                    elem::UPDATED => self.current_source.updated = parse_rfc3339_lax(&text),
830                    _ => {}
831                }
832            }
833            return Ok(Action::Continue);
834        }
835        // Entry-level core elements.
836        if self.in_entry {
837            let Some(text) = self.entry_acc.on_end(ns, local, text) else {
838                return Ok(Action::Continue);
839            };
840            if ns != Ns::Atom {
841                return Ok(Action::Continue);
842            }
843            match local {
844                elem::ID => {
845                    if let Some(id) = parse_uri(&text) {
846                        self.current_entry.id = id;
847                        self.current_entry_id_set = true;
848                    } else if self.strict {
849                        return Err(FeedParseError::new(format!(
850                            "Atom entry <id> could not be parsed as URI: {text:?}"
851                        )));
852                    }
853                }
854                elem::TITLE => {
855                    self.current_entry.title = make_atom_text(&self.current_title_type, text);
856                    self.current_entry_title_set = true;
857                }
858                elem::UPDATED => {
859                    if let Some(ts) = parse_rfc3339_lax(&text) {
860                        self.current_entry.updated = ts;
861                        self.current_entry_updated_parsed = true;
862                    } else if self.strict {
863                        return Err(FeedParseError::new(format!(
864                            "Atom entry <updated> could not be parsed as RFC 3339: {text:?}"
865                        )));
866                    }
867                }
868                elem::PUBLISHED => self.current_entry.published = parse_rfc3339_lax(&text),
869                elem::SUMMARY => {
870                    self.current_entry.summary =
871                        Some(make_atom_text(&self.current_summary_type, text));
872                }
873                elem::CONTENT => {
874                    self.current_entry.content = Some(AtomContent {
875                        value: make_atom_text(&self.current_content_type, text),
876                        src: None,
877                        out_of_line_type: None,
878                    });
879                }
880                elem::RIGHTS => {
881                    self.current_entry.rights =
882                        Some(make_atom_text(&self.current_rights_type, text));
883                }
884                elem::ENTRY => {
885                    if self.strict {
886                        if !self.current_entry_id_set {
887                            return Err(FeedParseError::new("Atom entry missing required <id>"));
888                        }
889                        if !self.current_entry_title_set {
890                            return Err(FeedParseError::new("Atom entry missing required <title>"));
891                        }
892                        if !self.current_entry_updated_parsed {
893                            return Err(FeedParseError::new(
894                                "Atom entry missing required <updated>",
895                            ));
896                        }
897                    }
898                    self.current_entry.extensions = std::mem::take(&mut self.entry_acc).finish();
899                    let entry = std::mem::replace(
900                        &mut self.current_entry,
901                        AtomEntry::new(
902                            Uri::from_static("urn:rama:missing"),
903                            AtomText::text(""),
904                            Timestamp::UNIX_EPOCH,
905                        ),
906                    );
907                    self.in_entry = false;
908                    return Ok(Action::EntryFinished(Box::new(entry)));
909                }
910                _ => {}
911            }
912            return Ok(Action::Continue);
913        }
914        // Feed-level core elements.
915        let Some(text) = self.feed_acc.on_end(ns, local, text) else {
916            return Ok(Action::Continue);
917        };
918        if ns != Ns::Atom {
919            return Ok(Action::Continue);
920        }
921        match local {
922            elem::ID => {
923                if let Some(id) = parse_uri(&text) {
924                    self.header.id = id;
925                    self.feed_id_set = true;
926                } else if self.strict {
927                    return Err(FeedParseError::new(format!(
928                        "Atom feed <id> could not be parsed as URI: {text:?}"
929                    )));
930                }
931            }
932            elem::TITLE => self.header.title = make_atom_text(&self.current_title_type, text),
933            elem::UPDATED => {
934                if let Some(ts) = parse_rfc3339_lax(&text) {
935                    self.header.updated = ts;
936                    self.feed_updated_parsed = true;
937                } else if self.strict {
938                    return Err(FeedParseError::new(format!(
939                        "Atom feed <updated> could not be parsed as RFC 3339: {text:?}"
940                    )));
941                }
942            }
943            elem::SUBTITLE => {
944                self.header.subtitle = Some(make_atom_text(&self.current_subtitle_type, text));
945            }
946            elem::RIGHTS => {
947                self.header.rights = Some(make_atom_text(&self.current_rights_type, text));
948            }
949            elem::LOGO => self.header.logo = parse_uri_reference(&text),
950            elem::ICON => self.header.icon = parse_uri_reference(&text),
951            elem::GENERATOR => {
952                if let Some(mut g) = self.pending_generator.take() {
953                    g.value = text;
954                    self.header.generator = Some(g);
955                }
956            }
957            _ => {}
958        }
959        Ok(Action::Continue)
960    }
961
962    fn handle_person_end(&mut self, local: &str, text: String, kind: &PersonKind) -> Action {
963        let person = match kind {
964            PersonKind::EntryAuthor | PersonKind::FeedAuthor => &mut self.current_author,
965            PersonKind::EntryContributor | PersonKind::FeedContributor => {
966                &mut self.current_contributor
967            }
968        };
969        match local {
970            elem::NAME => person.name = text,
971            elem::EMAIL => person.email = Some(text),
972            elem::URI => person.uri = parse_uri_reference(&text),
973            elem::AUTHOR | elem::CONTRIBUTOR => {
974                let finalised = std::mem::replace(person, AtomPerson::new(""));
975                match kind {
976                    PersonKind::EntryAuthor => {
977                        self.current_entry.authors.push(finalised);
978                        self.in_author = false;
979                    }
980                    PersonKind::FeedAuthor => {
981                        self.header.authors.push(finalised);
982                        self.in_feed_author = false;
983                    }
984                    PersonKind::EntryContributor => {
985                        self.current_entry.contributors.push(finalised);
986                        self.in_contributor = false;
987                    }
988                    PersonKind::FeedContributor => {
989                        self.header.contributors.push(finalised);
990                        self.in_feed_contributor = false;
991                    }
992                }
993            }
994            _ => {}
995        }
996        Action::Continue
997    }
998}
999
1000enum PersonKind {
1001    EntryAuthor,
1002    FeedAuthor,
1003    EntryContributor,
1004    FeedContributor,
1005}
1006
1007// ---------------------------------------------------------------------------
1008// XHTML subtree capture (preserved from the previous implementation).
1009// ---------------------------------------------------------------------------
1010
1011/// Consume events from `nsr` until the matching End closes the enclosing
1012/// `type="xhtml"` element, re-emitting child events into a string buffer.
1013///
1014/// RFC 4287 §3.1.1.3 requires the content of an xhtml text construct to be
1015/// **exactly one** XHTML-namespaced `<div>` element wrapping the real
1016/// markup. In `strict` mode this is enforced (missing wrapper, wrong
1017/// namespace, or non-`<div>` first element all fail). In lenient mode we
1018/// keep absorbing whatever the publisher emitted: a first `<div>` (any
1019/// namespace, or none) is treated as the wrapper; otherwise the inner
1020/// content is captured verbatim.
1021///
1022/// On the wire the writer always emits a correctly-namespaced wrapper, so
1023/// a round-tripped feed is always strict-clean.
1024async fn capture_xhtml_subtree_async<R>(
1025    nsr: &mut NsReader<R>,
1026    buf: &mut Vec<u8>,
1027    strict: bool,
1028) -> Result<String, FeedParseError>
1029where
1030    R: AsyncBufRead + Unpin,
1031{
1032    const XHTML_NS_BYTES: &[u8] = crate::protocols::rss::ns::XHTML_NS.as_bytes();
1033
1034    // The outer reader already runs with `trim_text(false)` (see
1035    // `AtomReader::new`), which is exactly what an xhtml subtree needs — inside
1036    // it every space matters: `<p>foo</p> <p>bar</p>` and
1037    // `<p>Hello <em>world</em> there</p>` both lose meaning if inter-element
1038    // whitespace is trimmed. No toggle needed.
1039    capture_xhtml_subtree_inner(nsr, buf, strict, XHTML_NS_BYTES).await
1040}
1041
1042async fn capture_xhtml_subtree_inner<R>(
1043    nsr: &mut NsReader<R>,
1044    buf: &mut Vec<u8>,
1045    strict: bool,
1046    xhtml_ns_bytes: &[u8],
1047) -> Result<String, FeedParseError>
1048where
1049    R: AsyncBufRead + Unpin,
1050{
1051    use quick_xml::Writer;
1052    use quick_xml::name::ResolveResult;
1053
1054    let mut captured = Vec::<u8>::new();
1055    let mut depth: i32 = 0;
1056    let mut saw_wrapper = false;
1057    let mut writer = Writer::new(&mut captured);
1058    loop {
1059        buf.clear();
1060        let (rr, ev) = nsr
1061            .read_resolved_event_into_async(buf)
1062            .await
1063            .map_err(|e| FeedParseError::new(format!("xhtml capture: {e}")))?;
1064        match ev {
1065            Event::Start(e) => {
1066                if depth == 0 && !saw_wrapper {
1067                    let is_div = e.local_name().as_ref() == elem::DIV.as_bytes();
1068                    let in_xhtml_ns =
1069                        matches!(rr, ResolveResult::Bound(n) if n.0 == xhtml_ns_bytes);
1070                    if strict && !(is_div && in_xhtml_ns) {
1071                        return Err(FeedParseError::new(
1072                            "Atom xhtml content must be wrapped in a single \
1073                             XHTML-namespaced <div> (RFC 4287 §3.1.1.3)",
1074                        ));
1075                    }
1076                    if is_div {
1077                        saw_wrapper = true;
1078                        depth += 1;
1079                        continue;
1080                    }
1081                }
1082                depth += 1;
1083                writer
1084                    .write_event(Event::Start(e))
1085                    .map_err(|err| FeedParseError::new(format!("xhtml write: {err}")))?;
1086            }
1087            Event::End(e) => {
1088                if depth == 0 {
1089                    if strict && !saw_wrapper {
1090                        return Err(FeedParseError::new(
1091                            "Atom xhtml content must be wrapped in a single \
1092                             XHTML-namespaced <div> (RFC 4287 §3.1.1.3)",
1093                        ));
1094                    }
1095                    drop(writer);
1096                    return String::from_utf8(captured).map_err(|err| {
1097                        FeedParseError::new(format!("xhtml inner is not utf-8: {err}"))
1098                    });
1099                }
1100                depth -= 1;
1101                if depth == 0 && saw_wrapper {
1102                    // closing the wrapper <div> — drop it
1103                } else {
1104                    writer
1105                        .write_event(Event::End(e))
1106                        .map_err(|err| FeedParseError::new(format!("xhtml write: {err}")))?;
1107                }
1108            }
1109            Event::Empty(e) => {
1110                if depth == 0 && !saw_wrapper && strict {
1111                    return Err(FeedParseError::new(
1112                        "Atom xhtml content must be wrapped in a single \
1113                         XHTML-namespaced <div> (RFC 4287 §3.1.1.3)",
1114                    ));
1115                }
1116                writer
1117                    .write_event(Event::Empty(e))
1118                    .map_err(|err| FeedParseError::new(format!("xhtml write: {err}")))?;
1119            }
1120            Event::Text(e) => {
1121                // Non-whitespace text outside the wrapper violates the spec.
1122                if depth == 0 && !saw_wrapper && strict {
1123                    let s = e.decode().map(|c| c.into_owned()).unwrap_or_default();
1124                    if !s.trim().is_empty() {
1125                        return Err(FeedParseError::new(
1126                            "Atom xhtml content must be wrapped in a single \
1127                             XHTML-namespaced <div> (RFC 4287 §3.1.1.3)",
1128                        ));
1129                    }
1130                }
1131                writer
1132                    .write_event(Event::Text(e))
1133                    .map_err(|err| FeedParseError::new(format!("xhtml write: {err}")))?;
1134            }
1135            Event::CData(e) => writer
1136                .write_event(Event::CData(e))
1137                .map_err(|err| FeedParseError::new(format!("xhtml write: {err}")))?,
1138            Event::Comment(e) => writer
1139                .write_event(Event::Comment(e))
1140                .map_err(|err| FeedParseError::new(format!("xhtml write: {err}")))?,
1141            // Entity references are valid xhtml text content; re-emit them
1142            // verbatim (quick-xml 0.40 no longer keeps them inside `Text`).
1143            Event::GeneralRef(e) => writer
1144                .write_event(Event::GeneralRef(e))
1145                .map_err(|err| FeedParseError::new(format!("xhtml write: {err}")))?,
1146            Event::Eof => {
1147                return Err(FeedParseError::new("unexpected EOF in xhtml content"));
1148            }
1149            // DocType / PI / Decl are not legal inside Atom xhtml content; drop.
1150            _ => {}
1151        }
1152    }
1153}
rama_http/protocols/rss/atom/read.rs

rama_http/protocols/rss/atom/
read.rs
